diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a87ea957cd..8016766812 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -679,6 +679,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 diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 109a9367d7..a1890adb56 100644 --- a/crates/buzz-db/src/channel.rs +++ b/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, diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 5d682d7843..dcfc97c844 100644 --- a/crates/buzz-db/src/event.rs +++ b/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, +) -> 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, +) -> 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, ) -> 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() { diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 3ff230f950..cbce6d3753 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -8,6 +8,21 @@ //! - Events table is partitioned by month on `created_at`. //! - No FK references to partitioned tables. //! - Uses `sqlx::query()` (runtime) not `sqlx::query!()` (compile-time). +//! +//! ## Runtime and store ownership +//! This crate intentionally keeps database runtime and Buzz domain persistence +//! together while maintaining an internal boundary between them: +//! +//! - Runtime concerns own pool construction, writer/replica routing, transaction +//! creation, session invariants, metrics, and health support. +//! - Store concerns own domain-specific SQL, row mapping, locking, mutation +//! rules, indexes, and focused persistence tests. +//! +//! Transaction-required store operations accept [`sqlx::Transaction`] so their +//! composition requirement is visible in the type. Private connection helpers +//! are reserved for SQL primitives that are valid on any same-session +//! connection. New domains should prove this boundary incrementally instead of +//! exposing raw pools or introducing broad store traits. /// Explicit deployment-global admin report reads. pub mod admin_moderation; @@ -45,6 +60,8 @@ pub mod reaction; pub mod relay_invite; /// Relay-level membership persistence (NIP-43). pub mod relay_members; +/// Replaceable-event persistence and coordinate locking. +pub mod replaceable; /// Replica freshness fence for keyset-cursor read routing. pub mod replica_fence; /// Thread metadata persistence. @@ -68,37 +85,12 @@ use uuid::Uuid; use buzz_core::{CommunityId, StoredEvent}; -pub(crate) fn event_replacement_lock_key( - community_id: CommunityId, - kind: i32, - pubkey: &[u8], - coordinate: Option<&[u8]>, -) -> i64 { - let mut hash: u64 = 0xcbf29ce484222325; - let kind_bytes = kind.to_le_bytes(); - for bytes in [ - community_id.as_uuid().as_bytes().as_slice(), - kind_bytes.as_slice(), - pubkey, - ] { - for byte in bytes { - hash ^= *byte as u64; - hash = hash.wrapping_mul(0x100000001b3); - } - } - if let Some(coordinate) = coordinate { - for byte in coordinate { - hash ^= *byte as u64; - hash = hash.wrapping_mul(0x100000001b3); - } - } - hash as i64 -} - /// Extract p-tag mentions from an event and insert into the `event_mentions` table. /// -/// Called after event insertion. Failures are logged but do not block event storage. -/// Uses `INSERT ... ON CONFLICT DO NOTHING` so duplicate inserts are silently skipped. +/// This pool-owning wrapper propagates failures to its caller. Replacement writes +/// use the transaction-bound helper below so event storage and mention indexing +/// commit or roll back together. Duplicate inserts are silently skipped with +/// `INSERT ... ON CONFLICT DO NOTHING`. pub async fn insert_mentions( pool: &PgPool, community_id: CommunityId, @@ -4875,7 +4867,7 @@ impl Db { .ok_or(DbError::InvalidTimestamp(created_at_secs))?; // Collisions only cause extra serialization; they cannot change behavior. - let lock_key = event_replacement_lock_key( + let lock_key = replaceable::event_replacement_lock_key( community_id, kind_i32, pubkey_bytes.as_slice(), @@ -5058,8 +5050,12 @@ impl Db { let kind_i32 = buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST as i32; let pubkey_bytes = relay_keypair.public_key().to_bytes(); - let lock_key = - event_replacement_lock_key(community_id, kind_i32, pubkey_bytes.as_slice(), None); + let lock_key = replaceable::event_replacement_lock_key( + community_id, + kind_i32, + pubkey_bytes.as_slice(), + None, + ); let mut tx = self.pool.begin().await?; @@ -5166,243 +5162,6 @@ impl Db { member_count, )) } - - /// Atomically replace a NIP-33 parameterized replaceable event (kind 30000–39999). - /// - /// Keeps only the event with the highest `created_at` per `(kind, pubkey, d_tag)`. - /// Same-second ties are broken by lowest event `id` (deterministic ordering). - /// The entire check → retire old payload → insert runs in a single transaction - /// with an advisory lock to prevent concurrent-insert races. NIP-RS read-state - /// coordinates hard-delete the superseded payload and preserve a compact - /// ordering watermark. Buzz mesh status coordinates also hard-delete their - /// superseded heartbeat payload because only the live head has product - /// value; other NIP-33 kinds retain soft-deleted history. - /// - /// **Channel policy:** NIP-33 replacement keys on `(kind, pubkey, d_tag)` globally — - /// `channel_id` is NOT part of the replacement key. This matches the Nostr spec: - /// an author's parameterized replaceable event is a single global resource identified - /// by its d-tag, regardless of which channel it was submitted to. The `channel_id` - /// parameter is stored on the new row for query scoping but does not affect replacement. - /// - /// Note: `replace_addressable_event()` keys on `channel_id` because it serves - /// relay-signed NIP-29 group metadata (kind 39000–39002) where the relay is the - /// author and channel_id distinguishes groups. User-submitted NIP-33 events use - /// this function instead, where the author's pubkey + d-tag is the natural key. - #[datastore_span(name = "replace_parameterized_event", system = "postgresql")] - pub async fn replace_parameterized_event( - &self, - community_id: CommunityId, - event: &nostr::Event, - d_tag: &str, - channel_id: Option, - ) -> Result<(StoredEvent, bool)> { - let kind_i32 = buzz_core::kind::event_kind_i32(event); - let pubkey_bytes = event.pubkey.to_bytes(); - let created_at_secs = event.created_at.as_secs() as i64; - let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0) - .ok_or(DbError::InvalidTimestamp(created_at_secs))?; - - let lock_key = event_replacement_lock_key( - community_id, - kind_i32, - pubkey_bytes.as_slice(), - Some(d_tag.as_bytes()), - ); - - let mut tx = self.pool.begin().await?; - - sqlx::query("SELECT pg_advisory_xact_lock($1)") - .bind(lock_key) - .execute(&mut *tx) - .await?; - - let d_tag_count = event - .tags - .iter() - .filter(|tag| tag.as_slice().first().is_some_and(|part| part == "d")) - .count(); - let has_exact_d_tag = event.tags.iter().any(|tag| { - let parts = tag.as_slice(); - parts.len() >= 2 && parts[0] == "d" && parts[1] == d_tag - }); - let read_state_t_tag_count = event - .tags - .iter() - .filter(|tag| { - let parts = tag.as_slice(); - parts.len() == 2 && parts[0] == "t" && parts[1] == "read-state" - }) - .count(); - let is_nip_rs = kind_i32 == buzz_core::kind::KIND_READ_STATE as i32 - && d_tag_count == 1 - && has_exact_d_tag - && d_tag.strip_prefix("read-state:").is_some_and(|slot| { - slot.len() == 32 - && slot - .bytes() - .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) - }) - && read_state_t_tag_count == 1; - let is_buzz_mesh_status = kind_i32 == buzz_core::kind::KIND_BOOKMARK_SET as i32 - && d_tag.starts_with("buzz-mesh-member-status:") - && event.tags.iter().any(|tag| { - let parts = tag.as_slice(); - parts.len() == 2 && parts[0] == "k" && parts[1] == "buzz-mesh-status" - }); - let hard_delete_superseded = is_nip_rs || is_buzz_mesh_status; - - // Check the live head and, for NIP-RS, the compact historical ordering - // watermark. The watermark remains after a NIP-09 coordinate deletion, - // preventing a previously accepted signed blob from being resurrected. - let existing: Option<(chrono::DateTime, Vec)> = sqlx::query_as( - "SELECT created_at, id FROM events \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \ - ORDER BY created_at DESC, id ASC LIMIT 1", - ) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(d_tag) - .fetch_optional(&mut *tx) - .await?; - let watermark: Option<(chrono::DateTime, Vec)> = if is_nip_rs { - sqlx::query_as( - "SELECT created_at, event_id FROM parameterized_event_watermarks \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4", - ) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(d_tag) - .fetch_optional(&mut *tx) - .await? - } else { - None - }; - - // Stale-write protection: reject if either durable ordering source - // dominates the incoming tuple. Equal timestamps use lowest event id. - let incoming_id = event.id.as_bytes().as_slice(); - let dominated = - existing - .iter() - .chain(watermark.iter()) - .any(|(accepted_ts, accepted_id)| { - created_at < *accepted_ts - || (created_at == *accepted_ts && incoming_id >= accepted_id.as_slice()) - }); - if dominated { - tx.rollback().await?; - let received_at = chrono::Utc::now(); - return Ok(( - StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), - false, - )); - } - - if existing.is_some() { - if is_nip_rs { - // Migration 0011 rejects regex-coordinate hard deletes from - // pre-fix writers. Authorize only this corrected NIP-RS delete, - // transaction-locally so pooled connections cannot leak it. - sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)") - .execute(&mut *tx) - .await?; - } - let statement = if hard_delete_superseded { - "DELETE FROM events \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL" - } else { - "UPDATE events SET deleted_at = NOW() \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL" - }; - sqlx::query(statement) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(d_tag) - .execute(&mut *tx) - .await?; - - if hard_delete_superseded { - if let Some((_, existing_id)) = &existing { - // Event first, mentions second: migration 0009's live-event - // fence uses this global lock order to avoid deadlocks. - sqlx::query( - "DELETE FROM event_mentions WHERE community_id = $1 AND event_id = $2", - ) - .bind(community_id.as_uuid()) - .bind(existing_id) - .execute(&mut *tx) - .await?; - } - } - } - - // Insert the new event inside the transaction. - let sig_bytes = event.sig.serialize(); - let tags_json = serde_json::to_value(&event.tags)?; - let received_at = chrono::Utc::now(); - - let insert_result = sqlx::query( - "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \ - ON CONFLICT DO NOTHING", - ) - .bind(community_id.as_uuid()) - .bind(event.id.as_bytes().as_slice()) - .bind(pubkey_bytes.as_slice()) - .bind(created_at) - .bind(kind_i32) - .bind(&tags_json) - .bind(&event.content) - .bind(sig_bytes.as_slice()) - .bind(received_at) - .bind(channel_id) - .bind(d_tag) - .bind(event::extract_not_before(event)) - .execute(&mut *tx) - .await?; - - let was_inserted = insert_result.rows_affected() > 0; - if !was_inserted { - tx.rollback().await?; - return Ok(( - StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), - false, - )); - } - - if is_nip_rs { - sqlx::query( - "INSERT INTO parameterized_event_watermarks \ - (community_id, kind, pubkey, d_tag, created_at, event_id) \ - VALUES ($1, $2, $3, $4, $5, $6) \ - ON CONFLICT (community_id, kind, pubkey, d_tag) DO UPDATE SET \ - created_at = EXCLUDED.created_at, event_id = EXCLUDED.event_id", - ) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(d_tag) - .bind(created_at) - .bind(incoming_id) - .execute(&mut *tx) - .await?; - } - - tx.commit().await?; - - // Mentions are a denormalized index — safe outside the transaction. - if let Err(e) = crate::insert_mentions(&self.pool, community_id, event, channel_id).await { - tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); - } - - Ok(( - StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), - true, - )) - } } /// A full API token record. @@ -6014,6 +5773,456 @@ mod tests { assert_eq!(live, 0, "watermark must block stale resurrection"); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn nip_rs_transaction_operation_restores_hard_delete_opt_in() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let base = Timestamp::now().as_secs(); + let replace_d_tag = format!("read-state:{}", "b".repeat(32)); + let victim_d_tag = format!("read-state:{}", "c".repeat(32)); + let event = |d_tag: &str, content: &str, timestamp: u64| { + EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), + content, + ) + .tags(vec![ + Tag::parse(["d", d_tag]).expect("d tag"), + Tag::parse(["t", "read-state"]).expect("t tag"), + ]) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&keys) + .expect("sign read state") + }; + let old = event(&replace_d_tag, "old", base); + let new = event(&replace_d_tag, "new", base + 1); + let victim = event(&victim_d_tag, "victim", base); + + assert!( + db.replace_parameterized_event(community, &old, &replace_d_tag, None) + .await + .expect("insert old head") + .1 + ); + assert!( + db.replace_parameterized_event(community, &victim, &victim_d_tag, None) + .await + .expect("insert victim head") + .1 + ); + + let mut tx = db + .begin_transaction() + .await + .expect("begin caller transaction"); + let result = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &new, + &replace_d_tag, + None, + replaceable::ParameterizedReplacePrecondition::Unconditional, + ) + .await + .expect("replace inside caller transaction"); + assert_eq!( + result.status, + replaceable::ParameterizedReplaceStatus::Inserted + ); + + let leaked: Option = sqlx::query_scalar( + "SELECT NULLIF(current_setting('buzz.nip_rs_hard_delete', true), '')", + ) + .fetch_one(&mut *tx) + .await + .expect("read hard-delete opt-in after replacement"); + assert_ne!(leaked.as_deref(), Some("on")); + + let unauthorized = sqlx::query( + "DELETE FROM events WHERE community_id=$1 AND kind=30078 \ + AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(&victim_d_tag) + .execute(&mut *tx) + .await; + assert!( + unauthorized.is_err(), + "replacement opt-in must not authorize later caller SQL" + ); + tx.rollback().await.expect("roll back caller transaction"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn parameterized_replacement_in_existing_transaction_honors_revision_and_rollback() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let d_tag = format!("transactional-project-{}", Uuid::new_v4().simple()); + let base = Timestamp::now().as_secs(); + let event = |content: &str, timestamp: u64| { + EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content) + .tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")]) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&keys) + .expect("sign project") + }; + let old = event("old", base); + let new = event("new", base + 1); + + assert!( + db.replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("insert old head") + .1 + ); + + let mut tx = db.begin_transaction().await.expect("begin replacement tx"); + let outcome = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &new, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExpectedRevision( + old.id.as_bytes().as_slice(), + ), + ) + .await + .expect("replace inside caller transaction"); + assert_eq!( + outcome.status, + replaceable::ParameterizedReplaceStatus::Inserted + ); + tx.rollback().await.expect("roll back replacement tx"); + + let live_id: Vec = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(buzz_core::kind::KIND_PROJECT as i32) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_one(&db.pool) + .await + .expect("load live head after rollback"); + assert_eq!(live_id, old.id.as_bytes().to_vec()); + + let mut tx = db + .begin_transaction() + .await + .expect("begin stale revision tx"); + let mismatch = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &new, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExpectedRevision( + [0x42; 32].as_slice(), + ), + ) + .await + .expect("evaluate stale revision"); + assert_eq!( + mismatch.status, + replaceable::ParameterizedReplaceStatus::RevisionMismatch + ); + tx.rollback().await.expect("roll back stale revision tx"); + + let missing_d_tag = format!("missing-project-{}", Uuid::new_v4().simple()); + let missing = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), + "missing", + ) + .tags(vec![ + Tag::parse(["d", missing_d_tag.as_str()]).expect("missing d tag") + ]) + .custom_created_at(Timestamp::from(base + 2)) + .sign_with_keys(&keys) + .expect("sign missing project"); + let mut tx = db + .begin_transaction() + .await + .expect("begin missing revision tx"); + let missing_result = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &missing, + &missing_d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExpectedRevision( + [0x24; 32].as_slice(), + ), + ) + .await + .expect("evaluate missing revision"); + assert_eq!( + missing_result.status, + replaceable::ParameterizedReplaceStatus::RevisionMissing + ); + tx.rollback().await.expect("roll back missing revision tx"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn parameterized_replacement_rolls_back_when_mention_indexing_fails() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (pool, scratch_name) = create_scratch_db(&admin, "atomic_parameterized").await; + let db = Db::from_pool(pool.clone()); + let community = CommunityId::from_uuid(make_community(&pool).await); + let keys = Keys::generate(); + let mentioned = Keys::generate().public_key().to_hex(); + let d_tag = format!("mention-project-{}", Uuid::new_v4().simple()); + let tags = || { + vec![ + Tag::parse(["d", d_tag.as_str()]).expect("d tag"), + Tag::parse(["p", mentioned.as_str()]).expect("p tag"), + ] + }; + let base = Timestamp::now().as_secs(); + let old = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), "old") + .tags(tags()) + .custom_created_at(Timestamp::from(base)) + .sign_with_keys(&keys) + .expect("sign old project"); + let new = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), "new") + .tags(tags()) + .custom_created_at(Timestamp::from(base + 1)) + .sign_with_keys(&keys) + .expect("sign new project"); + + assert!( + db.replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("insert old project") + .1 + ); + sqlx::query( + "CREATE FUNCTION reject_test_mention() RETURNS trigger AS $$ \ + BEGIN RAISE EXCEPTION 'injected mention failure'; END; \ + $$ LANGUAGE plpgsql", + ) + .execute(&pool) + .await + .expect("create failure function"); + sqlx::query( + "CREATE TRIGGER reject_test_mention BEFORE INSERT ON event_mentions \ + FOR EACH ROW EXECUTE FUNCTION reject_test_mention()", + ) + .execute(&pool) + .await + .expect("install failure injection"); + + let mut tx = db + .begin_transaction() + .await + .expect("begin caller transaction"); + let error = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &new, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::Unconditional, + ) + .await + .expect_err("mention failure must fail replacement"); + assert!(error.to_string().contains("injected mention failure")); + + let probe: i32 = sqlx::query_scalar("SELECT 1") + .fetch_one(&mut *tx) + .await + .expect("inner failure must leave caller transaction usable"); + assert_eq!(probe, 1); + + let live_id: Vec = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(buzz_core::kind::KIND_PROJECT as i32) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_one(&mut *tx) + .await + .expect("load live project after failed indexing"); + assert_eq!(live_id, old.id.as_bytes().to_vec()); + let new_rows: i64 = + sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(new.id.as_bytes().as_slice()) + .fetch_one(&mut *tx) + .await + .expect("count rolled-back project"); + assert_eq!(new_rows, 0); + tx.commit().await.expect("commit usable caller transaction"); + + drop_scratch_db(&admin, pool, &scratch_name).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn parameterized_duplicate_restores_live_head_inside_caller_transaction() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let d_tag = format!("duplicate-project-{}", Uuid::new_v4().simple()); + let base = Timestamp::now().as_secs(); + let event = |content: &str, timestamp: u64| { + EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content) + .tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")]) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&keys) + .expect("sign project") + }; + let old = event("old-live-head", base); + let duplicate = event("soft-deleted-duplicate", base + 1); + + assert!( + db.replace_parameterized_event(community, &duplicate, &d_tag, None) + .await + .expect("insert future duplicate") + .1 + ); + sqlx::query("UPDATE events SET deleted_at=NOW() WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(duplicate.id.as_bytes().as_slice()) + .execute(&db.pool) + .await + .expect("soft-delete duplicate row"); + + let mut seed_tx = db + .begin_transaction() + .await + .expect("begin seed transaction"); + let (_, was_inserted) = + event::insert_event_in_transaction(&mut seed_tx, community, &old, None) + .await + .expect("insert older live head"); + assert!(was_inserted); + seed_tx.commit().await.expect("commit older live head"); + + let mut tx = db + .begin_transaction() + .await + .expect("begin caller transaction"); + let result = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &duplicate, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::Unconditional, + ) + .await + .expect("evaluate soft-deleted duplicate"); + assert_eq!( + result.status, + replaceable::ParameterizedReplaceStatus::Duplicate + ); + + let live_id: Vec = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(buzz_core::kind::KIND_PROJECT as i32) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_one(&mut *tx) + .await + .expect("caller transaction remains usable after duplicate"); + assert_eq!(live_id, old.id.as_bytes().to_vec()); + tx.rollback().await.expect("roll back caller transaction"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn concurrent_parameterized_replacement_keeps_deterministic_head() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let d_tag = format!("concurrent-project-{}", Uuid::new_v4().simple()); + let created_at = Timestamp::now().as_secs(); + let event = |content: &str, timestamp: u64| { + EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content) + .tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")]) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&keys) + .expect("sign project") + }; + let first = event("first", created_at); + let second = event("second", created_at); + let expected = if first.id.as_bytes() < second.id.as_bytes() { + &first + } else { + &second + }; + + let (first_result, second_result) = tokio::join!( + db.replace_parameterized_event(community, &first, &d_tag, None), + db.replace_parameterized_event(community, &second, &d_tag, None), + ); + let first_inserted = first_result.expect("first concurrent writer").1; + let second_inserted = second_result.expect("second concurrent writer").1; + assert!( + first_inserted || second_inserted, + "at least one concurrent writer must insert", + ); + + let live_ids: Vec> = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(buzz_core::kind::KIND_PROJECT as i32) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_all(&db.pool) + .await + .expect("load concurrent live head"); + assert_eq!(live_ids, vec![expected.id.as_bytes().to_vec()]); + + assert!( + !db.replace_parameterized_event(community, expected, &d_tag, None) + .await + .expect("replay winning event") + .1, + "replaying the live event must be idempotent", + ); + let stale = event("stale", created_at.saturating_sub(1)); + assert!( + !db.replace_parameterized_event(community, &stale, &d_tag, None) + .await + .expect("submit stale event") + .1, + "an older event must not replace the live head", + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn mesh_status_replacement_keeps_one_physical_row() { diff --git a/crates/buzz-db/src/replaceable.rs b/crates/buzz-db/src/replaceable.rs new file mode 100644 index 0000000000..aa57a51666 --- /dev/null +++ b/crates/buzz-db/src/replaceable.rs @@ -0,0 +1,417 @@ +//! Replaceable-event persistence and coordinate locking. + +use buzz_core::{CommunityId, StoredEvent}; +use buzz_datastore_tracing::datastore_span; +use chrono::{DateTime, Utc}; +use sqlx::{Acquire, Postgres, Transaction}; +use uuid::Uuid; + +use crate::{Db, DbError, Result}; + +/// Result category for a parameterized-replaceable event write. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ParameterizedReplaceStatus { + /// The incoming event was inserted as the coordinate's live head. + Inserted, + /// The exact event was already accepted. + Duplicate, + /// A newer event, or lower-ID same-second event, already dominates it. + Superseded, + /// A requested current revision has no live coordinate head. + RevisionMissing, + /// The live coordinate head differs from the requested revision. + RevisionMismatch, + /// An exact replay was required, but the event is not the live head. + ReplayOnlyMiss, +} + +/// Structural precondition for a parameterized-replaceable write. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ParameterizedReplacePrecondition<'a> { + /// Apply normal NIP-33 ordering without a revision precondition. + Unconditional, + /// Require the live head to match this validated event ID. + ExpectedRevision(&'a [u8]), + /// Accept only an exact live-head replay and perform no mutation otherwise. + ExactReplayOnly, +} + +/// Result of a transaction-bound parameterized-replaceable event write. +#[derive(Clone, Debug)] +pub struct ParameterizedReplaceResult { + /// Stored representation of the submitted event. + pub event: StoredEvent, + /// Whether and why the coordinate accepted the event. + pub status: ParameterizedReplaceStatus, +} + +impl ParameterizedReplaceResult { + fn new( + event: &nostr::Event, + received_at: DateTime, + channel_id: Option, + status: ParameterizedReplaceStatus, + ) -> Self { + Self { + event: StoredEvent::with_received_at( + event.clone(), + received_at, + channel_id, + status == ParameterizedReplaceStatus::Inserted, + ), + status, + } + } +} + +/// Derive the transaction-scoped advisory-lock key for an event coordinate. +/// +/// Hash collisions only add serialization; the SQL predicates still determine +/// which rows are read or changed. +pub(crate) fn event_replacement_lock_key( + community_id: CommunityId, + kind: i32, + pubkey: &[u8], + coordinate: Option<&[u8]>, +) -> i64 { + let mut hash: u64 = 0xcbf29ce484222325; + let kind_bytes = kind.to_le_bytes(); + for bytes in [ + community_id.as_uuid().as_bytes().as_slice(), + kind_bytes.as_slice(), + pubkey, + ] { + for byte in bytes { + hash ^= *byte as u64; + hash = hash.wrapping_mul(0x100000001b3); + } + } + if let Some(coordinate) = coordinate { + for byte in coordinate { + hash ^= *byte as u64; + hash = hash.wrapping_mul(0x100000001b3); + } + } + hash as i64 +} + +/// Replace a parameterized event in a caller-owned transaction. +/// +/// This function acquires a transaction-scoped advisory lock but never commits +/// or rolls back the outer transaction. The typed precondition can require the +/// current live head to have an exact event ID or restrict the operation to an +/// idempotent replay. +async fn replace_parameterized_event_in_transaction_impl( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + event: &nostr::Event, + d_tag: &str, + channel_id: Option, + precondition: ParameterizedReplacePrecondition<'_>, +) -> Result { + let kind_i32 = buzz_core::kind::event_kind_i32(event); + let pubkey_bytes = event.pubkey.to_bytes(); + let created_at_secs = event.created_at.as_secs() as i64; + let created_at = DateTime::from_timestamp(created_at_secs, 0) + .ok_or(DbError::InvalidTimestamp(created_at_secs))?; + let received_at = Utc::now(); + + let lock_key = event_replacement_lock_key( + community_id, + kind_i32, + pubkey_bytes.as_slice(), + Some(d_tag.as_bytes()), + ); + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_key) + .execute(&mut **tx) + .await?; + + let d_tag_count = event + .tags + .iter() + .filter(|tag| tag.as_slice().first().is_some_and(|part| part == "d")) + .count(); + let has_exact_d_tag = event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.len() >= 2 && parts[0] == "d" && parts[1] == d_tag + }); + let read_state_t_tag_count = event + .tags + .iter() + .filter(|tag| { + let parts = tag.as_slice(); + parts.len() == 2 && parts[0] == "t" && parts[1] == "read-state" + }) + .count(); + let is_nip_rs = kind_i32 == buzz_core::kind::KIND_READ_STATE as i32 + && d_tag_count == 1 + && has_exact_d_tag + && d_tag.strip_prefix("read-state:").is_some_and(|slot| { + slot.len() == 32 + && slot + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + }) + && read_state_t_tag_count == 1; + let is_buzz_mesh_status = kind_i32 == buzz_core::kind::KIND_BOOKMARK_SET as i32 + && d_tag.starts_with("buzz-mesh-member-status:") + && event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.len() == 2 && parts[0] == "k" && parts[1] == "buzz-mesh-status" + }); + let hard_delete_superseded = is_nip_rs || is_buzz_mesh_status; + + let existing: Option<(DateTime, Vec)> = sqlx::query_as( + "SELECT created_at, id FROM events \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \ + ORDER BY created_at DESC, id ASC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(d_tag) + .fetch_optional(&mut **tx) + .await?; + let watermark: Option<(DateTime, Vec)> = if is_nip_rs { + sqlx::query_as( + "SELECT created_at, event_id FROM parameterized_event_watermarks \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4", + ) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(d_tag) + .fetch_optional(&mut **tx) + .await? + } else { + None + }; + + let incoming_id = event.id.as_bytes().as_slice(); + if existing + .as_ref() + .is_some_and(|(_, existing_id)| existing_id.as_slice() == incoming_id) + || watermark + .as_ref() + .is_some_and(|(_, event_id)| event_id.as_slice() == incoming_id) + { + return Ok(ParameterizedReplaceResult::new( + event, + received_at, + channel_id, + ParameterizedReplaceStatus::Duplicate, + )); + } + + if precondition == ParameterizedReplacePrecondition::ExactReplayOnly { + return Ok(ParameterizedReplaceResult::new( + event, + received_at, + channel_id, + ParameterizedReplaceStatus::ReplayOnlyMiss, + )); + } + + if let ParameterizedReplacePrecondition::ExpectedRevision(expected_revision) = precondition { + let status = match existing.as_ref() { + None => Some(ParameterizedReplaceStatus::RevisionMissing), + Some((_, existing_id)) if existing_id.as_slice() != expected_revision => { + Some(ParameterizedReplaceStatus::RevisionMismatch) + } + Some(_) => None, + }; + if let Some(status) = status { + return Ok(ParameterizedReplaceResult::new( + event, + received_at, + channel_id, + status, + )); + } + } + + let dominated = existing + .iter() + .chain(watermark.iter()) + .any(|(accepted_ts, accepted_id)| { + created_at < *accepted_ts + || (created_at == *accepted_ts && incoming_id >= accepted_id.as_slice()) + }); + if dominated { + return Ok(ParameterizedReplaceResult::new( + event, + received_at, + channel_id, + ParameterizedReplaceStatus::Superseded, + )); + } + + let mut savepoint = tx.begin().await?; + if existing.is_some() { + let previous_nip_rs_hard_delete: Option = if is_nip_rs { + sqlx::query_scalar( + "SELECT NULLIF(current_setting('buzz.nip_rs_hard_delete', true), '')", + ) + .fetch_one(&mut *savepoint) + .await? + } else { + None + }; + if is_nip_rs { + sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)") + .execute(&mut *savepoint) + .await?; + } + let statement = if hard_delete_superseded { + "DELETE FROM events \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL" + } else { + "UPDATE events SET deleted_at = NOW() \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL" + }; + sqlx::query(statement) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(d_tag) + .execute(&mut *savepoint) + .await?; + + if is_nip_rs { + let previous_value = previous_nip_rs_hard_delete.as_deref().unwrap_or_default(); + sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', $1, true)") + .bind(previous_value) + .execute(&mut *savepoint) + .await?; + } + + if hard_delete_superseded { + if let Some((_, existing_id)) = &existing { + sqlx::query("DELETE FROM event_mentions WHERE community_id = $1 AND event_id = $2") + .bind(community_id.as_uuid()) + .bind(existing_id) + .execute(&mut *savepoint) + .await?; + } + } + } + + let sig_bytes = event.sig.serialize(); + let tags_json = serde_json::to_value(&event.tags)?; + let insert_result = sqlx::query( + "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \ + ON CONFLICT DO NOTHING", + ) + .bind(community_id.as_uuid()) + .bind(incoming_id) + .bind(pubkey_bytes.as_slice()) + .bind(created_at) + .bind(kind_i32) + .bind(&tags_json) + .bind(&event.content) + .bind(sig_bytes.as_slice()) + .bind(received_at) + .bind(channel_id) + .bind(d_tag) + .bind(crate::event::extract_not_before(event)) + .execute(&mut *savepoint) + .await?; + + if insert_result.rows_affected() == 0 { + savepoint.rollback().await?; + return Ok(ParameterizedReplaceResult::new( + event, + received_at, + channel_id, + ParameterizedReplaceStatus::Duplicate, + )); + } + + if is_nip_rs { + sqlx::query( + "INSERT INTO parameterized_event_watermarks \ + (community_id, kind, pubkey, d_tag, created_at, event_id) \ + VALUES ($1, $2, $3, $4, $5, $6) \ + ON CONFLICT (community_id, kind, pubkey, d_tag) DO UPDATE SET \ + created_at = EXCLUDED.created_at, event_id = EXCLUDED.event_id", + ) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(d_tag) + .bind(created_at) + .bind(incoming_id) + .execute(&mut *savepoint) + .await?; + } + + crate::insert_mentions_in_transaction(&mut savepoint, community_id, event, channel_id).await?; + savepoint.commit().await?; + + Ok(ParameterizedReplaceResult::new( + event, + received_at, + channel_id, + ParameterizedReplaceStatus::Inserted, + )) +} + +impl Db { + /// Replace a NIP-33 event inside a caller-owned transaction. + /// + /// The caller owns commit or rollback. Requiring [`Transaction`] here and + /// in the internal state machine makes the advisory-lock contract explicit. + pub async fn replace_parameterized_event_in_transaction( + &self, + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + event: &nostr::Event, + d_tag: &str, + channel_id: Option, + precondition: ParameterizedReplacePrecondition<'_>, + ) -> Result { + replace_parameterized_event_in_transaction_impl( + tx, + community_id, + event, + d_tag, + channel_id, + precondition, + ) + .await + } + + /// Atomically replace a NIP-33 parameterized replaceable event. + /// + /// Replacement keys on `(kind, pubkey, d_tag)` across channels. The + /// highest timestamp wins; same-second ties use the lowest event ID. + #[datastore_span(name = "replace_parameterized_event", system = "postgresql")] + pub async fn replace_parameterized_event( + &self, + community_id: CommunityId, + event: &nostr::Event, + d_tag: &str, + channel_id: Option, + ) -> Result<(StoredEvent, bool)> { + let mut tx = self.pool.begin().await?; + let result = self + .replace_parameterized_event_in_transaction( + &mut tx, + community_id, + event, + d_tag, + channel_id, + ParameterizedReplacePrecondition::Unconditional, + ) + .await?; + let was_inserted = result.status == ParameterizedReplaceStatus::Inserted; + if was_inserted { + tx.commit().await?; + } else { + tx.rollback().await?; + } + Ok((result.event, was_inserted)) + } +} diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index d8569a7a86..007db43ffd 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -105,8 +105,9 @@ async fn persist_command_event( event: &Event, channel_id_override: Option, ) -> Result { - let channel_id = channel_id_override.or_else(|| extract_channel_id(event)); + use buzz_db::replaceable::{ParameterizedReplacePrecondition, ParameterizedReplaceStatus}; + let channel_id = channel_id_override.or_else(|| extract_channel_id(event)); let mut tx = db .begin_transaction() .await @@ -118,22 +119,8 @@ async fn persist_command_event( IngestError::Rejected(format!("restricted: community writes are fenced: {error}")) })?; - // INSERT with ON CONFLICT DO NOTHING — idempotency guard. - let id_bytes = event.id.as_bytes(); - let pubkey_bytes = event.pubkey.to_bytes(); - let sig_bytes = event.sig.serialize(); - let tags_json = serde_json::to_value(&event.tags) - .map_err(|e| IngestError::Internal(format!("error: serialize tags: {e}")))?; - let kind_i32 = event.kind.as_u16() as i32; - let created_at_secs = event.created_at.as_secs() as i64; - let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0).ok_or_else(|| { - IngestError::Rejected(format!("invalid: bad timestamp {created_at_secs}")) - })?; - let received_at = chrono::Utc::now(); - - // Extract d_tag for parameterized replaceable kinds (NIP-33). let d_tag = buzz_db::event::extract_d_tag(event); - if let Some(ref d_tag) = d_tag { + if let Some(d_tag) = d_tag.as_deref() { if d_tag.len() > buzz_db::event::D_TAG_MAX_LEN { return Err(IngestError::Rejected(format!( "invalid: d tag too long ({} bytes, max {})", @@ -142,130 +129,81 @@ async fn persist_command_event( ))); } - // Command kinds normally use plain insert semantics, but workflow - // definitions are NIP-33 events. Serialize writers for the same - // coordinate and reject stale writes before executing the domain - // mutation, otherwise old updates can overwrite newer workflow state. - let lock_key = { - let mut h: u64 = 0xcbf29ce484222325; - for b in tenant.community().as_uuid().as_bytes() { - h ^= *b as u64; - h = h.wrapping_mul(0x100000001b3); - } - for b in kind_i32.to_le_bytes() { - h ^= b as u64; - h = h.wrapping_mul(0x100000001b3); - } - for b in pubkey_bytes.as_slice() { - h ^= *b as u64; - h = h.wrapping_mul(0x100000001b3); - } - for b in d_tag.as_bytes() { - h ^= *b as u64; - h = h.wrapping_mul(0x100000001b3); - } - h as i64 + let kind = event.kind.as_u16() as i32; + let (expected_revision, revision_error) = match parse_expected_workflow_revision( + kind, + extract_tag(event, "expected-revision").as_deref(), + ) { + Ok(expected_revision) => (expected_revision, None), + Err(error) => (None, Some(error)), }; - - sqlx::query("SELECT pg_advisory_xact_lock($1)") - .bind(lock_key) - .execute(tx.as_mut()) - .await - .map_err(|e| IngestError::Internal(format!("error: lock event coordinate: {e}")))?; - - let existing: Option<(chrono::DateTime, Vec)> = sqlx::query_as( - "SELECT created_at, id FROM events \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \ - ORDER BY created_at DESC, id ASC LIMIT 1", - ) - .bind(tenant.community().as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(d_tag) - .fetch_optional(tx.as_mut()) - .await - .map_err(|e| IngestError::Internal(format!("error: query event coordinate: {e}")))?; - - let incoming_id = event.id.as_bytes().as_slice(); - if existing - .as_ref() - .is_some_and(|(_, existing_id)| existing_id.as_slice() == incoming_id) - { - return Ok(PersistResult::Duplicate); - } - - let expected_revision = extract_tag(event, "expected-revision"); - validate_workflow_revision( - kind_i32, - expected_revision.as_deref(), - existing.as_ref().map(|(_, id)| id.as_slice()), - )?; - if let Some((existing_ts, existing_id)) = existing { - let dominated = created_at < existing_ts - || (created_at == existing_ts && incoming_id >= existing_id.as_slice()); - if dominated { - if kind_i32 == KIND_WORKFLOW_DEF as i32 && expected_revision.is_some() { - return Err(IngestError::Rejected( - "conflict: workflow update was superseded; refresh and try again".into(), - )); - } - return Ok(PersistResult::Duplicate); - } - - sqlx::query( - "UPDATE events SET deleted_at = NOW() \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL", + let precondition = if revision_error.is_some() { + ParameterizedReplacePrecondition::ExactReplayOnly + } else if let Some(expected_revision) = expected_revision.as_deref() { + ParameterizedReplacePrecondition::ExpectedRevision(expected_revision) + } else { + ParameterizedReplacePrecondition::Unconditional + }; + let result = db + .replace_parameterized_event_in_transaction( + &mut tx, + tenant.community(), + event, + d_tag, + channel_id, + precondition, ) - .bind(tenant.community().as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(d_tag) - .execute(tx.as_mut()) .await - .map_err(|e| IngestError::Internal(format!("error: replace old event: {e}")))?; - } + .map_err(|e| { + IngestError::Internal(format!("error: replace parameterized event: {e}")) + })?; + + return match result.status { + ParameterizedReplaceStatus::Inserted => Ok(PersistResult::Inserted(tx)), + ParameterizedReplaceStatus::Duplicate => Ok(PersistResult::Duplicate), + ParameterizedReplaceStatus::Superseded + if kind == KIND_WORKFLOW_DEF as i32 && expected_revision.is_some() => + { + Err(IngestError::Rejected( + "conflict: workflow update was superseded; refresh and try again".into(), + )) + } + ParameterizedReplaceStatus::Superseded => Ok(PersistResult::Duplicate), + ParameterizedReplaceStatus::RevisionMissing => Err(IngestError::Rejected( + "conflict: workflow revision does not exist".into(), + )), + ParameterizedReplaceStatus::RevisionMismatch => Err(IngestError::Rejected( + "conflict: workflow changed since it was loaded".into(), + )), + ParameterizedReplaceStatus::ReplayOnlyMiss => match revision_error { + Some(error) => Err(error), + None => Err(IngestError::Internal( + "error: replay-only replacement lacked a revision error".into(), + )), + }, + }; } - let result = sqlx::query( - r#" - INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) - ON CONFLICT DO NOTHING - "#, - ) - .bind(tenant.community().as_uuid()) - .bind(id_bytes.as_slice()) - .bind(pubkey_bytes.as_slice()) - .bind(created_at) - .bind(kind_i32) - .bind(&tags_json) - .bind(&event.content) - .bind(sig_bytes.as_slice()) - .bind(received_at) - .bind(channel_id) - .bind(d_tag.as_deref()) - .execute(tx.as_mut()) - .await - .map_err(|e| IngestError::Internal(format!("error: insert event: {e}")))?; - - if result.rows_affected() == 0 { - // Duplicate — rollback (implicit on drop) and signal idempotent success. - Ok(PersistResult::Duplicate) - } else { + let (_, was_inserted) = + buzz_db::event::insert_event_in_transaction(&mut tx, tenant.community(), event, channel_id) + .await + .map_err(|e| IngestError::Internal(format!("error: insert event: {e}")))?; + if was_inserted { Ok(PersistResult::Inserted(tx)) + } else { + Ok(PersistResult::Duplicate) } } -fn validate_workflow_revision( +fn parse_expected_workflow_revision( kind: i32, expected_revision: Option<&str>, - existing_id: Option<&[u8]>, -) -> Result<(), IngestError> { +) -> Result>, IngestError> { if kind != KIND_WORKFLOW_DEF as i32 { - return Ok(()); + return Ok(None); } - let expected_id = expected_revision + expected_revision .map(|expected| { let id = hex::decode(expected).map_err(|_| { IngestError::Rejected("invalid: bad expected workflow revision".into()) @@ -277,18 +215,7 @@ fn validate_workflow_revision( } Ok(id) }) - .transpose()?; - - match (expected_id.as_deref(), existing_id) { - (None, _) => Ok(()), - (Some(_), None) => Err(IngestError::Rejected( - "conflict: workflow revision does not exist".into(), - )), - (Some(expected), Some(existing)) if expected != existing => Err(IngestError::Rejected( - "conflict: workflow changed since it was loaded".into(), - )), - (Some(_), Some(_)) => Ok(()), - } + .transpose() } /// Extract all `p` tag values (hex pubkeys) from an event. @@ -425,7 +352,7 @@ async fn handle_dm_open( .await .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; - // Commit: event + mutation succeeded atomically. + // Finalize the idempotency record after the separate mutation succeeds. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; @@ -586,7 +513,7 @@ async fn handle_dm_add_member( .await .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; - // Commit: event + mutation succeeded atomically. + // Finalize the idempotency record after the separate mutation succeeds. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; @@ -691,7 +618,7 @@ async fn handle_dm_hide( .await .map_err(|e| IngestError::Internal(format!("error: db hide_dm: {e}")))?; - // Commit: event + mutation succeeded atomically. + // Finalize the idempotency record after the separate mutation succeeds. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; @@ -994,7 +921,7 @@ async fn handle_workflow_trigger( .await .map_err(|e| IngestError::Internal(format!("error: db create_workflow_run: {e}")))?; - // Commit: event + run creation succeeded atomically. + // Finalize the idempotency record after the separate run creation succeeds. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; @@ -1169,7 +1096,7 @@ async fn handle_approval_grant( )); } - // Commit: event + approval update succeeded atomically. + // Finalize the idempotency record after the separate approval update succeeds. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; @@ -1280,7 +1207,7 @@ async fn handle_approval_deny( )); } - // Commit: event + approval denial succeeded atomically. + // Finalize the idempotency record after the separate approval denial succeeds. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; @@ -1489,57 +1416,40 @@ mod tests { .expect("workflow event") } - fn rejection_message(result: Result<(), IngestError>) -> String { + fn rejection_message(result: Result>, IngestError>) -> String { match result { Err(IngestError::Rejected(message)) => message, Err(IngestError::AuthFailed(message)) => panic!("unexpected auth failure: {message}"), Err(IngestError::Internal(message)) => panic!("unexpected internal failure: {message}"), - Ok(()) => panic!("expected revision validation to fail"), + Ok(_) => panic!("expected revision parsing to fail"), } } #[test] - fn workflow_revision_accepts_create_and_matching_update() { - let existing = [0x42; 32]; - assert!(validate_workflow_revision(KIND_WORKFLOW_DEF as i32, None, None).is_ok()); - assert!(validate_workflow_revision( - KIND_WORKFLOW_DEF as i32, - Some(&hex::encode(existing)), - Some(&existing), - ) - .is_ok()); + fn workflow_revision_parser_accepts_create_and_valid_update() { + let revision = [0x42; 32]; + assert_eq!( + parse_expected_workflow_revision(KIND_WORKFLOW_DEF as i32, None) + .expect("tagless workflow"), + None + ); + assert_eq!( + parse_expected_workflow_revision( + KIND_WORKFLOW_DEF as i32, + Some(&hex::encode(revision)), + ) + .expect("valid revision"), + Some(revision.to_vec()) + ); } #[test] - fn workflow_revision_rejects_stale_and_malformed_updates() { - let existing = [0x42; 32]; - let stale = [0x24; 32]; - assert_eq!( - rejection_message(validate_workflow_revision( - KIND_WORKFLOW_DEF as i32, - Some(&hex::encode(stale)), - Some(&existing), - )), - "conflict: workflow changed since it was loaded", - ); - assert!( - validate_workflow_revision(KIND_WORKFLOW_DEF as i32, None, Some(&existing)).is_ok(), - "tagless legacy workflow updates remain compatible during rollout", - ); + fn workflow_revision_parser_rejects_malformed_values() { for malformed in ["not-hex", "42"] { assert_eq!( - rejection_message(validate_workflow_revision( + rejection_message(parse_expected_workflow_revision( KIND_WORKFLOW_DEF as i32, Some(malformed), - Some(&existing), - )), - "invalid: bad expected workflow revision", - ); - assert_eq!( - rejection_message(validate_workflow_revision( - KIND_WORKFLOW_DEF as i32, - Some(malformed), - None, )), "invalid: bad expected workflow revision", ); @@ -1547,14 +1457,11 @@ mod tests { } #[test] - fn workflow_revision_rejects_update_for_missing_coordinate() { + fn revision_tag_does_not_change_other_command_kinds() { assert_eq!( - rejection_message(validate_workflow_revision( - KIND_WORKFLOW_DEF as i32, - Some(&hex::encode([0x42; 32])), - None, - )), - "conflict: workflow revision does not exist", + parse_expected_workflow_revision(KIND_DM_OPEN as i32, Some("not-hex")) + .expect("non-workflow revision tag"), + None ); } @@ -1567,6 +1474,25 @@ mod tests { let created_at = Timestamp::now().as_secs(); let create = workflow_event(&keys, workflow_id, created_at, None, "create"); + let missing_revision = hex::encode([0x24; 32]); + let missing_revision_update = workflow_event( + &keys, + Uuid::new_v4(), + created_at, + Some(&missing_revision), + "missing-revision", + ); + let error = match persist_command_event(&db, &tenant, &missing_revision_update, None).await + { + Err(error) => error, + Ok(_) => panic!("missing revision must not create a workflow"), + }; + assert!(matches!( + error, + IngestError::Rejected(ref message) + if message == "conflict: workflow revision does not exist" + )); + let PersistResult::Inserted(tx) = persist_command_event(&db, &tenant, &create, None) .await .expect("persist create") @@ -1621,6 +1547,23 @@ mod tests { PersistResult::Duplicate )); + let stale_revision_update = workflow_event( + &keys, + workflow_id, + created_at + 1, + Some(&create_revision), + "stale-revision", + ); + let error = match persist_command_event(&db, &tenant, &stale_revision_update, None).await { + Err(error) => error, + Ok(_) => panic!("stale revision must not replace the current workflow"), + }; + assert!(matches!( + error, + IngestError::Rejected(ref message) + if message == "conflict: workflow changed since it was loaded" + )); + let error = match persist_command_event(&db, &tenant, &dominated_update, None).await { Err(error) => error, Ok(_) => panic!("distinct dominated CAS update must not report duplicate success"), @@ -1632,8 +1575,55 @@ mod tests { )); } - #[test] - fn revision_tag_does_not_change_other_command_kinds() { - assert!(validate_workflow_revision(KIND_DM_OPEN as i32, Some("not-hex"), None).is_ok()); + #[tokio::test] + #[ignore = "requires Postgres"] + async fn workflow_persistence_replays_legacy_malformed_revision_before_validation() { + let (db, tenant) = persistence_test_context().await; + let keys = Keys::generate(); + let workflow_id = Uuid::new_v4(); + let created_at = Timestamp::now().as_secs(); + let legacy = workflow_event( + &keys, + workflow_id, + created_at, + Some("not-hex"), + "legacy-malformed", + ); + + let mut tx = db.begin_transaction().await.expect("begin legacy seed"); + let (_, was_inserted) = buzz_db::event::insert_event_in_transaction( + &mut tx, + tenant.community(), + &legacy, + extract_channel_id(&legacy), + ) + .await + .expect("seed legacy workflow event"); + assert!(was_inserted); + tx.commit().await.expect("commit legacy seed"); + + assert!(matches!( + persist_command_event(&db, &tenant, &legacy, None) + .await + .expect("exact legacy replay must remain idempotent"), + PersistResult::Duplicate + )); + + let distinct = workflow_event( + &keys, + workflow_id, + created_at + 1, + Some("not-hex"), + "distinct-malformed", + ); + let error = match persist_command_event(&db, &tenant, &distinct, None).await { + Err(error) => error, + Ok(_) => panic!("distinct malformed revision must remain rejected"), + }; + assert!(matches!( + error, + IngestError::Rejected(ref message) + if message == "invalid: bad expected workflow revision" + )); } }