feat(relay): add incremental artifact reads

crates/buzz-db/src/error.rs

@@ -77,6 +77,10 @@ pub enum DbError {
     #[error("deletion safety error: {0}")]
     DeletionSafety(String),
 
+    /// An artifact sync page exceeds its resource limit.
+    #[error("artifact sync limit exceeded")]
+    ArtifactSyncLimit,
+
     /// A complete read-state snapshot exceeds its bounded resource budget.
     #[error("read-state snapshot exceeds event or byte limit")]
     ReadStateSnapshotTooLarge,

crates/buzz-db/src/lib.rs

@@ -65,7 +65,7 @@ pub(crate) use runtime::{
     sweep_admitted_communities, ReadSessionInner, RouteDecision, RoutePredicate,
 };
 pub use store::{
-    admin_moderation, allowlist, api_token, archived_identities, artifact, channel,
+    admin_moderation, allowlist, api_token, archived_identities, artifact, artifact_sync, channel,
     channel_members, community, deletion, dm, entity_deletion, event, feed, git_repo, moderation,
     nip_fi_binding, operator_listener, partition, personal_read, product_feedback, push, reaction,
     read_state, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder,

crates/buzz-db/src/runtime/migration.rs

@@ -776,7 +776,7 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 75);
+        assert_eq!(migrations.len(), 78);
         for (index, migration) in migrations.iter().enumerate() {
             assert_eq!(migration.version, (index + 1) as i64);
         }
@@ -2918,6 +2918,10 @@ mod postgres_tests {
             "relay_operator_audit",
             "entity_deletion_approvals",
             "entity_deletion_requests",
+            "artifact_heads",
+            "artifact_revisions",
+            "channel_members",
+            "communities",
             "entity_deletion_stage_history",
             "user_deletion_fences",
         ] {

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

@@ -185,14 +185,15 @@ impl Db {
             }
         }
         sqlx::query(
-            "INSERT INTO artifact_revisions (community_id,event_id,artifact_id) VALUES ($1,$2,$3)",
+            "INSERT INTO artifact_revisions (community_id,event_id,artifact_id,source_channel_id,modified_at) VALUES ($1,$2,$3,$4,clock_timestamp())",
         )
         .bind(community.as_uuid())
         .bind(event.id.as_bytes().as_slice())
         .bind(env.id)
+        .bind(if env.op == ArtifactOp::Move { source } else { None })
         .execute(tx.conn())
         .await?;
-        sqlx::query("INSERT INTO artifact_heads (community_id,artifact_id,event_id,channel_id,artifact_type,root,deleted) VALUES ($1,$2,$3,$4,$5,$6,$7) ON CONFLICT (community_id,artifact_id) DO UPDATE SET event_id=EXCLUDED.event_id,channel_id=EXCLUDED.channel_id,root=EXCLUDED.root,deleted=EXCLUDED.deleted")
+        sqlx::query("INSERT INTO artifact_heads (community_id,artifact_id,event_id,channel_id,artifact_type,root,deleted,modified_at) VALUES ($1,$2,$3,$4,$5,$6,$7,clock_timestamp()) ON CONFLICT (community_id,artifact_id) DO UPDATE SET event_id=EXCLUDED.event_id,channel_id=EXCLUDED.channel_id,root=EXCLUDED.root,deleted=EXCLUDED.deleted,modified_at=clock_timestamp()")
             .bind(community.as_uuid()).bind(env.id).bind(event.id.as_bytes().as_slice()).bind(env.home).bind(&env.artifact_type).bind(&env.root).bind(env.op == ArtifactOp::Delete).execute(tx.conn()).await?;
         let (stored, _) =
             crate::event::insert_event_in_transaction(&mut tx, event, Some(env.home)).await?;

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

@@ -33,46 +33,7 @@ impl Db {
         } 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 "
         });
-        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<uuid::Uuid> = 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]]));
-            }
-            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("))");
-        }
+        predicates(&mut qb, community, reader, query);
         if count {
             let n: i64 = qb.build_query_scalar().fetch_one(&mut *tx).await?;
             tx.commit().await?;
@@ -93,3 +54,68 @@ impl Db {
         Ok((events, 0))
     }
 }
