neiam /dms

//! Failed-job retry surface.
//!
//! Lists every `upload_jobs` row in `failed` state for the authenticated
//! user and exposes a "retry all" + per-job retry. Retry is mechanically
//! identical to the existing `enqueue_job` flow: flip the row back to
//! pending and republish a `JobMessage` to RabbitMQ.

use axum::{
    extract::{Path, State},
    http::StatusCode,
    response::{Html, Redirect},
    Json,
};
use minijinja::context;
use sea_orm::*;
use serde::Serialize;
use tracing::{error, info};
use uuid::Uuid;

use crate::entities::{upload_jobs, uploads};
use crate::jobs;
use crate::middleware::UserContext;
use crate::queue::{publish_job, JobKind, JobMessage};
use crate::AppState;

#[derive(Debug, Serialize, Clone)]
pub struct FailedRow {
    pub job_id: Uuid,
    pub upload_id: Uuid,
    pub kind: String,
    pub attempts: i32,
    pub error: Option<String>,
    pub updated_at: chrono::DateTime<chrono::FixedOffset>,
    pub file_name_original: String,
    pub item_path: String,
}

/// Pull every failed job for the user, joined to the parent upload so the
/// listing has filenames + thumbnails to render.
async fn load_failed(
    db: &DatabaseConnection,
    user_id: Uuid,
) -> Result<Vec<FailedRow>, DbErr> {
    let rows = upload_jobs::Entity::find()
        .filter(upload_jobs::Column::Status.eq(jobs::STATUS_FAILED))
        .find_also_related(uploads::Entity)
        .order_by_desc(upload_jobs::Column::UpdatedAt)
        .all(db)
        .await?;
    let mut out = Vec::with_capacity(rows.len());
    for (job, upload) in rows {
        let Some(upload) = upload else { continue };
        if upload.user_id != user_id {
            continue;
        }
        out.push(FailedRow {
            job_id: job.id,
            upload_id: job.upload_id,
            kind: job.kind,
            attempts: job.attempts,
            error: job.error,
            updated_at: job.updated_at,
            file_name_original: upload.file_name_original,
            item_path: upload.item_path,
        });
    }
    Ok(out)
}

pub async fn show_failed(
    State(state): State<AppState>,
    user: UserContext,
) -> Result<Html<String>, StatusCode> {
    let rows = load_failed(state.db.as_ref(), user.user.id).await.map_err(|e| {
        error!(error = %e, "load failed jobs failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    let template = state.env.get_template("failed.html").map_err(|e| {
        error!(error = %e, "load failed.html failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    let rendered = template
        .render(context! {
            title => "Failed jobs",
            user => user.user,
            rows => rows,
        })
        .map_err(|e| {
            error!(error = %e, "render failed.html failed");
            StatusCode::INTERNAL_SERVER_ERROR
        })?;
    Ok(Html(rendered))
}

pub async fn api_list_failed(
    State(state): State<AppState>,
    user: UserContext,
) -> Result<Json<serde_json::Value>, StatusCode> {
    let rows = load_failed(state.db.as_ref(), user.user.id).await.map_err(|e| {
        error!(error = %e, "load failed jobs failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    Ok(Json(serde_json::json!({ "total": rows.len(), "items": rows })))
}

/// Enqueue retries for every failed job belonging to the user. Returns the
/// number of jobs that were queued; failures to publish individual messages
/// are logged but don't fail the whole batch.
async fn retry_failed_for(
    state: &AppState,
    user_id: Uuid,
) -> Result<usize, StatusCode> {
    let rows = load_failed(state.db.as_ref(), user_id).await.map_err(|e| {
        error!(error = %e, "load failed jobs failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    if rows.is_empty() {
        return Ok(0);
    }
    let conn = crate::queue::connect(&state.config.amqp_url).await.map_err(|e| {
        error!(error = %e, "amqp connect failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    let mut requeued = 0usize;
    for row in rows {
        let Some(kind) = JobKind::from_str(&row.kind) else {
            error!(kind = %row.kind, "unknown job kind on failed row, skipping");
            continue;
        };
        if let Err(e) = jobs::upsert_pending(state.db.as_ref(), row.upload_id, kind).await {
            error!(error = %e, upload_id = %row.upload_id, kind = ?kind, "upsert pending failed during retry");
            continue;
        }
        let msg = JobMessage::new(kind, row.upload_id, user_id);
        if let Err(e) = publish_job(&conn, &msg).await {
            error!(error = %e, upload_id = %row.upload_id, kind = ?kind, "publish failed during retry");
            continue;
        }
        requeued += 1;
    }
    info!(user_id = %user_id, requeued, "retry-all-failed completed");
    Ok(requeued)
}

pub async fn retry_all_failed(
    State(state): State<AppState>,
    user: UserContext,
) -> Result<Redirect, StatusCode> {
    let n = retry_failed_for(&state, user.user.id).await?;
    let msg = if n == 0 {
        "No failed jobs to retry.".to_string()
    } else {
        format!("Re-queued {n} failed job(s).")
    };
    let encoded = urlencode(&msg);
    Ok(Redirect::to(&format!("/failed?notice={encoded}")))
}

pub async fn api_retry_all_failed(
    State(state): State<AppState>,
    user: UserContext,
) -> Result<Json<serde_json::Value>, StatusCode> {
    let n = retry_failed_for(&state, user.user.id).await?;
    Ok(Json(serde_json::json!({ "requeued": n })))
}

/// Retry one specific failed (upload, kind) row. The upload must belong to
/// the caller; non-failed rows still requeue (idempotent — same effect as
/// the existing /uploads/{id}/jobs endpoint with no `force` flag).
pub async fn retry_one(
    State(state): State<AppState>,
    user: UserContext,
    Path((upload_id, kind_str)): Path<(Uuid, String)>,
) -> Result<Redirect, StatusCode> {
    let upload = uploads::Entity::find_by_id(upload_id)
        .filter(uploads::Column::UserId.eq(user.user.id))
        .one(state.db.as_ref())
        .await
        .map_err(|e| {
            error!(error = %e, "retry_one: find upload failed");
            StatusCode::INTERNAL_SERVER_ERROR
        })?
        .ok_or(StatusCode::NOT_FOUND)?;
    let kind = JobKind::from_str(&kind_str).ok_or(StatusCode::BAD_REQUEST)?;
    if let Err(e) = jobs::upsert_pending(state.db.as_ref(), upload.id, kind).await {
        error!(error = %e, "upsert pending failed");
        return Err(StatusCode::INTERNAL_SERVER_ERROR);
    }
    let conn = crate::queue::connect(&state.config.amqp_url).await.map_err(|e| {
        error!(error = %e, "amqp connect failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    let msg = JobMessage::new(kind, upload.id, upload.user_id);
    publish_job(&conn, &msg).await.map_err(|e| {
        error!(error = %e, "publish failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    info!(upload_id = %upload.id, kind = ?kind, "single-failed retry queued");
    Ok(Redirect::to("/failed?notice=Retry+queued."))
}

fn urlencode(s: &str) -> String {
    s.bytes()
        .map(|b| match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                (b as char).to_string()
            }
            b' ' => "+".to_string(),
            _ => format!("%{:02X}", b),
        })
        .collect()
}