From 4766d6bc629223144069fc766e4a189caef0d122 Mon Sep 17 00:00:00 2001 From: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co> Date: Fri, 26 Jun 2026 17:08:38 -0400 Subject: [PATCH] fix(db): scope relay rows by community Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- crates/buzz-db/src/channel.rs | 239 +++-- crates/buzz-db/src/event.rs | 112 ++- crates/buzz-db/src/lib.rs | 297 ++++++- crates/buzz-db/src/migration.rs | 821 +++++++++++++----- crates/buzz-db/src/thread.rs | 77 +- crates/buzz-db/src/user.rs | 156 ++-- crates/buzz-db/src/workflow.rs | 414 ++++++++- .../src/handlers/command_executor.rs | 12 +- 8 files changed, 1745 insertions(+), 383 deletions(-) diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index a9cecc141..ef2a1fa8e 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -9,6 +9,7 @@ use sqlx::{PgPool, Postgres, Row, Transaction}; use uuid::Uuid; use crate::error::{DbError, Result}; +use buzz_core::CommunityId; // Re-export the canonical enum definitions from buzz-core. // These live in core (zero I/O deps) so the SDK can share them @@ -84,6 +85,7 @@ pub struct MemberRecord { /// Creates a new channel, bootstraps the creator as owner, and returns the record. pub async fn create_channel( pool: &PgPool, + community_id: CommunityId, name: &str, channel_type: ChannelType, visibility: ChannelVisibility, @@ -104,12 +106,13 @@ pub async fn create_channel( sqlx::query( r#" - INSERT INTO channels (id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline) - VALUES ($1, $2, $3::channel_type, $4::channel_visibility, $5, $6, $7, - CASE WHEN $7 IS NOT NULL THEN NOW() + ($7 || ' seconds')::interval ELSE NULL END) + INSERT INTO channels (id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline) + VALUES ($1, $2, $3, $4::channel_type, $5::channel_visibility, $6, $7, $8, + CASE WHEN $8 IS NOT NULL THEN NOW() + ($8 || ' seconds')::interval ELSE NULL END) "#, ) .bind(id) + .bind(community_id.as_uuid()) .bind(name) .bind(channel_type.as_str()) .bind(visibility.as_str()) @@ -121,14 +124,15 @@ pub async fn create_channel( sqlx::query( r#" - INSERT INTO channel_members (channel_id, pubkey, role, invited_by) - VALUES ($1, $2, 'owner', $3) - ON CONFLICT (channel_id, pubkey) DO UPDATE SET + INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) + VALUES ($1, $2, $3, 'owner', $4) + ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET removed_at = NULL, removed_by = NULL, role = EXCLUDED.role "#, ) + .bind(community_id.as_uuid()) .bind(id) .bind(created_by) .bind(created_by) @@ -144,9 +148,10 @@ pub async fn create_channel( topic, topic_set_by, topic_set_at, purpose, purpose_set_by, purpose_set_at, ttl_seconds, ttl_deadline - FROM channels WHERE id = $1 + FROM channels WHERE community_id = $1 AND id = $2 "#, ) + .bind(community_id.as_uuid()) .bind(id) .fetch_one(&mut *tx) .await?; @@ -163,6 +168,7 @@ pub async fn create_channel( #[allow(clippy::too_many_arguments)] pub async fn create_channel_with_id( pool: &PgPool, + community_id: CommunityId, channel_id: Uuid, name: &str, channel_type: ChannelType, @@ -188,13 +194,14 @@ pub async fn create_channel_with_id( let rows_affected = sqlx::query( r#" - INSERT INTO channels (id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline) - VALUES ($1, $2, $3::channel_type, $4::channel_visibility, $5, $6, $7, - CASE WHEN $7 IS NOT NULL THEN NOW() + ($7 || ' seconds')::interval ELSE NULL END) - ON CONFLICT (id) DO NOTHING + INSERT INTO channels (id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline) + VALUES ($1, $2, $3, $4::channel_type, $5::channel_visibility, $6, $7, $8, + CASE WHEN $8 IS NOT NULL THEN NOW() + ($8 || ' seconds')::interval ELSE NULL END) + ON CONFLICT (community_id, id) DO NOTHING "#, ) .bind(channel_id) + .bind(community_id.as_uuid()) .bind(name) .bind(channel_type.as_str()) .bind(visibility.as_str()) @@ -211,14 +218,15 @@ pub async fn create_channel_with_id( // Bootstrap the creator as owner. sqlx::query( r#" - INSERT INTO channel_members (channel_id, pubkey, role, invited_by) - VALUES ($1, $2, 'owner', $3) - ON CONFLICT (channel_id, pubkey) DO UPDATE SET + INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) + VALUES ($1, $2, $3, 'owner', $4) + ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET removed_at = NULL, removed_by = NULL, role = EXCLUDED.role "#, ) + .bind(community_id.as_uuid()) .bind(channel_id) .bind(created_by) .bind(created_by) @@ -235,9 +243,10 @@ pub async fn create_channel_with_id( topic, topic_set_by, topic_set_at, purpose, purpose_set_by, purpose_set_at, ttl_seconds, ttl_deadline - FROM channels WHERE id = $1 + FROM channels WHERE community_id = $1 AND id = $2 "#, ) + .bind(community_id.as_uuid()) .bind(channel_id) .fetch_one(&mut *tx) .await?; @@ -305,6 +314,7 @@ pub async fn set_canvas(pool: &PgPool, channel_id: Uuid, canvas: Option<&str>) - /// races (e.g. the inviter being removed between the role check and the INSERT). pub async fn add_member( pool: &PgPool, + community_id: CommunityId, channel_id: Uuid, pubkey: &[u8], role: MemberRole, @@ -330,7 +340,7 @@ pub async fn add_member( let is_creator_bootstrap = inviter == pubkey && inviter == channel.created_by.as_slice(); if !is_creator_bootstrap { - let inviter_role_str = get_active_role_tx(&mut tx, channel_id, inviter) + let inviter_role_str = get_active_role_tx(&mut tx, community_id, channel_id, inviter) .await? .ok_or_else(|| { DbError::AccessDenied("inviter is not an active member".to_string()) @@ -354,7 +364,7 @@ pub async fn add_member( // elevated roles. Self-join always gets Member. if role.is_elevated() { let granter_role = match invited_by { - Some(inv) => get_active_role_tx(&mut tx, channel_id, inv).await?, + Some(inv) => get_active_role_tx(&mut tx, community_id, channel_id, inv).await?, None => None, }; match granter_role.as_deref() { @@ -372,14 +382,15 @@ pub async fn add_member( sqlx::query( r#" - INSERT INTO channel_members (channel_id, pubkey, role, invited_by) - VALUES ($1, $2, $3::member_role, $4) - ON CONFLICT (channel_id, pubkey) DO UPDATE SET + INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) + VALUES ($1, $2, $3, $4::member_role, $5) + ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET removed_at = NULL, removed_by = NULL, role = EXCLUDED.role "#, ) + .bind(community_id.as_uuid()) .bind(channel_id) .bind(pubkey) .bind(effective_role.as_str()) @@ -390,9 +401,10 @@ pub async fn add_member( let row = sqlx::query( r#" SELECT channel_id, pubkey, role::text AS role, joined_at, invited_by, removed_at - FROM channel_members WHERE channel_id = $1 AND pubkey = $2 + FROM channel_members WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 "#, ) + .bind(community_id.as_uuid()) .bind(channel_id) .bind(pubkey) .fetch_one(&mut *tx) @@ -415,6 +427,7 @@ pub async fn add_member( /// because `agent_owner_pubkey` is immutable (set once at token mint). pub async fn remove_member( pool: &PgPool, + community_id: CommunityId, channel_id: Uuid, pubkey: &[u8], actor_pubkey: &[u8], @@ -423,7 +436,7 @@ pub async fn remove_member( let is_self_remove = pubkey == actor_pubkey; if !is_self_remove { - let actor_role_str = get_active_role_tx(&mut tx, channel_id, actor_pubkey) + let actor_role_str = get_active_role_tx(&mut tx, community_id, channel_id, actor_pubkey) .await? .ok_or_else(|| DbError::AccessDenied("actor is not an active member".to_string()))?; let actor_role: MemberRole = actor_role_str.parse().map_err(|_| { @@ -432,7 +445,7 @@ pub async fn remove_member( // Safe to query outside the transaction: agent_owner_pubkey is immutable // (set once at token mint, first-mint-wins). if !actor_role.is_elevated() - && !crate::user::is_agent_owner(pool, pubkey, actor_pubkey).await? + && !crate::user::is_agent_owner(pool, community_id, pubkey, actor_pubkey).await? { return Err(DbError::AccessDenied( "only owners/admins or the agent's owner may remove other members".to_string(), @@ -443,12 +456,13 @@ pub async fn remove_member( // Defense-in-depth: prevent removing the last owner regardless of caller. // Callers (REST handlers, NIP-29 handlers) also check this, but the DB // layer enforces it as the final safety net. - let target_role = get_active_role_tx(&mut tx, channel_id, pubkey).await?; + let target_role = get_active_role_tx(&mut tx, community_id, channel_id, pubkey).await?; if target_role.as_deref() == Some("owner") { let row = sqlx::query( "SELECT COUNT(*) as cnt FROM channel_members \ - WHERE channel_id = $1 AND role = 'owner' AND removed_at IS NULL", + WHERE community_id = $1 AND channel_id = $2 AND role = 'owner' AND removed_at IS NULL", ) + .bind(community_id.as_uuid()) .bind(channel_id) .fetch_one(&mut *tx) .await?; @@ -464,10 +478,11 @@ pub async fn remove_member( r#" UPDATE channel_members SET removed_at = NOW(), removed_by = $1 - WHERE channel_id = $2 AND pubkey = $3 AND removed_at IS NULL + WHERE community_id = $2 AND channel_id = $3 AND pubkey = $4 AND removed_at IS NULL "#, ) .bind(actor_pubkey) + .bind(community_id.as_uuid()) .bind(channel_id) .bind(pubkey) .execute(&mut *tx) @@ -619,13 +634,15 @@ pub async fn list_channels(pool: &PgPool, visibility: Option<&str>) -> Result, + community_id: CommunityId, channel_id: Uuid, pubkey: &[u8], ) -> Result> { let row = sqlx::query( "SELECT role::text AS role FROM channel_members \ - WHERE channel_id = $1 AND pubkey = $2 AND removed_at IS NULL", + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 AND removed_at IS NULL", ) + .bind(community_id.as_uuid()) .bind(channel_id) .bind(pubkey) .fetch_optional(&mut **tx) @@ -1235,28 +1252,97 @@ mod tests { Keys::generate().public_key().to_bytes().to_vec() } + async fn make_test_community(pool: &PgPool) -> Uuid { + let id = Uuid::new_v4(); + let host = format!("channel-test-{}.example", id.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(host) + .execute(pool) + .await + .expect("insert test community"); + id + } + + async fn create_test_channel( + pool: &PgPool, + community_id: Uuid, + name: &str, + channel_type: ChannelType, + visibility: ChannelVisibility, + description: Option<&str>, + created_by: &[u8], + ttl_seconds: Option, + ) -> Result { + let id = Uuid::new_v4(); + + sqlx::query( + r#" + INSERT INTO channels + (id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline) + VALUES + ($1, $2, $3, $4::channel_type, $5::channel_visibility, $6, $7, $8, + CASE WHEN $8 IS NOT NULL THEN NOW() + ($8 || ' seconds')::interval ELSE NULL END) + "#, + ) + .bind(id) + .bind(community_id) + .bind(name) + .bind(channel_type.as_str()) + .bind(visibility.as_str()) + .bind(description) + .bind(created_by) + .bind(ttl_seconds) + .execute(pool) + .await + .expect("insert test channel"); + + sqlx::query( + r#" + INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) + VALUES ($1, $2, $3, 'owner', $4) + "#, + ) + .bind(community_id) + .bind(id) + .bind(created_by) + .bind(created_by) + .execute(pool) + .await + .expect("insert owner membership"); + + get_channel(pool, id).await + } + /// Agent owner (non-admin) can remove their own bot from a channel. #[tokio::test] #[ignore = "requires Postgres"] async fn test_agent_owner_can_remove_bot() { let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); let owner_pk = random_pubkey(); let agent_pk = random_pubkey(); // Create users and set agent ownership - ensure_user(&pool, &owner_pk).await.expect("ensure owner"); - ensure_user(&pool, &agent_pk).await.expect("ensure agent"); - set_agent_owner(&pool, &agent_pk, &owner_pk) + ensure_user(&pool, community, &owner_pk) + .await + .expect("ensure owner"); + ensure_user(&pool, community, &agent_pk) + .await + .expect("ensure agent"); + set_agent_owner(&pool, community, &agent_pk, &owner_pk) .await .expect("set agent owner"); // Create a channel owned by someone else entirely let channel_owner_pk = random_pubkey(); - ensure_user(&pool, &channel_owner_pk) + ensure_user(&pool, community, &channel_owner_pk) .await .expect("ensure channel owner"); - let channel = create_channel( + let channel = create_test_channel( &pool, + community_id, "test-bot-remove", ChannelType::Stream, ChannelVisibility::Open, @@ -1268,15 +1354,29 @@ mod tests { .expect("create channel"); // Add owner and agent as regular members - add_member(&pool, channel.id, &owner_pk, MemberRole::Member, None) - .await - .expect("add owner as member"); - add_member(&pool, channel.id, &agent_pk, MemberRole::Member, None) - .await - .expect("add agent as member"); + add_member( + &pool, + community, + channel.id, + &owner_pk, + MemberRole::Member, + None, + ) + .await + .expect("add owner as member"); + add_member( + &pool, + community, + channel.id, + &agent_pk, + MemberRole::Member, + None, + ) + .await + .expect("add agent as member"); // Owner should be able to remove their agent - remove_member(&pool, channel.id, &agent_pk, &owner_pk) + remove_member(&pool, community, channel.id, &agent_pk, &owner_pk) .await .expect("agent owner should be able to remove their bot"); @@ -1295,11 +1395,16 @@ mod tests { #[ignore = "requires Postgres"] async fn test_unarchive_expired_ephemeral_channel_renews_ttl_deadline() { let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); let owner_pk = random_pubkey(); - ensure_user(&pool, &owner_pk).await.expect("ensure owner"); + ensure_user(&pool, community, &owner_pk) + .await + .expect("ensure owner"); - let channel = create_channel( + let channel = create_test_channel( &pool, + community_id, "test-unarchive-renews-ttl", ChannelType::Stream, ChannelVisibility::Open, @@ -1311,8 +1416,9 @@ mod tests { .expect("create ephemeral channel"); sqlx::query( - "UPDATE channels SET archived_at = NOW(), ttl_deadline = NOW() - interval '1 second' WHERE id = $1", + "UPDATE channels SET archived_at = NOW(), ttl_deadline = NOW() - interval '1 second' WHERE community_id = $1 AND id = $2", ) + .bind(community_id) .bind(channel.id) .execute(&pool) .await @@ -1348,25 +1454,34 @@ mod tests { #[ignore = "requires Postgres"] async fn test_random_user_cannot_remove_bot() { let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); let owner_pk = random_pubkey(); let agent_pk = random_pubkey(); let random_pk = random_pubkey(); // Create users and set agent ownership - ensure_user(&pool, &owner_pk).await.expect("ensure owner"); - ensure_user(&pool, &agent_pk).await.expect("ensure agent"); - ensure_user(&pool, &random_pk).await.expect("ensure random"); - set_agent_owner(&pool, &agent_pk, &owner_pk) + ensure_user(&pool, community, &owner_pk) + .await + .expect("ensure owner"); + ensure_user(&pool, community, &agent_pk) + .await + .expect("ensure agent"); + ensure_user(&pool, community, &random_pk) + .await + .expect("ensure random"); + set_agent_owner(&pool, community, &agent_pk, &owner_pk) .await .expect("set agent owner"); // Create a channel let channel_owner_pk = random_pubkey(); - ensure_user(&pool, &channel_owner_pk) + ensure_user(&pool, community, &channel_owner_pk) .await .expect("ensure channel owner"); - let channel = create_channel( + let channel = create_test_channel( &pool, + community_id, "test-bot-no-remove", ChannelType::Stream, ChannelVisibility::Open, @@ -1378,15 +1493,29 @@ mod tests { .expect("create channel"); // Add random user and agent as regular members - add_member(&pool, channel.id, &random_pk, MemberRole::Member, None) - .await - .expect("add random as member"); - add_member(&pool, channel.id, &agent_pk, MemberRole::Member, None) - .await - .expect("add agent as member"); + add_member( + &pool, + community, + channel.id, + &random_pk, + MemberRole::Member, + None, + ) + .await + .expect("add random as member"); + add_member( + &pool, + community, + channel.id, + &agent_pk, + MemberRole::Member, + None, + ) + .await + .expect("add agent as member"); // Random user should NOT be able to remove the agent - let result = remove_member(&pool, channel.id, &agent_pk, &random_pk).await; + let result = remove_member(&pool, community, channel.id, &agent_pk, &random_pk).await; assert!( result.is_err(), "random user should not be able to remove someone else's bot" diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 9cef6c53c..29a23b51f 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -12,13 +12,15 @@ use uuid::Uuid; use buzz_core::kind::{ event_kind_i32, is_ephemeral, is_parameterized_replaceable, KIND_AUTH, KIND_EVENT_REMINDER, }; -use buzz_core::StoredEvent; +use buzz_core::{CommunityId, StoredEvent}; use crate::error::{DbError, Result}; /// Optional filters for [`query_events`]. -#[derive(Debug, Default, Clone)] +#[derive(Debug, Clone)] pub struct EventQuery { + /// Server-resolved community scope. + pub community_id: CommunityId, /// Restrict results to this channel. pub channel_id: Option, /// Restrict results to these kind values (stored as `i32` in Postgres). @@ -123,6 +125,7 @@ pub fn extract_not_before(event: &Event) -> Option { /// Returns `(StoredEvent, was_inserted)` — `was_inserted` is `false` on duplicate. pub async fn insert_event( pool: &PgPool, + community_id: CommunityId, event: &Event, channel_id: Option, ) -> Result<(StoredEvent, bool)> { @@ -150,11 +153,12 @@ pub async fn insert_event( let not_before = extract_not_before(event); let result = sqlx::query( r#" - INSERT INTO events (id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) + INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT DO NOTHING "#, ) + .bind(community_id.as_uuid()) .bind(id_bytes.as_slice()) .bind(pubkey_bytes.as_slice()) .bind(created_at) @@ -220,16 +224,22 @@ pub async fn query_events(pool: &PgPool, q: &EventQuery) -> Result Result { let mut qb: QueryBuilder = if let Some(ref p_hex) = q.p_tag_hex { let mut b = QueryBuilder::new( "SELECT COUNT(*) as cnt FROM events e \ - INNER JOIN event_mentions m ON e.id = m.event_id \ - WHERE e.deleted_at IS NULL AND m.pubkey_hex = ", + INNER JOIN event_mentions m \ + ON e.community_id = m.community_id AND e.id = m.event_id \ + WHERE e.community_id = ", ); + b.push_bind(q.community_id.as_uuid()); + b.push(" AND e.deleted_at IS NULL AND m.pubkey_hex = "); b.push_bind(p_hex.to_ascii_lowercase()); b } else { - QueryBuilder::new("SELECT COUNT(*) as cnt FROM events WHERE deleted_at IS NULL") + let mut b = QueryBuilder::new("SELECT COUNT(*) as cnt FROM events WHERE community_id = "); + b.push_bind(q.community_id.as_uuid()); + b.push(" AND deleted_at IS NULL"); + b }; let col_prefix = if q.p_tag_hex.is_some() { "e." } else { "" }; @@ -842,6 +858,7 @@ pub struct ThreadMetadataParams<'a> { /// Returns `(StoredEvent, was_inserted)`. pub async fn insert_event_with_thread_metadata( pool: &PgPool, + community_id: CommunityId, event: &Event, channel_id: Option, thread_meta: Option>, @@ -871,11 +888,12 @@ pub async fn insert_event_with_thread_metadata( let result = sqlx::query( r#" - INSERT INTO events (id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) + INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT DO NOTHING "#, ) + .bind(community_id.as_uuid()) .bind(id_bytes.as_slice()) .bind(pubkey_bytes.as_slice()) .bind(created_at) @@ -899,14 +917,15 @@ pub async fn insert_event_with_thread_metadata( let tm_result = sqlx::query( r#" INSERT INTO thread_metadata - (event_created_at, event_id, channel_id, + (community_id, event_created_at, event_id, channel_id, parent_event_id, parent_event_created_at, root_event_id, root_event_created_at, depth, broadcast) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) ON CONFLICT DO NOTHING "#, ) + .bind(community_id.as_uuid()) .bind(meta.event_created_at) .bind(meta.event_id) .bind(meta.channel_id) @@ -931,14 +950,15 @@ pub async fn insert_event_with_thread_metadata( sqlx::query( r#" INSERT INTO thread_metadata - (event_created_at, event_id, channel_id, + (community_id, event_created_at, event_id, channel_id, parent_event_id, parent_event_created_at, root_event_id, root_event_created_at, depth, broadcast) - VALUES ($1, $2, $3, NULL, NULL, NULL, NULL, 0, false) + VALUES ($1, $2, $3, $4, NULL, NULL, NULL, NULL, 0, false) ON CONFLICT DO NOTHING "#, ) + .bind(community_id.as_uuid()) .bind(parent_ts) .bind(pid) .bind(meta.channel_id) @@ -953,14 +973,15 @@ pub async fn insert_event_with_thread_metadata( sqlx::query( r#" INSERT INTO thread_metadata - (event_created_at, event_id, channel_id, + (community_id, event_created_at, event_id, channel_id, parent_event_id, parent_event_created_at, root_event_id, root_event_created_at, depth, broadcast) - VALUES ($1, $2, $3, NULL, NULL, NULL, NULL, 0, false) + VALUES ($1, $2, $3, $4, NULL, NULL, NULL, NULL, 0, false) ON CONFLICT DO NOTHING "#, ) + .bind(community_id.as_uuid()) .bind(root_ts) .bind(root_id) .bind(meta.channel_id) @@ -973,9 +994,10 @@ pub async fn insert_event_with_thread_metadata( r#" UPDATE thread_metadata SET reply_count = reply_count + 1, last_reply_at = NOW() - WHERE event_id = $1 + WHERE community_id = $1 AND event_id = $2 "#, ) + .bind(community_id.as_uuid()) .bind(pid) .execute(&mut *tx) .await?; @@ -985,9 +1007,10 @@ pub async fn insert_event_with_thread_metadata( r#" UPDATE thread_metadata SET descendant_count = descendant_count + 1 - WHERE event_id = $1 + WHERE community_id = $1 AND event_id = $2 "#, ) + .bind(community_id.as_uuid()) .bind(root_id) .execute(&mut *tx) .await?; @@ -1083,7 +1106,19 @@ pub async fn claim_due_reminder( event_id: &[u8], event_created_at: DateTime, ) -> Result { - let now_epoch = Utc::now().timestamp(); + claim_due_reminder_with_stamp(pool, event_id, event_created_at, Utc::now().timestamp()).await +} + +/// Atomically claim a due reminder using a caller-supplied delivery stamp. +/// +/// The same stamp should be passed to [`release_due_reminder`] if the publish +/// side effect fails, so rollback can compare-and-clear only this pod's claim. +pub async fn claim_due_reminder_with_stamp( + pool: &PgPool, + event_id: &[u8], + event_created_at: DateTime, + delivery_stamp: i64, +) -> Result { let result = sqlx::query( r#" UPDATE events @@ -1091,7 +1126,7 @@ pub async fn claim_due_reminder( WHERE created_at = $2 AND id = $3 AND delivered_at IS NULL "#, ) - .bind(now_epoch) + .bind(delivery_stamp) .bind(event_created_at) .bind(event_id) .execute(pool) @@ -1100,6 +1135,35 @@ pub async fn claim_due_reminder( Ok(result.rows_affected() > 0) } +/// Release a previously claimed reminder when publish fails. +/// +/// The `delivery_stamp` must be the exact value written by the claiming pod; +/// that compare-and-clear prevents one pod from rolling back another pod's +/// later claim after a retry/race. +pub async fn release_due_reminder( + pool: &PgPool, + event_id: &[u8], + event_created_at: DateTime, + delivery_stamp: i64, +) -> Result { + let result = sqlx::query( + r#" + UPDATE events + SET delivered_at = NULL + WHERE created_at = $1 + AND id = $2 + AND delivered_at = $3 + "#, + ) + .bind(event_created_at) + .bind(event_id) + .bind(delivery_stamp) + .execute(pool) + .await?; + + Ok(result.rows_affected() == 1) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9dd4d3e10..6b0a7199c 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -47,7 +47,7 @@ use sqlx::{PgPool, QueryBuilder, Row}; use std::time::Duration; use uuid::Uuid; -use buzz_core::StoredEvent; +use buzz_core::{CommunityId, StoredEvent}; /// Extract p-tag mentions from an event and insert into the `event_mentions` table. /// @@ -55,6 +55,7 @@ use buzz_core::StoredEvent; /// Uses `INSERT ... ON CONFLICT DO NOTHING` so duplicate inserts are silently skipped. pub async fn insert_mentions( pool: &PgPool, + community_id: CommunityId, event: &nostr::Event, channel_id: Option, ) -> Result<()> { @@ -106,11 +107,12 @@ pub async fn insert_mentions( // Single multi-row INSERT ... ON CONFLICT DO NOTHING — one round-trip regardless of mention count. let mut qb: QueryBuilder = QueryBuilder::new( "INSERT INTO event_mentions \ - (pubkey_hex, event_id, event_created_at, channel_id, event_kind) ", + (community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) ", ); qb.push_values(&valid_pubkeys, |mut b, pubkey| { - b.push_bind(pubkey.as_str()) + b.push_bind(community_id.as_uuid()) + .push_bind(pubkey.as_str()) .push_bind(event_id_bytes.as_slice()) .push_bind(created_at) .push_bind(channel_id) @@ -162,6 +164,15 @@ impl Default for DbConfig { } } +/// Community host-map row returned by [`Db::lookup_community_by_host`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CommunityRecord { + /// Stable server-resolved community id. + pub id: CommunityId, + /// Normalized host that maps to this community. + pub host: String, +} + /// Token summary returned by [`Db::list_active_tokens`]. #[derive(Debug, Clone)] pub struct TokenSummary { @@ -216,15 +227,101 @@ impl Db { self.pool.begin().await.map_err(Into::into) } + /// Returns the community mapped to a normalized request host, if one exists. + /// + /// The caller owns host normalization and turns `None` into the fail-closed + /// request/connection error. buzz-db only reads the durable host map. + pub async fn lookup_community_by_host( + &self, + normalized_host: &str, + ) -> Result> { + let row = sqlx::query( + r#" + SELECT id, host + FROM communities + WHERE host = $1 + "#, + ) + .bind(normalized_host) + .fetch_optional(&self.pool) + .await?; + + row.map(|row| { + let id: Uuid = row.try_get("id")?; + let host: String = row.try_get("host")?; + + Ok(CommunityRecord { + id: CommunityId::from_uuid(id), + host, + }) + }) + .transpose() + } + + /// Ensure a configured community host exists and return its row. + /// + /// This is the startup/config seeding path for N=1 deployments. Migrations + /// create the schema only; deployment-specific hosts are not hardcoded into + /// schema history. + pub async fn ensure_configured_community( + &self, + normalized_host: &str, + ) -> Result { + let row = sqlx::query( + r#" + INSERT INTO communities (host) + VALUES ($1) + ON CONFLICT (host) DO UPDATE SET host = EXCLUDED.host + RETURNING id, host + "#, + ) + .bind(normalized_host) + .fetch_one(&self.pool) + .await?; + + let id: Uuid = row.try_get("id")?; + let host: String = row.try_get("host")?; + + Ok(CommunityRecord { + id: CommunityId::from_uuid(id), + host, + }) + } + + /// Returns the community that owns a channel, if the channel exists. + /// + /// Internal relay producers use this to derive tenant context from the row + /// they are acting on, rather than falling back to an implicit default. + pub async fn community_of_channel(&self, channel_id: Uuid) -> Result> { + let row = sqlx::query( + r#" + SELECT community_id + FROM channels + WHERE id = $1 + AND deleted_at IS NULL + "#, + ) + .bind(channel_id) + .fetch_optional(&self.pool) + .await?; + + row.map(|row| { + let id: Uuid = row.try_get("community_id")?; + Ok(CommunityId::from_uuid(id)) + }) + .transpose() + } + /// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate. pub async fn insert_event( &self, + community_id: CommunityId, event: &nostr::Event, channel_id: Option, ) -> Result<(StoredEvent, bool)> { - let result = event::insert_event(&self.pool, event, channel_id).await?; + let result = event::insert_event(&self.pool, community_id, event, channel_id).await?; if result.1 { - if let Err(e) = insert_mentions(&self.pool, event, channel_id).await { + if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } } @@ -322,15 +419,21 @@ impl Db { /// Atomically insert an event AND its thread metadata in a single transaction. pub async fn insert_event_with_thread_metadata( &self, + community_id: CommunityId, event: &nostr::Event, channel_id: Option, thread_meta: Option>, ) -> Result<(StoredEvent, bool)> { - let result = - event::insert_event_with_thread_metadata(&self.pool, event, channel_id, thread_meta) - .await?; + let result = event::insert_event_with_thread_metadata( + &self.pool, + community_id, + event, + channel_id, + thread_meta, + ) + .await?; if result.1 { - if let Err(e) = insert_mentions(&self.pool, event, channel_id).await { + if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } } @@ -340,6 +443,7 @@ impl Db { /// Creates a new channel, bootstraps the creator as owner, and returns the record. pub async fn create_channel( &self, + community_id: CommunityId, name: &str, channel_type: channel::ChannelType, visibility: channel::ChannelVisibility, @@ -349,6 +453,7 @@ impl Db { ) -> Result { channel::create_channel( &self.pool, + community_id, name, channel_type, visibility, @@ -365,6 +470,7 @@ impl Db { #[allow(clippy::too_many_arguments)] pub async fn create_channel_with_id( &self, + community_id: CommunityId, channel_id: Uuid, name: &str, channel_type: channel::ChannelType, @@ -375,6 +481,7 @@ impl Db { ) -> Result<(channel::ChannelRecord, bool)> { channel::create_channel_with_id( &self.pool, + community_id, channel_id, name, channel_type, @@ -404,22 +511,32 @@ impl Db { /// Adds a member to a channel. pub async fn add_member( &self, + community_id: CommunityId, channel_id: Uuid, pubkey: &[u8], role: channel::MemberRole, invited_by: Option<&[u8]>, ) -> Result { - channel::add_member(&self.pool, channel_id, pubkey, role, invited_by).await + channel::add_member( + &self.pool, + community_id, + channel_id, + pubkey, + role, + invited_by, + ) + .await } /// Removes a member from a channel. pub async fn remove_member( &self, + community_id: CommunityId, channel_id: Uuid, pubkey: &[u8], actor_pubkey: &[u8], ) -> Result<()> { - channel::remove_member(&self.pool, channel_id, pubkey, actor_pubkey).await + channel::remove_member(&self.pool, community_id, channel_id, pubkey, actor_pubkey).await } /// Returns `true` if the pubkey is an active member. @@ -553,19 +670,45 @@ impl Db { event::claim_due_reminder(&self.pool, event_id, event_created_at).await } + /// Atomically claim a due reminder using a caller-supplied delivery stamp. + pub async fn claim_due_reminder_with_stamp( + &self, + event_id: &[u8], + event_created_at: chrono::DateTime, + delivery_stamp: i64, + ) -> Result { + event::claim_due_reminder_with_stamp(&self.pool, event_id, event_created_at, delivery_stamp) + .await + } + + /// Release a claimed due reminder after a publish failure. + pub async fn release_due_reminder( + &self, + event_id: &[u8], + event_created_at: chrono::DateTime, + delivery_stamp: i64, + ) -> Result { + event::release_due_reminder(&self.pool, event_id, event_created_at, delivery_stamp).await + } + /// Ensure a user record exists (upsert). - pub async fn ensure_user(&self, pubkey: &[u8]) -> Result<()> { - user::ensure_user(&self.pool, pubkey).await + pub async fn ensure_user(&self, community_id: CommunityId, pubkey: &[u8]) -> Result<()> { + user::ensure_user(&self.pool, community_id, pubkey).await } /// Get a single user record by pubkey. - pub async fn get_user(&self, pubkey: &[u8]) -> Result> { - user::get_user(&self.pool, pubkey).await + pub async fn get_user( + &self, + community_id: CommunityId, + pubkey: &[u8], + ) -> Result> { + user::get_user(&self.pool, community_id, pubkey).await } /// Update a user's profile fields. pub async fn update_user_profile( &self, + community_id: CommunityId, pubkey: &[u8], display_name: Option<&str>, avatar_url: Option<&str>, @@ -574,6 +717,7 @@ impl Db { ) -> Result<()> { user::update_user_profile( &self.pool, + community_id, pubkey, display_name, avatar_url, @@ -586,43 +730,61 @@ impl Db { /// Look up a user by NIP-05 handle. pub async fn get_user_by_nip05( &self, + community_id: CommunityId, local_part: &str, domain: &str, ) -> Result> { - user::get_user_by_nip05(&self.pool, local_part, domain).await + user::get_user_by_nip05(&self.pool, community_id, local_part, domain).await } /// Search users by display name, NIP-05 handle, or pubkey prefix. pub async fn search_users( &self, + community_id: CommunityId, query: &str, limit: u32, ) -> Result> { - user::search_users(&self.pool, query, limit).await + user::search_users(&self.pool, community_id, query, limit).await } /// Atomically set agent owner — only if no owner is currently assigned. /// Returns Ok(true) if set, Ok(false) if an owner already exists. - pub async fn set_agent_owner(&self, agent_pubkey: &[u8], owner_pubkey: &[u8]) -> Result { - user::set_agent_owner(&self.pool, agent_pubkey, owner_pubkey).await + pub async fn set_agent_owner( + &self, + community_id: CommunityId, + agent_pubkey: &[u8], + owner_pubkey: &[u8], + ) -> Result { + user::set_agent_owner(&self.pool, community_id, agent_pubkey, owner_pubkey).await } /// Get the channel_add_policy and agent_owner_pubkey for a user. pub async fn get_agent_channel_policy( &self, + community_id: CommunityId, pubkey: &[u8], ) -> Result>)>> { - user::get_agent_channel_policy(&self.pool, pubkey).await + user::get_agent_channel_policy(&self.pool, community_id, pubkey).await } /// Check whether `actor_pubkey` is the agent owner of `target_pubkey`. - pub async fn is_agent_owner(&self, target_pubkey: &[u8], actor_pubkey: &[u8]) -> Result { - user::is_agent_owner(&self.pool, target_pubkey, actor_pubkey).await + pub async fn is_agent_owner( + &self, + community_id: CommunityId, + target_pubkey: &[u8], + actor_pubkey: &[u8], + ) -> Result { + user::is_agent_owner(&self.pool, community_id, target_pubkey, actor_pubkey).await } /// Set the channel_add_policy for a user. - pub async fn set_channel_add_policy(&self, pubkey: &[u8], policy: &str) -> Result<()> { - user::set_channel_add_policy(&self.pool, pubkey, policy).await + pub async fn set_channel_add_policy( + &self, + community_id: CommunityId, + pubkey: &[u8], + policy: &str, + ) -> Result<()> { + user::set_channel_add_policy(&self.pool, community_id, pubkey, policy).await } /// Find an existing DM by its participant hash. @@ -1088,6 +1250,7 @@ impl Db { /// Create a new workflow. pub async fn create_workflow( &self, + community_id: CommunityId, channel_id: Option, owner_pubkey: &[u8], name: &str, @@ -1096,6 +1259,7 @@ impl Db { ) -> Result { workflow::create_workflow( &self.pool, + community_id, channel_id, owner_pubkey, name, @@ -1133,6 +1297,51 @@ impl Db { workflow::list_all_enabled_workflows(&self.pool).await } + /// Claim a scheduled workflow fire for an authoritative schedule instant. + /// + /// Returns `Some` only for the first pod to claim `(workflow_id, + /// scheduled_for)`; all other pods must skip creating a run. The claim SQL + /// resolves `community_id` from the workflow row; callers never supply it. + pub async fn claim_scheduled_workflow_fire( + &self, + workflow_id: Uuid, + scheduled_for: chrono::DateTime, + ) -> Result> { + workflow::claim_scheduled_workflow_fire(&self.pool, workflow_id, scheduled_for).await + } + + /// Fetch the latest claimed schedule instant for interval trigger anchoring. + pub async fn latest_scheduled_workflow_fire( + &self, + workflow_id: Uuid, + ) -> Result>> { + workflow::latest_scheduled_workflow_fire(&self.pool, workflow_id).await + } + + /// Attach the workflow run id created from a won scheduled-fire claim. + pub async fn attach_scheduled_workflow_run( + &self, + workflow_id: Uuid, + scheduled_for: chrono::DateTime, + workflow_run_id: Uuid, + ) -> Result { + workflow::attach_scheduled_workflow_run( + &self.pool, + workflow_id, + scheduled_for, + workflow_run_id, + ) + .await + } + + /// Delete old scheduled workflow fire claims before a retention cutoff. + pub async fn prune_scheduled_workflow_fires_before( + &self, + older_than: chrono::DateTime, + ) -> Result { + workflow::prune_scheduled_workflow_fires_before(&self.pool, older_than).await + } + /// Update a workflow's name, definition, and hash. pub async fn update_workflow( &self, @@ -1487,6 +1696,7 @@ impl Db { /// skip fan-out/dispatch when `was_inserted` is false. pub async fn replace_addressable_event( &self, + community_id: CommunityId, event: &nostr::Event, channel_id: Option, ) -> Result<(StoredEvent, bool)> { @@ -1501,6 +1711,10 @@ impl Db { // Collisions cause extra serialization, not incorrect behavior. let lock_key = { let mut h: u64 = 0xcbf29ce484222325; // FNV offset basis + for b in community_id.as_uuid().as_bytes() { + h ^= *b as u64; + h = h.wrapping_mul(0x100000001b3); + } for b in kind_i32.to_le_bytes() { h ^= b as u64; h = h.wrapping_mul(0x100000001b3); // FNV prime @@ -1531,11 +1745,12 @@ impl Db { // historical data where prior bugs may have left multiple live rows. let existing: Option<(chrono::DateTime, Vec)> = sqlx::query_as( "SELECT created_at, id FROM events \ - WHERE kind = $1 AND pubkey = $2 \ - AND channel_id IS NOT DISTINCT FROM $3 \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ + AND channel_id IS NOT DISTINCT FROM $4 \ AND deleted_at IS NULL \ ORDER BY created_at DESC, id ASC LIMIT 1", ) + .bind(community_id.as_uuid()) .bind(kind_i32) .bind(pubkey_bytes.as_slice()) .bind(channel_id) @@ -1562,10 +1777,11 @@ impl Db { // Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety. sqlx::query( "UPDATE events SET deleted_at = NOW() \ - WHERE kind = $1 AND pubkey = $2 \ - AND channel_id IS NOT DISTINCT FROM $3 \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ + AND channel_id IS NOT DISTINCT FROM $4 \ AND deleted_at IS NULL", ) + .bind(community_id.as_uuid()) .bind(kind_i32) .bind(pubkey_bytes.as_slice()) .bind(channel_id) @@ -1579,10 +1795,11 @@ impl Db { let d_tag = crate::event::extract_d_tag(event); let insert_result = sqlx::query( - "INSERT INTO events (id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) \ + "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \ ON CONFLICT DO NOTHING", ) + .bind(community_id.as_uuid()) .bind(event.id.as_bytes().as_slice()) .bind(pubkey_bytes.as_slice()) .bind(created_at) @@ -1611,7 +1828,7 @@ impl Db { // Mentions are a denormalized index — safe outside the transaction. // insert_event() normally handles this, but we inlined the INSERT above. - if let Err(e) = crate::insert_mentions(&self.pool, event, channel_id).await { + if let Err(e) = crate::insert_mentions(&self.pool, community_id, event, channel_id).await { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } @@ -1640,6 +1857,7 @@ impl Db { /// this function instead, where the author's pubkey + d-tag is the natural key. pub async fn replace_parameterized_event( &self, + community_id: CommunityId, event: &nostr::Event, d_tag: &str, channel_id: Option, @@ -1654,6 +1872,10 @@ impl Db { // Same algorithm as replace_addressable_event — deterministic across processes. let lock_key = { let mut h: u64 = 0xcbf29ce484222325; // FNV offset basis + for b in community_id.as_uuid().as_bytes() { + h ^= *b as u64; + h = h.wrapping_mul(0x100000001b3); + } for b in kind_i32.to_le_bytes() { h ^= b as u64; h = h.wrapping_mul(0x100000001b3); @@ -1679,9 +1901,10 @@ impl Db { // Check for existing event with same (kind, pubkey, d_tag). let existing: Option<(chrono::DateTime, Vec)> = sqlx::query_as( "SELECT created_at, id FROM events \ - WHERE kind = $1 AND pubkey = $2 AND d_tag = $3 AND deleted_at IS NULL \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \ ORDER BY created_at DESC, id ASC LIMIT 1", ) + .bind(community_id.as_uuid()) .bind(kind_i32) .bind(pubkey_bytes.as_slice()) .bind(d_tag) @@ -1705,8 +1928,9 @@ impl Db { // Soft-delete the older event(s). sqlx::query( "UPDATE events SET deleted_at = NOW() \ - WHERE kind = $1 AND pubkey = $2 AND d_tag = $3 AND deleted_at IS NULL", + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL", ) + .bind(community_id.as_uuid()) .bind(kind_i32) .bind(pubkey_bytes.as_slice()) .bind(d_tag) @@ -1720,10 +1944,11 @@ impl Db { let received_at = chrono::Utc::now(); let insert_result = sqlx::query( - "INSERT INTO events (id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \ + "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \ ON CONFLICT DO NOTHING", ) + .bind(community_id.as_uuid()) .bind(event.id.as_bytes().as_slice()) .bind(pubkey_bytes.as_slice()) .bind(created_at) @@ -1750,7 +1975,7 @@ impl Db { tx.commit().await?; // Mentions are a denormalized index — safe outside the transaction. - if let Err(e) = crate::insert_mentions(&self.pool, event, channel_id).await { + if let Err(e) = crate::insert_mentions(&self.pool, community_id, event, channel_id).await { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index f4f3c9ab3..401d067d8 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -1,9 +1,8 @@ //! Embedded SQLx migrations for Buzz. //! -//! Fresh deployments apply the checked-in SQL files under `migrations/`. -//! Existing pre-SQLx deployments are baselined when core Buzz tables already -//! exist but `_sqlx_migrations` does not, so startup will not try to replay the -//! initial schema over a live database. +//! Fresh deployments apply the checked-in SQL files under `migrations/`. The +//! multi-tenant rewrite owns a clean consolidated `0001`; legacy single-tenant +//! cutover/backfill is a separate operator script, not startup migration state. use sqlx::PgPool; @@ -11,154 +10,619 @@ use crate::Result; static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("../../migrations"); -#[cfg(test)] -static SCHEMA_SQL: &str = include_str!("../../../schema/schema.sql"); - -const BASELINE_MIGRATION_VERSIONS: &[i64] = &[1, 2]; - /// Run all pending Buzz database migrations. pub async fn run_migrations(pool: &PgPool) -> Result<()> { - baseline_existing_database(pool).await?; MIGRATOR.run(pool).await?; Ok(()) } -async fn baseline_existing_database(pool: &PgPool) -> Result<()> { - if migrations_table_exists(pool).await? || !pre_sqlx_schema_exists(pool).await? { - return Ok(()); - } - - ensure_migrations_table(pool).await?; - - for version in BASELINE_MIGRATION_VERSIONS { - let migration = MIGRATOR - .iter() - .find(|migration| migration.version == *version) - .expect("baseline migration version must exist in embedded migrator"); - - sqlx::query( - r#" - INSERT INTO _sqlx_migrations - (version, description, success, checksum, execution_time) - VALUES ($1, $2, TRUE, $3, 0) - ON CONFLICT (version) DO NOTHING - "#, - ) - .bind(migration.version) - .bind(&*migration.description) - .bind(&*migration.checksum) - .execute(pool) - .await?; - } - - tracing::info!( - versions = ?BASELINE_MIGRATION_VERSIONS, - "Baselined existing Buzz database for SQLx migrations" - ); - - Ok(()) -} - -async fn migrations_table_exists(pool: &PgPool) -> Result { - let exists = sqlx::query_scalar::<_, bool>( - r#" - SELECT EXISTS ( - SELECT 1 - FROM information_schema.tables - WHERE table_schema = 'public' - AND table_name = '_sqlx_migrations' - ) - "#, - ) - .fetch_one(pool) - .await?; - - Ok(exists) -} - -async fn pre_sqlx_schema_exists(pool: &PgPool) -> Result { - let exists = sqlx::query_scalar::<_, bool>( - r#" - SELECT EXISTS ( - SELECT 1 - FROM information_schema.tables - WHERE table_schema = 'public' - AND table_name = 'events' - ) AND EXISTS ( - SELECT 1 - FROM information_schema.tables - WHERE table_schema = 'public' - AND table_name = 'channels' - ) - "#, - ) - .fetch_one(pool) - .await?; - - Ok(exists) -} - -async fn ensure_migrations_table(pool: &PgPool) -> Result<()> { - sqlx::query( - r#" - CREATE TABLE IF NOT EXISTS _sqlx_migrations ( - version BIGINT PRIMARY KEY, - description TEXT NOT NULL, - installed_on TIMESTAMPTZ NOT NULL DEFAULT now(), - success BOOLEAN NOT NULL, - checksum BYTEA NOT NULL, - execution_time BIGINT NOT NULL - ) - "#, - ) - .execute(pool) - .await?; - - Ok(()) -} - #[cfg(test)] mod tests { use super::*; - use sqlx::PgPool; + use std::collections::BTreeSet; const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + enum ConstraintKind { + ForeignKey, + PrimaryKey, + Unique, + } + + #[derive(Debug, Clone, PartialEq, Eq)] + struct ConstraintLint { + table: String, + kind: ConstraintKind, + description: String, + columns: Vec, + } + + fn migration_sql() -> &'static str { + MIGRATOR + .iter() + .find(|migration| migration.version == 1) + .expect("initial migration must exist") + .sql + .as_str() + } + + fn strip_sql_comments(sql: &str) -> String { + sql.lines() + .map(|line| line.split_once("--").map_or(line, |(before, _)| before)) + .collect::>() + .join("\n") + } + + fn normalize_sql(sql: &str) -> String { + strip_sql_comments(sql) + .split_whitespace() + .collect::>() + .join(" ") + .to_ascii_lowercase() + } + + fn split_sql_statements(sql: &str) -> Vec { + let sql = strip_sql_comments(sql); + let bytes = sql.as_bytes(); + let mut statements = Vec::new(); + let mut start = 0usize; + let mut idx = 0usize; + let mut in_single_quote = false; + let mut in_dollar_quote = false; + + while idx < bytes.len() { + match bytes[idx] { + b'\'' if !in_dollar_quote => { + in_single_quote = !in_single_quote; + idx += 1; + } + b'$' if !in_single_quote && idx + 1 < bytes.len() && bytes[idx + 1] == b'$' => { + in_dollar_quote = !in_dollar_quote; + idx += 2; + } + b';' if !in_single_quote && !in_dollar_quote => { + let statement = sql[start..idx].trim(); + if !statement.is_empty() { + statements.push(statement.to_owned()); + } + start = idx + 1; + idx += 1; + } + _ => idx += 1, + } + } + + let tail = sql[start..].trim(); + if !tail.is_empty() { + statements.push(tail.to_owned()); + } + + statements + } + + fn find_matching_paren(sql: &str, open: usize) -> Option { + let mut depth = 0usize; + for (offset, byte) in sql.as_bytes()[open..].iter().enumerate() { + match byte { + b'(' => depth += 1, + b')' => { + depth = depth.checked_sub(1)?; + if depth == 0 { + return Some(open + offset); + } + } + _ => {} + } + } + None + } + + fn split_top_level_csv(input: &str) -> Vec { + let mut parts = Vec::new(); + let mut start = 0usize; + let mut depth = 0usize; + for (idx, byte) in input.bytes().enumerate() { + match byte { + b'(' => depth += 1, + b')' => depth = depth.saturating_sub(1), + b',' if depth == 0 => { + parts.push(input[start..idx].trim().to_owned()); + start = idx + 1; + } + _ => {} + } + } + let tail = input[start..].trim(); + if !tail.is_empty() { + parts.push(tail.to_owned()); + } + parts + } + + fn identifier_after_keyword(statement: &str, keyword: &str) -> Option { + let lower = statement.to_ascii_lowercase(); + let keyword_pos = lower.find(keyword)?; + let mut remainder = statement[keyword_pos + keyword.len()..].trim_start(); + for prefix in ["if not exists", "if exists", "only"] { + if remainder.to_ascii_lowercase().starts_with(prefix) { + remainder = remainder[prefix.len()..].trim_start(); + } + } + + let identifier = remainder + .split(|ch: char| ch.is_whitespace() || ch == '(') + .next()? + .trim_matches('"') + .rsplit('.') + .next()? + .trim_matches('"') + .to_ascii_lowercase(); + (!identifier.is_empty()).then_some(identifier) + } + + fn first_parenthesized_columns(input: &str) -> Vec { + let Some(open) = input.find('(') else { + return Vec::new(); + }; + let Some(close) = find_matching_paren(input, open) else { + return Vec::new(); + }; + + split_top_level_csv(&input[open + 1..close]) + .into_iter() + .filter_map(|column| { + let name = column + .trim() + .trim_matches('"') + .split_whitespace() + .next()? + .trim_matches('"') + .to_ascii_lowercase(); + (!name.is_empty()).then_some(name) + }) + .collect() + } + + fn column_definition_name(definition: &str) -> Option { + let trimmed = definition.trim(); + let lower = trimmed.to_ascii_lowercase(); + if lower.starts_with("constraint ") + || lower.starts_with("primary key") + || lower.starts_with("foreign key") + || lower.starts_with("unique") + || lower.starts_with("check ") + || lower.starts_with("exclude ") + { + return None; + } + + let name = trimmed + .split_whitespace() + .next()? + .trim_matches('"') + .to_ascii_lowercase(); + (!name.is_empty()).then_some(name) + } + + fn create_table_body(statement: &str) -> Option<(String, Vec)> { + let table = identifier_after_keyword(statement, "create table")?; + let open = statement.find('(')?; + let close = find_matching_paren(statement, open)?; + Some((table, split_top_level_csv(&statement[open + 1..close]))) + } + + fn create_table_definitions(sql: &str) -> Vec<(String, Vec)> { + split_sql_statements(sql) + .into_iter() + .filter_map(|statement| { + let normalized = statement.trim_start().to_ascii_lowercase(); + if !normalized.starts_with("create table") || normalized.contains(" partition of ") + { + return None; + } + create_table_body(&statement) + }) + .collect() + } + + fn create_tables(sql: &str) -> BTreeSet { + create_table_definitions(sql) + .into_iter() + .map(|(table, _)| table) + .collect() + } + + fn table_has_not_null_community_id(definitions: &[String]) -> bool { + definitions.iter().any(|definition| { + column_definition_name(definition).as_deref() == Some("community_id") + && normalize_sql(definition).contains("not null") + }) + } + + fn operator_global_tables(sql: &str) -> BTreeSet { + let mut globals = BTreeSet::new(); + let normalized = normalize_sql(sql); + let Some(insert_pos) = normalized.find("insert into _operator_global_tables") else { + return globals; + }; + + for value in [ + "communities", + "rate_limit_violations", + "_operator_global_tables", + ] { + if normalized[insert_pos..].contains(&format!("'{value}'")) { + globals.insert(value.to_owned()); + } + } + + globals + } + + fn scoped_tables(sql: &str) -> BTreeSet { + let globals = operator_global_tables(sql); + create_tables(sql) + .into_iter() + .filter(|table| !globals.contains(table)) + .collect() + } + + fn constraint_lint_for_definition(table: &str, definition: &str) -> Option { + let normalized = normalize_sql(definition); + let definition_without_name = if normalized.starts_with("constraint ") { + let after_constraint = definition + .trim_start() + .splitn(3, char::is_whitespace) + .nth(2) + .unwrap_or(""); + normalize_sql(after_constraint) + } else { + normalized.clone() + }; + + if definition_without_name.starts_with("primary key") { + Some(ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::PrimaryKey, + description: definition.to_owned(), + columns: first_parenthesized_columns(&definition_without_name), + }) + } else if definition_without_name.starts_with("unique") { + Some(ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::Unique, + description: definition.to_owned(), + columns: first_parenthesized_columns(&definition_without_name), + }) + } else if definition_without_name.starts_with("foreign key") { + Some(ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::ForeignKey, + description: definition.to_owned(), + columns: first_parenthesized_columns(&definition_without_name), + }) + } else if normalized.contains(" primary key") { + column_definition_name(definition).map(|column| ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::PrimaryKey, + description: definition.to_owned(), + columns: vec![column], + }) + } else if normalized.contains(" references ") { + column_definition_name(definition).map(|column| ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::ForeignKey, + description: definition.to_owned(), + columns: vec![column], + }) + } else if normalized.contains(" unique") { + column_definition_name(definition).map(|column| ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::Unique, + description: definition.to_owned(), + columns: vec![column], + }) + } else { + None + } + } + + fn table_constraints(sql: &str, scoped_tables: &BTreeSet) -> Vec { + create_table_definitions(sql) + .into_iter() + .filter(|(table, _)| scoped_tables.contains(table)) + .flat_map(|(table, definitions)| { + definitions.into_iter().filter_map(move |definition| { + constraint_lint_for_definition(&table, &definition) + }) + }) + .collect() + } + + fn alter_table_constraints(sql: &str, scoped_tables: &BTreeSet) -> Vec { + split_sql_statements(sql) + .into_iter() + .filter_map(|statement| { + let normalized = normalize_sql(&statement); + if !normalized.starts_with("alter table") { + return None; + } + + let table = identifier_after_keyword(&statement, "alter table")?; + if !scoped_tables.contains(&table) { + return None; + } + + let add_pos = normalized.find(" add ")?; + let definition = normalized[add_pos + " add ".len()..].trim(); + constraint_lint_for_definition(&table, definition) + }) + .collect() + } + + fn unique_indexes(sql: &str, scoped_tables: &BTreeSet) -> Vec { + split_sql_statements(sql) + .into_iter() + .filter_map(|statement| { + let normalized = normalize_sql(&statement); + if !normalized.starts_with("create unique index") { + return None; + } + + let lower_statement = statement.to_ascii_lowercase(); + let on_pos = lower_statement.find(" on ")?; + let table = statement[on_pos + " on ".len()..] + .trim_start() + .split(|ch: char| ch.is_whitespace() || ch == '(') + .next()? + .trim_matches('"') + .rsplit('.') + .next()? + .trim_matches('"') + .to_ascii_lowercase(); + + scoped_tables.contains(&table).then(|| ConstraintLint { + table, + kind: ConstraintKind::Unique, + description: statement.clone(), + columns: first_parenthesized_columns(&statement[on_pos + " on ".len()..]), + }) + }) + .collect() + } + + fn scoped_constraint_lints(sql: &str, scoped_tables: &BTreeSet) -> Vec { + let mut constraints = table_constraints(sql, scoped_tables); + constraints.extend(alter_table_constraints(sql, scoped_tables)); + constraints.extend(unique_indexes(sql, scoped_tables)); + constraints + } + + fn is_allowed_partition_primary_key_exception(constraint: &ConstraintLint) -> bool { + constraint.table == "delivery_log" + && constraint.kind == ConstraintKind::PrimaryKey + && constraint.columns == ["delivered_at", "id"] + } + + fn scoped_constraint_violations(sql: &str) -> Vec { + let scoped_tables = scoped_tables(sql); + scoped_constraint_lints(sql, &scoped_tables) + .into_iter() + .filter(|constraint| { + if is_allowed_partition_primary_key_exception(constraint) { + return false; + } + constraint.columns.first().map(String::as_str) != Some("community_id") + }) + .collect() + } + + fn has_channels_community_id_immutability_guard(sql: &str) -> bool { + let normalized = normalize_sql(sql); + normalized.contains("create trigger") + && normalized.contains("before update") + && normalized.contains(" on channels") + && normalized.contains("community_id") + && normalized.contains("old.community_id") + && normalized.contains("new.community_id") + && normalized.contains("raise exception") + } + + fn forbidden_channels_community_id_mutations(sql: &str) -> Vec { + split_sql_statements(sql) + .into_iter() + .filter(|statement| { + let normalized = normalize_sql(statement); + let updates_channels = + identifier_after_keyword(statement, "update").as_deref() == Some("channels"); + let mutates_with_update = updates_channels + && normalized.contains(" set ") + && normalized.contains("community_id"); + let alters_channels = identifier_after_keyword(statement, "alter table").as_deref() + == Some("channels"); + let drops_channels = identifier_after_keyword(statement, "drop table").as_deref() + == Some("channels"); + let drops_or_rewrites_column = alters_channels + && (normalized.contains("drop column community_id") + || normalized.contains("alter column community_id") + || normalized.contains("rename column community_id") + || normalized.contains("rename community_id") + || normalized.contains("drop trigger") + || normalized.contains("disable trigger")); + + mutates_with_update || drops_or_rewrites_column || drops_channels + }) + .collect() + } + #[test] - fn embedded_migrator_contains_all_schema_migrations() { + fn embedded_migrator_contains_consolidated_initial_schema() { let migrations: Vec<_> = MIGRATOR.iter().collect(); - assert_eq!(migrations.len(), 3); + assert_eq!(migrations.len(), 1); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); - assert!( - migrations[0].sql.as_str().contains("CREATE TABLE channels"), - "initial schema migration should include Buzz core tables" - ); - assert!( - migrations[0] - .sql - .as_str() - .contains("CREATE TABLE IF NOT EXISTS relay_members"), - "initial schema migration should include relay_members" - ); + assert!(migrations[0] + .sql + .as_str() + .contains("CREATE TABLE communities")); + assert!(migrations[0].sql.as_str().contains("CREATE TABLE channels")); + assert!(migrations[0] + .sql + .as_str() + .contains("CREATE TABLE scheduled_workflow_fires")); + assert!(migrations[0] + .sql + .as_str() + .contains("CREATE TABLE audit_log")); + assert!(migrations[0] + .sql + .as_str() + .contains("CREATE TABLE _operator_global_tables")); + assert!(migrations[0] + .sql + .as_str() + .contains("search_tsv TSVECTOR GENERATED ALWAYS")); + } - assert_eq!(migrations[1].version, 2); - assert_eq!(&*migrations[1].description, "backfill d tag"); - assert!( - migrations[1].sql.as_str().contains("UPDATE events"), - "second migration should backfill existing event rows" + #[test] + fn migration_lint_detects_tables_missing_community_id_by_default() { + let sql = r#" + CREATE TABLE communities (id UUID PRIMARY KEY); + CREATE TABLE widgets (id UUID PRIMARY KEY); + CREATE TABLE _operator_global_tables (table_name TEXT PRIMARY KEY, reason TEXT NOT NULL); + INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('communities', 'tenant registry'), + ('_operator_global_tables', 'registry'); + "#; + + let definitions = create_table_definitions(sql); + let scoped = scoped_tables(sql); + let missing = definitions + .into_iter() + .filter(|(table, _)| scoped.contains(table)) + .filter(|(_, definitions)| !table_has_not_null_community_id(definitions)) + .map(|(table, _)| table) + .collect::>(); + + assert_eq!(missing, vec!["widgets"]); + } + + #[test] + fn migration_lint_detects_scoped_key_constraints_not_led_by_community_id() { + let sql = r#" + CREATE TABLE widgets ( + community_id UUID NOT NULL, + id UUID PRIMARY KEY, + channel_id UUID REFERENCES channels(id), + slug TEXT, + CONSTRAINT widgets_name_unique UNIQUE (slug), + CONSTRAINT widgets_parent_fk FOREIGN KEY (channel_id) REFERENCES channels(id) + ); + CREATE UNIQUE INDEX idx_widgets_slug ON widgets (slug); + ALTER TABLE widgets ADD CONSTRAINT widgets_alter_slug_unique UNIQUE (slug); + ALTER TABLE widgets ADD CONSTRAINT widgets_alter_parent_fk FOREIGN KEY (channel_id) REFERENCES channels(id); + CREATE TABLE _operator_global_tables (table_name TEXT PRIMARY KEY, reason TEXT NOT NULL); + INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('_operator_global_tables', 'registry'); + "#; + + let violations = scoped_constraint_violations(sql); + + assert!(violations + .iter() + .any(|violation| violation.kind == ConstraintKind::PrimaryKey)); + assert_eq!( + violations + .iter() + .filter(|violation| violation.kind == ConstraintKind::ForeignKey) + .count(), + 3 ); + assert_eq!( + violations + .iter() + .filter(|violation| violation.kind == ConstraintKind::Unique) + .count(), + 3 + ); + } + + #[test] + fn migration_lint_accepts_scoped_key_constraints_led_by_community_id() { + let sql = r#" + CREATE TABLE widgets ( + community_id UUID NOT NULL, + id UUID NOT NULL, + channel_id UUID NOT NULL, + slug TEXT NOT NULL, + PRIMARY KEY (community_id, id), + UNIQUE (community_id, slug), + FOREIGN KEY (community_id, channel_id) REFERENCES channels(community_id, id) + ); + CREATE UNIQUE INDEX idx_widgets_slug ON widgets (community_id, slug); + ALTER TABLE widgets ADD CONSTRAINT widgets_alter_slug_unique UNIQUE (community_id, slug); + ALTER TABLE widgets ADD CONSTRAINT widgets_alter_parent_fk FOREIGN KEY (community_id, channel_id) REFERENCES channels(community_id, id); + CREATE TABLE _operator_global_tables (table_name TEXT PRIMARY KEY, reason TEXT NOT NULL); + INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('_operator_global_tables', 'registry'); + "#; + + assert!(scoped_constraint_violations(sql).is_empty()); + } + + #[test] + fn all_non_operator_global_tables_have_not_null_community_id() { + let sql = migration_sql(); + let scoped = scoped_tables(sql); + let missing = create_table_definitions(sql) + .into_iter() + .filter(|(table, _)| scoped.contains(table)) + .filter(|(_, definitions)| !table_has_not_null_community_id(definitions)) + .map(|(table, _)| table) + .collect::>(); - assert_eq!(migrations[2].version, 3); - assert_eq!(&*migrations[2].description, "event reminders"); assert!( - migrations[2] - .sql - .as_str() - .contains("ADD COLUMN not_before BIGINT") - && migrations[2].sql.as_str().contains("idx_events_not_before"), - "third migration should add the NIP-ER reminder columns and index" + missing.is_empty(), + "every table not listed in _operator_global_tables must carry NOT NULL community_id; missing: {}", + missing.join(", ") + ); + } + + #[test] + fn scoped_primary_key_unique_and_foreign_key_constraints_lead_with_community_id() { + let sql = migration_sql(); + let violations = scoped_constraint_violations(sql) + .into_iter() + .map(|constraint| { + format!( + "{}. {:?} constraint must lead with community_id: {}", + constraint.table, constraint.kind, constraint.description + ) + }) + .collect::>(); + + assert!( + violations.is_empty(), + "tenant-scoped tables are all tables not listed in _operator_global_tables; primary key, unique/FK constraints, and unique indexes on those tables must lead with community_id:\n{}", + violations.join("\n") + ); + } + + #[test] + fn channels_community_id_is_immutable_after_insert() { + let sql = migration_sql(); + let forbidden_mutations = forbidden_channels_community_id_mutations(sql); + + assert!( + forbidden_mutations.is_empty(), + "channels.community_id must not be re-tenanted after insert; forbidden migration statements:\n{}", + forbidden_mutations.join("\n---\n") + ); + assert!( + has_channels_community_id_immutability_guard(sql), + "migrations define channels.community_id but no BEFORE UPDATE trigger/function guard that rejects OLD.community_id <> NEW.community_id was found" ); } @@ -192,88 +656,35 @@ mod tests { .expect("read applied migrations") } - /// Returns `schema/schema.sql` with the NIP-ER reminder DDL removed, so it - /// models a pre-stack deployment whose `events` table lacks the reminder - /// columns and index. The strip is asserted: if the snapshot text drifts so - /// these fragments no longer match, the test fails loudly rather than - /// silently loading a snapshot that already carries the reminder columns - /// (which would make migration 0003 collide on re-add). - fn pre_reminder_schema_snapshot() -> String { - const REMINDER_COLUMNS: &str = " not_before BIGINT,\n delivered_at BIGINT,\n"; - const REMINDER_INDEX: &str = "CREATE INDEX idx_events_not_before ON events (not_before)\n WHERE not_before IS NOT NULL AND deleted_at IS NULL AND delivered_at IS NULL;\n"; - - assert!( - SCHEMA_SQL.contains(REMINDER_COLUMNS) && SCHEMA_SQL.contains(REMINDER_INDEX), - "schema.sql reminder DDL drifted; update pre_reminder_schema_snapshot to match" - ); - - SCHEMA_SQL - .replace(REMINDER_COLUMNS, "") - .replace(REMINDER_INDEX, "") - } - #[tokio::test] #[ignore = "requires Postgres"] - async fn run_migrations_applies_embedded_versions_on_fresh_database() { + async fn run_migrations_applies_consolidated_initial_schema_on_fresh_database() { let pool = connect_test_pool().await; reset_public_schema(&pool).await; run_migrations(&pool).await.expect("run migrations"); - assert_eq!(applied_versions(&pool).await, vec![1, 2, 3]); - let events_exists = sqlx::query_scalar::<_, bool>( - "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'events')", - ) - .fetch_one(&pool) - .await - .expect("check events table"); - assert!(events_exists); - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn run_migrations_baselines_existing_schema_and_preserves_allowlist_backfill_path() { - let pool = connect_test_pool().await; - reset_public_schema(&pool).await; - // Load a pre-stack snapshot (without the NIP-ER reminder DDL) so the - // events table matches a real pre-SQLx deployment, which never had the - // reminder columns. Migration 0003 must then add them — proving the - // genuine prod-upgrade path, not a snapshot that already carries them. - sqlx::raw_sql(sqlx::AssertSqlSafe(pre_reminder_schema_snapshot())) - .execute(&pool) - .await - .expect("load pre-SQLx schema snapshot"); - sqlx::query( - "INSERT INTO pubkey_allowlist (pubkey, added_at) VALUES (decode($1, 'hex'), now())", - ) - .bind("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") - .execute(&pool) - .await - .expect("seed legacy allowlist row"); - - run_migrations(&pool).await.expect("baseline migrations"); - - assert_eq!(applied_versions(&pool).await, vec![1, 2, 3]); - let allowlist_count = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM pubkey_allowlist") + assert_eq!(applied_versions(&pool).await, vec![1]); + let tables = create_tables(migration_sql()); + for table in [ + "communities", + "events", + "channels", + "scheduled_workflow_fires", + "audit_log", + ] { + let exists = sqlx::query_scalar::<_, bool>( + "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = $1)", + ) + .bind(table) .fetch_one(&pool) .await - .expect("count allowlist rows"); - assert_eq!( - allowlist_count, 1, - "baseline must not drop legacy allowlist rows before relay startup backfills them" - ); - - let inserted = crate::relay_members::backfill_from_allowlist(&pool) - .await - .expect("backfill legacy allowlist rows"); - assert_eq!(inserted, 1); - let relay_member_count = sqlx::query_scalar::<_, i64>( - "SELECT COUNT(*) FROM relay_members WHERE pubkey = $1 AND role = 'member'", - ) - .bind("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") - .fetch_one(&pool) - .await - .expect("count backfilled relay member"); - assert_eq!(relay_member_count, 1); + .unwrap_or_else(|err| panic!("check table {table}: {err}")); + assert!( + tables.contains(table), + "migration parser should see {table}" + ); + assert!(exists, "migration should create {table}"); + } } } diff --git a/crates/buzz-db/src/thread.rs b/crates/buzz-db/src/thread.rs index 4ccd8a248..2c7b07912 100644 --- a/crates/buzz-db/src/thread.rs +++ b/crates/buzz-db/src/thread.rs @@ -659,7 +659,7 @@ pub async fn get_thread_metadata_by_event( mod tests { use super::*; use crate::{ - channel::{create_channel, ChannelType, ChannelVisibility}, + channel::{ChannelType, ChannelVisibility}, event::{insert_event_with_thread_metadata, ThreadMetadataParams}, }; use nostr::{EventBuilder, Keys, Kind}; @@ -687,12 +687,76 @@ mod tests { .expect("event timestamp is valid") } + async fn make_test_community(pool: &PgPool) -> Uuid { + let id = Uuid::new_v4(); + let host = format!("thread-test-{}.example", id.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(host) + .execute(pool) + .await + .expect("insert test community"); + id + } + + async fn create_test_channel( + pool: &PgPool, + name: &str, + channel_type: ChannelType, + visibility: ChannelVisibility, + description: Option<&str>, + created_by: &[u8], + ttl_seconds: Option, + ) -> crate::error::Result<(crate::channel::ChannelRecord, buzz_core::CommunityId)> { + let id = Uuid::new_v4(); + let community_id = make_test_community(pool).await; + + sqlx::query( + r#" + INSERT INTO channels + (id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline) + VALUES + ($1, $2, $3, $4::channel_type, $5::channel_visibility, $6, $7, $8, + CASE WHEN $8 IS NOT NULL THEN NOW() + ($8 || ' seconds')::interval ELSE NULL END) + "#, + ) + .bind(id) + .bind(community_id) + .bind(name) + .bind(channel_type.as_str()) + .bind(visibility.as_str()) + .bind(description) + .bind(created_by) + .bind(ttl_seconds) + .execute(pool) + .await + .expect("insert test channel"); + + sqlx::query( + r#" + INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) + VALUES ($1, $2, $3, 'owner', $4) + "#, + ) + .bind(community_id) + .bind(id) + .bind(created_by) + .bind(created_by) + .execute(pool) + .await + .expect("insert owner membership"); + + crate::channel::get_channel(pool, id) + .await + .map(|channel| (channel, buzz_core::CommunityId::from_uuid(community_id))) + } + #[tokio::test] #[ignore = "requires Postgres"] async fn get_thread_replies_reconstructs_stored_events() { let pool = setup_pool().await; let author = Keys::generate(); - let channel = create_channel( + let (channel, community) = create_test_channel( &pool, &format!("thread-replies-{}", Uuid::new_v4()), ChannelType::Stream, @@ -706,7 +770,7 @@ mod tests { let root = make_stream_event(&author, "root"); let root_created_at = event_created_at(&root); - insert_event_with_thread_metadata(&pool, &root, Some(channel.id), None) + insert_event_with_thread_metadata(&pool, community, &root, Some(channel.id), None) .await .expect("insert root event"); @@ -715,6 +779,7 @@ mod tests { let reply_id = reply.id.to_hex(); insert_event_with_thread_metadata( &pool, + community, &reply, Some(channel.id), Some(ThreadMetadataParams { @@ -752,7 +817,7 @@ mod tests { async fn get_thread_replies_skips_unreconstructable_row() { let pool = setup_pool().await; let author = Keys::generate(); - let channel = create_channel( + let (channel, community) = create_test_channel( &pool, &format!("thread-replies-corrupt-{}", Uuid::new_v4()), ChannelType::Stream, @@ -766,7 +831,7 @@ mod tests { let root = make_stream_event(&author, "root"); let root_created_at = event_created_at(&root); - insert_event_with_thread_metadata(&pool, &root, Some(channel.id), None) + insert_event_with_thread_metadata(&pool, community, &root, Some(channel.id), None) .await .expect("insert root event"); @@ -776,6 +841,7 @@ mod tests { let good_created_at = event_created_at(&good); insert_event_with_thread_metadata( &pool, + community, &good, Some(channel.id), Some(ThreadMetadataParams { @@ -797,6 +863,7 @@ mod tests { let bad_created_at = event_created_at(&bad); insert_event_with_thread_metadata( &pool, + community, &bad, Some(channel.id), Some(ThreadMetadataParams { diff --git a/crates/buzz-db/src/user.rs b/crates/buzz-db/src/user.rs index 1e1eaf42a..d4c9395bc 100644 --- a/crates/buzz-db/src/user.rs +++ b/crates/buzz-db/src/user.rs @@ -1,6 +1,7 @@ //! User CRUD operations. use crate::error::Result; +use buzz_core::CommunityId; use sqlx::PgPool; use sqlx::Row; @@ -34,14 +35,15 @@ pub struct UserSearchProfile { /// Ensure a user record exists for the given pubkey (upsert). /// Creates with minimal fields if not present; no-op if already exists. -pub async fn ensure_user(pool: &PgPool, pubkey: &[u8]) -> Result<()> { +pub async fn ensure_user(pool: &PgPool, community_id: CommunityId, pubkey: &[u8]) -> Result<()> { sqlx::query( r#" - INSERT INTO users (pubkey) - VALUES ($1) + INSERT INTO users (community_id, pubkey) + VALUES ($1, $2) ON CONFLICT DO NOTHING "#, ) + .bind(community_id.as_uuid()) .bind(pubkey) .execute(pool) .await?; @@ -49,7 +51,11 @@ pub async fn ensure_user(pool: &PgPool, pubkey: &[u8]) -> Result<()> { } /// Get a single user record by pubkey. -pub async fn get_user(pool: &PgPool, pubkey: &[u8]) -> Result> { +pub async fn get_user( + pool: &PgPool, + community_id: CommunityId, + pubkey: &[u8], +) -> Result> { let row = sqlx::query_as::< _, ( @@ -63,9 +69,10 @@ pub async fn get_user(pool: &PgPool, pubkey: &[u8]) -> Result Result, avatar_url: Option<&str>, @@ -129,8 +137,9 @@ pub async fn update_user_profile( } let sql = format!( - "UPDATE users SET {} WHERE pubkey = ${param_idx}", - set_parts.join(", ") + "UPDATE users SET {} WHERE community_id = ${param_idx} AND pubkey = ${}", + set_parts.join(", "), + param_idx + 1 ); let mut query = sqlx::query(sqlx::AssertSqlSafe(sql)); if display_name.is_some() { @@ -145,6 +154,7 @@ pub async fn update_user_profile( if nip05_handle.is_some() { query = query.bind(empty_to_none(nip05_handle)); } + query = query.bind(community_id.as_uuid()); query = query.bind(pubkey); query.execute(pool).await?; Ok(()) @@ -154,6 +164,7 @@ pub async fn update_user_profile( /// Both `local_part` and `domain` must already be lowercased by the caller. pub async fn get_user_by_nip05( pool: &PgPool, + community_id: CommunityId, local_part: &str, domain: &str, ) -> Result> { @@ -171,10 +182,11 @@ pub async fn get_user_by_nip05( r#" SELECT pubkey, display_name, avatar_url, about, nip05_handle FROM users - WHERE LOWER(nip05_handle) = LOWER($1) + WHERE community_id = $1 AND LOWER(nip05_handle) = LOWER($2) LIMIT 1 "#, ) + .bind(community_id.as_uuid()) .bind(&handle) .fetch_optional(pool) .await?; @@ -207,6 +219,7 @@ fn escape_like(input: &str) -> String { /// Empty queries return an empty vec and do not hit the database. pub async fn search_users( pool: &PgPool, + community_id: CommunityId, query: &str, limit: u32, ) -> Result> { @@ -224,23 +237,25 @@ pub async fn search_users( r#" SELECT pubkey, display_name, avatar_url, nip05_handle FROM users - WHERE LOWER(COALESCE(display_name, '')) LIKE $1 ESCAPE '\' - OR LOWER(COALESCE(nip05_handle, '')) LIKE $1 ESCAPE '\' - OR LOWER(encode(pubkey, 'hex')) LIKE $1 ESCAPE '\' + WHERE community_id = $1 + AND (LOWER(COALESCE(display_name, '')) LIKE $2 ESCAPE '\' + OR LOWER(COALESCE(nip05_handle, '')) LIKE $2 ESCAPE '\' + OR LOWER(encode(pubkey, 'hex')) LIKE $2 ESCAPE '\') ORDER BY CASE - WHEN LOWER(COALESCE(display_name, '')) = $2 THEN 0 - WHEN LOWER(COALESCE(nip05_handle, '')) = $2 THEN 1 - WHEN LOWER(encode(pubkey, 'hex')) = $2 THEN 2 - WHEN LOWER(COALESCE(display_name, '')) LIKE $3 ESCAPE '\' THEN 3 - WHEN LOWER(COALESCE(nip05_handle, '')) LIKE $3 ESCAPE '\' THEN 4 - WHEN LOWER(encode(pubkey, 'hex')) LIKE $3 ESCAPE '\' THEN 5 + WHEN LOWER(COALESCE(display_name, '')) = $3 THEN 0 + WHEN LOWER(COALESCE(nip05_handle, '')) = $3 THEN 1 + WHEN LOWER(encode(pubkey, 'hex')) = $3 THEN 2 + WHEN LOWER(COALESCE(display_name, '')) LIKE $4 ESCAPE '\' THEN 3 + WHEN LOWER(COALESCE(nip05_handle, '')) LIKE $4 ESCAPE '\' THEN 4 + WHEN LOWER(encode(pubkey, 'hex')) LIKE $4 ESCAPE '\' THEN 5 ELSE 6 END, COALESCE(NULLIF(display_name, ''), NULLIF(nip05_handle, ''), LOWER(encode(pubkey, 'hex'))) - LIMIT $4 + LIMIT $5 "#, ) + .bind(community_id.as_uuid()) .bind(&contains_pattern) .bind(&normalized) .bind(&prefix_pattern) @@ -271,15 +286,17 @@ pub async fn search_users( /// agent pubkey doesn't exist in the users table. pub async fn set_agent_owner( pool: &PgPool, + community_id: CommunityId, agent_pubkey: &[u8], owner_pubkey: &[u8], ) -> Result { // Conditional UPDATE: only set owner if currently NULL. This makes // "first mint wins" atomic — no TOCTOU race between concurrent mints. let result = sqlx::query( - r#"UPDATE users SET agent_owner_pubkey = $1 WHERE pubkey = $2 AND agent_owner_pubkey IS NULL"#, + r#"UPDATE users SET agent_owner_pubkey = $1 WHERE community_id = $2 AND pubkey = $3 AND agent_owner_pubkey IS NULL"#, ) .bind(owner_pubkey) + .bind(community_id.as_uuid()) .bind(agent_pubkey) .execute(pool) .await?; @@ -287,7 +304,8 @@ pub async fn set_agent_owner( if result.rows_affected() == 0 { // Could be: (a) pubkey not found, or (b) owner already set. // Check which case by querying the row. - let exists = sqlx::query(r#"SELECT 1 FROM users WHERE pubkey = $1"#) + let exists = sqlx::query(r#"SELECT 1 FROM users WHERE community_id = $1 AND pubkey = $2"#) + .bind(community_id.as_uuid()) .bind(agent_pubkey) .fetch_optional(pool) .await?; @@ -307,11 +325,13 @@ pub async fn set_agent_owner( /// Returns Some((policy_str, owner_bytes_or_none)) if found. pub async fn get_agent_channel_policy( pool: &PgPool, + community_id: CommunityId, pubkey: &[u8], ) -> Result>)>> { let row = sqlx::query( - r#"SELECT channel_add_policy::text AS channel_add_policy, agent_owner_pubkey FROM users WHERE pubkey = $1"#, + r#"SELECT channel_add_policy::text AS channel_add_policy, agent_owner_pubkey FROM users WHERE community_id = $1 AND pubkey = $2"#, ) + .bind(community_id.as_uuid()) .bind(pubkey) .fetch_optional(pool) .await?; @@ -329,12 +349,14 @@ pub async fn get_agent_channel_policy( /// `get_agent_channel_policy`, which would fetch unrelated fields. pub async fn is_agent_owner( pool: &PgPool, + community_id: CommunityId, target_pubkey: &[u8], actor_pubkey: &[u8], ) -> Result { let row = sqlx::query_scalar::<_, bool>( - "SELECT agent_owner_pubkey = $2 FROM users WHERE pubkey = $1 AND agent_owner_pubkey IS NOT NULL", + "SELECT agent_owner_pubkey = $3 FROM users WHERE community_id = $1 AND pubkey = $2 AND agent_owner_pubkey IS NOT NULL", ) + .bind(community_id.as_uuid()) .bind(target_pubkey) .bind(actor_pubkey) .fetch_optional(pool) @@ -345,16 +367,22 @@ pub async fn is_agent_owner( /// Set the channel_add_policy for a user. /// Returns an error if the pubkey is not found (rows_affected == 0). /// Returns an error if `policy` is not one of the valid ENUM values. -pub async fn set_channel_add_policy(pool: &PgPool, pubkey: &[u8], policy: &str) -> Result<()> { +pub async fn set_channel_add_policy( + pool: &PgPool, + community_id: CommunityId, + pubkey: &[u8], + policy: &str, +) -> Result<()> { if !matches!(policy, "anyone" | "owner_only" | "nobody") { return Err(crate::error::DbError::InvalidData(format!( "invalid channel_add_policy: {policy}" ))); } let result = sqlx::query( - r#"UPDATE users SET channel_add_policy = $1::channel_add_policy WHERE pubkey = $2"#, + r#"UPDATE users SET channel_add_policy = $1::channel_add_policy WHERE community_id = $2 AND pubkey = $3"#, ) .bind(policy) + .bind(community_id.as_uuid()) .bind(pubkey) .execute(pool) .await?; @@ -385,28 +413,41 @@ mod tests { Keys::generate().public_key().to_bytes().to_vec() } + async fn make_community(pool: &PgPool) -> CommunityId { + let id = uuid::Uuid::new_v4(); + let host = format!("user-test-{}.example", id.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(host) + .execute(pool) + .await + .expect("insert test community"); + CommunityId::from_uuid(id) + } + /// Setting an agent owner then reading back the policy should return /// the default "anyone" policy and the owner pubkey. #[tokio::test] #[ignore = "requires Postgres"] async fn test_set_agent_owner_and_get_policy() { let db = setup_db().await; + let community = make_community(&db.pool).await; let agent_pk = random_pubkey(); let owner_pk = random_pubkey(); - ensure_user(&db.pool, &agent_pk) + ensure_user(&db.pool, community, &agent_pk) .await .expect("ensure agent"); - ensure_user(&db.pool, &owner_pk) + ensure_user(&db.pool, community, &owner_pk) .await .expect("ensure owner"); - let was_set = set_agent_owner(&db.pool, &agent_pk, &owner_pk) + let was_set = set_agent_owner(&db.pool, community, &agent_pk, &owner_pk) .await .expect("set_agent_owner"); assert!(was_set, "first set_agent_owner should return true"); - let result = get_agent_channel_policy(&db.pool, &agent_pk) + let result = get_agent_channel_policy(&db.pool, community, &agent_pk) .await .expect("get_agent_channel_policy"); @@ -424,14 +465,17 @@ mod tests { #[ignore = "requires Postgres"] async fn test_set_channel_add_policy() { let db = setup_db().await; + let community = make_community(&db.pool).await; let pk = random_pubkey(); - ensure_user(&db.pool, &pk).await.expect("ensure user"); + ensure_user(&db.pool, community, &pk) + .await + .expect("ensure user"); // owner_only - set_channel_add_policy(&db.pool, &pk, "owner_only") + set_channel_add_policy(&db.pool, community, &pk, "owner_only") .await .expect("set owner_only"); - let (policy, owner) = get_agent_channel_policy(&db.pool, &pk) + let (policy, owner) = get_agent_channel_policy(&db.pool, community, &pk) .await .expect("get policy") .expect("should be Some"); @@ -439,10 +483,10 @@ mod tests { assert!(owner.is_none(), "no owner was set"); // nobody - set_channel_add_policy(&db.pool, &pk, "nobody") + set_channel_add_policy(&db.pool, community, &pk, "nobody") .await .expect("set nobody"); - let (policy, owner) = get_agent_channel_policy(&db.pool, &pk) + let (policy, owner) = get_agent_channel_policy(&db.pool, community, &pk) .await .expect("get policy") .expect("should be Some"); @@ -450,10 +494,10 @@ mod tests { assert!(owner.is_none()); // anyone (reset to default) - set_channel_add_policy(&db.pool, &pk, "anyone") + set_channel_add_policy(&db.pool, community, &pk, "anyone") .await .expect("set anyone"); - let (policy, owner) = get_agent_channel_policy(&db.pool, &pk) + let (policy, owner) = get_agent_channel_policy(&db.pool, community, &pk) .await .expect("get policy") .expect("should be Some"); @@ -467,9 +511,10 @@ mod tests { #[ignore = "requires Postgres"] async fn test_get_policy_unknown_pubkey() { let db = setup_db().await; + let community = make_community(&db.pool).await; let pk = random_pubkey(); - let result = get_agent_channel_policy(&db.pool, &pk) + let result = get_agent_channel_policy(&db.pool, community, &pk) .await .expect("query should not error"); @@ -482,15 +527,16 @@ mod tests { #[ignore = "requires Postgres"] async fn test_set_agent_owner_nonexistent_agent() { let db = setup_db().await; + let community = make_community(&db.pool).await; let agent_pk = random_pubkey(); let owner_pk = random_pubkey(); // Only ensure the owner exists -- agent is intentionally absent. - ensure_user(&db.pool, &owner_pk) + ensure_user(&db.pool, community, &owner_pk) .await .expect("ensure owner"); - let result = set_agent_owner(&db.pool, &agent_pk, &owner_pk).await; + let result = set_agent_owner(&db.pool, community, &agent_pk, &owner_pk).await; assert!( result.is_err(), "should error when agent pubkey is not in users table" @@ -502,28 +548,33 @@ mod tests { #[ignore = "requires Postgres"] async fn test_set_agent_owner_already_owned() { let db = setup_db().await; + let community = make_community(&db.pool).await; let agent_pk = random_pubkey(); let owner1 = random_pubkey(); let owner2 = random_pubkey(); - ensure_user(&db.pool, &agent_pk) + ensure_user(&db.pool, community, &agent_pk) .await .expect("ensure agent"); - ensure_user(&db.pool, &owner1).await.expect("ensure owner1"); - ensure_user(&db.pool, &owner2).await.expect("ensure owner2"); + ensure_user(&db.pool, community, &owner1) + .await + .expect("ensure owner1"); + ensure_user(&db.pool, community, &owner2) + .await + .expect("ensure owner2"); - let first = set_agent_owner(&db.pool, &agent_pk, &owner1) + let first = set_agent_owner(&db.pool, community, &agent_pk, &owner1) .await .expect("first set"); assert!(first, "first set should succeed"); - let second = set_agent_owner(&db.pool, &agent_pk, &owner2) + let second = set_agent_owner(&db.pool, community, &agent_pk, &owner2) .await .expect("second set should not error"); assert!(!second, "second set should return false (already owned)"); // Verify original owner is preserved. - let (_, owner) = get_agent_channel_policy(&db.pool, &agent_pk) + let (_, owner) = get_agent_channel_policy(&db.pool, community, &agent_pk) .await .expect("get policy") .expect("should be Some"); @@ -536,9 +587,10 @@ mod tests { #[ignore = "requires Postgres"] async fn test_set_channel_add_policy_nonexistent_user() { let db = setup_db().await; + let community = make_community(&db.pool).await; let pk = random_pubkey(); - let result = set_channel_add_policy(&db.pool, &pk, "nobody").await; + let result = set_channel_add_policy(&db.pool, community, &pk, "nobody").await; assert!( result.is_err(), "should error when pubkey is not in users table" @@ -549,9 +601,10 @@ mod tests { #[ignore = "requires Postgres"] async fn test_set_channel_add_policy_rejects_invalid() { let db = setup_db().await; + let community = make_community(&db.pool).await; let pubkey = nostr::Keys::generate().public_key().to_bytes().to_vec(); - ensure_user(&db.pool, &pubkey).await.unwrap(); - let result = set_channel_add_policy(&db.pool, &pubkey, "invalid_policy").await; + ensure_user(&db.pool, community, &pubkey).await.unwrap(); + let result = set_channel_add_policy(&db.pool, community, &pubkey, "invalid_policy").await; assert!(result.is_err(), "should reject invalid policy value"); } @@ -595,14 +648,17 @@ mod tests { #[ignore = "requires Postgres"] async fn test_owner_only_with_no_owner() { let db = setup_db().await; + let community = make_community(&db.pool).await; let pk = random_pubkey(); - ensure_user(&db.pool, &pk).await.expect("ensure user"); + ensure_user(&db.pool, community, &pk) + .await + .expect("ensure user"); - set_channel_add_policy(&db.pool, &pk, "owner_only") + set_channel_add_policy(&db.pool, community, &pk, "owner_only") .await .expect("set owner_only"); - let result = get_agent_channel_policy(&db.pool, &pk) + let result = get_agent_channel_policy(&db.pool, community, &pk) .await .expect("get policy") .expect("should be Some"); diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index 993dbf19d..f7d64bda3 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -15,6 +15,8 @@ use sha2::{Digest, Sha256}; use sqlx::{PgPool, Row}; use uuid::Uuid; +use buzz_core::CommunityId; + use crate::error::{DbError, Result}; // -- Token hashing ------------------------------------------------------------ @@ -163,6 +165,8 @@ impl FromStr for ApprovalStatus { pub struct WorkflowRecord { /// Unique workflow identifier. pub id: Uuid, + /// Server-resolved community that owns this workflow. + pub community_id: CommunityId, /// Human-readable workflow name. pub name: String, /// Compressed public key bytes of the workflow owner. @@ -211,6 +215,23 @@ pub struct WorkflowRunRecord { pub created_at: DateTime, } +/// A winning scheduled workflow fire claim. +/// +/// The primary identity is `(workflow_id, scheduled_for)`. `community_id` is +/// resolved from the workflow row inside the claim SQL and returned for scoped +/// audit/logging; callers never supply it as a claim. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ScheduledWorkflowFireClaim { + /// Community that owns this scheduled fire. + pub community_id: CommunityId, + /// Workflow definition that should run. + pub workflow_id: Uuid, + /// Authoritative schedule instant this claim represents. + pub scheduled_for: DateTime, + /// Database timestamp for when this pod won the claim. + pub claimed_at: DateTime, +} + /// A pending or resolved approval gate for a workflow step. #[derive(Debug, Clone)] pub struct ApprovalRecord { @@ -244,6 +265,7 @@ pub struct ApprovalRecord { /// New workflows start as `active` and `enabled = TRUE`. pub async fn create_workflow( pool: &PgPool, + community_id: CommunityId, channel_id: Option, owner_pubkey: &[u8], name: &str, @@ -255,11 +277,12 @@ pub async fn create_workflow( sqlx::query( r#" INSERT INTO workflows - (id, name, owner_pubkey, channel_id, definition, definition_hash, status, enabled) - VALUES ($1, $2, $3, $4, $5::jsonb, $6, 'active', TRUE) + (id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, status, enabled) + VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, 'active', TRUE) "#, ) .bind(id) + .bind(community_id.as_uuid()) .bind(name) .bind(owner_pubkey) .bind(channel_id) @@ -275,7 +298,7 @@ pub async fn create_workflow( pub async fn get_workflow(pool: &PgPool, id: Uuid) -> Result { let row = sqlx::query( r#" - SELECT id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE id = $1 @@ -304,7 +327,7 @@ pub async fn list_channel_workflows( let rows = sqlx::query( r#" - SELECT id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE channel_id = $1 @@ -333,7 +356,7 @@ pub async fn list_enabled_channel_workflows( ) -> Result> { let rows = sqlx::query( r#" - SELECT id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE channel_id = $1 @@ -359,7 +382,7 @@ pub async fn list_enabled_channel_workflows( pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result> { let rows = sqlx::query( r#" - SELECT id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE status = 'active' @@ -376,6 +399,125 @@ pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result, +) -> Result> { + let row = sqlx::query( + r#" + INSERT INTO scheduled_workflow_fires (community_id, workflow_id, scheduled_for) + SELECT w.community_id, w.id, $2 + FROM workflows w + WHERE w.id = $1 + ON CONFLICT (community_id, workflow_id, scheduled_for) DO NOTHING + RETURNING community_id, workflow_id, scheduled_for, claimed_at + "#, + ) + .bind(workflow_id) + .bind(scheduled_for) + .fetch_optional(pool) + .await?; + + row.map(|row| { + let community_id: Uuid = row.try_get("community_id")?; + Ok(ScheduledWorkflowFireClaim { + community_id: CommunityId::from_uuid(community_id), + workflow_id: row.try_get("workflow_id")?, + scheduled_for: row.try_get("scheduled_for")?, + claimed_at: row.try_get("claimed_at")?, + }) + }) + .transpose() +} + +/// Fetch the greatest claimed schedule instant for a workflow. +/// +/// Interval schedulers use this as their DB-authoritative `last_fired` anchor. +/// It makes all pods compute the same next interval instant after a successful +/// claim, and preserves the interval clock across pod restarts. This intentionally +/// reads from `scheduled_workflow_fires`, not `workflow_runs`, because the claim +/// row is the source of truth for schedule deduplication. +pub async fn latest_scheduled_workflow_fire( + pool: &PgPool, + workflow_id: Uuid, +) -> Result>> { + let row = sqlx::query( + r#" + SELECT MAX(scheduled_for) AS scheduled_for + FROM scheduled_workflow_fires + WHERE workflow_id = $1 + "#, + ) + .bind(workflow_id) + .fetch_one(pool) + .await?; + + row.try_get("scheduled_for").map_err(Into::into) +} + +/// Link a won scheduled-fire claim to the workflow run it created. +/// +/// This is for ops/audit forensics only; the claim row remains the dedupe +/// boundary. If run creation succeeds, callers should attach the run id before +/// spawning execution. If run creation fails, leaving `workflow_run_id` NULL is +/// intentional: the schedule instant was claimed and must not duplicate later. +pub async fn attach_scheduled_workflow_run( + pool: &PgPool, + workflow_id: Uuid, + scheduled_for: DateTime, + workflow_run_id: Uuid, +) -> Result { + let result = sqlx::query( + r#" + UPDATE scheduled_workflow_fires + SET workflow_run_id = $3 + WHERE workflow_id = $1 + AND scheduled_for = $2 + AND workflow_run_id IS NULL + "#, + ) + .bind(workflow_id) + .bind(scheduled_for) + .bind(workflow_run_id) + .execute(pool) + .await?; + + Ok(result.rows_affected() == 1) +} + +/// Delete old scheduled workflow fire claims for retention. +/// +/// Schedule claim rows are correctness metadata, but they grow with every fire. +/// The relay/ops janitor should retain enough history for audits and interval +/// anchoring: the cutoff must be older than the largest interval schedule the +/// deployment supports, or interval workflows can lose their DB-authoritative +/// anchor after pruning. +pub async fn prune_scheduled_workflow_fires_before( + pool: &PgPool, + older_than: DateTime, +) -> Result { + let result = sqlx::query( + r#" + DELETE FROM scheduled_workflow_fires + WHERE claimed_at < $1 + "#, + ) + .bind(older_than) + .execute(pool) + .await?; + + Ok(result.rows_affected()) +} + /// Update a workflow's name, definition, and definition_hash. pub async fn update_workflow( pool: &PgPool, @@ -765,8 +907,11 @@ fn row_to_workflow_record(row: sqlx::postgres::PgRow) -> Result let enabled: bool = row.try_get("enabled")?; + let community_id: Uuid = row.try_get("community_id")?; + Ok(WorkflowRecord { id, + community_id: CommunityId::from_uuid(community_id), name: row.try_get("name")?, owner_pubkey: row.try_get("owner_pubkey")?, channel_id, @@ -831,7 +976,7 @@ pub async fn find_by_owner_and_name( ) -> Result> { let row = sqlx::query( r#" - SELECT id, name, owner_pubkey, channel_id, definition, definition_hash, + SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, status::text AS status, enabled, created_at, updated_at FROM workflows WHERE owner_pubkey = $1 AND name = $2 @@ -955,8 +1100,11 @@ mod tests { "steps": [{ "id": "s1", "action": "send_message", "text": "hi" }] }); + let community_id = CommunityId::from_uuid(Uuid::new_v4()); + let record = WorkflowRecord { id, + community_id, name: "My Workflow".to_owned(), owner_pubkey: vec![0xab; 32], channel_id: Some(channel_id), @@ -969,6 +1117,7 @@ mod tests { }; assert_eq!(record.id, id); + assert_eq!(record.community_id, community_id); assert_eq!(record.name, "My Workflow"); assert_eq!(record.owner_pubkey, vec![0xab; 32]); assert_eq!(record.channel_id, Some(channel_id)); @@ -985,6 +1134,7 @@ mod tests { let record = WorkflowRecord { id, + community_id: CommunityId::from_uuid(Uuid::new_v4()), name: "Global Workflow".to_owned(), owner_pubkey: vec![0x00; 32], channel_id: None, @@ -1006,6 +1156,7 @@ mod tests { let record = WorkflowRecord { id, + community_id: CommunityId::from_uuid(Uuid::new_v4()), name: "Original".to_owned(), owner_pubkey: vec![0x01; 32], channel_id: None, @@ -1034,6 +1185,7 @@ mod tests { ] { let record = WorkflowRecord { id: Uuid::new_v4(), + community_id: CommunityId::from_uuid(Uuid::new_v4()), name: "Test".to_owned(), owner_pubkey: vec![], channel_id: None, @@ -1053,6 +1205,7 @@ mod tests { let now = Utc::now(); let record = WorkflowRecord { id: Uuid::new_v4(), + community_id: CommunityId::from_uuid(Uuid::new_v4()), name: "Paused".to_owned(), owner_pubkey: vec![], channel_id: None, @@ -1302,4 +1455,251 @@ mod tests { assert_eq!(record.status, ApprovalStatus::Pending); assert_eq!(cloned.status, ApprovalStatus::Granted); } + + // -- F1 / S1 attack surface for scheduled workflow claims ------------------ + // + // These tests pin the locked spec from Eva [13] / Mari [12]: + // + // 1. `workflows.community_id` is row-owned, NOT NULL, immutable. + // 2. Claim resolves `community_id` server-side via `workflow_id`. + // 3. Claim uniqueness is `(workflow_id, scheduled_for)` — `workflow_id` + // is globally unique, so adding `community_id` to the key weakens it. + // 4. `latest_scheduled_workflow_fire` drops caller-supplied community. + // + // Until that lands, `claim_for_workflow_in_other_community_no_ops` is the + // S1 regression lock: today the schema permits a caller in community A to + // claim a workflow owned by community B and have `claimed.community_id` + // come back as A. That is a cross-tenant write surface (Theorem S1, + // `docs/multi-tenant-relay.md:248`: "the claimed community never appears + // in this function — only the resolved one"). + // + // The other two tests are characterization guards: same-window race must + // yield exactly one claim winner, and the retention primitive's docstring + // caveat (pruning below the largest interval breaks `latest_*`) is + // load-bearing for the §5c deployment-config rule Sami flagged. + + use crate::user::ensure_user; + + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; + + async fn setup_pool() -> PgPool { + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_owned()); + + PgPool::connect(&database_url) + .await + .expect("connect to test DB") + } + + /// Insert a community with a unique host. Returns its `CommunityId`. + async fn make_community(pool: &PgPool) -> CommunityId { + let id = Uuid::new_v4(); + let host = format!("test-{}.example", id.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(&host) + .execute(pool) + .await + .expect("insert community"); + CommunityId::from_uuid(id) + } + + /// Insert a channel under a community. Returns the channel id. + async fn make_channel(pool: &PgPool, community: CommunityId, owner: &[u8]) -> Uuid { + let id = Uuid::new_v4(); + sqlx::query( + r#" + INSERT INTO channels (id, community_id, name, created_by) + VALUES ($1, $2, $3, $4) + "#, + ) + .bind(id) + .bind(community.as_uuid()) + .bind(format!("ch-{}", id.simple())) + .bind(owner) + .execute(pool) + .await + .expect("insert channel"); + id + } + + /// Insert a workflow whose tenant is `community`'s channel. Returns the + /// workflow id and the owning community for callers that want to assert + /// the resolved tenant. + async fn make_workflow_in(pool: &PgPool, community: CommunityId) -> (Uuid, CommunityId) { + let owner = vec![0xa1; 32]; + ensure_user(pool, community, &owner) + .await + .expect("ensure owner"); + let channel_id = make_channel(pool, community, &owner).await; + let workflow_id = create_workflow( + pool, + community, + Some(channel_id), + &owner, + "f1-attack-workflow", + r#"{"trigger":{"on":"schedule"},"steps":[]}"#, + &[0u8; 32], + ) + .await + .expect("create workflow"); + (workflow_id, community) + } + + /// F1 attack: a caller in community A must NOT be able to claim a fire + /// for a workflow owned by community B and have the claim resolve under + /// A's tenant. The resolved community on the returned claim row MUST + /// equal the workflow's actual tenant (B). + /// + /// Post-fix (`1fa3d837f`) the claim signature no longer accepts a caller + /// tenant — the SQL resolves `community_id` from the `workflows` row via + /// `INSERT ... SELECT w.community_id, w.id, $2 FROM workflows w WHERE + /// w.id = $1`. This test still creates an "attacker_community" that the + /// caller is *not* permitted to name on the wire; the assertion is that + /// the resolved tenant equals the workflow owner's community, never the + /// attacker's. With the pre-fix signature this test was RED (the row's + /// `community_id` came back as the attacker's); under the locked spec it + /// must be GREEN. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn claim_for_workflow_in_other_community_no_ops() { + let pool = setup_pool().await; + + // The attacker community exists in the schema but the caller has no + // way to pass it to the claim API anymore — that *is* the S1 fix. + // Keeping the row in this test makes the no-influence invariant + // explicit: even with two real tenants in play, the resolved + // community is the workflow's owner. + let _attacker_community = make_community(&pool).await; + let owner_community = make_community(&pool).await; + let (workflow_id, expected_community) = make_workflow_in(&pool, owner_community).await; + + let scheduled_for = Utc.with_ymd_and_hms(2026, 6, 27, 0, 0, 0).unwrap(); + + let claim = claim_scheduled_workflow_fire(&pool, workflow_id, scheduled_for) + .await + .expect("claim should not error") + .expect("claim should succeed exactly once"); + + assert_eq!( + claim.community_id, + expected_community, + "claim must resolve community from workflow_id (server-side); \ + resolved={resolved:?} expected={expected_community:?}", + resolved = claim.community_id, + ); + assert_eq!(claim.workflow_id, workflow_id); + assert_eq!(claim.scheduled_for, scheduled_for); + } + + /// Same `(workflow_id, scheduled_for)` claimed concurrently by N tasks + /// must yield exactly one `Some` winner. Post-fix the PK is + /// `(workflow_id, scheduled_for)` (`workflow_id` is globally unique, + /// `community_id` is a scoped/audit label only) — exactly the locked + /// spec. Characterization guard: protects the dedup boundary against + /// regressions in the claim SQL. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn concurrent_same_window_claims_exactly_one_wins() { + let pool = setup_pool().await; + + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let scheduled_for = Utc.with_ymd_and_hms(2026, 6, 27, 0, 1, 0).unwrap(); + + const N: usize = 8; + let mut handles = Vec::with_capacity(N); + for _ in 0..N { + let pool = pool.clone(); + handles.push(tokio::spawn(async move { + claim_scheduled_workflow_fire(&pool, workflow_id, scheduled_for).await + })); + } + + let mut winners = 0usize; + for h in handles { + let result = h.await.expect("task did not panic").expect("claim ok"); + if result.is_some() { + winners += 1; + } + } + assert_eq!( + winners, 1, + "exactly one task must win the claim race for (workflow_id, scheduled_for)" + ); + } + + /// Documents the retention-vs-interval coupling Sami flagged for §5c: + /// pruning every claim below the workflow's interval makes + /// `latest_scheduled_workflow_fire` return `None`, which re-introduces the + /// per-pod-clock anchor bug F5 was meant to fix. Test is GREEN today and + /// MUST stay green — it pins the deployment-config rule that the janitor + /// cutoff must exceed `MAX(interval_secs) + safety margin`. If a future + /// change makes `latest_*` resilient to pruning (e.g. by reading the most + /// recent workflow_run instead, or by retaining a sentinel row), this + /// test's assertion encodes the contract that must be updated alongside. + /// + /// Test isolation: the prune primitive is global (filters only on + /// `claimed_at`), so to avoid colliding with parallel claim tests we + /// back-date this workflow's `claimed_at` into the deep past and use a + /// past cutoff that cannot match any other test's `claimed_at = NOW()`. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn latest_after_prune_below_interval_breaks_anchor() { + let pool = setup_pool().await; + + let community = make_community(&pool).await; + let (workflow_id, _) = make_workflow_in(&pool, community).await; + let scheduled_for = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap(); + + claim_scheduled_workflow_fire(&pool, workflow_id, scheduled_for) + .await + .expect("claim ok") + .expect("first claim wins"); + + // Backdate this row's `claimed_at` so the global prune below targets + // only this workflow's row and cannot race-delete other tests' rows. + let backdated_claimed_at = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap(); + sqlx::query( + "UPDATE scheduled_workflow_fires SET claimed_at = $1 \ + WHERE community_id = $2 AND workflow_id = $3 AND scheduled_for = $4", + ) + .bind(backdated_claimed_at) + .bind(community.as_uuid()) + .bind(workflow_id) + .bind(scheduled_for) + .execute(&pool) + .await + .expect("backdate ok"); + + let latest_before = latest_scheduled_workflow_fire(&pool, workflow_id) + .await + .expect("latest ok"); + assert_eq!( + latest_before, + Some(scheduled_for), + "latest must reflect the claim before pruning", + ); + + // Janitor cutoff above only the back-dated row: prunes the anchor row + // without touching anything claimed at wall-clock NOW. + let cutoff = backdated_claimed_at + chrono::Duration::seconds(1); + let pruned = prune_scheduled_workflow_fires_before(&pool, cutoff) + .await + .expect("prune ok"); + assert!( + pruned >= 1, + "expected at least one row pruned, got {pruned}" + ); + + let latest_after = latest_scheduled_workflow_fire(&pool, workflow_id) + .await + .expect("latest ok"); + assert_eq!( + latest_after, None, + "pruning below the largest interval breaks the DB anchor; \ + retention cutoff MUST exceed MAX(interval_secs) + safety margin (§5c)", + ); + } } diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index eb36f61be..c6b9f972e 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -612,10 +612,20 @@ async fn handle_workflow_def( PersistResult::Inserted(tx) => tx, }; - // 4. Execute: create_workflow + // 4. Execute: create_workflow. The workflow's community is resolved from + // the server-owned channel row, not from the client-supplied event. The DB + // also enforces `(community_id, channel_id)` as a composite FK. + let community_id = state + .db + .community_of_channel(channel_id) + .await + .map_err(|e| IngestError::Internal(format!("error: db channel community lookup: {e}")))? + .ok_or_else(|| IngestError::Rejected("invalid: workflow channel not found".into()))?; + let workflow_id = state .db .create_workflow( + community_id, Some(channel_id), &self_bytes, &workflow_name,