From ceb2b026b62887bfc83fc3c07b126bd5fa995505 Mon Sep 17 00:00:00 2001 From: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> Date: Mon, 10 Aug 2026 15:40:22 -0400 Subject: [PATCH] refactor(db): narrow SQLite community foundation Signed-off-by: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> --- crates/buzz-db/src/lib.rs | 935 ++++++------------ crates/buzz-db/src/sqlite.rs | 860 +--------------- crates/buzz-relay/src/api/bridge.rs | 8 +- .../src/handlers/command_executor.rs | 73 +- crates/buzz-relay/src/handlers/event.rs | 3 +- .../buzz-relay/src/handlers/side_effects.rs | 14 - 6 files changed, 366 insertions(+), 1527 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index cf4aaf94e7..6de335aa42 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -180,7 +180,6 @@ enum DbBackend { #[derive(Clone, Debug)] pub struct Db { backend: DbBackend, - // Retained for Postgres test helpers. All production method access goes through `pg_pool`. pub(crate) pool: PgPool, /// Maximum connections configured for this pool (from [`DbConfig::max_connections`]). pub(crate) max_connections: u32, @@ -649,17 +648,6 @@ pub struct TokenSummary { } impl Db { - /// Returns the PostgreSQL pool or a typed error for local SQLite handles. - fn pg_pool(&self) -> Result<&PgPool> { - match self.backend { - DbBackend::Postgres => Ok(&self.pool), - DbBackend::SQLite(_) => Err(DbError::UnsupportedBackend( - "PostgreSQL operation on SQLite Db", - )), - } - } - - /// Returns the SQLite pool or a typed error for PostgreSQL handles. fn sqlite_pool(&self) -> Result<&sqlx::SqlitePool> { match &self.backend { DbBackend::SQLite(pool) => Ok(pool), @@ -670,8 +658,6 @@ impl Db { } /// Creates a local single-node database backed by SQLite. - /// - /// The embedded local schema is applied before this constructor returns. pub async fn new_sqlite(database_url: &str) -> Result { let sqlite = sqlite::connect(database_url).await?; let pool = PgPoolOptions::new() @@ -688,7 +674,6 @@ impl Db { 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 @@ -887,10 +872,10 @@ impl Db { if self.read_pool.is_none() { return Ok(false); } - replica_fence::verify_floor_guard_catalog(self.pg_pool()?).await?; - replica_fence::verify_floor_guard_behavior(self.pg_pool()?).await?; + replica_fence::verify_floor_guard_catalog(&self.pool).await?; + replica_fence::verify_floor_guard_behavior(&self.pool).await?; tokio::spawn(replica_fence::run_probe( - self.pg_pool()?.clone(), + self.pool.clone(), std::sync::Arc::clone(&self.fence), )); Ok(true) @@ -905,8 +890,8 @@ impl Db { /// reads must go through [`Db::route_read`]-backed entry points; this /// remains only for the fence's own plumbing tests. #[cfg(test)] - fn read(&self) -> Result<&PgPool> { - Ok(self.read_pool.as_ref().unwrap_or(self.pg_pool()?)) + fn read(&self) -> &PgPool { + self.read_pool.as_ref().unwrap_or(&self.pool) } /// Whether a distinct read-replica pool is configured. @@ -1064,7 +1049,7 @@ impl Db { /// Run pending database migrations. pub async fn migrate(&self) -> Result<()> { match &self.backend { - DbBackend::Postgres => migration::run_migrations(self.pg_pool()?).await, + DbBackend::Postgres => migration::run_migrations(&self.pool).await, DbBackend::SQLite(_) => sqlite::migrate(self.sqlite_pool()?).await, } } @@ -1083,19 +1068,9 @@ impl Db { /// `idle` — connections available for immediate reuse /// `max` — pool ceiling set at construction pub fn pool_stats(&self) -> DbPoolStats { - let Some(pool) = (match &self.backend { - DbBackend::Postgres => Some(&self.pool), - DbBackend::SQLite(_) => None, - }) else { - return DbPoolStats { - size: 0, - idle: 0, - max: 0, - }; - }; DbPoolStats { - size: pool.size(), - idle: pool.num_idle() as u32, + size: self.pool.size(), + idle: self.pool.num_idle() as u32, max: self.max_connections, } } @@ -1126,7 +1101,7 @@ impl Db { &self, lock_key: i64, ) -> Result> { - let mut connection = self.pg_pool()?.acquire().await?; + let mut connection = self.pool.acquire().await?; let acquired = sqlx::query_scalar::<_, bool>("SELECT pg_try_advisory_lock($1)") .bind(lock_key) .fetch_one(&mut *connection) @@ -1154,7 +1129,7 @@ impl Db { limit: i64, ) -> Result> { admin_moderation::list_reports( - self.pg_pool()?, + &self.pool, community_id, status, report_type, @@ -1172,7 +1147,7 @@ impl Db { &self, id: Uuid, ) -> Result> { - admin_moderation::get_report(self.pg_pool()?, id).await + admin_moderation::get_report(&self.pool, id).await } /// List feedback for the deployment-global read-only admin plane. @@ -1180,7 +1155,7 @@ impl Db { &self, limit: i64, ) -> Result> { - admin_moderation::list_feedback(self.pg_pool()?, limit).await + admin_moderation::list_feedback(&self.pool, limit).await } /// Fetch one feedback submission for the deployment-global admin plane. @@ -1188,42 +1163,42 @@ impl Db { &self, id: Uuid, ) -> Result> { - admin_moderation::get_feedback(self.pg_pool()?, id).await + admin_moderation::get_feedback(&self.pool, id).await } /// Return total number of communities on this relay. pub async fn usage_community_count(&self) -> Result { - usage::community_count(self.pg_pool()?).await + usage::community_count(&self.pool).await } /// Return per-community user counts split by human/agent. pub async fn usage_user_counts(&self) -> Result> { - usage::user_counts(self.pg_pool()?).await + usage::user_counts(&self.pool).await } /// Return per-community channel counts by type. pub async fn usage_channel_counts(&self) -> Result> { - usage::channel_counts(self.pg_pool()?).await + usage::channel_counts(&self.pool).await } /// Return per-community kind=9 message counts. pub async fn usage_message_counts(&self) -> Result> { - usage::message_counts(self.pg_pool()?).await + usage::message_counts(&self.pool).await } /// Return per-community relay-member counts by role. pub async fn usage_relay_member_counts(&self) -> Result> { - usage::relay_member_counts(self.pg_pool()?).await + usage::relay_member_counts(&self.pool).await } /// Return per-community workflow counts by status. pub async fn usage_workflow_counts(&self) -> Result> { - usage::workflow_counts(self.pg_pool()?).await + usage::workflow_counts(&self.pool).await } /// Return per-community git-repo counts. pub async fn usage_git_repo_counts(&self) -> Result> { - usage::git_repo_counts(self.pg_pool()?).await + usage::git_repo_counts(&self.pool).await } /// Return per-community distinct active-user counts for a given SQL interval. @@ -1233,7 +1208,7 @@ impl Db { &self, interval_sql: &'static str, ) -> Result> { - usage::active_user_counts(self.pg_pool()?, interval_sql).await + usage::active_user_counts(&self.pool, interval_sql).await } /// Return per-community active-channel counts for a given SQL interval. @@ -1241,12 +1216,12 @@ impl Db { &self, interval_sql: &'static str, ) -> Result> { - usage::active_channel_counts(self.pg_pool()?, interval_sql).await + usage::active_channel_counts(&self.pool, interval_sql).await } /// Return all community id → host mappings. pub async fn usage_community_hosts(&self) -> Result> { - usage::community_hosts(self.pg_pool()?).await + usage::community_hosts(&self.pool).await } /// Begin a database transaction for atomic multi-statement operations. @@ -1254,7 +1229,7 @@ impl Db { /// Returns a `'static` transaction because `PgPool` is `Arc`-backed internally. /// The transaction holds an owned pool handle, not a borrow. pub async fn begin_transaction(&self) -> Result> { - self.pg_pool()?.begin().await.map_err(Into::into) + self.pool.begin().await.map_err(Into::into) } /// Returns the community mapped to a normalized request host, if one exists. @@ -1277,7 +1252,7 @@ impl Db { "#, ) .bind(normalized_host) - .fetch_optional(self.pg_pool()?) + .fetch_optional(&self.pool) .await?; row.map(|row| { @@ -1301,7 +1276,7 @@ impl Db { "SELECT EXISTS(SELECT 1 FROM communities WHERE id = $1 AND archived_at IS NULL)", ) .bind(community_id.as_uuid()) - .fetch_one(self.pg_pool()?) + .fetch_one(&self.pool) .await?; Ok(active) } @@ -1313,7 +1288,7 @@ impl Db { ) -> Result> { let row = sqlx::query("SELECT id, host FROM communities WHERE lower(host) = lower($1)") .bind(normalized_host) - .fetch_optional(self.pg_pool()?) + .fetch_optional(&self.pool) .await?; row.map(|row| { Ok(CommunityRecord { @@ -1344,7 +1319,7 @@ impl Db { "#, ) .bind(owner_pubkey) - .fetch_all(self.pg_pool()?) + .fetch_all(&self.pool) .await?; rows.into_iter() @@ -1386,7 +1361,7 @@ impl Db { "#, ) .bind(community_id.as_uuid()) - .fetch_optional(self.pg_pool()?) + .fetch_optional(&self.pool) .await?; row.map(|row| { @@ -1396,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 @@ -1409,7 +1400,7 @@ impl Db { "#, ) .bind(community_id.as_uuid()) - .fetch_optional(self.pg_pool()?) + .fetch_optional(&self.pool) .await?; Ok(row @@ -1434,32 +1425,11 @@ impl Db { ) .bind(community_id.as_uuid()) .bind(icon) - .execute(self.pg_pool()?) + .execute(&self.pool) .await?; Ok(()) } - /// Rebind the durable loopback community to this single-node process's - /// current ephemeral authority while preserving its stable UUID and data. - /// SQLite-only: production deployments never rewrite tenant host mappings. - pub async fn rebind_single_node_community_host( - &self, - normalized_host: &str, - owner_pubkey: &str, - ) -> Result> { - if matches!(&self.backend, DbBackend::SQLite(_)) { - return sqlite::rebind_single_node_community_host( - self.sqlite_pool()?, - normalized_host, - owner_pubkey, - ) - .await; - } - Err(DbError::UnsupportedBackend( - "single-node community rebind requires SQLite", - )) - } - /// Ensure a configured community host exists and return its row. /// /// This is the startup/config seeding path for N=1 deployments. Migrations @@ -1481,7 +1451,7 @@ impl Db { "#, ) .bind(normalized_host) - .fetch_one(self.pg_pool()?) + .fetch_one(&self.pool) .await?; let id: Uuid = row.try_get("id")?; @@ -1506,7 +1476,7 @@ impl Db { owner_pubkey: &str, ) -> Result { let owner_pubkey = owner_pubkey.to_ascii_lowercase(); - let mut tx = self.pg_pool()?.begin().await?; + let mut tx = self.pool.begin().await?; // Serialize on the owner pubkey so concurrent creates to the same // owner cannot both pass the ownership count check. @@ -1605,7 +1575,7 @@ impl Db { .bind(normalized_host) .bind(owner_pubkey) .bind(protected_deployment_host) - .fetch_optional(self.pg_pool()?) + .fetch_optional(&self.pool) .await?; row.map(|row| { Ok(ArchivedCommunityRecord { @@ -1635,7 +1605,7 @@ impl Db { ) .bind(normalized_host) .bind(owner_pubkey) - .fetch_optional(self.pg_pool()?) + .fetch_optional(&self.pool) .await?; row.map(|row| { Ok(UnarchivedCommunityRecord { @@ -1660,7 +1630,7 @@ impl Db { "#, ) .bind(channel_id) - .fetch_optional(self.pg_pool()?) + .fetch_optional(&self.pool) .await?; row.map(|row| { @@ -1695,30 +1665,25 @@ impl Db { if channel_ids.is_empty() { return Ok(std::collections::HashMap::new()); } - match &self.backend { - DbBackend::SQLite(pool) => sqlite::communities_of_channels(pool, channel_ids).await, - DbBackend::Postgres => { - let rows = sqlx::query( - r#" + let rows = sqlx::query( + r#" SELECT id, community_id FROM channels WHERE id = ANY($1) AND deleted_at IS NULL "#, - ) - .bind(channel_ids) - .fetch_all(self.pg_pool()?) - .await?; + ) + .bind(channel_ids) + .fetch_all(&self.pool) + .await?; - let mut out = std::collections::HashMap::with_capacity(rows.len()); - for row in rows { - let ch: Uuid = row.try_get("id")?; - let cm: Uuid = row.try_get("community_id")?; - out.insert(ch, CommunityId::from_uuid(cm)); - } - Ok(out) - } + let mut out = std::collections::HashMap::with_capacity(rows.len()); + for row in rows { + let ch: Uuid = row.try_get("id")?; + let cm: Uuid = row.try_get("community_id")?; + out.insert(ch, CommunityId::from_uuid(cm)); } + Ok(out) } /// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate. @@ -1728,13 +1693,9 @@ impl Db { event: &nostr::Event, channel_id: Option, ) -> Result<(StoredEvent, bool)> { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::insert_event(pool, community_id, event, channel_id).await; - } - let result = event::insert_event(self.pg_pool()?, community_id, event, channel_id).await?; + let result = event::insert_event(&self.pool, community_id, event, channel_id).await?; if result.1 { - if let Err(e) = insert_mentions(self.pg_pool()?, community_id, event, channel_id).await - { + if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } } @@ -1749,10 +1710,7 @@ impl Db { /// [`Db::query_events_routed`] instead — converting a caller is an /// explicit, per-callsite decision, never a change to this method. pub async fn query_events(&self, q: &EventQuery) -> Result> { - match &self.backend { - DbBackend::SQLite(pool) => sqlite::query_events(pool, q).await, - DbBackend::Postgres => event::query_events(self.pg_pool()?, q).await, - } + event::query_events(&self.pool, q).await } /// [`Db::query_events`] with replica routing — the opt-in fast path for @@ -1776,9 +1734,6 @@ impl Db { path: &'static str, q: &EventQuery, ) -> Result> { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::query_events(pool, q).await; - } let predicate = RoutePredicate::for_query(q, self.replica_read_max_age.is_some()); match self.route_read(path, predicate).await { RouteDecision::Replica(mut tx, _entry, reason) => { @@ -1792,11 +1747,11 @@ impl Db { // writer rather than surfacing a routed error. tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); - event::query_events(self.pg_pool()?, q).await + event::query_events(&self.pool, q).await } } } - RouteDecision::Writer => event::query_events(self.pg_pool()?, q).await, + RouteDecision::Writer => event::query_events(&self.pool, q).await, } } @@ -1813,9 +1768,6 @@ impl Db { path: &'static str, q: &EventQuery, ) -> Result> { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::query_events(pool, q).await; - } match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { match event::query_events_on(&mut tx, q).await { @@ -1826,11 +1778,11 @@ impl Db { Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); - event::query_events(self.pg_pool()?, q).await + event::query_events(&self.pool, q).await } } } - RouteDecision::Writer => event::query_events(self.pg_pool()?, q).await, + RouteDecision::Writer => event::query_events(&self.pool, q).await, } } @@ -1839,7 +1791,7 @@ impl Db { /// Always reads from the WRITER pool — see [`Db::query_events`] for the /// writer-vs-routed rule. pub async fn count_events(&self, q: &EventQuery) -> Result { - event::count_events(self.pg_pool()?, q).await + event::count_events(&self.pool, q).await } /// [`Db::count_events`] with replica routing — same contract, rules, @@ -1863,11 +1815,11 @@ impl Db { Err(e) => { tracing::warn!(path, "replica count failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); - event::count_events(self.pg_pool()?, q).await + event::count_events(&self.pool, q).await } } } - RouteDecision::Writer => event::count_events(self.pg_pool()?, q).await, + RouteDecision::Writer => event::count_events(&self.pool, q).await, } } @@ -1881,7 +1833,7 @@ impl Db { creator_pubkey: &[u8], ) -> Result { event::huddle_started_link_exists( - self.pg_pool()?, + &self.pool, community_id, parent_channel_id, ephemeral_channel_id, @@ -1901,8 +1853,7 @@ impl Db { kind: i32, pubkey_bytes: &[u8], ) -> Result> { - event::get_latest_global_replaceable(self.pg_pool()?, community_id, kind, pubkey_bytes) - .await + event::get_latest_global_replaceable(&self.pool, community_id, kind, pubkey_bytes).await } /// Fetches a single non-deleted event by its raw ID bytes. @@ -1913,12 +1864,7 @@ impl Db { community_id: CommunityId, id_bytes: &[u8], ) -> Result> { - match &self.backend { - DbBackend::SQLite(pool) => sqlite::get_event_by_id(pool, community_id, id_bytes).await, - DbBackend::Postgres => { - event::get_event_by_id(self.pg_pool()?, community_id, id_bytes).await - } - } + event::get_event_by_id(&self.pool, community_id, id_bytes).await } /// Fetches a single event by its raw ID bytes, **including soft-deleted rows**. @@ -1927,13 +1873,7 @@ impl Db { community_id: CommunityId, id_bytes: &[u8], ) -> Result> { - match &self.backend { - DbBackend::SQLite(pool) => sqlite::get_event_by_id(pool, community_id, id_bytes).await, - DbBackend::Postgres => { - event::get_event_by_id_including_deleted(self.pg_pool()?, community_id, id_bytes) - .await - } - } + event::get_event_by_id_including_deleted(&self.pool, community_id, id_bytes).await } /// Soft-deletes an event. Returns `Ok(true)` if deleted, `Ok(false)` if already deleted. @@ -1942,7 +1882,7 @@ impl Db { community_id: CommunityId, event_id: &[u8], ) -> Result { - event::soft_delete_event(self.pg_pool()?, community_id, event_id).await + event::soft_delete_event(&self.pool, community_id, event_id).await } /// Soft-delete the live row for an addressable coordinate `(kind, pubkey, d_tag)` @@ -1958,7 +1898,7 @@ impl Db { deletion_created_at_secs: i64, ) -> Result { event::soft_delete_by_coordinate( - self.pg_pool()?, + &self.pool, community_id, kind, pubkey, @@ -1976,21 +1916,14 @@ impl Db { parent_event_id: Option<&[u8]>, root_event_id: Option<&[u8]>, ) -> Result { - match &self.backend { - DbBackend::SQLite(pool) => { - sqlite::soft_delete_event(pool, community_id, event_id).await - } - DbBackend::Postgres => { - event::soft_delete_event_and_update_thread( - self.pg_pool()?, - community_id, - event_id, - parent_event_id, - root_event_id, - ) - .await - } - } + event::soft_delete_event_and_update_thread( + &self.pool, + community_id, + event_id, + parent_event_id, + root_event_id, + ) + .await } /// Returns the most recent `created_at` for a channel. @@ -1999,7 +1932,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result>> { - event::get_last_message_at(self.pg_pool()?, community_id, channel_id).await + event::get_last_message_at(&self.pool, community_id, channel_id).await } /// Bulk-fetch the most recent `created_at` for a set of channel IDs. @@ -2008,7 +1941,7 @@ impl Db { community_id: CommunityId, channel_ids: &[Uuid], ) -> Result>> { - event::get_last_message_at_bulk(self.pg_pool()?, community_id, channel_ids).await + event::get_last_message_at_bulk(&self.pool, community_id, channel_ids).await } /// Batch-fetch non-deleted events by their raw IDs. @@ -2017,7 +1950,7 @@ impl Db { community_id: CommunityId, ids: &[&[u8]], ) -> Result> { - event::get_events_by_ids(self.pg_pool()?, community_id, ids).await + event::get_events_by_ids(&self.pool, community_id, ids).await } /// [`Db::get_events_by_ids`] with replica routing — same contract and @@ -2043,13 +1976,11 @@ impl Db { Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); - event::get_events_by_ids(self.pg_pool()?, community_id, ids).await + event::get_events_by_ids(&self.pool, community_id, ids).await } } } - RouteDecision::Writer => { - event::get_events_by_ids(self.pg_pool()?, community_id, ids).await - } + RouteDecision::Writer => event::get_events_by_ids(&self.pool, community_id, ids).await, } } @@ -2059,7 +1990,7 @@ impl Db { limit: i64, lease_until: DateTime, ) -> Result> { - push::claim_due_match_batch(self.pg_pool()?, limit, lease_until).await + push::claim_due_match_batch(&self.pool, limit, lease_until).await } /// Load active endpoint-enabled leases eligible for push matching. @@ -2067,7 +1998,7 @@ impl Db { &self, community: CommunityId, ) -> Result> { - push::active_match_leases(self.pg_pool()?, community).await + push::active_match_leases(&self.pool, community).await } /// Complete matcher jobs from one claimed batch while the fence holds. @@ -2077,7 +2008,7 @@ impl Db { claim_id: uuid::Uuid, event_ids: &[Vec], ) -> Result { - push::complete_match_batch(self.pg_pool()?, community, claim_id, event_ids).await + push::complete_match_batch(&self.pool, community, claim_id, event_ids).await } /// Release fenced matcher claims from one batch for retry. @@ -2088,12 +2019,12 @@ impl Db { event_ids: &[Vec], next: DateTime, ) -> Result { - push::retry_match_batch(self.pg_pool()?, community, claim_id, event_ids, next).await + push::retry_match_batch(&self.pool, community, claim_id, event_ids, next).await } /// Delete exhausted matcher jobs (periodic sweep, off the claim path). pub async fn reap_exhausted_push_matches(&self) -> Result { - push::reap_exhausted_matches(self.pg_pool()?).await + push::reap_exhausted_matches(&self.pool).await } /// Idempotently enqueue a wake for a matched lease and event. @@ -2104,7 +2035,7 @@ impl Db { installation_id: &str, wake: push::NewWake<'_>, ) -> Result { - push::enqueue_wake(self.pg_pool()?, community, author, installation_id, wake).await + push::enqueue_wake(&self.pool, community, author, installation_id, wake).await } /// Set-wise [`Self::enqueue_push_wake`]: one transaction per batch. @@ -2113,7 +2044,7 @@ impl Db { community: CommunityId, requests: &[push::WakeRequest], ) -> Result> { - push::enqueue_wakes(self.pg_pool()?, community, requests).await + push::enqueue_wakes(&self.pool, community, requests).await } /// Exclusively claim due wake jobs for one community. @@ -2123,7 +2054,7 @@ impl Db { limit: i64, lease_until: DateTime, ) -> Result> { - push::claim_due_wakes(self.pg_pool()?, community, limit, lease_until).await + push::claim_due_wakes(&self.pool, community, limit, lease_until).await } /// Revalidate a wake's claim, source event, and current lease before send. @@ -2133,7 +2064,7 @@ impl Db { id: Uuid, claim_id: Uuid, ) -> Result { - push::revalidate_wake_for_send(self.pg_pool()?, community, id, claim_id).await + push::revalidate_wake_for_send(&self.pool, community, id, claim_id).await } /// Mark a fenced wake claim delivered. @@ -2143,7 +2074,7 @@ impl Db { id: Uuid, claim_id: Uuid, ) -> Result { - push::complete_wake(self.pg_pool()?, community, id, claim_id).await + push::complete_wake(&self.pool, community, id, claim_id).await } /// Release a fenced wake claim for retry at the supplied time. @@ -2154,7 +2085,7 @@ impl Db { claim_id: Uuid, next: DateTime, ) -> Result { - push::retry_wake(self.pg_pool()?, community, id, claim_id, next).await + push::retry_wake(&self.pool, community, id, claim_id, next).await } /// Mark a fenced wake claim terminally failed. @@ -2164,7 +2095,7 @@ impl Db { id: Uuid, claim_id: Uuid, ) -> Result { - push::fail_wake(self.pg_pool()?, community, id, claim_id).await + push::fail_wake(&self.pool, community, id, claim_id).await } /// Disable an endpoint only if the specified lease generation is current. @@ -2176,7 +2107,7 @@ impl Db { generation: i64, ) -> Result { push::disable_endpoint_generation( - self.pg_pool()?, + &self.pool, community, author, installation_id, @@ -2197,7 +2128,7 @@ impl Db { max_active_leases: i64, ) -> Result { push::accept_lease_event( - self.pg_pool()?, + &self.pool, community, event, installation_id, @@ -2216,15 +2147,8 @@ impl Db { channel_id: Option, thread_meta: Option>, ) -> Result<(StoredEvent, bool)> { - if let DbBackend::SQLite(pool) = &self.backend { - // SQLite Phase 1 persists the canonical event; thread relationships - // remain durable in the event's NIP-10 tags until Phase 2 ports the - // denormalized thread metadata tables. - let _ = thread_meta; - return sqlite::insert_event(pool, community_id, event, channel_id).await; - } let result = event::insert_event_with_thread_metadata( - self.pg_pool()?, + &self.pool, community_id, event, channel_id, @@ -2232,8 +2156,7 @@ impl Db { ) .await?; if result.1 { - if let Err(e) = insert_mentions(self.pg_pool()?, community_id, event, channel_id).await - { + if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } } @@ -2252,21 +2175,8 @@ impl Db { actor_pubkey: &[u8], emoji: &str, ) -> Result { - if let DbBackend::SQLite(pool) = &self.backend { - let _ = thread_meta; - return sqlite::insert_reaction_event( - pool, - community_id, - event, - channel_id, - target_event_id, - actor_pubkey, - emoji, - ) - .await; - } let outcome = event::insert_reaction_event_with_thread_metadata( - self.pg_pool()?, + &self.pool, community_id, event, channel_id, @@ -2280,8 +2190,7 @@ impl Db { was_inserted: true, .. } = &outcome { - if let Err(e) = insert_mentions(self.pg_pool()?, community_id, event, channel_id).await - { + if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } } @@ -2301,7 +2210,7 @@ impl Db { ttl_seconds: Option, ) -> Result { channel::create_channel( - self.pg_pool()?, + &self.pool, community_id, name, channel_type, @@ -2343,7 +2252,7 @@ impl Db { .await; } channel::create_channel_with_id( - self.pg_pool()?, + &self.pool, community_id, channel_id, name, @@ -2364,9 +2273,7 @@ impl Db { ) -> Result { match &self.backend { DbBackend::SQLite(pool) => sqlite::get_channel(pool, community_id, channel_id).await, - DbBackend::Postgres => { - channel::get_channel(self.pg_pool()?, community_id, channel_id).await - } + DbBackend::Postgres => channel::get_channel(&self.pool, community_id, channel_id).await, } } @@ -2376,7 +2283,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result> { - channel::get_canvas(self.pg_pool()?, community_id, channel_id).await + channel::get_canvas(&self.pool, community_id, channel_id).await } /// Sets or clears the canvas content for a channel. @@ -2386,7 +2293,7 @@ impl Db { channel_id: Uuid, canvas: Option<&str>, ) -> Result<()> { - channel::set_canvas(self.pg_pool()?, community_id, channel_id, canvas).await + channel::set_canvas(&self.pool, community_id, channel_id, canvas).await } /// Adds a member to a channel. @@ -2403,7 +2310,7 @@ impl Db { .await; } channel::add_member( - self.pg_pool()?, + &self.pool, community_id, channel_id, pubkey, @@ -2421,14 +2328,7 @@ impl Db { pubkey: &[u8], actor_pubkey: &[u8], ) -> Result<()> { - channel::remove_member( - self.pg_pool()?, - community_id, - channel_id, - pubkey, - actor_pubkey, - ) - .await + channel::remove_member(&self.pool, community_id, channel_id, pubkey, actor_pubkey).await } /// Returns `true` if the pubkey is an active member. @@ -2443,7 +2343,7 @@ impl Db { sqlite::is_member(pool, community_id, channel_id, pubkey).await } DbBackend::Postgres => { - channel::is_member(self.pg_pool()?, community_id, channel_id, pubkey).await + channel::is_member(&self.pool, community_id, channel_id, pubkey).await } } } @@ -2456,7 +2356,7 @@ impl Db { channel_ids: &[Uuid], pubkeys: &[Vec], ) -> Result)>> { - channel::membership_pairs(self.pg_pool()?, community_id, channel_ids, pubkeys).await + channel::membership_pairs(&self.pool, community_id, channel_ids, pubkeys).await } /// Returns all active members of a channel. @@ -2467,9 +2367,7 @@ impl Db { ) -> Result> { match &self.backend { DbBackend::SQLite(pool) => sqlite::get_members(pool, community_id, channel_id).await, - DbBackend::Postgres => { - channel::get_members(self.pg_pool()?, community_id, channel_id).await - } + DbBackend::Postgres => channel::get_members(&self.pool, community_id, channel_id).await, } } @@ -2479,7 +2377,7 @@ impl Db { community_id: CommunityId, channel_ids: &[Uuid], ) -> Result> { - channel::get_members_bulk(self.pg_pool()?, community_id, channel_ids).await + channel::get_members_bulk(&self.pool, community_id, channel_ids).await } /// Get all channel IDs accessible to a pubkey. @@ -2493,7 +2391,7 @@ impl Db { sqlite::get_accessible_channel_ids(pool, community_id, pubkey).await } DbBackend::Postgres => { - channel::get_accessible_channel_ids(self.pg_pool()?, community_id, pubkey).await + channel::get_accessible_channel_ids(&self.pool, community_id, pubkey).await } } } @@ -2504,7 +2402,7 @@ impl Db { community_id: CommunityId, visibility: Option<&str>, ) -> Result> { - channel::list_channels(self.pg_pool()?, community_id, visibility).await + channel::list_channels(&self.pool, community_id, visibility).await } /// Returns full channel records for all channels a user can access. @@ -2516,7 +2414,7 @@ impl Db { member_only: Option, ) -> Result> { channel::get_accessible_channels( - self.pg_pool()?, + &self.pool, community_id, pubkey, visibility_filter, @@ -2530,7 +2428,7 @@ impl Db { &self, community_id: CommunityId, ) -> Result> { - channel::get_bot_members(self.pg_pool()?, community_id).await + channel::get_bot_members(&self.pool, community_id).await } /// Bulk-fetch user records by pubkey. @@ -2539,7 +2437,7 @@ impl Db { community_id: CommunityId, pubkeys: &[Vec], ) -> Result> { - channel::get_users_bulk(self.pg_pool()?, community_id, pubkeys).await + channel::get_users_bulk(&self.pool, community_id, pubkeys).await } /// Updates a channel's name and/or description. @@ -2549,7 +2447,7 @@ impl Db { channel_id: Uuid, updates: channel::ChannelUpdate, ) -> Result { - channel::update_channel(self.pg_pool()?, community_id, channel_id, updates).await + channel::update_channel(&self.pool, community_id, channel_id, updates).await } /// Sets the topic for a channel. @@ -2560,7 +2458,7 @@ impl Db { topic: &str, set_by: &[u8], ) -> Result<()> { - channel::set_topic(self.pg_pool()?, community_id, channel_id, topic, set_by).await + channel::set_topic(&self.pool, community_id, channel_id, topic, set_by).await } /// Sets the purpose for a channel. @@ -2571,12 +2469,12 @@ impl Db { purpose: &str, set_by: &[u8], ) -> Result<()> { - channel::set_purpose(self.pg_pool()?, community_id, channel_id, purpose, set_by).await + channel::set_purpose(&self.pool, community_id, channel_id, purpose, set_by).await } /// Archives a channel. pub async fn archive_channel(&self, community_id: CommunityId, channel_id: Uuid) -> Result<()> { - channel::archive_channel(self.pg_pool()?, community_id, channel_id).await + channel::archive_channel(&self.pool, community_id, channel_id).await } /// Unarchives a channel. @@ -2585,7 +2483,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result<()> { - channel::unarchive_channel(self.pg_pool()?, community_id, channel_id).await + channel::unarchive_channel(&self.pool, community_id, channel_id).await } /// Soft-delete a channel. @@ -2594,7 +2492,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result { - channel::soft_delete_channel(self.pg_pool()?, community_id, channel_id).await + channel::soft_delete_channel(&self.pool, community_id, channel_id).await } /// Returns the count of active members in a channel. @@ -2603,7 +2501,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result { - channel::get_member_count(self.pg_pool()?, community_id, channel_id).await + channel::get_member_count(&self.pool, community_id, channel_id).await } /// Bulk-fetch member counts for a set of channel IDs. @@ -2612,7 +2510,7 @@ impl Db { community_id: CommunityId, channel_ids: &[Uuid], ) -> Result> { - channel::get_member_counts_bulk(self.pg_pool()?, community_id, channel_ids).await + channel::get_member_counts_bulk(&self.pool, community_id, channel_ids).await } /// Get the active role of a pubkey in a channel. @@ -2622,21 +2520,14 @@ impl Db { channel_id: Uuid, pubkey: &[u8], ) -> Result> { - match &self.backend { - DbBackend::SQLite(pool) => { - sqlite::get_member_role(pool, community_id, channel_id, pubkey).await - } - DbBackend::Postgres => { - channel::get_member_role(self.pg_pool()?, community_id, channel_id, pubkey).await - } - } + channel::get_member_role(&self.pool, community_id, channel_id, pubkey).await } /// Archive ephemeral channels whose TTL deadline has passed. pub async fn reap_expired_ephemeral_channels( &self, ) -> Result> { - channel::reap_expired_ephemeral_channels(self.pg_pool()?).await + channel::reap_expired_ephemeral_channels(&self.pool).await } /// Query due reminders ready for delivery. @@ -2645,7 +2536,7 @@ impl Db { now_secs: i64, batch_limit: i64, ) -> Result> { - event::query_due_reminders(self.pg_pool()?, now_secs, batch_limit).await + event::query_due_reminders(&self.pool, now_secs, batch_limit).await } /// Atomically claim a due reminder for delivery (cross-pod dedup). @@ -2655,7 +2546,7 @@ impl Db { event_id: &[u8], event_created_at: chrono::DateTime, ) -> Result { - event::claim_due_reminder(self.pg_pool()?, community_id, event_id, event_created_at).await + event::claim_due_reminder(&self.pool, community_id, event_id, event_created_at).await } /// Atomically claim a due reminder using a caller-supplied delivery stamp. @@ -2667,7 +2558,7 @@ impl Db { delivery_stamp: i64, ) -> Result { event::claim_due_reminder_with_stamp( - self.pg_pool()?, + &self.pool, community_id, event_id, event_created_at, @@ -2685,7 +2576,7 @@ impl Db { delivery_stamp: i64, ) -> Result { event::release_due_reminder( - self.pg_pool()?, + &self.pool, community_id, event_id, event_created_at, @@ -2700,10 +2591,7 @@ impl Db { /// already existed. Callers use the `true` return to increment /// `buzz_users_created_total`. pub async fn ensure_user(&self, community_id: CommunityId, pubkey: &[u8]) -> Result { - match &self.backend { - DbBackend::SQLite(pool) => sqlite::ensure_user(pool, community_id, pubkey).await, - DbBackend::Postgres => user::ensure_user(self.pg_pool()?, community_id, pubkey).await, - } + user::ensure_user(&self.pool, community_id, pubkey).await } /// Get a single user record by pubkey. @@ -2712,10 +2600,7 @@ impl Db { community_id: CommunityId, pubkey: &[u8], ) -> Result> { - match &self.backend { - DbBackend::SQLite(pool) => sqlite::get_user(pool, community_id, pubkey).await, - DbBackend::Postgres => user::get_user(self.pg_pool()?, community_id, pubkey).await, - } + user::get_user(&self.pool, community_id, pubkey).await } /// Update a user's profile fields. @@ -2728,32 +2613,16 @@ impl Db { about: Option<&str>, nip05_handle: Option<&str>, ) -> Result<()> { - match &self.backend { - DbBackend::SQLite(pool) => { - sqlite::update_user_profile( - pool, - community_id, - pubkey, - display_name, - avatar_url, - about, - nip05_handle, - ) - .await - } - DbBackend::Postgres => { - user::update_user_profile( - self.pg_pool()?, - community_id, - pubkey, - display_name, - avatar_url, - about, - nip05_handle, - ) - .await - } - } + user::update_user_profile( + &self.pool, + community_id, + pubkey, + display_name, + avatar_url, + about, + nip05_handle, + ) + .await } /// Look up a user by NIP-05 handle. @@ -2763,7 +2632,7 @@ impl Db { local_part: &str, domain: &str, ) -> Result> { - user::get_user_by_nip05(self.pg_pool()?, community_id, local_part, domain).await + user::get_user_by_nip05(&self.pool, community_id, local_part, domain).await } /// Search users by display name, NIP-05 handle, or pubkey prefix. @@ -2773,7 +2642,7 @@ impl Db { query: &str, limit: u32, ) -> Result> { - user::search_users(self.pg_pool()?, community_id, query, limit).await + user::search_users(&self.pool, community_id, query, limit).await } /// Atomically set agent owner — only if no owner is currently assigned. @@ -2784,15 +2653,7 @@ impl Db { agent_pubkey: &[u8], owner_pubkey: &[u8], ) -> Result { - match &self.backend { - DbBackend::SQLite(pool) => { - sqlite::set_agent_owner(pool, community_id, agent_pubkey, owner_pubkey).await - } - DbBackend::Postgres => { - user::set_agent_owner(self.pg_pool()?, community_id, agent_pubkey, owner_pubkey) - .await - } - } + user::set_agent_owner(&self.pool, community_id, agent_pubkey, owner_pubkey).await } /// Get the channel_add_policy and agent_owner_pubkey for a user. @@ -2801,14 +2662,7 @@ impl Db { community_id: CommunityId, pubkey: &[u8], ) -> Result>)>> { - match &self.backend { - DbBackend::SQLite(pool) => { - sqlite::get_agent_channel_policy(pool, community_id, pubkey).await - } - DbBackend::Postgres => { - user::get_agent_channel_policy(self.pg_pool()?, community_id, pubkey).await - } - } + user::get_agent_channel_policy(&self.pool, community_id, pubkey).await } /// Check whether `actor_pubkey` is the agent owner of `target_pubkey`. @@ -2818,15 +2672,7 @@ impl Db { target_pubkey: &[u8], actor_pubkey: &[u8], ) -> Result { - match &self.backend { - DbBackend::SQLite(pool) => { - sqlite::is_agent_owner(pool, community_id, target_pubkey, actor_pubkey).await - } - DbBackend::Postgres => { - user::is_agent_owner(self.pg_pool()?, community_id, target_pubkey, actor_pubkey) - .await - } - } + user::is_agent_owner(&self.pool, community_id, target_pubkey, actor_pubkey).await } /// Set the channel_add_policy for a user. @@ -2836,7 +2682,7 @@ impl Db { pubkey: &[u8], policy: &str, ) -> Result<()> { - user::set_channel_add_policy(self.pg_pool()?, community_id, pubkey, policy).await + user::set_channel_add_policy(&self.pool, community_id, pubkey, policy).await } /// Find an existing DM by its participant hash. @@ -2845,7 +2691,7 @@ impl Db { community_id: CommunityId, participant_hash: &[u8], ) -> Result> { - dm::find_dm_by_participants(self.pg_pool()?, community_id, participant_hash).await + dm::find_dm_by_participants(&self.pool, community_id, participant_hash).await } /// Create or return an existing DM channel. @@ -2855,7 +2701,7 @@ impl Db { participants: &[&[u8]], created_by: &[u8], ) -> Result { - dm::create_dm(self.pg_pool()?, community_id, participants, created_by).await + dm::create_dm(&self.pool, community_id, participants, created_by).await } /// List all DMs for a user. @@ -2866,7 +2712,7 @@ impl Db { limit: u32, cursor: Option, ) -> Result> { - dm::list_dms_for_user(self.pg_pool()?, community_id, pubkey, limit, cursor).await + dm::list_dms_for_user(&self.pool, community_id, pubkey, limit, cursor).await } /// Open or retrieve a DM for the given participants. @@ -2876,33 +2722,7 @@ impl Db { pubkeys: &[&[u8]], created_by: &[u8], ) -> Result<(channel::ChannelRecord, bool)> { - match &self.backend { - DbBackend::SQLite(pool) => { - sqlite::open_dm(pool, community_id, pubkeys, created_by, None) - .await? - .ok_or_else(|| DbError::InvalidData("unexpected duplicate DM open".into())) - } - DbBackend::Postgres => { - dm::open_dm(self.pg_pool()?, community_id, pubkeys, created_by).await - } - } - } - - /// Atomically persist a SQLite DM-open command and its channel mutation. - /// `None` means the command event was already processed. - pub async fn open_dm_sqlite_command( - &self, - community_id: CommunityId, - pubkeys: &[&[u8]], - created_by: &[u8], - event: &nostr::Event, - ) -> Result> { - match &self.backend { - DbBackend::SQLite(pool) => { - sqlite::open_dm(pool, community_id, pubkeys, created_by, Some(event)).await - } - DbBackend::Postgres => Ok(None), - } + dm::open_dm(&self.pool, community_id, pubkeys, created_by).await } /// Hide a DM channel for a specific user. @@ -2915,7 +2735,7 @@ impl Db { channel_id: Uuid, pubkey: &[u8], ) -> Result<()> { - dm::hide_dm(self.pg_pool()?, community_id, channel_id, pubkey).await + dm::hide_dm(&self.pool, community_id, channel_id, pubkey).await } /// Unhide a DM channel for a specific user. @@ -2925,7 +2745,7 @@ impl Db { channel_id: Uuid, pubkey: &[u8], ) -> Result<()> { - dm::unhide_dm(self.pg_pool()?, community_id, channel_id, pubkey).await + dm::unhide_dm(&self.pool, community_id, channel_id, pubkey).await } /// List the channel IDs of all DMs the given user currently has hidden. @@ -2934,7 +2754,7 @@ impl Db { community_id: CommunityId, pubkey: &[u8], ) -> Result> { - dm::list_hidden_dms(self.pg_pool()?, community_id, pubkey).await + dm::list_hidden_dms(&self.pool, community_id, pubkey).await } /// Insert thread metadata. @@ -2953,7 +2773,7 @@ impl Db { broadcast: bool, ) -> Result<()> { thread::insert_thread_metadata( - self.pg_pool()?, + &self.pool, community_id, event_id, event_created_at, @@ -3052,7 +2872,7 @@ impl Db { } } thread::get_thread_replies( - self.pg_pool()?, + &self.pool, community_id, root_event_id, depth_limit, @@ -3068,7 +2888,7 @@ impl Db { community_id: CommunityId, event_id: &[u8], ) -> Result> { - thread::get_thread_summary(self.pg_pool()?, community_id, event_id).await + thread::get_thread_summary(&self.pool, community_id, event_id).await } /// One channel window: top-level rows + summaries + server `has_more`. @@ -3152,7 +2972,7 @@ impl Db { ReadSession { inner: ReadSessionInner::Replica { tx, - writer: self.pg_pool()?.clone(), + writer: self.pool.clone(), }, }, )); @@ -3176,7 +2996,7 @@ impl Db { RouteDecision::Writer => {} } let window = thread::get_channel_window( - self.pg_pool()?, + &self.pool, community_id, channel_id, limit, @@ -3187,7 +3007,7 @@ impl Db { Ok(( window, ReadSession { - inner: ReadSessionInner::Writer(self.pg_pool()?.clone()), + inner: ReadSessionInner::Writer(self.pool.clone()), }, )) } @@ -3284,12 +3104,7 @@ impl Db { community_id: CommunityId, event_id: &[u8], ) -> Result> { - if matches!(&self.backend, DbBackend::SQLite(_)) { - // SQLite Phase 1 derives ancestry from durable NIP-10 event tags; - // denormalized counters and metadata arrive with the Phase 2 port. - return Ok(None); - } - thread::get_thread_metadata_by_event(self.pg_pool()?, community_id, event_id).await + thread::get_thread_metadata_by_event(&self.pool, community_id, event_id).await } /// Decrement reply counts. @@ -3299,13 +3114,8 @@ impl Db { parent_event_id: &[u8], root_event_id: Option<&[u8]>, ) -> Result<()> { - thread::decrement_reply_count( - self.pg_pool()?, - community_id, - parent_event_id, - root_event_id, - ) - .await + thread::decrement_reply_count(&self.pool, community_id, parent_event_id, root_event_id) + .await } /// Add (or re-activate) a reaction. @@ -3319,7 +3129,7 @@ impl Db { reaction_event_id: Option<&[u8]>, ) -> Result { reaction::add_reaction( - self.pg_pool()?, + &self.pool, community, event_id, event_created_at, @@ -3339,19 +3149,8 @@ impl Db { pubkey: &[u8], emoji: &str, ) -> Result { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::remove_reaction( - pool, - community, - event_id, - event_created_at, - pubkey, - emoji, - ) - .await; - } reaction::remove_reaction( - self.pg_pool()?, + &self.pool, community, event_id, event_created_at, @@ -3367,12 +3166,7 @@ impl Db { community: CommunityId, reaction_event_id: &[u8], ) -> Result { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::remove_reaction_by_source_event_id(pool, community, reaction_event_id) - .await; - } - reaction::remove_reaction_by_source_event_id(self.pg_pool()?, community, reaction_event_id) - .await + reaction::remove_reaction_by_source_event_id(&self.pool, community, reaction_event_id).await } /// Look up the active reaction row for one actor + emoji + target tuple. @@ -3385,7 +3179,7 @@ impl Db { emoji: &str, ) -> Result> { reaction::get_active_reaction_record( - self.pg_pool()?, + &self.pool, community, event_id, event_created_at, @@ -3406,7 +3200,7 @@ impl Db { reaction_event_id: &[u8], ) -> Result { reaction::set_reaction_event_id( - self.pg_pool()?, + &self.pool, community, event_id, event_created_at, @@ -3427,7 +3221,7 @@ impl Db { cursor: Option<&str>, ) -> Result> { reaction::get_reactions( - self.pg_pool()?, + &self.pool, community, event_id, event_created_at, @@ -3443,7 +3237,7 @@ impl Db { community: CommunityId, event_ids: &[(&[u8], DateTime)], ) -> Result> { - reaction::get_reactions_bulk(self.pg_pool()?, community, event_ids).await + reaction::get_reactions_bulk(&self.pool, community, event_ids).await } /// Find events that @mention the given pubkey. @@ -3456,7 +3250,7 @@ impl Db { limit: i64, ) -> Result> { feed::query_mentions( - self.pg_pool()?, + &self.pool, community, pubkey_bytes, accessible_channel_ids, @@ -3482,17 +3276,6 @@ impl Db { since: Option>, limit: i64, ) -> Result> { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::query_feed_mentions( - pool, - community, - pubkey_bytes, - accessible_channel_ids, - since, - limit, - ) - .await; - } match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { match feed::query_mentions_on( @@ -3513,7 +3296,7 @@ impl Db { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); feed::query_mentions( - self.pg_pool()?, + &self.pool, community, pubkey_bytes, accessible_channel_ids, @@ -3526,7 +3309,7 @@ impl Db { } RouteDecision::Writer => { feed::query_mentions( - self.pg_pool()?, + &self.pool, community, pubkey_bytes, accessible_channel_ids, @@ -3548,7 +3331,7 @@ impl Db { limit: i64, ) -> Result> { feed::query_needs_action( - self.pg_pool()?, + &self.pool, community, pubkey_bytes, accessible_channel_ids, @@ -3570,17 +3353,6 @@ impl Db { since: Option>, limit: i64, ) -> Result> { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::query_feed_needs_action( - pool, - community, - pubkey_bytes, - accessible_channel_ids, - since, - limit, - ) - .await; - } match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { match feed::query_needs_action_on( @@ -3601,7 +3373,7 @@ impl Db { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); feed::query_needs_action( - self.pg_pool()?, + &self.pool, community, pubkey_bytes, accessible_channel_ids, @@ -3614,7 +3386,7 @@ impl Db { } RouteDecision::Writer => { feed::query_needs_action( - self.pg_pool()?, + &self.pool, community, pubkey_bytes, accessible_channel_ids, @@ -3634,14 +3406,7 @@ impl Db { since: Option>, limit: i64, ) -> Result> { - feed::query_activity( - self.pg_pool()?, - community, - accessible_channel_ids, - since, - limit, - ) - .await + feed::query_activity(&self.pool, community, accessible_channel_ids, since, limit).await } /// [`Db::query_feed_activity`] with replica routing — BOUNDED arm only; @@ -3655,16 +3420,6 @@ impl Db { since: Option>, limit: i64, ) -> Result> { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::query_feed_activity( - pool, - community, - accessible_channel_ids, - since, - limit, - ) - .await; - } match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { match feed::query_activity_on( @@ -3684,7 +3439,7 @@ impl Db { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); feed::query_activity( - self.pg_pool()?, + &self.pool, community, accessible_channel_ids, since, @@ -3695,14 +3450,8 @@ impl Db { } } RouteDecision::Writer => { - feed::query_activity( - self.pg_pool()?, - community, - accessible_channel_ids, - since, - limit, - ) - .await + feed::query_activity(&self.pool, community, accessible_channel_ids, since, limit) + .await } } } @@ -3720,7 +3469,7 @@ impl Db { expires_at: Option>, ) -> Result { api_token::create_api_token( - self.pg_pool()?, + &self.pool, *community_id.as_uuid(), token_hash, owner_pubkey, @@ -3745,7 +3494,7 @@ impl Db { expires_at: Option>, ) -> Result> { api_token::create_api_token_if_under_limit( - self.pg_pool()?, + &self.pool, *community_id.as_uuid(), token_hash, owner_pubkey, @@ -3778,7 +3527,7 @@ impl Db { ) .bind(community_id.as_uuid()) .bind(hash) - .fetch_optional(self.pg_pool()?) + .fetch_optional(&self.pool) .await?; match row { @@ -3794,7 +3543,7 @@ impl Db { hash: &[u8], ) -> Result> { api_token::get_api_token_by_hash_including_revoked( - self.pg_pool()?, + &self.pool, *community_id.as_uuid(), hash, ) @@ -3808,7 +3557,7 @@ impl Db { ) .bind(community_id.as_uuid()) .bind(hash) - .execute(self.pg_pool()?) + .execute(&self.pool) .await?; Ok(()) } @@ -3834,7 +3583,7 @@ impl Db { "#, ) .bind(community_id.as_uuid()) - .fetch_all(self.pg_pool()?) + .fetch_all(&self.pool) .await?; let mut out = Vec::with_capacity(rows.len()); @@ -3862,7 +3611,7 @@ impl Db { community_id: CommunityId, pubkey: &[u8], ) -> Result> { - api_token::list_tokens_by_owner(self.pg_pool()?, *community_id.as_uuid(), pubkey).await + api_token::list_tokens_by_owner(&self.pool, *community_id.as_uuid(), pubkey).await } /// Revoke a single token by ID, scoped to (community, owner). @@ -3874,7 +3623,7 @@ impl Db { revoked_by: &[u8], ) -> Result { api_token::revoke_token( - self.pg_pool()?, + &self.pool, *community_id.as_uuid(), id, owner_pubkey, @@ -3891,7 +3640,7 @@ impl Db { revoked_by: &[u8], ) -> Result { api_token::revoke_all_tokens( - self.pg_pool()?, + &self.pool, *community_id.as_uuid(), owner_pubkey, revoked_by, @@ -3910,7 +3659,7 @@ impl Db { definition_hash: &[u8], ) -> Result { workflow::create_workflow( - self.pg_pool()?, + &self.pool, community_id, channel_id, owner_pubkey, @@ -3934,7 +3683,7 @@ impl Db { definition_hash: &[u8], ) -> Result<()> { workflow::upsert_workflow( - self.pg_pool()?, + &self.pool, community_id, id, channel_id, @@ -3952,7 +3701,7 @@ impl Db { community_id: CommunityId, id: Uuid, ) -> Result { - workflow::get_workflow(self.pg_pool()?, community_id, id).await + workflow::get_workflow(&self.pool, community_id, id).await } /// List workflows for a channel. @@ -3963,8 +3712,7 @@ impl Db { limit: Option, offset: Option, ) -> Result> { - workflow::list_channel_workflows(self.pg_pool()?, community_id, channel_id, limit, offset) - .await + workflow::list_channel_workflows(&self.pool, community_id, channel_id, limit, offset).await } /// List active, enabled workflows for a channel. @@ -3973,12 +3721,12 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result> { - workflow::list_enabled_channel_workflows(self.pg_pool()?, community_id, channel_id).await + workflow::list_enabled_channel_workflows(&self.pool, community_id, channel_id).await } /// List all active, enabled schedule-triggered workflows. pub async fn list_all_enabled_workflows(&self) -> Result> { - workflow::list_all_enabled_workflows(self.pg_pool()?).await + workflow::list_all_enabled_workflows(&self.pool).await } /// Claim a scheduled workflow fire for an authoritative schedule instant. @@ -3996,7 +3744,7 @@ impl Db { scheduled_for: chrono::DateTime, ) -> Result> { workflow::claim_scheduled_workflow_fire( - self.pg_pool()?, + &self.pool, community_id, workflow_id, scheduled_for, @@ -4010,7 +3758,7 @@ impl Db { community_id: CommunityId, workflow_id: Uuid, ) -> Result>> { - workflow::latest_scheduled_workflow_fire(self.pg_pool()?, community_id, workflow_id).await + workflow::latest_scheduled_workflow_fire(&self.pool, community_id, workflow_id).await } /// Attach the workflow run id created from a won scheduled-fire claim. @@ -4022,7 +3770,7 @@ impl Db { workflow_run_id: Uuid, ) -> Result { workflow::attach_scheduled_workflow_run( - self.pg_pool()?, + &self.pool, community_id, workflow_id, scheduled_for, @@ -4036,7 +3784,7 @@ impl Db { &self, older_than: chrono::DateTime, ) -> Result { - workflow::prune_scheduled_workflow_fires_before(self.pg_pool()?, older_than).await + workflow::prune_scheduled_workflow_fires_before(&self.pool, older_than).await } /// Update a workflow's name, definition, and hash. @@ -4049,7 +3797,7 @@ impl Db { definition_hash: &[u8], ) -> Result<()> { workflow::update_workflow( - self.pg_pool()?, + &self.pool, community_id, id, name, @@ -4066,7 +3814,7 @@ impl Db { id: Uuid, status: workflow::WorkflowStatus, ) -> Result<()> { - workflow::update_workflow_status(self.pg_pool()?, community_id, id, status).await + workflow::update_workflow_status(&self.pool, community_id, id, status).await } /// Enable or disable a workflow. @@ -4076,7 +3824,7 @@ impl Db { id: Uuid, enabled: bool, ) -> Result<()> { - workflow::set_workflow_enabled(self.pg_pool()?, community_id, id, enabled).await + workflow::set_workflow_enabled(&self.pool, community_id, id, enabled).await } /// Disable all of an owner's workflows in a channel (SEC-006, on @@ -4088,7 +3836,7 @@ impl Db { owner_pubkey: &[u8], ) -> Result { workflow::disable_workflows_for_owner_in_channel( - self.pg_pool()?, + &self.pool, community_id, channel_id, owner_pubkey, @@ -4098,7 +3846,7 @@ impl Db { /// Delete a workflow and all its runs/approvals. pub async fn delete_workflow(&self, community_id: CommunityId, id: Uuid) -> Result<()> { - workflow::delete_workflow(self.pg_pool()?, community_id, id).await + workflow::delete_workflow(&self.pool, community_id, id).await } /// Delete a workflow only when it belongs to the provided owner. @@ -4109,7 +3857,7 @@ impl Db { id: Uuid, owner_pubkey: &[u8], ) -> Result> { - workflow::delete_workflow_for_owner(self.pg_pool()?, community_id, id, owner_pubkey).await + workflow::delete_workflow_for_owner(&self.pool, community_id, id, owner_pubkey).await } /// Find a workflow by owner pubkey and name within a community. Used for @@ -4120,7 +3868,7 @@ impl Db { owner_pubkey: &[u8], name: &str, ) -> Result> { - workflow::find_by_owner_and_name(self.pg_pool()?, community_id, owner_pubkey, name).await + workflow::find_by_owner_and_name(&self.pool, community_id, owner_pubkey, name).await } /// Create a new workflow run. @@ -4132,7 +3880,7 @@ impl Db { trigger_context: Option<&serde_json::Value>, ) -> Result { workflow::create_workflow_run( - self.pg_pool()?, + &self.pool, community_id, workflow_id, trigger_event_id, @@ -4147,7 +3895,7 @@ impl Db { community_id: CommunityId, id: Uuid, ) -> Result { - workflow::get_workflow_run(self.pg_pool()?, community_id, id).await + workflow::get_workflow_run(&self.pool, community_id, id).await } /// List runs for a workflow. @@ -4157,7 +3905,7 @@ impl Db { workflow_id: Uuid, limit: i64, ) -> Result> { - workflow::list_workflow_runs(self.pg_pool()?, community_id, workflow_id, limit).await + workflow::list_workflow_runs(&self.pool, community_id, workflow_id, limit).await } /// Update a workflow run's status. @@ -4171,7 +3919,7 @@ impl Db { error: Option<&str>, ) -> Result<()> { workflow::update_workflow_run( - self.pg_pool()?, + &self.pool, community_id, id, status, @@ -4184,7 +3932,7 @@ impl Db { /// Create an approval request. pub async fn create_approval(&self, params: workflow::CreateApprovalParams<'_>) -> Result<()> { - workflow::create_approval(self.pg_pool()?, params).await + workflow::create_approval(&self.pool, params).await } /// Fetch an approval by raw token. @@ -4193,7 +3941,7 @@ impl Db { community_id: CommunityId, token: &str, ) -> Result { - workflow::get_approval(self.pg_pool()?, community_id, token).await + workflow::get_approval(&self.pool, community_id, token).await } /// Fetch an approval by its already-hashed token (no re-hashing). @@ -4202,7 +3950,7 @@ impl Db { community_id: CommunityId, token_hash: &[u8], ) -> Result { - workflow::get_approval_by_stored_hash(self.pg_pool()?, community_id, token_hash).await + workflow::get_approval_by_stored_hash(&self.pool, community_id, token_hash).await } /// Fetch all approvals for a workflow run. @@ -4212,7 +3960,7 @@ impl Db { workflow_id: uuid::Uuid, run_id: uuid::Uuid, ) -> Result> { - workflow::get_run_approvals(self.pg_pool()?, community_id, workflow_id, run_id).await + workflow::get_run_approvals(&self.pool, community_id, workflow_id, run_id).await } /// Update an approval's status. @@ -4225,7 +3973,7 @@ impl Db { note: Option<&str>, ) -> Result { workflow::update_approval( - self.pg_pool()?, + &self.pool, community_id, token, status, @@ -4245,7 +3993,7 @@ impl Db { note: Option<&str>, ) -> Result { workflow::update_approval_by_stored_hash( - self.pg_pool()?, + &self.pool, community_id, token_hash, status, @@ -4257,7 +4005,7 @@ impl Db { /// Ensures monthly partitions exist for the next N months. pub async fn ensure_future_partitions(&self, months_ahead: u32) -> Result<()> { - partition::ensure_future_partitions(self.pg_pool()?, months_ahead).await + partition::ensure_future_partitions(&self.pool, months_ahead).await } /// Backfill `d_tag` for existing NIP-33 events (kind 30000–39999) that have `d_tag IS NULL`. @@ -4274,7 +4022,7 @@ impl Db { ) \ WHERE kind BETWEEN 30000 AND 39999 AND d_tag IS NULL", ) - .execute(self.pg_pool()?) + .execute(&self.pool) .await?; Ok(result.rows_affected()) } @@ -4286,7 +4034,7 @@ impl Db { ) .bind(community.as_uuid()) .bind(pubkey) - .fetch_one(self.pg_pool()?) + .fetch_one(&self.pool) .await?; let cnt: i64 = row.try_get("cnt")?; Ok(cnt > 0) @@ -4297,7 +4045,7 @@ impl Db { let row = sqlx::query("SELECT COUNT(*) as cnt FROM pubkey_allowlist WHERE community_id = $1") .bind(community.as_uuid()) - .fetch_one(self.pg_pool()?) + .fetch_one(&self.pool) .await?; let cnt: i64 = row.try_get("cnt")?; Ok(cnt > 0) @@ -4319,7 +4067,7 @@ impl Db { .bind(pubkey) .bind(added_by) .bind(note) - .execute(self.pg_pool()?) + .execute(&self.pool) .await?; Ok(result.rows_affected() > 0) } @@ -4334,7 +4082,7 @@ impl Db { sqlx::query("DELETE FROM pubkey_allowlist WHERE community_id = $1 AND pubkey = $2") .bind(community.as_uuid()) .bind(pubkey) - .execute(self.pg_pool()?) + .execute(&self.pool) .await?; Ok(result.rows_affected() > 0) } @@ -4345,7 +4093,7 @@ impl Db { "SELECT pubkey, added_by, added_at, note FROM pubkey_allowlist WHERE community_id = $1 ORDER BY added_at DESC", ) .bind(community.as_uuid()) - .fetch_all(self.pg_pool()?) + .fetch_all(&self.pool) .await?; let mut out = Vec::with_capacity(rows.len()); @@ -4369,33 +4117,27 @@ 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 { - match &self.backend { - DbBackend::SQLite(pool) => sqlite::is_relay_member(pool, community, pubkey).await, - DbBackend::Postgres => { - let path = "relay_membership"; - match self.route_read(path, RoutePredicate::Bounded).await { - RouteDecision::Replica(mut tx, _entry, reason) => { - match relay_members::is_relay_member_on(&mut tx, community, pubkey).await { - Ok(is_member) => { - Self::record_route(path, "replica", reason); - Ok(is_member) - } - Err(e) => { - tracing::warn!( - path, - "replica read failed; re-running on writer: {e}" - ); - Self::record_route(path, "writer", "replica_error"); - relay_members::is_relay_member(self.pg_pool()?, community, pubkey) - .await - } - } + 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) => { + match relay_members::is_relay_member_on(&mut tx, community, pubkey).await { + Ok(is_member) => { + Self::record_route(path, "replica", reason); + Ok(is_member) } - RouteDecision::Writer => { - relay_members::is_relay_member(self.pg_pool()?, community, pubkey).await + Err(e) => { + tracing::warn!(path, "replica read failed; re-running on writer: {e}"); + Self::record_route(path, "writer", "replica_error"); + relay_members::is_relay_member(&self.pool, community, pubkey).await } } } + RouteDecision::Writer => { + relay_members::is_relay_member(&self.pool, community, pubkey).await + } } } @@ -4405,12 +4147,10 @@ impl Db { community: CommunityId, pubkey: &str, ) -> Result> { - match &self.backend { - DbBackend::SQLite(pool) => sqlite::get_relay_member(pool, community, pubkey).await, - DbBackend::Postgres => { - relay_members::get_relay_member(self.pg_pool()?, community, pubkey).await - } + 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 } /// Returns all relay members of `community` ordered by `created_at` ascending. @@ -4418,12 +4158,10 @@ impl Db { &self, community: CommunityId, ) -> Result> { - match &self.backend { - DbBackend::SQLite(pool) => sqlite::list_relay_members(pool, community).await, - DbBackend::Postgres => { - relay_members::list_relay_members(self.pg_pool()?, community).await - } + if let DbBackend::SQLite(pool) = &self.backend { + return sqlite::list_relay_members(pool, community).await; } + relay_members::list_relay_members(&self.pool, community).await } /// Adds a new relay member to `community`. @@ -4437,15 +4175,10 @@ impl Db { role: &str, added_by: Option<&str>, ) -> Result { - match &self.backend { - DbBackend::SQLite(pool) => { - sqlite::add_relay_member(pool, community, pubkey, role, added_by).await - } - DbBackend::Postgres => { - relay_members::add_relay_member(self.pg_pool()?, community, pubkey, role, added_by) - .await - } + 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 } /// Claims relay membership via an invite and atomically persists the @@ -4457,14 +4190,8 @@ impl Db { role: &str, policy_version: Option<&str>, ) -> Result { - relay_members::claim_relay_membership( - self.pg_pool()?, - community, - pubkey, - role, - policy_version, - ) - .await + relay_members::claim_relay_membership(&self.pool, community, pubkey, role, policy_version) + .await } /// Returns whether a member has persisted acceptance evidence for a policy version. @@ -4474,13 +4201,8 @@ impl Db { pubkey: &str, policy_version: &str, ) -> Result { - relay_members::has_join_policy_acceptance( - self.pg_pool()?, - community, - pubkey, - policy_version, - ) - .await + relay_members::has_join_policy_acceptance(&self.pool, community, pubkey, policy_version) + .await } /// Removes a relay member from `community` atomically, refusing to delete the owner. @@ -4489,7 +4211,7 @@ impl Db { community: CommunityId, pubkey: &str, ) -> Result { - relay_members::remove_relay_member(self.pg_pool()?, community, pubkey).await + relay_members::remove_relay_member(&self.pool, community, pubkey).await } /// Removes a relay member from `community` only if their current role matches `expected_role`. @@ -4502,13 +4224,8 @@ impl Db { pubkey: &str, expected_role: &str, ) -> Result { - relay_members::remove_relay_member_if_role( - self.pg_pool()?, - community, - pubkey, - expected_role, - ) - .await + relay_members::remove_relay_member_if_role(&self.pool, community, pubkey, expected_role) + .await } /// Updates the role of an existing relay member in `community`. Returns `true` if updated. @@ -4518,17 +4235,15 @@ impl Db { pubkey: &str, new_role: &str, ) -> Result { - relay_members::update_relay_member_role(self.pg_pool()?, community, pubkey, new_role).await + relay_members::update_relay_member_role(&self.pool, community, pubkey, new_role).await } /// 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<()> { - match &self.backend { - DbBackend::SQLite(pool) => sqlite::bootstrap_owner(pool, community, owner_pubkey).await, - DbBackend::Postgres => { - relay_members::bootstrap_owner(self.pg_pool()?, community, owner_pubkey).await - } + 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 } /// Returns `true` if any member of `community` holds the `admin` or @@ -4548,7 +4263,7 @@ impl Db { expected_owner_pubkey: &str, ) -> Result { relay_members::transfer_ownership( - self.pg_pool()?, + &self.pool, community, new_owner_pubkey, expected_owner_pubkey, @@ -4561,7 +4276,7 @@ impl Db { /// Idempotent — uses `ON CONFLICT DO NOTHING`. Returns the number of rows /// inserted, or 0 if the `pubkey_allowlist` table doesn't exist. pub async fn backfill_from_allowlist(&self, community: CommunityId) -> Result { - relay_members::backfill_from_allowlist(self.pg_pool()?, community).await + relay_members::backfill_from_allowlist(&self.pool, community).await } /// Mints a v2 use-limited relay invite. The plaintext code is returned @@ -4576,8 +4291,7 @@ impl Db { ttl_secs: u64, max_uses: Option, ) -> Result { - relay_invite::mint_relay_invite(self.pg_pool()?, community, created_by, ttl_secs, max_uses) - .await + relay_invite::mint_relay_invite(&self.pool, community, created_by, ttl_secs, max_uses).await } /// Delete one bounded batch of invites expired before `cutoff`. @@ -4585,7 +4299,7 @@ impl Db { &self, cutoff: chrono::DateTime, ) -> Result { - relay_invite::reap_expired_relay_invites(self.pg_pool()?, cutoff).await + relay_invite::reap_expired_relay_invites(&self.pool, cutoff).await } /// Atomically claims a v2 relay invite. The full redemption (membership @@ -4601,7 +4315,7 @@ impl Db { policy_version: Option<&str>, ) -> Result { relay_invite::claim_relay_invite( - self.pg_pool()?, + &self.pool, community, token_hash, claimer_pubkey, @@ -4616,7 +4330,7 @@ impl Db { community: CommunityId, feedback: product_feedback::NewProductFeedback<'_>, ) -> Result { - product_feedback::insert(self.pg_pool()?, community, feedback).await + product_feedback::insert(&self.pool, community, feedback).await } /// List product feedback across the deployment, newest first. @@ -4624,7 +4338,7 @@ impl Db { &self, limit: i64, ) -> Result> { - product_feedback::list(self.pg_pool()?, limit).await + product_feedback::list(&self.pool, limit).await } /// Insert a tenant-scoped NIP-56 report row, idempotent by report event id. @@ -4633,7 +4347,7 @@ impl Db { community: CommunityId, report: moderation::NewReport<'_>, ) -> Result { - moderation::insert_report(self.pg_pool()?, community, report).await + moderation::insert_report(&self.pool, community, report).await } /// List moderation reports for a community, newest first. @@ -4643,7 +4357,7 @@ impl Db { status: Option<&str>, limit: i64, ) -> Result> { - moderation::list_reports(self.pg_pool()?, community, status, limit).await + moderation::list_reports(&self.pool, community, status, limit).await } /// Fetch one moderation report by row id. @@ -4652,7 +4366,7 @@ impl Db { community: CommunityId, report_id: Uuid, ) -> Result> { - moderation::get_report(self.pg_pool()?, community, report_id).await + moderation::get_report(&self.pool, community, report_id).await } /// Fetch one moderation report by signed NIP-56 report event id. @@ -4661,7 +4375,7 @@ impl Db { community: CommunityId, report_event_id: &[u8], ) -> Result> { - moderation::get_report_by_event(self.pg_pool()?, community, report_event_id).await + moderation::get_report_by_event(&self.pool, community, report_event_id).await } /// Resolve, dismiss, or escalate an open moderation report. @@ -4674,7 +4388,7 @@ impl Db { action_id: Option, ) -> Result { moderation::resolve_report( - self.pg_pool()?, + &self.pool, community, report_id, status, @@ -4693,15 +4407,7 @@ impl Db { reason: Option<&str>, expires_at: Option>, ) -> Result<()> { - moderation::ban_member( - self.pg_pool()?, - community, - pubkey, - actor, - reason, - expires_at, - ) - .await + moderation::ban_member(&self.pool, community, pubkey, actor, reason, expires_at).await } /// Lift a community ban for a member pubkey. @@ -4711,7 +4417,7 @@ impl Db { pubkey: &[u8], actor: &[u8], ) -> Result { - moderation::unban_member(self.pg_pool()?, community, pubkey, actor).await + moderation::unban_member(&self.pool, community, pubkey, actor).await } /// Upsert a community timeout/write-block for a member pubkey. @@ -4723,15 +4429,7 @@ impl Db { muted_until: DateTime, reason: Option<&str>, ) -> Result<()> { - moderation::timeout_member( - self.pg_pool()?, - community, - pubkey, - actor, - muted_until, - reason, - ) - .await + moderation::timeout_member(&self.pool, community, pubkey, actor, muted_until, reason).await } /// Clear a community timeout/write-block for a member pubkey. @@ -4741,7 +4439,7 @@ impl Db { pubkey: &[u8], actor: &[u8], ) -> Result { - moderation::untimeout_member(self.pg_pool()?, community, pubkey, actor).await + moderation::untimeout_member(&self.pool, community, pubkey, actor).await } /// Fetch the active ban/timeout restriction state for enforcement hot paths. @@ -4750,12 +4448,7 @@ impl Db { community: CommunityId, pubkey: &[u8], ) -> Result { - match &self.backend { - DbBackend::SQLite(_) => Ok(moderation::RestrictionState::default()), - DbBackend::Postgres => { - moderation::restriction_state(self.pg_pool()?, community, pubkey).await - } - } + moderation::restriction_state(&self.pool, community, pubkey).await } /// Fetch the full ban/timeout row for a member pubkey. @@ -4764,7 +4457,7 @@ impl Db { community: CommunityId, pubkey: &[u8], ) -> Result> { - moderation::get_ban(self.pg_pool()?, community, pubkey).await + moderation::get_ban(&self.pool, community, pubkey).await } /// List currently restricted members in a community. @@ -4772,7 +4465,7 @@ impl Db { &self, community: CommunityId, ) -> Result> { - moderation::list_restricted(self.pg_pool()?, community).await + moderation::list_restricted(&self.pool, community).await } /// Insert a moderation audit action row. @@ -4781,7 +4474,7 @@ impl Db { community: CommunityId, action: moderation::NewAction<'_>, ) -> Result { - moderation::insert_action(self.pg_pool()?, community, action).await + moderation::insert_action(&self.pool, community, action).await } /// List moderation audit action rows, newest first. @@ -4790,7 +4483,7 @@ impl Db { community: CommunityId, limit: i64, ) -> Result> { - moderation::list_actions(self.pg_pool()?, community, limit).await + moderation::list_actions(&self.pool, community, limit).await } /// Return the current owner of git repo name `repo_id` in `community`, or @@ -4800,7 +4493,7 @@ impl Db { community: CommunityId, repo_id: &str, ) -> Result> { - git_repo::repo_name_owner(self.pg_pool()?, community, repo_id).await + git_repo::repo_name_owner(&self.pool, community, repo_id).await } /// Reserve a git repo name for `owner_pubkey` in `community` (NIP-34). @@ -4813,7 +4506,7 @@ impl Db { repo_id: &str, owner_pubkey: &str, ) -> Result { - git_repo::reserve_repo_name(self.pg_pool()?, community, repo_id, owner_pubkey).await + git_repo::reserve_repo_name(&self.pool, community, repo_id, owner_pubkey).await } /// Count git repos reserved by `owner_pubkey` in `community` (quota check). @@ -4822,7 +4515,7 @@ impl Db { community: CommunityId, owner_pubkey: &str, ) -> Result { - git_repo::count_repos_for_owner(self.pg_pool()?, community, owner_pubkey).await + git_repo::count_repos_for_owner(&self.pool, community, owner_pubkey).await } /// Release a git repo name reservation held by `owner_pubkey` (rollback). @@ -4834,12 +4527,12 @@ impl Db { repo_id: &str, owner_pubkey: &str, ) -> Result { - git_repo::release_repo_name(self.pg_pool()?, community, repo_id, owner_pubkey).await + git_repo::release_repo_name(&self.pool, community, repo_id, owner_pubkey).await } /// Returns `true` if `pubkey` (64-char hex) is archived in `community_id`. pub async fn is_archived(&self, community_id: CommunityId, pubkey: &str) -> Result { - archived_identities::is_archived(self.pg_pool()?, community_id, pubkey).await + archived_identities::is_archived(&self.pool, community_id, pubkey).await } /// Archives an identity in `community_id`. Returns `true` if inserted, `false` if already archived. @@ -4855,7 +4548,7 @@ impl Db { request_event_id: &str, ) -> Result { archived_identities::archive( - self.pg_pool()?, + &self.pool, community_id, pubkey, consent_path, @@ -4869,7 +4562,7 @@ impl Db { /// Unarchives an identity from `community_id`. Returns `true` if deleted, `false` if absent. pub async fn unarchive(&self, community_id: CommunityId, pubkey: &str) -> Result { - archived_identities::unarchive(self.pg_pool()?, community_id, pubkey).await + archived_identities::unarchive(&self.pool, community_id, pubkey).await } /// Returns all identities archived in `community_id`, ordered by archive time ascending. @@ -4877,7 +4570,7 @@ impl Db { &self, community_id: CommunityId, ) -> Result> { - archived_identities::list_archived(self.pg_pool()?, community_id).await + archived_identities::list_archived(&self.pool, community_id).await } /// Soft-delete NIP-29 discovery events for a channel created by a specific relay pubkey. @@ -4894,7 +4587,7 @@ impl Db { .bind(community_id.as_uuid()) .bind(channel_id) .bind(relay_pubkey) - .execute(self.pg_pool()?) + .execute(&self.pool) .await?; Ok(result.rows_affected()) } @@ -4912,9 +4605,6 @@ impl Db { event: &nostr::Event, channel_id: Option, ) -> Result<(StoredEvent, bool)> { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::replace_event(pool, community_id, event, channel_id, None).await; - } let kind_i32 = buzz_core::kind::event_kind_i32(event); let pubkey_bytes = event.pubkey.to_bytes(); let created_at_secs = event.created_at.as_secs() as i64; @@ -4929,7 +4619,7 @@ impl Db { channel_id.as_ref().map(|id| id.as_bytes().as_slice()), ); - let mut tx = self.pg_pool()?.begin().await?; + let mut tx = self.pool.begin().await?; // Serialize all writers for the same (kind, pubkey, channel_id) tuple. // Advisory lock is transaction-scoped — released on commit/rollback. @@ -5025,9 +4715,7 @@ impl Db { // Mentions are a denormalized index — safe outside the transaction. // insert_event() normally handles this, but we inlined the INSERT above. - if let Err(e) = - crate::insert_mentions(self.pg_pool()?, community_id, event, channel_id).await - { + if let Err(e) = crate::insert_mentions(&self.pool, community_id, event, channel_id).await { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } @@ -5106,7 +4794,7 @@ impl Db { let lock_key = event_replacement_lock_key(community_id, kind_i32, pubkey_bytes.as_slice(), None); - let mut tx = self.pg_pool()?.begin().await?; + let mut tx = self.pool.begin().await?; // Acquire the per-community snapshot lock BEFORE reading members. // This serializes the entire read-build-write cycle: a concurrent @@ -5201,7 +4889,7 @@ impl Db { tx.commit().await?; - if let Err(e) = crate::insert_mentions(self.pg_pool()?, community_id, &event, None).await { + if let Err(e) = crate::insert_mentions(&self.pool, community_id, &event, None).await { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } @@ -5240,9 +4928,6 @@ impl Db { d_tag: &str, channel_id: Option, ) -> Result<(StoredEvent, bool)> { - if let DbBackend::SQLite(pool) = &self.backend { - return sqlite::replace_event(pool, community_id, event, channel_id, Some(d_tag)).await; - } let kind_i32 = buzz_core::kind::event_kind_i32(event); let pubkey_bytes = event.pubkey.to_bytes(); let created_at_secs = event.created_at.as_secs() as i64; @@ -5256,7 +4941,7 @@ impl Db { Some(d_tag.as_bytes()), ); - let mut tx = self.pg_pool()?.begin().await?; + let mut tx = self.pool.begin().await?; sqlx::query("SELECT pg_advisory_xact_lock($1)") .bind(lock_key) @@ -5441,9 +5126,7 @@ impl Db { tx.commit().await?; // Mentions are a denormalized index — safe outside the transaction. - if let Err(e) = - crate::insert_mentions(self.pg_pool()?, community_id, event, channel_id).await - { + if let Err(e) = crate::insert_mentions(&self.pool, community_id, event, channel_id).await { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } @@ -7059,7 +6742,7 @@ mod tests { let db = Db::from_pool(pool); assert!(!db.has_read_pool()); assert!( - std::ptr::eq(db.read().expect("Postgres test DB"), &db.pool), + std::ptr::eq(db.read(), &db.pool), "read() must be the writer pool when no replica is configured" ); assert!(db.read_pool_stats().is_none()); diff --git a/crates/buzz-db/src/sqlite.rs b/crates/buzz-db/src/sqlite.rs index c3e4dfdc8f..2c2e6f6358 100644 --- a/crates/buzz-db/src/sqlite.rs +++ b/crates/buzz-db/src/sqlite.rs @@ -1,7 +1,7 @@ -//! SQLite storage for the single-node relay profile. +//! SQLite storage for the single-node community profile. //! -//! This module deliberately contains a separate, minimal schema rather than -//! attempting to translate the production PostgreSQL migrations. +//! 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}; @@ -32,21 +32,6 @@ CREATE TABLE IF NOT EXISTS relay_members ( FOREIGN KEY (community_id) REFERENCES communities(id) ON DELETE CASCADE ); -CREATE TABLE IF NOT EXISTS users ( - community_id TEXT NOT NULL, - pubkey BLOB NOT NULL, - display_name TEXT, - avatar_url TEXT, - about TEXT, - nip05_handle TEXT, - channel_add_policy TEXT NOT NULL DEFAULT 'anyone', - is_agent INTEGER NOT NULL DEFAULT 0, - agent_owner_pubkey BLOB, - created_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, @@ -86,52 +71,6 @@ CREATE TABLE IF NOT EXISTS channel_members ( PRIMARY KEY (channel_id, pubkey), FOREIGN KEY (channel_id) REFERENCES channels(id) ON DELETE CASCADE ); - -CREATE TABLE IF NOT EXISTS moderation_restrictions ( - community_id TEXT NOT NULL, - pubkey TEXT NOT NULL COLLATE NOCASE, - restriction_type TEXT NOT NULL, - expires_at INTEGER, - PRIMARY KEY (community_id, pubkey, restriction_type), - FOREIGN KEY (community_id) REFERENCES communities(id) ON DELETE CASCADE -); - -CREATE TABLE IF NOT EXISTS events ( - community_id TEXT NOT NULL, - id BLOB NOT NULL, - pubkey BLOB NOT NULL, - created_at INTEGER NOT NULL, - kind INTEGER NOT NULL, - tags_json TEXT NOT NULL, - content TEXT NOT NULL, - sig BLOB NOT NULL, - channel_id TEXT, - received_at INTEGER NOT NULL, - event_json TEXT NOT NULL, - PRIMARY KEY (community_id, id), - FOREIGN KEY (community_id) REFERENCES communities(id) ON DELETE CASCADE -); -CREATE INDEX IF NOT EXISTS idx_events_community_created - ON events (community_id, created_at DESC, id); -CREATE INDEX IF NOT EXISTS idx_events_community_kind - ON events (community_id, kind, created_at DESC); -CREATE INDEX IF NOT EXISTS idx_events_channel_created - ON events (community_id, channel_id, created_at DESC); - -CREATE TABLE IF NOT EXISTS reactions ( - community_id TEXT NOT NULL, - event_created_at INTEGER NOT NULL, - event_id BLOB NOT NULL, - pubkey BLOB NOT NULL, - emoji TEXT NOT NULL, - reaction_event_id BLOB, - removed_at INTEGER, - PRIMARY KEY (community_id, event_created_at, event_id, pubkey, emoji), - FOREIGN KEY (community_id) REFERENCES communities(id) ON DELETE CASCADE -); -CREATE UNIQUE INDEX IF NOT EXISTS idx_reactions_source_event - ON reactions (community_id, reaction_event_id) - WHERE reaction_event_id IS NOT NULL; "#; pub(crate) async fn connect(path_or_url: &str) -> Result { @@ -152,78 +91,9 @@ pub(crate) async fn connect(path_or_url: &str) -> Result { } pub(crate) async fn migrate(pool: &SqlitePool) -> Result<()> { - let had_application_schema = sqlx::query_scalar::<_, i64>( - "SELECT count(*) FROM sqlite_master WHERE type = 'table' AND name = 'channels'", - ) - .fetch_one(pool) - .await? - != 0; - for statement in SCHEMA.split(';').map(str::trim).filter(|s| !s.is_empty()) { sqlx::query(statement).execute(pool).await?; } - sqlx::query( - "CREATE TABLE IF NOT EXISTS schema_version (singleton INTEGER PRIMARY KEY CHECK (singleton = 1), version INTEGER NOT NULL)", - ) - .execute(pool) - .await?; - let initial_version = if had_application_schema { 1_i64 } else { 2_i64 }; - sqlx::query( - "INSERT INTO schema_version (singleton, version) VALUES (1, ?1) ON CONFLICT(singleton) DO NOTHING", - ) - .bind(initial_version) - .execute(pool) - .await?; - - let mut version = - sqlx::query_scalar::<_, i64>("SELECT version FROM schema_version WHERE singleton = 1") - .fetch_one(pool) - .await?; - if version < 2 { - let mut tx = pool.begin().await?; - ensure_column_on( - &mut tx, - "channels", - "participant_hash", - "ALTER TABLE channels ADD COLUMN participant_hash BLOB", - ) - .await?; - ensure_column_on( - &mut tx, - "channel_members", - "hidden_at", - "ALTER TABLE channel_members ADD COLUMN hidden_at INTEGER", - ) - .await?; - sqlx::query("UPDATE schema_version SET version = 2 WHERE singleton = 1") - .execute(&mut *tx) - .await?; - tx.commit().await?; - version = 2; - } - if version != 2 { - return Err(crate::DbError::InvalidData(format!( - "unsupported SQLite schema version {version}" - ))); - } - Ok(()) -} - -async fn ensure_column_on( - tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, - table: &str, - column: &str, - alter_sql: &'static str, -) -> Result<()> { - let pragma = format!("PRAGMA table_info({table})"); - let exists = sqlx::query(sqlx::AssertSqlSafe(pragma)) - .fetch_all(&mut **tx) - .await? - .iter() - .any(|row| row.get::("name") == column); - if !exists { - sqlx::query(alter_sql).execute(&mut **tx).await?; - } Ok(()) } @@ -274,8 +144,8 @@ pub(crate) async fn rebind_single_node_community_host( // Defect-era databases may contain several loopback communities, one per // ephemeral port. Prefer a community owned by this desktop identity, then - // the one carrying the most events. This preserves the most useful UUID and - // its channels/messages while making the current Host header resolvable. + // 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 @@ -283,11 +153,9 @@ pub(crate) async fn rebind_single_node_community_host( ON rm.community_id = c.id AND lower(rm.pubkey) = lower(?1) AND rm.role = 'owner' - LEFT JOIN events e ON e.community_id = c.id WHERE c.archived_at IS NULL AND c.host LIKE '127.0.0.1:%' - GROUP BY c.id, c.host, c.created_at - ORDER BY (rm.pubkey IS NOT NULL) DESC, count(e.id) DESC, c.created_at ASC, c.id ASC + 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) @@ -369,487 +237,6 @@ pub(crate) async fn ensure_configured_community( created: inserted, }) } - -pub(crate) async fn insert_event( - pool: &SqlitePool, - community: CommunityId, - event: &nostr::Event, - channel_id: Option, -) -> Result<(buzz_core::StoredEvent, bool)> { - let kind = u32::from(event.kind.as_u16()); - if kind == buzz_core::kind::KIND_AUTH { - return Err(crate::DbError::AuthEventRejected); - } - if buzz_core::kind::is_ephemeral(kind) { - return Err(crate::DbError::EphemeralEventRejected(event.kind.as_u16())); - } - let received_at = chrono::Utc::now(); - let inserted = sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags_json, content, sig, channel_id, received_at, event_json) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) ON CONFLICT DO NOTHING") - .bind(community.as_uuid().to_string()).bind(event.id.as_bytes().as_slice()).bind(event.pubkey.to_bytes().as_slice()) - .bind(event.created_at.as_secs() as i64).bind(event.kind.as_u16() as i32).bind(serde_json::to_string(&event.tags)?) - .bind(&event.content).bind(event.sig.serialize().as_slice()).bind(channel_id.map(|id| id.to_string())) - .bind(received_at.timestamp()).bind(serde_json::to_string(event)?).execute(pool).await?.rows_affected() == 1; - Ok(( - buzz_core::StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), - inserted, - )) -} - -pub(crate) async fn replace_event( - pool: &SqlitePool, - community: CommunityId, - event: &nostr::Event, - channel_id: Option, - d_tag: Option<&str>, -) -> Result<(buzz_core::StoredEvent, bool)> { - let mut tx = pool.begin().await?; - let rows = sqlx::query("SELECT id, event_json FROM events WHERE community_id = ?1 AND kind = ?2 AND pubkey = ?3 AND channel_id IS ?4") - .bind(community.as_uuid().to_string()) - .bind(event.kind.as_u16() as i32) - .bind(event.pubkey.to_bytes().as_slice()) - .bind(channel_id.map(|id| id.to_string())) - .fetch_all(&mut *tx) - .await?; - let mut replaced_ids = Vec::new(); - for row in rows { - let existing: nostr::Event = serde_json::from_str(row.try_get("event_json")?)?; - let existing_d = crate::event::extract_d_tag(&existing).unwrap_or_default(); - if d_tag.is_some_and(|expected| existing_d != expected) { - continue; - } - if event.created_at < existing.created_at - || (event.created_at == existing.created_at && event.id <= existing.id) - { - return Ok(( - buzz_core::StoredEvent::with_received_at( - existing, - chrono::Utc::now(), - channel_id, - true, - ), - false, - )); - } - replaced_ids.push(row.try_get::, _>("id")?); - } - for id in replaced_ids { - sqlx::query("DELETE FROM events WHERE community_id = ?1 AND id = ?2") - .bind(community.as_uuid().to_string()) - .bind(id) - .execute(&mut *tx) - .await?; - } - let received_at = chrono::Utc::now(); - sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags_json, content, sig, channel_id, received_at, event_json) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)") - .bind(community.as_uuid().to_string()).bind(event.id.as_bytes().as_slice()).bind(event.pubkey.to_bytes().as_slice()) - .bind(event.created_at.as_secs() as i64).bind(event.kind.as_u16() as i32).bind(serde_json::to_string(&event.tags)?) - .bind(&event.content).bind(event.sig.serialize().as_slice()).bind(channel_id.map(|id| id.to_string())) - .bind(received_at.timestamp()).bind(serde_json::to_string(event)?).execute(&mut *tx).await?; - tx.commit().await?; - Ok(( - buzz_core::StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), - true, - )) -} - -pub(crate) async fn soft_delete_event( - pool: &SqlitePool, - community: CommunityId, - event_id: &[u8], -) -> Result { - Ok( - sqlx::query("DELETE FROM events WHERE community_id = ?1 AND id = ?2") - .bind(community.as_uuid().to_string()) - .bind(event_id) - .execute(pool) - .await? - .rows_affected() - != 0, - ) -} - -pub(crate) async fn get_event_by_id( - pool: &SqlitePool, - community: CommunityId, - id: &[u8], -) -> Result> { - let row = sqlx::query("SELECT event_json, received_at, channel_id FROM events WHERE community_id = ?1 AND id = ?2") - .bind(community.as_uuid().to_string()).bind(id).fetch_optional(pool).await?; - row.map(stored_event).transpose() -} - -pub(crate) async fn query_events( - pool: &SqlitePool, - q: &crate::EventQuery, -) -> Result> { - if q.before_id.is_some() && q.until.is_none() { - return Err(crate::DbError::InvalidData( - "before_id requires until to be set".into(), - )); - } - if q.global_only && q.channel_id.is_some() { - return Err(crate::DbError::InvalidData( - "global_only and channel_id are mutually exclusive".into(), - )); - } - if q.kinds.as_ref().is_some_and(Vec::is_empty) - || q.authors.as_ref().is_some_and(Vec::is_empty) - || q.ids.as_ref().is_some_and(Vec::is_empty) - || q.e_tags.as_ref().is_some_and(Vec::is_empty) - { - return Ok(vec![]); - } - let rows = sqlx::query("SELECT event_json, received_at, channel_id FROM events WHERE community_id = ?1 ORDER BY created_at DESC, id ASC") - .bind(q.community_id.as_uuid().to_string()).fetch_all(pool).await?; - let mut events = Vec::new(); - for row in rows { - let stored = stored_event(row)?; - let event = &stored.event; - let created = event.created_at.as_secs() as i64; - let id = event.id.as_bytes().as_slice(); - let tags: Vec> = event - .tags - .iter() - .map(|tag| tag.as_slice().to_vec()) - .collect(); - let has_tag = |name: &str, value: &str| { - tags.iter().any(|tag| { - tag.first().is_some_and(|v| v == name) && tag.get(1).is_some_and(|v| v == value) - }) - }; - if q.channel_id.is_some_and(|ch| stored.channel_id != Some(ch)) - || (q.global_only && stored.channel_id.is_some()) - { - continue; - } - if q.channel_ids - .as_ref() - .is_some_and(|ids| stored.channel_id.is_some_and(|id| !ids.contains(&id))) - { - continue; - } - if q.kinds - .as_ref() - .is_some_and(|ks| !ks.contains(&(event.kind.as_u16() as i32))) - { - continue; - } - if q.pubkey - .as_ref() - .is_some_and(|pk| pk.as_slice() != event.pubkey.to_bytes().as_slice()) - { - continue; - } - if q.authors.as_ref().is_some_and(|authors| { - !authors - .iter() - .any(|pk| pk.as_slice() == event.pubkey.to_bytes().as_slice()) - }) { - continue; - } - if q.ids - .as_ref() - .is_some_and(|ids| !ids.iter().any(|candidate| candidate.as_slice() == id)) - { - continue; - } - if q.since.is_some_and(|since| created < since.timestamp()) - || q.until.is_some_and(|until| created > until.timestamp()) - { - continue; - } - if q.before_id.as_ref().is_some_and(|before| { - q.until.is_some_and(|until| created == until.timestamp()) && id <= before.as_slice() - }) { - continue; - } - if q.p_tag_hex - .as_ref() - .is_some_and(|p| !has_tag("p", &p.to_ascii_lowercase())) - { - continue; - } - if q.e_tags - .as_ref() - .is_some_and(|values| !values.iter().any(|value| has_tag("e", value))) - { - continue; - } - let d_tag = tags - .iter() - .find(|tag| tag.first().is_some_and(|v| v == "d")) - .and_then(|tag| tag.get(1)); - if q.d_tag.as_ref().is_some_and(|d| d_tag != Some(d)) { - continue; - } - if q.d_tags - .as_ref() - .is_some_and(|ds| !d_tag.is_some_and(|d| ds.contains(d))) - { - continue; - } - if q.shared_gated_reader.as_ref().is_some_and(|reader| { - buzz_core::kind::SHARED_GATED_KINDS.contains(&(event.kind.as_u16() as u32)) - && reader.as_slice() != event.pubkey.to_bytes().as_slice() - && !has_tag("shared", "true") - }) { - continue; - } - events.push(stored); - } - let offset = q.offset.unwrap_or(0).max(0) as usize; - let limit = q - .limit - .unwrap_or(100) - .min(q.max_limit.unwrap_or(crate::DEFAULT_MAX_PAGE_LIMIT)) - .max(0) as usize; - Ok(events.into_iter().skip(offset).take(limit).collect()) -} - -pub(crate) async fn query_feed_mentions( - pool: &SqlitePool, - community: CommunityId, - pubkey_bytes: &[u8], - accessible_channel_ids: &[Uuid], - since: Option>, - limit: i64, -) -> Result> { - let mut query = crate::EventQuery::for_community(community); - query.kinds = Some(vec![ - buzz_core::kind::KIND_STREAM_MESSAGE as i32, - buzz_core::kind::KIND_STREAM_MESSAGE_V2 as i32, - buzz_core::kind::KIND_TEXT_NOTE as i32, - buzz_core::kind::KIND_FORUM_POST as i32, - buzz_core::kind::KIND_FORUM_COMMENT as i32, - buzz_core::kind::KIND_GIT_PULL_REQUEST as i32, - buzz_core::kind::KIND_GIT_PR_UPDATE as i32, - buzz_core::kind::KIND_GIT_ISSUE as i32, - buzz_core::kind::KIND_GIT_STATUS_OPEN as i32, - buzz_core::kind::KIND_GIT_STATUS_MERGED as i32, - buzz_core::kind::KIND_GIT_STATUS_CLOSED as i32, - buzz_core::kind::KIND_GIT_STATUS_DRAFT as i32, - ]); - query.p_tag_hex = Some(hex::encode(pubkey_bytes)); - query.channel_ids = Some(accessible_channel_ids.to_vec()); - query.since = since; - query.limit = Some(limit.min(crate::feed::FEED_MAX_LIMIT)); - query_events(pool, &query).await -} - -pub(crate) async fn query_feed_needs_action( - pool: &SqlitePool, - community: CommunityId, - pubkey_bytes: &[u8], - accessible_channel_ids: &[Uuid], - since: Option>, - limit: i64, -) -> Result> { - let mut query = crate::EventQuery::for_community(community); - query.kinds = Some(vec![ - buzz_core::kind::KIND_WORKFLOW_APPROVAL_REQUESTED as i32, - buzz_core::kind::KIND_STREAM_REMINDER as i32, - ]); - query.p_tag_hex = Some(hex::encode(pubkey_bytes)); - query.channel_ids = Some(accessible_channel_ids.to_vec()); - query.since = since; - query.limit = Some(limit.min(crate::feed::FEED_MAX_LIMIT)); - query_events(pool, &query).await -} - -pub(crate) async fn query_feed_activity( - pool: &SqlitePool, - community: CommunityId, - accessible_channel_ids: &[Uuid], - since: Option>, - limit: i64, -) -> Result> { - let mut query = crate::EventQuery::for_community(community); - query.kinds = Some(vec![ - buzz_core::kind::KIND_STREAM_MESSAGE as i32, - buzz_core::kind::KIND_STREAM_MESSAGE_V2 as i32, - buzz_core::kind::KIND_FORUM_POST as i32, - buzz_core::kind::KIND_JOB_REQUEST as i32, - buzz_core::kind::KIND_JOB_PROGRESS as i32, - buzz_core::kind::KIND_JOB_RESULT as i32, - ]); - query.channel_ids = Some(accessible_channel_ids.to_vec()); - query.since = since; - query.limit = Some(limit.min(crate::feed::FEED_MAX_LIMIT)); - query_events(pool, &query).await -} - -pub(crate) async fn insert_reaction_event( - pool: &SqlitePool, - community: CommunityId, - reaction_event: &nostr::Event, - channel_id: Option, - target_event_id: &[u8], - actor_pubkey: &[u8], - emoji: &str, -) -> Result { - let mut tx = pool.begin().await?; - let target_created_at: Option = sqlx::query_scalar( - "SELECT created_at FROM events WHERE community_id = ?1 AND id = ?2 LIMIT 1", - ) - .bind(community.as_uuid().to_string()) - .bind(target_event_id) - .fetch_optional(&mut *tx) - .await?; - let Some(target_created_at) = target_created_at else { - return Ok(crate::event::ReactionEventInsertOutcome::TargetMissing); - }; - let changed = sqlx::query("INSERT INTO reactions (community_id, event_created_at, event_id, pubkey, emoji, reaction_event_id) VALUES (?1, ?2, ?3, ?4, ?5, ?6) ON CONFLICT (community_id, event_created_at, event_id, pubkey, emoji) DO UPDATE SET removed_at = NULL, reaction_event_id = excluded.reaction_event_id WHERE reactions.removed_at IS NOT NULL") - .bind(community.as_uuid().to_string()).bind(target_created_at).bind(target_event_id) - .bind(actor_pubkey).bind(emoji).bind(reaction_event.id.as_bytes().as_slice()) - .execute(&mut *tx).await?.rows_affected() != 0; - if !changed { - return Ok(crate::event::ReactionEventInsertOutcome::Duplicate); - } - let received_at = chrono::Utc::now(); - let inserted = sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags_json, content, sig, channel_id, received_at, event_json) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) ON CONFLICT DO NOTHING") - .bind(community.as_uuid().to_string()).bind(reaction_event.id.as_bytes().as_slice()).bind(reaction_event.pubkey.to_bytes().as_slice()) - .bind(reaction_event.created_at.as_secs() as i64).bind(reaction_event.kind.as_u16() as i32).bind(serde_json::to_string(&reaction_event.tags)?) - .bind(&reaction_event.content).bind(reaction_event.sig.serialize().as_slice()).bind(channel_id.map(|id| id.to_string())) - .bind(received_at.timestamp()).bind(serde_json::to_string(reaction_event)?).execute(&mut *tx).await?.rows_affected() == 1; - tx.commit().await?; - Ok(crate::event::ReactionEventInsertOutcome::Inserted { - stored_event: Box::new(buzz_core::StoredEvent::with_received_at( - reaction_event.clone(), - received_at, - channel_id, - true, - )), - was_inserted: inserted, - }) -} - -pub(crate) async fn remove_reaction( - pool: &SqlitePool, - community: CommunityId, - event_id: &[u8], - event_created_at: chrono::DateTime, - pubkey: &[u8], - emoji: &str, -) -> Result { - Ok(sqlx::query("UPDATE reactions SET removed_at = unixepoch() WHERE community_id = ?1 AND event_created_at = ?2 AND event_id = ?3 AND pubkey = ?4 AND emoji = ?5 AND removed_at IS NULL") - .bind(community.as_uuid().to_string()).bind(event_created_at.timestamp()).bind(event_id).bind(pubkey).bind(emoji) - .execute(pool).await?.rows_affected() != 0) -} - -pub(crate) async fn remove_reaction_by_source_event_id( - pool: &SqlitePool, - community: CommunityId, - reaction_event_id: &[u8], -) -> Result { - Ok(sqlx::query("UPDATE reactions SET removed_at = unixepoch() WHERE community_id = ?1 AND reaction_event_id = ?2 AND removed_at IS NULL") - .bind(community.as_uuid().to_string()).bind(reaction_event_id) - .execute(pool).await?.rows_affected() != 0) -} - -pub(crate) async fn open_dm( - pool: &SqlitePool, - community: CommunityId, - pubkeys: &[&[u8]], - created_by: &[u8], - command_event: Option<&nostr::Event>, -) -> Result> { - let mut participants = pubkeys.to_vec(); - if !participants.contains(&created_by) { - participants.push(created_by); - } - participants.sort_unstable(); - participants.dedup(); - if !(2..=9).contains(&participants.len()) { - return Err(crate::DbError::InvalidData( - "DM requires 2-9 unique participants".into(), - )); - } - if participants.iter().any(|pubkey| pubkey.len() != 32) { - return Err(crate::DbError::InvalidData( - "DM participant pubkeys must be 32 bytes".into(), - )); - } - let hash = crate::dm::compute_participant_hash(&participants); - let mut tx = pool.begin().await?; - if let Some(event) = command_event { - let received_at = chrono::Utc::now(); - let inserted = sqlx::query("INSERT INTO events (community_id, id, pubkey, created_at, kind, tags_json, content, sig, channel_id, received_at, event_json) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL, ?9, ?10) ON CONFLICT DO NOTHING") - .bind(community.as_uuid().to_string()).bind(event.id.as_bytes().as_slice()).bind(event.pubkey.to_bytes().as_slice()) - .bind(event.created_at.as_secs() as i64).bind(event.kind.as_u16() as i32).bind(serde_json::to_string(&event.tags)?) - .bind(&event.content).bind(event.sig.serialize().as_slice()).bind(received_at.timestamp()).bind(serde_json::to_string(event)?) - .execute(&mut *tx).await?.rows_affected() != 0; - if !inserted { - return Ok(None); - } - } - let select = "SELECT id, name, channel_type, visibility, description, canvas, created_by, created_at, updated_at, archived_at, deleted_at, nip29_group_id, topic_required, max_members, topic, topic_set_by, topic_set_at, purpose, purpose_set_by, purpose_set_at, ttl_seconds, ttl_deadline FROM channels WHERE community_id = ?1 AND participant_hash = ?2 AND channel_type = 'dm' AND deleted_at IS NULL LIMIT 1"; - if let Some(row) = sqlx::query(select) - .bind(community.as_uuid().to_string()) - .bind(hash.as_slice()) - .fetch_optional(&mut *tx) - .await? - { - sqlx::query("UPDATE channel_members SET hidden_at = NULL WHERE channel_id = ?1 AND pubkey = ?2 AND removed_at IS NULL") - .bind(row.try_get::("id")?) - .bind(created_by) - .execute(&mut *tx) - .await?; - let record = channel_record(row)?; - tx.commit().await?; - return Ok(Some((record, false))); - } - - let id = Uuid::new_v4(); - let name = if participants.len() == 2 { - "DM".to_owned() - } else { - format!("Group DM ({})", participants.len()) - }; - sqlx::query("INSERT INTO channels (id, community_id, name, channel_type, visibility, participant_hash, created_by) VALUES (?1, ?2, ?3, 'dm', 'private', ?4, ?5)") - .bind(id.to_string()) - .bind(community.as_uuid().to_string()) - .bind(name) - .bind(hash.as_slice()) - .bind(created_by) - .execute(&mut *tx) - .await?; - for participant in participants { - sqlx::query("INSERT INTO channel_members (channel_id, pubkey, role, invited_by) VALUES (?1, ?2, 'member', ?3)") - .bind(id.to_string()) - .bind(participant) - .bind(created_by) - .execute(&mut *tx) - .await?; - } - let row = sqlx::query(select) - .bind(community.as_uuid().to_string()) - .bind(hash.as_slice()) - .fetch_one(&mut *tx) - .await?; - let record = channel_record(row)?; - tx.commit().await?; - Ok(Some((record, true))) -} - -fn stored_event(row: sqlx::sqlite::SqliteRow) -> Result { - let json: String = row.try_get("event_json")?; - let event: nostr::Event = serde_json::from_str(&json)?; - let received: i64 = row.try_get("received_at")?; - let channel: Option = row.try_get("channel_id")?; - let channel_id = channel - .map(|id| { - Uuid::parse_str(&id) - .map_err(|e| crate::DbError::InvalidData(format!("invalid SQLite channel id: {e}"))) - }) - .transpose()?; - Ok(buzz_core::StoredEvent::with_received_at( - event, - timestamp(received)?, - channel_id, - true, - )) -} - #[allow(clippy::too_many_arguments)] pub(crate) async fn create_channel_with_id( pool: &SqlitePool, @@ -960,38 +347,6 @@ pub(crate) async fn get_accessible_channel_ids( .collect() } -pub(crate) async fn communities_of_channels( - pool: &SqlitePool, - channel_ids: &[Uuid], -) -> Result> { - let mut out = std::collections::HashMap::with_capacity(channel_ids.len()); - for channel_id in channel_ids { - let row = sqlx::query_scalar::<_, String>( - "SELECT community_id FROM channels WHERE id = ?1 AND deleted_at IS NULL", - ) - .bind(channel_id.to_string()) - .fetch_optional(pool) - .await?; - if let Some(community_id) = row { - let community_id = Uuid::parse_str(&community_id).map_err(|e| { - crate::DbError::InvalidData(format!("invalid SQLite community id: {e}")) - })?; - out.insert(*channel_id, CommunityId::from_uuid(community_id)); - } - } - Ok(out) -} - -pub(crate) async fn get_member_role( - pool: &SqlitePool, - community: CommunityId, - channel_id: Uuid, - pubkey: &[u8], -) -> Result> { - Ok(sqlx::query_scalar("SELECT cm.role 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_optional(pool).await?) -} - fn timestamp(value: i64) -> Result> { chrono::DateTime::from_timestamp(value, 0).ok_or(crate::DbError::InvalidTimestamp(value)) } @@ -1038,117 +393,6 @@ fn member_record(row: sqlx::sqlite::SqliteRow) -> Result Result { - Ok(sqlx::query( - "INSERT INTO users (community_id, pubkey) VALUES (?1, ?2) ON CONFLICT DO NOTHING", - ) - .bind(community.as_uuid().to_string()) - .bind(pubkey) - .execute(pool) - .await? - .rows_affected() - == 1) -} - -pub(crate) async fn update_user_profile( - pool: &SqlitePool, - community: CommunityId, - pubkey: &[u8], - display_name: Option<&str>, - avatar_url: Option<&str>, - about: Option<&str>, - nip05_handle: Option<&str>, -) -> Result<()> { - sqlx::query( - "UPDATE users SET display_name = COALESCE(?3, display_name), avatar_url = COALESCE(?4, avatar_url), about = COALESCE(?5, about), nip05_handle = COALESCE(?6, nip05_handle) WHERE community_id = ?1 AND pubkey = ?2", - ) - .bind(community.as_uuid().to_string()) - .bind(pubkey) - .bind(display_name) - .bind(avatar_url) - .bind(about) - .bind(nip05_handle) - .execute(pool) - .await?; - Ok(()) -} - -pub(crate) async fn get_user( - pool: &SqlitePool, - community: CommunityId, - pubkey: &[u8], -) -> Result> { - let row = sqlx::query( - "SELECT pubkey, display_name, avatar_url, about, nip05_handle FROM users WHERE community_id = ?1 AND pubkey = ?2", - ) - .bind(community.as_uuid().to_string()) - .bind(pubkey) - .fetch_optional(pool) - .await?; - row.map(|row| { - Ok(crate::user::UserProfile { - pubkey: row.try_get("pubkey")?, - display_name: row.try_get("display_name")?, - avatar_url: row.try_get("avatar_url")?, - about: row.try_get("about")?, - nip05_handle: row.try_get("nip05_handle")?, - }) - }) - .transpose() -} - -pub(crate) async fn set_agent_owner( - pool: &SqlitePool, - community: CommunityId, - agent_pubkey: &[u8], - owner_pubkey: &[u8], -) -> Result { - Ok(sqlx::query( - "UPDATE users SET agent_owner_pubkey = ?3, is_agent = 1 WHERE community_id = ?1 AND pubkey = ?2 AND agent_owner_pubkey IS NULL", - ) - .bind(community.as_uuid().to_string()) - .bind(agent_pubkey) - .bind(owner_pubkey) - .execute(pool) - .await? - .rows_affected() == 1) -} - -pub(crate) async fn get_agent_channel_policy( - pool: &SqlitePool, - community: CommunityId, - pubkey: &[u8], -) -> Result>)>> { - Ok(sqlx::query_as( - "SELECT channel_add_policy, agent_owner_pubkey FROM users WHERE community_id = ?1 AND pubkey = ?2", - ) - .bind(community.as_uuid().to_string()) - .bind(pubkey) - .fetch_optional(pool) - .await?) -} - -pub(crate) async fn is_agent_owner( - pool: &SqlitePool, - community: CommunityId, - target_pubkey: &[u8], - actor_pubkey: &[u8], -) -> Result { - Ok(sqlx::query_scalar::<_, i64>( - "SELECT count(*) FROM users WHERE community_id = ?1 AND pubkey = ?2 AND agent_owner_pubkey = ?3", - ) - .bind(community.as_uuid().to_string()) - .bind(target_pubkey) - .bind(actor_pubkey) - .fetch_one(pool) - .await? != 0) -} - pub(crate) async fn is_relay_member( pool: &SqlitePool, community: CommunityId, @@ -1264,63 +508,16 @@ fn community_record(row: sqlx::sqlite::SqliteRow) -> Result { #[cfg(test)] mod tests { use super::*; - use nostr::{EventBuilder, Keys, Kind}; + use nostr::Keys; #[tokio::test] - async fn upgrades_phase_1_core_schema_before_dm_use() { - let path = std::env::temp_dir().join(format!("buzz-db-upgrade-{}.sqlite", Uuid::new_v4())); - let options = SqliteConnectOptions::from_str(&format!("sqlite://{}", path.display())) - .unwrap() - .create_if_missing(true) - .foreign_keys(true); - let pool = SqlitePoolOptions::new() - .max_connections(1) - .connect_with(options) - .await - .unwrap(); - sqlx::query("CREATE TABLE communities (id TEXT PRIMARY KEY NOT NULL, host TEXT NOT NULL COLLATE NOCASE UNIQUE, icon TEXT, created_at INTEGER NOT NULL DEFAULT (unixepoch()), archived_at INTEGER)") - .execute(&pool).await.unwrap(); - sqlx::query("CREATE TABLE channels (id TEXT PRIMARY KEY NOT NULL, community_id TEXT NOT NULL, name TEXT NOT NULL, description TEXT, canvas TEXT, channel_type TEXT NOT NULL, visibility TEXT NOT NULL, created_by BLOB NOT NULL, created_at INTEGER NOT NULL DEFAULT (unixepoch()), updated_at INTEGER NOT NULL DEFAULT (unixepoch()), archived_at INTEGER, deleted_at INTEGER, nip29_group_id TEXT, topic_required INTEGER NOT NULL DEFAULT 0, max_members INTEGER, topic TEXT, topic_set_by BLOB, topic_set_at INTEGER, purpose TEXT, purpose_set_by BLOB, purpose_set_at INTEGER, ttl_seconds INTEGER, ttl_deadline INTEGER)") - .execute(&pool).await.unwrap(); - sqlx::query("CREATE TABLE channel_members (channel_id TEXT NOT NULL, pubkey BLOB NOT NULL, role TEXT NOT NULL, joined_at INTEGER NOT NULL DEFAULT (unixepoch()), invited_by BLOB, removed_at INTEGER, PRIMARY KEY (channel_id, pubkey))") - .execute(&pool).await.unwrap(); - pool.close().await; - - let upgraded = connect(path.to_str().unwrap()).await.unwrap(); - assert_eq!( - sqlx::query_scalar::<_, i64>("SELECT version FROM schema_version WHERE singleton = 1") - .fetch_one(&upgraded) - .await - .unwrap(), - 2 - ); - for (table, column) in [ - ("channels", "participant_hash"), - ("channel_members", "hidden_at"), - ] { - let pragma = format!("PRAGMA table_info({table})"); - assert!(sqlx::query(sqlx::AssertSqlSafe(pragma)) - .fetch_all(&upgraded) - .await - .unwrap() - .iter() - .any(|row| row.get::("name") == column)); - } - upgraded.close().await; - std::fs::remove_file(path).unwrap(); - } - - #[tokio::test] - async fn core_slice_survives_temporary_file_reopen() { + 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 event = EventBuilder::new(Kind::Custom(9), "durable message") - .sign_with_keys(&owner) - .unwrap(); let pool = connect(&path_string).await.unwrap(); let ensured = ensure_configured_community(&pool, "Local.Buzz") @@ -1329,7 +526,6 @@ mod tests { bootstrap_owner(&pool, ensured.id, &owner_hex) .await .unwrap(); - assert!(ensure_user(&pool, ensured.id, &owner_bytes).await.unwrap()); let (channel, created) = create_channel_with_id( &pool, ensured.id, @@ -1348,12 +544,6 @@ mod tests { assert!(is_member(&pool, ensured.id, channel_id, &owner_bytes) .await .unwrap()); - assert!( - insert_event(&pool, ensured.id, &event, Some(channel_id)) - .await - .unwrap() - .1 - ); pool.close().await; let reopened = connect(&path_string).await.unwrap(); @@ -1372,16 +562,32 @@ mod tests { .unwrap(), vec![channel_id] ); - let stored = get_event_by_id(&reopened, ensured.id, event.id.as_bytes()) - .await - .unwrap() - .unwrap(); - assert_eq!(stored.event.content, "durable message"); - let mut query = crate::EventQuery::for_community(ensured.id); - query.channel_id = Some(channel_id); - query.kinds = Some(vec![9]); - assert_eq!(query_events(&reopened, &query).await.unwrap().len(), 1); reopened.close().await; std::fs::remove_file(path).unwrap(); } + + #[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-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 4981845b17..a118ff453f 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1732,13 +1732,7 @@ async fn handle_bridge_search( .search .search(&search_query) .await - .map_err(|e| match e { - buzz_search::SearchError::Unsupported => api_error( - StatusCode::NOT_IMPLEMENTED, - "unsupported_feature: search is unavailable in this runtime profile", - ), - other => internal_error(&format!("search error: {other}")), - })?; + .map_err(|e| internal_error(&format!("search error: {e}")))?; // Fetch full events from DB by ID. Hit ids are already raw 32-byte // arrays from the FTS layer — no hex decode. diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 8f0cd0a250..2d82736807 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -80,23 +80,8 @@ pub async fn handle_command( enum PersistResult { /// Event was already processed — return idempotent success. Duplicate, - /// Event inserted — PostgreSQL remains open until mutations finish; SQLite - /// committed the idempotency row atomically before the local mutation. - Inserted(CommandTransaction), -} - -enum CommandTransaction { - Sqlite, - Postgres(sqlx::Transaction<'static, sqlx::Postgres>), -} - -impl CommandTransaction { - async fn commit(self) -> Result<(), sqlx::Error> { - match self { - Self::Sqlite => Ok(()), - Self::Postgres(tx) => tx.commit().await, - } - } + /// Event inserted — transaction is open, handler must commit after mutations. + Inserted(sqlx::Transaction<'static, sqlx::Postgres>), } /// Persist a command event inside a transaction. Returns the OPEN transaction @@ -120,12 +105,6 @@ async fn persist_command_event( ) -> Result { let channel_id = channel_id_override.or_else(|| extract_channel_id(event)); - if state.config.profile.is_single_node() { - return Err(IngestError::Rejected( - "unsupported_feature: this command is unavailable in the single-node profile".into(), - )); - } - let mut tx = state .db .begin_transaction() @@ -248,7 +227,7 @@ async fn persist_command_event( // Duplicate — rollback (implicit on drop) and signal idempotent success. Ok(PersistResult::Duplicate) } else { - Ok(PersistResult::Inserted(CommandTransaction::Postgres(tx))) + Ok(PersistResult::Inserted(tx)) } } @@ -366,35 +345,27 @@ async fn handle_dm_open( } } - // SQLite couples the command idempotency row and DM mutation in one - // transaction. PostgreSQL retains the existing open-transaction path. - let all_refs: Vec<&[u8]> = all_bytes.iter().map(|b| b.as_slice()).collect(); - let sqlite_result = state - .db - .open_dm_sqlite_command(tenant.community(), &all_refs, &self_bytes, event) - .await - .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; - let (channel, was_created, tx) = if let Some(result) = sqlite_result { - (result.0, result.1, CommandTransaction::Sqlite) - } else { - let tx = match persist_command_event(state, tenant, event, None).await? { - PersistResult::Duplicate => { - return Ok(IngestResult { - event_id: event.id.to_hex(), - accepted: true, - message: "duplicate: already processed".into(), - }); - } - PersistResult::Inserted(tx) => tx, - }; - let (channel, was_created) = state - .db - .open_dm(tenant.community(), &all_refs, &self_bytes) - .await - .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; - (channel, was_created, tx) + // Persist the command event (idempotency) — returns open transaction + let tx = match persist_command_event(state, tenant, event, None).await? { + PersistResult::Duplicate => { + return Ok(IngestResult { + event_id: event.id.to_hex(), + accepted: true, + message: "duplicate: already processed".into(), + }); + } + PersistResult::Inserted(tx) => tx, }; + // 4. Execute: open_dm + let all_refs: Vec<&[u8]> = all_bytes.iter().map(|b| b.as_slice()).collect(); + let (channel, was_created) = state + .db + .open_dm(tenant.community(), &all_refs, &self_bytes) + .await + .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; + + // Commit: event + mutation succeeded atomically. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index cf295bdfa3..a67797385b 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -525,8 +525,7 @@ async fn dispatch_persistent_event_inner( .iter() .any(|t| t.as_slice().first().map(|s| s.as_str()) == Some("buzz:workflow")); - if !state.config.profile.is_single_node() - && !buzz_core::kind::is_workflow_execution_kind(kind_u32) + if !buzz_core::kind::is_workflow_execution_kind(kind_u32) && !buzz_core::kind::is_command_kind(kind_u32) && !is_relay_workflow_msg && kind_u32 != KIND_GIFT_WRAP diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index ac24e74c23..88a9f0c731 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -809,9 +809,6 @@ pub fn emit_live_thread_summary( channel_id: Uuid, root_id: Vec, ) { - if state.config.profile.is_single_node() { - return; - } let tenant = tenant.clone(); let state = Arc::clone(state); tokio::spawn(async move { @@ -2935,12 +2932,6 @@ pub async fn publish_nip43_membership_list( tenant: &TenantContext, state: &Arc, ) -> anyhow::Result<()> { - // The production implementation relies on a PostgreSQL advisory lock for - // snapshot serialization. Single-node still emits the durable per-member - // delta above, but deliberately omits this denormalized list snapshot. - if state.config.profile.is_single_node() { - return Ok(()); - } let started_at = std::time::Instant::now(); metrics::counter!("buzz_nip43_membership_publications_total", "result" => "attempted") .increment(1); @@ -2999,11 +2990,6 @@ async fn publish_nip43_delta( target_pubkey_hex: &str, label: &str, ) -> anyhow::Result<()> { - // These are globally scoped events. Keep local relay membership private to - // the loopback process rather than exposing deltas to every subscriber. - if state.config.profile.is_single_node() { - return Ok(()); - } let relay_pubkey_hex = state.relay_keypair.public_key().to_hex(); let tags = vec![