Upload files to "backend/app"

backend/app/models.py

@@ -0,0 +1,63 @@
+import json
+from typing import Annotated, Any, Literal
+
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, StringConstraints, field_validator, model_validator
+
+RunStatus = Literal["running", "success", "warning", "failed"]
+Name = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=200)]
+
+MAX_MESSAGE_CHARS = 32_000
+MAX_DETAILS_BYTES = 16_000
+
+
+class Strict(BaseModel):
+    # Reject unknown fields so a typo in a collector script fails loudly instead of being dropped.
+    model_config = ConfigDict(extra="forbid")
+
+
+class JobIn(Strict):
+    external_id: Name = Field(description="The source system's own stable ID for the job")
+    name: Name
+    # Omit to leave the stored value untouched; send null to clear it.
+    next_run_at: AwareDatetime | None = None
+
+
+class RunIn(Strict):
+    external_id: Name | None = Field(
+        None, description="The source's ID for this run. Defaults to started_at, so a 'running' "
+                          "report and its later completion report merge into one run.")
+    status: RunStatus
+    started_at: AwareDatetime
+    # Omit for a finished run to use the time the report was received.
+    finished_at: AwareDatetime | None = None
+    message: str = ""
+    size_bytes: int | None = Field(None, ge=0)
+    details: dict[str, Any] = Field(default_factory=dict)
+
+    @field_validator("message")
+    @classmethod
+    def truncate_message(cls, v: str) -> str:
+        # Backup logs can be huge; keep the head, which is where the summary usually is.
+        return v if len(v) <= MAX_MESSAGE_CHARS else v[:MAX_MESSAGE_CHARS] + "\n… [truncated]"
+
+    @field_validator("details")
+    @classmethod
+    def limit_details(cls, v: dict[str, Any]) -> dict[str, Any]:
+        if len(json.dumps(v)) > MAX_DETAILS_BYTES:
+            raise ValueError(f"details must serialize to at most {MAX_DETAILS_BYTES} bytes")
+        return v
+
+    @model_validator(mode="after")
+    def check_times(self) -> "RunIn":
+        if self.status == "running" and self.finished_at is not None:
+            raise ValueError("a running run cannot have finished_at")
+        if self.finished_at is not None and self.finished_at < self.started_at:
+            raise ValueError("finished_at is before started_at")
+        return self
+
+
+class IngestIn(Strict):
+    tenant: Name = Field(description="Company name as shown on the dashboard")
+    source_system: Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=100)]
+    job: JobIn
+    run: RunIn | None = Field(None, description="Omit to only register/update the job (e.g. next_run_at)")

backend/app/repo.py

@@ -0,0 +1,162 @@
+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 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

backend/app/settings.py

@@ -0,0 +1,22 @@
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+# The modal shows this many days of run history; retention can never be shorter.
+HISTORY_DAYS = 30
+
+
+@dataclass(frozen=True)
+class Settings:
+    db_path: Path
+    retention_days: int = 90
+    # Dev only: serve ../public from the API so you can run without nginx.
+    serve_static: bool = False
+
+
+def load_settings() -> Settings:
+    return Settings(
+        db_path=Path(os.environ.get("BACKUP_DASH_DB", "backup-dashboard.db")),
+        retention_days=max(HISTORY_DAYS, int(os.environ.get("BACKUP_DASH_RETENTION_DAYS", "90"))),
+        serve_static=os.environ.get("BACKUP_DASH_SERVE_STATIC") == "1",
+    )

backend/app/timeutil.py

@@ -0,0 +1,12 @@
+from datetime import UTC, datetime
+
+# Every timestamp in the DB uses this exact format, so string comparison == chronological comparison.
+DB_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
+
+
+def utcnow() -> datetime:
+    return datetime.now(UTC).replace(microsecond=0)
+
+
+def to_db(dt: datetime) -> str:
+    return dt.astimezone(UTC).strftime(DB_FORMAT)