diff --git a/Cargo.lock b/Cargo.lock index fc2bb35..b2ff9b1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2327,6 +2327,7 @@ dependencies = [ name = "ely_sync_client" version = "0.1.0" dependencies = [ + "ed25519-dalek", "ely_domain", "serde", "serde_json", diff --git a/cloudflare/src/bindings.ts b/cloudflare/src/bindings.ts index 4175600..69f2d1a 100644 --- a/cloudflare/src/bindings.ts +++ b/cloudflare/src/bindings.ts @@ -1,5 +1,6 @@ export interface ElyKvNamespace { get(key: string): Promise; + put(key: string, value: string): Promise; delete(key: string): Promise; } diff --git a/cloudflare/src/devices.ts b/cloudflare/src/devices.ts index 800f5b9..1e6b788 100644 --- a/cloudflare/src/devices.ts +++ b/cloudflare/src/devices.ts @@ -1,4 +1,5 @@ import type { AuthContext } from "./auth.js"; +import { authSessionCacheKvKey } from "./auth.js"; import type { Env } from "./bindings.js"; import { type DeviceApprovalDocument, @@ -227,6 +228,7 @@ export async function registerDeviceDocument( throw new DevicePersistenceError("device_registration_missing"); } await bindSessionDeviceContext(env, context, registration.deviceId, nowSeconds); + await refreshSessionDeviceCache(env, context, registration.deviceId); return { version: 1, @@ -246,6 +248,23 @@ async function bindSessionDeviceContext( .run(); } +async function refreshSessionDeviceCache( + env: Env, + context: AuthContext, + deviceId: string, +): Promise { + await env.ELY_KV.put( + authSessionCacheKvKey(env.ELY_ENVIRONMENT, context.tokenHash), + JSON.stringify({ + version: 1, + user_id: context.userId, + session_id: context.sessionId, + device_id: deviceId, + expires_at: context.expiresAt, + }), + ); +} + export async function approveDeviceDocument( request: Request, env: Env, diff --git a/cloudflare/tests/api_controls.test.ts b/cloudflare/tests/api_controls.test.ts index 94e5a2d..5160415 100644 --- a/cloudflare/tests/api_controls.test.ts +++ b/cloudflare/tests/api_controls.test.ts @@ -319,6 +319,10 @@ function testEnv(options: TestEnvOptions = {}): Env { options.kvReads?.push(key); return Promise.resolve(values.get(key) ?? null); }, + put(key: string, value: string): Promise { + values.set(key, value); + return Promise.resolve(); + }, delete(key: string): Promise { values.delete(key); return Promise.resolve(); diff --git a/cloudflare/tests/better_auth_email_otp.test.ts b/cloudflare/tests/better_auth_email_otp.test.ts index c35f659..8fef5e5 100644 --- a/cloudflare/tests/better_auth_email_otp.test.ts +++ b/cloudflare/tests/better_auth_email_otp.test.ts @@ -47,6 +47,9 @@ function testEnv(sentEmails: ElyEmailMessageBuilder[] | null): Env { get() { return Promise.resolve(null); }, + put() { + return Promise.resolve(); + }, delete() { return Promise.resolve(); }, diff --git a/cloudflare/tests/devices_routes.test.ts b/cloudflare/tests/devices_routes.test.ts index a92c4ac..89547e4 100644 --- a/cloudflare/tests/devices_routes.test.ts +++ b/cloudflare/tests/devices_routes.test.ts @@ -157,6 +157,8 @@ describe("device routes", () => { it("registers the current device as a pending idempotent D1 write", async () => { const tokenHash = await authTokenHash(ACCESS_TOKEN); + const sessionCacheKey = authSessionCacheKvKey("local", tokenHash); + const kvPuts: [string, string][] = []; const d1 = testD1Database([ { device_id: "device-01", @@ -182,7 +184,8 @@ describe("device routes", () => { }), testEnv({ d1, - kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument()]], + kvEntries: [[sessionCacheKey, sessionDocument()]], + kvPuts, }), ); @@ -220,6 +223,46 @@ describe("device routes", () => { assert.equal(d1.binds[0]?.[7], IDEMPOTENCY_KEY); assert.deepEqual(d1.binds[1], ["user-01", IDEMPOTENCY_KEY]); assert.deepEqual(d1.binds[2]?.slice(0, 3), ["session-01", "user-01", "device-01"]); + assert.deepEqual(kvPuts, [[sessionCacheKey, sessionDocument("device-01")]]); + }); + + it("registers and caches device context for sessions without a current device", async () => { + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const sessionCacheKey = authSessionCacheKvKey("local", tokenHash); + const kvPuts: [string, string][] = []; + const d1 = testD1Database([ + { + device_id: "device-01", + public_key: PUBLIC_KEY, + device_name: "MacBook Pro", + platform: "macOS", + approval_status: "pending", + created_at: 1_780_000_100, + approved_at: null, + last_active_at: 1_780_000_100, + revoked_at: null, + }, + ]); + + const response = await handleRequest( + new Request("https://elydora.test/api/devices/register", { + method: "POST", + headers: { + authorization: `Bearer ${ACCESS_TOKEN}`, + "content-type": "application/json", + }, + body: JSON.stringify(deviceRegistrationBody()), + }), + testEnv({ + d1, + kvEntries: [[sessionCacheKey, sessionDocument(null)]], + kvPuts, + }), + ); + + assert.equal(response.status, 201); + assert.deepEqual(d1.binds[2]?.slice(0, 3), ["session-01", "user-01", "device-01"]); + assert.deepEqual(kvPuts, [[sessionCacheKey, sessionDocument("device-01")]]); }); it("rejects invalid device registration payloads before D1 writes", async () => { diff --git a/cloudflare/tests/devices_test_support.ts b/cloudflare/tests/devices_test_support.ts index 9f79861..fca2acb 100644 --- a/cloudflare/tests/devices_test_support.ts +++ b/cloudflare/tests/devices_test_support.ts @@ -15,6 +15,7 @@ export interface TestEnvOptions { d1?: RecordedD1Database; kvEntries?: [string, string][]; kvDeletes?: string[]; + kvPuts?: [string, string][]; kvReads?: string[]; r2Deletes?: string[]; r2Gets?: string[]; @@ -51,6 +52,11 @@ export function testEnv(options: TestEnvOptions): Env { options.kvReads?.push(key); return Promise.resolve(values.get(key) ?? null); }, + put(key: string, value: string): Promise { + options.kvPuts?.push([key, value]); + values.set(key, value); + return Promise.resolve(); + }, delete(key: string): Promise { options.kvDeletes?.push(key); values.delete(key); @@ -110,12 +116,12 @@ export function testD1Database(rows: unknown[] | TestD1DatabaseOptions): Recorde }; } -export function sessionDocument(deviceId = "device-01"): string { +export function sessionDocument(deviceId: string | null = "device-01"): string { return JSON.stringify({ version: 1, user_id: "user-01", session_id: "session-01", - device_id: deviceId, + ...(deviceId === null ? {} : { device_id: deviceId }), expires_at: "2099-01-01T00:00:00.000Z", }); } diff --git a/cloudflare/tests/index.test.ts b/cloudflare/tests/index.test.ts index 70cd1ff..3570650 100644 --- a/cloudflare/tests/index.test.ts +++ b/cloudflare/tests/index.test.ts @@ -359,6 +359,10 @@ function testEnv( get(key: string): Promise { return Promise.resolve(values.get(key) ?? null); }, + put(key: string, value: string): Promise { + values.set(key, value); + return Promise.resolve(); + }, delete(key: string): Promise { values.delete(key); return Promise.resolve(); diff --git a/crates/ely_app/src/shell/mod.rs b/crates/ely_app/src/shell/mod.rs index 8c73e23..368461c 100644 --- a/crates/ely_app/src/shell/mod.rs +++ b/crates/ely_app/src/shell/mod.rs @@ -277,6 +277,7 @@ impl ElyShell { let mut latest_connection: Option = None; let mut auth_changed = false; let mut trigger_initial_sync = false; + let mut trigger_merged_upload = None; while let Ok(update) = self.sync_inbox_rx.try_recv() { match update { SyncStateUpdate::SignedOut => { @@ -286,6 +287,28 @@ impl ElyShell { latest_connection = Some(ely_domain::SyncConnectionState::AwaitingDeviceApproval); } + SyncStateUpdate::RemoteSnapshot { bytes, logical_clock } => { + if let ShellState::Ready(core) = &mut self.state { + match core.apply_sync_snapshot_bytes(&bytes) { + Ok(summary) => { + tracing::info!( + target: "ely::sync", + imported = summary.imported(), + updated = summary.updated(), + skipped = summary.skipped(), + "remote snapshot applied", + ); + trigger_merged_upload = Some(logical_clock); + } + Err(error) => { + latest_connection = + Some(ely_domain::SyncConnectionState::SyncError { + message: error.to_string(), + }); + } + } + } + } SyncStateUpdate::SyncReady { last_synced_at_secs } => { latest_connection = Some(ely_domain::SyncConnectionState::SyncReady { last_synced_at_secs }); @@ -317,7 +340,11 @@ impl ElyShell { if trigger_initial_sync { self.trigger_cloud_sync_upload(); } - auth_changed || trigger_initial_sync + let merged_upload_requested = trigger_merged_upload.is_some(); + if let Some(logical_clock_floor) = trigger_merged_upload { + self.trigger_cloud_sync_upload_after_remote(logical_clock_floor); + } + auth_changed || trigger_initial_sync || merged_upload_requested } fn focus_command_mode(&mut self, window: &mut Window, cx: &mut Context) { diff --git a/crates/ely_app/src/shell/settings_actions.rs b/crates/ely_app/src/shell/settings_actions.rs index dec1aea..5c9bbd4 100644 --- a/crates/ely_app/src/shell/settings_actions.rs +++ b/crates/ely_app/src/shell/settings_actions.rs @@ -242,12 +242,15 @@ impl ElyShell { } } - /// Push the active profile's bookmarks to `ely-browser-cloud` as a - /// snapshot. The HTTP round-trip runs on a dedicated worker thread - /// (the UI thread never blocks on the network), and the worker - /// reports back through the shell's `sync_inbox` so the sync page - /// reflects the new state without waiting for a manual refresh. pub(crate) fn trigger_cloud_sync_upload(&mut self) { + self.trigger_cloud_sync_upload_with_clock_floor(None); + } + + pub(crate) fn trigger_cloud_sync_upload_after_remote(&mut self, logical_clock_floor: u64) { + self.trigger_cloud_sync_upload_with_clock_floor(Some(logical_clock_floor)); + } + + fn trigger_cloud_sync_upload_with_clock_floor(&mut self, logical_clock_floor: Option) { let ShellState::Ready(core) = &self.state else { return; }; @@ -273,9 +276,13 @@ impl ElyShell { } }; let tx = self.sync_inbox_tx.clone(); + let thread_name = + if logical_clock_floor.is_some() { "ely-sync-merge-upload" } else { "ely-sync-upload" }; std::thread::Builder::new() - .name("ely-sync-upload".to_string()) - .spawn(move || run_sync_upload(profile_dir, device_name, bytes, tx)) + .name(thread_name.to_string()) + .spawn(move || { + run_sync_upload(profile_dir, device_name, bytes, logical_clock_floor, tx) + }) .map(|_| ()) .unwrap_or_else(|error| { tracing::warn!( @@ -317,6 +324,7 @@ fn run_sync_upload( profile_dir: std::path::PathBuf, device_name: String, bytes: Vec, + logical_clock_floor: Option, inbox: std::sync::mpsc::Sender, ) { let mut engine = match SyncEngine::for_profile_dir( @@ -332,11 +340,60 @@ fn run_sync_upload( return; } }; - match engine.upload_bytes(bytes) { + let outcome = match logical_clock_floor { + Some(floor) => engine.upload_merged_bytes(bytes, floor), + None => engine.sync_bytes(bytes), + }; + match outcome { Ok(ely_browser_core::SyncOutcome::SignedOut) => { tracing::info!(target: "ely::sync", "no bearer token on disk; sync skipped"); let _ = inbox.send(SyncStateUpdate::SignedOut); } + Ok(ely_browser_core::SyncOutcome::AwaitingDeviceApproval { device_id }) => { + tracing::info!( + target: "ely::sync", + device_id = %device_id, + "sync device is awaiting approval", + ); + let _ = inbox.send(SyncStateUpdate::AwaitingDeviceApproval); + } + Ok(ely_browser_core::SyncOutcome::RemoteSnapshot { + snapshot_id, + logical_clock, + payload_bytes, + device_id, + bytes, + }) => { + tracing::info!( + target: "ely::sync", + snapshot_id = %snapshot_id, + logical_clock, + payload_bytes, + device_id = %device_id, + "remote snapshot downloaded", + ); + let _ = inbox.send(SyncStateUpdate::RemoteSnapshot { bytes, logical_clock }); + } + Ok(ely_browser_core::SyncOutcome::AlreadyCurrent { + snapshot_id, + logical_clock, + payload_bytes, + device_id, + }) => { + tracing::info!( + target: "ely::sync", + snapshot_id = %snapshot_id, + logical_clock, + payload_bytes, + device_id = %device_id, + "snapshot already current", + ); + let last_synced_at_secs = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + let _ = inbox.send(SyncStateUpdate::SyncReady { last_synced_at_secs }); + } Ok(ely_browser_core::SyncOutcome::Uploaded { snapshot_id, logical_clock, diff --git a/crates/ely_app/src/shell/sync_state.rs b/crates/ely_app/src/shell/sync_state.rs index 9ef8915..e6452c5 100644 --- a/crates/ely_app/src/shell/sync_state.rs +++ b/crates/ely_app/src/shell/sync_state.rs @@ -7,6 +7,7 @@ pub(crate) enum SyncStateUpdate { SignedOut, AwaitingDeviceApproval, + RemoteSnapshot { bytes: Vec, logical_clock: u64 }, SyncReady { last_synced_at_secs: u64 }, SyncError { message: String }, AuthOtpSent { email: String }, diff --git a/crates/ely_browser_core/src/state/sync.rs b/crates/ely_browser_core/src/state/sync.rs index 95c3c25..b3a9f34 100644 --- a/crates/ely_browser_core/src/state/sync.rs +++ b/crates/ely_browser_core/src/state/sync.rs @@ -1,9 +1,13 @@ +use std::time::{Duration, UNIX_EPOCH}; + use ely_domain::{ - SyncConnectionState, SyncObjectKind, SyncObjectPolicy, SyncObjectState, SyncObjectStatus, - SyncStatus, + BookmarkEntry, BookmarkId, ProfileId, SpaceId, SyncConnectionState, SyncObjectKind, + SyncObjectPolicy, SyncObjectState, SyncObjectStatus, SyncStatus, UrlText, }; +use ely_sync_client::SyncClientError; use super::BrowserCore; +use crate::sync_engine::{BookmarkSyncRecord, SyncSnapshotApplySummary, SyncSnapshotBody}; #[derive(Clone, Debug)] pub(super) struct SyncObjectPolicies { @@ -82,6 +86,24 @@ impl BrowserCore { self.sync_connection_state = state; } + pub(crate) fn sync_space_name_for(&self, space_id: &SpaceId) -> Option { + self.spaces + .iter() + .find(|space| space.id() == space_id) + .map(|space| space.name().to_string()) + } + + pub(crate) fn apply_sync_snapshot_body( + &mut self, + body: SyncSnapshotBody, + ) -> Result { + let mut summary = SyncSnapshotApplySummary::default(); + for record in body.bookmarks { + self.apply_bookmark_sync_record(record, &mut summary)?; + } + Ok(summary) + } + pub(super) fn sync_status(&self) -> SyncStatus { let enabled_state = match &self.sync_connection_state { SyncConnectionState::SyncReady { .. } => SyncObjectState::Synced, @@ -149,4 +171,94 @@ impl BrowserCore { fn sync_enabled_tab_count(&self) -> usize { self.tabs.iter().filter(|tab| tab.sync_enabled()).count() } + + fn apply_bookmark_sync_record( + &mut self, + record: BookmarkSyncRecord, + summary: &mut SyncSnapshotApplySummary, + ) -> Result<(), SyncClientError> { + let bookmark_id = parse_bookmark_id(&record.id)?; + let profile_id = self.sync_profile_id(&record.profile_id)?; + let space_id = self.sync_space_id(&record.space_id, record.space_name.as_deref())?; + let url = UrlText::parse(&record.url).map_err(snapshot_schema_error)?; + let added_at = UNIX_EPOCH + Duration::from_secs(record.added_at_secs); + let existing_index = + self.bookmarks.iter().position(|bookmark| bookmark.id() == &bookmark_id).or_else( + || { + self.bookmarks.iter().position(|bookmark| { + bookmark.profile_id() == &profile_id + && bookmark.space_id() == &space_id + && bookmark.url() == &url + }) + }, + ); + + let id = existing_index + .and_then(|index| self.bookmarks.get(index).map(|bookmark| bookmark.id().clone())) + .unwrap_or(bookmark_id); + let mut bookmark = BookmarkEntry::restore( + id, + profile_id, + space_id, + record.collection_name, + record.title, + url, + added_at, + ) + .map_err(snapshot_schema_error)?; + bookmark.set_tags(record.tags).map_err(snapshot_schema_error)?; + if let Some(note) = record.note { + bookmark.set_note(note).map_err(snapshot_schema_error)?; + } + if let Some(thumbnail_key) = record.thumbnail_key { + bookmark.set_thumbnail_key(thumbnail_key).map_err(snapshot_schema_error)?; + } + + match existing_index { + Some(index) if self.bookmarks[index] == bookmark => summary.record_skipped(), + Some(index) => { + self.bookmarks[index] = bookmark; + summary.record_updated(); + } + None => { + self.bookmarks.push(bookmark); + summary.record_imported(); + } + } + Ok(()) + } + + fn sync_profile_id(&self, raw: &str) -> Result { + let profile_id = ProfileId::parse(raw).map_err(snapshot_schema_error)?; + if self.profiles.iter().any(|profile| profile.id() == &profile_id) { + return Ok(profile_id); + } + Ok(self.active_profile_id.clone()) + } + + fn sync_space_id( + &self, + raw: &str, + space_name: Option<&str>, + ) -> Result { + let space_id = SpaceId::parse(raw).map_err(snapshot_schema_error)?; + if self.spaces.iter().any(|space| space.id() == &space_id) { + return Ok(space_id); + } + if let Some(space_name) = space_name + && let Some(space) = + self.spaces.iter().find(|space| space.name().eq_ignore_ascii_case(space_name)) + { + return Ok(space.id().clone()); + } + Ok(self.active_space_id.clone()) + } +} + +fn parse_bookmark_id(raw: &str) -> Result { + BookmarkId::parse(raw).map_err(snapshot_schema_error) +} + +fn snapshot_schema_error(error: impl ToString) -> SyncClientError { + SyncClientError::SnapshotSchema(error.to_string()) } diff --git a/crates/ely_browser_core/src/sync_engine.rs b/crates/ely_browser_core/src/sync_engine.rs index a0fc558..6c952e8 100644 --- a/crates/ely_browser_core/src/sync_engine.rs +++ b/crates/ely_browser_core/src/sync_engine.rs @@ -6,7 +6,7 @@ use std::{ use ely_domain::BookmarkEntry; use ely_sync_client::{ ApiClientConfig, BearerToken, BearerTokenStore, DeviceIdentity, SnapshotPayload, - SnapshotUploadRequest, SyncApiClient, SyncClientError, + SnapshotUploadRequest, SyncApiClient, SyncClientError, SyncLatestSnapshotDocument, }; use serde::{Deserialize, Serialize}; @@ -75,22 +75,99 @@ impl SyncEngine { self.bearer_store.load().map(|token| token.is_some()) } - /// Ship a pre-serialised snapshot payload to the worker. Callers - /// usually pair this with `BrowserCore::build_sync_snapshot_bytes` - /// — building the bytes on the UI thread and only crossing the - /// thread boundary with `Vec` keeps `BrowserCore` itself - /// single-threaded. - pub fn upload_bytes(&mut self, bytes: Vec) -> Result { + /// Run the snapshot sync plan for a pre-serialised local payload. + /// The engine registers the device, checks the worker's latest + /// snapshot, downloads a newer remote payload when another device + /// wrote one, and uploads when the local payload is ready to win. + pub fn sync_bytes(&mut self, bytes: Vec) -> Result { let Some(bearer) = self.bearer_store.load()? else { let outcome = SyncOutcome::SignedOut; self.last_outcome = Some(outcome.clone()); return Ok(outcome); }; let payload = SnapshotPayload::new(bytes)?; - let logical_clock = current_logical_clock(); - let snapshot_id = snapshot_id_for_user(&self.identity); - let client = SyncApiClient::new(self.api_config.clone(), bearer)?; + let Some(client) = self.approved_client(client)? else { + let outcome = + SyncOutcome::AwaitingDeviceApproval { device_id: self.identity.device_id.clone() }; + self.last_outcome = Some(outcome.clone()); + return Ok(outcome); + }; + + let status = client.sync_status()?; + let outcome = match status.snapshots.latest { + Some(latest) if latest.payload_hash == payload.payload_hash() => { + SyncOutcome::AlreadyCurrent { + snapshot_id: latest.snapshot_id, + logical_clock: latest.logical_clock, + payload_bytes: latest.size_bytes, + device_id: latest.device_id, + } + } + Some(latest) if latest.device_id != self.identity.device_id => { + self.download_remote_snapshot(&client, latest)? + } + Some(latest) => self.upload_payload(&client, payload, latest.logical_clock)?, + None => self.upload_payload(&client, payload, 0)?, + }; + self.last_outcome = Some(outcome.clone()); + Ok(outcome) + } + + /// Upload a local payload after the UI thread has applied a remote + /// snapshot. The caller passes the remote logical clock so the new + /// merged snapshot is ordered after the downloaded one. + pub fn upload_merged_bytes( + &mut self, + bytes: Vec, + logical_clock_floor: u64, + ) -> Result { + let Some(bearer) = self.bearer_store.load()? else { + let outcome = SyncOutcome::SignedOut; + self.last_outcome = Some(outcome.clone()); + return Ok(outcome); + }; + let payload = SnapshotPayload::new(bytes)?; + let client = SyncApiClient::new(self.api_config.clone(), bearer)?; + let Some(client) = self.approved_client(client)? else { + let outcome = + SyncOutcome::AwaitingDeviceApproval { device_id: self.identity.device_id.clone() }; + self.last_outcome = Some(outcome.clone()); + return Ok(outcome); + }; + let outcome = self.upload_payload(&client, payload, logical_clock_floor)?; + self.last_outcome = Some(outcome.clone()); + Ok(outcome) + } + + fn approved_client( + &self, + client: SyncApiClient, + ) -> Result, SyncClientError> { + let registration = client.register_device( + &self.identity, + &device_registration_idempotency_key(&self.identity), + )?; + if registration.device.is_approved() { + return Ok(Some(client)); + } + if registration.device.approval_status == "pending" { + return Ok(None); + } + Err(SyncClientError::DeviceApprovalStatus { + device_id: registration.device.device_id, + status: registration.device.approval_status, + }) + } + + fn upload_payload( + &self, + client: &SyncApiClient, + payload: SnapshotPayload, + logical_clock_floor: u64, + ) -> Result { + let logical_clock = current_logical_clock().max(logical_clock_floor.saturating_add(1)); + let snapshot_id = snapshot_id_for_user(&self.identity); let request = SnapshotUploadRequest::new( &snapshot_id, self.api_config.region(), @@ -99,21 +176,92 @@ impl SyncEngine { &payload, ); let document = client.upload_snapshot(&request)?; - let outcome = SyncOutcome::Uploaded { + Ok(SyncOutcome::Uploaded { snapshot_id: document.snapshot.snapshot_id, logical_clock: document.snapshot.logical_clock, payload_bytes: document.snapshot.size_bytes, device_id: document.device_id, - }; - self.last_outcome = Some(outcome.clone()); - Ok(outcome) + }) + } + + fn download_remote_snapshot( + &self, + client: &SyncApiClient, + latest: SyncLatestSnapshotDocument, + ) -> Result { + let download = client.download_snapshot(&latest.snapshot_id)?; + let payload = download.payload()?; + Ok(SyncOutcome::RemoteSnapshot { + snapshot_id: latest.snapshot_id, + logical_clock: latest.logical_clock, + payload_bytes: latest.size_bytes, + device_id: latest.device_id, + bytes: payload.into_bytes(), + }) } } #[derive(Clone, Debug, Eq, PartialEq)] pub enum SyncOutcome { SignedOut, - Uploaded { snapshot_id: String, logical_clock: u64, payload_bytes: u64, device_id: String }, + AwaitingDeviceApproval { + device_id: String, + }, + AlreadyCurrent { + snapshot_id: String, + logical_clock: u64, + payload_bytes: u64, + device_id: String, + }, + RemoteSnapshot { + snapshot_id: String, + logical_clock: u64, + payload_bytes: u64, + device_id: String, + bytes: Vec, + }, + Uploaded { + snapshot_id: String, + logical_clock: u64, + payload_bytes: u64, + device_id: String, + }, +} + +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct SyncSnapshotApplySummary { + imported: usize, + updated: usize, + skipped: usize, +} + +impl SyncSnapshotApplySummary { + #[must_use] + pub fn imported(self) -> usize { + self.imported + } + + #[must_use] + pub fn updated(self) -> usize { + self.updated + } + + #[must_use] + pub fn skipped(self) -> usize { + self.skipped + } + + pub(crate) fn record_imported(&mut self) { + self.imported += 1; + } + + pub(crate) fn record_updated(&mut self) { + self.updated += 1; + } + + pub(crate) fn record_skipped(&mut self) { + self.skipped += 1; + } } const SNAPSHOT_SCHEMA_REV: u32 = 1; @@ -123,18 +271,17 @@ fn current_logical_clock() -> u64 { } fn snapshot_id_for_user(identity: &DeviceIdentity) -> String { - // The Cloudflare worker requires `^[a-z0-9][a-z0-9._-]{0,127}$`. - // The device ID already satisfies the pattern (lowercased prefix - // + UUIDv7 simple form) and is per-user-stable, so we reuse it as - // the snapshot id. Future work can extend this to per-object-type - // snapshots without disturbing the existing one. identity.device_id.clone() } +fn device_registration_idempotency_key(identity: &DeviceIdentity) -> String { + format!("device-register:{}", identity.device_id) +} + #[derive(Serialize, Deserialize)] -struct SyncSnapshotBody { - schema_rev: u32, - bookmarks: Vec, +pub(crate) struct SyncSnapshotBody { + pub(crate) schema_rev: u32, + pub(crate) bookmarks: Vec, } impl SyncSnapshotBody { @@ -144,7 +291,12 @@ impl SyncSnapshotBody { bookmarks: core .visible_bookmarks_for_sync() .into_iter() - .map(BookmarkSyncRecord::from) + .map(|entry| { + BookmarkSyncRecord::from_entry( + entry, + core.sync_space_name_for(entry.space_id()), + ) + }) .collect(), } } @@ -153,30 +305,36 @@ impl SyncSnapshotBody { /// Wire representation of a bookmark. We keep this struct stable so a /// future deserializer can read snapshots written by earlier app /// versions; new fields must default-fill on read. -#[derive(Serialize, Deserialize)] -struct BookmarkSyncRecord { - id: String, - title: String, - url: String, - profile_id: String, - space_id: String, - collection_name: String, - tags: Vec, - note: Option, - added_at_secs: u64, +#[derive(Clone, Debug, Serialize, Deserialize)] +pub(crate) struct BookmarkSyncRecord { + pub(crate) id: String, + pub(crate) title: String, + pub(crate) url: String, + pub(crate) profile_id: String, + pub(crate) space_id: String, + #[serde(default)] + pub(crate) space_name: Option, + pub(crate) collection_name: String, + pub(crate) tags: Vec, + pub(crate) note: Option, + #[serde(default)] + pub(crate) thumbnail_key: Option, + pub(crate) added_at_secs: u64, } -impl From<&BookmarkEntry> for BookmarkSyncRecord { - fn from(entry: &BookmarkEntry) -> Self { +impl BookmarkSyncRecord { + fn from_entry(entry: &BookmarkEntry, space_name: Option) -> Self { Self { id: entry.id().as_str().to_string(), title: entry.title().to_string(), url: entry.url().as_str().to_string(), profile_id: entry.profile_id().as_str().to_string(), space_id: entry.space_id().as_str().to_string(), + space_name, collection_name: entry.collection_name().to_string(), tags: entry.tags().to_vec(), note: entry.note().map(str::to_string), + thumbnail_key: entry.thumbnail_key().map(str::to_string), added_at_secs: entry .added_at() .duration_since(UNIX_EPOCH) @@ -211,4 +369,20 @@ impl BrowserCore { source: error, }) } + + pub fn apply_sync_snapshot_bytes( + &mut self, + bytes: &[u8], + ) -> Result { + let body: SyncSnapshotBody = serde_json::from_slice(bytes).map_err(|error| { + SyncClientError::Json { endpoint: "snapshot".to_string(), source: error } + })?; + if body.schema_rev != SNAPSHOT_SCHEMA_REV { + return Err(SyncClientError::SnapshotSchema(format!( + "unsupported schema_rev {}", + body.schema_rev + ))); + } + self.apply_sync_snapshot_body(body) + } } diff --git a/crates/ely_browser_core/tests/sync.rs b/crates/ely_browser_core/tests/sync.rs index b676f15..aa66aa4 100644 --- a/crates/ely_browser_core/tests/sync.rs +++ b/crates/ely_browser_core/tests/sync.rs @@ -86,3 +86,72 @@ fn tab_sync_status_counts_sync_enabled_tabs() -> Result<(), Box> { assert_eq!(tabs_status.local_count(), 1); Ok(()) } + +#[test] +fn sync_snapshot_imports_remote_bookmarks_into_active_scope() -> Result<(), Box> { + let mut source = BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?; + source.open_tab(UrlText::parse("https://example.com/research")?); + let bookmark_id = source.bookmark_active_tab()?; + source.set_bookmark_collection_name(&bookmark_id, "Research")?; + source.set_bookmark_tags(&bookmark_id, vec!["rust".to_string(), "gpui".to_string()])?; + source.set_bookmark_note(&bookmark_id, "Read later")?; + let bytes = source.build_sync_snapshot_bytes()?; + + let mut target = BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?; + let target_profile_id = target.snapshot()?.active_profile_id; + let target_space_id = target.snapshot()?.active_space_id; + let summary = target.apply_sync_snapshot_bytes(&bytes)?; + let snapshot = target.snapshot()?; + + assert_eq!(summary.imported(), 1); + assert_eq!(summary.updated(), 0); + assert_eq!(summary.skipped(), 0); + assert_eq!(snapshot.bookmarks.len(), 1); + assert_eq!(snapshot.bookmarks[0].profile_id(), &target_profile_id); + assert_eq!(snapshot.bookmarks[0].space_id(), &target_space_id); + assert_eq!(snapshot.bookmarks[0].collection_name(), "Research"); + assert_eq!(snapshot.bookmarks[0].tags(), &["rust".to_string(), "gpui".to_string()]); + assert_eq!(snapshot.bookmarks[0].note(), Some("Read later")); + Ok(()) +} + +#[test] +fn sync_snapshot_updates_existing_bookmark_metadata() -> Result<(), Box> { + let mut source = BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?; + source.open_tab(UrlText::parse("https://example.com/research")?); + let source_bookmark_id = source.bookmark_active_tab()?; + source.set_bookmark_collection_name(&source_bookmark_id, "Research")?; + source.set_bookmark_tags(&source_bookmark_id, vec!["servo".to_string()])?; + source.set_bookmark_note(&source_bookmark_id, "Canonical")?; + let bytes = source.build_sync_snapshot_bytes()?; + + let mut target = BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?; + target.open_tab(UrlText::parse("https://example.com/research")?); + let target_bookmark_id = target.bookmark_active_tab()?; + target.set_bookmark_collection_name(&target_bookmark_id, "Inbox")?; + let summary = target.apply_sync_snapshot_bytes(&bytes)?; + let snapshot = target.snapshot()?; + + assert_eq!(summary.imported(), 0); + assert_eq!(summary.updated(), 1); + assert_eq!(summary.skipped(), 0); + assert_eq!(snapshot.bookmarks.len(), 1); + assert_eq!(snapshot.bookmarks[0].id(), &target_bookmark_id); + assert_eq!(snapshot.bookmarks[0].collection_name(), "Research"); + assert_eq!(snapshot.bookmarks[0].tags(), &["servo".to_string()]); + assert_eq!(snapshot.bookmarks[0].note(), Some("Canonical")); + Ok(()) +} + +#[test] +fn sync_snapshot_rejects_unknown_schema_rev() -> Result<(), Box> { + let mut core = BrowserCore::new(InitialBrowserConfig::ely_defaults()?)?; + let bytes = br#"{"schema_rev":999,"bookmarks":[]}"#; + + let Err(error) = core.apply_sync_snapshot_bytes(bytes) else { + return Err("expected sync snapshot schema error".into()); + }; + + assert!(error.to_string().contains("unsupported schema_rev 999")); + Ok(()) +} diff --git a/crates/ely_domain/src/appearance.rs b/crates/ely_domain/src/appearance.rs index b2fc659..d6cbef5 100644 --- a/crates/ely_domain/src/appearance.rs +++ b/crates/ely_domain/src/appearance.rs @@ -111,19 +111,20 @@ mod tests { } #[test] - fn json_round_trip_preserves_kebab_case_variants() { + fn json_round_trip_preserves_kebab_case_variants() -> Result<(), serde_json::Error> { let mut settings = AppearanceSettings::default(); settings.set_wallpaper(WallpaperTheme::Mint); settings.set_theme_mode(ThemeMode::Light); settings.set_reduce_motion(true); settings.set_translucency_pct(60); - let json = serde_json::to_string(&settings).unwrap(); + let json = serde_json::to_string(&settings)?; assert!(json.contains("\"wallpaper\":\"mint\"")); assert!(json.contains("\"theme_mode\":\"light\"")); assert!(json.contains("\"translucency_pct\":60")); - let restored: AppearanceSettings = serde_json::from_str(&json).unwrap(); + let restored: AppearanceSettings = serde_json::from_str(&json)?; assert_eq!(restored, settings); + Ok(()) } } diff --git a/crates/ely_domain/src/bookmark.rs b/crates/ely_domain/src/bookmark.rs index 7bbd799..1b38f98 100644 --- a/crates/ely_domain/src/bookmark.rs +++ b/crates/ely_domain/src/bookmark.rs @@ -42,6 +42,32 @@ impl BookmarkEntry { }) } + pub fn restore( + id: BookmarkId, + profile_id: ProfileId, + space_id: SpaceId, + collection_name: impl Into, + title: impl Into, + url: UrlText, + added_at: SystemTime, + ) -> Result { + let collection_name = non_empty_text("bookmark collection", collection_name.into())?; + let title = non_empty_text("bookmark title", title.into())?; + + Ok(Self { + id, + profile_id, + space_id, + collection_name, + title, + url, + tags: Vec::new(), + note: None, + thumbnail_key: None, + added_at, + }) + } + #[must_use] pub fn id(&self) -> &BookmarkId { &self.id diff --git a/crates/ely_sync_client/Cargo.toml b/crates/ely_sync_client/Cargo.toml index 5be8b37..4dd1492 100644 --- a/crates/ely_sync_client/Cargo.toml +++ b/crates/ely_sync_client/Cargo.toml @@ -6,6 +6,7 @@ license.workspace = true rust-version.workspace = true [dependencies] +ed25519-dalek.workspace = true ely_domain = { path = "../ely_domain" } serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/ely_sync_client/src/client.rs b/crates/ely_sync_client/src/client.rs index 835f902..aa3d08e 100644 --- a/crates/ely_sync_client/src/client.rs +++ b/crates/ely_sync_client/src/client.rs @@ -106,6 +106,19 @@ impl SyncApiClient { read_json_response::(&endpoint, response) } + /// `GET /api/sync/status` — return the worker-side cursor, + /// object, snapshot, and device summary for the authenticated + /// approved device. + pub fn sync_status(&self) -> Result { + let endpoint = self.endpoint("/api/sync/status"); + let response = self + .agent + .get(&endpoint) + .set("Authorization", &format!("Bearer {}", self.bearer.as_str())) + .call(); + read_json_response::(&endpoint, response) + } + /// `POST /api/sync/snapshot` — push the full per-user state. The /// worker enforces logical-clock monotonicity, so callers must /// pass a value strictly greater than the last accepted snapshot. @@ -162,6 +175,55 @@ pub struct SnapshotUploadDocument { pub snapshot: crate::snapshot::SnapshotDocument, } +#[derive(Clone, Debug, serde::Deserialize)] +pub struct SyncStatusDocument { + pub version: u32, + pub user_id: String, + pub device_id: String, + pub cursor: SyncCursorStatusDocument, + pub objects: Vec, + pub snapshots: SyncSnapshotStatusDocument, + pub devices: SyncDeviceStatusDocument, +} + +#[derive(Clone, Debug, serde::Deserialize)] +pub struct SyncCursorStatusDocument { + pub latest_change_id: u64, + pub total_changes: u64, +} + +#[derive(Clone, Debug, serde::Deserialize)] +pub struct SyncObjectStatusDocument { + pub object_type: String, + pub active_count: u64, + pub deleted_count: u64, + pub latest_logical_clock: u64, + pub latest_updated_at: u64, +} + +#[derive(Clone, Debug, serde::Deserialize)] +pub struct SyncSnapshotStatusDocument { + pub total_snapshots: u64, + pub latest: Option, +} + +#[derive(Clone, Debug, serde::Deserialize)] +pub struct SyncLatestSnapshotDocument { + pub snapshot_id: String, + pub payload_hash: String, + pub logical_clock: u64, + pub device_id: String, + pub size_bytes: u64, + pub created_at: u64, +} + +#[derive(Clone, Debug, serde::Deserialize)] +pub struct SyncDeviceStatusDocument { + pub approved_count: u64, + pub current_device_id: String, + pub current_device_approved: bool, +} + fn read_json_response( endpoint: &str, response: Result, diff --git a/crates/ely_sync_client/src/device.rs b/crates/ely_sync_client/src/device.rs index b363697..b48c2f8 100644 --- a/crates/ely_sync_client/src/device.rs +++ b/crates/ely_sync_client/src/device.rs @@ -4,6 +4,7 @@ use std::{ path::Path, }; +use ed25519_dalek::SigningKey; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -51,10 +52,7 @@ impl DeviceIdentity { pub fn generate(device_name: impl Into, platform: impl Into) -> Self { let device_id = format!("ely-{}", Uuid::now_v7().simple()); - // Placeholder public key — Ed25519 device-bound signing is a - // backend feature still in design. The worker validates the - // shape but does not currently challenge it. - let public_key = format!("ed25519:{}", Uuid::now_v7().simple()); + let public_key = public_key_hex(); Self { device_id, public_key, device_name: device_name.into(), platform: platform.into() } } @@ -102,6 +100,18 @@ fn io_err(error: io::Error) -> SyncClientError { SyncClientError::TokenStorage(error.to_string()) } +fn public_key_hex() -> String { + let mut seed = [0_u8; 32]; + seed[..16].copy_from_slice(Uuid::now_v7().as_bytes()); + seed[16..].copy_from_slice(Uuid::now_v7().as_bytes()); + let signing_key = SigningKey::from_bytes(&seed); + hex_string(&signing_key.verifying_key().to_bytes()) +} + +fn hex_string(bytes: &[u8]) -> String { + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} + #[derive(Clone, Debug, Serialize)] pub struct DeviceRegistration<'a> { pub device_id: &'a str, @@ -147,6 +157,8 @@ mod tests { let path = dir.join("device.json"); let identity = DeviceIdentity::load_or_create(&path, "Test", "macos")?; identity.validate()?; + assert_eq!(identity.public_key.len(), 64); + assert!(identity.public_key.as_bytes().iter().all(u8::is_ascii_hexdigit)); let again = DeviceIdentity::load_or_create(&path, "ignored", "ignored")?; assert_eq!(identity, again); diff --git a/crates/ely_sync_client/src/error.rs b/crates/ely_sync_client/src/error.rs index 3885884..585bdd4 100644 --- a/crates/ely_sync_client/src/error.rs +++ b/crates/ely_sync_client/src/error.rs @@ -30,4 +30,10 @@ pub enum SyncClientError { #[error("Snapshot base64 decode failed: {0}")] SnapshotBase64(String), + + #[error("Snapshot schema is invalid: {0}")] + SnapshotSchema(String), + + #[error("Device {device_id} cannot sync with approval status {status}")] + DeviceApprovalStatus { device_id: String, status: String }, } diff --git a/crates/ely_sync_client/src/lib.rs b/crates/ely_sync_client/src/lib.rs index 0f5bad4..478552d 100644 --- a/crates/ely_sync_client/src/lib.rs +++ b/crates/ely_sync_client/src/lib.rs @@ -10,19 +10,11 @@ //! - Bearer-token authenticated requests via `ureq`. //! - Device registration (`POST /api/devices/register`) and listing //! (`GET /api/devices`). -//! - Sync snapshot upload (`POST /api/sync/snapshot`) and download +//! - Sync status (`GET /api/sync/status`), snapshot upload +//! (`POST /api/sync/snapshot`), and snapshot download //! (`GET /api/sync/snapshot?snapshot_id=…`). -//! -//! Intentionally omitted (kept for follow-up work, not papered over here): //! - The full Better Auth handshake (email + OTP / OAuth). Callers obtain //! the bearer token out-of-band and hand it to the client. -//! - First-device approval bootstrap. The Cloudflare API rejects sync from -//! an unapproved device; the user must approve a freshly-registered -//! device from another already-approved device (or via direct D1 -//! operation), exactly as the backend enforces. -//! - Incremental change-log push/pull (`/api/sync/push` and `/api/sync/pull`). -//! The snapshot path is the simplest contract that round-trips the user's -//! entire state, so we start there. pub mod auth; pub mod client; @@ -32,7 +24,7 @@ pub mod error; pub mod snapshot; pub use auth::{BearerToken, BearerTokenStore}; -pub use client::{ApiClientConfig, SyncApiClient}; +pub use client::{ApiClientConfig, SyncApiClient, SyncLatestSnapshotDocument, SyncStatusDocument}; pub use device::{DeviceIdentity, DeviceListResponse, DeviceRecord, DeviceRegistration}; pub use email_otp::{send_email_otp, verify_email_otp}; pub use error::SyncClientError; diff --git a/crates/ely_sync_client/src/snapshot.rs b/crates/ely_sync_client/src/snapshot.rs index e008665..5df5d31 100644 --- a/crates/ely_sync_client/src/snapshot.rs +++ b/crates/ely_sync_client/src/snapshot.rs @@ -44,6 +44,10 @@ impl SnapshotPayload { pub fn payload_hash(&self) -> &str { &self.payload_hash } + + pub fn into_bytes(self) -> Vec { + self.bytes + } } #[derive(Clone, Debug, Serialize)]