From ef85d468f7cf2feeea5139050307ace84d40b75f Mon Sep 17 00:00:00 2001 From: Ravneet Arora Date: Thu, 29 Oct 2026 13:59:07 -0400 Subject: [PATCH 1/3] Add standalone user message deletion handler Signed-off-by: Ravneet Arora --- crates/buzz-db/src/store/thread_window.rs | 10 +- .../store/user_deletion/messages/cleanup.rs | 247 +++++ .../src/store/user_deletion/messages/mod.rs | 203 +++++ .../store/user_deletion/messages/select.sql | 67 ++ crates/buzz-db/src/store/user_deletion/mod.rs | 2 + crates/buzz-deletion/src/user.rs | 2 + crates/buzz-deletion/src/user/messages.rs | 66 ++ .../buzz-deletion/src/user/messages/README.md | 111 +++ .../src/user/messages/postgres_tests.rs | 846 ++++++++++++++++++ 9 files changed, 1553 insertions(+), 1 deletion(-) create mode 100644 crates/buzz-db/src/store/user_deletion/messages/cleanup.rs create mode 100644 crates/buzz-db/src/store/user_deletion/messages/mod.rs create mode 100644 crates/buzz-db/src/store/user_deletion/messages/select.sql create mode 100644 crates/buzz-deletion/src/user/messages.rs create mode 100644 crates/buzz-deletion/src/user/messages/README.md create mode 100644 crates/buzz-deletion/src/user/messages/postgres_tests.rs diff --git a/crates/buzz-db/src/store/thread_window.rs b/crates/buzz-db/src/store/thread_window.rs index 926ffae8df0..3d98f9a32df 100644 --- a/crates/buzz-db/src/store/thread_window.rs +++ b/crates/buzz-db/src/store/thread_window.rs @@ -88,7 +88,15 @@ async fn select_window( // Retain tombstones: deleting a root must not make its readable replies // disappear. Never follow a root belonging to another channel/community. let root_in_channel: bool = sqlx::query_scalar( - "SELECT EXISTS(SELECT 1 FROM events WHERE community_id = $1 AND channel_id = $2 AND id = $3 AND kind = ANY($4))") + "SELECT EXISTS(SELECT 1 FROM events WHERE community_id = $1 AND channel_id = $2 AND id = $3 AND kind = ANY($4)) + OR EXISTS(SELECT 1 FROM thread_metadata root WHERE root.community_id = $1 + AND root.channel_id = $2 AND root.event_id = $3 AND root.depth = 0 + AND root.parent_event_id IS NULL + AND NOT EXISTS (SELECT 1 FROM events e WHERE e.community_id=$1 AND e.id=$3) + AND EXISTS (SELECT 1 FROM thread_metadata reply JOIN events e + ON e.community_id=reply.community_id AND e.id=reply.event_id AND e.created_at=reply.event_created_at + WHERE reply.community_id=$1 AND reply.channel_id=$2 AND reply.root_event_id=$3 + AND e.channel_id=$2 AND e.kind=ANY($4) AND e.deleted_at IS NULL))") .bind(community.as_uuid()).bind(request.channel).bind(&root) .bind(ROW_KINDS.map(|k| k as i32).to_vec()) .fetch_one(&mut *conn).await?; diff --git a/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs b/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs new file mode 100644 index 00000000000..dc81fa7e758 --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs @@ -0,0 +1,247 @@ +use super::*; + +pub(super) async fn remove( + tx: &mut AdmittedTx, + community: CommunityId, + candidates: &[Candidate], + result: &mut MessageBatch, +) -> Result<()> { + let thread = candidates + .first() + .ok_or_else(|| safety("message batch unexpectedly empty"))?; + if candidates.iter().any(|row| row.has_media) { + return Err(safety("message has attachment or link references; shared-file cleanup is required before deletion")); + } + let mut ids: Vec> = candidates.iter().map(|row| row.id.clone()).collect(); + // Edits and votes use the first valid e tag, reactions use the last. Never treat any + // arbitrary e tag as authority to delete another person's signed content. + let mut frontier = ids.clone(); + for round in 0..16 { + let dependents: Vec<(Vec, bool)> = sqlx::query_as( + "SELECT e.id, (e.content ~* 'https?://|/media/' OR EXISTS ( + SELECT 1 FROM jsonb_array_elements(e.tags) t + WHERE t->>0 IN ('imeta','link-preview','url','thumb','image','x'))) AS has_media + FROM events e + CROSS JOIN LATERAL ( + SELECT decode(t->>1, 'hex') AS id + FROM jsonb_array_elements(e.tags) WITH ORDINALITY a(t,n) + WHERE t->>0 = 'e' AND t->>1 ~ '^[0-9a-fA-F]{64}$' + ORDER BY CASE WHEN e.kind IN (40003,45002) THEN n ELSE -n END LIMIT 1 + ) target + WHERE e.community_id = $1 AND e.kind IN (7,40003,45002) + AND target.id = ANY($2) AND NOT (e.id = ANY($5)) + AND e.channel_id IS NOT DISTINCT FROM $3::uuid + ORDER BY e.created_at,e.id LIMIT $4", + ) + .bind(community.as_uuid()) + .bind(&frontier) + .bind(thread.channel_id) + .bind((MAX_DEPENDENTS + 1) as i64) + .bind(&ids) + .fetch_all(tx.conn()) + .await?; + if ids.len() + dependents.len() > MAX_DEPENDENTS { + return Err(safety( + "too many dependent events in one message batch; reduce the batch size", + )); + } + if dependents.iter().any(|(_, media)| *media) { + return Err(safety( + "message edit or reaction has file references; shared-file cleanup is required", + )); + } + if dependents.is_empty() { + break; + } + if round == 15 { + return Err(safety("dependent event chain exceeds 16 levels")); + } + frontier = dependents.into_iter().map(|(id, _)| id).collect(); + ids.extend(frontier.iter().cloned()); + ids.sort(); + ids.dedup(); + } + + // A worker may already have fetched the payload even though the user is + // fenced. Do not delete its recovery record while it can still deliver. + let delivering: bool = sqlx::query_scalar( + "SELECT EXISTS ( + SELECT 1 FROM push_match_queue WHERE community_id=$1 AND event_id=ANY($2) AND lease_until>clock_timestamp() + UNION ALL SELECT 1 FROM push_wake_outbox WHERE community_id=$1 AND event_id=ANY($2) AND lease_until>clock_timestamp() + UNION ALL SELECT 1 FROM operator_listener_outbox WHERE community_id=$1 AND event_id=ANY($2) AND lease_until>clock_timestamp())", + ).bind(community.as_uuid()).bind(&ids).fetch_one(tx.conn()).await?; + if delivering { + return Err(safety( + "message delivery workers must drain before deletion", + )); + } + let runs: Vec<(Uuid, String)> = sqlx::query_as( + "SELECT id, status::text FROM workflow_runs WHERE community_id=$1 + AND (trigger_event_id=ANY($2) OR lower(trigger_context->>'message_id')=ANY($3)) + ORDER BY id LIMIT $4 FOR UPDATE", + ) + .bind(community.as_uuid()) + .bind(&ids) + .bind(ids.iter().map(hex::encode).collect::>()) + .bind((MAX_DEPENDENTS + 1) as i64) + .fetch_all(tx.conn()) + .await?; + if runs.len() > MAX_DEPENDENTS { + return Err(safety( + "too many workflow runs in one message batch; reduce the batch size", + )); + } + if runs + .iter() + .any(|(_, status)| !matches!(status.as_str(), "completed" | "failed" | "cancelled")) + { + return Err(safety( + "workflow runs triggered by these messages must stop before deletion", + )); + } + let run_ids: Vec<_> = runs.into_iter().map(|(id, _)| id).collect(); + result.workflow_runs_deleted = delete_runs(tx, community, &run_ids).await?; + result.reactions_deleted = sqlx::query( + "DELETE FROM reactions WHERE community_id=$1 AND (event_id=ANY($2) OR reaction_event_id=ANY($2))", + ).bind(community.as_uuid()).bind(&ids).execute(tx.conn()).await?.rows_affected(); + result.mentions_deleted = + sqlx::query("DELETE FROM event_mentions WHERE community_id=$1 AND event_id=ANY($2)") + .bind(community.as_uuid()) + .bind(&ids) + .execute(tx.conn()) + .await? + .rows_affected(); + for query in [ + "DELETE FROM push_match_queue WHERE community_id=$1 AND event_id=ANY($2)", + "DELETE FROM push_wake_outbox WHERE community_id=$1 AND event_id=ANY($2)", + "DELETE FROM operator_listener_outbox WHERE community_id=$1 AND event_id=ANY($2)", + "DELETE FROM delivery_log WHERE community_id=$1 AND event_id=ANY($2)", + ] { + result.deliveries_deleted += sqlx::query(query) + .bind(community.as_uuid()) + .bind(&ids) + .execute(tx.conn()) + .await? + .rows_affected(); + } + result.events_deleted = sqlx::query("DELETE FROM events WHERE community_id=$1 AND id=ANY($2)") + .bind(community.as_uuid()) + .bind(&ids) + .execute(tx.conn()) + .await? + .rows_affected(); + repair_thread(tx, community, thread).await +} + +async fn repair_thread( + tx: &mut AdmittedTx, + community: CommunityId, + thread: &Candidate, +) -> Result<()> { + // Preserve only the ancestry needed by retained events. The missing event + // rows carry no content, author, tags or signature; metadata keeps replies + // reachable without manufacturing a signed replacement event. + sqlx::query( + "WITH RECURSIVE needed AS ( + SELECT tm.* FROM thread_metadata tm + WHERE tm.community_id=$1 AND tm.channel_id=$2 AND (tm.event_id=$3 OR tm.root_event_id=$3) + AND EXISTS (SELECT 1 FROM events e WHERE e.community_id=tm.community_id + AND e.id=tm.event_id AND e.created_at=tm.event_created_at) + UNION + SELECT p.* FROM thread_metadata p JOIN needed n ON p.community_id=n.community_id + AND p.channel_id=n.channel_id AND (p.event_id=n.parent_event_id OR p.event_id=n.root_event_id) + ) DELETE FROM thread_metadata tm WHERE tm.community_id=$1 AND tm.channel_id=$2 + AND (tm.event_id=$3 OR tm.root_event_id=$3) + AND NOT EXISTS (SELECT 1 FROM needed n WHERE n.event_id=tm.event_id AND n.event_created_at=tm.event_created_at)", + ).bind(community.as_uuid()).bind(thread.channel_id).bind(&thread.root_id).execute(tx.conn()).await?; + // Recompute, never subtract old counts. Already-soft-deleted messages do + // not count; received_at matches the server-arrival activity clock. + sqlx::query( + "WITH visible AS MATERIALIZED ( + SELECT tm.parent_event_id,tm.root_event_id,e.received_at FROM thread_metadata tm + JOIN events e ON e.community_id=tm.community_id AND e.id=tm.event_id AND e.created_at=tm.event_created_at + WHERE tm.community_id=$1 AND tm.channel_id=$2 AND tm.root_event_id=$3 AND e.deleted_at IS NULL + ), direct AS ( + SELECT parent_event_id,count(*)::int AS n,max(received_at) AS last_at FROM visible GROUP BY parent_event_id + ), descendants AS ( + SELECT root_event_id,count(*)::int AS n,max(received_at) AS last_at FROM visible GROUP BY root_event_id + ), counts AS ( + SELECT tm.event_id,tm.event_created_at,COALESCE(d.n,0) AS replies,COALESCE(r.n,0) AS descendants, + GREATEST(d.last_at,r.last_at) AS last_at + FROM thread_metadata tm LEFT JOIN direct d ON d.parent_event_id=tm.event_id + LEFT JOIN descendants r ON r.root_event_id=tm.event_id + WHERE tm.community_id=$1 AND tm.channel_id=$2 AND (tm.event_id=$3 OR tm.root_event_id=$3) + ) UPDATE thread_metadata tm SET reply_count=c.replies,descendant_count=c.descendants,last_reply_at=c.last_at + FROM counts c WHERE tm.community_id=$1 AND tm.event_id=c.event_id AND tm.event_created_at=c.event_created_at", + ).bind(community.as_uuid()).bind(thread.channel_id).bind(&thread.root_id).execute(tx.conn()).await?; + Ok(()) +} + +// The source event may have been removed before this handler existed. A saved +// trigger author still identifies the copy without guessing from mention tags. +pub(super) async fn remove_orphan_runs( + tx: &mut AdmittedTx, + community: CommunityId, + subject: &str, + limit: u16, +) -> Result { + let runs: Vec<(Uuid, String)> = sqlx::query_as( + "SELECT id,status::text FROM workflow_runs WHERE community_id=$1 + AND lower(trigger_context->>'author')=$2 ORDER BY id LIMIT $3 FOR UPDATE", + ) + .bind(community.as_uuid()) + .bind(subject) + .bind(i64::from(limit)) + .fetch_all(tx.conn()) + .await?; + if runs + .iter() + .any(|(_, status)| !matches!(status.as_str(), "completed" | "failed" | "cancelled")) + { + return Err(safety( + "workflow runs triggered by this user must stop before deletion", + )); + } + delete_runs( + tx, + community, + &runs.into_iter().map(|(id, _)| id).collect::>(), + ) + .await +} + +async fn delete_runs(tx: &mut AdmittedTx, community: CommunityId, run_ids: &[Uuid]) -> Result { + // Retain the scheduled-fire claim: removing it would allow cron to fire + // again. Only clear its optional reference to the run being removed. + sqlx::query("UPDATE scheduled_workflow_fires SET workflow_run_id=NULL WHERE community_id=$1 AND workflow_run_id=ANY($2)") + .bind(community.as_uuid()).bind(run_ids).execute(tx.conn()).await?; + let deleted = sqlx::query("DELETE FROM workflow_runs WHERE community_id=$1 AND id=ANY($2)") + .bind(community.as_uuid()) + .bind(run_ids) + .execute(tx.conn()) + .await? + .rows_affected(); + // workflow_approvals cascade from each removed run; the workflow definition + // (possibly owned by a different user) is left intact. + Ok(deleted) +} + +// A reaction can outlive its signed source event in historical data. +pub(super) async fn remove_orphan_reactions( + tx: &mut AdmittedTx, + community: CommunityId, + subject: &[u8], + limit: u16, +) -> Result { + Ok(sqlx::query( + "DELETE FROM reactions WHERE (community_id,event_created_at,event_id,pubkey,emoji) IN ( + SELECT community_id,event_created_at,event_id,pubkey,emoji FROM reactions + WHERE community_id = $1 AND pubkey = $2 LIMIT $3)", + ) + .bind(community.as_uuid()) + .bind(subject) + .bind(i64::from(limit)) + .execute(tx.conn()) + .await? + .rows_affected()) +} diff --git a/crates/buzz-db/src/store/user_deletion/messages/mod.rs b/crates/buzz-db/src/store/user_deletion/messages/mod.rs new file mode 100644 index 00000000000..f7cdca2a3ab --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/messages/mod.rs @@ -0,0 +1,203 @@ +//! Standalone, bounded message removal. No workflow router calls this module. +//! +//! Each call owns one transaction and selects remaining rows, so retries need no +//! message cursor. A community deletion lock excludes admitted database writers +//! for the short batch; it does not replace prior user gating or external drain. + +use super::{normalize_user_deletion_pubkey, UserDeletionInventory, UserDeletionStore}; +use crate::entity_deletion::{lock_owned, validate_approved_inventory, EntityLease, EntityStage}; +use crate::{AdmittedTx, DbError, Result}; +use buzz_core::CommunityId; +use serde::Serialize; +use uuid::Uuid; + +mod cleanup; + +/// Maximum authored events removed by one transaction. Dependents have a separate cap. +pub const MAX_BATCH_SIZE: u16 = 100; +const MAX_DEPENDENTS: usize = 2000; +const SELECT: &str = include_str!("select.sql"); + +/// Result of one committed batch, not completion of the user's overall deletion. +#[derive(Debug, Default, Serialize)] +pub struct MessageBatch { + /// Thread processed; absent when the community has no remaining message work. + pub root_id: Option, + /// Authored and dependent event rows removed in this transaction. + pub events_deleted: u64, + /// Materialized reactions removed, including already removed reactions. + pub reactions_deleted: u64, + /// Mention-index entries belonging to deleted events. + pub mentions_deleted: u64, + /// Queue and delivery-history entries for deleted events. + pub deliveries_deleted: u64, + /// Terminal workflow executions removed with their approvals. + pub workflow_runs_deleted: u64, + /// No supported message events, orphan reactions or identifiable workflow copies remain. + pub exhausted: bool, +} + +#[derive(Clone, sqlx::FromRow)] +struct Candidate { + id: Vec, + channel_id: Option, + root_id: Vec, + has_media: bool, +} + +fn safety(message: &str) -> DbError { + DbError::DeletionSafety(message.into()) +} + +impl UserDeletionStore { + /// Remove one batch in one thread under the current `purge_messages` stage. + /// + /// Requires the exact approved inventory, a current lease, a global user + /// fence from this request, and drained external community writes. The + /// caller must already have stopped user writes and workflow execution; + /// fence storage alone does not yet enforce serving paths. The relay signer + /// must come from trusted relay configuration, never user-supplied tags. + /// + /// Attachment/link-bearing events and nonterminal workflow runs fail closed + /// before any deletion. Media, audit/moderation, personal state and other + /// users' signed messages are outside this operation. No stage advancement, + /// lease release, network call or main-workflow integration occurs here. + pub async fn purge_message_batch( + &self, + lease: &EntityLease, + stage_id: Uuid, + community: CommunityId, + relay_pubkey: &[u8; 32], + batch_size: u16, + ) -> Result { + if !(1..=MAX_BATCH_SIZE).contains(&batch_size) { + return Err(safety("message batch size must be 1..=100")); + } + let mut tx = self.pool.begin().await?; + sqlx::query("SET LOCAL statement_timeout = '10s'") + .execute(&mut *tx) + .await?; + sqlx::query("SET LOCAL lock_timeout = '2s'") + .execute(&mut *tx) + .await?; + // Existing lock order: community, request, identities. A shared lock + // followed by an upgrade would deadlock two deletion workers. + sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))") + .bind(community.as_uuid()) + .execute(&mut *tx) + .await?; + crate::deletion::guard_community_write(&mut tx, community).await?; + let request = lock_owned(&mut tx, lease).await?; + validate_approved_inventory(&mut tx, &request).await?; + let inventory = UserDeletionInventory::read(&request)?; + let expected_inputs = inventory.inputs(&request.target, Some(*community.as_uuid()))?; + let stage: EntityStage = sqlx::query_as( + "SELECT id, request_id, target_community_id, stage, status, is_current, + started_at, finished_at, metadata_json, last_error, updated_at + FROM entity_deletion_stage_history WHERE id = $1 FOR UPDATE", + ) + .bind(stage_id) + .fetch_one(&mut *tx) + .await?; + if stage.request_id != lease.request_id + || stage.stage != "purge_messages" + || stage.status != "running" + || !stage.is_current + || stage.target_community_id != Some(*community.as_uuid()) + || stage.metadata_json["inputs"] != expected_inputs + { + return Err(safety("expected the current approved purge_messages stage")); + } + let subject_hex = normalize_user_deletion_pubkey(&request.target)?; + let subject = hex::decode(&subject_hex).map_err(|_| safety("invalid subject"))?; + if subject == relay_pubkey { + return Err(safety("cannot purge the relay signing identity")); + } + let identity = crate::user_deletion_fences::FenceIdentity::user( + nostr::PublicKey::from_slice(&subject).map_err(|_| safety("invalid subject"))?, + ); + let fence = crate::user_deletion_fences::find(&mut tx, identity) + .await? + .ok_or_else(|| safety("user must be fenced before message deletion"))?; + if fence.request_id != lease.request_id { + return Err(safety("user fence belongs to another deletion request")); + } + let active: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM community_serving_write_leases + WHERE community_id = $1 AND lease_until > clock_timestamp())", + ) + .bind(community.as_uuid()) + .fetch_one(&mut *tx) + .await?; + if active { + return Err(safety( + "external community writes must drain before message deletion", + )); + } + let mut tx = AdmittedTx::admit( + tx, + &crate::deletion::DeletionStore::new(self.pool.clone()), + community, + ) + .await?; + let seed = select(&mut tx, community, &subject, relay_pubkey, None, 1).await?; + let mut result = MessageBatch::default(); + if let Some(seed) = seed.first() { + result.root_id = Some(hex::encode(&seed.root_id)); + let candidates = select( + &mut tx, + community, + &subject, + relay_pubkey, + Some(seed), + batch_size, + ) + .await?; + cleanup::remove(&mut tx, community, &candidates, &mut result).await?; + } else { + // A historic reaction can outlive its source event. Drain these in + // bounded batches too; deleting only signed events would miss them. + result.reactions_deleted = + cleanup::remove_orphan_reactions(&mut tx, community, &subject, batch_size).await?; + result.workflow_runs_deleted = + cleanup::remove_orphan_runs(&mut tx, community, &subject_hex, batch_size).await?; + result.exhausted = result.reactions_deleted == 0 && result.workflow_runs_deleted == 0; + } + // Expiry during a long query must roll back even though the parent row + // lock prevented a replacement worker from claiming in the meantime. + // Ownership cannot change while the request row is locked. Only the + // clock can invalidate the lease after admission. + let unexpired: bool = sqlx::query_scalar( + "SELECT lease_expires_at > clock_timestamp() FROM entity_deletion_requests WHERE id=$1", + ) + .bind(lease.request_id) + .fetch_one(tx.conn()) + .await?; + if !unexpired { + return Err(DbError::AccessDenied( + "deletion lease expired during message batch".into(), + )); + } + tx.commit().await?; + Ok(result) + } +} + +async fn select( + tx: &mut AdmittedTx, + community: CommunityId, + subject: &[u8], + relay: &[u8; 32], + thread: Option<&Candidate>, + limit: u16, +) -> Result> { + Ok(sqlx::query_as(SELECT) + .bind(community.as_uuid()) + .bind(subject) + .bind(relay.as_slice()) + .bind(thread.map(|t| &t.root_id)) + .bind(i64::from(limit)) + .bind(thread.and_then(|t| t.channel_id)) + .fetch_all(tx.conn()) + .await?) +} diff --git a/crates/buzz-db/src/store/user_deletion/messages/select.sql b/crates/buzz-db/src/store/user_deletion/messages/select.sql new file mode 100644 index 00000000000..83cd3a0dfb9 --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/messages/select.sql @@ -0,0 +1,67 @@ +-- $1 community, $2 subject, $3 trusted relay signer, $4 optional root, +-- $5 limit, $6 optional channel (used only with $4). +WITH authored AS ( + SELECT e.* + FROM events e + WHERE e.community_id = $1 AND e.pubkey IN ($2, $3) + AND e.kind IN (9, 40001, 40002, 45001, 45003, 7, 45002, 40003, 40004, 40005, 40006, 40007, 40008) + AND CASE WHEN e.pubkey = $3 THEN COALESCE( + -- Match effective_message_author: first actor, then first valid p. + (SELECT CASE WHEN t->>1 ~ '^[0-9a-fA-F]{64}$' THEN decode(t->>1, 'hex') END + FROM jsonb_array_elements(e.tags) t WHERE t->>0 = 'actor' LIMIT 1), + (SELECT decode(t->>1, 'hex') FROM jsonb_array_elements(e.tags) t + WHERE t->>0 = 'p' AND t->>1 ~ '^[0-9a-fA-F]{64}$' LIMIT 1), e.pubkey) + ELSE e.pubkey END = $2 +), located AS ( + SELECT e.id, e.created_at, e.channel_id, + CASE WHEN e.kind IN (9, 40001, 40002, 45001, 45003, 40006, 40007) THEN + COALESCE(tm.root_event_id, + CASE WHEN markers.reply IS NOT NULL THEN COALESCE(markers.root, parent.root_event_id, markers.reply) END, e.id) + ELSE COALESCE(target.root_event_id, + CASE WHEN target_markers.reply IS NOT NULL THEN COALESCE(target_markers.root, target_markers.reply) END, + refs.target_id, e.id) END AS root_id, + -- Deliberately conservative until shared-file cleanup is available. Never + -- discard the only source of a blob/preview reference and claim success. + (e.content ~* 'https?://|/media/' OR EXISTS ( + SELECT 1 FROM jsonb_array_elements(e.tags) t + WHERE t->>0 IN ('imeta', 'link-preview', 'url', 'thumb', 'image', 'x') + )) AS has_media + FROM authored e + LEFT JOIN thread_metadata tm ON tm.community_id = e.community_id + AND tm.event_id = e.id AND tm.event_created_at = e.created_at AND tm.channel_id = e.channel_id + LEFT JOIN LATERAL ( + SELECT + (array_agg(decode(t->>1, 'hex') ORDER BY n DESC) FILTER (WHERE t->>3 = 'root'))[1] AS root, + (array_agg(decode(t->>1, 'hex') ORDER BY n DESC) FILTER (WHERE t->>3 = 'reply'))[1] AS reply + FROM jsonb_array_elements(e.tags) WITH ORDINALITY a(t,n) + WHERE t->>0 = 'e' AND t->>1 ~ '^[0-9a-fA-F]{64}$' + ) markers ON true + LEFT JOIN LATERAL ( + SELECT root_event_id FROM thread_metadata p WHERE p.community_id = e.community_id + AND p.channel_id = e.channel_id AND p.event_id = markers.reply LIMIT 1 + ) parent ON true + LEFT JOIN LATERAL ( + SELECT decode(t->>1, 'hex') AS target_id + FROM jsonb_array_elements(e.tags) WITH ORDINALITY a(t,n) + WHERE t->>0 = 'e' AND t->>1 ~ '^[0-9a-fA-F]{64}$' + ORDER BY CASE WHEN e.kind IN (40003,45002) THEN n ELSE -n END LIMIT 1 + ) refs ON true + LEFT JOIN LATERAL ( + SELECT root_event_id FROM thread_metadata p WHERE p.community_id = e.community_id + AND p.channel_id = e.channel_id AND p.event_id = refs.target_id LIMIT 1 + ) target ON true + LEFT JOIN LATERAL ( + SELECT p.tags FROM events p WHERE p.community_id = e.community_id + AND p.channel_id IS NOT DISTINCT FROM e.channel_id AND p.id = refs.target_id LIMIT 1 + ) target_event ON true + LEFT JOIN LATERAL ( + SELECT + (array_agg(decode(t->>1, 'hex') ORDER BY n DESC) FILTER (WHERE t->>3 = 'root'))[1] AS root, + (array_agg(decode(t->>1, 'hex') ORDER BY n DESC) FILTER (WHERE t->>3 = 'reply'))[1] AS reply + FROM jsonb_array_elements(target_event.tags) WITH ORDINALITY a(t,n) + WHERE t->>0 = 'e' AND t->>1 ~ '^[0-9a-fA-F]{64}$' + ) target_markers ON true +) +SELECT id, created_at, channel_id, root_id, has_media FROM located +WHERE ($4::bytea IS NULL OR (root_id = $4 AND channel_id IS NOT DISTINCT FROM $6::uuid)) +ORDER BY root_id, channel_id NULLS FIRST, created_at, id LIMIT $5 diff --git a/crates/buzz-db/src/store/user_deletion/mod.rs b/crates/buzz-db/src/store/user_deletion/mod.rs index 1c093946945..aa3de5dbeec 100644 --- a/crates/buzz-db/src/store/user_deletion/mod.rs +++ b/crates/buzz-db/src/store/user_deletion/mod.rs @@ -3,6 +3,8 @@ use crate::Db; use sqlx::PgPool; mod discovery; mod inventory; +/// Standalone per-thread message cleanup. +pub mod messages; mod requests; pub use discovery::normalize_user_deletion_pubkey; pub use inventory::UserDeletionInventory; diff --git a/crates/buzz-deletion/src/user.rs b/crates/buzz-deletion/src/user.rs index 93d1766fdb3..4ebb4040fcb 100644 --- a/crates/buzz-deletion/src/user.rs +++ b/crates/buzz-deletion/src/user.rs @@ -152,6 +152,8 @@ pub async fn run(command: Command) -> Result { mod handlers; mod inventory; +/// Standalone message cleanup; not dispatched by the dummy workflow. +pub mod messages; /// Standalone delegation of an owned community to community deletion. pub mod owned_community; /// Read-only summaries of locally saved request progress and external waits. diff --git a/crates/buzz-deletion/src/user/messages.rs b/crates/buzz-deletion/src/user/messages.rs new file mode 100644 index 00000000000..7eb8bf5d831 --- /dev/null +++ b/crates/buzz-deletion/src/user/messages.rs @@ -0,0 +1,66 @@ +//! Standalone message handler. Deliberately absent from dispatch and routing. + +use anyhow::{bail, Context, Result}; +use buzz_core::CommunityId; +use buzz_db::{ + entity_deletion::{EntityLease, EntityStage}, + user_deletion::messages::MessageBatch, + Db, +}; + +/// Run one atomic thread batch and record failure or final stage success. +/// +/// The caller owns heartbeat/release and invokes this again while `exhausted` +/// is false. Restart selects remaining database rows; no message cursor or +/// message content is written to stage metadata. This handler never chooses +/// another stage or community. See `purge_message_batch` for drain prerequisites +/// and the intentionally blocked attachment/workflow cases. +pub async fn handle( + db: &Db, + lease: &EntityLease, + stage: &EntityStage, + relay_pubkey: &[u8; 32], + batch_size: u16, +) -> Result { + if stage.request_id != lease.request_id || stage.stage != "purge_messages" { + bail!("expected this request's purge_messages stage"); + } + let community = CommunityId::from_uuid( + stage + .target_community_id + .context("missing message community")?, + ); + let store = db.entity_deletion_store(); + let stage = store.start_stage(lease, stage.id).await?; + match db + .user_deletion_store() + .purge_message_batch(lease, stage.id, community, relay_pubkey, batch_size) + .await + { + Ok(batch) => { + if batch.exhausted { + store + .save_stage_success( + lease, + stage.id, + serde_json::json!({"message_rows_exhausted": true}), + ) + .await?; + } + Ok(batch) + } + Err(error) => { + if let Err(save_error) = store + .fail_stage(lease, stage.id, &error.to_string(), crate::RETRY_DELAY) + .await + { + return Err(anyhow::Error::new(error) + .context(format!("also could not save retry: {save_error}"))); + } + Err(error.into()) + } + } +} + +#[cfg(test)] +mod postgres_tests; diff --git a/crates/buzz-deletion/src/user/messages/README.md b/crates/buzz-deletion/src/user/messages/README.md new file mode 100644 index 00000000000..d3c6c64dbe0 --- /dev/null +++ b/crates/buzz-deletion/src/user/messages/README.md @@ -0,0 +1,111 @@ +# Standalone message deletion + +`buzz_deletion::user::messages::handle` processes one database batch for a +`purge_messages` stage. It is intentionally absent from the CLI, dummy dispatch, +transition rules and background executor. Adding this module does not enable +real deletion in any existing workflow. + +The caller supplies a claimed `EntityLease`, the current community stage, the +trusted relay public key from configuration, and a batch size from 1 to 100. +Call again while `MessageBatch.exhausted` is false. The handler selects remaining +rows, ordered by thread, rather than saving a cursor. It never changes community, +chooses the next stage, releases ownership or starts a heartbeat. The caller owns +those operations. A successful final empty scan marks only this stage complete. +Failures are recorded through the existing stage retry mechanism and propagated. + +## Preconditions + +The approved request must target a real hex key or npub and include this +participant/former-participant community. Stage inputs must match that exact +inventory. A global user fence from this request must exist. User admission, +workflow execution and external delivery must already be gated and drained. +That includes writes targeting the removed messages, such as other people's +reactions/edits; a pre-transaction target lookup must not later recreate a +dependent row after the batch commits. +**The current fence table alone does not enforce those serving paths.** Do not +invoke this handler against serving production until that integration exists. + +Each batch briefly takes the existing community deletion lock exclusively, +then the request lock, then the user identity lock. This waits for admitted +database writes, prevents counter races during the transaction, and does not +change the community lifecycle. It temporarily blocks other writes to this +community. Lock waits are limited to two seconds and statements to ten seconds. +Active external write leases, claimed deliveries and nonterminal affected +workflow runs prevent deletion. These checks do not recall already delivered +messages or replace application-level drain. + +## What one batch removes + +- Authored stream messages (9, 40001, 40002), forum posts/comments (45001, + 45003), reactions, forum votes, edits, individual pin/bookmark events, + scheduled-message/reminder events (40006/40007), and diff events (40008). + Old soft-deleted rows are included. Community-global private reminders, + personal lists and other account state belong to later handlers. +- Relay-signed attributed messages only when the signer is the trusted relay + key. Attribution follows the existing first-actor/first-valid-p rules. An + arbitrary author's actor or mention tags never confer authorship. +- Dependent edits, votes and reactions on removed events, including reactions + from other people. The first valid e tag selects an edit/vote target; the + last selects a reaction target. Other people's messages, diffs, pins and lists + are not removed merely because they reference a deleted message. +- Materialized reactions, event_mentions, push_match_queue, push_wake_outbox, + operator_listener_outbox and delivery_log entries for the removed events. +- Terminal workflow runs whose trigger event or saved message_id matches a + removed event, including saved text, traces and errors. Their approvals + cascade. Workflow definitions remain. Scheduled-fire claims remain with + their optional run pointer cleared, preventing duplicate scheduled execution. + +Authored candidates are capped at 100; the complete dependent event set and +matched runs are capped at 2,000 each. Dependent traversal stops with an error +past 16 levels, never commits an incomplete traversal. A smaller batch can +resolve a fan-out limit; a single over-limit message requires further work. +Only IDs and flags are loaded, not message bodies or workflow traces. + +All deletions and thread repairs commit together. Any SQL failure or lease +expiry rolls back the batch. Re-running after a successful commit cannot +subtract counts twice. A final bounded sweep removes orphan materialized +reactions authored by the subject even if their source events no longer exist. +It also removes terminal workflow copies identifiable by the saved trigger +author when their source event is already missing; active runs block completion. + +## What remains + +Thread metadata retains only ancestry needed by remaining events. It has no +message body, tags, author or signature. Counts are recomputed from visible +remaining messages using the server-arrival clock. Thread-window reads recognize +a missing root through trusted metadata and still return original signed +replies. No fake signed tombstone is manufactured. Posting new replies to a +physically missing parent and client refresh/notification behavior are separate +integration work; this PR tests retrieval of surviving replies. + +Other people's messages, their mention tags and mention-index entries remain. +Moderation reports/actions, administrative records and the hash-chain audit log +remain. Other people's copied text and messages generated by their workflows +are not classified as the subject's messages just because the source was a +trigger. Profile, membership, personal-state, agent and storage cleanup remain +separate. This stage's result is not an assertion of complete account erasure. + +## Files are an explicit blocker + +Shared physical files are not safe to delete based solely on an author's +message. Until the media handler can prove which files remain in use, any +selected event or dependent edit/reaction with imeta, link-preview, URL/image/ +thumbnail/hash tags or a content URL fails before deletion. This deliberately +also blocks some ordinary links. The source events remain available for a retry +once file cleanup exists; the handler never drops file references and reports +success. This PR does not add a bypass or a pretend storage adapter. + +## Local tests + +Use the repository's isolated PostgreSQL lane: + +```sh +scripts/postgres-test-run.sh -p buzz-deletion --lib \ + -E 'test(user::messages::postgres_tests)' +``` + +Tests exercise the real handler, signed events and the database. They cover +shared roots and replies, edits, both directions of reactions, mentions, +delivery records, workflow copies, failures/restart, tenant boundaries, +trusted-relay attribution, fences and stale leases. No production connection +or workflow integration is needed. diff --git a/crates/buzz-deletion/src/user/messages/postgres_tests.rs b/crates/buzz-deletion/src/user/messages/postgres_tests.rs new file mode 100644 index 00000000000..63517f475bd --- /dev/null +++ b/crates/buzz-deletion/src/user/messages/postgres_tests.rs @@ -0,0 +1,846 @@ +use super::*; +use buzz_core::channel::{ChannelType, ChannelVisibility}; +use buzz_db::entity_deletion::{inventory_digest, NewEntityStage, NextStage, StageTransition}; +use buzz_db::event::ThreadMetadataParams; +use buzz_db::user_deletion_fences::{self, FenceIdentity}; +use chrono::{DateTime, Utc}; +use nostr::{Event, EventBuilder, Keys, Kind, Tag}; +use serde_json::json; +use sqlx::PgPool; +use std::time::Duration; +use uuid::Uuid; + +struct Fixture { + db: Db, + community: CommunityId, + channel: Uuid, + alice: Keys, + bob: Keys, + relay: Keys, + lease: EntityLease, + stage: EntityStage, +} + +fn database_url() -> String { + std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap() +} + +async fn fixture() -> Fixture { + let url = database_url(); + let db = Db::from_pool(PgPool::connect(&url).await.unwrap()); + let community = db + .ensure_configured_community(&format!("messages-{}.example", Uuid::new_v4())) + .await + .unwrap() + .id; + let alice = Keys::generate(); + let bob = Keys::generate(); + let relay = Keys::generate(); + let channel = Uuid::new_v4(); + db.create_channel_with_id( + community, + channel, + "messages", + ChannelType::Stream, + ChannelVisibility::Private, + None, + &bob.public_key().to_bytes(), + None, + ) + .await + .unwrap(); + let target = alice.public_key().to_hex(); + let item = + json!({"community_id": community.as_uuid(), "relationship":"participant", "label":"test"}); + let inventory = json!({"communities":[item]}); + let digest = inventory_digest(&inventory).unwrap(); + let request: Uuid = sqlx::query_scalar( + "INSERT INTO entity_deletion_requests (target,requested_by,status,inventory_manifest,inventory_digest) + VALUES ($1,'submitter','inventoried',$2,$3) RETURNING id", + ).bind(&target).bind(&inventory).bind(&digest).fetch_one(db.pool()).await.unwrap(); + let store = db.entity_deletion_store(); + store.approve(request, "approver", &digest).await.unwrap(); + let lease = store + .claim(request, "messages-test", Duration::from_secs(600)) + .await + .unwrap() + .unwrap(); + let NextStage::Stage(stage) = store + .transition_stage(&lease, None, |_, _| { + Ok(StageTransition::Initialize(NewEntityStage { + stage: "purge_messages".into(), + community: Some(*community.as_uuid()), + inputs: json!({"target":target,"community":item}), + })) + }) + .await + .unwrap() + else { + panic!("stage expected") + }; + let mut tx = db.pool().begin().await.unwrap(); + user_deletion_fences::insert( + &mut tx, + FenceIdentity::user(alice.public_key()), + &lease, + None, + ) + .await + .unwrap(); + tx.commit().await.unwrap(); + Fixture { + db, + community, + channel, + alice, + bob, + relay, + lease, + stage, + } +} + +fn tag(parts: &[&str]) -> Tag { + Tag::parse(parts.iter().copied()).unwrap() +} +fn timestamp(e: &Event) -> DateTime { + DateTime::from_timestamp(e.created_at.as_secs() as i64, 0).unwrap() +} + +impl Fixture { + async fn event( + &self, + who: &Keys, + kind: u16, + text: &str, + mut tags: Vec, + thread: Option<(&Event, &Event, i32)>, + ) -> Event { + tags.push(tag(&["h", &self.channel.to_string()])); + if let Some((root, parent, _)) = thread { + tags.push(tag(&["e", &root.id.to_hex(), "", "root"])); + tags.push(tag(&["e", &parent.id.to_hex(), "", "reply"])); + } + let event = EventBuilder::new(Kind::Custom(kind), text) + .tags(tags) + .sign_with_keys(who) + .unwrap(); + let metadata = if [9, 40002, 45001, 45003].contains(&kind) { + Some(ThreadMetadataParams { + event_id: event.id.as_bytes(), + event_created_at: timestamp(&event), + channel_id: self.channel, + parent_event_id: thread.map(|(_, p, _)| p.id.as_bytes().as_slice()), + parent_event_created_at: thread.map(|(_, p, _)| timestamp(p)), + root_event_id: thread.map(|(r, _, _)| r.id.as_bytes().as_slice()), + root_event_created_at: thread.map(|(r, _, _)| timestamp(r)), + depth: thread.map_or(0, |(_, _, d)| d), + broadcast: false, + }) + } else { + None + }; + self.db + .insert_event_with_thread_metadata(self.community, &event, Some(self.channel), metadata) + .await + .unwrap(); + event + } + async fn batch(&self, size: u16) -> Result { + handle( + &self.db, + &self.lease, + &self.stage, + &self.relay.public_key().to_bytes(), + size, + ) + .await + } + async fn finish(&self, size: u16) { + for _ in 0..30 { + if self.batch(size).await.unwrap().exhausted { + return; + } + } + panic!("handler did not exhaust bounded fixture"); + } + async fn exists(&self, e: &Event) -> bool { + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM events WHERE community_id=$1 AND id=$2)") + .bind(self.community.as_uuid()) + .bind(e.id.as_bytes().as_slice()) + .fetch_one(self.db.pool()) + .await + .unwrap() + } + async fn reaction(&self, actor: &Keys, target: &Event) -> Event { + let e = self + .event(actor, 7, "👍", vec![tag(&["e", &target.id.to_hex()])], None) + .await; + sqlx::query("INSERT INTO reactions (community_id,event_created_at,event_id,pubkey,emoji,reaction_event_id) VALUES ($1,$2,$3,$4,'👍',$5)") + .bind(self.community.as_uuid()).bind(timestamp(target)).bind(target.id.as_bytes().as_slice()) + .bind(actor.public_key().to_bytes().as_slice()).bind(e.id.as_bytes().as_slice()).execute(self.db.pool()).await.unwrap(); + e + } + async fn count(&self, query: &str) -> i64 { + sqlx::query_scalar(sqlx::AssertSqlSafe(query)) + .bind(self.community.as_uuid()) + .fetch_one(self.db.pool()) + .await + .unwrap() + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn removes_messages_edits_reactions_mentions_and_delivery_but_keeps_others() { + let f = fixture().await; + let root = f.event(&f.bob, 9, "Bob root", vec![], None).await; + let alice = f + .event( + &f.alice, + 9, + "Alice secret", + vec![tag(&["p", &f.bob.public_key().to_hex()])], + Some((&root, &root, 1)), + ) + .await; + let bob = f + .event( + &f.bob, + 9, + "Bob mentions Alice", + vec![tag(&["p", &f.alice.public_key().to_hex()])], + Some((&root, &alice, 2)), + ) + .await; + let edit1 = f + .event( + &f.alice, + 40003, + "old edit secret", + vec![tag(&["e", &alice.id.to_hex()])], + None, + ) + .await; + let edit2 = f + .event( + &f.alice, + 40003, + "new edit secret", + vec![tag(&["e", &alice.id.to_hex()])], + None, + ) + .await; + sqlx::query("UPDATE events SET deleted_at=now() WHERE community_id=$1 AND id=$2") + .bind(f.community.as_uuid()) + .bind(edit1.id.as_bytes().as_slice()) + .execute(f.db.pool()) + .await + .unwrap(); + let own_reaction = f.reaction(&f.alice, &root).await; + let other_reaction = f.reaction(&f.bob, &alice).await; + let kept_reaction = f.reaction(&f.bob, &root).await; + sqlx::query("INSERT INTO push_match_queue (community_id,event_id) VALUES ($1,$2) ON CONFLICT DO NOTHING") + .bind(f.community.as_uuid()).bind(alice.id.as_bytes().as_slice()).execute(f.db.pool()).await.unwrap(); + sqlx::query("INSERT INTO delivery_log (community_id,event_id,success) VALUES ($1,$2,true)") + .bind(f.community.as_uuid()) + .bind(alice.id.as_bytes().as_slice()) + .execute(f.db.pool()) + .await + .unwrap(); + sqlx::query("INSERT INTO operator_listener_outbox (listener_pubkey,target_pubkey,community_id,event_id,event_kind,event_created_at) VALUES ($1,$2,$3,$4,9,$5)") + .bind(f.bob.public_key().to_bytes().as_slice()).bind(f.alice.public_key().to_bytes().as_slice()) + .bind(f.community.as_uuid()).bind(alice.id.as_bytes().as_slice()).bind(timestamp(&alice)).execute(f.db.pool()).await.unwrap(); + sqlx::query("INSERT INTO push_leases (community_id,author,installation_id,source_event_id,source_created_at,generation,active,expires_at) VALUES ($1,$2,'test',$3,0,1,false,0)") + .bind(f.community.as_uuid()).bind(f.bob.public_key().to_bytes().as_slice()).bind([88u8;32].as_slice()).execute(f.db.pool()).await.unwrap(); + sqlx::query("INSERT INTO push_wake_outbox (community_id,author,installation_id,lease_generation,endpoint_hash,event_id,class,expires_at) VALUES ($1,$2,'test',1,$3,$4,'default',0)") + .bind(f.community.as_uuid()).bind(f.bob.public_key().to_bytes().as_slice()).bind([89u8;32].as_slice()).bind(alice.id.as_bytes().as_slice()).execute(f.db.pool()).await.unwrap(); + f.finish(1).await; + for event in [&alice, &edit1, &edit2, &own_reaction, &other_reaction] { + assert!(!f.exists(event).await); + } + for event in [&root, &bob, &kept_reaction] { + assert!(f.exists(event).await); + } + assert_eq!( + f.count("SELECT count(*) FROM reactions WHERE community_id=$1") + .await, + 1 + ); + assert_eq!( + f.count("SELECT count(*) FROM event_mentions WHERE community_id=$1") + .await, + 1 + ); + for table in [ + "push_match_queue", + "push_wake_outbox", + "delivery_log", + "operator_listener_outbox", + ] { + assert_eq!( + f.count(&format!( + "SELECT count(*) FROM {table} WHERE community_id=$1" + )) + .await, + 0 + ); + } + let summary = + f.db.get_thread_summary(f.community, root.id.as_bytes()) + .await + .unwrap() + .unwrap(); + assert_eq!((summary.reply_count, summary.descendant_count), (0, 1)); + let intermediate = + f.db.get_thread_metadata_by_event(f.community, alice.id.as_bytes()) + .await + .unwrap() + .unwrap(); + assert_eq!(intermediate.reply_count, 1); + let completed = + f.db.entity_deletion_store() + .get(f.lease.request_id) + .await + .unwrap(); + assert_ne!(completed.status, "completed"); // handler never advances the request + let (status, current): (String, bool) = + sqlx::query_as("SELECT status,is_current FROM entity_deletion_stage_history WHERE id=$1") + .bind(f.stage.id) + .fetch_one(f.db.pool()) + .await + .unwrap(); + assert_eq!(status, "completed"); + assert!(current); // the central transition function still owns routing +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn purged_root_keeps_signed_replies_readable_without_retaining_its_payload() { + let f = fixture().await; + let root = f.event(&f.alice, 9, "private root", vec![], None).await; + let reply = f + .event(&f.bob, 9, "retained reply", vec![], Some((&root, &root, 1))) + .await; + f.finish(10).await; + assert!(!f.exists(&root).await); + let req=buzz_core::thread_window::Request::parse(&json!({"thread_window":true,"#h":[f.channel],"#e":[root.id.to_hex()],"kinds":[9],"limit":10})).unwrap(); + let (window, _) = + f.db.get_thread_window_with_session( + f.community, + &req, + &mut buzz_db::thread_window::ScanBudget::default(), + ) + .await + .unwrap(); + assert!(window.root_in_channel); + assert_eq!(window.rows.len(), 1); + assert_eq!(window.rows[0].event, reply); + let mut wrong = req.clone(); + wrong.channel = Uuid::new_v4(); + let (window, _) = + f.db.get_thread_window_with_session( + f.community, + &wrong, + &mut buzz_db::thread_window::ScanBudget::default(), + ) + .await + .unwrap(); + assert!(!window.root_in_channel); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn attachment_in_an_old_edit_preserves_source_and_records_retry() { + let f = fixture().await; + let root = f.event(&f.alice, 9, "message", vec![], None).await; + let edit = f + .event( + &f.alice, + 40003, + "old attachment", + vec![ + tag(&["e", &root.id.to_hex()]), + tag(&["imeta", "url https://files.example/a"]), + ], + None, + ) + .await; + let error = f.batch(10).await.unwrap_err(); + assert!(error.to_string().contains("file") || error.to_string().contains("attachment")); + assert!(f.exists(&root).await && f.exists(&edit).await); + let (status, error): (String, Option) = + sqlx::query_as("SELECT status,last_error FROM entity_deletion_stage_history WHERE id=$1") + .bind(f.stage.id) + .fetch_one(f.db.pool()) + .await + .unwrap(); + assert_eq!(status, "failed"); + assert!(error.is_some()); + assert_eq!( + f.db.entity_deletion_store() + .get(f.lease.request_id) + .await + .unwrap() + .status, + "retry_wait" + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn database_failure_rolls_back_dependents_and_counts_then_restart_finishes() { + let f = fixture().await; + let root = f.event(&f.bob, 9, "root", vec![], None).await; + let alice = f + .event( + &f.alice, + 9, + "reply", + vec![tag(&["p", &f.bob.public_key().to_hex()])], + Some((&root, &root, 1)), + ) + .await; + let reaction = f.reaction(&f.bob, &alice).await; + let trigger = format!("fail_message_purge_{}", Uuid::new_v4().simple()); + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "CREATE FUNCTION {trigger}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'injected failure'; END $$; + CREATE TRIGGER {trigger} BEFORE UPDATE ON thread_metadata FOR EACH ROW + WHEN (NEW.community_id = '{}') EXECUTE FUNCTION {trigger}();", + f.community.as_uuid(), + ))).execute(f.db.pool()).await.unwrap(); + assert!(f + .batch(10) + .await + .unwrap_err() + .to_string() + .contains("injected failure")); + assert!(f.exists(&alice).await && f.exists(&reaction).await); + assert_eq!( + f.count("SELECT count(*) FROM reactions WHERE community_id=$1") + .await, + 1 + ); + assert_eq!( + f.count("SELECT count(*) FROM event_mentions WHERE community_id=$1") + .await, + 1 + ); + assert_eq!( + f.db.get_thread_summary(f.community, root.id.as_bytes()) + .await + .unwrap() + .unwrap() + .reply_count, + 1 + ); + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "DROP TRIGGER {trigger} ON thread_metadata; DROP FUNCTION {trigger}();" + ))) + .execute(f.db.pool()) + .await + .unwrap(); + sqlx::query( + "UPDATE entity_deletion_requests SET next_attempt_at=clock_timestamp() WHERE id=$1", + ) + .bind(f.lease.request_id) + .execute(f.db.pool()) + .await + .unwrap(); + let restarted = Db::from_pool(PgPool::connect(&database_url()).await.unwrap()); + assert!( + !handle( + &restarted, + &f.lease, + &f.stage, + &f.relay.public_key().to_bytes(), + 1 + ) + .await + .unwrap() + .exhausted + ); + f.finish(1).await; + assert!(!f.exists(&alice).await && !f.exists(&reaction).await); + assert_eq!( + f.db.get_thread_summary(f.community, root.id.as_bytes()) + .await + .unwrap() + .unwrap() + .reply_count, + 0 + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn only_trusted_relay_attribution_counts_and_other_community_survives() { + let f = fixture().await; + let alice = f.event(&f.alice, 9, "direct", vec![], None).await; + let relayed = f + .event( + &f.relay, + 9, + "legacy attributed", + vec![tag(&["p", &f.alice.public_key().to_hex()])], + None, + ) + .await; + let bob = f + .event( + &f.bob, + 9, + "not Alice", + vec![tag(&["actor", &f.alice.public_key().to_hex()])], + None, + ) + .await; + let other = + f.db.ensure_configured_community("other.example") + .await + .unwrap() + .id; + // Same signed event in another tenant, with no shared channel provenance. + f.db.insert_event(other, &alice, None).await.unwrap(); + f.finish(10).await; + assert!(!f.exists(&alice).await && !f.exists(&relayed).await); + assert!(f.exists(&bob).await); + let count: i64 = + sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2") + .bind(other.as_uuid()) + .bind(alice.id.as_bytes().as_slice()) + .fetch_one(f.db.pool()) + .await + .unwrap(); + assert_eq!(count, 1); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn missing_fence_wrong_inventory_and_stale_lease_cannot_delete() { + let f = fixture().await; + let event = f + .event(&f.alice, 9, "keep until authorized", vec![], None) + .await; + f.db.entity_deletion_store() + .start_stage(&f.lease, f.stage.id) + .await + .unwrap(); + let mut stale = f.lease.clone(); + stale.generation += 1; + assert!(f + .db + .user_deletion_store() + .purge_message_batch( + &stale, + f.stage.id, + f.community, + &f.relay.public_key().to_bytes(), + 1 + ) + .await + .is_err()); + let other = + f.db.ensure_configured_community("unapproved.example") + .await + .unwrap() + .id; + assert!(f + .db + .user_deletion_store() + .purge_message_batch( + &f.lease, + f.stage.id, + other, + &f.relay.public_key().to_bytes(), + 1 + ) + .await + .is_err()); + sqlx::query("DELETE FROM user_deletion_fences WHERE request_id=$1") + .bind(f.lease.request_id) + .execute(f.db.pool()) + .await + .unwrap(); + assert!(f + .db + .user_deletion_store() + .purge_message_batch( + &f.lease, + f.stage.id, + f.community, + &f.relay.public_key().to_bytes(), + 1 + ) + .await + .unwrap_err() + .to_string() + .contains("fenced")); + assert!(f.exists(&event).await); +} + +async fn workflow(f: &Fixture, trigger: &Event, status: &str) -> (Uuid, Uuid) { + sqlx::query("INSERT INTO users (community_id,pubkey) VALUES ($1,$2) ON CONFLICT DO NOTHING") + .bind(f.community.as_uuid()) + .bind(f.bob.public_key().to_bytes().as_slice()) + .execute(f.db.pool()) + .await + .unwrap(); + let workflow:Uuid=sqlx::query_scalar("INSERT INTO workflows (community_id,name,owner_pubkey,definition,definition_hash) VALUES ($1,'Bob workflow',$2,'{}','\\x01') RETURNING id") + .bind(f.community.as_uuid()).bind(f.bob.public_key().to_bytes().as_slice()).fetch_one(f.db.pool()).await.unwrap(); + let run:Uuid=sqlx::query_scalar("INSERT INTO workflow_runs (community_id,workflow_id,status,trigger_event_id,trigger_context,execution_trace) VALUES ($1,$2,$3::run_status,$4,$5,$6) RETURNING id") + .bind(f.community.as_uuid()).bind(workflow).bind(status).bind(trigger.id.as_bytes().as_slice()) + .bind(json!({"text":"saved secret","author":f.alice.public_key().to_hex(),"message_id":trigger.id.to_hex()})) + .bind(json!([{"output":"copied secret"}])).fetch_one(f.db.pool()).await.unwrap(); + sqlx::query("INSERT INTO workflow_approvals (community_id,token,workflow_id,run_id,step_id,step_index,approver_spec,expires_at) VALUES ($1,'\\x01',$2,$3,'approval',0,'owner',now())") + .bind(f.community.as_uuid()).bind(workflow).bind(run).execute(f.db.pool()).await.unwrap(); + sqlx::query("INSERT INTO scheduled_workflow_fires (community_id,workflow_id,scheduled_for,workflow_run_id) VALUES ($1,$2,now(),$3)") + .bind(f.community.as_uuid()).bind(workflow).bind(run).execute(f.db.pool()).await.unwrap(); + (workflow, run) +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn workflow_copies_and_approvals_are_deleted_but_definition_and_cron_claim_remain() { + let f = fixture().await; + let event = f.event(&f.alice, 9, "workflow trigger", vec![], None).await; + let (_workflow, _run) = workflow(&f, &event, "completed").await; + f.finish(10).await; + assert_eq!( + f.count("SELECT count(*) FROM workflow_runs WHERE community_id=$1") + .await, + 0 + ); + assert_eq!( + f.count("SELECT count(*) FROM workflow_approvals WHERE community_id=$1") + .await, + 0 + ); + assert_eq!( + f.count("SELECT count(*) FROM workflows WHERE community_id=$1") + .await, + 1 + ); + assert_eq!(f.count("SELECT count(*) FROM scheduled_workflow_fires WHERE community_id=$1 AND workflow_run_id IS NULL").await,1); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn active_workflow_and_claimed_delivery_prevent_message_removal() { + let f = fixture().await; + let event = f.event(&f.alice, 9, "still in use", vec![], None).await; + let (_, run) = workflow(&f, &event, "running").await; + assert!(f + .batch(10) + .await + .unwrap_err() + .to_string() + .contains("workflow runs")); + assert!(f.exists(&event).await); + sqlx::query("UPDATE workflow_runs SET status='cancelled' WHERE community_id=$1 AND id=$2") + .bind(f.community.as_uuid()) + .bind(run) + .execute(f.db.pool()) + .await + .unwrap(); + sqlx::query( + "UPDATE entity_deletion_requests SET next_attempt_at=clock_timestamp() WHERE id=$1", + ) + .bind(f.lease.request_id) + .execute(f.db.pool()) + .await + .unwrap(); + sqlx::query("INSERT INTO push_match_queue (community_id,event_id,state,lease_until) VALUES ($1,$2,'matching',now()+interval '1 hour') ON CONFLICT (community_id,event_id) DO UPDATE SET state='matching',lease_until=now()+interval '1 hour'") + .bind(f.community.as_uuid()).bind(event.id.as_bytes().as_slice()).execute(f.db.pool()).await.unwrap(); + assert!(f + .batch(10) + .await + .unwrap_err() + .to_string() + .contains("delivery workers")); + assert!(f.exists(&event).await); + assert_eq!( + f.count("SELECT count(*) FROM workflow_runs WHERE community_id=$1") + .await, + 1 + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn reaction_only_threads_and_orphan_rows_are_not_missed() { + let f = fixture().await; + let root = f.event(&f.bob, 9, "Bob only", vec![], None).await; + let reaction = f.reaction(&f.alice, &root).await; + sqlx::query("INSERT INTO reactions (community_id,event_created_at,event_id,pubkey,emoji,removed_at) VALUES ($1,now(),$2,$3,'old',now())") + .bind(f.community.as_uuid()).bind([42u8;32].as_slice()).bind(f.alice.public_key().to_bytes().as_slice()).execute(f.db.pool()).await.unwrap(); + f.finish(1).await; + assert!(f.exists(&root).await); + assert!(!f.exists(&reaction).await); + assert_eq!( + f.count("SELECT count(*) FROM reactions WHERE community_id=$1") + .await, + 0 + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn existing_soft_delete_does_not_decrement_remaining_counts_twice() { + let f = fixture().await; + let root = f.event(&f.bob, 9, "Bob root", vec![], None).await; + let alice = f + .event( + &f.alice, + 9, + "already hidden", + vec![], + Some((&root, &root, 1)), + ) + .await; + f.event(&f.bob, 9, "retained", vec![], Some((&root, &root, 1))) + .await; + sqlx::query("UPDATE events SET deleted_at=now() WHERE community_id=$1 AND id=$2") + .bind(f.community.as_uuid()) + .bind(alice.id.as_bytes().as_slice()) + .execute(f.db.pool()) + .await + .unwrap(); + // Seed the correct post-soft-delete counts. Hard deletion must not subtract again. + sqlx::query("UPDATE thread_metadata SET reply_count=1,descendant_count=1 WHERE community_id=$1 AND event_id=$2") + .bind(f.community.as_uuid()).bind(root.id.as_bytes().as_slice()).execute(f.db.pool()).await.unwrap(); + f.finish(1).await; + let summary = + f.db.get_thread_summary(f.community, root.id.as_bytes()) + .await + .unwrap() + .unwrap(); + assert_eq!((summary.reply_count, summary.descendant_count), (1, 1)); + assert!(summary.last_reply_at.is_some()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn uses_the_authoritative_target_and_preserves_other_peoples_diffs_and_pins() { + let f = fixture().await; + let alice = f.event(&f.alice, 45001, "Alice forum", vec![], None).await; + let bob = f.event(&f.bob, 45001, "Bob forum", vec![], None).await; + let keep_vote = f + .event( + &f.bob, + 45002, + "+", + vec![ + tag(&["e", &bob.id.to_hex()]), + tag(&["e", &alice.id.to_hex()]), + ], + None, + ) + .await; + let delete_vote = f + .event( + &f.bob, + 45002, + "+", + vec![ + tag(&["e", &alice.id.to_hex()]), + tag(&["e", &bob.id.to_hex()]), + ], + None, + ) + .await; + let keep_reaction = f + .event( + &f.bob, + 7, + "+", + vec![ + tag(&["e", &alice.id.to_hex()]), + tag(&["e", &bob.id.to_hex()]), + ], + None, + ) + .await; + let delete_reaction = f + .event( + &f.bob, + 7, + "+", + vec![ + tag(&["e", &bob.id.to_hex()]), + tag(&["e", &alice.id.to_hex()]), + ], + None, + ) + .await; + let diff = f + .event( + &f.bob, + 40008, + "Bob diff", + vec![tag(&["e", &alice.id.to_hex()])], + None, + ) + .await; + let pin = f + .event( + &f.bob, + 40004, + "Bob pin", + vec![tag(&["e", &alice.id.to_hex()])], + None, + ) + .await; + f.finish(10).await; + for event in [&alice, &delete_vote, &delete_reaction] { + assert!(!f.exists(event).await); + } + for event in [&bob, &keep_vote, &keep_reaction, &diff, &pin] { + assert!(f.exists(event).await); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn legacy_messages_without_metadata_and_fully_removed_threads_exhaust() { + let f = fixture().await; + let root = f.event(&f.alice, 9, "root", vec![], None).await; + let reply = f + .event(&f.alice, 9, "legacy reply", vec![], Some((&root, &root, 1))) + .await; + sqlx::query("DELETE FROM thread_metadata WHERE community_id=$1 AND event_id=$2") + .bind(f.community.as_uuid()) + .bind(reply.id.as_bytes().as_slice()) + .execute(f.db.pool()) + .await + .unwrap(); + f.finish(1).await; + assert!(!f.exists(&root).await && !f.exists(&reply).await); + assert_eq!( + f.count("SELECT count(*) FROM thread_metadata WHERE community_id=$1") + .await, + 0 + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn saved_workflow_author_finds_copies_after_the_source_event_is_gone() { + let f = fixture().await; + let source = f.event(&f.alice, 9, "old source", vec![], None).await; + workflow(&f, &source, "completed").await; + sqlx::query("DELETE FROM events WHERE community_id=$1 AND id=$2") + .bind(f.community.as_uuid()) + .bind(source.id.as_bytes().as_slice()) + .execute(f.db.pool()) + .await + .unwrap(); + f.finish(1).await; + assert_eq!( + f.count("SELECT count(*) FROM workflow_runs WHERE community_id=$1") + .await, + 0 + ); +} From d2f31e85ea4fa9c97b063581f265d523bb3601fd Mon Sep 17 00:00:00 2001 From: Ravneet Arora Date: Thu, 29 Oct 2026 17:09:48 -0400 Subject: [PATCH 2/3] Preserve custom emoji references during user deletion Signed-off-by: Ravneet Arora --- .../store/user_deletion/messages/cleanup.rs | 2 +- .../store/user_deletion/messages/select.sql | 2 +- .../buzz-deletion/src/user/messages/README.md | 21 +- .../src/user/messages/postgres_tests.rs | 2 + .../postgres_tests/safety_postgres_tests.rs | 278 ++++++++++++++++++ 5 files changed, 301 insertions(+), 4 deletions(-) create mode 100644 crates/buzz-deletion/src/user/messages/postgres_tests/safety_postgres_tests.rs diff --git a/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs b/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs index dc81fa7e758..ae13d168ef6 100644 --- a/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs +++ b/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs @@ -20,7 +20,7 @@ pub(super) async fn remove( let dependents: Vec<(Vec, bool)> = sqlx::query_as( "SELECT e.id, (e.content ~* 'https?://|/media/' OR EXISTS ( SELECT 1 FROM jsonb_array_elements(e.tags) t - WHERE t->>0 IN ('imeta','link-preview','url','thumb','image','x'))) AS has_media + WHERE t->>0 IN ('imeta','link-preview','url','thumb','image','x','emoji'))) AS has_media FROM events e CROSS JOIN LATERAL ( SELECT decode(t->>1, 'hex') AS id diff --git a/crates/buzz-db/src/store/user_deletion/messages/select.sql b/crates/buzz-db/src/store/user_deletion/messages/select.sql index 83cd3a0dfb9..f14dacde1fa 100644 --- a/crates/buzz-db/src/store/user_deletion/messages/select.sql +++ b/crates/buzz-db/src/store/user_deletion/messages/select.sql @@ -24,7 +24,7 @@ WITH authored AS ( -- discard the only source of a blob/preview reference and claim success. (e.content ~* 'https?://|/media/' OR EXISTS ( SELECT 1 FROM jsonb_array_elements(e.tags) t - WHERE t->>0 IN ('imeta', 'link-preview', 'url', 'thumb', 'image', 'x') + WHERE t->>0 IN ('imeta', 'link-preview', 'url', 'thumb', 'image', 'x', 'emoji') )) AS has_media FROM authored e LEFT JOIN thread_metadata tm ON tm.community_id = e.community_id diff --git a/crates/buzz-deletion/src/user/messages/README.md b/crates/buzz-deletion/src/user/messages/README.md index d3c6c64dbe0..484711ba355 100644 --- a/crates/buzz-deletion/src/user/messages/README.md +++ b/crates/buzz-deletion/src/user/messages/README.md @@ -34,6 +34,15 @@ Active external write leases, claimed deliveries and nonterminal affected workflow runs prevent deletion. These checks do not recall already delivered messages or replace application-level drain. +The caller must renew its lease before a batch and leave enough time for the +entire transaction. The ten-second statement timeout is not a whole-batch +deadline. The request-row lock blocks heartbeat renewal during a batch; a +heartbeat waiting on that lock cannot revive a lease that has expired. Expiry +rolls back the batch. If the lease has actually expired, the old worker also +cannot save a failure: the current running stage remains durable and a new +worker can reclaim it with a higher generation. Reduce batch size if repeated +expiry prevents progress; this handler does not impose a new lease duration. + ## What one batch removes - Authored stream messages (9, 40001, 40002), forum posts/comments (45001, @@ -90,8 +99,9 @@ separate. This stage's result is not an assertion of complete account erasure. Shared physical files are not safe to delete based solely on an author's message. Until the media handler can prove which files remain in use, any selected event or dependent edit/reaction with imeta, link-preview, URL/image/ -thumbnail/hash tags or a content URL fails before deletion. This deliberately -also blocks some ordinary links. The source events remain available for a retry +thumbnail/hash/emoji tags or a content URL fails before deletion. NIP-30 custom +emoji tags carry image URLs even when the content contains only a shortcode. +This deliberately also blocks some ordinary links. The source events remain available for a retry once file cleanup exists; the handler never drops file references and reports success. This PR does not add a bypass or a pretend storage adapter. @@ -109,3 +119,10 @@ shared roots and replies, edits, both directions of reactions, mentions, delivery records, workflow copies, failures/restart, tenant boundaries, trusted-relay attribution, fences and stale leases. No production connection or workflow integration is needed. + +Separate safety cases cover saved-input mismatch, live external-write leases, +and expiry injected during count repair after admission and deletion. Each +asserts that source/dependent records survive and retry state is saved. A +replacement-worker case expires the lease between committed batches, reclaims +through the real API, rejects the old worker/heartbeat, and finishes with the +new generation. This does not simulate heartbeat contention inside a live batch. diff --git a/crates/buzz-deletion/src/user/messages/postgres_tests.rs b/crates/buzz-deletion/src/user/messages/postgres_tests.rs index 63517f475bd..43582d590c8 100644 --- a/crates/buzz-deletion/src/user/messages/postgres_tests.rs +++ b/crates/buzz-deletion/src/user/messages/postgres_tests.rs @@ -10,6 +10,8 @@ use sqlx::PgPool; use std::time::Duration; use uuid::Uuid; +mod safety_postgres_tests; + struct Fixture { db: Db, community: CommunityId, diff --git a/crates/buzz-deletion/src/user/messages/postgres_tests/safety_postgres_tests.rs b/crates/buzz-deletion/src/user/messages/postgres_tests/safety_postgres_tests.rs new file mode 100644 index 00000000000..abbdf9a39ee --- /dev/null +++ b/crates/buzz-deletion/src/user/messages/postgres_tests/safety_postgres_tests.rs @@ -0,0 +1,278 @@ +use super::*; + +async fn conversation(f: &Fixture) -> (Event, Event, Event) { + let root = f.event(&f.bob, 9, "root", vec![], None).await; + let source = f + .event( + &f.alice, + 9, + "reply", + vec![tag(&["p", &f.bob.public_key().to_hex()])], + Some((&root, &root, 1)), + ) + .await; + let reaction = f.reaction(&f.bob, &source).await; + (root, source, reaction) +} + +async fn assert_preserved(f: &Fixture, root: &Event, source: &Event, reaction: &Event) { + assert!(f.exists(root).await && f.exists(source).await && f.exists(reaction).await); + assert_eq!( + f.count("SELECT count(*) FROM reactions WHERE community_id=$1") + .await, + 1 + ); + assert_eq!( + f.count("SELECT count(*) FROM event_mentions WHERE community_id=$1") + .await, + 1 + ); + assert_eq!( + f.db.get_thread_summary(f.community, root.id.as_bytes()) + .await + .unwrap() + .unwrap() + .reply_count, + 1 + ); +} + +async fn assert_retry(f: &Fixture, expected: &str) { + let (status, current, finished, error): (String, bool, Option>, String) = + sqlx::query_as("SELECT status,is_current,finished_at,last_error FROM entity_deletion_stage_history WHERE id=$1") + .bind(f.stage.id).fetch_one(f.db.pool()).await.unwrap(); + assert_eq!(status, "failed"); + assert!(current && finished.is_none()); + assert!(error.contains(expected), "{error}"); + let (status, retries, scheduled): (String, i32, bool) = sqlx::query_as( + "SELECT status,retry_count,next_attempt_at IS NOT NULL FROM entity_deletion_requests WHERE id=$1", + ).bind(f.lease.request_id).fetch_one(f.db.pool()).await.unwrap(); + assert_eq!(status, "retry_wait"); + assert_eq!(retries, 1); + assert!(scheduled); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn authored_nip30_emoji_preserves_image_reference_and_saves_retry() { + let f = fixture().await; + let source = f + .event( + &f.alice, + 9, + ":party:", + vec![tag(&["emoji", "party", "https://files.example/party.png"])], + None, + ) + .await; + let reaction = f.reaction(&f.bob, &source).await; + let error = f.batch(10).await.unwrap_err(); + assert!(error.to_string().contains("attachment"), "{error}"); + assert!(f.exists(&source).await && f.exists(&reaction).await); + assert_eq!( + f.count("SELECT count(*) FROM reactions WHERE community_id=$1") + .await, + 1 + ); + assert_retry(&f, "attachment").await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn dependent_nip30_emoji_preserves_source_and_saves_retry() { + let f = fixture().await; + let (root, source, reaction) = conversation(&f).await; + let emoji = f + .event( + &f.bob, + 7, + ":party:", + vec![ + tag(&["e", &source.id.to_hex()]), + tag(&["emoji", "party", "https://files.example/party.png"]), + ], + None, + ) + .await; + let error = f.batch(10).await.unwrap_err(); + assert!(error.to_string().contains("file references"), "{error}"); + assert!(f.exists(&emoji).await); + assert_preserved(&f, &root, &source, &reaction).await; + assert_retry(&f, "file references").await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn mismatched_saved_inputs_refuse_the_correct_approved_community() { + let f = fixture().await; + let (root, source, reaction) = conversation(&f).await; + // Keep the target community, request, digest, approval and fence valid. + // Only the saved input snapshot differs from the approved inventory. + sqlx::query("UPDATE entity_deletion_stage_history SET metadata_json=jsonb_set(metadata_json,'{inputs,target}','\"different-user\"') WHERE id=$1") + .bind(f.stage.id).execute(f.db.pool()).await.unwrap(); + let error = f.batch(10).await.unwrap_err(); + assert!( + error + .to_string() + .contains("current approved purge_messages stage"), + "{error}" + ); + assert_preserved(&f, &root, &source, &reaction).await; + assert_retry(&f, "current approved purge_messages stage").await; +} + +async fn external_lease(f: &Fixture, community: CommunityId) { + sqlx::query("INSERT INTO community_serving_write_leases (community_id,operation,owner,fence_generation,lease_until) VALUES ($1,'test-upload','external-worker',0,clock_timestamp()+interval '10 minutes')") + .bind(community.as_uuid()).execute(f.db.pool()).await.unwrap(); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn active_external_write_lease_refuses_only_its_community() { + let f = fixture().await; + let (root, source, reaction) = conversation(&f).await; + external_lease(&f, f.community).await; + let error = f.batch(10).await.unwrap_err(); + assert!( + error + .to_string() + .contains("external community writes must drain"), + "{error}" + ); + assert_preserved(&f, &root, &source, &reaction).await; + assert_retry(&f, "external community writes must drain").await; + + // A live lease on another tenant must not stop this tenant's cleanup. + let other = fixture().await; + external_lease(&other, other.community).await; + sqlx::query("DELETE FROM community_serving_write_leases WHERE community_id=$1") + .bind(f.community.as_uuid()) + .execute(f.db.pool()) + .await + .unwrap(); + sqlx::query( + "UPDATE entity_deletion_requests SET next_attempt_at=clock_timestamp() WHERE id=$1", + ) + .bind(f.lease.request_id) + .execute(f.db.pool()) + .await + .unwrap(); + f.finish(10).await; + assert!(!f.exists(&source).await && !f.exists(&reaction).await); + assert!(f.exists(&root).await); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn expiry_after_admission_rolls_back_deletions_and_count_repairs() { + let f = fixture().await; + let (root, source, reaction) = conversation(&f).await; + let trigger = format!("expire_message_lease_{}", Uuid::new_v4().simple()); + // Inject expiry at count repair, after all admission checks and deletes. + // No wall-clock sleep or race with the initial lease check is needed. + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "CREATE FUNCTION {trigger}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN + UPDATE entity_deletion_requests SET lease_expires_at=clock_timestamp()-interval '1 second' WHERE id='{}'; + RETURN NEW; END $$; + CREATE TRIGGER {trigger} AFTER UPDATE ON thread_metadata FOR EACH ROW + WHEN (NEW.community_id='{}') EXECUTE FUNCTION {trigger}();", + f.lease.request_id, f.community.as_uuid(), + ))).execute(f.db.pool()).await.unwrap(); + let outcome = f.batch(10).await; + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "DROP TRIGGER {trigger} ON thread_metadata; DROP FUNCTION {trigger}();", + ))) + .execute(f.db.pool()) + .await + .unwrap(); + let error = outcome.unwrap_err(); + assert!( + error + .to_string() + .contains("lease expired during message batch"), + "{error}" + ); + assert_preserved(&f, &root, &source, &reaction).await; + // The injected deadline is transactional too. Rollback restores the live + // lease, so this case can assert durable failure recording independently. + assert_retry(&f, "lease expired during message batch").await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn replacement_worker_reclaims_and_resumes_without_deleting_twice() { + let f = fixture().await; + let (root, first, reaction) = conversation(&f).await; + let second = f + .event(&f.alice, 9, "second reply", vec![], Some((&root, &root, 1))) + .await; + let batch = f.batch(1).await.unwrap(); + assert!(!batch.exhausted); + assert_ne!(f.exists(&first).await, f.exists(&second).await); + let remaining = if f.exists(&first).await { + &first + } else { + &second + }; + sqlx::query("UPDATE entity_deletion_requests SET lease_expires_at=clock_timestamp()-interval '1 second' WHERE id=$1") + .bind(f.lease.request_id).execute(f.db.pool()).await.unwrap(); + let restarted = Db::from_pool(PgPool::connect(&database_url()).await.unwrap()); + let successor = restarted + .entity_deletion_store() + .claim( + f.lease.request_id, + "replacement-worker", + Duration::from_secs(600), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(successor.generation, f.lease.generation + 1); + assert!(f + .batch(1) + .await + .unwrap_err() + .to_string() + .contains("stale deletion lease")); + assert!(f.exists(remaining).await); + assert!(restarted + .entity_deletion_store() + .heartbeat(&f.lease, Duration::from_secs(600)) + .await + .is_err()); + let mut exhausted = false; + for _ in 0..3 { + if handle( + &restarted, + &successor, + &f.stage, + &f.relay.public_key().to_bytes(), + 1, + ) + .await + .unwrap() + .exhausted + { + exhausted = true; + break; + } + } + assert!(exhausted); + assert!(!f.exists(&first).await && !f.exists(&second).await && !f.exists(&reaction).await); + assert_eq!( + f.db.get_thread_summary(f.community, root.id.as_bytes()) + .await + .unwrap() + .unwrap() + .reply_count, + 0 + ); + let (status, current): (String, bool) = + sqlx::query_as("SELECT status,is_current FROM entity_deletion_stage_history WHERE id=$1") + .bind(f.stage.id) + .fetch_one(f.db.pool()) + .await + .unwrap(); + assert_eq!(status, "completed"); + assert!(current); +} From 644b7b599646318c0462035b48deae230ded2f38 Mon Sep 17 00:00:00 2001 From: Ravneet Arora Date: Fri, 30 Oct 2026 11:44:22 -0400 Subject: [PATCH 3/3] Repair every thread affected by user deletion Signed-off-by: Ravneet Arora --- .../store/user_deletion/messages/cleanup.rs | 41 +++- .../store/user_deletion/messages/select.sql | 4 +- .../buzz-deletion/src/user/messages/README.md | 35 ++- .../src/user/messages/postgres_tests.rs | 3 +- .../thread_repair_postgres_tests.rs | 209 ++++++++++++++++++ 5 files changed, 280 insertions(+), 12 deletions(-) create mode 100644 crates/buzz-deletion/src/user/messages/postgres_tests/thread_repair_postgres_tests.rs diff --git a/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs b/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs index ae13d168ef6..999ac5d913f 100644 --- a/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs +++ b/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs @@ -1,5 +1,7 @@ use super::*; +const MAX_AFFECTED_THREADS: usize = 32; + pub(super) async fn remove( tx: &mut AdmittedTx, community: CommunityId, @@ -62,6 +64,28 @@ pub(super) async fn remove( ids.dedup(); } + // A vote/edit's target can differ from its stored reply parent. Capture + // every affected stored thread before deleting its source event, plus the + // seed for legacy events without metadata. Repair all of them atomically. + let mut threads: Vec<(Option, Vec)> = sqlx::query_as( + "SELECT DISTINCT channel_id,COALESCE(root_event_id,event_id) AS root_id + FROM thread_metadata WHERE community_id=$1 AND event_id=ANY($2) + ORDER BY channel_id,root_id LIMIT $3", + ) + .bind(community.as_uuid()) + .bind(&ids) + .bind((MAX_AFFECTED_THREADS + 1) as i64) + .fetch_all(tx.conn()) + .await?; + threads.push((thread.channel_id, thread.root_id.clone())); + threads.sort(); + threads.dedup(); + if threads.len() > MAX_AFFECTED_THREADS { + return Err(safety( + "too many affected threads in one message batch; reduce the batch size", + )); + } + // A worker may already have fetched the payload even though the user is // fenced. Do not delete its recovery record while it can still deliver. let delivering: bool = sqlx::query_scalar( @@ -130,13 +154,17 @@ pub(super) async fn remove( .execute(tx.conn()) .await? .rows_affected(); - repair_thread(tx, community, thread).await + for (channel, root) in threads { + repair_thread(tx, community, channel, &root).await?; + } + Ok(()) } async fn repair_thread( tx: &mut AdmittedTx, community: CommunityId, - thread: &Candidate, + channel: Option, + root: &[u8], ) -> Result<()> { // Preserve only the ancestry needed by retained events. The missing event // rows carry no content, author, tags or signature; metadata keeps replies @@ -153,9 +181,10 @@ async fn repair_thread( ) DELETE FROM thread_metadata tm WHERE tm.community_id=$1 AND tm.channel_id=$2 AND (tm.event_id=$3 OR tm.root_event_id=$3) AND NOT EXISTS (SELECT 1 FROM needed n WHERE n.event_id=tm.event_id AND n.event_created_at=tm.event_created_at)", - ).bind(community.as_uuid()).bind(thread.channel_id).bind(&thread.root_id).execute(tx.conn()).await?; + ).bind(community.as_uuid()).bind(channel).bind(root).execute(tx.conn()).await?; // Recompute, never subtract old counts. Already-soft-deleted messages do // not count; received_at matches the server-arrival activity clock. + // Like live writes, last_reply_at tracks direct replies, not descendants. sqlx::query( "WITH visible AS MATERIALIZED ( SELECT tm.parent_event_id,tm.root_event_id,e.received_at FROM thread_metadata tm @@ -164,16 +193,16 @@ async fn repair_thread( ), direct AS ( SELECT parent_event_id,count(*)::int AS n,max(received_at) AS last_at FROM visible GROUP BY parent_event_id ), descendants AS ( - SELECT root_event_id,count(*)::int AS n,max(received_at) AS last_at FROM visible GROUP BY root_event_id + SELECT root_event_id,count(*)::int AS n FROM visible GROUP BY root_event_id ), counts AS ( SELECT tm.event_id,tm.event_created_at,COALESCE(d.n,0) AS replies,COALESCE(r.n,0) AS descendants, - GREATEST(d.last_at,r.last_at) AS last_at + d.last_at FROM thread_metadata tm LEFT JOIN direct d ON d.parent_event_id=tm.event_id LEFT JOIN descendants r ON r.root_event_id=tm.event_id WHERE tm.community_id=$1 AND tm.channel_id=$2 AND (tm.event_id=$3 OR tm.root_event_id=$3) ) UPDATE thread_metadata tm SET reply_count=c.replies,descendant_count=c.descendants,last_reply_at=c.last_at FROM counts c WHERE tm.community_id=$1 AND tm.event_id=c.event_id AND tm.event_created_at=c.event_created_at", - ).bind(community.as_uuid()).bind(thread.channel_id).bind(&thread.root_id).execute(tx.conn()).await?; + ).bind(community.as_uuid()).bind(channel).bind(root).execute(tx.conn()).await?; Ok(()) } diff --git a/crates/buzz-db/src/store/user_deletion/messages/select.sql b/crates/buzz-db/src/store/user_deletion/messages/select.sql index f14dacde1fa..e058c926084 100644 --- a/crates/buzz-db/src/store/user_deletion/messages/select.sql +++ b/crates/buzz-db/src/store/user_deletion/messages/select.sql @@ -6,9 +6,9 @@ WITH authored AS ( WHERE e.community_id = $1 AND e.pubkey IN ($2, $3) AND e.kind IN (9, 40001, 40002, 45001, 45003, 7, 45002, 40003, 40004, 40005, 40006, 40007, 40008) AND CASE WHEN e.pubkey = $3 THEN COALESCE( - -- Match effective_message_author: first actor, then first valid p. + -- Match effective_message_author: first actor with a value, then first valid p. (SELECT CASE WHEN t->>1 ~ '^[0-9a-fA-F]{64}$' THEN decode(t->>1, 'hex') END - FROM jsonb_array_elements(e.tags) t WHERE t->>0 = 'actor' LIMIT 1), + FROM jsonb_array_elements(e.tags) t WHERE t->>0 = 'actor' AND t->>1 IS NOT NULL LIMIT 1), (SELECT decode(t->>1, 'hex') FROM jsonb_array_elements(e.tags) t WHERE t->>0 = 'p' AND t->>1 ~ '^[0-9a-fA-F]{64}$' LIMIT 1), e.pubkey) ELSE e.pubkey END = $2 diff --git a/crates/buzz-deletion/src/user/messages/README.md b/crates/buzz-deletion/src/user/messages/README.md index 484711ba355..6dd00995307 100644 --- a/crates/buzz-deletion/src/user/messages/README.md +++ b/crates/buzz-deletion/src/user/messages/README.md @@ -41,7 +41,8 @@ heartbeat waiting on that lock cannot revive a lease that has expired. Expiry rolls back the batch. If the lease has actually expired, the old worker also cannot save a failure: the current running stage remains durable and a new worker can reclaim it with a higher generation. Reduce batch size if repeated -expiry prevents progress; this handler does not impose a new lease duration. +expiry prevents progress, but note the scan limitation below; this handler +does not impose a new lease duration. ## What one batch removes @@ -51,7 +52,8 @@ expiry prevents progress; this handler does not impose a new lease duration. Old soft-deleted rows are included. Community-global private reminders, personal lists and other account state belong to later handlers. - Relay-signed attributed messages only when the signer is the trusted relay - key. Attribution follows the existing first-actor/first-valid-p rules. An + key. Attribution uses the first actor tag with a value, skipping valueless + actor tags, then falls back to the first valid p tag if that value is invalid. An arbitrary author's actor or mention tags never confer authorship. - Dependent edits, votes and reactions on removed events, including reactions from other people. The first valid e tag selects an edit/vote target; the @@ -70,6 +72,12 @@ past 16 levels, never commits an incomplete traversal. A smaller batch can resolve a fan-out limit; a single over-limit message requires further work. Only IDs and flags are loaded, not message bodies or workflow traces. +Before deleting events, collect their stored thread roots as well as the selected +thread. A vote or edit can target one post while its stored reply parent belongs +to another thread. Both threads must be repaired. At most 32 distinct threads +are repaired in a batch; exceeding that limit fails before deleting anything. +A single event that exceeds the limit requires further work, even at batch size 1. + All deletions and thread repairs commit together. Any SQL failure or lease expiry rolls back the batch. Re-running after a successful commit cannot subtract counts twice. A final bounded sweep removes orphan materialized @@ -81,7 +89,9 @@ author when their source event is already missing; active runs block completion. Thread metadata retains only ancestry needed by remaining events. It has no message body, tags, author or signature. Counts are recomputed from visible -remaining messages using the server-arrival clock. Thread-window reads recognize +remaining messages using the server-arrival clock. `last_reply_at` is the +latest direct reply, matching normal writes; a newer nested reply only changes +its immediate parent's timestamp. Thread-window reads recognize a missing root through trusted metadata and still return original signed replies. No fake signed tombstone is manufactured. Posting new replies to a physically missing parent and client refresh/notification behavior are separate @@ -105,6 +115,21 @@ This deliberately also blocks some ordinary links. The source events remain avai once file cleanup exists; the handler never drops file references and reports success. This PR does not add a bypass or a pretend storage adapter. +## Performance to measure before integration + +The batch size bounds deleted authored events, not all database work. Candidate +selection can scan many authored events and inspect their tags; repairing a +thread scans its remaining messages. Even a batch size of one can time out for +a large account or thread. The 32-thread limit bounds the number of repairs, +not the size of each repair. Repeated retries may therefore make no progress. + +Before integrating this handler, measure query plans and batch duration with +large accounts, large threads and dependent events spread across threads. +Compare these timings with statement timeouts and the caller's lease budget. +These workloads have not been benchmarked here; smaller batches alone are not +a demonstrated solution. Query optimization or different batching should be +based on those measurements. + ## Local tests Use the repository's isolated PostgreSQL lane: @@ -126,3 +151,7 @@ asserts that source/dependent records survive and retry state is saved. A replacement-worker case expires the lease between committed batches, reclaims through the real API, rejects the old worker/heartbeat, and finishes with the new generation. This does not simulate heartbeat contention inside a live batch. + +Regression cases also cover votes/edits stored under a different thread, +rollback during another thread's repair, the affected-thread limit, direct-reply +timestamps and valueless actor tags. diff --git a/crates/buzz-deletion/src/user/messages/postgres_tests.rs b/crates/buzz-deletion/src/user/messages/postgres_tests.rs index 43582d590c8..d87ef35126d 100644 --- a/crates/buzz-deletion/src/user/messages/postgres_tests.rs +++ b/crates/buzz-deletion/src/user/messages/postgres_tests.rs @@ -11,6 +11,7 @@ use std::time::Duration; use uuid::Uuid; mod safety_postgres_tests; +mod thread_repair_postgres_tests; struct Fixture { db: Db, @@ -129,7 +130,7 @@ impl Fixture { .tags(tags) .sign_with_keys(who) .unwrap(); - let metadata = if [9, 40002, 45001, 45003].contains(&kind) { + let metadata = if thread.is_some() || [9, 40002, 45001, 45003].contains(&kind) { Some(ThreadMetadataParams { event_id: event.id.as_bytes(), event_created_at: timestamp(&event), diff --git a/crates/buzz-deletion/src/user/messages/postgres_tests/thread_repair_postgres_tests.rs b/crates/buzz-deletion/src/user/messages/postgres_tests/thread_repair_postgres_tests.rs new file mode 100644 index 00000000000..5be8ebbc4d3 --- /dev/null +++ b/crates/buzz-deletion/src/user/messages/postgres_tests/thread_repair_postgres_tests.rs @@ -0,0 +1,209 @@ +use super::*; + +async fn counts(f: &Fixture, event: &Event) -> (i32, i32, Option>) { + sqlx::query_as("SELECT reply_count,descendant_count,last_reply_at FROM thread_metadata WHERE community_id=$1 AND event_id=$2") + .bind(f.community.as_uuid()).bind(event.id.as_bytes().as_slice()) + .fetch_one(f.db.pool()).await.unwrap() +} + +async fn metadata_exists(f: &Fixture, event: &Event) -> bool { + sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM thread_metadata WHERE community_id=$1 AND event_id=$2)", + ) + .bind(f.community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(f.db.pool()) + .await + .unwrap() +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn dependent_vote_repairs_its_stored_thread_not_only_its_target() { + let f = fixture().await; + let target = f + .event(&f.alice, 45001, "delete this post", vec![], None) + .await; + let other = f.event(&f.bob, 45001, "keep this post", vec![], None).await; + let vote = f + .event( + &f.bob, + 45002, + "+", + vec![tag(&["e", &target.id.to_hex()])], + Some((&other, &other, 1)), + ) + .await; + assert_eq!(counts(&f, &other).await.0, 1); + assert_eq!(f.batch(1).await.unwrap().events_deleted, 2); + assert!(!f.exists(&target).await && !f.exists(&vote).await); + assert!(f.exists(&other).await); + assert!(!metadata_exists(&f, &vote).await); + assert_eq!(counts(&f, &other).await, (0, 0, None)); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn authored_edit_and_vote_repair_their_stored_thread() { + for kind in [40003, 45002] { + let f = fixture().await; + let target = f.event(&f.bob, 45001, "target", vec![], None).await; + let other = f.event(&f.bob, 45001, "stored thread", vec![], None).await; + let event = f + .event( + &f.alice, + kind, + "change", + vec![tag(&["e", &target.id.to_hex()])], + Some((&other, &other, 1)), + ) + .await; + assert_eq!(counts(&f, &other).await.0, 1); + assert_eq!(f.batch(1).await.unwrap().events_deleted, 1); + assert!(!f.exists(&event).await && !metadata_exists(&f, &event).await); + assert!(f.exists(&target).await && f.exists(&other).await); + assert_eq!(counts(&f, &other).await, (0, 0, None)); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn last_reply_time_uses_direct_replies_not_newer_nested_replies() { + let f = fixture().await; + let root = f.event(&f.bob, 9, "root", vec![], None).await; + let direct = f + .event(&f.bob, 9, "direct", vec![], Some((&root, &root, 1))) + .await; + let nested = f + .event(&f.bob, 9, "nested", vec![], Some((&root, &direct, 2))) + .await; + let removed = f + .event(&f.alice, 9, "remove", vec![], Some((&root, &root, 1))) + .await; + let earlier = DateTime::from_timestamp(1_700_000_000, 0).unwrap(); + let later = DateTime::from_timestamp(1_700_000_100, 0).unwrap(); + for (event, received) in [(&direct, earlier), (&nested, later)] { + sqlx::query("UPDATE events SET received_at=$3 WHERE community_id=$1 AND id=$2") + .bind(f.community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .bind(received) + .execute(f.db.pool()) + .await + .unwrap(); + } + f.batch(1).await.unwrap(); + assert!(!f.exists(&removed).await); + assert_eq!(counts(&f, &root).await, (1, 2, Some(earlier))); + assert_eq!(counts(&f, &direct).await, (1, 0, Some(later))); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn valueless_actor_tags_are_skipped_before_using_p_fallback() { + let f = fixture().await; + let removed = f + .event( + &f.relay, + 9, + "belongs to subject", + vec![ + tag(&["actor"]), + tag(&["actor", &f.alice.public_key().to_hex()]), + tag(&["p", &f.bob.public_key().to_hex()]), + ], + None, + ) + .await; + let retained = f + .event( + &f.relay, + 9, + "belongs to other", + vec![ + tag(&["actor"]), + tag(&["actor", &f.bob.public_key().to_hex()]), + tag(&["p", &f.alice.public_key().to_hex()]), + ], + None, + ) + .await; + f.finish(1).await; + assert!(!f.exists(&removed).await); + assert!(f.exists(&retained).await); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn affected_thread_limit_preserves_every_source_and_saves_retry() { + let f = fixture().await; + let target = f.event(&f.alice, 45001, "target", vec![], None).await; + let mut dependents = Vec::new(); + // The seed plus 32 different stored roots exceeds the 32-thread limit. + for i in 0..32 { + let root = f + .event(&f.bob, 45001, &format!("root {i}"), vec![], None) + .await; + let vote = f + .event( + &f.bob, + 45002, + "+", + vec![tag(&["e", &target.id.to_hex()])], + Some((&root, &root, 1)), + ) + .await; + dependents.push((root, vote)); + } + let error = f.batch(1).await.unwrap_err(); + assert!( + error.to_string().contains("too many affected threads"), + "{error}" + ); + assert!(f.exists(&target).await); + for (root, vote) in dependents { + assert!(f.exists(&root).await && f.exists(&vote).await); + assert!(metadata_exists(&f, &vote).await); + assert_eq!(counts(&f, &root).await.0, 1); + } + let status: (String, String) = sqlx::query_as("SELECT r.status,s.status FROM entity_deletion_requests r JOIN entity_deletion_stage_history s ON s.request_id=r.id WHERE s.id=$1") + .bind(f.stage.id).fetch_one(f.db.pool()).await.unwrap(); + assert_eq!(status, ("retry_wait".into(), "failed".into())); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn failure_repairing_another_thread_rolls_back_the_entire_batch() { + let f = fixture().await; + let target = f.event(&f.alice, 45001, "target", vec![], None).await; + let other = f.event(&f.bob, 45001, "other", vec![], None).await; + let vote = f + .event( + &f.bob, + 45002, + "+", + vec![tag(&["e", &target.id.to_hex()])], + Some((&other, &other, 1)), + ) + .await; + let trigger = format!("fail_other_thread_{}", Uuid::new_v4().simple()); + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "CREATE FUNCTION {trigger}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'injected other thread repair failure'; END $$; + CREATE TRIGGER {trigger} AFTER UPDATE ON thread_metadata FOR EACH ROW + WHEN (NEW.community_id='{}' AND NEW.event_id=decode('{}','hex')) EXECUTE FUNCTION {trigger}();", + f.community.as_uuid(), other.id.to_hex(), + ))).execute(f.db.pool()).await.unwrap(); + let outcome = f.batch(1).await; + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "DROP TRIGGER {trigger} ON thread_metadata; DROP FUNCTION {trigger}();" + ))) + .execute(f.db.pool()) + .await + .unwrap(); + assert!(outcome + .unwrap_err() + .to_string() + .contains("injected other thread")); + assert!(f.exists(&target).await && f.exists(&vote).await && f.exists(&other).await); + assert!(metadata_exists(&f, &target).await && metadata_exists(&f, &vote).await); + assert_eq!(counts(&f, &other).await.0, 1); +}