From f6919fc08a86611327ef687ca2619be52bbdaf95 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 21 Oct 2026 15:56:36 -0400 Subject: [PATCH] feat(db): add the artifact head view and the sync snapshot read Add the NIP-AR "head" query view, which returns the current revision of each named artifact, deletions included. Add the sync filter parser and ArtifactSnapshot: keyset pages ordered by d inside one REPEATABLE READ READ ONLY transaction, so every page sees the same cut. Opening the snapshot first checks, in the same snapshot, that the reader can read every named channel. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-core/src/artifact.rs | 63 ++-- crates/buzz-core/src/artifact_sync.rs | 77 +++++ crates/buzz-core/src/lib.rs | 2 + crates/buzz-db/src/lib.rs | 1 + .../src/store/artifact_postgres_tests.rs | 124 ++++++++ crates/buzz-db/src/store/artifact_query.rs | 274 ++++++++++++++---- crates/buzz-db/src/store/mod.rs | 1 + 7 files changed, 461 insertions(+), 81 deletions(-) create mode 100644 crates/buzz-core/src/artifact_sync.rs diff --git a/crates/buzz-core/src/artifact.rs b/crates/buzz-core/src/artifact.rs index 74f33c08d8..50e8f6070d 100644 --- a/crates/buzz-core/src/artifact.rs +++ b/crates/buzz-core/src/artifact.rs @@ -248,6 +248,8 @@ mod tests { ); let query = parse_query(&json!({"artifact":"current","#project":["p"]})).unwrap(); assert_eq!(query.view, ArtifactView::Current); + let head = parse_query(&json!({"artifact":"head","#d":["x"]})).unwrap(); + assert_eq!(head.view, ArtifactView::Head); for invalid in [ json!({"artifact":"lookup","#d":["x"]}), json!({"artifact":"history"}), @@ -256,6 +258,8 @@ mod tests { json!({"artifact":"current","limit":1001}), json!({"artifact":"current","#assignee":[]}), json!({"artifact":"history","#d":[]}), + json!({"artifact":"head"}), + json!({"artifact":"sync","#h":["c"]}), ] { assert!(parse_query(&invalid).is_err(), "{invalid}"); } @@ -276,6 +280,9 @@ pub enum ArtifactView { Current, /// Every stored, unredacted revision of the artifacts named by `#d`. History, + /// The current revision of each artifact named by `#d`, including a + /// deletion. + Head, } /// Explicit HTTP artifact query. All `#name` predicates compare the first /// value of the same tag. @@ -331,8 +338,38 @@ pub fn parse_query(value: &serde_json::Value) -> Result ArtifactView::Current, Some("history") => ArtifactView::History, - _ => return Err("artifact must be \"current\" or \"history\""), + Some("head") => ArtifactView::Head, + _ => return Err("artifact must be \"current\", \"history\" or \"head\""), }; + let tags = parse_predicates(object, &["artifact", "kinds", "limit", "offset"])?; + let number = |name: &str, default: u64| { + object + .get(name) + .map(|v| v.as_u64().ok_or("invalid limit or offset")) + .transpose() + .map(|v| v.unwrap_or(default)) + }; + let limit = number("limit", 100)?; + let offset = number("offset", 0)?; + if limit == 0 || limit > MAX_PAGE_SIZE as u64 || offset > MAX_OFFSET { + return Err("artifact page limit exceeded"); + } + if view != ArtifactView::Current && !tags.iter().any(|(n, _)| n == "d") { + return Err("history and head require #d"); + } + Ok(ArtifactQuery { + view, + tags, + limit: limit as i64, + offset: offset as i64, + }) +} +/// Parse the `kinds` and `#name` predicates shared by every artifact filter. +/// Keys outside `#name` and `allowed` are rejected. +pub(crate) fn parse_predicates( + object: &serde_json::Map, + allowed: &[&str], +) -> Result)>, &'static str> { if let Some(kinds) = object.get("kinds") { if kinds.as_array().map(Vec::as_slice) != Some(&[serde_json::json!(45010)]) { return Err("artifact queries accept only kinds [45010]"); @@ -361,32 +398,12 @@ pub fn parse_query(value: &serde_json::Value) -> Result MAX_PREDICATES || total > MAX_QUERY_VALUES || bytes > MAX_TAG_BYTES { return Err("artifact predicate limits exceeded"); } - let number = |name: &str, default: u64| { - object - .get(name) - .map(|v| v.as_u64().ok_or("invalid limit or offset")) - .transpose() - .map(|v| v.unwrap_or(default)) - }; - let limit = number("limit", 100)?; - let offset = number("offset", 0)?; - if limit == 0 || limit > MAX_PAGE_SIZE as u64 || offset > MAX_OFFSET { - return Err("artifact page limit exceeded"); - } - if view == ArtifactView::History && !tags.iter().any(|(n, _)| n == "d") { - return Err("history requires #d"); - } - Ok(ArtifactQuery { - view, - tags, - limit: limit as i64, - offset: offset as i64, - }) + Ok(tags) } diff --git a/crates/buzz-core/src/artifact_sync.rs b/crates/buzz-core/src/artifact_sync.rs new file mode 100644 index 0000000000..c7e95f496c --- /dev/null +++ b/crates/buzz-core/src/artifact_sync.rs @@ -0,0 +1,77 @@ +//! NIP-AR sync filter: `{"artifact":"sync", "#h":[...], ...}` on a `REQ`. +use crate::artifact::{canonical_uuid, parse_predicates}; +use uuid::Uuid; + +/// Largest set one sync serves: it bounds each initial read and the held +/// list of changed artifacts. Advertised in NIP-11 as `sync.max_events`. +pub const MAX_SYNC_EVENTS: usize = 4096; + +/// A parsed sync filter. `channels` are the `#h` values; `tags` holds every +/// `#name` predicate, `#h` included, with NIP-AR match semantics. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ArtifactSyncFilter { + /// Named channels (`#h`), deduplicated in request order. Never empty. + pub channels: Vec, + /// Exact tag-name/value predicates (AND across names, OR within one). + pub tags: Vec<(String, Vec)>, +} + +/// Whether a raw filter asks for sync rather than a one-shot artifact view. +pub fn is_sync_filter(value: &serde_json::Value) -> bool { + value.get("artifact").and_then(|v| v.as_str()) == Some("sync") +} + +/// Parse a sync filter. It requires `#h` with canonical channel UUIDs and +/// accepts the current-state predicates and `kinds: [45010]`, but no `limit` +/// or `offset`: a sync always serves the whole set. +pub fn parse_sync(value: &serde_json::Value) -> Result { + let object = value + .as_object() + .ok_or("artifact filter must be an object")?; + if !is_sync_filter(value) { + return Err("artifact must be \"sync\""); + } + let tags = parse_predicates(object, &["artifact", "kinds"])?; + let mut channels = Vec::new(); + for (_, values) in tags.iter().filter(|(name, _)| name == "h") { + for value in values { + let id = canonical_uuid(value).map_err(|_| "#h values must be channel UUIDs")?; + if !channels.contains(&id) { + channels.push(id); + } + } + } + if channels.is_empty() { + return Err("sync requires #h"); + } + Ok(ArtifactSyncFilter { channels, tags }) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn sync_filter_requires_channels_and_rejects_paging() { + let a = Uuid::new_v4().to_string(); + let parsed = parse_sync( + &json!({"artifact":"sync","#h":[a, a],"#type":["buzz.task"],"kinds":[45010]}), + ) + .unwrap(); + assert_eq!(parsed.channels.len(), 1); + assert_eq!(parsed.tags.len(), 2); + for invalid in [ + json!({"artifact":"sync"}), + json!({"artifact":"sync","#type":["buzz.task"]}), + json!({"artifact":"sync","#h":[a],"limit":10}), + json!({"artifact":"sync","#h":[a],"offset":0}), + json!({"artifact":"sync","#h":[a],"kinds":[45011]}), + json!({"artifact":"sync","#h":["not-a-uuid"]}), + json!({"artifact":"sync","#h":[a],"since":1}), + json!({"artifact":"current","#h":[a]}), + ] { + assert!(parse_sync(&invalid).is_err(), "{invalid}"); + } + } +} diff --git a/crates/buzz-core/src/lib.rs b/crates/buzz-core/src/lib.rs index e7669c874b..99b4aa6803 100644 --- a/crates/buzz-core/src/lib.rs +++ b/crates/buzz-core/src/lib.rs @@ -84,3 +84,5 @@ pub mod test_helpers { /// NIP-AR channel artifact envelope and limits. pub mod artifact; +/// NIP-AR sync filter parsing and limits. +pub mod artifact_sync; diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index c85b7dc6b9..f19656e237 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -85,6 +85,7 @@ pub use event::{EventQuery, DEFAULT_MAX_PAGE_LIMIT}; pub use nip_fi_binding::{NipFiBindingRecord, NipFiUnbind, RowCapExceeded}; pub use reaction::ReactionEventInsertOutcome; pub use reminder::DueReminder; +pub use store::{ArtifactSnapshot, SnapshotDenied}; pub use usage::SessionLeader; use buzz_core::CommunityId; diff --git a/crates/buzz-db/src/store/artifact_postgres_tests.rs b/crates/buzz-db/src/store/artifact_postgres_tests.rs index 97534792e7..26ebf9a4ec 100644 --- a/crates/buzz-db/src/store/artifact_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_postgres_tests.rs @@ -698,3 +698,127 @@ async fn head_outside_authorized_channel_matches_missing_head() { ArtifactOutcome::Accepted(_) )); } + +/// Sync's complete-set read: keyset pages by `d` share one snapshot, so a +/// change committed after it opens is in no page. Mutation: drop the +/// `REPEATABLE READ` statement → the late create appears in a later page → +/// RED. Mutation: drop the `a.artifact_id>` keyset bound → pages repeat → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sync_snapshot_pages_are_one_cut_and_head_includes_deletions() { + let f = Fixture::new().await; + let mut ids: Vec = (0..5).map(|_| Uuid::new_v4()).collect(); + ids.sort(); + let mut heads = vec![]; + for d in &ids { + let create = f.revision(*d, "create", f.a, None, &f.owner, vec![]); + assert!(matches!( + f.accept(&create, None).await, + ArtifactOutcome::Accepted(_) + )); + heads.push(create); + } + // A deleted artifact leaves the set but stays visible to the head view. + let deleted = f.revision(ids[4], "delete", f.a, Some(&heads[4]), &f.owner, vec![]); + assert!(matches!( + f.accept(&deleted, None).await, + ArtifactOutcome::Accepted(_) + )); + let reader = f.owner.public_key(); + let mut snapshot = + f.db.open_artifact_snapshot(f.community, reader.as_bytes(), &[f.a]) + .await + .unwrap() + .unwrap(); + let first = snapshot.next_page(&[], 2).await.unwrap(); + // Committed after the snapshot opened, and its `d` sorts after the first + // page, so only a fresh snapshot per page would return it. + let late = loop { + let d = Uuid::new_v4(); + if d > ids[1] { + break d; + } + }; + let late_create = f.revision(late, "create", f.a, None, &f.owner, vec![]); + assert!(matches!( + f.accept(&late_create, None).await, + ArtifactOutcome::Accepted(_) + )); + let mut seen: Vec<_> = first.iter().map(|e| e.event.id).collect(); + loop { + let page = snapshot.next_page(&[], 2).await.unwrap(); + if page.is_empty() { + break; + } + seen.extend(page.iter().map(|e| e.event.id)); + } + snapshot.finish().await.unwrap(); + assert_eq!(seen, heads[..4].iter().map(|e| e.id).collect::>()); + + // Predicates apply inside the snapshot. + let mut filtered = + f.db.open_artifact_snapshot(f.community, reader.as_bytes(), &[f.a]) + .await + .unwrap() + .unwrap(); + let none = filtered + .next_page(&[("type".into(), vec!["other".into()])], 10) + .await + .unwrap(); + assert!(none.is_empty()); + + // A private channel the reader cannot read refuses the whole sync. + assert!(matches!( + f.db.open_artifact_snapshot(f.community, f.peer.public_key().as_bytes(), &[f.a, f.b]) + .await + .unwrap(), + Err(super::artifact_query::SnapshotDenied::Restricted) + )); + // So does a channel that does not exist. + assert!(matches!( + f.db.open_artifact_snapshot(f.community, reader.as_bytes(), &[Uuid::new_v4()]) + .await + .unwrap(), + Err(super::artifact_query::SnapshotDenied::Restricted) + )); + + let head = + f.db.artifact_head(f.community, reader.as_bytes(), ids[4], &[]) + .await + .unwrap() + .unwrap(); + assert_eq!(head.event.id, deleted.id); + let (view, _) = f + .query( + &f.owner, + serde_json::json!({"artifact":"head","#d":[ids[4], ids[0]]}), + false, + ) + .await; + let mut view: Vec<_> = view.iter().map(|e| e.event.id).collect(); + view.sort(); + let mut expected = vec![deleted.id, heads[0].id]; + expected.sort(); + assert_eq!(view, expected); + assert_eq!( + f.count( + &f.owner, + serde_json::json!({"artifact":"current","#d":[ids[4]]}) + ) + .await, + 0 + ); + // A reader who cannot read the head's channel learns nothing. + let private = Uuid::new_v4(); + let create = f.revision(private, "create", f.b, None, &f.owner, vec![]); + assert!(matches!( + f.accept(&create, None).await, + ArtifactOutcome::Accepted(_) + )); + assert!(f + .db + .artifact_head(f.community, f.peer.public_key().as_bytes(), private, &[]) + .await + .unwrap() + .is_none()); +} diff --git a/crates/buzz-db/src/store/artifact_query.rs b/crates/buzz-db/src/store/artifact_query.rs index f359726d92..e74669c7f2 100644 --- a/crates/buzz-db/src/store/artifact_query.rs +++ b/crates/buzz-db/src/store/artifact_query.rs @@ -4,11 +4,146 @@ use buzz_core::{ artifact::{ArtifactQuery, ArtifactView}, CommunityId, StoredEvent, }; -use sqlx::QueryBuilder; +use sqlx::{pool::PoolConnection, Postgres, QueryBuilder, Transaction}; +use uuid::Uuid; + +const EVENT_COLUMNS: &str = + "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 "; + +/// Append the reader's permission scope and the exact tag predicates. A +/// system channel, or an unlisted open channel the reader is not a member +/// of, is readable only when `named_channels` names it (NIP-CL). +fn push_scope( + qb: &mut QueryBuilder, + community: CommunityId, + reader: &[u8], + tags: &[(String, Vec)], + named_channels: Vec, +) { + // The explicit kind lets the community/kind index exclude chat before + // the join, even for an unfiltered current-state page. + qb.push("WHERE e.community_id=").push_bind(community.as_uuid()).push(" AND e.kind=45010 AND e.deleted_at IS NULL AND EXISTS (SELECT 1 FROM channels c WHERE c.community_id=e.community_id AND c.id=e.channel_id AND c.deleted_at IS NULL AND (") + .push(crate::store::channel::legacy_channel_type_sql("c.channel_type")) + .push(" OR c.id=ANY(") + .push_bind(named_channels.clone()) + .push(")) AND ((c.visibility='open' AND (NOT c.unlisted OR c.id=ANY(") + .push_bind(named_channels) + .push("))) OR EXISTS (SELECT 1 FROM channel_members m WHERE m.community_id=c.community_id AND m.channel_id=c.id AND m.pubkey=") + .push_bind(reader.to_vec()).push(" AND m.removed_at IS NULL)))"); + for (name, values) in tags { + // GIN-usable necessary prefilter; retain the positional recheck + // because JSON array containment alone also matches swapped values. + qb.push(" AND ("); + for (index, value) in values.iter().enumerate() { + if index != 0 { + qb.push(" OR "); + } + qb.push("e.tags @> ") + .push_bind(serde_json::json!([[name, value]])); + } + qb.push(")"); + qb.push(" AND EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) tag WHERE tag->>0=") + .push_bind(name) + .push(" AND tag->>1=ANY(") + .push_bind(values) + .push("))"); + } +} + +/// Channel UUIDs named by `#h`; malformed values name no channel. +fn named_channels(tags: &[(String, Vec)]) -> Vec { + tags.iter() + .filter(|(name, _)| name == "h") + .flat_map(|(_, values)| values.iter().filter_map(|value| value.parse().ok())) + .collect() +} + +fn rows_to_events(rows: Vec) -> Result> { + let mut events = vec![]; + for row in rows { + if let Some(event) = crate::event::row_to_stored_event(row)? { + events.push(event); + } + } + Ok(events) +} + +/// Read-only: a plain writer-pool connection, labeled as the history read it +/// is, so it neither takes community admission nor counts as an event write. +/// The writer is always at least as fresh as every accepted change. +async fn read_connection(db: &Db) -> Result> { + crate::observability::acquire_writer( + &db.pool, + crate::observability::WriterOperation::SubscriptionHistory, + ) + .await + .map_err(Into::into) +} + +/// Why a sync snapshot did not open. +#[derive(Debug, PartialEq, Eq)] +pub enum SnapshotDenied { + /// The reader cannot read a named channel, or it does not exist. + Restricted, +} + +/// One read-only `REPEATABLE READ` snapshot of a sync's current heads, read +/// in keyset pages ordered by artifact identity. Every page sees the same +/// database snapshot, so the pages together are one consistent cut. +pub struct ArtifactSnapshot { + tx: Transaction<'static, Postgres>, + community: CommunityId, + reader: Vec, + channels: Vec, + after: Option, + done: bool, +} + +impl ArtifactSnapshot { + /// Read the next page of at most `page` heads that match `tags`, ordered + /// by `d`. Returns an empty page once the set is exhausted. + pub async fn next_page( + &mut self, + tags: &[(String, Vec)], + page: i64, + ) -> Result> { + if self.done { + return Ok(vec![]); + } + 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,a.artifact_id FROM events e ", + ); + qb.push("JOIN artifact_heads a ON a.community_id=e.community_id AND a.event_id=e.id AND NOT a.deleted "); + push_scope( + &mut qb, + self.community, + &self.reader, + tags, + self.channels.clone(), + ); + if let Some(after) = self.after { + qb.push(" AND a.artifact_id>").push_bind(after); + } + qb.push(" ORDER BY a.artifact_id LIMIT ").push_bind(page); + let rows = qb.build().fetch_all(&mut *self.tx).await?; + self.done = (rows.len() as i64) < page; + if let Some(last) = rows.last() { + self.after = Some(sqlx::Row::try_get(last, "artifact_id")?); + } + rows_to_events(rows) + } + + /// End the snapshot. Dropping it rolls the read-only transaction back. + pub async fn finish(self) -> Result<()> { + self.tx.commit().await?; + Ok(()) + } +} impl Db { - /// Execute a bounded query against current artifacts or revision history. - /// Permission is resolved in the query snapshot. + /// Execute a bounded query against current artifacts, current heads + /// including deletions, or revision history. Permission is resolved in + /// the query snapshot. pub async fn query_artifacts( &self, community: CommunityId, @@ -16,63 +151,34 @@ impl Db { query: &ArtifactQuery, count: bool, ) -> Result<(Vec, i64)> { - // Read-only: a plain writer-pool snapshot transaction, labeled as the - // history read it is, so it neither takes community admission nor - // counts as an event write. - let connection = crate::observability::acquire_writer( - &self.pool, - crate::observability::WriterOperation::SubscriptionHistory, - ) - .await?; + let connection = read_connection(self).await?; let mut tx = sqlx::Transaction::begin(connection, None).await?; sqlx::query("SET LOCAL statement_timeout='2s'") .execute(&mut *tx) .await?; - let mut qb = QueryBuilder::::new(if count { + let mut qb = QueryBuilder::::new(if count { "SELECT count(*) AS count FROM events e " } else { - "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 " + EVENT_COLUMNS }); - if query.view == ArtifactView::Current { - qb.push("JOIN artifact_heads a ON a.community_id=e.community_id AND a.event_id=e.id AND NOT a.deleted "); - } - // A system channel, or an unlisted open channel the reader is not a - // member of, is read only by a query whose `#h` names it (NIP-CL); - // malformed `#h` values name no channel. - let named_channels: Vec = query - .tags - .iter() - .filter(|(name, _)| name == "h") - .flat_map(|(_, values)| values.iter().filter_map(|value| value.parse().ok())) - .collect(); - // The explicit kind lets the community/kind index exclude chat before - // the join, even for an unfiltered current-state page. - qb.push("WHERE e.community_id=").push_bind(community.as_uuid()).push(" AND e.kind=45010 AND e.deleted_at IS NULL AND EXISTS (SELECT 1 FROM channels c WHERE c.community_id=e.community_id AND c.id=e.channel_id AND c.deleted_at IS NULL AND (") - .push(crate::store::channel::legacy_channel_type_sql("c.channel_type")) - .push(" OR c.id=ANY(") - .push_bind(named_channels.clone()) - .push(")) AND ((c.visibility='open' AND (NOT c.unlisted OR c.id=ANY(") - .push_bind(named_channels) - .push("))) OR EXISTS (SELECT 1 FROM channel_members m WHERE m.community_id=c.community_id AND m.channel_id=c.id AND m.pubkey=") - .push_bind(reader.to_vec()).push(" AND m.removed_at IS NULL)))"); - for (name, values) in &query.tags { - // GIN-usable necessary prefilter; retain the positional recheck - // because JSON array containment alone also matches swapped values. - qb.push(" AND ("); - for (index, value) in values.iter().enumerate() { - if index != 0 { - qb.push(" OR "); - } - qb.push("e.tags @> ") - .push_bind(serde_json::json!([[name, value]])); + match query.view { + ArtifactView::Current => { + qb.push("JOIN artifact_heads a ON a.community_id=e.community_id AND a.event_id=e.id AND NOT a.deleted "); } - qb.push(")"); - qb.push(" AND EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) tag WHERE tag->>0=") - .push_bind(name) - .push(" AND tag->>1=ANY(") - .push_bind(values) - .push("))"); + ArtifactView::Head => { + qb.push( + "JOIN artifact_heads a ON a.community_id=e.community_id AND a.event_id=e.id ", + ); + } + ArtifactView::History => {} } + push_scope( + &mut qb, + community, + reader, + &query.tags, + named_channels(&query.tags), + ); if count { let n: i64 = qb.build_query_scalar().fetch_one(&mut *tx).await?; tx.commit().await?; @@ -82,14 +188,66 @@ impl Db { .push_bind(query.limit) .push(" OFFSET ") .push_bind(query.offset); - let rows = qb.build().fetch_all(&mut *tx).await?; - let mut events = vec![]; - for row in rows { - if let Some(event) = crate::event::row_to_stored_event(row)? { - events.push(event); - } - } + let events = rows_to_events(qb.build().fetch_all(&mut *tx).await?)?; tx.commit().await?; Ok((events, 0)) } + + /// The current head of one artifact, including a deletion, when the + /// reader can read the head's channel. `named_channels` are the system + /// channels the caller's request names. + pub async fn artifact_head( + &self, + community: CommunityId, + reader: &[u8], + artifact: Uuid, + named_channels: &[Uuid], + ) -> Result> { + let mut connection = read_connection(self).await?; + let mut qb = QueryBuilder::::new(EVENT_COLUMNS); + qb.push("JOIN artifact_heads a ON a.community_id=e.community_id AND a.event_id=e.id AND a.artifact_id=") + .push_bind(artifact) + .push(" "); + push_scope(&mut qb, community, reader, &[], named_channels.to_vec()); + let rows = qb.build().fetch_all(&mut *connection).await?; + Ok(rows_to_events(rows)?.into_iter().next()) + } + + /// Open a sync snapshot over `channels`, after confirming in the same + /// snapshot that the reader can read every one of them. + /// `statement_timeout` bounds each page read. + pub async fn open_artifact_snapshot( + &self, + community: CommunityId, + reader: &[u8], + channels: &[Uuid], + ) -> Result> { + let connection = read_connection(self).await?; + let mut tx = sqlx::Transaction::begin(connection, None).await?; + sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY") + .execute(&mut *tx) + .await?; + sqlx::query("SET LOCAL statement_timeout='2s'") + .execute(&mut *tx) + .await?; + let readable: i64 = sqlx::query_scalar( + "SELECT count(*) FROM channels c WHERE c.community_id=$1 AND c.id=ANY($2) AND c.deleted_at IS NULL AND (c.visibility='open' OR EXISTS (SELECT 1 FROM channel_members m WHERE m.community_id=c.community_id AND m.channel_id=c.id AND m.pubkey=$3 AND m.removed_at IS NULL))", + ) + .bind(community.as_uuid()) + .bind(channels) + .bind(reader) + .fetch_one(&mut *tx) + .await?; + if readable != channels.len() as i64 { + return Ok(Err(SnapshotDenied::Restricted)); + } + Ok(Ok(ArtifactSnapshot { + tx, + community, + reader: reader.to_vec(), + channels: channels.to_vec(), + after: None, + done: false, + })) + } } diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index f3ea643ba4..b7214dc97c 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -81,6 +81,7 @@ pub mod workflow; pub mod artifact; mod artifact_query; +pub use artifact_query::{ArtifactSnapshot, SnapshotDenied}; #[cfg(test)] mod artifact_postgres_tests;