1use std::collections::HashMap;
9
10use crate::types::Attachment;
11
12const SELECT_COLS: &str = "event_id, att_index, hash, key, nonce, extension, name, url, \
13 path, size, img_meta, downloaded, webxdc_topic, group_id, original_hash, fallback_urls";
14
15fn row_to_attachment(row: &rusqlite::Row) -> rusqlite::Result<(String, Attachment)> {
18 let event_id: String = row.get(0)?;
19 let img_meta_json: Option<String> = row.get(10)?;
20 let att = Attachment {
21 id: row.get(2)?,
22 key: row.get(3)?,
23 nonce: row.get(4)?,
24 extension: row.get(5)?,
25 name: row.get(6)?,
26 url: row.get(7)?,
27 path: row.get(8)?,
28 size: row.get::<_, i64>(9)? as u64,
29 img_meta: img_meta_json.and_then(|j| serde_json::from_str(&j).ok()),
30 downloading: false,
31 downloaded: row.get::<_, i64>(11)? != 0,
32 webxdc_topic: row.get(12)?,
33 group_id: row.get(13)?,
34 original_hash: row.get(14)?,
35 fallback_urls: row
36 .get::<_, String>(15)?
37 .split_whitespace()
38 .map(|s| s.to_string())
39 .collect(),
40 };
41 Ok((event_id, att))
42}
43
44pub fn insert_attachment_rows(conn: &rusqlite::Connection, event_id: &str, attachments: &[Attachment]) -> Result<(), String> {
54 if attachments.is_empty() {
55 return Ok(());
56 }
57 let mut stmt = conn.prepare_cached(
60 "INSERT INTO attachments (event_id, att_index, hash, key, nonce, extension, name, url, \
61 path, size, img_meta, downloaded, webxdc_topic, group_id, original_hash, fallback_urls) \
62 VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16) \
63 ON CONFLICT(event_id, att_index) DO UPDATE SET \
64 key=excluded.key, nonce=excluded.nonce, extension=excluded.extension, \
65 name=excluded.name, url=excluded.url, size=excluded.size, img_meta=excluded.img_meta, \
66 webxdc_topic=excluded.webxdc_topic, group_id=excluded.group_id, \
67 original_hash=excluded.original_hash, \
68 fallback_urls=CASE WHEN excluded.fallback_urls='' THEN fallback_urls ELSE excluded.fallback_urls END, \
69 downloaded=MAX(downloaded, excluded.downloaded), \
70 hash=CASE WHEN excluded.downloaded=1 THEN excluded.hash ELSE hash END, \
71 path=CASE WHEN excluded.downloaded=1 THEN excluded.path ELSE path END",
72 ).map_err(|e| format!("prepare insert attachment: {e}"))?;
73 for (i, a) in attachments.iter().enumerate() {
74 let img_meta_json = a.img_meta.as_ref().and_then(|m| serde_json::to_string(m).ok());
75 stmt.execute(
76 rusqlite::params![
77 event_id, i as i64, a.id, a.key, a.nonce, a.extension, a.name, a.url,
78 a.path, a.size as i64, img_meta_json, a.downloaded as i64,
79 a.webxdc_topic, a.group_id, a.original_hash,
80 a.fallback_urls.join(" "),
81 ],
82 ).map_err(|e| format!("insert attachment: {e}"))?;
83 }
84 Ok(())
85}
86
87pub fn get_attachments_for_events(event_ids: &[String]) -> Result<HashMap<String, Vec<Attachment>>, String> {
90 let mut out: HashMap<String, Vec<Attachment>> = HashMap::new();
91 if event_ids.is_empty() {
92 return Ok(out);
93 }
94 let conn = super::get_db_connection_guard_static()?;
95 let sql = format!(
96 "SELECT {SELECT_COLS} FROM attachments WHERE event_id IN (SELECT value FROM json_each(?1)) ORDER BY event_id, att_index"
97 );
98 let mut stmt = conn.prepare_cached(&sql).map_err(|e| format!("prepare get_attachments: {e}"))?;
99 let rows = stmt.query_map([super::events::id_list_param(event_ids)], row_to_attachment)
100 .map_err(|e| format!("query get_attachments: {e}"))?;
101 for r in rows.flatten() {
102 out.entry(r.0).or_default().push(r.1);
103 }
104 Ok(out)
105}
106
107pub fn get_attachments_for_event(event_id: &str) -> Result<Vec<Attachment>, String> {
109 let map = get_attachments_for_events(std::slice::from_ref(&event_id.to_string()))?;
110 Ok(map.into_values().next().unwrap_or_default())
111}
112
113pub fn find_by_webxdc_topic_in(chat_identifier: &str, topic: &str) -> Result<Option<(String, Attachment)>, String> {
117 let Ok(chat) = super::id_cache::get_chat_id_by_identifier(chat_identifier) else { return Ok(None) };
118 let conn = super::get_db_connection_guard_static()?;
119 let cols = SELECT_COLS.split(", ").map(|c| format!("a.{}", c.trim())).collect::<Vec<_>>().join(", ");
120 let mut stmt = conn
121 .prepare(&format!(
122 "SELECT {cols} FROM attachments a JOIN events e ON e.id = a.event_id \
123 WHERE a.webxdc_topic = ?1 AND e.chat_id = ?2 ORDER BY e.created_at ASC LIMIT 1"
124 ))
125 .map_err(|e| format!("prepare topic lookup: {e}"))?;
126 let mut rows = stmt.query(rusqlite::params![topic, chat]).map_err(|e| format!("topic lookup: {e}"))?;
127 match rows.next().map_err(|e| format!("topic lookup: {e}"))? {
128 Some(row) => row_to_attachment(row).map(Some).map_err(|e| format!("topic row: {e}")),
129 None => Ok(None),
130 }
131}
132
133pub fn events_with_attachment_hash(hash: &str) -> Result<Vec<String>, String> {
140 let conn = super::get_db_connection_guard_static()?;
141 let mut stmt = conn
142 .prepare(
143 "SELECT a.event_id FROM attachments a JOIN events e ON e.id = a.event_id \
147 WHERE a.hash = ?1 ORDER BY e.created_at DESC, e.received_at DESC, e.rowid DESC",
148 )
149 .map_err(|e| format!("prepare events_with_attachment_hash: {e}"))?;
150 let rows = stmt
151 .query_map(rusqlite::params![hash], |r| r.get::<_, String>(0))
152 .map_err(|e| format!("query events_with_attachment_hash: {e}"))?;
153 Ok(rows.flatten().collect())
154}
155
156pub fn set_attachment_downloaded(event_id: &str, hash: &str, downloaded: bool, path: &str) -> Result<(), String> {
159 let conn = super::get_write_connection_guard_static()?;
160 conn.execute(
161 "UPDATE attachments SET downloaded=?1, path=?2 WHERE event_id=?3 AND hash=?4",
162 rusqlite::params![downloaded as i64, path, event_id, hash],
163 ).map_err(|e| format!("set_attachment_downloaded: {e}"))?;
164 Ok(())
165}
166
167pub fn backfill_downloaded_by_hash(hash: &str, path: &str, exclude_event_id: &str) -> Result<Vec<String>, String> {
171 let conn = super::get_write_connection_guard_static()?;
172 let affected: Vec<String> = {
173 let mut stmt = conn.prepare(
174 "SELECT DISTINCT event_id FROM attachments WHERE hash=?1 AND event_id!=?2 AND downloaded=0"
175 ).map_err(|e| format!("prepare backfill_by_hash: {e}"))?;
176 let rows = stmt.query_map(rusqlite::params![hash, exclude_event_id], |r| r.get::<_, String>(0))
177 .map_err(|e| format!("query backfill_by_hash: {e}"))?;
178 rows.flatten().collect()
179 };
180 conn.execute(
181 "UPDATE attachments SET downloaded=1, path=?1 WHERE hash=?2 AND event_id!=?3 AND downloaded=0",
182 rusqlite::params![path, hash, exclude_event_id],
183 ).map_err(|e| format!("backfill_by_hash update: {e}"))?;
184 Ok(affected)
185}
186
187pub fn downloaded_attachment_paths() -> Result<Vec<(String, String, String)>, String> {
191 let conn = super::get_db_connection_guard_static()?;
192 let mut stmt = conn.prepare(
193 "SELECT event_id, hash, path FROM attachments WHERE downloaded=1 AND path!=''"
194 ).map_err(|e| format!("prepare downloaded_paths: {e}"))?;
195 let rows = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?, r.get::<_, String>(2)?)))
196 .map_err(|e| format!("query downloaded_paths: {e}"))?;
197 Ok(rows.flatten().collect())
198}
199
200pub fn clear_attachment_download(event_id: &str, hash: &str) -> Result<(), String> {
202 let conn = super::get_write_connection_guard_static()?;
203 conn.execute(
204 "UPDATE attachments SET downloaded=0, path='' WHERE event_id=?1 AND hash=?2",
205 rusqlite::params![event_id, hash],
206 ).map_err(|e| format!("clear_attachment_download: {e}"))?;
207 Ok(())
208}
209
210pub fn rewrite_downloaded_paths(old_prefixes: &[String], new_dir: &std::path::Path) -> Result<Vec<String>, String> {
214 let conn = super::get_write_connection_guard_static()?;
215 let rows: Vec<(i64, String, String)> = {
216 let mut stmt = conn.prepare("SELECT id, event_id, path FROM attachments WHERE downloaded=1 AND path!=''")
217 .map_err(|e| format!("prepare rewrite_paths: {e}"))?;
218 let mapped = stmt.query_map([], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, String>(1)?, r.get::<_, String>(2)?)))
219 .map_err(|e| format!("query rewrite_paths: {e}"))?;
220 mapped.flatten().collect()
221 };
222 let mut affected = Vec::new();
223 for (rowid, event_id, path) in rows {
224 if !old_prefixes.iter().any(|p| path.starts_with(p.as_str())) {
225 continue;
226 }
227 let Some(name) = std::path::Path::new(&path).file_name() else { continue };
228 let new_path = new_dir.join(name);
229 let res = if new_path.exists() {
230 conn.execute("UPDATE attachments SET path=?1 WHERE id=?2",
231 rusqlite::params![new_path.to_string_lossy().to_string(), rowid])
232 } else {
233 conn.execute("UPDATE attachments SET downloaded=0, path='' WHERE id=?1", rusqlite::params![rowid])
234 };
235 if res.is_ok() {
236 affected.push(event_id);
237 }
238 }
239 Ok(affected)
240}
241
242#[derive(Debug, Clone)]
245pub struct ReusableUpload {
246 pub url: String,
247 pub key: String,
248 pub nonce: String,
249 pub size: u64,
250 pub mine: bool,
253}
254
255pub async fn verify_local_copy(att: &crate::types::Attachment) -> Option<String> {
269 let expected = att.id.clone();
270 let mut candidates: Vec<std::path::PathBuf> = downloaded_paths_by_hash(&expected)
271 .unwrap_or_default()
272 .into_iter()
273 .map(Into::into)
274 .collect();
275 if !att.name.is_empty() {
276 candidates.push(crate::db::get_download_dir().join(&att.name));
277 }
278 let size = att.size;
279 #[cfg(target_arch = "wasm32")]
280 {
281 for p in candidates {
282 let Some(len) = crate::webfiles::size(&p).await else { continue };
283 if size != 0 && len != size && len + 16 != size {
284 continue;
285 }
286 if let Ok(bytes) = crate::webfiles::read(&p).await {
287 if crate::crypto::sha256_hex(&bytes) == expected {
288 return Some(p.to_string_lossy().to_string());
289 }
290 }
291 }
292 None
293 }
294 #[cfg(not(target_arch = "wasm32"))]
295 {
296 candidates.retain(|p| match std::fs::metadata(p) {
297 Ok(m) => size == 0 || m.len() == size || m.len() + 16 == size,
298 Err(_) => false,
299 });
300 if candidates.is_empty() {
301 return None;
302 }
303 crate::rt::spawn_blocking(move || {
304 candidates.into_iter().find_map(|p| {
305 let bytes = std::fs::read(&p).ok()?;
306 (crate::crypto::sha256_hex(&bytes) == expected)
307 .then(|| p.to_string_lossy().to_string())
308 })
309 })
310 .await
311 .ok()
312 .flatten()
313 }
314}
315
316pub async fn verify_message_attachments(msg: &mut crate::types::Message, chat_hint: Option<&str>) {
324 let mut converted: Vec<(String, String)> = Vec::new();
325 for att in &mut msg.attachments {
326 if !att.downloaded {
327 if let Some(path) = verify_local_copy(att).await {
328 att.downloaded = true;
329 att.path = path.clone();
330 converted.push((att.id.clone(), path));
331 }
332 }
333 }
334 if let Some(chat) = chat_hint {
335 if !converted.is_empty() {
336 let mut state = crate::state::STATE.lock().await;
337 for (att_id, path) in converted {
338 state.update_attachment(chat, &msg.id, &att_id, |a| {
339 a.set_downloaded(true);
340 a.path = path.into_boxed_str();
341 });
342 }
343 }
344 }
345}
346
347pub fn downloaded_paths_by_hash(hash: &str) -> Result<Vec<String>, String> {
352 let conn = crate::db::get_db_connection_guard_static()?;
353 let mut stmt = conn
354 .prepare("SELECT DISTINCT path FROM attachments WHERE hash = ?1 AND downloaded = 1 AND path != ''")
355 .map_err(|e| e.to_string())?;
356 let rows = stmt
357 .query_map([hash], |r| r.get::<_, String>(0))
358 .map_err(|e| e.to_string())?
359 .filter_map(|r| r.ok())
360 .collect();
361 Ok(rows)
362}
363
364pub fn find_reusable_by_hash(hash: &str) -> Result<Option<ReusableUpload>, String> {
365 if hash.is_empty() {
366 return Ok(None);
367 }
368 let conn = super::get_db_connection_guard_static()?;
369 let mut stmt = conn
370 .prepare(
371 "SELECT a.url, a.key, a.nonce, a.size, COALESCE(e.mine, 0)
372 FROM attachments a JOIN events e ON e.id = a.event_id
373 WHERE a.hash = ?1 AND a.url <> '' AND a.key <> '' AND a.nonce <> '' AND a.downloaded = 1
374 ORDER BY COALESCE(e.mine, 0) DESC, a.id DESC LIMIT 1",
375 )
376 .map_err(|e| format!("find_reusable_by_hash prepare: {e}"))?;
377 let row = stmt
378 .query_row(rusqlite::params![hash], |r| {
379 Ok(ReusableUpload {
380 url: r.get(0)?,
381 key: r.get(1)?,
382 nonce: r.get(2)?,
383 size: r.get::<_, i64>(3)? as u64,
384 mine: r.get::<_, i64>(4)? != 0,
385 })
386 })
387 .ok();
388 Ok(row)
389}
390
391pub fn urls_unreferenced_elsewhere(urls: &[String], excluding_event_id: &str) -> Result<Vec<String>, String> {
397 if urls.is_empty() {
398 return Ok(Vec::new());
399 }
400 let conn = super::get_db_connection_guard_static()?;
401 let mut stmt = conn
402 .prepare("SELECT 1 FROM attachments WHERE url = ?1 AND event_id <> ?2 LIMIT 1")
403 .map_err(|e| format!("urls_unreferenced prepare: {e}"))?;
404 let mut out = Vec::with_capacity(urls.len());
405 for url in urls {
406 let referenced = stmt.exists(rusqlite::params![url, excluding_event_id])
407 .map_err(|e| format!("urls_unreferenced query: {e}"))?;
408 if !referenced {
409 out.push(url.clone());
410 }
411 }
412 Ok(out)
413}
414
415pub fn hash_referenced_elsewhere(hash: &str, excluding_event_id: &str) -> Result<bool, String> {
419 if hash.is_empty() {
420 return Ok(false);
421 }
422 let conn = super::get_db_connection_guard_static()?;
423 let mut stmt = conn
424 .prepare("SELECT 1 FROM attachments WHERE hash = ?1 AND event_id <> ?2 LIMIT 1")
425 .map_err(|e| format!("hash_referenced prepare: {e}"))?;
426 stmt.exists(rusqlite::params![hash, excluding_event_id])
427 .map_err(|e| format!("hash_referenced query: {e}"))
428}
429
430#[cfg(test)]
431mod tests {
432 use super::*;
433 use crate::types::{Attachment, Message};
434
435 static TEST_COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(74000);
436
437 fn make_test_npub(n: u32) -> String {
438 const BECH32: &[u8] = b"qpzry9x8gf2tvdw0s3jn54khce6mua7l";
439 let mut payload = vec![b'q'; 58];
440 let mut x = n as u64;
441 let mut i = 58;
442 while x > 0 && i > 0 {
443 i -= 1;
444 payload[i] = BECH32[(x as usize) % 32];
445 x /= 32;
446 }
447 format!("npub1{}", std::str::from_utf8(&payload).unwrap())
448 }
449
450 fn init_test_db() -> (tempfile::TempDir, std::sync::MutexGuard<'static, ()>) {
451 let guard = crate::db::DB_TEST_GUARD.lock().unwrap_or_else(|e| e.into_inner());
452 crate::db::close_database();
453 crate::db::clear_id_caches();
454 let tmp = tempfile::tempdir().unwrap();
455 let n = TEST_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
456 let account = make_test_npub(n);
457 std::fs::create_dir_all(tmp.path().join(&account)).unwrap();
458 crate::db::set_app_data_dir(crate::db::shared_test_data_dir().to_path_buf());
459 crate::db::set_current_account(account.clone()).unwrap();
460 crate::db::init_database(&account).unwrap();
461 (tmp, guard)
462 }
463
464 fn att(hash: &str, url: &str, keyed: bool, downloaded: bool) -> Attachment {
465 Attachment {
466 id: hash.to_string(),
467 key: if keyed { "k".into() } else { String::new() },
468 nonce: if keyed { "n".into() } else { String::new() },
469 url: url.to_string(),
470 size: 42,
471 downloaded,
472 ..Default::default()
473 }
474 }
475
476 async fn save(chat: &str, event_id: &str, mine: bool, a: Attachment) {
477 let msg = Message {
478 id: event_id.to_string(),
479 content: String::new(),
480 mine,
481 at: 1_000_000,
482 attachments: vec![a],
483 ..Default::default()
484 };
485 crate::db::events::save_message(chat, &msg).await.unwrap();
486 }
487
488 #[tokio::test]
491 async fn a_topic_resolves_to_its_app_message_only_within_its_chat() {
492 let (_tmp, _guard) = init_test_db();
493 let mut game = att("HG", "https://b.example/g", true, true);
494 game.extension = "xdc".into();
495 game.webxdc_topic = Some("TOPICA".into());
496 save("npub1chatq", "evt_game", false, game).await;
497 save("npub1chatq", "evt_photo", false, att("HP", "https://b.example/p", true, true)).await;
498
499 let (event_id, found) = find_by_webxdc_topic_in("npub1chatq", "TOPICA").unwrap().expect("the topic's message");
500 assert_eq!(event_id, "evt_game");
501 assert_eq!(found.id, "HG");
502 assert!(find_by_webxdc_topic_in("npub1chatq", "TOPICB").unwrap().is_none());
503 let mut copy = att("HG", "https://b.example/g2", true, true);
505 copy.extension = "xdc".into();
506 copy.webxdc_topic = Some("TOPICA".into());
507 save("npub1stranger", "evt_copy", false, copy).await;
508 assert_eq!(find_by_webxdc_topic_in("npub1chatq", "TOPICA").unwrap().unwrap().0, "evt_game");
509 assert_eq!(find_by_webxdc_topic_in("npub1stranger", "TOPICA").unwrap().unwrap().0, "evt_copy");
510 assert!(find_by_webxdc_topic_in("npub1nobody", "TOPICA").unwrap().is_none());
511 }
512
513 #[tokio::test]
517 async fn the_ledger_finds_a_prior_upload_and_prefers_our_own() {
518 let (_tmp, _guard) = init_test_db();
519 save("chat_a", "evt_foreign", false, att("H1", "https://b.example/foreign", true, true)).await;
520 save("chat_b", "evt_mine", true, att("H1", "https://b.example/mine", true, true)).await;
521 save("chat_c", "evt_unverified", false, att("H2", "https://b.example/x", true, false)).await;
522 save("chat_d", "evt_keyless", true, att("H3", "https://b.example/y", false, true)).await;
523
524 let hit = find_reusable_by_hash("H1").unwrap().expect("H1 is reusable");
525 assert_eq!(hit.url, "https://b.example/mine", "own upload preferred");
526 assert!(hit.mine);
527 assert_eq!(hit.key, "k");
528 assert_eq!(hit.size, 42);
529
530 assert!(find_reusable_by_hash("H2").unwrap().is_none(), "unverified rows never qualify");
531 assert!(find_reusable_by_hash("H3").unwrap().is_none(), "keyless rows never qualify");
532 assert!(find_reusable_by_hash("H9").unwrap().is_none());
533 }
534
535 #[tokio::test]
539 async fn the_ledger_knows_where_the_bytes_already_landed() {
540 let (_tmp, _guard) = init_test_db();
541 let mut on_disk = att("HP", "https://b.example/p", true, true);
542 on_disk.path = "/dl/counter-strike-1.xdc".to_string();
543 save("chat_a", "evt_disk", true, on_disk).await;
544 let mut ghost = att("HP", "https://b.example/p2", true, false);
545 ghost.path = "/dl/never-finished.xdc".to_string();
546 save("chat_b", "evt_ghost", false, ghost).await;
547 save("chat_c", "evt_pathless", true, att("HP", "https://b.example/p3", true, true)).await;
548
549 assert_eq!(
550 downloaded_paths_by_hash("HP").unwrap(),
551 vec!["/dl/counter-strike-1.xdc".to_string()],
552 "only the downloaded row with a real path answers"
553 );
554 assert!(downloaded_paths_by_hash("H9").unwrap().is_empty());
555 }
556
557 #[tokio::test]
561 async fn a_claim_verifies_against_real_bytes_or_not_at_all() {
562 let (_tmp, _guard) = init_test_db();
563 let dir = tempfile::tempdir().unwrap();
564 let good = dir.path().join("counter-strike.xdc");
565 std::fs::write(&good, b"the actual game bytes").unwrap();
566 let real_hash = crate::crypto::sha256_hex(b"the actual game bytes");
567
568 let mut row = att(&real_hash, "https://b.example/cs", true, true);
569 row.path = good.to_string_lossy().to_string();
570 save("chat_a", "evt_cs", true, row).await;
571
572 let claim = |hash: &str, size: u64| crate::types::Attachment {
573 id: hash.to_string(),
574 size,
575 ..Default::default()
576 };
577 let plain_len = b"the actual game bytes".len() as u64;
579 assert!(verify_local_copy(&claim(&real_hash, plain_len)).await.is_some());
580 assert!(verify_local_copy(&claim(&real_hash, plain_len + 16)).await.is_some());
581 assert!(verify_local_copy(&claim(&real_hash, 0)).await.is_some(), "sizeless claims still verify by content");
582 assert!(verify_local_copy(&claim(&real_hash, plain_len + 1)).await.is_none());
584
585 std::fs::write(&good, b"not the game").unwrap();
587 assert!(verify_local_copy(&claim(&real_hash, plain_len)).await.is_none());
588 }
589
590 #[tokio::test]
594 async fn a_shared_blob_survives_every_delete_but_the_last() {
595 let (_tmp, _guard) = init_test_db();
596 let url = "https://b.example/shared".to_string();
597 save("chat_a", "evt_first", true, att("HH", &url, true, true)).await;
598 save("chat_b", "evt_second", true, att("HH", &url, true, true)).await;
599
600 assert!(urls_unreferenced_elsewhere(&[url.clone()], "evt_first").unwrap().is_empty());
602 assert!(urls_unreferenced_elsewhere(&[url.clone()], "evt_second").unwrap().is_empty());
603 assert!(hash_referenced_elsewhere("HH", "evt_first").unwrap());
604
605 crate::db::events::delete_event("evt_second").await.unwrap();
607 assert_eq!(urls_unreferenced_elsewhere(&[url.clone()], "evt_first").unwrap(), vec![url]);
608 assert!(!hash_referenced_elsewhere("HH", "evt_first").unwrap());
609 }
610}