bamboo_memory/ledger_store/
store.rs

1use std::collections::HashSet;
2use std::io;
3use std::path::PathBuf;
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::sync::{Arc, OnceLock};
6use std::time::{SystemTime, UNIX_EPOCH};
7
8use bamboo_domain::ledger::{LedgerRecord, LedgerScope, RecordKind, RecordStatus};
9use bamboo_domain::TaskPriority;
10use chrono::{DateTime, Duration, Utc};
11use dashmap::DashMap;
12use serde::{Deserialize, Serialize};
13use tokio::fs;
14use tokio::sync::Mutex;
15
16use crate::atomic_fs::{atomic_write, atomic_write_batch};
17
18use super::paths::LedgerPathResolver;
19use super::{
20    build_status_index, build_time_index, build_todo_markdown_view, parse_record_document,
21    render_record_document, validate_record_id, validate_record_title, LedgerAuditEntry,
22    LedgerRecordDocument, TimeIndex, AGENDA_VIEW_FILE, AUDIT_LOG_FILE, BY_STATUS_INDEX_FILE,
23    BY_TIME_INDEX_FILE, TODO_VIEW_FILE,
24};
25
26/// Process-wide per-scope write locks, mirroring the memory store's
27/// serialization discipline: every read-modify-write plus artifact refresh for
28/// a scope happens under its lock.
29fn scope_locks() -> &'static DashMap<PathBuf, Arc<Mutex<()>>> {
30    static SCOPE_LOCKS: OnceLock<DashMap<PathBuf, Arc<Mutex<()>>>> = OnceLock::new();
31    SCOPE_LOCKS.get_or_init(DashMap::new)
32}
33
34static RECORD_SEQ: AtomicU64 = AtomicU64::new(0);
35
36/// A process-unique record id (`rec_<nanos-hex><seq>`); no uuid dependency
37/// needed at human ledger volumes.
38pub fn new_record_id() -> String {
39    let nanos = SystemTime::now()
40        .duration_since(UNIX_EPOCH)
41        .map(|d| d.as_nanos())
42        .unwrap_or(0);
43    let seq = RECORD_SEQ.fetch_add(1, Ordering::Relaxed) & 0xfff;
44    format!("rec_{nanos:x}{seq:03x}")
45}
46
47/// Filter for [`LedgerStore::list_records`]. Empty filter = every non-terminal
48/// record.
49#[derive(Debug, Clone, Default)]
50pub struct RecordFilter {
51    pub statuses: Option<HashSet<RecordStatus>>,
52    pub kinds: Option<HashSet<RecordKind>>,
53    pub tags: Vec<String>,
54    pub parent_id: Option<String>,
55    pub anchor_before: Option<DateTime<Utc>>,
56    pub anchor_after: Option<DateTime<Utc>>,
57    /// Include Done/Cancelled/Expired records when no explicit `statuses`
58    /// filter is set.
59    pub include_terminal: bool,
60    pub limit: Option<usize>,
61}
62
63impl RecordFilter {
64    fn matches(&self, record: &LedgerRecord) -> bool {
65        if let Some(statuses) = &self.statuses {
66            if !statuses.contains(&record.status) {
67                return false;
68            }
69        } else if !self.include_terminal && record.status.is_terminal() {
70            return false;
71        }
72        if let Some(kinds) = &self.kinds {
73            if !kinds.contains(&record.kind) {
74                return false;
75            }
76        }
77        if let Some(parent_id) = &self.parent_id {
78            if record.relations.parent_id.as_deref() != Some(parent_id.as_str()) {
79                return false;
80            }
81        }
82        if !self.tags.is_empty()
83            && !self
84                .tags
85                .iter()
86                .all(|tag| record.tags.iter().any(|have| have == tag))
87        {
88            return false;
89        }
90        let anchor = record.time_anchor();
91        if let Some(before) = self.anchor_before {
92            match anchor {
93                Some(anchor) if anchor < before => {}
94                _ => return false,
95            }
96        }
97        if let Some(after) = self.anchor_after {
98            match anchor {
99                Some(anchor) if anchor >= after => {}
100                _ => return false,
101            }
102        }
103        true
104    }
105}
106
107/// One agenda line: a record plus where it came from, so multi-scope agendas
108/// (personal + current project) stay attributable.
109#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
110pub struct AgendaItem {
111    pub id: String,
112    pub scope: LedgerScope,
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub project_key: Option<String>,
115    pub kind: RecordKind,
116    pub title: String,
117    pub status: RecordStatus,
118    pub priority: TaskPriority,
119    #[serde(default, skip_serializing_if = "Option::is_none")]
120    pub anchor_at: Option<DateTime<Utc>>,
121    #[serde(default, skip_serializing_if = "Option::is_none")]
122    pub due_at: Option<DateTime<Utc>>,
123}
124
125/// Time-bucketed agenda: what the assistant leads with.
126#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq, Eq)]
127pub struct AgendaSnapshot {
128    pub generated_at: DateTime<Utc>,
129    /// Non-terminal records whose anchor is in the past.
130    pub overdue: Vec<AgendaItem>,
131    /// Anchored within the next 24 hours.
132    pub today: Vec<AgendaItem>,
133    /// Anchored after 24h and within the horizon.
134    pub upcoming: Vec<AgendaItem>,
135    /// Open records with no time anchor (top priorities first, capped).
136    pub undated: Vec<AgendaItem>,
137}
138
139impl AgendaSnapshot {
140    pub fn is_empty(&self) -> bool {
141        self.overdue.is_empty()
142            && self.today.is_empty()
143            && self.upcoming.is_empty()
144            && self.undated.is_empty()
145    }
146}
147
148const UNDATED_AGENDA_CAP: usize = 10;
149
150fn priority_rank(priority: &TaskPriority) -> u8 {
151    match priority {
152        TaskPriority::Critical => 3,
153        TaskPriority::High => 2,
154        TaskPriority::Medium => 1,
155        TaskPriority::Low => 0,
156    }
157}
158
159fn agenda_item(record: &LedgerRecord) -> AgendaItem {
160    AgendaItem {
161        id: record.id.clone(),
162        scope: record.scope,
163        project_key: record.project_key.clone(),
164        kind: record.kind.clone(),
165        title: record.title.clone(),
166        status: record.status,
167        priority: record.priority.clone(),
168        anchor_at: record.time_anchor(),
169        due_at: record.time.due_at,
170    }
171}
172
173/// Bucket records into an agenda. Pure so views, the tool, and the prompt
174/// layer all share one definition of "overdue"/"today".
175pub fn build_agenda_snapshot(
176    records: &[LedgerRecord],
177    now: DateTime<Utc>,
178    horizon_days: i64,
179) -> AgendaSnapshot {
180    let day_ahead = now + Duration::hours(24);
181    let horizon = now + Duration::days(horizon_days.max(1));
182    let mut snapshot = AgendaSnapshot {
183        generated_at: now,
184        ..AgendaSnapshot::default()
185    };
186
187    for record in records {
188        if record.status.is_terminal() {
189            continue;
190        }
191        match record.time_anchor() {
192            Some(anchor) if anchor < now => snapshot.overdue.push(agenda_item(record)),
193            Some(anchor) if anchor < day_ahead => snapshot.today.push(agenda_item(record)),
194            Some(anchor) if anchor < horizon => snapshot.upcoming.push(agenda_item(record)),
195            Some(_) => {}
196            None => snapshot.undated.push(agenda_item(record)),
197        }
198    }
199
200    let by_anchor = |left: &AgendaItem, right: &AgendaItem| {
201        left.anchor_at
202            .cmp(&right.anchor_at)
203            .then_with(|| left.id.cmp(&right.id))
204    };
205    snapshot.overdue.sort_by(by_anchor);
206    snapshot.today.sort_by(by_anchor);
207    snapshot.upcoming.sort_by(by_anchor);
208    snapshot.undated.sort_by(|left, right| {
209        priority_rank(&right.priority)
210            .cmp(&priority_rank(&left.priority))
211            .then_with(|| left.id.cmp(&right.id))
212    });
213    snapshot.undated.truncate(UNDATED_AGENDA_CAP);
214    snapshot
215}
216
217/// Render an agenda as markdown — used for `AGENDA.md` and reused by the
218/// prompt-injection layer.
219pub fn build_agenda_markdown(snapshot: &AgendaSnapshot) -> String {
220    fn push_section(out: &mut String, heading: &str, items: &[AgendaItem]) {
221        if items.is_empty() {
222            return;
223        }
224        out.push_str(&format!("## {heading}\n"));
225        for item in items {
226            let when = item
227                .anchor_at
228                .map(|at| format!(" — {}", at.format("%Y-%m-%d %H:%M UTC")))
229                .unwrap_or_default();
230            out.push_str(&format!(
231                "- `{}` [{}] {}{}\n",
232                item.id,
233                item.kind.as_str(),
234                item.title,
235                when,
236            ));
237        }
238        out.push('\n');
239    }
240
241    let mut out = String::from("# Ledger Agenda\n\n");
242    if snapshot.is_empty() {
243        out.push_str("_(nothing scheduled or open)_\n");
244        return out;
245    }
246    push_section(&mut out, "Overdue", &snapshot.overdue);
247    push_section(&mut out, "Next 24 hours", &snapshot.today);
248    push_section(&mut out, "Upcoming", &snapshot.upcoming);
249    push_section(&mut out, "Open (no date)", &snapshot.undated);
250    out
251}
252
253/// Persistence for ledger records under `{data_dir}/ledger/v1`.
254#[derive(Debug, Clone)]
255pub struct LedgerStore {
256    resolver: LedgerPathResolver,
257}
258
259impl Default for LedgerStore {
260    fn default() -> Self {
261        Self::with_defaults()
262    }
263}
264
265impl LedgerStore {
266    pub fn new(data_dir: impl Into<PathBuf>) -> Self {
267        Self {
268            resolver: LedgerPathResolver::from_data_dir(data_dir),
269        }
270    }
271
272    pub fn with_defaults() -> Self {
273        Self {
274            resolver: LedgerPathResolver::default(),
275        }
276    }
277
278    pub fn resolver(&self) -> &LedgerPathResolver {
279        &self.resolver
280    }
281
282    fn scope_lock(&self, scope: LedgerScope, project_key: Option<&str>) -> Arc<Mutex<()>> {
283        scope_locks()
284            .entry(self.resolver.scope_root(scope, project_key))
285            .or_insert_with(|| Arc::new(Mutex::new(())))
286            .clone()
287    }
288
289    /// Create or replace a record document. Preserves `created_at` (and the
290    /// body, when `body` is `None`) of an existing record with the same id;
291    /// always bumps `updated_at`.
292    pub async fn write_record(
293        &self,
294        mut record: LedgerRecord,
295        body: Option<String>,
296    ) -> io::Result<LedgerRecordDocument> {
297        validate_record_id(&record.id)?;
298        validate_record_title(&record.title)?;
299        if record.scope == LedgerScope::Project && record.project_key.is_none() {
300            return Err(io::Error::new(
301                io::ErrorKind::InvalidInput,
302                "project-scoped records require a project_key",
303            ));
304        }
305
306        let scope = record.scope;
307        let project_key = record.project_key.clone();
308        let lock = self.scope_lock(scope, project_key.as_deref());
309        let _guard = lock.lock().await;
310
311        let path = self
312            .resolver
313            .record_path(scope, project_key.as_deref(), &record.id);
314        let existing = self.load_record_at(&path).await?;
315        let action = if existing.is_some() {
316            "update"
317        } else {
318            "create"
319        };
320        let body = match (body, &existing) {
321            (Some(body), _) => body,
322            (None, Some(existing)) => existing.body.clone(),
323            (None, None) => String::new(),
324        };
325        if let Some(existing) = &existing {
326            record.created_at = existing.record.created_at;
327        }
328        record.updated_at = Utc::now();
329
330        let rendered = render_record_document(&record, &body)?;
331        atomic_write(&path, rendered.as_bytes()).await?;
332        self.refresh_scope_artifacts(scope, project_key.as_deref())
333            .await?;
334        self.append_audit(
335            scope,
336            project_key.as_deref(),
337            &record.id,
338            action,
339            &record.title,
340        )
341        .await?;
342
343        Ok(LedgerRecordDocument { record, body, path })
344    }
345
346    pub async fn get_record(
347        &self,
348        scope: LedgerScope,
349        project_key: Option<&str>,
350        record_id: &str,
351    ) -> io::Result<Option<LedgerRecordDocument>> {
352        let record_id = validate_record_id(record_id)?;
353        let path = self.resolver.record_path(scope, project_key, record_id);
354        self.load_record_at(&path).await
355    }
356
357    /// Transition a record's status (recording history) and refresh artifacts.
358    /// Returns the updated document, or `None` when either the record does not
359    /// exist or the status is already the target (no-op).
360    pub async fn transition_record(
361        &self,
362        scope: LedgerScope,
363        project_key: Option<&str>,
364        record_id: &str,
365        status: RecordStatus,
366        reason: Option<&str>,
367    ) -> io::Result<Option<LedgerRecordDocument>> {
368        let record_id = validate_record_id(record_id)?;
369        let lock = self.scope_lock(scope, project_key);
370        let _guard = lock.lock().await;
371
372        let path = self.resolver.record_path(scope, project_key, record_id);
373        let Some(mut doc) = self.load_record_at(&path).await? else {
374            return Ok(None);
375        };
376        if !doc.record.transition_to(status, reason) {
377            return Ok(None);
378        }
379
380        let rendered = render_record_document(&doc.record, &doc.body)?;
381        atomic_write(&path, rendered.as_bytes()).await?;
382        self.refresh_scope_artifacts(scope, project_key).await?;
383        self.append_audit(
384            scope,
385            project_key,
386            record_id,
387            &format!("transition:{}", status.as_str()),
388            reason.unwrap_or(""),
389        )
390        .await?;
391        Ok(Some(doc))
392    }
393
394    /// Load and filter a scope's records, anchored records first (ascending),
395    /// undated records after (most recently updated first).
396    pub async fn list_records(
397        &self,
398        scope: LedgerScope,
399        project_key: Option<&str>,
400        filter: &RecordFilter,
401    ) -> io::Result<Vec<LedgerRecordDocument>> {
402        let mut docs: Vec<LedgerRecordDocument> = self
403            .load_scope_records(scope, project_key)
404            .await?
405            .into_iter()
406            .filter(|doc| filter.matches(&doc.record))
407            .collect();
408        docs.sort_by(|left, right| {
409            match (left.record.time_anchor(), right.record.time_anchor()) {
410                (Some(l), Some(r)) => l.cmp(&r),
411                (Some(_), None) => std::cmp::Ordering::Less,
412                (None, Some(_)) => std::cmp::Ordering::Greater,
413                (None, None) => right.record.updated_at.cmp(&left.record.updated_at),
414            }
415            .then_with(|| left.record.id.cmp(&right.record.id))
416        });
417        if let Some(limit) = filter.limit {
418            docs.truncate(limit);
419        }
420        Ok(docs)
421    }
422
423    /// Build a time-bucketed agenda across one or more scopes (typically the
424    /// global scope plus the current project's).
425    pub async fn agenda(
426        &self,
427        scopes: &[(LedgerScope, Option<String>)],
428        now: DateTime<Utc>,
429        horizon_days: i64,
430    ) -> io::Result<AgendaSnapshot> {
431        let mut records: Vec<LedgerRecord> = Vec::new();
432        for (scope, project_key) in scopes {
433            let docs = self
434                .load_scope_records(*scope, project_key.as_deref())
435                .await?;
436            records.extend(docs.into_iter().map(|doc| doc.record));
437        }
438        Ok(build_agenda_snapshot(&records, now, horizon_days))
439    }
440
441    /// Cheap agenda read for the prompt layer: derived purely from
442    /// `by_time.json` plus the undated slice of `by_status.json` would lose
443    /// priority ordering, so today it re-reads records; scale is human-sized.
444    pub async fn read_time_index(
445        &self,
446        scope: LedgerScope,
447        project_key: Option<&str>,
448    ) -> io::Result<Option<TimeIndex>> {
449        let path = self
450            .resolver
451            .indexes_dir(scope, project_key)
452            .join(BY_TIME_INDEX_FILE);
453        match fs::read_to_string(&path).await {
454            Ok(raw) => Ok(serde_json::from_str(&raw).ok()),
455            Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None),
456            Err(error) => Err(error),
457        }
458    }
459
460    /// Project keys that have a ledger scope directory on disk — the iteration
461    /// surface for background maintenance across every scope.
462    pub async fn list_project_keys(&self) -> io::Result<Vec<String>> {
463        let dir = self.resolver.projects_root();
464        let mut reader = match fs::read_dir(&dir).await {
465            Ok(reader) => reader,
466            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
467            Err(error) => return Err(error),
468        };
469        let mut keys = Vec::new();
470        while let Some(entry) = reader.next_entry().await? {
471            if entry
472                .file_type()
473                .await
474                .map(|ft| ft.is_dir())
475                .unwrap_or(false)
476            {
477                if let Some(name) = entry.file_name().to_str() {
478                    keys.push(name.to_string());
479                }
480            }
481        }
482        keys.sort();
483        Ok(keys)
484    }
485
486    /// Rebuild every derived artifact for a scope from the record documents.
487    pub async fn rebuild_scope(
488        &self,
489        scope: LedgerScope,
490        project_key: Option<&str>,
491    ) -> io::Result<()> {
492        let lock = self.scope_lock(scope, project_key);
493        let _guard = lock.lock().await;
494        self.refresh_scope_artifacts(scope, project_key).await
495    }
496
497    async fn load_record_at(&self, path: &PathBuf) -> io::Result<Option<LedgerRecordDocument>> {
498        let raw = match fs::read_to_string(path).await {
499            Ok(raw) => raw,
500            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
501            Err(error) => return Err(error),
502        };
503        let (record, body) = parse_record_document(&raw)?;
504        Ok(Some(LedgerRecordDocument {
505            record,
506            body,
507            path: path.clone(),
508        }))
509    }
510
511    async fn load_scope_records(
512        &self,
513        scope: LedgerScope,
514        project_key: Option<&str>,
515    ) -> io::Result<Vec<LedgerRecordDocument>> {
516        let dir = self.resolver.records_dir(scope, project_key);
517        let mut reader = match fs::read_dir(&dir).await {
518            Ok(reader) => reader,
519            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
520            Err(error) => return Err(error),
521        };
522        let mut docs = Vec::new();
523        while let Some(entry) = reader.next_entry().await? {
524            let path = entry.path();
525            if path.extension().and_then(|ext| ext.to_str()) != Some("md") {
526                continue;
527            }
528            match self.load_record_at(&path).await {
529                Ok(Some(doc)) => docs.push(doc),
530                Ok(None) => {}
531                // A corrupt document must not take the whole ledger down; it
532                // stays on disk for manual repair and is skipped here.
533                Err(error) => {
534                    tracing::warn!(path = %path.display(), error = %error, "skipping unreadable ledger record");
535                }
536            }
537        }
538        Ok(docs)
539    }
540
541    async fn refresh_scope_artifacts(
542        &self,
543        scope: LedgerScope,
544        project_key: Option<&str>,
545    ) -> io::Result<()> {
546        let records: Vec<LedgerRecord> = self
547            .load_scope_records(scope, project_key)
548            .await?
549            .into_iter()
550            .map(|doc| doc.record)
551            .collect();
552        let now = Utc::now();
553
554        let time_index = build_time_index(&records, now);
555        let status_index = build_status_index(&records, now);
556        let agenda = build_agenda_snapshot(&records, now, 7);
557
558        let indexes_dir = self.resolver.indexes_dir(scope, project_key);
559        let views_dir = self.resolver.views_dir(scope, project_key);
560        atomic_write_batch(vec![
561            (
562                indexes_dir.join(BY_TIME_INDEX_FILE),
563                serde_json::to_vec_pretty(&time_index)
564                    .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?,
565            ),
566            (
567                indexes_dir.join(BY_STATUS_INDEX_FILE),
568                serde_json::to_vec_pretty(&status_index)
569                    .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?,
570            ),
571            (
572                views_dir.join(AGENDA_VIEW_FILE),
573                build_agenda_markdown(&agenda).into_bytes(),
574            ),
575            (
576                views_dir.join(TODO_VIEW_FILE),
577                build_todo_markdown_view(&records).into_bytes(),
578            ),
579        ])
580        .await
581    }
582
583    async fn append_audit(
584        &self,
585        scope: LedgerScope,
586        project_key: Option<&str>,
587        record_id: &str,
588        action: &str,
589        summary: &str,
590    ) -> io::Result<()> {
591        let entry = LedgerAuditEntry {
592            timestamp: Utc::now().to_rfc3339(),
593            action: action.to_string(),
594            scope,
595            project_key: project_key.map(ToOwned::to_owned),
596            record_id: record_id.to_string(),
597            summary: summary.chars().take(200).collect(),
598        };
599        let path = self
600            .resolver
601            .logs_dir(scope, project_key)
602            .join(AUDIT_LOG_FILE);
603        if let Some(parent) = path.parent() {
604            fs::create_dir_all(parent).await?;
605        }
606        let mut line = serde_json::to_string(&entry)
607            .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
608        line.push('\n');
609        use tokio::io::AsyncWriteExt;
610        let mut file = fs::OpenOptions::new()
611            .create(true)
612            .append(true)
613            .open(&path)
614            .await?;
615        file.write_all(line.as_bytes()).await?;
616        file.sync_all().await
617    }
618}
619
620#[cfg(test)]
621mod tests {
622    use super::*;
623    use bamboo_domain::ledger::RecordKind;
624    use chrono::TimeZone;
625    use tempfile::tempdir;
626
627    fn utc(y: i32, mo: u32, d: u32, h: u32) -> DateTime<Utc> {
628        Utc.with_ymd_and_hms(y, mo, d, h, 0, 0).unwrap()
629    }
630
631    fn todo(id: &str, title: &str, due: Option<DateTime<Utc>>) -> LedgerRecord {
632        let mut record = LedgerRecord::new(id, RecordKind::Todo, title);
633        record.time.due_at = due;
634        record
635    }
636
637    #[tokio::test]
638    async fn write_get_transition_round_trip() {
639        let dir = tempdir().unwrap();
640        let store = LedgerStore::new(dir.path());
641
642        let record = todo("rec_passport", "Renew passport", Some(utc(2026, 8, 1, 9)));
643        let written = store
644            .write_record(record, Some("Bring the old one.".to_string()))
645            .await
646            .unwrap();
647        assert!(written
648            .path
649            .ends_with("scopes/global/records/rec_passport.md"));
650
651        let fetched = store
652            .get_record(LedgerScope::Global, None, "rec_passport")
653            .await
654            .unwrap()
655            .unwrap();
656        assert_eq!(fetched.record.title, "Renew passport");
657        assert_eq!(fetched.body, "Bring the old one.");
658
659        let done = store
660            .transition_record(
661                LedgerScope::Global,
662                None,
663                "rec_passport",
664                RecordStatus::Done,
665                Some("picked it up"),
666            )
667            .await
668            .unwrap()
669            .unwrap();
670        assert_eq!(done.record.status, RecordStatus::Done);
671        assert_eq!(done.record.transitions.len(), 1);
672
673        // No-op transition returns None and records nothing further.
674        let noop = store
675            .transition_record(
676                LedgerScope::Global,
677                None,
678                "rec_passport",
679                RecordStatus::Done,
680                None,
681            )
682            .await
683            .unwrap();
684        assert!(noop.is_none());
685    }
686
687    #[tokio::test]
688    async fn update_preserves_created_at_and_body() {
689        let dir = tempdir().unwrap();
690        let store = LedgerStore::new(dir.path());
691
692        let first = store
693            .write_record(
694                todo("rec_1", "Original", None),
695                Some("original body".to_string()),
696            )
697            .await
698            .unwrap();
699
700        let mut updated = first.record.clone();
701        updated.title = "Renamed".to_string();
702        let second = store.write_record(updated, None).await.unwrap();
703
704        assert_eq!(second.record.created_at, first.record.created_at);
705        assert_eq!(second.body, "original body");
706        assert!(second.record.updated_at >= first.record.updated_at);
707        assert_eq!(second.record.title, "Renamed");
708    }
709
710    #[tokio::test]
711    async fn project_scope_requires_project_key_and_separates_records() {
712        let dir = tempdir().unwrap();
713        let store = LedgerStore::new(dir.path());
714
715        let mut orphan = todo("rec_orphan", "No key", None);
716        orphan.scope = LedgerScope::Project;
717        assert!(store.write_record(orphan, None).await.is_err());
718
719        let mut scoped = todo("rec_scoped", "Project item", None);
720        scoped.scope = LedgerScope::Project;
721        scoped.project_key = Some("proj-1".to_string());
722        store.write_record(scoped, None).await.unwrap();
723
724        assert!(store
725            .get_record(LedgerScope::Global, None, "rec_scoped")
726            .await
727            .unwrap()
728            .is_none());
729        assert!(store
730            .get_record(LedgerScope::Project, Some("proj-1"), "rec_scoped")
731            .await
732            .unwrap()
733            .is_some());
734    }
735
736    #[tokio::test]
737    async fn list_records_filters_and_sorts_anchored_first() {
738        let dir = tempdir().unwrap();
739        let store = LedgerStore::new(dir.path());
740
741        store
742            .write_record(todo("rec_later", "Later", Some(utc(2026, 8, 2, 9))), None)
743            .await
744            .unwrap();
745        store
746            .write_record(todo("rec_soon", "Soon", Some(utc(2026, 7, 20, 9))), None)
747            .await
748            .unwrap();
749        store
750            .write_record(todo("rec_undated", "Undated", None), None)
751            .await
752            .unwrap();
753        store
754            .transition_record(
755                LedgerScope::Global,
756                None,
757                "rec_later",
758                RecordStatus::Cancelled,
759                None,
760            )
761            .await
762            .unwrap();
763
764        let open = store
765            .list_records(LedgerScope::Global, None, &RecordFilter::default())
766            .await
767            .unwrap();
768        let ids: Vec<&str> = open.iter().map(|doc| doc.record.id.as_str()).collect();
769        assert_eq!(ids, vec!["rec_soon", "rec_undated"]);
770
771        let terminal_only = store
772            .list_records(
773                LedgerScope::Global,
774                None,
775                &RecordFilter {
776                    statuses: Some(HashSet::from([RecordStatus::Cancelled])),
777                    ..RecordFilter::default()
778                },
779            )
780            .await
781            .unwrap();
782        assert_eq!(terminal_only.len(), 1);
783        assert_eq!(terminal_only[0].record.id, "rec_later");
784    }
785
786    #[tokio::test]
787    async fn writes_refresh_indexes_views_and_audit_log() {
788        let dir = tempdir().unwrap();
789        let store = LedgerStore::new(dir.path());
790
791        // Relative due date: the on-write artifact refresh buckets against the
792        // real clock, so a fixed date would drift out of the 7-day window.
793        store
794            .write_record(
795                todo("rec_1", "Send report", Some(Utc::now() + Duration::days(2))),
796                None,
797            )
798            .await
799            .unwrap();
800
801        let scope_root = store.resolver().scope_root(LedgerScope::Global, None);
802        let time_index: TimeIndex = serde_json::from_str(
803            &std::fs::read_to_string(scope_root.join("indexes").join(BY_TIME_INDEX_FILE)).unwrap(),
804        )
805        .unwrap();
806        assert_eq!(time_index.items.len(), 1);
807        assert_eq!(time_index.items[0].id, "rec_1");
808
809        let agenda_view =
810            std::fs::read_to_string(scope_root.join("views").join(AGENDA_VIEW_FILE)).unwrap();
811        assert!(agenda_view.contains("Send report"));
812
813        let audit = std::fs::read_to_string(scope_root.join("logs").join(AUDIT_LOG_FILE)).unwrap();
814        assert!(audit.contains("\"action\":\"create\""));
815    }
816
817    #[tokio::test]
818    async fn agenda_buckets_by_time_distance() {
819        let dir = tempdir().unwrap();
820        let store = LedgerStore::new(dir.path());
821        let now = utc(2026, 7, 13, 12);
822
823        store
824            .write_record(
825                todo("rec_overdue", "Yesterday", Some(utc(2026, 7, 12, 9))),
826                None,
827            )
828            .await
829            .unwrap();
830        store
831            .write_record(
832                todo("rec_today", "Tonight", Some(utc(2026, 7, 13, 20))),
833                None,
834            )
835            .await
836            .unwrap();
837        store
838            .write_record(
839                todo("rec_week", "This week", Some(utc(2026, 7, 16, 9))),
840                None,
841            )
842            .await
843            .unwrap();
844        store
845            .write_record(
846                todo("rec_far", "Next month", Some(utc(2026, 8, 20, 9))),
847                None,
848            )
849            .await
850            .unwrap();
851        store
852            .write_record(todo("rec_undated", "Someday", None), None)
853            .await
854            .unwrap();
855
856        let snapshot = store
857            .agenda(&[(LedgerScope::Global, None)], now, 7)
858            .await
859            .unwrap();
860        let ids =
861            |items: &[AgendaItem]| items.iter().map(|item| item.id.clone()).collect::<Vec<_>>();
862        assert_eq!(ids(&snapshot.overdue), vec!["rec_overdue"]);
863        assert_eq!(ids(&snapshot.today), vec!["rec_today"]);
864        assert_eq!(ids(&snapshot.upcoming), vec!["rec_week"]);
865        assert_eq!(ids(&snapshot.undated), vec!["rec_undated"]);
866
867        let markdown = build_agenda_markdown(&snapshot);
868        assert!(markdown.contains("## Overdue"));
869        assert!(markdown.contains("rec_today"));
870        assert!(!markdown.contains("rec_far"));
871    }
872
873    #[test]
874    fn new_record_ids_are_unique_and_path_safe() {
875        let a = new_record_id();
876        let b = new_record_id();
877        assert_ne!(a, b);
878        assert!(validate_record_id(&a).is_ok());
879    }
880}
bamboo_memory/ledger_store/store.rs

bamboo_memory/ledger_store/
store.rs
