From 1a17b09485a12190a25a971607efd7276fb9771d Mon Sep 17 00:00:00 2001 From: David Grochowski Date: Wed, 29 Jul 2026 20:27:09 -0400 Subject: [PATCH] feat(tracing): add PostgreSQL tracing spans Trace logical database operations across the DB facade, search, audit, replica fencing, and command persistence without recording arguments or raw errors. Signed-off-by: David Grochowski Co-authored-by: Amp --- Cargo.lock | 1 + crates/buzz-audit/src/service.rs | 10 +- crates/buzz-db/src/event.rs | 6 +- crates/buzz-db/src/lib.rs | 241 ++++++++++++++++-- crates/buzz-db/src/replica_fence.rs | 5 + .../src/handlers/command_executor.rs | 6 + crates/buzz-search/Cargo.toml | 1 + crates/buzz-search/src/query.rs | 12 +- 8 files changed, 256 insertions(+), 26 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 49104b22d2..204bb8c763 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1231,6 +1231,7 @@ dependencies = [ "sqlx", "thiserror 2.0.18", "tokio", + "tracing", "uuid", ] diff --git a/crates/buzz-audit/src/service.rs b/crates/buzz-audit/src/service.rs index fa0e0fb443..85265b6899 100644 --- a/crates/buzz-audit/src/service.rs +++ b/crates/buzz-audit/src/service.rs @@ -49,7 +49,7 @@ impl AuditService { /// Serialized per-community via `pg_advisory_lock`. Postgres advisory locks /// are session-scoped, so we acquire before the transaction and release /// after commit (or on any error path). - #[instrument(skip(self, entry), fields(action = %entry.action))] + #[instrument(target = "buzz_datastore", name = "audit_log", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn log(&self, entry: NewAuditEntry) -> Result { let mut conn = self.pool.acquire().await?; @@ -125,7 +125,7 @@ impl AuditService { audit_entry.hash = compute_hash(&audit_entry)?.to_vec(); - debug!(seq, "writing audit entry"); + debug!("writing audit entry"); sqlx::query( r#" @@ -156,7 +156,7 @@ impl AuditService { /// Reads exactly that community's chain — it can never observe another /// community's entries or head. Returns `Ok(false)` if the range is empty, /// `Ok(true)` if the segment is internally consistent. - #[instrument(skip(self))] + #[instrument(target = "buzz_datastore", name = "audit_verify_chain", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn verify_chain( &self, community: CommunityId, @@ -208,7 +208,7 @@ impl AuditService { /// Returns up to `limit` entries from one community's chain starting at /// `from_seq`, ordered by sequence number. Scoped to `community` — never /// returns another community's rows. - #[instrument(skip(self))] + #[instrument(target = "buzz_datastore", name = "audit_get_entries", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_entries( &self, community: CommunityId, @@ -274,7 +274,7 @@ mod tests { async fn test_pool() -> Option { let url = std::env::var("DATABASE_URL") - .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".into()); + .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".into()); // sadscan:disable np.postgres.1 -- local test fixture PgPool::connect(&url).await.ok() } diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 0e54196d11..f45f091c9c 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -589,8 +589,8 @@ pub(crate) fn row_to_stored_event(row: sqlx::postgres::PgRow) -> Result e, - Err(e) => { - tracing::warn!("failed to reconstruct event from DB row: {e}"); + Err(_e) => { + tracing::warn!("failed to reconstruct event from DB row"); return Ok(None); } }; @@ -1506,7 +1506,7 @@ mod tests { use super::*; use nostr::{EventBuilder, Keys, Kind, Tag}; - const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test fixture async fn setup_pool() -> PgPool { let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 0c6ea36dac..c221a77615 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -130,11 +130,7 @@ pub async fn insert_mentions( .into_iter() .filter(|pk| { if pk.len() != 64 || !pk.chars().all(|c| c.is_ascii_hexdigit()) { - tracing::debug!( - event_id = %event.id, - invalid_ptag = pk, - "skipping malformed p-tag in insert_mentions" - ); + tracing::debug!("skipping malformed p-tag in insert_mentions"); false } else { true @@ -252,6 +248,7 @@ impl ReadSession { /// the degraded follow-up can only observe *more* than the proof-time /// snapshot, never less — fresher aux rows, the same failure semantics /// as a request that routed to the writer to begin with. + #[tracing::instrument(target = "buzz_datastore", name = "read_session_query_events", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn query_events(&mut self, q: &EventQuery) -> Result> { let degraded = match &mut self.inner { ReadSessionInner::Replica { tx, writer } => { @@ -491,6 +488,7 @@ impl UsageMetricsLeader { /// /// Bounded to 5 seconds — a blackholed connection (no RST) would otherwise /// stall the entire poller tick until the OS TCP timeout. + #[tracing::instrument(target = "buzz_datastore", name = "usage_metrics_leader_is_live", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn is_live(&mut self) -> bool { tokio::time::timeout(std::time::Duration::from_secs(5), self.connection.ping()) .await @@ -1009,6 +1007,7 @@ impl Db { } /// Run pending database migrations. + #[tracing::instrument(target = "buzz_datastore", name = "migrate", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn migrate(&self) -> Result<()> { migration::run_migrations(&self.pool).await } @@ -1053,6 +1052,7 @@ impl Db { /// detached from the shared pool so a stable leader neither returns a locked /// session to other callers nor permanently consumes a pool slot. Dropping the /// guard closes the connection and releases the session-scoped lock. + #[tracing::instrument(target = "buzz_datastore", name = "try_lock_usage_metrics", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn try_lock_usage_metrics( &self, lock_key: i64, @@ -1073,6 +1073,7 @@ impl Db { /// List reports for the deployment-global read-only admin plane. #[allow(clippy::too_many_arguments)] + #[tracing::instrument(target = "buzz_datastore", name = "admin_list_reports", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn admin_list_reports( &self, community_id: Option, @@ -1099,6 +1100,7 @@ impl Db { } /// Fetch one report for the deployment-global read-only admin plane. + #[tracing::instrument(target = "buzz_datastore", name = "admin_get_report", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn admin_get_report( &self, id: Uuid, @@ -1107,6 +1109,7 @@ impl Db { } /// List feedback for the deployment-global read-only admin plane. + #[tracing::instrument(target = "buzz_datastore", name = "admin_list_feedback", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn admin_list_feedback( &self, limit: i64, @@ -1115,6 +1118,7 @@ impl Db { } /// Fetch one feedback submission for the deployment-global admin plane. + #[tracing::instrument(target = "buzz_datastore", name = "admin_get_feedback", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn admin_get_feedback( &self, id: Uuid, @@ -1123,36 +1127,43 @@ impl Db { } /// Return total number of communities on this relay. + #[tracing::instrument(target = "buzz_datastore", name = "usage_community_count", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn usage_community_count(&self) -> Result { usage::community_count(&self.pool).await } /// Return per-community user counts split by human/agent. + #[tracing::instrument(target = "buzz_datastore", name = "usage_user_counts", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn usage_user_counts(&self) -> Result> { usage::user_counts(&self.pool).await } /// Return per-community channel counts by type. + #[tracing::instrument(target = "buzz_datastore", name = "usage_channel_counts", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn usage_channel_counts(&self) -> Result> { usage::channel_counts(&self.pool).await } /// Return per-community kind=9 message counts. + #[tracing::instrument(target = "buzz_datastore", name = "usage_message_counts", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn usage_message_counts(&self) -> Result> { usage::message_counts(&self.pool).await } /// Return per-community relay-member counts by role. + #[tracing::instrument(target = "buzz_datastore", name = "usage_relay_member_counts", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn usage_relay_member_counts(&self) -> Result> { usage::relay_member_counts(&self.pool).await } /// Return per-community workflow counts by status. + #[tracing::instrument(target = "buzz_datastore", name = "usage_workflow_counts", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn usage_workflow_counts(&self) -> Result> { usage::workflow_counts(&self.pool).await } /// Return per-community git-repo counts. + #[tracing::instrument(target = "buzz_datastore", name = "usage_git_repo_counts", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn usage_git_repo_counts(&self) -> Result> { usage::git_repo_counts(&self.pool).await } @@ -1160,6 +1171,7 @@ impl Db { /// Return per-community distinct active-user counts for a given SQL interval. /// /// `interval_sql` must be a trusted literal such as `"1 day"` or `"7 days"`. + #[tracing::instrument(target = "buzz_datastore", name = "usage_active_user_counts", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn usage_active_user_counts( &self, interval_sql: &'static str, @@ -1168,6 +1180,7 @@ impl Db { } /// Return per-community active-channel counts for a given SQL interval. + #[tracing::instrument(target = "buzz_datastore", name = "usage_active_channel_counts", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn usage_active_channel_counts( &self, interval_sql: &'static str, @@ -1176,6 +1189,7 @@ impl Db { } /// Return all community id → host mappings. + #[tracing::instrument(target = "buzz_datastore", name = "usage_community_hosts", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn usage_community_hosts(&self) -> Result> { usage::community_hosts(&self.pool).await } @@ -1192,6 +1206,7 @@ impl Db { /// /// The caller owns host normalization and turns `None` into the fail-closed /// request/connection error. buzz-db only reads the durable host map. + #[tracing::instrument(target = "buzz_datastore", name = "lookup_community_by_host", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn lookup_community_by_host( &self, normalized_host: &str, @@ -1221,6 +1236,7 @@ impl Db { } /// Returns whether a community id still exists in the active lifecycle state. + #[tracing::instrument(target = "buzz_datastore", name = "is_community_active", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn is_community_active(&self, community_id: CommunityId) -> Result { let active = sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM communities WHERE id = $1 AND archived_at IS NULL)", @@ -1232,6 +1248,7 @@ impl Db { } /// Returns a community by host regardless of lifecycle state. Operator-plane only. + #[tracing::instrument(target = "buzz_datastore", name = "lookup_community_by_host_for_management", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn lookup_community_by_host_for_management( &self, normalized_host: &str, @@ -1253,6 +1270,7 @@ impl Db { /// /// This is an operator-plane helper, not a tenant-scoped data-plane read: /// callers must gate it on deployment-level operator auth before exposing it. + #[tracing::instrument(target = "buzz_datastore", name = "list_communities_owned_by", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_communities_owned_by( &self, owner_pubkey: &str, @@ -1298,6 +1316,7 @@ impl Db { /// fan out under *that* community rather than the deployment default. The /// community is authoritative; the host is read back for labelling only and /// is never used to re-derive the community. + #[tracing::instrument(target = "buzz_datastore", name = "lookup_community_host", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn lookup_community_host(&self, community_id: CommunityId) -> Result> { let row = sqlx::query( r#" @@ -1322,6 +1341,7 @@ impl Db { /// /// Set by relay admins/owners via the kind:9033 command; the value is /// validated and size-capped at that write path. + #[tracing::instrument(target = "buzz_datastore", name = "get_community_icon", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_community_icon(&self, community_id: CommunityId) -> Result> { let row = sqlx::query( r#" @@ -1342,6 +1362,7 @@ impl Db { } /// Sets or clears (`None`) the community's workspace icon. + #[tracing::instrument(target = "buzz_datastore", name = "set_community_icon", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn set_community_icon( &self, community_id: CommunityId, @@ -1366,6 +1387,7 @@ impl Db { /// This is the startup/config seeding path for N=1 deployments. Migrations /// create the schema only; deployment-specific hosts are not hardcoded into /// schema history. + #[tracing::instrument(target = "buzz_datastore", name = "ensure_configured_community", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn ensure_configured_community( &self, normalized_host: &str, @@ -1398,6 +1420,7 @@ impl Db { /// Holds a per-owner advisory lock while enforcing the ownership limit. /// Identical create retries return the original record; host collisions and /// limit failures remain distinguishable to the operator API. + #[tracing::instrument(target = "buzz_datastore", name = "create_community_with_owner", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn create_community_with_owner( &self, normalized_host: &str, @@ -1483,6 +1506,7 @@ impl Db { } /// Idempotently archives a community when the asserted pubkey is its current owner. + #[tracing::instrument(target = "buzz_datastore", name = "archive_community_owned_by", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn archive_community_owned_by( &self, normalized_host: &str, @@ -1516,6 +1540,7 @@ impl Db { } /// Idempotently restores a community when the asserted pubkey is its current owner. + #[tracing::instrument(target = "buzz_datastore", name = "unarchive_community_owned_by", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn unarchive_community_owned_by( &self, normalized_host: &str, @@ -1548,6 +1573,7 @@ impl Db { /// /// Internal relay producers use this to derive tenant context from the row /// they are acting on, rather than falling back to an implicit default. + #[tracing::instrument(target = "buzz_datastore", name = "community_of_channel", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn community_of_channel(&self, channel_id: Uuid) -> Result> { let row = sqlx::query( r#" @@ -1586,6 +1612,7 @@ impl Db { /// are intentionally not present rather than mapped to a default — /// callers MUST treat "channel-id not in map" as a coverage breach, /// never as "use the resolved community". + #[tracing::instrument(target = "buzz_datastore", name = "communities_of_channels", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn communities_of_channels( &self, channel_ids: &[Uuid], @@ -1615,6 +1642,7 @@ impl Db { } /// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate. + #[tracing::instrument(target = "buzz_datastore", name = "insert_event", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn insert_event( &self, community_id: CommunityId, @@ -1623,8 +1651,8 @@ impl Db { ) -> Result<(StoredEvent, bool)> { let result = event::insert_event(&self.pool, community_id, event, channel_id).await?; if result.1 { - 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}"); + if let Err(_e) = insert_mentions(&self.pool, community_id, event, channel_id).await { + tracing::warn!("mention insertion failed"); } } Ok(result) @@ -1637,6 +1665,7 @@ impl Db { /// callers that tolerate bounded staleness should use /// [`Db::query_events_routed`] instead — converting a caller is an /// explicit, per-callsite decision, never a change to this method. + #[tracing::instrument(target = "buzz_datastore", name = "query_events", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn query_events(&self, q: &EventQuery) -> Result> { event::query_events(&self.pool, q).await } @@ -1657,6 +1686,7 @@ impl Db { /// unset, even covered-eligible queries stay on the writer, so merging /// this seam is a true no-op until the budget is configured. Every /// failure fails closed to the writer. + #[tracing::instrument(target = "buzz_datastore", name = "query_events_routed", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn query_events_routed( &self, path: &'static str, @@ -1691,6 +1721,7 @@ impl Db { /// display page absorbs that per-row; a number derived from the rows /// does not. Same classification-table requirement as /// [`Db::query_events_routed`]. + #[tracing::instrument(target = "buzz_datastore", name = "query_events_routed_bounded", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn query_events_routed_bounded( &self, path: &'static str, @@ -1718,6 +1749,7 @@ impl Db { /// /// Always reads from the WRITER pool — see [`Db::query_events`] for the /// writer-vs-routed rule. + #[tracing::instrument(target = "buzz_datastore", name = "count_events", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn count_events(&self, q: &EventQuery) -> Result { event::count_events(&self.pool, q).await } @@ -1732,6 +1764,7 @@ impl Db { /// inflated number for up to `FENCE_STALENESS` is a different product /// statement than a page briefly showing a deleted row. `Bounded` ties /// the error to the accepted budget `B`. + #[tracing::instrument(target = "buzz_datastore", name = "count_events_routed", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn count_events_routed(&self, path: &'static str, q: &EventQuery) -> Result { match self.route_read(path, RoutePredicate::Bounded).await { RouteDecision::Replica(mut tx, _entry, reason) => { @@ -1753,6 +1786,7 @@ impl Db { /// Return whether a creator-signed huddle-start event links a parent /// channel to an ephemeral huddle channel. + #[tracing::instrument(target = "buzz_datastore", name = "huddle_started_link_exists", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn huddle_started_link_exists( &self, community_id: CommunityId, @@ -1775,6 +1809,7 @@ impl Db { /// Uses canonical NIP-16 ordering: `created_at DESC, id ASC`. /// This matches the write path in [`replace_addressable_event`] and handles /// historical duplicate survivors correctly. + #[tracing::instrument(target = "buzz_datastore", name = "get_latest_global_replaceable", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_latest_global_replaceable( &self, community_id: CommunityId, @@ -1787,6 +1822,7 @@ impl Db { /// Fetches a single non-deleted event by its raw ID bytes. /// /// Returns `None` if the event does not exist or has been soft-deleted. + #[tracing::instrument(target = "buzz_datastore", name = "get_event_by_id", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_event_by_id( &self, community_id: CommunityId, @@ -1796,6 +1832,7 @@ impl Db { } /// Fetches a single event by its raw ID bytes, **including soft-deleted rows**. + #[tracing::instrument(target = "buzz_datastore", name = "get_event_by_id_including_deleted", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_event_by_id_including_deleted( &self, community_id: CommunityId, @@ -1805,6 +1842,7 @@ impl Db { } /// Soft-deletes an event. Returns `Ok(true)` if deleted, `Ok(false)` if already deleted. + #[tracing::instrument(target = "buzz_datastore", name = "soft_delete_event", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn soft_delete_event( &self, community_id: CommunityId, @@ -1815,6 +1853,7 @@ impl Db { /// Soft-delete the live row for an addressable coordinate `(kind, pubkey, d_tag)`. /// Used by NIP-09 a-tag deletion for parameterized-replaceable kinds. + #[tracing::instrument(target = "buzz_datastore", name = "soft_delete_by_coordinate", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn soft_delete_by_coordinate( &self, community_id: CommunityId, @@ -1826,6 +1865,7 @@ impl Db { } /// Atomically soft-delete an event and decrement thread reply counters. + #[tracing::instrument(target = "buzz_datastore", name = "soft_delete_event_and_update_thread", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn soft_delete_event_and_update_thread( &self, community_id: CommunityId, @@ -1844,6 +1884,7 @@ impl Db { } /// Returns the most recent `created_at` for a channel. + #[tracing::instrument(target = "buzz_datastore", name = "get_last_message_at", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_last_message_at( &self, community_id: CommunityId, @@ -1853,6 +1894,7 @@ impl Db { } /// Bulk-fetch the most recent `created_at` for a set of channel IDs. + #[tracing::instrument(target = "buzz_datastore", name = "get_last_message_at_bulk", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_last_message_at_bulk( &self, community_id: CommunityId, @@ -1862,6 +1904,7 @@ impl Db { } /// Batch-fetch non-deleted events by their raw IDs. + #[tracing::instrument(target = "buzz_datastore", name = "get_events_by_ids", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_events_by_ids( &self, community_id: CommunityId, @@ -1877,6 +1920,7 @@ impl Db { /// channel pin, so no fence floor can prove insert-completeness — the /// covered arm is structurally unavailable. Used for FTS hit hydration, /// where a missing row degrades to a skipped search hit downstream. + #[tracing::instrument(target = "buzz_datastore", name = "get_events_by_ids_routed", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_events_by_ids_routed( &self, path: &'static str, @@ -1902,6 +1946,7 @@ impl Db { } /// Exclusively claim a batch of due matcher jobs from one community. + #[tracing::instrument(target = "buzz_datastore", name = "claim_due_push_match_batch", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn claim_due_push_match_batch( &self, limit: i64, @@ -1911,6 +1956,7 @@ impl Db { } /// Load active endpoint-enabled leases eligible for push matching. + #[tracing::instrument(target = "buzz_datastore", name = "active_push_match_leases", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn active_push_match_leases( &self, community: CommunityId, @@ -1919,6 +1965,7 @@ impl Db { } /// Complete matcher jobs from one claimed batch while the fence holds. + #[tracing::instrument(target = "buzz_datastore", name = "complete_push_match_batch", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn complete_push_match_batch( &self, community: CommunityId, @@ -1929,6 +1976,7 @@ impl Db { } /// Release fenced matcher claims from one batch for retry. + #[tracing::instrument(target = "buzz_datastore", name = "retry_push_match_batch", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn retry_push_match_batch( &self, community: CommunityId, @@ -1940,11 +1988,13 @@ impl Db { } /// Delete exhausted matcher jobs (periodic sweep, off the claim path). + #[tracing::instrument(target = "buzz_datastore", name = "reap_exhausted_push_matches", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn reap_exhausted_push_matches(&self) -> Result { push::reap_exhausted_matches(&self.pool).await } /// Idempotently enqueue a wake for a matched lease and event. + #[tracing::instrument(target = "buzz_datastore", name = "enqueue_push_wake", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn enqueue_push_wake( &self, community: CommunityId, @@ -1956,6 +2006,7 @@ impl Db { } /// Set-wise [`Self::enqueue_push_wake`]: one transaction per batch. + #[tracing::instrument(target = "buzz_datastore", name = "enqueue_push_wakes", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn enqueue_push_wakes( &self, community: CommunityId, @@ -1965,6 +2016,7 @@ impl Db { } /// Exclusively claim due wake jobs for one community. + #[tracing::instrument(target = "buzz_datastore", name = "claim_due_push_wakes", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn claim_due_push_wakes( &self, community: CommunityId, @@ -1975,6 +2027,7 @@ impl Db { } /// Revalidate a wake's claim, source event, and current lease before send. + #[tracing::instrument(target = "buzz_datastore", name = "revalidate_push_wake", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn revalidate_push_wake( &self, community: CommunityId, @@ -1985,6 +2038,7 @@ impl Db { } /// Mark a fenced wake claim delivered. + #[tracing::instrument(target = "buzz_datastore", name = "complete_push_wake", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn complete_push_wake( &self, community: CommunityId, @@ -1995,6 +2049,7 @@ impl Db { } /// Release a fenced wake claim for retry at the supplied time. + #[tracing::instrument(target = "buzz_datastore", name = "retry_push_wake", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn retry_push_wake( &self, community: CommunityId, @@ -2006,6 +2061,7 @@ impl Db { } /// Mark a fenced wake claim terminally failed. + #[tracing::instrument(target = "buzz_datastore", name = "fail_push_wake", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn fail_push_wake( &self, community: CommunityId, @@ -2016,6 +2072,7 @@ impl Db { } /// Disable an endpoint only if the specified lease generation is current. + #[tracing::instrument(target = "buzz_datastore", name = "disable_push_endpoint", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn disable_push_endpoint( &self, community: CommunityId, @@ -2035,6 +2092,7 @@ impl Db { /// Atomically persist a validated kind:30350 event and its effective lease. #[allow(clippy::too_many_arguments)] + #[tracing::instrument(target = "buzz_datastore", name = "accept_push_lease_event", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn accept_push_lease_event( &self, community: CommunityId, @@ -2057,6 +2115,7 @@ impl Db { } /// Atomically insert an event AND its thread metadata in a single transaction. + #[tracing::instrument(target = "buzz_datastore", name = "insert_event_with_thread_metadata", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn insert_event_with_thread_metadata( &self, community_id: CommunityId, @@ -2073,8 +2132,8 @@ impl Db { ) .await?; if result.1 { - 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}"); + if let Err(_e) = insert_mentions(&self.pool, community_id, event, channel_id).await { + tracing::warn!("mention insertion failed"); } } Ok(result) @@ -2082,6 +2141,7 @@ impl Db { /// Atomically insert a kind:7 reaction event and its reaction row. #[allow(clippy::too_many_arguments)] + #[tracing::instrument(target = "buzz_datastore", name = "insert_reaction_event_with_thread_metadata", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn insert_reaction_event_with_thread_metadata( &self, community_id: CommunityId, @@ -2107,8 +2167,8 @@ impl Db { was_inserted: true, .. } = &outcome { - 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}"); + if let Err(_e) = insert_mentions(&self.pool, community_id, event, channel_id).await { + tracing::warn!("mention insertion failed"); } } Ok(outcome) @@ -2116,6 +2176,7 @@ impl Db { /// Creates a new channel, bootstraps the creator as owner, and returns the record. #[allow(clippy::too_many_arguments)] + #[tracing::instrument(target = "buzz_datastore", name = "create_channel", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn create_channel( &self, community_id: CommunityId, @@ -2143,6 +2204,7 @@ impl Db { /// /// Returns `(record, true)` if newly created, `(record, false)` if already exists. #[allow(clippy::too_many_arguments)] + #[tracing::instrument(target = "buzz_datastore", name = "create_channel_with_id", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn create_channel_with_id( &self, community_id: CommunityId, @@ -2169,6 +2231,7 @@ impl Db { } /// Fetches a channel record by ID. + #[tracing::instrument(target = "buzz_datastore", name = "get_channel", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_channel( &self, community_id: CommunityId, @@ -2178,6 +2241,7 @@ impl Db { } /// Returns the canvas content for a channel, if any. + #[tracing::instrument(target = "buzz_datastore", name = "get_canvas", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_canvas( &self, community_id: CommunityId, @@ -2187,6 +2251,7 @@ impl Db { } /// Sets or clears the canvas content for a channel. + #[tracing::instrument(target = "buzz_datastore", name = "set_canvas", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn set_canvas( &self, community_id: CommunityId, @@ -2197,6 +2262,7 @@ impl Db { } /// Adds a member to a channel. + #[tracing::instrument(target = "buzz_datastore", name = "add_member", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn add_member( &self, community_id: CommunityId, @@ -2217,6 +2283,7 @@ impl Db { } /// Removes a member from a channel. + #[tracing::instrument(target = "buzz_datastore", name = "remove_member", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn remove_member( &self, community_id: CommunityId, @@ -2228,6 +2295,7 @@ impl Db { } /// Returns `true` if the pubkey is an active member. + #[tracing::instrument(target = "buzz_datastore", name = "is_member", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn is_member( &self, community_id: CommunityId, @@ -2239,6 +2307,7 @@ impl Db { /// Return the active (channel, pubkey) membership pairs among the given /// sets, in one statement. + #[tracing::instrument(target = "buzz_datastore", name = "membership_pairs", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn membership_pairs( &self, community_id: CommunityId, @@ -2249,6 +2318,7 @@ impl Db { } /// Returns all active members of a channel. + #[tracing::instrument(target = "buzz_datastore", name = "get_members", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_members( &self, community_id: CommunityId, @@ -2258,6 +2328,7 @@ impl Db { } /// Returns active members for multiple channels in a single query. + #[tracing::instrument(target = "buzz_datastore", name = "get_members_bulk", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_members_bulk( &self, community_id: CommunityId, @@ -2267,6 +2338,7 @@ impl Db { } /// Get all channel IDs accessible to a pubkey. + #[tracing::instrument(target = "buzz_datastore", name = "get_accessible_channel_ids", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_accessible_channel_ids( &self, community_id: CommunityId, @@ -2276,6 +2348,7 @@ impl Db { } /// Lists channels, optionally filtered by visibility. + #[tracing::instrument(target = "buzz_datastore", name = "list_channels", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_channels( &self, community_id: CommunityId, @@ -2285,6 +2358,7 @@ impl Db { } /// Returns full channel records for all channels a user can access. + #[tracing::instrument(target = "buzz_datastore", name = "get_accessible_channels", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_accessible_channels( &self, community_id: CommunityId, @@ -2303,6 +2377,7 @@ impl Db { } /// Returns all bot-role members with their aggregated channel names in one community. + #[tracing::instrument(target = "buzz_datastore", name = "get_bot_members", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_bot_members( &self, community_id: CommunityId, @@ -2311,6 +2386,7 @@ impl Db { } /// Bulk-fetch user records by pubkey. + #[tracing::instrument(target = "buzz_datastore", name = "get_users_bulk", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_users_bulk( &self, community_id: CommunityId, @@ -2320,6 +2396,7 @@ impl Db { } /// Updates a channel's name and/or description. + #[tracing::instrument(target = "buzz_datastore", name = "update_channel", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn update_channel( &self, community_id: CommunityId, @@ -2330,6 +2407,7 @@ impl Db { } /// Sets the topic for a channel. + #[tracing::instrument(target = "buzz_datastore", name = "set_topic", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn set_topic( &self, community_id: CommunityId, @@ -2341,6 +2419,7 @@ impl Db { } /// Sets the purpose for a channel. + #[tracing::instrument(target = "buzz_datastore", name = "set_purpose", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn set_purpose( &self, community_id: CommunityId, @@ -2352,11 +2431,13 @@ impl Db { } /// Archives a channel. + #[tracing::instrument(target = "buzz_datastore", name = "archive_channel", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn archive_channel(&self, community_id: CommunityId, channel_id: Uuid) -> Result<()> { channel::archive_channel(&self.pool, community_id, channel_id).await } /// Unarchives a channel. + #[tracing::instrument(target = "buzz_datastore", name = "unarchive_channel", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn unarchive_channel( &self, community_id: CommunityId, @@ -2366,6 +2447,7 @@ impl Db { } /// Soft-delete a channel. + #[tracing::instrument(target = "buzz_datastore", name = "soft_delete_channel", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn soft_delete_channel( &self, community_id: CommunityId, @@ -2375,6 +2457,7 @@ impl Db { } /// Returns the count of active members in a channel. + #[tracing::instrument(target = "buzz_datastore", name = "get_member_count", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_member_count( &self, community_id: CommunityId, @@ -2384,6 +2467,7 @@ impl Db { } /// Bulk-fetch member counts for a set of channel IDs. + #[tracing::instrument(target = "buzz_datastore", name = "get_member_counts_bulk", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_member_counts_bulk( &self, community_id: CommunityId, @@ -2393,6 +2477,7 @@ impl Db { } /// Get the active role of a pubkey in a channel. + #[tracing::instrument(target = "buzz_datastore", name = "get_member_role", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_member_role( &self, community_id: CommunityId, @@ -2403,6 +2488,7 @@ impl Db { } /// Archive ephemeral channels whose TTL deadline has passed. + #[tracing::instrument(target = "buzz_datastore", name = "reap_expired_ephemeral_channels", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn reap_expired_ephemeral_channels( &self, ) -> Result> { @@ -2410,6 +2496,7 @@ impl Db { } /// Query due reminders ready for delivery. + #[tracing::instrument(target = "buzz_datastore", name = "query_due_reminders", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn query_due_reminders( &self, now_secs: i64, @@ -2419,6 +2506,7 @@ impl Db { } /// Atomically claim a due reminder for delivery (cross-pod dedup). + #[tracing::instrument(target = "buzz_datastore", name = "claim_due_reminder", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn claim_due_reminder( &self, community_id: CommunityId, @@ -2429,6 +2517,7 @@ impl Db { } /// Atomically claim a due reminder using a caller-supplied delivery stamp. + #[tracing::instrument(target = "buzz_datastore", name = "claim_due_reminder_with_stamp", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn claim_due_reminder_with_stamp( &self, community_id: CommunityId, @@ -2447,6 +2536,7 @@ impl Db { } /// Release a claimed due reminder after a publish failure. + #[tracing::instrument(target = "buzz_datastore", name = "release_due_reminder", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn release_due_reminder( &self, community_id: CommunityId, @@ -2469,11 +2559,13 @@ impl Db { /// Returns `true` if a new row was inserted (first time), `false` if it /// already existed. Callers use the `true` return to increment /// `buzz_users_created_total`. + #[tracing::instrument(target = "buzz_datastore", name = "ensure_user", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn ensure_user(&self, community_id: CommunityId, pubkey: &[u8]) -> Result { user::ensure_user(&self.pool, community_id, pubkey).await } /// Get a single user record by pubkey. + #[tracing::instrument(target = "buzz_datastore", name = "get_user", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_user( &self, community_id: CommunityId, @@ -2483,6 +2575,7 @@ impl Db { } /// Update a user's profile fields. + #[tracing::instrument(target = "buzz_datastore", name = "update_user_profile", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn update_user_profile( &self, community_id: CommunityId, @@ -2505,6 +2598,7 @@ impl Db { } /// Look up a user by NIP-05 handle. + #[tracing::instrument(target = "buzz_datastore", name = "get_user_by_nip05", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_user_by_nip05( &self, community_id: CommunityId, @@ -2515,6 +2609,7 @@ impl Db { } /// Search users by display name, NIP-05 handle, or pubkey prefix. + #[tracing::instrument(target = "buzz_datastore", name = "search_users", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn search_users( &self, community_id: CommunityId, @@ -2526,6 +2621,7 @@ impl Db { /// Atomically set agent owner — only if no owner is currently assigned. /// Returns Ok(true) if set, Ok(false) if an owner already exists. + #[tracing::instrument(target = "buzz_datastore", name = "set_agent_owner", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn set_agent_owner( &self, community_id: CommunityId, @@ -2536,6 +2632,7 @@ impl Db { } /// Get the channel_add_policy and agent_owner_pubkey for a user. + #[tracing::instrument(target = "buzz_datastore", name = "get_agent_channel_policy", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_agent_channel_policy( &self, community_id: CommunityId, @@ -2545,6 +2642,7 @@ impl Db { } /// Check whether `actor_pubkey` is the agent owner of `target_pubkey`. + #[tracing::instrument(target = "buzz_datastore", name = "is_agent_owner", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn is_agent_owner( &self, community_id: CommunityId, @@ -2555,6 +2653,7 @@ impl Db { } /// Set the channel_add_policy for a user. + #[tracing::instrument(target = "buzz_datastore", name = "set_channel_add_policy", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn set_channel_add_policy( &self, community_id: CommunityId, @@ -2565,6 +2664,7 @@ impl Db { } /// Find an existing DM by its participant hash. + #[tracing::instrument(target = "buzz_datastore", name = "find_dm_by_participants", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn find_dm_by_participants( &self, community_id: CommunityId, @@ -2574,6 +2674,7 @@ impl Db { } /// Create or return an existing DM channel. + #[tracing::instrument(target = "buzz_datastore", name = "create_dm", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn create_dm( &self, community_id: CommunityId, @@ -2584,6 +2685,7 @@ impl Db { } /// List all DMs for a user. + #[tracing::instrument(target = "buzz_datastore", name = "list_dms_for_user", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_dms_for_user( &self, community_id: CommunityId, @@ -2595,6 +2697,7 @@ impl Db { } /// Open or retrieve a DM for the given participants. + #[tracing::instrument(target = "buzz_datastore", name = "open_dm", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn open_dm( &self, community_id: CommunityId, @@ -2608,6 +2711,7 @@ impl Db { /// /// The DM is not deleted — it can be restored by opening a new DM with /// the same participants. + #[tracing::instrument(target = "buzz_datastore", name = "hide_dm", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn hide_dm( &self, community_id: CommunityId, @@ -2618,6 +2722,7 @@ impl Db { } /// Unhide a DM channel for a specific user. + #[tracing::instrument(target = "buzz_datastore", name = "unhide_dm", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn unhide_dm( &self, community_id: CommunityId, @@ -2628,6 +2733,7 @@ impl Db { } /// List the channel IDs of all DMs the given user currently has hidden. + #[tracing::instrument(target = "buzz_datastore", name = "list_hidden_dms", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_hidden_dms( &self, community_id: CommunityId, @@ -2638,6 +2744,7 @@ impl Db { /// Insert thread metadata. #[allow(clippy::too_many_arguments)] + #[tracing::instrument(target = "buzz_datastore", name = "insert_thread_metadata", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn insert_thread_metadata( &self, community_id: CommunityId, @@ -2688,6 +2795,7 @@ impl Db { /// A head fetch routed under Predicate A skips the re-run: bounded /// staleness (missing at most the freshest budget-window of replies) is /// exactly the semantic the head gate accepts. + #[tracing::instrument(target = "buzz_datastore", name = "get_thread_replies", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_thread_replies( &self, community_id: CommunityId, @@ -2762,6 +2870,7 @@ impl Db { } /// Fetch aggregated thread stats. + #[tracing::instrument(target = "buzz_datastore", name = "get_thread_summary", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_thread_summary( &self, community_id: CommunityId, @@ -2813,6 +2922,7 @@ impl Db { /// /// Every failure fails closed to the writer and is recorded in /// `buzz_db_route_decision`. + #[tracing::instrument(target = "buzz_datastore", name = "get_channel_window", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_channel_window_with_session( &self, community_id: CommunityId, @@ -2978,6 +3088,7 @@ impl Db { } /// Look up a single thread_metadata row by event_id. + #[tracing::instrument(target = "buzz_datastore", name = "get_thread_metadata_by_event", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_thread_metadata_by_event( &self, community_id: CommunityId, @@ -2987,6 +3098,7 @@ impl Db { } /// Decrement reply counts. + #[tracing::instrument(target = "buzz_datastore", name = "decrement_reply_count", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn decrement_reply_count( &self, community_id: CommunityId, @@ -2998,6 +3110,7 @@ impl Db { } /// Add (or re-activate) a reaction. + #[tracing::instrument(target = "buzz_datastore", name = "add_reaction", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn add_reaction( &self, community: CommunityId, @@ -3020,6 +3133,7 @@ impl Db { } /// Soft-delete a reaction. + #[tracing::instrument(target = "buzz_datastore", name = "remove_reaction", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn remove_reaction( &self, community: CommunityId, @@ -3040,6 +3154,7 @@ impl Db { } /// Soft-delete a reaction by its source event ID. + #[tracing::instrument(target = "buzz_datastore", name = "remove_reaction_by_source_event_id", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn remove_reaction_by_source_event_id( &self, community: CommunityId, @@ -3049,6 +3164,7 @@ impl Db { } /// Look up the active reaction row for one actor + emoji + target tuple. + #[tracing::instrument(target = "buzz_datastore", name = "get_active_reaction_record", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_active_reaction_record( &self, community: CommunityId, @@ -3069,6 +3185,7 @@ impl Db { } /// Backfill the source event ID on an active reaction row. + #[tracing::instrument(target = "buzz_datastore", name = "set_reaction_event_id", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn set_reaction_event_id( &self, community: CommunityId, @@ -3091,6 +3208,7 @@ impl Db { } /// Get all active reactions for an event, grouped by emoji. + #[tracing::instrument(target = "buzz_datastore", name = "get_reactions", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_reactions( &self, community: CommunityId, @@ -3111,6 +3229,7 @@ impl Db { } /// Batch-fetch emoji counts for a set of (event_id, event_created_at) pairs. + #[tracing::instrument(target = "buzz_datastore", name = "get_reactions_bulk", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_reactions_bulk( &self, community: CommunityId, @@ -3120,6 +3239,7 @@ impl Db { } /// Find events that @mention the given pubkey. + #[tracing::instrument(target = "buzz_datastore", name = "query_feed_mentions", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn query_feed_mentions( &self, community: CommunityId, @@ -3146,6 +3266,7 @@ impl Db { /// parameter admits community-global rows alongside channel rows, so no /// single channel's fence floor can prove completeness — the covered arm /// is structurally unavailable, not merely unchosen. + #[tracing::instrument(target = "buzz_datastore", name = "query_feed_mentions_routed", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn query_feed_mentions_routed( &self, path: &'static str, @@ -3201,6 +3322,7 @@ impl Db { } /// Find events that require action from the given pubkey. + #[tracing::instrument(target = "buzz_datastore", name = "query_feed_needs_action", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn query_feed_needs_action( &self, community: CommunityId, @@ -3223,6 +3345,7 @@ impl Db { /// [`Db::query_feed_needs_action`] with replica routing — BOUNDED arm /// only; see [`Db::query_feed_mentions_routed`] for why the covered arm /// is structurally unavailable to feed queries. + #[tracing::instrument(target = "buzz_datastore", name = "query_feed_needs_action_routed", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn query_feed_needs_action_routed( &self, path: &'static str, @@ -3278,6 +3401,7 @@ impl Db { } /// Find recent activity across accessible channels. + #[tracing::instrument(target = "buzz_datastore", name = "query_feed_activity", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn query_feed_activity( &self, community: CommunityId, @@ -3291,6 +3415,7 @@ impl Db { /// [`Db::query_feed_activity`] with replica routing — BOUNDED arm only; /// see [`Db::query_feed_mentions_routed`] for why the covered arm is /// structurally unavailable to feed queries. + #[tracing::instrument(target = "buzz_datastore", name = "query_feed_activity_routed", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn query_feed_activity_routed( &self, path: &'static str, @@ -3337,6 +3462,7 @@ impl Db { /// Create a new API token record. #[allow(clippy::too_many_arguments)] + #[tracing::instrument(target = "buzz_datastore", name = "create_api_token", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn create_api_token( &self, community_id: CommunityId, @@ -3362,6 +3488,7 @@ impl Db { /// Atomic conditional INSERT with 10-token limit (per (community, owner)). #[allow(clippy::too_many_arguments)] + #[tracing::instrument(target = "buzz_datastore", name = "create_api_token_if_under_limit", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn create_api_token_if_under_limit( &self, community_id: CommunityId, @@ -3391,6 +3518,7 @@ impl Db { /// See [`api_token::get_api_token_by_hash_including_revoked`] for the /// row-44 conformance rationale — the `(community_id, token_hash)` key /// is enforced both by the storage UNIQUE index and by this WHERE clause. + #[tracing::instrument(target = "buzz_datastore", name = "get_api_token_by_hash", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_api_token_by_hash( &self, community_id: CommunityId, @@ -3416,6 +3544,7 @@ impl Db { } /// Look up an API token by hash, including revoked, scoped to community. + #[tracing::instrument(target = "buzz_datastore", name = "get_api_token_by_hash_including_revoked", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_api_token_by_hash_including_revoked( &self, community_id: CommunityId, @@ -3430,6 +3559,7 @@ impl Db { } /// Record a token usage (update `last_used_at`), scoped to community. + #[tracing::instrument(target = "buzz_datastore", name = "touch_api_token", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn touch_api_token(&self, community_id: CommunityId, hash: &[u8]) -> Result<()> { sqlx::query( "UPDATE api_tokens SET last_used_at = NOW() WHERE community_id = $1 AND token_hash = $2", @@ -3451,6 +3581,7 @@ impl Db { } /// List all active (non-revoked) tokens in a community, newest first. + #[tracing::instrument(target = "buzz_datastore", name = "list_active_tokens", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_active_tokens(&self, community_id: CommunityId) -> Result> { let rows = sqlx::query( r#" @@ -3485,6 +3616,7 @@ impl Db { } /// List all tokens for a (community, owner) pair (including revoked). + #[tracing::instrument(target = "buzz_datastore", name = "list_tokens_by_owner", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_tokens_by_owner( &self, community_id: CommunityId, @@ -3494,6 +3626,7 @@ impl Db { } /// Revoke a single token by ID, scoped to (community, owner). + #[tracing::instrument(target = "buzz_datastore", name = "revoke_token", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn revoke_token( &self, community_id: CommunityId, @@ -3512,6 +3645,7 @@ impl Db { } /// Revoke all active tokens for a (community, owner) pair. + #[tracing::instrument(target = "buzz_datastore", name = "revoke_all_tokens", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn revoke_all_tokens( &self, community_id: CommunityId, @@ -3528,6 +3662,7 @@ impl Db { } /// Create a new workflow. + #[tracing::instrument(target = "buzz_datastore", name = "create_workflow", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn create_workflow( &self, community_id: CommunityId, @@ -3551,6 +3686,7 @@ impl Db { /// Insert or update a workflow using its NIP-33 `d`-tag UUID. #[allow(clippy::too_many_arguments)] + #[tracing::instrument(target = "buzz_datastore", name = "upsert_workflow", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn upsert_workflow( &self, community_id: CommunityId, @@ -3575,6 +3711,7 @@ impl Db { } /// Fetch a single workflow by ID, scoped to its community. + #[tracing::instrument(target = "buzz_datastore", name = "get_workflow", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_workflow( &self, community_id: CommunityId, @@ -3584,6 +3721,7 @@ impl Db { } /// List workflows for a channel. + #[tracing::instrument(target = "buzz_datastore", name = "list_channel_workflows", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_channel_workflows( &self, community_id: CommunityId, @@ -3595,6 +3733,7 @@ impl Db { } /// List active, enabled workflows for a channel. + #[tracing::instrument(target = "buzz_datastore", name = "list_enabled_channel_workflows", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_enabled_channel_workflows( &self, community_id: CommunityId, @@ -3604,6 +3743,7 @@ impl Db { } /// List all active, enabled schedule-triggered workflows. + #[tracing::instrument(target = "buzz_datastore", name = "list_all_enabled_workflows", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_all_enabled_workflows(&self) -> Result> { workflow::list_all_enabled_workflows(&self.pool).await } @@ -3616,6 +3756,7 @@ impl Db { /// from the scheduler scan), never client-supplied — `workflows` is keyed /// `(community_id, id)`, so the claim must bind both to avoid fanning /// across communities that share the workflow UUID. + #[tracing::instrument(target = "buzz_datastore", name = "claim_scheduled_workflow_fire", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn claim_scheduled_workflow_fire( &self, community_id: CommunityId, @@ -3632,6 +3773,7 @@ impl Db { } /// Fetch the latest claimed schedule instant for interval trigger anchoring. + #[tracing::instrument(target = "buzz_datastore", name = "latest_scheduled_workflow_fire", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn latest_scheduled_workflow_fire( &self, community_id: CommunityId, @@ -3641,6 +3783,7 @@ impl Db { } /// Attach the workflow run id created from a won scheduled-fire claim. + #[tracing::instrument(target = "buzz_datastore", name = "attach_scheduled_workflow_run", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn attach_scheduled_workflow_run( &self, community_id: CommunityId, @@ -3659,6 +3802,7 @@ impl Db { } /// Delete old scheduled workflow fire claims before a retention cutoff. + #[tracing::instrument(target = "buzz_datastore", name = "prune_scheduled_workflow_fires_before", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn prune_scheduled_workflow_fires_before( &self, older_than: chrono::DateTime, @@ -3667,6 +3811,7 @@ impl Db { } /// Update a workflow's name, definition, and hash. + #[tracing::instrument(target = "buzz_datastore", name = "update_workflow", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn update_workflow( &self, community_id: CommunityId, @@ -3687,6 +3832,7 @@ impl Db { } /// Update a workflow's status. + #[tracing::instrument(target = "buzz_datastore", name = "update_workflow_status", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn update_workflow_status( &self, community_id: CommunityId, @@ -3697,6 +3843,7 @@ impl Db { } /// Enable or disable a workflow. + #[tracing::instrument(target = "buzz_datastore", name = "set_workflow_enabled", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn set_workflow_enabled( &self, community_id: CommunityId, @@ -3724,12 +3871,14 @@ impl Db { } /// Delete a workflow and all its runs/approvals. + #[tracing::instrument(target = "buzz_datastore", name = "delete_workflow", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn delete_workflow(&self, community_id: CommunityId, id: Uuid) -> Result<()> { workflow::delete_workflow(&self.pool, community_id, id).await } /// Delete a workflow only when it belongs to the provided owner. /// Returns the deleted workflow's `channel_id`. + #[tracing::instrument(target = "buzz_datastore", name = "delete_workflow_for_owner", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn delete_workflow_for_owner( &self, community_id: CommunityId, @@ -3741,6 +3890,7 @@ impl Db { /// Find a workflow by owner pubkey and name within a community. Used for /// NIP-09 a-tag deletion where the d-tag is the workflow name (not UUID). + #[tracing::instrument(target = "buzz_datastore", name = "find_workflow_by_owner_and_name", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn find_workflow_by_owner_and_name( &self, community_id: CommunityId, @@ -3751,6 +3901,7 @@ impl Db { } /// Create a new workflow run. + #[tracing::instrument(target = "buzz_datastore", name = "create_workflow_run", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn create_workflow_run( &self, community_id: CommunityId, @@ -3769,6 +3920,7 @@ impl Db { } /// Fetch a single workflow run, scoped to its community. + #[tracing::instrument(target = "buzz_datastore", name = "get_workflow_run", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_workflow_run( &self, community_id: CommunityId, @@ -3778,6 +3930,7 @@ impl Db { } /// List runs for a workflow. + #[tracing::instrument(target = "buzz_datastore", name = "list_workflow_runs", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_workflow_runs( &self, community_id: CommunityId, @@ -3788,6 +3941,7 @@ impl Db { } /// Update a workflow run's status. + #[tracing::instrument(target = "buzz_datastore", name = "update_workflow_run", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn update_workflow_run( &self, community_id: CommunityId, @@ -3810,11 +3964,13 @@ impl Db { } /// Create an approval request. + #[tracing::instrument(target = "buzz_datastore", name = "create_approval", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn create_approval(&self, params: workflow::CreateApprovalParams<'_>) -> Result<()> { workflow::create_approval(&self.pool, params).await } /// Fetch an approval by raw token. + #[tracing::instrument(target = "buzz_datastore", name = "get_approval", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_approval( &self, community_id: CommunityId, @@ -3824,6 +3980,7 @@ impl Db { } /// Fetch an approval by its already-hashed token (no re-hashing). + #[tracing::instrument(target = "buzz_datastore", name = "get_approval_by_stored_hash", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_approval_by_stored_hash( &self, community_id: CommunityId, @@ -3833,6 +3990,7 @@ impl Db { } /// Fetch all approvals for a workflow run. + #[tracing::instrument(target = "buzz_datastore", name = "get_run_approvals", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_run_approvals( &self, community_id: CommunityId, @@ -3843,6 +4001,7 @@ impl Db { } /// Update an approval's status. + #[tracing::instrument(target = "buzz_datastore", name = "update_approval", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn update_approval( &self, community_id: CommunityId, @@ -3863,6 +4022,7 @@ impl Db { } /// Update an approval by its already-hashed token (no re-hashing). + #[tracing::instrument(target = "buzz_datastore", name = "update_approval_by_stored_hash", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn update_approval_by_stored_hash( &self, community_id: CommunityId, @@ -3883,6 +4043,7 @@ impl Db { } /// Ensures monthly partitions exist for the next N months. + #[tracing::instrument(target = "buzz_datastore", name = "ensure_future_partitions", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn ensure_future_partitions(&self, months_ahead: u32) -> Result<()> { partition::ensure_future_partitions(&self.pool, months_ahead).await } @@ -3891,6 +4052,7 @@ impl Db { /// /// Idempotent — safe to call on every startup. No-ops when all rows are already populated. /// Runs a single UPDATE touching only NIP-33 rows with NULL d_tag. + #[tracing::instrument(target = "buzz_datastore", name = "backfill_d_tags", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn backfill_d_tags(&self) -> Result { let result = sqlx::query( "UPDATE events \ @@ -3907,6 +4069,7 @@ impl Db { } /// Check if a pubkey is in the allowlist for `community`. + #[tracing::instrument(target = "buzz_datastore", name = "is_pubkey_allowed", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn is_pubkey_allowed(&self, community: CommunityId, pubkey: &[u8]) -> Result { let row = sqlx::query( "SELECT COUNT(*) as cnt FROM pubkey_allowlist WHERE community_id = $1 AND pubkey = $2", @@ -3920,6 +4083,7 @@ impl Db { } /// Check if the community allowlist has any entries (i.e. is enforcement active). + #[tracing::instrument(target = "buzz_datastore", name = "has_allowlist_entries", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn has_allowlist_entries(&self, community: CommunityId) -> Result { let row = sqlx::query("SELECT COUNT(*) as cnt FROM pubkey_allowlist WHERE community_id = $1") @@ -3931,6 +4095,7 @@ impl Db { } /// Add a pubkey to the community allowlist. + #[tracing::instrument(target = "buzz_datastore", name = "add_to_allowlist", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn add_to_allowlist( &self, community: CommunityId, @@ -3952,6 +4117,7 @@ impl Db { } /// Remove a pubkey from the community allowlist. + #[tracing::instrument(target = "buzz_datastore", name = "remove_from_allowlist", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn remove_from_allowlist( &self, community: CommunityId, @@ -3967,6 +4133,7 @@ impl Db { } /// List all pubkeys in the community allowlist. + #[tracing::instrument(target = "buzz_datastore", name = "list_allowlist", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_allowlist(&self, community: CommunityId) -> Result> { let rows = sqlx::query( "SELECT pubkey, added_by, added_at, note FROM pubkey_allowlist WHERE community_id = $1 ORDER BY added_at DESC", @@ -3988,11 +4155,13 @@ impl Db { } /// Returns `true` if `pubkey` (64-char hex) is a member of `community`. + #[tracing::instrument(target = "buzz_datastore", name = "is_relay_member", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn is_relay_member(&self, community: CommunityId, pubkey: &str) -> Result { relay_members::is_relay_member(&self.pool, community, pubkey).await } /// Returns the relay member record for `pubkey` in `community`, or `None` if not found. + #[tracing::instrument(target = "buzz_datastore", name = "get_relay_member", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_relay_member( &self, community: CommunityId, @@ -4002,6 +4171,7 @@ impl Db { } /// Returns all relay members of `community` ordered by `created_at` ascending. + #[tracing::instrument(target = "buzz_datastore", name = "list_relay_members", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_relay_members( &self, community: CommunityId, @@ -4013,6 +4183,7 @@ impl Db { /// /// Returns `true` if the row was actually inserted, `false` if the pubkey /// already existed in `community` (idempotent — `ON CONFLICT DO NOTHING`). + #[tracing::instrument(target = "buzz_datastore", name = "add_relay_member", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn add_relay_member( &self, community: CommunityId, @@ -4025,6 +4196,7 @@ impl Db { /// Claims relay membership via an invite and atomically persists the /// accepted policy version when a policy is configured. + #[tracing::instrument(target = "buzz_datastore", name = "claim_relay_membership", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn claim_relay_membership( &self, community: CommunityId, @@ -4037,6 +4209,7 @@ impl Db { } /// Returns whether a member has persisted acceptance evidence for a policy version. + #[tracing::instrument(target = "buzz_datastore", name = "has_join_policy_acceptance", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn has_join_policy_acceptance( &self, community: CommunityId, @@ -4048,6 +4221,7 @@ impl Db { } /// Removes a relay member from `community` atomically, refusing to delete the owner. + #[tracing::instrument(target = "buzz_datastore", name = "remove_relay_member", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn remove_relay_member( &self, community: CommunityId, @@ -4060,6 +4234,7 @@ impl Db { /// /// Atomic conditional delete — eliminates the TOCTOU race between a /// prior role read and the delete. See [`relay_members::remove_relay_member_if_role`]. + #[tracing::instrument(target = "buzz_datastore", name = "remove_relay_member_if_role", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn remove_relay_member_if_role( &self, community: CommunityId, @@ -4071,6 +4246,7 @@ impl Db { } /// Updates the role of an existing relay member in `community`. Returns `true` if updated. + #[tracing::instrument(target = "buzz_datastore", name = "update_relay_member_role", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn update_relay_member_role( &self, community: CommunityId, @@ -4081,6 +4257,7 @@ impl Db { } /// Ensures the owner pubkey exists with role `"owner"` in `community`. Called at startup. + #[tracing::instrument(target = "buzz_datastore", name = "bootstrap_owner", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn bootstrap_owner(&self, community: CommunityId, owner_pubkey: &str) -> Result<()> { relay_members::bootstrap_owner(&self.pool, community, owner_pubkey).await } @@ -4089,6 +4266,7 @@ impl Db { /// demoting the previous owner(s) to `member`. Verifies /// `expected_owner_pubkey` matches the current owner inside the same /// transaction to prevent stale-owner races. + #[tracing::instrument(target = "buzz_datastore", name = "transfer_ownership", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn transfer_ownership( &self, community: CommunityId, @@ -4108,6 +4286,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. + #[tracing::instrument(target = "buzz_datastore", name = "backfill_from_allowlist", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn backfill_from_allowlist(&self, community: CommunityId) -> Result { relay_members::backfill_from_allowlist(&self.pool, community).await } @@ -4158,6 +4337,7 @@ impl Db { } /// Sidecar an accepted product-feedback event, idempotent by event id. + #[tracing::instrument(target = "buzz_datastore", name = "insert_product_feedback", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn insert_product_feedback( &self, community: CommunityId, @@ -4167,6 +4347,7 @@ impl Db { } /// List product feedback across the deployment, newest first. + #[tracing::instrument(target = "buzz_datastore", name = "list_product_feedback", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_product_feedback( &self, limit: i64, @@ -4175,6 +4356,7 @@ impl Db { } /// Insert a tenant-scoped NIP-56 report row, idempotent by report event id. + #[tracing::instrument(target = "buzz_datastore", name = "insert_moderation_report", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn insert_moderation_report( &self, community: CommunityId, @@ -4184,6 +4366,7 @@ impl Db { } /// List moderation reports for a community, newest first. + #[tracing::instrument(target = "buzz_datastore", name = "list_moderation_reports", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_moderation_reports( &self, community: CommunityId, @@ -4194,6 +4377,7 @@ impl Db { } /// Fetch one moderation report by row id. + #[tracing::instrument(target = "buzz_datastore", name = "get_moderation_report", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_moderation_report( &self, community: CommunityId, @@ -4203,6 +4387,7 @@ impl Db { } /// Fetch one moderation report by signed NIP-56 report event id. + #[tracing::instrument(target = "buzz_datastore", name = "get_moderation_report_by_event", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_moderation_report_by_event( &self, community: CommunityId, @@ -4212,6 +4397,7 @@ impl Db { } /// Resolve, dismiss, or escalate an open moderation report. + #[tracing::instrument(target = "buzz_datastore", name = "resolve_moderation_report", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn resolve_moderation_report( &self, community: CommunityId, @@ -4232,6 +4418,7 @@ impl Db { } /// Upsert a community ban for a member pubkey. + #[tracing::instrument(target = "buzz_datastore", name = "ban_community_member", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn ban_community_member( &self, community: CommunityId, @@ -4244,6 +4431,7 @@ impl Db { } /// Lift a community ban for a member pubkey. + #[tracing::instrument(target = "buzz_datastore", name = "unban_community_member", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn unban_community_member( &self, community: CommunityId, @@ -4254,6 +4442,7 @@ impl Db { } /// Upsert a community timeout/write-block for a member pubkey. + #[tracing::instrument(target = "buzz_datastore", name = "timeout_community_member", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn timeout_community_member( &self, community: CommunityId, @@ -4266,6 +4455,7 @@ impl Db { } /// Clear a community timeout/write-block for a member pubkey. + #[tracing::instrument(target = "buzz_datastore", name = "untimeout_community_member", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn untimeout_community_member( &self, community: CommunityId, @@ -4276,6 +4466,7 @@ impl Db { } /// Fetch the active ban/timeout restriction state for enforcement hot paths. + #[tracing::instrument(target = "buzz_datastore", name = "moderation_restriction_state", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn moderation_restriction_state( &self, community: CommunityId, @@ -4285,6 +4476,7 @@ impl Db { } /// Fetch the full ban/timeout row for a member pubkey. + #[tracing::instrument(target = "buzz_datastore", name = "get_community_ban", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn get_community_ban( &self, community: CommunityId, @@ -4294,6 +4486,7 @@ impl Db { } /// List currently restricted members in a community. + #[tracing::instrument(target = "buzz_datastore", name = "list_community_restrictions", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_community_restrictions( &self, community: CommunityId, @@ -4302,6 +4495,7 @@ impl Db { } /// Insert a moderation audit action row. + #[tracing::instrument(target = "buzz_datastore", name = "insert_moderation_action", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn insert_moderation_action( &self, community: CommunityId, @@ -4311,6 +4505,7 @@ impl Db { } /// List moderation audit action rows, newest first. + #[tracing::instrument(target = "buzz_datastore", name = "list_moderation_actions", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_moderation_actions( &self, community: CommunityId, @@ -4321,6 +4516,7 @@ impl Db { /// Return the current owner of git repo name `repo_id` in `community`, or /// `None` if unreserved. See [`git_repo::repo_name_owner`]. + #[tracing::instrument(target = "buzz_datastore", name = "repo_name_owner", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn repo_name_owner( &self, community: CommunityId, @@ -4333,6 +4529,7 @@ impl Db { /// /// See [`git_repo::reserve_repo_name`] for the outcome semantics. The /// per-pubkey quota is enforced by the caller against `count_repos_for_owner`. + #[tracing::instrument(target = "buzz_datastore", name = "reserve_repo_name", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn reserve_repo_name( &self, community: CommunityId, @@ -4343,6 +4540,7 @@ impl Db { } /// Count git repos reserved by `owner_pubkey` in `community` (quota check). + #[tracing::instrument(target = "buzz_datastore", name = "count_repos_for_owner", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn count_repos_for_owner( &self, community: CommunityId, @@ -4354,6 +4552,7 @@ impl Db { /// Release a git repo name reservation held by `owner_pubkey` (rollback). /// /// Returns the number of rows removed (0 or 1). See [`git_repo::release_repo_name`]. + #[tracing::instrument(target = "buzz_datastore", name = "release_repo_name", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn release_repo_name( &self, community: CommunityId, @@ -4364,12 +4563,14 @@ impl Db { } /// Returns `true` if `pubkey` (64-char hex) is archived in `community_id`. + #[tracing::instrument(target = "buzz_datastore", name = "is_archived", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn is_archived(&self, community_id: CommunityId, pubkey: &str) -> Result { archived_identities::is_archived(&self.pool, community_id, pubkey).await } /// Archives an identity in `community_id`. Returns `true` if inserted, `false` if already archived. #[allow(clippy::too_many_arguments)] + #[tracing::instrument(target = "buzz_datastore", name = "archive", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn archive( &self, community_id: CommunityId, @@ -4394,11 +4595,13 @@ impl Db { } /// Unarchives an identity from `community_id`. Returns `true` if deleted, `false` if absent. + #[tracing::instrument(target = "buzz_datastore", name = "unarchive", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn unarchive(&self, community_id: CommunityId, pubkey: &str) -> Result { archived_identities::unarchive(&self.pool, community_id, pubkey).await } /// Returns all identities archived in `community_id`, ordered by archive time ascending. + #[tracing::instrument(target = "buzz_datastore", name = "list_archived", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn list_archived( &self, community_id: CommunityId, @@ -4407,6 +4610,7 @@ impl Db { } /// Soft-delete NIP-29 discovery events for a channel created by a specific relay pubkey. + #[tracing::instrument(target = "buzz_datastore", name = "soft_delete_discovery_events", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn soft_delete_discovery_events( &self, community_id: CommunityId, @@ -4432,6 +4636,7 @@ impl Db { /// Same-second ties are broken by lowest event `id` (NIP-16 deterministic ordering). /// Returns `(event, false)` for stale writes and duplicate IDs — callers should /// skip fan-out/dispatch when `was_inserted` is false. + #[tracing::instrument(target = "buzz_datastore", name = "replace_addressable_event", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn replace_addressable_event( &self, community_id: CommunityId, @@ -4548,8 +4753,8 @@ 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 { - tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); + if let Err(_e) = crate::insert_mentions(&self.pool, community_id, event, channel_id).await { + tracing::warn!("mention insertion failed"); } Ok(( @@ -4614,6 +4819,7 @@ impl Db { /// prevents the stale-snapshot race where a concurrent publication reads /// older state and overwrites a newer snapshot by arrival order. /// + #[tracing::instrument(target = "buzz_datastore", name = "publish_nip43_membership_locked", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn publish_nip43_membership_locked( &self, community_id: CommunityId, @@ -4722,8 +4928,8 @@ impl Db { tx.commit().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}"); + if let Err(_e) = crate::insert_mentions(&self.pool, community_id, &event, None).await { + tracing::warn!("mention insertion failed"); } Ok(( @@ -4754,6 +4960,7 @@ impl Db { /// relay-signed NIP-29 group metadata (kind 39000–39002) where the relay is the /// author and channel_id distinguishes groups. User-submitted NIP-33 events use /// this function instead, where the author's pubkey + d-tag is the natural key. + #[tracing::instrument(target = "buzz_datastore", name = "replace_parameterized_event", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn replace_parameterized_event( &self, community_id: CommunityId, @@ -4959,8 +5166,8 @@ 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 { - tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); + if let Err(_e) = crate::insert_mentions(&self.pool, community_id, event, channel_id).await { + tracing::warn!("mention insertion failed"); } Ok(( diff --git a/crates/buzz-db/src/replica_fence.rs b/crates/buzz-db/src/replica_fence.rs index c840db393e..5dde365835 100644 --- a/crates/buzz-db/src/replica_fence.rs +++ b/crates/buzz-db/src/replica_fence.rs @@ -315,6 +315,7 @@ impl ReplicaFence { /// /// This is a name-and-shape check only; it cannot detect a sabotaged /// function body. [`verify_floor_guard_behavior`] proves the semantics. +#[tracing::instrument(target = "buzz_datastore", name = "replica_fence_verify_catalog", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn verify_floor_guard_catalog(pool: &PgPool) -> crate::Result<()> { // tgtype bits: 1 = ROW, 2 = BEFORE, 4 = INSERT, 16 = UPDATE, 64 = INSTEAD. // Required: ROW + INSERT + UPDATE set, BEFORE + INSTEAD clear. @@ -369,6 +370,7 @@ pub async fn verify_floor_guard_catalog(pool: &PgPool) -> crate::Result<()> { /// `SET CONSTRAINTS ALL IMMEDIATE` makes the deferred trigger fire per /// statement so each adversary is observable under a savepoint; deferral to /// COMMIT is separately pinned by the held-transaction fixture. +#[tracing::instrument(target = "buzz_datastore", name = "replica_fence_verify_behavior", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn verify_floor_guard_behavior(pool: &PgPool) -> crate::Result<()> { use crate::error::DbError; @@ -542,6 +544,7 @@ pub enum ProbeError { /// The statements are separately awaited on a single pinned connection; /// a single SELECT would not guarantee evaluation order across the /// subexpressions, reopening the race this ordering exists to close. +#[tracing::instrument(target = "buzz_datastore", name = "replica_fence_sample_writer", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] async fn sample_writer(writer: &PgPool) -> Result { let mut conn = writer.acquire().await?; @@ -698,6 +701,7 @@ pub const AURORA_IDENTITY_FN: &str = "aurora_db_instance_identifier"; /// (undefined_function, SQLSTATE 42883); transient errors surface as `Err` /// so the caller can retry the probe on a later request instead of caching /// a wrong answer. +#[tracing::instrument(target = "buzz_datastore", name = "replica_fence_reader_supports_aurora_identity", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn reader_supports_aurora_identity(conn: &mut PgConnection) -> Result { match sqlx::query(sqlx::AssertSqlSafe(format!( "SELECT {AURORA_IDENTITY_FN}()" @@ -719,6 +723,7 @@ pub async fn reader_supports_aurora_identity(conn: &mut PgConnection) -> Result< /// [`reader_supports_aurora_identity`] confirmed it — the function /// reference fails at parse time on plain Postgres). `None` when the row /// is missing there (migration not yet replayed): fail closed. +#[tracing::instrument(target = "buzz_datastore", name = "replica_fence_observe_heartbeat", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))] pub async fn observe_heartbeat( conn: &mut PgConnection, aurora: bool, diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 2d82736807..349faa848b 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -97,6 +97,12 @@ enum PersistResult { /// persists without the event record. On retry, the event INSERT succeeds /// (no conflict), and the mutation re-executes — which is safe for idempotent /// operations (open_dm, hide_dm, update_approval, upsert_workflow). +#[tracing::instrument( + target = "buzz_datastore", + name = "persist_command_event", + skip_all, + fields(otel.kind = "client", db.system.name = "postgresql") +)] async fn persist_command_event( state: &Arc, tenant: &TenantContext, diff --git a/crates/buzz-search/Cargo.toml b/crates/buzz-search/Cargo.toml index e42bcc8041..72b65a026b 100644 --- a/crates/buzz-search/Cargo.toml +++ b/crates/buzz-search/Cargo.toml @@ -12,6 +12,7 @@ buzz-core = { workspace = true } sqlx = { workspace = true } uuid = { workspace = true } thiserror = { workspace = true } +tracing = { workspace = true } [dev-dependencies] tokio = { workspace = true } diff --git a/crates/buzz-search/src/query.rs b/crates/buzz-search/src/query.rs index 7f33b660c4..f8b2c51237 100644 --- a/crates/buzz-search/src/query.rs +++ b/crates/buzz-search/src/query.rs @@ -10,6 +10,7 @@ use buzz_core::CommunityId; use sqlx::{PgPool, QueryBuilder, Row}; +use tracing::Instrument; use uuid::Uuid; use crate::error::SearchError; @@ -297,7 +298,16 @@ pub async fn search(pool: &PgPool, query: &SearchQuery) -> Result