BUZZ-413: Fence stream and forum message writes

crates/buzz-db/src/error.rs

@@ -37,6 +37,10 @@ pub enum DbError {
     #[error("access denied: {0}")]
     AccessDenied(String),
 
+    /// A permanent user-deletion restriction refuses this serving write.
+    #[error("identity is restricted for user deletion")]
+    UserDeletionRestricted,
+
     /// JSON serialization or deserialization failed.
     #[error("serialization error: {0}")]
     Serde(#[from] serde_json::Error),

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

@@ -145,6 +145,25 @@ impl AdmittedTx {
         &mut self.tx
     }
 
+    pub(crate) async fn guard_message_author(&mut self, event: &nostr::Event) -> Result<()> {
+        crate::store::user_deletion_write_guard::guard_message(&mut self.tx, self.community, event)
+            .await
+    }
+
+    pub(crate) async fn guard_agent_owner_assignment(
+        &mut self,
+        agent: &[u8],
+        owner: &[u8],
+    ) -> Result<()> {
+        crate::store::user_deletion_write_guard::guard_owner_assignment(
+            &mut self.tx,
+            self.community,
+            agent,
+            owner,
+        )
+        .await
+    }
+
     /// Record that an event committed with this transaction belongs to
     /// `channel_id`, so [`AdmittedTx::commit`] refreshes the channel's TTL.
     pub(crate) fn record_channel_event(&mut self, channel_id: Option<Uuid>, kind: i32) {

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

@@ -477,6 +477,7 @@ async fn insert_event_on(
         return Err(DbError::EphemeralEventRejected(kind_u16));
     }
 
+    tx.guard_message_author(event).await?;
     let id_bytes = event.id.as_bytes();
     let pubkey_bytes = event.pubkey.to_bytes();
     let sig_bytes = event.sig.serialize();
@@ -1865,6 +1866,7 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
         return Err(DbError::EphemeralEventRejected(kind_u16));
     }
 
+    tx.guard_message_author(event).await?;
     let id_bytes = event.id.as_bytes();
     let pubkey_bytes = event.pubkey.to_bytes();
     let sig_bytes = event.sig.serialize();

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

@@ -30,6 +30,10 @@ pub mod nip_fi_binding;
 pub mod user_deletion;
 /// Operator-global user restrictions and identity-lock primitives.
 pub mod user_deletion_fences;
+pub(crate) mod user_deletion_write_guard;
+
+#[cfg(test)]
+mod user_deletion_write_guard_postgres_tests;
 
 #[cfg(test)]
 mod event_follow_up_postgres_tests;

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

@@ -354,6 +354,8 @@ async fn set_agent_owner_with_operation(
     operation: crate::observability::WriterOperation,
 ) -> Result<bool> {
     let mut tx = crate::begin_community_write_transaction(pool, community_id, operation).await?;
+    tx.guard_agent_owner_assignment(agent_pubkey, owner_pubkey)
+        .await?;
     // Conditional UPDATE: only set owner if currently NULL. This makes
     // "first mint wins" atomic — no TOCTOU race between concurrent mints.
     let result = sqlx::query(

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

@@ -1,5 +1,6 @@
-//! Operator-global restriction storage and identity locks; no serving or worker
-//! path calls these helpers yet. Installing fences does not itself enforce them.
+//! Operator-global restriction storage and identity locks. New stream/forum
+//! messages and agent-owner assignment consult these through the write guard.
+//! Other serving paths and the real fencing worker are not integrated yet.
 //!
 //! Callers use a writer-pool READ COMMITTED transaction. Acquire existing
 //! community guards first, then `lock_request` for the worker, then all human
@@ -150,19 +151,20 @@ async fn lock(
 /// For multiple identities, acquire the full shared set first. The caller must
 /// refuse a restricted operation and retain this transaction through any write.
 /// An agent lookup is scoped; it does not resolve or check the human owner.
-/// The planned agent-write path locks the verified owner and agent together,
-/// then reads the agent restriction only. This requires the gate to install the
+/// The message-write path locks the verified owner and agent together, then
+/// reads the signer's restrictions, not the owner's. This requires the gate to install the
 /// human and all owned-agent restrictions atomically, and creation/assignment
-/// paths to reject restricted owners. These callers are not wired in yet.
+/// paths to reject restricted owners. The real gate is not wired in yet.
 pub async fn find(
     tx: &mut Transaction<'_, Postgres>,
     identity: FenceIdentity,
 ) -> Result<Option<UserDeletionFence>> {
     lock_shared(tx, &[identity]).await?;
-    read(tx, identity).await
+    read_locked(tx, identity).await
 }
 
-async fn read(
+/// Read one exact restriction after the caller has locked its identity.
+pub(super) async fn read_locked(
     tx: &mut Transaction<'_, Postgres>,
     identity: FenceIdentity,
 ) -> Result<Option<UserDeletionFence>> {
@@ -226,7 +228,7 @@ pub async fn insert(
     )
     .bind(identity.pubkey.to_hex()).bind(identity.scope_id()).bind(request_id).bind(&owner)
     .execute(&mut **tx).await?;
-    let saved = read(tx, identity)
+    let saved = read_locked(tx, identity)
         .await?
         .ok_or_else(|| DbError::DeletionSafety("fence insertion returned no row".into()))?;
     if saved.request_id != request_id || saved.source_owner_pubkey != owner {

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

@@ -0,0 +1,108 @@
+//! First serving integration: new stream/forum messages and agent ownership.
+//! Other writes are not fenced here; real deletion must remain disabled until
+//! every serving path and external-work drain has been integrated.
+
+use buzz_core::{kind, CommunityId};
+use nostr::{Event, PublicKey};
+use sqlx::{Postgres, Transaction};
+
+use super::user_deletion_fences::{lock_exclusive, lock_shared, read_locked, FenceIdentity};
+use crate::{DbError, Result};
+
+pub(crate) async fn guard_message(
+    tx: &mut Transaction<'_, Postgres>,
+    community: CommunityId,
+    event: &Event,
+) -> Result<()> {
+    if !matches!(
+        u32::from(event.kind.as_u16()),
+        kind::KIND_STREAM_MESSAGE
+            | kind::KIND_STREAM_MESSAGE_V2
+            | kind::KIND_FORUM_POST
+            | kind::KIND_FORUM_COMMENT
+    ) {
+        return Ok(());
+    }
+
+    let signer = event.pubkey;
+    let owner = owner(tx, community, signer).await?;
+    let mut identities = vec![
+        FenceIdentity::user(signer),
+        FenceIdentity::agent(community, signer),
+    ];
+    if let Some(owner) = owner {
+        identities.push(FenceIdentity::user(owner));
+    }
+    // Global signer lock also coordinates the first assignment of an owner.
+    // Never derive the owner from a cached session or an event tag.
+    lock_shared(tx, &identities).await?;
+    if owner != self::owner(tx, community, signer).await? {
+        // Restart in a new transaction: adding another owner lock now could
+        // invert the global-before-scoped ordering. No write has occurred.
+        return Err(DbError::AccessDenied(
+            "agent ownership changed; retry the write".into(),
+        ));
+    }
+    check_signer(tx, community, signer).await
+}
+
+pub(crate) async fn guard_owner_assignment(
+    tx: &mut Transaction<'_, Postgres>,
+    community: CommunityId,
+    agent: &[u8],
+    owner: &[u8],
+) -> Result<()> {
+    let agent = parse_key(agent)?;
+    let owner = parse_key(owner)?;
+    lock_shared(
+        tx,
+        &[FenceIdentity::user(agent), FenceIdentity::user(owner)],
+    )
+    .await?;
+    // Ownership must not change while a message holds the scoped shared lock.
+    lock_exclusive(tx, &[FenceIdentity::agent(community, agent)]).await?;
+    if read_locked(tx, FenceIdentity::user(owner)).await?.is_some() {
+        return Err(DbError::UserDeletionRestricted);
+    }
+    check_signer(tx, community, agent).await
+}
+
+async fn owner(
+    tx: &mut Transaction<'_, Postgres>,
+    community: CommunityId,
+    signer: PublicKey,
+) -> Result<Option<PublicKey>> {
+    let owner: Option<Option<Vec<u8>>> = sqlx::query_scalar(
+        "SELECT agent_owner_pubkey FROM users WHERE community_id = $1 AND pubkey = $2",
+    )
+    .bind(community.as_uuid())
+    .bind(signer.to_bytes().as_slice())
+    .fetch_optional(&mut **tx)
+    .await?;
+    owner.flatten().map(|key| parse_key(&key)).transpose()
+}
+
+fn parse_key(key: &[u8]) -> Result<PublicKey> {
+    PublicKey::from_slice(key)
+        .map_err(|_| DbError::InvalidData("invalid user-deletion identity key".into()))
+}
+
+async fn check_signer(
+    tx: &mut Transaction<'_, Postgres>,
+    community: CommunityId,
+    signer: PublicKey,
+) -> Result<()> {
+    // Read the signer's restrictions, not the owner's. Check both possible
+    // scopes even if the profile/ownership row is absent: those mutable rows
+    // must not determine whether a permanent restriction still applies.
+    if read_locked(tx, FenceIdentity::user(signer))
+        .await?
+        .is_some()
+        || read_locked(tx, FenceIdentity::agent(community, signer))
+            .await?
+            .is_some()
+    {
+        return Err(DbError::UserDeletionRestricted);
+    }
+    Ok(())
+}