neiam /dms

//! Per-upload, per-kind job state tracking.
//!
//! Backed by the `upload_jobs` table; lifecycle:
//!   pending → processing → complete (terminal, success)
//!                       ↘ failed   (terminal, error recorded)
//!
//! After every terminal transition we recompute the parent upload's
//! `processing_status` so the listing UI reflects reality:
//!   - any row pending/processing → "processing"
//!   - all rows terminal, none failed → "complete"
//!   - all rows terminal, ≥1 failed → "partial"

use chrono::Utc;
use sea_orm::sea_query::{Expr, OnConflict};
use sea_orm::*;

use crate::entities::{upload_jobs, uploads};
use crate::queue::JobKind;

pub const STATUS_PENDING: &str = "pending";
pub const STATUS_PROCESSING: &str = "processing";
pub const STATUS_COMPLETE: &str = "complete";
pub const STATUS_FAILED: &str = "failed";

pub const UPLOAD_STATUS_PROCESSING: &str = "processing";
pub const UPLOAD_STATUS_COMPLETE: &str = "complete";
pub const UPLOAD_STATUS_PARTIAL: &str = "partial";

/// Insert a pending row for (upload, kind), or reset an existing row to
/// pending (used when a job is re-enqueued — manual force, follow-up jobs,
/// etc.). Idempotent.
pub async fn upsert_pending(
    db: &DatabaseConnection,
    upload_id: i64,
    kind: JobKind,
) -> Result<(), DbErr> {
    let active = upload_jobs::ActiveModel {
        upload_id: Set(upload_id),
        kind: Set(kind.as_str().to_string()),
        status: Set(STATUS_PENDING.to_string()),
        error: Set(None),
        completed_at: Set(None),
        started_at: Set(None),
        ..Default::default()
    };
    upload_jobs::Entity::insert(active)
        .on_conflict(
            OnConflict::columns([upload_jobs::Column::UploadId, upload_jobs::Column::Kind])
                .update_columns([
                    upload_jobs::Column::Status,
                    upload_jobs::Column::Error,
                    upload_jobs::Column::StartedAt,
                    upload_jobs::Column::CompletedAt,
                    upload_jobs::Column::UpdatedAt,
                ])
                .to_owned(),
        )
        .exec(db)
        .await?;
    Ok(())
}

pub async fn mark_processing(
    db: &DatabaseConnection,
    upload_id: i64,
    kind: JobKind,
) -> Result<(), DbErr> {
    let now = Utc::now();
    upload_jobs::Entity::update_many()
        .filter(upload_jobs::Column::UploadId.eq(upload_id))
        .filter(upload_jobs::Column::Kind.eq(kind.as_str()))
        .col_expr(upload_jobs::Column::Status, Expr::value(STATUS_PROCESSING))
        .col_expr(upload_jobs::Column::StartedAt, Expr::value(Some(now)))
        .col_expr(upload_jobs::Column::UpdatedAt, Expr::value(now))
        .col_expr(upload_jobs::Column::Attempts, Expr::col(upload_jobs::Column::Attempts).add(1))
        .exec(db)
        .await?;
    Ok(())
}

pub async fn mark_complete(
    db: &DatabaseConnection,
    upload_id: i64,
    kind: JobKind,
) -> Result<(), DbErr> {
    let now = Utc::now();
    upload_jobs::Entity::update_many()
        .filter(upload_jobs::Column::UploadId.eq(upload_id))
        .filter(upload_jobs::Column::Kind.eq(kind.as_str()))
        .col_expr(upload_jobs::Column::Status, Expr::value(STATUS_COMPLETE))
        .col_expr(upload_jobs::Column::CompletedAt, Expr::value(Some(now)))
        .col_expr(upload_jobs::Column::UpdatedAt, Expr::value(now))
        .col_expr(upload_jobs::Column::Error, Expr::value::<Option<String>>(None))
        .exec(db)
        .await?;
    recompute_aggregate(db, upload_id).await
}

pub async fn mark_failed(
    db: &DatabaseConnection,
    upload_id: i64,
    kind: JobKind,
    err: &str,
) -> Result<(), DbErr> {
    let now = Utc::now();
    upload_jobs::Entity::update_many()
        .filter(upload_jobs::Column::UploadId.eq(upload_id))
        .filter(upload_jobs::Column::Kind.eq(kind.as_str()))
        .col_expr(upload_jobs::Column::Status, Expr::value(STATUS_FAILED))
        .col_expr(upload_jobs::Column::Error, Expr::value(Some(err.to_string())))
        .col_expr(upload_jobs::Column::CompletedAt, Expr::value(Some(now)))
        .col_expr(upload_jobs::Column::UpdatedAt, Expr::value(now))
        .exec(db)
        .await?;
    recompute_aggregate(db, upload_id).await
}

/// Whether the (upload, kind) job is already in the terminal "complete"
/// state. Workers consult this for idempotency before doing real work.
pub async fn is_complete(
    db: &DatabaseConnection,
    upload_id: i64,
    kind: JobKind,
) -> Result<bool, DbErr> {
    let row = upload_jobs::Entity::find()
        .filter(upload_jobs::Column::UploadId.eq(upload_id))
        .filter(upload_jobs::Column::Kind.eq(kind.as_str()))
        .one(db)
        .await?;
    Ok(row.map(|r| r.status == STATUS_COMPLETE).unwrap_or(false))
}

/// Recompute uploads.processing_status from the row set in upload_jobs.
async fn recompute_aggregate(db: &DatabaseConnection, upload_id: i64) -> Result<(), DbErr> {
    let rows = upload_jobs::Entity::find()
        .filter(upload_jobs::Column::UploadId.eq(upload_id))
        .all(db)
        .await?;
    if rows.is_empty() {
        return Ok(());
    }

    let any_open = rows
        .iter()
        .any(|r| r.status == STATUS_PENDING || r.status == STATUS_PROCESSING);
    let any_failed = rows.iter().any(|r| r.status == STATUS_FAILED);

    let new_status = if any_open {
        UPLOAD_STATUS_PROCESSING
    } else if any_failed {
        UPLOAD_STATUS_PARTIAL
    } else {
        UPLOAD_STATUS_COMPLETE
    };

    uploads::Entity::update_many()
        .filter(uploads::Column::Id.eq(upload_id))
        .col_expr(uploads::Column::ProcessingStatus, Expr::value(new_status))
        .col_expr(uploads::Column::DateUpdated, Expr::value(Utc::now()))
        .exec(db)
        .await?;
    Ok(())
}

pub async fn jobs_for_upload(
    db: &DatabaseConnection,
    upload_id: i64,
) -> Result<Vec<upload_jobs::Model>, DbErr> {
    upload_jobs::Entity::find()
        .filter(upload_jobs::Column::UploadId.eq(upload_id))
        .order_by_asc(upload_jobs::Column::Kind)
        .all(db)
        .await
}