use super::*;
pub(crate) fn processes_changed_since_conn(
conn: &Connection,
cursor: ProcessChangeCursor,
limit: usize,
) -> Result<(Vec<ProcessRecord>, ProcessChangeCursor), lash_core::PluginError> {
let mut stmt = conn
.prepare(
"SELECT change_seq, record_json FROM processes
WHERE change_seq > ?1
ORDER BY change_seq ASC, process_id ASC
LIMIT ?2",
)
.map_err(process_sqlite_error)?;
let rows = stmt
.query_map(
params![cursor.store_sequence() as i64, limit as i64],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
)
.map_err(process_sqlite_error)?;
let mut records = Vec::new();
let mut next_cursor = cursor;
for row in rows {
let (change_seq, record_json) = row.map_err(process_sqlite_error)?;
let record: ProcessRecord =
serde_json::from_str(&record_json).map_err(process_decode_error)?;
next_cursor = ProcessChangeCursor::from_store_sequence(change_seq as u64);
records.push(record);
}
Ok((records, next_cursor))
}
pub(crate) fn prune_terminal_processes_conn(
conn: &Connection,
cutoff: i64,
filter: Option<ProcessListFilter>,
max_change_seq: Option<u64>,
) -> Result<ProcessPruneReport, lash_core::PluginError> {
let max_change_seq = max_change_seq.map(|seq| seq as i64);
let trigger_deliveries_exists = conn
.query_row(
"SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'trigger_deliveries'",
[],
|_| Ok(()),
)
.optional()
.map_err(process_sqlite_error)?
.is_some();
let mut stmt = conn
.prepare(
"SELECT process_id, record_json FROM processes
WHERE status != 'running'
AND updated_at_ms < ?1
AND (?2 IS NULL OR change_seq <= ?2)
ORDER BY process_id ASC",
)
.map_err(process_sqlite_error)?;
let rows = stmt
.query_map(params![cutoff, max_change_seq], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})
.map_err(process_sqlite_error)?;
let mut prunable = Vec::new();
for row in rows {
let (process_id, record_json) = row.map_err(process_sqlite_error)?;
let record: ProcessRecord =
serde_json::from_str(&record_json).map_err(process_decode_error)?;
if filter
.as_ref()
.is_none_or(|filter| filter.matches_record(&record))
{
prunable.push(process_id);
}
}
let mut pruned_events = 0;
let mut pruned_processes = 0;
for process_id in prunable {
let process_id = process_id.as_str();
pruned_events += conn
.execute(
"DELETE FROM process_events WHERE process_id = ?1",
params![process_id],
)
.map_err(process_sqlite_error)?;
conn.execute(
"DELETE FROM process_wake_acks WHERE process_id = ?1",
params![process_id],
)
.map_err(process_sqlite_error)?;
conn.execute(
"DELETE FROM process_handle_grants WHERE process_id = ?1",
params![process_id],
)
.map_err(process_sqlite_error)?;
conn.execute(
"DELETE FROM process_leases WHERE process_id = ?1",
params![process_id],
)
.map_err(process_sqlite_error)?;
if trigger_deliveries_exists {
conn.execute(
"DELETE FROM trigger_deliveries WHERE process_id = ?1",
params![process_id],
)
.map_err(process_sqlite_error)?;
}
pruned_processes += conn
.execute(
"DELETE FROM processes WHERE process_id = ?1",
params![process_id],
)
.map_err(process_sqlite_error)?;
}
Ok(ProcessPruneReport {
pruned_processes,
pruned_events,
})
}