mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
fix(archive): atomic remove-kind path + split test modules
Fix A: add remove_owner_p_kind / remove_save_subscription_kind Closes the non-atomic toggle-OFF path (Wes note 1). The toggle handlers in LocalArchiveSettingsCard were doing a TS-side read-modify-overwrite on the shared owner_p row — on toggle-OFF they called deleteSaveSubscription for the whole row, which would silently drop the *other* kind (24200 vs 44200) if subs state was stale. - store.rs: add remove_owner_p_kind mirroring merge_owner_p_kinds — same BEGIN IMMEDIATE → closure → COMMIT/ROLLBACK shape; reads current kinds, removes the target kind, then DELETEs the row if the list becomes empty or UPDATEs kinds to the reduced list otherwise. - mod.rs: add remove_save_subscription_kind Tauri command delegating to store::remove_owner_p_kind; registered in lib.rs alongside the merge cmd. - tauriArchive.ts: add removeSaveSubscriptionKind wrapper calling the new command + notifySubscriptionChange(). - LocalArchiveSettingsCard.tsx: rewrite handleObserverToggle and handleMetricToggle to use atomic commands — toggle-ON calls mergeSaveSubscriptionKinds(KIND), toggle-OFF calls removeSaveSubscriptionKind(KIND). The TS-side read-modify-overwrite and the whole-row deleteSaveSubscription branch are removed entirely. subs dependency dropped from both useCallback dep arrays. - store_tests (new file): 4 unit tests for remove_owner_p_kind: removes one kind leaving the other, deletes row on last kind, no-op when row absent, no-op when kind absent. Fix B: split archive test modules (Wes note 3) The check-file-sizes.mjs override for archive/mod.rs had ratcheted 1465→1705 across the PR series. Split both oversized test blocks: - archive/mod_tests.rs: extracted #[cfg(test)] mod_tests module from mod.rs (~1208 lines, test-only content). Wired via #[cfg(test)] #[path = "mod_tests.rs"] mod mod_tests in mod.rs. - archive/store_tests.rs: extracted #[cfg(test)] store_tests module from store.rs (~732 lines, fits under 1000 — no override needed). Wired via #[cfg(test)] #[path = "store_tests.rs"] mod store_tests in store.rs. mod.rs is now 532 lines, store.rs is 599 lines — both under 1000. The two mod.rs/store.rs overrides are replaced by a single mod_tests.rs override (1208). All 934 Rust tests pass unchanged. Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
co-authored by
Will Pfleger
Will Pfleger
parent
0f93c1160b
commit
2bbac40fe3
@@ -60,34 +60,13 @@ const overrides = new Map([
|
||||
// config-bridge: get_agent_config_surface/write_agent_config_field/put_agent_session_config
|
||||
// commands add ~40 lines. Queued to split.
|
||||
// branch cut; override bumped to cover the merged total. Queued to split.
|
||||
// archive/mod.rs carries the full test module: 899 unit tests + 4 real-relay
|
||||
// integration tests (ignored, live-relay only, wrapped in
|
||||
// #[cfg(not(target_os = "windows"))] mod real_relay). The test module is the
|
||||
// source of the overage — production logic is ~408 lines. The fix-round added
|
||||
// 2 regression tests (F2: out-of-range kind + F3: atomicity invariant).
|
||||
// read_archived_events Tauri command added ~39 lines (Phase 1 read-back).
|
||||
// E2E test-depth hardening added the owner_p content round-trip assert and
|
||||
// two empty-table drop asserts (~24 lines). Queued to split the test module
|
||||
// into archive/mod_tests.rs in a follow-up.
|
||||
// agent-metric-archive PR added 4 new unit tests (owner_p+44200 routing,
|
||||
// decrypt-success plaintext storage, decrypt-fail-closed, 24200 still
|
||||
// ephemeral) + run_batch_sync_with_keys helper (~175 lines). Same test-growth
|
||||
// category as above. Still queued to split.
|
||||
// merge_save_subscription_kinds command + owner_p-kinds TOCTOU fix adds ~30
|
||||
// lines. Atomic merge to close the concurrent-seed race. Still queued to split.
|
||||
// IMMEDIATE-tx fix: BEGIN IMMEDIATE replaces DEFERRED unchecked_transaction
|
||||
// in merge_owner_p_kinds comment block (~5 lines). Still queued to split.
|
||||
// doc-comment: mixed row-shape invariant on read_archived_events (~5 lines).
|
||||
["src-tauri/src/archive/mod.rs", 1705],
|
||||
// archive/store.rs: merge_owner_p_kinds fn (read+union+upsert under a single
|
||||
// SQLite tx) + 4 unit tests (create-when-none, adds-kind, idempotent,
|
||||
// concurrent-interleave). Load-bearing TOCTOU fix for the owner_p shared row.
|
||||
// Queued to split test module into store_tests.rs in a follow-up.
|
||||
// IMMEDIATE-tx fix: replaces mislabeled sequential test with a real
|
||||
// two-connection WAL regression test (tempfile + std::thread + Barrier,
|
||||
// ~85 lines). The new test exercises the actual concurrent write path and
|
||||
// fails fast if IMMEDIATE guard is removed. Still queued to split.
|
||||
["src-tauri/src/archive/store.rs", 1179],
|
||||
// archive/mod_tests.rs carries the full test module for archive/mod.rs:
|
||||
// unit tests + 4 real-relay integration tests (ignored, live-relay only).
|
||||
// Production logic in mod.rs is now ~527 lines (under 1000). mod_tests.rs
|
||||
// is test-only content; the override covers the test growth accumulated
|
||||
// across the local-archive + agent-metric-archive PR series. store_tests.rs
|
||||
// (~731 lines) is under 1000 so needs no override.
|
||||
["src-tauri/src/archive/mod_tests.rs", 1208],
|
||||
["src-tauri/src/commands/agents.rs", 1437],
|
||||
// #1418 read-path fix: get_thread_replies' blocker fix (shared TIMELINE_KINDS
|
||||
// const + build_thread_replies_filter helper, mirroring the channel sibling so
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -274,6 +274,93 @@ pub fn merge_owner_p_kinds(
|
||||
result
|
||||
}
|
||||
|
||||
/// Atomically remove `kind` from the `owner_p` save subscription for the
|
||||
/// given identity + relay + scope_value.
|
||||
///
|
||||
/// Reads the current `kinds` array, removes `kind` if present, then:
|
||||
/// - if the resulting list is **empty**, deletes the `owner_p` row entirely
|
||||
/// (keeping parity with the UI behavior where the last kind off removes the
|
||||
/// subscription row).
|
||||
/// - otherwise, updates `kinds` to the reduced list.
|
||||
///
|
||||
/// Uses `BEGIN IMMEDIATE` for the same reason as `merge_owner_p_kinds`: the
|
||||
/// write lock is acquired before the SELECT so concurrent callers serialize
|
||||
/// rather than racing to a `BUSY_SNAPSHOT` error.
|
||||
///
|
||||
/// No-op (Ok) if the row does not exist or the kind is not present.
|
||||
pub fn remove_owner_p_kind(
|
||||
conn: &Connection,
|
||||
identity_pubkey: &str,
|
||||
relay_url: &str,
|
||||
scope_value: &str,
|
||||
kind: u32,
|
||||
) -> Result<(), String> {
|
||||
conn.execute_batch("BEGIN IMMEDIATE")
|
||||
.map_err(|e| format!("remove_owner_p_kind begin immediate: {e}"))?;
|
||||
|
||||
let result = (|| -> Result<(), String> {
|
||||
// Read the current kinds, if any.
|
||||
let existing_json: Option<String> = conn
|
||||
.query_row(
|
||||
"SELECT kinds FROM save_subscriptions
|
||||
WHERE identity_pubkey = ?1
|
||||
AND relay_url = ?2
|
||||
AND scope_type = 'owner_p'
|
||||
AND scope_value = ?3",
|
||||
params![identity_pubkey, relay_url, scope_value],
|
||||
|row| row.get::<_, String>(0),
|
||||
)
|
||||
.optional()
|
||||
.map_err(|e| format!("remove_owner_p_kind read: {e}"))?;
|
||||
|
||||
let existing_json = match existing_json {
|
||||
Some(j) => j,
|
||||
// Row doesn't exist — nothing to remove.
|
||||
None => return Ok(()),
|
||||
};
|
||||
|
||||
let mut kinds: Vec<u32> = serde_json::from_str(&existing_json).unwrap_or_default();
|
||||
kinds.retain(|&k| k != kind);
|
||||
|
||||
if kinds.is_empty() {
|
||||
// Last kind removed — delete the row entirely.
|
||||
conn.execute(
|
||||
"DELETE FROM save_subscriptions
|
||||
WHERE identity_pubkey = ?1
|
||||
AND relay_url = ?2
|
||||
AND scope_type = 'owner_p'
|
||||
AND scope_value = ?3",
|
||||
params![identity_pubkey, relay_url, scope_value],
|
||||
)
|
||||
.map_err(|e| format!("remove_owner_p_kind delete: {e}"))?;
|
||||
} else {
|
||||
let kinds_json = serde_json::to_string(&kinds)
|
||||
.map_err(|e| format!("remove_owner_p_kind serialize: {e}"))?;
|
||||
conn.execute(
|
||||
"UPDATE save_subscriptions
|
||||
SET kinds = ?4
|
||||
WHERE identity_pubkey = ?1
|
||||
AND relay_url = ?2
|
||||
AND scope_type = 'owner_p'
|
||||
AND scope_value = ?3",
|
||||
params![identity_pubkey, relay_url, scope_value, kinds_json],
|
||||
)
|
||||
.map_err(|e| format!("remove_owner_p_kind update: {e}"))?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
})();
|
||||
|
||||
if result.is_ok() {
|
||||
conn.execute_batch("COMMIT")
|
||||
.map_err(|e| format!("remove_owner_p_kind commit: {e}"))?;
|
||||
} else {
|
||||
let _ = conn.execute_batch("ROLLBACK");
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
/// Return the `kinds` JSON string for a matching save subscription, or `None`
|
||||
/// if no subscription exists.
|
||||
pub fn get_subscription_kinds(
|
||||
@@ -506,673 +593,5 @@ pub fn gc_orphaned_events(
|
||||
// ── Tests ───────────────────────────────────────────────────────────────────
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn in_memory() -> Connection {
|
||||
let conn = Connection::open_in_memory().unwrap();
|
||||
conn.pragma_update(None, "journal_mode", "WAL").unwrap();
|
||||
conn.pragma_update(None, "busy_timeout", 5000).unwrap();
|
||||
conn.execute_batch(SCHEMA).unwrap();
|
||||
conn
|
||||
}
|
||||
|
||||
// ── Schema init ──────────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_schema_init_creates_all_tables() {
|
||||
let conn = in_memory();
|
||||
// Verify all three tables exist by inserting a row in each.
|
||||
conn.execute(
|
||||
"INSERT INTO save_subscriptions VALUES ('pk','relay','channel_h','abc','[1]',0)",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO archived_events VALUES ('pk','relay','id1',1,'author',0,'{}',0)",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO archived_event_scopes VALUES ('pk','relay','id1','channel_h','abc',0)",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_schema_init_is_idempotent() {
|
||||
// Running SCHEMA twice must not error.
|
||||
let conn = Connection::open_in_memory().unwrap();
|
||||
conn.execute_batch(SCHEMA).unwrap();
|
||||
conn.execute_batch(SCHEMA).unwrap();
|
||||
}
|
||||
|
||||
// ── Save subscriptions ───────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_upsert_save_subscription_inserts_and_updates_kinds() {
|
||||
let conn = in_memory();
|
||||
upsert_save_subscription(&conn, "pk", "wss://r", "channel_h", "abc", "[1]", 1).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1);
|
||||
assert_eq!(subs[0].kinds, "[1]");
|
||||
|
||||
// Update kinds.
|
||||
upsert_save_subscription(&conn, "pk", "wss://r", "channel_h", "abc", "[1,6]", 2).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1);
|
||||
assert_eq!(subs[0].kinds, "[1,6]");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_list_save_subscriptions_scoped_to_identity_and_relay() {
|
||||
let conn = in_memory();
|
||||
upsert_save_subscription(&conn, "pk1", "wss://r1", "channel_h", "a", "[1]", 1).unwrap();
|
||||
upsert_save_subscription(&conn, "pk2", "wss://r1", "channel_h", "b", "[1]", 2).unwrap();
|
||||
upsert_save_subscription(&conn, "pk1", "wss://r2", "channel_h", "c", "[1]", 3).unwrap();
|
||||
|
||||
let subs = list_save_subscriptions(&conn, "pk1", "wss://r1").unwrap();
|
||||
assert_eq!(subs.len(), 1);
|
||||
assert_eq!(subs[0].scope_value, "a");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_delete_save_subscription_removes_row() {
|
||||
let conn = in_memory();
|
||||
upsert_save_subscription(&conn, "pk", "wss://r", "channel_h", "abc", "[1]", 1).unwrap();
|
||||
let deleted = delete_save_subscription(&conn, "pk", "wss://r", "channel_h", "abc").unwrap();
|
||||
assert!(deleted);
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert!(subs.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_delete_save_subscription_returns_false_when_not_found() {
|
||||
let conn = in_memory();
|
||||
let deleted =
|
||||
delete_save_subscription(&conn, "pk", "wss://r", "channel_h", "nope").unwrap();
|
||||
assert!(!deleted);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_has_save_subscription_true_and_false() {
|
||||
let conn = in_memory();
|
||||
upsert_save_subscription(&conn, "pk", "wss://r", "owner_p", "mypk", "[24200]", 1).unwrap();
|
||||
assert!(has_save_subscription(&conn, "pk", "wss://r", "owner_p", "mypk").unwrap());
|
||||
assert!(!has_save_subscription(&conn, "pk", "wss://r", "owner_p", "other").unwrap());
|
||||
}
|
||||
|
||||
// ── merge_owner_p_kinds ──────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_merge_owner_p_kinds_creates_row_when_none_exists() {
|
||||
let conn = in_memory();
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1);
|
||||
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
|
||||
assert_eq!(kinds, [24200]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_merge_owner_p_kinds_adds_new_kind_to_existing_row() {
|
||||
let conn = in_memory();
|
||||
// Seed with 24200 first.
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1).unwrap();
|
||||
// Now merge 44200 in — must produce [24200, 44200].
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1);
|
||||
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
|
||||
assert!(kinds.contains(&24200), "must still contain 24200");
|
||||
assert!(kinds.contains(&44200), "must now contain 44200");
|
||||
assert_eq!(kinds.len(), 2, "no duplicates");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_merge_owner_p_kinds_idempotent_on_existing_kind() {
|
||||
let conn = in_memory();
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 1).unwrap();
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
|
||||
assert_eq!(kinds, [44200], "no duplicates after idempotent call");
|
||||
}
|
||||
|
||||
/// Two-connection WAL regression test for the BEGIN IMMEDIATE fix.
|
||||
///
|
||||
/// Opens TWO separate connections to the same WAL file (mirroring the
|
||||
/// real scenario where the observer-archive seed hook and the
|
||||
/// metric-archive seed hook each open their own connection via
|
||||
/// `open_archive_db`). Both threads call `merge_owner_p_kinds`
|
||||
/// concurrently from an empty row. With `BEGIN IMMEDIATE` the losing
|
||||
/// thread blocks on `busy_timeout` until the winner commits, then
|
||||
/// reads the committed row and merges its kind in. Both calls must
|
||||
/// resolve `Ok` and the final row must contain exactly `[24200, 44200]`.
|
||||
///
|
||||
/// A `DEFERRED` transaction would produce `SQLITE_BUSY_SNAPSHOT` on the
|
||||
/// loser (not retried by busy_timeout), causing one kind to be silently
|
||||
/// dropped. This test fails in <10 ms if the IMMEDIATE guard is removed.
|
||||
#[test]
|
||||
fn test_merge_owner_p_kinds_two_conn_wal_both_kinds_survive() {
|
||||
use std::sync::{Arc, Barrier};
|
||||
use std::thread;
|
||||
use tempfile::NamedTempFile;
|
||||
|
||||
// A real file DB is required for WAL mode (in-memory dbs don't support
|
||||
// shared-cache WAL across multiple connections in the same process).
|
||||
let db_file = NamedTempFile::new().unwrap();
|
||||
let db_path = db_file.path().to_path_buf();
|
||||
|
||||
// Initialise schema on the file DB via conn-A so both threads see it.
|
||||
let init_conn = open_archive_db(&db_path).unwrap();
|
||||
drop(init_conn);
|
||||
|
||||
// Barrier ensures both threads are inside `merge_owner_p_kinds` before
|
||||
// either one issues `BEGIN IMMEDIATE`, maximising the race window.
|
||||
let barrier = Arc::new(Barrier::new(2));
|
||||
|
||||
let path_a = db_path.clone();
|
||||
let path_b = db_path.clone();
|
||||
let bar_a = Arc::clone(&barrier);
|
||||
let bar_b = Arc::clone(&barrier);
|
||||
|
||||
let handle_observer = thread::spawn(move || {
|
||||
let conn = open_archive_db(&path_a).unwrap();
|
||||
bar_a.wait(); // sync: both threads ready
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1)
|
||||
});
|
||||
|
||||
let handle_metric = thread::spawn(move || {
|
||||
let conn = open_archive_db(&path_b).unwrap();
|
||||
bar_b.wait(); // sync: both threads ready
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2)
|
||||
});
|
||||
|
||||
let res_observer = handle_observer.join().expect("observer thread panicked");
|
||||
let res_metric = handle_metric.join().expect("metric thread panicked");
|
||||
|
||||
assert!(
|
||||
res_observer.is_ok(),
|
||||
"observer seed must succeed: {:?}",
|
||||
res_observer
|
||||
);
|
||||
assert!(
|
||||
res_metric.is_ok(),
|
||||
"metric seed must succeed: {:?}",
|
||||
res_metric
|
||||
);
|
||||
|
||||
// Verify the final row contains both kinds.
|
||||
let verify_conn = open_archive_db(&db_path).unwrap();
|
||||
let subs = list_save_subscriptions(&verify_conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1, "exactly one owner_p row");
|
||||
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
|
||||
assert!(
|
||||
kinds.contains(&24200),
|
||||
"observer kind 24200 must survive concurrent metric seed; got {:?}",
|
||||
kinds
|
||||
);
|
||||
assert!(
|
||||
kinds.contains(&44200),
|
||||
"metric kind 44200 must be present after concurrent seed; got {:?}",
|
||||
kinds
|
||||
);
|
||||
assert_eq!(
|
||||
kinds.len(),
|
||||
2,
|
||||
"exactly two kinds, no duplicates; got {:?}",
|
||||
kinds
|
||||
);
|
||||
}
|
||||
|
||||
// ── Archived events ──────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_upsert_archived_event_is_idempotent() {
|
||||
let conn = in_memory();
|
||||
upsert_archived_event(&conn, "pk", "wss://r", "id1", 1, "author", 100, "{}", 200).unwrap();
|
||||
// Second call must not error or duplicate.
|
||||
upsert_archived_event(&conn, "pk", "wss://r", "id1", 1, "author", 100, "{}", 201).unwrap();
|
||||
let count: i64 = conn
|
||||
.query_row("SELECT COUNT(*) FROM archived_events", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(count, 1);
|
||||
}
|
||||
|
||||
// ── Many-to-many scope rows ──────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_one_event_gets_multiple_scope_rows() {
|
||||
let conn = in_memory();
|
||||
upsert_archived_event(&conn, "pk", "wss://r", "id1", 1, "author", 100, "{}", 200).unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "channel_h", "chan1", 200).unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "referenced_e", "evref", 200).unwrap();
|
||||
// Idempotent second insert.
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "channel_h", "chan1", 201).unwrap();
|
||||
|
||||
let count: i64 = conn
|
||||
.query_row(
|
||||
"SELECT COUNT(*) FROM archived_event_scopes WHERE id = 'id1'",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(count, 2);
|
||||
}
|
||||
|
||||
// ── GC ───────────────────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_gc_removes_event_when_last_scope_deleted() {
|
||||
let conn = in_memory();
|
||||
upsert_archived_event(&conn, "pk", "wss://r", "id1", 1, "author", 100, "{}", 200).unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "channel_h", "c1", 200).unwrap();
|
||||
// Delete the only scope row manually.
|
||||
conn.execute("DELETE FROM archived_event_scopes WHERE id = 'id1'", [])
|
||||
.unwrap();
|
||||
let removed = gc_orphaned_events(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(removed, 1);
|
||||
let count: i64 = conn
|
||||
.query_row("SELECT COUNT(*) FROM archived_events", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(count, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_gc_leaves_event_with_remaining_scope() {
|
||||
let conn = in_memory();
|
||||
upsert_archived_event(&conn, "pk", "wss://r", "id1", 1, "author", 100, "{}", 200).unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "channel_h", "c1", 200).unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "referenced_e", "ref", 200).unwrap();
|
||||
// Delete only one scope row.
|
||||
conn.execute(
|
||||
"DELETE FROM archived_event_scopes WHERE scope_type = 'referenced_e'",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
let removed = gc_orphaned_events(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(removed, 0);
|
||||
}
|
||||
|
||||
// ── read_archived_events ─────────────────────────────────────────────────
|
||||
|
||||
fn seed_events(conn: &Connection) {
|
||||
// Three events in scope "channel_h/chan1" for identity "pk"/"wss://r".
|
||||
// created_at: 300 (newest), 200, 100 (oldest).
|
||||
for (id, kind, created_at, raw) in &[
|
||||
("e1", 9i64, 300i64, r#"{"id":"e1","created_at":300}"#),
|
||||
("e2", 9i64, 200i64, r#"{"id":"e2","created_at":200}"#),
|
||||
("e3", 42i64, 100i64, r#"{"id":"e3","created_at":100}"#),
|
||||
] {
|
||||
upsert_archived_event(
|
||||
conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
id,
|
||||
*kind,
|
||||
"author",
|
||||
*created_at,
|
||||
raw,
|
||||
999,
|
||||
)
|
||||
.unwrap();
|
||||
upsert_event_scope(conn, "pk", "wss://r", id, "channel_h", "chan1", 999).unwrap();
|
||||
}
|
||||
// One event in a different scope — must never appear in chan1 results.
|
||||
upsert_archived_event(
|
||||
conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"e4",
|
||||
9,
|
||||
"author",
|
||||
250,
|
||||
r#"{"id":"e4"}"#,
|
||||
999,
|
||||
)
|
||||
.unwrap();
|
||||
upsert_event_scope(conn, "pk", "wss://r", "e4", "channel_h", "chan2", 999).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_returns_newest_first() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(rows.len(), 3);
|
||||
// Newest first: e1 (300), e2 (200), e3 (100).
|
||||
let ids: Vec<&str> = rows
|
||||
.iter()
|
||||
.map(|r| {
|
||||
if r.contains("\"e1\"") {
|
||||
"e1"
|
||||
} else if r.contains("\"e2\"") {
|
||||
"e2"
|
||||
} else {
|
||||
"e3"
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(ids, ["e1", "e2", "e3"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_keyset_cursor_excludes_at_boundary() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// Compound cursor at e1 (created_at=300, id="e1"): excludes e1 itself,
|
||||
// returns e2 and e3.
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
Some(300),
|
||||
Some("e1"),
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(rows.len(), 2);
|
||||
assert!(rows.iter().all(|r| !r.contains("\"e1\"")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_keyset_cursor_advances_correctly() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// Page 1: before=None/None, limit=2 → e1, e2.
|
||||
let page1 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
2,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(page1.len(), 2);
|
||||
// Page 2: compound cursor at e2 (created_at=200, id="e2") → e3 only.
|
||||
let page2 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
Some(200),
|
||||
Some("e2"),
|
||||
2,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(page2.len(), 1);
|
||||
assert!(page2[0].contains("\"e3\""));
|
||||
// No overlap between pages.
|
||||
assert!(page1.iter().all(|r| !r.contains("\"e3\"")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_kind_filter() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// Only kind 9 (e1 and e2); e3 is kind 42.
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
Some(&[9]),
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(rows.len(), 2);
|
||||
assert!(rows.iter().all(|r| !r.contains("\"e3\"")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_scope_isolation() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// chan2 has only e4; chan1 results must not include e4.
|
||||
let chan1 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert!(chan1.iter().all(|r| !r.contains("\"e4\"")));
|
||||
|
||||
let chan2 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan2",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(chan2.len(), 1);
|
||||
assert!(chan2[0].contains("\"e4\""));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_identity_isolation() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// Different identity — must see no rows.
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk2",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert!(rows.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_relay_isolation() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// Different relay — must see no rows.
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://other",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert!(rows.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_empty_result() {
|
||||
let conn = in_memory();
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"nope",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert!(rows.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_limit_respected() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
1,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(rows.len(), 1);
|
||||
// Must be the newest (e1, created_at=300).
|
||||
assert!(rows[0].contains("\"e1\""));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_no_duplicates_across_pages() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
let page1 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
2,
|
||||
)
|
||||
.unwrap();
|
||||
// Compound cursor at e2 (the oldest in page1: created_at=200, id="e2").
|
||||
let page2 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
Some(200),
|
||||
Some("e2"),
|
||||
2,
|
||||
)
|
||||
.unwrap();
|
||||
// All event ids across both pages are unique.
|
||||
let all: Vec<_> = page1.iter().chain(page2.iter()).collect();
|
||||
assert_eq!(all.len(), 3); // 2 + 1 = 3 total, no duplication.
|
||||
}
|
||||
|
||||
/// Regression for the scalar-cursor same-second skip defect (Thufir IMPORTANT).
|
||||
///
|
||||
/// The writer stores `created_at` in whole seconds, so two events can share
|
||||
/// the same timestamp. The sort order is `(created_at DESC, id DESC)`, so
|
||||
/// a page split exactly at a same-second boundary leaves one sibling on each
|
||||
/// side. With only `created_at < before` the second-page sibling would be
|
||||
/// permanently excluded. The compound `(created_at < ?) OR (created_at = ?
|
||||
/// AND id < ?)` predicate mirrors the sort key exactly and avoids the skip.
|
||||
#[test]
|
||||
fn test_read_archived_events_same_second_cursor_no_skip() {
|
||||
let conn = in_memory();
|
||||
// Two events share created_at=1000. Sort order: "z" (id "z") > "a" (id "a"),
|
||||
// so ORDER BY created_at DESC, id DESC yields: ("z", 1000) first, ("a", 1000) second.
|
||||
// A third event has created_at=500.
|
||||
for (id, kind, created_at, raw) in &[
|
||||
("z", 9i64, 1000i64, r#"{"id":"z","created_at":1000}"#),
|
||||
("a", 9i64, 1000i64, r#"{"id":"a","created_at":1000}"#),
|
||||
("old", 9i64, 500i64, r#"{"id":"old","created_at":500}"#),
|
||||
] {
|
||||
upsert_archived_event(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
id,
|
||||
*kind,
|
||||
"author",
|
||||
*created_at,
|
||||
raw,
|
||||
999,
|
||||
)
|
||||
.unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", id, "channel_h", "same_sec", 999).unwrap();
|
||||
}
|
||||
|
||||
// Page 1: limit=1 → should return ("z", 1000) only (newest by compound sort).
|
||||
let page1 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"same_sec",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
1,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(page1.len(), 1);
|
||||
assert!(page1[0].contains("\"z\""), "page1 must be the 'z' row");
|
||||
|
||||
// Page 2: compound cursor at ("z", 1000).
|
||||
// With a scalar cursor (created_at < 1000), row "a" would be SKIPPED.
|
||||
// With the compound cursor, "a" must appear on page 2.
|
||||
let page2 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"same_sec",
|
||||
None,
|
||||
Some(1000),
|
||||
Some("z"),
|
||||
2,
|
||||
)
|
||||
.unwrap();
|
||||
// Must contain "a" (same-second sibling) and "old" (strictly older).
|
||||
assert_eq!(page2.len(), 2, "page2 must return both remaining rows");
|
||||
assert!(
|
||||
page2.iter().any(|r| r.contains("\"a\"")),
|
||||
"same-second sibling 'a' must not be skipped"
|
||||
);
|
||||
assert!(
|
||||
page2.iter().any(|r| r.contains("\"old\"")),
|
||||
"'old' row must appear on page2"
|
||||
);
|
||||
// No overlap with page1.
|
||||
assert!(page2.iter().all(|r| !r.contains("\"z\"")));
|
||||
}
|
||||
}
|
||||
#[path = "store_tests.rs"]
|
||||
mod store_tests;
|
||||
|
||||
@@ -0,0 +1,730 @@
|
||||
//! Unit tests for `archive/store.rs`.
|
||||
//!
|
||||
//! Kept in a sibling file so `store.rs` stays under the 1000-line gate;
|
||||
//! `#[path]`-included from there.
|
||||
|
||||
use super::*;
|
||||
|
||||
fn in_memory() -> Connection {
|
||||
let conn = Connection::open_in_memory().unwrap();
|
||||
conn.pragma_update(None, "journal_mode", "WAL").unwrap();
|
||||
conn.pragma_update(None, "busy_timeout", 5000).unwrap();
|
||||
conn.execute_batch(SCHEMA).unwrap();
|
||||
conn
|
||||
}
|
||||
|
||||
// ── Schema init ──────────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_schema_init_creates_all_tables() {
|
||||
let conn = in_memory();
|
||||
// Verify all three tables exist by inserting a row in each.
|
||||
conn.execute(
|
||||
"INSERT INTO save_subscriptions VALUES ('pk','relay','channel_h','abc','[1]',0)",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO archived_events VALUES ('pk','relay','id1',1,'author',0,'{}',0)",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO archived_event_scopes VALUES ('pk','relay','id1','channel_h','abc',0)",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_schema_init_is_idempotent() {
|
||||
// Running SCHEMA twice must not error.
|
||||
let conn = Connection::open_in_memory().unwrap();
|
||||
conn.execute_batch(SCHEMA).unwrap();
|
||||
conn.execute_batch(SCHEMA).unwrap();
|
||||
}
|
||||
|
||||
// ── Save subscriptions ───────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_upsert_save_subscription_inserts_and_updates_kinds() {
|
||||
let conn = in_memory();
|
||||
upsert_save_subscription(&conn, "pk", "wss://r", "channel_h", "abc", "[1]", 1).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1);
|
||||
assert_eq!(subs[0].kinds, "[1]");
|
||||
|
||||
// Update kinds.
|
||||
upsert_save_subscription(&conn, "pk", "wss://r", "channel_h", "abc", "[1,6]", 2).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1);
|
||||
assert_eq!(subs[0].kinds, "[1,6]");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_list_save_subscriptions_scoped_to_identity_and_relay() {
|
||||
let conn = in_memory();
|
||||
upsert_save_subscription(&conn, "pk1", "wss://r1", "channel_h", "a", "[1]", 1).unwrap();
|
||||
upsert_save_subscription(&conn, "pk2", "wss://r1", "channel_h", "b", "[1]", 2).unwrap();
|
||||
upsert_save_subscription(&conn, "pk1", "wss://r2", "channel_h", "c", "[1]", 3).unwrap();
|
||||
|
||||
let subs = list_save_subscriptions(&conn, "pk1", "wss://r1").unwrap();
|
||||
assert_eq!(subs.len(), 1);
|
||||
assert_eq!(subs[0].scope_value, "a");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_delete_save_subscription_removes_row() {
|
||||
let conn = in_memory();
|
||||
upsert_save_subscription(&conn, "pk", "wss://r", "channel_h", "abc", "[1]", 1).unwrap();
|
||||
let deleted = delete_save_subscription(&conn, "pk", "wss://r", "channel_h", "abc").unwrap();
|
||||
assert!(deleted);
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert!(subs.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_delete_save_subscription_returns_false_when_not_found() {
|
||||
let conn = in_memory();
|
||||
let deleted = delete_save_subscription(&conn, "pk", "wss://r", "channel_h", "nope").unwrap();
|
||||
assert!(!deleted);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_has_save_subscription_true_and_false() {
|
||||
let conn = in_memory();
|
||||
upsert_save_subscription(&conn, "pk", "wss://r", "owner_p", "mypk", "[24200]", 1).unwrap();
|
||||
assert!(has_save_subscription(&conn, "pk", "wss://r", "owner_p", "mypk").unwrap());
|
||||
assert!(!has_save_subscription(&conn, "pk", "wss://r", "owner_p", "other").unwrap());
|
||||
}
|
||||
|
||||
// ── merge_owner_p_kinds ──────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_merge_owner_p_kinds_creates_row_when_none_exists() {
|
||||
let conn = in_memory();
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1);
|
||||
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
|
||||
assert_eq!(kinds, [24200]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_merge_owner_p_kinds_adds_new_kind_to_existing_row() {
|
||||
let conn = in_memory();
|
||||
// Seed with 24200 first.
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1).unwrap();
|
||||
// Now merge 44200 in — must produce [24200, 44200].
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1);
|
||||
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
|
||||
assert!(kinds.contains(&24200), "must still contain 24200");
|
||||
assert!(kinds.contains(&44200), "must now contain 44200");
|
||||
assert_eq!(kinds.len(), 2, "no duplicates");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_merge_owner_p_kinds_idempotent_on_existing_kind() {
|
||||
let conn = in_memory();
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 1).unwrap();
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
|
||||
assert_eq!(kinds, [44200], "no duplicates after idempotent call");
|
||||
}
|
||||
|
||||
/// Two-connection WAL regression test for the BEGIN IMMEDIATE fix.
|
||||
///
|
||||
/// Opens TWO separate connections to the same WAL file (mirroring the
|
||||
/// real scenario where the observer-archive seed hook and the
|
||||
/// metric-archive seed hook each open their own connection via
|
||||
/// `open_archive_db`). Both threads call `merge_owner_p_kinds`
|
||||
/// concurrently from an empty row. With `BEGIN IMMEDIATE` the losing
|
||||
/// thread blocks on `busy_timeout` until the winner commits, then
|
||||
/// reads the committed row and merges its kind in. Both calls must
|
||||
/// resolve `Ok` and the final row must contain exactly `[24200, 44200]`.
|
||||
///
|
||||
/// A `DEFERRED` transaction would produce `SQLITE_BUSY_SNAPSHOT` on the
|
||||
/// loser (not retried by busy_timeout), causing one kind to be silently
|
||||
/// dropped. This test fails in <10 ms if the IMMEDIATE guard is removed.
|
||||
#[test]
|
||||
fn test_merge_owner_p_kinds_two_conn_wal_both_kinds_survive() {
|
||||
use std::sync::{Arc, Barrier};
|
||||
use std::thread;
|
||||
use tempfile::NamedTempFile;
|
||||
|
||||
// A real file DB is required for WAL mode (in-memory dbs don't support
|
||||
// shared-cache WAL across multiple connections in the same process).
|
||||
let db_file = NamedTempFile::new().unwrap();
|
||||
let db_path = db_file.path().to_path_buf();
|
||||
|
||||
// Initialise schema on the file DB via conn-A so both threads see it.
|
||||
let init_conn = open_archive_db(&db_path).unwrap();
|
||||
drop(init_conn);
|
||||
|
||||
// Barrier ensures both threads are inside `merge_owner_p_kinds` before
|
||||
// either one issues `BEGIN IMMEDIATE`, maximising the race window.
|
||||
let barrier = Arc::new(Barrier::new(2));
|
||||
|
||||
let path_a = db_path.clone();
|
||||
let path_b = db_path.clone();
|
||||
let bar_a = Arc::clone(&barrier);
|
||||
let bar_b = Arc::clone(&barrier);
|
||||
|
||||
let handle_observer = thread::spawn(move || {
|
||||
let conn = open_archive_db(&path_a).unwrap();
|
||||
bar_a.wait(); // sync: both threads ready
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1)
|
||||
});
|
||||
|
||||
let handle_metric = thread::spawn(move || {
|
||||
let conn = open_archive_db(&path_b).unwrap();
|
||||
bar_b.wait(); // sync: both threads ready
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2)
|
||||
});
|
||||
|
||||
let res_observer = handle_observer.join().expect("observer thread panicked");
|
||||
let res_metric = handle_metric.join().expect("metric thread panicked");
|
||||
|
||||
assert!(
|
||||
res_observer.is_ok(),
|
||||
"observer seed must succeed: {:?}",
|
||||
res_observer
|
||||
);
|
||||
assert!(
|
||||
res_metric.is_ok(),
|
||||
"metric seed must succeed: {:?}",
|
||||
res_metric
|
||||
);
|
||||
|
||||
// Verify the final row contains both kinds.
|
||||
let verify_conn = open_archive_db(&db_path).unwrap();
|
||||
let subs = list_save_subscriptions(&verify_conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1, "exactly one owner_p row");
|
||||
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
|
||||
assert!(
|
||||
kinds.contains(&24200),
|
||||
"observer kind 24200 must survive concurrent metric seed; got {:?}",
|
||||
kinds
|
||||
);
|
||||
assert!(
|
||||
kinds.contains(&44200),
|
||||
"metric kind 44200 must be present after concurrent seed; got {:?}",
|
||||
kinds
|
||||
);
|
||||
assert_eq!(
|
||||
kinds.len(),
|
||||
2,
|
||||
"exactly two kinds, no duplicates; got {:?}",
|
||||
kinds
|
||||
);
|
||||
}
|
||||
|
||||
// ── remove_owner_p_kind ──────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_remove_owner_p_kind_removes_one_kind_leaving_other() {
|
||||
let conn = in_memory();
|
||||
// Seed both kinds.
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1).unwrap();
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 2).unwrap();
|
||||
|
||||
// Remove 44200 — 24200 must survive.
|
||||
remove_owner_p_kind(&conn, "pk", "wss://r", "mypk", 44200).unwrap();
|
||||
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1, "row must still exist");
|
||||
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
|
||||
assert_eq!(
|
||||
kinds,
|
||||
[24200],
|
||||
"only 24200 remains after removing 44200; got {kinds:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_remove_owner_p_kind_deletes_row_when_last_kind_removed() {
|
||||
let conn = in_memory();
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 44200, 1).unwrap();
|
||||
|
||||
// Remove the only kind — row must be deleted.
|
||||
remove_owner_p_kind(&conn, "pk", "wss://r", "mypk", 44200).unwrap();
|
||||
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert!(
|
||||
subs.is_empty(),
|
||||
"row must be deleted when last kind removed"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_remove_owner_p_kind_noop_when_row_absent() {
|
||||
let conn = in_memory();
|
||||
// No row exists — must succeed silently.
|
||||
remove_owner_p_kind(&conn, "pk", "wss://r", "mypk", 24200).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert!(subs.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_remove_owner_p_kind_noop_when_kind_absent() {
|
||||
let conn = in_memory();
|
||||
// Row exists with only 24200 — removing 44200 must leave row unchanged.
|
||||
merge_owner_p_kinds(&conn, "pk", "wss://r", "mypk", 24200, 1).unwrap();
|
||||
remove_owner_p_kind(&conn, "pk", "wss://r", "mypk", 44200).unwrap();
|
||||
let subs = list_save_subscriptions(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(subs.len(), 1);
|
||||
let kinds: Vec<u32> = serde_json::from_str(&subs[0].kinds).unwrap();
|
||||
assert_eq!(kinds, [24200]);
|
||||
}
|
||||
|
||||
// ── Archived events ──────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_upsert_archived_event_is_idempotent() {
|
||||
let conn = in_memory();
|
||||
upsert_archived_event(&conn, "pk", "wss://r", "id1", 1, "author", 100, "{}", 200).unwrap();
|
||||
// Second call must not error or duplicate.
|
||||
upsert_archived_event(&conn, "pk", "wss://r", "id1", 1, "author", 100, "{}", 201).unwrap();
|
||||
let count: i64 = conn
|
||||
.query_row("SELECT COUNT(*) FROM archived_events", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(count, 1);
|
||||
}
|
||||
|
||||
// ── Many-to-many scope rows ──────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_one_event_gets_multiple_scope_rows() {
|
||||
let conn = in_memory();
|
||||
upsert_archived_event(&conn, "pk", "wss://r", "id1", 1, "author", 100, "{}", 200).unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "channel_h", "chan1", 200).unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "referenced_e", "evref", 200).unwrap();
|
||||
// Idempotent second insert.
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "channel_h", "chan1", 201).unwrap();
|
||||
|
||||
let count: i64 = conn
|
||||
.query_row(
|
||||
"SELECT COUNT(*) FROM archived_event_scopes WHERE id = 'id1'",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(count, 2);
|
||||
}
|
||||
|
||||
// ── GC ───────────────────────────────────────────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_gc_removes_event_when_last_scope_deleted() {
|
||||
let conn = in_memory();
|
||||
upsert_archived_event(&conn, "pk", "wss://r", "id1", 1, "author", 100, "{}", 200).unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "channel_h", "c1", 200).unwrap();
|
||||
// Delete the only scope row manually.
|
||||
conn.execute("DELETE FROM archived_event_scopes WHERE id = 'id1'", [])
|
||||
.unwrap();
|
||||
let removed = gc_orphaned_events(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(removed, 1);
|
||||
let count: i64 = conn
|
||||
.query_row("SELECT COUNT(*) FROM archived_events", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(count, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_gc_leaves_event_with_remaining_scope() {
|
||||
let conn = in_memory();
|
||||
upsert_archived_event(&conn, "pk", "wss://r", "id1", 1, "author", 100, "{}", 200).unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "channel_h", "c1", 200).unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", "id1", "referenced_e", "ref", 200).unwrap();
|
||||
// Delete only one scope row.
|
||||
conn.execute(
|
||||
"DELETE FROM archived_event_scopes WHERE scope_type = 'referenced_e'",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
let removed = gc_orphaned_events(&conn, "pk", "wss://r").unwrap();
|
||||
assert_eq!(removed, 0);
|
||||
}
|
||||
|
||||
// ── read_archived_events ─────────────────────────────────────────────────
|
||||
|
||||
fn seed_events(conn: &Connection) {
|
||||
// Three events in scope "channel_h/chan1" for identity "pk"/"wss://r".
|
||||
// created_at: 300 (newest), 200, 100 (oldest).
|
||||
for (id, kind, created_at, raw) in &[
|
||||
("e1", 9i64, 300i64, r#"{"id":"e1","created_at":300}"#),
|
||||
("e2", 9i64, 200i64, r#"{"id":"e2","created_at":200}"#),
|
||||
("e3", 42i64, 100i64, r#"{"id":"e3","created_at":100}"#),
|
||||
] {
|
||||
upsert_archived_event(
|
||||
conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
id,
|
||||
*kind,
|
||||
"author",
|
||||
*created_at,
|
||||
raw,
|
||||
999,
|
||||
)
|
||||
.unwrap();
|
||||
upsert_event_scope(conn, "pk", "wss://r", id, "channel_h", "chan1", 999).unwrap();
|
||||
}
|
||||
// One event in a different scope — must never appear in chan1 results.
|
||||
upsert_archived_event(
|
||||
conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"e4",
|
||||
9,
|
||||
"author",
|
||||
250,
|
||||
r#"{"id":"e4"}"#,
|
||||
999,
|
||||
)
|
||||
.unwrap();
|
||||
upsert_event_scope(conn, "pk", "wss://r", "e4", "channel_h", "chan2", 999).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_returns_newest_first() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(rows.len(), 3);
|
||||
// Newest first: e1 (300), e2 (200), e3 (100).
|
||||
let ids: Vec<&str> = rows
|
||||
.iter()
|
||||
.map(|r| {
|
||||
if r.contains("\"e1\"") {
|
||||
"e1"
|
||||
} else if r.contains("\"e2\"") {
|
||||
"e2"
|
||||
} else {
|
||||
"e3"
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(ids, ["e1", "e2", "e3"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_keyset_cursor_excludes_at_boundary() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// Compound cursor at e1 (created_at=300, id="e1"): excludes e1 itself,
|
||||
// returns e2 and e3.
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
Some(300),
|
||||
Some("e1"),
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(rows.len(), 2);
|
||||
assert!(rows.iter().all(|r| !r.contains("\"e1\"")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_keyset_cursor_advances_correctly() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// Page 1: before=None/None, limit=2 → e1, e2.
|
||||
let page1 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
2,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(page1.len(), 2);
|
||||
// Page 2: compound cursor at e2 (created_at=200, id="e2") → e3 only.
|
||||
let page2 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
Some(200),
|
||||
Some("e2"),
|
||||
2,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(page2.len(), 1);
|
||||
assert!(page2[0].contains("\"e3\""));
|
||||
// No overlap between pages.
|
||||
assert!(page1.iter().all(|r| !r.contains("\"e3\"")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_kind_filter() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// Only kind 9 (e1 and e2); e3 is kind 42.
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
Some(&[9]),
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(rows.len(), 2);
|
||||
assert!(rows.iter().all(|r| !r.contains("\"e3\"")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_scope_isolation() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// chan2 has only e4; chan1 results must not include e4.
|
||||
let chan1 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert!(chan1.iter().all(|r| !r.contains("\"e4\"")));
|
||||
|
||||
let chan2 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan2",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(chan2.len(), 1);
|
||||
assert!(chan2[0].contains("\"e4\""));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_identity_isolation() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// Different identity — must see no rows.
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk2",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert!(rows.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_relay_isolation() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
// Different relay — must see no rows.
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://other",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert!(rows.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_empty_result() {
|
||||
let conn = in_memory();
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"nope",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
10,
|
||||
)
|
||||
.unwrap();
|
||||
assert!(rows.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_limit_respected() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
let rows = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
1,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(rows.len(), 1);
|
||||
// Must be the newest (e1, created_at=300).
|
||||
assert!(rows[0].contains("\"e1\""));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_read_archived_events_no_duplicates_across_pages() {
|
||||
let conn = in_memory();
|
||||
seed_events(&conn);
|
||||
let page1 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
2,
|
||||
)
|
||||
.unwrap();
|
||||
// Compound cursor at e2 (the oldest in page1: created_at=200, id="e2").
|
||||
let page2 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"chan1",
|
||||
None,
|
||||
Some(200),
|
||||
Some("e2"),
|
||||
2,
|
||||
)
|
||||
.unwrap();
|
||||
// All event ids across both pages are unique.
|
||||
let all: Vec<_> = page1.iter().chain(page2.iter()).collect();
|
||||
assert_eq!(all.len(), 3); // 2 + 1 = 3 total, no duplication.
|
||||
}
|
||||
|
||||
/// Regression for the scalar-cursor same-second skip defect (Thufir IMPORTANT).
|
||||
///
|
||||
/// The writer stores `created_at` in whole seconds, so two events can share
|
||||
/// the same timestamp. The sort order is `(created_at DESC, id DESC)`, so
|
||||
/// a page split exactly at a same-second boundary leaves one sibling on each
|
||||
/// side. With only `created_at < before` the second-page sibling would be
|
||||
/// permanently excluded. The compound `(created_at < ?) OR (created_at = ?
|
||||
/// AND id < ?)` predicate mirrors the sort key exactly and avoids the skip.
|
||||
#[test]
|
||||
fn test_read_archived_events_same_second_cursor_no_skip() {
|
||||
let conn = in_memory();
|
||||
// Two events share created_at=1000. Sort order: "z" (id "z") > "a" (id "a"),
|
||||
// so ORDER BY created_at DESC, id DESC yields: ("z", 1000) first, ("a", 1000) second.
|
||||
// A third event has created_at=500.
|
||||
for (id, kind, created_at, raw) in &[
|
||||
("z", 9i64, 1000i64, r#"{"id":"z","created_at":1000}"#),
|
||||
("a", 9i64, 1000i64, r#"{"id":"a","created_at":1000}"#),
|
||||
("old", 9i64, 500i64, r#"{"id":"old","created_at":500}"#),
|
||||
] {
|
||||
upsert_archived_event(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
id,
|
||||
*kind,
|
||||
"author",
|
||||
*created_at,
|
||||
raw,
|
||||
999,
|
||||
)
|
||||
.unwrap();
|
||||
upsert_event_scope(&conn, "pk", "wss://r", id, "channel_h", "same_sec", 999).unwrap();
|
||||
}
|
||||
|
||||
// Page 1: limit=1 → should return ("z", 1000) only (newest by compound sort).
|
||||
let page1 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"same_sec",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
1,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(page1.len(), 1);
|
||||
assert!(page1[0].contains("\"z\""), "page1 must be the 'z' row");
|
||||
|
||||
// Page 2: compound cursor at ("z", 1000).
|
||||
// With a scalar cursor (created_at < 1000), row "a" would be SKIPPED.
|
||||
// With the compound cursor, "a" must appear on page 2.
|
||||
let page2 = read_archived_events(
|
||||
&conn,
|
||||
"pk",
|
||||
"wss://r",
|
||||
"channel_h",
|
||||
"same_sec",
|
||||
None,
|
||||
Some(1000),
|
||||
Some("z"),
|
||||
2,
|
||||
)
|
||||
.unwrap();
|
||||
// Must contain "a" (same-second sibling) and "old" (strictly older).
|
||||
assert_eq!(page2.len(), 2, "page2 must return both remaining rows");
|
||||
assert!(
|
||||
page2.iter().any(|r| r.contains("\"a\"")),
|
||||
"same-second sibling 'a' must not be skipped"
|
||||
);
|
||||
assert!(
|
||||
page2.iter().any(|r| r.contains("\"old\"")),
|
||||
"'old' row must appear on page2"
|
||||
);
|
||||
// No overlap with page1.
|
||||
assert!(page2.iter().all(|r| !r.contains("\"z\"")));
|
||||
}
|
||||
@@ -619,6 +619,7 @@ pub fn run() {
|
||||
archive::archive_events,
|
||||
archive::create_save_subscription,
|
||||
archive::merge_save_subscription_kinds,
|
||||
archive::remove_save_subscription_kind,
|
||||
archive::list_save_subscriptions,
|
||||
archive::delete_save_subscription,
|
||||
archive::read_archived_events,
|
||||
|
||||
@@ -6,6 +6,8 @@ import {
|
||||
createSaveSubscription,
|
||||
deleteSaveSubscription,
|
||||
listSaveSubscriptions,
|
||||
mergeSaveSubscriptionKinds,
|
||||
removeSaveSubscriptionKind,
|
||||
type SaveSubscription,
|
||||
type ScopeType,
|
||||
} from "@/shared/api/tauriArchive";
|
||||
@@ -447,21 +449,10 @@ export function LocalArchiveSettingsCard() {
|
||||
if (!pubkey) return;
|
||||
setObserverToggling(true);
|
||||
try {
|
||||
// The owner_p row is keyed by (scope_type, scope_value) — both observer
|
||||
// (24200) and metric (44200) share the same row. Merge kinds atomically:
|
||||
// read current kinds, add or remove 24200, upsert the result.
|
||||
const currentKinds =
|
||||
subs
|
||||
.find((s) => s.scopeType === "owner_p" && s.scopeValue === pubkey)
|
||||
?.kinds.filter((k) => k !== KIND_AGENT_OBSERVER_FRAME) ?? [];
|
||||
const nextKinds = checked
|
||||
? [...currentKinds, KIND_AGENT_OBSERVER_FRAME]
|
||||
: currentKinds;
|
||||
|
||||
if (nextKinds.length > 0) {
|
||||
await createSaveSubscription("owner_p", pubkey, nextKinds);
|
||||
if (checked) {
|
||||
await mergeSaveSubscriptionKinds(KIND_AGENT_OBSERVER_FRAME);
|
||||
} else {
|
||||
await deleteSaveSubscription("owner_p", pubkey);
|
||||
await removeSaveSubscriptionKind(KIND_AGENT_OBSERVER_FRAME);
|
||||
}
|
||||
setExplicitObserverArchiveChoice(pubkey, checked);
|
||||
toast.success(
|
||||
@@ -480,7 +471,7 @@ export function LocalArchiveSettingsCard() {
|
||||
setObserverToggling(false);
|
||||
}
|
||||
},
|
||||
[pubkey, subs, reload],
|
||||
[pubkey, reload],
|
||||
);
|
||||
|
||||
const handleMetricToggle = React.useCallback(
|
||||
@@ -488,19 +479,10 @@ export function LocalArchiveSettingsCard() {
|
||||
if (!pubkey) return;
|
||||
setMetricToggling(true);
|
||||
try {
|
||||
// Same row as observer — merge 44200 in or out.
|
||||
const currentKinds =
|
||||
subs
|
||||
.find((s) => s.scopeType === "owner_p" && s.scopeValue === pubkey)
|
||||
?.kinds.filter((k) => k !== KIND_AGENT_TURN_METRIC) ?? [];
|
||||
const nextKinds = checked
|
||||
? [...currentKinds, KIND_AGENT_TURN_METRIC]
|
||||
: currentKinds;
|
||||
|
||||
if (nextKinds.length > 0) {
|
||||
await createSaveSubscription("owner_p", pubkey, nextKinds);
|
||||
if (checked) {
|
||||
await mergeSaveSubscriptionKinds(KIND_AGENT_TURN_METRIC);
|
||||
} else {
|
||||
await deleteSaveSubscription("owner_p", pubkey);
|
||||
await removeSaveSubscriptionKind(KIND_AGENT_TURN_METRIC);
|
||||
}
|
||||
setExplicitAgentMetricArchiveChoice(pubkey, checked);
|
||||
toast.success(
|
||||
@@ -519,7 +501,7 @@ export function LocalArchiveSettingsCard() {
|
||||
setMetricToggling(false);
|
||||
}
|
||||
},
|
||||
[pubkey, subs, reload],
|
||||
[pubkey, reload],
|
||||
);
|
||||
|
||||
// Non-owner_p subscriptions shown in the active-subscriptions list.
|
||||
|
||||
@@ -126,6 +126,24 @@ export async function mergeSaveSubscriptionKinds(kind: number): Promise<void> {
|
||||
notifySubscriptionChange();
|
||||
}
|
||||
|
||||
/**
|
||||
* Atomically remove `kind` from the `owner_p` save subscription for the
|
||||
* current identity + relay.
|
||||
*
|
||||
* Mirrors `mergeSaveSubscriptionKinds`: reads existing kinds, removes `kind`,
|
||||
* then deletes the row if the list becomes empty or updates it otherwise.
|
||||
* Uses `BEGIN IMMEDIATE` on the Rust side for the same reason as the merge
|
||||
* path — concurrent toggle-OFF callers serialize rather than racing.
|
||||
*
|
||||
* Called by toggle-OFF handlers in `LocalArchiveSettingsCard` for both
|
||||
* kind 24200 and kind 44200, replacing the former TS-side read-modify-overwrite
|
||||
* that would drop the *other* kind if `subs` state was stale.
|
||||
*/
|
||||
export async function removeSaveSubscriptionKind(kind: number): Promise<void> {
|
||||
await invokeTauri("remove_save_subscription_kind", { kind });
|
||||
notifySubscriptionChange();
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a save subscription.
|
||||
* Runs an access probe on the backend (channel membership, event readability).
|
||||
|
||||
Reference in New Issue
Block a user