feat(relay): record where media is shared and who uploaded it

ARCHITECTURE.md

@@ -504,6 +504,8 @@ shape is an explicit transition, not a flag day:
   never spans communities. Serving leases use the same application admission.
   Startup and deletion catalog validation tolerate the universal fence objects
   being absent; inventories still record live fences for older executors.
+  Desired-state bootstraps install only explicitly declared legacy fences.
+  New tenant-owned tables rely on application admission, without a fence trigger.
 - **Replica floor (transition foundation):** runtime now provides a shared
   replica-floor lock helper for compliant channel-event writers, and the writer
   probe handshake takes the exclusive counterpart before sampling `S`, scanning

crates/buzz-admin/src/main.rs

@@ -100,6 +100,22 @@ enum Command {
         #[arg(long, default_value_t = 10_000_000)]
         max_objects: u64,
     },
+    /// Record media shares for events stored before share recording existed.
+    ///
+    /// Run after every relay pod records shares and
+    /// before enforcing channel-scoped media reads. Historical references are
+    /// trusted only before each community's schema-recorded cutoff.
+    /// Restores uploader proof from surviving server media_uploaded audit rows;
+    /// files without that evidence still need an upload of the exact bytes.
+    /// Later references use current permissions, so any run can grant references
+    /// live recording previously refused if their authors have since gained access.
+    /// Accept the resulting lasting grants before the first backfill; otherwise
+    /// defer it. Each invocation restarts the scan and deduplicates existing rows.
+    BackfillMediaShares {
+        /// Events read and recorded per transaction.
+        #[arg(long, default_value_t = 1000, value_parser = clap::value_parser!(i64).range(1..))]
+        batch_size: i64,
+    },
     /// Run the partition catalog audit using a read-only database session.
     PartitionAudit {
         /// Future months to include in the coverage check.
@@ -193,6 +209,14 @@ async fn run(cli: Cli) -> Result<i32> {
             Ok(0)
         }
         Command::StorageSnapshot { max_objects } => cmd_storage_snapshot(max_objects).await,
+        Command::BackfillMediaShares { batch_size } => {
+            let added = connect_db()
+                .await?
+                .backfill_media_shares(batch_size)
+                .await?;
+            println!("Recorded {added} media shares.");
+            Ok(0)
+        }
         Command::PartitionAudit { months_ahead } => cmd_partition_audit(months_ahead).await,
         Command::AddMember { pubkey, role } => cmd_add_member(pubkey, role).await,
         Command::RemoveMember { pubkey, role } => cmd_remove_member(pubkey, role).await,

crates/buzz-core/src/lib.rs

@@ -24,6 +24,8 @@ pub mod git_perms;
 pub mod invite;
 /// Buzz kind number registry — custom event type constants.
 pub mod kind;
+/// Which media blobs an event references, for channel-scoped media access.
+pub mod media;
 /// Network utilities — SSRF-safe IP classification.
 pub mod network;
 /// NIP-10 thread-marker parsing — shared `root`/`reply` marker resolver.

crates/buzz-core/src/media.rs

@@ -0,0 +1,151 @@
+//! Which media blobs an event shares, for channel-scoped media access.
+
+use std::collections::BTreeSet;
+
+use crate::kind::{
+    event_is_shared, is_author_only_event_kind, is_shared_gated_kind, KIND_AGENT_ENGRAM,
+    P_GATED_KINDS,
+};
+
+/// Hashes of every blob `event` shares into its channel, or with the whole
+/// community when it is channel-less: each local media path `/media/<sha256>`
+/// anywhere in its content or tags (attachments, posters, thumbnails, link
+/// previews, avatars, custom emoji, pasted links).
+///
+/// Empty for an event only its author or tagged recipients can read, since
+/// referencing a blob there shows it to no one new.
+pub fn shared_blob_hashes(event: &nostr::Event) -> Vec<String> {
+    let kind = u32::from(event.kind.as_u16());
+    if is_author_only_event_kind(event)
+        || P_GATED_KINDS.contains(&kind)
+        || kind == KIND_AGENT_ENGRAM
+        || (is_shared_gated_kind(kind) && !event_is_shared(event))
+    {
+        return Vec::new();
+    }
+    // JSON content may escape `/` as `\/`; scan the path as clients will see it.
+    let content = event.content.replace("\\/", "/");
+    let mut hashes: BTreeSet<String> = media_path_hashes(&content).collect();
+    for tag in event.tags.iter() {
+        hashes.extend(
+            tag.as_slice()
+                .iter()
+                .flat_map(|part| media_path_hashes(part)),
+        );
+    }
+    hashes.into_iter().collect()
+}
+
+/// The media blob used as a kind:0 profile's picture. A registered agent may
+/// publish this one field using its owner's upload proof; arbitrary content
+/// and tags must not inherit that authority.
+pub fn profile_picture_blob_hash(event: &nostr::Event) -> Option<String> {
+    if event.kind != nostr::Kind::Metadata {
+        return None;
+    }
+    let profile: serde_json::Value = serde_json::from_str(&event.content).ok()?;
+    let hash = media_path_hashes(profile.get("picture")?.as_str()?).next();
+    hash
+}
+
+fn is_sha256_hex(value: &str) -> bool {
+    value.len() == 64 && value.chars().all(|c| matches!(c, '0'..='9' | 'a'..='f'))
+}
+
+/// The hash in each `/media/<sha256>` path in `text`, where the hash is not
+/// followed by another hex digit.
+fn media_path_hashes(text: &str) -> impl Iterator<Item = String> + '_ {
+    text.match_indices("/media/").filter_map(|(at, path)| {
+        let rest = &text[at + path.len()..];
+        let hash = rest.get(..64)?;
+        let bounded = !rest[64..].starts_with(|c: char| matches!(c, '0'..='9' | 'a'..='f'));
+        (is_sha256_hex(hash) && bounded).then(|| hash.to_owned())
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::kind::{KIND_EVENT_REMINDER, KIND_PERSONA};
+    use nostr::{EventBuilder, Keys, Kind, Tag};
+
+    fn media_url(sha256: &str, ext: &str) -> String {
+        format!("https://relay.example/media/{sha256}.{ext}")
+    }
+
+    fn signed(builder: EventBuilder) -> nostr::Event {
+        builder.sign_with_keys(&Keys::generate()).unwrap()
+    }
+
+    #[test]
+    fn shares_every_local_media_path() {
+        let (video, poster, embed) = ("a".repeat(64), "c".repeat(64), "d".repeat(64));
+        let event = signed(
+            EventBuilder::new(
+                Kind::TextNote,
+                format!("look ![image]({})", media_url(&embed, "png")),
+            )
+            .tags([
+                Tag::parse([
+                    "imeta",
+                    &format!("url {}", media_url(&video, "mp4")),
+                    &format!("x {video}"),
+                    &format!("image /media/{poster}.jpg"),
+                ])
+                .unwrap(),
+                Tag::parse([
+                    "r",
+                    &format!("https://elsewhere.example/{}", "e".repeat(64)),
+                ])
+                .unwrap(),
+                Tag::parse(["r", &format!("/media/{}0", "f".repeat(64))]).unwrap(),
+            ]),
+        );
+        assert_eq!(shared_blob_hashes(&event), vec![video, poster, embed]);
+    }
+
+    #[test]
+    fn shares_avatars_emoji_and_shared_personas_only() {
+        let (avatar, emoji) = ("c".repeat(64), "d".repeat(64));
+        let content = format!(r#"{{"picture":"{}"}}"#, media_url(&avatar, "jpg"));
+        let profile = signed(EventBuilder::new(Kind::Metadata, &content));
+        assert_eq!(shared_blob_hashes(&profile), vec![avatar.clone()]);
+        let escaped = signed(EventBuilder::new(
+            Kind::Metadata,
+            content.replace('/', "\\/"),
+        ));
+        assert_eq!(shared_blob_hashes(&escaped), vec![avatar.clone()]);
+
+        let emoji_set = signed(
+            EventBuilder::new(Kind::from(30030), "").tags([Tag::parse([
+                "emoji",
+                "party",
+                &media_url(&emoji, "png"),
+            ])
+            .unwrap()]),
+        );
+        assert_eq!(shared_blob_hashes(&emoji_set), vec![emoji]);
+
+        let persona = |shared: bool| {
+            let mut tags = vec![Tag::parse(["d", "helper"]).unwrap()];
+            if shared {
+                tags.push(Tag::parse(["shared", "true"]).unwrap());
+            }
+            signed(EventBuilder::new(Kind::from(KIND_PERSONA as u16), &content).tags(tags))
+        };
+        assert_eq!(shared_blob_hashes(&persona(true)), vec![avatar]);
+        assert!(shared_blob_hashes(&persona(false)).is_empty());
+    }
+
+    #[test]
+    fn author_only_events_share_nothing() {
+        let reminder = signed(
+            EventBuilder::new(Kind::from(KIND_EVENT_REMINDER as u16), "").tags([Tag::parse([
+                "imeta",
+                &format!("url /media/{}.jpg", "a".repeat(64)),
+            ])
+            .unwrap()]),
+        );
+        assert!(shared_blob_hashes(&reminder).is_empty());
+    }
+}

crates/buzz-db/src/lib.rs

@@ -66,10 +66,10 @@ pub(crate) use runtime::{
 };
 pub use store::{
     admin_moderation, allowlist, api_token, archived_identities, artifact, channel,
-    channel_members, community, deletion, dm, event, feed, git_repo, moderation, nip_fi_binding,
-    operator_listener, partition, personal_read, product_feedback, push, reaction, read_state,
-    relay_admin_actions, relay_invite, relay_members, relay_operators, reminder, replaceable,
-    storage_accounting, thread, thread_window, usage, user, workflow,
+    channel_members, community, deletion, dm, event, feed, git_repo, media_cleanup, moderation,
+    nip_fi_binding, operator_listener, partition, personal_read, product_feedback, push, reaction,
+    read_state, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder,
+    replaceable, storage_accounting, thread, thread_window, usage, user, workflow,
 };
 
 pub use allowlist::AllowlistEntry;

crates/buzz-db/src/runtime/migration.rs

@@ -14,6 +14,10 @@ use crate::Result;
 
 static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("../../migrations");
 
+#[cfg(test)]
+#[path = "tests/schema_fence_bootstrap_postgres_tests.rs"]
+mod schema_fence_bootstrap_postgres_tests;
+
 /// Run all pending Buzz database migrations.
 ///
 /// The entire run holds the exclusive [`SCHEMA_DESTRUCTION_LOCK_KEY`] session
@@ -769,7 +773,7 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 67);
+        assert_eq!(migrations.len(), 68);
         assert_eq!(migrations[64].version, 65);
         let retire_ttl = strip_sql_comments(migrations[64].sql.as_str());
         for statement in [
@@ -806,6 +810,16 @@ mod postgres_tests {
             .sql
             .as_str()
             .contains("CREATE TABLE community_nip_fi_issuers"));
+        assert_eq!(migrations[67].version, 68);
+        assert!(strip_sql_comments(migrations[67].sql.as_str())
+            .trim_start()
+            .starts_with("SET LOCAL lock_timeout = '5s';"));
+        for table in ["media_uploaders", "media_shares"] {
+            assert!(migrations[67]
+                .sql
+                .as_str()
+                .contains(&format!("CREATE TABLE {table}")));
+        }
         assert_eq!(migrations[55].version, 56);
         assert!(migrations[55]
             .sql

crates/buzz-db/src/runtime/tests/schema_fence_bootstrap_postgres_tests.rs

@@ -0,0 +1,39 @@
+use sqlx::PgPool;
+
+/// A fresh bootstrap keeps the declared backstops but never discovers new
+/// community tables and silently attaches a trigger to them.
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn migration_schema_bootstrap_keeps_only_explicit_legacy_fences() {
+    let pool = PgPool::connect(&crate::test_support::database_url())
+        .await
+        .unwrap();
+    sqlx::raw_sql("CREATE TABLE new_tenant_probe (community_id UUID NOT NULL)")
+        .execute(&pool)
+        .await
+        .unwrap();
+    // Only checked-in schema files contribute SQL to this test bootstrap.
+    sqlx::raw_sql(sqlx::AssertSqlSafe(
+        crate::test_support::desired_state_schema_sql(),
+    ))
+    .execute(&pool)
+    .await
+    .unwrap();
+
+    for (table, expected) in [
+        ("new_tenant_probe", false),
+        ("events", true),
+        ("users", true),
+    ] {
+        let fenced: bool = sqlx::query_scalar(
+            "SELECT EXISTS (SELECT 1 FROM pg_trigger \
+             WHERE tgrelid = $1::regclass \
+               AND tgfoid = 'enforce_community_write_fence()'::regprocedure)",
+        )
+        .bind(table)
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+        assert_eq!(fenced, expected, "unexpected fence on {table}");
+    }
+}

crates/buzz-db/src/store/channel_members.rs

@@ -556,13 +556,8 @@ impl LockedMemberSnapshot {
                 "member snapshot event id already exists".into(),
             ));
         }
-        crate::store::event_follow_up::after_admitted_insert(
-            &mut self.tx,
-            event.id.as_bytes().as_slice(),
-            kind,
-            Some(channel_id),
-        )
-        .await?;
+        crate::store::event_follow_up::after_admitted_insert(&mut self.tx, event, Some(channel_id))
+            .await?;
         crate::insert_mentions_in_transaction(&mut self.tx, event, Some(channel_id)).await?;
         Ok((
             buzz_core::StoredEvent::with_received_at(