diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 1e8cb1e571..8798828f23 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -1222,175 +1222,6 @@ impl Db { Ok(result) } - /// 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, - lease_until: DateTime, - ) -> Result> { - push::claim_due_match_batch(&self.pool, limit, lease_until).await - } - - /// 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, - ) -> Result> { - push::active_match_leases(&self.pool, community).await - } - - /// 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, - claim_id: uuid::Uuid, - event_ids: &[Vec], - ) -> Result { - push::complete_match_batch(&self.pool, community, claim_id, event_ids).await - } - - /// 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, - claim_id: uuid::Uuid, - event_ids: &[Vec], - next: DateTime, - ) -> Result { - push::retry_match_batch(&self.pool, community, claim_id, event_ids, next).await - } - - /// 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, - author: &[u8], - installation_id: &str, - wake: push::NewWake<'_>, - ) -> Result { - push::enqueue_wake(&self.pool, community, author, installation_id, wake).await - } - - /// 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, - requests: &[push::WakeRequest], - ) -> Result> { - push::enqueue_wakes(&self.pool, community, requests).await - } - - /// 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, - limit: i64, - lease_until: DateTime, - ) -> Result> { - push::claim_due_wakes(&self.pool, community, limit, lease_until).await - } - - /// 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, - id: Uuid, - claim_id: Uuid, - ) -> Result { - push::revalidate_wake_for_send(&self.pool, community, id, claim_id).await - } - - /// Mark a fenced wake claim delivered. - #[datastore_span(name = "complete_push_wake", system = "postgresql")] - pub async fn complete_push_wake( - &self, - community: CommunityId, - id: Uuid, - claim_id: Uuid, - ) -> Result { - push::complete_wake(&self.pool, community, id, claim_id).await - } - - /// 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, - id: Uuid, - claim_id: Uuid, - next: DateTime, - ) -> Result { - push::retry_wake(&self.pool, community, id, claim_id, next).await - } - - /// Mark a fenced wake claim terminally failed. - #[datastore_span(name = "fail_push_wake", system = "postgresql")] - pub async fn fail_push_wake( - &self, - community: CommunityId, - id: Uuid, - claim_id: Uuid, - ) -> Result { - push::fail_wake(&self.pool, community, id, claim_id).await - } - - /// 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, - author: &[u8], - installation_id: &str, - generation: i64, - ) -> Result { - push::disable_endpoint_generation( - &self.pool, - community, - author, - installation_id, - generation, - ) - .await - } - - /// 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, - event: &nostr::Event, - installation_id: &str, - version: push::LeaseVersion<'_>, - active: Option>, - max_active_leases: i64, - ) -> Result { - push::accept_lease_event( - &self.pool, - community, - event, - installation_id, - version, - active, - max_active_leases, - ) - .await - } - /// Shared route decision for one read: evaluate the predicate against a /// proved reader session and record the decision. Fail closed to the /// writer everywhere. diff --git a/crates/buzz-db/src/push.rs b/crates/buzz-db/src/push.rs index 3aa6cd9b3f..fc94843a9c 100644 --- a/crates/buzz-db/src/push.rs +++ b/crates/buzz-db/src/push.rs @@ -11,6 +11,8 @@ use sqlx::{PgPool, Row as _}; use uuid::Uuid; use crate::error::Result; +use crate::Db; +use buzz_datastore_tracing::datastore_span; /// Namespace for the per-community push-gate advisory lock. Must match the /// key built inside the `enqueue_push_match_job` trigger (migration 0023): @@ -1278,6 +1280,177 @@ fn row_to_claimed_wake(row: sqlx::postgres::PgRow) -> Result { }) } +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, + lease_until: DateTime, + ) -> Result> { + crate::push::claim_due_match_batch(&self.pool, limit, lease_until).await + } + + /// 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, + ) -> Result> { + crate::push::active_match_leases(&self.pool, community).await + } + + /// 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, + claim_id: uuid::Uuid, + event_ids: &[Vec], + ) -> Result { + crate::push::complete_match_batch(&self.pool, community, claim_id, event_ids).await + } + + /// 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, + claim_id: uuid::Uuid, + event_ids: &[Vec], + next: DateTime, + ) -> Result { + crate::push::retry_match_batch(&self.pool, community, claim_id, event_ids, next).await + } + + /// 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 { + crate::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, + author: &[u8], + installation_id: &str, + wake: crate::push::NewWake<'_>, + ) -> Result { + crate::push::enqueue_wake(&self.pool, community, author, installation_id, wake).await + } + + /// 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, + requests: &[crate::push::WakeRequest], + ) -> Result> { + crate::push::enqueue_wakes(&self.pool, community, requests).await + } + + /// 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, + limit: i64, + lease_until: DateTime, + ) -> Result> { + crate::push::claim_due_wakes(&self.pool, community, limit, lease_until).await + } + + /// 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, + id: Uuid, + claim_id: Uuid, + ) -> Result { + crate::push::revalidate_wake_for_send(&self.pool, community, id, claim_id).await + } + + /// Mark a fenced wake claim delivered. + #[datastore_span(name = "complete_push_wake", system = "postgresql")] + pub async fn complete_push_wake( + &self, + community: CommunityId, + id: Uuid, + claim_id: Uuid, + ) -> Result { + crate::push::complete_wake(&self.pool, community, id, claim_id).await + } + + /// 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, + id: Uuid, + claim_id: Uuid, + next: DateTime, + ) -> Result { + crate::push::retry_wake(&self.pool, community, id, claim_id, next).await + } + + /// Mark a fenced wake claim terminally failed. + #[datastore_span(name = "fail_push_wake", system = "postgresql")] + pub async fn fail_push_wake( + &self, + community: CommunityId, + id: Uuid, + claim_id: Uuid, + ) -> Result { + crate::push::fail_wake(&self.pool, community, id, claim_id).await + } + + /// 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, + author: &[u8], + installation_id: &str, + generation: i64, + ) -> Result { + crate::push::disable_endpoint_generation( + &self.pool, + community, + author, + installation_id, + generation, + ) + .await + } + + /// 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, + event: &nostr::Event, + installation_id: &str, + version: crate::push::LeaseVersion<'_>, + active: Option>, + max_active_leases: i64, + ) -> Result { + crate::push::accept_lease_event( + &self.pool, + community, + event, + installation_id, + version, + active, + max_active_leases, + ) + .await + } +} + #[cfg(test)] mod tests { use super::*;