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/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/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 44ab8fe..4ee3da8 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(); @@ -545,6 +541,7 @@ async fn main() -> Result<(), Box> { &user_email, &cli.api_key, &cli.api_secret, + None, ) .await { @@ -630,6 +627,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, + ); + } } }) { 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/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/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/client.rs b/replicant-client/src/client.rs index 132309e..87b8a7f 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,33 +106,65 @@ impl Client { let (reconnect_sync_tx, reconnect_sync_rx) = mpsc::channel(10); let is_connected = Arc::new(AtomicBool::new(false)); - // Try to connect to WebSocket, but don't fail if offline - let (ws_client, initial_ping_time) = 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())) + + // 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); + 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) + (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) + } + } + } else { + (None, None) }; + let mut engine = Self { db: db.clone(), ws_client: Arc::new(Mutex::new(ws_client)), @@ -139,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 @@ -360,6 +406,13 @@ impl Client { self.create_document_with_id(Uuid::new_v4(), content).await } + /// 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 + } + pub async fn create_document_with_id( &self, id: Uuid, @@ -367,7 +420,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, @@ -1622,6 +1675,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(); diff --git a/replicant-client/src/database.rs b/replicant-client/src/database.rs index 523db01..017ff69 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; @@ -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) @@ -109,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) @@ -128,6 +125,53 @@ 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. + /// + /// # 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( + "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) + .bind(canonical_id.to_string()) + .bind(old_id.to_string()) + .execute(&mut *tx) + .await?; + + 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(()) + } + pub async fn get_document(&self, id: &Uuid) -> SyncResult { let row = sqlx::query(Queries::GET_DOCUMENT) .bind(id.to_string()) @@ -577,26 +621,143 @@ impl ClientDatabase { } #[cfg(test)] -mod identity_freeze_tests { +mod identity_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" - ); + async fn fresh_db() -> ClientDatabase { + let db = ClientDatabase::new(":memory:").await.unwrap(); + db.run_migrations().await.unwrap(); + db } - #[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" - ); + #[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); + } + + #[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); + } + + 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); + } + + #[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/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..4994322 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}; @@ -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 @@ -815,6 +818,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 diff --git a/replicant-client/src/queries.rs b/replicant-client/src/queries.rs index 9a2d716..40b8a60 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 WHERE user_id = ?2"; + // Document queries pub const GET_DOCUMENT: &'static str = r#" SELECT id, user_id, content, sync_revision, diff --git a/replicant-client/src/websocket.rs b/replicant-client/src/websocket.rs index 953bf68..ba06649 100644 --- a/replicant-client/src/websocket.rs +++ b/replicant-client/src/websocket.rs @@ -27,7 +27,6 @@ pub struct WebSocketClient { channel: Arc, _public_channel: Arc, tx: mpsc::Sender, - user_id: Uuid, } pub struct WebSocketReceiver { @@ -87,15 +86,11 @@ impl WebSocketClient { ws_err(format!("Connect failed: {:?}", e)) })?; - // Join channel with HMAC auth + // 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 join_payload = json!({ - "email": email, - "api_key": api_key, - "signature": signature, - "timestamp": timestamp - }); + let join_payload = Self::build_join_payload(email, api_key, &signature, timestamp, user_id); // Join per-user channel let channel = socket @@ -106,13 +101,25 @@ 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)) })?; + // 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!({})); + 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; + return Err(ws_err(msg)); + } + // Join public channel for public document events let public_channel = socket .channel( @@ -135,8 +142,8 @@ 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(), is_connected.clone()); + Self::setup_broadcast_handlers(&public_channel, tx.clone(), is_connected); // Emit auth success let _ = tx @@ -151,12 +158,50 @@ impl WebSocketClient { channel, _public_channel: public_channel, tx, - user_id, }, WebSocketReceiver { rx }, )) } + /// 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: Uuid, + ) -> Value { + json!({ + "email": email, + "api_key": api_key, + "signature": signature, + "timestamp": timestamp, + "user_id": user_id.to_string(), + }) + } + + /// `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 { let url = match server_url { s if s.starts_with("http://") => s.replace("http://", "ws://"), @@ -175,7 +220,6 @@ impl WebSocketClient { fn setup_broadcast_handlers( channel: &Arc, tx: mpsc::Sender, - user_id: Uuid, is_connected: Arc, ) { let events = channel.events(); @@ -191,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 }) @@ -346,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; } } @@ -481,12 +523,20 @@ 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 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()?), }; Some(Document { id: Uuid::parse_str(j.get("id")?.as_str()?).ok()?, @@ -565,7 +615,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()); @@ -575,11 +625,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 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 join_payload_always_includes_user_id() { + let uid = Uuid::new_v4(); + 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 reply_identity_matching_id_is_ok() { + let local = Uuid::new_v4(); + let reply = serde_json::json!({ "user_id": local.to_string() }); + assert!(WebSocketClient::verify_reply_identity(local, &reply).is_ok()); + } + + #[test] + fn reply_identity_mismatch_is_drift_error() { + let local = Uuid::new_v4(); + 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 reply_identity_absent_is_tolerated() { + let local = Uuid::new_v4(); + let reply = serde_json::json!({ "email": "a@b.com" }); + 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..6d8333a --- /dev/null +++ b/replicant-client/tests/constructor_identity_tests.rs @@ -0,0 +1,226 @@ +//! 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 std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use uuid::Uuid; + +/// 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 { + let path = std::env::temp_dir().join(format!("replicant-ctor-test-{}.sqlite3", Uuid::new_v4())); + 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 +} + +#[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); +} + +/// Empty credentials must yield a fully usable local-only client: no +/// 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, &server_url, "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(), + "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 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, + &server_url, + "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(), + "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" + ); +} 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/identity_adoption_test.rs b/replicant-client/tests/phoenix_integration/identity_adoption_test.rs new file mode 100644 index 0000000..8173260 --- /dev/null +++ b/replicant-client/tests/phoenix_integration/identity_adoption_test.rs @@ -0,0 +1,132 @@ +//! 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` 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`, 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) + .unwrap() + .as_nanos(); + format!( + "databases/identity_adopt_{}_{}.sqlite3", + std::process::id(), + nanos + ) +} + +#[tokio::test] +#[serial] +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(); + 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" + ); + assert_ne!(provisional, canonical); + + 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 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 adopt and connect to the live server"); + + // 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:"); + + // 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) + .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 1c00ae9..6b6ef56 100644 --- a/replicant-client/tests/phoenix_integration/mod.rs +++ b/replicant-client/tests/phoenix_integration/mod.rs @@ -22,6 +22,8 @@ mod basic_sync_test; mod conflict_test; +mod credential_enrollment_test; +mod identity_adoption_test; mod live_sync_test; mod multi_client_test; @@ -59,22 +61,19 @@ pub fn server_url() -> String { .unwrap_or_else(|_| "ws://localhost:4000/socket/websocket".to_string()) } -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) +/// 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) } -#[test] -fn deterministic_user_id_normalizes_like_client_and_server() { - assert_eq!( - deterministic_user_id(" Integration-Test@Example.COM "), - deterministic_user_id(TEST_EMAIL) - ); +pub fn skip_if_no_server() -> bool { + std::env::var("RUN_INTEGRATION_TESTS").is_err() } /// A broadcast event received from the server @@ -104,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 = deterministic_user_id(email); + let user_id = test_user_id(); let socket = Socket::spawn(url, None, None) .await 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 145b940..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,6 +204,7 @@ async fn main() -> Result<()> { &args.user, &args.api_key, &args.api_secret, + 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 6793654..38eaf80 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}"; } @@ -40,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)" @@ -47,56 +69,109 @@ 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 + [ -n "$BOOT_SCRIPT" ] && rm -f "$BOOT_SCRIPT" } 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 ) -# --- Start server ------------------------------------------------------------ -log "Starting Elixir server on port $SERVER_PORT (log: $SERVER_LOG)" +# --- 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 (stderr: $SERVER_LOG)" : > "$SERVER_LOG" -( cd "$SERVER_DIR" && PORT="$SERVER_PORT" DATABASE_URL="$DATABASE_URL" MIX_ENV=dev \ - exec mix phx.server ) >> "$SERVER_LOG" 2>&1 & +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>>"$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"; tail -30 "$SERVER_LOG"; exit 1; } +log "Enrolled user_id=$TEST_USER_ID" + +# --- Start server (minimal endpoint mounting the sync socket) ---------------- +BOOT_SCRIPT="$(mktemp /tmp/replicant_interop_boot.XXXXXX.exs)" +cat > "$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 +180,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 +190,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