Add new sync

Cargo.toml

@@ -1,11 +1,11 @@
 [workspace]
 members = [
-    "atuin",
-    "atuin-client",
-    "atuin-server",
-    "atuin-server-postgres",
-    "atuin-server-database",
-    "atuin-common",
+  "atuin",
+  "atuin-client",
+  "atuin-server",
+  "atuin-server-postgres",
+  "atuin-server-database",
+  "atuin-common",
 ]
 
 [workspace.package]
@@ -35,9 +35,10 @@ semver = "1.0.14"
 serde = { version = "1.0.145", features = ["derive"] }
 serde_json = "1.0.86"
 tokio = { version = "1", features = ["full"] }
-uuid = { version = "1.3", features = ["v4"] }
+uuid = { version = "1.3", features = ["v4", "serde"] }
 whoami = "1.1.2"
 typed-builder = "0.14.0"
+pretty_assertions = "1.3.0"
 
 [workspace.dependencies.reqwest]
 version = "0.11"
@@ -46,4 +47,4 @@ default-features = false
 
 [workspace.dependencies.sqlx]
 version = "0.6"
-features = ["runtime-tokio-rustls", "chrono", "postgres"]
+features = ["runtime-tokio-rustls", "chrono", "postgres", "uuid"]

atuin-client/Cargo.toml

@@ -57,7 +57,10 @@ semver = { workspace = true }
 
 # encryption
 rusty_paseto = { version = "0.5.0", default-features = false }
-rusty_paserk = { version = "0.2.0", default-features = false, features = ["v4", "serde"] }
+rusty_paserk = { version = "0.2.0", default-features = false, features = [
+  "v4",
+  "serde",
+] }
 
 # sync
 urlencoding = { version = "2.1.0", optional = true }
@@ -69,3 +72,4 @@ generic-array = { version = "0.14", optional = true, features = ["serde"] }
 
 [dev-dependencies]
 tokio = { version = "1", features = ["full"] }
+pretty_assertions = { workspace = true }

atuin-client/record-migrations/20230531212437_create-records.sql

@@ -7,7 +7,8 @@ create table if not exists records (
   timestamp integer not null,
   tag text not null,
   version text not null,
-  data blob not null
+  data blob not null,
+  cek blob not null
 );
 
 create index host_idx on records (host);

atuin-client/src/api_client.rs

@@ -1,5 +1,6 @@
 use std::collections::HashMap;
 use std::env;
+use uuid::Uuid;
 
 use chrono::Utc;
 use eyre::{bail, Result};
@@ -8,9 +9,13 @@ use reqwest::{
     StatusCode, Url,
 };
 
