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;