perf(relay): deliver operator-listener mentions on commit

crates/buzz-db/src/runtime/admitted_tx.rs

@@ -76,6 +76,8 @@ pub struct AdmittedTx {
     ttl_refresh_channels: BTreeSet<Uuid>,
     push_enqueue: Option<super::push_enqueue::PushEnqueue>,
     push_events: Vec<Vec<u8>>,
+    operator_listener_commits: Option<crate::operator_listener::CommittedDeliveries>,
+    operator_listener_deliveries: Vec<Uuid>,
 }
 
 impl AdmittedTx {
@@ -92,6 +94,8 @@ impl AdmittedTx {
             ttl_refresh_channels: BTreeSet::new(),
             push_enqueue: None,
             push_events: Vec::new(),
+            operator_listener_commits: None,
+            operator_listener_deliveries: Vec::new(),
         })
     }
 
@@ -111,13 +115,27 @@ impl AdmittedTx {
             ttl_refresh_channels: BTreeSet::new(),
             push_enqueue: None,
             push_events: Vec::new(),
+            operator_listener_commits: None,
+            operator_listener_deliveries: Vec::new(),
         })
     }
 
     pub(crate) fn set_push_enqueue(&mut self, producer: Option<super::push_enqueue::PushEnqueue>) {
         self.push_enqueue = producer;
     }
 
+    /// Arm the post-commit hand-offs a [`crate::Db`] handle owns.
+    pub(crate) fn set_post_commit(&mut self, db: &crate::Db) {
+        self.set_push_enqueue(db.push_enqueue.clone());
+        self.operator_listener_commits = Some(db.operator_listener_commits.clone());
+    }
+
+    pub(crate) fn record_operator_listener_deliveries(&mut self, ids: Vec<Uuid>) {
+        if self.operator_listener_commits.is_some() {
+            self.operator_listener_deliveries.extend(ids);
+        }
+    }
+
     pub(crate) fn record_push_event(&mut self, event_id: &[u8], kind: i32) {
         if self.push_enqueue.is_some()
             && crate::store::event_follow_up::PUSH_MATCH_KINDS.contains(&kind)
@@ -174,6 +192,9 @@ impl AdmittedTx {
                 producer.submit(self.community, event_id);
             }
         }
+        if let Some(commits) = self.operator_listener_commits {
+            commits.submit(self.operator_listener_deliveries);
+        }
         Ok(())
     }
 

crates/buzz-db/src/runtime/mod.rs

@@ -262,6 +262,9 @@ pub struct Db {
     pub(crate) pool: PgPool,
     /// Optional bounded post-commit producer, shared across handle clones.
     pub(crate) push_enqueue: Option<push_enqueue::PushEnqueue>,
+    /// Post-commit hand-off of new operator-listener outbox rows to this
+    /// process's delivery worker, shared across handle clones.
+    pub(crate) operator_listener_commits: crate::operator_listener::CommittedDeliveries,
     /// Maximum connections configured for this pool (from [`DbConfig::max_connections`]).
     pub(crate) max_connections: u32,
     /// Optional read-replica pool (from [`DbConfig::read_database_url`]).
@@ -783,6 +786,7 @@ impl Db {
             .then(|| push_enqueue::PushEnqueue::new(&pool));
         Ok(Self {
             push_enqueue,
+            operator_listener_commits: Default::default(),
             pool,
             max_connections: config.max_connections,
             read_pool,
@@ -1026,6 +1030,7 @@ impl Db {
     pub fn from_pool(pool: PgPool) -> Self {
         Self {
             push_enqueue: None,
+            operator_listener_commits: Default::default(),
             max_connections: pool.options().get_max_connections(),
             read_max_connections: pool.options().get_max_connections(),
             pool,
@@ -1047,6 +1052,7 @@ impl Db {
     pub fn from_pools(pool: PgPool, read_pool: PgPool) -> Self {
         Self {
             push_enqueue: None,
+            operator_listener_commits: Default::default(),
             max_connections: pool.options().get_max_connections(),
             read_max_connections: read_pool.options().get_max_connections(),
             pool,
@@ -1413,7 +1419,7 @@ impl Db {
             observability::WriterOperation::EventWrite,
         )
         .await?;
-        tx.set_push_enqueue(self.push_enqueue.clone());
+        tx.set_post_commit(self);
         Ok(tx)
     }
 
@@ -1458,7 +1464,7 @@ impl Db {
         let tx = sqlx::Transaction::begin(connection, None).await?;
         let mut tx =
             AdmittedTx::admit_with_serving_lease(tx, &self.deletion_store(), lease).await?;
-        tx.set_push_enqueue(self.push_enqueue.clone());
+        tx.set_post_commit(self);
         event::acquire_canvas_event_write_lock_if_needed(&mut tx, event, channel_id).await?;
         let result =
             event::insert_event_with_thread_metadata_tx(&mut tx, event, channel_id, None).await?;

crates/buzz-db/src/store/event.rs

@@ -2177,7 +2177,7 @@ impl Db {
             crate::observability::WriterOperation::EventWrite,
         )
         .await?;
-        tx.set_push_enqueue(self.push_enqueue.clone());
+        tx.set_post_commit(self);
         let result = crate::event::insert_event_in_transaction(&mut tx, event, channel_id).await?;
         if result.1 {
             crate::insert_mentions_in_transaction(&mut tx, event, channel_id).await?;
@@ -2703,7 +2703,7 @@ impl Db {
             crate::observability::WriterOperation::EventWrite,
         )
         .await?;
-        tx.set_push_enqueue(self.push_enqueue.clone());
+        tx.set_post_commit(self);
         crate::event::acquire_canvas_event_write_lock_if_needed(&mut tx, event, channel_id).await?;
         let result = crate::event::insert_event_with_thread_metadata_tx(
             &mut tx,