aboutsummaryrefslogtreecommitdiffstats
path: root/atuin-client/src
diff options
context:
space:
mode:
Diffstat (limited to 'atuin-client/src')
-rw-r--r--atuin-client/src/api_client.rs59
-rw-r--r--atuin-client/src/database.rs2
-rw-r--r--atuin-client/src/kv.rs22
-rw-r--r--atuin-client/src/record/encryption.rs65
-rw-r--r--atuin-client/src/record/mod.rs2
-rw-r--r--atuin-client/src/record/sqlite_store.rs99
-rw-r--r--atuin-client/src/record/store.rs18
-rw-r--r--atuin-client/src/record/sync.rs421
-rw-r--r--atuin-client/src/settings.rs13
9 files changed, 610 insertions, 91 deletions
diff --git a/atuin-client/src/api_client.rs b/atuin-client/src/api_client.rs
index 350c419d..5ae1ed0a 100644
--- a/atuin-client/src/api_client.rs
+++ b/atuin-client/src/api_client.rs
@@ -8,9 +8,13 @@ use reqwest::{
StatusCode, Url,
};
-use atuin_common::api::{
- AddHistoryRequest, CountResponse, DeleteHistoryRequest, ErrorResponse, IndexResponse,
- LoginRequest, LoginResponse, RegisterResponse, StatusResponse, SyncHistoryResponse,
+use atuin_common::record::{EncryptedData, HostId, Record, RecordId};
+use atuin_common::{
+ api::{
+ AddHistoryRequest, CountResponse, DeleteHistoryRequest, ErrorResponse, IndexResponse,
+ LoginRequest, LoginResponse, RegisterResponse, StatusResponse, SyncHistoryResponse,
+ },
+ record::RecordIndex,
};
use semver::Version;
@@ -195,6 +199,55 @@ 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: HostId,
+ tag: String,
+ start: Option<RecordId>,
+ count: u64,
+ ) -> Result<Vec<Record<EncryptedData>>> {
+ let url = format!(
+ "{}/record/next?host={}&tag={}&count={}",
+ self.sync_addr, host.0, 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.0, tag, count, start.0
+ )
+ .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())?;
diff --git a/atuin-client/src/database.rs b/atuin-client/src/database.rs
index b7b44409..218c1d6e 100644
--- a/atuin-client/src/database.rs
+++ b/atuin-client/src/database.rs
@@ -57,7 +57,7 @@ pub fn current_context() -> Context {
session,
hostname,
cwd,
- host_id,
+ host_id: host_id.0.as_simple().to_string(),
}
}
diff --git a/atuin-client/src/kv.rs b/atuin-client/src/kv.rs
index c365a385..30018d63 100644
--- a/atuin-client/src/kv.rs
+++ b/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;
}
diff --git a/atuin-client/src/record/encryption.rs b/atuin-client/src/record/encryption.rs
index f14bf027..6760d97b 100644
--- a/atuin-client/src/record/encryption.rs
+++ b/atuin-client/src/record/encryption.rs
@@ -1,4 +1,6 @@
-use atuin_common::record::{AdditionalData, DecryptedData, EncryptedData, Encryption};
+use atuin_common::record::{
+ AdditionalData, DecryptedData, EncryptedData, Encryption, HostId, RecordId,
+};
use base64::{engine::general_purpose, Engine};
use eyre::{ensure, Context, Result};
use rusty_paserk::{Key, KeyId, Local, PieWrappedKey};
@@ -158,10 +160,11 @@ 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 RecordId,
version: &'a str,
tag: &'a str,
- host: &'a str,
+ host: &'a HostId,
+ parent: Option<&'a RecordId>,
}
impl<'a> From<AdditionalData<'a>> for Assertions<'a> {
@@ -171,6 +174,7 @@ impl<'a> From<AdditionalData<'a>> for Assertions<'a> {
version: ad.version,
tag: ad.tag,
host: ad.host,
+ parent: ad.parent,
}
}
}
@@ -183,7 +187,7 @@ impl Assertions<'_> {
#[cfg(test)]
mod tests {
- use atuin_common::record::Record;
+ use atuin_common::{record::Record, utils::uuid_v7};
use super::*;
@@ -192,10 +196,11 @@ mod tests {
let key = Key::<V4, Local>::new_os_random();
let ad = AdditionalData {
- id: "foo",
+ id: &RecordId(uuid_v7()),
version: "v0",
tag: "kv",
- host: "1234",
+ host: &HostId(uuid_v7()),
+ parent: None,
};
let data = DecryptedData(vec![1, 2, 3, 4]);
@@ -210,10 +215,11 @@ mod tests {
let key = Key::<V4, Local>::new_os_random();
let ad = AdditionalData {
- id: "foo",
+ id: &RecordId(uuid_v7()),
version: "v0",
tag: "kv",
- host: "1234",
+ host: &HostId(uuid_v7()),
+ parent: None,
};
let data = DecryptedData(vec![1, 2, 3, 4]);
@@ -233,10 +239,11 @@ mod tests {
let fake_key = Key::<V4, Local>::new_os_random();
let ad = AdditionalData {
- id: "foo",
+ id: &RecordId(uuid_v7()),
version: "v0",
tag: "kv",
- host: "1234",
+ host: &HostId(uuid_v7()),
+ parent: None,
};
let data = DecryptedData(vec![1, 2, 3, 4]);
@@ -250,10 +257,11 @@ mod tests {
let key = Key::<V4, Local>::new_os_random();
let ad = AdditionalData {
- id: "foo",
+ id: &RecordId(uuid_v7()),
version: "v0",
tag: "kv",
- host: "1234",
+ host: &HostId(uuid_v7()),
+ parent: None,
};
let data = DecryptedData(vec![1, 2, 3, 4]);
@@ -261,10 +269,8 @@ mod tests {
let encrypted = PASETO_V4::encrypt(data, ad, &key.to_bytes());
let ad = AdditionalData {
- id: "foo1",
- version: "v0",
- tag: "kv",
- host: "1234",
+ id: &RecordId(uuid_v7()),
+ ..ad
};
let _ = PASETO_V4::decrypt(encrypted, ad, &key.to_bytes()).unwrap_err();
}
@@ -275,10 +281,11 @@ mod tests {
let key2 = Key::<V4, Local>::new_os_random();
let ad = AdditionalData {
- id: "foo",
+ id: &RecordId(uuid_v7()),
version: "v0",
tag: "kv",
- host: "1234",
+ host: &HostId(uuid_v7()),
+ parent: None,
};
let data = DecryptedData(vec![1, 2, 3, 4]);
@@ -304,10 +311,10 @@ mod tests {
fn full_record_round_trip() {
let key = [0x55; 32];
let record = Record::builder()
- .id("1".to_owned())
+ .id(RecordId(uuid_v7()))
.version("v0".to_owned())
.tag("kv".to_owned())
- .host("host1".to_owned())
+ .host(HostId(uuid_v7()))
.timestamp(1687244806000000)
.data(DecryptedData(vec![1, 2, 3, 4]))
.build();
@@ -316,30 +323,20 @@ mod tests {
assert!(!encrypted.data.data.is_empty());
assert!(!encrypted.data.content_encryption_key.is_empty());
- assert_eq!(encrypted.id, "1");
- assert_eq!(encrypted.host, "host1");
- assert_eq!(encrypted.version, "v0");
- assert_eq!(encrypted.tag, "kv");
- assert_eq!(encrypted.timestamp, 1687244806000000);
let decrypted = encrypted.decrypt::<PASETO_V4>(&key).unwrap();
assert_eq!(decrypted.data.0, [1, 2, 3, 4]);
- assert_eq!(decrypted.id, "1");
- assert_eq!(decrypted.host, "host1");
- assert_eq!(decrypted.version, "v0");
- assert_eq!(decrypted.tag, "kv");
- assert_eq!(decrypted.timestamp, 1687244806000000);
}
#[test]
fn full_record_round_trip_fail() {
let key = [0x55; 32];
let record = Record::builder()
- .id("1".to_owned())
+ .id(RecordId(uuid_v7()))
.version("v0".to_owned())
.tag("kv".to_owned())
- .host("host1".to_owned())
+ .host(HostId(uuid_v7()))
.timestamp(1687244806000000)
.data(DecryptedData(vec![1, 2, 3, 4]))
.build();
@@ -347,13 +344,13 @@ mod tests {
let encrypted = record.encrypt::<PASETO_V4>(&key);
let mut enc1 = encrypted.clone();
- enc1.host = "host2".to_owned();
+ enc1.host = HostId(uuid_v7());
let _ = enc1
.decrypt::<PASETO_V4>(&key)
.expect_err("tampering with the host should result in auth failure");
let mut enc2 = encrypted;
- enc2.id = "2".to_owned();
+ enc2.id = RecordId(uuid_v7());
let _ = enc2
.decrypt::<PASETO_V4>(&key)
.expect_err("tampering with the id should result in auth failure");
diff --git a/atuin-client/src/record/mod.rs b/atuin-client/src/record/mod.rs
index 9ac2c541..8bc816ae 100644
--- a/atuin-client/src/record/mod.rs
+++ b/atuin-client/src/record/mod.rs
@@ -1,3 +1,5 @@
pub mod encryption;
pub mod sqlite_store;
pub mod store;
+#[cfg(feature = "sync")]
+pub mod sync;
diff --git a/atuin-client/src/record/sqlite_store.rs b/atuin-client/src/record/sqlite_store.rs
index f692c0c2..14a7e277 100644
--- a/atuin-client/src/record/sqlite_store.rs
+++ b/atuin-client/src/record/sqlite_store.rs
@@ -8,12 +8,14 @@ use std::str::FromStr;
use async_trait::async_trait;
use eyre::{eyre, Result};
use fs_err as fs;
+use futures::TryStreamExt;
use sqlx::{
sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePool, SqlitePoolOptions, SqliteRow},
Row,
};
-use atuin_common::record::{EncryptedData, Record};
+use atuin_common::record::{EncryptedData, HostId, Record, RecordId, RecordIndex};
+use uuid::Uuid;
use super::store::Store;
@@ -62,11 +64,11 @@ impl SqliteStore {
"insert or ignore into records(id, host, tag, timestamp, parent, version, data, cek)
values(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
)
- .bind(r.id.as_str())
- .bind(r.host.as_str())
+ .bind(r.id.0.as_simple().to_string())
+ .bind(r.host.0.as_simple().to_string())
.bind(r.tag.as_str())
.bind(r.timestamp as i64)
- .bind(r.parent.as_ref())
+ .bind(r.parent.map(|p| p.0.as_simple().to_string()))
.bind(r.version.as_str())
.bind(r.data.data.as_str())
.bind(r.data.content_encryption_key.as_str())
@@ -79,10 +81,18 @@ impl SqliteStore {
fn query_row(row: SqliteRow) -> Record<EncryptedData> {
let timestamp: i64 = row.get("timestamp");
+ // tbh at this point things are pretty fucked so just panic
+ let id = Uuid::from_str(row.get("id")).expect("invalid id UUID format in sqlite DB");
+ let host = Uuid::from_str(row.get("host")).expect("invalid host UUID format in sqlite DB");
+ let parent: Option<&str> = row.get("parent");
+
+ let parent = parent
+ .map(|parent| Uuid::from_str(parent).expect("invalid parent UUID format in sqlite DB"));
+
Record {
- id: row.get("id"),
- host: row.get("host"),
- parent: row.get("parent"),
+ id: RecordId(id),
+ host: HostId(host),
+ parent: parent.map(RecordId),
timestamp: timestamp as u64,
tag: row.get("tag"),
version: row.get("version"),
@@ -111,9 +121,9 @@ impl Store for SqliteStore {
Ok(())
}
- async fn get(&self, id: &str) -> Result<Record<EncryptedData>> {
+ async fn get(&self, id: RecordId) -> Result<Record<EncryptedData>> {
let res = sqlx::query("select * from records where id = ?1")
- .bind(id)
+ .bind(id.0.as_simple().to_string())
.map(Self::query_row)
.fetch_one(&self.pool)
.await?;
@@ -121,10 +131,10 @@ impl Store for SqliteStore {
Ok(res)
}
- async fn len(&self, host: &str, tag: &str) -> Result<u64> {
+ async fn len(&self, host: HostId, tag: &str) -> Result<u64> {
let res: (i64,) =
sqlx::query_as("select count(1) from records where host = ?1 and tag = ?2")
- .bind(host)
+ .bind(host.0.as_simple().to_string())
.bind(tag)
.fetch_one(&self.pool)
.await?;
@@ -134,7 +144,7 @@ impl Store for SqliteStore {
async fn next(&self, record: &Record<EncryptedData>) -> Result<Option<Record<EncryptedData>>> {
let res = sqlx::query("select * from records where parent = ?1")
- .bind(record.id.clone())
+ .bind(record.id.0.as_simple().to_string())
.map(Self::query_row)
.fetch_one(&self.pool)
.await;
@@ -146,11 +156,11 @@ impl Store for SqliteStore {
}
}
- async fn first(&self, host: &str, tag: &str) -> Result<Option<Record<EncryptedData>>> {
+ async fn head(&self, host: HostId, tag: &str) -> Result<Option<Record<EncryptedData>>> {
let res = sqlx::query(
"select * from records where host = ?1 and tag = ?2 and parent is null limit 1",
)
- .bind(host)
+ .bind(host.0.as_simple().to_string())
.bind(tag)
.map(Self::query_row)
.fetch_optional(&self.pool)
@@ -159,23 +169,53 @@ impl Store for SqliteStore {
Ok(res)
}
- async fn last(&self, host: &str, tag: &str) -> Result<Option<Record<EncryptedData>>> {
+ async fn tail(&self, host: HostId, tag: &str) -> Result<Option<Record<EncryptedData>>> {
let res = sqlx::query(
"select * from records rp where tag=?1 and host=?2 and (select count(1) from records where parent=rp.id) = 0;",
)
.bind(tag)
- .bind(host)
+ .bind(host.0.as_simple().to_string())
.map(Self::query_row)
.fetch_optional(&self.pool)
.await?;
Ok(res)
}
+
+ async fn tag_tails(&self, tag: &str) -> Result<Vec<Record<EncryptedData>>> {
+ let res = sqlx::query(
+ "select * from records rp where tag=?1 and (select count(1) from records where parent=rp.id) = 0;",
+ )
+ .bind(tag)
+ .map(Self::query_row)
+ .fetch_all(&self.pool)
+ .await?;
+
+ Ok(res)
+ }
+
+ async fn tail_records(&self) -> Result<RecordIndex> {
+ let res = sqlx::query(
+ "select host, tag, id from records rp where (select count(1) from records where parent=rp.id) = 0;",
+ )
+ .map(|row: SqliteRow| {
+ let host: Uuid= Uuid::from_str(row.get("host")).expect("invalid uuid in db host");
+ let tag: String= row.get("tag");
+ let id: Uuid= Uuid::from_str(row.get("id")).expect("invalid uuid in db id");
+
+ (HostId(host), tag, RecordId(id))
+ })
+ .fetch(&self.pool)
+ .try_collect()
+ .await?;
+
+ Ok(res)
+ }
}
#[cfg(test)]
mod tests {
- use atuin_common::record::{EncryptedData, Record};
+ use atuin_common::record::{EncryptedData, HostId, Record};
use crate::record::{encryption::PASETO_V4, store::Store};
@@ -183,7 +223,7 @@ mod tests {
fn test_record() -> Record<EncryptedData> {
Record::builder()
- .host(atuin_common::utils::uuid_v7().simple().to_string())
+ .host(HostId(atuin_common::utils::uuid_v7()))
.version("v1".into())
.tag(atuin_common::utils::uuid_v7().simple().to_string())
.data(EncryptedData {
@@ -218,10 +258,7 @@ mod tests {
let record = test_record();
db.push(&record).await.unwrap();
- let new_record = db
- .get(record.id.as_str())
- .await
- .expect("failed to fetch record");
+ let new_record = db.get(record.id).await.expect("failed to fetch record");
assert_eq!(record, new_record, "records are not equal");
}
@@ -233,7 +270,7 @@ mod tests {
db.push(&record).await.unwrap();
let len = db
- .len(record.host.as_str(), record.tag.as_str())
+ .len(record.host, record.tag.as_str())
.await
.expect("failed to get store len");
@@ -253,14 +290,8 @@ mod tests {
db.push(&first).await.unwrap();
db.push(&second).await.unwrap();
- let first_len = db
- .len(first.host.as_str(), first.tag.as_str())
- .await
- .unwrap();
- let second_len = db
- .len(second.host.as_str(), second.tag.as_str())
- .await
- .unwrap();
+ let first_len = db.len(first.host, first.tag.as_str()).await.unwrap();
+ let second_len = db.len(second.host, second.tag.as_str()).await.unwrap();
assert_eq!(first_len, 1, "expected length of 1 after insert");
assert_eq!(second_len, 1, "expected length of 1 after insert");
@@ -281,7 +312,7 @@ mod tests {
}
assert_eq!(
- db.len(tail.host.as_str(), tail.tag.as_str()).await.unwrap(),
+ db.len(tail.host, tail.tag.as_str()).await.unwrap(),
100,
"failed to insert 100 records"
);
@@ -304,7 +335,7 @@ mod tests {
db.push_batch(records.iter()).await.unwrap();
assert_eq!(
- db.len(tail.host.as_str(), tail.tag.as_str()).await.unwrap(),
+ db.len(tail.host, tail.tag.as_str()).await.unwrap(),
10000,
"failed to insert 10k records"
);
@@ -327,7 +358,7 @@ mod tests {
db.push_batch(records.iter()).await.unwrap();
let mut record = db
- .first(tail.host.as_str(), tail.tag.as_str())
+ .head(tail.host, tail.tag.as_str())
.await
.expect("in memory sqlite should not fail")
.expect("entry exists");
diff --git a/atuin-client/src/record/store.rs b/atuin-client/src/record/store.rs
index 9ea7007a..45d554ef 100644
--- a/atuin-client/src/record/store.rs
+++ b/atuin-client/src/record/store.rs
@@ -1,7 +1,7 @@
use async_trait::async_trait;
use eyre::Result;
-use atuin_common::record::{EncryptedData, Record};
+use atuin_common::record::{EncryptedData, HostId, Record, RecordId, RecordIndex};
/// A record store stores records
/// In more detail - we tend to need to process this into _another_ format to actually query it.
@@ -20,14 +20,22 @@ pub trait Store {
records: impl Iterator<Item = &Record<EncryptedData>> + Send + Sync,
) -> Result<()>;
- async fn get(&self, id: &str) -> Result<Record<EncryptedData>>;
- async fn len(&self, host: &str, tag: &str) -> Result<u64>;
+ async fn get(&self, id: RecordId) -> Result<Record<EncryptedData>>;
+ async fn len(&self, host: HostId, tag: &str) -> Result<u64>;
/// Get the record that follows this record
async fn next(&self, record: &Record<EncryptedData>) -> Result<Option<Record<EncryptedData>>>;
/// Get the first record for a given host and tag
- async fn first(&self, host: &str, tag: &str) -> Result<Option<Record<EncryptedData>>>;
+ async fn head(&self, host: HostId, tag: &str) -> Result<Option<Record<EncryptedData>>>;
+
/// Get the last record for a given host and tag
- async fn last(&self, host: &str, tag: &str) -> Result<Option<Record<EncryptedData>>>;
+ async fn tail(&self, host: HostId, tag: &str) -> Result<Option<Record<EncryptedData>>>;
+
+ // Get the last record for all hosts for a given tag, useful for the read path of apps.
+ async fn tag_tails(&self, tag: &str) -> Result<Vec<Record<EncryptedData>>>;
+
+ // Get the latest host/tag/record tuple for every set in the store. useful for building an
+ // index
+ async fn tail_records(&self) -> Result<RecordIndex>;
}
diff --git a/atuin-client/src/record/sync.rs b/atuin-client/src/record/sync.rs
new file mode 100644
index 00000000..ebdb8eb2
--- /dev/null
+++ b/atuin-client/src/record/sync.rs
@@ -0,0 +1,421 @@
+// do a sync :O
+use eyre::Result;
+
+use super::store::Store;
+use crate::{api_client::Client, settings::Settings};
+
+use atuin_common::record::{Diff, HostId, RecordId, RecordIndex};
+
+#[derive(Debug, Eq, PartialEq)]
+pub enum Operation {
+ // Either upload or download until the tail matches the below
+ Upload {
+ tail: RecordId,
+ host: HostId,
+ tag: String,
+ },
+ Download {
+ tail: RecordId,
+ host: HostId,
+ tag: String,
+ },
+}
+
+pub async fn diff(settings: &Settings, store: &mut impl Store) -> Result<(Vec<Diff>, RecordIndex)> {
+ let client = Client::new(&settings.sync_address, &settings.session_token)?;
+
+ let local_index = store.tail_records().await?;
+ let remote_index = client.record_index().await?;
+
+ let diff = local_index.diff(&remote_index);
+
+ Ok((diff, remote_index))
+}
+
+// Take a diff, along with a local store, and resolve it into a set of operations.
+// With the store as context, we can determine if a tail exists locally or not and therefore if it needs uploading or download.
+// In theory this could be done as a part of the diffing stage, but it's easier to reason
+// about and test this way
+pub async fn operations(diffs: Vec<Diff>, store: &impl Store) -> Result<Vec<Operation>> {
+ let mut operations = Vec::with_capacity(diffs.len());
+
+ for diff in diffs {
+ // First, try to fetch the tail
+ // If it exists locally, then that means we need to update the remote
+ // host until it has the same tail. Ie, upload.
+ // If it does not exist locally, that means remote is ahead of us.
+ // Therefore, we need to download until our local tail matches
+ let record = store.get(diff.tail).await;
+
+ let op = if record.is_ok() {
+ // if local has the ID, then we should find the actual tail of this
+ // store, so we know what we need to update the remote to.
+ let tail = store
+ .tail(diff.host, diff.tag.as_str())
+ .await?
+ .expect("failed to fetch last record, expected tag/host to exist");
+
+ // TODO(ellie) update the diffing so that it stores the context of the current tail
+ // that way, we can determine how much we need to upload.
+ // For now just keep uploading until tails match
+
+ Operation::Upload {
+ tail: tail.id,
+ host: diff.host,
+ tag: diff.tag,
+ }
+ } else {
+ Operation::Download {
+ tail: diff.tail,
+ host: diff.host,
+ tag: diff.tag,
+ }
+ };
+
+ operations.push(op);
+ }
+
+ // sort them - purely so we have a stable testing order, and can rely on
+ // same input = same output
+ // We can sort by ID so long as we continue to use UUIDv7 or something
+ // with the same properties
+
+ operations.sort_by_key(|op| match op {
+ Operation::Upload { tail, host, .. } => ("upload", *host, *tail),
+ Operation::Download { tail, host, .. } => ("download", *host, *tail),
+ });
+
+ Ok(operations)
+}
+
+async fn sync_upload(
+ store: &mut impl Store,
+ remote_index: &RecordIndex,
+ client: &Client<'_>,
+ op: (HostId, String, RecordId),
+) -> Result<i64> {
+ let upload_page_size = 100;
+ let mut total = 0;
+
+ // so. we have an upload operation, with the tail representing the state
+ // we want to get the remote to
+ let current_tail = remote_index.get(op.0, op.1.clone());
+
+ println!(
+ "Syncing local {:?}/{}/{:?}, remote has {:?}",
+ op.0, op.1, op.2, current_tail
+ );
+
+ let start = if let Some(current_tail) = current_tail {
+ current_tail
+ } else {
+ store
+ .head(op.0, op.1.as_str())
+ .await
+ .expect("failed to fetch host/tag head")
+ .expect("host/tag not in current index")
+ .id
+ };
+
+ debug!("starting push to remote from: {:?}", start);
+
+ // we have the start point for sync. it is either the head of the store if
+ // the remote has no data for it, or the tail that the remote has
+ // we need to iterate from the remote tail, and keep going until
+ // remote tail = current local tail
+
+ let mut record = Some(store.get(start).await.unwrap());
+
+ let mut buf = Vec::with_capacity(upload_page_size);
+
+ while let Some(r) = record {
+ if buf.len() < upload_page_size {
+ buf.push(r.clone());
+ } else {
+ client.post_records(&buf).await?;
+
+ // can we reset what we have? len = 0 but keep capacity
+ buf = Vec::with_capacity(upload_page_size);
+ }
+ record = store.next(&r).await?;
+
+ total += 1;
+ }
+
+ if !buf.is_empty() {
+ client.post_records(&buf).await?;
+ }
+
+ Ok(total)
+}
+
+async fn sync_download(
+ store: &mut impl Store,
+ remote_index: &RecordIndex,
+ client: &Client<'_>,
+ op: (HostId, String, RecordId),
+) -> Result<i64> {
+ // TODO(ellie): implement variable page sizing like on history sync
+ let download_page_size = 1000;
+
+ let mut total = 0;
+
+ // We know that the remote is ahead of us, so let's keep downloading until both
+ // 1) The remote stops returning full pages
+ // 2) The tail equals what we expect
+ //
+ // If (1) occurs without (2), then something is wrong with our index calculation
+ // and we should bail.
+ let remote_tail = remote_index
+ .get(op.0, op.1.clone())
+ .expect("remote index does not contain expected tail during download");
+ let local_tail = store.tail(op.0, op.1.as_str()).await?;
+ //
+ // We expect that the operations diff will represent the desired state
+ // In this case, that contains the remote tail.
+ assert_eq!(remote_tail, op.2);
+
+ println!("Downloading {:?}/{}/{:?} to local", op.0, op.1, op.2);
+
+ let mut records = client
+ .next_records(
+ op.0,
+ op.1.clone(),
+ local_tail.map(|r| r.id),
+ download_page_size,
+ )
+ .await?;
+
+ while !records.is_empty() {
+ total += std::cmp::min(download_page_size, records.len() as u64);
+ store.push_batch(records.iter()).await?;
+
+ if records.last().unwrap().id == remote_tail {
+ break;
+ }
+
+ records = client
+ .next_records(
+ op.0,
+ op.1.clone(),
+ records.last().map(|r| r.id),
+ download_page_size,
+ )
+ .await?;
+ }
+
+ Ok(total as i64)
+}
+
+pub async fn sync_remote(
+ operations: Vec<Operation>,
+ remote_index: &RecordIndex,
+ local_store: &mut impl Store,
+ settings: &Settings,
+) -> Result<(i64, i64)> {
+ let client = Client::new(&settings.sync_address, &settings.session_token)?;
+
+ let mut uploaded = 0;
+ let mut downloaded = 0;
+
+ // this can totally run in parallel, but lets get it working first
+ for i in operations {
+ match i {
+ Operation::Upload { tail, host, tag } => {
+ uploaded +=
+ sync_upload(local_store, remote_index, &client, (host, tag, tail)).await?
+ }
+ Operation::Download { tail, host, tag } => {
+ downloaded +=
+ sync_download(local_store, remote_index, &client, (host, tag, tail)).await?
+ }
+ }
+ }
+
+ Ok((uploaded, downloaded))
+}
+
+#[cfg(test)]
+mod tests {
+ use atuin_common::record::{Diff, EncryptedData, HostId, Record};
+ use pretty_assertions::assert_eq;
+
+ use crate::record::{
+ encryption::PASETO_V4,
+ sqlite_store::SqliteStore,
+ store::Store,
+ sync::{self, Operation},
+ };
+
+ fn test_record() -> Record<EncryptedData> {
+ Record::builder()
+ .host(HostId(atuin_common::utils::uuid_v7()))
+ .version("v1".into())
+ .tag(atuin_common::utils::uuid_v7().simple().to_string())
+ .data(EncryptedData {
+ data: String::new(),
+ content_encryption_key: String::new(),
+ })
+ .build()
+ }
+
+ // Take a list of local records, and a list of remote records.
+ // Return the local database, and a diff of local/remote, ready to build
+ // ops
+ async fn build_test_diff(
+ local_records: Vec<Record<EncryptedData>>,
+ remote_records: Vec<Record<EncryptedData>>,
+ ) -> (SqliteStore, Vec<Diff>) {
+ let local_store = SqliteStore::new(":memory:")
+ .await
+ .expect("failed to open in memory sqlite");
+ let remote_store = SqliteStore::new(":memory:")
+ .await
+ .expect("failed to open in memory sqlite"); // "remote"
+
+ for i in local_records {
+ local_store.push(&i).await.unwrap();
+ }
+
+ for i in remote_records {
+ remote_store.push(&i).await.unwrap();
+ }
+
+ let local_index = local_store.tail_records().await.unwrap();
+ let remote_index = remote_store.tail_records().await.unwrap();
+
+ let diff = local_index.diff(&remote_index);
+
+ (local_store, diff)
+ }
+
+ #[tokio::test]
+ async fn test_basic_diff() {
+ // a diff where local is ahead of remote. nothing else.
+
+ let record = test_record();
+ let (store, diff) = build_test_diff(vec![record.clone()], vec![]).await;
+
+ assert_eq!(diff.len(), 1);
+
+ let operations = sync::operations(diff, &store).await.unwrap();
+
+ assert_eq!(operations.len(), 1);
+
+ assert_eq!(
+ operations[0],
+ Operation::Upload {
+ host: record.host,
+ tag: record.tag,
+ tail: record.id
+ }
+ );
+ }
+
+ #[tokio::test]
+ async fn build_two_way_diff() {
+ // a diff where local is ahead of remote for one, and remote for
+ // another. One upload, one download
+
+ let shared_record = test_record();
+
+ let remote_ahead = test_record();
+ let local_ahead = shared_record
+ .new_child(vec![1, 2, 3])
+ .encrypt::<PASETO_V4>(&[0; 32]);
+
+ let local = vec![shared_record.clone(), local_ahead.clone()]; // local knows about the already synced, and something newer in the same store
+ let remote = vec![shared_record.clone(), remote_ahead.clone()]; // remote knows about the already-synced, and one new record in a new store
+
+ let (store, diff) = build_test_diff(local, remote).await;
+ let operations = sync::operations(diff, &store).await.unwrap();
+
+ assert_eq!(operations.len(), 2);
+
+ assert_eq!(
+ operations,
+ vec![
+ Operation::Download {
+ tail: remote_ahead.id,
+ host: remote_ahead.host,
+ tag: remote_ahead.tag,
+ },
+ Operation::Upload {
+ tail: local_ahead.id,
+ host: local_ahead.host,
+ tag: local_ahead.tag,
+ },
+ ]
+ );
+ }
+
+ #[tokio::test]
+ async fn build_complex_diff() {
+ // One shared, ahead but known only by remote
+ // One known only by local
+ // One known only by remote
+
+ let shared_record = test_record();
+
+ let remote_known = test_record();
+ let local_known = test_record();
+
+ let second_shared = test_record();
+ let second_shared_remote_ahead = second_shared
+ .new_child(vec![1, 2, 3])
+ .encrypt::<PASETO_V4>(&[0; 32]);
+
+ let local_ahead = shared_record
+ .new_child(vec![1, 2, 3])
+ .encrypt::<PASETO_V4>(&[0; 32]);
+
+ let local = vec![
+ shared_record.clone(),
+ second_shared.clone(),
+ local_known.clone(),
+ local_ahead.clone(),
+ ];
+
+ let remote = vec![
+ shared_record.clone(),
+ second_shared.clone(),
+ second_shared_remote_ahead.clone(),
+ remote_known.clone(),
+ ]; // remote knows about the already-synced, and one new record in a new store
+
+ let (store, diff) = build_test_diff(local, remote).await;
+ let operations = sync::operations(diff, &store).await.unwrap();
+
+ assert_eq!(operations.len(), 4);
+
+ let mut result_ops = vec![
+ Operation::Download {
+ tail: remote_known.id,
+ host: remote_known.host,
+ tag: remote_known.tag,
+ },
+ Operation::Download {
+ tail: second_shared_remote_ahead.id,
+ host: second_shared.host,
+ tag: second_shared.tag,
+ },
+ Operation::Upload {
+ tail: local_ahead.id,
+ host: local_ahead.host,
+ tag: local_ahead.tag,
+ },
+ Operation::Upload {
+ tail: local_known.id,
+ host: local_known.host,
+ tag: local_known.tag,
+ },
+ ];
+
+ result_ops.sort_by_key(|op| match op {
+ Operation::Upload { tail, host, .. } => ("upload", *host, *tail),
+ Operation::Download { tail, host, .. } => ("download", *host, *tail),
+ });
+
+ assert_eq!(operations, result_ops);
+ }
+}
diff --git a/atuin-client/src/settings.rs b/atuin-client/src/settings.rs
index dd072451..bb41a890 100644
--- a/atuin-client/src/settings.rs
+++ b/atuin-client/src/settings.rs
@@ -1,8 +1,10 @@
use std::{
io::prelude::*,
path::{Path, PathBuf},
+ str::FromStr,
};
+use atuin_common::record::HostId;
use chrono::{prelude::*, Utc};
use clap::ValueEnum;
use config::{Config, Environment, File as ConfigFile, FileFormat};
@@ -12,6 +14,7 @@ use parse_duration::parse;
use regex::RegexSet;
use semver::Version;
use serde::Deserialize;
+use uuid::Uuid;
pub const HISTORY_PAGE_SIZE: i64 = 100;
pub const LAST_SYNC_FILENAME: &str = "last_sync_time";
@@ -228,11 +231,13 @@ impl Settings {
Settings::load_time_from_file(LAST_VERSION_CHECK_FILENAME)
}
- pub fn host_id() -> Option<String> {
+ pub fn host_id() -> Option<HostId> {
let id = Settings::read_from_data_dir(HOST_ID_FILENAME);
- if id.is_some() {
- return id;
+ if let Some(id) = id {
+ let parsed =
+ Uuid::from_str(id.as_str()).expect("failed to parse host ID from local directory");
+ return Some(HostId(parsed));
}
let uuid = atuin_common::utils::uuid_v7();
@@ -240,7 +245,7 @@ impl Settings {
Settings::save_to_data_dir(HOST_ID_FILENAME, uuid.as_simple().to_string().as_ref())
.expect("Could not write host ID to data dir");
- Some(uuid.as_simple().to_string())
+ Some(HostId(uuid))
}
pub fn should_sync(&self) -> Result<bool> {
generated by cgit v1.3.1 (git 2.54.0) at 2026-06-14 10:24:00 +0000