jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(db): split channel membership store (#6782)

Browse Source 
## Current reconstructed head

Exact base: `codex/issue-7-roster-test-fixtures` at
`5df440c411be9705eb29a57f0c41f7239767e007`
Exact head: `codex/issue-7-channel-membership-store` at
`8ad0782ee311f5f51b714494ce750c5937f127cc`

This current head removes `crates/buzz-db/tests/store_ownership.rs`; no
replacement path-sensitive ownership test is introduced. Apart from
removing that complete test-file diff, the production patch is
byte-for-byte identical to the previously reviewed slice. This remains
part of tracker #2 and the #17/#19 acceptance work.

Independent exact-head review from a separate clean Blox workstation
found no issues. Current-head evidence passed formatting, strict
`buzz-db` clippy, 111 non-PostgreSQL library tests with 200 PostgreSQL
tests ignored, the observability source test, relay consumer
compilation, exact ownership/unique-span review checks, and 3 channel
and 19 membership PostgreSQL tests on native PostgreSQL where
applicable.

## Why
Complete the channel ownership slice of [tracker
#2](https://github.com/TheSentinel454/buzz/issues/2) and [domain issue
#7](https://github.com/TheSentinel454/buzz/issues/7) while preserving
the runtime/store boundary established by #6660 and #6668. This child
stacks on the test-only fixture prerequisite #6819 above #6777 and
carries forward PR #6700's membership/replacement lock timing without
changing lock or transaction behavior.

## What
- Keep channel lifecycle, metadata, TTL advisory locking, and lifecycle
tests in `channel.rs`
- Move membership/roster records, SQL, advisory-lock helpers, `Db`
methods, focused tests, and datastore spans to a dedicated
`channel_members.rs`
- Preserve existing `buzz_db::channel::*` paths with compatibility
re-exports while exposing the dedicated module
- Move the four roster-fence PostgreSQL tests out of `lib.rs`

## Stack
- Exact base: codex/issue-7-roster-test-fixtures at
21d1b265c133292e6707e766cd4204e6a43f08af
([#6819](https://github.com/block/buzz/pull/6819))
- Exact head: codex/issue-7-channel-membership-store at
25138bfd6588e046170dbdbc4ed953bdc3cf7ed1
- Tracker: https://github.com/TheSentinel454/buzz/issues/2
- Domain: https://github.com/TheSentinel454/buzz/issues/7
- Test/span acceptance: https://github.com/TheSentinel454/buzz/issues/17
and https://github.com/TheSentinel454/buzz/issues/19

## Non-goals
- No SQL, schema, retry, timeout, lock ordering, transaction boundary,
or client-visible behavior changes
- No change to channel TTL lifecycle ownership merely because lifecycle
bootstrap writes an owner membership row
- No store traits, domain-handle redesign, broad `PgExecutor` migration,
raw pool accessor, new crate, or directory-wide reorganization
- No changes to, retargeting of, or merge action on PR #6700 or #6777

## Risk Assessment
Moderate review surface, low semantic risk. The file split is large, but
method signatures, SQL, bind order, membership and replacement lock
namespaces, transaction boundaries, and span names remain unchanged.
Compatibility re-exports preserve existing `buzz_db::channel::*`
consumers.

## Blox Verification
Author workstation: `buzz-tornquist-issue-2-store-stack` (`2046520`),
exact head `8376e19d0da3ec77550590cd91cc3dfe284d95d6`.

- `cargo fmt --all --check` — passed
- `cargo clippy -p buzz-db -p buzz-relay --all-targets -- -D warnings` —
passed
- Native PostgreSQL channel lifecycle suite — 3 passed
- Native PostgreSQL membership/roster suite — 17 passed; two
pre-existing ignored-test fixture failures reproduced identically on the
untouched parent `2de5444`:
`large_roster_reconciliation_candidates_respect_snapshot_count_and_signer`
and `locked_member_snapshot_blocks_post_capture_membership_mutation`
both receive the migration-0032 `23514` invalid-`p`-tag rejection. This
extraction intentionally does not fold a test-behavior fix into the
move.
- `cargo test -p buzz-relay --lib -- --test-threads=1` — 908 passed, 48
ignored; the existing load-sensitive mesh demo test returned 504,
matching the #6700/parent baseline
- `cargo test -p buzz-relay --lib
api::mesh_demo::tests::demo_join_forwarded_arm_round_trips_echo --
--exact --test-threads=1` — passed

Independent exact-head review: `buzz-tornquist-pr-6782-review`
(`2048397`) found no remaining critical, important, or minor issues. The
full implementation review also independently reproduced both stated
PostgreSQL fixture failures on the exact parent and passed the relay
library suite (909 passed, 48 ignored).

Generated with Codex

## Superseded pre-comment restack verification

PR #6700 merged before publication completed. This layer was restacked
onto current main through the exact parent named above; the final
cumulative tip is 2ddcc8a29f95c8c8c9cb87bb6cf59335f2190fae. Cumulative
author gates passed: formatting and diff checks; buzz-db and buzz-relay
all-target clippy with -D warnings; DB lib 111 passed / 200 ignored;
ownership 22/22; observability 1/1; the full isolated PostgreSQL domain
matrix; and relay lib 910 passed / 49 ignored.

- Workstation: `buzz-tornquist-pr-6782-final-review` (`2057620`), fresh
shallow checkout
- Base: `c60e793eadde79d9eab9f48bbb2ede0ad4831f9b`
- Head: `fa09b6c81c4db3b3e1940a2117a97ab2186e49f7`
- Findings: none

Reviewed both commits in `base..head`. Channel lifecycle/metadata, TTL
transitions, and their lock rationale remain in `channel.rs`; membership
authorization, roster fencing/snapshots, membership advisory locking,
membership records, and focused tests move together to
`channel_members.rs`. SQL, transaction, and lock sequences are
preserved.

Verification: format and diff checks passed; `buzz-db --all-targets`
clippy passed with `-D warnings`; DB lib tests passed (111 passed, 200
PostgreSQL tests ignored); ownership (2/2) and observability (1/1)
guards passed; native PostgreSQL 17 passed 3 channel lifecycle tests
plus 19 membership/roster tests with migrations 1-32 successful; relay
lib test target compiled successfully. Final worktree was detached at
the exact head and clean.

Complete evidence archive SHA-256:
`81ae374095f649ca7a25d8b9a4fc864257b7925d1b44657a69ca111523adf36e`.

## Comment-addressed restack

Review follow-up on #6777 removed only the low-value replaceable
ownership source test. This PR was restacked onto its rewritten parent;
its production patch is unchanged.

- Exact base: `21d1b265c133292e6707e766cd4204e6a43f08af`
- Exact head: `25138bfd6588e046170dbdbc4ed953bdc3cf7ed1`
- Final cumulative tip: `6fa2f104d42c6ba85bdf62e7ccb74ceaf4a84f67`
- Per-layer patch-ID and tree audits confirm this PR’s production diff
is unchanged from its pre-comment head.
- Cumulative Blox gate: formatting and diff checks; strict
`buzz-db`/`buzz-relay` Clippy; DB lib 111 passed / 200 ignored;
ownership 21/21; observability 1/1; every moved PostgreSQL test; relay
lib 910 passed / 49 ignored.
- Independent re-review at this exact head: no findings; fresh
exact-parent/head Blox review passed fmt/diff, strict Clippy, DB lib 111
passed / 200 ignored, current ownership/observability guards, 3 channel
plus 19 membership PostgreSQL tests, and relay compilation.

Signed-off-by: OpenAI Codex <codex@openai.com>
Co-authored-by: OpenAI Codex <codex@openai.com>
This commit is contained in:
Luke Tornquist
2026-08-28 10:55:48 -04:00
committed by GitHub
parent 953a1798b2
commit e76c81968b
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() {