use std::path::Path;
use libsql::TransactionBehavior;
use crate::error::{Result, WriteOp};
use crate::schema::ddl::ARCHIVE_SESSION_MARKER;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ArchiveReport {
pub links_archived: usize,
pub concepts_archived: usize,
pub log_entries_archived: usize,
pub horizon: Option<i64>,
}
const COLD_SCHEMA: &[&str] = &[
r#"CREATE TABLE IF NOT EXISTS cold.links (
source_id TEXT NOT NULL,
target_id TEXT NOT NULL,
edge_type TEXT NOT NULL,
valid_from TEXT NOT NULL,
recorded_at TEXT NOT NULL,
valid_to TEXT NOT NULL,
weight REAL NOT NULL CHECK (weight >= 0.0 AND weight < 9e999 AND typeof(weight) = 'real'),
properties TEXT NOT NULL,
PRIMARY KEY (source_id, target_id, edge_type, valid_from, recorded_at)
)"#,
r#"CREATE TABLE IF NOT EXISTS cold.concepts (
rowid_pk INTEGER,
id TEXT NOT NULL PRIMARY KEY,
title TEXT NOT NULL,
content TEXT NOT NULL DEFAULT '',
embedding_model TEXT,
valid_from TEXT NOT NULL,
valid_to TEXT NOT NULL,
recorded_at TEXT NOT NULL,
retired INTEGER NOT NULL DEFAULT 0
)"#,
r#"CREATE TABLE IF NOT EXISTS cold.transaction_log (
seq_id INTEGER PRIMARY KEY,
table_name TEXT NOT NULL,
entity_id TEXT NOT NULL,
operation TEXT NOT NULL,
payload TEXT NOT NULL,
recorded_at TEXT NOT NULL
)"#,
"CREATE INDEX IF NOT EXISTS cold.idx_cold_txlog_entity ON transaction_log (entity_id)",
"CREATE INDEX IF NOT EXISTS cold.idx_cold_txlog_time ON transaction_log (recorded_at)",
r#"CREATE TABLE IF NOT EXISTS cold.archive_horizon (
archived_at TEXT NOT NULL,
cutoff TEXT NOT NULL,
horizon INTEGER
)"#,
];
const LINKS_ARCHIVABLE: &str = r#"
recorded_at < :cutoff AND (
EXISTS (
SELECT 1 FROM links newer
WHERE newer.source_id = links.source_id
AND newer.target_id = links.target_id
AND newer.edge_type = links.edge_type
AND newer.valid_from = links.valid_from
AND newer.recorded_at > links.recorded_at
)
OR (valid_to <> '9999-12-31T23:59:59.999999Z' AND valid_to <= :cutoff)
)
"#;
const LOG_ARCHIVABLE: &str = r#"
recorded_at < :cutoff AND EXISTS (
SELECT 1 FROM transaction_log newer
WHERE newer.entity_id = transaction_log.entity_id
AND newer.seq_id > transaction_log.seq_id
)
"#;
const CONCEPTS_ARCHIVABLE: &str = r#"
retired = 1
AND recorded_at < :cutoff
AND valid_to < :cutoff
AND NOT EXISTS (
SELECT 1 FROM links
WHERE links.source_id = concepts.id
OR links.target_id = concepts.id
)
"#;
pub async fn archivable_concepts(conn: &libsql::Connection, cutoff: &str) -> Result<Vec<String>> {
let mut rows = conn
.query(
&format!("SELECT id FROM concepts WHERE {CONCEPTS_ARCHIVABLE} ORDER BY id"),
libsql::named_params! {":cutoff": cutoff},
)
.await?;
let mut ids = Vec::new();
while let Some(row) = rows.next().await? {
ids.push(row.get::<String>(0)?);
}
Ok(ids)
}
pub async fn archive(
conn: &libsql::Connection,
cutoff: &str,
archived_at: &str,
archive_path: &Path,
) -> Result<ArchiveReport> {
crate::temporal::replay::detach_stale_cold(conn).await;
conn.execute(
"ATTACH DATABASE ?1 AS cold",
libsql::params![archive_path.to_string_lossy().as_ref()],
)
.await?;
let result = archive_session(conn, cutoff, archived_at).await;
if let Err(e) = conn.execute("DETACH DATABASE cold", ()).await {
tracing::warn!("archive: failed to DETACH cold database: {e}");
}
result
}
async fn archive_session(
conn: &libsql::Connection,
cutoff: &str,
archived_at: &str,
) -> Result<ArchiveReport> {
for ddl in COLD_SCHEMA {
conn.execute(ddl, ()).await?;
}
let tx = conn
.transaction_with_behavior(TransactionBehavior::Immediate)
.await?;
tx.execute(&format!("CREATE TABLE {ARCHIVE_SESSION_MARKER} (x)"), ())
.await?;
let links_archived = tx
.execute(
&format!(
"INSERT OR IGNORE INTO cold.links
(source_id, target_id, edge_type, valid_from, recorded_at,
valid_to, weight, properties)
SELECT source_id, target_id, edge_type, valid_from, recorded_at,
valid_to, weight, properties
FROM links WHERE {LINKS_ARCHIVABLE}"
),
libsql::named_params! {":cutoff": cutoff},
)
.await? as usize;
let links_deleted = delete_guarded(
&tx,
conn,
&format!("DELETE FROM links WHERE {LINKS_ARCHIVABLE}"),
cutoff,
"links",
)
.await?;
if links_deleted > 0 {
crate::integrity::rebuild::rebuild_within(&tx, crate::integrity::rebuild::Verify::No)
.await?;
}
let concepts_archived = archive_concepts(&tx, conn, cutoff).await?;
let log_entries_archived = tx
.execute(
&format!(
"INSERT OR IGNORE INTO cold.transaction_log
(seq_id, table_name, entity_id, operation, payload, recorded_at)
SELECT seq_id, table_name, entity_id, operation, payload, recorded_at
FROM transaction_log WHERE {LOG_ARCHIVABLE}"
),
libsql::named_params! {":cutoff": cutoff},
)
.await? as usize;
delete_guarded(
&tx,
conn,
&format!("DELETE FROM transaction_log WHERE {LOG_ARCHIVABLE}"),
cutoff,
"transaction_log",
)
.await?;
let horizon: Option<i64> = tx
.query("SELECT MIN(seq_id) FROM transaction_log", ())
.await?
.next()
.await?
.and_then(|row| row.get(0).ok());
tx.execute(
"INSERT INTO cold.archive_horizon (archived_at, cutoff, horizon) VALUES (?1, ?2, ?3)",
libsql::params![archived_at, cutoff, horizon],
)
.await?;
tx.execute(&format!("DROP TABLE {ARCHIVE_SESSION_MARKER}"), ())
.await?;
tx.commit().await?;
Ok(ArchiveReport {
links_archived,
concepts_archived,
log_entries_archived,
horizon,
})
}
async fn archive_concepts(
tx: &libsql::Transaction,
conn: &libsql::Connection,
cutoff: &str,
) -> Result<usize> {
let moved = tx
.execute(
&format!(
"INSERT OR IGNORE INTO cold.concepts
(rowid_pk, id, title, content, embedding_model,
valid_from, valid_to, recorded_at, retired)
SELECT rowid_pk, id, title, content, embedding_model,
valid_from, valid_to, recorded_at, retired
FROM concepts WHERE {CONCEPTS_ARCHIVABLE}"
),
libsql::named_params! {":cutoff": cutoff},
)
.await? as usize;
if moved == 0 {
return Ok(0);
}
let mut derived: Vec<String> = vec!["analytics_annotations".to_string()];
let mut rows = tx
.query(
"SELECT name FROM sqlite_master WHERE type = 'table' \
AND name LIKE 'embeddings\\_%' ESCAPE '\\'",
(),
)
.await?;
while let Some(row) = rows.next().await? {
derived.push(row.get::<String>(0)?);
}
drop(rows);
for table in &derived {
tx.execute(
&format!(
"DELETE FROM {table} WHERE concept_id IN \
(SELECT id FROM cold.concepts)"
),
(),
)
.await?;
}
let deleted = delete_guarded(
tx,
conn,
&format!("DELETE FROM concepts WHERE {CONCEPTS_ARCHIVABLE}"),
cutoff,
"concepts",
)
.await? as usize;
debug_assert_eq!(
moved, deleted,
"the predicate selected a different set for the copy than for the delete"
);
Ok(deleted)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RehydrateReport {
pub concepts_rehydrated: usize,
pub rowids_reassigned: usize,
}
pub async fn rehydrate(
conn: &libsql::Connection,
ids: &[&str],
archive_path: &Path,
) -> Result<RehydrateReport> {
if ids.is_empty() {
return Ok(RehydrateReport {
concepts_rehydrated: 0,
rowids_reassigned: 0,
});
}
crate::temporal::replay::detach_stale_cold(conn).await;
conn.execute(
"ATTACH DATABASE ?1 AS cold",
libsql::params![archive_path.to_string_lossy().as_ref()],
)
.await?;
let result = rehydrate_session(conn, ids).await;
if let Err(e) = conn.execute("DETACH DATABASE cold", ()).await {
tracing::warn!("rehydrate: failed to DETACH cold database: {e}");
}
result
}
async fn rehydrate_session(conn: &libsql::Connection, ids: &[&str]) -> Result<RehydrateReport> {
let tx = conn
.transaction_with_behavior(TransactionBehavior::Immediate)
.await?;
tx.execute(&format!("CREATE TABLE {ARCHIVE_SESSION_MARKER} (x)"), ())
.await?;
let mut rehydrated = 0usize;
let mut reassigned = 0usize;
for id in ids {
let Some(row) = tx
.query(
"SELECT rowid_pk, id, title, content, embedding_model, \
valid_from, valid_to, recorded_at, retired \
FROM cold.concepts WHERE id = ?1",
libsql::params![*id],
)
.await?
.next()
.await?
else {
continue;
};
let old_rowid: i64 = row.get(0)?;
let title: String = row.get(2)?;
let content: String = row.get(3)?;
let model: Option<String> = row.get(4)?;
let valid_from: String = row.get(5)?;
let valid_to: String = row.get(6)?;
let recorded_at: String = row.get(7)?;
let retired: i64 = row.get(8)?;
let taken: i64 = tx
.query(
"SELECT COUNT(*) FROM concepts WHERE rowid_pk = ?1",
libsql::params![old_rowid],
)
.await?
.next()
.await?
.expect("COUNT(*) always returns a row")
.get(0)?;
if taken == 0 {
tx.execute(
"INSERT INTO concepts (rowid_pk, id, title, content, embedding_model, \
valid_from, valid_to, recorded_at, retired) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
libsql::params![
old_rowid,
*id,
title.clone(),
content.clone(),
model,
valid_from,
valid_to,
recorded_at,
retired
],
)
.await?;
} else {
tx.execute(
"INSERT INTO concepts (id, title, content, embedding_model, \
valid_from, valid_to, recorded_at, retired) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
libsql::params![
*id,
title.clone(),
content.clone(),
model,
valid_from,
valid_to,
recorded_at,
retired
],
)
.await?;
tx.execute(
"INSERT INTO concepts_fts (concepts_fts, rowid, title, content) \
VALUES ('delete', ?1, ?2, ?3)",
libsql::params![old_rowid, title, content],
)
.await?;
reassigned += 1;
}
tx.execute(
"DELETE FROM cold.concepts WHERE id = ?1",
libsql::params![*id],
)
.await?;
rehydrated += 1;
}
tx.execute(&format!("DROP TABLE {ARCHIVE_SESSION_MARKER}"), ())
.await?;
tx.commit().await?;
Ok(RehydrateReport {
concepts_rehydrated: rehydrated,
rowids_reassigned: reassigned,
})
}
async fn delete_guarded(
tx: &libsql::Transaction,
conn: &libsql::Connection,
sql: &str,
cutoff: &str,
table: &str,
) -> Result<u64> {
match tx
.execute(sql, libsql::named_params! {":cutoff": cutoff})
.await
{
Ok(n) => Ok(n),
Err(e) => Err(crate::error::classify(conn, e, WriteOp::Delete { table }).await),
}
}