External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(db): require community scope for row lookups

Browse Source 
Co-authored-by: Mari <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co>
Signed-off-by: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com>
This commit is contained in:
tlongwell-block
2026-06-26 18:59:03 -04:00
co-authored by Mari
parent 1850af8424
commit e55c2ceae0
4 changed files with 708 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-db/src/channel.rs
+261 -70
View File
@@ -256,8 +256,12 @@ pub async fn create_channel_with_id(
Ok((record, was_created))
}
/// Fetches a channel record by ID. Returns `ChannelNotFound` if missing or deleted.
pub async fn get_channel(pool: &PgPool, channel_id: Uuid) -> Result<ChannelRecord> {
/// Fetches a channel record by `(community_id, id)`. Returns `ChannelNotFound` if missing or deleted.
pub async fn get_channel(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<ChannelRecord> {
let row = sqlx::query(
r#"
SELECT id, name, channel_type::text AS channel_type, visibility::text AS visibility,
@@ -267,9 +271,10 @@ pub async fn get_channel(pool: &PgPool, channel_id: Uuid) -> Result<ChannelRecor
topic, topic_set_by, topic_set_at,
purpose, purpose_set_by, purpose_set_at,
ttl_seconds, ttl_deadline
FROM channels WHERE id = $1 AND deleted_at IS NULL
FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_optional(pool)
.await?
@@ -279,19 +284,34 @@ pub async fn get_channel(pool: &PgPool, channel_id: Uuid) -> Result<ChannelRecor
}
/// Returns the canvas content for a channel, if any.
pub async fn get_canvas(pool: &PgPool, channel_id: Uuid) -> Result<Option<String>> {
let row = sqlx::query("SELECT canvas FROM channels WHERE id = $1 AND deleted_at IS NULL")
.bind(channel_id)
.fetch_optional(pool)
.await?
.ok_or(DbError::ChannelNotFound(channel_id))?;
pub async fn get_canvas(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<Option<String>> {
let row = sqlx::query(
"SELECT canvas FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_optional(pool)
.await?
.ok_or(DbError::ChannelNotFound(channel_id))?;
Ok(row.try_get("canvas")?)
}
/// Sets or clears the canvas content for a channel.
pub async fn set_canvas(pool: &PgPool, channel_id: Uuid, canvas: Option<&str>) -> Result<()> {
let rows = sqlx::query("UPDATE channels SET canvas = $1 WHERE id = $2 AND deleted_at IS NULL")
pub async fn set_canvas(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
canvas: Option<&str>,
) -> Result<()> {
let rows = sqlx::query(
"UPDATE channels SET canvas = $1 WHERE community_id = $2 AND id = $3 AND deleted_at IS NULL",
)
.bind(canvas)
.bind(community_id.as_uuid())
.bind(channel_id)
.execute(pool)
.await?;
@@ -329,7 +349,7 @@ pub async fn add_member(
let mut tx = pool.begin().await?;
let channel = get_channel_tx(&mut tx, channel_id).await?;
let channel = get_channel_tx(&mut tx, community_id, channel_id).await?;
let effective_role = if channel.visibility == "private" {
let inviter = invited_by.ok_or_else(|| {
@@ -497,12 +517,18 @@ pub async fn remove_member(
}
/// Returns `true` if the given pubkey is an active member of the channel.
pub async fn is_member(pool: &PgPool, channel_id: Uuid, pubkey: &[u8]) -> Result<bool> {
pub async fn is_member(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<bool> {
let row = sqlx::query(
"SELECT COUNT(*) as cnt FROM channel_members cm \
JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL \
WHERE cm.channel_id = $1 AND cm.pubkey = $2 AND cm.removed_at IS NULL",
JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \
WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.pubkey = $3 AND cm.removed_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(pubkey)
.fetch_one(pool)
@@ -514,17 +540,22 @@ pub async fn is_member(pool: &PgPool, channel_id: Uuid, pubkey: &[u8]) -> Result
/// Returns all active members of the given channel.
///
/// Returns an empty list if the channel has been soft-deleted.
pub async fn get_members(pool: &PgPool, channel_id: Uuid) -> Result<Vec<MemberRecord>> {
pub async fn get_members(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<Vec<MemberRecord>> {
let rows = sqlx::query(
r#"
SELECT cm.channel_id, cm.pubkey, cm.role::text AS role, cm.joined_at, cm.invited_by, cm.removed_at
FROM channel_members cm
JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL
WHERE cm.channel_id = $1 AND cm.removed_at IS NULL
JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL
WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.removed_at IS NULL
ORDER BY cm.joined_at ASC
LIMIT 1000
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_all(pool)
.await?;
@@ -538,7 +569,11 @@ pub async fn get_members(pool: &PgPool, channel_id: Uuid) -> Result<Vec<MemberRe
/// Returns a flat `Vec<MemberRecord>` ordered by `joined_at`; callers should
/// group by `channel_id` if per-channel access is needed.
/// Returns an empty vec immediately when `channel_ids` is empty.
pub async fn get_members_bulk(pool: &PgPool, channel_ids: &[Uuid]) -> Result<Vec<MemberRecord>> {
pub async fn get_members_bulk(
pool: &PgPool,
community_id: CommunityId,
channel_ids: &[Uuid],
) -> Result<Vec<MemberRecord>> {
if channel_ids.is_empty() {
return Ok(Vec::new());
}
@@ -546,11 +581,12 @@ pub async fn get_members_bulk(pool: &PgPool, channel_ids: &[Uuid]) -> Result<Vec
r#"
SELECT cm.channel_id, cm.pubkey, cm.role::text AS role, cm.joined_at, cm.invited_by, cm.removed_at
FROM channel_members cm
JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL
WHERE cm.channel_id = ANY($1) AND cm.removed_at IS NULL
JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL
WHERE cm.community_id = $1 AND cm.channel_id = ANY($2) AND cm.removed_at IS NULL
ORDER BY cm.joined_at ASC
"#,
)
.bind(community_id.as_uuid())
.bind(channel_ids)
.fetch_all(pool)
.await?;
@@ -561,20 +597,25 @@ pub async fn get_members_bulk(pool: &PgPool, channel_ids: &[Uuid]) -> Result<Vec
///
/// Includes channels where the pubkey is an active member AND all open channels.
/// Open channels must be included in REQ filter resolution.
pub async fn get_accessible_channel_ids(pool: &PgPool, pubkey: &[u8]) -> Result<Vec<Uuid>> {
pub async fn get_accessible_channel_ids(
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 cm.channel_id = c.id AND c.deleted_at IS NULL
WHERE cm.pubkey = $1 AND cm.removed_at IS NULL
JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL
WHERE cm.community_id = $1 AND cm.pubkey = $2 AND cm.removed_at IS NULL
UNION
SELECT id AS channel_id
FROM channels
WHERE visibility = 'open' AND deleted_at IS NULL
WHERE community_id = $1 AND visibility = 'open' AND deleted_at IS NULL
LIMIT 1000
"#,
)
.bind(community_id.as_uuid())
.bind(pubkey)
.fetch_all(pool)
.await?;
@@ -587,8 +628,12 @@ pub async fn get_accessible_channel_ids(pool: &PgPool, pubkey: &[u8]) -> Result<
.collect()
}
/// Lists channels, optionally filtered by visibility string.
pub async fn list_channels(pool: &PgPool, visibility: Option<&str>) -> Result<Vec<ChannelRecord>> {
/// Lists channels in a community, optionally filtered by visibility string.
pub async fn list_channels(
pool: &PgPool,
community_id: CommunityId,
visibility: Option<&str>,
) -> Result<Vec<ChannelRecord>> {
let rows = if let Some(vis) = visibility {
sqlx::query(
r#"
@@ -600,11 +645,12 @@ pub async fn list_channels(pool: &PgPool, visibility: Option<&str>) -> Result<Ve
purpose, purpose_set_by, purpose_set_at,
ttl_seconds, ttl_deadline
FROM channels
WHERE deleted_at IS NULL AND visibility::text = $1
WHERE community_id = $1 AND deleted_at IS NULL AND visibility::text = $2
ORDER BY created_at DESC
LIMIT 1000
"#,
)
.bind(community_id.as_uuid())
.bind(vis)
.fetch_all(pool)
.await?
@@ -619,11 +665,12 @@ pub async fn list_channels(pool: &PgPool, visibility: Option<&str>) -> Result<Ve
purpose, purpose_set_by, purpose_set_at,
ttl_seconds, ttl_deadline
FROM channels
WHERE deleted_at IS NULL
WHERE community_id = $1 AND deleted_at IS NULL
ORDER BY created_at DESC
LIMIT 1000
"#,
)
.bind(community_id.as_uuid())
.fetch_all(pool)
.await?
};
@@ -653,6 +700,7 @@ async fn get_active_role_tx(
/// Transaction-aware variant of [`get_channel`].
async fn get_channel_tx(
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<ChannelRecord> {
let row = sqlx::query(
@@ -664,9 +712,10 @@ async fn get_channel_tx(
topic, topic_set_by, topic_set_at,
purpose, purpose_set_by, purpose_set_at,
ttl_seconds, ttl_deadline
FROM channels WHERE id = $1 AND deleted_at IS NULL
FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_optional(&mut **tx)
.await?
@@ -683,6 +732,15 @@ pub struct BotChannelEntry {
pub id: String,
}
/// A channel archived by the ephemeral-channel reaper.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ReapedEphemeralChannel {
/// Community that owns the archived channel.
pub community_id: CommunityId,
/// Archived channel UUID.
pub channel_id: Uuid,
}
/// Bot member record — a user with role=bot, with their channel memberships aggregated.
#[derive(Debug, Clone)]
pub struct BotMemberRecord {
@@ -730,6 +788,7 @@ pub struct AccessibleChannel {
/// that visibility value are returned. `None` returns all accessible channels.
pub async fn get_accessible_channels(
pool: &PgPool,
community_id: CommunityId,
pubkey: &[u8],
visibility_filter: Option<&str>,
member_only: Option<bool>,
@@ -756,20 +815,22 @@ pub async fn get_accessible_channels(
(cm.channel_id IS NOT NULL) AS is_member
FROM channels c
LEFT JOIN channel_members cm
ON c.id = cm.channel_id AND cm.pubkey = $1 AND cm.removed_at IS NULL
WHERE c.deleted_at IS NULL
ON c.community_id = cm.community_id AND c.id = cm.channel_id AND cm.pubkey = $2 AND cm.removed_at IS NULL
WHERE c.community_id = $1 AND c.deleted_at IS NULL
{membership_clause}
AND (c.channel_type != 'dm' OR cm.hidden_at IS NULL)
"#
);
let sql = if visibility_filter.is_some() {
format!("{base} AND c.visibility::text = $2\n ORDER BY array_position(ARRAY['stream','forum','dm']::text[], c.channel_type::text), c.name\n LIMIT 1000")
format!("{base} AND c.visibility::text = $3\n ORDER BY array_position(ARRAY['stream','forum','dm']::text[], c.channel_type::text), c.name\n LIMIT 1000")
} else {
format!("{base} ORDER BY array_position(ARRAY['stream','forum','dm']::text[], c.channel_type::text), c.name\n LIMIT 1000")
};
let query = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(pubkey);
let query = sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(community_id.as_uuid())
.bind(pubkey);
let query = if let Some(vis) = visibility_filter {
query.bind(vis)
} else {
@@ -786,24 +847,28 @@ pub async fn get_accessible_channels(
.collect()
}
/// Returns all bot-role members with their channel memberships.
/// Returns all bot-role members with their channel memberships in one community.
///
/// Channels are returned as a JSON array of `{name, id}` objects via `json_agg`,
/// preserving the 1:1 name↔UUID pairing. No separate string_agg ordering issues.
/// Members with no active channel memberships are excluded (INNER JOIN on channels).
pub async fn get_bot_members(pool: &PgPool) -> Result<Vec<BotMemberRecord>> {
pub async fn get_bot_members(
pool: &PgPool,
community_id: CommunityId,
) -> Result<Vec<BotMemberRecord>> {
let rows = sqlx::query(
r#"
SELECT cm.pubkey, u.display_name, u.agent_type, u.capabilities,
COALESCE(json_agg(DISTINCT jsonb_build_object('name', c.name, 'id', c.id::text)), '[]') AS channels_json
FROM channel_members cm
LEFT JOIN users u ON cm.pubkey = u.pubkey
JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL
WHERE cm.role = 'bot' AND cm.removed_at IS NULL
LEFT JOIN users u ON cm.community_id = u.community_id AND cm.pubkey = u.pubkey
JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL
WHERE cm.community_id = $1 AND cm.role = 'bot' AND cm.removed_at IS NULL
GROUP BY cm.pubkey, u.display_name, u.agent_type, u.capabilities
LIMIT 1000
"#,
)
.bind(community_id.as_uuid())
.fetch_all(pool)
.await?;
@@ -939,6 +1004,7 @@ pub struct ChannelUpdate {
/// Returns the updated `ChannelRecord` on success.
pub async fn update_channel(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
updates: ChannelUpdate,
) -> Result<ChannelRecord> {
@@ -980,8 +1046,9 @@ pub async fn update_channel(
None => set_parts.push("ttl_deadline = NULL".to_string()),
}
}
let channel_param_idx = param_idx + 1;
let sql = format!(
"UPDATE channels SET {}, updated_at = NOW() WHERE id = ${param_idx} AND deleted_at IS NULL",
"UPDATE channels SET {}, updated_at = NOW() WHERE community_id = ${param_idx} AND id = ${channel_param_idx} AND deleted_at IS NULL",
set_parts.join(", ")
);
@@ -998,6 +1065,7 @@ pub async fn update_channel(
if let Some(ref ttl) = updates.ttl_seconds {
q = q.bind(*ttl);
}
q = q.bind(community_id.as_uuid());
q = q.bind(channel_id);
let result = q.execute(pool).await?;
@@ -1005,17 +1073,24 @@ pub async fn update_channel(
return Err(DbError::ChannelNotFound(channel_id));
}
get_channel(pool, channel_id).await
get_channel(pool, community_id, channel_id).await
}
/// Sets the topic for a channel, recording who set it and when.
pub async fn set_topic(pool: &PgPool, channel_id: Uuid, topic: &str, set_by: &[u8]) -> Result<()> {
pub async fn set_topic(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
topic: &str,
set_by: &[u8],
) -> Result<()> {
let result = sqlx::query(
"UPDATE channels SET topic = $1, topic_set_by = $2, topic_set_at = NOW() \
WHERE id = $3 AND deleted_at IS NULL",
WHERE community_id = $3 AND id = $4 AND deleted_at IS NULL",
)
.bind(topic)
.bind(set_by)
.bind(community_id.as_uuid())
.bind(channel_id)
.execute(pool)
.await?;
@@ -1028,16 +1103,18 @@ pub async fn set_topic(pool: &PgPool, channel_id: Uuid, topic: &str, set_by: &[u
/// Sets the purpose for a channel, recording who set it and when.
pub async fn set_purpose(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
purpose: &str,
set_by: &[u8],
) -> Result<()> {
let result = sqlx::query(
"UPDATE channels SET purpose = $1, purpose_set_by = $2, purpose_set_at = NOW() \
WHERE id = $3 AND deleted_at IS NULL",
WHERE community_id = $3 AND id = $4 AND deleted_at IS NULL",
)
.bind(purpose)
.bind(set_by)
.bind(community_id.as_uuid())
.bind(channel_id)
.execute(pool)
.await?;
@@ -1051,9 +1128,16 @@ pub async fn set_purpose(
///
/// Returns `AccessDenied` if the channel is already archived.
/// Returns `ChannelNotFound` if the channel does not exist or is deleted.
pub async fn archive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
pub async fn archive_channel(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<()> {
// First check: does the channel exist and what is its state?
let row = sqlx::query("SELECT archived_at FROM channels WHERE id = $1 AND deleted_at IS NULL")
let row = sqlx::query(
"SELECT archived_at FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_optional(pool)
.await?;
@@ -1072,8 +1156,9 @@ pub async fn archive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
sqlx::query(
"UPDATE channels SET archived_at = NOW() \
WHERE id = $1 AND deleted_at IS NULL AND archived_at IS NULL",
WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.execute(pool)
.await?;
@@ -1085,9 +1170,16 @@ pub async fn archive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
///
/// Returns `AccessDenied` if the channel is not currently archived.
/// Returns `ChannelNotFound` if the channel does not exist or is deleted.
pub async fn unarchive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
pub async fn unarchive_channel(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<()> {
// First check: does the channel exist and what is its state?
let row = sqlx::query("SELECT archived_at FROM channels WHERE id = $1 AND deleted_at IS NULL")
let row = sqlx::query(
"SELECT archived_at FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_optional(pool)
.await?;
@@ -1108,8 +1200,9 @@ pub async fn unarchive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
WHEN ttl_seconds IS NOT NULL THEN NOW() + (ttl_seconds || ' seconds')::interval \
ELSE ttl_deadline \
END \
WHERE id = $1 AND deleted_at IS NULL AND archived_at IS NOT NULL",
WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NOT NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.execute(pool)
.await?;
@@ -1121,9 +1214,15 @@ pub async fn unarchive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> {
///
/// Returns `Ok(true)` if the channel was deleted, `Ok(false)` if already
/// deleted or not found.
pub async fn soft_delete_channel(pool: &PgPool, channel_id: Uuid) -> Result<bool> {
let result =
sqlx::query("UPDATE channels SET deleted_at = NOW() WHERE id = $1 AND deleted_at IS NULL")
pub async fn soft_delete_channel(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<bool> {
let result = sqlx::query(
"UPDATE channels SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.execute(pool)
.await?;
@@ -1132,10 +1231,15 @@ pub async fn soft_delete_channel(pool: &PgPool, channel_id: Uuid) -> Result<bool
}
/// Returns the count of active (non-removed) members in a channel.
pub async fn get_member_count(pool: &PgPool, channel_id: Uuid) -> Result<i64> {
pub async fn get_member_count(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<i64> {
let row = sqlx::query(
"SELECT COUNT(*) as cnt FROM channel_members WHERE channel_id = $1 AND removed_at IS NULL",
"SELECT COUNT(*) as cnt FROM channel_members WHERE community_id = $1 AND channel_id = $2 AND removed_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_one(pool)
.await?;
@@ -1148,6 +1252,7 @@ pub async fn get_member_count(pool: &PgPool, channel_id: Uuid) -> Result<i64> {
/// Single query regardless of input size.
pub async fn get_member_counts_bulk(
pool: &PgPool,
community_id: CommunityId,
channel_ids: &[Uuid],
) -> Result<std::collections::HashMap<Uuid, i64>> {
if channel_ids.is_empty() {
@@ -1156,8 +1261,10 @@ pub async fn get_member_counts_bulk(
let mut qb: sqlx::QueryBuilder<sqlx::Postgres> = sqlx::QueryBuilder::new(
"SELECT channel_id, COUNT(*) as cnt FROM channel_members \
WHERE removed_at IS NULL AND channel_id IN (",
WHERE community_id = ",
);
qb.push_bind(community_id.as_uuid());
qb.push(" AND removed_at IS NULL AND channel_id IN (");
let mut sep = qb.separated(", ");
for id in channel_ids {
sep.push_bind(*id);
@@ -1180,14 +1287,16 @@ pub async fn get_member_counts_bulk(
/// Returns `None` if the pubkey is not an active member.
pub async fn get_member_role(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<Option<String>> {
let row = sqlx::query(
"SELECT cm.role::text AS role FROM channel_members cm \
JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL \
WHERE cm.channel_id = $1 AND cm.pubkey = $2 AND cm.removed_at IS NULL",
JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \
WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.pubkey = $3 AND cm.removed_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(pubkey)
.fetch_optional(pool)
@@ -1198,11 +1307,16 @@ pub async fn get_member_role(
/// Bump the TTL deadline for an ephemeral channel after a new message.
///
/// No-op for permanent channels or channels that are already archived/deleted.
pub async fn bump_ttl_deadline(pool: &PgPool, channel_id: Uuid) -> Result<()> {
pub async fn bump_ttl_deadline(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<()> {
sqlx::query(
"UPDATE channels SET ttl_deadline = NOW() + (ttl_seconds || ' seconds')::interval \
WHERE id = $1 AND ttl_seconds IS NOT NULL AND archived_at IS NULL AND deleted_at IS NULL",
WHERE community_id = $1 AND id = $2 AND ttl_seconds IS NOT NULL AND archived_at IS NULL AND deleted_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.execute(pool)
.await?;
@@ -1211,25 +1325,29 @@ pub async fn bump_ttl_deadline(pool: &PgPool, channel_id: Uuid) -> Result<()> {
/// Archive ephemeral channels whose TTL deadline has passed.
///
/// Returns the list of channel IDs that were archived. Idempotent — the
/// Returns the `(community_id, channel_id)` list that was archived. Idempotent — the
/// `archived_at IS NULL` guard prevents double-archiving even if called
/// concurrently from multiple relay pods.
pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result<Vec<Uuid>> {
pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result<Vec<ReapedEphemeralChannel>> {
let rows = sqlx::query(
"UPDATE channels SET archived_at = NOW() \
WHERE ttl_seconds IS NOT NULL \
AND ttl_deadline < NOW() \
AND archived_at IS NULL \
AND deleted_at IS NULL \
RETURNING id",
RETURNING community_id, id",
)
.fetch_all(pool)
.await?;
rows.into_iter()
.map(|row| {
let id: Uuid = row.try_get("id")?;
Ok(id)
let community_id: Uuid = row.try_get("community_id")?;
let channel_id: Uuid = row.try_get("id")?;
Ok(ReapedEphemeralChannel {
community_id: CommunityId::from_uuid(community_id),
channel_id,
})
})
.collect()
}
@@ -1311,7 +1429,78 @@ mod tests {
.await
.expect("insert owner membership");
get_channel(pool, id).await
get_channel(pool, CommunityId::from_uuid(community_id), id).await
}
async fn insert_channel_with_id(
pool: &PgPool,
community_id: Uuid,
id: Uuid,
name: &str,
created_by: &[u8],
) {
sqlx::query(
r#"
INSERT INTO channels
(id, community_id, name, channel_type, visibility, created_by)
VALUES
($1, $2, $3, 'stream', 'open', $4)
"#,
)
.bind(id)
.bind(community_id)
.bind(name)
.bind(created_by)
.execute(pool)
.await
.expect("insert channel with fixed id");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn get_channel_is_scoped_when_channel_uuid_collides_across_communities() {
let pool = setup_pool().await;
let community_a = make_test_community(&pool).await;
let community_b = make_test_community(&pool).await;
let channel_id = Uuid::new_v4();
let creator = random_pubkey();
insert_channel_with_id(
&pool,
community_a,
channel_id,
"community-a-channel",
&creator,
)
.await;
insert_channel_with_id(
&pool,
community_b,
channel_id,
"community-b-channel",
&creator,
)
.await;
let a = get_channel(&pool, CommunityId::from_uuid(community_a), channel_id)
.await
.expect("community A channel should resolve");
let b = get_channel(&pool, CommunityId::from_uuid(community_b), channel_id)
.await
.expect("community B channel should resolve");
assert_eq!(a.name, "community-a-channel");
assert_eq!(b.name, "community-b-channel");
let listed_a = list_channels(&pool, CommunityId::from_uuid(community_a), None)
.await
.expect("list community A channels");
assert!(listed_a
.iter()
.any(|row| row.id == channel_id && row.name == "community-a-channel"));
assert!(!listed_a
.iter()
.any(|row| row.id == channel_id && row.name == "community-b-channel"));
}
/// Agent owner (non-admin) can remove their own bot from a channel.
@@ -1382,7 +1571,7 @@ mod tests {
// Verify the agent is no longer a member
assert!(
!is_member(&pool, channel.id, &agent_pk)
!is_member(&pool, community, channel.id, &agent_pk)
.await
.expect("is_member check"),
"agent should no longer be a member"
@@ -1424,11 +1613,11 @@ mod tests {
.await
.expect("expire and archive channel");
unarchive_channel(&pool, channel.id)
unarchive_channel(&pool, community, channel.id)
.await
.expect("unarchive expired ephemeral channel");
let channel = get_channel(&pool, channel.id)
let channel = get_channel(&pool, community, channel.id)
.await
.expect("reload channel");
assert!(
@@ -1444,7 +1633,9 @@ mod tests {
.await
.expect("run reaper");
assert!(
!reaped.contains(&channel.id),
!reaped
.iter()
.any(|row| row.community_id == community && row.channel_id == channel.id),
"reaper should not immediately rearchive renewed channel"
);
}
crates/buzz-db/src/event.rs
+122 -19
View File
@@ -580,9 +580,15 @@ pub async fn count_events(pool: &PgPool, q: &EventQuery) -> Result<i64> {
/// Returns `Ok(true)` if the event was deleted, `Ok(false)` if already deleted
/// or not found. Callers are responsible for decrementing thread reply counts
/// when the deleted event is a thread reply.
pub async fn soft_delete_event(pool: &PgPool, event_id: &[u8]) -> Result<bool> {
let result =
sqlx::query("UPDATE events SET deleted_at = NOW() WHERE id = $1 AND deleted_at IS NULL")
pub async fn soft_delete_event(
pool: &PgPool,
community_id: CommunityId,
event_id: &[u8],
) -> Result<bool> {
let result = sqlx::query(
"UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(event_id)
.execute(pool)
.await?;
@@ -603,14 +609,16 @@ pub async fn soft_delete_event(pool: &PgPool, event_id: &[u8]) -> Result<bool> {
/// (already deleted, or never existed).
pub async fn soft_delete_by_coordinate(
pool: &PgPool,
community_id: CommunityId,
kind: i32,
pubkey: &[u8],
d_tag: &str,
) -> Result<bool> {
let result = sqlx::query(
"UPDATE events SET deleted_at = NOW() \
WHERE kind = $1 AND pubkey = $2 AND d_tag = $3 AND deleted_at IS NULL",
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(kind)
.bind(pubkey)
.bind(d_tag)
@@ -627,17 +635,20 @@ pub async fn soft_delete_by_coordinate(
/// event was deleted this call.
pub async fn soft_delete_event_and_update_thread(
pool: &PgPool,
community_id: CommunityId,
event_id: &[u8],
parent_event_id: Option<&[u8]>,
root_event_id: Option<&[u8]>,
) -> Result<bool> {
let mut tx = pool.begin().await?;
let result =
sqlx::query("UPDATE events SET deleted_at = NOW() WHERE id = $1 AND deleted_at IS NULL")
.bind(event_id)
.execute(&mut *tx)
.await?;
let result = sqlx::query(
"UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(event_id)
.execute(&mut *tx)
.await?;
let deleted = result.rows_affected() > 0;
@@ -646,8 +657,9 @@ pub async fn soft_delete_event_and_update_thread(
sqlx::query(
"UPDATE thread_metadata \
SET reply_count = GREATEST(reply_count - 1, 0) \
WHERE event_id = $1",
WHERE community_id = $1 AND event_id = $2",
)
.bind(community_id.as_uuid())
.bind(pid)
.execute(&mut *tx)
.await?;
@@ -656,8 +668,9 @@ pub async fn soft_delete_event_and_update_thread(
sqlx::query(
"UPDATE thread_metadata \
SET descendant_count = GREATEST(descendant_count - 1, 0) \
WHERE event_id = $1",
WHERE community_id = $1 AND event_id = $2",
)
.bind(community_id.as_uuid())
.bind(root_id)
.execute(&mut *tx)
.await?;
@@ -672,13 +685,15 @@ pub async fn soft_delete_event_and_update_thread(
/// Returns the `created_at` timestamp of the most recent non-deleted event in a channel.
pub async fn get_last_message_at(
pool: &PgPool,
community_id: CommunityId,
channel_id: uuid::Uuid,
) -> Result<Option<DateTime<Utc>>> {
let row = sqlx::query(
"SELECT created_at FROM events \
WHERE channel_id = $1 AND deleted_at IS NULL \
WHERE community_id = $1 AND channel_id = $2 AND deleted_at IS NULL \
ORDER BY created_at DESC LIMIT 1",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_optional(pool)
.await?;
@@ -695,6 +710,7 @@ pub async fn get_last_message_at(
/// Single query regardless of input size.
pub async fn get_last_message_at_bulk(
pool: &PgPool,
community_id: CommunityId,
channel_ids: &[uuid::Uuid],
) -> Result<std::collections::HashMap<uuid::Uuid, DateTime<Utc>>> {
if channel_ids.is_empty() {
@@ -703,8 +719,10 @@ pub async fn get_last_message_at_bulk(
let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
"SELECT channel_id, MAX(created_at) as last_at FROM events \
WHERE deleted_at IS NULL AND channel_id IN (",
WHERE community_id = ",
);
qb.push_bind(community_id.as_uuid());
qb.push(" AND deleted_at IS NULL AND channel_id IN (");
let mut sep = qb.separated(", ");
for id in channel_ids {
sep.push_bind(*id);
@@ -727,11 +745,16 @@ pub async fn get_last_message_at_bulk(
/// Returns `None` if the event does not exist or has been soft-deleted.
/// Use [`get_event_by_id_including_deleted`] when you need to inspect
/// tombstoned rows (e.g. audit, undelete).
pub async fn get_event_by_id(pool: &PgPool, id_bytes: &[u8]) -> Result<Option<StoredEvent>> {
pub async fn get_event_by_id(
pool: &PgPool,
community_id: CommunityId,
id_bytes: &[u8],
) -> Result<Option<StoredEvent>> {
let row = sqlx::query(
"SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
FROM events WHERE id = $1 AND deleted_at IS NULL ORDER BY created_at DESC LIMIT 1",
FROM events WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL ORDER BY created_at DESC LIMIT 1",
)
.bind(community_id.as_uuid())
.bind(id_bytes)
.fetch_optional(pool)
.await?;
@@ -750,16 +773,18 @@ pub async fn get_event_by_id(pool: &PgPool, id_bytes: &[u8]) -> Result<Option<St
/// duplicate survivors where multiple live rows share the same timestamp.
pub async fn get_latest_global_replaceable(
pool: &PgPool,
community_id: CommunityId,
kind: i32,
pubkey_bytes: &[u8],
) -> Result<Option<StoredEvent>> {
let row = sqlx::query(
"SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
FROM events \
WHERE kind = $1 AND pubkey = $2 AND channel_id IS NULL AND deleted_at IS NULL \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND channel_id IS NULL AND deleted_at IS NULL \
ORDER BY created_at DESC, id ASC \
LIMIT 1",
)
.bind(community_id.as_uuid())
.bind(kind)
.bind(pubkey_bytes)
.fetch_optional(pool)
@@ -778,12 +803,14 @@ pub async fn get_latest_global_replaceable(
/// audit trails, compliance queries).
pub async fn get_event_by_id_including_deleted(
pool: &PgPool,
community_id: CommunityId,
id_bytes: &[u8],
) -> Result<Option<StoredEvent>> {
let row = sqlx::query(
"SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
FROM events WHERE id = $1 ORDER BY created_at DESC LIMIT 1",
FROM events WHERE community_id = $1 AND id = $2 ORDER BY created_at DESC LIMIT 1",
)
.bind(community_id.as_uuid())
.bind(id_bytes)
.fetch_optional(pool)
.await?;
@@ -798,7 +825,11 @@ pub async fn get_event_by_id_including_deleted(
///
/// Returns events in arbitrary order — callers reorder as needed.
/// Uses a single `WHERE id IN (...)` query regardless of input size.
pub async fn get_events_by_ids(pool: &PgPool, ids: &[&[u8]]) -> Result<Vec<StoredEvent>> {
pub async fn get_events_by_ids(
pool: &PgPool,
community_id: CommunityId,
ids: &[&[u8]],
) -> Result<Vec<StoredEvent>> {
if ids.is_empty() {
return Ok(vec![]);
}
@@ -806,8 +837,10 @@ pub async fn get_events_by_ids(pool: &PgPool, ids: &[&[u8]]) -> Result<Vec<Store
let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
"SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
FROM events WHERE deleted_at IS NULL AND id IN (",
FROM events WHERE community_id = ",
);
qb.push_bind(community_id.as_uuid());
qb.push(" AND deleted_at IS NULL AND id IN (");
let mut sep = qb.separated(", ");
for id in ids {
sep.push_bind(id.to_vec());
@@ -1169,6 +1202,76 @@ mod tests {
use super::*;
use nostr::{EventBuilder, Keys, Kind, Tag};
const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz";
async fn setup_pool() -> PgPool {
let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
.or_else(|_| std::env::var("DATABASE_URL"))
.unwrap_or_else(|_| TEST_DB_URL.to_owned());
PgPool::connect(&database_url)
.await
.expect("connect to test DB")
}
async fn make_test_community(pool: &PgPool) -> Uuid {
let id = Uuid::new_v4();
let host = format!("event-test-{}.example", id.simple());
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(id)
.bind(host)
.execute(pool)
.await
.expect("insert test community");
id
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn get_event_by_id_is_scoped_when_event_id_collides_across_communities() {
let pool = setup_pool().await;
let community_a = CommunityId::from_uuid(make_test_community(&pool).await);
let community_b = CommunityId::from_uuid(make_test_community(&pool).await);
let keys = Keys::generate();
let event = EventBuilder::new(Kind::Custom(9), "same signed event")
.sign_with_keys(&keys)
.expect("sign event");
insert_event(&pool, community_a, &event, None)
.await
.expect("insert in community A");
insert_event(&pool, community_b, &event, None)
.await
.expect("insert same event in community B");
sqlx::query("UPDATE events SET content = $1 WHERE community_id = $2 AND id = $3")
.bind("community-a-copy")
.bind(community_a.as_uuid())
.bind(event.id.as_bytes())
.execute(&pool)
.await
.expect("mark community A row");
sqlx::query("UPDATE events SET content = $1 WHERE community_id = $2 AND id = $3")
.bind("community-b-copy")
.bind(community_b.as_uuid())
.bind(event.id.as_bytes())
.execute(&pool)
.await
.expect("mark community B row");
let a = get_event_by_id(&pool, community_a, event.id.as_bytes())
.await
.expect("lookup community A")
.expect("community A row exists");
let b = get_event_by_id(&pool, community_b, event.id.as_bytes())
.await
.expect("lookup community B")
.expect("community B row exists");
assert_eq!(a.event.content, "community-a-copy");
assert_eq!(b.event.content, "community-b-copy");
}
fn make_event_with_kind_and_tags(kind: u16, tags: Vec<Tag>) -> nostr::Event {
let keys = Keys::generate();
EventBuilder::new(Kind::Custom(kind), "test")
crates/buzz-db/src/lib.rs
+171 -54
View File
@@ -345,52 +345,65 @@ impl Db {
/// historical duplicate survivors correctly.
pub async fn get_latest_global_replaceable(
&self,
community_id: CommunityId,
kind: i32,
pubkey_bytes: &[u8],
) -> Result<Option<StoredEvent>> {
event::get_latest_global_replaceable(&self.pool, kind, pubkey_bytes).await
event::get_latest_global_replaceable(&self.pool, community_id, kind, pubkey_bytes).await
}
/// Fetches a single non-deleted event by its raw ID bytes.
///
/// Returns `None` if the event does not exist or has been soft-deleted.
pub async fn get_event_by_id(&self, id_bytes: &[u8]) -> Result<Option<StoredEvent>> {
event::get_event_by_id(&self.pool, id_bytes).await
pub async fn get_event_by_id(
&self,
community_id: CommunityId,
id_bytes: &[u8],
) -> Result<Option<StoredEvent>> {
event::get_event_by_id(&self.pool, community_id, id_bytes).await
}
/// Fetches a single event by its raw ID bytes, **including soft-deleted rows**.
pub async fn get_event_by_id_including_deleted(
&self,
community_id: CommunityId,
id_bytes: &[u8],
) -> Result<Option<StoredEvent>> {
event::get_event_by_id_including_deleted(&self.pool, id_bytes).await
event::get_event_by_id_including_deleted(&self.pool, community_id, id_bytes).await
}
/// Soft-deletes an event. Returns `Ok(true)` if deleted, `Ok(false)` if already deleted.
pub async fn soft_delete_event(&self, event_id: &[u8]) -> Result<bool> {
event::soft_delete_event(&self.pool, event_id).await
pub async fn soft_delete_event(
&self,
community_id: CommunityId,
event_id: &[u8],
) -> Result<bool> {
event::soft_delete_event(&self.pool, community_id, event_id).await
}
/// Soft-delete the live row for an addressable coordinate `(kind, pubkey, d_tag)`.
/// Used by NIP-09 a-tag deletion for parameterized-replaceable kinds.
pub async fn soft_delete_by_coordinate(
&self,
community_id: CommunityId,
kind: i32,
pubkey: &[u8],
d_tag: &str,
) -> Result<bool> {
event::soft_delete_by_coordinate(&self.pool, kind, pubkey, d_tag).await
event::soft_delete_by_coordinate(&self.pool, community_id, kind, pubkey, d_tag).await
}
/// Atomically soft-delete an event and decrement thread reply counters.
pub async fn soft_delete_event_and_update_thread(
&self,
community_id: CommunityId,
event_id: &[u8],
parent_event_id: Option<&[u8]>,
root_event_id: Option<&[u8]>,
) -> Result<bool> {
event::soft_delete_event_and_update_thread(
&self.pool,
community_id,
event_id,
parent_event_id,
root_event_id,
@@ -399,21 +412,30 @@ impl Db {
}
/// Returns the most recent `created_at` for a channel.
pub async fn get_last_message_at(&self, channel_id: Uuid) -> Result<Option<DateTime<Utc>>> {
event::get_last_message_at(&self.pool, channel_id).await
pub async fn get_last_message_at(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<Option<DateTime<Utc>>> {
event::get_last_message_at(&self.pool, community_id, channel_id).await
}
/// Bulk-fetch the most recent `created_at` for a set of channel IDs.
pub async fn get_last_message_at_bulk(
&self,
community_id: CommunityId,
channel_ids: &[Uuid],
) -> Result<std::collections::HashMap<Uuid, DateTime<Utc>>> {
event::get_last_message_at_bulk(&self.pool, channel_ids).await
event::get_last_message_at_bulk(&self.pool, community_id, channel_ids).await
}
/// Batch-fetch non-deleted events by their raw IDs.
pub async fn get_events_by_ids(&self, ids: &[&[u8]]) -> Result<Vec<StoredEvent>> {
event::get_events_by_ids(&self.pool, ids).await
pub async fn get_events_by_ids(
&self,
community_id: CommunityId,
ids: &[&[u8]],
) -> Result<Vec<StoredEvent>> {
event::get_events_by_ids(&self.pool, community_id, ids).await
}
/// Atomically insert an event AND its thread metadata in a single transaction.
@@ -494,18 +516,31 @@ impl Db {
}
/// Fetches a channel record by ID.
pub async fn get_channel(&self, channel_id: Uuid) -> Result<channel::ChannelRecord> {
channel::get_channel(&self.pool, channel_id).await
pub async fn get_channel(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<channel::ChannelRecord> {
channel::get_channel(&self.pool, community_id, channel_id).await
}
/// Returns the canvas content for a channel, if any.
pub async fn get_canvas(&self, channel_id: Uuid) -> Result<Option<String>> {
channel::get_canvas(&self.pool, channel_id).await
pub async fn get_canvas(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<Option<String>> {
channel::get_canvas(&self.pool, community_id, channel_id).await
}
/// Sets or clears the canvas content for a channel.
pub async fn set_canvas(&self, channel_id: Uuid, canvas: Option<&str>) -> Result<()> {
channel::set_canvas(&self.pool, channel_id, canvas).await
pub async fn set_canvas(
&self,
community_id: CommunityId,
channel_id: Uuid,
canvas: Option<&str>,
) -> Result<()> {
channel::set_canvas(&self.pool, community_id, channel_id, canvas).await
}
/// Adds a member to a channel.
@@ -540,49 +575,75 @@ impl Db {
}
/// Returns `true` if the pubkey is an active member.
pub async fn is_member(&self, channel_id: Uuid, pubkey: &[u8]) -> Result<bool> {
channel::is_member(&self.pool, channel_id, pubkey).await
pub async fn is_member(
&self,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<bool> {
channel::is_member(&self.pool, community_id, channel_id, pubkey).await
}
/// Returns all active members of a channel.
pub async fn get_members(&self, channel_id: Uuid) -> Result<Vec<channel::MemberRecord>> {
channel::get_members(&self.pool, channel_id).await
pub async fn get_members(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<Vec<channel::MemberRecord>> {
channel::get_members(&self.pool, community_id, channel_id).await
}
/// Returns active members for multiple channels in a single query.
pub async fn get_members_bulk(
&self,
community_id: CommunityId,
channel_ids: &[Uuid],
) -> Result<Vec<channel::MemberRecord>> {
channel::get_members_bulk(&self.pool, channel_ids).await
channel::get_members_bulk(&self.pool, community_id, channel_ids).await
}
/// Get all channel IDs accessible to a pubkey.
pub async fn get_accessible_channel_ids(&self, pubkey: &[u8]) -> Result<Vec<Uuid>> {
channel::get_accessible_channel_ids(&self.pool, pubkey).await
pub async fn get_accessible_channel_ids(
&self,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Vec<Uuid>> {
channel::get_accessible_channel_ids(&self.pool, community_id, pubkey).await
}
/// Lists channels, optionally filtered by visibility.
pub async fn list_channels(
&self,
community_id: CommunityId,
visibility: Option<&str>,
) -> Result<Vec<channel::ChannelRecord>> {
channel::list_channels(&self.pool, visibility).await
channel::list_channels(&self.pool, community_id, visibility).await
}
/// Returns full channel records for all channels a user can access.
pub async fn get_accessible_channels(
&self,
community_id: CommunityId,
pubkey: &[u8],
visibility_filter: Option<&str>,
member_only: Option<bool>,
) -> Result<Vec<channel::AccessibleChannel>> {
channel::get_accessible_channels(&self.pool, pubkey, visibility_filter, member_only).await
channel::get_accessible_channels(
&self.pool,
community_id,
pubkey,
visibility_filter,
member_only,
)
.await
}
/// Returns all bot-role members with their aggregated channel names.
pub async fn get_bot_members(&self) -> Result<Vec<channel::BotMemberRecord>> {
channel::get_bot_members(&self.pool).await
/// Returns all bot-role members with their aggregated channel names in one community.
pub async fn get_bot_members(
&self,
community_id: CommunityId,
) -> Result<Vec<channel::BotMemberRecord>> {
channel::get_bot_members(&self.pool, community_id).await
}
/// Bulk-fetch user records by pubkey.
@@ -593,62 +654,99 @@ impl Db {
/// Updates a channel's name and/or description.
pub async fn update_channel(
&self,
community_id: CommunityId,
channel_id: Uuid,
updates: channel::ChannelUpdate,
) -> Result<channel::ChannelRecord> {
channel::update_channel(&self.pool, channel_id, updates).await
channel::update_channel(&self.pool, community_id, channel_id, updates).await
}
/// Sets the topic for a channel.
pub async fn set_topic(&self, channel_id: Uuid, topic: &str, set_by: &[u8]) -> Result<()> {
channel::set_topic(&self.pool, channel_id, topic, set_by).await
pub async fn set_topic(
&self,
community_id: CommunityId,
channel_id: Uuid,
topic: &str,
set_by: &[u8],
) -> Result<()> {
channel::set_topic(&self.pool, community_id, channel_id, topic, set_by).await
}
/// Sets the purpose for a channel.
pub async fn set_purpose(&self, channel_id: Uuid, purpose: &str, set_by: &[u8]) -> Result<()> {
channel::set_purpose(&self.pool, channel_id, purpose, set_by).await
pub async fn set_purpose(
&self,
community_id: CommunityId,
channel_id: Uuid,
purpose: &str,
set_by: &[u8],
) -> Result<()> {
channel::set_purpose(&self.pool, community_id, channel_id, purpose, set_by).await
}
/// Archives a channel.
pub async fn archive_channel(&self, channel_id: Uuid) -> Result<()> {
channel::archive_channel(&self.pool, channel_id).await
pub async fn archive_channel(&self, community_id: CommunityId, channel_id: Uuid) -> Result<()> {
channel::archive_channel(&self.pool, community_id, channel_id).await
}
/// Unarchives a channel.
pub async fn unarchive_channel(&self, channel_id: Uuid) -> Result<()> {
channel::unarchive_channel(&self.pool, channel_id).await
pub async fn unarchive_channel(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<()> {
channel::unarchive_channel(&self.pool, community_id, channel_id).await
}
/// Soft-delete a channel.
pub async fn soft_delete_channel(&self, channel_id: Uuid) -> Result<bool> {
channel::soft_delete_channel(&self.pool, channel_id).await
pub async fn soft_delete_channel(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<bool> {
channel::soft_delete_channel(&self.pool, community_id, channel_id).await
}
/// Returns the count of active members in a channel.
pub async fn get_member_count(&self, channel_id: Uuid) -> Result<i64> {
channel::get_member_count(&self.pool, channel_id).await
pub async fn get_member_count(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<i64> {
channel::get_member_count(&self.pool, community_id, channel_id).await
}
/// Bulk-fetch member counts for a set of channel IDs.
pub async fn get_member_counts_bulk(
&self,
community_id: CommunityId,
channel_ids: &[Uuid],
) -> Result<std::collections::HashMap<Uuid, i64>> {
channel::get_member_counts_bulk(&self.pool, channel_ids).await
channel::get_member_counts_bulk(&self.pool, community_id, channel_ids).await
}
/// Get the active role of a pubkey in a channel.
pub async fn get_member_role(&self, channel_id: Uuid, pubkey: &[u8]) -> Result<Option<String>> {
channel::get_member_role(&self.pool, channel_id, pubkey).await
pub async fn get_member_role(
&self,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<Option<String>> {
channel::get_member_role(&self.pool, community_id, channel_id, pubkey).await
}
/// Bump the TTL deadline for an ephemeral channel after a new message.
pub async fn bump_ttl_deadline(&self, channel_id: Uuid) -> Result<()> {
channel::bump_ttl_deadline(&self.pool, channel_id).await
pub async fn bump_ttl_deadline(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<()> {
channel::bump_ttl_deadline(&self.pool, community_id, channel_id).await
}
/// Archive ephemeral channels whose TTL deadline has passed.
pub async fn reap_expired_ephemeral_channels(&self) -> Result<Vec<Uuid>> {
pub async fn reap_expired_ephemeral_channels(
&self,
) -> Result<Vec<channel::ReapedEphemeralChannel>> {
channel::reap_expired_ephemeral_channels(&self.pool).await
}
@@ -845,6 +943,7 @@ impl Db {
#[allow(clippy::too_many_arguments)]
pub async fn insert_thread_metadata(
&self,
community_id: CommunityId,
event_id: &[u8],
event_created_at: DateTime<Utc>,
channel_id: Uuid,
@@ -857,6 +956,7 @@ impl Db {
) -> Result<()> {
thread::insert_thread_metadata(
&self.pool,
community_id,
event_id,
event_created_at,
channel_id,
@@ -873,25 +973,36 @@ impl Db {
/// Fetch replies under a root event.
pub async fn get_thread_replies(
&self,
community_id: CommunityId,
root_event_id: &[u8],
depth_limit: Option<u32>,
limit: u32,
cursor: Option<&[u8]>,
) -> Result<Vec<thread::ThreadReply>> {
thread::get_thread_replies(&self.pool, root_event_id, depth_limit, limit, cursor).await
thread::get_thread_replies(
&self.pool,
community_id,
root_event_id,
depth_limit,
limit,
cursor,
)
.await
}
/// Fetch aggregated thread stats.
pub async fn get_thread_summary(
&self,
community_id: CommunityId,
event_id: &[u8],
) -> Result<Option<thread::ThreadSummary>> {
thread::get_thread_summary(&self.pool, event_id).await
thread::get_thread_summary(&self.pool, community_id, event_id).await
}
/// Top-level messages for a channel.
pub async fn get_channel_messages_top_level(
&self,
community_id: CommunityId,
channel_id: Uuid,
limit: u32,
before_cursor: Option<DateTime<Utc>>,
@@ -900,6 +1011,7 @@ impl Db {
) -> Result<Vec<thread::TopLevelMessage>> {
thread::get_channel_messages_top_level(
&self.pool,
community_id,
channel_id,
limit,
before_cursor,
@@ -912,18 +1024,21 @@ impl Db {
/// Look up a single thread_metadata row by event_id.
pub async fn get_thread_metadata_by_event(
&self,
community_id: CommunityId,
event_id: &[u8],
) -> Result<Option<thread::ThreadMetadataRecord>> {
thread::get_thread_metadata_by_event(&self.pool, event_id).await
thread::get_thread_metadata_by_event(&self.pool, community_id, event_id).await
}
/// Decrement reply counts.
pub async fn decrement_reply_count(
&self,
community_id: CommunityId,
parent_event_id: &[u8],
root_event_id: Option<&[u8]>,
) -> Result<()> {
thread::decrement_reply_count(&self.pool, parent_event_id, root_event_id).await
thread::decrement_reply_count(&self.pool, community_id, parent_event_id, root_event_id)
.await
}
/// Add (or re-activate) a reaction.
@@ -1673,13 +1788,15 @@ impl Db {
/// Soft-delete NIP-29 discovery events for a channel created by a specific relay pubkey.
pub async fn soft_delete_discovery_events(
&self,
community_id: CommunityId,
channel_id: Uuid,
relay_pubkey: &[u8],
) -> Result<u64> {
let result = sqlx::query(
"UPDATE events SET deleted_at = NOW() \
WHERE channel_id = $1 AND pubkey = $2 AND deleted_at IS NULL AND kind IN (39000, 39001, 39002)",
WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 AND deleted_at IS NULL AND kind IN (39000, 39001, 39002)",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(relay_pubkey)
.execute(&self.pool)
crates/buzz-db/src/thread.rs
+154 -26
View File
@@ -9,6 +9,8 @@ use chrono::{DateTime, Utc};
use sqlx::{PgPool, Row};
use uuid::Uuid;
use buzz_core::CommunityId;
use crate::{error::Result, event::row_to_stored_event};
// -- Structs ------------------------------------------------------------------
@@ -110,6 +112,7 @@ pub struct ThreadMetadataRecord {
#[allow(clippy::too_many_arguments)]
pub async fn insert_thread_metadata(
pool: &PgPool,
community_id: CommunityId,
event_id: &[u8],
event_created_at: DateTime<Utc>,
channel_id: Uuid,
@@ -125,14 +128,15 @@ pub async fn insert_thread_metadata(
let result = sqlx::query(
r#"
INSERT INTO thread_metadata
(event_created_at, event_id, channel_id,
(community_id, event_created_at, event_id, channel_id,
parent_event_id, parent_event_created_at,
root_event_id, root_event_created_at,
depth, broadcast)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
ON CONFLICT DO NOTHING
"#,
)
.bind(community_id.as_uuid())
.bind(event_created_at)
.bind(event_id)
.bind(channel_id)
@@ -156,14 +160,15 @@ pub async fn insert_thread_metadata(
sqlx::query(
r#"
INSERT INTO thread_metadata
(event_created_at, event_id, channel_id,
(community_id, event_created_at, event_id, channel_id,
parent_event_id, parent_event_created_at,
root_event_id, root_event_created_at,
depth, broadcast)
VALUES ($1, $2, $3, NULL, NULL, NULL, NULL, 0, false)
VALUES ($1, $2, $3, $4, NULL, NULL, NULL, NULL, 0, false)
ON CONFLICT DO NOTHING
"#,
)
.bind(community_id.as_uuid())
.bind(parent_ts)
.bind(pid)
.bind(channel_id)
@@ -185,6 +190,7 @@ pub async fn insert_thread_metadata(
ON CONFLICT DO NOTHING
"#,
)
.bind(community_id.as_uuid())
.bind(root_ts)
.bind(root_id)
.bind(channel_id)
@@ -199,9 +205,10 @@ pub async fn insert_thread_metadata(
UPDATE thread_metadata
SET reply_count = reply_count + 1,
last_reply_at = NOW()
WHERE event_id = $1
WHERE community_id = $1 AND event_id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(pid)
.execute(&mut *tx)
.await?;
@@ -212,9 +219,10 @@ pub async fn insert_thread_metadata(
r#"
UPDATE thread_metadata
SET descendant_count = descendant_count + 1
WHERE event_id = $1
WHERE community_id = $1 AND event_id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(root_id)
.execute(&mut *tx)
.await?;
@@ -239,6 +247,7 @@ pub async fn insert_thread_metadata(
#[allow(dead_code)]
pub async fn increment_reply_count(
pool: &PgPool,
community_id: CommunityId,
parent_event_id: &[u8],
root_event_id: Option<&[u8]>,
) -> Result<()> {
@@ -248,9 +257,10 @@ pub async fn increment_reply_count(
UPDATE thread_metadata
SET reply_count = reply_count + 1,
last_reply_at = NOW()
WHERE event_id = $1
WHERE community_id = $1 AND event_id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(parent_event_id)
.execute(pool)
.await?;
@@ -261,9 +271,10 @@ pub async fn increment_reply_count(
r#"
UPDATE thread_metadata
SET descendant_count = descendant_count + 1
WHERE event_id = $1
WHERE community_id = $1 AND event_id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(root_id)
.execute(pool)
.await?;
@@ -277,6 +288,7 @@ pub async fn increment_reply_count(
/// root -- even when root == parent. Mirrors the increment logic exactly.
pub async fn decrement_reply_count(
pool: &PgPool,
community_id: CommunityId,
parent_event_id: &[u8],
root_event_id: Option<&[u8]>,
) -> Result<()> {
@@ -285,9 +297,10 @@ pub async fn decrement_reply_count(
r#"
UPDATE thread_metadata
SET reply_count = GREATEST(reply_count - 1, 0)
WHERE event_id = $1
WHERE community_id = $1 AND event_id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(parent_event_id)
.execute(pool)
.await?;
@@ -298,9 +311,10 @@ pub async fn decrement_reply_count(
r#"
UPDATE thread_metadata
SET descendant_count = GREATEST(descendant_count - 1, 0)
WHERE event_id = $1
WHERE community_id = $1 AND event_id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(root_id)
.execute(pool)
.await?;
@@ -320,6 +334,7 @@ pub async fn decrement_reply_count(
/// - `limit` -- maximum rows returned (caller should cap this).
pub async fn get_thread_replies(
pool: &PgPool,
community_id: CommunityId,
root_event_id: &[u8],
depth_limit: Option<u32>,
limit: u32,
@@ -336,7 +351,7 @@ pub async fn get_thread_replies(
// Build the query dynamically based on optional filters.
// Track the next positional parameter index.
let mut param_idx = 2u32; // $1 is root_event_id
let mut param_idx = 3u32; // $1 is community_id, $2 is root_event_id
let mut sql = String::from(
r#"
SELECT
@@ -357,9 +372,11 @@ pub async fn get_thread_replies(
tm.broadcast
FROM thread_metadata tm
JOIN events e
ON e.created_at = tm.event_created_at
ON e.community_id = tm.community_id
AND e.created_at = tm.event_created_at
AND e.id = tm.event_id
WHERE tm.root_event_id = $1
WHERE tm.community_id = $1
AND tm.root_event_id = $2
AND e.deleted_at IS NULL
"#,
);
@@ -377,7 +394,9 @@ pub async fn get_thread_replies(
" ORDER BY tm.event_created_at ASC LIMIT ${param_idx}"
));
let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(root_event_id);
let mut q = sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(community_id.as_uuid())
.bind(root_event_id);
if let Some(dl) = depth_limit {
q = q.bind(dl as i32);
@@ -428,15 +447,20 @@ pub async fn get_thread_replies(
}
/// Fetch aggregated thread stats for a single event, plus up to 10 participant pubkeys.
pub async fn get_thread_summary(pool: &PgPool, event_id: &[u8]) -> Result<Option<ThreadSummary>> {
pub async fn get_thread_summary(
pool: &PgPool,
community_id: CommunityId,
event_id: &[u8],
) -> Result<Option<ThreadSummary>> {
let row = sqlx::query(
r#"
SELECT reply_count, descendant_count, last_reply_at
FROM thread_metadata
WHERE event_id = $1
WHERE community_id = $1 AND event_id = $2
LIMIT 1
"#,
)
.bind(community_id.as_uuid())
.bind(event_id)
.fetch_optional(pool)
.await?;
@@ -457,9 +481,11 @@ pub async fn get_thread_summary(pool: &PgPool, event_id: &[u8]) -> Result<Option
SELECT DISTINCT e.pubkey, MAX(e.created_at) AS last_seen
FROM thread_metadata tm
JOIN events e
ON e.created_at = tm.event_created_at
ON e.community_id = tm.community_id
AND e.created_at = tm.event_created_at
AND e.id = tm.event_id
WHERE tm.root_event_id = $1
WHERE tm.community_id = $1
AND tm.root_event_id = $2
AND e.deleted_at IS NULL
GROUP BY e.pubkey
) sub
@@ -467,6 +493,7 @@ pub async fn get_thread_summary(pool: &PgPool, event_id: &[u8]) -> Result<Option
LIMIT 10
"#,
)
.bind(community_id.as_uuid())
.bind(event_id)
.fetch_all(pool)
.await?;
@@ -500,13 +527,14 @@ pub async fn get_thread_summary(pool: &PgPool, event_id: &[u8]) -> Result<Option
/// polling (returns only messages created after the given timestamp).
pub async fn get_channel_messages_top_level(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
limit: u32,
before_cursor: Option<DateTime<Utc>>,
since_cursor: Option<DateTime<Utc>>,
kind_filter: Option<&[u32]>,
) -> Result<Vec<TopLevelMessage>> {
let mut param_idx = 2u32; // $1 is channel_id
let mut param_idx = 3u32; // $1 is community_id, $2 is channel_id
let mut sql = String::from(
r#"
SELECT
@@ -519,9 +547,11 @@ pub async fn get_channel_messages_top_level(
e.channel_id
FROM events e
LEFT JOIN thread_metadata tm
ON tm.event_created_at = e.created_at
ON tm.community_id = e.community_id
AND tm.event_created_at = e.created_at
AND tm.event_id = e.id
WHERE e.channel_id = $1
WHERE e.community_id = $1
AND e.channel_id = $2
AND e.deleted_at IS NULL
AND (
tm.depth IS NULL
@@ -561,7 +591,9 @@ pub async fn get_channel_messages_top_level(
" ORDER BY e.created_at {order} LIMIT ${param_idx}"
));
let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(channel_id);
let mut q = sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(community_id.as_uuid())
.bind(channel_id);
if let Some(cursor) = before_cursor {
q = q.bind(cursor);
@@ -604,6 +636,7 @@ pub async fn get_channel_messages_top_level(
/// can be decremented.
pub async fn get_thread_metadata_by_event(
pool: &PgPool,
community_id: CommunityId,
event_id: &[u8],
) -> Result<Option<ThreadMetadataRecord>> {
let row = sqlx::query(
@@ -619,10 +652,11 @@ pub async fn get_thread_metadata_by_event(
descendant_count,
broadcast
FROM thread_metadata
WHERE event_id = $1
WHERE community_id = $1 AND event_id = $2
LIMIT 1
"#,
)
.bind(community_id.as_uuid())
.bind(event_id)
.fetch_optional(pool)
.await?;
@@ -746,11 +780,105 @@ mod tests {
.await
.expect("insert owner membership");
crate::channel::get_channel(pool, id)
crate::channel::get_channel(pool, buzz_core::CommunityId::from_uuid(community_id), id)
.await
.map(|channel| (channel, buzz_core::CommunityId::from_uuid(community_id)))
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn get_thread_metadata_by_event_is_scoped_when_event_id_collides_across_communities() {
let pool = setup_pool().await;
let author = Keys::generate();
let channel_id = Uuid::new_v4();
let community_a = make_test_community(&pool).await;
let community_b = make_test_community(&pool).await;
let community_a = buzz_core::CommunityId::from_uuid(community_a);
let community_b = buzz_core::CommunityId::from_uuid(community_b);
crate::channel::create_channel_with_id(
&pool,
community_a,
channel_id,
&format!("thread-collision-a-{channel_id}"),
ChannelType::Stream,
ChannelVisibility::Open,
None,
author.public_key().to_bytes().as_slice(),
None,
)
.await
.expect("create community A channel");
crate::channel::create_channel_with_id(
&pool,
community_b,
channel_id,
&format!("thread-collision-b-{channel_id}"),
ChannelType::Stream,
ChannelVisibility::Open,
None,
author.public_key().to_bytes().as_slice(),
None,
)
.await
.expect("create community B channel");
let event = make_stream_event(&author, "same id in both communities");
let created_at = event_created_at(&event);
insert_event_with_thread_metadata(
&pool,
community_a,
&event,
Some(channel_id),
Some(ThreadMetadataParams {
event_id: event.id.as_bytes(),
event_created_at: created_at,
channel_id,
parent_event_id: None,
parent_event_created_at: None,
root_event_id: None,
root_event_created_at: None,
depth: 0,
broadcast: true,
}),
)
.await
.expect("insert community A metadata");
insert_event_with_thread_metadata(
&pool,
community_b,
&event,
Some(channel_id),
Some(ThreadMetadataParams {
event_id: event.id.as_bytes(),
event_created_at: created_at,
channel_id,
parent_event_id: None,
parent_event_created_at: None,
root_event_id: None,
root_event_created_at: None,
depth: 3,
broadcast: false,
}),
)
.await
.expect("insert community B metadata");
let a = get_thread_metadata_by_event(&pool, community_a, event.id.as_bytes())
.await
.expect("lookup community A metadata")
.expect("community A metadata exists");
let b = get_thread_metadata_by_event(&pool, community_b, event.id.as_bytes())
.await
.expect("lookup community B metadata")
.expect("community B metadata exists");
assert_eq!(a.depth, 0);
assert!(a.broadcast);
assert_eq!(b.depth, 3);
assert!(!b.broadcast);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn get_thread_replies_reconstructs_stored_events() {
@@ -797,7 +925,7 @@ mod tests {
.await
.expect("insert reply event and metadata");
let replies = get_thread_replies(&pool, root.id.as_bytes(), Some(10), 10, None)
let replies = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 10, None)
.await
.expect("fetch thread replies");
@@ -893,7 +1021,7 @@ mod tests {
.rows_affected();
assert_eq!(rows_changed, 1, "expected to corrupt exactly one row");
let replies = get_thread_replies(&pool, root.id.as_bytes(), Some(10), 10, None)
let replies = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 10, None)
.await
.expect("fetch thread replies must succeed despite a corrupt row");