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, queue_depths, requeue_all_dlqs, requeue_dlq, JobKind, JobMessage, QueueDepth,
};
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)
}

#[derive(Debug, Serialize, Clone)]
pub struct CompletedRow {
    pub upload_id: Uuid,
    pub kind: String,
    pub completed_at: Option<chrono::DateTime<chrono::FixedOffset>>,
    pub file_name_original: String,
    pub item_path: String,
}

/// Last N completed (terminal-success) jobs per kind, scoped to the user.
/// Pulled per kind so a single noisy queue doesn't crowd the others out;
/// 5 kinds × `per_kind` rows total.
async fn load_recent_completed(
    db: &DatabaseConnection,
    user_id: Uuid,
    per_kind: u64,
) -> Result<std::collections::BTreeMap<String, Vec<CompletedRow>>, DbErr> {
    let mut by_kind = std::collections::BTreeMap::new();
    for kind in [
        JobKind::Thumbnail,
        JobKind::Ocr,
        JobKind::Describe,
        JobKind::ExtractTags,
        JobKind::SimilaritySearch,
    ] {
        let kind_str = kind.as_str().to_string();
        // Pull a generous slice and filter by upload-owner in Rust — joining
        // through find_also_related is the same shape as load_failed and
        // keeps the post-filter step uniform across the page.
        let rows = upload_jobs::Entity::find()
            .filter(upload_jobs::Column::Status.eq(jobs::STATUS_COMPLETE))
            .filter(upload_jobs::Column::Kind.eq(kind_str.clone()))
            .find_also_related(uploads::Entity)
            .order_by_desc(upload_jobs::Column::CompletedAt)
            .limit(per_kind * 5) // headroom for non-owner rows
            .all(db)
            .await?;
        let mut taken: Vec<CompletedRow> = Vec::new();
        for (job, upload) in rows {
            let Some(upload) = upload else { continue };
            if upload.user_id != user_id {
                continue;
            }
            taken.push(CompletedRow {
                upload_id: job.upload_id,
                kind: job.kind,
                completed_at: job.completed_at,
                file_name_original: upload.file_name_original,
                item_path: upload.item_path,
            });
            if taken.len() as u64 >= per_kind {
                break;
            }
        }
        by_kind.insert(kind_str, taken);
    }
    Ok(by_kind)
}

async fn load_queue_depths(state: &AppState) -> Vec<QueueDepth> {
    match crate::queue::connect(&state.config.amqp_url).await {
        Ok(conn) => match queue_depths(&conn).await {
            Ok(d) => d,
            Err(e) => {
                error!(error = %e, "queue_depths failed");
                Vec::new()
            }
        },
        Err(e) => {
            error!(error = %e, "amqp connect for queue_depths failed");
            Vec::new()
        }
    }
}

pub async fn show_failed(
    State(state): State<AppState>,
    user: UserContext,
) -> Result<Html<String>, StatusCode> {
    let user_id = user.user.id;
    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
    })?;
    let recent = load_recent_completed(state.db.as_ref(), user_id, 3).await.map_err(|e| {
        error!(error = %e, "load recent completed jobs failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    let depths = load_queue_depths(&state).await;
    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,
            recent => recent,
            depths => depths,
        })
        .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 user_id = user.user.id;
    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
    })?;
    let recent = load_recent_completed(state.db.as_ref(), user_id, 3).await.map_err(|e| {
        error!(error = %e, "load recent completed jobs failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    let depths = load_queue_depths(&state).await;
    Ok(Json(serde_json::json!({
        "failed": { "total": rows.len(), "items": rows },
        "recent_completed": recent,
        "queues": depths,
    })))
}

/// 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."))
}

/// Drain every per-kind DLQ back into its main queue and redirect to the
/// Work page. Operates broker-globally (DLQs aren't user-scoped — neither
/// is the depth display).
pub async fn requeue_all_dlqs_html(
    State(state): State<AppState>,
    _user: UserContext,
) -> Result<Redirect, StatusCode> {
    let conn = crate::queue::connect(&state.config.amqp_url).await.map_err(|e| {
        error!(error = %e, "amqp connect failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    let n = requeue_all_dlqs(&conn).await.map_err(|e| {
        error!(error = %e, "requeue_all_dlqs failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    info!(moved = n, "requeue all DLQs");
    let msg = if n == 0 {
        "No DLQ messages to requeue.".to_string()
    } else {
        format!("Requeued {n} DLQ message(s).")
    };
    Ok(Redirect::to(&format!("/failed?notice={}", urlencode(&msg))))
}

pub async fn api_requeue_all_dlqs(
    State(state): State<AppState>,
    _user: UserContext,
) -> Result<Json<serde_json::Value>, StatusCode> {
    let conn = crate::queue::connect(&state.config.amqp_url).await.map_err(|e| {
        error!(error = %e, "amqp connect failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    let n = requeue_all_dlqs(&conn).await.map_err(|e| {
        error!(error = %e, "requeue_all_dlqs failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    Ok(Json(serde_json::json!({ "requeued": n })))
}

pub async fn requeue_dlq_kind_html(
    State(state): State<AppState>,
    _user: UserContext,
    Path(kind_str): Path<String>,
) -> Result<Redirect, StatusCode> {
    let kind = JobKind::from_str(&kind_str).ok_or(StatusCode::BAD_REQUEST)?;
    let conn = crate::queue::connect(&state.config.amqp_url).await.map_err(|e| {
        error!(error = %e, "amqp connect failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    let n = requeue_dlq(&conn, kind).await.map_err(|e| {
        error!(error = %e, kind = ?kind, "requeue_dlq failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    info!(moved = n, kind = ?kind, "requeue one DLQ");
    let msg = if n == 0 {
        format!("No DLQ messages for {} to requeue.", kind.as_str())
    } else {
        format!("Requeued {n} {} DLQ message(s).", kind.as_str())
    };
    Ok(Redirect::to(&format!("/failed?notice={}", urlencode(&msg))))
}

pub async fn api_requeue_dlq_kind(
    State(state): State<AppState>,
    _user: UserContext,
    Path(kind_str): Path<String>,
) -> Result<Json<serde_json::Value>, StatusCode> {
    let kind = JobKind::from_str(&kind_str).ok_or(StatusCode::BAD_REQUEST)?;
    let conn = crate::queue::connect(&state.config.amqp_url).await.map_err(|e| {
        error!(error = %e, "amqp connect failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    let n = requeue_dlq(&conn, kind).await.map_err(|e| {
        error!(error = %e, kind = ?kind, "requeue_dlq failed");
        StatusCode::INTERNAL_SERVER_ERROR
    })?;
    Ok(Json(serde_json::json!({ "kind": kind.as_str(), "requeued": n })))
}

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()
}