Make reaction ingest atomic (#1458)

Signed-off-by: npub1t2tgm7d8f995uqvmnm8h88sg3wnpp9a5xysjf6dg3tjmgt3ltulqdp8ehr <5a968df9a7494b4e019b9ecf739e088ba61097b4312124e9a88ae5b42e3f5f3e@sprout-oss.stage.blox.sqprod.co>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
Co-authored-by: npub1t2tgm7d8f995uqvmnm8h88sg3wnpp9a5xysjf6dg3tjmgt3ltulqdp8ehr <5a968df9a7494b4e019b9ecf739e088ba61097b4312124e9a88ae5b42e3f5f3e@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: Tyler Longwell <tlongwell@block.xyz>

crates/buzz-db/src/event.rs

@@ -6,7 +6,7 @@
 
 use chrono::{DateTime, Utc};
 use nostr::Event;
-use sqlx::{PgPool, QueryBuilder, Row};
+use sqlx::{PgPool, Postgres, QueryBuilder, Row, Transaction};
 use uuid::Uuid;
 
 use buzz_core::kind::{
@@ -102,6 +102,22 @@ impl EventQuery {
     }
 }
 
+/// Result of atomically inserting a kind:7 reaction event and its reaction row.
+#[derive(Debug)]
+pub enum ReactionEventInsertOutcome {
+    /// Target event was absent in this community, or was soft-deleted. No writes committed.
+    TargetMissing,
+    /// The active `(target, actor, emoji)` reaction already exists. No event was stored.
+    Duplicate,
+    /// Reaction row and event transaction committed.
+    Inserted {
+        /// Stored reaction event.
+        stored_event: Box<StoredEvent>,
+        /// Whether the event row itself was newly inserted.
+        was_inserted: bool,
+    },
+}
+
 /// Maximum length for a `d_tag` value (bytes). NIP-33 d-tags are short identifiers;
 /// anything beyond this is either a bug or abuse.
 pub const D_TAG_MAX_LEN: usize = 1024;
@@ -982,16 +998,8 @@ pub struct ThreadMetadataParams<'a> {
     pub broadcast: bool,
 }
 
