diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index e984b2a836..2564f116c3 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -493,6 +493,7 @@ mod postgres_tests { "relay_banners", "relay_banner_communities", "relay_banner_user_state", + "relay_banner_view_acks", ] { if normalized[insert_pos..].contains(&format!("'{value}'")) { globals.insert(value.to_owned()); @@ -1833,6 +1834,7 @@ mod postgres_tests { expected_fences.remove("rate_limit_violations"); expected_fences.insert("relay_banner_communities".to_owned()); expected_fences.insert("relay_banner_user_state".to_owned()); + expected_fences.insert("relay_banner_view_acks".to_owned()); assert_eq!( expected_fences, schema.fence_attachments, "write-fence attachment targets differ after recovery policy" @@ -2275,6 +2277,7 @@ mod postgres_tests { "relay_banners", "relay_banner_communities", "relay_banner_user_state", + "relay_banner_view_acks", ] { assert_eq!( columns(&desired, table).await, diff --git a/crates/buzz-db/src/store/relay_banners.rs b/crates/buzz-db/src/store/relay_banners.rs index df623a17cf..2d4e71660a 100644 --- a/crates/buzz-db/src/store/relay_banners.rs +++ b/crates/buzz-db/src/store/relay_banners.rs @@ -9,6 +9,7 @@ use buzz_datastore_tracing::datastore_span; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use sqlx::{Postgres, Row as _, Transaction}; +use std::collections::HashSet; use uuid::Uuid; use crate::{Db, Result}; @@ -139,6 +140,8 @@ pub enum RelayBannerViewOutcome { Accepted { /// Display count after accepting this view. display_count: i32, + /// True only when this acknowledgement consumed a new display. + changed: bool, }, /// The banner was already permanently dismissed by this user. Dismissed, @@ -188,6 +191,12 @@ impl RelayBannerUpsert { "relay banner community scope must be non-empty".to_owned(), )); } + let mut seen = HashSet::with_capacity(ids.len()); + if ids.iter().any(|id| !seen.insert(*id)) { + return Err(crate::DbError::InvalidData( + "relay banner community scope must not contain duplicates".to_owned(), + )); + } } Ok(()) } @@ -203,16 +212,18 @@ async fn acquire_banner_lock(tx: &mut Transaction<'_, Postgres>) -> Result<()> { async fn banner_internal_id_for_public_id( tx: &mut Transaction<'_, Postgres>, public_id: Uuid, + require_active: bool, ) -> Result> { sqlx::query_scalar::<_, i64>( r#" SELECT id FROM relay_banners WHERE public_id = $1 - AND disabled_at IS NULL + AND (NOT $2 OR disabled_at IS NULL) "#, ) .bind(public_id) + .bind(require_active) .fetch_optional(&mut **tx) .await .map_err(Into::into) @@ -249,6 +260,47 @@ async fn banner_targets_community( Ok(eligible) } +async fn relay_banner_user_eligible_in_tx( + tx: &mut Transaction<'_, Postgres>, + banner_id: i64, + community_id: CommunityId, + pubkey: &[u8], + require_active: bool, +) -> Result { + let eligible = sqlx::query_scalar::<_, bool>( + r#" + SELECT EXISTS ( + SELECT 1 + FROM relay_banners b + LEFT JOIN relay_banner_user_state us + ON us.banner_id = b.id + AND us.community_id = $2 + AND us.pubkey = $3 + WHERE b.id = $1 + AND (NOT $4 OR b.disabled_at IS NULL) + AND ( + b.target_all_communities + OR EXISTS ( + SELECT 1 + FROM relay_banner_communities bc + WHERE bc.banner_id = b.id + AND bc.community_id = $2 + ) + ) + AND us.dismissed_at IS NULL + AND COALESCE(us.display_count, 0) < b.max_displays + ) + "#, + ) + .bind(banner_id) + .bind(community_id.as_uuid()) + .bind(pubkey) + .bind(require_active) + .fetch_one(&mut **tx) + .await?; + Ok(eligible) +} + impl Db { /// Lists active communities for the admin banner community picker. #[datastore_span(name = "admin_list_banner_communities", system = "postgresql")] @@ -318,6 +370,44 @@ impl Db { rows.into_iter().map(row_to_banner).collect() } + /// Returns a banner by its public id, including disabled history. + #[datastore_span(name = "relay_banner_by_public_id", system = "postgresql")] + pub async fn relay_banner_by_public_id( + &self, + public_id: Uuid, + ) -> Result> { + let mut connection = crate::observability::acquire_writer( + &self.pool, + crate::observability::WriterOperation::Authorization, + ) + .await?; + let row = sqlx::query( + r#" + SELECT + b.id, + b.public_id, + b.severity, + b.message, + b.max_displays, + b.target_all_communities, + b.created_by, + b.created_at, + b.updated_at, + b.disabled_at, + COALESCE(array_agg(bc.community_id ORDER BY bc.community_id) + FILTER (WHERE bc.community_id IS NOT NULL), ARRAY[]::uuid[]) AS community_ids + FROM relay_banners b + LEFT JOIN relay_banner_communities bc ON bc.banner_id = b.id + WHERE b.public_id = $1 + GROUP BY b.id + "#, + ) + .bind(public_id) + .fetch_optional(&mut *connection) + .await?; + row.map(row_to_banner).transpose() + } + /// Returns the currently active deployment banner, if any. #[datastore_span(name = "admin_get_active_relay_banner", system = "postgresql")] pub async fn admin_get_active_relay_banner(&self) -> Result> { @@ -439,7 +529,10 @@ impl Db { /// Disables the currently active deployment banner, preserving history and state. #[datastore_span(name = "admin_disable_active_relay_banner", system = "postgresql")] - pub async fn admin_disable_active_relay_banner(&self, actor_pubkey: &[u8]) -> Result { + pub async fn admin_disable_active_relay_banner( + &self, + actor_pubkey: &[u8], + ) -> Result> { let connection = crate::observability::acquire_writer( &self.pool, crate::observability::WriterOperation::Authorization, @@ -447,18 +540,79 @@ impl Db { .await?; let mut tx = sqlx::Transaction::begin(connection, None).await?; acquire_banner_lock(&mut tx).await?; - let result = sqlx::query( + let Some(active) = sqlx::query( + r#" + SELECT + b.id, + b.public_id, + b.severity, + b.message, + b.max_displays, + b.target_all_communities, + b.created_by, + b.created_at, + b.updated_at, + b.disabled_at, + COALESCE(array_agg(bc.community_id ORDER BY bc.community_id) + FILTER (WHERE bc.community_id IS NOT NULL), ARRAY[]::uuid[]) AS community_ids + FROM relay_banners b + LEFT JOIN relay_banner_communities bc ON bc.banner_id = b.id + WHERE b.disabled_at IS NULL + GROUP BY b.id + ORDER BY b.id DESC + LIMIT 1 + "#, + ) + .fetch_optional(&mut *tx) + .await? + else { + tx.commit().await?; + return Ok(None); + }; + let mut active = row_to_banner(active)?; + let updated = sqlx::query( r#" UPDATE relay_banners SET disabled_at = now(), disabled_by = $1, updated_at = now(), updated_by = $1 - WHERE disabled_at IS NULL + WHERE id = $2 AND disabled_at IS NULL + RETURNING updated_at, disabled_at "#, ) .bind(actor_pubkey) - .execute(&mut *tx) + .bind(active.id) + .fetch_one(&mut *tx) + .await?; + active.updated_at = updated.try_get("updated_at")?; + active.disabled_at = updated.try_get("disabled_at")?; + tx.commit().await?; + Ok(Some(active)) + } + + /// Returns whether this user would receive the banner in this community. + #[datastore_span(name = "relay_banner_user_eligible", system = "postgresql")] + pub async fn relay_banner_user_eligible( + &self, + banner_id: i64, + community_id: CommunityId, + pubkey: &[u8], + require_active: bool, + ) -> Result { + let connection = crate::observability::acquire_writer( + &self.pool, + crate::observability::WriterOperation::Authorization, + ) + .await?; + let mut tx = sqlx::Transaction::begin(connection, None).await?; + let eligible = relay_banner_user_eligible_in_tx( + &mut tx, + banner_id, + community_id, + pubkey, + require_active, + ) .await?; tx.commit().await?; - Ok(result.rows_affected() > 0) + Ok(eligible) } /// Resolves the active banner eligible for this user in this community. @@ -522,6 +676,7 @@ impl Db { community_id: CommunityId, public_id: Uuid, pubkey: &[u8], + view_id: Uuid, ) -> Result { let connection = crate::observability::acquire_writer( &self.pool, @@ -529,7 +684,8 @@ impl Db { ) .await?; let mut tx = sqlx::Transaction::begin(connection, None).await?; - let Some(banner_id) = banner_internal_id_for_public_id(&mut tx, public_id).await? else { + let Some(banner_id) = banner_internal_id_for_public_id(&mut tx, public_id, true).await? + else { tx.rollback().await?; return Ok(RelayBannerViewOutcome::NotEligible); }; @@ -538,6 +694,60 @@ impl Db { return Ok(RelayBannerViewOutcome::NotEligible); } + let inserted_view = sqlx::query_scalar::<_, bool>( + r#" + INSERT INTO relay_banner_view_acks + (banner_id, community_id, pubkey, view_id) + SELECT $1, $2, $3, $4 + WHERE EXISTS ( + SELECT 1 + FROM relay_banners b + WHERE b.id = $1 + AND b.disabled_at IS NULL + AND (b.target_all_communities OR EXISTS ( + SELECT 1 + FROM relay_banner_communities bc + WHERE bc.banner_id = b.id AND bc.community_id = $2 + )) + ) + ON CONFLICT (banner_id, community_id, pubkey, view_id) DO NOTHING + RETURNING TRUE + "#, + ) + .bind(banner_id) + .bind(community_id.as_uuid()) + .bind(pubkey) + .bind(view_id) + .fetch_optional(&mut *tx) + .await? + .unwrap_or(false); + + if !inserted_view { + let state = sqlx::query( + r#" + SELECT us.display_count, us.dismissed_at IS NOT NULL AS dismissed + FROM relay_banner_user_state us + WHERE us.banner_id = $1 AND us.community_id = $2 AND us.pubkey = $3 + "#, + ) + .bind(banner_id) + .bind(community_id.as_uuid()) + .bind(pubkey) + .fetch_optional(&mut *tx) + .await?; + tx.commit().await?; + return Ok(match state { + Some(row) if row.try_get::("dismissed")? => { + RelayBannerViewOutcome::Dismissed + } + Some(row) => RelayBannerViewOutcome::Accepted { + display_count: row.try_get("display_count")?, + changed: false, + }, + None => RelayBannerViewOutcome::NotEligible, + }); + } + let row = sqlx::query( r#" INSERT INTO relay_banner_user_state @@ -568,6 +778,7 @@ impl Db { let outcome = if let Some(row) = row { RelayBannerViewOutcome::Accepted { display_count: row.try_get("display_count")?, + changed: true, } } else { let state = sqlx::query( @@ -592,6 +803,13 @@ impl Db { None => RelayBannerViewOutcome::NotEligible, } }; + if !matches!( + outcome, + RelayBannerViewOutcome::Accepted { changed: true, .. } + ) { + tx.rollback().await?; + return Ok(outcome); + } tx.commit().await?; Ok(outcome) } @@ -610,7 +828,8 @@ impl Db { ) .await?; let mut tx = sqlx::Transaction::begin(connection, None).await?; - let Some(banner_id) = banner_internal_id_for_public_id(&mut tx, public_id).await? else { + let Some(banner_id) = banner_internal_id_for_public_id(&mut tx, public_id, true).await? + else { tx.rollback().await?; return Ok(RelayBannerDismissOutcome::NotEligible); }; @@ -686,17 +905,21 @@ mod tests { Db::from_pool(pool) } - async fn make_community(pool: &PgPool) -> CommunityId { + pub(crate) async fn insert_test_community(pool: &PgPool, host_prefix: &str) -> CommunityId { let id = Uuid::new_v4(); sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") .bind(id) - .bind(format!("banner-{}.example", id.simple())) + .bind(format!("{host_prefix}-{}.example", id.simple())) .execute(pool) .await .expect("insert community"); CommunityId::from_uuid(id) } + async fn make_community(pool: &PgPool) -> CommunityId { + insert_test_community(pool, "banner").await + } + fn actor(byte: u8) -> Vec { vec![byte; 32] } @@ -770,6 +993,60 @@ mod tests { .is_none()); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn duplicate_community_scope_is_rejected() { + let db = setup_db().await; + let community = make_community(&db.pool).await; + let result = db + .admin_upsert_relay_banner(RelayBannerUpsert { + severity: RelayBannerSeverity::Info, + message: "duplicates".to_owned(), + max_displays: 1, + scope: RelayBannerScope::Communities(vec![community, community]), + actor_pubkey: actor(15), + }) + .await; + + assert!( + matches!(result, Err(crate::DbError::InvalidData(message)) if message.contains("duplicates")) + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn disabled_banner_remains_eligible_when_active_requirement_is_lifted() { + let db = setup_db().await; + let community = make_community(&db.pool).await; + let user = actor(16); + let banner = db + .admin_upsert_relay_banner(RelayBannerUpsert { + severity: RelayBannerSeverity::Warning, + message: "disable eligibility".to_owned(), + max_displays: 2, + scope: RelayBannerScope::AllCommunities, + actor_pubkey: actor(17), + }) + .await + .expect("insert banner"); + let disabled = db + .admin_disable_active_relay_banner(&actor(17)) + .await + .expect("disable banner") + .expect("active banner"); + + assert_eq!(disabled.id, banner.id); + assert!(disabled.disabled_at.is_some()); + assert!(db + .relay_banner_user_eligible(banner.id, community, &user, false) + .await + .expect("pre-disable eligibility lookup")); + assert!(!db + .relay_banner_user_eligible(banner.id, community, &user, true) + .await + .expect("active eligibility lookup")); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn view_ack_consumes_display_and_then_exhausts() { @@ -806,13 +1083,16 @@ mod tests { "repeated delivery lookups without view ack must not exhaust max_displays" ); assert_eq!( - db.ack_relay_banner_view(community, banner.public_id, &user) + db.ack_relay_banner_view(community, banner.public_id, &user, Uuid::new_v4()) .await .expect("first view"), - RelayBannerViewOutcome::Accepted { display_count: 1 } + RelayBannerViewOutcome::Accepted { + display_count: 1, + changed: true + } ); assert_eq!( - db.ack_relay_banner_view(community, banner.public_id, &user) + db.ack_relay_banner_view(community, banner.public_id, &user, Uuid::new_v4()) .await .expect("second view"), RelayBannerViewOutcome::Exhausted { display_count: 1 } @@ -824,6 +1104,135 @@ mod tests { .is_none()); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn view_ack_same_key_retry_does_not_consume_again() { + let db = setup_db().await; + let community = make_community(&db.pool).await; + let user = actor(9); + let view_id = Uuid::new_v4(); + let banner = db + .admin_upsert_relay_banner(RelayBannerUpsert { + severity: RelayBannerSeverity::Info, + message: "retry".to_owned(), + max_displays: 2, + scope: RelayBannerScope::AllCommunities, + actor_pubkey: actor(10), + }) + .await + .expect("insert banner"); + + assert_eq!( + db.ack_relay_banner_view(community, banner.public_id, &user, view_id) + .await + .expect("first view"), + RelayBannerViewOutcome::Accepted { + display_count: 1, + changed: true + } + ); + assert_eq!( + db.ack_relay_banner_view(community, banner.public_id, &user, view_id) + .await + .expect("retry view"), + RelayBannerViewOutcome::Accepted { + display_count: 1, + changed: false + } + ); + assert!(db + .active_relay_banner_for_user(community, &user) + .await + .expect("still eligible after duplicate") + .is_some()); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn distinct_view_keys_increment_until_cap() { + let db = setup_db().await; + let community = make_community(&db.pool).await; + let user = actor(11); + let banner = db + .admin_upsert_relay_banner(RelayBannerUpsert { + severity: RelayBannerSeverity::Warning, + message: "cap".to_owned(), + max_displays: 2, + scope: RelayBannerScope::AllCommunities, + actor_pubkey: actor(12), + }) + .await + .expect("insert banner"); + + assert_eq!( + db.ack_relay_banner_view(community, banner.public_id, &user, Uuid::new_v4()) + .await + .expect("first view"), + RelayBannerViewOutcome::Accepted { + display_count: 1, + changed: true + } + ); + assert_eq!( + db.ack_relay_banner_view(community, banner.public_id, &user, Uuid::new_v4()) + .await + .expect("second view"), + RelayBannerViewOutcome::Accepted { + display_count: 2, + changed: true + } + ); + assert_eq!( + db.ack_relay_banner_view(community, banner.public_id, &user, Uuid::new_v4()) + .await + .expect("third view"), + RelayBannerViewOutcome::Exhausted { display_count: 2 } + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn concurrent_same_view_key_consumes_once() { + let db = setup_db().await; + let community = make_community(&db.pool).await; + let user = actor(13); + let view_id = Uuid::new_v4(); + let banner = db + .admin_upsert_relay_banner(RelayBannerUpsert { + severity: RelayBannerSeverity::Urgent, + message: "concurrent".to_owned(), + max_displays: 2, + scope: RelayBannerScope::AllCommunities, + actor_pubkey: actor(14), + }) + .await + .expect("insert banner"); + + let (first, second) = tokio::join!( + db.ack_relay_banner_view(community, banner.public_id, &user, view_id), + db.ack_relay_banner_view(community, banner.public_id, &user, view_id), + ); + let changed = [first.expect("first"), second.expect("second")] + .into_iter() + .filter(|outcome| { + matches!( + outcome, + RelayBannerViewOutcome::Accepted { changed: true, .. } + ) + }) + .count(); + assert_eq!(changed, 1); + assert_eq!( + db.ack_relay_banner_view(community, banner.public_id, &user, Uuid::new_v4()) + .await + .expect("second distinct view"), + RelayBannerViewOutcome::Accepted { + display_count: 2, + changed: true + } + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn dismiss_permanently_suppresses_banner() { @@ -854,7 +1263,7 @@ mod tests { RelayBannerDismissOutcome::Dismissed { changed: false } ); assert_eq!( - db.ack_relay_banner_view(community, banner.public_id, &user) + db.ack_relay_banner_view(community, banner.public_id, &user, Uuid::new_v4()) .await .expect("view after dismiss"), RelayBannerViewOutcome::Dismissed diff --git a/crates/buzz-relay/src/api/admin/mod.rs b/crates/buzz-relay/src/api/admin/mod.rs index 25ddf26155..408b69a8df 100644 --- a/crates/buzz-relay/src/api/admin/mod.rs +++ b/crates/buzz-relay/src/api/admin/mod.rs @@ -7,6 +7,7 @@ mod auth; mod error; +use std::collections::HashSet; use std::sync::Arc; use auth::{ @@ -568,13 +569,16 @@ async fn upsert_banner( } buzz_db::RelayBannerScope::AllCommunities } else { - let ids = body.community_ids.unwrap_or_default(); + let mut ids = body.community_ids.unwrap_or_default(); + ids.sort_unstable(); + ids.dedup(); if ids.is_empty() { return Err(ApiError::bad_request( "invalid_scope", "communityIds must be non-empty when targetScope is communities", )); } + validate_active_banner_communities(&state, &ids).await?; buzz_db::RelayBannerScope::Communities( ids.into_iter() .map(buzz_core::CommunityId::from_uuid) @@ -593,6 +597,7 @@ async fn upsert_banner( }) .await .map_err(map_banner_db_error)?; + fan_out_relay_banner_update(&state, &banner, false).await; Ok(Json(AdminBannerResponse::from(banner))) } @@ -613,11 +618,112 @@ async fn disable_banner( .await?, )?; require_operator(&principal)?; - let disabled = state + let disabled_banner = state .db .admin_disable_active_relay_banner(&principal.pubkey) .await?; - Ok(Json(serde_json::json!({ "disabled": disabled }))) + if let Some(banner) = disabled_banner.as_ref() { + fan_out_relay_banner_update(&state, banner, true).await; + } + Ok(Json( + serde_json::json!({ "disabled": disabled_banner.is_some() }), + )) +} + +async fn validate_active_banner_communities( + state: &crate::state::AppState, + ids: &[Uuid], +) -> Result<(), ApiError> { + let active = state.db.admin_list_banner_communities().await?; + let active: HashSet = active + .into_iter() + .map(|community| *community.id.as_uuid()) + .collect(); + if ids.iter().all(|id| active.contains(id)) { + Ok(()) + } else { + Err(ApiError::bad_request( + "invalid_scope", + "communityIds must refer to active communities", + )) + } +} + +async fn fan_out_relay_banner_update( + state: &crate::state::AppState, + banner: &buzz_db::RelayBannerRecord, + disabled: bool, +) { + let created_at = nostr::Timestamp::from(banner.updated_at.timestamp().max(0) as u64); + let event = if disabled { + match crate::api::banners::banner_disabled_event( + &state.relay_keypair, + banner, + Some(created_at), + ) { + Ok(event) => event, + Err(error) => { + tracing::warn!("Relay banner disable event signing failed: {error}"); + return; + } + } + } else { + match crate::api::banners::banner_event(&state.relay_keypair, banner, Some(created_at)) { + Ok(event) => event, + Err(error) => { + tracing::warn!("Relay banner event signing failed: {error}"); + return; + } + } + }; + publish_relay_banner_update(state, banner, &event).await; + crate::handlers::event::fan_out_relay_banner_event(state, banner, &event, !disabled).await; +} + +async fn publish_relay_banner_update( + state: &crate::state::AppState, + banner: &buzz_db::RelayBannerRecord, + event: &nostr::Event, +) { + let communities = if banner.target_all_communities { + match state.db.admin_list_banner_communities().await { + Ok(communities) => communities + .into_iter() + .map(|community| community.id) + .collect(), + Err(error) => { + tracing::warn!("Relay banner community list failed during fan-out: {error}"); + return; + } + } + } else { + banner.community_ids.clone() + }; + for community_id in communities { + let Some(host) = state + .db + .lookup_community_host(community_id) + .await + .unwrap_or_else(|error| { + tracing::warn!(%community_id, "Relay banner community host lookup failed: {error}"); + None + }) + else { + continue; + }; + let tenant = buzz_core::TenantContext::resolved(community_id, host); + state.mark_local_event(community_id, &event.id); + if let Err(error) = state + .pubsub + .publish_event(&tenant, buzz_pubsub::EventTopic::Global, event) + .await + { + state + .local_event_ids + .invalidate(&(community_id, event.id.to_bytes())); + tracing::warn!(%community_id, "Relay banner Redis publish failed: {error}"); + } + } } fn map_banner_db_error(error: buzz_db::DbError) -> ApiError { diff --git a/crates/buzz-relay/src/api/banners.rs b/crates/buzz-relay/src/api/banners.rs index 8a68027420..56d8d2d1a7 100644 --- a/crates/buzz-relay/src/api/banners.rs +++ b/crates/buzz-relay/src/api/banners.rs @@ -11,6 +11,7 @@ use axum::{ use buzz_core::TenantContext; use serde::Serialize; use serde_json::Value; +use uuid::Uuid; use crate::api::{api_error, bridge, internal_error, relay_members}; use crate::state::AppState; @@ -19,6 +20,8 @@ use crate::state::AppState; pub(crate) const BANNER_ACTIVE_PATH: &str = "/api/banners/current"; /// Client route for acknowledging a successful banner render/view. pub(crate) const BANNER_VIEW_ROUTE: &str = "/api/banners/{banner_id}/view"; +/// Header carrying a client-stable UUID for idempotent view retries. +pub(crate) const BANNER_VIEW_ID_HEADER: &str = "x-buzz-banner-view-id"; /// Client route for dismissing a banner. pub(crate) const BANNER_DISMISS_ROUTE: &str = "/api/banners/{banner_id}/dismiss"; @@ -43,6 +46,8 @@ pub(crate) struct BannerResponse { #[serde(rename_all = "camelCase")] pub(crate) struct BannerAckResponse { status: &'static str, + #[serde(skip_serializing_if = "Option::is_none")] + changed: Option, } impl From for BannerResponse { @@ -61,6 +66,7 @@ impl From for BannerResponse { pub(crate) fn banner_event( relay_keypair: &nostr::Keys, banner: &buzz_db::RelayBannerRecord, + created_at: Option, ) -> Result { let id = banner.public_id.to_string(); let scope = if banner.target_all_communities { @@ -80,13 +86,50 @@ pub(crate) fn banner_event( nostr::Tag::parse(["d", id.as_str()]).map_err(|e| e.to_string())?, nostr::Tag::parse(["scope", scope]).map_err(|e| e.to_string())?, ]; - nostr::EventBuilder::new( + let builder = nostr::EventBuilder::new( nostr::Kind::Custom(buzz_core::kind::KIND_RELAY_BANNER as u16), content, ) - .tags(tags) - .sign_with_keys(relay_keypair) - .map_err(|e| e.to_string()) + .tags(tags); + let builder = if let Some(created_at) = created_at { + builder.custom_created_at(created_at) + } else { + builder + }; + builder + .sign_with_keys(relay_keypair) + .map_err(|e| e.to_string()) +} + +pub(crate) fn banner_disabled_event( + relay_keypair: &nostr::Keys, + banner: &buzz_db::RelayBannerRecord, + created_at: Option, +) -> Result { + let id = banner.public_id.to_string(); + let scope = if banner.target_all_communities { + "all" + } else { + "communities" + }; + let tags = vec![ + nostr::Tag::parse(["d", id.as_str()]).map_err(|e| e.to_string())?, + nostr::Tag::parse(["scope", scope]).map_err(|e| e.to_string())?, + nostr::Tag::parse(["status", "disabled"]).map_err(|e| e.to_string())?, + ]; + let builder = nostr::EventBuilder::new( + nostr::Kind::Custom(buzz_core::kind::KIND_RELAY_BANNER as u16), + "", + ) + .tags(tags); + let builder = if let Some(created_at) = created_at { + builder.custom_created_at(created_at) + } else { + builder + }; + builder + .sign_with_keys(relay_keypair) + .map_err(|e| e.to_string()) } pub(crate) async fn active_banner_event_for_user( @@ -102,7 +145,7 @@ pub(crate) async fn active_banner_event_for_user( else { return Ok(None); }; - banner_event(&state.relay_keypair, &banner).map(Some) + banner_event(&state.relay_keypair, &banner, None).map(Some) } pub(crate) async fn get_active_banner( @@ -144,25 +187,33 @@ pub(crate) async fn ack_banner_view( if !body.is_empty() { return Err(non_empty_ack_body_error()); } + let view_id = parse_view_id(&headers)?; match state .db - .ack_relay_banner_view(tenant.community(), banner_id, pubkey.as_bytes()) + .ack_relay_banner_view(tenant.community(), banner_id, pubkey.as_bytes(), view_id) .await .map_err(|e| internal_error(&format!("banner view ack: {e}")))? { - buzz_db::RelayBannerViewOutcome::Accepted { .. } => { - metrics::counter!( - "buzz_relay_banner_views_total", - "community" => tenant.host().to_owned() - ) - .increment(1); - Ok(Json(BannerAckResponse { status: "accepted" })) + buzz_db::RelayBannerViewOutcome::Accepted { changed, .. } => { + if changed { + metrics::counter!( + "buzz_relay_banner_views_total", + "community" => tenant.host().to_owned() + ) + .increment(1); + } + Ok(Json(BannerAckResponse { + status: "accepted", + changed: Some(changed), + })) } buzz_db::RelayBannerViewOutcome::Dismissed => Ok(Json(BannerAckResponse { status: "dismissed", + changed: None, })), buzz_db::RelayBannerViewOutcome::Exhausted { .. } => Ok(Json(BannerAckResponse { status: "exhausted", + changed: None, })), buzz_db::RelayBannerViewOutcome::NotEligible => Err(api_error( axum::http::StatusCode::NOT_FOUND, @@ -207,6 +258,7 @@ pub(crate) async fn ack_banner_dismiss( } Ok(Json(BannerAckResponse { status: "dismissed", + changed: Some(changed), })) } buzz_db::RelayBannerDismissOutcome::NotEligible => Err(api_error( @@ -257,6 +309,29 @@ async fn authenticate_client_request( )) } +fn parse_view_id(headers: &HeaderMap) -> Result)> { + let Some(raw) = headers + .get(BANNER_VIEW_ID_HEADER) + .and_then(|value| value.to_str().ok()) + else { + return Err(invalid_view_id_error()); + }; + let Ok(view_id) = Uuid::parse_str(raw) else { + return Err(invalid_view_id_error()); + }; + if raw != view_id.hyphenated().to_string() || view_id.is_nil() { + return Err(invalid_view_id_error()); + } + Ok(view_id) +} + +fn invalid_view_id_error() -> (axum::http::StatusCode, Json) { + api_error( + axum::http::StatusCode::BAD_REQUEST, + "missing or invalid banner view id", + ) +} + fn non_empty_ack_body_error() -> (axum::http::StatusCode, Json) { api_error( axum::http::StatusCode::BAD_REQUEST, @@ -288,7 +363,7 @@ mod tests { fn banner_event_uses_published_contract() { let keys = nostr::Keys::generate(); let banner = banner_record(false); - let event = banner_event(&keys, &banner).expect("banner event"); + let event = banner_event(&keys, &banner, None).expect("banner event"); assert_eq!( event.kind.as_u16() as u32, @@ -322,11 +397,44 @@ mod tests { assert!(content.get("message").is_none()); } + #[test] + fn disabled_banner_event_uses_clear_contract() { + let keys = nostr::Keys::generate(); + let banner = banner_record(false); + let event = banner_disabled_event(&keys, &banner, None).expect("disabled event"); + + assert_eq!( + event.kind.as_u16() as u32, + buzz_core::kind::KIND_RELAY_BANNER + ); + assert!(event.content.is_empty()); + assert_eq!( + event + .tags + .iter() + .find(|tag| tag.kind().to_string() == "scope") + .and_then(|tag| tag.content()), + Some("communities") + ); + assert_eq!( + event + .tags + .iter() + .find(|tag| tag.kind().to_string() == "status") + .and_then(|tag| tag.content()), + Some("disabled") + ); + assert!(event + .tags + .iter() + .all(|tag| tag.kind().to_string() != "text")); + } + #[test] fn all_community_banner_event_has_all_scope() { let keys = nostr::Keys::generate(); let banner = banner_record(true); - let event = banner_event(&keys, &banner).expect("banner event"); + let event = banner_event(&keys, &banner, None).expect("banner event"); assert_eq!( event @@ -344,6 +452,40 @@ mod tests { assert_eq!(BANNER_DISMISS_ROUTE, "/api/banners/{banner_id}/dismiss"); } + #[test] + fn parse_view_id_requires_canonical_non_nil_uuid() { + let mut headers = HeaderMap::new(); + assert!(parse_view_id(&headers).is_err()); + + headers.insert(BANNER_VIEW_ID_HEADER, "not-a-uuid".parse().expect("header")); + assert!(parse_view_id(&headers).is_err()); + + headers.insert( + BANNER_VIEW_ID_HEADER, + "00000000-0000-0000-0000-000000000000" + .parse() + .expect("header"), + ); + assert!(parse_view_id(&headers).is_err()); + + headers.insert( + BANNER_VIEW_ID_HEADER, + "12345678123456781234567812345678".parse().expect("header"), + ); + assert!(parse_view_id(&headers).is_err()); + + headers.insert( + BANNER_VIEW_ID_HEADER, + "12345678-1234-5678-1234-567812345678" + .parse() + .expect("header"), + ); + assert_eq!( + parse_view_id(&headers).expect("valid canonical uuid"), + Uuid::parse_str("12345678-1234-5678-1234-567812345678").expect("uuid") + ); + } + #[test] fn non_empty_ack_body_is_rejected() { let (status, body) = non_empty_ack_body_error(); diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 3f767c1874..01d6451fe9 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -36,7 +36,7 @@ pub(crate) fn bounded_kind_label(kind: u32) -> String { match kind { 0..=9 | 1059 | 1063 => kind.to_string(), 8000..=8003 | 9000..=9022 | 9030..=9036 => kind.to_string(), - 13534..=13535 => kind.to_string(), + 13534..=13536 => kind.to_string(), 20000..=29999 => kind.to_string(), 30023 | 30315 | 39000..=39003 => kind.to_string(), 40002..=40100 => kind.to_string(), @@ -221,6 +221,88 @@ pub async fn filter_fanout_by_access( allowed } +/// Fan out a relay banner control event to live banner subscribers whose authenticated +/// user is eligible for that banner in their bound community. +pub(crate) async fn fan_out_relay_banner_event( + state: &AppState, + banner: &buzz_db::RelayBannerRecord, + event: &Event, + require_active: bool, +) { + let communities: Vec = if banner.target_all_communities { + state + .conn_manager + .per_community_ws_connections() + .keys() + .copied() + .collect() + } else { + banner.community_ids.clone() + }; + fan_out_relay_banner_event_to_communities(state, banner, event, require_active, communities) + .await; +} + +async fn fan_out_relay_banner_event_to_communities( + state: &AppState, + banner: &buzz_db::RelayBannerRecord, + event: &Event, + require_active: bool, + communities: Vec, +) { + let stored = StoredEvent::with_received_at(event.clone(), chrono::Utc::now(), None, true); + let mut matches = Vec::new(); + let mut seen = std::collections::HashSet::new(); + for community_id in communities { + for (conn_id, sub_id) in state.sub_registry.fan_out_scoped(community_id, &stored) { + if !seen.insert((conn_id, sub_id.clone())) { + continue; + } + let Some(pubkey) = state.conn_manager.pubkey_for_conn(conn_id) else { + continue; + }; + match state + .db + .relay_banner_user_eligible(banner.id, community_id, &pubkey, require_active) + .await + { + Ok(true) => matches.push((conn_id, sub_id)), + Ok(false) => {} + Err(error) => { + tracing::warn!(%community_id, "Relay banner eligibility lookup failed: {error}"); + } + } + } + } + if matches.is_empty() { + return; + } + let event_json = match serde_json::to_string(event) { + Ok(json) => json, + Err(error) => { + error!("Failed to serialize relay banner event for fan-out: {error}"); + return; + } + }; + let frames = fanout_frame_cache( + matches.iter().map(|(_, sub_id)| sub_id.as_str()), + &event_json, + ); + let drop_count = send_fanout_frames( + state, + matches + .iter() + .map(|(conn_id, sub_id)| (*conn_id, sub_id.as_str())), + &frames, + ); + if drop_count > 0 { + tracing::warn!( + drop_count, + "relay banner fan-out: {drop_count} connection(s) cancelled due to full/closed buffers" + ); + } +} + /// Deliver one event to this relay's local subscribers through the access gate. /// /// This is the single guarded send path for relay-local EVENT delivery. It runs @@ -303,6 +385,11 @@ pub async fn fan_out_pubsub_event(state: &Arc, channel_event: buzz_pub return; } + if event_kind_u32(&stored.event) == buzz_core::kind::KIND_RELAY_BANNER { + fan_out_pubsub_relay_banner_event(state, community_id, &stored.event).await; + return; + } + let matches = state.sub_registry.fan_out_scoped(community_id, &stored); let matches = filter_fanout_by_access(state, community_id, &stored, matches, None).await; metrics::counter!("buzz_multinode_fanout_total").increment(1); @@ -337,6 +424,38 @@ pub async fn fan_out_pubsub_event(state: &Arc, channel_event: buzz_pub } } +async fn fan_out_pubsub_relay_banner_event( + state: &AppState, + community_id: CommunityId, + event: &Event, +) { + let disabled = event + .tags + .iter() + .any(|tag| tag.kind().to_string() == "status" && tag.content() == Some("disabled")); + let banner_id = event + .tags + .iter() + .find(|tag| tag.kind().to_string() == "d") + .and_then(|tag| tag.content()) + .and_then(|value| uuid::Uuid::parse_str(value).ok()); + let Some(banner_id) = banner_id else { + tracing::warn!("Relay banner pubsub event missing valid d tag"); + return; + }; + let Some(banner) = (match state.db.relay_banner_by_public_id(banner_id).await { + Ok(banner) => banner, + Err(error) => { + tracing::warn!(%banner_id, "Relay banner pubsub lookup failed: {error}"); + return; + } + }) else { + return; + }; + fan_out_relay_banner_event_to_communities(state, &banner, event, !disabled, vec![community_id]) + .await; +} + /// Schedule post-commit delivery/side effects for a stored event. /// /// This intentionally returns after only the bounded audit enqueue has completed: @@ -2859,3 +2978,161 @@ mod tests { } } } + +#[cfg(test)] +mod relay_banner_fanout_tests { + use super::*; + use axum::extract::ws::Message as WsMessage; + use std::collections::HashMap; + use tokio::sync::{mpsc, Mutex}; + use tokio_util::sync::CancellationToken; + use uuid::Uuid; + + async fn setup_state() -> (Arc, sqlx::PgPool) { + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect test DB"); + let state = crate::state::tests::test_state_with_database_pool(pool.clone()).await; + (state, pool) + } + + async fn insert_test_community(pool: &sqlx::PgPool, host_prefix: &str) -> CommunityId { + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(format!("{host_prefix}-{}.example", id.simple())) + .execute(pool) + .await + .expect("insert community"); + CommunityId::from_uuid(id) + } + + fn register_banner_sub( + state: &AppState, + community: CommunityId, + pubkey: Vec, + ) -> (Uuid, mpsc::Receiver) { + let conn_id = Uuid::new_v4(); + let (tx, rx) = mpsc::channel(4); + let (ctrl_tx, _ctrl_rx) = mpsc::channel(4); + state.conn_manager.register( + conn_id, + tx, + ctrl_tx, + None, + CancellationToken::new(), + community, + Arc::new(std::sync::atomic::AtomicU8::new(0)), + Arc::new(Mutex::new(HashMap::new())), + 3, + ); + state.conn_manager.set_authenticated_pubkey(conn_id, pubkey); + state.sub_registry.register_scoped( + community, + conn_id, + "banner".to_owned(), + vec![nostr::Filter::new().kind(nostr::Kind::Custom( + buzz_core::kind::KIND_RELAY_BANNER as u16, + ))], + None, + ); + (conn_id, rx) + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn relay_banner_upsert_live_routes_only_to_eligible_users() { + let (state, pool) = setup_state().await; + let included = insert_test_community(&pool, "banner-fanout-included").await; + let excluded = insert_test_community(&pool, "banner-fanout-excluded").await; + let (_eligible_conn, mut eligible_rx) = register_banner_sub(&state, included, vec![7; 32]); + let (_excluded_conn, mut excluded_rx) = register_banner_sub(&state, excluded, vec![8; 32]); + let banner = state + .db + .admin_upsert_relay_banner(buzz_db::RelayBannerUpsert { + severity: buzz_db::RelayBannerSeverity::Info, + message: "live".to_owned(), + max_displays: 1, + scope: buzz_db::RelayBannerScope::Communities(vec![included]), + actor_pubkey: vec![1; 32], + }) + .await + .expect("upsert banner"); + let event = crate::api::banners::banner_event(&state.relay_keypair, &banner, None) + .expect("banner event"); + + fan_out_relay_banner_event(&state, &banner, &event, true).await; + + let frame = eligible_rx.recv().await.expect("live banner frame"); + let WsMessage::Text(frame) = frame else { + panic!("expected text frame"); + }; + assert!(frame.contains("\"live\"")); + assert!(excluded_rx.try_recv().is_err()); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn relay_banner_disable_live_routes_using_pre_disable_eligibility() { + let (state, pool) = setup_state().await; + let eligible_community = insert_test_community(&pool, "banner-disable-eligible").await; + let excluded_community = insert_test_community(&pool, "banner-disable-excluded").await; + let exhausted_user = vec![8; 32]; + let eligible_user = vec![9; 32]; + let excluded_user = vec![10; 32]; + let (_exhausted_conn, mut exhausted_rx) = + register_banner_sub(&state, eligible_community, exhausted_user.clone()); + let (_eligible_conn, mut eligible_rx) = + register_banner_sub(&state, eligible_community, eligible_user); + let (_excluded_conn, mut excluded_rx) = + register_banner_sub(&state, excluded_community, excluded_user); + let banner = state + .db + .admin_upsert_relay_banner(buzz_db::RelayBannerUpsert { + severity: buzz_db::RelayBannerSeverity::Warning, + message: "clear".to_owned(), + max_displays: 1, + scope: buzz_db::RelayBannerScope::Communities(vec![eligible_community]), + actor_pubkey: vec![1; 32], + }) + .await + .expect("upsert banner"); + assert_eq!( + state + .db + .ack_relay_banner_view( + eligible_community, + banner.public_id, + &exhausted_user, + Uuid::new_v4(), + ) + .await + .expect("exhaust user"), + buzz_db::RelayBannerViewOutcome::Accepted { + display_count: 1, + changed: true, + } + ); + let disabled = state + .db + .admin_disable_active_relay_banner(&[1; 32]) + .await + .expect("disable banner") + .expect("disabled banner"); + assert_eq!(disabled.id, banner.id); + let event = + crate::api::banners::banner_disabled_event(&state.relay_keypair, &disabled, None) + .expect("disabled event"); + + fan_out_relay_banner_event(&state, &disabled, &event, false).await; + + let frame = eligible_rx.recv().await.expect("live disable frame"); + let WsMessage::Text(frame) = frame else { + panic!("expected text frame"); + }; + assert!(frame.contains("\"status\",\"disabled\"")); + assert!(frame.contains("\"scope\",\"communities\"")); + assert!(exhausted_rx.try_recv().is_err()); + assert!(excluded_rx.try_recv().is_err()); + } +} diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index ded8348a58..ea3958b434 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -222,6 +222,15 @@ pub async fn handle_req( } if filters_are_relay_banner_only(&filters) { + register_subscription( + &state, + &conn, + conn_id, + &sub_id, + &filters, + authorized_requested_channels.as_ref(), + ) + .await; match crate::api::banners::active_banner_event_for_user(&state, &conn.tenant, &pubkey_bytes) .await { @@ -306,46 +315,15 @@ pub async fn handle_req( return; } - { - let mut subs = conn.subscriptions.lock().await; - subs.insert(sub_id.clone(), filters.clone()); - } - - let replaced = if let Some(channel_ids) = authorized_requested_channels.as_ref() { - state.sub_registry.register_channels_scoped( - conn.tenant.community(), - conn_id, - sub_id.clone(), - filters.clone(), - channel_ids.clone(), - ) - } else { - state.sub_registry.register_scoped( - conn.tenant.community(), - conn_id, - sub_id.clone(), - filters.clone(), - None, - ) - }; - if let Some(replaced) = replaced { - release_subscription_topics(&state, &conn.tenant, &replaced.scope).await; - } - if let Some(channel_ids) = authorized_requested_channels.as_ref() { - for &channel_id in channel_ids { - state - .pubsub - .retain_topic(&conn.tenant, EventTopic::Channel(channel_id)) - .await; - } - } else { - state - .pubsub - .retain_topic(&conn.tenant, EventTopic::Global) - .await; - } - - debug!(conn_id = %conn_id, sub_id = %sub_id, "Subscription registered"); + register_subscription( + &state, + &conn, + conn_id, + &sub_id, + &filters, + authorized_requested_channels.as_ref(), + ) + .await; // NIP-01 OR semantics: execute one DB query per filter and deduplicate results // by event ID. Collapsing all filters into a single query would merge their @@ -1171,6 +1149,56 @@ pub(crate) fn extract_channel_ids_from_filters(filters: &[Filter]) -> Option>, +) { + { + let mut subs = conn.subscriptions.lock().await; + subs.insert(sub_id.to_owned(), filters.to_vec()); + } + + let replaced = if let Some(channel_ids) = authorized_requested_channels { + state.sub_registry.register_channels_scoped( + conn.tenant.community(), + conn_id, + sub_id.to_owned(), + filters.to_vec(), + channel_ids.clone(), + ) + } else { + state.sub_registry.register_scoped( + conn.tenant.community(), + conn_id, + sub_id.to_owned(), + filters.to_vec(), + None, + ) + }; + if let Some(replaced) = replaced { + release_subscription_topics(state, &conn.tenant, &replaced.scope).await; + } + if let Some(channel_ids) = authorized_requested_channels { + for &channel_id in channel_ids { + state + .pubsub + .retain_topic(&conn.tenant, EventTopic::Channel(channel_id)) + .await; + } + } else { + state + .pubsub + .retain_topic(&conn.tenant, EventTopic::Global) + .await; + } + + debug!(conn_id = %conn_id, sub_id = %sub_id, "Subscription registered"); +} + fn filters_are_huddle_liveness_only(filters: &[Filter]) -> bool { !filters.is_empty() && filters.iter().all(|filter| { diff --git a/migrations/0046_relay_banners.sql b/migrations/0046_relay_banners.sql index 2be963e9ab..d93f4f6ca0 100644 --- a/migrations/0046_relay_banners.sql +++ b/migrations/0046_relay_banners.sql @@ -53,10 +53,24 @@ CREATE TABLE relay_banner_user_state ( CREATE INDEX idx_relay_banner_user_state_pubkey ON relay_banner_user_state (community_id, pubkey, banner_id); +CREATE TABLE relay_banner_view_acks ( + banner_id BIGINT NOT NULL REFERENCES relay_banners(id) ON DELETE CASCADE, + community_id UUID NOT NULL REFERENCES communities(id) ON DELETE CASCADE, + pubkey BYTEA NOT NULL CHECK (length(pubkey) = 32), + view_id UUID NOT NULL, + viewed_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (banner_id, community_id, pubkey, view_id) +); + +CREATE INDEX idx_relay_banner_view_acks_pubkey + ON relay_banner_view_acks (community_id, pubkey, banner_id); + INSERT INTO _operator_global_tables (table_name, reason) VALUES ('relay_banners', 'deployment-global operator-configured banner; no community_id intentionally'), ('relay_banner_communities', 'deployment-global banner targeting allowlist; community_id is target provenance only'), - ('relay_banner_user_state', 'deployment-global per-user banner display state; community_id scopes the targeted impression only'); + ('relay_banner_user_state', 'deployment-global per-user banner display state; community_id scopes the targeted impression only'), + ('relay_banner_view_acks', 'deployment-global per-render banner idempotency keys; community_id scopes the targeted impression only'); SELECT attach_community_write_fence('relay_banner_communities'); SELECT attach_community_write_fence('relay_banner_user_state'); +SELECT attach_community_write_fence('relay_banner_view_acks'); diff --git a/schema/schema.sql b/schema/schema.sql index 6e4e375088..8aa3775b3b 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -1757,10 +1757,23 @@ CREATE TABLE relay_banner_user_state ( CREATE INDEX idx_relay_banner_user_state_pubkey ON relay_banner_user_state (community_id, pubkey, banner_id); +CREATE TABLE relay_banner_view_acks ( + banner_id BIGINT NOT NULL REFERENCES relay_banners(id) ON DELETE CASCADE, + community_id UUID NOT NULL REFERENCES communities(id) ON DELETE CASCADE, + pubkey BYTEA NOT NULL CHECK (length(pubkey) = 32), + view_id UUID NOT NULL, + viewed_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (banner_id, community_id, pubkey, view_id) +); + +CREATE INDEX idx_relay_banner_view_acks_pubkey + ON relay_banner_view_acks (community_id, pubkey, banner_id); + INSERT INTO _operator_global_tables (table_name, reason) VALUES ('relay_banners', 'deployment-global operator-configured banner; no community_id intentionally'), ('relay_banner_communities', 'deployment-global banner targeting allowlist; community_id is target provenance only'), - ('relay_banner_user_state', 'deployment-global per-user banner display state; community_id scopes the targeted impression only'); + ('relay_banner_user_state', 'deployment-global per-user banner display state; community_id scopes the targeted impression only'), + ('relay_banner_view_acks', 'deployment-global per-render banner idempotency keys; community_id scopes the targeted impression only'); -- Attach the universal fence to every existing table carrying community_id, @@ -1814,6 +1827,7 @@ SELECT attach_community_write_fence('relay_invites'); SELECT attach_community_write_fence('relay_members'); SELECT attach_community_write_fence('relay_banner_communities'); SELECT attach_community_write_fence('relay_banner_user_state'); +SELECT attach_community_write_fence('relay_banner_view_acks'); SELECT attach_community_write_fence('scheduled_workflow_fires'); SELECT attach_community_write_fence('subscriptions'); SELECT attach_community_write_fence('thread_metadata');