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..999ac5d913f --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/messages/cleanup.rs @@ -0,0 +1,276 @@ +use super::*; + +const MAX_AFFECTED_THREADS: usize = 32; + +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','emoji'))) 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 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( + "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(); + for (channel, root) in threads { + repair_thread(tx, community, channel, &root).await?; + } + Ok(()) +} + +async fn repair_thread( + tx: &mut AdmittedTx, + community: CommunityId, + 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 + // 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(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 + 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 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, + 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(channel).bind(root).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..e058c926084 --- /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 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' 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 +), 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', 'emoji') + )) 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..6dd00995307 --- /dev/null +++ b/crates/buzz-deletion/src/user/messages/README.md @@ -0,0 +1,157 @@ +# 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. + +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, but note the scan limitation below; this handler +does not impose a new lease duration. + +## 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 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 + 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. + +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 +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. `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 +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/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. + +## 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: + +```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. + +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. + +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 new file mode 100644 index 00000000000..d87ef35126d --- /dev/null +++ b/crates/buzz-deletion/src/user/messages/postgres_tests.rs @@ -0,0 +1,849 @@ +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; + +mod safety_postgres_tests; +mod thread_repair_postgres_tests; + +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 thread.is_some() || [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 + ); +} 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); +} 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); +}