-/// Atomically insert an event AND its thread metadata in a single transaction.
-///
-/// This prevents the race condition where a concurrent delete between separate
-/// `insert_event` and `insert_thread_metadata` calls could leave reply counters
-/// permanently inflated (the metadata insert increments counters for an event
-/// that was already soft-deleted).
-///
-/// Returns `(StoredEvent, was_inserted)`.
-pub async fn insert_event_with_thread_metadata(
-    pool: &PgPool,
+async fn insert_event_with_thread_metadata_tx(
+    tx: &mut Transaction<'_, Postgres>,
     community_id: CommunityId,
     event: &Event,
     channel_id: Option<Uuid>,
@@ -1018,7 +1026,6 @@ pub async fn insert_event_with_thread_metadata(
     let received_at = Utc::now();
     let d_tag = extract_d_tag(event);
     let not_before = extract_not_before(event);
-    let mut tx = pool.begin().await?;
 
     let result = sqlx::query(
         r#"
@@ -1039,7 +1046,7 @@ pub async fn insert_event_with_thread_metadata(
     .bind(channel_id)
     .bind(d_tag.as_deref())
     .bind(not_before)
-    .execute(&mut *tx)
+    .execute(&mut **tx)
     .await?;
 
     let was_inserted = result.rows_affected() > 0;
@@ -1069,7 +1076,7 @@ pub async fn insert_event_with_thread_metadata(
             .bind(meta.root_event_created_at)
             .bind(meta.depth)
             .bind(broadcast_val)
-            .execute(&mut *tx)
+            .execute(&mut **tx)
             .await?;
 
             // Only bump reply counts if the metadata row was actually inserted.
@@ -1096,7 +1103,7 @@ pub async fn insert_event_with_thread_metadata(
                     .bind(parent_ts)
                     .bind(pid)
                     .bind(meta.channel_id)
-                    .execute(&mut *tx)
+                    .execute(&mut **tx)
                     .await?;
 
                     // Ensure the root also has a row (may differ from parent for nested replies).
@@ -1119,7 +1126,7 @@ pub async fn insert_event_with_thread_metadata(
                             .bind(root_ts)
                             .bind(root_id)
                             .bind(meta.channel_id)
-                            .execute(&mut *tx)
+                            .execute(&mut **tx)
                             .await?;
                         }
                     }
@@ -1133,7 +1140,7 @@ pub async fn insert_event_with_thread_metadata(
                     )
                     .bind(community_id.as_uuid())
                     .bind(pid)
-                    .execute(&mut *tx)
+                    .execute(&mut **tx)
                     .await?;
 
                     if let Some(root_id) = meta.root_event_id {
@@ -1146,7 +1153,7 @@ pub async fn insert_event_with_thread_metadata(
                         )
                         .bind(community_id.as_uuid())
                         .bind(root_id)
-                        .execute(&mut *tx)
+                        .execute(&mut **tx)
                         .await?;
                     }
                 }
@@ -1154,14 +1161,102 @@ pub async fn insert_event_with_thread_metadata(
         }
     }
 
-    tx.commit().await?;
-
     Ok((
         StoredEvent::with_received_at(event.clone(), received_at, channel_id, true),
         was_inserted,
     ))
 }
 
+/// Atomically insert an event and its optional thread metadata.
+///
+/// `insert_event` and `insert_thread_metadata` calls could leave reply counters
+/// inconsistent if one succeeded and the other failed. Keep this as one
+/// transaction so reply metadata and counters commit together with the event.
+///
+/// Returns `(StoredEvent, was_inserted)`.
+pub async fn insert_event_with_thread_metadata(
+    pool: &PgPool,
+    community_id: CommunityId,
+    event: &Event,
+    channel_id: Option<Uuid>,
+    thread_meta: Option<ThreadMetadataParams<'_>>,
+) -> Result<(StoredEvent, bool)> {
+    let mut tx = pool.begin().await?;
+    let result =
+        insert_event_with_thread_metadata_tx(&mut tx, community_id, event, channel_id, thread_meta)
+            .await?;
+    tx.commit().await?;
+    Ok(result)
+}
+
+/// Atomically insert a kind:7 reaction event and its reaction row.
+///
+/// Ordering is load-bearing: resolve target, upsert/reactivate the reaction row,
+/// check `rows_affected`, then insert the kind:7 event. Active duplicates return
+/// before event insertion so duplicate reactions never store a duplicate kind:7.
+#[allow(clippy::too_many_arguments)]
+pub async fn insert_reaction_event_with_thread_metadata(
+    pool: &PgPool,
+    community_id: CommunityId,
+    reaction_event: &Event,
+    channel_id: Option<Uuid>,
+    thread_meta: Option<ThreadMetadataParams<'_>>,
+    target_event_id: &[u8],
+    actor_pubkey: &[u8],
+    emoji: &str,
+) -> Result<ReactionEventInsertOutcome> {
+    let mut tx = pool.begin().await?;
+
+    let target_row = sqlx::query(
+        "SELECT created_at FROM events \
+         WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL \
+         ORDER BY created_at DESC LIMIT 1",
+    )
+    .bind(community_id.as_uuid())
+    .bind(target_event_id)
+    .fetch_optional(&mut *tx)
+    .await?;
+
+    let Some(target_row) = target_row else {
+        tx.rollback().await?;
+        return Ok(ReactionEventInsertOutcome::TargetMissing);
+    };
+    let target_created_at: DateTime<Utc> = target_row.get("created_at");
+
+    // Preserve add_reaction's exact new / re-activate / active-duplicate semantics.
+    let reaction_inserted = crate::reaction::add_reaction_tx(
+        &mut tx,
+        community_id,
+        target_event_id,
+        target_created_at,
+        actor_pubkey,
+        emoji,
+        Some(reaction_event.id.as_bytes()),
+    )
+    .await?;
+
+    if !reaction_inserted {
+        tx.rollback().await?;
+        return Ok(ReactionEventInsertOutcome::Duplicate);
+    }
+
+    let (stored_event, was_inserted) = insert_event_with_thread_metadata_tx(
+        &mut tx,
+        community_id,
+        reaction_event,
+        channel_id,
+        thread_meta,
+    )
+    .await?;
+
+    tx.commit().await?;
+
+    Ok(ReactionEventInsertOutcome::Inserted {
+        stored_event: Box::new(stored_event),
+        was_inserted,
+    })
+}
+
 /// A due reminder row returned by [`query_due_reminders`].
 #[derive(Debug)]
 pub struct DueReminder {
@@ -1410,6 +1505,269 @@ mod tests {
             .expect("sign")
     }
 
+    fn make_text_event(content: &str) -> nostr::Event {
+        let keys = Keys::generate();
+        EventBuilder::new(Kind::Custom(9), content)
+            .sign_with_keys(&keys)
+            .expect("sign text event")
+    }
+
+    fn make_reaction_event(keys: &Keys, target_id_hex: &str, emoji: &str) -> nostr::Event {
+        let nonce = Uuid::new_v4().to_string();
+        EventBuilder::new(Kind::Custom(7), emoji)
+            .tags(vec![
+                Tag::parse(["e", target_id_hex]).expect("reaction e tag"),
+                Tag::parse(["nonce", nonce.as_str()]).expect("nonce tag"),
+            ])
+            .sign_with_keys(keys)
+            .expect("sign reaction event")
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn reaction_single_tx_duplicate_short_circuit_stores_no_event() {
+        let pool = setup_pool().await;
+        let community = CommunityId::from_uuid(make_test_community(&pool).await);
+        let target = make_text_event("reaction target");
+        insert_event(&pool, community, &target, None)
+            .await
+            .expect("insert target");
+
+        let actor = Keys::generate();
+        let actor_pubkey = actor.public_key().to_bytes();
+        let target_hex = target.id.to_hex();
+        let first = make_reaction_event(&actor, &target_hex, "👍");
+        let second = make_reaction_event(&actor, &target_hex, "👍");
+
+        let first_outcome = insert_reaction_event_with_thread_metadata(
+            &pool,
+            community,
+            &first,
+            None,
+            None,
+            target.id.as_bytes(),
+            &actor_pubkey,
+            "👍",
+        )
+        .await
+        .expect("first reaction insert");
+        assert!(matches!(
+            first_outcome,
+            ReactionEventInsertOutcome::Inserted {
+                was_inserted: true,
+                ..
+            }
+        ));
+
+        let duplicate = insert_reaction_event_with_thread_metadata(
+            &pool,
+            community,
+            &second,
+            None,
+            None,
+            target.id.as_bytes(),
+            &actor_pubkey,
+            "👍",
+        )
+        .await
+        .expect("duplicate reaction insert");
+        assert!(matches!(duplicate, ReactionEventInsertOutcome::Duplicate));
+
+        let duplicate_event = get_event_by_id(&pool, community, second.id.as_bytes())
+            .await
+            .expect("lookup duplicate reaction event");
+        assert!(
+            duplicate_event.is_none(),
+            "active duplicate reaction must short-circuit before storing kind:7 event"
+        );
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn reaction_single_tx_cross_community_target_rejected() {
+        let pool = setup_pool().await;
+        let community_a = CommunityId::from_uuid(make_test_community(&pool).await);
+        let community_b = CommunityId::from_uuid(make_test_community(&pool).await);
+        let target = make_text_event("community A target only");
+        insert_event(&pool, community_a, &target, None)
+            .await
+            .expect("insert target in A");
+
+        let actor = Keys::generate();
+        let actor_pubkey = actor.public_key().to_bytes();
+        let reaction = make_reaction_event(&actor, &target.id.to_hex(), "👍");
+
+        let outcome = insert_reaction_event_with_thread_metadata(
+            &pool,
+            community_b,
+            &reaction,
+            None,
+            None,
+            target.id.as_bytes(),
+            &actor_pubkey,
+            "👍",
+        )
+        .await
+        .expect("cross-community reaction attempt");
+        assert!(matches!(outcome, ReactionEventInsertOutcome::TargetMissing));
+
+        assert!(
+            get_event_by_id(&pool, community_b, reaction.id.as_bytes())
+                .await
+                .expect("lookup B reaction event")
+                .is_none(),
+            "reaction event must not store when target exists only in another community"
+        );
+        assert!(
+            crate::reaction::get_active_reaction_record(
+                &pool,
+                community_b,
+                target.id.as_bytes(),
+                DateTime::from_timestamp(target.created_at.as_secs() as i64, 0).unwrap(),
+                &actor_pubkey,
+                "👍",
+            )
+            .await
+            .expect("lookup B reaction row")
+            .is_none(),
+            "reaction row must not be inserted for cross-community target miss"
+        );
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn reaction_single_tx_event_insert_failure_rolls_back_reaction() {
+        let pool = setup_pool().await;
+        let community = CommunityId::from_uuid(make_test_community(&pool).await);
+        let target = make_text_event("rollback target");
+        insert_event(&pool, community, &target, None)
+            .await
+            .expect("insert target");
+
+        let actor = Keys::generate();
+        let actor_pubkey = actor.public_key().to_bytes();
+        let target_hex = target.id.to_hex();
+        let bad_reaction = EventBuilder::new(Kind::Custom(20000), "👍")
+            .tags(vec![
+                Tag::parse(["e", target_hex.as_str()]).expect("reaction e tag")
+            ])
+            .sign_with_keys(&actor)
+            .expect("sign ephemeral reaction-shaped event");
+        let target_created_at = DateTime::from_timestamp(target.created_at.as_secs() as i64, 0)
+            .expect("target timestamp");
+
+        let err = insert_reaction_event_with_thread_metadata(
+            &pool,
+            community,
+            &bad_reaction,
+            None,
+            None,
+            target.id.as_bytes(),
+            &actor_pubkey,
+            "👍",
+        )
+        .await
+        .expect_err("ephemeral event insert must fail after reaction upsert attempt");
+        assert!(matches!(err, DbError::EphemeralEventRejected(20000)));
+
+        assert!(
+            crate::reaction::get_active_reaction_record(
+                &pool,
+                community,
+                target.id.as_bytes(),
+                target_created_at,
+                &actor_pubkey,
+                "👍",
+            )
+            .await
+            .expect("lookup reaction row after rollback")
+            .is_none(),
+            "transaction rollback must remove the reaction row when event insert fails"
+        );
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn reaction_single_tx_reactivates_soft_deleted_reaction() {
+        let pool = setup_pool().await;
+        let community = CommunityId::from_uuid(make_test_community(&pool).await);
+        let target = make_text_event("reactivation target");
+        insert_event(&pool, community, &target, None)
+            .await
+            .expect("insert target");
+
+        let actor = Keys::generate();
+        let actor_pubkey = actor.public_key().to_bytes();
+        let target_hex = target.id.to_hex();
+        let target_created_at = DateTime::from_timestamp(target.created_at.as_secs() as i64, 0)
+            .expect("target timestamp");
+        let first = make_reaction_event(&actor, &target_hex, "👍");
+        let second = make_reaction_event(&actor, &target_hex, "👍");
+
+        assert!(matches!(
+            insert_reaction_event_with_thread_metadata(
+                &pool,
+                community,
+                &first,
+                None,
+                None,
+                target.id.as_bytes(),
+                &actor_pubkey,
+                "👍",
+            )
+            .await
+            .expect("first reaction insert"),
+            ReactionEventInsertOutcome::Inserted { .. }
+        ));
+        assert!(crate::reaction::remove_reaction(
+            &pool,
+            community,
+            target.id.as_bytes(),
+            target_created_at,
+            &actor_pubkey,
+            "👍",
+        )
+        .await
+        .expect("soft delete reaction"));
+
+        let outcome = insert_reaction_event_with_thread_metadata(
+            &pool,
+            community,
+            &second,
+            None,
+            None,
+            target.id.as_bytes(),
+            &actor_pubkey,
+            "👍",
+        )
+        .await
+        .expect("reactivate reaction");
+        assert!(matches!(
+            outcome,
+            ReactionEventInsertOutcome::Inserted {
+                was_inserted: true,
+                ..
+            }
+        ));
+
+        let active = crate::reaction::get_active_reaction_record(
+            &pool,
+            community,
+            target.id.as_bytes(),
+            target_created_at,
+            &actor_pubkey,
+            "👍",
+        )
+        .await
+        .expect("active record after reactivation")
+        .expect("reaction active after reactivation");
+        assert_eq!(
+            active.reaction_event_id.as_deref(),
+            Some(second.id.as_bytes().as_slice()),
+            "reactivation through the tx path must preserve add_reaction's source-id update semantics"
+        );
+    }
+
     #[test]
     fn extract_d_tag_from_nip33_event() {
         let event = make_event_with_kind_and_tags(

crates/buzz-db/src/lib.rs

@@ -41,7 +41,7 @@ pub mod user;
 pub mod workflow;
 
 pub use error::{DbError, Result};
-pub use event::EventQuery;
+pub use event::{EventQuery, ReactionEventInsertOutcome};
 
 use chrono::{DateTime, Utc};
 use sqlx::postgres::PgPoolOptions;
@@ -590,6 +590,40 @@ impl Db {
         Ok(result)
     }
 
+    /// Atomically insert a kind:7 reaction event and its reaction row.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn insert_reaction_event_with_thread_metadata(
+        &self,
+        community_id: CommunityId,
+        event: &nostr::Event,
+        channel_id: Option<Uuid>,
+        thread_meta: Option<event::ThreadMetadataParams<'_>>,
+        target_event_id: &[u8],
+        actor_pubkey: &[u8],
+        emoji: &str,
+    ) -> Result<event::ReactionEventInsertOutcome> {
+        let outcome = event::insert_reaction_event_with_thread_metadata(
+            &self.pool,
+            community_id,
+            event,
+            channel_id,
+            thread_meta,
+            target_event_id,
+            actor_pubkey,
+            emoji,
+        )
+        .await?;
+        if let event::ReactionEventInsertOutcome::Inserted {
+            was_inserted: true, ..
+        } = &outcome
+        {
+            if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await {
+                tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
+            }
+        }
+        Ok(outcome)
+    }
+
     /// Creates a new channel, bootstraps the creator as owner, and returns the record.
     #[allow(clippy::too_many_arguments)]
     pub async fn create_channel(

crates/buzz-db/src/reaction.rs

@@ -3,7 +3,7 @@
 //! One reaction per user per emoji per event. Soft-delete via removed_at.
 
 use chrono::{DateTime, Utc};
-use sqlx::{PgPool, Row};
+use sqlx::{PgPool, Postgres, Row, Transaction};
 
 use crate::error::Result;
 use crate::CommunityId;
@@ -62,6 +62,16 @@ pub struct ActiveReactionRecord {
 
 // -- Write operations ---------------------------------------------------------
 
+const ADD_REACTION_SQL: &str = r#"
+        INSERT INTO reactions (community_id, event_created_at, event_id, pubkey, emoji, reaction_event_id)
+        VALUES ($1, $2, $3, $4, $5, $6)
+        ON CONFLICT (community_id, event_created_at, event_id, pubkey, emoji) DO UPDATE SET
+            created_at = NOW(),
+            removed_at = NULL,
+            reaction_event_id = COALESCE(EXCLUDED.reaction_event_id, reactions.reaction_event_id)
+        WHERE reactions.removed_at IS NOT NULL
+        "#;
+
 /// Add (or re-activate) a reaction.
 ///
 /// Returns `Ok(true)` if the reaction was added or re-activated, `Ok(false)` if
@@ -78,25 +88,15 @@ pub async fn add_reaction(
     emoji: &str,
     reaction_event_id: Option<&[u8]>,
 ) -> Result<bool> {
-    let result = sqlx::query(
-        r#"
-        INSERT INTO reactions (community_id, event_created_at, event_id, pubkey, emoji, reaction_event_id)
-        VALUES ($1, $2, $3, $4, $5, $6)
-        ON CONFLICT (community_id, event_created_at, event_id, pubkey, emoji) DO UPDATE SET
-            created_at = NOW(),
-            removed_at = NULL,
-            reaction_event_id = COALESCE(EXCLUDED.reaction_event_id, reactions.reaction_event_id)
-        WHERE reactions.removed_at IS NOT NULL
-        "#,
-    )
-    .bind(community.as_uuid())
-    .bind(event_created_at)
-    .bind(event_id)
-    .bind(pubkey)
-    .bind(emoji)
-    .bind(reaction_event_id)
-    .execute(pool)
-    .await?;
+    let result = sqlx::query(ADD_REACTION_SQL)
+        .bind(community.as_uuid())
+        .bind(event_created_at)
+        .bind(event_id)
+        .bind(pubkey)
+        .bind(emoji)
+        .bind(reaction_event_id)
+        .execute(pool)
+        .await?;
 
     // Three cases:
     // (a) New reaction (no existing row): INSERT succeeds → rows_affected = 1 → true.
@@ -107,6 +107,33 @@ pub async fn add_reaction(
     Ok(result.rows_affected() != 0)
 }
 
+/// Add (or re-activate) a reaction inside an existing transaction.
+///
+/// Uses the same `INSERT ... ON CONFLICT DO UPDATE ... WHERE removed_at IS NOT NULL`
+/// statement as [`add_reaction`], preserving the new / re-activate / active-duplicate
+/// semantics while letting callers atomically couple the reaction row to other writes.
+pub(crate) async fn add_reaction_tx(
+    tx: &mut Transaction<'_, Postgres>,
+    community: CommunityId,
+    event_id: &[u8],
+    event_created_at: DateTime<Utc>,
+    pubkey: &[u8],
+    emoji: &str,
+    reaction_event_id: Option<&[u8]>,
+) -> Result<bool> {
+    let result = sqlx::query(ADD_REACTION_SQL)
+        .bind(community.as_uuid())
+        .bind(event_created_at)
+        .bind(event_id)
+        .bind(pubkey)
+        .bind(emoji)
+        .bind(reaction_event_id)
+        .execute(&mut **tx)
+        .await?;
+
+    Ok(result.rows_affected() != 0)
+}
+
 /// Soft-delete a reaction by setting `removed_at`.
 ///
 /// Returns `true` if a row was updated, `false` if not found or already removed.

crates/buzz-relay/src/handlers/ingest.rs

@@ -1833,10 +1833,9 @@ async fn ingest_event_inner(
         ));
     }
 
-    // Resolve target event, insert the reaction row (dedup via ON CONFLICT),
-    // store the event, then backfill the reaction_event_id. If the event insert
-    // fails, compensate by removing the reaction row so state stays consistent.
-    // This replaces the post-storage side-effect handler for kind:7.
+    // Resolve the target reference, then use one DB transaction to upsert the
+    // reaction row (dedup via ON CONFLICT) with reaction_event_id already set and
+    // store the kind:7 event. This replaces the post-storage side-effect handler.
     if kind_u32 == KIND_REACTION {
         // Extract target event hex from last e-tag (NIP-25).
         let target_hex = event
@@ -1865,19 +1864,6 @@ async fn ingest_event_inner(
         let target_id = hex::decode(&target_hex)
             .map_err(|_| IngestError::Rejected("invalid: malformed reaction target id".into()))?;
 
-        let target_event = state
-            .db
-            .get_event_by_id(tenant.community(), &target_id)
-            .await
-            .map_err(|e| IngestError::Internal(format!("error: {e}")))?
-            .ok_or_else(|| {
-                IngestError::Rejected("invalid: reaction target event not found".into())
-            })?;
-
-        let target_created_at =
-            chrono::DateTime::from_timestamp(target_event.event.created_at.as_secs() as i64, 0)
-                .unwrap_or_else(chrono::Utc::now);
-
         let actor_bytes = effective_message_author(&event, &state.relay_keypair.public_key());
         let emoji = if event.content.is_empty() {
             "+"
@@ -1897,79 +1883,42 @@ async fn ingest_event_inner(
             )));
         }
 
-        // add_reaction returns false if the (target, actor, emoji) tuple already
-        // exists — short-circuit without storing the event.
-        let inserted = state
-            .db
-            .add_reaction(
-                tenant.community(),
-                &target_id,
-                target_created_at,
-                &actor_bytes,
-                emoji,
-                None,
-            )
-            .await
-            .map_err(|e| IngestError::Internal(format!("error: {e}")))?;
-
-        if !inserted {
-            return Ok(IngestResult {
-                event_id: event_id_hex,
-                accepted: false,
-                message: "duplicate: reaction already exists".into(),
-            });
-        }
-
-        // Store the event; on failure compensate by removing the reaction row.
+        // Atomically upsert the reaction row with this kind:7 event id, then store
+        // the event in the same transaction. Ordering is load-bearing: active
+        // duplicate reactions must return before storing a duplicate kind:7 event.
         let thread_params = thread_meta.as_ref().map(|m| m.as_params());
         let (stored_event, was_inserted) = match state
             .db
-            .insert_event_with_thread_metadata(
+            .insert_reaction_event_with_thread_metadata(
                 tenant.community(),
                 &event,
                 channel_id,
                 thread_params,
+                &target_id,
+                &actor_bytes,
+                emoji,
             )
             .await
+            .map_err(|e| IngestError::Internal(format!("error: {e}")))?
         {
-            Ok(result) => result,
-            Err(e) => {
-                // Compensate: undo the reaction row so state stays consistent.
-                if let Err(re) = state
-                    .db
-                    .remove_reaction(
-                        tenant.community(),
-                        &target_id,
-                        target_created_at,
-                        &actor_bytes,
-                        emoji,
-                    )
-                    .await
-                {
-                    warn!(event_id = %event_id_hex, "reaction compensation failed: {re}");
-                }
-                return Err(IngestError::Internal(format!("error: database error: {e}")));
+            buzz_db::ReactionEventInsertOutcome::TargetMissing => {
+                return Err(IngestError::Rejected(
+                    "invalid: reaction target event not found".into(),
+                ));
             }
+            buzz_db::ReactionEventInsertOutcome::Duplicate => {
+                return Ok(IngestResult {
+                    event_id: event_id_hex,
+                    accepted: false,
+                    message: "duplicate: reaction already exists".into(),
+                });
+            }
+            buzz_db::ReactionEventInsertOutcome::Inserted {
+                stored_event,
+                was_inserted,
+            } => (stored_event, was_inserted),
         };
 
-        if was_inserted {
-            // Backfill the reaction_event_id so the row is fully linked.
-            if let Err(e) = state
-                .db
-                .set_reaction_event_id(
-                    tenant.community(),
-                    &target_id,
-                    target_created_at,
-                    &actor_bytes,
-                    emoji,
-                    event.id.as_bytes(),
-                )
-                .await
-            {
-                warn!(event_id = %event_id_hex, "set_reaction_event_id failed: {e}");
-            }
-        }
-
         let pubkey_hex = auth.pubkey().to_hex();
         // Spec WriteInsert (line 514) / WriteDuplicate (line 606): emit
         // the abstract write action. The persist API returns