diff --git a/crates/buzz-db/src/dm.rs b/crates/buzz-db/src/dm.rs index 3df7e87db5..2dadb911c3 100644 --- a/crates/buzz-db/src/dm.rs +++ b/crates/buzz-db/src/dm.rs @@ -5,7 +5,7 @@ use chrono::{DateTime, Utc}; use sha2::{Digest, Sha256}; -use sqlx::{PgPool, Row}; +use sqlx::{PgPool, Postgres, Row, Transaction}; use uuid::Uuid; use crate::channel::ChannelRecord; @@ -505,6 +505,20 @@ pub async fn unhide_dm_recipients( community_id: CommunityId, channel_id: Uuid, sender_pubkey: &[u8], +) -> Result>> { + let mut tx = pool.begin().await?; + let viewers = unhide_dm_recipients_tx(&mut tx, community_id, channel_id, sender_pubkey).await?; + tx.commit().await?; + Ok(viewers) +} + +/// Transactional form of [`unhide_dm_recipients`] used when accepted-message +/// persistence and the canonical visibility mutation must commit together. +pub(crate) async fn unhide_dm_recipients_tx( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + channel_id: Uuid, + sender_pubkey: &[u8], ) -> Result>> { let rows = sqlx::query( r#" @@ -539,7 +553,7 @@ pub async fn unhide_dm_recipients( .bind(community_id.as_uuid()) .bind(channel_id) .bind(sender_pubkey) - .fetch_all(pool) + .fetch_all(&mut **tx) .await?; rows.into_iter() @@ -602,17 +616,97 @@ pub async fn claim_dm_visibility_dirty_viewers( ORDER BY q.lease_until, q.community_id, q.viewer FOR UPDATE OF q SKIP LOCKED LIMIT $3 + ), overflow_fresh AS ( + SELECT q.community_id, q.viewer + FROM dm_visibility_dirty_viewers q + JOIN communities c ON c.id = q.community_id + WHERE c.archived_at IS NULL + AND c.deleted_at IS NULL + AND c.deletion_state = 'active' + AND community_write_allowed(q.community_id) + AND q.state = 'pending' + AND q.attempts = 0 + AND q.next_attempt_at <= NOW() + AND NOT EXISTS ( + SELECT 1 FROM fresh_candidates reserved + WHERE reserved.community_id = q.community_id + AND reserved.viewer = q.viewer + ) + ORDER BY q.dirty_at, q.community_id, q.viewer + FOR UPDATE OF q SKIP LOCKED + LIMIT GREATEST( + $4 - (SELECT COUNT(*) FROM fresh_candidates) + - (SELECT COUNT(*) FROM retry_candidates) + - (SELECT COUNT(*) FROM recovery_candidates), + 0 + ) + ), overflow_retry AS ( + SELECT q.community_id, q.viewer + FROM dm_visibility_dirty_viewers q + JOIN communities c ON c.id = q.community_id + WHERE c.archived_at IS NULL + AND c.deleted_at IS NULL + AND c.deletion_state = 'active' + AND community_write_allowed(q.community_id) + AND q.state = 'pending' + AND q.attempts > 0 + AND q.next_attempt_at <= NOW() + AND NOT EXISTS ( + SELECT 1 FROM retry_candidates reserved + WHERE reserved.community_id = q.community_id + AND reserved.viewer = q.viewer + ) + ORDER BY q.next_attempt_at, q.dirty_at, q.community_id, q.viewer + FOR UPDATE OF q SKIP LOCKED + LIMIT GREATEST( + $4 - (SELECT COUNT(*) FROM fresh_candidates) + - (SELECT COUNT(*) FROM retry_candidates) + - (SELECT COUNT(*) FROM recovery_candidates) + - (SELECT COUNT(*) FROM overflow_fresh), + 0 + ) + ), overflow_recovery AS ( + SELECT q.community_id, q.viewer + FROM dm_visibility_dirty_viewers q + JOIN communities c ON c.id = q.community_id + WHERE c.archived_at IS NULL + AND c.deleted_at IS NULL + AND c.deletion_state = 'active' + AND community_write_allowed(q.community_id) + AND q.state = 'publishing' + AND q.lease_until <= NOW() + AND NOT EXISTS ( + SELECT 1 FROM recovery_candidates reserved + WHERE reserved.community_id = q.community_id + AND reserved.viewer = q.viewer + ) + ORDER BY q.lease_until, q.community_id, q.viewer + FOR UPDATE OF q SKIP LOCKED + LIMIT GREATEST( + $4 - (SELECT COUNT(*) FROM fresh_candidates) + - (SELECT COUNT(*) FROM retry_candidates) + - (SELECT COUNT(*) FROM recovery_candidates) + - (SELECT COUNT(*) FROM overflow_fresh) + - (SELECT COUNT(*) FROM overflow_retry), + 0 + ) ), candidates AS ( SELECT community_id, viewer FROM fresh_candidates UNION ALL SELECT community_id, viewer FROM retry_candidates UNION ALL SELECT community_id, viewer FROM recovery_candidates + UNION ALL + SELECT community_id, viewer FROM overflow_fresh + UNION ALL + SELECT community_id, viewer FROM overflow_retry + UNION ALL + SELECT community_id, viewer FROM overflow_recovery ) UPDATE dm_visibility_dirty_viewers q SET state = 'publishing', - claim_id = $4, - lease_until = NOW() + make_interval(secs => $5::int), + claim_id = $5, + lease_until = NOW() + make_interval(secs => $6::int), attempts = q.attempts + 1 FROM candidates candidate, communities c WHERE q.community_id = candidate.community_id @@ -624,6 +718,7 @@ pub async fn claim_dm_visibility_dirty_viewers( .bind(fresh_limit) .bind(retry_limit) .bind(recovery_limit) + .bind(limit) .bind(claim_id) .bind(lease_seconds.clamp(1, 3_600) as i32) .fetch_all(pool) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 8a620089af..a52acd319d 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -2265,6 +2265,48 @@ impl Db { Ok(result) } + /// Atomically inserts a DM message, its optional thread metadata, and the + /// recipient visibility mutations that make the accepted message visible. + /// + /// Duplicate events do not repeat the visibility mutation. The returned + /// viewers are the recipients whose canonical hidden state changed and + /// whose relay-authored visibility snapshots should be published after + /// this transaction commits. + #[datastore_span(name = "insert_dm_event_with_thread_metadata", system = "postgresql")] + pub async fn insert_dm_event_with_thread_metadata( + &self, + community_id: CommunityId, + event: &nostr::Event, + channel_id: Uuid, + thread_meta: Option>, + sender_pubkey: &[u8], + ) -> Result<(StoredEvent, bool, Vec>)> { + let mut tx = self.pool.begin().await?; + let (stored_event, was_inserted) = event::insert_event_with_thread_metadata_tx( + &mut tx, + community_id, + event, + Some(channel_id), + thread_meta, + ) + .await?; + let resurfaced_viewers = if was_inserted { + dm::unhide_dm_recipients_tx(&mut tx, community_id, channel_id, sender_pubkey).await? + } else { + Vec::new() + }; + tx.commit().await?; + + if was_inserted { + if let Err(e) = insert_mentions(&self.pool, community_id, event, Some(channel_id)).await + { + tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}"); + } + } + + Ok((stored_event, was_inserted, resurfaced_viewers)) + } + /// Atomically insert a kind:7 reaction event and its reaction row. #[allow(clippy::too_many_arguments)] #[datastore_span( @@ -7500,6 +7542,156 @@ mod tests { ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn dm_event_insert_and_recipient_resurface_roll_back_together() { + use nostr::{EventBuilder, Keys, Kind, Tag}; + + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (pool, scratch_name) = create_scratch_db(&admin, "dm_insert_resurface").await; + let db = Db::from_pool(pool.clone()); + let community_uuid = Uuid::new_v4(); + let community = CommunityId::from_uuid(community_uuid); + let dm_channel = Uuid::new_v4(); + let unrelated_dm = Uuid::new_v4(); + let sender = Keys::generate(); + let recipient = Keys::generate(); + let sender_bytes = sender.public_key().to_bytes(); + let recipient_bytes = recipient.public_key().to_bytes(); + + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_uuid) + .bind(format!("dm-insert-{}.example", community_uuid.simple())) + .execute(&pool) + .await + .expect("insert community"); + for channel_id in [dm_channel, unrelated_dm] { + sqlx::query( + "INSERT INTO channels \ + (id, community_id, name, channel_type, visibility, created_by) \ + VALUES ($1, $2, 'atomic DM', 'dm', 'private', $3)", + ) + .bind(channel_id) + .bind(community_uuid) + .bind(sender_bytes.as_slice()) + .execute(&pool) + .await + .expect("insert DM"); + } + for (channel_id, pubkey) in [ + (dm_channel, sender_bytes.as_slice()), + (dm_channel, recipient_bytes.as_slice()), + (unrelated_dm, recipient_bytes.as_slice()), + ] { + sqlx::query( + "INSERT INTO channel_members \ + (community_id, channel_id, pubkey, hidden_at) \ + VALUES ($1, $2, $3, NOW())", + ) + .bind(community_uuid) + .bind(channel_id) + .bind(pubkey) + .execute(&pool) + .await + .expect("insert hidden DM member"); + } + + sqlx::raw_sql( + "CREATE FUNCTION fail_dm_visibility_enqueue() RETURNS trigger \ + LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'injected enqueue failure'; END $$; \ + CREATE TRIGGER fail_dm_visibility_enqueue \ + BEFORE INSERT OR UPDATE ON dm_visibility_dirty_viewers \ + FOR EACH ROW EXECUTE FUNCTION fail_dm_visibility_enqueue();", + ) + .execute(&pool) + .await + .expect("install visibility failure injection"); + + let event = EventBuilder::new(Kind::Custom(9), "accepted DM") + .tags(vec![ + Tag::parse(["h", &dm_channel.to_string()]).expect("h tag") + ]) + .sign_with_keys(&sender) + .expect("sign DM event"); + db.insert_dm_event_with_thread_metadata( + community, + &event, + dm_channel, + None, + sender_bytes.as_slice(), + ) + .await + .expect_err("injected enqueue failure must abort accepted-message transaction"); + + let persisted_after_failure: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE community_id = $1 AND id = $2") + .bind(community_uuid) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("count rolled-back event"); + assert_eq!(persisted_after_failure, 0); + let hidden_after_failure: bool = sqlx::query_scalar( + "SELECT hidden_at IS NOT NULL FROM channel_members \ + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", + ) + .bind(community_uuid) + .bind(dm_channel) + .bind(recipient_bytes.as_slice()) + .fetch_one(&pool) + .await + .expect("read recipient after rollback"); + assert!(hidden_after_failure); + + sqlx::raw_sql( + "DROP TRIGGER fail_dm_visibility_enqueue ON dm_visibility_dirty_viewers; \ + DROP FUNCTION fail_dm_visibility_enqueue();", + ) + .execute(&pool) + .await + .expect("remove visibility failure injection"); + let (_stored, inserted, resurfaced) = db + .insert_dm_event_with_thread_metadata( + community, + &event, + dm_channel, + None, + sender_bytes.as_slice(), + ) + .await + .expect("retry accepted DM transaction"); + assert!(inserted); + assert_eq!(resurfaced, vec![recipient_bytes.to_vec()]); + + let hidden_states: Vec<(Uuid, Vec, bool)> = sqlx::query_as( + "SELECT channel_id, pubkey, hidden_at IS NOT NULL \ + FROM channel_members WHERE community_id = $1 ORDER BY channel_id, pubkey", + ) + .bind(community_uuid) + .fetch_all(&pool) + .await + .expect("read final hidden states"); + assert!(hidden_states + .iter() + .any(|(channel, pubkey, hidden)| *channel == dm_channel + && pubkey == sender_bytes.as_slice() + && *hidden)); + assert!(hidden_states + .iter() + .any(|(channel, pubkey, hidden)| *channel == dm_channel + && pubkey == recipient_bytes.as_slice() + && !*hidden)); + assert!(hidden_states + .iter() + .any(|(channel, pubkey, hidden)| *channel == unrelated_dm + && pubkey == recipient_bytes.as_slice() + && *hidden)); + + drop_scratch_db(&admin, pool, &scratch_name).await; + } + #[tokio::test] #[ignore = "requires Postgres"] async fn dm_visibility_fresh_claims_are_not_starved_by_poison_retries() { @@ -7547,8 +7739,8 @@ mod tests { ); assert_eq!( claims.len(), - 26, - "the batch reserves half for fresh work and a quarter for retries" + 100, + "reserved fairness capacity must not leave the rest of the batch idle" ); let retry_claim = claims @@ -7563,7 +7755,7 @@ mod tests { .await .expect("release retry claim"); let retry_delay_seconds: f64 = sqlx::query_scalar( - "SELECT EXTRACT(EPOCH FROM (next_attempt_at - NOW())) \ + "SELECT EXTRACT(EPOCH FROM (next_attempt_at - NOW()))::DOUBLE PRECISION \ FROM dm_visibility_dirty_viewers \ WHERE community_id = $1 AND viewer = $2", ) diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index ec83757b09..23f4ee72f3 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -2038,11 +2038,36 @@ mod tests { .execute(&pool) .await .expect("insert hidden legacy viewer"); + sqlx::query("UPDATE communities SET archived_at = NOW() WHERE id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("archive community before queue migration"); run_migrations(&pool) .await .expect("apply DM visibility queue migration"); let db = Db::from_pool(pool.clone()); + let queued_while_archived: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM dm_visibility_dirty_viewers WHERE community_id = $1", + ) + .bind(community_id) + .fetch_one(&pool) + .await + .expect("count archived brownfield viewer"); + assert_eq!(queued_while_archived, 1); + assert!( + db.claim_dm_visibility_dirty_viewers(100, 60) + .await + .expect("defer archived viewer") + .is_empty(), + "archived communities must retain repair work without publishing" + ); + sqlx::query("UPDATE communities SET archived_at = NULL WHERE id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("unarchive community after migration"); let claims = db .claim_dm_visibility_dirty_viewers(100, 60) .await diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 556153949f..352161286f 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -3036,65 +3036,96 @@ async fn ingest_event_inner( }); } - let (stored_event, was_inserted) = if buzz_core::kind::is_replaceable(kind_u32) { - // NIP-16 replaceable event — atomic replace with stale-write protection. - // channel_id is None for global kinds (0, 1, 3) due to step 5b above. - state - .db - .replace_addressable_event(tenant.community(), &event, channel_id) - .await - .map_err(|e| IngestError::Internal(format!("error: {e}")))? - } else if is_parameterized_replaceable(kind_u32) { - // NIP-33 parameterized replaceable — keyed by (kind, pubkey, d_tag). - let d_tag = buzz_db::event::extract_d_tag(&event).unwrap_or_default(); - if d_tag.len() > buzz_db::event::D_TAG_MAX_LEN { - return Err(IngestError::Rejected(format!( - "invalid: d tag too long ({} bytes, max {})", - d_tag.len(), - buzz_db::event::D_TAG_MAX_LEN, - ))); - } - state - .db - .replace_parameterized_event(tenant.community(), &event, &d_tag, channel_id) - .await - .map_err(|e| IngestError::Internal(format!("error: {e}")))? + let dm_message_sender = if matches!(kind_u32, KIND_STREAM_MESSAGE | KIND_STREAM_MESSAGE_V2) + && channel_row + .as_ref() + .is_some_and(|channel| channel.channel_type == "dm") + { + Some(effective_message_author( + &event, + &state.relay_keypair.public_key(), + )) } else { - let thread_params = thread_meta.as_ref().map(|m| m.as_params()); - match state - .db - .insert_event_with_thread_metadata( - tenant.community(), - &event, - channel_id, - thread_params, - ) - .await - { - Ok(result) => result, - Err(e) => { - // Compensate: if we pre-created a channel for kind:9007, - // soft-delete it so no orphaned channel row remains. - if let Some(ch_id) = pre_created_channel { - if let Err(re) = state - .db - .soft_delete_channel(tenant.community(), ch_id) - .await - { - warn!(event_id = %event_id_hex, "channel compensation failed: {re}"); - } - state.invalidate_channel_deleted(tenant); - } - return Err(match e { - buzz_db::DbError::AuthEventRejected => { - IngestError::Rejected("invalid: AUTH events cannot be stored".into()) - } - other => IngestError::Internal(format!("error: database error: {other}")), - }); - } - } + None }; + let (stored_event, was_inserted, resurfaced_viewers) = + if buzz_core::kind::is_replaceable(kind_u32) { + // NIP-16 replaceable event — atomic replace with stale-write protection. + // channel_id is None for global kinds (0, 1, 3) due to step 5b above. + let (stored, inserted) = state + .db + .replace_addressable_event(tenant.community(), &event, channel_id) + .await + .map_err(|e| IngestError::Internal(format!("error: {e}")))?; + (stored, inserted, Vec::new()) + } else if is_parameterized_replaceable(kind_u32) { + // NIP-33 parameterized replaceable — keyed by (kind, pubkey, d_tag). + let d_tag = buzz_db::event::extract_d_tag(&event).unwrap_or_default(); + if d_tag.len() > buzz_db::event::D_TAG_MAX_LEN { + return Err(IngestError::Rejected(format!( + "invalid: d tag too long ({} bytes, max {})", + d_tag.len(), + buzz_db::event::D_TAG_MAX_LEN, + ))); + } + let (stored, inserted) = state + .db + .replace_parameterized_event(tenant.community(), &event, &d_tag, channel_id) + .await + .map_err(|e| IngestError::Internal(format!("error: {e}")))?; + (stored, inserted, Vec::new()) + } else { + let thread_params = thread_meta.as_ref().map(|m| m.as_params()); + let insert_result = + if let (Some(ch_id), Some(sender)) = (channel_id, dm_message_sender.as_ref()) { + state + .db + .insert_dm_event_with_thread_metadata( + tenant.community(), + &event, + ch_id, + thread_params, + sender, + ) + .await + } else { + state + .db + .insert_event_with_thread_metadata( + tenant.community(), + &event, + channel_id, + thread_params, + ) + .await + .map(|(stored, inserted)| (stored, inserted, Vec::new())) + }; + match insert_result { + Ok(result) => result, + Err(e) => { + // Compensate: if we pre-created a channel for kind:9007, + // soft-delete it so no orphaned channel row remains. + if let Some(ch_id) = pre_created_channel { + if let Err(re) = state + .db + .soft_delete_channel(tenant.community(), ch_id) + .await + { + warn!(event_id = %event_id_hex, "channel compensation failed: {re}"); + } + state.invalidate_channel_deleted(tenant); + } + return Err(match e { + buzz_db::DbError::AuthEventRejected => { + IngestError::Rejected("invalid: AUTH events cannot be stored".into()) + } + other => IngestError::Internal(format!("error: database error: {other}")), + }); + } + } + }; + if !was_inserted { return Ok(IngestResult { event_id: event_id_hex, @@ -3103,22 +3134,19 @@ async fn ingest_event_inner( }); } - if matches!(kind_u32, KIND_STREAM_MESSAGE | KIND_STREAM_MESSAGE_V2) { - if let (Some(ch_id), Some(channel)) = (channel_id, channel_row.as_ref()) { - if channel.channel_type == "dm" { - let sender = effective_message_author(&event, &state.relay_keypair.public_key()); - if let Err(e) = crate::handlers::side_effects::resurface_dm_for_message_recipients( - tenant, state, ch_id, &sender, - ) - .await - { - error!( - event_id = %event_id_hex, - channel_id = %ch_id, - "Failed to resurface DM for message recipients: {e}" - ); - } - } + if !resurfaced_viewers.is_empty() { + if let Err(e) = + crate::handlers::side_effects::publish_dm_visibility_snapshots_for_recipients( + tenant, + state, + &resurfaced_viewers, + ) + .await + { + error!( + event_id = %event_id_hex, + "Failed to publish resurfaced DM visibility snapshots: {e}" + ); } } diff --git a/crates/buzz-relay/src/handlers/moderation_notices.rs b/crates/buzz-relay/src/handlers/moderation_notices.rs index 11af521820..fe41e65c65 100644 --- a/crates/buzz-relay/src/handlers/moderation_notices.rs +++ b/crates/buzz-relay/src/handlers/moderation_notices.rs @@ -29,7 +29,11 @@ use buzz_core::kind::{event_kind_u32, KIND_STREAM_MESSAGE}; use buzz_core::tenant::TenantContext; use super::event::dispatch_persistent_event; -use super::side_effects::{emit_group_discovery_events, publish_dm_visibility_snapshot}; +#[cfg(test)] +use super::side_effects::publish_dm_visibility_snapshot; +use super::side_effects::{ + emit_group_discovery_events, publish_dm_visibility_snapshots_for_recipients, +}; use crate::state::AppState; /// Tag naming the moderation source row (report/action) a notice was derived @@ -142,18 +146,6 @@ pub async fn send_moderation_notice( .increment(1); } - // Resurface the moderation DM for the recipient. `open_dm` only clears - // `hidden_at` for `created_by` (the relay key), so a user who hid the - // "{host} Moderation" thread would never see a later ban/resolution notice. - // The closed-loop trust requirement needs the notice to reappear. - state - .db - .unhide_dm(tenant.community(), dm_channel_id, recipient_pubkey) - .await?; - if let Err(e) = publish_dm_visibility_snapshot(tenant, state, recipient_pubkey).await { - warn!(error = %e, "moderation DM visibility snapshot failed (continuing)"); - } - // 2. Ensure the relay's "{host} Moderation" kind:0 profile exists, and 3. // the DM's kind:39000 discovery (with `hidden` / `t=dm` / `p`). Both are // replaceable events, so we emit them on EVERY send rather than gating on @@ -182,11 +174,26 @@ pub async fn send_moderation_notice( .sign_with_keys(&state.relay_keypair) .map_err(|e| anyhow::anyhow!("failed to sign moderation notice: {e}"))?; - let (stored, _inserted) = state + // Commit the accepted notice and recipient resurface together. Discovery + // runs first, so any failure before this transaction leaves a previously + // hidden moderation conversation hidden rather than exposing stale content. + let (stored, _inserted, resurfaced_viewers) = state .db - .insert_event(tenant.community(), &event, Some(dm_channel_id)) + .insert_dm_event_with_thread_metadata( + tenant.community(), + &event, + dm_channel_id, + None, + relay_pubkey_bytes.as_slice(), + ) .await?; + if let Err(e) = + publish_dm_visibility_snapshots_for_recipients(tenant, state, &resurfaced_viewers).await + { + warn!(error = %e, "moderation DM visibility snapshot failed (continuing)"); + } + let kind_u32 = event_kind_u32(&stored.event); dispatch_persistent_event(tenant, state, &stored, kind_u32, &relay_pubkey_hex, None).await; @@ -489,6 +496,64 @@ mod tests { kind: "timeout".into(), public_reason: "Cool off.".into(), }; + + let relay_pubkey_bytes = relay_keys.public_key().to_bytes(); + let (preexisting_dm, _) = state + .db + .open_dm( + tenant.community(), + &[recipient_bytes.as_slice()], + relay_pubkey_bytes.as_slice(), + ) + .await + .expect("open moderation DM before injected failure"); + state + .db + .hide_dm( + tenant.community(), + preexisting_dm.id, + recipient_bytes.as_slice(), + ) + .await + .expect("hide moderation DM before injected failure"); + sqlx::raw_sql( + "CREATE FUNCTION fail_moderation_notice_insert() RETURNS trigger \ + LANGUAGE plpgsql AS $$ BEGIN \ + IF NEW.kind = 9 THEN RAISE EXCEPTION 'injected notice failure'; END IF; \ + RETURN NEW; \ + END $$; \ + CREATE TRIGGER fail_moderation_notice_insert \ + BEFORE INSERT ON events FOR EACH ROW \ + EXECUTE FUNCTION fail_moderation_notice_insert();", + ) + .execute(&pool) + .await + .expect("install notice failure injection"); + send_moderation_notice(&tenant, &state, recipient_bytes.as_slice(), notice.clone()) + .await + .expect_err("notice insertion failure must be returned"); + let hidden_after_failure: bool = sqlx::query_scalar( + "SELECT hidden_at IS NOT NULL FROM channel_members \ + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", + ) + .bind(tenant.community().as_uuid()) + .bind(preexisting_dm.id) + .bind(recipient_bytes.as_slice()) + .fetch_one(&pool) + .await + .expect("read hidden state after notice failure"); + assert!( + hidden_after_failure, + "failed notice insertion must not expose stale moderation content" + ); + sqlx::raw_sql( + "DROP TRIGGER fail_moderation_notice_insert ON events; \ + DROP FUNCTION fail_moderation_notice_insert();", + ) + .execute(&pool) + .await + .expect("remove notice failure injection"); + send_moderation_notice(&tenant, &state, recipient_bytes.as_slice(), notice.clone()) .await .expect("deliver notice"); diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index a174a2be7a..0924074fd2 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3467,22 +3467,16 @@ pub async fn reconcile_dm_visibility_snapshots(state: &Arc) -> anyhow: Ok(reconciled) } -/// Resurface a DM for recipients of a newly accepted message. +/// Publish fresh NIP-DV snapshots after recipient visibility mutations commit. /// -/// Hidden state is per viewer, so only active members other than the effective -/// message author are changed. Fresh NIP-DV snapshots make that server-side -/// change observable immediately to every connected client and across devices. -pub async fn resurface_dm_for_message_recipients( +/// The caller supplies only viewers whose canonical hidden state changed in +/// the accepted-message transaction. Publication remains recoverable through +/// the durable dirty-viewer queue if any snapshot attempt fails. +pub async fn publish_dm_visibility_snapshots_for_recipients( tenant: &TenantContext, state: &Arc, - channel_id: Uuid, - sender_pubkey: &[u8], + recipients: &[Vec], ) -> anyhow::Result<()> { - let recipients = state - .db - .unhide_dm_recipients(tenant.community(), channel_id, sender_pubkey) - .await?; - let results = futures_util::future::join_all( recipients .iter() diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 2e10473205..8608eae2c1 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -17,7 +17,7 @@ use tracing::{info, warn}; use uuid::Uuid; use crate::handlers::event::dispatch_persistent_event; -use crate::handlers::side_effects::resurface_dm_for_message_recipients; +use crate::handlers::side_effects::publish_dm_visibility_snapshots_for_recipients; use crate::state::AppState; /// Resolves `@Name` mentions in workflow message text to the pubkeys of the @@ -388,35 +388,47 @@ impl ActionSink for RelayActionSink { }, }); - let (stored_event, was_inserted) = state - .db - .insert_event_with_thread_metadata( - tenant.community(), - &event, - Some(channel_uuid), - thread_meta, - ) - .await + let (stored_event, was_inserted, resurfaced_viewers) = + if channel.channel_type == "dm" { + state + .db + .insert_dm_event_with_thread_metadata( + tenant.community(), + &event, + channel_uuid, + thread_meta, + &author_pubkey_bytes, + ) + .await + } else { + state + .db + .insert_event_with_thread_metadata( + tenant.community(), + &event, + Some(channel_uuid), + thread_meta, + ) + .await + .map(|(stored, inserted)| (stored, inserted, Vec::new())) + } .map_err(|e| ActionSinkError::Database(e.to_string()))?; // 5. Post-persist side effects (fan-out, search, audit) // Only if actually inserted (idempotency guard). if was_inserted { - if channel.channel_type == "dm" { - if let Err(e) = resurface_dm_for_message_recipients( - &tenant, - &state, - channel_uuid, - &author_pubkey_bytes, - ) - .await - { - warn!( - event_id = %event_id_hex, - channel_id = %channel_uuid, - "Workflow DM recipient resurface failed: {e}" - ); - } + if let Err(e) = publish_dm_visibility_snapshots_for_recipients( + &tenant, + &state, + &resurfaced_viewers, + ) + .await + { + warn!( + event_id = %event_id_hex, + channel_id = %channel_uuid, + "Workflow DM visibility snapshot publication failed: {e}" + ); } let _ = dispatch_persistent_event( diff --git a/migrations/0033_dm_visibility_dirty_viewers.sql b/migrations/0033_dm_visibility_dirty_viewers.sql index c2d8234196..7b0c3eebcf 100644 --- a/migrations/0033_dm_visibility_dirty_viewers.sql +++ b/migrations/0033_dm_visibility_dirty_viewers.sql @@ -37,7 +37,8 @@ SELECT attach_community_write_fence('dm_visibility_dirty_viewers'); -- Brownfield deployments can already contain canonical hidden-DM state whose -- relay-authored snapshot drifted before the durable queue existed. Enqueue --- each active DM viewer once so the bounded worker repairs that legacy state. +-- each DM viewer once so the bounded worker repairs that legacy state. Rows +-- for archived communities remain durable but unclaimable until unarchive. INSERT INTO dm_visibility_dirty_viewers (community_id, viewer) SELECT cm.community_id, cm.pubkey FROM channel_members cm @@ -48,7 +49,6 @@ JOIN communities community ON community.id = cm.community_id WHERE cm.removed_at IS NULL AND c.channel_type = 'dm' AND c.deleted_at IS NULL - AND community.archived_at IS NULL AND community.deleted_at IS NULL AND community.deletion_state = 'active' GROUP BY cm.community_id, cm.pubkey