jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-13 04:41:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(relay): scope DM command writes by community

Browse Source 
Co-authored-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co>
Signed-off-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf
2026-06-26 23:12:04 -04:00
committed by npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d
parent 989f87e0ab
commit 2738faf9b2
4 changed files with 124 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -10,6 +10,7 @@ use uuid::Uuid;
use crate::channel::ChannelRecord;
use crate::error::{DbError, Result};
use buzz_core::CommunityId;
// -- Public structs -----------------------------------------------------------
@@ -63,6 +64,7 @@ pub fn compute_participant_hash(pubkeys: &[&[u8]]) -> [u8; 32] {
/// Returns `None` if no matching DM exists or if it has been deleted.
pub async fn find_dm_by_participants(
pool: &PgPool,
community_id: CommunityId,
participant_hash: &[u8],
) -> Result<Option<ChannelRecord>> {
let row = sqlx::query(
@@ -74,12 +76,14 @@ pub async fn find_dm_by_participants(
topic, topic_set_by, topic_set_at,
purpose, purpose_set_by, purpose_set_at
FROM channels
WHERE participant_hash = $1
WHERE community_id = $1
AND participant_hash = $2
AND channel_type = 'dm'
AND deleted_at IS NULL
LIMIT 1
"#,
)
.bind(community_id.as_uuid())
.bind(participant_hash)
.fetch_optional(pool)
.await?;
@@ -96,6 +100,7 @@ pub async fn find_dm_by_participants(
/// - The operation is idempotent: same participant set -> same channel returned.
pub async fn create_dm(
pool: &PgPool,
community_id: CommunityId,
participants: &[&[u8]],
created_by: &[u8],
) -> Result<ChannelRecord> {
@@ -132,12 +137,14 @@ pub async fn create_dm(
topic, topic_set_by, topic_set_at,
purpose, purpose_set_by, purpose_set_at
FROM channels
WHERE participant_hash = $1
WHERE community_id = $1
AND participant_hash = $2
AND channel_type = 'dm'
AND deleted_at IS NULL
LIMIT 1
"#,
)
.bind(community_id.as_uuid())
.bind(hash.as_slice())
.fetch_optional(&mut *tx)
.await?;
@@ -159,11 +166,12 @@ pub async fn create_dm(
sqlx::query(
r#"
INSERT INTO channels
(id, name, channel_type, visibility, created_by, participant_hash)
VALUES ($1, $2, 'dm', 'private', $3, $4)
(id, community_id, name, channel_type, visibility, created_by, participant_hash)
VALUES ($1, $2, $3, 'dm', 'private', $4, $5)
"#,
)
.bind(id)
.bind(community_id.as_uuid())
.bind(&name)
.bind(created_by)
.bind(hash.as_slice())
@@ -174,14 +182,15 @@ pub async fn create_dm(
for pk in participants {
sqlx::query(
r#"
INSERT INTO channel_members (channel_id, pubkey, role, invited_by)
VALUES ($1, $2, 'member', $3)
ON CONFLICT (channel_id, pubkey) DO UPDATE SET
INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by)
VALUES ($1, $2, $3, 'member', $4)
ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET
removed_at = NULL,
removed_by = NULL,
role = EXCLUDED.role
"#,
)
.bind(community_id.as_uuid())
.bind(id)
.bind(*pk)
.bind(created_by)
@@ -197,9 +206,10 @@ pub async fn create_dm(
nip29_group_id, topic_required, max_members,
topic, topic_set_by, topic_set_at,
purpose, purpose_set_by, purpose_set_at
FROM channels WHERE id = $1
FROM channels WHERE community_id = $1 AND id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(id)
.fetch_one(&mut *tx)
.await?;
@@ -215,6 +225,7 @@ pub async fn create_dm(
/// using `updated_at` ordering.
pub async fn list_dms_for_user(
pool: &PgPool,
community_id: CommunityId,
pubkey: &[u8],
limit: u32,
cursor: Option<Uuid>,
@@ -223,10 +234,12 @@ pub async fn list_dms_for_user(
// Resolve cursor to a timestamp for keyset pagination.
let cursor_ts: Option<DateTime<Utc>> = if let Some(cid) = cursor {
let row = sqlx::query("SELECT updated_at FROM channels WHERE id = $1")
.bind(cid)
.fetch_optional(pool)
.await?;
let row =
sqlx::query("SELECT updated_at FROM channels WHERE community_id = $1 AND id = $2")
.bind(community_id.as_uuid())
.bind(cid)
.fetch_optional(pool)
.await?;
row.map(|r| r.try_get::<DateTime<Utc>, _>("updated_at"))
.transpose()?
} else {
@@ -240,17 +253,20 @@ pub async fn list_dms_for_user(
SELECT c.id, c.created_at, c.updated_at
FROM channels c
JOIN channel_members cm
ON c.id = cm.channel_id
AND cm.pubkey = $1
ON c.community_id = cm.community_id
AND c.id = cm.channel_id
AND cm.pubkey = $2
AND cm.removed_at IS NULL
AND cm.hidden_at IS NULL
WHERE c.channel_type = 'dm'
WHERE c.community_id = $1
AND c.channel_type = 'dm'
AND c.deleted_at IS NULL
AND c.updated_at < $2
AND c.updated_at < $3
ORDER BY c.updated_at DESC
LIMIT $3
LIMIT $4
"#,
)
.bind(community_id.as_uuid())
.bind(pubkey)
.bind(ts)
.bind(limit)
@@ -262,16 +278,19 @@ pub async fn list_dms_for_user(
SELECT c.id, c.created_at, c.updated_at
FROM channels c
JOIN channel_members cm
ON c.id = cm.channel_id
AND cm.pubkey = $1
ON c.community_id = cm.community_id
AND c.id = cm.channel_id
AND cm.pubkey = $2
AND cm.removed_at IS NULL
AND cm.hidden_at IS NULL
WHERE c.channel_type = 'dm'
WHERE c.community_id = $1
AND c.channel_type = 'dm'
AND c.deleted_at IS NULL
ORDER BY c.updated_at DESC
LIMIT $2
LIMIT $3
"#,
)
.bind(community_id.as_uuid())
.bind(pubkey)
.bind(limit)
.fetch_all(pool)
@@ -290,12 +309,16 @@ pub async fn list_dms_for_user(
r#"
SELECT cm.pubkey, cm.role::text AS role, u.display_name
FROM channel_members cm
LEFT JOIN users u ON cm.pubkey = u.pubkey
WHERE cm.channel_id = $1
LEFT JOIN users u
ON u.community_id = cm.community_id
AND u.pubkey = cm.pubkey
WHERE cm.community_id = $1
AND cm.channel_id = $2
AND cm.removed_at IS NULL
ORDER BY cm.joined_at ASC
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_all(pool)
.await?;
@@ -332,6 +355,7 @@ pub async fn list_dms_for_user(
/// - `was_created = false` -- an existing DM was returned.
pub async fn open_dm(
pool: &PgPool,
community_id: CommunityId,
pubkeys: &[&[u8]],
created_by: &[u8],
) -> Result<(ChannelRecord, bool)> {
@@ -351,14 +375,14 @@ pub async fn open_dm(
let hash = compute_participant_hash(&all);
// Check for existing DM first (fast path, no transaction).
if let Some(existing) = find_dm_by_participants(pool, &hash).await? {
if let Some(existing) = find_dm_by_participants(pool, community_id, &hash).await? {
// Clear hidden_at for the caller so the DM reappears in their sidebar.
unhide_dm(pool, existing.id, created_by).await?;
unhide_dm(pool, community_id, existing.id, created_by).await?;
return Ok((existing, false));
}
// Create new DM.
let channel = create_dm(pool, &all, created_by).await?;
let channel = create_dm(pool, community_id, &all, created_by).await?;
Ok((channel, true))
}
@@ -370,14 +394,20 @@ pub async fn open_dm(
/// The DM is not deleted — it can be restored by opening a new DM with the
/// same participants (which clears `hidden_at`). Returns an error if the user
/// is not an active member of the channel.
pub async fn hide_dm(pool: &PgPool, channel_id: Uuid, pubkey: &[u8]) -> Result<()> {
pub async fn hide_dm(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<()> {
let result = sqlx::query(
r#"
UPDATE channel_members
SET hidden_at = NOW()
WHERE channel_id = $1 AND pubkey = $2 AND removed_at IS NULL
WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 AND removed_at IS NULL
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(pubkey)
.execute(pool)
@@ -396,14 +426,20 @@ pub async fn hide_dm(pool: &PgPool, channel_id: Uuid, pubkey: &[u8]) -> Result<(
///
/// This is called automatically when a user re-opens a DM via [`open_dm`].
/// It is a no-op if the membership is not currently hidden.
pub async fn unhide_dm(pool: &PgPool, channel_id: Uuid, pubkey: &[u8]) -> Result<()> {
pub async fn unhide_dm(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<()> {
sqlx::query(
r#"
UPDATE channel_members
SET hidden_at = NULL
WHERE channel_id = $1 AND pubkey = $2 AND removed_at IS NULL
WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 AND removed_at IS NULL
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(pubkey)
.execute(pool)
@@ -415,13 +451,20 @@ pub async fn unhide_dm(pool: &PgPool, channel_id: Uuid, pubkey: &[u8]) -> Result
/// Return the channel IDs of all DMs the given user currently has hidden
/// (`hidden_at IS NOT NULL`) while still being an active member. Used to build
/// the relay-signed NIP-DV visibility snapshot.
pub async fn list_hidden_dms(pool: &PgPool, pubkey: &[u8]) -> Result<Vec<Uuid>> {
pub async fn list_hidden_dms(
pool: &PgPool,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Vec<Uuid>> {
let rows = sqlx::query(
r#"
SELECT cm.channel_id
FROM channel_members cm
JOIN channels c ON c.id = cm.channel_id
WHERE cm.pubkey = $1
JOIN channels c
ON c.community_id = cm.community_id
AND c.id = cm.channel_id
WHERE cm.community_id = $1
AND cm.pubkey = $2
AND cm.removed_at IS NULL
AND cm.hidden_at IS NOT NULL
AND c.channel_type = 'dm'
@@ -429,6 +472,7 @@ pub async fn list_hidden_dms(pool: &PgPool, pubkey: &[u8]) -> Result<Vec<Uuid>>
ORDER BY cm.channel_id
"#,
)
.bind(community_id.as_uuid())
.bind(pubkey)
.fetch_all(pool)
.await?;

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

@@ -889,55 +889,73 @@ impl Db {
/// Find an existing DM by its participant hash.
pub async fn find_dm_by_participants(
&self,
community_id: CommunityId,
participant_hash: &[u8],
) -> Result<Option<channel::ChannelRecord>> {
dm::find_dm_by_participants(&self.pool, participant_hash).await
dm::find_dm_by_participants(&self.pool, community_id, participant_hash).await
}
/// Create or return an existing DM channel.
pub async fn create_dm(
&self,
community_id: CommunityId,
participants: &[&[u8]],
created_by: &[u8],
) -> Result<channel::ChannelRecord> {
dm::create_dm(&self.pool, participants, created_by).await
dm::create_dm(&self.pool, community_id, participants, created_by).await
}
/// List all DMs for a user.
pub async fn list_dms_for_user(
&self,
community_id: CommunityId,
pubkey: &[u8],
limit: u32,
cursor: Option<Uuid>,
) -> Result<Vec<dm::DmRecord>> {
dm::list_dms_for_user(&self.pool, pubkey, limit, cursor).await
dm::list_dms_for_user(&self.pool, community_id, pubkey, limit, cursor).await
}
/// Open or retrieve a DM for the given participants.
pub async fn open_dm(
&self,
community_id: CommunityId,
pubkeys: &[&[u8]],
created_by: &[u8],
) -> Result<(channel::ChannelRecord, bool)> {
dm::open_dm(&self.pool, pubkeys, created_by).await
dm::open_dm(&self.pool, community_id, pubkeys, created_by).await
}
/// Hide a DM channel for a specific user.
///
/// The DM is not deleted — it can be restored by opening a new DM with
/// the same participants.
pub async fn hide_dm(&self, channel_id: Uuid, pubkey: &[u8]) -> Result<()> {
dm::hide_dm(&self.pool, channel_id, pubkey).await
pub async fn hide_dm(
&self,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<()> {
dm::hide_dm(&self.pool, community_id, channel_id, pubkey).await
}
/// Unhide a DM channel for a specific user.
pub async fn unhide_dm(&self, channel_id: Uuid, pubkey: &[u8]) -> Result<()> {
dm::unhide_dm(&self.pool, channel_id, pubkey).await
pub async fn unhide_dm(
&self,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<()> {
dm::unhide_dm(&self.pool, community_id, channel_id, pubkey).await
}
/// List the channel IDs of all DMs the given user currently has hidden.
pub async fn list_hidden_dms(&self, pubkey: &[u8]) -> Result<Vec<Uuid>> {
dm::list_hidden_dms(&self.pool, pubkey).await
pub async fn list_hidden_dms(
&self,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Vec<Uuid>> {
dm::list_hidden_dms(&self.pool, community_id, pubkey).await
}
/// Insert thread metadata.

32
crates/buzz-relay/src/handlers/command_executor.rs
View File

@@ -56,8 +56,8 @@ pub async fn handle_command(
KIND_DM_HIDE => handle_dm_hide(tenant, state, &event, &auth).await,
KIND_WORKFLOW_DEF => handle_workflow_def(tenant, state, &event, &auth).await,
KIND_WORKFLOW_TRIGGER => handle_workflow_trigger(tenant, state, &event, &auth).await,
KIND_APPROVAL_GRANT => handle_approval_grant(state, &event, &auth).await,
KIND_APPROVAL_DENY => handle_approval_deny(state, &event, &auth).await,
KIND_APPROVAL_GRANT => handle_approval_grant(tenant, state, &event, &auth).await,
KIND_APPROVAL_DENY => handle_approval_deny(tenant, state, &event, &auth).await,
_ => Err(IngestError::Rejected(format!(
"unknown command kind: {kind}"
))),
@@ -91,6 +91,7 @@ enum PersistResult {
/// as the natural dedup key.
async fn persist_command_event(
state: &Arc<AppState>,
tenant: &TenantContext,
event: &Event,
) -> Result<PersistResult, IngestError> {
let channel_id = extract_channel_id(event);
@@ -129,11 +130,12 @@ async fn persist_command_event(
let result = sqlx::query(
r#"
INSERT INTO events (id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
ON CONFLICT DO NOTHING
"#,
)
.bind(tenant.community().as_uuid())
.bind(id_bytes.as_slice())
.bind(pubkey_bytes.as_slice())
.bind(created_at)
@@ -271,7 +273,7 @@ async fn handle_dm_open(
}
// Persist the command event (idempotency) — returns open transaction
let tx = match persist_command_event(state, event).await? {
let tx = match persist_command_event(state, tenant, event).await? {
PersistResult::Duplicate => {
return Ok(IngestResult {
event_id: event.id.to_hex(),
@@ -286,7 +288,7 @@ async fn handle_dm_open(
let all_refs: Vec<&[u8]> = all_bytes.iter().map(|b| b.as_slice()).collect();
let (channel, was_created) = state
.db
.open_dm(&all_refs, &self_bytes)
.open_dm(tenant.community(), &all_refs, &self_bytes)
.await
.map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?;
@@ -425,7 +427,7 @@ async fn handle_dm_add_member(
}
// Persist the command event — returns open transaction
let tx = match persist_command_event(state, event).await? {
let tx = match persist_command_event(state, tenant, event).await? {
PersistResult::Duplicate => {
return Ok(IngestResult {
event_id: event.id.to_hex(),
@@ -440,7 +442,7 @@ async fn handle_dm_add_member(
let all_refs: Vec<&[u8]> = all_bytes.iter().map(|b| b.as_slice()).collect();
let (new_channel, was_created) = state
.db
.open_dm(&all_refs, &self_bytes)
.open_dm(tenant.community(), &all_refs, &self_bytes)
.await
.map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?;
@@ -524,7 +526,7 @@ async fn handle_dm_hide(
}
// Persist the command event — returns open transaction
let tx = match persist_command_event(state, event).await? {
let tx = match persist_command_event(state, tenant, event).await? {
PersistResult::Duplicate => {
return Ok(IngestResult {
event_id: event.id.to_hex(),
@@ -538,7 +540,7 @@ async fn handle_dm_hide(
// 4. Execute: hide_dm
state
.db
.hide_dm(channel_id, &self_bytes)
.hide_dm(tenant.community(), channel_id, &self_bytes)
.await
.map_err(|e| IngestError::Internal(format!("error: db hide_dm: {e}")))?;
@@ -614,7 +616,7 @@ async fn handle_workflow_def(
let hash = compute_definition_hash(&definition_json_final);
// Persist the command event — returns open transaction
let tx = match persist_command_event(state, event).await? {
let tx = match persist_command_event(state, tenant, event).await? {
PersistResult::Duplicate => {
return Ok(IngestResult {
event_id: event.id.to_hex(),
@@ -710,7 +712,7 @@ async fn handle_workflow_trigger(
}
// Persist the command event — returns open transaction
let tx = match persist_command_event(state, event).await? {
let tx = match persist_command_event(state, tenant, event).await? {
PersistResult::Duplicate => {
return Ok(IngestResult {
event_id: event.id.to_hex(),
@@ -843,6 +845,7 @@ fn check_approver_spec(approver_spec: &str, requester_hex: &str) -> Result<(), I
}
async fn handle_approval_grant(
tenant: &TenantContext,
state: &Arc<AppState>,
event: &Event,
auth: &IngestAuth,
@@ -885,7 +888,7 @@ async fn handle_approval_grant(
check_approver_spec(&approval.approver_spec, &self_hex)?;
// Persist the command event — returns open transaction
let tx = match persist_command_event(state, event).await? {
let tx = match persist_command_event(state, tenant, event).await? {
PersistResult::Duplicate => {
return Ok(IngestResult {
event_id: event.id.to_hex(),
@@ -951,6 +954,7 @@ async fn handle_approval_grant(
}
async fn handle_approval_deny(
tenant: &TenantContext,
state: &Arc<AppState>,
event: &Event,
auth: &IngestAuth,
@@ -992,7 +996,7 @@ async fn handle_approval_deny(
check_approver_spec(&approval.approver_spec, &self_hex)?;
// Persist the command event — returns open transaction
let tx = match persist_command_event(state, event).await? {
let tx = match persist_command_event(state, tenant, event).await? {
PersistResult::Duplicate => {
return Ok(IngestResult {
event_id: event.id.to_hex(),

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

@@ -2568,7 +2568,7 @@ pub async fn publish_dm_visibility_snapshot(
viewer: &[u8],
) -> anyhow::Result<()> {
let viewer_hex = hex::encode(viewer);
let hidden = state.db.list_hidden_dms(viewer).await?;
let hidden = state.db.list_hidden_dms(tenant.community(), viewer).await?;
let relay_pubkey_hex = state.relay_keypair.public_key().to_hex();
let mut tags: Vec<Tag> = Vec::with_capacity(hidden.len() + 2);