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/// Backfill a wrapper timestamp onto an EXISTING ledger row (pre-migration-17 rows hold 0).
37/// UPDATE-only by design: the ledger is the negentropy fingerprint set, and message wrappers
38/// may be deferred (batch-buffered) — an INSERT here could ledger a message before its row
39/// lands, marking it "have" forever. Inserting belongs to the save paths, never this backfill.
40pub fn update_wrapper_timestamp(wrapper_id_bytes: &[u8; 32], wrapper_created_at: u64) -> Result<(), String> {
41    let conn = super::get_write_connection_guard_static()?;
42    conn.execute(
43        "UPDATE processed_wrappers SET wrapper_created_at = ?2 \
44         WHERE wrapper_id = ?1 AND wrapper_created_at = 0",
45        rusqlite::params![&wrapper_id_bytes[..], wrapper_created_at as i64],
46    ).map_err(|e| format!("Failed to backfill wrapper timestamp: {}", e))?;
47    Ok(())
48}
49
50/// Load all processed wrapper IDs as raw bytes for the dedup cache.
51pub fn load_processed_wrappers() -> Result<Vec<[u8; 32]>, String> {
52    let conn = match super::get_db_connection_guard_static() {
53        Ok(c) => c,
54        Err(_) => return Ok(Vec::new()),
55    };
56    // NIP-17 only: this feeds the WRAPPER_ID_CACHE, the DM gift-wrap dedup. Concord uses the
57    // synchronous ledger check (processed_wrapper_exists), not this in-memory cache.
58    let mut stmt = conn.prepare("SELECT wrapper_id FROM processed_wrappers WHERE transport = 0")
59        .map_err(|e| format!("Failed to prepare processed_wrappers query: {}", e))?;
60    let rows = stmt.query_map([], |row| {
61        let blob: Vec<u8> = row.get(0)?;
62        if blob.len() == 32 {
63            let mut arr = [0u8; 32];
64            arr.copy_from_slice(&blob);
65            Ok(arr)
66        } else {
67            Err(rusqlite::Error::InvalidParameterCount(blob.len(), 32))
68        }
69    }).map_err(|e| format!("Failed to query processed_wrappers: {}", e))?;
70
71    Ok(rows.flatten().collect())
72}
73
74/// [`load_processed_wrappers`] bounded to `wrapper_created_at >= since_secs` —
75/// the dedup cache only needs the window the planned reconciles can touch;
76/// anything older that still arrives dedups through the DB fallback.
77pub fn load_processed_wrappers_since(since_secs: u64) -> Result<Vec<[u8; 32]>, String> {
78    let conn = match super::get_db_connection_guard_static() {
79        Ok(c) => c,
80        Err(_) => return Ok(Vec::new()),
81    };
82    let mut stmt = conn.prepare(
83        "SELECT wrapper_id FROM processed_wrappers WHERE transport = 0 AND wrapper_created_at >= ?1",
84    ).map_err(|e| format!("Failed to prepare processed_wrappers query: {}", e))?;
85    let rows = stmt.query_map(rusqlite::params![since_secs as i64], |row| {
86        let blob: Vec<u8> = row.get(0)?;
87        if blob.len() == 32 {
88            let mut arr = [0u8; 32];
89            arr.copy_from_slice(&blob);
90            Ok(arr)
91        } else {
92            Err(rusqlite::Error::InvalidParameterCount(blob.len(), 32))
93        }
94    }).map_err(|e| format!("Failed to query processed_wrappers: {}", e))?;
95
96    Ok(rows.flatten().collect())
97}
98
99/// Load recent wrapper IDs from events table (last N days) as raw bytes.
100pub fn load_recent_wrapper_ids(days: u64) -> Result<Vec<[u8; 32]>, String> {
101    let conn = match super::get_db_connection_guard_static() {
102        Ok(c) => c,
103        Err(_) => return Ok(Vec::new()),
104    };
105
106    let cutoff_secs = std::time::SystemTime::now()
107        .duration_since(std::time::UNIX_EPOCH).unwrap()
108        .as_secs()
109        .saturating_sub(days * 24 * 60 * 60);
110
111    // DM cache only: exclude Community chats (chat_type 2). Concord stamps its OUTER id on
112    // `events.wrapper_event_id` too (atomic message dedup), so without this join those ids would warm
113    // the DM gift-wrap cache — harmless (they'd never match a gift-wrap lookup) but wasteful. Concord
114    // dedup uses the synchronous `processed_wrapper_exists` ledger, not this cache.
115    let mut stmt = conn.prepare(
116        "SELECT e.wrapper_event_id FROM events e \
117         JOIN chats c ON e.chat_id = c.id \
118         WHERE e.wrapper_event_id IS NOT NULL AND e.wrapper_event_id != '' \
119         AND e.created_at >= ?1 AND c.chat_type != 2"
120    ).map_err(|e| format!("Failed to prepare wrapper_id query: {}", e))?;
121
122    let hex_ids: Vec<String> = stmt.query_map(rusqlite::params![cutoff_secs as i64], |row| {
123        row.get::<_, String>(0)
124    }).map_err(|e| format!("Failed to query wrapper_ids: {}", e))?
125    .flatten().collect();
126
127    let mut result = Vec::with_capacity(hex_ids.len());
128    for hex in hex_ids {
129        if hex.len() == 64 {
130            result.push(crate::simd::hex::hex_to_bytes_32(&hex));
131        }
132    }
133    Ok(result)
134}
135
136/// Load all processed wrappers as (EventId, Timestamp) pairs for negentropy (NIP-77).
137pub fn load_negentropy_items() -> Result<Vec<(EventId, Timestamp)>, String> {
138    let conn = super::get_db_connection_guard_static()
139        .map_err(|_| "No DB connection".to_string())?;
140
141    // NIP-77 reconciles gift-wraps for our pubkey, so fingerprint ONLY the 'nip17' carrier.
142    // Concord outer events share the ledger for dedup but must never enter DM negentropy.
143    let mut stmt = conn.prepare(
144        "SELECT wrapper_id, wrapper_created_at FROM processed_wrappers WHERE transport = 0"
145    ).map_err(|e| format!("Failed to prepare negentropy query: {}", e))?;
146
147    let items: Vec<_> = stmt.query_map([], |row| {
148        let blob: Vec<u8> = row.get(0)?;
149        let created_at: i64 = row.get(1)?;
150        Ok((blob, created_at))
151    }).map_err(|e| format!("Failed to query processed_wrappers: {}", e))?
152    .flatten()
153    .filter_map(|(blob, ts)| {
154        if blob.len() == 32 {
155            let mut arr = [0u8; 32];
156            arr.copy_from_slice(&blob);
157            Some((
158                EventId::from_byte_array(arr),
159                Timestamp::from_secs(ts as u64),
160            ))
161        } else {
162            None
163        }
164    })
165    .collect();
166
167    Ok(items)
168}
vector_core/db/wrappers.rs

vector_core/db/
wrappers.rs
