External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

Bound NIP-RS retention and search indexing (#1771)

Browse Source 
Signed-off-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Tyler
2026-07-13 17:58:20 -04:00
committed by GitHub
co-authored by npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757
parent d75c2e913e
commit 1b4703021d
9 changed files with 1560 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-db/src/lib.rs
+749 -26
View File
@@ -2781,8 +2781,10 @@ impl Db {
///
/// Keeps only the event with the highest `created_at` per `(kind, pubkey, d_tag)`.
/// Same-second ties are broken by lowest event `id` (deterministic ordering).
/// The entire check → soft-delete → insert runs in a single transaction with
/// an advisory lock to prevent concurrent-insert races.
/// The entire check → retire old payload → insert runs in a single transaction
/// with an advisory lock to prevent concurrent-insert races. NIP-RS read-state
/// coordinates hard-delete the superseded payload and preserve a compact
/// ordering watermark; other NIP-33 kinds retain soft-deleted history.
///
/// **Channel policy:** NIP-33 replacement keys on `(kind, pubkey, d_tag)` globally —
/// `channel_id` is NOT part of the replacement key. This matches the Nostr spec:
@@ -2837,7 +2839,37 @@ impl Db {
.execute(&mut *tx)
.await?;
// Check for existing event with same (kind, pubkey, d_tag).
let d_tag_count = event
.tags
.iter()
.filter(|tag| tag.as_slice().first().is_some_and(|part| part == "d"))
.count();
let has_exact_d_tag = event.tags.iter().any(|tag| {
let parts = tag.as_slice();
parts.len() >= 2 && parts[0] == "d" && parts[1] == d_tag
});
let read_state_t_tag_count = event
.tags
.iter()
.filter(|tag| {
let parts = tag.as_slice();
parts.len() == 2 && parts[0] == "t" && parts[1] == "read-state"
})
.count();
let is_nip_rs = kind_i32 == buzz_core::kind::KIND_READ_STATE as i32
&& d_tag_count == 1
&& has_exact_d_tag
&& d_tag.strip_prefix("read-state:").is_some_and(|slot| {
slot.len() == 32
&& slot
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
})
&& read_state_t_tag_count == 1;
// Check the live head and, for NIP-RS, the compact historical ordering
// watermark. The watermark remains after a NIP-09 coordinate deletion,
// preventing a previously accepted signed blob from being resurrected.
let existing: Option<(chrono::DateTime<chrono::Utc>, Vec<u8>)> = sqlx::query_as(
"SELECT created_at, id FROM events \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL \
@@ -2849,32 +2881,76 @@ impl Db {
.bind(d_tag)
.fetch_optional(&mut *tx)
.await?;
// Stale-write protection: reject if incoming is not newer.
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 older event(s).
sqlx::query(
"UPDATE events SET deleted_at = NOW() \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL",
let watermark: Option<(chrono::DateTime<chrono::Utc>, Vec<u8>)> = if is_nip_rs {
sqlx::query_as(
"SELECT created_at, event_id FROM parameterized_event_watermarks \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4",
)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
.execute(&mut *tx)
.await?;
.fetch_optional(&mut *tx)
.await?
} else {
None
};
// Stale-write protection: reject if either durable ordering source
// dominates the incoming tuple. Equal timestamps use lowest event id.
let incoming_id = event.id.as_bytes().as_slice();
let dominated =
existing
.iter()
.chain(watermark.iter())
.any(|(accepted_ts, accepted_id)| {
created_at < *accepted_ts
|| (created_at == *accepted_ts && incoming_id >= accepted_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,
));
}
if existing.is_some() {
let statement = if is_nip_rs {
// Migration 0011 rejects regex-coordinate hard deletes from
// pre-fix writers. Authorize only this corrected NIP-RS delete,
// transaction-locally so pooled connections cannot leak it.
sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)")
.execute(&mut *tx)
.await?;
"DELETE FROM events \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL"
} else {
"UPDATE events SET deleted_at = NOW() \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL"
};
sqlx::query(statement)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
.execute(&mut *tx)
.await?;
if is_nip_rs {
if let Some((_, existing_id)) = &existing {
// Event first, mentions second: migration 0009's live-event
// fence uses this global lock order to avoid deadlocks.
sqlx::query(
"DELETE FROM event_mentions WHERE community_id = $1 AND event_id = $2",
)
.bind(community_id.as_uuid())
.bind(existing_id)
.execute(&mut *tx)
.await?;
}
}
}
// Insert the new event inside the transaction.
@@ -2911,6 +2987,24 @@ impl Db {
));
}
if is_nip_rs {
sqlx::query(
"INSERT INTO parameterized_event_watermarks \
(community_id, kind, pubkey, d_tag, created_at, event_id) \
VALUES ($1, $2, $3, $4, $5, $6) \
ON CONFLICT (community_id, kind, pubkey, d_tag) DO UPDATE SET \
created_at = EXCLUDED.created_at, event_id = EXCLUDED.event_id",
)
.bind(community_id.as_uuid())
.bind(kind_i32)
.bind(pubkey_bytes.as_slice())
.bind(d_tag)
.bind(created_at)
.bind(incoming_id)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
// Mentions are a denormalized index — safe outside the transaction.
@@ -3009,13 +3103,15 @@ mod tests {
//! channels, that fail-closed chain would go blind.
use super::*;
use buzz_core::CommunityId;
use sqlx::PgPool;
use sqlx::{Acquire, PgPool};
use uuid::Uuid;
const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz";
async fn setup_db() -> Db {
let pool = PgPool::connect(TEST_DB_URL)
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)
@@ -3033,6 +3129,633 @@ mod tests {
id
}
#[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 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<String> = 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() {
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:{}", "b".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 a = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "A")
.tags(tags.clone())
.custom_created_at(Timestamp::from(base))
.sign_with_keys(&keys)
.expect("sign A");
let x = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "X")
.tags(tags.clone())
.custom_created_at(Timestamp::from(base + 1))
.sign_with_keys(&keys)
.expect("sign X");
let b = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "B")
.tags(tags.clone())
.custom_created_at(Timestamp::from(base + 2))
.sign_with_keys(&keys)
.expect("sign B");
let c = EventBuilder::new(Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16), "C")
.tags(tags)
.custom_created_at(Timestamp::from(base + 3))
.sign_with_keys(&keys)
.expect("sign C");
async fn legacy_insert(
pool: &PgPool,
community: CommunityId,
event: &nostr::Event,
d_tag: &str,
) -> std::result::Result<sqlx::postgres::PgQueryResult, sqlx::Error> {
sqlx::query(
"INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, d_tag) \
VALUES ($1, $2, $3, to_timestamp($4), $5, $6, $7, $8, NOW(), $9) ON CONFLICT DO NOTHING",
)
.bind(community.as_uuid())
.bind(event.id.as_bytes().as_slice())
.bind(event.pubkey.to_bytes())
.bind(event.created_at.as_secs() as f64)
.bind(buzz_core::kind::KIND_READ_STATE as i32)
.bind(serde_json::to_value(&event.tags).expect("serialize tags"))
.bind(&event.content)
.bind(event.sig.serialize().as_slice())
.bind(d_tag)
.execute(pool)
.await
}
legacy_insert(&db.pool, community, &a, &d_tag)
.await
.expect("legacy insert A");
let duplicate = legacy_insert(&db.pool, community, &a, &d_tag)
.await
.expect("legacy duplicate A remains idempotent");
assert_eq!(duplicate.rows_affected(), 0);
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("c".repeat(64))
.bind(a.id.as_bytes().as_slice())
.bind(a.created_at.as_secs() as f64)
.execute(&db.pool)
.await
.expect("insert live mention");
// Emulate the pre-PR replacement path after migration 0007: soft-delete
// the live row, then insert B without any application watermark write.
sqlx::query(
"UPDATE events SET deleted_at=NOW() \
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)
.execute(&db.pool)
.await
.expect("legacy soft-delete A");
let mentions_after_delete: i64 = sqlx::query_scalar(
"SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2",
)
.bind(community.as_uuid())
.bind(a.id.as_bytes().as_slice())
.fetch_one(&db.pool)
.await
.expect("count mentions after delete");
assert_eq!(mentions_after_delete, 0);
let stale_mention = 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("d".repeat(64))
.bind(a.id.as_bytes().as_slice())
.bind(a.created_at.as_secs() as f64)
.execute(&db.pool)
.await
.expect("stale post-commit mention is skipped");
assert_eq!(stale_mention.rows_affected(), 0);
legacy_insert(&db.pool, community, &b, &d_tag)
.await
.expect("legacy insert B");
let duplicate_b = legacy_insert(&db.pool, community, &b, &d_tag)
.await
.expect("live duplicate B is skipped");
assert_eq!(duplicate_b.rows_affected(), 0);
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("e".repeat(64))
.bind(b.id.as_bytes().as_slice())
.bind(b.created_at.as_secs() as f64)
.execute(&db.pool)
.await
.expect("insert B mention");
// Exercise the new Rust hard-delete path independently. An in-flight
// mention holds KEY SHARE on B, so replacement by C must block, then
// complete after the mention commits and remove both B and its mention.
let mut rust_mention_tx = db
.pool
.begin()
.await
.expect("begin Rust mention transaction");
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) ON CONFLICT DO NOTHING",
)
.bind(community.as_uuid())
.bind("e".repeat(64))
.bind(b.id.as_bytes().as_slice())
.bind(b.created_at.as_secs() as f64)
.execute(&mut *rust_mention_tx)
.await
.expect("hold B live-event key-share lock");
let replace_db = db.clone();
let replace_d_tag = d_tag.clone();
let replace_c = c.clone();
let replace_task = tokio::spawn(async move {
replace_db
.replace_parameterized_event(community, &replace_c, &replace_d_tag, None)
.await
});
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
assert!(
!replace_task.is_finished(),
"Rust hard delete should wait for mention lock"
);
rust_mention_tx
.commit()
.await
.expect("release Rust mention lock");
let replaced = tokio::time::timeout(std::time::Duration::from_secs(2), replace_task)
.await
.expect("Rust hard delete deadlocked with mention insert")
.expect("replacement task panicked")
.expect("replace B with C");
assert!(replaced.1, "C must replace B");
let b_mentions: i64 = sqlx::query_scalar(
"SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2",
)
.bind(community.as_uuid())
.bind(b.id.as_bytes().as_slice())
.fetch_one(&db.pool)
.await
.expect("count B mentions after Rust replacement");
assert_eq!(b_mentions, 0);
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("f".repeat(64))
.bind(c.id.as_bytes().as_slice())
.bind(c.created_at.as_secs() as f64)
.execute(&db.pool)
.await
.expect("insert C mention");
// Exercise legacy UPDATE-trigger deletion with the same barrier. While
// deletion waits on C's KEY SHARE lock, an exact replay must already be
// a zero-row trigger no-op; it must not wait for deletion or resurrect C.
let mut legacy_mention_tx = db
.pool
.begin()
.await
.expect("begin legacy mention transaction");
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) ON CONFLICT DO NOTHING",
)
.bind(community.as_uuid())
.bind("f".repeat(64))
.bind(c.id.as_bytes().as_slice())
.bind(c.created_at.as_secs() as f64)
.execute(&mut *legacy_mention_tx)
.await
.expect("hold C live-event key-share lock");
let delete_pool = db.pool.clone();
let delete_pubkey = keys.public_key().to_bytes();
let delete_d_tag = d_tag.clone();
let delete_task = tokio::spawn(async move {
sqlx::query(
"UPDATE events SET deleted_at=NOW() \
WHERE community_id=$1 AND kind=30078 AND pubkey=$2 AND d_tag=$3 AND deleted_at IS NULL",
)
.bind(community.as_uuid())
.bind(delete_pubkey)
.bind(delete_d_tag)
.execute(&delete_pool)
.await
});
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
assert!(
!delete_task.is_finished(),
"legacy delete should wait for mention lock"
);
let replay_while_delete_waits = legacy_insert(&db.pool, community, &c, &d_tag)
.await
.expect("concurrent exact C replay is skipped");
assert_eq!(replay_while_delete_waits.rows_affected(), 0);
legacy_mention_tx
.commit()
.await
.expect("release legacy mention lock");
tokio::time::timeout(std::time::Duration::from_secs(2), delete_task)
.await
.expect("legacy delete deadlocked with mention insert")
.expect("delete task panicked")
.expect("legacy NIP-09 delete C");
let payloads: 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 retained payloads");
assert_eq!(
payloads, 0,
"legacy soft deletes must not retain NIP-RS payloads"
);
// Opposite commit order: deletion has committed before exact replay.
// Equality remains an observable zero-row no-op, never a resurrection.
let replay_c = legacy_insert(&db.pool, community, &c, &d_tag)
.await
.expect("post-delete exact C replay is skipped");
assert_eq!(replay_c.rows_affected(), 0);
let payloads_after_exact_replay: 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 payloads after exact replay");
assert_eq!(payloads_after_exact_replay, 0);
let replay = legacy_insert(&db.pool, community, &x, &d_tag).await;
assert!(
replay.is_err(),
"database guard must reject A < X < C replay"
);
let watermark: (chrono::DateTime<chrono::Utc>, Vec<u8>) = sqlx::query_as(
"SELECT created_at, event_id 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("read C watermark");
assert_eq!(watermark.0.timestamp(), base as i64 + 3);
assert_eq!(watermark.1, c.id.as_bytes().as_slice());
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn lookup_community_by_host_matches_case_insensitive_host_index() {
crates/buzz-db/src/migration.rs
+246 -1
View File
@@ -12,10 +12,81 @@ static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("../../migrations");
/// Run all pending Buzz database migrations.
pub async fn run_migrations(pool: &PgPool) -> Result<()> {
reject_legacy_nip_rs_cardinality_ambiguity(pool).await?;
MIGRATOR.run(pool).await?;
Ok(())
}
/// Migration 0007 is checksum-frozen and predates exact NIP-RS tag-cardinality
/// enforcement. A populated database still on 0001-0006 must not let 0007
/// irreversibly purge duplicate-tag history. Fail before sqlx starts its
/// migration transaction so an operator can inspect and repair those rows.
async fn reject_legacy_nip_rs_cardinality_ambiguity(pool: &PgPool) -> Result<()> {
let migrations_table: Option<String> =
sqlx::query_scalar("SELECT to_regclass('_sqlx_migrations')::text")
.fetch_one(pool)
.await?;
if migrations_table.is_none() {
return Ok(());
}
let applied: Option<i64> =
sqlx::query_scalar("SELECT max(version) FROM _sqlx_migrations WHERE success")
.fetch_one(pool)
.await?;
if applied.is_none_or(|version| version >= 7) {
return Ok(());
}
let ambiguous: bool = sqlx::query_scalar(
"SELECT EXISTS (\
SELECT 1 FROM events e \
WHERE e.kind = 30078 \
AND e.d_tag ~ '^read-state:[0-9a-f]{32}$' \
AND (\
jsonb_typeof(e.tags) IS DISTINCT FROM 'array' \
OR (\
EXISTS (\
SELECT 1 FROM jsonb_array_elements(\
CASE WHEN jsonb_typeof(e.tags) = 'array' THEN e.tags ELSE '[]'::jsonb END\
) tag \
WHERE tag = '[\"t\", \"read-state\"]'::jsonb\
) \
AND (\
(SELECT count(*) FROM jsonb_array_elements(\
CASE WHEN jsonb_typeof(e.tags) = 'array' THEN e.tags ELSE '[]'::jsonb END\
) tag \
WHERE jsonb_typeof(tag) = 'array' \
AND tag->0 = '\"d\"'::jsonb) <> 1 \
OR NOT EXISTS (\
SELECT 1 FROM jsonb_array_elements(\
CASE WHEN jsonb_typeof(e.tags) = 'array' THEN e.tags ELSE '[]'::jsonb END\
) tag \
WHERE jsonb_typeof(tag) = 'array' \
AND jsonb_array_length(tag) >= 2 \
AND jsonb_typeof(tag->1) = 'string' \
AND tag->>0 = 'd' \
AND tag->>1 = e.d_tag\
) \
OR (SELECT count(*) FROM jsonb_array_elements(\
CASE WHEN jsonb_typeof(e.tags) = 'array' THEN e.tags ELSE '[]'::jsonb END\
) tag WHERE tag = '[\"t\", \"read-state\"]'::jsonb) <> 1\
)\
)\
)\
)",
)
.fetch_one(pool)
.await?;
if ambiguous {
return Err(crate::DbError::InvalidData(
"NIP-RS migration blocked: pre-0007 database contains kind-30078 rows with ambiguous d/t tag cardinality; repair or remove those nonconforming rows before retrying"
.into(),
));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
@@ -471,7 +542,7 @@ mod tests {
let mut migrations: Vec<_> = MIGRATOR.iter().collect();
migrations.sort_by_key(|migration| migration.version);
assert_eq!(migrations.len(), 6);
assert_eq!(migrations.len(), 11);
assert_eq!(migrations[0].version, 1);
assert_eq!(&*migrations[0].description, "initial schema");
assert!(migrations[0]
@@ -555,6 +626,90 @@ mod tests {
);
}
assert!(!migrations[0].sql.as_str().contains("moderation_reports"));
// NIP-RS retention is additive and boot-safe: seed replay watermarks
// before deleting payload history, without rewriting search storage.
assert_eq!(migrations[6].version, 7);
assert!(migrations[6]
.sql
.as_str()
.contains("LOCK TABLE events IN SHARE ROW EXCLUSIVE MODE"));
assert!(migrations[6]
.sql
.as_str()
.contains("CREATE TABLE parameterized_event_watermarks"));
assert!(migrations[6]
.sql
.as_str()
.contains("INSERT INTO parameterized_event_watermarks"));
assert!(migrations[6]
.sql
.as_str()
.contains("CREATE INDEX idx_event_mentions_community_event"));
assert!(migrations[6]
.sql
.as_str()
.contains("NIP-RS retention blocked: deleted event outranks live head"));
assert!(migrations[6]
.sql
.as_str()
.contains("DELETE FROM events old"));
assert!(!migrations[6]
.sql
.as_str()
.contains("ALTER TABLE events DROP COLUMN search_tsv"));
// Fresh installs opt into the positive search allowlist without making
// populated databases rewrite their events heap during relay startup.
assert_eq!(migrations[7].version, 8);
assert!(migrations[7]
.sql
.as_str()
.contains("IF NOT EXISTS (SELECT 1 FROM events LIMIT 1)"));
assert!(migrations[7]
.sql
.as_str()
.contains("CASE WHEN kind IN (0, 9, 40002, 45001, 45003)"));
assert!(migrations[7].sql.as_str().contains("ELSE NULL::tsvector"));
// Mixed-version guards are additive because 0007/0008 may already be
// recorded by a running relay and their sqlx checksums are immutable.
assert_eq!(migrations[8].version, 9);
assert!(migrations[8]
.sql
.as_str()
.contains("CREATE TRIGGER trg_events_nip_rs_watermark"));
assert!(migrations[8]
.sql
.as_str()
.contains("stale NIP-RS event rejected by durable watermark"));
assert!(migrations[8]
.sql
.as_str()
.contains("CREATE TRIGGER trg_events_purge_soft_deleted_nip_rs"));
assert!(migrations[8]
.sql
.as_str()
.contains("CREATE TRIGGER trg_event_mentions_require_live_event"));
assert_eq!(migrations[9].version, 10);
assert!(migrations[9]
.sql
.as_str()
.contains("CREATE OR REPLACE FUNCTION guard_nip_rs_watermark"));
assert!(migrations[9].sql.as_str().contains("RETURN NULL"));
assert_eq!(migrations[10].version, 11);
assert!(migrations[10]
.sql
.as_str()
.contains("CREATE OR REPLACE FUNCTION guard_nip_rs_watermark"));
assert!(migrations[10]
.sql
.as_str()
.contains("CREATE OR REPLACE FUNCTION purge_soft_deleted_nip_rs"));
assert!(migrations[10].sql.as_str().contains("tag->>0 = 'd'"));
assert!(migrations[10].sql.as_str().contains(") = 1"));
}
#[test]
@@ -730,6 +885,76 @@ mod tests {
.expect("read applied migrations")
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn pre_0007_ambiguous_nip_rs_data_blocks_without_mutation_and_allows_retry() {
let pool = connect_test_pool().await;
reset_public_schema(&pool).await;
MIGRATOR
.run_to(6, &pool)
.await
.expect("apply migrations 1-6");
let community_id = uuid::Uuid::new_v4();
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(community_id)
.bind(format!("pre-0007-{}.example", community_id.simple()))
.execute(&pool)
.await
.expect("insert community");
let event_id = vec![1_u8; 32];
let pubkey = vec![2_u8; 32];
let d_tag = format!("read-state:{}", "a".repeat(32));
let ambiguous_tags = serde_json::json!([["d", d_tag], ["d", "other"], ["t", "read-state"]]);
sqlx::query(
"INSERT INTO events \
(community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, d_tag) \
VALUES ($1, $2, $3, NOW(), 30078, $4, 'ambiguous', $5, NOW(), $6)",
)
.bind(community_id)
.bind(&event_id)
.bind(&pubkey)
.bind(&ambiguous_tags)
.bind(vec![3_u8; 64])
.bind(&d_tag)
.execute(&pool)
.await
.expect("insert ambiguous NIP-RS row");
let before_versions = applied_versions(&pool).await;
let before_row: (serde_json::Value, String) =
sqlx::query_as("SELECT tags, content FROM events WHERE community_id=$1 AND id=$2")
.bind(community_id)
.bind(&event_id)
.fetch_one(&pool)
.await
.expect("read ambiguous row before blocked migration");
let blocked = run_migrations(&pool).await;
assert!(blocked.is_err(), "ambiguous pre-0007 data must fail closed");
assert_eq!(applied_versions(&pool).await, before_versions);
let after_row: (serde_json::Value, String) =
sqlx::query_as("SELECT tags, content FROM events WHERE community_id=$1 AND id=$2")
.bind(community_id)
.bind(&event_id)
.fetch_one(&pool)
.await
.expect("blocked migration must preserve source row");
assert_eq!(after_row, before_row);
let repaired_tags = serde_json::json!([["d", d_tag], ["t", "read-state"]]);
sqlx::query("UPDATE events SET tags=$1 WHERE community_id=$2 AND id=$3")
.bind(repaired_tags)
.bind(community_id)
.bind(&event_id)
.execute(&pool)
.await
.expect("repair ambiguous row");
run_migrations(&pool)
.await
.expect("retry succeeds after operator repair");
assert_eq!(applied_versions(&pool).await.last().copied(), Some(11));
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn run_migrations_applies_consolidated_initial_schema_on_fresh_database() {
@@ -769,5 +994,25 @@ mod tests {
);
assert!(exists, "migration should create {table}");
}
let search_expression: String = sqlx::query_scalar(
"SELECT pg_get_expr(adbin, adrelid) \
FROM pg_attrdef \
WHERE adrelid = 'events'::regclass \
AND adnum = (SELECT attnum FROM pg_attribute \
WHERE attrelid = 'events'::regclass \
AND attname = 'search_tsv')",
)
.fetch_one(&pool)
.await
.expect("read fresh-install search expression");
assert!(
search_expression.contains("ARRAY[0, 9, 40002, 45001, 45003]"),
"fresh-install search allowlist has the wrong kinds: {search_expression}"
);
assert!(
search_expression.contains("ELSE NULL::tsvector"),
"fresh installs must default non-allowlisted kinds to NULL: {search_expression}"
);
}
}
crates/buzz-search/tests/fts_integration.rs
+30 -17
View File
@@ -2,9 +2,9 @@
//!
//! Run with a local PG: `BUZZ_TEST_DATABASE_URL=postgres://buzz:buzz_dev@localhost:5432/buzz cargo test -p buzz-search --tests -- --include-ignored`
//!
//! Each test creates a uniquely-named schema, applies all five migrations in
//! order (0001 → 0002 → 0003 → 0004 → 0005) into it, exercises a scenario, and drops
//! it. Tests are parallel-safe.
//! Each test creates a uniquely-named schema, applies the full migration chain
//! (0001 through 0008) into it, exercises a scenario, and drops it. Tests are
//! parallel-safe.
use buzz_core::{
kind::{
@@ -23,6 +23,10 @@ const MIGRATION_0002_SQL: &str = include_str!("../../../migrations/0002_git_repo
const MIGRATION_0003_SQL: &str = include_str!("../../../migrations/0003_community_icon.sql");
const MIGRATION_0004_SQL: &str = include_str!("../../../migrations/0004_events_tags_gin.sql");
const MIGRATION_0005_SQL: &str = include_str!("../../../migrations/0005_agent_turn_metric_fts.sql");
const MIGRATION_0006_SQL: &str = include_str!("../../../migrations/0006_moderation.sql");
const MIGRATION_0007_SQL: &str = include_str!("../../../migrations/0007_nip_rs_retention.sql");
const MIGRATION_0008_SQL: &str =
include_str!("../../../migrations/0008_fresh_install_search_allowlist.sql");
async fn setup() -> (PgPool, String) {
let url = std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.to_string());
@@ -64,6 +68,15 @@ async fn setup() -> (PgPool, String) {
pool.execute(MIGRATION_0005_SQL)
.await
.expect("apply 0005 migration");
pool.execute(MIGRATION_0006_SQL)
.await
.expect("apply 0006 migration");
pool.execute(MIGRATION_0007_SQL)
.await
.expect("apply 0007 migration");
pool.execute(MIGRATION_0008_SQL)
.await
.expect("apply 0008 migration");
(pool, schema)
}
@@ -148,7 +161,7 @@ async fn search_finds_event_in_same_community() {
c_a,
evt_id,
pk,
1,
9,
"hello wonderland — buzz everyone",
None,
1700000000,
@@ -174,7 +187,7 @@ async fn search_finds_event_in_same_community() {
assert_eq!(result.hits.len(), 1);
assert_eq!(result.hits[0].event_id, evt_id);
assert_eq!(result.hits[0].kind, 1);
assert_eq!(result.hits[0].kind, 9);
assert_eq!(result.hits[0].created_at, 1700000000);
assert!(result.hits[0].rank > 0.0);
@@ -196,7 +209,7 @@ async fn search_does_not_return_other_community_events() {
c_a,
rand_bytes32(),
pk,
1,
9,
"only-in-a unique-token-xyz",
None,
1700000000,
@@ -547,7 +560,7 @@ async fn channel_scope_restricts_results() {
c,
rand_bytes32(),
pk,
1,
9,
"shared-token in ch-a",
Some(ch_a),
1700000000,
@@ -558,7 +571,7 @@ async fn channel_scope_restricts_results() {
c,
rand_bytes32(),
pk,
1,
9,
"shared-token in ch-b",
Some(ch_b),
1700000001,
@@ -569,7 +582,7 @@ async fn channel_scope_restricts_results() {
c,
rand_bytes32(),
pk,
1,
9,
"shared-token global",
None,
1700000002,
@@ -662,7 +675,7 @@ async fn deleted_events_are_excluded() {
let c = mk_community(&pool, "a.example").await;
let evt_id = rand_bytes32();
let pk = rand_bytes32();
insert_event(&pool, c, evt_id, pk, 1, "deleted-token-q", None, 1700000000).await;
insert_event(&pool, c, evt_id, pk, 9, "deleted-token-q", None, 1700000000).await;
// Soft-delete
sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2")
@@ -740,7 +753,7 @@ async fn since_until_filters() {
c,
rand_bytes32(),
pk,
1,
9,
"time-token-zz at A",
None,
1_700_000_000,
@@ -751,7 +764,7 @@ async fn since_until_filters() {
c,
rand_bytes32(),
pk,
1,
9,
"time-token-zz at B",
None,
1_700_010_000,
@@ -762,7 +775,7 @@ async fn since_until_filters() {
c,
rand_bytes32(),
pk,
1,
9,
"time-token-zz at C",
None,
1_700_020_000,
@@ -805,7 +818,7 @@ async fn pagination_works() {
c,
rand_bytes32(),
pk,
1,
9,
"page-token-qq",
None,
1_700_000_000 + i,
@@ -893,7 +906,7 @@ async fn channel_less_only_excludes_per_channel_events() {
c,
rand_bytes32(),
pk,
1,
9,
"fence-token in ch-a",
Some(ch_a),
1_700_000_000,
@@ -904,7 +917,7 @@ async fn channel_less_only_excludes_per_channel_events() {
c,
rand_bytes32(),
pk,
1,
9,
"fence-token in ch-b",
Some(ch_b),
1_700_000_001,
@@ -915,7 +928,7 @@ async fn channel_less_only_excludes_per_channel_events() {
c,
rand_bytes32(),
pk,
1,
9,
"fence-token channel-less",
None,
1_700_000_002,
migrations/0007_nip_rs_retention.sql
+140
View File
@@ -0,0 +1,140 @@
-- Bound NIP-RS storage while preserving NIP-33 replay ordering.
--
-- The payload table previously retained every superseded kind:30078 event as a
-- soft-deleted row. Besides keeping the encrypted blob, search_tsv tokenized it
-- and the GIN index amplified it further. A compact ordering watermark retains
-- the only historical fact replacement needs without retaining user payloads.
-- The relay may still have old instances writing during a rolling deploy. Hold a
-- table-level writer lock for this transaction so the seed is a complete
-- high-water mark: without it, an old instance could insert between the seed
-- and purge, then a later NIP-09 deletion could reopen a replay window. Reads
-- remain available; inserts, updates, and deletes wait for migration commit.
LOCK TABLE events IN SHARE ROW EXCLUSIVE MODE;
CREATE TABLE parameterized_event_watermarks (
community_id UUID NOT NULL REFERENCES communities(id),
kind INT NOT NULL,
pubkey BYTEA NOT NULL,
d_tag TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL,
event_id BYTEA NOT NULL,
PRIMARY KEY (community_id, kind, pubkey, d_tag)
);
-- Superseded read-state events normally have no p-tags, but malformed/legacy
-- rows can. Serve defensive mention cleanup without a per-replacement seq scan.
CREATE INDEX idx_event_mentions_community_event
ON event_mentions (community_id, event_id);
-- Fail closed on legacy anomalies that would make a deleted tuple outrank a
-- live head. Seeding that tuple would freeze legitimate writes; ignoring it
-- would weaken replay protection. Operators must inspect and repair such a
-- coordinate before retrying the migration.
DO $$
BEGIN
IF EXISTS (
SELECT 1
FROM events dead
JOIN LATERAL (
SELECT live.created_at, live.id
FROM events live
WHERE live.community_id = dead.community_id
AND live.kind = dead.kind
AND live.pubkey = dead.pubkey
AND live.d_tag = dead.d_tag
AND live.deleted_at IS NULL
ORDER BY live.created_at DESC, live.id ASC
LIMIT 1
) live ON TRUE
WHERE dead.kind = 30078
AND dead.deleted_at IS NOT NULL
AND dead.d_tag ~ '^read-state:[0-9a-f]{32}$'
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(dead.tags) tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) = 2
AND tag->>0 = 't'
AND tag->>1 = 'read-state'
)
AND (dead.created_at > live.created_at
OR (dead.created_at = live.created_at AND dead.id < live.id))
) THEN
RAISE EXCEPTION 'NIP-RS retention blocked: deleted event outranks live head';
END IF;
END $$;
-- Seed the greatest accepted tuple (newest created_at; lowest id wins ties)
-- from live and historical NIP-RS rows before removing payload history.
INSERT INTO parameterized_event_watermarks
(community_id, kind, pubkey, d_tag, created_at, event_id)
SELECT DISTINCT ON (community_id, kind, pubkey, d_tag)
community_id, kind, pubkey, d_tag, created_at, id
FROM events e
WHERE kind = 30078
AND d_tag ~ '^read-state:[0-9a-f]{32}$'
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(e.tags) tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) = 2
AND tag->>0 = 't'
AND tag->>1 = 'read-state'
)
ORDER BY community_id, kind, pubkey, d_tag, created_at DESC, id ASC;
-- Mentions are denormalized and do not have a foreign key to the partitioned
-- events table. Delete any defensive/legacy rows for the exact purge set first.
DELETE FROM event_mentions mention
USING events old
WHERE mention.community_id = old.community_id
AND mention.event_id = old.id
AND mention.event_created_at = old.created_at
AND old.kind = 30078
AND old.deleted_at IS NOT NULL
AND old.d_tag ~ '^read-state:[0-9a-f]{32}$'
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(old.tags) tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) = 2
AND tag->>0 = 't'
AND tag->>1 = 'read-state'
)
AND EXISTS (
SELECT 1
FROM events live
WHERE live.community_id = old.community_id
AND live.kind = old.kind
AND live.pubkey = old.pubkey
AND live.d_tag = old.d_tag
AND live.deleted_at IS NULL
AND (live.created_at > old.created_at
OR (live.created_at = old.created_at AND live.id < old.id))
);
-- Purge only replacement history with a strictly dominating live head. Rows
-- deleted explicitly through NIP-09 have no live head and remain untouched.
DELETE FROM events old
WHERE old.kind = 30078
AND old.deleted_at IS NOT NULL
AND old.d_tag ~ '^read-state:[0-9a-f]{32}$'
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(old.tags) tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) = 2
AND tag->>0 = 't'
AND tag->>1 = 'read-state'
)
AND EXISTS (
SELECT 1
FROM events live
WHERE live.community_id = old.community_id
AND live.kind = old.kind
AND live.pubkey = old.pubkey
AND live.d_tag = old.d_tag
AND live.deleted_at IS NULL
AND (live.created_at > old.created_at
OR (live.created_at = old.created_at AND live.id < old.id))
);
migrations/0008_fresh_install_search_allowlist.sql
+23
View File
@@ -0,0 +1,23 @@
-- Give new, empty installations the positive FTS allowlist without rewriting
-- populated databases during relay startup. Existing installations keep their
-- current search_tsv expression until an operator runs the sized out-of-band
-- maintenance script in scripts/maintenance/nip_rs_search_allowlist.sql.
--
-- Serialize the emptiness check with event writers. Reads remain available on
-- populated databases; an actually empty table upgrades briefly to ACCESS
-- EXCLUSIVE for the generated-column replacement and index build.
LOCK TABLE events IN SHARE ROW EXCLUSIVE MODE;
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM events LIMIT 1) THEN
ALTER TABLE events DROP COLUMN search_tsv;
ALTER TABLE events ADD COLUMN search_tsv TSVECTOR GENERATED ALWAYS AS (
CASE WHEN kind IN (0, 9, 40002, 45001, 45003)
THEN to_tsvector('simple', content)
ELSE NULL::tsvector
END
) STORED;
CREATE INDEX idx_events_search_tsv ON events USING GIN (search_tsv);
END IF;
END $$;
migrations/0009_nip_rs_database_guards.sql
+135
View File
@@ -0,0 +1,135 @@
-- Enforce NIP-RS retention across mixed relay versions.
--
-- Migration 0007 is already published and checksum-frozen. These database
-- triggers are additive so databases that applied 0007/0008 can upgrade safely,
-- while pre-PR relay binaries cannot bypass watermark or payload-retention rules.
-- Keep the invariant in PostgreSQL so it also covers pre-migration relay
-- binaries during a rolling deployment. Every conforming NIP-RS insert must
-- advance the watermark; an insert older than the greatest accepted tuple is
-- rejected even when no live row remains.
CREATE FUNCTION guard_nip_rs_watermark() RETURNS trigger AS $$
DECLARE
advanced BOOLEAN;
BEGIN
IF NEW.kind = 30078
AND NEW.d_tag ~ '^read-state:[0-9a-f]{32}$'
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(NEW.tags) tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) = 2
AND tag->>0 = 't'
AND tag->>1 = 'read-state'
) THEN
INSERT INTO parameterized_event_watermarks
(community_id, kind, pubkey, d_tag, created_at, event_id)
VALUES
(NEW.community_id, NEW.kind, NEW.pubkey, NEW.d_tag, NEW.created_at, NEW.id)
ON CONFLICT (community_id, kind, pubkey, d_tag) DO UPDATE SET
created_at = EXCLUDED.created_at,
event_id = EXCLUDED.event_id
WHERE EXCLUDED.created_at > parameterized_event_watermarks.created_at
OR (EXCLUDED.created_at = parameterized_event_watermarks.created_at
AND EXCLUDED.event_id < parameterized_event_watermarks.event_id)
RETURNING TRUE INTO advanced;
IF NOT COALESCE(advanced, FALSE) THEN
-- Let an exact duplicate reach the events uniqueness constraint so
-- legacy `ON CONFLICT DO NOTHING` keeps its existing idempotence.
IF EXISTS (
SELECT 1
FROM parameterized_event_watermarks watermark
JOIN events live
ON live.community_id = watermark.community_id
AND live.kind = watermark.kind
AND live.pubkey = watermark.pubkey
AND live.d_tag = watermark.d_tag
AND live.created_at = watermark.created_at
AND live.id = watermark.event_id
AND live.deleted_at IS NULL
WHERE watermark.community_id = NEW.community_id
AND watermark.kind = NEW.kind
AND watermark.pubkey = NEW.pubkey
AND watermark.d_tag = NEW.d_tag
AND watermark.created_at = NEW.created_at
AND watermark.event_id = NEW.id
) THEN
RETURN NEW;
END IF;
RAISE EXCEPTION 'stale NIP-RS event rejected by durable watermark'
USING ERRCODE = 'check_violation';
END IF;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trg_events_nip_rs_watermark
BEFORE INSERT ON events
FOR EACH ROW EXECUTE FUNCTION guard_nip_rs_watermark();
-- NIP-RS payloads have no historical product value. Enforce physical removal
-- in the database when old relay binaries use their legacy soft-delete path,
-- including NIP-09 coordinate deletion during a mixed-version rollout.
CREATE FUNCTION purge_soft_deleted_nip_rs() RETURNS trigger AS $$
BEGIN
IF OLD.deleted_at IS NULL
AND NEW.deleted_at IS NOT NULL
AND NEW.kind = 30078
AND NEW.d_tag ~ '^read-state:[0-9a-f]{32}$'
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(NEW.tags) tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) = 2
AND tag->>0 = 't'
AND tag->>1 = 'read-state'
) THEN
DELETE FROM events
WHERE community_id = NEW.community_id
AND created_at = NEW.created_at
AND id = NEW.id;
DELETE FROM event_mentions
WHERE community_id = NEW.community_id AND event_id = NEW.id;
END IF;
RETURN NULL;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trg_events_purge_soft_deleted_nip_rs
AFTER UPDATE OF deleted_at ON events
FOR EACH ROW EXECUTE FUNCTION purge_soft_deleted_nip_rs();
-- Mention indexing runs after the event transaction commits. Lock the live event
-- row while a mention is inserted so a concurrent hard delete cannot leave an
-- orphan behind; if deletion already won, silently skip the stale index row.
CREATE FUNCTION guard_event_mention_live() RETURNS trigger AS $$
BEGIN
IF NEW.event_kind IS DISTINCT FROM 30078 THEN
RETURN NEW;
END IF;
PERFORM 1
FROM events
WHERE community_id = NEW.community_id
AND id = NEW.event_id
AND created_at = NEW.event_created_at
AND deleted_at IS NULL
FOR KEY SHARE;
IF NOT FOUND THEN
RETURN NULL;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trg_event_mentions_require_live_event
BEFORE INSERT ON event_mentions
FOR EACH ROW EXECUTE FUNCTION guard_event_mention_live();
migrations/0010_nip_rs_exact_replay_guard.sql
+55
View File
@@ -0,0 +1,55 @@
-- Replace the published 0009 watermark guard without changing its checksum.
-- Exact replay is a durable coordinate-level no-op, independent of whether the
-- physically retained payload still exists.
CREATE OR REPLACE FUNCTION guard_nip_rs_watermark() RETURNS trigger AS $$
DECLARE
advanced BOOLEAN;
BEGIN
IF NEW.kind = 30078
AND NEW.d_tag ~ '^read-state:[0-9a-f]{32}$'
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(NEW.tags) tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) = 2
AND tag->>0 = 't'
AND tag->>1 = 'read-state'
) THEN
INSERT INTO parameterized_event_watermarks
(community_id, kind, pubkey, d_tag, created_at, event_id)
VALUES
(NEW.community_id, NEW.kind, NEW.pubkey, NEW.d_tag, NEW.created_at, NEW.id)
ON CONFLICT (community_id, kind, pubkey, d_tag) DO UPDATE SET
created_at = EXCLUDED.created_at,
event_id = EXCLUDED.event_id
WHERE EXCLUDED.created_at > parameterized_event_watermarks.created_at
OR (EXCLUDED.created_at = parameterized_event_watermarks.created_at
AND EXCLUDED.event_id < parameterized_event_watermarks.event_id)
RETURNING TRUE INTO advanced;
IF NOT COALESCE(advanced, FALSE) THEN
-- Exact equality is idempotent at the durable coordinate level,
-- whether or not its payload is still live. Skip it in the trigger
-- so concurrent physical deletion cannot create a resurrection
-- window between an existence check and uniqueness enforcement.
IF EXISTS (
SELECT 1
FROM parameterized_event_watermarks
WHERE community_id = NEW.community_id
AND kind = NEW.kind
AND pubkey = NEW.pubkey
AND d_tag = NEW.d_tag
AND created_at = NEW.created_at
AND event_id = NEW.id
) THEN
RETURN NULL;
END IF;
RAISE EXCEPTION 'stale NIP-RS event rejected by durable watermark'
USING ERRCODE = 'check_violation';
END IF;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
migrations/0011_nip_rs_exact_tag_cardinality.sql
+162
View File
@@ -0,0 +1,162 @@
-- Match NIP-RS's exact tag cardinality in mixed-version database guards.
-- Published migrations 0007-0010 remain checksum-frozen.
-- Remove polluted watermarks only when their exact source payload still exists
-- and proves the event was nonconforming. Missing source payloads are left
-- untouched: they may be legitimate NIP-09-deleted read state and have no
-- remaining provenance that permits safe automatic classification.
DELETE FROM parameterized_event_watermarks watermark
USING events source
WHERE source.community_id = watermark.community_id
AND source.kind = watermark.kind
AND source.pubkey = watermark.pubkey
AND source.d_tag = watermark.d_tag
AND source.created_at = watermark.created_at
AND source.id = watermark.event_id
AND source.kind = 30078
AND NOT (
source.d_tag ~ '^read-state:[0-9a-f]{32}$'
AND (
SELECT count(*)
FROM jsonb_array_elements(CASE WHEN jsonb_typeof(source.tags) = 'array' THEN source.tags ELSE '[]'::jsonb END) tag
WHERE jsonb_typeof(tag) = 'array'
AND tag->0 = '"d"'::jsonb
) = 1
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(CASE WHEN jsonb_typeof(source.tags) = 'array' THEN source.tags ELSE '[]'::jsonb END) tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) >= 2
AND jsonb_typeof(tag->1) = 'string'
AND tag->>0 = 'd'
AND tag->>1 = source.d_tag
)
AND (
SELECT count(*)
FROM jsonb_array_elements(CASE WHEN jsonb_typeof(source.tags) = 'array' THEN source.tags ELSE '[]'::jsonb END) tag
WHERE tag = '["t", "read-state"]'::jsonb
) = 1
);
-- A relay binary from before this migration can classify an incoming event by
-- broad EXISTS predicates and hard-delete the current coordinate before its
-- corrected INSERT guard runs. Fail the whole old-writer transaction rather
-- than silently skipping the DELETE (which would permit two live rows and
-- strip the retained row's mentions). Corrected paths opt in transaction-locally.
CREATE FUNCTION guard_nip_rs_hard_delete() RETURNS trigger AS $$
BEGIN
IF current_setting('buzz.nip_rs_hard_delete', true) IS DISTINCT FROM 'on' THEN
RAISE EXCEPTION 'NIP-RS hard delete requires corrected writer opt-in'
USING ERRCODE = 'check_violation';
END IF;
RETURN OLD;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trg_events_guard_nip_rs_hard_delete
BEFORE DELETE ON events
FOR EACH ROW
WHEN (OLD.kind = 30078 AND OLD.d_tag ~ '^read-state:[0-9a-f]{32}$')
EXECUTE FUNCTION guard_nip_rs_hard_delete();
CREATE OR REPLACE FUNCTION guard_nip_rs_watermark() RETURNS trigger AS $$
DECLARE
advanced BOOLEAN;
BEGIN
IF NEW.kind = 30078
AND NEW.d_tag ~ '^read-state:[0-9a-f]{32}$'
AND (
SELECT count(*)
FROM jsonb_array_elements(CASE WHEN jsonb_typeof(NEW.tags) = 'array' THEN NEW.tags ELSE '[]'::jsonb END) tag
WHERE jsonb_typeof(tag) = 'array'
AND tag->0 = '"d"'::jsonb
) = 1
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(CASE WHEN jsonb_typeof(NEW.tags) = 'array' THEN NEW.tags ELSE '[]'::jsonb END) tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) >= 2
AND jsonb_typeof(tag->1) = 'string'
AND tag->>0 = 'd'
AND tag->>1 = NEW.d_tag
)
AND (
SELECT count(*)
FROM jsonb_array_elements(CASE WHEN jsonb_typeof(NEW.tags) = 'array' THEN NEW.tags ELSE '[]'::jsonb END) tag
WHERE tag = '["t", "read-state"]'::jsonb
) = 1 THEN
INSERT INTO parameterized_event_watermarks
(community_id, kind, pubkey, d_tag, created_at, event_id)
VALUES
(NEW.community_id, NEW.kind, NEW.pubkey, NEW.d_tag, NEW.created_at, NEW.id)
ON CONFLICT (community_id, kind, pubkey, d_tag) DO UPDATE SET
created_at = EXCLUDED.created_at,
event_id = EXCLUDED.event_id
WHERE EXCLUDED.created_at > parameterized_event_watermarks.created_at
OR (EXCLUDED.created_at = parameterized_event_watermarks.created_at
AND EXCLUDED.event_id < parameterized_event_watermarks.event_id)
RETURNING TRUE INTO advanced;
IF NOT COALESCE(advanced, FALSE) THEN
IF EXISTS (
SELECT 1
FROM parameterized_event_watermarks
WHERE community_id = NEW.community_id
AND kind = NEW.kind
AND pubkey = NEW.pubkey
AND d_tag = NEW.d_tag
AND created_at = NEW.created_at
AND event_id = NEW.id
) THEN
RETURN NULL;
END IF;
RAISE EXCEPTION 'stale NIP-RS event rejected by durable watermark'
USING ERRCODE = 'check_violation';
END IF;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE OR REPLACE FUNCTION purge_soft_deleted_nip_rs() RETURNS trigger AS $$
BEGIN
IF OLD.deleted_at IS NULL
AND NEW.deleted_at IS NOT NULL
AND NEW.kind = 30078
AND NEW.d_tag ~ '^read-state:[0-9a-f]{32}$'
AND (
SELECT count(*)
FROM jsonb_array_elements(CASE WHEN jsonb_typeof(NEW.tags) = 'array' THEN NEW.tags ELSE '[]'::jsonb END) tag
WHERE jsonb_typeof(tag) = 'array'
AND tag->0 = '"d"'::jsonb
) = 1
AND EXISTS (
SELECT 1
FROM jsonb_array_elements(CASE WHEN jsonb_typeof(NEW.tags) = 'array' THEN NEW.tags ELSE '[]'::jsonb END) tag
WHERE jsonb_typeof(tag) = 'array'
AND jsonb_array_length(tag) >= 2
AND jsonb_typeof(tag->1) = 'string'
AND tag->>0 = 'd'
AND tag->>1 = NEW.d_tag
)
AND (
SELECT count(*)
FROM jsonb_array_elements(CASE WHEN jsonb_typeof(NEW.tags) = 'array' THEN NEW.tags ELSE '[]'::jsonb END) tag
WHERE tag = '["t", "read-state"]'::jsonb
) = 1 THEN
PERFORM set_config('buzz.nip_rs_hard_delete', 'on', true);
DELETE FROM events
WHERE community_id = NEW.community_id
AND created_at = NEW.created_at
AND id = NEW.id;
DELETE FROM event_mentions
WHERE community_id = NEW.community_id AND event_id = NEW.id;
END IF;
RETURN NULL;
END;
$$ LANGUAGE plpgsql;
scripts/maintenance/nip_rs_search_allowlist.sql
+20
View File
@@ -0,0 +1,20 @@
-- OUT-OF-BAND MAINTENANCE: do not run from relay startup migrations.
--
-- This rewrites every events partition and rebuilds the partitioned GIN index.
-- Run only in a maintenance window after confirming enough free space for the
-- replacement heap/TOAST/index files plus WAL. ALTER TABLE takes ACCESS
-- EXCLUSIVE, so event reads and writes block until this transaction commits.
-- Consider combining this with the planned partition repack/reclaim operation.
BEGIN;
SET LOCAL lock_timeout = '5s';
ALTER TABLE events DROP COLUMN search_tsv;
ALTER TABLE events ADD COLUMN search_tsv TSVECTOR GENERATED ALWAYS AS (
CASE WHEN kind IN (0, 9, 40002, 45001, 45003)
THEN to_tsvector('simple', content)
ELSE NULL::tsvector
END
) STORED;
CREATE INDEX idx_events_search_tsv ON events USING GIN (search_tsv);
COMMIT;