From cb971e0e2b5060581a63464c6286cc2d6b528c88 Mon Sep 17 00:00:00 2001 From: Ravneet Arora Date: Tue, 10 Nov 2026 14:50:49 -0500 Subject: [PATCH] Add read-only user cleanup batch preview Print ordered cleanup batches for a public key and community using a read-only database snapshot. Keep this diagnostic path separate from deletion execution. Co-authored-by: maincourse <37ce551b493d4a998caf030929c47bce0b6a0e43f1178b3a63c4151d7c16b76e@buzz.block.builderlab.xyz> Signed-off-by: Ravneet Arora --- crates/buzz-admin/src/user_deletions.rs | 27 ++ crates/buzz-db/src/store/user_deletion/mod.rs | 2 + .../store/user_deletion/preview/authored.sql | 12 + .../src/store/user_deletion/preview/mod.rs | 373 +++++++++++++++ .../user_deletion/preview/postgres_tests.rs | 434 ++++++++++++++++++ .../src/store/user_deletion/preview/rows.rs | 168 +++++++ crates/buzz-deletion/src/user.rs | 44 ++ crates/buzz-deletion/src/user/preview.rs | 27 ++ docs/user-content-batch-preview.md | 95 ++++ 9 files changed, 1182 insertions(+) create mode 100644 crates/buzz-db/src/store/user_deletion/preview/authored.sql create mode 100644 crates/buzz-db/src/store/user_deletion/preview/mod.rs create mode 100644 crates/buzz-db/src/store/user_deletion/preview/postgres_tests.rs create mode 100644 crates/buzz-db/src/store/user_deletion/preview/rows.rs create mode 100644 crates/buzz-deletion/src/user/preview.rs create mode 100644 docs/user-content-batch-preview.md diff --git a/crates/buzz-admin/src/user_deletions.rs b/crates/buzz-admin/src/user_deletions.rs index 594f9189b37..f8ecdbd4298 100644 --- a/crates/buzz-admin/src/user_deletions.rs +++ b/crates/buzz-admin/src/user_deletions.rs @@ -11,6 +11,33 @@ pub async fn run(command: UserDeletionsCommand) -> anyhow::Result { mod tests { use clap::Parser; + #[test] + fn batch_preview_requires_read_only_scope_and_valid_keys() { + let args = [ + "buzz-admin", + "user-deletions", + "preview-batches", + "--pubkey", + "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + "--community", + "00000000-0000-0000-0000-000000000001", + "--read-only", + ]; + assert!(crate::Cli::try_parse_from(args).is_ok()); + assert!(crate::Cli::try_parse_from(&args[..7]).is_err()); + for extra in [ + ["--batch-size", "0"], + ["--batch-size", "101"], + ["--pubkey", "demo"], + ["--max-items", "0"], + ["--max-items", "1000001"], + ["--relay-pubkey", "invalid"], + ] { + assert!(crate::Cli::try_parse_from(args.into_iter().chain(extra)).is_err()); + } + assert!(crate::Cli::try_parse_from(args.into_iter().chain(["--execute"])).is_err()); + } + #[test] fn user_workflow_commands_are_registered() { for args in [ diff --git a/crates/buzz-db/src/store/user_deletion/mod.rs b/crates/buzz-db/src/store/user_deletion/mod.rs index 1c093946945..4eef5ab37ea 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; +/// Read-only, ordered previews of human content cleanup batches. +pub mod preview; mod requests; pub use discovery::normalize_user_deletion_pubkey; pub use inventory::UserDeletionInventory; diff --git a/crates/buzz-db/src/store/user_deletion/preview/authored.sql b/crates/buzz-db/src/store/user_deletion/preview/authored.sql new file mode 100644 index 00000000000..484fa2218b4 --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/preview/authored.sql @@ -0,0 +1,12 @@ +-- Scope shared by source selection and the excluded-kind inventory. +-- Relay attribution is opt-in: only the supplied trusted signer may use actor/p. +SELECT e.* FROM events e +WHERE e.community_id = $1 AND e.pubkey IN (decode($2, 'hex'), decode($3, 'hex')) + AND CASE WHEN e.pubkey = decode($3, 'hex') THEN COALESCE( + (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 = decode($2, 'hex') + -- Notes/articles use direct authorship, not relay attribution. + AND (e.kind NOT IN (1, 30023) OR e.pubkey = decode($2, 'hex')) diff --git a/crates/buzz-db/src/store/user_deletion/preview/mod.rs b/crates/buzz-db/src/store/user_deletion/preview/mod.rs new file mode 100644 index 00000000000..98ef19c86d8 --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/preview/mod.rs @@ -0,0 +1,373 @@ +//! Observational batch planning. No mutation executor or durable progress exists here. +//! +//! A repeatable-read, read-only snapshot supplies rows. An in-memory set models +//! rows already emitted, so parents can follow children without deleting either. + +use super::{normalize_user_deletion_pubkey, UserDeletionStore}; +use crate::{DbError, Result}; +use chrono::{DateTime, Utc}; +use serde::Serialize; +use serde_json::{json, Value}; +use sqlx::PgConnection; +use std::collections::HashSet; +use std::io::Write; +use uuid::Uuid; + +#[cfg(test)] +mod postgres_tests; +mod rows; + +const KINDS: &[i32] = &[ + 1, 7, 9, 30023, 40001, 40002, 40003, 40004, 40005, 40006, 40007, 40008, 45001, 45002, 45003, +]; +const MAX_DEPTH: usize = 64; +const PAGE_SIZE: i64 = 100; +const EVENT_COLUMNS: &str = "e.id,e.created_at,e.channel_id,e.kind, + (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"; + +/// Input to the standalone preview; never authorizes a deletion. +pub struct PreviewOptions { + /// Explicit tenant scope, including content from former memberships. + pub community: Uuid, + /// Human public key, as npub or hex. + pub pubkey: String, + /// Optional trusted relay public key for historical actor/p attribution. + pub relay_pubkey: Option, + /// Maximum explicit row operations in each printed batch (1–100). + pub batch_size: u16, + /// Maximum unique operations retained in the simulation (1–1,000,000). + /// Exceeding this fails visibly; it never returns a complete partial plan. + pub max_items: u32, +} + +/// Final marker proving that all batches in the declared scope were printed. +#[derive(Debug, Serialize)] +pub struct PreviewSummary { + /// Always `preview_complete`; absent if the preview fails or is interrupted. + pub record: &'static str, + /// Number of batches printed. + pub batches: u64, + /// Number of unique proposed row operations. + pub items: usize, + /// Operations carrying a media, active-work or delivery warning. + pub warnings: usize, + /// Authored event kinds outside this handler, with counts. + pub excluded_event_kinds: Vec, +} + +#[derive(Clone, sqlx::FromRow)] +struct Event { + id: Vec, + created_at: DateTime, + channel_id: Option, + kind: i32, + has_media: bool, +} + +#[derive(Default, Clone)] +struct Cursor(Option>, Option>); + +impl Cursor { + fn advance(&mut self, event: &Event) { + self.0 = Some(event.created_at); + self.1 = Some(event.id.clone()); + } +} + +struct Frame { + event: Event, + children: Cursor, + reason: Value, +} + +struct Output<'a, W> { + writer: &'a mut W, + community: Uuid, + limit: usize, + max_items: usize, + seen: HashSet, + events: HashSet<(Vec, DateTime)>, + runs: HashSet, + batch: Vec, + repairs: Vec, + batches: u64, + warnings: usize, +} + +fn invalid(message: impl Into) -> DbError { + DbError::DeletionSafety(message.into()) +} + +fn line(writer: &mut impl Write, value: &impl Serialize) -> Result<()> { + serde_json::to_writer(&mut *writer, value)?; + writer + .write_all(b"\n") + .and_then(|_| writer.flush()) + .map_err(|error| invalid(format!("preview output failed: {error}"))) +} + +async fn snapshot(pool: &sqlx::PgPool) -> Result> { + let mut tx = pool.begin().await?; + sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY") + .execute(&mut *tx) + .await?; + sqlx::query("SET LOCAL statement_timeout = '30s'") + .execute(&mut *tx) + .await?; + sqlx::query("SET LOCAL lock_timeout = '3s'") + .execute(&mut *tx) + .await?; + sqlx::query("SET LOCAL TIME ZONE 'UTC'") + .execute(&mut *tx) + .await?; + Ok(tx) +} + +impl Output<'_, W> { + fn add( + &mut self, + table: &str, + key: Value, + action: &str, + reason: &Value, + warning: Option, + ) -> Result<()> { + let identity = format!("{table}:{key}"); + if self.seen.contains(&identity) { + return Ok(()); + } + if self.seen.len() >= self.max_items { + return Err(invalid( + "preview item limit exceeded; output is incomplete; increase --max-items", + )); + } + self.seen.insert(identity); + self.warnings += usize::from(warning.is_some()); + self.batch.push(json!({ + "table": table, "key": key, "action": action, + "reason": reason, "warning": warning, + })); + if self.batch.len() == self.limit { + self.flush()?; + } + Ok(()) + } + + fn flush(&mut self) -> Result<()> { + if self.batch.is_empty() { + return Ok(()); + } + self.batches += 1; + line( + self.writer, + &json!({ + "record": "batch", "number": self.batches, "community_id": self.community, + "items": self.batch, "thread_repairs": self.repairs, + }), + )?; + self.batch.clear(); + self.repairs.clear(); + Ok(()) + } +} + +impl UserDeletionStore { + /// Print JSON lines for every proposed batch in one read-only snapshot. + /// + /// This does not save a request, acquire a worker lease, update counters, + /// migrate the schema, or delete rows. The completion marker describes only + /// the stated handler scope; it is not evidence that deletion is safe to run. + pub async fn preview_message_batches( + &self, + options: &PreviewOptions, + writer: &mut impl Write, + ) -> Result { + if !(1..=100).contains(&options.batch_size) || !(1..=1_000_000).contains(&options.max_items) + { + return Err(invalid( + "batch size must be 1–100 and max items must be 1–1,000,000", + )); + } + let subject = normalize_user_deletion_pubkey(&options.pubkey)?; + let relay = options + .relay_pubkey + .as_deref() + .map(normalize_user_deletion_pubkey) + .transpose()?; + if relay.as_deref() == Some(&subject) { + return Err(invalid( + "the human and trusted relay public keys must differ", + )); + } + let mut tx = snapshot(&self.pool).await?; + let host: String = sqlx::query_scalar("SELECT host FROM communities WHERE id=$1") + .bind(options.community) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| invalid("unknown community"))?; + line( + writer, + &json!({ + "record": "preview_start", "version": 1, "read_only": true, + "community_id": options.community, "host": host, "pubkey": subject, + "trusted_relay_pubkey": relay, "included_event_kinds": KINDS, + "batch_size": options.batch_size, "max_items": options.max_items, + "scope": "human content events, permitted dependents, reaction rows and workflow copies", + "relay_attribution": if relay.is_some() { "enabled" } else { "not included; supply --relay-pubkey for relay-authored history" }, + "execution": "unavailable; media, worker drain, write fences and thread repair require separate validation", + "thread_repairs": "reported separately from row items; metadata mutations are not expanded", + }), + )?; + let mut output = Output { + writer, + community: options.community, + limit: usize::from(options.batch_size), + max_items: options.max_items as usize, + seen: HashSet::new(), + events: HashSet::new(), + runs: HashSet::new(), + batch: Vec::new(), + repairs: Vec::new(), + batches: 0, + warnings: 0, + }; + let mut cursor = Cursor::default(); + loop { + let query = format!( + "WITH authored AS ({}) SELECT {EVENT_COLUMNS} FROM authored e + WHERE e.kind=ANY($4) AND ($5::timestamptz IS NULL OR (e.created_at,e.id)>($5,$6)) + ORDER BY e.created_at,e.id LIMIT 100", + include_str!("authored.sql") + ); + // Only static SQL fragments are interpolated; every value is bound. + let events: Vec = sqlx::query_as(sqlx::AssertSqlSafe(query)) + .bind(options.community) + .bind(&subject) + .bind(&relay) + .bind(KINDS) + .bind(cursor.0) + .bind(&cursor.1) + .fetch_all(&mut *tx) + .await?; + if events.is_empty() { + break; + } + for event in events { + cursor.advance(&event); + walk(&mut tx, event, &mut output).await?; + } + } + rows::orphan_sweeps(&mut tx, &subject, &mut output).await?; + let query = format!( + "WITH authored AS ({}) SELECT jsonb_build_object('kind',kind,'count',count(*)) + FROM authored WHERE NOT(kind=ANY($4)) GROUP BY kind ORDER BY kind", + include_str!("authored.sql") + ); + let excluded: Vec = sqlx::query_scalar(sqlx::AssertSqlSafe(query)) + .bind(options.community) + .bind(&subject) + .bind(&relay) + .bind(KINDS) + .fetch_all(&mut *tx) + .await?; + output.flush()?; + let summary = PreviewSummary { + record: "preview_complete", + batches: output.batches, + items: output.seen.len(), + warnings: output.warnings, + excluded_event_kinds: excluded, + }; + // End the observational snapshot before claiming complete output. + tx.rollback().await?; + line(output.writer, &summary)?; + Ok(summary) + } +} + +async fn walk( + conn: &mut PgConnection, + seed: Event, + output: &mut Output<'_, impl Write>, +) -> Result<()> { + if output.events.contains(&(seed.id.clone(), seed.created_at)) { + return Ok(()); + } + let mut stack = vec![Frame { + event: seed, + children: Cursor::default(), + reason: json!({"type": "authored_event"}), + }]; + while let Some(frame) = stack.last_mut() { + let query = format!("SELECT {EVENT_COLUMNS} 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=$2 + AND e.channel_id IS NOT DISTINCT FROM $3::uuid + AND ($4::timestamptz IS NULL OR (e.created_at,e.id)>($4,$5)) + ORDER BY e.created_at,e.id LIMIT 1"); + let child: Option = sqlx::query_as(sqlx::AssertSqlSafe(query)) + .bind(output.community) + .bind(&frame.event.id) + .bind(frame.event.channel_id) + .bind(frame.children.0) + .bind(&frame.children.1) + .fetch_optional(&mut *conn) + .await?; + if let Some(child) = child { + frame.children.advance(&child); + let reason = + json!({"type": "dependent_event", "target_event_id": hex::encode(&frame.event.id)}); + if stack.iter().any(|f| f.event.id == child.id) { + return Err(invalid("cyclic event dependencies; preview is incomplete")); + } + if output + .events + .contains(&(child.id.clone(), child.created_at)) + { + continue; + } + if stack.len() >= MAX_DEPTH { + return Err(invalid( + "event dependency depth exceeds 64; preview is incomplete", + )); + } + stack.push(Frame { + event: child, + children: Cursor::default(), + reason, + }); + } else { + let frame = stack + .pop() + .ok_or_else(|| invalid("missing preview frame"))?; + rows::event_rows(conn, &frame.event, output).await?; + let roots: Vec = sqlx::query_scalar("SELECT jsonb_build_object( + 'channel_id',channel_id,'root_event_id',encode(COALESCE(root_event_id,event_id),'hex')) + FROM thread_metadata WHERE community_id=$1 AND event_id=$2 AND event_created_at=$3 + ORDER BY channel_id,root_event_id") + .bind(output.community).bind(&frame.event.id).bind(frame.event.created_at) + .fetch_all(&mut *conn).await?; + output.repairs.push(json!({"event_id": hex::encode(&frame.event.id), + "channel_id": frame.event.channel_id, "stored_threads": roots, + "action": "recompute affected thread metadata; preserve surviving reply ancestry"})); + output.add( + "events", + json!({"id": hex::encode(&frame.event.id), "created_at": frame.event.created_at}), + "would_delete", + &json!({"selection": frame.reason, "kind": frame.event.kind}), + frame + .event + .has_media + .then(|| "media references require prior cleanup".to_owned()), + )?; + output + .events + .insert((frame.event.id, frame.event.created_at)); + } + } + Ok(()) +} diff --git a/crates/buzz-db/src/store/user_deletion/preview/postgres_tests.rs b/crates/buzz-db/src/store/user_deletion/preview/postgres_tests.rs new file mode 100644 index 00000000000..2da77d4f13c --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/preview/postgres_tests.rs @@ -0,0 +1,434 @@ +use super::*; +use crate::Db; +use nostr::ToBech32; +use sqlx::PgPool; + +const HUMAN: &str = "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798"; +const OTHER: &str = "c6047f9441ed7d6d3045406e95c07cd85c778e4b8cef3ca7abac09b95c709ee5"; +const RELAY: &str = "f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9"; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn production_snapshot_rejects_writes_and_does_not_see_later_inserts() { + let (pool, options) = setup().await; + let mut tx = super::snapshot(&pool).await.unwrap(); + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1") + .bind(options.community) + .fetch_one(&mut *tx) + .await + .unwrap(); + assert_eq!(count, 0); + event(&pool, options.community, 1, HUMAN, 9, json!([])).await; + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1") + .bind(options.community) + .fetch_one(&mut *tx) + .await + .unwrap(); + assert_eq!(count, 0); + let error = sqlx::query("UPDATE communities SET icon='unexpected write' WHERE id=$1") + .bind(options.community) + .execute(&mut *tx) + .await + .unwrap_err(); + assert_eq!( + error.as_database_error().unwrap().code().as_deref(), + Some("25006") + ); + tx.rollback().await.unwrap(); +} + +async fn setup() -> (PgPool, PreviewOptions) { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let community = Uuid::new_v4(); + sqlx::query("INSERT INTO communities(id,host) VALUES($1,$2)") + .bind(community) + .bind(format!("{community}.preview.example.org")) + .execute(&pool) + .await + .unwrap(); + ( + pool, + PreviewOptions { + community, + pubkey: HUMAN.to_owned(), + relay_pubkey: None, + batch_size: 100, + max_items: 10_000, + }, + ) +} + +fn id(n: u32) -> String { + format!("{n:064x}") +} + +async fn event(pool: &PgPool, c: Uuid, n: u32, author: &str, kind: i32, tags: Value) { + sqlx::query("INSERT INTO events(community_id,id,pubkey,created_at,kind,tags,content,sig) + VALUES($1,decode($2,'hex'),decode($3,'hex'),date_trunc('month',now())+make_interval(secs=>$4),$5,$6,'fixture content',decode(repeat('00',64),'hex'))") + .bind(c).bind(id(n)).bind(author).bind(f64::from(n)).bind(kind).bind(tags) + .execute(pool).await.unwrap(); +} + +async fn preview(pool: &PgPool, options: &PreviewOptions) -> (PreviewSummary, Vec) { + let mut bytes = Vec::new(); + let summary = Db::from_pool(pool.clone()) + .user_deletion_store() + .preview_message_batches(options, &mut bytes) + .await + .unwrap(); + let lines = String::from_utf8(bytes) + .unwrap() + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + (summary, lines) +} + +fn items(lines: &[Value]) -> Vec { + lines + .iter() + .filter_map(|line| line["items"].as_array()) + .flatten() + .cloned() + .collect() +} + +async fn snapshot(pool: &PgPool) -> Vec { + let mut values = Vec::new(); + for table in [ + "events", + "reactions", + "event_mentions", + "workflow_runs", + "workflow_approvals", + "scheduled_workflow_fires", + "delivery_log", + "push_match_queue", + "push_wake_outbox", + "operator_listener_outbox", + "thread_metadata", + "entity_deletion_requests", + "entity_deletion_stage_history", + "entity_deletion_approvals", + ] { + let query = format!("SELECT COALESCE(jsonb_agg(to_jsonb(t) ORDER BY to_jsonb(t)::text),'[]'::jsonb) FROM {table} t"); + values.push( + sqlx::query_scalar(sqlx::AssertSqlSafe(query)) + .fetch_one(pool) + .await + .unwrap(), + ); + } + values +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn mixed_batches_are_complete_ordered_deduplicated_and_read_only() { + let (pool, options) = setup().await; + let c = options.community; + event(&pool, c, 1, HUMAN, 9, json!([])).await; + event(&pool, c, 2, HUMAN, 40003, json!([["e", id(1)]])).await; + event(&pool, c, 3, OTHER, 7, json!([["e", id(2)], ["p", OTHER]])).await; + event(&pool, c, 4, OTHER, 9, json!([["e", id(1)]])).await; + event(&pool, c, 5, HUMAN, 7, json!([["e", id(4)], ["p", OTHER]])).await; + for (source, target, author) in [(3, 2, OTHER), (5, 4, HUMAN)] { + sqlx::query("INSERT INTO reactions(community_id,event_created_at,event_id,pubkey,emoji,reaction_event_id) + SELECT community_id,created_at,id,decode($3,'hex'),'👍',decode($4,'hex') FROM events WHERE community_id=$1 AND id=decode($2,'hex')") + .bind(c).bind(id(target)).bind(author).bind(id(source)).execute(&pool).await.unwrap(); + sqlx::query("INSERT INTO event_mentions(community_id,pubkey_hex,event_id,event_created_at,event_kind) + SELECT community_id,$3,id,created_at,kind FROM events WHERE community_id=$1 AND id=decode($2,'hex')") + .bind(c).bind(id(source)).bind(OTHER).execute(&pool).await.unwrap(); + } + // Multiple side-table pages, including a mention on Bob's retained reply. + sqlx::query("INSERT INTO event_mentions(community_id,pubkey_hex,event_id,event_created_at,event_kind) + SELECT e.community_id,lpad(to_hex(n),64,'0'),e.id,e.created_at,e.kind + FROM events e CROSS JOIN generate_series(1,205) n WHERE community_id=$1 AND id=decode($2,'hex')") + .bind(c).bind(id(1)).execute(&pool).await.unwrap(); + sqlx::query("INSERT INTO event_mentions(community_id,pubkey_hex,event_id,event_created_at) + SELECT community_id,$3,id,created_at FROM events WHERE community_id=$1 AND id=decode($2,'hex')") + .bind(c).bind(id(4)).bind(HUMAN).execute(&pool).await.unwrap(); + let before = snapshot(&pool).await; + let (summary, lines) = preview(&pool, &options).await; + let operations = items(&lines); + assert_eq!(summary.items, 213); // 4 events, 2 reactions, 207 mentions + assert_eq!(summary.batches, 3); + let sizes: Vec<_> = lines + .iter() + .filter_map(|v| v["items"].as_array().map(Vec::len)) + .collect(); + assert_eq!(sizes, [100, 100, 13]); + let deleted_events: Vec<_> = operations + .iter() + .filter(|v| v["table"] == "events") + .map(|v| v["key"]["id"].clone()) + .collect(); + assert_eq!(deleted_events, [id(3), id(2), id(1), id(5)]); + let position = |table: &str, field: &str, value: &str| { + operations + .iter() + .position(|v| v["table"] == table && v["key"][field] == value) + .unwrap() + }; + assert!(position("event_mentions", "event_id", &id(3)) < position("events", "id", &id(3))); + assert!(!operations + .iter() + .any(|v| v["table"] == "event_mentions" && v["key"]["event_id"] == id(4))); + assert_eq!(snapshot(&pool).await, before); + let (_, again) = preview(&pool, &options).await; + assert_eq!(again, lines); + assert_eq!(lines.last().unwrap()["record"], "preview_complete"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn sources_include_all_declared_kinds_and_history_but_preserve_other_authors_and_tenants() { + let (pool, mut options) = setup().await; + let (_, other_scope) = setup().await; + for (n, kind) in KINDS.iter().enumerate() { + event( + &pool, + options.community, + n as u32 + 1, + HUMAN, + *kind, + json!([]), + ) + .await; + event( + &pool, + other_scope.community, + n as u32 + 1, + HUMAN, + *kind, + json!([]), + ) + .await; + } + event( + &pool, + options.community, + 80, + HUMAN, + 30023, + json!([["d", "same-article"]]), + ) + .await; + event( + &pool, + options.community, + 81, + HUMAN, + 30023, + json!([["d", "same-article"]]), + ) + .await; + sqlx::query("UPDATE events SET deleted_at=now() WHERE community_id=$1 AND id=decode($2,'hex')") + .bind(options.community) + .bind(id(80)) + .execute(&pool) + .await + .unwrap(); + event(&pool, options.community, 90, HUMAN, 46020, json!([])).await; + event( + &pool, + options.community, + 91, + OTHER, + 9, + json!([["actor", HUMAN], ["p", HUMAN]]), + ) + .await; + event( + &pool, + options.community, + 92, + RELAY, + 9, + json!([["actor", HUMAN]]), + ) + .await; + options.relay_pubkey = Some(RELAY.to_owned()); + options.pubkey = nostr::PublicKey::parse(HUMAN).unwrap().to_bech32().unwrap(); + let (summary, lines) = preview(&pool, &options).await; + assert_eq!(summary.items, KINDS.len() + 3); + assert_eq!( + summary.excluded_event_kinds, + [json!({"kind":46020,"count":1})] + ); + assert!(items(&lines).iter().any(|v| v["key"]["id"] == id(80))); + assert!(!items(&lines).iter().any(|v| v["key"]["id"] == id(91))); + assert!(lines + .iter() + .filter(|v| v["record"] == "batch") + .all(|v| v["community_id"] == options.community.to_string())); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn authoritative_targets_and_channel_boundaries_control_recursion() { + let (pool, options) = setup().await; + let c = options.community; + event(&pool, c, 1, HUMAN, 9, json!([])).await; + event(&pool, c, 2, OTHER, 9, json!([])).await; + event(&pool, c, 3, OTHER, 7, json!([["e", id(2)], ["e", id(1)]])).await; // last + event( + &pool, + c, + 4, + OTHER, + 40003, + json!([["e", id(1)], ["e", id(2)]]), + ) + .await; // first + event( + &pool, + c, + 5, + OTHER, + 45002, + json!([["e", "bad"], ["e", id(1)]]), + ) + .await; + event(&pool, c, 6, OTHER, 7, json!([["e", id(1)], ["e", id(2)]])).await; // preserve + event( + &pool, + c, + 7, + OTHER, + 40003, + json!([["e", id(2)], ["e", id(1)]]), + ) + .await; + event(&pool, c, 8, OTHER, 7, json!([["e", id(1)]])).await; + sqlx::query("UPDATE events SET channel_id=$3 WHERE community_id=$1 AND id=decode($2,'hex')") + .bind(c) + .bind(id(8)) + .bind(Uuid::new_v4()) + .execute(&pool) + .await + .unwrap(); + let (_, lines) = preview(&pool, &options).await; + let selected: Vec<_> = items(&lines) + .into_iter() + .map(|v| v["key"]["id"].clone()) + .collect(); + assert_eq!(selected, [id(3), id(4), id(5), id(1)]); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn workflow_children_orphans_and_warnings_are_printed_without_writes() { + let (pool, options) = setup().await; + let c = options.community; + event( + &pool, + c, + 1, + HUMAN, + 9, + json!([["image", "https://example.org/photo"]]), + ) + .await; + sqlx::query("INSERT INTO users(community_id,pubkey) VALUES($1,decode($2,'hex'))") + .bind(c) + .bind(OTHER) + .execute(&pool) + .await + .unwrap(); + let workflow = Uuid::new_v4(); + sqlx::query( + "INSERT INTO workflows(community_id,id,name,owner_pubkey,definition,definition_hash) + VALUES($1,$2,'preview',decode($3,'hex'),'{}','\\x00')", + ) + .bind(c) + .bind(workflow) + .bind(OTHER) + .execute(&pool) + .await + .unwrap(); + for (run, event_id, status) in [ + (Uuid::from_u128(1), id(1), "completed"), + (Uuid::from_u128(2), id(999), "running"), + ] { + sqlx::query("INSERT INTO workflow_runs(community_id,id,workflow_id,status,trigger_event_id,trigger_context) + VALUES($1,$2,$3,$4::run_status,decode($5,'hex'),$6)") + .bind(c).bind(run).bind(workflow).bind(status).bind(event_id) + .bind(json!({"author": HUMAN})).execute(&pool).await.unwrap(); + } + sqlx::query("INSERT INTO workflow_approvals(community_id,token,workflow_id,run_id,step_id,step_index,approver_spec,expires_at) + SELECT $1,decode(lpad(to_hex(n),64,'0'),'hex'),$2,$3,'step',n,'owner',now()+interval '1 hour' FROM generate_series(1,205) n") + .bind(c).bind(workflow).bind(Uuid::from_u128(1)).execute(&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(c).bind(workflow).bind(Uuid::from_u128(1)).execute(&pool).await.unwrap(); + sqlx::query("INSERT INTO reactions(community_id,event_created_at,event_id,pubkey,emoji) VALUES($1,now(),decode($2,'hex'),decode($3,'hex'),'x')") + .bind(c).bind(id(999)).bind(HUMAN).execute(&pool).await.unwrap(); + sqlx::query("INSERT INTO push_match_queue(community_id,event_id,lease_until) VALUES($1,decode($2,'hex'),now()+interval '1 hour')") + .bind(c).bind(id(1)).execute(&pool).await.unwrap(); + let before = snapshot(&pool).await; + let (summary, lines) = preview(&pool, &options).await; + let operations = items(&lines); + assert_eq!(summary.items, 211); + assert_eq!(summary.warnings, 3); + assert!(operations[..205] + .iter() + .all(|v| v["table"] == "workflow_approvals")); + assert_eq!(operations[205]["action"], "would_clear_workflow_run_id"); + assert_eq!(operations[206]["table"], "workflow_runs"); + assert_eq!(operations.last().unwrap()["table"], "workflow_runs"); + assert_eq!(snapshot(&pool).await, before); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn cycles_limits_empty_scope_and_output_failures_never_report_false_completion() { + let (pool, mut options) = setup().await; + let (summary, _) = preview(&pool, &options).await; + assert_eq!(summary.items, 0); + event(&pool, options.community, 1, HUMAN, 7, json!([["e", id(2)]])).await; + event(&pool, options.community, 2, OTHER, 7, json!([["e", id(1)]])).await; + let store = Db::from_pool(pool.clone()).user_deletion_store(); + let mut bytes = Vec::new(); + assert!(store + .preview_message_batches(&options, &mut bytes) + .await + .unwrap_err() + .to_string() + .contains("cyclic")); + assert!(!String::from_utf8(bytes) + .unwrap() + .contains("preview_complete")); + sqlx::query("UPDATE events SET tags='[]' WHERE community_id=$1 AND id=decode($2,'hex')") + .bind(options.community) + .bind(id(1)) + .execute(&pool) + .await + .unwrap(); + options.max_items = 1; + options.batch_size = 1; + let mut bytes = Vec::new(); + assert!(store + .preview_message_batches(&options, &mut bytes) + .await + .unwrap_err() + .to_string() + .contains("item limit")); + assert!(!String::from_utf8(bytes) + .unwrap() + .contains("preview_complete")); + struct Broken; + impl Write for Broken { + fn write(&mut self, _: &[u8]) -> std::io::Result { + Err(std::io::ErrorKind::BrokenPipe.into()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + assert!(store + .preview_message_batches(&options, &mut Broken) + .await + .is_err()); +} diff --git a/crates/buzz-db/src/store/user_deletion/preview/rows.rs b/crates/buzz-db/src/store/user_deletion/preview/rows.rs new file mode 100644 index 00000000000..bdb48775fda --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/preview/rows.rs @@ -0,0 +1,168 @@ +//! Exact side-table keys, paged without mutating the snapshot. + +use super::*; + +const REACTION_KEY: &str = "jsonb_build_object('event_created_at',event_created_at, + 'event_id',encode(event_id,'hex'),'pubkey',encode(pubkey,'hex'),'emoji',emoji)"; + +// All SQL fragments are static program text. Subject/anchor/cursor are bound. +async fn rows( + conn: &mut PgConnection, + output: &mut Output<'_, impl Write>, + table: &str, + select: &str, + anchor: &str, + reason: &Value, + action: &str, +) -> Result<()> { + let query = format!( + "SELECT row_key,warning FROM ({select}) selected + WHERE ($3::jsonb IS NULL OR row_key>$3) ORDER BY row_key LIMIT {PAGE_SIZE}" + ); + let mut cursor: Option = None; + loop { + let page: Vec<(Value, Option)> = sqlx::query_as(sqlx::AssertSqlSafe(query.clone())) + .bind(output.community) + .bind(anchor) + .bind(&cursor) + .fetch_all(&mut *conn) + .await?; + if page.is_empty() { + break; + } + for (key, warning) in page { + cursor = Some(key.clone()); + output.add(table, key, action, reason, warning)?; + } + } + Ok(()) +} + +pub(super) async fn event_rows( + conn: &mut PgConnection, + event: &Event, + output: &mut Output<'_, impl Write>, +) -> Result<()> { + let id = hex::encode(&event.id); + runs(conn, &id, false, output).await?; + let reason = json!({"type": "event_record", "event_id": id}); + rows(conn, output, "reactions", &format!("SELECT {REACTION_KEY} AS row_key,NULL::text AS warning + FROM reactions WHERE community_id=$1 AND (event_id=decode($2,'hex') OR reaction_event_id=decode($2,'hex'))"), + &id, &reason, "would_delete").await?; + for (table, key, warning) in [ + ( + "event_mentions", + "jsonb_build_object('pubkey_hex',pubkey_hex,'event_id',encode(event_id,'hex'))", + "NULL::text", + ), + ( + "push_match_queue", + "jsonb_build_object('event_id',encode(event_id,'hex'))", + "CASE WHEN lease_until>transaction_timestamp() THEN 'delivery worker is active' END", + ), + ( + "push_wake_outbox", + "jsonb_build_object('id',id)", + "CASE WHEN lease_until>transaction_timestamp() THEN 'delivery worker is active' END", + ), + ( + "operator_listener_outbox", + "jsonb_build_object('id',id)", + "CASE WHEN lease_until>transaction_timestamp() THEN 'delivery worker is active' END", + ), + ( + "delivery_log", + "jsonb_build_object('delivered_at',delivered_at,'id',id)", + "NULL::text", + ), + ] { + rows( + conn, + output, + table, + &format!( + "SELECT {key} AS row_key,{warning} AS warning + FROM {table} WHERE community_id=$1 AND event_id=decode($2,'hex')" + ), + &id, + &reason, + "would_delete", + ) + .await?; + } + Ok(()) +} + +async fn runs( + conn: &mut PgConnection, + anchor: &str, + by_author: bool, + output: &mut Output<'_, impl Write>, +) -> Result<()> { + let predicate = if by_author { + "lower(trigger_context->>'author')=$2" + } else { + "(trigger_event_id=decode($2,'hex') OR lower(trigger_context->>'message_id')=$2)" + }; + let query = format!( + "SELECT id,status::text FROM workflow_runs WHERE community_id=$1 AND {predicate} + AND ($3::uuid IS NULL OR id>$3) ORDER BY id LIMIT {PAGE_SIZE}" + ); + let mut cursor: Option = None; + loop { + let page: Vec<(Uuid, String)> = sqlx::query_as(sqlx::AssertSqlSafe(query.clone())) + .bind(output.community) + .bind(anchor) + .bind(cursor) + .fetch_all(&mut *conn) + .await?; + if page.is_empty() { + break; + } + for (run, status) in page { + cursor = Some(run); + if output.runs.contains(&run) { + continue; + } + let reason = json!({"type": if by_author {"saved_trigger_author"} else {"trigger_event"}, "source": anchor, "run_id": run}); + rows(conn, output, "workflow_approvals", "SELECT jsonb_build_object('token_hash',encode(token,'hex')) AS row_key, + NULL::text AS warning FROM workflow_approvals WHERE community_id=$1 AND run_id=$2::uuid", + &run.to_string(), &reason, "would_delete").await?; + rows(conn, output, "scheduled_workflow_fires", "SELECT jsonb_build_object('workflow_id',workflow_id,'scheduled_for',scheduled_for) AS row_key, + NULL::text AS warning FROM scheduled_workflow_fires WHERE community_id=$1 AND workflow_run_id=$2::uuid", + &run.to_string(), &reason, "would_clear_workflow_run_id").await?; + output.add( + "workflow_runs", + json!({"id": run}), + "would_delete", + &reason, + (!matches!(status.as_str(), "completed" | "failed" | "cancelled")).then(|| { + format!("workflow run is {status}; must stop before any of its cleanup") + }), + )?; + output.runs.insert(run); + } + } + Ok(()) +} + +pub(super) async fn orphan_sweeps( + conn: &mut PgConnection, + subject: &str, + output: &mut Output<'_, impl Write>, +) -> Result<()> { + rows( + conn, + output, + "reactions", + &format!( + "SELECT {REACTION_KEY} AS row_key,NULL::text AS warning + FROM reactions WHERE community_id=$1 AND pubkey=decode($2,'hex')" + ), + subject, + &json!({"type": "reaction_author", "pubkey": subject}), + "would_delete", + ) + .await?; + runs(conn, subject, true, output).await +} diff --git a/crates/buzz-deletion/src/user.rs b/crates/buzz-deletion/src/user.rs index 93d1766fdb3..50ff63fbd8e 100644 --- a/crates/buzz-deletion/src/user.rs +++ b/crates/buzz-deletion/src/user.rs @@ -7,6 +7,27 @@ use uuid::Uuid; /// Preview real memberships or prepare and execute synthetic deletion requests. #[derive(Debug, Subcommand)] pub enum Command { + /// Print all proposed human-content cleanup batches; never changes data. + PreviewBatches { + /// Human public key, as npub or 64-character hex. + #[arg(long, value_parser = preview::parse_pubkey)] + pubkey: String, + /// Explicit community UUID; includes historical content after leaving. + #[arg(long)] + community: Uuid, + /// Required acknowledgement of the only supported mode. + #[arg(long, required = true)] + read_only: bool, + /// Trusted relay public key for historical actor/p attribution, if needed. + #[arg(long, value_parser = preview::parse_pubkey)] + relay_pubkey: Option, + /// Maximum row operations per printed batch. + #[arg(long, default_value_t = 100, value_parser = clap::value_parser!(u16).range(1..=100))] + batch_size: u16, + /// Fail visibly if the preview would exceed this many unique row operations. + #[arg(long, default_value_t = 1_000_000, value_parser = clap::value_parser!(u32).range(1..=1_000_000))] + max_items: u32, + }, /// Print real current memberships as JSON; does not save or execute a request. Inventory { /// Demo fixture (default), hex public key, or npub. Current memberships only. @@ -74,6 +95,25 @@ pub enum Command { /// Run inventory discovery or synthetic workflow commands against PostgreSQL. pub async fn run(command: Command) -> Result { + if let Command::PreviewBatches { + pubkey, + community, + read_only, + relay_pubkey, + batch_size, + max_items, + } = command + { + anyhow::ensure!(read_only, "only --read-only is supported"); + return preview::run(buzz_db::user_deletion::preview::PreviewOptions { + community, + pubkey, + relay_pubkey, + batch_size, + max_items, + }) + .await; + } let db = if matches!(command, Command::Inventory { .. }) { buzz_db::Db::connect_cold_start( buzz_db::DbConfig { @@ -91,6 +131,9 @@ pub async fn run(command: Command) -> Result { }; let store = db.entity_deletion_store(); match command { + Command::PreviewBatches { .. } => { + anyhow::bail!("preview must use the read-only connection") + } Command::Inventory { pubkey } => { crate::print_json(&inventory::generate(&db, &pubkey).await?)?; } @@ -154,6 +197,7 @@ mod handlers; mod inventory; /// Standalone delegation of an owned community to community deletion. pub mod owned_community; +mod preview; /// Read-only summaries of locally saved request progress and external waits. pub mod summary; mod transition; diff --git a/crates/buzz-deletion/src/user/preview.rs b/crates/buzz-deletion/src/user/preview.rs new file mode 100644 index 00000000000..b772cdb2433 --- /dev/null +++ b/crates/buzz-deletion/src/user/preview.rs @@ -0,0 +1,27 @@ +//! Standalone observational handler, deliberately outside the durable worker. + +use anyhow::Result; +use buzz_db::user_deletion::preview::PreviewOptions; + +pub(super) fn parse_pubkey(input: &str) -> std::result::Result { + buzz_db::user_deletion::normalize_user_deletion_pubkey(input).map_err(|error| error.to_string()) +} + +pub(super) async fn run(options: PreviewOptions) -> Result { + let db = buzz_db::Db::connect_cold_start( + buzz_db::DbConfig { + database_url: crate::required_env("DATABASE_URL")?, + max_connections: 1, + min_connections: 0, + default_transaction_read_only: true, + ..buzz_db::DbConfig::default() + }, + "user_batch_preview", + ) + .await?; + let stdout = std::io::stdout(); + db.user_deletion_store() + .preview_message_batches(&options, &mut stdout.lock()) + .await?; + Ok(0) +} diff --git a/docs/user-content-batch-preview.md b/docs/user-content-batch-preview.md new file mode 100644 index 00000000000..ade7ee3f605 --- /dev/null +++ b/docs/user-content-batch-preview.md @@ -0,0 +1,95 @@ +# Preview human-content cleanup batches + +This operator command prints a proposed cleanup sequence for one public key in +one community. It has no deletion executor. It neither creates a deletion request +nor changes workflow progress, counters, rows, media or delivery state. + +```sh +# DATABASE_URL must already identify the database to inspect. +cargo run -p buzz-admin -- user-deletions preview-batches \ + --pubkey '' \ + --community '' \ + --read-only > batches.jsonl +``` + +Only public keys are accepted. The community is required even when a key currently +belongs to just one community; historical content after leaving is included. +Use a database account with SELECT privileges for an additional boundary. +The command uses a read-only connection and a repeatable-read, read-only +transaction. It does not run migrations or the durable deletion worker. + +For historical events signed by a relay on behalf of a person, additionally pass +`--relay-pubkey ''`. Only this signer may use the existing +actor/p attribution rules. Without it, the header explicitly says that relay +attribution is not included. A private key is never needed. + +## Read the output + +Stdout contains JSON lines; startup diagnostics and errors use stderr: + +1. `preview_start`: subject, community, covered kinds, attribution and limits. +2. `batch`: sequential number, community and up to 100 ordered `items`. +3. `preview_complete`: total batches, unique items, warnings and excluded kinds. + +Each item names a table, exact key within the batch's community, proposed action, +selection reason and any warning. An approval's `token_hash` identifies its stored +`token` column; no plaintext approval tokens or event contents are printed. +Actions are `would_delete` and `would_clear_workflow_run_id`. A batch can mix tables +and event kinds. `thread_repairs` describes affected events and stored roots whose +metadata would need repair after that batch; it does not expand or simulate those +metadata mutations, and they are not counted in the 100 row operations. + +```sh +jq 'select(.record == "batch") | {number, items}' batches.jsonl +jq 'select(.record == "preview_complete")' batches.jsonl +``` + +A zero exit status **and** `preview_complete` mean the declared preview scope was +exhausted. They do not mean all account data is covered or that deletion is safe +to execute. On a query/output error, cycle, excessive depth or item limit, the +command fails without that marker. Discard partial output; rerun from the start. +No position is saved. + +## Selection and batching + +Sources are authored events in timestamp/ID order across threads: chat and +historical chat kinds, forum posts/comments, reactions, edits, votes, individual +pins/bookmarks, schedules, reminders, diffs, short notes and article versions. +The exact numeric kinds are printed in the header. Old and soft-deleted stored +versions are included. Notes/articles require direct authorship. + +For each source, the iterator follows incoming reactions, edits and votes within +the same community and channel, including a NULL channel. Edits/votes use the +first valid `e` tag and reactions the last. It never follows a source's outgoing +target to delete somebody else's message. Other people's ordinary replies stay. +Dependent cleanup comes before the selected event; workflow approvals and run +pointers come before their run. Direct author sweeps also find remaining reaction +rows and workflow trigger copies with missing source events. + +The iterator remembers already-emitted exact keys in memory. It fills each batch +before printing it, including across event boundaries, without changing the +database. For example, 50 reactions with exactly two stored rows each fill one +100-item batch. Additional mention or delivery records consume additional slots. +A message with many dependents spans batches and appears after the last dependent. + +`--batch-size` accepts 1–100 (default 100). `--max-items` bounds the in-memory +simulation to at most 1,000,000 unique row operations (also the default). +Exceeding the limit is an error, never a silently truncated success. Dependency +depth is limited to 64. Queries have a 30-second statement and 3-second lock +timeout. This is a diagnostic tool: snapshot duration, dependency lookups and +side-table key sorting still need measurement on large real datasets. + +## Scope and prerequisites + +This is a content-handler preview, not a whole-account erasure inventory. Shared +canvases/artifacts, command history, profiles, memberships, agents and other +non-message categories are not selected here. The footer counts authored event +kinds outside the handler. Media references, active workflow runs and claimed +delivery work are flagged while the preview continues to print the full proposed +sequence. Those warnings are not permission to execute any of it. Write fences, +media evidence, worker drain, authorization and surviving-thread repair remain +requirements for a future executor. + +The snapshot is observational: live writes after it starts are not included. +Compare previews for different users or batch sizes by rerunning the command; +never treat a saved preview as an executable deletion manifest.