//! Versioned durable-daemon migrations. Kept independent so the store can call it.
use rusqlite::{params, Connection, OptionalExtension};
use sha2::{Digest, Sha256};
pub const CURRENT_VERSION: i64 = 3;
pub const LEGACY_OWNER_UNKNOWN: &str = "legacy_owner_unknown";
#[allow(dead_code)]
pub trait MigrationStore {
fn connection(&self) -> &Connection;
}
pub fn apply(conn: &mut Connection, binary_digest: &str) -> rusqlite::Result<()> {
let tx = conn.transaction()?;
tx.execute_batch("CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY, applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, migration_digest TEXT NOT NULL);")?;
let exists: Option<i64> = tx
.query_row(
"SELECT version FROM schema_migrations WHERE version = ?1",
[CURRENT_VERSION],
|r| r.get(0),
)
.optional()?;
if exists.is_none() {
tx.execute_batch("CREATE TABLE IF NOT EXISTS daemon_instances (instance_id TEXT PRIMARY KEY, generation INTEGER NOT NULL UNIQUE, pid INTEGER NOT NULL, boot_id TEXT, executable_digest TEXT, started_at TEXT NOT NULL, heartbeat_at TEXT NOT NULL, clean_shutdown_at TEXT); CREATE TABLE IF NOT EXISTS run_publication_state (run_id TEXT PRIMARY KEY, state TEXT NOT NULL, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, reason TEXT); CREATE TABLE IF NOT EXISTS operator_receipts (receipt_id TEXT PRIMARY KEY, request_digest TEXT NOT NULL, action TEXT NOT NULL, resource_kind TEXT NOT NULL, resource_id TEXT NOT NULL, state_digest TEXT NOT NULL, operator_uid INTEGER NOT NULL, daemon_instance_id TEXT NOT NULL, nonce TEXT NOT NULL UNIQUE, issued_at TEXT NOT NULL, expires_at TEXT NOT NULL, consumed_at TEXT); CREATE INDEX IF NOT EXISTS idx_operator_receipts_nonce ON operator_receipts(nonce);")?;
let has_owner: Option<String> = tx
.query_row(
"SELECT 1 FROM pragma_table_info('executions') WHERE name='owner_instance_id'",
[],
|r| r.get(0),
)
.optional()?;
if has_owner.is_none() {
// Only add owner_instance_id if executions table exists.
// The store's own migration creates it; if it doesn't exist yet,
// the store migration will need to include the column.
let has_executions: Option<i64> = tx
.query_row(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name='executions'",
[],
|r| r.get(0),
)
.optional()?;
if has_executions.is_some() {
tx.execute_batch("ALTER TABLE executions ADD COLUMN owner_instance_id TEXT;")?;
tx.execute("UPDATE executions SET owner_instance_id = ?1 WHERE owner_instance_id IS NULL AND status IN ('accepted','running')", [LEGACY_OWNER_UNKNOWN])?;
}
}
let digest = migration_digest(binary_digest);
tx.execute(
"INSERT INTO schema_migrations(version,migration_digest) VALUES (?1,?2)",
params![CURRENT_VERSION, digest],
)?;
}
tx.commit()
}
pub fn migration_digest(binary_digest: &str) -> String {
let mut h = Sha256::new();
h.update(format!(
"agent-graph-mcp:migration:{CURRENT_VERSION}:{binary_digest}"
));
format!("{:x}", h.finalize())
}
#[allow(dead_code)]
pub fn owner_for_new_run<'a>(owner: &'a str) -> rusqlite::Result<&'a str> {
if owner.is_empty() {
Err(rusqlite::Error::InvalidParameterName(
"owner_instance_id".into(),
))
} else {
Ok(owner)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn legacy_active_rows_are_quarantined() {
let mut c = Connection::open_in_memory().unwrap();
c.execute_batch("CREATE TABLE executions(run_id TEXT PRIMARY KEY,status TEXT NOT NULL); INSERT INTO executions VALUES ('a','running'),('b','completed');").unwrap();
apply(&mut c, "bin").unwrap();
let owner: String = c
.query_row(
"SELECT owner_instance_id FROM executions WHERE run_id='a'",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(owner, LEGACY_OWNER_UNKNOWN);
apply(&mut c, "bin").unwrap();
assert_eq!(
c.query_row::<i64, _, _>("SELECT count(*) FROM schema_migrations", [], |r| r.get(0))
.unwrap(),
1
);
}
}