1use anyhow::{Context, Result, anyhow};
20use serde_json::Value;
21use std::collections::HashMap;
22use std::fs;
23use std::io::Write;
24use std::path::{Path, PathBuf};
25use std::sync::{Arc, Mutex, OnceLock};
26
27pub fn config_dir() -> Result<PathBuf> {
32 if let Ok(home) = std::env::var("WIRE_HOME") {
33 return Ok(PathBuf::from(home).join("config").join("wire"));
34 }
35 dirs::config_dir()
36 .map(|d| d.join("wire"))
37 .ok_or_else(|| anyhow!("could not resolve XDG_CONFIG_HOME — set WIRE_HOME"))
38}
39
40pub fn state_dir() -> Result<PathBuf> {
44 if let Ok(home) = std::env::var("WIRE_HOME") {
45 return Ok(PathBuf::from(home).join("state").join("wire"));
46 }
47 dirs::state_dir()
48 .or_else(dirs::data_local_dir)
49 .map(|d| d.join("wire"))
50 .ok_or_else(|| anyhow!("could not resolve XDG_STATE_HOME — set WIRE_HOME"))
51}
52
53pub fn private_key_path() -> Result<PathBuf> {
54 Ok(config_dir()?.join("private.key"))
55}
56pub fn agent_card_path() -> Result<PathBuf> {
57 Ok(config_dir()?.join("agent-card.json"))
58}
59pub fn trust_path() -> Result<PathBuf> {
60 Ok(config_dir()?.join("trust.json"))
61}
62pub fn config_toml_path() -> Result<PathBuf> {
63 Ok(config_dir()?.join("config.toml"))
64}
65pub fn inbox_dir() -> Result<PathBuf> {
66 Ok(state_dir()?.join("inbox"))
67}
68pub fn outbox_dir() -> Result<PathBuf> {
69 Ok(state_dir()?.join("outbox"))
70}
71
72static OUTBOX_LOCKS: OnceLock<Mutex<HashMap<PathBuf, Arc<Mutex<()>>>>> = OnceLock::new();
84
85fn outbox_lock(path: &Path) -> Arc<Mutex<()>> {
86 let registry = OUTBOX_LOCKS.get_or_init(|| Mutex::new(HashMap::new()));
87 let mut g = registry.lock().expect("OUTBOX_LOCKS poisoned");
88 g.entry(path.to_path_buf())
89 .or_insert_with(|| Arc::new(Mutex::new(())))
90 .clone()
91}
92
93pub fn append_pushed_log(peer: &str, event_id: &str, ts: &str) -> Result<PathBuf> {
125 ensure_dirs()?;
126 let normalized = crate::agent_card::bare_handle(peer);
127 let path = outbox_dir()?.join(format!("{normalized}.pushed.jsonl"));
128 let lock = outbox_lock(&path);
129 let _g = lock.lock().expect("pushed-log per-path mutex poisoned");
130 let mut f = fs::OpenOptions::new()
131 .create(true)
132 .append(true)
133 .open(&path)
134 .with_context(|| format!("opening pushed-log {path:?}"))?;
135 let line = serde_json::to_string(&serde_json::json!({
136 "ts": ts,
137 "event_id": event_id,
138 }))?;
139 f.write_all(line.as_bytes())
140 .with_context(|| format!("appending to {path:?}"))?;
141 f.write_all(b"\n")?;
142 Ok(path)
143}
144
145pub fn compute_pending_push_count() -> u64 {
156 compute_pending_push_breakdown()
157 .iter()
158 .map(|p| p.count)
159 .sum()
160}
161
162#[derive(Debug, Clone, serde::Serialize)]
176pub struct PendingPushPerPeer {
177 pub peer: String,
178 pub tier: String,
179 pub count: u64,
180}
181
182pub fn compute_pending_push_breakdown() -> Vec<PendingPushPerPeer> {
183 let trust = match read_trust() {
184 Ok(t) => t,
185 Err(_) => return Vec::new(),
186 };
187 let agents = match trust.get("agents").and_then(serde_json::Value::as_object) {
188 Some(a) => a.clone(),
189 None => return Vec::new(),
190 };
191 let relay_state = read_relay_state().unwrap_or_else(|_| serde_json::json!({"peers": {}}));
195 let mut out: Vec<PendingPushPerPeer> = Vec::new();
196 for (peer_handle, _agent) in agents.iter() {
197 let pushed_ids = read_pushed_event_ids(peer_handle);
198 let outbox_path = match outbox_dir() {
199 Ok(d) => d.join(format!("{peer_handle}.jsonl")),
200 Err(_) => continue,
201 };
202 let body = match fs::read_to_string(&outbox_path) {
203 Ok(b) => b,
204 Err(_) => continue,
205 };
206 let mut count: u64 = 0;
207 for line in body.lines() {
208 if let Some(eid) = serde_json::from_str::<serde_json::Value>(line)
209 .ok()
210 .and_then(|v| {
211 v.get("event_id")
212 .and_then(serde_json::Value::as_str)
213 .map(str::to_string)
214 })
215 && !pushed_ids.contains(&eid)
216 {
217 count += 1;
218 }
219 }
220 if count > 0 {
221 let tier = crate::trust::effective_tier(&trust, &relay_state, peer_handle);
226 out.push(PendingPushPerPeer {
227 peer: peer_handle.clone(),
228 tier,
229 count,
230 });
231 }
232 }
233 out.sort_by(|a, b| b.count.cmp(&a.count).then_with(|| a.peer.cmp(&b.peer)));
236 out
237}
238
239pub fn read_stream_state() -> serde_json::Value {
244 state_dir()
245 .ok()
246 .and_then(|d| fs::read_to_string(d.join("stream_state.json")).ok())
247 .and_then(|body| serde_json::from_str::<serde_json::Value>(&body).ok())
248 .unwrap_or(serde_json::Value::Null)
249}
250
251pub fn stale_sync(last_sync_age_seconds: Option<u64>) -> bool {
255 match last_sync_age_seconds {
256 Some(age) => age > 60,
257 None => true,
258 }
259}
260
261pub fn read_pushed_event_ids(peer: &str) -> std::collections::HashSet<String> {
266 let normalized = crate::agent_card::bare_handle(peer);
267 let path = match outbox_dir() {
268 Ok(d) => d.join(format!("{normalized}.pushed.jsonl")),
269 Err(_) => return std::collections::HashSet::new(),
270 };
271 let body = match fs::read_to_string(&path) {
272 Ok(b) => b,
273 Err(_) => return std::collections::HashSet::new(),
274 };
275 body.lines()
276 .filter_map(|line| {
277 serde_json::from_str::<serde_json::Value>(line)
278 .ok()?
279 .get("event_id")?
280 .as_str()
281 .map(str::to_string)
282 })
283 .collect()
284}
285
286pub fn drain_outbox_delivered(peer: &str) -> Result<()> {
295 let normalized = crate::agent_card::bare_handle(peer);
296 let path = outbox_dir()?.join(format!("{normalized}.jsonl"));
297 let delivered = read_pushed_event_ids(peer);
298 if delivered.is_empty() {
299 return Ok(());
300 }
301 let lock = outbox_lock(&path);
302 let _g = lock.lock().expect("outbox per-path mutex poisoned");
303 let body = match fs::read_to_string(&path) {
304 Ok(b) => b,
305 Err(_) => return Ok(()), };
307 let mut kept = String::with_capacity(body.len());
308 let mut dropped = 0usize;
309 for line in body.lines() {
310 let is_delivered = serde_json::from_str::<serde_json::Value>(line)
311 .ok()
312 .and_then(|v| {
313 v.get("event_id")
314 .and_then(|e| e.as_str())
315 .map(str::to_string)
316 })
317 .map(|id| delivered.contains(&id))
318 .unwrap_or(false); if is_delivered {
320 dropped += 1;
321 } else {
322 kept.push_str(line);
323 kept.push('\n');
324 }
325 }
326 if dropped == 0 {
327 return Ok(()); }
329 let tmp = path.with_extension("jsonl.tmp");
330 fs::write(&tmp, kept.as_bytes()).with_context(|| format!("writing {tmp:?}"))?;
331 fs::rename(&tmp, &path).with_context(|| format!("renaming {tmp:?} -> {path:?}"))?;
332 Ok(())
333}
334
335pub fn append_outbox_record(peer: &str, record_bytes: &[u8]) -> Result<PathBuf> {
336 ensure_dirs()?;
337 let normalized = crate::agent_card::bare_handle(peer);
338 let path = outbox_dir()?.join(format!("{normalized}.jsonl"));
339 let lock = outbox_lock(&path);
340 let _g = lock.lock().expect("outbox per-path mutex poisoned");
341 let mut f = fs::OpenOptions::new()
342 .create(true)
343 .append(true)
344 .open(&path)
345 .with_context(|| format!("opening outbox {path:?}"))?;
346 let mut buf = Vec::with_capacity(record_bytes.len() + 1);
347 buf.extend_from_slice(record_bytes);
348 buf.push(b'\n');
349 f.write_all(&buf)
350 .with_context(|| format!("appending to {path:?}"))?;
351 Ok(path)
352}
353
354pub fn is_initialized() -> Result<bool> {
356 Ok(private_key_path()?.exists() && agent_card_path()?.exists())
357}
358
359pub fn ensure_dirs() -> Result<()> {
361 let cfg = config_dir()?;
362 fs::create_dir_all(&cfg).with_context(|| format!("creating {cfg:?}"))?;
363 fs::create_dir_all(state_dir()?)?;
364 fs::create_dir_all(inbox_dir()?)?;
365 fs::create_dir_all(outbox_dir()?)?;
366 set_dir_mode_0700(&cfg)?;
367 Ok(())
368}
369
370#[cfg(unix)]
371fn set_dir_mode_0700(path: &Path) -> Result<()> {
372 use std::os::unix::fs::PermissionsExt;
373 let mut perms = fs::metadata(path)?.permissions();
374 perms.set_mode(0o700);
375 fs::set_permissions(path, perms)?;
376 Ok(())
377}
378
379#[cfg(not(unix))]
380fn set_dir_mode_0700(_: &Path) -> Result<()> {
381 Ok(())
382}
383
384pub fn write_private_key(seed: &[u8; 32]) -> Result<()> {
386 let path = private_key_path()?;
387 fs::write(&path, seed).with_context(|| format!("writing {path:?}"))?;
388 set_file_mode_0600(&path)?;
389 Ok(())
390}
391
392#[cfg(unix)]
393fn set_file_mode_0600(path: &Path) -> Result<()> {
394 use std::os::unix::fs::PermissionsExt;
395 let mut perms = fs::metadata(path)?.permissions();
396 perms.set_mode(0o600);
397 fs::set_permissions(path, perms)?;
398 Ok(())
399}
400
401#[cfg(not(unix))]
402fn set_file_mode_0600(_: &Path) -> Result<()> {
403 Ok(())
404}
405
406pub fn read_private_key() -> Result<[u8; 32]> {
408 let path = private_key_path()?;
409 let bytes = fs::read(&path).with_context(|| format!("reading {path:?}"))?;
410 if bytes.len() != 32 {
411 return Err(anyhow!(
412 "private key file has wrong length ({} != 32)",
413 bytes.len()
414 ));
415 }
416 let mut seed = [0u8; 32];
417 seed.copy_from_slice(&bytes);
418 Ok(seed)
419}
420
421pub fn op_key_path() -> Result<PathBuf> {
427 Ok(config_dir()?.join("op.key"))
428}
429
430fn did_filename(did: &str) -> String {
432 did.chars()
433 .map(|c| {
434 if c.is_ascii_alphanumeric() || c == '-' {
435 c
436 } else {
437 '_'
438 }
439 })
440 .collect()
441}
442
443pub fn org_key_path(org_did: &str) -> Result<PathBuf> {
444 Ok(config_dir()?
445 .join("orgs")
446 .join(format!("{}.key", did_filename(org_did))))
447}
448
449fn write_seed_0600(path: &Path, seed: &[u8; 32]) -> Result<()> {
450 if let Some(parent) = path.parent() {
451 fs::create_dir_all(parent)?;
452 }
453 fs::write(path, seed).with_context(|| format!("writing {path:?}"))?;
454 set_file_mode_0600(path)?;
455 Ok(())
456}
457
458fn read_seed(path: &Path) -> Result<[u8; 32]> {
459 let bytes = fs::read(path).with_context(|| format!("reading {path:?}"))?;
460 if bytes.len() != 32 {
461 return Err(anyhow!(
462 "key file {path:?} has wrong length ({} != 32)",
463 bytes.len()
464 ));
465 }
466 let mut seed = [0u8; 32];
467 seed.copy_from_slice(&bytes);
468 Ok(seed)
469}
470
471pub fn write_op_key(seed: &[u8; 32]) -> Result<()> {
472 write_seed_0600(&op_key_path()?, seed)
473}
474pub fn read_op_key() -> Result<[u8; 32]> {
475 read_seed(&op_key_path()?)
476}
477
478pub fn nostr_key_path() -> Result<PathBuf> {
481 Ok(config_dir()?.join("nostr.key"))
482}
483pub fn write_nostr_key(secret: &[u8; 32]) -> Result<()> {
484 write_seed_0600(&nostr_key_path()?, secret)
485}
486pub fn read_nostr_key() -> Result<[u8; 32]> {
487 read_seed(&nostr_key_path()?)
488}
489pub fn write_org_key(org_did: &str, seed: &[u8; 32]) -> Result<()> {
490 write_seed_0600(&org_key_path(org_did)?, seed)
491}
492pub fn read_org_key(org_did: &str) -> Result<[u8; 32]> {
493 read_seed(&org_key_path(org_did)?)
494}
495
496pub fn succession_log_path() -> Result<PathBuf> {
497 Ok(config_dir()?.join("succession.jsonl"))
498}
499
500pub fn append_succession_record(
504 kind: &str,
505 old_did: &str,
506 new_did: &str,
507 cert: &str,
508) -> Result<()> {
509 let path = succession_log_path()?;
510 if let Some(p) = path.parent() {
511 fs::create_dir_all(p)?;
512 }
513 let at_unix = std::time::SystemTime::now()
514 .duration_since(std::time::UNIX_EPOCH)
515 .map(|d| d.as_secs())
516 .unwrap_or(0);
517 let line = serde_json::to_string(&serde_json::json!({
518 "kind": kind,
519 "old_did": old_did,
520 "new_did": new_did,
521 "cert": cert,
522 "at_unix": at_unix,
523 }))?;
524 use std::io::Write;
525 let mut f = fs::OpenOptions::new()
526 .create(true)
527 .append(true)
528 .open(&path)
529 .with_context(|| format!("opening {path:?}"))?;
530 writeln!(f, "{line}")?;
531 set_file_mode_0600(&path)?;
532 Ok(())
533}
534
535pub fn op_meta_path() -> Result<PathBuf> {
536 Ok(config_dir()?.join("op.json"))
537}
538
539pub fn write_op_handle(handle: &str) -> Result<()> {
542 let path = op_meta_path()?;
543 if let Some(p) = path.parent() {
544 fs::create_dir_all(p)?;
545 }
546 fs::write(
547 &path,
548 serde_json::to_vec_pretty(&serde_json::json!({ "handle": handle }))?,
549 )?;
550 set_file_mode_0600(&path)?;
551 Ok(())
552}
553
554pub fn read_op_handle() -> Result<Option<String>> {
555 let Ok(bytes) = fs::read(op_meta_path()?) else {
556 return Ok(None);
557 };
558 let v: Value = serde_json::from_slice(&bytes)?;
559 Ok(v.get("handle").and_then(Value::as_str).map(str::to_string))
560}
561
562pub fn memberships_path() -> Result<PathBuf> {
563 Ok(config_dir()?.join("memberships.json"))
564}
565
566pub fn add_membership(org_did: &str, org_pubkey: &str, member_cert: &str) -> Result<()> {
570 let mut list = read_memberships()?;
571 list.retain(|m| m.get("org_did").and_then(Value::as_str) != Some(org_did));
572 list.push(serde_json::json!({
573 "org_did": org_did, "org_pubkey": org_pubkey, "member_cert": member_cert
574 }));
575 let path = memberships_path()?;
576 if let Some(p) = path.parent() {
577 fs::create_dir_all(p)?;
578 }
579 fs::write(&path, serde_json::to_vec_pretty(&Value::Array(list))?)?;
580 Ok(())
581}
582
583pub fn read_memberships() -> Result<Vec<Value>> {
585 let Ok(bytes) = fs::read(memberships_path()?) else {
586 return Ok(vec![]);
587 };
588 Ok(serde_json::from_slice::<Value>(&bytes)
589 .ok()
590 .and_then(|v| v.as_array().cloned())
591 .unwrap_or_default())
592}
593
594pub fn write_agent_card(card: &Value) -> Result<()> {
595 let path = agent_card_path()?;
596 let body = serde_json::to_vec_pretty(card)?;
597 let tmp = path.with_extension("json.tmp");
603 fs::write(&tmp, body).with_context(|| format!("writing tmp {tmp:?}"))?;
604 fs::rename(&tmp, &path).with_context(|| format!("atomic rename {tmp:?} → {path:?}"))?;
605 Ok(())
606}
607
608pub fn read_agent_card() -> Result<Value> {
609 let path = agent_card_path()?;
610 let body = fs::read(&path).with_context(|| format!("reading {path:?}"))?;
611 Ok(serde_json::from_slice(&body)?)
612}
613
614pub fn display_overrides_path() -> Result<PathBuf> {
622 Ok(config_dir()?.join("display.json"))
623}
624
625#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
626pub struct DisplayOverrides {
627 #[serde(default, skip_serializing_if = "Option::is_none")]
628 pub nickname: Option<String>,
629 #[serde(default, skip_serializing_if = "Option::is_none")]
630 pub emoji: Option<String>,
631}
632
633pub fn read_display_overrides() -> Result<DisplayOverrides> {
634 read_display_overrides_at(&display_overrides_path()?)
635}
636
637pub fn read_display_overrides_at(path: &Path) -> Result<DisplayOverrides> {
638 if !path.exists() {
639 return Ok(DisplayOverrides::default());
640 }
641 let body = fs::read(path).with_context(|| format!("reading {path:?}"))?;
642 Ok(serde_json::from_slice(&body)?)
643}
644
645pub fn write_display_overrides(overrides: &DisplayOverrides) -> Result<()> {
646 let path = display_overrides_path()?;
647 if let Some(parent) = path.parent() {
648 fs::create_dir_all(parent).with_context(|| format!("creating {parent:?}"))?;
649 }
650 let body = serde_json::to_vec_pretty(overrides)?;
651 let tmp = path.with_extension("json.tmp");
655 fs::write(&tmp, body).with_context(|| format!("writing tmp {tmp:?}"))?;
656 fs::rename(&tmp, &path).with_context(|| format!("atomic rename {tmp:?} → {path:?}"))?;
657 Ok(())
658}
659
660fn trust_state_lock_path() -> Result<PathBuf> {
664 Ok(config_dir()?.join("trust.lock"))
665}
666
667pub fn write_trust(trust: &Value) -> Result<()> {
681 use fs2::FileExt;
682 let lock_path = trust_state_lock_path()?;
683 if let Some(parent) = lock_path.parent() {
684 fs::create_dir_all(parent).with_context(|| format!("creating {parent:?}"))?;
685 }
686 let lock_file = fs::OpenOptions::new()
687 .create(true)
688 .truncate(false)
689 .read(true)
690 .write(true)
691 .open(&lock_path)
692 .with_context(|| format!("opening {lock_path:?}"))?;
693 lock_file
694 .lock_exclusive()
695 .with_context(|| format!("flock {lock_path:?}"))?;
696 let r = write_trust_unlocked(trust);
697 let _ = fs2::FileExt::unlock(&lock_file);
698 r
699}
700
701fn write_trust_unlocked(trust: &Value) -> Result<()> {
705 let path = trust_path()?;
706 let body = serde_json::to_vec_pretty(trust)?;
707 let tmp = path.with_extension("json.tmp");
708 fs::write(&tmp, &body).with_context(|| format!("writing tmp {tmp:?}"))?;
709 fs::rename(&tmp, &path).with_context(|| format!("atomic rename {tmp:?} → {path:?}"))?;
710 Ok(())
711}
712
713pub fn read_trust() -> Result<Value> {
714 let path = trust_path()?;
715 if !path.exists() {
716 return Ok(crate::trust::empty_trust());
717 }
718 let body = fs::read(&path).with_context(|| format!("reading {path:?}"))?;
719 Ok(serde_json::from_slice(&body)?)
720}
721
722pub fn update_trust<F>(modifier: F) -> Result<()>
730where
731 F: FnOnce(&mut Value) -> Result<()>,
732{
733 use fs2::FileExt;
734 let lock_path = trust_state_lock_path()?;
735 if let Some(parent) = lock_path.parent() {
736 fs::create_dir_all(parent).with_context(|| format!("creating {parent:?}"))?;
737 }
738 let lock_file = fs::OpenOptions::new()
739 .create(true)
740 .truncate(false)
741 .read(true)
742 .write(true)
743 .open(&lock_path)
744 .with_context(|| format!("opening {lock_path:?}"))?;
745 lock_file
746 .lock_exclusive()
747 .with_context(|| format!("flock {lock_path:?}"))?;
748
749 let mut trust = read_trust()?;
752 let result = modifier(&mut trust);
753 let write_result = if result.is_ok() {
754 write_trust_unlocked(&trust)
755 } else {
756 Ok(())
757 };
758 let _ = fs2::FileExt::unlock(&lock_file);
759 result?;
760 write_result?;
761 Ok(())
762}
763
764pub fn relay_state_path() -> Result<PathBuf> {
769 Ok(config_dir()?.join("relay.json"))
770}
771
772pub fn read_relay_state() -> Result<Value> {
773 let path = relay_state_path()?;
774 if !path.exists() {
775 return Ok(serde_json::json!({"self": Value::Null, "peers": {}}));
776 }
777 let body = fs::read(&path).with_context(|| format!("reading {path:?}"))?;
778 Ok(serde_json::from_slice(&body)?)
779}
780
781pub fn write_relay_state(state: &Value) -> Result<()> {
794 use fs2::FileExt;
795 let lock_file = acquire_relay_lock(std::process::id())?;
796 let r = write_relay_state_unlocked(state);
797 let _ = FileExt::unlock(&lock_file);
798 r
799}
800
801fn write_relay_state_unlocked(state: &Value) -> Result<()> {
806 let path = relay_state_path()?;
807 let body = serde_json::to_vec_pretty(state)?;
808 let tmp = path.with_extension("json.tmp");
809 fs::write(&tmp, &body).with_context(|| format!("writing tmp {tmp:?}"))?;
810 set_file_mode_0600(&tmp)?;
811 fs::rename(&tmp, &path).with_context(|| format!("atomic rename {tmp:?} → {path:?}"))?;
812 Ok(())
813}
814
815fn relay_state_lock_path() -> Result<PathBuf> {
820 Ok(config_dir()?.join("relay.lock"))
821}
822
823fn relay_lock_timeout() -> std::time::Duration {
828 std::env::var("WIRE_RELAY_LOCK_TIMEOUT_SECS")
829 .ok()
830 .and_then(|s| s.parse::<u64>().ok())
831 .map(std::time::Duration::from_secs)
832 .unwrap_or_else(|| std::time::Duration::from_secs(10))
833}
834
835#[derive(Debug, PartialEq, Eq)]
839pub(crate) enum LockAttemptOutcome {
840 HeldByAlive(u32),
843 HeldByDeadOrAbsent(Option<u32>),
848}
849
850pub(crate) fn classify_contention(
861 body: &[u8],
862 is_alive: impl Fn(u32) -> bool,
863) -> LockAttemptOutcome {
864 let pid = std::str::from_utf8(body)
865 .ok()
866 .and_then(|s| s.trim().parse::<u32>().ok());
867 match pid {
868 Some(p) if is_alive(p) => LockAttemptOutcome::HeldByAlive(p),
869 Some(p) => LockAttemptOutcome::HeldByDeadOrAbsent(Some(p)),
870 None => LockAttemptOutcome::HeldByDeadOrAbsent(None),
871 }
872}
873
874fn relay_state_lock_owner_path() -> Result<PathBuf> {
880 Ok(config_dir()?.join("relay.lock.owner"))
881}
882
883fn acquire_relay_lock(our_pid: u32) -> Result<fs::File> {
911 use fs2::FileExt;
912 let lock_path = relay_state_lock_path()?;
913 if let Some(parent) = lock_path.parent() {
914 fs::create_dir_all(parent).with_context(|| format!("creating {parent:?}"))?;
915 }
916 let owner_path = relay_state_lock_owner_path()?;
917 let deadline = std::time::Instant::now() + relay_lock_timeout();
918 let mut backoff = std::time::Duration::from_millis(10);
919 loop {
920 let lock_file = fs::OpenOptions::new()
921 .create(true)
922 .truncate(false)
923 .read(true)
924 .write(true)
925 .open(&lock_path)
926 .with_context(|| format!("opening {lock_path:?}"))?;
927 match lock_file.try_lock_exclusive() {
928 Ok(()) => {
929 let _ = fs::write(&owner_path, our_pid.to_string());
936 return Ok(lock_file);
937 }
938 Err(_) => {
939 drop(lock_file);
940 let body = fs::read(&owner_path).unwrap_or_default();
941 match classify_contention(&body, crate::platform::process_alive) {
942 LockAttemptOutcome::HeldByDeadOrAbsent(_) => {
943 std::thread::sleep(std::time::Duration::from_millis(1));
947 }
948 LockAttemptOutcome::HeldByAlive(holder_pid) => {
949 if std::time::Instant::now() >= deadline {
950 return Err(anyhow!(
951 "relay.lock held by live pid {holder_pid} after {}s — \
952 likely a hung wire process. Run `wire doctor`, or \
953 kill {holder_pid} and retry.",
954 relay_lock_timeout().as_secs(),
955 ));
956 }
957 std::thread::sleep(backoff);
958 backoff = (backoff * 2).min(std::time::Duration::from_millis(200));
959 }
960 }
961 }
962 }
963 }
964}
965
966pub fn update_relay_state<F>(modifier: F) -> Result<()>
981where
982 F: FnOnce(&mut Value) -> Result<()>,
983{
984 use fs2::FileExt;
985 let lock_file = acquire_relay_lock(std::process::id())?;
986
987 let mut state = read_relay_state()?;
990 let result = modifier(&mut state);
991 let write_result = if result.is_ok() {
992 write_relay_state_unlocked(&state)
995 } else {
996 Ok(())
997 };
998 let _ = FileExt::unlock(&lock_file);
1001 result?;
1002 write_result?;
1003 Ok(())
1004}
1005
1006#[cfg(test)]
1011pub(crate) mod test_support {
1012 use std::sync::Mutex;
1013
1014 pub static ENV_LOCK: Mutex<()> = Mutex::new(());
1015
1016 pub fn with_temp_home<F: FnOnce()>(f: F) {
1017 let _guard = ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
1019 let tmp = std::env::temp_dir().join(format!("wire-test-{}", rand::random::<u32>()));
1020 unsafe { std::env::set_var("WIRE_HOME", &tmp) };
1022 let _ = std::fs::remove_dir_all(&tmp);
1023 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
1024 unsafe { std::env::remove_var("WIRE_HOME") };
1025 let _ = std::fs::remove_dir_all(&tmp);
1026 if let Err(e) = result {
1027 std::panic::resume_unwind(e);
1028 }
1029 }
1030}
1031
1032#[cfg(test)]
1033mod tests {
1034 use super::*;
1035 use serde_json::json;
1036
1037 #[test]
1038 fn did_filename_sanitizes_did_punctuation() {
1039 assert_eq!(
1040 did_filename("did:wire:org:slanchaai-abc123"),
1041 "did_wire_org_slanchaai-abc123"
1042 );
1043 let f = did_filename("did:wire:org:x/../../etc");
1045 assert!(!f.contains('/') && !f.contains('.'));
1046 }
1047
1048 #[test]
1049 fn op_and_org_key_roundtrip() {
1050 with_temp_home(|| {
1051 let op_seed = [7u8; 32];
1052 write_op_key(&op_seed).unwrap();
1053 assert_eq!(read_op_key().unwrap(), op_seed);
1054
1055 let org_did = "did:wire:org:slanchaai-deadbeef";
1056 let org_seed = [9u8; 32];
1057 write_org_key(org_did, &org_seed).unwrap();
1058 assert_eq!(read_org_key(org_did).unwrap(), org_seed);
1059 });
1060 }
1061
1062 fn with_temp_home<F: FnOnce()>(f: F) {
1063 super::test_support::with_temp_home(f)
1064 }
1065
1066 #[test]
1067 fn read_trust_missing_is_ok_empty_but_corrupt_is_err() {
1068 with_temp_home(|| {
1073 let t = read_trust().unwrap();
1075 assert!(t.get("agents").is_some(), "missing trust → empty skeleton");
1076
1077 ensure_dirs().unwrap();
1079 std::fs::write(trust_path().unwrap(), b"{ this is not json").unwrap();
1080 assert!(
1081 read_trust().is_err(),
1082 "corrupt trust.json must Err, not swallow"
1083 );
1084 });
1085 }
1086
1087 #[test]
1088 fn drain_outbox_removes_only_delivered_lines() {
1089 with_temp_home(|| {
1090 let peer = "alpha-fox";
1091 for id in ["e1", "e2", "e3"] {
1092 append_outbox_record(peer, format!("{{\"event_id\":\"{id}\"}}").as_bytes())
1093 .unwrap();
1094 }
1095 append_pushed_log(peer, "e1", "t").unwrap();
1097 append_pushed_log(peer, "e3", "t").unwrap();
1098
1099 drain_outbox_delivered(peer).unwrap();
1100
1101 let body = fs::read_to_string(outbox_dir().unwrap().join("alpha-fox.jsonl")).unwrap();
1102 assert!(body.contains("\"e2\""), "undelivered line kept");
1103 assert!(!body.contains("\"e1\""), "delivered line dropped");
1104 assert!(!body.contains("\"e3\""), "delivered line dropped");
1105 assert_eq!(body.lines().count(), 1, "only the pending line remains");
1106 });
1107 }
1108
1109 #[test]
1110 fn config_dir_honors_wire_home() {
1111 with_temp_home(|| {
1112 let dir = config_dir().unwrap();
1113 assert!(dir.ends_with("wire"), "got {dir:?}");
1114 assert!(dir.to_string_lossy().contains("wire-test-"));
1115 });
1116 }
1117
1118 #[test]
1119 fn ensure_dirs_creates_layout() {
1120 with_temp_home(|| {
1121 ensure_dirs().unwrap();
1122 assert!(config_dir().unwrap().is_dir());
1123 assert!(state_dir().unwrap().is_dir());
1124 assert!(inbox_dir().unwrap().is_dir());
1125 assert!(outbox_dir().unwrap().is_dir());
1126 });
1127 }
1128
1129 #[test]
1130 fn private_key_roundtrip() {
1131 with_temp_home(|| {
1132 ensure_dirs().unwrap();
1133 let seed = [42u8; 32];
1134 write_private_key(&seed).unwrap();
1135 let read_back = read_private_key().unwrap();
1136 assert_eq!(seed, read_back);
1137 });
1138 }
1139
1140 #[test]
1141 fn agent_card_roundtrip() {
1142 with_temp_home(|| {
1143 ensure_dirs().unwrap();
1144 let card = json!({"did": "did:wire:paul", "name": "Paul"});
1145 write_agent_card(&card).unwrap();
1146 let read_back = read_agent_card().unwrap();
1147 assert_eq!(card, read_back);
1148 });
1149 }
1150
1151 #[test]
1152 fn trust_returns_empty_when_missing() {
1153 with_temp_home(|| {
1154 ensure_dirs().unwrap();
1155 let t = read_trust().unwrap();
1156 assert_eq!(t["version"], 1);
1157 assert!(t["agents"].is_object());
1158 });
1159 }
1160
1161 #[test]
1162 fn update_relay_state_writes_through_lock() {
1163 with_temp_home(|| {
1169 ensure_dirs().unwrap();
1170 let initial = json!({"self": null, "peers": {}});
1172 write_relay_state(&initial).unwrap();
1173 super::update_relay_state(|state| {
1175 state["self"] = json!({
1176 "relay_url": "https://test",
1177 "slot_id": "abc",
1178 "slot_token": "tok",
1179 });
1180 Ok(())
1181 })
1182 .unwrap();
1183 let after = read_relay_state().unwrap();
1185 assert_eq!(after["self"]["relay_url"], "https://test");
1186 assert_eq!(after["self"]["slot_id"], "abc");
1187 });
1188 }
1189
1190 #[test]
1191 fn write_relay_state_never_tears_under_concurrency() {
1192 with_temp_home(|| {
1199 ensure_dirs().unwrap();
1200 write_relay_state(&json!({"self": null, "peers": {}})).unwrap();
1201 let handles: Vec<_> = (0..8)
1202 .map(|w| {
1203 std::thread::spawn(move || {
1204 for j in 0..25 {
1205 let body = if j % 2 == 0 {
1206 json!({"self": {"w": w, "j": j, "pad": "x".repeat(2048)}})
1207 } else {
1208 json!({"self": {"w": w}})
1209 };
1210 write_relay_state(&body).unwrap();
1211 read_relay_state().expect("relay.json must always parse");
1213 }
1214 })
1215 })
1216 .collect();
1217 for h in handles {
1218 h.join().unwrap();
1219 }
1220 assert!(read_relay_state().unwrap().get("self").is_some());
1221 });
1222 }
1223
1224 #[test]
1225 fn write_trust_round_trips_and_leaves_no_tmp() {
1226 with_temp_home(|| {
1229 ensure_dirs().unwrap();
1230 let t = json!({"version": 1, "agents": {"did:wire:raven-kettle-465c3352": {"tier": "VERIFIED"}}});
1231 write_trust(&t).unwrap();
1232 let back = read_trust().unwrap();
1233 assert_eq!(
1234 back["agents"]["did:wire:raven-kettle-465c3352"]["tier"],
1235 "VERIFIED"
1236 );
1237 let tmp = trust_path().unwrap().with_extension("json.tmp");
1238 assert!(!tmp.exists(), "tmp file must be consumed by the rename");
1239 });
1240 }
1241
1242 #[test]
1243 fn write_trust_never_tears_under_concurrency() {
1244 with_temp_home(|| {
1248 ensure_dirs().unwrap();
1249 write_trust(&json!({"version": 1, "agents": {}})).unwrap();
1250 let handles: Vec<_> = (0..8)
1251 .map(|w| {
1252 std::thread::spawn(move || {
1253 for j in 0..25 {
1254 let body = if j % 2 == 0 {
1255 json!({"version": 1, "agents": {"a": {"w": w, "pad": "x".repeat(2048)}}})
1256 } else {
1257 json!({"version": 1, "agents": {"a": {"w": w}}})
1258 };
1259 write_trust(&body).unwrap();
1260 read_trust().expect("trust.json must always parse");
1261 }
1262 })
1263 })
1264 .collect();
1265 for h in handles {
1266 h.join().unwrap();
1267 }
1268 assert_eq!(read_trust().unwrap()["version"], 1);
1269 });
1270 }
1271
1272 #[test]
1273 fn update_trust_no_lost_update_under_concurrency() {
1274 with_temp_home(|| {
1279 ensure_dirs().unwrap();
1280 write_trust(&json!({"version": 1, "agents": {}})).unwrap();
1281 let handles: Vec<_> = (0..8)
1282 .map(|w| {
1283 std::thread::spawn(move || {
1284 for j in 0..15 {
1285 let key = format!("peer-{w}-{j}");
1286 update_trust(|t| {
1287 t["agents"][&key] = json!({"tier": "VERIFIED"});
1288 Ok(())
1289 })
1290 .unwrap();
1291 }
1292 })
1293 })
1294 .collect();
1295 for h in handles {
1296 h.join().unwrap();
1297 }
1298 let agents = read_trust().unwrap();
1299 let n = agents["agents"].as_object().unwrap().len();
1300 assert_eq!(
1301 n,
1302 8 * 15,
1303 "every concurrent add must survive (no lost update)"
1304 );
1305 });
1306 }
1307
1308 #[test]
1309 fn update_trust_modifier_error_does_not_clobber() {
1310 with_temp_home(|| {
1311 ensure_dirs().unwrap();
1312 write_trust(&json!({"version": 1, "agents": {"keep": {"tier": "VERIFIED"}}})).unwrap();
1313 let r = update_trust(|t| {
1314 t["agents"]["transient"] = json!({"tier": "X"});
1315 anyhow::bail!("simulated mid-RMW error")
1316 });
1317 assert!(r.is_err());
1318 let after = read_trust().unwrap();
1319 assert!(
1320 after["agents"]["keep"].is_object(),
1321 "prior pin must survive"
1322 );
1323 assert!(
1324 after["agents"]["transient"].is_null(),
1325 "aborted modifier must not persist"
1326 );
1327 });
1328 }
1329
1330 #[test]
1331 fn update_relay_state_modifier_error_does_not_clobber() {
1332 with_temp_home(|| {
1336 ensure_dirs().unwrap();
1337 let initial = json!({"self": {"relay_url": "https://prior"}, "peers": {}});
1338 write_relay_state(&initial).unwrap();
1339 let result = super::update_relay_state(|state| {
1340 state["self"] = json!({"relay_url": "https://NEVER_PERSIST"});
1342 anyhow::bail!("simulated mid-RMW error")
1344 });
1345 assert!(result.is_err());
1346 let after = read_relay_state().unwrap();
1347 assert_eq!(
1348 after["self"]["relay_url"], "https://prior",
1349 "state on disk must not reflect aborted modifier"
1350 );
1351 });
1352 }
1353
1354 #[test]
1355 fn is_initialized_true_only_after_both_files_written() {
1356 with_temp_home(|| {
1357 ensure_dirs().unwrap();
1358 assert!(!is_initialized().unwrap());
1359 write_private_key(&[0u8; 32]).unwrap();
1360 assert!(!is_initialized().unwrap()); write_agent_card(&json!({"did": "did:wire:paul"})).unwrap();
1362 assert!(is_initialized().unwrap());
1363 });
1364 }
1365
1366 #[cfg(unix)]
1367 #[test]
1368 fn append_outbox_record_normalizes_fqdn_to_bare_handle() {
1369 with_temp_home(|| {
1373 let path_fqdn = append_outbox_record("bob@wireup.net", b"{\"kind\":1100}").unwrap();
1374 let path_bare = append_outbox_record("bob", b"{\"kind\":1100}").unwrap();
1375 assert_eq!(path_fqdn, path_bare, "FQDN form should normalize to bare");
1377 assert!(
1378 path_fqdn.file_name().unwrap().to_string_lossy() == "bob.jsonl",
1379 "expected bob.jsonl, got {path_fqdn:?}"
1380 );
1381 let outbox = outbox_dir().unwrap();
1383 assert!(
1384 !outbox.join("bob@wireup.net.jsonl").exists(),
1385 "FQDN-named file must not be created"
1386 );
1387 let body = std::fs::read_to_string(&path_bare).unwrap();
1389 assert_eq!(body.matches("kind").count(), 2, "got: {body}");
1390 });
1391 }
1392
1393 #[test]
1394 fn pending_push_breakdown_attributes_per_peer_with_tier() {
1395 with_temp_home(|| {
1396 ensure_dirs().unwrap();
1397 let trust = json!({
1399 "agents": {
1400 "alpha-fox": {"tier": "VERIFIED"},
1401 "beta-newt": {"tier": "PENDING_ACK"},
1402 "gamma-otter": {"tier": "UNTRUSTED"},
1403 }
1404 });
1405 write_trust(&trust).unwrap();
1406 let relay = json!({
1412 "self": null,
1413 "peers": {
1414 "alpha-fox": {
1415 "bilateral_completed_at": "2026-06-01T00:00:00Z"
1416 }
1417 }
1418 });
1419 write_relay_state(&relay).unwrap();
1420 let out = outbox_dir().unwrap();
1428 std::fs::write(
1429 out.join("alpha-fox.jsonl"),
1430 "{\"event_id\":\"a1\"}\n{\"event_id\":\"a2\"}\n",
1431 )
1432 .unwrap();
1433 std::fs::write(
1434 out.join("alpha-fox.pushed.jsonl"),
1435 "{\"event_id\":\"a1\"}\n",
1436 )
1437 .unwrap();
1438 std::fs::write(
1439 out.join("beta-newt.jsonl"),
1440 "{\"event_id\":\"b1\"}\n{\"event_id\":\"b2\"}\n{\"event_id\":\"b3\"}\n",
1441 )
1442 .unwrap();
1443 let bd = compute_pending_push_breakdown();
1444 assert_eq!(bd.len(), 2, "got: {bd:?}");
1445 assert_eq!(bd[0].peer, "beta-newt");
1446 assert_eq!(bd[0].tier, "PENDING_ACK");
1447 assert_eq!(bd[0].count, 3);
1448 assert_eq!(bd[1].peer, "alpha-fox");
1449 assert_eq!(bd[1].tier, "VERIFIED");
1450 assert_eq!(bd[1].count, 1);
1451 assert_eq!(compute_pending_push_count(), 4);
1453 });
1454 }
1455
1456 #[cfg(unix)]
1457 #[test]
1458 fn private_key_is_mode_0600() {
1459 use std::os::unix::fs::PermissionsExt;
1460 with_temp_home(|| {
1461 ensure_dirs().unwrap();
1462 write_private_key(&[1u8; 32]).unwrap();
1463 let mode = fs::metadata(private_key_path().unwrap())
1464 .unwrap()
1465 .permissions()
1466 .mode();
1467 assert_eq!(mode & 0o777, 0o600, "got {:o}", mode & 0o777);
1468 });
1469 }
1470
1471 #[test]
1474 fn classify_contention_dead_pid_says_reclaim() {
1475 let body = b"12345";
1476 let outcome = classify_contention(body, |_| false);
1478 assert_eq!(outcome, LockAttemptOutcome::HeldByDeadOrAbsent(Some(12345)));
1479 }
1480
1481 #[test]
1482 fn classify_contention_live_pid_says_wait() {
1483 let body = b"54321";
1484 let outcome = classify_contention(body, |pid| pid == 54321);
1486 assert_eq!(outcome, LockAttemptOutcome::HeldByAlive(54321));
1487 }
1488
1489 #[test]
1490 fn classify_contention_empty_body_says_reclaim() {
1491 let outcome = classify_contention(b"", |_| true);
1492 assert_eq!(outcome, LockAttemptOutcome::HeldByDeadOrAbsent(None));
1493 }
1494
1495 #[test]
1496 fn classify_contention_garbage_body_says_reclaim() {
1497 let outcome = classify_contention(b"not-a-pid\n\0\xff", |_| true);
1498 assert_eq!(outcome, LockAttemptOutcome::HeldByDeadOrAbsent(None));
1499 }
1500
1501 #[test]
1502 fn classify_contention_trims_whitespace() {
1503 let body = b" 789\n";
1504 let outcome = classify_contention(body, |pid| pid == 789);
1505 assert_eq!(outcome, LockAttemptOutcome::HeldByAlive(789));
1506 }
1507
1508 #[test]
1509 fn acquire_relay_lock_stamps_our_pid_into_owner_sidecar() {
1510 use fs2::FileExt;
1511 with_temp_home(|| {
1512 ensure_dirs().unwrap();
1513 let pid = std::process::id();
1514 let lock = acquire_relay_lock(pid).expect("acquire fresh lock");
1515 let body = fs::read(relay_state_lock_owner_path().unwrap()).unwrap();
1518 assert_eq!(
1519 std::str::from_utf8(&body).unwrap().trim(),
1520 pid.to_string(),
1521 "owner sidecar must hold our PID after acquire"
1522 );
1523 let _ = FileExt::unlock(&lock);
1524 drop(lock);
1525 });
1526 }
1527
1528 #[test]
1529 fn acquire_relay_lock_reclaims_when_owner_pid_is_dead() {
1530 with_temp_home(|| {
1531 ensure_dirs().unwrap();
1532 let owner_path = relay_state_lock_owner_path().unwrap();
1539 if let Some(parent) = owner_path.parent() {
1540 fs::create_dir_all(parent).unwrap();
1541 }
1542 fs::write(&owner_path, u32::MAX.to_string()).unwrap();
1543
1544 unsafe { std::env::set_var("WIRE_RELAY_LOCK_TIMEOUT_SECS", "2") };
1547 let started = std::time::Instant::now();
1548 let lock =
1549 acquire_relay_lock(std::process::id()).expect("dead-owner lock must be reclaimed");
1550 assert!(
1551 started.elapsed() < std::time::Duration::from_secs(2),
1552 "reclaim should be fast (well inside timeout); took {:?}",
1553 started.elapsed()
1554 );
1555 drop(lock);
1556 unsafe { std::env::remove_var("WIRE_RELAY_LOCK_TIMEOUT_SECS") };
1557 });
1558 }
1559
1560 #[test]
1561 fn acquire_relay_lock_times_out_when_owner_is_alive() {
1562 use fs2::FileExt;
1563 with_temp_home(|| {
1564 ensure_dirs().unwrap();
1565 let lock_path = relay_state_lock_path().unwrap();
1570 if let Some(parent) = lock_path.parent() {
1571 fs::create_dir_all(parent).unwrap();
1572 }
1573 let holder = fs::OpenOptions::new()
1574 .create(true)
1575 .truncate(false)
1576 .read(true)
1577 .write(true)
1578 .open(&lock_path)
1579 .unwrap();
1580 holder.lock_exclusive().unwrap();
1581 let our_pid = std::process::id();
1585 fs::write(relay_state_lock_owner_path().unwrap(), our_pid.to_string()).unwrap();
1586
1587 unsafe { std::env::set_var("WIRE_RELAY_LOCK_TIMEOUT_SECS", "1") };
1589 let started = std::time::Instant::now();
1590 let result = acquire_relay_lock(our_pid);
1591 let elapsed = started.elapsed();
1592
1593 let _ = FileExt::unlock(&holder);
1596 unsafe { std::env::remove_var("WIRE_RELAY_LOCK_TIMEOUT_SECS") };
1597
1598 let err = result.expect_err("live-owner contention must time out");
1599 let msg = format!("{err}");
1600 assert!(
1601 msg.contains(&our_pid.to_string()),
1602 "timeout error must surface the live holder's PID; got: {msg}"
1603 );
1604 assert!(
1605 elapsed >= std::time::Duration::from_secs(1),
1606 "must respect the bounded timeout; elapsed={elapsed:?}"
1607 );
1608 assert!(
1609 elapsed < std::time::Duration::from_secs(3),
1610 "must not run wildly past the bounded timeout; elapsed={elapsed:?}"
1611 );
1612 });
1613 }
1614}