+
+/// Own the visibility and tag rules for artifact queries and current-ID reads.
+pub(super) fn predicates(
+    qb: &mut QueryBuilder<sqlx::Postgres>,
+    community: CommunityId,
+    reader: &[u8],
+    query: &ArtifactQuery,
+) {
+    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 ");
+    }
+    // System channels and unlisted open channels require an explicit `#h`
+    // unless the reader is a member of the unlisted channel (NIP-CL).
+    let named_channels: Vec<uuid::Uuid> = 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)))");
+    tag_predicates(qb, query);
+    if query.view == ArtifactView::Current {
+        for (_, values) in query.tags.iter().filter(|(name, _)| name == "d") {
+            qb.push(" AND a.artifact_id=ANY(")
+                .push_bind(
+                    values
+                        .iter()
+                        .filter_map(|id| id.parse::<uuid::Uuid>().ok())
+                        .collect::<Vec<_>>(),
+                )
+                .push(")");
+        }
+    }
+}
+
+/// Apply exact artifact tag predicates after access is resolved.
+pub(super) fn tag_predicates(qb: &mut QueryBuilder<sqlx::Postgres>, query: &ArtifactQuery) {
+    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]]));
+        }
+        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("))");
+    }
+}

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

@@ -0,0 +1,205 @@
+//! Indexed artifact changes. Each page reads current authority on the writer.
+use crate::{Db, DbError, Result};
+use buzz_core::{artifact::ArtifactQuery, CommunityId, StoredEvent};
+use chrono::{DateTime, Utc};
+use futures_util::TryStreamExt;
+use serde::{Deserialize, Serialize};
+use sqlx::{QueryBuilder, Row};
+use uuid::Uuid;
+
+/// Maximum changes or requested identities in one page.
+pub const PAGE_ROWS: usize = 100;
+
+/// A stable position within one timestamp window.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct Position {
+    /// Database timestamp of the change.
+    pub at: DateTime<Utc>,
+    /// Change source, which separates equal keys from different tables.
+    pub source: i32,
+    /// Stable identity within that source.
+    pub id: String,
+}
+
+/// A changed identity that requires a current, permission-scoped read.
+pub struct Candidate {
+    /// Position used after this candidate is processed.
+    pub position: Position,
+    /// Artifact to revalidate, absent for a removed membership.
+    pub artifact_id: Option<Uuid>,
+    /// Removed channel belonging to the authenticated reader's membership.
+    pub removed_channel: Option<Uuid>,
+}
+
+impl Db {
+    /// Read server time and the community's access-policy recovery boundary.
+    pub async fn artifact_sync_boundary(
+        &self,
+        community: CommunityId,
+    ) -> Result<(DateTime<Utc>, DateTime<Utc>)> {
+        sqlx::query_as("SELECT clock_timestamp(),artifact_policy_modified_at FROM communities WHERE id=$1 AND deletion_state='active' AND deleted_at IS NULL")
+            .bind(community.as_uuid()).fetch_optional(&self.pool).await?
+            .ok_or_else(|| DbError::InvalidData("artifact sync community unavailable".into()))
+    }
+
+    /// Find at most one page plus one look-ahead row, without a channel roster.
+    pub async fn artifact_sync_candidates(
+        &self,
+        community: CommunityId,
+        reader: &[u8],
+        since: Option<DateTime<Utc>>,
+        until: DateTime<Utc>,
+        after: Option<&Position>,
+        query: &ArtifactQuery,
+    ) -> Result<Vec<Candidate>> {
+        let connection = crate::observability::acquire_writer(
+            &self.pool,
+            crate::observability::WriterOperation::SubscriptionHistory,
+        )
+        .await?;
+        let mut tx = sqlx::Transaction::begin(connection, None).await?;
+        sqlx::query("SET LOCAL statement_timeout='2s'")
+            .execute(&mut *tx)
+            .await?;
+        let types = query
+            .tags
+            .iter()
+            .filter(|(n, _)| n == "type")
+            .flat_map(|(_, v)| v.iter().cloned())
+            .collect::<Vec<_>>();
+        let named = query
+            .tags
+            .iter()
+            .filter(|(n, _)| n == "h")
+            .flat_map(|(_, v)| v.iter().filter_map(|id| id.parse::<Uuid>().ok()))
+            .collect::<Vec<_>>();
+        let rows = sqlx::query(CANDIDATES)
+            .bind(community.as_uuid())
+            .bind(reader)
+            .bind(since)
+            .bind(until)
+            .bind(after.map(|p| p.at))
+            .bind(after.map(|p| p.source))
+            .bind(after.map(|p| p.id.as_str()))
+            .bind(types)
+            .bind(named)
+            .bind((PAGE_ROWS + 1) as i64)
+            .bind(buzz_core::channel::LEGACY_CHANNEL_TYPES.to_vec())
+            .bind(query.tags.iter().any(|(name, _)| name == "h"))
+            .fetch_all(&mut *tx)
+            .await?;
+        let candidates = rows
+            .into_iter()
+            .map(|r| {
+                Ok(Candidate {
+                    position: Position {
+                        at: r.try_get("at")?,
+                        source: r.try_get("source")?,
+                        id: r.try_get("id")?,
+                    },
+                    artifact_id: r.try_get("artifact_id")?,
+                    removed_channel: r.try_get("removed_channel")?,
+                })
+            })
+            .collect::<Result<Vec<_>>>()?;
+        tx.commit().await?;
+        Ok(candidates)
+    }
+
+    /// Return current matching events for bounded known identities.
+    /// Omitted identities have the same unavailable meaning for every cause.
+    pub async fn artifact_sync_current(
+        &self,
+        community: CommunityId,
+        reader: &[u8],
+        query: &ArtifactQuery,
+        ids: &[Uuid],
+    ) -> Result<Vec<StoredEvent>> {
+        if ids.len() > PAGE_ROWS {
+            return Err(DbError::ArtifactSyncLimit);
+        }
+        if ids.is_empty() {
+            return Ok(vec![]);
+        }
+        let connection = crate::observability::acquire_writer(
+            &self.pool,
+            crate::observability::WriterOperation::SubscriptionHistory,
+        )
+        .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 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, query);
+        qb.push(" AND a.artifact_id=ANY(").push_bind(ids).push(")");
+        qb.push(" AND EXISTS (SELECT 1 FROM channels c WHERE c.community_id=e.community_id AND c.id=e.channel_id AND c.ttl_seconds IS NULL)");
+        qb.push(" LIMIT ").push_bind(PAGE_ROWS as i64);
+        let mut events = Vec::new();
+        let mut bytes = 0;
+        {
+            let mut rows = qb.build().fetch(&mut *tx);
+            while let Some(row) = rows.try_next().await? {
+                if let Some(event) = crate::event::row_to_stored_event(row)? {
+                    bytes += serde_json::to_vec(&event.event)
+                        .map_err(|e| DbError::InvalidData(e.to_string()))?
+                        .len()
+                        + 512;
+                    if bytes > 8_388_608 {
+                        return Err(DbError::ArtifactSyncLimit);
+                    }
+                    events.push(event);
+                }
+            }
+        }
+        tx.commit().await?;
+        Ok(events)
+    }
+}
+
+// Keep candidate access rules aligned with artifact_query::predicates, which owns
+// current-state visibility. This query also reads retained move and membership rows.
+const CANDIDATES: &str = r#"
+WITH readable AS NOT MATERIALIZED (
+    SELECT c.id FROM channels c
+    WHERE c.community_id=$1 AND c.deleted_at IS NULL AND c.ttl_seconds IS NULL
+      AND (NOT $12::boolean OR c.id=ANY($9))
+      AND (c.channel_type::text=ANY($11) OR c.id=ANY($9))
+      AND ((c.visibility='open' AND (NOT c.unlisted OR c.id=ANY($9))) OR EXISTS (
+        SELECT 1 FROM channel_members m WHERE m.community_id=c.community_id
+        AND m.channel_id=c.id AND m.pubkey=$2 AND m.removed_at IS NULL))
+), changes AS (
+    SELECT a.modified_at AS at,1 AS source,a.artifact_id::text AS id,
+           a.artifact_id,NULL::uuid AS removed_channel
+    FROM artifact_heads a
+    WHERE a.community_id=$1 AND a.modified_at >= COALESCE($3::timestamptz,'epoch'::timestamptz)
+      AND a.modified_at <= $4 AND a.channel_id IN (SELECT id FROM readable)
+      AND ($3::timestamptz IS NOT NULL OR NOT a.deleted)
+      AND (cardinality($8::text[])=0 OR a.artifact_type=ANY($8))
+    UNION ALL
+    SELECT m.sync_modified_at,0,m.channel_id::text,NULL::uuid,m.channel_id
+    FROM channel_members m WHERE m.community_id=$1 AND m.pubkey=$2
+      AND $3::timestamptz IS NOT NULL AND m.sync_modified_at >= $3
+      AND m.sync_modified_at <= $4 AND m.removed_at IS NOT NULL
+      AND (NOT $12::boolean OR m.channel_id=ANY($9))
+      AND NOT EXISTS (SELECT 1 FROM readable r WHERE r.id=m.channel_id)
+    UNION ALL
+    SELECT m.sync_modified_at,2,a.artifact_id::text,a.artifact_id,NULL::uuid
+    FROM channel_members m JOIN artifact_heads a
+      ON a.community_id=m.community_id AND a.channel_id=m.channel_id
+    WHERE m.community_id=$1 AND m.pubkey=$2 AND m.removed_at IS NULL
+      AND $3::timestamptz IS NOT NULL AND m.sync_modified_at >= $3
+      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 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
+)
+SELECT * FROM changes WHERE $5::timestamptz IS NULL OR (at,source,id)>($5,$6,$7)
+ORDER BY at,source,id LIMIT $10
+"#;