diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 2b0eb25c688..2114e3d561d 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -775,6 +775,63 @@ jobs: ${{ steps.artifacts.outputs.exe }} ${{ steps.artifacts.outputs.sig }} + desktop-release-smoke: + name: Desktop release smoke + if: github.repository == 'block/buzz' + runs-on: ubuntu-latest + needs: setup + timeout-minutes: 20 + permissions: + contents: read + env: + PLAYWRIGHT_BROWSERS_PATH: ${{ github.workspace }}/.cache/ms-playwright + steps: + - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + with: + ref: ${{ needs.setup.outputs.source_sha }} + persist-credentials: false + - uses: cashapp/activate-hermit@cea9af7913204a965fd488637a8d1811bba2e616 # v1 + - name: Install desktop dependencies + run: just desktop-install-ci + - name: Get Playwright version + id: pw-version + run: echo "version=$(cd desktop && node -e \"console.log(require('@playwright/test/package.json').version)\")" >> "$GITHUB_OUTPUT" + - name: Restore Playwright browser cache + id: playwright-cache + uses: actions/cache/restore@caa296126883cff596d87d8935842f9db880ef25 # v5 + with: + path: ${{ env.PLAYWRIGHT_BROWSERS_PATH }} + key: playwright-${{ runner.os }}-${{ steps.pw-version.outputs.version }} + - name: Install Playwright Chromium + if: steps.playwright-cache.outputs.cache-hit != 'true' + run: cd desktop && pnpm exec playwright install chromium + - name: Install Playwright system dependencies + run: cd desktop && pnpm exec playwright install-deps chromium + - name: Save Playwright browser cache + if: steps.playwright-cache.outputs.cache-hit != 'true' + uses: actions/cache/save@caa296126883cff596d87d8935842f9db880ef25 # v5 + with: + path: ${{ env.PLAYWRIGHT_BROWSERS_PATH }} + key: playwright-${{ runner.os }}-${{ steps.pw-version.outputs.version }} + - name: Build test relay + run: cargo build --profile ci -p buzz-relay + - name: Run deterministic correctness smoke + env: + BUZZ_E2E_RELAY_BIN: ${{ github.workspace }}/target/ci/buzz-relay + BUZZ_RELEASE_SMOKE_ARTIFACT_DIR: ${{ github.workspace }}/release-smoke-artifacts + run: just desktop-release-smoke + - name: Upload release-smoke diagnostics + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7 + with: + name: desktop-release-smoke + path: | + release-smoke-artifacts + desktop/test-results + desktop/playwright-release-smoke-report + if-no-files-found: warn + retention-days: 14 + assemble-manifest: name: Assemble multi-platform latest.json # Only the tag-bound setup path can reach this job. @@ -785,9 +842,10 @@ jobs: needs.release-macos-x64.result == 'success' && needs.release-linux.result == 'success' && needs.release-windows.result == 'success' && + needs.desktop-release-smoke.result == 'success' && github.ref == format('refs/tags/desktop-v{0}', needs.setup.outputs.version) runs-on: ubuntu-latest - needs: [setup, release, release-macos-x64, release-linux, release-windows] + needs: [setup, release, release-macos-x64, release-linux, release-windows, desktop-release-smoke] timeout-minutes: 10 permissions: contents: write diff --git a/Cargo.lock b/Cargo.lock index ddab506c2a0..ab0253dbebf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -881,6 +881,7 @@ dependencies = [ "tokio", "tracing", "url", + "uuid", ] [[package]] diff --git a/Justfile b/Justfile index 6719d3c53f1..b3627f689c8 100644 --- a/Justfile +++ b/Justfile @@ -276,6 +276,10 @@ desktop-e2e-smoke: desktop-e2e-integration: _ensure-migrations cd {{desktop_dir}} && pnpm test:e2e:integration +# Run the deterministic desktop correctness smoke against an isolated local relay +desktop-release-smoke: + ./scripts/run-desktop-release-smoke.sh + # Run only the e2e specs changed vs origin/main (both projects) before pushing desktop-e2e-pre-push: _ensure-migrations git fetch origin main diff --git a/TESTING.md b/TESTING.md index 7c107da5754..29d07a80de0 100644 --- a/TESTING.md +++ b/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="" + +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= members=1502 late_pubkey= +PASS late-member-action event_id= +PASS targeted-repair-preserves-metadata-and-admin-events channel= +PASS discovery-after-republish channel= members=1502 late_pubkey= +``` + +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). diff --git a/crates/buzz-admin/Cargo.toml b/crates/buzz-admin/Cargo.toml index 00c2804cbcf..263ba4eb319 100644 --- a/crates/buzz-admin/Cargo.toml +++ b/crates/buzz-admin/Cargo.toml @@ -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"] } diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index 580d5865913..42a7de84f7c 100644 --- a/crates/buzz-admin/src/main.rs +++ b/crates/buzz-admin/src/main.rs @@ -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, + /// 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 { 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 { Ok(TenantContext::resolved(record.id, record.host)) } -async fn reconcile_channels(relay_key_arg: Option) -> Result<()> { +async fn reconcile_channels( + channel_arg: Option, + relay_key_arg: Option, +) -> 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) -> 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) -> 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 = 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])?); + // 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 mut tags: Vec = 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?; } - 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 = vec![Tag::parse(["d", &channel_id_str])?]; - for m in members - .iter() - .filter(|m| m.role == "owner" || m.role == "admin") + // kind:39001 — admins { - let pk = hex::encode(&m.pubkey); - tags.push(Tag::parse(["p", &pk, &m.role])?); + let mut tags: Vec = 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 diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 3ca9b3d901c..8035ab58adb 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -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"] diff --git a/crates/buzz-db/src/feed.rs b/crates/buzz-db/src/feed.rs index 40e58d0d060..6900e2061c5 100644 --- a/crates/buzz-db/src/feed.rs +++ b/crates/buzz-db/src/feed.rs @@ -886,4 +886,40 @@ mod tests { let unique: std::collections::HashSet> = 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 = (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" + ); + } } diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 351eb9d2856..49a646abbe7 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -104,6 +104,21 @@ pub async fn insert_mentions( community_id: CommunityId, event: &nostr::Event, channel_id: Option, +) -> 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, ) -> 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 = 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 = 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(()) } @@ -4051,6 +4073,27 @@ impl Db { workflow::list_workflow_runs(&self.pool, community_id, workflow_id, limit).await } + /// List one keyset-paginated page of workflow runs. + #[datastore_span(name = "list_workflow_runs_page", system = "postgresql")] + pub async fn list_workflow_runs_page( + &self, + community_id: CommunityId, + workflow_id: Uuid, + before: Option>, + before_id: Option, + limit: i64, + ) -> Result> { + workflow::list_workflow_runs_page( + &self.pool, + community_id, + workflow_id, + before, + before_id, + limit, + ) + .await + } + /// Update a workflow run's status. #[datastore_span(name = "update_workflow_run", system = "postgresql")] pub async fn update_workflow_run( @@ -4060,7 +4103,7 @@ impl Db { status: workflow::RunStatus, current_step: i32, trace: &serde_json::Value, - error: Option<&str>, + failure: Option>, ) -> Result<()> { workflow::update_workflow_run( &self.pool, @@ -4069,7 +4112,7 @@ impl Db { status, current_step, trace, - error, + failure, ) .await } @@ -4895,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), @@ -5438,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 = 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() { diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index b3bc5f31cf7..be87faa1ac2 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -625,7 +625,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 30); + assert_eq!(migrations.len(), 31); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1038,6 +1038,27 @@ mod tests { assert!(deletion_recovery.contains("SET LOCAL lock_timeout = '5s'")); } + #[test] + fn workflow_run_error_codes_are_additive_and_backfilled_without_parsing_diagnostics() { + let mut migrations: Vec<_> = MIGRATOR.iter().collect(); + migrations.sort_by_key(|migration| migration.version); + + assert_eq!(migrations[30].version, 31); + let sql = migrations[30].sql.as_str(); + assert!(sql.contains("ALTER TABLE workflow_runs ADD COLUMN error_code TEXT")); + assert!(sql.contains("SET error_code = 'legacy_unclassified'")); + assert!(sql.contains("status IN ('failed', 'cancelled')")); + assert!(!sql.contains("error_message LIKE")); + assert!(!MIGRATOR + .iter() + .find(|migration| migration.version == 1) + .expect("initial migration") + .sql + .as_str() + .contains("error_code")); + assert!(include_str!("../../../schema/schema.sql").contains("error_code TEXT")); + } + #[test] fn migration_lint_detects_tables_missing_community_id_by_default() { let sql = r#" diff --git a/crates/buzz-db/src/workflow.rs b/crates/buzz-db/src/workflow.rs index ad1fd3a9396..e970e978aaf 100644 --- a/crates/buzz-db/src/workflow.rs +++ b/crates/buzz-db/src/workflow.rs @@ -216,8 +216,11 @@ pub struct WorkflowRunRecord { pub started_at: Option>, /// When execution finished (success or failure). pub completed_at: Option>, - /// Error message if the run failed. + /// Redacted human-readable diagnostic for failed or cancelled runs. pub error_message: Option, + /// Stable machine-readable failure or cancellation classification. + /// Kept separate from `error_message` so callers never parse diagnostics. + pub error_code: Option, /// When the run record was created. pub created_at: DateTime, } @@ -831,7 +834,7 @@ pub async fn get_workflow_run( let row = sqlx::query( r#" SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step, - execution_trace, trigger_context, started_at, completed_at, error_message, created_at + execution_trace, trigger_context, started_at, completed_at, error_message, error_code, created_at FROM workflow_runs WHERE community_id = $1 AND id = $2 "#, @@ -845,26 +848,40 @@ pub async fn get_workflow_run( row_to_run_record(row) } -/// List runs for a workflow, newest first, up to `limit` rows. -pub async fn list_workflow_runs( +/// List runs for a workflow using a stable newest-first keyset. +/// +/// Rows are ordered by `(created_at DESC, id DESC)`. A cursor is valid only +/// when both `before` and `before_id` are supplied; callers should pass the +/// final row from the previous page. `limit` is clamped to the shared list +/// bounds. +pub async fn list_workflow_runs_page( pool: &PgPool, community_id: CommunityId, workflow_id: Uuid, + before: Option>, + before_id: Option, limit: i64, ) -> Result> { - let limit = limit.min(1000); + let limit = limit.clamp(1, LIST_MAX_LIMIT); let rows = sqlx::query( r#" SELECT community_id, id, workflow_id, status::text AS status, trigger_event_id, current_step, - execution_trace, trigger_context, started_at, completed_at, error_message, created_at + execution_trace, trigger_context, started_at, completed_at, error_message, error_code, created_at FROM workflow_runs WHERE community_id = $1 AND workflow_id = $2 - ORDER BY created_at DESC - LIMIT $3 + AND ( + $3::timestamptz IS NULL + OR $4::uuid IS NULL + OR (created_at, id) < ($3, $4) + ) + ORDER BY created_at DESC, id DESC + LIMIT $5 "#, ) .bind(community_id.as_uuid()) .bind(workflow_id) + .bind(before) + .bind(before_id) .bind(limit) .fetch_all(pool) .await?; @@ -872,7 +889,26 @@ pub async fn list_workflow_runs( rows.into_iter().map(row_to_run_record).collect() } -/// Update run status, current step, execution trace, and optional error message. +/// List runs for a workflow, newest first, up to `limit` rows. +pub async fn list_workflow_runs( + pool: &PgPool, + community_id: CommunityId, + workflow_id: Uuid, + limit: i64, +) -> Result> { + list_workflow_runs_page(pool, community_id, workflow_id, None, None, limit).await +} + +/// Structured failure persisted for a workflow run. +#[derive(Debug, Clone, Copy)] +pub struct WorkflowRunFailure<'a> { + /// Stable machine-readable failure code. + pub code: &'a str, + /// Human-readable failure detail. + pub message: &'a str, +} + +/// Update run status, current step, execution trace, and optional failure. /// /// Fix C3: `started_at` is set when the NEW status is 'running' and `started_at` /// has not yet been stamped (IS NULL). The original code read `status` from the @@ -885,26 +921,31 @@ pub async fn update_workflow_run( status: RunStatus, current_step: i32, trace: &serde_json::Value, - error: Option<&str>, + failure: Option>, ) -> Result<()> { let status_str = status.to_string(); + let (error_code, error) = failure + .map(|failure| (Some(failure.code), Some(failure.message))) + .unwrap_or((None, None)); let affected = sqlx::query( r#" UPDATE workflow_runs SET status = $1::run_status, current_step = $2, execution_trace = $3, - error_message = $4, - started_at = CASE WHEN $5 = 'running' AND started_at IS NULL + error_code = $4, + error_message = $5, + started_at = CASE WHEN $6 = 'running' AND started_at IS NULL THEN NOW() ELSE started_at END, - completed_at = CASE WHEN $6 IN ('completed','failed','cancelled') + completed_at = CASE WHEN $7 IN ('completed','failed','cancelled') THEN NOW() ELSE completed_at END - WHERE community_id = $7 AND id = $8 + WHERE community_id = $8 AND id = $9 "#, ) .bind(&status_str) .bind(current_step) .bind(trace) + .bind(error_code) .bind(error) .bind(&status_str) // for started_at CASE .bind(&status_str) // for completed_at CASE @@ -1169,6 +1210,7 @@ fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result { started_at: row.try_get("started_at")?, completed_at: row.try_get("completed_at")?, error_message: row.try_get("error_message")?, + error_code: row.try_get("error_code")?, created_at: row.try_get("created_at")?, }) } @@ -1473,6 +1515,7 @@ mod tests { started_at: Some(now), completed_at: None, error_message: None, + error_code: None, created_at: now, }; @@ -1501,6 +1544,7 @@ mod tests { started_at: None, completed_at: None, error_message: None, + error_code: None, created_at: now, }; @@ -1524,6 +1568,7 @@ mod tests { started_at: Some(now), completed_at: Some(now), error_message: Some("step timeout exceeded".to_owned()), + error_code: Some("step_timeout".to_owned()), created_at: now, }; @@ -1555,6 +1600,7 @@ mod tests { started_at: Some(now), completed_at: Some(now), error_message: None, + error_code: None, created_at: now, }; @@ -1577,6 +1623,7 @@ mod tests { started_at: None, completed_at: None, error_message: None, + error_code: None, created_at: now, }; diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 6ada39f5535..4d5bb772dba 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -21,7 +21,7 @@ use crate::state::AppState; use super::{api_error, internal_error, not_found}; -async fn enforce_http_admission( +pub(crate) async fn enforce_http_admission( state: &AppState, tenant: &TenantContext, pubkey: &nostr::PublicKey, @@ -1938,7 +1938,10 @@ pub async fn workflow_webhook( buzz_db::workflow::RunStatus::Failed, 0, &serde_json::json!([]), - Some(&format!("definition parse error: {e}")), + Some(buzz_db::workflow::WorkflowRunFailure { + code: "invalid_definition", + message: &format!("definition parse error: {e}"), + }), ) .await { diff --git a/crates/buzz-relay/src/api/mod.rs b/crates/buzz-relay/src/api/mod.rs index d9f829433b1..2a942bc8039 100644 --- a/crates/buzz-relay/src/api/mod.rs +++ b/crates/buzz-relay/src/api/mod.rs @@ -9,6 +9,7 @@ pub mod media; pub mod mesh_demo; pub mod nip05; pub mod operator; +pub mod workflows; // Re-export imeta helpers used by ingest pipeline. pub use crate::handlers::imeta::{validate_imeta_tags, verify_imeta_blobs}; diff --git a/crates/buzz-relay/src/api/workflows.rs b/crates/buzz-relay/src/api/workflows.rs new file mode 100644 index 00000000000..a3d5a6c729e --- /dev/null +++ b/crates/buzz-relay/src/api/workflows.rs @@ -0,0 +1,264 @@ +//! Authorized structured reads for workflow execution state. +//! +//! Runs and approvals are relay-owned database rows, not Nostr events. These +//! endpoints expose those read models without inventing synthetic events. + +use std::sync::Arc; + +use axum::{ + extract::{Path, Query, RawQuery, State}, + http::{HeaderMap, StatusCode}, + response::Json, +}; +use chrono::{DateTime, Utc}; +use serde::Deserialize; +use serde_json::Value; +use uuid::Uuid; + +use buzz_core::TenantContext; + +use crate::{ + api::{api_error, bridge, internal_error}, + state::AppState, +}; + +const DEFAULT_RUN_LIMIT: i64 = 20; +const MAX_RUN_LIMIT: i64 = 100; + +/// Pagination query for workflow run history. +#[derive(Debug, Deserialize, Default)] +pub struct RunsQuery { + before: Option>, + before_id: Option, + limit: Option, +} + +fn request_path(path: &str, raw_query: Option<&str>) -> String { + match raw_query { + Some(query) if !query.is_empty() => format!("{path}?{query}"), + _ => path.to_string(), + } +} + +async fn authorize_workflow_read( + state: &Arc, + headers: &HeaderMap, + path: &str, + raw_query: Option<&str>, + workflow_id: Uuid, +) -> Result)> { + let raw_host = headers + .get(axum::http::header::HOST) + .and_then(|value| value.to_str().ok()) + .unwrap_or(""); + let tenant = crate::tenant::bind_community(&state.db, raw_host) + .await + .map_err(|_| { + api_error( + StatusCode::NOT_FOUND, + "relay: no community is configured for this host", + ) + })?; + + let path_with_query = request_path(path, raw_query); + let url = bridge::nip98_expected_url(&state.config.relay_url, &tenant, &path_with_query); + let (pubkey, event_id_bytes) = + bridge::verify_bridge_auth(headers, "GET", &url, None, state.config.require_auth_token)?; + bridge::enforce_http_admission(state, &tenant, &pubkey).await?; + bridge::check_nip98_replay(state, &tenant, event_id_bytes).await?; + + let pubkey_bytes = pubkey.to_bytes().to_vec(); + let auth_tag = headers + .get("x-auth-tag") + .and_then(|value| value.to_str().ok()); + super::relay_members::enforce_relay_membership( + state, + tenant.community(), + &pubkey_bytes, + auth_tag, + ) + .await?; + + let workflow = state + .db + .get_workflow(tenant.community(), workflow_id) + .await + .map_err(|error| match error { + buzz_db::error::DbError::NotFound(_) => { + api_error(StatusCode::NOT_FOUND, "workflow not found") + } + other => internal_error(&format!("get workflow for run read: {other}")), + })?; + let channel_id = workflow + .channel_id + .ok_or_else(|| api_error(StatusCode::FORBIDDEN, "workflow is not channel-scoped"))?; + let accessible = state + .get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes) + .await + .map_err(|error| internal_error(&format!("workflow channel access lookup: {error}")))?; + if !accessible.contains(&channel_id) { + return Err(api_error( + StatusCode::FORBIDDEN, + "workflow is not accessible", + )); + } + + Ok(tenant) +} + +/// `GET /workflows/{workflow_id}/runs` — one authorized, keyset-paginated page. +pub async fn workflow_runs( + State(state): State>, + Path(workflow_id): Path, + headers: HeaderMap, + RawQuery(raw_query): RawQuery, + Query(query): Query, +) -> Result, (StatusCode, Json)> { + if query.before.is_some() != query.before_id.is_some() { + return Err(api_error( + StatusCode::BAD_REQUEST, + "before and before_id must be supplied together", + )); + } + let limit = query.limit.unwrap_or(DEFAULT_RUN_LIMIT); + if !(1..=MAX_RUN_LIMIT).contains(&limit) { + return Err(api_error( + StatusCode::BAD_REQUEST, + "limit must be between 1 and 100", + )); + } + + let path = format!("/workflows/{workflow_id}/runs"); + let tenant = + authorize_workflow_read(&state, &headers, &path, raw_query.as_deref(), workflow_id).await?; + let mut rows = state + .db + .list_workflow_runs_page( + tenant.community(), + workflow_id, + query.before, + query.before_id, + limit + 1, + ) + .await + .map_err(|error| internal_error(&format!("list workflow runs: {error}")))?; + + let has_more = rows.len() > limit as usize; + rows.truncate(limit as usize); + let next = if has_more { + rows.last().map(|last| { + serde_json::json!({ + "before": last.created_at, + "before_id": last.id, + }) + }) + } else { + None + }; + + Ok(Json(serde_json::json!({ + "runs": rows.iter().map(run_json).collect::>(), + "next": next, + }))) +} + +/// `GET /workflows/{workflow_id}/runs/{run_id}/approvals` — approvals for a run. +pub async fn run_approvals( + State(state): State>, + Path((workflow_id, run_id)): Path<(Uuid, Uuid)>, + headers: HeaderMap, +) -> Result, (StatusCode, Json)> { + let path = format!("/workflows/{workflow_id}/runs/{run_id}/approvals"); + let tenant = authorize_workflow_read(&state, &headers, &path, None, workflow_id).await?; + + let run = state + .db + .get_workflow_run(tenant.community(), run_id) + .await + .map_err(|error| match error { + buzz_db::error::DbError::NotFound(_) => { + api_error(StatusCode::NOT_FOUND, "workflow run not found") + } + other => internal_error(&format!("get workflow run for approval read: {other}")), + })?; + if run.workflow_id != workflow_id { + return Err(api_error(StatusCode::NOT_FOUND, "workflow run not found")); + } + + let approvals = state + .db + .get_run_approvals(tenant.community(), workflow_id, run_id) + .await + .map_err(|error| internal_error(&format!("list run approvals: {error}")))?; + Ok(Json(serde_json::json!({ + "approvals": approvals.iter().map(approval_json).collect::>(), + }))) +} + +fn run_json(run: &buzz_db::workflow::WorkflowRunRecord) -> Value { + serde_json::json!({ + "id": run.id, + "workflow_id": run.workflow_id, + "status": run.status, + "current_step": run.current_step, + "execution_trace": run.execution_trace, + "started_at": run.started_at.map(|value| value.timestamp()), + "completed_at": run.completed_at.map(|value| value.timestamp()), + "error_code": run.error_code, + "error_message": run.error_message, + "created_at": run.created_at.timestamp(), + }) +} + +fn approval_json(approval: &buzz_db::workflow::ApprovalRecord) -> Value { + serde_json::json!({ + "approval_ref": hex::encode(&approval.token), + "workflow_id": approval.workflow_id, + "run_id": approval.run_id, + "step_id": approval.step_id, + "step_index": approval.step_index, + "approver_spec": approval.approver_spec, + "status": approval.status, + "approver_pubkey": approval.approver_pubkey.as_ref().map(hex::encode), + "note": approval.note, + "expires_at": approval.expires_at, + "created_at": approval.created_at.timestamp(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn request_path_preserves_signed_query_verbatim() { + assert_eq!( + request_path("/workflows/id/runs", Some("limit=20&before_id=abc")), + "/workflows/id/runs?limit=20&before_id=abc" + ); + assert_eq!( + request_path("/workflows/id/runs", None), + "/workflows/id/runs" + ); + } + + #[test] + fn approval_wire_does_not_expose_hash_as_token() { + let approval = buzz_db::workflow::ApprovalRecord { + token: vec![0xab; 32], + workflow_id: Uuid::new_v4(), + run_id: Uuid::new_v4(), + step_id: "review".to_string(), + step_index: 1, + approver_spec: "any".to_string(), + status: buzz_db::workflow::ApprovalStatus::Pending, + approver_pubkey: None, + note: None, + expires_at: Utc::now(), + created_at: Utc::now(), + }; + let wire = approval_json(&approval); + assert!(wire.get("token").is_none()); + assert_eq!(wire["approval_ref"], hex::encode([0xab; 32])); + } +} diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index abb9bb20665..29abe9f27d4 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -964,7 +964,10 @@ async fn handle_workflow_trigger( RunStatus::Failed, 0, &serde_json::json!([]), - Some(&format!("definition parse error: {e}")), + Some(buzz_db::workflow::WorkflowRunFailure { + code: "invalid_definition", + message: &format!("definition parse error: {e}"), + }), ) .await { @@ -1261,7 +1264,10 @@ async fn handle_approval_deny( RunStatus::Cancelled, run.current_step, &run.execution_trace, - Some(&cancel_msg), + Some(buzz_db::workflow::WorkflowRunFailure { + code: "approval_denied", + message: &cancel_msg, + }), ) .await { @@ -1329,7 +1335,10 @@ async fn resume_workflow_after_approval( RunStatus::Failed, run.current_step, &run.execution_trace, - Some(&format!("definition parse error: {e}")), + Some(buzz_db::workflow::WorkflowRunFailure { + code: "invalid_definition", + message: &format!("definition parse error: {e}"), + }), ) .await { diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 88a9f0c731c..0dc6cbd5039 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -1033,6 +1033,18 @@ async fn emit_addressable_discovery_event( Ok(()) } +fn group_members_tags(group_id: &str, members: &[MemberRecord]) -> anyhow::Result> { + let mut tags: Vec = 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 = 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) -> EventTopic { mod tests { use super::*; + #[test] + fn group_members_snapshot_keeps_members_past_one_thousand() { + let channel_id = Uuid::new_v4(); + let members: Vec = (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 = diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 82ad9938a2f..1dce66e91e4 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -72,6 +72,14 @@ pub fn build_router(state: Arc) -> Router { .route("/events", post(api::bridge::submit_event)) .route("/query", post(api::bridge::query_events)) .route("/count", post(api::bridge::count_events)) + .route( + "/workflows/{workflow_id}/runs", + get(api::workflows::workflow_runs), + ) + .route( + "/workflows/{workflow_id}/runs/{run_id}/approvals", + get(api::workflows::run_approvals), + ) .route( "/operator/communities", get(api::operator::list_owned_communities).post(api::operator::provision_community), diff --git a/crates/buzz-workflow/src/error.rs b/crates/buzz-workflow/src/error.rs index 292f8dd027c..109d4a2cb3d 100644 --- a/crates/buzz-workflow/src/error.rs +++ b/crates/buzz-workflow/src/error.rs @@ -65,8 +65,50 @@ pub enum WorkflowError { NotImplemented(String), } +impl WorkflowError { + /// Stable run-level classification. Diagnostics remain in `Display` output. + pub const fn code(&self) -> &'static str { + match self { + Self::InvalidYaml(_) => "invalid_yaml", + Self::InvalidDefinition(_) => "invalid_definition", + Self::ConditionError(_) => "condition_evaluation_failed", + Self::TemplateError(_) => "template_resolution_failed", + Self::StepTimeout { .. } => "step_timeout", + Self::WebhookError(_) => "webhook_failed", + Self::CapacityExceeded => "capacity_exceeded", + Self::Database(_) => "database_error", + Self::Unauthorized(_) => "owner_unauthorized", + Self::NotImplemented(_) => "action_not_implemented", + } + } +} + impl From for WorkflowError { fn from(e: buzz_db::error::DbError) -> Self { WorkflowError::Database(e.to_string()) } } + +#[cfg(test)] +mod tests { + use super::WorkflowError; + + #[test] + fn workflow_error_codes_are_stable_and_separate_from_diagnostics() { + let timeout = WorkflowError::StepTimeout { + step_id: "notify".to_owned(), + timeout_secs: 30, + }; + assert_eq!(timeout.code(), "step_timeout"); + assert!(timeout.to_string().contains("notify")); + + let webhook = WorkflowError::WebhookError("secret-bearing detail".to_owned()); + assert_eq!(webhook.code(), "webhook_failed"); + assert!(!webhook.code().contains("secret-bearing detail")); + + assert_eq!( + WorkflowError::NotImplemented("SendDm".to_owned()).code(), + "action_not_implemented" + ); + } +} diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index e1422211690..fe8b477ba40 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -242,7 +242,10 @@ impl WorkflowEngine { RunStatus::Failed, step_count, &trace_json, - Some("approval gates not yet implemented — see WF-08"), + Some(buzz_db::workflow::WorkflowRunFailure { + code: "approval_not_supported", + message: "approval gates not yet implemented — see WF-08", + }), ) .await { @@ -285,7 +288,10 @@ impl WorkflowEngine { RunStatus::Failed, progress.step_index as i32, &trace_json, - Some(&e.to_string()), + Some(buzz_db::workflow::WorkflowRunFailure { + code: e.code(), + message: &e.to_string(), + }), ) .await { diff --git a/desktop/.gitignore b/desktop/.gitignore index 4d3e0c5ac5a..5dda9a099b5 100644 --- a/desktop/.gitignore +++ b/desktop/.gitignore @@ -14,6 +14,7 @@ dist-ssr playwright-report playwright-report.json test-results +playwright-release-smoke-report *.local playwright-report test-results diff --git a/desktop/package.json b/desktop/package.json index 3601f25185e..b6581d48057 100644 --- a/desktop/package.json +++ b/desktop/package.json @@ -20,6 +20,7 @@ "test:e2e": "pnpm build:e2e && playwright test", "test:e2e:smoke": "pnpm build:e2e && playwright test --project=smoke", "test:e2e:integration": "pnpm build:e2e && playwright test --project=integration", + "test:e2e:release-smoke": "pnpm build:e2e && playwright test --config=playwright.release-smoke.config.ts", "test:e2e:report": "playwright show-report", "tauri:build": "tauri build" }, diff --git a/desktop/playwright.release-smoke.config.ts b/desktop/playwright.release-smoke.config.ts new file mode 100644 index 00000000000..19ac3cbf06d --- /dev/null +++ b/desktop/playwright.release-smoke.config.ts @@ -0,0 +1,36 @@ +import { defineConfig, devices } from "@playwright/test"; + +const webPort = process.env.BUZZ_RELEASE_SMOKE_WEB_PORT ?? "4173"; +const webUrl = `http://127.0.0.1:${webPort}`; + +export default defineConfig({ + testDir: "./tests/e2e", + testMatch: [ + "**/release-smoke.spec.ts", + "**/dm-history-live-regression.spec.ts", + "**/foreground-responsiveness-regression.spec.ts", + ], + timeout: 10 * 60_000, + retries: 0, + workers: 1, + reporter: [ + ["list"], + ["json", { outputFile: "test-results/release-smoke/playwright.json" }], + [ + "html", + { open: "never", outputFolder: "playwright-release-smoke-report" }, + ], + ], + use: { + ...devices["Desktop Chrome"], + baseURL: webUrl, + screenshot: "only-on-failure", + trace: "retain-on-failure", + }, + webServer: { + command: `python3 -m http.server ${webPort} -d dist`, + cwd: ".", + reuseExistingServer: false, + url: webUrl, + }, +}); diff --git a/desktop/src-tauri/src/commands/workflows.rs b/desktop/src-tauri/src/commands/workflows.rs index 1d5f309fb5c..25e02980fa7 100644 --- a/desktop/src-tauri/src/commands/workflows.rs +++ b/desktop/src-tauri/src/commands/workflows.rs @@ -5,7 +5,7 @@ use tauri::State; use crate::{ app_state::AppState, events, - relay::{parse_command_response, query_relay, submit_event}, + relay::{get_relay_json, parse_command_response, query_relay, submit_event}, }; // ── Wire shapes (snake_case, consumed by tauriWorkflows.ts) ────────────────── @@ -47,6 +47,41 @@ pub struct WorkflowSaveWire { pub webhook_secret: Option, } +#[derive(Debug, Clone, serde::Deserialize, Serialize, PartialEq)] +pub struct WorkflowRunCursorWire { + pub before: String, + pub before_id: String, +} + +#[derive(Debug, Clone, serde::Deserialize, Serialize, PartialEq)] +pub struct WorkflowRunsWire { + pub runs: Vec, + pub next: Option, +} + +#[derive(Debug, Clone, serde::Deserialize, Serialize, PartialEq)] +pub struct WorkflowApprovalsWire { + pub approvals: Vec, +} + +/// Canonical trigger acknowledgement consumed by the Desktop client. +/// +/// The relay currently returns only `run_id`; the workflow id is the command +/// input and a newly-created run always begins pending. Keeping that adaptation +/// here prevents the frontend from guessing fields or confusing the trigger +/// event id with the persisted run id. +#[derive(Debug, Clone, Serialize, PartialEq)] +pub struct WorkflowTriggerWire { + pub run_id: String, + pub workflow_id: String, + pub status: String, +} + +#[derive(Debug, serde::Deserialize)] +struct WorkflowTriggerAck { + run_id: String, +} + // ── Reads ──────────────────────────────────────────────────────────────────── #[tauri::command] @@ -121,26 +156,16 @@ pub async fn get_workflow( pub async fn get_workflow_runs( workflow_id: String, limit: Option, - _state: State<'_, AppState>, -) -> Result, String> { - // TODO(workflow-runs): Run reconstruction is a clearly-scoped follow-up. - // The authoritative run record the frontend's `WorkflowRun` shape needs - // (status / current_step / execution_trace / error_message) lives in the - // relay DB and is not exposed to the desktop client as a single queryable - // record. If the relay starts emitting lifecycle events (46001–46007, …), - // folding that stream into `WorkflowRun` would be another viable design. - // The important bit for this command is that raw lifecycle events are not - // the `RawWorkflowRun` contract. - // - // Until then we return a bare empty array — NOT a raw-event wrapper. The - // frontend wrapper (`getWorkflowRuns`) does `raw.map(fromRawWorkflowRun)`, - // so it must receive an array; the wrapped `{ runs: [...] }` shape would - // make `.map()` throw and crash the detail panel (the same TypeError class - // as the original page bug). Raw lifecycle events also don't carry the - // `id`/`workflow_id`/`status`/… fields `RawWorkflowRun` expects, so an - // empty list is the honest, safe placeholder. - let _ = (workflow_id, limit); - Ok(Vec::new()) + state: State<'_, AppState>, +) -> Result { + let workflow_id = + uuid::Uuid::parse_str(&workflow_id).map_err(|_| "invalid workflow id".to_string())?; + let limit = limit.unwrap_or(20).clamp(1, 100); + get_relay_json( + &state, + &format!("/workflows/{workflow_id}/runs?limit={limit}"), + ) + .await } // ── Writes ─────────────────────────────────────────────────────────────────── @@ -242,10 +267,10 @@ pub async fn delete_workflow( pub async fn trigger_workflow( workflow_id: String, state: State<'_, AppState>, -) -> Result { +) -> Result { let builder = events::build_workflow_trigger(&workflow_id)?; let result = submit_event(builder, &state).await?; - Ok(serde_json::json!({ "event_id": result.event_id })) + trigger_wire_from_message(workflow_id, &result.message) } // ── Approvals ──────────────────────────────────────────────────────────────── @@ -254,15 +279,17 @@ pub async fn trigger_workflow( pub async fn get_run_approvals( workflow_id: String, run_id: String, - _state: State<'_, AppState>, -) -> Result, String> { - // TODO(workflow-runs): Like runs (see `get_workflow_runs`), reconstructing - // approvals into the frontend's `WorkflowApproval` shape from lifecycle - // events (46010/46011/46012) is a clearly-scoped follow-up tracked under - // TODO(workflow-runs). Return a bare empty array so the frontend's - // `getRunApprovals` (`raw.map(fromRawApproval)`) is safe. - let _ = (workflow_id, run_id); - Ok(Vec::new()) + state: State<'_, AppState>, +) -> Result { + let workflow_id = + uuid::Uuid::parse_str(&workflow_id).map_err(|_| "invalid workflow id".to_string())?; + let run_id = + uuid::Uuid::parse_str(&run_id).map_err(|_| "invalid workflow run id".to_string())?; + get_relay_json( + &state, + &format!("/workflows/{workflow_id}/runs/{run_id}/approvals"), + ) + .await } #[tauri::command] @@ -289,6 +316,21 @@ pub async fn deny_approval( // ── Helpers (pure, unit-tested in workflows_tests.rs) ───────────────────────── +fn trigger_wire_from_message( + workflow_id: String, + message: &str, +) -> Result { + let ack: WorkflowTriggerAck = parse_command_response(message)?; + if ack.run_id.trim().is_empty() { + return Err("workflow trigger response contained an empty run_id".to_string()); + } + Ok(WorkflowTriggerWire { + run_id: ack.run_id, + workflow_id, + status: "pending".to_string(), + }) +} + fn current_pubkey_hex(state: &AppState) -> Result { let keys = state.keys.lock().map_err(|e| e.to_string())?; Ok(keys.public_key().to_hex()) diff --git a/desktop/src-tauri/src/commands/workflows_tests.rs b/desktop/src-tauri/src/commands/workflows_tests.rs index f07f4b0f421..647cc687064 100644 --- a/desktop/src-tauri/src/commands/workflows_tests.rs +++ b/desktop/src-tauri/src/commands/workflows_tests.rs @@ -189,21 +189,41 @@ fn workflow_wire_serializes_with_snake_case_keys() { } #[test] -fn runs_and_approvals_serialize_to_bare_empty_array() { - // Regression guard for the crash class this fix closed. The frontend - // wrappers `getWorkflowRuns` / `getRunApprovals` do `raw.map(...)`, so the - // Rust side MUST return a bare JSON array. A wrapped `{ runs: [...] }` / - // `{ approvals: [...] }` shape would make `.map()` throw and crash the - // detail panel — the same TypeError class as the original page bug. - // - // The commands take `State`, so we can't invoke them directly in - // a unit test; instead we pin the exact value they return (`Vec::new()` of - // their `Vec` element type) and assert its serialized shape. - let runs: Vec = Vec::new(); - let approvals: Vec = Vec::new(); - assert_eq!(serde_json::to_string(&runs).expect("serialize runs"), "[]"); +fn trigger_response_uses_persisted_run_id_contract() { + let wire = trigger_wire_from_message( + WF.to_string(), + "response:{\"run_id\":\"33333333-3333-3333-3333-333333333333\"}", + ) + .expect("parse trigger response"); + + assert_eq!(wire.run_id, "33333333-3333-3333-3333-333333333333"); + assert_eq!(wire.workflow_id, WF); + assert_eq!(wire.status, "pending"); + let value = serde_json::to_value(wire).expect("serialize trigger response"); + assert!(value.get("event_id").is_none()); +} + +#[test] +fn trigger_response_rejects_missing_or_empty_run_id() { + assert!(trigger_wire_from_message(WF.to_string(), "response:{}").is_err()); + assert!(trigger_wire_from_message(WF.to_string(), "response:{\"run_id\":\" \"}",).is_err()); +} + +#[test] +fn run_reads_serialize_to_backend_envelopes() { + let runs = WorkflowRunsWire { + runs: Vec::new(), + next: None, + }; + let approvals = WorkflowApprovalsWire { + approvals: Vec::new(), + }; + assert_eq!( + serde_json::to_value(runs).expect("serialize runs"), + serde_json::json!({ "runs": [], "next": null }) + ); assert_eq!( - serde_json::to_string(&approvals).expect("serialize approvals"), - "[]" + serde_json::to_value(approvals).expect("serialize approvals"), + serde_json::json!({ "approvals": [] }) ); } diff --git a/desktop/src-tauri/src/relay.rs b/desktop/src-tauri/src/relay.rs index 7b636a4a822..685f83b7999 100644 --- a/desktop/src-tauri/src/relay.rs +++ b/desktop/src-tauri/src/relay.rs @@ -532,6 +532,9 @@ pub struct AgentProfileInfo { // ── Signed-event submission ───────────────────────────────────────────────── +mod get; +pub use get::get_relay_json; + mod submit; pub use submit::{ submit_event, submit_event_at_with_keys, submit_signed_event_at_with_keys, SubmitEventResponse, diff --git a/desktop/src-tauri/src/relay/get.rs b/desktop/src-tauri/src/relay/get.rs new file mode 100644 index 00000000000..7d0855f463f --- /dev/null +++ b/desktop/src-tauri/src/relay/get.rs @@ -0,0 +1,37 @@ +use reqwest::Method; +use serde::de::DeserializeOwned; + +use crate::app_state::AppState; + +use super::{ + build_nip98_auth_header, classify_request_error, parse_json_response, + relay_api_base_url_with_override, relay_error_message, +}; + +/// Execute an authenticated GET against the active relay and decode its JSON body. +pub async fn get_relay_json( + state: &AppState, + path_with_query: &str, +) -> Result { + if !path_with_query.starts_with('/') { + return Err("relay GET path must begin with '/'".to_string()); + } + crate::relay_admission::wait_for_rate_limit().await; + let url = format!( + "{}{}", + relay_api_base_url_with_override(state), + path_with_query + ); + let auth = build_nip98_auth_header(&Method::GET, &url, &[], state)?; + let response = state + .http_client + .get(&url) + .header("Authorization", auth) + .send() + .await + .map_err(|error| classify_request_error(&error))?; + if !response.status().is_success() { + return Err(relay_error_message(response).await); + } + parse_json_response(response).await +} diff --git a/desktop/src/app/AppHuddleShell.tsx b/desktop/src/app/AppHuddleShell.tsx index 29dcd26cdfb..68f9eacd6fd 100644 --- a/desktop/src/app/AppHuddleShell.tsx +++ b/desktop/src/app/AppHuddleShell.tsx @@ -68,7 +68,7 @@ export function AppHuddleShell({ {children} {isRoom || !isCompanionOpen ? ( -
+
{ assert.equal( shouldHideAgentFromMentions({ + ownerOnly: false, isAgent: false, isMember: false, pubkey: PUB_A, @@ -392,6 +395,7 @@ test("shouldHideAgentFromMentions: never hides non-agents", () => { test("shouldHideAgentFromMentions: shows invocable agents even when non-member", () => { assert.equal( shouldHideAgentFromMentions({ + ownerOnly: false, isAgent: true, isMember: false, pubkey: PUB_A, @@ -405,6 +409,7 @@ test("shouldHideAgentFromMentions: shows invocable agents even when non-member", test("shouldHideAgentFromMentions: hides non-member non-invocable agents", () => { assert.equal( shouldHideAgentFromMentions({ + ownerOnly: false, isAgent: true, isMember: false, pubkey: PUB_A, @@ -418,6 +423,7 @@ test("shouldHideAgentFromMentions: hides non-member non-invocable agents", () => test("shouldHideAgentFromMentions: hides member agents with an explicit not-invocable directory entry (Fizz)", () => { assert.equal( shouldHideAgentFromMentions({ + ownerOnly: false, isAgent: true, isMember: true, pubkey: PUB_A, @@ -428,14 +434,46 @@ test("shouldHideAgentFromMentions: hides member agents with an explicit not-invo ); }); -test("shouldHideAgentFromMentions: shows member agents with unknown invocability (not in directory)", () => { +test("shouldHideAgentFromMentions: hides unknown member agents while directories load", () => { assert.equal( shouldHideAgentFromMentions({ + ownerOnly: false, isAgent: true, isMember: true, pubkey: PUB_A, mentionableAgentPubkeys: new Set(), directoryAgentPubkeys: new Set(), + directoryReady: false, + }), + true, + ); +}); + +test("shouldHideAgentFromMentions: hides mentionable member agents while directories load", () => { + assert.equal( + shouldHideAgentFromMentions({ + ownerOnly: false, + isAgent: true, + isMember: true, + pubkey: PUB_A, + mentionableAgentPubkeys: new Set([PUB_A]), + directoryAgentPubkeys: new Set(), + directoryReady: false, + }), + true, + ); +}); + +test("shouldHideAgentFromMentions: shows non-agent members while directories load", () => { + assert.equal( + shouldHideAgentFromMentions({ + ownerOnly: false, + isAgent: false, + isMember: true, + pubkey: PUB_A, + mentionableAgentPubkeys: new Set(), + directoryAgentPubkeys: new Set([PUB_A]), + directoryReady: false, }), false, ); @@ -457,6 +495,7 @@ test("member agents: the allowed-list predicate is STRICTER than the hide rule", pubkey: PUB_A, mentionableAgentPubkeys: new Set(), directoryAgentPubkeys: new Set(), + ownerOnly: false, }; assert.equal( @@ -534,12 +573,26 @@ test("getAdmittedMemberAgentPubkeys: normalizes before gating and emitting", () ); }); +test("shouldHideAgentFromMentions: hides agents while owner policy loads", () => { + assert.equal( + shouldHideAgentFromMentions({ + isAgent: true, + pubkey: PUB_A, + mentionableAgentPubkeys: new Set([PUB_A]), + directoryReady: true, + ownerOnly: undefined, + }), + true, + ); +}); + test("shouldHideAgentFromMentions: normalizes the pubkey before lookup", () => { const mixedCase = "Ab".repeat(32); const normalized = mixedCase.toLowerCase(); assert.equal( shouldHideAgentFromMentions({ + ownerOnly: false, isAgent: true, isMember: true, pubkey: mixedCase, @@ -550,6 +603,58 @@ test("shouldHideAgentFromMentions: normalizes the pubkey before lookup", () => { ); }); +test("getAgentMentionAdmission: owner-only requires current verified ownership", () => { + const common = { + isAgent: true, + isManagedAgent: false, + pubkey: PUB_A, + currentPubkey: CURRENT_PUBKEY, + mentionableAgentPubkeys: new Set([PUB_A]), + directoryReady: true, + ownerOnly: true, + }; + + assert.equal( + getAgentMentionAdmission({ ...common, ownerPubkey: CURRENT_PUBKEY }), + "allow", + ); + assert.equal( + getAgentMentionAdmission({ ...common, ownerPubkey: OTHER_OWNER_PUBKEY }), + "deny", + ); + assert.equal( + getAgentMentionAdmission({ ...common, ownerPubkey: null }), + "unknown", + ); +}); + +test("getAgentMentionAdmission: unresolved directory state stays unknown", () => { + assert.equal( + getAgentMentionAdmission({ + isAgent: true, + isManagedAgent: false, + pubkey: PUB_A, + currentPubkey: CURRENT_PUBKEY, + ownerPubkey: CURRENT_PUBKEY, + mentionableAgentPubkeys: new Set([PUB_A]), + directoryReady: false, + ownerOnly: false, + }), + "unknown", + ); +}); + +test("filterAdmittedMentionPubkeys: rechecks agent admission without dropping people", () => { + assert.deepEqual( + filterAdmittedMentionPubkeys( + [PUB_A, PUB_B, PUB_C], + new Set([PUB_A, PUB_B]), + new Set([PUB_B]), + ), + [PUB_B, PUB_C], + ); +}); + test("coalesceAgentAutocompleteCandidates: keeps agents with the same persona id distinct", () => { const first = makeAgent({ pubkey: PUB_A, personaId: "pinky" }); const second = makeAgent({ diff --git a/desktop/src/features/agents/lib/agentAutocompleteEligibility.ts b/desktop/src/features/agents/lib/agentAutocompleteEligibility.ts index 2e0584c8944..741ac66ee9d 100644 --- a/desktop/src/features/agents/lib/agentAutocompleteEligibility.ts +++ b/desktop/src/features/agents/lib/agentAutocompleteEligibility.ts @@ -168,37 +168,158 @@ export function isAgentIdentityInAllowedList( ); } +export type AgentMentionAdmission = "allow" | "deny" | "unknown"; + +export function getAgentMentionAdmission({ + isAgent, + isManagedAgent, + isMember = false, + pubkey, + ownerPubkey, + currentPubkey, + mentionableAgentPubkeys, + directoryAgentPubkeys = new Set(), + directoryReady, + ownerOnly, +}: { + isAgent: boolean; + isManagedAgent: boolean; + isMember?: boolean; + pubkey: string; + ownerPubkey?: string | null; + currentPubkey?: string | null; + mentionableAgentPubkeys: ReadonlySet; + directoryAgentPubkeys?: ReadonlySet; + directoryReady: boolean; + ownerOnly: boolean | undefined; +}): AgentMentionAdmission { + if (!isAgent) return "allow"; + if (!directoryReady || ownerOnly === undefined) return "unknown"; + + const normalized = normalizePubkey(pubkey); + // Member (Option B): a channel-member agent with no relay directory + // (kind:10100) entry has unknown invocability rather than an explicit + // exclusion — treat it as mentionable rather than hiding every + // other-owner agent whose profile was never published. + const isLenientMember = + isMember && + !mentionableAgentPubkeys.has(normalized) && + !directoryAgentPubkeys.has(normalized); + if (!mentionableAgentPubkeys.has(normalized) && !isLenientMember) { + return "deny"; + } + if (!ownerOnly || isManagedAgent) return "allow"; + if (!ownerPubkey || !currentPubkey) return "unknown"; + + return normalizePubkey(ownerPubkey) === normalizePubkey(currentPubkey) + ? "allow" + : "deny"; +} + export function shouldHideAgentFromMentions({ isAgent, - isMember, + isManagedAgent = false, + isMember = false, pubkey, + ownerPubkey, + currentPubkey, mentionableAgentPubkeys, directoryAgentPubkeys, + directoryReady = true, + ownerOnly, }: { isAgent: boolean; - isMember: boolean; + isManagedAgent?: boolean; + isMember?: boolean; pubkey: string; + ownerPubkey?: string | null; + currentPubkey?: string | null; mentionableAgentPubkeys: ReadonlySet; - directoryAgentPubkeys: ReadonlySet; + directoryAgentPubkeys?: ReadonlySet; + directoryReady?: boolean; + ownerOnly: boolean | undefined; }) { - if (!isAgent) return false; - const normalized = normalizePubkey(pubkey); - // Invocable => always show. - if (mentionableAgentPubkeys.has(normalized)) return false; - // Non-member, non-invocable => hide (preserves prior behavior). - if (!isMember) return true; - // Member (Option B): hide only when we have an explicit not-invocable - // signal — a relay directory (kind:10100) entry that excludes us. - // Unknown invocability (not in directory) => show. - // - // NOTE: this assumes `directoryAgentPubkeys` and `mentionableAgentPubkeys` - // share the same source query (`relayAgentsQuery.data`), so directory - // presence without membership in `mentionableAgentPubkeys` is a real - // explicit-exclusion signal. If a future change sources the directory set - // from a different query, an agent that's directory-present but whose - // mentionability is still loading could be hidden prematurely — keep the - // two sets derived from the same query. - return directoryAgentPubkeys.has(normalized); + return ( + getAgentMentionAdmission({ + isAgent, + isManagedAgent, + isMember, + pubkey, + ownerPubkey, + currentPubkey, + mentionableAgentPubkeys, + directoryAgentPubkeys, + directoryReady, + ownerOnly, + }) !== "allow" + ); +} + +export function getAgentIdentityPubkeys({ + managedAgentPubkeys, + relayAgents, + members, + profileIsAgent, +}: { + managedAgentPubkeys: ReadonlySet; + relayAgents: readonly { pubkey: string }[]; + members: readonly { + pubkey: string; + isAgent?: boolean; + role?: string | null; + }[]; + profileIsAgent: (pubkey: string) => boolean; +}) { + return new Set([ + ...managedAgentPubkeys, + ...relayAgents.map(({ pubkey }) => normalizePubkey(pubkey)), + ...members + .filter( + (member) => + member.isAgent === true || + member.role === "bot" || + profileIsAgent(normalizePubkey(member.pubkey)), + ) + .map(({ pubkey }) => normalizePubkey(pubkey)), + ]); +} + +export function getAdmittedAgentPubkeys( + candidates: readonly { pubkey?: string; isAgent?: boolean }[], +) { + return new Set( + candidates.flatMap((candidate) => + candidate.isAgent && candidate.pubkey + ? [normalizePubkey(candidate.pubkey)] + : [], + ), + ); +} + +export function rememberSelectedAgentPubkeys( + target: Set, + selected: readonly { pubkey?: string; isAgent?: boolean }[], + selectionIsAgent: boolean, +) { + for (const candidate of selected) { + if (candidate.pubkey && (selectionIsAgent || candidate.isAgent === true)) { + target.add(normalizePubkey(candidate.pubkey)); + } + } +} + +export function filterAdmittedMentionPubkeys( + pubkeys: readonly string[], + agentIdentityPubkeys: ReadonlySet, + admittedAgentPubkeys: ReadonlySet, +) { + return pubkeys.filter((pubkey) => { + const normalized = normalizePubkey(pubkey); + return ( + !agentIdentityPubkeys.has(normalized) || + admittedAgentPubkeys.has(normalized) + ); + }); } /** @@ -235,6 +356,11 @@ export function getAdmittedMemberAgentPubkeys({ pubkey: normalized, mentionableAgentPubkeys, directoryAgentPubkeys, + // This helper only answers the invocability/membership question the + // picker's member branch depends on — owner-only visibility is + // already enforced by the picker itself before a pubkey ever + // reaches here. + ownerOnly: false, }) ) { continue; diff --git a/desktop/src/features/channels/hooks.ts b/desktop/src/features/channels/hooks.ts index 2d3dd22f05a..fb2b6864b36 100644 --- a/desktop/src/features/channels/hooks.ts +++ b/desktop/src/features/channels/hooks.ts @@ -484,24 +484,31 @@ export function useOpenDmMutation() { ); }, onSettled: () => { - void queryClient.invalidateQueries({ queryKey: channelsQueryKey }); + // The relay-returned DM is already in the cache. Mark the list stale so + // the normal live/poll refresh can reconcile it later without putting a + // full get_channels round-trip on the critical path to the conversation. + void queryClient.invalidateQueries({ + queryKey: channelsQueryKey, + refetchType: "none", + }); }, }); } /** - * Waits for any active channel-list refresh to settle, then restores a - * relay-returned channel to the shared cache before a caller depends on it for - * navigation. + * Reasserts a relay-returned channel in the shared cache before a caller + * depends on it for navigation. The open-DM mutation already made the relay + * write authoritative, so cancel any older list read and stay local rather + * than blocking on a read-after-write channel-list refresh. */ export function useUpsertCachedChannel() { const queryClient = useQueryClient(); return React.useCallback( async (channel: Channel) => { - await queryClient.refetchQueries({ + await queryClient.cancelQueries({ queryKey: channelsQueryKey, - type: "active", + exact: true, }); queryClient.setQueryData(channelsQueryKey, (current) => reconcileRefreshedCachedChannel(current, channel), diff --git a/desktop/src/features/forum/ui/ForumComposer.tsx b/desktop/src/features/forum/ui/ForumComposer.tsx index 6204186abe9..54e7d58c96c 100644 --- a/desktop/src/features/forum/ui/ForumComposer.tsx +++ b/desktop/src/features/forum/ui/ForumComposer.tsx @@ -58,6 +58,7 @@ export function ForumComposer({ const [isCompactExpanded, setIsCompactExpanded] = React.useState(!compact); const [isEmojiPickerOpen, setIsEmojiPickerOpen] = React.useState(false); const [isFormattingOpen, setIsFormattingOpen] = React.useState(false); + const [isSubmissionPending, setIsSubmissionPending] = React.useState(false); const [submitMode, setSubmitMode] = React.useState<"primary" | "secondary">( "primary", ); @@ -83,6 +84,7 @@ export function ForumComposer({ const disabledRef = React.useRef(disabled); const isSendingRef = React.useRef(isSending); const isUploadingRef = React.useRef(media.isUploading); + const isSubmissionPendingRef = React.useRef(false); const onSubmitRef = React.useRef(onSubmit); const onSecondarySubmitRef = React.useRef(onSecondarySubmit); const submitModeRef = React.useRef(submitMode); @@ -111,7 +113,7 @@ export function ForumComposer({ const richText = useRichTextEditor({ placeholder, - editable: !disabled, + editable: !disabled && !isSubmissionPending, mentionNames: mentions.knownNames, channelNames: channelLinks.knownChannelNames, messageLinkChannels: channelLinks.channels, @@ -139,6 +141,7 @@ export function ForumComposer({ // Native ProseMirror transactions — no markdown round-trip. const applyMentionInsert = React.useCallback( (suggestion: MentionSuggestion) => { + if (isSubmissionPendingRef.current) return; const { cursor } = richText.getPlainTextAndCursor(); const { replaceFromOffset, replaceToOffset, insertText } = mentions.insertMention(suggestion, cursor); @@ -157,6 +160,7 @@ export function ForumComposer({ const applyChannelInsert = React.useCallback( (suggestion: ChannelSuggestion) => { + if (isSubmissionPendingRef.current) return; const { cursor } = richText.getPlainTextAndCursor(); const { replaceFromOffset, replaceToOffset, insertText } = channelLinks.insertChannel(suggestion, cursor); @@ -175,7 +179,7 @@ export function ForumComposer({ const insertEmoji = React.useCallback( (emoji: string) => { - if (!richText.editor) return; + if (isSubmissionPendingRef.current || !richText.editor) return; richText.editor.chain().focus().insertContent(emoji).run(); setIsEmojiPickerOpen(false); mentions.clearMentions(); @@ -213,7 +217,7 @@ export function ForumComposer({ // ── Submit ────────────────────────────────────────────────────────── const submitMessage = React.useCallback( - (submitter = onSubmitRef.current) => { + async (submitter = onSubmitRef.current) => { const trimmed = contentRef.current.trim(); const currentPendingImeta = media.pendingImetaRef.current; const hasMedia = currentPendingImeta.length > 0; @@ -222,58 +226,68 @@ export function ForumComposer({ (!trimmed && !hasMedia) || disabledRef.current || isSendingRef.current || - isUploadingRef.current + isUploadingRef.current || + isSubmissionPendingRef.current ) { return; } - const pubkeys = mentions.extractMentionPubkeys(trimmed); - - // Reuse the shared send-path builder so forum/notes posts emit the same - // body + imeta as chat: generic files become `[filename](url)` links with a - // `filename` imeta tag (FileCard renderer), images/video stay inline. Send - // semantics use `undefined` for "no attachments" (no imeta tags emitted). - const { content: finalContent, mediaTags } = buildOutgoingMessage( - trimmed, - currentPendingImeta, - ); - - // Save draft state so we can restore on failure. - const savedContent = contentRef.current; - const savedImeta = [...currentPendingImeta]; - - setContent(""); - contentRef.current = ""; - richText.clearContent(); - media.setPendingImeta([]); - mentions.clearMentions(); + isSubmissionPendingRef.current = true; + setIsSubmissionPending(true); + mentions.cancelMentionAutocomplete(); channelLinks.clearChannels(); setIsEmojiPickerOpen(false); - - const result = submitter(finalContent, pubkeys, mediaTags); - const completeSubmission = () => { - setSubmitMode("primary"); - if (compact) setIsCompactExpanded(false); - }; - - // If onSubmit returns a promise, restore draft on failure. - if (result && typeof result.then === "function") { - result.then(completeSubmission).catch(() => { + try { + const pubkeys = await mentions.revalidateMentionPubkeys( + mentions.extractMentionPubkeys(trimmed), + ); + + // Reuse the shared send-path builder so forum/notes posts emit the same + // body + imeta as chat: generic files become `[filename](url)` links with a + // `filename` imeta tag (FileCard renderer), images/video stay inline. Send + // semantics use `undefined` for "no attachments" (no imeta tags emitted). + const { content: finalContent, mediaTags } = buildOutgoingMessage( + trimmed, + currentPendingImeta, + ); + + // Save draft state so we can restore on failure. + const savedContent = contentRef.current; + const savedImeta = [...currentPendingImeta]; + + setContent(""); + contentRef.current = ""; + richText.clearContent(); + media.setPendingImeta([]); + mentions.clearMentions(); + channelLinks.clearChannels(); + setIsEmojiPickerOpen(false); + + try { + await submitter(finalContent, pubkeys, mediaTags); + setSubmitMode("primary"); + if (compact) setIsCompactExpanded(false); + } catch { setContent(savedContent); contentRef.current = savedContent; richText.setContent(savedContent); media.setPendingImeta(savedImeta); if (compact) setIsCompactExpanded(true); - }); - } else { - completeSubmission(); + } + } catch { + // Keep the draft intact when authorization refresh fails. + } finally { + isSubmissionPendingRef.current = false; + setIsSubmissionPending(false); } }, [ compact, media.pendingImetaRef, media.setPendingImeta, + mentions.cancelMentionAutocomplete, mentions.extractMentionPubkeys, + mentions.revalidateMentionPubkeys, mentions.clearMentions, channelLinks.clearChannels, richText.clearContent, @@ -375,9 +389,16 @@ export function ForumComposer({ const sendDisabled = React.useMemo( () => disabled || + isSubmissionPending || media.isUploading || (content.trim().length === 0 && media.pendingImeta.length === 0), - [disabled, media.isUploading, content, media.pendingImeta.length], + [ + disabled, + isSubmissionPending, + media.isUploading, + content, + media.pendingImeta.length, + ], ); const hasComposerContent = content.trim().length > 0 || @@ -448,15 +469,30 @@ export function ForumComposer({ "relative rounded-2xl border border-input bg-card px-3 py-2 sm:px-4", className, )} + inert={isSubmissionPending ? true : undefined} onBlurCapture={handleFormBlur} onDragEnter={(event) => { + if (isSubmissionPending) { + event.preventDefault(); + return; + } expandCompactComposer(); media.handleDragEnter(event); }} onDragLeave={media.handleDragLeave} - onDragOver={media.handleDragOver} - onDrop={(e) => { - void media.handleDrop(e); + onDragOver={(event) => { + if (isSubmissionPending) { + event.preventDefault(); + return; + } + media.handleDragOver(event); + }} + onDrop={(event) => { + if (isSubmissionPending) { + event.preventDefault(); + return; + } + void media.handleDrop(event); }} onFocusCapture={expandCompactComposer} onSubmit={handleSubmit} @@ -466,7 +502,7 @@ export function ForumComposer({ @@ -496,7 +532,15 @@ export function ForumComposer({ position={autocompletePosition} /> - +
+ +
{/* biome-ignore lint/a11y/noStaticElementInteractions: keydown handler bridges Tiptap editor to autocomplete and submit */}
{onCancel ? (