-use atuin_common::api::{
-    AddHistoryRequest, CountResponse, DeleteHistoryRequest, ErrorResponse, IndexResponse,
-    LoginRequest, LoginResponse, RegisterResponse, StatusResponse, SyncHistoryResponse,
+use atuin_common::record::{EncryptedData, Record};
+use atuin_common::{
+    api::{
+        AddHistoryRequest, CountResponse, DeleteHistoryRequest, ErrorResponse, IndexResponse,
+        LoginRequest, LoginResponse, RegisterResponse, StatusResponse, SyncHistoryResponse,
+    },
+    record::RecordIndex,
 };
 use semver::Version;
 
@@ -195,6 +200,51 @@ impl<'a> Client<'a> {
         Ok(())
     }
 
+    pub async fn post_records(&self, records: &[Record<EncryptedData>]) -> Result<()> {
+        let url = format!("{}/record", self.sync_addr);
+        let url = Url::parse(url.as_str())?;
+
+        self.client.post(url).json(records).send().await?;
+
+        Ok(())
+    }
+
+    pub async fn next_records(
+        &self,
+        host: Uuid,
+        tag: String,
+        start: Option<Uuid>,
+        count: u64,
+    ) -> Result<Vec<Record<EncryptedData>>> {
+        let url = format!(
+            "{}/record/next?host={}&tag={}&count={}",
+            self.sync_addr, host, tag, count
+        );
+        let mut url = Url::parse(url.as_str())?;
+
+        if let Some(start) = start {
+            url.set_query(Some(
+                format!("host={}&tag={}&count={}&start={}", host, tag, count, start).as_str(),
+            ));
+        }
+
+        let resp = self.client.get(url).send().await?;
+
+        let records = resp.json::<Vec<Record<EncryptedData>>>().await?;
+
+        Ok(records)
+    }
+
+    pub async fn record_index(&self) -> Result<RecordIndex> {
+        let url = format!("{}/record", self.sync_addr);
+        let url = Url::parse(url.as_str())?;
+
+        let resp = self.client.get(url).send().await?;
+        let index = resp.json().await?;
+
+        Ok(index)
+    }
+
     pub async fn delete(&self) -> Result<()> {
         let url = format!("{}/account", self.sync_addr);
         let url = Url::parse(url.as_str())?;

atuin-client/src/database.rs

@@ -14,6 +14,7 @@ use sqlx::{
     Result, Row,
 };
 
+
 use super::{
     history::History,
     ordering,
@@ -57,7 +58,7 @@ pub fn current_context() -> Context {
         session,
         hostname,
         cwd,
-        host_id,
+        host_id: host_id.as_simple().to_string(),
     }
 }
 

atuin-client/src/kv.rs

@@ -101,10 +101,7 @@ impl KvStore {
 
         let bytes = record.serialize()?;
 
-        let parent = store
-            .last(host_id.as_str(), KV_TAG)
-            .await?
-            .map(|entry| entry.id);
+        let parent = store.tail(host_id, KV_TAG).await?.map(|entry| entry.id);
 
         let record = atuin_common::record::Record::builder()
             .host(host_id)
@@ -130,17 +127,22 @@ impl KvStore {
         namespace: &str,
         key: &str,
     ) -> Result<Option<KvRecord>> {
-        // TODO: don't load this from disk so much
-        let host_id = Settings::host_id().expect("failed to get host_id");
-
         // Currently, this is O(n). When we have an actual KV store, it can be better
         // Just a poc for now!
 
         // iterate records to find the value we want
         // start at the end, so we get the most recent version
-        let Some(mut record) = store.last(host_id.as_str(), KV_TAG).await? else {
+        let tails = store.tag_tails(KV_TAG).await?;
+
+        if tails.is_empty() {
             return Ok(None);
-        };
+        }
+
+        // first, decide on a record.
+        // try getting the newest first
+        // we always need a way of deciding the "winner" of a write
+        // TODO(ellie): something better than last-write-wins, what if two write at the same time?
+        let mut record = tails.iter().max_by_key(|r| r.timestamp).unwrap().clone();
 
         loop {
             let decrypted = match record.version.as_str() {
@@ -154,7 +156,7 @@ impl KvStore {
             }
 
             if let Some(parent) = decrypted.parent {
-                record = store.get(parent.as_str()).await?;
+                record = store.get(parent).await?;
             } else {
                 break;
             }

atuin-client/src/record/encryption.rs

@@ -6,6 +6,7 @@ use rusty_paseto::core::{
     ImplicitAssertion, Key as DataKey, Local as LocalPurpose, Paseto, PasetoNonce, Payload, V4,
 };
 use serde::{Deserialize, Serialize};
+use uuid::Uuid;
 
 /// Use PASETO V4 Local encryption using the additional data as an implicit assertion.
 #[allow(non_camel_case_types)]
@@ -158,10 +159,10 @@ struct AtuinFooter {
 // This cannot be changed, otherwise it breaks the authenticated encryption.
 #[derive(Debug, Copy, Clone, Serialize)]
 struct Assertions<'a> {
-    id: &'a str,
+    id: &'a Uuid,
     version: &'a str,
     tag: &'a str,
-    host: &'a str,
+    host: &'a Uuid,
 }
 
 impl<'a> From<AdditionalData<'a>> for Assertions<'a> {

atuin-client/src/record/mod.rs

@@ -1,3 +1,4 @@
 pub mod encryption;
 pub mod sqlite_store;
 pub mod store;
+pub mod sync;