//! Import job CRUD: create, update progress, complete, list.

use sqlx::PgPool;

use super::enums::ImportJobStatus;
use super::id_types::{ImportJobId, ProjectId, UserId};
use super::models::DbImportJob;
use crate::error::Result;
use crate::import::ImportSource;

/// Create a new import job in `pending` status.
#[tracing::instrument(skip_all)]
pub async fn create_import_job(
    pool: &PgPool,
    user_id: UserId,
    project_id: ProjectId,
    source: ImportSource,
    total_rows: i32,
) -> Result<DbImportJob> {
    let job = sqlx::query_as::<_, DbImportJob>(
        r"
        INSERT INTO import_jobs (user_id, project_id, source, total_rows)
        VALUES ($1, $2, $3, $4)
        RETURNING *
        ",
    )
    .bind(user_id)
    .bind(project_id)
    .bind(source)
    .bind(total_rows)
    .fetch_one(pool)
    .await?;

    Ok(job)
}

/// Update the processing progress of an import job.
#[tracing::instrument(skip_all)]
pub async fn update_import_progress(
    pool: &PgPool,
    job_id: ImportJobId,
    processed_rows: i32,
    created_rows: i32,
    skipped_rows: i32,
) -> Result<()> {
    sqlx::query(
        r"
        UPDATE import_jobs
        SET processed_rows = $2, created_rows = $3, skipped_rows = $4
        WHERE id = $1
        ",
    )
    .bind(job_id)
    .bind(processed_rows)
    .bind(created_rows)
    .bind(skipped_rows)
    .execute(pool)
    .await?;

    Ok(())
}

/// Update the status of an import job.
///
/// Typed on [`ImportJobStatus`] rather than a raw `&str` so a caller can't stamp
/// a status the reader ([`DbImportJob::status`]) can't decode. Entering
/// `Processing` also stamps `heartbeat_at = NOW()` so the reaper has a liveness
/// baseline from the moment the job starts running (the first chunk may take a
/// while); other transitions leave the heartbeat untouched.
#[tracing::instrument(skip_all)]
pub async fn update_import_status(
    pool: &PgPool,
    job_id: ImportJobId,
    status: ImportJobStatus,
) -> Result<()> {
    sqlx::query(
        "UPDATE import_jobs \
         SET status = $2, \
             heartbeat_at = CASE WHEN $2 = 'processing' THEN NOW() ELSE heartbeat_at END \
         WHERE id = $1",
    )
    .bind(job_id)
    .bind(status)
    .execute(pool)
    .await?;

    Ok(())
}

/// Refresh a running import's liveness heartbeat. Called after every processed
/// chunk so [`reap_stuck_import_jobs`] can tell a slow-but-progressing import
/// (fresh beat) from one whose owning process died (stale beat).
#[tracing::instrument(skip_all)]
pub async fn bump_import_heartbeat(pool: &PgPool, job_id: ImportJobId) -> Result<()> {
    sqlx::query(
        "UPDATE import_jobs SET heartbeat_at = NOW() WHERE id = $1 AND status = 'processing'",
    )
    .bind(job_id)
    .execute(pool)
    .await?;

    Ok(())
}

/// Fail every import job stuck in `processing` whose heartbeat has gone stale
/// past `max_age_secs`, i.e. the process running it died without reaching
/// `complete_import_job` / `fail_import_job`. Without this a crash mid-import
/// leaves the job `processing` forever. Returns the number of jobs reaped.
///
/// `COALESCE(heartbeat_at, created_at)` covers a job that crashed before its
/// first heartbeat (older rows and the window between insert and the first
/// `processing` stamp). Mirrors `scan_jobs::reap_stuck`'s liveness definition.
#[tracing::instrument(skip_all)]
pub async fn reap_stuck_import_jobs(pool: &PgPool, max_age_secs: i64) -> Result<u64> {
    let affected = sqlx::query(
        "UPDATE import_jobs \
         SET status = 'failed', \
             completed_at = NOW(), \
             error_log = COALESCE(error_log, '') || \
                 CASE WHEN error_log IS NULL OR error_log = '' THEN '' ELSE E'\\n' END || \
                 'import reaped: no heartbeat for over the stuck-job threshold (process likely crashed)' \
         WHERE status = 'processing' \
           AND COALESCE(heartbeat_at, created_at) < NOW() - ($1 || ' seconds')::interval",
    )
    .bind(max_age_secs.to_string())
    .execute(pool)
    .await?
    .rows_affected();

    Ok(affected)
}

/// Mark an import job as completed with optional error log.
#[tracing::instrument(skip_all)]
pub async fn complete_import_job(
    pool: &PgPool,
    job_id: ImportJobId,
    error_log: Option<String>,
) -> Result<()> {
    sqlx::query(
        r"
        UPDATE import_jobs
        SET status = 'completed', completed_at = NOW(), error_log = $2
        WHERE id = $1
        ",
    )
    .bind(job_id)
    .bind(error_log)
    .execute(pool)
    .await?;

    Ok(())
}

/// Mark an import job as failed with an error message.
#[tracing::instrument(skip_all)]
pub async fn fail_import_job(pool: &PgPool, job_id: ImportJobId, error: &str) -> Result<()> {
    sqlx::query(
        r"
        UPDATE import_jobs
        SET status = 'failed', completed_at = NOW(), error_log = $2
        WHERE id = $1
        ",
    )
    .bind(job_id)
    .bind(error)
    .execute(pool)
    .await?;

    Ok(())
}

/// Get a single import job by ID, scoped to a user.
#[tracing::instrument(skip_all)]
pub async fn get_import_job(
    pool: &PgPool,
    job_id: ImportJobId,
    user_id: UserId,
) -> Result<Option<DbImportJob>> {
    let job = sqlx::query_as::<_, DbImportJob>(
        "SELECT * FROM import_jobs WHERE id = $1 AND user_id = $2",
    )
    .bind(job_id)
    .bind(user_id)
    .fetch_optional(pool)
    .await?;

    Ok(job)
}

/// List import jobs for a user, most recent first.
#[tracing::instrument(skip_all)]
pub async fn list_import_jobs(pool: &PgPool, user_id: UserId) -> Result<Vec<DbImportJob>> {
    let jobs = sqlx::query_as::<_, DbImportJob>(
        "SELECT * FROM import_jobs WHERE user_id = $1 ORDER BY created_at DESC LIMIT 20",
    )
    .bind(user_id)
    .fetch_all(pool)
    .await?;

    Ok(jobs)
}