vector_core/db/
wrappers.rs

1//! Wrapper tracking — NIP-59 gift wrap dedup + NIP-77 negentropy.
2
3use nostr_sdk::prelude::{EventId, Timestamp};
4
5/// Transport carriers for the shared outer-event ledger — stored as a small INTEGER discriminator
6/// (cheaper than a per-row string, and the ledger can grow large). Never renumber an existing value.
7pub const TRANSPORT_NIP17: i64 = 0;
8pub const TRANSPORT_CONCORD: i64 = 1;
9
10/// Persist an outer-event id for cross-session dedup (INSERT OR IGNORE), tagged by `transport`
11/// so the ledger is shared across transports while negentropy stays NIP-17-scoped.
12pub fn save_processed_wrapper(wrapper_id_bytes: &[u8; 32], wrapper_created_at: u64, transport: i64) -> Result<(), String> {
13    let conn = super::get_write_connection_guard_static()?;
14    conn.execute(
15        "INSERT OR IGNORE INTO processed_wrappers (wrapper_id, wrapper_created_at, transport) VALUES (?1, ?2, ?3)",
16        rusqlite::params![&wrapper_id_bytes[..], wrapper_created_at as i64, transport],
17    ).map_err(|e| format!("Failed to save processed wrapper: {}", e))?;
18    Ok(())
19}
20
21/// Sync existence check against the ledger (any transport) — the DB half of the outer-event dedup
22/// for callers that can't reach the async `WRAPPER_ID_CACHE` (e.g. the synchronous Concord ingest).
23/// Returns false on a missing/closed DB so a dedup failure never drops a genuinely-new event.
24pub fn processed_wrapper_exists(wrapper_id_bytes: &[u8; 32]) -> bool {
25    let conn = match super::get_db_connection_guard_static() {
26        Ok(c) => c,
27        Err(_) => return false,
28    };
29    conn.query_row(
30        "SELECT EXISTS(SELECT 1 FROM processed_wrappers WHERE wrapper_id = ?1)",
31        rusqlite::params![&wrapper_id_bytes[..]],
32        |row| row.get(0),
33    ).unwrap_or(false)
34}
35
36/// Upsert a wrapper timestamp (backfill for pre-migration-17 wrappers).
37pub fn update_wrapper_timestamp(wrapper_id_bytes: &[u8; 32], wrapper_created_at: u64) -> Result<(), String> {
38    let conn = super::get_write_connection_guard_static()?;
39    conn.execute(
40        "INSERT INTO processed_wrappers (wrapper_id, wrapper_created_at) VALUES (?1, ?2) \
41         ON CONFLICT(wrapper_id) DO UPDATE SET wrapper_created_at = ?2 WHERE wrapper_created_at = 0",
42        rusqlite::params![&wrapper_id_bytes[..], wrapper_created_at as i64],
43    ).map_err(|e| format!("Failed to upsert wrapper timestamp: {}", e))?;
44    Ok(())
45}
46
47/// Load all processed wrapper IDs as raw bytes for the dedup cache.
48pub fn load_processed_wrappers() -> Result<Vec<[u8; 32]>, String> {
49    let conn = match super::get_db_connection_guard_static() {
50        Ok(c) => c,
51        Err(_) => return Ok(Vec::new()),
52    };
53    // NIP-17 only: this feeds the WRAPPER_ID_CACHE, the DM gift-wrap dedup. Concord uses the
54    // synchronous ledger check (processed_wrapper_exists), not this in-memory cache.
55    let mut stmt = conn.prepare("SELECT wrapper_id FROM processed_wrappers WHERE transport = 0")
56        .map_err(|e| format!("Failed to prepare processed_wrappers query: {}", e))?;
57    let rows = stmt.query_map([], |row| {
58        let blob: Vec<u8> = row.get(0)?;
59        if blob.len() == 32 {
60            let mut arr = [0u8; 32];
61            arr.copy_from_slice(&blob);
62            Ok(arr)
63        } else {
64            Err(rusqlite::Error::InvalidParameterCount(blob.len(), 32))
65        }
66    }).map_err(|e| format!("Failed to query processed_wrappers: {}", e))?;
67
68    Ok(rows.flatten().collect())
69}
70
71/// Load recent wrapper IDs from events table (last N days) as raw bytes.
72pub fn load_recent_wrapper_ids(days: u64) -> Result<Vec<[u8; 32]>, String> {
73    let conn = match super::get_db_connection_guard_static() {
74        Ok(c) => c,
75        Err(_) => return Ok(Vec::new()),
76    };
77
78    let cutoff_secs = std::time::SystemTime::now()
79        .duration_since(std::time::UNIX_EPOCH).unwrap()
80        .as_secs()
81        .saturating_sub(days * 24 * 60 * 60);
82
83    // DM cache only: exclude Community chats (chat_type 2). Concord stamps its OUTER id on
84    // `events.wrapper_event_id` too (atomic message dedup), so without this join those ids would warm
85    // the DM gift-wrap cache — harmless (they'd never match a gift-wrap lookup) but wasteful. Concord
86    // dedup uses the synchronous `processed_wrapper_exists` ledger, not this cache.
87    let mut stmt = conn.prepare(
88        "SELECT e.wrapper_event_id FROM events e \
89         JOIN chats c ON e.chat_id = c.id \
90         WHERE e.wrapper_event_id IS NOT NULL AND e.wrapper_event_id != '' \
91         AND e.created_at >= ?1 AND c.chat_type != 2"
92    ).map_err(|e| format!("Failed to prepare wrapper_id query: {}", e))?;
93
94    let hex_ids: Vec<String> = stmt.query_map(rusqlite::params![cutoff_secs as i64], |row| {
95        row.get::<_, String>(0)
96    }).map_err(|e| format!("Failed to query wrapper_ids: {}", e))?
97    .flatten().collect();
98
99    let mut result = Vec::with_capacity(hex_ids.len());
100    for hex in hex_ids {
101        if hex.len() == 64 {
102            result.push(crate::simd::hex::hex_to_bytes_32(&hex));
103        }
104    }
105    Ok(result)
106}
107
108/// Load all processed wrappers as (EventId, Timestamp) pairs for negentropy (NIP-77).
109pub fn load_negentropy_items() -> Result<Vec<(EventId, Timestamp)>, String> {
110    let conn = super::get_db_connection_guard_static()
111        .map_err(|_| "No DB connection".to_string())?;
112
113    // NIP-77 reconciles gift-wraps for our pubkey, so fingerprint ONLY the 'nip17' carrier.
114    // Concord outer events share the ledger for dedup but must never enter DM negentropy.
115    let mut stmt = conn.prepare(
116        "SELECT wrapper_id, wrapper_created_at FROM processed_wrappers WHERE transport = 0"
117    ).map_err(|e| format!("Failed to prepare negentropy query: {}", e))?;
118
119    let items: Vec<_> = stmt.query_map([], |row| {
120        let blob: Vec<u8> = row.get(0)?;
121        let created_at: i64 = row.get(1)?;
122        Ok((blob, created_at))
123    }).map_err(|e| format!("Failed to query processed_wrappers: {}", e))?
124    .flatten()
125    .filter_map(|(blob, ts)| {
126        if blob.len() == 32 {
127            let mut arr = [0u8; 32];
128            arr.copy_from_slice(&blob);
129            Some((
130                EventId::from_byte_array(arr),
131                Timestamp::from_secs(ts as u64),
132            ))
133        } else {
134            None
135        }
136    })
137    .collect();
138
139    Ok(items)
140}
vector_core/db/wrappers.rs

vector_core/db/
wrappers.rs
