fix(relay): serialize archival snapshot publication

Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>

crates/buzz-db/src/lib.rs

@@ -4960,6 +4960,129 @@ impl Db {
         Ok(snapshot_members != canonical_members)
     }
 
+    /// Atomically publish a NIP-IA archived-identities snapshot under the
+    /// replaceable event's transaction-scoped advisory lock.
+    ///
+    /// The lock is acquired before canonical archive state and the prior event
+    /// head are read, so concurrent publishers cannot capture different states,
+    /// choose the same timestamp, and fall back to event-ID ordering.
+    #[datastore_span(name = "publish_nipia_archival_list_locked", system = "postgresql")]
+    pub async fn publish_nipia_archival_list_locked(
+        &self,
+        community_id: CommunityId,
+        relay_keypair: &nostr::Keys,
+    ) -> Result<(StoredEvent, bool, usize)> {
+        use nostr::{EventBuilder, Kind, Tag};
+
+        let kind_i32 = buzz_core::kind::KIND_IA_ARCHIVED_LIST as i32;
+        let pubkey_bytes = relay_keypair.public_key().to_bytes();
+        let lock_key =
+            event_replacement_lock_key(community_id, kind_i32, pubkey_bytes.as_slice(), None);
+        let mut tx = self.pool.begin().await?;
+
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(lock_key)
+            .execute(&mut *tx)
+            .await?;
+
+        let archived_pubkeys: Vec<String> = sqlx::query_scalar(
+            "SELECT pubkey FROM archived_identities \
+             WHERE community_id = $1 ORDER BY archived_at ASC",
+        )
+        .bind(community_id.as_uuid())
+        .fetch_all(&mut *tx)
+        .await?;
+
+        let existing_created_at: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
+            "SELECT created_at FROM events \
+             WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \
+             AND channel_id IS NULL AND deleted_at IS NULL \
+             ORDER BY created_at DESC, id ASC LIMIT 1",
+        )
+        .bind(community_id.as_uuid())
+        .bind(kind_i32)
+        .bind(pubkey_bytes.as_slice())
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        let now = chrono::Utc::now().timestamp().max(0);
+        let created_at = existing_created_at
+            .map(|timestamp| timestamp.timestamp().saturating_add(1).max(now))
+            .unwrap_or(now);
+
+        let mut tags = Vec::with_capacity(archived_pubkeys.len() + 1);
+        tags.push(
+            Tag::parse(["-"])
+                .map_err(|e| DbError::InvalidData(format!("failed to build '-' tag: {e}")))?,
+        );
+        for pubkey in &archived_pubkeys {
+            tags.push(
+                Tag::parse(["p", pubkey])
+                    .map_err(|e| DbError::InvalidData(format!("failed to build p tag: {e}")))?,
+            );
+        }
+
+        let event = EventBuilder::new(Kind::Custom(kind_i32 as u16), "")
+            .tags(tags)
+            .custom_created_at(nostr::Timestamp::from(created_at as u64))
+            .sign_with_keys(relay_keypair)
+            .map_err(|e| DbError::InvalidData(format!("failed to sign kind:{kind_i32}: {e}")))?;
+        let event_created_at = chrono::DateTime::from_timestamp(created_at, 0)
+            .ok_or(DbError::InvalidTimestamp(created_at))?;
+        let sig_bytes = event.sig.serialize();
+        let tags_json = serde_json::to_value(&event.tags)?;
+        let received_at = chrono::Utc::now();
+
+        sqlx::query(
+            "UPDATE events SET deleted_at = NOW() \
+             WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \
+             AND channel_id IS NULL AND deleted_at IS NULL",
+        )
+        .bind(community_id.as_uuid())
+        .bind(kind_i32)
+        .bind(pubkey_bytes.as_slice())
+        .execute(&mut *tx)
+        .await?;
+
+        let insert_result = sqlx::query(
+            "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \
+             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NULL, NULL) \
+             ON CONFLICT DO NOTHING",
+        )
+        .bind(community_id.as_uuid())
+        .bind(event.id.as_bytes().as_slice())
+        .bind(pubkey_bytes.as_slice())
+        .bind(event_created_at)
+        .bind(kind_i32)
+        .bind(&tags_json)
+        .bind(&event.content)
+        .bind(sig_bytes.as_slice())
+        .bind(received_at)
+        .execute(&mut *tx)
+        .await?;
+
+        let was_inserted = insert_result.rows_affected() > 0;
+        if !was_inserted {
+            tx.rollback().await?;
+            return Ok((
+                StoredEvent::with_received_at(event, received_at, None, false),
+                false,
+                archived_pubkeys.len(),
+            ));
+        }
+
+        tx.commit().await?;
+        if let Err(e) = crate::insert_mentions(&self.pool, community_id, &event, None).await {
+            tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
+        }
+
+        Ok((
+            StoredEvent::with_received_at(event, received_at, None, true),
+            true,
+            archived_pubkeys.len(),
+        ))
+    }
+
     /// Atomically publish a NIP-43 membership snapshot under a single
     /// transaction-scoped advisory lock.
     ///
