From c5973335a059451f680390d2d70628da7708efeb Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 6 Jul 2026 10:05:33 +0100 Subject: [PATCH 01/16] Add identity_adopted column to user_config Claude-Session: https://claude.ai/code/session_01VpgSNzaVPefKfHKjERxTSQ --- .../migrations/011_add_identity_adopted.sql | 1 + replicant-client/src/database.rs | 24 +++++++++++++++++++ 2 files changed, 25 insertions(+) create mode 100644 replicant-client/migrations/011_add_identity_adopted.sql diff --git a/replicant-client/migrations/011_add_identity_adopted.sql b/replicant-client/migrations/011_add_identity_adopted.sql new file mode 100644 index 0000000..cd1b79a --- /dev/null +++ b/replicant-client/migrations/011_add_identity_adopted.sql @@ -0,0 +1 @@ +ALTER TABLE user_config ADD COLUMN identity_adopted INTEGER NOT NULL DEFAULT 0; diff --git a/replicant-client/src/database.rs b/replicant-client/src/database.rs index 523db01..5bb4a7b 100644 --- a/replicant-client/src/database.rs +++ b/replicant-client/src/database.rs @@ -600,3 +600,27 @@ mod identity_freeze_tests { ); } } + +#[cfg(test)] +mod identity_tests { + use super::*; + + async fn fresh_db() -> ClientDatabase { + let db = ClientDatabase::new(":memory:").await.unwrap(); + db.run_migrations().await.unwrap(); + db + } + + #[tokio::test] + async fn user_config_has_identity_adopted_defaulting_to_zero() { + let db = fresh_db().await; + db.ensure_user_config("ws://localhost/ws").await.unwrap(); + + let row = sqlx::query("SELECT identity_adopted FROM user_config LIMIT 1") + .fetch_one(&db.pool) + .await + .unwrap(); + let adopted: i64 = row.try_get("identity_adopted").unwrap(); + assert_eq!(adopted, 0); + } +} From 70ed678fc69c9dd58d56059ea44a26846d139670 Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 6 Jul 2026 10:09:42 +0100 Subject: [PATCH 02/16] Generate a random provisional user id; remove email derivation Claude-Session: https://claude.ai/code/session_01VpgSNzaVPefKfHKjERxTSQ --- .../examples/task_list_example.rs | 8 +-- replicant-client/src/database.rs | 64 +++++++------------ .../tests/phoenix_integration/mod.rs | 16 +---- 3 files changed, 27 insertions(+), 61 deletions(-) diff --git a/replicant-client/examples/task_list_example.rs b/replicant-client/examples/task_list_example.rs index 44ab8fe..02999ad 100644 --- a/replicant-client/examples/task_list_example.rs +++ b/replicant-client/examples/task_list_example.rs @@ -451,12 +451,8 @@ async fn main() -> Result<(), Box> { let user_id = match db.get_user_id().await { Ok(id) => id, Err(_) => { - // Deterministic user ID from the identifier, or random for anonymous use - let id = if let Some(user_identifier) = &cli.user { - ClientDatabase::generate_deterministic_user_id(user_identifier) - } else { - Uuid::new_v4() - }; + // Provisional random id; the server assigns a canonical id on first contact. + let id = Uuid::new_v4(); // Client ID should always be unique per client instance let client_id = Uuid::new_v4(); diff --git a/replicant-client/src/database.rs b/replicant-client/src/database.rs index 5bb4a7b..7c4f334 100644 --- a/replicant-client/src/database.rs +++ b/replicant-client/src/database.rs @@ -62,7 +62,7 @@ impl ClientDatabase { pub async fn ensure_user_config_with_identifier( &self, server_url: &str, - user_identifier: &str, + _user_identifier: &str, ) -> SyncResult<()> { // Check if user_config already exists let exists = sqlx::query("SELECT COUNT(*) as count FROM user_config") @@ -72,12 +72,12 @@ impl ClientDatabase { let count: i64 = exists.try_get("count")?; if count == 0 { - // No user config exists, create with deterministic user ID - let user_id = Self::generate_deterministic_user_id(user_identifier); + // Provisional random id; the server's canonical id is adopted on first contact. + let user_id = Uuid::new_v4(); let client_id = Uuid::new_v4(); // Client ID should always be unique per instance sqlx::query( - "INSERT INTO user_config (user_id, client_id, server_url) VALUES (?1, ?2, ?3)", + "INSERT INTO user_config (user_id, client_id, server_url, identity_adopted) VALUES (?1, ?2, ?3, 0)", ) .bind(user_id.to_string()) .bind(client_id.to_string()) @@ -89,17 +89,6 @@ impl ClientDatabase { Ok(()) } - /// FROZEN — must match replicant-server Auth (namespace + email - /// normalization). The crate's only identity derivation; call this - /// instead of re-implementing it. - pub fn generate_deterministic_user_id(user_identifier: &str) -> Uuid { - const APP_ID: &str = "com.nodeaudio.entonal"; - - let normalized = user_identifier.trim().to_lowercase(); - let app_namespace = Uuid::new_v5(&Uuid::NAMESPACE_DNS, APP_ID.as_bytes()); - Uuid::new_v5(&app_namespace, normalized.as_bytes()) - } - pub async fn get_user_id(&self) -> SyncResult { let row = sqlx::query(Queries::GET_USER_ID) .fetch_one(&self.pool) @@ -576,31 +565,6 @@ impl ClientDatabase { } } -#[cfg(test)] -mod identity_freeze_tests { - use super::*; - - #[test] - fn deterministic_user_id_matches_frozen_vectors() { - assert_eq!( - ClientDatabase::generate_deterministic_user_id("test@example.com").to_string(), - "71b2b712-7878-56ee-8323-43809b8198a5" - ); - assert_eq!( - ClientDatabase::generate_deterministic_user_id("alice@example.com").to_string(), - "af665bed-e8e7-5b1f-ba4f-9343fefde4bb" - ); - } - - #[test] - fn normalization_makes_case_and_whitespace_irrelevant() { - assert_eq!( - ClientDatabase::generate_deterministic_user_id(" Alice@Example.COM ").to_string(), - "af665bed-e8e7-5b1f-ba4f-9343fefde4bb" - ); - } -} - #[cfg(test)] mod identity_tests { use super::*; @@ -623,4 +587,24 @@ mod identity_tests { let adopted: i64 = row.try_get("identity_adopted").unwrap(); assert_eq!(adopted, 0); } + + #[tokio::test] + async fn ensure_user_config_with_identifier_generates_random_v4_id() { + let db = fresh_db().await; + db.ensure_user_config_with_identifier("ws://localhost/ws", "test@example.com") + .await + .unwrap(); + + let user_id = db.get_user_id().await.unwrap(); + // No longer derived from the email. + assert_ne!(user_id.to_string(), "71b2b712-7878-56ee-8323-43809b8198a5"); + assert_eq!(user_id.get_version(), Some(uuid::Version::Random)); + + let row = sqlx::query("SELECT identity_adopted FROM user_config LIMIT 1") + .fetch_one(&db.pool) + .await + .unwrap(); + let adopted: i64 = row.try_get("identity_adopted").unwrap(); + assert_eq!(adopted, 0); + } } diff --git a/replicant-client/tests/phoenix_integration/mod.rs b/replicant-client/tests/phoenix_integration/mod.rs index 1c00ae9..84b5a81 100644 --- a/replicant-client/tests/phoenix_integration/mod.rs +++ b/replicant-client/tests/phoenix_integration/mod.rs @@ -63,20 +63,6 @@ pub fn skip_if_no_server() -> bool { std::env::var("RUN_INTEGRATION_TESTS").is_err() } -/// Deterministic user ID for an email — the client's frozen derivation, -/// which matches the server's Auth.deterministic_user_id/1. -pub fn deterministic_user_id(email: &str) -> Uuid { - replicant_client::ClientDatabase::generate_deterministic_user_id(email) -} - -#[test] -fn deterministic_user_id_normalizes_like_client_and_server() { - assert_eq!( - deterministic_user_id(" Integration-Test@Example.COM "), - deterministic_user_id(TEST_EMAIL) - ); -} - /// A broadcast event received from the server #[derive(Debug)] pub struct BroadcastEvent { @@ -104,7 +90,7 @@ impl TestClient { api_secret: &str, ) -> Result { let url = Url::parse(&server_url()).map_err(|e| format!("Invalid URL: {}", e))?; - let user_id = deterministic_user_id(email); + let user_id = Uuid::new_v4(); let socket = Socket::spawn(url, None, None) .await From 8fcae661e7b9c8c1fb82206e41cbc63d12583abd Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 6 Jul 2026 10:11:30 +0100 Subject: [PATCH 03/16] Add atomic identity-adoption transaction (re-stamp docs + flip flag) Claude-Session: https://claude.ai/code/session_01VpgSNzaVPefKfHKjERxTSQ --- replicant-client/src/database.rs | 101 +++++++++++++++++++++++++++++++ replicant-client/src/queries.rs | 6 ++ 2 files changed, 107 insertions(+) diff --git a/replicant-client/src/database.rs b/replicant-client/src/database.rs index 7c4f334..995a53c 100644 --- a/replicant-client/src/database.rs +++ b/replicant-client/src/database.rs @@ -117,6 +117,27 @@ impl ClientDatabase { Ok((Uuid::parse_str(&user_id)?, Uuid::parse_str(&client_id)?)) } + /// Atomically adopt the server's canonical id: re-stamp local documents + /// owned by `old_id` and flip `user_config` to `canonical_id` with + /// `identity_adopted = 1`. A crash mid-adoption leaves the old id intact. + pub async fn adopt_identity(&self, old_id: Uuid, canonical_id: Uuid) -> SyncResult<()> { + let mut tx = self.pool.begin().await?; + + sqlx::query(Queries::RESTAMP_DOCUMENTS_USER_ID) + .bind(canonical_id.to_string()) + .bind(old_id.to_string()) + .execute(&mut *tx) + .await?; + + sqlx::query(Queries::ADOPT_USER_CONFIG_IDENTITY) + .bind(canonical_id.to_string()) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(()) + } + pub async fn get_document(&self, id: &Uuid) -> SyncResult { let row = sqlx::query(Queries::GET_DOCUMENT) .bind(id.to_string()) @@ -607,4 +628,84 @@ mod identity_tests { let adopted: i64 = row.try_get("identity_adopted").unwrap(); assert_eq!(adopted, 0); } + + async fn seed_document(db: &ClientDatabase, owner: Option) -> Uuid { + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO documents (id, user_id, content) VALUES (?1, ?2, ?3)") + .bind(id.to_string()) + .bind(owner.map(|u| u.to_string())) + .bind("{}") + .execute(&db.pool) + .await + .unwrap(); + id + } + + async fn count_docs_for(db: &ClientDatabase, owner: Uuid) -> i64 { + sqlx::query("SELECT COUNT(*) as c FROM documents WHERE user_id = ?1") + .bind(owner.to_string()) + .fetch_one(&db.pool) + .await + .unwrap() + .try_get("c") + .unwrap() + } + + #[tokio::test] + async fn adopt_identity_restamps_docs_and_flips_flag() { + let db = fresh_db().await; + db.ensure_user_config("ws://localhost/ws").await.unwrap(); + let provisional = db.get_user_id().await.unwrap(); + let canonical = Uuid::new_v4(); + + seed_document(&db, Some(provisional)).await; + seed_document(&db, Some(provisional)).await; + let public_id = seed_document(&db, None).await; + + db.adopt_identity(provisional, canonical).await.unwrap(); + + // Identity flipped in user_config. + assert_eq!(db.get_user_id().await.unwrap(), canonical); + let adopted: i64 = sqlx::query("SELECT identity_adopted FROM user_config LIMIT 1") + .fetch_one(&db.pool) + .await + .unwrap() + .try_get("identity_adopted") + .unwrap(); + assert_eq!(adopted, 1); + + // Owned documents re-stamped; none remain under the provisional id. + assert_eq!(count_docs_for(&db, canonical).await, 2); + assert_eq!(count_docs_for(&db, provisional).await, 0); + + // Public (null-owner) document untouched. + let pub_null: i64 = + sqlx::query("SELECT COUNT(*) as c FROM documents WHERE id = ?1 AND user_id IS NULL") + .bind(public_id.to_string()) + .fetch_one(&db.pool) + .await + .unwrap() + .try_get("c") + .unwrap(); + assert_eq!(pub_null, 1); + } + + #[tokio::test] + async fn adopt_identity_with_no_documents_still_flips_flag() { + let db = fresh_db().await; + db.ensure_user_config("ws://localhost/ws").await.unwrap(); + let provisional = db.get_user_id().await.unwrap(); + let canonical = Uuid::new_v4(); + + db.adopt_identity(provisional, canonical).await.unwrap(); + + assert_eq!(db.get_user_id().await.unwrap(), canonical); + let adopted: i64 = sqlx::query("SELECT identity_adopted FROM user_config LIMIT 1") + .fetch_one(&db.pool) + .await + .unwrap() + .try_get("identity_adopted") + .unwrap(); + assert_eq!(adopted, 1); + } } diff --git a/replicant-client/src/queries.rs b/replicant-client/src/queries.rs index 9a2d716..db8087a 100644 --- a/replicant-client/src/queries.rs +++ b/replicant-client/src/queries.rs @@ -82,6 +82,12 @@ impl Queries { pub const UPDATE_LAST_SYNC: &'static str = "UPDATE user_config SET last_sync_at = ?1 WHERE user_id = ?2"; + pub const RESTAMP_DOCUMENTS_USER_ID: &'static str = + "UPDATE documents SET user_id = ?1 WHERE user_id = ?2"; + + pub const ADOPT_USER_CONFIG_IDENTITY: &'static str = + "UPDATE user_config SET user_id = ?1, identity_adopted = 1"; + // Document queries pub const GET_DOCUMENT: &'static str = r#" SELECT id, user_id, content, sync_revision, From be6b50f465fb77b6ce67547ca420f9a2645db46a Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 6 Jul 2026 10:14:57 +0100 Subject: [PATCH 04/16] Store client user_id as shared interior-mutable state Claude-Session: https://claude.ai/code/session_01VpgSNzaVPefKfHKjERxTSQ --- replicant-client/src/client.rs | 22 ++++++++++++++++------ 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/replicant-client/src/client.rs b/replicant-client/src/client.rs index 132309e..84f5865 100644 --- a/replicant-client/src/client.rs +++ b/replicant-client/src/client.rs @@ -10,7 +10,7 @@ use sqlx::Row; use std::collections::HashMap; use std::sync::{ atomic::{AtomicBool, Ordering}, - Arc, + Arc, RwLock, }; use std::time::{Duration, Instant}; use tokio::sync::{mpsc, Mutex, Notify}; @@ -35,7 +35,7 @@ enum UploadType { pub struct Client { db: Arc, ws_client: Arc>>, - user_id: Uuid, + user_id: Arc>, client_id: Uuid, message_rx: Option>, event_dispatcher: Arc, @@ -123,7 +123,7 @@ impl Client { let mut engine = Self { db: db.clone(), ws_client: Arc::new(Mutex::new(ws_client)), - user_id, + user_id: Arc::new(RwLock::new(user_id)), client_id, message_rx: Some(rx), event_dispatcher: event_dispatcher.clone(), @@ -360,6 +360,12 @@ impl Client { self.create_document_with_id(Uuid::new_v4(), content).await } + /// Current server-authoritative user id. Interior-mutable so identity + /// adoption updates every subsequent local write and reconnect. + pub fn user_id(&self) -> Uuid { + *self.user_id.read().expect("user_id lock poisoned") + } + pub async fn create_document_with_id( &self, id: Uuid, @@ -367,7 +373,7 @@ impl Client { ) -> SyncResult { let doc = Document { id, - user_id: Some(self.user_id), + user_id: Some(self.user_id()), content, sync_revision: 1, content_hash: None, @@ -1629,7 +1635,7 @@ impl Client { let api_key = self.api_key.clone(); let api_secret = self.api_secret.clone(); let client_id = self.client_id; - let user_id = self.user_id; + let user_id = self.user_id.clone(); let event_dispatcher = self.event_dispatcher.clone(); let db = self.db.clone(); let pending_uploads = self.pending_uploads.clone(); @@ -1660,12 +1666,16 @@ impl Client { server_url ); + // Read the current (possibly just-adopted) id; copy out so no + // lock guard is held across the await. + let current_user_id = *user_id.read().expect("user_id lock poisoned"); + // Try to connect match WebSocketClient::connect( &server_url, &email, client_id, - user_id, + current_user_id, &api_key, &api_secret, Some(event_dispatcher.clone()), From 4e90a88ede107939bd9c66ce79459b50bf28033f Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 6 Jul 2026 10:24:05 +0100 Subject: [PATCH 05/16] Handle join reply: gate user_id payload, adopt canonical id, rejoin Claude-Session: https://claude.ai/code/session_01VpgSNzaVPefKfHKjERxTSQ --- replicant-client/src/client.rs | 37 ++++++- replicant-client/src/database.rs | 8 ++ replicant-client/src/websocket.rs | 163 +++++++++++++++++++++++++++--- 3 files changed, 190 insertions(+), 18 deletions(-) diff --git a/replicant-client/src/client.rs b/replicant-client/src/client.rs index 84f5865..e506aac 100644 --- a/replicant-client/src/client.rs +++ b/replicant-client/src/client.rs @@ -93,8 +93,9 @@ impl Client { let (reconnect_sync_tx, reconnect_sync_rx) = mpsc::channel(10); let is_connected = Arc::new(AtomicBool::new(false)); + let identity_adopted = db.is_identity_adopted().await.unwrap_or(false); // Try to connect to WebSocket, but don't fail if offline - let (ws_client, initial_ping_time) = match WebSocketClient::connect( + let (ws_client, initial_ping_time, adoption) = match WebSocketClient::connect( server_url, email, client_id, @@ -103,23 +104,35 @@ impl Client { api_secret, Some(event_dispatcher.clone()), is_connected.clone(), + identity_adopted, ) .await { - Ok((client, receiver)) => { + Ok((client, receiver, adoption)) => { // Start forwarding WebSocket messages to our channel tokio::spawn(async move { if let Err(e) = receiver.forward_to(tx).await { tracing::error!("WebSocket receiver error: {}", e); } }); - (Some(client), Some(Instant::now())) + (Some(client), Some(Instant::now()), adoption) } Err(e) => { eprintln!("Failed to connect to server (will retry): {}", e); - (None, None) + (None, None, None) } }; + + // Adopt the server's canonical id on first contact: re-stamp local docs + // and switch our in-memory identity to it. + let user_id = if let Some(a) = adoption { + db.adopt_identity(a.old_user_id, a.canonical_user_id) + .await?; + a.canonical_user_id + } else { + user_id + }; + let mut engine = Self { db: db.clone(), ws_client: Arc::new(Mutex::new(ws_client)), @@ -1669,6 +1682,7 @@ impl Client { // Read the current (possibly just-adopted) id; copy out so no // lock guard is held across the await. let current_user_id = *user_id.read().expect("user_id lock poisoned"); + let identity_adopted = db.is_identity_adopted().await.unwrap_or(false); // Try to connect match WebSocketClient::connect( @@ -1680,10 +1694,11 @@ impl Client { &api_secret, Some(event_dispatcher.clone()), is_connected.clone(), + identity_adopted, ) .await { - Ok((new_client, receiver)) => { + Ok((new_client, receiver, adoption)) => { tracing::info!( "✅ CLIENT {}: Reconnection successful after {} attempts!", client_id, @@ -1691,6 +1706,18 @@ impl Client { ); connection_attempts = 0; + // Adopt the server's canonical id if it differs. + if let Some(a) = adoption { + if let Err(e) = + db.adopt_identity(a.old_user_id, a.canonical_user_id).await + { + tracing::error!("Identity adoption failed: {}", e); + } else { + *user_id.write().expect("user_id lock poisoned") = + a.canonical_user_id; + } + } + // Update the client *ws_client.lock().await = Some(new_client); is_connected.store(true, Ordering::Relaxed); diff --git a/replicant-client/src/database.rs b/replicant-client/src/database.rs index 995a53c..1654592 100644 --- a/replicant-client/src/database.rs +++ b/replicant-client/src/database.rs @@ -98,6 +98,14 @@ impl ClientDatabase { Ok(Uuid::parse_str(&user_id)?) } + pub async fn is_identity_adopted(&self) -> SyncResult { + let row = sqlx::query("SELECT identity_adopted FROM user_config LIMIT 1") + .fetch_one(&self.pool) + .await?; + let adopted: i64 = row.try_get("identity_adopted")?; + Ok(adopted != 0) + } + pub async fn get_client_id(&self) -> SyncResult { let row = sqlx::query(Queries::GET_CLIENT_ID) .fetch_one(&self.pool) diff --git a/replicant-client/src/websocket.rs b/replicant-client/src/websocket.rs index 953bf68..fdcfb79 100644 --- a/replicant-client/src/websocket.rs +++ b/replicant-client/src/websocket.rs @@ -30,6 +30,15 @@ pub struct WebSocketClient { user_id: Uuid, } +/// The server assigned a canonical id differing from the local one; the caller +/// must re-stamp local documents and update its in-memory identity. +#[derive(Debug, Clone, PartialEq)] +pub struct Adoption { + pub old_user_id: Uuid, + pub canonical_user_id: Uuid, + pub email: String, +} + pub struct WebSocketReceiver { rx: mpsc::Receiver, } @@ -44,7 +53,8 @@ impl WebSocketClient { api_secret: &str, event_dispatcher: Option>, is_connected: Arc, - ) -> SyncResult<(Self, WebSocketReceiver)> { + identity_adopted: bool, + ) -> SyncResult<(Self, WebSocketReceiver, Option)> { Self::connect_with_hmac( server_url, email, @@ -54,6 +64,7 @@ impl WebSocketClient { api_secret, event_dispatcher, is_connected, + identity_adopted, ) .await } @@ -67,7 +78,8 @@ impl WebSocketClient { api_secret: &str, event_dispatcher: Option>, is_connected: Arc, - ) -> SyncResult<(Self, WebSocketReceiver)> { + identity_adopted: bool, + ) -> SyncResult<(Self, WebSocketReceiver, Option)> { let ws_url = Self::to_websocket_url(server_url)?; if let Some(ref d) = event_dispatcher { @@ -87,15 +99,17 @@ impl WebSocketClient { ws_err(format!("Connect failed: {:?}", e)) })?; - // Join channel with HMAC auth + // Join channel with HMAC auth. Send user_id only once identity is adopted + // (steady-state); a provisional client bootstrap-joins by email. let timestamp = chrono::Utc::now().timestamp(); let signature = Self::create_hmac_signature(api_secret, timestamp, email, api_key, ""); - let join_payload = json!({ - "email": email, - "api_key": api_key, - "signature": signature, - "timestamp": timestamp - }); + let payload_user_id = if identity_adopted { + Some(user_id) + } else { + None + }; + let join_payload = + Self::build_join_payload(email, api_key, &signature, timestamp, payload_user_id); // Join per-user channel let channel = socket @@ -106,13 +120,55 @@ impl WebSocketClient { .await .map_err(|e| ws_err(format!("Channel create failed: {:?}", e)))?; - channel.join(JOIN_TIMEOUT).await.map_err(|e| { + let join_reply = channel.join(JOIN_TIMEOUT).await.map_err(|e| { if let Some(ref d) = event_dispatcher { d.emit_sync_error(&format!("Join failed: {:?}", e)); } ws_err(format!("Join failed: {:?}", e)) })?; + // Detect a server-assigned canonical id; if it differs, leave the + // provisional topic and rejoin the canonical one (now steady-state). + let reply_value = payload_to_value(&join_reply).unwrap_or_else(|| json!({})); + let adoption = Self::should_adopt(user_id, &reply_value).map(|canonical| Adoption { + old_user_id: user_id, + canonical_user_id: canonical, + email: reply_value + .get("email") + .and_then(|v| v.as_str()) + .map(String::from) + .unwrap_or_default(), + }); + + let (channel, effective_user_id) = if let Some(ref a) = adoption { + let canonical = a.canonical_user_id; + let _ = channel.leave().await; + + let ts = chrono::Utc::now().timestamp(); + let sig = Self::create_hmac_signature(api_secret, ts, email, api_key, ""); + let canonical_payload = + Self::build_join_payload(email, api_key, &sig, ts, Some(canonical)); + + let canonical_channel = socket + .channel( + Topic::from_string(format!("sync:user:{}", canonical)), + Some(to_payload(&canonical_payload)?), + ) + .await + .map_err(|e| ws_err(format!("Canonical channel create failed: {:?}", e)))?; + + canonical_channel.join(JOIN_TIMEOUT).await.map_err(|e| { + if let Some(ref d) = event_dispatcher { + d.emit_sync_error(&format!("Canonical join failed: {:?}", e)); + } + ws_err(format!("Canonical join failed: {:?}", e)) + })?; + + (canonical_channel, canonical) + } else { + (channel, user_id) + }; + // Join public channel for public document events let public_channel = socket .channel( @@ -135,8 +191,18 @@ impl WebSocketClient { } let (tx, rx) = mpsc::channel::(100); - Self::setup_broadcast_handlers(&channel, tx.clone(), user_id, is_connected.clone()); - Self::setup_broadcast_handlers(&public_channel, tx.clone(), user_id, is_connected); + Self::setup_broadcast_handlers( + &channel, + tx.clone(), + effective_user_id, + is_connected.clone(), + ); + Self::setup_broadcast_handlers( + &public_channel, + tx.clone(), + effective_user_id, + is_connected, + ); // Emit auth success let _ = tx @@ -151,12 +217,44 @@ impl WebSocketClient { channel, _public_channel: public_channel, tx, - user_id, + user_id: effective_user_id, }, WebSocketReceiver { rx }, + adoption, )) } + /// Build the channel join payload. `user_id` is included only for a + /// steady-state (already-adopted) join; a bootstrap join omits it. + pub(crate) fn build_join_payload( + email: &str, + api_key: &str, + signature: &str, + timestamp: i64, + user_id: Option, + ) -> Value { + let mut payload = json!({ + "email": email, + "api_key": api_key, + "signature": signature, + "timestamp": timestamp, + }); + if let Some(uid) = user_id { + payload["user_id"] = json!(uid.to_string()); + } + payload + } + + /// The canonical id to adopt, if the join reply carries a `user_id` that + /// differs from the local one. `None` means no adoption is needed. + pub(crate) fn should_adopt(local: Uuid, reply: &Value) -> Option { + let canonical = reply + .get("user_id") + .and_then(|v| v.as_str()) + .and_then(|s| Uuid::parse_str(s).ok())?; + (canonical != local).then_some(canonical) + } + fn to_websocket_url(server_url: &str) -> SyncResult { let url = match server_url { s if s.starts_with("http://") => s.replace("http://", "ws://"), @@ -582,4 +680,43 @@ mod tests { assert_eq!(doc.author_name, None); assert_eq!(doc.visibility, None); } + + #[test] + fn provisional_join_payload_omits_user_id() { + let payload = WebSocketClient::build_join_payload("a@b.com", "key", "sig", 123, None); + assert!(payload.get("user_id").is_none()); + assert_eq!(payload["email"], "a@b.com"); + } + + #[test] + fn adopted_join_payload_includes_user_id() { + let uid = Uuid::new_v4(); + let payload = WebSocketClient::build_join_payload("a@b.com", "key", "sig", 123, Some(uid)); + assert_eq!(payload["user_id"], uid.to_string()); + } + + #[test] + fn should_adopt_detects_canonical_mismatch() { + let local = Uuid::new_v4(); + let canonical = Uuid::new_v4(); + let reply = serde_json::json!({ "user_id": canonical.to_string(), "email": "a@b.com" }); + assert_eq!( + WebSocketClient::should_adopt(local, &reply), + Some(canonical) + ); + } + + #[test] + fn should_adopt_noop_when_ids_match() { + let local = Uuid::new_v4(); + let reply = serde_json::json!({ "user_id": local.to_string() }); + assert_eq!(WebSocketClient::should_adopt(local, &reply), None); + } + + #[test] + fn should_adopt_noop_when_reply_has_no_user_id() { + let local = Uuid::new_v4(); + let reply = serde_json::json!({ "email": "a@b.com" }); + assert_eq!(WebSocketClient::should_adopt(local, &reply), None); + } } From 0a32a23fbdbd94cf4ae8a487881f094d57a6374e Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 6 Jul 2026 10:35:31 +0100 Subject: [PATCH 06/16] Emit identity_changed event across Rust and FFI on adoption Adds EventType::IdentityChanged, SyncEvent::IdentityChanged, the emit/register/dispatch surface, a C-FFI IdentityEventCallback + replicant_register_identity_callback, and regenerates the cbindgen C header. The emit is wired into both adoption paths in client.rs. Claude-Session: https://claude.ai/code/session_01VpgSNzaVPefKfHKjERxTSQ --- replicant-client/include/replicant.h | 38 +++++++ replicant-client/src/client.rs | 6 + replicant-client/src/events.rs | 161 +++++++++++++++++++++++++++ replicant-client/src/ffi.rs | 35 +++++- 4 files changed, 239 insertions(+), 1 deletion(-) diff --git a/replicant-client/include/replicant.h b/replicant-client/include/replicant.h index b543f3a..402aa15 100644 --- a/replicant-client/include/replicant.h +++ b/replicant-client/include/replicant.h @@ -62,6 +62,10 @@ typedef enum ReplicantEventType { * Successfully connected to the server */ ConnectionSucceeded = 9, + /** + * The server-authoritative user id was adopted, replacing the local one + */ + IdentityChanged = 10, } ReplicantEventType; /** @@ -157,6 +161,22 @@ typedef void (*ConflictEventCallback)(enum ReplicantEventType event_type, const char *losing_content, void *context); +/** + * Identity event callback for IdentityChanged + * + * # Parameters + * * `event_type` - Always IdentityChanged + * * `old_user_id` - The provisional/previous user id (always non-null) + * * `new_user_id` - The adopted canonical user id (always non-null) + * * `email` - The email the server resolved the id from (may be empty) + * * `context` - User-defined context pointer + */ +typedef void (*IdentityEventCallback)(enum ReplicantEventType event_type, + const char *old_user_id, + const char *new_user_id, + const char *email, + void *context); + /** * Document structure for C API */ @@ -402,6 +422,24 @@ enum ReplicantSyncResult replicant_register_conflict_callback(struct Replicant * ConflictEventCallback callback, void *context); +/** + * Register a callback for IdentityChanged events + * + * # Arguments + * * `engine` - Sync engine instance + * * `callback` - Function to call when the server-authoritative id is adopted + * * `context` - User-defined context pointer passed to callback + * + * # Returns + * * SyncResult indicating success or failure + * + * # Safety + * Caller must ensure engine is valid, callback is a valid function pointer, and context pointer outlives the callback registration + */ +enum ReplicantSyncResult replicant_register_identity_callback(struct Replicant *engine, + IdentityEventCallback callback, + void *context); + /** * Process all queued events on the current thread * diff --git a/replicant-client/src/client.rs b/replicant-client/src/client.rs index e506aac..e23bf54 100644 --- a/replicant-client/src/client.rs +++ b/replicant-client/src/client.rs @@ -128,6 +128,7 @@ impl Client { let user_id = if let Some(a) = adoption { db.adopt_identity(a.old_user_id, a.canonical_user_id) .await?; + event_dispatcher.emit_identity_changed(&a.old_user_id, &a.canonical_user_id, &a.email); a.canonical_user_id } else { user_id @@ -1715,6 +1716,11 @@ impl Client { } else { *user_id.write().expect("user_id lock poisoned") = a.canonical_user_id; + event_dispatcher.emit_identity_changed( + &a.old_user_id, + &a.canonical_user_id, + &a.email, + ); } } diff --git a/replicant-client/src/events.rs b/replicant-client/src/events.rs index 526e16b..2dc4c8c 100644 --- a/replicant-client/src/events.rs +++ b/replicant-client/src/events.rs @@ -61,6 +61,8 @@ pub enum EventType { ConnectionAttempted = 8, /// Successfully connected to the server ConnectionSucceeded = 9, + /// The server-authoritative user id was adopted, replacing the local one + IdentityChanged = 10, } // ============================================================================= @@ -131,6 +133,12 @@ pub enum SyncEvent { ConnectionAttempted { server_url: String }, /// Successfully connected to server ConnectionSucceeded { server_url: String }, + /// The server-authoritative user id was adopted, replacing the local one + IdentityChanged { + old_user_id: String, + new_user_id: String, + email: String, + }, } impl SyncEvent { @@ -147,6 +155,7 @@ impl SyncEvent { SyncEvent::ConnectionLost { .. } => EventType::ConnectionLost, SyncEvent::ConnectionAttempted { .. } => EventType::ConnectionAttempted, SyncEvent::ConnectionSucceeded { .. } => EventType::ConnectionSucceeded, + SyncEvent::IdentityChanged { .. } => EventType::IdentityChanged, } } @@ -210,6 +219,11 @@ impl SyncEvent { EventType::ConnectionSucceeded => SyncEvent::ConnectionSucceeded { server_url: event.title.clone().unwrap_or_default(), }, + EventType::IdentityChanged => SyncEvent::IdentityChanged { + old_user_id: event.document_id.clone().unwrap_or_default(), + new_user_id: event.user_id.clone().unwrap_or_default(), + email: event.title.clone().unwrap_or_default(), + }, } } } @@ -288,6 +302,22 @@ pub type ConflictEventCallback = extern "C" fn( context: *mut c_void, ); +/// Identity event callback for IdentityChanged +/// +/// # Parameters +/// * `event_type` - Always IdentityChanged +/// * `old_user_id` - The provisional/previous user id (always non-null) +/// * `new_user_id` - The adopted canonical user id (always non-null) +/// * `email` - The email the server resolved the id from (may be empty) +/// * `context` - User-defined context pointer +pub type IdentityEventCallback = extern "C" fn( + event_type: EventType, + old_user_id: *const c_char, + new_user_id: *const c_char, + email: *const c_char, + context: *mut c_void, +); + // ============================================================================= // Callback Entry Types (Internal) // ============================================================================= @@ -318,6 +348,11 @@ struct ConflictCallbackEntry { context: *mut c_void, } +struct IdentityCallbackEntry { + callback: IdentityEventCallback, + context: *mut c_void, +} + // Safety: Callback entries are only accessed from the registered thread unsafe impl Send for DocumentCallbackEntry {} unsafe impl Sync for DocumentCallbackEntry {} @@ -329,6 +364,8 @@ unsafe impl Send for ConnectionCallbackEntry {} unsafe impl Sync for ConnectionCallbackEntry {} unsafe impl Send for ConflictCallbackEntry {} unsafe impl Sync for ConflictCallbackEntry {} +unsafe impl Send for IdentityCallbackEntry {} +unsafe impl Sync for IdentityCallbackEntry {} // ============================================================================= // Rust Callback Entry (Internal) @@ -401,6 +438,7 @@ pub struct EventDispatcher { error_callbacks: Mutex>, connection_callbacks: Mutex>, conflict_callbacks: Mutex>, + identity_callbacks: Mutex>, // Rust-native callback storage rust_callbacks: Mutex>, // Event queue @@ -418,6 +456,7 @@ impl EventDispatcher { error_callbacks: Mutex::new(Vec::new()), connection_callbacks: Mutex::new(Vec::new()), conflict_callbacks: Mutex::new(Vec::new()), + identity_callbacks: Mutex::new(Vec::new()), rust_callbacks: Mutex::new(Vec::new()), event_queue: Mutex::new(receiver), event_sender: sender, @@ -557,6 +596,23 @@ impl EventDispatcher { Ok(()) } + pub fn register_identity_callback( + &self, + callback: IdentityEventCallback, + context: *mut c_void, + ) -> SyncResult<()> { + self.ensure_callback_thread()?; + + let mut callbacks = self + .identity_callbacks + .lock() + .map_err(|_| ClientError::LockError("identity_callbacks".into()))?; + + callbacks.push(IdentityCallbackEntry { callback, context }); + + Ok(()) + } + /// Register a Rust-native callback for all events /// /// This provides an idiomatic Rust interface using the `SyncEvent` enum. @@ -820,6 +876,22 @@ impl EventDispatcher { ); } + pub fn emit_identity_changed(&self, old_user_id: &Uuid, new_user_id: &Uuid, email: &str) { + // document_id carries the old id, title the email, user_id the new id. + self.queue_event( + EventType::IdentityChanged, + Some(old_user_id), + Some(email), + None, + None, + 0, + false, + Some(new_user_id.to_string()), + None, + None, + ); + } + /// Queue an event for later processing on the callback thread #[allow(clippy::too_many_arguments)] // FFI callback constraints fn queue_event( @@ -875,6 +947,10 @@ impl EventDispatcher { .conflict_callbacks .lock() .map_err(|_| ClientError::LockError("conflict_callbacks".into()))?; + let identity = self + .identity_callbacks + .lock() + .map_err(|_| ClientError::LockError("identity_callbacks".into()))?; let rust = self .rust_callbacks .lock() @@ -885,6 +961,7 @@ impl EventDispatcher { || !error.is_empty() || !conn.is_empty() || !conflict.is_empty() + || !identity.is_empty() || !rust.is_empty()) } @@ -930,6 +1007,10 @@ impl EventDispatcher { .conflict_callbacks .lock() .map_err(|_| ClientError::LockError("conflict_callbacks".into()))?; + let identity_callbacks = self + .identity_callbacks + .lock() + .map_err(|_| ClientError::LockError("identity_callbacks".into()))?; let rust_callbacks = self .rust_callbacks .lock() @@ -1087,6 +1168,22 @@ impl EventDispatcher { ); } } + + EventType::IdentityChanged => { + let old_ptr = document_id_cstr.unwrap_or(std::ptr::null()); + let new_ptr = user_id_cstr.unwrap_or(std::ptr::null()); + let email_ptr = title_cstr.unwrap_or(std::ptr::null()); + + for entry in identity_callbacks.iter() { + (entry.callback)( + queued_event.event_type, + old_ptr, + new_ptr, + email_ptr, + entry.context, + ); + } + } } processed_count += 1; @@ -1325,6 +1422,70 @@ mod tests { assert_eq!(processed, 0); } + #[test] + fn identity_changed_delivers_to_rust_callback() { + let dispatcher = EventDispatcher::new(); + let captured: Arc>> = Arc::new(Mutex::new(None)); + let c = captured.clone(); + dispatcher + .register_rust_callback(move |e| *c.lock().unwrap() = Some(e)) + .unwrap(); + + let old = Uuid::new_v4(); + let new = Uuid::new_v4(); + dispatcher.emit_identity_changed(&old, &new, "user@example.com"); + assert_eq!(dispatcher.process_events().unwrap(), 1); + + let result = captured.lock().unwrap().clone(); + match result { + Some(SyncEvent::IdentityChanged { + old_user_id, + new_user_id, + email, + }) => { + assert_eq!(old_user_id, old.to_string()); + assert_eq!(new_user_id, new.to_string()); + assert_eq!(email, "user@example.com"); + } + other => panic!("expected IdentityChanged, got {:?}", other), + } + } + + #[test] + fn identity_changed_delivers_to_ffi_callback() { + let dispatcher = EventDispatcher::new(); + let count = Arc::new(AtomicUsize::new(0)); + let count_clone = count.clone(); + + extern "C" fn identity_callback( + event_type: EventType, + old_user_id: *const c_char, + new_user_id: *const c_char, + email: *const c_char, + context: *mut c_void, + ) { + assert_eq!(event_type, EventType::IdentityChanged); + assert!(!old_user_id.is_null()); + assert!(!new_user_id.is_null()); + assert!(!email.is_null()); + let c = unsafe { &*(context as *const AtomicUsize) }; + c.fetch_add(1, Ordering::SeqCst); + } + + dispatcher + .register_identity_callback( + identity_callback, + &*count_clone as *const AtomicUsize as *mut c_void, + ) + .unwrap(); + + let old = Uuid::new_v4(); + let new = Uuid::new_v4(); + dispatcher.emit_identity_changed(&old, &new, "user@example.com"); + assert_eq!(dispatcher.process_events().unwrap(), 1); + assert_eq!(count.load(Ordering::SeqCst), 1); + } + #[test] fn test_error_callback() { let dispatcher = EventDispatcher::new(); diff --git a/replicant-client/src/ffi.rs b/replicant-client/src/ffi.rs index ab36656..44c4a52 100644 --- a/replicant-client/src/ffi.rs +++ b/replicant-client/src/ffi.rs @@ -13,7 +13,7 @@ use uuid::Uuid; use crate::events::{ ConflictEventCallback, ConnectionEventCallback, DocumentEventCallback, ErrorEventCallback, - EventDispatcher, EventType, SyncEventCallback, + EventDispatcher, EventType, IdentityEventCallback, SyncEventCallback, }; use crate::{Client as CoreClient, ClientDatabase}; @@ -815,6 +815,39 @@ pub unsafe extern "C" fn replicant_register_conflict_callback( } } +/// Register a callback for IdentityChanged events +/// +/// # Arguments +/// * `engine` - Sync engine instance +/// * `callback` - Function to call when the server-authoritative id is adopted +/// * `context` - User-defined context pointer passed to callback +/// +/// # Returns +/// * SyncResult indicating success or failure +/// +/// # Safety +/// Caller must ensure engine is valid, callback is a valid function pointer, and context pointer outlives the callback registration +#[no_mangle] +pub unsafe extern "C" fn replicant_register_identity_callback( + engine: *mut Replicant, + callback: IdentityEventCallback, + context: *mut c_void, +) -> SyncResult { + if engine.is_null() { + return SyncResult::ErrorInvalidInput; + } + + let engine = &*engine; + + match engine + .event_dispatcher + .register_identity_callback(callback, context) + { + Ok(_) => SyncResult::Success, + Err(_) => SyncResult::ErrorUnknown, + } +} + /// Process all queued events on the current thread /// /// # Arguments From ed029328ead994b6db7be0d2160f2b212b0579f7 Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 6 Jul 2026 10:39:37 +0100 Subject: [PATCH 07/16] Handle IdentityChanged in task_list example match Claude-Session: https://claude.ai/code/session_01VpgSNzaVPefKfHKjERxTSQ --- replicant-client/examples/task_list_example.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/replicant-client/examples/task_list_example.rs b/replicant-client/examples/task_list_example.rs index 02999ad..dd38b56 100644 --- a/replicant-client/examples/task_list_example.rs +++ b/replicant-client/examples/task_list_example.rs @@ -626,6 +626,15 @@ async fn main() -> Result<(), Box> { ActivityType::Error, ); } + SyncEvent::IdentityChanged { new_user_id, .. } => { + app_state.add_activity( + format!( + "Identity adopted: {}...", + &new_user_id[..8.min(new_user_id.len())] + ), + ActivityType::Connected, + ); + } } }) { From 6b8d0a13ddb9d5e0e023fcd105a8398aebd74862 Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 6 Jul 2026 14:53:21 +0100 Subject: [PATCH 08/16] Add gated end-to-end identity-adoption integration test Drives the real Client through the join-reply adoption path against a live server: pre-seeds a provisional id + document, then asserts the document is re-stamped to the server's canonical id, identity_adopted flips, and the in-memory id updates. Gated behind RUN_INTEGRATION_TESTS. Claude-Session: https://claude.ai/code/session_01VpgSNzaVPefKfHKjERxTSQ --- .../identity_adoption_test.rs | 110 ++++++++++++++++++ .../tests/phoenix_integration/mod.rs | 1 + 2 files changed, 111 insertions(+) create mode 100644 replicant-client/tests/phoenix_integration/identity_adoption_test.rs diff --git a/replicant-client/tests/phoenix_integration/identity_adoption_test.rs b/replicant-client/tests/phoenix_integration/identity_adoption_test.rs new file mode 100644 index 0000000..a467a39 --- /dev/null +++ b/replicant-client/tests/phoenix_integration/identity_adoption_test.rs @@ -0,0 +1,110 @@ +//! End-to-end identity adoption against a live server. +//! +//! Pre-seeds a provisional (random v4) identity plus a document owned by it, +//! then constructs the real `Client` pointed at the deployed server. Connecting +//! runs the join-reply → adoption path, which must re-stamp the document to the +//! server's canonical id, flip `identity_adopted`, and update the in-memory id. +//! +//! Gated behind `RUN_INTEGRATION_TESTS` (see `skip_if_no_server`); needs +//! `SYNC_SERVER_URL`, `REPLICANT_API_KEY`, `REPLICANT_API_SECRET`. + +use super::{serial, server_url, skip_if_no_server, test_api_key, test_api_secret, TEST_EMAIL}; +use replicant_client::{Client, ClientDatabase}; +use sqlx::Row; +use uuid::Uuid; + +fn temp_db_path() -> String { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + format!( + "databases/identity_adopt_{}_{}.sqlite3", + std::process::id(), + nanos + ) +} + +#[tokio::test] +#[serial] +async fn adoption_restamps_document_and_flips_flag_against_live_server() { + if skip_if_no_server() { + eprintln!("skipping identity adoption test: RUN_INTEGRATION_TESTS not set"); + return; + } + + std::fs::create_dir_all("databases").ok(); + let db_file = temp_db_path(); + let db_url = format!("sqlite:{}?mode=rwc", db_file); + + // 1. Pre-seed the offline state: a provisional random id + a document under it. + let provisional; + let doc_id = Uuid::new_v4(); + { + let db = ClientDatabase::new(&db_url).await.unwrap(); + db.run_migrations().await.unwrap(); + db.ensure_user_config(&server_url()).await.unwrap(); + + provisional = db.get_user_id().await.unwrap(); + assert_eq!( + provisional.get_version(), + Some(uuid::Version::Random), + "seeded id should be a random v4 provisional id" + ); + + sqlx::query("INSERT INTO documents (id, user_id, content) VALUES (?1, ?2, ?3)") + .bind(doc_id.to_string()) + .bind(provisional.to_string()) + .bind("{}") + .execute(&db.pool) + .await + .unwrap(); + } + + // 2. Construct the real client on the same db, pointed at the live server. + // Adoption runs synchronously inside construction, on first join reply. + let client = Client::with_event_dispatcher( + &db_url, + &server_url(), + TEST_EMAIL, + &test_api_key(), + &test_api_secret(), + None, + ) + .await + .expect("client should connect to the live server"); + + // 3. The in-memory id switched to the server's canonical id. + let canonical = client.user_id(); + assert_ne!( + canonical, provisional, + "client should have adopted the server's canonical id" + ); + + // 4. The DB reflects the adoption and the pre-existing doc was re-stamped. + let db = ClientDatabase::new(&db_url).await.unwrap(); + let adopted: i64 = sqlx::query("SELECT identity_adopted FROM user_config LIMIT 1") + .fetch_one(&db.pool) + .await + .unwrap() + .try_get("identity_adopted") + .unwrap(); + assert_eq!(adopted, 1, "identity_adopted should be flipped"); + assert_eq!(db.get_user_id().await.unwrap(), canonical); + + let doc_owner: String = sqlx::query("SELECT user_id FROM documents WHERE id = ?1") + .bind(doc_id.to_string()) + .fetch_one(&db.pool) + .await + .unwrap() + .try_get("user_id") + .unwrap(); + assert_eq!( + doc_owner, + canonical.to_string(), + "the pre-existing document should be re-stamped to the canonical id" + ); + + drop(client); + std::fs::remove_file(&db_file).ok(); +} diff --git a/replicant-client/tests/phoenix_integration/mod.rs b/replicant-client/tests/phoenix_integration/mod.rs index 84b5a81..a15abad 100644 --- a/replicant-client/tests/phoenix_integration/mod.rs +++ b/replicant-client/tests/phoenix_integration/mod.rs @@ -22,6 +22,7 @@ mod basic_sync_test; mod conflict_test; +mod identity_adoption_test; mod live_sync_test; mod multi_client_test; From 72d8728457b0d2cba246493f876e5c4694065ae0 Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 13 Jul 2026 15:33:29 +0100 Subject: [PATCH 09/16] fix(identity): gate adoption on non-nil id and single adoption adopt_identity now rejects a nil canonical_id, old_id == canonical_id, and a second adoption attempt. The user_config UPDATE is scoped to the expected old_id and errors if rows_affected != 1, rolling back the transaction. --- replicant-client/src/database.rs | 42 ++++++++++++++++++++++++++++++-- replicant-client/src/queries.rs | 2 +- 2 files changed, 41 insertions(+), 3 deletions(-) diff --git a/replicant-client/src/database.rs b/replicant-client/src/database.rs index 1654592..63c3533 100644 --- a/replicant-client/src/database.rs +++ b/replicant-client/src/database.rs @@ -3,7 +3,7 @@ use json_patch; use replicant_core::protocol::ChangeEventType; use replicant_core::{ models::{Document, SyncStatus}, - SyncResult, + SyncError, SyncResult, }; use sqlx::{sqlite::SqlitePoolOptions, Row, SqlitePool}; use uuid::Uuid; @@ -129,6 +129,17 @@ impl ClientDatabase { /// owned by `old_id` and flip `user_config` to `canonical_id` with /// `identity_adopted = 1`. A crash mid-adoption leaves the old id intact. pub async fn adopt_identity(&self, old_id: Uuid, canonical_id: Uuid) -> SyncResult<()> { + if canonical_id.is_nil() || old_id == canonical_id { + return Err(SyncError::InvalidOperation( + "adopt_identity: canonical_id must be non-nil and differ from old_id".to_string(), + )); + } + if self.is_identity_adopted().await? { + return Err(SyncError::InvalidOperation( + "adopt_identity: identity has already been adopted".to_string(), + )); + } + let mut tx = self.pool.begin().await?; sqlx::query(Queries::RESTAMP_DOCUMENTS_USER_ID) @@ -137,11 +148,20 @@ impl ClientDatabase { .execute(&mut *tx) .await?; - sqlx::query(Queries::ADOPT_USER_CONFIG_IDENTITY) + let result = sqlx::query(Queries::ADOPT_USER_CONFIG_IDENTITY) .bind(canonical_id.to_string()) + .bind(old_id.to_string()) .execute(&mut *tx) .await?; + if result.rows_affected() != 1 { + return Err(SyncError::InvalidOperation(format!( + "adopt_identity: expected to update 1 user_config row for old_id {}, got {}", + old_id, + result.rows_affected() + ))); + } + tx.commit().await?; Ok(()) } @@ -716,4 +736,22 @@ mod identity_tests { .unwrap(); assert_eq!(adopted, 1); } + + #[tokio::test] + async fn adopt_identity_rejects_nil_canonical_id() { + let db = fresh_db().await; + db.ensure_user_config("ws://localhost/ws").await.unwrap(); + let provisional = db.get_user_id().await.unwrap(); + assert!(db.adopt_identity(provisional, Uuid::nil()).await.is_err()); + } + + #[tokio::test] + async fn adopt_identity_rejects_second_adoption() { + let db = fresh_db().await; + db.ensure_user_config("ws://localhost/ws").await.unwrap(); + let provisional = db.get_user_id().await.unwrap(); + let canonical = Uuid::new_v4(); + db.adopt_identity(provisional, canonical).await.unwrap(); + assert!(db.adopt_identity(canonical, Uuid::new_v4()).await.is_err()); + } } diff --git a/replicant-client/src/queries.rs b/replicant-client/src/queries.rs index db8087a..40b8a60 100644 --- a/replicant-client/src/queries.rs +++ b/replicant-client/src/queries.rs @@ -86,7 +86,7 @@ impl Queries { "UPDATE documents SET user_id = ?1 WHERE user_id = ?2"; pub const ADOPT_USER_CONFIG_IDENTITY: &'static str = - "UPDATE user_config SET user_id = ?1, identity_adopted = 1"; + "UPDATE user_config SET user_id = ?1, identity_adopted = 1 WHERE user_id = ?2"; // Document queries pub const GET_DOCUMENT: &'static str = r#" From 648431f5667c836de92a7b671d980bbe88bdf6e2 Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 13 Jul 2026 15:54:54 +0100 Subject: [PATCH 10/16] feat(identity)!: adopt canonical id at construction; remove bootstrap join The canonical user id (delivered by enrollment claim) is now passed into Client construction and adopted before any WebSocket work, so adoption cannot race live document creation. The provisional-topic bootstrap join and post-connect adoption paths are removed; the join reply's user_id is a drift check that refuses to sync on mismatch. Sync is attempted only with credentials and an adopted identity; account switch on an adopted database is rejected. Document payloads without an ownership envelope are rejected instead of stamped with the local owner. --- .../examples/interactive_client.rs | 1 + replicant-client/examples/simple_sync_test.rs | 1 + .../examples/task_list_example.rs | 1 + .../examples/test_rust_callbacks.rs | 4 + replicant-client/src/client.rs | 138 +++++++----- replicant-client/src/ffi.rs | 3 + replicant-client/src/websocket.rs | 208 +++++++----------- .../tests/constructor_identity_tests.rs | 76 +++++++ .../identity_adoption_test.rs | 54 +++-- 9 files changed, 286 insertions(+), 200 deletions(-) create mode 100644 replicant-client/tests/constructor_identity_tests.rs diff --git a/replicant-client/examples/interactive_client.rs b/replicant-client/examples/interactive_client.rs index 241b953..73bb0ab 100644 --- a/replicant-client/examples/interactive_client.rs +++ b/replicant-client/examples/interactive_client.rs @@ -93,6 +93,7 @@ async fn main() -> Result<(), Box> { &cli.email, &cli.api_key, &cli.api_secret, + None, ) .await { diff --git a/replicant-client/examples/simple_sync_test.rs b/replicant-client/examples/simple_sync_test.rs index 2213d3a..beca0f9 100644 --- a/replicant-client/examples/simple_sync_test.rs +++ b/replicant-client/examples/simple_sync_test.rs @@ -111,6 +111,7 @@ async fn main() -> Result<(), Box> { &user_email, "test-key", "test-secret", + None, ) .await?; diff --git a/replicant-client/examples/task_list_example.rs b/replicant-client/examples/task_list_example.rs index dd38b56..4ee3da8 100644 --- a/replicant-client/examples/task_list_example.rs +++ b/replicant-client/examples/task_list_example.rs @@ -541,6 +541,7 @@ async fn main() -> Result<(), Box> { &user_email, &cli.api_key, &cli.api_secret, + None, ) .await { diff --git a/replicant-client/examples/test_rust_callbacks.rs b/replicant-client/examples/test_rust_callbacks.rs index ef59999..c35421f 100644 --- a/replicant-client/examples/test_rust_callbacks.rs +++ b/replicant-client/examples/test_rust_callbacks.rs @@ -60,6 +60,7 @@ async fn main() -> Result<(), Box> { "test-user@example.com", "test-key", "test-secret", + None, ) .await { @@ -100,6 +101,9 @@ async fn main() -> Result<(), Box> { SyncEvent::ConflictDetected { document_id, .. } => { format!("⚠️ Conflict detected: {}", &document_id[..8]) } + SyncEvent::IdentityChanged { new_user_id, .. } => { + format!("🪪 Identity changed: {}", new_user_id) + } }; if let Ok(mut t) = tracker_clone.lock() { diff --git a/replicant-client/src/client.rs b/replicant-client/src/client.rs index e23bf54..9075692 100644 --- a/replicant-client/src/client.rs +++ b/replicant-client/src/client.rs @@ -53,6 +53,9 @@ pub struct Client { reconnect_sync_rx: Option>, // Queue for deferred sync messages during upload protection deferred_messages: Arc>>, + // Sync is possible for this instance (credentials + adopted identity). + // Immutable for the client's lifetime: enrollment recreates the client. + sync_enabled: bool, } impl Client { @@ -62,9 +65,18 @@ impl Client { email: &str, api_key: &str, api_secret: &str, + canonical_user_id: Option, ) -> SyncResult { - Self::with_event_dispatcher(database_url, server_url, email, api_key, api_secret, None) - .await + Self::with_event_dispatcher( + database_url, + server_url, + email, + api_key, + api_secret, + canonical_user_id, + None, + ) + .await } pub async fn with_event_dispatcher( @@ -73,6 +85,7 @@ impl Client { email: &str, api_key: &str, api_secret: &str, + canonical_user_id: Option, event_dispatcher: Option>, ) -> SyncResult { let db = Arc::new(ClientDatabase::new(database_url).await?); @@ -93,45 +106,63 @@ impl Client { let (reconnect_sync_tx, reconnect_sync_rx) = mpsc::channel(10); let is_connected = Arc::new(AtomicBool::new(false)); + + // Adopt the canonical identity (from stored credentials) BEFORE any + // WebSocket work: no live sync exists yet and the handle hasn't been + // returned, so adoption cannot race document creation. let identity_adopted = db.is_identity_adopted().await.unwrap_or(false); - // Try to connect to WebSocket, but don't fail if offline - let (ws_client, initial_ping_time, adoption) = match WebSocketClient::connect( - server_url, - email, - client_id, - user_id, - api_key, - api_secret, - Some(event_dispatcher.clone()), - is_connected.clone(), - identity_adopted, - ) - .await - { - Ok((client, receiver, adoption)) => { - // Start forwarding WebSocket messages to our channel - tokio::spawn(async move { - if let Err(e) = receiver.forward_to(tx).await { - tracing::error!("WebSocket receiver error: {}", e); - } - }); - (Some(client), Some(Instant::now()), adoption) + match (canonical_user_id, identity_adopted) { + (Some(canonical), false) => { + db.adopt_identity(user_id, canonical).await?; + event_dispatcher.emit_identity_changed(&user_id, &canonical, email); } - Err(e) => { - eprintln!("Failed to connect to server (will retry): {}", e); - (None, None, None) + (Some(canonical), true) if canonical != user_id => { + return Err(replicant_core::errors::SyncError::InvalidOperation( + format!( + "account switch not supported: credentials belong to user {} but this \ + database is owned by user {}; reset local data to enroll a different account", + canonical, user_id + ), + )); + } + _ => {} + } + let user_id = db.get_user_id().await?; + + // Sync requires credentials and an adopted (server-confirmed) + // identity; otherwise stay local-only and never join a sync topic. + let sync_enabled = !api_key.is_empty() && db.is_identity_adopted().await.unwrap_or(false); + + let (ws_client, initial_ping_time) = if sync_enabled { + // Try to connect to WebSocket, but don't fail if offline + match WebSocketClient::connect( + server_url, + email, + client_id, + user_id, + api_key, + api_secret, + Some(event_dispatcher.clone()), + is_connected.clone(), + ) + .await + { + Ok((client, receiver)) => { + // Start forwarding WebSocket messages to our channel + tokio::spawn(async move { + if let Err(e) = receiver.forward_to(tx).await { + tracing::error!("WebSocket receiver error: {}", e); + } + }); + (Some(client), Some(Instant::now())) + } + Err(e) => { + eprintln!("Failed to connect to server (will retry): {}", e); + (None, None) + } } - }; - - // Adopt the server's canonical id on first contact: re-stamp local docs - // and switch our in-memory identity to it. - let user_id = if let Some(a) = adoption { - db.adopt_identity(a.old_user_id, a.canonical_user_id) - .await?; - event_dispatcher.emit_identity_changed(&a.old_user_id, &a.canonical_user_id, &a.email); - a.canonical_user_id } else { - user_id + (None, None) }; let mut engine = Self { @@ -153,6 +184,7 @@ impl Client { reconnect_sync_tx, reconnect_sync_rx: Some(reconnect_sync_rx), deferred_messages: Arc::new(Mutex::new(Vec::new())), + sync_enabled, }; // Automatically start background tasks @@ -1642,6 +1674,14 @@ impl Client { /// Start the reconnection loop if not already running fn start_reconnection_loop(&self) { + if !self.sync_enabled { + tracing::info!( + "CLIENT {}: sync disabled (no credentials or identity not adopted) — \ + skipping reconnection monitor", + self.client_id + ); + return; + } let is_connected = self.is_connected.clone(); let ws_client = self.ws_client.clone(); let server_url = self.server_url.clone(); @@ -1680,10 +1720,8 @@ impl Client { server_url ); - // Read the current (possibly just-adopted) id; copy out so no - // lock guard is held across the await. + // Copy the id out so no lock guard is held across the await. let current_user_id = *user_id.read().expect("user_id lock poisoned"); - let identity_adopted = db.is_identity_adopted().await.unwrap_or(false); // Try to connect match WebSocketClient::connect( @@ -1695,11 +1733,10 @@ impl Client { &api_secret, Some(event_dispatcher.clone()), is_connected.clone(), - identity_adopted, ) .await { - Ok((new_client, receiver, adoption)) => { + Ok((new_client, receiver)) => { tracing::info!( "✅ CLIENT {}: Reconnection successful after {} attempts!", client_id, @@ -1707,23 +1744,6 @@ impl Client { ); connection_attempts = 0; - // Adopt the server's canonical id if it differs. - if let Some(a) = adoption { - if let Err(e) = - db.adopt_identity(a.old_user_id, a.canonical_user_id).await - { - tracing::error!("Identity adoption failed: {}", e); - } else { - *user_id.write().expect("user_id lock poisoned") = - a.canonical_user_id; - event_dispatcher.emit_identity_changed( - &a.old_user_id, - &a.canonical_user_id, - &a.email, - ); - } - } - // Update the client *ws_client.lock().await = Some(new_client); is_connected.store(true, Ordering::Relaxed); diff --git a/replicant-client/src/ffi.rs b/replicant-client/src/ffi.rs index 44c4a52..4994322 100644 --- a/replicant-client/src/ffi.rs +++ b/replicant-client/src/ffi.rs @@ -150,6 +150,9 @@ pub unsafe extern "C" fn replicant_create( &email, &api_key, &api_secret, + // Canonical id plumbed through the C ABI in a follow-up change; + // until then adoption never triggers via FFI construction. + None, Some(event_dispatcher_clone.clone()), ) .await diff --git a/replicant-client/src/websocket.rs b/replicant-client/src/websocket.rs index fdcfb79..e27cffd 100644 --- a/replicant-client/src/websocket.rs +++ b/replicant-client/src/websocket.rs @@ -27,16 +27,6 @@ pub struct WebSocketClient { channel: Arc, _public_channel: Arc, tx: mpsc::Sender, - user_id: Uuid, -} - -/// The server assigned a canonical id differing from the local one; the caller -/// must re-stamp local documents and update its in-memory identity. -#[derive(Debug, Clone, PartialEq)] -pub struct Adoption { - pub old_user_id: Uuid, - pub canonical_user_id: Uuid, - pub email: String, } pub struct WebSocketReceiver { @@ -53,8 +43,7 @@ impl WebSocketClient { api_secret: &str, event_dispatcher: Option>, is_connected: Arc, - identity_adopted: bool, - ) -> SyncResult<(Self, WebSocketReceiver, Option)> { + ) -> SyncResult<(Self, WebSocketReceiver)> { Self::connect_with_hmac( server_url, email, @@ -64,7 +53,6 @@ impl WebSocketClient { api_secret, event_dispatcher, is_connected, - identity_adopted, ) .await } @@ -78,8 +66,7 @@ impl WebSocketClient { api_secret: &str, event_dispatcher: Option>, is_connected: Arc, - identity_adopted: bool, - ) -> SyncResult<(Self, WebSocketReceiver, Option)> { + ) -> SyncResult<(Self, WebSocketReceiver)> { let ws_url = Self::to_websocket_url(server_url)?; if let Some(ref d) = event_dispatcher { @@ -99,17 +86,11 @@ impl WebSocketClient { ws_err(format!("Connect failed: {:?}", e)) })?; - // Join channel with HMAC auth. Send user_id only once identity is adopted - // (steady-state); a provisional client bootstrap-joins by email. + // Join channel with HMAC auth; the topic and payload both carry the + // caller's (already-adopted) user id. let timestamp = chrono::Utc::now().timestamp(); let signature = Self::create_hmac_signature(api_secret, timestamp, email, api_key, ""); - let payload_user_id = if identity_adopted { - Some(user_id) - } else { - None - }; - let join_payload = - Self::build_join_payload(email, api_key, &signature, timestamp, payload_user_id); + let join_payload = Self::build_join_payload(email, api_key, &signature, timestamp, user_id); // Join per-user channel let channel = socket @@ -127,47 +108,17 @@ impl WebSocketClient { ws_err(format!("Join failed: {:?}", e)) })?; - // Detect a server-assigned canonical id; if it differs, leave the - // provisional topic and rejoin the canonical one (now steady-state). + // Identity drift check: every join reply names the credential's user + // id. A mismatch means the local identity diverged from the account — + // refuse to sync rather than silently re-stamp anything. let reply_value = payload_to_value(&join_reply).unwrap_or_else(|| json!({})); - let adoption = Self::should_adopt(user_id, &reply_value).map(|canonical| Adoption { - old_user_id: user_id, - canonical_user_id: canonical, - email: reply_value - .get("email") - .and_then(|v| v.as_str()) - .map(String::from) - .unwrap_or_default(), - }); - - let (channel, effective_user_id) = if let Some(ref a) = adoption { - let canonical = a.canonical_user_id; + if let Err(msg) = Self::verify_reply_identity(user_id, &reply_value) { + if let Some(ref d) = event_dispatcher { + d.emit_sync_error(&msg); + } let _ = channel.leave().await; - - let ts = chrono::Utc::now().timestamp(); - let sig = Self::create_hmac_signature(api_secret, ts, email, api_key, ""); - let canonical_payload = - Self::build_join_payload(email, api_key, &sig, ts, Some(canonical)); - - let canonical_channel = socket - .channel( - Topic::from_string(format!("sync:user:{}", canonical)), - Some(to_payload(&canonical_payload)?), - ) - .await - .map_err(|e| ws_err(format!("Canonical channel create failed: {:?}", e)))?; - - canonical_channel.join(JOIN_TIMEOUT).await.map_err(|e| { - if let Some(ref d) = event_dispatcher { - d.emit_sync_error(&format!("Canonical join failed: {:?}", e)); - } - ws_err(format!("Canonical join failed: {:?}", e)) - })?; - - (canonical_channel, canonical) - } else { - (channel, user_id) - }; + return Err(ws_err(msg)); + } // Join public channel for public document events let public_channel = socket @@ -191,18 +142,8 @@ impl WebSocketClient { } let (tx, rx) = mpsc::channel::(100); - Self::setup_broadcast_handlers( - &channel, - tx.clone(), - effective_user_id, - is_connected.clone(), - ); - Self::setup_broadcast_handlers( - &public_channel, - tx.clone(), - effective_user_id, - is_connected, - ); + Self::setup_broadcast_handlers(&channel, tx.clone(), is_connected.clone()); + Self::setup_broadcast_handlers(&public_channel, tx.clone(), is_connected); // Emit auth success let _ = tx @@ -217,42 +158,48 @@ impl WebSocketClient { channel, _public_channel: public_channel, tx, - user_id: effective_user_id, }, WebSocketReceiver { rx }, - adoption, )) } - /// Build the channel join payload. `user_id` is included only for a - /// steady-state (already-adopted) join; a bootstrap join omits it. + /// Build the channel join payload; always carries the adopted user id. pub(crate) fn build_join_payload( email: &str, api_key: &str, signature: &str, timestamp: i64, - user_id: Option, + user_id: Uuid, ) -> Value { - let mut payload = json!({ + json!({ "email": email, "api_key": api_key, "signature": signature, "timestamp": timestamp, - }); - if let Some(uid) = user_id { - payload["user_id"] = json!(uid.to_string()); - } - payload + "user_id": user_id.to_string(), + }) } - /// The canonical id to adopt, if the join reply carries a `user_id` that - /// differs from the local one. `None` means no adoption is needed. - pub(crate) fn should_adopt(local: Uuid, reply: &Value) -> Option { - let canonical = reply - .get("user_id") - .and_then(|v| v.as_str()) - .and_then(|s| Uuid::parse_str(s).ok())?; - (canonical != local).then_some(canonical) + /// `Ok` when the join reply's `user_id` (if present) matches ours; `Err` + /// with a description when it differs or cannot be parsed. A reply naming + /// a different user means local identity diverged from the account — the + /// caller must refuse to sync, never silently re-stamp. + pub(crate) fn verify_reply_identity(local: Uuid, reply: &Value) -> Result<(), String> { + match reply.get("user_id").and_then(|v| v.as_str()) { + None => Ok(()), + Some(s) => match Uuid::parse_str(s) { + Ok(server_id) if server_id == local => Ok(()), + Ok(server_id) => Err(format!( + "identity drift: server reports user {} but local identity is {}; \ + refusing to sync", + server_id, local + )), + Err(_) => Err(format!( + "identity drift: server sent unparseable user_id {:?}; refusing to sync", + s + )), + }, + } } fn to_websocket_url(server_url: &str) -> SyncResult { @@ -273,7 +220,6 @@ impl WebSocketClient { fn setup_broadcast_handlers( channel: &Arc, tx: mpsc::Sender, - user_id: Uuid, is_connected: Arc, ) { let events = channel.events(); @@ -289,9 +235,7 @@ impl WebSocketClient { match event_name.as_str() { "document_created" => { - if let Some(doc) = payload_json - .as_ref() - .and_then(|j| json_to_document(j, user_id)) + if let Some(doc) = payload_json.as_ref().and_then(json_to_document) { let _ = tx_clone .send(ServerMessage::DocumentCreated { document: doc }) @@ -444,7 +388,7 @@ impl WebSocketClient { Ok(j) => { if let Some(docs) = j.get("documents").and_then(|v| v.as_array()) { for doc_json in docs { - if let Some(document) = json_to_document(doc_json, self.user_id) { + if let Some(document) = json_to_document(doc_json) { let _ = self.tx.send(ServerMessage::SyncDocument { document }).await; } } @@ -579,12 +523,13 @@ fn payload_to_value(p: &Payload) -> Option { } } -fn json_to_document(j: &Value, default_user_id: Uuid) -> Option { - // Use server-provided user_id if present (null = public doc), otherwise fall back to default - let user_id = if let Some(uid_value) = j.get("user_id") { - uid_value.as_str().and_then(|s| Uuid::parse_str(s).ok()) - } else { - Some(default_user_id) +fn json_to_document(j: &Value) -> Option { + // The server always carries ownership in the sync envelope: a string + // user_id (owned doc) or null (public doc). A payload missing the key + // entirely is malformed — reject it rather than guess an owner. + let user_id = match j.get("user_id")? { + Value::Null => None, + uid_value => Some(Uuid::parse_str(uid_value.as_str()?).ok()?), }; Some(Document { id: Uuid::parse_str(j.get("id")?.as_str()?).ok()?, @@ -663,7 +608,7 @@ mod tests { "visibility": "public", "provenance": {"copied_from": "x"} }); - let doc = json_to_document(&j, uuid::Uuid::nil()).unwrap(); + let doc = json_to_document(&j).unwrap(); assert_eq!(doc.author_name.as_deref(), Some("Sevish")); assert_eq!(doc.visibility.as_deref(), Some("public")); assert!(doc.provenance.is_some()); @@ -673,50 +618,63 @@ mod tests { fn json_to_document_tolerates_missing_attribution() { let j = serde_json::json!({ "id": "71b2b712-7878-56ee-8323-43809b8198a5", + "user_id": null, "content": {"title": "T"}, "sync_revision": 1 }); - let doc = json_to_document(&j, uuid::Uuid::nil()).unwrap(); + let doc = json_to_document(&j).unwrap(); + assert_eq!(doc.user_id, None, "null user_id means a public document"); assert_eq!(doc.author_name, None); assert_eq!(doc.visibility, None); } #[test] - fn provisional_join_payload_omits_user_id() { - let payload = WebSocketClient::build_join_payload("a@b.com", "key", "sig", 123, None); - assert!(payload.get("user_id").is_none()); - assert_eq!(payload["email"], "a@b.com"); + fn json_to_document_rejects_payload_without_user_id_key() { + let j = serde_json::json!({ + "id": "71b2b712-7878-56ee-8323-43809b8198a5", + "content": {"title": "T"}, + "sync_revision": 1 + }); + assert!( + json_to_document(&j).is_none(), + "ownership must come from the envelope, never be guessed" + ); } #[test] - fn adopted_join_payload_includes_user_id() { + fn join_payload_always_includes_user_id() { let uid = Uuid::new_v4(); - let payload = WebSocketClient::build_join_payload("a@b.com", "key", "sig", 123, Some(uid)); + let payload = WebSocketClient::build_join_payload("a@b.com", "key", "sig", 123, uid); assert_eq!(payload["user_id"], uid.to_string()); + assert_eq!(payload["email"], "a@b.com"); } #[test] - fn should_adopt_detects_canonical_mismatch() { + fn reply_identity_matching_id_is_ok() { let local = Uuid::new_v4(); - let canonical = Uuid::new_v4(); - let reply = serde_json::json!({ "user_id": canonical.to_string(), "email": "a@b.com" }); - assert_eq!( - WebSocketClient::should_adopt(local, &reply), - Some(canonical) - ); + let reply = serde_json::json!({ "user_id": local.to_string() }); + assert!(WebSocketClient::verify_reply_identity(local, &reply).is_ok()); } #[test] - fn should_adopt_noop_when_ids_match() { + fn reply_identity_mismatch_is_drift_error() { let local = Uuid::new_v4(); - let reply = serde_json::json!({ "user_id": local.to_string() }); - assert_eq!(WebSocketClient::should_adopt(local, &reply), None); + let other = Uuid::new_v4(); + let reply = serde_json::json!({ "user_id": other.to_string() }); + assert!(WebSocketClient::verify_reply_identity(local, &reply).is_err()); } #[test] - fn should_adopt_noop_when_reply_has_no_user_id() { + fn reply_identity_absent_is_tolerated() { let local = Uuid::new_v4(); let reply = serde_json::json!({ "email": "a@b.com" }); - assert_eq!(WebSocketClient::should_adopt(local, &reply), None); + assert!(WebSocketClient::verify_reply_identity(local, &reply).is_ok()); + } + + #[test] + fn reply_identity_garbage_is_drift_error() { + let local = Uuid::new_v4(); + let reply = serde_json::json!({ "user_id": "not-a-uuid" }); + assert!(WebSocketClient::verify_reply_identity(local, &reply).is_err()); } } diff --git a/replicant-client/tests/constructor_identity_tests.rs b/replicant-client/tests/constructor_identity_tests.rs new file mode 100644 index 0000000..1ce1ebb --- /dev/null +++ b/replicant-client/tests/constructor_identity_tests.rs @@ -0,0 +1,76 @@ +//! Constructor-time identity adoption: the canonical user id (from stored +//! credentials) is adopted during `Client` construction, before any WebSocket +//! work and before the handle is returned — so adoption can never race live +//! document creation. + +use replicant_client::Client; +use uuid::Uuid; + +/// Unreachable server: connect fails fast, keeping tests offline. +const DEAD_SERVER: &str = "http://127.0.0.1:1"; + +fn temp_db_url() -> String { + let path = std::env::temp_dir().join(format!("replicant-ctor-test-{}.sqlite3", Uuid::new_v4())); + format!("sqlite://{}?mode=rwc", path.display()) +} + +async fn open_offline(db_url: &str, canonical: Option) -> replicant_core::SyncResult { + Client::new(db_url, DEAD_SERVER, "ctor@test.com", "", "", canonical).await +} + +#[tokio::test] +async fn constructor_adopts_canonical_id_before_returning() { + let db_url = temp_db_url(); + + // First run: offline, no canonical id → provisional identity. + let client = open_offline(&db_url, None).await.unwrap(); + let provisional = client.user_id(); + let doc = client + .create_document(serde_json::json!({"title": "offline tuning"})) + .await + .unwrap(); + assert_eq!(doc.user_id, Some(provisional)); + drop(client); + + // Second run: credentials arrived with a canonical id → constructor adopts. + let canonical = Uuid::new_v4(); + let client = open_offline(&db_url, Some(canonical)).await.unwrap(); + assert_eq!(client.user_id(), canonical); + + let docs = client.get_all_documents().await.unwrap(); + let restamped = docs.iter().find(|d| d.id == doc.id).unwrap(); + assert_eq!( + restamped.user_id, + Some(canonical), + "offline-created document must be re-stamped to the canonical id" + ); +} + +#[tokio::test] +async fn constructor_rejects_account_switch() { + let db_url = temp_db_url(); + let canonical = Uuid::new_v4(); + + let client = open_offline(&db_url, Some(canonical)).await.unwrap(); + assert_eq!(client.user_id(), canonical); + drop(client); + + // A different canonical id on an already-adopted install is an account + // switch — refuse rather than re-stamp another account's documents. + let other_account = Uuid::new_v4(); + let result = open_offline(&db_url, Some(other_account)).await; + assert!(result.is_err(), "account switch must be rejected"); +} + +#[tokio::test] +async fn constructor_same_id_after_adoption_is_noop() { + let db_url = temp_db_url(); + let canonical = Uuid::new_v4(); + + let client = open_offline(&db_url, Some(canonical)).await.unwrap(); + drop(client); + + // Same canonical id again (e.g. credential rotation): proceed normally. + let client = open_offline(&db_url, Some(canonical)).await.unwrap(); + assert_eq!(client.user_id(), canonical); +} diff --git a/replicant-client/tests/phoenix_integration/identity_adoption_test.rs b/replicant-client/tests/phoenix_integration/identity_adoption_test.rs index a467a39..8173260 100644 --- a/replicant-client/tests/phoenix_integration/identity_adoption_test.rs +++ b/replicant-client/tests/phoenix_integration/identity_adoption_test.rs @@ -1,18 +1,28 @@ -//! End-to-end identity adoption against a live server. +//! End-to-end claim-time identity adoption against a live server. //! //! Pre-seeds a provisional (random v4) identity plus a document owned by it, -//! then constructs the real `Client` pointed at the deployed server. Connecting -//! runs the join-reply → adoption path, which must re-stamp the document to the -//! server's canonical id, flip `identity_adopted`, and update the in-memory id. +//! then constructs the real `Client` with the canonical user id (as delivered +//! by enrollment claim; in CI it is seeded and exported alongside the API +//! credentials). Construction must adopt BEFORE connecting: re-stamp the +//! document, flip `identity_adopted`, switch the in-memory id — and then join +//! `sync:user:` successfully, passing the join-reply drift check. //! //! Gated behind `RUN_INTEGRATION_TESTS` (see `skip_if_no_server`); needs -//! `SYNC_SERVER_URL`, `REPLICANT_API_KEY`, `REPLICANT_API_SECRET`. +//! `SYNC_SERVER_URL`, `REPLICANT_API_KEY`, `REPLICANT_API_SECRET`, and +//! `REPLICANT_TEST_USER_ID` (the canonical id bound to those credentials). use super::{serial, server_url, skip_if_no_server, test_api_key, test_api_secret, TEST_EMAIL}; use replicant_client::{Client, ClientDatabase}; use sqlx::Row; +use std::time::Duration; use uuid::Uuid; +fn canonical_user_id_from_env() -> Option { + std::env::var("REPLICANT_TEST_USER_ID") + .ok() + .and_then(|s| Uuid::parse_str(&s).ok()) +} + fn temp_db_path() -> String { let nanos = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -27,11 +37,14 @@ fn temp_db_path() -> String { #[tokio::test] #[serial] -async fn adoption_restamps_document_and_flips_flag_against_live_server() { +async fn claim_time_adoption_restamps_and_connects_to_canonical_topic() { if skip_if_no_server() { eprintln!("skipping identity adoption test: RUN_INTEGRATION_TESTS not set"); return; } + let Some(canonical) = canonical_user_id_from_env() else { + panic!("RUN_INTEGRATION_TESTS is set but REPLICANT_TEST_USER_ID is missing/invalid"); + }; std::fs::create_dir_all("databases").ok(); let db_file = temp_db_path(); @@ -51,6 +64,7 @@ async fn adoption_restamps_document_and_flips_flag_against_live_server() { Some(uuid::Version::Random), "seeded id should be a random v4 provisional id" ); + assert_ne!(provisional, canonical); sqlx::query("INSERT INTO documents (id, user_id, content) VALUES (?1, ?2, ?3)") .bind(doc_id.to_string()) @@ -61,27 +75,35 @@ async fn adoption_restamps_document_and_flips_flag_against_live_server() { .unwrap(); } - // 2. Construct the real client on the same db, pointed at the live server. - // Adoption runs synchronously inside construction, on first join reply. + // 2. Construct the real client with the canonical id (claim-time adoption). + // Adoption runs inside construction, before any WebSocket work. let client = Client::with_event_dispatcher( &db_url, &server_url(), TEST_EMAIL, &test_api_key(), &test_api_secret(), + Some(canonical), None, ) .await - .expect("client should connect to the live server"); + .expect("client should adopt and connect to the live server"); - // 3. The in-memory id switched to the server's canonical id. - let canonical = client.user_id(); - assert_ne!( - canonical, provisional, - "client should have adopted the server's canonical id" - ); + // 3. The in-memory id switched to the canonical id. + assert_eq!(client.user_id(), canonical); + + // 4. The join to sync:user: succeeded (drift check passed). + let mut connected = client.is_connected(); + for _ in 0..50 { + if connected { + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + connected = client.is_connected(); + } + assert!(connected, "client should join sync:user:"); - // 4. The DB reflects the adoption and the pre-existing doc was re-stamped. + // 5. The DB reflects the adoption and the pre-existing doc was re-stamped. let db = ClientDatabase::new(&db_url).await.unwrap(); let adopted: i64 = sqlx::query("SELECT identity_adopted FROM user_config LIMIT 1") .fetch_one(&db.pool) From ef7efb0a08886d9ed9dfb4b378a6b3988373eef2 Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 13 Jul 2026 16:00:13 +0100 Subject: [PATCH 11/16] chore(sync): log dropped ownerless document payloads --- replicant-client/src/websocket.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/replicant-client/src/websocket.rs b/replicant-client/src/websocket.rs index e27cffd..ba06649 100644 --- a/replicant-client/src/websocket.rs +++ b/replicant-client/src/websocket.rs @@ -527,7 +527,14 @@ fn json_to_document(j: &Value) -> Option { // The server always carries ownership in the sync envelope: a string // user_id (owned doc) or null (public doc). A payload missing the key // entirely is malformed — reject it rather than guess an owner. - let user_id = match j.get("user_id")? { + let Some(uid_value) = j.get("user_id") else { + tracing::warn!( + "dropping document payload without ownership envelope (id: {:?})", + j.get("id") + ); + return None; + }; + let user_id = match uid_value { Value::Null => None, uid_value => Some(Uuid::parse_str(uid_value.as_str()?).ok()?), }; From d7a54ab95c54f37c5fc7bfaf90b9dff3d8aed0f9 Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 13 Jul 2026 16:02:47 +0100 Subject: [PATCH 12/16] test(offline): lock in credential-less local-only operation --- .../tests/constructor_identity_tests.rs | 98 +++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/replicant-client/tests/constructor_identity_tests.rs b/replicant-client/tests/constructor_identity_tests.rs index 1ce1ebb..75fada7 100644 --- a/replicant-client/tests/constructor_identity_tests.rs +++ b/replicant-client/tests/constructor_identity_tests.rs @@ -74,3 +74,101 @@ async fn constructor_same_id_after_adoption_is_noop() { let client = open_offline(&db_url, Some(canonical)).await.unwrap(); assert_eq!(client.user_id(), canonical); } + +/// Empty credentials must yield a fully usable local-only client: no +/// WebSocket connect attempt, no reconnect loop, full CRUD over the local +/// DB, and a clean drop with no hung background task. +#[tokio::test] +async fn open_without_credentials_is_local_only_and_usable() { + let result = tokio::time::timeout(std::time::Duration::from_secs(10), async { + let db_url = temp_db_url(); + let client = Client::new(&db_url, DEAD_SERVER, "offline@test.com", "", "", None) + .await + .unwrap(); + + assert!(!client.is_connected(), "no credentials must mean offline"); + + let doc = client + .create_document(serde_json::json!({"title": "offline-only"})) + .await + .unwrap(); + + let docs = client.get_all_documents().await.unwrap(); + assert!(docs.iter().any(|d| d.id == doc.id)); + + client + .update_document(doc.id, serde_json::json!({"title": "offline-only-edited"})) + .await + .unwrap(); + let docs = client.get_all_documents().await.unwrap(); + let updated = docs.iter().find(|d| d.id == doc.id).unwrap(); + assert_eq!(updated.content["title"], "offline-only-edited"); + + client.delete_document(doc.id).await.unwrap(); + let docs = client.get_all_documents().await.unwrap(); + assert!(!docs.iter().any(|d| d.id == doc.id)); + + assert!( + !client.is_connected(), + "still offline after local CRUD activity" + ); + + drop(client); + }) + .await; + + assert!( + result.is_ok(), + "client with no credentials hung instead of completing/dropping cleanly" + ); +} + +/// Reviewer-flagged gap (Task-3 decision row 5): credentials are present but +/// the identity has never been adopted (canonical_user_id = None, never +/// adopted before). Sync must stay disabled — no reconnect-loop spam — even +/// though api_key/api_secret are non-empty. Uses the dead-server URL so even +/// a wrongly-attempted connection cannot succeed. +#[tokio::test] +async fn open_with_credentials_but_unadopted_identity_stays_local() { + let result = tokio::time::timeout(std::time::Duration::from_secs(10), async { + let db_url = temp_db_url(); + let client = Client::new( + &db_url, + DEAD_SERVER, + "unadopted@test.com", + "some-api-key", + "some-api-secret", + None, + ) + .await + .unwrap(); + + assert!( + !client.is_connected(), + "unadopted identity must not connect even with credentials present" + ); + + let doc = client + .create_document(serde_json::json!({"title": "unadopted-but-local"})) + .await + .unwrap(); + let docs = client.get_all_documents().await.unwrap(); + assert!(docs.iter().any(|d| d.id == doc.id)); + + // Give a hypothetical reconnect loop time to spin and misbehave; it + // must not, since sync_enabled requires an adopted identity. + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + assert!( + !client.is_connected(), + "reconnection loop must not have connected/spammed while identity is unadopted" + ); + + drop(client); + }) + .await; + + assert!( + result.is_ok(), + "client with unadopted identity hung instead of completing/dropping cleanly" + ); +} From 5f9f002946cdbe85607d7bebb18338f6652f2444 Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 13 Jul 2026 16:10:23 +0100 Subject: [PATCH 13/16] test(offline): make sync-gating regression detectable with a counting listener --- .../tests/constructor_identity_tests.rs | 72 ++++++++++++++++--- 1 file changed, 62 insertions(+), 10 deletions(-) diff --git a/replicant-client/tests/constructor_identity_tests.rs b/replicant-client/tests/constructor_identity_tests.rs index 75fada7..6d8333a 100644 --- a/replicant-client/tests/constructor_identity_tests.rs +++ b/replicant-client/tests/constructor_identity_tests.rs @@ -4,9 +4,13 @@ //! document creation. use replicant_client::Client; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; use uuid::Uuid; -/// Unreachable server: connect fails fast, keeping tests offline. +/// Unreachable server: connect fails fast, keeping tests offline. Only used +/// by tests that don't need to prove "zero connection attempts" (a dead +/// address fails identically whether or not the sync gate ran). const DEAD_SERVER: &str = "http://127.0.0.1:1"; fn temp_db_url() -> String { @@ -14,6 +18,31 @@ fn temp_db_url() -> String { format!("sqlite://{}?mode=rwc", path.display()) } +/// Binds a local listener and counts accepted connections, so tests can +/// assert "no connection attempt was made" instead of relying on a dead +/// address (which fails identically whether or not the gate ran). Returns +/// the reachable server URL and the shared counter. +async fn counting_listener() -> (String, Arc) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind local listener"); + let addr = listener.local_addr().expect("local addr"); + let count = Arc::new(AtomicUsize::new(0)); + let count_clone = Arc::clone(&count); + tokio::spawn(async move { + loop { + match listener.accept().await { + Ok((socket, _)) => { + count_clone.fetch_add(1, Ordering::SeqCst); + drop(socket); + } + Err(_) => break, + } + } + }); + (format!("http://{}", addr), count) +} + async fn open_offline(db_url: &str, canonical: Option) -> replicant_core::SyncResult { Client::new(db_url, DEAD_SERVER, "ctor@test.com", "", "", canonical).await } @@ -76,13 +105,20 @@ async fn constructor_same_id_after_adoption_is_noop() { } /// Empty credentials must yield a fully usable local-only client: no -/// WebSocket connect attempt, no reconnect loop, full CRUD over the local -/// DB, and a clean drop with no hung background task. +/// connection attempt against the sync server, and full CRUD over the local +/// DB. The 10s timeout is a suite-hang safety net only — it does not prove +/// the client dropped cleanly (there is no `Drop` impl on `Client` and any +/// spawned task is detached, so a hung reconnect task would not fail this +/// wrapper). What actually falsifies a regressed sync gate is the counting +/// listener below: if `sync_enabled` were wrongly true, the constructor's own +/// connect attempt would register on the counter within milliseconds. #[tokio::test] async fn open_without_credentials_is_local_only_and_usable() { + let (server_url, connections) = counting_listener().await; + let result = tokio::time::timeout(std::time::Duration::from_secs(10), async { let db_url = temp_db_url(); - let client = Client::new(&db_url, DEAD_SERVER, "offline@test.com", "", "", None) + let client = Client::new(&db_url, &server_url, "offline@test.com", "", "", None) .await .unwrap(); @@ -119,22 +155,32 @@ async fn open_without_credentials_is_local_only_and_usable() { assert!( result.is_ok(), - "client with no credentials hung instead of completing/dropping cleanly" + "test did not complete within the 10s suite-hang safety timeout" + ); + + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + assert_eq!( + connections.load(Ordering::SeqCst), + 0, + "no connection attempt must be made against the sync server without credentials" ); } /// Reviewer-flagged gap (Task-3 decision row 5): credentials are present but /// the identity has never been adopted (canonical_user_id = None, never -/// adopted before). Sync must stay disabled — no reconnect-loop spam — even -/// though api_key/api_secret are non-empty. Uses the dead-server URL so even -/// a wrongly-attempted connection cannot succeed. +/// adopted before). Sync must stay disabled — no connection attempt at all — +/// even though api_key/api_secret are non-empty. Uses a real local listener +/// (not the dead-server address) so a wrongly-attempted connection would +/// actually register instead of merely failing to connect either way. #[tokio::test] async fn open_with_credentials_but_unadopted_identity_stays_local() { + let (server_url, connections) = counting_listener().await; + let result = tokio::time::timeout(std::time::Duration::from_secs(10), async { let db_url = temp_db_url(); let client = Client::new( &db_url, - DEAD_SERVER, + &server_url, "unadopted@test.com", "some-api-key", "some-api-secret", @@ -169,6 +215,12 @@ async fn open_with_credentials_but_unadopted_identity_stays_local() { assert!( result.is_ok(), - "client with unadopted identity hung instead of completing/dropping cleanly" + "test did not complete within the 10s suite-hang safety timeout" + ); + + assert_eq!( + connections.load(Ordering::SeqCst), + 0, + "no connection attempt must be made against the sync server while identity is unadopted" ); } From e04136d5d56c81e19d840fdedc647e0d86be3bfe Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 13 Jul 2026 16:11:55 +0100 Subject: [PATCH 14/16] fix(seed): update Client::new call for canonical_user_id parameter --- replicant-seed/src/main.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/replicant-seed/src/main.rs b/replicant-seed/src/main.rs index 145b940..37de08b 100644 --- a/replicant-seed/src/main.rs +++ b/replicant-seed/src/main.rs @@ -198,6 +198,7 @@ async fn main() -> Result<()> { &args.user, &args.api_key, &args.api_secret, + None, ) .await .map_err(|e| anyhow!("Replicant connect failed: {}", e))?; From 4086df3b2da8da4d91efebb3eedbe50d5b3280dc Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 13 Jul 2026 17:01:06 +0100 Subject: [PATCH 15/16] ci: boot stripped server with test endpoint; seed enrolled credentials --- .github/workflows/tests.yml | 72 ++------- .gitignore | 3 + .../credential_enrollment_test.rs | 46 ++++++ .../tests/phoenix_integration/mod.rs | 14 +- test/run_phoenix_interop_local.sh | 147 +++++++++++++----- 5 files changed, 188 insertions(+), 94 deletions(-) create mode 100644 replicant-client/tests/phoenix_integration/credential_enrollment_test.rs diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 68dc89d..367c67f 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -88,71 +88,27 @@ jobs: path: target key: ${{ runner.os }}-cargo-build-target-${{ hashFiles('**/Cargo.lock') }} - # Set up Elixir/Phoenix server - - name: Clone Phoenix server - run: git clone --depth 1 https://github.com/replicant-sync/replicant-server.git /tmp/replicant-server - - name: Set up Elixir uses: erlef/setup-beam@v1 with: elixir-version: '1.17' otp-version: '27' - - name: Cache Elixir deps - uses: actions/cache@v3 - with: - path: /tmp/replicant-server/deps - key: ${{ runner.os }}-mix-${{ hashFiles('/tmp/replicant-server/mix.lock') }} - - - name: Install Phoenix dependencies - working-directory: /tmp/replicant-server - run: mix deps.get - - - name: Set up Phoenix database - working-directory: /tmp/replicant-server - env: - DATABASE_URL: postgres://postgres:postgres@localhost:5432/replicant_server_test - SECRET_KEY_BASE: test-secret-key-base-that-is-at-least-64-characters-long-for-testing - run: | - mix ecto.create - mix ecto.migrate - - - name: Generate API credentials - id: credentials - working-directory: /tmp/replicant-server - env: - DATABASE_URL: postgres://postgres:postgres@localhost:5432/replicant_server_test - SECRET_KEY_BASE: test-secret-key-base-that-is-at-least-64-characters-long-for-testing - run: | - OUTPUT=$(mix replicant.gen.credentials --name "CI Test") - API_KEY=$(echo "$OUTPUT" | grep "API Key:" | awk '{print $3}') - API_SECRET=$(echo "$OUTPUT" | grep "Secret:" | awk '{print $2}') - echo "api_key=$API_KEY" >> $GITHUB_OUTPUT - echo "api_secret=$API_SECRET" >> $GITHUB_OUTPUT - - - name: Start Phoenix server - working-directory: /tmp/replicant-server - env: - DATABASE_URL: postgres://postgres:postgres@localhost:5432/replicant_server_test - SECRET_KEY_BASE: test-secret-key-base-that-is-at-least-64-characters-long-for-testing - run: | - mix phx.server & - for i in $(seq 1 60); do - if curl -fs http://localhost:4000/health >/dev/null; then - echo "Server is up" - exit 0 - fi - sleep 1 - done - echo "Server did not come up within 60s" - exit 1 - - - name: Run integration tests + # The harness boots the stripped server (a library: no HTTP server of its + # own) behind a minimal endpoint that mounts ReplicantServer.Sync.Socket on + # :4000, seeds an enrolled user + credential (and one legacy nil-user + # credential for the negative test), and runs the gated integration suite. + # It clones the server itself at the pinned SERVER_REF, so no separate + # checkout/credential/health steps are needed here. + - name: Boot stripped server + run integration suite env: - RUN_INTEGRATION_TESTS: 1 - REPLICANT_API_KEY: ${{ steps.credentials.outputs.api_key }} - REPLICANT_API_SECRET: ${{ steps.credentials.outputs.api_secret }} - run: cargo test --package replicant-client --test integration -- --nocapture + REPLICANT_SERVER_REF: '332a8ba' # origin/main (PR #7 merged: claim -> user_id) + REPLICANT_SERVER_DIR: /tmp/replicant-server-interop + INTEROP_DB_HOST: localhost + INTEROP_DB_USER: postgres + INTEROP_DB_PASS: postgres + INTEROP_DB_NAME: replicant_server_test + run: ./test/run_phoenix_interop_local.sh lint: name: Lint diff --git a/.gitignore b/.gitignore index cc9ce76..6fded76 100644 --- a/.gitignore +++ b/.gitignore @@ -52,3 +52,6 @@ target-docker/ # Development test scripts test_*.sh + +# Claude local session/handoff files +.claude/ diff --git a/replicant-client/tests/phoenix_integration/credential_enrollment_test.rs b/replicant-client/tests/phoenix_integration/credential_enrollment_test.rs new file mode 100644 index 0000000..b954e75 --- /dev/null +++ b/replicant-client/tests/phoenix_integration/credential_enrollment_test.rs @@ -0,0 +1,46 @@ +//! Negative enrollment guard: a credential with no bound user (the legacy +//! shared-style secret minted via `ReplicantServer.Auth.create_credential/1`, +//! which leaves `user_id` nil) must be REJECTED at channel join. +//! +//! Post-#6 the server resolves identity from the credential's `user_id`; a nil +//! user cannot resolve identity, so join is refused with `credential_not_enrolled` +//! (server: `Sync.Channel.require_enrolled/1`, locked by f4d37db) before any +//! topic check. This asserts that guard end-to-end. +//! +//! Gated behind `RUN_INTEGRATION_TESTS`. The harness seeds the legacy credential +//! and exports `REPLICANT_LEGACY_API_KEY` / `REPLICANT_LEGACY_API_SECRET`. + +use super::{serial, skip_if_no_server, TestClient, TEST_EMAIL}; + +fn legacy_credentials() -> Option<(String, String)> { + let key = std::env::var("REPLICANT_LEGACY_API_KEY").ok()?; + let secret = std::env::var("REPLICANT_LEGACY_API_SECRET").ok()?; + Some((key, secret)) +} + +#[tokio::test] +#[serial] +async fn test_unenrolled_credential_rejected_at_join() { + if skip_if_no_server() { + return; + } + + let Some((legacy_key, legacy_secret)) = legacy_credentials() else { + panic!( + "RUN_INTEGRATION_TESTS is set but REPLICANT_LEGACY_API_KEY/SECRET are missing; \ + the harness must seed a nil-user credential and export them" + ); + }; + + let result = + TestClient::connect_with_credentials(TEST_EMAIL, &legacy_key, &legacy_secret).await; + + let err = result + .err() + .expect("join with an unenrolled (nil-user) credential must be rejected, but it connected"); + assert!( + err.contains("credential_not_enrolled"), + "expected credential_not_enrolled rejection, got: {}", + err + ); +} diff --git a/replicant-client/tests/phoenix_integration/mod.rs b/replicant-client/tests/phoenix_integration/mod.rs index a15abad..6b6ef56 100644 --- a/replicant-client/tests/phoenix_integration/mod.rs +++ b/replicant-client/tests/phoenix_integration/mod.rs @@ -22,6 +22,7 @@ mod basic_sync_test; mod conflict_test; +mod credential_enrollment_test; mod identity_adoption_test; mod live_sync_test; mod multi_client_test; @@ -60,6 +61,17 @@ pub fn server_url() -> String { .unwrap_or_else(|_| "ws://localhost:4000/socket/websocket".to_string()) } +/// The canonical user id bound to `REPLICANT_API_KEY`. Post-#6 the server +/// binds each credential to a user and rejects `sync:user:` joins whose +/// id does not match the credential's user (`topic_user_mismatch`), so the +/// test client must join under this seeded id, not a random one. +pub fn test_user_id() -> Uuid { + std::env::var("REPLICANT_TEST_USER_ID") + .ok() + .and_then(|s| Uuid::parse_str(&s).ok()) + .unwrap_or_else(Uuid::new_v4) +} + pub fn skip_if_no_server() -> bool { std::env::var("RUN_INTEGRATION_TESTS").is_err() } @@ -91,7 +103,7 @@ impl TestClient { api_secret: &str, ) -> Result { let url = Url::parse(&server_url()).map_err(|e| format!("Invalid URL: {}", e))?; - let user_id = Uuid::new_v4(); + let user_id = test_user_id(); let socket = Socket::spawn(url, None, None) .await diff --git a/test/run_phoenix_interop_local.sh b/test/run_phoenix_interop_local.sh index 6793654..567cef2 100755 --- a/test/run_phoenix_interop_local.sh +++ b/test/run_phoenix_interop_local.sh @@ -1,18 +1,22 @@ #!/bin/bash # -# Phoenix interop test runner (Rust client <-> Elixir server). +# Phoenix interop test runner (Rust client <-> stripped Elixir server). # -# Stands up the Elixir replicant-server against a throwaway clean database, -# generates HMAC API credentials, and runs the client's phoenix_integration -# suite against it. The clean DB proves the frozen-identity contract holds -# end-to-end (same email -> same user_id on both sides) without touching the -# developer's local dev database. +# The merged replicant-server (>= #6) is a LIBRARY: it ships the sync socket +# (`lib/replicant_server/sync/socket.ex`) but no endpoint/router/HTTP server of +# its own — production hosts (entonal-web-app) mount the socket in their own +# endpoint. This harness supplies that host: it boots a minimal endpoint +# (modelled on the server's `test/support/test_endpoint.ex`) that mounts +# `ReplicantServer.Sync.Socket` on :4000, seeds ONE enrolled user + credential +# (plus one legacy nil-user credential for the negative test) by calling +# `ReplicantServer.Auth` directly, and runs the client's phoenix_integration +# suite against it on a throwaway clean database. # # Usage: # test/run_phoenix_interop_local.sh # run the full suite # test/run_phoenix_interop_local.sh test_name # run a single test filter # -# Requires: a running PostgreSQL, Elixir/mix, and cargo. +# Requires: a running PostgreSQL, Elixir/mix, cargo, and curl. set -euo pipefail @@ -20,19 +24,36 @@ set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" CLIENT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" CLIENT_CRATE="$CLIENT_ROOT/replicant-client" -SERVER_DIR="${REPLICANT_SERVER_DIR:-$CLIENT_ROOT/../replicant-server}" + +# --- Server pin -------------------------------------------------------------- +# PINNED to origin/main of replicant-server. This SHA must include the +# claim_enrollment -> user_id change (PR #7, merged as 332a8ba) so the seed can +# export REPLICANT_TEST_USER_ID. Re-pin this to a newer main SHA as the server +# advances. +SERVER_REF="${REPLICANT_SERVER_REF:-332a8ba}" + +# Where to prepare the server checkout. Locally we make a detached git worktree +# from a sibling replicant-server clone; in CI (no local clone) we git-clone. +SERVER_SRC="${REPLICANT_SERVER_SRC:-$CLIENT_ROOT/../replicant-server}" +SERVER_CLONE_URL="${REPLICANT_SERVER_CLONE_URL:-https://github.com/replicant-sync/replicant-server.git}" +SERVER_DIR="${REPLICANT_SERVER_DIR:-/tmp/replicant-server-interop}" + +# Hex/Mix caches. Redirected to writable paths so `mix deps.get` can persist its +# registry cache even where $HOME/.hex is not writable (sandboxed shells). +export HEX_HOME="${INTEROP_HEX_HOME:-/tmp/replicant-interop-hex}" +export MIX_HOME="${INTEROP_MIX_HOME:-/tmp/replicant-interop-mix}" +mkdir -p "$HEX_HOME" "$MIX_HOME" # --- Configuration ----------------------------------------------------------- -DB_NAME="${INTEROP_DB_NAME:-replicant_interop_test}" +DB_NAME="${INTEROP_DB_NAME:-replicant_server_test}" DB_USER="${INTEROP_DB_USER:-postgres}" DB_PASS="${INTEROP_DB_PASS:-postgres}" DB_HOST="${INTEROP_DB_HOST:-localhost}" -SERVER_PORT="${INTEROP_SERVER_PORT:-4010}" +SERVER_PORT="${INTEROP_SERVER_PORT:-4000}" DATABASE_URL="ecto://$DB_USER:$DB_PASS@$DB_HOST/$DB_NAME" SERVER_LOG="${INTEROP_SERVER_LOG:-/tmp/replicant_interop_server.log}" - -# Basic-auth vars are only required when PHX_SERVER is set (releases). Dev-mode -# `mix phx.server` does not set it, so the web UI guard stays inactive here. +TEST_EMAIL="${INTEROP_TEST_EMAIL:-integration-test@example.com}" +export MIX_ENV=test RED='\033[0;31m'; GREEN='\033[0;32m'; YELLOW='\033[1;33m'; NC='\033[0m' log() { echo -e "${GREEN}[$(date +'%H:%M:%S')] $1${NC}"; } @@ -47,7 +68,6 @@ cleanup() { sleep 1 kill -9 "$SERVER_PID" 2>/dev/null || true fi - # Sweep any stragglers bound to the interop port. local pids pids="$(lsof -ti :"$SERVER_PORT" 2>/dev/null || true)" [ -n "$pids" ] && kill -9 $pids 2>/dev/null || true @@ -55,48 +75,102 @@ cleanup() { trap cleanup EXIT INT TERM # --- Preflight --------------------------------------------------------------- -[ -d "$SERVER_DIR" ] || { err "Server dir not found: $SERVER_DIR (set REPLICANT_SERVER_DIR)"; exit 1; } command -v mix >/dev/null || { err "mix not found on PATH"; exit 1; } command -v cargo >/dev/null || { err "cargo not found on PATH"; exit 1; } +command -v curl >/dev/null || { err "curl not found on PATH"; exit 1; } if ! PGPASSWORD="$DB_PASS" psql -U "$DB_USER" -h "$DB_HOST" -d postgres -c "SELECT 1;" >/dev/null 2>&1; then err "PostgreSQL not reachable as $DB_USER@$DB_HOST"; exit 1 fi -# Free the port before we start. existing="$(lsof -ti :"$SERVER_PORT" 2>/dev/null || true)" [ -n "$existing" ] && { warn "Killing processes on port $SERVER_PORT: $existing"; kill -9 $existing 2>/dev/null || true; sleep 1; } +# --- Prepare pinned server checkout ------------------------------------------ +if [ -d "$SERVER_DIR/.git" ] || [ -f "$SERVER_DIR/.git" ]; then + log "Reusing server checkout at $SERVER_DIR (pinning to $SERVER_REF)" + git -C "$SERVER_DIR" checkout --quiet --detach "$SERVER_REF" 2>/dev/null || \ + git -C "$SERVER_DIR" checkout --quiet "$SERVER_REF" +elif [ -d "$SERVER_SRC/.git" ]; then + log "Creating detached worktree at $SERVER_DIR from $SERVER_SRC @ $SERVER_REF" + git -C "$SERVER_SRC" worktree prune + git -C "$SERVER_SRC" worktree add --force --detach "$SERVER_DIR" "$SERVER_REF" +else + log "Cloning $SERVER_CLONE_URL into $SERVER_DIR @ $SERVER_REF" + git clone "$SERVER_CLONE_URL" "$SERVER_DIR" + git -C "$SERVER_DIR" checkout "$SERVER_REF" +fi + +# The stripped server ships no HTTP adapter dependency (its own channel tests run +# with `server: false`). Inject Bandit so the endpoint can actually bind a WS +# port — mirroring what a production host app brings. Harness-local only. +if ! grep -q ':bandit' "$SERVER_DIR/mix.exs"; then + log "Injecting Bandit HTTP adapter dependency (harness-only)" + perl -0pi -e 's/(\{:jsonpatch,\s*"[^"]*"\})/$1,\n {:bandit, "~> 1.0"}/' "$SERVER_DIR/mix.exs" + grep -q ':bandit' "$SERVER_DIR/mix.exs" || { err "Failed to inject bandit dep"; exit 1; } +fi + +# --- Build server ------------------------------------------------------------ +log "Fetching + compiling server deps (MIX_ENV=test)" +( cd "$SERVER_DIR" && mix deps.get >/dev/null && mix compile >/dev/null ) + # --- Clean database ---------------------------------------------------------- log "Recreating clean database '$DB_NAME'" -export DATABASE_URL MIX_ENV=dev +export DATABASE_URL ( cd "$SERVER_DIR" mix ecto.drop --quiet 2>/dev/null || true mix ecto.create --quiet - mix ecto.migrate ) + mix ecto.migrate >/dev/null ) + +# --- Seed credentials -------------------------------------------------------- +# One enrolled user+credential (bound user_id) via the enrollment flow, and one +# legacy nil-user credential via create_credential/1 for the negative test. +log "Seeding enrolled + legacy credentials for $TEST_EMAIL" +SEED_LINE="$( cd "$SERVER_DIR" && TEST_EMAIL="$TEST_EMAIL" mix run -e ' + Ecto.Adapters.SQL.Sandbox.mode(ReplicantServer.Repo, :auto) + email = System.get_env("TEST_EMAIL") + {:ok, token} = ReplicantServer.Auth.request_enrollment(email) + {:ok, creds} = ReplicantServer.Auth.claim_enrollment(email, token) + {:ok, legacy} = ReplicantServer.Auth.create_credential("interop-legacy-shared") + IO.puts("SEED #{creds.api_key} #{creds.secret} #{creds.user_id} #{legacy.api_key} #{legacy.secret}") +' 2>/dev/null | grep '^SEED ' )" +read -r _ API_KEY API_SECRET TEST_USER_ID LEGACY_API_KEY LEGACY_API_SECRET <<<"$SEED_LINE" +[ -n "$API_KEY" ] && [ -n "$API_SECRET" ] && [ -n "$TEST_USER_ID" ] && \ +[ -n "$LEGACY_API_KEY" ] && [ -n "$LEGACY_API_SECRET" ] || { + err "Failed to seed credentials"; echo "$SEED_LINE"; exit 1; } +log "Enrolled user_id=$TEST_USER_ID" -# --- Start server ------------------------------------------------------------ -log "Starting Elixir server on port $SERVER_PORT (log: $SERVER_LOG)" +# --- Start server (minimal endpoint mounting the sync socket) ---------------- +BOOT_SCRIPT="$(mktemp /tmp/replicant_interop_boot.XXXXXX.exs)" +cat > "$BOOT_SCRIPT" < "$SERVER_LOG" -( cd "$SERVER_DIR" && PORT="$SERVER_PORT" DATABASE_URL="$DATABASE_URL" MIX_ENV=dev \ - exec mix phx.server ) >> "$SERVER_LOG" 2>&1 & +( cd "$SERVER_DIR" && exec mix run --no-halt "$BOOT_SCRIPT" ) >> "$SERVER_LOG" 2>&1 & SERVER_PID=$! -# Wait for the port to accept connections. +# Health-check: wait for the socket port to accept HTTP connections. The stripped +# server has no /health route, so any HTTP response (curl exit 0) means "up". for i in $(seq 1 60); do - if lsof -ti :"$SERVER_PORT" >/dev/null 2>&1; then break; fi + if curl -s -o /dev/null --max-time 2 "http://127.0.0.1:$SERVER_PORT/" 2>/dev/null; then + log "Server is up"; break + fi if ! kill -0 "$SERVER_PID" 2>/dev/null; then err "Server exited early:"; tail -30 "$SERVER_LOG"; exit 1; fi sleep 1 [ "$i" -eq 60 ] && { err "Server did not come up within 60s"; tail -30 "$SERVER_LOG"; exit 1; } done -log "Server is up" - -# --- Credentials ------------------------------------------------------------- -log "Generating API credentials" -CRED_OUTPUT="$( cd "$SERVER_DIR" && DATABASE_URL="$DATABASE_URL" MIX_ENV=dev \ - mix replicant.gen.credentials --name "integration-test" 2>/dev/null )" -API_KEY="$(echo "$CRED_OUTPUT" | grep -Eo 'rpa_[a-f0-9]+' | head -1)" -API_SECRET="$(echo "$CRED_OUTPUT" | grep -Eo 'rps_[a-f0-9]+' | head -1)" -[ -n "$API_KEY" ] && [ -n "$API_SECRET" ] || { err "Failed to parse credentials"; echo "$CRED_OUTPUT"; exit 1; } # --- Run interop suite ------------------------------------------------------- log "Running phoenix_integration suite against clean DB" @@ -105,6 +179,9 @@ set +e RUN_INTEGRATION_TESTS=1 \ REPLICANT_API_KEY="$API_KEY" \ REPLICANT_API_SECRET="$API_SECRET" \ + REPLICANT_TEST_USER_ID="$TEST_USER_ID" \ + REPLICANT_LEGACY_API_KEY="$LEGACY_API_KEY" \ + REPLICANT_LEGACY_API_SECRET="$LEGACY_API_SECRET" \ SYNC_SERVER_URL="ws://localhost:$SERVER_PORT/socket/websocket" \ cargo test --test integration ${1:+"$1"} -- --test-threads=1 ) test_exit=$? @@ -112,9 +189,9 @@ set -e echo "" if [ $test_exit -eq 0 ]; then - log "✅ Interop suite passed" + log "Interop suite passed" else - err "❌ Interop suite failed (server log tail below)" + err "Interop suite failed (server log tail below)" tail -40 "$SERVER_LOG" fi exit $test_exit From dd652b3e5ec30ab29bf813cd857bc7be90ddc7ad Mon Sep 17 00:00:00 2001 From: Adam Wilson Date: Mon, 13 Jul 2026 18:44:31 +0100 Subject: [PATCH 16/16] fix: address final-review findings for #37 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - interop harness: fold boot-script removal into cleanup() — the second trap EXIT was replacing the server-cleanup trap, leaking the booted server on :4000 after every successful run; seed step stderr now goes to the server log instead of /dev/null - client: user_id is write-once after construction — replace the RwLock with a plain field and correct the stale doc comment; document adopt_identity error conditions - replicant-seed: require --user-id (REPLICANT_USER_ID) and pass it as the canonical id — passing None left the client unadopted/local-only, so nothing was ever uploaded Claude-Session: https://claude.ai/code/session_01GwMbERgLK4KAuLrz4b4Liw --- Cargo.lock | 1 + replicant-client/src/client.rs | 20 +++++++++----------- replicant-client/src/database.rs | 6 ++++++ replicant-seed/Cargo.toml | 1 + replicant-seed/src/main.rs | 8 +++++++- test/run_phoenix_interop_local.sh | 11 ++++++----- 6 files changed, 30 insertions(+), 17 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 738e52b..10b020b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2104,6 +2104,7 @@ dependencies = [ "tokio", "tracing", "tracing-subscriber", + "uuid", "walkdir", ] diff --git a/replicant-client/src/client.rs b/replicant-client/src/client.rs index 9075692..87b8a7f 100644 --- a/replicant-client/src/client.rs +++ b/replicant-client/src/client.rs @@ -10,7 +10,7 @@ use sqlx::Row; use std::collections::HashMap; use std::sync::{ atomic::{AtomicBool, Ordering}, - Arc, RwLock, + Arc, }; use std::time::{Duration, Instant}; use tokio::sync::{mpsc, Mutex, Notify}; @@ -35,7 +35,7 @@ enum UploadType { pub struct Client { db: Arc, ws_client: Arc>>, - user_id: Arc>, + user_id: Uuid, client_id: Uuid, message_rx: Option>, event_dispatcher: Arc, @@ -168,7 +168,7 @@ impl Client { let mut engine = Self { db: db.clone(), ws_client: Arc::new(Mutex::new(ws_client)), - user_id: Arc::new(RwLock::new(user_id)), + user_id, client_id, message_rx: Some(rx), event_dispatcher: event_dispatcher.clone(), @@ -406,10 +406,11 @@ impl Client { self.create_document_with_id(Uuid::new_v4(), content).await } - /// Current server-authoritative user id. Interior-mutable so identity - /// adoption updates every subsequent local write and reconnect. + /// Server-authoritative user id, fixed for the client's lifetime. + /// Identity adoption runs in `Client::new` before this is read; + /// enrolling afterwards recreates the client. pub fn user_id(&self) -> Uuid { - *self.user_id.read().expect("user_id lock poisoned") + self.user_id } pub async fn create_document_with_id( @@ -1689,7 +1690,7 @@ impl Client { let api_key = self.api_key.clone(); let api_secret = self.api_secret.clone(); let client_id = self.client_id; - let user_id = self.user_id.clone(); + let user_id = self.user_id; let event_dispatcher = self.event_dispatcher.clone(); let db = self.db.clone(); let pending_uploads = self.pending_uploads.clone(); @@ -1720,15 +1721,12 @@ impl Client { server_url ); - // Copy the id out so no lock guard is held across the await. - let current_user_id = *user_id.read().expect("user_id lock poisoned"); - // Try to connect match WebSocketClient::connect( &server_url, &email, client_id, - current_user_id, + user_id, &api_key, &api_secret, Some(event_dispatcher.clone()), diff --git a/replicant-client/src/database.rs b/replicant-client/src/database.rs index 63c3533..017ff69 100644 --- a/replicant-client/src/database.rs +++ b/replicant-client/src/database.rs @@ -128,6 +128,12 @@ impl ClientDatabase { /// Atomically adopt the server's canonical id: re-stamp local documents /// owned by `old_id` and flip `user_config` to `canonical_id` with /// `identity_adopted = 1`. A crash mid-adoption leaves the old id intact. + /// + /// # Errors + /// + /// Returns [`SyncError::InvalidOperation`] if `canonical_id` is nil, equals + /// `old_id`, the identity was already adopted, or no `user_config` row + /// matches `old_id`; database failures surface as the underlying error. pub async fn adopt_identity(&self, old_id: Uuid, canonical_id: Uuid) -> SyncResult<()> { if canonical_id.is_nil() || old_id == canonical_id { return Err(SyncError::InvalidOperation( diff --git a/replicant-seed/Cargo.toml b/replicant-seed/Cargo.toml index 1134727..5bce950 100644 --- a/replicant-seed/Cargo.toml +++ b/replicant-seed/Cargo.toml @@ -14,6 +14,7 @@ tokio = { workspace = true } serde_json = { workspace = true } clap = { version = "4.4", features = ["derive", "env"] } anyhow = "1.0" +uuid = { workspace = true } walkdir = "2.5" tracing = { workspace = true } tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/replicant-seed/src/main.rs b/replicant-seed/src/main.rs index 37de08b..b1d6495 100644 --- a/replicant-seed/src/main.rs +++ b/replicant-seed/src/main.rs @@ -41,6 +41,12 @@ struct Args { #[arg(long, env = "REPLICANT_API_SECRET")] api_secret: String, + /// Canonical user id bound to the credential at enrollment. Required for + /// sync: without an adopted identity the client stays local-only and + /// pushes nothing. + #[arg(long, env = "REPLICANT_USER_ID")] + user_id: uuid::Uuid, + /// Directory containing the JSON documents to push (recursively). #[arg(long)] json_dir: PathBuf, @@ -198,7 +204,7 @@ async fn main() -> Result<()> { &args.user, &args.api_key, &args.api_secret, - None, + Some(args.user_id), ) .await .map_err(|e| anyhow!("Replicant connect failed: {}", e))?; diff --git a/test/run_phoenix_interop_local.sh b/test/run_phoenix_interop_local.sh index 567cef2..38eaf80 100755 --- a/test/run_phoenix_interop_local.sh +++ b/test/run_phoenix_interop_local.sh @@ -61,6 +61,7 @@ warn() { echo -e "${YELLOW}[$(date +'%H:%M:%S')] WARN: $1${NC}"; } err() { echo -e "${RED}[$(date +'%H:%M:%S')] ERROR: $1${NC}"; } SERVER_PID="" +BOOT_SCRIPT="" cleanup() { if [ -n "$SERVER_PID" ] && kill -0 "$SERVER_PID" 2>/dev/null; then log "Stopping server (PID $SERVER_PID)" @@ -71,6 +72,7 @@ cleanup() { local pids pids="$(lsof -ti :"$SERVER_PORT" 2>/dev/null || true)" [ -n "$pids" ] && kill -9 $pids 2>/dev/null || true + [ -n "$BOOT_SCRIPT" ] && rm -f "$BOOT_SCRIPT" } trap cleanup EXIT INT TERM @@ -124,7 +126,8 @@ export DATABASE_URL # --- Seed credentials -------------------------------------------------------- # One enrolled user+credential (bound user_id) via the enrollment flow, and one # legacy nil-user credential via create_credential/1 for the negative test. -log "Seeding enrolled + legacy credentials for $TEST_EMAIL" +log "Seeding enrolled + legacy credentials for $TEST_EMAIL (stderr: $SERVER_LOG)" +: > "$SERVER_LOG" SEED_LINE="$( cd "$SERVER_DIR" && TEST_EMAIL="$TEST_EMAIL" mix run -e ' Ecto.Adapters.SQL.Sandbox.mode(ReplicantServer.Repo, :auto) email = System.get_env("TEST_EMAIL") @@ -132,11 +135,11 @@ SEED_LINE="$( cd "$SERVER_DIR" && TEST_EMAIL="$TEST_EMAIL" mix run -e ' {:ok, creds} = ReplicantServer.Auth.claim_enrollment(email, token) {:ok, legacy} = ReplicantServer.Auth.create_credential("interop-legacy-shared") IO.puts("SEED #{creds.api_key} #{creds.secret} #{creds.user_id} #{legacy.api_key} #{legacy.secret}") -' 2>/dev/null | grep '^SEED ' )" +' 2>>"$SERVER_LOG" | grep '^SEED ' )" read -r _ API_KEY API_SECRET TEST_USER_ID LEGACY_API_KEY LEGACY_API_SECRET <<<"$SEED_LINE" [ -n "$API_KEY" ] && [ -n "$API_SECRET" ] && [ -n "$TEST_USER_ID" ] && \ [ -n "$LEGACY_API_KEY" ] && [ -n "$LEGACY_API_SECRET" ] || { - err "Failed to seed credentials"; echo "$SEED_LINE"; exit 1; } + err "Failed to seed credentials"; echo "$SEED_LINE"; tail -30 "$SERVER_LOG"; exit 1; } log "Enrolled user_id=$TEST_USER_ID" # --- Start server (minimal endpoint mounting the sync socket) ---------------- @@ -154,10 +157,8 @@ Ecto.Adapters.SQL.Sandbox.mode(ReplicantServer.Repo, :auto) IO.puts("ENDPOINT_STARTED") Process.sleep(:infinity) EOF -trap 'rm -f "$BOOT_SCRIPT"' EXIT log "Starting minimal endpoint on port $SERVER_PORT (log: $SERVER_LOG)" -: > "$SERVER_LOG" ( cd "$SERVER_DIR" && exec mix run --no-halt "$BOOT_SCRIPT" ) >> "$SERVER_LOG" 2>&1 & SERVER_PID=$!