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, Result};
18use fs2::FileExt;
19use serde::{Deserialize, Serialize};
20use std::collections::HashMap;
21use std::fs::{self, OpenOptions};
22use std::io::Write;
23use std::path::{Path, PathBuf};
24use std::sync::{
25    atomic::{AtomicU64, Ordering},
26    Arc, Mutex, OnceLock,
27};
28use std::time::{SystemTime, UNIX_EPOCH};
29
30use crate::config::Layout;
31
32static LAST_SEQ: AtomicU64 = AtomicU64::new(0);
33static LAST_EMIT_MS_BY_DIR: OnceLock<Mutex<HashMap<PathBuf, u64>>> = OnceLock::new();
34static PROCESS_LOCKS: OnceLock<Mutex<HashMap<PathBuf, Arc<Mutex<()>>>>> = OnceLock::new();
35
36const LOCK_NAME: &str = ".vault-events.lock";
37
38fn with_events_lock<R, F>(dir: &Path, f: F) -> Result<R>
39where
40    F: FnOnce() -> Result<R>,
41{
42    let key = dir.to_path_buf();
43    let mutex = {
44        let mut map = PROCESS_LOCKS
45            .get_or_init(|| Mutex::new(HashMap::new()))
46            .lock()
47            .unwrap_or_else(|p| p.into_inner());
48        map.entry(key)
49            .or_insert_with(|| Arc::new(Mutex::new(())))
50            .clone()
51    };
52    let _proc = mutex.lock().unwrap_or_else(|p| p.into_inner());
53    fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
54    let lock_path = dir.join(LOCK_NAME);
55    let file = OpenOptions::new()
56        .create(true)
57        .read(true)
58        .append(true)
59        .open(&lock_path)
60        .with_context(|| format!("open {}", lock_path.display()))?;
61    file.lock_exclusive()
62        .with_context(|| format!("lock {}", lock_path.display()))?;
63    let result = f();
64    let _ = FileExt::unlock(&file);
65    let _ = file;
66    result
67}
68
69/// Repeated writes inside this window bump the generation without appending a
70/// line, so an editor or agent saving in a burst does not flood the log.
71const DEBOUNCE_MS: u64 = 2000;
72
73const LOG_NAME: &str = ".vault-events.jsonl";
74const GEN_NAME: &str = ".vault-events.gen";
75
76/// One entry in the change log.
77#[derive(Debug, Clone, Serialize, Deserialize)]
78pub struct Event {
79    /// Monotonic sequence, mirrored in the generation file.
80    pub seq: u64,
81    /// Unix seconds.
82    pub ts: u64,
83    /// `issues_write`, `ping`, and so on.
84    pub kind: String,
85    pub project: Option<String>,
86    pub id: Option<String>,
87    pub path: Option<String>,
88    pub detail: Option<String>,
89}
90
91/// Whether emission is enabled. `VISSUE_EVENTS=0` disables it.
92pub fn enabled() -> bool {
93    !matches!(
94        std::env::var("VISSUE_EVENTS").as_deref(),
95        Ok("0") | Ok("false") | Ok("off")
96    )
97}
98
99pub fn log_path(dir: &Path) -> PathBuf {
100    dir.join(LOG_NAME)
101}
102
103pub fn gen_path(dir: &Path) -> PathBuf {
104    dir.join(GEN_NAME)
105}
106
107fn now_secs() -> u64 {
108    SystemTime::now()
109        .duration_since(UNIX_EPOCH)
110        .map(|d| d.as_secs())
111        .unwrap_or(0)
112}
113
114fn now_millis() -> u64 {
115    SystemTime::now()
116        .duration_since(UNIX_EPOCH)
117        .map(|d| d.as_millis() as u64)
118        .unwrap_or(0)
119}
120
121fn read_gen(dir: &Path) -> u64 {
122    fs::read_to_string(gen_path(dir))
123        .ok()
124        .and_then(|s| s.trim().parse().ok())
125        .unwrap_or(0)
126}
127
128fn write_gen(dir: &Path, seq: u64) -> Result<()> {
129    fs::create_dir_all(dir)?;
130    let target = gen_path(dir);
131    // A shared temporary name races: concurrent emitters would rename each
132    // other's file out from under themselves.
133    let tmp = dir.join(format!("{GEN_NAME}.tmp.{}-{}", std::process::id(), seq));
134    fs::write(&tmp, format!("{seq}\n"))?;
135    if let Err(e) = fs::rename(&tmp, &target) {
136        let _ = fs::remove_file(&tmp);
137        return Err(e).with_context(|| format!("rename {} -> {}", tmp.display(), target.display()));
138    }
139    Ok(())
140}
141
142/// The current generation. A poller compares this against the last value it
143/// saw; a difference means something changed.
144pub fn generation_in(dir: &Path) -> u64 {
145    let g = read_gen(dir);
146    let _ = LAST_SEQ.fetch_max(g, Ordering::Relaxed);
147    g
148}
149
150/// Append one event and return the new sequence.
151pub fn emit_in(
152    dir: &Path,
153    kind: &str,
154    project: Option<&str>,
155    id: Option<&str>,
156    path: Option<&Path>,
157    detail: Option<&str>,
158) -> Result<u64> {
159    with_events_lock(dir, || {
160        fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?;
161
162        let prev = read_gen(dir).max(LAST_SEQ.load(Ordering::Relaxed));
163        let seq = prev.saturating_add(1);
164        LAST_SEQ.store(seq, Ordering::Relaxed);
165
166        let event = Event {
167            seq,
168            ts: now_secs(),
169            kind: kind.to_string(),
170            project: project.map(|s| s.to_string()),
171            id: id.map(|s| s.to_string()),
172            path: path.map(|p| p.display().to_string()),
173            detail: detail.map(|s| s.to_string()),
174        };
175
176        if kind == "issues_write" {
177            let now_ms = now_millis();
178            let key = dir.to_path_buf();
179            let mut last_by_dir = LAST_EMIT_MS_BY_DIR
180                .get_or_init(|| Mutex::new(HashMap::new()))
181                .lock()
182                .unwrap_or_else(|poisoned| poisoned.into_inner());
183            let previous = last_by_dir.get(&key).copied().unwrap_or(0);
184            if previous > 0 && now_ms.saturating_sub(previous) < DEBOUNCE_MS {
185                // Inside the window: advance the generation so pollers still wake,
186                // but leave the log alone.
187                write_gen(dir, seq)?;
188                LAST_SEQ.store(seq, Ordering::Relaxed);
189                last_by_dir.insert(key, now_ms);
190                return Ok(seq);
191            }
192            last_by_dir.insert(key, now_ms);
193        }
194
195        let line = serde_json::to_string(&event)?;
196        let mut file = OpenOptions::new()
197            .create(true)
198            .append(true)
199            .open(log_path(dir))?;
200        writeln!(file, "{line}")?;
201        file.flush()?;
202        write_gen(dir, seq)?;
203        Ok(seq)
204    })
205}
206
207/// Record that a project's issues.org was rewritten. Called by the store after
208/// a successful write; failure here never fails the write.
209pub fn emit_issues_write(dir: &Path, project: &str, path: &Path) -> Result<u64> {
210    emit_in(
211        dir,
212        "issues_write",
213        Some(project),
214        None,
215        Some(path),
216        Some("issues.org updated"),
217    )
218}
219
220/// Events with a sequence above `since_seq`, most recent last.
221pub fn since_in(dir: &Path, since_seq: u64, limit: usize) -> Result<Vec<Event>> {
222    let log = log_path(dir);
223    if !log.is_file() {
224        return Ok(Vec::new());
225    }
226    let text = fs::read_to_string(&log)?;
227    let mut out = Vec::new();
228    for line in text.lines().rev() {
229        if line.trim().is_empty() {
230            continue;
231        }
232        let event: Event = match serde_json::from_str(line) {
233            Ok(e) => e,
234            Err(_) => continue,
235        };
236        if event.seq <= since_seq {
237            break;
238        }
239        out.push(event);
240        if out.len() >= limit {
241            break;
242        }
243    }
244    out.reverse();
245    Ok(out)
246}
247
248/// As [`since_in`], narrowed to a project, a kind, or both.
249pub fn since_filtered_in(
250    dir: &Path,
251    since_seq: u64,
252    limit: usize,
253    project: Option<&str>,
254    kind: Option<&str>,
255) -> Result<Vec<Event>> {
256    let mut events = since_in(dir, since_seq, limit.saturating_mul(4).max(limit))?;
257    if let Some(p) = project {
258        events.retain(|e| e.project.as_deref() == Some(p));
259    }
260    if let Some(k) = kind {
261        events.retain(|e| e.kind == k);
262    }
263    events.truncate(limit);
264    Ok(events)
265}
266
267// --- Layout-facing wrappers, called by the CLI and MCP server. ---
268
269/// The directory holding the event files: the same one holding the projects.
270pub fn events_dir(layout: &Layout) -> PathBuf {
271    layout.projects_dir()
272}
273
274pub fn generation(layout: &Layout) -> u64 {
275    generation_in(&events_dir(layout))
276}
277
278pub fn since(layout: &Layout, since_seq: u64, limit: usize) -> Result<Vec<Event>> {
279    since_in(&events_dir(layout), since_seq, limit)
280}
281
282/// Text plus a trailing JSON block, the shape existing readers parse.
283pub fn since_report(layout: &Layout, since_seq: u64, limit: usize) -> Result<String> {
284    let dir = events_dir(layout);
285    let events = since_in(&dir, since_seq, limit)?;
286    Ok(render_events(&dir, since_seq, events))
287}
288
289fn render_events(dir: &Path, since_seq: u64, events: Vec<Event>) -> String {
290    let generation_now = generation_in(dir);
291    let mut text = format!(
292        "generation={} since={} count={}\n",
293        generation_now,
294        since_seq,
295        events.len()
296    );
297    for e in &events {
298        text.push_str(&format!(
299            "{}\t{}\t{}\t{:?}\t{:?}\t{:?}\n",
300            e.seq, e.ts, e.kind, e.project, e.id, e.path
301        ));
302    }
303    let data = serde_json::json!({
304        "generation": generation_now,
305        "since": since_seq,
306        "events": events,
307        "log": log_path(dir).display().to_string(),
308        "gen_file": gen_path(dir).display().to_string(),
309    });
310    text.push_str("\n---json---\n");
311    text.push_str(&data.to_string());
312    text.push('\n');
313    text
314}
315
316/// Append a manual event, waking pollers without touching an issues.org.
317pub fn ping_report(layout: &Layout, detail: Option<&str>) -> Result<String> {
318    let dir = events_dir(layout);
319    let seq = emit_in(
320        &dir,
321        "ping",
322        None,
323        None,
324        None,
325        detail.or(Some("manual ping")),
326    )?;
327    Ok(format!(
328        "ping seq={} generation={}\nlog={}\n",
329        seq,
330        generation_in(&dir),
331        log_path(&dir).display()
332    ))
333}
334
335/// Block until the generation passes `last`, or the timeout expires. Returns
336/// the generation either way; the caller compares it against `last` to tell
337/// which happened.
338pub fn wait_generation(layout: &Layout, last: u64, poll_ms: u64, timeout_ms: u64) -> Result<u64> {
339    let dir = events_dir(layout);
340    let start = std::time::Instant::now();
341    loop {
342        let g = generation_in(&dir);
343        if g > last {
344            return Ok(g);
345        }
346        if start.elapsed().as_millis() as u64 >= timeout_ms {
347            return Ok(g);
348        }
349        std::thread::sleep(std::time::Duration::from_millis(poll_ms.max(50)));
350    }
351}
352
353/// The last `n` events in the log.
354///
355/// Counted from the end of the log rather than back from the generation: a
356/// debounced burst advances the generation without appending, so a sequence
357/// window that wide can hold fewer lines than the caller asked for, or none.
358pub fn tail_in(dir: &Path, n: usize) -> Result<Vec<Event>> {
359    let log = log_path(dir);
360    if !log.is_file() {
361        return Ok(Vec::new());
362    }
363    let text = fs::read_to_string(&log)?;
364    let mut out: Vec<Event> = text
365        .lines()
366        .rev()
367        .filter(|line| !line.trim().is_empty())
368        .filter_map(|line| serde_json::from_str(line).ok())
369        .take(n)
370        .collect();
371    out.reverse();
372    Ok(out)
373}
374
375/// The last `n` events, for a reader that only wants what is recent.
376pub fn tail_report(layout: &Layout, n: usize) -> Result<String> {
377    let dir = events_dir(layout);
378    let events = tail_in(&dir, n)?;
379    let since_seq = events.first().map(|e| e.seq.saturating_sub(1)).unwrap_or(0);
380    Ok(render_events(&dir, since_seq, events))
381}
382
383/// Add the event files to a `.gitignore` that already exists beside them. Best
384/// effort: a tracker without one is left alone.
385pub fn ensure_gitignore_hint(dir: &Path) -> Result<()> {
386    let gitignore = dir.join(".gitignore");
387    if !gitignore.is_file() {
388        return Ok(());
389    }
390    let current = fs::read_to_string(&gitignore)?;
391    if current.contains(".vault-events") {
392        return Ok(());
393    }
394    let mut file = OpenOptions::new().append(true).open(&gitignore)?;
395    writeln!(
396        file,
397        "\n# agent ping stream (local)\n{LOG_NAME}\n{GEN_NAME}\n"
398    )?;
399    Ok(())
400}
401
402#[cfg(test)]
403mod tests {
404    use super::*;
405    use crate::config::DEFAULT_PREFIX;
406
407    #[test]
408    fn a_sequence_advances_and_reads_back() {
409        let dir = tempfile::tempdir().unwrap();
410        let d = dir.path();
411        let first = emit_in(d, "ping", None, None, None, Some("a")).unwrap();
412        let second = emit_in(
413            d,
414            "issues_write",
415            Some("atlas"),
416            Some("atlas-1a2b"),
417            Some(Path::new("atlas/issues.org")),
418            None,
419        )
420        .unwrap();
421        assert!(second > first);
422        assert_eq!(generation_in(d), second);
423
424        let events = since_in(d, first, 10).unwrap();
425        assert_eq!(events.len(), 1);
426        assert_eq!(events[0].kind, "issues_write");
427        assert_eq!(events[0].project.as_deref(), Some("atlas"));
428    }
429
430    #[test]
431    fn filters_narrow_by_project_and_kind() {
432        let dir = tempfile::tempdir().unwrap();
433        let d = dir.path();
434        emit_in(d, "ping", None, None, None, None).unwrap();
435        emit_in(d, "manual", Some("atlas"), None, None, None).unwrap();
436        emit_in(d, "manual", Some("beacon"), None, None, None).unwrap();
437
438        let by_project = since_filtered_in(d, 0, 10, Some("atlas"), None).unwrap();
439        assert_eq!(by_project.len(), 1);
440        let by_kind = since_filtered_in(d, 0, 10, None, Some("ping")).unwrap();
441        assert_eq!(by_kind.len(), 1);
442        assert_eq!(by_kind[0].kind, "ping");
443    }
444
445    #[test]
446    fn the_report_carries_a_json_block() {
447        let dir = tempfile::tempdir().unwrap();
448        let layout = Layout::new(dir.path(), DEFAULT_PREFIX);
449        ping_report(&layout, Some("hello")).unwrap();
450        let text = since_report(&layout, 0, 10).unwrap();
451        assert!(text.starts_with("generation="), "{text}");
452        let (_, json) = text.split_once("---json---").expect("json block present");
453        let parsed: serde_json::Value = serde_json::from_str(json.trim()).unwrap();
454        assert_eq!(parsed["events"][0]["kind"], "ping");
455        assert_eq!(parsed["events"][0]["detail"], "hello");
456    }
457
458    #[test]
459    fn the_gitignore_hint_only_touches_an_existing_file() {
460        let dir = tempfile::tempdir().unwrap();
461        ensure_gitignore_hint(dir.path()).unwrap();
462        assert!(!dir.path().join(".gitignore").exists());
463
464        fs::write(dir.path().join(".gitignore"), "target\n").unwrap();
465        ensure_gitignore_hint(dir.path()).unwrap();
466        let text = fs::read_to_string(dir.path().join(".gitignore")).unwrap();
467        assert!(text.contains(LOG_NAME), "{text}");
468        assert!(text.contains(GEN_NAME), "{text}");
469
470        // Running again must not append a second copy.
471        ensure_gitignore_hint(dir.path()).unwrap();
472        let again = fs::read_to_string(dir.path().join(".gitignore")).unwrap();
473        assert_eq!(text, again);
474    }
475
476    #[test]
477    fn concurrent_emits_assign_unique_sequences() {
478        use std::sync::Arc;
479        use std::thread;
480
481        let dir = tempfile::tempdir().unwrap();
482        let d = Arc::new(dir.path().to_path_buf());
483        let handles: Vec<_> = (0..16)
484            .map(|i| {
485                let d = Arc::clone(&d);
486                thread::spawn(move || emit_in(&d, "ping", None, None, None, Some(&format!("{i}"))))
487            })
488            .collect();
489        let mut seqs = Vec::new();
490        for handle in handles {
491            seqs.push(handle.join().unwrap().unwrap());
492        }
493        seqs.sort_unstable();
494        seqs.dedup();
495        assert_eq!(seqs.len(), 16, "duplicate event sequences: {seqs:?}");
496        assert_eq!(generation_in(&d), *seqs.last().unwrap());
497    }
498
499    #[test]
500    fn a_tail_counts_lines_not_sequence_numbers() {
501        let dir = tempfile::tempdir().unwrap();
502        let d = dir.path();
503        for i in 0..5 {
504            emit_in(d, "manual", None, None, None, Some(&format!("event {i}"))).unwrap();
505        }
506        // A ping with no log line of its own would still move the generation
507        // past the window a sequence-derived tail would look in.
508        let tailed = tail_in(d, 3).unwrap();
509        assert_eq!(tailed.len(), 3, "{tailed:?}");
510        assert_eq!(tailed[2].detail.as_deref(), Some("event 4"));
511        assert_eq!(tailed[0].detail.as_deref(), Some("event 2"));
512        assert!(tail_in(d, 50).unwrap().len() == 5);
513    }
514
515    #[test]
516    fn waiting_returns_the_current_generation_on_timeout() {
517        let dir = tempfile::tempdir().unwrap();
518        let layout = Layout::new(dir.path(), DEFAULT_PREFIX);
519        let g = generation(&layout);
520        let waited = wait_generation(&layout, g + 100, 50, 120).unwrap();
521        assert!(waited <= g + 100, "timed out without advancing");
522    }
523}
vissue_core/events.rs

vissue_core/
events.rs
