From 9c6b00bb21be1cb0ae060535d809090adce0bf60 Mon Sep 17 00:00:00 2001 From: Ravneet Arora Date: Tue, 10 Nov 2026 13:27:28 -0500 Subject: [PATCH 1/2] feat(deletion): add resumable media inventory Save atomic per-file inventory and expose print-only inspection with a protected exact-version execution loop. Co-authored-by: dessert <7d616e7e143e5d408159614dccff1371350a1c1bd36c8de838a45ece5aca0aca@buzz.block.builderlab.xyz> Signed-off-by: Ravneet Arora --- .../tests/postgres_media_inventory.rs | 180 +++++++++++++ crates/buzz-core/src/lib.rs | 3 + crates/buzz-core/src/media_inventory.rs | 120 +++++++++ crates/buzz-db/src/lib.rs | 3 + crates/buzz-db/src/runtime/migration.rs | 5 +- crates/buzz-db/src/store/deletion.rs | 11 + .../src/store/media_deletion/execution.rs | 92 +++++++ .../buzz-db/src/store/media_deletion/mod.rs | 241 +++++++++++++++++ .../store/media_deletion/postgres_tests.rs | 252 ++++++++++++++++++ .../buzz-db/src/store/media_deletion/scan.rs | 230 ++++++++++++++++ .../src/store/media_deletion/sources.sql | 36 +++ crates/buzz-db/src/store/mod.rs | 3 + crates/buzz-deletion/src/user.rs | 10 + .../buzz-deletion/src/user/media/command.rs | 116 ++++++++ crates/buzz-deletion/src/user/media/mod.rs | 237 ++++++++++++++++ .../src/user/media/postgres_tests.rs | 145 ++++++++++ .../buzz-deletion/src/user/media/storage.rs | 144 ++++++++++ crates/buzz-media/src/storage.rs | 35 +++ docs/user-media-deletion.md | 160 +++++++++++ migrations/0087_entity_deletion_media.sql | 24 ++ schema/schema.sql | 2 + schema/seeds/entity_deletion_media_files.sql | 3 + .../public/entity_deletion_media_files.sql | 17 ++ 23 files changed, 2068 insertions(+), 1 deletion(-) create mode 100644 crates/buzz-admin/tests/postgres_media_inventory.rs create mode 100644 crates/buzz-core/src/media_inventory.rs create mode 100644 crates/buzz-db/src/store/media_deletion/execution.rs create mode 100644 crates/buzz-db/src/store/media_deletion/mod.rs create mode 100644 crates/buzz-db/src/store/media_deletion/postgres_tests.rs create mode 100644 crates/buzz-db/src/store/media_deletion/scan.rs create mode 100644 crates/buzz-db/src/store/media_deletion/sources.sql create mode 100644 crates/buzz-deletion/src/user/media/command.rs create mode 100644 crates/buzz-deletion/src/user/media/mod.rs create mode 100644 crates/buzz-deletion/src/user/media/postgres_tests.rs create mode 100644 crates/buzz-deletion/src/user/media/storage.rs create mode 100644 docs/user-media-deletion.md create mode 100644 migrations/0087_entity_deletion_media.sql create mode 100644 schema/seeds/entity_deletion_media_files.sql create mode 100644 schema/tables/public/entity_deletion_media_files.sql diff --git a/crates/buzz-admin/tests/postgres_media_inventory.rs b/crates/buzz-admin/tests/postgres_media_inventory.rs new file mode 100644 index 00000000000..057a7a7f2f6 --- /dev/null +++ b/crates/buzz-admin/tests/postgres_media_inventory.rs @@ -0,0 +1,180 @@ +//! Real command against PostgreSQL and a read-only S3 protocol fixture. +use axum::{ + extract::State, + http::{Method, StatusCode, Uri}, + response::IntoResponse, + routing::any, + Router, +}; +use serde_json::Value; +use sqlx::PgPool; +use std::sync::{Arc, Mutex}; +use tokio::process::Command; +use uuid::Uuid; + +#[derive(Clone)] +struct S3Fixture { + hash: String, + upload_key: String, + upload_record: String, + methods: Arc>>, +} +async fn storage(State(state): State, method: Method, uri: Uri) -> impl IntoResponse { + state.methods.lock().unwrap().push(method.clone()); + if method != Method::GET { + return (StatusCode::FORBIDDEN, "unexpected storage operation".into()); + } + if uri.query().unwrap_or_default().contains("versions") { + return (StatusCode::OK,format!("false{}.jpgfixture-v14true{}upload-v1{}false",state.hash,state.upload_key,state.upload_record.len())); + } + if uri.path().ends_with(&state.upload_key) + && uri + .query() + .unwrap_or_default() + .contains("versionId=upload-v1") + { + return (StatusCode::OK, state.upload_record); + } + ( + StatusCode::FORBIDDEN, + "exact upload version required".into(), + ) +} +fn binary() -> std::ffi::OsString { + std::env::var_os("NEXTEST_BIN_EXE_buzz_admin") + .unwrap_or_else(|| env!("CARGO_BIN_EXE_buzz-admin").into()) +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn preview_cli_saves_inventory_resumes_and_never_calls_delete() { + let url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap(); + let pool = PgPool::connect(&url).await.unwrap(); + let human = nostr::Keys::generate().public_key().to_hex(); + let community = Uuid::new_v4(); + let run = Uuid::new_v4(); + let hash = "a".repeat(64); + sqlx::query("INSERT INTO communities (id,host) VALUES ($1,'preview-fixture.example.org')") + .bind(community) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO media_uploaders (community_id,sha256,pubkey) VALUES ($1,$2,decode($3,'hex'))", + ) + .bind(community) + .bind(&hash) + .bind(&human) + .execute(&pool) + .await + .unwrap(); + let state = S3Fixture { + hash: hash.clone(), + upload_key: format!( + "_uploads/{community}/{}/01ARZ3NDEKTSV4RRFFQ69G5FAV.json", + "b".repeat(64) + ), + upload_record: serde_json::json!({ + "version":1,"community_id":community,"sha256":"b".repeat(64), + "event_id":"01ARZ3NDEKTSV4RRFFQ69G5FAV","uploader_id":human + }) + .to_string(), + methods: Default::default(), + }; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let app = Router::new() + .fallback(any(storage)) + .with_state(state.clone()); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let invoke = || { + let mut command = Command::new(binary()); + command + .args([ + "user-deletions", + "media", + "--pubkey", + &human, + "--run-id", + &run.to_string(), + "--mode", + "print-only", + ]) + .env("DATABASE_URL", &url) + .env("BUZZ_MEDIA_SOURCE_DATABASE_URL", &url) + .env("BUZZ_S3_ENDPOINT", format!("http://{addr}")) + .env("BUZZ_S3_BUCKET", "fixture") + .env("BUZZ_S3_ACCESS_KEY", "fixture") + .env("BUZZ_S3_SECRET_KEY", "fixture-secret") + .env("BUZZ_S3_ADDRESSING_STYLE", "path"); + command + }; + let first = invoke().output().await.unwrap(); + assert_eq!( + first.status.code(), + Some(2), + "{}", + String::from_utf8_lossy(&first.stderr) + ); + let inventory: Value = serde_json::from_slice(&first.stdout).unwrap(); + assert_eq!(inventory["discovery_complete"], true); + assert_eq!(inventory["candidate_count"], 2); + assert_eq!(inventory["execution_authorized"], false); + assert_eq!(inventory["files"][0]["file_hash"], hash); + assert_eq!(inventory["files"][0]["status"], "unknown"); + assert_eq!( + inventory["files"][0]["evidence_json"]["observed_objects"][0]["version_id"], + "fixture-v1" + ); + // This file has no database uploader or reference. Its historical upload + // record alone must add it, and must be fetched at the listed version. + assert_eq!(inventory["files"][1]["file_hash"], "b".repeat(64)); + assert_eq!( + inventory["files"][1]["evidence_json"]["sources"][0]["table"], + "storage_uploads" + ); + let methods = state.methods.lock().unwrap().clone(); + assert!(!methods.is_empty()); + assert!(methods.iter().all(|m| *m == Method::GET)); + server.abort(); // Recovery must use the saved inventory even with S3 offline. + let second = invoke().output().await.unwrap(); + assert_eq!( + second.status.code(), + Some(2), + "{}", + String::from_utf8_lossy(&second.stderr) + ); + let repeated: Value = serde_json::from_slice(&second.stdout).unwrap(); + assert_eq!(inventory, repeated); + let remaining: i64 = + sqlx::query_scalar("SELECT count(*) FROM media_uploaders WHERE community_id=$1") + .bind(community) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(remaining, 1); + let count: i64 = + sqlx::query_scalar("SELECT count(*) FROM entity_deletion_media_files WHERE stage_id=$1") + .bind(run) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(count, 2); + let rejected = Command::new(binary()) + .args([ + "user-deletions", + "media", + "--pubkey", + &human, + "--mode", + "print", + ]) + .env_remove("DATABASE_URL") + .output() + .await + .unwrap(); + assert!(!rejected.status.success()); + assert!(String::from_utf8_lossy(&rejected.stderr).contains("ProtectedMedia")); +} diff --git a/crates/buzz-core/src/lib.rs b/crates/buzz-core/src/lib.rs index f1330a65655..88d45167f39 100644 --- a/crates/buzz-core/src/lib.rs +++ b/crates/buzz-core/src/lib.rs @@ -89,3 +89,6 @@ pub mod artifact_sync; /// Local media reference extraction. pub mod media; + +/// Media inventory extraction (does not grant read access). +pub mod media_inventory; diff --git a/crates/buzz-core/src/media_inventory.rs b/crates/buzz-core/src/media_inventory.rs new file mode 100644 index 00000000000..a2d90049a52 --- /dev/null +++ b/crates/buzz-core/src/media_inventory.rs @@ -0,0 +1,120 @@ +//! Conservative inventory extraction, separate from read authorization. +//! Free-text sightings do not grant media access or prove upload ownership. +use serde_json::Value; +use std::collections::BTreeSet; + +/// Extract local media hashes with JSON-pointer-like field paths. Nested JSON +/// strings and animated-avatar descriptors are decoded. Resource/format errors +/// must not be interpreted as an empty reference set by the caller. +pub fn references(value: &Value, bases: &[String]) -> Result, String> { + let mut out = BTreeSet::new(); + visit(value, bases, "", 0, &mut out)?; + Ok(out) +} + +fn visit( + value: &Value, + bases: &[String], + path: &str, + depth: usize, + out: &mut BTreeSet<(String, String)>, +) -> Result<(), String> { + if depth > 32 || out.len() > 10_000 { + return Err("media reference parsing limit exceeded".into()); + } + match value { + Value::Array(items) => { + for (i, v) in items.iter().enumerate() { + visit(v, bases, &format!("{path}/{i}"), depth + 1, out)?; + } + } + Value::Object(items) => { + for (k, v) in items { + visit( + v, + bases, + &format!("{path}/{}", k.replace('~', "~0").replace('/', "~1")), + depth + 1, + out, + )?; + } + } + Value::String(s) => { + // JSON embedded in event content is commonplace. Decode before URL tokenization. + if (s.trim_start().starts_with('{') || s.trim_start().starts_with('[')) + && s.len() <= 1_048_576 + { + if let Ok(decoded) = serde_json::from_str::(s) { + visit(&decoded, bases, &format!("{path}/json"), depth + 1, out)?; + } + } + for token in s.split(|c: char| { + c.is_whitespace() + || matches!( + c, + '"' | '\'' | '<' | '>' | '(' | ')' | '[' | ']' | '{' | '}' | '`' | '\\' + ) + }) { + let token = token.trim_end_matches([',', ';']); + if token.contains("#buzz-anim=") { + let (poster, animation) = crate::media::animated_avatar_urls(token) + .ok_or_else(|| "invalid animated-avatar descriptor".to_string())?; + for (suffix, url) in [("poster", poster), ("animation", animation)] { + add_url(&url, bases, &format!("{path}/{suffix}"), out); + } + } else { + add_url(token, bases, path, out); + } + } + if let Some(hash) = s.strip_prefix("x ").filter(|h| valid_hash(h)) { + out.insert((hash.into(), path.into())); + } + } + _ => {} + } + Ok(()) +} + +fn add_url(url: &str, bases: &[String], path: &str, out: &mut BTreeSet<(String, String)>) { + let url = url + .split(['?', '#']) + .next() + .unwrap_or(url) + .trim_end_matches('.'); + // Always validate absolute URLs against configured/local hosts. A foreign + // /media path is not an object in this namespace. + for base in bases { + if let Some(hash) = crate::media::local_hash(url, base) { + out.insert((hash, path.into())); + break; + } + } +} + +/// Whether a value is a canonical file hash. +pub fn valid_hash(value: &str) -> bool { + value.len() == 64 + && value + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn attachments_nested_json_animation_and_foreign_hosts() { + let a = "a".repeat(64); + let b = "b".repeat(64); + let c = "c".repeat(64); + let value = serde_json::json!({"tags":[["imeta",format!("url /media/{a}.jpg")]],"content":serde_json::json!({"picture":format!("https://example.org/media/{b}.jpg#buzz-anim=https%3A%2F%2Fexample.org%2Fmedia%2F{c}.gif")}).to_string(),"external":format!("https://elsewhere.org/media/{}.png","d".repeat(64))}); + let refs = references(&value, &["https://example.org/media".into()]).unwrap(); + let hashes: BTreeSet<_> = refs.into_iter().map(|(h, _)| h).collect(); + assert_eq!(hashes, BTreeSet::from([a, b, c])); + assert!(references( + &serde_json::json!("https://example.org/media/a#buzz-anim=%QQ"), + &[] + ) + .is_err()); + } +} diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 92eeb42f882..9d8cf5ecd20 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -134,3 +134,6 @@ use buzz_core::CommunityId; /// Durable NIP-FI verified-profile projection state. pub use store::verified_profiles; + +/// Operator media inventory and deletion journal. +pub use store::media_deletion; diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index f7f0888e96f..34fbcbe224a 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -674,6 +674,7 @@ mod postgres_tests { "entity_deletion_approvals", "entity_deletion_requests", "entity_deletion_stage_history", + "entity_deletion_media_files", "user_deletion_fences", "community_deletion_approvals", "community_deletion_checkpoints", @@ -902,7 +903,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 86); + assert_eq!(migrations.len(), 87); assert!(migrations[83] .sql .as_str() @@ -3056,6 +3057,7 @@ mod postgres_tests { "channel_members", "communities", "entity_deletion_stage_history", + "entity_deletion_media_files", "user_deletion_fences", ] { assert_eq!( @@ -3118,6 +3120,7 @@ mod postgres_tests { "community_auth_upgrades", "entity_deletion_requests", "entity_deletion_stage_history", + "entity_deletion_media_files", "user_deletion_fences", "entity_deletion_approvals", ] { diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 3a3d4e6e4fd..269b8c07b11 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -8703,6 +8703,8 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); .fetch_one(&db.pool) .await .unwrap(); + sqlx::query("INSERT INTO entity_deletion_media_files (stage_id,storage_namespace,file_hash,evidence_json) VALUES ($1,'fixture',repeat('a',64),$2)") + .bind(history_id).bind(serde_json::json!({"source_community":community})).execute(&db.pool).await.unwrap(); let live = store .inventory_schema(request.community_id) .await @@ -8751,6 +8753,15 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); .await .unwrap(); assert_eq!(retained, community); + let media_evidence: serde_json::Value = sqlx::query_scalar( + "SELECT evidence_json FROM entity_deletion_media_files WHERE stage_id=$1", + ) + .bind(history_id) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(media_evidence["source_community"], community.to_string()); + let approvals_after = db .entity_deletion_store() .inspect(entity.id) diff --git a/crates/buzz-db/src/store/media_deletion/execution.rs b/crates/buzz-db/src/store/media_deletion/execution.rs new file mode 100644 index 00000000000..c2fa4471d03 --- /dev/null +++ b/crates/buzz-db/src/store/media_deletion/execution.rs @@ -0,0 +1,92 @@ +//! Owned workflow writes. Preview rows cannot be promoted into an execution run. +use super::*; +use crate::entity_deletion::EntityLease; + +impl MediaDeletionStore { + /// Bind a caller-started, approved purge_media stage to a protected inventory. + /// The workflow supplies and maintains serving/media protections separately + /// from its request lease; a request lease alone never fences uploads. + pub async fn execution_run( + &self, + lease: &EntityLease, + stage: Uuid, + binding: Value, + ) -> Result { + let mut tx = self.pool.begin().await?; + let request = super::super::entity_deletion::lock_owned(&mut tx, lease).await?; + super::super::entity_deletion::validate_approved_inventory(&mut tx, &request).await?; + let metadata:Value=sqlx::query_scalar("SELECT metadata_json FROM entity_deletion_stage_history WHERE id=$1 AND request_id=$2 AND stage='purge_media' AND is_current AND status='running' FOR UPDATE") + .bind(stage).bind(lease.request_id).fetch_one(&mut *tx).await?; + if metadata.get("binding").is_some_and(|b| b != &binding) { + return Err(invalid("execution media binding changed")); + } + sqlx::query("UPDATE entity_deletion_stage_history SET metadata_json=metadata_json || $2,updated_at=now() WHERE id=$1") + .bind(stage).bind(json!({"binding":binding,"generation":lease.generation})).execute(&mut *tx).await?; + tx.commit().await?; + Ok(MediaRun { + id: stage, + generation: lease.generation, + binding, + lease: Some(lease.clone()), + }) + } + + /// Atomically save a complete decision with its immutable target list. + /// Only unknown rows may acquire a plan; retries never replace saved targets. + pub async fn save_execution_decision(&self, run: &MediaRun, file: &MediaFile) -> Result<()> { + if run.lease.is_none() { + return Err(invalid("preview is not executable")); + } + if !matches!( + file.status.as_str(), + "unknown" | "dont_delete" | "to_be_deleted" + ) || serde_json::to_vec(file)?.len() > MAX_FILE_BYTES + { + return Err(invalid("invalid media decision")); + } + let targets: Vec = serde_json::from_value(file.delete_targets_json.clone())?; + if (file.status == "to_be_deleted") != !targets.is_empty() { + return Err(invalid("decision and targets disagree")); + } + for target in &targets { + if target.version_id.is_empty() || target.version_id == "null" || target.key.is_empty() + { + return Err(invalid("deletion requires immutable version IDs")); + } + } + let mut tx = self.pool.begin().await?; + let metadata = self.lock(&mut tx, run).await?; + if metadata["media_discovery"]["complete"] != true { + return Err(invalid("discovery unfinished")); + } + let changed=sqlx::query("UPDATE entity_deletion_media_files SET status=$4,evidence_json=$5,delete_targets_json=$6,reason=$7,last_error=$8,updated_at=now() WHERE stage_id=$1 AND storage_namespace=$2 AND file_hash=$3 AND status='unknown'") + .bind(run.id).bind(&file.storage_namespace).bind(&file.file_hash).bind(&file.status).bind(&file.evidence_json).bind(&file.delete_targets_json).bind(&file.reason).bind(&file.last_error).execute(&mut *tx).await?; + if changed.rows_affected() != 1 { + return Err(invalid("decision row is no longer unknown")); + } + tx.commit().await?; + Ok(()) + } + + /// Record confirmed absence or a retry error without changing saved targets. + pub async fn save_execution_outcome( + &self, + run: &MediaRun, + file: &MediaFile, + error: Option<&str>, + ) -> Result<()> { + if run.lease.is_none() { + return Err(invalid("preview is not executable")); + } + let mut tx = self.pool.begin().await?; + self.lock(&mut tx, run).await?; + let error = error.map(|s| s.chars().take(2048).collect::()); + let changed=sqlx::query("UPDATE entity_deletion_media_files SET status=CASE WHEN $5::text IS NULL THEN 'deleted' ELSE 'to_be_deleted' END,last_error=$5,updated_at=now() WHERE stage_id=$1 AND storage_namespace=$2 AND file_hash=$3 AND delete_targets_json=$4 AND status='to_be_deleted'") + .bind(run.id).bind(&file.storage_namespace).bind(&file.file_hash).bind(&file.delete_targets_json).bind(error).execute(&mut *tx).await?; + if changed.rows_affected() != 1 { + return Err(invalid("execution row or targets changed")); + } + tx.commit().await?; + Ok(()) + } +} diff --git a/crates/buzz-db/src/store/media_deletion/mod.rs b/crates/buzz-db/src/store/media_deletion/mod.rs new file mode 100644 index 00000000000..de31d059019 --- /dev/null +++ b/crates/buzz-db/src/store/media_deletion/mod.rs @@ -0,0 +1,241 @@ +//! One-table media journal. Discovery publication is all-or-none, including zero files. +use crate::{Db, DbError, Result}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use sqlx::PgPool; +use uuid::Uuid; + +mod execution; +mod scan; +pub use scan::{MediaScan, MediaSource, SubjectScope}; + +/// Maximum journal payload per file; exceeding it fails rather than truncates. +pub const MAX_FILE_BYTES: usize = 1_048_576; +/// Maximum number of distinct candidates in an atomic publication. +pub const MAX_FILES: usize = 100_000; + +/// Exact, immutable storage target. A null/unversioned identifier is not executable. +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct MediaTarget { + /// Storage object path. + pub key: String, + /// Immutable storage version. + pub version_id: String, +} + +/// One logical file and its evidence, decision and selected objects. +#[derive(Clone, Debug, Serialize, Deserialize, sqlx::FromRow)] +pub struct MediaFile { + /// Shared bucket/namespace identity. + pub storage_namespace: String, + /// SHA-256 of file content. + pub file_hash: String, + /// unknown, dont_delete, to_be_deleted or deleted. + pub status: String, + /// Source keys, fingerprints and checks; never source bodies. + pub evidence_json: Value, + /// Selected storage keys and versions. + pub delete_targets_json: Value, + /// Explanation of the decision, including any retained components. + pub reason: Option, + /// Unresolved check or execution error. + pub last_error: Option, +} + +/// Fenced ownership of a standalone observation stage. +#[derive(Clone, Debug)] +pub struct MediaRun { + /// Stable stage identifier supplied by the operator for recovery. + pub id: Uuid, + /// Increments on every invocation; stale workers cannot publish. + pub generation: i64, + pub(super) lease: Option, + /// Bound subject, namespace, source identity, parser version and mode. + pub binding: Value, +} + +/// Writer connection to the operator control database (may differ from source DB). +#[derive(Clone)] +pub struct MediaDeletionStore { + pool: PgPool, +} +impl Db { + /// Access the operator-global media journal. + pub fn media_deletion_store(&self) -> MediaDeletionStore { + MediaDeletionStore { + pool: self.pool().clone(), + } + } +} +fn invalid(message: &str) -> DbError { + DbError::DeletionSafety(message.into()) +} + +impl MediaDeletionStore { + /// Start/resume an isolated preview. It cannot be approved or executed by the + /// user worker: its parent is terminal and no stage is current. + pub async fn open_preview(&self, id: Uuid, binding: Value) -> Result { + let mut tx = self.pool.begin().await?; + sqlx::query("SET LOCAL lock_timeout = '5s'") + .execute(&mut *tx) + .await?; + sqlx::query( + "INSERT INTO entity_deletion_requests + (id,target,status,requested_by,reason,completed_at) + VALUES ($1,$2,'completed','media-preview','Non-executable media observation',now()) + ON CONFLICT (id) DO NOTHING", + ) + .bind(id) + .bind(format!("media-preview:{id}")) + .execute(&mut *tx) + .await?; + let parent: (String, String) = sqlx::query_as( + "SELECT target,status FROM entity_deletion_requests WHERE id=$1 FOR UPDATE", + ) + .bind(id) + .fetch_one(&mut *tx) + .await?; + if parent != (format!("media-preview:{id}"), "completed".into()) { + return Err(invalid("run ID belongs to another workflow")); + } + sqlx::query( + "INSERT INTO entity_deletion_stage_history + (id,request_id,stage,is_current,metadata_json) + VALUES ($1,$1,'media_preview',false,$2) ON CONFLICT (id) DO NOTHING", + ) + .bind(id) + .bind(json!({"binding":binding,"generation":0})) + .execute(&mut *tx) + .await?; + let (metadata, stage, current): (Value,String,bool) = sqlx::query_as("SELECT metadata_json,stage,is_current FROM entity_deletion_stage_history WHERE id=$1 AND request_id=$1 FOR UPDATE") + .bind(id).fetch_one(&mut *tx).await?; + if metadata["binding"] != binding || stage != "media_preview" || current { + return Err(invalid( + "run binding changed; use a new run ID for a new observation", + )); + } + let generation = metadata["generation"] + .as_i64() + .and_then(|v| v.checked_add(1)) + .ok_or_else(|| invalid("invalid media generation"))?; + sqlx::query("UPDATE entity_deletion_stage_history SET metadata_json=jsonb_set(metadata_json,'{generation}',$2), updated_at=now() WHERE id=$1") + .bind(id).bind(json!(generation)).execute(&mut *tx).await?; + tx.commit().await?; + Ok(MediaRun { + id, + generation, + binding, + lease: None, + }) + } + + async fn lock( + &self, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + run: &MediaRun, + ) -> Result { + if let Some(lease) = &run.lease { + let request = super::entity_deletion::lock_owned(tx, lease).await?; + super::entity_deletion::validate_approved_inventory(tx, &request).await?; + } + let metadata: Value = sqlx::query_scalar("SELECT metadata_json FROM entity_deletion_stage_history WHERE id=$1 AND ((stage='media_preview' AND NOT is_current AND request_id=$1 AND $2 IS NULL) OR (stage='purge_media' AND is_current AND request_id=$2 AND status='running')) FOR UPDATE") + .bind(run.id).bind(run.lease.as_ref().map(|l|l.request_id)).fetch_one(&mut **tx).await?; + if metadata["generation"] != run.generation || metadata["binding"] != run.binding { + return Err(invalid("media stage ownership lost")); + } + Ok(metadata) + } + + /// Read completion plus rows from one snapshot. Counts detect missing rows; + /// an empty table without the committed marker is never successful discovery. + pub async fn load(&self, run: &MediaRun) -> Result>> { + let mut tx = self.pool.begin().await?; + let metadata = self.lock(&mut tx, run).await?; + let rows: Vec = sqlx::query_as("SELECT storage_namespace,file_hash,status,evidence_json,delete_targets_json,reason,last_error FROM entity_deletion_media_files WHERE stage_id=$1 ORDER BY storage_namespace,file_hash") + .bind(run.id).fetch_all(&mut *tx).await?; + let complete = metadata["media_discovery"]["complete"] == true; + if (!complete && !rows.is_empty()) + || (complete && metadata["media_discovery"]["candidate_count"] != json!(rows.len())) + { + return Err(invalid("media discovery marker/count disagrees with rows")); + } + tx.commit().await?; + Ok(complete.then_some(rows)) + } + + /// Return the committed scan receipt, including scope and zero-result scans. + pub async fn receipt(&self, run: &MediaRun) -> Result { + let mut tx = self.pool.begin().await?; + let metadata = self.lock(&mut tx, run).await?; + tx.commit().await?; + Ok(metadata["media_discovery"].clone()) + } + + /// Insert every candidate and the completion marker in ONE transaction. + /// Retrying after a lost commit acknowledgement preserves the first inventory. + pub async fn publish(&self, run: &MediaRun, files: &[MediaFile], scans: Value) -> Result<()> { + if files.len() > MAX_FILES { + return Err(invalid("media candidate limit exceeded")); + } + let total_bytes = files.iter().try_fold(0usize, |n, f| { + serde_json::to_vec(f).map(|bytes| n.saturating_add(bytes.len())) + })?; + if total_bytes > 128 * 1024 * 1024 { + return Err(invalid("media inventory exceeds 128 MiB")); + } + let mut tx = self.pool.begin().await?; + let metadata = self.lock(&mut tx, run).await?; + let existing: i64 = sqlx::query_scalar( + "SELECT count(*) FROM entity_deletion_media_files WHERE stage_id=$1", + ) + .bind(run.id) + .fetch_one(&mut *tx) + .await?; + if metadata["media_discovery"]["complete"] == true { + if metadata["media_discovery"]["candidate_count"] != json!(existing) { + return Err(invalid("published media row count changed")); + } + return Ok(()); + } + if existing != 0 { + return Err(invalid("unpublished media rows exist")); + } + for file in files { + if file.status != "unknown" + || file.delete_targets_json != json!([]) + || serde_json::to_vec(file)?.len() > MAX_FILE_BYTES + { + return Err(invalid("invalid or oversized discovery candidate")); + } + sqlx::query("INSERT INTO entity_deletion_media_files (stage_id,storage_namespace,file_hash,evidence_json) VALUES ($1,$2,$3,$4)") + .bind(run.id).bind(&file.storage_namespace).bind(&file.file_hash).bind(&file.evidence_json).execute(&mut *tx).await?; + } + sqlx::query("UPDATE entity_deletion_stage_history SET status='running', started_at=COALESCE(started_at,now()), metadata_json=jsonb_set(metadata_json,'{media_discovery}',$2),updated_at=now() WHERE id=$1") + .bind(run.id).bind(json!({"complete":true,"candidate_count":files.len(),"scans":scans})).execute(&mut *tx).await?; + tx.commit().await?; + Ok(()) + } + + /// Save a decision and its targets together. Preview rows may only be kept + /// or unresolved; execution cannot promote these unfenced observations. + pub async fn save_preview_decision(&self, run: &MediaRun, file: &MediaFile) -> Result<()> { + if !matches!(file.status.as_str(), "unknown" | "dont_delete") + || file.delete_targets_json != json!([]) + || serde_json::to_vec(file)?.len() > MAX_FILE_BYTES + { + return Err(invalid("a preview cannot authorize or report deletion")); + } + let mut tx = self.pool.begin().await?; + let metadata = self.lock(&mut tx, run).await?; + if metadata["media_discovery"]["complete"] != true { + return Err(invalid("discovery unfinished")); + } + sqlx::query("UPDATE entity_deletion_media_files SET status=$4,evidence_json=$5,reason=$6,last_error=$7,updated_at=now() WHERE stage_id=$1 AND storage_namespace=$2 AND file_hash=$3 AND status='unknown'") + .bind(run.id).bind(&file.storage_namespace).bind(&file.file_hash).bind(&file.status).bind(&file.evidence_json).bind(&file.reason).bind(&file.last_error).execute(&mut *tx).await?; + tx.commit().await?; + Ok(()) + } +} + +#[cfg(test)] +mod postgres_tests; diff --git a/crates/buzz-db/src/store/media_deletion/postgres_tests.rs b/crates/buzz-db/src/store/media_deletion/postgres_tests.rs new file mode 100644 index 00000000000..2d0d96190f0 --- /dev/null +++ b/crates/buzz-db/src/store/media_deletion/postgres_tests.rs @@ -0,0 +1,252 @@ +use super::*; +use crate::entity_deletion::{NewEntityStage, NextStage, StageTransition}; +use std::{collections::BTreeMap, time::Duration}; + +fn candidate(hash: &str) -> MediaFile { + MediaFile { + storage_namespace: "fixture".into(), + file_hash: hash.repeat(64), + status: "unknown".into(), + evidence_json: json!({"sources":[]}), + delete_targets_json: json!([]), + reason: None, + last_error: None, + } +} +async fn db() -> Db { + Db::from_pool( + PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(), + ) +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn atomic_discovery_zero_inventory_and_stale_worker() { + let db = db().await; + let store = db.media_deletion_store(); + let id = Uuid::new_v4(); + let run = store + .open_preview(id, json!({"mode":"print-only","subject":"fixture"})) + .await + .unwrap(); + assert!(store.load(&run).await.unwrap().is_none()); + let file = candidate("a"); + // Second insert violates uniqueness AFTER the first insert. Neither the row + // nor the marker may survive that transaction's failure. + assert!(store + .publish(&run, &[file.clone(), file], json!({})) + .await + .is_err()); + assert!(store.load(&run).await.unwrap().is_none()); + store.publish(&run, &[], json!({"events":0})).await.unwrap(); + assert!(store.load(&run).await.unwrap().unwrap().is_empty()); + let replacement = store.open_preview(id, run.binding.clone()).await.unwrap(); + assert!(store.load(&run).await.is_err()); + store + .publish(&replacement, &[candidate("b")], json!({})) + .await + .unwrap(); + assert!(store.load(&replacement).await.unwrap().unwrap().is_empty()); + assert!(store + .open_preview(id, json!({"mode":"print"})) + .await + .is_err()); + assert!(db + .entity_deletion_store() + .claim(id, "worker", Duration::from_secs(60)) + .await + .unwrap() + .is_none()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn preview_cannot_authorize_deletion_and_preserves_control_evidence() { + let db = db().await; + let store = db.media_deletion_store(); + let run = store + .open_preview(Uuid::new_v4(), json!({"mode":"print-only"})) + .await + .unwrap(); + let mut file = candidate("a"); + store + .publish(&run, &[file.clone()], json!({})) + .await + .unwrap(); + file.status = "to_be_deleted".into(); + file.delete_targets_json = json!([{"key":"a.jpg","version_id":"v1"}]); + assert!(store.save_preview_decision(&run, &file).await.is_err()); + assert!(store.save_execution_decision(&run, &file).await.is_err()); + assert!(store + .save_execution_outcome(&run, &file, None) + .await + .is_err()); + let rows = store.load(&run).await.unwrap().unwrap(); + assert_eq!(rows[0].status, "unknown"); + file = rows[0].clone(); + file.status = "dont_delete".into(); + file.reason = Some("surviving private reference".into()); + store.save_preview_decision(&run, &file).await.unwrap(); + assert_eq!( + store.load(&run).await.unwrap().unwrap()[0].status, + "dont_delete" + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn exact_execution_plan_survives_failure_and_cannot_be_replaced() { + let db = db().await; + let requests = db.entity_deletion_store(); + let request = db + .user_deletion_store() + .submit( + &format!("demo:media-{}", Uuid::new_v4().simple()), + "operator", + None, + ) + .await + .unwrap(); + requests + .approve( + request.id, + "reviewer", + request.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + let lease = requests + .claim(request.id, "worker", Duration::from_secs(120)) + .await + .unwrap() + .unwrap(); + let NextStage::Stage(stage) = requests + .transition_stage(&lease, None, |_, _| { + Ok(StageTransition::Initialize(NewEntityStage { + stage: "purge_media".into(), + community: None, + inputs: json!({}), + })) + }) + .await + .unwrap() + else { + panic!("missing stage") + }; + requests.start_stage(&lease, stage.id).await.unwrap(); + let store = db.media_deletion_store(); + let run = store + .execution_run(&lease, stage.id, json!({"scope":"fixture"})) + .await + .unwrap(); + let mut file = candidate("a"); + store + .publish(&run, &[file.clone()], json!({})) + .await + .unwrap(); + file.status = "to_be_deleted".into(); + file.delete_targets_json = json!([{"key":format!("{}.jpg",file.file_hash),"version_id":"v1"}]); + store.save_execution_decision(&run, &file).await.unwrap(); + let mut changed = file.clone(); + changed.delete_targets_json = json!([{"key":"replacement","version_id":"v2"}]); + assert!(store.save_execution_decision(&run, &changed).await.is_err()); + assert!(store + .save_execution_outcome(&run, &changed, None) + .await + .is_err()); + store + .save_execution_outcome(&run, &file, Some("timeout after external delete")) + .await + .unwrap(); + let row = store.load(&run).await.unwrap().unwrap().remove(0); + assert_eq!(row.status, "to_be_deleted"); + assert_eq!(row.delete_targets_json, file.delete_targets_json); + store + .save_execution_outcome(&run, &file, None) + .await + .unwrap(); + assert_eq!( + store.load(&run).await.unwrap().unwrap()[0].status, + "deleted" + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn source_scan_finds_unposted_uploads_private_survivors_and_scoped_agents() { + let db = db().await; + let pool = db.pool(); + let human = nostr::Keys::generate().public_key().to_hex(); + let agent = nostr::Keys::generate().public_key().to_hex(); + let other = nostr::Keys::generate().public_key().to_hex(); + let a = Uuid::new_v4(); + let b = Uuid::new_v4(); + for c in [a, b] { + sqlx::query("INSERT INTO communities (id,host) VALUES ($1,$2)") + .bind(c) + .bind(format!("{c}.example.org")) + .execute(pool) + .await + .unwrap(); + for key in [&human, &agent, &other] { + sqlx::query("INSERT INTO users (community_id,pubkey) VALUES ($1,decode($2,'hex'))") + .bind(c) + .bind(key) + .execute(pool) + .await + .unwrap(); + } + } + sqlx::query("UPDATE users SET agent_owner_pubkey=decode($2,'hex') WHERE community_id=$1 AND pubkey=decode($3,'hex')").bind(a).bind(&human).bind(&agent).execute(pool).await.unwrap(); + // No membership rows: historical source discovery must still find these. + for (c, key, hash) in [ + (a, &human, "a"), + (a, &agent, "b"), + (b, &agent, "b"), + (a, &human, "c"), + ] { + sqlx::query("INSERT INTO media_uploaders (community_id,pubkey,sha256) VALUES ($1,decode($2,'hex'),$3)").bind(c).bind(key).bind(hash.repeat(64)).execute(pool).await.unwrap(); + } + sqlx::query("UPDATE users SET avatar_url=$3 WHERE community_id=$1 AND pubkey=decode($2,'hex')") + .bind(b) + .bind(&other) + .bind(format!("/media/{}.jpg", "a".repeat(64))) + .execute(pool) + .await + .unwrap(); + sqlx::query("UPDATE users SET about=$3 WHERE community_id=$1 AND pubkey=decode($2,'hex')") + .bind(a) + .bind(&human) + .bind(format!("only a reference /media/{}.bin", "d".repeat(64))) + .execute(pool) + .await + .unwrap(); + let scan = db + .scan_user_media(&human, true, "fixture", &[]) + .await + .unwrap(); + let files: BTreeMap<_, _> = scan + .files + .iter() + .map(|f| (f.file_hash.clone(), f)) + .collect(); + for hash in ["a", "b", "c", "d"] { + assert!(files.contains_key(&hash.repeat(64))); + } + assert_eq!(scan.scope.agents.len(), 1); + for hash in ["a", "b"] { + assert!(files[&hash.repeat(64)].evidence_json["sources"] + .as_array() + .unwrap() + .iter() + .any(|s| s["surviving"] == true)); + } + assert!(!files[&"c".repeat(64)].evidence_json["sources"] + .as_array() + .unwrap() + .iter() + .any(|s| s["surviving"] == true)); + assert!(scan.scanned.contains_key("events")); +} diff --git a/crates/buzz-db/src/store/media_deletion/scan.rs b/crates/buzz-db/src/store/media_deletion/scan.rs new file mode 100644 index 00000000000..c2b526e78f7 --- /dev/null +++ b/crates/buzz-db/src/store/media_deletion/scan.rs @@ -0,0 +1,230 @@ +//! Consistent read-only PostgreSQL inventory. No membership-only filter: departed +//! users can still have events/uploads after leaving every channel. +use super::{invalid, MediaFile, MAX_FILES, MAX_FILE_BYTES}; +use crate::{Db, Result}; +use buzz_core::media_inventory::{references, valid_hash}; +use futures_util::TryStreamExt; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use std::collections::{BTreeMap, BTreeSet}; +use uuid::Uuid; + +/// Exact subject pairs. The human is selected across the database; agent +/// ownership is community-specific and never propagated to another community. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SubjectScope { + /// Normalized human public key. + pub human: String, + /// Explicitly included current owned-agent pairs. + pub agents: BTreeSet<(Uuid, String)>, +} +impl SubjectScope { + /// Whether a source is associated with the requested person/agent scope. + pub fn contains(&self, community: Option, key: Option<&str>) -> bool { + key == Some(self.human.as_str()) + || community + .zip(key) + .is_some_and(|(c, k)| self.agents.contains(&(c, k.into()))) + } +} +/// Evidence locates a source without persisting message bodies or private URLs. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MediaSource { + /// Table or storage-record source name. + pub table: String, + /// Origin community, nullable for operator-global records. + pub community: Option, + /// Complete source primary key. + pub key: Value, + /// Hash of inspected payload, for later source revalidation. + pub fingerprint: String, + /// Field path containing the reference. + pub field: String, + /// Source is associated with the requested subject. + pub selected_subject: bool, + /// Whether this is an uploader claim rather than a file reference. + pub upload_claim: bool, + /// True for a definite survivor, independent of future subject deletion. + pub surviving: bool, +} +/// Output of completed DB scans. S3 scan and execution prerequisites are separate. +#[derive(Debug, Serialize, Deserialize)] +pub struct MediaScan { + /// Database snapshot time; storage listing occurs separately afterwards. + pub database_snapshot_at: String, + /// Subject selection captured in this snapshot. + pub scope: SubjectScope, + /// Deduplicated reference hashes before joining storage-only upload candidates. + pub files: Vec, + /// Per-source inspected row counts (including sources with zero references). + pub scanned: BTreeMap, + /// Known coverage limitations; never converted into successful deletion checks. + pub unresolved: Vec, +} +#[derive(sqlx::FromRow)] +struct SourceRow { + source_table: String, + community_id: Option, + source_key: Value, + subject: Option, + payload: Value, + direct_hash: Option, + upload_claim: bool, + subject_owned: bool, +} + +impl Db { + /// Scan media-bearing records in all communities under one read-only repeatable + /// read transaction. Bounds fail the whole scan; no partial inventory is returned. + pub async fn scan_user_media( + &self, + human: &str, + include_agents: bool, + namespace: &str, + extra_bases: &[String], + ) -> Result { + let human = crate::user_deletion::normalize_user_deletion_pubkey(human)?; + let mut tx = self.pool().begin().await?; + sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY") + .execute(&mut *tx) + .await?; + sqlx::query("SET LOCAL statement_timeout = '120s'") + .execute(&mut *tx) + .await?; + let database_snapshot_at: String = + sqlx::query_scalar("SELECT transaction_timestamp()::text") + .fetch_one(&mut *tx) + .await?; + let agents: Vec<(Uuid, String)> = if include_agents { + sqlx::query_as("SELECT community_id,encode(pubkey,'hex') FROM users WHERE agent_owner_pubkey=decode($1,'hex') ORDER BY community_id,pubkey LIMIT 10001") + .bind(&human).fetch_all(&mut *tx).await? + } else { + vec![] + }; + if agents.len() > 10_000 { + return Err(invalid("owned-agent limit exceeded")); + } + let scope = SubjectScope { + human, + agents: agents.into_iter().collect(), + }; + let hosts: Vec = + sqlx::query_scalar("SELECT host FROM communities ORDER BY id LIMIT 10001") + .fetch_all(&mut *tx) + .await?; + if hosts.len() > 10_000 { + return Err(invalid("community limit exceeded")); + } + let mut bases = extra_bases.to_vec(); + bases.extend(hosts.into_iter().map(|h| format!("https://{h}/media"))); + // Root-relative references remain discoverable in an otherwise empty host registry. + bases.push("https://invalid.example/media".into()); + let mut scanned = BTreeMap::new(); + let mut sightings: BTreeMap> = BTreeMap::new(); + let mut total_bytes = 0usize; + let mut count = 0usize; + { + let mut rows = + sqlx::query_as::<_, SourceRow>(include_str!("sources.sql")).fetch(&mut *tx); + while let Some(row) = rows.try_next().await? { + count += 1; + if count > 1_000_000 { + return Err(invalid( + "source scan exceeds one million rows; no partial inventory published", + )); + } + *scanned.entry(row.source_table.clone()).or_insert(0) += 1; + let payload = serde_json::to_vec(&row.payload)?; + if payload.len() > MAX_FILE_BYTES { + return Err(invalid("media source exceeds parser byte limit")); + } + let mut refs = references(&row.payload, &bases).map_err(|e| invalid(&e))?; + if let Some(hash) = &row.direct_hash { + if !valid_hash(hash) { + return Err(invalid("invalid direct media hash")); + } + if row.upload_claim && row.payload["sha256"].as_str().is_some_and(|h| h != hash) + { + return Err(invalid("contradictory upload audit hash")); + } + refs.insert((hash.clone(), "direct_hash".into())); + } + let selected = scope.contains(row.community_id, row.subject.as_deref()); + let fingerprint = hex::encode(Sha256::digest(serde_json::to_vec( + &json!({"key":row.source_key,"community":row.community_id,"subject":row.subject,"payload":row.payload,"direct_hash":row.direct_hash,"upload_claim":row.upload_claim,"subject_owned":row.subject_owned}), + )?)); + for (hash, field) in refs { + let evidence = MediaSource { + table: row.source_table.clone(), + community: row.community_id, + key: row.source_key.clone(), + fingerprint: fingerprint.clone(), + field, + selected_subject: selected, + upload_claim: row.upload_claim, + surviving: !selected || !row.subject_owned, + }; + total_bytes += serde_json::to_vec(&evidence)?.len(); + if total_bytes > 128 * 1024 * 1024 { + return Err(invalid( + "media evidence exceeds 128 MiB; no partial inventory published", + )); + } + sightings.entry(hash).or_default().push(evidence); + if sightings.len() > MAX_FILES { + return Err(invalid("media reference file limit exceeded")); + } + } + } + } + // The parser/selector names are version-bound in the stage; zero table + // counts still appear in the scan receipt. + for table in [ + "events", + "users", + "nip_fi_profile_heads", + "workflows", + "workflow_runs", + "workflow_trigger_runs", + "workflow_approvals", + "subscriptions", + "delivery_log", + "product_feedback", + "channels", + "communities", + "moderation_reports", + "moderation_report_events", + "moderation_actions", + "relay_admin_actions", + "relay_admin_outbox", + "artifact_heads", + "media_uploaders", + "audit_log.media_uploaded", + ] { + scanned.entry(table.into()).or_insert(0); + } + tx.commit().await?; + let mut files = Vec::new(); + for (hash, sources) in sightings { + let file = MediaFile { + storage_namespace: namespace.into(), + file_hash: hash, + status: "unknown".into(), + evidence_json: json!({"sources":sources}), + delete_targets_json: json!([]), + reason: None, + last_error: None, + }; + if serde_json::to_vec(&file)?.len() > MAX_FILE_BYTES { + return Err(invalid("per-file evidence limit exceeded")); + } + files.push(file); + } + Ok(MediaScan {database_snapshot_at,scope,files,scanned,unresolved:vec![ + "historical unreferenced uploads without attribution cannot be assigned to a subject".into(), + "exact downstream source-removal approval is not available in a standalone preview".into(), + "all-writer file reservations are not installed".into(), + ]}) + } +} diff --git a/crates/buzz-db/src/store/media_deletion/sources.sql b/crates/buzz-db/src/store/media_deletion/sources.sql new file mode 100644 index 00000000000..e9f43a244d9 --- /dev/null +++ b/crates/buzz-db/src/store/media_deletion/sources.sql @@ -0,0 +1,36 @@ +-- Operator-only scan: all communities, private and historical records included. +-- Select media-bearing payloads explicitly; do not copy credentials or signing keys. +SELECT 'events'::text AS source_table, e.community_id, + jsonb_build_array(e.created_at,encode(e.id,'hex')) AS source_key, CASE WHEN e.kind=30382 AND e.d_tag ~ '^[0-9a-f]{64}$' THEN e.d_tag ELSE encode(e.pubkey,'hex') END AS subject, + jsonb_build_object('content',e.content,'tags',e.tags) AS payload, + NULL::text AS direct_hash, false AS upload_claim, true AS subject_owned +FROM events e +UNION ALL SELECT 'users',community_id,to_jsonb(encode(pubkey,'hex')),encode(pubkey,'hex'), + jsonb_build_object('avatar_url',avatar_url,'about',about,'capabilities',capabilities),NULL,false,true FROM users +UNION ALL SELECT 'nip_fi_profile_heads',community_id,to_jsonb(pubkey),pubkey,event,NULL,false,true FROM nip_fi_profile_heads +UNION ALL SELECT 'workflows',community_id,to_jsonb(id),encode(owner_pubkey,'hex'),definition,NULL,false,true FROM workflows +UNION ALL SELECT 'workflow_runs',r.community_id,to_jsonb(r.id),encode(w.owner_pubkey,'hex'), + jsonb_build_object('trigger_context',r.trigger_context,'execution_trace',r.execution_trace,'error_message',r.error_message),NULL,false,true + FROM workflow_runs r LEFT JOIN workflows w ON (w.community_id,w.id)=(r.community_id,r.workflow_id) +UNION ALL SELECT 'workflow_approvals',a.community_id,to_jsonb(encode(a.token,'hex')),encode(w.owner_pubkey,'hex'), + jsonb_build_object('note',a.note,'approver_spec',a.approver_spec),NULL,false,true + FROM workflow_approvals a LEFT JOIN workflows w ON (w.community_id,w.id)=(a.community_id,a.workflow_id) +UNION ALL SELECT 'subscriptions',community_id,to_jsonb(id),encode(owner_pubkey,'hex'),jsonb_build_object('delivery_url',delivery_url),NULL,false,true FROM subscriptions +UNION ALL SELECT 'delivery_log',l.community_id,jsonb_build_array(l.delivered_at,l.id),encode(s.owner_pubkey,'hex'),jsonb_build_object('error_message',l.error_message),NULL,false,true + FROM delivery_log l LEFT JOIN subscriptions s ON (s.community_id,s.id)=(l.community_id,l.subscription_id) +UNION ALL SELECT 'product_feedback',community_id,to_jsonb(id),encode(submitter_pubkey,'hex'),jsonb_build_object('body',body,'tags',tags),NULL,false,false FROM product_feedback +UNION ALL SELECT 'channels',community_id,to_jsonb(id),NULL,jsonb_build_object('canvas',canvas,'description',description,'topic',topic,'purpose',purpose),NULL,false,false FROM channels +UNION ALL SELECT 'communities',id,to_jsonb(id),NULL,jsonb_build_object('icon',icon),NULL,false,false FROM communities +UNION ALL SELECT 'moderation_reports',community_id,to_jsonb(id),encode(reporter_pubkey,'hex'),jsonb_build_object('note',note),encode(target_blob_sha256,'hex'),false,false FROM moderation_reports +UNION ALL SELECT 'moderation_report_events',r.community_id,to_jsonb(r.id),encode(r.reporter_pubkey,'hex'),jsonb_build_object('content',e.content,'tags',e.tags),NULL,false,false + FROM moderation_reports r JOIN events e ON e.community_id=r.community_id AND e.id=r.target_event_id +UNION ALL SELECT 'moderation_actions',community_id,to_jsonb(id),encode(actor_pubkey,'hex'),jsonb_build_object('public_reason',public_reason,'private_reason',private_reason),NULL,false,false FROM moderation_actions +UNION ALL SELECT 'relay_admin_actions',report_community_id,to_jsonb(id),encode(actor_pubkey,'hex'),jsonb_build_object('reason',reason,'error_message',error_message),NULL,false,false FROM relay_admin_actions +UNION ALL SELECT 'relay_admin_outbox',a.report_community_id,to_jsonb(o.id),encode(a.actor_pubkey,'hex'),jsonb_build_object('payload',o.payload,'error_message',o.error_message),NULL,false,false + FROM relay_admin_outbox o LEFT JOIN relay_admin_actions a ON a.id=o.action_id +UNION ALL SELECT 'artifact_heads',h.community_id,to_jsonb(h.artifact_id),NULL,jsonb_build_object('content',e.content,'tags',e.tags),NULL,false,false + FROM artifact_heads h JOIN events e ON e.community_id=h.community_id AND e.id=h.event_id WHERE NOT h.deleted +UNION ALL SELECT 'media_uploaders',community_id,jsonb_build_array(sha256,encode(pubkey,'hex')),encode(pubkey,'hex'),'{}'::jsonb,sha256,true,true FROM media_uploaders +UNION ALL SELECT 'audit_log.media_uploaded',community_id,to_jsonb(seq),encode(actor_pubkey,'hex'),jsonb_build_object('sha256',detail->'sha256'),object_id,true,true FROM audit_log WHERE action='media_uploaded' + +UNION ALL SELECT 'workflow_trigger_runs',r.community_id,to_jsonb(r.id),r.trigger_context->>'author',jsonb_build_object('trigger_context',r.trigger_context,'execution_trace',r.execution_trace,'error_message',r.error_message),NULL,false,false FROM workflow_runs r WHERE r.trigger_context->>'author' IS NOT NULL diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index d1cb647694b..b19dc6e32cf 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -121,3 +121,6 @@ pub mod media; mod media_postgres_tests; #[cfg(test)] mod message_edit_state_postgres_tests; + +/// Operator media inventory and deletion journal. +pub mod media_deletion; diff --git a/crates/buzz-deletion/src/user.rs b/crates/buzz-deletion/src/user.rs index 93d1766fdb3..542c4333981 100644 --- a/crates/buzz-deletion/src/user.rs +++ b/crates/buzz-deletion/src/user.rs @@ -7,6 +7,8 @@ use uuid::Uuid; /// Preview real memberships or prepare and execute synthetic deletion requests. #[derive(Debug, Subcommand)] pub enum Command { + /// Direct media discovery and inventory; print-only stops after step 2. + Media(media::Options), /// 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 +76,10 @@ pub enum Command { /// Run inventory discovery or synthetic workflow commands against PostgreSQL. pub async fn run(command: Command) -> Result { + let command = match command { + Command::Media(options) => return media::run(options).await, + other => other, + }; let db = if matches!(command, Command::Inventory { .. }) { buzz_db::Db::connect_cold_start( buzz_db::DbConfig { @@ -91,6 +97,7 @@ pub async fn run(command: Command) -> Result { }; let store = db.entity_deletion_store(); match command { + Command::Media(_) => anyhow::bail!("media command was already dispatched"), Command::Inventory { pubkey } => { crate::print_json(&inventory::generate(&db, &pubkey).await?)?; } @@ -159,3 +166,6 @@ pub mod summary; mod transition; mod worker; mod workflow; + +/// Media discovery, classification and protected execution. +pub mod media; diff --git a/crates/buzz-deletion/src/user/media/command.rs b/crates/buzz-deletion/src/user/media/command.rs new file mode 100644 index 00000000000..eb35404a24b --- /dev/null +++ b/crates/buzz-deletion/src/user/media/command.rs @@ -0,0 +1,116 @@ +//! Standalone preview: read-only source sessions, separate journal connection. +use super::*; +use buzz_db::{Db, DbConfig}; +use serde_json::Value; + +/// Direct command arguments. The mode is bound to the run and cannot change on retry. +#[derive(Debug, clap::Args)] +pub struct Options { + /// Human public key (hex or npub); demo input is not accepted. + #[arg(long,value_parser=buzz_db::user_deletion::normalize_user_deletion_pubkey)] + pub pubkey: String, + /// Stable UUID for retries. Omit to begin a fresh observation. + #[arg(long)] + pub run_id: Option, + /// Include agents currently owned by this person, separately in each community. + #[arg(long)] + pub include_owned_agents: bool, + /// Current direct execution supports print-only. Other modes require a protected workflow caller. + #[arg(long, value_enum, default_value = "print-only")] + pub mode: Mode, + /// Additional canonical media base URLs (including /media) for deployment aliases. + #[arg(long)] + pub media_base: Vec, +} + +/// Run the direct command. It never migrates either DB, fences subjects, mutates +/// source tables, removes uploader proofs, or calls storage deletion. +pub async fn run(options: Options) -> Result { + if options.mode != Mode::PrintOnly { + bail!("direct execution requires print-only; destructive modes require a workflow ProtectedMedia provider with durable writer reservations"); + } + let source_url = crate::required_env("BUZZ_MEDIA_SOURCE_DATABASE_URL")?; + let journal_url = crate::required_env("DATABASE_URL")?; + let endpoint = crate::required_env("BUZZ_S3_ENDPOINT")?; + let bucket = crate::required_env("BUZZ_S3_BUCKET")?; + let namespace = + buzz_db::entity_deletion::inventory_digest(&json!({"endpoint":endpoint,"bucket":bucket}))?; + let source_identity = buzz_db::entity_deletion::inventory_digest(&json!(source_url))?; + let run_id = options.run_id.unwrap_or_else(Uuid::new_v4); + eprintln!("media inventory run {run_id}; retry with --run-id {run_id}"); + let journal = Db::connect_cold_start( + DbConfig { + database_url: journal_url, + ..DbConfig::default() + }, + "media_journal", + ) + .await?; + let store = journal.media_deletion_store(); + let binding = json!({"version":1,"mode":"print-only","human":options.pubkey,"include_owned_agents":options.include_owned_agents,"source":source_identity,"namespace":namespace,"media_bases":options.media_base}); + let run = store.open_preview(run_id, binding).await?; + if store.load(&run).await?.is_none() { + let source = Db::connect_cold_start( + DbConfig { + database_url: source_url, + max_connections: 1, + min_connections: 0, + default_transaction_read_only: true, + ..DbConfig::default() + }, + "media_source", + ) + .await?; + let mut scan = source + .scan_user_media( + &options.pubkey, + options.include_owned_agents, + &namespace, + &options.media_base, + ) + .await?; + let (s3_access_key, s3_secret_key) = crate::s3_key_pair_from_env(); + let media = MediaStorage::new(&buzz_media::MediaConfig { + s3_endpoint: endpoint, + s3_bucket: bucket, + s3_access_key, + s3_secret_key, + s3_region: crate::s3_region_from_env(), + s3_addressing_style: std::env::var("BUZZ_S3_ADDRESSING_STYLE") + .unwrap_or_else(|_| "path".into()) + .parse() + .map_err(anyhow::Error::msg)?, + max_image_bytes: 1, + max_gif_bytes: 1, + max_video_bytes: 1, + max_file_bytes: 1, + public_base_url: "https://example.org/media".into(), + upload_records_enabled: false, + upload_ip_header: None, + upload_port_header: None, + })?; + storage::supplement(&mut scan, &media, &namespace).await?; + store + .publish( + &run, + &scan.files, + json!({"tables":scan.scanned,"scope":scan.scope,"unresolved":scan.unresolved,"database_snapshot_at":scan.database_snapshot_at,"storage_listing_finished_at":chrono::Utc::now()}), + ) + .await?; + } + let files = store + .load(&run) + .await? + .context("discovery did not publish")?; + for file in files.into_iter().filter(|f| f.status == "unknown") { + store + .save_preview_decision(&run, &classify_preview(file)?) + .await?; + } + let files = store.load(&run).await?.context("discovery missing")?; + let unknown = files.iter().filter(|f| f.status == "unknown").count(); + let discovery = store.receipt(&run).await?; + let output: Value = json!({"discovery":discovery,"run_id":run_id,"mode":"print-only","discovery_complete":true,"candidate_count":files.len(),"decision_complete":unknown==0,"execution_authorized":false,"unknown_count":unknown,"files":files}); + crate::print_json(&output)?; + Ok(if unknown == 0 { 0 } else { 2 }) +} diff --git a/crates/buzz-deletion/src/user/media/mod.rs b/crates/buzz-deletion/src/user/media/mod.rs new file mode 100644 index 00000000000..c9e957016cb --- /dev/null +++ b/crates/buzz-deletion/src/user/media/mod.rs @@ -0,0 +1,237 @@ +//! Direct media inventory and the shared three-step handler. The direct command +//! is observational; a workflow must supply durable protection before execution. +use anyhow::{bail, Context, Result}; +use buzz_db::media_deletion::{MediaDeletionStore, MediaFile, MediaRun, MediaTarget}; +use buzz_media::{MediaStorage, ObjectVersionRef}; +use clap::ValueEnum; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use uuid::Uuid; +mod command; +mod storage; +pub use command::{run, Options}; + +/// Output/execution policy for the media stage. +#[derive(Clone, Copy, Debug, Serialize, Deserialize, ValueEnum, PartialEq, Eq)] +#[serde(rename_all = "kebab-case")] +pub enum Mode { + /// Print saved step-2 inventory and stop before any deletion. + PrintOnly, + /// Print saved step-2 inventory, then execute its protected targets. + Print, + /// Execute protected targets without printing inventory. + NoPrint, +} + +/// The workflow's safety boundary, distinct from its worker lease. The provider +/// must verify approved scope/removal fingerprints, storage retention/holds, +/// complete cross-community scans and crash-persistent all-writer reservations. +/// There is deliberately no permissive production implementation until those +/// writer integrations exist. The standalone command cannot mint this authority. +#[allow(async_fn_in_trait)] +pub trait ProtectedMedia { + /// Validate the same durable protection and scope before each external action. + async fn verify(&self, run: &MediaRun) -> Result<()>; + /// Perform the final checks and return keep, unresolved or exact delete targets. + /// Failure must leave the saved row unknown, never manufacture a keep decision. + async fn decide(&self, run: &MediaRun, file: &MediaFile) -> Result; + /// Confirm whether this exact version (including delete markers) is present. + async fn present(&self, target: &MediaTarget) -> Result; + /// Delete only this exact version. Errors/timeout are retryable, never absence. + async fn delete(&self, target: &MediaTarget) -> Result<()>; + /// Remove only the exact departing uploader-proof rows after all selected + /// storage work succeeds. Must be idempotent; preserve media_publications. + async fn cleanup(&self, run: &MediaRun) -> Result<()>; +} + +/// Run steps 2 and 3 over a committed inventory. The caller publishes step 1 +/// through MediaDeletionStore::publish and owns workflow claim/heartbeat/routing. +/// PrintOnly always returns after printing, including when checks are unresolved. +pub async fn handle( + store: &MediaDeletionStore, + run: &MediaRun, + mode: Mode, + protected: &P, +) -> Result<()> { + let files = store + .load(run) + .await? + .context("media discovery has not completed")?; + for file in files.iter().filter(|f| f.status == "unknown") { + protected.verify(run).await?; + let decision = match protected.decide(run, file).await { + Ok(decision) => decision, + Err(error) => { + let mut pending = file.clone(); + pending.last_error = Some(format!("{error:#}")); + pending + } + }; + if decision.file_hash != file.file_hash + || decision.storage_namespace != file.storage_namespace + { + bail!("media decision changed file identity"); + } + store.save_execution_decision(run, &decision).await?; + } + let files = store + .load(run) + .await? + .context("media discovery disappeared")?; + if mode != Mode::NoPrint { + crate::print_json(&json!({"stage_id":run.id,"mode":mode,"files":files}))?; + } + if mode == Mode::PrintOnly { + return Ok(()); + } + if files.iter().any(|f| f.status == "unknown") { + bail!("media decisions unresolved; step 3 cannot start"); + } + for file in files.iter().filter(|f| f.status == "to_be_deleted") { + let targets: Vec = serde_json::from_value(file.delete_targets_json.clone())?; + let result = delete_targets(run, &targets, protected).await; + match result { + Ok(()) => store.save_execution_outcome(run, file, None).await?, + Err(error) => { + store + .save_execution_outcome(run, file, Some(&format!("{error:#}"))) + .await?; + return Err(error); + } + } + } + protected.verify(run).await?; + protected.cleanup(run).await?; + Ok(()) +} + +async fn delete_targets( + run: &MediaRun, + targets: &[MediaTarget], + protected: &P, +) -> Result<()> { + if targets.is_empty() { + bail!("delete decision has no targets"); + } + for target in targets { + if target.version_id.is_empty() || target.version_id == "null" { + bail!("unversioned deletion target"); + } + protected.verify(run).await?; + if protected.present(target).await? { + protected.verify(run).await?; + protected.delete(target).await?; + if protected.present(target).await? { + bail!("storage version still present after deletion"); + } + } + } + protected.verify(run).await?; + Ok(()) +} + +/// Exact-version storage operations for a workflow's protected adapter. +/// Listing proves absence of object versions and delete markers alike; a 403, +/// missing continuation or repeated page is an error, never an absent version. +pub async fn version_present(media: &MediaStorage, target: &MediaTarget) -> Result { + let (mut key, mut version) = (None, None); + let mut seen = std::collections::BTreeSet::new(); + for _ in 0..10_000 { + let page = media + .list_prefix_versions_page(&target.key, key.clone(), version.clone(), 1000) + .await?; + if page + .entries + .iter() + .any(|e| e.key == target.key && e.version_id == target.version_id) + { + return Ok(true); + } + if !page.is_truncated { + return Ok(false); + } + let next = (page.next_key_marker, page.next_version_id_marker); + if next.0.is_none() || !seen.insert(next.clone()) { + bail!("invalid storage listing continuation"); + } + (key, version) = next; + } + bail!("version verification exceeded page limit") +} + +/// Delete a single selected version, preserving per-version error attribution. +pub async fn delete_version(media: &MediaStorage, target: &MediaTarget) -> Result<()> { + if target.version_id.is_empty() || target.version_id == "null" { + bail!("immutable version required"); + } + let outcome = media + .delete_object_versions(&[ObjectVersionRef { + key: target.key.clone(), + version_id: target.version_id.clone(), + }]) + .await?; + if !outcome.failed.is_empty() { + bail!("storage refused selected version deletion"); + } + Ok(()) +} + +/// Classify an unfenced observation. Shared references are concrete keep evidence; +/// absence of known survivors cannot authorize deletion of a live file. +pub fn classify_preview(mut file: MediaFile) -> Result { + let sources = file.evidence_json["sources"] + .as_array() + .context("missing media source evidence")?; + let surviving = sources.iter().any(|s| s["surviving"] == true); + file.evidence_json["preview_checks"] = json!({ + "database_sources": "scanned", + "storage_versions": "listed", + "surviving_reference_or_upload": surviving, + "exact_source_removal_approval": "unresolved", + "historical_upload_coverage": "unresolved", + "version_retention_and_holds": "unresolved", + "upload_record_retention_policy": "unresolved", + "durable_all_writer_reservation": "unavailable" + }); + if surviving { + file.status = "dont_delete".into(); + file.reason = + Some("surviving reference or uploader claim; retain all components in preview".into()); + file.last_error = None; + } else { + file.last_error=Some("unresolved: exact source-removal approval, historical upload coverage, version retention/holds, upload-record retention policy, and durable file reservation".into()); + file.reason = Some( + "associated with subject; no known surviving reference in this observation".into(), + ); + } + Ok(file) +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn preview_does_not_confuse_missing_checks_with_keep_or_delete() { + let file = MediaFile { + storage_namespace: "fixture".into(), + file_hash: "a".repeat(64), + status: "unknown".into(), + evidence_json: json!({"sources":[{"selected_subject":true,"surviving":false}]}), + delete_targets_json: json!([]), + reason: None, + last_error: None, + }; + let unknown = classify_preview(file.clone()).unwrap(); + assert_eq!(unknown.status, "unknown"); + assert!(unknown.last_error.is_some()); + let mut shared = file; + shared.evidence_json["sources"] + .as_array_mut() + .unwrap() + .push(json!({"surviving":true})); + assert_eq!(classify_preview(shared).unwrap().status, "dont_delete"); + } +} + +#[cfg(test)] +mod postgres_tests; diff --git a/crates/buzz-deletion/src/user/media/postgres_tests.rs b/crates/buzz-deletion/src/user/media/postgres_tests.rs new file mode 100644 index 00000000000..656eeb0ddc5 --- /dev/null +++ b/crates/buzz-deletion/src/user/media/postgres_tests.rs @@ -0,0 +1,145 @@ +use super::*; +use buzz_db::{ + entity_deletion::{NewEntityStage, NextStage, StageTransition}, + Db, +}; +use std::{ + collections::BTreeSet, + sync::{ + atomic::{AtomicBool, AtomicUsize, Ordering}, + Mutex, + }, + time::Duration, +}; + +struct Backend { + present: Mutex>, + fail_after_delete: AtomicBool, + protected: AtomicBool, + deletes: AtomicUsize, + cleanups: AtomicUsize, +} +impl ProtectedMedia for Backend { + async fn verify(&self, _: &MediaRun) -> Result<()> { + if !self.protected.load(Ordering::SeqCst) { + bail!("protection lost"); + } + Ok(()) + } + async fn decide(&self, _: &MediaRun, file: &MediaFile) -> Result { + let mut file = file.clone(); + file.status = "to_be_deleted".into(); + file.delete_targets_json = json!([{"key":format!("{}.jpg",file.file_hash),"version_id":"v1"},{"key":format!("{}.thumb.webp",file.file_hash),"version_id":"v2"}]); + Ok(file) + } + async fn present(&self, t: &MediaTarget) -> Result { + Ok(self.present.lock().unwrap().contains(&t.version_id)) + } + async fn delete(&self, t: &MediaTarget) -> Result<()> { + self.present.lock().unwrap().remove(&t.version_id); + self.deletes.fetch_add(1, Ordering::SeqCst); + if self.fail_after_delete.swap(false, Ordering::SeqCst) { + bail!("lost acknowledgement"); + } + Ok(()) + } + async fn cleanup(&self, _: &MediaRun) -> Result<()> { + self.cleanups.fetch_add(1, Ordering::SeqCst); + Ok(()) + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn print_only_skips_deletion_and_crash_recovery_uses_saved_versions() { + let url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap(); + let db = Db::from_pool(sqlx::PgPool::connect(&url).await.unwrap()); + let request = db + .user_deletion_store() + .submit( + &format!("demo:media-loop-{}", Uuid::new_v4().simple()), + "operator", + None, + ) + .await + .unwrap(); + let requests = db.entity_deletion_store(); + requests + .approve( + request.id, + "approver", + request.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + let lease = requests + .claim(request.id, "worker", Duration::from_secs(120)) + .await + .unwrap() + .unwrap(); + let NextStage::Stage(stage) = requests + .transition_stage(&lease, None, |_, _| { + Ok(StageTransition::Initialize(NewEntityStage { + stage: "purge_media".into(), + community: None, + inputs: json!({}), + })) + }) + .await + .unwrap() + else { + panic!("missing stage") + }; + requests.start_stage(&lease, stage.id).await.unwrap(); + let store = db.media_deletion_store(); + let run = store + .execution_run(&lease, stage.id, json!({"scope":"fixture"})) + .await + .unwrap(); + store + .publish( + &run, + &[MediaFile { + storage_namespace: "fixture".into(), + file_hash: "a".repeat(64), + status: "unknown".into(), + evidence_json: json!({"sources":[]}), + delete_targets_json: json!([]), + reason: None, + last_error: None, + }], + json!({}), + ) + .await + .unwrap(); + let backend = Backend { + present: Mutex::new(BTreeSet::from(["v1".into(), "v2".into()])), + fail_after_delete: AtomicBool::new(true), + protected: AtomicBool::new(true), + deletes: AtomicUsize::new(0), + cleanups: AtomicUsize::new(0), + }; + handle(&store, &run, Mode::PrintOnly, &backend) + .await + .unwrap(); + assert_eq!(backend.deletes.load(Ordering::SeqCst), 0); + assert_eq!(backend.cleanups.load(Ordering::SeqCst), 0); + backend.protected.store(false, Ordering::SeqCst); + assert!(handle(&store, &run, Mode::NoPrint, &backend).await.is_err()); + assert_eq!(backend.deletes.load(Ordering::SeqCst), 0); + backend.protected.store(true, Ordering::SeqCst); + assert!(handle(&store, &run, Mode::NoPrint, &backend).await.is_err()); + let pending = store.load(&run).await.unwrap().unwrap().remove(0); + assert_eq!(pending.status, "to_be_deleted"); + assert!(pending.last_error.is_some()); + assert_eq!(backend.cleanups.load(Ordering::SeqCst), 0); + handle(&store, &run, Mode::NoPrint, &backend).await.unwrap(); + assert_eq!( + store.load(&run).await.unwrap().unwrap()[0].status, + "deleted" + ); + assert_eq!(backend.deletes.load(Ordering::SeqCst), 2); // v1 was not deleted twice. + assert_eq!(backend.cleanups.load(Ordering::SeqCst), 1); +} diff --git a/crates/buzz-deletion/src/user/media/storage.rs b/crates/buzz-deletion/src/user/media/storage.rs new file mode 100644 index 00000000000..c93e2e373b1 --- /dev/null +++ b/crates/buzz-deletion/src/user/media/storage.rs @@ -0,0 +1,144 @@ +//! Read-only object inventory. This adapter never has to call a delete method. +use anyhow::{bail, Context, Result}; +use buzz_db::media_deletion::{MediaFile, MediaScan, MediaSource, MAX_FILES}; +use buzz_media::{MediaStorage, ObjectVersionEntry, ObjectVersionKind}; +use serde_json::json; +use std::collections::{BTreeMap, BTreeSet}; +use uuid::Uuid; + +/// Enumerate originals, derivatives, sidecars and attribution versions. Unknown +/// layouts do not become deletion targets. Failed listings abort discovery. +pub(super) async fn supplement( + scan: &mut MediaScan, + media: &MediaStorage, + namespace: &str, +) -> Result<()> { + let mut files: BTreeMap = std::mem::take(&mut scan.files) + .into_iter() + .map(|f| (f.file_hash.clone(), f)) + .collect(); + let mut objects: BTreeMap> = BTreeMap::new(); + let mut key_marker = None; + let mut version_marker = None; + let mut seen = BTreeSet::new(); + let mut count = 0usize; + let mut object_bytes = 0usize; + loop { + let page = media + .list_prefix_versions_page("", key_marker.clone(), version_marker.clone(), 1000) + .await?; + for entry in page.entries { + object_bytes += serde_json::to_vec(&entry)?.len(); + if object_bytes > 128 * 1024 * 1024 { + bail!("storage inventory exceeds 128 MiB"); + } + count += 1; + if count > 1_000_000 { + bail!( + "storage listing exceeds one million versions; no partial discovery committed" + ); + } + let segments: Vec<_> = entry.key.split('/').collect(); + let hash = if segments.len() == 1 { + segments[0].split('.').next() + } else if segments.first() == Some(&"_uploads") && segments.len() == 4 { + Some(segments[2]) + } else if segments.first() == Some(&"_meta") && segments.len() == 3 { + segments[2].strip_suffix(".json") + } else { + None + }; + let Some(hash) = hash + .filter(|h| buzz_core::media_inventory::valid_hash(h)) + .map(str::to_owned) + else { + continue; + }; + if segments.first() == Some(&"_uploads") && entry.kind == ObjectVersionKind::Object { + let bytes = media + .get_small_version(&entry.key, &entry.version_id, 65_536) + .await?; + let record: serde_json::Value = + serde_json::from_slice(&bytes).context("invalid storage upload record")?; + let community = Uuid::parse_str( + record["community_id"] + .as_str() + .context("upload record missing community")?, + )?; + let actor = record["uploader_id"] + .as_str() + .context("upload record missing uploader")?; + let actor = buzz_db::user_deletion::normalize_user_deletion_pubkey(actor)?; + let event = record["event_id"] + .as_str() + .context("upload record missing ID")?; + let valid_ulid = event.len() == 26 + && event.as_bytes()[0] <= b'7' + && event + .bytes() + .all(|b| b"0123456789ABCDEFGHJKMNPQRSTVWXYZ".contains(&b)); + if record["version"] != 1 + || record["sha256"] != hash + || segments[1] != community.to_string() + || segments[3] != format!("{event}.json") + || !valid_ulid + { + bail!("storage upload record/path disagreement"); + } + let selected = scan.scope.contains(Some(community), Some(&actor)); + let source = MediaSource { + table: "storage_uploads".into(), + community: Some(community), + key: json!({"key":entry.key,"version_id":entry.version_id}), + fingerprint: buzz_db::entity_deletion::inventory_digest(&record)?, + field: "uploader_id".into(), + selected_subject: selected, + upload_claim: true, + surviving: !selected, + }; + let file = files.entry(hash.clone()).or_insert_with(|| MediaFile { + storage_namespace: namespace.into(), + file_hash: hash.clone(), + status: "unknown".into(), + evidence_json: json!({"sources":[]}), + delete_targets_json: json!([]), + reason: None, + last_error: None, + }); + file.evidence_json["sources"] + .as_array_mut() + .context("invalid file evidence")? + .push(serde_json::to_value(source)?); + } + objects.entry(hash).or_default().push(entry); + if files.len() > MAX_FILES || objects.len() > MAX_FILES { + bail!("storage file inventory limit exceeded"); + } + } + if !page.is_truncated { + break; + } + let next = (page.next_key_marker, page.next_version_id_marker); + if next.0.is_none() || !seen.insert(next.clone()) { + bail!("storage listing continuation missing or repeated"); + } + (key_marker, version_marker) = next; + } + scan.scanned.insert("storage_versions".into(), count); + for (hash, mut file) in files { + let selected = file.evidence_json["sources"] + .as_array() + .context("invalid source evidence")? + .iter() + .any(|s| s["selected_subject"] == true); + if !selected { + continue; + } + file.evidence_json["observed_objects"] = json!(objects.remove(&hash).unwrap_or_default()); + if serde_json::to_vec(&file)?.len() > buzz_db::media_deletion::MAX_FILE_BYTES { + bail!("file evidence/versions exceed size bound"); + } + scan.files.push(file); + } + Ok(()) +} diff --git a/crates/buzz-media/src/storage.rs b/crates/buzz-media/src/storage.rs index f12c0f00519..29abd0967e6 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -1119,3 +1119,38 @@ pub struct BlobMeta { #[serde(default, skip_serializing_if = "Option::is_none")] pub duration_secs: Option, } + +impl MediaStorage { + /// Read a small attribution record at its exact immutable version. A bounded + /// Range request avoids fetching arbitrary media bytes as upload metadata. + pub async fn get_small_version( + &self, + key: &str, + version_id: &str, + max_bytes: usize, + ) -> Result, MediaError> { + let bucket = self + .bucket + .with_extra_query(HashMap::from([( + "versionId".to_owned(), + version_id.to_owned(), + )])) + .map_err(|e| MediaError::StorageError(e.to_string()))?; + let response = bucket + .get_object_range(key, 0, Some(max_bytes as u64)) + .await + .map_err(|e| MediaError::StorageError(e.to_string()))?; + if response.status_code() != 200 && response.status_code() != 206 { + return Err(MediaError::StorageError(format!( + "version read failed: {}", + response.status_code() + ))); + } + if response.as_slice().len() > max_bytes { + return Err(MediaError::StorageError( + "upload record exceeds size bound".into(), + )); + } + Ok(response.to_vec()) + } +} diff --git a/docs/user-media-deletion.md b/docs/user-media-deletion.md new file mode 100644 index 00000000000..9d6f19f440e --- /dev/null +++ b/docs/user-media-deletion.md @@ -0,0 +1,160 @@ +# User media inventory and deletion + +`buzz-admin user-deletions media --pubkey --mode print-only` +collects media associated with a person across the source database's communities, +saves one row per file, classifies the observation, and prints JSON. Add +`--include-owned-agents` to include current agent ownership from +`users.agent_owner_pubkey`, separately in each community. Current memberships +are not required: old events and unposted uploads remain discoverable. + +This direct command **never deletes**. It does not install a user fence, approve +a deletion request, migrate a database, or enqueue the subject for deletion. + +## Connections and a first test + +Use an already-migrated local control database for `DATABASE_URL`. Set +`BUZZ_MEDIA_SOURCE_DATABASE_URL` to the source database; the command connects with +`default_transaction_read_only=true` and scans in a read-only repeatable-read +transaction. Source access must include private and historical rows in every +community sharing the configured storage namespace. Use SELECT-only credentials +when testing against a live deployment. Never run the migrator against that +source merely to make a preview work. + +Configure `BUZZ_S3_ENDPOINT`, `BUZZ_S3_BUCKET`, region/addressing style, and the +normal S3 credentials or workload identity. The direct command requires only +version listing and GET access. It scans historical upload-record versions as +well as current ones. A missing table, unreadable record, denied storage request, +malformed upload record, or exceeded resource limit fails discovery; none is +interpreted as an empty result. + +```sh +# Configure connection and storage environment through the operator's normal setup. +cargo run -p buzz-admin -- user-deletions media \ + --pubkey --include-owned-agents --mode print-only +``` + +Stdout is inventory JSON. Stderr prints a run UUID immediately; retry the same +command with `--run-id ` after an interruption. Use a **new** run UUID to +observe changed source data. A retry intentionally reprints the saved observation. +Additional canonical storage aliases can be supplied with repeated +`--media-base https://files.example.org/media` arguments. The run binds the +subject, agent-selection flag, source-connection fingerprint, storage namespace, +aliases and parser version. Changing any of these requires a new run. + +Exit 0 means no file decisions remain unknown, including a valid empty inventory. +Exit 2 prints the inventory but reports unresolved decisions. Other failures +produce an error, not partial discovery success. `execution_authorized` is always +false in a standalone observation, even when every file is kept. + +## Tables read + +The executable query catalog is +[`sources.sql`](../crates/buzz-db/src/store/media_deletion/sources.sql). +The database scan uses no visibility, membership, or soft-deletion filter. + +| Tables | Fields/relationship inspected | +|---|---| +| `users` | Current `agent_owner_pubkey`; avatar, about and capabilities | +| `events` | Content/tags for every stored kind, including messages, edits, profiles, notes, canvases, personas, teams and artifacts | +| `nip_fi_profile_heads` | Stored verified-profile event | +| `workflows` | Definition, attributed to `owner_pubkey` | +| `workflow_runs` | Trigger context, trace and errors; owner joined through `(community_id, workflow_id)`; separate association through `trigger_context.author` | +| `workflow_approvals` | Note/spec, with the same composite owner join | +| `subscriptions`, `delivery_log` | Delivery URL and delivery errors, attributed through subscription ownership | +| `product_feedback` | Body/tags; retained feedback protects its files | +| `channels`, `communities` | Canvas copy, descriptions and community icon; shared state protects its files | +| `moderation_reports` | Notes, direct blob hashes and referenced event payloads; retained evidence protects files | +| `moderation_actions`, `relay_admin_actions`, `relay_admin_outbox` | Reasons, errors and notice payloads; retained state protects files | +| `artifact_heads` | Live head's referenced event; shared head protects its files | +| `media_uploaders` | Recorded uploader/hash pairs, including unposted uploads | +| `audit_log` | `media_uploaded` actor/hash evidence; conflicting hashes fail the scan | + +The application parses JSON strings, attachment tags, free-text local media URLs +and encoded animated-avatar descriptors. Absolute foreign-host URLs are not +mapped to local objects. Evidence contains source keys, field paths and payload +fingerprints; it does not persist message bodies, upload IP addresses or signed +URLs. Source association is not a claim that the author originally uploaded the +file or that the source is approved for removal. + +Storage reads enumerate versions and delete markers. `_uploads//` +JSON records are read at their listed versions and validated against their path, +format, public key and ULID. Original/thumbnail keys and `_meta` sidecars are +recorded as observed objects. They are **not** executable targets until the +protected workflow checks each disposition and retention requirement. + +## Tables written and crash behavior + +Only one new table is added: `entity_deletion_media_files`. Its primary key is +`(stage_id, storage_namespace, file_hash)`. Every file row has one of +`unknown`, `dont_delete`, `to_be_deleted`, `deleted`, with per-file evidence, +targets, decision reason and error. + +The direct command creates an isolated `media-preview:` control record +in the existing `entity_deletion_requests`, terminal and ineligible for worker +claims. Its `media_preview` row in `entity_deletion_stage_history` is never the +current workflow stage. This records the observation without enrolling the +person in a destructive request. Control evidence intentionally survives +community deletion; no tenant data is owned by the new table. + +1. **Discover.** Collect and deduplicate the complete bounded observation in + memory. After DB and storage scans succeed, one transaction inserts all file + rows as `unknown` and writes `metadata_json.media_discovery` with `complete`, + count and scan receipt. A failed insert rolls back every row and the marker. + Zero files still commits the marker with count zero. A lost commit response is + resolved by reading the same stage. Partial rows never count as completion. +2. **Decide.** Process only unknown rows. Standalone preview keeps a file when a + definite surviving source or uploader claim protects it. Otherwise it records + the outstanding checks and leaves it unknown. Protected execution atomically + writes a complete immutable target list with `to_be_deleted`, or a keep reason + with `dont_delete`. A failed check remains unknown. A restart processes the + remaining unknown rows, without replacing a published inventory. +3. **Execute.** Require the discovery marker, matching row count, no unknown + decisions, current request ownership and continuous external-write protection. + Delete the exact recorded versions; verify their absence; only then change the + file row to `deleted`. A crash after storage deletion but before the row update + leaves `to_be_deleted`; retry verifies the same versions and finishes the row. + If only a thumbnail was removed, retry still handles every saved target. + Permission errors and timeouts are never absence. Finally remove the exact + approved uploader proofs idempotently; preserve sticky `media_publications`. + Source-message removal and workflow advancement belong to later handlers. + +`deleted` describes **all selected targets being confirmed absent**. It does not +claim that intentionally retained components or a later re-upload were erased. +A shared original and separately eligible metadata need separate dispositions +inside that file's target/evidence fields. + +Every control write checks the run generation. An older invocation cannot +publish after a replacement starts. Workflow writes additionally validate the +existing request lease and inventory approval. A worker lease does not by itself +protect media from new uploads or references. + +## Shared handler modes and remaining integration + +The library handler supports `print-only` (print and return), `print` (print then +execute) and `no-print` (execute without inventory output). Its exact-version +deletion and recovery loop is implemented and tested. The standalone command +rejects the latter two modes before connecting to any service. + +The live deployment adapter must implement `ProtectedMedia`: verify approved +scope, exact downstream removal selectors/fingerprints, complete source and +uploader coverage, durable reservations respected by **all** writers, object +holds/retention and upload-record retention policy; perform idempotent uploader +cleanup. Main does not yet have that reservation protocol or the complete +removal/retention proof. There is no permissive production adapter or flag that +asserts these checks passed. Consequently standalone preview deliberately does +not produce `to_be_deleted` rows. Its observed storage versions are useful for +inspection, not approval to delete later. These are explicit integration gaps, +not functionality claimed by this PR. + +The source catalog also needs downstream-selector integration for dependent +records owned by another author and historical relay-signed attribution. Missing +historical orphan-upload evidence cannot be reconstructed from file hashes. +The discovery marker certifies that the **defined scans** finished, not that +historical data was fully recorded or that all future source formats are covered. + +Initial limits are one million source rows, one million storage versions, +100,000 distinct hashes, 128 MiB of evidence/storage descriptors, 1 MiB per +source/file document and 64 KiB per upload record. Exceeding a limit reports an +error without publishing partial discovery. Scale measurements may justify a +later unpublished-generation design; this implementation never quietly commits +individual discovery pages. diff --git a/migrations/0087_entity_deletion_media.sql b/migrations/0087_entity_deletion_media.sql new file mode 100644 index 00000000000..66611ddccab --- /dev/null +++ b/migrations/0087_entity_deletion_media.sql @@ -0,0 +1,24 @@ +SET LOCAL lock_timeout = '5s'; +SET LOCAL statement_timeout = '30s'; + +-- Operator-global recovery evidence, retained after community deletion. +-- One logical file per stage; JSON is bounded per file by the application. +CREATE TABLE entity_deletion_media_files ( + stage_id UUID NOT NULL, + storage_namespace TEXT NOT NULL, + file_hash TEXT NOT NULL CHECK (file_hash ~ '^[0-9a-f]{64}$'), + status TEXT NOT NULL DEFAULT 'unknown' + CHECK (status IN ('unknown', 'dont_delete', 'to_be_deleted', 'deleted')), + evidence_json JSONB NOT NULL, + delete_targets_json JSONB NOT NULL DEFAULT '[]'::jsonb + CHECK (jsonb_typeof(delete_targets_json) = 'array'), + reason TEXT, + last_error TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (stage_id, storage_namespace, file_hash), + CHECK (status NOT IN ('to_be_deleted', 'deleted') OR jsonb_array_length(delete_targets_json) > 0) +); + +INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('entity_deletion_media_files', 'Operator-global media inventory and recovery evidence; retained after community deletion') +ON CONFLICT (table_name) DO NOTHING; diff --git a/schema/schema.sql b/schema/schema.sql index f804fb0bc4a..af7dbdecb82 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -223,3 +223,5 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i tables/public/media_uploaders.sql \i tables/public/media_publications.sql +\i tables/public/entity_deletion_media_files.sql +\i seeds/entity_deletion_media_files.sql diff --git a/schema/seeds/entity_deletion_media_files.sql b/schema/seeds/entity_deletion_media_files.sql new file mode 100644 index 00000000000..4b71891533e --- /dev/null +++ b/schema/seeds/entity_deletion_media_files.sql @@ -0,0 +1,3 @@ +INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('entity_deletion_media_files', 'Operator-global media inventory and recovery evidence; retained after community deletion') +ON CONFLICT (table_name) DO NOTHING; diff --git a/schema/tables/public/entity_deletion_media_files.sql b/schema/tables/public/entity_deletion_media_files.sql new file mode 100644 index 00000000000..c668fe21e18 --- /dev/null +++ b/schema/tables/public/entity_deletion_media_files.sql @@ -0,0 +1,17 @@ +-- Operator-global recovery evidence, retained after community deletion. +-- One logical file per stage; JSON is bounded per file by the application. +CREATE TABLE entity_deletion_media_files ( + stage_id UUID NOT NULL, + storage_namespace TEXT NOT NULL, + file_hash TEXT NOT NULL CHECK (file_hash ~ '^[0-9a-f]{64}$'), + status TEXT NOT NULL DEFAULT 'unknown' + CHECK (status IN ('unknown', 'dont_delete', 'to_be_deleted', 'deleted')), + evidence_json JSONB NOT NULL, + delete_targets_json JSONB NOT NULL DEFAULT '[]'::jsonb + CHECK (jsonb_typeof(delete_targets_json) = 'array'), + reason TEXT, + last_error TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (stage_id, storage_namespace, file_hash), + CHECK (status NOT IN ('to_be_deleted', 'deleted') OR jsonb_array_length(delete_targets_json) > 0) +); From 4f33216073bd8dfe95b61141fe4e11bd9e668c17 Mon Sep 17 00:00:00 2001 From: Ravneet Arora Date: Tue, 10 Nov 2026 13:35:50 -0500 Subject: [PATCH 2/2] feat(deletion): configure media scan budgets Allow larger read-only inventories while preserving atomic failure when the source scan exceeds its budget. Co-authored-by: dessert <7d616e7e143e5d408159614dccff1371350a1c1bd36c8de838a45ece5aca0aca@buzz.block.builderlab.xyz> Signed-off-by: Ravneet Arora --- .../tests/postgres_media_inventory.rs | 20 ++++++-- .../buzz-db/src/store/media_deletion/mod.rs | 2 +- .../buzz-db/src/store/media_deletion/scan.rs | 47 +++++++++++++++++-- .../buzz-deletion/src/user/media/command.rs | 18 ++++++- docs/user-media-deletion.md | 5 ++ 5 files changed, 82 insertions(+), 10 deletions(-) diff --git a/crates/buzz-admin/tests/postgres_media_inventory.rs b/crates/buzz-admin/tests/postgres_media_inventory.rs index 057a7a7f2f6..81b703ca520 100644 --- a/crates/buzz-admin/tests/postgres_media_inventory.rs +++ b/crates/buzz-admin/tests/postgres_media_inventory.rs @@ -89,7 +89,7 @@ async fn preview_cli_saves_inventory_resumes_and_never_calls_delete() { .fallback(any(storage)) .with_state(state.clone()); let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); - let invoke = || { + let invoke = |run_id: Uuid, max_source_rows: usize| { let mut command = Command::new(binary()); command .args([ @@ -98,9 +98,13 @@ async fn preview_cli_saves_inventory_resumes_and_never_calls_delete() { "--pubkey", &human, "--run-id", - &run.to_string(), + &run_id.to_string(), "--mode", "print-only", + "--max-source-rows", + &max_source_rows.to_string(), + "--scan-timeout-secs", + "900", ]) .env("DATABASE_URL", &url) .env("BUZZ_MEDIA_SOURCE_DATABASE_URL", &url) @@ -111,7 +115,15 @@ async fn preview_cli_saves_inventory_resumes_and_never_calls_delete() { .env("BUZZ_S3_ADDRESSING_STYLE", "path"); command }; - let first = invoke().output().await.unwrap(); + let capped_run = Uuid::new_v4(); + let capped = invoke(capped_run, 1).output().await.unwrap(); + assert!(!capped.status.success()); + assert!(String::from_utf8_lossy(&capped.stderr).contains("max-source-rows")); + assert!(state.methods.lock().unwrap().is_empty()); + let unpublished: bool = sqlx::query_scalar("SELECT NOT (metadata_json ? 'media_discovery') AND NOT EXISTS (SELECT 1 FROM entity_deletion_media_files WHERE stage_id=$1) FROM entity_deletion_stage_history WHERE id=$1") + .bind(capped_run).fetch_one(&pool).await.unwrap(); + assert!(unpublished); + let first = invoke(run, 10_000_000).output().await.unwrap(); assert_eq!( first.status.code(), Some(2), @@ -139,7 +151,7 @@ async fn preview_cli_saves_inventory_resumes_and_never_calls_delete() { assert!(!methods.is_empty()); assert!(methods.iter().all(|m| *m == Method::GET)); server.abort(); // Recovery must use the saved inventory even with S3 offline. - let second = invoke().output().await.unwrap(); + let second = invoke(run, 10_000_000).output().await.unwrap(); assert_eq!( second.status.code(), Some(2), diff --git a/crates/buzz-db/src/store/media_deletion/mod.rs b/crates/buzz-db/src/store/media_deletion/mod.rs index de31d059019..44582208363 100644 --- a/crates/buzz-db/src/store/media_deletion/mod.rs +++ b/crates/buzz-db/src/store/media_deletion/mod.rs @@ -7,7 +7,7 @@ use uuid::Uuid; mod execution; mod scan; -pub use scan::{MediaScan, MediaSource, SubjectScope}; +pub use scan::{MediaScan, MediaScanBudget, MediaSource, SubjectScope}; /// Maximum journal payload per file; exceeding it fails rather than truncates. pub const MAX_FILE_BYTES: usize = 1_048_576; diff --git a/crates/buzz-db/src/store/media_deletion/scan.rs b/crates/buzz-db/src/store/media_deletion/scan.rs index c2b526e78f7..e1de668586b 100644 --- a/crates/buzz-db/src/store/media_deletion/scan.rs +++ b/crates/buzz-db/src/store/media_deletion/scan.rs @@ -74,6 +74,23 @@ struct SourceRow { subject_owned: bool, } +/// Operator-selected work limits; exceeding either aborts the entire observation. +#[derive(Clone, Copy, Debug, Serialize, Deserialize)] +pub struct MediaScanBudget { + /// Maximum inspected database records, including records without references. + pub max_source_rows: usize, + /// Per-statement deadline for the read-only source scan. + pub statement_timeout_secs: u64, +} +impl Default for MediaScanBudget { + fn default() -> Self { + Self { + max_source_rows: 1_000_000, + statement_timeout_secs: 120, + } + } +} + impl Db { /// Scan media-bearing records in all communities under one read-only repeatable /// read transaction. Bounds fail the whole scan; no partial inventory is returned. @@ -84,12 +101,36 @@ impl Db { namespace: &str, extra_bases: &[String], ) -> Result { + self.scan_user_media_with_budget( + human, + include_agents, + namespace, + extra_bases, + MediaScanBudget::default(), + ) + .await + } + + /// Scan with explicit limits for a larger source database. Memory and per-file + /// bounds still apply; a larger row budget never permits partial publication. + pub async fn scan_user_media_with_budget( + &self, + human: &str, + include_agents: bool, + namespace: &str, + extra_bases: &[String], + budget: MediaScanBudget, + ) -> Result { + if budget.max_source_rows == 0 || budget.statement_timeout_secs == 0 { + return Err(invalid("media scan budgets must be positive")); + } let human = crate::user_deletion::normalize_user_deletion_pubkey(human)?; let mut tx = self.pool().begin().await?; sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY") .execute(&mut *tx) .await?; - sqlx::query("SET LOCAL statement_timeout = '120s'") + sqlx::query("SELECT set_config('statement_timeout', $1, true)") + .bind(format!("{}s", budget.statement_timeout_secs)) .execute(&mut *tx) .await?; let database_snapshot_at: String = @@ -129,9 +170,9 @@ impl Db { sqlx::query_as::<_, SourceRow>(include_str!("sources.sql")).fetch(&mut *tx); while let Some(row) = rows.try_next().await? { count += 1; - if count > 1_000_000 { + if count > budget.max_source_rows { return Err(invalid( - "source scan exceeds one million rows; no partial inventory published", + "source scan exceeds max-source-rows; no partial inventory published", )); } *scanned.entry(row.source_table.clone()).or_insert(0) += 1; diff --git a/crates/buzz-deletion/src/user/media/command.rs b/crates/buzz-deletion/src/user/media/command.rs index eb35404a24b..e0775125de0 100644 --- a/crates/buzz-deletion/src/user/media/command.rs +++ b/crates/buzz-deletion/src/user/media/command.rs @@ -21,6 +21,12 @@ pub struct Options { /// Additional canonical media base URLs (including /media) for deployment aliases. #[arg(long)] pub media_base: Vec, + /// Maximum source records to inspect; exceeding it publishes no inventory. + #[arg(long, default_value_t = 1_000_000)] + pub max_source_rows: usize, + /// Per-statement timeout for the read-only source scan. + #[arg(long, default_value_t = 120)] + pub scan_timeout_secs: u64, } /// Run the direct command. It never migrates either DB, fences subjects, mutates @@ -29,6 +35,13 @@ pub async fn run(options: Options) -> Result { if options.mode != Mode::PrintOnly { bail!("direct execution requires print-only; destructive modes require a workflow ProtectedMedia provider with durable writer reservations"); } + if options.max_source_rows == 0 || options.scan_timeout_secs == 0 { + bail!("media scan budgets must be positive"); + } + let budget = buzz_db::media_deletion::MediaScanBudget { + max_source_rows: options.max_source_rows, + statement_timeout_secs: options.scan_timeout_secs, + }; let source_url = crate::required_env("BUZZ_MEDIA_SOURCE_DATABASE_URL")?; let journal_url = crate::required_env("DATABASE_URL")?; let endpoint = crate::required_env("BUZZ_S3_ENDPOINT")?; @@ -47,7 +60,7 @@ pub async fn run(options: Options) -> Result { ) .await?; let store = journal.media_deletion_store(); - let binding = json!({"version":1,"mode":"print-only","human":options.pubkey,"include_owned_agents":options.include_owned_agents,"source":source_identity,"namespace":namespace,"media_bases":options.media_base}); + let binding = json!({"version":1,"mode":"print-only","human":options.pubkey,"include_owned_agents":options.include_owned_agents,"source":source_identity,"namespace":namespace,"media_bases":options.media_base,"scan_budget":budget}); let run = store.open_preview(run_id, binding).await?; if store.load(&run).await?.is_none() { let source = Db::connect_cold_start( @@ -62,11 +75,12 @@ pub async fn run(options: Options) -> Result { ) .await?; let mut scan = source - .scan_user_media( + .scan_user_media_with_budget( &options.pubkey, options.include_owned_agents, &namespace, &options.media_base, + budget, ) .await?; let (s3_access_key, s3_secret_key) = crate::s3_key_pair_from_env(); diff --git a/docs/user-media-deletion.md b/docs/user-media-deletion.md index 9d6f19f440e..6ae9f450f53 100644 --- a/docs/user-media-deletion.md +++ b/docs/user-media-deletion.md @@ -158,3 +158,8 @@ source/file document and 64 KiB per upload record. Exceeding a limit reports an error without publishing partial discovery. Scale measurements may justify a later unpublished-generation design; this implementation never quietly commits individual discovery pages. + +For a larger database, set `--max-source-rows 10000000 --scan-timeout-secs 900` +on a new run. These positive limits are bound to that run and do not relax the +memory, per-file or storage limits. Choose a reader endpoint and a budget suited +to the deployment; increasing the budget allows more work, not partial success.