"""Read-only Gitea repository bundle backup readback. This service turns backup textfile metrics into an operator-facing evidence payload. It never executes a backup, restore, offsite sync, credential read, or repository mutation. """ from __future__ import annotations import json from datetime import datetime from pathlib import Path from typing import Any from urllib.parse import urlencode from urllib.request import Request, urlopen from zoneinfo import ZoneInfo from src.core.config import settings from src.services.snapshot_paths import default_operations_dir SCHEMA_VERSION = "gitea_repo_bundle_backup_readback_v1" TZ_TAIPEI = ZoneInfo("Asia/Taipei") DEFAULT_HOST = "188" PROMETHEUS_TIMEOUT_SECONDS = 4 _DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__)) _DEV_PROD_REPO_READBACK_FILE = "gitea-all-product-dev-prod-repo-readback.snapshot.json" def build_gitea_repo_bundle_backup_readback( *, metric_samples: list[dict[str, Any]], generated_at: str | None = None, host: str = DEFAULT_HOST, metrics_error: str = "", dev_prod_repo_refs: list[str] | None = None, dev_prod_repo_truth_ref: str = "", dev_prod_repo_truth_error: str = "", ) -> dict[str, Any]: """Build the readback payload from Prometheus vector samples.""" now = generated_at or datetime.now(TZ_TAIPEI).isoformat(timespec="seconds") filtered = [ sample for sample in metric_samples if str(sample.get("labels", {}).get("host", host)) == host ] metrics_present = bool(filtered) and not metrics_error scalar = _scalar_index(filtered) repo_rows = _repo_rows(filtered) dev_prod_coverage = _dev_prod_repo_truth_coverage( repo_rows=repo_rows, expected_repo_refs=dev_prod_repo_refs, source_ref=dev_prod_repo_truth_ref, source_error=dev_prod_repo_truth_error, ) expected_repo_count = int(_first_scalar(scalar, "awoooi_gitea_bundle_expected_repo_count", 0)) missing_count = int(_first_scalar(scalar, "awoooi_gitea_bundle_expected_repo_missing_count", 0)) failed_count = int(_first_scalar(scalar, "awoooi_gitea_bundle_failed_repo_count", 0)) checksum_missing_count = int(_first_scalar(scalar, "awoooi_gitea_bundle_checksum_missing_count", 0)) age_seconds = int(_first_scalar(scalar, "awoooi_gitea_bundle_age_seconds", 0)) fresh = _first_scalar(scalar, "awoooi_gitea_bundle_fresh", 0) == 1 all_expected_ok = _first_scalar(scalar, "awoooi_gitea_bundle_all_expected_ok", 0) == 1 sample_restore_value = _first_scalar( scalar, "awoooi_gitea_bundle_sample_restore_dry_run_ok", None, ) sample_restore_present = sample_restore_value is not None sample_restore_ok = sample_restore_value == 1 blockers: list[str] = [] if not metrics_present: blockers.append("gitea_bundle_metrics_readback_unavailable") if missing_count > 0: blockers.append("gitea_bundle_expected_repo_missing") if failed_count > 0: blockers.append("gitea_bundle_repo_failed") if checksum_missing_count > 0: blockers.append("gitea_bundle_checksum_missing") if metrics_present and not fresh: blockers.append("gitea_bundle_backup_stale") if metrics_present and not all_expected_ok: blockers.append("gitea_bundle_all_expected_not_ok") if metrics_present and not sample_restore_present: blockers.append("gitea_bundle_sample_restore_dry_run_missing") elif metrics_present and not sample_restore_ok: blockers.append("gitea_bundle_sample_restore_dry_run_failed") if dev_prod_coverage["source_error"]: blockers.append("gitea_bundle_dev_prod_repo_truth_unavailable") elif ( dev_prod_coverage["evaluated"] and not dev_prod_coverage["all_dev_prod_repos_have_bundle_backup"] ): blockers.append("gitea_bundle_dev_prod_repo_truth_missing") ready = not blockers if ready: status_value = "gitea_repo_bundle_backup_readback_ready" safe_next_step = "keep_daily_bundle_backup_and_restore_dry_run_monitoring" elif "gitea_bundle_metrics_readback_unavailable" in blockers: status_value = "blocked_gitea_bundle_metrics_readback_unavailable" safe_next_step = "restore_188_backup_metrics_readback_then_rerun_gitea_bundle_readback" elif "gitea_bundle_backup_stale" in blockers: status_value = "blocked_gitea_bundle_backup_stale" safe_next_step = "run_gitea_repo_bundle_backup_and_sample_restore_dry_run_no_secret_no_prod_restore" elif "gitea_bundle_sample_restore_dry_run_missing" in blockers: status_value = "blocked_gitea_bundle_restore_dry_run_missing" safe_next_step = "run_gitea_bundle_sample_restore_dry_run_no_secret_no_prod_restore" elif "gitea_bundle_dev_prod_repo_truth_missing" in blockers: status_value = "blocked_gitea_bundle_dev_prod_repo_truth_gap" safe_next_step = "run_gitea_bundle_backup_until_all_dev_prod_repos_are_covered" elif "gitea_bundle_dev_prod_repo_truth_unavailable" in blockers: status_value = "blocked_gitea_bundle_dev_prod_repo_truth_unavailable" safe_next_step = "restore_committed_dev_prod_repo_truth_snapshot_then_rerun_readback" else: status_value = "blocked_gitea_bundle_integrity_gap" safe_next_step = "inspect_awoooi_gitea_bundle_metrics_before_any_restore_or_repo_change" return { "schema_version": SCHEMA_VERSION, "generated_at": now, "status": status_value, "host": host, "readback_present": metrics_present, "ready": ready, "active_blockers": blockers, "active_blocker_count": len(blockers), "safe_next_step": safe_next_step, "operation_boundaries": { "read_only_api_allowed": True, "backup_execution_allowed": False, "restore_to_production_allowed": False, "offsite_sync_execution_allowed": False, "credential_read_allowed": False, "repo_visibility_change_allowed": False, "repo_delete_allowed": False, }, "summary": { "expected_repo_count": expected_repo_count, "repo_row_count": len(repo_rows), "expected_repo_missing_count": missing_count, "failed_repo_count": failed_count, "checksum_missing_count": checksum_missing_count, "bundle_age_seconds": age_seconds, "bundle_fresh": fresh, "all_expected_ok": all_expected_ok, "sample_restore_dry_run_present": sample_restore_present, "sample_restore_dry_run_ok": sample_restore_ok, "dev_prod_repo_truth_expected_count": dev_prod_coverage[ "expected_repo_count" ], "dev_prod_repo_truth_backup_covered_count": dev_prod_coverage[ "backup_covered_repo_count" ], "missing_dev_prod_repo_backup_count": dev_prod_coverage[ "missing_backup_repo_count" ], "all_dev_prod_repos_have_bundle_backup": dev_prod_coverage[ "all_dev_prod_repos_have_bundle_backup" ], }, "repos": repo_rows, "dev_prod_repo_truth_coverage": dev_prod_coverage, "metrics_error": metrics_error, "restore_dry_run": { "metric_name": "awoooi_gitea_bundle_sample_restore_dry_run_ok", "present": sample_restore_present, "ok": sample_restore_ok, "verifier_command": ( "scripts/backup/gitea-bundle-sample-restore-dry-run.sh " "--repo wooo/awoooi --write-textfile" ), "production_restore_performed": False, }, } def load_latest_gitea_repo_bundle_backup_readback( *, host: str = DEFAULT_HOST, prometheus_url: str | None = None, operations_dir: Path | None = None, ) -> dict[str, Any]: """Load live Prometheus readback, failing closed when metrics are unavailable.""" url = (prometheus_url or settings.PROMETHEUS_URL).rstrip("/") queries = [ f"awoooi_gitea_bundle_root_exists{{host=\"{host}\"}}", f"awoooi_gitea_bundle_manifest_present{{host=\"{host}\"}}", f"awoooi_gitea_bundle_age_seconds{{host=\"{host}\"}}", f"awoooi_gitea_bundle_fresh{{host=\"{host}\"}}", f"awoooi_gitea_bundle_expected_repo_count{{host=\"{host}\"}}", f"awoooi_gitea_bundle_expected_repo_missing_count{{host=\"{host}\"}}", f"awoooi_gitea_bundle_failed_repo_count{{host=\"{host}\"}}", f"awoooi_gitea_bundle_checksum_missing_count{{host=\"{host}\"}}", f"awoooi_gitea_bundle_all_expected_ok{{host=\"{host}\"}}", f"awoooi_gitea_bundle_repo_present{{host=\"{host}\"}}", f"awoooi_gitea_bundle_repo_ok{{host=\"{host}\"}}", f"awoooi_gitea_bundle_repo_head_count{{host=\"{host}\"}}", f"awoooi_gitea_bundle_checksum_digest_present{{host=\"{host}\"}}", f"awoooi_gitea_bundle_sample_restore_dry_run_ok{{host=\"{host}\"}}", ] samples: list[dict[str, Any]] = [] errors: list[str] = [] dev_prod_refs: list[str] = [] dev_prod_ref = f"docs/operations/{_DEV_PROD_REPO_READBACK_FILE}" dev_prod_error = "" try: dev_prod_refs = _load_dev_prod_repo_refs( (operations_dir or _DEFAULT_OPERATIONS_DIR) / _DEV_PROD_REPO_READBACK_FILE ) except Exception as exc: # noqa: BLE001 - fail closed with redacted error class dev_prod_error = exc.__class__.__name__ for query in queries: try: samples.extend(_query_prometheus_vector(url, query)) except Exception as exc: # noqa: BLE001 - fail closed with redacted error class errors.append(f"{query.split('{', 1)[0]}:{exc.__class__.__name__}") return build_gitea_repo_bundle_backup_readback( metric_samples=samples, host=host, metrics_error=",".join(errors), dev_prod_repo_refs=dev_prod_refs, dev_prod_repo_truth_ref=dev_prod_ref, dev_prod_repo_truth_error=dev_prod_error, ) def _query_prometheus_vector(prometheus_url: str, query: str) -> list[dict[str, Any]]: params = urlencode({"query": query}) request = Request( f"{prometheus_url}/api/v1/query?{params}", headers={"User-Agent": "awoooi-gitea-bundle-readback"}, ) with urlopen(request, timeout=PROMETHEUS_TIMEOUT_SECONDS) as response: # noqa: S310 payload = json.loads(response.read().decode("utf-8", errors="replace")) if payload.get("status") != "success": raise ValueError("prometheus_status_not_success") rows = payload.get("data", {}).get("result") or [] results: list[dict[str, Any]] = [] for row in rows: metric = row.get("metric") or {} value = row.get("value") or [] try: numeric = float(value[1]) except (IndexError, TypeError, ValueError): continue labels = {str(key): str(val) for key, val in metric.items() if key != "__name__"} results.append( { "name": str(metric.get("__name__") or query.split("{", 1)[0]), "labels": labels, "value": numeric, } ) return results def _scalar_index(samples: list[dict[str, Any]]) -> dict[str, list[float]]: indexed: dict[str, list[float]] = {} for sample in samples: indexed.setdefault(str(sample.get("name")), []).append(float(sample.get("value") or 0)) return indexed def _first_scalar( indexed: dict[str, list[float]], name: str, default: float | None, ) -> float | None: values = indexed.get(name) or [] return values[0] if values else default def _repo_rows(samples: list[dict[str, Any]]) -> list[dict[str, Any]]: repos: dict[str, dict[str, Any]] = {} for sample in samples: labels = sample.get("labels") or {} repo = str(labels.get("repo") or "") if not repo: continue row = repos.setdefault( repo, { "repo": repo, "present": False, "ok": False, "head_count": 0, "checksum_digest_present": False, }, ) name = str(sample.get("name")) value = float(sample.get("value") or 0) if name == "awoooi_gitea_bundle_repo_present": row["present"] = value == 1 elif name == "awoooi_gitea_bundle_repo_ok": row["ok"] = value == 1 elif name == "awoooi_gitea_bundle_repo_head_count": row["head_count"] = int(value) elif name == "awoooi_gitea_bundle_checksum_digest_present": row["checksum_digest_present"] = value == 1 return [repos[key] for key in sorted(repos)] def _load_dev_prod_repo_refs(path: Path) -> list[str]: payload = json.loads(path.read_text(encoding="utf-8")) products = payload.get("products") if isinstance(payload, dict) else None if not isinstance(products, list): raise ValueError("dev_prod_products_not_list") refs = [ str(product.get("gitea_repo") or "") for product in products if isinstance(product, dict) and product.get("gitea_repo") ] if not refs: raise ValueError("dev_prod_repo_refs_empty") return sorted(set(refs)) def _dev_prod_repo_truth_coverage( *, repo_rows: list[dict[str, Any]], expected_repo_refs: list[str] | None, source_ref: str, source_error: str, ) -> dict[str, Any]: expected = sorted(set(expected_repo_refs or [])) covered = sorted( str(row.get("repo")) for row in repo_rows if row.get("repo") and row.get("present") is True and row.get("ok") is True and row.get("checksum_digest_present") is True ) covered_set = set(covered) expected_set = set(expected) missing = sorted(expected_set - covered_set) unexpected = sorted(covered_set - expected_set) if expected else [] evaluated = bool(expected) or bool(source_error) return { "schema_version": "gitea_bundle_dev_prod_repo_truth_coverage_v1", "source_ref": source_ref, "source_error": source_error, "evaluated": evaluated, "expected_repo_count": len(expected), "backup_covered_repo_count": len(expected_set & covered_set), "missing_backup_repo_count": len(missing), "unexpected_backup_repo_count": len(unexpected), "all_dev_prod_repos_have_bundle_backup": evaluated and not source_error and bool(expected) and not missing, "expected_repos": expected, "missing_backup_repos": missing, "unexpected_backup_repos": unexpected, }