diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index e78d832129..4d28503edd 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -13,7 +13,7 @@ use crate::{deletion, event, DbError, EventQuery, Result}; use buzz_datastore_tracing::datastore_span; use chrono::{DateTime, Utc}; use sqlx::postgres::PgPoolOptions; -use sqlx::{PgPool, QueryBuilder}; +use sqlx::{Acquire, PgPool, QueryBuilder}; use std::time::Duration; use uuid::Uuid; @@ -1348,12 +1348,23 @@ impl Db { None, ) .await?; - tx.commit().await?; if result.1 { - 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}"); + // Index mentions in the event's own transaction so a concurrent + // deletion cannot strand an orphan row. A savepoint keeps the + // existing best-effort contract: an indexing failure never rejects + // an otherwise valid event. + let mut savepoint = tx.begin().await?; + match insert_mentions_in_transaction(&mut savepoint, community_id, event, channel_id) + .await + { + Ok(()) => savepoint.commit().await?, + Err(e) => { + savepoint.rollback().await?; + tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); + } } } + tx.commit().await?; Ok(result) } diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 14886aee2c..3998e10b3b 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -3811,3 +3811,159 @@ async fn e_tag_any_runs_with_real_binds_on_p_join_and_count() { #[path = "tests/thread_window_postgres_tests.rs"] mod thread_window_postgres_tests; + +/// `insert_event_with_serving_write_guard` indexes mentions in the event's own +/// transaction, and an indexing failure stays best-effort: the savepoint rolls +/// back only the mention rows, never the event. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn serving_write_guard_insert_indexes_mentions_in_event_transaction() { + use nostr::{EventBuilder, Keys, Kind, Tag}; + + let db = setup_db().await; + let community = db + .ensure_configured_community(&format!( + "serving-mentions-{}.example", + Uuid::new_v4().simple() + )) + .await + .expect("create serving-write mention community") + .id; + let store = db.deletion_store(); + let lease = store + .acquire_serving_write_lease( + community, + "mention_index_test", + "mention-index-test", + std::time::Duration::from_secs(30), + ) + .await + .expect("acquire serving-write lease"); + let mentioned = Keys::generate().public_key().to_hex(); + let build = |content: &str| { + EventBuilder::new(Kind::TextNote, content) + .tags([Tag::parse(["p", mentioned.as_str()]).expect("p tag")]) + .sign_with_keys(&Keys::generate()) + .expect("sign event") + }; + let counts = |event_id: Vec| { + let pool = db.pool.clone(); + async move { + let live: i64 = sqlx::query_scalar( + "SELECT count(*) FROM events \ + WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(&event_id) + .fetch_one(&pool) + .await + .expect("count live events"); + let mentions: i64 = sqlx::query_scalar( + "SELECT count(*) FROM event_mentions WHERE community_id = $1 AND event_id = $2", + ) + .bind(community.as_uuid()) + .bind(&event_id) + .fetch_one(&pool) + .await + .expect("count mentions"); + (live, mentions) + } + }; + + // Pin in-transaction indexing: the mention insert must run in the same + // top-level transaction that inserted the event row. A post-commit index + // runs in a later transaction, raises here, is logged as a warning, and + // leaves zero mentions. + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE FUNCTION test_require_event_tx_mentions() RETURNS trigger AS $$ \ + BEGIN \ + IF NEW.community_id = '{}'::uuid AND NOT EXISTS ( \ + SELECT 1 FROM events \ + WHERE community_id = NEW.community_id AND id = NEW.event_id \ + AND xmin = pg_current_xact_id()::xid) THEN \ + RAISE EXCEPTION 'test: post-commit mention'; \ + END IF; \ + RETURN NEW; \ + END; $$ LANGUAGE plpgsql", + community.as_uuid() + ))) + .execute(&db.pool) + .await + .expect("create same-transaction mention function"); + sqlx::query( + "CREATE TRIGGER trg_test_require_event_tx_mentions BEFORE INSERT ON event_mentions \ + FOR EACH ROW EXECUTE FUNCTION test_require_event_tx_mentions()", + ) + .execute(&db.pool) + .await + .expect("install same-transaction mention trigger"); + + let indexed = build("indexed"); + let result = db + .insert_event_with_serving_write_guard(&lease, &indexed, None) + .await; + + sqlx::query("DROP TRIGGER trg_test_require_event_tx_mentions ON event_mentions") + .execute(&db.pool) + .await + .expect("drop same-transaction mention trigger"); + sqlx::query("DROP FUNCTION test_require_event_tx_mentions()") + .execute(&db.pool) + .await + .expect("drop same-transaction mention function"); + + let (_, inserted) = result.expect("guarded insert"); + assert!(inserted); + assert_eq!( + counts(indexed.id.as_bytes().to_vec()).await, + (1, 1), + "mentions must be indexed in the event's own transaction" + ); + + // Make mention indexing fail for this community only. Without the + // savepoint the aborted statement poisons the event transaction and the + // commit fails. + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE FUNCTION test_reject_mentions() RETURNS trigger AS $$ \ + BEGIN \ + IF NEW.community_id = '{}'::uuid THEN \ + RAISE EXCEPTION 'test: mention indexing rejected'; \ + END IF; \ + RETURN NEW; \ + END; $$ LANGUAGE plpgsql", + community.as_uuid() + ))) + .execute(&db.pool) + .await + .expect("create mention rejection function"); + sqlx::query( + "CREATE TRIGGER trg_test_reject_mentions BEFORE INSERT ON event_mentions \ + FOR EACH ROW EXECUTE FUNCTION test_reject_mentions()", + ) + .execute(&db.pool) + .await + .expect("install mention rejection trigger"); + + let rejected = build("mention indexing fails"); + let result = db + .insert_event_with_serving_write_guard(&lease, &rejected, None) + .await; + + sqlx::query("DROP TRIGGER trg_test_reject_mentions ON event_mentions") + .execute(&db.pool) + .await + .expect("drop mention rejection trigger"); + sqlx::query("DROP FUNCTION test_reject_mentions()") + .execute(&db.pool) + .await + .expect("drop mention rejection function"); + + let (_, inserted) = result.expect("mention failure must not reject the event"); + assert!(inserted); + assert_eq!(counts(rejected.id.as_bytes().to_vec()).await, (1, 0)); + + assert!(store + .release_serving_write_lease(&lease) + .await + .expect("release serving-write lease")); +} diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 89aa1f1f51..11218114f6 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -1089,6 +1089,37 @@ pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuer Ok(cnt) } +/// SQL predicate over an `events` row for coordinates that carry no +/// historical value: NIP-RS read state and Buzz mesh heartbeats. Their +/// deletion is physical, including the mention index, so soft-deleted payloads +/// never accumulate. This is exactly the classification used by the migration +/// 0011 / 0019 purge triggers this path replaces. +macro_rules! retention_free_event_predicate { + () => { + "((kind = 30078 \ + AND d_tag ~ '^read-state:[0-9a-f]{32}$' \ + AND (SELECT count(*) \ + FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \ + WHERE jsonb_typeof(tag) = 'array' AND tag->0 = '\"d\"'::jsonb) = 1 \ + AND EXISTS (SELECT 1 \ + FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \ + WHERE jsonb_typeof(tag) = 'array' AND jsonb_array_length(tag) >= 2 \ + AND jsonb_typeof(tag->1) = 'string' AND tag->>0 = 'd' AND tag->>1 = d_tag) \ + AND (SELECT count(*) \ + FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \ + WHERE tag = '[\"t\", \"read-state\"]'::jsonb) = 1) \ + OR (kind = 30003 \ + AND d_tag LIKE 'buzz-mesh-member-status:%' \ + AND tags @> '[[\"k\", \"buzz-mesh-status\"]]'::jsonb))" + }; +} + +/// Kinds that can hold a retention-free coordinate. Cheap Rust gate so +/// ordinary deletions skip the purge statement entirely. +fn may_be_retention_free(kind: i32) -> bool { + kind == 30078 || kind == 30003 +} + /// Soft-delete the live row for an addressable coordinate /// `(kind, pubkey, d_tag)` — the NIP-33 replacement key — provided it is not /// newer than the deletion request. @@ -1134,11 +1165,34 @@ pub async fn soft_delete_by_coordinate( crate::observability::WriterOperation::EventWrite, ) .await?; - let result = sqlx::query( + let purged = if may_be_retention_free(kind) { + purge_retention_free_events( + &mut tx, + community_id, + RetentionFreeTarget::Coordinate { + kind, + pubkey, + d_tag, + created_at_or_before: deletion_created_at, + }, + ) + .await? + } else { + 0 + }; + // `AND NOT COALESCE(, false)` holds the invariant in SQL rather than through + // statement order: under READ COMMITTED this UPDATE takes a fresh snapshot, + // so a retention-free head committed by a racing replacement after the + // purge ran is spared instead of soft-deleted (the "deletion arrived + // first" outcome documented above). COALESCE keeps a NULL predicate + // (e.g. NULL `d_tag`) on the ordinary soft-delete path, as the triggers do. + let result = sqlx::query(concat!( "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 \ - AND created_at <= $5", - ) + AND created_at <= $5 AND NOT COALESCE(", + retention_free_event_predicate!(), + ", false)" + )) .bind(community_id.as_uuid()) .bind(kind) .bind(pubkey) @@ -1149,7 +1203,86 @@ pub async fn soft_delete_by_coordinate( tx.commit().await?; - Ok(result.rows_affected() > 0) + Ok(purged > 0 || result.rows_affected() > 0) +} + +/// Purge live retention-free rows matching `$target`, plus their mention rows, +/// and return the number of events removed. +macro_rules! purge_retention_free_sql { + ($target:literal) => { + concat!( + "WITH purged AS ( \ + DELETE FROM events \ + WHERE community_id = $1 AND ", + $target, + " AND deleted_at IS NULL AND ", + retention_free_event_predicate!(), + " RETURNING id \ + ), purged_mentions AS ( \ + DELETE FROM event_mentions mention USING purged \ + WHERE mention.community_id = $1 AND mention.event_id = purged.id \ + ) \ + SELECT count(*) FROM purged" + ) + }; +} + +/// Which live rows a deletion targets. +enum RetentionFreeTarget<'a> { + Id(&'a [u8]), + Coordinate { + kind: i32, + pubkey: &'a [u8], + d_tag: &'a str, + created_at_or_before: DateTime, + }, +} + +/// Physically delete live retention-free rows matched by `target`, plus their +/// mention rows, in the caller's transaction. The generic soft delete that +/// follows excludes the same predicate, so a retention-free row is either +/// purged here or left live, never soft-deleted. Returns the number of events +/// removed. +async fn purge_retention_free_events( + tx: &mut PgConnection, + community_id: CommunityId, + target: RetentionFreeTarget<'_>, +) -> Result { + // Migration 0011 fences NIP-RS hard deletes behind a transaction-local + // opt-in. The fence and this opt-in are removed together once the + // migration-only triggers are dropped. + sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)") + .execute(&mut *tx) + .await?; + + let purged: i64 = match target { + RetentionFreeTarget::Id(event_id) => { + sqlx::query_scalar(purge_retention_free_sql!("id = $2")) + .bind(community_id.as_uuid()) + .bind(event_id) + .fetch_one(&mut *tx) + .await? + } + RetentionFreeTarget::Coordinate { + kind, + pubkey, + d_tag, + created_at_or_before, + } => { + sqlx::query_scalar(purge_retention_free_sql!( + "kind = $2 AND pubkey = $3 AND d_tag = $4 AND created_at <= $5" + )) + .bind(community_id.as_uuid()) + .bind(kind) + .bind(pubkey) + .bind(d_tag) + .bind(created_at_or_before) + .fetch_one(&mut *tx) + .await? + } + }; + + Ok(purged as u64) } /// Atomically soft-delete an event and decrement thread reply counters. @@ -1236,15 +1369,24 @@ pub(crate) async fn soft_delete_event_and_update_thread_in_tx( } } - let result = sqlx::query( - "UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL", - ) + let purged = if target.is_some_and(|(kind, _)| may_be_retention_free(kind)) { + purge_retention_free_events(&mut *tx, community_id, RetentionFreeTarget::Id(event_id)) + .await? + } else { + 0 + }; + let result = sqlx::query(concat!( + "UPDATE events SET deleted_at = NOW() \ + WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND NOT COALESCE(", + retention_free_event_predicate!(), + ", false)" + )) .bind(community_id.as_uuid()) .bind(event_id) .execute(&mut *tx) .await?; - let deleted = result.rows_affected() > 0; + let deleted = purged > 0 || result.rows_affected() > 0; if deleted { if let Some(pid) = parent_event_id { diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index 34912e9f30..9c65ee21aa 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -1871,4 +1871,484 @@ mod postgres_tests { assert_ne!(leaked.as_deref(), Some("on")); } } + + /// NIP-09 deletion of a retention-free coordinate (NIP-RS read state, mesh + /// status) is physical in application code, so it holds on desired-state + /// databases that never installed the migration 0009/0019 purge triggers. + /// Ordinary addressable events, and look-alikes that fail classification, + /// keep the soft-delete contract. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn nip09_deletion_purges_retention_free_rows_and_mentions() { + assert_nip09_deletion_purges_retention_free_rows_and_mentions(false).await; + } + + /// The same contract on a migration-built database, where the 0009/0011/ + /// 0019 purge and hard-delete fence triggers are still installed. Each row + /// shape is also soft-deleted the way a pre-0009 writer would, so the + /// table proves the app classification agrees with the triggers. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn migration_schema_nip09_deletion_purges_retention_free_rows_and_mentions() { + assert_nip09_deletion_purges_retention_free_rows_and_mentions(true).await; + } + + async fn assert_nip09_deletion_purges_retention_free_rows_and_mentions(migration: bool) { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + #[derive(Clone, Copy, Debug)] + enum Delete { + Coordinate, + Id, + } + + let db = setup_db().await; + let triggers_installed: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_trigger \ + WHERE tgname = 'trg_events_purge_soft_deleted_nip_rs')", + ) + .fetch_one(&db.pool) + .await + .expect("inspect purge trigger"); + assert_eq!( + triggers_installed, migration, + "schema mode must match the test variant (run through the Postgres wrapper)" + ); + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let nip_rs = buzz_core::kind::KIND_READ_STATE as u16; + let mesh = buzz_core::kind::KIND_BOOKMARK_SET as u16; + let read_state_d = format!("read-state:{}", "b".repeat(32)); + let mesh_d = "buzz-mesh-member-status:purge-test".to_string(); + let tag = |parts: &[&str]| parts.iter().map(|p| p.to_string()).collect::>(); + let t_read_state = tag(&["t", "read-state"]); + let k_mesh = tag(&["k", "buzz-mesh-status"]); + // (label, kind, column d_tag, tags, physically purged) + type Case = (&'static str, u16, String, Vec>, bool); + let cases: Vec = vec![ + ( + "nip-rs", + nip_rs, + read_state_d.clone(), + vec![tag(&["d", &read_state_d]), t_read_state.clone()], + true, + ), + ( + "nip-rs without t tag", + nip_rs, + read_state_d.clone(), + vec![tag(&["d", &read_state_d])], + false, + ), + ( + "nip-rs uppercase hex slot", + nip_rs, + format!("read-state:{}", "B".repeat(32)), + vec![ + tag(&["d", &format!("read-state:{}", "B".repeat(32))]), + t_read_state.clone(), + ], + false, + ), + ( + "nip-rs 31-char slot", + nip_rs, + format!("read-state:{}", "b".repeat(31)), + vec![ + tag(&["d", &format!("read-state:{}", "b".repeat(31))]), + t_read_state.clone(), + ], + false, + ), + ( + "nip-rs duplicate d tag", + nip_rs, + read_state_d.clone(), + vec![ + tag(&["d", &read_state_d]), + tag(&["d", &read_state_d]), + t_read_state.clone(), + ], + false, + ), + ( + "nip-rs duplicate t tag", + nip_rs, + read_state_d.clone(), + vec![ + tag(&["d", &read_state_d]), + t_read_state.clone(), + t_read_state.clone(), + ], + false, + ), + ( + "nip-rs d tag differs from column", + nip_rs, + read_state_d.clone(), + vec![ + tag(&["d", &format!("read-state:{}", "c".repeat(32))]), + t_read_state.clone(), + ], + false, + ), + ( + "mesh-status", + mesh, + mesh_d.clone(), + vec![tag(&["d", &mesh_d]), k_mesh.clone()], + true, + ), + ( + // Containment semantics (`tags @>`), shared by trigger and app. + "mesh-status k tag with extra element", + mesh, + mesh_d.clone(), + vec![ + tag(&["d", &mesh_d]), + tag(&["k", "buzz-mesh-status", "extra"]), + ], + true, + ), + ( + "mesh-status wrong d prefix", + mesh, + "buzz-mesh-member:purge-test".into(), + vec![tag(&["d", "buzz-mesh-member:purge-test"]), k_mesh.clone()], + false, + ), + ( + "mesh-status without k tag", + mesh, + mesh_d.clone(), + vec![tag(&["d", &mesh_d])], + false, + ), + ( + "mesh-status wrong k value", + mesh, + mesh_d.clone(), + vec![tag(&["d", &mesh_d]), tag(&["k", "buzz-mesh-other"])], + false, + ), + ( + "ordinary addressable", + buzz_core::kind::KIND_PROJECT as u16, + "retained-project".into(), + vec![tag(&["d", "retained-project"])], + false, + ), + ]; + + // Store one event of this shape with one mention; return + // (keys, id, created_at secs). + let store = |label: &'static str, kind: u16, d_tag: String, tags: Vec>| { + let db = &db; + async move { + let keys = Keys::generate(); + let created_at = Timestamp::now().as_secs(); + let event = EventBuilder::new(Kind::Custom(kind), label) + .tags( + tags.iter() + .map(|tag| Tag::parse(tag.clone()).expect("tag")) + .collect::>(), + ) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(&keys) + .expect("sign event"); + assert!( + db.replace_parameterized_event(community, &event, &d_tag, None) + .await + .expect("store event") + .1, + "{label}: event must be stored" + ); + let id = event.id.as_bytes().to_vec(); + sqlx::query( + "INSERT INTO event_mentions \ + (community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) \ + SELECT community_id, $3, id, created_at, NULL, kind \ + FROM events WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(&id) + .bind("c".repeat(64)) + .execute(&db.pool) + .await + .expect("index mention"); + (keys, id, created_at) + } + }; + // (event rows, live event rows, mention rows) for one id. + let state = |id: Vec| { + let pool = db.pool.clone(); + async move { + let (rows, live): (i64, i64) = sqlx::query_as( + "SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) \ + FROM events WHERE community_id=$1 AND id=$2", + ) + .bind(community.as_uuid()) + .bind(&id) + .fetch_one(&pool) + .await + .expect("count event rows"); + let mentions: i64 = sqlx::query_scalar( + "SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2", + ) + .bind(community.as_uuid()) + .bind(&id) + .fetch_one(&pool) + .await + .expect("count mentions"); + (rows, live, mentions) + } + }; + let expected = |purged: bool| -> (i64, i64, i64) { + if purged { + (0, 0, 0) + } else { + (1, 0, 1) + } + }; + + for (label, kind, d_tag, tags, purged) in cases { + if migration { + // Trigger half of the parity table: a pre-0009 writer's plain + // soft delete, classified only by the 0009/0011/0019 triggers. + let (_, id, _) = store(label, kind, d_tag.clone(), tags.clone()).await; + sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(&id) + .execute(&db.pool) + .await + .expect("legacy soft delete"); + assert_eq!( + state(id).await, + expected(purged), + "{label}: trigger classification" + ); + } + + for mode in [Delete::Coordinate, Delete::Id] { + let (keys, id, created_at) = store(label, kind, d_tag.clone(), tags.clone()).await; + let delete = || async { + match mode { + Delete::Coordinate => event::soft_delete_by_coordinate( + &db.pool, + community, + i32::from(kind), + &keys.public_key().to_bytes(), + &d_tag, + (created_at + 1) as i64, + ) + .await + .expect("coordinate delete"), + Delete::Id => event::soft_delete_event_and_update_thread( + &db.pool, community, &id, None, None, + ) + .await + .expect("id delete"), + } + }; + assert!( + delete().await, + "{label}/{mode:?}: deletion must report success" + ); + assert_eq!( + state(id.clone()).await, + expected(purged), + "{label}/{mode:?}: app classification" + ); + assert!( + !delete().await, + "{label}/{mode:?}: repeated deletion is a no-op" + ); + } + } + + // A mesh-status-shaped row whose `d_tag` column was never backfilled: + // `kind = 30003 AND NULL LIKE … AND tags @> …` makes the purge + // predicate NULL (a read-state row cannot: its tag/column EXISTS is + // false), which the triggers treat as false. The id path must still + // soft-delete it (the `COALESCE` in the generic soft delete), not + // silently skip it. Coordinate deletes key on `d_tag`, so only the id + // path can reach this row. + let null_d_tag_row = || async { + let (_, id, _) = store( + "mesh-status with NULL d_tag column", + mesh, + mesh_d.clone(), + vec![tag(&["d", &mesh_d]), k_mesh.clone()], + ) + .await; + sqlx::query("UPDATE events SET d_tag = NULL WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(&id) + .execute(&db.pool) + .await + .expect("clear d_tag column"); + id + }; + if migration { + let id = null_d_tag_row().await; + sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(&id) + .execute(&db.pool) + .await + .expect("legacy soft delete"); + assert_eq!( + state(id).await, + expected(false), + "NULL d_tag: trigger classification" + ); + } + let id = null_d_tag_row().await; + assert!( + event::soft_delete_event_and_update_thread(&db.pool, community, &id, None, None) + .await + .expect("id delete"), + "NULL d_tag/Id: deletion must report success" + ); + assert_eq!( + state(id).await, + expected(false), + "NULL d_tag/Id: app classification" + ); + } + + /// A coordinate deletion racing a NIP-RS replacement must never leave a + /// soft-deleted read-state row behind. The purge waits on the old head's + /// row lock; the replacement commits a newer head that the purge's snapshot + /// cannot see; the generic soft delete then runs on a fresh snapshot and + /// must spare that head rather than soft-delete it. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn coordinate_deletion_racing_nip_rs_replacement_never_soft_deletes_read_state() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + use sqlx::postgres::PgConnectOptions; + use std::str::FromStr; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let d_tag = format!("read-state:{}", "d".repeat(32)); + let created_at = Timestamp::now().as_secs(); + let old_head = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), + "old head", + ) + .tags([ + Tag::parse(["d", d_tag.as_str()]).expect("d tag"), + Tag::parse(["t", "read-state"]).expect("t tag"), + ]) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(&keys) + .expect("sign old head"); + assert!( + db.replace_parameterized_event(community, &old_head, &d_tag, None) + .await + .expect("store old head") + .1 + ); + let old_id = old_head.id.as_bytes().to_vec(); + let new_id = vec![0x5a_u8; 32]; + + // The replacement: insert a newer head (still at or before the + // tombstone), hard-delete the old head, and hold the transaction open. + let mut replacement = db.pool.begin().await.expect("begin replacement"); + sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)") + .execute(&mut *replacement) + .await + .expect("opt in to hard delete"); + sqlx::query( + "INSERT INTO events \ + (community_id, id, pubkey, created_at, kind, tags, content, sig, channel_id, d_tag) \ + SELECT community_id, $3, pubkey, created_at + interval '1 second', kind, tags, \ + 'new head', sig, channel_id, d_tag \ + FROM events WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(&old_id) + .bind(&new_id) + .execute(&mut *replacement) + .await + .expect("insert new head"); + sqlx::query("DELETE FROM events WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(&old_id) + .execute(&mut *replacement) + .await + .expect("delete old head"); + + let database_url = + std::env::var("TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.into()); + let application_name = format!("nip-rs-race-{}", Uuid::new_v4().simple()); + let deleter_pool = PgPoolOptions::new() + .max_connections(1) + .connect_with( + PgConnectOptions::from_str(&database_url) + .expect("parse test database URL") + .application_name(&application_name), + ) + .await + .expect("connect deleter pool"); + let pubkey = keys.public_key().to_bytes(); + let deletion = tokio::spawn({ + let d_tag = d_tag.clone(); + async move { + event::soft_delete_by_coordinate( + &deleter_pool, + community, + buzz_core::kind::KIND_READ_STATE as i32, + &pubkey, + &d_tag, + (created_at + 5) as i64, + ) + .await + } + }); + + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_stat_activity \ + WHERE datname = current_database() AND application_name = $1 \ + AND wait_event_type = 'Lock')", + ) + .bind(&application_name) + .fetch_one(&db.pool) + .await + .expect("inspect deleter lock"); + if waiting { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("deletion reached the old head's row lock"); + + replacement.commit().await.expect("commit replacement"); + deletion + .await + .expect("join deletion") + .expect("coordinate delete"); + + let rows: Vec<(Vec, bool)> = sqlx::query_as( + "SELECT id, deleted_at IS NOT NULL FROM events \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4", + ) + .bind(community.as_uuid()) + .bind(buzz_core::kind::KIND_READ_STATE as i32) + .bind(keys.public_key().to_bytes().as_slice()) + .bind(&d_tag) + .fetch_all(&db.pool) + .await + .expect("read coordinate rows"); + assert_eq!( + rows, + vec![(new_id, false)], + "the racing head is spared and no soft-deleted read-state row remains" + ); + } }