diff --git a/.env.example b/.env.example index a8742bdf5..0f7bbba6f 100644 --- a/.env.example +++ b/.env.example @@ -69,39 +69,6 @@ RELAY_URL=ws://localhost:3000 # BUZZ_RATE_LIMIT_AGENT_ELEVATED_MESSAGES_PER_MIN=300 # BUZZ_RATE_LIMIT_AGENT_PLATFORM_MESSAGES_PER_MIN=600 -# Relay-verified identity (disabled by default). When enabled, the relay -# requires authenticated requests to present a valid corporate JWT, then binds -# the configured uid claim to the Nostr pubkey proven by NIP-42/NIP-98. The JWT -# may be injected by a trusted proxy or attached by a first-party client; the -# relay treats both as the same header. Clients must forward the configured -# token header on every authenticated HTTP request and session handshake. -# -# Operational notes for the initial implementation: -# - When a trusted proxy injects this header, it MUST overwrite any inbound -# client-supplied value before forwarding to the relay. -# - Revocation and rotation are explicit database lifecycle operations; -# ordinary authentication never silently replaces a key. -# - JWKS outages fail closed for human JWT authentication. Delegated agent -# admission can still work when the owner binding is already present. -# - DISPLAY_CLAIM is private binding metadata. It is never projected publicly -# unless PUBLIC_DISPLAY_CLAIM is separately configured. -# BUZZ_REQUIRE_CORPORATE_IDENTITY=false -# BUZZ_CORPORATE_IDENTITY_JWT_HEADER=x-forwarded-identity-token -# BUZZ_ALLOW_CORPORATE_IDENTITY_DELEGATION=true -# When a request carries both a JWT and a verified NIP-OA owner declaration, -# choose whether the JWT identifies the signer or the owner binding delegates -# access. Defaults to direct; deployments that inject an owner's JWT into agent -# requests can explicitly select delegated. -# BUZZ_CORPORATE_IDENTITY_AUTH_PRECEDENCE=direct -# BUZZ_CORPORATE_IDENTITY_JWKS_URI=https://idp.example/.well-known/jwks.json -# BUZZ_CORPORATE_IDENTITY_ISSUER=https://idp.example -# BUZZ_CORPORATE_IDENTITY_AUDIENCE=buzz-relay -# BUZZ_CORPORATE_IDENTITY_UID_CLAIM=sub -# BUZZ_CORPORATE_IDENTITY_DISPLAY_CLAIM=email -# Optional, public NIP-85 label. Unset by default to keep identity claims private. -# BUZZ_CORPORATE_IDENTITY_PUBLIC_DISPLAY_CLAIM=display_name -# BUZZ_CORPORATE_IDENTITY_NPUB_CLAIM=buzz_npub - # ----------------------------------------------------------------------------- # Git (NIP-34 bare repositories) # ----------------------------------------------------------------------------- diff --git a/Cargo.lock b/Cargo.lock index 93fe292b4..2d86e05aa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -916,6 +916,7 @@ name = "buzz-audit" version = "0.1.0" dependencies = [ "buzz-core", + "buzz-datastore-tracing", "chrono", "futures-util", "hex", @@ -1032,11 +1033,27 @@ dependencies = [ "zeroize", ] +[[package]] +name = "buzz-datastore-tracing" +version = "0.1.0" +dependencies = [ + "opentelemetry 0.32.0", + "opentelemetry_sdk 0.32.1", + "proc-macro2", + "quote", + "syn 2.0.117", + "tokio", + "tracing", + "tracing-opentelemetry", + "tracing-subscriber", +] + [[package]] name = "buzz-db" version = "0.1.0" dependencies = [ "buzz-core", + "buzz-datastore-tracing", "chrono", "hex", "metrics", @@ -1225,6 +1242,7 @@ dependencies = [ "buzz-auth", "buzz-conformance", "buzz-core", + "buzz-datastore-tracing", "buzz-db", "buzz-media", "buzz-pubsub", @@ -1322,9 +1340,11 @@ name = "buzz-search" version = "0.1.0" dependencies = [ "buzz-core", + "buzz-datastore-tracing", "sqlx", "thiserror 2.0.18", "tokio", + "tracing", "uuid", ] diff --git a/Cargo.toml b/Cargo.toml index 3b22f2a81..c228a754d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,6 +28,7 @@ members = [ "crates/buzz-dev-mcp", "crates/buzz-voice", "crates/buzz-backend-kubernetes", + "crates/buzz-datastore-tracing", "examples/countdown-bot", ] exclude = ["desktop/src-tauri"] @@ -145,6 +146,7 @@ buzz-media = { path = "crates/buzz-media" } buzz-sdk = { path = "crates/buzz-sdk" } buzz-ws-client = { path = "crates/buzz-ws-client" } buzz-relay-mesh = { path = "crates/buzz-relay-mesh" } +buzz-datastore-tracing = { path = "crates/buzz-datastore-tracing" } # CI profile — builds the relay for desktop e2e. Dependencies keep full # release optimization (warm from main's cache; they carry the runtime hot diff --git a/crates/buzz-audit/Cargo.toml b/crates/buzz-audit/Cargo.toml index ff7bafb37..dfa73353d 100644 --- a/crates/buzz-audit/Cargo.toml +++ b/crates/buzz-audit/Cargo.toml @@ -9,6 +9,7 @@ description = "Hash-chain audit log for Buzz" [dependencies] buzz-core = { workspace = true } +buzz-datastore-tracing = { workspace = true } sqlx = { workspace = true } tokio = { workspace = true } serde = { workspace = true } diff --git a/crates/buzz-audit/src/service.rs b/crates/buzz-audit/src/service.rs index fa0e0fb44..9ae1d1685 100644 --- a/crates/buzz-audit/src/service.rs +++ b/crates/buzz-audit/src/service.rs @@ -1,10 +1,11 @@ use chrono::{DateTime, Utc}; use futures_util::FutureExt as _; use sqlx::{Acquire, PgPool, Row}; -use tracing::{debug, instrument, warn}; +use tracing::{debug, warn}; use uuid::Uuid; use buzz_core::CommunityId; +use buzz_datastore_tracing::datastore_span; use crate::{ action::AuditAction, @@ -49,7 +50,11 @@ 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))] + #[datastore_span( + name = "audit_log", + system = "postgresql", + fields(action = %entry.action) + )] pub async fn log(&self, entry: NewAuditEntry) -> Result { let mut conn = self.pool.acquire().await?; @@ -156,7 +161,11 @@ 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))] + #[datastore_span( + name = "audit_verify_chain", + system = "postgresql", + fields(from_seq = from_seq, to_seq = to_seq) + )] pub async fn verify_chain( &self, community: CommunityId, @@ -208,7 +217,11 @@ 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))] + #[datastore_span( + name = "audit_get_entries", + system = "postgresql", + fields(from_seq = from_seq, limit = limit) + )] pub async fn get_entries( &self, community: CommunityId, @@ -274,7 +287,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-datastore-tracing/Cargo.toml b/crates/buzz-datastore-tracing/Cargo.toml new file mode 100644 index 000000000..e93900c54 --- /dev/null +++ b/crates/buzz-datastore-tracing/Cargo.toml @@ -0,0 +1,24 @@ +[package] +name = "buzz-datastore-tracing" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Privacy-preserving datastore tracing policy macros for Buzz" + +[lib] +proc-macro = true + +[dependencies] +proc-macro2 = "1" +quote = "1" +syn = { version = "2", features = ["full"] } + +[dev-dependencies] +opentelemetry = { workspace = true } +opentelemetry_sdk = { workspace = true, features = ["testing"] } +tokio = { workspace = true } +tracing = { workspace = true } +tracing-opentelemetry = { workspace = true } +tracing-subscriber = { workspace = true } diff --git a/crates/buzz-datastore-tracing/src/lib.rs b/crates/buzz-datastore-tracing/src/lib.rs new file mode 100644 index 000000000..f2645cb8f --- /dev/null +++ b/crates/buzz-datastore-tracing/src/lib.rs @@ -0,0 +1,168 @@ +//! Policy-enforcing instrumentation for logical datastore operations. + +use proc_macro::TokenStream; +use proc_macro2::TokenStream as TokenStream2; +use quote::{format_ident, quote}; +use syn::parse::{Parse, ParseStream}; +use syn::{ + parenthesized, parse_macro_input, Error, Ident, ItemFn, LitStr, Result, ReturnType, Token, Type, +}; + +struct DatastoreArgs { + name: LitStr, + system: LitStr, + fields: Option, +} + +impl Parse for DatastoreArgs { + fn parse(input: ParseStream<'_>) -> Result { + let mut name = None; + let mut system = None; + let mut fields = None; + + while !input.is_empty() { + let key: Ident = input.parse()?; + match key.to_string().as_str() { + "name" => { + if name.is_some() { + return Err(Error::new(key.span(), "duplicate `name` argument")); + } + input.parse::()?; + name = Some(input.parse()?); + } + "system" => { + if system.is_some() { + return Err(Error::new(key.span(), "duplicate `system` argument")); + } + input.parse::()?; + system = Some(input.parse()?); + } + "fields" => { + if fields.is_some() { + return Err(Error::new(key.span(), "duplicate `fields` argument")); + } + let content; + parenthesized!(content in input); + fields = Some(content.parse()?); + } + _ => { + return Err(Error::new( + key.span(), + "expected `name`, `system`, or `fields`", + )) + } + } + if !input.is_empty() { + input.parse::()?; + } + } + + Ok(Self { + name: name.ok_or_else(|| Error::new(input.span(), "missing `name`"))?, + system: system.ok_or_else(|| Error::new(input.span(), "missing `system`"))?, + fields, + }) + } +} + +/// Instruments an async logical datastore operation according to Buzz policy. +/// +/// PostgreSQL spans always omit function arguments, use the `buzz_datastore` +/// target, and expose only canonical semantic fields plus explicitly supplied +/// safe fields. An `Err` sets `otel.status_code` without inspecting the error. +#[proc_macro_attribute] +pub fn datastore_span(args: TokenStream, item: TokenStream) -> TokenStream { + let args = parse_macro_input!(args as DatastoreArgs); + let mut function = parse_macro_input!(item as ItemFn); + + if function.sig.asyncness.is_none() { + return Error::new_spanned( + function.sig.fn_token, + "`datastore_span` requires an async function", + ) + .into_compile_error() + .into(); + } + if args.system.value() != "postgresql" { + return Error::new_spanned( + args.system, + "unsupported datastore system; only `postgresql` is currently supported", + ) + .into_compile_error() + .into(); + } + + let name = args.name; + let extra_fields = args.fields.map(|tokens| quote!(, #tokens)); + function.attrs.push(syn::parse_quote!( + #[::tracing::instrument( + target = "buzz_datastore", + name = #name, + skip_all, + fields( + otel.kind = "client", + db.system.name = "postgresql", + otel.status_code = ::tracing::field::Empty + #extra_fields + ) + )] + )); + + let return_type: Type = match &function.sig.output { + ReturnType::Type(_, ty) => *ty.clone(), + ReturnType::Default => syn::parse_quote!(()), + }; + let returns_result = match &return_type { + Type::Path(path) => path + .path + .segments + .last() + .is_some_and(|segment| segment.ident == "Result"), + _ => false, + }; + let original_body = function.block; + let result = format_ident!("__buzz_datastore_result_7f3a9c"); + let record_error = returns_result.then(|| { + quote! { + if #result.is_err() { + ::tracing::Span::current().record("otel.status_code", "ERROR"); + } + } + }); + function.block = Box::new(syn::parse_quote!({ + let #result: #return_type = (async #original_body).await; + #record_error + #result + })); + + quote!(#function).into() +} + +#[cfg(test)] +mod tests { + use super::DatastoreArgs; + + #[test] + fn parses_safe_tracing_fields() { + let args: DatastoreArgs = syn::parse_str( + r#"name = "audit", system = "postgresql", fields(action = %entry.action, from_seq = from_seq, limit = limit)"#, + ) + .expect("valid arguments"); + assert!(args.fields.is_some()); + } + + #[test] + fn rejects_duplicate_arguments() { + for args in [ + r#"name = "one", name = "two", system = "postgresql""#, + r#"name = "one", system = "postgresql", system = "postgresql""#, + r#"name = "one", system = "postgresql", fields(a = 1), fields(b = 2)"#, + ] { + let error = match syn::parse_str::(args) { + Ok(_) => panic!("duplicate accepted"), + Err(error) => error, + }; + assert!(error.to_string().starts_with("duplicate `")); + } + } +} diff --git a/crates/buzz-datastore-tracing/tests/runtime.rs b/crates/buzz-datastore-tracing/tests/runtime.rs new file mode 100644 index 000000000..b58dca871 --- /dev/null +++ b/crates/buzz-datastore-tracing/tests/runtime.rs @@ -0,0 +1,80 @@ +use buzz_datastore_tracing::datastore_span; +use opentelemetry::trace::{SpanKind, Status, TracerProvider as _}; +use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider}; +use tracing_subscriber::prelude::*; + +const DIRECT_ERROR: &str = "raw-secret-direct-error"; +const QUESTION_ERROR: &str = "raw-secret-question-error"; + +fn question_path(fail: bool) -> Result<(), &'static str> { + if fail { + Err(QUESTION_ERROR) + } else { + Ok(()) + } +} + +#[datastore_span(name = "test_operation", system = "postgresql", fields(limit = limit))] +async fn operation( + limit: usize, + direct_error: bool, + question_error: bool, +) -> Result { + if direct_error { + return Err(DIRECT_ERROR); + } + question_path(question_error)?; + Ok(limit) +} + +#[tokio::test(flavor = "current_thread")] +async fn exports_policy_fields_without_error_or_argument_data() { + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let subscriber = tracing_subscriber::registry() + .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("datastore-macro-test"))); + let _subscriber_guard = tracing::subscriber::set_default(subscriber); + + assert_eq!(operation(7, false, false).await, Ok(7)); + assert_eq!(operation(8, true, false).await, Err(DIRECT_ERROR)); + assert_eq!(operation(9, false, true).await, Err(QUESTION_ERROR)); + + provider.force_flush().expect("spans flush"); + let spans = exporter.get_finished_spans().expect("exported spans"); + assert_eq!(spans.len(), 3); + + for (span, (expected_limit, expected_status)) in spans.iter().zip([ + (7_i64, Status::Unset), + (8_i64, Status::error("")), + (9_i64, Status::error("")), + ]) { + assert_eq!(span.name, "test_operation"); + assert_eq!(span.span_kind, SpanKind::Client); + assert_eq!(span.status, expected_status); + + let attributes = span + .attributes + .iter() + .map(|attribute| (attribute.key.as_str(), attribute.value.to_string())) + .collect::>(); + assert!(attributes.contains(&("target", "buzz_datastore".to_owned()))); + assert!(attributes.contains(&("db.system.name", "postgresql".to_owned()))); + assert!(attributes.contains(&("limit", expected_limit.to_string()))); + assert!(!attributes + .iter() + .any(|(key, _)| { matches!(*key, "direct_error" | "question_error") })); + + let exported = format!("{span:?}"); + assert!(!exported.contains(DIRECT_ERROR)); + assert!(!exported.contains(QUESTION_ERROR)); + assert!(span.events.iter().all(|event| { + !format!("{event:?}").contains(DIRECT_ERROR) + && !format!("{event:?}").contains(QUESTION_ERROR) + })); + if let Status::Error { description } = &span.status { + assert!(description.is_empty()); + } + } +} diff --git a/crates/buzz-db/Cargo.toml b/crates/buzz-db/Cargo.toml index 6f76a11bc..380605728 100644 --- a/crates/buzz-db/Cargo.toml +++ b/crates/buzz-db/Cargo.toml @@ -9,6 +9,7 @@ description = "Postgres event store and data access layer for Buzz" [dependencies] buzz-core = { workspace = true } +buzz-datastore-tracing = { workspace = true } sqlx = { workspace = true } tokio = { workspace = true } serde = { workspace = true } diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 4900650d8..96eeea7ad 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -61,6 +61,7 @@ pub mod workflow; pub use error::{DbError, Result}; pub use event::{EventQuery, ReactionEventInsertOutcome, DEFAULT_MAX_PAGE_LIMIT}; +use buzz_datastore_tracing::datastore_span; use chrono::{DateTime, Utc}; use sqlx::postgres::{PgConnection, PgPoolOptions}; use sqlx::{Connection, PgPool, QueryBuilder, Row}; @@ -256,6 +257,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. + #[datastore_span(name = "read_session_query_events", system = "postgresql")] pub async fn query_events(&mut self, q: &EventQuery) -> Result> { let degraded = match &mut self.inner { ReadSessionInner::Replica { tx, writer } => { @@ -1013,6 +1015,7 @@ impl Db { } /// Run pending database migrations. + #[datastore_span(name = "migrate", system = "postgresql")] pub async fn migrate(&self) -> Result<()> { migration::run_migrations(&self.pool).await } @@ -1057,6 +1060,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. + #[datastore_span(name = "try_lock_usage_metrics", system = "postgresql")] pub async fn try_lock_usage_metrics( &self, lock_key: i64, @@ -1077,6 +1081,7 @@ impl Db { /// List reports for the deployment-global read-only admin plane. #[allow(clippy::too_many_arguments)] + #[datastore_span(name = "admin_list_reports", system = "postgresql")] pub async fn admin_list_reports( &self, community_id: Option, @@ -1103,6 +1108,7 @@ impl Db { } /// Fetch one report for the deployment-global read-only admin plane. + #[datastore_span(name = "admin_get_report", system = "postgresql")] pub async fn admin_get_report( &self, id: Uuid, @@ -1111,6 +1117,7 @@ impl Db { } /// List feedback for the deployment-global read-only admin plane. + #[datastore_span(name = "admin_list_feedback", system = "postgresql")] pub async fn admin_list_feedback( &self, limit: i64, @@ -1119,6 +1126,7 @@ impl Db { } /// Fetch one feedback submission for the deployment-global admin plane. + #[datastore_span(name = "admin_get_feedback", system = "postgresql")] pub async fn admin_get_feedback( &self, id: Uuid, @@ -1127,36 +1135,43 @@ impl Db { } /// Return total number of communities on this relay. + #[datastore_span(name = "usage_community_count", system = "postgresql")] pub async fn usage_community_count(&self) -> Result { usage::community_count(&self.pool).await } /// Return per-community user counts split by human/agent. + #[datastore_span(name = "usage_user_counts", system = "postgresql")] pub async fn usage_user_counts(&self) -> Result> { usage::user_counts(&self.pool).await } /// Return per-community channel counts by type. + #[datastore_span(name = "usage_channel_counts", system = "postgresql")] pub async fn usage_channel_counts(&self) -> Result> { usage::channel_counts(&self.pool).await } /// Return per-community kind=9 message counts. + #[datastore_span(name = "usage_message_counts", system = "postgresql")] pub async fn usage_message_counts(&self) -> Result> { usage::message_counts(&self.pool).await } /// Return per-community relay-member counts by role. + #[datastore_span(name = "usage_relay_member_counts", system = "postgresql")] pub async fn usage_relay_member_counts(&self) -> Result> { usage::relay_member_counts(&self.pool).await } /// Return per-community workflow counts by status. + #[datastore_span(name = "usage_workflow_counts", system = "postgresql")] pub async fn usage_workflow_counts(&self) -> Result> { usage::workflow_counts(&self.pool).await } /// Return per-community git-repo counts. + #[datastore_span(name = "usage_git_repo_counts", system = "postgresql")] pub async fn usage_git_repo_counts(&self) -> Result> { usage::git_repo_counts(&self.pool).await } @@ -1164,6 +1179,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"`. + #[datastore_span(name = "usage_active_user_counts", system = "postgresql")] pub async fn usage_active_user_counts( &self, interval_sql: &'static str, @@ -1172,6 +1188,7 @@ impl Db { } /// Return per-community active-channel counts for a given SQL interval. + #[datastore_span(name = "usage_active_channel_counts", system = "postgresql")] pub async fn usage_active_channel_counts( &self, interval_sql: &'static str, @@ -1180,6 +1197,7 @@ impl Db { } /// Return all community id → host mappings. + #[datastore_span(name = "usage_community_hosts", system = "postgresql")] pub async fn usage_community_hosts(&self) -> Result> { usage::community_hosts(&self.pool).await } @@ -1196,6 +1214,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. + #[datastore_span(name = "lookup_community_by_host", system = "postgresql")] pub async fn lookup_community_by_host( &self, normalized_host: &str, @@ -1225,6 +1244,7 @@ impl Db { } /// Returns whether a community id still exists in the active lifecycle state. + #[datastore_span(name = "is_community_active", system = "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)", @@ -1236,6 +1256,10 @@ impl Db { } /// Returns a community by host regardless of lifecycle state. Operator-plane only. + #[datastore_span( + name = "lookup_community_by_host_for_management", + system = "postgresql" + )] pub async fn lookup_community_by_host_for_management( &self, normalized_host: &str, @@ -1257,6 +1281,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. + #[datastore_span(name = "list_communities_owned_by", system = "postgresql")] pub async fn list_communities_owned_by( &self, owner_pubkey: &str, @@ -1302,6 +1327,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. + #[datastore_span(name = "lookup_community_host", system = "postgresql")] pub async fn lookup_community_host(&self, community_id: CommunityId) -> Result> { let row = sqlx::query( r#" @@ -1326,6 +1352,7 @@ impl Db { /// /// Set by relay admins/owners via the kind:9033 command; the value is /// validated and size-capped at that write path. + #[datastore_span(name = "get_community_icon", system = "postgresql")] pub async fn get_community_icon(&self, community_id: CommunityId) -> Result> { let row = sqlx::query( r#" @@ -1346,6 +1373,7 @@ impl Db { } /// Sets or clears (`None`) the community's workspace icon. + #[datastore_span(name = "set_community_icon", system = "postgresql")] pub async fn set_community_icon( &self, community_id: CommunityId, @@ -1370,6 +1398,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. + #[datastore_span(name = "ensure_configured_community", system = "postgresql")] pub async fn ensure_configured_community( &self, normalized_host: &str, @@ -1402,6 +1431,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. + #[datastore_span(name = "create_community_with_owner", system = "postgresql")] pub async fn create_community_with_owner( &self, normalized_host: &str, @@ -1487,6 +1517,7 @@ impl Db { } /// Idempotently archives a community when the asserted pubkey is its current owner. + #[datastore_span(name = "archive_community_owned_by", system = "postgresql")] pub async fn archive_community_owned_by( &self, normalized_host: &str, @@ -1520,6 +1551,7 @@ impl Db { } /// Idempotently restores a community when the asserted pubkey is its current owner. + #[datastore_span(name = "unarchive_community_owned_by", system = "postgresql")] pub async fn unarchive_community_owned_by( &self, normalized_host: &str, @@ -1552,6 +1584,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. + #[datastore_span(name = "community_of_channel", system = "postgresql")] pub async fn community_of_channel(&self, channel_id: Uuid) -> Result> { let row = sqlx::query( r#" @@ -1590,6 +1623,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". + #[datastore_span(name = "communities_of_channels", system = "postgresql")] pub async fn communities_of_channels( &self, channel_ids: &[Uuid], @@ -1619,6 +1653,7 @@ impl Db { } /// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate. + #[datastore_span(name = "insert_event", system = "postgresql")] pub async fn insert_event( &self, community_id: CommunityId, @@ -1641,6 +1676,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. + #[datastore_span(name = "query_events", system = "postgresql")] pub async fn query_events(&self, q: &EventQuery) -> Result> { event::query_events(&self.pool, q).await } @@ -1661,6 +1697,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. + #[datastore_span(name = "query_events_routed", system = "postgresql")] pub async fn query_events_routed( &self, path: &'static str, @@ -1695,6 +1732,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`]. + #[datastore_span(name = "query_events_routed_bounded", system = "postgresql")] pub async fn query_events_routed_bounded( &self, path: &'static str, @@ -1722,6 +1760,7 @@ impl Db { /// /// Always reads from the WRITER pool — see [`Db::query_events`] for the /// writer-vs-routed rule. + #[datastore_span(name = "count_events", system = "postgresql")] pub async fn count_events(&self, q: &EventQuery) -> Result { event::count_events(&self.pool, q).await } @@ -1736,6 +1775,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`. + #[datastore_span(name = "count_events_routed", system = "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) => { @@ -1757,6 +1797,7 @@ impl Db { /// Return whether a creator-signed huddle-start event links a parent /// channel to an ephemeral huddle channel. + #[datastore_span(name = "huddle_started_link_exists", system = "postgresql")] pub async fn huddle_started_link_exists( &self, community_id: CommunityId, @@ -1779,6 +1820,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. + #[datastore_span(name = "get_latest_global_replaceable", system = "postgresql")] pub async fn get_latest_global_replaceable( &self, community_id: CommunityId, @@ -1791,6 +1833,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. + #[datastore_span(name = "get_event_by_id", system = "postgresql")] pub async fn get_event_by_id( &self, community_id: CommunityId, @@ -1800,6 +1843,7 @@ impl Db { } /// Fetches a single event by its raw ID bytes, **including soft-deleted rows**. + #[datastore_span(name = "get_event_by_id_including_deleted", system = "postgresql")] pub async fn get_event_by_id_including_deleted( &self, community_id: CommunityId, @@ -1809,6 +1853,7 @@ impl Db { } /// Soft-deletes an event. Returns `Ok(true)` if deleted, `Ok(false)` if already deleted. + #[datastore_span(name = "soft_delete_event", system = "postgresql")] pub async fn soft_delete_event( &self, community_id: CommunityId, @@ -1821,6 +1866,7 @@ impl Db { /// when it is not newer than the deletion request. /// Used by NIP-09 a-tag deletion for parameterized-replaceable kinds; /// `deletion_created_at_secs` is the deletion event's `created_at`. + #[datastore_span(name = "soft_delete_by_coordinate", system = "postgresql")] pub async fn soft_delete_by_coordinate( &self, community_id: CommunityId, @@ -1841,6 +1887,7 @@ impl Db { } /// Atomically soft-delete an event and decrement thread reply counters. + #[datastore_span(name = "soft_delete_event_and_update_thread", system = "postgresql")] pub async fn soft_delete_event_and_update_thread( &self, community_id: CommunityId, @@ -1859,6 +1906,7 @@ impl Db { } /// Returns the most recent `created_at` for a channel. + #[datastore_span(name = "get_last_message_at", system = "postgresql")] pub async fn get_last_message_at( &self, community_id: CommunityId, @@ -1868,6 +1916,7 @@ impl Db { } /// Bulk-fetch the most recent `created_at` for a set of channel IDs. + #[datastore_span(name = "get_last_message_at_bulk", system = "postgresql")] pub async fn get_last_message_at_bulk( &self, community_id: CommunityId, @@ -1877,6 +1926,7 @@ impl Db { } /// Batch-fetch non-deleted events by their raw IDs. + #[datastore_span(name = "get_events_by_ids", system = "postgresql")] pub async fn get_events_by_ids( &self, community_id: CommunityId, @@ -1892,6 +1942,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. + #[datastore_span(name = "get_events_by_ids_routed", system = "postgresql")] pub async fn get_events_by_ids_routed( &self, path: &'static str, @@ -1917,6 +1968,7 @@ impl Db { } /// Exclusively claim a batch of due matcher jobs from one community. + #[datastore_span(name = "claim_due_push_match_batch", system = "postgresql")] pub async fn claim_due_push_match_batch( &self, limit: i64, @@ -1926,6 +1978,7 @@ impl Db { } /// Load active endpoint-enabled leases eligible for push matching. + #[datastore_span(name = "active_push_match_leases", system = "postgresql")] pub async fn active_push_match_leases( &self, community: CommunityId, @@ -1934,6 +1987,7 @@ impl Db { } /// Complete matcher jobs from one claimed batch while the fence holds. + #[datastore_span(name = "complete_push_match_batch", system = "postgresql")] pub async fn complete_push_match_batch( &self, community: CommunityId, @@ -1944,6 +1998,7 @@ impl Db { } /// Release fenced matcher claims from one batch for retry. + #[datastore_span(name = "retry_push_match_batch", system = "postgresql")] pub async fn retry_push_match_batch( &self, community: CommunityId, @@ -1955,11 +2010,13 @@ impl Db { } /// Delete exhausted matcher jobs (periodic sweep, off the claim path). + #[datastore_span(name = "reap_exhausted_push_matches", system = "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. + #[datastore_span(name = "enqueue_push_wake", system = "postgresql")] pub async fn enqueue_push_wake( &self, community: CommunityId, @@ -1971,6 +2028,7 @@ impl Db { } /// Set-wise [`Self::enqueue_push_wake`]: one transaction per batch. + #[datastore_span(name = "enqueue_push_wakes", system = "postgresql")] pub async fn enqueue_push_wakes( &self, community: CommunityId, @@ -1980,6 +2038,7 @@ impl Db { } /// Exclusively claim due wake jobs for one community. + #[datastore_span(name = "claim_due_push_wakes", system = "postgresql")] pub async fn claim_due_push_wakes( &self, community: CommunityId, @@ -1990,6 +2049,7 @@ impl Db { } /// Revalidate a wake's claim, source event, and current lease before send. + #[datastore_span(name = "revalidate_push_wake", system = "postgresql")] pub async fn revalidate_push_wake( &self, community: CommunityId, @@ -2000,6 +2060,7 @@ impl Db { } /// Mark a fenced wake claim delivered. + #[datastore_span(name = "complete_push_wake", system = "postgresql")] pub async fn complete_push_wake( &self, community: CommunityId, @@ -2010,6 +2071,7 @@ impl Db { } /// Release a fenced wake claim for retry at the supplied time. + #[datastore_span(name = "retry_push_wake", system = "postgresql")] pub async fn retry_push_wake( &self, community: CommunityId, @@ -2021,6 +2083,7 @@ impl Db { } /// Mark a fenced wake claim terminally failed. + #[datastore_span(name = "fail_push_wake", system = "postgresql")] pub async fn fail_push_wake( &self, community: CommunityId, @@ -2031,6 +2094,7 @@ impl Db { } /// Disable an endpoint only if the specified lease generation is current. + #[datastore_span(name = "disable_push_endpoint", system = "postgresql")] pub async fn disable_push_endpoint( &self, community: CommunityId, @@ -2050,6 +2114,7 @@ impl Db { /// Atomically persist a validated kind:30350 event and its effective lease. #[allow(clippy::too_many_arguments)] + #[datastore_span(name = "accept_push_lease_event", system = "postgresql")] pub async fn accept_push_lease_event( &self, community: CommunityId, @@ -2072,6 +2137,7 @@ impl Db { } /// Atomically insert an event AND its thread metadata in a single transaction. + #[datastore_span(name = "insert_event_with_thread_metadata", system = "postgresql")] pub async fn insert_event_with_thread_metadata( &self, community_id: CommunityId, @@ -2097,6 +2163,10 @@ impl Db { /// Atomically insert a kind:7 reaction event and its reaction row. #[allow(clippy::too_many_arguments)] + #[datastore_span( + name = "insert_reaction_event_with_thread_metadata", + system = "postgresql" + )] pub async fn insert_reaction_event_with_thread_metadata( &self, community_id: CommunityId, @@ -2131,6 +2201,7 @@ impl Db { /// Creates a new channel, bootstraps the creator as owner, and returns the record. #[allow(clippy::too_many_arguments)] + #[datastore_span(name = "create_channel", system = "postgresql")] pub async fn create_channel( &self, community_id: CommunityId, @@ -2158,6 +2229,7 @@ impl Db { /// /// Returns `(record, true)` if newly created, `(record, false)` if already exists. #[allow(clippy::too_many_arguments)] + #[datastore_span(name = "create_channel_with_id", system = "postgresql")] pub async fn create_channel_with_id( &self, community_id: CommunityId, @@ -2184,6 +2256,7 @@ impl Db { } /// Fetches a channel record by ID. + #[datastore_span(name = "get_channel", system = "postgresql")] pub async fn get_channel( &self, community_id: CommunityId, @@ -2193,6 +2266,7 @@ impl Db { } /// Returns the canvas content for a channel, if any. + #[datastore_span(name = "get_canvas", system = "postgresql")] pub async fn get_canvas( &self, community_id: CommunityId, @@ -2202,6 +2276,7 @@ impl Db { } /// Sets or clears the canvas content for a channel. + #[datastore_span(name = "set_canvas", system = "postgresql")] pub async fn set_canvas( &self, community_id: CommunityId, @@ -2212,6 +2287,7 @@ impl Db { } /// Adds a member to a channel. + #[datastore_span(name = "add_member", system = "postgresql")] pub async fn add_member( &self, community_id: CommunityId, @@ -2254,6 +2330,7 @@ impl Db { } /// Removes a member from a channel. + #[datastore_span(name = "remove_member", system = "postgresql")] pub async fn remove_member( &self, community_id: CommunityId, @@ -2265,6 +2342,7 @@ impl Db { } /// Returns `true` if the pubkey is an active member. + #[datastore_span(name = "is_member", system = "postgresql")] pub async fn is_member( &self, community_id: CommunityId, @@ -2276,6 +2354,7 @@ impl Db { /// Return the active (channel, pubkey) membership pairs among the given /// sets, in one statement. + #[datastore_span(name = "membership_pairs", system = "postgresql")] pub async fn membership_pairs( &self, community_id: CommunityId, @@ -2286,6 +2365,7 @@ impl Db { } /// Returns all active members of a channel. + #[datastore_span(name = "get_members", system = "postgresql")] pub async fn get_members( &self, community_id: CommunityId, @@ -2295,6 +2375,7 @@ impl Db { } /// Returns active members for multiple channels in a single query. + #[datastore_span(name = "get_members_bulk", system = "postgresql")] pub async fn get_members_bulk( &self, community_id: CommunityId, @@ -2304,6 +2385,7 @@ impl Db { } /// Get all channel IDs accessible to a pubkey. + #[datastore_span(name = "get_accessible_channel_ids", system = "postgresql")] pub async fn get_accessible_channel_ids( &self, community_id: CommunityId, @@ -2313,6 +2395,7 @@ impl Db { } /// Lists channels, optionally filtered by visibility. + #[datastore_span(name = "list_channels", system = "postgresql")] pub async fn list_channels( &self, community_id: CommunityId, @@ -2322,6 +2405,7 @@ impl Db { } /// Returns full channel records for all channels a user can access. + #[datastore_span(name = "get_accessible_channels", system = "postgresql")] pub async fn get_accessible_channels( &self, community_id: CommunityId, @@ -2340,6 +2424,7 @@ impl Db { } /// Returns all bot-role members with their aggregated channel names in one community. + #[datastore_span(name = "get_bot_members", system = "postgresql")] pub async fn get_bot_members( &self, community_id: CommunityId, @@ -2348,6 +2433,7 @@ impl Db { } /// Bulk-fetch user records by pubkey. + #[datastore_span(name = "get_users_bulk", system = "postgresql")] pub async fn get_users_bulk( &self, community_id: CommunityId, @@ -2357,6 +2443,7 @@ impl Db { } /// Updates a channel's name and/or description. + #[datastore_span(name = "update_channel", system = "postgresql")] pub async fn update_channel( &self, community_id: CommunityId, @@ -2367,6 +2454,7 @@ impl Db { } /// Sets the topic for a channel. + #[datastore_span(name = "set_topic", system = "postgresql")] pub async fn set_topic( &self, community_id: CommunityId, @@ -2378,6 +2466,7 @@ impl Db { } /// Sets the purpose for a channel. + #[datastore_span(name = "set_purpose", system = "postgresql")] pub async fn set_purpose( &self, community_id: CommunityId, @@ -2389,11 +2478,13 @@ impl Db { } /// Archives a channel. + #[datastore_span(name = "archive_channel", system = "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. + #[datastore_span(name = "unarchive_channel", system = "postgresql")] pub async fn unarchive_channel( &self, community_id: CommunityId, @@ -2403,6 +2494,7 @@ impl Db { } /// Soft-delete a channel. + #[datastore_span(name = "soft_delete_channel", system = "postgresql")] pub async fn soft_delete_channel( &self, community_id: CommunityId, @@ -2412,6 +2504,7 @@ impl Db { } /// Returns the count of active members in a channel. + #[datastore_span(name = "get_member_count", system = "postgresql")] pub async fn get_member_count( &self, community_id: CommunityId, @@ -2421,6 +2514,7 @@ impl Db { } /// Bulk-fetch member counts for a set of channel IDs. + #[datastore_span(name = "get_member_counts_bulk", system = "postgresql")] pub async fn get_member_counts_bulk( &self, community_id: CommunityId, @@ -2430,6 +2524,7 @@ impl Db { } /// Get the active role of a pubkey in a channel. + #[datastore_span(name = "get_member_role", system = "postgresql")] pub async fn get_member_role( &self, community_id: CommunityId, @@ -2440,6 +2535,7 @@ impl Db { } /// Archive ephemeral channels whose TTL deadline has passed. + #[datastore_span(name = "reap_expired_ephemeral_channels", system = "postgresql")] pub async fn reap_expired_ephemeral_channels( &self, ) -> Result> { @@ -2447,6 +2543,7 @@ impl Db { } /// Query due reminders ready for delivery. + #[datastore_span(name = "query_due_reminders", system = "postgresql")] pub async fn query_due_reminders( &self, now_secs: i64, @@ -2456,6 +2553,7 @@ impl Db { } /// Atomically claim a due reminder for delivery (cross-pod dedup). + #[datastore_span(name = "claim_due_reminder", system = "postgresql")] pub async fn claim_due_reminder( &self, community_id: CommunityId, @@ -2466,6 +2564,7 @@ impl Db { } /// Atomically claim a due reminder using a caller-supplied delivery stamp. + #[datastore_span(name = "claim_due_reminder_with_stamp", system = "postgresql")] pub async fn claim_due_reminder_with_stamp( &self, community_id: CommunityId, @@ -2484,6 +2583,7 @@ impl Db { } /// Release a claimed due reminder after a publish failure. + #[datastore_span(name = "release_due_reminder", system = "postgresql")] pub async fn release_due_reminder( &self, community_id: CommunityId, @@ -2506,11 +2606,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`. + #[datastore_span(name = "ensure_user", system = "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. + #[datastore_span(name = "get_user", system = "postgresql")] pub async fn get_user( &self, community_id: CommunityId, @@ -2520,6 +2622,7 @@ impl Db { } /// Update a user's profile fields. + #[datastore_span(name = "update_user_profile", system = "postgresql")] pub async fn update_user_profile( &self, community_id: CommunityId, @@ -2542,6 +2645,7 @@ impl Db { } /// Look up a user by NIP-05 handle. + #[datastore_span(name = "get_user_by_nip05", system = "postgresql")] pub async fn get_user_by_nip05( &self, community_id: CommunityId, @@ -2552,6 +2656,7 @@ impl Db { } /// Search users by display name, NIP-05 handle, or pubkey prefix. + #[datastore_span(name = "search_users", system = "postgresql")] pub async fn search_users( &self, community_id: CommunityId, @@ -2656,6 +2761,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. + #[datastore_span(name = "set_agent_owner", system = "postgresql")] pub async fn set_agent_owner( &self, community_id: CommunityId, @@ -2666,6 +2772,7 @@ impl Db { } /// Get the channel_add_policy and agent_owner_pubkey for a user. + #[datastore_span(name = "get_agent_channel_policy", system = "postgresql")] pub async fn get_agent_channel_policy( &self, community_id: CommunityId, @@ -2675,6 +2782,7 @@ impl Db { } /// Check whether `actor_pubkey` is the agent owner of `target_pubkey`. + #[datastore_span(name = "is_agent_owner", system = "postgresql")] pub async fn is_agent_owner( &self, community_id: CommunityId, @@ -2685,6 +2793,7 @@ impl Db { } /// Set the channel_add_policy for a user. + #[datastore_span(name = "set_channel_add_policy", system = "postgresql")] pub async fn set_channel_add_policy( &self, community_id: CommunityId, @@ -2695,6 +2804,7 @@ impl Db { } /// Find an existing DM by its participant hash. + #[datastore_span(name = "find_dm_by_participants", system = "postgresql")] pub async fn find_dm_by_participants( &self, community_id: CommunityId, @@ -2704,6 +2814,7 @@ impl Db { } /// Create or return an existing DM channel. + #[datastore_span(name = "create_dm", system = "postgresql")] pub async fn create_dm( &self, community_id: CommunityId, @@ -2714,6 +2825,7 @@ impl Db { } /// List all DMs for a user. + #[datastore_span(name = "list_dms_for_user", system = "postgresql")] pub async fn list_dms_for_user( &self, community_id: CommunityId, @@ -2725,6 +2837,7 @@ impl Db { } /// Open or retrieve a DM for the given participants. + #[datastore_span(name = "open_dm", system = "postgresql")] pub async fn open_dm( &self, community_id: CommunityId, @@ -2738,6 +2851,7 @@ impl Db { /// /// The DM is not deleted — it can be restored by opening a new DM with /// the same participants. + #[datastore_span(name = "hide_dm", system = "postgresql")] pub async fn hide_dm( &self, community_id: CommunityId, @@ -2748,6 +2862,7 @@ impl Db { } /// Unhide a DM channel for a specific user. + #[datastore_span(name = "unhide_dm", system = "postgresql")] pub async fn unhide_dm( &self, community_id: CommunityId, @@ -2758,6 +2873,7 @@ impl Db { } /// List the channel IDs of all DMs the given user currently has hidden. + #[datastore_span(name = "list_hidden_dms", system = "postgresql")] pub async fn list_hidden_dms( &self, community_id: CommunityId, @@ -2768,6 +2884,7 @@ impl Db { /// Insert thread metadata. #[allow(clippy::too_many_arguments)] + #[datastore_span(name = "insert_thread_metadata", system = "postgresql")] pub async fn insert_thread_metadata( &self, community_id: CommunityId, @@ -2818,6 +2935,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. + #[datastore_span(name = "get_thread_replies", system = "postgresql")] pub async fn get_thread_replies( &self, community_id: CommunityId, @@ -2892,6 +3010,7 @@ impl Db { } /// Fetch aggregated thread stats. + #[datastore_span(name = "get_thread_summary", system = "postgresql")] pub async fn get_thread_summary( &self, community_id: CommunityId, @@ -2943,6 +3062,7 @@ impl Db { /// /// Every failure fails closed to the writer and is recorded in /// `buzz_db_route_decision`. + #[datastore_span(name = "get_channel_window", system = "postgresql")] pub async fn get_channel_window_with_session( &self, community_id: CommunityId, @@ -3108,6 +3228,7 @@ impl Db { } /// Look up a single thread_metadata row by event_id. + #[datastore_span(name = "get_thread_metadata_by_event", system = "postgresql")] pub async fn get_thread_metadata_by_event( &self, community_id: CommunityId, @@ -3117,6 +3238,7 @@ impl Db { } /// Decrement reply counts. + #[datastore_span(name = "decrement_reply_count", system = "postgresql")] pub async fn decrement_reply_count( &self, community_id: CommunityId, @@ -3128,6 +3250,7 @@ impl Db { } /// Add (or re-activate) a reaction. + #[datastore_span(name = "add_reaction", system = "postgresql")] pub async fn add_reaction( &self, community: CommunityId, @@ -3150,6 +3273,7 @@ impl Db { } /// Soft-delete a reaction. + #[datastore_span(name = "remove_reaction", system = "postgresql")] pub async fn remove_reaction( &self, community: CommunityId, @@ -3170,6 +3294,7 @@ impl Db { } /// Soft-delete a reaction by its source event ID. + #[datastore_span(name = "remove_reaction_by_source_event_id", system = "postgresql")] pub async fn remove_reaction_by_source_event_id( &self, community: CommunityId, @@ -3179,6 +3304,7 @@ impl Db { } /// Look up the active reaction row for one actor + emoji + target tuple. + #[datastore_span(name = "get_active_reaction_record", system = "postgresql")] pub async fn get_active_reaction_record( &self, community: CommunityId, @@ -3199,6 +3325,7 @@ impl Db { } /// Backfill the source event ID on an active reaction row. + #[datastore_span(name = "set_reaction_event_id", system = "postgresql")] pub async fn set_reaction_event_id( &self, community: CommunityId, @@ -3221,6 +3348,7 @@ impl Db { } /// Get all active reactions for an event, grouped by emoji. + #[datastore_span(name = "get_reactions", system = "postgresql")] pub async fn get_reactions( &self, community: CommunityId, @@ -3241,6 +3369,7 @@ impl Db { } /// Batch-fetch emoji counts for a set of (event_id, event_created_at) pairs. + #[datastore_span(name = "get_reactions_bulk", system = "postgresql")] pub async fn get_reactions_bulk( &self, community: CommunityId, @@ -3250,6 +3379,7 @@ impl Db { } /// Find events that @mention the given pubkey. + #[datastore_span(name = "query_feed_mentions", system = "postgresql")] pub async fn query_feed_mentions( &self, community: CommunityId, @@ -3276,6 +3406,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. + #[datastore_span(name = "query_feed_mentions_routed", system = "postgresql")] pub async fn query_feed_mentions_routed( &self, path: &'static str, @@ -3331,6 +3462,7 @@ impl Db { } /// Find events that require action from the given pubkey. + #[datastore_span(name = "query_feed_needs_action", system = "postgresql")] pub async fn query_feed_needs_action( &self, community: CommunityId, @@ -3353,6 +3485,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. + #[datastore_span(name = "query_feed_needs_action_routed", system = "postgresql")] pub async fn query_feed_needs_action_routed( &self, path: &'static str, @@ -3408,6 +3541,7 @@ impl Db { } /// Find recent activity across accessible channels. + #[datastore_span(name = "query_feed_activity", system = "postgresql")] pub async fn query_feed_activity( &self, community: CommunityId, @@ -3421,6 +3555,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. + #[datastore_span(name = "query_feed_activity_routed", system = "postgresql")] pub async fn query_feed_activity_routed( &self, path: &'static str, @@ -3467,6 +3602,7 @@ impl Db { /// Create a new API token record. #[allow(clippy::too_many_arguments)] + #[datastore_span(name = "create_api_token", system = "postgresql")] pub async fn create_api_token( &self, community_id: CommunityId, @@ -3492,6 +3628,7 @@ impl Db { /// Atomic conditional INSERT with 10-token limit (per (community, owner)). #[allow(clippy::too_many_arguments)] + #[datastore_span(name = "create_api_token_if_under_limit", system = "postgresql")] pub async fn create_api_token_if_under_limit( &self, community_id: CommunityId, @@ -3521,6 +3658,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. + #[datastore_span(name = "get_api_token_by_hash", system = "postgresql")] pub async fn get_api_token_by_hash( &self, community_id: CommunityId, @@ -3546,6 +3684,10 @@ impl Db { } /// Look up an API token by hash, including revoked, scoped to community. + #[datastore_span( + name = "get_api_token_by_hash_including_revoked", + system = "postgresql" + )] pub async fn get_api_token_by_hash_including_revoked( &self, community_id: CommunityId, @@ -3560,6 +3702,7 @@ impl Db { } /// Record a token usage (update `last_used_at`), scoped to community. + #[datastore_span(name = "touch_api_token", system = "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", @@ -3581,6 +3724,7 @@ impl Db { } /// List all active (non-revoked) tokens in a community, newest first. + #[datastore_span(name = "list_active_tokens", system = "postgresql")] pub async fn list_active_tokens(&self, community_id: CommunityId) -> Result> { let rows = sqlx::query( r#" @@ -3615,6 +3759,7 @@ impl Db { } /// List all tokens for a (community, owner) pair (including revoked). + #[datastore_span(name = "list_tokens_by_owner", system = "postgresql")] pub async fn list_tokens_by_owner( &self, community_id: CommunityId, @@ -3624,6 +3769,7 @@ impl Db { } /// Revoke a single token by ID, scoped to (community, owner). + #[datastore_span(name = "revoke_token", system = "postgresql")] pub async fn revoke_token( &self, community_id: CommunityId, @@ -3642,6 +3788,7 @@ impl Db { } /// Revoke all active tokens for a (community, owner) pair. + #[datastore_span(name = "revoke_all_tokens", system = "postgresql")] pub async fn revoke_all_tokens( &self, community_id: CommunityId, @@ -3658,6 +3805,7 @@ impl Db { } /// Create a new workflow. + #[datastore_span(name = "create_workflow", system = "postgresql")] pub async fn create_workflow( &self, community_id: CommunityId, @@ -3681,6 +3829,7 @@ impl Db { /// Insert or update a workflow using its NIP-33 `d`-tag UUID. #[allow(clippy::too_many_arguments)] + #[datastore_span(name = "upsert_workflow", system = "postgresql")] pub async fn upsert_workflow( &self, community_id: CommunityId, @@ -3705,6 +3854,7 @@ impl Db { } /// Fetch a single workflow by ID, scoped to its community. + #[datastore_span(name = "get_workflow", system = "postgresql")] pub async fn get_workflow( &self, community_id: CommunityId, @@ -3714,6 +3864,7 @@ impl Db { } /// List workflows for a channel. + #[datastore_span(name = "list_channel_workflows", system = "postgresql")] pub async fn list_channel_workflows( &self, community_id: CommunityId, @@ -3725,6 +3876,7 @@ impl Db { } /// List active, enabled workflows for a channel. + #[datastore_span(name = "list_enabled_channel_workflows", system = "postgresql")] pub async fn list_enabled_channel_workflows( &self, community_id: CommunityId, @@ -3734,6 +3886,7 @@ impl Db { } /// List all active, enabled schedule-triggered workflows. + #[datastore_span(name = "list_all_enabled_workflows", system = "postgresql")] pub async fn list_all_enabled_workflows(&self) -> Result> { workflow::list_all_enabled_workflows(&self.pool).await } @@ -3746,6 +3899,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. + #[datastore_span(name = "claim_scheduled_workflow_fire", system = "postgresql")] pub async fn claim_scheduled_workflow_fire( &self, community_id: CommunityId, @@ -3762,6 +3916,7 @@ impl Db { } /// Fetch the latest claimed schedule instant for interval trigger anchoring. + #[datastore_span(name = "latest_scheduled_workflow_fire", system = "postgresql")] pub async fn latest_scheduled_workflow_fire( &self, community_id: CommunityId, @@ -3771,6 +3926,7 @@ impl Db { } /// Attach the workflow run id created from a won scheduled-fire claim. + #[datastore_span(name = "attach_scheduled_workflow_run", system = "postgresql")] pub async fn attach_scheduled_workflow_run( &self, community_id: CommunityId, @@ -3789,6 +3945,7 @@ impl Db { } /// Delete old scheduled workflow fire claims before a retention cutoff. + #[datastore_span(name = "prune_scheduled_workflow_fires_before", system = "postgresql")] pub async fn prune_scheduled_workflow_fires_before( &self, older_than: chrono::DateTime, @@ -3797,6 +3954,7 @@ impl Db { } /// Update a workflow's name, definition, and hash. + #[datastore_span(name = "update_workflow", system = "postgresql")] pub async fn update_workflow( &self, community_id: CommunityId, @@ -3817,6 +3975,7 @@ impl Db { } /// Update a workflow's status. + #[datastore_span(name = "update_workflow_status", system = "postgresql")] pub async fn update_workflow_status( &self, community_id: CommunityId, @@ -3827,6 +3986,7 @@ impl Db { } /// Enable or disable a workflow. + #[datastore_span(name = "set_workflow_enabled", system = "postgresql")] pub async fn set_workflow_enabled( &self, community_id: CommunityId, @@ -3838,6 +3998,7 @@ impl Db { /// Disable all of an owner's workflows in a channel (SEC-006, on /// membership loss). Returns the number of workflows disabled. + #[datastore_span(name = "disable_workflows_for_owner_in_channel", system = "postgresql")] pub async fn disable_workflows_for_owner_in_channel( &self, community_id: CommunityId, @@ -3854,12 +4015,14 @@ impl Db { } /// Delete a workflow and all its runs/approvals. + #[datastore_span(name = "delete_workflow", system = "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`. + #[datastore_span(name = "delete_workflow_for_owner", system = "postgresql")] pub async fn delete_workflow_for_owner( &self, community_id: CommunityId, @@ -3871,6 +4034,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). + #[datastore_span(name = "find_workflow_by_owner_and_name", system = "postgresql")] pub async fn find_workflow_by_owner_and_name( &self, community_id: CommunityId, @@ -3881,6 +4045,7 @@ impl Db { } /// Create a new workflow run. + #[datastore_span(name = "create_workflow_run", system = "postgresql")] pub async fn create_workflow_run( &self, community_id: CommunityId, @@ -3899,6 +4064,7 @@ impl Db { } /// Fetch a single workflow run, scoped to its community. + #[datastore_span(name = "get_workflow_run", system = "postgresql")] pub async fn get_workflow_run( &self, community_id: CommunityId, @@ -3908,6 +4074,7 @@ impl Db { } /// List runs for a workflow. + #[datastore_span(name = "list_workflow_runs", system = "postgresql")] pub async fn list_workflow_runs( &self, community_id: CommunityId, @@ -3918,6 +4085,7 @@ impl Db { } /// Update a workflow run's status. + #[datastore_span(name = "update_workflow_run", system = "postgresql")] pub async fn update_workflow_run( &self, community_id: CommunityId, @@ -3940,11 +4108,13 @@ impl Db { } /// Create an approval request. + #[datastore_span(name = "create_approval", system = "postgresql")] pub async fn create_approval(&self, params: workflow::CreateApprovalParams<'_>) -> Result<()> { workflow::create_approval(&self.pool, params).await } /// Fetch an approval by raw token. + #[datastore_span(name = "get_approval", system = "postgresql")] pub async fn get_approval( &self, community_id: CommunityId, @@ -3954,6 +4124,7 @@ impl Db { } /// Fetch an approval by its already-hashed token (no re-hashing). + #[datastore_span(name = "get_approval_by_stored_hash", system = "postgresql")] pub async fn get_approval_by_stored_hash( &self, community_id: CommunityId, @@ -3963,6 +4134,7 @@ impl Db { } /// Fetch all approvals for a workflow run. + #[datastore_span(name = "get_run_approvals", system = "postgresql")] pub async fn get_run_approvals( &self, community_id: CommunityId, @@ -3973,6 +4145,7 @@ impl Db { } /// Update an approval's status. + #[datastore_span(name = "update_approval", system = "postgresql")] pub async fn update_approval( &self, community_id: CommunityId, @@ -3993,6 +4166,7 @@ impl Db { } /// Update an approval by its already-hashed token (no re-hashing). + #[datastore_span(name = "update_approval_by_stored_hash", system = "postgresql")] pub async fn update_approval_by_stored_hash( &self, community_id: CommunityId, @@ -4013,6 +4187,7 @@ impl Db { } /// Ensures monthly partitions exist for the next N months. + #[datastore_span(name = "ensure_future_partitions", system = "postgresql")] pub async fn ensure_future_partitions(&self, months_ahead: u32) -> Result<()> { partition::ensure_future_partitions(&self.pool, months_ahead).await } @@ -4021,6 +4196,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. + #[datastore_span(name = "backfill_d_tags", system = "postgresql")] pub async fn backfill_d_tags(&self) -> Result { let result = sqlx::query( "UPDATE events \ @@ -4037,6 +4213,7 @@ impl Db { } /// Check if a pubkey is in the allowlist for `community`. + #[datastore_span(name = "is_pubkey_allowed", system = "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", @@ -4050,6 +4227,7 @@ impl Db { } /// Check if the community allowlist has any entries (i.e. is enforcement active). + #[datastore_span(name = "has_allowlist_entries", system = "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") @@ -4061,6 +4239,7 @@ impl Db { } /// Add a pubkey to the community allowlist. + #[datastore_span(name = "add_to_allowlist", system = "postgresql")] pub async fn add_to_allowlist( &self, community: CommunityId, @@ -4082,6 +4261,7 @@ impl Db { } /// Remove a pubkey from the community allowlist. + #[datastore_span(name = "remove_from_allowlist", system = "postgresql")] pub async fn remove_from_allowlist( &self, community: CommunityId, @@ -4097,6 +4277,7 @@ impl Db { } /// List all pubkeys in the community allowlist. + #[datastore_span(name = "list_allowlist", system = "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", @@ -4125,6 +4306,7 @@ impl Db { /// `B`; everything else fails closed to the writer, exactly like /// [`Db::query_events_routed_bounded`]. Not precedent for routing other /// permission reads. + #[datastore_span(name = "is_relay_member", system = "postgresql")] pub async fn is_relay_member(&self, community: CommunityId, pubkey: &str) -> Result { let path = "relay_membership"; match self.route_read(path, RoutePredicate::Bounded).await { @@ -4148,6 +4330,7 @@ impl Db { } /// Returns the relay member record for `pubkey` in `community`, or `None` if not found. + #[datastore_span(name = "get_relay_member", system = "postgresql")] pub async fn get_relay_member( &self, community: CommunityId, @@ -4157,6 +4340,7 @@ impl Db { } /// Returns all relay members of `community` ordered by `created_at` ascending. + #[datastore_span(name = "list_relay_members", system = "postgresql")] pub async fn list_relay_members( &self, community: CommunityId, @@ -4168,6 +4352,7 @@ impl Db { /// /// Returns `true` if the row was actually inserted, `false` if the pubkey /// already existed in `community` (idempotent — `ON CONFLICT DO NOTHING`). + #[datastore_span(name = "add_relay_member", system = "postgresql")] pub async fn add_relay_member( &self, community: CommunityId, @@ -4180,6 +4365,7 @@ impl Db { /// Claims relay membership via an invite and atomically persists the /// accepted policy version when a policy is configured. + #[datastore_span(name = "claim_relay_membership", system = "postgresql")] pub async fn claim_relay_membership( &self, community: CommunityId, @@ -4213,6 +4399,7 @@ impl Db { } /// Returns whether a member has persisted acceptance evidence for a policy version. + #[datastore_span(name = "has_join_policy_acceptance", system = "postgresql")] pub async fn has_join_policy_acceptance( &self, community: CommunityId, @@ -4224,6 +4411,7 @@ impl Db { } /// Removes a relay member from `community` atomically, refusing to delete the owner. + #[datastore_span(name = "remove_relay_member", system = "postgresql")] pub async fn remove_relay_member( &self, community: CommunityId, @@ -4236,6 +4424,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`]. + #[datastore_span(name = "remove_relay_member_if_role", system = "postgresql")] pub async fn remove_relay_member_if_role( &self, community: CommunityId, @@ -4247,6 +4436,7 @@ impl Db { } /// Updates the role of an existing relay member in `community`. Returns `true` if updated. + #[datastore_span(name = "update_relay_member_role", system = "postgresql")] pub async fn update_relay_member_role( &self, community: CommunityId, @@ -4257,6 +4447,7 @@ impl Db { } /// Ensures the owner pubkey exists with role `"owner"` in `community`. Called at startup. + #[datastore_span(name = "bootstrap_owner", system = "postgresql")] pub async fn bootstrap_owner(&self, community: CommunityId, owner_pubkey: &str) -> Result<()> { relay_members::bootstrap_owner(&self.pool, community, owner_pubkey).await } @@ -4271,6 +4462,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. + #[datastore_span(name = "transfer_ownership", system = "postgresql")] pub async fn transfer_ownership( &self, community: CommunityId, @@ -4290,6 +4482,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. + #[datastore_span(name = "backfill_from_allowlist", system = "postgresql")] pub async fn backfill_from_allowlist(&self, community: CommunityId) -> Result { relay_members::backfill_from_allowlist(&self.pool, community).await } @@ -4299,6 +4492,7 @@ impl Db { /// /// `max_uses` is `None` for unlimited or `Some(1..=10000)`. /// `ttl_secs` must be in the shared invite lifetime range. + #[datastore_span(name = "mint_relay_invite", system = "postgresql")] pub async fn mint_relay_invite( &self, community: CommunityId, @@ -4310,6 +4504,7 @@ impl Db { } /// Delete one bounded batch of invites expired before `cutoff`. + #[datastore_span(name = "reap_expired_relay_invites", system = "postgresql")] pub async fn reap_expired_relay_invites( &self, cutoff: chrono::DateTime, @@ -4322,6 +4517,7 @@ impl Db { /// transaction with `FOR UPDATE` on the invite row. /// /// `token_hash` is the SHA-256 of the presented v2 code (32 bytes). + #[datastore_span(name = "claim_relay_invite", system = "postgresql")] pub async fn claim_relay_invite( &self, community: CommunityId, @@ -4361,6 +4557,7 @@ impl Db { } /// Sidecar an accepted product-feedback event, idempotent by event id. + #[datastore_span(name = "insert_product_feedback", system = "postgresql")] pub async fn insert_product_feedback( &self, community: CommunityId, @@ -4370,6 +4567,7 @@ impl Db { } /// List product feedback across the deployment, newest first. + #[datastore_span(name = "list_product_feedback", system = "postgresql")] pub async fn list_product_feedback( &self, limit: i64, @@ -4378,6 +4576,7 @@ impl Db { } /// Insert a tenant-scoped NIP-56 report row, idempotent by report event id. + #[datastore_span(name = "insert_moderation_report", system = "postgresql")] pub async fn insert_moderation_report( &self, community: CommunityId, @@ -4387,6 +4586,7 @@ impl Db { } /// List moderation reports for a community, newest first. + #[datastore_span(name = "list_moderation_reports", system = "postgresql")] pub async fn list_moderation_reports( &self, community: CommunityId, @@ -4397,6 +4597,7 @@ impl Db { } /// Fetch one moderation report by row id. + #[datastore_span(name = "get_moderation_report", system = "postgresql")] pub async fn get_moderation_report( &self, community: CommunityId, @@ -4406,6 +4607,7 @@ impl Db { } /// Fetch one moderation report by signed NIP-56 report event id. + #[datastore_span(name = "get_moderation_report_by_event", system = "postgresql")] pub async fn get_moderation_report_by_event( &self, community: CommunityId, @@ -4415,6 +4617,7 @@ impl Db { } /// Resolve, dismiss, or escalate an open moderation report. + #[datastore_span(name = "resolve_moderation_report", system = "postgresql")] pub async fn resolve_moderation_report( &self, community: CommunityId, @@ -4435,6 +4638,7 @@ impl Db { } /// Upsert a community ban for a member pubkey. + #[datastore_span(name = "ban_community_member", system = "postgresql")] pub async fn ban_community_member( &self, community: CommunityId, @@ -4447,6 +4651,7 @@ impl Db { } /// Lift a community ban for a member pubkey. + #[datastore_span(name = "unban_community_member", system = "postgresql")] pub async fn unban_community_member( &self, community: CommunityId, @@ -4457,6 +4662,7 @@ impl Db { } /// Upsert a community timeout/write-block for a member pubkey. + #[datastore_span(name = "timeout_community_member", system = "postgresql")] pub async fn timeout_community_member( &self, community: CommunityId, @@ -4469,6 +4675,7 @@ impl Db { } /// Clear a community timeout/write-block for a member pubkey. + #[datastore_span(name = "untimeout_community_member", system = "postgresql")] pub async fn untimeout_community_member( &self, community: CommunityId, @@ -4479,6 +4686,7 @@ impl Db { } /// Fetch the active ban/timeout restriction state for enforcement hot paths. + #[datastore_span(name = "moderation_restriction_state", system = "postgresql")] pub async fn moderation_restriction_state( &self, community: CommunityId, @@ -4488,6 +4696,7 @@ impl Db { } /// Fetch the full ban/timeout row for a member pubkey. + #[datastore_span(name = "get_community_ban", system = "postgresql")] pub async fn get_community_ban( &self, community: CommunityId, @@ -4497,6 +4706,7 @@ impl Db { } /// List currently restricted members in a community. + #[datastore_span(name = "list_community_restrictions", system = "postgresql")] pub async fn list_community_restrictions( &self, community: CommunityId, @@ -4505,6 +4715,7 @@ impl Db { } /// Insert a moderation audit action row. + #[datastore_span(name = "insert_moderation_action", system = "postgresql")] pub async fn insert_moderation_action( &self, community: CommunityId, @@ -4514,6 +4725,7 @@ impl Db { } /// List moderation audit action rows, newest first. + #[datastore_span(name = "list_moderation_actions", system = "postgresql")] pub async fn list_moderation_actions( &self, community: CommunityId, @@ -4524,6 +4736,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`]. + #[datastore_span(name = "repo_name_owner", system = "postgresql")] pub async fn repo_name_owner( &self, community: CommunityId, @@ -4536,6 +4749,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`. + #[datastore_span(name = "reserve_repo_name", system = "postgresql")] pub async fn reserve_repo_name( &self, community: CommunityId, @@ -4546,6 +4760,7 @@ impl Db { } /// Count git repos reserved by `owner_pubkey` in `community` (quota check). + #[datastore_span(name = "count_repos_for_owner", system = "postgresql")] pub async fn count_repos_for_owner( &self, community: CommunityId, @@ -4557,6 +4772,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`]. + #[datastore_span(name = "release_repo_name", system = "postgresql")] pub async fn release_repo_name( &self, community: CommunityId, @@ -4567,12 +4783,14 @@ impl Db { } /// Returns `true` if `pubkey` (64-char hex) is archived in `community_id`. + #[datastore_span(name = "is_archived", system = "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)] + #[datastore_span(name = "archive", system = "postgresql")] pub async fn archive( &self, community_id: CommunityId, @@ -4597,11 +4815,13 @@ impl Db { } /// Unarchives an identity from `community_id`. Returns `true` if deleted, `false` if absent. + #[datastore_span(name = "unarchive", system = "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. + #[datastore_span(name = "list_archived", system = "postgresql")] pub async fn list_archived( &self, community_id: CommunityId, @@ -4610,6 +4830,7 @@ impl Db { } /// Soft-delete NIP-29 discovery events for a channel created by a specific relay pubkey. + #[datastore_span(name = "soft_delete_discovery_events", system = "postgresql")] pub async fn soft_delete_discovery_events( &self, community_id: CommunityId, @@ -4635,6 +4856,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. + #[datastore_span(name = "replace_addressable_event", system = "postgresql")] pub async fn replace_addressable_event( &self, community_id: CommunityId, @@ -4767,6 +4989,10 @@ impl Db { /// Snapshot and canonical rows are compared directly rather than by /// timestamp: relay membership events use whole-second Nostr timestamps, /// and multiple mutations within one second must still be repaired. + #[datastore_span( + name = "nip43_membership_snapshot_needs_reconciliation", + system = "postgresql" + )] pub async fn nip43_membership_snapshot_needs_reconciliation( &self, community_id: CommunityId, @@ -4817,6 +5043,7 @@ impl Db { /// prevents the stale-snapshot race where a concurrent publication reads /// older state and overwrites a newer snapshot by arrival order. /// + #[datastore_span(name = "publish_nip43_membership_locked", system = "postgresql")] pub async fn publish_nip43_membership_locked( &self, community_id: CommunityId, @@ -4957,6 +5184,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. + #[datastore_span(name = "replace_parameterized_event", system = "postgresql")] pub async fn replace_parameterized_event( &self, community_id: CommunityId, diff --git a/crates/buzz-db/src/replica_fence.rs b/crates/buzz-db/src/replica_fence.rs index c840db393..98bdd9850 100644 --- a/crates/buzz-db/src/replica_fence.rs +++ b/crates/buzz-db/src/replica_fence.rs @@ -63,6 +63,8 @@ use chrono::{DateTime, Utc}; use sqlx::{PgConnection, PgPool, Row}; use uuid::Uuid; +use buzz_datastore_tracing::datastore_span; + /// Seconds of `created_at` history the commit-time floor guard tolerates. /// /// Must exceed the relay's ingest envelope (±900 s) by enough slack that a @@ -315,6 +317,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. +#[datastore_span(name = "replica_fence_verify_catalog", system = "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 +372,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. +#[datastore_span(name = "replica_fence_verify_behavior", system = "postgresql")] pub async fn verify_floor_guard_behavior(pool: &PgPool) -> crate::Result<()> { use crate::error::DbError; @@ -698,6 +702,10 @@ 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. +#[datastore_span( + name = "replica_fence_reader_supports_aurora_identity", + system = "postgresql" +)] pub async fn reader_supports_aurora_identity(conn: &mut PgConnection) -> Result { match sqlx::query(sqlx::AssertSqlSafe(format!( "SELECT {AURORA_IDENTITY_FN}()" diff --git a/crates/buzz-relay/Cargo.toml b/crates/buzz-relay/Cargo.toml index ec4992e4c..582feaa7f 100644 --- a/crates/buzz-relay/Cargo.toml +++ b/crates/buzz-relay/Cargo.toml @@ -19,6 +19,7 @@ path = "src/main.rs" buzz-core = { workspace = true } buzz-conformance = { workspace = true } buzz-db = { workspace = true } +buzz-datastore-tracing = { workspace = true } buzz-auth = { workspace = true } buzz-pubsub = { workspace = true } buzz-audit = { workspace = true } diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 2d8273680..efdb307e1 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -19,6 +19,7 @@ use uuid::Uuid; use buzz_core::kind::*; use buzz_core::tenant::{CommunityId, TenantContext}; +use buzz_datastore_tracing::datastore_span; use buzz_db::workflow::{ApprovalStatus, RunStatus}; use buzz_db::DbError; use buzz_workflow::executor::TriggerContext; @@ -97,6 +98,7 @@ 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). +#[datastore_span(name = "persist_command_event", system = "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 e42bcc804..e28c5b684 100644 --- a/crates/buzz-search/Cargo.toml +++ b/crates/buzz-search/Cargo.toml @@ -9,9 +9,11 @@ description = "Postgres full-text search for Buzz, scoped by community" [dependencies] buzz-core = { workspace = true } +buzz-datastore-tracing = { 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 d191353af..bd95e8cdb 100644 --- a/crates/buzz-search/src/query.rs +++ b/crates/buzz-search/src/query.rs @@ -8,10 +8,12 @@ //! //! See conformance row 50. -use buzz_core::CommunityId; use sqlx::{PgPool, QueryBuilder, Row}; use uuid::Uuid; +use buzz_core::CommunityId; +use buzz_datastore_tracing::datastore_span; + use crate::error::SearchError; /// Channel-scope filter for a community-scoped FTS query. @@ -213,6 +215,7 @@ fn normalized_search_text(q: &str) -> Option { /// /// `community_id = $ctx` is the first predicate and is non-negotiable. There /// is no code path through this function that omits it. +#[datastore_span(name = "search", system = "postgresql")] pub async fn search(pool: &PgPool, query: &SearchQuery) -> Result { let Some(search_text) = normalized_search_text(&query.q) else { return Ok(SearchResult { diff --git a/docs/CORPORATE_IDENTITY.md b/docs/CORPORATE_IDENTITY.md deleted file mode 100644 index 00b47b42b..000000000 --- a/docs/CORPORATE_IDENTITY.md +++ /dev/null @@ -1,78 +0,0 @@ -# Corporate identity - -Corporate identity is an optional relay policy enabled with -`BUZZ_REQUIRE_CORPORATE_IDENTITY=true`. The relay verifies an asymmetric JWT -after the request proves control of a Nostr key, then admits the request only -when the existing community policy also succeeds. - -## Required JWT policy - -- `BUZZ_CORPORATE_IDENTITY_JWKS_URI` must be HTTPS and contain no credentials. -- JWTs must have a supported asymmetric algorithm, a `kid`, and valid `exp`, - `iss`, and `aud` claims. A present `nbf` claim is enforced. -- `BUZZ_CORPORATE_IDENTITY_NPUB_CLAIM`, when configured, is mandatory and must - equal the authenticated Nostr key. Leaving it unset enables first-use - uid-to-key enrollment in the private binding table. -- JWKS requests have connect and total timeouts, reject redirects, cap the - response at 1 MiB, cache keys for five minutes, and coalesce refreshes. - -`BUZZ_REQUIRE_CORPORATE_IDENTITY` and -`BUZZ_ALLOW_CORPORATE_IDENTITY_DELEGATION` are strict booleans. Misspellings and -non-UTF-8 values stop configuration loading instead of silently disabling a -gate. - -## Binding and revocation lifecycle - -JWT validation is read-only. The relay creates or refreshes a binding only -after admission, allowlist, role, and community membership checks succeed. -Invite claims commit the binding, membership, policy evidence, and invite use -in one PostgreSQL transaction. - -Revocation has three explicit meanings: - -- `principal` disables every key for an issuer-qualified uid. Normal - authentication cannot re-enroll the principal with another key. -- `key` revokes one key but does not silently authorize a replacement. -- `rotation` is the audit state written by an explicit atomic old-key to - new-key rotation. - -WebSocket and audio sessions revalidate the authoritative binding at least -every 30 seconds. Direct sessions also close at JWT expiry. Delegated sessions -check the owner's binding, so disabling an owner evicts the owner's agents as -well as the direct owner session. - -Corporate NIP-OA delegation is transport-wide and therefore accepts only an -empty conditions string. Conditional tags must be evaluated for a specific -operation and are not treated as blanket corporate identity authority. - -## Privacy and public assertions - -`BUZZ_CORPORATE_IDENTITY_DISPLAY_CLAIM` is private. Its default (`email`) is -stored only in the community-scoped binding table and audit data; it is not -published to Nostr. - -Public projection is separately opt-in with -`BUZZ_CORPORATE_IDENTITY_PUBLIC_DISPLAY_CLAIM`. When set, that claim is -published as a relay-signed NIP-85 label. Assertions carry both `active=true` -and an `expiration` no later than one hour or the JWT's expiry, whichever comes -first. Clients require the relay signature, active status, and a future -expiration. Removing the opt-in publishes an inactive replacement only when a -prior public assertion exists. - -NIP-85 events are replaceable relay events and may also remain in downstream -caches or archives after replacement. Operators must choose a non-sensitive, -user-approved public label and account for that retention when configuring the -public claim. - -## Route policy - -Corporate identity applies to authenticated WebSocket and audio connections, -the NIP-98 event/query/count bridge, moderation reads, invite mint and claim, -Git smart HTTP, media uploads, and protected media reads. - -Intentional exemptions are public media reads when media GET authentication is -disabled, health/readiness/metrics endpoints, NIP-11 and NIP-05 discovery, -operator and admin control planes with their own authentication, secret-backed -workflow hooks, public join-policy documents, invite policy-acceptance -callbacks, and static local web callbacks. These exemptions must remain in the -central route-policy test matrix when routes change.