From 021ad7c6d7f977163310c818512e461e4091a5ca Mon Sep 17 00:00:00 2001 From: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> Date: Tue, 11 Aug 2026 11:45:14 -0400 Subject: [PATCH] fix: preserve PR1 invariants in local message slice Signed-off-by: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> --- crates/buzz-db/src/lib.rs | 104 +--- crates/buzz-db/src/sqlite.rs | 588 ++++++++++-------- .../src/handlers/command_executor.rs | 73 +-- crates/buzz-relay/src/main.rs | 9 +- 4 files changed, 391 insertions(+), 383 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9b9a2d32ac..e077830afd 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -1396,6 +1396,22 @@ impl Db { .transpose() } + /// Rebinds a stale loopback community to the current single-node authority. + pub async fn rebind_single_node_community_host( + &self, + normalized_host: &str, + owner_pubkey: &str, + ) -> Result> { + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::rebind_single_node_community_host(pool, normalized_host, owner_pubkey).await + } + DbBackend::Postgres => Err(DbError::UnsupportedBackend( + "single-node community rebind on PostgreSQL Db", + )), + } + } + /// Returns the community's workspace icon (NIP-11 `icon`), if set. /// /// Set by relay admins/owners via the kind:9033 command; the value is @@ -2231,19 +2247,6 @@ impl Db { actor_pubkey: &[u8], emoji: &str, ) -> Result { - if let DbBackend::SQLite(pool) = &self.backend { - let _ = thread_meta; - return sqlite::insert_reaction_event( - pool, - community_id, - event, - channel_id, - target_event_id, - actor_pubkey, - emoji, - ) - .await; - } let outcome = event::insert_reaction_event_with_thread_metadata( self.pg_pool()?, community_id, @@ -2855,33 +2858,7 @@ impl Db { pubkeys: &[&[u8]], created_by: &[u8], ) -> Result<(channel::ChannelRecord, bool)> { - match &self.backend { - DbBackend::SQLite(pool) => { - sqlite::open_dm(pool, community_id, pubkeys, created_by, None) - .await? - .ok_or_else(|| DbError::InvalidData("unexpected duplicate DM open".into())) - } - DbBackend::Postgres => { - dm::open_dm(self.pg_pool()?, community_id, pubkeys, created_by).await - } - } - } - - /// Atomically persist a SQLite DM-open command and its channel mutation. - /// `None` means the command event was already processed. - pub async fn open_dm_sqlite_command( - &self, - community_id: CommunityId, - pubkeys: &[&[u8]], - created_by: &[u8], - event: &nostr::Event, - ) -> Result> { - match &self.backend { - DbBackend::SQLite(pool) => { - sqlite::open_dm(pool, community_id, pubkeys, created_by, Some(event)).await - } - DbBackend::Postgres => Ok(None), - } + dm::open_dm(self.pg_pool()?, community_id, pubkeys, created_by).await } /// Hide a DM channel for a specific user. @@ -3318,17 +3295,6 @@ impl Db { pubkey: &[u8], emoji: &str, ) -> Result { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::remove_reaction( - pool, - community, - event_id, - event_created_at, - pubkey, - emoji, - ) - .await; - } reaction::remove_reaction( self.pg_pool()?, community, @@ -3346,10 +3312,6 @@ impl Db { community: CommunityId, reaction_event_id: &[u8], ) -> Result { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::remove_reaction_by_source_event_id(pool, community, reaction_event_id) - .await; - } reaction::remove_reaction_by_source_event_id(self.pg_pool()?, community, reaction_event_id) .await } @@ -3461,17 +3423,6 @@ impl Db { since: Option>, limit: i64, ) -> Result> { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::query_feed_mentions( - pool, - community, - pubkey_bytes, - accessible_channel_ids, - since, - limit, - ) - .await; - } match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { match feed::query_mentions_on( @@ -3549,17 +3500,6 @@ impl Db { since: Option>, limit: i64, ) -> Result> { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::query_feed_needs_action( - pool, - community, - pubkey_bytes, - accessible_channel_ids, - since, - limit, - ) - .await; - } match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { match feed::query_needs_action_on( @@ -3634,16 +3574,6 @@ impl Db { since: Option>, limit: i64, ) -> Result> { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::query_feed_activity( - pool, - community, - accessible_channel_ids, - since, - limit, - ) - .await; - } match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { match feed::query_activity_on( diff --git a/crates/buzz-db/src/sqlite.rs b/crates/buzz-db/src/sqlite.rs index 8134bf04d0..77816efa08 100644 --- a/crates/buzz-db/src/sqlite.rs +++ b/crates/buzz-db/src/sqlite.rs @@ -118,20 +118,6 @@ CREATE INDEX IF NOT EXISTS idx_events_community_kind CREATE INDEX IF NOT EXISTS idx_events_channel_created ON events (community_id, channel_id, created_at DESC); -CREATE TABLE IF NOT EXISTS reactions ( - community_id TEXT NOT NULL, - event_created_at INTEGER NOT NULL, - event_id BLOB NOT NULL, - pubkey BLOB NOT NULL, - emoji TEXT NOT NULL, - reaction_event_id BLOB, - removed_at INTEGER, - PRIMARY KEY (community_id, event_created_at, event_id, pubkey, emoji), - FOREIGN KEY (community_id) REFERENCES communities(id) ON DELETE CASCADE -); -CREATE UNIQUE INDEX IF NOT EXISTS idx_reactions_source_event - ON reactions (community_id, reaction_event_id) - WHERE reaction_event_id IS NOT NULL; "#; pub(crate) async fn connect(path_or_url: &str) -> Result { @@ -265,6 +251,83 @@ pub(crate) async fn is_community_active( Ok(count != 0) } +pub(crate) async fn rebind_single_node_community_host( + pool: &SqlitePool, + normalized_host: &str, + owner_pubkey: &str, +) -> Result> { + let mut tx = pool.begin().await?; + + // Defect-era databases may contain several loopback communities, one per + // ephemeral port. Prefer a community owned by this desktop identity, then + // the oldest community. This keeps ownership stable without depending on + // event persistence, which arrives with PR 2. + let row = sqlx::query( + r#"SELECT c.id, c.host + FROM communities c + LEFT JOIN relay_members rm + ON rm.community_id = c.id + AND lower(rm.pubkey) = lower(?1) + AND rm.role = 'owner' + WHERE c.archived_at IS NULL + AND c.host LIKE '127.0.0.1:%' + ORDER BY CASE WHEN rm.pubkey IS NULL THEN 0 ELSE 1 END DESC, c.created_at ASC, c.id ASC + LIMIT 1"#, + ) + .bind(owner_pubkey) + .fetch_optional(&mut *tx) + .await?; + let Some(row) = row else { + tx.commit().await?; + return Ok(None); + }; + let record = community_record(row)?; + if record.host.eq_ignore_ascii_case(normalized_host) { + tx.commit().await?; + return Ok(Some(record)); + } + + // The OS may reuse a port that belongs to another stale defect-era row. + // Swap that collision to the selected row's old authority transactionally, + // using a temporary unique host to satisfy the case-insensitive constraint. + if let Some(collision_id) = sqlx::query_scalar::<_, String>( + "SELECT id FROM communities WHERE host = ?1 COLLATE NOCASE AND id <> ?2", + ) + .bind(normalized_host) + .bind(record.id.as_uuid().to_string()) + .fetch_optional(&mut *tx) + .await? + { + let temporary_host = format!("rebind-{}.invalid", Uuid::new_v4()); + sqlx::query("UPDATE communities SET host = ?2 WHERE id = ?1") + .bind(&collision_id) + .bind(&temporary_host) + .execute(&mut *tx) + .await?; + sqlx::query("UPDATE communities SET host = ?2 WHERE id = ?1") + .bind(record.id.as_uuid().to_string()) + .bind(normalized_host) + .execute(&mut *tx) + .await?; + sqlx::query("UPDATE communities SET host = ?2 WHERE id = ?1") + .bind(collision_id) + .bind(&record.host) + .execute(&mut *tx) + .await?; + } else { + sqlx::query("UPDATE communities SET host = ?2 WHERE id = ?1") + .bind(record.id.as_uuid().to_string()) + .bind(normalized_host) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + Ok(Some(CommunityRecord { + id: record.id, + host: normalized_host.to_string(), + })) +} + pub(crate) async fn ensure_configured_community( pool: &SqlitePool, normalized_host: &str, @@ -527,231 +590,6 @@ pub(crate) async fn query_events( Ok(events.into_iter().skip(offset).take(limit).collect()) } -pub(crate) async fn query_feed_mentions( - pool: &SqlitePool, - community: CommunityId, - pubkey_bytes: &[u8], - accessible_channel_ids: &[Uuid], - since: Option>, - limit: i64, -) -> Result> { - let mut query = crate::EventQuery::for_community(community); - query.kinds = Some(vec![ - buzz_core::kind::KIND_STREAM_MESSAGE as i32, - buzz_core::kind::KIND_STREAM_MESSAGE_V2 as i32, - buzz_core::kind::KIND_TEXT_NOTE as i32, - buzz_core::kind::KIND_FORUM_POST as i32, - buzz_core::kind::KIND_FORUM_COMMENT as i32, - buzz_core::kind::KIND_GIT_PULL_REQUEST as i32, - buzz_core::kind::KIND_GIT_PR_UPDATE as i32, - buzz_core::kind::KIND_GIT_ISSUE as i32, - buzz_core::kind::KIND_GIT_STATUS_OPEN as i32, - buzz_core::kind::KIND_GIT_STATUS_MERGED as i32, - buzz_core::kind::KIND_GIT_STATUS_CLOSED as i32, - buzz_core::kind::KIND_GIT_STATUS_DRAFT as i32, - ]); - query.p_tag_hex = Some(hex::encode(pubkey_bytes)); - query.channel_ids = Some(accessible_channel_ids.to_vec()); - query.since = since; - query.limit = Some(limit.min(crate::feed::FEED_MAX_LIMIT)); - query_events(pool, &query).await -} - -pub(crate) async fn query_feed_needs_action( - pool: &SqlitePool, - community: CommunityId, - pubkey_bytes: &[u8], - accessible_channel_ids: &[Uuid], - since: Option>, - limit: i64, -) -> Result> { - let mut query = crate::EventQuery::for_community(community); - query.kinds = Some(vec![ - buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED as i32, - buzz_core::kind::KIND_STREAM_REMINDER as i32, - ]); - query.p_tag_hex = Some(hex::encode(pubkey_bytes)); - query.channel_ids = Some(accessible_channel_ids.to_vec()); - query.since = since; - query.limit = Some(limit.min(crate::feed::FEED_MAX_LIMIT)); - query_events(pool, &query).await -} - -pub(crate) async fn query_feed_activity( - pool: &SqlitePool, - community: CommunityId, - accessible_channel_ids: &[Uuid], - since: Option>, - limit: i64, -) -> Result> { - let mut query = crate::EventQuery::for_community(community); - query.kinds = Some(vec![ - buzz_core::kind::KIND_STREAM_MESSAGE as i32, - buzz_core::kind::KIND_STREAM_MESSAGE_V2 as i32, - buzz_core::kind::KIND_FORUM_POST as i32, - buzz_core::kind::KIND_JOB_REQUEST as i32, - buzz_core::kind::KIND_JOB_PROGRESS as i32, - buzz_core::kind::KIND_JOB_RESULT as i32, - ]); - query.channel_ids = Some(accessible_channel_ids.to_vec()); - query.since = since; - query.limit = Some(limit.min(crate::feed::FEED_MAX_LIMIT)); - query_events(pool, &query).await -} - -pub(crate) async fn insert_reaction_event( - pool: &SqlitePool, - community: CommunityId, - reaction_event: &nostr::Event, - channel_id: Option, - target_event_id: &[u8], - actor_pubkey: &[u8], - emoji: &str, -) -> Result { - let mut tx = pool.begin().await?; - let target_created_at: Option = sqlx::query_scalar( - "SELECT created_at FROM events WHERE community_id = ?1 AND id = ?2 LIMIT 1", - ) - .bind(community.as_uuid().to_string()) - .bind(target_event_id) - .fetch_optional(&mut *tx) - .await?; - let Some(target_created_at) = target_created_at else { - return Ok(crate::event::ReactionEventInsertOutcome::TargetMissing); - }; - let changed = sqlx::query("INSERT INTO reactions (community_id, event_created_at, event_id, pubkey, emoji, reaction_event_id) VALUES (?1, ?2, ?3, ?4, ?5, ?6) ON CONFLICT (community_id, event_created_at, event_id, pubkey, emoji) DO UPDATE SET removed_at = NULL, reaction_event_id = excluded.reaction_event_id WHERE reactions.removed_at IS NOT NULL") - .bind(community.as_uuid().to_string()).bind(target_created_at).bind(target_event_id) - .bind(actor_pubkey).bind(emoji).bind(reaction_event.id.as_bytes().as_slice()) - .execute(&mut *tx).await?.rows_affected() != 0; - if !changed { - return Ok(crate::event::ReactionEventInsertOutcome::Duplicate); - } - let received_at = chrono::Utc::now(); - let inserted = sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags_json, content, sig, channel_id, received_at, event_json) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) ON CONFLICT DO NOTHING") - .bind(community.as_uuid().to_string()).bind(reaction_event.id.as_bytes().as_slice()).bind(reaction_event.pubkey.to_bytes().as_slice()) - .bind(reaction_event.created_at.as_secs() as i64).bind(reaction_event.kind.as_u16() as i32).bind(serde_json::to_string(&reaction_event.tags)?) - .bind(&reaction_event.content).bind(reaction_event.sig.serialize().as_slice()).bind(channel_id.map(|id| id.to_string())) - .bind(received_at.timestamp()).bind(serde_json::to_string(reaction_event)?).execute(&mut *tx).await?.rows_affected() == 1; - tx.commit().await?; - Ok(crate::event::ReactionEventInsertOutcome::Inserted { - stored_event: Box::new(buzz_core::StoredEvent::with_received_at( - reaction_event.clone(), - received_at, - channel_id, - true, - )), - was_inserted: inserted, - }) -} - -pub(crate) async fn remove_reaction( - pool: &SqlitePool, - community: CommunityId, - event_id: &[u8], - event_created_at: chrono::DateTime, - pubkey: &[u8], - emoji: &str, -) -> Result { - Ok(sqlx::query("UPDATE reactions SET removed_at = unixepoch() WHERE community_id = ?1 AND event_created_at = ?2 AND event_id = ?3 AND pubkey = ?4 AND emoji = ?5 AND removed_at IS NULL") - .bind(community.as_uuid().to_string()).bind(event_created_at.timestamp()).bind(event_id).bind(pubkey).bind(emoji) - .execute(pool).await?.rows_affected() != 0) -} - -pub(crate) async fn remove_reaction_by_source_event_id( - pool: &SqlitePool, - community: CommunityId, - reaction_event_id: &[u8], -) -> Result { - Ok(sqlx::query("UPDATE reactions SET removed_at = unixepoch() WHERE community_id = ?1 AND reaction_event_id = ?2 AND removed_at IS NULL") - .bind(community.as_uuid().to_string()).bind(reaction_event_id) - .execute(pool).await?.rows_affected() != 0) -} - -pub(crate) async fn open_dm( - pool: &SqlitePool, - community: CommunityId, - pubkeys: &[&[u8]], - created_by: &[u8], - command_event: Option<&nostr::Event>, -) -> Result> { - let mut participants = pubkeys.to_vec(); - if !participants.contains(&created_by) { - participants.push(created_by); - } - participants.sort_unstable(); - participants.dedup(); - if !(2..=9).contains(&participants.len()) { - return Err(crate::DbError::InvalidData( - "DM requires 2-9 unique participants".into(), - )); - } - if participants.iter().any(|pubkey| pubkey.len() != 32) { - return Err(crate::DbError::InvalidData( - "DM participant pubkeys must be 32 bytes".into(), - )); - } - let hash = crate::dm::compute_participant_hash(&participants); - let mut tx = pool.begin().await?; - if let Some(event) = command_event { - let received_at = chrono::Utc::now(); - let inserted = sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags_json, content, sig, channel_id, received_at, event_json) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL, ?9, ?10) ON CONFLICT DO NOTHING") - .bind(community.as_uuid().to_string()).bind(event.id.as_bytes().as_slice()).bind(event.pubkey.to_bytes().as_slice()) - .bind(event.created_at.as_secs() as i64).bind(event.kind.as_u16() as i32).bind(serde_json::to_string(&event.tags)?) - .bind(&event.content).bind(event.sig.serialize().as_slice()).bind(received_at.timestamp()).bind(serde_json::to_string(event)?) - .execute(&mut *tx).await?.rows_affected() != 0; - if !inserted { - return Ok(None); - } - } - let select = "SELECT id, name, channel_type, visibility, description, canvas, created_by, created_at, updated_at, archived_at, deleted_at, nip29_group_id, topic_required, max_members, topic, topic_set_by, topic_set_at, purpose, purpose_set_by, purpose_set_at, ttl_seconds, ttl_deadline FROM channels WHERE community_id = ?1 AND participant_hash = ?2 AND channel_type = 'dm' AND deleted_at IS NULL LIMIT 1"; - if let Some(row) = sqlx::query(select) - .bind(community.as_uuid().to_string()) - .bind(hash.as_slice()) - .fetch_optional(&mut *tx) - .await? - { - sqlx::query("UPDATE channel_members SET hidden_at = NULL WHERE channel_id = ?1 AND pubkey = ?2 AND removed_at IS NULL") - .bind(row.try_get::("id")?) - .bind(created_by) - .execute(&mut *tx) - .await?; - let record = channel_record(row)?; - tx.commit().await?; - return Ok(Some((record, false))); - } - - let id = Uuid::new_v4(); - let name = if participants.len() == 2 { - "DM".to_owned() - } else { - format!("Group DM ({})", participants.len()) - }; - sqlx::query("INSERT INTO channels (id, community_id, name, channel_type, visibility, participant_hash, created_by) VALUES (?1, ?2, ?3, 'dm', 'private', ?4, ?5)") - .bind(id.to_string()) - .bind(community.as_uuid().to_string()) - .bind(name) - .bind(hash.as_slice()) - .bind(created_by) - .execute(&mut *tx) - .await?; - for participant in participants { - sqlx::query("INSERT INTO channel_members (channel_id, pubkey, role, invited_by) VALUES (?1, ?2, 'member', ?3)") - .bind(id.to_string()) - .bind(participant) - .bind(created_by) - .execute(&mut *tx) - .await?; - } - let row = sqlx::query(select) - .bind(community.as_uuid().to_string()) - .bind(hash.as_slice()) - .fetch_one(&mut *tx) - .await?; - let record = channel_record(row)?; - tx.commit().await?; - Ok(Some((record, true))) -} - fn stored_event(row: sqlx::sqlite::SqliteRow) -> Result { let json: String = row.try_get("event_json")?; let event: nostr::Event = serde_json::from_str(&json)?; @@ -839,11 +677,110 @@ pub(crate) async fn add_member( role: crate::channel::MemberRole, invited_by: Option<&[u8]>, ) -> Result { - get_channel(pool, community, channel_id).await?; - sqlx::query("INSERT INTO channel_members (channel_id, pubkey, role, invited_by) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(channel_id, pubkey) DO UPDATE SET role = excluded.role, invited_by = excluded.invited_by, removed_at = NULL") - .bind(channel_id.to_string()).bind(pubkey).bind(role.as_str()).bind(invited_by).execute(pool).await?; - member_record(sqlx::query("SELECT channel_id, pubkey, role, joined_at, invited_by, removed_at FROM channel_members WHERE channel_id = ?1 AND pubkey = ?2") - .bind(channel_id.to_string()).bind(pubkey).fetch_one(pool).await?) + if pubkey.len() != 32 { + return Err(crate::DbError::InvalidData(format!( + "pubkey must be 32 bytes, got {}", + pubkey.len() + ))); + } + + let mut tx = pool.begin().await?; + let channel = sqlx::query( + "SELECT * FROM channels WHERE community_id = ?1 AND id = ?2 AND deleted_at IS NULL", + ) + .bind(community.as_uuid().to_string()) + .bind(channel_id.to_string()) + .fetch_optional(&mut *tx) + .await? + .ok_or(crate::DbError::ChannelNotFound(channel_id))?; + let channel = channel_record(channel)?; + + let effective_role = if channel.visibility == "private" { + let inviter = invited_by.ok_or_else(|| { + crate::DbError::AccessDenied("private channel requires an invite".to_string()) + })?; + let creator_bootstrap = inviter == pubkey && inviter == channel.created_by.as_slice(); + if !creator_bootstrap { + let inviter_role = active_member_role(&mut tx, channel_id, inviter) + .await? + .ok_or_else(|| { + crate::DbError::AccessDenied("inviter is not an active member".to_string()) + })?; + if role.is_elevated() && !is_elevated_role(&inviter_role) { + return Err(crate::DbError::AccessDenied( + "only owners/admins may grant elevated roles".to_string(), + )); + } + } + role + } else if role.is_elevated() { + let granter_role = match invited_by { + Some(inviter) => active_member_role(&mut tx, channel_id, inviter).await?, + None => None, + }; + if !granter_role.is_some_and(|role| is_elevated_role(&role)) { + return Err(crate::DbError::AccessDenied( + "only owners/admins may grant elevated roles".to_string(), + )); + } + role + } else { + role + }; + + if let Some(current_role) = active_member_role(&mut tx, channel_id, pubkey).await? { + if current_role != effective_role.as_str() { + let actor_role = match invited_by { + Some(inviter) => active_member_role(&mut tx, channel_id, inviter).await?, + None => None, + }; + if !actor_role.is_some_and(|role| is_elevated_role(&role)) { + return Err(crate::DbError::AccessDenied( + "only owners/admins may change an active member's role".to_string(), + )); + } + if current_role == "owner" && effective_role != crate::channel::MemberRole::Owner { + let owner_count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM channel_members WHERE channel_id = ?1 AND role = 'owner' AND removed_at IS NULL", + ) + .bind(channel_id.to_string()) + .fetch_one(&mut *tx) + .await?; + if owner_count <= 1 { + return Err(crate::DbError::AccessDenied( + "cannot demote the last owner — transfer ownership first".to_string(), + )); + } + } + } + } + + sqlx::query("INSERT INTO channel_members (channel_id, pubkey, role, invited_by) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(channel_id, pubkey) DO UPDATE SET role = excluded.role, removed_at = NULL") + .bind(channel_id.to_string()).bind(pubkey).bind(effective_role.as_str()).bind(invited_by).execute(&mut *tx).await?; + let row = sqlx::query("SELECT channel_id, pubkey, role, joined_at, invited_by, removed_at FROM channel_members WHERE channel_id = ?1 AND pubkey = ?2") + .bind(channel_id.to_string()).bind(pubkey).fetch_one(&mut *tx).await?; + let record = member_record(row)?; + tx.commit().await?; + Ok(record) +} + +async fn active_member_role( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + channel_id: Uuid, + pubkey: &[u8], +) -> Result> { + sqlx::query_scalar( + "SELECT role FROM channel_members WHERE channel_id = ?1 AND pubkey = ?2 AND removed_at IS NULL", + ) + .bind(channel_id.to_string()) + .bind(pubkey) + .fetch_optional(&mut **tx) + .await + .map_err(Into::into) +} + +fn is_elevated_role(role: &str) -> bool { + matches!(role, "owner" | "admin") } pub(crate) async fn is_member( @@ -1187,6 +1124,169 @@ mod tests { use super::*; use nostr::{EventBuilder, Keys, Kind}; + #[tokio::test] + async fn rebind_prefers_owned_then_oldest_community() { + let pool = connect(":memory:").await.unwrap(); + let first = ensure_configured_community(&pool, "127.0.0.1:4000") + .await + .unwrap(); + let second = ensure_configured_community(&pool, "127.0.0.1:4001") + .await + .unwrap(); + bootstrap_owner(&pool, second.id, "owner").await.unwrap(); + let rebound = rebind_single_node_community_host(&pool, "127.0.0.1:5000", "owner") + .await + .unwrap() + .unwrap(); + assert_eq!(rebound.id, second.id); + assert_eq!( + lookup_community_host(&pool, first.id) + .await + .unwrap() + .unwrap(), + "127.0.0.1:4000" + ); + assert_eq!(rebound.host, "127.0.0.1:5000"); + } + + #[tokio::test] + async fn sqlite_membership_preserves_public_contract_invariants() { + let pool = connect(":memory:").await.unwrap(); + let community = ensure_configured_community(&pool, "local.buzz") + .await + .unwrap(); + let owner = Keys::generate().public_key().to_bytes(); + let member = Keys::generate().public_key().to_bytes(); + let admin = Keys::generate().public_key().to_bytes(); + let other = Keys::generate().public_key().to_bytes(); + let channel_id = Uuid::new_v4(); + create_channel_with_id( + &pool, + community.id, + channel_id, + "private", + crate::channel::ChannelType::Stream, + crate::channel::ChannelVisibility::Private, + None, + &owner, + None, + ) + .await + .unwrap(); + + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &[0; 31], + crate::channel::MemberRole::Member, + Some(&owner), + ) + .await, + Err(crate::DbError::InvalidData(_)) + )); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Member, + None, + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Member, + Some(&owner), + ) + .await + .unwrap(); + add_member( + &pool, + community.id, + channel_id, + &admin, + crate::channel::MemberRole::Admin, + Some(&owner), + ) + .await + .unwrap(); + let readded = add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Member, + Some(&admin), + ) + .await + .unwrap(); + assert_eq!(readded.invited_by, Some(owner.to_vec())); + sqlx::query::( + "UPDATE channel_members SET removed_at = unixepoch() WHERE channel_id = ?1 AND pubkey = ?2", + ) + .bind(channel_id.to_string()) + .bind(&member[..]) + .execute(&pool) + .await + .unwrap(); + let reactivated = add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Member, + Some(&admin), + ) + .await + .unwrap(); + assert_eq!(reactivated.invited_by, Some(owner.to_vec())); + assert!(reactivated.removed_at.is_none()); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &other, + crate::channel::MemberRole::Admin, + Some(&member), + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Guest, + Some(&member), + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &owner, + crate::channel::MemberRole::Member, + Some(&owner), + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + } + #[tokio::test] async fn upgrades_phase_1_core_schema_before_dm_use() { let path = std::env::temp_dir().join(format!("buzz-db-upgrade-{}.sqlite", Uuid::new_v4())); diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 8f0cd0a250..2d82736807 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -80,23 +80,8 @@ pub async fn handle_command( enum PersistResult { /// Event was already processed — return idempotent success. Duplicate, - /// Event inserted — PostgreSQL remains open until mutations finish; SQLite - /// committed the idempotency row atomically before the local mutation. - Inserted(CommandTransaction), -} - -enum CommandTransaction { - Sqlite, - Postgres(sqlx::Transaction<'static, sqlx::Postgres>), -} - -impl CommandTransaction { - async fn commit(self) -> Result<(), sqlx::Error> { - match self { - Self::Sqlite => Ok(()), - Self::Postgres(tx) => tx.commit().await, - } - } + /// Event inserted — transaction is open, handler must commit after mutations. + Inserted(sqlx::Transaction<'static, sqlx::Postgres>), } /// Persist a command event inside a transaction. Returns the OPEN transaction @@ -120,12 +105,6 @@ async fn persist_command_event( ) -> Result { let channel_id = channel_id_override.or_else(|| extract_channel_id(event)); - if state.config.profile.is_single_node() { - return Err(IngestError::Rejected( - "unsupported_feature: this command is unavailable in the single-node profile".into(), - )); - } - let mut tx = state .db .begin_transaction() @@ -248,7 +227,7 @@ async fn persist_command_event( // Duplicate — rollback (implicit on drop) and signal idempotent success. Ok(PersistResult::Duplicate) } else { - Ok(PersistResult::Inserted(CommandTransaction::Postgres(tx))) + Ok(PersistResult::Inserted(tx)) } } @@ -366,35 +345,27 @@ async fn handle_dm_open( } } - // SQLite couples the command idempotency row and DM mutation in one - // transaction. PostgreSQL retains the existing open-transaction path. - let all_refs: Vec<&[u8]> = all_bytes.iter().map(|b| b.as_slice()).collect(); - let sqlite_result = state - .db - .open_dm_sqlite_command(tenant.community(), &all_refs, &self_bytes, event) - .await - .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; - let (channel, was_created, tx) = if let Some(result) = sqlite_result { - (result.0, result.1, CommandTransaction::Sqlite) - } else { - let tx = match persist_command_event(state, tenant, event, None).await? { - PersistResult::Duplicate => { - return Ok(IngestResult { - event_id: event.id.to_hex(), - accepted: true, - message: "duplicate: already processed".into(), - }); - } - PersistResult::Inserted(tx) => tx, - }; - let (channel, was_created) = state - .db - .open_dm(tenant.community(), &all_refs, &self_bytes) - .await - .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; - (channel, was_created, tx) + // Persist the command event (idempotency) — returns open transaction + let tx = match persist_command_event(state, tenant, event, None).await? { + PersistResult::Duplicate => { + return Ok(IngestResult { + event_id: event.id.to_hex(), + accepted: true, + message: "duplicate: already processed".into(), + }); + } + PersistResult::Inserted(tx) => tx, }; + // 4. Execute: open_dm + let all_refs: Vec<&[u8]> = all_bytes.iter().map(|b| b.as_slice()).collect(); + let (channel, was_created) = state + .db + .open_dm(tenant.community(), &all_refs, &self_bytes) + .await + .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; + + // Commit: event + mutation succeeded atomically. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index fb5aacc152..f972caaff4 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1231,7 +1231,14 @@ async fn run_single_node(config: Config, tracer_init: telemetry::TracerInit) -> "Cannot derive community host from BUZZ_RELAY_URL" )); } - let community = db.ensure_configured_community(&host).await?.id; + let community = if let Some(owner) = config.relay_owner_pubkey.as_deref() { + match db.rebind_single_node_community_host(&host, owner).await? { + Some(record) => record.id, + None => db.ensure_configured_community(&host).await?.id, + } + } else { + db.ensure_configured_community(&host).await?.id + }; if let Some(owner) = config.relay_owner_pubkey.as_deref() { db.bootstrap_owner(community, owner).await?; }