mirror_log/
decay.rs

1use rusqlite::{params, Connection, Result};
2use std::time::UNIX_EPOCH;
3
4const DECAY_THRESHOLD_DAYS: i64 = 30;
5const ACCESS_COUNT_THRESHOLD: i64 = 10;
6
7/// Initialize the decay tracking table
8pub fn init_decay_table(conn: &Connection) -> Result<()> {
9    conn.execute(
10        "CREATE TABLE IF NOT EXISTS decay (
11            event_id TEXT PRIMARY KEY,
12            access_count INTEGER NOT NULL DEFAULT 0,
13            last_accessed INTEGER NOT NULL,
14            pinned BOOLEAN NOT NULL DEFAULT 0,
15            created_at INTEGER NOT NULL DEFAULT (unixepoch())
16        )",
17        [],
18    )?;
19    Ok(())
20}
21
22/// Initialize the shadow events table
23pub fn init_shadow_table(conn: &Connection) -> Result<()> {
24    conn.execute(
25        "CREATE TABLE IF NOT EXISTS shadow (
26            id TEXT PRIMARY KEY,
27            timestamp INTEGER NOT NULL,
28            source TEXT NOT NULL,
29            content TEXT NOT NULL,
30            meta TEXT,
31            ingested_at INTEGER NOT NULL,
32            content_hash TEXT,
33            decay_score REAL NOT NULL,
34            flagged_at INTEGER NOT NULL DEFAULT (unixepoch()),
35            FOREIGN KEY (id) REFERENCES events(id) ON DELETE CASCADE
36        )",
37        [],
38    )?;
39    Ok(())
40}
41
42/// Initialize decay-related tables
43pub fn init_decay_tables(conn: &Connection) -> Result<()> {
44    init_decay_table(conn)?;
45    init_shadow_table(conn)?;
46    Ok(())
47}
48
49/// Increment access count for an event and update last_accessed timestamp
50pub fn track_access(conn: &Connection, event_id: &str) -> Result<()> {
51    conn.execute(
52        "INSERT INTO decay (event_id, last_accessed)
53         VALUES (?1, unixepoch())
54         ON CONFLICT(event_id) DO UPDATE SET
55            access_count = decay.access_count + 1,
56            last_accessed = unixepoch(),
57            pinned = decay.pinned",
58        [event_id],
59    )?;
60    Ok(())
61}
62
63/// Get decay score for an event: access_count / days_since_logged
64pub fn get_decay_score(conn: &Connection, event_id: &str) -> Result<f64> {
65    let score: Option<f64> = conn.query_row(
66        "SELECT
67            CASE
68                WHEN last_accessed = 0 THEN 0
69                ELSE CAST(access_count AS REAL) /
70                     GREATEST(1, CAST((unixepoch() - last_accessed) / 86400 AS INTEGER))
71            END
72         FROM decay
73         WHERE event_id = ?1",
74        [event_id],
75        |row| row.get(0),
76    )?;
77    Ok(score.unwrap_or(0.0))
78}
79
80/// Check if an event is flagged for decay (below threshold)
81pub fn is_flagged(conn: &Connection, event_id: &str) -> Result<bool> {
82    let flagged: bool = conn.query_row(
83        "SELECT EXISTS(
84            SELECT 1 FROM decay
85            WHERE event_id = ?1
86            AND access_count < ?2
87            AND (unixepoch() - last_accessed) > ?3 * 86400
88            AND pinned = 0
89        )",
90        params![event_id, ACCESS_COUNT_THRESHOLD, DECAY_THRESHOLD_DAYS],
91        |row| row.get(0),
92    )?;
93    Ok(flagged)
94}
95
96/// Get all events that meet decay criteria
97pub fn get_flagged_events(conn: &Connection) -> Result<Vec<String>> {
98    let mut ids = Vec::new();
99
100    let mut stmt = conn.prepare(
101        "SELECT e.id FROM events e
102         JOIN decay d ON e.id = d.event_id
103         WHERE d.access_count < ?
104         AND (unixepoch() - d.last_accessed) > ? * 86400
105         AND d.pinned = 0
106         ORDER BY d.last_accessed ASC",
107    )?;
108
109    let rows = stmt.query_map(
110        params![ACCESS_COUNT_THRESHOLD, DECAY_THRESHOLD_DAYS],
111        |row| row.get(0),
112    )?;
113
114    for row in rows {
115        ids.push(row?);
116    }
117
118    Ok(ids)
119}
120
121/// Move flagged events to shadow table
122pub fn move_to_shadow(conn: &Connection) -> Result<usize> {
123    let flagged_ids = get_flagged_events(conn)?;
124
125    if flagged_ids.is_empty() {
126        return Ok(0);
127    }
128
129    // Get decay scores for flagged events
130    let mut scores: Vec<(String, f64)> = Vec::new();
131    for id in &flagged_ids {
132        if let Ok(score) = get_decay_score(conn, id) {
133            scores.push((id.clone(), score));
134        }
135    }
136
137    // Move events to shadow table
138    let mut moved = 0;
139    for (id, score) in scores {
140        let now = UNIX_EPOCH.elapsed().unwrap().as_secs() as i64;
141
142        conn.execute(
143            "INSERT INTO shadow (id, timestamp, source, content, meta, ingested_at, content_hash, decay_score, flagged_at)
144             SELECT id, timestamp, source, content, meta, ingested_at, content_hash, ?1, ?2
145             FROM events
146             WHERE id = ?3",
147            params![score, now, id],
148        )?;
149
150        moved += 1;
151    }
152
153    Ok(moved)
154}
155
156/// Get all shadowed events
157pub fn get_shadow_events(conn: &Connection) -> Result<Vec<ShadowEvent>> {
158    let mut events = Vec::new();
159
160    let mut stmt = conn.prepare(
161        "SELECT id, timestamp, source, content, meta, ingested_at, content_hash, decay_score, flagged_at
162         FROM shadow
163         ORDER BY flagged_at DESC"
164    )?;
165
166    let rows = stmt.query_map([], |row| {
167        Ok(ShadowEvent {
168            id: row.get(0)?,
169            timestamp: row.get(1)?,
170            source: row.get(2)?,
171            content: row.get(3)?,
172            meta: row.get(4)?,
173            ingested_at: row.get(5)?,
174            content_hash: row.get(6)?,
175            decay_score: row.get(7)?,
176            flagged_at: row.get(8)?,
177        })
178    })?;
179
180    for row in rows {
181        events.push(row?);
182    }
183
184    Ok(events)
185}
186
187/// Pin an event (immune to decay)
188pub fn pin_event(conn: &Connection, event_id: &str) -> Result<()> {
189    conn.execute(
190        "UPDATE decay SET pinned = 1 WHERE event_id = ?1",
191        [event_id],
192    )?;
193    Ok(())
194}
195
196/// Unpin an event
197pub fn unpin_event(conn: &Connection, event_id: &str) -> Result<()> {
198    conn.execute(
199        "UPDATE decay SET pinned = 0 WHERE event_id = ?1",
200        [event_id],
201    )?;
202    Ok(())
203}
204
205/// Restore an event from shadow back to main table
206pub fn restore_from_shadow(conn: &Connection, event_id: &str) -> Result<()> {
207    // Remove from shadow
208    conn.execute("DELETE FROM shadow WHERE id = ?1", [event_id])?;
209
210    // Reset decay tracking
211    conn.execute(
212        "INSERT INTO decay (event_id, access_count, last_accessed, pinned)
213         VALUES (?1, 0, unixepoch(), 0)
214         ON CONFLICT(event_id) DO UPDATE SET
215            access_count = 0,
216            last_accessed = unixepoch(),
217            pinned = 0",
218        [event_id],
219    )?;
220
221    Ok(())
222}
223
224/// Get decay statistics
225pub fn get_decay_stats(conn: &Connection) -> Result<DecayStats> {
226    let total_events: i64 = conn.query_row("SELECT COUNT(*) FROM decay", [], |row| row.get(0))?;
227
228    let total_access: i64 =
229        conn.query_row("SELECT SUM(access_count) FROM decay", [], |row| row.get(0))?;
230
231    let pinned_count: i64 =
232        conn.query_row("SELECT COUNT(*) FROM decay WHERE pinned = 1", [], |row| {
233            row.get(0)
234        })?;
235
236    let avg_access: f64 = if total_events > 0 {
237        total_access as f64 / total_events as f64
238    } else {
239        0.0
240    };
241
242    let flagged_count: i64 = conn.query_row(
243        "SELECT COUNT(*) FROM decay WHERE access_count < ? AND (unixepoch() - last_accessed) > ? * 86400 AND pinned = 0",
244        params![ACCESS_COUNT_THRESHOLD, DECAY_THRESHOLD_DAYS],
245        |row| row.get(0),
246    )?;
247
248    Ok(DecayStats {
249        total_events,
250        total_access,
251        pinned_count,
252        avg_access,
253        flagged_count,
254    })
255}
256
257#[derive(Debug, Clone)]
258pub struct ShadowEvent {
259    pub id: String,
260    pub timestamp: i64,
261    pub source: String,
262    pub content: String,
263    pub meta: Option<String>,
264    pub ingested_at: i64,
265    pub content_hash: Option<String>,
266    pub decay_score: f64,
267    pub flagged_at: i64,
268}
269
270#[derive(Debug, Clone)]
271pub struct DecayStats {
272    pub total_events: i64,
273    pub total_access: i64,
274    pub pinned_count: i64,
275    pub avg_access: f64,
276    pub flagged_count: i64,
277}
mirror_log/decay.rs

mirror_log/
decay.rs
