jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-07 18:01:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge refreshed Bestie sidebar base

Browse Source 
On-behalf-of: mahanti <arjun.mahanti@gmail.com>
Signed-off-by: Fizz <fizz@buzz.local>
This commit is contained in:
Fizz
2026-08-28 11:10:40 -04:00
parent f91e8009b0 ee16f8ec40
commit 3c110cd535
3 changed files with 3203 additions and 3017 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2641
crates/buzz-db/src/channel.rs
View File

File diff suppressed because it is too large Load Diff

3017
crates/buzz-db/src/channel_members.rs Normal file
View File

File diff suppressed because it is too large Load Diff

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

@@ -30,8 +30,10 @@ pub mod admin_moderation;
pub mod api_token;
/// Relay-scoped archived identity persistence (NIP-IA).
pub mod archived_identities;
/// Channel and membership persistence.
/// Channel lifecycle and metadata persistence.
pub mod channel;
/// Channel membership and roster persistence.
pub mod channel_members;
/// Community lifecycle and host-map persistence.
pub mod community;
/// Durable whole-community deletion lifecycle and PostgreSQL adapter.
@@ -1776,363 +1778,6 @@ impl Db {
Ok(outcome)
}
/// Creates a new channel, bootstraps the creator as owner, and returns the record.
#[allow(clippy::too_many_arguments)]
#[datastore_span(name = "create_channel", system = "postgresql")]
pub async fn create_channel(
&self,
community_id: CommunityId,
name: &str,
channel_type: channel::ChannelType,
visibility: channel::ChannelVisibility,
description: Option<&str>,
created_by: &[u8],
ttl_seconds: Option<i32>,
) -> Result<channel::ChannelRecord> {
channel::create_channel(
&self.pool,
community_id,
name,
channel_type,
visibility,
description,
created_by,
ttl_seconds,
)
.await
}
/// Creates a channel with a client-supplied UUID.
///
/// Returns `(record, true)` if newly created, `(record, false)` if already exists.
#[allow(clippy::too_many_arguments)]
#[datastore_span(name = "create_channel_with_id", system = "postgresql")]
pub async fn create_channel_with_id(
&self,
community_id: CommunityId,
channel_id: Uuid,
name: &str,
channel_type: channel::ChannelType,
visibility: channel::ChannelVisibility,
description: Option<&str>,
created_by: &[u8],
ttl_seconds: Option<i32>,
) -> Result<(channel::ChannelRecord, bool)> {
channel::create_channel_with_id(
&self.pool,
community_id,
channel_id,
name,
channel_type,
visibility,
description,
created_by,
ttl_seconds,
)
.await
}
/// Fetches a channel record by ID.
#[datastore_span(name = "get_channel", system = "postgresql")]
pub async fn get_channel(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<channel::ChannelRecord> {
channel::get_channel(&self.pool, community_id, channel_id).await
}
/// Returns the canvas content for a channel, if any.
#[datastore_span(name = "get_canvas", system = "postgresql")]
pub async fn get_canvas(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<Option<String>> {
channel::get_canvas(&self.pool, community_id, channel_id).await
}
/// Sets or clears the canvas content for a channel.
#[datastore_span(name = "set_canvas", system = "postgresql")]
pub async fn set_canvas(
&self,
community_id: CommunityId,
channel_id: Uuid,
canvas: Option<&str>,
) -> Result<()> {
channel::set_canvas(&self.pool, community_id, channel_id, canvas).await
}
/// Verify the mixed-version channel-roster database fence end to end.
#[datastore_span(name = "verify_channel_roster_fence", system = "postgresql")]
pub async fn verify_channel_roster_fence(&self) -> Result<()> {
channel::verify_channel_roster_fence_catalog(&self.pool).await?;
channel::verify_channel_roster_fence_behavior(&self.pool).await
}
/// Capture the active roster while holding the membership-writer lock.
#[datastore_span(name = "lock_member_snapshot", system = "postgresql")]
pub async fn lock_member_snapshot(
&self,
community_id: CommunityId,
channel_id: Uuid,
relay_pubkey: &[u8],
) -> Result<channel::LockedMemberSnapshot> {
channel::lock_member_snapshot(&self.pool, community_id, channel_id, relay_pubkey).await
}
/// Adds a member to a channel.
#[datastore_span(name = "add_member", system = "postgresql")]
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,
community_id,
channel_id,
pubkey,
role,
invited_by,
)
.await
}
/// Removes a member from a channel.
#[datastore_span(name = "remove_member", system = "postgresql")]
pub async fn remove_member(
&self,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
actor_pubkey: &[u8],
) -> Result<()> {
channel::remove_member(&self.pool, community_id, channel_id, pubkey, actor_pubkey).await
}
/// Returns `true` if the pubkey is an active member.
#[datastore_span(name = "is_member", system = "postgresql")]
pub async fn is_member(
&self,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<bool> {
channel::is_member(&self.pool, community_id, channel_id, pubkey).await
}
/// Return the active (channel, pubkey) membership pairs among the given
/// sets, in one statement.
#[datastore_span(name = "membership_pairs", system = "postgresql")]
pub async fn membership_pairs(
&self,
community_id: CommunityId,
channel_ids: &[Uuid],
pubkeys: &[Vec<u8>],
) -> Result<Vec<(Uuid, Vec<u8>)>> {
channel::membership_pairs(&self.pool, community_id, channel_ids, pubkeys).await
}
/// Returns all active members of a channel.
#[datastore_span(name = "get_members", system = "postgresql")]
pub async fn get_members(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<Vec<channel::MemberRecord>> {
channel::get_members(&self.pool, community_id, channel_id).await
}
/// Returns active members for multiple channels in a single query.
#[datastore_span(name = "get_members_bulk", system = "postgresql")]
pub async fn get_members_bulk(
&self,
community_id: CommunityId,
channel_ids: &[Uuid],
) -> Result<Vec<channel::MemberRecord>> {
channel::get_members_bulk(&self.pool, community_id, channel_ids).await
}
/// Get all channel IDs accessible to a pubkey.
#[datastore_span(name = "get_accessible_channel_ids", system = "postgresql")]
pub async fn get_accessible_channel_ids(
&self,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Vec<Uuid>> {
channel::get_accessible_channel_ids(&self.pool, community_id, pubkey).await
}
/// Returns large active-channel rosters whose relay-authored snapshots differ.
#[datastore_span(
name = "list_large_channel_rosters_needing_reconciliation",
system = "postgresql"
)]
pub async fn list_large_channel_rosters_needing_reconciliation(
&self,
minimum_members: i64,
relay_pubkey: &[u8],
) -> Result<Vec<channel::LargeChannelRoster>> {
channel::list_large_channel_rosters_needing_reconciliation(
&self.pool,
minimum_members,
relay_pubkey,
)
.await
}
/// Lists channels, optionally filtered by visibility.
#[datastore_span(name = "list_channels", system = "postgresql")]
pub async fn list_channels(
&self,
community_id: CommunityId,
visibility: Option<&str>,
) -> Result<Vec<channel::ChannelRecord>> {
channel::list_channels(&self.pool, community_id, visibility).await
}
/// Returns full channel records for all channels a user can access.
#[datastore_span(name = "get_accessible_channels", system = "postgresql")]
pub async fn get_accessible_channels(
&self,
community_id: CommunityId,
pubkey: &[u8],
visibility_filter: Option<&str>,
member_only: Option<bool>,
) -> Result<Vec<channel::AccessibleChannel>> {
channel::get_accessible_channels(
&self.pool,
community_id,
pubkey,
visibility_filter,
member_only,
)
.await
}
/// Returns all bot-role members with their aggregated channel names in one community.
#[datastore_span(name = "get_bot_members", system = "postgresql")]
pub async fn get_bot_members(
&self,
community_id: CommunityId,
) -> Result<Vec<channel::BotMemberRecord>> {
channel::get_bot_members(&self.pool, community_id).await
}
/// Bulk-fetch user records by pubkey.
#[datastore_span(name = "get_users_bulk", system = "postgresql")]
pub async fn get_users_bulk(
&self,
community_id: CommunityId,
pubkeys: &[Vec<u8>],
) -> Result<Vec<channel::UserRecord>> {
channel::get_users_bulk(&self.pool, community_id, pubkeys).await
}
/// Updates a channel's name and/or description.
#[datastore_span(name = "update_channel", system = "postgresql")]
pub async fn update_channel(
&self,
community_id: CommunityId,
channel_id: Uuid,
updates: channel::ChannelUpdate,
) -> Result<channel::ChannelRecord> {
channel::update_channel(&self.pool, community_id, channel_id, updates).await
}
/// Sets the topic for a channel.
#[datastore_span(name = "set_topic", system = "postgresql")]
pub async fn set_topic(
&self,
community_id: CommunityId,
channel_id: Uuid,
topic: &str,
set_by: &[u8],
) -> Result<()> {
channel::set_topic(&self.pool, community_id, channel_id, topic, set_by).await
}
/// Sets the purpose for a channel.
#[datastore_span(name = "set_purpose", system = "postgresql")]
pub async fn set_purpose(
&self,
community_id: CommunityId,
channel_id: Uuid,
purpose: &str,
set_by: &[u8],
) -> Result<()> {
channel::set_purpose(&self.pool, community_id, channel_id, purpose, set_by).await
}
/// Archives a channel.
#[datastore_span(name = "archive_channel", system = "postgresql")]
pub async fn archive_channel(&self, community_id: CommunityId, channel_id: Uuid) -> Result<()> {
channel::archive_channel(&self.pool, community_id, channel_id).await
}
/// Unarchives a channel.
#[datastore_span(name = "unarchive_channel", system = "postgresql")]
pub async fn unarchive_channel(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<()> {
channel::unarchive_channel(&self.pool, community_id, channel_id).await
}
/// Soft-delete a channel.
#[datastore_span(name = "soft_delete_channel", system = "postgresql")]
pub async fn soft_delete_channel(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<bool> {
channel::soft_delete_channel(&self.pool, community_id, channel_id).await
}
/// Returns the count of active members in a channel.
#[datastore_span(name = "get_member_count", system = "postgresql")]
pub async fn get_member_count(
&self,
community_id: CommunityId,
channel_id: Uuid,
) -> Result<i64> {
channel::get_member_count(&self.pool, community_id, channel_id).await
}
/// Bulk-fetch member counts for a set of channel IDs.
#[datastore_span(name = "get_member_counts_bulk", system = "postgresql")]
pub async fn get_member_counts_bulk(
&self,
community_id: CommunityId,
channel_ids: &[Uuid],
) -> Result<std::collections::HashMap<Uuid, i64>> {
channel::get_member_counts_bulk(&self.pool, community_id, channel_ids).await
}
/// Get the active role of a pubkey in a channel.
#[datastore_span(name = "get_member_role", system = "postgresql")]
pub async fn get_member_role(
&self,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> Result<Option<String>> {
channel::get_member_role(&self.pool, community_id, channel_id, pubkey).await
}
/// Archive ephemeral channels whose TTL deadline has passed.
#[datastore_span(name = "reap_expired_ephemeral_channels", system = "postgresql")]
pub async fn reap_expired_ephemeral_channels(
&self,
) -> Result<Vec<channel::ReapedEphemeralChannel>> {
channel::reap_expired_ephemeral_channels(&self.pool).await
}
/// Query due reminders ready for delivery.
#[datastore_span(name = "query_due_reminders", system = "postgresql")]
pub async fn query_due_reminders(
@@ -4625,207 +4270,6 @@ mod tests {
id
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn unmigrated_roster_fence_blocks_startup_until_0032_is_applied() {
let admin = PgPool::connect(&admin_url().await)
.await
.expect("connect admin");
let (pool, scratch_name) =
create_scratch_db_through(&admin, "roster_fence_unmigrated", Some(31)).await;
let db = Db::from_pool(pool.clone());
let error = db
.verify_channel_roster_fence()
.await
.expect_err("pre-0032 schema must block roster publishers");
assert!(
error.to_string().contains("channel roster fence trigger"),
"startup gate must report the missing schema fence: {error}"
);
let rows_before: i64 = sqlx::query_scalar("SELECT count(*) FROM events WHERE kind = 39002")
.fetch_one(&pool)
.await
.expect("count pre-migration rosters");
assert_eq!(
rows_before, 0,
"failed startup gate must not publish a roster"
);
migration::run_migrations(&pool)
.await
.expect("apply migration 0032");
db.verify_channel_roster_fence()
.await
.expect("0032 must open the startup gate");
drop_scratch_db(&admin, pool, &scratch_name).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn channel_roster_fence_behavior_verification_detects_inert_function() {
let admin = PgPool::connect(&admin_url().await)
.await
.expect("connect admin");
let (pool, scratch_name) = create_scratch_db(&admin, "roster_fence_inert").await;
let db = Db::from_pool(pool.clone());
sqlx::raw_sql(
"CREATE OR REPLACE FUNCTION guard_channel_roster_snapshot() \
RETURNS TRIGGER AS $$ BEGIN RETURN NEW; END; $$ LANGUAGE plpgsql;",
)
.execute(&pool)
.await
.expect("replace roster fence with inert body");
let error = db
.verify_channel_roster_fence()
.await
.expect_err("inert roster fence must fail closed");
assert!(
error
.to_string()
.contains("stale probe roster was accepted"),
"behavior probe must identify inert semantics: {error}"
);
drop_scratch_db(&admin, pool, &scratch_name).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn channel_roster_fence_catalog_verification_fails_closed() {
let admin = PgPool::connect(&admin_url().await)
.await
.expect("connect admin");
let (pool, scratch_name) = create_scratch_db(&admin, "roster_fence_catalog").await;
let db = Db::from_pool(pool.clone());
db.verify_channel_roster_fence()
.await
.expect("migrated roster fence must verify");
let child: String = sqlx::query_scalar(
"SELECT n.nspname || '.' || c.relname \
FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid \
JOIN pg_namespace n ON n.oid = c.relnamespace \
WHERE i.inhparent = 'public.events'::regclass ORDER BY i.inhrelid LIMIT 1",
)
.fetch_one(&pool)
.await
.expect("load event partition");
sqlx::query(sqlx::AssertSqlSafe(format!(
"ALTER TABLE {child} DISABLE TRIGGER trg_events_guard_channel_roster_snapshot"
)))
.execute(&pool)
.await
.expect("disable partition roster trigger");
let error = db
.verify_channel_roster_fence()
.await
.expect_err("disabled partition roster fence must fail closed");
assert!(
error.to_string().contains(&child),
"verification must identify the unfenced partition: {error}"
);
drop_scratch_db(&admin, pool, &scratch_name).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn desired_schema_rejects_stale_legacy_roster_role() {
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
let admin = PgPool::connect(&admin_url().await)
.await
.expect("connect admin");
let scratch_name = format!("schema_roster_role_{}", Uuid::new_v4().simple());
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE DATABASE {scratch_name}"
)))
.execute(&admin)
.await
.expect("create desired-schema scratch db");
let base_url = admin_url().await;
let slash = base_url.rfind('/').expect("database URL has path segment");
let scratch_url = format!("{}/{}", &base_url[..slash], scratch_name);
let pool = PgPoolOptions::new()
.max_connections(1)
.connect(&scratch_url)
.await
.expect("connect desired-schema scratch db");
sqlx::raw_sql(include_str!("../../../schema/schema.sql"))
.execute(&pool)
.await
.expect("apply desired-state schema");
let db = Db::from_pool(pool.clone());
let community_uuid = Uuid::new_v4();
let community = CommunityId::from_uuid(community_uuid);
let channel = Uuid::new_v4();
let relay_keys = Keys::generate();
let owner_keys = Keys::generate();
let owner = owner_keys.public_key().to_bytes();
seed_community_channel(&pool, community_uuid, channel, &owner_keys).await;
let member = Keys::generate().public_key().to_bytes();
sqlx::query(
"INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) \
VALUES ($1, $2, $3, 'admin', $4)",
)
.bind(community_uuid)
.bind(channel)
.bind(member.as_slice())
.bind(owner.as_slice())
.execute(&pool)
.await
.expect("seed canonical admin");
let roster = |role: &str, timestamp| {
EventBuilder::new(Kind::Custom(39002), "")
.tags(vec![
Tag::parse(["d", channel.to_string().as_str()]).expect("d tag"),
Tag::parse(["p", hex::encode(owner).as_str(), "", "owner"])
.expect("owner p tag"),
Tag::parse(["p", hex::encode(member).as_str(), "", role])
.expect("member p tag"),
])
.custom_created_at(Timestamp::from(timestamp))
.sign_with_keys(&relay_keys)
.expect("sign roster")
};
let base = Timestamp::now().as_secs();
let fresh = roster("admin", base);
assert!(
db.replace_addressable_event(community, &fresh, Some(channel))
.await
.expect("publish canonical role")
.1
);
let stale = roster("member", base + 1);
let error = db
.replace_addressable_event(community, &stale, Some(channel))
.await
.expect_err("desired-state fence must reject stale role");
assert!(matches!(
error,
DbError::Sqlx(sqlx::Error::Database(ref db_error))
if db_error.code().as_deref() == Some("23514")
));
let live_id: Vec<u8> = sqlx::query_scalar(
"SELECT id FROM events WHERE community_id=$1 AND channel_id=$2 \
AND kind=39002 AND deleted_at IS NULL",
)
.bind(community_uuid)
.bind(channel)
.fetch_one(&pool)
.await
.expect("load desired-state live roster");
assert_eq!(live_id, fresh.id.as_bytes().to_vec());
drop_scratch_db(&admin, pool, &scratch_name).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn coordinate_delete_spares_head_newer_than_the_deletion() {