Merge branch 'task/async-capture-backend'

crates/archivr-core/src/archive.rs

@@ -60,6 +60,17 @@ pub struct RunSummary {
     pub error_summary: Option<String>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
+pub struct CaptureJobSummary {
+    pub job_uid: String,
+    pub archive_id: String,
+    pub run_uid: Option<String>,
+    pub status: String,
+    pub error_text: Option<String>,
+    pub created_at: String,
+    pub updated_at: String,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
 pub struct Tag {
     pub tag_uid: String,
@@ -299,6 +310,21 @@ pub fn list_runs(conn: &rusqlite::Connection) -> Result<Vec<RunSummary>> {
     Ok(runs)
 }
 
+pub fn get_capture_job(
+    conn: &rusqlite::Connection,
+    job_uid: &str,
+) -> Result<Option<CaptureJobSummary>> {
+    Ok(database::get_capture_job(conn, job_uid)?.map(|r| CaptureJobSummary {
+        job_uid: r.job_uid,
+        archive_id: r.archive_id,
+        run_uid: r.run_uid,
+        status: r.status,
+        error_text: r.error_text,
+        created_at: r.created_at,
+        updated_at: r.updated_at,
+    }))
+}
+
 /// Resolves an artifact to its absolute on-disk path under `store_path`.
 ///
 /// `artifact.relpath` is a store-relative path (e.g. `raw/a/b/abc.pdf`).

crates/archivr-core/src/database.rs

@@ -98,6 +98,17 @@ pub struct ApiTokenRecord {
     pub last_used_at: Option<String>,
 }
 
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct CaptureJobRecord {
+    pub job_uid: String,
+    pub archive_id: String,
+    pub run_uid: Option<String>,
+    pub status: String,
+    pub error_text: Option<String>,
+    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<i64> {
     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<String> {
+    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<Option<CaptureJobRecord>> {
+    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<usize> {
+    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"));
+    }
 }

crates/archivr-server/src/main.rs

@@ -26,6 +26,17 @@ async fn main() -> Result<()> {
 
     let app = routes::app(registry.clone(), auth_db_path.clone());
 
+    // On startup, mark any jobs that were 'running' when the server last stopped as 'failed'.
+    for archive in &registry.archives {
+        if let Ok(conn) = archivr_core::database::open_or_initialize(&archive.archive_path) {
+            match archivr_core::database::fail_stalled_capture_jobs(&conn) {
+                Ok(n) if n > 0 => eprintln!("info: marked {n} stalled capture job(s) as failed in '{}'", archive.id),
+                Err(e) => eprintln!("warn: stalled job cleanup failed for '{}': {e:#}", archive.id),
+                _ => {}
+            }
+        }
+    }
+
     // Spawn session cleanup: runs at startup and every 24h.
     let cleanup_auth_path = auth_db_path.clone();
     tokio::spawn(async move {

crates/archivr-server/src/routes.rs

@@ -103,6 +103,10 @@ pub fn app(registry: ServerRegistry, auth_db_path: std::path::PathBuf) -> Router
         )
         .route("/api/archives/:archive_id/runs", get(list_runs))
         .route("/api/archives/:archive_id/captures", post(capture_handler))
+        .route(
+            "/api/archives/:archive_id/capture_jobs/:job_uid",
+            get(get_capture_job_handler),
+        )
         .route("/api/archives/:archive_id/tags", get(list_tags).post(create_tag_handler))
         .route(
             "/api/archives/:archive_id/entries/:entry_uid/tags",
@@ -366,7 +370,7 @@ async fn capture_handler(
     auth_user: AuthUser,
     Path(archive_id): Path<String>,
     Json(body): Json<CaptureBody>,
-) -> Result<Json<capture::CaptureResult>, ApiError> {
+) -> Result<(StatusCode, Json<serde_json::Value>), ApiError> {
     auth_user.require_role(ROLE_USER)?;
     if body.locator.trim().is_empty() {
         return Err(ApiError::bad_request("locator must not be empty"));
@@ -374,9 +378,67 @@ async fn capture_handler(
     let mounted = mounted_archive(&state, &archive_id)?;
     let archive_paths = archive::read_archive_paths(&mounted.archive_path)
         .map_err(ApiError::from)?;
-    let result = capture::perform_capture(&archive_paths, &body.locator, Some(&archive_id))
-        .map_err(ApiError::from)?;
-    Ok(Json(result))
+
+    // Create job record in the archive DB.
+    let conn = database::open_or_initialize(&mounted.archive_path)?;
+    let job_uid = database::create_capture_job(&conn, &archive_id)?;
+    drop(conn);
+
+    // Spawn background capture.
+    let locator = body.locator.trim().to_string();
+    let archive_path = mounted.archive_path.clone();
+    let job_uid_bg = job_uid.clone();
+    let archive_id_bg = archive_id.clone();
+    tokio::task::spawn_blocking(move || {
+        let conn = match database::open_or_initialize(&archive_path) {
+            Ok(c) => c,
+            Err(e) => {
+                eprintln!("warn: capture job {job_uid_bg}: db open failed: {e:#}");
+                return;
+            }
+        };
+        database::update_capture_job_status(&conn, &job_uid_bg, "running", None, None).ok();
+        match capture::perform_capture(&archive_paths, &locator, Some(&archive_id_bg)) {
+            Ok(result) => {
+                database::update_capture_job_status(
+                    &conn,
+                    &job_uid_bg,
+                    "completed",
+                    Some(&result.run_uid),
+                    None,
+                )
+                .ok();
+            }
+            Err(e) => {
+                database::update_capture_job_status(
+                    &conn,
+                    &job_uid_bg,
+                    "failed",
+                    None,
+                    Some(&format!("{e:#}")),
+                )
+                .ok();
+            }
+        }
+    });
+
+    Ok((
+        StatusCode::ACCEPTED,
+        Json(serde_json::json!({ "job_uid": job_uid, "status": "pending" })),
+    ))
+}
+
+async fn get_capture_job_handler(
+    State(state): State<AppState>,
+    auth_user: AuthUser,
+    Path((archive_id, job_uid)): Path<(String, String)>,
+) -> Result<Json<archive::CaptureJobSummary>, ApiError> {
+    auth_user.require_role(ROLE_USER)?;
+    let mounted = mounted_archive(&state, &archive_id)?;
+    let conn = database::open_or_initialize(&mounted.archive_path)?;
+    archive::get_capture_job(&conn, &job_uid)?
+        .map(Json)
+        .ok_or_else(|| ApiError::not_found("capture job not found"))
 }
 
 async fn auth_setup_status(
@@ -1421,4 +1483,28 @@ mod tests {
         assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
     }
 
+
+    #[tokio::test]
+    async fn capture_post_returns_accepted_with_job_uid() {
+        let dir = tempfile::tempdir().unwrap();
+        let (registry, _, auth_path) = make_test_registry(&dir);
+        let session_cookie = make_test_session(&auth_path);
+        let response = app(registry, auth_path)
+            .oneshot(
+                Request::builder()
+                    .method("POST")
+                    .uri("/api/archives/test/captures")
+                    .header("content-type", "application/json")
+                    .header("cookie", &session_cookie)
+                    .body(Body::from(r#"{"locator":"local:/nonexistent"}"#))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::ACCEPTED);
+        let body = axum::body::to_bytes(response.into_body(), usize::MAX).await.unwrap();
+        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
+        assert!(json["job_uid"].as_str().is_some(), "response must have job_uid");
+        assert_eq!(json["status"], "pending");
+    }
 }