Backup-Dashboard/backend/app/repo.py

import json
import sqlite3
from datetime import datetime, timedelta
from typing import Any

from .db import transaction
from .models import IngestIn
from .timeutil import to_db, utcnow

# One row per job: `cur` is the most recent run (may be in progress), `fin` the most recent
# finished one. "Last run" on the dashboard is fin.finished_at, so a job that's currently
# running still shows when its previous run completed.
_JOB_SQL = """
SELECT j.id, t.name AS company_name, j.name AS job_name, j.source_system, j.next_run_at,
       cur.status,
       CASE WHEN cur.status = 'running' THEN cur.started_at END AS running_since,
       fin.status      AS last_status,
       fin.started_at  AS last_started_at,
       fin.finished_at AS last_run_at,
       unixepoch(fin.finished_at) - unixepoch(fin.started_at) AS duration_seconds,
       fin.size_bytes,
       fin.details,
       COALESCE(fin.message, cur.message, '') AS message
FROM jobs j
JOIN tenants t ON t.id = j.tenant_id
LEFT JOIN runs cur ON cur.id = (
    SELECT r.id FROM runs r WHERE r.job_id = j.id
    ORDER BY r.started_at DESC, r.id DESC LIMIT 1)
LEFT JOIN runs fin ON fin.id = (
    SELECT r.id FROM runs r WHERE r.job_id = j.id AND r.finished_at IS NOT NULL
    ORDER BY r.finished_at DESC, r.id DESC LIMIT 1)
"""

_SUMMARY_FIELDS = (
    "id", "company_name", "job_name", "source_system", "status",
    "running_since", "last_run_at", "next_run_at", "message",
)


def list_jobs(conn: sqlite3.Connection) -> list[dict[str, Any]]:
    rows = conn.execute(_JOB_SQL + " ORDER BY t.name, j.name").fetchall()
    return [{k: row[k] for k in _SUMMARY_FIELDS} for row in rows]


def list_tenant_jobs(conn: sqlite3.Connection, tenant_id: int) -> list[dict[str, Any]]:
    """A company's jobs with their current status, for reports."""
    rows = conn.execute(_JOB_SQL + " WHERE j.tenant_id = ? ORDER BY j.name", (tenant_id,)).fetchall()
    fields = (*_SUMMARY_FIELDS, "last_status")
    return [{k: row[k] for k in fields} for row in rows]


def get_job(conn: sqlite3.Connection, job_id: int) -> dict[str, Any] | None:
    row = conn.execute(_JOB_SQL + " WHERE j.id = ?", (job_id,)).fetchone()
    if row is None:
        return None
    job = dict(row)
    job["details"] = json.loads(job["details"]) if job["details"] else {}
    return job


def list_runs(conn: sqlite3.Connection, job_id: int, since: datetime) -> list[dict[str, Any]]:
    # Messages/details are left out: an hourly job has ~720 runs a month and logs can be large.
    # The UI fetches a single run's full record on demand.
    rows = conn.execute(
        """
        SELECT id, status, started_at, finished_at,
               unixepoch(finished_at) - unixepoch(started_at) AS duration_seconds, size_bytes
        FROM runs WHERE job_id = ? AND started_at >= ?
        ORDER BY started_at DESC, id DESC
        """,
        (job_id, to_db(since)),
    ).fetchall()
    return [dict(r) for r in rows]


def get_run(conn: sqlite3.Connection, run_id: int) -> dict[str, Any] | None:
    row = conn.execute(
        """
        SELECT id, job_id, status, started_at, finished_at,
               unixepoch(finished_at) - unixepoch(started_at) AS duration_seconds,
               size_bytes, message, details
        FROM runs WHERE id = ?
        """,
        (run_id,),
    ).fetchone()
    if row is None:
        return None
    run = dict(row)
    run["details"] = json.loads(run["details"])
    return run


def ingest(conn: sqlite3.Connection, item: IngestIn) -> tuple[int, int | None]:
    """Upsert tenant → job → run. Safe to call repeatedly with the same payload."""
    now = utcnow()
    now_s = to_db(now)
    with transaction(conn):
        conn.execute("INSERT INTO tenants (name) VALUES (?) ON CONFLICT (name) DO NOTHING", (item.tenant,))
        tenant_id = conn.execute("SELECT id FROM tenants WHERE name = ?", (item.tenant,)).fetchone()[0]

        job_id = conn.execute(
            """
            INSERT INTO jobs (tenant_id, source_system, external_id, name, next_run_at, created_at, updated_at)
            VALUES (:tenant_id, :source, :ext, :name, :next_run, :now, :now)
            ON CONFLICT (tenant_id, source_system, external_id) DO UPDATE SET
                name        = excluded.name,
                next_run_at = CASE WHEN :set_next_run THEN excluded.next_run_at ELSE jobs.next_run_at END,
                updated_at  = excluded.updated_at
            RETURNING id
            """,
            {
                "tenant_id": tenant_id,
                "source": item.source_system,
                "ext": item.job.external_id,
                "name": item.job.name,
                "next_run": to_db(item.job.next_run_at) if item.job.next_run_at else None,
                "set_next_run": "next_run_at" in item.job.model_fields_set,
                "now": now_s,
            },
        ).fetchone()[0]

        if item.run is None:
            return job_id, None

        run = item.run
        started = to_db(run.started_at)
        finished = None
        if run.status != "running":
            finished = to_db(run.finished_at or max(now, run.started_at))
        params = {
            "job_id": job_id,
            "ext": run.external_id or started,
            "status": run.status,
            "started": started,
            "finished": finished,
            "message": run.message,
            "size": run.size_bytes,
            "details": json.dumps(run.details),
            "now": now_s,
        }
        # The WHERE stops a late/retried "running" report from reopening a run that already finished.
        row = conn.execute(
            """
            INSERT INTO runs (job_id, external_id, status, started_at, finished_at, message,
                              size_bytes, details, received_at)
            VALUES (:job_id, :ext, :status, :started, :finished, :message, :size, :details, :now)
            ON CONFLICT (job_id, external_id) DO UPDATE SET
                status      = excluded.status,
                started_at  = excluded.started_at,
                finished_at = excluded.finished_at,
                message     = excluded.message,
                size_bytes  = excluded.size_bytes,
                details     = excluded.details,
                received_at = excluded.received_at
            WHERE NOT (runs.status != 'running' AND excluded.status = 'running')
            RETURNING id
            """,
            params,
        ).fetchone()
        if row is None:
            row = conn.execute(
                "SELECT id FROM runs WHERE job_id = :job_id AND external_id = :ext", params
            ).fetchone()
        return job_id, row[0]


def purge_old_runs(conn: sqlite3.Connection, retention_days: int) -> int:
    cutoff = to_db(utcnow() - timedelta(days=retention_days))
    return conn.execute("DELETE FROM runs WHERE started_at < ?", (cutoff,)).rowcount