feat(relay): NIP-AR sync subscriptions

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<ArtifactQuery, &'static
     let view = match object.get("artifact").and_then(|v| v.as_str()) {
         Some("current") => 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<String, serde_json::Value>,
+    allowed: &[&str],
+) -> Result<Vec<(String, Vec<String>)>, &'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<ArtifactQuery, &'static
                 parsed.push(value.into());
             }
             tags.push((name.into(), parsed));
-        } else if !["artifact", "kinds", "limit", "offset"].contains(&name.as_str()) {
+        } else if !allowed.contains(&name.as_str()) {
             return Err("unsupported artifact predicate");
         }
     }
     if tags.len() > 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)
 }

crates/buzz-core/src/artifact_sync.rs

@@ -0,0 +1,132 @@
+//! NIP-AR sync filter: `{"artifact":"current", "#h":[...], ...}` on a `REQ`.
+//! The same current-state filter as HTTP `/query`; the `REQ` transport makes
+//! it a sync (complete set, `EOSE`, then live heads).
+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<Uuid>,
+    /// Exact tag-name/value predicates (AND across names, OR within one).
+    pub tags: Vec<(String, Vec<String>)>,
+}
+
+impl ArtifactSyncFilter {
+    /// Whether `head` matches every predicate with the NIP-AR match rule: a
+    /// tag with that name whose first value is one of the requested values.
+    /// The same rule the current-state query applies (NIP-AR sync rule 3).
+    pub fn matches(&self, head: &nostr::Event) -> bool {
+        self.tags.iter().all(|(name, values)| {
+            head.tags.iter().any(|tag| match tag.as_slice() {
+                [tag_name, value, ..] => tag_name == name && values.contains(value),
+                _ => false,
+            })
+        })
+    }
+}
+
+/// Whether a raw `REQ` filter is a sync: the current-state view. Other
+/// artifact views have no live form and stay HTTP-only.
+pub fn is_sync_filter(value: &serde_json::Value) -> bool {
+    value.get("artifact").and_then(|v| v.as_str()) == Some("current")
+}
+
+/// 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<ArtifactSyncFilter, &'static str> {
+    let object = value
+        .as_object()
+        .ok_or("artifact filter must be an object")?;
+    if !is_sync_filter(value) {
+        return Err("artifact must be \"current\"");
+    }
+    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":"current","#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":"current"}),
+            json!({"artifact":"current","#type":["buzz.task"]}),
+            json!({"artifact":"current","#h":[a],"limit":10}),
+            json!({"artifact":"current","#h":[a],"offset":0}),
+            json!({"artifact":"current","#h":[a],"kinds":[45011]}),
+            json!({"artifact":"current","#h":["not-a-uuid"]}),
+            json!({"artifact":"current","#h":[a],"since":1}),
+            json!({"artifact":"history","#h":[a]}),
+        ] {
+            assert!(parse_sync(&invalid).is_err(), "{invalid}");
+        }
+    }
+
+    #[test]
+    fn matches_applies_every_predicate() {
+        let h = Uuid::new_v4().to_string();
+        let filter = parse_sync(
+            &json!({"artifact":"current","#h":[h],"#type":["work.set"],"#status":["open"]}),
+        )
+        .unwrap();
+        let keys = nostr::Keys::generate();
+        let head = |tags: &[[&str; 2]]| {
+            nostr::EventBuilder::new(nostr::Kind::Custom(45010), "")
+                .tags(tags.iter().map(|t| nostr::Tag::parse(t.to_vec()).unwrap()))
+                .sign_with_keys(&keys)
+                .unwrap()
+        };
+        assert!(filter.matches(&head(&[
+            ["h", &h],
+            ["type", "work.set"],
+            ["status", "open"]
+        ])));
+        // A mutable tag that does not match, a missing tag, another type or
+        // another channel: not in scope.
+        for tags in [
+            &[["h", &h], ["type", "work.set"], ["status", "done"]][..],
+            &[["h", &h], ["type", "work.set"]][..],
+            &[["h", &h], ["type", "work.goals"], ["status", "open"]][..],
+            &[["h", "other"], ["type", "work.set"], ["status", "open"]][..],
+        ] {
+            assert!(!filter.matches(&head(tags)), "{tags:?}");
+        }
+        // Any tag of the name can match, as in the query.
+        assert!(filter.matches(&head(&[
+            ["h", &h],
+            ["type", "work.set"],
+            ["status", "done"],
+            ["status", "open"]
+        ])));
+    }
+}

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;

crates/buzz-db/src/lib.rs

@@ -86,6 +86,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;

crates/buzz-db/src/store/artifact_postgres_tests.rs

@@ -698,3 +698,135 @@ 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<Uuid> = (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::<Vec<_>>());
+
+    // 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, is_deletion) =
+        f.db.artifact_head(f.community, reader.as_bytes(), ids[4], &[])
+            .await
+            .unwrap()
+            .unwrap();
+    assert_eq!(head.event.id, deleted.id);
+    assert!(is_deletion);
+    let (head, is_deletion) =
+        f.db.artifact_head(f.community, reader.as_bytes(), ids[0], &[])
+            .await
+            .unwrap()
+            .unwrap();
+    assert_eq!(head.event.id, heads[0].id);
+    assert!(!is_deletion);
+    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());
+}