diff --git a/client/src/lib.rs b/client/src/lib.rs index 6ceef56..94b8ce2 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -107,8 +107,7 @@ pub fn run_download_once( let storage_dir = &PathBuf::from(storage_dir); let chunk_cache = InMemoryCache::new(INMEMORY_CACHE_MAX_REC, INMEMORY_CACHE_MAX_MEM); - let chunker = &mut Chunker::new(chunk_cache, storage_dir.clone()); - let chunker = Arc::new(Mutex::new(chunker)); + let chunker = Arc::new(Mutex::new(Chunker::new(chunk_cache, storage_dir.clone()))); let remote = &remote::Remote::new(api_endpoint, remote_token); let pool = connection::get_connection_pool(db_file_path)?; @@ -138,8 +137,7 @@ pub fn run_upload_once( ) -> Result<(), errors::SyncError> { let storage_dir = &PathBuf::from(storage_dir); let chunk_cache = InMemoryCache::new(INMEMORY_CACHE_MAX_REC, INMEMORY_CACHE_MAX_MEM); - let chunker = &mut Chunker::new(chunk_cache, storage_dir.clone()); - let chunker = Arc::new(Mutex::new(chunker)); + let chunker = Arc::new(Mutex::new(Chunker::new(chunk_cache, storage_dir.clone()))); let remote = &remote::Remote::new(api_endpoint, remote_token); let pool = connection::get_connection_pool(db_file_path)?; @@ -188,7 +186,7 @@ pub async fn run_async( let storage_dir = &PathBuf::from(storage_dir); let chunk_cache = InMemoryCache::new(INMEMORY_CACHE_MAX_REC, INMEMORY_CACHE_MAX_MEM); - let chunker = &mut Chunker::new(chunk_cache, storage_dir.clone()); + let chunker = Chunker::new(chunk_cache, storage_dir.clone()); let remote = &remote::Remote::new(api_endpoint, remote_token); let pool = connection::get_connection_pool(db_file_path)?; diff --git a/client/src/models.rs b/client/src/models.rs index f3e87b4..c713094 100644 --- a/client/src/models.rs +++ b/client/src/models.rs @@ -219,7 +219,7 @@ mod tests { assert_eq!(form.jid, Some(42)); assert_eq!(form.path, "recipes/test.cook"); - assert_eq!(form.deleted, false); + assert!(!form.deleted); assert_eq!(form.size, 512); assert_eq!(form.namespace_id, 5); } @@ -238,7 +238,7 @@ mod tests { }; assert_eq!(form.path, "recipes/deleted.cook"); - assert_eq!(form.deleted, true); + assert!(form.deleted); assert_eq!(form.size, 0); } diff --git a/client/src/registry.rs b/client/src/registry.rs index a86d688..d78f51a 100644 --- a/client/src/registry.rs +++ b/client/src/registry.rs @@ -10,7 +10,7 @@ use crate::schema::*; type Result = std::result::Result; -pub fn create(conn: &mut Connection, forms: &Vec) -> Result { +pub fn create(conn: &mut Connection, forms: &[CreateForm]) -> Result { trace!("inserting {:?}", forms); insert_into(file_records::table).values(forms).execute(conn) @@ -25,7 +25,7 @@ pub fn update_jid(conn: &mut Connection, record: &FileRecord, jid: i32) -> Resul .execute(conn) } -pub fn delete(conn: &mut Connection, forms: &Vec) -> Result { +pub fn delete(conn: &mut Connection, forms: &[DeleteForm]) -> Result { trace!("marking as deleted {:?}", forms); insert_into(file_records::table).values(forms).execute(conn) diff --git a/client/src/syncer.rs b/client/src/syncer.rs index 840fde8..65b24b1 100644 --- a/client/src/syncer.rs +++ b/client/src/syncer.rs @@ -31,7 +31,7 @@ pub async fn run( pool: &ConnectionPool, storage_path: &Path, namespace_id: i32, - chunker: &mut Chunker, + chunker: Chunker, remote: &Remote, local_registry_updated_rx: Receiver, read_only: bool, @@ -78,7 +78,7 @@ async fn download_loop( token: CancellationToken, listener: Option>, pool: &ConnectionPool, - chunker: Arc>, + chunker: Arc>, remote: &Remote, storage_path: &Path, namespace_id: i32, @@ -134,7 +134,7 @@ pub async fn upload_loop( token: CancellationToken, listener: Option>, pool: &ConnectionPool, - chunker: Arc>, + chunker: Arc>, remote: &Remote, namespace_id: i32, mut local_registry_updated_rx: Receiver, @@ -182,7 +182,7 @@ pub async fn upload_loop( pub async fn check_upload_once( pool: &ConnectionPool, - chunker: Arc>, + chunker: Arc>, remote: &Remote, namespace_id: i32, ) -> Result { @@ -246,7 +246,7 @@ pub async fn check_upload_once( pub async fn check_download_once( pool: &ConnectionPool, - chunker: Arc>, + chunker: Arc>, remote: &Remote, storage_path: &Path, namespace_id: i32, diff --git a/client/tests/common/mod.rs b/client/tests/common/mod.rs index 7f8f26a..22d7ad3 100644 --- a/client/tests/common/mod.rs +++ b/client/tests/common/mod.rs @@ -54,3 +54,23 @@ fn base64_url_nopad(bytes: &[u8]) -> String { use base64::Engine; URL_SAFE_NO_PAD.encode(bytes) } + +use cooklang_sync_client::chunker::{Chunker, InMemoryCache}; + +/// Bundle of the three things every syncer integration test needs: +/// a fresh SQLite pool, a tempdir root, and a `Chunker` rooted at that dir. +/// +/// The tempdir is returned separately (not moved into the `Chunker`) so the +/// test can write fixture files into it before invoking the syncer. +pub struct ClientBase { + pub pool: cooklang_sync_client::connection::ConnectionPool, + pub dir: TempDir, + pub chunker: Chunker, +} + +pub fn client_base() -> ClientBase { + let (pool, dir) = fresh_client_pool(); + let cache = InMemoryCache::new(100, 10_000_000); + let chunker = Chunker::new(cache, dir.path().to_path_buf()); + ClientBase { pool, dir, chunker } +} diff --git a/client/tests/connection_tests.rs b/client/tests/connection_tests.rs index ec83d78..547d735 100644 --- a/client/tests/connection_tests.rs +++ b/client/tests/connection_tests.rs @@ -42,8 +42,7 @@ fn get_connection_pool_returns_error_for_unwritable_path() { // directory, causing SQLite to fail to open the file. let bogus = "/dev/null/does_not_exist/db.sqlite3"; let err = get_connection_pool(bogus) - .err() - .expect("pool creation should fail on unwritable parent"); + .expect_err("pool creation should fail on unwritable parent"); let msg = format!("{err}"); assert!( msg.to_lowercase().contains("connection"), diff --git a/client/tests/indexer_tests.rs b/client/tests/indexer_tests.rs index 8337bcc..545b4dc 100644 --- a/client/tests/indexer_tests.rs +++ b/client/tests/indexer_tests.rs @@ -260,3 +260,16 @@ fn check_index_once_does_not_tombstone_just_downloaded_file() { assert_eq!(live[0].path, rel); assert!(!live[0].deleted, "downloaded file must not be soft-deleted by the indexer"); } + +#[test] +fn check_index_once_on_empty_dir_is_noop() { + let (pool, _dir) = common::fresh_client_pool(); + let storage = TempDir::new().expect("tempdir"); + + let changed = check_index_once(&pool, storage.path(), 1).expect("check_index_once"); + assert!(!changed, "empty dir must return Ok(false)"); + + let conn = &mut get_connection(&pool).expect("checkout"); + let rows = registry::non_deleted(conn, 1).expect("non_deleted"); + assert!(rows.is_empty(), "empty dir must not produce any registry rows"); +} diff --git a/client/tests/registry_tests.rs b/client/tests/registry_tests.rs index 3321143..1f0de01 100644 --- a/client/tests/registry_tests.rs +++ b/client/tests/registry_tests.rs @@ -263,3 +263,38 @@ fn latest_jid_returns_highest_jid_in_namespace_and_ignores_null_jid_rows() { assert_eq!(registry::latest_jid(conn, 1).unwrap(), 7); assert_eq!(registry::latest_jid(conn, 2).unwrap(), 100); } + +#[test] +fn updated_locally_surfaces_unsynced_tombstone() { + let (pool, _dir) = common::fresh_client_pool(); + let conn = &mut get_connection(&pool).expect("checkout"); + + // Create then delete (tombstone is appended; jid still None on both rows). + registry::create(conn, &vec![sample_create("gone.cook", 10, 1)]).expect("create"); + let existing: Vec = registry::non_deleted(conn, 1).expect("non_deleted"); + let sample = existing.first().expect("row present"); + registry::delete(conn, &vec![sample_delete(sample)]).expect("delete"); + + // Latest row per path is the tombstone, which still has jid=None, so + // updated_locally must surface it - this is what lets the upload path + // retry tombstones that never reached the server. + let unsynced = registry::updated_locally(conn, 1).expect("updated_locally"); + assert_eq!(unsynced.len(), 1, "tombstone with jid=None must appear in updated_locally"); + assert_eq!(unsynced[0].path, "gone.cook"); + assert!(unsynced[0].deleted, "latest row is the tombstone"); + assert!(unsynced[0].jid.is_none()); +} + +#[test] +fn latest_jid_returns_zero_for_explicit_jid_zero() { + let (pool, _dir) = common::fresh_client_pool(); + let conn = &mut get_connection(&pool).expect("checkout"); + + // Insert a single row with jid=Some(0). + let mut form = sample_create("a.cook", 5, 1); + form.jid = Some(0); + registry::create(conn, &vec![form]).expect("create"); + + let latest = registry::latest_jid(conn, 1).expect("latest_jid"); + assert_eq!(latest, 0, "Some(0) must unwrap to 0, not NotFound"); +} diff --git a/client/tests/remote_tests.rs b/client/tests/remote_tests.rs new file mode 100644 index 0000000..823544c --- /dev/null +++ b/client/tests/remote_tests.rs @@ -0,0 +1,412 @@ +//! Integration tests for `cooklang_sync_client::remote::Remote`. +//! +//! Every test spins up a fresh `wiremock::MockServer` and points a `Remote` +//! at its URL. Tests assert on URL shape, headers, and body — *not* on the +//! per-instance `uuid` that `Remote` mints at construction. + +use cooklang_sync_client::errors::SyncError; +use cooklang_sync_client::remote::{CommitResultStatus, Remote, ResponseFileRecord, REQUEST_TIMEOUT_SECS}; +use futures::StreamExt; +use std::time::Duration; +use wiremock::matchers::{ + body_string_contains, header, header_exists, method, path, query_param, query_param_contains, +}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +const TOKEN: &str = "test-token"; + +/// Build a `Remote` wired to `server`'s base URL. +fn new_remote(server: &MockServer) -> Remote { + Remote::new(&server.uri(), TOKEN) +} + +#[tokio::test] +async fn commit_returns_success_on_2xx_with_success_payload() { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/metadata/commit")) + .and(header("authorization", format!("Bearer {}", TOKEN).as_str())) + .and(header_exists("user-agent")) + .and(header_exists("x-client-version")) + .and(query_param_contains("uuid", "-")) // v4 UUID always contains hyphens + .and(body_string_contains("path=recipes%2Fa.cook")) + .and(body_string_contains("deleted=false")) + .and(body_string_contains("chunk_ids=abc%2Cdef")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "Success": 42 + }))) + .expect(1) + .mount(&server) + .await; + + let remote = new_remote(&server); + let result = remote + .commit("recipes/a.cook", false, "abc,def") + .await + .expect("commit"); + assert!(matches!(result, CommitResultStatus::Success(42))); +} + +#[tokio::test] +async fn commit_returns_need_chunks_on_2xx_with_need_chunks_payload() { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/metadata/commit")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "NeedChunks": "abc,def" + }))) + .expect(1) + .mount(&server) + .await; + + let remote = new_remote(&server); + let result = remote.commit("a.cook", false, "abc,def").await.expect("commit"); + match result { + CommitResultStatus::NeedChunks(s) => assert_eq!(s, "abc,def"), + other => panic!("expected NeedChunks, got {:?}", other), + } +} + +#[tokio::test] +async fn commit_maps_401_to_unauthorized() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/metadata/commit")) + .respond_with(ResponseTemplate::new(401)) + .mount(&server) + .await; + + let remote = new_remote(&server); + let err = remote.commit("a.cook", false, "").await.unwrap_err(); + assert!( + matches!(err, SyncError::Unauthorized), + "expected SyncError::Unauthorized on 401, got {:?}", + err + ); +} + +#[tokio::test] +async fn commit_maps_5xx_to_unknown_with_status_in_message() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/metadata/commit")) + .respond_with(ResponseTemplate::new(503)) + .mount(&server) + .await; + + let remote = new_remote(&server); + let err = remote.commit("a.cook", false, "").await.unwrap_err(); + match err { + SyncError::Unknown(msg) => assert!(msg.contains("503"), "expected status in message, got {msg:?}"), + other => panic!("expected SyncError::Unknown on 5xx, got {:?}", other), + } +} + +#[tokio::test] +async fn list_parses_response_records_and_preserves_order() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/metadata/list")) + .and(query_param("jid", "7")) + .and(header("authorization", format!("Bearer {}", TOKEN).as_str())) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([ + { "id": 8, "path": "a.cook", "deleted": false, "chunk_ids": "abc" }, + { "id": 9, "path": "b.cook", "deleted": true, "chunk_ids": "" } + ]))) + .expect(1) + .mount(&server) + .await; + + let remote = new_remote(&server); + let records: Vec = remote.list(7).await.expect("list"); + assert_eq!(records.len(), 2); + assert_eq!(records[0].id, 8); + assert_eq!(records[0].path, "a.cook"); + assert!(!records[0].deleted); + assert_eq!(records[0].chunk_ids, "abc"); + assert_eq!(records[1].id, 9); + assert!(records[1].deleted); +} + +#[tokio::test] +async fn list_returns_empty_vec_on_empty_json_array() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/metadata/list")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([]))) + .mount(&server) + .await; + + let remote = new_remote(&server); + let records = remote.list(0).await.expect("list"); + assert!(records.is_empty()); +} + +#[tokio::test] +async fn list_maps_401_to_unauthorized() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/metadata/list")) + .respond_with(ResponseTemplate::new(401)) + .mount(&server) + .await; + + let remote = new_remote(&server); + let err = remote.list(0).await.unwrap_err(); + assert!(matches!(err, SyncError::Unauthorized)); +} + +#[tokio::test] +async fn poll_returns_ok_on_200() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/metadata/poll")) + .and(query_param_contains("uuid", "-")) + .and(header("authorization", format!("Bearer {}", TOKEN).as_str())) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + + let remote = new_remote(&server); + remote.poll().await.expect("poll should succeed on 200"); +} + +#[tokio::test] +async fn poll_maps_401_to_unauthorized() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/metadata/poll")) + .respond_with(ResponseTemplate::new(401)) + .mount(&server) + .await; + + let remote = new_remote(&server); + let err = remote.poll().await.unwrap_err(); + assert!(matches!(err, SyncError::Unauthorized)); +} + +#[ignore = "deliberately waits for REQUEST_TIMEOUT_SECS (~60 s); run with -- --ignored"] +#[tokio::test] +async fn poll_treats_client_timeout_as_ok() { + let server = MockServer::start().await; + // Respond *after* the client's request timeout expires. + Mock::given(method("GET")) + .and(path("/metadata/poll")) + .respond_with( + ResponseTemplate::new(200) + .set_delay(Duration::from_secs(REQUEST_TIMEOUT_SECS + 5)), + ) + .mount(&server) + .await; + + let remote = new_remote(&server); + // `poll` deliberately swallows reqwest::Error::is_timeout and returns Ok(()). + remote.poll().await.expect("timeout should be mapped to Ok(())"); +} + +#[tokio::test] +async fn upload_posts_raw_body_to_chunk_path() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chunks/abc123")) + .and(header("authorization", format!("Bearer {}", TOKEN).as_str())) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + + let remote = new_remote(&server); + remote.upload("abc123", b"hello".to_vec()).await.expect("upload"); +} + +#[tokio::test] +async fn upload_maps_401_to_unauthorized() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chunks/abc123")) + .respond_with(ResponseTemplate::new(401)) + .mount(&server) + .await; + + let remote = new_remote(&server); + let err = remote.upload("abc123", b"hello".to_vec()).await.unwrap_err(); + assert!(matches!(err, SyncError::Unauthorized)); +} + +#[tokio::test] +async fn upload_maps_5xx_to_unknown_with_status_in_message() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chunks/abc123")) + .respond_with(ResponseTemplate::new(503)) + .mount(&server) + .await; + + let remote = new_remote(&server); + let err = remote.upload("abc123", b"hello".to_vec()).await.unwrap_err(); + match err { + SyncError::Unknown(msg) => assert!(msg.contains("503"), "expected status in message, got {msg:?}"), + other => panic!("expected SyncError::Unknown on 5xx, got {:?}", other), + } +} + +#[tokio::test] +async fn upload_batch_posts_multipart_with_each_chunk_as_named_part() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chunks/upload")) + .and(header("authorization", format!("Bearer {}", TOKEN).as_str())) + // Content-Type includes the generated boundary. + .and(header_exists("content-type")) + // Each chunk is a form-data part whose `name=` is its chunk_id. + .and(body_string_contains(r#"Content-Disposition: form-data; name="c1""#)) + .and(body_string_contains(r#"Content-Disposition: form-data; name="c2""#)) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + + let remote = new_remote(&server); + let chunks = vec![ + ("c1".to_string(), b"hello".to_vec()), + ("c2".to_string(), b"world".to_vec()), + ]; + remote.upload_batch(chunks).await.expect("upload_batch"); +} + +#[tokio::test] +async fn upload_batch_maps_401_to_unauthorized() { + use cooklang_sync_client::errors::SyncError; + + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chunks/upload")) + .respond_with(ResponseTemplate::new(401)) + .mount(&server) + .await; + + let remote = new_remote(&server); + let err = remote.upload_batch(vec![("c1".into(), b"x".to_vec())]).await.unwrap_err(); + assert!(matches!(err, SyncError::Unauthorized)); +} + +#[tokio::test] +async fn upload_batch_maps_5xx_to_unknown_with_status_in_message() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chunks/upload")) + .respond_with(ResponseTemplate::new(500)) + .mount(&server) + .await; + + let remote = new_remote(&server); + let err = remote.upload_batch(vec![("c1".into(), b"x".to_vec())]).await.unwrap_err(); + match err { + SyncError::Unknown(msg) => assert!(msg.contains("500"), "expected status in message, got {msg:?}"), + other => panic!("expected SyncError::Unknown on 5xx, got {:?}", other), + } +} + +#[tokio::test] +async fn download_returns_body_bytes_on_200() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/chunks/xyz")) + .and(header("authorization", format!("Bearer {}", TOKEN).as_str())) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"payload".to_vec())) + .expect(1) + .mount(&server) + .await; + + let remote = new_remote(&server); + let bytes = remote.download("xyz").await.expect("download"); + assert_eq!(bytes, b"payload"); +} + +#[tokio::test] +async fn download_maps_401_to_unauthorized() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/chunks/xyz")) + .respond_with(ResponseTemplate::new(401)) + .mount(&server) + .await; + + let remote = new_remote(&server); + let err = remote.download("xyz").await.unwrap_err(); + assert!(matches!(err, SyncError::Unauthorized)); +} + +#[tokio::test] +async fn download_batch_streams_parts_keyed_by_x_chunk_id_header() { + let server = MockServer::start().await; + + // Hand-assemble a tiny multipart body that mirrors the server's format. + // The parser in remote.rs looks for `--{boundary}` separators, then a + // `X-Chunk-ID:` line inside the part headers, then the bytes up to the + // next boundary. + let boundary = "testboundary"; + let body = format!( + "--{b}\r\n\ + X-Chunk-ID: c1\r\n\ + Content-Type: application/octet-stream\r\n\r\n\ + hello\r\n\ + --{b}\r\n\ + X-Chunk-ID: c2\r\n\ + Content-Type: application/octet-stream\r\n\r\n\ + world\r\n\ + --{b}--\r\n", + b = boundary + ); + + Mock::given(method("POST")) + .and(path("/chunks/download")) + .and(body_string_contains("chunk_ids%5B%5D=c1")) + .and(body_string_contains("chunk_ids%5B%5D=c2")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", format!("multipart/form-data; boundary={}", boundary).as_str()) + .set_body_bytes(body.into_bytes()), + ) + .expect(1) + .mount(&server) + .await; + + let remote = new_remote(&server); + let mut stream = remote.download_batch(vec!["c1", "c2"]).await; + + let mut got: Vec<(String, Vec)> = Vec::new(); + while let Some(item) = stream.next().await { + got.push(item.expect("part ok")); + } + + // Order is not guaranteed by the parser, so sort by chunk id. + got.sort_by(|a, b| a.0.cmp(&b.0)); + assert_eq!(got.len(), 2); + assert_eq!(got[0].0, "c1"); + assert_eq!(got[0].1, b"hello"); + assert_eq!(got[1].0, "c2"); + assert_eq!(got[1].1, b"world"); +} + +#[tokio::test] +async fn download_batch_errors_when_content_type_is_missing() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/chunks/download")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"whatever".to_vec())) + .mount(&server) + .await; + + let remote = new_remote(&server); + let mut stream = remote.download_batch(vec!["c1"]).await; + let first = stream.next().await.expect("at least one item").unwrap_err(); + assert!( + matches!(first, SyncError::BatchDownloadError(_)), + "expected BatchDownloadError on missing content-type, got {:?}", + first + ); +} diff --git a/client/tests/syncer_tests.rs b/client/tests/syncer_tests.rs new file mode 100644 index 0000000..5f0e540 --- /dev/null +++ b/client/tests/syncer_tests.rs @@ -0,0 +1,371 @@ +//! Integration tests for `syncer::check_upload_once` and `check_download_once`. +//! +//! Drives each function with: +//! * real SQLite pool via `common::fresh_client_pool()` +//! * real `Chunker` rooted at a tempdir (`common::client_base()`) +//! * `wiremock::MockServer` posing as the remote +//! +//! Tests pin observable side effects (DB rows, files on disk, HTTP requests +//! the server actually received) rather than internals. + +mod common; + +use cooklang_sync_client::chunker::{Chunker, InMemoryCache}; +use cooklang_sync_client::connection::get_connection; +use cooklang_sync_client::errors::SyncError; +use cooklang_sync_client::models::{CreateForm, DeleteForm, FileRecord}; +use cooklang_sync_client::registry; +use cooklang_sync_client::remote::Remote; +use cooklang_sync_client::syncer::{check_download_once, check_upload_once}; +use std::sync::Arc; +use time::OffsetDateTime; +use tokio::sync::Mutex; +use wiremock::matchers::{body_string_contains, method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +const NS: i32 = 1; +const TOKEN: &str = "test-token"; + +fn sample_create(path: &str, size: i64) -> CreateForm { + CreateForm { + jid: None, + path: path.to_string(), + deleted: false, + size, + modified_at: OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap(), + namespace_id: NS, + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn check_upload_once_commits_success_and_marks_jid() { + let server = MockServer::start().await; + // Always return Success(100). No NeedChunks branch => no upload_batch. + Mock::given(method("POST")) + .and(path("/metadata/commit")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "Success": 100 }))) + .expect(1) + .mount(&server) + .await; + + let base = common::client_base(); + // Seed a real file so hashify can read it. + tokio::fs::write(base.dir.path().join("a.cook"), b"Eggs\n").await.expect("write file"); + // Seed a registry row with jid=None so `updated_locally` picks it up. + { + let conn = &mut get_connection(&base.pool).expect("checkout"); + registry::create(conn, &[sample_create("a.cook", 5)]).expect("create"); + } + + let remote = Remote::new(&server.uri(), TOKEN); + let chunker_arc = Arc::new(Mutex::new(base.chunker)); + let all_committed = check_upload_once(&base.pool, Arc::clone(&chunker_arc), &remote, NS) + .await + .expect("check_upload_once"); + assert!(all_committed, "all rows should commit in one pass"); + + // Row now has the returned jid. + let conn = &mut get_connection(&base.pool).expect("checkout"); + let after: Vec = registry::updated_locally(conn, NS).expect("updated_locally"); + assert!(after.is_empty(), "no rows should remain unsynced after Success"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn check_upload_once_triggers_upload_batch_when_server_asks_for_chunks() { + let server = MockServer::start().await; + + // First /metadata/commit returns NeedChunks. We don't care which chunk ids + // are listed — we just need *some* comma-separated ids for the client to + // push through. To keep it simple, we echo back "CHUNK1" and expect the + // client to attempt to read that chunk from its cache (the hashify call + // warms the cache, so the id we echo must be one of the hashes the + // chunker computed). Workaround: the test file is *text*, so only one + // chunk id is produced; we read it out, then program the mock to return + // exactly that id in NeedChunks. + // + // This two-phase mocking is unavoidable because chunk ids are content- + // derived and we don't want to hard-code them. + + let mut base = common::client_base(); + tokio::fs::write(base.dir.path().join("a.cook"), b"Eggs\n").await.expect("write file"); + let computed_ids = base.chunker.hashify("a.cook").await.expect("hashify"); + assert!(!computed_ids.is_empty(), "text chunker must produce ids"); + let chunk_id = computed_ids.first().expect("first id").clone(); + + Mock::given(method("POST")) + .and(path("/metadata/commit")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "NeedChunks": chunk_id.clone() + }))) + .expect(1) + .mount(&server) + .await; + + Mock::given(method("POST")) + .and(path("/chunks/upload")) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + + { + let conn = &mut get_connection(&base.pool).expect("checkout"); + registry::create(conn, &[sample_create("a.cook", 5)]).expect("create"); + } + + let remote = Remote::new(&server.uri(), TOKEN); + let chunker_arc = Arc::new(Mutex::new(base.chunker)); + let all_committed = check_upload_once(&base.pool, Arc::clone(&chunker_arc), &remote, NS) + .await + .expect("check_upload_once"); + // NeedChunks path means we did *not* fully commit this pass — caller will + // retry on the next loop iteration (that's why `lib::run_upload_once` + // calls this function twice). + assert!(!all_committed, "NeedChunks path should return false"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn check_upload_once_commits_tombstone_without_hashifying() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/metadata/commit")) + .and(body_string_contains("deleted=true")) + .and(body_string_contains("chunk_ids=&")) // empty chunk_ids, followed by next field + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "Success": 7 }))) + .expect(1) + .mount(&server) + .await; + + let base = common::client_base(); + // Deliberately do NOT write the file to disk. + { + let conn = &mut get_connection(&base.pool).expect("checkout"); + // Seed a first record + a tombstone that supersedes it. `updated_locally` + // returns the latest (id-max) row per path. + registry::create(conn, &[sample_create("gone.cook", 10)]).expect("create"); + let live: Vec = registry::non_deleted(conn, NS).expect("non_deleted"); + let latest = live.first().expect("live row").clone(); + let tombstone = DeleteForm { + path: latest.path.clone(), + jid: None, + size: 0, + modified_at: latest.modified_at, + deleted: true, + namespace_id: NS, + }; + registry::delete(conn, &[tombstone]).expect("delete"); + } + + let remote = Remote::new(&server.uri(), TOKEN); + let chunker_arc = Arc::new(Mutex::new(base.chunker)); + let ok = check_upload_once(&base.pool, Arc::clone(&chunker_arc), &remote, NS) + .await + .expect("check_upload_once"); + assert!(ok, "tombstone commit is a Success => all_commited stays true"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn check_download_once_writes_file_and_inserts_registry_row() { + let server = MockServer::start().await; + + // We need two mocks to be wired in a particular order: + // 1. GET /metadata/list?jid=0 → return one record referencing chunk "c1" + // 2. POST /chunks/download → stream back a multipart body containing c1 + // The chunk id must be content-derived so the chunker accepts the save. + // Easiest: we hashify a known text file *first* in a scratch chunker rooted + // at a throwaway dir, grab the chunk id, then drive the real download against + // a fresh base. + let scratch_dir = tempfile::TempDir::new().unwrap(); + tokio::fs::write(scratch_dir.path().join("a.cook"), b"Eggs\n").await.unwrap(); + let scratch_cache = InMemoryCache::new(10, 1_000_000); + let mut scratch = Chunker::new(scratch_cache, scratch_dir.path().to_path_buf()); + let ids = scratch.hashify("a.cook").await.unwrap(); + assert_eq!(ids.len(), 1, "text file produces one line-per-chunk id"); + let chunk_id = ids[0].clone(); + + Mock::given(method("GET")) + .and(path("/metadata/list")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([ + { "id": 11, "path": "a.cook", "deleted": false, "chunk_ids": chunk_id } + ]))) + .expect(1) + .mount(&server) + .await; + + let boundary = "downloadbound"; + let body = format!( + "--{b}\r\nX-Chunk-ID: {id}\r\nContent-Type: application/octet-stream\r\n\r\nEggs\n\r\n--{b}--\r\n", + b = boundary, + id = chunk_id + ); + Mock::given(method("POST")) + .and(path("/chunks/download")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", format!("multipart/form-data; boundary={}", boundary).as_str()) + .set_body_bytes(body.into_bytes()), + ) + .expect(1) + .mount(&server) + .await; + + let base = common::client_base(); + let remote = Remote::new(&server.uri(), TOKEN); + let chunker_arc = Arc::new(Mutex::new(base.chunker)); + let downloaded = check_download_once( + &base.pool, + Arc::clone(&chunker_arc), + &remote, + base.dir.path(), + NS, + ) + .await + .expect("check_download_once"); + assert!(downloaded, "non-empty remote list => returns true"); + + // File was written. + let bytes = tokio::fs::read(base.dir.path().join("a.cook")).await.expect("read"); + assert_eq!(bytes, b"Eggs\n"); + + // Registry has a live row with jid=11. + let conn = &mut get_connection(&base.pool).expect("checkout"); + let rows = registry::non_deleted(conn, NS).expect("non_deleted"); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].path, "a.cook"); + assert_eq!(rows[0].jid, Some(11)); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn check_download_once_removes_local_file_and_appends_tombstone() { + let server = MockServer::start().await; + + Mock::given(method("GET")) + .and(path("/metadata/list")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([ + { "id": 22, "path": "gone.cook", "deleted": true, "chunk_ids": "" } + ]))) + .expect(1) + .mount(&server) + .await; + // No /chunks/download is expected — deleted records skip the download queue. + + let base = common::client_base(); + tokio::fs::write(base.dir.path().join("gone.cook"), b"bye\n").await.unwrap(); + { + let conn = &mut get_connection(&base.pool).expect("checkout"); + registry::create(conn, &[sample_create("gone.cook", 4)]).expect("create"); + } + + let remote = Remote::new(&server.uri(), TOKEN); + let chunker_arc = Arc::new(Mutex::new(base.chunker)); + check_download_once( + &base.pool, + Arc::clone(&chunker_arc), + &remote, + base.dir.path(), + NS, + ) + .await + .expect("check_download_once"); + + // File is gone on disk. + assert!( + !base.dir.path().join("gone.cook").exists(), + "local file should be deleted when remote says deleted" + ); + // Registry has a tombstone appended (latest row for this path has deleted=true). + let conn = &mut get_connection(&base.pool).expect("checkout"); + let live = registry::non_deleted(conn, NS).expect("non_deleted"); + assert!( + live.iter().all(|r| r.path != "gone.cook"), + "gone.cook should not be in non_deleted after tombstone applied" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn check_download_once_empty_remote_list_is_noop() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/metadata/list")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([]))) + .expect(1) + .mount(&server) + .await; + // No /chunks/download should be hit; wiremock will error on unmatched paths + // when we assert on a mounted mock. We omit that mock deliberately. + + let base = common::client_base(); + let remote = Remote::new(&server.uri(), TOKEN); + let chunker_arc = Arc::new(Mutex::new(base.chunker)); + let downloaded = check_download_once( + &base.pool, + Arc::clone(&chunker_arc), + &remote, + base.dir.path(), + NS, + ) + .await + .expect("check_download_once"); + assert!(!downloaded, "empty list => returns false"); + + // Registry is still empty. + let conn = &mut get_connection(&base.pool).expect("checkout"); + let rows = registry::non_deleted(conn, NS).expect("non_deleted"); + assert!(rows.is_empty()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn check_download_once_propagates_unauthorized_from_list() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/metadata/list")) + .respond_with(ResponseTemplate::new(401)) + .mount(&server) + .await; + + let base = common::client_base(); + let remote = Remote::new(&server.uri(), TOKEN); + let chunker_arc = Arc::new(Mutex::new(base.chunker)); + let err = check_download_once( + &base.pool, + Arc::clone(&chunker_arc), + &remote, + base.dir.path(), + NS, + ) + .await + .unwrap_err(); + assert!( + matches!(err, SyncError::Unauthorized), + "expected SyncError::Unauthorized, got {:?}", + err + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn check_upload_once_propagates_unauthorized_from_commit() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/metadata/commit")) + .respond_with(ResponseTemplate::new(401)) + .mount(&server) + .await; + + let base = common::client_base(); + tokio::fs::write(base.dir.path().join("a.cook"), b"Eggs\n").await.expect("write file"); + { + let conn = &mut get_connection(&base.pool).expect("checkout"); + registry::create(conn, &[sample_create("a.cook", 5)]).expect("create"); + } + + let remote = Remote::new(&server.uri(), TOKEN); + let chunker_arc = Arc::new(Mutex::new(base.chunker)); + let err = check_upload_once(&base.pool, Arc::clone(&chunker_arc), &remote, NS) + .await + .unwrap_err(); + assert!( + matches!(err, SyncError::Unauthorized), + "expected SyncError::Unauthorized, got {:?}", + err + ); +}