mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
fix(channels): return complete member rosters (#5765)
## Summary - return complete channel rosters instead of truncating at 1,000 members - chunk `event_mentions` inserts inside one transaction so large kind `39002` snapshots remain discoverable by every `p` tag - add a targeted `buzz-admin reconcile-channels --channel <uuid>` force-republish path for stale discovery snapshots - cover a 1,501-member roster, 11,000-tag mention index, and kind `39002` tag construction past member 1,000 ## Why The relay builds NIP-29 discovery and several authorization decisions from `get_members()`, but that helper silently returned only the first 1,000 active members. Desktop then counted the truncated kind `39002` event, while late members could be rejected by roster-scanning member actions. Removing the roster cap exposes PostgreSQL's 65,535 bind-parameter ceiling in mention indexing, so the insert is chunked transactionally to preserve all-or-nothing indexing. The existing reconcilers only fill missing discovery events. The targeted admin option bypasses the separately known 1,000-channel reconciliation-list ceiling and replaces an existing channel snapshot using the configured production relay key. ## Attribution This supersedes and builds on #3166 by @LordMelkor. Thank you for identifying the roster boundary and contributing the original complete-roster and mention-index patch. The production roster/query changes and the two PostgreSQL regressions retain that work's shape; this PR rebases it onto current `main`, adds relay coverage, and adds the targeted repair operation requested for rollout. ## Validation Exact pushed head: `24d02e4f3824150ed84913c9d230e675502e5b12` - `cargo check -p buzz-db -p buzz-admin` - `cargo test -p buzz-db channel::tests::get_members_returns_full_roster_beyond_1000 -- --ignored --exact --nocapture` - `cargo test -p buzz-db feed::tests::insert_mentions_indexes_rosters_past_bind_parameter_cap -- --ignored --exact --nocapture` - `cargo test -p buzz-relay --lib handlers::side_effects::tests::group_members_snapshot_keeps_members_past_one_thousand -- --exact` - `cargo run -q -p buzz-admin -- reconcile-channels --help` - mandatory pre-push hook: branch-skew, desktop checks/typecheck/tests, mobile tests, Rust tests, and desktop Tauri checks all passed on the pushed head ## Rollout 1. Deploy the relay/backend build. 2. Run `buzz-admin reconcile-channels --channel <general-channel-uuid>` with `BUZZ_RELAY_PRIVATE_KEY` configured. 3. Verify the replacement kind `39002` roster count matches the active database membership count. No schema migration or desktop release is required. Fixes #3156 Supersedes #3166 --------- Signed-off-by: Wes <wesbillman@users.noreply.github.com> Co-authored-by: Carl <c7ebe626f000404285d3686e1dc74cc07cc60a9754a150041ba132e14bd3e2ec@buzz.block.builderlab.xyz>
This commit is contained in:
Wes
1
Cargo.lock
generated
1
Cargo.lock
generated
@@ -881,6 +881,7 @@ dependencies = [
|
||||
"tokio",
|
||||
"tracing",
|
||||
"url",
|
||||
"uuid",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
44
TESTING.md
44
TESTING.md
@@ -155,9 +155,49 @@ buzz messages thread --channel "$CHANNEL" --event "$EVENT_ID" | jq .
|
||||
|
||||
A successful run prints `{"event_id":"…","accepted":true,"message":""}` for
|
||||
the send, and the message body in the `get` output. `thread` returns `[]`
|
||||
for a leaf message — populated only after a reply comes in (see §5).
|
||||
for a leaf message — populated only after a reply comes in (see §6).
|
||||
|
||||
### 5. Going deeper
|
||||
### 5. Verify a roster beyond 1,000 members
|
||||
|
||||
Use the focused live-relay script when changing channel membership, discovery,
|
||||
or reconciliation. It proves the three boundaries that DB-only tests cannot:
|
||||
the relay-served kind 39002 includes a member at roster position 1,501, that
|
||||
identity can publish a channel message, and targeted reconciliation preserves
|
||||
its discoverability.
|
||||
|
||||
Run this only against an isolated local database. The script inserts fixture
|
||||
members directly, then drives discovery and messaging through the release CLI
|
||||
and relay. Keep the release relay from step 3 running and use its configured
|
||||
relay key for authoritative replacement:
|
||||
|
||||
```bash
|
||||
export PATH="$PWD/target/release:$PATH"
|
||||
export DATABASE_URL="postgres://buzz:buzz_dev@localhost:5432/buzz_roster_e2e"
|
||||
export BUZZ_RELAY_URL="http://localhost:3030" # match the relay from step 3
|
||||
export RELAY_URL="ws://localhost:3030"
|
||||
export BUZZ_RELAY_PRIVATE_KEY="<same key used by buzz-relay>"
|
||||
|
||||
scripts/e2e-large-channel-roster.sh
|
||||
```
|
||||
|
||||
Success is directly observable as four `PASS` lines. The first and fourth
|
||||
include a member count greater than 1,000 and the same late-member pubkey; the
|
||||
second includes the accepted kind 9 event ID, and the third proves targeted
|
||||
repair left kind 39000/39001 IDs and tags unchanged:
|
||||
|
||||
```text
|
||||
PASS discovery-before-republish channel=<uuid> members=1502 late_pubkey=<hex>
|
||||
PASS late-member-action event_id=<hex>
|
||||
PASS targeted-repair-preserves-metadata-and-admin-events channel=<uuid>
|
||||
PASS discovery-after-republish channel=<uuid> members=1502 late_pubkey=<hex>
|
||||
```
|
||||
|
||||
The script refuses debug binaries and refuses a `buzz` or `buzz-admin` resolved
|
||||
outside this checkout's `target/release`. It also requires the targeted admin
|
||||
operation to use `BUZZ_RELAY_PRIVATE_KEY`; never substitute an ephemeral signer
|
||||
for an authoritative replacement.
|
||||
|
||||
### 6. Going deeper
|
||||
|
||||
For full coverage of every CLI command (54 subcommands across 12 groups),
|
||||
follow [`crates/buzz-cli/TESTING.md`](crates/buzz-cli/TESTING.md).
|
||||
|
||||
@@ -35,4 +35,5 @@ rustls = { version = "0.23", default-features = false, features = ["ring", "std"
|
||||
tracing = { workspace = true }
|
||||
sqlx = { workspace = true }
|
||||
url = { workspace = true }
|
||||
uuid = { workspace = true }
|
||||
clap = { version = "4", features = ["derive"] }
|
||||
|
||||
@@ -88,12 +88,17 @@ enum Command {
|
||||
#[command(subcommand)]
|
||||
command: deletions::DeletionsCommand,
|
||||
},
|
||||
/// Emit kind:39000/39002 events for channels missing them.
|
||||
/// Emit missing kind:39000/39001/39002 channel discovery events, or
|
||||
/// republish only a targeted channel's kind:39002 roster.
|
||||
///
|
||||
/// Channels created via direct SQL (seed scripts, pre-migration data) won't
|
||||
/// have Nostr discovery events. This command creates them so pure-nostr
|
||||
/// clients can see those channels. Idempotent — safe to run multiple times.
|
||||
/// Without `--channel`, only channels missing discovery metadata are
|
||||
/// reconciled. With `--channel`, only that channel's member snapshot is
|
||||
/// replaced; canonical metadata and admin events remain untouched.
|
||||
ReconcileChannels {
|
||||
/// Optional channel UUID to force-republish.
|
||||
#[arg(long)]
|
||||
channel: Option<String>,
|
||||
|
||||
/// Relay private key (hex) for signing events. Falls back to
|
||||
/// BUZZ_RELAY_PRIVATE_KEY env var. If neither is set, generates
|
||||
/// an ephemeral key (events will be unverifiable after restart).
|
||||
@@ -156,8 +161,8 @@ async fn run(cli: Cli) -> Result<i32> {
|
||||
command: ProductFeedbackCommand::List { limit },
|
||||
} => cmd_list_product_feedback(limit).await,
|
||||
Command::Deletions { command } => deletions::run(command).await,
|
||||
Command::ReconcileChannels { relay_key } => {
|
||||
reconcile_channels(relay_key).await?;
|
||||
Command::ReconcileChannels { channel, relay_key } => {
|
||||
reconcile_channels(channel, relay_key).await?;
|
||||
Ok(0)
|
||||
}
|
||||
}
|
||||
@@ -466,14 +471,26 @@ async fn resolve_admin_tenant(db: &Db) -> Result<TenantContext> {
|
||||
Ok(TenantContext::resolved(record.id, record.host))
|
||||
}
|
||||
|
||||
async fn reconcile_channels(relay_key_arg: Option<String>) -> Result<()> {
|
||||
async fn reconcile_channels(
|
||||
channel_arg: Option<String>,
|
||||
relay_key_arg: Option<String>,
|
||||
) -> Result<()> {
|
||||
use buzz_core::kind::KIND_NIP29_GROUP_ADMINS;
|
||||
use buzz_db::event::EventQuery;
|
||||
|
||||
let db = connect_db().await?;
|
||||
|
||||
// Resolve relay signing key: arg > env > ephemeral
|
||||
let relay_keys = match relay_key_arg.or_else(|| std::env::var("BUZZ_RELAY_PRIVATE_KEY").ok()) {
|
||||
// Resolve relay signing key: arg > env > ephemeral. Force-republish must
|
||||
// never use an ephemeral key because it replaces an existing authoritative
|
||||
// snapshot.
|
||||
let configured_relay_key =
|
||||
relay_key_arg.or_else(|| std::env::var("BUZZ_RELAY_PRIVATE_KEY").ok());
|
||||
if channel_arg.is_some() && configured_relay_key.is_none() {
|
||||
return Err(anyhow::anyhow!(
|
||||
"--channel requires --relay-key or BUZZ_RELAY_PRIVATE_KEY"
|
||||
));
|
||||
}
|
||||
let relay_keys = match configured_relay_key {
|
||||
Some(key_hex) => {
|
||||
Keys::parse(&key_hex).map_err(|e| anyhow::anyhow!("invalid relay key: {e}"))?
|
||||
}
|
||||
@@ -490,7 +507,21 @@ async fn reconcile_channels(relay_key_arg: Option<String>) -> Result<()> {
|
||||
};
|
||||
|
||||
let tenant = resolve_admin_tenant(&db).await?;
|
||||
let channels = db.list_channels(tenant.community(), None).await?;
|
||||
let target_channel = channel_arg
|
||||
.as_deref()
|
||||
.map(uuid::Uuid::parse_str)
|
||||
.transpose()
|
||||
.map_err(|e| anyhow::anyhow!("invalid --channel UUID: {e}"))?;
|
||||
let channels = if let Some(target) = target_channel {
|
||||
vec![db
|
||||
.get_channel(tenant.community(), target)
|
||||
.await
|
||||
.map_err(|_| {
|
||||
anyhow::anyhow!("channel {target} not found in community {}", tenant.host())
|
||||
})?]
|
||||
} else {
|
||||
db.list_channels(tenant.community(), None).await?
|
||||
};
|
||||
if channels.is_empty() {
|
||||
println!("No channels in database.");
|
||||
return Ok(());
|
||||
@@ -513,57 +544,64 @@ async fn reconcile_channels(relay_key_arg: Option<String>) -> Result<()> {
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
|
||||
if !existing.is_empty() {
|
||||
if !existing.is_empty() && target_channel.is_none() {
|
||||
skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
let members = db.get_members(tenant.community(), channel.id).await?;
|
||||
|
||||
// kind:39000 — channel metadata
|
||||
{
|
||||
let mut tags: Vec<Tag> = vec![Tag::parse(["d", &channel_id_str])?];
|
||||
tags.push(Tag::parse(["name", &channel.name])?);
|
||||
if let Some(ref desc) = channel.description {
|
||||
if !desc.is_empty() {
|
||||
tags.push(Tag::parse(["about", desc])?);
|
||||
}
|
||||
}
|
||||
if channel.visibility == "private" {
|
||||
tags.push(Tag::parse(["private"])?);
|
||||
} else {
|
||||
tags.push(Tag::parse(["public"])?);
|
||||
}
|
||||
if channel.channel_type == "dm" {
|
||||
tags.push(Tag::parse(["hidden"])?);
|
||||
}
|
||||
tags.push(Tag::parse(["closed"])?);
|
||||
tags.push(Tag::parse(["t", &channel.channel_type])?);
|
||||
|
||||
let event = EventBuilder::new(Kind::Custom(39000), "")
|
||||
.tags(tags)
|
||||
.sign_with_keys(&relay_keys)
|
||||
.map_err(|e| anyhow::anyhow!("sign kind:39000: {e}"))?;
|
||||
db.replace_addressable_event(tenant.community(), &event, Some(channel.id))
|
||||
.await?;
|
||||
}
|
||||
|
||||
// kind:39001 — admins
|
||||
{
|
||||
let mut tags: Vec<Tag> = vec![Tag::parse(["d", &channel_id_str])?];
|
||||
for m in members
|
||||
.iter()
|
||||
.filter(|m| m.role == "owner" || m.role == "admin")
|
||||
// A targeted repair is deliberately roster-only. kind:39000 metadata
|
||||
// is richer than this legacy backfill builder, and kind:39001 is not
|
||||
// part of the stale-roster incident; replacing either can destroy
|
||||
// canonical state. Full backfill still creates all three event kinds
|
||||
// for channels with no discovery metadata.
|
||||
if target_channel.is_none() {
|
||||
// kind:39000 — channel metadata
|
||||
{
|
||||
let pk = hex::encode(&m.pubkey);
|
||||
tags.push(Tag::parse(["p", &pk, &m.role])?);
|
||||
let mut tags: Vec<Tag> = vec![Tag::parse(["d", &channel_id_str])?];
|
||||
tags.push(Tag::parse(["name", &channel.name])?);
|
||||
if let Some(ref desc) = channel.description {
|
||||
if !desc.is_empty() {
|
||||
tags.push(Tag::parse(["about", desc])?);
|
||||
}
|
||||
}
|
||||
if channel.visibility == "private" {
|
||||
tags.push(Tag::parse(["private"])?);
|
||||
} else {
|
||||
tags.push(Tag::parse(["public"])?);
|
||||
}
|
||||
if channel.channel_type == "dm" {
|
||||
tags.push(Tag::parse(["hidden"])?);
|
||||
}
|
||||
tags.push(Tag::parse(["closed"])?);
|
||||
tags.push(Tag::parse(["t", &channel.channel_type])?);
|
||||
|
||||
let event = EventBuilder::new(Kind::Custom(39000), "")
|
||||
.tags(tags)
|
||||
.sign_with_keys(&relay_keys)
|
||||
.map_err(|e| anyhow::anyhow!("sign kind:39000: {e}"))?;
|
||||
db.replace_addressable_event(tenant.community(), &event, Some(channel.id))
|
||||
.await?;
|
||||
}
|
||||
|
||||
// kind:39001 — admins
|
||||
{
|
||||
let mut tags: Vec<Tag> = vec![Tag::parse(["d", &channel_id_str])?];
|
||||
for m in members
|
||||
.iter()
|
||||
.filter(|m| m.role == "owner" || m.role == "admin")
|
||||
{
|
||||
let pk = hex::encode(&m.pubkey);
|
||||
tags.push(Tag::parse(["p", &pk, &m.role])?);
|
||||
}
|
||||
let event = EventBuilder::new(Kind::Custom(KIND_NIP29_GROUP_ADMINS as u16), "")
|
||||
.tags(tags)
|
||||
.sign_with_keys(&relay_keys)
|
||||
.map_err(|e| anyhow::anyhow!("sign kind:39001: {e}"))?;
|
||||
db.replace_addressable_event(tenant.community(), &event, Some(channel.id))
|
||||
.await?;
|
||||
}
|
||||
let event = EventBuilder::new(Kind::Custom(KIND_NIP29_GROUP_ADMINS as u16), "")
|
||||
.tags(tags)
|
||||
.sign_with_keys(&relay_keys)
|
||||
.map_err(|e| anyhow::anyhow!("sign kind:39001: {e}"))?;
|
||||
db.replace_addressable_event(tenant.community(), &event, Some(channel.id))
|
||||
.await?;
|
||||
}
|
||||
|
||||
// kind:39002 — members
|
||||
|
||||
@@ -687,7 +687,12 @@ pub async fn membership_pairs(
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Returns all active members of the given channel.
|
||||
/// Returns all active members of the given channel, ordered by `joined_at`.
|
||||
///
|
||||
/// The roster is returned in full and is never truncated: callers use it to
|
||||
/// build the kind 39002 (NIP-29 group members) snapshot and to resolve actor
|
||||
/// roles for admin-event authorization, so a partial list silently hides late
|
||||
/// joiners from channel discovery and makes them read as non-members.
|
||||
///
|
||||
/// Returns an empty list if the channel has been soft-deleted.
|
||||
pub async fn get_members(
|
||||
@@ -702,7 +707,6 @@ pub async fn get_members(
|
||||
JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL
|
||||
WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.removed_at IS NULL
|
||||
ORDER BY cm.joined_at ASC
|
||||
LIMIT 1000
|
||||
"#,
|
||||
)
|
||||
.bind(community_id.as_uuid())
|
||||
@@ -1532,7 +1536,7 @@ mod tests {
|
||||
use crate::user::{ensure_user, set_agent_owner};
|
||||
use nostr::Keys;
|
||||
|
||||
const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz";
|
||||
const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials
|
||||
|
||||
async fn setup_pool() -> PgPool {
|
||||
PgPool::connect(TEST_DB_URL)
|
||||
@@ -1933,6 +1937,85 @@ mod tests {
|
||||
assert_eq!(channel_ids.len(), channel_count as usize);
|
||||
}
|
||||
|
||||
/// `get_members` must return the complete roster, not a truncated prefix.
|
||||
///
|
||||
/// The relay builds the kind 39002 (NIP-29 group members) snapshot and every
|
||||
/// admin role lookup from this list, so a cap silently hides late joiners:
|
||||
/// their clients never discover the channel, and an owner past the cutoff
|
||||
/// reads as a non-member.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn get_members_returns_full_roster_beyond_1000() {
|
||||
let database_url =
|
||||
std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.to_string());
|
||||
let pool = PgPool::connect(&database_url)
|
||||
.await
|
||||
.expect("connect to test DB");
|
||||
let community_id = make_test_community(&pool).await;
|
||||
let community = CommunityId::from_uuid(community_id);
|
||||
let creator = random_pubkey();
|
||||
|
||||
// create_test_channel also inserts the creator as the first (owner) member.
|
||||
let channel = create_test_channel(
|
||||
&pool,
|
||||
community_id,
|
||||
"high-volume-roster",
|
||||
ChannelType::Stream,
|
||||
ChannelVisibility::Open,
|
||||
None,
|
||||
&creator,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("create test channel");
|
||||
|
||||
// Bulk-insert additional members with strictly increasing `joined_at`, so
|
||||
// member N lands at roster position N (the creator holds position 0).
|
||||
// The final member is an owner joining well past the old 1000-row cutoff.
|
||||
let extra_members = 1_500;
|
||||
sqlx::query(
|
||||
r#"
|
||||
INSERT INTO channel_members (community_id, channel_id, pubkey, role, joined_at)
|
||||
SELECT
|
||||
$1,
|
||||
$2,
|
||||
decode(lpad(to_hex(n), 64, '0'), 'hex'),
|
||||
(CASE WHEN n = $3 THEN 'owner' ELSE 'member' END)::member_role,
|
||||
NOW() + (n || ' seconds')::interval
|
||||
FROM generate_series(1, $3) n
|
||||
"#,
|
||||
)
|
||||
.bind(community_id)
|
||||
.bind(channel.id)
|
||||
.bind(extra_members)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.expect("insert high-volume channel members");
|
||||
|
||||
let members = get_members(&pool, community, channel.id)
|
||||
.await
|
||||
.expect("load channel members");
|
||||
|
||||
assert_eq!(
|
||||
members.len(),
|
||||
extra_members as usize + 1,
|
||||
"get_members truncated the roster"
|
||||
);
|
||||
|
||||
// The last joiner sits at the final roster position — past any
|
||||
// 1000-row cap — which also pins the documented `joined_at` ordering.
|
||||
let late_owner = hex::decode(format!("{:064x}", extra_members)).expect("hex pubkey");
|
||||
let late = members.last().expect("roster is non-empty");
|
||||
assert_eq!(
|
||||
late.pubkey, late_owner,
|
||||
"member who joined after the 1000th must be present and ordered last"
|
||||
);
|
||||
assert_eq!(
|
||||
late.role, "owner",
|
||||
"role of a late-joining owner must resolve correctly"
|
||||
);
|
||||
}
|
||||
|
||||
/// A random non-admin, non-owner user cannot remove someone else's bot.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
|
||||
@@ -886,4 +886,40 @@ mod tests {
|
||||
let unique: std::collections::HashSet<Vec<u8>> = byte_seqs.into_iter().collect();
|
||||
assert_eq!(unique.len(), 5, "all channel IDs must be distinct");
|
||||
}
|
||||
|
||||
/// `insert_mentions` must index every p-tag even past Postgres's
|
||||
/// bind-parameter statement cap.
|
||||
///
|
||||
/// Relay-signed kind 39002 member snapshots carry one p-tag per channel
|
||||
/// member, and a multi-row INSERT binds 6 parameters per row — a single
|
||||
/// statement tops out at ~10.9k rows against the 65,535-parameter limit.
|
||||
/// Clients discover their channels via `{kinds:[39002], "#p":[me]}`, so a
|
||||
/// failed insert silently breaks discovery for the whole channel.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn insert_mentions_indexes_rosters_past_bind_parameter_cap() {
|
||||
let pool = setup_pool().await;
|
||||
let community = CommunityId::from_uuid(make_test_community(&pool).await);
|
||||
let channel = insert_test_channel(&pool, community).await;
|
||||
|
||||
// 11,000 rows x 6 binds = 66,000 > 65,535: overflows a single statement.
|
||||
let mention_count = 11_000usize;
|
||||
let tags: Vec<Tag> = (1..=mention_count)
|
||||
.map(|n| Tag::parse(["p", &format!("{n:064x}")]).expect("p tag"))
|
||||
.collect();
|
||||
let event = store_feed_event(&pool, community, 39002, "", Some(channel), tags).await;
|
||||
|
||||
let indexed: i64 = sqlx::query_scalar(
|
||||
"SELECT COUNT(*) FROM event_mentions WHERE community_id = $1 AND event_id = $2",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(event.id.as_bytes().as_slice())
|
||||
.fetch_one(&pool)
|
||||
.await
|
||||
.expect("count indexed mentions");
|
||||
assert_eq!(
|
||||
indexed as usize, mention_count,
|
||||
"every roster p-tag must land in event_mentions"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -104,6 +104,21 @@ pub async fn insert_mentions(
|
||||
community_id: CommunityId,
|
||||
event: &nostr::Event,
|
||||
channel_id: Option<Uuid>,
|
||||
) -> Result<()> {
|
||||
let mut tx = pool.begin().await?;
|
||||
insert_mentions_in_transaction(&mut tx, community_id, event, channel_id).await?;
|
||||
tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Insert mention rows on the caller's transaction. Replacement writes use
|
||||
/// this so the authoritative event and its discovery index commit or roll back
|
||||
/// as one unit.
|
||||
async fn insert_mentions_in_transaction(
|
||||
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
|
||||
community_id: CommunityId,
|
||||
event: &nostr::Event,
|
||||
channel_id: Option<Uuid>,
|
||||
) -> Result<()> {
|
||||
let p_tags: Vec<&str> = event
|
||||
.tags
|
||||
@@ -150,24 +165,31 @@ pub async fn insert_mentions(
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// 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 \
|
||||
(community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) ",
|
||||
);
|
||||
// Multi-row INSERT ... ON CONFLICT DO NOTHING, chunked to stay under
|
||||
// Postgres's 65,535 bind-parameter statement cap (6 binds per row caps a
|
||||
// single statement at ~10.9k rows). Relay-signed kind 39002 rosters carry
|
||||
// one p-tag per channel member and can exceed that. The caller owns the
|
||||
// transaction so all chunks share its commit boundary.
|
||||
const MENTION_INSERT_CHUNK_ROWS: usize = 5_000;
|
||||
for chunk in valid_pubkeys.chunks(MENTION_INSERT_CHUNK_ROWS) {
|
||||
let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
|
||||
"INSERT INTO event_mentions \
|
||||
(community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) ",
|
||||
);
|
||||
|
||||
qb.push_values(&valid_pubkeys, |mut b, pubkey| {
|
||||
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)
|
||||
.push_bind(kind as i32);
|
||||
});
|
||||
qb.push_values(chunk, |mut b, pubkey| {
|
||||
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)
|
||||
.push_bind(kind as i32);
|
||||
});
|
||||
|
||||
qb.push(" ON CONFLICT DO NOTHING");
|
||||
qb.push(" ON CONFLICT DO NOTHING");
|
||||
|
||||
qb.build().execute(pool).await?;
|
||||
qb.build().execute(&mut **tx).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -4916,13 +4938,12 @@ impl Db {
|
||||
));
|
||||
}
|
||||
|
||||
tx.commit().await?;
|
||||
// The replaceable event and its denormalized mention index are one
|
||||
// authoritative discovery write. An indexing error must roll back the
|
||||
// new event and restore the previously-live event.
|
||||
crate::insert_mentions_in_transaction(&mut tx, community_id, event, channel_id).await?;
|
||||
|
||||
// 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, community_id, event, channel_id).await {
|
||||
tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
|
||||
}
|
||||
tx.commit().await?;
|
||||
|
||||
Ok((
|
||||
StoredEvent::with_received_at(event.clone(), received_at, channel_id, true),
|
||||
@@ -5459,6 +5480,87 @@ mod tests {
|
||||
id
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn addressable_replacement_rolls_back_when_mention_indexing_fails() {
|
||||
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
|
||||
|
||||
let admin = PgPool::connect(&admin_url().await)
|
||||
.await
|
||||
.expect("connect admin");
|
||||
let (pool, scratch_name) = create_scratch_db(&admin, "atomic_addressable").await;
|
||||
let db = Db::from_pool(pool.clone());
|
||||
let community_uuid = Uuid::new_v4();
|
||||
let channel = Uuid::new_v4();
|
||||
let keys = Keys::generate();
|
||||
seed_community_channel(&pool, community_uuid, channel, &keys).await;
|
||||
let community = CommunityId::from_uuid(community_uuid);
|
||||
let member = Keys::generate().public_key().to_hex();
|
||||
let tags = || {
|
||||
vec![
|
||||
Tag::parse(["d", channel.to_string().as_str()]).expect("d tag"),
|
||||
Tag::parse(["p", member.as_str(), "", "member"]).expect("p tag"),
|
||||
]
|
||||
};
|
||||
let base = Timestamp::now().as_secs();
|
||||
let old = EventBuilder::new(Kind::Custom(39002), "old")
|
||||
.tags(tags())
|
||||
.custom_created_at(Timestamp::from(base))
|
||||
.sign_with_keys(&keys)
|
||||
.expect("sign old");
|
||||
db.replace_addressable_event(community, &old, Some(channel))
|
||||
.await
|
||||
.expect("insert old roster");
|
||||
|
||||
sqlx::query(
|
||||
"CREATE FUNCTION reject_test_mention() RETURNS trigger AS $$ \
|
||||
BEGIN RAISE EXCEPTION 'injected mention failure'; END; \
|
||||
$$ LANGUAGE plpgsql",
|
||||
)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.expect("create failure function");
|
||||
sqlx::query(
|
||||
"CREATE TRIGGER reject_test_mention BEFORE INSERT ON event_mentions \
|
||||
FOR EACH ROW EXECUTE FUNCTION reject_test_mention()",
|
||||
)
|
||||
.execute(&pool)
|
||||
.await
|
||||
.expect("install failure injection");
|
||||
|
||||
let new = EventBuilder::new(Kind::Custom(39002), "new")
|
||||
.tags(tags())
|
||||
.custom_created_at(Timestamp::from(base + 1))
|
||||
.sign_with_keys(&keys)
|
||||
.expect("sign new");
|
||||
let error = db
|
||||
.replace_addressable_event(community, &new, Some(channel))
|
||||
.await
|
||||
.expect_err("mention failure must fail replacement");
|
||||
assert!(error.to_string().contains("injected mention failure"));
|
||||
|
||||
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.as_uuid())
|
||||
.bind(channel)
|
||||
.fetch_one(&pool)
|
||||
.await
|
||||
.expect("query live roster");
|
||||
assert_eq!(live_id, old.id.as_bytes(), "old roster must remain live");
|
||||
let new_rows: i64 =
|
||||
sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2")
|
||||
.bind(community.as_uuid())
|
||||
.bind(new.id.as_bytes().as_slice())
|
||||
.fetch_one(&pool)
|
||||
.await
|
||||
.expect("count rolled-back event");
|
||||
assert_eq!(new_rows, 0, "new roster must roll back with its index");
|
||||
|
||||
drop_scratch_db(&admin, pool, &scratch_name).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn nip_rs_replacement_hard_deletes_payload_and_watermark_rejects_replay() {
|
||||
|
||||
@@ -1033,6 +1033,18 @@ async fn emit_addressable_discovery_event(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn group_members_tags(group_id: &str, members: &[MemberRecord]) -> anyhow::Result<Vec<Tag>> {
|
||||
let mut tags: Vec<Tag> = Vec::with_capacity(members.len() + 1);
|
||||
tags.push(Tag::parse(["d", group_id])?);
|
||||
for member in members {
|
||||
let pubkey_hex = hex::encode(&member.pubkey);
|
||||
// NIP-29 convention: ["p", pubkey, relay_url, role]. Empty relay_url
|
||||
// because the canonical relay is implicit (this event is signed by it).
|
||||
tags.push(Tag::parse(["p", &pubkey_hex, "", &member.role])?);
|
||||
}
|
||||
Ok(tags)
|
||||
}
|
||||
|
||||
/// Emit NIP-29 group discovery events (39000, 39001, 39002) signed by the relay keypair.
|
||||
/// Called after group creation, metadata changes, or membership changes.
|
||||
/// Events are stored channel-scoped (`channel_id = Some(...)`) so that existing
|
||||
@@ -1136,13 +1148,7 @@ pub async fn emit_group_discovery_events(
|
||||
}
|
||||
|
||||
{
|
||||
let mut tags: Vec<Tag> = vec![Tag::parse(["d", &group_id])?];
|
||||
for m in &members {
|
||||
let pubkey_hex = hex::encode(&m.pubkey);
|
||||
// NIP-29 convention: ["p", pubkey, relay_url, role]. Empty relay_url
|
||||
// because the canonical relay is implicit (this event is signed by it).
|
||||
tags.push(Tag::parse(["p", &pubkey_hex, "", &m.role])?);
|
||||
}
|
||||
let tags = group_members_tags(&group_id, &members)?;
|
||||
emit_addressable_discovery_event(
|
||||
tenant,
|
||||
state,
|
||||
@@ -3372,6 +3378,33 @@ fn topic_for_subscription(channel_id: Option<Uuid>) -> EventTopic {
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn group_members_snapshot_keeps_members_past_one_thousand() {
|
||||
let channel_id = Uuid::new_v4();
|
||||
let members: Vec<MemberRecord> = (0_u16..1_501)
|
||||
.map(|index| MemberRecord {
|
||||
channel_id,
|
||||
pubkey: vec![(index >> 8) as u8, index as u8],
|
||||
role: if index == 1_500 { "owner" } else { "member" }.to_string(),
|
||||
joined_at: chrono::Utc::now(),
|
||||
invited_by: None,
|
||||
removed_at: None,
|
||||
})
|
||||
.collect();
|
||||
|
||||
let tags = group_members_tags(&channel_id.to_string(), &members).expect("build tags");
|
||||
assert_eq!(tags.len(), 1_502, "d tag plus every member p tag");
|
||||
|
||||
let late_pubkey = hex::encode(&members[1_500].pubkey);
|
||||
assert!(tags.iter().any(|tag| {
|
||||
let fields = tag.as_slice();
|
||||
fields.len() == 4
|
||||
&& fields[0] == "p"
|
||||
&& fields[1] == late_pubkey
|
||||
&& fields[3] == "owner"
|
||||
}));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn delete_tombstone_omits_absent_moderation_metadata() {
|
||||
let content =
|
||||
|
||||
127
scripts/e2e-large-channel-roster.sh
Executable file
127
scripts/e2e-large-channel-roster.sh
Executable file
@@ -0,0 +1,127 @@
|
||||
#!/usr/bin/env bash
|
||||
# Prove a channel member past the historical 1,000-row boundary is present in
|
||||
# relay-served discovery, can write, and survives an authoritative republish.
|
||||
set -euo pipefail
|
||||
|
||||
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
cd "$REPO_ROOT"
|
||||
|
||||
: "${DATABASE_URL:?set DATABASE_URL to the isolated relay database}"
|
||||
: "${BUZZ_RELAY_URL:=http://localhost:3030}"
|
||||
: "${RELAY_URL:=ws://localhost:3030}"
|
||||
: "${BUZZ_RELAY_PRIVATE_KEY:=0000000000000000000000000000000000000000000000000000000000000001}"
|
||||
export BUZZ_RELAY_URL RELAY_URL BUZZ_RELAY_PRIVATE_KEY
|
||||
unset BUZZ_AUTH_TAG
|
||||
|
||||
for binary in buzz buzz-admin; do
|
||||
resolved="$(command -v "$binary" || true)"
|
||||
[[ "$resolved" == "$REPO_ROOT/target/release/$binary" ]] || {
|
||||
echo "error: $binary must resolve to $REPO_ROOT/target/release/$binary (got ${resolved:-not found})" >&2
|
||||
exit 1
|
||||
}
|
||||
done
|
||||
command -v jq >/dev/null || { echo "error: jq is required" >&2; exit 1; }
|
||||
command -v psql >/dev/null || { echo "error: psql is required" >&2; exit 1; }
|
||||
|
||||
key_field() {
|
||||
awk -v label="$1" '$1 == label && $2 == "key:" { print $3 }'
|
||||
}
|
||||
|
||||
OWNER_GEN="$(buzz-admin generate-key)"
|
||||
OWNER_SK="$(printf '%s\n' "$OWNER_GEN" | key_field Secret)"
|
||||
export BUZZ_PRIVATE_KEY="$OWNER_SK"
|
||||
|
||||
CHANNEL="$(buzz channels create \
|
||||
--name "roster-boundary-$$" --type stream --visibility open | jq -er '.channel_id')"
|
||||
|
||||
LATE_GEN="$(buzz-admin generate-key)"
|
||||
LATE_SK="$(printf '%s\n' "$LATE_GEN" | key_field Secret)"
|
||||
LATE_PUBKEY="$(printf '%s\n' "$LATE_GEN" | key_field Public)"
|
||||
|
||||
# The creator is roster position 1. Add 1,499 fixtures followed by the real
|
||||
# late identity at position 1,501. A final API-added member forces the relay to
|
||||
# emit a fresh discovery snapshot through the normal membership-change path.
|
||||
psql "$DATABASE_URL" -v ON_ERROR_STOP=1 \
|
||||
--set=channel="$CHANNEL" --set=late_pubkey="$LATE_PUBKEY" <<'SQL'
|
||||
WITH target AS (
|
||||
SELECT community_id, id
|
||||
FROM channels
|
||||
WHERE id = :'channel'::uuid
|
||||
), fixtures AS (
|
||||
SELECT n, decode(lpad(to_hex(n + 65536), 64, '0'), 'hex') AS pubkey
|
||||
FROM generate_series(1, 1499) AS n
|
||||
)
|
||||
INSERT INTO channel_members (community_id, channel_id, pubkey, role, joined_at)
|
||||
SELECT target.community_id, target.id, fixtures.pubkey, 'member',
|
||||
NOW() + fixtures.n * interval '1 millisecond'
|
||||
FROM target CROSS JOIN fixtures;
|
||||
|
||||
INSERT INTO channel_members (community_id, channel_id, pubkey, role, joined_at)
|
||||
SELECT community_id, id, decode(:'late_pubkey', 'hex'), 'member',
|
||||
NOW() + interval '2 seconds'
|
||||
FROM channels
|
||||
WHERE id = :'channel'::uuid;
|
||||
SQL
|
||||
|
||||
TRIGGER_GEN="$(buzz-admin generate-key)"
|
||||
TRIGGER_PUBKEY="$(printf '%s\n' "$TRIGGER_GEN" | key_field Public)"
|
||||
buzz channels add-member --channel "$CHANNEL" --pubkey "$TRIGGER_PUBKEY" --role member >/dev/null
|
||||
|
||||
BEFORE="$(buzz channels members --channel "$CHANNEL")"
|
||||
BEFORE_COUNT="$(jq 'length' <<<"$BEFORE")"
|
||||
jq -e --arg pk "$LATE_PUBKEY" 'any(.[]; .pubkey == $pk and .role == "member")' \
|
||||
<<<"$BEFORE" >/dev/null
|
||||
(( BEFORE_COUNT > 1000 ))
|
||||
printf 'PASS discovery-before-republish channel=%s members=%s late_pubkey=%s\n' \
|
||||
"$CHANNEL" "$BEFORE_COUNT" "$LATE_PUBKEY"
|
||||
|
||||
export BUZZ_PRIVATE_KEY="$LATE_SK"
|
||||
ACTION="$(buzz messages send --channel "$CHANNEL" --content "member-1501-action")"
|
||||
jq -e '.accepted == true' <<<"$ACTION" >/dev/null
|
||||
ACTION_ID="$(jq -er '.event_id' <<<"$ACTION")"
|
||||
buzz messages get --channel "$CHANNEL" --limit 10 \
|
||||
| jq -e --arg id "$ACTION_ID" 'any(.[]; .id == $id and .content == "member-1501-action")' >/dev/null
|
||||
printf 'PASS late-member-action event_id=%s\n' "$ACTION_ID"
|
||||
|
||||
# Targeted roster repair must not replace canonical metadata or admin events.
|
||||
# Preserve both the event ID and complete tags, including fields this backfill
|
||||
# command does not know how to rebuild (DM participants, topic, TTL, etc.).
|
||||
DISCOVERY_BEFORE="$(psql "$DATABASE_URL" -AtX -v ON_ERROR_STOP=1 \
|
||||
--set=channel="$CHANNEL" <<'SQL'
|
||||
SELECT jsonb_object_agg(kind::text, jsonb_build_object(
|
||||
'id', encode(id, 'hex'),
|
||||
'tags', tags
|
||||
) ORDER BY kind)::text
|
||||
FROM events
|
||||
WHERE channel_id = :'channel'::uuid
|
||||
AND kind IN (39000, 39001)
|
||||
AND deleted_at IS NULL;
|
||||
SQL
|
||||
)"
|
||||
jq -e 'has("39000") and has("39001")' <<<"$DISCOVERY_BEFORE" >/dev/null
|
||||
|
||||
DATABASE_URL="$DATABASE_URL" RELAY_URL="$RELAY_URL" \
|
||||
buzz-admin reconcile-channels --channel "$CHANNEL" >/dev/null
|
||||
|
||||
DISCOVERY_AFTER="$(psql "$DATABASE_URL" -AtX -v ON_ERROR_STOP=1 \
|
||||
--set=channel="$CHANNEL" <<'SQL'
|
||||
SELECT jsonb_object_agg(kind::text, jsonb_build_object(
|
||||
'id', encode(id, 'hex'),
|
||||
'tags', tags
|
||||
) ORDER BY kind)::text
|
||||
FROM events
|
||||
WHERE channel_id = :'channel'::uuid
|
||||
AND kind IN (39000, 39001)
|
||||
AND deleted_at IS NULL;
|
||||
SQL
|
||||
)"
|
||||
[[ "$DISCOVERY_AFTER" == "$DISCOVERY_BEFORE" ]]
|
||||
printf 'PASS targeted-repair-preserves-metadata-and-admin-events channel=%s\n' "$CHANNEL"
|
||||
|
||||
AFTER="$(buzz channels members --channel "$CHANNEL")"
|
||||
AFTER_COUNT="$(jq 'length' <<<"$AFTER")"
|
||||
jq -e --arg pk "$LATE_PUBKEY" 'any(.[]; .pubkey == $pk and .role == "member")' \
|
||||
<<<"$AFTER" >/dev/null
|
||||
(( AFTER_COUNT == BEFORE_COUNT ))
|
||||
printf 'PASS discovery-after-republish channel=%s members=%s late_pubkey=%s\n' \
|
||||
"$CHANNEL" "$AFTER_COUNT" "$LATE_PUBKEY"
|
||||
Reference in New Issue
Block a user