369 lines
12 KiB
Rust
369 lines
12 KiB
Rust
//! Harness-local reminder store — the persistent, DB-backed half of the
|
|
//! in-container reminders migration. Mirrors `todos.rs`'s shape: one
|
|
//! sqlite db under the harness dir, single-agent
|
|
//! scope (no `agent` column — every row belongs to this agent, unlike the
|
|
//! old c0re-side store which served every agent in the hive).
|
|
//!
|
|
//! Unlike todos (discovered passively via `get_loose_ends`), a reminder is
|
|
//! *active push*: it must fire into the turn loop at `due_at` with its own
|
|
//! message body (see `reminder_timer.rs`). Delivery is *soft*-deleted
|
|
//! (`sent_at` set, not `DELETE`) so the harness's own `/api/stats` reminder
|
|
//! rollup can still report scheduled/delivered/pending counts over a
|
|
//! trailing window (mirrors the old c0re `ReminderStats` shape exactly). A
|
|
//! periodic prune (`vacuum.rs`) reaps old delivered rows so the table
|
|
//! doesn't grow unbounded.
|
|
|
|
use std::path::Path;
|
|
use std::sync::Mutex;
|
|
|
|
use anyhow::{Context, Result};
|
|
use chrono::{DateTime, Utc};
|
|
use hive_sh4re::approvals::ReminderStats;
|
|
use hive_sh4re::wire_time;
|
|
use rusqlite::{Connection, params};
|
|
|
|
const SCHEMA: &str = r"
|
|
CREATE TABLE IF NOT EXISTS reminders (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
message TEXT NOT NULL,
|
|
file_path TEXT,
|
|
due_at INTEGER NOT NULL,
|
|
created_at INTEGER NOT NULL,
|
|
sent_at INTEGER
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_reminders_due ON reminders (due_at) WHERE sent_at IS NULL;
|
|
";
|
|
|
|
/// One reminder row. `file_path`, when set, is where the harness persists
|
|
/// the body instead of inlining it at delivery time (see the old
|
|
/// `store_remind`/`prepare_body` split this mirrors, now folded into
|
|
/// `reminder_timer.rs`).
|
|
#[derive(Debug, Clone)]
|
|
pub struct Reminder {
|
|
pub id: i64,
|
|
pub message: String,
|
|
pub file_path: Option<String>,
|
|
pub due_at: DateTime<Utc>,
|
|
pub created_at: DateTime<Utc>,
|
|
}
|
|
|
|
/// The harness-local reminder store. Cheap to share behind an `Arc`; the
|
|
/// inner connection is guarded by a `Mutex` (reminder ops are short sqlite
|
|
/// writes, same as `Todos`).
|
|
pub struct Reminders {
|
|
conn: Mutex<Connection>,
|
|
}
|
|
|
|
impl Reminders {
|
|
/// Open (creating if needed) the reminder store at `path`.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates sqlite open / schema-apply failures.
|
|
pub fn open(path: &Path) -> Result<Self> {
|
|
let conn = Connection::open(path)
|
|
.with_context(|| format!("open reminders db {}", path.display()))?;
|
|
conn.execute_batch(SCHEMA)
|
|
.context("apply reminders schema")?;
|
|
Ok(Self {
|
|
conn: Mutex::new(conn),
|
|
})
|
|
}
|
|
|
|
/// Store a new pending reminder, due at `due_at` (unix seconds).
|
|
/// Returns the new row id.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite insert failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn store(&self, message: &str, file_path: Option<&str>, due_at: i64) -> Result<i64> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let now = Utc::now().timestamp();
|
|
conn.execute(
|
|
"INSERT INTO reminders (message, file_path, due_at, created_at, sent_at) \
|
|
VALUES (?1, ?2, ?3, ?4, NULL)",
|
|
params![message, file_path, due_at, now],
|
|
)?;
|
|
Ok(conn.last_insert_rowid())
|
|
}
|
|
|
|
/// Count of currently-pending (undelivered) reminders.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite query failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn count_pending(&self) -> Result<u64> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let count: i64 = conn.query_row(
|
|
"SELECT COUNT(*) FROM reminders WHERE sent_at IS NULL",
|
|
[],
|
|
|row| row.get(0),
|
|
)?;
|
|
Ok(u64::try_from(count).unwrap_or(0))
|
|
}
|
|
|
|
/// List pending reminders, soonest-due first — for `get_loose_ends`
|
|
/// rendering.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite prepare / query failures.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn list_pending(&self) -> Result<Vec<Reminder>> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let mut stmt = conn.prepare(
|
|
"SELECT id, message, file_path, due_at, created_at \
|
|
FROM reminders WHERE sent_at IS NULL ORDER BY due_at ASC",
|
|
)?;
|
|
let rows = stmt
|
|
.query_map([], row_to_reminder)?
|
|
.collect::<rusqlite::Result<Vec<_>>>()?;
|
|
Ok(rows)
|
|
}
|
|
|
|
/// Pending reminders due at or before `now`, oldest-due first, capped
|
|
/// at `limit` rows — the batch the delivery timer drains per tick.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite prepare / query failures.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn due(&self, now: i64, limit: u64) -> Result<Vec<Reminder>> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let mut stmt = conn.prepare(
|
|
"SELECT id, message, file_path, due_at, created_at \
|
|
FROM reminders WHERE sent_at IS NULL AND due_at <= ?1 \
|
|
ORDER BY due_at ASC LIMIT ?2",
|
|
)?;
|
|
let limit = i64::try_from(limit).unwrap_or(i64::MAX);
|
|
let rows = stmt
|
|
.query_map(params![now, limit], row_to_reminder)?
|
|
.collect::<rusqlite::Result<Vec<_>>>()?;
|
|
Ok(rows)
|
|
}
|
|
|
|
/// Soft-delete: stamp `sent_at` so the row drops out of
|
|
/// `count_pending`/`list_pending`/`due` but survives for the rollup
|
|
/// stats until `prune_delivered_older_than` reaps it.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite update failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn mark_delivered(&self, id: i64) -> Result<()> {
|
|
let conn = self.conn.lock().unwrap();
|
|
conn.execute(
|
|
"UPDATE reminders SET sent_at = ?1 WHERE id = ?2",
|
|
params![Utc::now().timestamp(), id],
|
|
)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Cancel a still-pending reminder by id (hard delete — it never
|
|
/// fired, so there's nothing to keep for the rollup). Returns the
|
|
/// number of rows deleted (0 = unknown id, or already delivered).
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite delete failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn cancel(&self, id: i64) -> Result<usize> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let n = conn.execute(
|
|
"DELETE FROM reminders WHERE id = ?1 AND sent_at IS NULL",
|
|
params![id],
|
|
)?;
|
|
Ok(n)
|
|
}
|
|
|
|
/// Scheduled/delivered/pending counts within the trailing
|
|
/// `since_secs` window (`0` = all time) — the local equivalent of the
|
|
/// old c0re `ReminderRollup` query, same `sent_at IS NULL/NOT NULL`
|
|
/// shape.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite query failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn rollup(&self, since_secs: i64) -> Result<ReminderStats> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let cutoff = if since_secs > 0 {
|
|
Utc::now().timestamp().saturating_sub(since_secs)
|
|
} else {
|
|
i64::MIN
|
|
};
|
|
let scheduled: i64 = conn.query_row(
|
|
"SELECT COUNT(*) FROM reminders WHERE created_at >= ?1",
|
|
params![cutoff],
|
|
|row| row.get(0),
|
|
)?;
|
|
let delivered: i64 = conn.query_row(
|
|
"SELECT COUNT(*) FROM reminders WHERE created_at >= ?1 AND sent_at IS NOT NULL",
|
|
params![cutoff],
|
|
|row| row.get(0),
|
|
)?;
|
|
let pending: i64 = conn.query_row(
|
|
"SELECT COUNT(*) FROM reminders WHERE created_at >= ?1 AND sent_at IS NULL",
|
|
params![cutoff],
|
|
|row| row.get(0),
|
|
)?;
|
|
Ok(ReminderStats {
|
|
scheduled: u64::try_from(scheduled).unwrap_or(0),
|
|
delivered: u64::try_from(delivered).unwrap_or(0),
|
|
pending: u64::try_from(pending).unwrap_or(0),
|
|
})
|
|
}
|
|
|
|
/// Reap delivered rows older than `cutoff` (unix seconds) — called
|
|
/// from `vacuum.rs`'s periodic sweep so the table doesn't grow
|
|
/// unbounded. Returns the number of rows deleted.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Propagates the sqlite delete failure.
|
|
///
|
|
/// # Panics
|
|
///
|
|
/// Panics if the connection mutex is poisoned.
|
|
pub fn prune_delivered_older_than(&self, cutoff: i64) -> Result<usize> {
|
|
let conn = self.conn.lock().unwrap();
|
|
let n = conn.execute(
|
|
"DELETE FROM reminders WHERE sent_at IS NOT NULL AND sent_at < ?1",
|
|
params![cutoff],
|
|
)?;
|
|
Ok(n)
|
|
}
|
|
}
|
|
|
|
fn row_to_reminder(row: &rusqlite::Row) -> rusqlite::Result<Reminder> {
|
|
let due_at_secs: i64 = row.get(3)?;
|
|
let created_at_secs: i64 = row.get(4)?;
|
|
Ok(Reminder {
|
|
id: row.get(0)?,
|
|
message: row.get(1)?,
|
|
file_path: row.get(2)?,
|
|
due_at: wire_time::from_secs(due_at_secs),
|
|
created_at: wire_time::from_secs(created_at_secs),
|
|
})
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
// Return the `TempDir` alongside the store so it outlives the test —
|
|
// dropping it early deletes the dir and SQLite fails with
|
|
// `SQLITE_READONLY_DBMOVED`.
|
|
fn store() -> (tempfile::TempDir, Reminders) {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let db = Reminders::open(&dir.path().join("reminders.sqlite")).unwrap();
|
|
(dir, db)
|
|
}
|
|
|
|
#[test]
|
|
fn store_and_list_pending() {
|
|
let (_dir, s) = store();
|
|
let id = s.store("check on x", None, 1000).unwrap();
|
|
let pending = s.list_pending().unwrap();
|
|
assert_eq!(pending.len(), 1);
|
|
assert_eq!(pending[0].id, id);
|
|
assert_eq!(pending[0].due_at.timestamp(), 1000);
|
|
assert_eq!(s.count_pending().unwrap(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn due_filters_by_time_and_limit() {
|
|
let (_dir, s) = store();
|
|
s.store("a", None, 100).unwrap();
|
|
s.store("b", None, 200).unwrap();
|
|
s.store("c", None, 300).unwrap();
|
|
let due = s.due(200, 10).unwrap();
|
|
assert_eq!(due.len(), 2, "only a and b are due by t=200");
|
|
let capped = s.due(300, 1).unwrap();
|
|
assert_eq!(capped.len(), 1, "limit caps the batch");
|
|
}
|
|
|
|
#[test]
|
|
fn mark_delivered_removes_from_pending_but_counts_in_rollup() {
|
|
let (_dir, s) = store();
|
|
let id = s.store("a", None, 100).unwrap();
|
|
s.mark_delivered(id).unwrap();
|
|
assert_eq!(s.count_pending().unwrap(), 0);
|
|
assert!(s.list_pending().unwrap().is_empty());
|
|
let rollup = s.rollup(0).unwrap();
|
|
assert_eq!(rollup.scheduled, 1);
|
|
assert_eq!(rollup.delivered, 1);
|
|
assert_eq!(rollup.pending, 0);
|
|
}
|
|
|
|
#[test]
|
|
fn cancel_only_removes_pending() {
|
|
let (_dir, s) = store();
|
|
let pending_id = s.store("a", None, 100).unwrap();
|
|
let delivered_id = s.store("b", None, 100).unwrap();
|
|
s.mark_delivered(delivered_id).unwrap();
|
|
assert_eq!(s.cancel(pending_id).unwrap(), 1);
|
|
assert_eq!(
|
|
s.cancel(delivered_id).unwrap(),
|
|
0,
|
|
"already-delivered rows aren't cancellable"
|
|
);
|
|
assert_eq!(s.cancel(999).unwrap(), 0, "unknown id is a no-op");
|
|
}
|
|
|
|
#[test]
|
|
fn prune_reaps_only_old_delivered_rows() {
|
|
let (_dir, s) = store();
|
|
let old = s.store("old", None, 100).unwrap();
|
|
let recent = s.store("recent", None, 100).unwrap();
|
|
let still_pending = s.store("pending", None, 100).unwrap();
|
|
s.mark_delivered(old).unwrap();
|
|
s.mark_delivered(recent).unwrap();
|
|
// Backdate `old`'s sent_at directly so the cutoff test is deterministic.
|
|
{
|
|
let conn = s.conn.lock().unwrap();
|
|
conn.execute(
|
|
"UPDATE reminders SET sent_at = 1 WHERE id = ?1",
|
|
params![old],
|
|
)
|
|
.unwrap();
|
|
}
|
|
let cutoff = Utc::now().timestamp() - 10;
|
|
let n = s.prune_delivered_older_than(cutoff).unwrap();
|
|
assert_eq!(n, 1, "only the backdated row is older than cutoff");
|
|
let remaining_ids: Vec<i64> = {
|
|
let conn = s.conn.lock().unwrap();
|
|
let mut stmt = conn
|
|
.prepare("SELECT id FROM reminders ORDER BY id")
|
|
.unwrap();
|
|
stmt.query_map([], |row| row.get(0))
|
|
.unwrap()
|
|
.collect::<rusqlite::Result<Vec<_>>>()
|
|
.unwrap()
|
|
};
|
|
assert_eq!(remaining_ids, vec![recent, still_pending]);
|
|
}
|
|
}
|