diff --git a/backend/app/models.py b/backend/app/models.py new file mode 100644 index 0000000..53f5070 --- /dev/null +++ b/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)") diff --git a/backend/app/repo.py b/backend/app/repo.py new file mode 100644 index 0000000..7328cf5 --- /dev/null +++ b/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 diff --git a/backend/app/settings.py b/backend/app/settings.py new file mode 100644 index 0000000..005a73a --- /dev/null +++ b/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", + ) diff --git a/backend/app/timeutil.py b/backend/app/timeutil.py new file mode 100644 index 0000000..18acd07 --- /dev/null +++ b/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)