diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 7ff26b4ef74..34f2afa2d5a 100644 --- a/ARCHITECTURE.md +++ b/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 diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index 4f91369d6df..9d7f396dbf5 100644 --- a/crates/buzz-admin/src/main.rs +++ b/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 { 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, diff --git a/crates/buzz-core/src/lib.rs b/crates/buzz-core/src/lib.rs index e7669c874b0..0c89cd62c0f 100644 --- a/crates/buzz-core/src/lib.rs +++ b/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. diff --git a/crates/buzz-core/src/media.rs b/crates/buzz-core/src/media.rs new file mode 100644 index 00000000000..1685ae5f566 --- /dev/null +++ b/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/` +/// 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 { + 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 = 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 { + 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/` path in `text`, where the hash is not +/// followed by another hex digit. +fn media_path_hashes(text: &str) -> impl Iterator + '_ { + 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()); + } +} diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index f495983a28d..7fa87929f45 100644 --- a/crates/buzz-db/src/lib.rs +++ b/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; diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index af877332529..12f89c0504c 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/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 diff --git a/crates/buzz-db/src/runtime/tests/schema_fence_bootstrap_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/schema_fence_bootstrap_postgres_tests.rs new file mode 100644 index 00000000000..67eb9a9a990 --- /dev/null +++ b/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}"); + } +} diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 28e64ff61c8..111016ede1a 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/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( diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index c5d62451278..c236d4158c4 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -90,6 +90,8 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[ "events", "git_repo_names", "join_policy_acceptances", + "media_shares", + "media_uploaders", "moderation_actions", "moderation_reports", "parameterized_event_watermarks", @@ -156,6 +158,8 @@ pub const PURGE_SCOPED_TABLES: &[&str] = &[ "audit_log", "community_bans", "community_nip_fi_issuers", + "media_shares", + "media_uploaders", "pubkey_allowlist", "relay_members", "users", @@ -4430,7 +4434,7 @@ mod postgres_tests { }) .await .expect("connect isolated migration DB"); - crate::migration::run_migrations_through(&db.pool, 66) + crate::migration::run_migrations_through(&db.pool, 68) .await .expect("migrate with fence objects present"); let store = db.deletion_store(); @@ -4458,6 +4462,8 @@ DROP TRIGGER IF EXISTS community_write_fence_event_mentions ON event_mentions; DROP TRIGGER IF EXISTS community_write_fence_events ON events; DROP TRIGGER IF EXISTS community_write_fence_git_repo_names ON git_repo_names; DROP TRIGGER IF EXISTS community_write_fence_join_policy_acceptances ON join_policy_acceptances; +DROP TRIGGER IF EXISTS community_write_fence_media_shares ON media_shares; +DROP TRIGGER IF EXISTS community_write_fence_media_uploaders ON media_uploaders; DROP TRIGGER IF EXISTS community_write_fence_moderation_actions ON moderation_actions; DROP TRIGGER IF EXISTS community_write_fence_moderation_reports ON moderation_reports; DROP TRIGGER IF EXISTS community_write_fence_parameterized_event_watermarks ON parameterized_event_watermarks; @@ -4501,14 +4507,13 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); store.validate_catalog().await.expect("deletion catalog"); let inventory = store.inventory_schema(community).await.expect("inventory"); assert_eq!(inventory.scoped_tables.len(), EXPECTED_SCOPED_TABLES.len()); - assert_eq!( - inventory.fenced_tables.len(), - if objects_present { - EXPECTED_SCOPED_TABLES.len() - } else { - 0 - } - ); + // Tables added after fence retirement are admitted in the app and + // carry no fence, so only legacy tables are fenced. + assert_eq!(inventory.fenced_tables.is_empty(), !objects_present); + assert!(inventory + .fenced_tables + .iter() + .all(|table| inventory.scoped_tables.contains(table))); let lease = store .acquire_serving_write_lease( community, @@ -8489,6 +8494,31 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); .expect("count relay admin rows") } + async fn seed_unfenced_media(db: &Db, community: Uuid) -> nostr::Event { + let community = CommunityId::from_uuid(community); + let keys = nostr::Keys::generate(); + let sha = "ab".repeat(32); + let store = db.deletion_store(); + let lease = store + .acquire_serving_write_lease( + community, + "media_upload", + "purge-test", + Duration::from_secs(60), + ) + .await + .unwrap(); + db.record_media_uploader(&lease, &sha, keys.public_key().as_bytes()) + .await + .unwrap(); + store.release_serving_write_lease(&lease).await.unwrap(); + let event = nostr::EventBuilder::new(nostr::Kind::TextNote, format!("/media/{sha}")) + .sign_with_keys(&keys) + .unwrap(); + db.insert_event(community, &event, None).await.unwrap(); + event + } + /// Report-backed staff actions FK `moderation_reports`, and outbox rows FK /// their action. Purge must delete the community's report-backed and direct /// rows in FK order and leave other communities' rows alone. @@ -8498,6 +8528,21 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); let (db, store) = store().await; let (request, inventory) = inventoried_request(&db, &store).await; let community = *request.community_id.as_uuid(); + let media = seed_unfenced_media(&db, community).await; + let backstops: i64 = sqlx::query_scalar( + "SELECT (SELECT count(*) FROM pg_constraint WHERE contype = 'f' \ + AND conrelid IN ('media_shares'::regclass, 'media_uploaders'::regclass)) \ + + (SELECT count(*) FROM pg_trigger \ + WHERE tgrelid IN ('media_shares'::regclass, 'media_uploaders'::regclass) \ + AND tgfoid = 'enforce_community_write_fence()'::regprocedure)", + ) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!( + backstops, 0, + "new media tables rely on application admission" + ); seed_relay_admin_rows(&db, community).await; let bystander = db .ensure_configured_community(&format!("bystander-{}.example", Uuid::new_v4().simple())) @@ -8507,6 +8552,7 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); .as_uuid() .to_owned(); seed_relay_admin_rows(&db, bystander).await; + seed_unfenced_media(&db, bystander).await; let live = store .inventory_schema(request.community_id) .await @@ -8524,6 +8570,10 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); .expect("claim") .expect("won claim"); store.begin_quiescing(&claim.lease).await.expect("quiesce"); + assert!(db + .insert_event(request.community_id, &media, None) + .await + .is_err()); let generation = store.fence(&claim.lease).await.expect("fence"); let token = LeaseToken { fence_generation: Some(generation), @@ -8540,6 +8590,18 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); .expect("bindings"); let deleted = store.purge_postgres(&token).await.expect("purge postgres"); + for table in ["media_shares", "media_uploaders"] { + assert_eq!(deleted.get(table), Some(&1)); + } + let media_rows: (i64, i64) = sqlx::query_as( + "SELECT (SELECT count(*) FROM media_shares WHERE community_id = $1), \ + (SELECT count(*) FROM media_uploaders WHERE community_id = $1)", + ) + .bind(bystander) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(media_rows, (1, 1), "another community's media must survive"); assert_eq!(deleted.get("relay_admin_outbox"), Some(&2)); assert_eq!(deleted.get("relay_admin_actions"), Some(&2)); assert_eq!(relay_admin_row_counts(&db, community).await, (0, 0)); diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index c279e4831d0..3aba787a523 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -515,13 +515,7 @@ async fn insert_event_on( let was_inserted = result.rows_affected() > 0; if was_inserted { - crate::store::event_follow_up::after_admitted_insert( - tx, - id_bytes.as_slice(), - kind_i32, - channel_id, - ) - .await?; + crate::store::event_follow_up::after_admitted_insert(tx, event, channel_id).await?; } Ok(( @@ -1905,13 +1899,7 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( let was_inserted = result.rows_affected() > 0; if was_inserted { - crate::store::event_follow_up::after_admitted_insert( - tx, - id_bytes.as_slice(), - kind_i32, - channel_id, - ) - .await?; + crate::store::event_follow_up::after_admitted_insert(tx, event, channel_id).await?; if let Some(ref meta) = thread_meta { let broadcast_val: bool = meta.broadcast; diff --git a/crates/buzz-db/src/store/event_follow_up.rs b/crates/buzz-db/src/store/event_follow_up.rs index d2853b4d89d..9079e496539 100644 --- a/crates/buzz-db/src/store/event_follow_up.rs +++ b/crates/buzz-db/src/store/event_follow_up.rs @@ -1,7 +1,10 @@ //! App-owned follow-up work for a newly inserted `events` row. //! -//! Two pieces of work follow every durable event insert: +//! Three pieces of work follow every durable event insert: //! +//! - **Media grants.** References grant access only after their author passes +//! the sharing check. Replacements record these before retiring the old head, +//! inside their savepoint; delivery is staged only after it succeeds. //! - **Push match enqueue.** Eligible event IDs are staged in memory. Commit //! submits them to a bounded best-effort worker using a dedicated connection. //! Push errors, delays, overload, and shutdown can drop wakes, never messages. @@ -37,18 +40,53 @@ const SQLSTATE_LOCK_NOT_AVAILABLE: &str = "55P03"; /// Follow-up for an event row inserted in an admitted transaction. /// /// Call only when the INSERT affected a row, after releasing any savepoint -/// that may roll it back. Stages push work for after commit and records TTL work. +/// that may roll it back. Stages push work for after commit, records TTL work, +/// and shares the blobs the event references into its channel in the same +/// transaction, so a stored reference always carries its share. pub(crate) async fn after_admitted_insert( tx: &mut AdmittedTx, - event_id: &[u8], - kind: i32, + event: &nostr::Event, channel_id: Option, ) -> Result<()> { - tx.record_push_event(event_id, kind); - tx.record_channel_event(channel_id, kind); + record_admitted_media(tx, event, channel_id).await?; + stage_admitted_insert(tx, event, channel_id); + Ok(()) +} + +/// Record media inside a replacement's savepoint, before retiring its old head. +/// This preserves authority carried by a reference the new head keeps. +pub(crate) async fn record_admitted_media( + tx: &mut AdmittedTx, + event: &nostr::Event, + channel_id: Option, +) -> Result<()> { + let hashes = buzz_core::media::shared_blob_hashes(event); + if !hashes.is_empty() { + let community = tx.community(); + crate::store::media_reference::record_media_shares( + tx.conn(), + community, + &hashes, + event, + channel_id, + ) + .await?; + } Ok(()) } +/// Stage delivery only after the insertion's savepoint has succeeded. +/// Call after [`record_admitted_media`] when splitting the two phases. +pub(crate) fn stage_admitted_insert( + tx: &mut AdmittedTx, + event: &nostr::Event, + channel_id: Option, +) { + let kind = i32::from(event.kind.as_u16()); + tx.record_push_event(event.id.as_bytes(), kind); + tx.record_channel_event(channel_id, kind); +} + /// The channel whose TTL an event of `kind` in `channel_id` refreshes, if any. pub(crate) fn ttl_refresh_channel(channel_id: Option, kind: i32) -> Option { channel_id.filter(|_| kind != KIND_CHANNEL_CREATE) diff --git a/crates/buzz-db/src/store/media_cleanup.rs b/crates/buzz-db/src/store/media_cleanup.rs new file mode 100644 index 00000000000..a07c51b52b4 --- /dev/null +++ b/crates/buzz-db/src/store/media_cleanup.rs @@ -0,0 +1,120 @@ +//! Bounded, tenant-admitted reclamation of grants whose sources are gone. + +use std::collections::BTreeMap; + +use buzz_datastore_tracing::datastore_span; +use sqlx::FromRow; +use uuid::Uuid; + +use crate::{Db, Result}; + +/// Volatile primary-key position for the periodic media-share reaper. +/// Restarting a scan is safe: cleanup never creates an authorization grant. +#[derive(Clone, Debug, FromRow)] +pub struct MediaShareCleanupCursor { + community_id: Uuid, + sha256: String, + event_id: Vec, +} + +/// Committed cleanup and the next scan position; `None` starts another cycle. +pub struct MediaShareCleanupBatch { + /// Number of inactive grant rows removed. + pub removed: u64, + /// Continue after this key, or wrap to the beginning on the next tick. + pub next_cursor: Option, +} + +impl Db { + /// Inspect at most `batch_size` grants in primary-key order and remove + /// inactive ones under each community's write admission. + /// + /// Keep the first channel grant per blob, even if inactive: otherwise + /// removing its last historical channel row would restore uploader preview + /// access. This bounds dead channel history to one row per blob; public + /// profile/emoji grants need no such marker. Live sources are never removed. + /// Tenant-local errors retry in a later scan cycle, as do skipped locks. + #[datastore_span(name = "reap_inactive_media_shares", system = "postgresql")] + pub async fn reap_inactive_media_shares( + &self, + after: Option<&MediaShareCleanupCursor>, + batch_size: u16, + ) -> Result { + if !(1..=1000).contains(&batch_size) { + return Err(crate::DbError::InvalidData( + "media cleanup batch size must be between 1 and 1000".into(), + )); + } + let mut conn = crate::observability::acquire_writer( + &self.pool, + crate::observability::WriterOperation::Maintenance, + ) + .await?; + // Intentional fleet-wide candidate discovery, bounded by the leading + // primary key. It reads only keys; every mutation is admitted below. + let rows = sqlx::query_as::<_, MediaShareCleanupCursor>( + "SELECT community_id, sha256, event_id FROM media_shares \ + WHERE (community_id, sha256, event_id) > ($1, $2, $3) \ + ORDER BY community_id, sha256, event_id LIMIT $4", + ) + .bind(after.map_or(Uuid::nil(), |key| key.community_id)) + .bind(after.map_or("", |key| key.sha256.as_str())) + .bind(after.map_or(&[][..], |key| key.event_id.as_slice())) + .bind(i64::from(batch_size)) + .fetch_all(&mut *conn) + .await?; + drop(conn); + let next_cursor = (rows.len() == usize::from(batch_size)) + .then(|| rows.last().cloned()) + .flatten(); + let mut groups = BTreeMap::, Vec>)>::new(); + for row in rows { + let (hashes, ids) = groups.entry(row.community_id).or_default(); + hashes.push(row.sha256); + ids.push(row.event_id); + } + let groups = &groups; + let removed = crate::sweep_admitted_communities( + &self.pool, + groups.keys().copied(), + crate::observability::WriterOperation::Maintenance, + "media_share_reap", + |community, mut tx| async move { + sqlx::raw_sql("SET LOCAL lock_timeout = '1s'; SET LOCAL statement_timeout = '5s'") + .execute(tx.conn()) + .await?; + let (hashes, ids) = &groups[&community]; + let removed = sqlx::query(concat!( + "DELETE FROM media_shares s \ + USING unnest($2::text[], $3::bytea[]) AS batch(sha256, event_id) \ + WHERE s.community_id = $1 AND s.sha256 = batch.sha256 \ + AND s.event_id = batch.event_id \ + AND NOT EXISTS (SELECT 1 FROM events e \ + WHERE e.community_id = s.community_id AND e.id = s.event_id \ + AND e.created_at = s.event_created_at AND ", + crate::store::media_reference::live_source_sql!(), + ") AND (s.channel_id IS NULL OR EXISTS ( \ + SELECT 1 FROM media_shares previous \ + WHERE previous.community_id = s.community_id \ + AND previous.sha256 = s.sha256 AND previous.channel_id IS NOT NULL \ + AND previous.event_id < s.event_id))", + )) + .bind(community) + .bind(hashes) + .bind(ids) + .execute(tx.conn()) + .await? + .rows_affected(); + tx.commit().await?; + Ok(removed) + }, + ) + .await + .into_iter() + .sum(); + Ok(MediaShareCleanupBatch { + removed, + next_cursor, + }) + } +} diff --git a/crates/buzz-db/src/store/media_grant.rs b/crates/buzz-db/src/store/media_grant.rs new file mode 100644 index 00000000000..b1aa93977a2 --- /dev/null +++ b/crates/buzz-db/src/store/media_grant.rs @@ -0,0 +1,143 @@ +//! Provenance shared by live media recording, legacy replacement, and backfill. + +use buzz_core::{CommunityId, StoredEvent}; +use chrono::{DateTime, Utc}; +use nostr::Event; +use sqlx::PgConnection; +use uuid::Uuid; + +use crate::{AdmittedTx, Result}; + +/// Resolve the first valid edit target, as the relay's edit validator does. +/// Nested edits and missing/deleted/cross-channel targets grant no file access. +pub(crate) async fn edit_target( + conn: &mut PgConnection, + community: CommunityId, + event: &Event, + channel: Option, +) -> Result, DateTime)>> { + if event.kind.as_u16() != buzz_core::kind::KIND_STREAM_MESSAGE_EDIT as u16 { + return Ok(None); + } + let Some(channel) = channel else { + return Ok(None); + }; + let Some(target) = event.tags.iter().find_map(|tag| { + let parts = tag.as_slice(); + (parts.first().is_some_and(|name| name == "e")) + .then(|| { + parts + .get(1) + .and_then(|id| nostr::EventId::from_hex(id).ok()) + }) + .flatten() + }) else { + return Ok(None); + }; + Ok(sqlx::query_as( + "SELECT id, created_at FROM events \ + WHERE community_id = $1 AND id = $2 AND channel_id = $3 \ + AND deleted_at IS NULL AND kind <> 40003 \ + ORDER BY created_at DESC LIMIT 1", + ) + .bind(community.as_uuid()) + .bind(target.as_bytes().as_slice()) + .bind(channel) + .fetch_optional(conn) + .await?) +} + +/// Record only server-received pre-cutoff references, with their live source +/// and (for edits) target. The receipt check is repeated in the INSERT. +pub(crate) async fn record_history( + conn: &mut PgConnection, + community: CommunityId, + events: &[StoredEvent], +) -> Result { + let mut hashes = Vec::new(); + let mut channels = Vec::new(); + let mut ids = Vec::new(); + let mut times = Vec::new(); + let mut targets = Vec::new(); + let mut target_times = Vec::new(); + for stored in events { + let event = &stored.event; + let blobs = buzz_core::media::shared_blob_hashes(event); + if blobs.is_empty() { + continue; + } + let target = edit_target(conn, community, event, stored.channel_id).await?; + if event.kind.as_u16() == 40003 && target.is_none() { + continue; + } + let seconds = event.created_at.as_secs() as i64; + let created_at = DateTime::from_timestamp(seconds, 0) + .ok_or(crate::DbError::InvalidTimestamp(seconds))?; + for hash in blobs { + hashes.push(hash); + channels.push(stored.channel_id); + ids.push(event.id.to_bytes().to_vec()); + times.push(created_at); + targets.push(target.as_ref().map(|(id, _)| id.clone())); + target_times.push(target.as_ref().map(|(_, at)| *at)); + } + } + if hashes.is_empty() { + return Ok(0); + } + Ok(sqlx::query( + "INSERT INTO media_shares \ + (community_id, sha256, channel_id, event_id, event_created_at, \ + target_event_id, target_event_created_at) \ + SELECT $1, t.h, t.c, e.id, e.created_at, t.target_id, t.target_at \ + FROM unnest($2::text[], $3::uuid[], $4::bytea[], $5::timestamptz[], \ + $6::bytea[], $7::timestamptz[]) \ + AS t(h, c, event_id, event_created_at, target_id, target_at) \ + JOIN events e ON e.community_id = $1 AND e.id = t.event_id \ + AND e.created_at = t.event_created_at AND e.deleted_at IS NULL \ + JOIN communities c ON c.id = e.community_id \ + AND e.received_at < c.media_shares_recording_started_at \ + ORDER BY t.h, e.id ON CONFLICT DO NOTHING", + ) + .bind(community.as_uuid()) + .bind(hashes) + .bind(channels) + .bind(ids) + .bind(times) + .bind(targets) + .bind(target_times) + .execute(conn) + .await? + .rows_affected()) +} + +/// Recover an eligible legacy head inside its replacement transaction, while +/// it still grants access. Only references the replacement keeps survive it. +pub(crate) async fn bootstrap_replacement( + tx: &mut AdmittedTx, + incoming: &Event, + previous_id: &[u8], +) -> Result<()> { + if buzz_core::media::shared_blob_hashes(incoming).is_empty() { + return Ok(()); + } + let community = tx.community(); + let row = sqlx::query( + "SELECT e.id, e.pubkey, e.created_at, e.kind, e.tags, e.content, e.sig, \ + e.received_at, e.channel_id \ + FROM events e JOIN communities c ON c.id = e.community_id \ + WHERE e.community_id = $1 AND e.id = $2 AND e.deleted_at IS NULL \ + AND e.received_at < c.media_shares_recording_started_at \ + ORDER BY e.created_at DESC LIMIT 1", + ) + .bind(community.as_uuid()) + .bind(previous_id) + .fetch_optional(tx.conn()) + .await?; + if let Some(row) = row { + if let Some(stored) = crate::event::row_to_stored_event(row)? { + record_history(tx.conn(), community, &[stored]).await?; + } + } + Ok(()) +} diff --git a/crates/buzz-db/src/store/media_reference.rs b/crates/buzz-db/src/store/media_reference.rs new file mode 100644 index 00000000000..3bd57f746a2 --- /dev/null +++ b/crates/buzz-db/src/store/media_reference.rs @@ -0,0 +1,960 @@ +//! Where each media blob is shared, recorded with the events that reference it. + +use buzz_datastore_tracing::datastore_span; +use chrono::{DateTime, Utc}; +use sqlx::{PgConnection, Row}; +use uuid::Uuid; + +use crate::{error::Result, Db}; +use buzz_core::CommunityId; + +/// Both the reference and its edited message must survive. The target lookup +/// uses its recorded timestamp to reach the original event's partition. +macro_rules! live_source_sql { + () => { + "e.deleted_at IS NULL AND (e.kind <> 40003 OR EXISTS ( \ + SELECT 1 FROM events target \ + WHERE target.community_id = s.community_id AND target.id = s.target_event_id \ + AND target.created_at = s.target_event_created_at AND target.deleted_at IS NULL \ + AND target.kind <> 40003 AND target.channel_id IS NOT DISTINCT FROM e.channel_id))" + }; +} +pub(crate) use live_source_sql; + +/// SQL predicate: whether the reader `$3` may read the blob `b.sha256` in community `$1`. +/// +/// A share admits everyone when it is community-wide, or readers of its +/// channel (open, or one they actively belong to). An unknown or unshared hash +/// grants nothing. Deleted or purged source events grant nothing. +macro_rules! readable_sql { + () => { + concat!( + "( \ + EXISTS (SELECT 1 FROM media_shares s \ + JOIN events e ON e.community_id = s.community_id AND e.id = s.event_id \ + AND e.created_at = s.event_created_at AND ", + crate::store::media_reference::live_source_sql!(), + " \ + WHERE s.community_id = $1 AND s.sha256 = b.sha256 AND s.channel_id IS NULL) \ + OR EXISTS (SELECT 1 FROM media_shares s \ + JOIN events e ON e.community_id = s.community_id AND e.id = s.event_id \ + AND e.created_at = s.event_created_at AND ", + crate::store::media_reference::live_source_sql!(), + " \ + JOIN channels c ON c.community_id = $1 AND c.id = s.channel_id \ + WHERE s.community_id = $1 AND s.sha256 = b.sha256 AND c.deleted_at IS NULL \ + AND (c.visibility = 'open' \ + OR EXISTS (SELECT 1 FROM channel_members cm \ + WHERE cm.community_id = $1 AND cm.channel_id = c.id \ + AND cm.pubkey = $3 AND cm.removed_at IS NULL))))" + ) + }; +} + +/// Share into `channel_id` (or with the whole community when `None`) each blob +/// in `sha256s` that the event's author can read or has uploaded (uploading +/// the exact bytes proves possession). An agent's profile picture may also +/// use its registered owner's upload proof. Any other blob is left unshared, +/// so a reference built from a leaked hash shows it to no one new. Called with the +/// transaction that stores the referencing event; returns the shares added. +pub(crate) async fn record_media_shares( + conn: &mut PgConnection, + community_id: CommunityId, + sha256s: &[String], + event: &nostr::Event, + channel_id: Option, +) -> Result { + let created_at_secs = event.created_at.as_secs() as i64; + let created_at = DateTime::from_timestamp(created_at_secs, 0) + .ok_or(crate::DbError::InvalidTimestamp(created_at_secs))?; + let target = super::media_grant::edit_target(conn, community_id, event, channel_id).await?; + if event.kind.as_u16() == 40003 && target.is_none() { + return Ok(0); + } + let added = sqlx::query(concat!( + "INSERT INTO media_shares (community_id, sha256, channel_id, event_id, event_created_at, \ + target_event_id, target_event_created_at) \ + SELECT $1, b.sha256, $4, $6, $7, $8, $9 FROM (SELECT DISTINCT unnest($2::text[]) AS sha256) b \ + WHERE (EXISTS (SELECT 1 FROM media_uploaders u \ + WHERE u.community_id = $1 AND u.sha256 = b.sha256 AND u.pubkey = $3) \ + OR (b.sha256 = $5 AND EXISTS ( \ + SELECT 1 FROM users agent JOIN media_uploaders u \ + ON u.community_id = agent.community_id AND u.pubkey = agent.agent_owner_pubkey \ + WHERE agent.community_id = $1 AND agent.pubkey = $3 AND u.sha256 = b.sha256)) \ + OR ", + readable_sql!(), + ") AND EXISTS (SELECT 1 FROM events e \ + WHERE e.community_id = $1 AND e.id = $6 AND e.created_at = $7 \ + AND e.deleted_at IS NULL) \ + ORDER BY b.sha256 ON CONFLICT DO NOTHING" + )) + .bind(community_id.as_uuid()) + .bind(sha256s) + .bind(event.pubkey.as_bytes().as_slice()) + .bind(channel_id) + .bind(buzz_core::media::profile_picture_blob_hash(event)) + .bind(event.id.as_bytes().as_slice()) + .bind(created_at) + .bind(target.as_ref().map(|(id, _)| id)) + .bind(target.as_ref().map(|(_, at)| *at)) + .execute(conn) + .await? + .rows_affected(); + Ok(added) +} + +impl Db { + /// Record the shares of undeleted events stored in an active community, + /// reading `batch_size` events at a time; returns + /// the number of shares added. + /// + /// Retained server upload audits restore uploader proof before references + /// are evaluated. Missing, malformed, or unattributed audit rows establish + /// no ownership; those uploads still need the exact bytes uploaded again. + /// + /// Shares come from the same rule as live writes. Events received before + /// share recording was installed predate the readability check, so their + /// references are recorded as they stand; later events get the live, + /// checked recording using the author's current permissions. Earlier refusals + /// are not remembered: any run can grant old references after access changes. + /// Rows are deduplicated, but authorization decisions are not preserved. + #[datastore_span(name = "backfill_media_shares", system = "postgresql")] + pub async fn backfill_media_shares(&self, batch_size: i64) -> Result { + if batch_size <= 0 { + return Err(crate::DbError::InvalidData( + "batch size must be positive".into(), + )); + } + let communities: Vec<(Uuid, DateTime)> = sqlx::query_as( + "SELECT id, media_shares_recording_started_at FROM communities \ + WHERE deleted_at IS NULL AND deletion_state = 'active' ORDER BY id", + ) + .fetch_all(&self.pool) + .await?; + let mut added = 0; + for (community, installed) in communities { + let community = CommunityId::from_uuid(community); + self.backfill_media_uploaders(community, batch_size).await?; + let mut cursor: Option<(DateTime, Vec)> = None; + loop { + let rows = sqlx::query( + "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, \ + channel_id \ + FROM events \ + WHERE community_id = $1 AND deleted_at IS NULL \ + AND ($2::timestamptz IS NULL OR (created_at, id) > ($2, $3)) \ + ORDER BY created_at, id LIMIT $4", + ) + .bind(community.as_uuid()) + .bind(cursor.as_ref().map(|(at, _)| *at)) + .bind( + cursor + .as_ref() + .map(|(_, id)| id.clone()) + .unwrap_or_default(), + ) + .bind(batch_size) + .fetch_all(&self.pool) + .await?; + let Some(last) = rows.last() else { break }; + cursor = Some((last.try_get("created_at")?, last.try_get("id")?)); + let full_batch = i64::try_from(rows.len()).unwrap_or(i64::MAX) == batch_size; + + let mut history = Vec::new(); + let mut recorded_live = Vec::new(); + for row in rows { + let received_at: DateTime = row.try_get("received_at")?; + let Some(stored) = crate::event::row_to_stored_event(row)? else { + continue; + }; + let hashes = buzz_core::media::shared_blob_hashes(&stored.event); + if hashes.is_empty() { + continue; + } + if received_at < installed { + history.push(stored); + } else { + recorded_live.push((stored.event, hashes, stored.channel_id)); + } + } + if !history.is_empty() { + let mut tx = self.begin_media_share_transaction(community).await?; + added += + super::media_grant::record_history(tx.conn(), community, &history).await?; + tx.commit().await?; + } + for (event, hashes, channel) in recorded_live { + let mut tx = self.begin_media_share_transaction(community).await?; + added += + record_media_shares(tx.conn(), community, &hashes, &event, channel).await?; + tx.commit().await?; + } + if !full_batch { + break; + } + } + } + Ok(added) + } + + async fn begin_media_share_transaction( + &self, + community: CommunityId, + ) -> Result { + crate::begin_community_write_transaction( + &self.pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await + } +} + +#[cfg(test)] +mod postgres_tests { + use std::time::Duration; + + use super::*; + use crate::channel::{ChannelType, ChannelVisibility}; + use nostr::{EventBuilder, Keys, Kind, Tag}; + use sqlx::PgPool; + + async fn fixture() -> (Db, CommunityId) { + let db = Db::from_pool( + PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to test DB"), + ); + let host = format!("media-shares-{}.example", Uuid::new_v4().simple()); + let community = db + .ensure_configured_community(&host) + .await + .expect("seed community") + .id; + (db, community) + } + + async fn channel( + db: &Db, + community: CommunityId, + owner: &Keys, + visibility: ChannelVisibility, + ) -> Uuid { + db.create_channel( + community, + &format!("media-{}", Uuid::new_v4().simple()), + ChannelType::Stream, + visibility, + None, + &owner.public_key().to_bytes(), + None, + ) + .await + .expect("create channel") + .id + } + + fn attachment(keys: &Keys, sha256: &str, channel: Option) -> nostr::Event { + let mut tags = vec![Tag::parse(["imeta", &format!("url /media/{sha256}.jpg")]).unwrap()]; + if let Some(channel) = channel { + tags.push(Tag::parse(["h", &channel.to_string()]).unwrap()); + } + EventBuilder::new(Kind::from(9), format!("attachment {}", Uuid::new_v4())) + .tags(tags) + .sign_with_keys(keys) + .unwrap() + } + + async fn shares(db: &Db, community: CommunityId, sha256: &str) -> Vec> { + let mut shares: Vec> = sqlx::query_scalar( + "SELECT DISTINCT channel_id FROM media_shares WHERE community_id = $1 AND sha256 = $2", + ) + .bind(community.as_uuid()) + .bind(sha256) + .fetch_all(&db.pool) + .await + .expect("read shares"); + shares.sort(); + shares + } + + fn unique_sha() -> String { + format!("{:064x}", Uuid::new_v4().as_u128()) + } + + async fn upload(db: &Db, community: CommunityId, uploader: &Keys, sha256: &str) { + let lease = db + .deletion_store() + .acquire_serving_write_lease( + community, + "media_upload", + "media-shares-test", + Duration::from_secs(60), + ) + .await + .expect("acquire lease"); + db.record_media_uploader(&lease, sha256, &uploader.public_key().to_bytes()) + .await + .expect("record uploader"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn stored_references_share_only_blobs_their_author_can_read() { + let (db, community) = fixture().await; + let (alice, bob) = (Keys::generate(), Keys::generate()); + let sha = unique_sha(); + let private = channel(&db, community, &alice, ChannelVisibility::Private).await; + let bobs = channel(&db, community, &bob, ChannelVisibility::Open).await; + + for uploader_present in [false, true] { + if uploader_present { + upload(&db, community, &alice, &sha).await; + } + db.insert_event(community, &attachment(&bob, &sha, None), None) + .await + .expect("bob references an unshared hash"); + assert!( + shares(&db, community, &sha).await.is_empty(), + "a hash alone grants nothing, even before its uploader posts it" + ); + } + db.insert_event( + community, + &attachment(&alice, &sha, Some(private)), + Some(private), + ) + .await + .expect("alice posts privately"); + assert_eq!(shares(&db, community, &sha).await, vec![Some(private)]); + + db.insert_event(community, &attachment(&bob, &sha, None), None) + .await + .expect("bob references the hash"); + assert_eq!( + shares(&db, community, &sha).await, + vec![Some(private)], + "a hash alone must not share a blob its author cannot read" + ); + + upload(&db, community, &bob, &sha).await; + db.insert_event(community, &attachment(&bob, &sha, Some(bobs)), Some(bobs)) + .await + .expect("bob posts his copy"); + let mut expected = vec![Some(private), Some(bobs)]; + expected.sort(); + assert_eq!( + shares(&db, community, &sha).await, + expected, + "an uploader may share their own bytes" + ); + + let excluded = unique_sha(); + upload(&db, community, &alice, &excluded).await; + use buzz_core::kind::{ + AUTHOR_ONLY_KINDS, KIND_AGENT_ENGRAM, KIND_DELETION, KIND_EVENT_REMINDER, + P_GATED_KINDS, SHARED_GATED_KINDS, + }; + let kinds = AUTHOR_ONLY_KINDS + .iter() + .chain(P_GATED_KINDS) + .chain(SHARED_GATED_KINDS) + .copied() + .chain([KIND_AGENT_ENGRAM, KIND_DELETION]); + for kind in kinds { + if (20_000..30_000).contains(&kind) { + continue; // Ephemeral frames have no stored reference to backfill. + } + let mut tags = vec![Tag::parse(["d", "private-media"]).unwrap()]; + if kind == KIND_DELETION { + tags.push(Tag::parse(["k", &KIND_EVENT_REMINDER.to_string()]).unwrap()); + } + let event = EventBuilder::new(Kind::from(kind as u16), format!("/media/{excluded}")) + .tags(tags) + .sign_with_keys(&alice) + .unwrap(); + db.insert_event(community, &event, None).await.unwrap(); + assert!( + db.get_event_by_id(community, event.id.as_bytes()) + .await + .unwrap() + .is_some(), + "kind {kind} must exercise a stored event" + ); + assert!( + shares(&db, community, &excluded).await.is_empty(), + "private kind {kind} must create no grant even for its uploader" + ); + } + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn backfill_trusts_history_and_rechecks_recorded_events() { + check_backfill().await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn deleting_a_reference_revokes_only_its_own_media_grant() { + let (db, community) = fixture().await; + let (owner, reader) = (Keys::generate(), Keys::generate()); + let open = channel(&db, community, &owner, ChannelVisibility::Open).await; + for channel_id in [Some(open), None] { + let sha = unique_sha(); + upload(&db, community, &owner, &sha).await; + let first = attachment(&owner, &sha, channel_id); + let last = attachment(&owner, &sha, channel_id); + for event in [&first, &last] { + db.insert_event(community, event, channel_id).await.unwrap(); + } + let sources: Vec> = sqlx::query_scalar( + "SELECT event_id FROM media_shares WHERE community_id = $1 AND sha256 = $2", + ) + .bind(community.as_uuid()) + .bind(&sha) + .fetch_all(&db.pool) + .await + .unwrap(); + assert_eq!(sources.len(), 2, "each message must retain its own grant"); + for event in [&first, &last] { + assert!(sources.contains(&event.id.to_bytes().to_vec())); + } + let edit_only = if let Some(channel) = channel_id { + let edit_sha = unique_sha(); + upload(&db, community, &owner, &edit_sha).await; + for target in [&first, &last] { + let edit = EventBuilder::new( + Kind::from(40003), + format!("retained /media/{sha}, added /media/{edit_sha}"), + ) + .tags([ + Tag::parse(["h", &channel.to_string()]).unwrap(), + Tag::parse(["e", &target.id.to_hex()]).unwrap(), + ]) + .sign_with_keys(&owner) + .unwrap(); + db.insert_event(community, &edit, channel_id).await.unwrap(); + // The backfill must also reject surviving legacy edits + // once their logical message has disappeared. + let mut tx = db.begin_media_share_transaction(community).await.unwrap(); + sqlx::query( + "UPDATE events SET received_at = (SELECT media_shares_recording_started_at \ + FROM communities WHERE id = $1) - interval '1 second' \ + WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(edit.id.as_bytes().as_slice()) + .execute(tx.conn()) + .await + .unwrap(); + tx.commit().await.unwrap(); + } + assert!(readable_share(&db, community, &edit_sha, &reader).await); + Some(edit_sha) + } else { + None + }; + assert!(readable_share(&db, community, &sha, &reader).await); + for (event, still_shared) in [(&first, true), (&last, false)] { + assert!( + db.soft_delete_event_and_update_thread( + community, + event.id.as_bytes(), + None, + None, + ) + .await + .unwrap() + ); + assert_eq!( + readable_share(&db, community, &sha, &reader).await, + still_shared, + "only an undeleted source message can grant access" + ); + if let Some(edit_sha) = &edit_only { + assert_eq!( + readable_share(&db, community, edit_sha, &reader).await, + still_shared, + "edit-only attachments must share the target message's lifetime" + ); + } + } + db.insert_event(community, &attachment(&reader, &sha, None), None) + .await + .unwrap(); + assert!(!readable_share(&db, community, &sha, &reader).await); + db.backfill_media_shares(1).await.unwrap(); + assert!(!readable_share(&db, community, &sha, &reader).await); + if let Some(edit_sha) = &edit_only { + assert!(!readable_share(&db, community, edit_sha, &reader).await); + } + + let purged = attachment(&owner, &sha, channel_id); + db.insert_event(community, &purged, channel_id) + .await + .unwrap(); + assert!(readable_share(&db, community, &sha, &reader).await); + let mut tx = db.begin_media_share_transaction(community).await.unwrap(); + sqlx::query("DELETE FROM events WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(purged.id.as_bytes().as_slice()) + .execute(tx.conn()) + .await + .unwrap(); + tx.commit().await.unwrap(); + assert!( + !readable_share(&db, community, &sha, &reader).await, + "a purged source cannot leave a dangling grant" + ); + assert!(reap_all(&db).await > 0); + assert_eq!( + shares(&db, community, &sha).await.len(), + usize::from(channel_id.is_some()), + "cleanup retains at most one channel boundary, never public history" + ); + if let Some(edit_sha) = &edit_only { + assert_eq!( + shares(&db, community, edit_sha).await.len(), + 1, + "inactive edit grants compact to one channel marker" + ); + } + assert!(!readable_share(&db, community, &sha, &reader).await); + assert_eq!(reap_all(&db).await, 0, "cleanup is idempotent"); + } + + // Fleet cleanup skips a quiescing community while reclaiming another + // community's dead copy of the same hash and keeping its live copy. + let sha = unique_sha(); + upload(&db, community, &owner, &sha).await; + let retired = attachment(&owner, &sha, None); + db.insert_event(community, &retired, None).await.unwrap(); + db.soft_delete_event_and_update_thread(community, retired.id.as_bytes(), None, None) + .await + .unwrap(); + quiesce(&db, community).await; + let (other, other_community) = fixture().await; + upload(&other, other_community, &owner, &sha).await; + let old = attachment(&owner, &sha, None); + let current = attachment(&owner, &sha, None); + for event in [&old, ¤t] { + other + .insert_event(other_community, event, None) + .await + .unwrap(); + } + other + .soft_delete_event_and_update_thread(other_community, old.id.as_bytes(), None, None) + .await + .unwrap(); + assert!(reap_all(&db).await > 0); + assert_eq!( + shares(&db, community, &sha).await.len(), + 1, + "cleanup may not write into a quiescing community" + ); + assert_eq!(shares(&other, other_community, &sha).await.len(), 1); + assert!(readable_share(&other, other_community, &sha, &reader).await); + } + + async fn readable_share(db: &Db, community: CommunityId, sha: &str, reader: &Keys) -> bool { + sqlx::query_scalar(concat!( + "SELECT ", + readable_sql!(), + " FROM (SELECT $2::text AS sha256) b" + )) + .bind(community.as_uuid()) + .bind(sha) + .bind(reader.public_key().to_bytes().to_vec()) + .fetch_one(&db.pool) + .await + .unwrap() + } + + async fn quiesce(db: &Db, community: CommunityId) { + let mut lifecycle = db.pool.begin().await.unwrap(); + sqlx::query( + "SELECT set_config('buzz.deletion_executor_community', $1, true), \ + set_config('buzz.deletion_fence_generation', '0', true)", + ) + .bind(community.as_uuid().to_string()) + .execute(&mut *lifecycle) + .await + .unwrap(); + sqlx::query("UPDATE communities SET deletion_state = 'quiescing' WHERE id = $1") + .bind(community.as_uuid()) + .execute(&mut *lifecycle) + .await + .unwrap(); + lifecycle.commit().await.unwrap(); + } + + async fn reap_all(db: &Db) -> u64 { + let mut cursor = None; + let mut removed = 0; + // One row per page exercises progress across live rows and retained + // channel markers as well as rows deleted by a previous page. + for _ in 0..10_000 { + let batch = db + .reap_inactive_media_shares(cursor.as_ref(), 1) + .await + .unwrap(); + removed += batch.removed; + cursor = batch.next_cursor; + if cursor.is_none() { + return removed; + } + } + panic!("cleanup must finish its bounded primary-key scan"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn replacement_preserves_only_the_media_it_continues_to_reference() { + let (db, community) = fixture().await; + let viewer = Keys::generate(); + for kind in [Kind::Metadata, Kind::from(30030)] { + let (uploader, publisher) = (Keys::generate(), Keys::generate()); + let sha = unique_sha(); + upload(&db, community, &uploader, &sha).await; + let base = nostr::Timestamp::now().as_secs(); + let make_for = |sha: &str, keys: &Keys, kind: Kind, step: u64, picture: bool| { + let url = if picture { + format!("/media/{sha}.jpg") + } else { + String::new() + }; + let mut tags = Vec::new(); + let content = if kind == Kind::Metadata { + serde_json::json!({"picture": url, "name": format!("profile-{step}")}) + .to_string() + } else { + tags.push(Tag::parse(["d", "media-continuity"]).unwrap()); + if picture { + tags.push(Tag::parse(["emoji", "sample", &url]).unwrap()); + } + String::new() + }; + EventBuilder::new(kind, content) + .tags(tags) + .custom_created_at(nostr::Timestamp::from_secs(base + step)) + .sign_with_keys(keys) + .unwrap() + }; + let make = |keys: &Keys, kind: Kind, step: u64, picture: bool| { + make_for(&sha, keys, kind, step, picture) + }; + let publish = async |event: &nostr::Event| { + if event.kind == Kind::Metadata { + db.replace_addressable_event(community, event, None) + .await + .unwrap() + .1 + } else { + db.replace_parameterized_event(community, event, "media-continuity", None) + .await + .unwrap() + .1 + } + }; + assert!(publish(&make(&uploader, Kind::Metadata, 0, true)).await); + assert!(publish(&make(&publisher, kind, 0, true)).await); + assert!(publish(&make(&uploader, Kind::Metadata, 1, false)).await); + assert!(readable_share(&db, community, &sha, &viewer).await); + assert!(publish(&make(&publisher, kind, 1, true)).await); + assert!(reap_all(&db).await > 0); + assert_eq!( + shares(&db, community, &sha).await.len(), + 1, + "cleanup keeps the surviving profile/emoji head" + ); + assert!( + readable_share(&db, community, &sha, &viewer).await, + "a replacement must carry forward an authorized reference" + ); + assert!(publish(&make(&publisher, kind, 2, false)).await); + assert!(!readable_share(&db, community, &sha, &viewer).await); + assert!(publish(&make(&publisher, kind, 3, true)).await); + assert!( + !readable_share(&db, community, &sha, &viewer).await, + "a later replacement cannot revive a previously removed grant" + ); + + // An unbackfilled legacy head has neither proof nor shares. A + // normal name-only update must retain the eligible old reference. + let legacy_keys = Keys::generate(); + let legacy_sha = unique_sha(); + let legacy_make = + |step, picture| make_for(&legacy_sha, &legacy_keys, kind, step, picture); + let legacy = legacy_make(0, true); + assert!(publish(&legacy).await); + assert!(!readable_share(&db, community, &legacy_sha, &viewer).await); + let mut tx = db.begin_media_share_transaction(community).await.unwrap(); + sqlx::query( + "UPDATE events SET received_at = (SELECT media_shares_recording_started_at \ + FROM communities WHERE id = $1) - interval '1 second' \ + WHERE community_id = $1 AND id = $2", + ) + .bind(community.as_uuid()) + .bind(legacy.id.as_bytes().as_slice()) + .execute(tx.conn()) + .await + .unwrap(); + tx.commit().await.unwrap(); + assert!(publish(&legacy_make(1, true)).await); + assert!(reap_all(&db).await > 0); + assert_eq!(shares(&db, community, &legacy_sha).await.len(), 1); + db.backfill_media_shares(1).await.unwrap(); + assert!( + readable_share(&db, community, &legacy_sha, &viewer).await, + "a retained legacy avatar/emoji must survive replacement before backfill" + ); + assert!(publish(&legacy_make(2, false)).await); + assert!(!readable_share(&db, community, &legacy_sha, &viewer).await); + assert!(publish(&legacy_make(3, true)).await); + assert!(!readable_share(&db, community, &legacy_sha, &viewer).await); + assert!(reap_all(&db).await > 0); + assert!( + shares(&db, community, &legacy_sha).await.is_empty(), + "replaced avatar/emoji heads must not accumulate inactive rows" + ); + assert!(shares(&db, community, &sha).await.is_empty()); + } + } + + async fn check_backfill() { + let (db, community) = fixture().await; + let (alice, bob, carol) = (Keys::generate(), Keys::generate(), Keys::generate()); + let (sha, historical, deleted_history) = (unique_sha(), unique_sha(), unique_sha()); + let private = channel(&db, community, &alice, ChannelVisibility::Private).await; + upload(&db, community, &alice, &sha).await; + db.insert_event( + community, + &attachment(&alice, &sha, Some(private)), + Some(private), + ) + .await + .expect("alice posts privately"); + db.insert_event(community, &attachment(&bob, &sha, None), None) + .await + .expect("bob references the hash"); + let old = attachment(&carol, &historical, None); + db.insert_event(community, &old, None) + .await + .expect("carol posts before share recording"); + let deleted = attachment(&carol, &deleted_history, None); + db.insert_event(community, &deleted, None).await.unwrap(); + db.soft_delete_event_and_update_thread(community, deleted.id.as_bytes(), None, None) + .await + .unwrap(); + + let mut tx = crate::begin_community_write_transaction( + &db.pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await + .expect("admit"); + sqlx::query( + "UPDATE events SET received_at = 'epoch' WHERE community_id = $1 AND id = ANY($2)", + ) + .bind(community.as_uuid()) + .bind(vec![ + old.id.to_bytes().to_vec(), + deleted.id.to_bytes().to_vec(), + ]) + .execute(tx.conn()) + .await + .expect("age carol's event"); + sqlx::query("DELETE FROM media_shares WHERE community_id = $1") + .bind(community.as_uuid()) + .execute(tx.conn()) + .await + .expect("forget shares"); + tx.commit().await.expect("commit"); + + // Surviving server upload audits recover legacy ownership; malformed, + // missing, or unrelated audit evidence never creates uploader proof. + let (recovered, unproven) = (unique_sha(), unique_sha()); + let (uploader, stranger) = (Keys::generate(), Keys::generate()); + db.insert_event(community, &attachment(&uploader, &recovered, None), None) + .await + .unwrap(); + db.insert_event(community, &attachment(&stranger, &unproven, None), None) + .await + .unwrap(); + let mut tx = db.begin_media_share_transaction(community).await.unwrap(); + for (seq, action, actor, object) in [ + ( + 1_i64, + "media_uploaded", + Some(uploader.public_key().to_bytes().to_vec()), + recovered.clone(), + ), + (2, "media_uploaded", None, unproven.clone()), + (3, "media_uploaded", Some(vec![1; 31]), unproven.clone()), + ( + 4, + "event_created", + Some(stranger.public_key().to_bytes().to_vec()), + unproven.clone(), + ), + ( + 5, + "media_uploaded", + Some(stranger.public_key().to_bytes().to_vec()), + "bad-hash".into(), + ), + ] { + sqlx::query( + "INSERT INTO audit_log (community_id, seq, hash, action, actor_pubkey, object_id) \ + VALUES ($1,$2,$3,$4,$5,$6)", + ) + .bind(community.as_uuid()) + .bind(seq) + .bind(vec![seq as u8; 32]) + .bind(action) + .bind(actor) + .bind(object) + .execute(tx.conn()) + .await + .unwrap(); + } + tx.commit().await.unwrap(); + let (fenced, fenced_community) = fixture().await; + let mut tx = fenced + .begin_media_share_transaction(fenced_community) + .await + .unwrap(); + sqlx::query( + "INSERT INTO audit_log (community_id, seq, hash, action, actor_pubkey, object_id) \ + VALUES ($1,1,$2,'media_uploaded',$3,$4)", + ) + .bind(fenced_community.as_uuid()) + .bind(vec![1_u8; 32]) + .bind(stranger.public_key().to_bytes().as_slice()) + .bind(&unproven) + .execute(tx.conn()) + .await + .unwrap(); + tx.commit().await.unwrap(); + quiesce(&fenced, fenced_community).await; + + db.backfill_media_shares(1).await.expect("backfill"); + assert_eq!(db.backfill_media_shares(1).await.expect("rerun"), 0); + assert_eq!( + shares(&db, community, &sha).await, + vec![Some(private)], + "a reference live recording refused must stay unshared" + ); + assert_eq!(shares(&db, community, &historical).await, vec![None]); + let fenced_proofs: i64 = + sqlx::query_scalar("SELECT count(*) FROM media_uploaders WHERE community_id = $1") + .bind(fenced_community.as_uuid()) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!( + fenced_proofs, 0, + "audit recovery must not write across a fenced tenant boundary" + ); + assert_eq!( + shares(&db, community, &recovered).await, + vec![None], + "retained upload evidence must recover an uploader's intended share" + ); + assert!(shares(&db, community, &unproven).await.is_empty()); + let proofs: Vec<(String, Vec)> = sqlx::query_as( + "SELECT sha256, pubkey FROM media_uploaders WHERE community_id = $1 AND sha256 = ANY($2)", + ).bind(community.as_uuid()).bind(vec![recovered.clone(), unproven]) + .fetch_all(&db.pool).await.unwrap(); + assert_eq!( + proofs, + vec![(recovered, uploader.public_key().to_bytes().to_vec())] + ); + assert!( + shares(&db, community, &deleted_history).await.is_empty(), + "legacy deleted messages must never gain a share" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn migration_schema_backfill_trusts_history_and_rechecks_recorded_events() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to migration DB"); + crate::migration::run_migrations_through(&pool, 68) + .await + .expect("migrate"); + pool.close().await; + check_backfill().await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn agent_profile_picture_uses_only_its_registered_owners_upload_proof() { + let (db, community) = fixture().await; + let (owner, agent) = (Keys::generate(), Keys::generate()); + let (avatar, secret, unknown) = (unique_sha(), unique_sha(), unique_sha()); + for sha in [&avatar, &secret] { + upload(&db, community, &owner, sha).await; + } + let profile = |picture: &str| { + EventBuilder::new( + Kind::Metadata, + serde_json::json!({ + "name": Uuid::new_v4().to_string(), + "picture": format!("/media/{picture}.jpg"), + "about": format!("/media/{secret}.jpg"), + }) + .to_string(), + ) + .tags([Tag::parse(["p", &owner.public_key().to_hex()]).unwrap()]) + .sign_with_keys(&agent) + .unwrap() + }; + db.insert_event(community, &profile(&avatar), None) + .await + .unwrap(); + assert!( + shares(&db, community, &avatar).await.is_empty(), + "an unverified owner tag grants nothing" + ); + for keys in [&owner, &agent] { + db.ensure_user(community, &keys.public_key().to_bytes()) + .await + .unwrap(); + } + db.set_agent_owner( + community, + &agent.public_key().to_bytes(), + &owner.public_key().to_bytes(), + ) + .await + .unwrap(); + db.insert_event(community, &profile(&avatar), None) + .await + .unwrap(); + assert_eq!(shares(&db, community, &avatar).await, vec![None]); + assert!( + shares(&db, community, &secret).await.is_empty(), + "owner proof applies only to picture" + ); + db.insert_event(community, &profile(&unknown), None) + .await + .unwrap(); + assert!( + shares(&db, community, &unknown).await.is_empty(), + "the owner must have uploaded these bytes" + ); + db.insert_event(community, &attachment(&agent, &secret, None), None) + .await + .unwrap(); + assert!( + shares(&db, community, &secret).await.is_empty(), + "ordinary events cannot borrow the owner's uploads" + ); + } +} diff --git a/crates/buzz-db/src/store/media_uploader.rs b/crates/buzz-db/src/store/media_uploader.rs new file mode 100644 index 00000000000..4ced9087b59 --- /dev/null +++ b/crates/buzz-db/src/store/media_uploader.rs @@ -0,0 +1,173 @@ +//! Who has uploaded each blob, as proof of possession for sharing it. + +use buzz_datastore_tracing::datastore_span; + +use crate::deletion::ServingWriteLease; +use crate::{error::Result, observability, Db}; +use buzz_core::CommunityId; + +impl Db { + /// Recover possession proof from retained server-authored upload audits. + /// Scan a finite sequence range using audit_log's tenant-leading primary + /// key; malformed or unattributed records never establish ownership. + #[datastore_span(name = "backfill_media_uploaders", system = "postgresql")] + pub(crate) async fn backfill_media_uploaders( + &self, + community: CommunityId, + batch_size: i64, + ) -> Result { + let mut conn = + observability::acquire_writer(&self.pool, observability::WriterOperation::Maintenance) + .await?; + let upper: Option = + sqlx::query_scalar("SELECT max(seq) FROM audit_log WHERE community_id = $1") + .bind(community.as_uuid()) + .fetch_one(&mut *conn) + .await?; + drop(conn); + let Some(upper) = upper else { + return Ok(0); + }; + let (mut cursor, mut added) = (0_i64, 0_u64); + while cursor < upper { + let mut tx = crate::begin_community_write_transaction( + &self.pool, + community, + observability::WriterOperation::Maintenance, + ) + .await?; + let page: Vec = sqlx::query_scalar( + "SELECT seq FROM audit_log WHERE community_id = $1 AND seq > $2 AND seq <= $3 \ + ORDER BY seq LIMIT $4", + ) + .bind(community.as_uuid()) + .bind(cursor) + .bind(upper) + .bind(batch_size) + .fetch_all(tx.conn()) + .await?; + let Some(&last) = page.last() else { + break; + }; + added += sqlx::query( + "INSERT INTO media_uploaders (community_id, sha256, pubkey, created_at) \ + SELECT community_id, object_id, actor_pubkey, created_at FROM audit_log \ + WHERE community_id = $1 AND seq > $2 AND seq <= $3 \ + AND action = 'media_uploaded' AND octet_length(actor_pubkey) = 32 \ + AND object_id ~ '^[0-9a-f]{64}$' \ + ORDER BY object_id, actor_pubkey ON CONFLICT DO NOTHING", + ) + .bind(community.as_uuid()) + .bind(cursor) + .bind(last) + .execute(tx.conn()) + .await? + .rows_affected(); + tx.commit().await?; + cursor = last; + } + Ok(added) + } + + /// Record that `pubkey` uploaded the blob `sha256`; repeat uploads are + /// no-ops. The write presents the upload's serving lease, so it may finish + /// while the community quiesces but never after the lease is gone. + #[datastore_span(name = "record_media_uploader", system = "postgresql")] + pub async fn record_media_uploader( + &self, + lease: &ServingWriteLease, + sha256: &str, + pubkey: &[u8], + ) -> Result<()> { + let connection = + observability::acquire_writer(&self.pool, observability::WriterOperation::EventWrite) + .await?; + let mut tx = sqlx::Transaction::begin(connection, None).await?; + self.deletion_store() + .guard_transaction_with_serving_lease(&mut tx, lease) + .await?; + sqlx::query( + "INSERT INTO media_uploaders (community_id, sha256, pubkey) VALUES ($1, $2, $3) \ + ON CONFLICT DO NOTHING", + ) + .bind(lease.community_id.as_uuid()) + .bind(sha256) + .bind(pubkey) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) + } +} + +#[cfg(test)] +mod postgres_tests { + use std::time::Duration; + + use super::*; + use buzz_core::CommunityId; + use sqlx::PgPool; + use uuid::Uuid; + + const SHA: &str = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"; + + async fn db_with_lease() -> (Db, CommunityId, ServingWriteLease) { + let db = Db::from_pool( + PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to test DB"), + ); + let host = format!("uploader-{}.example", Uuid::new_v4().simple()); + let community_id = db + .ensure_configured_community(&host) + .await + .expect("seed community") + .id; + let lease = db + .deletion_store() + .acquire_serving_write_lease( + community_id, + "media_upload", + "media-uploader-test", + Duration::from_secs(60), + ) + .await + .expect("acquire lease"); + (db, community_id, lease) + } + + async fn uploader_rows(db: &Db, community_id: CommunityId) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM media_uploaders WHERE community_id = $1") + .bind(community_id.as_uuid()) + .fetch_one(&db.pool) + .await + .expect("count uploaders") + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn repeat_upload_records_one_row() { + let (db, community_id, lease) = db_with_lease().await; + for _ in 0..2 { + db.record_media_uploader(&lease, SHA, &[7; 32]) + .await + .expect("record uploader"); + } + assert_eq!(uploader_rows(&db, community_id).await, 1); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn released_lease_refuses_the_write() { + let (db, community_id, lease) = db_with_lease().await; + db.deletion_store() + .release_serving_write_lease(&lease) + .await + .expect("release lease"); + assert!(db + .record_media_uploader(&lease, SHA, &[7; 32]) + .await + .is_err()); + assert_eq!(uploader_rows(&db, community_id).await, 0); + } +} diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 6850b625cf1..f1e290fb15b 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -32,6 +32,13 @@ mod event_follow_up_postgres_tests; pub mod feed; /// Git repository name registry (NIP-34 kind:30617). pub mod git_repo; +/// Bounded reclamation of inactive media grants. +pub mod media_cleanup; +/// Where each media blob is shared, recorded with the events that reference it. +mod media_grant; +pub mod media_reference; +/// Who has uploaded each blob, as proof of possession for sharing it. +pub mod media_uploader; /// Community moderation: reports, bans/timeouts, audit actions. pub mod moderation; /// Deployment-global operator-listener mention registrations and delivery queues. diff --git a/crates/buzz-db/src/store/push.rs b/crates/buzz-db/src/store/push.rs index 24ae312521e..d4ae0b86b3c 100644 --- a/crates/buzz-db/src/store/push.rs +++ b/crates/buzz-db/src/store/push.rs @@ -360,13 +360,7 @@ pub async fn accept_lease_event( return Err(error.into()); } // The row above is always kind 30350 with no channel. - crate::store::event_follow_up::after_admitted_insert( - &mut tx, - event.id.as_bytes().as_slice(), - 30350, - None, - ) - .await?; + crate::store::event_follow_up::after_admitted_insert(&mut tx, event, None).await?; let (is_active, endpoint_hash, endpoint_grant, max_class, subscriptions) = active .map_or((false, None, None, None, None), |active| { diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 6bc4de1a959..43154fe4baf 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -1170,13 +1170,7 @@ pub(crate) async fn write_nip43_membership_snapshot( .bind(d_tag.as_deref()) .execute(tx.conn()) .await?; - crate::store::event_follow_up::after_admitted_insert( - tx, - event.id.as_bytes().as_slice(), - kind_i32, - None, - ) - .await?; + crate::store::event_follow_up::after_admitted_insert(tx, &event, None).await?; Ok(StoredEvent::with_received_at( event, diff --git a/crates/buzz-db/src/store/replaceable.rs b/crates/buzz-db/src/store/replaceable.rs index 1ccee163320..4db5e590514 100644 --- a/crates/buzz-db/src/store/replaceable.rs +++ b/crates/buzz-db/src/store/replaceable.rs @@ -262,6 +262,57 @@ async fn replace_parameterized_event_in_transaction_impl( let tags_json = serde_json::to_value(&event.tags)?; crate::channel_members::guard_channel_roster_event(tx, kind_i32, channel_id, &tags_json) .await?; + + let sig_bytes = event.sig.serialize(); + let insert_result = sqlx::query( + "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \ + ON CONFLICT DO NOTHING", + ) + .bind(community_id.as_uuid()) + .bind(incoming_id) + .bind(pubkey_bytes.as_slice()) + .bind(created_at) + .bind(kind_i32) + .bind(&tags_json) + .bind(&event.content) + .bind(sig_bytes.as_slice()) + .bind(received_at) + .bind(channel_id) + .bind(d_tag) + .bind(crate::event::extract_not_before(event)) + .execute(tx.conn()) + .await?; + + if insert_result.rows_affected() == 0 { + return Ok(false); + } + + if is_nip_rs { + sqlx::query( + "INSERT INTO parameterized_event_watermarks \ + (community_id, kind, pubkey, d_tag, created_at, event_id) \ + VALUES ($1, $2, $3, $4, $5, $6) \ + ON CONFLICT (community_id, kind, pubkey, d_tag) DO UPDATE SET \ + created_at = EXCLUDED.created_at, event_id = EXCLUDED.event_id", + ) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(d_tag) + .bind(created_at) + .bind(incoming_id) + .execute(tx.conn()) + .await?; + } + + crate::insert_mentions_in_transaction(&mut *tx, event, channel_id).await?; + // Recover eligible legacy references before retiring the old head. + if let Some((_, previous_id)) = &existing { + crate::store::media_grant::bootstrap_replacement(tx, event, previous_id).await?; + } + // Authorize retained references while the previous head is still live. + crate::store::event_follow_up::record_admitted_media(tx, event, channel_id).await?; if existing.is_some() { let previous_nip_rs_hard_delete: Option = if is_nip_rs { sqlx::query_scalar( @@ -279,16 +330,17 @@ async fn replace_parameterized_event_in_transaction_impl( } let statement = if hard_delete_superseded { "DELETE FROM events \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL" + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL AND id <> $5" } else { "UPDATE events SET deleted_at = NOW() \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL" + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL AND id <> $5" }; sqlx::query(statement) .bind(community_id.as_uuid()) .bind(kind_i32) .bind(pubkey_bytes.as_slice()) .bind(d_tag) + .bind(incoming_id) .execute(tx.conn()) .await?; @@ -311,50 +363,6 @@ async fn replace_parameterized_event_in_transaction_impl( } } - let sig_bytes = event.sig.serialize(); - let insert_result = sqlx::query( - "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \ - ON CONFLICT DO NOTHING", - ) - .bind(community_id.as_uuid()) - .bind(incoming_id) - .bind(pubkey_bytes.as_slice()) - .bind(created_at) - .bind(kind_i32) - .bind(&tags_json) - .bind(&event.content) - .bind(sig_bytes.as_slice()) - .bind(received_at) - .bind(channel_id) - .bind(d_tag) - .bind(crate::event::extract_not_before(event)) - .execute(tx.conn()) - .await?; - - if insert_result.rows_affected() == 0 { - return Ok(false); - } - - if is_nip_rs { - sqlx::query( - "INSERT INTO parameterized_event_watermarks \ - (community_id, kind, pubkey, d_tag, created_at, event_id) \ - VALUES ($1, $2, $3, $4, $5, $6) \ - ON CONFLICT (community_id, kind, pubkey, d_tag) DO UPDATE SET \ - created_at = EXCLUDED.created_at, event_id = EXCLUDED.event_id", - ) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(d_tag) - .bind(created_at) - .bind(incoming_id) - .execute(tx.conn()) - .await?; - } - - crate::insert_mentions_in_transaction(&mut *tx, event, channel_id).await?; Ok(true) } .await; @@ -383,8 +391,7 @@ async fn replace_parameterized_event_in_transaction_impl( .execute(tx.conn()) .await?; if status == ParameterizedReplaceStatus::Inserted { - crate::store::event_follow_up::after_admitted_insert(tx, incoming_id, kind_i32, channel_id) - .await?; + crate::store::event_follow_up::stage_admitted_insert(tx, event, channel_id); } Ok(ParameterizedReplaceResult::new( @@ -468,9 +475,9 @@ impl Db { // NIP-16: created_at is second-resolution. On same-second tie, lowest // event id (lexicographic) wins — deterministic across relays. let incoming_id = event.id.as_bytes().as_slice(); - if let Some((existing_ts, existing_id)) = existing { - let dominated = created_at < existing_ts - || (created_at == existing_ts + if let Some((existing_ts, existing_id)) = &existing { + let dominated = created_at < *existing_ts + || (created_at == *existing_ts && incoming_id >= existing_id.as_slice()); if dominated { tx.rollback().await?; @@ -493,20 +500,6 @@ impl Db { ) .await?; - // Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety. - sqlx::query( - "UPDATE events SET deleted_at = NOW() \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ - AND channel_id IS NOT DISTINCT FROM $4 \ - AND deleted_at IS NULL", - ) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(channel_id) - .execute(tx.conn()) - .await?; - // Insert the new event inside the same transaction. let sig_bytes = event.sig.serialize(); let received_at = chrono::Utc::now(); @@ -533,8 +526,7 @@ impl Db { let was_inserted = insert_result.rows_affected() > 0; if !was_inserted { - // ON CONFLICT fired — the event ID already exists. Rollback the - // soft-delete so we don't lose the previous replaceable event. + // ON CONFLICT fired — leave the previously-live event intact. tx.rollback().await?; return Ok(( StoredEvent::with_received_at( @@ -547,12 +539,27 @@ impl Db { )); } - crate::store::event_follow_up::after_admitted_insert( - &mut tx, - event.id.as_bytes().as_slice(), - kind_i32, - channel_id, + if let Some((_, previous_id)) = &existing { + crate::store::media_grant::bootstrap_replacement(&mut tx, event, previous_id) + .await?; + } + // Authorize retained references while the previous head is still live. + crate::store::event_follow_up::after_admitted_insert(&mut tx, event, channel_id) + .await?; + + // Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety. + sqlx::query( + "UPDATE events SET deleted_at = NOW() \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ + AND channel_id IS NOT DISTINCT FROM $4 \ + AND deleted_at IS NULL AND id <> $5", ) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(channel_id) + .bind(incoming_id) + .execute(tx.conn()) .await?; // The replaceable event and its denormalized mention index are one diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index b1c95d7f1c7..dbf96bd621b 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -1643,12 +1643,14 @@ fn events_insert_tails(source: &str) -> Vec { /// An events insert must run both follow-ups: the push enqueue and the /// pre-commit TTL refresh. `after_admitted_insert` does both; a writer that -/// inserts inside a savepoint calls the two halves separately. +/// inserts inside a savepoint records media there and stages delivery after it. /// /// Only real calls count: a hook in a comment or a string literal does not. fn runs_event_follow_ups(function_source: &str) -> bool { let code = mask_comments_and_literals(function_source); code.contains("event_follow_up::after_admitted_insert(") + || (code.contains("event_follow_up::record_admitted_media(") + && code.contains("event_follow_up::stage_admitted_insert(")) || (code.contains("event_follow_up::enqueue_push_match(") && code.contains(".record_channel_event(")) } @@ -1696,6 +1698,12 @@ pub(crate) async fn hooked_writer(tx: &mut AdmittedTx) { .await .expect("follow up"); } +pub(crate) async fn media_savepoint_writer(tx: &mut AdmittedTx) { + sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") + .execute(tx.conn()).await.expect("write"); + crate::store::event_follow_up::record_admitted_media(tx, event, channel).await.expect("media"); + crate::store::event_follow_up::stage_admitted_insert(tx, event, channel); +} pub(crate) async fn savepoint_writer(tx: &mut AdmittedTx) { sqlx::query("INSERT INTO events (community_id, id) VALUES ($1, $2)") .execute(tx.conn()) diff --git a/crates/buzz-relay/src/api/media.rs b/crates/buzz-relay/src/api/media.rs index 9c5f3dd93b3..137f5acf01c 100644 --- a/crates/buzz-relay/src/api/media.rs +++ b/crates/buzz-relay/src/api/media.rs @@ -579,6 +579,24 @@ async fn upload_blob_result( auth.tenant.host(), ); + state + .db + .record_media_uploader( + serving_write.lease(), + &descriptor.sha256, + auth.auth_event.pubkey.as_bytes(), + ) + .await + .map_err(|error| { + let error = anyhow::Error::from(error); + if buzz_deletion::ServingWriteGuard::is_lease_lost(&error) { + serving_lease_lost(error) + } else { + tracing::error!(%error, "media uploader recording failed"); + MediaError::ServiceUnavailable + } + })?; + // Normalize MIME to a known set to bound label cardinality. let mime_label = match descriptor.mime_type.as_str() { "image/jpeg" | "image/png" | "image/gif" | "image/webp" | "video/mp4" => { @@ -2144,11 +2162,13 @@ mod tests { const TEST_ISSUER: &str = "https://issuer.example"; const TEST_AUDIENCE: &str = "https://relay.example"; const TEST_KID: &str = "test-key-1"; - const TEST_EC_PKCS8_PEM: &str = "-----BEGIN PRIVATE KEY-----\n\ - MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgcnxDM4EiirH9dHUE\n\ - WZc759TX4s5PAn8kO5ovXSnGxCWhRANCAARFb6ZnsfkqOOXyEhj3KBQphGKF4vTa\n\ - zhebbavbZ1ZoklqkF1cGg+jTO7rONAVEzXvXUWtV6CdDV+rybiVmFP2w\n\ - -----END PRIVATE KEY-----\n"; + const TEST_EC_PKCS8_PEM: &str = concat!( + "-----BEGIN PRIVATE KEY-----\n", // sadscan:disable kingfisher.privkey.2 -- static test-only key + "MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgcnxDM4EiirH9dHUE\n", + "WZc759TX4s5PAn8kO5ovXSnGxCWhRANCAARFb6ZnsfkqOOXyEhj3KBQphGKF4vTa\n", + "zhebbavbZ1ZoklqkF1cGg+jTO7rONAVEzXvXUWtV6CdDV+rybiVmFP2w\n", + "-----END PRIVATE KEY-----\n", + ); /// Build an AppState with NIP-FI Enforce and a real injected P-256 verifier. async fn media_enforce_test_state() -> Option> { diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index fa181f2a473..725ca1d8d61 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -134,6 +134,7 @@ const _: () = assert!(FLEET_STOCK_COLLECTION_DEADLINE.as_secs() > 5); const _: () = assert!(FLEET_ACTIVITY_COLLECTION_DEADLINE.as_secs() > 15); struct FleetUsageSchedule { + media_cleanup_cursor: Option, next_stock: std::time::Instant, next_activity: std::time::Instant, last_stock_success: Option, @@ -167,6 +168,7 @@ impl FleetUsageSchedule { retry_interval: std::time::Duration, ) -> Self { Self { + media_cleanup_cursor: None, next_stock: now, next_activity: now, last_stock_success: None, @@ -2505,6 +2507,20 @@ async fn run_usage_metrics_tick( } } if leader.is_some() { + // Run maintenance before optional telemetry collection. + match state + .db + .reap_inactive_media_shares(fleet_schedule.media_cleanup_cursor.as_ref(), 1000) + .await + { + Ok(batch) => { + fleet_schedule.media_cleanup_cursor = batch.next_cursor; + if batch.removed > 0 { + info!(removed = batch.removed, "reaped inactive media shares"); + } + } + Err(error) => warn!(error = %error, "failed to reap inactive media shares"), + } if emission_scope.includes_per_community() { let result = emit_db_usage_metrics(state, emission_scope, &host_map).await; record_legacy_usage_collection(result.is_ok(), fleet_schedule); diff --git a/crates/buzz-test-client/tests/e2e_media_extended.rs b/crates/buzz-test-client/tests/e2e_media_extended.rs index 9bd73b13fcb..aefee104e4a 100644 --- a/crates/buzz-test-client/tests/e2e_media_extended.rs +++ b/crates/buzz-test-client/tests/e2e_media_extended.rs @@ -37,6 +37,12 @@ fn http_client() -> Client { .expect("http client") } +async fn media_db_pool() -> sqlx::PgPool { + let database = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".into()); // sadscan:disable np.postgres.1 -- local test-only credentials + sqlx::PgPool::connect(&database).await.unwrap() +} + fn sign_blossom_auth(keys: &Keys, sha256: &str) -> nostr::Event { let now = Timestamp::now().as_secs(); let server = relay_server_authority(); @@ -200,6 +206,24 @@ async fn test_upload_png_roundtrip() { // GET back — reads are authenticated, so scope a token to the uploaded hash. let sha256 = desc["sha256"].as_str().expect("descriptor sha256"); + // Bind uploader recording to the real HTTP handler, independently of + // whether this relay currently enforces file reads. + let pool = media_db_pool().await; + let recorded: i64 = sqlx::query_scalar( + "SELECT count(*) FROM media_uploaders u JOIN communities c ON c.id = u.community_id \ + WHERE c.host = $1 AND u.sha256 = $2 AND u.pubkey = $3", + ) + .bind(relay_server_authority()) + .bind(sha256) + .bind(keys.public_key().to_bytes().as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + recorded, 1, + "successful upload must persist its authenticated uploader" + ); + pool.close().await; let get = client .get(desc["url"].as_str().unwrap()) .header( diff --git a/docs/database.md b/docs/database.md index 78ad9486dd4..0a3c9b58e70 100644 --- a/docs/database.md +++ b/docs/database.md @@ -100,7 +100,7 @@ writer of a fenced table already goes through admission. The remaining fence-retirement work drops the fence triggers and functions, and MUST also update ARCHITECTURE.md § *Dual enforcement*. -**Today this blocks new tenant-owned tables.** Startup and deletion validation +**New tenant-owned tables use application admission.** Startup and deletion validation no longer require a fence on any table; they pass with every universal fence object dropped (`migration_schema_fence_objects_are_optional_for_serving_and_deletion` in `crates/buzz-db/src/store/deletion.rs`). Two things still matter: @@ -110,21 +110,19 @@ in `crates/buzz-db/src/store/deletion.rs`). Two things still matter: (`community deletion catalog drift (missing=…, unknown=…)`), and deletion fails closed for every community, not only the new table's. Listing the table is enough to pass. -- Desired-state bootstraps run `schema/community_write_fence.sql`, which - attaches a fence trigger to every `community_id` table not excluded by - `community_write_fence_excluded_table`. A fresh desired-state database would - put a trigger on the new table and a migrated one would not, so the two - schema paths would diverge. - -So before a feature can add a tenant-owned `community_id` table, the -desired-state bootstrap must stop attaching fence triggers. That change belongs -to the fence-retirement work. Land it before the feature, or as the feature's -first PR. Do not work around it with a trigger, or with an entry in +- Desired-state bootstraps install only the legacy fences explicitly listed + under `schema/triggers/`. They do not discover and fence new tables. + Keep those existing backstops until their own retirement is verified; + new tables receive no fence in either schema path. + +The blanket desired-state fence attachment has been removed. A feature may +add a tenant-owned `community_id` table when it follows the checklist below. +Do not add a trigger, or an entry in `community_write_fence_excluded_table` for a table the community owns. #### Checklist: adding a tenant-owned `community_id` table -After the desired-state change above has landed, all in the same PR: +All in the same PR: - [ ] Every writer of the table goes through community admission (`begin_community_write_transaction` or another constructor that returns diff --git a/migrations/0068_media_access.sql b/migrations/0068_media_access.sql new file mode 100644 index 00000000000..2e8fd7289c9 --- /dev/null +++ b/migrations/0068_media_access.sql @@ -0,0 +1,34 @@ +SET LOCAL lock_timeout = '5s'; + +-- Durable historical backfill cutoff, independent of the schema install tool. +ALTER TABLE communities + ADD COLUMN media_shares_recording_started_at TIMESTAMPTZ NOT NULL DEFAULT now(); + +-- Channel-scoped media access: who may read each blob. +-- +-- media_uploaders: who has uploaded each blob. Uploading the exact bytes +-- proves possession, so an uploader may share a blob into their own channels +-- even when the same bytes are also shared somewhere they cannot read. +CREATE TABLE media_uploaders ( + community_id UUID NOT NULL, + sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'), + pubkey BYTEA NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (community_id, sha256, pubkey) +); + +-- media_shares: one grant per referencing event and blob. A NULL channel +-- shares with the whole community. Deleted or purged source events grant +-- nothing; event_created_at locates the source in the partitioned table. +CREATE TABLE media_shares ( + community_id UUID NOT NULL, + sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'), + channel_id UUID, + event_id BYTEA NOT NULL, + event_created_at TIMESTAMPTZ NOT NULL, + target_event_id BYTEA, + target_event_created_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CHECK ((target_event_id IS NULL) = (target_event_created_at IS NULL)), + PRIMARY KEY (community_id, sha256, event_id) +); diff --git a/schema/community_write_fence.sql b/schema/community_write_fence.sql deleted file mode 100644 index 2ebc9fd1f98..00000000000 --- a/schema/community_write_fence.sql +++ /dev/null @@ -1,31 +0,0 @@ --- Attach the universal fence to every existing table carrying community_id, --- including deployment-private sidecars whose community_id is provenance. --- --- Desired-state schema application does not replay migration history, so the --- explicit `SELECT attach_community_write_fence('')` calls in --- `triggers/
.sql` stay as first-class catalog declarations. They also --- make the fence contract visible to migration linting instead of hiding it --- only in this dynamic bootstrap loop. This loop runs last, so it fences every --- desired-state table regardless: a fenced desired state does not prove the --- migration path attaches the fence. Keep the explicit calls. -DO $$ -DECLARE - target REGCLASS; -BEGIN - FOR target IN - SELECT c.oid::REGCLASS - FROM pg_class c - JOIN pg_namespace n ON n.oid = c.relnamespace - JOIN pg_attribute a ON a.attrelid = c.oid - WHERE n.nspname = current_schema() - AND c.relkind IN ('r', 'p') - AND NOT c.relispartition - AND a.attname = 'community_id' - AND NOT a.attisdropped - AND NOT community_write_fence_excluded_table(c.relname) - ORDER BY c.oid::REGCLASS::TEXT - LOOP - PERFORM attach_community_write_fence(target); - END LOOP; -END -$$; diff --git a/schema/schema.sql b/schema/schema.sql index d252cbdb7ea..1ee65c207b2 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -7,9 +7,9 @@ -- -- Everything SchemaBot cannot apply lives in this directory's other -- subdirectories: `types/`, `functions/`, `partitions/`, `triggers/` (including --- write-fence attachment), `seeds/`, and the `community_write_fence.sql` --- bootstrap. This file composes all of them into one fresh database, in load --- order: types → functions → tables → partitions → triggers → seeds → fence. +-- explicit legacy write-fence attachment), and `seeds/`. This file composes +-- them into one fresh database, in load order: +-- types → functions → tables → partitions → triggers → seeds. -- -- `./bin/pgschema apply --file schema/schema.sql` resolves the `\i` includes -- relative to this file (it refuses `..`). `pgschema` skips INSERTs and some @@ -107,6 +107,8 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i tables/public/moderation_reports.sql \i tables/public/community_bans.sql \i tables/public/community_nip_fi_issuers.sql +\i tables/public/media_shares.sql +\i tables/public/media_uploaders.sql \i tables/public/_operator_global_tables.sql \i tables/public/git_repo_names.sql \i tables/public/parameterized_event_watermarks.sql @@ -209,4 +211,3 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i seeds/storage_accounting_snapshots.sql -- Fence any community-scoped table that lacks an explicit attach. -\i community_write_fence.sql diff --git a/schema/tables/public/communities.sql b/schema/tables/public/communities.sql index 06f74c3f585..ea27392437b 100644 --- a/schema/tables/public/communities.sql +++ b/schema/tables/public/communities.sql @@ -20,6 +20,8 @@ CREATE TABLE communities ( -- Added by migration 0003; kept here so desired-state applies match. icon TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + -- The share-recording cutoff exists in both desired-state and migrated DBs. + media_shares_recording_started_at TIMESTAMPTZ NOT NULL DEFAULT now(), archived_at TIMESTAMPTZ, deletion_state TEXT NOT NULL DEFAULT 'active' CHECK (deletion_state IN ('active', 'quiescing', 'fenced', 'tombstone')), deletion_fence_generation BIGINT NOT NULL DEFAULT 0 CHECK (deletion_fence_generation >= 0), diff --git a/schema/tables/public/media_shares.sql b/schema/tables/public/media_shares.sql new file mode 100644 index 00000000000..600574f766c --- /dev/null +++ b/schema/tables/public/media_shares.sql @@ -0,0 +1,17 @@ +-- ── Media shares ────────────────────────────────────────────────────────────── +-- Each event's grant to a channel, or the whole community for a NULL channel. +-- Reads join the source event: deleted or purged events grant no access. +-- Keep its timestamp for an indexed lookup in the partitioned events table. + +CREATE TABLE media_shares ( + community_id UUID NOT NULL, + sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'), + channel_id UUID, + event_id BYTEA NOT NULL, + event_created_at TIMESTAMPTZ NOT NULL, + target_event_id BYTEA, + target_event_created_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CHECK ((target_event_id IS NULL) = (target_event_created_at IS NULL)), + PRIMARY KEY (community_id, sha256, event_id) +); diff --git a/schema/tables/public/media_uploaders.sql b/schema/tables/public/media_uploaders.sql new file mode 100644 index 00000000000..1add946f3e2 --- /dev/null +++ b/schema/tables/public/media_uploaders.sql @@ -0,0 +1,11 @@ +-- ── Media uploaders ─────────────────────────────────────────────────────────── +-- Who has uploaded each blob in a community. Uploading the exact bytes proves +-- possession, so an uploader may share a blob into their own channels. + +CREATE TABLE media_uploaders ( + community_id UUID NOT NULL, + sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'), + pubkey BYTEA NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (community_id, sha256, pubkey) +);