diff --git a/Cargo.lock b/Cargo.lock index f2c5f0050d..13c9a631eb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4549,6 +4549,7 @@ version = "0.30.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149" dependencies = [ + "cc", "pkg-config", "vcpkg", ] diff --git a/Cargo.toml b/Cargo.toml index be9cb5aa7b..69b6e1aad0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -53,7 +53,7 @@ tower-http = { version = "0.6", features = ["trace", "cors", "compression-gzip" # Database sqlx = { version = "0.9", features = [ - "runtime-tokio", "tls-rustls", "postgres", "uuid", "chrono", "json" + "runtime-tokio", "tls-rustls", "postgres", "sqlite", "uuid", "chrono", "json" ] } # Redis diff --git a/Justfile b/Justfile index 0a43249d5f..598027c71d 100644 --- a/Justfile +++ b/Justfile @@ -155,7 +155,7 @@ _ensure-sidecar-stubs: set -euo pipefail TARGET=$(rustc -vV | sed -n 's|host: ||p') mkdir -p desktop/src-tauri/binaries - SIDECARS=(buzz-acp buzz-agent buzz-dev-mcp git-credential-nostr buzz) + SIDECARS=(buzz-acp buzz-agent buzz-dev-mcp git-credential-nostr buzz buzz-relay) if [[ "$TARGET" != *windows* ]]; then SIDECARS+=(buzz-backend-kubernetes) fi @@ -258,6 +258,7 @@ desktop-release-build target="aarch64-apple-darwin": touch "desktop/src-tauri/binaries/buzz-dev-mcp-$TARGET" touch "desktop/src-tauri/binaries/git-credential-nostr-$TARGET" touch "desktop/src-tauri/binaries/buzz-$TARGET" + touch "desktop/src-tauri/binaries/buzz-relay-$TARGET" pnpm install cd {{desktop_dir}} && pnpm tauri build --features mesh-llm --target {{target}} @@ -305,6 +306,10 @@ test-unit: # #[ignore]d, so --lib runs only the infra-free set. Without this gate a # stray file in migrations/ or a broken lint ships green. cargo nextest run -p buzz-db --lib + # Embedded single-node community foundation. This is intentionally + # narrower than the PostgreSQL suite: only SQLite community, channel, + # membership, and relay-owner durability belongs to PR 1A. + cargo nextest run -p buzz-db sqlite::tests # Multi-tenant conformance gate (buzz-conformance): the independent # replay checker + golden fixtures. No infra — pure in-process trace # replay — so it belongs in the unit job. Run all targets (lib + the @@ -503,10 +508,10 @@ desktop-standalone *ARGS: _ensure-sidecar-stubs #!/usr/bin/env bash set -euo pipefail export PATH="{{justfile_directory()}}/bin:$PATH" - cargo build -p buzz-acp -p buzz-agent -p buzz-backend-kubernetes -p buzz-dev-mcp -p buzz-cli -p git-credential-nostr + cargo build -p buzz-acp -p buzz-agent -p buzz-backend-kubernetes -p buzz-dev-mcp -p buzz-cli -p git-credential-nostr -p buzz-relay TARGET=$(rustc -vV | sed -n 's|host: ||p') TARGET_DIR=$(cargo metadata --format-version 1 --no-deps | node -p "JSON.parse(require('fs').readFileSync(0, 'utf8')).target_directory") - for bin in buzz-acp buzz-agent buzz-backend-kubernetes buzz-dev-mcp git-credential-nostr buzz; do + for bin in buzz-acp buzz-agent buzz-backend-kubernetes buzz-dev-mcp git-credential-nostr buzz buzz-relay; do cp "${TARGET_DIR}/debug/${bin}" "desktop/src-tauri/binaries/${bin}-${TARGET}" chmod +x "desktop/src-tauri/binaries/${bin}-${TARGET}" done diff --git a/PHASE1C_BREAKAGE_CATALOGUE.md b/PHASE1C_BREAKAGE_CATALOGUE.md new file mode 100644 index 0000000000..06fd2f3365 --- /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/error.rs b/crates/buzz-db/src/error.rs index f8b8a2eb56..2dbfa210ab 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -13,6 +13,10 @@ pub enum DbError { #[error("migration error: {0}")] Migrate(#[from] sqlx::migrate::MigrateError), + /// Operation is unsupported by the selected database backend. + #[error("unsupported database backend operation: {0}")] + UnsupportedBackend(&'static str), + /// Attempted to store an AUTH event (kind 22242), which is forbidden. #[error("AUTH events (kind 22242) must not be stored")] AuthEventRejected, diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9b26876747..6de335aa42 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -45,6 +45,7 @@ pub mod relay_invite; pub mod relay_members; /// Replica freshness fence for keyset-cursor read routing. pub mod replica_fence; +mod sqlite; /// Thread metadata persistence. pub mod thread; /// Per-community usage rollup queries for Prometheus gauges. @@ -168,9 +169,17 @@ pub async fn insert_mentions( Ok(()) } +/// Selected backing store for a [`Db`] handle. +#[derive(Clone, Debug)] +enum DbBackend { + Postgres, + SQLite(sqlx::SqlitePool), +} + /// Database handle. Clone is cheap (Arc-backed pool). #[derive(Clone, Debug)] pub struct Db { + backend: DbBackend, pub(crate) pool: PgPool, /// Maximum connections configured for this pool (from [`DbConfig::max_connections`]). pub(crate) max_connections: u32, @@ -639,6 +648,32 @@ pub struct TokenSummary { } impl Db { + fn sqlite_pool(&self) -> Result<&sqlx::SqlitePool> { + match &self.backend { + DbBackend::SQLite(pool) => Ok(pool), + DbBackend::Postgres => Err(DbError::UnsupportedBackend( + "SQLite operation on PostgreSQL Db", + )), + } + } + + /// Creates a local single-node database backed by SQLite. + pub async fn new_sqlite(database_url: &str) -> Result { + let sqlite = sqlite::connect(database_url).await?; + let pool = PgPoolOptions::new() + .max_connections(1) + .connect_lazy("postgres://unused:unused@127.0.0.1/unused")?; + Ok(Self { + backend: DbBackend::SQLite(sqlite), + pool, + max_connections: 0, + read_pool: None, + read_max_connections: 0, + fence: std::sync::Arc::new(replica_fence::ReplicaFence::new()), + replica_read_max_age: None, + reader_aurora_identity: std::sync::Arc::new(std::sync::OnceLock::new()), + }) + } /// Creates a new `Db` by connecting a Postgres pool with the given config. /// /// When `config.read_database_url` is set, a second pool with the same @@ -659,6 +694,7 @@ impl Db { }; let replica_read_max_age = read_budget_from_ms(config.replica_read_max_age_ms); Ok(Self { + backend: DbBackend::Postgres, pool, max_connections: config.max_connections, read_pool, @@ -774,6 +810,7 @@ impl Db { /// Creates a `Db` from an existing `PgPool` (useful in tests). pub fn from_pool(pool: PgPool) -> Self { Self { + backend: DbBackend::Postgres, max_connections: pool.options().get_max_connections(), read_max_connections: pool.options().get_max_connections(), pool, @@ -793,6 +830,7 @@ impl Db { /// [`Db::fence`]). pub fn from_pools(pool: PgPool, read_pool: PgPool) -> Self { Self { + backend: DbBackend::Postgres, max_connections: pool.options().get_max_connections(), read_max_connections: read_pool.options().get_max_connections(), pool, @@ -1010,12 +1048,18 @@ impl Db { /// Run pending database migrations. pub async fn migrate(&self) -> Result<()> { - migration::run_migrations(&self.pool).await + match &self.backend { + DbBackend::Postgres => migration::run_migrations(&self.pool).await, + DbBackend::SQLite(_) => sqlite::migrate(self.sqlite_pool()?).await, + } } /// Returns `true` if the database is reachable (used by readiness probes). pub async fn ping(&self) -> bool { - sqlx::query("SELECT 1").execute(&self.pool).await.is_ok() + match &self.backend { + DbBackend::Postgres => sqlx::query("SELECT 1").execute(&self.pool).await.is_ok(), + DbBackend::SQLite(pool) => sqlx::query("SELECT 1").execute(pool).await.is_ok(), + } } /// Returns pool utilisation stats for metrics emission. @@ -1196,6 +1240,9 @@ impl Db { &self, normalized_host: &str, ) -> Result> { + if matches!(&self.backend, DbBackend::SQLite(_)) { + return sqlite::lookup_community_by_host(self.sqlite_pool()?, normalized_host).await; + } let row = sqlx::query( r#" SELECT id, host @@ -1222,6 +1269,9 @@ impl Db { /// Returns whether a community id still exists in the active lifecycle state. pub async fn is_community_active(&self, community_id: CommunityId) -> Result { + if matches!(&self.backend, DbBackend::SQLite(_)) { + return sqlite::is_community_active(self.sqlite_pool()?, community_id).await; + } let active = sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM communities WHERE id = $1 AND archived_at IS NULL)", ) @@ -1299,6 +1349,9 @@ impl Db { /// community is authoritative; the host is read back for labelling only and /// is never used to re-derive the community. pub async fn lookup_community_host(&self, community_id: CommunityId) -> Result> { + if matches!(&self.backend, DbBackend::SQLite(_)) { + return sqlite::lookup_community_host(self.sqlite_pool()?, community_id).await; + } let row = sqlx::query( r#" SELECT host @@ -1318,6 +1371,22 @@ impl Db { .transpose() } + /// Rebinds a stale loopback community to the current single-node authority. + pub async fn rebind_single_node_community_host( + &self, + normalized_host: &str, + owner_pubkey: &str, + ) -> Result> { + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::rebind_single_node_community_host(pool, normalized_host, owner_pubkey).await + } + DbBackend::Postgres => Err(DbError::UnsupportedBackend( + "single-node community rebind on PostgreSQL Db", + )), + } + } + /// Returns the community's workspace icon (NIP-11 `icon`), if set. /// /// Set by relay admins/owners via the kind:9033 command; the value is @@ -1370,6 +1439,9 @@ impl Db { &self, normalized_host: &str, ) -> Result { + if matches!(&self.backend, DbBackend::SQLite(_)) { + return sqlite::ensure_configured_community(self.sqlite_pool()?, normalized_host).await; + } let row = sqlx::query( r#" INSERT INTO communities (host) @@ -2165,6 +2237,20 @@ impl Db { created_by: &[u8], ttl_seconds: Option, ) -> Result<(channel::ChannelRecord, bool)> { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::create_channel_with_id( + pool, + community_id, + channel_id, + name, + channel_type, + visibility, + description, + created_by, + ttl_seconds, + ) + .await; + } channel::create_channel_with_id( &self.pool, community_id, @@ -2185,7 +2271,10 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result { - channel::get_channel(&self.pool, community_id, channel_id).await + match &self.backend { + DbBackend::SQLite(pool) => sqlite::get_channel(pool, community_id, channel_id).await, + DbBackend::Postgres => channel::get_channel(&self.pool, community_id, channel_id).await, + } } /// Returns the canvas content for a channel, if any. @@ -2216,6 +2305,10 @@ impl Db { role: channel::MemberRole, invited_by: Option<&[u8]>, ) -> Result { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::add_member(pool, community_id, channel_id, pubkey, role, invited_by) + .await; + } channel::add_member( &self.pool, community_id, @@ -2245,7 +2338,14 @@ impl Db { channel_id: Uuid, pubkey: &[u8], ) -> Result { - channel::is_member(&self.pool, community_id, channel_id, pubkey).await + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::is_member(pool, community_id, channel_id, pubkey).await + } + DbBackend::Postgres => { + channel::is_member(&self.pool, community_id, channel_id, pubkey).await + } + } } /// Return the active (channel, pubkey) membership pairs among the given @@ -2265,7 +2365,10 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result> { - channel::get_members(&self.pool, community_id, channel_id).await + match &self.backend { + DbBackend::SQLite(pool) => sqlite::get_members(pool, community_id, channel_id).await, + DbBackend::Postgres => channel::get_members(&self.pool, community_id, channel_id).await, + } } /// Returns active members for multiple channels in a single query. @@ -2283,7 +2386,14 @@ impl Db { community_id: CommunityId, pubkey: &[u8], ) -> Result> { - channel::get_accessible_channel_ids(&self.pool, community_id, pubkey).await + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::get_accessible_channel_ids(pool, community_id, pubkey).await + } + DbBackend::Postgres => { + channel::get_accessible_channel_ids(&self.pool, community_id, pubkey).await + } + } } /// Lists channels, optionally filtered by visibility. @@ -4007,6 +4117,9 @@ impl Db { /// [`Db::query_events_routed_bounded`]. Not precedent for routing other /// permission reads. pub async fn is_relay_member(&self, community: CommunityId, pubkey: &str) -> Result { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::is_relay_member(pool, community, pubkey).await; + } let path = "relay_membership"; match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { @@ -4034,6 +4147,9 @@ impl Db { community: CommunityId, pubkey: &str, ) -> Result> { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::get_relay_member(pool, community, pubkey).await; + } relay_members::get_relay_member(&self.pool, community, pubkey).await } @@ -4042,6 +4158,9 @@ impl Db { &self, community: CommunityId, ) -> Result> { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::list_relay_members(pool, community).await; + } relay_members::list_relay_members(&self.pool, community).await } @@ -4056,6 +4175,9 @@ impl Db { role: &str, added_by: Option<&str>, ) -> Result { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::add_relay_member(pool, community, pubkey, role, added_by).await; + } relay_members::add_relay_member(&self.pool, community, pubkey, role, added_by).await } @@ -4118,6 +4240,9 @@ impl Db { /// Ensures the owner pubkey exists with role `"owner"` in `community`. Called at startup. pub async fn bootstrap_owner(&self, community: CommunityId, owner_pubkey: &str) -> Result<()> { + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::bootstrap_owner(pool, community, owner_pubkey).await; + } relay_members::bootstrap_owner(&self.pool, community, owner_pubkey).await } diff --git a/crates/buzz-db/src/sqlite.rs b/crates/buzz-db/src/sqlite.rs new file mode 100644 index 0000000000..c6103ae85d --- /dev/null +++ b/crates/buzz-db/src/sqlite.rs @@ -0,0 +1,830 @@ +//! SQLite storage for the single-node community profile. +//! +//! This intentionally covers only community bootstrap, channels, memberships, +//! and relay ownership. Event and user persistence lands with PR 2. + +use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; +use sqlx::{Row, SqlitePool}; +use std::str::FromStr; +use uuid::Uuid; + +use buzz_core::CommunityId; + +use crate::{CommunityRecord, EnsuredCommunityRecord, Result}; + +const SCHEMA: &str = r#" +CREATE TABLE IF NOT EXISTS 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 +); + +CREATE TABLE IF NOT EXISTS relay_members ( + community_id TEXT NOT NULL, + pubkey TEXT NOT NULL COLLATE NOCASE, + role TEXT NOT NULL, + added_by TEXT, + created_at INTEGER NOT NULL DEFAULT (unixepoch()), + updated_at INTEGER NOT NULL DEFAULT (unixepoch()), + PRIMARY KEY (community_id, pubkey), + FOREIGN KEY (community_id) REFERENCES communities(id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS 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, + participant_hash BLOB, + 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, + FOREIGN KEY (community_id) REFERENCES communities(id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS 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, + hidden_at INTEGER, + removed_at INTEGER, + PRIMARY KEY (channel_id, pubkey), + FOREIGN KEY (channel_id) REFERENCES channels(id) ON DELETE CASCADE +); +"#; + +pub(crate) async fn connect(path_or_url: &str) -> Result { + let database_url = if path_or_url.starts_with("sqlite:") { + path_or_url.to_owned() + } else { + format!("sqlite://{path_or_url}") + }; + let options = SqliteConnectOptions::from_str(&database_url)? + .create_if_missing(true) + .foreign_keys(true); + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect_with(options) + .await?; + migrate(&pool).await?; + Ok(pool) +} + +pub(crate) async fn migrate(pool: &SqlitePool) -> Result<()> { + for statement in SCHEMA.split(';').map(str::trim).filter(|s| !s.is_empty()) { + sqlx::query(statement).execute(pool).await?; + } + Ok(()) +} + +pub(crate) async fn lookup_community_by_host( + pool: &SqlitePool, + normalized_host: &str, +) -> Result> { + let row = sqlx::query( + "SELECT id, host FROM communities WHERE host = ?1 COLLATE NOCASE AND archived_at IS NULL", + ) + .bind(normalized_host) + .fetch_optional(pool) + .await?; + row.map(community_record).transpose() +} + +pub(crate) async fn lookup_community_host( + pool: &SqlitePool, + community_id: CommunityId, +) -> Result> { + Ok( + sqlx::query_scalar("SELECT host FROM communities WHERE id = ?1 AND archived_at IS NULL") + .bind(community_id.as_uuid().to_string()) + .fetch_optional(pool) + .await?, + ) +} + +pub(crate) async fn is_community_active( + pool: &SqlitePool, + community_id: CommunityId, +) -> Result { + let count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM communities WHERE id = ?1 AND archived_at IS NULL", + ) + .bind(community_id.as_uuid().to_string()) + .fetch_one(pool) + .await?; + Ok(count != 0) +} + +pub(crate) async fn rebind_single_node_community_host( + pool: &SqlitePool, + normalized_host: &str, + owner_pubkey: &str, +) -> Result> { + let mut tx = pool.begin().await?; + + // Defect-era databases may contain several loopback communities, one per + // ephemeral port. Prefer a community owned by this desktop identity, then + // the oldest community. This keeps ownership stable without depending on + // event persistence, which arrives with PR 2. + let row = sqlx::query( + r#"SELECT c.id, c.host + FROM communities c + LEFT JOIN relay_members rm + ON rm.community_id = c.id + AND lower(rm.pubkey) = lower(?1) + AND rm.role = 'owner' + WHERE c.archived_at IS NULL + AND c.host LIKE '127.0.0.1:%' + ORDER BY CASE WHEN rm.pubkey IS NULL THEN 0 ELSE 1 END DESC, c.created_at ASC, c.id ASC + LIMIT 1"#, + ) + .bind(owner_pubkey) + .fetch_optional(&mut *tx) + .await?; + let Some(row) = row else { + tx.commit().await?; + return Ok(None); + }; + let record = community_record(row)?; + if record.host.eq_ignore_ascii_case(normalized_host) { + tx.commit().await?; + return Ok(Some(record)); + } + + // The OS may reuse a port that belongs to another stale defect-era row. + // Swap that collision to the selected row's old authority transactionally, + // using a temporary unique host to satisfy the case-insensitive constraint. + if let Some(collision_id) = sqlx::query_scalar::<_, String>( + "SELECT id FROM communities WHERE host = ?1 COLLATE NOCASE AND id <> ?2", + ) + .bind(normalized_host) + .bind(record.id.as_uuid().to_string()) + .fetch_optional(&mut *tx) + .await? + { + let temporary_host = format!("rebind-{}.invalid", Uuid::new_v4()); + sqlx::query("UPDATE communities SET host = ?2 WHERE id = ?1") + .bind(&collision_id) + .bind(&temporary_host) + .execute(&mut *tx) + .await?; + sqlx::query("UPDATE communities SET host = ?2 WHERE id = ?1") + .bind(record.id.as_uuid().to_string()) + .bind(normalized_host) + .execute(&mut *tx) + .await?; + sqlx::query("UPDATE communities SET host = ?2 WHERE id = ?1") + .bind(collision_id) + .bind(&record.host) + .execute(&mut *tx) + .await?; + } else { + sqlx::query("UPDATE communities SET host = ?2 WHERE id = ?1") + .bind(record.id.as_uuid().to_string()) + .bind(normalized_host) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + Ok(Some(CommunityRecord { + id: record.id, + host: normalized_host.to_string(), + })) +} + +pub(crate) async fn ensure_configured_community( + pool: &SqlitePool, + normalized_host: &str, +) -> Result { + let id = Uuid::new_v4(); + let inserted = + sqlx::query("INSERT INTO communities (id, host) VALUES (?1, ?2) ON CONFLICT DO NOTHING") + .bind(id.to_string()) + .bind(normalized_host) + .execute(pool) + .await? + .rows_affected() + == 1; + + let row = sqlx::query("SELECT id, host FROM communities WHERE host = ?1 COLLATE NOCASE") + .bind(normalized_host) + .fetch_one(pool) + .await?; + let record = community_record(row)?; + Ok(EnsuredCommunityRecord { + id: record.id, + host: record.host, + created: inserted, + }) +} +#[allow(clippy::too_many_arguments)] +pub(crate) async fn create_channel_with_id( + pool: &SqlitePool, + community: CommunityId, + channel_id: Uuid, + name: &str, + channel_type: crate::channel::ChannelType, + visibility: crate::channel::ChannelVisibility, + description: Option<&str>, + created_by: &[u8], + ttl_seconds: Option, +) -> Result<(crate::channel::ChannelRecord, bool)> { + if created_by.len() != 32 { + return Err(crate::DbError::InvalidData(format!( + "pubkey must be 32 bytes, got {}", + created_by.len() + ))); + } + if channel_id.is_nil() { + return Err(crate::DbError::InvalidData( + "channel_id must not be nil (reserved for global fan-out)".into(), + )); + } + let name = buzz_core::channel::canonical_channel_name(name); + if name.trim().is_empty() { + return Err(crate::DbError::InvalidData( + "channel name is required".into(), + )); + } + let mut tx = pool.begin().await?; + let inserted = sqlx::query( + "INSERT INTO channels (id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, CASE WHEN ?8 IS NULL THEN NULL ELSE unixepoch() + ?8 END) ON CONFLICT DO NOTHING", + ) + .bind(channel_id.to_string()).bind(community.as_uuid().to_string()).bind(&name) + .bind(channel_type.as_str()).bind(visibility.as_str()).bind(description).bind(created_by) + .bind(ttl_seconds).execute(&mut *tx).await?.rows_affected() == 1; + if inserted { + sqlx::query("INSERT INTO channel_members (channel_id, pubkey, role, invited_by) VALUES (?1, ?2, 'owner', ?2)") + .bind(channel_id.to_string()).bind(created_by).execute(&mut *tx).await?; + } + tx.commit().await?; + Ok((get_channel(pool, community, channel_id).await?, inserted)) +} + +pub(crate) async fn get_channel( + pool: &SqlitePool, + community: CommunityId, + channel_id: Uuid, +) -> Result { + let row = sqlx::query( + "SELECT * FROM channels WHERE community_id = ?1 AND id = ?2 AND deleted_at IS NULL", + ) + .bind(community.as_uuid().to_string()) + .bind(channel_id.to_string()) + .fetch_optional(pool) + .await? + .ok_or(crate::DbError::ChannelNotFound(channel_id))?; + channel_record(row) +} + +pub(crate) async fn add_member( + pool: &SqlitePool, + community: CommunityId, + channel_id: Uuid, + pubkey: &[u8], + role: crate::channel::MemberRole, + invited_by: Option<&[u8]>, +) -> Result { + if pubkey.len() != 32 { + return Err(crate::DbError::InvalidData(format!( + "pubkey must be 32 bytes, got {}", + pubkey.len() + ))); + } + + let mut tx = pool.begin().await?; + let channel = sqlx::query( + "SELECT * FROM channels WHERE community_id = ?1 AND id = ?2 AND deleted_at IS NULL", + ) + .bind(community.as_uuid().to_string()) + .bind(channel_id.to_string()) + .fetch_optional(&mut *tx) + .await? + .ok_or(crate::DbError::ChannelNotFound(channel_id))?; + let channel = channel_record(channel)?; + + let effective_role = if channel.visibility == "private" { + let inviter = invited_by.ok_or_else(|| { + crate::DbError::AccessDenied("private channel requires an invite".to_string()) + })?; + let creator_bootstrap = inviter == pubkey && inviter == channel.created_by.as_slice(); + if !creator_bootstrap { + let inviter_role = active_member_role(&mut tx, channel_id, inviter) + .await? + .ok_or_else(|| { + crate::DbError::AccessDenied("inviter is not an active member".to_string()) + })?; + if role.is_elevated() && !is_elevated_role(&inviter_role) { + return Err(crate::DbError::AccessDenied( + "only owners/admins may grant elevated roles".to_string(), + )); + } + } + role + } else if role.is_elevated() { + let granter_role = match invited_by { + Some(inviter) => active_member_role(&mut tx, channel_id, inviter).await?, + None => None, + }; + if !granter_role.is_some_and(|role| is_elevated_role(&role)) { + return Err(crate::DbError::AccessDenied( + "only owners/admins may grant elevated roles".to_string(), + )); + } + role + } else { + role + }; + + if let Some(current_role) = active_member_role(&mut tx, channel_id, pubkey).await? { + if current_role != effective_role.as_str() { + let actor_role = match invited_by { + Some(inviter) => active_member_role(&mut tx, channel_id, inviter).await?, + None => None, + }; + if !actor_role.is_some_and(|role| is_elevated_role(&role)) { + return Err(crate::DbError::AccessDenied( + "only owners/admins may change an active member's role".to_string(), + )); + } + if current_role == "owner" && effective_role != crate::channel::MemberRole::Owner { + let owner_count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM channel_members WHERE channel_id = ?1 AND role = 'owner' AND removed_at IS NULL", + ) + .bind(channel_id.to_string()) + .fetch_one(&mut *tx) + .await?; + if owner_count <= 1 { + return Err(crate::DbError::AccessDenied( + "cannot demote the last owner — transfer ownership first".to_string(), + )); + } + } + } + } + + sqlx::query("INSERT INTO channel_members (channel_id, pubkey, role, invited_by) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(channel_id, pubkey) DO UPDATE SET role = excluded.role, removed_at = NULL") + .bind(channel_id.to_string()).bind(pubkey).bind(effective_role.as_str()).bind(invited_by).execute(&mut *tx).await?; + let row = sqlx::query("SELECT channel_id, pubkey, role, joined_at, invited_by, removed_at FROM channel_members WHERE channel_id = ?1 AND pubkey = ?2") + .bind(channel_id.to_string()).bind(pubkey).fetch_one(&mut *tx).await?; + let record = member_record(row)?; + tx.commit().await?; + Ok(record) +} + +async fn active_member_role( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + channel_id: Uuid, + pubkey: &[u8], +) -> Result> { + sqlx::query_scalar( + "SELECT role FROM channel_members WHERE channel_id = ?1 AND pubkey = ?2 AND removed_at IS NULL", + ) + .bind(channel_id.to_string()) + .bind(pubkey) + .fetch_optional(&mut **tx) + .await + .map_err(Into::into) +} + +fn is_elevated_role(role: &str) -> bool { + matches!(role, "owner" | "admin") +} + +pub(crate) async fn is_member( + pool: &SqlitePool, + community: CommunityId, + channel_id: Uuid, + pubkey: &[u8], +) -> Result { + Ok(sqlx::query_scalar::<_, i64>("SELECT count(*) FROM channel_members cm JOIN channels c ON c.id = cm.channel_id WHERE c.community_id = ?1 AND c.id = ?2 AND cm.pubkey = ?3 AND cm.removed_at IS NULL AND c.deleted_at IS NULL") + .bind(community.as_uuid().to_string()).bind(channel_id.to_string()).bind(pubkey).fetch_one(pool).await? != 0) +} + +pub(crate) async fn get_members( + pool: &SqlitePool, + community: CommunityId, + channel_id: Uuid, +) -> Result> { + get_channel(pool, community, channel_id).await?; + sqlx::query("SELECT channel_id, pubkey, role, joined_at, invited_by, removed_at FROM channel_members WHERE channel_id = ?1 AND removed_at IS NULL ORDER BY joined_at, pubkey") + .bind(channel_id.to_string()).fetch_all(pool).await?.into_iter().map(member_record).collect() +} + +pub(crate) async fn get_accessible_channel_ids( + pool: &SqlitePool, + community: CommunityId, + pubkey: &[u8], +) -> Result> { + let rows = sqlx::query_scalar::<_, String>("SELECT c.id FROM channels c JOIN channel_members cm ON cm.channel_id = c.id WHERE c.community_id = ?1 AND cm.pubkey = ?2 AND cm.removed_at IS NULL AND c.deleted_at IS NULL ORDER BY c.created_at, c.id") + .bind(community.as_uuid().to_string()).bind(pubkey).fetch_all(pool).await?; + rows.into_iter() + .map(|id| { + Uuid::parse_str(&id) + .map_err(|e| crate::DbError::InvalidData(format!("invalid SQLite channel id: {e}"))) + }) + .collect() +} + +fn timestamp(value: i64) -> Result> { + chrono::DateTime::from_timestamp(value, 0).ok_or(crate::DbError::InvalidTimestamp(value)) +} +fn optional_timestamp(value: Option) -> Result>> { + value.map(timestamp).transpose() +} +fn channel_record(row: sqlx::sqlite::SqliteRow) -> Result { + let id: String = row.try_get("id")?; + Ok(crate::channel::ChannelRecord { + id: Uuid::parse_str(&id) + .map_err(|e| crate::DbError::InvalidData(format!("invalid SQLite channel id: {e}")))?, + name: row.try_get("name")?, + channel_type: row.try_get("channel_type")?, + visibility: row.try_get("visibility")?, + description: row.try_get("description")?, + canvas: row.try_get("canvas")?, + created_by: row.try_get("created_by")?, + created_at: timestamp(row.try_get("created_at")?)?, + updated_at: timestamp(row.try_get("updated_at")?)?, + archived_at: optional_timestamp(row.try_get("archived_at")?)?, + deleted_at: optional_timestamp(row.try_get("deleted_at")?)?, + nip29_group_id: row.try_get("nip29_group_id")?, + topic_required: row.try_get::("topic_required")? != 0, + max_members: row.try_get("max_members")?, + topic: row.try_get("topic")?, + topic_set_by: row.try_get("topic_set_by")?, + topic_set_at: optional_timestamp(row.try_get("topic_set_at")?)?, + purpose: row.try_get("purpose")?, + purpose_set_by: row.try_get("purpose_set_by")?, + purpose_set_at: optional_timestamp(row.try_get("purpose_set_at")?)?, + ttl_seconds: row.try_get("ttl_seconds")?, + ttl_deadline: optional_timestamp(row.try_get("ttl_deadline")?)?, + }) +} +fn member_record(row: sqlx::sqlite::SqliteRow) -> Result { + let id: String = row.try_get("channel_id")?; + Ok(crate::channel::MemberRecord { + channel_id: Uuid::parse_str(&id) + .map_err(|e| crate::DbError::InvalidData(format!("invalid SQLite channel id: {e}")))?, + pubkey: row.try_get("pubkey")?, + role: row.try_get("role")?, + joined_at: timestamp(row.try_get("joined_at")?)?, + invited_by: row.try_get("invited_by")?, + removed_at: optional_timestamp(row.try_get("removed_at")?)?, + }) +} +pub(crate) async fn is_relay_member( + pool: &SqlitePool, + community: CommunityId, + pubkey: &str, +) -> Result { + Ok(sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM relay_members WHERE community_id = ?1 AND pubkey = ?2 COLLATE NOCASE", + ) + .bind(community.as_uuid().to_string()) + .bind(pubkey) + .fetch_one(pool) + .await? + != 0) +} + +pub(crate) async fn get_relay_member( + pool: &SqlitePool, + community: CommunityId, + pubkey: &str, +) -> Result> { + let row = sqlx::query( + "SELECT pubkey, role, added_by, created_at, updated_at FROM relay_members WHERE community_id = ?1 AND pubkey = ?2 COLLATE NOCASE", + ) + .bind(community.as_uuid().to_string()) + .bind(pubkey) + .fetch_optional(pool) + .await?; + row.map(relay_member).transpose() +} + +pub(crate) async fn list_relay_members( + pool: &SqlitePool, + community: CommunityId, +) -> Result> { + sqlx::query( + "SELECT pubkey, role, added_by, created_at, updated_at FROM relay_members WHERE community_id = ?1 ORDER BY created_at ASC, pubkey ASC", + ) + .bind(community.as_uuid().to_string()) + .fetch_all(pool) + .await? + .into_iter() + .map(relay_member) + .collect() +} + +pub(crate) async fn add_relay_member( + pool: &SqlitePool, + community: CommunityId, + pubkey: &str, + role: &str, + added_by: Option<&str>, +) -> Result { + Ok(sqlx::query( + "INSERT INTO relay_members (community_id, pubkey, role, added_by) VALUES (?1, lower(?2), ?3, ?4) ON CONFLICT DO NOTHING", + ) + .bind(community.as_uuid().to_string()) + .bind(pubkey) + .bind(role) + .bind(added_by) + .execute(pool) + .await? + .rows_affected() == 1) +} + +pub(crate) async fn bootstrap_owner( + pool: &SqlitePool, + community: CommunityId, + owner_pubkey: &str, +) -> Result<()> { + let mut tx = pool.begin().await?; + sqlx::query( + "UPDATE relay_members SET role = 'admin', updated_at = unixepoch() WHERE community_id = ?1 AND role = 'owner' AND pubkey <> ?2 COLLATE NOCASE", + ) + .bind(community.as_uuid().to_string()) + .bind(owner_pubkey) + .execute(&mut *tx) + .await?; + sqlx::query( + "INSERT INTO relay_members (community_id, pubkey, role) VALUES (?1, lower(?2), 'owner') ON CONFLICT(community_id, pubkey) DO UPDATE SET role = 'owner', updated_at = unixepoch()", + ) + .bind(community.as_uuid().to_string()) + .bind(owner_pubkey) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) +} + +fn relay_member(row: sqlx::sqlite::SqliteRow) -> Result { + let created_at: i64 = row.try_get("created_at")?; + let updated_at: i64 = row.try_get("updated_at")?; + Ok(crate::relay_members::RelayMember { + pubkey: row.try_get("pubkey")?, + role: row.try_get("role")?, + added_by: row.try_get("added_by")?, + created_at: chrono::DateTime::from_timestamp(created_at, 0) + .ok_or(crate::DbError::InvalidTimestamp(created_at))?, + updated_at: chrono::DateTime::from_timestamp(updated_at, 0) + .ok_or(crate::DbError::InvalidTimestamp(updated_at))?, + }) +} + +fn community_record(row: sqlx::sqlite::SqliteRow) -> Result { + let id: String = row.try_get("id")?; + let id = Uuid::parse_str(&id) + .map_err(|e| crate::DbError::InvalidData(format!("invalid SQLite community id: {e}")))?; + Ok(CommunityRecord { + id: CommunityId::from_uuid(id), + host: row.try_get("host")?, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use nostr::Keys; + + #[tokio::test] + async fn community_slice_survives_temporary_file_reopen() { + let path = std::env::temp_dir().join(format!("buzz-db-{}.sqlite", Uuid::new_v4())); + let path_string = path.to_string_lossy().into_owned(); + let owner = Keys::generate(); + let owner_bytes = owner.public_key().to_bytes(); + let owner_hex = owner.public_key().to_hex(); + let channel_id = Uuid::new_v4(); + + let pool = connect(&path_string).await.unwrap(); + let ensured = ensure_configured_community(&pool, "Local.Buzz") + .await + .unwrap(); + bootstrap_owner(&pool, ensured.id, &owner_hex) + .await + .unwrap(); + let (channel, created) = create_channel_with_id( + &pool, + ensured.id, + channel_id, + "general", + crate::channel::ChannelType::Stream, + crate::channel::ChannelVisibility::Private, + None, + &owner_bytes, + None, + ) + .await + .unwrap(); + assert!(created); + assert_eq!(channel.id, channel_id); + assert!(is_member(&pool, ensured.id, channel_id, &owner_bytes) + .await + .unwrap()); + pool.close().await; + + let reopened = connect(&path_string).await.unwrap(); + let found = lookup_community_by_host(&reopened, "local.buzz") + .await + .unwrap() + .unwrap(); + assert_eq!(found.id, ensured.id); + assert!(is_community_active(&reopened, ensured.id).await.unwrap()); + assert!(is_relay_member(&reopened, ensured.id, &owner_hex) + .await + .unwrap()); + assert_eq!( + get_accessible_channel_ids(&reopened, ensured.id, &owner_bytes) + .await + .unwrap(), + vec![channel_id] + ); + reopened.close().await; + std::fs::remove_file(path).unwrap(); + } + + #[tokio::test] + async fn sqlite_membership_preserves_public_contract_invariants() { + let pool = connect(":memory:").await.unwrap(); + let community = ensure_configured_community(&pool, "local.buzz") + .await + .unwrap(); + let owner = Keys::generate().public_key().to_bytes(); + let member = Keys::generate().public_key().to_bytes(); + let admin = Keys::generate().public_key().to_bytes(); + let other = Keys::generate().public_key().to_bytes(); + let channel_id = Uuid::new_v4(); + create_channel_with_id( + &pool, + community.id, + channel_id, + "private", + crate::channel::ChannelType::Stream, + crate::channel::ChannelVisibility::Private, + None, + &owner, + None, + ) + .await + .unwrap(); + + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &[0; 31], + crate::channel::MemberRole::Member, + Some(&owner), + ) + .await, + Err(crate::DbError::InvalidData(_)) + )); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Member, + None, + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Member, + Some(&owner), + ) + .await + .unwrap(); + add_member( + &pool, + community.id, + channel_id, + &admin, + crate::channel::MemberRole::Admin, + Some(&owner), + ) + .await + .unwrap(); + let readded = add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Member, + Some(&admin), + ) + .await + .unwrap(); + assert_eq!(readded.invited_by, Some(owner.to_vec())); + sqlx::query::( + "UPDATE channel_members SET removed_at = unixepoch() WHERE channel_id = ?1 AND pubkey = ?2", + ) + .bind(channel_id.to_string()) + .bind(&member[..]) + .execute(&pool) + .await + .unwrap(); + let reactivated = add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Member, + Some(&admin), + ) + .await + .unwrap(); + assert_eq!(reactivated.invited_by, Some(owner.to_vec())); + assert!(reactivated.removed_at.is_none()); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &other, + crate::channel::MemberRole::Admin, + Some(&member), + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Guest, + Some(&member), + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &owner, + crate::channel::MemberRole::Member, + Some(&owner), + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + } + + #[tokio::test] + async fn rebind_prefers_owned_then_oldest_community() { + let pool = connect(":memory:").await.unwrap(); + let first = ensure_configured_community(&pool, "127.0.0.1:4000") + .await + .unwrap(); + let second = ensure_configured_community(&pool, "127.0.0.1:4001") + .await + .unwrap(); + bootstrap_owner(&pool, second.id, "owner").await.unwrap(); + let rebound = rebind_single_node_community_host(&pool, "127.0.0.1:5000", "owner") + .await + .unwrap() + .unwrap(); + assert_eq!(rebound.id, second.id); + assert_eq!( + lookup_community_host(&pool, first.id) + .await + .unwrap() + .unwrap(), + "127.0.0.1:4000" + ); + assert_eq!(rebound.host, "127.0.0.1:5000"); + } +} diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 323b7b949f..27605568f2 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 858f0f5ace..99baa76a77 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/main.rs b/crates/buzz-relay/src/main.rs index 6c70f0b565..a6adbbf083 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1,4 +1,5 @@ use std::collections::{HashMap, HashSet}; +use std::net::{IpAddr, Ipv4Addr}; use std::sync::atomic::Ordering; use std::sync::Arc; @@ -11,16 +12,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::config::{Config, RelayProfile, 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 +155,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 +1199,116 @@ 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_loopback(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 = if let Some(owner) = config.relay_owner_pubkey.as_deref() { + match db.rebind_single_node_community_host(&host, owner).await? { + Some(record) => record.id, + None => db.ensure_configured_community(&host).await?.id, + } + } else { + db.ensure_configured_community(&host).await?.id + }; + if let Some(owner) = config.relay_owner_pubkey.as_deref() { + db.bootstrap_owner(community, owner).await?; + } + + 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 @@ -1234,6 +1341,24 @@ async fn run_periodic_until_cancelled( /// roughly the 5s grace plus the ack wait. const GRACEFUL_DRAIN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); +fn health_listener_ip(profile: RelayProfile) -> IpAddr { + match profile { + RelayProfile::SingleNode => IpAddr::V4(Ipv4Addr::LOCALHOST), + RelayProfile::Production => IpAddr::V4(Ipv4Addr::UNSPECIFIED), + } +} + +#[cfg(test)] +mod listener_tests { + use super::{health_listener_ip, RelayProfile}; + + #[test] + fn single_node_health_listener_is_loopback_only() { + assert!(health_listener_ip(RelayProfile::SingleNode).is_loopback()); + assert!(health_listener_ip(RelayProfile::Production).is_unspecified()); + } +} + async fn serve( router: axum::Router, health_router: axum::Router, @@ -1241,10 +1366,11 @@ async fn serve( ) -> anyhow::Result<()> { let config = &state.config; - let health_listener = tokio::net::TcpListener::bind(("0.0.0.0", config.health_port)) + let health_host = health_listener_ip(config.profile); + let health_listener = tokio::net::TcpListener::bind((health_host, config.health_port)) .await .map_err(|e| anyhow::anyhow!("Failed to bind health port {}: {e}", config.health_port))?; - info!(port = config.health_port, "Health probe listener started"); + info!(host = %health_host, port = config.health_port, "Health probe listener started"); tokio::spawn(async move { axum::serve(health_listener, health_router).await.ok(); }); diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 16e521a44e..21d1ff43c4 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -14,6 +14,7 @@ //! recorded by [`track_metrics`] middleware on the app router. Buzz-specific //! metrics are recorded inline at their call sites. +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::time::{Duration, Instant}; use axum::{ @@ -56,6 +57,17 @@ const GIT_PACK_BUCKETS: [f64; 9] = [0.0, 1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 1 /// Integer-count buckets for fan-out recipient histograms. const FANOUT_BUCKETS: [f64; 9] = [0.0, 1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 500.0, 1000.0]; +fn prometheus_listener(port: u16, loopback_only: bool) -> SocketAddr { + SocketAddr::new( + IpAddr::V4(if loopback_only { + Ipv4Addr::LOCALHOST + } else { + Ipv4Addr::UNSPECIFIED + }), + port, + ) +} + /// Install the global metrics recorder and spawn the Prometheus HTTP exporter. /// /// `build()` returns the recorder + exporter future and internally spawns @@ -64,8 +76,17 @@ const FANOUT_BUCKETS: [f64; 9] = [0.0, 1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 500.0, /// Must be called from within a Tokio runtime. /// Panics if a recorder is already installed or the port is in use. pub fn install(port: u16, gauge_idle_timeout_secs: u64) { + install_on(prometheus_listener(port, false), gauge_idle_timeout_secs); +} + +/// Install the metrics exporter on loopback for the single-node profile. +pub fn install_loopback(port: u16, gauge_idle_timeout_secs: u64) { + install_on(prometheus_listener(port, true), gauge_idle_timeout_secs); +} + +fn install_on(listener: SocketAddr, gauge_idle_timeout_secs: u64) { let (recorder, exporter) = PrometheusBuilder::new() - .with_http_listener(([0, 0, 0, 0], port)) + .with_http_listener(listener) // Remove gauge series that the relay intentionally stops emitting. .idle_timeout( MetricKindMask::GAUGE, @@ -205,3 +226,14 @@ pub async fn track_metrics(req: Request, next: Next) -> Response { response } + +#[cfg(test)] +mod tests { + use super::prometheus_listener; + + #[test] + fn single_node_metrics_listener_is_loopback_only() { + assert!(prometheus_listener(9102, true).ip().is_loopback()); + assert!(prometheus_listener(9102, false).ip().is_unspecified()); + } +} diff --git a/desktop/src-tauri/src/commands/agent_auth.rs b/desktop/src-tauri/src/commands/agent_auth.rs index bba0f33860..d90c9ad7f6 100644 --- a/desktop/src-tauri/src/commands/agent_auth.rs +++ b/desktop/src-tauri/src/commands/agent_auth.rs @@ -13,6 +13,51 @@ use serde::{Deserialize, Serialize}; use crate::managed_agents::{ default_agent_workdir, known_acp_runtime_exact, normalize_agent_args, resolve_command, }; +use crate::{ + app_state::AppState, + managed_agents::{load_managed_agents, save_managed_agents}, + util::now_iso, +}; + +pub(super) fn backfill_missing_agent_auth_tags( + app: &tauri::AppHandle, + state: &AppState, +) -> Result { + let owner_keys = state.signing_keys()?; + let compat_owner = nostr::Keys::parse(&owner_keys.secret_key().to_secret_hex()) + .map_err(|e| format!("failed to bridge owner keys: {e}"))?; + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let mut records = load_managed_agents(app)?; + let mut changed = 0; + for record in &mut records { + if record.auth_tag.is_some() + || record + .pubkey + .eq_ignore_ascii_case(&owner_keys.public_key().to_hex()) + { + continue; + } + let agent = nostr::PublicKey::from_hex(&record.pubkey) + .map_err(|e| format!("invalid managed agent pubkey {}: {e}", record.pubkey))?; + record.auth_tag = Some( + buzz_sdk_pkg::nip_oa::compute_auth_tag(&compat_owner, &agent, "").map_err(|e| { + format!( + "failed to compute NIP-OA auth tag for {}: {e}", + record.pubkey + ) + })?, + ); + record.updated_at = now_iso(); + changed += 1; + } + if changed > 0 { + save_managed_agents(app, &records)?; + } + Ok(changed) +} #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "camelCase")] diff --git a/desktop/src-tauri/src/commands/identity.rs b/desktop/src-tauri/src/commands/identity.rs index ec2357b85e..d6350a7ee8 100644 --- a/desktop/src-tauri/src/commands/identity.rs +++ b/desktop/src-tauri/src/commands/identity.rs @@ -51,8 +51,11 @@ pub fn get_identity(state: State<'_, AppState>) -> Result } #[tauri::command] -pub fn get_default_relay_url() -> String { - relay::relay_ws_url() +pub fn get_default_relay_url(state: State<'_, AppState>) -> Result { + // A hosted default remains just that: resolving it never spawns an unused + // local sidecar. The explicit local workspace sentinel starts its sidecar + // during `apply_workspace`. + Ok(relay::relay_ws_url_with_override(&state)) } #[tauri::command] diff --git a/desktop/src-tauri/src/commands/workspace.rs b/desktop/src-tauri/src/commands/workspace.rs index aa88bfe39a..26739bfab8 100644 --- a/desktop/src-tauri/src/commands/workspace.rs +++ b/desktop/src-tauri/src/commands/workspace.rs @@ -135,7 +135,9 @@ pub async fn apply_workspace( tokio::task::spawn_blocking(move || { let state = app.state::(); - // ── Validate before mutating ────────────────────────────────────────── + // Validate credentials before starting a local sidecar: an onboarding + // import may replace the in-memory identity, and the relay's durable + // nest/owner must match the identity that this workspace will use. let parsed_keys = match nsec.as_deref().map(str::trim).filter(|s| !s.is_empty()) { Some(nsec_trimmed) => { Some(Keys::parse(nsec_trimmed).map_err(|e| format!("invalid nsec: {e}"))?) @@ -143,6 +145,19 @@ pub async fn apply_workspace( None => None, }; + let is_local_workspace = crate::local_relay::is_local_relay_url(&relay_url); + if is_local_workspace { + // Only the UI-created local sentinel can launch the bundled relay. + // An arbitrary loopback URL remains a normal remote community. + let keys = parsed_keys.clone().unwrap_or(state.signing_keys()?); + let runtime = app.state::(); + crate::local_relay::ensure_started(&app, &runtime, &keys, None)?; + } else { + // A desktop-owned relay is private to the active local workspace; + // do not leave it serving after a switch to any remote workspace. + crate::local_relay::stop(&app.state::()); + } + // Decide the effective repos_dir from the candidate. A bad path does NOT // reject — it is treated as if no override were set: relay/keys still // apply, the bad value is not persisted, and a `repos-dir-error` surfaces @@ -165,8 +180,17 @@ pub async fn apply_workspace( // ── Apply all state changes (nothing below can fail) ────────────────── { + // The local sentinel persists across restarts while the sidecar + // endpoint is deliberately ephemeral. Resolve it only after + // `ensure_started` above, before any agent can be restored. + let effective_relay_url = if crate::local_relay::is_local_relay_url(&relay_url) { + crate::local_relay::relay_url(&app.state::()) + .ok_or("local relay did not start")? + } else { + relay_url + }; let mut override_guard = state.relay_url_override.lock().map_err(|e| e.to_string())?; - *override_guard = Some(relay_url); + *override_guard = Some(effective_relay_url); } // Reset the Rust-side admission gate when switching workspace/community, // matching `resetRateLimitGate()` on the TS side (useCommunityInit.ts:38). @@ -205,6 +229,12 @@ pub async fn apply_workspace( } try_regenerate_nest(&app); + if is_local_workspace { + let changed = super::agent_auth::backfill_missing_agent_auth_tags(&app, &state)?; + if changed > 0 { + eprintln!("buzz-desktop: backfilled NIP-OA auth tags for {changed} local agent(s)"); + } + } Ok::<(), String>(()) }) diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 7aa954ce8e..b3444601b7 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -14,6 +14,7 @@ mod initial_window; mod key_backup; mod link_preview_tags; mod linux_media; +mod local_relay; #[cfg(target_os = "macos")] mod macos_notifications; mod managed_agents; @@ -301,6 +302,7 @@ pub fn run() { }); }) .manage(build_app_state()) + .manage(local_relay::RuntimeState::new(None)) .manage(ClipboardState::new()) .manage(PendingCommunityDeepLinks::default()) .manage(BuilderlabSession::default()) @@ -968,9 +970,11 @@ pub fn run() { if is_restart_request(code) { restart_requested.store(true, Ordering::SeqCst); } + local_relay::stop(&app_handle.state::()); shut_down_app(app_handle, &run_shutdown_done); } RunEvent::Exit => { + local_relay::stop(&app_handle.state::()); shut_down_app(app_handle, &run_shutdown_done); app_handle.state::().release(); diff --git a/desktop/src-tauri/src/local_relay.rs b/desktop/src-tauri/src/local_relay.rs new file mode 100644 index 0000000000..7fd6497ace --- /dev/null +++ b/desktop/src-tauri/src/local_relay.rs @@ -0,0 +1,507 @@ +//! Lifecycle management for the embedded single-node relay. +//! +//! The relay remains the source of truth for local-mode behavior. Desktop only +//! chooses private loopback ports, supplies its identity and durable paths, and +//! supervises the bundled `buzz-relay` binary. + +use std::{ + io::Read, + net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener}, + path::{Path, PathBuf}, + process::{Child, Command, Stdio}, + sync::{Arc, Mutex}, + time::{Duration, Instant}, +}; + +use nostr::Keys; +use tauri::{AppHandle, Manager}; + +const STARTUP_TIMEOUT: Duration = Duration::from_secs(15); +const POLL_INTERVAL: Duration = Duration::from_millis(100); +const STDERR_TAIL_BYTES: usize = 8 * 1024; + +type StderrTail = Arc>>; + +pub(crate) struct LocalRelayRuntime { + child: Child, + url: String, + owner_pubkey: String, +} + +pub(crate) type RuntimeState = Mutex>; + +#[derive(Debug, PartialEq, Eq)] +struct LocalRelayConfig { + relay_addr: SocketAddr, + health_port: u16, + metrics_port: u16, + data_dir: PathBuf, + owner_pubkey: String, + relay_private_key: String, +} + +impl LocalRelayConfig { + fn url(&self) -> String { + format!("ws://{}", self.relay_addr) + } + + fn environment(&self) -> Vec<(&'static str, String)> { + let db_path = self.data_dir.join("relay.sqlite3"); + let media_dir = self.data_dir.join("media"); + vec![ + ("BUZZ_PROFILE", "single-node".to_string()), + ("BUZZ_BIND_ADDR", self.relay_addr.to_string()), + ("BUZZ_HEALTH_PORT", self.health_port.to_string()), + ("BUZZ_METRICS_PORT", self.metrics_port.to_string()), + ("RELAY_URL", self.url()), + ("BUZZ_LOCAL_DB", sqlite_url(&db_path)), + ("BUZZ_LOCAL_MEDIA_DIR", media_dir.display().to_string()), + ("BUZZ_REQUIRE_RELAY_MEMBERSHIP", "true".to_string()), + ("BUZZ_ALLOW_NIP_OA_AUTH", "true".to_string()), + ("RELAY_OWNER_PUBKEY", self.owner_pubkey.clone()), + ("BUZZ_RELAY_PRIVATE_KEY", self.relay_private_key.clone()), + ] + } +} + +impl LocalRelayRuntime { + pub(crate) fn url(&self) -> &str { + &self.url + } + + fn is_running(&mut self) -> Result { + self.child + .try_wait() + .map(|status| status.is_none()) + .map_err(|error| format!("inspect local relay: {error}")) + } + + fn stop(&mut self) { + // The relay handles SIGTERM with a WebSocket drain. It is our child, so + // waiting here prevents a restart from inheriting a stale listener. + #[cfg(unix)] + unsafe { + libc::kill(self.child.id() as i32, libc::SIGTERM); + } + #[cfg(windows)] + let _ = self.child.kill(); + + let deadline = Instant::now() + Duration::from_secs(8); + while Instant::now() < deadline { + if self.child.try_wait().ok().flatten().is_some() { + return; + } + std::thread::sleep(Duration::from_millis(50)); + } + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +/// Starts a relay for this desktop identity and waits for the public relay-info +/// endpoint. `Db::new_sqlite` in the relay applies its versioned SQLite +/// migrations before this endpoint can become available. +pub(crate) fn start( + app: &AppHandle, + keys: &Keys, + requested_port: Option, +) -> Result { + let owner_pubkey = keys.public_key().to_hex(); + let data_dir = local_data_dir(app, &owner_pubkey)?; + let relay_keys = load_or_create_relay_keys(&data_dir)?; + let media_dir = data_dir.join("media"); + std::fs::create_dir_all(&media_dir) + .map_err(|e| format!("create local relay media dir: {e}"))?; + let binary = relay_binary()?; + + // The relay owns its TCP listeners, so the desktop cannot safely retain a + // reservation across `spawn`. Retry a complete, distinct three-port set + // when an automatically-selected port loses that race; explicit user ports + // fail with the child diagnostic instead of silently moving the endpoint. + let attempts = if requested_port.is_some() { 1 } else { 3 }; + let mut last_error = None; + for _ in 0..attempts { + let (relay_port, health_port, metrics_port) = + allocate_distinct_loopback_ports(requested_port)?; + let config = LocalRelayConfig { + relay_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), relay_port), + health_port, + metrics_port, + data_dir: data_dir.clone(), + owner_pubkey: owner_pubkey.clone(), + relay_private_key: relay_keys.secret_key().to_secret_hex(), + }; + + match spawn_and_wait(&binary, config) { + Ok(runtime) => return Ok(runtime), + Err(error) if requested_port.is_none() && is_bind_conflict(&error) => { + last_error = Some(error); + } + Err(error) => return Err(error), + } + } + Err(last_error.expect("at least one local relay startup attempt")) +} + +pub(crate) fn ensure_started( + app: &AppHandle, + runtime: &RuntimeState, + keys: &Keys, + requested_port: Option, +) -> Result { + let mut runtime = runtime.lock().map_err(|error| error.to_string())?; + if let Some(running) = runtime.as_mut() { + let same_port = requested_port.is_none_or(|port| running.url == local_relay_url(port)); + let same_owner = running.owner_pubkey == keys.public_key().to_hex(); + let is_running = running.is_running()?; + if same_port && same_owner && is_running { + return Ok(running.url().to_string()); + } + if !same_owner { + // An onboarding identity import changed the owner. Its local relay + // must use a separate identity-scoped nest, not retain the prior + // identity's private sidecar. `try_wait` above already reaped an + // exited child, so never signal its potentially recycled PID. + if let Some(mut previous) = runtime.take() { + if is_running { + previous.stop(); + } + } + } else if !is_running { + // `try_wait` reaped the crashed child. Drop its stale handle so the + // replacement below receives a fresh loopback port and child. + runtime.take(); + } else { + return Err("a different local relay port is already running".to_string()); + } + } + let started = start(app, keys, requested_port)?; + let url = started.url().to_string(); + *runtime = Some(started); + Ok(url) +} + +pub(crate) fn stop(runtime: &RuntimeState) { + if let Ok(mut runtime) = runtime.lock() { + if let Some(mut runtime) = runtime.take() { + runtime.stop(); + } + } +} + +/// The persisted sentinel for the desktop-owned local workspace. It is never a +/// network endpoint: `apply_workspace` resolves it to the current supervised +/// loopback URL before any client or managed agent is restored. +pub(crate) const LOCAL_RELAY_SENTINEL: &str = "buzz-local://on-this-device"; + +pub(crate) fn is_local_relay_url(url: &str) -> bool { + url == LOCAL_RELAY_SENTINEL +} + +fn local_relay_url(port: u16) -> String { + format!("ws://127.0.0.1:{port}") +} + +pub(crate) fn relay_url(runtime: &RuntimeState) -> Option { + runtime + .lock() + .ok() + .and_then(|runtime| runtime.as_ref().map(|runtime| runtime.url().to_string())) +} + +fn local_data_dir(app: &AppHandle, owner_pubkey: &str) -> Result { + // Relay state is both durable and identity-scoped. A different desktop + // identity must never inherit this identity's SQLite membership or media. + app.path() + .app_data_dir() + .map(|path| path.join("local-relay").join(owner_pubkey)) + .map_err(|e| format!("resolve app data directory for local relay: {e}")) +} + +fn sqlite_url(path: &Path) -> String { + // sqlx accepts an absolute sqlite URL with three slashes. Path display is + // intentional: app-data paths come from the OS, not user relay input. + format!("sqlite://{}", path.display()) +} + +/// Load the relay's service identity from its identity-scoped nest, or create +/// it once with the same atomic, owner-only file semantics used for private +/// desktop identities. This key signs relay-originated metadata; it is never +/// the human/owner key supplied by `apply_workspace`. +fn load_or_create_relay_keys(data_dir: &Path) -> Result { + let path = data_dir.join("relay-service.key"); + if path.exists() { + let value = std::fs::read_to_string(&path) + .map_err(|e| format!("read local relay service key: {e}"))?; + return Keys::parse(value.trim()) + .map_err(|e| format!("parse local relay service key: {e}")); + } + + std::fs::create_dir_all(data_dir).map_err(|e| format!("create local relay data dir: {e}"))?; + let keys = Keys::generate(); + crate::app_state::save_key_file(&path, &keys) + .map_err(|e| format!("persist local relay service key: {e}"))?; + Ok(keys) +} + +fn is_bind_conflict(error: &str) -> bool { + error.contains("Address already in use") || error.contains("address already in use") +} + +fn spawn_and_wait(binary: &Path, config: LocalRelayConfig) -> Result { + let url = config.url(); + let info_url = format!("http://{}/info", config.relay_addr); + let stderr = Arc::new(Mutex::new(Vec::new())); + let mut command = Command::new(binary); + command + .envs(config.environment()) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()); + let mut child = command + .spawn() + .map_err(|e| format!("start bundled buzz-relay at {}: {e}", binary.display()))?; + let stderr_reader = capture_stderr(child.stderr.take(), stderr.clone()); + + let client = reqwest::blocking::Client::new(); + let deadline = Instant::now() + STARTUP_TIMEOUT; + while Instant::now() < deadline { + if let Some(status) = child + .try_wait() + .map_err(|e| format!("inspect local relay: {e}"))? + { + let _ = stderr_reader.join(); + return Err(format!( + "bundled buzz-relay exited before becoming ready ({status}){}", + stderr_context(&stderr) + )); + } + if client + .get(&info_url) + .header("Accept", "application/nostr+json") + .send() + .is_ok_and(|response| response.status().is_success()) + { + return Ok(LocalRelayRuntime { + child, + url, + owner_pubkey: config.owner_pubkey, + }); + } + std::thread::sleep(POLL_INTERVAL); + } + + let _ = child.kill(); + let _ = child.wait(); + let _ = stderr_reader.join(); + Err(format!( + "bundled buzz-relay did not become ready within 15 seconds{}", + stderr_context(&stderr) + )) +} + +fn capture_stderr( + stderr: Option, + tail: StderrTail, +) -> std::thread::JoinHandle<()> { + std::thread::spawn(move || { + if let Some(mut stderr) = stderr { + let mut buffer = [0; 1024]; + while let Ok(read) = stderr.read(&mut buffer) { + if read == 0 { + break; + } + let mut tail = tail.lock().expect("stderr tail mutex poisoned"); + tail.extend_from_slice(&buffer[..read]); + if tail.len() > STDERR_TAIL_BYTES { + let excess = tail.len() - STDERR_TAIL_BYTES; + tail.drain(..excess); + } + } + } + }) +} + +fn stderr_context(stderr: &StderrTail) -> String { + let tail = stderr.lock().expect("stderr tail mutex poisoned"); + let text = String::from_utf8_lossy(&tail).trim().to_string(); + if text.is_empty() { + String::new() + } else { + format!("; stderr: {text}") + } +} + +fn allocate_distinct_loopback_ports( + requested_port: Option, +) -> Result<(u16, u16, u16), String> { + let relay_port = requested_port.unwrap_or(free_loopback_port()?); + let mut ports = vec![relay_port]; + while ports.len() < 3 { + let port = free_loopback_port()?; + if !ports.contains(&port) { + ports.push(port); + } + } + Ok((ports[0], ports[1], ports[2])) +} + +fn free_loopback_port() -> Result { + let listener = + TcpListener::bind("127.0.0.1:0").map_err(|e| format!("pick local relay port: {e}"))?; + listener + .local_addr() + .map(|address| address.port()) + .map_err(|e| format!("read local relay port: {e}")) +} + +fn relay_binary() -> Result { + let exe = std::env::current_exe().map_err(|e| format!("resolve desktop executable: {e}"))?; + let parent = exe + .parent() + .ok_or("desktop executable has no parent directory")?; + let name = if cfg!(windows) { + "buzz-relay.exe" + } else { + "buzz-relay" + }; + let binary = parent.join(name); + if binary.is_file() { + Ok(binary) + } else { + Err(format!( + "bundled buzz-relay is missing at {}", + binary.display() + )) + } +} + +#[cfg(test)] +mod tests { + use super::{ + allocate_distinct_loopback_ports, is_local_relay_url, load_or_create_relay_keys, + local_relay_url, sqlite_url, stderr_context, LocalRelayConfig, StderrTail, + LOCAL_RELAY_SENTINEL, + }; + use std::{ + collections::HashMap, + net::{IpAddr, Ipv4Addr, SocketAddr}, + path::{Path, PathBuf}, + process::{Command, Stdio}, + sync::{Arc, Mutex}, + }; + + #[test] + fn selected_ports_are_distinct_in_one_startup_attempt() { + let (relay, health, metrics) = allocate_distinct_loopback_ports(None).expect("ports"); + assert_ne!(relay, health); + assert_ne!(relay, metrics); + assert_ne!(health, metrics); + } + + #[test] + fn startup_error_includes_bounded_stderr_tail() { + let tail: StderrTail = Arc::new(Mutex::new(b"bind: address already in use\n".to_vec())); + assert_eq!( + stderr_context(&tail), + "; stderr: bind: address already in use" + ); + } + + #[test] + fn sqlite_url_is_absolute() { + assert_eq!( + sqlite_url(Path::new("/tmp/buzz.sqlite3")), + "sqlite:///tmp/buzz.sqlite3" + ); + } + + #[test] + fn local_relay_sentinel_is_the_only_managed_workspace_address() { + assert_eq!(local_relay_url(4317), "ws://127.0.0.1:4317"); + assert!(is_local_relay_url(LOCAL_RELAY_SENTINEL)); + assert!(!is_local_relay_url("ws://127.0.0.1:4317")); + assert!(!is_local_relay_url("ws://localhost:4317")); + assert!(!is_local_relay_url("wss://127.0.0.1:4317")); + } + + #[test] + fn exited_relay_is_not_running() { + let child = Command::new(std::env::current_exe().expect("test executable")) + .arg("--help") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("start test child"); + let mut runtime = super::LocalRelayRuntime { + child, + url: "ws://127.0.0.1:4317".to_string(), + owner_pubkey: "owner".to_string(), + }; + + runtime.child.wait().expect("wait for test child"); + assert!(!runtime.is_running().expect("inspect exited child")); + } + + #[test] + fn relay_service_key_is_durable_and_distinct_from_owner_identity() { + let directory = tempfile::tempdir().expect("temp data dir"); + let owner = nostr::Keys::generate(); + let first = load_or_create_relay_keys(directory.path()).expect("create service key"); + let second = load_or_create_relay_keys(directory.path()).expect("reload service key"); + + assert_ne!(first.public_key(), owner.public_key()); + assert_eq!(first.public_key(), second.public_key()); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(directory.path().join("relay-service.key")) + .expect("service key metadata") + .permissions() + .mode() + & 0o777; + assert_eq!(mode, 0o600); + } + } + + #[test] + fn local_relay_environment_uses_only_the_single_node_contract() { + let config = LocalRelayConfig { + relay_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4317), + health_port: 4318, + metrics_port: 4319, + data_dir: PathBuf::from("/tmp/buzz/local-relay"), + owner_pubkey: "owner".to_string(), + relay_private_key: "private".to_string(), + }; + let env: HashMap<_, _> = config.environment().into_iter().collect(); + + assert_eq!(env.get("BUZZ_PROFILE"), Some(&"single-node".to_string())); + assert_eq!( + env.get("BUZZ_BIND_ADDR"), + Some(&"127.0.0.1:4317".to_string()) + ); + assert_eq!( + env.get("RELAY_URL"), + Some(&"ws://127.0.0.1:4317".to_string()) + ); + assert_eq!( + env.get("BUZZ_LOCAL_DB"), + Some(&"sqlite:///tmp/buzz/local-relay/relay.sqlite3".to_string()) + ); + assert_eq!( + env.get("BUZZ_LOCAL_MEDIA_DIR"), + Some(&"/tmp/buzz/local-relay/media".to_string()) + ); + assert_eq!( + env.get("BUZZ_REQUIRE_RELAY_MEMBERSHIP"), + Some(&"true".to_string()) + ); + assert_eq!(env.get("BUZZ_ALLOW_NIP_OA_AUTH"), Some(&"true".to_string())); + assert_eq!(env.get("RELAY_OWNER_PUBKEY"), Some(&"owner".to_string())); + assert_eq!( + env.get("BUZZ_RELAY_PRIVATE_KEY"), + Some(&"private".to_string()) + ); + } +} diff --git a/desktop/src-tauri/tauri.conf.json b/desktop/src-tauri/tauri.conf.json index 3f332ddbf1..75d7c7c594 100644 --- a/desktop/src-tauri/tauri.conf.json +++ b/desktop/src-tauri/tauri.conf.json @@ -58,7 +58,8 @@ "binaries/buzz-backend-kubernetes", "binaries/buzz-dev-mcp", "binaries/git-credential-nostr", - "binaries/buzz" + "binaries/buzz", + "binaries/buzz-relay" ], "icon": [ "icons/32x32.png", diff --git a/desktop/src/app/App.tsx b/desktop/src/app/App.tsx index 0f311f3a65..528e397f36 100644 --- a/desktop/src/app/App.tsx +++ b/desktop/src/app/App.tsx @@ -42,7 +42,10 @@ import { PendingInviteGate } from "@/features/onboarding/ui/PendingInviteGate"; import { KeyringLockedScreen } from "@/features/onboarding/ui/KeyringLockedScreen"; import { RelaunchRequiredScreen } from "@/features/onboarding/ui/RelaunchRequiredScreen"; import { ResetFailedScreen } from "@/features/onboarding/ui/ResetFailedScreen"; -import { loadCommunityDiscoveryAfterLeave } from "@/features/communities/communityStorage"; +import { + isLocalCommunityRelayUrl, + loadCommunityDiscoveryAfterLeave, +} from "@/features/communities/communityStorage"; import { useCommunityInit } from "@/features/communities/useCommunityInit"; import { useNestNotifications } from "@/features/communities/useNestNotifications"; import { useCommunities } from "@/features/communities/useCommunities"; @@ -387,13 +390,17 @@ function CommunityApp({ return; } const previousCommunityId = activeCommunity?.id; + const isLocal = isLocalCommunityRelayUrl(transaction.relayUrl); const relayAlreadyExists = communities.some( - (community) => community.relayUrl === transaction.relayUrl, + (community) => + community.relayUrl === transaction.relayUrl || + (isLocal && community.local === true), ); const id = addCommunity({ id: crypto.randomUUID(), name: transaction.communityName, relayUrl: transaction.relayUrl, + local: isLocal, token: transaction.token, reposDir: transaction.reposDir, pubkey: currentPubkey ?? undefined, diff --git a/desktop/src/features/communities/communityStorage.test.mjs b/desktop/src/features/communities/communityStorage.test.mjs index fca1909137..94e21cd84c 100644 --- a/desktop/src/features/communities/communityStorage.test.mjs +++ b/desktop/src/features/communities/communityStorage.test.mjs @@ -3,9 +3,13 @@ import test from "node:test"; import { clearCommunityStorage, + deriveCommunityName, initFirstCommunity, + isLocalCommunityRelayUrl, loadCommunities, loadCommunityDiscoveryAfterLeave, + LOCAL_COMMUNITY_NAME, + LOCAL_COMMUNITY_RELAY_URL, markCommunityDiscoveryAfterLeave, migrateLegacyCommunityStorage, saveCommunities, @@ -53,6 +57,16 @@ test("migrateLegacyCommunityStorage does not overwrite new community state", () assert.equal(storage.getItem("buzz-active-community-id"), "new"); }); +test("only the local sentinel identifies the desktop-managed workspace", () => { + assert.equal(isLocalCommunityRelayUrl(LOCAL_COMMUNITY_RELAY_URL), true); + assert.equal( + deriveCommunityName(LOCAL_COMMUNITY_RELAY_URL), + LOCAL_COMMUNITY_NAME, + ); + assert.equal(isLocalCommunityRelayUrl("ws://127.0.0.1:4317"), false); + assert.equal(isLocalCommunityRelayUrl("ws://localhost:4317"), false); +}); + test("signed-build relay defaults auto-connect during first-run onboarding", () => { assert.equal( shouldAutoConnectDefaultRelay("wss://buzz.block.builderlab.xyz"), diff --git a/desktop/src/features/communities/communityStorage.ts b/desktop/src/features/communities/communityStorage.ts index 6f99ce8d33..1cf2033be9 100644 --- a/desktop/src/features/communities/communityStorage.ts +++ b/desktop/src/features/communities/communityStorage.ts @@ -10,6 +10,13 @@ const LEGACY_ACTIVE_WORKSPACE_KEY = "buzz-active-workspace-id"; const COMMUNITY_DISCOVERY_AFTER_LEAVE_KEY = "buzz-community-discovery-after-leave"; +export const LOCAL_COMMUNITY_RELAY_URL = "buzz-local://on-this-device"; +export const LOCAL_COMMUNITY_NAME = "On this device"; + +export function isLocalCommunityRelayUrl(relayUrl: string): boolean { + return relayUrl === LOCAL_COMMUNITY_RELAY_URL; +} + /** * Expand a leading `~` to the user's home directory. The backend rejects * `~`-prefixed paths (`std::fs` does not expand the shell tilde), so the UI @@ -186,13 +193,16 @@ export function shouldAutoConnectDefaultRelay(relayUrl: string): boolean { } export function deriveCommunityName(relayUrl: string): string { + if (isLocalCommunityRelayUrl(relayUrl)) { + return LOCAL_COMMUNITY_NAME; + } try { const url = new URL( relayUrl.replace("ws://", "http://").replace("wss://", "https://"), ); const host = url.hostname; if (isLocalRelayHost(host)) { - return "Local Dev"; + return "On this device"; } const parts = host.split("."); // Detect staging environments (e.g. buzz-oss.stage.blox.sqprod.co) diff --git a/desktop/src/features/communities/types.ts b/desktop/src/features/communities/types.ts index 7087ca7ce3..620b84056d 100644 --- a/desktop/src/features/communities/types.ts +++ b/desktop/src/features/communities/types.ts @@ -17,6 +17,8 @@ export type Community = { * `REPOS` directory inside the nest. */ reposDir?: string; + /** True only for the relay supervised by this desktop instance. */ + local?: boolean; /** * @deprecated Never read. Kept on the type so old localStorage entries * deserialise without errors. New entries never set this field, and diff --git a/desktop/src/features/communities/ui/AddCommunityDialog.tsx b/desktop/src/features/communities/ui/AddCommunityDialog.tsx index a83456652c..b636aaf31c 100644 --- a/desktop/src/features/communities/ui/AddCommunityDialog.tsx +++ b/desktop/src/features/communities/ui/AddCommunityDialog.tsx @@ -2,9 +2,16 @@ import * as React from "react"; import { ArrowLeft, ChevronRight, Link2, Plus } from "lucide-react"; import type { AddCommunityPrefillRequest } from "@/features/communities/addCommunityPrefill"; +import { + LOCAL_COMMUNITY_NAME, + LOCAL_COMMUNITY_RELAY_URL, +} from "@/features/communities/communityStorage"; +import { useCommunities } from "@/features/communities/useCommunities"; +import { CommunityCreationChoice } from "@/features/communities/ui/CommunityCreationChoice"; import { HostedCommunityCreateFlow } from "@/features/communities/ui/HostedCommunityCreateFlow"; import { useCommunityOnboarding } from "@/features/onboarding/communityOnboarding"; import { InviteRedeemForm } from "@/features/onboarding/ui/InviteRedeemForm"; +import { useFeatureEnabled } from "@/shared/features"; import { Dialog, DialogContent, @@ -22,7 +29,7 @@ type AddCommunityDialogProps = { onOpenChange: (open: boolean) => void; }; -type AddCommunityMode = "choose" | "create" | "join"; +type AddCommunityMode = "choose" | "create-choose" | "create-hosted" | "join"; const OPTION_CLASS = "flex w-full items-center gap-3 rounded-xl border border-border/70 bg-muted/30 px-4 py-4 text-left transition-colors duration-150 ease-out hover:bg-muted/60 focus-visible:outline-hidden focus-visible:ring-1 focus-visible:ring-ring"; @@ -33,6 +40,9 @@ export function AddCommunityDialog({ onOpenChange, }: AddCommunityDialogProps) { const communityOnboarding = useCommunityOnboarding(); + const localCommunitiesEnabled = useFeatureEnabled("localCommunities"); + const { communities } = useCommunities(); + const hasLocalCommunity = communities.some((community) => community.local); const [mode, setMode] = React.useState("choose"); const [joinError, setJoinError] = React.useState(null); const appliedPrefillId = React.useRef(null); @@ -78,19 +88,38 @@ export function AddCommunityDialog({ [communityOnboarding, handleClose, prefill?.name], ); + const startLocalCommunity = React.useCallback(() => { + const started = communityOnboarding.start({ + source: "add-community", + communityName: LOCAL_COMMUNITY_NAME, + relayUrl: LOCAL_COMMUNITY_RELAY_URL, + }); + if (!started) { + setJoinError( + "Finish connecting the community already in progress, then try again.", + ); + return; + } + handleClose(); + }, [communityOnboarding, handleClose]); + const title = - mode === "create" + mode === "create-choose" ? "Create a new community" - : mode === "join" - ? "Join an existing community" - : "Add community"; + : mode === "create-hosted" + ? "Host a community online" + : mode === "join" + ? "Join an existing community" + : "Add community"; const description = - mode === "create" - ? "Opens Builderlab in your browser." - : mode === "join" - ? "Use the community URL or invite link you received." - : "Create a new community or join one you already have."; + mode === "create-choose" + ? "Choose where your community will live." + : mode === "create-hosted" + ? "Opens Builderlab in your browser." + : mode === "join" + ? "Use the community URL or invite link you received." + : "Create a new community or join one you already have."; return ( {title} {description} @@ -135,7 +164,7 @@ export function AddCommunityDialog({ + + {showLocalOption ? ( + + + + ) : null} + + ); + } + + return ( +
+ + {showLocalOption ? ( + + ) : null} +
+ ); +} diff --git a/desktop/src/features/communities/ui/WelcomeSetup.tsx b/desktop/src/features/communities/ui/WelcomeSetup.tsx index 530933acc2..3b00fa823f 100644 --- a/desktop/src/features/communities/ui/WelcomeSetup.tsx +++ b/desktop/src/features/communities/ui/WelcomeSetup.tsx @@ -1,6 +1,11 @@ import * as React from "react"; import { Check, Copy } from "lucide-react"; +import { + LOCAL_COMMUNITY_NAME, + LOCAL_COMMUNITY_RELAY_URL, +} from "@/features/communities/communityStorage"; +import { CommunityCreationChoice } from "@/features/communities/ui/CommunityCreationChoice"; import { HostedCommunityOnboarding } from "@/features/communities/ui/HostedCommunityOnboarding"; import { useCommunityOnboarding } from "@/features/onboarding/communityOnboarding"; import { InviteRedeemForm } from "@/features/onboarding/ui/InviteRedeemForm"; @@ -14,6 +19,7 @@ import { OnboardingSlideTransition, } from "@/features/onboarding/ui/OnboardingSlideTransition"; import { useIdentityQuery } from "@/shared/api/hooks"; +import { useFeatureEnabled } from "@/shared/features"; import { writeTextToClipboard } from "@/shared/lib/clipboard"; import { pubkeyToNpub } from "@/shared/lib/nostrUtils"; import { useSystemColorScheme } from "@/shared/theme/useSystemColorScheme"; @@ -21,7 +27,13 @@ import { Button } from "@/shared/ui/button"; import { Card } from "@/shared/ui/card"; import { StartupWindowDragRegion } from "@/shared/ui/StartupWindowDragRegion"; -type WelcomeSetupPage = "welcome" | "existing" | "join" | "member" | "owned"; +type WelcomeSetupPage = + | "welcome" + | "create" + | "existing" + | "join" + | "member" + | "owned"; type WelcomeTransitionMode = "initial" | OnboardingTransitionDirection; type WelcomeSetupProps = { @@ -47,6 +59,7 @@ export function WelcomeSetup({ const [isHostedSignInOpen, setIsHostedSignInOpen] = React.useState(false); const [copiedNpub, setCopiedNpub] = React.useState(false); const communityOnboarding = useCommunityOnboarding(); + const localCommunitiesEnabled = useFeatureEnabled("localCommunities"); const identityQuery = useIdentityQuery(); const systemColorScheme = useSystemColorScheme(); const npub = identityQuery.data?.pubkey @@ -92,6 +105,15 @@ export function WelcomeSetup({ [communityOnboarding, page], ); + const startLocalCommunity = React.useCallback(() => { + communityOnboarding.start({ + source: "first-community", + firstCommunityPage: "join", + communityName: LOCAL_COMMUNITY_NAME, + relayUrl: LOCAL_COMMUNITY_RELAY_URL, + }); + }, [communityOnboarding]); + const transitionDirection = transitionMode === "backward" ? "backward" : "forward"; const welcomeEffect = @@ -144,7 +166,7 @@ export function WelcomeSetup({ > + + ) : page === "existing" ? ( { test("capture: create a new community", async ({ page }) => { await page.getByTestId("add-community-create").click(); + await page.getByTestId("community-create-hosted").click(); const dialog = page.getByTestId("add-community-dialog"); await page.getByLabel("Community address").waitFor(); await waitForAnimations(page); diff --git a/desktop/tests/e2e/onboarding.spec.ts b/desktop/tests/e2e/onboarding.spec.ts index 403edbcda1..d652675c02 100644 --- a/desktop/tests/e2e/onboarding.spec.ts +++ b/desktop/tests/e2e/onboarding.spec.ts @@ -3,6 +3,7 @@ import { expect, test, type Page } from "@playwright/test"; import { nsecEncode } from "nostr-tools/nip19"; import { installMockBridge, TEST_IDENTITIES } from "../helpers/bridge"; +import { FEATURE_OVERRIDES_STORAGE_KEY } from "../helpers/features"; import { installFakeCamera } from "../helpers/fakeCamera"; import { E2E_IDENTITY_OVERRIDE_STORAGE_KEY, @@ -80,6 +81,18 @@ const FIRST_RUN_ALICE = { username: "", }; +async function setLocalCommunitiesFeature(page: Page, enabled: boolean) { + await page.addInitScript( + ({ key, enabled }) => { + window.localStorage.setItem( + key, + JSON.stringify({ localCommunities: enabled }), + ); + }, + { key: FEATURE_OVERRIDES_STORAGE_KEY, enabled }, + ); +} + async function seedOnboardingCompletion(page: Page, pubkey: string) { await page.addInitScript( ({ storageKey }) => { @@ -1064,6 +1077,53 @@ test("non-local default auto-connects when the release flag is enabled", async ( }); }); +test("first-community hides the local choice by default", async ({ page }) => { + await seedActiveIdentity(page, BLANK_TYLER_IDENTITY); + await page.addInitScript((pubkey) => { + window.localStorage.setItem( + `buzz-machine-onboarding-complete.v2:${pubkey}`, + "true", + ); + }, BLANK_TYLER_IDENTITY.pubkey); + await installMockBridge(page, undefined, { + relayWsUrl: "ws://localhost:3000", + seedPreviewFeatures: false, + skipOnboardingSeed: true, + skipCommunitySeed: true, + }); + await setLocalCommunitiesFeature(page, false); + await page.goto("/"); + + await page.getByTestId("community-choice-create").click(); + await expect(page.getByTestId("community-create-hosted")).toBeVisible(); + await expect(page.getByTestId("community-choice-local")).toHaveCount(0); +}); + +test("first-community shows the local choice when its preview is enabled", async ({ + page, +}) => { + await seedActiveIdentity(page, BLANK_TYLER_IDENTITY); + await setLocalCommunitiesFeature(page, true); + await page.addInitScript((pubkey) => { + window.localStorage.setItem( + `buzz-machine-onboarding-complete.v2:${pubkey}`, + "true", + ); + }, BLANK_TYLER_IDENTITY.pubkey); + await installMockBridge(page, undefined, { + relayWsUrl: "ws://localhost:3000", + seedPreviewFeatures: false, + skipOnboardingSeed: true, + skipCommunitySeed: true, + }); + await page.goto("/"); + + await page.getByTestId("community-choice-create").click(); + await expect(page.getByTestId("community-choice-local")).toContainText( + "Keep it on this device", + ); +}); + test("first-community choices route join, create, owner, and member intents", async ({ page, }) => { @@ -1087,6 +1147,15 @@ test("first-community choices route join, create, owner, and member intents", as await expect( page.getByRole("button", { name: /Create a community/ }), ).toBeVisible(); + await expect(page.getByTestId("community-choice-local")).toHaveCount(0); + await page.getByTestId("community-choice-create").click(); + await expect(page.getByTestId("community-create-hosted")).toContainText( + "Host it online", + ); + await expect(page.getByTestId("community-choice-local")).toContainText( + "Keep it on this device", + ); + await page.getByTestId("community-create-back").click(); const existing = page.getByRole("button", { name: /I already have a community/, }); @@ -1169,6 +1238,7 @@ test("first-community owner can connect an existing hosted community", async ({ await page.goto("/"); await page.getByTestId("community-choice-create").click(); + await page.getByTestId("community-create-hosted").click(); await expect(page.getByText("North Star")).toBeVisible(); await page.getByRole("button", { name: "Connect", exact: true }).click(); await expect( @@ -1227,6 +1297,7 @@ test("first-community owner can create and connect a hosted community", async ({ await page.goto("/"); await page.getByTestId("community-choice-create").click(); + await page.getByTestId("community-create-hosted").click(); await page.getByRole("button", { name: "Sign in to continue" }).click(); await expect( page.getByRole("heading", { name: "Finish connecting Buzz" }), @@ -1303,6 +1374,7 @@ test("hosted community address line stays within the card for a long name", asyn await page.goto("/"); await page.getByTestId("community-choice-create").click(); + await page.getByTestId("community-create-hosted").click(); await page.getByRole("button", { name: "Sign in to continue" }).click(); await expect( page.getByRole("heading", { name: "Finish connecting Buzz" }), @@ -1372,6 +1444,7 @@ test("first-community reports a created community without a relay address", asyn await page.goto("/"); await page.getByTestId("community-choice-create").click(); + await page.getByTestId("community-create-hosted").click(); await page.getByRole("textbox", { name: "Community name" }).fill("bee-lab"); await expect(page.getByText("That address is available.")).toBeVisible(); await page.getByRole("button", { name: "Next" }).click(); @@ -1403,6 +1476,7 @@ test("first-community X cancels a pending sign-in", async ({ page }) => { await page.goto("/"); await page.getByTestId("community-choice-create").click(); + await page.getByTestId("community-create-hosted").click(); await page.getByRole("button", { name: "Sign in to continue" }).click(); await expect(page.getByText("Waiting for your browser…")).toBeVisible(); await expect( @@ -1410,8 +1484,9 @@ test("first-community X cancels a pending sign-in", async ({ page }) => { ).toHaveCount(0); await page.getByRole("button", { name: "Close" }).click(); await expect( - page.getByRole("button", { name: /Create a community/ }), + page.getByRole("heading", { name: "Create a community" }), ).toBeVisible(); + await expect(page.getByTestId("community-create-hosted")).toBeVisible(); await expect .poll(() => page.evaluate(() => window.__BUZZ_E2E_COMMANDS__ ?? [])) .toEqual(expect.arrayContaining(["cancel_builderlab_login"])); @@ -1445,6 +1520,7 @@ test("first-community owner can replace a mismatched account identity", async ({ await page.goto("/"); await page.getByTestId("community-choice-create").click(); + await page.getByTestId("community-create-hosted").click(); await expect( page.getByRole("heading", { name: "This account uses a different Buzz identity", @@ -1495,6 +1571,7 @@ test("first-community explains when the local identity belongs to another accoun await page.goto("/"); await page.getByTestId("community-choice-create").click(); + await page.getByTestId("community-create-hosted").click(); await page .getByRole("button", { name: "Use this device's identity" }) .click(); @@ -1536,8 +1613,10 @@ test("back clears Builderlab auth before returning to first-community choices", await page.goto("/"); await page.getByTestId("community-choice-create").click(); + await page.getByTestId("community-create-hosted").click(); await page.getByRole("button", { name: "Back" }).click(); await page.getByTestId("community-choice-create").click(); + await page.getByTestId("community-create-hosted").click(); await expect(page.getByRole("button", { name: "Continue" })).toBeVisible(); }); diff --git a/desktop/tests/e2e/sidebar.spec.ts b/desktop/tests/e2e/sidebar.spec.ts index 9dce5e7598..b11ef5ed59 100644 --- a/desktop/tests/e2e/sidebar.spec.ts +++ b/desktop/tests/e2e/sidebar.spec.ts @@ -1,6 +1,7 @@ import { expect, test, type Page } from "@playwright/test"; import { installMockBridge } from "../helpers/bridge"; +import { FEATURE_OVERRIDES_STORAGE_KEY } from "../helpers/features"; import { openSettings } from "../helpers/settings"; const SIDEBAR_WIDTH_STORAGE_KEY = "buzz-sidebar-width"; @@ -39,6 +40,18 @@ async function openAddCommunityDialog(page: Page) { await page.getByRole("menuitem", { name: "Add a community" }).click(); } +async function setLocalCommunitiesFeature(page: Page, enabled: boolean) { + await page.addInitScript( + ({ key, enabled }) => { + window.localStorage.setItem( + key, + JSON.stringify({ localCommunities: enabled }), + ); + }, + { key: FEATURE_OVERRIDES_STORAGE_KEY, enabled }, + ); +} + // Regression guard for the "Leave channel" lockup: with two bundled copies of // @radix-ui/react-dismissable-layer, opening a modal AlertDialog from a modal // Radix ContextMenu left `pointer-events: none` stuck on after the @@ -74,6 +87,33 @@ async function dragSidebarRail(page: Page, deltaX: number) { await page.mouse.up(); } +test("add community hides the local choice by default", async ({ page }) => { + await installMockBridge(page, {}, { seedPreviewFeatures: false }); + await setLocalCommunitiesFeature(page, false); + await page.goto("/"); + + await openAddCommunityDialog(page); + await page.getByTestId("add-community-create").click(); + + await expect(page.getByTestId("community-create-hosted")).toBeVisible(); + await expect(page.getByTestId("community-choice-local")).toHaveCount(0); +}); + +test("add community shows the local choice when its preview is enabled", async ({ + page, +}) => { + await setLocalCommunitiesFeature(page, true); + await installMockBridge(page, {}, { seedPreviewFeatures: false }); + await page.goto("/"); + + await openAddCommunityDialog(page); + await page.getByTestId("add-community-create").click(); + + await expect(page.getByTestId("community-choice-local")).toContainText( + "Keep it on this device", + ); +}); + test("add community starts with create and join choices", async ({ page }) => { await installMockBridge(page, {}); await page.goto("/"); @@ -103,11 +143,111 @@ test("add community starts with create and join choices", async ({ page }) => { await expect( page.getByRole("heading", { name: "Create a new community" }), ).toBeVisible(); + await expect(page.getByTestId("community-create-hosted")).toContainText( + "Host it online", + ); + await expect(page.getByTestId("community-choice-local")).toContainText( + "Keep it on this device", + ); + await page.getByTestId("community-create-hosted").click(); await page.getByRole("button", { name: "Continue to Builderlab" }).click(); await page.getByRole("button", { name: "Connect and continue" }).click(); await expect(page.getByLabel("Community address")).toBeVisible(); }); +test("add community can start an on-this-device community", async ({ + page, +}) => { + await setLocalCommunitiesFeature(page, true); + await installMockBridge( + page, + { applyCommunityDelayMs: 1_000 }, + { seedPreviewFeatures: false }, + ); + await page.goto("/"); + + await openAddCommunityDialog(page); + await page.getByTestId("add-community-create").click(); + await page.getByTestId("community-choice-local").click(); + + await expect + .poll(() => + page.evaluate((key) => { + const raw = window.localStorage.getItem(key); + if (!raw) return null; + const transaction = JSON.parse(raw) as { + source?: string; + communityName?: string; + relayUrl?: string; + }; + return { + source: transaction.source, + communityName: transaction.communityName, + relayUrl: transaction.relayUrl, + }; + }, COMMUNITY_ONBOARDING_STORAGE_KEY), + ) + .toEqual({ + source: "add-community", + communityName: "On this device", + relayUrl: "buzz-local://on-this-device", + }); +}); + +test("create choice offers to open an existing on-this-device community", async ({ + page, +}) => { + await page.addInitScript(() => { + const communities = [ + { + id: "hosted", + name: "Hosted", + relayUrl: "wss://hosted.example.com", + addedAt: "2026-01-01T00:00:00.000Z", + }, + { + id: "local", + name: "On this device", + relayUrl: "buzz-local://on-this-device", + local: true, + addedAt: "2026-01-02T00:00:00.000Z", + }, + ]; + window.localStorage.setItem( + "buzz-communities", + JSON.stringify(communities), + ); + window.localStorage.setItem("buzz-active-community-id", "hosted"); + }); + await setLocalCommunitiesFeature(page, true); + await installMockBridge( + page, + { applyCommunityDelayMs: 1_000 }, + { seedPreviewFeatures: false, skipCommunitySeed: true }, + ); + await page.goto("/"); + + await openAddCommunityDialog(page); + await page.getByTestId("add-community-create").click(); + await expect(page.getByTestId("community-choice-local")).toContainText( + "Open your on-this-device community", + ); + await page.getByTestId("community-choice-local").click(); + await expect + .poll(() => + page.evaluate(() => localStorage.getItem("buzz-active-community-id")), + ) + .toBe("local"); + await expect + .poll(() => + page.evaluate(() => { + const raw = localStorage.getItem("buzz-communities"); + return raw ? JSON.parse(raw).length : 0; + }), + ) + .toBe(2); +}); + test("automatically shows community join requirements near the community URL", async ({ page, }) => { diff --git a/preview-features.json b/preview-features.json index 388f1c39b0..de991d93cd 100644 --- a/preview-features.json +++ b/preview-features.json @@ -30,6 +30,12 @@ "name": "Agent-managed profiles", "description": "Let agents manage their own relay name and avatar instead of restoring the desktop copy", "platforms": ["desktop"] + }, + { + "id": "localCommunities", + "name": "Local communities", + "description": "Create a community that lives entirely on this device (local relay)", + "platforms": ["desktop"] } ] } diff --git a/scripts/bundle-sidecars.sh b/scripts/bundle-sidecars.sh index 8ea5fe2bbe..739c323ddf 100755 --- a/scripts/bundle-sidecars.sh +++ b/scripts/bundle-sidecars.sh @@ -1,14 +1,14 @@ #!/usr/bin/env bash set -euo pipefail -SIDECARS=(buzz-acp buzz-agent buzz-dev-mcp git-credential-nostr buzz) +SIDECARS=(buzz-acp buzz-agent buzz-dev-mcp git-credential-nostr buzz buzz-relay) HOST=$(rustc -vV | sed -n 's|host: ||p') TARGET=${1:-$HOST} if [[ "$TARGET" != *windows* ]]; then SIDECARS+=(buzz-backend-kubernetes) - BUILD_HINT="cargo build --release -p buzz-acp -p buzz-agent -p buzz-backend-kubernetes -p buzz-dev-mcp -p git-credential-nostr -p buzz-cli" + BUILD_HINT="cargo build --release -p buzz-acp -p buzz-agent -p buzz-backend-kubernetes -p buzz-dev-mcp -p git-credential-nostr -p buzz-cli -p buzz-relay" else - BUILD_HINT="cargo build --release -p buzz-acp -p buzz-agent -p buzz-dev-mcp -p git-credential-nostr -p buzz-cli" + BUILD_HINT="cargo build --release -p buzz-acp -p buzz-agent -p buzz-dev-mcp -p git-credential-nostr -p buzz-cli -p buzz-relay" fi BINARIES_DIR="desktop/src-tauri/binaries"