feat(deletion): add resumable media inventory

crates/buzz-admin/tests/postgres_media_inventory.rs

@@ -0,0 +1,192 @@
+//! 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<Mutex<Vec<Method>>>,
+}
+async fn storage(State(state): State<S3Fixture>, 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!("<ListVersionsResult><IsTruncated>false</IsTruncated><Version><Key>{}.jpg</Key><VersionId>fixture-v1</VersionId><Size>4</Size><IsLatest>true</IsLatest></Version><Version><Key>{}</Key><VersionId>upload-v1</VersionId><Size>{}</Size><IsLatest>false</IsLatest></Version></ListVersionsResult>",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 = |run_id: Uuid, max_source_rows: usize| {
+        let mut command = Command::new(binary());
+        command
+            .args([
+                "user-deletions",
+                "media",
+                "--pubkey",
+                &human,
+                "--run-id",
+                &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)
+            .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 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),
+        "{}",
+        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(run, 10_000_000).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"));
+}

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;

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<BTreeSet<(String, String)>, 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::<Value>(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());
+    }
+}

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;

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",
         ] {

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)