Revalidate import job commands and error downloads

crates/horae/src/jobs.rs

@@ -15,6 +15,7 @@ use crate::{config::JobPolicy, state::AppState};
 const LEASE: Duration = Duration::from_secs(300);
 const POLL: Duration = Duration::from_secs(2);
 
+pub(crate) mod access;
 mod lease;
 pub(crate) mod outbox;
 pub(crate) mod report;
@@ -214,8 +215,8 @@ impl JobPayload {
 #[error("Idempotency key conflicts with another import request")]
 pub struct RequestConflict;
 
-pub async fn request_exists(
-    pool: &sqlx::PgPool,
+pub async fn request_exists<'e, E: sqlx::PgExecutor<'e>>(
+    exec: E,
     org_id: Uuid,
     kind: &str,
     key: &str,
@@ -227,7 +228,7 @@ pub async fn request_exists(
         kind,
         key,
     )
-    .fetch_one(pool)
+    .fetch_one(exec)
     .await?)
 }
 
@@ -250,6 +251,7 @@ pub async fn enqueue(
     Ok(id)
 }
 
+#[cfg(test)]
 pub async fn enqueue_api(
     pool: &sqlx::PgPool,
     org_id: Uuid,
@@ -258,20 +260,8 @@ pub async fn enqueue_api(
     policy: JobPolicy,
     generation: i64,
 ) -> anyhow::Result<Uuid> {
-    use crate::importers::harvest::account_switch::{self, ChangeError};
     let mut tx = pool.begin().await?;
-    let current = account_switch::gate(&mut tx, org_id).await?;
-    anyhow::ensure!(
-        generation == current.account_generation,
-        ChangeError::OldImport
-    );
-    let existing = sqlx::query_scalar!("SELECT id FROM horae_jobs WHERE org_id = $1 AND kind = 'harvest_api_import' AND idempotency_key = $2", org_id, idempotency_key)
-        .fetch_optional(&mut *tx).await?;
-    if existing.is_none() {
-        let connection = account_switch::status(&mut *tx, org_id, true).await?;
-        anyhow::ensure!(connection.connected, ChangeError::NotConnected);
-    }
-    let id = enqueue_in(
+    let id = enqueue_api_in(
         &mut tx,
         org_id,
         payload,
@@ -284,6 +274,31 @@ pub async fn enqueue_api(
     Ok(id)
 }
 
+/// Enqueue within the caller's authorized transaction; never acquire a worker reservation.
+pub async fn enqueue_api_in(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    org_id: Uuid,
+    payload: &JobPayload,
+    idempotency_key: &str,
+    policy: JobPolicy,
+    generation: i64,
+) -> anyhow::Result<Uuid> {
+    use crate::importers::harvest::account_switch::{self, ChangeError};
+    let current = account_switch::gate(tx, org_id).await?;
+    anyhow::ensure!(
+        generation == current.account_generation,
+        ChangeError::OldImport
+    );
+    let existing = sqlx::query_scalar!("SELECT id FROM horae_jobs WHERE org_id = $1 AND kind = 'harvest_api_import' AND idempotency_key = $2", org_id, idempotency_key)
+        .fetch_optional(&mut **tx).await?;
+    if existing.is_none() {
+        let connection = account_switch::status(&mut **tx, org_id, true).await?;
+        anyhow::ensure!(connection.connected, ChangeError::NotConnected);
+    }
+    let id = enqueue_in(tx, org_id, payload, idempotency_key, policy, generation).await?;
+    Ok(id)
+}
+
 async fn enqueue_in(
     tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     org_id: Uuid,
@@ -318,21 +333,36 @@ async fn enqueue_in(
     Ok(row.id)
 }
 
+#[cfg(test)]
 pub async fn enqueue_csv(
     pool: &sqlx::PgPool,
     org_id: Uuid,
     mode: ImportMode,
     body: Vec<u8>,
     idempotency_key: &str,
     policy: JobPolicy,
+) -> anyhow::Result<Uuid> {
+    let mut tx = pool.begin().await?;
+    let id = enqueue_csv_in(&mut tx, org_id, mode, body, idempotency_key, policy).await?;
+    tx.commit().await?;
+    Ok(id)
+}
+
+/// Persist the accepted upload and job atomically in the authorization transaction.
+pub async fn enqueue_csv_in(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    org_id: Uuid,
+    mode: ImportMode,
+    body: Vec<u8>,
+    idempotency_key: &str,
+    policy: JobPolicy,
 ) -> anyhow::Result<Uuid> {
     let policy = policy.validate()?;
     let id = Uuid::now_v7();
     let payload = JobPayload::HarvestCsv { mode };
     let report = payload.initial_report()?;
     let payload = encode_payload(&payload)?;
-    let mut tx = pool.begin().await?;
-    crate::importers::harvest::account_switch::gate(&mut tx, org_id).await?;
+    crate::importers::harvest::account_switch::gate(tx, org_id).await?;
     let row = sqlx::query!(
         r#"INSERT INTO horae_jobs (id, org_id, kind, payload, idempotency_key, max_attempts, report)
            VALUES ($1, $2, 'harvest_csv_import',
@@ -350,7 +380,7 @@ pub async fn enqueue_csv(
         report,
         &body,
     )
-    .fetch_optional(&mut *tx)
+    .fetch_optional(&mut **tx)
     .await?
     .ok_or(RequestConflict)?;
     // An identical resubmission must not restore an upload already removed by
@@ -364,10 +394,9 @@ pub async fn enqueue_csv(
             org_id,
             body,
         )
-        .execute(&mut *tx)
+        .execute(&mut **tx)
         .await?;
     }
-    tx.commit().await?;
     Ok(row.id)
 }
 
@@ -398,8 +427,8 @@ fn retry_availability(
     }
 }
 
-pub async fn status(
-    pool: &sqlx::PgPool,
+pub async fn status<'e, E: sqlx::PgExecutor<'e>>(
+    exec: E,
     org_id: Uuid,
     id: Uuid,
 ) -> anyhow::Result<Option<crate::models::JobStatus>> {
@@ -418,7 +447,7 @@ pub async fn status(
         id,
         org_id,
     )
-    .fetch_optional(pool)
+    .fetch_optional(exec)
     .await?;
     Ok(row.map(|r| crate::models::JobStatus {
         retry_availability: retry_availability(
@@ -442,8 +471,8 @@ pub async fn status(
     }))
 }
 
-pub async fn list(
-    pool: &sqlx::PgPool,
+pub async fn list<'e, E: sqlx::PgExecutor<'e>>(
+    exec: E,
     org_id: Uuid,
     limit: i64,
     before: Option<Uuid>,
@@ -469,7 +498,7 @@ pub async fn list(
         limit.clamp(1, 100),
         before,
     )
-    .fetch_all(pool)
+    .fetch_all(exec)
     .await?;
     Ok(rows
         .into_iter()
@@ -518,11 +547,23 @@ pub async fn cancel<'e, E: sqlx::PgExecutor<'e>>(
     Ok(result.rows_affected() == 1)
 }
 
+#[cfg(test)]
 pub async fn retry(pool: &sqlx::PgPool, org_id: Uuid, id: Uuid) -> anyhow::Result<bool> {
     let mut tx = pool.begin().await?;
-    let current = crate::importers::harvest::account_switch::gate(&mut tx, org_id).await?;
+    let retried = retry_in(&mut tx, org_id, id).await?;
+    tx.commit().await?;
+    Ok(retried)
+}
+
+/// Retry within the caller's authorization transaction, retaining generation fencing.
+pub async fn retry_in(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    org_id: Uuid,
+    id: Uuid,
+) -> anyhow::Result<bool> {
+    let current = crate::importers::harvest::account_switch::gate(tx, org_id).await?;
     let generation = sqlx::query_scalar!("SELECT account_generation FROM horae_jobs WHERE id = $1 AND org_id = $2 AND kind = 'harvest_api_import'", id, org_id)
-        .fetch_optional(&mut *tx).await?;
+        .fetch_optional(&mut **tx).await?;
     anyhow::ensure!(
         generation.is_none_or(|generation| generation == current.account_generation),
         crate::importers::harvest::account_switch::ChangeError::OldImport
@@ -540,9 +581,8 @@ pub async fn retry(pool: &sqlx::PgPool, org_id: Uuid, id: Uuid) -> anyhow::Resul
         id,
         org_id,
     )
-    .execute(&mut *tx)
+    .execute(&mut **tx)
     .await?;
-    tx.commit().await?;
     Ok(result.rows_affected() == 1)
 }
 

crates/horae/src/jobs/access.rs

@@ -0,0 +1,42 @@
+//! Current user authority for import commands and retained results, not worker leases.
+
+use horae_core::types::OrgRole;
+use sqlx::{PgPool, Postgres, Transaction};
+use uuid::Uuid;
+
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum AccessError {
+    #[error("Active administrator access required")]
+    Forbidden,
+    #[error(transparent)]
+    Database(#[from] sqlx::Error),
+}
+
+/// Hold current tenant-bound authority through a bounded operation's commit.
+pub(crate) async fn begin_admin_access(
+    pool: &PgPool,
+    org_id: Uuid,
+    actor_id: Uuid,
+) -> Result<Transaction<'_, Postgres>, AccessError> {
+    let mut tx = pool.begin().await?;
+    sqlx::query!("SET TRANSACTION ISOLATION LEVEL READ COMMITTED")
+        .execute(&mut *tx)
+        .await?;
+    crate::db::lock_organization(&mut tx, org_id, crate::db::OrganizationLock::Shared)
+        .await
+        .map_err(|error| match error {
+            sqlx::Error::RowNotFound => AccessError::Forbidden,
+            other => AccessError::Database(other),
+        })?;
+    sqlx::query_scalar!(
+        "SELECT id FROM users WHERE id = $1 AND org_id = $2
+           AND active AND org_role = $3 FOR SHARE",
+        actor_id,
+        org_id,
+        OrgRole::Admin as OrgRole,
+    )
+    .fetch_optional(&mut *tx)
+    .await?
+    .ok_or(AccessError::Forbidden)?;
+    Ok(tx)
+}