diff --git a/AGENTS.md b/AGENTS.md index b1f11bd3db..6d97f4dfd9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -226,7 +226,9 @@ channel and derives its containing root. An optional `thread` parameter is accepted only when it matches that derived root. The explicit `--channel --event ` form remains available. -All reads return sig-stripped JSON arrays; all writes return +All event reads return normalized JSON arrays. Normal output preserves the seven +canonical signed Nostr event fields (`id`, `pubkey`, `kind`, `content`, +`created_at`, `tags`, `sig`); all writes return `{event_id, accepted, message}`; creates add the entity ID. Exit codes: 0=ok, 1=input error, 2=network/relay, 3=auth, 4=other, 5=write conflict (NIP-33 LWW). diff --git a/admin-web/src/styles.css b/admin-web/src/styles.css index 93ebceeb14..8f0ba3d6c3 100644 --- a/admin-web/src/styles.css +++ b/admin-web/src/styles.css @@ -195,6 +195,7 @@ h1 { .feedback-filters label { display: grid; gap: 0.45rem; + min-width: 0; } .feedback-filters label > span { @@ -221,6 +222,8 @@ h1 { } .feedback-filters select { + width: 100%; + min-width: 0; padding: 0 2.25rem 0 1rem; } diff --git a/admin-web/tests/routes.spec.ts b/admin-web/tests/routes.spec.ts index 3c965dd2d8..ca6abe7143 100644 --- a/admin-web/tests/routes.spec.ts +++ b/admin-web/tests/routes.spec.ts @@ -224,6 +224,74 @@ test("feedback can be searched and filtered by community and time", async ({ await expect(page.getByText("Calls are much more reliable")).toHaveCount(0); }); +test("feedback filters keep long community names usable", async ({ page }) => { + await page.route("**/api/admin/v1/feedback", (route) => + route.fulfill({ + contentType: "application/json", + body: JSON.stringify([ + { + id: "long-community", + communityId: "long-community", + communityHost: `${"long-community-name.".repeat(4)}buzz.example.com`, + submitterPubkey: "21".repeat(32), + category: "bug", + bodySummary: "The filter row stays within its container", + receivedAt: new Date().toISOString(), + }, + ]), + }), + ); + + for (const viewport of [ + { name: "desktop", width: 1200 }, + { name: "mobile", width: 720 }, + ]) { + await test.step(viewport.name, async () => { + await page.setViewportSize({ width: viewport.width, height: 720 }); + await page.goto("/feedback"); + + const filters = page.locator(".feedback-filters"); + const search = page.getByRole("searchbox", { name: "Search feedback" }); + const community = page.getByRole("combobox", { name: "Community" }); + const status = page.getByLabel("Status"); + await expect(filters).toBeVisible(); + await expect(community).toBeVisible(); + await expect(status).toBeVisible(); + + const [filtersBox, searchBox, communityBox, statusBox] = + await Promise.all([ + filters.boundingBox(), + search.boundingBox(), + community.boundingBox(), + status.boundingBox(), + ]); + if (!filtersBox || !searchBox || !communityBox || !statusBox) { + throw new Error("feedback filter bounds were unavailable"); + } + expect(statusBox.x + statusBox.width).toBeLessThanOrEqual( + filtersBox.x + filtersBox.width, + ); + + if (viewport.name === "desktop") { + const rootFontSize = await page.evaluate(() => + Number.parseFloat( + getComputedStyle(document.documentElement).fontSize, + ), + ); + expect(communityBox.width).toBeGreaterThanOrEqual(14 * rootFontSize); + } else { + expect(Math.abs(communityBox.width - searchBox.width)).toBeLessThan(1); + } + + const pageWidths = await page.evaluate(() => ({ + client: document.documentElement.clientWidth, + scroll: document.documentElement.scrollWidth, + })); + expect(pageWidths.scroll).toBe(pageWidths.client); + }); + } +}); + test("feedback status is stored locally by feedback id", async ({ page }) => { await page.route("**/api/admin/v1/feedback", (route) => route.fulfill({ diff --git a/crates/buzz-cli/TESTING.md b/crates/buzz-cli/TESTING.md index 81ed36f62b..b7fa06d203 100644 --- a/crates/buzz-cli/TESTING.md +++ b/crates/buzz-cli/TESTING.md @@ -428,7 +428,8 @@ buzz workflows delete --workflow "$WF_ID" | jq . ```bash buzz feed get | jq . buzz feed get --limit 5 | jq . -# Expected: [{id,pubkey,kind,content,created_at,tags}] — sig-stripped, sorted newest-first +# Expected: complete signed Nostr events with +# {id,pubkey,kind,content,created_at,sig,tags}, sorted newest-first ``` ### 6.11 Forum & Voting diff --git a/crates/buzz-cli/src/client.rs b/crates/buzz-cli/src/client.rs index 3d727cd43d..76d0e6fb95 100644 --- a/crates/buzz-cli/src/client.rs +++ b/crates/buzz-cli/src/client.rs @@ -1323,20 +1323,24 @@ fn to_ws_url(http_url: &str) -> String { .replace("http://", "ws://") } -/// Normalize raw event JSON array into consistent shape. -/// Each event becomes: {id, pubkey, kind, content, created_at, tags} +/// Normalize raw event JSON array into the canonical Nostr event shape. +/// String signatures are preserved; absent or non-string signatures remain absent. pub fn normalize_events(events: &[serde_json::Value]) -> String { let normalized: Vec = events .iter() .map(|e| { - serde_json::json!({ + let mut event = serde_json::json!({ "id": e.get("id").and_then(|v| v.as_str()).unwrap_or(""), "pubkey": e.get("pubkey").and_then(|v| v.as_str()).unwrap_or(""), "kind": e.get("kind").and_then(|v| v.as_u64()).unwrap_or(0), "content": e.get("content").and_then(|v| v.as_str()).unwrap_or(""), "created_at": e.get("created_at").and_then(|v| v.as_u64()).unwrap_or(0), "tags": e.get("tags").cloned().unwrap_or(serde_json::json!([])), - }) + }); + if let Some(sig) = e.get("sig").and_then(|v| v.as_str()) { + event["sig"] = serde_json::json!(sig); + } + event }) .collect(); serde_json::to_string(&normalized).unwrap_or_default() @@ -2325,10 +2329,42 @@ mod retry_policy_tests { mod tests { use super::{ advance_query_cursor, create_response_with_id_if_accepted, extract_relay_response_field, - BuzzClient, + normalize_events, BuzzClient, }; use nostr::{EventBuilder, Keys, Kind, Tag}; + #[test] + fn normalize_events_preserves_the_complete_signed_event_shape() { + let signed_event = EventBuilder::new(Kind::TextNote, "signed content") + .tags([Tag::parse(["h", "channel-id"]).unwrap()]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let mut event = serde_json::to_value(&signed_event).unwrap(); + event["relay_internal"] = serde_json::json!("excluded"); + + let output: Vec = + serde_json::from_str(&normalize_events(&[event])).unwrap(); + let normalized = &output[0]; + let round_tripped: nostr::Event = serde_json::from_value(normalized.clone()).unwrap(); + + assert_eq!(round_tripped, signed_event); + round_tripped.verify().unwrap(); + assert!(normalized.get("sig").is_some()); + assert!(normalized.get("relay_internal").is_none()); + } + + #[test] + fn normalize_events_omits_missing_or_non_string_signatures() { + let output: Vec = serde_json::from_str(&normalize_events(&[ + serde_json::json!({}), + serde_json::json!({"sig": 42}), + ])) + .unwrap(); + + assert!(output[0].get("sig").is_none()); + assert!(output[1].get("sig").is_none()); + } + #[test] fn query_cursor_uses_last_events_composite_sort_key() { let mut filter = serde_json::json!({"kinds": [39000], "limit": 500}); diff --git a/crates/buzz-cli/src/commands/feed.rs b/crates/buzz-cli/src/commands/feed.rs index d3d5c7f81a..d5e1dae4b2 100644 --- a/crates/buzz-cli/src/commands/feed.rs +++ b/crates/buzz-cli/src/commands/feed.rs @@ -5,6 +5,27 @@ use crate::error::CliError; const VALID_FEED_TYPES: &[&str] = &["mentions", "needs_action", "activity", "agent_activity"]; +fn format_events(normalized: &str, format: &crate::OutputFormat) -> String { + match format { + crate::OutputFormat::Compact => { + let events: Vec = + serde_json::from_str(normalized).unwrap_or_default(); + let compact: Vec = events + .iter() + .map(|e| { + serde_json::json!({ + "id": e.get("id").cloned().unwrap_or_default(), + "content": e.get("content").cloned().unwrap_or_default(), + "created_at": e.get("created_at").cloned().unwrap_or_default(), + }) + }) + .collect(); + serde_json::to_string(&compact).unwrap_or_default() + } + crate::OutputFormat::Json => normalized.to_string(), + } +} + /// Get activity feed — query events mentioning our pubkey (via p-tag). pub async fn cmd_get_feed( client: &BuzzClient, @@ -42,25 +63,7 @@ pub async fn cmd_get_feed( let mut events: Vec = serde_json::from_str(&resp).unwrap_or_default(); events.sort_by_key(|e| Reverse(e.get("created_at").and_then(|v| v.as_u64()).unwrap_or(0))); let normalized = normalize_events(&events); - let output = match format { - crate::OutputFormat::Compact => { - let evts: Vec = - serde_json::from_str(&normalized).unwrap_or_default(); - let compact: Vec = evts - .iter() - .map(|e| { - serde_json::json!({ - "id": e.get("id").cloned().unwrap_or_default(), - "content": e.get("content").cloned().unwrap_or_default(), - "created_at": e.get("created_at").cloned().unwrap_or_default(), - }) - }) - .collect(); - serde_json::to_string(&compact).unwrap_or_default() - } - crate::OutputFormat::Json => normalized, - }; - println!("{output}"); + println!("{}", format_events(&normalized, format)); Ok(()) } @@ -78,3 +81,35 @@ pub async fn dispatch( } => cmd_get_feed(client, since, limit, types.as_deref(), format).await, } } + +#[cfg(test)] +mod tests { + use super::format_events; + + #[test] + fn compact_event_format_remains_the_three_key_contract() { + let normalized = serde_json::json!([{ + "id": "a".repeat(64), + "pubkey": "b".repeat(64), + "kind": 9, + "content": "compact content", + "created_at": 1_787_754_972_u64, + "tags": [["p", "c".repeat(64)]], + "sig": "d".repeat(128), + }]) + .to_string(); + + let output: Vec = + serde_json::from_str(&format_events(&normalized, &crate::OutputFormat::Compact)) + .unwrap(); + + assert_eq!( + output[0], + serde_json::json!({ + "id": "a".repeat(64), + "content": "compact content", + "created_at": 1_787_754_972_u64, + }) + ); + } +} diff --git a/crates/buzz-cli/src/commands/messages.rs b/crates/buzz-cli/src/commands/messages.rs index ea273336e3..9f41fbf751 100644 --- a/crates/buzz-cli/src/commands/messages.rs +++ b/crates/buzz-cli/src/commands/messages.rs @@ -1057,7 +1057,7 @@ pub async fn dispatch( mod tests { use super::{ channel_id_from_event, cmd_get_thread, event_mention_pubkeys, find_root_from_tags, - match_profiles_by_name, merge_message_mentions, missing_members, + format_events, match_profiles_by_name, merge_message_mentions, missing_members, normalize_explicit_mentions, parse_member_pubkeys, resolve_names_to_pubkeys, resolve_thread_target, thread_ref_from_event, thread_ref_from_parent_tags, BuzzClient, CliError, Uuid, @@ -1078,6 +1078,33 @@ mod tests { const PK_VALID_B: &str = "c6237ef84fa537c78dcee78efd2d4e59f728859c7f194da42ac51ededfa0be05"; const PK_VALID_C: &str = "f4a42a97e594b77bdbd8ee35191c8b28a94a4cb871d96f32921558275421fb68"; + #[test] + fn compact_event_format_remains_the_three_key_contract() { + let normalized = serde_json::json!([{ + "id": ID_A, + "pubkey": PUBKEY, + "kind": 9, + "content": "compact content", + "created_at": 1_787_754_972_u64, + "tags": [["h", "channel-id"]], + "sig": "d".repeat(128), + }]) + .to_string(); + + let output: Vec = + serde_json::from_str(&format_events(&normalized, &crate::OutputFormat::Compact)) + .unwrap(); + + assert_eq!( + output[0], + serde_json::json!({ + "id": ID_A, + "content": "compact content", + "created_at": 1_787_754_972_u64, + }) + ); + } + #[tokio::test] async fn malformed_channel_is_rejected_before_thread_fetch() { let client = diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index a6cd77c979..b37dcedff8 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -4327,150 +4327,6 @@ impl Db { Ok(result.rows_affected()) } - /// Atomically replace a replaceable event: NIP-16 kinds (0, 3, 41, 10000–19999) - /// and NIP-29 discovery state (39000–39002, called from side_effects.rs). - /// - /// Keeps only the event with the highest `created_at` per (kind, pubkey, channel_id). - /// Same-second ties are broken by lowest event `id` (NIP-16 deterministic ordering). - /// Returns `(event, false)` for stale writes and duplicate IDs — callers should - /// skip fan-out/dispatch when `was_inserted` is false. - #[datastore_span(name = "replace_addressable_event", system = "postgresql")] - pub async fn replace_addressable_event( - &self, - community_id: CommunityId, - event: &nostr::Event, - channel_id: Option, - ) -> Result<(StoredEvent, bool)> { - let kind_i32 = buzz_core::kind::event_kind_i32(event); - let pubkey_bytes = event.pubkey.to_bytes(); - let created_at_secs = event.created_at.as_secs() as i64; - let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0) - .ok_or(DbError::InvalidTimestamp(created_at_secs))?; - - // Collisions only cause extra serialization; they cannot change behavior. - let lock_key = replaceable::event_replacement_lock_key( - community_id, - kind_i32, - pubkey_bytes.as_slice(), - channel_id.as_ref().map(|id| id.as_bytes().as_slice()), - ); - - let (mut tx, transaction_timer) = observability::begin_transaction( - &self.pool, - observability::TransactionOperation::ReplaceAddressableEvent, - ) - .await?; - transaction_timer - .observe(async { - - // Serialize all writers for the same (kind, pubkey, channel_id) tuple. - // Advisory lock is transaction-scoped — released on commit/rollback. - observability::observe_advisory_lock( - observability::LockType::Replacement, - sqlx::query("SELECT pg_advisory_xact_lock($1)") - .bind(lock_key) - .execute(&mut *tx), - ) - .await?; - - // Check for the newest existing event. ORDER BY + LIMIT 1 is defensive against - // historical data where prior bugs may have left multiple live rows. - let existing: Option<(chrono::DateTime, Vec)> = sqlx::query_as( - "SELECT created_at, id FROM events \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ - AND channel_id IS NOT DISTINCT FROM $4 \ - AND deleted_at IS NULL \ - ORDER BY created_at DESC, id ASC LIMIT 1", - ) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(channel_id) - .fetch_optional(&mut *tx) - .await?; - - // Stale-write protection: reject if incoming is not newer. - // NIP-16: created_at is second-resolution. On same-second tie, lowest - // event id (lexicographic) wins — deterministic across relays. - let incoming_id = event.id.as_bytes().as_slice(); - if let Some((existing_ts, existing_id)) = existing { - let dominated = created_at < existing_ts - || (created_at == existing_ts && incoming_id >= existing_id.as_slice()); - if dominated { - tx.rollback().await?; - let received_at = chrono::Utc::now(); - return Ok(( - StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), - false, - )); - } - } - - // Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety. - sqlx::query( - "UPDATE events SET deleted_at = NOW() \ - WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ - AND channel_id IS NOT DISTINCT FROM $4 \ - AND deleted_at IS NULL", - ) - .bind(community_id.as_uuid()) - .bind(kind_i32) - .bind(pubkey_bytes.as_slice()) - .bind(channel_id) - .execute(&mut *tx) - .await?; - - // Insert the new event inside the same transaction. - let sig_bytes = event.sig.serialize(); - let tags_json = serde_json::to_value(&event.tags)?; - let received_at = chrono::Utc::now(); - let d_tag = crate::event::extract_d_tag(event); - - let insert_result = sqlx::query( - "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \ - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \ - ON CONFLICT DO NOTHING", - ) - .bind(community_id.as_uuid()) - .bind(event.id.as_bytes().as_slice()) - .bind(pubkey_bytes.as_slice()) - .bind(created_at) - .bind(kind_i32) - .bind(&tags_json) - .bind(&event.content) - .bind(sig_bytes.as_slice()) - .bind(received_at) - .bind(channel_id) - .bind(d_tag.as_deref()) - .execute(&mut *tx) - .await?; - - let was_inserted = insert_result.rows_affected() > 0; - if !was_inserted { - // ON CONFLICT fired — the event ID already exists. Rollback the - // soft-delete so we don't lose the previous replaceable event. - tx.rollback().await?; - return Ok(( - StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), - false, - )); - } - - // 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?; - - tx.commit().await?; - - Ok(( - StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), - true, - )) - }) - .await - } - /// Returns whether the relay-authored NIP-43 snapshot is absent or differs /// from the canonical membership rows for `community_id`. /// @@ -4743,7 +4599,7 @@ mod tests { use super::*; use buzz_core::CommunityId; use sqlx::postgres::PgPoolOptions; - use sqlx::{Acquire, PgPool}; + use sqlx::PgPool; use uuid::Uuid; const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; @@ -4876,226 +4732,6 @@ mod tests { drop_scratch_db(&admin, pool, &scratch_name).await; } - #[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(); - let owner_keys = Keys::generate(); - seed_community_channel(&pool, community_uuid, channel, &owner_keys).await; - let community = CommunityId::from_uuid(community_uuid); - let member = owner_keys.public_key().to_hex(); - let tags = || { - vec![ - Tag::parse(["d", channel.to_string().as_str()]).expect("d tag"), - Tag::parse(["p", member.as_str(), "", "owner"]).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 stale_legacy_roster_cannot_replace_new_locked_snapshot() { - use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; - - let admin = PgPool::connect(&admin_url().await) - .await - .expect("connect admin"); - let (setup_pool, scratch_name) = create_scratch_db(&admin, "mixed_roster_writer").await; - let base_url = admin_url().await; - let slash = base_url.rfind('/').expect("database URL has path segment"); - let scratch_url = format!("{}/{}", &base_url[..slash], scratch_name); - let pool = PgPoolOptions::new() - .max_connections(1) - .acquire_timeout(Duration::from_secs(1)) - .connect(&scratch_url) - .await - .expect("connect one-connection scratch pool"); - setup_pool.close().await; - let db = Db::from_pool(pool.clone()); - let community_uuid = Uuid::new_v4(); - let community = CommunityId::from_uuid(community_uuid); - let channel = Uuid::new_v4(); - let relay_keys = Keys::generate(); - let owner_keys = Keys::generate(); - let owner = owner_keys.public_key().to_bytes(); - seed_community_channel(&pool, community_uuid, channel, &owner_keys).await; - - // This is the old pod's unlocked capture A. It remains in process memory - // while a role-only canonical mutation advances and the new pod publishes B. - let base = Timestamp::now().as_secs(); - let roster = |members: &[(&[u8], &str)], timestamp| { - let tags = - std::iter::once(Tag::parse(["d", channel.to_string().as_str()]).expect("d tag")) - .chain(members.iter().map(|(member, role)| { - Tag::parse(["p", hex::encode(member).as_str(), "", *role]).expect("p tag") - })) - .collect::>(); - EventBuilder::new(Kind::Custom(39002), "") - .tags(tags) - .custom_created_at(Timestamp::from(timestamp)) - .sign_with_keys(&relay_keys) - .expect("sign roster") - }; - - let newcomer = Keys::generate().public_key().to_bytes(); - sqlx::query( - "INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) \ - VALUES ($1, $2, $3, 'member', $4)", - ) - .bind(community_uuid) - .bind(channel) - .bind(newcomer.as_slice()) - .bind(owner.as_slice()) - .execute(&pool) - .await - .expect("seed member before legacy capture"); - let stale_a = roster( - &[(owner.as_slice(), "owner"), (newcomer.as_slice(), "member")], - base + 2, - ); - - sqlx::query( - "UPDATE channel_members SET role = 'admin' \ - WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", - ) - .bind(community_uuid) - .bind(channel) - .bind(newcomer.as_slice()) - .execute(&pool) - .await - .expect("commit newer canonical role"); - - let relay_pubkey = relay_keys.public_key().to_bytes(); - let mut snapshot = db - .lock_member_snapshot(community, channel, &relay_pubkey) - .await - .expect("new writer captures locked roster B"); - let fresh_b = roster( - &[(owner.as_slice(), "owner"), (newcomer.as_slice(), "admin")], - base + 1, - ); - assert!( - snapshot - .replace_member_event(community, channel, &fresh_b) - .await - .expect("new writer publishes B") - .1 - ); - snapshot - .release() - .await - .expect("commit B and release locks"); - - // The legacy canonical path takes the replacement key, soft-deletes B, - // then attempts its newer-timestamp stale A. Migration 0032 rejects the - // INSERT; transaction rollback must restore B. A one-connection pool - // proves the lock order does not turn this compatibility path into a - // self-deadlock. - let error = tokio::time::timeout( - Duration::from_secs(3), - db.replace_addressable_event(community, &stale_a, Some(channel)), - ) - .await - .expect("legacy replacement must not deadlock") - .expect_err("stale captured roster A must be rejected"); - assert!( - matches!( - error, - DbError::Sqlx(sqlx::Error::Database(ref db_error)) - if db_error.code().as_deref() == Some("23514") - ), - "expected roster fence check violation, got {error:?}" - ); - - let live_ids: Vec> = sqlx::query_scalar( - "SELECT id FROM events WHERE community_id=$1 AND channel_id=$2 \ - AND kind=39002 AND pubkey=$3 AND deleted_at IS NULL", - ) - .bind(community_uuid) - .bind(channel) - .bind(relay_pubkey.as_slice()) - .fetch_all(&pool) - .await - .expect("load live roster heads"); - assert_eq!(live_ids, vec![fresh_b.id.as_bytes().to_vec()]); - let stale_rows: i64 = - sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2") - .bind(community_uuid) - .bind(stale_a.id.as_bytes().as_slice()) - .fetch_one(&pool) - .await - .expect("count rejected stale roster"); - assert_eq!(stale_rows, 0, "stale roster insert must roll back"); - - drop_scratch_db(&admin, pool, &scratch_name).await; - } - #[tokio::test] #[ignore = "requires Postgres"] async fn desired_schema_rejects_stale_legacy_roster_role() { @@ -5190,602 +4826,6 @@ mod tests { 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() { - use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; - - let db = setup_db().await; - let community = CommunityId::from_uuid(make_community(&db.pool).await); - let keys = Keys::generate(); - let d_tag = format!("read-state:{}", "a".repeat(32)); - let tags = vec![ - Tag::parse(["d", d_tag.as_str()]).expect("d tag"), - Tag::parse(["t", "read-state"]).expect("t tag"), - ]; - let base = Timestamp::now().as_secs(); - let old = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "old") - .tags(tags.clone()) - .custom_created_at(Timestamp::from(base)) - .sign_with_keys(&keys) - .expect("sign old"); - let new = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "new") - .tags(tags) - .custom_created_at(Timestamp::from(base + 1)) - .sign_with_keys(&keys) - .expect("sign new"); - - assert!( - db.replace_parameterized_event(community, &old, &d_tag, None) - .await - .expect("insert old") - .1 - ); - assert!( - db.replace_parameterized_event(community, &new, &d_tag, None) - .await - .expect("replace with new") - .1 - ); - - let rows: i64 = sqlx::query_scalar( - "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3", - ) - .bind(community.as_uuid()) - .bind(keys.public_key().to_bytes()) - .bind(&d_tag) - .fetch_one(&db.pool) - .await - .expect("count NIP-RS rows"); - assert_eq!(rows, 1, "superseded payload must be physically deleted"); - - sqlx::query( - "UPDATE events SET deleted_at=NOW() WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3", - ) - .bind(community.as_uuid()) - .bind(keys.public_key().to_bytes()) - .bind(&d_tag) - .execute(&db.pool) - .await - .expect("simulate NIP-09 coordinate deletion"); - - assert!( - !db.replace_parameterized_event(community, &old, &d_tag, None) - .await - .expect("replay old") - .1 - ); - let live: i64 = sqlx::query_scalar( - "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL", - ) - .bind(community.as_uuid()) - .bind(keys.public_key().to_bytes()) - .bind(&d_tag) - .fetch_one(&db.pool) - .await - .expect("count live NIP-RS rows"); - assert_eq!(live, 0, "watermark must block stale resurrection"); - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn nip_rs_transaction_operation_restores_hard_delete_opt_in() { - use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; - - let db = setup_db().await; - let community = CommunityId::from_uuid(make_community(&db.pool).await); - let keys = Keys::generate(); - let base = Timestamp::now().as_secs(); - let replace_d_tag = format!("read-state:{}", "b".repeat(32)); - let victim_d_tag = format!("read-state:{}", "c".repeat(32)); - let event = |d_tag: &str, content: &str, timestamp: u64| { - EventBuilder::new( - Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), - content, - ) - .tags(vec![ - Tag::parse(["d", d_tag]).expect("d tag"), - Tag::parse(["t", "read-state"]).expect("t tag"), - ]) - .custom_created_at(Timestamp::from(timestamp)) - .sign_with_keys(&keys) - .expect("sign read state") - }; - let old = event(&replace_d_tag, "old", base); - let new = event(&replace_d_tag, "new", base + 1); - let victim = event(&victim_d_tag, "victim", base); - - assert!( - db.replace_parameterized_event(community, &old, &replace_d_tag, None) - .await - .expect("insert old head") - .1 - ); - assert!( - db.replace_parameterized_event(community, &victim, &victim_d_tag, None) - .await - .expect("insert victim head") - .1 - ); - - let mut tx = db - .begin_transaction() - .await - .expect("begin caller transaction"); - let result = db - .replace_parameterized_event_in_transaction( - &mut tx, - community, - &new, - &replace_d_tag, - None, - replaceable::ParameterizedReplacePrecondition::Unconditional, - ) - .await - .expect("replace inside caller transaction"); - assert_eq!( - result.status, - replaceable::ParameterizedReplaceStatus::Inserted - ); - - let leaked: Option = sqlx::query_scalar( - "SELECT NULLIF(current_setting('buzz.nip_rs_hard_delete', true), '')", - ) - .fetch_one(&mut *tx) - .await - .expect("read hard-delete opt-in after replacement"); - assert_ne!(leaked.as_deref(), Some("on")); - - let unauthorized = sqlx::query( - "DELETE FROM events WHERE community_id=$1 AND kind=30078 \ - AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL", - ) - .bind(community.as_uuid()) - .bind(keys.public_key().to_bytes()) - .bind(&victim_d_tag) - .execute(&mut *tx) - .await; - assert!( - unauthorized.is_err(), - "replacement opt-in must not authorize later caller SQL" - ); - tx.rollback().await.expect("roll back caller transaction"); - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn parameterized_replacement_in_existing_transaction_honors_revision_and_rollback() { - use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; - - let db = setup_db().await; - let community = CommunityId::from_uuid(make_community(&db.pool).await); - let keys = Keys::generate(); - let d_tag = format!("transactional-project-{}", Uuid::new_v4().simple()); - let base = Timestamp::now().as_secs(); - let event = |content: &str, timestamp: u64| { - EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content) - .tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")]) - .custom_created_at(Timestamp::from(timestamp)) - .sign_with_keys(&keys) - .expect("sign project") - }; - let old = event("old", base); - let new = event("new", base + 1); - - assert!( - db.replace_parameterized_event(community, &old, &d_tag, None) - .await - .expect("insert old head") - .1 - ); - - let mut tx = db.begin_transaction().await.expect("begin replacement tx"); - let outcome = db - .replace_parameterized_event_in_transaction( - &mut tx, - community, - &new, - &d_tag, - None, - replaceable::ParameterizedReplacePrecondition::ExpectedRevision( - old.id.as_bytes().as_slice(), - ), - ) - .await - .expect("replace inside caller transaction"); - assert_eq!( - outcome.status, - replaceable::ParameterizedReplaceStatus::Inserted - ); - tx.rollback().await.expect("roll back replacement tx"); - - let live_id: Vec = sqlx::query_scalar( - "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ - AND d_tag=$4 AND deleted_at IS NULL", - ) - .bind(community.as_uuid()) - .bind(buzz_core::kind::KIND_PROJECT as i32) - .bind(keys.public_key().to_bytes()) - .bind(&d_tag) - .fetch_one(&db.pool) - .await - .expect("load live head after rollback"); - assert_eq!(live_id, old.id.as_bytes().to_vec()); - - let mut tx = db - .begin_transaction() - .await - .expect("begin stale revision tx"); - let mismatch = db - .replace_parameterized_event_in_transaction( - &mut tx, - community, - &new, - &d_tag, - None, - replaceable::ParameterizedReplacePrecondition::ExpectedRevision( - [0x42; 32].as_slice(), - ), - ) - .await - .expect("evaluate stale revision"); - assert_eq!( - mismatch.status, - replaceable::ParameterizedReplaceStatus::RevisionMismatch - ); - tx.rollback().await.expect("roll back stale revision tx"); - - let missing_d_tag = format!("missing-project-{}", Uuid::new_v4().simple()); - let missing = EventBuilder::new( - Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), - "missing", - ) - .tags(vec![ - Tag::parse(["d", missing_d_tag.as_str()]).expect("missing d tag") - ]) - .custom_created_at(Timestamp::from(base + 2)) - .sign_with_keys(&keys) - .expect("sign missing project"); - let mut tx = db - .begin_transaction() - .await - .expect("begin missing revision tx"); - let missing_result = db - .replace_parameterized_event_in_transaction( - &mut tx, - community, - &missing, - &missing_d_tag, - None, - replaceable::ParameterizedReplacePrecondition::ExpectedRevision( - [0x24; 32].as_slice(), - ), - ) - .await - .expect("evaluate missing revision"); - assert_eq!( - missing_result.status, - replaceable::ParameterizedReplaceStatus::RevisionMissing - ); - tx.rollback().await.expect("roll back missing revision tx"); - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn parameterized_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_parameterized").await; - let db = Db::from_pool(pool.clone()); - let community = CommunityId::from_uuid(make_community(&pool).await); - let keys = Keys::generate(); - let mentioned = Keys::generate().public_key().to_hex(); - let d_tag = format!("mention-project-{}", Uuid::new_v4().simple()); - let tags = || { - vec![ - Tag::parse(["d", d_tag.as_str()]).expect("d tag"), - Tag::parse(["p", mentioned.as_str()]).expect("p tag"), - ] - }; - let base = Timestamp::now().as_secs(); - let old = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), "old") - .tags(tags()) - .custom_created_at(Timestamp::from(base)) - .sign_with_keys(&keys) - .expect("sign old project"); - let new = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), "new") - .tags(tags()) - .custom_created_at(Timestamp::from(base + 1)) - .sign_with_keys(&keys) - .expect("sign new project"); - - assert!( - db.replace_parameterized_event(community, &old, &d_tag, None) - .await - .expect("insert old project") - .1 - ); - 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 mut tx = db - .begin_transaction() - .await - .expect("begin caller transaction"); - let error = db - .replace_parameterized_event_in_transaction( - &mut tx, - community, - &new, - &d_tag, - None, - replaceable::ParameterizedReplacePrecondition::Unconditional, - ) - .await - .expect_err("mention failure must fail replacement"); - assert!(error.to_string().contains("injected mention failure")); - - let probe: i32 = sqlx::query_scalar("SELECT 1") - .fetch_one(&mut *tx) - .await - .expect("inner failure must leave caller transaction usable"); - assert_eq!(probe, 1); - - let live_id: Vec = sqlx::query_scalar( - "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ - AND d_tag=$4 AND deleted_at IS NULL", - ) - .bind(community.as_uuid()) - .bind(buzz_core::kind::KIND_PROJECT as i32) - .bind(keys.public_key().to_bytes()) - .bind(&d_tag) - .fetch_one(&mut *tx) - .await - .expect("load live project after failed indexing"); - assert_eq!(live_id, old.id.as_bytes().to_vec()); - 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(&mut *tx) - .await - .expect("count rolled-back project"); - assert_eq!(new_rows, 0); - tx.commit().await.expect("commit usable caller transaction"); - - drop_scratch_db(&admin, pool, &scratch_name).await; - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn parameterized_duplicate_restores_live_head_inside_caller_transaction() { - use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; - - let db = setup_db().await; - let community = CommunityId::from_uuid(make_community(&db.pool).await); - let keys = Keys::generate(); - let d_tag = format!("duplicate-project-{}", Uuid::new_v4().simple()); - let base = Timestamp::now().as_secs(); - let event = |content: &str, timestamp: u64| { - EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content) - .tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")]) - .custom_created_at(Timestamp::from(timestamp)) - .sign_with_keys(&keys) - .expect("sign project") - }; - let old = event("old-live-head", base); - let duplicate = event("soft-deleted-duplicate", base + 1); - - assert!( - db.replace_parameterized_event(community, &duplicate, &d_tag, None) - .await - .expect("insert future duplicate") - .1 - ); - sqlx::query("UPDATE events SET deleted_at=NOW() WHERE community_id=$1 AND id=$2") - .bind(community.as_uuid()) - .bind(duplicate.id.as_bytes().as_slice()) - .execute(&db.pool) - .await - .expect("soft-delete duplicate row"); - - let mut seed_tx = db - .begin_transaction() - .await - .expect("begin seed transaction"); - let (_, was_inserted) = - event::insert_event_in_transaction(&mut seed_tx, community, &old, None) - .await - .expect("insert older live head"); - assert!(was_inserted); - seed_tx.commit().await.expect("commit older live head"); - - let mut tx = db - .begin_transaction() - .await - .expect("begin caller transaction"); - let result = db - .replace_parameterized_event_in_transaction( - &mut tx, - community, - &duplicate, - &d_tag, - None, - replaceable::ParameterizedReplacePrecondition::Unconditional, - ) - .await - .expect("evaluate soft-deleted duplicate"); - assert_eq!( - result.status, - replaceable::ParameterizedReplaceStatus::Duplicate - ); - - let live_id: Vec = sqlx::query_scalar( - "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ - AND d_tag=$4 AND deleted_at IS NULL", - ) - .bind(community.as_uuid()) - .bind(buzz_core::kind::KIND_PROJECT as i32) - .bind(keys.public_key().to_bytes()) - .bind(&d_tag) - .fetch_one(&mut *tx) - .await - .expect("caller transaction remains usable after duplicate"); - assert_eq!(live_id, old.id.as_bytes().to_vec()); - tx.rollback().await.expect("roll back caller transaction"); - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn concurrent_parameterized_replacement_keeps_deterministic_head() { - use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; - - let db = setup_db().await; - let community = CommunityId::from_uuid(make_community(&db.pool).await); - let keys = Keys::generate(); - let d_tag = format!("concurrent-project-{}", Uuid::new_v4().simple()); - let created_at = Timestamp::now().as_secs(); - let event = |content: &str, timestamp: u64| { - EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content) - .tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")]) - .custom_created_at(Timestamp::from(timestamp)) - .sign_with_keys(&keys) - .expect("sign project") - }; - let first = event("first", created_at); - let second = event("second", created_at); - let expected = if first.id.as_bytes() < second.id.as_bytes() { - &first - } else { - &second - }; - - let (first_result, second_result) = tokio::join!( - db.replace_parameterized_event(community, &first, &d_tag, None), - db.replace_parameterized_event(community, &second, &d_tag, None), - ); - let first_inserted = first_result.expect("first concurrent writer").1; - let second_inserted = second_result.expect("second concurrent writer").1; - assert!( - first_inserted || second_inserted, - "at least one concurrent writer must insert", - ); - - let live_ids: Vec> = sqlx::query_scalar( - "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ - AND d_tag=$4 AND deleted_at IS NULL", - ) - .bind(community.as_uuid()) - .bind(buzz_core::kind::KIND_PROJECT as i32) - .bind(keys.public_key().to_bytes()) - .bind(&d_tag) - .fetch_all(&db.pool) - .await - .expect("load concurrent live head"); - assert_eq!(live_ids, vec![expected.id.as_bytes().to_vec()]); - - assert!( - !db.replace_parameterized_event(community, expected, &d_tag, None) - .await - .expect("replay winning event") - .1, - "replaying the live event must be idempotent", - ); - let stale = event("stale", created_at.saturating_sub(1)); - assert!( - !db.replace_parameterized_event(community, &stale, &d_tag, None) - .await - .expect("submit stale event") - .1, - "an older event must not replace the live head", - ); - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn mesh_status_replacement_keeps_one_physical_row() { - use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; - - let db = setup_db().await; - let community = CommunityId::from_uuid(make_community(&db.pool).await); - let keys = Keys::generate(); - let d_tag = "buzz-mesh-member-status:owner-test"; - let tags = vec![ - Tag::parse(["d", d_tag]).expect("d tag"), - Tag::parse(["k", "buzz-mesh-status"]).expect("k tag"), - ]; - let base = Timestamp::now().as_secs(); - for (offset, content) in [(0, "running"), (1, "running-again"), (2, "stopped")] { - let event = EventBuilder::new( - Kind::Custom(buzz_core::kind::KIND_BOOKMARK_SET as u16), - content, - ) - .tags(tags.clone()) - .custom_created_at(Timestamp::from(base + offset)) - .sign_with_keys(&keys) - .expect("sign mesh status"); - assert!( - db.replace_parameterized_event(community, &event, d_tag, None) - .await - .expect("replace mesh status") - .1 - ); - } - - let (rows, live): (i64, i64) = sqlx::query_as( - "SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) FROM events \ - WHERE community_id=$1 AND kind=30003 AND pubkey=$2 AND d_tag=$3", - ) - .bind(community.as_uuid()) - .bind(keys.public_key().to_bytes()) - .bind(d_tag) - .fetch_one(&db.pool) - .await - .expect("count mesh status rows"); - assert_eq!((rows, live), (1, 1)); - - sqlx::query( - "UPDATE events SET deleted_at=NOW() \ - WHERE community_id=$1 AND kind=30003 AND pubkey=$2 AND d_tag=$3", - ) - .bind(community.as_uuid()) - .bind(keys.public_key().to_bytes()) - .bind(d_tag) - .execute(&db.pool) - .await - .expect("simulate old relay soft delete"); - let rows_after_legacy_delete: i64 = sqlx::query_scalar( - "SELECT count(*) FROM events \ - WHERE community_id=$1 AND kind=30003 AND pubkey=$2 AND d_tag=$3", - ) - .bind(community.as_uuid()) - .bind(keys.public_key().to_bytes()) - .bind(d_tag) - .fetch_one(&db.pool) - .await - .expect("count rows after old relay soft delete"); - assert_eq!( - rows_after_legacy_delete, 0, - "migration trigger must purge soft-deleted mesh status" - ); - } - #[tokio::test] #[ignore = "requires Postgres"] async fn coordinate_delete_spares_head_newer_than_the_deletion() { @@ -5855,246 +4895,6 @@ mod tests { ); } - #[tokio::test] - #[ignore = "requires Postgres"] - async fn duplicate_nip_rs_discriminator_tags_keep_legacy_retention() { - use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; - - let db = setup_db().await; - let community = CommunityId::from_uuid(make_community(&db.pool).await); - let keys = Keys::generate(); - let base = Timestamp::now().as_secs(); - - for (case, tags) in [ - ( - "duplicate-d", - vec![ - Tag::parse(["d", &format!("read-state:{}", "c".repeat(32))]) - .expect("first d tag"), - Tag::parse(["d", &format!("read-state:{}", "d".repeat(32))]) - .expect("second d tag"), - Tag::parse(["t", "read-state"]).expect("t tag"), - ], - ), - ( - "duplicate-t", - vec![ - Tag::parse(["d", &format!("read-state:{}", "e".repeat(32))]).expect("d tag"), - Tag::parse(["t", "read-state"]).expect("first t tag"), - Tag::parse(["t", "read-state"]).expect("second t tag"), - ], - ), - ] { - let d_tag = tags - .iter() - .find_map(|tag| { - let parts = tag.as_slice(); - (parts.first().is_some_and(|part| part == "d") && parts.len() >= 2) - .then(|| parts[1].clone()) - }) - .expect("first d-tag value"); - let old = EventBuilder::new( - Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), - format!("{case}-old"), - ) - .tags(tags.clone()) - .custom_created_at(Timestamp::from(base)) - .sign_with_keys(&keys) - .expect("sign old event"); - let new = EventBuilder::new( - Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), - format!("{case}-new"), - ) - .tags(tags) - .custom_created_at(Timestamp::from(base + 1)) - .sign_with_keys(&keys) - .expect("sign new event"); - - assert!( - db.replace_parameterized_event(community, &old, &d_tag, None) - .await - .expect("insert old event") - .1 - ); - assert!( - db.replace_parameterized_event(community, &new, &d_tag, None) - .await - .expect("replace with new event") - .1 - ); - - let (rows, live): (i64, i64) = sqlx::query_as( - "SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) FROM events \ - WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3", - ) - .bind(community.as_uuid()) - .bind(keys.public_key().to_bytes()) - .bind(&d_tag) - .fetch_one(&db.pool) - .await - .expect("count retained rows"); - assert_eq!((rows, live), (2, 1), "{case} must retain legacy history"); - - let watermarks: i64 = sqlx::query_scalar( - "SELECT count(*) FROM parameterized_event_watermarks \ - WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3", - ) - .bind(community.as_uuid()) - .bind(keys.public_key().to_bytes()) - .bind(&d_tag) - .fetch_one(&db.pool) - .await - .expect("count watermarks"); - assert_eq!(watermarks, 0, "{case} must not create a watermark"); - } - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn nip_rs_hard_delete_fence_fails_closed_and_scopes_opt_in_to_transaction() { - use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; - - let db = setup_db().await; - let community = CommunityId::from_uuid(make_community(&db.pool).await); - let keys = Keys::generate(); - let base = Timestamp::now().as_secs(); - let conforming_d = format!("read-state:{}", "6".repeat(32)); - let conforming = EventBuilder::new( - Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), - "fenced-conforming", - ) - .tags(vec![ - Tag::parse(["d", conforming_d.as_str()]).expect("d tag"), - Tag::parse(["t", "read-state"]).expect("t tag"), - ]) - .custom_created_at(Timestamp::from(base)) - .sign_with_keys(&keys) - .expect("sign conforming event"); - assert!( - db.replace_parameterized_event(community, &conforming, &conforming_d, None) - .await - .expect("insert conforming event") - .1 - ); - sqlx::query( - "INSERT INTO event_mentions \ - (community_id, pubkey_hex, event_id, event_created_at, event_kind) \ - VALUES ($1, $2, $3, to_timestamp($4), 30078)", - ) - .bind(community.as_uuid()) - .bind("6".repeat(64)) - .bind(conforming.id.as_bytes().as_slice()) - .bind(conforming.created_at.as_secs() as f64) - .execute(&db.pool) - .await - .expect("insert mention"); - - // Model ce10's first destructive statement. RAISE aborts the transaction, - // so its later mention delete and incoming insert can never commit. - let mut old_writer = db.pool.begin().await.expect("begin old-writer tx"); - let rejected = sqlx::query( - "DELETE FROM events WHERE community_id=$1 AND kind=30078 \ - AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL", - ) - .bind(community.as_uuid()) - .bind(keys.public_key().to_bytes()) - .bind(&conforming_d) - .execute(&mut *old_writer) - .await; - assert!(rejected.is_err(), "old-writer hard delete must be rejected"); - old_writer.rollback().await.expect("rollback rejected tx"); - let preserved: (i64, i64) = sqlx::query_as( - "SELECT (SELECT count(*) FROM events WHERE community_id=$1 AND id=$2), \ - (SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2)", - ) - .bind(community.as_uuid()) - .bind(conforming.id.as_bytes().as_slice()) - .fetch_one(&db.pool) - .await - .expect("count preserved payload and mention"); - assert_eq!(preserved, (1, 1)); - - let nonconforming_d = format!("read-state:{}", "7".repeat(32)); - let nonconforming = EventBuilder::new( - Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), - "fenced-nonconforming", - ) - .tags(vec![ - Tag::parse(["d", nonconforming_d.as_str()]).expect("first d tag"), - Tag::parse(["d", "other"]).expect("second d tag"), - Tag::parse(["t", "read-state"]).expect("t tag"), - ]) - .custom_created_at(Timestamp::from(base + 1)) - .sign_with_keys(&keys) - .expect("sign nonconforming event"); - assert!( - db.replace_parameterized_event(community, &nonconforming, &nonconforming_d, None,) - .await - .expect("insert nonconforming event") - .1 - ); - let rejected_nonconforming = sqlx::query( - "DELETE FROM events WHERE community_id=$1 AND id=$2 AND created_at=to_timestamp($3)", - ) - .bind(community.as_uuid()) - .bind(nonconforming.id.as_bytes().as_slice()) - .bind(nonconforming.created_at.as_secs() as f64) - .execute(&db.pool) - .await; - assert!( - rejected_nonconforming.is_err(), - "fence must cover a nonconforming OLD row at a regex coordinate" - ); - - let unrelated_d = format!("read-state:{}", "8".repeat(32)); - let unrelated = EventBuilder::new(Kind::Custom(30023), "unrelated") - .tags(vec![Tag::parse(["d", unrelated_d.as_str()]).expect("d tag")]) - .custom_created_at(Timestamp::from(base + 2)) - .sign_with_keys(&keys) - .expect("sign unrelated event"); - assert!( - db.replace_parameterized_event(community, &unrelated, &unrelated_d, None) - .await - .expect("insert unrelated event") - .1 - ); - let unrelated_delete = sqlx::query( - "DELETE FROM events WHERE community_id=$1 AND id=$2 AND created_at=to_timestamp($3)", - ) - .bind(community.as_uuid()) - .bind(unrelated.id.as_bytes().as_slice()) - .bind(unrelated.created_at.as_secs() as f64) - .execute(&db.pool) - .await - .expect("delete unrelated event"); - assert_eq!(unrelated_delete.rows_affected(), 1); - - // Check both transaction exits on one physical session; pool selection - // cannot accidentally hide a leaked session-local authorization value. - let mut conn = db.pool.acquire().await.expect("acquire dedicated session"); - for commit in [true, false] { - let mut tx = conn.begin().await.expect("begin GUC transaction"); - let value: String = - sqlx::query_scalar("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)") - .fetch_one(&mut *tx) - .await - .expect("set transaction-local GUC"); - assert_eq!(value, "on"); - if commit { - tx.commit().await.expect("commit GUC transaction"); - } else { - tx.rollback().await.expect("rollback GUC transaction"); - } - let leaked: Option = sqlx::query_scalar( - "SELECT NULLIF(current_setting('buzz.nip_rs_hard_delete', true), '')", - ) - .fetch_one(&mut *conn) - .await - .expect("read GUC after transaction"); - assert_ne!(leaked.as_deref(), Some("on")); - } - } - #[tokio::test] #[ignore = "requires Postgres"] async fn database_guard_covers_legacy_writer_and_nip09_deletion() { diff --git a/crates/buzz-db/src/replaceable.rs b/crates/buzz-db/src/replaceable.rs index 4904dfe209..9b575b6ea1 100644 --- a/crates/buzz-db/src/replaceable.rs +++ b/crates/buzz-db/src/replaceable.rs @@ -363,6 +363,163 @@ async fn replace_parameterized_event_in_transaction_impl( } impl Db { + /// Atomically replace a replaceable event: NIP-16 kinds (0, 3, 41, 10000–19999) + /// and NIP-29 discovery state (39000–39002, called from side_effects.rs). + /// + /// Keeps only the event with the highest `created_at` per (kind, pubkey, channel_id). + /// Same-second ties are broken by lowest event `id` (NIP-16 deterministic ordering). + /// Returns `(event, false)` for stale writes and duplicate IDs — callers should + /// skip fan-out/dispatch when `was_inserted` is false. + #[datastore_span(name = "replace_addressable_event", system = "postgresql")] + pub async fn replace_addressable_event( + &self, + community_id: CommunityId, + event: &nostr::Event, + channel_id: Option, + ) -> Result<(StoredEvent, bool)> { + let kind_i32 = buzz_core::kind::event_kind_i32(event); + let pubkey_bytes = event.pubkey.to_bytes(); + let created_at_secs = event.created_at.as_secs() as i64; + let created_at = chrono::DateTime::from_timestamp(created_at_secs, 0) + .ok_or(DbError::InvalidTimestamp(created_at_secs))?; + + // Collisions only cause extra serialization; they cannot change behavior. + let lock_key = event_replacement_lock_key( + community_id, + kind_i32, + pubkey_bytes.as_slice(), + channel_id.as_ref().map(|id| id.as_bytes().as_slice()), + ); + + let (mut tx, transaction_timer) = observability::begin_transaction( + &self.pool, + observability::TransactionOperation::ReplaceAddressableEvent, + ) + .await?; + + transaction_timer + .observe(async { + // Serialize all writers for the same (kind, pubkey, channel_id) tuple. + // Advisory lock is transaction-scoped — released on commit/rollback. + observability::observe_advisory_lock( + observability::LockType::Replacement, + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_key) + .execute(&mut *tx), + ) + .await?; + + // Check for the newest existing event. ORDER BY + LIMIT 1 is defensive against + // historical data where prior bugs may have left multiple live rows. + let existing: Option<(chrono::DateTime, Vec)> = + sqlx::query_as( + "SELECT created_at, id FROM events \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ + AND channel_id IS NOT DISTINCT FROM $4 \ + AND deleted_at IS NULL \ + ORDER BY created_at DESC, id ASC LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(channel_id) + .fetch_optional(&mut *tx) + .await?; + + // Stale-write protection: reject if incoming is not newer. + // NIP-16: created_at is second-resolution. On same-second tie, lowest + // event id (lexicographic) wins — deterministic across relays. + let incoming_id = event.id.as_bytes().as_slice(); + if let Some((existing_ts, existing_id)) = existing { + let dominated = created_at < existing_ts + || (created_at == existing_ts + && incoming_id >= existing_id.as_slice()); + if dominated { + tx.rollback().await?; + let received_at = chrono::Utc::now(); + return Ok(( + StoredEvent::with_received_at( + event.clone(), + received_at, + channel_id, + false, + ), + false, + )); + } + } + + // Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety. + sqlx::query( + "UPDATE events SET deleted_at = NOW() \ + WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \ + AND channel_id IS NOT DISTINCT FROM $4 \ + AND deleted_at IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(kind_i32) + .bind(pubkey_bytes.as_slice()) + .bind(channel_id) + .execute(&mut *tx) + .await?; + + // Insert the new event inside the same transaction. + let sig_bytes = event.sig.serialize(); + let tags_json = serde_json::to_value(&event.tags)?; + let received_at = chrono::Utc::now(); + let d_tag = crate::event::extract_d_tag(event); + + let insert_result = sqlx::query( + "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \ + ON CONFLICT DO NOTHING", + ) + .bind(community_id.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .bind(pubkey_bytes.as_slice()) + .bind(created_at) + .bind(kind_i32) + .bind(&tags_json) + .bind(&event.content) + .bind(sig_bytes.as_slice()) + .bind(received_at) + .bind(channel_id) + .bind(d_tag.as_deref()) + .execute(&mut *tx) + .await?; + + let was_inserted = insert_result.rows_affected() > 0; + if !was_inserted { + // ON CONFLICT fired — the event ID already exists. Rollback the + // soft-delete so we don't lose the previous replaceable event. + tx.rollback().await?; + return Ok(( + StoredEvent::with_received_at( + event.clone(), + received_at, + channel_id, + false, + ), + false, + )); + } + + // 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?; + + tx.commit().await?; + + Ok(( + StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), + true, + )) + }) + .await + } + /// Replace a NIP-33 event inside a caller-owned transaction. /// /// The caller owns commit or rollback. Requiring [`Transaction`] here and @@ -427,3 +584,1170 @@ impl Db { .await } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{event, migration, replaceable}; + use sqlx::postgres::PgPoolOptions; + use sqlx::{Acquire, PgPool}; + use std::time::Duration; + + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials + + async fn setup_db() -> Db { + let database_url = + std::env::var("TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.into()); + let pool = PgPool::connect(&database_url) + .await + .expect("connect to test DB"); + Db::from_pool(pool) + } + + async fn make_community(pool: &PgPool) -> Uuid { + let id = Uuid::new_v4(); + let host = format!("communities-of-channels-{}.example", id.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(host) + .execute(pool) + .await + .expect("insert community"); + id + } + + async fn admin_url() -> String { + std::env::var("TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.into()) + } + + /// Create a fresh scratch database on the same server and optionally run migrations. + async fn create_scratch_db_through( + admin: &PgPool, + prefix: &str, + target: Option, + ) -> (PgPool, String) { + let name = format!("{}_{}", prefix, Uuid::new_v4().simple()); + sqlx::query(sqlx::AssertSqlSafe(format!("CREATE DATABASE {name}"))) + .execute(admin) + .await + .expect("create scratch db"); + let base = admin_url().await; + // Swap the database path segment of the admin URL for the scratch name. + let scratch_url = { + let idx = base.rfind('/').expect("db url has a path segment"); + format!("{}/{}", &base[..idx], name) + }; + let pool = PgPool::connect(&scratch_url) + .await + .expect("connect scratch db"); + match target { + Some(target) => migration::run_migrations_through(&pool, target) + .await + .expect("migrate scratch db through target"), + None => migration::run_migrations(&pool) + .await + .expect("migrate scratch db"), + } + (pool, name) + } + + /// Create a fresh scratch database on the same server and run all migrations. + /// Returns (pool, db_name); callers should `drop_scratch_db` when done. + async fn create_scratch_db(admin: &PgPool, prefix: &str) -> (PgPool, String) { + create_scratch_db_through(admin, prefix, None).await + } + + async fn drop_scratch_db(admin: &PgPool, pool: PgPool, name: &str) { + pool.close().await; + let _ = sqlx::query(sqlx::AssertSqlSafe(format!( + "DROP DATABASE IF EXISTS {name} WITH (FORCE)" + ))) + .execute(admin) + .await; + } + + /// Insert identical community + channel rows into a database so the same + /// (community, channel) ids resolve in both writer and replica. + async fn seed_community_channel( + pool: &PgPool, + community: Uuid, + channel: Uuid, + author: &nostr::Keys, + ) { + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community) + .bind(format!("replica-routing-{}.example", community.simple())) + .execute(pool) + .await + .expect("insert community"); + crate::channel::create_channel_with_id( + pool, + CommunityId::from_uuid(community), + channel, + &format!("replica-routing-{channel}"), + crate::channel::ChannelType::Stream, + crate::channel::ChannelVisibility::Open, + None, + author.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("create channel"); + } + + #[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(); + let owner_keys = Keys::generate(); + seed_community_channel(&pool, community_uuid, channel, &owner_keys).await; + let community = CommunityId::from_uuid(community_uuid); + let member = owner_keys.public_key().to_hex(); + let tags = || { + vec![ + Tag::parse(["d", channel.to_string().as_str()]).expect("d tag"), + Tag::parse(["p", member.as_str(), "", "owner"]).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 stale_legacy_roster_cannot_replace_new_locked_snapshot() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let admin = PgPool::connect(&admin_url().await) + .await + .expect("connect admin"); + let (setup_pool, scratch_name) = create_scratch_db(&admin, "mixed_roster_writer").await; + let base_url = admin_url().await; + let slash = base_url.rfind('/').expect("database URL has path segment"); + let scratch_url = format!("{}/{}", &base_url[..slash], scratch_name); + let pool = PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(Duration::from_secs(1)) + .connect(&scratch_url) + .await + .expect("connect one-connection scratch pool"); + setup_pool.close().await; + let db = Db::from_pool(pool.clone()); + let community_uuid = Uuid::new_v4(); + let community = CommunityId::from_uuid(community_uuid); + let channel = Uuid::new_v4(); + let relay_keys = Keys::generate(); + let owner_keys = Keys::generate(); + let owner = owner_keys.public_key().to_bytes(); + seed_community_channel(&pool, community_uuid, channel, &owner_keys).await; + + // This is the old pod's unlocked capture A. It remains in process memory + // while a role-only canonical mutation advances and the new pod publishes B. + let base = Timestamp::now().as_secs(); + let roster = |members: &[(&[u8], &str)], timestamp| { + let tags = + std::iter::once(Tag::parse(["d", channel.to_string().as_str()]).expect("d tag")) + .chain(members.iter().map(|(member, role)| { + Tag::parse(["p", hex::encode(member).as_str(), "", *role]).expect("p tag") + })) + .collect::>(); + EventBuilder::new(Kind::Custom(39002), "") + .tags(tags) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&relay_keys) + .expect("sign roster") + }; + + let newcomer = Keys::generate().public_key().to_bytes(); + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) \ + VALUES ($1, $2, $3, 'member', $4)", + ) + .bind(community_uuid) + .bind(channel) + .bind(newcomer.as_slice()) + .bind(owner.as_slice()) + .execute(&pool) + .await + .expect("seed member before legacy capture"); + let stale_a = roster( + &[(owner.as_slice(), "owner"), (newcomer.as_slice(), "member")], + base + 2, + ); + + sqlx::query( + "UPDATE channel_members SET role = 'admin' \ + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", + ) + .bind(community_uuid) + .bind(channel) + .bind(newcomer.as_slice()) + .execute(&pool) + .await + .expect("commit newer canonical role"); + + let relay_pubkey = relay_keys.public_key().to_bytes(); + let mut snapshot = db + .lock_member_snapshot(community, channel, &relay_pubkey) + .await + .expect("new writer captures locked roster B"); + let fresh_b = roster( + &[(owner.as_slice(), "owner"), (newcomer.as_slice(), "admin")], + base + 1, + ); + assert!( + snapshot + .replace_member_event(community, channel, &fresh_b) + .await + .expect("new writer publishes B") + .1 + ); + snapshot + .release() + .await + .expect("commit B and release locks"); + + // The legacy canonical path takes the replacement key, soft-deletes B, + // then attempts its newer-timestamp stale A. Migration 0032 rejects the + // INSERT; transaction rollback must restore B. A one-connection pool + // proves the lock order does not turn this compatibility path into a + // self-deadlock. + let error = tokio::time::timeout( + Duration::from_secs(3), + db.replace_addressable_event(community, &stale_a, Some(channel)), + ) + .await + .expect("legacy replacement must not deadlock") + .expect_err("stale captured roster A must be rejected"); + assert!( + matches!( + error, + DbError::Sqlx(sqlx::Error::Database(ref db_error)) + if db_error.code().as_deref() == Some("23514") + ), + "expected roster fence check violation, got {error:?}" + ); + + let live_ids: Vec> = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND channel_id=$2 \ + AND kind=39002 AND pubkey=$3 AND deleted_at IS NULL", + ) + .bind(community_uuid) + .bind(channel) + .bind(relay_pubkey.as_slice()) + .fetch_all(&pool) + .await + .expect("load live roster heads"); + assert_eq!(live_ids, vec![fresh_b.id.as_bytes().to_vec()]); + let stale_rows: i64 = + sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND id=$2") + .bind(community_uuid) + .bind(stale_a.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("count rejected stale roster"); + assert_eq!(stale_rows, 0, "stale roster insert must roll back"); + + 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() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let d_tag = format!("read-state:{}", "a".repeat(32)); + let tags = vec![ + Tag::parse(["d", d_tag.as_str()]).expect("d tag"), + Tag::parse(["t", "read-state"]).expect("t tag"), + ]; + let base = Timestamp::now().as_secs(); + let old = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "old") + .tags(tags.clone()) + .custom_created_at(Timestamp::from(base)) + .sign_with_keys(&keys) + .expect("sign old"); + let new = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "new") + .tags(tags) + .custom_created_at(Timestamp::from(base + 1)) + .sign_with_keys(&keys) + .expect("sign new"); + + assert!( + db.replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("insert old") + .1 + ); + assert!( + db.replace_parameterized_event(community, &new, &d_tag, None) + .await + .expect("replace with new") + .1 + ); + + let rows: i64 = sqlx::query_scalar( + "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_one(&db.pool) + .await + .expect("count NIP-RS rows"); + assert_eq!(rows, 1, "superseded payload must be physically deleted"); + + sqlx::query( + "UPDATE events SET deleted_at=NOW() WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .execute(&db.pool) + .await + .expect("simulate NIP-09 coordinate deletion"); + + assert!( + !db.replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("replay old") + .1 + ); + let live: i64 = sqlx::query_scalar( + "SELECT count(*) FROM events WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_one(&db.pool) + .await + .expect("count live NIP-RS rows"); + assert_eq!(live, 0, "watermark must block stale resurrection"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn nip_rs_transaction_operation_restores_hard_delete_opt_in() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let base = Timestamp::now().as_secs(); + let replace_d_tag = format!("read-state:{}", "b".repeat(32)); + let victim_d_tag = format!("read-state:{}", "c".repeat(32)); + let event = |d_tag: &str, content: &str, timestamp: u64| { + EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), + content, + ) + .tags(vec![ + Tag::parse(["d", d_tag]).expect("d tag"), + Tag::parse(["t", "read-state"]).expect("t tag"), + ]) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&keys) + .expect("sign read state") + }; + let old = event(&replace_d_tag, "old", base); + let new = event(&replace_d_tag, "new", base + 1); + let victim = event(&victim_d_tag, "victim", base); + + assert!( + db.replace_parameterized_event(community, &old, &replace_d_tag, None) + .await + .expect("insert old head") + .1 + ); + assert!( + db.replace_parameterized_event(community, &victim, &victim_d_tag, None) + .await + .expect("insert victim head") + .1 + ); + + let mut tx = db + .begin_transaction() + .await + .expect("begin caller transaction"); + let result = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &new, + &replace_d_tag, + None, + replaceable::ParameterizedReplacePrecondition::Unconditional, + ) + .await + .expect("replace inside caller transaction"); + assert_eq!( + result.status, + replaceable::ParameterizedReplaceStatus::Inserted + ); + + let leaked: Option = sqlx::query_scalar( + "SELECT NULLIF(current_setting('buzz.nip_rs_hard_delete', true), '')", + ) + .fetch_one(&mut *tx) + .await + .expect("read hard-delete opt-in after replacement"); + assert_ne!(leaked.as_deref(), Some("on")); + + let unauthorized = sqlx::query( + "DELETE FROM events WHERE community_id=$1 AND kind=30078 \ + AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(&victim_d_tag) + .execute(&mut *tx) + .await; + assert!( + unauthorized.is_err(), + "replacement opt-in must not authorize later caller SQL" + ); + tx.rollback().await.expect("roll back caller transaction"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn parameterized_replacement_in_existing_transaction_honors_revision_and_rollback() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let d_tag = format!("transactional-project-{}", Uuid::new_v4().simple()); + let base = Timestamp::now().as_secs(); + let event = |content: &str, timestamp: u64| { + EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content) + .tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")]) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&keys) + .expect("sign project") + }; + let old = event("old", base); + let new = event("new", base + 1); + + assert!( + db.replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("insert old head") + .1 + ); + + let mut tx = db.begin_transaction().await.expect("begin replacement tx"); + let outcome = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &new, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExpectedRevision( + old.id.as_bytes().as_slice(), + ), + ) + .await + .expect("replace inside caller transaction"); + assert_eq!( + outcome.status, + replaceable::ParameterizedReplaceStatus::Inserted + ); + tx.rollback().await.expect("roll back replacement tx"); + + let live_id: Vec = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(buzz_core::kind::KIND_PROJECT as i32) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_one(&db.pool) + .await + .expect("load live head after rollback"); + assert_eq!(live_id, old.id.as_bytes().to_vec()); + + let mut tx = db + .begin_transaction() + .await + .expect("begin stale revision tx"); + let mismatch = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &new, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExpectedRevision( + [0x42; 32].as_slice(), + ), + ) + .await + .expect("evaluate stale revision"); + assert_eq!( + mismatch.status, + replaceable::ParameterizedReplaceStatus::RevisionMismatch + ); + tx.rollback().await.expect("roll back stale revision tx"); + + let missing_d_tag = format!("missing-project-{}", Uuid::new_v4().simple()); + let missing = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), + "missing", + ) + .tags(vec![ + Tag::parse(["d", missing_d_tag.as_str()]).expect("missing d tag") + ]) + .custom_created_at(Timestamp::from(base + 2)) + .sign_with_keys(&keys) + .expect("sign missing project"); + let mut tx = db + .begin_transaction() + .await + .expect("begin missing revision tx"); + let missing_result = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &missing, + &missing_d_tag, + None, + replaceable::ParameterizedReplacePrecondition::ExpectedRevision( + [0x24; 32].as_slice(), + ), + ) + .await + .expect("evaluate missing revision"); + assert_eq!( + missing_result.status, + replaceable::ParameterizedReplaceStatus::RevisionMissing + ); + tx.rollback().await.expect("roll back missing revision tx"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn parameterized_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_parameterized").await; + let db = Db::from_pool(pool.clone()); + let community = CommunityId::from_uuid(make_community(&pool).await); + let keys = Keys::generate(); + let mentioned = Keys::generate().public_key().to_hex(); + let d_tag = format!("mention-project-{}", Uuid::new_v4().simple()); + let tags = || { + vec![ + Tag::parse(["d", d_tag.as_str()]).expect("d tag"), + Tag::parse(["p", mentioned.as_str()]).expect("p tag"), + ] + }; + let base = Timestamp::now().as_secs(); + let old = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), "old") + .tags(tags()) + .custom_created_at(Timestamp::from(base)) + .sign_with_keys(&keys) + .expect("sign old project"); + let new = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), "new") + .tags(tags()) + .custom_created_at(Timestamp::from(base + 1)) + .sign_with_keys(&keys) + .expect("sign new project"); + + assert!( + db.replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("insert old project") + .1 + ); + 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 mut tx = db + .begin_transaction() + .await + .expect("begin caller transaction"); + let error = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &new, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::Unconditional, + ) + .await + .expect_err("mention failure must fail replacement"); + assert!(error.to_string().contains("injected mention failure")); + + let probe: i32 = sqlx::query_scalar("SELECT 1") + .fetch_one(&mut *tx) + .await + .expect("inner failure must leave caller transaction usable"); + assert_eq!(probe, 1); + + let live_id: Vec = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(buzz_core::kind::KIND_PROJECT as i32) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_one(&mut *tx) + .await + .expect("load live project after failed indexing"); + assert_eq!(live_id, old.id.as_bytes().to_vec()); + 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(&mut *tx) + .await + .expect("count rolled-back project"); + assert_eq!(new_rows, 0); + tx.commit().await.expect("commit usable caller transaction"); + + drop_scratch_db(&admin, pool, &scratch_name).await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn parameterized_duplicate_restores_live_head_inside_caller_transaction() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let d_tag = format!("duplicate-project-{}", Uuid::new_v4().simple()); + let base = Timestamp::now().as_secs(); + let event = |content: &str, timestamp: u64| { + EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content) + .tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")]) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&keys) + .expect("sign project") + }; + let old = event("old-live-head", base); + let duplicate = event("soft-deleted-duplicate", base + 1); + + assert!( + db.replace_parameterized_event(community, &duplicate, &d_tag, None) + .await + .expect("insert future duplicate") + .1 + ); + sqlx::query("UPDATE events SET deleted_at=NOW() WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(duplicate.id.as_bytes().as_slice()) + .execute(&db.pool) + .await + .expect("soft-delete duplicate row"); + + let mut seed_tx = db + .begin_transaction() + .await + .expect("begin seed transaction"); + let (_, was_inserted) = + event::insert_event_in_transaction(&mut seed_tx, community, &old, None) + .await + .expect("insert older live head"); + assert!(was_inserted); + seed_tx.commit().await.expect("commit older live head"); + + let mut tx = db + .begin_transaction() + .await + .expect("begin caller transaction"); + let result = db + .replace_parameterized_event_in_transaction( + &mut tx, + community, + &duplicate, + &d_tag, + None, + replaceable::ParameterizedReplacePrecondition::Unconditional, + ) + .await + .expect("evaluate soft-deleted duplicate"); + assert_eq!( + result.status, + replaceable::ParameterizedReplaceStatus::Duplicate + ); + + let live_id: Vec = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(buzz_core::kind::KIND_PROJECT as i32) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_one(&mut *tx) + .await + .expect("caller transaction remains usable after duplicate"); + assert_eq!(live_id, old.id.as_bytes().to_vec()); + tx.rollback().await.expect("roll back caller transaction"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn concurrent_parameterized_replacement_keeps_deterministic_head() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let d_tag = format!("concurrent-project-{}", Uuid::new_v4().simple()); + let created_at = Timestamp::now().as_secs(); + let event = |content: &str, timestamp: u64| { + EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_PROJECT as u16), content) + .tags(vec![Tag::parse(["d", d_tag.as_str()]).expect("d tag")]) + .custom_created_at(Timestamp::from(timestamp)) + .sign_with_keys(&keys) + .expect("sign project") + }; + let first = event("first", created_at); + let second = event("second", created_at); + let expected = if first.id.as_bytes() < second.id.as_bytes() { + &first + } else { + &second + }; + + let (first_result, second_result) = tokio::join!( + db.replace_parameterized_event(community, &first, &d_tag, None), + db.replace_parameterized_event(community, &second, &d_tag, None), + ); + let first_inserted = first_result.expect("first concurrent writer").1; + let second_inserted = second_result.expect("second concurrent writer").1; + assert!( + first_inserted || second_inserted, + "at least one concurrent writer must insert", + ); + + let live_ids: Vec> = sqlx::query_scalar( + "SELECT id FROM events WHERE community_id=$1 AND kind=$2 AND pubkey=$3 \ + AND d_tag=$4 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(buzz_core::kind::KIND_PROJECT as i32) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_all(&db.pool) + .await + .expect("load concurrent live head"); + assert_eq!(live_ids, vec![expected.id.as_bytes().to_vec()]); + + assert!( + !db.replace_parameterized_event(community, expected, &d_tag, None) + .await + .expect("replay winning event") + .1, + "replaying the live event must be idempotent", + ); + let stale = event("stale", created_at.saturating_sub(1)); + assert!( + !db.replace_parameterized_event(community, &stale, &d_tag, None) + .await + .expect("submit stale event") + .1, + "an older event must not replace the live head", + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn mesh_status_replacement_keeps_one_physical_row() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let d_tag = "buzz-mesh-member-status:owner-test"; + let tags = vec![ + Tag::parse(["d", d_tag]).expect("d tag"), + Tag::parse(["k", "buzz-mesh-status"]).expect("k tag"), + ]; + let base = Timestamp::now().as_secs(); + for (offset, content) in [(0, "running"), (1, "running-again"), (2, "stopped")] { + let event = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_BOOKMARK_SET as u16), + content, + ) + .tags(tags.clone()) + .custom_created_at(Timestamp::from(base + offset)) + .sign_with_keys(&keys) + .expect("sign mesh status"); + assert!( + db.replace_parameterized_event(community, &event, d_tag, None) + .await + .expect("replace mesh status") + .1 + ); + } + + let (rows, live): (i64, i64) = sqlx::query_as( + "SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) FROM events \ + WHERE community_id=$1 AND kind=30003 AND pubkey=$2 AND d_tag=$3", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(d_tag) + .fetch_one(&db.pool) + .await + .expect("count mesh status rows"); + assert_eq!((rows, live), (1, 1)); + + sqlx::query( + "UPDATE events SET deleted_at=NOW() \ + WHERE community_id=$1 AND kind=30003 AND pubkey=$2 AND d_tag=$3", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(d_tag) + .execute(&db.pool) + .await + .expect("simulate old relay soft delete"); + let rows_after_legacy_delete: i64 = sqlx::query_scalar( + "SELECT count(*) FROM events \ + WHERE community_id=$1 AND kind=30003 AND pubkey=$2 AND d_tag=$3", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(d_tag) + .fetch_one(&db.pool) + .await + .expect("count rows after old relay soft delete"); + assert_eq!( + rows_after_legacy_delete, 0, + "migration trigger must purge soft-deleted mesh status" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn duplicate_nip_rs_discriminator_tags_keep_legacy_retention() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let base = Timestamp::now().as_secs(); + + for (case, tags) in [ + ( + "duplicate-d", + vec![ + Tag::parse(["d", &format!("read-state:{}", "c".repeat(32))]) + .expect("first d tag"), + Tag::parse(["d", &format!("read-state:{}", "d".repeat(32))]) + .expect("second d tag"), + Tag::parse(["t", "read-state"]).expect("t tag"), + ], + ), + ( + "duplicate-t", + vec![ + Tag::parse(["d", &format!("read-state:{}", "e".repeat(32))]).expect("d tag"), + Tag::parse(["t", "read-state"]).expect("first t tag"), + Tag::parse(["t", "read-state"]).expect("second t tag"), + ], + ), + ] { + let d_tag = tags + .iter() + .find_map(|tag| { + let parts = tag.as_slice(); + (parts.first().is_some_and(|part| part == "d") && parts.len() >= 2) + .then(|| parts[1].clone()) + }) + .expect("first d-tag value"); + let old = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), + format!("{case}-old"), + ) + .tags(tags.clone()) + .custom_created_at(Timestamp::from(base)) + .sign_with_keys(&keys) + .expect("sign old event"); + let new = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), + format!("{case}-new"), + ) + .tags(tags) + .custom_created_at(Timestamp::from(base + 1)) + .sign_with_keys(&keys) + .expect("sign new event"); + + assert!( + db.replace_parameterized_event(community, &old, &d_tag, None) + .await + .expect("insert old event") + .1 + ); + assert!( + db.replace_parameterized_event(community, &new, &d_tag, None) + .await + .expect("replace with new event") + .1 + ); + + let (rows, live): (i64, i64) = sqlx::query_as( + "SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) FROM events \ + WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_one(&db.pool) + .await + .expect("count retained rows"); + assert_eq!((rows, live), (2, 1), "{case} must retain legacy history"); + + let watermarks: i64 = sqlx::query_scalar( + "SELECT count(*) FROM parameterized_event_watermarks \ + WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(&d_tag) + .fetch_one(&db.pool) + .await + .expect("count watermarks"); + assert_eq!(watermarks, 0, "{case} must not create a watermark"); + } + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn nip_rs_hard_delete_fence_fails_closed_and_scopes_opt_in_to_transaction() { + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + + let db = setup_db().await; + let community = CommunityId::from_uuid(make_community(&db.pool).await); + let keys = Keys::generate(); + let base = Timestamp::now().as_secs(); + let conforming_d = format!("read-state:{}", "6".repeat(32)); + let conforming = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), + "fenced-conforming", + ) + .tags(vec![ + Tag::parse(["d", conforming_d.as_str()]).expect("d tag"), + Tag::parse(["t", "read-state"]).expect("t tag"), + ]) + .custom_created_at(Timestamp::from(base)) + .sign_with_keys(&keys) + .expect("sign conforming event"); + assert!( + db.replace_parameterized_event(community, &conforming, &conforming_d, None) + .await + .expect("insert conforming event") + .1 + ); + sqlx::query( + "INSERT INTO event_mentions \ + (community_id, pubkey_hex, event_id, event_created_at, event_kind) \ + VALUES ($1, $2, $3, to_timestamp($4), 30078)", + ) + .bind(community.as_uuid()) + .bind("6".repeat(64)) + .bind(conforming.id.as_bytes().as_slice()) + .bind(conforming.created_at.as_secs() as f64) + .execute(&db.pool) + .await + .expect("insert mention"); + + // Model ce10's first destructive statement. RAISE aborts the transaction, + // so its later mention delete and incoming insert can never commit. + let mut old_writer = db.pool.begin().await.expect("begin old-writer tx"); + let rejected = sqlx::query( + "DELETE FROM events WHERE community_id=$1 AND kind=30078 \ + AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .bind(keys.public_key().to_bytes()) + .bind(&conforming_d) + .execute(&mut *old_writer) + .await; + assert!(rejected.is_err(), "old-writer hard delete must be rejected"); + old_writer.rollback().await.expect("rollback rejected tx"); + let preserved: (i64, i64) = sqlx::query_as( + "SELECT (SELECT count(*) FROM events WHERE community_id=$1 AND id=$2), \ + (SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2)", + ) + .bind(community.as_uuid()) + .bind(conforming.id.as_bytes().as_slice()) + .fetch_one(&db.pool) + .await + .expect("count preserved payload and mention"); + assert_eq!(preserved, (1, 1)); + + let nonconforming_d = format!("read-state:{}", "7".repeat(32)); + let nonconforming = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), + "fenced-nonconforming", + ) + .tags(vec![ + Tag::parse(["d", nonconforming_d.as_str()]).expect("first d tag"), + Tag::parse(["d", "other"]).expect("second d tag"), + Tag::parse(["t", "read-state"]).expect("t tag"), + ]) + .custom_created_at(Timestamp::from(base + 1)) + .sign_with_keys(&keys) + .expect("sign nonconforming event"); + assert!( + db.replace_parameterized_event(community, &nonconforming, &nonconforming_d, None,) + .await + .expect("insert nonconforming event") + .1 + ); + let rejected_nonconforming = sqlx::query( + "DELETE FROM events WHERE community_id=$1 AND id=$2 AND created_at=to_timestamp($3)", + ) + .bind(community.as_uuid()) + .bind(nonconforming.id.as_bytes().as_slice()) + .bind(nonconforming.created_at.as_secs() as f64) + .execute(&db.pool) + .await; + assert!( + rejected_nonconforming.is_err(), + "fence must cover a nonconforming OLD row at a regex coordinate" + ); + + let unrelated_d = format!("read-state:{}", "8".repeat(32)); + let unrelated = EventBuilder::new(Kind::Custom(30023), "unrelated") + .tags(vec![Tag::parse(["d", unrelated_d.as_str()]).expect("d tag")]) + .custom_created_at(Timestamp::from(base + 2)) + .sign_with_keys(&keys) + .expect("sign unrelated event"); + assert!( + db.replace_parameterized_event(community, &unrelated, &unrelated_d, None) + .await + .expect("insert unrelated event") + .1 + ); + let unrelated_delete = sqlx::query( + "DELETE FROM events WHERE community_id=$1 AND id=$2 AND created_at=to_timestamp($3)", + ) + .bind(community.as_uuid()) + .bind(unrelated.id.as_bytes().as_slice()) + .bind(unrelated.created_at.as_secs() as f64) + .execute(&db.pool) + .await + .expect("delete unrelated event"); + assert_eq!(unrelated_delete.rows_affected(), 1); + + // Check both transaction exits on one physical session; pool selection + // cannot accidentally hide a leaked session-local authorization value. + let mut conn = db.pool.acquire().await.expect("acquire dedicated session"); + for commit in [true, false] { + let mut tx = conn.begin().await.expect("begin GUC transaction"); + let value: String = + sqlx::query_scalar("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)") + .fetch_one(&mut *tx) + .await + .expect("set transaction-local GUC"); + assert_eq!(value, "on"); + if commit { + tx.commit().await.expect("commit GUC transaction"); + } else { + tx.rollback().await.expect("rollback GUC transaction"); + } + let leaked: Option = sqlx::query_scalar( + "SELECT NULLIF(current_setting('buzz.nip_rs_hard_delete', true), '')", + ) + .fetch_one(&mut *conn) + .await + .expect("read GUC after transaction"); + assert_ne!(leaked.as_deref(), Some("on")); + } + } +} diff --git a/desktop/src-tauri/src/managed_agents/nest_skill.md b/desktop/src-tauri/src/managed_agents/nest_skill.md index 79a5ea301d..01f7622915 100644 --- a/desktop/src-tauri/src/managed_agents/nest_skill.md +++ b/desktop/src-tauri/src/managed_agents/nest_skill.md @@ -51,7 +51,7 @@ Manage your repository's enforced branch and tag rules with `repos protect list| Output varies by command group — `--help` shows flags but not response shapes. -**Read commands** (messages, channels, users, feed, workflows): normalized JSON arrays with `sig` stripped. Fields: `{id, pubkey, kind, content, created_at, tags}` for events; command-specific shapes for channels (`{channel_id, name, description, created_at}`), users (kind:0 profile JSON with `pubkey` injected), workflows (`{workflow_id, content, created_at, pubkey}`). +**Read commands** return JSON arrays. Event reads (`messages get/thread/search`, `feed get`) return normalized, complete signed Nostr events with `{id, pubkey, kind, content, created_at, tags, sig}`. Other reads use command-specific shapes for channels (`{channel_id, name, description, created_at}`), users (kind:0 profile JSON with `pubkey` injected), and workflows (`{workflow_id, content, created_at, pubkey}`). **Write commands**: all return `{event_id, accepted, message}`. Create commands add the generated entity ID: `channels create` → `channel_id`, `dms open` → `dm_id`, `workflows create` → `workflow_id`. Agent draft commands add `{request_id, action, saved: false}` because they only open an owner-reviewed Desktop draft. diff --git a/desktop/src/features/messages/ui/MessageComposer.tsx b/desktop/src/features/messages/ui/MessageComposer.tsx index 05e568f174..e77e2e0d55 100644 --- a/desktop/src/features/messages/ui/MessageComposer.tsx +++ b/desktop/src/features/messages/ui/MessageComposer.tsx @@ -341,7 +341,6 @@ function MessageComposerImpl({ drafts, emojiAutocomplete, mentions, - onAddressedAgentsSendStarted: addressPulse.pulseMany, onAddressedAgentsComposerCleared: (pubkeys) => restoreAddressedAgentMentionsRef.current(pubkeys), onAddressedAgentsSendFailed: addressPulse.shakeMany, diff --git a/desktop/src/features/messages/ui/useMentionSendFlow.ts b/desktop/src/features/messages/ui/useMentionSendFlow.ts index 10ca033272..0a4d89c049 100644 --- a/desktop/src/features/messages/ui/useMentionSendFlow.ts +++ b/desktop/src/features/messages/ui/useMentionSendFlow.ts @@ -52,7 +52,6 @@ export function useMentionSendFlow({ emojiAutocomplete, mentions, onPrepareSendChannel, - onAddressedAgentsSendStarted, onAddressedAgentsComposerCleared, onAddressedAgentsSendFailed, onAddressedAgentsSendSucceeded, @@ -404,9 +403,6 @@ export function useMentionSendFlow({ draft.capturedChannelId === channelIdRef.current || channelIdRef.current === null ) { - if (draft.addressedAgentPubkeys.length > 0) { - onAddressedAgentsSendStarted?.(draft.addressedAgentPubkeys); - } clearComposer(); if (draft.addressedAgentPubkeys.length > 0) { optimisticComposerContent = @@ -640,7 +636,6 @@ export function useMentionSendFlow({ getManagedAgentsByPubkey, mentions.isAgentPubkey, mentions.revalidateMentionPubkeys, - onAddressedAgentsSendStarted, onAddressedAgentsComposerCleared, onAddressedAgentsSendFailed, onAddressedAgentsSendSucceeded, diff --git a/desktop/src/features/messages/ui/useMentionSendFlow.types.ts b/desktop/src/features/messages/ui/useMentionSendFlow.types.ts index 496a73a94a..fe16ba2f63 100644 --- a/desktop/src/features/messages/ui/useMentionSendFlow.types.ts +++ b/desktop/src/features/messages/ui/useMentionSendFlow.types.ts @@ -19,7 +19,6 @@ export type UseMentionSendFlowOptions = { emojiAutocomplete: Pick; mentions: UseMentionsResult; onPrepareSendChannel?: (pubkeys?: string[]) => Promise; - onAddressedAgentsSendStarted?: (pubkeys: readonly string[]) => void; onAddressedAgentsComposerCleared?: (pubkeys: readonly string[]) => string; onAddressedAgentsSendFailed?: (pubkeys: readonly string[]) => void; onAddressedAgentsSendSucceeded?: ( diff --git a/desktop/tests/e2e/persistent-agent-audience.spec.ts b/desktop/tests/e2e/persistent-agent-audience.spec.ts index 12a6f07819..0b957992e2 100644 --- a/desktop/tests/e2e/persistent-agent-audience.spec.ts +++ b/desktop/tests/e2e/persistent-agent-audience.spec.ts @@ -500,7 +500,7 @@ test("the mention button opens settings and can undo an address", async ({ ).toHaveCount(0); }); -test("always-mentioned agents remain in the mention button while Enter-send resolves", async ({ +test("always-mentioned agents remain selected without replaying their animation while Enter-send resolves", async ({ page, }) => { await installAudienceFixtures(page, { @@ -549,7 +549,7 @@ test("always-mentioned agents remain in the mention button while Enter-send reso .not.toContain(""); await expect(avatar).toHaveAttribute( "data-pulse-version", - String(initialPulseVersion + 1), + String(initialPulseVersion), { timeout: 500 }, ); await expect( @@ -587,7 +587,7 @@ test("always-mentioned agents remain in the mention button while Enter-send reso ).toHaveCount(1); }); -test("a failed always-mentioned send shakes the composer avatar", async ({ +test("a failed always-mentioned send shakes the composer avatar without replaying its selection animation", async ({ page, }) => { await installAudienceFixtures(page, { @@ -610,7 +610,7 @@ test("a failed always-mentioned send shakes the composer avatar", async ({ await expect(avatar).toHaveAttribute( "data-pulse-version", - String(initialPulseVersion + 1), + String(initialPulseVersion), { timeout: 500 }, ); await expect(input).toHaveText("@Morgarita please retry");