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): scope relay rows by community

Browse Source 
Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
This commit is contained in:
npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf
2026-06-26 17:08:38 -04:00
co-authored by Tyler Longwell
parent e349d76498
commit 4766d6bc62
8 changed files with 1745 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-db/src/channel.rs
+184 -55
View File
@@ -9,6 +9,7 @@ use sqlx::{PgPool, Postgres, Row, Transaction};
use uuid::Uuid;
use crate::error::{DbError, Result};
use buzz_core::CommunityId;
// Re-export the canonical enum definitions from buzz-core.
// These live in core (zero I/O deps) so the SDK can share them
@@ -84,6 +85,7 @@ pub struct MemberRecord {
/// Creates a new channel, bootstraps the creator as owner, and returns the record.
pub async fn create_channel(
pool: &PgPool,
community_id: CommunityId,
name: &str,
channel_type: ChannelType,
visibility: ChannelVisibility,
@@ -104,12 +106,13 @@ pub async fn create_channel(
sqlx::query(
r#"
INSERT INTO channels (id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline)
VALUES ($1, $2, $3::channel_type, $4::channel_visibility, $5, $6, $7,
CASE WHEN $7 IS NOT NULL THEN NOW() + ($7 || ' seconds')::interval ELSE NULL END)
INSERT INTO channels (id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline)
VALUES ($1, $2, $3, $4::channel_type, $5::channel_visibility, $6, $7, $8,
CASE WHEN $8 IS NOT NULL THEN NOW() + ($8 || ' seconds')::interval ELSE NULL END)
"#,
)
.bind(id)
.bind(community_id.as_uuid())
.bind(name)
.bind(channel_type.as_str())
.bind(visibility.as_str())
@@ -121,14 +124,15 @@ pub async fn create_channel(
sqlx::query(
r#"
INSERT INTO channel_members (channel_id, pubkey, role, invited_by)
VALUES ($1, $2, 'owner', $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, 'owner', $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(created_by)
.bind(created_by)
@@ -144,9 +148,10 @@ pub async fn create_channel(
topic, topic_set_by, topic_set_at,
purpose, purpose_set_by, purpose_set_at,
ttl_seconds, ttl_deadline
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?;
@@ -163,6 +168,7 @@ pub async fn create_channel(
#[allow(clippy::too_many_arguments)]
pub async fn create_channel_with_id(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
name: &str,
channel_type: ChannelType,
@@ -188,13 +194,14 @@ pub async fn create_channel_with_id(
let rows_affected = sqlx::query(
r#"
INSERT INTO channels (id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline)
VALUES ($1, $2, $3::channel_type, $4::channel_visibility, $5, $6, $7,
CASE WHEN $7 IS NOT NULL THEN NOW() + ($7 || ' seconds')::interval ELSE NULL END)
ON CONFLICT (id) DO NOTHING
INSERT INTO channels (id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline)
VALUES ($1, $2, $3, $4::channel_type, $5::channel_visibility, $6, $7, $8,
CASE WHEN $8 IS NOT NULL THEN NOW() + ($8 || ' seconds')::interval ELSE NULL END)
ON CONFLICT (community_id, id) DO NOTHING
"#,
)
.bind(channel_id)
.bind(community_id.as_uuid())
.bind(name)
.bind(channel_type.as_str())
.bind(visibility.as_str())
@@ -211,14 +218,15 @@ pub async fn create_channel_with_id(
// Bootstrap the creator as owner.
sqlx::query(
r#"
INSERT INTO channel_members (channel_id, pubkey, role, invited_by)
VALUES ($1, $2, 'owner', $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, 'owner', $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(channel_id)
.bind(created_by)
.bind(created_by)
@@ -235,9 +243,10 @@ pub async fn create_channel_with_id(
topic, topic_set_by, topic_set_at,
purpose, purpose_set_by, purpose_set_at,
ttl_seconds, ttl_deadline
FROM channels WHERE id = $1
FROM channels WHERE community_id = $1 AND id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_one(&mut *tx)
.await?;
@@ -305,6 +314,7 @@ pub async fn set_canvas(pool: &PgPool, channel_id: Uuid, canvas: Option<&str>) -
/// races (e.g. the inviter being removed between the role check and the INSERT).
pub async fn add_member(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
role: MemberRole,
@@ -330,7 +340,7 @@ pub async fn add_member(
let is_creator_bootstrap = inviter == pubkey && inviter == channel.created_by.as_slice();
if !is_creator_bootstrap {
let inviter_role_str = get_active_role_tx(&mut tx, channel_id, inviter)
let inviter_role_str = get_active_role_tx(&mut tx, community_id, channel_id, inviter)
.await?
.ok_or_else(|| {
DbError::AccessDenied("inviter is not an active member".to_string())
@@ -354,7 +364,7 @@ pub async fn add_member(
// elevated roles. Self-join always gets Member.
if role.is_elevated() {
let granter_role = match invited_by {
Some(inv) => get_active_role_tx(&mut tx, channel_id, inv).await?,
Some(inv) => get_active_role_tx(&mut tx, community_id, channel_id, inv).await?,
None => None,
};
match granter_role.as_deref() {
@@ -372,14 +382,15 @@ pub async fn add_member(
sqlx::query(
r#"
INSERT INTO channel_members (channel_id, pubkey, role, invited_by)
VALUES ($1, $2, $3::member_role, $4)
ON CONFLICT (channel_id, pubkey) DO UPDATE SET
INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by)
VALUES ($1, $2, $3, $4::member_role, $5)
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(channel_id)
.bind(pubkey)
.bind(effective_role.as_str())
@@ -390,9 +401,10 @@ pub async fn add_member(
let row = sqlx::query(
r#"
SELECT channel_id, pubkey, role::text AS role, joined_at, invited_by, removed_at
FROM channel_members WHERE channel_id = $1 AND pubkey = $2
FROM channel_members WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3
"#,
)
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(pubkey)
.fetch_one(&mut *tx)
@@ -415,6 +427,7 @@ pub async fn add_member(
/// because `agent_owner_pubkey` is immutable (set once at token mint).
pub async fn remove_member(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
actor_pubkey: &[u8],
@@ -423,7 +436,7 @@ pub async fn remove_member(
let is_self_remove = pubkey == actor_pubkey;
if !is_self_remove {
let actor_role_str = get_active_role_tx(&mut tx, channel_id, actor_pubkey)
let actor_role_str = get_active_role_tx(&mut tx, community_id, channel_id, actor_pubkey)
.await?
.ok_or_else(|| DbError::AccessDenied("actor is not an active member".to_string()))?;
let actor_role: MemberRole = actor_role_str.parse().map_err(|_| {
@@ -432,7 +445,7 @@ pub async fn remove_member(
// Safe to query outside the transaction: agent_owner_pubkey is immutable
// (set once at token mint, first-mint-wins).
if !actor_role.is_elevated()
&& !crate::user::is_agent_owner(pool, pubkey, actor_pubkey).await?
&& !crate::user::is_agent_owner(pool, community_id, pubkey, actor_pubkey).await?
{
return Err(DbError::AccessDenied(
"only owners/admins or the agent's owner may remove other members".to_string(),
@@ -443,12 +456,13 @@ pub async fn remove_member(
// Defense-in-depth: prevent removing the last owner regardless of caller.
// Callers (REST handlers, NIP-29 handlers) also check this, but the DB
// layer enforces it as the final safety net.
let target_role = get_active_role_tx(&mut tx, channel_id, pubkey).await?;
let target_role = get_active_role_tx(&mut tx, community_id, channel_id, pubkey).await?;
if target_role.as_deref() == Some("owner") {
let row = sqlx::query(
"SELECT COUNT(*) as cnt FROM channel_members \
WHERE channel_id = $1 AND role = 'owner' AND removed_at IS NULL",
WHERE community_id = $1 AND channel_id = $2 AND role = 'owner' AND removed_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_one(&mut *tx)
.await?;
@@ -464,10 +478,11 @@ pub async fn remove_member(
r#"
UPDATE channel_members
SET removed_at = NOW(), removed_by = $1
WHERE channel_id = $2 AND pubkey = $3 AND removed_at IS NULL
WHERE community_id = $2 AND channel_id = $3 AND pubkey = $4 AND removed_at IS NULL
"#,
)
.bind(actor_pubkey)
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(pubkey)
.execute(&mut *tx)
@@ -619,13 +634,15 @@ pub async fn list_channels(pool: &PgPool, visibility: Option<&str>) -> Result<Ve
/// Transaction-aware variant of [`get_active_role_tx`].
async fn get_active_role_tx(
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<Option<String>> {
let row = sqlx::query(
"SELECT role::text AS role FROM channel_members \
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)
.fetch_optional(&mut **tx)
@@ -1235,28 +1252,97 @@ mod tests {
Keys::generate().public_key().to_bytes().to_vec()
}
async fn make_test_community(pool: &PgPool) -> Uuid {
let id = Uuid::new_v4();
let host = format!("channel-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
}
async fn create_test_channel(
pool: &PgPool,
community_id: Uuid,
name: &str,
channel_type: ChannelType,
visibility: ChannelVisibility,
description: Option<&str>,
created_by: &[u8],
ttl_seconds: Option<i32>,
) -> Result<ChannelRecord> {
let id = Uuid::new_v4();
sqlx::query(
r#"
INSERT INTO channels
(id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline)
VALUES
($1, $2, $3, $4::channel_type, $5::channel_visibility, $6, $7, $8,
CASE WHEN $8 IS NOT NULL THEN NOW() + ($8 || ' seconds')::interval ELSE NULL END)
"#,
)
.bind(id)
.bind(community_id)
.bind(name)
.bind(channel_type.as_str())
.bind(visibility.as_str())
.bind(description)
.bind(created_by)
.bind(ttl_seconds)
.execute(pool)
.await
.expect("insert test channel");
sqlx::query(
r#"
INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by)
VALUES ($1, $2, $3, 'owner', $4)
"#,
)
.bind(community_id)
.bind(id)
.bind(created_by)
.bind(created_by)
.execute(pool)
.await
.expect("insert owner membership");
get_channel(pool, id).await
}
/// Agent owner (non-admin) can remove their own bot from a channel.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn test_agent_owner_can_remove_bot() {
let pool = setup_pool().await;
let community_id = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_id);
let owner_pk = random_pubkey();
let agent_pk = random_pubkey();
// Create users and set agent ownership
ensure_user(&pool, &owner_pk).await.expect("ensure owner");
ensure_user(&pool, &agent_pk).await.expect("ensure agent");
set_agent_owner(&pool, &agent_pk, &owner_pk)
ensure_user(&pool, community, &owner_pk)
.await
.expect("ensure owner");
ensure_user(&pool, community, &agent_pk)
.await
.expect("ensure agent");
set_agent_owner(&pool, community, &agent_pk, &owner_pk)
.await
.expect("set agent owner");
// Create a channel owned by someone else entirely
let channel_owner_pk = random_pubkey();
ensure_user(&pool, &channel_owner_pk)
ensure_user(&pool, community, &channel_owner_pk)
.await
.expect("ensure channel owner");
let channel = create_channel(
let channel = create_test_channel(
&pool,
community_id,
"test-bot-remove",
ChannelType::Stream,
ChannelVisibility::Open,
@@ -1268,15 +1354,29 @@ mod tests {
.expect("create channel");
// Add owner and agent as regular members
add_member(&pool, channel.id, &owner_pk, MemberRole::Member, None)
.await
.expect("add owner as member");
add_member(&pool, channel.id, &agent_pk, MemberRole::Member, None)
.await
.expect("add agent as member");
add_member(
&pool,
community,
channel.id,
&owner_pk,
MemberRole::Member,
None,
)
.await
.expect("add owner as member");
add_member(
&pool,
community,
channel.id,
&agent_pk,
MemberRole::Member,
None,
)
.await
.expect("add agent as member");
// Owner should be able to remove their agent
remove_member(&pool, channel.id, &agent_pk, &owner_pk)
remove_member(&pool, community, channel.id, &agent_pk, &owner_pk)
.await
.expect("agent owner should be able to remove their bot");
@@ -1295,11 +1395,16 @@ mod tests {
#[ignore = "requires Postgres"]
async fn test_unarchive_expired_ephemeral_channel_renews_ttl_deadline() {
let pool = setup_pool().await;
let community_id = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_id);
let owner_pk = random_pubkey();
ensure_user(&pool, &owner_pk).await.expect("ensure owner");
ensure_user(&pool, community, &owner_pk)
.await
.expect("ensure owner");
let channel = create_channel(
let channel = create_test_channel(
&pool,
community_id,
"test-unarchive-renews-ttl",
ChannelType::Stream,
ChannelVisibility::Open,
@@ -1311,8 +1416,9 @@ mod tests {
.expect("create ephemeral channel");
sqlx::query(
"UPDATE channels SET archived_at = NOW(), ttl_deadline = NOW() - interval '1 second' WHERE id = $1",
"UPDATE channels SET archived_at = NOW(), ttl_deadline = NOW() - interval '1 second' WHERE community_id = $1 AND id = $2",
)
.bind(community_id)
.bind(channel.id)
.execute(&pool)
.await
@@ -1348,25 +1454,34 @@ mod tests {
#[ignore = "requires Postgres"]
async fn test_random_user_cannot_remove_bot() {
let pool = setup_pool().await;
let community_id = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_id);
let owner_pk = random_pubkey();
let agent_pk = random_pubkey();
let random_pk = random_pubkey();
// Create users and set agent ownership
ensure_user(&pool, &owner_pk).await.expect("ensure owner");
ensure_user(&pool, &agent_pk).await.expect("ensure agent");
ensure_user(&pool, &random_pk).await.expect("ensure random");
set_agent_owner(&pool, &agent_pk, &owner_pk)
ensure_user(&pool, community, &owner_pk)
.await
.expect("ensure owner");
ensure_user(&pool, community, &agent_pk)
.await
.expect("ensure agent");
ensure_user(&pool, community, &random_pk)
.await
.expect("ensure random");
set_agent_owner(&pool, community, &agent_pk, &owner_pk)
.await
.expect("set agent owner");
// Create a channel
let channel_owner_pk = random_pubkey();
ensure_user(&pool, &channel_owner_pk)
ensure_user(&pool, community, &channel_owner_pk)
.await
.expect("ensure channel owner");
let channel = create_channel(
let channel = create_test_channel(
&pool,
community_id,
"test-bot-no-remove",
ChannelType::Stream,
ChannelVisibility::Open,
@@ -1378,15 +1493,29 @@ mod tests {
.expect("create channel");
// Add random user and agent as regular members
add_member(&pool, channel.id, &random_pk, MemberRole::Member, None)
.await
.expect("add random as member");
add_member(&pool, channel.id, &agent_pk, MemberRole::Member, None)
.await
.expect("add agent as member");
add_member(
&pool,
community,
channel.id,
&random_pk,
MemberRole::Member,
None,
)
.await
.expect("add random as member");
add_member(
&pool,
community,
channel.id,
&agent_pk,
MemberRole::Member,
None,
)
.await
.expect("add agent as member");
// Random user should NOT be able to remove the agent
let result = remove_member(&pool, channel.id, &agent_pk, &random_pk).await;
let result = remove_member(&pool, community, channel.id, &agent_pk, &random_pk).await;
assert!(
result.is_err(),
"random user should not be able to remove someone else's bot"
crates/buzz-db/src/event.rs
+88 -24
View File
@@ -12,13 +12,15 @@ use uuid::Uuid;
use buzz_core::kind::{
event_kind_i32, is_ephemeral, is_parameterized_replaceable, KIND_AUTH, KIND_EVENT_REMINDER,
};
use buzz_core::StoredEvent;
use buzz_core::{CommunityId, StoredEvent};
use crate::error::{DbError, Result};
/// Optional filters for [`query_events`].
#[derive(Debug, Default, Clone)]
#[derive(Debug, Clone)]
pub struct EventQuery {
/// Server-resolved community scope.
pub community_id: CommunityId,
/// Restrict results to this channel.
pub channel_id: Option<Uuid>,
/// Restrict results to these kind values (stored as `i32` in Postgres).
@@ -123,6 +125,7 @@ pub fn extract_not_before(event: &Event) -> Option<i64> {
/// Returns `(StoredEvent, was_inserted)` — `was_inserted` is `false` on duplicate.
pub async fn insert_event(
pool: &PgPool,
community_id: CommunityId,
event: &Event,
channel_id: Option<Uuid>,
) -> Result<(StoredEvent, bool)> {
@@ -150,11 +153,12 @@ pub async fn insert_event(
let not_before = extract_not_before(event);
let result = sqlx::query(
r#"
INSERT INTO events (id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
ON CONFLICT DO NOTHING
"#,
)
.bind(community_id.as_uuid())
.bind(id_bytes.as_slice())
.bind(pubkey_bytes.as_slice())
.bind(created_at)
@@ -220,16 +224,22 @@ pub async fn query_events(pool: &PgPool, q: &EventQuery) -> Result<Vec<StoredEve
"SELECT e.id, e.pubkey, e.created_at, e.kind, e.tags, e.content, \
e.sig, e.received_at, e.channel_id \
FROM events e \
INNER JOIN event_mentions m ON e.id = m.event_id \
WHERE e.deleted_at IS NULL AND m.pubkey_hex = ",
INNER JOIN event_mentions m \
ON e.community_id = m.community_id AND e.id = m.event_id \
WHERE e.community_id = ",
);
b.push_bind(q.community_id.as_uuid());
b.push(" AND e.deleted_at IS NULL AND m.pubkey_hex = ");
b.push_bind(p_hex.to_ascii_lowercase());
b
} else {
QueryBuilder::new(
let mut b = QueryBuilder::new(
"SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \
FROM events WHERE deleted_at IS NULL",
)
FROM events WHERE community_id = ",
);
b.push_bind(q.community_id.as_uuid());
b.push(" AND deleted_at IS NULL");
b
};
// Use unqualified column names when no join, qualified when joined.
@@ -444,13 +454,19 @@ pub async fn count_events(pool: &PgPool, q: &EventQuery) -> Result<i64> {
let mut qb: QueryBuilder<sqlx::Postgres> = if let Some(ref p_hex) = q.p_tag_hex {
let mut b = QueryBuilder::new(
"SELECT COUNT(*) as cnt FROM events e \
INNER JOIN event_mentions m ON e.id = m.event_id \
WHERE e.deleted_at IS NULL AND m.pubkey_hex = ",
INNER JOIN event_mentions m \
ON e.community_id = m.community_id AND e.id = m.event_id \
WHERE e.community_id = ",
);
b.push_bind(q.community_id.as_uuid());
b.push(" AND e.deleted_at IS NULL AND m.pubkey_hex = ");
b.push_bind(p_hex.to_ascii_lowercase());
b
} else {
QueryBuilder::new("SELECT COUNT(*) as cnt FROM events WHERE deleted_at IS NULL")
let mut b = QueryBuilder::new("SELECT COUNT(*) as cnt FROM events WHERE community_id = ");
b.push_bind(q.community_id.as_uuid());
b.push(" AND deleted_at IS NULL");
b
};
let col_prefix = if q.p_tag_hex.is_some() { "e." } else { "" };
@@ -842,6 +858,7 @@ pub struct ThreadMetadataParams<'a> {
/// Returns `(StoredEvent, was_inserted)`.
pub async fn insert_event_with_thread_metadata(
pool: &PgPool,
community_id: CommunityId,
event: &Event,
channel_id: Option<Uuid>,
thread_meta: Option<ThreadMetadataParams<'_>>,
@@ -871,11 +888,12 @@ pub async fn insert_event_with_thread_metadata(
let result = sqlx::query(
r#"
INSERT INTO events (id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
ON CONFLICT DO NOTHING
"#,
)
.bind(community_id.as_uuid())
.bind(id_bytes.as_slice())
.bind(pubkey_bytes.as_slice())
.bind(created_at)
@@ -899,14 +917,15 @@ pub async fn insert_event_with_thread_metadata(
let tm_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(meta.event_created_at)
.bind(meta.event_id)
.bind(meta.channel_id)
@@ -931,14 +950,15 @@ pub async fn insert_event_with_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(meta.channel_id)
@@ -953,14 +973,15 @@ pub async fn insert_event_with_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(root_ts)
.bind(root_id)
.bind(meta.channel_id)
@@ -973,9 +994,10 @@ pub async fn insert_event_with_thread_metadata(
r#"
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?;
@@ -985,9 +1007,10 @@ pub async fn insert_event_with_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?;
@@ -1083,7 +1106,19 @@ pub async fn claim_due_reminder(
event_id: &[u8],
event_created_at: DateTime<Utc>,
) -> Result<bool> {
let now_epoch = Utc::now().timestamp();
claim_due_reminder_with_stamp(pool, event_id, event_created_at, Utc::now().timestamp()).await
}
/// Atomically claim a due reminder using a caller-supplied delivery stamp.
///
/// The same stamp should be passed to [`release_due_reminder`] if the publish
/// side effect fails, so rollback can compare-and-clear only this pod's claim.
pub async fn claim_due_reminder_with_stamp(
pool: &PgPool,
event_id: &[u8],
event_created_at: DateTime<Utc>,
delivery_stamp: i64,
) -> Result<bool> {
let result = sqlx::query(
r#"
UPDATE events
@@ -1091,7 +1126,7 @@ pub async fn claim_due_reminder(
WHERE created_at = $2 AND id = $3 AND delivered_at IS NULL
"#,
)
.bind(now_epoch)
.bind(delivery_stamp)
.bind(event_created_at)
.bind(event_id)
.execute(pool)
@@ -1100,6 +1135,35 @@ pub async fn claim_due_reminder(
Ok(result.rows_affected() > 0)
}
/// Release a previously claimed reminder when publish fails.
///
/// The `delivery_stamp` must be the exact value written by the claiming pod;
/// that compare-and-clear prevents one pod from rolling back another pod's
/// later claim after a retry/race.
pub async fn release_due_reminder(
pool: &PgPool,
event_id: &[u8],
event_created_at: DateTime<Utc>,
delivery_stamp: i64,
) -> Result<bool> {
let result = sqlx::query(
r#"
UPDATE events
SET delivered_at = NULL
WHERE created_at = $1
AND id = $2
AND delivered_at = $3
"#,
)
.bind(event_created_at)
.bind(event_id)
.bind(delivery_stamp)
.execute(pool)
.await?;
Ok(result.rows_affected() == 1)
}
#[cfg(test)]
mod tests {
use super::*;
crates/buzz-db/src/lib.rs
+261 -36
View File
@@ -47,7 +47,7 @@ use sqlx::{PgPool, QueryBuilder, Row};
use std::time::Duration;
use uuid::Uuid;
use buzz_core::StoredEvent;
use buzz_core::{CommunityId, StoredEvent};
/// Extract p-tag mentions from an event and insert into the `event_mentions` table.
///
@@ -55,6 +55,7 @@ use buzz_core::StoredEvent;
/// Uses `INSERT ... ON CONFLICT DO NOTHING` so duplicate inserts are silently skipped.
pub async fn insert_mentions(
pool: &PgPool,
community_id: CommunityId,
event: &nostr::Event,
channel_id: Option<Uuid>,
) -> Result<()> {
@@ -106,11 +107,12 @@ pub async fn insert_mentions(
// Single multi-row INSERT ... ON CONFLICT DO NOTHING — one round-trip regardless of mention count.
let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
"INSERT INTO event_mentions \
(pubkey_hex, event_id, event_created_at, channel_id, event_kind) ",
(community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) ",
);
qb.push_values(&valid_pubkeys, |mut b, pubkey| {
b.push_bind(pubkey.as_str())
b.push_bind(community_id.as_uuid())
.push_bind(pubkey.as_str())
.push_bind(event_id_bytes.as_slice())
.push_bind(created_at)
.push_bind(channel_id)
@@ -162,6 +164,15 @@ impl Default for DbConfig {
}
}
/// Community host-map row returned by [`Db::lookup_community_by_host`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommunityRecord {
/// Stable server-resolved community id.
pub id: CommunityId,
/// Normalized host that maps to this community.
pub host: String,
}
/// Token summary returned by [`Db::list_active_tokens`].
#[derive(Debug, Clone)]
pub struct TokenSummary {
@@ -216,15 +227,101 @@ impl Db {
self.pool.begin().await.map_err(Into::into)
}
/// Returns the community mapped to a normalized request host, if one exists.
///
/// The caller owns host normalization and turns `None` into the fail-closed
/// request/connection error. buzz-db only reads the durable host map.
pub async fn lookup_community_by_host(
&self,
normalized_host: &str,
) -> Result<Option<CommunityRecord>> {
let row = sqlx::query(
r#"
SELECT id, host
FROM communities
WHERE host = $1
"#,
)
.bind(normalized_host)
.fetch_optional(&self.pool)
.await?;
row.map(|row| {
let id: Uuid = row.try_get("id")?;
let host: String = row.try_get("host")?;
Ok(CommunityRecord {
id: CommunityId::from_uuid(id),
host,
})
})
.transpose()
}
/// Ensure a configured community host exists and return its row.
///
/// This is the startup/config seeding path for N=1 deployments. Migrations
/// create the schema only; deployment-specific hosts are not hardcoded into
/// schema history.
pub async fn ensure_configured_community(
&self,
normalized_host: &str,
) -> Result<CommunityRecord> {
let row = sqlx::query(
r#"
INSERT INTO communities (host)
VALUES ($1)
ON CONFLICT (host) DO UPDATE SET host = EXCLUDED.host
RETURNING id, host
"#,
)
.bind(normalized_host)
.fetch_one(&self.pool)
.await?;
let id: Uuid = row.try_get("id")?;
let host: String = row.try_get("host")?;
Ok(CommunityRecord {
id: CommunityId::from_uuid(id),
host,
})
}
/// Returns the community that owns a channel, if the channel exists.
///
/// Internal relay producers use this to derive tenant context from the row
/// they are acting on, rather than falling back to an implicit default.
pub async fn community_of_channel(&self, channel_id: Uuid) -> Result<Option<CommunityId>> {
let row = sqlx::query(
r#"
SELECT community_id
FROM channels
WHERE id = $1
AND deleted_at IS NULL
"#,
)
.bind(channel_id)
.fetch_optional(&self.pool)
.await?;
row.map(|row| {
let id: Uuid = row.try_get("community_id")?;
Ok(CommunityId::from_uuid(id))
})
.transpose()
}
/// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate.
pub async fn insert_event(
&self,
community_id: CommunityId,
event: &nostr::Event,
channel_id: Option<Uuid>,
) -> Result<(StoredEvent, bool)> {
let result = event::insert_event(&self.pool, event, channel_id).await?;
let result = event::insert_event(&self.pool, community_id, event, channel_id).await?;
if result.1 {
if let Err(e) = insert_mentions(&self.pool, event, channel_id).await {
if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await {
tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
}
}
@@ -322,15 +419,21 @@ impl Db {
/// Atomically insert an event AND its thread metadata in a single transaction.
pub async fn insert_event_with_thread_metadata(
&self,
community_id: CommunityId,
event: &nostr::Event,
channel_id: Option<Uuid>,
thread_meta: Option<event::ThreadMetadataParams<'_>>,
) -> Result<(StoredEvent, bool)> {
let result =
event::insert_event_with_thread_metadata(&self.pool, event, channel_id, thread_meta)
.await?;
let result = event::insert_event_with_thread_metadata(
&self.pool,
community_id,
event,
channel_id,
thread_meta,
)
.await?;
if result.1 {
if let Err(e) = insert_mentions(&self.pool, event, channel_id).await {
if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await {
tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
}
}
@@ -340,6 +443,7 @@ impl Db {
/// Creates a new channel, bootstraps the creator as owner, and returns the record.
pub async fn create_channel(
&self,
community_id: CommunityId,
name: &str,
channel_type: channel::ChannelType,
visibility: channel::ChannelVisibility,
@@ -349,6 +453,7 @@ impl Db {
) -> Result<channel::ChannelRecord> {
channel::create_channel(
&self.pool,
community_id,
name,
channel_type,
visibility,
@@ -365,6 +470,7 @@ impl Db {
#[allow(clippy::too_many_arguments)]
pub async fn create_channel_with_id(
&self,
community_id: CommunityId,
channel_id: Uuid,
name: &str,
channel_type: channel::ChannelType,
@@ -375,6 +481,7 @@ impl Db {
) -> Result<(channel::ChannelRecord, bool)> {
channel::create_channel_with_id(
&self.pool,
community_id,
channel_id,
name,
channel_type,
@@ -404,22 +511,32 @@ impl Db {
/// Adds a member to a channel.
pub async fn add_member(
&self,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
role: channel::MemberRole,
invited_by: Option<&[u8]>,
) -> Result<channel::MemberRecord> {
channel::add_member(&self.pool, channel_id, pubkey, role, invited_by).await
channel::add_member(
&self.pool,
community_id,
channel_id,
pubkey,
role,
invited_by,
)
.await
}
/// Removes a member from a channel.
pub async fn remove_member(
&self,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
actor_pubkey: &[u8],
) -> Result<()> {
channel::remove_member(&self.pool, channel_id, pubkey, actor_pubkey).await
channel::remove_member(&self.pool, community_id, channel_id, pubkey, actor_pubkey).await
}
/// Returns `true` if the pubkey is an active member.
@@ -553,19 +670,45 @@ impl Db {
event::claim_due_reminder(&self.pool, event_id, event_created_at).await
}
/// Atomically claim a due reminder using a caller-supplied delivery stamp.
pub async fn claim_due_reminder_with_stamp(
&self,
event_id: &[u8],
event_created_at: chrono::DateTime<chrono::Utc>,
delivery_stamp: i64,
) -> Result<bool> {
event::claim_due_reminder_with_stamp(&self.pool, event_id, event_created_at, delivery_stamp)
.await
}
/// Release a claimed due reminder after a publish failure.
pub async fn release_due_reminder(
&self,
event_id: &[u8],
event_created_at: chrono::DateTime<chrono::Utc>,
delivery_stamp: i64,
) -> Result<bool> {
event::release_due_reminder(&self.pool, event_id, event_created_at, delivery_stamp).await
}
/// Ensure a user record exists (upsert).
pub async fn ensure_user(&self, pubkey: &[u8]) -> Result<()> {
user::ensure_user(&self.pool, pubkey).await
pub async fn ensure_user(&self, community_id: CommunityId, pubkey: &[u8]) -> Result<()> {
user::ensure_user(&self.pool, community_id, pubkey).await
}
/// Get a single user record by pubkey.
pub async fn get_user(&self, pubkey: &[u8]) -> Result<Option<user::UserProfile>> {
user::get_user(&self.pool, pubkey).await
pub async fn get_user(
&self,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Option<user::UserProfile>> {
user::get_user(&self.pool, community_id, pubkey).await
}
/// Update a user's profile fields.
pub async fn update_user_profile(
&self,
community_id: CommunityId,
pubkey: &[u8],
display_name: Option<&str>,
avatar_url: Option<&str>,
@@ -574,6 +717,7 @@ impl Db {
) -> Result<()> {
user::update_user_profile(
&self.pool,
community_id,
pubkey,
display_name,
avatar_url,
@@ -586,43 +730,61 @@ impl Db {
/// Look up a user by NIP-05 handle.
pub async fn get_user_by_nip05(
&self,
community_id: CommunityId,
local_part: &str,
domain: &str,
) -> Result<Option<user::UserProfile>> {
user::get_user_by_nip05(&self.pool, local_part, domain).await
user::get_user_by_nip05(&self.pool, community_id, local_part, domain).await
}
/// Search users by display name, NIP-05 handle, or pubkey prefix.
pub async fn search_users(
&self,
community_id: CommunityId,
query: &str,
limit: u32,
) -> Result<Vec<user::UserSearchProfile>> {
user::search_users(&self.pool, query, limit).await
user::search_users(&self.pool, community_id, query, limit).await
}
/// Atomically set agent owner — only if no owner is currently assigned.
/// Returns Ok(true) if set, Ok(false) if an owner already exists.
pub async fn set_agent_owner(&self, agent_pubkey: &[u8], owner_pubkey: &[u8]) -> Result<bool> {
user::set_agent_owner(&self.pool, agent_pubkey, owner_pubkey).await
pub async fn set_agent_owner(
&self,
community_id: CommunityId,
agent_pubkey: &[u8],
owner_pubkey: &[u8],
) -> Result<bool> {
user::set_agent_owner(&self.pool, community_id, agent_pubkey, owner_pubkey).await
}
/// Get the channel_add_policy and agent_owner_pubkey for a user.
pub async fn get_agent_channel_policy(
&self,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Option<(String, Option<Vec<u8>>)>> {
user::get_agent_channel_policy(&self.pool, pubkey).await
user::get_agent_channel_policy(&self.pool, community_id, pubkey).await
}
/// Check whether `actor_pubkey` is the agent owner of `target_pubkey`.
pub async fn is_agent_owner(&self, target_pubkey: &[u8], actor_pubkey: &[u8]) -> Result<bool> {
user::is_agent_owner(&self.pool, target_pubkey, actor_pubkey).await
pub async fn is_agent_owner(
&self,
community_id: CommunityId,
target_pubkey: &[u8],
actor_pubkey: &[u8],
) -> Result<bool> {
user::is_agent_owner(&self.pool, community_id, target_pubkey, actor_pubkey).await
}
/// Set the channel_add_policy for a user.
pub async fn set_channel_add_policy(&self, pubkey: &[u8], policy: &str) -> Result<()> {
user::set_channel_add_policy(&self.pool, pubkey, policy).await
pub async fn set_channel_add_policy(
&self,
community_id: CommunityId,
pubkey: &[u8],
policy: &str,
) -> Result<()> {
user::set_channel_add_policy(&self.pool, community_id, pubkey, policy).await
}
/// Find an existing DM by its participant hash.
@@ -1088,6 +1250,7 @@ impl Db {
/// Create a new workflow.
pub async fn create_workflow(
&self,
community_id: CommunityId,
channel_id: Option<Uuid>,
owner_pubkey: &[u8],
name: &str,
@@ -1096,6 +1259,7 @@ impl Db {
) -> Result<Uuid> {
workflow::create_workflow(
&self.pool,
community_id,
channel_id,
owner_pubkey,
name,
@@ -1133,6 +1297,51 @@ impl Db {
workflow::list_all_enabled_workflows(&self.pool).await
}
/// Claim a scheduled workflow fire for an authoritative schedule instant.
///
/// Returns `Some` only for the first pod to claim `(workflow_id,
/// scheduled_for)`; all other pods must skip creating a run. The claim SQL
/// resolves `community_id` from the workflow row; callers never supply it.
pub async fn claim_scheduled_workflow_fire(
&self,
workflow_id: Uuid,
scheduled_for: chrono::DateTime<chrono::Utc>,
) -> Result<Option<workflow::ScheduledWorkflowFireClaim>> {
workflow::claim_scheduled_workflow_fire(&self.pool, workflow_id, scheduled_for).await
}
/// Fetch the latest claimed schedule instant for interval trigger anchoring.
pub async fn latest_scheduled_workflow_fire(
&self,
workflow_id: Uuid,
) -> Result<Option<chrono::DateTime<chrono::Utc>>> {
workflow::latest_scheduled_workflow_fire(&self.pool, workflow_id).await
}
/// Attach the workflow run id created from a won scheduled-fire claim.
pub async fn attach_scheduled_workflow_run(
&self,
workflow_id: Uuid,
scheduled_for: chrono::DateTime<chrono::Utc>,
workflow_run_id: Uuid,
) -> Result<bool> {
workflow::attach_scheduled_workflow_run(
&self.pool,
workflow_id,
scheduled_for,
workflow_run_id,
)
.await
}
/// Delete old scheduled workflow fire claims before a retention cutoff.
pub async fn prune_scheduled_workflow_fires_before(
&self,
older_than: chrono::DateTime<chrono::Utc>,
) -> Result<u64> {
workflow::prune_scheduled_workflow_fires_before(&self.pool, older_than).await
}
/// Update a workflow's name, definition, and hash.
pub async fn update_workflow(
&self,
@@ -1487,6 +1696,7 @@ impl Db {
/// skip fan-out/dispatch when `was_inserted` is false.
pub async fn replace_addressable_event(
&self,
community_id: CommunityId,
event: &nostr::Event,
channel_id: Option<Uuid>,
) -> Result<(StoredEvent, bool)> {
@@ -1501,6 +1711,10 @@ impl Db {
// Collisions cause extra serialization, not incorrect behavior.
let lock_key = {
let mut h: u64 = 0xcbf29ce484222325; // FNV offset basis
for b in community_id.as_uuid().as_bytes() {
h ^= *b as u64;
h = h.wrapping_mul(0x100000001b3);
}
for b in kind_i32.to_le_bytes() {
h ^= b as u64;
h = h.wrapping_mul(0x100000001b3); // FNV prime
@@ -1531,11 +1745,12 @@ impl Db {
// historical data where prior bugs may have left multiple live rows.
let existing: Option<(chrono::DateTime<chrono::Utc>, Vec<u8>)> = sqlx::query_as(
"SELECT created_at, id FROM events \
WHERE kind = $1 AND pubkey = $2 \
AND channel_id IS NOT DISTINCT FROM $3 \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \
AND channel_id IS NOT DISTINCT FROM $4 \
AND deleted_at IS NULL \
ORDER BY created_at DESC, id ASC LIMIT 1",
)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(channel_id)
@@ -1562,10 +1777,11 @@ impl Db {
// Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety.
sqlx::query(
"UPDATE events SET deleted_at = NOW() \
WHERE kind = $1 AND pubkey = $2 \
AND channel_id IS NOT DISTINCT FROM $3 \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \
AND channel_id IS NOT DISTINCT FROM $4 \
AND deleted_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(channel_id)
@@ -1579,10 +1795,11 @@ impl Db {
let d_tag = crate::event::extract_d_tag(event);
let insert_result = sqlx::query(
"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(community_id.as_uuid())
.bind(event.id.as_bytes().as_slice())
.bind(pubkey_bytes.as_slice())
.bind(created_at)
@@ -1611,7 +1828,7 @@ impl Db {
// Mentions are a denormalized index — safe outside the transaction.
// insert_event() normally handles this, but we inlined the INSERT above.
if let Err(e) = crate::insert_mentions(&self.pool, event, channel_id).await {
if let Err(e) = crate::insert_mentions(&self.pool, community_id, event, channel_id).await {
tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
}
@@ -1640,6 +1857,7 @@ impl Db {
/// this function instead, where the author's pubkey + d-tag is the natural key.
pub async fn replace_parameterized_event(
&self,
community_id: CommunityId,
event: &nostr::Event,
d_tag: &str,
channel_id: Option<Uuid>,
@@ -1654,6 +1872,10 @@ impl Db {
// Same algorithm as replace_addressable_event — deterministic across processes.
let lock_key = {
let mut h: u64 = 0xcbf29ce484222325; // FNV offset basis
for b in community_id.as_uuid().as_bytes() {
h ^= *b as u64;
h = h.wrapping_mul(0x100000001b3);
}
for b in kind_i32.to_le_bytes() {
h ^= b as u64;
h = h.wrapping_mul(0x100000001b3);
@@ -1679,9 +1901,10 @@ impl Db {
// Check for existing event with same (kind, pubkey, d_tag).
let existing: Option<(chrono::DateTime<chrono::Utc>, Vec<u8>)> = sqlx::query_as(
"SELECT created_at, id FROM events \
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 \
ORDER BY created_at DESC, id ASC LIMIT 1",
)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
@@ -1705,8 +1928,9 @@ impl Db {
// Soft-delete the older event(s).
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_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
@@ -1720,10 +1944,11 @@ impl Db {
let received_at = chrono::Utc::now();
let insert_result = sqlx::query(
"INSERT INTO events (id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \
"INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \
ON CONFLICT DO NOTHING",
)
.bind(community_id.as_uuid())
.bind(event.id.as_bytes().as_slice())
.bind(pubkey_bytes.as_slice())
.bind(created_at)
@@ -1750,7 +1975,7 @@ impl Db {
tx.commit().await?;
// Mentions are a denormalized index — safe outside the transaction.
if let Err(e) = crate::insert_mentions(&self.pool, event, channel_id).await {
if let Err(e) = crate::insert_mentions(&self.pool, community_id, event, channel_id).await {
tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
}
crates/buzz-db/src/migration.rs
+616 -205
View File
@@ -1,9 +1,8 @@
//! Embedded SQLx migrations for Buzz.
//!
//! Fresh deployments apply the checked-in SQL files under `migrations/`.
//! Existing pre-SQLx deployments are baselined when core Buzz tables already
//! exist but `_sqlx_migrations` does not, so startup will not try to replay the
//! initial schema over a live database.
//! Fresh deployments apply the checked-in SQL files under `migrations/`. The
//! multi-tenant rewrite owns a clean consolidated `0001`; legacy single-tenant
//! cutover/backfill is a separate operator script, not startup migration state.
use sqlx::PgPool;
@@ -11,154 +10,619 @@ use crate::Result;
static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("../../migrations");
#[cfg(test)]
static SCHEMA_SQL: &str = include_str!("../../../schema/schema.sql");
const BASELINE_MIGRATION_VERSIONS: &[i64] = &[1, 2];
/// Run all pending Buzz database migrations.
pub async fn run_migrations(pool: &PgPool) -> Result<()> {
baseline_existing_database(pool).await?;
MIGRATOR.run(pool).await?;
Ok(())
}
async fn baseline_existing_database(pool: &PgPool) -> Result<()> {
if migrations_table_exists(pool).await? || !pre_sqlx_schema_exists(pool).await? {
return Ok(());
}
ensure_migrations_table(pool).await?;
for version in BASELINE_MIGRATION_VERSIONS {
let migration = MIGRATOR
.iter()
.find(|migration| migration.version == *version)
.expect("baseline migration version must exist in embedded migrator");
sqlx::query(
r#"
INSERT INTO _sqlx_migrations
(version, description, success, checksum, execution_time)
VALUES ($1, $2, TRUE, $3, 0)
ON CONFLICT (version) DO NOTHING
"#,
)
.bind(migration.version)
.bind(&*migration.description)
.bind(&*migration.checksum)
.execute(pool)
.await?;
}
tracing::info!(
versions = ?BASELINE_MIGRATION_VERSIONS,
"Baselined existing Buzz database for SQLx migrations"
);
Ok(())
}
async fn migrations_table_exists(pool: &PgPool) -> Result<bool> {
let exists = sqlx::query_scalar::<_, bool>(
r#"
SELECT EXISTS (
SELECT 1
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_name = '_sqlx_migrations'
)
"#,
)
.fetch_one(pool)
.await?;
Ok(exists)
}
async fn pre_sqlx_schema_exists(pool: &PgPool) -> Result<bool> {
let exists = sqlx::query_scalar::<_, bool>(
r#"
SELECT EXISTS (
SELECT 1
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_name = 'events'
) AND EXISTS (
SELECT 1
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_name = 'channels'
)
"#,
)
.fetch_one(pool)
.await?;
Ok(exists)
}
async fn ensure_migrations_table(pool: &PgPool) -> Result<()> {
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS _sqlx_migrations (
version BIGINT PRIMARY KEY,
description TEXT NOT NULL,
installed_on TIMESTAMPTZ NOT NULL DEFAULT now(),
success BOOLEAN NOT NULL,
checksum BYTEA NOT NULL,
execution_time BIGINT NOT NULL
)
"#,
)
.execute(pool)
.await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use sqlx::PgPool;
use std::collections::BTreeSet;
const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ConstraintKind {
ForeignKey,
PrimaryKey,
Unique,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct ConstraintLint {
table: String,
kind: ConstraintKind,
description: String,
columns: Vec<String>,
}
fn migration_sql() -> &'static str {
MIGRATOR
.iter()
.find(|migration| migration.version == 1)
.expect("initial migration must exist")
.sql
.as_str()
}
fn strip_sql_comments(sql: &str) -> String {
sql.lines()
.map(|line| line.split_once("--").map_or(line, |(before, _)| before))
.collect::<Vec<_>>()
.join("\n")
}
fn normalize_sql(sql: &str) -> String {
strip_sql_comments(sql)
.split_whitespace()
.collect::<Vec<_>>()
.join(" ")
.to_ascii_lowercase()
}
fn split_sql_statements(sql: &str) -> Vec<String> {
let sql = strip_sql_comments(sql);
let bytes = sql.as_bytes();
let mut statements = Vec::new();
let mut start = 0usize;
let mut idx = 0usize;
let mut in_single_quote = false;
let mut in_dollar_quote = false;
while idx < bytes.len() {
match bytes[idx] {
b'\'' if !in_dollar_quote => {
in_single_quote = !in_single_quote;
idx += 1;
}
b'$' if !in_single_quote && idx + 1 < bytes.len() && bytes[idx + 1] == b'$' => {
in_dollar_quote = !in_dollar_quote;
idx += 2;
}
b';' if !in_single_quote && !in_dollar_quote => {
let statement = sql[start..idx].trim();
if !statement.is_empty() {
statements.push(statement.to_owned());
}
start = idx + 1;
idx += 1;
}
_ => idx += 1,
}
}
let tail = sql[start..].trim();
if !tail.is_empty() {
statements.push(tail.to_owned());
}
statements
}
fn find_matching_paren(sql: &str, open: usize) -> Option<usize> {
let mut depth = 0usize;
for (offset, byte) in sql.as_bytes()[open..].iter().enumerate() {
match byte {
b'(' => depth += 1,
b')' => {
depth = depth.checked_sub(1)?;
if depth == 0 {
return Some(open + offset);
}
}
_ => {}
}
}
None
}
fn split_top_level_csv(input: &str) -> Vec<String> {
let mut parts = Vec::new();
let mut start = 0usize;
let mut depth = 0usize;
for (idx, byte) in input.bytes().enumerate() {
match byte {
b'(' => depth += 1,
b')' => depth = depth.saturating_sub(1),
b',' if depth == 0 => {
parts.push(input[start..idx].trim().to_owned());
start = idx + 1;
}
_ => {}
}
}
let tail = input[start..].trim();
if !tail.is_empty() {
parts.push(tail.to_owned());
}
parts
}
fn identifier_after_keyword(statement: &str, keyword: &str) -> Option<String> {
let lower = statement.to_ascii_lowercase();
let keyword_pos = lower.find(keyword)?;
let mut remainder = statement[keyword_pos + keyword.len()..].trim_start();
for prefix in ["if not exists", "if exists", "only"] {
if remainder.to_ascii_lowercase().starts_with(prefix) {
remainder = remainder[prefix.len()..].trim_start();
}
}
let identifier = remainder
.split(|ch: char| ch.is_whitespace() || ch == '(')
.next()?
.trim_matches('"')
.rsplit('.')
.next()?
.trim_matches('"')
.to_ascii_lowercase();
(!identifier.is_empty()).then_some(identifier)
}
fn first_parenthesized_columns(input: &str) -> Vec<String> {
let Some(open) = input.find('(') else {
return Vec::new();
};
let Some(close) = find_matching_paren(input, open) else {
return Vec::new();
};
split_top_level_csv(&input[open + 1..close])
.into_iter()
.filter_map(|column| {
let name = column
.trim()
.trim_matches('"')
.split_whitespace()
.next()?
.trim_matches('"')
.to_ascii_lowercase();
(!name.is_empty()).then_some(name)
})
.collect()
}
fn column_definition_name(definition: &str) -> Option<String> {
let trimmed = definition.trim();
let lower = trimmed.to_ascii_lowercase();
if lower.starts_with("constraint ")
|| lower.starts_with("primary key")
|| lower.starts_with("foreign key")
|| lower.starts_with("unique")
|| lower.starts_with("check ")
|| lower.starts_with("exclude ")
{
return None;
}
let name = trimmed
.split_whitespace()
.next()?
.trim_matches('"')
.to_ascii_lowercase();
(!name.is_empty()).then_some(name)
}
fn create_table_body(statement: &str) -> Option<(String, Vec<String>)> {
let table = identifier_after_keyword(statement, "create table")?;
let open = statement.find('(')?;
let close = find_matching_paren(statement, open)?;
Some((table, split_top_level_csv(&statement[open + 1..close])))
}
fn create_table_definitions(sql: &str) -> Vec<(String, Vec<String>)> {
split_sql_statements(sql)
.into_iter()
.filter_map(|statement| {
let normalized = statement.trim_start().to_ascii_lowercase();
if !normalized.starts_with("create table") || normalized.contains(" partition of ")
{
return None;
}
create_table_body(&statement)
})
.collect()
}
fn create_tables(sql: &str) -> BTreeSet<String> {
create_table_definitions(sql)
.into_iter()
.map(|(table, _)| table)
.collect()
}
fn table_has_not_null_community_id(definitions: &[String]) -> bool {
definitions.iter().any(|definition| {
column_definition_name(definition).as_deref() == Some("community_id")
&& normalize_sql(definition).contains("not null")
})
}
fn operator_global_tables(sql: &str) -> BTreeSet<String> {
let mut globals = BTreeSet::new();
let normalized = normalize_sql(sql);
let Some(insert_pos) = normalized.find("insert into _operator_global_tables") else {
return globals;
};
for value in [
"communities",
"rate_limit_violations",
"_operator_global_tables",
] {
if normalized[insert_pos..].contains(&format!("'{value}'")) {
globals.insert(value.to_owned());
}
}
globals
}
fn scoped_tables(sql: &str) -> BTreeSet<String> {
let globals = operator_global_tables(sql);
create_tables(sql)
.into_iter()
.filter(|table| !globals.contains(table))
.collect()
}
fn constraint_lint_for_definition(table: &str, definition: &str) -> Option<ConstraintLint> {
let normalized = normalize_sql(definition);
let definition_without_name = if normalized.starts_with("constraint ") {
let after_constraint = definition
.trim_start()
.splitn(3, char::is_whitespace)
.nth(2)
.unwrap_or("");
normalize_sql(after_constraint)
} else {
normalized.clone()
};
if definition_without_name.starts_with("primary key") {
Some(ConstraintLint {
table: table.to_owned(),
kind: ConstraintKind::PrimaryKey,
description: definition.to_owned(),
columns: first_parenthesized_columns(&definition_without_name),
})
} else if definition_without_name.starts_with("unique") {
Some(ConstraintLint {
table: table.to_owned(),
kind: ConstraintKind::Unique,
description: definition.to_owned(),
columns: first_parenthesized_columns(&definition_without_name),
})
} else if definition_without_name.starts_with("foreign key") {
Some(ConstraintLint {
table: table.to_owned(),
kind: ConstraintKind::ForeignKey,
description: definition.to_owned(),
columns: first_parenthesized_columns(&definition_without_name),
})
} else if normalized.contains(" primary key") {
column_definition_name(definition).map(|column| ConstraintLint {
table: table.to_owned(),
kind: ConstraintKind::PrimaryKey,
description: definition.to_owned(),
columns: vec![column],
})
} else if normalized.contains(" references ") {
column_definition_name(definition).map(|column| ConstraintLint {
table: table.to_owned(),
kind: ConstraintKind::ForeignKey,
description: definition.to_owned(),
columns: vec![column],
})
} else if normalized.contains(" unique") {
column_definition_name(definition).map(|column| ConstraintLint {
table: table.to_owned(),
kind: ConstraintKind::Unique,
description: definition.to_owned(),
columns: vec![column],
})
} else {
None
}
}
fn table_constraints(sql: &str, scoped_tables: &BTreeSet<String>) -> Vec<ConstraintLint> {
create_table_definitions(sql)
.into_iter()
.filter(|(table, _)| scoped_tables.contains(table))
.flat_map(|(table, definitions)| {
definitions.into_iter().filter_map(move |definition| {
constraint_lint_for_definition(&table, &definition)
})
})
.collect()
}
fn alter_table_constraints(sql: &str, scoped_tables: &BTreeSet<String>) -> Vec<ConstraintLint> {
split_sql_statements(sql)
.into_iter()
.filter_map(|statement| {
let normalized = normalize_sql(&statement);
if !normalized.starts_with("alter table") {
return None;
}
let table = identifier_after_keyword(&statement, "alter table")?;
if !scoped_tables.contains(&table) {
return None;
}
let add_pos = normalized.find(" add ")?;
let definition = normalized[add_pos + " add ".len()..].trim();
constraint_lint_for_definition(&table, definition)
})
.collect()
}
fn unique_indexes(sql: &str, scoped_tables: &BTreeSet<String>) -> Vec<ConstraintLint> {
split_sql_statements(sql)
.into_iter()
.filter_map(|statement| {
let normalized = normalize_sql(&statement);
if !normalized.starts_with("create unique index") {
return None;
}
let lower_statement = statement.to_ascii_lowercase();
let on_pos = lower_statement.find(" on ")?;
let table = statement[on_pos + " on ".len()..]
.trim_start()
.split(|ch: char| ch.is_whitespace() || ch == '(')
.next()?
.trim_matches('"')
.rsplit('.')
.next()?
.trim_matches('"')
.to_ascii_lowercase();
scoped_tables.contains(&table).then(|| ConstraintLint {
table,
kind: ConstraintKind::Unique,
description: statement.clone(),
columns: first_parenthesized_columns(&statement[on_pos + " on ".len()..]),
})
})
.collect()
}
fn scoped_constraint_lints(sql: &str, scoped_tables: &BTreeSet<String>) -> Vec<ConstraintLint> {
let mut constraints = table_constraints(sql, scoped_tables);
constraints.extend(alter_table_constraints(sql, scoped_tables));
constraints.extend(unique_indexes(sql, scoped_tables));
constraints
}
fn is_allowed_partition_primary_key_exception(constraint: &ConstraintLint) -> bool {
constraint.table == "delivery_log"
&& constraint.kind == ConstraintKind::PrimaryKey
&& constraint.columns == ["delivered_at", "id"]
}
fn scoped_constraint_violations(sql: &str) -> Vec<ConstraintLint> {
let scoped_tables = scoped_tables(sql);
scoped_constraint_lints(sql, &scoped_tables)
.into_iter()
.filter(|constraint| {
if is_allowed_partition_primary_key_exception(constraint) {
return false;
}
constraint.columns.first().map(String::as_str) != Some("community_id")
})
.collect()
}
fn has_channels_community_id_immutability_guard(sql: &str) -> bool {
let normalized = normalize_sql(sql);
normalized.contains("create trigger")
&& normalized.contains("before update")
&& normalized.contains(" on channels")
&& normalized.contains("community_id")
&& normalized.contains("old.community_id")
&& normalized.contains("new.community_id")
&& normalized.contains("raise exception")
}
fn forbidden_channels_community_id_mutations(sql: &str) -> Vec<String> {
split_sql_statements(sql)
.into_iter()
.filter(|statement| {
let normalized = normalize_sql(statement);
let updates_channels =
identifier_after_keyword(statement, "update").as_deref() == Some("channels");
let mutates_with_update = updates_channels
&& normalized.contains(" set ")
&& normalized.contains("community_id");
let alters_channels = identifier_after_keyword(statement, "alter table").as_deref()
== Some("channels");
let drops_channels = identifier_after_keyword(statement, "drop table").as_deref()
== Some("channels");
let drops_or_rewrites_column = alters_channels
&& (normalized.contains("drop column community_id")
|| normalized.contains("alter column community_id")
|| normalized.contains("rename column community_id")
|| normalized.contains("rename community_id")
|| normalized.contains("drop trigger")
|| normalized.contains("disable trigger"));
mutates_with_update || drops_or_rewrites_column || drops_channels
})
.collect()
}
#[test]
fn embedded_migrator_contains_all_schema_migrations() {
fn embedded_migrator_contains_consolidated_initial_schema() {
let migrations: Vec<_> = MIGRATOR.iter().collect();
assert_eq!(migrations.len(), 3);
assert_eq!(migrations.len(), 1);
assert_eq!(migrations[0].version, 1);
assert_eq!(&*migrations[0].description, "initial schema");
assert!(
migrations[0].sql.as_str().contains("CREATE TABLE channels"),
"initial schema migration should include Buzz core tables"
);
assert!(
migrations[0]
.sql
.as_str()
.contains("CREATE TABLE IF NOT EXISTS relay_members"),
"initial schema migration should include relay_members"
);
assert!(migrations[0]
.sql
.as_str()
.contains("CREATE TABLE communities"));
assert!(migrations[0].sql.as_str().contains("CREATE TABLE channels"));
assert!(migrations[0]
.sql
.as_str()
.contains("CREATE TABLE scheduled_workflow_fires"));
assert!(migrations[0]
.sql
.as_str()
.contains("CREATE TABLE audit_log"));
assert!(migrations[0]
.sql
.as_str()
.contains("CREATE TABLE _operator_global_tables"));
assert!(migrations[0]
.sql
.as_str()
.contains("search_tsv TSVECTOR GENERATED ALWAYS"));
}
assert_eq!(migrations[1].version, 2);
assert_eq!(&*migrations[1].description, "backfill d tag");
assert!(
migrations[1].sql.as_str().contains("UPDATE events"),
"second migration should backfill existing event rows"
#[test]
fn migration_lint_detects_tables_missing_community_id_by_default() {
let sql = r#"
CREATE TABLE communities (id UUID PRIMARY KEY);
CREATE TABLE widgets (id UUID PRIMARY KEY);
CREATE TABLE _operator_global_tables (table_name TEXT PRIMARY KEY, reason TEXT NOT NULL);
INSERT INTO _operator_global_tables (table_name, reason) VALUES
('communities', 'tenant registry'),
('_operator_global_tables', 'registry');
"#;
let definitions = create_table_definitions(sql);
let scoped = scoped_tables(sql);
let missing = definitions
.into_iter()
.filter(|(table, _)| scoped.contains(table))
.filter(|(_, definitions)| !table_has_not_null_community_id(definitions))
.map(|(table, _)| table)
.collect::<Vec<_>>();
assert_eq!(missing, vec!["widgets"]);
}
#[test]
fn migration_lint_detects_scoped_key_constraints_not_led_by_community_id() {
let sql = r#"
CREATE TABLE widgets (
community_id UUID NOT NULL,
id UUID PRIMARY KEY,
channel_id UUID REFERENCES channels(id),
slug TEXT,
CONSTRAINT widgets_name_unique UNIQUE (slug),
CONSTRAINT widgets_parent_fk FOREIGN KEY (channel_id) REFERENCES channels(id)
);
CREATE UNIQUE INDEX idx_widgets_slug ON widgets (slug);
ALTER TABLE widgets ADD CONSTRAINT widgets_alter_slug_unique UNIQUE (slug);
ALTER TABLE widgets ADD CONSTRAINT widgets_alter_parent_fk FOREIGN KEY (channel_id) REFERENCES channels(id);
CREATE TABLE _operator_global_tables (table_name TEXT PRIMARY KEY, reason TEXT NOT NULL);
INSERT INTO _operator_global_tables (table_name, reason) VALUES
('_operator_global_tables', 'registry');
"#;
let violations = scoped_constraint_violations(sql);
assert!(violations
.iter()
.any(|violation| violation.kind == ConstraintKind::PrimaryKey));
assert_eq!(
violations
.iter()
.filter(|violation| violation.kind == ConstraintKind::ForeignKey)
.count(),
3
);
assert_eq!(
violations
.iter()
.filter(|violation| violation.kind == ConstraintKind::Unique)
.count(),
3
);
}
#[test]
fn migration_lint_accepts_scoped_key_constraints_led_by_community_id() {
let sql = r#"
CREATE TABLE widgets (
community_id UUID NOT NULL,
id UUID NOT NULL,
channel_id UUID NOT NULL,
slug TEXT NOT NULL,
PRIMARY KEY (community_id, id),
UNIQUE (community_id, slug),
FOREIGN KEY (community_id, channel_id) REFERENCES channels(community_id, id)
);
CREATE UNIQUE INDEX idx_widgets_slug ON widgets (community_id, slug);
ALTER TABLE widgets ADD CONSTRAINT widgets_alter_slug_unique UNIQUE (community_id, slug);
ALTER TABLE widgets ADD CONSTRAINT widgets_alter_parent_fk FOREIGN KEY (community_id, channel_id) REFERENCES channels(community_id, id);
CREATE TABLE _operator_global_tables (table_name TEXT PRIMARY KEY, reason TEXT NOT NULL);
INSERT INTO _operator_global_tables (table_name, reason) VALUES
('_operator_global_tables', 'registry');
"#;
assert!(scoped_constraint_violations(sql).is_empty());
}
#[test]
fn all_non_operator_global_tables_have_not_null_community_id() {
let sql = migration_sql();
let scoped = scoped_tables(sql);
let missing = create_table_definitions(sql)
.into_iter()
.filter(|(table, _)| scoped.contains(table))
.filter(|(_, definitions)| !table_has_not_null_community_id(definitions))
.map(|(table, _)| table)
.collect::<Vec<_>>();
assert_eq!(migrations[2].version, 3);
assert_eq!(&*migrations[2].description, "event reminders");
assert!(
migrations[2]
.sql
.as_str()
.contains("ADD COLUMN not_before BIGINT")
&& migrations[2].sql.as_str().contains("idx_events_not_before"),
"third migration should add the NIP-ER reminder columns and index"
missing.is_empty(),
"every table not listed in _operator_global_tables must carry NOT NULL community_id; missing: {}",
missing.join(", ")
);
}
#[test]
fn scoped_primary_key_unique_and_foreign_key_constraints_lead_with_community_id() {
let sql = migration_sql();
let violations = scoped_constraint_violations(sql)
.into_iter()
.map(|constraint| {
format!(
"{}. {:?} constraint must lead with community_id: {}",
constraint.table, constraint.kind, constraint.description
)
})
.collect::<Vec<_>>();
assert!(
violations.is_empty(),
"tenant-scoped tables are all tables not listed in _operator_global_tables; primary key, unique/FK constraints, and unique indexes on those tables must lead with community_id:\n{}",
violations.join("\n")
);
}
#[test]
fn channels_community_id_is_immutable_after_insert() {
let sql = migration_sql();
let forbidden_mutations = forbidden_channels_community_id_mutations(sql);
assert!(
forbidden_mutations.is_empty(),
"channels.community_id must not be re-tenanted after insert; forbidden migration statements:\n{}",
forbidden_mutations.join("\n---\n")
);
assert!(
has_channels_community_id_immutability_guard(sql),
"migrations define channels.community_id but no BEFORE UPDATE trigger/function guard that rejects OLD.community_id <> NEW.community_id was found"
);
}
@@ -192,88 +656,35 @@ mod tests {
.expect("read applied migrations")
}
/// Returns `schema/schema.sql` with the NIP-ER reminder DDL removed, so it
/// models a pre-stack deployment whose `events` table lacks the reminder
/// columns and index. The strip is asserted: if the snapshot text drifts so
/// these fragments no longer match, the test fails loudly rather than
/// silently loading a snapshot that already carries the reminder columns
/// (which would make migration 0003 collide on re-add).
fn pre_reminder_schema_snapshot() -> String {
const REMINDER_COLUMNS: &str = " not_before BIGINT,\n delivered_at BIGINT,\n";
const REMINDER_INDEX: &str = "CREATE INDEX idx_events_not_before ON events (not_before)\n WHERE not_before IS NOT NULL AND deleted_at IS NULL AND delivered_at IS NULL;\n";
assert!(
SCHEMA_SQL.contains(REMINDER_COLUMNS) && SCHEMA_SQL.contains(REMINDER_INDEX),
"schema.sql reminder DDL drifted; update pre_reminder_schema_snapshot to match"
);
SCHEMA_SQL
.replace(REMINDER_COLUMNS, "")
.replace(REMINDER_INDEX, "")
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn run_migrations_applies_embedded_versions_on_fresh_database() {
async fn run_migrations_applies_consolidated_initial_schema_on_fresh_database() {
let pool = connect_test_pool().await;
reset_public_schema(&pool).await;
run_migrations(&pool).await.expect("run migrations");
assert_eq!(applied_versions(&pool).await, vec![1, 2, 3]);
let events_exists = sqlx::query_scalar::<_, bool>(
"SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'events')",
)
.fetch_one(&pool)
.await
.expect("check events table");
assert!(events_exists);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn run_migrations_baselines_existing_schema_and_preserves_allowlist_backfill_path() {
let pool = connect_test_pool().await;
reset_public_schema(&pool).await;
// Load a pre-stack snapshot (without the NIP-ER reminder DDL) so the
// events table matches a real pre-SQLx deployment, which never had the
// reminder columns. Migration 0003 must then add them — proving the
// genuine prod-upgrade path, not a snapshot that already carries them.
sqlx::raw_sql(sqlx::AssertSqlSafe(pre_reminder_schema_snapshot()))
.execute(&pool)
.await
.expect("load pre-SQLx schema snapshot");
sqlx::query(
"INSERT INTO pubkey_allowlist (pubkey, added_at) VALUES (decode($1, 'hex'), now())",
)
.bind("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
.execute(&pool)
.await
.expect("seed legacy allowlist row");
run_migrations(&pool).await.expect("baseline migrations");
assert_eq!(applied_versions(&pool).await, vec![1, 2, 3]);
let allowlist_count = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM pubkey_allowlist")
assert_eq!(applied_versions(&pool).await, vec![1]);
let tables = create_tables(migration_sql());
for table in [
"communities",
"events",
"channels",
"scheduled_workflow_fires",
"audit_log",
] {
let exists = sqlx::query_scalar::<_, bool>(
"SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = $1)",
)
.bind(table)
.fetch_one(&pool)
.await
.expect("count allowlist rows");
assert_eq!(
allowlist_count, 1,
"baseline must not drop legacy allowlist rows before relay startup backfills them"
);
let inserted = crate::relay_members::backfill_from_allowlist(&pool)
.await
.expect("backfill legacy allowlist rows");
assert_eq!(inserted, 1);
let relay_member_count = sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM relay_members WHERE pubkey = $1 AND role = 'member'",
)
.bind("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
.fetch_one(&pool)
.await
.expect("count backfilled relay member");
assert_eq!(relay_member_count, 1);
.unwrap_or_else(|err| panic!("check table {table}: {err}"));
assert!(
tables.contains(table),
"migration parser should see {table}"
);
assert!(exists, "migration should create {table}");
}
}
}
crates/buzz-db/src/thread.rs
+72 -5
View File
@@ -659,7 +659,7 @@ pub async fn get_thread_metadata_by_event(
mod tests {
use super::*;
use crate::{
channel::{create_channel, ChannelType, ChannelVisibility},
channel::{ChannelType, ChannelVisibility},
event::{insert_event_with_thread_metadata, ThreadMetadataParams},
};
use nostr::{EventBuilder, Keys, Kind};
@@ -687,12 +687,76 @@ mod tests {
.expect("event timestamp is valid")
}
async fn make_test_community(pool: &PgPool) -> Uuid {
let id = Uuid::new_v4();
let host = format!("thread-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
}
async fn create_test_channel(
pool: &PgPool,
name: &str,
channel_type: ChannelType,
visibility: ChannelVisibility,
description: Option<&str>,
created_by: &[u8],
ttl_seconds: Option<i32>,
) -> crate::error::Result<(crate::channel::ChannelRecord, buzz_core::CommunityId)> {
let id = Uuid::new_v4();
let community_id = make_test_community(pool).await;
sqlx::query(
r#"
INSERT INTO channels
(id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline)
VALUES
($1, $2, $3, $4::channel_type, $5::channel_visibility, $6, $7, $8,
CASE WHEN $8 IS NOT NULL THEN NOW() + ($8 || ' seconds')::interval ELSE NULL END)
"#,
)
.bind(id)
.bind(community_id)
.bind(name)
.bind(channel_type.as_str())
.bind(visibility.as_str())
.bind(description)
.bind(created_by)
.bind(ttl_seconds)
.execute(pool)
.await
.expect("insert test channel");
sqlx::query(
r#"
INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by)
VALUES ($1, $2, $3, 'owner', $4)
"#,
)
.bind(community_id)
.bind(id)
.bind(created_by)
.bind(created_by)
.execute(pool)
.await
.expect("insert owner membership");
crate::channel::get_channel(pool, id)
.await
.map(|channel| (channel, buzz_core::CommunityId::from_uuid(community_id)))
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn get_thread_replies_reconstructs_stored_events() {
let pool = setup_pool().await;
let author = Keys::generate();
let channel = create_channel(
let (channel, community) = create_test_channel(
&pool,
&format!("thread-replies-{}", Uuid::new_v4()),
ChannelType::Stream,
@@ -706,7 +770,7 @@ mod tests {
let root = make_stream_event(&author, "root");
let root_created_at = event_created_at(&root);
insert_event_with_thread_metadata(&pool, &root, Some(channel.id), None)
insert_event_with_thread_metadata(&pool, community, &root, Some(channel.id), None)
.await
.expect("insert root event");
@@ -715,6 +779,7 @@ mod tests {
let reply_id = reply.id.to_hex();
insert_event_with_thread_metadata(
&pool,
community,
&reply,
Some(channel.id),
Some(ThreadMetadataParams {
@@ -752,7 +817,7 @@ mod tests {
async fn get_thread_replies_skips_unreconstructable_row() {
let pool = setup_pool().await;
let author = Keys::generate();
let channel = create_channel(
let (channel, community) = create_test_channel(
&pool,
&format!("thread-replies-corrupt-{}", Uuid::new_v4()),
ChannelType::Stream,
@@ -766,7 +831,7 @@ mod tests {
let root = make_stream_event(&author, "root");
let root_created_at = event_created_at(&root);
insert_event_with_thread_metadata(&pool, &root, Some(channel.id), None)
insert_event_with_thread_metadata(&pool, community, &root, Some(channel.id), None)
.await
.expect("insert root event");
@@ -776,6 +841,7 @@ mod tests {
let good_created_at = event_created_at(&good);
insert_event_with_thread_metadata(
&pool,
community,
&good,
Some(channel.id),
Some(ThreadMetadataParams {
@@ -797,6 +863,7 @@ mod tests {
let bad_created_at = event_created_at(&bad);
insert_event_with_thread_metadata(
&pool,
community,
&bad,
Some(channel.id),
Some(ThreadMetadataParams {
crates/buzz-db/src/user.rs
+106 -50
View File
@@ -1,6 +1,7 @@
//! User CRUD operations.
use crate::error::Result;
use buzz_core::CommunityId;
use sqlx::PgPool;
use sqlx::Row;
@@ -34,14 +35,15 @@ pub struct UserSearchProfile {
/// Ensure a user record exists for the given pubkey (upsert).
/// Creates with minimal fields if not present; no-op if already exists.
pub async fn ensure_user(pool: &PgPool, pubkey: &[u8]) -> Result<()> {
pub async fn ensure_user(pool: &PgPool, community_id: CommunityId, pubkey: &[u8]) -> Result<()> {
sqlx::query(
r#"
INSERT INTO users (pubkey)
VALUES ($1)
INSERT INTO users (community_id, pubkey)
VALUES ($1, $2)
ON CONFLICT DO NOTHING
"#,
)
.bind(community_id.as_uuid())
.bind(pubkey)
.execute(pool)
.await?;
@@ -49,7 +51,11 @@ pub async fn ensure_user(pool: &PgPool, pubkey: &[u8]) -> Result<()> {
}
/// Get a single user record by pubkey.
pub async fn get_user(pool: &PgPool, pubkey: &[u8]) -> Result<Option<UserProfile>> {
pub async fn get_user(
pool: &PgPool,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Option<UserProfile>> {
let row = sqlx::query_as::<
_,
(
@@ -63,9 +69,10 @@ pub async fn get_user(pool: &PgPool, pubkey: &[u8]) -> Result<Option<UserProfile
r#"
SELECT pubkey, display_name, avatar_url, about, nip05_handle
FROM users
WHERE pubkey = $1
WHERE community_id = $1 AND pubkey = $2
"#,
)
.bind(community_id.as_uuid())
.bind(pubkey)
.fetch_optional(pool)
.await?;
@@ -91,6 +98,7 @@ pub async fn get_user(pool: &PgPool, pubkey: &[u8]) -> Result<Option<UserProfile
/// but multiple empty strings would violate uniqueness).
pub async fn update_user_profile(
pool: &PgPool,
community_id: CommunityId,
pubkey: &[u8],
display_name: Option<&str>,
avatar_url: Option<&str>,
@@ -129,8 +137,9 @@ pub async fn update_user_profile(
}
let sql = format!(
"UPDATE users SET {} WHERE pubkey = ${param_idx}",
set_parts.join(", ")
"UPDATE users SET {} WHERE community_id = ${param_idx} AND pubkey = ${}",
set_parts.join(", "),
param_idx + 1
);
let mut query = sqlx::query(sqlx::AssertSqlSafe(sql));
if display_name.is_some() {
@@ -145,6 +154,7 @@ pub async fn update_user_profile(
if nip05_handle.is_some() {
query = query.bind(empty_to_none(nip05_handle));
}
query = query.bind(community_id.as_uuid());
query = query.bind(pubkey);
query.execute(pool).await?;
Ok(())
@@ -154,6 +164,7 @@ pub async fn update_user_profile(
/// Both `local_part` and `domain` must already be lowercased by the caller.
pub async fn get_user_by_nip05(
pool: &PgPool,
community_id: CommunityId,
local_part: &str,
domain: &str,
) -> Result<Option<UserProfile>> {
@@ -171,10 +182,11 @@ pub async fn get_user_by_nip05(
r#"
SELECT pubkey, display_name, avatar_url, about, nip05_handle
FROM users
WHERE LOWER(nip05_handle) = LOWER($1)
WHERE community_id = $1 AND LOWER(nip05_handle) = LOWER($2)
LIMIT 1
"#,
)
.bind(community_id.as_uuid())
.bind(&handle)
.fetch_optional(pool)
.await?;
@@ -207,6 +219,7 @@ fn escape_like(input: &str) -> String {
/// Empty queries return an empty vec and do not hit the database.
pub async fn search_users(
pool: &PgPool,
community_id: CommunityId,
query: &str,
limit: u32,
) -> Result<Vec<UserSearchProfile>> {
@@ -224,23 +237,25 @@ pub async fn search_users(
r#"
SELECT pubkey, display_name, avatar_url, nip05_handle
FROM users
WHERE LOWER(COALESCE(display_name, '')) LIKE $1 ESCAPE '\'
OR LOWER(COALESCE(nip05_handle, '')) LIKE $1 ESCAPE '\'
OR LOWER(encode(pubkey, 'hex')) LIKE $1 ESCAPE '\'
WHERE community_id = $1
AND (LOWER(COALESCE(display_name, '')) LIKE $2 ESCAPE '\'
OR LOWER(COALESCE(nip05_handle, '')) LIKE $2 ESCAPE '\'
OR LOWER(encode(pubkey, 'hex')) LIKE $2 ESCAPE '\')
ORDER BY
CASE
WHEN LOWER(COALESCE(display_name, '')) = $2 THEN 0
WHEN LOWER(COALESCE(nip05_handle, '')) = $2 THEN 1
WHEN LOWER(encode(pubkey, 'hex')) = $2 THEN 2
WHEN LOWER(COALESCE(display_name, '')) LIKE $3 ESCAPE '\' THEN 3
WHEN LOWER(COALESCE(nip05_handle, '')) LIKE $3 ESCAPE '\' THEN 4
WHEN LOWER(encode(pubkey, 'hex')) LIKE $3 ESCAPE '\' THEN 5
WHEN LOWER(COALESCE(display_name, '')) = $3 THEN 0
WHEN LOWER(COALESCE(nip05_handle, '')) = $3 THEN 1
WHEN LOWER(encode(pubkey, 'hex')) = $3 THEN 2
WHEN LOWER(COALESCE(display_name, '')) LIKE $4 ESCAPE '\' THEN 3
WHEN LOWER(COALESCE(nip05_handle, '')) LIKE $4 ESCAPE '\' THEN 4
WHEN LOWER(encode(pubkey, 'hex')) LIKE $4 ESCAPE '\' THEN 5
ELSE 6
END,
COALESCE(NULLIF(display_name, ''), NULLIF(nip05_handle, ''), LOWER(encode(pubkey, 'hex')))
LIMIT $4
LIMIT $5
"#,
)
.bind(community_id.as_uuid())
.bind(&contains_pattern)
.bind(&normalized)
.bind(&prefix_pattern)
@@ -271,15 +286,17 @@ pub async fn search_users(
/// agent pubkey doesn't exist in the users table.
pub async fn set_agent_owner(
pool: &PgPool,
community_id: CommunityId,
agent_pubkey: &[u8],
owner_pubkey: &[u8],
) -> Result<bool> {
// Conditional UPDATE: only set owner if currently NULL. This makes
// "first mint wins" atomic — no TOCTOU race between concurrent mints.
let result = sqlx::query(
r#"UPDATE users SET agent_owner_pubkey = $1 WHERE pubkey = $2 AND agent_owner_pubkey IS NULL"#,
r#"UPDATE users SET agent_owner_pubkey = $1 WHERE community_id = $2 AND pubkey = $3 AND agent_owner_pubkey IS NULL"#,
)
.bind(owner_pubkey)
.bind(community_id.as_uuid())
.bind(agent_pubkey)
.execute(pool)
.await?;
@@ -287,7 +304,8 @@ pub async fn set_agent_owner(
if result.rows_affected() == 0 {
// Could be: (a) pubkey not found, or (b) owner already set.
// Check which case by querying the row.
let exists = sqlx::query(r#"SELECT 1 FROM users WHERE pubkey = $1"#)
let exists = sqlx::query(r#"SELECT 1 FROM users WHERE community_id = $1 AND pubkey = $2"#)
.bind(community_id.as_uuid())
.bind(agent_pubkey)
.fetch_optional(pool)
.await?;
@@ -307,11 +325,13 @@ pub async fn set_agent_owner(
/// Returns Some((policy_str, owner_bytes_or_none)) if found.
pub async fn get_agent_channel_policy(
pool: &PgPool,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Option<(String, Option<Vec<u8>>)>> {
let row = sqlx::query(
r#"SELECT channel_add_policy::text AS channel_add_policy, agent_owner_pubkey FROM users WHERE pubkey = $1"#,
r#"SELECT channel_add_policy::text AS channel_add_policy, agent_owner_pubkey FROM users WHERE community_id = $1 AND pubkey = $2"#,
)
.bind(community_id.as_uuid())
.bind(pubkey)
.fetch_optional(pool)
.await?;
@@ -329,12 +349,14 @@ pub async fn get_agent_channel_policy(
/// `get_agent_channel_policy`, which would fetch unrelated fields.
pub async fn is_agent_owner(
pool: &PgPool,
community_id: CommunityId,
target_pubkey: &[u8],
actor_pubkey: &[u8],
) -> Result<bool> {
let row = sqlx::query_scalar::<_, bool>(
"SELECT agent_owner_pubkey = $2 FROM users WHERE pubkey = $1 AND agent_owner_pubkey IS NOT NULL",
"SELECT agent_owner_pubkey = $3 FROM users WHERE community_id = $1 AND pubkey = $2 AND agent_owner_pubkey IS NOT NULL",
)
.bind(community_id.as_uuid())
.bind(target_pubkey)
.bind(actor_pubkey)
.fetch_optional(pool)
@@ -345,16 +367,22 @@ pub async fn is_agent_owner(
/// Set the channel_add_policy for a user.
/// Returns an error if the pubkey is not found (rows_affected == 0).
/// Returns an error if `policy` is not one of the valid ENUM values.
pub async fn set_channel_add_policy(pool: &PgPool, pubkey: &[u8], policy: &str) -> Result<()> {
pub async fn set_channel_add_policy(
pool: &PgPool,
community_id: CommunityId,
pubkey: &[u8],
policy: &str,
) -> Result<()> {
if !matches!(policy, "anyone" | "owner_only" | "nobody") {
return Err(crate::error::DbError::InvalidData(format!(
"invalid channel_add_policy: {policy}"
)));
}
let result = sqlx::query(
r#"UPDATE users SET channel_add_policy = $1::channel_add_policy WHERE pubkey = $2"#,
r#"UPDATE users SET channel_add_policy = $1::channel_add_policy WHERE community_id = $2 AND pubkey = $3"#,
)
.bind(policy)
.bind(community_id.as_uuid())
.bind(pubkey)
.execute(pool)
.await?;
@@ -385,28 +413,41 @@ mod tests {
Keys::generate().public_key().to_bytes().to_vec()
}
async fn make_community(pool: &PgPool) -> CommunityId {
let id = uuid::Uuid::new_v4();
let host = format!("user-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");
CommunityId::from_uuid(id)
}
/// Setting an agent owner then reading back the policy should return
/// the default "anyone" policy and the owner pubkey.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn test_set_agent_owner_and_get_policy() {
let db = setup_db().await;
let community = make_community(&db.pool).await;
let agent_pk = random_pubkey();
let owner_pk = random_pubkey();
ensure_user(&db.pool, &agent_pk)
ensure_user(&db.pool, community, &agent_pk)
.await
.expect("ensure agent");
ensure_user(&db.pool, &owner_pk)
ensure_user(&db.pool, community, &owner_pk)
.await
.expect("ensure owner");
let was_set = set_agent_owner(&db.pool, &agent_pk, &owner_pk)
let was_set = set_agent_owner(&db.pool, community, &agent_pk, &owner_pk)
.await
.expect("set_agent_owner");
assert!(was_set, "first set_agent_owner should return true");
let result = get_agent_channel_policy(&db.pool, &agent_pk)
let result = get_agent_channel_policy(&db.pool, community, &agent_pk)
.await
.expect("get_agent_channel_policy");
@@ -424,14 +465,17 @@ mod tests {
#[ignore = "requires Postgres"]
async fn test_set_channel_add_policy() {
let db = setup_db().await;
let community = make_community(&db.pool).await;
let pk = random_pubkey();
ensure_user(&db.pool, &pk).await.expect("ensure user");
ensure_user(&db.pool, community, &pk)
.await
.expect("ensure user");
// owner_only
set_channel_add_policy(&db.pool, &pk, "owner_only")
set_channel_add_policy(&db.pool, community, &pk, "owner_only")
.await
.expect("set owner_only");
let (policy, owner) = get_agent_channel_policy(&db.pool, &pk)
let (policy, owner) = get_agent_channel_policy(&db.pool, community, &pk)
.await
.expect("get policy")
.expect("should be Some");
@@ -439,10 +483,10 @@ mod tests {
assert!(owner.is_none(), "no owner was set");
// nobody
set_channel_add_policy(&db.pool, &pk, "nobody")
set_channel_add_policy(&db.pool, community, &pk, "nobody")
.await
.expect("set nobody");
let (policy, owner) = get_agent_channel_policy(&db.pool, &pk)
let (policy, owner) = get_agent_channel_policy(&db.pool, community, &pk)
.await
.expect("get policy")
.expect("should be Some");
@@ -450,10 +494,10 @@ mod tests {
assert!(owner.is_none());
// anyone (reset to default)
set_channel_add_policy(&db.pool, &pk, "anyone")
set_channel_add_policy(&db.pool, community, &pk, "anyone")
.await
.expect("set anyone");
let (policy, owner) = get_agent_channel_policy(&db.pool, &pk)
let (policy, owner) = get_agent_channel_policy(&db.pool, community, &pk)
.await
.expect("get policy")
.expect("should be Some");
@@ -467,9 +511,10 @@ mod tests {
#[ignore = "requires Postgres"]
async fn test_get_policy_unknown_pubkey() {
let db = setup_db().await;
let community = make_community(&db.pool).await;
let pk = random_pubkey();
let result = get_agent_channel_policy(&db.pool, &pk)
let result = get_agent_channel_policy(&db.pool, community, &pk)
.await
.expect("query should not error");
@@ -482,15 +527,16 @@ mod tests {
#[ignore = "requires Postgres"]
async fn test_set_agent_owner_nonexistent_agent() {
let db = setup_db().await;
let community = make_community(&db.pool).await;
let agent_pk = random_pubkey();
let owner_pk = random_pubkey();
// Only ensure the owner exists -- agent is intentionally absent.
ensure_user(&db.pool, &owner_pk)
ensure_user(&db.pool, community, &owner_pk)
.await
.expect("ensure owner");
let result = set_agent_owner(&db.pool, &agent_pk, &owner_pk).await;
let result = set_agent_owner(&db.pool, community, &agent_pk, &owner_pk).await;
assert!(
result.is_err(),
"should error when agent pubkey is not in users table"
@@ -502,28 +548,33 @@ mod tests {
#[ignore = "requires Postgres"]
async fn test_set_agent_owner_already_owned() {
let db = setup_db().await;
let community = make_community(&db.pool).await;
let agent_pk = random_pubkey();
let owner1 = random_pubkey();
let owner2 = random_pubkey();
ensure_user(&db.pool, &agent_pk)
ensure_user(&db.pool, community, &agent_pk)
.await
.expect("ensure agent");
ensure_user(&db.pool, &owner1).await.expect("ensure owner1");
ensure_user(&db.pool, &owner2).await.expect("ensure owner2");
ensure_user(&db.pool, community, &owner1)
.await
.expect("ensure owner1");
ensure_user(&db.pool, community, &owner2)
.await
.expect("ensure owner2");
let first = set_agent_owner(&db.pool, &agent_pk, &owner1)
let first = set_agent_owner(&db.pool, community, &agent_pk, &owner1)
.await
.expect("first set");
assert!(first, "first set should succeed");
let second = set_agent_owner(&db.pool, &agent_pk, &owner2)
let second = set_agent_owner(&db.pool, community, &agent_pk, &owner2)
.await
.expect("second set should not error");
assert!(!second, "second set should return false (already owned)");
// Verify original owner is preserved.
let (_, owner) = get_agent_channel_policy(&db.pool, &agent_pk)
let (_, owner) = get_agent_channel_policy(&db.pool, community, &agent_pk)
.await
.expect("get policy")
.expect("should be Some");
@@ -536,9 +587,10 @@ mod tests {
#[ignore = "requires Postgres"]
async fn test_set_channel_add_policy_nonexistent_user() {
let db = setup_db().await;
let community = make_community(&db.pool).await;
let pk = random_pubkey();
let result = set_channel_add_policy(&db.pool, &pk, "nobody").await;
let result = set_channel_add_policy(&db.pool, community, &pk, "nobody").await;
assert!(
result.is_err(),
"should error when pubkey is not in users table"
@@ -549,9 +601,10 @@ mod tests {
#[ignore = "requires Postgres"]
async fn test_set_channel_add_policy_rejects_invalid() {
let db = setup_db().await;
let community = make_community(&db.pool).await;
let pubkey = nostr::Keys::generate().public_key().to_bytes().to_vec();
ensure_user(&db.pool, &pubkey).await.unwrap();
let result = set_channel_add_policy(&db.pool, &pubkey, "invalid_policy").await;
ensure_user(&db.pool, community, &pubkey).await.unwrap();
let result = set_channel_add_policy(&db.pool, community, &pubkey, "invalid_policy").await;
assert!(result.is_err(), "should reject invalid policy value");
}
@@ -595,14 +648,17 @@ mod tests {
#[ignore = "requires Postgres"]
async fn test_owner_only_with_no_owner() {
let db = setup_db().await;
let community = make_community(&db.pool).await;
let pk = random_pubkey();
ensure_user(&db.pool, &pk).await.expect("ensure user");
ensure_user(&db.pool, community, &pk)
.await
.expect("ensure user");
set_channel_add_policy(&db.pool, &pk, "owner_only")
set_channel_add_policy(&db.pool, community, &pk, "owner_only")
.await
.expect("set owner_only");
let result = get_agent_channel_policy(&db.pool, &pk)
let result = get_agent_channel_policy(&db.pool, community, &pk)
.await
.expect("get policy")
.expect("should be Some");
crates/buzz-db/src/workflow.rs
+407 -7
View File
@@ -15,6 +15,8 @@ use sha2::{Digest, Sha256};
use sqlx::{PgPool, Row};
use uuid::Uuid;
use buzz_core::CommunityId;
use crate::error::{DbError, Result};
// -- Token hashing ------------------------------------------------------------
@@ -163,6 +165,8 @@ impl FromStr for ApprovalStatus {
pub struct WorkflowRecord {
/// Unique workflow identifier.
pub id: Uuid,
/// Server-resolved community that owns this workflow.
pub community_id: CommunityId,
/// Human-readable workflow name.
pub name: String,
/// Compressed public key bytes of the workflow owner.
@@ -211,6 +215,23 @@ pub struct WorkflowRunRecord {
pub created_at: DateTime<Utc>,
}
/// A winning scheduled workflow fire claim.
///
/// The primary identity is `(workflow_id, scheduled_for)`. `community_id` is
/// resolved from the workflow row inside the claim SQL and returned for scoped
/// audit/logging; callers never supply it as a claim.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ScheduledWorkflowFireClaim {
/// Community that owns this scheduled fire.
pub community_id: CommunityId,
/// Workflow definition that should run.
pub workflow_id: Uuid,
/// Authoritative schedule instant this claim represents.
pub scheduled_for: DateTime<Utc>,
/// Database timestamp for when this pod won the claim.
pub claimed_at: DateTime<Utc>,
}
/// A pending or resolved approval gate for a workflow step.
#[derive(Debug, Clone)]
pub struct ApprovalRecord {
@@ -244,6 +265,7 @@ pub struct ApprovalRecord {
/// New workflows start as `active` and `enabled = TRUE`.
pub async fn create_workflow(
pool: &PgPool,
community_id: CommunityId,
channel_id: Option<Uuid>,
owner_pubkey: &[u8],
name: &str,
@@ -255,11 +277,12 @@ pub async fn create_workflow(
sqlx::query(
r#"
INSERT INTO workflows
(id, name, owner_pubkey, channel_id, definition, definition_hash, status, enabled)
VALUES ($1, $2, $3, $4, $5::jsonb, $6, 'active', TRUE)
(id, community_id, name, owner_pubkey, channel_id, definition, definition_hash, status, enabled)
VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, 'active', TRUE)
"#,
)
.bind(id)
.bind(community_id.as_uuid())
.bind(name)
.bind(owner_pubkey)
.bind(channel_id)
@@ -275,7 +298,7 @@ pub async fn create_workflow(
pub async fn get_workflow(pool: &PgPool, id: Uuid) -> Result<WorkflowRecord> {
let row = sqlx::query(
r#"
SELECT id, name, owner_pubkey, channel_id, definition, definition_hash,
SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
status::text AS status, enabled, created_at, updated_at
FROM workflows
WHERE id = $1
@@ -304,7 +327,7 @@ pub async fn list_channel_workflows(
let rows = sqlx::query(
r#"
SELECT id, name, owner_pubkey, channel_id, definition, definition_hash,
SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
status::text AS status, enabled, created_at, updated_at
FROM workflows
WHERE channel_id = $1
@@ -333,7 +356,7 @@ pub async fn list_enabled_channel_workflows(
) -> Result<Vec<WorkflowRecord>> {
let rows = sqlx::query(
r#"
SELECT id, name, owner_pubkey, channel_id, definition, definition_hash,
SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
status::text AS status, enabled, created_at, updated_at
FROM workflows
WHERE channel_id = $1
@@ -359,7 +382,7 @@ pub async fn list_enabled_channel_workflows(
pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result<Vec<WorkflowRecord>> {
let rows = sqlx::query(
r#"
SELECT id, name, owner_pubkey, channel_id, definition, definition_hash,
SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
status::text AS status, enabled, created_at, updated_at
FROM workflows
WHERE status = 'active'
@@ -376,6 +399,125 @@ pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result<Vec<WorkflowRec
rows.into_iter().map(row_to_workflow_record).collect()
}
/// Claim a scheduled workflow fire for an authoritative schedule instant.
///
/// Returns `Some` only for the first pod that claims `(workflow_id,
/// scheduled_for)`. All other pods receive `None` and must skip creating a
/// workflow run. The `scheduled_for` value must come from an external
/// schedule anchor (cron expression) or DB-authoritative interval anchor; a
/// per-pod in-memory timestamp is not safe because different pods can compute
/// different claim keys.
pub async fn claim_scheduled_workflow_fire(
pool: &PgPool,
workflow_id: Uuid,
scheduled_for: DateTime<Utc>,
) -> Result<Option<ScheduledWorkflowFireClaim>> {
let row = sqlx::query(
r#"
INSERT INTO scheduled_workflow_fires (community_id, workflow_id, scheduled_for)
SELECT w.community_id, w.id, $2
FROM workflows w
WHERE w.id = $1
ON CONFLICT (community_id, workflow_id, scheduled_for) DO NOTHING
RETURNING community_id, workflow_id, scheduled_for, claimed_at
"#,
)
.bind(workflow_id)
.bind(scheduled_for)
.fetch_optional(pool)
.await?;
row.map(|row| {
let community_id: Uuid = row.try_get("community_id")?;
Ok(ScheduledWorkflowFireClaim {
community_id: CommunityId::from_uuid(community_id),
workflow_id: row.try_get("workflow_id")?,
scheduled_for: row.try_get("scheduled_for")?,
claimed_at: row.try_get("claimed_at")?,
})
})
.transpose()
}
/// Fetch the greatest claimed schedule instant for a workflow.
///
/// Interval schedulers use this as their DB-authoritative `last_fired` anchor.
/// It makes all pods compute the same next interval instant after a successful
/// claim, and preserves the interval clock across pod restarts. This intentionally
/// reads from `scheduled_workflow_fires`, not `workflow_runs`, because the claim
/// row is the source of truth for schedule deduplication.
pub async fn latest_scheduled_workflow_fire(
pool: &PgPool,
workflow_id: Uuid,
) -> Result<Option<DateTime<Utc>>> {
let row = sqlx::query(
r#"
SELECT MAX(scheduled_for) AS scheduled_for
FROM scheduled_workflow_fires
WHERE workflow_id = $1
"#,
)
.bind(workflow_id)
.fetch_one(pool)
.await?;
row.try_get("scheduled_for").map_err(Into::into)
}
/// Link a won scheduled-fire claim to the workflow run it created.
///
/// This is for ops/audit forensics only; the claim row remains the dedupe
/// boundary. If run creation succeeds, callers should attach the run id before
/// spawning execution. If run creation fails, leaving `workflow_run_id` NULL is
/// intentional: the schedule instant was claimed and must not duplicate later.
pub async fn attach_scheduled_workflow_run(
pool: &PgPool,
workflow_id: Uuid,
scheduled_for: DateTime<Utc>,
workflow_run_id: Uuid,
) -> Result<bool> {
let result = sqlx::query(
r#"
UPDATE scheduled_workflow_fires
SET workflow_run_id = $3
WHERE workflow_id = $1
AND scheduled_for = $2
AND workflow_run_id IS NULL
"#,
)
.bind(workflow_id)
.bind(scheduled_for)
.bind(workflow_run_id)
.execute(pool)
.await?;
Ok(result.rows_affected() == 1)
}
/// Delete old scheduled workflow fire claims for retention.
///
/// Schedule claim rows are correctness metadata, but they grow with every fire.
/// The relay/ops janitor should retain enough history for audits and interval
/// anchoring: the cutoff must be older than the largest interval schedule the
/// deployment supports, or interval workflows can lose their DB-authoritative
/// anchor after pruning.
pub async fn prune_scheduled_workflow_fires_before(
pool: &PgPool,
older_than: DateTime<Utc>,
) -> Result<u64> {
let result = sqlx::query(
r#"
DELETE FROM scheduled_workflow_fires
WHERE claimed_at < $1
"#,
)
.bind(older_than)
.execute(pool)
.await?;
Ok(result.rows_affected())
}
/// Update a workflow's name, definition, and definition_hash.
pub async fn update_workflow(
pool: &PgPool,
@@ -765,8 +907,11 @@ fn row_to_workflow_record(row: sqlx::postgres::PgRow) -> Result<WorkflowRecord>
let enabled: bool = row.try_get("enabled")?;
let community_id: Uuid = row.try_get("community_id")?;
Ok(WorkflowRecord {
id,
community_id: CommunityId::from_uuid(community_id),
name: row.try_get("name")?,
owner_pubkey: row.try_get("owner_pubkey")?,
channel_id,
@@ -831,7 +976,7 @@ pub async fn find_by_owner_and_name(
) -> Result<Option<WorkflowRecord>> {
let row = sqlx::query(
r#"
SELECT id, name, owner_pubkey, channel_id, definition, definition_hash,
SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
status::text AS status, enabled, created_at, updated_at
FROM workflows
WHERE owner_pubkey = $1 AND name = $2
@@ -955,8 +1100,11 @@ mod tests {
"steps": [{ "id": "s1", "action": "send_message", "text": "hi" }]
});
let community_id = CommunityId::from_uuid(Uuid::new_v4());
let record = WorkflowRecord {
id,
community_id,
name: "My Workflow".to_owned(),
owner_pubkey: vec![0xab; 32],
channel_id: Some(channel_id),
@@ -969,6 +1117,7 @@ mod tests {
};
assert_eq!(record.id, id);
assert_eq!(record.community_id, community_id);
assert_eq!(record.name, "My Workflow");
assert_eq!(record.owner_pubkey, vec![0xab; 32]);
assert_eq!(record.channel_id, Some(channel_id));
@@ -985,6 +1134,7 @@ mod tests {
let record = WorkflowRecord {
id,
community_id: CommunityId::from_uuid(Uuid::new_v4()),
name: "Global Workflow".to_owned(),
owner_pubkey: vec![0x00; 32],
channel_id: None,
@@ -1006,6 +1156,7 @@ mod tests {
let record = WorkflowRecord {
id,
community_id: CommunityId::from_uuid(Uuid::new_v4()),
name: "Original".to_owned(),
owner_pubkey: vec![0x01; 32],
channel_id: None,
@@ -1034,6 +1185,7 @@ mod tests {
] {
let record = WorkflowRecord {
id: Uuid::new_v4(),
community_id: CommunityId::from_uuid(Uuid::new_v4()),
name: "Test".to_owned(),
owner_pubkey: vec![],
channel_id: None,
@@ -1053,6 +1205,7 @@ mod tests {
let now = Utc::now();
let record = WorkflowRecord {
id: Uuid::new_v4(),
community_id: CommunityId::from_uuid(Uuid::new_v4()),
name: "Paused".to_owned(),
owner_pubkey: vec![],
channel_id: None,
@@ -1302,4 +1455,251 @@ mod tests {
assert_eq!(record.status, ApprovalStatus::Pending);
assert_eq!(cloned.status, ApprovalStatus::Granted);
}
// -- F1 / S1 attack surface for scheduled workflow claims ------------------
//
// These tests pin the locked spec from Eva [13] / Mari [12]:
//
// 1. `workflows.community_id` is row-owned, NOT NULL, immutable.
// 2. Claim resolves `community_id` server-side via `workflow_id`.
// 3. Claim uniqueness is `(workflow_id, scheduled_for)` — `workflow_id`
// is globally unique, so adding `community_id` to the key weakens it.
// 4. `latest_scheduled_workflow_fire` drops caller-supplied community.
//
// Until that lands, `claim_for_workflow_in_other_community_no_ops` is the
// S1 regression lock: today the schema permits a caller in community A to
// claim a workflow owned by community B and have `claimed.community_id`
// come back as A. That is a cross-tenant write surface (Theorem S1,
// `docs/multi-tenant-relay.md:248`: "the claimed community never appears
// in this function — only the resolved one").
//
// The other two tests are characterization guards: same-window race must
// yield exactly one claim winner, and the retention primitive's docstring
// caveat (pruning below the largest interval breaks `latest_*`) is
// load-bearing for the §5c deployment-config rule Sami flagged.
use crate::user::ensure_user;
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")
}
/// Insert a community with a unique host. Returns its `CommunityId`.
async fn make_community(pool: &PgPool) -> CommunityId {
let id = Uuid::new_v4();
let host = format!("test-{}.example", id.simple());
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(id)
.bind(&host)
.execute(pool)
.await
.expect("insert community");
CommunityId::from_uuid(id)
}
/// Insert a channel under a community. Returns the channel id.
async fn make_channel(pool: &PgPool, community: CommunityId, owner: &[u8]) -> Uuid {
let id = Uuid::new_v4();
sqlx::query(
r#"
INSERT INTO channels (id, community_id, name, created_by)
VALUES ($1, $2, $3, $4)
"#,
)
.bind(id)
.bind(community.as_uuid())
.bind(format!("ch-{}", id.simple()))
.bind(owner)
.execute(pool)
.await
.expect("insert channel");
id
}
/// Insert a workflow whose tenant is `community`'s channel. Returns the
/// workflow id and the owning community for callers that want to assert
/// the resolved tenant.
async fn make_workflow_in(pool: &PgPool, community: CommunityId) -> (Uuid, CommunityId) {
let owner = vec![0xa1; 32];
ensure_user(pool, community, &owner)
.await
.expect("ensure owner");
let channel_id = make_channel(pool, community, &owner).await;
let workflow_id = create_workflow(
pool,
community,
Some(channel_id),
&owner,
"f1-attack-workflow",
r#"{"trigger":{"on":"schedule"},"steps":[]}"#,
&[0u8; 32],
)
.await
.expect("create workflow");
(workflow_id, community)
}
/// F1 attack: a caller in community A must NOT be able to claim a fire
/// for a workflow owned by community B and have the claim resolve under
/// A's tenant. The resolved community on the returned claim row MUST
/// equal the workflow's actual tenant (B).
///
/// Post-fix (`1fa3d837f`) the claim signature no longer accepts a caller
/// tenant — the SQL resolves `community_id` from the `workflows` row via
/// `INSERT ... SELECT w.community_id, w.id, $2 FROM workflows w WHERE
/// w.id = $1`. This test still creates an "attacker_community" that the
/// caller is *not* permitted to name on the wire; the assertion is that
/// the resolved tenant equals the workflow owner's community, never the
/// attacker's. With the pre-fix signature this test was RED (the row's
/// `community_id` came back as the attacker's); under the locked spec it
/// must be GREEN.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn claim_for_workflow_in_other_community_no_ops() {
let pool = setup_pool().await;
// The attacker community exists in the schema but the caller has no
// way to pass it to the claim API anymore — that *is* the S1 fix.
// Keeping the row in this test makes the no-influence invariant
// explicit: even with two real tenants in play, the resolved
// community is the workflow's owner.
let _attacker_community = make_community(&pool).await;
let owner_community = make_community(&pool).await;
let (workflow_id, expected_community) = make_workflow_in(&pool, owner_community).await;
let scheduled_for = Utc.with_ymd_and_hms(2026, 6, 27, 0, 0, 0).unwrap();
let claim = claim_scheduled_workflow_fire(&pool, workflow_id, scheduled_for)
.await
.expect("claim should not error")
.expect("claim should succeed exactly once");
assert_eq!(
claim.community_id,
expected_community,
"claim must resolve community from workflow_id (server-side); \
resolved={resolved:?} expected={expected_community:?}",
resolved = claim.community_id,
);
assert_eq!(claim.workflow_id, workflow_id);
assert_eq!(claim.scheduled_for, scheduled_for);
}
/// Same `(workflow_id, scheduled_for)` claimed concurrently by N tasks
/// must yield exactly one `Some` winner. Post-fix the PK is
/// `(workflow_id, scheduled_for)` (`workflow_id` is globally unique,
/// `community_id` is a scoped/audit label only) — exactly the locked
/// spec. Characterization guard: protects the dedup boundary against
/// regressions in the claim SQL.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn concurrent_same_window_claims_exactly_one_wins() {
let pool = setup_pool().await;
let community = make_community(&pool).await;
let (workflow_id, _) = make_workflow_in(&pool, community).await;
let scheduled_for = Utc.with_ymd_and_hms(2026, 6, 27, 0, 1, 0).unwrap();
const N: usize = 8;
let mut handles = Vec::with_capacity(N);
for _ in 0..N {
let pool = pool.clone();
handles.push(tokio::spawn(async move {
claim_scheduled_workflow_fire(&pool, workflow_id, scheduled_for).await
}));
}
let mut winners = 0usize;
for h in handles {
let result = h.await.expect("task did not panic").expect("claim ok");
if result.is_some() {
winners += 1;
}
}
assert_eq!(
winners, 1,
"exactly one task must win the claim race for (workflow_id, scheduled_for)"
);
}
/// Documents the retention-vs-interval coupling Sami flagged for §5c:
/// pruning every claim below the workflow's interval makes
/// `latest_scheduled_workflow_fire` return `None`, which re-introduces the
/// per-pod-clock anchor bug F5 was meant to fix. Test is GREEN today and
/// MUST stay green — it pins the deployment-config rule that the janitor
/// cutoff must exceed `MAX(interval_secs) + safety margin`. If a future
/// change makes `latest_*` resilient to pruning (e.g. by reading the most
/// recent workflow_run instead, or by retaining a sentinel row), this
/// test's assertion encodes the contract that must be updated alongside.
///
/// Test isolation: the prune primitive is global (filters only on
/// `claimed_at`), so to avoid colliding with parallel claim tests we
/// back-date this workflow's `claimed_at` into the deep past and use a
/// past cutoff that cannot match any other test's `claimed_at = NOW()`.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn latest_after_prune_below_interval_breaks_anchor() {
let pool = setup_pool().await;
let community = make_community(&pool).await;
let (workflow_id, _) = make_workflow_in(&pool, community).await;
let scheduled_for = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
claim_scheduled_workflow_fire(&pool, workflow_id, scheduled_for)
.await
.expect("claim ok")
.expect("first claim wins");
// Backdate this row's `claimed_at` so the global prune below targets
// only this workflow's row and cannot race-delete other tests' rows.
let backdated_claimed_at = Utc.with_ymd_and_hms(2020, 1, 1, 0, 0, 0).unwrap();
sqlx::query(
"UPDATE scheduled_workflow_fires SET claimed_at = $1 \
WHERE community_id = $2 AND workflow_id = $3 AND scheduled_for = $4",
)
.bind(backdated_claimed_at)
.bind(community.as_uuid())
.bind(workflow_id)
.bind(scheduled_for)
.execute(&pool)
.await
.expect("backdate ok");
let latest_before = latest_scheduled_workflow_fire(&pool, workflow_id)
.await
.expect("latest ok");
assert_eq!(
latest_before,
Some(scheduled_for),
"latest must reflect the claim before pruning",
);
// Janitor cutoff above only the back-dated row: prunes the anchor row
// without touching anything claimed at wall-clock NOW.
let cutoff = backdated_claimed_at + chrono::Duration::seconds(1);
let pruned = prune_scheduled_workflow_fires_before(&pool, cutoff)
.await
.expect("prune ok");
assert!(
pruned >= 1,
"expected at least one row pruned, got {pruned}"
);
let latest_after = latest_scheduled_workflow_fire(&pool, workflow_id)
.await
.expect("latest ok");
assert_eq!(
latest_after, None,
"pruning below the largest interval breaks the DB anchor; \
retention cutoff MUST exceed MAX(interval_secs) + safety margin (§5c)",
);
}
}
crates/buzz-relay/src/handlers/command_executor.rs
+11 -1
View File
@@ -612,10 +612,20 @@ async fn handle_workflow_def(
PersistResult::Inserted(tx) => tx,
};
// 4. Execute: create_workflow
// 4. Execute: create_workflow. The workflow's community is resolved from
// the server-owned channel row, not from the client-supplied event. The DB
// also enforces `(community_id, channel_id)` as a composite FK.
let community_id = state
.db
.community_of_channel(channel_id)
.await
.map_err(|e| IngestError::Internal(format!("error: db channel community lookup: {e}")))?
.ok_or_else(|| IngestError::Rejected("invalid: workflow channel not found".into()))?;
let workflow_id = state
.db
.create_workflow(
community_id,
Some(channel_id),
&self_bytes,
&workflow_name,