Mangalord/backend/src/repo/crawl_metrics.rs

//! DB access for the `crawl_metrics` timing log (migration 0028).
//!
//! `record` is the best-effort write called from each crawl operation;
//! `summary` and `list_ops` are the admin Metrics-tab reads. Mirrors the
//! plain-fn + `query_as` style used across `repo`.

use chrono::{DateTime, Utc};
use sqlx::PgPool;
use uuid::Uuid;

use crate::domain::crawl_metrics::{MetricsBucket, OpRow, OpSummary};

/// Time-bucket granularity for trend series. A closed enum (not a free
/// string) so the `date_trunc` unit can never be attacker-controlled.
#[derive(Debug, Clone, Copy)]
pub enum Bucket {
    Hour,
    Day,
}

impl Bucket {
    pub fn as_str(self) -> &'static str {
        match self {
            Bucket::Hour => "hour",
            Bucket::Day => "day",
        }
    }
}

/// Operation kinds, kept as consts so call sites and the CHECK constraint
/// don't drift from a free-typed literal.
pub const OP_MANGA_LIST: &str = "manga_list";
pub const OP_MANGA_DETAIL: &str = "manga_detail";
pub const OP_MANGA_COVER: &str = "manga_cover";
pub const OP_CHAPTER: &str = "chapter";

/// Insert one completed-operation timing row. Best-effort: callers log and
/// continue on error so a metrics-write failure never fails the actual crawl
/// work. `outcome` is `"ok"` or `"failed"`.
#[allow(clippy::too_many_arguments)]
pub async fn record(
    pool: &PgPool,
    op: &str,
    manga_id: Option<Uuid>,
    chapter_id: Option<Uuid>,
    outcome: &str,
    duration_ms: i64,
    items: Option<i32>,
    error: Option<&str>,
) -> sqlx::Result<()> {
    sqlx::query(
        "INSERT INTO crawl_metrics \
            (op, manga_id, chapter_id, outcome, duration_ms, items, error) \
         VALUES ($1, $2, $3, $4, $5, $6, $7)",
    )
    .bind(op)
    .bind(manga_id)
    .bind(chapter_id)
    .bind(outcome)
    .bind(duration_ms)
    .bind(items)
    .bind(error)
    .execute(pool)
    .await?;
    Ok(())
}

/// Per-op average roll-up over an optional time window (`since = None` → all
/// time). One row per `op` that has any metric in the window, with mean
/// duration, counts, success split, and mean `items`.
pub async fn summary(
    pool: &PgPool,
    since: Option<DateTime<Utc>>,
) -> sqlx::Result<Vec<OpSummary>> {
    sqlx::query_as::<_, OpSummary>(
        r#"
        SELECT
            op,
            AVG(duration_ms)::float8                              AS avg_ms,
            COUNT(*)                                              AS n,
            COUNT(*) FILTER (WHERE outcome = 'ok')                AS ok,
            COUNT(*) FILTER (WHERE outcome = 'failed')            AS failed,
            AVG(items)::float8                                    AS avg_items
          FROM crawl_metrics
         WHERE ($1::timestamptz IS NULL OR finished_at >= $1)
         GROUP BY op
         ORDER BY op
        "#,
    )
    .bind(since)
    .fetch_all(pool)
    .await
}

/// Bucketed throughput/success/duration series over an optional window.
/// Plain `GROUP BY date_trunc(...)` — empty intervals are absent and the
/// client fills the continuous axis. Uses `crawl_metrics_time_idx`.
pub async fn series(
    pool: &PgPool,
    bucket: Bucket,
    since: Option<DateTime<Utc>>,
) -> sqlx::Result<Vec<MetricsBucket>> {
    sqlx::query_as::<_, MetricsBucket>(
        r#"
        SELECT
            date_trunc($1, finished_at)                  AS t,
            COUNT(*)                                     AS n,
            COUNT(*) FILTER (WHERE outcome = 'ok')       AS ok,
            COUNT(*) FILTER (WHERE outcome = 'failed')   AS failed,
            AVG(duration_ms)::float8                     AS avg_ms
          FROM crawl_metrics
         WHERE ($2::timestamptz IS NULL OR finished_at >= $2)
         GROUP BY t
         ORDER BY t
        "#,
    )
    .bind(bucket.as_str())
    .bind(since)
    .fetch_all(pool)
    .await
}

/// Filters for [`list_ops`]. `None`/empty widens the scope.
#[derive(Debug, Default, Clone)]
pub struct OpFilter<'a> {
    pub op: Option<&'a str>,
    pub outcome: Option<&'a str>,
    pub since: Option<DateTime<Utc>>,
}

/// Paginated, newest-first recent-operations log resolved to manga/chapter
/// labels. Returns the page slice plus the filtered total.
pub async fn list_ops(
    pool: &PgPool,
    filter: OpFilter<'_>,
    limit: i64,
    offset: i64,
) -> sqlx::Result<(Vec<OpRow>, i64)> {
    let items = sqlx::query_as::<_, OpRow>(
        r#"
        SELECT
            cm.id,
            cm.op,
            cm.manga_id,
            m.title           AS manga_title,
            cm.chapter_id,
            c.number          AS chapter_number,
            cm.outcome,
            cm.duration_ms,
            cm.items,
            cm.error,
            cm.finished_at
          FROM crawl_metrics cm
          LEFT JOIN mangas m   ON m.id = cm.manga_id
          LEFT JOIN chapters c ON c.id = cm.chapter_id
         WHERE ($1::text IS NULL OR cm.op = $1)
           AND ($2::text IS NULL OR cm.outcome = $2)
           AND ($3::timestamptz IS NULL OR cm.finished_at >= $3)
         ORDER BY cm.finished_at DESC
         LIMIT $4 OFFSET $5
        "#,
    )
    .bind(filter.op)
    .bind(filter.outcome)
    .bind(filter.since)
    .bind(limit)
    .bind(offset)
    .fetch_all(pool)
    .await?;

    let (total,): (i64,) = sqlx::query_as(
        r#"
        SELECT COUNT(*)
          FROM crawl_metrics cm
         WHERE ($1::text IS NULL OR cm.op = $1)
           AND ($2::text IS NULL OR cm.outcome = $2)
           AND ($3::timestamptz IS NULL OR cm.finished_at >= $3)
        "#,
    )
    .bind(filter.op)
    .bind(filter.outcome)
    .bind(filter.since)
    .fetch_one(pool)
    .await?;

    Ok((items, total))
}

/// Delete metric rows older than `retention_days`. `0` disables the reaper
/// (returns 0 without touching the table). Mirrors `jobs::reap_done`.
pub async fn reap(pool: &PgPool, retention_days: u32) -> sqlx::Result<u64> {
    if retention_days == 0 {
        return Ok(0);
    }
    let result = sqlx::query(
        "DELETE FROM crawl_metrics \
          WHERE finished_at < now() - ($1::bigint || ' days')::interval",
    )
    .bind(retention_days as i64)
    .execute(pool)
    .await?;
    Ok(result.rows_affected())
}