diff --git a/Cargo.lock b/Cargo.lock index 798825deaf..298ba4cee2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2605,7 +2605,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 2.0.117", + "syn 1.0.109", ] [[package]] @@ -8814,9 +8814,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.46" +version = "0.23.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf" +checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" dependencies = [ "aws-lc-rs", "log", diff --git a/crates/buzz-core/src/ifc.rs b/crates/buzz-core/src/ifc.rs new file mode 100644 index 0000000000..986688c7e3 --- /dev/null +++ b/crates/buzz-core/src/ifc.rs @@ -0,0 +1,257 @@ +//! Shared version-one remote-agent information-flow fence grammar. + +use std::collections::HashSet; + +use nostr::Event; +use thiserror::Error; +use uuid::Uuid; + +use crate::kind::{event_kind_u32, KIND_CANVAS, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2}; + +/// Version carried by all supported information-flow fence tags. +pub const IFC_VERSION: &str = "1"; +/// Maximum distinct source policies on one publication. +pub const IFC_MAX_SOURCES: usize = 8; +/// Maximum future lifetime of a new publication fence. +pub const IFC_MAX_TTL_SECONDS: i64 = 300; +/// Duration for which an accepted event may be retried as a historic duplicate. +pub const IFC_RECEIPT_RETENTION_SECONDS: i64 = 86_400; +/// Channel event kinds that may carry a version-one publication fence. +pub const IFC_CHANNEL_PUBLICATION_KINDS: [u32; 3] = + [KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2, KIND_CANVAS]; + +/// One channel policy coordinate asserted by the trusted broker. +#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)] +pub struct IfcPolicyRef { + /// Channel whose canonical policy is referenced. + pub channel_id: Uuid, + /// Positive monotone generation encoded canonically on the wire. + pub generation: i64, +} + +/// Strictly parsed version-one channel publication fence. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct IfcPublicationFence { + /// Host-bound community asserted in the signed event. + pub community_id: Uuid, + /// Fence expiry as Unix seconds. + pub expires_at: i64, + /// Sorted, distinct source policy coordinates. + pub sources: Vec, + /// Destination channel policy coordinate. + pub destination: IfcPolicyRef, +} + +/// Failure while parsing reserved information-flow tags. +#[derive(Clone, Debug, Error, Eq, PartialEq)] +#[error("{0}")] +pub struct IfcTagError(pub String); + +/// Return whether an event contains any reserved information-flow tag. +#[must_use] +pub fn has_ifc_tag(event: &Event) -> bool { + event.tags.iter().any(|tag| { + tag.as_slice() + .first() + .is_some_and(|name| name.starts_with("ifc")) + }) +} + +/// Parse the strict version-one channel publication fence. +/// +/// Absence of every reserved tag returns `Ok(None)`. Presence of any `ifc` or +/// `ifc-*` tag opts the complete event into this grammar; partial, unknown, +/// duplicate, noncanonical, or unsupported forms return an error. +pub fn parse_publication_fence(event: &Event) -> Result, IfcTagError> { + if !has_ifc_tag(event) { + return Ok(None); + } + let kind = event_kind_u32(event); + if !IFC_CHANNEL_PUBLICATION_KINDS.contains(&kind) { + return Err(invalid("event kind is not enabled for fenced publication")); + } + + let mut common = None; + let mut destination = None; + let mut sources = Vec::new(); + for tag in event.tags.iter() { + let parts = tag.as_slice(); + let Some(name) = parts.first().map(String::as_str) else { + continue; + }; + if !name.starts_with("ifc") { + continue; + } + match name { + "ifc" => { + if common.is_some() || parts.len() != 4 || parts[1] != IFC_VERSION { + return Err(invalid("invalid or duplicate ifc tag")); + } + let community_id = canonical_uuid(&parts[2], "ifc community")?; + let expires_at = canonical_positive_i64(&parts[3], "ifc expiry")?; + common = Some((community_id, expires_at)); + } + "ifc-source" => { + if parts.len() != 3 { + return Err(invalid("ifc-source tag requires channel and generation")); + } + sources.push(IfcPolicyRef { + channel_id: canonical_uuid(&parts[1], "ifc source channel")?, + generation: canonical_positive_i64(&parts[2], "ifc source generation")?, + }); + } + "ifc-destination" => { + if destination.is_some() || parts.len() != 3 { + return Err(invalid("invalid or duplicate ifc-destination tag")); + } + destination = Some(IfcPolicyRef { + channel_id: canonical_uuid(&parts[1], "ifc destination channel")?, + generation: canonical_positive_i64(&parts[2], "ifc destination generation")?, + }); + } + _ => return Err(invalid("unknown reserved ifc tag")), + } + } + + let (community_id, expires_at) = common.ok_or_else(|| invalid("missing ifc tag"))?; + let destination = destination.ok_or_else(|| invalid("missing ifc-destination tag"))?; + if sources.is_empty() || sources.len() > IFC_MAX_SOURCES { + return Err(invalid("ifc-source count is outside the supported range")); + } + let ordered = sources + .windows(2) + .all(|pair| pair[0].channel_id.to_string() < pair[1].channel_id.to_string()); + let unique = sources + .iter() + .map(|source| source.channel_id) + .collect::>() + .len() + == sources.len(); + if !ordered || !unique { + return Err(invalid( + "ifc-source tags must be distinct and ordered by channel UUID", + )); + } + + let h_tags: Vec<_> = event + .tags + .iter() + .map(nostr::Tag::as_slice) + .filter(|parts| parts.first().is_some_and(|name| name == "h")) + .collect(); + if h_tags.len() != 1 || h_tags[0].len() != 2 { + return Err(invalid("fenced publication requires exactly one h tag")); + } + let event_channel = canonical_uuid(&h_tags[0][1], "h channel")?; + if event_channel != destination.channel_id { + return Err(invalid("ifc destination does not match the h channel")); + } + + Ok(Some(IfcPublicationFence { + community_id, + expires_at, + sources, + destination, + })) +} + +fn canonical_uuid(value: &str, field: &str) -> Result { + let parsed = Uuid::parse_str(value).map_err(|_| invalid(&format!("{field} is not a UUID")))?; + if parsed.to_string() != value { + return Err(invalid(&format!("{field} is not canonical"))); + } + Ok(parsed) +} + +fn canonical_positive_i64(value: &str, field: &str) -> Result { + if value.is_empty() + || value.starts_with('0') + || !value.bytes().all(|byte| byte.is_ascii_digit()) + { + return Err(invalid(&format!("{field} is not canonical"))); + } + value + .parse::() + .ok() + .filter(|number| *number > 0) + .ok_or_else(|| invalid(&format!("{field} is outside the supported range"))) +} + +fn invalid(message: &str) -> IfcTagError { + IfcTagError(message.to_owned()) +} + +#[cfg(test)] +mod tests { + use nostr::{EventBuilder, Keys, Kind, Tag}; + + use super::*; + + fn event(kind: u32, tags: &[&[&str]]) -> Event { + let tags = tags + .iter() + .map(|parts| Tag::parse(parts.iter().copied()).expect("valid test tag")) + .collect::>(); + EventBuilder::new(Kind::Custom(kind as u16), "test") + .tags(tags) + .sign_with_keys(&Keys::generate()) + .expect("sign test event") + } + + #[test] + fn absent_reserved_tags_preserve_legacy_events() { + assert_eq!( + parse_publication_fence(&event(KIND_STREAM_MESSAGE, &[])), + Ok(None) + ); + } + + #[test] + fn parses_canonical_channel_fence() { + let source = "11111111-1111-4111-8111-111111111111"; + let destination = "22222222-2222-4222-8222-222222222222"; + let parsed = parse_publication_fence(&event( + KIND_STREAM_MESSAGE, + &[ + &["h", destination], + &[ + "ifc", + "1", + "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + "1792000000", + ], + &["ifc-source", source, "17"], + &["ifc-destination", destination, "23"], + ], + )) + .expect("parse") + .expect("fence"); + assert_eq!(parsed.sources[0].generation, 17); + assert_eq!(parsed.destination.generation, 23); + } + + #[test] + fn any_reserved_tag_requires_complete_supported_grammar() { + let owner = "1111111111111111111111111111111111111111111111111111111111111111"; + for tags in [ + vec![vec![ + "ifc-source", + "11111111-1111-4111-8111-111111111111", + "1", + ]], + vec![vec!["ifc-owner-memory", owner]], + vec![vec![ + "ifc", + "2", + "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + "10", + ]], + ] { + let borrowed = tags + .iter() + .map(|parts| parts.as_slice()) + .collect::>(); + assert!(parse_publication_fence(&event(KIND_STREAM_MESSAGE, &borrowed)).is_err()); + } + } +} diff --git a/crates/buzz-core/src/kind.rs b/crates/buzz-core/src/kind.rs index db05ea922f..fe56dd2fd6 100644 --- a/crates/buzz-core/src/kind.rs +++ b/crates/buzz-core/src/kind.rs @@ -590,6 +590,8 @@ pub const KIND_ARTIFACT_REMOVAL: u32 = 45011; pub const KIND_FORUM_VOTE: u32 = 45002; /// A comment reply on a forum post. pub const KIND_FORUM_COMMENT: u32 = 45003; +/// Relay-signed, response-only remote-agent policy attestation. +pub const KIND_IFC_ATTESTATION: u32 = 45900; // Workflow engine (46000–46999) /// Trigger workflow execution. @@ -776,6 +778,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_ARTIFACT_REMOVAL, KIND_FORUM_VOTE, KIND_FORUM_COMMENT, + KIND_IFC_ATTESTATION, KIND_WORKFLOW_TRIGGER, KIND_APPROVAL_GRANT, KIND_APPROVAL_DENY, @@ -876,6 +879,7 @@ pub const fn is_relay_only_kind(kind: u32) -> bool { matches!( kind, KIND_ARTIFACT_REMOVAL + | KIND_IFC_ATTESTATION | KIND_NIP43_MEMBERSHIP_LIST | KIND_CHANNEL_SUMMARY | KIND_PRESENCE_SNAPSHOT diff --git a/crates/buzz-core/src/lib.rs b/crates/buzz-core/src/lib.rs index e7669c874b..6b3de01e9e 100644 --- a/crates/buzz-core/src/lib.rs +++ b/crates/buzz-core/src/lib.rs @@ -20,6 +20,8 @@ pub mod event; pub mod filter; /// Git permission types — ref patterns, protection rules, policy evaluation. pub mod git_perms; +/// Remote-agent information-flow fence protocol types and validation. +pub mod ifc; /// Shared invite-link contract constants. pub mod invite; /// Buzz kind number registry — custom event type constants. diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 2f306c35d0..63796987f3 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, 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, ifc, moderation, + 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 b6ead54e06..cd309d2e10 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -705,7 +705,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 60); + assert_eq!(migrations.len(), 61); assert_eq!(migrations[58].version, 59); assert!(migrations[58].sql.as_str().contains("buzz.push_enabled")); assert_eq!(migrations[59].version, 60); @@ -713,6 +713,14 @@ mod postgres_tests { .sql .as_str() .contains("ALTER TABLE push_leases DROP COLUMN app_profile")); + assert_eq!(migrations[60].version, 61); + let ifc = migrations[60].sql.as_str(); + assert!(ifc.contains("CREATE TABLE channel_policy_generations")); + assert!(ifc.contains("CREATE TABLE ifc_event_acceptances")); + assert!(ifc.contains("CREATE FUNCTION prune_ifc_acceptances_after_insert")); + assert!(ifc.contains("CREATE TRIGGER trg_channels_ifc_policy_update")); + assert!(ifc.contains("CREATE TRIGGER trg_channel_members_ifc_policy_update")); + assert!(ifc.contains("CREATE TRIGGER trg_ifc_acceptance_prune")); assert_eq!(migrations[55].version, 56); assert!(migrations[55] .sql diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 183585e278..fa46fd121f 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -1916,6 +1916,78 @@ async fn acquire_canvas_coordinate_lock(tx: &mut AdmittedTx, channel_id: Uuid) - Ok(()) } +pub(crate) async fn insert_canvas_head_checked_tx( + tx: &mut AdmittedTx, + event: &Event, + channel_id: Uuid, + precondition: ChannelHeadPrecondition<'_>, +) -> Result<(StoredEvent, ChannelHeadWriteStatus)> { + let kind_i32 = buzz_core::kind::event_kind_i32(event); + if kind_i32 != KIND_CANVAS as i32 { + return Err(DbError::InvalidData(format!( + "insert_canvas_head_checked requires kind {KIND_CANVAS}, got {kind_i32}" + ))); + } + let received_at = Utc::now(); + let incoming_id = event.id.as_bytes(); + let community_id = tx.community(); + + acquire_canvas_coordinate_lock(tx, channel_id).await?; + let head: Option<(Vec, DateTime)> = sqlx::query_as( + "SELECT id, created_at FROM events \ + WHERE community_id = $1 AND kind = $2 AND channel_id = $3 AND deleted_at IS NULL \ + ORDER BY created_at DESC, id ASC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await?; + + if head + .as_ref() + .is_some_and(|(id, _)| id.as_slice() == incoming_id.as_slice()) + { + return Ok(( + StoredEvent::with_received_at(event.clone(), received_at, Some(channel_id), true), + ChannelHeadWriteStatus::Duplicate, + )); + } + + let status = match (precondition, head.as_ref()) { + (ChannelHeadPrecondition::ExpectNoHead, None) => None, + (ChannelHeadPrecondition::ExpectNoHead, Some(_)) => { + Some(ChannelHeadWriteStatus::RevisionMismatch) + } + (ChannelHeadPrecondition::ExpectedHead(_), None) => { + Some(ChannelHeadWriteStatus::RevisionMissing) + } + (ChannelHeadPrecondition::ExpectedHead(expected), Some((id, head_created_at))) => { + if id.as_slice() != expected { + Some(ChannelHeadWriteStatus::RevisionMismatch) + } else if !candidate_supersedes_head(event, incoming_id, *head_created_at, id) { + Some(ChannelHeadWriteStatus::SupersedeFailed) + } else { + None + } + } + }; + if let Some(status) = status { + return Ok(( + StoredEvent::with_received_at(event.clone(), received_at, Some(channel_id), false), + status, + )); + } + + let (stored, was_inserted) = + insert_event_with_thread_metadata_tx(tx, event, Some(channel_id), None).await?; + if !was_inserted { + return Ok((stored, ChannelHeadWriteStatus::RevisionMismatch)); + } + crate::insert_mentions_in_transaction(tx, event, Some(channel_id)).await?; + Ok((stored, ChannelHeadWriteStatus::Inserted)) +} + /// Atomically insert an event and its optional thread metadata. /// /// `insert_event` and `insert_thread_metadata` calls could leave reply counters @@ -2018,90 +2090,20 @@ pub async fn insert_canvas_head_checked( channel_id: Uuid, precondition: ChannelHeadPrecondition<'_>, ) -> Result<(StoredEvent, ChannelHeadWriteStatus)> { - let kind_i32 = buzz_core::kind::event_kind_i32(event); - if kind_i32 != KIND_CANVAS as i32 { - return Err(DbError::InvalidData(format!( - "insert_canvas_head_checked requires kind {KIND_CANVAS}, got {kind_i32}" - ))); - } - let received_at = Utc::now(); - let incoming_id = event.id.as_bytes(); - let mut tx = crate::begin_community_event_write_transaction( pool, community_id, crate::observability::WriterOperation::EventWrite, ) .await?; - acquire_canvas_coordinate_lock(&mut tx, channel_id).await?; - - let head: Option<(Vec, DateTime)> = sqlx::query_as( - "SELECT id, created_at FROM events \ - WHERE community_id = $1 AND kind = $2 AND channel_id = $3 AND deleted_at IS NULL \ - ORDER BY created_at DESC, id ASC LIMIT 1", - ) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(channel_id) - .fetch_optional(tx.conn()) - .await?; - - // Idempotent replay: the incoming event is already the live head. - if head - .as_ref() - .is_some_and(|(id, _)| id.as_slice() == incoming_id.as_slice()) - { + let (stored, status) = + insert_canvas_head_checked_tx(&mut tx, event, channel_id, precondition).await?; + if !matches!(status, ChannelHeadWriteStatus::Inserted) { tx.rollback().await?; - return Ok(( - StoredEvent::with_received_at(event.clone(), received_at, Some(channel_id), true), - ChannelHeadWriteStatus::Duplicate, - )); + return Ok((stored, status)); } - - let status = match (precondition, head.as_ref()) { - (ChannelHeadPrecondition::ExpectNoHead, None) => None, - (ChannelHeadPrecondition::ExpectNoHead, Some(_)) => { - Some(ChannelHeadWriteStatus::RevisionMismatch) - } - (ChannelHeadPrecondition::ExpectedHead(_), None) => { - Some(ChannelHeadWriteStatus::RevisionMissing) - } - (ChannelHeadPrecondition::ExpectedHead(expected), Some((id, head_created_at))) => { - if id.as_slice() != expected { - Some(ChannelHeadWriteStatus::RevisionMismatch) - } else if !candidate_supersedes_head(event, incoming_id, *head_created_at, id) { - Some(ChannelHeadWriteStatus::SupersedeFailed) - } else { - None - } - } - }; - if let Some(status) = status { - tx.rollback().await?; - return Ok(( - StoredEvent::with_received_at(event.clone(), received_at, Some(channel_id), false), - status, - )); - } - - let (stored, was_inserted) = - insert_event_with_thread_metadata_tx(&mut tx, event, Some(channel_id), None).await?; - if !was_inserted { - // The primary-key row already exists. The idempotent-replay branch above - // already returned `Duplicate` for the case where the incoming event is - // the canonical live head. Reaching here means the row exists but is NOT - // the live head (e.g. soft-deleted). Because every kind-40100 mutator - // serializes on this advisory key, no concurrent writer can have changed - // the live head since our read above, so the truthful result is - // `RevisionMismatch` — a false `Duplicate` would acknowledge a write - // that did not become live. - tx.rollback().await?; - return Ok((stored, ChannelHeadWriteStatus::RevisionMismatch)); - } - crate::insert_mentions_in_transaction(&mut tx, event, Some(channel_id)).await?; tx.commit().await?; - - Ok((stored, ChannelHeadWriteStatus::Inserted)) + Ok((stored, status)) } impl Db { diff --git a/crates/buzz-db/src/store/event_follow_up.rs b/crates/buzz-db/src/store/event_follow_up.rs index 5219af6062..fc8a21a966 100644 --- a/crates/buzz-db/src/store/event_follow_up.rs +++ b/crates/buzz-db/src/store/event_follow_up.rs @@ -118,7 +118,7 @@ pub(crate) async fn refresh_channel_ttls( Ok(()) } -async fn refresh_channel_ttl( +pub(crate) async fn refresh_channel_ttl( conn: &mut PgConnection, community: CommunityId, channel: Uuid, diff --git a/crates/buzz-db/src/store/ifc.rs b/crates/buzz-db/src/store/ifc.rs new file mode 100644 index 0000000000..660fc4ee8e --- /dev/null +++ b/crates/buzz-db/src/store/ifc.rs @@ -0,0 +1,1147 @@ +//! Canonical policy cuts and atomic fenced publication. + +use std::collections::BTreeMap; + +use buzz_core::ifc::IfcPublicationFence; +use buzz_core::{CommunityId, StoredEvent}; +use chrono::{DateTime, Utc}; +use nostr::Event; +use sqlx::Row; +use thiserror::Error; +use uuid::Uuid; + +use crate::event::{ + insert_canvas_head_checked_tx, insert_event_with_thread_metadata_tx, ChannelHeadPrecondition, + ChannelHeadWriteStatus, EventQuery, ThreadMetadataParams, +}; +use crate::{Db, DbError}; + +/// Largest canonical active-reader set included in one proof. +pub const IFC_MAX_POLICY_MEMBERS: i64 = 50_000; +/// Largest number of raw resources included in one proof. +pub const IFC_MAX_RESOURCES: i64 = 500; + +/// Canonical channel policy captured under its serialization lock. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct IfcPolicySnapshot { + /// Channel identity. + pub channel_id: Uuid, + /// Monotone policy generation. + pub generation: i64, + /// Canonical channel type. + pub channel_type: String, + /// Canonical visibility. + pub visibility: String, + /// Effective retained-history authorization. + pub history_access: String, + /// Whether publication is archived. + pub archived: bool, + /// Sorted active member public keys. + pub members: Vec, +} + +/// One serialized policy cut and the raw events read under it. +#[derive(Debug)] +pub struct IfcReadCut { + /// Canonical policy. + pub policy: IfcPolicySnapshot, + /// Exact signed resources selected by the supported query. + pub resources: Vec, +} + +/// Result of one fenced publication transaction. +#[derive(Debug)] +pub enum IfcPublishOutcome { + /// This transaction inserted and owns post-commit dispatch. + Inserted(Box), + /// The exact event was accepted by an earlier committed transaction. + HistoricDuplicate, + /// Canvas optimistic-concurrency validation rejected the candidate. + CanvasConflict(ChannelHeadWriteStatus), +} + +/// Stable fenced-read and publication failures. +#[derive(Debug, Error)] +pub enum IfcError { + /// A supplied policy generation is no longer current. + #[error("ifc-policy-changed")] + PolicyChanged, + /// A new publication fence has expired or exceeds its lifetime bound. + #[error("ifc-expired")] + Expired, + /// Canonical policy is missing, inactive, or unsupported. + #[error("ifc-invalid")] + Invalid, + /// The authenticated principal lacks ordinary channel access. + #[error("access denied")] + AccessDenied, + /// Database failure. + #[error(transparent)] + Database(#[from] DbError), +} + +async fn lock_policies( + tx: &mut crate::AdmittedTx, + channels: impl IntoIterator, +) -> Result<(), IfcError> { + let mut channels = channels.into_iter().collect::>(); + channels.sort_unstable(); + channels.dedup(); + for channel in channels { + sqlx::query("SELECT pg_advisory_xact_lock(channel_policy_lock_key($1, $2))") + .bind(tx.community().as_uuid()) + .bind(channel) + .execute(tx.conn()) + .await + .map_err(DbError::from)?; + } + Ok(()) +} + +async fn load_policy( + tx: &mut crate::AdmittedTx, + channel_id: Uuid, +) -> Result { + let row = sqlx::query( + "SELECT p.generation, p.deleted, c.channel_type::text, c.visibility::text, \ + c.history_access, c.archived_at IS NOT NULL AS archived \ + FROM channel_policy_generations p \ + JOIN channels c ON c.community_id = p.community_id AND c.id = p.channel_id \ + WHERE p.community_id = $1 AND p.channel_id = $2 AND c.deleted_at IS NULL", + ) + .bind(tx.community().as_uuid()) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await + .map_err(DbError::from)? + .ok_or(IfcError::Invalid)?; + if row.try_get::("deleted").map_err(DbError::from)? { + return Err(IfcError::Invalid); + } + let history_access: String = row.try_get("history_access").map_err(DbError::from)?; + if history_access != "all_current_members" { + return Err(IfcError::Invalid); + } + let member_rows = sqlx::query( + "SELECT encode(pubkey, 'hex') AS pubkey \ + FROM channel_members \ + WHERE community_id = $1 AND channel_id = $2 AND removed_at IS NULL \ + ORDER BY pubkey LIMIT $3", + ) + .bind(tx.community().as_uuid()) + .bind(channel_id) + .bind(IFC_MAX_POLICY_MEMBERS + 1) + .fetch_all(tx.conn()) + .await + .map_err(DbError::from)?; + if member_rows.len() as i64 > IFC_MAX_POLICY_MEMBERS { + return Err(IfcError::Invalid); + } + let members = member_rows + .into_iter() + .map(|member| member.try_get::("pubkey").map_err(DbError::from)) + .collect::, _>>()?; + Ok(IfcPolicySnapshot { + channel_id, + generation: row.try_get("generation").map_err(DbError::from)?, + channel_type: row.try_get("channel_type").map_err(DbError::from)?, + visibility: row.try_get("visibility").map_err(DbError::from)?, + history_access, + archived: row.try_get("archived").map_err(DbError::from)?, + members, + }) +} + +fn has_ordinary_read(policy: &IfcPolicySnapshot, principal_hex: &str) -> bool { + policy.visibility == "open" + || policy + .members + .binary_search_by(|member| member.as_str().cmp(principal_hex)) + .is_ok() +} + +async fn lock_event_receipt( + tx: &mut crate::AdmittedTx, + event_id: &[u8; 32], +) -> Result<(), IfcError> { + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(format!( + "buzz_ifc_event:{}:{}", + tx.community(), + hex::encode(event_id) + )) + .execute(tx.conn()) + .await + .map_err(DbError::from)?; + Ok(()) +} + +impl Db { + /// Capture a canonical policy and bounded raw history under one lock cut. + pub async fn ifc_read_cut( + &self, + community_id: CommunityId, + channel_id: Uuid, + principal_hex: &str, + expected_generation: Option, + query: &EventQuery, + ) -> Result { + if query.community_id != community_id + || query.channel_id != Some(channel_id) + || query.limit.unwrap_or(IFC_MAX_RESOURCES) > IFC_MAX_RESOURCES + { + return Err(IfcError::Invalid); + } + let mut tx = crate::begin_community_write_transaction( + &self.pool, + community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; + sqlx::query("SET LOCAL statement_timeout = '9s'") + .execute(tx.conn()) + .await + .map_err(DbError::from)?; + sqlx::query("SET LOCAL lock_timeout = '9s'") + .execute(tx.conn()) + .await + .map_err(DbError::from)?; + lock_policies(&mut tx, [channel_id]).await?; + let policy = load_policy(&mut tx, channel_id).await?; + if !has_ordinary_read(&policy, principal_hex) { + return Err(IfcError::AccessDenied); + } + if expected_generation.is_some_and(|generation| generation != policy.generation) { + return Err(IfcError::PolicyChanged); + } + let resources = crate::event::query_events_on(tx.conn(), query).await?; + tx.rollback().await?; + Ok(IfcReadCut { policy, resources }) + } + + /// Return whether an exact fenced event has a committed acceptance receipt. + pub async fn ifc_event_was_accepted( + &self, + community_id: CommunityId, + event_id: &[u8; 32], + ) -> Result { + let row: Option<(bool,)> = sqlx::query_as( + "SELECT TRUE FROM ifc_event_acceptances WHERE community_id = $1 AND event_id = $2", + ) + .bind(community_id.as_uuid()) + .bind(event_id.as_slice()) + .fetch_optional(&self.pool) + .await?; + Ok(row.is_some()) + } + + /// Atomically validate canonical policies, insert an event, and record acceptance. + pub async fn publish_ifc_event( + &self, + community_id: CommunityId, + event: &Event, + fence: &IfcPublicationFence, + principal_hex: &str, + thread_meta: Option>, + canvas_precondition: Option>, + ) -> Result { + if fence.community_id != *community_id.as_uuid() { + return Err(IfcError::Invalid); + } + let mut tx = crate::begin_community_event_write_transaction( + &self.pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + tx.set_push_enqueue(self.push_enqueue.clone()); + lock_event_receipt(&mut tx, event.id.as_bytes()).await?; + let accepted: Option<(bool,)> = sqlx::query_as( + "SELECT TRUE FROM ifc_event_acceptances WHERE community_id = $1 AND event_id = $2", + ) + .bind(community_id.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .fetch_optional(tx.conn()) + .await + .map_err(DbError::from)?; + if accepted.is_some() { + tx.rollback().await?; + return Ok(IfcPublishOutcome::HistoricDuplicate); + } + + let now: i64 = sqlx::query_scalar("SELECT EXTRACT(EPOCH FROM clock_timestamp())::BIGINT") + .fetch_one(tx.conn()) + .await + .map_err(DbError::from)?; + if fence.expires_at <= now || fence.expires_at > now + buzz_core::ifc::IFC_MAX_TTL_SECONDS { + return Err(IfcError::Expired); + } + + // Match the established event-write order before taking policy locks: + // canvas coordinate, channel TTL key, ephemeral channel row, policy. + // The normal insert and commit paths reacquire the same transaction + // locks, so this keeps their existing behavior without an inversion + // against channel metadata writers. + crate::event::acquire_canvas_event_write_lock_if_needed( + &mut tx, + event, + Some(fence.destination.channel_id), + ) + .await?; + if super::event_follow_up::ttl_refresh_channel( + Some(fence.destination.channel_id), + i32::from(event.kind.as_u16()), + ) + .is_some() + { + super::event_follow_up::refresh_channel_ttl( + tx.conn(), + community_id, + fence.destination.channel_id, + ) + .await?; + } + + let mut required = BTreeMap::new(); + for source in &fence.sources { + required.insert(source.channel_id, source.generation); + } + match required.insert(fence.destination.channel_id, fence.destination.generation) { + Some(source_generation) if source_generation != fence.destination.generation => { + return Err(IfcError::PolicyChanged); + } + _ => {} + } + lock_policies(&mut tx, required.keys().copied()).await?; + for (channel_id, generation) in required { + let policy = load_policy(&mut tx, channel_id).await?; + if policy.generation != generation { + return Err(IfcError::PolicyChanged); + } + if policy.archived || !has_ordinary_read(&policy, principal_hex) { + return Err(IfcError::AccessDenied); + } + } + + let commit_check_now: i64 = + sqlx::query_scalar("SELECT EXTRACT(EPOCH FROM clock_timestamp())::BIGINT") + .fetch_one(tx.conn()) + .await + .map_err(DbError::from)?; + if fence.expires_at <= commit_check_now { + return Err(IfcError::Expired); + } + + let outcome = if let Some(precondition) = canvas_precondition { + let (stored, status) = insert_canvas_head_checked_tx( + &mut tx, + event, + fence.destination.channel_id, + precondition, + ) + .await?; + if status != ChannelHeadWriteStatus::Inserted { + tx.rollback().await?; + return Ok(IfcPublishOutcome::CanvasConflict(status)); + } + IfcPublishOutcome::Inserted(Box::new(stored)) + } else { + let (stored, inserted) = insert_event_with_thread_metadata_tx( + &mut tx, + event, + Some(fence.destination.channel_id), + thread_meta, + ) + .await?; + if !inserted { + return Err(IfcError::Invalid); + } + crate::insert_mentions_in_transaction( + &mut tx, + event, + Some(fence.destination.channel_id), + ) + .await?; + IfcPublishOutcome::Inserted(Box::new(stored)) + }; + + let expires_at = + DateTime::::from_timestamp(fence.expires_at, 0).ok_or(IfcError::Invalid)?; + sqlx::query( + "INSERT INTO ifc_event_acceptances \ + (community_id, event_id, event_kind, fence_expires_at) VALUES ($1, $2, $3, $4)", + ) + .bind(community_id.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .bind(i32::from(event.kind.as_u16())) + .bind(expires_at) + .execute(tx.conn()) + .await + .map_err(DbError::from)?; + match tx.commit().await { + Ok(()) => {} + Err(DbError::Sqlx(sqlx::Error::Database(error))) + if error.constraint() == Some("ifc_acceptance_unexpired") => + { + return Err(IfcError::Expired); + } + Err(error) => return Err(IfcError::Database(error)), + } + Ok(outcome) + } + + /// Verify that the required generation catalog and database triggers exist. + pub async fn verify_ifc_policy_catalog(&self) -> Result<(), DbError> { + let rows: Vec<(String,)> = sqlx::query_as( + "SELECT tgname FROM pg_trigger \ + WHERE NOT tgisinternal AND tgenabled = 'O' AND tgname = ANY($1::text[]) ORDER BY tgname", + ) + .bind([ + "trg_channel_members_ifc_policy_delete", + "trg_channel_members_ifc_policy_identity", + "trg_channel_members_ifc_policy_insert", + "trg_channel_members_ifc_policy_update", + "trg_channels_ifc_policy_delete", + "trg_channels_ifc_policy_identity", + "trg_channels_ifc_policy_insert", + "trg_channels_ifc_policy_update", + ]) + .fetch_all(&self.pool) + .await?; + let catalog_ready: bool = sqlx::query_scalar( + "SELECT to_regclass('channel_policy_generations') IS NOT NULL \ + AND to_regclass('ifc_event_acceptances') IS NOT NULL \ + AND EXISTS (SELECT 1 FROM information_schema.columns \ + WHERE table_schema='public' AND table_name='channels' \ + AND column_name='history_access') \ + AND to_regprocedure('channel_policy_lock_key(uuid,uuid)') IS NOT NULL \ + AND to_regprocedure('initialize_channel_policy_generation()') IS NOT NULL \ + AND to_regprocedure('advance_channel_policy_for_channel_update()') IS NOT NULL \ + AND to_regprocedure('advance_channel_policy_for_member_update()') IS NOT NULL \ + AND to_regprocedure('guard_channel_policy_generation_row()') IS NOT NULL \ + AND to_regprocedure('guard_ifc_acceptance_row()') IS NOT NULL \ + AND to_regprocedure('enforce_ifc_acceptance_expiry()') IS NOT NULL \ + AND to_regprocedure('prune_ifc_acceptances_after_insert()') IS NOT NULL \ + AND EXISTS (SELECT 1 FROM pg_trigger \ + WHERE tgname='trg_ifc_acceptance_unexpired' AND tgenabled='O') \ + AND EXISTS (SELECT 1 FROM pg_trigger \ + WHERE tgname='trg_ifc_acceptance_prune' AND tgenabled='O') \ + AND EXISTS (SELECT 1 FROM pg_trigger \ + WHERE tgname='trg_channel_policy_generation_guard' AND tgenabled='O') \ + AND EXISTS (SELECT 1 FROM pg_trigger \ + WHERE tgname='trg_ifc_acceptance_guard' AND tgenabled='O')", + ) + .fetch_one(&self.pool) + .await?; + if rows.len() != 8 || !catalog_ready { + return Err(DbError::InvalidData( + "IFC policy catalog is incomplete".to_owned(), + )); + } + Ok(()) + } +} + +#[cfg(test)] +mod postgres_tests { + use super::*; + use crate::{migration, DbConfig}; + use buzz_core::ifc::parse_publication_fence; + use buzz_core::kind::KIND_STREAM_MESSAGE; + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + use sqlx::PgPool; + + async fn connect() -> Db { + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .expect("Postgres wrapper must provide an isolated database URL"); + let migration_schema = std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref() == Ok("migration"); + let db = Db::new(&DbConfig { + database_url, + ..DbConfig::default() + }) + .await + .expect("connect isolated database"); + if migration_schema { + migration::run_migrations(db.pool()) + .await + .expect("apply migrations"); + } + db.verify_ifc_policy_catalog() + .await + .expect("verify IFC policy catalog"); + db + } + + async fn fixture(pool: &PgPool) -> (CommunityId, Uuid, Keys) { + let community_uuid = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_uuid) + .bind(format!("ifc-{}.example", community_uuid.simple())) + .execute(pool) + .await + .expect("insert community"); + let community = CommunityId::from_uuid(community_uuid); + let keys = Keys::generate(); + let channel = crate::channel::create_channel( + pool, + community, + "ifc-test", + crate::channel::ChannelType::Stream, + crate::channel::ChannelVisibility::Private, + None, + &keys.public_key().to_bytes(), + None, + ) + .await + .expect("create channel"); + (community, channel.id, keys) + } + + async fn generation(pool: &PgPool, community: CommunityId, channel: Uuid) -> i64 { + sqlx::query_scalar( + "SELECT generation FROM channel_policy_generations WHERE community_id=$1 AND channel_id=$2", + ) + .bind(community.as_uuid()) + .bind(channel) + .fetch_one(pool) + .await + .expect("read generation") + } + + fn fenced_event( + keys: &Keys, + community: CommunityId, + channel: Uuid, + generation: i64, + expires_at: i64, + content: &str, + ) -> Event { + let channel_text = channel.to_string(); + let community_text = community.to_string(); + let generation_text = generation.to_string(); + let expiry_text = expires_at.to_string(); + EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), content) + .tags([ + Tag::parse(["h", channel_text.as_str()]).expect("h tag"), + Tag::parse(["ifc", "1", community_text.as_str(), expiry_text.as_str()]) + .expect("ifc tag"), + Tag::parse([ + "ifc-source", + channel_text.as_str(), + generation_text.as_str(), + ]) + .expect("source tag"), + Tag::parse([ + "ifc-destination", + channel_text.as_str(), + generation_text.as_str(), + ]) + .expect("destination tag"), + ]) + .sign_with_keys(keys) + .expect("sign event") + } + + fn fenced_event_between( + keys: &Keys, + community: CommunityId, + source: (Uuid, i64), + destination: (Uuid, i64), + content: &str, + ) -> Event { + let source_channel = source.0.to_string(); + let source_generation = source.1.to_string(); + let destination_channel = destination.0.to_string(); + let destination_generation = destination.1.to_string(); + let community_text = community.to_string(); + let expiry = (Utc::now().timestamp() + 120).to_string(); + EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), content) + .tags([ + Tag::parse(["h", destination_channel.as_str()]).expect("h tag"), + Tag::parse(["ifc", "1", community_text.as_str(), expiry.as_str()]) + .expect("ifc tag"), + Tag::parse([ + "ifc-source", + source_channel.as_str(), + source_generation.as_str(), + ]) + .expect("source tag"), + Tag::parse([ + "ifc-destination", + destination_channel.as_str(), + destination_generation.as_str(), + ]) + .expect("destination tag"), + ]) + .sign_with_keys(keys) + .expect("sign event") + } + + #[tokio::test] + #[ignore = "requires isolated Postgres"] + async fn policy_generation_tracks_semantic_changes_and_leave_rejoin_aba() { + let db = connect().await; + let (community, channel, keys) = fixture(db.pool()).await; + assert_eq!(generation(db.pool(), community, channel).await, 1); + + sqlx::query( + "UPDATE channels SET name=name, topic='new topic' WHERE community_id=$1 AND id=$2", + ) + .bind(community.as_uuid()) + .bind(channel) + .execute(db.pool()) + .await + .expect("metadata update"); + sqlx::query("UPDATE channel_members SET role=role WHERE community_id=$1 AND channel_id=$2") + .bind(community.as_uuid()) + .bind(channel) + .execute(db.pool()) + .await + .expect("no-op member update"); + assert_eq!(generation(db.pool(), community, channel).await, 1); + + sqlx::query("UPDATE channel_members SET removed_at=NOW() WHERE community_id=$1 AND channel_id=$2 AND pubkey=$3") + .bind(community.as_uuid()).bind(channel).bind(keys.public_key().to_bytes().as_slice()).execute(db.pool()).await.expect("leave"); + assert_eq!(generation(db.pool(), community, channel).await, 2); + sqlx::query("UPDATE channel_members SET removed_at=NULL WHERE community_id=$1 AND channel_id=$2 AND pubkey=$3") + .bind(community.as_uuid()).bind(channel).bind(keys.public_key().to_bytes().as_slice()).execute(db.pool()).await.expect("rejoin"); + assert_eq!(generation(db.pool(), community, channel).await, 3); + sqlx::query("UPDATE channels SET visibility='open' WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(channel) + .execute(db.pool()) + .await + .expect("visibility update"); + assert_eq!(generation(db.pool(), community, channel).await, 4); + assert!( + sqlx::query("UPDATE channels SET id=$3 WHERE community_id=$1 AND id=$2",) + .bind(community.as_uuid()) + .bind(channel) + .bind(Uuid::new_v4()) + .execute(db.pool()) + .await + .is_err() + ); + sqlx::query( + "UPDATE channel_members SET role='admin' \ + WHERE community_id=$1 AND channel_id=$2 AND pubkey=$3", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(keys.public_key().to_bytes().as_slice()) + .execute(db.pool()) + .await + .expect("change active authority"); + assert_eq!(generation(db.pool(), community, channel).await, 5); + sqlx::query("UPDATE channels SET archived_at=NOW() WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(channel) + .execute(db.pool()) + .await + .expect("archive channel"); + assert_eq!(generation(db.pool(), community, channel).await, 6); + sqlx::query("UPDATE channels SET archived_at=NULL WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(channel) + .execute(db.pool()) + .await + .expect("unarchive channel"); + assert_eq!(generation(db.pool(), community, channel).await, 7); + assert!(sqlx::query( + "UPDATE channels SET history_access='future_policy' \ + WHERE community_id=$1 AND id=$2", + ) + .bind(community.as_uuid()) + .bind(channel) + .execute(db.pool()) + .await + .is_err()); + assert_eq!(generation(db.pool(), community, channel).await, 7); + sqlx::query("DELETE FROM channels WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(channel) + .execute(db.pool()) + .await + .expect("hard delete channel"); + let tombstone: (i64, bool) = sqlx::query_as( + "SELECT generation, deleted FROM channel_policy_generations \ + WHERE community_id=$1 AND channel_id=$2", + ) + .bind(community.as_uuid()) + .bind(channel) + .fetch_one(db.pool()) + .await + .expect("policy tombstone"); + assert!(tombstone.0 > 7 && tombstone.1); + assert!(sqlx::query( + "INSERT INTO channels (id, community_id, name, created_by) VALUES ($1,$2,$3,$4)", + ) + .bind(channel) + .bind(community.as_uuid()) + .bind("reused-channel") + .bind(keys.public_key().to_bytes().as_slice()) + .execute(db.pool()) + .await + .is_err()); + assert!(sqlx::query( + "UPDATE channel_policy_generations SET generation=1 \ + WHERE community_id=$1 AND channel_id=$2", + ) + .bind(community.as_uuid()) + .bind(channel) + .execute(db.pool()) + .await + .is_err()); + } + + #[tokio::test] + #[ignore = "requires isolated Postgres"] + async fn receipt_retention_prunes_old_entries_only_through_trigger() { + let db = connect().await; + let (community, _, _) = fixture(db.pool()).await; + let old_event_id = [7_u8; 32]; + let mut cleanup = crate::begin_community_write_transaction( + db.pool(), + community, + crate::observability::WriterOperation::Maintenance, + ) + .await + .expect("begin receipt cleanup probe"); + sqlx::query( + "INSERT INTO ifc_event_acceptances \ + (community_id, event_id, event_kind, fence_expires_at, accepted_at) \ + VALUES ($1, $2, 9, NOW() + INTERVAL '1 minute', NOW() - INTERVAL '25 hours')", + ) + .bind(community.as_uuid()) + .bind(old_event_id.as_slice()) + .execute(cleanup.conn()) + .await + .expect("insert old receipt"); + cleanup + .commit() + .await + .expect("commit receipt cleanup probe"); + assert!(!db + .ifc_event_was_accepted(community, &old_event_id) + .await + .expect("query pruned receipt")); + } + + #[tokio::test] + #[ignore = "requires isolated Postgres"] + async fn exact_retry_keeps_historic_acceptance_after_policy_change() { + let db = connect().await; + let (community, channel, keys) = fixture(db.pool()).await; + let event = fenced_event( + &keys, + community, + channel, + 1, + Utc::now().timestamp() + 120, + "first", + ); + let fence = parse_publication_fence(&event) + .expect("parse") + .expect("fence"); + let first = db + .publish_ifc_event( + community, + &event, + &fence, + &keys.public_key().to_hex(), + None, + None, + ) + .await; + assert!( + matches!(first, Ok(IfcPublishOutcome::Inserted(_))), + "{first:?}" + ); + sqlx::query("UPDATE channels SET visibility='open' WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(channel) + .execute(db.pool()) + .await + .expect("policy change"); + assert!(matches!( + db.publish_ifc_event( + community, + &event, + &fence, + &keys.public_key().to_hex(), + None, + None + ) + .await, + Ok(IfcPublishOutcome::HistoricDuplicate) + )); + assert!(sqlx::query( + "DELETE FROM ifc_event_acceptances WHERE community_id=$1 AND event_id=$2", + ) + .bind(community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .execute(db.pool()) + .await + .is_err()); + let stale = fenced_event( + &keys, + community, + channel, + 1, + Utc::now().timestamp() + 120, + "stale", + ); + let stale_fence = parse_publication_fence(&stale) + .expect("parse") + .expect("fence"); + assert!(matches!( + db.publish_ifc_event( + community, + &stale, + &stale_fence, + &keys.public_key().to_hex(), + None, + None + ) + .await, + Err(IfcError::PolicyChanged) + )); + let expired = fenced_event( + &keys, + community, + channel, + 2, + Utc::now().timestamp() - 1, + "expired", + ); + let expired_fence = parse_publication_fence(&expired) + .expect("parse") + .expect("fence"); + assert!(matches!( + db.publish_ifc_event( + community, + &expired, + &expired_fence, + &keys.public_key().to_hex(), + None, + None + ) + .await, + Err(IfcError::Expired) + )); + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1") + .bind(community.as_uuid()) + .fetch_one(db.pool()) + .await + .expect("count events"); + assert_eq!(count, 1); + } + + #[tokio::test] + #[ignore = "requires isolated Postgres"] + async fn concurrent_exact_duplicates_have_one_insert_owner() { + let db = connect().await; + let (community, channel, keys) = fixture(db.pool()).await; + let event = fenced_event( + &keys, + community, + channel, + 1, + Utc::now().timestamp() + 120, + "race", + ); + let fence = parse_publication_fence(&event) + .expect("parse") + .expect("fence"); + let principal = keys.public_key().to_hex(); + let (left, right) = tokio::join!( + db.publish_ifc_event(community, &event, &fence, &principal, None, None), + db.publish_ifc_event(community, &event, &fence, &principal, None, None), + ); + let outcomes = [left, right]; + let inserted = outcomes + .iter() + .filter(|outcome| matches!(outcome, Ok(IfcPublishOutcome::Inserted(_)))) + .count(); + assert_eq!(inserted, 1, "{outcomes:?}"); + let count: i64 = + sqlx::query_scalar("SELECT count(*) FROM ifc_event_acceptances WHERE community_id=$1") + .bind(community.as_uuid()) + .fetch_one(db.pool()) + .await + .expect("count receipts"); + assert_eq!(count, 1); + } + + #[tokio::test] + #[ignore = "requires isolated Postgres"] + async fn current_member_reads_legacy_history_at_current_generation() { + let db = connect().await; + let (community, channel, keys) = fixture(db.pool()).await; + let channel_text = channel.to_string(); + let legacy = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "legacy") + .tags([Tag::parse(["h", channel_text.as_str()]).expect("h tag")]) + .sign_with_keys(&keys) + .expect("sign legacy event"); + db.insert_event_with_thread_metadata(community, &legacy, Some(channel), None) + .await + .expect("insert legacy event"); + let added = Keys::generate(); + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) \ + VALUES ($1, $2, $3, 'member', $4)", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(added.public_key().to_bytes().as_slice()) + .bind(keys.public_key().to_bytes().as_slice()) + .execute(db.pool()) + .await + .expect("add current reader"); + let mut query = EventQuery::for_community(community); + query.channel_id = Some(channel); + query.kinds = Some(vec![KIND_STREAM_MESSAGE as i32]); + query.limit = Some(10); + let cut = db + .ifc_read_cut( + community, + channel, + &added.public_key().to_hex(), + Some(2), + &query, + ) + .await + .expect("current full-history grant"); + assert_eq!(cut.policy.generation, 2); + assert_eq!(cut.resources.len(), 1); + assert_eq!(cut.resources[0].event.id, legacy.id); + let outsider = Keys::generate(); + assert!(matches!( + db.ifc_read_cut( + community, + channel, + &outsider.public_key().to_hex(), + Some(2), + &query + ) + .await, + Err(IfcError::AccessDenied) + )); + } + + #[tokio::test] + #[ignore = "requires isolated Postgres"] + async fn source_and_destination_generations_are_checked_independently() { + let db = connect().await; + let (community, source, keys) = fixture(db.pool()).await; + let destination = crate::channel::create_channel( + db.pool(), + community, + "ifc-destination", + crate::channel::ChannelType::Stream, + crate::channel::ChannelVisibility::Private, + None, + &keys.public_key().to_bytes(), + None, + ) + .await + .expect("create destination") + .id; + sqlx::query("UPDATE channels SET visibility='open' WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(source) + .execute(db.pool()) + .await + .expect("advance source"); + let stale_source = fenced_event_between( + &keys, + community, + (source, 1), + (destination, 1), + "stale source", + ); + let fence = parse_publication_fence(&stale_source) + .expect("parse") + .expect("fence"); + assert!(matches!( + db.publish_ifc_event( + community, + &stale_source, + &fence, + &keys.public_key().to_hex(), + None, + None + ) + .await, + Err(IfcError::PolicyChanged) + )); + sqlx::query("UPDATE channels SET visibility='open' WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(destination) + .execute(db.pool()) + .await + .expect("advance destination"); + let stale_destination = fenced_event_between( + &keys, + community, + (source, 2), + (destination, 1), + "stale destination", + ); + let fence = parse_publication_fence(&stale_destination) + .expect("parse") + .expect("fence"); + assert!(matches!( + db.publish_ifc_event( + community, + &stale_destination, + &fence, + &keys.public_key().to_hex(), + None, + None + ) + .await, + Err(IfcError::PolicyChanged) + )); + } + + #[tokio::test] + #[ignore = "requires isolated Postgres"] + async fn fenced_canvas_preserves_head_preconditions() { + let db = connect().await; + let (community, channel, keys) = fixture(db.pool()).await; + let channel_text = channel.to_string(); + let community_text = community.to_string(); + let expiry = (Utc::now().timestamp() + 120).to_string(); + let tags = || { + [ + Tag::parse(["h", channel_text.as_str()]).expect("h tag"), + Tag::parse(["ifc", "1", community_text.as_str(), expiry.as_str()]) + .expect("ifc tag"), + Tag::parse(["ifc-source", channel_text.as_str(), "1"]).expect("source tag"), + Tag::parse(["ifc-destination", channel_text.as_str(), "1"]) + .expect("destination tag"), + ] + }; + let now = Utc::now().timestamp() as u64; + let first = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_CANVAS as u16), "first") + .tags(tags()) + .custom_created_at(Timestamp::from(now)) + .sign_with_keys(&keys) + .expect("sign first canvas"); + let first_fence = parse_publication_fence(&first) + .expect("parse") + .expect("fence"); + assert!(matches!( + db.publish_ifc_event( + community, + &first, + &first_fence, + &keys.public_key().to_hex(), + None, + Some(ChannelHeadPrecondition::ExpectNoHead) + ) + .await, + Ok(IfcPublishOutcome::Inserted(_)) + )); + let second = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_CANVAS as u16), "second") + .tags(tags()) + .custom_created_at(Timestamp::from(now + 1)) + .sign_with_keys(&keys) + .expect("sign second canvas"); + let second_fence = parse_publication_fence(&second) + .expect("parse") + .expect("fence"); + assert!(matches!( + db.publish_ifc_event( + community, + &second, + &second_fence, + &keys.public_key().to_hex(), + None, + Some(ChannelHeadPrecondition::ExpectNoHead) + ) + .await, + Ok(IfcPublishOutcome::CanvasConflict( + ChannelHeadWriteStatus::RevisionMismatch + )) + )); + assert!(matches!( + db.publish_ifc_event( + community, + &second, + &second_fence, + &keys.public_key().to_hex(), + None, + Some(ChannelHeadPrecondition::ExpectedHead(first.id.as_bytes())) + ) + .await, + Ok(IfcPublishOutcome::Inserted(_)) + )); + } + + #[tokio::test] + #[ignore = "requires isolated Postgres"] + async fn membership_commit_serializes_read_and_fenced_write() { + let db = connect().await; + let (community, channel, keys) = fixture(db.pool()).await; + let event = fenced_event( + &keys, + community, + channel, + 1, + Utc::now().timestamp() + 120, + "racing write", + ); + let fence = parse_publication_fence(&event) + .expect("parse") + .expect("fence"); + let mut mutation = db.pool().begin().await.expect("begin membership change"); + sqlx::query( + "UPDATE channel_members SET removed_at=NOW() \ + WHERE community_id=$1 AND channel_id=$2 AND pubkey=$3", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(keys.public_key().to_bytes().as_slice()) + .execute(&mut *mutation) + .await + .expect("hold canonical membership change"); + + let mut query = EventQuery::for_community(community); + query.channel_id = Some(channel); + query.kinds = Some(vec![KIND_STREAM_MESSAGE as i32]); + query.limit = Some(10); + let read_db = db.clone(); + let reader = keys.public_key().to_hex(); + let read = tokio::spawn(async move { + read_db + .ifc_read_cut(community, channel, &reader, None, &query) + .await + }); + let write_db = db.clone(); + let writer = keys.public_key().to_hex(); + let write = tokio::spawn(async move { + write_db + .publish_ifc_event(community, &event, &fence, &writer, None, None) + .await + }); + tokio::task::yield_now().await; + mutation.commit().await.expect("commit membership change"); + + assert!(matches!( + read.await.expect("read task"), + Err(IfcError::AccessDenied) + )); + assert!(matches!( + write.await.expect("write task"), + Err(IfcError::PolicyChanged) + )); + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1") + .bind(community.as_uuid()) + .fetch_one(db.pool()) + .await + .expect("count events"); + assert_eq!(count, 0); + } +} diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index efa46f1748..e74736b664 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -30,6 +30,8 @@ mod event_follow_up_postgres_tests; pub mod feed; /// Git repository name registry (NIP-34 kind:30617). pub mod git_repo; +/// Canonical remote-agent information-flow policy and fenced writes. +pub mod ifc; /// Community moderation: reports, bans/timeouts, audit actions. pub mod moderation; /// Deployment-global operator-listener mention registrations and delivery queues. diff --git a/crates/buzz-relay/examples/generate_ifc_fixture.rs b/crates/buzz-relay/examples/generate_ifc_fixture.rs new file mode 100644 index 0000000000..afea3316ee --- /dev/null +++ b/crates/buzz-relay/examples/generate_ifc_fixture.rs @@ -0,0 +1,84 @@ +use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; +use serde_json::json; +use sha2::{Digest, Sha256}; + +fn main() -> anyhow::Result<()> { + let relay_secret = "0000000000000000000000000000000000000000000000000000000000000001"; + let requester_secret = "0000000000000000000000000000000000000000000000000000000000000002"; + let broker_secret = "0000000000000000000000000000000000000000000000000000000000000003"; + let relay = Keys::parse(relay_secret)?; + let requester = Keys::parse(requester_secret)?; + let broker = Keys::parse(broker_secret)?; + let community = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"; + let source = "11111111-1111-4111-8111-111111111111"; + let destination = "22222222-2222-4222-8222-222222222222"; + let nonce = "0123456789abcdef0123456789abcdef"; + let request_body = format!( + "[{{\"kinds\":[9,40002,40100],\"#h\":[\"{source}\"],\"limit\":13,\"ifc\":{{\"version\":1,\"nonce\":\"{nonce}\",\"expected_generation\":\"17\"}}}}]" + ); + let resource = EventBuilder::new(Kind::Custom(40002), "fixture source message") + .tags([Tag::parse(["h", source])?]) + .custom_created_at(Timestamp::from(1_791_999_700u64)) + .sign_with_keys(&requester)?; + let digest = hex::encode(Sha256::digest(request_body.as_bytes())); + let mut members = [ + requester.public_key().to_hex(), + broker.public_key().to_hex(), + ]; + members.sort(); + let content = json!({ + "version": 1, + "community_id": community, + "requester": requester.public_key().to_hex(), + "nonce": nonce, + "request_sha256": digest, + "policy": { + "channel_id": source, + "generation": "17", + "channel_type": "dm", + "visibility": "private", + "history_access": "all_current_members", + "archived": false, + "members": members + }, + "resources": [{ + "event": resource, + "effective_generation": "17", + "origin_generation": null + }] + }); + let attestation = EventBuilder::new(Kind::Custom(45900), serde_json::to_string(&content)?) + .tags([ + Tag::parse(["p", requester.public_key().to_hex().as_str()])?, + Tag::parse(["h", source])?, + ]) + .custom_created_at(Timestamp::from(1_791_999_800u64)) + .sign_with_keys(&relay)?; + let publication = EventBuilder::new(Kind::Custom(40002), "fixture remote-agent output") + .tags([ + Tag::parse(["h", destination])?, + Tag::parse(["ifc", "1", community, "1792000000"])?, + Tag::parse(["ifc-source", source, "17"])?, + Tag::parse(["ifc-destination", destination, "23"])?, + ]) + .custom_created_at(Timestamp::from(1_791_999_900u64)) + .sign_with_keys(&broker)?; + let fixture = json!({ + "contract": "buzz-ifc-v1", + "test_keys": { + "relay_secret_hex": relay_secret, + "relay_pubkey": relay.public_key().to_hex(), + "requester_secret_hex": requester_secret, + "requester_pubkey": requester.public_key().to_hex(), + "broker_secret_hex": broker_secret, + "broker_pubkey": broker.public_key().to_hex() + }, + "request_body_utf8": request_body, + "request_sha256": digest, + "response": {"ifc": 1, "attestation": attestation}, + "channel_publication_event": publication, + "owner_memory": {"enabled": false, "reason": "awaiting the backend's existing memory address grammar"} + }); + println!("{}", serde_json::to_string_pretty(&fixture)?); + Ok(()) +} diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index be6743599b..d7507fd535 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1370,6 +1370,12 @@ async fn query_events_authed( ) .await?; + if let Some(result) = crate::ifc::query(state, tenant, &pubkey, body).await { + return result + .map(Json) + .map_err(|(status, message)| api_error(status, &message)); + } + // Two-pass parse: preserve raw JSON for custom extension fields (before_id, // depth_limit, feed_types) that nostr::Filter silently drops. let raw_filters: Vec = serde_json::from_slice(body) diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 0324d6f2cc..0a419afb43 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -10,6 +10,7 @@ use tracing::{debug, error, info, warn}; use uuid::Uuid; use buzz_auth::Scope; +use buzz_core::ifc::{has_ifc_tag, parse_publication_fence}; use buzz_core::kind::{ event_kind_u32, is_identity_archive_request_kind, is_parameterized_replaceable, is_relay_admin_kind, KIND_AGENT_ENGRAM, KIND_AGENT_PROFILE, KIND_AGENT_TURN_METRIC, @@ -2423,6 +2424,10 @@ async fn ingest_event_inner( ))); } + if has_ifc_tag(&event) && parse_publication_fence(&event).is_err() { + return Err(IngestError::Rejected("ifc-invalid".into())); + } + if buzz_core::kind::is_relay_only_kind(kind_u32) { return Err(IngestError::Rejected("restricted: relay-only kind".into())); } @@ -2448,6 +2453,41 @@ async fn ingest_event_inner( } let event = std::sync::Arc::try_unwrap(event).unwrap_or_else(|arc| (*arc).clone()); + let is_gift_wrap = kind_u32 == KIND_GIFT_WRAP; + if event.pubkey != *auth.pubkey() && !is_gift_wrap { + return Err(IngestError::AuthFailed( + "invalid: event pubkey does not match authenticated identity".into(), + )); + } + let ifc_fence = + parse_publication_fence(&event).map_err(|_| IngestError::Rejected("ifc-invalid".into()))?; + if let Some(fence) = &ifc_fence { + if fence.community_id != *tenant.community().as_uuid() { + return Err(IngestError::Rejected("ifc-invalid".into())); + } + if state + .db + .ifc_event_was_accepted(tenant.community(), event.id.as_bytes()) + .await + .map_err(|error| IngestError::Internal(format!("error: {error}")))? + { + emit( + tracer, + TraceAction::WriteDuplicate { + msg_id: msg_id_label(event.id.as_bytes()), + channel: channel_label(fence.destination.channel_id), + claimed_community: claimed_community_from_event(&event), + }, + state_for_request(tenant, auth.pubkey()), + ); + return Ok(IngestResult { + event_id: event_id_hex, + accepted: true, + message: "duplicate:".into(), + }); + } + } + if kind_u32 == buzz_core::kind::KIND_ARTIFACT && event.pubkey == *auth.pubkey() && state @@ -2501,13 +2541,6 @@ async fn ingest_event_inner( ))); } - let is_gift_wrap = kind_u32 == KIND_GIFT_WRAP; - if event.pubkey != *auth.pubkey() && !is_gift_wrap { - return Err(IngestError::AuthFailed( - "invalid: event pubkey does not match authenticated identity".into(), - )); - } - let required = match required_scope_for_kind(kind_u32, &event) { Ok(scope) => scope, Err(msg) => return Err(IngestError::Rejected(msg.into())), @@ -2690,6 +2723,12 @@ async fn ingest_event_inner( IngestError::AuthFailed(e) } })?; + if let Some(fence) = &ifc_fence { + for source in &fence.sources { + check_token_channel_access(&auth, source.channel_id) + .map_err(IngestError::AuthFailed)?; + } + } } else if deletion_targets_event && auth.channel_ids().is_some() { return Err(deletion_denied()); } else if auth.channel_ids().is_some() { @@ -2741,7 +2780,8 @@ async fn ingest_event_inner( // member/open gate here lets the owning human act on private agent channels // without being a member (OQ1 decision; see validate_edit_ownership / // validate_admin_event for per-kind enforcement). - let skip_membership = kind_u32 == KIND_NIP29_JOIN_REQUEST + let skip_membership = ifc_fence.is_some() + || kind_u32 == KIND_NIP29_JOIN_REQUEST || kind_u32 == KIND_NIP29_CREATE_GROUP || kind_u32 == KIND_STREAM_MESSAGE_EDIT || kind_u32 == KIND_NIP29_EDIT_METADATA @@ -3398,7 +3438,76 @@ async fn ingest_event_inner( }; let workflow_deletion = crate::handlers::side_effects::is_workflow_deletion(&event); - let (stored_event, was_inserted) = if workflow_deletion { + let (stored_event, was_inserted) = if let Some(fence) = &ifc_fence { + let thread_params = thread_meta.as_ref().map(|metadata| metadata.as_params()); + let canvas_precondition = canvas_revision_spec.as_ref().map(|spec| match spec { + CanvasRevisionSpec::NoHead => buzz_db::ChannelHeadPrecondition::ExpectNoHead, + CanvasRevisionSpec::Head(id) => buzz_db::ChannelHeadPrecondition::ExpectedHead(id), + }); + match state + .db + .publish_ifc_event( + tenant.community(), + &event, + fence, + &auth.pubkey().to_hex(), + thread_params, + canvas_precondition, + ) + .await + { + Ok(buzz_db::ifc::IfcPublishOutcome::Inserted(stored)) => (*stored, true), + Ok(buzz_db::ifc::IfcPublishOutcome::HistoricDuplicate) => { + emit( + tracer, + TraceAction::WriteDuplicate { + msg_id: msg_id_label(event.id.as_bytes()), + channel: channel_label(fence.destination.channel_id), + claimed_community: claimed_community_from_event(&event), + }, + state_for_request(tenant, auth.pubkey()), + ); + return Ok(IngestResult { + event_id: event_id_hex, + accepted: true, + message: "duplicate:".into(), + }); + } + Ok(buzz_db::ifc::IfcPublishOutcome::CanvasConflict(status)) => { + let message = match status { + buzz_db::ChannelHeadWriteStatus::RevisionMissing => { + "conflict: canvas revision does not exist" + } + buzz_db::ChannelHeadWriteStatus::RevisionMismatch => { + "conflict: canvas changed since it was loaded" + } + buzz_db::ChannelHeadWriteStatus::SupersedeFailed => { + "conflict: canvas write does not supersede the current head" + } + buzz_db::ChannelHeadWriteStatus::Inserted + | buzz_db::ChannelHeadWriteStatus::Duplicate => "ifc-invalid", + }; + return Err(IngestError::CanvasConflict(message.into())); + } + Err(buzz_db::ifc::IfcError::PolicyChanged) => { + return Err(IngestError::CanvasConflict("ifc-policy-changed".into())); + } + Err(buzz_db::ifc::IfcError::Expired) => { + return Err(IngestError::Rejected("ifc-expired".into())); + } + Err(buzz_db::ifc::IfcError::Invalid) => { + return Err(IngestError::Rejected("ifc-invalid".into())); + } + Err(buzz_db::ifc::IfcError::AccessDenied) => { + return Err(IngestError::AuthFailed( + "restricted: channel access denied".into(), + )); + } + Err(buzz_db::ifc::IfcError::Database(error)) => { + return Err(IngestError::Internal(format!("error: {error}"))); + } + } + } else if workflow_deletion { // A single commit owns public acceptance, domain mutation, and dispatch. // Failure rolls everything back; identical concurrent requests cannot // divide insertion and repair ownership between two relay workers. @@ -6610,6 +6719,112 @@ mod postgres_tests { } } + /// A committed HTTP publication is a historic duplicate when retried through + /// WebSocket auth after policy advancement, while new stale bytes are denied. + #[tokio::test] + #[ignore = "requires Postgres and Redis"] + async fn fenced_ingest_accepts_exact_retry_across_http_and_websocket_auth() { + use buzz_db::channel::{ChannelType, ChannelUpdate, ChannelVisibility}; + use nostr::{EventBuilder, Keys, Kind, Tag}; + + let db_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); // sadscan:disable np.postgres.1 + let pool = sqlx::PgPool::connect(&db_url) + .await + .expect("connect test Postgres"); + let state = build_canvas_ingest_state(&db_url, &pool).await; + let host = format!("ifc-ingest-{}.test", Uuid::new_v4().simple()); + let community = state + .db + .ensure_configured_community(&host) + .await + .expect("ensure community") + .id; + let tenant = TenantContext::resolved(community, &host); + let author = Keys::generate(); + let channel_id = Uuid::new_v4(); + state + .db + .create_channel_with_id( + community, + channel_id, + &format!("ifc-ingest-{}", channel_id.simple()), + ChannelType::Stream, + ChannelVisibility::Open, + None, + author.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("create test channel"); + let channel = channel_id.to_string(); + let community_text = community.to_string(); + let expiry = (Utc::now().timestamp() + 120).to_string(); + let fenced = |content: &str| { + EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), content) + .tags([ + Tag::parse(["h", channel.as_str()]).expect("h tag"), + Tag::parse(["ifc", "1", community_text.as_str(), expiry.as_str()]) + .expect("ifc tag"), + Tag::parse(["ifc-source", channel.as_str(), "1"]).expect("source tag"), + Tag::parse(["ifc-destination", channel.as_str(), "1"]) + .expect("destination tag"), + ]) + .sign_with_keys(&author) + .expect("sign fenced event") + }; + let event = fenced("accepted once"); + let tracer: Arc = Arc::new(VecTracer::default()); + let http = IngestAuth::Http { + pubkey: author.public_key(), + scopes: vec![Scope::MessagesWrite], + auth_method: HttpAuthMethod::Nip98, + }; + ingest_event_inner(&state, &tracer, &tenant, event.clone(), http) + .await + .expect("HTTP fenced publication"); + + state + .db + .update_channel( + community, + channel_id, + ChannelUpdate { + visibility: Some("private".to_owned()), + ..ChannelUpdate::default() + }, + ) + .await + .expect("advance policy"); + let websocket = || IngestAuth::Nip42 { + pubkey: author.public_key(), + scopes: vec![Scope::MessagesWrite], + channel_ids: Some(vec![channel_id]), + conn_id: Uuid::new_v4(), + }; + let retry = ingest_event_inner(&state, &tracer, &tenant, event.clone(), websocket()) + .await + .expect("WebSocket exact retry"); + assert_eq!(retry.message, "duplicate:"); + + let stale = fenced("never inserted"); + let stale_id = stale.id.as_bytes().to_vec(); + assert!(matches!( + ingest_event_inner(&state, &tracer, &tenant, stale, websocket()).await, + Err(IngestError::CanvasConflict(message)) if message == "ifc-policy-changed" + )); + let stored: i64 = sqlx::query_scalar( + "SELECT count(*) FROM events WHERE community_id=$1 AND id = ANY($2::bytea[])", + ) + .bind(community.as_uuid()) + .bind(vec![event.id.as_bytes().as_slice(), stale_id.as_slice()]) + .fetch_one(&pool) + .await + .expect("count production ingest rows"); + assert_eq!(stored, 1); + } + /// End-to-end CAS dispatch wiring: a tagged write inserts, a stale same-head /// competitor returns the exact conflict: rejection, the loser is absent from /// the DB, and an untagged write still appends unconditionally. diff --git a/crates/buzz-relay/src/ifc.rs b/crates/buzz-relay/src/ifc.rs new file mode 100644 index 0000000000..f21baa2a13 --- /dev/null +++ b/crates/buzz-relay/src/ifc.rs @@ -0,0 +1,534 @@ +//! Strict remote-agent information-flow query proofs. + +use std::collections::HashSet; + +use axum::http::StatusCode; +use buzz_core::kind::{ + KIND_CANVAS, KIND_IFC_ATTESTATION, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2, +}; +use buzz_core::{CommunityId, TenantContext}; +use buzz_db::ifc::IfcError; +use buzz_db::EventQuery; +use chrono::{DateTime, Utc}; +use nostr::{Event, EventBuilder, Kind, PublicKey, Tag}; +use serde::de::{DeserializeSeed, MapAccess, SeqAccess, Visitor}; +use serde::Serialize; +use serde_json::{Map, Value}; +use sha2::{Digest, Sha256}; +use uuid::Uuid; + +use crate::state::AppState; + +const HISTORY_KINDS: [u32; 3] = [KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2, KIND_CANVAS]; +/// Maximum serialized attestation content returned by one query. +pub(crate) const MAX_RESPONSE_BYTES: usize = 8 * 1024 * 1024; +/// End-to-end deadline for one proof query. +pub(crate) const QUERY_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10); + +#[derive(Debug)] +struct Request { + channel_id: Uuid, + nonce: String, + expected_generation: Option, + policy_only: bool, + query: EventQuery, +} + +#[derive(Serialize)] +struct Content<'a> { + version: u8, + community_id: String, + requester: &'a str, + nonce: &'a str, + request_sha256: String, + policy: Policy<'a>, + resources: Vec>, +} + +#[derive(Serialize)] +struct Policy<'a> { + channel_id: String, + generation: String, + channel_type: &'a str, + visibility: &'a str, + history_access: &'a str, + archived: bool, + members: &'a [String], +} + +#[derive(Serialize)] +struct Resource<'a> { + event: &'a Event, + effective_generation: String, + origin_generation: Option, +} + +fn bad(message: impl Into) -> (StatusCode, String) { + (StatusCode::BAD_REQUEST, message.into()) +} + +fn positive_decimal(value: &Value, field: &str) -> Result { + let text = value + .as_str() + .ok_or_else(|| bad(format!("{field} must be a decimal string")))?; + if text.is_empty() || text.starts_with('0') || !text.bytes().all(|byte| byte.is_ascii_digit()) { + return Err(bad(format!("{field} is not canonical"))); + } + text.parse::() + .ok() + .filter(|value| *value > 0) + .ok_or_else(|| bad(format!("{field} is outside the supported range"))) +} + +fn string_array(value: &Value, field: &str) -> Result, (StatusCode, String)> { + value + .as_array() + .ok_or_else(|| bad(format!("{field} must be an array")))? + .iter() + .map(|item| { + item.as_str() + .map(ToOwned::to_owned) + .ok_or_else(|| bad(format!("{field} entries must be strings"))) + }) + .collect() +} + +fn decode_hex32(value: &str, field: &str) -> Result, (StatusCode, String)> { + if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Err(bad(format!("{field} entries must be 32-byte hex"))); + } + hex::decode(value).map_err(|_| bad(format!("{field} entries must be hex"))) +} + +fn parse_request(body: &[u8], community_id: CommunityId) -> Result { + let value = strict_json(body).map_err(|error| bad(format!("invalid IFC query: {error}")))?; + let filters = value + .as_array() + .filter(|filters| filters.len() == 1) + .ok_or_else(|| bad("IFC query requires exactly one filter"))?; + let object = filters[0] + .as_object() + .ok_or_else(|| bad("IFC filter must be an object"))?; + let allowed = [ + "kinds", "#h", "limit", "since", "until", "ids", "authors", "#e", "ifc", + ]; + if object.keys().any(|key| !allowed.contains(&key.as_str())) { + return Err(bad("IFC filter contains an unsupported field")); + } + let extension = object + .get("ifc") + .and_then(Value::as_object) + .ok_or_else(|| bad("IFC extension must be an object"))?; + if extension + .keys() + .any(|key| !["version", "nonce", "expected_generation"].contains(&key.as_str())) + { + return Err(bad("IFC extension contains an unsupported field")); + } + if extension.get("version").and_then(Value::as_u64) != Some(1) { + return Err(bad("unsupported IFC version")); + } + let nonce = extension + .get("nonce") + .and_then(Value::as_str) + .filter(|nonce| { + nonce.len() == 32 + && nonce + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + }) + .ok_or_else(|| bad("IFC nonce must be 32 lowercase hex characters"))? + .to_owned(); + let expected_generation = extension + .get("expected_generation") + .map(|value| positive_decimal(value, "expected_generation")) + .transpose()?; + let channels = string_array(object.get("#h").ok_or_else(|| bad("missing #h"))?, "#h")?; + if channels.len() != 1 { + return Err(bad("IFC query requires exactly one #h channel")); + } + let channel_id = Uuid::parse_str(&channels[0]).map_err(|_| bad("#h is not a UUID"))?; + if channel_id.to_string() != channels[0] { + return Err(bad("#h UUID is not canonical")); + } + let kinds = object + .get("kinds") + .and_then(Value::as_array) + .ok_or_else(|| bad("IFC query requires kinds"))? + .iter() + .map(|kind| { + kind.as_u64() + .and_then(|kind| u32::try_from(kind).ok()) + .ok_or_else(|| bad("kind is invalid")) + }) + .collect::, _>>()?; + let limit = object + .get("limit") + .and_then(Value::as_i64) + .ok_or_else(|| bad("IFC query requires a limit"))?; + let policy_only = kinds == [KIND_IFC_ATTESTATION] && limit == 1; + if !policy_only + && (kinds.is_empty() + || kinds.iter().any(|kind| !HISTORY_KINDS.contains(kind)) + || limit <= 0 + || limit > buzz_db::ifc::IFC_MAX_RESOURCES) + { + return Err(bad("unsupported IFC history query")); + } + if policy_only + && object + .keys() + .any(|key| !["kinds", "#h", "limit", "ifc"].contains(&key.as_str())) + { + return Err(bad("policy-only IFC query contains history filters")); + } + + let mut query = EventQuery::for_community(community_id); + query.channel_id = Some(channel_id); + query.kinds = Some(if policy_only { + Vec::new() + } else { + kinds.iter().map(|kind| *kind as i32).collect() + }); + query.limit = Some(if policy_only { 0 } else { limit }); + query.max_limit = Some(buzz_db::ifc::IFC_MAX_RESOURCES); + query.since = object + .get("since") + .map(|value| parse_timestamp(value, "since")) + .transpose()?; + query.until = object + .get("until") + .map(|value| parse_timestamp(value, "until")) + .transpose()?; + query.ids = object + .get("ids") + .map(|value| { + string_array(value, "ids")? + .iter() + .map(|value| decode_hex32(value, "ids")) + .collect() + }) + .transpose()?; + query.authors = object + .get("authors") + .map(|value| { + string_array(value, "authors")? + .iter() + .map(|value| decode_hex32(value, "authors")) + .collect() + }) + .transpose()?; + query.e_tags = object + .get("#e") + .map(|value| string_array(value, "#e")) + .transpose()?; + Ok(Request { + channel_id, + nonce, + expected_generation, + policy_only, + query, + }) +} + +fn parse_timestamp(value: &Value, field: &str) -> Result, (StatusCode, String)> { + let seconds = value + .as_i64() + .ok_or_else(|| bad(format!("{field} must be an integer")))?; + DateTime::from_timestamp(seconds, 0) + .ok_or_else(|| bad(format!("{field} is outside the supported range"))) +} + +/// Return `None` only when the request does not opt into IFC. +pub(crate) async fn query( + state: &AppState, + tenant: &TenantContext, + requester: &PublicKey, + body: &[u8], +) -> Option> { + let opts_in = serde_json::from_slice::(body) + .ok() + .and_then(|value| value.as_array().cloned()) + .is_some_and(|filters| { + filters.iter().any(|filter| { + filter + .as_object() + .is_some_and(|object| object.contains_key("ifc")) + }) + }); + if !opts_in { + return None; + } + Some( + tokio::time::timeout(QUERY_DEADLINE, query_inner(state, tenant, requester, body)) + .await + .unwrap_or_else(|_| { + Err(( + StatusCode::SERVICE_UNAVAILABLE, + "ifc-unavailable".to_owned(), + )) + }), + ) +} + +async fn query_inner( + state: &AppState, + tenant: &TenantContext, + requester: &PublicKey, + body: &[u8], +) -> Result { + let request = parse_request(body, tenant.community())?; + let requester_hex = requester.to_hex(); + let cut = state + .db + .ifc_read_cut( + tenant.community(), + request.channel_id, + &requester_hex, + request.expected_generation, + &request.query, + ) + .await + .map_err(map_db_error)?; + if request.policy_only && !cut.resources.is_empty() { + return Err((StatusCode::SERVICE_UNAVAILABLE, "ifc-invalid".to_owned())); + } + for resource in &cut.resources { + if resource.channel_id != Some(request.channel_id) + || !HISTORY_KINDS.contains(&(resource.event.kind.as_u16() as u32)) + || !resource.event.verify_id() + || !resource.event.verify_signature() + { + return Err((StatusCode::SERVICE_UNAVAILABLE, "ifc-invalid".to_owned())); + } + } + let generation = cut.policy.generation.to_string(); + let content = Content { + version: 1, + community_id: tenant.community().to_string(), + requester: &requester_hex, + nonce: &request.nonce, + request_sha256: hex::encode(Sha256::digest(body)), + policy: Policy { + channel_id: cut.policy.channel_id.to_string(), + generation: generation.clone(), + channel_type: &cut.policy.channel_type, + visibility: &cut.policy.visibility, + history_access: &cut.policy.history_access, + archived: cut.policy.archived, + members: &cut.policy.members, + }, + resources: cut + .resources + .iter() + .map(|resource| Resource { + event: &resource.event, + effective_generation: generation.clone(), + origin_generation: None, + }) + .collect(), + }; + let content = serde_json::to_string(&content) + .map_err(|_| (StatusCode::INTERNAL_SERVER_ERROR, "ifc-invalid".to_owned()))?; + if content.len() > MAX_RESPONSE_BYTES { + return Err(( + StatusCode::PAYLOAD_TOO_LARGE, + "ifc-response-too-large".to_owned(), + )); + } + let channel_text = request.channel_id.to_string(); + let tags = [ + Tag::parse(["p", requester_hex.as_str()]) + .map_err(|_| (StatusCode::INTERNAL_SERVER_ERROR, "ifc-invalid".to_owned()))?, + Tag::parse(["h", channel_text.as_str()]) + .map_err(|_| (StatusCode::INTERNAL_SERVER_ERROR, "ifc-invalid".to_owned()))?, + ]; + let attestation = EventBuilder::new(Kind::Custom(KIND_IFC_ATTESTATION as u16), content) + .tags(tags) + .sign_with_keys(&state.relay_keypair) + .map_err(|_| (StatusCode::INTERNAL_SERVER_ERROR, "ifc-invalid".to_owned()))?; + Ok(serde_json::json!({"ifc": 1, "attestation": attestation})) +} + +fn map_db_error(error: IfcError) -> (StatusCode, String) { + match error { + IfcError::PolicyChanged => (StatusCode::CONFLICT, "ifc-policy-changed".to_owned()), + IfcError::AccessDenied => ( + StatusCode::FORBIDDEN, + "restricted: channel access denied".to_owned(), + ), + IfcError::Invalid => (StatusCode::SERVICE_UNAVAILABLE, "ifc-invalid".to_owned()), + IfcError::Expired => (StatusCode::BAD_REQUEST, "ifc-expired".to_owned()), + IfcError::Database(_) => ( + StatusCode::SERVICE_UNAVAILABLE, + "ifc-unavailable".to_owned(), + ), + } +} + +fn strict_json(input: &[u8]) -> Result { + let mut deserializer = serde_json::Deserializer::from_slice(input); + let value = NoDuplicates.deserialize(&mut deserializer)?; + deserializer.end()?; + Ok(value) +} + +struct NoDuplicates; + +impl<'de> DeserializeSeed<'de> for NoDuplicates { + type Value = Value; + + fn deserialize(self, deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + deserializer.deserialize_any(NoDuplicateVisitor) + } +} + +struct NoDuplicateVisitor; + +impl<'de> Visitor<'de> for NoDuplicateVisitor { + type Value = Value; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("JSON without duplicate object fields") + } + + fn visit_bool(self, value: bool) -> Result { + Ok(Value::Bool(value)) + } + fn visit_i64(self, value: i64) -> Result { + Ok(Value::Number(value.into())) + } + fn visit_u64(self, value: u64) -> Result { + Ok(Value::Number(value.into())) + } + fn visit_f64(self, value: f64) -> Result + where + E: serde::de::Error, + { + serde_json::Number::from_f64(value) + .map(Value::Number) + .ok_or_else(|| E::custom("non-finite number")) + } + fn visit_str(self, value: &str) -> Result { + Ok(Value::String(value.to_owned())) + } + fn visit_string(self, value: String) -> Result { + Ok(Value::String(value)) + } + fn visit_none(self) -> Result { + Ok(Value::Null) + } + fn visit_unit(self) -> Result { + Ok(Value::Null) + } + fn visit_some(self, deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + NoDuplicates.deserialize(deserializer) + } + fn visit_seq(self, mut sequence: A) -> Result + where + A: SeqAccess<'de>, + { + let mut values = Vec::new(); + while let Some(value) = sequence.next_element_seed(NoDuplicates)? { + values.push(value); + } + Ok(Value::Array(values)) + } + fn visit_map(self, mut map: A) -> Result + where + A: MapAccess<'de>, + { + let mut values = Map::new(); + let mut seen = HashSet::new(); + while let Some(key) = map.next_key::()? { + if !seen.insert(key.clone()) { + return Err(serde::de::Error::custom(format!("duplicate field {key}"))); + } + values.insert(key, map.next_value_seed(NoDuplicates)?); + } + Ok(Value::Object(values)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use buzz_core::ifc::parse_publication_fence; + + #[test] + fn duplicate_ifc_fields_are_rejected() { + let body = br##"[{"kinds":[45900],"#h":["11111111-1111-4111-8111-111111111111"],"limit":1,"ifc":{"version":1,"version":1,"nonce":"0123456789abcdef0123456789abcdef"}}]"##; + assert!(strict_json(body).is_err()); + } + + #[test] + fn shared_history_request_is_strictly_parsed() { + let body = br##"[{"kinds":[9,40002,40100],"#h":["11111111-1111-4111-8111-111111111111"],"limit":13,"ifc":{"version":1,"nonce":"0123456789abcdef0123456789abcdef","expected_generation":"17"}}]"##; + let request = parse_request(body, CommunityId::from_uuid(Uuid::nil())).expect("request"); + assert_eq!(request.expected_generation, Some(17)); + assert!(!request.policy_only); + assert_eq!(request.query.limit, Some(13)); + assert_eq!(request.query.kinds, Some(vec![9, 40002, 40100])); + } + + #[test] + fn unknown_ifc_fields_and_noncanonical_generation_are_rejected() { + for body in [ + br##"[{"kinds":[45900],"#h":["11111111-1111-4111-8111-111111111111"],"limit":1,"ifc":{"version":1,"nonce":"0123456789abcdef0123456789abcdef","extra":true}}]"##.as_slice(), + br##"[{"kinds":[45900],"#h":["11111111-1111-4111-8111-111111111111"],"limit":1,"ifc":{"version":1,"nonce":"0123456789abcdef0123456789abcdef","expected_generation":"017"}}]"##.as_slice(), + ] { + assert!(parse_request(body, CommunityId::from_uuid(Uuid::nil())).is_err()); + } + } + + #[test] + fn http_and_websocket_publish_through_the_shared_ifc_gate() { + let ingest = include_str!("handlers/ingest.rs"); + let http = include_str!("api/bridge.rs"); + let websocket = include_str!("handlers/event.rs"); + assert!(ingest.contains("parse_publication_fence(&event)")); + assert!(ingest.contains(".publish_ifc_event(")); + assert!(http.contains("crate::handlers::ingest::ingest_event(")); + assert!(websocket.contains("super::ingest::ingest_event(")); + } + + #[test] + fn golden_fixture_has_valid_signatures_and_exact_request_digest() { + let fixture_text = include_str!("../tests/fixtures/ifc-v1-golden.json"); + assert!(!fixture_text.contains("REPLACE_WITH")); + let fixture: Value = serde_json::from_str(fixture_text).expect("parse fixture"); + let request = fixture["request_body_utf8"] + .as_str() + .expect("request bytes"); + assert_eq!( + fixture["request_sha256"].as_str(), + Some(hex::encode(Sha256::digest(request.as_bytes())).as_str()) + ); + let attestation: Event = serde_json::from_value(fixture["response"]["attestation"].clone()) + .expect("attestation event"); + assert!(attestation.verify_id()); + assert!(attestation.verify_signature()); + let content: Value = serde_json::from_str(&attestation.content).expect("content"); + assert_eq!(content["request_sha256"], fixture["request_sha256"]); + assert!(content["policy"]["generation"].is_string()); + assert!(content["resources"][0]["origin_generation"].is_null()); + let resource: Event = serde_json::from_value(content["resources"][0]["event"].clone()) + .expect("resource event"); + assert!(resource.verify_id()); + assert!(resource.verify_signature()); + let publication: Event = + serde_json::from_value(fixture["channel_publication_event"].clone()) + .expect("publication event"); + assert!(publication.verify_id()); + assert!(publication.verify_signature()); + assert!(parse_publication_fence(&publication) + .expect("parse publication") + .is_some()); + } +} diff --git a/crates/buzz-relay/src/lib.rs b/crates/buzz-relay/src/lib.rs index e0fd61d1f6..9b2acb2103 100644 --- a/crates/buzz-relay/src/lib.rs +++ b/crates/buzz-relay/src/lib.rs @@ -36,6 +36,8 @@ pub mod connection; pub mod error; /// WebSocket message handlers for NIP-01 client commands. pub mod handlers; +/// Remote-agent information-flow proof protocol. +pub(crate) mod ifc; /// Stateless HMAC-signed relay invite tokens (mint/verify). pub mod invite_token; /// Fixed-schema evidence for the relay's earliest startup steps. diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 4ed127dfff..0abec62242 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1026,6 +1026,15 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { )); } } + match state.db.verify_ifc_policy_catalog().await { + Ok(()) => info!("IFC policy catalog verified"), + Err(error) => { + error!(%error, "IFC policy catalog validation failed"); + return Err(anyhow::anyhow!( + "IFC policy catalog is unsafe; apply or repair migration 0061 before starting this relay: {error}" + )); + } + } step.finish(); // Repair legacy NIP-29 channel rosters that were persisted while the diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index d60771b2bc..fdce52046c 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -34,6 +34,9 @@ pub struct RelayInfo { /// Host-bound atomic read-state snapshot capability; absent on unresolved hosts. #[serde(skip_serializing_if = "Option::is_none")] pub read_state_snapshot: Option, + /// Remote-agent information-flow proof and fenced-publication capability. + #[serde(skip_serializing_if = "Option::is_none")] + pub ifc: Option, /// Opt-in, host-bound private accessory API; separate from legacy NIP-RS. #[serde(skip_serializing_if = "Option::is_none")] pub buzz_v1: Option, @@ -271,6 +274,7 @@ impl RelayInfo { description: "Buzz — private team communication relay".to_string(), icon: icon.filter(|s| !s.is_empty()).map(|s| s.to_string()), read_state_snapshot: None, + ifc: None, buzz_v1: None, artifacts: serde_json::json!({ "version": 1, "revision_kind": 45010, "removal_kind": 45011, @@ -381,6 +385,7 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st state.config.klipy.as_ref().map(|_| "klipy"), ); if let Ok(tenant) = crate::tenant::bind_community(&state.db, raw_host).await { + info.ifc = ifc_descriptor(state.config.relay_private_key.is_some()); if state.config.buzz_v1_enabled { use buzz_db::personal_read::{ELIGIBLE_KINDS, MAX_CHANNELS, MAX_INTENTS}; info.buzz_v1 = Some(BuzzV1Descriptor { @@ -422,6 +427,25 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st info } +fn ifc_descriptor(has_stable_relay_key: bool) -> Option { + has_stable_relay_key.then(|| { + serde_json::json!({ + "version": 1, + "attestation_kind": buzz_core::kind::KIND_IFC_ATTESTATION, + "query": "/query", + "max_sources": buzz_core::ifc::IFC_MAX_SOURCES, + "max_fence_ttl_seconds": buzz_core::ifc::IFC_MAX_TTL_SECONDS, + "receipt_retention_seconds": buzz_core::ifc::IFC_RECEIPT_RETENTION_SECONDS, + "max_policy_members": buzz_db::ifc::IFC_MAX_POLICY_MEMBERS, + "max_resources": buzz_db::ifc::IFC_MAX_RESOURCES, + "max_response_bytes": crate::ifc::MAX_RESPONSE_BYTES, + "query_timeout_ms": crate::ifc::QUERY_DEADLINE.as_millis(), + "history_access": ["all_current_members"], + "fenced_kinds": buzz_core::ifc::IFC_CHANNEL_PUBLICATION_KINDS, + }) + }) +} + /// Fetches the workspace icon for the community bound to `raw_host`, as the /// host-scoped scalar consumed by [`RelayInfo::build`]. /// @@ -532,6 +556,22 @@ mod tests { use super::*; + #[test] + fn ifc_descriptor_requires_stable_key_and_omits_owner_memory() { + assert!(ifc_descriptor(false).is_none()); + let descriptor = ifc_descriptor(true).expect("stable relay key advertises IFC"); + assert_eq!(descriptor["version"], 1); + assert_eq!( + descriptor["attestation_kind"], + buzz_core::kind::KIND_IFC_ATTESTATION + ); + assert_eq!( + descriptor["fenced_kinds"], + serde_json::json!(buzz_core::ifc::IFC_CHANNEL_PUBLICATION_KINDS) + ); + assert!(!descriptor.to_string().contains("30078")); + } + #[test] fn push_descriptor_is_gated_by_gateway_configuration_and_tenant_binding() { let keys = nostr::Keys::generate(); diff --git a/crates/buzz-relay/tests/fixtures/ifc-v1-golden.json b/crates/buzz-relay/tests/fixtures/ifc-v1-golden.json new file mode 100644 index 0000000000..8799b8f4db --- /dev/null +++ b/crates/buzz-relay/tests/fixtures/ifc-v1-golden.json @@ -0,0 +1,46 @@ +{ + "channel_publication_event": { + "content": "fixture remote-agent output", + "created_at": 1791999900, + "id": "102e1649791d5b6f0d77101571b9cabdafbbdcf3e8da82745c1e5e7b5f486e33", + "kind": 40002, + "pubkey": "f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9", + "sig": "f053bcacbcea6778646d9b794acade4ef7d7b1f8ae5bb2921100723f9ca8d5cf13d83d0829447ddea22656c6c71304996ea81eb408d2d63734de71f4c9353aa5", + "tags": [ + ["h", "22222222-2222-4222-8222-222222222222"], + ["ifc", "1", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", "1792000000"], + ["ifc-source", "11111111-1111-4111-8111-111111111111", "17"], + ["ifc-destination", "22222222-2222-4222-8222-222222222222", "23"] + ] + }, + "contract": "buzz-ifc-v1", + "owner_memory": { + "enabled": false, + "reason": "awaiting the backend's existing memory address grammar" + }, + "request_body_utf8": "[{\"kinds\":[9,40002,40100],\"#h\":[\"11111111-1111-4111-8111-111111111111\"],\"limit\":13,\"ifc\":{\"version\":1,\"nonce\":\"0123456789abcdef0123456789abcdef\",\"expected_generation\":\"17\"}}]", + "request_sha256": "6f1fbcbe47a1165230b8b9c622748af4c0a1fcc55374425e7b5688c5b0a6aa44", + "response": { + "attestation": { + "content": "{\"community_id\":\"aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa\",\"nonce\":\"0123456789abcdef0123456789abcdef\",\"policy\":{\"archived\":false,\"channel_id\":\"11111111-1111-4111-8111-111111111111\",\"channel_type\":\"dm\",\"generation\":\"17\",\"history_access\":\"all_current_members\",\"members\":[\"c6047f9441ed7d6d3045406e95c07cd85c778e4b8cef3ca7abac09b95c709ee5\",\"f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9\"],\"visibility\":\"private\"},\"request_sha256\":\"6f1fbcbe47a1165230b8b9c622748af4c0a1fcc55374425e7b5688c5b0a6aa44\",\"requester\":\"c6047f9441ed7d6d3045406e95c07cd85c778e4b8cef3ca7abac09b95c709ee5\",\"resources\":[{\"effective_generation\":\"17\",\"event\":{\"content\":\"fixture source message\",\"created_at\":1791999700,\"id\":\"9d66eb8ad4dbc6990b835f895dc7a0c4bb4bec7535848cd56f3c03ff59e67226\",\"kind\":40002,\"pubkey\":\"c6047f9441ed7d6d3045406e95c07cd85c778e4b8cef3ca7abac09b95c709ee5\",\"sig\":\"fa9624d82d13441e51f8de031f706b646aec33e6aee50dd31f6cb6f1cff2b47cd2282ec3abeedc8f8fa447fab264224d42bc613c78f4efcd452d24ed47be9ffb\",\"tags\":[[\"h\",\"11111111-1111-4111-8111-111111111111\"]]},\"origin_generation\":null}],\"version\":1}", + "created_at": 1791999800, + "id": "de2d939e7e9f596ab5b03a28c1dd2af0515416d83cb71e9a6cc2a976f6619f58", + "kind": 45900, + "pubkey": "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + "sig": "e54811cca3d75343a424673f79e86c8d1ae05009f683a073cc5f9f0236eaa7cdaa13defa2070dd36a238f0a4a83203bf1cc3dcd85af70f7fb223553b1e8ee94c", + "tags": [ + ["p", "c6047f9441ed7d6d3045406e95c07cd85c778e4b8cef3ca7abac09b95c709ee5"], + ["h", "11111111-1111-4111-8111-111111111111"] + ] + }, + "ifc": 1 + }, + "test_keys": { + "broker_pubkey": "f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9", + "broker_secret_hex": "0000000000000000000000000000000000000000000000000000000000000003", + "relay_pubkey": "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", + "relay_secret_hex": "0000000000000000000000000000000000000000000000000000000000000001", + "requester_pubkey": "c6047f9441ed7d6d3045406e95c07cd85c778e4b8cef3ca7abac09b95c709ee5", + "requester_secret_hex": "0000000000000000000000000000000000000000000000000000000000000002" + } +} diff --git a/migrations/0061_ifc_channel_policy.sql b/migrations/0061_ifc_channel_policy.sql new file mode 100644 index 0000000000..052c04634a --- /dev/null +++ b/migrations/0061_ifc_channel_policy.sql @@ -0,0 +1,357 @@ +-- Canonical monotone channel policy generations and durable fenced-event +-- receipts for remote-agent IFC version 1. + +ALTER TABLE channels + ADD COLUMN history_access TEXT NOT NULL DEFAULT 'all_current_members', + ADD CONSTRAINT chk_channels_history_access + CHECK (history_access = 'all_current_members'); + +CREATE TABLE channel_policy_generations ( + community_id UUID NOT NULL REFERENCES communities(id) ON DELETE CASCADE, + channel_id UUID NOT NULL, + generation BIGINT NOT NULL DEFAULT 1 CHECK (generation > 0), + deleted BOOLEAN NOT NULL DEFAULT FALSE, + created_xid xid8 NOT NULL DEFAULT pg_current_xact_id(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, channel_id) +); + +INSERT INTO channel_policy_generations + (community_id, channel_id, generation, deleted, created_xid) +SELECT community_id, id, 1, deleted_at IS NOT NULL, pg_current_xact_id() +FROM channels; + +CREATE TABLE ifc_event_acceptances ( + community_id UUID NOT NULL REFERENCES communities(id) ON DELETE CASCADE, + event_id BYTEA NOT NULL CHECK (octet_length(event_id) = 32), + event_kind INT NOT NULL, + fence_expires_at TIMESTAMPTZ NOT NULL, + accepted_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, event_id) +); + +CREATE INDEX idx_ifc_event_acceptances_accepted_at + ON ifc_event_acceptances (accepted_at); + +CREATE FUNCTION channel_policy_lock_key(target_community UUID, target_channel UUID) +RETURNS BIGINT +LANGUAGE SQL IMMUTABLE STRICT PARALLEL SAFE AS $$ + SELECT hashtextextended( + 'buzz_channel_membership:' || target_community::text || ':' || target_channel::text, + 0 + ) +$$; + +CREATE FUNCTION initialize_channel_policy_generation() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + SELECT community_id, id AS channel_id + FROM new_channels + ORDER BY community_id, id + LOOP + INSERT INTO channel_policy_generations + (community_id, channel_id, generation, deleted, created_xid) + VALUES + (target.community_id, target.channel_id, 1, FALSE, pg_current_xact_id()) + ON CONFLICT (community_id, channel_id) DO NOTHING; + IF NOT FOUND THEN + RAISE EXCEPTION 'deleted channel identity cannot be reused: %/%', + target.community_id, target.channel_id + USING ERRCODE = '23514'; + END IF; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION channel_member_policy_identity_immutable() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF ROW(NEW.community_id, NEW.channel_id, NEW.pubkey) + IS DISTINCT FROM ROW(OLD.community_id, OLD.channel_id, OLD.pubkey) THEN + RAISE EXCEPTION 'channel membership identity is immutable' + USING ERRCODE = '23514'; + END IF; + RETURN NEW; +END +$$; + +CREATE FUNCTION channel_policy_identity_immutable() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF ROW(NEW.community_id, NEW.id) + IS DISTINCT FROM ROW(OLD.community_id, OLD.id) THEN + RAISE EXCEPTION 'channel policy identity is immutable' + USING ERRCODE = '23514'; + END IF; + RETURN NEW; +END +$$; + +CREATE FUNCTION guard_channel_policy_generation_row() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'DELETE' THEN + IF pg_trigger_depth() <= 1 THEN + RAISE EXCEPTION 'channel policy generations cannot be deleted directly' + USING ERRCODE = '23514'; + END IF; + RETURN OLD; + END IF; + IF ROW(NEW.community_id, NEW.channel_id, NEW.created_xid) + IS DISTINCT FROM ROW(OLD.community_id, OLD.channel_id, OLD.created_xid) + OR NEW.generation < OLD.generation + OR (OLD.deleted AND NOT NEW.deleted) THEN + RAISE EXCEPTION 'channel policy generation is monotone and identity-bound' + USING ERRCODE = '23514'; + END IF; + RETURN NEW; +END +$$; + +CREATE FUNCTION guard_ifc_acceptance_row() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'UPDATE' OR pg_trigger_depth() <= 1 THEN + RAISE EXCEPTION 'IFC acceptance receipts are immutable' + USING ERRCODE = '23514'; + END IF; + RETURN OLD; +END +$$; + +CREATE FUNCTION advance_channel_policy_for_channel_update() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + SELECT n.community_id, n.id AS channel_id, n.deleted_at IS NOT NULL AS deleted + FROM new_channels n + JOIN old_channels o USING (community_id, id) + WHERE ROW(n.channel_type, n.visibility, n.history_access, n.created_by, + n.archived_at, n.deleted_at) + IS DISTINCT FROM + ROW(o.channel_type, o.visibility, o.history_access, o.created_by, + o.archived_at, o.deleted_at) + ORDER BY n.community_id, n.id + LOOP + PERFORM pg_advisory_xact_lock( + channel_policy_lock_key(target.community_id, target.channel_id)); + UPDATE channel_policy_generations + SET generation = generation + 1, + deleted = target.deleted, + updated_at = NOW() + WHERE community_id = target.community_id + AND channel_id = target.channel_id; + IF NOT FOUND THEN + RAISE EXCEPTION 'channel policy generation missing: %/%', + target.community_id, target.channel_id + USING ERRCODE = '23514'; + END IF; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION advance_channel_policy_for_channel_delete() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + SELECT community_id, id AS channel_id + FROM old_channels + ORDER BY community_id, id + LOOP + PERFORM pg_advisory_xact_lock( + channel_policy_lock_key(target.community_id, target.channel_id)); + UPDATE channel_policy_generations + SET generation = generation + 1, + deleted = TRUE, + updated_at = NOW() + WHERE community_id = target.community_id + AND channel_id = target.channel_id; + IF NOT FOUND THEN + RAISE EXCEPTION 'channel policy generation missing: %/%', + target.community_id, target.channel_id + USING ERRCODE = '23514'; + END IF; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION advance_channel_policy_for_member_insert() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + SELECT DISTINCT n.community_id, n.channel_id + FROM new_members n + JOIN channel_policy_generations p + ON p.community_id = n.community_id AND p.channel_id = n.channel_id + WHERE n.removed_at IS NULL + AND p.deleted = FALSE + AND p.created_xid <> pg_current_xact_id() + ORDER BY n.community_id, n.channel_id + LOOP + PERFORM pg_advisory_xact_lock( + channel_policy_lock_key(target.community_id, target.channel_id)); + UPDATE channel_policy_generations + SET generation = generation + 1, updated_at = NOW() + WHERE community_id = target.community_id + AND channel_id = target.channel_id + AND deleted = FALSE; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION advance_channel_policy_for_member_update() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + WITH changed AS ( + SELECT o.community_id AS old_community, o.channel_id AS old_channel, + o.role AS old_role, o.removed_at IS NULL AS old_active, + n.community_id AS new_community, n.channel_id AS new_channel, + n.role AS new_role, n.removed_at IS NULL AS new_active + FROM old_members o + JOIN new_members n + ON o.community_id = n.community_id + AND o.channel_id = n.channel_id + AND o.pubkey = n.pubkey + WHERE (o.removed_at IS NULL) IS DISTINCT FROM (n.removed_at IS NULL) + OR ((o.removed_at IS NULL) AND (n.removed_at IS NULL) + AND o.role IS DISTINCT FROM n.role) + ), affected AS ( + SELECT old_community AS community_id, old_channel AS channel_id + FROM changed WHERE old_active + UNION + SELECT new_community, new_channel FROM changed WHERE new_active + ) + SELECT community_id, channel_id FROM affected + ORDER BY community_id, channel_id + LOOP + PERFORM pg_advisory_xact_lock( + channel_policy_lock_key(target.community_id, target.channel_id)); + UPDATE channel_policy_generations + SET generation = generation + 1, updated_at = NOW() + WHERE community_id = target.community_id + AND channel_id = target.channel_id + AND deleted = FALSE; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION advance_channel_policy_for_member_delete() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + SELECT DISTINCT o.community_id, o.channel_id + FROM old_members o + JOIN channel_policy_generations p + ON p.community_id = o.community_id AND p.channel_id = o.channel_id + WHERE o.removed_at IS NULL AND p.deleted = FALSE + ORDER BY o.community_id, o.channel_id + LOOP + PERFORM pg_advisory_xact_lock( + channel_policy_lock_key(target.community_id, target.channel_id)); + UPDATE channel_policy_generations + SET generation = generation + 1, updated_at = NOW() + WHERE community_id = target.community_id + AND channel_id = target.channel_id + AND deleted = FALSE; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION enforce_ifc_acceptance_expiry() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF NEW.fence_expires_at <= clock_timestamp() THEN + RAISE EXCEPTION 'ifc-expired' + USING ERRCODE = '23514', CONSTRAINT = 'ifc_acceptance_unexpired'; + END IF; + RETURN NULL; +END +$$; + +CREATE FUNCTION prune_ifc_acceptances_after_insert() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + DELETE FROM ifc_event_acceptances + WHERE ctid IN ( + SELECT a.ctid + FROM ifc_event_acceptances a + JOIN (SELECT DISTINCT community_id FROM new_acceptances) n + ON n.community_id = a.community_id + WHERE a.accepted_at < clock_timestamp() - INTERVAL '24 hours' + ORDER BY a.accepted_at + LIMIT 1000 + ); + RETURN NULL; +END +$$; + +CREATE TRIGGER trg_channels_ifc_policy_insert + AFTER INSERT ON channels + REFERENCING NEW TABLE AS new_channels + FOR EACH STATEMENT EXECUTE FUNCTION initialize_channel_policy_generation(); +CREATE TRIGGER trg_channels_ifc_policy_identity + BEFORE UPDATE ON channels + FOR EACH ROW EXECUTE FUNCTION channel_policy_identity_immutable(); +CREATE TRIGGER trg_channels_ifc_policy_update + AFTER UPDATE ON channels + REFERENCING OLD TABLE AS old_channels NEW TABLE AS new_channels + FOR EACH STATEMENT EXECUTE FUNCTION advance_channel_policy_for_channel_update(); +CREATE TRIGGER trg_channels_ifc_policy_delete + AFTER DELETE ON channels + REFERENCING OLD TABLE AS old_channels + FOR EACH STATEMENT EXECUTE FUNCTION advance_channel_policy_for_channel_delete(); +CREATE TRIGGER trg_channel_members_ifc_policy_insert + AFTER INSERT ON channel_members + REFERENCING NEW TABLE AS new_members + FOR EACH STATEMENT EXECUTE FUNCTION advance_channel_policy_for_member_insert(); +CREATE TRIGGER trg_channel_members_ifc_policy_identity + BEFORE UPDATE ON channel_members + FOR EACH ROW EXECUTE FUNCTION channel_member_policy_identity_immutable(); +CREATE TRIGGER trg_channel_members_ifc_policy_update + AFTER UPDATE ON channel_members + REFERENCING OLD TABLE AS old_members NEW TABLE AS new_members + FOR EACH STATEMENT EXECUTE FUNCTION advance_channel_policy_for_member_update(); +CREATE TRIGGER trg_channel_members_ifc_policy_delete + AFTER DELETE ON channel_members + REFERENCING OLD TABLE AS old_members + FOR EACH STATEMENT EXECUTE FUNCTION advance_channel_policy_for_member_delete(); + +CREATE CONSTRAINT TRIGGER trg_ifc_acceptance_unexpired + AFTER INSERT ON ifc_event_acceptances + DEFERRABLE INITIALLY DEFERRED + FOR EACH ROW EXECUTE FUNCTION enforce_ifc_acceptance_expiry(); + +CREATE TRIGGER trg_ifc_acceptance_prune + AFTER INSERT ON ifc_event_acceptances + REFERENCING NEW TABLE AS new_acceptances + FOR EACH STATEMENT EXECUTE FUNCTION prune_ifc_acceptances_after_insert(); + +CREATE TRIGGER trg_channel_policy_generation_guard + BEFORE UPDATE OR DELETE ON channel_policy_generations + FOR EACH ROW EXECUTE FUNCTION guard_channel_policy_generation_row(); +CREATE TRIGGER trg_ifc_acceptance_guard + BEFORE UPDATE OR DELETE ON ifc_event_acceptances + FOR EACH ROW EXECUTE FUNCTION guard_ifc_acceptance_row(); + +SELECT attach_community_write_fence('channel_policy_generations'); +SELECT attach_community_write_fence('ifc_event_acceptances'); diff --git a/schema/functions/channel_policy_lock_key.sql b/schema/functions/channel_policy_lock_key.sql new file mode 100644 index 0000000000..a513d40c32 --- /dev/null +++ b/schema/functions/channel_policy_lock_key.sql @@ -0,0 +1,11 @@ +-- Same advisory coordinate historically named buzz_channel_membership. Policy +-- generation subsumes that serialization boundary so old and new writers +-- contend on one key during rolling migration. +CREATE FUNCTION channel_policy_lock_key(target_community UUID, target_channel UUID) +RETURNS BIGINT +LANGUAGE SQL IMMUTABLE STRICT PARALLEL SAFE AS $$ + SELECT hashtextextended( + 'buzz_channel_membership:' || target_community::text || ':' || target_channel::text, + 0 + ) +$$; diff --git a/schema/functions/maintain_channel_policy_generation.sql b/schema/functions/maintain_channel_policy_generation.sql new file mode 100644 index 0000000000..ec363dddcd --- /dev/null +++ b/schema/functions/maintain_channel_policy_generation.sql @@ -0,0 +1,264 @@ +CREATE FUNCTION initialize_channel_policy_generation() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + SELECT community_id, id AS channel_id + FROM new_channels + ORDER BY community_id, id + LOOP + INSERT INTO channel_policy_generations + (community_id, channel_id, generation, deleted, created_xid) + VALUES + (target.community_id, target.channel_id, 1, FALSE, pg_current_xact_id()) + ON CONFLICT (community_id, channel_id) DO NOTHING; + IF NOT FOUND THEN + RAISE EXCEPTION 'deleted channel identity cannot be reused: %/%', + target.community_id, target.channel_id + USING ERRCODE = '23514'; + END IF; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION channel_member_policy_identity_immutable() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF ROW(NEW.community_id, NEW.channel_id, NEW.pubkey) + IS DISTINCT FROM ROW(OLD.community_id, OLD.channel_id, OLD.pubkey) THEN + RAISE EXCEPTION 'channel membership identity is immutable' + USING ERRCODE = '23514'; + END IF; + RETURN NEW; +END +$$; + +CREATE FUNCTION channel_policy_identity_immutable() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF ROW(NEW.community_id, NEW.id) + IS DISTINCT FROM ROW(OLD.community_id, OLD.id) THEN + RAISE EXCEPTION 'channel policy identity is immutable' + USING ERRCODE = '23514'; + END IF; + RETURN NEW; +END +$$; + +CREATE FUNCTION guard_channel_policy_generation_row() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'DELETE' THEN + IF pg_trigger_depth() <= 1 THEN + RAISE EXCEPTION 'channel policy generations cannot be deleted directly' + USING ERRCODE = '23514'; + END IF; + RETURN OLD; + END IF; + IF ROW(NEW.community_id, NEW.channel_id, NEW.created_xid) + IS DISTINCT FROM ROW(OLD.community_id, OLD.channel_id, OLD.created_xid) + OR NEW.generation < OLD.generation + OR (OLD.deleted AND NOT NEW.deleted) THEN + RAISE EXCEPTION 'channel policy generation is monotone and identity-bound' + USING ERRCODE = '23514'; + END IF; + RETURN NEW; +END +$$; + +CREATE FUNCTION guard_ifc_acceptance_row() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'UPDATE' OR pg_trigger_depth() <= 1 THEN + RAISE EXCEPTION 'IFC acceptance receipts are immutable' + USING ERRCODE = '23514'; + END IF; + RETURN OLD; +END +$$; + +CREATE FUNCTION advance_channel_policy_for_channel_update() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + SELECT n.community_id, n.id AS channel_id, n.deleted_at IS NOT NULL AS deleted + FROM new_channels n + JOIN old_channels o USING (community_id, id) + WHERE ROW(n.channel_type, n.visibility, n.history_access, n.created_by, + n.archived_at, n.deleted_at) + IS DISTINCT FROM + ROW(o.channel_type, o.visibility, o.history_access, o.created_by, + o.archived_at, o.deleted_at) + ORDER BY n.community_id, n.id + LOOP + PERFORM pg_advisory_xact_lock( + channel_policy_lock_key(target.community_id, target.channel_id)); + UPDATE channel_policy_generations + SET generation = generation + 1, + deleted = target.deleted, + updated_at = NOW() + WHERE community_id = target.community_id + AND channel_id = target.channel_id; + IF NOT FOUND THEN + RAISE EXCEPTION 'channel policy generation missing: %/%', + target.community_id, target.channel_id + USING ERRCODE = '23514'; + END IF; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION advance_channel_policy_for_channel_delete() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + SELECT community_id, id AS channel_id + FROM old_channels + ORDER BY community_id, id + LOOP + PERFORM pg_advisory_xact_lock( + channel_policy_lock_key(target.community_id, target.channel_id)); + UPDATE channel_policy_generations + SET generation = generation + 1, + deleted = TRUE, + updated_at = NOW() + WHERE community_id = target.community_id + AND channel_id = target.channel_id; + IF NOT FOUND THEN + RAISE EXCEPTION 'channel policy generation missing: %/%', + target.community_id, target.channel_id + USING ERRCODE = '23514'; + END IF; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION advance_channel_policy_for_member_insert() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + SELECT DISTINCT n.community_id, n.channel_id + FROM new_members n + JOIN channel_policy_generations p + ON p.community_id = n.community_id AND p.channel_id = n.channel_id + WHERE n.removed_at IS NULL + AND p.deleted = FALSE + AND p.created_xid <> pg_current_xact_id() + ORDER BY n.community_id, n.channel_id + LOOP + PERFORM pg_advisory_xact_lock( + channel_policy_lock_key(target.community_id, target.channel_id)); + UPDATE channel_policy_generations + SET generation = generation + 1, updated_at = NOW() + WHERE community_id = target.community_id + AND channel_id = target.channel_id + AND deleted = FALSE; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION advance_channel_policy_for_member_update() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + WITH changed AS ( + SELECT o.community_id AS old_community, o.channel_id AS old_channel, + o.pubkey AS old_pubkey, o.role AS old_role, + o.removed_at IS NULL AS old_active, + n.community_id AS new_community, n.channel_id AS new_channel, + n.pubkey AS new_pubkey, n.role AS new_role, + n.removed_at IS NULL AS new_active + FROM old_members o + JOIN new_members n + ON o.community_id = n.community_id + AND o.channel_id = n.channel_id + AND o.pubkey = n.pubkey + WHERE (o.removed_at IS NULL) IS DISTINCT FROM (n.removed_at IS NULL) + OR ((o.removed_at IS NULL) AND (n.removed_at IS NULL) + AND o.role IS DISTINCT FROM n.role) + ), affected AS ( + SELECT old_community AS community_id, old_channel AS channel_id + FROM changed WHERE old_active + UNION + SELECT new_community, new_channel FROM changed WHERE new_active + ) + SELECT community_id, channel_id FROM affected + ORDER BY community_id, channel_id + LOOP + PERFORM pg_advisory_xact_lock( + channel_policy_lock_key(target.community_id, target.channel_id)); + UPDATE channel_policy_generations + SET generation = generation + 1, updated_at = NOW() + WHERE community_id = target.community_id + AND channel_id = target.channel_id + AND deleted = FALSE; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION advance_channel_policy_for_member_delete() RETURNS trigger +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; +BEGIN + FOR target IN + SELECT DISTINCT o.community_id, o.channel_id + FROM old_members o + JOIN channel_policy_generations p + ON p.community_id = o.community_id AND p.channel_id = o.channel_id + WHERE o.removed_at IS NULL AND p.deleted = FALSE + ORDER BY o.community_id, o.channel_id + LOOP + PERFORM pg_advisory_xact_lock( + channel_policy_lock_key(target.community_id, target.channel_id)); + UPDATE channel_policy_generations + SET generation = generation + 1, updated_at = NOW() + WHERE community_id = target.community_id + AND channel_id = target.channel_id + AND deleted = FALSE; + END LOOP; + RETURN NULL; +END +$$; + +CREATE FUNCTION enforce_ifc_acceptance_expiry() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF NEW.fence_expires_at <= clock_timestamp() THEN + RAISE EXCEPTION 'ifc-expired' + USING ERRCODE = '23514', CONSTRAINT = 'ifc_acceptance_unexpired'; + END IF; + RETURN NULL; +END +$$; + +CREATE FUNCTION prune_ifc_acceptances_after_insert() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + DELETE FROM ifc_event_acceptances + WHERE ctid IN ( + SELECT a.ctid + FROM ifc_event_acceptances a + JOIN (SELECT DISTINCT community_id FROM new_acceptances) n + ON n.community_id = a.community_id + WHERE a.accepted_at < clock_timestamp() - INTERVAL '24 hours' + ORDER BY a.accepted_at + LIMIT 1000 + ); + RETURN NULL; +END +$$; diff --git a/schema/schema.sql b/schema/schema.sql index 5ec8bd5829..ed4d0359b1 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -78,11 +78,14 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i functions/enforce_community_write_fence.sql \i functions/enforce_community_tombstone.sql \i functions/attach_community_write_fence.sql +\i functions/channel_policy_lock_key.sql +\i functions/maintain_channel_policy_generation.sql -- tables \i tables/public/communities.sql \i tables/public/channels.sql \i tables/public/channel_members.sql +\i tables/public/channel_policy_generations.sql \i tables/public/users.sql \i tables/public/personal_read_accounts.sql \i tables/public/personal_read_frontiers.sql @@ -135,6 +138,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i tables/public/operator_listener_outbox.sql \i tables/public/relay_operator_audit.sql \i tables/public/storage_accounting_snapshots.sql +\i tables/public/ifc_event_acceptances.sql \i tables/public/artifact_heads.sql \i tables/public/artifact_revisions.sql @@ -179,6 +183,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i triggers/workflows.sql \i triggers/artifact_heads.sql \i triggers/artifact_revisions.sql +\i triggers/channel_policy.sql -- seeds \i seeds/communities.sql diff --git a/schema/tables/public/channel_policy_generations.sql b/schema/tables/public/channel_policy_generations.sql new file mode 100644 index 0000000000..2ea63e1747 --- /dev/null +++ b/schema/tables/public/channel_policy_generations.sql @@ -0,0 +1,12 @@ +-- Monotone IFC policy coordinate retained after channel deletion. No foreign +-- key points at channels: retaining this row prevents a deleted UUID from ever +-- acquiring a fresh generation sequence inside the same community. +CREATE TABLE channel_policy_generations ( + community_id UUID NOT NULL REFERENCES communities(id) ON DELETE CASCADE, + channel_id UUID NOT NULL, + generation BIGINT NOT NULL DEFAULT 1 CHECK (generation > 0), + deleted BOOLEAN NOT NULL DEFAULT FALSE, + created_xid xid8 NOT NULL DEFAULT pg_current_xact_id(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, channel_id) +); diff --git a/schema/tables/public/channels.sql b/schema/tables/public/channels.sql index 5fe5659cfb..f8194e49e0 100644 --- a/schema/tables/public/channels.sql +++ b/schema/tables/public/channels.sql @@ -13,6 +13,7 @@ CREATE TABLE channels ( name VARCHAR(255) NOT NULL, channel_type channel_type NOT NULL DEFAULT 'stream', visibility channel_visibility NOT NULL DEFAULT 'open', + history_access TEXT NOT NULL DEFAULT 'all_current_members', description TEXT, canvas TEXT, created_by BYTEA NOT NULL, @@ -33,7 +34,8 @@ CREATE TABLE channels ( ttl_seconds INT, ttl_deadline TIMESTAMPTZ, PRIMARY KEY (community_id, id), - CONSTRAINT chk_channels_id_not_nil CHECK (id <> '00000000-0000-0000-0000-000000000000'::uuid) + CONSTRAINT chk_channels_id_not_nil CHECK (id <> '00000000-0000-0000-0000-000000000000'::uuid), + CONSTRAINT chk_channels_history_access CHECK (history_access = 'all_current_members') ); -- nip29 group id and DM participant hash are unique WITHIN a community, not globally. diff --git a/schema/tables/public/ifc_event_acceptances.sql b/schema/tables/public/ifc_event_acceptances.sql new file mode 100644 index 0000000000..af55e97038 --- /dev/null +++ b/schema/tables/public/ifc_event_acceptances.sql @@ -0,0 +1,13 @@ +-- Durable accepted-event receipt. It is intentionally independent of events +-- retention so an exact signed retry cannot become a new write after expiry. +CREATE TABLE ifc_event_acceptances ( + community_id UUID NOT NULL REFERENCES communities(id) ON DELETE CASCADE, + event_id BYTEA NOT NULL CHECK (octet_length(event_id) = 32), + event_kind INT NOT NULL, + fence_expires_at TIMESTAMPTZ NOT NULL, + accepted_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (community_id, event_id) +); + +CREATE INDEX idx_ifc_event_acceptances_accepted_at + ON ifc_event_acceptances (accepted_at); diff --git a/schema/triggers/channel_policy.sql b/schema/triggers/channel_policy.sql new file mode 100644 index 0000000000..3cc541f1ae --- /dev/null +++ b/schema/triggers/channel_policy.sql @@ -0,0 +1,58 @@ +CREATE TRIGGER trg_channels_ifc_policy_insert + AFTER INSERT ON channels + REFERENCING NEW TABLE AS new_channels + FOR EACH STATEMENT EXECUTE FUNCTION initialize_channel_policy_generation(); + +CREATE TRIGGER trg_channels_ifc_policy_identity + BEFORE UPDATE ON channels + FOR EACH ROW EXECUTE FUNCTION channel_policy_identity_immutable(); + +CREATE TRIGGER trg_channels_ifc_policy_update + AFTER UPDATE ON channels + REFERENCING OLD TABLE AS old_channels NEW TABLE AS new_channels + FOR EACH STATEMENT EXECUTE FUNCTION advance_channel_policy_for_channel_update(); + +CREATE TRIGGER trg_channels_ifc_policy_delete + AFTER DELETE ON channels + REFERENCING OLD TABLE AS old_channels + FOR EACH STATEMENT EXECUTE FUNCTION advance_channel_policy_for_channel_delete(); + +CREATE TRIGGER trg_channel_members_ifc_policy_insert + AFTER INSERT ON channel_members + REFERENCING NEW TABLE AS new_members + FOR EACH STATEMENT EXECUTE FUNCTION advance_channel_policy_for_member_insert(); + +CREATE TRIGGER trg_channel_members_ifc_policy_identity + BEFORE UPDATE ON channel_members + FOR EACH ROW EXECUTE FUNCTION channel_member_policy_identity_immutable(); + +CREATE TRIGGER trg_channel_members_ifc_policy_update + AFTER UPDATE ON channel_members + REFERENCING OLD TABLE AS old_members NEW TABLE AS new_members + FOR EACH STATEMENT EXECUTE FUNCTION advance_channel_policy_for_member_update(); + +CREATE TRIGGER trg_channel_members_ifc_policy_delete + AFTER DELETE ON channel_members + REFERENCING OLD TABLE AS old_members + FOR EACH STATEMENT EXECUTE FUNCTION advance_channel_policy_for_member_delete(); + +CREATE CONSTRAINT TRIGGER trg_ifc_acceptance_unexpired + AFTER INSERT ON ifc_event_acceptances + DEFERRABLE INITIALLY DEFERRED + FOR EACH ROW EXECUTE FUNCTION enforce_ifc_acceptance_expiry(); + +CREATE TRIGGER trg_ifc_acceptance_prune + AFTER INSERT ON ifc_event_acceptances + REFERENCING NEW TABLE AS new_acceptances + FOR EACH STATEMENT EXECUTE FUNCTION prune_ifc_acceptances_after_insert(); + +CREATE TRIGGER trg_channel_policy_generation_guard + BEFORE UPDATE OR DELETE ON channel_policy_generations + FOR EACH ROW EXECUTE FUNCTION guard_channel_policy_generation_row(); + +CREATE TRIGGER trg_ifc_acceptance_guard + BEFORE UPDATE OR DELETE ON ifc_event_acceptances + FOR EACH ROW EXECUTE FUNCTION guard_ifc_acceptance_row(); + +SELECT attach_community_write_fence('channel_policy_generations'); +SELECT attach_community_write_fence('ifc_event_acceptances');