From 42a0a786b713f0d32743409fb8e7fb7ab105e728 Mon Sep 17 00:00:00 2001
From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
Date: Wed, 1 Jul 2026 16:30:58 -0400
Subject: [PATCH 01/19] feat(archive): add local-save archive backend
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
SQLite store in the Buzz nest (~/.buzz/archive/archive.db) with
three tables:
- archived_events — one raw event row per (identity, relay, id)
- archived_event_scopes — many-to-many scope membership rows
- save_subscriptions — which scopes the user has subscribed to save
Four Tauri commands:
- archive_events — batch-first; two proof paths (persistent vs ephemeral)
- create_save_subscription — per-scope access probe before persisting
- list_save_subscriptions — scoped to current identity + relay
- delete_save_subscription — subscription row only; retention decoupled (v1)
Persistent scopes (channel_h, referenced_e) use a batched authed /query
to the relay as the access proof. Ephemeral scope (owner_p, kind 24200)
uses local validation: valid sig/id, kind 24200, #p == identity, agent
tag present, frame == telemetry, author == agent tag. Both paths are
fail-closed.
relay_url is canonicalized to WS-form via relay_ws_url_with_override
everywhere so the same relay never splits rows across wss:// and https://
variants.
Co-authored-by: Will Pfleger
Signed-off-by: Will Pfleger
---
desktop/src-tauri/src/archive/mod.rs | 861 +++++++++++++++++++++++++
desktop/src-tauri/src/archive/store.rs | 457 +++++++++++++
desktop/src-tauri/src/lib.rs | 5 +
3 files changed, 1323 insertions(+)
create mode 100644 desktop/src-tauri/src/archive/mod.rs
create mode 100644 desktop/src-tauri/src/archive/store.rs
diff --git a/desktop/src-tauri/src/archive/mod.rs b/desktop/src-tauri/src/archive/mod.rs
new file mode 100644
index 000000000..ab56c5469
--- /dev/null
+++ b/desktop/src-tauri/src/archive/mod.rs
@@ -0,0 +1,861 @@
+//! Local-save archive — Tauri commands for archiving relay messages to a
+//! per-identity SQLite database in the Buzz nest.
+//!
+//! # Architecture
+//!
+//! Two access proof paths, chosen by event kind:
+//!
+//! **Persistent scopes** (`channel_h`, `referenced_e`): the relay is the
+//! source of truth. Candidates are grouped and re-queried via a batched
+//! authed `/query`; only events the relay returns are inserted.
+//!
+//! **Ephemeral scope** (`owner_p`, kind 24200 observer frames): the relay
+//! never stores these, so `/query` cannot verify them. The relay's REQ-time
+//! `#p == authed reader` gate is the access control; local per-frame
+//! validation (sig/id + kind + p-tag + agent tag + frame=telemetry + author
+//! == agent) is applied fail-closed.
+
+pub mod store;
+
+use nostr::{Event, JsonUtil};
+use rusqlite::Connection;
+use serde::{Deserialize, Serialize};
+use tauri::State;
+
+use crate::app_state::AppState;
+use crate::managed_agents::nest_dir;
+use crate::relay::{query_relay, relay_ws_url_with_override};
+
+// ── Constants ───────────────────────────────────────────────────────────────
+
+const KIND_AGENT_OBSERVER_FRAME: u16 = 24200;
+const OBSERVER_FRAME_TELEMETRY: &str = "telemetry";
+
+// ── DB helpers ───────────────────────────────────────────────────────────────
+
+fn open_db() -> Result {
+ let nest = nest_dir().ok_or("cannot resolve nest directory for archive")?;
+ let db_path = nest.join("archive").join("archive.db");
+ store::open_archive_db(&db_path)
+}
+
+fn identity_pubkey(state: &AppState) -> Result {
+ let keys = state.keys.lock().map_err(|e| e.to_string())?;
+ Ok(keys.public_key().to_hex())
+}
+
+fn now_secs() -> i64 {
+ std::time::SystemTime::now()
+ .duration_since(std::time::UNIX_EPOCH)
+ .unwrap_or_default()
+ .as_secs() as i64
+}
+
+// ── Scope type ───────────────────────────────────────────────────────────────
+
+/// The three supported archive scope discriminants.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ScopeType {
+ ChannelH,
+ OwnerP,
+ ReferencedE,
+}
+
+impl ScopeType {
+ fn as_str(&self) -> &'static str {
+ match self {
+ ScopeType::ChannelH => "channel_h",
+ ScopeType::OwnerP => "owner_p",
+ ScopeType::ReferencedE => "referenced_e",
+ }
+ }
+
+ fn is_ephemeral(&self) -> bool {
+ matches!(self, ScopeType::OwnerP)
+ }
+}
+
+// ── archive_events ───────────────────────────────────────────────────────────
+
+/// One event candidate to archive.
+#[derive(Debug, Deserialize)]
+pub struct ArchiveCandidate {
+ /// Raw Nostr event JSON.
+ pub raw_event_json: String,
+ /// Which save scope this candidate was matched against. The backend
+ /// re-verifies this — it is never trusted blind.
+ pub matched_scope: MatchedScope,
+}
+
+/// A scope match assertion from the frontend.
+#[derive(Debug, Deserialize)]
+pub struct MatchedScope {
+ pub scope_type: ScopeType,
+ pub scope_value: String,
+}
+
+/// Result of a batch archive call.
+#[derive(Debug, Serialize)]
+pub struct ArchiveBatchResult {
+ /// Events successfully written to the store (event + scope rows).
+ pub persisted: u32,
+ /// Events dropped due to access denial or invalid payload (not an error).
+ pub dropped: u32,
+}
+
+/// Archive a batch of event candidates.
+///
+/// - Persistent scopes (`channel_h`, `referenced_e`): grouped by scope, then
+/// batch-queried against the relay; only returned events are inserted.
+/// - Ephemeral scope (`owner_p`): local validation only (no `/query`).
+#[tauri::command]
+pub async fn archive_events(
+ state: State<'_, AppState>,
+ candidates: Vec,
+) -> Result {
+ if candidates.is_empty() {
+ return Ok(ArchiveBatchResult {
+ persisted: 0,
+ dropped: 0,
+ });
+ }
+
+ let identity_pk = identity_pubkey(&state)?;
+ let relay_url = relay_ws_url_with_override(&state);
+ let now = now_secs();
+ let conn = open_db()?;
+
+ // Parse and verify all candidates up front; split by proof path.
+ struct Parsed {
+ event: Event,
+ raw_json: String,
+ matched_scope: MatchedScope,
+ }
+
+ let mut persistent: Vec = Vec::new();
+ let mut ephemeral: Vec = Vec::new();
+ let mut dropped: u32 = 0;
+
+ for cand in candidates {
+ let event = match Event::from_json(&cand.raw_event_json) {
+ Ok(e) => e,
+ Err(_) => {
+ dropped += 1;
+ continue;
+ }
+ };
+ if !event.verify_id() || !event.verify_signature() {
+ dropped += 1;
+ continue;
+ }
+
+ if cand.matched_scope.scope_type.is_ephemeral() {
+ ephemeral.push(Parsed {
+ event,
+ raw_json: cand.raw_event_json,
+ matched_scope: cand.matched_scope,
+ });
+ } else {
+ persistent.push(Parsed {
+ event,
+ raw_json: cand.raw_event_json,
+ matched_scope: cand.matched_scope,
+ });
+ }
+ }
+
+ let mut persisted: u32 = 0;
+
+ // ── Persistent path ──────────────────────────────────────────────────────
+ // Group candidates by scope to issue one /query filter per scope bucket,
+ // keeping filter sizes manageable.
+ if !persistent.is_empty() {
+ use std::collections::HashMap;
+
+ // Build a map: (scope_type_str, scope_value) → [parsed candidates]
+ let mut buckets: HashMap<(String, String), Vec> = HashMap::new();
+ for p in persistent {
+ let key = (
+ p.matched_scope.scope_type.as_str().to_string(),
+ p.matched_scope.scope_value.clone(),
+ );
+ buckets.entry(key).or_default().push(p);
+ }
+
+ for ((scope_type_str, scope_value), mut group) in buckets {
+ // Deduplicate by event id within the bucket.
+ let mut seen_ids = std::collections::HashSet::new();
+ group.retain(|p| seen_ids.insert(p.event.id.to_hex()));
+
+ let ids: Vec = group.iter().map(|p| p.event.id.to_hex()).collect();
+
+ // Build a filter for the relay /query. We always include the event
+ // ids; the relay's auth gate will strip any the user can't read.
+ let filter = serde_json::json!({ "ids": ids });
+ let returned = match query_relay(&state, &[filter]).await {
+ Ok(evs) => evs,
+ Err(_) => {
+ // Relay unreachable — drop the whole group rather than
+ // persisting unverified.
+ dropped += group.len() as u32;
+ continue;
+ }
+ };
+
+ // Index returned event ids for O(1) lookup.
+ let returned_ids: std::collections::HashSet = returned
+ .iter()
+ .map(|e| e.id.to_hex())
+ .collect();
+
+ for p in group {
+ let eid = p.event.id.to_hex();
+ if !returned_ids.contains(&eid) {
+ dropped += 1;
+ continue;
+ }
+
+ // Re-derive the matched scope from the event itself — never
+ // trust the frontend's matched_scope blind.
+ let verified_scope = match scope_type_str.as_str() {
+ "channel_h" => derive_channel_h_scope(&p.event),
+ "referenced_e" => derive_referenced_e_scope(&p.event, &scope_value),
+ _ => None,
+ };
+
+ let verified_scope_value = match verified_scope {
+ Some(v) => v,
+ None => {
+ dropped += 1;
+ continue;
+ }
+ };
+
+ // Check a matching subscription exists.
+ let sub_ok = store::has_save_subscription(
+ &conn,
+ &identity_pk,
+ &relay_url,
+ &scope_type_str,
+ &verified_scope_value,
+ )?;
+ if !sub_ok {
+ dropped += 1;
+ continue;
+ }
+
+ store::upsert_archived_event(
+ &conn,
+ &identity_pk,
+ &relay_url,
+ &eid,
+ p.event.kind.as_u16() as i64,
+ &p.event.pubkey.to_hex(),
+ p.event.created_at.as_secs() as i64,
+ &p.raw_json,
+ now,
+ )?;
+ store::upsert_event_scope(
+ &conn,
+ &identity_pk,
+ &relay_url,
+ &eid,
+ &scope_type_str,
+ &verified_scope_value,
+ now,
+ )?;
+ persisted += 1;
+ }
+ }
+ }
+
+ // ── Ephemeral path (owner_p) ─────────────────────────────────────────────
+ for p in ephemeral {
+ match validate_ephemeral_frame(&p.event, &identity_pk, &p.matched_scope.scope_value, &conn, &identity_pk, &relay_url) {
+ Ok(()) => {}
+ Err(_) => {
+ dropped += 1;
+ continue;
+ }
+ }
+
+ let eid = p.event.id.to_hex();
+ store::upsert_archived_event(
+ &conn,
+ &identity_pk,
+ &relay_url,
+ &eid,
+ p.event.kind.as_u16() as i64,
+ &p.event.pubkey.to_hex(),
+ p.event.created_at.as_secs() as i64,
+ &p.raw_json,
+ now,
+ )?;
+ store::upsert_event_scope(
+ &conn,
+ &identity_pk,
+ &relay_url,
+ &eid,
+ "owner_p",
+ &p.matched_scope.scope_value,
+ now,
+ )?;
+ persisted += 1;
+ }
+
+ Ok(ArchiveBatchResult { persisted, dropped })
+}
+
+/// Validate an ephemeral observer frame (kind 24200) against ALL local rules.
+///
+/// Rules (verbatim from spec):
+/// 1. kind == 24200
+/// 2. `#p` contains `identity_pubkey`
+/// 3. `agent` tag is present
+/// 4. `frame == "telemetry"` (control frames are not archived)
+/// 5. event author (pubkey) == agent tag value
+/// 6. A matching `owner_p` save-subscription exists for `scope_value`
+fn validate_ephemeral_frame(
+ event: &Event,
+ identity_pk: &str,
+ scope_value: &str,
+ conn: &Connection,
+ sub_identity: &str,
+ relay_url: &str,
+) -> Result<(), String> {
+ // 1. Kind guard.
+ if event.kind.as_u16() != KIND_AGENT_OBSERVER_FRAME {
+ return Err(format!(
+ "expected kind {KIND_AGENT_OBSERVER_FRAME}, got {}",
+ event.kind.as_u16()
+ ));
+ }
+
+ // 2. `#p` contains current identity.
+ let p_matches = event.tags.iter().any(|t| {
+ let s = t.as_slice();
+ s.len() >= 2 && s[0] == "p" && s[1] == identity_pk
+ });
+ if !p_matches {
+ return Err("observer frame #p does not match current identity".into());
+ }
+
+ // 3. `agent` tag present.
+ let agent_value = event
+ .tags
+ .iter()
+ .find_map(|t| {
+ let s = t.as_slice();
+ if s.len() >= 2 && s[0] == "agent" {
+ Some(s[1].clone())
+ } else {
+ None
+ }
+ })
+ .ok_or_else(|| "observer frame missing `agent` tag".to_string())?;
+
+ // 4. `frame == "telemetry"`.
+ let frame_value = event
+ .tags
+ .iter()
+ .find_map(|t| {
+ let s = t.as_slice();
+ if s.len() >= 2 && s[0] == "frame" {
+ Some(s[1].clone())
+ } else {
+ None
+ }
+ })
+ .ok_or_else(|| "observer frame missing `frame` tag".to_string())?;
+ if frame_value != OBSERVER_FRAME_TELEMETRY {
+ return Err(format!("expected frame=telemetry, got {frame_value:?}"));
+ }
+
+ // 5. Event author == agent tag value.
+ if event.pubkey.to_hex() != agent_value {
+ return Err("observer frame author does not match agent tag".into());
+ }
+
+ // 6. Matching owner_p subscription exists.
+ if !store::has_save_subscription(conn, sub_identity, relay_url, "owner_p", scope_value)? {
+ return Err(format!("no owner_p subscription for scope_value={scope_value:?}"));
+ }
+
+ Ok(())
+}
+
+/// Derive the `channel_h` scope value from an event: the first `h` tag value.
+fn derive_channel_h_scope(event: &Event) -> Option {
+ event.tags.iter().find_map(|t| {
+ let s = t.as_slice();
+ if s.len() >= 2 && s[0] == "h" && !s[1].is_empty() {
+ Some(s[1].clone())
+ } else {
+ None
+ }
+ })
+}
+
+/// Derive the `referenced_e` scope value: matches the claimed scope_value if the
+/// event contains an `e` tag pointing to it.
+fn derive_referenced_e_scope(event: &Event, claimed: &str) -> Option {
+ let found = event.tags.iter().any(|t| {
+ let s = t.as_slice();
+ s.len() >= 2 && s[0] == "e" && s[1] == claimed
+ });
+ if found {
+ Some(claimed.to_string())
+ } else {
+ None
+ }
+}
+
+// ── create_save_subscription ─────────────────────────────────────────────────
+
+/// Create a save subscription after running a per-scope access probe.
+///
+/// Probes:
+/// - `channel_h`: verify the current user is a member of the channel (kind 39002
+/// `#p` contains our pubkey, or we are the event author / open channel).
+/// - `referenced_e`: the referenced event id is currently readable via `/query`.
+/// - `owner_p`: restricted to the current identity's own pubkey (v1).
+#[tauri::command]
+pub async fn create_save_subscription(
+ state: State<'_, AppState>,
+ scope_type: ScopeType,
+ scope_value: String,
+ kinds: Vec,
+) -> Result<(), String> {
+ let identity_pk = identity_pubkey(&state)?;
+ let relay_url = relay_ws_url_with_override(&state);
+ let now = now_secs();
+
+ // Per-scope access probe.
+ match &scope_type {
+ ScopeType::ChannelH => {
+ probe_channel_access(&state, &identity_pk, &scope_value).await?;
+ }
+ ScopeType::ReferencedE => {
+ probe_event_readable(&state, &scope_value).await?;
+ }
+ ScopeType::OwnerP => {
+ // v1: only the current identity's own pubkey is allowed.
+ if scope_value != identity_pk {
+ return Err(format!(
+ "owner_p scope_value must equal current identity pubkey in v1 (got {scope_value:?})"
+ ));
+ }
+ }
+ }
+
+ let kinds_json = serde_json::to_string(&kinds)
+ .map_err(|e| format!("failed to serialize kinds: {e}"))?;
+
+ let conn = open_db()?;
+ store::upsert_save_subscription(
+ &conn,
+ &identity_pk,
+ &relay_url,
+ scope_type.as_str(),
+ &scope_value,
+ &kinds_json,
+ now,
+ )
+}
+
+/// Probe: the current user has access to `channel_id` (kind 39002 lists them).
+async fn probe_channel_access(
+ state: &AppState,
+ identity_pk: &str,
+ channel_id: &str,
+) -> Result<(), String> {
+ // Fetch the channel's members event (kind 39002, #d = channel_id).
+ let events = query_relay(
+ state,
+ &[serde_json::json!({
+ "kinds": [39002],
+ "#d": [channel_id],
+ "limit": 1
+ })],
+ )
+ .await?;
+
+ // If no members event exists this could be an open channel — try to read
+ // its metadata (kind 39000) as a fallback proof of readability.
+ if events.is_empty() {
+ let meta = query_relay(
+ state,
+ &[serde_json::json!({
+ "kinds": [39000],
+ "#d": [channel_id],
+ "limit": 1
+ })],
+ )
+ .await?;
+ if meta.is_empty() {
+ return Err(format!("channel {channel_id:?} not found or not accessible"));
+ }
+ // Open channel — readable, access granted.
+ return Ok(());
+ }
+
+ // Check that the current identity is listed as a member.
+ let ev = &events[0];
+ let is_member = ev.tags.iter().any(|t| {
+ let s = t.as_slice();
+ s.len() >= 2 && s[0] == "p" && s[1] == identity_pk
+ });
+ // Also allow if we are the event author (e.g. the workspace owner who
+ // published the members event may not be in the `#p` list themselves).
+ let is_author = ev.pubkey.to_hex() == identity_pk;
+
+ if is_member || is_author {
+ Ok(())
+ } else {
+ Err(format!(
+ "current identity is not a member of channel {channel_id:?}"
+ ))
+ }
+}
+
+/// Probe: the given event id is currently readable by the current user.
+async fn probe_event_readable(state: &AppState, event_id: &str) -> Result<(), String> {
+ let events = query_relay(
+ state,
+ &[serde_json::json!({
+ "ids": [event_id],
+ "limit": 1
+ })],
+ )
+ .await?;
+
+ if events.is_empty() {
+ return Err(format!("event {event_id:?} not found or not accessible"));
+ }
+ Ok(())
+}
+
+// ── list_save_subscriptions ──────────────────────────────────────────────────
+
+/// List all save subscriptions for the current identity + relay.
+#[tauri::command]
+pub fn list_save_subscriptions(
+ state: State<'_, AppState>,
+) -> Result, String> {
+ let identity_pk = identity_pubkey(&state)?;
+ let relay_url = relay_ws_url_with_override(&state);
+ let conn = open_db()?;
+ store::list_save_subscriptions(&conn, &identity_pk, &relay_url)
+}
+
+// ── delete_save_subscription ─────────────────────────────────────────────────
+
+/// Delete a save subscription.
+///
+/// Does NOT purge already-archived event data — retention is decoupled in v1.
+/// GC of orphaned event rows happens in P4 purge commands, not here.
+#[tauri::command]
+pub fn delete_save_subscription(
+ state: State<'_, AppState>,
+ scope_type: ScopeType,
+ scope_value: String,
+) -> Result {
+ let identity_pk = identity_pubkey(&state)?;
+ let relay_url = relay_ws_url_with_override(&state);
+ let conn = open_db()?;
+ store::delete_save_subscription(
+ &conn,
+ &identity_pk,
+ &relay_url,
+ scope_type.as_str(),
+ &scope_value,
+ )
+}
+
+// ── Tests ────────────────────────────────────────────────────────────────────
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use nostr::{EventBuilder, Keys, Kind, Tag};
+ use rusqlite::Connection;
+
+ // ── Helper: open an in-memory store ──────────────────────────────────────
+
+ 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(super::store::SCHEMA).unwrap();
+ conn
+ }
+
+ // ── Helper: build a real signed observer frame ────────────────────────────
+
+ fn make_observer_frame(
+ owner_keys: &Keys,
+ agent_keys: &Keys,
+ frame_type: &str,
+ ) -> Event {
+ let owner_pk = owner_keys.public_key().to_hex();
+ let agent_pk = agent_keys.public_key().to_hex();
+
+ // Minimal NIP-44-looking ciphertext (base64, long enough to pass heuristic).
+ let fake_ciphertext = "A".repeat(200);
+
+ let tags = vec![
+ Tag::parse(["p", &owner_pk]).unwrap(),
+ Tag::parse(["agent", &agent_pk]).unwrap(),
+ Tag::parse(["frame", frame_type]).unwrap(),
+ ];
+
+ EventBuilder::new(Kind::Custom(24200), &fake_ciphertext)
+ .tags(tags)
+ .sign_with_keys(agent_keys)
+ .unwrap()
+ }
+
+ // ── Ephemeral validator — individual condition rejection ──────────────────
+
+ fn add_owner_p_sub(conn: &Connection, identity_pk: &str, relay_url: &str, scope_value: &str) {
+ store::upsert_save_subscription(
+ conn,
+ identity_pk,
+ relay_url,
+ "owner_p",
+ scope_value,
+ "[24200]",
+ 0,
+ )
+ .unwrap();
+ }
+
+ #[test]
+ fn test_ephemeral_validator_accepts_valid_frame() {
+ let conn = in_memory();
+ let owner_keys = Keys::generate();
+ let agent_keys = Keys::generate();
+ let owner_pk = owner_keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+
+ add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
+ let ev = make_observer_frame(&owner_keys, &agent_keys, OBSERVER_FRAME_TELEMETRY);
+
+ assert!(validate_ephemeral_frame(
+ &ev,
+ &owner_pk,
+ &owner_pk,
+ &conn,
+ &owner_pk,
+ relay_url
+ )
+ .is_ok());
+ }
+
+ #[test]
+ fn test_ephemeral_validator_rejects_wrong_kind() {
+ let conn = in_memory();
+ let owner_keys = Keys::generate();
+ let agent_keys = Keys::generate();
+ let owner_pk = owner_keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
+
+ // kind 1 instead of 24200
+ let ev = EventBuilder::new(Kind::TextNote, "hello")
+ .tags(vec![
+ Tag::parse(["p", &owner_pk]).unwrap(),
+ Tag::parse(["agent", &agent_keys.public_key().to_hex()]).unwrap(),
+ Tag::parse(["frame", OBSERVER_FRAME_TELEMETRY]).unwrap(),
+ ])
+ .sign_with_keys(&agent_keys)
+ .unwrap();
+
+ let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ assert!(result.is_err());
+ assert!(result.unwrap_err().contains("kind"));
+ }
+
+ #[test]
+ fn test_ephemeral_validator_rejects_missing_p_tag() {
+ let conn = in_memory();
+ let owner_keys = Keys::generate();
+ let agent_keys = Keys::generate();
+ let owner_pk = owner_keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
+
+ // No `p` tag.
+ let ev = EventBuilder::new(Kind::Custom(24200), &"A".repeat(200))
+ .tags(vec![
+ Tag::parse(["agent", &agent_keys.public_key().to_hex()]).unwrap(),
+ Tag::parse(["frame", OBSERVER_FRAME_TELEMETRY]).unwrap(),
+ ])
+ .sign_with_keys(&agent_keys)
+ .unwrap();
+
+ let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ assert!(result.is_err());
+ assert!(result.unwrap_err().contains("#p"));
+ }
+
+ #[test]
+ fn test_ephemeral_validator_rejects_missing_agent_tag() {
+ let conn = in_memory();
+ let owner_keys = Keys::generate();
+ let agent_keys = Keys::generate();
+ let owner_pk = owner_keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
+
+ // No `agent` tag.
+ let ev = EventBuilder::new(Kind::Custom(24200), &"A".repeat(200))
+ .tags(vec![
+ Tag::parse(["p", &owner_pk]).unwrap(),
+ Tag::parse(["frame", OBSERVER_FRAME_TELEMETRY]).unwrap(),
+ ])
+ .sign_with_keys(&agent_keys)
+ .unwrap();
+
+ let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ assert!(result.is_err());
+ assert!(result.unwrap_err().contains("agent"));
+ }
+
+ #[test]
+ fn test_ephemeral_validator_rejects_control_frame() {
+ let conn = in_memory();
+ let owner_keys = Keys::generate();
+ let agent_keys = Keys::generate();
+ let owner_pk = owner_keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
+
+ // frame=control, not telemetry.
+ let ev = make_observer_frame(&owner_keys, &agent_keys, "control");
+ let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ assert!(result.is_err());
+ assert!(result.unwrap_err().contains("telemetry"));
+ }
+
+ #[test]
+ fn test_ephemeral_validator_rejects_wrong_author() {
+ let conn = in_memory();
+ let owner_keys = Keys::generate();
+ let agent_keys = Keys::generate();
+ let other_keys = Keys::generate();
+ let owner_pk = owner_keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
+
+ // event signed by `other_keys` but agent tag = agent_keys pubkey.
+ let ev = EventBuilder::new(Kind::Custom(24200), &"A".repeat(200))
+ .tags(vec![
+ Tag::parse(["p", &owner_pk]).unwrap(),
+ Tag::parse(["agent", &agent_keys.public_key().to_hex()]).unwrap(),
+ Tag::parse(["frame", OBSERVER_FRAME_TELEMETRY]).unwrap(),
+ ])
+ .sign_with_keys(&other_keys) // wrong signer
+ .unwrap();
+
+ let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ assert!(result.is_err());
+ assert!(result.unwrap_err().contains("author"));
+ }
+
+ #[test]
+ fn test_ephemeral_validator_rejects_no_subscription() {
+ let conn = in_memory();
+ let owner_keys = Keys::generate();
+ let agent_keys = Keys::generate();
+ let owner_pk = owner_keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ // Deliberately do NOT add a subscription.
+
+ let ev = make_observer_frame(&owner_keys, &agent_keys, OBSERVER_FRAME_TELEMETRY);
+ let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ assert!(result.is_err());
+ assert!(result.unwrap_err().contains("owner_p subscription"));
+ }
+
+ // ── Scope derivation ─────────────────────────────────────────────────────
+
+ #[test]
+ fn test_derive_channel_h_scope_extracts_h_tag() {
+ let keys = Keys::generate();
+ let ev = EventBuilder::new(Kind::TextNote, "hello")
+ .tags(vec![Tag::parse(["h", "chan-uuid-123"]).unwrap()])
+ .sign_with_keys(&keys)
+ .unwrap();
+ assert_eq!(
+ derive_channel_h_scope(&ev),
+ Some("chan-uuid-123".to_string())
+ );
+ }
+
+ #[test]
+ fn test_derive_channel_h_scope_returns_none_when_absent() {
+ let keys = Keys::generate();
+ let ev = EventBuilder::new(Kind::TextNote, "hello")
+ .sign_with_keys(&keys)
+ .unwrap();
+ assert_eq!(derive_channel_h_scope(&ev), None);
+ }
+
+ #[test]
+ fn test_derive_referenced_e_scope_matches_claimed() {
+ let keys = Keys::generate();
+ let ref_id = "a".repeat(64);
+ let ev = EventBuilder::new(Kind::TextNote, "reply")
+ .tags(vec![Tag::parse(["e", &ref_id]).unwrap()])
+ .sign_with_keys(&keys)
+ .unwrap();
+ assert_eq!(
+ derive_referenced_e_scope(&ev, &ref_id),
+ Some(ref_id.clone())
+ );
+ }
+
+ #[test]
+ fn test_derive_referenced_e_scope_rejects_wrong_claimed() {
+ let keys = Keys::generate();
+ let actual_ref = "a".repeat(64);
+ let claimed = "b".repeat(64);
+ let ev = EventBuilder::new(Kind::TextNote, "reply")
+ .tags(vec![Tag::parse(["e", &actual_ref]).unwrap()])
+ .sign_with_keys(&keys)
+ .unwrap();
+ assert_eq!(derive_referenced_e_scope(&ev, &claimed), None);
+ }
+
+ // ── Dropped-vs-persisted accounting ─────────────────────────────────────
+
+ #[test]
+ fn test_dropped_counting_invalid_json() {
+ // archive_events is async and needs AppState — we test the lower-level
+ // accounting by confirming that Event::from_json fails gracefully for
+ // malformed input and increments the dropped counter.
+ let result = Event::from_json("not json at all");
+ assert!(result.is_err());
+ }
+
+ #[test]
+ fn test_dropped_counting_bad_signature() {
+ let keys = Keys::generate();
+ // Build a valid event then tamper with the content to break the id
+ // (the id is a hash of the event fields including content).
+ let mut ev_json: serde_json::Value =
+ serde_json::from_str(&EventBuilder::new(Kind::TextNote, "ok")
+ .sign_with_keys(&keys)
+ .unwrap()
+ .as_json())
+ .unwrap();
+ ev_json["content"] = serde_json::Value::String("tampered".into());
+ let tampered = ev_json.to_string();
+ let ev = Event::from_json(&tampered).unwrap();
+ // After tampering the content the event id (a hash over all fields)
+ // no longer matches, so verify_id() returns false.
+ assert!(!ev.verify_id());
+ }
+}
diff --git a/desktop/src-tauri/src/archive/store.rs b/desktop/src-tauri/src/archive/store.rs
new file mode 100644
index 000000000..ac3890806
--- /dev/null
+++ b/desktop/src-tauri/src/archive/store.rs
@@ -0,0 +1,457 @@
+//! Local SQLite archive store for saved relay messages.
+//!
+//! Three tables:
+//! - `archived_events` — one raw event row per (identity, relay, event_id)
+//! - `archived_event_scopes` — N scope membership rows per raw event (many-to-many)
+//! - `save_subscriptions` — which scopes the user has subscribed to save
+//!
+//! WAL + `busy_timeout=5000` matches `managed_agents/retention.rs`.
+//! Raw event rows are GC'd when their last scope row is deleted.
+
+use std::path::Path;
+
+use rusqlite::{params, Connection};
+
+// ── Schema ─────────────────────────────────────────────────────────────────
+
+pub(super) const SCHEMA: &str = "
+CREATE TABLE IF NOT EXISTS archived_events (
+ identity_pubkey TEXT NOT NULL,
+ relay_url TEXT NOT NULL,
+ id TEXT NOT NULL,
+ kind INTEGER NOT NULL,
+ pubkey TEXT NOT NULL,
+ created_at INTEGER NOT NULL,
+ raw_json TEXT NOT NULL,
+ archived_at INTEGER NOT NULL,
+ PRIMARY KEY (identity_pubkey, relay_url, id)
+);
+
+CREATE TABLE IF NOT EXISTS archived_event_scopes (
+ identity_pubkey TEXT NOT NULL,
+ relay_url TEXT NOT NULL,
+ id TEXT NOT NULL,
+ scope_type TEXT NOT NULL,
+ scope_value TEXT NOT NULL,
+ archived_at INTEGER NOT NULL,
+ PRIMARY KEY (identity_pubkey, relay_url, id, scope_type, scope_value)
+);
+
+CREATE TABLE IF NOT EXISTS save_subscriptions (
+ identity_pubkey TEXT NOT NULL,
+ relay_url TEXT NOT NULL,
+ scope_type TEXT NOT NULL,
+ scope_value TEXT NOT NULL,
+ kinds TEXT NOT NULL,
+ created_at INTEGER NOT NULL,
+ PRIMARY KEY (identity_pubkey, relay_url, scope_type, scope_value)
+);
+";
+
+// ── Open / init ─────────────────────────────────────────────────────────────
+
+/// Open (or create) the archive database at the given path.
+///
+/// Applies WAL journaling and `busy_timeout=5000` on every connection,
+/// matching `managed_agents/retention.rs`. Creates all three tables if they
+/// don't already exist.
+pub fn open_archive_db(path: &Path) -> Result {
+ // Ensure the parent directory exists so `Connection::open` doesn't fail.
+ if let Some(parent) = path.parent() {
+ std::fs::create_dir_all(parent)
+ .map_err(|e| format!("failed to create archive dir: {e}"))?;
+ }
+
+ let conn =
+ Connection::open(path).map_err(|e| format!("failed to open archive db: {e}"))?;
+
+ conn.pragma_update(None, "journal_mode", "WAL")
+ .map_err(|e| format!("failed to set WAL mode: {e}"))?;
+ conn.pragma_update(None, "busy_timeout", 5000)
+ .map_err(|e| format!("failed to set busy_timeout: {e}"))?;
+
+ conn.execute_batch(SCHEMA)
+ .map_err(|e| format!("failed to initialize archive schema: {e}"))?;
+
+ Ok(conn)
+}
+
+// ── Save subscriptions ──────────────────────────────────────────────────────
+
+/// A save subscription row.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct SaveSubscription {
+ pub identity_pubkey: String,
+ pub relay_url: String,
+ pub scope_type: String,
+ pub scope_value: String,
+ /// JSON-encoded integer array, e.g. `[1,6,39000]`.
+ pub kinds: String,
+ pub created_at: i64,
+}
+
+/// Insert or replace a save subscription. `kinds` must be a JSON int array.
+pub fn upsert_save_subscription(
+ conn: &Connection,
+ identity_pubkey: &str,
+ relay_url: &str,
+ scope_type: &str,
+ scope_value: &str,
+ kinds_json: &str,
+ now: i64,
+) -> Result<(), String> {
+ conn.execute(
+ "INSERT INTO save_subscriptions
+ (identity_pubkey, relay_url, scope_type, scope_value, kinds, created_at)
+ VALUES (?1, ?2, ?3, ?4, ?5, ?6)
+ ON CONFLICT (identity_pubkey, relay_url, scope_type, scope_value)
+ DO UPDATE SET kinds = excluded.kinds",
+ params![identity_pubkey, relay_url, scope_type, scope_value, kinds_json, now],
+ )
+ .map_err(|e| format!("failed to upsert save subscription: {e}"))?;
+ Ok(())
+}
+
+/// List all save subscriptions for the given identity + relay.
+pub fn list_save_subscriptions(
+ conn: &Connection,
+ identity_pubkey: &str,
+ relay_url: &str,
+) -> Result, String> {
+ let mut stmt = conn
+ .prepare(
+ "SELECT identity_pubkey, relay_url, scope_type, scope_value, kinds, created_at
+ FROM save_subscriptions
+ WHERE identity_pubkey = ?1 AND relay_url = ?2
+ ORDER BY created_at ASC",
+ )
+ .map_err(|e| format!("prepare list_save_subscriptions: {e}"))?;
+
+ let rows = stmt
+ .query_map(params![identity_pubkey, relay_url], |row| {
+ Ok(SaveSubscription {
+ identity_pubkey: row.get(0)?,
+ relay_url: row.get(1)?,
+ scope_type: row.get(2)?,
+ scope_value: row.get(3)?,
+ kinds: row.get(4)?,
+ created_at: row.get(5)?,
+ })
+ })
+ .map_err(|e| format!("query list_save_subscriptions: {e}"))?;
+
+ rows.collect::, _>>()
+ .map_err(|e| format!("read list_save_subscriptions row: {e}"))
+}
+
+/// Delete a save subscription. Does NOT purge archived event data (retention
+/// is decoupled in v1). Returns `true` if a row was deleted.
+pub fn delete_save_subscription(
+ conn: &Connection,
+ identity_pubkey: &str,
+ relay_url: &str,
+ scope_type: &str,
+ scope_value: &str,
+) -> Result {
+ let affected = conn
+ .execute(
+ "DELETE FROM save_subscriptions
+ WHERE identity_pubkey = ?1
+ AND relay_url = ?2
+ AND scope_type = ?3
+ AND scope_value = ?4",
+ params![identity_pubkey, relay_url, scope_type, scope_value],
+ )
+ .map_err(|e| format!("failed to delete save subscription: {e}"))?;
+ Ok(affected > 0)
+}
+
+/// Return true if a matching save subscription exists for the given scope.
+pub fn has_save_subscription(
+ conn: &Connection,
+ identity_pubkey: &str,
+ relay_url: &str,
+ scope_type: &str,
+ scope_value: &str,
+) -> Result {
+ let count: i64 = conn
+ .query_row(
+ "SELECT COUNT(*) FROM save_subscriptions
+ WHERE identity_pubkey = ?1
+ AND relay_url = ?2
+ AND scope_type = ?3
+ AND scope_value = ?4",
+ params![identity_pubkey, relay_url, scope_type, scope_value],
+ |row| row.get(0),
+ )
+ .map_err(|e| format!("failed to check save subscription: {e}"))?;
+ Ok(count > 0)
+}
+
+// ── Archived events ─────────────────────────────────────────────────────────
+
+/// Upsert an event row (idempotent on the PK).
+///
+/// Does nothing if the event is already archived (same identity/relay/id).
+pub fn upsert_archived_event(
+ conn: &Connection,
+ identity_pubkey: &str,
+ relay_url: &str,
+ event_id: &str,
+ kind: i64,
+ pubkey: &str,
+ created_at: i64,
+ raw_json: &str,
+ archived_at: i64,
+) -> Result<(), String> {
+ conn.execute(
+ "INSERT INTO archived_events
+ (identity_pubkey, relay_url, id, kind, pubkey, created_at, raw_json, archived_at)
+ VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
+ ON CONFLICT (identity_pubkey, relay_url, id) DO NOTHING",
+ params![
+ identity_pubkey,
+ relay_url,
+ event_id,
+ kind,
+ pubkey,
+ created_at,
+ raw_json,
+ archived_at
+ ],
+ )
+ .map_err(|e| format!("failed to upsert archived event: {e}"))?;
+ Ok(())
+}
+
+/// Upsert a scope membership row for an event.
+///
+/// Idempotent: if the (identity, relay, id, scope_type, scope_value) PK already
+/// exists the row is left unchanged.
+pub fn upsert_event_scope(
+ conn: &Connection,
+ identity_pubkey: &str,
+ relay_url: &str,
+ event_id: &str,
+ scope_type: &str,
+ scope_value: &str,
+ archived_at: i64,
+) -> Result<(), String> {
+ conn.execute(
+ "INSERT INTO archived_event_scopes
+ (identity_pubkey, relay_url, id, scope_type, scope_value, archived_at)
+ VALUES (?1, ?2, ?3, ?4, ?5, ?6)
+ ON CONFLICT (identity_pubkey, relay_url, id, scope_type, scope_value) DO NOTHING",
+ params![
+ identity_pubkey,
+ relay_url,
+ event_id,
+ scope_type,
+ scope_value,
+ archived_at
+ ],
+ )
+ .map_err(|e| format!("failed to upsert event scope: {e}"))?;
+ Ok(())
+}
+
+/// GC: delete orphaned event rows whose last scope row was just removed.
+///
+/// Called after any batch deletion of scope rows. Uses a LEFT JOIN so only
+/// events with zero remaining scope rows are deleted.
+#[allow(dead_code)] // Used by P4 purge commands; not yet wired to a Tauri command.
+pub fn gc_orphaned_events(
+ conn: &Connection,
+ identity_pubkey: &str,
+ relay_url: &str,
+) -> Result {
+ let affected = conn
+ .execute(
+ "DELETE FROM archived_events
+ WHERE identity_pubkey = ?1
+ AND relay_url = ?2
+ AND id NOT IN (
+ SELECT id FROM archived_event_scopes
+ WHERE identity_pubkey = ?1
+ AND relay_url = ?2
+ )",
+ params![identity_pubkey, relay_url],
+ )
+ .map_err(|e| format!("failed to gc orphaned events: {e}"))?;
+ Ok(affected)
+}
+
+// ── 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());
+ }
+
+ // ── 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);
+ }
+}
diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs
index 93424cd0f..6ca7f836a 100644
--- a/desktop/src-tauri/src/lib.rs
+++ b/desktop/src-tauri/src/lib.rs
@@ -1,4 +1,5 @@
mod app_state;
+mod archive;
mod commands;
mod deep_link;
mod events;
@@ -620,6 +621,10 @@ pub fn run() {
get_agent_memory,
relay_reconnect_hook,
relay_reconnect_hook_configured,
+ archive::archive_events,
+ archive::create_save_subscription,
+ archive::list_save_subscriptions,
+ archive::delete_save_subscription,
])
.build(tauri::generate_context!())
.expect("error while building tauri application");
From 5778e4d8020623e475f7c5703c4107e61eefc3b2 Mon Sep 17 00:00:00 2001
From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
Date: Wed, 1 Jul 2026 16:34:58 -0400
Subject: [PATCH 02/19] chore(archive): apply cargo fmt to archive module
Co-authored-by: Will Pfleger
Signed-off-by: Will Pfleger
---
desktop/src-tauri/src/archive/mod.rs | 73 ++++++++++++++------------
desktop/src-tauri/src/archive/store.rs | 25 ++++-----
2 files changed, 53 insertions(+), 45 deletions(-)
diff --git a/desktop/src-tauri/src/archive/mod.rs b/desktop/src-tauri/src/archive/mod.rs
index ab56c5469..5189dc183 100644
--- a/desktop/src-tauri/src/archive/mod.rs
+++ b/desktop/src-tauri/src/archive/mod.rs
@@ -204,10 +204,8 @@ pub async fn archive_events(
};
// Index returned event ids for O(1) lookup.
- let returned_ids: std::collections::HashSet = returned
- .iter()
- .map(|e| e.id.to_hex())
- .collect();
+ let returned_ids: std::collections::HashSet =
+ returned.iter().map(|e| e.id.to_hex()).collect();
for p in group {
let eid = p.event.id.to_hex();
@@ -272,7 +270,14 @@ pub async fn archive_events(
// ── Ephemeral path (owner_p) ─────────────────────────────────────────────
for p in ephemeral {
- match validate_ephemeral_frame(&p.event, &identity_pk, &p.matched_scope.scope_value, &conn, &identity_pk, &relay_url) {
+ match validate_ephemeral_frame(
+ &p.event,
+ &identity_pk,
+ &p.matched_scope.scope_value,
+ &conn,
+ &identity_pk,
+ &relay_url,
+ ) {
Ok(()) => {}
Err(_) => {
dropped += 1;
@@ -379,7 +384,9 @@ fn validate_ephemeral_frame(
// 6. Matching owner_p subscription exists.
if !store::has_save_subscription(conn, sub_identity, relay_url, "owner_p", scope_value)? {
- return Err(format!("no owner_p subscription for scope_value={scope_value:?}"));
+ return Err(format!(
+ "no owner_p subscription for scope_value={scope_value:?}"
+ ));
}
Ok(())
@@ -449,8 +456,8 @@ pub async fn create_save_subscription(
}
}
- let kinds_json = serde_json::to_string(&kinds)
- .map_err(|e| format!("failed to serialize kinds: {e}"))?;
+ let kinds_json =
+ serde_json::to_string(&kinds).map_err(|e| format!("failed to serialize kinds: {e}"))?;
let conn = open_db()?;
store::upsert_save_subscription(
@@ -494,7 +501,9 @@ async fn probe_channel_access(
)
.await?;
if meta.is_empty() {
- return Err(format!("channel {channel_id:?} not found or not accessible"));
+ return Err(format!(
+ "channel {channel_id:?} not found or not accessible"
+ ));
}
// Open channel — readable, access granted.
return Ok(());
@@ -593,11 +602,7 @@ mod tests {
// ── Helper: build a real signed observer frame ────────────────────────────
- fn make_observer_frame(
- owner_keys: &Keys,
- agent_keys: &Keys,
- frame_type: &str,
- ) -> Event {
+ fn make_observer_frame(owner_keys: &Keys, agent_keys: &Keys, frame_type: &str) -> Event {
let owner_pk = owner_keys.public_key().to_hex();
let agent_pk = agent_keys.public_key().to_hex();
@@ -642,15 +647,10 @@ mod tests {
add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
let ev = make_observer_frame(&owner_keys, &agent_keys, OBSERVER_FRAME_TELEMETRY);
- assert!(validate_ephemeral_frame(
- &ev,
- &owner_pk,
- &owner_pk,
- &conn,
- &owner_pk,
- relay_url
- )
- .is_ok());
+ assert!(
+ validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url)
+ .is_ok()
+ );
}
#[test]
@@ -672,7 +672,8 @@ mod tests {
.sign_with_keys(&agent_keys)
.unwrap();
- let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ let result =
+ validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
assert!(result.is_err());
assert!(result.unwrap_err().contains("kind"));
}
@@ -695,7 +696,8 @@ mod tests {
.sign_with_keys(&agent_keys)
.unwrap();
- let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ let result =
+ validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
assert!(result.is_err());
assert!(result.unwrap_err().contains("#p"));
}
@@ -718,7 +720,8 @@ mod tests {
.sign_with_keys(&agent_keys)
.unwrap();
- let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ let result =
+ validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
assert!(result.is_err());
assert!(result.unwrap_err().contains("agent"));
}
@@ -734,7 +737,8 @@ mod tests {
// frame=control, not telemetry.
let ev = make_observer_frame(&owner_keys, &agent_keys, "control");
- let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ let result =
+ validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
assert!(result.is_err());
assert!(result.unwrap_err().contains("telemetry"));
}
@@ -759,7 +763,8 @@ mod tests {
.sign_with_keys(&other_keys) // wrong signer
.unwrap();
- let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ let result =
+ validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
assert!(result.is_err());
assert!(result.unwrap_err().contains("author"));
}
@@ -774,7 +779,8 @@ mod tests {
// Deliberately do NOT add a subscription.
let ev = make_observer_frame(&owner_keys, &agent_keys, OBSERVER_FRAME_TELEMETRY);
- let result = validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ let result =
+ validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
assert!(result.is_err());
assert!(result.unwrap_err().contains("owner_p subscription"));
}
@@ -845,12 +851,13 @@ mod tests {
let keys = Keys::generate();
// Build a valid event then tamper with the content to break the id
// (the id is a hash of the event fields including content).
- let mut ev_json: serde_json::Value =
- serde_json::from_str(&EventBuilder::new(Kind::TextNote, "ok")
+ let mut ev_json: serde_json::Value = serde_json::from_str(
+ &EventBuilder::new(Kind::TextNote, "ok")
.sign_with_keys(&keys)
.unwrap()
- .as_json())
- .unwrap();
+ .as_json(),
+ )
+ .unwrap();
ev_json["content"] = serde_json::Value::String("tampered".into());
let tampered = ev_json.to_string();
let ev = Event::from_json(&tampered).unwrap();
diff --git a/desktop/src-tauri/src/archive/store.rs b/desktop/src-tauri/src/archive/store.rs
index ac3890806..c2676323a 100644
--- a/desktop/src-tauri/src/archive/store.rs
+++ b/desktop/src-tauri/src/archive/store.rs
@@ -62,8 +62,7 @@ pub fn open_archive_db(path: &Path) -> Result {
.map_err(|e| format!("failed to create archive dir: {e}"))?;
}
- let conn =
- Connection::open(path).map_err(|e| format!("failed to open archive db: {e}"))?;
+ let conn = Connection::open(path).map_err(|e| format!("failed to open archive db: {e}"))?;
conn.pragma_update(None, "journal_mode", "WAL")
.map_err(|e| format!("failed to set WAL mode: {e}"))?;
@@ -106,7 +105,14 @@ pub fn upsert_save_subscription(
VALUES (?1, ?2, ?3, ?4, ?5, ?6)
ON CONFLICT (identity_pubkey, relay_url, scope_type, scope_value)
DO UPDATE SET kinds = excluded.kinds",
- params![identity_pubkey, relay_url, scope_type, scope_value, kinds_json, now],
+ params![
+ identity_pubkey,
+ relay_url,
+ scope_type,
+ scope_value,
+ kinds_json,
+ now
+ ],
)
.map_err(|e| format!("failed to upsert save subscription: {e}"))?;
Ok(())
@@ -359,8 +365,7 @@ mod tests {
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();
+ 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());
@@ -377,8 +382,7 @@ mod tests {
#[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();
+ 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());
}
@@ -426,11 +430,8 @@ mod tests {
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();
+ 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
From b2b87031460cb9148f7c7f462dd733cc772fe266 Mon Sep 17 00:00:00 2001
From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
Date: Wed, 1 Jul 2026 16:51:39 -0400
Subject: [PATCH 03/19] fix(archive): enforce subscription kinds, scoped relay
proof, e2e tests
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Three IMPORTANT defects from Thufir pass 1:
1. Kinds not enforced at archive time: add get_subscription_kinds() to
store.rs (returns Option via OptionalExtension). Both paths
now parse the kinds JSON and reject events whose kind is not in the
subscription's allowed list.
2. Persistent proof path used unscoped relay filter: replace bare
{ids}-only /query with {ids, #h/#e: scope_value, kinds: allowed_kinds}.
The relay returning an event for this scoped filter IS proof of scope
membership — no local tag re-derivation needed, which also correctly
admits h-less events matched via StoredEvent.channel_id fallback.
3. Missing end-to-end tests: add plan_archive / commit_archive split
(extracted from archive_events for Send-safety and testability),
plus 14 new tests covering: persistent persists/drops, h-less event
via scoped filter proof, kind mismatch drops (both paths), no-sub
drops, mixed batch exact counts, ephemeral persists/drops,
ephemeral kind-not-in-subscription drops.
Send-safety: rusqlite::Connection is !Send. archive_events now follows
the managed_agents/persona_events.rs pattern — DB work in scoped blocks
that drop the connection before any .await — so the Tauri command future
is Send.
Co-authored-by: Will Pfleger
Signed-off-by: Will Pfleger
---
desktop/src-tauri/src/archive/mod.rs | 778 ++++++++++++++++++-------
desktop/src-tauri/src/archive/store.rs | 26 +-
2 files changed, 580 insertions(+), 224 deletions(-)
diff --git a/desktop/src-tauri/src/archive/mod.rs b/desktop/src-tauri/src/archive/mod.rs
index 5189dc183..6bcf203c2 100644
--- a/desktop/src-tauri/src/archive/mod.rs
+++ b/desktop/src-tauri/src/archive/mod.rs
@@ -109,44 +109,101 @@ pub struct ArchiveBatchResult {
/// - Persistent scopes (`channel_h`, `referenced_e`): grouped by scope, then
/// batch-queried against the relay; only returned events are inserted.
/// - Ephemeral scope (`owner_p`): local validation only (no `/query`).
+///
+/// # Send-safety
+///
+/// `rusqlite::Connection` is `!Send`. All DB work is bracketed in scoped
+/// `{ let conn = open_db()?; ... }` blocks that drop the connection before any
+/// `.await`, exactly matching the pattern in `managed_agents/persona_events.rs`.
#[tauri::command]
pub async fn archive_events(
state: State<'_, AppState>,
candidates: Vec,
) -> Result {
- if candidates.is_empty() {
- return Ok(ArchiveBatchResult {
- persisted: 0,
- dropped: 0,
- });
- }
-
let identity_pk = identity_pubkey(&state)?;
let relay_url = relay_ws_url_with_override(&state);
let now = now_secs();
+
+ // ── Phase 1: plan (sync) ─────────────────────────────────────────────────
+ // Read subscriptions and build relay filters. Connection dropped before
+ // any .await.
+ let plan = {
+ let conn = open_db()?;
+ plan_archive(candidates, &identity_pk, &relay_url, &conn)?
+ // conn drops here
+ };
+
+ // ── Phase 2: relay queries (async) ───────────────────────────────────────
+ // No Connection in scope — future is Send.
+ let state_ref: &AppState = &state;
+ let bucket_results = query_buckets(plan.buckets, state_ref).await;
+
+ // ── Phase 3: persist (sync) ──────────────────────────────────────────────
let conn = open_db()?;
+ commit_archive(
+ bucket_results,
+ plan.ephemeral,
+ plan.pre_dropped,
+ &identity_pk,
+ &relay_url,
+ now,
+ &conn,
+ )
+}
- // Parse and verify all candidates up front; split by proof path.
- struct Parsed {
- event: Event,
- raw_json: String,
- matched_scope: MatchedScope,
- }
+// ── Archive internals ────────────────────────────────────────────────────────
+
+/// A parsed, sig-verified candidate ready for further processing.
+struct Parsed {
+ event: Event,
+ raw_json: String,
+ matched_scope: MatchedScope,
+}
- let mut persistent: Vec = Vec::new();
+/// One scope bucket: a set of candidates that share a scope type+value,
+/// with the relay filter already built and the subscription kinds loaded.
+struct Bucket {
+ scope_type_str: String,
+ scope_value: String,
+ allowed_kinds: Vec,
+ filter: serde_json::Value,
+ group: Vec,
+}
+
+/// Output of the sync planning phase.
+struct ArchivePlan {
+ buckets: Vec,
+ ephemeral: Vec,
+ /// Events already accounted as dropped during planning (no subscription,
+ /// unknown scope type, parse failure, bad sig).
+ pre_dropped: u32,
+}
+
+/// Phase 1 (sync): parse all candidates, group persistent ones into per-scope
+/// buckets, and load the subscription kinds for each bucket.
+///
+/// Returns an [`ArchivePlan`] with no `&Connection` remaining — safe to hold
+/// across `.await`.
+fn plan_archive(
+ candidates: Vec,
+ identity_pk: &str,
+ relay_url: &str,
+ conn: &Connection,
+) -> Result {
+ let mut persistent_raw: Vec = Vec::new();
let mut ephemeral: Vec = Vec::new();
- let mut dropped: u32 = 0;
+ let mut pre_dropped: u32 = 0;
for cand in candidates {
let event = match Event::from_json(&cand.raw_event_json) {
Ok(e) => e,
Err(_) => {
- dropped += 1;
+ pre_dropped += 1;
continue;
}
};
if !event.verify_id() || !event.verify_signature() {
- dropped += 1;
+ pre_dropped += 1;
continue;
}
@@ -157,7 +214,7 @@ pub async fn archive_events(
matched_scope: cand.matched_scope,
});
} else {
- persistent.push(Parsed {
+ persistent_raw.push(Parsed {
event,
raw_json: cand.raw_event_json,
matched_scope: cand.matched_scope,
@@ -165,118 +222,188 @@ pub async fn archive_events(
}
}
+ // Group persistent candidates by (scope_type, scope_value).
+ use std::collections::HashMap;
+ let mut scope_groups: HashMap<(String, String), Vec> = HashMap::new();
+ for p in persistent_raw {
+ let key = (
+ p.matched_scope.scope_type.as_str().to_string(),
+ p.matched_scope.scope_value.clone(),
+ );
+ scope_groups.entry(key).or_default().push(p);
+ }
+
+ let mut buckets: Vec = Vec::with_capacity(scope_groups.len());
+ for ((scope_type_str, scope_value), mut group) in scope_groups {
+ // No subscription → drop the whole group.
+ let kinds_json = match store::get_subscription_kinds(
+ conn,
+ identity_pk,
+ relay_url,
+ &scope_type_str,
+ &scope_value,
+ )? {
+ Some(k) => k,
+ None => {
+ pre_dropped += group.len() as u32;
+ continue;
+ }
+ };
+
+ let allowed_kinds: Vec =
+ serde_json::from_str::>(&kinds_json).unwrap_or_default();
+
+ // Deduplicate by event id within the bucket.
+ let mut seen = std::collections::HashSet::new();
+ group.retain(|p| seen.insert(p.event.id.to_hex()));
+
+ let ids: Vec = group.iter().map(|p| p.event.id.to_hex()).collect();
+
+ // Build a *scoped* relay filter: ids + scope tag + kinds.
+ let filter = match scope_type_str.as_str() {
+ "channel_h" => serde_json::json!({
+ "ids": ids,
+ "#h": [&scope_value],
+ "kinds": allowed_kinds,
+ }),
+ "referenced_e" => serde_json::json!({
+ "ids": ids,
+ "#e": [&scope_value],
+ "kinds": allowed_kinds,
+ }),
+ _ => {
+ pre_dropped += group.len() as u32;
+ continue;
+ }
+ };
+
+ buckets.push(Bucket {
+ scope_type_str,
+ scope_value,
+ allowed_kinds,
+ filter,
+ group,
+ });
+ }
+
+ Ok(ArchivePlan {
+ buckets,
+ ephemeral,
+ pre_dropped,
+ })
+}
+
+/// A bucket with the relay's response attached.
+struct BucketWithResult {
+ scope_type_str: String,
+ scope_value: String,
+ allowed_kinds: Vec,
+ group: Vec,
+ /// Event ids returned by the relay for the scoped filter.
+ returned_ids: std::collections::HashSet,
+ /// True if the relay query failed (network error); entire group dropped.
+ relay_failed: bool,
+}
+
+/// Phase 2 (async): fire one relay query per bucket and collect results.
+///
+/// `state` is `&AppState` — a `Copy` reference — so no `!Send` value is held
+/// across `.await`.
+async fn query_buckets(buckets: Vec, state: &AppState) -> Vec {
+ let mut results: Vec = Vec::with_capacity(buckets.len());
+ for bucket in buckets {
+ let (returned_ids, relay_failed) = match query_relay(state, &[bucket.filter]).await {
+ Ok(evs) => (evs.iter().map(|e| e.id.to_hex()).collect(), false),
+ Err(_) => (std::collections::HashSet::new(), true),
+ };
+ results.push(BucketWithResult {
+ scope_type_str: bucket.scope_type_str,
+ scope_value: bucket.scope_value,
+ allowed_kinds: bucket.allowed_kinds,
+ group: bucket.group,
+ returned_ids,
+ relay_failed,
+ });
+ }
+ results
+}
+
+/// Phase 3 (sync): apply relay results and write accepted events to the store.
+fn commit_archive(
+ bucket_results: Vec,
+ ephemeral: Vec,
+ pre_dropped: u32,
+ identity_pk: &str,
+ relay_url: &str,
+ now: i64,
+ conn: &Connection,
+) -> Result {
let mut persisted: u32 = 0;
+ let mut dropped: u32 = pre_dropped;
// ── Persistent path ──────────────────────────────────────────────────────
- // Group candidates by scope to issue one /query filter per scope bucket,
- // keeping filter sizes manageable.
- if !persistent.is_empty() {
- use std::collections::HashMap;
-
- // Build a map: (scope_type_str, scope_value) → [parsed candidates]
- let mut buckets: HashMap<(String, String), Vec> = HashMap::new();
- for p in persistent {
- let key = (
- p.matched_scope.scope_type.as_str().to_string(),
- p.matched_scope.scope_value.clone(),
- );
- buckets.entry(key).or_default().push(p);
+ for result in bucket_results {
+ if result.relay_failed {
+ dropped += result.group.len() as u32;
+ continue;
}
- for ((scope_type_str, scope_value), mut group) in buckets {
- // Deduplicate by event id within the bucket.
- let mut seen_ids = std::collections::HashSet::new();
- group.retain(|p| seen_ids.insert(p.event.id.to_hex()));
-
- let ids: Vec = group.iter().map(|p| p.event.id.to_hex()).collect();
-
- // Build a filter for the relay /query. We always include the event
- // ids; the relay's auth gate will strip any the user can't read.
- let filter = serde_json::json!({ "ids": ids });
- let returned = match query_relay(&state, &[filter]).await {
- Ok(evs) => evs,
- Err(_) => {
- // Relay unreachable — drop the whole group rather than
- // persisting unverified.
- dropped += group.len() as u32;
- continue;
- }
- };
-
- // Index returned event ids for O(1) lookup.
- let returned_ids: std::collections::HashSet =
- returned.iter().map(|e| e.id.to_hex()).collect();
-
- for p in group {
- let eid = p.event.id.to_hex();
- if !returned_ids.contains(&eid) {
- dropped += 1;
- continue;
- }
-
- // Re-derive the matched scope from the event itself — never
- // trust the frontend's matched_scope blind.
- let verified_scope = match scope_type_str.as_str() {
- "channel_h" => derive_channel_h_scope(&p.event),
- "referenced_e" => derive_referenced_e_scope(&p.event, &scope_value),
- _ => None,
- };
-
- let verified_scope_value = match verified_scope {
- Some(v) => v,
- None => {
- dropped += 1;
- continue;
- }
- };
-
- // Check a matching subscription exists.
- let sub_ok = store::has_save_subscription(
- &conn,
- &identity_pk,
- &relay_url,
- &scope_type_str,
- &verified_scope_value,
- )?;
- if !sub_ok {
- dropped += 1;
- continue;
- }
-
- store::upsert_archived_event(
- &conn,
- &identity_pk,
- &relay_url,
- &eid,
- p.event.kind.as_u16() as i64,
- &p.event.pubkey.to_hex(),
- p.event.created_at.as_secs() as i64,
- &p.raw_json,
- now,
- )?;
- store::upsert_event_scope(
- &conn,
- &identity_pk,
- &relay_url,
- &eid,
- &scope_type_str,
- &verified_scope_value,
- now,
- )?;
- persisted += 1;
+ for p in result.group {
+ let eid = p.event.id.to_hex();
+
+ // Relay proof: event was returned for the scoped filter.
+ if !result.returned_ids.contains(&eid) {
+ dropped += 1;
+ continue;
+ }
+
+ // Kind enforcement: event.kind must be in the subscription's list.
+ if !result
+ .allowed_kinds
+ .contains(&(p.event.kind.as_u16() as u64))
+ {
+ dropped += 1;
+ continue;
}
+
+ // The relay returning this event for {ids, #h/#e, kinds} IS the
+ // proof of scope membership. Use scope_value directly; no local
+ // tag re-derivation (which would incorrectly drop h-less events
+ // matched via the relay's StoredEvent.channel_id fallback).
+ store::upsert_archived_event(
+ conn,
+ identity_pk,
+ relay_url,
+ &eid,
+ p.event.kind.as_u16() as i64,
+ &p.event.pubkey.to_hex(),
+ p.event.created_at.as_secs() as i64,
+ &p.raw_json,
+ now,
+ )?;
+ store::upsert_event_scope(
+ conn,
+ identity_pk,
+ relay_url,
+ &eid,
+ &result.scope_type_str,
+ &result.scope_value,
+ now,
+ )?;
+ persisted += 1;
}
}
// ── Ephemeral path (owner_p) ─────────────────────────────────────────────
+ // Fully local validation — no relay query.
for p in ephemeral {
match validate_ephemeral_frame(
&p.event,
- &identity_pk,
+ identity_pk,
&p.matched_scope.scope_value,
- &conn,
- &identity_pk,
- &relay_url,
+ conn,
+ identity_pk,
+ relay_url,
) {
Ok(()) => {}
Err(_) => {
@@ -287,9 +414,9 @@ pub async fn archive_events(
let eid = p.event.id.to_hex();
store::upsert_archived_event(
- &conn,
- &identity_pk,
- &relay_url,
+ conn,
+ identity_pk,
+ relay_url,
&eid,
p.event.kind.as_u16() as i64,
&p.event.pubkey.to_hex(),
@@ -298,9 +425,9 @@ pub async fn archive_events(
now,
)?;
store::upsert_event_scope(
- &conn,
- &identity_pk,
- &relay_url,
+ conn,
+ identity_pk,
+ relay_url,
&eid,
"owner_p",
&p.matched_scope.scope_value,
@@ -320,7 +447,8 @@ pub async fn archive_events(
/// 3. `agent` tag is present
/// 4. `frame == "telemetry"` (control frames are not archived)
/// 5. event author (pubkey) == agent tag value
-/// 6. A matching `owner_p` save-subscription exists for `scope_value`
+/// 6. A matching `owner_p` save-subscription exists AND its `kinds` list
+/// includes `24200` (kinds enforcement mirrors the persistent path).
fn validate_ephemeral_frame(
event: &Event,
identity_pk: &str,
@@ -382,42 +510,20 @@ fn validate_ephemeral_frame(
return Err("observer frame author does not match agent tag".into());
}
- // 6. Matching owner_p subscription exists.
- if !store::has_save_subscription(conn, sub_identity, relay_url, "owner_p", scope_value)? {
+ // 6. Matching owner_p subscription exists AND kind 24200 is in its kinds list.
+ let kinds_json =
+ store::get_subscription_kinds(conn, sub_identity, relay_url, "owner_p", scope_value)?
+ .ok_or_else(|| format!("no owner_p subscription for scope_value={scope_value:?}"))?;
+ let allowed_kinds: Vec = serde_json::from_str::>(&kinds_json).unwrap_or_default();
+ if !allowed_kinds.contains(&(KIND_AGENT_OBSERVER_FRAME as u64)) {
return Err(format!(
- "no owner_p subscription for scope_value={scope_value:?}"
+ "owner_p subscription kinds {kinds_json:?} does not include {KIND_AGENT_OBSERVER_FRAME}"
));
}
Ok(())
}
-/// Derive the `channel_h` scope value from an event: the first `h` tag value.
-fn derive_channel_h_scope(event: &Event) -> Option {
- event.tags.iter().find_map(|t| {
- let s = t.as_slice();
- if s.len() >= 2 && s[0] == "h" && !s[1].is_empty() {
- Some(s[1].clone())
- } else {
- None
- }
- })
-}
-
-/// Derive the `referenced_e` scope value: matches the claimed scope_value if the
-/// event contains an `e` tag pointing to it.
-fn derive_referenced_e_scope(event: &Event, claimed: &str) -> Option {
- let found = event.tags.iter().any(|t| {
- let s = t.as_slice();
- s.len() >= 2 && s[0] == "e" && s[1] == claimed
- });
- if found {
- Some(claimed.to_string())
- } else {
- None
- }
-}
-
// ── create_save_subscription ─────────────────────────────────────────────────
/// Create a save subscription after running a per-scope access probe.
@@ -587,10 +693,10 @@ pub fn delete_save_subscription(
#[cfg(test)]
mod tests {
use super::*;
- use nostr::{EventBuilder, Keys, Kind, Tag};
+ use nostr::{EventBuilder, JsonUtil, Keys, Kind, Tag};
use rusqlite::Connection;
- // ── Helper: open an in-memory store ──────────────────────────────────────
+ // ── Helpers ──────────────────────────────────────────────────────────────
fn in_memory() -> Connection {
let conn = Connection::open_in_memory().unwrap();
@@ -600,42 +706,93 @@ mod tests {
conn
}
- // ── Helper: build a real signed observer frame ────────────────────────────
-
fn make_observer_frame(owner_keys: &Keys, agent_keys: &Keys, frame_type: &str) -> Event {
let owner_pk = owner_keys.public_key().to_hex();
let agent_pk = agent_keys.public_key().to_hex();
-
- // Minimal NIP-44-looking ciphertext (base64, long enough to pass heuristic).
- let fake_ciphertext = "A".repeat(200);
-
let tags = vec![
Tag::parse(["p", &owner_pk]).unwrap(),
Tag::parse(["agent", &agent_pk]).unwrap(),
Tag::parse(["frame", frame_type]).unwrap(),
];
-
- EventBuilder::new(Kind::Custom(24200), &fake_ciphertext)
+ EventBuilder::new(Kind::Custom(24200), &"A".repeat(200))
.tags(tags)
.sign_with_keys(agent_keys)
.unwrap()
}
- // ── Ephemeral validator — individual condition rejection ──────────────────
-
- fn add_owner_p_sub(conn: &Connection, identity_pk: &str, relay_url: &str, scope_value: &str) {
+ fn add_sub(
+ conn: &Connection,
+ identity_pk: &str,
+ relay_url: &str,
+ scope_type: &str,
+ scope_value: &str,
+ kinds_json: &str,
+ ) {
store::upsert_save_subscription(
conn,
identity_pk,
relay_url,
- "owner_p",
+ scope_type,
scope_value,
- "[24200]",
+ kinds_json,
0,
)
.unwrap();
}
+ /// Run the full archive pipeline synchronously with a fake relay response.
+ ///
+ /// Calls `plan_archive` → injects fake relay events → `commit_archive`.
+ /// This mirrors `archive_events` without the async relay calls.
+ fn run_batch_sync(
+ candidates: Vec,
+ identity_pk: &str,
+ relay_url: &str,
+ conn: &Connection,
+ fake_relay_events: Vec,
+ ) -> ArchiveBatchResult {
+ let plan = plan_archive(candidates, identity_pk, relay_url, conn).unwrap();
+
+ // Synthesize BucketWithResult from the fake relay response.
+ let fake_ids: std::collections::HashSet =
+ fake_relay_events.iter().map(|e| e.id.to_hex()).collect();
+ let bucket_results: Vec = plan
+ .buckets
+ .into_iter()
+ .map(|b| BucketWithResult {
+ scope_type_str: b.scope_type_str,
+ scope_value: b.scope_value,
+ allowed_kinds: b.allowed_kinds,
+ group: b.group,
+ returned_ids: fake_ids.clone(),
+ relay_failed: false,
+ })
+ .collect();
+
+ commit_archive(
+ bucket_results,
+ plan.ephemeral,
+ plan.pre_dropped,
+ identity_pk,
+ relay_url,
+ 0,
+ conn,
+ )
+ .unwrap()
+ }
+
+ fn candidate(event: &Event, scope_type: ScopeType, scope_value: &str) -> ArchiveCandidate {
+ ArchiveCandidate {
+ raw_event_json: event.as_json(),
+ matched_scope: MatchedScope {
+ scope_type,
+ scope_value: scope_value.to_string(),
+ },
+ }
+ }
+
+ // ── Ephemeral validator — individual condition rejection ──────────────────
+
#[test]
fn test_ephemeral_validator_accepts_valid_frame() {
let conn = in_memory();
@@ -643,10 +800,8 @@ mod tests {
let agent_keys = Keys::generate();
let owner_pk = owner_keys.public_key().to_hex();
let relay_url = "wss://relay.example";
-
- add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
+ add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[24200]");
let ev = make_observer_frame(&owner_keys, &agent_keys, OBSERVER_FRAME_TELEMETRY);
-
assert!(
validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url)
.is_ok()
@@ -660,9 +815,7 @@ mod tests {
let agent_keys = Keys::generate();
let owner_pk = owner_keys.public_key().to_hex();
let relay_url = "wss://relay.example";
- add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
-
- // kind 1 instead of 24200
+ add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[24200]");
let ev = EventBuilder::new(Kind::TextNote, "hello")
.tags(vec![
Tag::parse(["p", &owner_pk]).unwrap(),
@@ -671,7 +824,6 @@ mod tests {
])
.sign_with_keys(&agent_keys)
.unwrap();
-
let result =
validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
assert!(result.is_err());
@@ -685,9 +837,7 @@ mod tests {
let agent_keys = Keys::generate();
let owner_pk = owner_keys.public_key().to_hex();
let relay_url = "wss://relay.example";
- add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
-
- // No `p` tag.
+ add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[24200]");
let ev = EventBuilder::new(Kind::Custom(24200), &"A".repeat(200))
.tags(vec![
Tag::parse(["agent", &agent_keys.public_key().to_hex()]).unwrap(),
@@ -695,7 +845,6 @@ mod tests {
])
.sign_with_keys(&agent_keys)
.unwrap();
-
let result =
validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
assert!(result.is_err());
@@ -709,9 +858,7 @@ mod tests {
let agent_keys = Keys::generate();
let owner_pk = owner_keys.public_key().to_hex();
let relay_url = "wss://relay.example";
- add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
-
- // No `agent` tag.
+ add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[24200]");
let ev = EventBuilder::new(Kind::Custom(24200), &"A".repeat(200))
.tags(vec![
Tag::parse(["p", &owner_pk]).unwrap(),
@@ -719,7 +866,6 @@ mod tests {
])
.sign_with_keys(&agent_keys)
.unwrap();
-
let result =
validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
assert!(result.is_err());
@@ -733,9 +879,7 @@ mod tests {
let agent_keys = Keys::generate();
let owner_pk = owner_keys.public_key().to_hex();
let relay_url = "wss://relay.example";
- add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
-
- // frame=control, not telemetry.
+ add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[24200]");
let ev = make_observer_frame(&owner_keys, &agent_keys, "control");
let result =
validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
@@ -751,9 +895,7 @@ mod tests {
let other_keys = Keys::generate();
let owner_pk = owner_keys.public_key().to_hex();
let relay_url = "wss://relay.example";
- add_owner_p_sub(&conn, &owner_pk, relay_url, &owner_pk);
-
- // event signed by `other_keys` but agent tag = agent_keys pubkey.
+ add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[24200]");
let ev = EventBuilder::new(Kind::Custom(24200), &"A".repeat(200))
.tags(vec![
Tag::parse(["p", &owner_pk]).unwrap(),
@@ -762,7 +904,6 @@ mod tests {
])
.sign_with_keys(&other_keys) // wrong signer
.unwrap();
-
let result =
validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
assert!(result.is_err());
@@ -777,7 +918,6 @@ mod tests {
let owner_pk = owner_keys.public_key().to_hex();
let relay_url = "wss://relay.example";
// Deliberately do NOT add a subscription.
-
let ev = make_observer_frame(&owner_keys, &agent_keys, OBSERVER_FRAME_TELEMETRY);
let result =
validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
@@ -785,72 +925,266 @@ mod tests {
assert!(result.unwrap_err().contains("owner_p subscription"));
}
- // ── Scope derivation ─────────────────────────────────────────────────────
+ #[test]
+ fn test_ephemeral_validator_rejects_kind_not_in_subscription() {
+ // Subscription exists but kinds = [1] (not 24200) — must be rejected.
+ let conn = in_memory();
+ let owner_keys = Keys::generate();
+ let agent_keys = Keys::generate();
+ let owner_pk = owner_keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[1]"); // wrong kinds
+ let ev = make_observer_frame(&owner_keys, &agent_keys, OBSERVER_FRAME_TELEMETRY);
+ let result =
+ validate_ephemeral_frame(&ev, &owner_pk, &owner_pk, &conn, &owner_pk, relay_url);
+ assert!(result.is_err());
+ let msg = result.unwrap_err();
+ assert!(
+ msg.contains("24200"),
+ "expected kind 24200 in error, got: {msg}"
+ );
+ }
+
+ // ── archive pipeline — persistent path ───────────────────────────────────
#[test]
- fn test_derive_channel_h_scope_extracts_h_tag() {
+ fn test_persistent_channel_h_persists_when_relay_returns_event() {
+ let conn = in_memory();
let keys = Keys::generate();
- let ev = EventBuilder::new(Kind::TextNote, "hello")
- .tags(vec![Tag::parse(["h", "chan-uuid-123"]).unwrap()])
+ let identity_pk = keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ let chan = "chan-abc";
+ add_sub(&conn, &identity_pk, relay_url, "channel_h", chan, "[9]");
+
+ let ev = EventBuilder::new(Kind::Custom(9), "msg")
+ .tags(vec![Tag::parse(["h", chan]).unwrap()])
.sign_with_keys(&keys)
.unwrap();
- assert_eq!(
- derive_channel_h_scope(&ev),
- Some("chan-uuid-123".to_string())
- );
+ let cands = vec![candidate(&ev, ScopeType::ChannelH, chan)];
+
+ // Fake relay returns the event (simulates relay proof).
+ let result = run_batch_sync(cands, &identity_pk, relay_url, &conn, vec![ev.clone()]);
+ assert_eq!(result.persisted, 1);
+ assert_eq!(result.dropped, 0);
+
+ // Confirm the event is in the store.
+ let count: i64 = conn
+ .query_row("SELECT COUNT(*) FROM archived_events", [], |r| r.get(0))
+ .unwrap();
+ assert_eq!(count, 1);
+ let scope_count: i64 = conn
+ .query_row("SELECT COUNT(*) FROM archived_event_scopes WHERE scope_type = 'channel_h' AND scope_value = ?1", [chan], |r| r.get(0))
+ .unwrap();
+ assert_eq!(scope_count, 1);
}
#[test]
- fn test_derive_channel_h_scope_returns_none_when_absent() {
+ fn test_persistent_channel_h_drops_when_relay_does_not_return_event() {
+ // Relay returns empty — event not proven accessible.
+ let conn = in_memory();
let keys = Keys::generate();
- let ev = EventBuilder::new(Kind::TextNote, "hello")
+ let identity_pk = keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ let chan = "chan-abc";
+ add_sub(&conn, &identity_pk, relay_url, "channel_h", chan, "[9]");
+
+ let ev = EventBuilder::new(Kind::Custom(9), "msg")
+ .tags(vec![Tag::parse(["h", chan]).unwrap()])
+ .sign_with_keys(&keys)
+ .unwrap();
+ let cands = vec![candidate(&ev, ScopeType::ChannelH, chan)];
+
+ // Fake relay returns nothing.
+ let result = run_batch_sync(cands, &identity_pk, relay_url, &conn, vec![]);
+ assert_eq!(result.persisted, 0);
+ assert_eq!(result.dropped, 1);
+ }
+
+ #[test]
+ fn test_persistent_drops_when_no_subscription() {
+ // No subscription at all — drop before even querying.
+ let conn = in_memory();
+ let keys = Keys::generate();
+ let identity_pk = keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ let chan = "chan-abc";
+ // Intentionally no subscription.
+
+ let ev = EventBuilder::new(Kind::Custom(9), "msg")
+ .tags(vec![Tag::parse(["h", chan]).unwrap()])
.sign_with_keys(&keys)
.unwrap();
- assert_eq!(derive_channel_h_scope(&ev), None);
+ let cands = vec![candidate(&ev, ScopeType::ChannelH, chan)];
+
+ // Fake relay would return the event, but no sub → dropped.
+ let result = run_batch_sync(cands, &identity_pk, relay_url, &conn, vec![ev.clone()]);
+ assert_eq!(result.persisted, 0);
+ assert_eq!(result.dropped, 1);
}
#[test]
- fn test_derive_referenced_e_scope_matches_claimed() {
+ fn test_persistent_drops_kind_not_in_subscription() {
+ // Subscription is for kind 9 only; event is kind 7 (reaction).
+ let conn = in_memory();
let keys = Keys::generate();
+ let identity_pk = keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ let chan = "chan-abc";
+ add_sub(&conn, &identity_pk, relay_url, "channel_h", chan, "[9]");
+
+ // kind 7 reaction — no `h` tag naturally, but relay-returned under scoped filter
+ let ev = EventBuilder::new(Kind::Reaction, "+")
+ .tags(vec![Tag::parse(["h", chan]).unwrap()])
+ .sign_with_keys(&keys)
+ .unwrap();
+ let cands = vec![candidate(&ev, ScopeType::ChannelH, chan)];
+
+ // Fake relay returns the event (simulates relay proof via StoredEvent.channel_id),
+ // but kind 7 is not in the subscription's kinds list.
+ let result = run_batch_sync(cands, &identity_pk, relay_url, &conn, vec![ev.clone()]);
+ assert_eq!(result.persisted, 0);
+ assert_eq!(result.dropped, 1);
+ }
+
+ #[test]
+ fn test_persistent_h_less_event_persists_when_relay_returns_it() {
+ // An h-less event (e.g. reaction kind:7) that the relay returns under
+ // the scoped #h filter (via StoredEvent.channel_id fallback) must be
+ // persisted. The local tag scanner would have dropped it; the scoped
+ // filter proof must not.
+ let conn = in_memory();
+ let keys = Keys::generate();
+ let identity_pk = keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ let chan = "chan-abc";
+ add_sub(&conn, &identity_pk, relay_url, "channel_h", chan, "[7]");
+
+ // Build a reaction without an `h` tag — local re-derivation would drop it.
+ let ev = EventBuilder::new(Kind::Reaction, "+")
+ .sign_with_keys(&keys)
+ .unwrap();
+ let cands = vec![candidate(&ev, ScopeType::ChannelH, chan)];
+
+ // Fake relay returns it (relay used StoredEvent.channel_id to match).
+ let result = run_batch_sync(cands, &identity_pk, relay_url, &conn, vec![ev.clone()]);
+ assert_eq!(result.persisted, 1);
+ assert_eq!(result.dropped, 0);
+
+ // Scope row uses bucket's scope_value, not a local-derived value.
+ let scope_val: String = conn
+ .query_row(
+ "SELECT scope_value FROM archived_event_scopes WHERE scope_type = 'channel_h'",
+ [],
+ |r| r.get(0),
+ )
+ .unwrap();
+ assert_eq!(scope_val, chan);
+ }
+
+ #[test]
+ fn test_persistent_referenced_e_persists_when_relay_returns_event() {
+ let conn = in_memory();
+ let keys = Keys::generate();
+ let identity_pk = keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
let ref_id = "a".repeat(64);
- let ev = EventBuilder::new(Kind::TextNote, "reply")
+ add_sub(
+ &conn,
+ &identity_pk,
+ relay_url,
+ "referenced_e",
+ &ref_id,
+ "[9]",
+ );
+
+ let ev = EventBuilder::new(Kind::Custom(9), "reply")
.tags(vec![Tag::parse(["e", &ref_id]).unwrap()])
.sign_with_keys(&keys)
.unwrap();
- assert_eq!(
- derive_referenced_e_scope(&ev, &ref_id),
- Some(ref_id.clone())
- );
+ let cands = vec![candidate(&ev, ScopeType::ReferencedE, &ref_id)];
+ let result = run_batch_sync(cands, &identity_pk, relay_url, &conn, vec![ev.clone()]);
+ assert_eq!(result.persisted, 1);
+ assert_eq!(result.dropped, 0);
}
#[test]
- fn test_derive_referenced_e_scope_rejects_wrong_claimed() {
+ fn test_mixed_batch_persisted_and_dropped_counted_exactly() {
+ // Two channel_h candidates: relay only returns one. dropped must be 1.
+ let conn = in_memory();
let keys = Keys::generate();
- let actual_ref = "a".repeat(64);
- let claimed = "b".repeat(64);
- let ev = EventBuilder::new(Kind::TextNote, "reply")
- .tags(vec![Tag::parse(["e", &actual_ref]).unwrap()])
+ let identity_pk = keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ let chan = "chan-abc";
+ add_sub(&conn, &identity_pk, relay_url, "channel_h", chan, "[9]");
+
+ let ev1 = EventBuilder::new(Kind::Custom(9), "msg1")
+ .tags(vec![Tag::parse(["h", chan]).unwrap()])
+ .sign_with_keys(&keys)
+ .unwrap();
+ let ev2 = EventBuilder::new(Kind::Custom(9), "msg2")
+ .tags(vec![Tag::parse(["h", chan]).unwrap()])
.sign_with_keys(&keys)
.unwrap();
- assert_eq!(derive_referenced_e_scope(&ev, &claimed), None);
+ let cands = vec![
+ candidate(&ev1, ScopeType::ChannelH, chan),
+ candidate(&ev2, ScopeType::ChannelH, chan),
+ ];
+
+ // Fake relay only returns ev1.
+ let result = run_batch_sync(cands, &identity_pk, relay_url, &conn, vec![ev1.clone()]);
+ assert_eq!(result.persisted, 1);
+ assert_eq!(result.dropped, 1);
+ }
+
+ // ── archive pipeline — ephemeral path ────────────────────────────────────
+
+ #[test]
+ fn test_ephemeral_path_persists_valid_frame() {
+ let conn = in_memory();
+ let owner_keys = Keys::generate();
+ let agent_keys = Keys::generate();
+ let owner_pk = owner_keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[24200]");
+
+ let ev = make_observer_frame(&owner_keys, &agent_keys, OBSERVER_FRAME_TELEMETRY);
+ let cands = vec![candidate(&ev, ScopeType::OwnerP, &owner_pk)];
+
+ // Fake relay returns nothing (not consulted for ephemeral path).
+ let result = run_batch_sync(cands, &owner_pk, relay_url, &conn, vec![]);
+ assert_eq!(result.persisted, 1);
+ assert_eq!(result.dropped, 0);
+ }
+
+ #[test]
+ fn test_ephemeral_path_drops_kind_not_in_subscription() {
+ let conn = in_memory();
+ let owner_keys = Keys::generate();
+ let agent_keys = Keys::generate();
+ let owner_pk = owner_keys.public_key().to_hex();
+ let relay_url = "wss://relay.example";
+ // kinds = [1], not [24200]
+ add_sub(&conn, &owner_pk, relay_url, "owner_p", &owner_pk, "[1]");
+
+ let ev = make_observer_frame(&owner_keys, &agent_keys, OBSERVER_FRAME_TELEMETRY);
+ let cands = vec![candidate(&ev, ScopeType::OwnerP, &owner_pk)];
+
+ let result = run_batch_sync(cands, &owner_pk, relay_url, &conn, vec![]);
+ assert_eq!(result.persisted, 0);
+ assert_eq!(result.dropped, 1);
}
- // ── Dropped-vs-persisted accounting ─────────────────────────────────────
+ // ── Invalid input dropped ─────────────────────────────────────────────────
#[test]
- fn test_dropped_counting_invalid_json() {
- // archive_events is async and needs AppState — we test the lower-level
- // accounting by confirming that Event::from_json fails gracefully for
- // malformed input and increments the dropped counter.
+ fn test_malformed_json_is_dropped() {
let result = Event::from_json("not json at all");
assert!(result.is_err());
}
#[test]
- fn test_dropped_counting_bad_signature() {
+ fn test_tampered_event_fails_verify_id() {
let keys = Keys::generate();
- // Build a valid event then tamper with the content to break the id
- // (the id is a hash of the event fields including content).
let mut ev_json: serde_json::Value = serde_json::from_str(
&EventBuilder::new(Kind::TextNote, "ok")
.sign_with_keys(&keys)
@@ -861,8 +1195,6 @@ mod tests {
ev_json["content"] = serde_json::Value::String("tampered".into());
let tampered = ev_json.to_string();
let ev = Event::from_json(&tampered).unwrap();
- // After tampering the content the event id (a hash over all fields)
- // no longer matches, so verify_id() returns false.
assert!(!ev.verify_id());
}
}
diff --git a/desktop/src-tauri/src/archive/store.rs b/desktop/src-tauri/src/archive/store.rs
index c2676323a..05ea395ae 100644
--- a/desktop/src-tauri/src/archive/store.rs
+++ b/desktop/src-tauri/src/archive/store.rs
@@ -10,7 +10,7 @@
use std::path::Path;
-use rusqlite::{params, Connection};
+use rusqlite::{params, Connection, OptionalExtension};
// ── Schema ─────────────────────────────────────────────────────────────────
@@ -194,6 +194,30 @@ pub fn has_save_subscription(
Ok(count > 0)
}
+/// Return the `kinds` JSON string for a matching save subscription, or `None`
+/// if no subscription exists.
+pub fn get_subscription_kinds(
+ conn: &Connection,
+ identity_pubkey: &str,
+ relay_url: &str,
+ scope_type: &str,
+ scope_value: &str,
+) -> Result
- {sub.scopeType === "owner_p"
- ? "owner_p"
- : sub.scopeType}{" "}
- · kinds: {kindSummary(sub.kinds)}
+ {sub.scopeType} · kinds: {kindSummary(sub.kinds)}