feat(relay): complete Phase 1 single-node slice

Signed-off-by: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>

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.

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<Vec<StoredEvent>> {
+        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<Vec<StoredEvent>> {
+        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<Option<StoredEvent>> {
-        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<bool> {
-        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<Uuid>,
         thread_meta: Option<event::ThreadMetadataParams<'_>>,
     ) -> 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<event::ReactionEventInsertOutcome> {
+        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<Option<(channel::ChannelRecord, bool)>> {
+        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<Option<thread::ThreadMetadataRecord>> {
+        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<bool> {
+        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<bool> {
+        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<DateTime<Utc>>,
         limit: i64,
     ) -> Result<Vec<StoredEvent>> {
+        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<DateTime<Utc>>,
         limit: i64,
     ) -> Result<Vec<StoredEvent>> {
+        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<DateTime<Utc>>,
         limit: i64,
     ) -> Result<Vec<StoredEvent>> {
+        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<Uuid>,
     ) -> 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<Uuid>,
     ) -> 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;

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<SqlitePool> {
@@ -135,9 +152,78 @@ pub(crate) async fn connect(path_or_url: &str) -> Result<SqlitePool> {
 }
 
 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::<String, _>("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<Uuid>,
+    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::<Vec<u8>, _>("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<bool> {
+    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<chrono::DateTime<chrono::Utc>>,
+    limit: i64,
+) -> Result<Vec<buzz_core::StoredEvent>> {
+    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<chrono::DateTime<chrono::Utc>>,
+    limit: i64,
+) -> Result<Vec<buzz_core::StoredEvent>> {
+    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<chrono::DateTime<chrono::Utc>>,
+    limit: i64,
+) -> Result<Vec<buzz_core::StoredEvent>> {
+    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<Uuid>,
+    target_event_id: &[u8],
+    actor_pubkey: &[u8],
+    emoji: &str,
+) -> Result<crate::event::ReactionEventInsertOutcome> {
+    let mut tx = pool.begin().await?;
+    let target_created_at: Option<i64> = 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<chrono::Utc>,
+    pubkey: &[u8],
+    emoji: &str,
+) -> Result<bool> {
+    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<bool> {
+    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<Option<(crate::channel::ChannelRecord, bool)>> {
+    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::<String, _>("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<buzz_core::StoredEvent> {
     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<std::collections::HashMap<Uuid, CommunityId>> {
+    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::<String, _>("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()));

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.

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<String, std::time::Instant>,
+    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<Box<dyn std::future::Future<Output = Result<bool, AuthError>> + 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() {

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.

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<PersistResult, IngestError> {
     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}")))?;

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

crates/buzz-relay/src/handlers/side_effects.rs

@@ -809,6 +809,9 @@ pub fn emit_live_thread_summary(
     channel_id: Uuid,
     root_id: Vec<u8>,
 ) {
+    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<AppState>,
 ) -> 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![

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<Tick, TickFuture>(
 /// │  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<dyn Nip98ReplayGuard> = 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::<AuditService>,
+        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