jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-28 13:01:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(relay): carry the deleted original's slot in deletion notices

Browse Source 
Relay-signed message_deleted notices now record where the original lived
(created_at, parent, root, depth, broadcast) so clients can draw the tombstone
in place. One builder serves both the NIP-29 DELETE_EVENT path and the admin
outbox tombstone.

The NIP-29 soft-delete and its notice now commit in one transaction, so a
failed notice insert can no longer lose the marker. The admin path freezes the
slot on the action row with the mutation marker (migration 0050), and delivery
rebuilds from that payload, so retries sign identical bytes even after purge.
Rows from before the column existed freeze their slot once at finalization.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Hayt
2026-09-24 20:07:25 -04:00
parent 2013148852
commit d19dbe7652
10 changed files with 935 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/buzz-db/src/runtime/migration.rs
View File

@@ -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]

182
crates/buzz-db/src/store/event.rs
View File

@@ -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<String>,
/// Outer thread root event ID (hex); `None` for a top-level message.
pub root_event_id: Option<String>,
/// 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<Option<OriginalSlot>> {
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<Utc> = row.try_get("created_at")?;
let parent: Option<Vec<u8>> = row.try_get("parent_event_id")?;
let root: Option<Vec<u8>> = 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<bool> {
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<bool> {
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<u8>| async move {
sqlx::query_scalar::<_, Option<DateTime<Utc>>>(
"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, &notice, 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)

248
crates/buzz-db/src/store/relay_admin_actions.rs
View File

@@ -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<Uuid>,
/// Delete target's original slot, frozen at mutation time (or at
/// finalization for legacy rows). `None` = not captured yet.
pub original_slot: Option<FrozenSlot>,
/// Row creation time.
pub created_at: DateTime<Utc>,
/// Row last-updated time.
pub updated_at: DateTime<Utc>,
}
/// 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<OriginalSlot>,
}
/// 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<Option<sqlx::types::Json<FrozenSlot>>> = 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<Option<AdminAc
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 id = $1
"#,
@@ -1322,7 +1368,7 @@ pub async fn get_action_by_request(
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
@@ -1581,7 +1627,7 @@ pub async fn claim_stranded_action_batch(
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, created_at, updated_at
enforcement_channel_id, original_slot, created_at, updated_at
"#,
)
.bind(id)
@@ -1681,6 +1727,9 @@ fn row_to_action(row: sqlx::postgres::PgRow) -> Result<AdminActionRecord> {
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::<Option<sqlx::types::Json<FrozenSlot>>, _>("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<Utc> =
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

20
crates/buzz-relay/src/api/admin/mod.rs
View File

@@ -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<chrono::Utc> =
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 =

65
crates/buzz-relay/src/handlers/admin_outbox_worker.rs
View File

@@ -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<AppState>, 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<buzz_db::event::OriginalSlot> =
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, &notice,
)
.await
.map_err(|e| format!("tombstone: system message failed: {e}"))

249
crates/buzz-relay/src/handlers/deletion_tombstone.rs Normal file
View File

@@ -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<Utc>,
relay_keys: &Keys,
) -> anyhow::Result<Event> {
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<Vec<String>> {
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}");
}
}
}

2
crates/buzz-relay/src/handlers/mod.rs
View File

@@ -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).

276
crates/buzz-relay/src/handlers/side_effects.rs
View File

@@ -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<AppState>,
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=<event-in-B>.
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(),
&notice,
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), &notice)
.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<String> {
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<String>,
reason_code: Option<String>,
public_reason: Option<String>,
) -> 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<u8>, 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,
})
);
}
}

14
migrations/0050_relay_admin_action_original_slot.sql Normal file
View File

@@ -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');

4
schema/schema.sql
View File

@@ -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).