diff --git a/.env.example b/.env.example index db5a7ea25c..8619d77144 100644 --- a/.env.example +++ b/.env.example @@ -102,6 +102,10 @@ RELAY_URL=ws://localhost:3000 # How often the reaper checks for expired ephemeral channels (default: 60s). # BUZZ_REAPER_INTERVAL_SECS=5 +# Audited channel-owner recovery outbox delivery. +# BUZZ_RECOVERY_AUDIT_INTERVAL_SECS=10 +# BUZZ_RECOVERY_AUDIT_BATCH_LIMIT=100 + # ----------------------------------------------------------------------------- # Logging / Tracing # ----------------------------------------------------------------------------- diff --git a/README.md b/README.md index 72af92ce13..e4af2df980 100644 --- a/README.md +++ b/README.md @@ -222,6 +222,7 @@ A Rust workspace of focused crates. Single source of truth: the relay. See [ARCH - **[VISION.md](VISION.md)** · **[VISION_SOVEREIGN.md](VISION_SOVEREIGN.md)** · **[VISION_PROJECTS.md](VISION_PROJECTS.md)** · **[VISION_AGENT.md](VISION_AGENT.md)** — the four vision docs - **[ARCHITECTURE.md](ARCHITECTURE.md)** — system design, kind ranges, subsystem boundaries +- **[Channel-owner recovery](docs/channel-owner-recovery.md)** — exceptional recovery policy, event contract, and relay operations - **[TESTING.md](TESTING.md)** — multi-agent E2E test suite - **[CONTRIBUTING.md](CONTRIBUTING.md)** · **[CODE_OF_CONDUCT.md](CODE_OF_CONDUCT.md)** · **[SECURITY.md](SECURITY.md)** · **[GOVERNANCE.md](GOVERNANCE.md)** diff --git a/crates/buzz-cli/README.md b/crates/buzz-cli/README.md index a8c668cf06..3d6c50c086 100644 --- a/crates/buzz-cli/README.md +++ b/crates/buzz-cli/README.md @@ -47,6 +47,7 @@ buzz channels list buzz channels create --name "my-channel" --type stream --visibility open buzz channels join --channel buzz channels topic --channel --topic "New topic" +buzz channels recover-owner --channel --pubkey --reason "Prior consent recorded" # Reactions buzz reactions add --event --emoji "👍" @@ -95,6 +96,12 @@ buzz channels list | jq '.[].name' constraint omitted from the command is removed. `protect list` reports malformed stored rules in `validation_error` so an owner can remove and repair them. +`channels recover-owner` is an exceptional, promotion-only command. It requires +the target to be a current human member of the same community and channel, and +requires every current human owner to have self-archived with that exact target +as `replaced-by`. It does not recover a lost/deleted key without that prior +evidence. See [Channel-owner recovery](../../docs/channel-owner-recovery.md). + ## Commands | Group | Subcommand | Description | @@ -121,6 +128,7 @@ stored rules in `validation_error` so an owner can remove and repair them. | | `members` | List channel members | | | `add-member` | Add a member | | | `remove-member` | Remove a member | +| | `recover-owner` | Promote a prior-consented replacement through the audited recovery path | | `canvas` | `get` | Get channel canvas | | | `set` | Set channel canvas | | `reactions` | `add` | React to a message | diff --git a/crates/buzz-cli/src/commands/channels.rs b/crates/buzz-cli/src/commands/channels.rs index 42844bf1e0..3c1e94fbfb 100644 --- a/crates/buzz-cli/src/commands/channels.rs +++ b/crates/buzz-cli/src/commands/channels.rs @@ -1001,6 +1001,23 @@ pub async fn cmd_remove_channel_member( Ok(()) } +/// Submit the dedicated, prior-self-consent-only channel-owner recovery event. +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 +1178,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 6ab81a082d..70c2478f1e 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 afec52305a..bca7dd9894 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -348,6 +348,12 @@ 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; +/// Marker on relay-signed, non-deletable channel-owner recovery audit events. +pub const CHANNEL_OWNER_RECOVERY_AUDIT_MARKER: &str = "channel-owner-recovery-v1"; // NIP-IA identity archival announcement events (relay-signed) /// NIP-IA: Archived-identity delta (relay-signed). @@ -612,6 +618,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 941c0fc735..c6a61c4e56 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 396244cc96..5cf8b5068b 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 0000000000..2cf04123ef --- /dev/null +++ b/crates/buzz-db/src/channel_owner_recovery.rs @@ -0,0 +1,2060 @@ +//! 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, StoredEvent}; +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(), + )); + } + + // Lock the actor and target community memberships in stable pubkey order. + // Community removal deletes relay_members without touching channel_members, + // so the target's current relay row is the authoritative same-community + // admission check. + let relay_member_rows = sqlx::query( + "SELECT pubkey, role FROM relay_members \ + WHERE community_id = $1 AND pubkey IN ($2, $3) \ + ORDER BY pubkey FOR UPDATE", + ) + .bind(community_id.as_uuid()) + .bind(&actor_hex) + .bind(&target_hex) + .fetch_all(&mut *tx) + .await?; + let mut actor_role = None; + let mut target_is_community_member = false; + for row in relay_member_rows { + let pubkey: String = row.try_get("pubkey")?; + if pubkey == actor_hex { + actor_role = Some(row.try_get::("role")?); + } + if pubkey == target_hex { + target_is_community_member = true; + } + } + let actor_role = actor_role + .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(), + )); + } + if !target_is_community_member { + return Err(DbError::AccessDenied( + "target must be a current member of the same community".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, + }) +} + +/// Atomically store a relay-signed audit event and link it to its immutable recovery record. +/// +/// The exact event-ID linkage prevents historical or mixed-version clients +/// from forging marker-shaped events that later become undeletable. +pub async fn store_recovery_audit_event( + pool: &PgPool, + community_id: CommunityId, + request_event_id: &[u8], + event: &Event, + channel_id: Uuid, +) -> Result<(StoredEvent, bool)> { + let event_id = event.id.as_bytes(); + let event_pubkey = event.pubkey.to_bytes(); + let event_sig = event.sig.serialize(); + let event_tags = serde_json::to_value(&event.tags)?; + let event_created_at_secs = event.created_at.as_secs() as i64; + let event_created_at = DateTime::from_timestamp(event_created_at_secs, 0) + .ok_or(DbError::InvalidTimestamp(event_created_at_secs))?; + let received_at = Utc::now(); + let mut tx = pool.begin().await?; + + 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,$9,$10) \ + ON CONFLICT DO NOTHING", + ) + .bind(community_id.as_uuid()) + .bind(event_id.as_slice()) + .bind(event_pubkey.as_slice()) + .bind(event_created_at) + .bind(i32::from(event.kind.as_u16())) + .bind(event_tags) + .bind(&event.content) + .bind(event_sig.as_slice()) + .bind(received_at) + .bind(channel_id) + .execute(&mut *tx) + .await? + .rows_affected() + == 1; + + let linked = sqlx::query( + "UPDATE channel_owner_recovery_outbox \ + SET audit_event_id = $3, delivered_at = NOW(), \ + attempts = attempts + 1, last_error = NULL \ + WHERE community_id = $1 AND request_event_id = $2 \ + AND channel_id = $4 \ + AND (audit_event_id IS NULL OR audit_event_id = $3)", + ) + .bind(community_id.as_uuid()) + .bind(request_event_id) + .bind(event_id.as_slice()) + .bind(channel_id) + .execute(&mut *tx) + .await? + .rows_affected(); + if linked != 1 { + return Err(DbError::InvalidData( + "recovery outbox is missing or linked to a different audit event".into(), + )); + } + + tx.commit().await?; + if inserted { + if let Err(error) = + crate::insert_mentions(pool, community_id, event, Some(channel_id)).await + { + tracing::warn!( + event_id = %event.id, + error = %error, + "Failed to insert recovery audit mentions" + ); + } + } + Ok(( + StoredEvent::with_received_at(event.clone(), received_at, Some(channel_id), true), + inserted, + )) +} + +/// Return whether an event ID is durably linked to an immutable recovery audit. +pub async fn is_recovery_audit_event( + pool: &PgPool, + community_id: CommunityId, + event_id: &[u8], +) -> Result { + Ok(sqlx::query_scalar( + "SELECT EXISTS ( \ + SELECT 1 FROM channel_owner_recovery_outbox \ + WHERE community_id = $1 AND audit_event_id = $2 \ + )", + ) + .bind(community_id.as_uuid()) + .bind(event_id) + .fetch_one(pool) + .await?) +} + +/// 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::{ + CHANNEL_OWNER_RECOVERY_AUDIT_MARKER, KIND_CHANNEL_OWNER_RECOVERY, KIND_SYSTEM_MESSAGE, + }; + use nostr::{EventBuilder, Keys, Kind, Tag}; + + 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 { + let database_url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".into()); + PgPool::connect(&database_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; + } + for (keys, role) in [(&actor, "owner"), (&owner, "member"), (&target, "member")] { + sqlx::query("INSERT INTO relay_members (community_id,pubkey,role) VALUES ($1,$2,$3)") + .bind(community.as_uuid()) + .bind(keys.public_key().to_hex()) + .bind(role) + .execute(pool) + .await + .expect("insert community member"); + } + + 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 linked_audit_rejects_legacy_unconditional_soft_delete() { + let pool = setup_pool().await; + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + let request = request(&fixture, "legacy delete guard"); + let recovery = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "legacy delete guard", + &request, + ) + .await + .expect("recover channel owner"); + let channel_id = fixture.channel.to_string(); + let audit = EventBuilder::new( + Kind::Custom(KIND_SYSTEM_MESSAGE as u16), + serde_json::to_string(&recovery.payload).expect("encode audit payload"), + ) + .tags([ + Tag::parse(["h", &channel_id]).unwrap(), + Tag::parse(["audit", CHANNEL_OWNER_RECOVERY_AUDIT_MARKER]).unwrap(), + ]) + .sign_with_keys(&Keys::generate()) + .expect("sign recovery audit"); + store_recovery_audit_event( + &pool, + fixture.community, + request.id.as_bytes().as_slice(), + &audit, + fixture.channel, + ) + .await + .expect("store and link recovery audit"); + + let db = crate::Db::from_pool(pool.clone()); + let error = db + .soft_delete_event_and_update_thread( + fixture.community, + audit.id.as_bytes().as_slice(), + None, + None, + ) + .await + .expect_err("legacy unconditional deletion must fail closed"); + assert!( + error + .to_string() + .contains("channel owner recovery audit events are immutable"), + "unexpected deletion error: {error}" + ); + + let hard_delete = sqlx::query("DELETE FROM events WHERE community_id=$1 AND id=$2") + .bind(fixture.community.as_uuid()) + .bind(audit.id.as_bytes().as_slice()) + .execute(&pool) + .await + .expect_err("hard deletion must also fail closed"); + assert!( + hard_delete + .to_string() + .contains("channel owner recovery audit events are immutable"), + "unexpected hard-deletion error: {hard_delete}" + ); + + let deleted_at: Option> = + sqlx::query_scalar("SELECT deleted_at FROM events WHERE community_id=$1 AND id=$2") + .bind(fixture.community.as_uuid()) + .bind(audit.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("read linked audit event"); + assert!(deleted_at.is_none(), "linked audit must remain visible"); + } + + #[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 former_community_member_with_stale_channel_row_cannot_be_promoted() { + let pool = setup_pool().await; + let fixture = setup_fixture(&pool).await; + record_owner_consent(&pool, &fixture).await; + + let removed = crate::relay_members::remove_relay_member( + &pool, + fixture.community, + &fixture.target.public_key().to_hex(), + ) + .await + .expect("remove target from community"); + assert_eq!(removed, crate::relay_members::RemoveResult::Removed); + + let stale_channel_row: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 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(fixture.target.public_key().to_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("read stale channel membership"); + assert!( + stale_channel_row, + "community removal must leave the regression's channel row intact" + ); + + let denied = recover_channel_owner( + &pool, + fixture.community, + fixture.channel, + fixture.target.public_key().to_bytes().as_slice(), + "former member must not be promoted", + &request(&fixture, "former member must not be promoted"), + ) + .await; + assert!(matches!( + denied, + Err(DbError::AccessDenied(message)) + if message == "target must be a current member of the same community" + )); + assert_no_recovery_side_effects(&pool, &fixture).await; + } + + #[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(_)))); + 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("bot target role"); + assert_eq!(target_role, "bot"); + 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("bot target audit count"); + assert_eq!(audit_count, 0); + + 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::AccessDenied(message)) + if message == "target must be a current member of the same community" + )); + 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 9c63b2e8ab..1c77b89654 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,77 @@ 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 + } + + /// Atomically store and durably link a relay-signed recovery audit event. + pub async fn store_recovery_audit_event( + &self, + community_id: CommunityId, + request_event_id: &[u8], + event: &nostr::Event, + channel_id: Uuid, + ) -> Result<(StoredEvent, bool)> { + channel_owner_recovery::store_recovery_audit_event( + &self.pool, + community_id, + request_event_id, + event, + channel_id, + ) + .await + } + + /// Return whether an event is durably linked to an immutable recovery audit. + pub async fn is_recovery_audit_event( + &self, + community_id: CommunityId, + event_id: &[u8], + ) -> Result { + channel_owner_recovery::is_recovery_audit_event(&self.pool, community_id, 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 1674b0ec4d..18cf7a5c04 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,49 @@ 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("audit_event_id")); + assert!(owner_recovery.contains("uq_channel_owner_recovery_audit_event")); + assert!(owner_recovery.contains("reason_code")); + assert!(owner_recovery.contains("BEFORE UPDATE OR DELETE")); + assert!(owner_recovery.contains("CREATE FUNCTION channel_owner_recovery_event_immutable")); + assert!(owner_recovery + .contains("CREATE TRIGGER trg_events_channel_owner_recovery_soft_delete_guard")); + assert!(owner_recovery.contains("BEFORE UPDATE OF deleted_at ON events")); + assert!(owner_recovery + .contains("CREATE TRIGGER trg_events_channel_owner_recovery_hard_delete_guard")); + assert!(owner_recovery.contains("BEFORE DELETE ON events")); + assert!(!migrations[0] + .sql + .as_str() + .contains("channel_owner_recovery_audit")); + } + + #[test] + fn desired_state_schema_includes_channel_owner_recovery_state() { + let schema = include_str!("../../../schema/schema.sql"); + assert!(schema.contains("CREATE TABLE channel_owner_recovery_audit")); + assert!(schema.contains("CREATE TABLE channel_owner_recovery_outbox")); + assert!(schema.contains("audit_event_id")); + assert!(schema.contains("uq_channel_owner_recovery_audit_event")); + assert!(schema.contains("CREATE TRIGGER trg_channel_owner_recovery_audit_immutable")); + assert!(schema.contains("BEFORE UPDATE OR DELETE ON channel_owner_recovery_audit")); + assert!(schema.contains("CREATE FUNCTION channel_owner_recovery_event_immutable")); + assert!( + schema.contains("CREATE TRIGGER trg_events_channel_owner_recovery_soft_delete_guard") + ); + assert!(schema.contains("BEFORE UPDATE OF deleted_at ON events")); + assert!( + schema.contains("CREATE TRIGGER trg_events_channel_owner_recovery_hard_delete_guard") + ); + assert!(schema.contains("BEFORE DELETE ON events")); } #[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 0000000000..d7812c2be1 --- /dev/null +++ b/crates/buzz-relay/src/handlers/channel_owner_recovery.rs @@ -0,0 +1,568 @@ +//! 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::{CHANNEL_OWNER_RECOVERY_AUDIT_MARKER, 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 + .store_recovery_audit_event( + tenant.community(), + request_event_id, + &audit, + 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; + } + 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])?, + Tag::parse(["audit", CHANNEL_OWNER_RECOVERY_AUDIT_MARKER])?, + ]) + .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], + vec!["audit".into(), CHANNEL_OWNER_RECOVERY_AUDIT_MARKER.into(),], + ] + ); + 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"); + } + for (keys, role) in [(&actor, "owner"), (&owner, "member"), (&target, "member")] { + sqlx::query("INSERT INTO relay_members (community_id,pubkey,role) VALUES ($1,$2,$3)") + .bind(community.as_uuid()) + .bind(keys.public_key().to_hex()) + .bind(role) + .execute(&pool) + .await + .expect("insert community member"); + } + 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, audit_event_id): (Option>, Option>) = sqlx::query_as( + "SELECT delivered_at, audit_event_id 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"); + assert_eq!( + audit_event_id.as_deref(), + Some(audit.event.id.as_bytes().as_slice()) + ); + assert!(state + .db + .is_recovery_audit_event(community, audit.event.id.as_bytes().as_slice()) + .await + .expect("read durable audit link")); + } +} diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 88dd5f5180..c15bc0c9b6 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 497adc6c73..f089fa2c9f 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -14,26 +14,26 @@ 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, - RELAY_ADMIN_REMOVE_MEMBER, RELAY_ADMIN_SET_WORKSPACE_PROFILE, + KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, KIND_SYSTEM_MESSAGE, KIND_TEAM, KIND_TEXT_NOTE, + KIND_USER_STATUS, KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, RELAY_ADMIN_ADD_MEMBER, + RELAY_ADMIN_CHANGE_ROLE, RELAY_ADMIN_REMOVE_MEMBER, RELAY_ADMIN_SET_WORKSPACE_PROFILE, }; use buzz_core::tenant::TenantContext; use buzz_core::verification::verify_event; @@ -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 @@ -1481,6 +1483,17 @@ async fn ingest_event_inner( if buzz_core::kind::is_relay_only_kind(kind_u32) { return Err(IngestError::Rejected("restricted: relay-only kind".into())); } + // Kind 40099 also carries legitimate user-submitted system-style messages, + // so it cannot be relay-only wholesale. Reserve only the recovery-audit + // marker: otherwise a member could forge an event that deletion handlers + // must preserve forever. + if kind_u32 == KIND_SYSTEM_MESSAGE + && super::side_effects::has_channel_owner_recovery_audit_marker(&event) + { + return Err(IngestError::Rejected( + "restricted: channel owner recovery audit marker is relay-only".into(), + )); + } // Share the event with the verify task via Arc instead of deep-cloning it // (tags + up to 256 KB of content). spawn_blocking only needs 'static, not @@ -1798,7 +1811,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 +1844,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 98a5e6c51d..87cb238bcd 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/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 3112e9a559..5678cd1914 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -216,6 +216,12 @@ pub async fn validate_standard_deletion_event( .await? .ok_or_else(|| anyhow::anyhow!("target event not found"))?; + if is_immutable_channel_owner_recovery_audit(tenant, state, &target_event.event).await? { + return Err(anyhow::anyhow!( + "channel owner recovery audit events are immutable" + )); + } + let target_author = effective_message_author(&target_event.event, &state.relay_keypair.public_key()); if target_author != actor_bytes @@ -585,6 +591,12 @@ pub async fn validate_admin_event( } _ => {} // Same channel — OK } + if is_immutable_channel_owner_recovery_audit(tenant, state, &target_event.event).await? + { + return Err(anyhow::anyhow!( + "channel owner recovery audit events are immutable" + )); + } // Check if actor is the event author. // For relay-signed REST messages, the real author is in the p tag. @@ -1605,6 +1617,11 @@ async fn handle_delete_event_side_effect( } _ => {} // Same channel — OK } + if is_immutable_channel_owner_recovery_audit(tenant, state, &target_event.event).await? { + return Err(anyhow::anyhow!( + "channel owner recovery audit events are immutable" + )); + } } // Look up thread metadata so we can pass parent/root IDs to the @@ -1657,6 +1674,39 @@ async fn handle_delete_event_side_effect( Ok(()) } +pub(crate) fn has_channel_owner_recovery_audit_marker(event: &Event) -> bool { + event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.first().map(String::as_str) == Some("audit") + && parts.get(1).map(String::as_str) + == Some(buzz_core::kind::CHANNEL_OWNER_RECOVERY_AUDIT_MARKER) + }) +} + +fn has_channel_owner_recovery_audit_shape(event: &Event) -> bool { + event.kind.as_u16() as u32 == buzz_core::kind::KIND_SYSTEM_MESSAGE + && has_channel_owner_recovery_audit_marker(event) + && serde_json::from_str::(&event.content).is_ok_and(|value| { + value.get("type").and_then(serde_json::Value::as_str) == Some("channel_owner_recovered") + }) +} + +async fn is_immutable_channel_owner_recovery_audit( + tenant: &TenantContext, + state: &Arc, + event: &Event, +) -> anyhow::Result { + if !has_channel_owner_recovery_audit_shape(event) { + return Ok(false); + } + // Exact durable linkage is the trust boundary. Shape alone is insufficient: + // an older relay version may have accepted a client-authored lookalike. + Ok(state + .db + .is_recovery_audit_event(tenant.community(), event.id.as_bytes().as_slice()) + .await?) +} + async fn handle_create_group( tenant: &TenantContext, event: &Event, @@ -2127,6 +2177,13 @@ async fn handle_standard_deletion_event( Some(target) => target, None => continue, }; + // Defense in depth: validation runs before storage, but the mutation + // path must preserve immutable audits even if it is called directly. + if is_immutable_channel_owner_recovery_audit(tenant, state, &target_event.event).await? { + return Err(anyhow::anyhow!( + "channel owner recovery audit events are immutable" + )); + } if u32::from(target_event.event.kind.as_u16()) == super::push_lease::KIND_PUSH_LEASE { tracing::debug!( target_id = %hex::encode(&target_id), @@ -3267,6 +3324,43 @@ fn topic_for_subscription(channel_id: Option) -> EventTopic { mod tests { use super::*; + #[test] + fn recovery_audit_shape_does_not_depend_on_current_relay_key() { + let old_relay = nostr::Keys::generate(); + let current_relay = nostr::Keys::generate(); + let content = serde_json::json!({ + "schema_version": 1, + "type": "channel_owner_recovered", + }) + .to_string(); + let marked = |keys: &nostr::Keys| { + EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_SYSTEM_MESSAGE as u16), + &content, + ) + .tags([Tag::parse([ + "audit", + buzz_core::kind::CHANNEL_OWNER_RECOVERY_AUDIT_MARKER, + ]) + .unwrap()]) + .sign_with_keys(keys) + .unwrap() + }; + + assert!(has_channel_owner_recovery_audit_shape(&marked(&old_relay))); + assert!(has_channel_owner_recovery_audit_shape(&marked( + ¤t_relay + ))); + + let unmarked = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_SYSTEM_MESSAGE as u16), + content, + ) + .sign_with_keys(¤t_relay) + .unwrap(); + assert!(!has_channel_owner_recovery_audit_shape(&unmarked)); + } + #[test] fn delete_tombstone_omits_absent_moderation_metadata() { let content = diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 22101219eb..efdb0f0fbe 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -680,6 +680,50 @@ 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!( + error = %error, + "Channel-owner recovery audit worker tick failed" + ); + } + } + 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 f9e54de9c5..c761763234 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/crates/buzz-test-client/tests/e2e_relay.rs b/crates/buzz-test-client/tests/e2e_relay.rs index 7a7f7e19fa..5b68dbb5da 100644 --- a/crates/buzz-test-client/tests/e2e_relay.rs +++ b/crates/buzz-test-client/tests/e2e_relay.rs @@ -43,6 +43,14 @@ fn relay_http_url() -> String { .to_string() } +fn relay_host() -> String { + relay_url() + .trim_start_matches("wss://") + .trim_start_matches("ws://") + .trim_end_matches('/') + .to_string() +} + fn test_owner_keys() -> Keys { std::env::var("BUZZ_TEST_OWNER_PRIVATE_KEY") .ok() @@ -119,6 +127,30 @@ async fn seed_relay_member(host: &str, keys: &Keys, role: &str) { .unwrap_or_else(|e| panic!("seed relay member {role}: {e}")); } +async fn seed_preexisting_channel_event(host: &str, channel_id: Uuid, event: &nostr::Event) { + let pool = e2e_db_pool().await; + let community_id = ensure_test_community(host).await; + let created_at = chrono::DateTime::from_timestamp(event.created_at.as_secs() as i64, 0) + .expect("valid forged-event timestamp"); + 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)", + ) + .bind(community_id) + .bind(event.id.as_bytes().as_slice()) + .bind(event.pubkey.to_bytes().as_slice()) + .bind(created_at) + .bind(i32::from(event.kind.as_u16())) + .bind(serde_json::to_value(&event.tags).expect("serialize forged-event tags")) + .bind(&event.content) + .bind(event.sig.serialize().as_slice()) + .bind(channel_id) + .execute(&pool) + .await + .expect("seed pre-upgrade marked event"); +} + async fn seed_relay_owner(keys: &Keys) { seed_relay_member("localhost:3000", keys, "owner").await; } @@ -201,6 +233,76 @@ async fn create_test_channel(keys: &Keys) -> String { channel_uuid.to_string() } +async fn post_test_event(keys: &Keys, event: &nostr::Event) -> (reqwest::StatusCode, String) { + let url = format!("{}/events", relay_http_url()); + let body = serde_json::to_string(event).expect("serialize event"); + let response = reqwest::Client::new() + .post(&url) + .header("Authorization", nip98_post_header(keys, &url, &body)) + .header("Content-Type", "application/json") + .body(body) + .send() + .await + .expect("submit test event"); + let status = response.status(); + let body = response.text().await.expect("read event response"); + (status, body) +} + +async fn create_authenticated_test_channel(keys: &Keys) -> String { + let channel_uuid = Uuid::new_v4(); + let event = EventBuilder::new(Kind::Custom(9007), "") + .tags([ + Tag::parse(["h", &channel_uuid.to_string()]).expect("channel h tag"), + Tag::parse(["name", &format!("relay-e2e-auth-{channel_uuid}")]) + .expect("channel name tag"), + Tag::parse(["channel_type", "stream"]).expect("channel type tag"), + Tag::parse(["visibility", "open"]).expect("channel visibility tag"), + ]) + .sign_with_keys(keys) + .expect("sign authenticated create-channel event"); + let (status, body) = post_test_event(keys, &event).await; + assert_event_accepted(status, &body); + channel_uuid.to_string() +} + +fn assert_event_accepted(status: reqwest::StatusCode, body: &str) { + assert!( + status.is_success(), + "event submission failed: status={status}, body={body}" + ); + let json: serde_json::Value = serde_json::from_str(body).expect("parse event response"); + assert!( + json["accepted"].as_bool().unwrap_or(false), + "event was not accepted: {body}" + ); +} + +async fn query_channel_events(keys: &Keys, channel_id: &str, kind: u32) -> serde_json::Value { + let filters = serde_json::json!([{ + "kinds": [kind], + "#h": [channel_id], + "limit": 50, + }]); + let url = format!("{}/query", relay_http_url()); + let body = filters.to_string(); + let response = reqwest::Client::new() + .post(&url) + .header("Authorization", nip98_post_header(keys, &url, &body)) + .header("Content-Type", "application/json") + .body(body) + .send() + .await + .expect("query channel events"); + let status = response.status(); + let body = response.text().await.expect("read query response"); + assert!( + status.is_success(), + "channel query failed: status={status}, body={body}" + ); + serde_json::from_str(&body).expect("parse channel query") +} + #[tokio::test] #[ignore] async fn test_connect_and_authenticate() { @@ -214,6 +316,180 @@ async fn test_connect_and_authenticate() { client.disconnect().await.expect("clean disconnect"); } +#[tokio::test] +#[ignore] +async fn test_owner_recovery_public_ingest_preserves_immutable_visible_audit() { + let actor = Keys::generate(); + let prior_owner = Keys::generate(); + let target = Keys::generate(); + let host = relay_host(); + for (keys, role) in [ + (&actor, "owner"), + (&prior_owner, "member"), + (&target, "member"), + ] { + seed_relay_member(&host, keys, role).await; + let profile = EventBuilder::new( + Kind::Custom(0), + serde_json::json!({ + "display_name": format!("recovery-{}", keys.public_key().to_hex()) + }) + .to_string(), + ) + .sign_with_keys(keys) + .expect("sign profile"); + let (status, body) = post_test_event(keys, &profile).await; + assert_event_accepted(status, &body); + } + + let channel_id = create_authenticated_test_channel(&prior_owner).await; + let channel_uuid = Uuid::parse_str(&channel_id).expect("channel UUID"); + let target_hex = target.public_key().to_hex(); + let add_target = buzz_sdk::build_add_member( + channel_uuid, + &target_hex, + Some(buzz_core::channel::MemberRole::Member), + ) + .expect("build add-member") + .sign_with_keys(&prior_owner) + .expect("sign add-member"); + let (status, body) = post_test_event(&prior_owner, &add_target).await; + assert_event_accepted(status, &body); + + let forged_audit = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_SYSTEM_MESSAGE as u16), + serde_json::json!({"type": "channel_owner_recovered"}).to_string(), + ) + .tags([ + Tag::parse(["h", &channel_id]).expect("forged audit h tag"), + Tag::parse([ + "audit", + buzz_core::kind::CHANNEL_OWNER_RECOVERY_AUDIT_MARKER, + ]) + .expect("forged audit marker"), + ]) + .sign_with_keys(&target) + .expect("sign forged recovery audit"); + let (status, body) = post_test_event(&target, &forged_audit).await; + assert_eq!(status, reqwest::StatusCode::BAD_REQUEST); + assert!( + body.contains("channel owner recovery audit marker is relay-only"), + "unexpected forged audit rejection: {body}" + ); + + // Simulate a marker-shaped event accepted by an older pod before the marker + // became reserved. Shape alone must not make this historical forgery + // immutable after a rolling upgrade. + seed_preexisting_channel_event(&host, channel_uuid, &forged_audit).await; + let forged_audit_id = forged_audit.id.to_hex(); + let delete_forgery = EventBuilder::new(Kind::EventDeletion, "") + .tags([Tag::parse(["e", &forged_audit_id]).expect("forgery delete e tag")]) + .sign_with_keys(&target) + .expect("sign historical-forgery deletion"); + let (status, body) = post_test_event(&target, &delete_forgery).await; + assert_event_accepted(status, &body); + let after_forgery_delete = + query_channel_events(&target, &channel_id, buzz_core::kind::KIND_SYSTEM_MESSAGE).await; + assert!( + after_forgery_delete + .as_array() + .expect("channel query array") + .iter() + .all(|event| event["id"].as_str() != Some(forged_audit_id.as_str())), + "unlinked historical forgery must remain deletable" + ); + + let prior_owner_hex = prior_owner.public_key().to_hex(); + let archive = buzz_sdk::build_archive_identity_request( + &prior_owner_hex, + "Prior owner designated a replacement before retiring.", + Some("retired"), + Some(&target_hex), + None, + ) + .expect("build self-archive") + .sign_with_keys(&prior_owner) + .expect("sign self-archive"); + let (status, body) = post_test_event(&prior_owner, &archive).await; + assert_event_accepted(status, &body); + + let recovery = buzz_sdk::build_channel_owner_recovery( + channel_uuid, + &target_hex, + "Public-ingest owner recovery regression", + ) + .expect("build owner recovery") + .sign_with_keys(&actor) + .expect("sign owner recovery"); + let (status, body) = post_test_event(&actor, &recovery).await; + assert_event_accepted(status, &body); + + let visible = + query_channel_events(&target, &channel_id, buzz_core::kind::KIND_SYSTEM_MESSAGE).await; + let audit = visible + .as_array() + .expect("channel query array") + .iter() + .find(|event| { + event["content"].as_str().is_some_and(|content| { + serde_json::from_str::(content).is_ok_and(|content| { + content.get("type").and_then(serde_json::Value::as_str) + == Some("channel_owner_recovered") + }) + }) + }) + .expect("channel-visible owner recovery audit"); + let audit_id = audit["id"].as_str().expect("audit event id"); + assert!( + audit["tags"].as_array().is_some_and(|tags| { + tags.iter().any(|tag| { + tag.as_array().is_some_and(|parts| { + parts.first().and_then(serde_json::Value::as_str) == Some("audit") + && parts.get(1).and_then(serde_json::Value::as_str) + == Some(buzz_core::kind::CHANNEL_OWNER_RECOVERY_AUDIT_MARKER) + }) + }) + }), + "visible audit must carry the immutable relay marker" + ); + + let delete = EventBuilder::new(Kind::Custom(9005), "") + .tags([ + Tag::parse(["h", &channel_id]).expect("delete h tag"), + Tag::parse(["e", audit_id]).expect("delete e tag"), + ]) + .sign_with_keys(&target) + .expect("sign attempted audit deletion"); + let (status, body) = post_test_event(&target, &delete).await; + assert_eq!(status, reqwest::StatusCode::BAD_REQUEST); + assert!( + body.contains("channel owner recovery audit events are immutable"), + "unexpected audit deletion rejection: {body}" + ); + + let standard_delete = EventBuilder::new(Kind::EventDeletion, "") + .tags([Tag::parse(["e", audit_id]).expect("standard delete e tag")]) + .sign_with_keys(&actor) + .expect("sign standard audit deletion attempt"); + let (status, body) = post_test_event(&actor, &standard_delete).await; + assert_eq!(status, reqwest::StatusCode::BAD_REQUEST); + assert!( + body.contains("channel owner recovery audit events are immutable"), + "unexpected standard audit deletion rejection: {body}" + ); + + let after_delete_attempt = + query_channel_events(&target, &channel_id, buzz_core::kind::KIND_SYSTEM_MESSAGE).await; + assert!( + after_delete_attempt + .as_array() + .expect("channel query array") + .iter() + .any(|event| event["id"].as_str() == Some(audit_id)), + "recovery audit must remain channel-visible after deletion attempt" + ); +} + #[tokio::test] #[ignore] async fn test_client_submitted_nip43_membership_snapshots_are_rejected() { diff --git a/desktop/playwright.config.ts b/desktop/playwright.config.ts index e59c68bfaf..b5b4ac88ff 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 59c80c4807..b2cbaeac42 100644 --- a/desktop/src-tauri/src/commands/channels.rs +++ b/desktop/src-tauri/src/commands/channels.rs @@ -848,6 +848,21 @@ pub async fn change_channel_member_role( Ok(()) } +#[tauri::command] +/// Recover channel ownership through the relay's dedicated audited 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 c5b71987ce..1b99e7d391 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 9003d0f5a5..55ddeea32d 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 3c1727e00a..154faffdeb 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 0000000000..9568091275 --- /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"} +
+
+ + + +