diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 20496ac1cf..01bb5f5761 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -703,12 +703,14 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 49); + assert_eq!(migrations.len(), 50); assert_eq!(migrations[48].version, 49); assert!(migrations[48] .sql .as_str() .contains("idx_thread_metadata_window")); + assert_eq!(migrations[49].version, 50); + assert!(migrations[49].sql.as_str().contains("original_slot")); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 411c518240..9e195b203a 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -1091,6 +1091,99 @@ pub(crate) async fn soft_delete_event_and_update_thread_in_tx( Ok(deleted) } +/// Public structural position of an event: where it sat in its channel and +/// thread. Captured when the event is deleted so a relay-signed deletion notice +/// can stand in the original's slot. Carries no content, author, or tags. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct OriginalSlot { + /// Original event `created_at` (unix seconds). + pub created_at: i64, + /// Direct parent event ID (hex); `None` for a top-level message. + pub parent_event_id: Option, + /// Outer thread root event ID (hex); `None` for a top-level message. + pub root_event_id: Option, + /// Nesting depth (top-level = 0). + pub depth: i32, + /// Whether the reply was also broadcast to the channel timeline. + pub broadcast: bool, +} + +/// Read `event_id`'s [`OriginalSlot`] inside the caller's transaction. +/// +/// Soft-deleted rows still have a slot. `None` means the event row is gone +/// (purged). A row without `thread_metadata` is top-level: replies always get +/// a metadata row at ingest. +pub(crate) async fn capture_original_slot_in_tx( + tx: &mut PgConnection, + community_id: CommunityId, + event_id: &[u8], +) -> Result> { + let row = sqlx::query( + "SELECT e.created_at, tm.parent_event_id, tm.root_event_id, \ + COALESCE(tm.depth, 0) AS depth, COALESCE(tm.broadcast, false) AS broadcast \ + FROM events e \ + LEFT JOIN thread_metadata tm \ + ON tm.community_id = e.community_id AND tm.event_id = e.id \ + WHERE e.community_id = $1 AND e.id = $2 \ + LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(event_id) + .fetch_optional(&mut *tx) + .await?; + let Some(row) = row else { + return Ok(None); + }; + let created_at: DateTime = row.try_get("created_at")?; + let parent: Option> = row.try_get("parent_event_id")?; + let root: Option> = row.try_get("root_event_id")?; + Ok(Some(OriginalSlot { + created_at: created_at.timestamp(), + parent_event_id: parent.map(hex::encode), + root_event_id: root.map(hex::encode), + depth: row.try_get("depth")?, + broadcast: row.try_get("broadcast")?, + })) +} + +/// Soft-delete `event_id` (with thread counters) and persist `notice` in one +/// transaction. +/// +/// Returns `false` without writing anything when the event was already deleted +/// or absent. A notice insert failure rolls the deletion back, so a retry +/// performs the whole mutation again instead of losing the notice. +pub async fn soft_delete_event_with_notice( + pool: &PgPool, + community_id: CommunityId, + event_id: &[u8], + parent_event_id: Option<&[u8]>, + root_event_id: Option<&[u8]>, + notice: &Event, + notice_channel_id: Uuid, +) -> Result { + let connection = crate::observability::acquire_writer( + pool, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + let mut tx = sqlx::Transaction::begin(connection, None).await?; + let deleted = soft_delete_event_and_update_thread_in_tx( + &mut tx, + community_id, + event_id, + parent_event_id, + root_event_id, + ) + .await?; + if !deleted { + tx.rollback().await?; + return Ok(false); + } + insert_event_in_transaction(&mut tx, community_id, notice, Some(notice_channel_id)).await?; + tx.commit().await?; + Ok(true) +} + /// Returns the `created_at` timestamp of the most recent non-deleted event in a channel. pub async fn get_last_message_at( pool: &PgPool, @@ -2151,6 +2244,30 @@ impl Db { .await } + /// Atomically soft-delete an event, decrement thread counters, and persist + /// its deletion notice. See [`soft_delete_event_with_notice`]. + #[datastore_span(name = "soft_delete_event_with_notice", system = "postgresql")] + pub async fn soft_delete_event_with_notice( + &self, + community_id: CommunityId, + event_id: &[u8], + parent_event_id: Option<&[u8]>, + root_event_id: Option<&[u8]>, + notice: &nostr::Event, + notice_channel_id: Uuid, + ) -> Result { + crate::event::soft_delete_event_with_notice( + &self.pool, + community_id, + event_id, + parent_event_id, + root_event_id, + notice, + notice_channel_id, + ) + .await + } + /// Returns the most recent `created_at` for a channel. #[datastore_span(name = "get_last_message_at", system = "postgresql")] pub async fn get_last_message_at( @@ -2813,6 +2930,71 @@ mod postgres_tests { ); } + /// A failed notice insert rolls the deletion back; the retry deletes once + /// and stores the notice; a later attempt is a no-op. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn soft_delete_with_notice_is_atomic_with_the_notice() { + let pool = setup_pool().await; + let community_uuid = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_uuid); + let channel = make_test_channel(&pool, community_uuid, None).await; + let target = make_text_event("to be deleted"); + insert_event(&pool, community, &target, Some(channel)) + .await + .expect("insert target"); + let keys = Keys::generate(); + let deleted_at = |pool: PgPool, id: Vec| async move { + sqlx::query_scalar::<_, Option>>( + "SELECT deleted_at FROM events WHERE community_id = $1 AND id = $2", + ) + .bind(community_uuid) + .bind(id) + .fetch_one(&pool) + .await + .expect("target row") + }; + + // KIND_AUTH is rejected by the insert path: a deterministic notice failure. + let rejected = EventBuilder::new(Kind::Custom(KIND_AUTH as u16), "") + .sign_with_keys(&keys) + .expect("sign"); + let target_id = target.id.as_bytes().to_vec(); + assert!(soft_delete_event_with_notice( + &pool, community, &target_id, None, None, &rejected, channel + ) + .await + .is_err()); + assert!( + deleted_at(pool.clone(), target_id.clone()).await.is_none(), + "failed notice must roll back the deletion" + ); + + let notice = EventBuilder::new(Kind::Custom(40099), "{}") + .tags(vec![Tag::parse(["h", &channel.to_string()]).expect("h")]) + .sign_with_keys(&keys) + .expect("sign"); + for expected in [true, false] { + assert_eq!( + soft_delete_event_with_notice( + &pool, community, &target_id, None, None, ¬ice, channel + ) + .await + .expect("delete with notice"), + expected + ); + } + assert!(deleted_at(pool.clone(), target_id).await.is_some()); + let notices: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE community_id = $1 AND id = $2") + .bind(community_uuid) + .bind(notice.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("notice count"); + assert_eq!(notices, 1); + } + fn make_text_event(content: &str) -> nostr::Event { let keys = Keys::generate(); EventBuilder::new(Kind::Custom(9), content) diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index cc85f30db3..e454cbf4eb 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -16,6 +16,7 @@ use sqlx::{PgPool, Row as _}; use uuid::Uuid; use crate::error::Result; +use crate::event::OriginalSlot; use crate::CommunityId; /// A row in `relay_admin_actions`. @@ -53,12 +54,25 @@ pub struct AdminActionRecord { /// Authoritative channel persisted at claim time (kick actions). /// `None` for community-wide actions. pub enforcement_channel_id: Option, + /// Delete target's original slot, frozen at mutation time (or at + /// finalization for legacy rows). `None` = not captured yet. + pub original_slot: Option, /// Row creation time. pub created_at: DateTime, /// Row last-updated time. pub updated_at: DateTime, } +/// A delete target's [`OriginalSlot`] as frozen on the action row. +/// +/// `slot: None` records that capture ran but the target row was already gone, +/// so a retry can never substitute different data. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct FrozenSlot { + /// The captured slot, or `None` when the target no longer existed. + pub slot: Option, +} + /// A row in `relay_admin_outbox`. #[derive(Debug, Clone)] pub struct OutboxRecord { @@ -247,7 +261,7 @@ pub async fn claim_report( r#" SELECT id, report_id, report_community_id, request_id, actor_pubkey, actor_role, action, reason, timeout_until, state, step_marker, cancelled_by, error_message, - enforcement_target_pubkey, enforcement_channel_id, + enforcement_target_pubkey, enforcement_channel_id, original_slot, created_at, updated_at FROM relay_admin_actions WHERE report_community_id = $1 AND report_id = $2 AND request_id = $3 @@ -280,7 +294,7 @@ pub async fn claim_report( ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'pending', $9, $10) RETURNING id, report_id, report_community_id, request_id, actor_pubkey, actor_role, action, reason, timeout_until, state, step_marker, cancelled_by, error_message, - enforcement_target_pubkey, enforcement_channel_id, + enforcement_target_pubkey, enforcement_channel_id, original_slot, created_at, updated_at "#, ) @@ -726,6 +740,9 @@ pub async fn execute_kick_with_marker( /// /// The delete is idempotent: if the event is already deleted the marker is still /// committed (soft-delete is already-done = success). +/// +/// The target's [`OriginalSlot`] is frozen on the action row with the marker, so +/// the deletion notice is rebuilt from the same data on every delivery retry. pub async fn execute_delete_with_marker( pool: &PgPool, action_id: Uuid, @@ -764,6 +781,11 @@ pub async fn execute_delete_with_marker( .fetch_optional(&mut *tx) .await?; + let slot = FrozenSlot { + slot: crate::event::capture_original_slot_in_tx(&mut tx, community_id, target_event_id) + .await?, + }; + // Canonical delete + thread_metadata counters, fenced by this transaction. // Counters move only when the row transitions to deleted, so a second // action against an already-deleted target leaves them unchanged. @@ -779,7 +801,7 @@ pub async fn execute_delete_with_marker( let marker = sqlx::query( r#" UPDATE relay_admin_actions - SET step_marker = 'mutation_committed', updated_at = now() + SET step_marker = 'mutation_committed', original_slot = $3, updated_at = now() WHERE id = $1 AND action_lease_token = $2 AND action_lease_expires_at > clock_timestamp() @@ -789,6 +811,7 @@ pub async fn execute_delete_with_marker( ) .bind(action_id) .bind(lease_token) + .bind(sqlx::types::Json(&slot)) .execute(&mut *tx) .await?; @@ -851,21 +874,22 @@ pub async fn finalize_success( let mut tx = pool.begin().await?; // Require step_marker = 'mutation_committed' to prevent premature finalization. - let updated_action = sqlx::query( + let frozen_slot: Option>> = sqlx::query_scalar( r#" UPDATE relay_admin_actions SET state = 'succeeded', step_marker = 'artifacts_done', updated_at = now() WHERE id = $1 AND state = 'enforcing' AND step_marker = 'mutation_committed' + RETURNING original_slot "#, ) .bind(action_id) - .execute(&mut *tx) + .fetch_optional(&mut *tx) .await?; - if updated_action.rows_affected() == 0 { + let Some(frozen_slot) = frozen_slot else { tx.rollback().await?; return Ok(false); - } + }; // Transition report to terminal status. Requires active_action_id = this action, // which prevents a stale or wrong action from closing the report. @@ -902,6 +926,27 @@ pub async fn finalize_success( if action_name == "delete" { if let (Some(target_eid), Some(ch)) = (target_event_id, channel_id) { + // Actions committed before migration 0050 have no frozen slot: + // freeze it once here so every delivery attempt agrees. + let slot = match frozen_slot { + Some(json) => json.0, + None => { + let slot = FrozenSlot { + slot: crate::event::capture_original_slot_in_tx( + &mut tx, + community_id, + target_eid, + ) + .await?, + }; + sqlx::query("UPDATE relay_admin_actions SET original_slot = $2 WHERE id = $1") + .bind(action_id) + .bind(sqlx::types::Json(&slot)) + .execute(&mut *tx) + .await?; + slot + } + }; let payload = serde_json::json!({ "community_id": community_str, "channel_id": ch.to_string(), @@ -909,6 +954,7 @@ pub async fn finalize_success( "action_id": action_str, "actor": hex::encode(actor_pubkey), "reason_code": reason.unwrap_or(""), + "original": slot.slot, }); sqlx::query( r#" @@ -1300,7 +1346,7 @@ pub async fn get_action(pool: &PgPool, action_id: Uuid) -> Result Result { error_message: row.try_get("error_message")?, enforcement_target_pubkey: row.try_get("enforcement_target_pubkey")?, enforcement_channel_id: row.try_get("enforcement_channel_id")?, + original_slot: row + .try_get::>, _>("original_slot")? + .map(|json| json.0), created_at: row.try_get("created_at")?, updated_at: row.try_get("updated_at")?, }) @@ -3139,6 +3188,185 @@ mod postgres_tests { assert_eq!(counts(&pool, cid, &root).await, (1, 2)); } + /// Finalize a committed delete of `target` and return the tombstone payload. + async fn finalize_delete_payload( + pool: &PgPool, + community_id: Uuid, + action_id: Uuid, + target: &[u8], + channel: Uuid, + ) -> serde_json::Value { + let report_id: Uuid = + sqlx::query_scalar("SELECT report_id FROM relay_admin_actions WHERE id = $1") + .bind(action_id) + .fetch_one(pool) + .await + .expect("report id"); + assert!(finalize_success( + pool, + action_id, + CommunityId::from_uuid(community_id), + report_id, + "resolved", + &actor(), + "delete", + None, + Some(target), + Some(channel), + Some("spam"), + None, + ) + .await + .expect("finalize")); + sqlx::query_scalar( + "SELECT payload FROM relay_admin_outbox WHERE action_id = $1 AND task_type = 'tombstone'", + ) + .bind(action_id) + .fetch_one(pool) + .await + .expect("tombstone payload") + } + + async fn event_channel(pool: &PgPool, community_id: Uuid, id: &[u8]) -> Uuid { + sqlx::query_scalar("SELECT channel_id FROM events WHERE community_id = $1 AND id = $2") + .bind(community_id) + .bind(id) + .fetch_one(pool) + .await + .expect("event channel") + } + + async fn purge_event(pool: &PgPool, community_id: Uuid, id: &[u8]) { + for sql in [ + "DELETE FROM thread_metadata WHERE community_id = $1 AND event_id = $2", + "DELETE FROM events WHERE community_id = $1 AND id = $2", + ] { + sqlx::query(sql) + .bind(community_id) + .bind(id) + .execute(pool) + .await + .expect("purge"); + } + } + + /// The slot is frozen with the mutation marker, so a target purged before + /// finalization still yields the original slot in the tombstone payload. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn admin_delete_freezes_original_slot_before_purge() { + let pool = setup_pool().await; + let community_id = make_community(&pool).await; + let cid = CommunityId::from_uuid(community_id); + let [root, reply, nested] = make_thread(&pool, cid).await; + let channel = event_channel(&pool, community_id, &nested).await; + let created_at: DateTime = + sqlx::query_scalar("SELECT created_at FROM events WHERE community_id = $1 AND id = $2") + .bind(community_id) + .bind(&nested) + .fetch_one(&pool) + .await + .expect("created_at"); + + let lease_until = Utc::now() + chrono::Duration::seconds(60); + let (action_id, token) = enforcing_action(&pool, community_id, lease_until).await; + assert!(execute_delete_with_marker( + &pool, + action_id, + token, + cid, + &nested, + Some(&reply), + Some(&root), + ) + .await + .expect("delete")); + + let expected = OriginalSlot { + created_at: created_at.timestamp(), + parent_event_id: Some(hex::encode(&reply)), + root_event_id: Some(hex::encode(&root)), + depth: 2, + broadcast: false, + }; + let frozen = get_action(&pool, action_id) + .await + .expect("get_action") + .expect("action") + .original_slot; + assert_eq!( + frozen, + Some(FrozenSlot { + slot: Some(expected.clone()) + }) + ); + + purge_event(&pool, community_id, &nested).await; + let payload = + finalize_delete_payload(&pool, community_id, action_id, &nested, channel).await; + assert_eq!( + payload["original"], + serde_json::to_value(&expected).expect("slot json") + ); + } + + /// Actions committed before the slot column existed freeze it once at + /// finalization; a target already purged by then freezes "unavailable". + #[tokio::test] + #[ignore = "requires Postgres"] + async fn legacy_admin_delete_freezes_slot_at_finalization() { + let pool = setup_pool().await; + let community_id = make_community(&pool).await; + let cid = CommunityId::from_uuid(community_id); + let lease_until = Utc::now() + chrono::Duration::seconds(60); + + for purged in [false, true] { + let [root, reply, _nested] = make_thread(&pool, cid).await; + let channel = event_channel(&pool, community_id, &reply).await; + let (action_id, token) = enforcing_action(&pool, community_id, lease_until).await; + assert!(execute_delete_with_marker( + &pool, + action_id, + token, + cid, + &reply, + Some(&root), + Some(&root), + ) + .await + .expect("delete")); + sqlx::query("UPDATE relay_admin_actions SET original_slot = NULL WHERE id = $1") + .bind(action_id) + .execute(&pool) + .await + .expect("simulate pre-0050 row"); + if purged { + purge_event(&pool, community_id, &reply).await; + } + + let payload = + finalize_delete_payload(&pool, community_id, action_id, &reply, channel).await; + let frozen = get_action(&pool, action_id) + .await + .expect("get_action") + .expect("action") + .original_slot + .expect("frozen at finalization"); + assert_eq!( + payload["original"], + serde_json::to_value(&frozen.slot).expect("slot json"), + "payload and row agree" + ); + if purged { + assert_eq!(frozen.slot, None, "purged target freezes unavailable"); + } else { + let slot = frozen.slot.expect("slot captured"); + assert_eq!(slot.parent_event_id, Some(hex::encode(&root))); + assert_eq!(slot.depth, 1); + } + } + } + /// Regression: lease live at transaction entry, expires while the event write blocks. /// The target event row is locked externally; the worker blocks at the `UPDATE events` /// domain write (after it has already acquired the action-row lock). The test confirms diff --git a/crates/buzz-relay/src/api/admin/mod.rs b/crates/buzz-relay/src/api/admin/mod.rs index e59591f771..da96b5618a 100644 --- a/crates/buzz-relay/src/api/admin/mod.rs +++ b/crates/buzz-relay/src/api/admin/mod.rs @@ -6463,6 +6463,26 @@ mod postgres_tests { Some("e2e test"), "tombstone public_reason mirrors the operator reason" ); + // The shared builder carries the top-level target's frozen slot. + let target_created_at: chrono::DateTime = + sqlx::query_scalar("SELECT created_at FROM events WHERE community_id = $1 AND id = $2") + .bind(community_id) + .bind(target_event_id.as_slice()) + .fetch_one(&pool) + .await + .expect("target created_at"); + assert_eq!( + parsed["original"], + serde_json::json!({ + "version": 1, + "created_at": target_created_at.timestamp(), + "parent_event_id": null, + "root_event_id": null, + "depth": 0, + "broadcast": false, + }), + "admin tombstone must carry the original slot" + ); // Assert: tombstone outbox row is now delivered. let tombstone_state: String = diff --git a/crates/buzz-relay/src/handlers/admin_outbox_worker.rs b/crates/buzz-relay/src/handlers/admin_outbox_worker.rs index a71197301e..45ab99719d 100644 --- a/crates/buzz-relay/src/handlers/admin_outbox_worker.rs +++ b/crates/buzz-relay/src/handlers/admin_outbox_worker.rs @@ -21,6 +21,7 @@ use uuid::Uuid; use buzz_db::relay_admin_actions::OutboxRecord; +use crate::handlers::deletion_tombstone::{build_delete_tombstone, DeleteTombstone}; use crate::state::AppState; /// Lease duration: if the worker pod dies mid-delivery, another pod picks up @@ -154,13 +155,15 @@ async fn resolve_tenant( /// Deliver a tombstone: publish an admin-deletion system message in the channel. /// -/// The emitted system message matches the channel-moderation tombstone schema -/// (`side_effects.rs` NIP-29 DELETE_EVENT: `type: "message_deleted"` with -/// `actor`, `target_event_id`, and an optional public reason) so the room -/// renders it identically. Without `target_event_id` the room cannot tell which -/// message was removed, and without a reason it renders as a bare self-delete -/// rather than a moderator removal — `SystemMessageRow` keys "Removed by -/// community moderators" on `public_reason`. +/// Built by the shared `deletion_tombstone` builder — the same contract as the +/// NIP-29 DELETE_EVENT notice (`type: "message_deleted"` with `actor`, +/// `target_event_id`, an optional public reason, and the original slot) — so +/// the room renders both identically. Rows enqueued before the slot existed +/// have no `original` and deliver the legacy notice. Without `target_event_id` +/// the room cannot tell which message was removed, and without a reason it +/// renders as a bare self-delete rather than a moderator removal — +/// `SystemMessageRow` keys "Removed by community moderators" on +/// `public_reason`. /// /// The `reason_code`/`public_reason` fields are the operator's `reason` string /// verbatim (an operator-authored public reason, not a sanitized derivative); @@ -187,29 +190,33 @@ async fn deliver_tombstone(state: &Arc, row: &OutboxRecord) -> Result< let community_id = buzz_core::CommunityId::from_uuid(community_uuid); let tenant = resolve_tenant(state, community_id, "tombstone").await?; - // Match the established channel-moderation `message_deleted` schema - // (`side_effects.rs`): `type`, `actor` (the acting operator's pubkey hex), - // and `target_event_id`, plus the admin `action_id`. `reason_code` is the - // operator's `reason` string (see `finalize_success`) — forwarded as the - // room-facing public reason; the room renders the moderator-removal template - // only when a non-empty reason is present. - let mut content = serde_json::json!({ - "type": "message_deleted", - "actor": actor, - "target_event_id": target_event_id, - "action_id": row.action_id.to_string(), - }); - if let Some(reason_code) = payload["reason_code"].as_str().filter(|r| !r.is_empty()) { - content["reason_code"] = serde_json::Value::String(reason_code.to_string()); - content["public_reason"] = serde_json::Value::String(reason_code.to_string()); - } - - crate::handlers::side_effects::emit_system_message( - &tenant, - state, - channel_id, - content, + // Rebuilt from the payload frozen at finalization — never a fresh target + // lookup — so every retry signs identical bytes (same event ID) even after + // the target is purged. `reason_code` is the operator's `reason` string + // (see `finalize_success`), forwarded as the room-facing public reason; the + // room renders the moderator-removal template only when it is non-empty. + let original: Option = + serde_json::from_value(payload["original"].clone()) + .map_err(|e| format!("tombstone: invalid original slot: {e}"))?; + let action_id = row.action_id.to_string(); + let reason = payload["reason_code"].as_str().filter(|r| !r.is_empty()); + let notice = build_delete_tombstone( + &DeleteTombstone { + channel_id, + target_event_id, + actor, + action_id: Some(&action_id), + reason_code: reason, + public_reason: reason, + original: original.as_ref(), + }, row.created_at, + &state.relay_keypair, + ) + .map_err(|e| format!("tombstone: {e}"))?; + + crate::handlers::side_effects::persist_and_publish_system_event( + &tenant, state, channel_id, ¬ice, ) .await .map_err(|e| format!("tombstone: system message failed: {e}")) diff --git a/crates/buzz-relay/src/handlers/deletion_tombstone.rs b/crates/buzz-relay/src/handlers/deletion_tombstone.rs new file mode 100644 index 0000000000..aa4684d323 --- /dev/null +++ b/crates/buzz-relay/src/handlers/deletion_tombstone.rs @@ -0,0 +1,249 @@ +//! The relay-signed `message_deleted` deletion notice (kind 40099). +//! +//! Both deletion emitters — NIP-29 `DELETE_EVENT` (kind 9005) and the admin +//! outbox tombstone — build their notice here so the wire contract has one +//! implementation. +//! +//! The notice's `created_at` is always the deletion-notice time, never the +//! original's time: subscribers filter live fan-out by `since`, so a backdated +//! notice would never reach them. The original's position travels in the +//! optional, versioned `original` object instead. Tags are exactly one `h` +//! plus unmarked `e` references (target, root, parent; deduplicated) for +//! indexed lookup — no `reply`/`root` markers, so no client treats the notice +//! as a conversational reply, and no `p` tags, so nobody is notified. + +use buzz_core::kind::KIND_SYSTEM_MESSAGE; +use buzz_db::event::OriginalSlot; +use buzz_db::thread::ThreadMetadataRecord; +use chrono::{DateTime, Utc}; +use nostr::{Event, EventBuilder, Keys, Kind, Tag, Timestamp}; +use uuid::Uuid; + +/// Version of the `original` slot object emitted by this relay. +pub const ORIGINAL_SLOT_VERSION: u32 = 1; + +/// Inputs for one deletion notice. +pub struct DeleteTombstone<'a> { + /// Channel the deleted event belonged to (already authorized by the caller). + pub channel_id: Uuid, + /// Deleted event ID (64-char hex). + pub target_event_id: &'a str, + /// Acting principal pubkey (hex). + pub actor: &'a str, + /// Admin action ID, when the deletion came from moderation. + pub action_id: Option<&'a str>, + /// Moderation reason code. + pub reason_code: Option<&'a str>, + /// Room-facing moderation reason. + pub public_reason: Option<&'a str>, + /// The deleted event's slot, when known. An invalid slot is dropped and the + /// notice degrades to the legacy form rather than inventing structure. + pub original: Option<&'a OriginalSlot>, +} + +/// Sign the deletion notice at `created_at` with the relay keys. +pub fn build_delete_tombstone( + tombstone: &DeleteTombstone<'_>, + created_at: DateTime, + relay_keys: &Keys, +) -> anyhow::Result { + let original = tombstone + .original + .filter(|slot| valid_original_slot(slot, tombstone.target_event_id)); + + let mut content = serde_json::json!({ + "type": "message_deleted", + "actor": tombstone.actor, + "target_event_id": tombstone.target_event_id, + }); + for (field, value) in [ + ("action_id", tombstone.action_id), + ("reason_code", tombstone.reason_code), + ("public_reason", tombstone.public_reason), + ] { + if let Some(value) = value { + content[field] = value.into(); + } + } + if let Some(slot) = original { + content["original"] = serde_json::json!({ + "version": ORIGINAL_SLOT_VERSION, + "created_at": slot.created_at, + "parent_event_id": slot.parent_event_id, + "root_event_id": slot.root_event_id, + "depth": slot.depth, + "broadcast": slot.broadcast, + }); + } + + let mut references = vec![tombstone.target_event_id]; + if let Some(slot) = original { + for id in [&slot.root_event_id, &slot.parent_event_id] + .into_iter() + .flatten() + { + if !references.contains(&id.as_str()) { + references.push(id); + } + } + } + let mut tags = vec![Tag::parse(["h", &tombstone.channel_id.to_string()])?]; + for id in references { + tags.push(Tag::parse(["e", id])?); + } + + EventBuilder::new( + Kind::Custom(KIND_SYSTEM_MESSAGE as u16), + content.to_string(), + ) + .tags(tags) + .custom_created_at(Timestamp::from(created_at.timestamp().max(0) as u64)) + .sign_with_keys(relay_keys) + .map_err(|e| anyhow::anyhow!("failed to sign deletion notice: {e}")) +} + +/// The slot of a still-stored event: its `created_at` plus canonical thread +/// metadata. No metadata row means a top-level message. +pub fn original_slot(target_created_at: i64, meta: Option<&ThreadMetadataRecord>) -> OriginalSlot { + OriginalSlot { + created_at: target_created_at, + parent_event_id: meta.and_then(|m| m.parent_event_id.as_deref().map(hex::encode)), + root_event_id: meta.and_then(|m| m.root_event_id.as_deref().map(hex::encode)), + depth: meta.map_or(0, |m| m.depth), + broadcast: meta.is_some_and(|m| m.broadcast), + } +} + +/// All-or-nothing v1 shape check: top-level has neither ID at depth 0; a direct +/// reply has parent == root at depth 1; a nested reply has distinct parent and +/// root at depth > 1. IDs are full hex and never the target itself. +fn valid_original_slot(slot: &OriginalSlot, target_event_id: &str) -> bool { + let is_id = |id: &str| { + id.len() == 64 + && id + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + }; + let shape = match ( + slot.parent_event_id.as_deref(), + slot.root_event_id.as_deref(), + ) { + (None, None) => slot.depth == 0, + (Some(parent), Some(root)) => { + is_id(parent) + && is_id(root) + && parent != target_event_id + && root != target_event_id + && match slot.depth { + 1 => parent == root, + d => d > 1 && parent != root, + } + } + _ => false, + }; + shape && slot.created_at >= 0 +} + +#[cfg(test)] +mod tests { + use super::*; + + const TARGET: &str = "1111111111111111111111111111111111111111111111111111111111111111"; + const ROOT: &str = "2222222222222222222222222222222222222222222222222222222222222222"; + const PARENT: &str = "3333333333333333333333333333333333333333333333333333333333333333"; + + fn slot(parent: Option<&str>, root: Option<&str>, depth: i32) -> OriginalSlot { + OriginalSlot { + created_at: 100, + parent_event_id: parent.map(str::to_owned), + root_event_id: root.map(str::to_owned), + depth, + broadcast: false, + } + } + + fn notice(original: Option<&OriginalSlot>) -> Event { + build_delete_tombstone( + &DeleteTombstone { + channel_id: Uuid::nil(), + target_event_id: TARGET, + actor: "actor", + action_id: None, + reason_code: None, + public_reason: None, + original, + }, + DateTime::from_timestamp(300, 0).expect("ts"), + &Keys::generate(), + ) + .expect("build") + } + + fn tags(event: &Event) -> Vec> { + event.tags.iter().map(|t| t.as_slice().to_vec()).collect() + } + + fn content(event: &Event) -> serde_json::Value { + serde_json::from_str(&event.content).expect("json") + } + + #[test] + fn nested_reply_notice_carries_slot_and_unmarked_references() { + let event = notice(Some(&slot(Some(PARENT), Some(ROOT), 2))); + assert_eq!(event.created_at.as_secs(), 300, "notice time, not original"); + assert_eq!( + tags(&event), + vec![ + vec!["h".to_owned(), Uuid::nil().to_string()], + vec!["e".to_owned(), TARGET.to_owned()], + vec!["e".to_owned(), ROOT.to_owned()], + vec!["e".to_owned(), PARENT.to_owned()], + ] + ); + assert_eq!( + content(&event)["original"], + serde_json::json!({ + "version": 1, "created_at": 100, "parent_event_id": PARENT, + "root_event_id": ROOT, "depth": 2, "broadcast": false, + }) + ); + } + + #[test] + fn direct_reply_references_are_deduplicated() { + let event = notice(Some(&slot(Some(ROOT), Some(ROOT), 1))); + assert_eq!(tags(&event).len(), 3, "h + target + root once"); + } + + #[test] + fn malformed_slots_degrade_to_legacy_notice() { + for bad in [ + slot(Some(ROOT), None, 1), + slot(None, None, 1), + slot(Some(PARENT), Some(ROOT), 1), + slot(Some(ROOT), Some(ROOT), 2), + slot(Some(TARGET), Some(TARGET), 1), + slot(Some("ab"), Some("ab"), 1), + slot(Some(&"A".repeat(64)), Some(&"A".repeat(64)), 1), + OriginalSlot { + created_at: -1, + ..slot(None, None, 0) + }, + ] { + let event = notice(Some(&bad)); + assert!(content(&event).get("original").is_none(), "{bad:?}"); + assert_eq!(tags(&event).len(), 2, "{bad:?}"); + } + } + + #[test] + fn legacy_notice_omits_absent_moderation_metadata() { + let event = notice(None); + let content = content(&event); + assert_eq!(content["type"], "message_deleted"); + assert_eq!(content["target_event_id"], TARGET); + for field in ["action_id", "reason_code", "public_reason", "original"] { + assert!(content.get(field).is_none(), "{field}"); + } + } +} diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 2f4aa00b59..db821287ee 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -12,6 +12,8 @@ pub mod command_executor; pub mod community_provisioning; /// NIP-45 COUNT handler. pub mod count; +/// Relay-signed `message_deleted` deletion notice builder (kind 40099). +pub mod deletion_tombstone; /// EVENT handler — WS dispatcher → ingest pipeline → fan-out. pub mod event; /// NIP-IA identity archive request handler (kinds 9035–9036). diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index f63c174d96..78f787ada8 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -17,6 +17,7 @@ use buzz_core::StoredEvent; use buzz_db::channel::{MemberRecord, MemberRole}; use super::channel_authz::{self, ChannelAuthzError, PutUserDecision, RemoveOtherDecision}; +use super::deletion_tombstone::{build_delete_tombstone, original_slot, DeleteTombstone}; use super::event::dispatch_persistent_event; use crate::protocol::RelayMessage; use crate::state::AppState; @@ -840,17 +841,30 @@ pub async fn emit_system_message( .sign_with_keys(&state.relay_keypair) .map_err(|e| anyhow::anyhow!("failed to sign system message: {e}"))?; + persist_and_publish_system_event(tenant, state, channel_id, &event).await +} + +/// Durably insert a signed relay event, then best-effort fan it out. +/// +/// Returns `Err` only if the insert fails; a duplicate insert (same event ID) +/// is success, which makes retries of a stable event idempotent. +pub async fn persist_and_publish_system_event( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + event: &Event, +) -> anyhow::Result<()> { // Durable insert is the completion boundary — propagate failure. state .db - .insert_event(tenant.community(), &event, Some(channel_id)) + .insert_event(tenant.community(), event, Some(channel_id)) .await .map_err(|e| anyhow::anyhow!("system message insert failed: {e}"))?; // Fan out to subscribers: best-effort, clients can retrieve the persisted event. if let Err(e) = state .pubsub - .publish_event(tenant, EventTopic::Channel(channel_id), &event) + .publish_event(tenant, EventTopic::Channel(channel_id), event) .await { warn!("System message fan-out failed: {e}"); @@ -1829,12 +1843,12 @@ async fn handle_delete_event_side_effect( // Verify the target event belongs to the same channel as the h-tag. // Without this check, an admin of channel A could delete events in channel B // by sending h=A, e=. - if let Some(target_event) = state + let target_event = state .db .get_event_by_id_including_deleted_for_event_write(tenant.community(), &target_id) .await - .map_err(|e| anyhow::anyhow!("get_event_by_id failed: {e}"))? - { + .map_err(|e| anyhow::anyhow!("get_event_by_id failed: {e}"))?; + if let Some(target_event) = &target_event { match target_event.channel_id { Some(target_ch) if target_ch != channel_id => { return Err(anyhow::anyhow!( @@ -1859,20 +1873,45 @@ async fn handle_delete_event_side_effect( let parent_id = meta.as_ref().and_then(|m| m.parent_event_id.clone()); let root_id = meta.as_ref().and_then(|m| m.root_event_id.clone()); - // Atomically soft-delete the event and decrement thread counters in one transaction. + let target_hex = hex::encode(&target_id); + let actor_hex = hex::encode(event.pubkey.to_bytes()); + let action_id = extract_tag_value(event, "action_id"); + let reason_code = extract_tag_value(event, "reason_code"); + let public_reason = extract_tag_value(event, "public_reason"); + let original = target_event + .as_ref() + .map(|t| original_slot(t.event.created_at.as_secs() as i64, meta.as_ref())); + let notice = build_delete_tombstone( + &DeleteTombstone { + channel_id, + target_event_id: &target_hex, + actor: &actor_hex, + action_id: action_id.as_deref(), + reason_code: reason_code.as_deref(), + public_reason: public_reason.as_deref(), + original: original.as_ref(), + }, + chrono::Utc::now(), + &state.relay_keypair, + )?; + + // Soft-delete, thread counters, and the notice commit together: a failed + // notice insert rolls the deletion back so a retry cannot lose the notice. let deleted = state .db - .soft_delete_event_and_update_thread( + .soft_delete_event_with_notice( tenant.community(), &target_id, parent_id.as_deref(), root_id.as_deref(), + ¬ice, + channel_id, ) .await .map_err(|e| anyhow::anyhow!("soft_delete_event failed: {e}"))?; if !deleted { - warn!(target_event = %hex::encode(&target_id), "event already deleted or not found"); + warn!(target_event = %target_hex, "event already deleted or not found"); return Ok(()); // No-op: skip system message to avoid false audit records. } @@ -1882,17 +1921,14 @@ async fn handle_delete_event_side_effect( emit_live_thread_summary(tenant, state, channel_id, root_id); } - let actor_hex = hex::encode(event.pubkey.to_bytes()); - let mut tombstone = serde_json::json!({ - "type": "message_deleted", - "actor": actor_hex, - "target_event_id": hex::encode(&target_id), - }); - copy_optional_string_field(event, &mut tombstone, "action_id"); - copy_optional_string_field(event, &mut tombstone, "reason_code"); - copy_optional_string_field(event, &mut tombstone, "public_reason"); - - emit_system_message(tenant, state, channel_id, tombstone, chrono::Utc::now()).await?; + // Fan out to subscribers: best-effort, clients can retrieve the persisted event. + if let Err(e) = state + .pubsub + .publish_event(tenant, EventTopic::Channel(channel_id), ¬ice) + .await + { + warn!("Deletion notice fan-out failed: {e}"); + } info!(target_event = %hex::encode(&target_id), "NIP-29 DELETE_EVENT processed"); Ok(()) @@ -2588,19 +2624,6 @@ fn extract_tag_value(event: &Event, tag_name: &str) -> Option { None } -fn copy_optional_string_field(event: &Event, object: &mut serde_json::Value, tag_name: &str) { - let Some(value) = extract_tag_value(event, tag_name) else { - return; - }; - copy_optional_string_value(object, tag_name, value); -} - -fn copy_optional_string_value(object: &mut serde_json::Value, field_name: &str, value: String) { - if let Some(map) = object.as_object_mut() { - map.insert(field_name.to_string(), serde_json::Value::String(value)); - } -} - fn has_moderation_delete_metadata(event: &Event) -> bool { ["action_id", "reason_code", "public_reason"] .iter() @@ -2617,31 +2640,6 @@ fn actor_is_channel_owner_or_admin(members: &[MemberRecord], actor: &[u8]) -> bo .any(|m| m.pubkey == actor && (m.role == "owner" || m.role == "admin")) } -#[cfg(test)] -fn delete_tombstone_content( - actor_hex: String, - target_event_id: String, - action_id: Option, - reason_code: Option, - public_reason: Option, -) -> serde_json::Value { - let mut tombstone = serde_json::json!({ - "type": "message_deleted", - "actor": actor_hex, - "target_event_id": target_event_id, - }); - if let Some(action_id) = action_id { - copy_optional_string_value(&mut tombstone, "action_id", action_id); - } - if let Some(reason_code) = reason_code { - copy_optional_string_value(&mut tombstone, "reason_code", reason_code); - } - if let Some(public_reason) = public_reason { - copy_optional_string_value(&mut tombstone, "public_reason", public_reason); - } - tombstone -} - /// Validate a git repo identifier (d-tag value from kind:30617). /// /// Rules: `[a-zA-Z0-9._-]{1,64}`, no leading dots, no `..`. @@ -3872,38 +3870,6 @@ mod tests { })); } - #[test] - fn delete_tombstone_omits_absent_moderation_metadata() { - let content = - delete_tombstone_content("actor".to_string(), "target".to_string(), None, None, None); - - assert_eq!(content["type"], "message_deleted"); - assert_eq!(content["actor"], "actor"); - assert_eq!(content["target_event_id"], "target"); - assert!(content.get("action_id").is_none()); - assert!(content.get("reason_code").is_none()); - assert!(content.get("public_reason").is_none()); - } - - #[test] - fn delete_tombstone_carries_optional_moderation_metadata() { - let content = delete_tombstone_content( - "actor".to_string(), - "target".to_string(), - Some("550e8400-e29b-41d4-a716-446655440000".to_string()), - Some("spam".to_string()), - Some("Removed for spam.".to_string()), - ); - - assert_eq!(content["type"], "message_deleted"); - assert_eq!(content["actor"], "actor"); - assert_eq!(content["target_event_id"], "target"); - assert_eq!(content["action_id"], "550e8400-e29b-41d4-a716-446655440000"); - assert_eq!(content["reason_code"], "spam"); - assert_eq!(content["public_reason"], "Removed for spam."); - assert!(!content.to_string().contains("reporter")); - } - #[test] fn author_self_delete_with_moderation_metadata_skips_self_delete_path() { let keys = nostr::Keys::generate(); @@ -3950,3 +3916,135 @@ mod tests { assert!(actor_is_channel_owner_or_admin(&members, &actor)); } } + +#[cfg(test)] +mod postgres_tests { + use super::*; + use buzz_db::channel::{ChannelType, ChannelVisibility}; + use buzz_db::event::ThreadMetadataParams; + use nostr::{EventBuilder, Keys, Kind, Tag}; + + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 + + /// NIP-29 DELETE_EVENT stores exactly one relay-signed notice carrying the + /// deleted reply's slot, committed with the deletion itself. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn nip29_delete_notice_carries_original_slot() { + let url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_string()); + let pool = sqlx::PgPool::connect(&url).await.expect("connect"); + let state = crate::state::tests::test_state_with_database_pool(pool.clone()).await; + let community_uuid = Uuid::new_v4(); + let host = format!("tombstone-{}.example", community_uuid.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_uuid) + .bind(&host) + .execute(&pool) + .await + .expect("community"); + let community = buzz_core::CommunityId::from_uuid(community_uuid); + let tenant = TenantContext::resolved(community, host); + + let author = Keys::generate(); + let channel = Uuid::new_v4(); + state + .db + .create_channel_with_id( + community, + channel, + "tombstone", + ChannelType::Stream, + ChannelVisibility::Open, + None, + author.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("channel"); + let root = EventBuilder::new(Kind::Custom(9), "root") + .sign_with_keys(&author) + .expect("sign"); + let reply = EventBuilder::new(Kind::Custom(9), "reply") + .custom_created_at(root.created_at + 1) + .sign_with_keys(&author) + .expect("sign"); + let at = |e: &Event| { + chrono::DateTime::from_timestamp(e.created_at.as_secs() as i64, 0).expect("ts") + }; + for (event, parent) in [(&root, None), (&reply, Some(&root))] { + state + .db + .insert_event_with_thread_metadata( + community, + event, + Some(channel), + Some(ThreadMetadataParams { + event_id: event.id.as_bytes(), + event_created_at: at(event), + channel_id: channel, + parent_event_id: parent.map(|p| p.id.as_bytes().as_slice()), + parent_event_created_at: parent.map(at), + root_event_id: parent.map(|p| p.id.as_bytes().as_slice()), + root_event_created_at: parent.map(at), + depth: i32::from(parent.is_some()), + broadcast: false, + }), + ) + .await + .expect("thread event"); + } + + let delete = EventBuilder::new(Kind::Custom(9005), "") + .tags(vec![ + Tag::parse(["h", &channel.to_string()]).expect("h"), + Tag::parse(["e", &reply.id.to_hex()]).expect("e"), + Tag::parse(["public_reason", "Removed for spam."]).expect("reason"), + ]) + .sign_with_keys(&author) + .expect("sign"); + for _ in 0..2 { + handle_delete_event_side_effect(&tenant, &delete, &state) + .await + .expect("delete side effect"); + } + + let rows: Vec<(Vec, serde_json::Value, String)> = sqlx::query_as( + "SELECT pubkey, tags, content FROM events \ + WHERE community_id = $1 AND channel_id = $2 AND kind = 40099", + ) + .bind(community_uuid) + .bind(channel) + .fetch_all(&pool) + .await + .expect("notices"); + assert_eq!(rows.len(), 1, "a repeated delete adds no second notice"); + let (pubkey, tags, content) = &rows[0]; + assert_eq!( + pubkey, + &state.relay_keypair.public_key().to_bytes().to_vec() + ); + assert_eq!( + tags, + &serde_json::json!([ + ["h", channel.to_string()], + ["e", reply.id.to_hex()], + ["e", root.id.to_hex()], + ]) + ); + let content: serde_json::Value = serde_json::from_str(content).expect("json"); + assert_eq!(content["public_reason"], "Removed for spam."); + assert_eq!( + content["original"], + serde_json::json!({ + "version": 1, + "created_at": reply.created_at.as_secs(), + "parent_event_id": root.id.to_hex(), + "root_event_id": root.id.to_hex(), + "depth": 1, + "broadcast": false, + }) + ); + } +} diff --git a/migrations/0050_relay_admin_action_original_slot.sql b/migrations/0050_relay_admin_action_original_slot.sql new file mode 100644 index 0000000000..6c8259f09b --- /dev/null +++ b/migrations/0050_relay_admin_action_original_slot.sql @@ -0,0 +1,14 @@ +-- Freeze a delete action's original message slot (time + thread position) at +-- mutation time so the relay-signed deletion notice can be rebuilt identically +-- on every outbox retry, even after the target row is purged. +-- +-- NULL = not captured yet (pre-migration rows, non-delete actions). +-- {"slot": {...}} = captured slot. +-- {"slot": null} = captured, but the target row no longer existed. +-- +-- Additive and nullable; no backfill. Legacy delete actions freeze their slot +-- once during finalization. + +ALTER TABLE relay_admin_actions + ADD COLUMN original_slot JSONB + CHECK (original_slot IS NULL OR jsonb_typeof(original_slot) = 'object'); diff --git a/schema/schema.sql b/schema/schema.sql index 1a0a849f7a..160be8c911 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -1817,6 +1817,10 @@ CREATE TABLE relay_admin_actions ( enforcement_target_pubkey BYTEA CHECK (enforcement_target_pubkey IS NULL OR length(enforcement_target_pubkey) = 32), enforcement_channel_id UUID, + -- Frozen original slot of a delete target (migration 0050): NULL = not + -- captured, {"slot": {...}} = captured, {"slot": null} = target was gone. + original_slot JSONB + CHECK (original_slot IS NULL OR jsonb_typeof(original_slot) = 'object'), created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), -- Report-scoped idempotency: one action per (report, request_id).