Merge main and reconcile owner deletion migration assertions

Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>

crates/buzz-db/src/lib.rs

@@ -63,9 +63,10 @@ pub(crate) use runtime::{
 };
 pub use store::{
     admin_moderation, allowlist, api_token, archived_identities, channel, channel_members,
-    community, deletion, dm, event, feed, git_repo, moderation, partition, product_feedback, push,
-    reaction, read_state, relay_admin_actions, relay_invite, relay_members, relay_operators,
-    reminder, replaceable, storage_accounting, thread, thread_window, usage, user, workflow,
+    community, deletion, dm, event, feed, git_repo, moderation, operator_listener, partition,
+    product_feedback, push, reaction, read_state, relay_admin_actions, relay_invite, relay_members,
+    relay_operators, reminder, replaceable, storage_accounting, thread, thread_window, usage, user,
+    workflow,
 };
 
 pub use allowlist::AllowlistEntry;

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

@@ -490,6 +490,8 @@ mod postgres_tests {
             "relay_admin_outbox",
             "relay_operator_audit",
             "storage_accounting_snapshots",
+            "operator_listener_pubkeys",
+            "operator_listener_outbox",
         ] {
             if normalized[insert_pos..].contains(&format!("'{value}'")) {
                 globals.insert(value.to_owned());
@@ -703,14 +705,14 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 50);
+        assert_eq!(migrations.len(), 51);
         assert_eq!(migrations[48].version, 49);
-        assert_eq!(migrations[49].version, 51);
+        assert_eq!(migrations[50].version, 51);
         assert!(migrations[48]
             .sql
             .as_str()
             .contains("idx_thread_metadata_window"));
-        assert!(migrations[49]
+        assert!(migrations[50]
             .sql
             .as_str()
             .contains("community_deletion_owner_provenance"));
@@ -1298,10 +1300,63 @@ mod postgres_tests {
             .sql
             .as_str()
             .contains("CREATE TABLE storage_accounting_snapshots"));
-        // schema.sql exclusion list must match the restored (pre-0041) body.
+        assert_eq!(migrations[47].version, 48);
+        assert!(migrations[47]
+            .sql
+            .as_str()
+            .contains("ADD COLUMN hash_version SMALLINT"));
+        assert_eq!(migrations[49].version, 50);
+        assert!(migrations[49]
+            .sql
+            .as_str()
+            .contains("CREATE TABLE operator_listener_pubkeys"));
+        assert!(migrations[49]
+            .sql
+            .as_str()
+            .contains("CREATE TABLE operator_listener_outbox"));
+        assert!(migrations[49]
+            .sql
+            .as_str()
+            .contains("CREATE OR REPLACE FUNCTION community_write_fence_excluded_table"));
+        assert!(migrations[49]
+            .sql
+            .as_str()
+            .contains("'rate_limit_violations', 'operator_listener_outbox'"));
+        for index in [
+            "operator_listener_pubkeys_target",
+            "operator_listener_pubkeys_created_at",
+            "operator_listener_outbox_due",
+            "operator_listener_outbox_recovery",
+            "operator_listener_outbox_created_at",
+        ] {
+            assert!(
+                migrations[49].sql.as_str().contains(index),
+                "migration 0050 must declare {index}"
+            );
+            assert!(
+                desired_schema.contains(index),
+                "schema.sql must declare {index}"
+            );
+        }
+        for index_shape in [
+            "ON operator_listener_outbox (next_attempt_at, created_at, id)",
+            "ON operator_listener_outbox (lease_until, created_at, id)",
+        ] {
+            assert!(
+                migrations[49].sql.as_str().contains(index_shape),
+                "migration 0050 must declare {index_shape}"
+            );
+            assert!(
+                desired_schema.contains(index_shape),
+                "schema.sql must declare {index_shape}"
+            );
+        }
+        // schema.sql keeps the restored (pre-0041) deletion exclusions and also
+        // leaves the deployment-global listener outbox outside tenant fencing.
         assert!(
-            desired_schema.contains("'rate_limit_violations'\n    ]::TEXT[])"),
-            "schema.sql exclusion list must match the pre-0041 body after ledger removal"
+            desired_schema
+                .contains("'rate_limit_violations', 'operator_listener_outbox'\n    ]::TEXT[])"),
+            "schema.sql must exclude the deployment-global listener outbox from tenant fencing"
         );
     }
 

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

@@ -638,6 +638,26 @@ async fn migration_schema_database_guard_covers_legacy_writer_and_nip09_deletion
     assert_eq!(watermark.1, c.id.as_bytes().as_slice());
 }
 
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn migration_schema_operator_listener_outbox_is_not_tenant_scoped() {
+    let db = setup_db().await;
+    let community = CommunityId::from_uuid(make_community(&db.pool).await);
+    let inventory = db
+        .deletion_store()
+        .inventory_schema(community)
+        .await
+        .expect("migrated deletion catalog should validate");
+
+    assert!(
+        !inventory
+            .scoped_tables
+            .iter()
+            .any(|table| table == "operator_listener_outbox"),
+        "operator-listener outbox is deployment-global, not tenant-scoped"
+    );
+}
+
 // ---- Read-replica routing ------------------------------------------------
 //
 // These tests pin the routing contract of `Db::read()` and the two routed

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

@@ -360,7 +360,14 @@ pub async fn insert_event(
         crate::observability::WriterOperation::EventWrite,
     )
     .await?;
-    insert_event_on(&mut connection, community_id, event, channel_id).await
+    if crate::operator_listener::is_listener_mention_kind(u32::from(event.kind.as_u16())) {
+        let mut tx = Transaction::begin(connection, None).await?;
+        let result = insert_event_in_transaction(&mut tx, community_id, event, channel_id).await?;
+        tx.commit().await?;
+        Ok(result)
+    } else {
+        insert_event_on(&mut connection, community_id, event, channel_id).await
+    }
 }
 
 /// Insert a Nostr event in a caller-owned PostgreSQL transaction.
@@ -374,7 +381,11 @@ pub async fn insert_event_in_transaction(
     event: &Event,
     channel_id: Option<Uuid>,
 ) -> Result<(StoredEvent, bool)> {
-    insert_event_on(tx.as_mut(), community_id, event, channel_id).await
+    let result = insert_event_on(tx.as_mut(), community_id, event, channel_id).await?;
+    if result.1 {
+        crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?;
+    }
+    Ok(result)
 }
 
 async fn insert_event_on(
@@ -1528,6 +1539,8 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
                 }
             }
         }
+
+        crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?;
     }
 
     Ok((

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

@@ -26,6 +26,8 @@ pub mod feed;
 pub mod git_repo;
 /// Community moderation: reports, bans/timeouts, audit actions.
 pub mod moderation;
+/// Deployment-global operator-listener mention registrations and delivery queues.
+pub mod operator_listener;
 /// Monthly table partition management.
 pub mod partition;
 /// Buzz product-feedback sidecar persistence.