jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-31 06:21:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make DM resurfacing atomic with message acceptance

Browse Source 
Signed-off-by: kenny lopez <klopez4212@gmail.com>
This commit is contained in:
kenny lopez
2026-08-21 12:55:52 +01:00
parent 1101c6f20e
commit 896378c902
8 changed files with 544 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
crates/buzz-db/src/dm.rs
View File

@@ -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<Vec<Vec<u8>>> {
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<Vec<Vec<u8>>> {
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)

198
crates/buzz-db/src/lib.rs
View File

@@ -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<event::ThreadMetadataParams<'_>>,
sender_pubkey: &[u8],
) -> Result<(StoredEvent, bool, Vec<Vec<u8>>)> {
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<u8>, 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",
)

25
crates/buzz-db/src/migration.rs
View File

@@ -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

172
crates/buzz-relay/src/handlers/ingest.rs
View File

@@ -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}"
);
}
}

95
crates/buzz-relay/src/handlers/moderation_notices.rs
View File

@@ -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");

18
crates/buzz-relay/src/handlers/side_effects.rs
View File

@@ -3467,22 +3467,16 @@ pub async fn reconcile_dm_visibility_snapshots(state: &Arc<AppState>) -> 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<AppState>,
channel_id: Uuid,
sender_pubkey: &[u8],
recipients: &[Vec<u8>],
) -> 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()

62
crates/buzz-relay/src/workflow_sink.rs
View File

@@ -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(

4
migrations/0033_dm_visibility_dirty_viewers.sql
View File

@@ -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