mati_core/store/db/
history.rs

1//! Version history queries (M-14).
2
3use super::*;
4
5/// A single versioned entry from the SurrealKV history iterator.
6///
7/// Timestamps come from SurrealKV's internal clock (nanoseconds since epoch).
8/// Both seconds and nanoseconds are exposed for callers that need either
9/// precision level.
10#[derive(Debug, Clone, Serialize, Deserialize)]
11pub struct HistoryEntry {
12    /// Timestamp in whole seconds (nanosecond timestamp / 1_000_000_000).
13    pub timestamp_secs: u64,
14    /// Raw nanosecond timestamp from SurrealKV.
15    pub timestamp_ns: u64,
16    /// Deserialized record, `None` for tombstones or corrupt values.
17    pub record: Option<Record>,
18    /// `true` when this version represents a deletion.
19    pub is_tombstone: bool,
20}
21
22/// Shared synchronous implementation for key history queries.
23///
24/// Iterates all versions of `key` using `history_with_options` with the tight
25/// upper bound `key + \0` (not `prefix_end`) to guarantee no adjacent key
26/// spills. Returns entries sorted newest first.
27fn history_impl(txn: &Transaction, key: &str, opts: &HistoryOptions) -> Result<Vec<HistoryEntry>> {
28    // Upper bound: key + NUL byte — tighter than prefix_end which increments
29    // the last byte. This ensures only exact-key versions are returned.
30    let mut upper = key.as_bytes().to_vec();
31    upper.push(0x00);
32
33    let mut cursor = txn.history_with_options(key.as_bytes(), upper.as_slice(), opts)?;
34
35    let mut entries = Vec::new();
36    while cursor.next()? {
37        let key_ref = cursor.key();
38
39        // Guard: only process entries whose user_key matches exactly
40        if key_ref.user_key() != key.as_bytes() {
41            continue;
42        }
43
44        let is_tombstone = key_ref.is_tombstone();
45        let ts_ns = key_ref.timestamp();
46        let ts_secs = ts_ns / 1_000_000_000;
47
48        let record = if is_tombstone {
49            None
50        } else {
51            match cursor.value() {
52                Ok(bytes) => rmps::from_slice::<Record>(&bytes).ok(),
53                Err(_) => None,
54            }
55        };
56
57        entries.push(HistoryEntry {
58            timestamp_secs: ts_secs,
59            timestamp_ns: ts_ns,
60            record,
61            is_tombstone,
62        });
63    }
64
65    // Newest first — SurrealKV history iterator order is not guaranteed to be
66    // reverse-chronological, so sort explicitly.
67    entries.sort_by_key(|e| std::cmp::Reverse(e.timestamp_ns));
68    Ok(entries)
69}
70
71impl Store {
72    /// Return version history for a single key, newest first.
73    ///
74    /// Includes tombstones (deletions). Uses the tight upper bound `key + \0`
75    /// so adjacent keys never spill into the result set.
76    ///
77    /// `limit` caps the number of entries returned; `0` means unlimited.
78    pub fn history(&self, key: &str, limit: usize) -> Result<Vec<HistoryEntry>> {
79        anyhow::ensure!(!key.is_empty(), "history key must not be empty");
80        let tree = self.tree_for(key);
81        let txn = tree.begin_with_mode(Mode::ReadOnly)?;
82
83        let mut opts = HistoryOptions::new().with_tombstones(true);
84        if limit > 0 {
85            opts = opts.with_limit(limit);
86        }
87
88        history_impl(&txn, key, &opts)
89    }
90
91    /// Return version history for a single key since `since_ts` (seconds),
92    /// newest first.
93    ///
94    /// Timestamps are converted to nanoseconds for the SurrealKV range filter.
95    pub fn history_since(
96        &self,
97        key: &str,
98        since_ts: u64,
99        limit: usize,
100    ) -> Result<Vec<HistoryEntry>> {
101        anyhow::ensure!(!key.is_empty(), "history key must not be empty");
102        let tree = self.tree_for(key);
103        let txn = tree.begin_with_mode(Mode::ReadOnly)?;
104
105        let since_ns = since_ts.saturating_mul(1_000_000_000);
106        let mut opts = HistoryOptions::new()
107            .with_tombstones(true)
108            .with_ts_range(since_ns, u64::MAX);
109        if limit > 0 {
110            opts = opts.with_limit(limit);
111        }
112
113        history_impl(&txn, key, &opts)
114    }
115
116    /// Return all records updated since `since_ts` (seconds), newest first.
117    ///
118    /// Scans every knowledge namespace (including `dep:`) and returns records
119    /// whose `updated_at >= since_ts`. Results are sorted by `updated_at`
120    /// descending with secondary sort by key for deterministic ordering.
121    pub async fn records_since(&self, since_ts: u64, limit: usize) -> Result<Vec<Record>> {
122        let mut results = Vec::new();
123        for ns in KNOWLEDGE_NAMESPACES {
124            let records = self.scan_prefix(ns).await?;
125            for r in records {
126                if r.updated_at >= since_ts {
127                    results.push(r);
128                }
129            }
130        }
131        // Newest first, secondary sort by key for determinism
132        results.sort_by(|a, b| {
133            b.updated_at
134                .cmp(&a.updated_at)
135                .then_with(|| a.key.cmp(&b.key))
136        });
137        if limit > 0 && results.len() > limit {
138            results.truncate(limit);
139        }
140        Ok(results)
141    }
142}
mati_core/store/db/history.rs

mati_core/store/db/
history.rs
