vissue_core/
events.rs

1//! Append-only change log and generation counter, so a poller notices a write
2//! without re-parsing every issues.org.
3//!
4//! Two files sit beside the project directories:
5//!
6//! - `.vault-events.gen`, a monotonic counter; comparing it against the last
7//!   value seen is the cheap change check.
8//! - `.vault-events.jsonl`, one event per line, read with a `since` sequence.
9//!
10//! The names are fixed rather than derived from the product name: existing
11//! readers watch these paths, and renaming them would silence every poller.
12//!
13//! Emission is best-effort and never fails a write. Set `VISSUE_EVENTS=0` to
14//! turn it off, for a caller that needs the tracker to stay
15//! untouched.
16
17use anyhow::Context;
18
19use crate::error::{Error, Result};
20use fs2::FileExt;
21use serde::{Deserialize, Serialize};
22use std::collections::HashMap;
23use std::fs::{self, OpenOptions};
24use std::io::Write;
25use std::path::{Path, PathBuf};
26use std::sync::{
27    Arc, Mutex, OnceLock,
28    atomic::{AtomicU64, Ordering},
29};
30use std::time::{SystemTime, UNIX_EPOCH};
31
32use crate::config::Layout;
33
34static LAST_SEQ: AtomicU64 = AtomicU64::new(0);
35static LAST_EMIT_MS_BY_DIR: OnceLock<Mutex<HashMap<PathBuf, u64>>> = OnceLock::new();
36static PROCESS_LOCKS: OnceLock<Mutex<HashMap<PathBuf, Arc<Mutex<()>>>>> = OnceLock::new();
37
38const LOCK_NAME: &str = ".vault-events.lock";
39
40fn with_events_lock<R, F>(dir: &Path, f: F) -> Result<R>
41where
42    F: FnOnce() -> Result<R>,
43{
44    let key = dir.to_path_buf();
45    let mutex = {
46        let mut map = PROCESS_LOCKS
47            .get_or_init(|| Mutex::new(HashMap::new()))
48            .lock()
49            .unwrap_or_else(|p| p.into_inner());
50        map.entry(key)
51            .or_insert_with(|| Arc::new(Mutex::new(())))
52            .clone()
53    };
54    let _proc = mutex.lock().unwrap_or_else(|p| p.into_inner());
55    fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
56    let lock_path = dir.join(LOCK_NAME);
57    let file = OpenOptions::new()
58        .create(true)
59        .read(true)
60        .append(true)
61        .open(&lock_path)
62        .with_context(|| format!("open {}", lock_path.display()))?;
63    file.lock_exclusive()
64        .with_context(|| format!("lock {}", lock_path.display()))?;
65    let result = f();
66    let _ = FileExt::unlock(&file);
67    let _ = file;
68    result
69}
70
71/// Repeated writes inside this window bump the generation without appending a
72/// line, so an editor or agent saving in a burst does not flood the log.
73const DEBOUNCE_MS: u64 = 2000;
74
75const LOG_NAME: &str = ".vault-events.jsonl";
76const GEN_NAME: &str = ".vault-events.gen";
77
78/// One entry in the change log.
79#[derive(Debug, Clone, Serialize, Deserialize)]
80pub struct Event {
81    /// Monotonic sequence, mirrored in the generation file.
82    pub seq: u64,
83    /// Unix seconds.
84    pub ts: u64,
85    /// `issues_write`, `ping`, and so on.
86    pub kind: String,
87    /// Project touched by the event, when the kind is project-scoped.
88    pub project: Option<String>,
89    /// Issue id touched by the event, when the kind is issue-scoped.
90    pub id: Option<String>,
91    /// Path of the file that changed, when one did.
92    pub path: Option<String>,
93    /// Free-form extra text (a ping message, say).
94    pub detail: Option<String>,
95}
96
97/// Whether emission is enabled. `VISSUE_EVENTS=0` disables it.
98pub fn enabled() -> bool {
99    !matches!(
100        crate::process_env::var("VISSUE_EVENTS").as_deref(),
101        Ok("0") | Ok("false") | Ok("off")
102    )
103}
104
105/// Path of the JSONL change log under `dir`.
106pub fn log_path(dir: &Path) -> PathBuf {
107    dir.join(LOG_NAME)
108}
109
110/// Path of the generation counter file under `dir`.
111pub fn gen_path(dir: &Path) -> PathBuf {
112    dir.join(GEN_NAME)
113}
114
115fn now_secs() -> u64 {
116    SystemTime::now()
117        .duration_since(UNIX_EPOCH)
118        .map(|d| d.as_secs())
119        .unwrap_or(0)
120}
121
122fn now_millis() -> u64 {
123    SystemTime::now()
124        .duration_since(UNIX_EPOCH)
125        .map(|d| d.as_millis() as u64)
126        .unwrap_or(0)
127}
128
129fn read_gen(dir: &Path) -> u64 {
130    fs::read_to_string(gen_path(dir))
131        .ok()
132        .and_then(|s| s.trim().parse().ok())
133        .unwrap_or(0)
134}
135
136fn write_gen(dir: &Path, seq: u64) -> Result<()> {
137    fs::create_dir_all(dir)?;
138    let target = gen_path(dir);
139    // A shared temporary name races: concurrent emitters would rename each
140    // other's file out from under themselves.
141    let tmp = dir.join(format!("{GEN_NAME}.tmp.{}-{}", std::process::id(), seq));
142    fs::write(&tmp, format!("{seq}\n"))?;
143    if let Err(e) = fs::rename(&tmp, &target) {
144        let _ = fs::remove_file(&tmp);
145        return Err(e)
146            .with_context(|| format!("rename {} -> {}", tmp.display(), target.display()))
147            .map_err(crate::error::Error::from);
148    }
149    Ok(())
150}
151
152/// The current generation. A poller compares this against the last value it
153/// saw; a difference means something changed.
154pub fn generation_in(dir: &Path) -> u64 {
155    let g = read_gen(dir);
156    let _ = LAST_SEQ.fetch_max(g, Ordering::Relaxed);
157    g
158}
159
160/// Append one event and return the new sequence.
161///
162/// # Errors
163///
164/// Returns an error if the event directory cannot be created, locked, or
165/// written.
166pub fn emit_in(
167    dir: &Path,
168    kind: &str,
169    project: Option<&str>,
170    id: Option<&str>,
171    path: Option<&Path>,
172    detail: Option<&str>,
173) -> Result<u64> {
174    with_events_lock(dir, || {
175        fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
176
177        let prev = read_gen(dir).max(LAST_SEQ.load(Ordering::Relaxed));
178        let seq = prev.saturating_add(1);
179        LAST_SEQ.store(seq, Ordering::Relaxed);
180
181        let event = Event {
182            seq,
183            ts: now_secs(),
184            kind: kind.to_string(),
185            project: project.map(|s| s.to_string()),
186            id: id.map(|s| s.to_string()),
187            path: path.map(|p| p.display().to_string()),
188            detail: detail.map(|s| s.to_string()),
189        };
190
191        if kind == "issues_write" {
192            let now_ms = now_millis();
193            let key = dir.to_path_buf();
194            let mut last_by_dir = LAST_EMIT_MS_BY_DIR
195                .get_or_init(|| Mutex::new(HashMap::new()))
196                .lock()
197                .unwrap_or_else(|poisoned| poisoned.into_inner());
198            let previous = last_by_dir.get(&key).copied().unwrap_or(0);
199            if previous > 0 && now_ms.saturating_sub(previous) < DEBOUNCE_MS {
200                // Inside the window: advance the generation so pollers still wake,
201                // but leave the log alone.
202                write_gen(dir, seq)?;
203                LAST_SEQ.store(seq, Ordering::Relaxed);
204                last_by_dir.insert(key, now_ms);
205                return Ok(seq);
206            }
207            last_by_dir.insert(key, now_ms);
208        }
209
210        let line = serde_json::to_string(&event)?;
211        let mut file = OpenOptions::new()
212            .create(true)
213            .append(true)
214            .open(log_path(dir))?;
215        writeln!(file, "{line}")?;
216        file.flush()?;
217        write_gen(dir, seq)?;
218        Ok(seq)
219    })
220}
221
222/// Record that a project's issues.org was rewritten. Called by the store after
223/// a successful write; failure here never fails the write.
224///
225/// # Errors
226///
227/// Returns an error if the event directory cannot be created, locked, or
228/// written.
229pub fn emit_issues_write(dir: &Path, project: &str, path: &Path) -> Result<u64> {
230    emit_in(
231        dir,
232        "issues_write",
233        Some(project),
234        None,
235        Some(path),
236        Some("issues.org updated"),
237    )
238}
239
240/// kind=state_change, id=Some(id), detail=Some("FROM->TO"), project set.
241/// NOT debounced (unlike issues_write).
242///
243/// # Errors
244///
245/// Returns an error if the event directory cannot be created, locked, or
246/// written.
247pub fn emit_state_change(
248    layout: &Layout,
249    project: &str,
250    id: &str,
251    from: &str,
252    to: &str,
253) -> Result<u64> {
254    emit_in(
255        &events_dir(layout),
256        "state_change",
257        Some(project),
258        Some(id),
259        None,
260        Some(&format!("{from}->{to}")),
261    )
262}
263
264/// Events with a sequence above `since_seq`, most recent last.
265///
266/// # Errors
267///
268/// Returns an error if the log file exists but cannot be read.
269pub fn since_in(dir: &Path, since_seq: u64, limit: usize) -> Result<Vec<Event>> {
270    let log = log_path(dir);
271    if !log.is_file() {
272        return Ok(Vec::new());
273    }
274    let text = fs::read_to_string(&log)?;
275    let mut out = Vec::new();
276    for line in text.lines().rev() {
277        if line.trim().is_empty() {
278            continue;
279        }
280        let event: Event = match serde_json::from_str(line) {
281            Ok(e) => e,
282            Err(_) => continue,
283        };
284        if event.seq <= since_seq {
285            break;
286        }
287        out.push(event);
288        if out.len() >= limit {
289            break;
290        }
291    }
292    out.reverse();
293    Ok(out)
294}
295
296/// As [`since_in`], narrowed to a project, a kind, or both.
297///
298/// # Errors
299///
300/// Returns an error if the log file exists but cannot be read.
301pub fn since_filtered_in(
302    dir: &Path,
303    since_seq: u64,
304    limit: usize,
305    project: Option<&str>,
306    kind: Option<&str>,
307) -> Result<Vec<Event>> {
308    let mut events = since_in(dir, since_seq, limit.saturating_mul(4).max(limit))?;
309    if let Some(p) = project {
310        events.retain(|e| e.project.as_deref() == Some(p));
311    }
312    if let Some(k) = kind {
313        events.retain(|e| e.kind == k);
314    }
315    events.truncate(limit);
316    Ok(events)
317}
318
319// --- Layout-facing wrappers, called by the CLI and MCP server. ---
320
321/// The directory holding the event files: the same one holding the projects.
322pub fn events_dir(layout: &Layout) -> PathBuf {
323    layout.projects_dir()
324}
325
326/// The current generation for this layout's event directory.
327pub fn generation(layout: &Layout) -> u64 {
328    generation_in(&events_dir(layout))
329}
330
331/// Events with a sequence above `since_seq` for this layout.
332///
333/// # Errors
334///
335/// Returns an error if the log file exists but cannot be read.
336pub fn since(layout: &Layout, since_seq: u64, limit: usize) -> Result<Vec<Event>> {
337    since_in(&events_dir(layout), since_seq, limit)
338}
339
340/// Text plus a trailing JSON block, the shape existing readers parse.
341///
342/// # Errors
343///
344/// Returns an error if the log file exists but cannot be read.
345pub fn since_report(layout: &Layout, since_seq: u64, limit: usize) -> Result<String> {
346    let dir = events_dir(layout);
347    let events = since_in(&dir, since_seq, limit)?;
348    Ok(render_events(&dir, since_seq, events))
349}
350
351fn render_events(dir: &Path, since_seq: u64, events: Vec<Event>) -> String {
352    let generation_now = generation_in(dir);
353    let mut text = format!(
354        "generation={} since={} count={}\n",
355        generation_now,
356        since_seq,
357        events.len()
358    );
359    for e in &events {
360        text.push_str(&format!(
361            "{}\t{}\t{}\t{:?}\t{:?}\t{:?}\n",
362            e.seq, e.ts, e.kind, e.project, e.id, e.path
363        ));
364    }
365    let data = serde_json::json!({
366        "generation": generation_now,
367        "since": since_seq,
368        "events": events,
369        "log": log_path(dir).display().to_string(),
370        "gen_file": gen_path(dir).display().to_string(),
371    });
372    text.push_str("\n---json---\n");
373    text.push_str(&data.to_string());
374    text.push('\n');
375    text
376}
377
378/// Append a manual event, waking pollers without touching an issues.org.
379///
380/// # Errors
381///
382/// Returns an error if the event directory cannot be created, locked, or
383/// written.
384pub fn ping_report(layout: &Layout, detail: Option<&str>) -> Result<String> {
385    let dir = events_dir(layout);
386    let seq = emit_in(
387        &dir,
388        "ping",
389        None,
390        None,
391        None,
392        detail.or(Some("manual ping")),
393    )?;
394    Ok(format!(
395        "ping seq={} generation={}\nlog={}\n",
396        seq,
397        generation_in(&dir),
398        log_path(&dir).display()
399    ))
400}
401
402/// Block until the generation passes `last`, or the timeout expires. Returns
403/// the generation either way; the caller compares it against `last` to tell
404/// which happened. A `timeout_ms` of 0 is a peek: it never sleeps.
405///
406/// # Errors
407///
408/// The signature is `Result` to match the other event verbs. This path does
409/// not fail.
410pub fn wait_generation(layout: &Layout, last: u64, poll_ms: u64, timeout_ms: u64) -> Result<u64> {
411    let dir = events_dir(layout);
412    let start = std::time::Instant::now();
413    let poll = poll_ms.max(1);
414    loop {
415        let g = generation_in(&dir);
416        if g > last {
417            return Ok(g);
418        }
419        let elapsed = start.elapsed().as_millis() as u64;
420        if elapsed >= timeout_ms {
421            return Ok(g);
422        }
423        let remain = timeout_ms - elapsed;
424        std::thread::sleep(std::time::Duration::from_millis(poll.min(remain)));
425    }
426}
427
428/// Outcome of [`wait_until_terminal`].
429#[derive(Debug, Clone, PartialEq, Eq)]
430pub enum TerminalWait {
431    /// The issue reached DONE.
432    Done {
433        /// Generation at the moment the terminal state was observed.
434        generation: u64,
435    },
436    /// The issue reached CANCELLED.
437    Cancelled {
438        /// Generation at the moment the terminal state was observed.
439        generation: u64,
440    },
441    /// The timeout expired while the issue was still non-terminal.
442    Timeout {
443        /// Generation at the moment the timeout expired.
444        generation: u64,
445        /// Heading state when the wait gave up.
446        state: String,
447    },
448}
449
450/// Poll the issue state. Re-read on generation change or poll interval.
451/// Missing id is an error (IssueNotFound).
452///
453/// # Errors
454///
455/// Returns [`Error::IssueNotFound`] if `id` is not in the catalog, or an
456/// error if a project file cannot be read or parsed.
457pub fn wait_until_terminal(
458    layout: &Layout,
459    id: &str,
460    poll_ms: u64,
461    timeout_ms: u64,
462) -> Result<TerminalWait> {
463    let dir = events_dir(layout);
464    let start = std::time::Instant::now();
465    let mut last_gen = generation_in(&dir);
466    loop {
467        let heading = crate::store::find_by_id(layout, id)?
468            .ok_or_else(|| Error::IssueNotFound { id: id.to_string() })?
469            .0;
470        let generation = generation_in(&dir);
471        match heading.state.as_str() {
472            "DONE" => return Ok(TerminalWait::Done { generation }),
473            "CANCELLED" => return Ok(TerminalWait::Cancelled { generation }),
474            _ => {}
475        }
476        if start.elapsed().as_millis() as u64 >= timeout_ms {
477            return Ok(TerminalWait::Timeout {
478                generation,
479                state: heading.state,
480            });
481        }
482        // Wake on a generation bump or when the poll interval elapses.
483        let poll = poll_ms.max(50);
484        let slice = 50_u64.min(poll);
485        let wake = std::time::Instant::now();
486        loop {
487            let now_gen = generation_in(&dir);
488            if now_gen != last_gen {
489                last_gen = now_gen;
490                break;
491            }
492            if wake.elapsed().as_millis() as u64 >= poll {
493                break;
494            }
495            if start.elapsed().as_millis() as u64 >= timeout_ms {
496                break;
497            }
498            std::thread::sleep(std::time::Duration::from_millis(slice));
499        }
500    }
501}
502
503/// The last `n` events in the log.
504///
505/// Counted from the end of the log rather than back from the generation: a
506/// debounced burst advances the generation without appending, so a sequence
507/// window that wide can hold fewer lines than the caller asked for, or none.
508///
509/// # Errors
510///
511/// Returns an error if the log file exists but cannot be read.
512pub fn tail_in(dir: &Path, n: usize) -> Result<Vec<Event>> {
513    let log = log_path(dir);
514    if !log.is_file() {
515        return Ok(Vec::new());
516    }
517    let text = fs::read_to_string(&log)?;
518    let mut out: Vec<Event> = text
519        .lines()
520        .rev()
521        .filter(|line| !line.trim().is_empty())
522        .filter_map(|line| serde_json::from_str(line).ok())
523        .take(n)
524        .collect();
525    out.reverse();
526    Ok(out)
527}
528
529/// The last `n` events, for a reader that only wants what is recent.
530///
531/// # Errors
532///
533/// Returns an error if the log file exists but cannot be read.
534pub fn tail_report(layout: &Layout, n: usize) -> Result<String> {
535    let dir = events_dir(layout);
536    let events = tail_in(&dir, n)?;
537    let since_seq = events.first().map(|e| e.seq.saturating_sub(1)).unwrap_or(0);
538    Ok(render_events(&dir, since_seq, events))
539}
540
541/// Add the event files to a `.gitignore` that already exists beside them. Best
542/// effort: a tracker without one is left alone.
543///
544/// # Errors
545///
546/// Returns an error if an existing `.gitignore` cannot be read or appended.
547pub fn ensure_gitignore_hint(dir: &Path) -> Result<()> {
548    let gitignore = dir.join(".gitignore");
549    if !gitignore.is_file() {
550        return Ok(());
551    }
552    let current = fs::read_to_string(&gitignore)?;
553    if current.contains(".vault-events") {
554        return Ok(());
555    }
556    let mut file = OpenOptions::new().append(true).open(&gitignore)?;
557    writeln!(
558        file,
559        "\n# agent ping stream (local)\n{LOG_NAME}\n{GEN_NAME}\n"
560    )?;
561    Ok(())
562}
563
564#[cfg(test)]
565mod tests {
566    use super::*;
567    use crate::config::DEFAULT_PREFIX;
568    use crate::model::TODO_HEADER;
569
570    #[test]
571    fn a_sequence_advances_and_reads_back() {
572        let dir = tempfile::tempdir().unwrap();
573        let d = dir.path();
574        let first = emit_in(d, "ping", None, None, None, Some("a")).unwrap();
575        let second = emit_in(
576            d,
577            "issues_write",
578            Some("atlas"),
579            Some("atlas-1a2b"),
580            Some(Path::new("atlas/issues.org")),
581            None,
582        )
583        .unwrap();
584        assert!(second > first);
585        assert_eq!(generation_in(d), second);
586
587        let events = since_in(d, first, 10).unwrap();
588        assert_eq!(events.len(), 1);
589        assert_eq!(events[0].kind, "issues_write");
590        assert_eq!(events[0].project.as_deref(), Some("atlas"));
591    }
592
593    #[test]
594    fn filters_narrow_by_project_and_kind() {
595        let dir = tempfile::tempdir().unwrap();
596        let d = dir.path();
597        emit_in(d, "ping", None, None, None, None).unwrap();
598        emit_in(d, "manual", Some("atlas"), None, None, None).unwrap();
599        emit_in(d, "manual", Some("beacon"), None, None, None).unwrap();
600
601        let by_project = since_filtered_in(d, 0, 10, Some("atlas"), None).unwrap();
602        assert_eq!(by_project.len(), 1);
603        let by_kind = since_filtered_in(d, 0, 10, None, Some("ping")).unwrap();
604        assert_eq!(by_kind.len(), 1);
605        assert_eq!(by_kind[0].kind, "ping");
606    }
607
608    #[test]
609    fn the_report_carries_a_json_block() {
610        let dir = tempfile::tempdir().unwrap();
611        let layout = Layout::new(dir.path(), DEFAULT_PREFIX);
612        ping_report(&layout, Some("hello")).unwrap();
613        let text = since_report(&layout, 0, 10).unwrap();
614        assert!(text.starts_with("generation="), "{text}");
615        let (_, json) = text.split_once("---json---").expect("json block present");
616        let parsed: serde_json::Value = serde_json::from_str(json.trim()).unwrap();
617        assert_eq!(parsed["events"][0]["kind"], "ping");
618        assert_eq!(parsed["events"][0]["detail"], "hello");
619    }
620
621    #[test]
622    fn the_gitignore_hint_only_touches_an_existing_file() {
623        let dir = tempfile::tempdir().unwrap();
624        ensure_gitignore_hint(dir.path()).unwrap();
625        assert!(!dir.path().join(".gitignore").exists());
626
627        fs::write(dir.path().join(".gitignore"), "target\n").unwrap();
628        ensure_gitignore_hint(dir.path()).unwrap();
629        let text = fs::read_to_string(dir.path().join(".gitignore")).unwrap();
630        assert!(text.contains(LOG_NAME), "{text}");
631        assert!(text.contains(GEN_NAME), "{text}");
632
633        // Running again must not append a second copy.
634        ensure_gitignore_hint(dir.path()).unwrap();
635        let again = fs::read_to_string(dir.path().join(".gitignore")).unwrap();
636        assert_eq!(text, again);
637    }
638
639    #[test]
640    fn concurrent_emits_assign_unique_sequences() {
641        use std::sync::Arc;
642        use std::thread;
643
644        let dir = tempfile::tempdir().unwrap();
645        let d = Arc::new(dir.path().to_path_buf());
646        let handles: Vec<_> = (0..16)
647            .map(|i| {
648                let d = Arc::clone(&d);
649                thread::spawn(move || emit_in(&d, "ping", None, None, None, Some(&format!("{i}"))))
650            })
651            .collect();
652        let mut seqs = Vec::new();
653        for handle in handles {
654            seqs.push(handle.join().unwrap().unwrap());
655        }
656        seqs.sort_unstable();
657        seqs.dedup();
658        assert_eq!(seqs.len(), 16, "duplicate event sequences: {seqs:?}");
659        assert_eq!(generation_in(&d), *seqs.last().unwrap());
660    }
661
662    #[test]
663    fn a_tail_counts_lines_not_sequence_numbers() {
664        let dir = tempfile::tempdir().unwrap();
665        let d = dir.path();
666        for i in 0..5 {
667            emit_in(d, "manual", None, None, None, Some(&format!("event {i}"))).unwrap();
668        }
669        // A ping with no log line of its own would still move the generation
670        // past the window a sequence-derived tail would look in.
671        let tailed = tail_in(d, 3).unwrap();
672        assert_eq!(tailed.len(), 3, "{tailed:?}");
673        assert_eq!(tailed[2].detail.as_deref(), Some("event 4"));
674        assert_eq!(tailed[0].detail.as_deref(), Some("event 2"));
675        assert!(tail_in(d, 50).unwrap().len() == 5);
676    }
677
678    #[test]
679    fn waiting_returns_the_current_generation_on_timeout() {
680        let dir = tempfile::tempdir().unwrap();
681        let layout = Layout::new(dir.path(), DEFAULT_PREFIX);
682        let g = generation(&layout);
683        let waited = wait_generation(&layout, g + 100, 50, 120).unwrap();
684        assert!(waited <= g + 100, "timed out without advancing");
685    }
686
687    #[test]
688    fn a_zero_timeout_does_not_sleep() {
689        let dir = tempfile::tempdir().unwrap();
690        let layout = Layout::new(dir.path(), DEFAULT_PREFIX);
691        let start = std::time::Instant::now();
692        let _ = wait_generation(&layout, u64::MAX, 200, 0).unwrap();
693        assert!(
694            start.elapsed() < std::time::Duration::from_millis(50),
695            "timeout 0 must not wait out the poll interval"
696        );
697    }
698
699    #[test]
700    fn a_short_timeout_does_not_wait_the_poll_interval() {
701        let dir = tempfile::tempdir().unwrap();
702        let layout = Layout::new(dir.path(), DEFAULT_PREFIX);
703        let start = std::time::Instant::now();
704        let _ = wait_generation(&layout, u64::MAX, 200, 1).unwrap();
705        assert!(
706            start.elapsed() < std::time::Duration::from_millis(50),
707            "a 1ms timeout must not sleep the 200ms poll"
708        );
709    }
710
711    fn layout_with_issue(state: &str, id: &str) -> (tempfile::TempDir, Layout) {
712        let dir = tempfile::tempdir().unwrap();
713        let layout = Layout::new(dir.path(), DEFAULT_PREFIX);
714        let path = layout.project_issues_path("sample");
715        fs::create_dir_all(path.parent().unwrap()).unwrap();
716        fs::write(
717            &path,
718            format!(
719                "#+TITLE: sample issues\n{TODO_HEADER}\n\n* {state} [#B] wait target\n:PROPERTIES:\n:ID:         {id}\n:END:\n"
720            ),
721        )
722        .unwrap();
723        (dir, layout)
724    }
725
726    #[test]
727    fn emit_state_change_writes_kind_id_and_detail() {
728        let dir = tempfile::tempdir().unwrap();
729        let layout = Layout::new(dir.path(), DEFAULT_PREFIX);
730        let seq = emit_state_change(&layout, "sample", "sample-aaaa", "TODO", "CANCELLED").unwrap();
731        let events = since(&layout, 0, 10).unwrap();
732        assert_eq!(events.len(), 1);
733        assert_eq!(events[0].seq, seq);
734        assert_eq!(events[0].kind, "state_change");
735        assert_eq!(events[0].id.as_deref(), Some("sample-aaaa"));
736        assert_eq!(events[0].project.as_deref(), Some("sample"));
737        assert_eq!(events[0].detail.as_deref(), Some("TODO->CANCELLED"));
738    }
739
740    #[test]
741    fn two_rapid_state_changes_both_appear_in_the_log() {
742        let dir = tempfile::tempdir().unwrap();
743        let layout = Layout::new(dir.path(), DEFAULT_PREFIX);
744        emit_state_change(&layout, "sample", "sample-aaaa", "TODO", "STARTED").unwrap();
745        emit_state_change(&layout, "sample", "sample-aaaa", "STARTED", "DONE").unwrap();
746        let events = since(&layout, 0, 10).unwrap();
747        assert_eq!(events.len(), 2, "{events:?}");
748        assert_eq!(events[0].kind, "state_change");
749        assert_eq!(events[1].kind, "state_change");
750        assert_eq!(events[0].detail.as_deref(), Some("TODO->STARTED"));
751        assert_eq!(events[1].detail.as_deref(), Some("STARTED->DONE"));
752        assert_eq!(events[0].id.as_deref(), Some("sample-aaaa"));
753        assert_eq!(events[1].id.as_deref(), Some("sample-aaaa"));
754    }
755
756    #[test]
757    fn wait_until_terminal_returns_done_immediately() {
758        let (_dir, layout) = layout_with_issue("DONE", "sample-done");
759        let waited = wait_until_terminal(&layout, "sample-done", 50, 120).unwrap();
760        match waited {
761            TerminalWait::Done { .. } => {}
762            other => panic!("expected Done, got {other:?}"),
763        }
764    }
765
766    #[test]
767    fn wait_until_terminal_returns_cancelled() {
768        let (_dir, layout) = layout_with_issue("CANCELLED", "sample-canc");
769        let waited = wait_until_terminal(&layout, "sample-canc", 50, 120).unwrap();
770        match waited {
771            TerminalWait::Cancelled { .. } => {}
772            other => panic!("expected Cancelled, got {other:?}"),
773        }
774    }
775
776    #[test]
777    fn wait_until_terminal_times_out_on_started() {
778        let (_dir, layout) = layout_with_issue("STARTED", "sample-work");
779        let waited = wait_until_terminal(&layout, "sample-work", 50, 120).unwrap();
780        match waited {
781            TerminalWait::Timeout { state, .. } => assert_eq!(state, "STARTED"),
782            other => panic!("expected Timeout, got {other:?}"),
783        }
784    }
785}
vissue_core/events.rs

vissue_core/
events.rs
