diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 04a4dfd8d..b6e7f7e0a 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -300,7 +300,7 @@ impl Db { r#" INSERT INTO communities (host) VALUES ($1) - ON CONFLICT (host) DO UPDATE SET host = EXCLUDED.host + ON CONFLICT (lower(host)) DO UPDATE SET host = EXCLUDED.host RETURNING id, host "#, ) @@ -1154,6 +1154,7 @@ impl Db { /// Add (or re-activate) a reaction. pub async fn add_reaction( &self, + community: CommunityId, event_id: &[u8], event_created_at: DateTime, pubkey: &[u8], @@ -1162,6 +1163,7 @@ impl Db { ) -> Result { reaction::add_reaction( &self.pool, + community, event_id, event_created_at, pubkey, @@ -1174,37 +1176,56 @@ impl Db { /// Soft-delete a reaction. pub async fn remove_reaction( &self, + community: CommunityId, event_id: &[u8], event_created_at: DateTime, pubkey: &[u8], emoji: &str, ) -> Result { - reaction::remove_reaction(&self.pool, event_id, event_created_at, pubkey, emoji).await + reaction::remove_reaction( + &self.pool, + community, + event_id, + event_created_at, + pubkey, + emoji, + ) + .await } /// Soft-delete a reaction by its source event ID. pub async fn remove_reaction_by_source_event_id( &self, + community: CommunityId, reaction_event_id: &[u8], ) -> Result { - reaction::remove_reaction_by_source_event_id(&self.pool, reaction_event_id).await + reaction::remove_reaction_by_source_event_id(&self.pool, community, reaction_event_id).await } /// Look up the active reaction row for one actor + emoji + target tuple. pub async fn get_active_reaction_record( &self, + community: CommunityId, event_id: &[u8], event_created_at: DateTime, pubkey: &[u8], emoji: &str, ) -> Result> { - reaction::get_active_reaction_record(&self.pool, event_id, event_created_at, pubkey, emoji) - .await + reaction::get_active_reaction_record( + &self.pool, + community, + event_id, + event_created_at, + pubkey, + emoji, + ) + .await } /// Backfill the source event ID on an active reaction row. pub async fn set_reaction_event_id( &self, + community: CommunityId, event_id: &[u8], event_created_at: DateTime, pubkey: &[u8], @@ -1213,6 +1234,7 @@ impl Db { ) -> Result { reaction::set_reaction_event_id( &self.pool, + community, event_id, event_created_at, pubkey, @@ -1225,20 +1247,30 @@ impl Db { /// Get all active reactions for an event, grouped by emoji. pub async fn get_reactions( &self, + community: CommunityId, event_id: &[u8], event_created_at: DateTime, limit: u32, cursor: Option<&str>, ) -> Result> { - reaction::get_reactions(&self.pool, event_id, event_created_at, limit, cursor).await + reaction::get_reactions( + &self.pool, + community, + event_id, + event_created_at, + limit, + cursor, + ) + .await } /// Batch-fetch emoji counts for a set of (event_id, event_created_at) pairs. pub async fn get_reactions_bulk( &self, + community: CommunityId, event_ids: &[(&[u8], DateTime)], ) -> Result> { - reaction::get_reactions_bulk(&self.pool, event_ids).await + reaction::get_reactions_bulk(&self.pool, community, event_ids).await } /// Find events that @mention the given pubkey. @@ -1851,36 +1883,43 @@ impl Db { Ok(result.rows_affected()) } - /// Check if a pubkey is in the allowlist. - pub async fn is_pubkey_allowed(&self, pubkey: &[u8]) -> Result { - let row = sqlx::query("SELECT COUNT(*) as cnt FROM pubkey_allowlist WHERE pubkey = $1") - .bind(pubkey) - .fetch_one(&self.pool) - .await?; + /// Check if a pubkey is in the allowlist for `community`. + pub async fn is_pubkey_allowed(&self, community: CommunityId, pubkey: &[u8]) -> Result { + let row = sqlx::query( + "SELECT COUNT(*) as cnt FROM pubkey_allowlist WHERE community_id = $1 AND pubkey = $2", + ) + .bind(community.as_uuid()) + .bind(pubkey) + .fetch_one(&self.pool) + .await?; let cnt: i64 = row.try_get("cnt")?; Ok(cnt > 0) } - /// Check if the allowlist has any entries (i.e. is enforcement active). - pub async fn has_allowlist_entries(&self) -> Result { - let row = sqlx::query("SELECT COUNT(*) as cnt FROM pubkey_allowlist") - .fetch_one(&self.pool) - .await?; + /// Check if the community allowlist has any entries (i.e. is enforcement active). + pub async fn has_allowlist_entries(&self, community: CommunityId) -> Result { + let row = + sqlx::query("SELECT COUNT(*) as cnt FROM pubkey_allowlist WHERE community_id = $1") + .bind(community.as_uuid()) + .fetch_one(&self.pool) + .await?; let cnt: i64 = row.try_get("cnt")?; Ok(cnt > 0) } - /// Add a pubkey to the allowlist. + /// Add a pubkey to the community allowlist. pub async fn add_to_allowlist( &self, + community: CommunityId, pubkey: &[u8], added_by: &[u8], note: Option<&str>, ) -> Result { let result = sqlx::query( - "INSERT INTO pubkey_allowlist (pubkey, added_by, note) VALUES ($1, $2, $3) \ + "INSERT INTO pubkey_allowlist (community_id, pubkey, added_by, note) VALUES ($1, $2, $3, $4) \ ON CONFLICT DO NOTHING", ) + .bind(community.as_uuid()) .bind(pubkey) .bind(added_by) .bind(note) @@ -1889,20 +1928,27 @@ impl Db { Ok(result.rows_affected() > 0) } - /// Remove a pubkey from the allowlist. - pub async fn remove_from_allowlist(&self, pubkey: &[u8]) -> Result { - let result = sqlx::query("DELETE FROM pubkey_allowlist WHERE pubkey = $1") - .bind(pubkey) - .execute(&self.pool) - .await?; + /// Remove a pubkey from the community allowlist. + pub async fn remove_from_allowlist( + &self, + community: CommunityId, + pubkey: &[u8], + ) -> Result { + let result = + sqlx::query("DELETE FROM pubkey_allowlist WHERE community_id = $1 AND pubkey = $2") + .bind(community.as_uuid()) + .bind(pubkey) + .execute(&self.pool) + .await?; Ok(result.rows_affected() > 0) } - /// List all pubkeys in the allowlist. - pub async fn list_allowlist(&self) -> Result> { + /// List all pubkeys in the community allowlist. + pub async fn list_allowlist(&self, community: CommunityId) -> Result> { let rows = sqlx::query( - "SELECT pubkey, added_by, added_at, note FROM pubkey_allowlist ORDER BY added_at DESC", + "SELECT pubkey, added_by, added_at, note FROM pubkey_allowlist WHERE community_id = $1 ORDER BY added_at DESC", ) + .bind(community.as_uuid()) .fetch_all(&self.pool) .await?; @@ -2488,6 +2534,72 @@ mod tests { .expect("insert channel"); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn allowlist_is_scoped_to_community() { + let db = setup_db().await; + let community_a = CommunityId::from_uuid(make_community(&db.pool).await); + let community_b = CommunityId::from_uuid(make_community(&db.pool).await); + let pubkey = [7u8; 32]; + let added_by = [9u8; 32]; + + assert!(db + .add_to_allowlist(community_a, &pubkey, &added_by, Some("a-only")) + .await + .expect("add allowlist row")); + assert!(!db + .add_to_allowlist(community_a, &pubkey, &added_by, Some("duplicate")) + .await + .expect("duplicate allowlist row is idempotent")); + + assert!( + db.is_pubkey_allowed(community_a, &pubkey) + .await + .expect("allowlist check A"), + "pubkey added to A must be allowed in A" + ); + assert!( + !db.is_pubkey_allowed(community_b, &pubkey) + .await + .expect("allowlist check B"), + "pubkey added only to A must not be allowed in B" + ); + assert!(db + .has_allowlist_entries(community_a) + .await + .expect("A has entries")); + assert!(!db + .has_allowlist_entries(community_b) + .await + .expect("B has no entries")); + + let listed = db + .list_allowlist(community_a) + .await + .expect("list A allowlist"); + assert_eq!(listed.len(), 1); + assert_eq!(listed[0].pubkey, pubkey); + + assert!( + !db.remove_from_allowlist(community_b, &pubkey) + .await + .expect("remove from B is no-op"), + "removing from B must not delete A's row" + ); + assert!(db + .is_pubkey_allowed(community_a, &pubkey) + .await + .expect("A still allowed after B remove")); + assert!(db + .remove_from_allowlist(community_a, &pubkey) + .await + .expect("remove from A")); + assert!(!db + .is_pubkey_allowed(community_a, &pubkey) + .await + .expect("A not allowed after remove")); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn communities_of_channels_present_for_existing_absent_for_missing() { @@ -2529,4 +2641,133 @@ mod tests { "result map must contain only existing channels" ); } + + /// BUG-5 regression: the `reactions` table is community-scoped + /// (`PK (community_id, event_created_at, event_id, pubkey, emoji)`), so a + /// reaction added under community A must be invisible and unremovable from + /// community B — even for the *identical* `(event_id, pubkey, emoji)` shape. + /// Before the fix, `add_reaction` omitted `community_id` (NOT NULL → 500) and + /// every read/remove filtered `event_id` only (latent cross-tenant bleed). + #[tokio::test] + #[ignore = "requires Postgres"] + async fn reactions_are_scoped_to_community() { + let db = setup_db().await; + let community_a = CommunityId::from_uuid(make_community(&db.pool).await); + let community_b = CommunityId::from_uuid(make_community(&db.pool).await); + + // Identical referenced-event shape across both tenants. + let event_id = [0xABu8; 32]; + let event_created_at = Utc::now(); + let pubkey = [7u8; 32]; + let emoji = "👍"; + + // (1) Add succeeds under A (this INSERT 500'd before the fix). + assert!( + db.add_reaction( + community_a, + &event_id, + event_created_at, + &pubkey, + emoji, + None + ) + .await + .expect("add reaction under A"), + "first reaction under A must be inserted" + ); + // Idempotent: re-adding the same active reaction is a no-op. + assert!( + !db.add_reaction( + community_a, + &event_id, + event_created_at, + &pubkey, + emoji, + None + ) + .await + .expect("duplicate reaction under A"), + "active duplicate under A must not re-insert" + ); + + // (2) Visible on A, invisible on B (grouped read path). + let groups_a = db + .get_reactions(community_a, &event_id, event_created_at, 100, None) + .await + .expect("get reactions A"); + assert_eq!(groups_a.len(), 1, "A must see its own reaction group"); + assert_eq!(groups_a[0].emoji, emoji); + assert_eq!(groups_a[0].count, 1); + + let groups_b = db + .get_reactions(community_b, &event_id, event_created_at, 100, None) + .await + .expect("get reactions B"); + assert!( + groups_b.is_empty(), + "B must NOT see A's reaction for the same event shape, got {groups_b:?}" + ); + + // (3) Active-record lookup is scoped: present on A, absent on B. + assert!( + db.get_active_reaction_record(community_a, &event_id, event_created_at, &pubkey, emoji) + .await + .expect("active record A") + .is_some(), + "A's active reaction record must be present" + ); + assert!( + db.get_active_reaction_record(community_b, &event_id, event_created_at, &pubkey, emoji) + .await + .expect("active record B") + .is_none(), + "B must not find A's active reaction record" + ); + + // (4) B can add the identical shape independently (no PK collision). + assert!( + db.add_reaction( + community_b, + &event_id, + event_created_at, + &pubkey, + emoji, + None + ) + .await + .expect("add reaction under B"), + "B must be able to add the same shape as its own scoped row" + ); + + // (5) Removing from B does not touch A's row. + assert!( + db.remove_reaction(community_b, &event_id, event_created_at, &pubkey, emoji) + .await + .expect("remove under B"), + "B remove must affect B's own row" + ); + assert!( + db.get_active_reaction_record(community_a, &event_id, event_created_at, &pubkey, emoji) + .await + .expect("active record A after B remove") + .is_some(), + "A's reaction must survive a B-side removal" + ); + + // (6) A remove affects only A; A's read now empty. + assert!( + db.remove_reaction(community_a, &event_id, event_created_at, &pubkey, emoji) + .await + .expect("remove under A"), + "A remove must affect A's row" + ); + let groups_a_after = db + .get_reactions(community_a, &event_id, event_created_at, 100, None) + .await + .expect("get reactions A after remove"); + assert!( + groups_a_after.is_empty(), + "A's reaction must be gone after A removes it" + ); + } } diff --git a/crates/buzz-db/src/partition.rs b/crates/buzz-db/src/partition.rs index 480da3b55..b3803f1b3 100644 --- a/crates/buzz-db/src/partition.rs +++ b/crates/buzz-db/src/partition.rs @@ -100,7 +100,7 @@ async fn ensure_partition( ))); } - let partition_name = format!("{table_name}_{suffix}"); + let partition_name = format!("{table_name}_p{suffix}"); let row = sqlx::query( r#" @@ -127,10 +127,26 @@ async fn ensure_partition( FOR VALUES FROM ('{start_date_str}') TO ('{end_date_str}')" ); - sqlx::query(sqlx::AssertSqlSafe(sql)).execute(pool).await?; - info!("added partition {partition_name}"); - - Ok(()) + match sqlx::query(sqlx::AssertSqlSafe(sql)).execute(pool).await { + Ok(_) => { + info!("added partition {partition_name}"); + Ok(()) + } + Err(sqlx::Error::Database(db_err)) + if db_err.code().as_deref() == Some("42P17") + && db_err.message().contains("would overlap partition") => + { + // Fresh schemas include a right-edge catch-all partition (`*_p_future`). + // If it already covers this month, the table is still safe for writes; + // treat the overlap as "ensured" rather than failing startup. + info!( + partition_name, + "partition range already covered by an existing partition" + ); + Ok(()) + } + Err(e) => Err(e.into()), + } } #[cfg(test)] diff --git a/crates/buzz-db/src/reaction.rs b/crates/buzz-db/src/reaction.rs index f35344833..fca1f3de7 100644 --- a/crates/buzz-db/src/reaction.rs +++ b/crates/buzz-db/src/reaction.rs @@ -6,6 +6,7 @@ use chrono::{DateTime, Utc}; use sqlx::{PgPool, Row}; use crate::error::Result; +use crate::CommunityId; // -- Public structs ----------------------------------------------------------- @@ -70,6 +71,7 @@ pub struct ActiveReactionRecord { /// two concurrent adds both see no existing row and then race to INSERT. pub async fn add_reaction( pool: &PgPool, + community: CommunityId, event_id: &[u8], event_created_at: DateTime, pubkey: &[u8], @@ -78,15 +80,16 @@ pub async fn add_reaction( ) -> Result { let result = sqlx::query( r#" - INSERT INTO reactions (event_created_at, event_id, pubkey, emoji, reaction_event_id) - VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (event_created_at, event_id, pubkey, emoji) DO UPDATE SET + INSERT INTO reactions (community_id, event_created_at, event_id, pubkey, emoji, reaction_event_id) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (community_id, event_created_at, event_id, pubkey, emoji) DO UPDATE SET created_at = NOW(), removed_at = NULL, reaction_event_id = COALESCE(EXCLUDED.reaction_event_id, reactions.reaction_event_id) WHERE reactions.removed_at IS NOT NULL "#, ) + .bind(community.as_uuid()) .bind(event_created_at) .bind(event_id) .bind(pubkey) @@ -109,6 +112,7 @@ pub async fn add_reaction( /// Returns `true` if a row was updated, `false` if not found or already removed. pub async fn remove_reaction( pool: &PgPool, + community: CommunityId, event_id: &[u8], event_created_at: DateTime, pubkey: &[u8], @@ -118,13 +122,15 @@ pub async fn remove_reaction( r#" UPDATE reactions SET removed_at = NOW() - WHERE event_created_at = $1 - AND event_id = $2 - AND pubkey = $3 - AND emoji = $4 + WHERE community_id = $1 + AND event_created_at = $2 + AND event_id = $3 + AND pubkey = $4 + AND emoji = $5 AND removed_at IS NULL "#, ) + .bind(community.as_uuid()) .bind(event_created_at) .bind(event_id) .bind(pubkey) @@ -140,16 +146,19 @@ pub async fn remove_reaction( /// Returns `true` if a row was updated, `false` if not found or already removed. pub async fn remove_reaction_by_source_event_id( pool: &PgPool, + community: CommunityId, reaction_event_id: &[u8], ) -> Result { let result = sqlx::query( r#" UPDATE reactions SET removed_at = NOW() - WHERE reaction_event_id = $1 + WHERE community_id = $1 + AND reaction_event_id = $2 AND removed_at IS NULL "#, ) + .bind(community.as_uuid()) .bind(reaction_event_id) .execute(pool) .await?; @@ -160,6 +169,7 @@ pub async fn remove_reaction_by_source_event_id( /// Look up the active reaction row for one actor + emoji + target tuple. pub async fn get_active_reaction_record( pool: &PgPool, + community: CommunityId, event_id: &[u8], event_created_at: DateTime, pubkey: &[u8], @@ -169,14 +179,16 @@ pub async fn get_active_reaction_record( r#" SELECT reaction_event_id FROM reactions - WHERE event_id = $1 - AND event_created_at = $2 - AND pubkey = $3 - AND emoji = $4 + WHERE community_id = $1 + AND event_id = $2 + AND event_created_at = $3 + AND pubkey = $4 + AND emoji = $5 AND removed_at IS NULL LIMIT 1 "#, ) + .bind(community.as_uuid()) .bind(event_id) .bind(event_created_at) .bind(pubkey) @@ -198,6 +210,7 @@ pub async fn get_active_reaction_record( /// reaction row to its source event. Returns `true` if the row was updated. pub async fn set_reaction_event_id( pool: &PgPool, + community: CommunityId, event_id: &[u8], event_created_at: DateTime, pubkey: &[u8], @@ -208,14 +221,16 @@ pub async fn set_reaction_event_id( r#" UPDATE reactions SET reaction_event_id = $1 - WHERE event_created_at = $2 - AND event_id = $3 - AND pubkey = $4 - AND emoji = $5 + WHERE community_id = $2 + AND event_created_at = $3 + AND event_id = $4 + AND pubkey = $5 + AND emoji = $6 AND removed_at IS NULL "#, ) .bind(reaction_event_id) + .bind(community.as_uuid()) .bind(event_created_at) .bind(event_id) .bind(pubkey) @@ -237,6 +252,7 @@ pub async fn set_reaction_event_id( /// `cursor` is reserved for future keyset pagination (currently unused). pub async fn get_reactions( pool: &PgPool, + community: CommunityId, event_id: &[u8], event_created_at: DateTime, limit: u32, @@ -253,18 +269,21 @@ pub async fn get_reactions( INNER JOIN ( SELECT DISTINCT emoji FROM reactions - WHERE event_id = $1 - AND event_created_at = $2 + WHERE community_id = $1 + AND event_id = $2 + AND event_created_at = $3 AND removed_at IS NULL ORDER BY emoji - LIMIT $3 + LIMIT $4 ) g ON g.emoji = r.emoji - WHERE r.event_id = $1 - AND r.event_created_at = $2 + WHERE r.community_id = $1 + AND r.event_id = $2 + AND r.event_created_at = $3 AND r.removed_at IS NULL ORDER BY r.emoji, r.created_at "#, ) + .bind(community.as_uuid()) .bind(event_id) .bind(event_created_at) .bind(limit as i64) @@ -319,6 +338,7 @@ pub async fn get_reactions( /// active reaction. Pairs with no reactions are omitted. pub async fn get_reactions_bulk( pool: &PgPool, + community: CommunityId, event_ids: &[(&[u8], DateTime)], ) -> Result> { if event_ids.is_empty() { @@ -335,13 +355,15 @@ pub async fn get_reactions_bulk( r#" SELECT emoji, COUNT(*) AS count FROM reactions - WHERE event_id = $1 - AND event_created_at = $2 + WHERE community_id = $1 + AND event_id = $2 + AND event_created_at = $3 AND removed_at IS NULL GROUP BY emoji ORDER BY emoji "#, ) + .bind(community.as_uuid()) .bind(*event_id) .bind(event_created_at) .fetch_all(pool) diff --git a/crates/buzz-db/src/relay_members.rs b/crates/buzz-db/src/relay_members.rs index e4999ac1b..8a71d8609 100644 --- a/crates/buzz-db/src/relay_members.rs +++ b/crates/buzz-db/src/relay_members.rs @@ -322,6 +322,7 @@ pub async fn backfill_from_allowlist(pool: &PgPool, community: CommunityId) -> R "INSERT INTO relay_members (community_id, pubkey, role, added_by, created_at) \ SELECT $1, encode(pubkey, 'hex'), 'member', NULL, added_at \ FROM pubkey_allowlist \ + WHERE community_id = $1 \ ON CONFLICT (community_id, pubkey) DO NOTHING", ) .bind(community.as_uuid()) diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index 7c9ee79bd..4ead112e1 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/crates/buzz-relay/src/api/media.rs @@ -699,7 +699,7 @@ async fn resolve_upload_scopes( let pubkey_bytes = blossom_pubkey.to_bytes().to_vec(); if !state .db - .is_pubkey_allowed(&pubkey_bytes) + .is_pubkey_allowed(tenant.community(), &pubkey_bytes) .await .unwrap_or(false) { diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 48a8f0354..143dfbee9 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -88,7 +88,11 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: if state.config.pubkey_allowlist_enabled && auth_ctx.auth_method == buzz_auth::AuthMethod::Nip42 { - let allowed = match state.db.is_pubkey_allowed(pubkey.as_bytes()).await { + let allowed = match state + .db + .is_pubkey_allowed(conn.tenant.community(), pubkey.as_bytes()) + .await + { Ok(v) => v, Err(e) => { warn!(conn_id = %conn_id, pubkey = %pubkey.to_hex(), error = %e, diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index e11380faf..05f823c6e 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -1832,7 +1832,14 @@ async fn ingest_event_inner( // exists — short-circuit without storing the event. let inserted = state .db - .add_reaction(&target_id, target_created_at, &actor_bytes, emoji, None) + .add_reaction( + tenant.community(), + &target_id, + target_created_at, + &actor_bytes, + emoji, + None, + ) .await .map_err(|e| IngestError::Internal(format!("error: {e}")))?; @@ -1861,7 +1868,13 @@ async fn ingest_event_inner( // Compensate: undo the reaction row so state stays consistent. if let Err(re) = state .db - .remove_reaction(&target_id, target_created_at, &actor_bytes, emoji) + .remove_reaction( + tenant.community(), + &target_id, + target_created_at, + &actor_bytes, + emoji, + ) .await { warn!(event_id = %event_id_hex, "reaction compensation failed: {re}"); @@ -1875,6 +1888,7 @@ async fn ingest_event_inner( if let Err(e) = state .db .set_reaction_event_id( + tenant.community(), &target_id, target_created_at, &actor_bytes, diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index fd91f0488..56e4316cb 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1883,7 +1883,7 @@ async fn handle_standard_deletion_event( // if the backfill was missed (set_reaction_event_id is best-effort). let removed = state .db - .remove_reaction_by_source_event_id(&target_id) + .remove_reaction_by_source_event_id(tenant.community(), &target_id) .await .unwrap_or(false); if !removed { @@ -1925,7 +1925,13 @@ async fn handle_standard_deletion_event( .unwrap_or_else(chrono::Utc::now); if let Err(e) = state .db - .remove_reaction(&react_target_id, react_target_ts, &actor, emoji) + .remove_reaction( + tenant.community(), + &react_target_id, + react_target_ts, + &actor, + emoji, + ) .await { tracing::warn!( @@ -2087,10 +2093,11 @@ async fn handle_git_repo_announcement( // (see `api::git::hydrate`). Announce only (1) reserves the repo name and // (2) seeds the empty-manifest pointer that makes the repo clone-able. // - // `.names/` is the relay's name registry. Each reservation holds - // an `owner` file naming the announcer. It serves three jobs at once: + // `.names//` is the relay's name registry. Each + // reservation holds an `owner` file naming the announcer. It serves three + // jobs at once inside the server-resolved community boundary: // - uniqueness: `create_dir` is atomic, so concurrent kind:30617 events - // for the same name can't both claim it (TOCTOU-free); + // for the same community/name can't both claim it (TOCTOU-free); // - idempotent re-announce: a reservation owned by the same pubkey is an // update, not a collision; // - per-pubkey quota: count the reservations whose `owner` matches. @@ -2100,7 +2107,9 @@ async fn handle_git_repo_announcement( // pointer (not this registry) preventing actual ref-state corruption. A // CAS-backed name index is the multi-instance follow-up. let git_repo_root = &state.config.git_repo_path; - let names_dir = git_repo_root.join(".names"); + let names_dir = git_repo_root + .join(".names") + .join(tenant.community().to_string()); std::fs::create_dir_all(&names_dir) .map_err(|e| anyhow::anyhow!("failed to create name reservation index: {e}"))?; diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index e14bff902..eaa484165 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -111,8 +111,8 @@ async fn main() -> anyhow::Result<()> { // (`relay_url_authority` → `normalize_host`), so the bootstrapped owner lands // in exactly the community that live requests for this host will resolve to. // - // `ensure_configured_community` is idempotent (`ON CONFLICT (host)`), so this - // is safe to run every startup. An empty authority (unparseable `relay_url`) + // `ensure_configured_community` is idempotent, so this is safe to run every + // startup. An empty authority (unparseable `relay_url`) // is a misconfiguration — fail fast when membership is enforced rather than // seeding an empty-host community that no request can ever resolve to. let deployment_community = { diff --git a/crates/buzz-test-client/tests/conformance_multitenant.rs b/crates/buzz-test-client/tests/conformance_multitenant.rs index 06dbf931d..004eb8bba 100644 --- a/crates/buzz-test-client/tests/conformance_multitenant.rs +++ b/crates/buzz-test-client/tests/conformance_multitenant.rs @@ -1699,12 +1699,11 @@ mod workflows { } /// Fire a workflow by id on `http_base`'s community (kind:46020, `d`=id). - /// Returns the parsed `{accepted, message}` body so the caller can assert on - /// the *wire-observable* accept/reject and message. The relay resolves the - /// workflow with `get_workflow(host_community, id)` — community-scoped — so a - /// foreign-community id fails closed with a generic `invalid: workflow not - /// found`, indistinguishable from "no such id at all" (no cross-tenant - /// enumeration oracle). + /// Returns a normalized `{accepted, message}` body so the caller can assert + /// on the *wire-observable* accept/reject and message. The HTTP bridge maps + /// `IngestError::Rejected` to HTTP 400 + `{error}` while the WS door maps the + /// same condition to `OK false`; for this conformance row either envelope is + /// acceptable. The safety property is the scoped lookup and generic message. async fn trigger_workflow( http_base: &str, keys: &Keys, @@ -1714,7 +1713,31 @@ mod workflows { .tags(vec![Tag::parse(["d", workflow_id]).unwrap()]) .sign_with_keys(keys) .unwrap(); - submit_event(http_base, keys, event).await + + let client = reqwest::Client::new(); + let resp = client + .post(format!("{http_base}/events")) + .header("X-Pubkey", keys.public_key().to_hex()) + .header("Content-Type", "application/json") + .body(serde_json::to_string(&event).expect("serialize event")) + .send() + .await + .unwrap_or_else(|e| panic!("POST workflow trigger to {http_base} failed: {e}")); + let status = resp.status(); + let body = resp.text().await.expect("read workflow trigger body"); + let parsed: serde_json::Value = serde_json::from_str(&body).unwrap_or_else(|e| { + panic!("parse workflow trigger JSON from {http_base}: {e} (body: {body})") + }); + if status.is_success() { + return parsed; + } + if status == reqwest::StatusCode::BAD_REQUEST { + return serde_json::json!({ + "accepted": false, + "message": parsed["error"].as_str().unwrap_or_default(), + }); + } + panic!("POST workflow trigger to {http_base} returned HTTP {status}: {body}"); } /// Obligation (trigger-confinement half): a workflow id defined under diff --git a/crates/buzz-test-client/tests/e2e_git.rs b/crates/buzz-test-client/tests/e2e_git.rs index 263c97ce5..b632c46ff 100644 --- a/crates/buzz-test-client/tests/e2e_git.rs +++ b/crates/buzz-test-client/tests/e2e_git.rs @@ -143,6 +143,9 @@ impl GitS3Probe { fn pointer_key(owner: &str, repo: &str) -> String { let repo = repo.strip_suffix(".git").unwrap_or(repo); + if let Ok(community) = std::env::var("BUZZ_E2E_GIT_COMMUNITY_ID") { + return format!("repos/{community}/{owner}/{repo}/pointer"); + } format!("repos/{owner}/{repo}/pointer") }