diff --git a/crates/buzz-db/src/store/artifact_sync.rs b/crates/buzz-db/src/store/artifact_sync.rs index 32067986e5..18b9323276 100644 --- a/crates/buzz-db/src/store/artifact_sync.rs +++ b/crates/buzz-db/src/store/artifact_sync.rs @@ -120,14 +120,6 @@ impl Db { if ids.is_empty() { return Ok(vec![]); } - let mut tags = query.tags.clone(); - tags.push(("d".into(), ids.iter().map(Uuid::to_string).collect())); - let narrowed = ArtifactQuery { - view: query.view, - tags, - limit: PAGE_ROWS as i64, - offset: 0, - }; let connection = crate::observability::acquire_writer( &self.pool, crate::observability::WriterOperation::SubscriptionHistory, @@ -141,7 +133,8 @@ impl Db { .execute(&mut *tx) .await?; let mut qb = QueryBuilder::new("SELECT e.id,e.pubkey,e.created_at,e.kind,e.tags,e.content,e.sig,e.received_at,e.channel_id FROM events e "); - super::artifact_query::predicates(&mut qb, community, reader, &narrowed); + super::artifact_query::predicates(&mut qb, community, reader, query); + qb.push(" AND a.artifact_id=ANY(").push_bind(ids).push(")"); qb.push(" LIMIT ").push_bind(PAGE_ROWS as i64); let mut events = Vec::new(); let mut bytes = 0; @@ -204,7 +197,7 @@ WITH readable AS NOT MATERIALIZED ( AND m.sync_modified_at <= $4 AND a.channel_id IN (SELECT id FROM readable) AND (cardinality($8::text[])=0 OR a.artifact_type=ANY($8)) UNION ALL - SELECT r.modified_at,3,encode(r.event_id,'hex'),r.artifact_id,NULL::uuid + SELECT DISTINCT r.modified_at,3,r.artifact_id::text,r.artifact_id,NULL::uuid FROM artifact_revisions r WHERE r.community_id=$1 AND r.source_channel_id IS NOT NULL AND r.source_channel_id IN (SELECT id FROM readable) AND $3::timestamptz IS NOT NULL AND r.modified_at >= $3 AND r.modified_at <= $4 diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs index 0855cd1c3a..af52ff02e1 100644 --- a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs @@ -384,3 +384,30 @@ async fn sync_progress_replays_queries_with_escaped_values() { assert!(status.is_success(), "{status}: {next}"); assert_eq!(next["changes"], json!([])); } + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_current_intersects_known_ids_with_query_scope() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let included = Uuid::new_v4(); + let excluded = Uuid::new_v4(); + let event = f.revision(included, "create", None); + accept(&f, &event).await; + accept(&f, &f.revision(excluded, "create", None)).await; + let (status, body) = f + .request( + "/query", + json!([{"artifact":"sync","action":"current", + "query":{"artifact":"current","#type":["buzz.task"],"#d":[included]}, + "artifactIds":[included,excluded]}]), + ) + .await; + assert!(status.is_success(), "{status}: {body}"); + assert_eq!(body["changes"][0]["artifactId"], included.to_string()); + assert_eq!(body["changes"][0]["event"]["id"], event.id.to_hex()); + assert_eq!( + body["changes"][1], + json!({"type":"remove","artifactId":excluded}) + ); +} diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs new file mode 100644 index 0000000000..4f2b855e57 --- /dev/null +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs @@ -0,0 +1,102 @@ +use super::*; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_move_cursor_hides_private_and_redacted_revision_ids() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let (_, empty) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"sync","query":query()}]), + ) + .await; + let mut current = std::collections::HashMap::new(); + let mut private_ids = Vec::new(); + for _ in 0..101 { + let id = Uuid::new_v4(); + let created = f.revision(id, "create", None); + accept(&f, &created).await; + let moved = f.revision_as(&f.owner, f.private, id, "move", Some(&created)); + accept(&f, &moved).await; + private_ids.push(moved.id.to_hex()); + current.insert(id, moved); + } + let (&id, previous) = current.iter().next().unwrap(); + let back = f.revision(id, "move", Some(previous)); + accept(&f, &back).await; + let moved = f.revision_as(&f.owner, f.private, id, "move", Some(&back)); + accept(&f, &moved).await; + private_ids.push(moved.id.to_hex()); + current.insert(id, moved); + let stamp: chrono::DateTime = sqlx::query_scalar("SELECT clock_timestamp()") + .fetch_one(&f.pool) + .await + .unwrap(); + sqlx::query("UPDATE artifact_revisions SET modified_at=$2 WHERE community_id=$1") + .bind(f.community.as_uuid()) + .bind(stamp) + .execute(&f.pool) + .await + .unwrap(); + for redacted in [false, true] { + let (status, first) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"sync","query":query(),"checkpoint":empty["checkpoint"]}]), + ) + .await; + assert!(status.is_success(), "{status}: {first}"); + let cursor = first["nextCursor"].as_str().unwrap(); + let payload = cursor.split_once('.').unwrap().0; + let decoded = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(payload) + .unwrap(); + let decoded = String::from_utf8(decoded).unwrap(); + for private_id in &private_ids { + assert!( + !decoded.contains(private_id), + "cursor exposes a private revision ID (redacted={redacted})" + ); + } + assert_eq!(first["changes"].as_array().unwrap().len(), 100); + let (status, last) = f + .request_as( + &f.peer, + "/query", + json!([{"artifact":"sync","query":query(),"cursor":cursor}]), + ) + .await; + assert!(status.is_success(), "{status}: {last}"); + assert!(last["checkpoint"].is_string()); + let mut ids = std::collections::HashSet::new(); + for change in first["changes"] + .as_array() + .unwrap() + .iter() + .chain(last["changes"].as_array().unwrap()) + { + assert_eq!(change["type"], "remove"); + assert!(ids.insert(change["artifactId"].as_str().unwrap().to_owned())); + } + assert_eq!(ids.len(), 101); + if !redacted { + let boundary = first["changes"].as_array().unwrap().last().unwrap()["artifactId"] + .as_str() + .unwrap() + .parse::() + .unwrap(); + assert!(buzz_db::event::soft_delete_event_and_update_thread( + &f.pool, + f.community, + current[&boundary].id.as_bytes(), + None, + None + ) + .await + .unwrap()); + } + } +} diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs index 3a776dc158..aaac2be864 100644 --- a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs @@ -2,6 +2,8 @@ use super::*; use axum::http::StatusCode; #[path = "sync_access_postgres_tests.rs"] mod access_postgres_tests; +#[path = "sync_cursor_postgres_tests.rs"] +mod cursor_postgres_tests; #[path = "sync_policy_postgres_tests.rs"] mod policy_postgres_tests; #[path = "sync_scale_postgres_tests.rs"]