1use std::path::{Path, PathBuf};
38
39use anyhow::{Context, Result, bail};
40use jiff::Timestamp;
41use serde::{Deserialize, Serialize};
42
43pub const SCHEMA: u32 = 1;
46
47pub const CAP: usize = 200;
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
53#[serde(rename_all = "lowercase")]
54pub enum Severity {
55 Info,
57 Warn,
59 Error,
61}
62
63#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
65#[serde(tag = "kind", rename_all = "lowercase")]
66pub enum Link {
67 Run {
69 id: String,
71 },
72 Task {
74 id: String,
76 },
77 Url {
79 url: String,
81 },
82}
83
84#[derive(Debug, Clone, Serialize, Deserialize)]
86pub struct Notice {
87 pub id: String,
89 pub key: String,
91 pub severity: Severity,
93 pub message: String,
95 #[serde(default)]
97 pub link: Option<Link>,
98 pub first_at: Timestamp,
100 pub last_at: Timestamp,
102 #[serde(default = "one")]
104 pub count: u32,
105 #[serde(default)]
107 pub read_at: Option<Timestamp>,
108 #[serde(default)]
110 pub dismissed_at: Option<Timestamp>,
111 #[serde(default)]
114 pub subjects: Vec<String>,
115 #[serde(default)]
119 pub covered_by: Option<String>,
120 #[serde(default)]
124 pub since: Option<Timestamp>,
125 #[serde(default = "schema")]
127 pub schema: u32,
128}
129
130fn one() -> u32 {
131 1
132}
133
134fn schema() -> u32 {
135 SCHEMA
136}
137
138impl Notice {
139 pub fn new(severity: Severity, key: &str, message: impl Into<String>) -> Self {
141 let now = Timestamp::now();
142 Self {
143 id: id_of(key),
144 key: key.to_owned(),
145 severity,
146 message: message.into(),
147 link: None,
148 first_at: now,
149 last_at: now,
150 count: 1,
151 read_at: None,
152 dismissed_at: None,
153 subjects: Vec::new(),
154 covered_by: None,
155 since: None,
156 schema: SCHEMA,
157 }
158 }
159
160 pub fn about<I, S>(mut self, subjects: I) -> Self
162 where
163 I: IntoIterator<Item = S>,
164 S: Into<String>,
165 {
166 self.subjects = subjects.into_iter().map(Into::into).collect();
167 self
168 }
169
170 pub fn info(key: &str, message: impl Into<String>) -> Self {
172 Self::new(Severity::Info, key, message)
173 }
174
175 pub fn warn(key: &str, message: impl Into<String>) -> Self {
177 Self::new(Severity::Warn, key, message)
178 }
179
180 pub fn error(key: &str, message: impl Into<String>) -> Self {
182 Self::new(Severity::Error, key, message)
183 }
184
185 pub fn since(mut self, at: Option<Timestamp>) -> Self {
187 self.since = at;
188 self
189 }
190
191 pub fn link(mut self, link: Link) -> Self {
193 self.link = Some(link);
194 self
195 }
196
197 pub fn unread(&self) -> bool {
199 self.read_at.is_none() && self.dismissed_at.is_none()
200 }
201
202 pub fn raise_again(&mut self, again: &Notice, now: Timestamp) -> bool {
209 self.count = self.count.saturating_add(1);
210 self.last_at = now;
211 if again.link.is_some() {
212 self.link = again.link.clone();
213 }
214 if !again.subjects.is_empty() {
215 self.subjects = again.subjects.clone();
216 }
217 let escalated = again.severity > self.severity;
218 let newer_cause = match (self.since, again.since) {
224 (Some(old), Some(new)) => new > old,
225 (None, Some(_)) => self.covered_by.is_some() && again.covered_by.is_none(),
226 _ => false,
227 };
228 if again.since.is_some() {
229 self.since = again.since;
230 }
231 let relit = again.message != self.message || escalated || newer_cause;
232 if relit {
233 self.message = again.message.clone();
234 self.severity = self.severity.max(again.severity);
235 self.read_at = None;
236 self.dismissed_at = None;
237 self.covered_by = None;
238 }
239 if let Some(q) = &again.covered_by
242 && !escalated
243 && self.unread()
244 {
245 self.read_at = Some(now);
246 self.covered_by = Some(q.clone());
247 }
248 relit && self.covered_by.is_none()
249 }
250
251 pub fn mark_read(&mut self, now: Timestamp) {
253 if self.read_at.is_none() {
254 self.read_at = Some(now);
255 }
256 }
257
258 pub fn dismiss(&mut self, now: Timestamp) {
260 self.mark_read(now);
261 if self.dismissed_at.is_none() {
262 self.dismissed_at = Some(now);
263 }
264 }
265
266 fn obsolete(&self) -> bool {
269 self.key.starts_with("deputy:")
270 }
271
272 fn visible(&self) -> bool {
274 !self.obsolete() && self.dismissed_at.is_none()
275 }
276
277 fn keep_rank(&self) -> u8 {
279 if self.obsolete() {
280 0
281 } else if self.dismissed_at.is_some() {
282 1
283 } else if self.read_at.is_some() {
284 2
285 } else {
286 3
287 }
288 }
289}
290
291pub fn id_of(key: &str) -> String {
295 let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
296 for b in key.bytes() {
297 hash ^= u64::from(b);
298 hash = hash.wrapping_mul(0x0100_0000_01b3);
299 }
300 let slug: String = key
301 .chars()
302 .map(|c| {
303 if c.is_ascii_alphanumeric() {
304 c.to_ascii_lowercase()
305 } else {
306 '-'
307 }
308 })
309 .take(32)
310 .collect();
311 format!("{slug}-{hash:016x}")
312}
313
314fn valid_id(id: &str) -> bool {
317 !id.is_empty()
318 && id.len() <= 64
319 && id
320 .bytes()
321 .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
322}
323
324const LOCK_WAIT: std::time::Duration = std::time::Duration::from_secs(5);
326
327const LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);
329
330static TMP_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
332
333#[derive(Debug, Clone)]
335pub struct Notices {
336 root: PathBuf,
337}
338
339impl Notices {
340 pub fn open() -> Self {
342 Self::at(crate::run::home().join("notifications"))
343 }
344
345 pub fn at(root: PathBuf) -> Self {
347 Self { root }
348 }
349
350 pub fn root(&self) -> &Path {
352 &self.root
353 }
354
355 fn path_of(&self, id: &str) -> PathBuf {
356 self.root.join(format!("{id}.json"))
357 }
358
359 fn put(&self, n: &Notice) -> Result<()> {
360 std::fs::create_dir_all(&self.root)
361 .with_context(|| format!("create {}", self.root.display()))?;
362 let body = serde_json::to_string_pretty(n).context("serialize notice")?;
363 let path = self.path_of(&n.id);
364 let tmp = path.with_extension(format!(
367 "json.{}.{}.tmp",
368 std::process::id(),
369 TMP_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
370 ));
371 std::fs::write(&tmp, body).with_context(|| format!("write {}", tmp.display()))?;
372 if let Err(e) = std::fs::rename(&tmp, &path) {
373 let _ = std::fs::remove_file(&tmp);
374 return Err(e).with_context(|| format!("replace {}", path.display()));
375 }
376 Ok(())
377 }
378
379 pub fn get(&self, id: &str) -> Result<Notice> {
381 if !valid_id(id) {
382 bail!("`{id}` is not a notification id");
383 }
384 read_path(&self.path_of(id))
385 }
386
387 fn locked<T>(&self, id: &str, f: impl FnOnce() -> Result<T>) -> Result<T> {
396 std::fs::create_dir_all(&self.root)
397 .with_context(|| format!("create {}", self.root.display()))?;
398 let lock = self.root.join(format!("{id}.lock"));
399 let start = std::time::Instant::now();
400 let mut held = false;
401 while start.elapsed() < LOCK_WAIT {
402 match std::fs::OpenOptions::new()
403 .write(true)
404 .create_new(true)
405 .open(&lock)
406 {
407 Ok(_) => {
408 held = true;
409 break;
410 }
411 Err(_) => {
412 let stale = std::fs::metadata(&lock)
413 .and_then(|m| m.modified())
414 .ok()
415 .and_then(|t| t.elapsed().ok())
416 .is_some_and(|age| age > LOCK_STALE);
417 if stale {
418 let _ = std::fs::remove_file(&lock);
419 } else {
420 std::thread::sleep(std::time::Duration::from_millis(5));
421 }
422 }
423 }
424 }
425 let out = f();
426 if held {
427 let _ = std::fs::remove_file(&lock);
428 }
429 out
430 }
431
432 pub fn raise(&self, incoming: Notice) -> Result<Notice> {
434 self.raise_paged(incoming).map(|(n, _)| n)
435 }
436
437 pub fn raise_paged(&self, incoming: Notice) -> Result<(Notice, bool)> {
439 let out = self.locked(&incoming.id.clone(), || {
440 let now = Timestamp::now();
441 let (stored, page) = match read_path(&self.path_of(&incoming.id)) {
442 Ok(mut existing) => {
443 let page = existing.raise_again(&incoming, now);
444 (existing, page)
445 }
446 Err(_) => {
447 let mut fresh = incoming;
448 if fresh.covered_by.is_some() {
449 fresh.read_at = Some(now);
450 }
451 let page = fresh.covered_by.is_none();
452 (fresh, page)
453 }
454 };
455 self.put(&stored)?;
456 Ok((stored, page))
457 })?;
458 self.prune();
459 Ok(out)
460 }
461
462 pub fn cover(&self, question: &crate::ask::Question) {
465 for n in self.all() {
466 if n.unread() && covers(question, &n) {
467 let _ = self.update(&n.id, |n| {
468 if n.unread() && covers(question, n) {
471 n.mark_read(Timestamp::now());
472 n.covered_by = Some(question.id.clone());
473 }
474 });
475 }
476 }
477 }
478
479 fn update(&self, id: &str, change: impl FnOnce(&mut Notice)) -> Result<Notice> {
481 if !valid_id(id) {
482 bail!("`{id}` is not a notification id");
483 }
484 self.locked(id, || {
485 let mut n = read_path(&self.path_of(id))?;
486 change(&mut n);
487 self.put(&n)?;
488 Ok(n)
489 })
490 }
491
492 fn all(&self) -> Vec<Notice> {
494 let mut all: Vec<Notice> = std::fs::read_dir(&self.root)
495 .into_iter()
496 .flatten()
497 .flatten()
498 .map(|e| e.path())
499 .filter(|p| p.extension().is_some_and(|x| x == "json"))
500 .filter_map(|p| read_path(&p).ok())
501 .collect();
502 all.sort_by(|a, b| b.last_at.cmp(&a.last_at).then_with(|| a.id.cmp(&b.id)));
503 all
504 }
505
506 pub fn list(&self) -> Vec<Notice> {
508 self.all().into_iter().filter(Notice::visible).collect()
509 }
510
511 pub fn count_unread(&self) -> usize {
513 self.list().iter().filter(|n| n.unread()).count()
514 }
515
516 pub fn mark_read(&self, id: &str) -> Result<Notice> {
518 let now = Timestamp::now();
519 self.update(id, |n| n.mark_read(now))
520 }
521
522 pub fn mark_all_read(&self) -> Result<usize> {
524 let now = Timestamp::now();
525 let mut changed = 0;
526 for n in self.list().into_iter().filter(Notice::unread) {
527 let done = self.update(&n.id, |n| n.mark_read(now));
530 if done.is_ok() {
531 changed += 1;
532 }
533 }
534 Ok(changed)
535 }
536
537 pub fn dismiss(&self, id: &str) -> Result<Notice> {
539 let now = Timestamp::now();
540 self.update(id, |n| n.dismiss(now))
541 }
542
543 fn prune(&self) {
545 let mut all = self.all();
546 if all.len() <= CAP {
547 return;
548 }
549 all.sort_by(|a, b| {
551 b.keep_rank()
552 .cmp(&a.keep_rank())
553 .then_with(|| b.last_at.cmp(&a.last_at))
554 });
555 for n in all.split_off(CAP) {
556 let _ = std::fs::remove_file(self.path_of(&n.id));
557 }
558 }
559
560 pub fn revision(&self) -> u64 {
564 use std::hash::{Hash as _, Hasher as _};
565 let mut entries: Vec<(std::ffi::OsString, u128)> = std::fs::read_dir(&self.root)
566 .into_iter()
567 .flatten()
568 .flatten()
569 .filter(|e| e.path().extension().is_some_and(|x| x == "json"))
570 .map(|e| {
571 let at = e
572 .metadata()
573 .and_then(|m| m.modified())
574 .ok()
575 .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
576 .map_or(0, |d| d.as_nanos());
577 (e.file_name(), at)
578 })
579 .collect();
580 if entries.is_empty() {
581 return 0;
582 }
583 entries.sort();
584 let mut h = std::collections::hash_map::DefaultHasher::new();
585 entries.hash(&mut h);
586 h.finish()
587 }
588}
589
590fn read_path(path: &Path) -> Result<Notice> {
591 let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
592 let n: Notice =
593 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
594 if n.schema > SCHEMA {
595 bail!(
596 "notice {} was written by a newer magi (schema {}, this build speaks up to {SCHEMA})",
597 n.id,
598 n.schema
599 );
600 }
601 Ok(n)
602}
603
604pub fn run_ended(state: &crate::run::RunState) -> Option<Notice> {
610 use crate::run::RunStatus;
611 let failed = matches!(
612 state.status,
613 RunStatus::Blocked | RunStatus::Stalled | RunStatus::Failed
614 );
615 (failed && !state.parked).then(|| {
616 Notice::error(
617 &format!("run:{}", state.id),
618 format!("Run {} ended {}.", state.short(), state.status.as_str()),
619 )
620 .link(Link::Run {
621 id: state.id.clone(),
622 })
623 .about([state.id.clone()])
624 })
625}
626
627pub fn merged_red(run_id: &str, summary: &str) -> Notice {
633 Notice::warn(&format!("merged-red:{run_id}"), summary).link(Link::Run {
634 id: run_id.to_owned(),
635 })
636}
637
638pub fn task_held(task: &crate::queue::Task) -> Option<Notice> {
652 use crate::queue::{HoldSource, TaskStatus};
653 if task.status != TaskStatus::Held || task.hold_source != Some(HoldSource::Machine) {
654 return None;
655 }
656 let why = task
657 .hold_reason
658 .as_deref()
659 .or(task.last_error.as_deref())
660 .unwrap_or("no reason recorded");
661 Some(
662 Notice::warn(
663 &format!("task:{}", task.id),
664 format!("Task {} is held: {why}.", task.short()),
665 )
666 .link(Link::Task {
667 id: task.id.clone(),
668 })
669 .about([task.id.clone()])
670 .since(task.held_at),
671 )
672}
673
674pub fn run_stopped(id: &str, state: &crate::run::RunState) -> Notice {
676 Notice::error(
677 &format!("run:{id}"),
678 format!("Run {} stopped with an error.", state.short()),
679 )
680 .link(Link::Run { id: id.to_owned() })
681 .about([id.to_owned()])
682}
683
684#[derive(Debug, Clone, PartialEq, Eq)]
686pub struct Page {
687 pub run: String,
689 pub summary: String,
691}
692
693static PAGER: std::sync::OnceLock<()> = std::sync::OnceLock::new();
696
697pub fn install_pager() {
699 let _ = PAGER.set(());
700}
701
702fn send_page(page: Page, notify: Option<crate::config::Notify>) {
707 if PAGER.get().is_none() {
708 return;
709 }
710 let spawned = std::thread::Builder::new()
711 .name("notice-pager".into())
712 .spawn(move || {
713 let notify = notify.unwrap_or_else(machine_notify);
714 if notify.command.is_empty() {
715 return;
716 }
717 let rt = match tokio::runtime::Builder::new_current_thread()
718 .enable_all()
719 .build()
720 {
721 Ok(rt) => rt,
722 Err(e) => return tracing::warn!("could not page about a notification: {e:#}"),
723 };
724 if let Err(e) = rt.block_on(crate::ask::notify_text(¬ify, &page.run, &page.summary))
725 {
726 tracing::warn!("could not page about a notification: {e:#}");
727 }
728 });
729 if let Err(e) = spawned {
730 tracing::warn!("could not page about a notification: {e:#}");
731 }
732}
733
734fn machine_notify() -> crate::config::Notify {
737 let layers: Vec<PathBuf> = crate::config::Config::machine_layer()
738 .into_iter()
739 .filter(|p| p.is_file())
740 .collect();
741 match crate::config::Config::load_layers(&layers) {
742 Ok(c) => c.notify,
743 Err(e) => {
744 tracing::warn!("could not read [notify] for a notification: {e:#}");
745 Default::default()
746 }
747 }
748}
749
750pub fn raise(notice: Notice) {
754 if let Some(home) = crate::run::try_home() {
755 raise_in(&home, notice);
756 }
757}
758
759pub fn raise_with(notice: Notice, notify: &crate::config::Notify) {
761 if let Some(home) = crate::run::try_home() {
762 raise_in_with(&home, notice, &|page| send_page(page, Some(notify.clone())));
763 }
764}
765
766pub fn covers(q: &crate::ask::Question, n: &Notice) -> bool {
783 if !q.status.open() || !n.subjects.contains(&q.run) {
784 return false;
785 }
786 if n.since.is_some_and(|since| q.asked_at < since) {
787 return false;
788 }
789 let about_task = n.key.starts_with("task:") || n.key.starts_with("handover:");
790 let about_run = n.key.starts_with("run:");
791 match q.node.as_str() {
792 crate::bump::NOTICE_NODE | crate::land::APPROVAL_NODE => false,
793 _ if q.run_names_task() => about_task,
794 _ => about_run,
795 }
796}
797
798fn covering_question(home: &Path, n: &Notice) -> Option<String> {
799 if n.subjects.is_empty() {
800 return None;
801 }
802 crate::ask::Questions::at(home.join("questions"))
803 .list()
804 .into_iter()
805 .find(|q| covers(q, n))
806 .map(|q| q.id)
807}
808
809pub fn quiet_for(home: &Path, question: &crate::ask::Question) {
811 Notices::at(home.join("notifications")).cover(question);
812}
813
814pub fn raise_in(home: &Path, notice: Notice) {
817 raise_in_with(home, notice, &|page| send_page(page, None));
818}
819
820pub fn raise_in_with(home: &Path, mut notice: Notice, send: &dyn Fn(Page)) {
823 if let Some(q) = covering_question(home, ¬ice) {
824 notice.covered_by = Some(q);
825 }
826 match Notices::at(home.join("notifications")).raise_paged(notice) {
827 Ok((n, true)) => send(Page {
828 run: match &n.link {
829 Some(Link::Run { id }) => id.clone(),
830 _ => String::new(),
831 },
832 summary: n.message.clone(),
833 }),
834 Ok(_) => {}
835 Err(e) => tracing::warn!("could not file a notification: {e:#}"),
836 }
837}
838
839#[cfg(test)]
840mod tests {
841 use super::*;
842
843 #[test]
844 fn legacy_deputy_warnings_are_hidden_without_hiding_other_namespaces() {
845 let (_dir, s) = store();
846 for key in ["deputy:open", "deputy:answered", "deputy:deleted"] {
848 s.raise(Notice::warn(key, "old exhausted warning")).unwrap();
849 }
850 let keys = ["deputy-turn:open", "deputyish:open", "deputy", "task:open"];
851 for key in keys {
852 s.raise(Notice::warn(key, "still relevant")).unwrap();
853 }
854 assert_eq!(s.list().len(), keys.len());
855 assert_eq!(s.count_unread(), keys.len());
856 assert!(s.list().iter().all(|n| keys.contains(&n.key.as_str())));
857 assert_eq!(s.mark_all_read().unwrap(), keys.len());
858 assert_eq!(s.count_unread(), 0);
859 assert!(s.get(&id_of("deputy:open")).unwrap().unread());
860 }
861
862 #[test]
863 fn obsolete_deputy_warnings_are_pruned_before_current_notices() {
864 let (_dir, s) = store();
865 for i in 0..CAP {
866 let id = s
867 .raise(Notice::info(&format!("current:{i}"), "keep"))
868 .unwrap()
869 .id;
870 s.dismiss(&id).unwrap();
871 }
872 s.raise(Notice::warn("deputy:old", "obsolete")).unwrap();
874 assert!(s.get(&id_of("deputy:old")).is_err());
875 assert_eq!(s.all().len(), CAP);
876 assert!(s.all().iter().all(|n| n.key.starts_with("current:")));
877 }
878
879 fn store() -> (tempfile::TempDir, Notices) {
880 let dir = tempfile::tempdir().unwrap();
881 let s = Notices::at(dir.path().join("notifications"));
882 (dir, s)
883 }
884
885 #[test]
886 fn a_red_merge_notice_is_keyed_on_the_run_and_raised_once() {
887 let (_d, s) = store();
888 let n = merged_red("run-1", "Merged a/b PR #1 with red checks: x (u)");
889 assert_eq!(n.key, "merged-red:run-1");
890 assert_eq!(
891 n.link,
892 Some(Link::Run {
893 id: "run-1".to_owned()
894 })
895 );
896 s.raise(merged_red(
897 "run-1",
898 "Merged a/b PR #1 with red checks: x (u)",
899 ))
900 .unwrap();
901 s.raise(merged_red(
902 "run-1",
903 "Merged a/b PR #1 with red checks: x (u)",
904 ))
905 .unwrap();
906 assert_eq!(s.list().len(), 1);
907 }
908
909 #[test]
910 fn a_repeat_counts_but_stays_read_and_a_change_relights_it() {
911 let now = Timestamp::now();
912 let mut n = Notice::warn("task:1", "held");
913 n.mark_read(now);
914 n.raise_again(&Notice::warn("task:1", "held"), now);
915 assert_eq!(n.count, 2);
916 assert!(n.read_at.is_some(), "identical repeat stays read");
917 n.raise_again(&Notice::warn("task:1", "held differently"), now);
918 assert!(n.unread());
919 assert_eq!(n.message, "held differently");
920 }
921
922 #[test]
923 fn a_rise_in_severity_resurrects_a_dismissed_notice_but_a_repeat_does_not() {
924 let now = Timestamp::now();
925 let mut n = Notice::warn("task:x", "m");
926 n.dismiss(now);
927 n.raise_again(&Notice::warn("task:x", "m"), now);
928 assert!(n.dismissed_at.is_some(), "tombstone holds");
929 n.raise_again(&Notice::error("k", "m"), now);
930 assert!(n.unread());
931 assert_eq!(n.severity, Severity::Error);
932 n.raise_again(&Notice::info("k", "other"), now);
934 assert_eq!(n.severity, Severity::Error);
935 }
936
937 #[test]
938 fn adding_link_on_repeat_updates_link_without_relighting_read_or_dismissed() {
939 let now = Timestamp::now();
940 let mut read_notice = Notice::warn("task:1", "held");
942 read_notice.mark_read(now);
943 let link_to_add = Link::Task { id: "1".to_owned() };
944 let again = Notice::warn("task:1", "held").link(link_to_add.clone());
945 let relit = read_notice.raise_again(&again, now);
946 assert!(!relit, "link-only update must not page/relight");
947 assert!(read_notice.read_at.is_some(), "read notice stays read");
948 assert_eq!(read_notice.link, Some(link_to_add.clone()));
949
950 let mut dismissed_notice = Notice::warn("task:2", "held");
952 dismissed_notice.dismiss(now);
953 let link_to_add_2 = Link::Task { id: "2".to_owned() };
954 let again_2 = Notice::warn("task:2", "held").link(link_to_add_2.clone());
955 let relit_2 = dismissed_notice.raise_again(&again_2, now);
956 assert!(!relit_2, "link-only update must not page/relight");
957 assert!(dismissed_notice.dismissed_at.is_some(), "tombstone holds");
958 assert_eq!(dismissed_notice.link, Some(link_to_add_2));
959 }
960
961 #[test]
962 fn identical_raises_share_one_file() {
963 let (_d, s) = store();
964 for _ in 0..5 {
965 s.raise(Notice::error("run:abc", "blocked")).unwrap();
966 }
967 let all = s.list();
968 assert_eq!(all.len(), 1);
969 assert_eq!(all[0].count, 5);
970 assert_eq!(s.count_unread(), 1);
971 }
972
973 #[test]
974 fn transitions_persist_and_dismissed_leave_the_list() {
975 let (_d, s) = store();
976 let a = s.raise(Notice::info("a", "one")).unwrap();
977 let b = s.raise(Notice::warn("b", "two")).unwrap();
978 assert_eq!(s.count_unread(), 2);
979 s.mark_read(&a.id).unwrap();
980 assert_eq!(s.count_unread(), 1);
981 s.dismiss(&b.id).unwrap();
982 assert_eq!(s.count_unread(), 0);
983 assert_eq!(s.list().len(), 1);
984 s.raise(Notice::warn("b", "two")).unwrap();
985 assert_eq!(
986 s.list().len(),
987 1,
988 "a tombstone survives a same-message raise"
989 );
990 s.raise(Notice::info("c", "three")).unwrap();
991 assert_eq!(s.mark_all_read().unwrap(), 1);
992 assert_eq!(s.count_unread(), 0);
993 }
994
995 #[test]
996 fn the_list_is_newest_first() {
997 let (_d, s) = store();
998 let mut old = Notice::info("old", "old");
999 old.last_at = "2020-01-01T00:00:00Z".parse().unwrap();
1000 s.put(&old).unwrap();
1001 s.raise(Notice::info("new", "new")).unwrap();
1002 let keys: Vec<_> = s.list().into_iter().map(|n| n.key).collect();
1003 assert_eq!(keys, ["new", "old"]);
1004 }
1005
1006 #[test]
1007 fn the_cap_prunes_dismissed_then_read_then_oldest() {
1008 let (_d, s) = store();
1009 let keep = s.raise(Notice::error("keep", "unread")).unwrap();
1010 let gone = s.raise(Notice::info("gone", "dismissed")).unwrap();
1011 s.dismiss(&gone.id).unwrap();
1012 for i in 0..CAP - 1 {
1013 s.raise(Notice::info(&format!("k{i}"), "x")).unwrap();
1014 }
1015 let files = std::fs::read_dir(s.root()).unwrap().count();
1016 assert_eq!(files, CAP);
1017 assert!(
1018 s.get(&keep.id).is_ok(),
1019 "an unread notice outlives a dismissed one"
1020 );
1021 assert!(s.get(&gone.id).is_err(), "the dismissed one went first");
1022 }
1023
1024 #[test]
1025 fn revision_moves_on_write_and_on_removal() {
1026 let (_d, s) = store();
1027 assert_eq!(s.revision(), 0);
1028 let a = s.raise(Notice::info("a", "m")).unwrap();
1029 let r1 = s.revision();
1030 assert_ne!(r1, 0);
1031 s.raise(Notice::info("b", "m")).unwrap();
1032 let r2 = s.revision();
1033 assert_ne!(r1, r2);
1034 std::fs::remove_file(s.path_of(&a.id)).unwrap();
1035 assert_ne!(s.revision(), r2);
1036 }
1037
1038 #[test]
1039 fn ids_are_stable_and_untrusted_ids_are_refused() {
1040 assert_eq!(id_of("run:1"), id_of("run:1"));
1041 assert_ne!(id_of("run:1"), id_of("run-1"));
1042 assert!(valid_id(&id_of("release-bump:20260101-abc")));
1043 let (_d, s) = store();
1044 for bad in ["", "../x", "a/b", "A", "x.json"] {
1045 assert!(s.get(bad).is_err(), "{bad}");
1046 }
1047 }
1048
1049 #[test]
1050 fn a_newer_schema_is_refused_and_an_older_reads() {
1051 let (_d, s) = store();
1052 let mut n = Notice::info("k", "m");
1053 n.schema = SCHEMA + 1;
1054 s.put(&n).unwrap();
1055 assert!(s.get(&n.id).is_err());
1056 let old = r#"{"id":"x-1","key":"x","severity":"warn","message":"m",
1057 "first_at":"2020-01-01T00:00:00Z","last_at":"2020-01-01T00:00:00Z"}"#;
1058 std::fs::write(s.path_of("x-1"), old).unwrap();
1059 assert_eq!(s.get("x-1").unwrap().count, 1);
1060 }
1061
1062 fn state(status: crate::run::RunStatus) -> crate::run::RunState {
1063 let mut st = crate::run::RunState::new(
1064 std::path::PathBuf::from("/repo"),
1065 "main".to_owned(),
1066 "abc1234def".to_owned(),
1067 "task".to_owned(),
1068 crate::config::Config::default(),
1069 );
1070 st.status = status;
1071 st
1072 }
1073
1074 #[test]
1075 fn a_run_that_ended_badly_is_news_unless_it_only_parked() {
1076 use crate::run::RunStatus;
1077 for bad in [RunStatus::Blocked, RunStatus::Stalled, RunStatus::Failed] {
1078 let st = state(bad);
1079 let n = run_ended(&st).expect("news");
1080 assert_eq!(n.key, format!("run:{}", st.id));
1081 assert_eq!(n.severity, Severity::Error);
1082 }
1083 assert!(run_ended(&state(RunStatus::Merged)).is_none());
1084 let mut parked = state(RunStatus::Stalled);
1085 parked.parked = true;
1086 assert!(run_ended(&parked).is_none());
1087 }
1088
1089 #[test]
1090 fn concurrent_writers_of_one_key_all_succeed() {
1091 let (_d, s) = store();
1092 let handles: Vec<_> = (0..8)
1093 .map(|_| {
1094 let s = s.clone();
1095 std::thread::spawn(move || {
1096 for _ in 0..20 {
1097 s.raise(Notice::warn("same", "m")).unwrap();
1098 }
1099 })
1100 })
1101 .collect();
1102 for h in handles {
1103 h.join().unwrap();
1104 }
1105 assert_eq!(s.list().len(), 1);
1106 let stray = std::fs::read_dir(s.root())
1107 .unwrap()
1108 .flatten()
1109 .filter(|e| e.path().extension().is_some_and(|x| x == "tmp"))
1110 .count();
1111 assert_eq!(stray, 0);
1112 }
1113
1114 #[test]
1115 fn a_dead_holders_lock_is_broken_and_no_lock_is_left_behind() {
1116 let (_d, s) = store();
1117 let n = s.raise(Notice::info("k", "m")).unwrap();
1118 let lock = s.root().join(format!("{}.lock", n.id));
1119 std::fs::write(&lock, "").unwrap();
1120 let old = std::time::SystemTime::now() - std::time::Duration::from_secs(60);
1121 std::fs::File::options()
1122 .write(true)
1123 .open(&lock)
1124 .unwrap()
1125 .set_modified(old)
1126 .unwrap();
1127 s.mark_read(&n.id).unwrap();
1128 assert!(!lock.exists());
1129 assert!(s.get(&n.id).unwrap().read_at.is_some());
1130 }
1131
1132 #[test]
1133 fn only_a_machine_hold_is_a_task_notice() {
1134 let mut t = crate::queue::Task::new(
1135 "t".to_owned(),
1136 "do it".to_owned(),
1137 std::path::PathBuf::from("/repo"),
1138 crate::queue::Source::Human,
1139 );
1140 assert!(task_held(&t).is_none());
1141 t.hold_manual(None);
1142 assert!(task_held(&t).is_none(), "the operator's own hold");
1143 t.hold_machine(Some("missing blocker".to_owned()));
1144 let n = task_held(&t).expect("machine hold");
1145 assert_eq!(n.key, format!("task:{}", t.id));
1146 assert!(n.message.contains("missing blocker"));
1147 }
1148
1149 fn held_task() -> crate::queue::Task {
1150 let mut t = crate::queue::Task::new(
1151 "t".to_owned(),
1152 "do it".to_owned(),
1153 std::path::PathBuf::from("/repo"),
1154 crate::queue::Source::Human,
1155 );
1156 t.hold_machine(Some("branch b is checked out".to_owned()));
1157 t
1158 }
1159
1160 fn ask(home: &Path, run: &str, node: &str) -> crate::ask::Question {
1161 let mut q = crate::ask::Question::new(
1162 run.to_owned(),
1163 node.to_owned(),
1164 "conductor".to_owned(),
1165 "cannot resume".to_owned(),
1166 String::new(),
1167 Vec::new(),
1168 );
1169 crate::ask::Questions::at(home.join("questions"))
1170 .put(&mut q)
1171 .unwrap();
1172 q
1173 }
1174
1175 fn unread(home: &Path) -> usize {
1176 Notices::at(home.join("notifications")).count_unread()
1177 }
1178
1179 #[test]
1180 fn a_hold_with_an_open_question_for_the_task_pages_once() {
1181 let dir = tempfile::tempdir().unwrap();
1182 let mut t = held_task();
1183 let q = ask(dir.path(), &t.id, "conduct");
1184 crate::queue::Queue::at(dir.path().join("queue"))
1185 .put(&mut t)
1186 .unwrap();
1187 let s = Notices::at(dir.path().join("notifications"));
1188 assert_eq!(s.count_unread(), 0);
1189 let n = s.list().pop().expect("the record is kept");
1190 assert_eq!(n.covered_by.as_deref(), Some(q.id.as_str()));
1191 assert!(n.message.contains("checked out"));
1192 }
1193
1194 #[test]
1195 fn a_refused_handover_without_a_question_pages_once() {
1196 let dir = tempfile::tempdir().unwrap();
1197 let q = crate::queue::Queue::at(dir.path().join("queue"));
1198 let mut t = held_task();
1199 t.hold_for_handover(Some("magi/x/A".to_owned()), "checked out".to_owned());
1200 q.put(&mut t).unwrap();
1201 q.put(&mut t).unwrap();
1202 let s = Notices::at(dir.path().join("notifications"));
1203 assert_eq!(s.list().len(), 1);
1204 assert_eq!(s.list()[0].key, format!("task:{}", t.id));
1205 assert_eq!(s.count_unread(), 1);
1206 }
1207
1208 #[test]
1209 fn an_older_question_does_not_silence_a_new_hold() {
1210 for node in ["deps", "conduct", "triage"] {
1211 let dir = tempfile::tempdir().unwrap();
1212 let mut t = crate::queue::Task::new(
1213 "t".to_owned(),
1214 "do it".to_owned(),
1215 std::path::PathBuf::from("/repo"),
1216 crate::queue::Source::Human,
1217 );
1218 let old = ask(dir.path(), &t.id, node);
1219 std::thread::sleep(std::time::Duration::from_millis(5));
1220 t.hold_machine(Some("new failure".to_owned()));
1221 crate::queue::Queue::at(dir.path().join("queue"))
1222 .put(&mut t)
1223 .unwrap();
1224 let s = Notices::at(dir.path().join("notifications"));
1225 let n = s.list().pop().unwrap();
1226 assert_eq!(n.covered_by, None, "{node} question {}", old.id);
1227 assert_eq!(s.count_unread(), 1);
1228 }
1229 }
1230
1231 #[test]
1232 fn a_legacy_covered_notice_pages_for_a_new_uncovered_hold() {
1233 let (_d, s) = store();
1234 let mut legacy = Notice::warn("task:x", "held");
1235 legacy.covered_by = Some("q1".to_owned());
1236 let id = legacy.id.clone();
1237 s.raise(legacy).unwrap();
1238 assert!(!s.get(&id).unwrap().unread());
1239 s.raise(Notice::warn("task:x", "held").since(Some(Timestamp::now())))
1240 .unwrap();
1241 let n = s.get(&id).unwrap();
1242 assert!(n.unread());
1243 assert_eq!(n.covered_by, None);
1244 }
1245
1246 #[test]
1247 fn a_newer_cause_relights_a_dismissed_notice() {
1248 let (_d, s) = store();
1249 let t0 = Timestamp::now();
1250 let first = Notice::warn("task:x", "held").since(Some(t0));
1251 let id = first.id.clone();
1252 s.raise(first).unwrap();
1253 s.dismiss(&id).unwrap();
1254 s.raise(Notice::warn("task:x", "held").since(Some(t0)))
1255 .unwrap();
1256 assert!(!s.get(&id).unwrap().unread(), "same cause stays dismissed");
1257 let later = t0 + std::time::Duration::from_secs(1);
1258 s.raise(Notice::warn("task:x", "held").since(Some(later)))
1259 .unwrap();
1260 assert!(s.get(&id).unwrap().unread());
1261 }
1262
1263 #[test]
1264 fn a_hold_with_no_question_still_notifies() {
1265 let dir = tempfile::tempdir().unwrap();
1266 let mut t = held_task();
1267 crate::queue::Queue::at(dir.path().join("queue"))
1268 .put(&mut t)
1269 .unwrap();
1270 assert_eq!(unread(dir.path()), 1);
1271 }
1272
1273 #[test]
1274 fn a_question_for_another_task_does_not_suppress() {
1275 let dir = tempfile::tempdir().unwrap();
1276 let mut t = held_task();
1277 ask(dir.path(), "some-other-task", "conduct");
1278 crate::queue::Queue::at(dir.path().join("queue"))
1279 .put(&mut t)
1280 .unwrap();
1281 assert_eq!(unread(dir.path()), 1);
1282 }
1283
1284 #[test]
1285 fn a_question_filed_after_the_hold_quiets_it_once() {
1286 let dir = tempfile::tempdir().unwrap();
1287 let mut t = held_task();
1288 crate::queue::Queue::at(dir.path().join("queue"))
1289 .put(&mut t)
1290 .unwrap();
1291 assert_eq!(unread(dir.path()), 1);
1292 let q = ask(dir.path(), &t.id, "conduct");
1293 assert_eq!(unread(dir.path()), 0);
1294 let s = Notices::at(dir.path().join("notifications"));
1296 let id = s.list()[0].id.clone();
1297 s.update(&id, |n| n.read_at = None).unwrap();
1298 let mut again = crate::ask::Questions::at(dir.path().join("questions"))
1299 .get(&q.id)
1300 .unwrap();
1301 crate::ask::Questions::at(dir.path().join("questions"))
1302 .put(&mut again)
1303 .unwrap();
1304 assert_eq!(unread(dir.path()), 1);
1305 }
1306
1307 #[test]
1308 fn a_blocked_run_with_an_open_question_is_quiet() {
1309 let dir = tempfile::tempdir().unwrap();
1310 ask(dir.path(), "run-1", "implement");
1311 raise_in(
1312 dir.path(),
1313 Notice::error("run:run-1", "Run r ended blocked.").about(["run-1"]),
1314 );
1315 assert_eq!(unread(dir.path()), 0);
1316 raise_in(
1317 dir.path(),
1318 Notice::error("run:run-2", "Run r ended blocked.").about(["run-2"]),
1319 );
1320 assert_eq!(unread(dir.path()), 1);
1321 }
1322
1323 #[test]
1324 fn escalation_makes_a_covered_notice_unread_again() {
1325 let dir = tempfile::tempdir().unwrap();
1326 ask(dir.path(), "t1", "conduct");
1327 raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
1328 assert_eq!(unread(dir.path()), 0);
1329 raise_in(dir.path(), Notice::error("task:t1", "held").about(["t1"]));
1330 assert_eq!(unread(dir.path()), 1);
1331 }
1332
1333 #[test]
1334 fn covers_matches_run_exactly_and_ignores_release_questions() {
1335 let dir = tempfile::tempdir().unwrap();
1336 let n = Notice::warn("task:x", "m").about(["t1"]);
1337 let q = ask(dir.path(), "t1", "conduct");
1338 assert!(covers(&q, &n));
1339 assert!(!covers(&q, &Notice::warn("task:x", "m")));
1340 assert!(!covers(&q, &Notice::warn("task:x", "m").about(["t"])));
1341 let release = ask(dir.path(), "t1", crate::bump::NOTICE_NODE);
1342 assert!(!covers(&release, &n));
1343 }
1344
1345 #[test]
1346 fn a_run_question_does_not_swallow_a_task_hold_and_vice_versa() {
1347 let dir = tempfile::tempdir().unwrap();
1348 ask(dir.path(), "t1", "implement");
1349 raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
1350 assert_eq!(unread(dir.path()), 1);
1351 ask(dir.path(), "r1", "conduct");
1352 raise_in(dir.path(), Notice::error("run:r1", "ended").about(["r1"]));
1353 assert_eq!(unread(dir.path()), 2);
1354 }
1355
1356 #[test]
1357 fn a_conduct_question_covers_the_hold_however_late_it_was_filed() {
1358 let dir = tempfile::tempdir().unwrap();
1359 let mut q = crate::ask::Question::new(
1360 "t1".to_owned(),
1361 "conduct".to_owned(),
1362 "conductor".to_owned(),
1363 "s".to_owned(),
1364 String::new(),
1365 Vec::new(),
1366 );
1367 q.asked_at = Timestamp::from_second(Timestamp::now().as_second() - 3600).unwrap();
1368 crate::ask::Questions::at(dir.path().join("questions"))
1369 .put(&mut q)
1370 .unwrap();
1371 raise_in(dir.path(), Notice::warn("task:t1", "held").about(["t1"]));
1372 assert_eq!(unread(dir.path()), 0);
1373 }
1374
1375 fn pages(home: &Path, n: Notice) -> Vec<Page> {
1376 let sent = std::cell::RefCell::new(Vec::new());
1377 raise_in_with(home, n, &|p| sent.borrow_mut().push(p));
1378 sent.into_inner()
1379 }
1380
1381 #[test]
1382 fn a_page_fires_on_new_changed_and_escalated_only() {
1383 let dir = tempfile::tempdir().unwrap();
1384 let h = dir.path();
1385 let n = || Notice::warn("run:p1", "stopped").link(Link::Run { id: "p1".into() });
1386 let first = pages(h, n());
1387 assert_eq!(
1388 first,
1389 vec![Page {
1390 run: "p1".into(),
1391 summary: "stopped".into()
1392 }]
1393 );
1394 assert!(pages(h, n()).is_empty(), "identical re-raise is silent");
1395 assert_eq!(pages(h, Notice::warn("run:p1", "stopped again")).len(), 1);
1396 assert_eq!(pages(h, Notice::error("run:p1", "stopped again")).len(), 1);
1397 assert!(pages(h, Notice::error("run:p1", "stopped again")).is_empty());
1398 }
1399
1400 #[test]
1401 fn a_non_run_link_leaves_run_empty() {
1402 let dir = tempfile::tempdir().unwrap();
1403 let n = Notice::warn("task:t1", "held").link(Link::Task { id: "t1".into() });
1404 assert_eq!(pages(dir.path(), n)[0].run, "");
1405 }
1406
1407 #[test]
1408 fn a_dismissed_tombstone_stays_silent_until_it_changes() {
1409 let dir = tempfile::tempdir().unwrap();
1410 let h = dir.path();
1411 pages(h, Notice::warn("disk:r", "low"));
1412 Notices::at(h.join("notifications"))
1413 .dismiss(&id_of("disk:r"))
1414 .unwrap();
1415 assert!(pages(h, Notice::warn("disk:r", "low")).is_empty());
1416 assert_eq!(pages(h, Notice::warn("disk:r", "lower")).len(), 1);
1417 }
1418
1419 #[test]
1420 fn a_newer_cause_pages_again() {
1421 let dir = tempfile::tempdir().unwrap();
1422 let h = dir.path();
1423 let at = |s: i64| Some(Timestamp::from_second(s).unwrap());
1424 let n = |s| Notice::warn("task:t2", "held").since(at(s));
1425 assert_eq!(pages(h, n(100)).len(), 1);
1426 assert!(pages(h, n(100)).is_empty());
1427 assert_eq!(pages(h, n(200)).len(), 1);
1428 }
1429
1430 #[test]
1431 fn a_covered_notice_does_not_page() {
1432 let dir = tempfile::tempdir().unwrap();
1433 let h = dir.path();
1434 let q = ask(h, "r9", "implement");
1435 let n = || Notice::error("run:r9", "ended").about(["r9".to_owned()]);
1436 assert!(pages(h, n()).is_empty());
1437 let stored = Notices::at(h.join("notifications"))
1438 .get(&id_of("run:r9"))
1439 .unwrap();
1440 assert_eq!(stored.covered_by.as_deref(), Some(q.id.as_str()));
1441 assert!(pages(h, n()).is_empty());
1442 }
1443
1444 #[test]
1445 fn a_failing_sender_path_never_propagates() {
1446 let notify = crate::config::Notify {
1448 command: vec!["magi-no-such-program-xyz".into()],
1449 };
1450 let rt = tokio::runtime::Builder::new_current_thread()
1451 .enable_all()
1452 .build()
1453 .unwrap();
1454 let r = rt.block_on(crate::ask::notify_text(¬ify, "", "x"));
1455 assert!(r.is_err());
1456 send_page(
1458 Page {
1459 run: String::new(),
1460 summary: "x".into(),
1461 },
1462 Some(notify),
1463 );
1464 }
1465}