From d86a4d92a81f6f8a7e9f84ca6a4b76f8b2227513 Mon Sep 17 00:00:00 2001 From: TheGeneralist <180094941+thegeneralist01@users.noreply.github.com> Date: Fri, 26 Jun 2026 12:56:10 +0200 Subject: [PATCH] feat(capture): capture_jobs schema and DB helpers --- crates/archivr-core/src/database.rs | 120 ++++++++++++++++++++++++++++ 1 file changed, 120 insertions(+) diff --git a/crates/archivr-core/src/database.rs b/crates/archivr-core/src/database.rs index 7038272..b23d782 100644 --- a/crates/archivr-core/src/database.rs +++ b/crates/archivr-core/src/database.rs @@ -98,6 +98,17 @@ pub struct ApiTokenRecord { pub last_used_at: Option, } +#[derive(Debug, Clone, serde::Serialize)] +pub struct CaptureJobRecord { + pub job_uid: String, + pub archive_id: String, + pub run_uid: Option, + pub status: String, + pub error_text: Option, + pub created_at: String, + pub updated_at: String, +} + pub fn database_path(archive_path: &Path) -> PathBuf { archive_path.join(DATABASE_FILE_NAME) } @@ -240,6 +251,17 @@ pub fn initialize_schema(conn: &Connection) -> Result<()> { PRIMARY KEY (entry_id, tag_id) ); + CREATE TABLE IF NOT EXISTS capture_jobs ( + id INTEGER PRIMARY KEY, + job_uid TEXT NOT NULL UNIQUE, + archive_id TEXT NOT NULL, + run_uid TEXT, + status TEXT NOT NULL CHECK (status IN ('pending','running','completed','failed')) DEFAULT 'pending', + error_text TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_capture_jobs_status ON capture_jobs(status); CREATE INDEX IF NOT EXISTS idx_archive_run_items_run_id ON archive_run_items(run_id); CREATE INDEX IF NOT EXISTS idx_archived_entries_source_identity_id ON archived_entries(source_identity_id); CREATE INDEX IF NOT EXISTS idx_archived_entries_created_by_user_id ON archived_entries(created_by_user_id); @@ -582,6 +604,71 @@ pub fn ensure_default_user(conn: &Connection) -> Result { Ok(conn.last_insert_rowid()) } +/// Creates a pending capture job. Returns the new `job_uid`. +pub fn create_capture_job(conn: &Connection, archive_id: &str) -> Result { + let job_uid = public_id("job"); + let now = now_timestamp(); + conn.execute( + "INSERT INTO capture_jobs (job_uid, archive_id, run_uid, status, error_text, created_at, updated_at) + VALUES (?1, ?2, NULL, 'pending', NULL, ?3, ?3)", + rusqlite::params![job_uid, archive_id, now], + )?; + Ok(job_uid) +} + +/// Updates the status (and optionally run_uid / error_text) of a capture job. +pub fn update_capture_job_status( + conn: &Connection, + job_uid: &str, + status: &str, + run_uid: Option<&str>, + error_text: Option<&str>, +) -> Result<()> { + let now = now_timestamp(); + conn.execute( + "UPDATE capture_jobs SET status = ?1, run_uid = COALESCE(?2, run_uid), + error_text = ?3, updated_at = ?4 WHERE job_uid = ?5", + rusqlite::params![status, run_uid, error_text, now, job_uid], + )?; + Ok(()) +} + +/// Returns a capture job by uid. +pub fn get_capture_job(conn: &Connection, job_uid: &str) -> Result> { + conn.query_row( + "SELECT job_uid, archive_id, run_uid, status, error_text, created_at, updated_at + FROM capture_jobs WHERE job_uid = ?1", + [job_uid], + |row| { + Ok(CaptureJobRecord { + job_uid: row.get(0)?, + archive_id: row.get(1)?, + run_uid: row.get(2)?, + status: row.get(3)?, + error_text: row.get(4)?, + created_at: row.get(5)?, + updated_at: row.get(6)?, + }) + }, + ) + .optional() + .map_err(Into::into) +} + +/// Marks all 'running' capture jobs as 'failed' with a restart message. +/// Called at server startup to clean up jobs interrupted by a previous shutdown. +pub fn fail_stalled_capture_jobs(conn: &Connection) -> Result { + let now = now_timestamp(); + let n = conn.execute( + "UPDATE capture_jobs SET status = 'failed', + error_text = 'interrupted by server restart', + updated_at = ?1 + WHERE status = 'running'", + [now], + )?; + Ok(n) +} + pub fn create_archive_run( conn: &Connection, created_by_user_id: i64, @@ -1568,4 +1655,37 @@ mod tests { let conn = make_auth_conn(); assert!(get_user_for_token(&conn, "unknown").unwrap().is_none()); } + + #[test] + fn capture_job_create_and_get() { + let conn = conn(); + let job_uid = create_capture_job(&conn, "personal").unwrap(); + let job = get_capture_job(&conn, &job_uid).unwrap().unwrap(); + assert_eq!(job.status, "pending"); + assert_eq!(job.archive_id, "personal"); + assert!(job.run_uid.is_none()); + } + + #[test] + fn capture_job_status_transitions() { + let conn = conn(); + let job_uid = create_capture_job(&conn, "test").unwrap(); + update_capture_job_status(&conn, &job_uid, "running", None, None).unwrap(); + update_capture_job_status(&conn, &job_uid, "completed", Some("run_abc"), None).unwrap(); + let job = get_capture_job(&conn, &job_uid).unwrap().unwrap(); + assert_eq!(job.status, "completed"); + assert_eq!(job.run_uid.as_deref(), Some("run_abc")); + } + + #[test] + fn fail_stalled_jobs_on_restart() { + let conn = conn(); + let uid = create_capture_job(&conn, "test").unwrap(); + update_capture_job_status(&conn, &uid, "running", None, None).unwrap(); + let n = fail_stalled_capture_jobs(&conn).unwrap(); + assert_eq!(n, 1); + let job = get_capture_job(&conn, &uid).unwrap().unwrap(); + assert_eq!(job.status, "failed"); + assert!(job.error_text.as_deref().unwrap().contains("interrupted")); + } }