use anyhow::Result;
use rusqlite::{params, Connection, OptionalExtension};
#[derive(Debug, Clone)]
pub(crate) struct MemoryOpRecord {
pub(crate) id: String,
pub(crate) command: String,
pub(crate) request_fingerprint: String,
pub(crate) plan_json: String,
pub(crate) staged_at_epoch_s: u64,
pub(crate) updated_at_epoch_s: u64,
pub(crate) actor_id: Option<String>,
pub(crate) session_id: Option<String>,
pub(crate) reconciled: bool,
pub(crate) outcome: Option<String>,
pub(crate) authored_entry_ids_json: String,
pub(crate) authored_target_path: Option<String>,
pub(crate) authored_source_path: Option<String>,
pub(crate) snapshot_json: Option<String>,
}
pub(crate) fn get_by_id(conn: &Connection, id: &str) -> Result<Option<MemoryOpRecord>> {
conn.query_row(
"SELECT id, command, request_fingerprint, plan_json, staged_at_epoch_s,
updated_at_epoch_s, actor_id, session_id, reconciled, outcome,
authored_entry_ids, authored_target_path, authored_source_path, snapshot_json
FROM memory_op_queue
WHERE id = ?1",
[id],
row_to_record,
)
.optional()
.map_err(Into::into)
}
pub(crate) fn insert_staged(conn: &Connection, record: &MemoryOpRecord) -> Result<()> {
conn.execute(
"INSERT INTO memory_op_queue
(id, command, request_fingerprint, plan_json, staged_at_epoch_s, updated_at_epoch_s,
actor_id, session_id, reconciled, outcome, authored_entry_ids,
authored_target_path, authored_source_path, snapshot_json)
VALUES
(?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
params![
record.id,
record.command,
record.request_fingerprint,
record.plan_json,
record.staged_at_epoch_s,
record.updated_at_epoch_s,
record.actor_id,
record.session_id,
i64::from(record.reconciled),
record.outcome,
record.authored_entry_ids_json,
record.authored_target_path,
record.authored_source_path,
record.snapshot_json,
],
)?;
Ok(())
}
pub(crate) fn mark_reconciled(
conn: &Connection,
id: &str,
updated_at_epoch_s: u64,
outcome: &str,
authored_entry_ids_json: &str,
authored_target_path: &str,
authored_source_path: Option<&str>,
) -> Result<()> {
conn.execute(
"UPDATE memory_op_queue
SET reconciled = 1,
updated_at_epoch_s = ?2,
outcome = ?3,
authored_entry_ids = ?4,
authored_target_path = ?5,
authored_source_path = ?6
WHERE id = ?1",
params![
id,
updated_at_epoch_s,
outcome,
authored_entry_ids_json,
authored_target_path,
authored_source_path,
],
)?;
Ok(())
}
pub(crate) fn update_snapshot(
conn: &Connection,
id: &str,
updated_at_epoch_s: u64,
snapshot_json: &str,
) -> Result<()> {
conn.execute(
"UPDATE memory_op_queue
SET updated_at_epoch_s = ?2,
snapshot_json = ?3
WHERE id = ?1",
params![id, updated_at_epoch_s, snapshot_json],
)?;
Ok(())
}
pub(crate) fn list_by_session(conn: &Connection, session_id: &str) -> Result<Vec<MemoryOpRecord>> {
let mut stmt = conn.prepare(
"SELECT id, command, request_fingerprint, plan_json, staged_at_epoch_s,
updated_at_epoch_s, actor_id, session_id, reconciled, outcome,
authored_entry_ids, authored_target_path, authored_source_path, snapshot_json
FROM memory_op_queue
WHERE session_id = ?1",
)?;
let rows = stmt.query_map([session_id], row_to_record)?;
let mut records = Vec::new();
for row in rows {
records.push(row?);
}
Ok(records)
}
fn row_to_record(row: &rusqlite::Row<'_>) -> rusqlite::Result<MemoryOpRecord> {
Ok(MemoryOpRecord {
id: row.get(0)?,
command: row.get(1)?,
request_fingerprint: row.get(2)?,
plan_json: row.get(3)?,
staged_at_epoch_s: row.get(4)?,
updated_at_epoch_s: row.get(5)?,
actor_id: row.get(6)?,
session_id: row.get(7)?,
reconciled: row.get::<_, i64>(8)? != 0,
outcome: row.get(9)?,
authored_entry_ids_json: row.get(10)?,
authored_target_path: row.get(11)?,
authored_source_path: row.get(12)?,
snapshot_json: row.get(13)?,
})
}