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::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/// Events with a sequence above `since_seq`, most recent last.
241///
242/// # Errors
243///
244/// Returns an error if the log file exists but cannot be read.
245pub fn since_in(dir: &Path, since_seq: u64, limit: usize) -> Result<Vec<Event>> {
246    let log = log_path(dir);
247    if !log.is_file() {
248        return Ok(Vec::new());
249    }
250    let text = fs::read_to_string(&log)?;
251    let mut out = Vec::new();
252    for line in text.lines().rev() {
253        if line.trim().is_empty() {
254            continue;
255        }
256        let event: Event = match serde_json::from_str(line) {
257            Ok(e) => e,
258            Err(_) => continue,
259        };
260        if event.seq <= since_seq {
261            break;
262        }
263        out.push(event);
264        if out.len() >= limit {
265            break;
266        }
267    }
268    out.reverse();
269    Ok(out)
270}
271
272/// As [`since_in`], narrowed to a project, a kind, or both.
273///
274/// # Errors
275///
276/// Returns an error if the log file exists but cannot be read.
277pub fn since_filtered_in(
278    dir: &Path,
279    since_seq: u64,
280    limit: usize,
281    project: Option<&str>,
282    kind: Option<&str>,
283) -> Result<Vec<Event>> {
284    let mut events = since_in(dir, since_seq, limit.saturating_mul(4).max(limit))?;
285    if let Some(p) = project {
286        events.retain(|e| e.project.as_deref() == Some(p));
287    }
288    if let Some(k) = kind {
289        events.retain(|e| e.kind == k);
290    }
291    events.truncate(limit);
292    Ok(events)
293}
294
295// --- Layout-facing wrappers, called by the CLI and MCP server. ---
296
297/// The directory holding the event files: the same one holding the projects.
298pub fn events_dir(layout: &Layout) -> PathBuf {
299    layout.projects_dir()
300}
301
302/// The current generation for this layout's event directory.
303pub fn generation(layout: &Layout) -> u64 {
304    generation_in(&events_dir(layout))
305}
306
307/// Events with a sequence above `since_seq` for this layout.
308///
309/// # Errors
310///
311/// Returns an error if the log file exists but cannot be read.
312pub fn since(layout: &Layout, since_seq: u64, limit: usize) -> Result<Vec<Event>> {
313    since_in(&events_dir(layout), since_seq, limit)
314}
315
316/// Text plus a trailing JSON block, the shape existing readers parse.
317///
318/// # Errors
319///
320/// Returns an error if the log file exists but cannot be read.
321pub fn since_report(layout: &Layout, since_seq: u64, limit: usize) -> Result<String> {
322    let dir = events_dir(layout);
323    let events = since_in(&dir, since_seq, limit)?;
324    Ok(render_events(&dir, since_seq, events))
325}
326
327fn render_events(dir: &Path, since_seq: u64, events: Vec<Event>) -> String {
328    let generation_now = generation_in(dir);
329    let mut text = format!(
330        "generation={} since={} count={}\n",
331        generation_now,
332        since_seq,
333        events.len()
334    );
335    for e in &events {
336        text.push_str(&format!(
337            "{}\t{}\t{}\t{:?}\t{:?}\t{:?}\n",
338            e.seq, e.ts, e.kind, e.project, e.id, e.path
339        ));
340    }
341    let data = serde_json::json!({
342        "generation": generation_now,
343        "since": since_seq,
344        "events": events,
345        "log": log_path(dir).display().to_string(),
346        "gen_file": gen_path(dir).display().to_string(),
347    });
348    text.push_str("\n---json---\n");
349    text.push_str(&data.to_string());
350    text.push('\n');
351    text
352}
353
354/// Append a manual event, waking pollers without touching an issues.org.
355///
356/// # Errors
357///
358/// Returns an error if the event directory cannot be created, locked, or
359/// written.
360pub fn ping_report(layout: &Layout, detail: Option<&str>) -> Result<String> {
361    let dir = events_dir(layout);
362    let seq = emit_in(
363        &dir,
364        "ping",
365        None,
366        None,
367        None,
368        detail.or(Some("manual ping")),
369    )?;
370    Ok(format!(
371        "ping seq={} generation={}\nlog={}\n",
372        seq,
373        generation_in(&dir),
374        log_path(&dir).display()
375    ))
376}
377
378/// Block until the generation passes `last`, or the timeout expires. Returns
379/// the generation either way; the caller compares it against `last` to tell
380/// which happened.
381///
382/// # Errors
383///
384/// The signature is `Result` to match the other event verbs. This path does
385/// not fail.
386pub fn wait_generation(layout: &Layout, last: u64, poll_ms: u64, timeout_ms: u64) -> Result<u64> {
387    let dir = events_dir(layout);
388    let start = std::time::Instant::now();
389    loop {
390        let g = generation_in(&dir);
391        if g > last {
392            return Ok(g);
393        }
394        if start.elapsed().as_millis() as u64 >= timeout_ms {
395            return Ok(g);
396        }
397        std::thread::sleep(std::time::Duration::from_millis(poll_ms.max(50)));
398    }
399}
400
401/// The last `n` events in the log.
402///
403/// Counted from the end of the log rather than back from the generation: a
404/// debounced burst advances the generation without appending, so a sequence
405/// window that wide can hold fewer lines than the caller asked for, or none.
406///
407/// # Errors
408///
409/// Returns an error if the log file exists but cannot be read.
410pub fn tail_in(dir: &Path, n: usize) -> Result<Vec<Event>> {
411    let log = log_path(dir);
412    if !log.is_file() {
413        return Ok(Vec::new());
414    }
415    let text = fs::read_to_string(&log)?;
416    let mut out: Vec<Event> = text
417        .lines()
418        .rev()
419        .filter(|line| !line.trim().is_empty())
420        .filter_map(|line| serde_json::from_str(line).ok())
421        .take(n)
422        .collect();
423    out.reverse();
424    Ok(out)
425}
426
427/// The last `n` events, for a reader that only wants what is recent.
428///
429/// # Errors
430///
431/// Returns an error if the log file exists but cannot be read.
432pub fn tail_report(layout: &Layout, n: usize) -> Result<String> {
433    let dir = events_dir(layout);
434    let events = tail_in(&dir, n)?;
435    let since_seq = events.first().map(|e| e.seq.saturating_sub(1)).unwrap_or(0);
436    Ok(render_events(&dir, since_seq, events))
437}
438
439/// Add the event files to a `.gitignore` that already exists beside them. Best
440/// effort: a tracker without one is left alone.
441///
442/// # Errors
443///
444/// Returns an error if an existing `.gitignore` cannot be read or appended.
445pub fn ensure_gitignore_hint(dir: &Path) -> Result<()> {
446    let gitignore = dir.join(".gitignore");
447    if !gitignore.is_file() {
448        return Ok(());
449    }
450    let current = fs::read_to_string(&gitignore)?;
451    if current.contains(".vault-events") {
452        return Ok(());
453    }
454    let mut file = OpenOptions::new().append(true).open(&gitignore)?;
455    writeln!(
456        file,
457        "\n# agent ping stream (local)\n{LOG_NAME}\n{GEN_NAME}\n"
458    )?;
459    Ok(())
460}
461
462#[cfg(test)]
463mod tests {
464    use super::*;
465    use crate::config::DEFAULT_PREFIX;
466
467    #[test]
468    fn a_sequence_advances_and_reads_back() {
469        let dir = tempfile::tempdir().unwrap();
470        let d = dir.path();
471        let first = emit_in(d, "ping", None, None, None, Some("a")).unwrap();
472        let second = emit_in(
473            d,
474            "issues_write",
475            Some("atlas"),
476            Some("atlas-1a2b"),
477            Some(Path::new("atlas/issues.org")),
478            None,
479        )
480        .unwrap();
481        assert!(second > first);
482        assert_eq!(generation_in(d), second);
483
484        let events = since_in(d, first, 10).unwrap();
485        assert_eq!(events.len(), 1);
486        assert_eq!(events[0].kind, "issues_write");
487        assert_eq!(events[0].project.as_deref(), Some("atlas"));
488    }
489
490    #[test]
491    fn filters_narrow_by_project_and_kind() {
492        let dir = tempfile::tempdir().unwrap();
493        let d = dir.path();
494        emit_in(d, "ping", None, None, None, None).unwrap();
495        emit_in(d, "manual", Some("atlas"), None, None, None).unwrap();
496        emit_in(d, "manual", Some("beacon"), None, None, None).unwrap();
497
498        let by_project = since_filtered_in(d, 0, 10, Some("atlas"), None).unwrap();
499        assert_eq!(by_project.len(), 1);
500        let by_kind = since_filtered_in(d, 0, 10, None, Some("ping")).unwrap();
501        assert_eq!(by_kind.len(), 1);
502        assert_eq!(by_kind[0].kind, "ping");
503    }
504
505    #[test]
506    fn the_report_carries_a_json_block() {
507        let dir = tempfile::tempdir().unwrap();
508        let layout = Layout::new(dir.path(), DEFAULT_PREFIX);
509        ping_report(&layout, Some("hello")).unwrap();
510        let text = since_report(&layout, 0, 10).unwrap();
511        assert!(text.starts_with("generation="), "{text}");
512        let (_, json) = text.split_once("---json---").expect("json block present");
513        let parsed: serde_json::Value = serde_json::from_str(json.trim()).unwrap();
514        assert_eq!(parsed["events"][0]["kind"], "ping");
515        assert_eq!(parsed["events"][0]["detail"], "hello");
516    }
517
518    #[test]
519    fn the_gitignore_hint_only_touches_an_existing_file() {
520        let dir = tempfile::tempdir().unwrap();
521        ensure_gitignore_hint(dir.path()).unwrap();
522        assert!(!dir.path().join(".gitignore").exists());
523
524        fs::write(dir.path().join(".gitignore"), "target\n").unwrap();
525        ensure_gitignore_hint(dir.path()).unwrap();
526        let text = fs::read_to_string(dir.path().join(".gitignore")).unwrap();
527        assert!(text.contains(LOG_NAME), "{text}");
528        assert!(text.contains(GEN_NAME), "{text}");
529
530        // Running again must not append a second copy.
531        ensure_gitignore_hint(dir.path()).unwrap();
532        let again = fs::read_to_string(dir.path().join(".gitignore")).unwrap();
533        assert_eq!(text, again);
534    }
535
536    #[test]
537    fn concurrent_emits_assign_unique_sequences() {
538        use std::sync::Arc;
539        use std::thread;
540
541        let dir = tempfile::tempdir().unwrap();
542        let d = Arc::new(dir.path().to_path_buf());
543        let handles: Vec<_> = (0..16)
544            .map(|i| {
545                let d = Arc::clone(&d);
546                thread::spawn(move || emit_in(&d, "ping", None, None, None, Some(&format!("{i}"))))
547            })
548            .collect();
549        let mut seqs = Vec::new();
550        for handle in handles {
551            seqs.push(handle.join().unwrap().unwrap());
552        }
553        seqs.sort_unstable();
554        seqs.dedup();
555        assert_eq!(seqs.len(), 16, "duplicate event sequences: {seqs:?}");
556        assert_eq!(generation_in(&d), *seqs.last().unwrap());
557    }
558
559    #[test]
560    fn a_tail_counts_lines_not_sequence_numbers() {
561        let dir = tempfile::tempdir().unwrap();
562        let d = dir.path();
563        for i in 0..5 {
564            emit_in(d, "manual", None, None, None, Some(&format!("event {i}"))).unwrap();
565        }
566        // A ping with no log line of its own would still move the generation
567        // past the window a sequence-derived tail would look in.
568        let tailed = tail_in(d, 3).unwrap();
569        assert_eq!(tailed.len(), 3, "{tailed:?}");
570        assert_eq!(tailed[2].detail.as_deref(), Some("event 4"));
571        assert_eq!(tailed[0].detail.as_deref(), Some("event 2"));
572        assert!(tail_in(d, 50).unwrap().len() == 5);
573    }
574
575    #[test]
576    fn waiting_returns_the_current_generation_on_timeout() {
577        let dir = tempfile::tempdir().unwrap();
578        let layout = Layout::new(dir.path(), DEFAULT_PREFIX);
579        let g = generation(&layout);
580        let waited = wait_generation(&layout, g + 100, 50, 120).unwrap();
581        assert!(waited <= g + 100, "timed out without advancing");
582    }
583}
vissue_core/events.rs

vissue_core/
events.rs
