feat(auth): persist current client status projections

Signed-off-by: Cea Stapleton Cordasco <261786559+cea-block@users.noreply.github.com>

crates/buzz-core/src/kind.rs

@@ -85,6 +85,12 @@ pub const KIND_AUTH: u32 = 22242;
 pub const KIND_BLOSSOM_AUTH: u32 = 24242;
 /// Buzz custom one-time identity binding proof (ephemeral, not stored).
 pub const KIND_NOSTR_IDENTITY_BINDING: u32 = 24243;
+/// Buzz relay-authenticated client binding status (ephemeral, not stored).
+///
+/// This provisional allocation carries short-lived, display-only status. It
+/// is intentionally absent from relay ingest and storage allowlists until the
+/// binding lifecycle and client-presentation joins are complete.
+pub const KIND_CLIENT_BINDING_STATUS: u32 = 24244;
 /// NIP-98: HTTP auth event (used in nip98.rs, not stored).
 pub const KIND_HTTP_AUTH: u32 = 27235;
 
@@ -823,6 +829,7 @@ pub const fn is_relay_only_kind(kind: u32) -> bool {
     matches!(
         kind,
         KIND_NIP43_MEMBERSHIP_LIST
+            | KIND_CLIENT_BINDING_STATUS
             | KIND_CHANNEL_SUMMARY
             | KIND_PRESENCE_SNAPSHOT
             | KIND_DM_VISIBILITY
@@ -904,6 +911,12 @@ mod tests {
         assert!(!is_relay_only_kind(KIND_NIP43_LEAVE_REQUEST));
     }
 
+    #[test]
+    fn client_binding_status_is_relay_only() {
+        assert!(is_relay_only_kind(KIND_CLIENT_BINDING_STATUS));
+        assert!(is_ephemeral(KIND_CLIENT_BINDING_STATUS));
+    }
+
     #[test]
     fn parameterized_replaceable_range() {
         assert!(!is_parameterized_replaceable(29999));

crates/buzz-core/src/lib.rs

@@ -9,6 +9,8 @@
 pub mod agent_turn_metric;
 /// Channel and membership enums shared across crates.
 pub mod channel;
+/// Relay-authenticated, display-only client binding status contract.
+pub mod client_binding_status;
 /// NIP-AE Agent Engrams — slug grammar, conversation key, d-tag derivation,
 /// body parse/serialize, envelope build/validate, head selection.
 pub mod engram;

crates/buzz-db/src/client_status.rs

@@ -0,0 +1,710 @@
+//! Durable, current-only client verification-status revisions.
+//!
+//! Allocation is transaction-owned: the exact active binding, membership,
+//! invalidation generation/floors, database-clock freshness, revision row,
+//! authority epoch, and idempotency receipt are committed together.
+
+use buzz_core::CommunityId;
+use sqlx::{Postgres, Row, Transaction};
+use thiserror::Error;
+use uuid::Uuid;
+
+use crate::authorization_invalidation::AuthorizationSelector;
+use crate::Db;
+
+const CURRENT_KIND: &str = "client.status.current.v1";
+const WITHDRAW_KIND: &str = "client.status.withdraw.v1";
+
+/// Exact private requirement for one current-status issuance.
+pub struct CurrentStatusAllocation<'a> {
+    /// Server-resolved authorization domain.
+    pub community_id: CommunityId,
+    /// Exact event-author key.
+    pub event_author_pubkey: &'a [u8; 32],
+    /// Stable active binding ID.
+    pub binding_id: Uuid,
+    /// Exact positive binding version.
+    pub binding_version: u64,
+    /// Opaque current provider policy version.
+    pub policy_version: &'a str,
+    /// Invalidation generation captured before provider evaluation.
+    pub evaluation_generation: u64,
+    /// Database-clock freshness boundary.
+    pub fresh_until: u64,
+    /// Stable issuance operation ID.
+    pub operation_id: Uuid,
+    /// Exact event-input fingerprint.
+    pub request_fingerprint: [u8; 32],
+}
+
+/// Exact private requirement for an opaque withdrawal.
+pub struct WithdrawalStatusAllocation<'a> {
+    /// Server-resolved authorization domain.
+    pub community_id: CommunityId,
+    /// Exact event-author key.
+    pub event_author_pubkey: &'a [u8; 32],
+    /// Revision of the actual current issuance being withdrawn.
+    pub supersedes_revision: u64,
+    /// Fingerprint of the durable current issuance receipt.
+    pub issuance_fingerprint: [u8; 32],
+    /// Stable withdrawal operation ID.
+    pub operation_id: Uuid,
+    /// Exact withdrawal-input fingerprint.
+    pub request_fingerprint: [u8; 32],
+}
+
+/// Result of a transaction-owned revision allocation.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct AllocatedStatusRevision {
+    /// Strictly positive revision.
+    pub revision: u64,
+    /// Domain-wide durable status floor after this allocation.
+    pub floor: u64,
+}
+
+/// Allocation failure classified by whether PostgreSQL commit was attempted.
+#[derive(Debug, Error)]
+pub enum ClientStatusAllocationError {
+    /// Current private authority did not satisfy the exact requirement.
+    #[error("client status authority is not current")]
+    NotCurrent,
+    /// A stable operation ID was reused for different input.
+    #[error("client status operation conflicts with a committed request")]
+    ConflictingRetry,
+    /// Input could not be represented safely.
+    #[error("client status allocation input is invalid")]
+    InvalidInput,
+    /// PostgreSQL failed before commit was attempted; the transaction rolls back.
+    #[error("client status allocation failed before commit")]
+    Database(#[source] sqlx::Error),
+    /// PostgreSQL commit acknowledgement was ambiguous. The receipt decides.
+    #[error("client status commit acknowledgement is ambiguous")]
+    CommitUnknown(#[source] sqlx::Error),
+}
+
+impl From<sqlx::Error> for ClientStatusAllocationError {
+    fn from(error: sqlx::Error) -> Self {
+        Self::Database(error)
+    }
+}
+
+impl Db {
+    /// Read an exact committed status allocation after ambiguous commit acknowledgement.
+    pub async fn committed_status_revision(
+        &self,
+        community_id: CommunityId,
+        operation_id: Uuid,
+        request_fingerprint: [u8; 32],
+    ) -> Result<Option<AllocatedStatusRevision>, ClientStatusAllocationError> {
+        let row = sqlx::query(
+            "SELECT request_fingerprint, result_payload FROM authorization_operation_receipts \
+             WHERE community_id=$1 AND operation_id=$2",
+        )
+        .bind(community_id.as_uuid())
+        .bind(operation_id)
+        .fetch_optional(&self.pool)
+        .await?;
+        let Some(row) = row else { return Ok(None) };
+        let fingerprint: Vec<u8> = row.try_get("request_fingerprint")?;
+        let payload: Vec<u8> = row.try_get("result_payload")?;
+        if fingerprint.as_slice() != request_fingerprint {
+            return Err(ClientStatusAllocationError::ConflictingRetry);
+        }
+        Ok(Some(decode_receipt_payload(&payload)?))
+    }
+
+    /// Allocate or replay one strictly monotonic current-status revision.
+    pub async fn allocate_current_status_revision(
+        &self,
+        request: CurrentStatusAllocation<'_>,
+    ) -> Result<AllocatedStatusRevision, ClientStatusAllocationError> {
+        validate_common(
+            request.community_id,
+            request.event_author_pubkey,
+            request.operation_id,
+            request.binding_version,
+        )?;
+        if request.policy_version.is_empty()
+            || request.evaluation_generation > i64::MAX as u64
+            || request.fresh_until > i64::MAX as u64
+        {
+            return Err(ClientStatusAllocationError::InvalidInput);
+        }
+        let mut tx = self.begin_transaction().await.map_err(db_error)?;
+        lock_scope(&mut tx, request.community_id, request.event_author_pubkey).await?;
+        let (issuer, subject) = validate_current_authority(&mut tx, &request).await?;
+        validate_invalidation(
+            &mut tx,
+            request.community_id,
+            request.evaluation_generation,
+            request.binding_id,
+            request.binding_version,
+            request.event_author_pubkey,
+            request.policy_version,
+            &issuer,
+            &subject,
+        )
+        .await?;
+        if let Some(revision) = replay_revision(
+            &mut tx,
+            request.community_id,
+            request.operation_id,
+            CURRENT_KIND,
+            request.request_fingerprint,
+        )
+        .await?
+        {
+            tx.commit()
+                .await
+                .map_err(ClientStatusAllocationError::CommitUnknown)?;
+            return Ok(revision);
+        }
+        let allocated = next_revision(&mut tx, request.community_id).await?;
+        sqlx::query(
+            "INSERT INTO client_status_revisions \
+             (community_id, event_author_pubkey, revision, disposition, binding_id, binding_version) \
+             VALUES ($1, $2, $3, 'current', $4, $5) \
+         ON CONFLICT (community_id, event_author_pubkey) DO UPDATE SET \
+           revision=EXCLUDED.revision, disposition='current', binding_id=EXCLUDED.binding_id, \
+           binding_version=EXCLUDED.binding_version, supersedes_revision=NULL, \
+           updated_at=clock_timestamp()",
+        )
+        .bind(request.community_id.as_uuid())
+        .bind(request.event_author_pubkey.as_slice())
+        .bind(allocated as i64)
+        .bind(request.binding_id)
+        .bind(request.binding_version as i64)
+        .execute(&mut *tx)
+        .await?;
+        insert_receipt(
+            &mut tx,
+            request.community_id,
+            request.operation_id,
+            CURRENT_KIND,
+            request.request_fingerprint,
+            allocated,
+        )
+        .await?;
+        tx.commit()
+            .await
+            .map_err(ClientStatusAllocationError::CommitUnknown)?;
+        Ok(AllocatedStatusRevision {
+            revision: allocated,
+            floor: allocated,
+        })
+    }
+
+    /// Allocate or replay a withdrawal strictly after its exact current receipt.
+    pub async fn allocate_withdrawn_status_revision(
+        &self,
+        request: WithdrawalStatusAllocation<'_>,
+    ) -> Result<AllocatedStatusRevision, ClientStatusAllocationError> {
+        validate_common(
+            request.community_id,
+            request.event_author_pubkey,
+            request.operation_id,
+            request.supersedes_revision,
+        )?;
+        let mut tx = self.begin_transaction().await.map_err(db_error)?;
+        lock_scope(&mut tx, request.community_id, request.event_author_pubkey).await?;
+        if let Some(revision) = replay_revision(
+            &mut tx,
+            request.community_id,
+            request.operation_id,
+            WITHDRAW_KIND,
+            request.request_fingerprint,
+        )
+        .await?
+        {
+            tx.commit()
+                .await
+                .map_err(ClientStatusAllocationError::CommitUnknown)?;
+            return Ok(revision);
+        }
+        let issuance_exists: bool = sqlx::query_scalar(
+            "SELECT EXISTS(SELECT 1 FROM authorization_operation_receipts \
+             WHERE community_id=$1 AND operation_kind=$2 AND request_fingerprint=$3 \
+               AND octet_length(result_payload) IN (8,16) \
+               AND substring(result_payload FROM 1 FOR 8)=$4)",
+        )
+        .bind(request.community_id.as_uuid())
+        .bind(CURRENT_KIND)
+        .bind(request.issuance_fingerprint.as_slice())
+        .bind(request.supersedes_revision.to_be_bytes().as_slice())
+        .fetch_one(&mut *tx)
+        .await?;
+        if !issuance_exists {
+            return Err(ClientStatusAllocationError::NotCurrent);
+        }
+        let row: Option<(i64, String, Option<i64>)> = sqlx::query_as(
+            "SELECT revision, disposition, supersedes_revision FROM client_status_revisions \
+             WHERE community_id=$1 AND event_author_pubkey=$2 FOR UPDATE",
+        )
+        .bind(request.community_id.as_uuid())
+        .bind(request.event_author_pubkey.as_slice())
+        .fetch_optional(&mut *tx)
+        .await?;
+        let Some((revision, disposition, prior_supersedes)) = row else {
+            return Err(ClientStatusAllocationError::NotCurrent);
+        };
+        let receipt_revision = request.supersedes_revision as i64;
+        let superseded_revision = if disposition == "current" {
+            if receipt_revision > revision {
+                return Err(ClientStatusAllocationError::NotCurrent);
+            }
+            revision
+        } else if disposition == "withdrawn"
+            && prior_supersedes.is_some_and(|superseded| receipt_revision <= superseded)
+            && revision > receipt_revision
+        {
+            prior_supersedes.expect("withdrawn status has a superseded revision")
+        } else {
+            return Err(ClientStatusAllocationError::NotCurrent);
+        };
+        let floor: i64 = sqlx::query_scalar(
+            "SELECT status_revision FROM authorization_authority_epochs \
+             WHERE community_id=$1 FOR UPDATE",
+        )
+        .bind(request.community_id.as_uuid())
+        .fetch_one(&mut *tx)
+        .await?;
+        let allocated = if disposition == "current" || revision < floor {
+            next_revision(&mut tx, request.community_id).await?
+        } else {
+            revision as u64
+        };
+        if allocated <= superseded_revision as u64 {
+            return Err(ClientStatusAllocationError::NotCurrent);
+        }
+        sqlx::query(
+            "UPDATE client_status_revisions SET revision=$3, disposition='withdrawn', \
+             binding_id=NULL, binding_version=NULL, supersedes_revision=$4, \
+             updated_at=clock_timestamp() \
+             WHERE community_id=$1 AND event_author_pubkey=$2",
+        )
+        .bind(request.community_id.as_uuid())
+        .bind(request.event_author_pubkey.as_slice())
+        .bind(allocated as i64)
+        .bind(superseded_revision)
+        .execute(&mut *tx)
+        .await?;
+        insert_receipt(
+            &mut tx,
+            request.community_id,
+            request.operation_id,
+            WITHDRAW_KIND,
+            request.request_fingerprint,
+            allocated,
+        )
+        .await?;
+        tx.commit()
+            .await
+            .map_err(ClientStatusAllocationError::CommitUnknown)?;
+        Ok(AllocatedStatusRevision {
+            revision: allocated,
+            floor: allocated,
+        })
+    }
+}
+
+fn validate_common(
+    community_id: CommunityId,
+    pubkey: &[u8; 32],
+    operation_id: Uuid,
+    positive: u64,
+) -> Result<(), ClientStatusAllocationError> {
+    if community_id.as_uuid().is_nil()
+        || operation_id.is_nil()
+        || positive == 0
+        || positive > i64::MAX as u64
+        || pubkey.iter().all(|byte| *byte == 0)
+    {
+        return Err(ClientStatusAllocationError::InvalidInput);
+    }
+    Ok(())
+}
+
+fn db_error(error: crate::DbError) -> ClientStatusAllocationError {
+    match error {
+        crate::DbError::Sqlx(error) => ClientStatusAllocationError::Database(error),
+        _ => ClientStatusAllocationError::InvalidInput,
+    }
+}
+
+async fn lock_scope(
+    tx: &mut Transaction<'static, Postgres>,
+    community_id: CommunityId,
+    pubkey: &[u8; 32],
+) -> Result<(), ClientStatusAllocationError> {
+    sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
+        .bind(format!(
+            "client-status:{}:{}",
+            community_id,
+            hex::encode(pubkey)
+        ))
+        .execute(&mut **tx)
+        .await?;
+    Ok(())
+}
+
+async fn validate_current_authority(
+    tx: &mut Transaction<'static, Postgres>,
+    request: &CurrentStatusAllocation<'_>,
+) -> Result<(String, String), ClientStatusAllocationError> {
+    let row: Option<(String, String)> = sqlx::query_as(
+        "SELECT binding.issuer, binding.uid FROM identity_bindings binding \
+         JOIN identity_principals principal ON principal.community_id=binding.community_id \
+           AND principal.issuer=binding.issuer AND principal.uid=binding.uid \
+         JOIN relay_members member ON member.community_id=binding.community_id \
+           AND member.pubkey=encode(binding.pubkey, 'hex') \
+         WHERE binding.community_id=$1 AND binding.binding_id=$2 AND binding.pubkey=$3 \
+           AND binding.binding_version=$4 AND binding.binding_state='active' \
+           AND principal.disabled_at IS NULL \
+           AND NOT EXISTS (SELECT 1 FROM identity_revoked_keys revoked \
+             WHERE revoked.community_id=binding.community_id AND revoked.pubkey=binding.pubkey) \
+         FOR SHARE OF binding, principal, member",
+    )
+    .bind(request.community_id.as_uuid())
+    .bind(request.binding_id)
+    .bind(request.event_author_pubkey.as_slice())
+    .bind(request.binding_version as i64)
+    .fetch_optional(&mut **tx)
+    .await?;
+    let Some(principal) = row else {
+        return Err(ClientStatusAllocationError::NotCurrent);
+    };
+    let fresh: bool =
+        sqlx::query_scalar("SELECT clock_timestamp() < to_timestamp($1::double precision)")
+            .bind(request.fresh_until as f64)
+            .fetch_one(&mut **tx)
+            .await?;
+    if !fresh {
+        return Err(ClientStatusAllocationError::NotCurrent);
+    }
+    Ok(principal)
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn validate_invalidation(
+    tx: &mut Transaction<'static, Postgres>,
+    community_id: CommunityId,
+    evaluation_generation: u64,
+    binding_id: Uuid,
+    binding_version: u64,
+    pubkey: &[u8; 32],
+    policy_version: &str,
+    issuer: &str,
+    subject: &str,
+) -> Result<(), ClientStatusAllocationError> {
+    let generation: Option<i64> = sqlx::query_scalar(
+        "SELECT generation FROM authorization_invalidation_domains \
+         WHERE community_id=$1 FOR SHARE",
+    )
+    .bind(community_id.as_uuid())
+    .fetch_optional(&mut **tx)
+    .await?;
+    if generation != Some(evaluation_generation as i64) {
+        return Err(ClientStatusAllocationError::NotCurrent);
+    }
+    let selectors = [
+        AuthorizationSelector::domain(),
+        AuthorizationSelector::principal(issuer, subject)
+            .map_err(|_| ClientStatusAllocationError::InvalidInput)?,
+        AuthorizationSelector::nostr_key(*pubkey),
+        AuthorizationSelector::binding(binding_id, binding_version)
+            .map_err(|_| ClientStatusAllocationError::InvalidInput)?,
+        AuthorizationSelector::policy_version(policy_version)
+            .map_err(|_| ClientStatusAllocationError::InvalidInput)?,
+    ];
+    for selector in selectors {
+        let row = sqlx::query(
+            "SELECT generation, sticky_deny, binding_version_floor \
+             FROM authorization_invalidation_floors WHERE community_id=$1 \
+               AND selector_kind=$2 AND selector_fingerprint=$3 FOR SHARE",
+        )
+        .bind(community_id.as_uuid())
+        .bind(selector.kind().as_str())
+        .bind(selector.fingerprint().as_slice())
+        .fetch_optional(&mut **tx)
+        .await?;
+        let Some(row) = row else { continue };
+        let floor_generation: i64 = row.try_get("generation")?;
+        let sticky: bool = row.try_get("sticky_deny")?;
+        let version_floor: Option<i64> = row.try_get("binding_version_floor")?;
+        if sticky
+            || floor_generation > evaluation_generation as i64
+            || version_floor.is_some_and(|floor| binding_version <= floor as u64)
+        {
+            return Err(ClientStatusAllocationError::NotCurrent);
+        }
+    }
+    Ok(())
+}
+
+async fn replay_revision(
+    tx: &mut Transaction<'static, Postgres>,
+    community_id: CommunityId,
+    operation_id: Uuid,
+    operation_kind: &str,
+    request_fingerprint: [u8; 32],
+) -> Result<Option<AllocatedStatusRevision>, ClientStatusAllocationError> {
+    let row = sqlx::query(
+        "SELECT operation_kind, request_fingerprint, result_payload \
+         FROM authorization_operation_receipts \
+         WHERE community_id=$1 AND operation_id=$2 FOR SHARE",
+    )
+    .bind(community_id.as_uuid())
+    .bind(operation_id)
+    .fetch_optional(&mut **tx)
+    .await?;
+    let Some(row) = row else { return Ok(None) };
+    let kind: String = row.try_get("operation_kind")?;
+    let fingerprint: Vec<u8> = row.try_get("request_fingerprint")?;
+    let payload: Vec<u8> = row.try_get("result_payload")?;
+    if kind != operation_kind || fingerprint.as_slice() != request_fingerprint {
+        return Err(ClientStatusAllocationError::ConflictingRetry);
+    }
+    Ok(Some(decode_receipt_payload(&payload)?))
+}
+
+fn decode_receipt_payload(
+    payload: &[u8],
+) -> Result<AllocatedStatusRevision, ClientStatusAllocationError> {
+    let (revision_bytes, floor_bytes) = match payload.len() {
+        // Compatibility with receipts written before allocation-time floors
+        // were retained. The allocation revision was also its floor.
+        8 => (&payload[..8], &payload[..8]),
+        16 => (&payload[..8], &payload[8..16]),
+        _ => return Err(ClientStatusAllocationError::ConflictingRetry),
+    };
+    let revision = u64::from_be_bytes(
+        revision_bytes
+            .try_into()
+            .map_err(|_| ClientStatusAllocationError::ConflictingRetry)?,
+    );
+    let floor = u64::from_be_bytes(
+        floor_bytes
+            .try_into()
+            .map_err(|_| ClientStatusAllocationError::ConflictingRetry)?,
+    );
+    if revision == 0 || floor == 0 || revision < floor {
+        return Err(ClientStatusAllocationError::ConflictingRetry);
+    }
+    Ok(AllocatedStatusRevision { revision, floor })
+}
+
+async fn next_revision(
+    tx: &mut Transaction<'static, Postgres>,
+    community_id: CommunityId,
+) -> Result<u64, ClientStatusAllocationError> {
+    let revision: i64 = sqlx::query_scalar(
+        "UPDATE authorization_authority_epochs SET \
+           authority_epoch=authority_epoch+1, status_revision=status_revision+1, \
+           updated_at=clock_timestamp() WHERE community_id=$1 RETURNING status_revision",
+    )
+    .bind(community_id.as_uuid())
+    .fetch_one(&mut **tx)
+    .await?;
+    u64::try_from(revision).map_err(|_| ClientStatusAllocationError::InvalidInput)
+}
+
+async fn insert_receipt(
+    tx: &mut Transaction<'static, Postgres>,
+    community_id: CommunityId,
+    operation_id: Uuid,
+    operation_kind: &str,
+    request_fingerprint: [u8; 32],
+    revision: u64,
+) -> Result<(), ClientStatusAllocationError> {
+    let mut result_payload = Vec::with_capacity(16);
+    result_payload.extend_from_slice(&revision.to_be_bytes());
+    result_payload.extend_from_slice(&revision.to_be_bytes());
+    sqlx::query(
+        "INSERT INTO authorization_operation_receipts \
+         (community_id, operation_id, operation_kind, request_fingerprint, \
+          result_version, result_payload, lease_expires_at) \
+         VALUES ($1,$2,$3,$4,1,$5,clock_timestamp()+interval '100 years')",
+    )
+    .bind(community_id.as_uuid())
+    .bind(operation_id)
+    .bind(operation_kind)
+    .bind(request_fingerprint.as_slice())
+    .bind(result_payload)
+    .execute(&mut **tx)
+    .await?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    #[ignore = "requires migrated Postgres"]
+    async fn current_replay_withdrawal_and_revocation_are_transaction_owned() {
+        let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
+            .or_else(|_| std::env::var("DATABASE_URL"))
+            .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_owned());
+        let pool = sqlx::PgPool::connect(&database_url)
+            .await
+            .expect("test database");
+        crate::migration::run_migrations(&pool)
+            .await
+            .expect("migrations");
+        let db = Db::from_pool(pool);
+        let community = CommunityId::from_uuid(Uuid::new_v4());
+        let binding_id = Uuid::new_v4();
+        let author = [0x41; 32];
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1,$2)")
+            .bind(community.as_uuid())
+            .bind(format!("status-{}.example", community.as_uuid()))
+            .execute(&db.pool)
+            .await
+            .expect("community");
+        sqlx::query("INSERT INTO identity_principals (community_id,issuer,uid) VALUES ($1,$2,$3)")
+            .bind(community.as_uuid())
+            .bind("https://idp.example")
+            .bind("subject")
+            .execute(&db.pool)
+            .await
+            .expect("principal");
+        sqlx::query(
+            "INSERT INTO identity_bindings \
+             (community_id,issuer,uid,pubkey,source,binding_id,binding_version, \
+              binding_state,binding_provenance) \
+             VALUES ($1,$2,$3,$4,'jwt_npub',$5,1,'active','attested_key')",
+        )
+        .bind(community.as_uuid())
+        .bind("https://idp.example")
+        .bind("subject")
+        .bind(author.as_slice())
+        .bind(binding_id)
+        .execute(&db.pool)
+        .await
+        .expect("binding");
+        sqlx::query("INSERT INTO relay_members (community_id,pubkey,role) VALUES ($1,$2,'member')")
+            .bind(community.as_uuid())
+            .bind(hex::encode(author))
+            .execute(&db.pool)
+            .await
+            .expect("member");
+        sqlx::query(
+            "INSERT INTO authorization_invalidation_domains (community_id) VALUES ($1) \
+             ON CONFLICT (community_id) DO NOTHING",
+        )
+        .bind(community.as_uuid())
+        .execute(&db.pool)
+        .await
+        .expect("invalidation domain");
+        let generation: i64 = sqlx::query_scalar(
+            "SELECT generation FROM authorization_invalidation_domains WHERE community_id=$1",
+        )
+        .bind(community.as_uuid())
+        .fetch_one(&db.pool)
+        .await
+        .expect("generation");
+        let fresh_until = chrono::Utc::now().timestamp() as u64 + 300;
+        let operation_id = Uuid::new_v4();
+        let allocate = |operation_id, fingerprint| CurrentStatusAllocation {
+            community_id: community,
+            event_author_pubkey: &author,
+            binding_id,
+            binding_version: 1,
+            policy_version: "policy-v1",
+            evaluation_generation: generation as u64,
+            fresh_until,
+            operation_id,
+            request_fingerprint: fingerprint,
+        };
+        let first = db
+            .allocate_current_status_revision(allocate(operation_id, [1; 32]))
+            .await
+            .expect("first current");
+        let replay = db
+            .allocate_current_status_revision(allocate(operation_id, [1; 32]))
+            .await
+            .expect("exact replay");
+        assert_eq!(first, replay);
+        let second = db
+            .allocate_current_status_revision(allocate(Uuid::new_v4(), [2; 32]))
+            .await
+            .expect("new issuance");
+        assert!(second.revision > first.revision);
+        let withdrawal_operation = Uuid::new_v4();
+        let withdrawn = db
+            .allocate_withdrawn_status_revision(WithdrawalStatusAllocation {
+                community_id: community,
+                event_author_pubkey: &author,
+                supersedes_revision: second.revision,
+                issuance_fingerprint: [2; 32],
+                operation_id: withdrawal_operation,
+                request_fingerprint: [3; 32],
+            })
+            .await
+            .expect("withdrawal");
+        assert!(withdrawn.revision > second.revision);
+        let fanout = db
+            .allocate_withdrawn_status_revision(WithdrawalStatusAllocation {
+                community_id: community,
+                event_author_pubkey: &author,
+                supersedes_revision: first.revision,
+                issuance_fingerprint: [1; 32],
+                operation_id: Uuid::new_v4(),
+                request_fingerprint: [4; 32],
+            })
+            .await
+            .expect("older displayed current receives the same withdrawal");
+        assert_eq!(fanout, withdrawn);
+
+        sqlx::query(
+            "UPDATE authorization_authority_epochs \
+             SET authority_epoch=authority_epoch+1, status_revision=status_revision+1 \
+             WHERE community_id=$1",
+        )
+        .bind(community.as_uuid())
+        .execute(&db.pool)
+        .await
+        .expect("advance unrelated durable status floor");
+        let delayed_replay = db
+            .allocate_withdrawn_status_revision(WithdrawalStatusAllocation {
+                community_id: community,
+                event_author_pubkey: &author,
+                supersedes_revision: second.revision,
+                issuance_fingerprint: [2; 32],
+                operation_id: withdrawal_operation,
+                request_fingerprint: [3; 32],
+            })
+            .await
+            .expect("exact delayed fan-out replay retains allocation-time floor");
+        assert_eq!(delayed_replay, withdrawn);
+
+        let reissued = db
+            .allocate_current_status_revision(allocate(Uuid::new_v4(), [6; 32]))
+            .await
+            .expect("a fresh current status can replace a withdrawn projection");
+        assert!(reissued.revision > withdrawn.revision);
+        let row: (String, Option<i64>) = sqlx::query_as(
+            "SELECT disposition, supersedes_revision FROM client_status_revisions \
+             WHERE community_id=$1 AND event_author_pubkey=$2",
+        )
+        .bind(community.as_uuid())
+        .bind(author.as_slice())
+        .fetch_one(&db.pool)
+        .await
+        .expect("reissued projection");
+        assert_eq!(row, ("current".to_owned(), None));
+
+        assert!(matches!(
+            db.allocate_withdrawn_status_revision(WithdrawalStatusAllocation {
+                community_id: community,
+                event_author_pubkey: &author,
+                supersedes_revision: second.revision,
+                issuance_fingerprint: [9; 32],
+                operation_id: Uuid::new_v4(),
+                request_fingerprint: [5; 32],
+            })
+            .await,
+            Err(ClientStatusAllocationError::NotCurrent)
+        ));
+    }
+}

crates/buzz-db/src/push.rs

@@ -821,10 +821,21 @@ pub async fn claim_due_match_batch(
     limit: i64,
     lease_until: DateTime<Utc>,
 ) -> Result<Option<ClaimedMatchBatch>> {
-    claim_due_match_batch_with_loader(
+    claim_due_match_batch_excluding(pool, limit, lease_until, &[]).await
+}
+
+/// Claim a matcher batch without touching exact protected Enforce domains.
+pub async fn claim_due_match_batch_excluding(
+    pool: &PgPool,
+    limit: i64,
+    lease_until: DateTime<Utc>,
+    excluded_communities: &[Uuid],
+) -> Result<Option<ClaimedMatchBatch>> {
+    claim_due_match_batch_with_loader_excluding(
         pool,
         limit,
         lease_until,
+        excluded_communities,
         |pool, community, ids| async move {
             let refs: Vec<&[u8]> = ids.iter().map(Vec::as_slice).collect();
             crate::event::get_events_by_ids(&pool, community, &refs).await
@@ -833,10 +844,11 @@ pub async fn claim_due_match_batch(
     .await
 }
 
-async fn claim_due_match_batch_with_loader<F, Fut>(
+async fn claim_due_match_batch_with_loader_excluding<F, Fut>(
     pool: &PgPool,
     limit: i64,
     lease_until: DateTime<Utc>,
+    excluded_communities: &[Uuid],
     load: F,
 ) -> Result<Option<ClaimedMatchBatch>>
 where
@@ -850,6 +862,7 @@ where
             SELECT community_id
             FROM push_match_queue
             WHERE attempts < $3
+              AND NOT (community_id = ANY($5::uuid[]))
               AND next_attempt_at <= now()
               AND (state = 'pending' OR (state = 'matching' AND lease_until < now()))
             ORDER BY next_attempt_at, created_at
@@ -860,6 +873,7 @@ where
             FROM push_match_queue q
             JOIN target t ON q.community_id = t.community_id
             WHERE q.attempts < $3
+              AND NOT (q.community_id = ANY($5::uuid[]))
               AND q.next_attempt_at <= now()
               AND (q.state = 'pending' OR (q.state = 'matching' AND q.lease_until < now()))
             ORDER BY q.next_attempt_at, q.created_at
@@ -877,6 +891,7 @@ where
     .bind(lease_until)
     .bind(MAX_MATCH_ATTEMPTS)
     .bind(limit)
+    .bind(excluded_communities)
     .fetch_all(pool)
     .await?;
     if rows.is_empty() {
@@ -931,11 +946,21 @@ where
 /// served by the due partial index, so putting it in every claim made claims
 /// slower exactly when a backlog needed them fastest.
 pub async fn reap_exhausted_matches(pool: &PgPool) -> Result<u64> {
+    reap_exhausted_matches_excluding(pool, &[]).await
+}
+
+/// Reap exhausted matcher jobs outside exact protected Enforce domains.
+pub async fn reap_exhausted_matches_excluding(
+    pool: &PgPool,
+    excluded_communities: &[Uuid],
+) -> Result<u64> {
     Ok(sqlx::query(
         "DELETE FROM push_match_queue WHERE attempts >= $1 \
+         AND NOT (community_id = ANY($2::uuid[])) \
          AND (state='pending' OR (state='matching' AND lease_until < now()))",
     )
     .bind(MAX_MATCH_ATTEMPTS)
+    .bind(excluded_communities)
     .execute(pool)
     .await?
     .rows_affected())
@@ -1891,6 +1916,18 @@ mod tests {
         .await
         .expect("read matcher queue");
         assert_eq!(queued, vec![9]);
+        assert!(
+            claim_due_match_batch_excluding(
+                &pool,
+                16,
+                Utc::now() + chrono::Duration::minutes(1),
+                &[*community.as_uuid()],
+            )
+            .await
+            .expect("excluded protected matcher claim")
+            .is_none(),
+            "an excluded domain must remain unclaimed"
+        );
 
         sqlx::query("UPDATE events SET deleted_at=now() WHERE community_id=$1 AND id=$2")
             .bind(community.as_uuid())
@@ -1927,10 +1964,11 @@ mod tests {
             .await
             .expect("insert event");
 
-        let error = claim_due_match_batch_with_loader(
+        let error = claim_due_match_batch_with_loader_excluding(
             &pool,
             16,
             Utc::now() - chrono::Duration::seconds(1),
+            &[],
             |_pool, _community, _event_ids| async {
                 Err(crate::DbError::InvalidData("injected load failure".into()))
             },

desktop/src-tauri/src/commands/profile.rs

@@ -89,33 +89,58 @@ fn verified_identities(
             let parts = tag.as_slice();
             (parts.len() == 2).then(|| parts[1].as_str())
         };
+        let canonical_tag_set = |active: bool| {
+            let allowed: &[&str] = if active {
+                &["d", "p", "verified", "active", "expiration", "display_name"]
+            } else {
+                &["d", "p", "verified", "active", "expiration"]
+            };
+            event.tags.len() == allowed.len()
+                && event.tags.iter().all(|tag| {
+                    let parts = tag.as_slice();
+                    parts.len() == 2
+                        && allowed.contains(&parts[0].as_str())
+                        && event
+                            .tags
+                            .iter()
+                            .filter(|candidate| {
+                                candidate.as_slice().first() == parts.first()
+                            })
+                            .count()
+                            == 1
+                })
+        };
         // Select the signed replaceable-event head before validating its
         // payload. Otherwise a newer malformed assertion could be skipped and
         // silently resurrect the older active label returned alongside it.
-        let identity = match (tag_value("d"), tag_value("verified"), tag_value("p")) {
-            (Some(assertion_d), Some("relay"), Some(asserted_subject))
-                if assertion_d == subject && asserted_subject == subject =>
-            {
-                match tag_value("active") {
-                    Some("false") => None,
-                    Some("true") => match (
-                        tag_value("expiration")
-                            .and_then(|value| value.parse::<u64>().ok())
-                            .filter(|expiration| *expiration > now),
-                        tag_value("display_name")
-                            .map(str::trim)
-                            .filter(|value| !value.is_empty()),
-                    ) {
-                        (Some(expires_at), Some(display_name)) => Some(VerifiedIdentity {
-                            display_name: display_name.to_string(),
-                            expires_at,
-                        }),
+        let identity = if event.content.is_empty() {
+            match (tag_value("d"), tag_value("verified"), tag_value("p")) {
+                (Some(assertion_d), Some("relay"), Some(asserted_subject))
+                    if assertion_d == subject && asserted_subject == subject =>
+                {
+                    match tag_value("active") {
+                        Some("false") if canonical_tag_set(false) => None,
+                        Some("true") if canonical_tag_set(true) => match (
+                            tag_value("expiration")
+                                .and_then(|value| value.parse::<u64>().ok())
+                                .filter(|expiration| *expiration > now),
+                            tag_value("display_name")
+                                .map(str::trim)
+                                .filter(|value| !value.is_empty()),
+                        ) {
+                            (Some(expires_at), Some(display_name)) => Some(VerifiedIdentity {
+                                display_name: display_name.to_string(),
+                                expires_at,
+                            }),
+                            _ => None,
+                        },
                         _ => None,
-                    },
-                    _ => None,
+                    }
                 }
+                _ => None,
             }
-            _ => None,
+        } else {
+            None
         };
         let created_at = event.created_at.as_secs();
         let event_id = event.id.to_hex();
@@ -650,6 +675,86 @@ mod tests {
         );
     }
 
+    #[test]
+    fn newer_nonempty_projection_removes_verified_identity() {
+        let relay = nostr::Keys::generate();
+        let subject = nostr::Keys::generate().public_key().to_hex();
+        let created_at = nostr::Timestamp::now().as_secs();
+        let expires_at = created_at + 60;
+        let canonical_tags = || {
+            [
+                nostr::Tag::parse(["d", subject.as_str()]).unwrap(),
+                nostr::Tag::parse(["p", subject.as_str()]).unwrap(),
+                nostr::Tag::parse(["verified", "relay"]).unwrap(),
+                nostr::Tag::parse(["active", "true"]).unwrap(),
+                nostr::Tag::parse(["expiration", &expires_at.to_string()]).unwrap(),
+                nostr::Tag::parse(["display_name", "Example User"]).unwrap(),
+            ]
+        };
+        let active =
+            nostr::EventBuilder::new(nostr::Kind::Custom(KIND_USER_TRUSTED_ASSERTION as u16), "")
+                .tags(canonical_tags())
+                .custom_created_at(nostr::Timestamp::from(created_at))
+                .sign_with_keys(&relay)
+                .unwrap();
+        let nonempty = nostr::EventBuilder::new(
+            nostr::Kind::Custom(KIND_USER_TRUSTED_ASSERTION as u16),
+            "private content must never be projected",
+        )
+        .tags(canonical_tags())
+        .custom_created_at(nostr::Timestamp::from(created_at + 1))
+        .sign_with_keys(&relay)
+        .unwrap();
+
+        assert!(
+            verified_identities(&[active, nonempty], Some(&relay.public_key().to_hex())).is_empty(),
+            "a malformed newer head must withdraw rather than reveal or resurrect a label"
+        );
+    }
+
+    #[test]
+    fn newer_projection_with_unknown_or_duplicate_tags_removes_verified_identity() {
+        let relay = nostr::Keys::generate();
+        let subject = nostr::Keys::generate().public_key().to_hex();
+        let created_at = nostr::Timestamp::now().as_secs();
+        let expires_at = created_at + 60;
+        let canonical = nostr::EventBuilder::new(
+            nostr::Kind::Custom(KIND_USER_TRUSTED_ASSERTION as u16),
+            "",
+        )
+        .tags([
+            nostr::Tag::parse(["d", subject.as_str()]).unwrap(),
+            nostr::Tag::parse(["p", subject.as_str()]).unwrap(),
+            nostr::Tag::parse(["verified", "relay"]).unwrap(),
+            nostr::Tag::parse(["active", "true"]).unwrap(),
+            nostr::Tag::parse(["expiration", &expires_at.to_string()]).unwrap(),
+            nostr::Tag::parse(["display_name", "Example User"]).unwrap(),
+        ])
+        .custom_created_at(nostr::Timestamp::from(created_at))
+        .sign_with_keys(&relay)
+        .unwrap();
+
+        for extra in [
+            nostr::Tag::parse(["issuer", "private.invalid"]).unwrap(),
+            nostr::Tag::parse(["display_name", "Replacement"]).unwrap(),
+        ] {
+            let mut tags = canonical.tags.clone().to_vec();
+            tags.push(extra);
+            let malformed = nostr::EventBuilder::new(
+                nostr::Kind::Custom(KIND_USER_TRUSTED_ASSERTION as u16),
+                "",
+            )
+            .tags(tags)
+            .custom_created_at(nostr::Timestamp::from(created_at + 1))
+            .sign_with_keys(&relay)
+            .unwrap();
+            assert!(
+                verified_identities(&[canonical.clone(), malformed], Some(&relay.public_key().to_hex()))
+                    .is_empty()
+            );
+        }
+    }
+
     #[test]
     fn newer_malformed_assertion_does_not_resurrect_older_identity() {
         let relay = nostr::Keys::generate();

migrations/0038_client_status_fanout_withdrawals.sql

@@ -0,0 +1,21 @@
+-- Retain the exact current revision superseded by a withdrawal so every
+-- authenticated connection that displayed that author can receive a
+-- strictly newer opaque withdrawal. This remains server-side reconciliation
+-- state and is never serialized into the client projection.
+
+ALTER TABLE client_status_revisions
+    ADD COLUMN supersedes_revision BIGINT;
+
+UPDATE client_status_revisions
+SET supersedes_revision = revision - 1
+WHERE disposition = 'withdrawn';
+
+ALTER TABLE client_status_revisions
+    ADD CONSTRAINT client_status_revisions_withdrawal CHECK (
+        (disposition = 'current' AND supersedes_revision IS NULL)
+        OR
+        (disposition = 'withdrawn'
+            AND supersedes_revision IS NOT NULL
+            AND supersedes_revision > 0
+            AND revision > supersedes_revision)
+    );

migrations/0044_identity_public_projection_retirement.sql

@@ -0,0 +1,80 @@
+-- Durable, provider-neutral reconciliation for the optional public identity
+-- projection. O3 lifecycle rows remain the authority; these tables contain
+-- only public event coordinates and opaque binding generations.
+
+CREATE TABLE identity_public_projection_heads (
+    community_id          UUID NOT NULL REFERENCES communities(id),
+    relay_pubkey          BYTEA NOT NULL,
+    subject_pubkey        BYTEA NOT NULL,
+    event_id              BYTEA NOT NULL,
+    event_created_at      TIMESTAMPTZ NOT NULL,
+    disposition           TEXT NOT NULL,
+    source_binding_id     UUID,
+    source_binding_version BIGINT,
+    updated_at            TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (community_id, relay_pubkey, subject_pubkey),
+    FOREIGN KEY (community_id, source_binding_id)
+        REFERENCES identity_bindings (community_id, binding_id),
+    CHECK (length(relay_pubkey) = 32),
+    CHECK (length(subject_pubkey) = 32),
+    CHECK (length(event_id) = 32),
+    CHECK (disposition IN ('active', 'inactive')),
+    CHECK (source_binding_version IS NULL OR source_binding_version > 0),
+    CHECK (
+        (source_binding_id IS NULL AND source_binding_version IS NULL)
+        OR
+        (source_binding_id IS NOT NULL AND source_binding_version IS NOT NULL)
+    )
+);
+
+CREATE TABLE identity_public_projection_retirements (
+    community_id          UUID NOT NULL REFERENCES communities(id),
+    operation_id          UUID NOT NULL,
+    relay_pubkey          BYTEA NOT NULL,
+    old_pubkey            BYTEA NOT NULL,
+    source_binding_id     UUID,
+    source_binding_version BIGINT,
+    operation_kind        TEXT NOT NULL,
+    phase                 TEXT NOT NULL DEFAULT 'projection',
+    outcome               TEXT,
+    event_id              BYTEA,
+    attempts              BIGINT NOT NULL DEFAULT 0,
+    next_attempt_at       TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    claim_token           UUID,
+    lease_until           TIMESTAMPTZ,
+    completed_at          TIMESTAMPTZ,
+    created_at            TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at            TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (community_id, operation_id, relay_pubkey),
+    FOREIGN KEY (community_id, operation_id)
+        REFERENCES identity_lifecycle_operations (community_id, operation_id),
+    FOREIGN KEY (community_id, source_binding_id)
+        REFERENCES identity_bindings (community_id, binding_id),
+    CHECK (length(relay_pubkey) = 32),
+    CHECK (length(old_pubkey) = 32),
+    CHECK (source_binding_version IS NULL OR source_binding_version > 0),
+    CHECK (
+        (source_binding_id IS NULL AND source_binding_version IS NULL)
+        OR
+        (source_binding_id IS NOT NULL AND source_binding_version IS NOT NULL)
+    ),
+    CHECK (operation_kind IN ('revoke_key', 'rotate')),
+    CHECK (phase IN ('projection', 'delivery', 'completed', 'superseded')),
+    CHECK (outcome IS NULL OR outcome IN (
+        'no_projection', 'already_inactive', 'replaced_inactive',
+        'newer_binding', 'newer_projection'
+    )),
+    CHECK (event_id IS NULL OR length(event_id) = 32),
+    CHECK (attempts >= 0),
+    CHECK ((claim_token IS NULL) = (lease_until IS NULL)),
+    CHECK (
+        (phase IN ('completed', 'superseded') AND completed_at IS NOT NULL)
+        OR
+        (phase IN ('projection', 'delivery') AND completed_at IS NULL)
+    )
+);
+
+CREATE INDEX idx_identity_public_projection_retirements_ready
+    ON identity_public_projection_retirements
+        (phase, next_attempt_at, community_id, operation_id)
+    WHERE phase IN ('projection', 'delivery');