qualia_core_db/services/
webtorrent_seeder.rs

1//! Qualia-native WebTorrent seeder — HTTP web seeds for Permissive Commons
2//! `.q42` volumes and legacy `.c.q42` ontology artifacts.
3//!
4//! This is Commons ICN transport, not a dump of every Q42 file. Unified `.q42`
5//! volumes are classified before they enter the registry: Sanctuary, medical,
6//! bilateral, mixed, and unmarked personal volumes are refused. They travel
7//! on SocialWebNet (pairwise DID / WireGuard) or stay local.
8//!
9//! Browser WebTorrent clients leech via the `ws=` magnet parameter pointing at
10//! `/torrent/webseed/{info_hash}`. One file, one SHA-1 info-hash. A Q42 volume
11//! set publishes the root plus each child as its own magnet.
12
13#![cfg(not(target_arch = "wasm32"))]
14
15use std::collections::HashMap;
16use std::fs;
17use std::path::{Path, PathBuf};
18use std::sync::atomic::{AtomicU64, Ordering};
19use std::sync::{OnceLock, RwLock};
20
21use serde::{Deserialize, Serialize};
22
23static REGISTRY: OnceLock<RwLock<HashMap<String, SeedRecord>>> = OnceLock::new();
24static BYTES_SERVED_SESSION: AtomicU64 = AtomicU64::new(0);
25static GLOBAL_POLICY: OnceLock<RwLock<SeederBandwidthPolicy>> = OnceLock::new();
26
27#[derive(Debug, Clone, Serialize, Deserialize)]
28pub struct SeederBandwidthPolicy {
29    pub global_limit_kbps: u32,
30    pub metered_mode: bool,
31}
32
33impl Default for SeederBandwidthPolicy {
34    fn default() -> Self {
35        Self {
36            global_limit_kbps: 1024,
37            metered_mode: true,
38        }
39    }
40}
41
42#[derive(Debug, Clone, Serialize)]
43pub struct SeedRecord {
44    pub info_hash: String,
45    pub file_path: PathBuf,
46    pub display_name: String,
47    pub ontology_id: String,
48    pub file_size: u64,
49    pub bytes_uploaded_total: u64,
50    pub bytes_uploaded_session: u64,
51    pub download_count: u64,
52    pub deprecated: bool,
53    /// Human principal asserted this is a Commons catalog. Never overrides
54    /// Quin-level Sanctuary / medical / bilateral bits.
55    pub commons_asserted: bool,
56}
57
58#[derive(Debug, Clone, Deserialize)]
59pub struct RegisterSeedRequest {
60    pub info_hash: String,
61    pub file_path: String,
62    pub display_name: String,
63    pub ontology_id: String,
64    #[serde(default)]
65    pub bandwidth_limit_kbps: u32,
66    /// Assert this file is a Permissive Commons catalog (ontologies, not a
67    /// person's medical record). Default false = fail closed.
68    #[serde(default)]
69    pub commons_asserted: bool,
70}
71
72#[derive(Debug, Clone, Deserialize)]
73pub struct UnregisterSeedRequest {
74    pub info_hash: String,
75}
76
77fn registry() -> &'static RwLock<HashMap<String, SeedRecord>> {
78    REGISTRY.get_or_init(|| RwLock::new(HashMap::new()))
79}
80
81fn policy_lock() -> &'static RwLock<SeederBandwidthPolicy> {
82    GLOBAL_POLICY.get_or_init(|| RwLock::new(SeederBandwidthPolicy::default()))
83}
84
85pub fn normalize_info_hash(raw: &str) -> String {
86    raw.trim().to_ascii_lowercase()
87}
88
89pub fn build_magnet_uri(info_hash_sha1: &str, display_name: &str, daemon_port: u16) -> String {
90    let hash = normalize_info_hash(info_hash_sha1);
91    let dn = urlencoding::encode(display_name);
92    let ws_url = format!("http://127.0.0.1:{daemon_port}/torrent/webseed/{hash}");
93    let ws = urlencoding::encode(&ws_url);
94    format!("magnet:?xt=urn:btih:{hash}&dn={dn}&ws={ws}")
95}
96
97pub fn ensure_magnet_webseed(
98    magnet: &str,
99    info_hash: &str,
100    display_name: &str,
101    port: u16,
102) -> String {
103    if magnet.contains("&ws=") || magnet.contains("?ws=") {
104        return magnet.to_string();
105    }
106    build_magnet_uri(info_hash, display_name, port)
107}
108
109pub fn sha1_file(path: &Path) -> Result<String, String> {
110    use sha1::{Digest, Sha1};
111    let mut file = fs::File::open(path).map_err(|e| e.to_string())?;
112    let mut hasher = Sha1::new();
113    let mut buf = [0u8; 8192];
114    loop {
115        let n = std::io::Read::read(&mut file, &mut buf).map_err(|e| e.to_string())?;
116        if n == 0 {
117            break;
118        }
119        hasher.update(&buf[..n]);
120    }
121    Ok(hasher
122        .finalize()
123        .iter()
124        .map(|b| format!("{b:02x}"))
125        .collect())
126}
127
128pub fn set_bandwidth_policy(policy: SeederBandwidthPolicy) {
129    *policy_lock().write().unwrap() = policy;
130}
131
132pub fn get_bandwidth_policy() -> SeederBandwidthPolicy {
133    policy_lock().read().unwrap().clone()
134}
135
136pub fn register_seed(req: RegisterSeedRequest) -> Result<SeedRecord, String> {
137    let hash = normalize_info_hash(&req.info_hash);
138    let path = PathBuf::from(&req.file_path);
139    if !path.is_file() {
140        return Err(format!("Seed file not found: {}", path.display()));
141    }
142    if crate::q42_volume::is_unified_volume(&path).unwrap_or(false) {
143        let intent = if req.commons_asserted {
144            crate::q42_volume::PublicationIntent::CommonsCatalog
145        } else {
146            crate::q42_volume::PublicationIntent::Default
147        };
148        let verdict = crate::q42_volume::classify_q42_path(&path, intent)
149            .map_err(|e| format!("Q42 publication denied: {e}"))?;
150        if !verdict.may_http_webseed {
151            return Err(verdict.reason);
152        }
153    }
154    let computed = sha1_file(&path)?;
155    if computed != hash {
156        return Err(format!(
157            "Info hash mismatch: file SHA1 {computed} != expected {hash}"
158        ));
159    }
160    let file_size = fs::metadata(&path).map_err(|e| e.to_string())?.len();
161    let record = SeedRecord {
162        info_hash: hash.clone(),
163        file_path: path,
164        display_name: req.display_name,
165        ontology_id: req.ontology_id,
166        file_size,
167        bytes_uploaded_total: 0,
168        bytes_uploaded_session: 0,
169        download_count: 0,
170        deprecated: false,
171        commons_asserted: req.commons_asserted,
172    };
173    registry().write().unwrap().insert(hash, record.clone());
174    println!(
175        "[Qualia WebTorrent] Seeding {} ({} bytes) as {}",
176        record.ontology_id, record.file_size, record.info_hash
177    );
178    Ok(record)
179}
180
181pub fn unregister_seed(info_hash: &str) -> bool {
182    let hash = normalize_info_hash(info_hash);
183    registry().write().unwrap().remove(&hash).is_some()
184}
185
186pub fn deprecate_seed(info_hash: &str) -> bool {
187    let hash = normalize_info_hash(info_hash);
188    if let Some(rec) = registry().write().unwrap().get_mut(&hash) {
189        rec.deprecated = true;
190        // DHT Caching for intermittent connections: We retain the record to resume seeding
191        // seamlessly if the node drops offline and reconnects, but mark it as deprecated.
192        return true;
193    }
194    false
195}
196
197pub fn list_active_seeds() -> Vec<SeedRecord> {
198    registry().read().unwrap().values().cloned().collect()
199}
200
201pub fn lookup_seed(info_hash: &str) -> Option<SeedRecord> {
202    registry()
203        .read()
204        .unwrap()
205        .get(&normalize_info_hash(info_hash))
206        .cloned()
207}
208
209pub fn record_bytes_served(info_hash: &str, bytes: u64) {
210    let hash = normalize_info_hash(info_hash);
211    BYTES_SERVED_SESSION.fetch_add(bytes, Ordering::Relaxed);
212    if let Some(rec) = registry().write().unwrap().get_mut(&hash) {
213        rec.bytes_uploaded_session += bytes;
214        rec.bytes_uploaded_total += bytes;
215    }
216}
217
218pub fn record_full_download(info_hash: &str) {
219    let hash = normalize_info_hash(info_hash);
220    if let Some(rec) = registry().write().unwrap().get_mut(&hash) {
221        rec.download_count += 1;
222    }
223}
224
225#[derive(Debug, Clone, Serialize)]
226pub struct SeederTelemetry {
227    pub seeders: usize,
228    pub leechers: u64,
229    pub speed: String,
230    pub status: String,
231    pub uploaded_session_kb: u64,
232    pub uploaded_total_kb: u64,
233    pub active_ontologies: Vec<String>,
234    pub global_bandwidth_kbps: u32,
235    pub metered_mode: bool,
236    pub seeder: String,
237}
238
239pub fn telemetry() -> SeederTelemetry {
240    let seeds = list_active_seeds();
241    let policy = get_bandwidth_policy();
242    let uploaded_session = BYTES_SERVED_SESSION.load(Ordering::Relaxed);
243    let uploaded_total: u64 = seeds.iter().map(|s| s.bytes_uploaded_total).sum();
244    let limit = policy.global_limit_kbps;
245    let speed = if limit == 0 {
246        "unlimited".to_string()
247    } else {
248        format!("{limit} KiB/s cap")
249    };
250    SeederTelemetry {
251        seeders: seeds.len(),
252        leechers: seeds.iter().map(|s| s.download_count).sum(),
253        speed,
254        status: if seeds.is_empty() { "idle" } else { "seeding" }.to_string(),
255        uploaded_session_kb: uploaded_session / 1024,
256        uploaded_total_kb: uploaded_total / 1024,
257        active_ontologies: seeds.iter().map(|s| s.ontology_id.clone()).collect(),
258        global_bandwidth_kbps: policy.global_limit_kbps,
259        metered_mode: policy.metered_mode,
260        seeder: "qualia-daemon".to_string(),
261    }
262}
263
264/// Reload active seeds from `{storage}/Index/workbench.jsonl` on daemon boot.
265pub fn sync_from_workbench(storage_path: &str, daemon_port: u16) {
266    let path = Path::new(storage_path)
267        .join("Index")
268        .join("workbench.jsonl");
269    if !path.is_file() {
270        return;
271    }
272    let Ok(text) = fs::read_to_string(&path) else {
273        return;
274    };
275    for line in text.lines() {
276        if line.trim().is_empty() {
277            continue;
278        }
279        let Ok(v) = serde_json::from_str::<serde_json::Value>(line) else {
280            continue;
281        };
282        let seed_active = v["seed_active"].as_bool().unwrap_or(false);
283        let seed_enabled = v["torrent"]["seed_enabled"].as_bool().unwrap_or(false);
284        if !seed_active || !seed_enabled {
285            continue;
286        }
287        let info_hash = v["info_hash_sha1"].as_str().unwrap_or_default();
288        let file_path = v["c_q42_path"].as_str().unwrap_or_default();
289        let ontology_id = v["ontology_id"].as_str().unwrap_or_default();
290        let title = v["title"].as_str().unwrap_or(ontology_id);
291        if info_hash.is_empty() || file_path.is_empty() {
292            continue;
293        }
294        let _ = register_seed(RegisterSeedRequest {
295            info_hash: info_hash.to_string(),
296            file_path: file_path.to_string(),
297            display_name: format!("{title}.c.q42"),
298            ontology_id: ontology_id.to_string(),
299            bandwidth_limit_kbps: v["torrent"]["bandwidth_limit_kbps"].as_u64().unwrap_or(512)
300                as u32,
301            commons_asserted: true,
302        });
303        if let Some(magnet) = v["magnet_uri"].as_str() {
304            let updated =
305                ensure_magnet_webseed(magnet, info_hash, &format!("{title}.c.q42"), daemon_port);
306            if updated != magnet {
307                // Magnet refresh is persisted by the client on next policy save.
308                let _ = updated;
309            }
310        }
311    }
312    let n = list_active_seeds().len();
313    if n > 0 {
314        println!("[Qualia WebTorrent] Restored {n} active seed(s) from workbench");
315    }
316}
317
318#[cfg(test)]
319mod tests {
320    use super::*;
321
322    #[test]
323    fn magnet_includes_qualia_webseed() {
324        let m = build_magnet_uri("abc123def", "demo.q42", 4242);
325        assert!(m.contains("urn:btih:abc123def"));
326        assert!(m.contains("ws=http"));
327        assert!(m.contains("abc123def"));
328        assert!(m.contains("torrent"));
329        assert!(m.contains("webseed"));
330    }
331
332    #[test]
333    fn normalize_hash_lowercase() {
334        assert_eq!(normalize_info_hash("ABCD"), "abcd");
335    }
336
337    #[test]
338    fn register_seed_refuses_medical_q42() {
339        use crate::q42_volume::write_unified_volume;
340        use crate::NQuin;
341        use std::collections::HashMap;
342
343        let file = tempfile::NamedTempFile::new().unwrap();
344        let mut quin = NQuin {
345            subject: 1,
346            predicate: 2,
347            object: 3,
348            context: 0,
349            metadata: 0,
350            parity: 0,
351        };
352        quin.set_sensitivity_byte(NQuin::SENSITIVITY_CLASSIFIED);
353        quin.set_sensitivity_tier(NQuin::SENSITIVITY_TIER_MEDICAL);
354        write_unified_volume(
355            file.path(),
356            &HashMap::new(),
357            &[(3, 3)],
358            &[vec![quin]],
359        )
360        .unwrap();
361        let hash = sha1_file(file.path()).unwrap();
362        let err = register_seed(RegisterSeedRequest {
363            info_hash: hash,
364            file_path: file.path().display().to_string(),
365            display_name: "pep-record.q42".into(),
366            ontology_id: "should-not-seed".into(),
367            bandwidth_limit_kbps: 1,
368            commons_asserted: true,
369        })
370        .unwrap_err();
371        assert!(
372            err.contains("publication denied") || err.contains("Selfhood") || err.contains("medical"),
373            "{err}"
374        );
375    }
376}
qualia_core_db/services/webtorrent_seeder.rs

qualia_core_db/services/
webtorrent_seeder.rs
