From bb3c39049d313a40cbaca445096e9078d6a82a65 Mon Sep 17 00:00:00 2001 From: npub1z3hmzc9ryehxzedl5wzlvpyvja0d483peaja5zt6pd0209f9x2jspe2dxh <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> Date: Sat, 1 Aug 2026 10:30:35 -0400 Subject: [PATCH] refactor(db): gate postgres operations by backend Signed-off-by: npub1z3hmzc9ryehxzedl5wzlvpyvja0d483peaja5zt6pd0209f9x2jspe2dxh <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> --- crates/buzz-db/src/error.rs | 4 + crates/buzz-db/src/lib.rs | 601 +++++++++++++++++++++--------------- 2 files changed, 361 insertions(+), 244 deletions(-) diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index f8b8a2eb5..2dbfa210a 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -13,6 +13,10 @@ pub enum DbError { #[error("migration error: {0}")] Migrate(#[from] sqlx::migrate::MigrateError), + /// Operation is unsupported by the selected database backend. + #[error("unsupported database backend operation: {0}")] + UnsupportedBackend(&'static str), + /// Attempted to store an AUTH event (kind 22242), which is forbidden. #[error("AUTH events (kind 22242) must not be stored")] AuthEventRejected, diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9b2687674..0475fb4d6 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -168,9 +168,18 @@ pub async fn insert_mentions( Ok(()) } +/// Selected backing store for a [`Db`] handle. +#[derive(Clone, Debug)] +enum DbBackend { + Postgres, + SQLite(sqlx::SqlitePool), +} + /// Database handle. Clone is cheap (Arc-backed pool). #[derive(Clone, Debug)] pub struct Db { + backend: DbBackend, + // 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, @@ -639,6 +648,24 @@ 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 for the local profile, if selected. + fn sqlite_pool(&self) -> Option<&sqlx::SqlitePool> { + match &self.backend { + DbBackend::SQLite(pool) => Some(pool), + DbBackend::Postgres => None, + } + } + /// Creates a new `Db` by connecting a Postgres pool with the given config. /// /// When `config.read_database_url` is set, a second pool with the same @@ -659,6 +686,7 @@ impl Db { }; let replica_read_max_age = read_budget_from_ms(config.replica_read_max_age_ms); Ok(Self { + backend: DbBackend::Postgres, pool, max_connections: config.max_connections, read_pool, @@ -774,6 +802,7 @@ impl Db { /// Creates a `Db` from an existing `PgPool` (useful in tests). pub fn from_pool(pool: PgPool) -> Self { Self { + backend: DbBackend::Postgres, max_connections: pool.options().get_max_connections(), read_max_connections: pool.options().get_max_connections(), pool, @@ -793,6 +822,7 @@ impl Db { /// [`Db::fence`]). pub fn from_pools(pool: PgPool, read_pool: PgPool) -> Self { Self { + backend: DbBackend::Postgres, max_connections: pool.options().get_max_connections(), read_max_connections: read_pool.options().get_max_connections(), pool, @@ -834,10 +864,10 @@ impl Db { if self.read_pool.is_none() { return Ok(false); } - replica_fence::verify_floor_guard_catalog(&self.pool).await?; - replica_fence::verify_floor_guard_behavior(&self.pool).await?; + replica_fence::verify_floor_guard_catalog(self.pg_pool()?).await?; + replica_fence::verify_floor_guard_behavior(self.pg_pool()?).await?; tokio::spawn(replica_fence::run_probe( - self.pool.clone(), + self.pg_pool()?.clone(), std::sync::Arc::clone(&self.fence), )); Ok(true) @@ -852,8 +882,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) -> &PgPool { - self.read_pool.as_ref().unwrap_or(&self.pool) + fn read(&self) -> Result<&PgPool> { + Ok(self.read_pool.as_ref().unwrap_or(self.pg_pool()?)) } /// Whether a distinct read-replica pool is configured. @@ -1010,12 +1040,15 @@ impl Db { /// Run pending database migrations. pub async fn migrate(&self) -> Result<()> { - migration::run_migrations(&self.pool).await + migration::run_migrations(self.pg_pool()?).await } /// Returns `true` if the database is reachable (used by readiness probes). pub async fn ping(&self) -> bool { - sqlx::query("SELECT 1").execute(&self.pool).await.is_ok() + match self.pg_pool() { + Ok(pool) => sqlx::query("SELECT 1").execute(pool).await.is_ok(), + Err(_) => false, + } } /// Returns pool utilisation stats for metrics emission. @@ -1024,9 +1057,19 @@ 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: self.pool.size(), - idle: self.pool.num_idle() as u32, + size: pool.size(), + idle: pool.num_idle() as u32, max: self.max_connections, } } @@ -1057,7 +1100,7 @@ impl Db { &self, lock_key: i64, ) -> Result> { - let mut connection = self.pool.acquire().await?; + let mut connection = self.pg_pool()?.acquire().await?; let acquired = sqlx::query_scalar::<_, bool>("SELECT pg_try_advisory_lock($1)") .bind(lock_key) .fetch_one(&mut *connection) @@ -1085,7 +1128,7 @@ impl Db { limit: i64, ) -> Result> { admin_moderation::list_reports( - &self.pool, + self.pg_pool()?, community_id, status, report_type, @@ -1103,7 +1146,7 @@ impl Db { &self, id: Uuid, ) -> Result> { - admin_moderation::get_report(&self.pool, id).await + admin_moderation::get_report(self.pg_pool()?, id).await } /// List feedback for the deployment-global read-only admin plane. @@ -1111,7 +1154,7 @@ impl Db { &self, limit: i64, ) -> Result> { - admin_moderation::list_feedback(&self.pool, limit).await + admin_moderation::list_feedback(self.pg_pool()?, limit).await } /// Fetch one feedback submission for the deployment-global admin plane. @@ -1119,42 +1162,42 @@ impl Db { &self, id: Uuid, ) -> Result> { - admin_moderation::get_feedback(&self.pool, id).await + admin_moderation::get_feedback(self.pg_pool()?, id).await } /// Return total number of communities on this relay. pub async fn usage_community_count(&self) -> Result { - usage::community_count(&self.pool).await + usage::community_count(self.pg_pool()?).await } /// Return per-community user counts split by human/agent. pub async fn usage_user_counts(&self) -> Result> { - usage::user_counts(&self.pool).await + usage::user_counts(self.pg_pool()?).await } /// Return per-community channel counts by type. pub async fn usage_channel_counts(&self) -> Result> { - usage::channel_counts(&self.pool).await + usage::channel_counts(self.pg_pool()?).await } /// Return per-community kind=9 message counts. pub async fn usage_message_counts(&self) -> Result> { - usage::message_counts(&self.pool).await + usage::message_counts(self.pg_pool()?).await } /// Return per-community relay-member counts by role. pub async fn usage_relay_member_counts(&self) -> Result> { - usage::relay_member_counts(&self.pool).await + usage::relay_member_counts(self.pg_pool()?).await } /// Return per-community workflow counts by status. pub async fn usage_workflow_counts(&self) -> Result> { - usage::workflow_counts(&self.pool).await + usage::workflow_counts(self.pg_pool()?).await } /// Return per-community git-repo counts. pub async fn usage_git_repo_counts(&self) -> Result> { - usage::git_repo_counts(&self.pool).await + usage::git_repo_counts(self.pg_pool()?).await } /// Return per-community distinct active-user counts for a given SQL interval. @@ -1164,7 +1207,7 @@ impl Db { &self, interval_sql: &'static str, ) -> Result> { - usage::active_user_counts(&self.pool, interval_sql).await + usage::active_user_counts(self.pg_pool()?, interval_sql).await } /// Return per-community active-channel counts for a given SQL interval. @@ -1172,12 +1215,12 @@ impl Db { &self, interval_sql: &'static str, ) -> Result> { - usage::active_channel_counts(&self.pool, interval_sql).await + usage::active_channel_counts(self.pg_pool()?, interval_sql).await } /// Return all community id → host mappings. pub async fn usage_community_hosts(&self) -> Result> { - usage::community_hosts(&self.pool).await + usage::community_hosts(self.pg_pool()?).await } /// Begin a database transaction for atomic multi-statement operations. @@ -1185,7 +1228,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.pool.begin().await.map_err(Into::into) + self.pg_pool()?.begin().await.map_err(Into::into) } /// Returns the community mapped to a normalized request host, if one exists. @@ -1205,7 +1248,7 @@ impl Db { "#, ) .bind(normalized_host) - .fetch_optional(&self.pool) + .fetch_optional(self.pg_pool()?) .await?; row.map(|row| { @@ -1226,7 +1269,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.pool) + .fetch_one(self.pg_pool()?) .await?; Ok(active) } @@ -1238,7 +1281,7 @@ impl Db { ) -> Result> { let row = sqlx::query("SELECT id, host FROM communities WHERE lower(host) = lower($1)") .bind(normalized_host) - .fetch_optional(&self.pool) + .fetch_optional(self.pg_pool()?) .await?; row.map(|row| { Ok(CommunityRecord { @@ -1269,7 +1312,7 @@ impl Db { "#, ) .bind(owner_pubkey) - .fetch_all(&self.pool) + .fetch_all(self.pg_pool()?) .await?; rows.into_iter() @@ -1308,7 +1351,7 @@ impl Db { "#, ) .bind(community_id.as_uuid()) - .fetch_optional(&self.pool) + .fetch_optional(self.pg_pool()?) .await?; row.map(|row| { @@ -1331,7 +1374,7 @@ impl Db { "#, ) .bind(community_id.as_uuid()) - .fetch_optional(&self.pool) + .fetch_optional(self.pg_pool()?) .await?; Ok(row @@ -1356,7 +1399,7 @@ impl Db { ) .bind(community_id.as_uuid()) .bind(icon) - .execute(&self.pool) + .execute(self.pg_pool()?) .await?; Ok(()) } @@ -1379,7 +1422,7 @@ impl Db { "#, ) .bind(normalized_host) - .fetch_one(&self.pool) + .fetch_one(self.pg_pool()?) .await?; let id: Uuid = row.try_get("id")?; @@ -1404,7 +1447,7 @@ impl Db { owner_pubkey: &str, ) -> Result { let owner_pubkey = owner_pubkey.to_ascii_lowercase(); - let mut tx = self.pool.begin().await?; + let mut tx = self.pg_pool()?.begin().await?; // Serialize on the owner pubkey so concurrent creates to the same // owner cannot both pass the ownership count check. @@ -1503,7 +1546,7 @@ impl Db { .bind(normalized_host) .bind(owner_pubkey) .bind(protected_deployment_host) - .fetch_optional(&self.pool) + .fetch_optional(self.pg_pool()?) .await?; row.map(|row| { Ok(ArchivedCommunityRecord { @@ -1533,7 +1576,7 @@ impl Db { ) .bind(normalized_host) .bind(owner_pubkey) - .fetch_optional(&self.pool) + .fetch_optional(self.pg_pool()?) .await?; row.map(|row| { Ok(UnarchivedCommunityRecord { @@ -1558,7 +1601,7 @@ impl Db { "#, ) .bind(channel_id) - .fetch_optional(&self.pool) + .fetch_optional(self.pg_pool()?) .await?; row.map(|row| { @@ -1602,7 +1645,7 @@ impl Db { "#, ) .bind(channel_ids) - .fetch_all(&self.pool) + .fetch_all(self.pg_pool()?) .await?; let mut out = std::collections::HashMap::with_capacity(rows.len()); @@ -1621,9 +1664,10 @@ impl Db { event: &nostr::Event, channel_id: Option, ) -> Result<(StoredEvent, bool)> { - let result = event::insert_event(&self.pool, community_id, event, channel_id).await?; + let result = event::insert_event(self.pg_pool()?, community_id, event, channel_id).await?; if result.1 { - if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await { + if let Err(e) = insert_mentions(self.pg_pool()?, community_id, event, channel_id).await + { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } } @@ -1638,7 +1682,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> { - event::query_events(&self.pool, q).await + event::query_events(self.pg_pool()?, q).await } /// [`Db::query_events`] with replica routing — the opt-in fast path for @@ -1675,11 +1719,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.pool, q).await + event::query_events(self.pg_pool()?, q).await } } } - RouteDecision::Writer => event::query_events(&self.pool, q).await, + RouteDecision::Writer => event::query_events(self.pg_pool()?, q).await, } } @@ -1706,11 +1750,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.pool, q).await + event::query_events(self.pg_pool()?, q).await } } } - RouteDecision::Writer => event::query_events(&self.pool, q).await, + RouteDecision::Writer => event::query_events(self.pg_pool()?, q).await, } } @@ -1719,7 +1763,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.pool, q).await + event::count_events(self.pg_pool()?, q).await } /// [`Db::count_events`] with replica routing — same contract, rules, @@ -1743,11 +1787,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.pool, q).await + event::count_events(self.pg_pool()?, q).await } } } - RouteDecision::Writer => event::count_events(&self.pool, q).await, + RouteDecision::Writer => event::count_events(self.pg_pool()?, q).await, } } @@ -1761,7 +1805,7 @@ impl Db { creator_pubkey: &[u8], ) -> Result { event::huddle_started_link_exists( - &self.pool, + self.pg_pool()?, community_id, parent_channel_id, ephemeral_channel_id, @@ -1781,7 +1825,8 @@ impl Db { kind: i32, pubkey_bytes: &[u8], ) -> Result> { - event::get_latest_global_replaceable(&self.pool, community_id, kind, pubkey_bytes).await + event::get_latest_global_replaceable(self.pg_pool()?, community_id, kind, pubkey_bytes) + .await } /// Fetches a single non-deleted event by its raw ID bytes. @@ -1792,7 +1837,7 @@ impl Db { community_id: CommunityId, id_bytes: &[u8], ) -> Result> { - event::get_event_by_id(&self.pool, community_id, id_bytes).await + event::get_event_by_id(self.pg_pool()?, community_id, id_bytes).await } /// Fetches a single event by its raw ID bytes, **including soft-deleted rows**. @@ -1801,7 +1846,7 @@ impl Db { community_id: CommunityId, id_bytes: &[u8], ) -> Result> { - event::get_event_by_id_including_deleted(&self.pool, community_id, id_bytes).await + event::get_event_by_id_including_deleted(self.pg_pool()?, community_id, id_bytes).await } /// Soft-deletes an event. Returns `Ok(true)` if deleted, `Ok(false)` if already deleted. @@ -1810,7 +1855,7 @@ impl Db { community_id: CommunityId, event_id: &[u8], ) -> Result { - event::soft_delete_event(&self.pool, community_id, event_id).await + event::soft_delete_event(self.pg_pool()?, community_id, event_id).await } /// Soft-delete the live row for an addressable coordinate `(kind, pubkey, d_tag)` @@ -1826,7 +1871,7 @@ impl Db { deletion_created_at_secs: i64, ) -> Result { event::soft_delete_by_coordinate( - &self.pool, + self.pg_pool()?, community_id, kind, pubkey, @@ -1845,7 +1890,7 @@ impl Db { root_event_id: Option<&[u8]>, ) -> Result { event::soft_delete_event_and_update_thread( - &self.pool, + self.pg_pool()?, community_id, event_id, parent_event_id, @@ -1860,7 +1905,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result>> { - event::get_last_message_at(&self.pool, community_id, channel_id).await + event::get_last_message_at(self.pg_pool()?, community_id, channel_id).await } /// Bulk-fetch the most recent `created_at` for a set of channel IDs. @@ -1869,7 +1914,7 @@ impl Db { community_id: CommunityId, channel_ids: &[Uuid], ) -> Result>> { - event::get_last_message_at_bulk(&self.pool, community_id, channel_ids).await + event::get_last_message_at_bulk(self.pg_pool()?, community_id, channel_ids).await } /// Batch-fetch non-deleted events by their raw IDs. @@ -1878,7 +1923,7 @@ impl Db { community_id: CommunityId, ids: &[&[u8]], ) -> Result> { - event::get_events_by_ids(&self.pool, community_id, ids).await + event::get_events_by_ids(self.pg_pool()?, community_id, ids).await } /// [`Db::get_events_by_ids`] with replica routing — same contract and @@ -1904,11 +1949,13 @@ 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.pool, community_id, ids).await + 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, + RouteDecision::Writer => { + event::get_events_by_ids(self.pg_pool()?, community_id, ids).await + } } } @@ -1918,7 +1965,7 @@ impl Db { limit: i64, lease_until: DateTime, ) -> Result> { - push::claim_due_match_batch(&self.pool, limit, lease_until).await + push::claim_due_match_batch(self.pg_pool()?, limit, lease_until).await } /// Load active endpoint-enabled leases eligible for push matching. @@ -1926,7 +1973,7 @@ impl Db { &self, community: CommunityId, ) -> Result> { - push::active_match_leases(&self.pool, community).await + push::active_match_leases(self.pg_pool()?, community).await } /// Complete matcher jobs from one claimed batch while the fence holds. @@ -1936,7 +1983,7 @@ impl Db { claim_id: uuid::Uuid, event_ids: &[Vec], ) -> Result { - push::complete_match_batch(&self.pool, community, claim_id, event_ids).await + push::complete_match_batch(self.pg_pool()?, community, claim_id, event_ids).await } /// Release fenced matcher claims from one batch for retry. @@ -1947,12 +1994,12 @@ impl Db { event_ids: &[Vec], next: DateTime, ) -> Result { - push::retry_match_batch(&self.pool, community, claim_id, event_ids, next).await + push::retry_match_batch(self.pg_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.pool).await + push::reap_exhausted_matches(self.pg_pool()?).await } /// Idempotently enqueue a wake for a matched lease and event. @@ -1963,7 +2010,7 @@ impl Db { installation_id: &str, wake: push::NewWake<'_>, ) -> Result { - push::enqueue_wake(&self.pool, community, author, installation_id, wake).await + push::enqueue_wake(self.pg_pool()?, community, author, installation_id, wake).await } /// Set-wise [`Self::enqueue_push_wake`]: one transaction per batch. @@ -1972,7 +2019,7 @@ impl Db { community: CommunityId, requests: &[push::WakeRequest], ) -> Result> { - push::enqueue_wakes(&self.pool, community, requests).await + push::enqueue_wakes(self.pg_pool()?, community, requests).await } /// Exclusively claim due wake jobs for one community. @@ -1982,7 +2029,7 @@ impl Db { limit: i64, lease_until: DateTime, ) -> Result> { - push::claim_due_wakes(&self.pool, community, limit, lease_until).await + push::claim_due_wakes(self.pg_pool()?, community, limit, lease_until).await } /// Revalidate a wake's claim, source event, and current lease before send. @@ -1992,7 +2039,7 @@ impl Db { id: Uuid, claim_id: Uuid, ) -> Result { - push::revalidate_wake_for_send(&self.pool, community, id, claim_id).await + push::revalidate_wake_for_send(self.pg_pool()?, community, id, claim_id).await } /// Mark a fenced wake claim delivered. @@ -2002,7 +2049,7 @@ impl Db { id: Uuid, claim_id: Uuid, ) -> Result { - push::complete_wake(&self.pool, community, id, claim_id).await + push::complete_wake(self.pg_pool()?, community, id, claim_id).await } /// Release a fenced wake claim for retry at the supplied time. @@ -2013,7 +2060,7 @@ impl Db { claim_id: Uuid, next: DateTime, ) -> Result { - push::retry_wake(&self.pool, community, id, claim_id, next).await + push::retry_wake(self.pg_pool()?, community, id, claim_id, next).await } /// Mark a fenced wake claim terminally failed. @@ -2023,7 +2070,7 @@ impl Db { id: Uuid, claim_id: Uuid, ) -> Result { - push::fail_wake(&self.pool, community, id, claim_id).await + push::fail_wake(self.pg_pool()?, community, id, claim_id).await } /// Disable an endpoint only if the specified lease generation is current. @@ -2035,7 +2082,7 @@ impl Db { generation: i64, ) -> Result { push::disable_endpoint_generation( - &self.pool, + self.pg_pool()?, community, author, installation_id, @@ -2056,7 +2103,7 @@ impl Db { max_active_leases: i64, ) -> Result { push::accept_lease_event( - &self.pool, + self.pg_pool()?, community, event, installation_id, @@ -2076,7 +2123,7 @@ impl Db { thread_meta: Option>, ) -> Result<(StoredEvent, bool)> { let result = event::insert_event_with_thread_metadata( - &self.pool, + self.pg_pool()?, community_id, event, channel_id, @@ -2084,7 +2131,8 @@ impl Db { ) .await?; if result.1 { - if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await { + if let Err(e) = insert_mentions(self.pg_pool()?, community_id, event, channel_id).await + { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } } @@ -2104,7 +2152,7 @@ impl Db { emoji: &str, ) -> Result { let outcome = event::insert_reaction_event_with_thread_metadata( - &self.pool, + self.pg_pool()?, community_id, event, channel_id, @@ -2118,7 +2166,8 @@ impl Db { was_inserted: true, .. } = &outcome { - if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await { + if let Err(e) = insert_mentions(self.pg_pool()?, community_id, event, channel_id).await + { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } } @@ -2138,7 +2187,7 @@ impl Db { ttl_seconds: Option, ) -> Result { channel::create_channel( - &self.pool, + self.pg_pool()?, community_id, name, channel_type, @@ -2166,7 +2215,7 @@ impl Db { ttl_seconds: Option, ) -> Result<(channel::ChannelRecord, bool)> { channel::create_channel_with_id( - &self.pool, + self.pg_pool()?, community_id, channel_id, name, @@ -2185,7 +2234,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result { - channel::get_channel(&self.pool, community_id, channel_id).await + channel::get_channel(self.pg_pool()?, community_id, channel_id).await } /// Returns the canvas content for a channel, if any. @@ -2194,7 +2243,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result> { - channel::get_canvas(&self.pool, community_id, channel_id).await + channel::get_canvas(self.pg_pool()?, community_id, channel_id).await } /// Sets or clears the canvas content for a channel. @@ -2204,7 +2253,7 @@ impl Db { channel_id: Uuid, canvas: Option<&str>, ) -> Result<()> { - channel::set_canvas(&self.pool, community_id, channel_id, canvas).await + channel::set_canvas(self.pg_pool()?, community_id, channel_id, canvas).await } /// Adds a member to a channel. @@ -2217,7 +2266,7 @@ impl Db { invited_by: Option<&[u8]>, ) -> Result { channel::add_member( - &self.pool, + self.pg_pool()?, community_id, channel_id, pubkey, @@ -2235,7 +2284,14 @@ impl Db { pubkey: &[u8], actor_pubkey: &[u8], ) -> Result<()> { - channel::remove_member(&self.pool, community_id, channel_id, pubkey, actor_pubkey).await + channel::remove_member( + self.pg_pool()?, + community_id, + channel_id, + pubkey, + actor_pubkey, + ) + .await } /// Returns `true` if the pubkey is an active member. @@ -2245,7 +2301,7 @@ impl Db { channel_id: Uuid, pubkey: &[u8], ) -> Result { - channel::is_member(&self.pool, community_id, channel_id, pubkey).await + channel::is_member(self.pg_pool()?, community_id, channel_id, pubkey).await } /// Return the active (channel, pubkey) membership pairs among the given @@ -2256,7 +2312,7 @@ impl Db { channel_ids: &[Uuid], pubkeys: &[Vec], ) -> Result)>> { - channel::membership_pairs(&self.pool, community_id, channel_ids, pubkeys).await + channel::membership_pairs(self.pg_pool()?, community_id, channel_ids, pubkeys).await } /// Returns all active members of a channel. @@ -2265,7 +2321,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result> { - channel::get_members(&self.pool, community_id, channel_id).await + channel::get_members(self.pg_pool()?, community_id, channel_id).await } /// Returns active members for multiple channels in a single query. @@ -2274,7 +2330,7 @@ impl Db { community_id: CommunityId, channel_ids: &[Uuid], ) -> Result> { - channel::get_members_bulk(&self.pool, community_id, channel_ids).await + channel::get_members_bulk(self.pg_pool()?, community_id, channel_ids).await } /// Get all channel IDs accessible to a pubkey. @@ -2283,7 +2339,7 @@ impl Db { community_id: CommunityId, pubkey: &[u8], ) -> Result> { - channel::get_accessible_channel_ids(&self.pool, community_id, pubkey).await + channel::get_accessible_channel_ids(self.pg_pool()?, community_id, pubkey).await } /// Lists channels, optionally filtered by visibility. @@ -2292,7 +2348,7 @@ impl Db { community_id: CommunityId, visibility: Option<&str>, ) -> Result> { - channel::list_channels(&self.pool, community_id, visibility).await + channel::list_channels(self.pg_pool()?, community_id, visibility).await } /// Returns full channel records for all channels a user can access. @@ -2304,7 +2360,7 @@ impl Db { member_only: Option, ) -> Result> { channel::get_accessible_channels( - &self.pool, + self.pg_pool()?, community_id, pubkey, visibility_filter, @@ -2318,7 +2374,7 @@ impl Db { &self, community_id: CommunityId, ) -> Result> { - channel::get_bot_members(&self.pool, community_id).await + channel::get_bot_members(self.pg_pool()?, community_id).await } /// Bulk-fetch user records by pubkey. @@ -2327,7 +2383,7 @@ impl Db { community_id: CommunityId, pubkeys: &[Vec], ) -> Result> { - channel::get_users_bulk(&self.pool, community_id, pubkeys).await + channel::get_users_bulk(self.pg_pool()?, community_id, pubkeys).await } /// Updates a channel's name and/or description. @@ -2337,7 +2393,7 @@ impl Db { channel_id: Uuid, updates: channel::ChannelUpdate, ) -> Result { - channel::update_channel(&self.pool, community_id, channel_id, updates).await + channel::update_channel(self.pg_pool()?, community_id, channel_id, updates).await } /// Sets the topic for a channel. @@ -2348,7 +2404,7 @@ impl Db { topic: &str, set_by: &[u8], ) -> Result<()> { - channel::set_topic(&self.pool, community_id, channel_id, topic, set_by).await + channel::set_topic(self.pg_pool()?, community_id, channel_id, topic, set_by).await } /// Sets the purpose for a channel. @@ -2359,12 +2415,12 @@ impl Db { purpose: &str, set_by: &[u8], ) -> Result<()> { - channel::set_purpose(&self.pool, community_id, channel_id, purpose, set_by).await + channel::set_purpose(self.pg_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.pool, community_id, channel_id).await + channel::archive_channel(self.pg_pool()?, community_id, channel_id).await } /// Unarchives a channel. @@ -2373,7 +2429,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result<()> { - channel::unarchive_channel(&self.pool, community_id, channel_id).await + channel::unarchive_channel(self.pg_pool()?, community_id, channel_id).await } /// Soft-delete a channel. @@ -2382,7 +2438,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result { - channel::soft_delete_channel(&self.pool, community_id, channel_id).await + channel::soft_delete_channel(self.pg_pool()?, community_id, channel_id).await } /// Returns the count of active members in a channel. @@ -2391,7 +2447,7 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result { - channel::get_member_count(&self.pool, community_id, channel_id).await + channel::get_member_count(self.pg_pool()?, community_id, channel_id).await } /// Bulk-fetch member counts for a set of channel IDs. @@ -2400,7 +2456,7 @@ impl Db { community_id: CommunityId, channel_ids: &[Uuid], ) -> Result> { - channel::get_member_counts_bulk(&self.pool, community_id, channel_ids).await + channel::get_member_counts_bulk(self.pg_pool()?, community_id, channel_ids).await } /// Get the active role of a pubkey in a channel. @@ -2410,14 +2466,14 @@ impl Db { channel_id: Uuid, pubkey: &[u8], ) -> Result> { - channel::get_member_role(&self.pool, community_id, channel_id, pubkey).await + channel::get_member_role(self.pg_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.pool).await + channel::reap_expired_ephemeral_channels(self.pg_pool()?).await } /// Query due reminders ready for delivery. @@ -2426,7 +2482,7 @@ impl Db { now_secs: i64, batch_limit: i64, ) -> Result> { - event::query_due_reminders(&self.pool, now_secs, batch_limit).await + event::query_due_reminders(self.pg_pool()?, now_secs, batch_limit).await } /// Atomically claim a due reminder for delivery (cross-pod dedup). @@ -2436,7 +2492,7 @@ impl Db { event_id: &[u8], event_created_at: chrono::DateTime, ) -> Result { - event::claim_due_reminder(&self.pool, community_id, event_id, event_created_at).await + event::claim_due_reminder(self.pg_pool()?, community_id, event_id, event_created_at).await } /// Atomically claim a due reminder using a caller-supplied delivery stamp. @@ -2448,7 +2504,7 @@ impl Db { delivery_stamp: i64, ) -> Result { event::claim_due_reminder_with_stamp( - &self.pool, + self.pg_pool()?, community_id, event_id, event_created_at, @@ -2466,7 +2522,7 @@ impl Db { delivery_stamp: i64, ) -> Result { event::release_due_reminder( - &self.pool, + self.pg_pool()?, community_id, event_id, event_created_at, @@ -2481,7 +2537,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 { - user::ensure_user(&self.pool, community_id, pubkey).await + user::ensure_user(self.pg_pool()?, community_id, pubkey).await } /// Get a single user record by pubkey. @@ -2490,7 +2546,7 @@ impl Db { community_id: CommunityId, pubkey: &[u8], ) -> Result> { - user::get_user(&self.pool, community_id, pubkey).await + user::get_user(self.pg_pool()?, community_id, pubkey).await } /// Update a user's profile fields. @@ -2504,7 +2560,7 @@ impl Db { nip05_handle: Option<&str>, ) -> Result<()> { user::update_user_profile( - &self.pool, + self.pg_pool()?, community_id, pubkey, display_name, @@ -2522,7 +2578,7 @@ impl Db { local_part: &str, domain: &str, ) -> Result> { - user::get_user_by_nip05(&self.pool, community_id, local_part, domain).await + user::get_user_by_nip05(self.pg_pool()?, community_id, local_part, domain).await } /// Search users by display name, NIP-05 handle, or pubkey prefix. @@ -2532,7 +2588,7 @@ impl Db { query: &str, limit: u32, ) -> Result> { - user::search_users(&self.pool, community_id, query, limit).await + user::search_users(self.pg_pool()?, community_id, query, limit).await } /// Atomically set agent owner — only if no owner is currently assigned. @@ -2543,7 +2599,7 @@ impl Db { agent_pubkey: &[u8], owner_pubkey: &[u8], ) -> Result { - user::set_agent_owner(&self.pool, community_id, agent_pubkey, owner_pubkey).await + user::set_agent_owner(self.pg_pool()?, community_id, agent_pubkey, owner_pubkey).await } /// Get the channel_add_policy and agent_owner_pubkey for a user. @@ -2552,7 +2608,7 @@ impl Db { community_id: CommunityId, pubkey: &[u8], ) -> Result>)>> { - user::get_agent_channel_policy(&self.pool, community_id, pubkey).await + user::get_agent_channel_policy(self.pg_pool()?, community_id, pubkey).await } /// Check whether `actor_pubkey` is the agent owner of `target_pubkey`. @@ -2562,7 +2618,7 @@ impl Db { target_pubkey: &[u8], actor_pubkey: &[u8], ) -> Result { - user::is_agent_owner(&self.pool, community_id, target_pubkey, actor_pubkey).await + user::is_agent_owner(self.pg_pool()?, community_id, target_pubkey, actor_pubkey).await } /// Set the channel_add_policy for a user. @@ -2572,7 +2628,7 @@ impl Db { pubkey: &[u8], policy: &str, ) -> Result<()> { - user::set_channel_add_policy(&self.pool, community_id, pubkey, policy).await + user::set_channel_add_policy(self.pg_pool()?, community_id, pubkey, policy).await } /// Find an existing DM by its participant hash. @@ -2581,7 +2637,7 @@ impl Db { community_id: CommunityId, participant_hash: &[u8], ) -> Result> { - dm::find_dm_by_participants(&self.pool, community_id, participant_hash).await + dm::find_dm_by_participants(self.pg_pool()?, community_id, participant_hash).await } /// Create or return an existing DM channel. @@ -2591,7 +2647,7 @@ impl Db { participants: &[&[u8]], created_by: &[u8], ) -> Result { - dm::create_dm(&self.pool, community_id, participants, created_by).await + dm::create_dm(self.pg_pool()?, community_id, participants, created_by).await } /// List all DMs for a user. @@ -2602,7 +2658,7 @@ impl Db { limit: u32, cursor: Option, ) -> Result> { - dm::list_dms_for_user(&self.pool, community_id, pubkey, limit, cursor).await + dm::list_dms_for_user(self.pg_pool()?, community_id, pubkey, limit, cursor).await } /// Open or retrieve a DM for the given participants. @@ -2612,7 +2668,7 @@ impl Db { pubkeys: &[&[u8]], created_by: &[u8], ) -> Result<(channel::ChannelRecord, bool)> { - dm::open_dm(&self.pool, community_id, pubkeys, created_by).await + dm::open_dm(self.pg_pool()?, community_id, pubkeys, created_by).await } /// Hide a DM channel for a specific user. @@ -2625,7 +2681,7 @@ impl Db { channel_id: Uuid, pubkey: &[u8], ) -> Result<()> { - dm::hide_dm(&self.pool, community_id, channel_id, pubkey).await + dm::hide_dm(self.pg_pool()?, community_id, channel_id, pubkey).await } /// Unhide a DM channel for a specific user. @@ -2635,7 +2691,7 @@ impl Db { channel_id: Uuid, pubkey: &[u8], ) -> Result<()> { - dm::unhide_dm(&self.pool, community_id, channel_id, pubkey).await + dm::unhide_dm(self.pg_pool()?, community_id, channel_id, pubkey).await } /// List the channel IDs of all DMs the given user currently has hidden. @@ -2644,7 +2700,7 @@ impl Db { community_id: CommunityId, pubkey: &[u8], ) -> Result> { - dm::list_hidden_dms(&self.pool, community_id, pubkey).await + dm::list_hidden_dms(self.pg_pool()?, community_id, pubkey).await } /// Insert thread metadata. @@ -2663,7 +2719,7 @@ impl Db { broadcast: bool, ) -> Result<()> { thread::insert_thread_metadata( - &self.pool, + self.pg_pool()?, community_id, event_id, event_created_at, @@ -2762,7 +2818,7 @@ impl Db { } } thread::get_thread_replies( - &self.pool, + self.pg_pool()?, community_id, root_event_id, depth_limit, @@ -2778,7 +2834,7 @@ impl Db { community_id: CommunityId, event_id: &[u8], ) -> Result> { - thread::get_thread_summary(&self.pool, community_id, event_id).await + thread::get_thread_summary(self.pg_pool()?, community_id, event_id).await } /// One channel window: top-level rows + summaries + server `has_more`. @@ -2862,7 +2918,7 @@ impl Db { ReadSession { inner: ReadSessionInner::Replica { tx, - writer: self.pool.clone(), + writer: self.pg_pool()?.clone(), }, }, )); @@ -2886,7 +2942,7 @@ impl Db { RouteDecision::Writer => {} } let window = thread::get_channel_window( - &self.pool, + self.pg_pool()?, community_id, channel_id, limit, @@ -2897,7 +2953,7 @@ impl Db { Ok(( window, ReadSession { - inner: ReadSessionInner::Writer(self.pool.clone()), + inner: ReadSessionInner::Writer(self.pg_pool()?.clone()), }, )) } @@ -2994,7 +3050,7 @@ impl Db { community_id: CommunityId, event_id: &[u8], ) -> Result> { - thread::get_thread_metadata_by_event(&self.pool, community_id, event_id).await + thread::get_thread_metadata_by_event(self.pg_pool()?, community_id, event_id).await } /// Decrement reply counts. @@ -3004,8 +3060,13 @@ impl Db { parent_event_id: &[u8], root_event_id: Option<&[u8]>, ) -> Result<()> { - thread::decrement_reply_count(&self.pool, community_id, parent_event_id, root_event_id) - .await + thread::decrement_reply_count( + self.pg_pool()?, + community_id, + parent_event_id, + root_event_id, + ) + .await } /// Add (or re-activate) a reaction. @@ -3019,7 +3080,7 @@ impl Db { reaction_event_id: Option<&[u8]>, ) -> Result { reaction::add_reaction( - &self.pool, + self.pg_pool()?, community, event_id, event_created_at, @@ -3040,7 +3101,7 @@ impl Db { emoji: &str, ) -> Result { reaction::remove_reaction( - &self.pool, + self.pg_pool()?, community, event_id, event_created_at, @@ -3056,7 +3117,8 @@ impl Db { community: CommunityId, reaction_event_id: &[u8], ) -> Result { - reaction::remove_reaction_by_source_event_id(&self.pool, community, reaction_event_id).await + reaction::remove_reaction_by_source_event_id(self.pg_pool()?, community, reaction_event_id) + .await } /// Look up the active reaction row for one actor + emoji + target tuple. @@ -3069,7 +3131,7 @@ impl Db { emoji: &str, ) -> Result> { reaction::get_active_reaction_record( - &self.pool, + self.pg_pool()?, community, event_id, event_created_at, @@ -3090,7 +3152,7 @@ impl Db { reaction_event_id: &[u8], ) -> Result { reaction::set_reaction_event_id( - &self.pool, + self.pg_pool()?, community, event_id, event_created_at, @@ -3111,7 +3173,7 @@ impl Db { cursor: Option<&str>, ) -> Result> { reaction::get_reactions( - &self.pool, + self.pg_pool()?, community, event_id, event_created_at, @@ -3127,7 +3189,7 @@ impl Db { community: CommunityId, event_ids: &[(&[u8], DateTime)], ) -> Result> { - reaction::get_reactions_bulk(&self.pool, community, event_ids).await + reaction::get_reactions_bulk(self.pg_pool()?, community, event_ids).await } /// Find events that @mention the given pubkey. @@ -3140,7 +3202,7 @@ impl Db { limit: i64, ) -> Result> { feed::query_mentions( - &self.pool, + self.pg_pool()?, community, pubkey_bytes, accessible_channel_ids, @@ -3186,7 +3248,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.pool, + self.pg_pool()?, community, pubkey_bytes, accessible_channel_ids, @@ -3199,7 +3261,7 @@ impl Db { } RouteDecision::Writer => { feed::query_mentions( - &self.pool, + self.pg_pool()?, community, pubkey_bytes, accessible_channel_ids, @@ -3221,7 +3283,7 @@ impl Db { limit: i64, ) -> Result> { feed::query_needs_action( - &self.pool, + self.pg_pool()?, community, pubkey_bytes, accessible_channel_ids, @@ -3263,7 +3325,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.pool, + self.pg_pool()?, community, pubkey_bytes, accessible_channel_ids, @@ -3276,7 +3338,7 @@ impl Db { } RouteDecision::Writer => { feed::query_needs_action( - &self.pool, + self.pg_pool()?, community, pubkey_bytes, accessible_channel_ids, @@ -3296,7 +3358,14 @@ impl Db { since: Option>, limit: i64, ) -> Result> { - feed::query_activity(&self.pool, community, accessible_channel_ids, since, limit).await + feed::query_activity( + self.pg_pool()?, + community, + accessible_channel_ids, + since, + limit, + ) + .await } /// [`Db::query_feed_activity`] with replica routing — BOUNDED arm only; @@ -3329,7 +3398,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.pool, + self.pg_pool()?, community, accessible_channel_ids, since, @@ -3340,8 +3409,14 @@ impl Db { } } RouteDecision::Writer => { - feed::query_activity(&self.pool, community, accessible_channel_ids, since, limit) - .await + feed::query_activity( + self.pg_pool()?, + community, + accessible_channel_ids, + since, + limit, + ) + .await } } } @@ -3359,7 +3434,7 @@ impl Db { expires_at: Option>, ) -> Result { api_token::create_api_token( - &self.pool, + self.pg_pool()?, *community_id.as_uuid(), token_hash, owner_pubkey, @@ -3384,7 +3459,7 @@ impl Db { expires_at: Option>, ) -> Result> { api_token::create_api_token_if_under_limit( - &self.pool, + self.pg_pool()?, *community_id.as_uuid(), token_hash, owner_pubkey, @@ -3417,7 +3492,7 @@ impl Db { ) .bind(community_id.as_uuid()) .bind(hash) - .fetch_optional(&self.pool) + .fetch_optional(self.pg_pool()?) .await?; match row { @@ -3433,7 +3508,7 @@ impl Db { hash: &[u8], ) -> Result> { api_token::get_api_token_by_hash_including_revoked( - &self.pool, + self.pg_pool()?, *community_id.as_uuid(), hash, ) @@ -3447,7 +3522,7 @@ impl Db { ) .bind(community_id.as_uuid()) .bind(hash) - .execute(&self.pool) + .execute(self.pg_pool()?) .await?; Ok(()) } @@ -3473,7 +3548,7 @@ impl Db { "#, ) .bind(community_id.as_uuid()) - .fetch_all(&self.pool) + .fetch_all(self.pg_pool()?) .await?; let mut out = Vec::with_capacity(rows.len()); @@ -3501,7 +3576,7 @@ impl Db { community_id: CommunityId, pubkey: &[u8], ) -> Result> { - api_token::list_tokens_by_owner(&self.pool, *community_id.as_uuid(), pubkey).await + api_token::list_tokens_by_owner(self.pg_pool()?, *community_id.as_uuid(), pubkey).await } /// Revoke a single token by ID, scoped to (community, owner). @@ -3513,7 +3588,7 @@ impl Db { revoked_by: &[u8], ) -> Result { api_token::revoke_token( - &self.pool, + self.pg_pool()?, *community_id.as_uuid(), id, owner_pubkey, @@ -3530,7 +3605,7 @@ impl Db { revoked_by: &[u8], ) -> Result { api_token::revoke_all_tokens( - &self.pool, + self.pg_pool()?, *community_id.as_uuid(), owner_pubkey, revoked_by, @@ -3549,7 +3624,7 @@ impl Db { definition_hash: &[u8], ) -> Result { workflow::create_workflow( - &self.pool, + self.pg_pool()?, community_id, channel_id, owner_pubkey, @@ -3573,7 +3648,7 @@ impl Db { definition_hash: &[u8], ) -> Result<()> { workflow::upsert_workflow( - &self.pool, + self.pg_pool()?, community_id, id, channel_id, @@ -3591,7 +3666,7 @@ impl Db { community_id: CommunityId, id: Uuid, ) -> Result { - workflow::get_workflow(&self.pool, community_id, id).await + workflow::get_workflow(self.pg_pool()?, community_id, id).await } /// List workflows for a channel. @@ -3602,7 +3677,8 @@ impl Db { limit: Option, offset: Option, ) -> Result> { - workflow::list_channel_workflows(&self.pool, community_id, channel_id, limit, offset).await + workflow::list_channel_workflows(self.pg_pool()?, community_id, channel_id, limit, offset) + .await } /// List active, enabled workflows for a channel. @@ -3611,12 +3687,12 @@ impl Db { community_id: CommunityId, channel_id: Uuid, ) -> Result> { - workflow::list_enabled_channel_workflows(&self.pool, community_id, channel_id).await + workflow::list_enabled_channel_workflows(self.pg_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.pool).await + workflow::list_all_enabled_workflows(self.pg_pool()?).await } /// Claim a scheduled workflow fire for an authoritative schedule instant. @@ -3634,7 +3710,7 @@ impl Db { scheduled_for: chrono::DateTime, ) -> Result> { workflow::claim_scheduled_workflow_fire( - &self.pool, + self.pg_pool()?, community_id, workflow_id, scheduled_for, @@ -3648,7 +3724,7 @@ impl Db { community_id: CommunityId, workflow_id: Uuid, ) -> Result>> { - workflow::latest_scheduled_workflow_fire(&self.pool, community_id, workflow_id).await + workflow::latest_scheduled_workflow_fire(self.pg_pool()?, community_id, workflow_id).await } /// Attach the workflow run id created from a won scheduled-fire claim. @@ -3660,7 +3736,7 @@ impl Db { workflow_run_id: Uuid, ) -> Result { workflow::attach_scheduled_workflow_run( - &self.pool, + self.pg_pool()?, community_id, workflow_id, scheduled_for, @@ -3674,7 +3750,7 @@ impl Db { &self, older_than: chrono::DateTime, ) -> Result { - workflow::prune_scheduled_workflow_fires_before(&self.pool, older_than).await + workflow::prune_scheduled_workflow_fires_before(self.pg_pool()?, older_than).await } /// Update a workflow's name, definition, and hash. @@ -3687,7 +3763,7 @@ impl Db { definition_hash: &[u8], ) -> Result<()> { workflow::update_workflow( - &self.pool, + self.pg_pool()?, community_id, id, name, @@ -3704,7 +3780,7 @@ impl Db { id: Uuid, status: workflow::WorkflowStatus, ) -> Result<()> { - workflow::update_workflow_status(&self.pool, community_id, id, status).await + workflow::update_workflow_status(self.pg_pool()?, community_id, id, status).await } /// Enable or disable a workflow. @@ -3714,7 +3790,7 @@ impl Db { id: Uuid, enabled: bool, ) -> Result<()> { - workflow::set_workflow_enabled(&self.pool, community_id, id, enabled).await + workflow::set_workflow_enabled(self.pg_pool()?, community_id, id, enabled).await } /// Disable all of an owner's workflows in a channel (SEC-006, on @@ -3726,7 +3802,7 @@ impl Db { owner_pubkey: &[u8], ) -> Result { workflow::disable_workflows_for_owner_in_channel( - &self.pool, + self.pg_pool()?, community_id, channel_id, owner_pubkey, @@ -3736,7 +3812,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.pool, community_id, id).await + workflow::delete_workflow(self.pg_pool()?, community_id, id).await } /// Delete a workflow only when it belongs to the provided owner. @@ -3747,7 +3823,7 @@ impl Db { id: Uuid, owner_pubkey: &[u8], ) -> Result> { - workflow::delete_workflow_for_owner(&self.pool, community_id, id, owner_pubkey).await + workflow::delete_workflow_for_owner(self.pg_pool()?, community_id, id, owner_pubkey).await } /// Find a workflow by owner pubkey and name within a community. Used for @@ -3758,7 +3834,7 @@ impl Db { owner_pubkey: &[u8], name: &str, ) -> Result> { - workflow::find_by_owner_and_name(&self.pool, community_id, owner_pubkey, name).await + workflow::find_by_owner_and_name(self.pg_pool()?, community_id, owner_pubkey, name).await } /// Create a new workflow run. @@ -3770,7 +3846,7 @@ impl Db { trigger_context: Option<&serde_json::Value>, ) -> Result { workflow::create_workflow_run( - &self.pool, + self.pg_pool()?, community_id, workflow_id, trigger_event_id, @@ -3785,7 +3861,7 @@ impl Db { community_id: CommunityId, id: Uuid, ) -> Result { - workflow::get_workflow_run(&self.pool, community_id, id).await + workflow::get_workflow_run(self.pg_pool()?, community_id, id).await } /// List runs for a workflow. @@ -3795,7 +3871,7 @@ impl Db { workflow_id: Uuid, limit: i64, ) -> Result> { - workflow::list_workflow_runs(&self.pool, community_id, workflow_id, limit).await + workflow::list_workflow_runs(self.pg_pool()?, community_id, workflow_id, limit).await } /// Update a workflow run's status. @@ -3809,7 +3885,7 @@ impl Db { error: Option<&str>, ) -> Result<()> { workflow::update_workflow_run( - &self.pool, + self.pg_pool()?, community_id, id, status, @@ -3822,7 +3898,7 @@ impl Db { /// Create an approval request. pub async fn create_approval(&self, params: workflow::CreateApprovalParams<'_>) -> Result<()> { - workflow::create_approval(&self.pool, params).await + workflow::create_approval(self.pg_pool()?, params).await } /// Fetch an approval by raw token. @@ -3831,7 +3907,7 @@ impl Db { community_id: CommunityId, token: &str, ) -> Result { - workflow::get_approval(&self.pool, community_id, token).await + workflow::get_approval(self.pg_pool()?, community_id, token).await } /// Fetch an approval by its already-hashed token (no re-hashing). @@ -3840,7 +3916,7 @@ impl Db { community_id: CommunityId, token_hash: &[u8], ) -> Result { - workflow::get_approval_by_stored_hash(&self.pool, community_id, token_hash).await + workflow::get_approval_by_stored_hash(self.pg_pool()?, community_id, token_hash).await } /// Fetch all approvals for a workflow run. @@ -3850,7 +3926,7 @@ impl Db { workflow_id: uuid::Uuid, run_id: uuid::Uuid, ) -> Result> { - workflow::get_run_approvals(&self.pool, community_id, workflow_id, run_id).await + workflow::get_run_approvals(self.pg_pool()?, community_id, workflow_id, run_id).await } /// Update an approval's status. @@ -3863,7 +3939,7 @@ impl Db { note: Option<&str>, ) -> Result { workflow::update_approval( - &self.pool, + self.pg_pool()?, community_id, token, status, @@ -3883,7 +3959,7 @@ impl Db { note: Option<&str>, ) -> Result { workflow::update_approval_by_stored_hash( - &self.pool, + self.pg_pool()?, community_id, token_hash, status, @@ -3895,7 +3971,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.pool, months_ahead).await + partition::ensure_future_partitions(self.pg_pool()?, months_ahead).await } /// Backfill `d_tag` for existing NIP-33 events (kind 30000–39999) that have `d_tag IS NULL`. @@ -3912,7 +3988,7 @@ impl Db { ) \ WHERE kind BETWEEN 30000 AND 39999 AND d_tag IS NULL", ) - .execute(&self.pool) + .execute(self.pg_pool()?) .await?; Ok(result.rows_affected()) } @@ -3924,7 +4000,7 @@ impl Db { ) .bind(community.as_uuid()) .bind(pubkey) - .fetch_one(&self.pool) + .fetch_one(self.pg_pool()?) .await?; let cnt: i64 = row.try_get("cnt")?; Ok(cnt > 0) @@ -3935,7 +4011,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.pool) + .fetch_one(self.pg_pool()?) .await?; let cnt: i64 = row.try_get("cnt")?; Ok(cnt > 0) @@ -3957,7 +4033,7 @@ impl Db { .bind(pubkey) .bind(added_by) .bind(note) - .execute(&self.pool) + .execute(self.pg_pool()?) .await?; Ok(result.rows_affected() > 0) } @@ -3972,7 +4048,7 @@ impl Db { sqlx::query("DELETE FROM pubkey_allowlist WHERE community_id = $1 AND pubkey = $2") .bind(community.as_uuid()) .bind(pubkey) - .execute(&self.pool) + .execute(self.pg_pool()?) .await?; Ok(result.rows_affected() > 0) } @@ -3983,7 +4059,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.pool) + .fetch_all(self.pg_pool()?) .await?; let mut out = Vec::with_capacity(rows.len()); @@ -4018,12 +4094,12 @@ impl Db { 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 + relay_members::is_relay_member(self.pg_pool()?, community, pubkey).await } } } RouteDecision::Writer => { - relay_members::is_relay_member(&self.pool, community, pubkey).await + relay_members::is_relay_member(self.pg_pool()?, community, pubkey).await } } } @@ -4034,7 +4110,7 @@ impl Db { community: CommunityId, pubkey: &str, ) -> Result> { - relay_members::get_relay_member(&self.pool, community, pubkey).await + relay_members::get_relay_member(self.pg_pool()?, community, pubkey).await } /// Returns all relay members of `community` ordered by `created_at` ascending. @@ -4042,7 +4118,7 @@ impl Db { &self, community: CommunityId, ) -> Result> { - relay_members::list_relay_members(&self.pool, community).await + relay_members::list_relay_members(self.pg_pool()?, community).await } /// Adds a new relay member to `community`. @@ -4056,7 +4132,7 @@ impl Db { role: &str, added_by: Option<&str>, ) -> Result { - relay_members::add_relay_member(&self.pool, community, pubkey, role, added_by).await + relay_members::add_relay_member(self.pg_pool()?, community, pubkey, role, added_by).await } /// Claims relay membership via an invite and atomically persists the @@ -4068,8 +4144,14 @@ impl Db { role: &str, policy_version: Option<&str>, ) -> Result { - relay_members::claim_relay_membership(&self.pool, community, pubkey, role, policy_version) - .await + relay_members::claim_relay_membership( + self.pg_pool()?, + community, + pubkey, + role, + policy_version, + ) + .await } /// Returns whether a member has persisted acceptance evidence for a policy version. @@ -4079,8 +4161,13 @@ impl Db { pubkey: &str, policy_version: &str, ) -> Result { - relay_members::has_join_policy_acceptance(&self.pool, community, pubkey, policy_version) - .await + relay_members::has_join_policy_acceptance( + self.pg_pool()?, + community, + pubkey, + policy_version, + ) + .await } /// Removes a relay member from `community` atomically, refusing to delete the owner. @@ -4089,7 +4176,7 @@ impl Db { community: CommunityId, pubkey: &str, ) -> Result { - relay_members::remove_relay_member(&self.pool, community, pubkey).await + relay_members::remove_relay_member(self.pg_pool()?, community, pubkey).await } /// Removes a relay member from `community` only if their current role matches `expected_role`. @@ -4102,8 +4189,13 @@ impl Db { pubkey: &str, expected_role: &str, ) -> Result { - relay_members::remove_relay_member_if_role(&self.pool, community, pubkey, expected_role) - .await + relay_members::remove_relay_member_if_role( + self.pg_pool()?, + community, + pubkey, + expected_role, + ) + .await } /// Updates the role of an existing relay member in `community`. Returns `true` if updated. @@ -4113,12 +4205,12 @@ impl Db { pubkey: &str, new_role: &str, ) -> Result { - relay_members::update_relay_member_role(&self.pool, community, pubkey, new_role).await + relay_members::update_relay_member_role(self.pg_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<()> { - relay_members::bootstrap_owner(&self.pool, community, owner_pubkey).await + relay_members::bootstrap_owner(self.pg_pool()?, community, owner_pubkey).await } /// Returns `true` if any member of `community` holds the `admin` or @@ -4138,7 +4230,7 @@ impl Db { expected_owner_pubkey: &str, ) -> Result { relay_members::transfer_ownership( - &self.pool, + self.pg_pool()?, community, new_owner_pubkey, expected_owner_pubkey, @@ -4151,7 +4243,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.pool, community).await + relay_members::backfill_from_allowlist(self.pg_pool()?, community).await } /// Mints a v2 use-limited relay invite. The plaintext code is returned @@ -4166,7 +4258,8 @@ impl Db { ttl_secs: u64, max_uses: Option, ) -> Result { - relay_invite::mint_relay_invite(&self.pool, community, created_by, ttl_secs, max_uses).await + relay_invite::mint_relay_invite(self.pg_pool()?, community, created_by, ttl_secs, max_uses) + .await } /// Delete one bounded batch of invites expired before `cutoff`. @@ -4174,7 +4267,7 @@ impl Db { &self, cutoff: chrono::DateTime, ) -> Result { - relay_invite::reap_expired_relay_invites(&self.pool, cutoff).await + relay_invite::reap_expired_relay_invites(self.pg_pool()?, cutoff).await } /// Atomically claims a v2 relay invite. The full redemption (membership @@ -4190,7 +4283,7 @@ impl Db { policy_version: Option<&str>, ) -> Result { relay_invite::claim_relay_invite( - &self.pool, + self.pg_pool()?, community, token_hash, claimer_pubkey, @@ -4205,7 +4298,7 @@ impl Db { community: CommunityId, feedback: product_feedback::NewProductFeedback<'_>, ) -> Result { - product_feedback::insert(&self.pool, community, feedback).await + product_feedback::insert(self.pg_pool()?, community, feedback).await } /// List product feedback across the deployment, newest first. @@ -4213,7 +4306,7 @@ impl Db { &self, limit: i64, ) -> Result> { - product_feedback::list(&self.pool, limit).await + product_feedback::list(self.pg_pool()?, limit).await } /// Insert a tenant-scoped NIP-56 report row, idempotent by report event id. @@ -4222,7 +4315,7 @@ impl Db { community: CommunityId, report: moderation::NewReport<'_>, ) -> Result { - moderation::insert_report(&self.pool, community, report).await + moderation::insert_report(self.pg_pool()?, community, report).await } /// List moderation reports for a community, newest first. @@ -4232,7 +4325,7 @@ impl Db { status: Option<&str>, limit: i64, ) -> Result> { - moderation::list_reports(&self.pool, community, status, limit).await + moderation::list_reports(self.pg_pool()?, community, status, limit).await } /// Fetch one moderation report by row id. @@ -4241,7 +4334,7 @@ impl Db { community: CommunityId, report_id: Uuid, ) -> Result> { - moderation::get_report(&self.pool, community, report_id).await + moderation::get_report(self.pg_pool()?, community, report_id).await } /// Fetch one moderation report by signed NIP-56 report event id. @@ -4250,7 +4343,7 @@ impl Db { community: CommunityId, report_event_id: &[u8], ) -> Result> { - moderation::get_report_by_event(&self.pool, community, report_event_id).await + moderation::get_report_by_event(self.pg_pool()?, community, report_event_id).await } /// Resolve, dismiss, or escalate an open moderation report. @@ -4263,7 +4356,7 @@ impl Db { action_id: Option, ) -> Result { moderation::resolve_report( - &self.pool, + self.pg_pool()?, community, report_id, status, @@ -4282,7 +4375,15 @@ impl Db { reason: Option<&str>, expires_at: Option>, ) -> Result<()> { - moderation::ban_member(&self.pool, community, pubkey, actor, reason, expires_at).await + moderation::ban_member( + self.pg_pool()?, + community, + pubkey, + actor, + reason, + expires_at, + ) + .await } /// Lift a community ban for a member pubkey. @@ -4292,7 +4393,7 @@ impl Db { pubkey: &[u8], actor: &[u8], ) -> Result { - moderation::unban_member(&self.pool, community, pubkey, actor).await + moderation::unban_member(self.pg_pool()?, community, pubkey, actor).await } /// Upsert a community timeout/write-block for a member pubkey. @@ -4304,7 +4405,15 @@ impl Db { muted_until: DateTime, reason: Option<&str>, ) -> Result<()> { - moderation::timeout_member(&self.pool, community, pubkey, actor, muted_until, reason).await + moderation::timeout_member( + self.pg_pool()?, + community, + pubkey, + actor, + muted_until, + reason, + ) + .await } /// Clear a community timeout/write-block for a member pubkey. @@ -4314,7 +4423,7 @@ impl Db { pubkey: &[u8], actor: &[u8], ) -> Result { - moderation::untimeout_member(&self.pool, community, pubkey, actor).await + moderation::untimeout_member(self.pg_pool()?, community, pubkey, actor).await } /// Fetch the active ban/timeout restriction state for enforcement hot paths. @@ -4323,7 +4432,7 @@ impl Db { community: CommunityId, pubkey: &[u8], ) -> Result { - moderation::restriction_state(&self.pool, community, pubkey).await + moderation::restriction_state(self.pg_pool()?, community, pubkey).await } /// Fetch the full ban/timeout row for a member pubkey. @@ -4332,7 +4441,7 @@ impl Db { community: CommunityId, pubkey: &[u8], ) -> Result> { - moderation::get_ban(&self.pool, community, pubkey).await + moderation::get_ban(self.pg_pool()?, community, pubkey).await } /// List currently restricted members in a community. @@ -4340,7 +4449,7 @@ impl Db { &self, community: CommunityId, ) -> Result> { - moderation::list_restricted(&self.pool, community).await + moderation::list_restricted(self.pg_pool()?, community).await } /// Insert a moderation audit action row. @@ -4349,7 +4458,7 @@ impl Db { community: CommunityId, action: moderation::NewAction<'_>, ) -> Result { - moderation::insert_action(&self.pool, community, action).await + moderation::insert_action(self.pg_pool()?, community, action).await } /// List moderation audit action rows, newest first. @@ -4358,7 +4467,7 @@ impl Db { community: CommunityId, limit: i64, ) -> Result> { - moderation::list_actions(&self.pool, community, limit).await + moderation::list_actions(self.pg_pool()?, community, limit).await } /// Return the current owner of git repo name `repo_id` in `community`, or @@ -4368,7 +4477,7 @@ impl Db { community: CommunityId, repo_id: &str, ) -> Result> { - git_repo::repo_name_owner(&self.pool, community, repo_id).await + git_repo::repo_name_owner(self.pg_pool()?, community, repo_id).await } /// Reserve a git repo name for `owner_pubkey` in `community` (NIP-34). @@ -4381,7 +4490,7 @@ impl Db { repo_id: &str, owner_pubkey: &str, ) -> Result { - git_repo::reserve_repo_name(&self.pool, community, repo_id, owner_pubkey).await + git_repo::reserve_repo_name(self.pg_pool()?, community, repo_id, owner_pubkey).await } /// Count git repos reserved by `owner_pubkey` in `community` (quota check). @@ -4390,7 +4499,7 @@ impl Db { community: CommunityId, owner_pubkey: &str, ) -> Result { - git_repo::count_repos_for_owner(&self.pool, community, owner_pubkey).await + git_repo::count_repos_for_owner(self.pg_pool()?, community, owner_pubkey).await } /// Release a git repo name reservation held by `owner_pubkey` (rollback). @@ -4402,12 +4511,12 @@ impl Db { repo_id: &str, owner_pubkey: &str, ) -> Result { - git_repo::release_repo_name(&self.pool, community, repo_id, owner_pubkey).await + git_repo::release_repo_name(self.pg_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.pool, community_id, pubkey).await + archived_identities::is_archived(self.pg_pool()?, community_id, pubkey).await } /// Archives an identity in `community_id`. Returns `true` if inserted, `false` if already archived. @@ -4423,7 +4532,7 @@ impl Db { request_event_id: &str, ) -> Result { archived_identities::archive( - &self.pool, + self.pg_pool()?, community_id, pubkey, consent_path, @@ -4437,7 +4546,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.pool, community_id, pubkey).await + archived_identities::unarchive(self.pg_pool()?, community_id, pubkey).await } /// Returns all identities archived in `community_id`, ordered by archive time ascending. @@ -4445,7 +4554,7 @@ impl Db { &self, community_id: CommunityId, ) -> Result> { - archived_identities::list_archived(&self.pool, community_id).await + archived_identities::list_archived(self.pg_pool()?, community_id).await } /// Soft-delete NIP-29 discovery events for a channel created by a specific relay pubkey. @@ -4462,7 +4571,7 @@ impl Db { .bind(community_id.as_uuid()) .bind(channel_id) .bind(relay_pubkey) - .execute(&self.pool) + .execute(self.pg_pool()?) .await?; Ok(result.rows_affected()) } @@ -4494,7 +4603,7 @@ impl Db { channel_id.as_ref().map(|id| id.as_bytes().as_slice()), ); - let mut tx = self.pool.begin().await?; + let mut tx = self.pg_pool()?.begin().await?; // Serialize all writers for the same (kind, pubkey, channel_id) tuple. // Advisory lock is transaction-scoped — released on commit/rollback. @@ -4590,7 +4699,9 @@ 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.pool, community_id, event, channel_id).await { + if let Err(e) = + crate::insert_mentions(self.pg_pool()?, community_id, event, channel_id).await + { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } @@ -4669,7 +4780,7 @@ impl Db { let lock_key = event_replacement_lock_key(community_id, kind_i32, pubkey_bytes.as_slice(), None); - let mut tx = self.pool.begin().await?; + let mut tx = self.pg_pool()?.begin().await?; // Acquire the per-community snapshot lock BEFORE reading members. // This serializes the entire read-build-write cycle: a concurrent @@ -4764,7 +4875,7 @@ impl Db { tx.commit().await?; - if let Err(e) = crate::insert_mentions(&self.pool, community_id, &event, None).await { + if let Err(e) = crate::insert_mentions(self.pg_pool()?, community_id, &event, None).await { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } @@ -4816,7 +4927,7 @@ impl Db { Some(d_tag.as_bytes()), ); - let mut tx = self.pool.begin().await?; + let mut tx = self.pg_pool()?.begin().await?; sqlx::query("SELECT pg_advisory_xact_lock($1)") .bind(lock_key) @@ -5001,7 +5112,9 @@ impl Db { tx.commit().await?; // Mentions are a denormalized index — safe outside the transaction. - if let Err(e) = crate::insert_mentions(&self.pool, community_id, event, channel_id).await { + if let Err(e) = + crate::insert_mentions(self.pg_pool()?, community_id, event, channel_id).await + { tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); } @@ -6617,7 +6730,7 @@ mod tests { let db = Db::from_pool(pool); assert!(!db.has_read_pool()); assert!( - std::ptr::eq(db.read(), &db.pool), + std::ptr::eq(db.read().expect("Postgres test DB"), &db.pool), "read() must be the writer pool when no replica is configured" ); assert!(db.read_pool_stats().is_none());