From d780ae6cf9ab7d4ce19d14d28aa647d889d5f6d8 Mon Sep 17 00:00:00 2001 From: Jreevo Date: Sat, 25 Jul 2026 22:21:36 -0700 Subject: [PATCH 1/3] feat(channels): add audited owner recovery Co-authored-by: Jreevo Signed-off-by: Jreevo --- crates/buzz-cli/src/commands/channels.rs | 21 + crates/buzz-cli/src/lib.rs | 23 +- crates/buzz-core/src/kind.rs | 5 + crates/buzz-db/src/archived_identities.rs | 35 +- crates/buzz-db/src/channel.rs | 17 + crates/buzz-db/src/channel_owner_recovery.rs | 1800 +++++++++++++++++ crates/buzz-db/src/lib.rs | 56 + crates/buzz-db/src/migration.rs | 16 +- .../src/handlers/channel_owner_recovery.rs | 553 +++++ crates/buzz-relay/src/handlers/event.rs | 2 +- crates/buzz-relay/src/handlers/ingest.rs | 50 +- crates/buzz-relay/src/handlers/mod.rs | 2 + crates/buzz-relay/src/main.rs | 41 + crates/buzz-sdk/src/builders.rs | 90 +- desktop/playwright.config.ts | 1 + desktop/src-tauri/src/commands/channels.rs | 14 + desktop/src-tauri/src/lib.rs | 1 + desktop/src/features/channels/hooks.ts | 27 + .../channels/ui/ChannelManagementSheet.tsx | 81 +- .../ui/ChannelOwnerRecoveryDialog.tsx | 223 ++ .../features/messages/ui/SystemMessageRow.tsx | 15 + desktop/src/shared/api/tauriChannels.ts | 12 + desktop/src/testing/e2eBridge.ts | 135 +- .../tests/e2e/channel-owner-recovery.spec.ts | 145 ++ desktop/tests/helpers/bridge.ts | 6 + migrations/0025_channel_owner_recovery.sql | 57 + 26 files changed, 3383 insertions(+), 45 deletions(-) create mode 100644 crates/buzz-db/src/channel_owner_recovery.rs create mode 100644 crates/buzz-relay/src/handlers/channel_owner_recovery.rs create mode 100644 desktop/src/features/channels/ui/ChannelOwnerRecoveryDialog.tsx create mode 100644 desktop/tests/e2e/channel-owner-recovery.spec.ts create mode 100644 migrations/0025_channel_owner_recovery.sql diff --git a/crates/buzz-cli/src/commands/channels.rs b/crates/buzz-cli/src/commands/channels.rs index 42844bf1e02..c37b4fedd17 100644 --- a/crates/buzz-cli/src/commands/channels.rs +++ b/crates/buzz-cli/src/commands/channels.rs @@ -1001,6 +1001,22 @@ pub async fn cmd_remove_channel_member( Ok(()) } +pub async fn cmd_recover_channel_owner( + client: &BuzzClient, + channel_id: &str, + target_pubkey: &str, + reason: &str, +) -> Result<(), CliError> { + let channel = parse_uuid(channel_id)?; + validate_hex64(target_pubkey)?; + let builder = buzz_sdk::build_channel_owner_recovery(channel, target_pubkey, reason) + .map_err(|error| CliError::Usage(error.to_string()))?; + let event = client.sign_event(builder)?; + let response = client.submit_event(event).await?; + println!("{}", normalize_write_response(&response)); + Ok(()) +} + /// Set the channel addition policy — sign and submit a kind:10100 (agent profile) event. pub async fn cmd_set_add_policy(client: &BuzzClient, policy: &str) -> Result<(), CliError> { match policy { @@ -1161,6 +1177,11 @@ pub async fn dispatch( ChannelsCmd::RemoveMember { channel, pubkey } => { cmd_remove_channel_member(client, &channel, &pubkey).await } + ChannelsCmd::RecoverOwner { + channel, + pubkey, + reason, + } => cmd_recover_channel_owner(client, &channel, &pubkey, &reason).await, ChannelsCmd::SetAddPolicy { policy } => cmd_set_add_policy(client, &policy).await, } } diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 6ab81a082df..70c2478f1ea 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -666,6 +666,26 @@ pub enum ChannelsCmd { #[arg(long)] pubkey: String, }, + /// Recover an orphaned channel owner using prior durable self-consent. + #[command( + name = "recover-owner", + after_help = "The relay denies this command unless every current human owner \ +self-archived and named the target as replacement before becoming unavailable. \ +Lost or deleted keys without that durable evidence are out of scope.\n\n\ +Example:\n buzz channels recover-owner --channel --pubkey \ +--reason \"All owners recorded replacement consent\"" + )] + RecoverOwner { + /// Channel UUID. + #[arg(long)] + channel: String, + /// Existing human member to promote (64-char hex). + #[arg(long)] + pubkey: String, + /// Human-readable immutable audit reason (1-500 bytes). + #[arg(long)] + reason: String, + }, /// Set your channel addition policy #[command(name = "set-add-policy")] SetAddPolicy { @@ -1904,6 +1924,7 @@ mod tests { "list", "members", "purpose", + "recover-owner", "remove-member", "search", "set-add-policy", @@ -1997,7 +2018,7 @@ mod tests { let expected: Vec<(&str, usize)> = vec![ ("agents", 5), ("canvas", 2), - ("channels", 16), + ("channels", 17), ("dms", 4), ("emoji", 5), ("feed", 1), diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index afec52305a0..0badc44486f 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -348,6 +348,10 @@ pub const KIND_NIP43_LEAVE_REQUEST: u32 = 28936; pub const KIND_IA_ARCHIVE_REQUEST: u32 = 9035; /// NIP-IA: Request that the relay unarchive a target identity. pub const KIND_IA_UNARCHIVE_REQUEST: u32 = 9036; +/// Buzz: Request audited recovery of an orphaned channel owner. +/// +/// This is a dedicated command, separate from generic channel role changes. +pub const KIND_CHANNEL_OWNER_RECOVERY: u32 = 9038; // NIP-IA identity archival announcement events (relay-signed) /// NIP-IA: Archived-identity delta (relay-signed). @@ -612,6 +616,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_NIP43_LEAVE_REQUEST, KIND_IA_ARCHIVE_REQUEST, KIND_IA_UNARCHIVE_REQUEST, + KIND_CHANNEL_OWNER_RECOVERY, KIND_IA_ARCHIVED, KIND_IA_UNARCHIVED, KIND_IA_ARCHIVED_LIST, diff --git a/crates/buzz-db/src/archived_identities.rs b/crates/buzz-db/src/archived_identities.rs index 941c0fc7358..c6a61c4e56b 100644 --- a/crates/buzz-db/src/archived_identities.rs +++ b/crates/buzz-db/src/archived_identities.rs @@ -7,10 +7,25 @@ use buzz_core::CommunityId; use chrono::{DateTime, Utc}; -use sqlx::{PgPool, Row as _}; +use sqlx::{PgPool, Postgres, Row as _, Transaction}; use crate::error::Result; +/// Acquire the transaction-scoped identity lock shared by archive, unarchive, +/// and owner recovery. Callers that lock multiple identities must sort them. +pub(crate) async fn lock_identity( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + pubkey: &str, +) -> Result<()> { + let key = format!("archived-identity:{community_id}:{pubkey}"); + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(key) + .execute(&mut **tx) + .await?; + Ok(()) +} + /// A single archived identity record. #[derive(Debug, Clone)] pub struct ArchivedIdentity { @@ -57,6 +72,8 @@ pub async fn archive( replaced_by: Option<&str>, request_event_id: &str, ) -> Result { + let mut tx = pool.begin().await?; + lock_identity(&mut tx, community_id, pubkey).await?; let result = sqlx::query( "INSERT INTO archived_identities \ (community_id, pubkey, consent_path, actor, reason, replaced_by, request_event_id) \ @@ -70,10 +87,11 @@ pub async fn archive( .bind(reason) .bind(replaced_by) .bind(request_event_id) - .execute(pool) + .execute(&mut *tx) .await?; - - Ok(result.rows_affected() > 0) + let changed = result.rows_affected() > 0; + tx.commit().await?; + Ok(changed) } /// Unarchives an identity from `community_id`. @@ -81,14 +99,17 @@ pub async fn archive( /// Returns `true` if a row was deleted, `false` if the identity was not archived /// in that community. pub async fn unarchive(pool: &PgPool, community_id: CommunityId, pubkey: &str) -> Result { + let mut tx = pool.begin().await?; + lock_identity(&mut tx, community_id, pubkey).await?; let result = sqlx::query("DELETE FROM archived_identities WHERE community_id = $1 AND pubkey = $2") .bind(community_id.as_uuid()) .bind(pubkey) - .execute(pool) + .execute(&mut *tx) .await?; - - Ok(result.rows_affected() > 0) + let changed = result.rows_affected() > 0; + tx.commit().await?; + Ok(changed) } /// Returns all identities archived in `community_id`, ordered by archive time ascending. diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 396244cc967..5cf8b5068b3 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -16,6 +16,21 @@ use buzz_core::CommunityId; // without pulling in sqlx/tokio. pub use buzz_core::channel::{ChannelType, ChannelVisibility, MemberRole}; +/// Acquire the transaction-scoped lock shared by every channel membership +/// mutation and the owner-recovery transaction. +pub(crate) async fn lock_channel_membership( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + channel_id: Uuid, +) -> Result<()> { + let key = format!("channel-membership:{community_id}:{channel_id}"); + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(key) + .execute(&mut **tx) + .await?; + Ok(()) +} + /// A channel row as returned from the database. #[derive(Debug, Clone)] pub struct ChannelRecord { @@ -359,6 +374,7 @@ pub async fn add_member( } let mut tx = pool.begin().await?; + lock_channel_membership(&mut tx, community_id, channel_id).await?; let channel = get_channel_tx(&mut tx, community_id, channel_id).await?; @@ -464,6 +480,7 @@ pub async fn remove_member( actor_pubkey: &[u8], ) -> Result<()> { let mut tx = pool.begin().await?; + lock_channel_membership(&mut tx, community_id, channel_id).await?; let is_self_remove = pubkey == actor_pubkey; if !is_self_remove { diff --git a/crates/buzz-db/src/channel_owner_recovery.rs b/crates/buzz-db/src/channel_owner_recovery.rs new file mode 100644 index 00000000000..0f4a55b079c --- /dev/null +++ b/crates/buzz-db/src/channel_owner_recovery.rs @@ -0,0 +1,1800 @@ +//! Atomic, audited orphaned-channel ownership recovery. +//! +//! Recovery is intentionally separate from generic role mutation. The only +//! supported predicate is durable prior self-consent from every current human +//! owner naming the nominated replacement. + +use buzz_core::CommunityId; +use chrono::{DateTime, Utc}; +use nostr::Event; +use serde::{Deserialize, Serialize}; +use sqlx::{PgPool, Row as _}; +use uuid::Uuid; + +use crate::error::{DbError, Result}; + +/// Stable identifier recorded in every audit event. +pub const RECOVERY_PREDICATE_ID: &str = "all_current_human_owners_self_archived_for_target_v1"; +/// Stable machine-readable reason recorded for this dedicated recovery path. +pub const RECOVERY_REASON_CODE: &str = "orphaned_owner_prior_self_consent"; + +/// An elevated membership captured before promotion. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct PriorElevatedRole { + /// Lowercase hex public key. + pub pubkey: String, + /// Existing channel role (`owner` or `admin`). + pub role: String, +} + +/// Durable payload used to construct the relay-signed channel audit event. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RecoveryAuditPayload { + /// Payload schema version. + pub schema_version: u8, + /// System-event type rendered by channel clients. + #[serde(rename = "type")] + pub event_type: String, + /// Community selected from the relay tenant context. + pub community_id: Uuid, + /// Channel whose ownership was recovered. + pub channel_id: Uuid, + /// Original signed recovery request ID. + pub request_event_id: String, + /// Community owner who requested recovery. + pub actor: String, + /// Existing member promoted to channel owner. + pub target: String, + /// Eligibility predicate applied under lock. + pub predicate_id: String, + /// Stable machine-readable recovery reason. + pub reason_code: String, + /// Human-supplied reason. + pub reason: String, + /// Elevated membership set immediately before promotion. + pub prior_elevated_roles: Vec, + /// Transaction timestamp. + pub created_at: DateTime, +} + +/// Result of an applied or replayed recovery request. +#[derive(Debug, Clone)] +pub struct RecoveryRecord { + /// Whether this request performed the promotion. + pub applied: bool, + /// Durable audit payload. + pub payload: RecoveryAuditPayload, + /// Whether the channel audit event has been delivered. + pub delivered: bool, +} + +/// Pending relay delivery loaded from the durable recovery outbox. +#[derive(Debug, Clone)] +pub struct PendingRecoveryDelivery { + /// Server-resolved community that owns the recovery. + pub community_id: CommunityId, + /// Community host used to construct a tenant context for fan-out. + pub host: String, + /// Original signed request ID. + pub request_event_id: Vec, + /// Immutable audit payload to publish. + pub payload: RecoveryAuditPayload, +} + +fn validate_reason(reason: &str) -> Result<&str> { + let trimmed = reason.trim(); + if trimmed.is_empty() { + return Err(DbError::InvalidData( + "recovery reason must not be empty".into(), + )); + } + if trimmed.len() > 500 { + return Err(DbError::InvalidData( + "recovery reason must not exceed 500 bytes".into(), + )); + } + if trimmed.chars().any(char::is_control) { + return Err(DbError::InvalidData( + "recovery reason must not contain control characters".into(), + )); + } + Ok(trimmed) +} + +fn payload_from_row(row: &sqlx::postgres::PgRow) -> Result { + let community_id: Uuid = row.try_get("community_id")?; + let request_event_id: Vec = row.try_get("request_event_id")?; + let actor_pubkey: Vec = row.try_get("actor_pubkey")?; + let target_pubkey: Vec = row.try_get("target_pubkey")?; + let prior_elevated_roles: serde_json::Value = row.try_get("prior_elevated_roles")?; + Ok(RecoveryAuditPayload { + schema_version: 1, + event_type: "channel_owner_recovered".into(), + community_id, + channel_id: row.try_get("channel_id")?, + request_event_id: hex::encode(request_event_id), + actor: hex::encode(actor_pubkey), + target: hex::encode(target_pubkey), + predicate_id: row.try_get("predicate_id")?, + reason_code: row.try_get("reason_code")?, + reason: row.try_get("reason")?, + prior_elevated_roles: serde_json::from_value(prior_elevated_roles)?, + created_at: row.try_get("created_at")?, + }) +} + +/// Apply or idempotently replay a protected channel-owner recovery request. +pub async fn recover_channel_owner( + pool: &PgPool, + community_id: CommunityId, + channel_id: Uuid, + target_pubkey: &[u8], + reason: &str, + request: &Event, +) -> Result { + if target_pubkey.len() != 32 { + return Err(DbError::InvalidData( + "target pubkey must be 32 bytes".into(), + )); + } + let reason = validate_reason(reason)?; + let actor_pubkey = request.pubkey.to_bytes(); + let actor_hex = request.pubkey.to_hex(); + let target_hex = hex::encode(target_pubkey); + let request_id = request.id.as_bytes(); + let request_id_hex = request.id.to_hex(); + + let mut tx = pool.begin().await?; + crate::channel::lock_channel_membership(&mut tx, community_id, channel_id).await?; + + if let Some(row) = sqlx::query( + "SELECT a.community_id, a.request_event_id, a.channel_id, \ + a.actor_pubkey, a.target_pubkey, a.predicate_id, \ + a.reason_code, a.reason, a.prior_elevated_roles, a.created_at, \ + o.delivered_at \ + FROM channel_owner_recovery_audit a \ + JOIN channel_owner_recovery_outbox o \ + ON o.community_id = a.community_id \ + AND o.request_event_id = a.request_event_id \ + WHERE a.community_id = $1 AND a.request_event_id = $2", + ) + .bind(community_id.as_uuid()) + .bind(request_id.as_slice()) + .fetch_optional(&mut *tx) + .await? + { + let payload = payload_from_row(&row)?; + if payload.community_id != *community_id.as_uuid() + || payload.channel_id != channel_id + || payload.request_event_id != request_id_hex + || payload.actor != actor_hex + || payload.target != target_hex + || payload.reason != reason + { + return Err(DbError::InvalidData( + "recovery replay does not match the committed audit".into(), + )); + } + let delivered_at: Option> = row.try_get("delivered_at")?; + tx.commit().await?; + return Ok(RecoveryRecord { + applied: false, + payload, + delivered: delivered_at.is_some(), + }); + } + + // Freshness applies only to a new state transition. An exact cryptographic + // replay of an already-committed request remains eligible to drain the + // durable outbox and refresh discovery after the original 120-second + // window has elapsed. + let request_ts = request.created_at.as_secs() as i64; + let now = Utc::now().timestamp(); + if (request_ts - now).abs() > 120 { + return Err(DbError::AccessDenied(format!( + "recovery request timestamp out of range (delta={}s, max ±120s)", + request_ts - now + ))); + } + + let channel = sqlx::query( + "SELECT channel_type::text AS channel_type, archived_at, deleted_at \ + FROM channels WHERE community_id = $1 AND id = $2 FOR UPDATE", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_optional(&mut *tx) + .await? + .ok_or(DbError::ChannelNotFound(channel_id))?; + let channel_type: String = channel.try_get("channel_type")?; + let archived_at: Option> = channel.try_get("archived_at")?; + let deleted_at: Option> = channel.try_get("deleted_at")?; + if archived_at.is_some() || deleted_at.is_some() { + return Err(DbError::AccessDenied( + "channel must be active for owner recovery".into(), + )); + } + if channel_type == "dm" { + return Err(DbError::AccessDenied( + "direct-message channels do not support owner recovery".into(), + )); + } + + let actor_role = sqlx::query_scalar::<_, String>( + "SELECT role FROM relay_members \ + WHERE community_id = $1 AND pubkey = $2 FOR UPDATE", + ) + .bind(community_id.as_uuid()) + .bind(&actor_hex) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| DbError::AccessDenied("actor is not a community member".into()))?; + if actor_role != "owner" { + return Err(DbError::AccessDenied( + "recovery requires an active, known human community owner".into(), + )); + } + + let target_role = sqlx::query_scalar::<_, String>( + "SELECT role::text FROM channel_members \ + WHERE community_id = $1 AND channel_id = $2 \ + AND pubkey = $3 AND removed_at IS NULL FOR UPDATE", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(target_pubkey) + .fetch_optional(&mut *tx) + .await? + .ok_or(DbError::MemberNotFound(channel_id))?; + if !matches!(target_role.as_str(), "member" | "guest") { + return Err(DbError::AccessDenied( + "target must be an active, known human, non-elevated channel member".into(), + )); + } + + let elevated_rows = sqlx::query( + "SELECT pubkey, role::text AS role FROM channel_members \ + WHERE community_id = $1 AND channel_id = $2 \ + AND removed_at IS NULL AND role IN ('owner', 'admin') \ + ORDER BY pubkey FOR UPDATE", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_all(&mut *tx) + .await?; + if elevated_rows.is_empty() { + return Err(DbError::AccessDenied( + "channel has no current owner record to authorize recovery".into(), + )); + } + + let mut identities = vec![actor_hex.clone(), target_hex.clone()]; + for row in &elevated_rows { + let pubkey: Vec = row.try_get("pubkey")?; + identities.push(hex::encode(pubkey)); + } + identities.sort(); + identities.dedup(); + for identity in &identities { + crate::archived_identities::lock_identity(&mut tx, community_id, identity).await?; + } + + let actor = sqlx::query( + "SELECT agent_owner_pubkey, deactivated_at FROM users \ + WHERE community_id = $1 AND pubkey = $2 FOR SHARE", + ) + .bind(community_id.as_uuid()) + .bind(actor_pubkey.as_slice()) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| { + DbError::AccessDenied("recovery requires an active, known human community owner".into()) + })?; + let actor_agent_owner: Option> = actor.try_get("agent_owner_pubkey")?; + let actor_deactivated: Option> = actor.try_get("deactivated_at")?; + let actor_archived = sqlx::query_scalar::<_, bool>( + "SELECT EXISTS(SELECT 1 FROM archived_identities \ + WHERE community_id = $1 AND pubkey = $2)", + ) + .bind(community_id.as_uuid()) + .bind(&actor_hex) + .fetch_one(&mut *tx) + .await?; + if actor_agent_owner.is_some() || actor_deactivated.is_some() || actor_archived { + return Err(DbError::AccessDenied( + "recovery requires an active, known human community owner".into(), + )); + } + + let target = sqlx::query( + "SELECT agent_owner_pubkey, deactivated_at FROM users \ + WHERE community_id = $1 AND pubkey = $2 FOR SHARE", + ) + .bind(community_id.as_uuid()) + .bind(target_pubkey) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| { + DbError::AccessDenied( + "target must be an active, known human, non-elevated channel member".into(), + ) + })?; + let target_agent_owner: Option> = target.try_get("agent_owner_pubkey")?; + let target_deactivated: Option> = target.try_get("deactivated_at")?; + if target_agent_owner.is_some() || target_deactivated.is_some() { + return Err(DbError::AccessDenied( + "target must be an active, known human, non-elevated channel member".into(), + )); + } + + let target_archived = sqlx::query_scalar::<_, bool>( + "SELECT EXISTS(SELECT 1 FROM archived_identities \ + WHERE community_id = $1 AND pubkey = $2)", + ) + .bind(community_id.as_uuid()) + .bind(&target_hex) + .fetch_one(&mut *tx) + .await?; + if target_archived { + return Err(DbError::AccessDenied("target identity is archived".into())); + } + + let mut prior_elevated_roles = Vec::with_capacity(elevated_rows.len()); + for row in elevated_rows { + let pubkey: Vec = row.try_get("pubkey")?; + let pubkey_hex = hex::encode(&pubkey); + let role: String = row.try_get("role")?; + let user = sqlx::query( + "SELECT agent_owner_pubkey, deactivated_at FROM users \ + WHERE community_id = $1 AND pubkey = $2 FOR SHARE", + ) + .bind(community_id.as_uuid()) + .bind(&pubkey) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| { + DbError::AccessDenied("owner recovery rejects unknown elevated identities".into()) + })?; + let agent_owner: Option> = user.try_get("agent_owner_pubkey")?; + let deactivated_at: Option> = user.try_get("deactivated_at")?; + if agent_owner.is_some() { + return Err(DbError::AccessDenied( + "owner recovery is disabled while an owner/admin agent exists".into(), + )); + } + if deactivated_at.is_some() { + return Err(DbError::AccessDenied( + "owner recovery rejects unknown or deactivated elevated identities".into(), + )); + } + if role == "admin" { + // Identity archival is a UI hint and does not revoke channel + // authority. Any active admin membership can still use the normal + // role path, regardless of archive state. + return Err(DbError::AccessDenied( + "use the ordinary role path while an active human channel admin exists".into(), + )); + } else { + let eligible = sqlx::query_scalar::<_, bool>( + "SELECT EXISTS(SELECT 1 FROM archived_identities \ + WHERE community_id = $1 AND pubkey = $2 \ + AND consent_path = 'self' AND actor = $2 AND replaced_by = $3)", + ) + .bind(community_id.as_uuid()) + .bind(&pubkey_hex) + .bind(&target_hex) + .fetch_one(&mut *tx) + .await?; + if !eligible { + return Err(DbError::AccessDenied(format!( + "owner {pubkey_hex} lacks prior durable self-consent naming the target" + ))); + } + } + prior_elevated_roles.push(PriorElevatedRole { + pubkey: pubkey_hex, + role, + }); + } + + let promoted = sqlx::query( + "UPDATE channel_members SET role = 'owner' \ + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 \ + AND removed_at IS NULL AND role IN ('member', 'guest')", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(target_pubkey) + .execute(&mut *tx) + .await? + .rows_affected(); + if promoted != 1 { + return Err(DbError::InvalidData( + "target membership changed before owner promotion".into(), + )); + } + + let created_at_secs = request.created_at.as_secs() as i64; + let request_created_at = DateTime::from_timestamp(created_at_secs, 0) + .ok_or(DbError::InvalidTimestamp(created_at_secs))?; + let tags = serde_json::to_value(&request.tags)?; + let inserted = sqlx::query( + "INSERT INTO events \ + (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id) \ + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,NOW(),$9) ON CONFLICT DO NOTHING", + ) + .bind(community_id.as_uuid()) + .bind(request_id.as_slice()) + .bind(actor_pubkey.as_slice()) + .bind(request_created_at) + .bind(i32::from(request.kind.as_u16())) + .bind(tags) + .bind(&request.content) + .bind(request.sig.serialize().as_slice()) + .bind(channel_id) + .execute(&mut *tx) + .await? + .rows_affected(); + if inserted != 1 { + return Err(DbError::InvalidData( + "request event already exists without a matching recovery audit".into(), + )); + } + + let created_at = Utc::now(); + let payload = RecoveryAuditPayload { + schema_version: 1, + event_type: "channel_owner_recovered".into(), + community_id: *community_id.as_uuid(), + channel_id, + request_event_id: request_id_hex, + actor: actor_hex, + target: target_hex, + predicate_id: RECOVERY_PREDICATE_ID.into(), + reason_code: RECOVERY_REASON_CODE.into(), + reason: reason.to_string(), + prior_elevated_roles, + created_at, + }; + let prior_roles_json = serde_json::to_value(&payload.prior_elevated_roles)?; + sqlx::query( + "INSERT INTO channel_owner_recovery_audit \ + (community_id,request_event_id,channel_id,actor_pubkey,target_pubkey,predicate_id,reason_code,reason,prior_elevated_roles,created_at) \ + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)", + ) + .bind(community_id.as_uuid()) + .bind(request_id.as_slice()) + .bind(channel_id) + .bind(actor_pubkey.as_slice()) + .bind(target_pubkey) + .bind(RECOVERY_PREDICATE_ID) + .bind(RECOVERY_REASON_CODE) + .bind(reason) + .bind(prior_roles_json) + .bind(created_at) + .execute(&mut *tx) + .await?; + sqlx::query( + "INSERT INTO channel_owner_recovery_outbox \ + (community_id,request_event_id,channel_id) VALUES ($1,$2,$3)", + ) + .bind(community_id.as_uuid()) + .bind(request_id.as_slice()) + .bind(channel_id) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(RecoveryRecord { + applied: true, + payload, + delivered: false, + }) +} + +/// Mark the channel audit event as durably delivered. +pub async fn mark_recovery_delivered( + pool: &PgPool, + community_id: CommunityId, + request_event_id: &[u8], +) -> Result<()> { + sqlx::query( + "UPDATE channel_owner_recovery_outbox SET delivered_at = NOW(), \ + attempts = attempts + 1, last_error = NULL \ + WHERE community_id = $1 AND request_event_id = $2", + ) + .bind(community_id.as_uuid()) + .bind(request_event_id) + .execute(pool) + .await?; + Ok(()) +} + +/// Record a retryable channel audit delivery failure. +pub async fn record_recovery_delivery_failure( + pool: &PgPool, + community_id: CommunityId, + request_event_id: &[u8], + error: &str, +) -> Result<()> { + sqlx::query( + "UPDATE channel_owner_recovery_outbox SET attempts = attempts + 1, last_error = $3 \ + WHERE community_id = $1 AND request_event_id = $2 AND delivered_at IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(request_event_id) + .bind(error) + .execute(pool) + .await?; + Ok(()) +} + +/// Load a bounded batch of undelivered channel recovery audits. +pub async fn pending_recovery_deliveries( + pool: &PgPool, + limit: i64, +) -> Result> { + let rows = sqlx::query( + "SELECT a.community_id, c.host, a.request_event_id, a.channel_id, \ + a.actor_pubkey, a.target_pubkey, a.predicate_id, \ + a.reason_code, a.reason, a.prior_elevated_roles, a.created_at \ + FROM channel_owner_recovery_outbox o \ + JOIN channel_owner_recovery_audit a \ + ON a.community_id = o.community_id \ + AND a.request_event_id = o.request_event_id \ + JOIN communities c ON c.id = o.community_id \ + WHERE o.delivered_at IS NULL \ + ORDER BY o.created_at, o.community_id, o.request_event_id \ + LIMIT $1", + ) + .bind(limit.clamp(1, 1_000)) + .fetch_all(pool) + .await?; + + rows.into_iter() + .map(|row| { + let community_id: Uuid = row.try_get("community_id")?; + Ok(PendingRecoveryDelivery { + community_id: CommunityId::from_uuid(community_id), + host: row.try_get("host")?, + request_event_id: row.try_get("request_event_id")?, + payload: payload_from_row(&row)?, + }) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use buzz_core::channel::{ChannelType, ChannelVisibility, MemberRole}; + use buzz_core::kind::KIND_CHANNEL_OWNER_RECOVERY; + use nostr::{EventBuilder, Keys, Kind, Tag}; + + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; + + struct Fixture { + community: CommunityId, + channel: Uuid, + actor: Keys, + owner: Keys, + target: Keys, + } + + struct ContinuityRecords { + root_event_id: Vec, + reply_event_id: Vec, + workflow_id: Uuid, + agent: Keys, + } + + async fn setup_pool() -> PgPool { + PgPool::connect(TEST_DB_URL) + .await + .expect("connect to test DB") + } + + async fn insert_human(pool: &PgPool, community: CommunityId, keys: &Keys) { + sqlx::query("INSERT INTO users (community_id,pubkey) VALUES ($1,$2)") + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes().as_slice()) + .execute(pool) + .await + .expect("insert human"); + } + + async fn setup_fixture(pool: &PgPool) -> Fixture { + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id,host) VALUES ($1,$2)") + .bind(id) + .bind(format!("owner-recovery-{}.example", id.simple())) + .execute(pool) + .await + .expect("insert community"); + let community = CommunityId::from_uuid(id); + let actor = Keys::generate(); + let owner = Keys::generate(); + let target = Keys::generate(); + for keys in [&actor, &owner, &target] { + insert_human(pool, community, keys).await; + } + sqlx::query("INSERT INTO relay_members (community_id,pubkey,role) VALUES ($1,$2,'owner')") + .bind(community.as_uuid()) + .bind(actor.public_key().to_hex()) + .execute(pool) + .await + .expect("insert community owner"); + + let channel = crate::channel::create_channel( + pool, + community, + "orphaned", + ChannelType::Stream, + ChannelVisibility::Open, + Some("continuity fixture"), + owner.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("create channel") + .id; + crate::channel::add_member( + pool, + community, + channel, + target.public_key().to_bytes().as_slice(), + MemberRole::Member, + Some(owner.public_key().to_bytes().as_slice()), + ) + .await + .expect("add target"); + + Fixture { + community, + channel, + actor, + owner, + target, + } + } + + fn request(fixture: &Fixture, reason: &str) -> Event { + request_as(&fixture.actor, fixture, reason) + } + + fn request_as(actor: &Keys, fixture: &Fixture, reason: &str) -> Event { + request_as_at(actor, fixture, reason, nostr::Timestamp::now()) + } + + fn request_as_at( + actor: &Keys, + fixture: &Fixture, + reason: &str, + created_at: nostr::Timestamp, + ) -> Event { + EventBuilder::new(Kind::Custom(KIND_CHANNEL_OWNER_RECOVERY as u16), "") + .tags([ + Tag::parse(["-"]).unwrap(), + Tag::parse(["h", &fixture.channel.to_string()]).unwrap(), + Tag::parse(["p", &fixture.target.public_key().to_hex()]).unwrap(), + Tag::parse(["reason", reason]).unwrap(), + ]) + .custom_created_at(created_at) + .sign_with_keys(actor) + .expect("sign request") + } + + async fn record_owner_consent(pool: &PgPool, fixture: &Fixture) { + crate::archived_identities::archive( + pool, + fixture.community, + &fixture.owner.public_key().to_hex(), + "self", + &fixture.owner.public_key().to_hex(), + Some("retired"), + Some(&fixture.target.public_key().to_hex()), + &"aa".repeat(32), + ) + .await + .expect("archive owner"); + } + + async fn seed_continuity_records(pool: &PgPool, fixture: &Fixture) -> ContinuityRecords { + crate::channel::set_canvas( + pool, + fixture.community, + fixture.channel, + Some("# Recovery continuity\n\nKeep this canvas."), + ) + .await + .expect("seed canvas"); + + let channel = fixture.channel.to_string(); + let root = EventBuilder::new(Kind::Custom(9), "continuity root") + .tags([Tag::parse(["h", &channel]).unwrap()]) + .sign_with_keys(&fixture.owner) + .expect("sign root"); + let reply = EventBuilder::new(Kind::Custom(9), "continuity reply") + .tags([ + Tag::parse(["h", &channel]).unwrap(), + Tag::parse(["e", &root.id.to_hex(), "", "root"]).unwrap(), + Tag::parse(["e", &root.id.to_hex(), "", "reply"]).unwrap(), + ]) + .custom_created_at(nostr::Timestamp::from(root.created_at.as_secs() + 1)) + .sign_with_keys(&fixture.target) + .expect("sign reply"); + for event in [&root, &reply] { + crate::event::insert_event(pool, fixture.community, event, Some(fixture.channel)) + .await + .expect("seed message"); + } + + let root_created_at = + DateTime::from_timestamp(root.created_at.as_secs() as i64, 0).expect("root timestamp"); + let reply_created_at = DateTime::from_timestamp(reply.created_at.as_secs() as i64, 0) + .expect("reply timestamp"); + sqlx::query( + "INSERT INTO thread_metadata \ + (community_id,event_created_at,event_id,channel_id,depth,reply_count,descendant_count,last_reply_at) \ + VALUES ($1,$2,$3,$4,0,1,1,$5)", + ) + .bind(fixture.community.as_uuid()) + .bind(root_created_at) + .bind(root.id.as_bytes().as_slice()) + .bind(fixture.channel) + .bind(reply_created_at) + .execute(pool) + .await + .expect("seed root metadata"); + sqlx::query( + "INSERT INTO thread_metadata \ + (community_id,event_created_at,event_id,channel_id,parent_event_id,parent_event_created_at,root_event_id,root_event_created_at,depth) \ + VALUES ($1,$2,$3,$4,$5,$6,$5,$6,1)", + ) + .bind(fixture.community.as_uuid()) + .bind(reply_created_at) + .bind(reply.id.as_bytes().as_slice()) + .bind(fixture.channel) + .bind(root.id.as_bytes().as_slice()) + .bind(root_created_at) + .execute(pool) + .await + .expect("seed reply metadata"); + + let workflow_id = Uuid::new_v4(); + sqlx::query( + "INSERT INTO workflows \ + (community_id,id,name,owner_pubkey,channel_id,definition,definition_hash) \ + VALUES ($1,$2,'continuity workflow',$3,$4,$5,$6)", + ) + .bind(fixture.community.as_uuid()) + .bind(workflow_id) + .bind(fixture.actor.public_key().to_bytes().as_slice()) + .bind(fixture.channel) + .bind(serde_json::json!({"name":"continuity workflow","steps":[]})) + .bind(vec![7_u8; 32]) + .execute(pool) + .await + .expect("seed workflow"); + + let agent = Keys::generate(); + insert_human(pool, fixture.community, &agent).await; + sqlx::query( + "UPDATE users SET agent_owner_pubkey=$3 \ + WHERE community_id=$1 AND pubkey=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(agent.public_key().to_bytes().as_slice()) + .bind(fixture.actor.public_key().to_bytes().as_slice()) + .execute(pool) + .await + .expect("classify continuity agent"); + crate::channel::add_member( + pool, + fixture.community, + fixture.channel, + agent.public_key().to_bytes().as_slice(), + MemberRole::Bot, + Some(fixture.owner.public_key().to_bytes().as_slice()), + ) + .await + .expect("seed agent membership"); + + ContinuityRecords { + root_event_id: root.id.as_bytes().to_vec(), + reply_event_id: reply.id.as_bytes().to_vec(), + workflow_id, + agent, + } + } + + async fn assert_continuity_preserved( + pool: &PgPool, + fixture: &Fixture, + records: &ContinuityRecords, + ) { + let canvas: Option = + sqlx::query_scalar("SELECT canvas FROM channels WHERE community_id=$1 AND id=$2") + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .fetch_one(pool) + .await + .expect("read canvas"); + assert_eq!( + canvas.as_deref(), + Some("# Recovery continuity\n\nKeep this canvas.") + ); + + let message_count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM events \ + WHERE community_id=$1 AND channel_id=$2 AND (id=$3 OR id=$4)", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .bind(&records.root_event_id) + .bind(&records.reply_event_id) + .fetch_one(pool) + .await + .expect("read messages"); + assert_eq!(message_count, 2); + + let thread_rows: i64 = sqlx::query_scalar( + "SELECT count(*) FROM thread_metadata \ + WHERE community_id=$1 AND channel_id=$2 \ + AND (event_id=$3 OR event_id=$4)", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .bind(&records.root_event_id) + .bind(&records.reply_event_id) + .fetch_one(pool) + .await + .expect("read thread metadata"); + assert_eq!(thread_rows, 2); + + let workflow_channel: Uuid = + sqlx::query_scalar("SELECT channel_id FROM workflows WHERE community_id=$1 AND id=$2") + .bind(fixture.community.as_uuid()) + .bind(records.workflow_id) + .fetch_one(pool) + .await + .expect("read workflow"); + assert_eq!(workflow_channel, fixture.channel); + + let agent_role: String = sqlx::query_scalar( + "SELECT role::text FROM channel_members \ + WHERE community_id=$1 AND channel_id=$2 AND pubkey=$3 AND removed_at IS NULL", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .bind(records.agent.public_key().to_bytes().as_slice()) + .fetch_one(pool) + .await + .expect("read agent membership"); + assert_eq!(agent_role, "bot"); + } + + async fn assert_no_recovery_side_effects(pool: &PgPool, fixture: &Fixture) { + let target_role: String = sqlx::query_scalar( + "SELECT role::text FROM channel_members \ + WHERE community_id=$1 AND channel_id=$2 AND pubkey=$3", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .bind(fixture.target.public_key().to_bytes().as_slice()) + .fetch_one(pool) + .await + .expect("target role"); + assert_eq!(target_role, "member"); + let audit_count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM channel_owner_recovery_audit \ + WHERE community_id=$1 AND channel_id=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .fetch_one(pool) + .await + .expect("audit count"); + assert_eq!(audit_count, 0); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn eligible_recovery_is_promotion_only_atomic_audited_and_idempotent() { + let pool = setup_pool().await; + let fixture = setup_fixture(&pool).await; + let continuity = seed_continuity_records(&pool, &fixture).await; + record_owner_consent(&pool, &fixture).await; + let request = request(&fixture, "approved continuity recovery"); + + let first = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "approved continuity recovery", + &request, + ) + .await + .expect("eligible recovery"); + assert!(first.applied); + assert!(!first.delivered); + assert_eq!(first.payload.predicate_id, RECOVERY_PREDICATE_ID); + assert_eq!(first.payload.reason_code, RECOVERY_REASON_CODE); + assert_eq!(first.payload.event_type, "channel_owner_recovered"); + assert_continuity_preserved(&pool, &fixture, &continuity).await; + let pending = pending_recovery_deliveries(&pool, 1_000) + .await + .expect("load pending recovery audit"); + let pending = pending + .iter() + .find(|item| item.request_event_id.as_slice() == request.id.as_bytes().as_slice()) + .expect("recovery audit remains available for worker delivery"); + assert_eq!(pending.community_id, fixture.community); + assert_eq!(pending.payload, first.payload); + + let roles: Vec = sqlx::query_scalar( + "SELECT role::text FROM channel_members \ + WHERE community_id=$1 AND channel_id=$2 AND removed_at IS NULL \ + ORDER BY role::text,pubkey", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .fetch_all(&pool) + .await + .expect("read roles"); + assert_eq!(roles, vec!["bot", "owner", "owner"]); + + let channel_name: String = sqlx::query_scalar( + "SELECT name FROM channels WHERE community_id=$1 AND id=$2 \ + AND archived_at IS NULL AND deleted_at IS NULL", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .fetch_one(&pool) + .await + .expect("channel preserved"); + assert_eq!(channel_name, "orphaned"); + + let mismatched_replay = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "different reason", + &request, + ) + .await; + assert!(matches!( + mismatched_replay, + Err(DbError::InvalidData(message)) + if message == "recovery replay does not match the committed audit" + )); + + let replay = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "approved continuity recovery", + &request, + ) + .await + .expect("idempotent replay"); + assert!(!replay.applied); + let audit_count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM channel_owner_recovery_audit \ + WHERE community_id=$1 AND request_event_id=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(request.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("count audit"); + assert_eq!(audit_count, 1); + + let immutable = sqlx::query( + "UPDATE channel_owner_recovery_audit SET reason='changed' \ + WHERE community_id=$1 AND request_event_id=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(request.id.as_bytes().as_slice()) + .execute(&pool) + .await; + assert!(immutable.is_err(), "audit row must be immutable"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn stale_new_request_is_denied_but_committed_replay_remains_retryable() { + let pool = setup_pool().await; + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + + let stale = request_as_at( + &fixture.actor, + &fixture, + "stale new request", + nostr::Timestamp::from(nostr::Timestamp::now().as_secs().saturating_sub(121)), + ); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "stale new request", + &stale, + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + + let retryable = request_as_at( + &fixture.actor, + &fixture, + "retryable committed request", + nostr::Timestamp::from(nostr::Timestamp::now().as_secs().saturating_sub(119)), + ); + recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "retryable committed request", + &retryable, + ) + .await + .expect("fresh original request"); + tokio::time::sleep(std::time::Duration::from_secs(3)).await; + let replay = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "retryable committed request", + &retryable, + ) + .await + .expect("stale committed replay"); + assert!(!replay.applied); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn every_current_owner_must_self_consent_to_the_same_target() { + let pool = setup_pool().await; + let fixture = setup_fixture(&pool).await; + let second_owner = Keys::generate(); + insert_human(&pool, fixture.community, &second_owner).await; + crate::channel::add_member( + &pool, + fixture.community, + fixture.channel, + second_owner.public_key().to_bytes().as_slice(), + MemberRole::Owner, + Some(fixture.owner.public_key().to_bytes().as_slice()), + ) + .await + .expect("add second owner"); + record_owner_consent(&pool, &fixture).await; + + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "second owner has not consented", + &request(&fixture, "second owner has not consented"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + + crate::archived_identities::archive( + &pool, + fixture.community, + &second_owner.public_key().to_hex(), + "self", + &second_owner.public_key().to_hex(), + Some("second owner retired"), + Some(&fixture.target.public_key().to_hex()), + &"12".repeat(32), + ) + .await + .expect("archive second owner"); + let accepted = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "all owners consented", + &request(&fixture, "all owners consented"), + ) + .await + .expect("all owners consented"); + assert!(accepted.applied); + assert_eq!(accepted.payload.prior_elevated_roles.len(), 2); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn missing_self_consent_and_archived_target_fail_without_side_effects() { + let pool = setup_pool().await; + let fixture = setup_fixture(&pool).await; + let denied_request = request(&fixture, "must be denied"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "must be denied", + &denied_request, + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + + crate::archived_identities::archive( + &pool, + fixture.community, + &fixture.target.public_key().to_hex(), + "self", + &fixture.target.public_key().to_hex(), + Some("target retired"), + None, + &"bb".repeat(32), + ) + .await + .expect("archive target"); + record_owner_consent(&pool, &fixture).await; + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "target archived", + &request(&fixture, "target archived"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + + assert_no_recovery_side_effects(&pool, &fixture).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn unknown_actor_and_unknown_target_fail_closed() { + let pool = setup_pool().await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + let unknown_actor = Keys::generate(); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "unknown actor", + &request_as(&unknown_actor, &fixture, "unknown actor"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + sqlx::query("DELETE FROM users WHERE community_id=$1 AND pubkey=$2") + .bind(fixture.community.as_uuid()) + .bind(fixture.target.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("remove target user classification"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "unknown target", + &request(&fixture, "unknown target"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn community_admin_member_and_agent_actor_are_denied() { + let pool = setup_pool().await; + + for role in ["admin", "member"] { + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + sqlx::query( + "UPDATE relay_members SET role=$3 \ + WHERE community_id=$1 AND pubkey=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.actor.public_key().to_hex()) + .bind(role) + .execute(&pool) + .await + .expect("change community role"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "unauthorized community role", + &request(&fixture, "unauthorized community role"), + ) + .await; + assert!( + matches!(denied, Err(DbError::AccessDenied(_))), + "community {role} must be denied" + ); + assert_no_recovery_side_effects(&pool, &fixture).await; + } + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + sqlx::query( + "UPDATE users SET agent_owner_pubkey=$3 \ + WHERE community_id=$1 AND pubkey=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.actor.public_key().to_bytes().as_slice()) + .bind(fixture.owner.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("classify actor as agent"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "agent actor", + &request(&fixture, "agent actor"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn deactivated_actor_target_and_owner_are_denied() { + let pool = setup_pool().await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + sqlx::query( + "UPDATE users SET deactivated_at=NOW() \ + WHERE community_id=$1 AND pubkey=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.actor.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("deactivate actor"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "deactivated actor", + &request(&fixture, "deactivated actor"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + sqlx::query( + "UPDATE users SET deactivated_at=NOW() \ + WHERE community_id=$1 AND pubkey=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.target.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("deactivate target"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "deactivated target", + &request(&fixture, "deactivated target"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + sqlx::query( + "UPDATE users SET deactivated_at=NOW() \ + WHERE community_id=$1 AND pubkey=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.owner.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("deactivate current owner"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "deactivated current owner", + &request(&fixture, "deactivated current owner"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn removed_bot_and_agent_targets_are_denied() { + let pool = setup_pool().await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + crate::channel::remove_member( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + fixture.owner.public_key().to_bytes().as_slice(), + ) + .await + .expect("remove target"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "removed target", + &request(&fixture, "removed target"), + ) + .await; + assert!(matches!(denied, Err(DbError::MemberNotFound(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + sqlx::query( + "UPDATE channel_members SET role='bot' \ + WHERE community_id=$1 AND channel_id=$2 AND pubkey=$3", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .bind(fixture.target.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("classify target membership as bot"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "bot target", + &request(&fixture, "bot target"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + sqlx::query( + "UPDATE users SET agent_owner_pubkey=$3 \ + WHERE community_id=$1 AND pubkey=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.target.public_key().to_bytes().as_slice()) + .bind(fixture.actor.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("classify target as agent"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "agent target", + &request(&fixture, "agent target"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn administrative_archive_and_wrong_replacement_do_not_qualify_owner() { + let pool = setup_pool().await; + + let fixture = setup_fixture(&pool).await; + crate::archived_identities::archive( + &pool, + fixture.community, + &fixture.owner.public_key().to_hex(), + "admin", + &fixture.actor.public_key().to_hex(), + Some("administrative archive"), + Some(&fixture.target.public_key().to_hex()), + &"dd".repeat(32), + ) + .await + .expect("archive through administrative path"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "administrative archive cannot qualify", + &request(&fixture, "administrative archive cannot qualify"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + + let fixture = setup_fixture(&pool).await; + crate::archived_identities::archive( + &pool, + fixture.community, + &fixture.owner.public_key().to_hex(), + "self", + &fixture.owner.public_key().to_hex(), + Some("named another replacement"), + Some(&Keys::generate().public_key().to_hex()), + &"ee".repeat(32), + ) + .await + .expect("archive with another replacement"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "wrong replacement", + &request(&fixture, "wrong replacement"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn active_admin_and_elevated_agent_each_block_recovery() { + let pool = setup_pool().await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + let admin = Keys::generate(); + insert_human(&pool, fixture.community, &admin).await; + crate::channel::add_member( + &pool, + fixture.community, + fixture.channel, + admin.public_key().to_bytes().as_slice(), + MemberRole::Admin, + Some(fixture.owner.public_key().to_bytes().as_slice()), + ) + .await + .expect("add active human admin"); + crate::archived_identities::archive( + &pool, + fixture.community, + &admin.public_key().to_hex(), + "self", + &admin.public_key().to_hex(), + Some("admin retired but membership still active"), + Some(&fixture.target.public_key().to_hex()), + &"ff".repeat(32), + ) + .await + .expect("archive active admin"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "admin can use ordinary path", + &request(&fixture, "admin can use ordinary path"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + let agent = Keys::generate(); + insert_human(&pool, fixture.community, &agent).await; + sqlx::query( + "UPDATE users SET agent_owner_pubkey=$3 \ + WHERE community_id=$1 AND pubkey=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(agent.public_key().to_bytes().as_slice()) + .bind(fixture.actor.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("classify elevated identity as agent"); + crate::channel::add_member( + &pool, + fixture.community, + fixture.channel, + agent.public_key().to_bytes().as_slice(), + MemberRole::Admin, + Some(fixture.owner.public_key().to_bytes().as_slice()), + ) + .await + .expect("add elevated agent"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "agent must block", + &request(&fixture, "agent must block"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn archived_deleted_and_cross_tenant_channels_are_rejected() { + let pool = setup_pool().await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + sqlx::query( + "UPDATE channels SET archived_at=NOW() \ + WHERE community_id=$1 AND id=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .execute(&pool) + .await + .expect("archive channel"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "archived channel", + &request(&fixture, "archived channel"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + + sqlx::query( + "UPDATE channels SET archived_at=NULL, deleted_at=NOW() \ + WHERE community_id=$1 AND id=$2", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .execute(&pool) + .await + .expect("delete channel"); + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "deleted channel", + &request(&fixture, "deleted channel"), + ) + .await; + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + + let other_id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id,host) VALUES ($1,$2)") + .bind(other_id) + .bind(format!("cross-tenant-{}.example", other_id.simple())) + .execute(&pool) + .await + .expect("insert other community"); + let denied = recover_channel_owner( + &pool, + CommunityId::from_uuid(other_id), + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "cross tenant", + &request(&fixture, "cross tenant"), + ) + .await; + assert!(matches!(denied, Err(DbError::ChannelNotFound(_)))); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn cross_community_target_is_not_a_channel_member() { + let pool = setup_pool().await; + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + + let other_id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id,host) VALUES ($1,$2)") + .bind(other_id) + .bind(format!("cross-target-{}.example", other_id.simple())) + .execute(&pool) + .await + .expect("insert target community"); + let other_community = CommunityId::from_uuid(other_id); + let other_target = Keys::generate(); + insert_human(&pool, other_community, &other_target).await; + let cross_target_request = + EventBuilder::new(Kind::Custom(KIND_CHANNEL_OWNER_RECOVERY as u16), "") + .tags([ + Tag::parse(["-"]).unwrap(), + Tag::parse(["h", &fixture.channel.to_string()]).unwrap(), + Tag::parse(["p", &other_target.public_key().to_hex()]).unwrap(), + Tag::parse(["reason", "cross-community target"]).unwrap(), + ]) + .sign_with_keys(&fixture.actor) + .expect("sign cross-community target request"); + + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + other_target.public_key().to_bytes().as_slice(), + "cross-community target", + &cross_target_request, + ) + .await; + assert!(matches!(denied, Err(DbError::MemberNotFound(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn concurrent_requests_allow_exactly_one_promotion_and_audit() { + let pool = setup_pool().await; + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + let request_a = request(&fixture, "concurrent A"); + let request_b = request(&fixture, "concurrent B"); + let target = fixture.target.public_key().to_bytes(); + let community = fixture.community; + let channel = fixture.channel; + let pool_a = pool.clone(); + let pool_b = pool.clone(); + + let (result_a, result_b) = tokio::join!( + recover_channel_owner( + &pool_a, + community, + channel, + target.as_slice(), + "concurrent A", + &request_a, + ), + recover_channel_owner( + &pool_b, + community, + channel, + target.as_slice(), + "concurrent B", + &request_b, + ) + ); + assert_eq!( + [result_a.is_ok(), result_b.is_ok()] + .into_iter() + .filter(|ok| *ok) + .count(), + 1 + ); + let audit_count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM channel_owner_recovery_audit \ + WHERE community_id=$1 AND channel_id=$2", + ) + .bind(community.as_uuid()) + .bind(channel) + .fetch_one(&pool) + .await + .expect("count audits"); + assert_eq!(audit_count, 1); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn target_archive_and_removal_races_cannot_partially_recover() { + let pool = setup_pool().await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + let mut archive_tx = pool.begin().await.expect("begin archive race"); + crate::archived_identities::lock_identity( + &mut archive_tx, + fixture.community, + &fixture.target.public_key().to_hex(), + ) + .await + .expect("hold target archive lock"); + let recovery_pool = pool.clone(); + let community = fixture.community; + let channel = fixture.channel; + let target = fixture.target.public_key().to_bytes(); + let recovery_request = request(&fixture, "archive race"); + let recovery = tokio::spawn(async move { + recover_channel_owner( + &recovery_pool, + community, + channel, + target.as_slice(), + "archive race", + &recovery_request, + ) + .await + }); + tokio::task::yield_now().await; + sqlx::query( + "INSERT INTO archived_identities \ + (community_id,pubkey,consent_path,actor,reason,request_event_id) \ + VALUES ($1,$2,'self',$2,'archive won race',$3)", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.target.public_key().to_hex()) + .bind("cc".repeat(32)) + .execute(&mut *archive_tx) + .await + .expect("archive target while recovery waits"); + archive_tx.commit().await.expect("commit archive"); + let denied = recovery.await.expect("join archive race"); + assert!(matches!(denied, Err(DbError::AccessDenied(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + let mut removal_tx = pool.begin().await.expect("begin removal race"); + crate::channel::lock_channel_membership( + &mut removal_tx, + fixture.community, + fixture.channel, + ) + .await + .expect("hold membership lock"); + let recovery_pool = pool.clone(); + let community = fixture.community; + let channel = fixture.channel; + let target = fixture.target.public_key().to_bytes(); + let recovery_request = request(&fixture, "removal race"); + let recovery = tokio::spawn(async move { + recover_channel_owner( + &recovery_pool, + community, + channel, + target.as_slice(), + "removal race", + &recovery_request, + ) + .await + }); + tokio::task::yield_now().await; + sqlx::query( + "UPDATE channel_members SET removed_at=NOW() \ + WHERE community_id=$1 AND channel_id=$2 AND pubkey=$3", + ) + .bind(fixture.community.as_uuid()) + .bind(fixture.channel) + .bind(fixture.target.public_key().to_bytes().as_slice()) + .execute(&mut *removal_tx) + .await + .expect("remove target while recovery waits"); + removal_tx.commit().await.expect("commit removal"); + let denied = recovery.await.expect("join removal race"); + assert!(matches!(denied, Err(DbError::MemberNotFound(_)))); + assert_no_recovery_side_effects(&pool, &fixture).await; + } +} diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9c63b2e8ab2..28fa24c9560 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -17,6 +17,8 @@ pub mod api_token; pub mod archived_identities; /// Channel and membership persistence. pub mod channel; +/// Atomic orphaned-channel ownership recovery. +pub mod channel_owner_recovery; /// Direct message channel persistence. pub mod dm; /// Database error types. @@ -1546,6 +1548,60 @@ impl Db { channel::remove_member(&self.pool, community_id, channel_id, pubkey, actor_pubkey).await } + /// Apply or idempotently replay a dedicated channel-owner recovery request. + pub async fn recover_channel_owner( + &self, + community_id: CommunityId, + channel_id: Uuid, + target_pubkey: &[u8], + reason: &str, + request: &nostr::Event, + ) -> Result { + channel_owner_recovery::recover_channel_owner( + &self.pool, + community_id, + channel_id, + target_pubkey, + reason, + request, + ) + .await + } + + /// Mark a recovery audit outbox record delivered. + pub async fn mark_recovery_delivered( + &self, + community_id: CommunityId, + request_event_id: &[u8], + ) -> Result<()> { + channel_owner_recovery::mark_recovery_delivered(&self.pool, community_id, request_event_id) + .await + } + + /// Record a retryable recovery audit delivery failure. + pub async fn record_recovery_delivery_failure( + &self, + community_id: CommunityId, + request_event_id: &[u8], + error: &str, + ) -> Result<()> { + channel_owner_recovery::record_recovery_delivery_failure( + &self.pool, + community_id, + request_event_id, + error, + ) + .await + } + + /// Load a bounded batch of pending channel recovery audit deliveries. + pub async fn pending_recovery_deliveries( + &self, + limit: i64, + ) -> Result> { + channel_owner_recovery::pending_recovery_deliveries(&self.pool, limit).await + } + /// Returns `true` if the pubkey is an active member. pub async fn is_member( &self, diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 1674b0ec4d4..8a2c58524ba 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -560,7 +560,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 24); + assert_eq!(migrations.len(), 25); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -879,6 +879,20 @@ mod tests { .to_lowercase() .contains("for update")); assert!(ttl_shared.contains("NEW.kind <> 9007")); + + // Channel-owner recovery is additive and keeps its promotion audit and + // retryable relay-delivery state community-scoped and immutable. + assert_eq!(migrations[24].version, 25); + let owner_recovery = migrations[24].sql.as_str(); + assert!(owner_recovery.contains("CREATE TABLE channel_owner_recovery_audit")); + assert!(owner_recovery.contains("CREATE TABLE channel_owner_recovery_outbox")); + assert!(owner_recovery.contains("PRIMARY KEY (community_id, request_event_id)")); + assert!(owner_recovery.contains("reason_code")); + assert!(owner_recovery.contains("BEFORE UPDATE OR DELETE")); + assert!(!migrations[0] + .sql + .as_str() + .contains("channel_owner_recovery_audit")); } #[test] diff --git a/crates/buzz-relay/src/handlers/channel_owner_recovery.rs b/crates/buzz-relay/src/handlers/channel_owner_recovery.rs new file mode 100644 index 00000000000..a422c36dfa2 --- /dev/null +++ b/crates/buzz-relay/src/handlers/channel_owner_recovery.rs @@ -0,0 +1,553 @@ +//! Dedicated orphaned-channel ownership recovery handler (kind 9038). + +use std::sync::Arc; + +use nostr::{Event, EventBuilder, Kind, Tag}; +use uuid::Uuid; + +use buzz_core::kind::KIND_SYSTEM_MESSAGE; +use buzz_core::tenant::TenantContext; + +use crate::handlers::event::dispatch_persistent_event; +use crate::handlers::side_effects::emit_group_discovery_events; +use crate::state::AppState; + +/// Validate, atomically apply, and deliver a protected recovery request. +pub async fn handle_channel_owner_recovery( + tenant: &TenantContext, + state: &Arc, + event: &Event, +) -> Result { + let (channel_id, target_hex, reason) = parse_request(event)?; + let target = hex::decode(&target_hex).map_err(|_| "invalid target pubkey".to_string())?; + + let record = state + .db + .recover_channel_owner(tenant.community(), channel_id, &target, &reason, event) + .await + .map_err(|e| e.to_string())?; + + // Repeat both cache invalidation and discovery publication on idempotent + // request replay. That gives a committed recovery a convergence path after + // a transient post-commit failure without repeating the promotion. + state.invalidate_membership(tenant, channel_id, &target); + if let Err(error) = emit_group_discovery_events(tenant, state, channel_id).await { + tracing::warn!( + channel = %channel_id, + request = %event.id, + error = %error, + "owner recovery committed but group discovery refresh failed" + ); + } + + if !record.delivered { + if let Err(error) = deliver_audit_event( + tenant, + state, + event.id.as_bytes().as_slice(), + &record.payload, + ) + .await + { + let error_text = error.to_string(); + tracing::warn!( + channel = %channel_id, + request = %event.id, + error = %error_text, + "owner recovery committed; audit event remains in durable outbox" + ); + if let Err(db_error) = state + .db + .record_recovery_delivery_failure( + tenant.community(), + event.id.as_bytes().as_slice(), + &error_text, + ) + .await + { + tracing::error!( + request = %event.id, + error = %db_error, + "failed to record owner recovery audit delivery failure" + ); + } + return Ok("recovered; channel audit delivery pending retry".into()); + } + } + + Ok(if record.applied { + "channel ownership recovered".into() + } else { + "channel ownership recovery already applied".into() + }) +} + +async fn deliver_audit_event( + tenant: &TenantContext, + state: &Arc, + request_event_id: &[u8], + payload: &buzz_db::channel_owner_recovery::RecoveryAuditPayload, +) -> anyhow::Result<()> { + let audit = build_audit_event(&state.relay_keypair, payload)?; + let (stored, inserted) = state + .db + .insert_event(tenant.community(), &audit, Some(payload.channel_id)) + .await?; + if inserted { + let relay_pubkey = state.relay_keypair.public_key().to_hex(); + dispatch_persistent_event( + tenant, + state, + &stored, + KIND_SYSTEM_MESSAGE, + &relay_pubkey, + None, + ) + .await; + } + state + .db + .mark_recovery_delivered(tenant.community(), request_event_id) + .await?; + Ok(()) +} + +/// Drain a bounded batch from the durable recovery-audit outbox. +/// +/// The relay-signed audit event is deterministic, so concurrent relay pods +/// converge on one persisted event ID before each marks the outbox row +/// delivered. +pub async fn drain_pending_recovery_audits( + state: &Arc, + limit: i64, +) -> anyhow::Result { + let pending = state.db.pending_recovery_deliveries(limit).await?; + let mut delivered = 0; + for item in pending { + let tenant = TenantContext::resolved(item.community_id, item.host); + match deliver_audit_event(&tenant, state, &item.request_event_id, &item.payload).await { + Ok(()) => delivered += 1, + Err(error) => { + let error_text = error.to_string(); + tracing::warn!( + community = %item.community_id, + channel = %item.payload.channel_id, + request = %hex::encode(&item.request_event_id), + error = %error_text, + "pending owner recovery audit delivery failed" + ); + if let Err(db_error) = state + .db + .record_recovery_delivery_failure( + item.community_id, + &item.request_event_id, + &error_text, + ) + .await + { + tracing::error!( + request = %hex::encode(&item.request_event_id), + error = %db_error, + "failed to record pending recovery audit delivery failure" + ); + } + } + } + } + Ok(delivered) +} + +fn build_audit_event( + relay_keys: &nostr::Keys, + payload: &buzz_db::channel_owner_recovery::RecoveryAuditPayload, +) -> anyhow::Result { + let content = serde_json::to_string(payload)?; + let channel = payload.channel_id.to_string(); + let created_at = payload.created_at.timestamp().max(0) as u64; + Ok( + EventBuilder::new(Kind::Custom(KIND_SYSTEM_MESSAGE as u16), content) + .tags([ + Tag::parse(["h", &channel])?, + Tag::parse(["e", &payload.request_event_id])?, + Tag::parse(["p", &payload.actor])?, + Tag::parse(["p", &payload.target])?, + Tag::parse(["predicate", &payload.predicate_id])?, + Tag::parse(["reason-code", &payload.reason_code])?, + ]) + .custom_created_at(nostr::Timestamp::from(created_at)) + .sign_with_keys(relay_keys)?, + ) +} + +fn parse_request(event: &Event) -> Result<(Uuid, String, String), String> { + if !event.content.is_empty() { + return Err("recovery request content must be empty".into()); + } + let mut protected = 0; + let mut channel = None; + let mut target = None; + let mut reason = None; + for tag in event.tags.iter() { + let parts = tag.as_slice(); + match parts.first().map(String::as_str) { + Some("-") if parts.len() == 1 => protected += 1, + Some("h") if parts.len() == 2 && channel.is_none() => { + channel = Some( + Uuid::parse_str(&parts[1]).map_err(|_| "invalid channel h tag".to_string())?, + ); + } + Some("p") if parts.len() == 2 && target.is_none() => { + let value = &parts[1]; + if value.len() != 64 + || !value.chars().all(|ch| ch.is_ascii_hexdigit()) + || value.to_ascii_lowercase() != *value + { + return Err("target p tag must be lowercase 64-character hex".into()); + } + target = Some(value.clone()); + } + Some("reason") if parts.len() == 2 && reason.is_none() => { + let value = parts[1].trim(); + if value.is_empty() || value.len() > 500 || value.chars().any(char::is_control) { + return Err("reason must be 1–500 bytes without control characters".into()); + } + reason = Some(value.to_string()); + } + _ => return Err("malformed, duplicate, or unsupported recovery request tag".into()), + } + } + if protected != 1 { + return Err("request must include exactly one NIP-70 protected tag".into()); + } + let channel = channel.ok_or_else(|| "missing channel h tag".to_string())?; + if channel.is_nil() { + return Err("channel h tag must not be nil".into()); + } + Ok(( + channel, + target.ok_or_else(|| "missing target p tag".to_string())?, + reason.ok_or_else(|| "missing reason tag".to_string())?, + )) +} + +#[cfg(test)] +mod tests { + use chrono::{DateTime, TimeZone as _, Utc}; + use nostr::{EventBuilder, Keys, Kind, Tag}; + + use buzz_core::kind::KIND_CHANNEL_OWNER_RECOVERY; + + use super::*; + + fn request(tags: Vec, content: &str) -> Event { + EventBuilder::new(Kind::Custom(KIND_CHANNEL_OWNER_RECOVERY as u16), content) + .tags(tags) + .sign_with_keys(&Keys::generate()) + .expect("sign") + } + + #[test] + fn request_parser_requires_exact_protected_shape() { + let channel = Uuid::new_v4(); + let target = "11".repeat(32); + let event = request( + vec![ + Tag::parse(["-"]).unwrap(), + Tag::parse(["h", &channel.to_string()]).unwrap(), + Tag::parse(["p", &target]).unwrap(), + Tag::parse(["reason", "durable self-consent recorded"]).unwrap(), + ], + "", + ); + assert_eq!( + parse_request(&event).unwrap(), + (channel, target, "durable self-consent recorded".to_string()) + ); + } + + #[test] + fn request_parser_rejects_duplicates_unknown_tags_and_content() { + let channel = Uuid::new_v4().to_string(); + let target = "11".repeat(32); + for tags in [ + vec![ + Tag::parse(["-"]).unwrap(), + Tag::parse(["-"]).unwrap(), + Tag::parse(["h", &channel]).unwrap(), + Tag::parse(["p", &target]).unwrap(), + Tag::parse(["reason", "reason"]).unwrap(), + ], + vec![ + Tag::parse(["-"]).unwrap(), + Tag::parse(["h", &channel]).unwrap(), + Tag::parse(["p", &target]).unwrap(), + Tag::parse(["reason", "reason"]).unwrap(), + Tag::parse(["x", "unexpected"]).unwrap(), + ], + ] { + assert!(parse_request(&request(tags, "")).is_err()); + } + let valid_tags = vec![ + Tag::parse(["-"]).unwrap(), + Tag::parse(["h", &channel]).unwrap(), + Tag::parse(["p", &target]).unwrap(), + Tag::parse(["reason", "reason"]).unwrap(), + ]; + assert!(parse_request(&request(valid_tags, "not empty")).is_err()); + } + + #[test] + fn request_parser_rejects_missing_and_malformed_required_tags() { + let channel = Uuid::new_v4().to_string(); + let target = "11".repeat(32); + let valid = || { + vec![ + Tag::parse(["-"]).unwrap(), + Tag::parse(["h", &channel]).unwrap(), + Tag::parse(["p", &target]).unwrap(), + Tag::parse(["reason", "reason"]).unwrap(), + ] + }; + + for missing_index in 0..4 { + let mut tags = valid(); + tags.remove(missing_index); + assert!(parse_request(&request(tags, "")).is_err()); + } + + for replacement in [ + Tag::parse(["h", "not-a-uuid"]).unwrap(), + Tag::parse(["h", &Uuid::nil().to_string()]).unwrap(), + Tag::parse(["p", &"1".repeat(63)]).unwrap(), + Tag::parse(["p", &"AA".repeat(32)]).unwrap(), + Tag::parse(["reason", "bad\nreason"]).unwrap(), + ] { + let mut tags = valid(); + let tag_name = replacement.as_slice()[0].clone(); + let index = tags + .iter() + .position(|tag| tag.as_slice()[0] == tag_name) + .expect("replace required tag"); + tags[index] = replacement; + assert!(parse_request(&request(tags, "")).is_err()); + } + } + + #[test] + fn audit_event_is_exact_relay_signed_and_deterministic() { + let relay = Keys::generate(); + let payload = buzz_db::channel_owner_recovery::RecoveryAuditPayload { + schema_version: 1, + event_type: "channel_owner_recovered".into(), + community_id: Uuid::new_v4(), + channel_id: Uuid::new_v4(), + request_event_id: "22".repeat(32), + actor: "33".repeat(32), + target: "44".repeat(32), + predicate_id: buzz_db::channel_owner_recovery::RECOVERY_PREDICATE_ID.into(), + reason_code: buzz_db::channel_owner_recovery::RECOVERY_REASON_CODE.into(), + reason: "durable consent recovery".into(), + prior_elevated_roles: vec![buzz_db::channel_owner_recovery::PriorElevatedRole { + pubkey: "55".repeat(32), + role: "owner".into(), + }], + created_at: Utc.timestamp_opt(1_750_000_000, 0).unwrap(), + }; + + let first = build_audit_event(&relay, &payload).expect("build audit event"); + let replay = build_audit_event(&relay, &payload).expect("rebuild audit event"); + assert_eq!(first.id, replay.id); + assert_eq!(first.pubkey, relay.public_key()); + assert_eq!(first.kind, Kind::Custom(KIND_SYSTEM_MESSAGE as u16)); + assert_eq!( + serde_json::from_str::( + &first.content + ) + .expect("decode payload"), + payload + ); + let tags: Vec> = first + .tags + .iter() + .map(|tag| tag.as_slice().to_vec()) + .collect(); + assert_eq!( + tags, + vec![ + vec!["h".into(), payload.channel_id.to_string()], + vec!["e".into(), payload.request_event_id], + vec!["p".into(), payload.actor], + vec!["p".into(), payload.target], + vec!["predicate".into(), payload.predicate_id], + vec!["reason-code".into(), payload.reason_code], + ] + ); + buzz_core::verification::verify_event(&first).expect("verify relay signature"); + } + + async fn test_pool() -> sqlx::PgPool { + let url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".into()); + sqlx::PgPool::connect(&url) + .await + .expect("connect to recovery integration test database") + } + + async fn test_state(pool: sqlx::PgPool) -> Arc { + let db = buzz_db::Db::from_pool(pool.clone()); + let config = crate::config::Config::from_env().expect("load relay test configuration"); + let redis_pool = deadpool_redis::Config::from_url(&config.redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("create recovery integration Redis pool"); + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone()) + .await + .expect("connect recovery integration pubsub"), + ); + let audit = buzz_audit::AuditService::new(pool.clone()); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool.clone()); + let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let media_storage = + buzz_media::MediaStorage::new(&config.media).expect("create test media storage"); + let (state, _audit_shutdown) = crate::state::AppState::new( + config, + db, + redis_pool, + audit, + pubsub, + auth, + search, + workflow_engine, + Keys::generate(), + media_storage, + ); + Arc::new(state) + } + + #[tokio::test] + #[ignore = "requires migrated Postgres and Redis"] + async fn successful_handler_persists_and_delivers_channel_visible_audit() { + let pool = test_pool().await; + sqlx::query("SELECT 1 FROM channel_owner_recovery_audit LIMIT 1") + .execute(&pool) + .await + .expect("owner recovery migration must be applied"); + let state = test_state(pool.clone()).await; + + let community_uuid = Uuid::new_v4(); + let host = format!("owner-recovery-handler-{}.example", community_uuid.simple()); + sqlx::query("INSERT INTO communities (id,host) VALUES ($1,$2)") + .bind(community_uuid) + .bind(&host) + .execute(&pool) + .await + .expect("insert community"); + let community = buzz_core::CommunityId::from_uuid(community_uuid); + let tenant = TenantContext::resolved(community, host); + let actor = Keys::generate(); + let owner = Keys::generate(); + let target = Keys::generate(); + for keys in [&actor, &owner, &target] { + sqlx::query("INSERT INTO users (community_id,pubkey) VALUES ($1,$2)") + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("insert human"); + } + sqlx::query("INSERT INTO relay_members (community_id,pubkey,role) VALUES ($1,$2,'owner')") + .bind(community.as_uuid()) + .bind(actor.public_key().to_hex()) + .execute(&pool) + .await + .expect("insert community owner"); + let channel = buzz_db::channel::create_channel( + &pool, + community, + "recoverable", + buzz_core::channel::ChannelType::Stream, + buzz_core::channel::ChannelVisibility::Open, + None, + owner.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("create channel") + .id; + buzz_db::channel::add_member( + &pool, + community, + channel, + target.public_key().to_bytes().as_slice(), + buzz_core::channel::MemberRole::Member, + Some(owner.public_key().to_bytes().as_slice()), + ) + .await + .expect("add target"); + state + .db + .archive( + community, + &owner.public_key().to_hex(), + "self", + &owner.public_key().to_hex(), + Some("retired"), + Some(&target.public_key().to_hex()), + &"ab".repeat(32), + ) + .await + .expect("archive owner"); + + let channel_tag = channel.to_string(); + let target_tag = target.public_key().to_hex(); + let request = EventBuilder::new(Kind::Custom(KIND_CHANNEL_OWNER_RECOVERY as u16), "") + .tags([ + Tag::parse(["-"]).unwrap(), + Tag::parse(["h", &channel_tag]).unwrap(), + Tag::parse(["p", &target_tag]).unwrap(), + Tag::parse(["reason", "handler delivery test"]).unwrap(), + ]) + .sign_with_keys(&actor) + .expect("sign request"); + let message = handle_channel_owner_recovery(&tenant, &state, &request) + .await + .expect("handle recovery"); + assert_eq!(message, "channel ownership recovered"); + + let delivered: Option> = sqlx::query_scalar( + "SELECT delivered_at FROM channel_owner_recovery_outbox \ + WHERE community_id=$1 AND request_event_id=$2", + ) + .bind(community.as_uuid()) + .bind(request.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("read outbox"); + assert!(delivered.is_some()); + + let events = state + .db + .query_events(&buzz_db::EventQuery { + channel_id: Some(channel), + kinds: Some(vec![KIND_SYSTEM_MESSAGE as i32]), + limit: Some(10), + ..buzz_db::EventQuery::for_community(community) + }) + .await + .expect("query channel audit"); + let audit = events + .into_iter() + .find(|stored| stored.event.content.contains("channel_owner_recovered")) + .expect("channel-visible recovery audit"); + let payload: buzz_db::channel_owner_recovery::RecoveryAuditPayload = + serde_json::from_str(&audit.event.content).expect("decode audit"); + assert_eq!(payload.actor, actor.public_key().to_hex()); + assert_eq!(payload.target, target.public_key().to_hex()); + assert_eq!(payload.reason, "handler delivery test"); + buzz_core::verification::verify_event(&audit.event).expect("verify relay signature"); + } +} diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 88dd5f5180e..c15bc0c9b6e 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -35,7 +35,7 @@ fn reject(reason: &'static str) { pub(crate) fn bounded_kind_label(kind: u32) -> String { match kind { 0..=9 | 1059 | 1063 => kind.to_string(), - 8000..=8003 | 9000..=9022 | 9030..=9036 => kind.to_string(), + 8000..=8003 | 9000..=9022 | 9030..=9036 | 9038 => kind.to_string(), 13534..=13535 => kind.to_string(), 20000..=29999 => kind.to_string(), 30023 | 30315 | 39000..=39003 => kind.to_string(), diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 497adc6c737..b7ccb76b98b 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -14,22 +14,22 @@ use buzz_core::kind::{ event_kind_u32, is_identity_archive_request_kind, is_parameterized_replaceable, is_relay_admin_kind, KIND_AGENT_ENGRAM, KIND_AGENT_PROFILE, KIND_AGENT_TURN_METRIC, KIND_APPROVAL_DENY, KIND_APPROVAL_GRANT, KIND_AUTH, KIND_BOOKMARK_LIST, KIND_BOOKMARK_SET, - KIND_CANVAS, KIND_CONTACT_LIST, KIND_DELETION, KIND_DM_ADD_MEMBER, KIND_DM_HIDE, KIND_DM_OPEN, - KIND_EMOJI_LIST, KIND_EMOJI_SET, KIND_EVENT_REMINDER, KIND_FOLLOW_SET, KIND_FORUM_COMMENT, - KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_GIFT_WRAP, KIND_GIT_ISSUE, KIND_GIT_PATCH, - KIND_GIT_PR_UPDATE, KIND_GIT_PULL_REQUEST, KIND_GIT_REPO_ANNOUNCEMENT, KIND_GIT_REPO_STATE, - KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, KIND_GIT_STATUS_MERGED, KIND_GIT_STATUS_OPEN, - KIND_HUDDLE_ENDED, KIND_HUDDLE_GUIDELINES, KIND_HUDDLE_PARTICIPANT_JOINED, - KIND_HUDDLE_PARTICIPANT_LEFT, KIND_HUDDLE_STARTED, KIND_IA_ARCHIVE_REQUEST, - KIND_IA_UNARCHIVE_REQUEST, KIND_LONG_FORM, KIND_MANAGED_AGENT, KIND_MEMBER_ADDED_NOTIFICATION, - KIND_MEMBER_REMOVED_NOTIFICATION, KIND_MODERATION_BAN, KIND_MODERATION_RESOLVE_REPORT, - KIND_MODERATION_TIMEOUT, KIND_MODERATION_UNBAN, KIND_MODERATION_UNTIMEOUT, KIND_MUTE_LIST, - KIND_NIP29_CREATE_GROUP, KIND_NIP29_DELETE_EVENT, KIND_NIP29_DELETE_GROUP, - KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST, KIND_NIP29_LEAVE_REQUEST, - KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER, KIND_NIP43_LEAVE_REQUEST, - KIND_NIP65_RELAY_LIST_METADATA, KIND_PERSONA, KIND_PIN_LIST, KIND_PRESENCE_UPDATE, - KIND_PRODUCT_FEEDBACK, KIND_PROFILE, KIND_REACTION, KIND_READ_STATE, KIND_REPORT, - KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_DIFF, + KIND_CANVAS, KIND_CHANNEL_OWNER_RECOVERY, KIND_CONTACT_LIST, KIND_DELETION, KIND_DM_ADD_MEMBER, + KIND_DM_HIDE, KIND_DM_OPEN, KIND_EMOJI_LIST, KIND_EMOJI_SET, KIND_EVENT_REMINDER, + KIND_FOLLOW_SET, KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_GIFT_WRAP, + KIND_GIT_ISSUE, KIND_GIT_PATCH, KIND_GIT_PR_UPDATE, KIND_GIT_PULL_REQUEST, + KIND_GIT_REPO_ANNOUNCEMENT, KIND_GIT_REPO_STATE, KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, + KIND_GIT_STATUS_MERGED, KIND_GIT_STATUS_OPEN, KIND_HUDDLE_ENDED, KIND_HUDDLE_GUIDELINES, + KIND_HUDDLE_PARTICIPANT_JOINED, KIND_HUDDLE_PARTICIPANT_LEFT, KIND_HUDDLE_STARTED, + KIND_IA_ARCHIVE_REQUEST, KIND_IA_UNARCHIVE_REQUEST, KIND_LONG_FORM, KIND_MANAGED_AGENT, + KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_MODERATION_BAN, + KIND_MODERATION_RESOLVE_REPORT, KIND_MODERATION_TIMEOUT, KIND_MODERATION_UNBAN, + KIND_MODERATION_UNTIMEOUT, KIND_MUTE_LIST, KIND_NIP29_CREATE_GROUP, KIND_NIP29_DELETE_EVENT, + KIND_NIP29_DELETE_GROUP, KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST, + KIND_NIP29_LEAVE_REQUEST, KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER, + KIND_NIP43_LEAVE_REQUEST, KIND_NIP65_RELAY_LIST_METADATA, KIND_PERSONA, KIND_PIN_LIST, + KIND_PRESENCE_UPDATE, KIND_PRODUCT_FEEDBACK, KIND_PROFILE, KIND_REACTION, KIND_READ_STATE, + KIND_REPORT, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_PINNED, KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, KIND_TEAM, KIND_TEXT_NOTE, KIND_USER_STATUS, KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE, @@ -246,6 +246,7 @@ fn required_scope_for_kind(kind: u32, event: &Event) -> Result { Ok(Scope::AdminChannels) } + KIND_CHANNEL_OWNER_RECOVERY => Ok(Scope::AdminChannels), // NIP-43: relay membership admin commands (9030–9032) + Buzz // workspace-profile command (9033). k if k == RELAY_ADMIN_ADD_MEMBER @@ -473,6 +474,7 @@ pub(crate) fn requires_h_channel_scope(kind: u32) -> bool { | KIND_NIP29_EDIT_METADATA | KIND_NIP29_DELETE_EVENT | KIND_NIP29_DELETE_GROUP + | KIND_CHANNEL_OWNER_RECOVERY | KIND_NIP29_LEAVE_REQUEST // Huddle lifecycle events + guidelines | KIND_HUDDLE_STARTED @@ -1798,7 +1800,8 @@ async fn ingest_event_inner( || kind_u32 == KIND_STREAM_MESSAGE_EDIT || kind_u32 == KIND_NIP29_EDIT_METADATA || kind_u32 == KIND_NIP29_DELETE_EVENT - || kind_u32 == KIND_NIP29_DELETE_GROUP; + || kind_u32 == KIND_NIP29_DELETE_GROUP + || kind_u32 == KIND_CHANNEL_OWNER_RECOVERY; if !skip_membership { // Spec AuthCheck (line 794): emit the verdict at the actual // call site. claimed_community comes from the event's h tag @@ -1830,6 +1833,19 @@ async fn ingest_event_inner( } } + if kind_u32 == KIND_CHANNEL_OWNER_RECOVERY { + let message = crate::handlers::channel_owner_recovery::handle_channel_owner_recovery( + tenant, state, &event, + ) + .await + .map_err(|error| IngestError::Rejected(format!("invalid: {error}")))?; + return Ok(IngestResult { + event_id: event_id_hex, + accepted: true, + message, + }); + } + // Handled directly — these mutate relay_members and do NOT get stored. if is_relay_admin_kind(event.kind.as_u16() as u32) { crate::handlers::relay_admin::handle_relay_admin_event(tenant, state, &event) diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 98a5e6c51d2..87cb238bcd6 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -1,5 +1,7 @@ /// NIP-42 authentication handler. pub mod auth; +/// Dedicated orphaned-channel ownership recovery (kind 9038). +pub mod channel_owner_recovery; /// Subscription close (CLOSE) handler. pub mod close; /// Command executor — transactional processing for command kinds. diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 22101219ebf..b4da7a728d0 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -680,6 +680,47 @@ async fn main() -> anyhow::Result<()> { }); } + // Durable channel-owner recovery audit delivery. The ownership promotion, + // immutable audit row, and outbox entry commit together; this bounded + // sweeper makes the relay-signed channel event converge after a crash or a + // transient post-commit delivery failure. Deterministic event IDs make + // concurrent relay pods harmless. + { + let recovery_audit_state = Arc::clone(&state); + let interval_secs: u64 = std::env::var("BUZZ_RECOVERY_AUDIT_INTERVAL_SECS") + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(10) + .max(1); + let batch_limit: i64 = std::env::var("BUZZ_RECOVERY_AUDIT_BATCH_LIMIT") + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(100); + tokio::spawn(async move { + info!( + interval_secs, + batch_limit, "Channel-owner recovery audit worker started" + ); + loop { + match buzz_relay::handlers::channel_owner_recovery::drain_pending_recovery_audits( + &recovery_audit_state, + batch_limit, + ) + .await + { + Ok(0) => {} + Ok(delivered) => { + info!(delivered, "Delivered pending channel-owner recovery audits"); + } + Err(error) => { + warn!("Channel-owner recovery audit worker tick failed: {error}"); + } + } + tokio::time::sleep(std::time::Duration::from_secs(interval_secs)).await; + } + }); + } + // NIP-PL matcher and worker are enabled as one unit. Lease acceptance is // already disabled without the exact gateway URL, so discovery and runtime // cannot advertise or accumulate work for an undeliverable configuration. diff --git a/crates/buzz-sdk/src/builders.rs b/crates/buzz-sdk/src/builders.rs index f9e54de9c52..c7617632349 100644 --- a/crates/buzz-sdk/src/builders.rs +++ b/crates/buzz-sdk/src/builders.rs @@ -5,14 +5,14 @@ use buzz_core::{ kind::{ - KIND_AGENT_OBSERVER_FRAME, KIND_APPROVAL_DENY, KIND_APPROVAL_GRANT, KIND_DELETION, - KIND_DM_ADD_MEMBER, KIND_DM_OPEN, KIND_EMOJI_SET, KIND_GIT_ISSUE, KIND_GIT_PATCH, - KIND_GIT_PR_UPDATE, KIND_GIT_PULL_REQUEST, KIND_GIT_REPO_ANNOUNCEMENT, - KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, KIND_GIT_STATUS_MERGED, - KIND_GIT_STATUS_OPEN, KIND_IA_ARCHIVE_REQUEST, KIND_IA_UNARCHIVE_REQUEST, - KIND_MODERATION_BAN, KIND_MODERATION_RESOLVE_REPORT, KIND_MODERATION_TIMEOUT, - KIND_MODERATION_UNBAN, KIND_MODERATION_UNTIMEOUT, KIND_PRESENCE_UPDATE, KIND_WORKFLOW_DEF, - KIND_WORKFLOW_TRIGGER, + KIND_AGENT_OBSERVER_FRAME, KIND_APPROVAL_DENY, KIND_APPROVAL_GRANT, + KIND_CHANNEL_OWNER_RECOVERY, KIND_DELETION, KIND_DM_ADD_MEMBER, KIND_DM_OPEN, + KIND_EMOJI_SET, KIND_GIT_ISSUE, KIND_GIT_PATCH, KIND_GIT_PR_UPDATE, KIND_GIT_PULL_REQUEST, + KIND_GIT_REPO_ANNOUNCEMENT, KIND_GIT_STATUS_CLOSED, KIND_GIT_STATUS_DRAFT, + KIND_GIT_STATUS_MERGED, KIND_GIT_STATUS_OPEN, KIND_IA_ARCHIVE_REQUEST, + KIND_IA_UNARCHIVE_REQUEST, KIND_MODERATION_BAN, KIND_MODERATION_RESOLVE_REPORT, + KIND_MODERATION_TIMEOUT, KIND_MODERATION_UNBAN, KIND_MODERATION_UNTIMEOUT, + KIND_PRESENCE_UPDATE, KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, }, observer::{ content_looks_like_nip44, OBSERVER_AGENT_TAG, OBSERVER_FRAME_CONTROL, OBSERVER_FRAME_TAG, @@ -75,6 +75,45 @@ fn check_pubkey_hex(s: &str, field: &str) -> Result { Ok(s.to_ascii_lowercase()) } +/// Build a protected request to recover an orphaned channel's ownership. +/// +/// The relay applies the narrow `all_current_human_owners_self_archived_for_target_v1` +/// predicate. This builder cannot grant owner through the generic role-change +/// path and does not imply recovery for lost or deleted keys. +pub fn build_channel_owner_recovery( + channel_id: Uuid, + target_pubkey: &str, + reason: &str, +) -> Result { + if channel_id.is_nil() { + return Err(SdkError::InvalidInput( + "channel_id must not be nil".to_string(), + )); + } + let target = check_pubkey_hex(target_pubkey, "target_pubkey")?; + let reason = reason.trim(); + if reason.is_empty() { + return Err(SdkError::InvalidInput( + "recovery reason must not be empty".to_string(), + )); + } + check_content(reason, 500)?; + if reason.chars().any(char::is_control) { + return Err(SdkError::InvalidInput( + "recovery reason must not contain control characters".to_string(), + )); + } + + Ok( + EventBuilder::new(Kind::Custom(KIND_CHANNEL_OWNER_RECOVERY as u16), "").tags([ + tag(&["-"])?, + tag(&["h", &channel_id.to_string()])?, + tag(&["p", &target])?, + tag(&["reason", reason])?, + ]), + ) +} + /// Validate an exact-length hex string (event ids), returning it lowercased. fn check_hex_exact(s: &str, len: usize, field: &str) -> Result { if s.len() != len || !s.chars().all(|c| c.is_ascii_hexdigit()) { @@ -3821,4 +3860,39 @@ mod tests { .iter() .any(|t| t.as_slice().first().map(String::as_str) == Some("replaced-by"))); } + + #[test] + fn channel_owner_recovery_has_dedicated_protected_wire_shape() { + let channel = Uuid::new_v4(); + let target = "ab".repeat(32); + let event = + sign(build_channel_owner_recovery(channel, &target, "Prior consent recorded").unwrap()); + let tags: Vec> = event + .tags + .iter() + .map(|tag| tag.as_slice().to_vec()) + .collect(); + assert_eq!(event.kind, Kind::Custom(KIND_CHANNEL_OWNER_RECOVERY as u16)); + assert_eq!( + tags, + vec![ + vec!["-"], + vec!["h", &channel.to_string()], + vec!["p", &target], + vec!["reason", "Prior consent recorded"], + ] + ); + assert!(event.content.is_empty()); + } + + #[test] + fn channel_owner_recovery_rejects_unsafe_inputs() { + let channel = Uuid::new_v4(); + let target = "ab".repeat(32); + assert!(build_channel_owner_recovery(Uuid::nil(), &target, "reason").is_err()); + assert!(build_channel_owner_recovery(channel, "not-a-key", "reason").is_err()); + assert!(build_channel_owner_recovery(channel, &target, " ").is_err()); + assert!(build_channel_owner_recovery(channel, &target, "bad\nreason").is_err()); + assert!(build_channel_owner_recovery(channel, &target, &"x".repeat(501)).is_err()); + } } diff --git a/desktop/playwright.config.ts b/desktop/playwright.config.ts index e59c68bfafb..b5b4ac88ff2 100644 --- a/desktop/playwright.config.ts +++ b/desktop/playwright.config.ts @@ -40,6 +40,7 @@ export default defineConfig({ "**/channel-mute.spec.ts", "**/channel-star.spec.ts", "**/channel-controls.spec.ts", + "**/channel-owner-recovery.spec.ts", "**/active-turn-resilience.spec.ts", "**/profile-active-turn.spec.ts", "**/config-bridge-screenshots.spec.ts", diff --git a/desktop/src-tauri/src/commands/channels.rs b/desktop/src-tauri/src/commands/channels.rs index 59c80c48076..698d10f6d05 100644 --- a/desktop/src-tauri/src/commands/channels.rs +++ b/desktop/src-tauri/src/commands/channels.rs @@ -848,6 +848,20 @@ pub async fn change_channel_member_role( Ok(()) } +#[tauri::command] +pub async fn recover_channel_owner( + channel_id: String, + target_pubkey: String, + reason: String, + state: State<'_, AppState>, +) -> Result<(), String> { + let uuid = parse_channel_uuid(&channel_id)?; + let builder = buzz_sdk_pkg::build_channel_owner_recovery(uuid, &target_pubkey, &reason) + .map_err(|error| error.to_string())?; + submit_event(builder, &state).await?; + Ok(()) +} + #[tauri::command] pub async fn join_channel(channel_id: String, state: State<'_, AppState>) -> Result<(), String> { let uuid = parse_channel_uuid(&channel_id)?; diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index c5b71987cec..1b99e7d391a 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -730,6 +730,7 @@ pub fn run() { add_channel_members, remove_channel_member, change_channel_member_role, + recover_channel_owner, join_channel, leave_channel, get_canvas, diff --git a/desktop/src/features/channels/hooks.ts b/desktop/src/features/channels/hooks.ts index 9003d0f5a58..55ddeea32df 100644 --- a/desktop/src/features/channels/hooks.ts +++ b/desktop/src/features/channels/hooks.ts @@ -15,6 +15,7 @@ import { leaveChannel, openDm, removeChannelMember, + recoverChannelOwner, setCanvas, setChannelPurpose, setChannelTopic, @@ -547,6 +548,32 @@ export function useRemoveChannelMemberMutation(channelId: string | null) { }); } +export function useRecoverChannelOwnerMutation(channelId: string | null) { + const queryClient = useQueryClient(); + + return useMutation({ + mutationFn: async ({ + reason, + targetPubkey, + }: { + reason: string; + targetPubkey: string; + }) => { + if (!channelId) { + throw new Error("No channel selected."); + } + await recoverChannelOwner({ + channelId, + reason, + targetPubkey, + }); + }, + onSettled: async () => { + await invalidateChannelState(queryClient, channelId); + }, + }); +} + export function useJoinChannelMutation(channelId: string | null) { const queryClient = useQueryClient(); diff --git a/desktop/src/features/channels/ui/ChannelManagementSheet.tsx b/desktop/src/features/channels/ui/ChannelManagementSheet.tsx index 3c1727e00a0..154faffdeb1 100644 --- a/desktop/src/features/channels/ui/ChannelManagementSheet.tsx +++ b/desktop/src/features/channels/ui/ChannelManagementSheet.tsx @@ -11,6 +11,7 @@ import { MessageSquare, Pencil, Radio, + ShieldCheck, Type, Users, Zap, @@ -27,10 +28,16 @@ import { useDeleteChannelMutation, useJoinChannelMutation, useLeaveChannelMutation, + useRecoverChannelOwnerMutation, useUnarchiveChannelMutation, useUpdateChannelMutation, } from "@/features/channels/hooks"; -import { compareMembersByRole } from "@/features/channels/lib/memberUtils"; +import { useMyRelayMembershipQuery } from "@/features/community-members/hooks"; +import { useUsersBatchQuery } from "@/features/profile/hooks"; +import { + compareMembersByRole, + formatMemberName, +} from "@/features/channels/lib/memberUtils"; import { DEFAULT_EPHEMERAL_TTL_SECONDS, formatTtlDuration, @@ -69,6 +76,7 @@ import { CHANNEL_FORM_FIELD_SHELL_CLASS, } from "./channelFormStyles"; import { ChannelTypeSettings } from "./ChannelTypeSettings"; +import { ChannelOwnerRecoveryDialog } from "./ChannelOwnerRecoveryDialog"; import { ChannelPermissionsSettings } from "./ChannelPermissionsSettings"; import { ChannelHero, @@ -124,6 +132,8 @@ export function ChannelManagementSheet({ const deleteChannelMutation = useDeleteChannelMutation(channelId); const joinChannelMutation = useJoinChannelMutation(channelId); const leaveChannelMutation = useLeaveChannelMutation(channelId); + const recoverChannelOwnerMutation = useRecoverChannelOwnerMutation(channelId); + const myRelayMembershipQuery = useMyRelayMembershipQuery(); const channelIdRef = React.useRef(channelId); channelIdRef.current = channelId; @@ -137,6 +147,15 @@ export function ChannelManagementSheet({ const selfMember = members.find((member) => member.pubkey === currentPubkey) ?? null; const hasResolvedMembership = membersQuery.data !== undefined; + const recoveryIdentitiesQuery = useUsersBatchQuery( + [ + ...(currentPubkey ? [currentPubkey] : []), + ...members.map((member) => member.pubkey), + ], + { + enabled: open && myRelayMembershipQuery.data?.role === "owner", + }, + ); const { canDeleteChannel, canManageChannel } = useChannelModerationCapabilities(membersQuery.data, currentPubkey, open); @@ -167,6 +186,7 @@ export function ChannelManagementSheet({ ); const [isDeleteDialogOpen, setIsDeleteDialogOpen] = React.useState(false); const [isEditDialogOpen, setIsEditDialogOpen] = React.useState(false); + const [isRecoveryDialogOpen, setIsRecoveryDialogOpen] = React.useState(false); const [isConvertingVisibility, setIsConvertingVisibility] = React.useState(false); const [hasUserEditedChannelDraft, setHasUserEditedChannelDraft] = @@ -184,6 +204,7 @@ export function ChannelManagementSheet({ syncedForRef.current = null; setIsDeleteDialogOpen(false); setIsEditDialogOpen(false); + setIsRecoveryDialogOpen(false); setActiveView("summary"); return; } @@ -263,6 +284,34 @@ export function ChannelManagementSheet({ ? getMarkdownPreviewText(canvasContent) : undefined; const canOpenCanvas = hasCanvas || canEditNarrative; + function isKnownHuman(pubkey: string) { + const normalizedPubkey = pubkey.toLowerCase(); + const profile = + recoveryIdentitiesQuery.data?.profiles[normalizedPubkey] ?? undefined; + return ( + profile !== undefined && + profile.isAgent === false && + profile.ownerPubkey === null && + !recoveryIdentitiesQuery.data?.missing.some( + (missingPubkey) => missingPubkey.toLowerCase() === normalizedPubkey, + ) + ); + } + const recoveryCandidates = members.filter( + (member) => + isKnownHuman(member.pubkey) && + (member.role === "member" || member.role === "guest"), + ); + const currentOwners = members.filter((member) => member.role === "owner"); + const currentIdentityIsKnownHuman = + currentPubkey !== undefined && isKnownHuman(currentPubkey); + const canRequestOwnerRecovery = + hasResolvedMembership && + myRelayMembershipQuery.data?.role === "owner" && + currentIdentityIsKnownHuman && + !isArchived && + resolvedChannel.channelType !== "dm" && + recoveryCandidates.length > 0; function handleEditDialogOpenChange(next: boolean) { if (!next) { @@ -377,6 +426,8 @@ export function ChannelManagementSheet({ resolvedChannel={resolvedChannel} setActiveView={setActiveView} setIsEditDialogOpen={setIsEditDialogOpen} + setIsRecoveryDialogOpen={setIsRecoveryDialogOpen} + canRequestOwnerRecovery={canRequestOwnerRecovery} unarchiveChannelMutation={unarchiveChannelMutation} /> @@ -423,6 +474,8 @@ export function ChannelManagementSheet({ resolvedChannel={resolvedChannel} setActiveView={setActiveView} setIsEditDialogOpen={setIsEditDialogOpen} + setIsRecoveryDialogOpen={setIsRecoveryDialogOpen} + canRequestOwnerRecovery={canRequestOwnerRecovery} unarchiveChannelMutation={unarchiveChannelMutation} /> @@ -565,6 +618,20 @@ export function ChannelManagementSheet({ ) : null} + + + formatMemberName(owner, currentPubkey), + )} + mutation={recoverChannelOwnerMutation} + onOpenChange={(next) => { + recoverChannelOwnerMutation.reset(); + setIsRecoveryDialogOpen(next); + }} + open={isRecoveryDialogOpen} + /> ); } @@ -604,6 +671,8 @@ type ChannelManagementPanelContentProps = { resolvedChannel: Channel; setActiveView: React.Dispatch>; setIsEditDialogOpen: React.Dispatch>; + setIsRecoveryDialogOpen: React.Dispatch>; + canRequestOwnerRecovery: boolean; unarchiveChannelMutation: ChannelMutation; }; @@ -636,6 +705,8 @@ function ChannelManagementPanelContent({ resolvedChannel, setActiveView, setIsEditDialogOpen, + setIsRecoveryDialogOpen, + canRequestOwnerRecovery, unarchiveChannelMutation, }: ChannelManagementPanelContentProps) { const scrollRef = React.useRef(null); @@ -755,6 +826,14 @@ function ChannelManagementPanelContent({ testId="channel-management-edit" /> ) : null} + {canRequestOwnerRecovery ? ( + setIsRecoveryDialogOpen(true)} + testId="channel-owner-recovery-open" + /> + ) : null} {joinChannelMutation.error instanceof Error ? ( diff --git a/desktop/src/features/channels/ui/ChannelOwnerRecoveryDialog.tsx b/desktop/src/features/channels/ui/ChannelOwnerRecoveryDialog.tsx new file mode 100644 index 00000000000..95680912750 --- /dev/null +++ b/desktop/src/features/channels/ui/ChannelOwnerRecoveryDialog.tsx @@ -0,0 +1,223 @@ +import * as React from "react"; +import { toast } from "sonner"; + +import type { ChannelMember } from "@/shared/api/types"; +import { Button } from "@/shared/ui/button"; +import { + Dialog, + DialogContent, + DialogHeader, + DialogTitle, +} from "@/shared/ui/dialog"; +import { Textarea } from "@/shared/ui/textarea"; + +type RecoveryMutation = { + error: unknown; + isPending: boolean; + mutateAsync: (input: { + reason: string; + targetPubkey: string; + }) => Promise; + reset: () => void; +}; + +export function ChannelOwnerRecoveryDialog({ + candidates, + channelName, + currentOwners, + mutation, + onOpenChange, + open, +}: { + candidates: ChannelMember[]; + channelName: string; + currentOwners: string[]; + mutation: RecoveryMutation; + onOpenChange: (open: boolean) => void; + open: boolean; +}) { + const [targetPubkey, setTargetPubkey] = React.useState(""); + const [reason, setReason] = React.useState(""); + const [confirmed, setConfirmed] = React.useState(false); + + React.useEffect(() => { + if (!open) { + setTargetPubkey(""); + setReason(""); + setConfirmed(false); + } else if (!targetPubkey && candidates[0]) { + setTargetPubkey(candidates[0].pubkey); + } + }, [candidates, open, targetPubkey]); + + const trimmedReason = reason.trim(); + const reasonByteLength = new TextEncoder().encode(trimmedReason).length; + const canSubmit = + targetPubkey.length === 64 && + reasonByteLength > 0 && + reasonByteLength <= 500 && + confirmed && + !mutation.isPending; + const selectedTarget = candidates.find( + (candidate) => candidate.pubkey === targetPubkey, + ); + const selectedTargetName = + selectedTarget?.displayName?.trim() || selectedTarget?.pubkey || "None"; + const ownerSummary = + currentOwners.length > 0 ? currentOwners.join(", ") : "None listed"; + + async function submit() { + if (!canSubmit) { + return; + } + try { + await mutation.mutateAsync({ + targetPubkey, + reason: trimmedReason, + }); + toast.success("Channel owner recovered"); + onOpenChange(false); + } catch { + // Preserve the relay's authoritative denial in the inline error below. + } + } + + return ( + + + + Recover channel owner + + +
+

+ Every current human owner must have self-archived and named this + target as their replacement. Recovery is denied while an active + human admin or any owner/admin agent exists. A lost or deleted key + without that durable consent is not recoverable. +

+ +

+ This only promotes the selected member. It does not remove or demote + anyone, and the channel ID, messages, threads, roster, canvas, and + workflows remain unchanged. +

+ +
+
Channel
+
#{channelName}
+
Current owners
+
+ {ownerSummary} +
+
Replacement
+
+ {selectedTargetName} +
+
Audit reason
+
+ {trimmedReason || "Enter a reason below"} +
+
+ + + +