@@ -5438,6 +5561,97 @@ mod tests {
         id
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn nipia_snapshot_publication_serializes_state_capture_with_replacement() {
+        use nostr::Keys;
+
+        let db = setup_db().await;
+        let community = CommunityId::from_uuid(make_community(&db.pool).await);
+        let relay_keys = Keys::generate();
+        let target = Keys::generate().public_key().to_hex();
+
+        db.publish_nipia_archival_list_locked(community, &relay_keys)
+            .await
+            .expect("publish initial empty snapshot");
+        db.archive(
+            community,
+            &target,
+            "owner",
+            &"a".repeat(64),
+            None,
+            None,
+            &"b".repeat(64),
+        )
+        .await
+        .expect("archive target");
+
+        let lock_key = event_replacement_lock_key(
+            community,
+            buzz_core::kind::KIND_IA_ARCHIVED_LIST as i32,
+            relay_keys.public_key().to_bytes().as_slice(),
+            None,
+        );
+        let mut gate = db.pool.begin().await.expect("begin publication gate");
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(lock_key)
+            .execute(&mut *gate)
+            .await
+            .expect("hold publication lock");
+
+        let first_db = db.clone();
+        let first_keys = relay_keys.clone();
+        let first = tokio::spawn(async move {
+            first_db
+                .publish_nipia_archival_list_locked(community, &first_keys)
+                .await
+        });
+
+        // The first publisher is blocked before canonical state capture. Model
+        // the final transition while it waits, then queue its publisher too.
+        tokio::task::yield_now().await;
+        db.unarchive(community, &target)
+            .await
+            .expect("unarchive target");
+        let second_db = db.clone();
+        let second_keys = relay_keys.clone();
+        let second = tokio::spawn(async move {
+            second_db
+                .publish_nipia_archival_list_locked(community, &second_keys)
+                .await
+        });
+
+        gate.commit().await.expect("release publication gate");
+        first
+            .await
+            .expect("join first publisher")
+            .expect("first publish");
+        second
+            .await
+            .expect("join second publisher")
+            .expect("second publish");
+
+        let stored = db
+            .get_latest_global_replaceable(
+                community,
+                buzz_core::kind::KIND_IA_ARCHIVED_LIST as i32,
+                relay_keys.public_key().to_bytes().as_slice(),
+            )
+            .await
+            .expect("query snapshot")
+            .expect("stored snapshot");
+        let archived_tags = stored
+            .event
+            .tags
+            .iter()
+            .filter(|tag| tag.as_slice().first().map(String::as_str) == Some("p"))
+            .count();
+        assert_eq!(
+            archived_tags, 0,
+            "snapshot must match final unarchived state"
+        );
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn nip_rs_replacement_hard_deletes_payload_and_watermark_rejects_replay() {

crates/buzz-relay/src/handlers/side_effects.rs

@@ -3106,68 +3106,22 @@ pub async fn reconcile_channel_events(
     Ok(())
 }
 
-fn next_snapshot_timestamp(now: u64, current: Option<u64>) -> u64 {
-    current
-        .map(|created_at| created_at.saturating_add(1).max(now))
-        .unwrap_or(now)
-}
-
 /// Publish a kind:13535 archived identities list event (NIP-IA).
 ///
-/// Queries all current archived identities and emits a relay-signed,
-/// NIP-70-protected replaceable-by-convention snapshot with bare `p` tags.
+/// The canonical archive-state read, timestamp allocation, and replacement are
+/// serialized inside the database under the event's advisory lock. That durable
+/// boundary prevents overlapping publishers on different relay nodes from
+/// capturing different states at the same replacement timestamp.
 pub async fn publish_nipia_archival_list(
     tenant: &TenantContext,
     state: &Arc<AppState>,
 ) -> anyhow::Result<()> {
-    let archived = state.db.list_archived(tenant.community()).await?;
-    let relay_pubkey = state.relay_keypair.public_key();
-    let relay_pubkey_hex = relay_pubkey.to_hex();
-
-    let mut tags: Vec<Tag> = Vec::with_capacity(archived.len() + 1);
-    tags.push(Tag::parse(["-"]).map_err(|e| anyhow::anyhow!("failed to build '-' tag: {e}"))?);
-
-    for identity in &archived {
-        tags.push(
-            Tag::parse(["p", &identity.pubkey])
-                .map_err(|e| anyhow::anyhow!("failed to build p tag: {e}"))?,
-        );
-    }
-
-    // Nostr timestamps have one-second resolution, while archive and unarchive
-    // can complete within the same second. A same-second snapshot would enter
-    // NIP-16's event-id tie-break and could lose to the prior snapshot, leaving
-    // clients with stale archive state. Advance strictly past the stored head so
-    // every accepted identity-state transition produces the authoritative head.
-    let now = std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .unwrap_or_default()
-        .as_secs();
-    let existing = state
+    let relay_pubkey_hex = state.relay_keypair.public_key().to_hex();
+    let (stored, was_inserted, archived_count) = state
         .db
-        .get_latest_global_replaceable(
-            tenant.community(),
-            KIND_IA_ARCHIVED_LIST as i32,
-            relay_pubkey.to_bytes().as_slice(),
-        )
+        .publish_nipia_archival_list_locked(tenant.community(), &state.relay_keypair)
         .await?;
-    let created_at = next_snapshot_timestamp(
-        now,
-        existing
-            .as_ref()
-            .map(|stored| stored.event.created_at.as_secs()),
-    );
 
-    let event = EventBuilder::new(Kind::Custom(KIND_IA_ARCHIVED_LIST as u16), "")
-        .tags(tags)
-        .custom_created_at(nostr::Timestamp::from(created_at))
-        .sign_with_keys(&state.relay_keypair)
-        .map_err(|e| anyhow::anyhow!("failed to sign kind:{KIND_IA_ARCHIVED_LIST}: {e}"))?;
-
-    let (stored, was_inserted) = state
-        .db
-        .replace_addressable_event(tenant.community(), &event, None)
-        .await?;
     if was_inserted {
         dispatch_persistent_event(
             tenant,
@@ -3180,10 +3134,7 @@ pub async fn publish_nipia_archival_list(
         .await;
     }
 
-    info!(
-        archived_count = archived.len(),
-        "NIP-IA archived identities list published"
-    );
+    info!(archived_count, "NIP-IA archived identities list published");
     Ok(())
 }
 
@@ -3404,23 +3355,6 @@ fn topic_for_subscription(channel_id: Option<Uuid>) -> EventTopic {
 mod tests {
     use super::*;
 
-    #[test]
-    fn identity_archive_snapshot_advances_past_same_second_head() {
-        assert_eq!(next_snapshot_timestamp(1_000, Some(1_000)), 1_001);
-        assert_eq!(next_snapshot_timestamp(1_000, Some(1_001)), 1_002);
-    }
-
-    #[test]
-    fn identity_archive_snapshot_uses_wall_clock_when_it_is_newer() {
-        assert_eq!(next_snapshot_timestamp(1_001, Some(1_000)), 1_001);
-        assert_eq!(next_snapshot_timestamp(1_001, None), 1_001);
-    }
-
-    #[test]
-    fn identity_archive_snapshot_timestamp_saturates() {
-        assert_eq!(next_snapshot_timestamp(u64::MAX, Some(u64::MAX)), u64::MAX);
-    }
-
     #[test]
     fn delete_tombstone_omits_absent_moderation_metadata() {
         let content =