From 4e6321097408edf905ccf29a4dc5d44e4285c4fd Mon Sep 17 00:00:00 2001 From: jm Date: Wed, 14 Oct 2026 22:18:14 +0000 Subject: [PATCH] feat(relay): complete IFC proof semantics Co-authored-by: Codex Signed-off-by: jm --- crates/buzz-db/src/runtime/migration.rs | 10 +- crates/buzz-db/src/store/ifc.rs | 335 +++++++++++++- crates/buzz-db/src/store/thread.rs | 30 ++ crates/buzz-relay/src/api/bridge.rs | 425 ++++++++++++++++++ crates/buzz-relay/src/ifc.rs | 191 +++++++- crates/buzz-relay/src/main.rs | 2 +- migrations/0062_ifc_event_origin.sql | 28 ++ .../guard_event_origin_generation.sql | 16 + schema/schema.sql | 1 + schema/tables/public/events.sql | 1 + schema/triggers/events.sql | 3 + 11 files changed, 1022 insertions(+), 20 deletions(-) create mode 100644 migrations/0062_ifc_event_origin.sql create mode 100644 schema/functions/guard_event_origin_generation.sql diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index cd309d2e10..266a9e54e2 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(), 61); + assert_eq!(migrations.len(), 62); assert_eq!(migrations[58].version, 59); assert!(migrations[58].sql.as_str().contains("buzz.push_enabled")); assert_eq!(migrations[59].version, 60); @@ -718,6 +718,14 @@ mod postgres_tests { 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_eq!(migrations[61].version, 62); + let origin = migrations[61].sql.as_str(); + assert!(origin.contains("ADD COLUMN origin_generation BIGINT")); + assert!(origin.contains("CREATE FUNCTION guard_event_origin_generation")); + let desired = crate::test_support::desired_state_schema_sql(); + assert!(desired.contains("origin_generation BIGINT")); + assert!(desired.contains("CREATE FUNCTION guard_event_origin_generation")); + assert!(desired.contains("trg_events_guard_origin_generation")); 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")); diff --git a/crates/buzz-db/src/store/ifc.rs b/crates/buzz-db/src/store/ifc.rs index ef9c494a67..5dbae246b2 100644 --- a/crates/buzz-db/src/store/ifc.rs +++ b/crates/buzz-db/src/store/ifc.rs @@ -1,6 +1,6 @@ //! Canonical policy cuts and atomic fenced publication. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use buzz_core::ifc::{IfcMemoryExpectedHead, IfcPublicationFence, IfcPublicationTarget}; use buzz_core::{CommunityId, StoredEvent}; @@ -47,7 +47,29 @@ pub struct IfcReadCut { /// Canonical policy. pub policy: IfcPolicySnapshot, /// Exact signed resources selected by the supported query. - pub resources: Vec, + pub resources: Vec, +} + +/// One signed resource and its relay-authenticated publication origin, when known. +#[derive(Debug)] +pub struct IfcReadResource { + /// Exact signed event stored by the relay. + pub event: StoredEvent, + /// Destination policy generation atomically recorded for a fenced publication. + pub origin_generation: Option, +} + +/// Existing thread traversal semantics applied inside an IFC policy read cut. +#[derive(Debug)] +pub struct IfcThreadQuery { + /// Selected thread root event ID. + pub root_event_id: Vec, + /// Maximum reply nesting depth. + pub depth_limit: u32, + /// Composite ascending cursor: timestamp seconds followed by event ID. + pub cursor: Option>, + /// Message kinds admitted by the caller's signed filter. + pub kinds: Vec, } /// Result of one fenced publication transaction. @@ -215,10 +237,13 @@ impl Db { principal_hex: &str, expected_generation: Option, query: &EventQuery, + thread_query: Option<&IfcThreadQuery>, ) -> Result { if query.community_id != community_id || query.channel_id != Some(channel_id) || query.limit.unwrap_or(IFC_MAX_RESOURCES) > IFC_MAX_RESOURCES + || (thread_query.is_some() + && !matches!(query.limit, Some(limit) if limit > 0 && limit <= IFC_MAX_RESOURCES)) { return Err(IfcError::Invalid); } @@ -244,7 +269,62 @@ impl Db { 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?; + let events = if let Some(thread) = thread_query { + if thread.root_event_id.len() != 32 + || thread.depth_limit > i32::MAX as u32 + || thread.kinds.is_empty() + || thread + .cursor + .as_ref() + .is_some_and(|cursor| cursor.len() != 40) + { + return Err(IfcError::Invalid); + } + crate::thread::get_thread_replies_filtered_on( + tx.conn(), + community_id, + &thread.root_event_id, + Some(thread.depth_limit), + query.limit.ok_or(IfcError::Invalid)? as u32, + thread.cursor.as_deref(), + Some(&thread.kinds), + ) + .await? + .into_iter() + .map(|reply| reply.stored_event) + .collect() + } else { + crate::event::query_events_on(tx.conn(), query).await? + }; + let event_ids = events + .iter() + .map(|resource| resource.event.id.as_bytes().to_vec()) + .collect::>(); + let origins = if event_ids.is_empty() { + HashMap::new() + } else { + sqlx::query_as::<_, (Vec, Option)>( + "SELECT id, origin_generation FROM events \ + WHERE community_id = $1 AND id = ANY($2::bytea[]) AND deleted_at IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(&event_ids) + .fetch_all(tx.conn()) + .await + .map_err(DbError::from)? + .into_iter() + .collect() + }; + let resources = events + .into_iter() + .map(|event| IfcReadResource { + origin_generation: origins + .get(event.event.id.as_bytes().as_slice()) + .copied() + .flatten(), + event, + }) + .collect(); tx.rollback().await?; Ok(IfcReadCut { policy, resources }) } @@ -546,6 +626,23 @@ impl Db { IfcPublicationTarget::OwnerTelemetry { .. } => return Err(IfcError::Invalid), }; + if let IfcPublicationTarget::Channel(destination) = &fence.target { + let updated = sqlx::query( + "UPDATE events SET origin_generation = $3 \ + WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL \ + AND origin_generation IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .bind(destination.generation) + .execute(tx.conn()) + .await + .map_err(DbError::from)?; + if updated.rows_affected() != 1 { + return Err(IfcError::Invalid); + } + } + let expires_at = DateTime::::from_timestamp(fence.expires_at, 0).ok_or(IfcError::Invalid)?; sqlx::query( @@ -574,7 +671,7 @@ impl Db { /// 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 \ + "SELECT DISTINCT tgname FROM pg_trigger \ WHERE NOT tgisinternal AND tgenabled = 'O' AND tgname = ANY($1::text[]) ORDER BY tgname", ) .bind([ @@ -586,6 +683,7 @@ impl Db { "trg_channels_ifc_policy_identity", "trg_channels_ifc_policy_insert", "trg_channels_ifc_policy_update", + "trg_events_guard_origin_generation", ]) .fetch_all(&self.pool) .await?; @@ -595,12 +693,16 @@ impl Db { AND EXISTS (SELECT 1 FROM information_schema.columns \ WHERE table_schema='public' AND table_name='channels' \ AND column_name='history_access') \ + AND EXISTS (SELECT 1 FROM information_schema.columns \ + WHERE table_schema='public' AND table_name='events' \ + AND column_name='origin_generation') \ 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('guard_event_origin_generation()') 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 \ @@ -614,7 +716,7 @@ impl Db { ) .fetch_one(&self.pool) .await?; - if rows.len() != 8 || !catalog_ready { + if rows.len() != 9 || !catalog_ready { return Err(DbError::InvalidData( "IFC policy catalog is incomplete".to_owned(), )); @@ -1362,12 +1464,14 @@ mod postgres_tests { &added.public_key().to_hex(), Some(2), &query, + None, ) .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); + assert_eq!(cut.resources[0].event.event.id, legacy.id); + assert_eq!(cut.resources[0].origin_generation, None); let outsider = Keys::generate(); assert!(matches!( db.ifc_read_cut( @@ -1375,13 +1479,219 @@ mod postgres_tests { channel, &outsider.public_key().to_hex(), Some(2), - &query + &query, + None, ) .await, Err(IfcError::AccessDenied) )); } + #[tokio::test] + #[ignore = "requires isolated Postgres"] + async fn fenced_origin_survives_policy_rotation_and_is_immutable() { + let db = connect().await; + let (community, channel, keys) = fixture(db.pool()).await; + let event = fenced_event( + &keys, + community, + channel, + 1, + Utc::now().timestamp() + 120, + "known origin", + ); + let fence = parse_publication_fence(&event) + .expect("parse") + .expect("fence"); + assert!(matches!( + db.publish_ifc_event( + community, + &event, + &fence, + &keys.public_key().to_hex(), + None, + None, + ) + .await + .expect("publish at generation one"), + IfcPublishOutcome::Inserted(_) + )); + 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("rotate policy"); + + let mut query = EventQuery::for_community(community); + query.channel_id = Some(channel); + query.kinds = Some(vec![KIND_STREAM_MESSAGE as i32]); + query.ids = Some(vec![event.id.as_bytes().to_vec()]); + query.limit = Some(1); + let cut = db + .ifc_read_cut( + community, + channel, + &keys.public_key().to_hex(), + Some(2), + &query, + None, + ) + .await + .expect("read after rotation"); + assert_eq!(cut.policy.generation, 2); + assert_eq!(cut.resources.len(), 1); + assert_eq!(cut.resources[0].origin_generation, Some(1)); + + let error = + sqlx::query("UPDATE events SET origin_generation=2 WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .execute(db.pool()) + .await + .expect_err("known origin must be immutable"); + assert!(error + .to_string() + .contains("event origin generation is immutable")); + } + + #[tokio::test] + #[ignore = "requires isolated Postgres"] + async fn locked_read_cut_applies_thread_depth_and_composite_cursor() { + let db = connect().await; + let (community, channel, keys) = fixture(db.pool()).await; + let channel_text = channel.to_string(); + let timestamp = Utc::now().timestamp(); + let created_at = DateTime::::from_timestamp(timestamp, 0).expect("valid timestamp"); + let root = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "root") + .tags([Tag::parse(["h", channel_text.as_str()]).expect("h tag")]) + .custom_created_at(Timestamp::from(timestamp as u64)) + .sign_with_keys(&keys) + .expect("sign root"); + db.insert_event_with_thread_metadata( + community, + &root, + Some(channel), + Some(ThreadMetadataParams { + event_id: root.id.as_bytes(), + event_created_at: created_at, + channel_id: channel, + parent_event_id: None, + parent_event_created_at: None, + root_event_id: None, + root_event_created_at: None, + depth: 0, + broadcast: true, + }), + ) + .await + .expect("insert root"); + let mut replies = Vec::new(); + for content in ["same-second-a", "same-second-b"] { + let reply = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), content) + .tags([ + Tag::parse(["h", channel_text.as_str()]).expect("h tag"), + Tag::parse(["e", root.id.to_hex().as_str()]).expect("root tag"), + ]) + .custom_created_at(Timestamp::from(timestamp as u64)) + .sign_with_keys(&keys) + .expect("sign reply"); + db.insert_event_with_thread_metadata( + community, + &reply, + Some(channel), + Some(ThreadMetadataParams { + event_id: reply.id.as_bytes(), + event_created_at: created_at, + channel_id: channel, + parent_event_id: Some(root.id.as_bytes()), + parent_event_created_at: Some(created_at), + root_event_id: Some(root.id.as_bytes()), + root_event_created_at: Some(created_at), + depth: 1, + broadcast: false, + }), + ) + .await + .expect("insert reply"); + replies.push(reply); + } + replies.sort_by_key(|event| event.id); + + let nested = EventBuilder::new(Kind::Custom(KIND_STREAM_MESSAGE as u16), "nested") + .tags([ + Tag::parse(["h", channel_text.as_str()]).expect("h tag"), + Tag::parse(["e", root.id.to_hex().as_str()]).expect("root tag"), + ]) + .custom_created_at(Timestamp::from((timestamp + 1) as u64)) + .sign_with_keys(&keys) + .expect("sign nested reply"); + db.insert_event_with_thread_metadata( + community, + &nested, + Some(channel), + Some(ThreadMetadataParams { + event_id: nested.id.as_bytes(), + event_created_at: DateTime::::from_timestamp(timestamp + 1, 0) + .expect("valid nested timestamp"), + channel_id: channel, + parent_event_id: Some(replies[0].id.as_bytes()), + parent_event_created_at: Some(created_at), + root_event_id: Some(root.id.as_bytes()), + root_event_created_at: Some(created_at), + depth: 2, + broadcast: false, + }), + ) + .await + .expect("insert nested reply"); + + let mut query = EventQuery::for_community(community); + query.channel_id = Some(channel); + query.kinds = Some(vec![KIND_STREAM_MESSAGE as i32]); + query.limit = Some(1); + let thread = IfcThreadQuery { + root_event_id: root.id.as_bytes().to_vec(), + depth_limit: 1, + cursor: None, + kinds: vec![KIND_STREAM_MESSAGE as i32], + }; + let first = db + .ifc_read_cut( + community, + channel, + &keys.public_key().to_hex(), + Some(1), + &query, + Some(&thread), + ) + .await + .expect("first thread page"); + assert_eq!(first.resources.len(), 1); + assert_eq!(first.resources[0].event.event.id, replies[0].id); + + let mut cursor = timestamp.to_be_bytes().to_vec(); + cursor.extend(replies[0].id.as_bytes()); + let after = IfcThreadQuery { + cursor: Some(cursor), + ..thread + }; + let second = db + .ifc_read_cut( + community, + channel, + &keys.public_key().to_hex(), + Some(1), + &query, + Some(&after), + ) + .await + .expect("second thread page"); + assert_eq!(second.resources.len(), 1); + assert_eq!(second.resources[0].event.event.id, replies[1].id); + assert_ne!(second.resources[0].event.event.id, nested.id); + } + #[tokio::test] #[ignore = "requires isolated Postgres"] async fn source_and_destination_generations_are_checked_independently() { @@ -1531,6 +1841,15 @@ mod postgres_tests { .await, Ok(IfcPublishOutcome::Inserted(_)) )); + let origin: Option = sqlx::query_scalar( + "SELECT origin_generation FROM events WHERE community_id=$1 AND id=$2", + ) + .bind(community.as_uuid()) + .bind(second.id.as_bytes().as_slice()) + .fetch_one(db.pool()) + .await + .expect("read canvas origin"); + assert_eq!(origin, Some(1)); } #[tokio::test] @@ -1931,7 +2250,7 @@ mod postgres_tests { let reader = keys.public_key().to_hex(); let read = tokio::spawn(async move { read_db - .ifc_read_cut(community, channel, &reader, None, &query) + .ifc_read_cut(community, channel, &reader, None, &query, None) .await }); let write_db = db.clone(); diff --git a/crates/buzz-db/src/store/thread.rs b/crates/buzz-db/src/store/thread.rs index 719861241c..6bd721fb9b 100644 --- a/crates/buzz-db/src/store/thread.rs +++ b/crates/buzz-db/src/store/thread.rs @@ -299,6 +299,29 @@ pub(crate) async fn get_thread_replies_on( depth_limit: Option, limit: u32, cursor: Option<&[u8]>, +) -> Result> { + get_thread_replies_filtered_on( + conn, + community_id, + root_event_id, + depth_limit, + limit, + cursor, + None, + ) + .await +} + +/// [`get_thread_replies_on`] with an optional event-kind restriction applied +/// before ordering and limiting. +pub(crate) async fn get_thread_replies_filtered_on( + conn: &mut sqlx::PgConnection, + community_id: CommunityId, + root_event_id: &[u8], + depth_limit: Option, + limit: u32, + cursor: Option<&[u8]>, + kinds: Option<&[i32]>, ) -> Result> { // Decode cursor bytes -> keyset (timestamp, optional event_id) for the // WHERE condition. Layout: 8-byte BE i64 seconds, then the raw event_id. @@ -363,6 +386,10 @@ pub(crate) async fn get_thread_replies_on( sql.push_str(&format!(" AND tm.depth <= ${param_idx}")); param_idx += 1; } + if kinds.is_some() { + sql.push_str(&format!(" AND e.kind = ANY(${param_idx}::int[])")); + param_idx += 1; + } match &cursor_key { Some((_, Some(_))) => { // Composite keyset: strict row comparison with an event_id tiebreak @@ -393,6 +420,9 @@ pub(crate) async fn get_thread_replies_on( if let Some(dl) = depth_limit { q = q.bind(dl as i32); } + if let Some(kinds) = kinds { + q = q.bind(kinds); + } match &cursor_key { Some((ts, Some(id))) => { q = q.bind(*ts).bind(id.clone()); diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index cec7369025..43d66caf3b 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -8295,6 +8295,8 @@ pub(crate) mod postgres_tests { assert!(attestation.verify_signature()); let content: Value = serde_json::from_str(&attestation.content).expect("parse proof"); assert_eq!(content["requester"], broker.public_key().to_hex()); + assert_eq!(content["policy"]["channel_type"], "channel"); + assert_eq!(content["policy"]["visibility"], "private"); assert_ne!(content["requester"], human.public_key().to_hex()); assert_eq!( attestation @@ -8334,6 +8336,429 @@ pub(crate) mod postgres_tests { ); } + #[tokio::test] + #[ignore = "requires Postgres and Redis"] + async fn ifc_policy_route_maps_stream_forum_and_dm_and_allows_public_nonmembers() { + use buzz_db::channel::{ChannelType, ChannelVisibility}; + use buzz_db::channel_members::MemberRole; + use nostr::Event; + use uuid::Uuid; + + let state = nip_fi_off_test_state() + .await + .expect("build HTTP query state"); + let host = format!("ifc-policy-map-{}.local", Uuid::new_v4().simple()); + let community = state + .db + .ensure_configured_community(&host) + .await + .expect("ensure community") + .id; + let creator = Keys::generate(); + let requester = Keys::generate(); + for keys in [&creator, &requester] { + state + .db + .ensure_user(community, keys.public_key().as_bytes()) + .await + .expect("ensure policy principal"); + } + + for (index, channel_type, visibility, wire_type, wire_visibility) in [ + ( + 0, + ChannelType::Stream, + ChannelVisibility::Private, + "channel", + "private", + ), + ( + 1, + ChannelType::Stream, + ChannelVisibility::Open, + "channel", + "public", + ), + ( + 2, + ChannelType::Forum, + ChannelVisibility::Private, + "channel", + "private", + ), + ( + 3, + ChannelType::Forum, + ChannelVisibility::Open, + "channel", + "public", + ), + ( + 4, + ChannelType::Dm, + ChannelVisibility::Private, + "dm", + "private", + ), + ] { + let channel = Uuid::new_v4(); + state + .db + .create_channel_with_id( + community, + channel, + &format!("ifc-policy-map-{index}-{channel}"), + channel_type, + visibility, + None, + creator.public_key().as_bytes(), + None, + ) + .await + .expect("create mapped channel"); + if visibility == ChannelVisibility::Private { + state + .db + .add_member( + community, + channel, + requester.public_key().as_bytes(), + MemberRole::Member, + Some(creator.public_key().as_bytes()), + ) + .await + .expect("authorize private reader"); + } + let generation: 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(state.db.pool()) + .await + .expect("read policy generation"); + let body = format!( + "[{{\"kinds\":[45900],\"#h\":[\"{channel}\"],\"limit\":1,\"ifc\":{{\"version\":1,\"nonce\":\"{index:032x}\",\"expected_generation\":\"{generation}\"}}}}]" + ); + let url = format!("https://{host}/query"); + let headers = make_nip98_headers(&requester, &url, "POST", body.as_bytes()); + let (status, _, response) = oneshot_request_full( + state.clone(), + "POST", + "/query", + &host, + headers, + body.as_bytes(), + ) + .await; + assert_eq!( + status, + StatusCode::OK, + "{}", + String::from_utf8_lossy(&response) + ); + let response: Value = serde_json::from_slice(&response).expect("parse IFC response"); + let attestation: Event = serde_json::from_value(response["attestation"].clone()) + .expect("parse policy attestation"); + let content: Value = + serde_json::from_str(&attestation.content).expect("parse policy content"); + assert_eq!(content["policy"]["channel_type"], wire_type); + assert_eq!(content["policy"]["visibility"], wire_visibility); + let members = content["policy"]["members"] + .as_array() + .expect("policy members"); + assert_eq!( + members + .iter() + .any(|member| member == &Value::String(requester.public_key().to_hex())), + visibility == ChannelVisibility::Private + ); + if channel_type == ChannelType::Dm { + let dm = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_STREAM_MESSAGE_V2 as u16), + "dm bootstrap", + ) + .tags([Tag::parse(["h", channel.to_string().as_str()]).expect("h tag")]) + .sign_with_keys(&creator) + .expect("sign DM message"); + state + .db + .insert_event_with_thread_metadata(community, &dm, Some(channel), None) + .await + .expect("insert DM message"); + let history_body = format!( + "[{{\"kinds\":[9,40002],\"#h\":[\"{channel}\"],\"limit\":10,\"ifc\":{{\"version\":1,\"nonce\":\"d0000000000000000000000000000000\",\"expected_generation\":\"{generation}\"}}}}]" + ); + let headers = make_nip98_headers(&requester, &url, "POST", history_body.as_bytes()); + let (status, _, response) = oneshot_request_full( + state.clone(), + "POST", + "/query", + &host, + headers, + history_body.as_bytes(), + ) + .await; + assert_eq!(status, StatusCode::OK); + let response: Value = + serde_json::from_slice(&response).expect("parse DM IFC response"); + let attestation: Event = serde_json::from_value(response["attestation"].clone()) + .expect("parse DM attestation"); + let content: Value = + serde_json::from_str(&attestation.content).expect("parse DM proof"); + assert_eq!(content["resources"][0]["event"]["id"], dm.id.to_hex()); + } + } + } + + #[tokio::test] + #[ignore = "requires Postgres and Redis"] + async fn ifc_query_route_applies_message_thread_and_strong_canvas_cursors() { + use buzz_db::channel::{ChannelType, ChannelVisibility}; + use buzz_db::event::ThreadMetadataParams; + use chrono::{DateTime, Utc}; + use nostr::Event; + use uuid::Uuid; + + async fn post_ifc( + state: Arc, + host: &str, + requester: &Keys, + body: &str, + ) -> Value { + let url = format!("https://{host}/query"); + let headers = make_nip98_headers(requester, &url, "POST", body.as_bytes()); + let (status, _, response) = + oneshot_request_full(state, "POST", "/query", host, headers, body.as_bytes()).await; + assert_eq!( + status, + StatusCode::OK, + "{}", + String::from_utf8_lossy(&response) + ); + let response: Value = serde_json::from_slice(&response).expect("parse IFC response"); + let attestation: Event = serde_json::from_value(response["attestation"].clone()) + .expect("parse IFC attestation"); + serde_json::from_str(&attestation.content).expect("parse IFC content") + } + + let state = nip_fi_off_test_state() + .await + .expect("build HTTP query state"); + let host = format!("ifc-query-cursors-{}.local", Uuid::new_v4().simple()); + let community = state + .db + .ensure_configured_community(&host) + .await + .expect("ensure community") + .id; + let requester = Keys::generate(); + state + .db + .ensure_user(community, requester.public_key().as_bytes()) + .await + .expect("ensure requester"); + let channel = Uuid::new_v4(); + state + .db + .create_channel_with_id( + community, + channel, + &format!("ifc-query-cursors-{channel}"), + ChannelType::Stream, + ChannelVisibility::Private, + None, + requester.public_key().as_bytes(), + None, + ) + .await + .expect("create channel"); + let timestamp = Utc::now().timestamp(); + let channel_text = channel.to_string(); + let mut messages = Vec::new(); + for content in ["page-a", "page-b", "page-c"] { + let event = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_STREAM_MESSAGE_V2 as u16), + content, + ) + .tags([Tag::parse(["h", channel_text.as_str()]).expect("h tag")]) + .custom_created_at(nostr::Timestamp::from(timestamp as u64)) + .sign_with_keys(&requester) + .expect("sign paged event"); + state + .db + .insert_event_with_thread_metadata(community, &event, Some(channel), None) + .await + .expect("insert paged event"); + messages.push(event); + } + messages.sort_by_key(|event| event.id); + let first_page = format!( + "[{{\"kinds\":[9,40002],\"#h\":[\"{channel}\"],\"limit\":1,\"until\":{timestamp},\"ifc\":{{\"version\":1,\"nonce\":\"10000000000000000000000000000000\"}}}}]" + ); + let first = post_ifc(state.clone(), &host, &requester, &first_page).await; + assert_eq!( + first["resources"][0]["event"]["id"], + messages[0].id.to_hex() + ); + let next_page = format!( + "[{{\"kinds\":[9,40002],\"#h\":[\"{channel}\"],\"limit\":1,\"until\":{timestamp},\"before_id\":\"{}\",\"ifc\":{{\"version\":1,\"nonce\":\"20000000000000000000000000000000\"}}}}]", + messages[0].id.to_hex() + ); + let next = post_ifc(state.clone(), &host, &requester, &next_page).await; + assert_eq!(next["resources"][0]["event"]["id"], messages[1].id.to_hex()); + + let root = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_STREAM_MESSAGE_V2 as u16), + "root", + ) + .tags([Tag::parse(["h", channel_text.as_str()]).expect("h tag")]) + .custom_created_at(nostr::Timestamp::from((timestamp + 1) as u64)) + .sign_with_keys(&requester) + .expect("sign root"); + let root_created = + DateTime::::from_timestamp(timestamp + 1, 0).expect("valid root timestamp"); + state + .db + .insert_event_with_thread_metadata( + community, + &root, + Some(channel), + Some(ThreadMetadataParams { + event_id: root.id.as_bytes(), + event_created_at: root_created, + channel_id: channel, + parent_event_id: None, + parent_event_created_at: None, + root_event_id: None, + root_event_created_at: None, + depth: 0, + broadcast: true, + }), + ) + .await + .expect("insert root"); + let reply_timestamp = timestamp + 2; + let reply_created = + DateTime::::from_timestamp(reply_timestamp, 0).expect("valid reply timestamp"); + let mut replies = Vec::new(); + for content in ["reply-a", "reply-b"] { + let reply = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_STREAM_MESSAGE_V2 as u16), + content, + ) + .tags([ + Tag::parse(["h", channel_text.as_str()]).expect("h tag"), + Tag::parse(["e", root.id.to_hex().as_str()]).expect("root tag"), + ]) + .custom_created_at(nostr::Timestamp::from(reply_timestamp as u64)) + .sign_with_keys(&requester) + .expect("sign reply"); + state + .db + .insert_event_with_thread_metadata( + community, + &reply, + Some(channel), + Some(ThreadMetadataParams { + event_id: reply.id.as_bytes(), + event_created_at: reply_created, + channel_id: channel, + parent_event_id: Some(root.id.as_bytes()), + parent_event_created_at: Some(root_created), + root_event_id: Some(root.id.as_bytes()), + root_event_created_at: Some(root_created), + depth: 1, + broadcast: false, + }), + ) + .await + .expect("insert reply"); + replies.push(reply); + } + replies.sort_by_key(|event| event.id); + let nested = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_STREAM_MESSAGE_V2 as u16), + "nested", + ) + .tags([ + Tag::parse(["h", channel_text.as_str()]).expect("h tag"), + Tag::parse(["e", root.id.to_hex().as_str()]).expect("root tag"), + ]) + .custom_created_at(nostr::Timestamp::from((reply_timestamp + 1) as u64)) + .sign_with_keys(&requester) + .expect("sign nested reply"); + state + .db + .insert_event_with_thread_metadata( + community, + &nested, + Some(channel), + Some(ThreadMetadataParams { + event_id: nested.id.as_bytes(), + event_created_at: DateTime::::from_timestamp(reply_timestamp + 1, 0) + .expect("valid nested timestamp"), + channel_id: channel, + parent_event_id: Some(replies[0].id.as_bytes()), + parent_event_created_at: Some(reply_created), + root_event_id: Some(root.id.as_bytes()), + root_event_created_at: Some(root_created), + depth: 2, + broadcast: false, + }), + ) + .await + .expect("insert nested reply"); + + let root_id = root.id.to_hex(); + let thread_first = format!( + "[{{\"kinds\":[9,40002],\"#h\":[\"{channel}\"],\"#e\":[\"{root_id}\"],\"limit\":1,\"depth_limit\":1,\"ifc\":{{\"version\":1,\"nonce\":\"30000000000000000000000000000000\"}}}}]" + ); + let first = post_ifc(state.clone(), &host, &requester, &thread_first).await; + assert_eq!(first["resources"][0]["event"]["id"], replies[0].id.to_hex()); + let thread_after = format!( + "[{{\"kinds\":[9,40002],\"#h\":[\"{channel}\"],\"#e\":[\"{root_id}\"],\"limit\":1,\"depth_limit\":1,\"thread_cursor\":{reply_timestamp},\"thread_cursor_id\":\"{}\",\"ifc\":{{\"version\":1,\"nonce\":\"40000000000000000000000000000000\"}}}}]", + replies[0].id.to_hex() + ); + let after = post_ifc(state.clone(), &host, &requester, &thread_after).await; + assert_eq!(after["resources"][0]["event"]["id"], replies[1].id.to_hex()); + let selected = format!( + "[{{\"kinds\":[9,40002],\"#h\":[\"{channel}\"],\"ids\":[\"{}\"],\"limit\":1,\"ifc\":{{\"version\":1,\"nonce\":\"50000000000000000000000000000000\"}}}}]", + nested.id.to_hex() + ); + let selected = post_ifc(state.clone(), &host, &requester, &selected).await; + assert_eq!(selected["resources"][0]["event"]["id"], nested.id.to_hex()); + let root_lookup = format!( + "[{{\"kinds\":[9,40002],\"#h\":[\"{channel}\"],\"ids\":[\"{root_id}\"],\"limit\":1,\"ifc\":{{\"version\":1,\"nonce\":\"60000000000000000000000000000000\"}}}}]" + ); + let root_result = post_ifc(state.clone(), &host, &requester, &root_lookup).await; + assert_eq!(root_result["resources"][0]["event"]["id"], root_id); + + let canvas = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_CANVAS as u16), + "strong canvas", + ) + .tags([Tag::parse(["h", channel_text.as_str()]).expect("h tag")]) + .custom_created_at(nostr::Timestamp::from((reply_timestamp + 2) as u64)) + .sign_with_keys(&requester) + .expect("sign canvas"); + state + .db + .insert_event_with_thread_metadata(community, &canvas, Some(channel), None) + .await + .expect("insert canvas"); + let canvas_query = format!( + "[{{\"kinds\":[40100],\"#h\":[\"{channel}\"],\"ids\":[\"{}\"],\"limit\":1,\"consistency\":\"strong\",\"ifc\":{{\"version\":1,\"nonce\":\"70000000000000000000000000000000\"}}}}]", + canvas.id.to_hex() + ); + let canvas_result = post_ifc(state, &host, &requester, &canvas_query).await; + assert_eq!( + canvas_result["resources"][0]["event"]["id"], + canvas.id.to_hex() + ); + } + // ── Bridge dispatch test: writer-pin routes to the writer pool ──────────── // // Exercises the catchall dispatch loop's `ReadRoute` match at the shipping diff --git a/crates/buzz-relay/src/ifc.rs b/crates/buzz-relay/src/ifc.rs index 42d3f8a9f3..d79c0d75d1 100644 --- a/crates/buzz-relay/src/ifc.rs +++ b/crates/buzz-relay/src/ifc.rs @@ -7,7 +7,7 @@ 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::ifc::{IfcError, IfcThreadQuery}; use buzz_db::EventQuery; use chrono::{DateTime, Utc}; use nostr::{Event, EventBuilder, Kind, PublicKey, Tag}; @@ -32,6 +32,7 @@ struct Request { expected_generation: Option, policy_only: bool, query: EventQuery, + thread_query: Option, } #[derive(Serialize)] @@ -100,6 +101,17 @@ fn decode_hex32(value: &str, field: &str) -> Result, (StatusCode, String hex::decode(value).map_err(|_| bad(format!("{field} entries must be hex"))) } +fn decode_canonical_hex32(value: &str, field: &str) -> Result, (StatusCode, String)> { + if value.len() != 64 + || !value + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + { + return Err(bad(format!("{field} must be 64 lowercase hex characters"))); + } + hex::decode(value).map_err(|_| bad(format!("{field} 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 @@ -110,7 +122,20 @@ fn parse_request(body: &[u8], community_id: CommunityId) -> Result Result Result Result { + let seconds = timestamp + .as_i64() + .filter(|seconds| *seconds >= 0) + .ok_or_else(|| bad("thread_cursor must be a nonnegative integer"))?; + let event_id = event_id + .as_str() + .ok_or_else(|| bad("thread_cursor_id must be a string")) + .and_then(|value| decode_canonical_hex32(value, "thread_cursor_id"))?; + let mut cursor = seconds.to_be_bytes().to_vec(); + cursor.extend(event_id); + Some(cursor) + } + (None, None) => None, + _ => return Err(bad("thread cursor fields must be supplied together")), + }; + let thread_query = if let Some(depth_limit) = depth_limit { + let root = query + .e_tags + .as_ref() + .filter(|roots| roots.len() == 1) + .and_then(|roots| roots.first()) + .ok_or_else(|| bad("thread traversal requires exactly one #e root"))?; + let root_event_id = decode_canonical_hex32(root, "#e")?; + if query.ids.is_some() + || query.authors.is_some() + || query.since.is_some() + || query.until.is_some() + || query.before_id.is_some() + || kinds.is_empty() + || kinds + .iter() + .any(|kind| ![KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_V2].contains(kind)) + { + return Err(bad("unsupported IFC thread traversal")); + } + Some(IfcThreadQuery { + root_event_id, + depth_limit, + cursor, + kinds: kinds.iter().map(|kind| *kind as i32).collect(), + }) + } else { + if cursor.is_some() { + return Err(bad("thread cursor requires depth_limit")); + } + None + }; Ok(Request { channel_id, nonce, expected_generation, policy_only, query, + thread_query, }) } +fn wire_channel_type(value: &str) -> Result<&'static str, (StatusCode, String)> { + match value { + "stream" | "forum" | "workflow" => Ok("channel"), + "dm" => Ok("dm"), + _ => Err((StatusCode::SERVICE_UNAVAILABLE, "ifc-invalid".to_owned())), + } +} + +fn wire_visibility(value: &str) -> Result<&'static str, (StatusCode, String)> { + match value { + "open" => Ok("public"), + "private" => Ok("private"), + _ => Err((StatusCode::SERVICE_UNAVAILABLE, "ifc-invalid".to_owned())), + } +} + fn parse_timestamp(value: &Value, field: &str) -> Result, (StatusCode, String)> { let seconds = value .as_i64() @@ -287,6 +411,7 @@ async fn query_inner( &requester_hex, request.expected_generation, &request.query, + request.thread_query.as_ref(), ) .await .map_err(map_db_error)?; @@ -294,15 +419,17 @@ async fn query_inner( 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() + if resource.event.channel_id != Some(request.channel_id) + || !HISTORY_KINDS.contains(&(resource.event.event.kind.as_u16() as u32)) + || !resource.event.event.verify_id() + || !resource.event.event.verify_signature() { return Err((StatusCode::SERVICE_UNAVAILABLE, "ifc-invalid".to_owned())); } } let generation = cut.policy.generation.to_string(); + let channel_type = wire_channel_type(&cut.policy.channel_type)?; + let visibility = wire_visibility(&cut.policy.visibility)?; let content = Content { version: 1, community_id: tenant.community().to_string(), @@ -312,8 +439,8 @@ async fn query_inner( policy: Policy { channel_id: cut.policy.channel_id.to_string(), generation: generation.clone(), - channel_type: &cut.policy.channel_type, - visibility: &cut.policy.visibility, + channel_type, + visibility, history_access: &cut.policy.history_access, archived: cut.policy.archived, members: &cut.policy.members, @@ -322,9 +449,9 @@ async fn query_inner( .resources .iter() .map(|resource| Resource { - event: &resource.event, + event: &resource.event.event, effective_generation: generation.clone(), - origin_generation: None, + origin_generation: resource.origin_generation.map(|value| value.to_string()), }) .collect(), }; @@ -482,6 +609,50 @@ mod tests { assert_eq!(request.query.kinds, Some(vec![9, 40002, 40100])); } + #[test] + fn adapter_pagination_and_strong_read_fields_are_applied() { + let body = br##"[{"kinds":[40100],"#h":["11111111-1111-4111-8111-111111111111"],"limit":1,"until":42,"before_id":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","consistency":"strong","ifc":{"version":1,"nonce":"0123456789abcdef0123456789abcdef"}}]"##; + let request = parse_request(body, CommunityId::from_uuid(Uuid::nil())).expect("request"); + assert_eq!(request.query.until.map(|value| value.timestamp()), Some(42)); + assert_eq!(request.query.before_id, Some(vec![0xaa; 32])); + assert!(request.thread_query.is_none()); + + for rejected in [ + br##"[{"kinds":[40100],"#h":["11111111-1111-4111-8111-111111111111"],"limit":1,"consistency":"eventual","ifc":{"version":1,"nonce":"0123456789abcdef0123456789abcdef"}}]"##.as_slice(), + br##"[{"kinds":[9],"#h":["11111111-1111-4111-8111-111111111111"],"limit":1,"before_id":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","ifc":{"version":1,"nonce":"0123456789abcdef0123456789abcdef"}}]"##.as_slice(), + ] { + assert!(parse_request(rejected, CommunityId::from_uuid(Uuid::nil())).is_err()); + } + } + + #[test] + fn adapter_thread_cursor_is_composite_and_requires_depth_traversal() { + let root = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + let after = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + let body = format!( + "[{{\"kinds\":[9,40002],\"#h\":[\"11111111-1111-4111-8111-111111111111\"],\"#e\":[\"{root}\"],\"limit\":13,\"depth_limit\":2147483647,\"thread_cursor\":42,\"thread_cursor_id\":\"{after}\",\"ifc\":{{\"version\":1,\"nonce\":\"0123456789abcdef0123456789abcdef\"}}}}]" + ); + let request = parse_request(body.as_bytes(), CommunityId::from_uuid(Uuid::nil())) + .expect("thread request"); + let thread = request.thread_query.expect("thread traversal"); + assert_eq!(thread.root_event_id, vec![0xaa; 32]); + assert_eq!(thread.depth_limit, i32::MAX as u32); + assert_eq!(&thread.cursor.expect("cursor")[..8], &42i64.to_be_bytes()); + assert_eq!(thread.kinds, vec![9, 40002]); + } + + #[test] + fn policy_wire_mapping_is_closed_and_canonical() { + for database_value in ["stream", "forum", "workflow"] { + assert_eq!(wire_channel_type(database_value), Ok("channel")); + } + assert_eq!(wire_channel_type("dm"), Ok("dm")); + assert_eq!(wire_visibility("open"), Ok("public")); + assert_eq!(wire_visibility("private"), Ok("private")); + assert!(wire_channel_type("unknown").is_err()); + assert!(wire_visibility("unknown").is_err()); + } + #[test] fn unknown_ifc_fields_and_noncanonical_generation_are_rejected() { for body in [ diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 0abec62242..7979d3d903 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1031,7 +1031,7 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { 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}" + "IFC policy catalog is unsafe; apply or repair migrations through 0062 before starting this relay: {error}" )); } } diff --git a/migrations/0062_ifc_event_origin.sql b/migrations/0062_ifc_event_origin.sql new file mode 100644 index 0000000000..afa8206ca2 --- /dev/null +++ b/migrations/0062_ifc_event_origin.sql @@ -0,0 +1,28 @@ +-- Durable relay-authenticated destination generation for newly fenced channel +-- and canvas publications. Legacy and ordinary publications remain unknown. + +ALTER TABLE events + ADD COLUMN origin_generation BIGINT, + ADD CONSTRAINT chk_events_origin_generation + CHECK (origin_generation IS NULL OR origin_generation > 0); + +CREATE FUNCTION guard_event_origin_generation() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'INSERT' AND NEW.origin_generation IS NOT NULL THEN + RAISE EXCEPTION 'event origin generation must be assigned after fenced validation' + USING ERRCODE = '23514'; + END IF; + IF TG_OP = 'UPDATE' + AND OLD.origin_generation IS NOT NULL + AND NEW.origin_generation IS DISTINCT FROM OLD.origin_generation THEN + RAISE EXCEPTION 'event origin generation is immutable' + USING ERRCODE = '23514'; + END IF; + RETURN NEW; +END +$$; + +CREATE TRIGGER trg_events_guard_origin_generation + BEFORE INSERT OR UPDATE OF origin_generation ON events + FOR EACH ROW EXECUTE FUNCTION guard_event_origin_generation(); diff --git a/schema/functions/guard_event_origin_generation.sql b/schema/functions/guard_event_origin_generation.sql new file mode 100644 index 0000000000..f73cfb5035 --- /dev/null +++ b/schema/functions/guard_event_origin_generation.sql @@ -0,0 +1,16 @@ +CREATE FUNCTION guard_event_origin_generation() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'INSERT' AND NEW.origin_generation IS NOT NULL THEN + RAISE EXCEPTION 'event origin generation must be assigned after fenced validation' + USING ERRCODE = '23514'; + END IF; + IF TG_OP = 'UPDATE' + AND OLD.origin_generation IS NOT NULL + AND NEW.origin_generation IS DISTINCT FROM OLD.origin_generation THEN + RAISE EXCEPTION 'event origin generation is immutable' + USING ERRCODE = '23514'; + END IF; + RETURN NEW; +END +$$; diff --git a/schema/schema.sql b/schema/schema.sql index ed4d0359b1..8b2d5f5fe8 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -67,6 +67,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i functions/enqueue_push_match_job.sql \i functions/refresh_channel_ttl_after_event_insert.sql \i functions/guard_channel_roster_snapshot.sql +\i functions/guard_event_origin_generation.sql \i functions/events_created_at_floor_guard.sql \i functions/prevent_community_deletion_request_retargeting.sql \i functions/prevent_community_deletion_approval_removal.sql diff --git a/schema/tables/public/events.sql b/schema/tables/public/events.sql index 10c1adc19e..4c3f9d9c45 100644 --- a/schema/tables/public/events.sql +++ b/schema/tables/public/events.sql @@ -37,6 +37,7 @@ CREATE TABLE events ( d_tag TEXT, not_before BIGINT, delivered_at BIGINT, + origin_generation BIGINT CHECK (origin_generation IS NULL OR origin_generation > 0), PRIMARY KEY (community_id, created_at, id) ) PARTITION BY RANGE (created_at); diff --git a/schema/triggers/events.sql b/schema/triggers/events.sql index 5491d1631d..eeb1feff0a 100644 --- a/schema/triggers/events.sql +++ b/schema/triggers/events.sql @@ -9,6 +9,9 @@ FOR EACH ROW EXECUTE FUNCTION refresh_channel_ttl_after_event_insert(); CREATE TRIGGER trg_events_guard_channel_roster_snapshot BEFORE INSERT ON events FOR EACH ROW EXECUTE FUNCTION guard_channel_roster_snapshot(); +CREATE TRIGGER trg_events_guard_origin_generation + BEFORE INSERT OR UPDATE OF origin_generation ON events + FOR EACH ROW EXECUTE FUNCTION guard_event_origin_generation(); -- INSERT OR UPDATE OF: an UPDATE can move a previously exempt row into the -- guarded set (channel_id NULL -> NOT NULL) or move a channel row's