From 0c85626194068aa2f9c4e1b419d7c8caff00ede4 Mon Sep 17 00:00:00 2001 From: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Date: Sat, 1 Aug 2026 12:18:07 -0400 Subject: [PATCH] feat(relay): complete Phase 1 single-node slice Signed-off-by: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> --- PHASE1C_BREAKAGE_CATALOGUE.md | 75 +++ crates/buzz-db/src/lib.rs | 208 ++++++-- crates/buzz-db/src/sqlite.rs | 473 ++++++++++++++++++ crates/buzz-pubsub/src/lib.rs | 2 +- crates/buzz-pubsub/src/nip98_replay.rs | 93 ++++ crates/buzz-relay/src/api/bridge.rs | 8 +- .../src/handlers/command_executor.rs | 67 ++- crates/buzz-relay/src/handlers/event.rs | 3 +- .../buzz-relay/src/handlers/side_effects.rs | 14 + crates/buzz-relay/src/main.rs | 115 ++++- 10 files changed, 996 insertions(+), 62 deletions(-) create mode 100644 PHASE1C_BREAKAGE_CATALOGUE.md diff --git a/PHASE1C_BREAKAGE_CATALOGUE.md b/PHASE1C_BREAKAGE_CATALOGUE.md new file mode 100644 index 000000000..06fd2f336 --- /dev/null +++ b/PHASE1C_BREAKAGE_CATALOGUE.md @@ -0,0 +1,75 @@ +# Phase 1c single-node breakage catalogue + +Branch: `local-mode/phase-1`, based on committed SQLite core slice `88b9644` and the Phase 0 seam stack through `a680665`. This catalogue describes the production-router single-node slice, not the retired Phase 0 alternate router. + +## Result + +`BUZZ_PROFILE=single-node` enters the normal production router through `crates/buzz-relay/src/main.rs:164,1208-1305`, rejects non-loopback binding (`main.rs:1209-1214`), and constructs SQLite, in-process pub/sub and replay fencing, filesystem media, unsupported search, and permissive admission without opening PostgreSQL, Redis, or S3 connections (`main.rs:1231-1278`). The readiness log enumerates intentionally disabled background/service families (`main.rs:1295-1296`). + +The fresh WP-B matrix passed profiles, public relay-member admission (kind 9030), channel creation/membership, owner↔agent chat, thread/reply/mention, reaction add/get/remove/get, DM open and sends in both directions, feeds, explicit search denial, canonical filesystem media byte round-trip, NIP-42/history/live fan-out, and restart durability. Evidence is recorded below. + +## Worked as-is + +| Surface | File:line evidence | Result | +|---|---|---| +| Production protocol/router and NIP-42 | `crates/buzz-relay/src/main.rs:164,1284-1294`; `crates/buzz-test-client/src/bin/local_mode_smoke.rs:7-54` | Normal router and client protocol; no local-only wire protocol. | +| PostgreSQL production profile | `crates/buzz-db/src/lib.rs:172-221`; backend matches such as `lib.rs:1730-1734` | Existing PostgreSQL construction and paths remain the default backend arms. | +| Search failure shape | `crates/buzz-relay/src/api/bridge.rs:1732-1741` | Local unsupported search returns explicit HTTP 501 `unsupported_feature`, not 500 or false success. | +| Loopback safety fence | `crates/buzz-relay/src/main.rs:1209-1214` | Single-node startup fails closed on a non-loopback bind address. | + +## Worked with local backend/shim + +| Surface | File:line evidence | Fidelity / limitation | +|---|---|---| +| Startup/service aggregate | `crates/buzz-relay/src/main.rs:1208-1305` | Production router with a local backend bundle; production-only workers are not started. | +| Durable SQLite core | `crates/buzz-db/src/sqlite.rs:137-458`; dispatch at `crates/buzz-db/src/lib.rs:1710-1812,1891-1961,2187-2256,4848-4850,5176-5178` | Real normalized community/channel/member/event/reaction/DM tables and SQL filtering. Thread ancestry remains canonical NIP-10 tags; denormalized thread metadata/counters are deferred. | +| Conformance row projection | `crates/buzz-db/src/lib.rs:1670-1699`; `crates/buzz-db/src/sqlite.rs:824-844` | SQLite now resolves each returned row's channel community, removing the PG-only warning while retaining the non-interference trace seam. | +| Pub/sub + presence | `crates/buzz-pubsub/src/lib.rs:133-160,994-1057`; startup `main.rs:1252,1286-1292` | Process-local, community-scoped fan-out and expiring presence leases; no cross-process Redis semantics. | +| NIP-98 replay | `crates/buzz-pubsub/src/nip98_replay.rs:1-213`; startup `main.rs:1270,1277` | Bounded in-process replay fence; restart does not retain replay history. | +| Media | `crates/buzz-media/src/storage.rs:89-98,104-313`; startup `main.rs:1275` | Filesystem CAS path supports canonical upload/download, sidecars, ranges and pages; no S3 replication. | +| Profiles/users/channels/membership | SQLite methods `crates/buzz-db/src/sqlite.rs:715-844,903-1079`; dispatch `crates/buzz-db/src/lib.rs:2299-2467,2672-2779,4334-4467` | Startup owner bootstrap plus public relay-admin and channel membership flows work locally. | +| Feed | `crates/buzz-db/src/sqlite.rs:461-530`; routed dispatch `crates/buzz-db/src/lib.rs:3444-3470,3532-3558,3618-3642` | WP-B mentions/needs-action/activity query shapes work; advanced non-routed feed methods remain PostgreSQL-only. | +| Reactions | `crates/buzz-db/src/sqlite.rs:533-599`; routed writes/removal `crates/buzz-db/src/lib.rs:2213-2256,3302-3344` | Atomic reaction/event insertion, dedupe, and removal used by CLI matrix. Aggregate/direct helper family is not fully ported. | +| DM open | `crates/buzz-db/src/sqlite.rs:601-713`; command seam `crates/buzz-db/src/lib.rs:2845-2858`; `crates/buzz-relay/src/handlers/command_executor.rs:373-413` | DM open and its command idempotency execute transactionally in SQLite. Other command kinds are rejected before persistence; broader DM list/find/create public methods remain PostgreSQL-only. | +| Workflow dispatch gate | `crates/buzz-relay/src/handlers/event.rs:520-559` | Local profile does not invoke the PostgreSQL workflow engine, matching startup's disabled declaration. | + +## Stubbed / deliberately permissive + +| Surface | File:line evidence | Local behavior | +|---|---|---| +| Search | `crates/buzz-relay/src/main.rs:1274`; `api/bridge.rs:1732-1741` | Service is unavailable and explicitly returns 501; FTS5 is deferred. | +| Git, push, workflows, reapers, usage, replica tasks | `crates/buzz-relay/src/main.rs:1295-1296`; Git route gate `crates/buzz-relay/src/router.rs:49-52` | Not started/exposed in the single-node profile. | +| Admission rate limiting | `crates/buzz-pubsub/src/rate_limiter.rs:97-103`; startup `main.rs:1278` | Explicit permissive in-process policy, not Redis quotas. | +| Audit | `crates/buzz-relay/src/main.rs:1280-1282` | No audit service/worker in single-node startup. | +| Relay signing identity | `crates/buzz-relay/src/main.rs:1245-1251` | If no key is configured, local mode uses a public deterministic development key; safe only with the enforced loopback bind and unsuitable for shared/production use. | +| Unsupported command families | `crates/buzz-relay/src/handlers/command_executor.rs:120-128` | Commands other than the dedicated atomic DM-open path are explicitly rejected before an idempotency event can be persisted. | +| Thread summaries and NIP-43 publications | `crates/buzz-relay/src/handlers/side_effects.rs` (`emit_live_thread_summary`, `publish_nip43_membership_list`, `publish_nip43_delta`) | PostgreSQL-backed thread/list snapshots are omitted locally; globally scoped NIP-43 deltas are also suppressed to avoid leaking the loopback roster. Durable NIP-10 tags and the SQLite relay-members table remain authoritative. | + +## Blocked PostgreSQL-only public DB methods + +These calls fail with typed `DbError::UnsupportedBackend` if reached on SQLite; they are not silently emulated. The list is representative of the remaining families rather than a claim that every `Db` method is ported. + +| Family | File:line evidence | Consequence | +|---|---|---| +| Direct deletion helpers | `crates/buzz-db/src/lib.rs:1919-1937` | Standard event deletion uses the routed atomic helper, but callers of these direct methods remain blocked. | +| Channel policy | `crates/buzz-db/src/lib.rs:2785-2793` | Channel add-policy mutation is unavailable locally. | +| Direct DM helpers/list | `crates/buzz-db/src/lib.rs:2795-2823` | WP-B DM open uses the SQLite command seam; separate find/create/list APIs remain blocked. | +| Thread counters | `crates/buzz-db/src/lib.rs:3263-3277` | No denormalized decrement; tag-based ancestry only. | +| Direct/aggregate reaction helpers | `crates/buzz-db/src/lib.rs:3280-3299,3347-3416` | Ingest/removal tracer path works; direct add, active-record lookup, and aggregate reads are not routed. | +| Non-routed feed helpers | `crates/buzz-db/src/lib.rs:3418-3436,3510-3528,3598-3614` | HTTP/CLI routed feed works; callers choosing writer-only helpers remain blocked. | + +## Verification artifacts + +Fresh runtime evidence is under `.scratch/wp-b-matrix/` (ephemeral, not committed): + +```text +PASS wp_b_cli_matrix +PASS media_byte_comparison +PASS kind_9030_admission +PASS nip42,event_insert,req_history,live_fanout +PASS sqlite_restart_history +PASS no_postgres_workflow_conformance_leakage +PASS no_global_nip43_publications +``` + +The matrix covers profiles, channel creation/membership, owner↔agent chat, reply/thread/mention, reactions through removal, DM open plus both-direction sends, owner/agent feeds, explicit search 501, and canonical media upload/download byte comparison. A scan of fresh first-boot and restart logs found no `Workflow trigger failed`, `conformance row-community lookup failed`, or `PostgreSQL operation on SQLite` lines; the fresh SQLite file contains zero kind 8000/8001/13534 global NIP-43 publication rows. Package test counts and the exact verified commit are reported with the final commit because line numbers above describe the formatted final working tree. diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 036015a69..9b9a2d32a 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -1674,25 +1674,30 @@ impl Db { if channel_ids.is_empty() { return Ok(std::collections::HashMap::new()); } - let rows = sqlx::query( - r#" + match &self.backend { + DbBackend::SQLite(pool) => sqlite::communities_of_channels(pool, channel_ids).await, + DbBackend::Postgres => { + let rows = sqlx::query( + r#" SELECT id, community_id FROM channels WHERE id = ANY($1) AND deleted_at IS NULL "#, - ) - .bind(channel_ids) - .fetch_all(self.pg_pool()?) - .await?; + ) + .bind(channel_ids) + .fetch_all(self.pg_pool()?) + .await?; - let mut out = std::collections::HashMap::with_capacity(rows.len()); - for row in rows { - let ch: Uuid = row.try_get("id")?; - let cm: Uuid = row.try_get("community_id")?; - out.insert(ch, CommunityId::from_uuid(cm)); + let mut out = std::collections::HashMap::with_capacity(rows.len()); + for row in rows { + let ch: Uuid = row.try_get("id")?; + let cm: Uuid = row.try_get("community_id")?; + out.insert(ch, CommunityId::from_uuid(cm)); + } + Ok(out) + } } - Ok(out) } /// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate. @@ -1750,6 +1755,9 @@ impl Db { path: &'static str, q: &EventQuery, ) -> Result> { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::query_events(pool, q).await; + } let predicate = RoutePredicate::for_query(q, self.replica_read_max_age.is_some()); match self.route_read(path, predicate).await { RouteDecision::Replica(mut tx, _entry, reason) => { @@ -1784,6 +1792,9 @@ impl Db { path: &'static str, q: &EventQuery, ) -> Result> { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::query_events(pool, q).await; + } match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { match event::query_events_on(&mut tx, q).await { @@ -1895,7 +1906,13 @@ impl Db { community_id: CommunityId, id_bytes: &[u8], ) -> Result> { - event::get_event_by_id_including_deleted(self.pg_pool()?, community_id, id_bytes).await + match &self.backend { + DbBackend::SQLite(pool) => sqlite::get_event_by_id(pool, community_id, id_bytes).await, + DbBackend::Postgres => { + event::get_event_by_id_including_deleted(self.pg_pool()?, community_id, id_bytes) + .await + } + } } /// Soft-deletes an event. Returns `Ok(true)` if deleted, `Ok(false)` if already deleted. @@ -1938,14 +1955,21 @@ impl Db { parent_event_id: Option<&[u8]>, root_event_id: Option<&[u8]>, ) -> Result { - event::soft_delete_event_and_update_thread( - self.pg_pool()?, - community_id, - event_id, - parent_event_id, - root_event_id, - ) - .await + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::soft_delete_event(pool, community_id, event_id).await + } + DbBackend::Postgres => { + event::soft_delete_event_and_update_thread( + self.pg_pool()?, + community_id, + event_id, + parent_event_id, + root_event_id, + ) + .await + } + } } /// Returns the most recent `created_at` for a channel. @@ -2171,6 +2195,13 @@ impl Db { channel_id: Option, thread_meta: Option>, ) -> Result<(StoredEvent, bool)> { + if let DbBackend::SQLite(pool) = &self.backend { + // SQLite Phase 1 persists the canonical event; thread relationships + // remain durable in the event's NIP-10 tags until Phase 2 ports the + // denormalized thread metadata tables. + let _ = thread_meta; + return sqlite::insert_event(pool, community_id, event, channel_id).await; + } let result = event::insert_event_with_thread_metadata( self.pg_pool()?, community_id, @@ -2200,6 +2231,19 @@ impl Db { actor_pubkey: &[u8], emoji: &str, ) -> Result { + if let DbBackend::SQLite(pool) = &self.backend { + let _ = thread_meta; + return sqlite::insert_reaction_event( + pool, + community_id, + event, + channel_id, + target_event_id, + actor_pubkey, + emoji, + ) + .await; + } let outcome = event::insert_reaction_event_with_thread_metadata( self.pg_pool()?, community_id, @@ -2663,16 +2707,32 @@ impl Db { about: Option<&str>, nip05_handle: Option<&str>, ) -> Result<()> { - user::update_user_profile( - self.pg_pool()?, - community_id, - pubkey, - display_name, - avatar_url, - about, - nip05_handle, - ) - .await + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::update_user_profile( + pool, + community_id, + pubkey, + display_name, + avatar_url, + about, + nip05_handle, + ) + .await + } + DbBackend::Postgres => { + user::update_user_profile( + self.pg_pool()?, + community_id, + pubkey, + display_name, + avatar_url, + about, + nip05_handle, + ) + .await + } + } } /// Look up a user by NIP-05 handle. @@ -2795,7 +2855,33 @@ impl Db { pubkeys: &[&[u8]], created_by: &[u8], ) -> Result<(channel::ChannelRecord, bool)> { - dm::open_dm(self.pg_pool()?, community_id, pubkeys, created_by).await + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::open_dm(pool, community_id, pubkeys, created_by, None) + .await? + .ok_or_else(|| DbError::InvalidData("unexpected duplicate DM open".into())) + } + DbBackend::Postgres => { + dm::open_dm(self.pg_pool()?, community_id, pubkeys, created_by).await + } + } + } + + /// Atomically persist a SQLite DM-open command and its channel mutation. + /// `None` means the command event was already processed. + pub async fn open_dm_sqlite_command( + &self, + community_id: CommunityId, + pubkeys: &[&[u8]], + created_by: &[u8], + event: &nostr::Event, + ) -> Result> { + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::open_dm(pool, community_id, pubkeys, created_by, Some(event)).await + } + DbBackend::Postgres => Ok(None), + } } /// Hide a DM channel for a specific user. @@ -3177,6 +3263,11 @@ impl Db { community_id: CommunityId, event_id: &[u8], ) -> Result> { + if matches!(&self.backend, DbBackend::SQLite(_)) { + // SQLite Phase 1 derives ancestry from durable NIP-10 event tags; + // denormalized counters and metadata arrive with the Phase 2 port. + return Ok(None); + } thread::get_thread_metadata_by_event(self.pg_pool()?, community_id, event_id).await } @@ -3227,6 +3318,17 @@ impl Db { pubkey: &[u8], emoji: &str, ) -> Result { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::remove_reaction( + pool, + community, + event_id, + event_created_at, + pubkey, + emoji, + ) + .await; + } reaction::remove_reaction( self.pg_pool()?, community, @@ -3244,6 +3346,10 @@ impl Db { community: CommunityId, reaction_event_id: &[u8], ) -> Result { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::remove_reaction_by_source_event_id(pool, community, reaction_event_id) + .await; + } reaction::remove_reaction_by_source_event_id(self.pg_pool()?, community, reaction_event_id) .await } @@ -3355,6 +3461,17 @@ impl Db { since: Option>, limit: i64, ) -> Result> { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::query_feed_mentions( + pool, + community, + pubkey_bytes, + accessible_channel_ids, + since, + limit, + ) + .await; + } match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { match feed::query_mentions_on( @@ -3432,6 +3549,17 @@ impl Db { since: Option>, limit: i64, ) -> Result> { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::query_feed_needs_action( + pool, + community, + pubkey_bytes, + accessible_channel_ids, + since, + limit, + ) + .await; + } match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { match feed::query_needs_action_on( @@ -3506,6 +3634,16 @@ impl Db { since: Option>, limit: i64, ) -> Result> { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::query_feed_activity( + pool, + community, + accessible_channel_ids, + since, + limit, + ) + .await; + } match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { match feed::query_activity_on( @@ -4753,6 +4891,9 @@ impl Db { event: &nostr::Event, channel_id: Option, ) -> Result<(StoredEvent, bool)> { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::replace_event(pool, community_id, event, channel_id, None).await; + } let kind_i32 = buzz_core::kind::event_kind_i32(event); let pubkey_bytes = event.pubkey.to_bytes(); let created_at_secs = event.created_at.as_secs() as i64; @@ -5078,6 +5219,9 @@ impl Db { d_tag: &str, channel_id: Option, ) -> Result<(StoredEvent, bool)> { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::replace_event(pool, community_id, event, channel_id, Some(d_tag)).await; + } let kind_i32 = buzz_core::kind::event_kind_i32(event); let pubkey_bytes = event.pubkey.to_bytes(); let created_at_secs = event.created_at.as_secs() as i64; diff --git a/crates/buzz-db/src/sqlite.rs b/crates/buzz-db/src/sqlite.rs index da2aa68e9..922657ad1 100644 --- a/crates/buzz-db/src/sqlite.rs +++ b/crates/buzz-db/src/sqlite.rs @@ -55,6 +55,7 @@ CREATE TABLE IF NOT EXISTS channels ( canvas TEXT, channel_type TEXT NOT NULL, visibility TEXT NOT NULL, + participant_hash BLOB, created_by BLOB NOT NULL, created_at INTEGER NOT NULL DEFAULT (unixepoch()), updated_at INTEGER NOT NULL DEFAULT (unixepoch()), @@ -80,6 +81,7 @@ CREATE TABLE IF NOT EXISTS channel_members ( role TEXT NOT NULL, joined_at INTEGER NOT NULL DEFAULT (unixepoch()), invited_by BLOB, + hidden_at INTEGER, removed_at INTEGER, PRIMARY KEY (channel_id, pubkey), FOREIGN KEY (channel_id) REFERENCES channels(id) ON DELETE CASCADE @@ -115,6 +117,21 @@ CREATE INDEX IF NOT EXISTS idx_events_community_kind ON events (community_id, kind, created_at DESC); CREATE INDEX IF NOT EXISTS idx_events_channel_created ON events (community_id, channel_id, created_at DESC); + +CREATE TABLE IF NOT EXISTS reactions ( + community_id TEXT NOT NULL, + event_created_at INTEGER NOT NULL, + event_id BLOB NOT NULL, + pubkey BLOB NOT NULL, + emoji TEXT NOT NULL, + reaction_event_id BLOB, + removed_at INTEGER, + PRIMARY KEY (community_id, event_created_at, event_id, pubkey, emoji), + FOREIGN KEY (community_id) REFERENCES communities(id) ON DELETE CASCADE +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_reactions_source_event + ON reactions (community_id, reaction_event_id) + WHERE reaction_event_id IS NOT NULL; "#; pub(crate) async fn connect(path_or_url: &str) -> Result { @@ -135,9 +152,78 @@ pub(crate) async fn connect(path_or_url: &str) -> Result { } pub(crate) async fn migrate(pool: &SqlitePool) -> Result<()> { + let had_application_schema = sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM sqlite_master WHERE type = 'table' AND name = 'channels'", + ) + .fetch_one(pool) + .await? + != 0; + for statement in SCHEMA.split(';').map(str::trim).filter(|s| !s.is_empty()) { sqlx::query(statement).execute(pool).await?; } + sqlx::query( + "CREATE TABLE IF NOT EXISTS schema_version (singleton INTEGER PRIMARY KEY CHECK (singleton = 1), version INTEGER NOT NULL)", + ) + .execute(pool) + .await?; + let initial_version = if had_application_schema { 1_i64 } else { 2_i64 }; + sqlx::query( + "INSERT INTO schema_version (singleton, version) VALUES (1, ?1) ON CONFLICT(singleton) DO NOTHING", + ) + .bind(initial_version) + .execute(pool) + .await?; + + let mut version = + sqlx::query_scalar::<_, i64>("SELECT version FROM schema_version WHERE singleton = 1") + .fetch_one(pool) + .await?; + if version < 2 { + let mut tx = pool.begin().await?; + ensure_column_on( + &mut tx, + "channels", + "participant_hash", + "ALTER TABLE channels ADD COLUMN participant_hash BLOB", + ) + .await?; + ensure_column_on( + &mut tx, + "channel_members", + "hidden_at", + "ALTER TABLE channel_members ADD COLUMN hidden_at INTEGER", + ) + .await?; + sqlx::query("UPDATE schema_version SET version = 2 WHERE singleton = 1") + .execute(&mut *tx) + .await?; + tx.commit().await?; + version = 2; + } + if version != 2 { + return Err(crate::DbError::InvalidData(format!( + "unsupported SQLite schema version {version}" + ))); + } + Ok(()) +} + +async fn ensure_column_on( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + table: &str, + column: &str, + alter_sql: &'static str, +) -> Result<()> { + let pragma = format!("PRAGMA table_info({table})"); + let exists = sqlx::query(sqlx::AssertSqlSafe(pragma)) + .fetch_all(&mut **tx) + .await? + .iter() + .any(|row| row.get::("name") == column); + if !exists { + sqlx::query(alter_sql).execute(&mut **tx).await?; + } Ok(()) } @@ -230,6 +316,79 @@ pub(crate) async fn insert_event( )) } +pub(crate) async fn replace_event( + pool: &SqlitePool, + community: CommunityId, + event: &nostr::Event, + channel_id: Option, + d_tag: Option<&str>, +) -> Result<(buzz_core::StoredEvent, bool)> { + let mut tx = pool.begin().await?; + let rows = sqlx::query("SELECT id, event_json FROM events WHERE community_id = ?1 AND kind = ?2 AND pubkey = ?3 AND channel_id IS ?4") + .bind(community.as_uuid().to_string()) + .bind(event.kind.as_u16() as i32) + .bind(event.pubkey.to_bytes().as_slice()) + .bind(channel_id.map(|id| id.to_string())) + .fetch_all(&mut *tx) + .await?; + let mut replaced_ids = Vec::new(); + for row in rows { + let existing: nostr::Event = serde_json::from_str(row.try_get("event_json")?)?; + let existing_d = crate::event::extract_d_tag(&existing).unwrap_or_default(); + if d_tag.is_some_and(|expected| existing_d != expected) { + continue; + } + if event.created_at < existing.created_at + || (event.created_at == existing.created_at && event.id <= existing.id) + { + return Ok(( + buzz_core::StoredEvent::with_received_at( + existing, + chrono::Utc::now(), + channel_id, + true, + ), + false, + )); + } + replaced_ids.push(row.try_get::, _>("id")?); + } + for id in replaced_ids { + sqlx::query("DELETE FROM events WHERE community_id = ?1 AND id = ?2") + .bind(community.as_uuid().to_string()) + .bind(id) + .execute(&mut *tx) + .await?; + } + let received_at = chrono::Utc::now(); + sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags_json, content, sig, channel_id, received_at, event_json) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)") + .bind(community.as_uuid().to_string()).bind(event.id.as_bytes().as_slice()).bind(event.pubkey.to_bytes().as_slice()) + .bind(event.created_at.as_secs() as i64).bind(event.kind.as_u16() as i32).bind(serde_json::to_string(&event.tags)?) + .bind(&event.content).bind(event.sig.serialize().as_slice()).bind(channel_id.map(|id| id.to_string())) + .bind(received_at.timestamp()).bind(serde_json::to_string(event)?).execute(&mut *tx).await?; + tx.commit().await?; + Ok(( + buzz_core::StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), + true, + )) +} + +pub(crate) async fn soft_delete_event( + pool: &SqlitePool, + community: CommunityId, + event_id: &[u8], +) -> Result { + Ok( + sqlx::query("DELETE FROM events WHERE community_id = ?1 AND id = ?2") + .bind(community.as_uuid().to_string()) + .bind(event_id) + .execute(pool) + .await? + .rows_affected() + != 0, + ) +} + pub(crate) async fn get_event_by_id( pool: &SqlitePool, community: CommunityId, @@ -368,6 +527,231 @@ pub(crate) async fn query_events( Ok(events.into_iter().skip(offset).take(limit).collect()) } +pub(crate) async fn query_feed_mentions( + pool: &SqlitePool, + community: CommunityId, + pubkey_bytes: &[u8], + accessible_channel_ids: &[Uuid], + since: Option>, + limit: i64, +) -> Result> { + let mut query = crate::EventQuery::for_community(community); + query.kinds = Some(vec![ + buzz_core::kind::KIND_STREAM_MESSAGE as i32, + buzz_core::kind::KIND_STREAM_MESSAGE_V2 as i32, + buzz_core::kind::KIND_TEXT_NOTE as i32, + buzz_core::kind::KIND_FORUM_POST as i32, + buzz_core::kind::KIND_FORUM_COMMENT as i32, + buzz_core::kind::KIND_GIT_PULL_REQUEST as i32, + buzz_core::kind::KIND_GIT_PR_UPDATE as i32, + buzz_core::kind::KIND_GIT_ISSUE as i32, + buzz_core::kind::KIND_GIT_STATUS_OPEN as i32, + buzz_core::kind::KIND_GIT_STATUS_MERGED as i32, + buzz_core::kind::KIND_GIT_STATUS_CLOSED as i32, + buzz_core::kind::KIND_GIT_STATUS_DRAFT as i32, + ]); + query.p_tag_hex = Some(hex::encode(pubkey_bytes)); + query.channel_ids = Some(accessible_channel_ids.to_vec()); + query.since = since; + query.limit = Some(limit.min(crate::feed::FEED_MAX_LIMIT)); + query_events(pool, &query).await +} + +pub(crate) async fn query_feed_needs_action( + pool: &SqlitePool, + community: CommunityId, + pubkey_bytes: &[u8], + accessible_channel_ids: &[Uuid], + since: Option>, + limit: i64, +) -> Result> { + let mut query = crate::EventQuery::for_community(community); + query.kinds = Some(vec![ + buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED as i32, + buzz_core::kind::KIND_STREAM_REMINDER as i32, + ]); + query.p_tag_hex = Some(hex::encode(pubkey_bytes)); + query.channel_ids = Some(accessible_channel_ids.to_vec()); + query.since = since; + query.limit = Some(limit.min(crate::feed::FEED_MAX_LIMIT)); + query_events(pool, &query).await +} + +pub(crate) async fn query_feed_activity( + pool: &SqlitePool, + community: CommunityId, + accessible_channel_ids: &[Uuid], + since: Option>, + limit: i64, +) -> Result> { + let mut query = crate::EventQuery::for_community(community); + query.kinds = Some(vec![ + buzz_core::kind::KIND_STREAM_MESSAGE as i32, + buzz_core::kind::KIND_STREAM_MESSAGE_V2 as i32, + buzz_core::kind::KIND_FORUM_POST as i32, + buzz_core::kind::KIND_JOB_REQUEST as i32, + buzz_core::kind::KIND_JOB_PROGRESS as i32, + buzz_core::kind::KIND_JOB_RESULT as i32, + ]); + query.channel_ids = Some(accessible_channel_ids.to_vec()); + query.since = since; + query.limit = Some(limit.min(crate::feed::FEED_MAX_LIMIT)); + query_events(pool, &query).await +} + +pub(crate) async fn insert_reaction_event( + pool: &SqlitePool, + community: CommunityId, + reaction_event: &nostr::Event, + channel_id: Option, + target_event_id: &[u8], + actor_pubkey: &[u8], + emoji: &str, +) -> Result { + let mut tx = pool.begin().await?; + let target_created_at: Option = sqlx::query_scalar( + "SELECT created_at FROM events WHERE community_id = ?1 AND id = ?2 LIMIT 1", + ) + .bind(community.as_uuid().to_string()) + .bind(target_event_id) + .fetch_optional(&mut *tx) + .await?; + let Some(target_created_at) = target_created_at else { + return Ok(crate::event::ReactionEventInsertOutcome::TargetMissing); + }; + let changed = sqlx::query("INSERT INTO reactions (community_id, event_created_at, event_id, pubkey, emoji, reaction_event_id) VALUES (?1, ?2, ?3, ?4, ?5, ?6) ON CONFLICT (community_id, event_created_at, event_id, pubkey, emoji) DO UPDATE SET removed_at = NULL, reaction_event_id = excluded.reaction_event_id WHERE reactions.removed_at IS NOT NULL") + .bind(community.as_uuid().to_string()).bind(target_created_at).bind(target_event_id) + .bind(actor_pubkey).bind(emoji).bind(reaction_event.id.as_bytes().as_slice()) + .execute(&mut *tx).await?.rows_affected() != 0; + if !changed { + return Ok(crate::event::ReactionEventInsertOutcome::Duplicate); + } + let received_at = chrono::Utc::now(); + let inserted = sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags_json, content, sig, channel_id, received_at, event_json) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) ON CONFLICT DO NOTHING") + .bind(community.as_uuid().to_string()).bind(reaction_event.id.as_bytes().as_slice()).bind(reaction_event.pubkey.to_bytes().as_slice()) + .bind(reaction_event.created_at.as_secs() as i64).bind(reaction_event.kind.as_u16() as i32).bind(serde_json::to_string(&reaction_event.tags)?) + .bind(&reaction_event.content).bind(reaction_event.sig.serialize().as_slice()).bind(channel_id.map(|id| id.to_string())) + .bind(received_at.timestamp()).bind(serde_json::to_string(reaction_event)?).execute(&mut *tx).await?.rows_affected() == 1; + tx.commit().await?; + Ok(crate::event::ReactionEventInsertOutcome::Inserted { + stored_event: Box::new(buzz_core::StoredEvent::with_received_at( + reaction_event.clone(), + received_at, + channel_id, + true, + )), + was_inserted: inserted, + }) +} + +pub(crate) async fn remove_reaction( + pool: &SqlitePool, + community: CommunityId, + event_id: &[u8], + event_created_at: chrono::DateTime, + pubkey: &[u8], + emoji: &str, +) -> Result { + Ok(sqlx::query("UPDATE reactions SET removed_at = unixepoch() WHERE community_id = ?1 AND event_created_at = ?2 AND event_id = ?3 AND pubkey = ?4 AND emoji = ?5 AND removed_at IS NULL") + .bind(community.as_uuid().to_string()).bind(event_created_at.timestamp()).bind(event_id).bind(pubkey).bind(emoji) + .execute(pool).await?.rows_affected() != 0) +} + +pub(crate) async fn remove_reaction_by_source_event_id( + pool: &SqlitePool, + community: CommunityId, + reaction_event_id: &[u8], +) -> Result { + Ok(sqlx::query("UPDATE reactions SET removed_at = unixepoch() WHERE community_id = ?1 AND reaction_event_id = ?2 AND removed_at IS NULL") + .bind(community.as_uuid().to_string()).bind(reaction_event_id) + .execute(pool).await?.rows_affected() != 0) +} + +pub(crate) async fn open_dm( + pool: &SqlitePool, + community: CommunityId, + pubkeys: &[&[u8]], + created_by: &[u8], + command_event: Option<&nostr::Event>, +) -> Result> { + let mut participants = pubkeys.to_vec(); + if !participants.contains(&created_by) { + participants.push(created_by); + } + participants.sort_unstable(); + participants.dedup(); + if !(2..=9).contains(&participants.len()) { + return Err(crate::DbError::InvalidData( + "DM requires 2-9 unique participants".into(), + )); + } + if participants.iter().any(|pubkey| pubkey.len() != 32) { + return Err(crate::DbError::InvalidData( + "DM participant pubkeys must be 32 bytes".into(), + )); + } + let hash = crate::dm::compute_participant_hash(&participants); + let mut tx = pool.begin().await?; + if let Some(event) = command_event { + let received_at = chrono::Utc::now(); + let inserted = sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags_json, content, sig, channel_id, received_at, event_json) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL, ?9, ?10) ON CONFLICT DO NOTHING") + .bind(community.as_uuid().to_string()).bind(event.id.as_bytes().as_slice()).bind(event.pubkey.to_bytes().as_slice()) + .bind(event.created_at.as_secs() as i64).bind(event.kind.as_u16() as i32).bind(serde_json::to_string(&event.tags)?) + .bind(&event.content).bind(event.sig.serialize().as_slice()).bind(received_at.timestamp()).bind(serde_json::to_string(event)?) + .execute(&mut *tx).await?.rows_affected() != 0; + if !inserted { + return Ok(None); + } + } + let select = "SELECT id, name, channel_type, visibility, description, canvas, created_by, created_at, updated_at, archived_at, deleted_at, nip29_group_id, topic_required, max_members, topic, topic_set_by, topic_set_at, purpose, purpose_set_by, purpose_set_at, ttl_seconds, ttl_deadline FROM channels WHERE community_id = ?1 AND participant_hash = ?2 AND channel_type = 'dm' AND deleted_at IS NULL LIMIT 1"; + if let Some(row) = sqlx::query(select) + .bind(community.as_uuid().to_string()) + .bind(hash.as_slice()) + .fetch_optional(&mut *tx) + .await? + { + sqlx::query("UPDATE channel_members SET hidden_at = NULL WHERE channel_id = ?1 AND pubkey = ?2 AND removed_at IS NULL") + .bind(row.try_get::("id")?) + .bind(created_by) + .execute(&mut *tx) + .await?; + let record = channel_record(row)?; + tx.commit().await?; + return Ok(Some((record, false))); + } + + let id = Uuid::new_v4(); + let name = if participants.len() == 2 { + "DM".to_owned() + } else { + format!("Group DM ({})", participants.len()) + }; + sqlx::query("INSERT INTO channels (id, community_id, name, channel_type, visibility, participant_hash, created_by) VALUES (?1, ?2, ?3, 'dm', 'private', ?4, ?5)") + .bind(id.to_string()) + .bind(community.as_uuid().to_string()) + .bind(name) + .bind(hash.as_slice()) + .bind(created_by) + .execute(&mut *tx) + .await?; + for participant in participants { + sqlx::query("INSERT INTO channel_members (channel_id, pubkey, role, invited_by) VALUES (?1, ?2, 'member', ?3)") + .bind(id.to_string()) + .bind(participant) + .bind(created_by) + .execute(&mut *tx) + .await?; + } + let row = sqlx::query(select) + .bind(community.as_uuid().to_string()) + .bind(hash.as_slice()) + .fetch_one(&mut *tx) + .await?; + let record = channel_record(row)?; + tx.commit().await?; + Ok(Some((record, true))) +} + fn stored_event(row: sqlx::sqlite::SqliteRow) -> Result { let json: String = row.try_get("event_json")?; let event: nostr::Event = serde_json::from_str(&json)?; @@ -497,6 +881,28 @@ pub(crate) async fn get_accessible_channel_ids( .collect() } +pub(crate) async fn communities_of_channels( + pool: &SqlitePool, + channel_ids: &[Uuid], +) -> Result> { + let mut out = std::collections::HashMap::with_capacity(channel_ids.len()); + for channel_id in channel_ids { + let row = sqlx::query_scalar::<_, String>( + "SELECT community_id FROM channels WHERE id = ?1 AND deleted_at IS NULL", + ) + .bind(channel_id.to_string()) + .fetch_optional(pool) + .await?; + if let Some(community_id) = row { + let community_id = Uuid::parse_str(&community_id).map_err(|e| { + crate::DbError::InvalidData(format!("invalid SQLite community id: {e}")) + })?; + out.insert(*channel_id, CommunityId::from_uuid(community_id)); + } + } + Ok(out) +} + pub(crate) async fn get_member_role( pool: &SqlitePool, community: CommunityId, @@ -570,6 +976,29 @@ pub(crate) async fn ensure_user( == 1) } +pub(crate) async fn update_user_profile( + pool: &SqlitePool, + community: CommunityId, + pubkey: &[u8], + display_name: Option<&str>, + avatar_url: Option<&str>, + about: Option<&str>, + nip05_handle: Option<&str>, +) -> Result<()> { + sqlx::query( + "UPDATE users SET display_name = COALESCE(?3, display_name), avatar_url = COALESCE(?4, avatar_url), about = COALESCE(?5, about), nip05_handle = COALESCE(?6, nip05_handle) WHERE community_id = ?1 AND pubkey = ?2", + ) + .bind(community.as_uuid().to_string()) + .bind(pubkey) + .bind(display_name) + .bind(avatar_url) + .bind(about) + .bind(nip05_handle) + .execute(pool) + .await?; + Ok(()) +} + pub(crate) async fn get_user( pool: &SqlitePool, community: CommunityId, @@ -758,6 +1187,50 @@ mod tests { use super::*; use nostr::{EventBuilder, Keys, Kind}; + #[tokio::test] + async fn upgrades_phase_1_core_schema_before_dm_use() { + let path = std::env::temp_dir().join(format!("buzz-db-upgrade-{}.sqlite", Uuid::new_v4())); + let options = SqliteConnectOptions::from_str(&format!("sqlite://{}", path.display())) + .unwrap() + .create_if_missing(true) + .foreign_keys(true); + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + .unwrap(); + sqlx::query("CREATE TABLE communities (id TEXT PRIMARY KEY NOT NULL, host TEXT NOT NULL COLLATE NOCASE UNIQUE, icon TEXT, created_at INTEGER NOT NULL DEFAULT (unixepoch()), archived_at INTEGER)") + .execute(&pool).await.unwrap(); + sqlx::query("CREATE TABLE channels (id TEXT PRIMARY KEY NOT NULL, community_id TEXT NOT NULL, name TEXT NOT NULL, description TEXT, canvas TEXT, channel_type TEXT NOT NULL, visibility TEXT NOT NULL, created_by BLOB NOT NULL, created_at INTEGER NOT NULL DEFAULT (unixepoch()), updated_at INTEGER NOT NULL DEFAULT (unixepoch()), archived_at INTEGER, deleted_at INTEGER, nip29_group_id TEXT, topic_required INTEGER NOT NULL DEFAULT 0, max_members INTEGER, topic TEXT, topic_set_by BLOB, topic_set_at INTEGER, purpose TEXT, purpose_set_by BLOB, purpose_set_at INTEGER, ttl_seconds INTEGER, ttl_deadline INTEGER)") + .execute(&pool).await.unwrap(); + sqlx::query("CREATE TABLE channel_members (channel_id TEXT NOT NULL, pubkey BLOB NOT NULL, role TEXT NOT NULL, joined_at INTEGER NOT NULL DEFAULT (unixepoch()), invited_by BLOB, removed_at INTEGER, PRIMARY KEY (channel_id, pubkey))") + .execute(&pool).await.unwrap(); + pool.close().await; + + let upgraded = connect(path.to_str().unwrap()).await.unwrap(); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT version FROM schema_version WHERE singleton = 1") + .fetch_one(&upgraded) + .await + .unwrap(), + 2 + ); + for (table, column) in [ + ("channels", "participant_hash"), + ("channel_members", "hidden_at"), + ] { + let pragma = format!("PRAGMA table_info({table})"); + assert!(sqlx::query(sqlx::AssertSqlSafe(pragma)) + .fetch_all(&upgraded) + .await + .unwrap() + .iter() + .any(|row| row.get::("name") == column)); + } + upgraded.close().await; + std::fs::remove_file(path).unwrap(); + } + #[tokio::test] async fn core_slice_survives_temporary_file_reopen() { let path = std::env::temp_dir().join(format!("buzz-db-{}.sqlite", Uuid::new_v4())); diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 323b7b949..27605568f 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -29,7 +29,7 @@ pub mod conn_control; pub mod error; /// Redis-backed NIP-98 replay seen-set. pub mod nip98_replay; -pub use nip98_replay::RedisNip98ReplayGuard; +pub use nip98_replay::{InProcessNip98ReplayGuard, RedisNip98ReplayGuard}; /// Online/offline presence tracking in Redis. pub mod presence; /// Redis PUBLISH for channel event fan-out. diff --git a/crates/buzz-pubsub/src/nip98_replay.rs b/crates/buzz-pubsub/src/nip98_replay.rs index 858f0f5ac..99baa76a7 100644 --- a/crates/buzz-pubsub/src/nip98_replay.rs +++ b/crates/buzz-pubsub/src/nip98_replay.rs @@ -12,6 +12,76 @@ use buzz_auth::{ }; use nostr::EventId; +/// Process-local NIP-98 replay seen-set for a single relay process. +pub struct InProcessNip98ReplayGuard { + seen: dashmap::DashMap, + max_entries: usize, +} + +impl InProcessNip98ReplayGuard { + /// Create a bounded, expiring process-local replay guard. + pub fn new() -> Self { + Self { + seen: dashmap::DashMap::new(), + max_entries: 100_000, + } + } + + #[cfg(test)] + fn with_capacity(max_entries: usize) -> Self { + Self { + seen: dashmap::DashMap::new(), + max_entries, + } + } +} + +impl Default for InProcessNip98ReplayGuard { + fn default() -> Self { + Self::new() + } +} + +impl Nip98ReplayGuard for InProcessNip98ReplayGuard { + fn try_mark_in_scope<'a>( + &'a self, + scope: &'a str, + event_id: &'a EventId, + ttl_secs: u64, + ) -> std::pin::Pin> + Send + 'a>> + { + Box::pin(async move { + let ttl = std::time::Duration::from_secs( + ttl_secs.clamp(DEFAULT_REPLAY_TTL_SECS, MAX_REPLAY_TTL_SECS), + ); + let now = std::time::Instant::now(); + let key = nip98_replay_key_for_scope(scope, event_id); + if self.seen.len() >= self.max_entries && !self.seen.contains_key(&key) { + self.seen + .retain(|_, marked| now.duration_since(*marked) < ttl); + if self.seen.len() >= self.max_entries { + return Err(AuthError::Internal( + "process-local replay guard capacity exhausted".to_string(), + )); + } + } + match self.seen.entry(key) { + dashmap::mapref::entry::Entry::Vacant(entry) => { + entry.insert(now); + Ok(true) + } + dashmap::mapref::entry::Entry::Occupied(mut entry) + if now.duration_since(*entry.get()) >= ttl => + { + entry.insert(now); + Ok(true) + } + dashmap::mapref::entry::Entry::Occupied(_) => Ok(false), + } + }) + } +} + /// Redis-backed NIP-98 replay seen-set. /// /// Each `try_mark(ctx, event_id, ttl)` issues a single @@ -124,6 +194,29 @@ mod tests { .id } + #[tokio::test] + async fn in_process_guard_rejects_replay_and_fails_closed_at_capacity() { + let guard = InProcessNip98ReplayGuard::with_capacity(1); + let ctx = fresh_ctx(); + let first = fresh_event_id(); + assert!(guard + .try_mark(&ctx, &first, DEFAULT_REPLAY_TTL_SECS) + .await + .expect("first mark")); + assert!(!guard + .try_mark(&ctx, &first, DEFAULT_REPLAY_TTL_SECS) + .await + .expect("replay")); + + let second = fresh_event_id(); + assert!(guard + .try_mark(&ctx, &second, DEFAULT_REPLAY_TTL_SECS) + .await + .expect_err("full guard must fail closed") + .to_string() + .contains("capacity exhausted")); + } + #[tokio::test] #[ignore = "requires Redis"] async fn first_claim_succeeds_replay_fails() { diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index a118ff453..4981845b1 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1732,7 +1732,13 @@ async fn handle_bridge_search( .search .search(&search_query) .await - .map_err(|e| internal_error(&format!("search error: {e}")))?; + .map_err(|e| match e { + buzz_search::SearchError::Unsupported => api_error( + StatusCode::NOT_IMPLEMENTED, + "unsupported_feature: search is unavailable in this runtime profile", + ), + other => internal_error(&format!("search error: {other}")), + })?; // Fetch full events from DB by ID. Hit ids are already raw 32-byte // arrays from the FTS layer — no hex decode. diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 2d8273680..8f0cd0a25 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -80,8 +80,23 @@ pub async fn handle_command( enum PersistResult { /// Event was already processed — return idempotent success. Duplicate, - /// Event inserted — transaction is open, handler must commit after mutations. - Inserted(sqlx::Transaction<'static, sqlx::Postgres>), + /// Event inserted — PostgreSQL remains open until mutations finish; SQLite + /// committed the idempotency row atomically before the local mutation. + Inserted(CommandTransaction), +} + +enum CommandTransaction { + Sqlite, + Postgres(sqlx::Transaction<'static, sqlx::Postgres>), +} + +impl CommandTransaction { + async fn commit(self) -> Result<(), sqlx::Error> { + match self { + Self::Sqlite => Ok(()), + Self::Postgres(tx) => tx.commit().await, + } + } } /// Persist a command event inside a transaction. Returns the OPEN transaction @@ -105,6 +120,12 @@ async fn persist_command_event( ) -> Result { let channel_id = channel_id_override.or_else(|| extract_channel_id(event)); + if state.config.profile.is_single_node() { + return Err(IngestError::Rejected( + "unsupported_feature: this command is unavailable in the single-node profile".into(), + )); + } + let mut tx = state .db .begin_transaction() @@ -227,7 +248,7 @@ async fn persist_command_event( // Duplicate — rollback (implicit on drop) and signal idempotent success. Ok(PersistResult::Duplicate) } else { - Ok(PersistResult::Inserted(tx)) + Ok(PersistResult::Inserted(CommandTransaction::Postgres(tx))) } } @@ -345,27 +366,35 @@ async fn handle_dm_open( } } - // Persist the command event (idempotency) — returns open transaction - let tx = match persist_command_event(state, tenant, event, None).await? { - PersistResult::Duplicate => { - return Ok(IngestResult { - event_id: event.id.to_hex(), - accepted: true, - message: "duplicate: already processed".into(), - }); - } - PersistResult::Inserted(tx) => tx, - }; - - // 4. Execute: open_dm + // SQLite couples the command idempotency row and DM mutation in one + // transaction. PostgreSQL retains the existing open-transaction path. let all_refs: Vec<&[u8]> = all_bytes.iter().map(|b| b.as_slice()).collect(); - let (channel, was_created) = state + let sqlite_result = state .db - .open_dm(tenant.community(), &all_refs, &self_bytes) + .open_dm_sqlite_command(tenant.community(), &all_refs, &self_bytes, event) .await .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; + let (channel, was_created, tx) = if let Some(result) = sqlite_result { + (result.0, result.1, CommandTransaction::Sqlite) + } else { + let tx = match persist_command_event(state, tenant, event, None).await? { + PersistResult::Duplicate => { + return Ok(IngestResult { + event_id: event.id.to_hex(), + accepted: true, + message: "duplicate: already processed".into(), + }); + } + PersistResult::Inserted(tx) => tx, + }; + let (channel, was_created) = state + .db + .open_dm(tenant.community(), &all_refs, &self_bytes) + .await + .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; + (channel, was_created, tx) + }; - // Commit: event + mutation succeeded atomically. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index a67797385..cf295bdfa 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -525,7 +525,8 @@ async fn dispatch_persistent_event_inner( .iter() .any(|t| t.as_slice().first().map(|s| s.as_str()) == Some("buzz:workflow")); - if !buzz_core::kind::is_workflow_execution_kind(kind_u32) + if !state.config.profile.is_single_node() + && !buzz_core::kind::is_workflow_execution_kind(kind_u32) && !buzz_core::kind::is_command_kind(kind_u32) && !is_relay_workflow_msg && kind_u32 != KIND_GIFT_WRAP diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 88a9f0c73..ac24e74c2 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -809,6 +809,9 @@ pub fn emit_live_thread_summary( channel_id: Uuid, root_id: Vec, ) { + if state.config.profile.is_single_node() { + return; + } let tenant = tenant.clone(); let state = Arc::clone(state); tokio::spawn(async move { @@ -2932,6 +2935,12 @@ pub async fn publish_nip43_membership_list( tenant: &TenantContext, state: &Arc, ) -> anyhow::Result<()> { + // The production implementation relies on a PostgreSQL advisory lock for + // snapshot serialization. Single-node still emits the durable per-member + // delta above, but deliberately omits this denormalized list snapshot. + if state.config.profile.is_single_node() { + return Ok(()); + } let started_at = std::time::Instant::now(); metrics::counter!("buzz_nip43_membership_publications_total", "result" => "attempted") .increment(1); @@ -2990,6 +2999,11 @@ async fn publish_nip43_delta( target_pubkey_hex: &str, label: &str, ) -> anyhow::Result<()> { + // These are globally scoped events. Keep local relay membership private to + // the loopback process rather than exposing deltas to every subscriber. + if state.config.profile.is_single_node() { + return Ok(()); + } let relay_pubkey_hex = state.relay_keypair.public_key().to_hex(); let tags = vec![ diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 6c70f0b56..fb5aacc15 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -11,16 +11,17 @@ fn log_env_filter(rust_log: Option<&str>) -> EnvFilter { use uuid::Uuid; use buzz_audit::AuditService; +use buzz_auth::nip98_replay::Nip98ReplayGuard; use buzz_auth::AuthService; use buzz_core::CommunityId; use buzz_db::{Db, DbConfig}; -use buzz_pubsub::PubSubManager; +use buzz_pubsub::{rate_limiter::AdmissionRateLimiter, InProcessNip98ReplayGuard, PubSubManager}; use buzz_search::SearchService; use buzz_relay::config::{Config, MAX_DRAIN_JITTER_MS}; use buzz_relay::metrics as relay_metrics; use buzz_relay::router::{build_health_router, build_router}; -use buzz_relay::state::AppState; +use buzz_relay::state::{AppBackends, AppState}; use buzz_relay::storage_sweep; use buzz_relay::telemetry; use buzz_workflow::WorkflowEngine; @@ -153,13 +154,8 @@ async fn main() -> anyhow::Result<()> { "Config loaded" ); - // Phase 1 constructor fence: never fall through to production's eager - // Postgres/Redis/S3 graph for a single-node profile. The local backend - // constructor replaces this explicit denial as its implementations land. if config.profile.is_single_node() { - return Err(anyhow::anyhow!( - "BUZZ_PROFILE=single-node local backend bundle is not installed yet" - )); + return run_single_node(config, tracer_init).await; } let usage_interval_secs = usage_metrics_interval_secs(); @@ -1202,6 +1198,109 @@ async fn run_periodic_until_cancelled( /// │ SIGTERM → shutting_down=true → readiness 503 │ /// │ → graceful drain (30s) → exit │ /// └─────────────────────────────────────────────────────────┘ +/// Start the production router with only embedded/process-local backends. +async fn run_single_node(config: Config, tracer_init: telemetry::TracerInit) -> anyhow::Result<()> { + if !config.bind_addr.ip().is_loopback() { + return Err(anyhow::anyhow!( + "BUZZ_PROFILE=single-node requires a loopback BUZZ_BIND_ADDR, got {}", + config.bind_addr + )); + } + if config.require_relay_membership && config.relay_owner_pubkey.is_none() { + return Err(anyhow::anyhow!( + "RELAY_OWNER_PUBKEY required when BUZZ_REQUIRE_RELAY_MEMBERSHIP=true" + )); + } + if config.require_relay_membership && config.relay_private_key.is_none() { + return Err(anyhow::anyhow!( + "BUZZ_RELAY_PRIVATE_KEY is required when BUZZ_REQUIRE_RELAY_MEMBERSHIP=true" + )); + } + + relay_metrics::install(config.metrics_port, usage_metrics_idle_timeout_secs(60)); + let db_path = + std::env::var("BUZZ_LOCAL_DB").unwrap_or_else(|_| "buzz-local.sqlite".to_string()); + let media_root = + std::env::var("BUZZ_LOCAL_MEDIA_DIR").unwrap_or_else(|_| "buzz-local-media".to_string()); + let db = Db::new_sqlite(&db_path) + .await + .map_err(|e| anyhow::anyhow!("SQLite initialization failed: {e}"))?; + let host = buzz_relay::tenant::relay_url_authority(&config.relay_url); + if host.is_empty() { + return Err(anyhow::anyhow!( + "Cannot derive community host from BUZZ_RELAY_URL" + )); + } + let community = db.ensure_configured_community(&host).await?.id; + if let Some(owner) = config.relay_owner_pubkey.as_deref() { + db.bootstrap_owner(community, owner).await?; + } + + let relay_keypair = if let Some(hex) = &config.relay_private_key { + nostr::Keys::parse(hex) + .map_err(|e| anyhow::anyhow!("invalid BUZZ_RELAY_PRIVATE_KEY: {e}"))? + } else { + nostr::Keys::parse("0000000000000000000000000000000000000000000000000000000000000001") + .expect("hardcoded development key") + }; + let pubsub = Arc::new(PubSubManager::in_process()); + let workflow_engine = Arc::new(WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + // Git routes and probes are disabled by the profile. A never-used store + // satisfies the concrete state field until GitStore gains a local backend. + let git_store = buzz_relay::api::git::store::GitStore::new( + "http://127.0.0.1:1", + "local", + "local", + "disabled", + "local", + buzz_media::S3AddressingStyle::Path, + )?; + let replay: Arc = Arc::new(InProcessNip98ReplayGuard::new()); + let (app_state, audit_shutdown) = AppState::new_with_backends( + config, + AppBackends { + db, + redis_pool: None, + pubsub, + search: SearchService::unsupported(), + media_storage: buzz_media::MediaStorage::filesystem(&media_root), + git_store, + nip98_replay: replay, + admission_rate_limiter: Arc::new(AdmissionRateLimiter::permissive()), + }, + None::, + AuthService::new(buzz_auth::AuthConfig::default()), + workflow_engine, + relay_keypair, + ); + let state = Arc::new(app_state); + + let state_for_fanout = Arc::clone(&state); + let mut events = state.pubsub.subscribe_local(); + tokio::spawn(async move { + while let Ok(event) = events.recv().await { + buzz_relay::handlers::event::fan_out_pubsub_event(&state_for_fanout, event).await; + } + }); + + info!(database = %db_path, media = %media_root, community = %community, + "single-node profile ready; search, Git, push, workflows, reapers, usage and replica tasks disabled"); + let router = build_router(Arc::clone(&state)); + let health_router = build_health_router(Arc::clone(&state)); + serve(router, health_router, Arc::clone(&state)).await?; + state.community_revalidator_cancel.cancel(); + audit_shutdown + .drain(std::time::Duration::from_secs(5)) + .await; + if let telemetry::TracerInit::Enabled(tp) = tracer_init { + let _ = tp.shutdown(); + } + Ok(()) +} + /// ``` /// /// ## Shutdown budget