Centralize replaceable event persistence

.github/workflows/ci.yml

@@ -674,6 +674,19 @@ jobs:
           VALUES ('00000000-0000-4000-8000-00000000c0de', 'localhost:3000')
           ON CONFLICT (lower(host)) DO NOTHING
           ;"
+      - name: Replaceable persistence PostgreSQL tests
+        # Transaction, concurrency, and mention-index coverage for the
+        # replaceable-event store seam. These tests require real Postgres and
+        # are ignored by the infrastructure-free unit-test job.
+        run: |
+          filter='package(buzz-db) and test(/tests::(parameterized_|concurrent_parameterized_)/)'
+          cargo nextest run \
+            --archive-file target/ci/backend-integration-tests.tar.zst \
+            -E "${filter}" \
+            --run-ignored ignored-only
+        env:
+          DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz
+          TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz
       - name: Start relay
         run: |
           chmod +x ./target/ci/buzz-relay

crates/buzz-db/src/channel.rs

@@ -625,7 +625,7 @@ pub async fn lock_member_snapshot(
     // this key before INSERT; migration 0032 then takes the membership key in
     // the INSERT trigger. Taking both in that order avoids mixed-version
     // duplicate heads without introducing a lock-order inversion.
-    let replacement_lock = crate::event_replacement_lock_key(
+    let replacement_lock = crate::replaceable::event_replacement_lock_key(
         community_id,
         39002,
         relay_pubkey,

crates/buzz-db/src/event.rs

@@ -6,7 +6,7 @@
 
 use chrono::{DateTime, Utc};
 use nostr::Event;
-use sqlx::{PgPool, Postgres, QueryBuilder, Row, Transaction};
+use sqlx::{PgConnection, PgPool, Postgres, QueryBuilder, Row, Transaction};
 use uuid::Uuid;
 
 use buzz_core::kind::{
@@ -275,6 +275,30 @@ pub async fn insert_event(
     community_id: CommunityId,
     event: &Event,
     channel_id: Option<Uuid>,
+) -> Result<(StoredEvent, bool)> {
+    let mut connection = pool.acquire().await?;
+    insert_event_on(&mut connection, community_id, event, channel_id).await
+}
+
+/// Insert a Nostr event in a caller-owned PostgreSQL transaction.
+///
+/// This is the transaction-composition seam for callers that must keep the
+/// event insert open while performing related work. The caller owns commit or
+/// rollback.
+pub async fn insert_event_in_transaction(
+    tx: &mut Transaction<'_, Postgres>,
+    community_id: CommunityId,
+    event: &Event,
+    channel_id: Option<Uuid>,
+) -> Result<(StoredEvent, bool)> {
+    insert_event_on(tx.as_mut(), community_id, event, channel_id).await
+}
+
+async fn insert_event_on(
+    connection: &mut PgConnection,
+    community_id: CommunityId,
+    event: &Event,
+    channel_id: Option<Uuid>,
 ) -> Result<(StoredEvent, bool)> {
     let kind_u16 = event.kind.as_u16();
     let kind_u32 = u32::from(kind_u16);
@@ -317,7 +341,7 @@ pub async fn insert_event(
     .bind(channel_id)
     .bind(d_tag.as_deref())
     .bind(not_before)
-    .execute(pool)
+    .execute(connection)
     .await?;
 
     let was_inserted = result.rows_affected() > 0;
@@ -1607,6 +1631,31 @@ mod tests {
         id
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn event_insert_in_existing_transaction_rolls_back_with_caller() {
+        let pool = setup_pool().await;
+        let community_uuid = make_test_community(&pool).await;
+        let community = CommunityId::from_uuid(community_uuid);
+        let event = make_text_event("caller-owned transaction");
+
+        let mut tx = pool.begin().await.expect("begin event insert transaction");
+        let (_, was_inserted) = insert_event_in_transaction(&mut tx, community, &event, None)
+            .await
+            .expect("insert event in caller transaction");
+        assert!(was_inserted);
+        tx.rollback().await.expect("roll back event insert");
+
+        let persisted: i64 =
+            sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id = $1 AND id = $2")
+                .bind(community_uuid)
+                .bind(event.id.as_bytes().as_slice())
+                .fetch_one(&pool)
+                .await
+                .expect("count rolled-back event");
+        assert_eq!(persisted, 0);
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn event_insert_ttl_trigger_handles_permanent_ephemeral_duplicate_and_activation_race() {