1528 lines
67 KiB
Python
1528 lines
67 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
from collections import Counter
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from datetime import UTC, datetime, timedelta
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from .models import (
|
|
ArchitectureGraph,
|
|
RemediationSignal,
|
|
Repository,
|
|
RepositoryDetail,
|
|
RepositoryPolicy,
|
|
RepositoryState,
|
|
RepositorySummary,
|
|
RepositorySyncResult,
|
|
ScanComparison,
|
|
ScanJob,
|
|
ScanJobStatus,
|
|
ScanReport,
|
|
ScanSummary,
|
|
ScanTrigger,
|
|
Severity,
|
|
SourceOrigin,
|
|
TrendPoint,
|
|
finding_fingerprint,
|
|
)
|
|
from .rules import enrich_findings
|
|
|
|
MAX_HISTORY_PER_REPOSITORY = 200
|
|
SCHEMA_VERSION = 4
|
|
JOB_LEASE_SECONDS = 900
|
|
ACTIVE_JOB_STATUSES = (
|
|
ScanJobStatus.QUEUED.value,
|
|
ScanJobStatus.ACQUIRING_SOURCE.value,
|
|
ScanJobStatus.SCANNING.value,
|
|
ScanJobStatus.ANALYZING.value,
|
|
ScanJobStatus.PERSISTING.value,
|
|
)
|
|
|
|
|
|
class QueueFullError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class MaintenanceModeError(RuntimeError):
|
|
"""Raised when a new scan is requested while durable maintenance is active."""
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
def _iso(value: datetime | None) -> str | None:
|
|
return value.isoformat() if value else None
|
|
|
|
|
|
def _datetime(value: str | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(value)
|
|
except ValueError:
|
|
return None
|
|
return parsed.replace(tzinfo=UTC) if parsed.tzinfo is None else parsed
|
|
|
|
|
|
def local_repository_id(path: str) -> str:
|
|
resolved = str(Path(path).expanduser().resolve())
|
|
return f"local:{sha256(resolved.encode('utf-8')).hexdigest()[:24]}"
|
|
|
|
|
|
def _bounded_setting(name: str, default: int, *, minimum: int, maximum: int) -> int:
|
|
try:
|
|
value = int(os.getenv(name, str(default)))
|
|
except ValueError:
|
|
value = default
|
|
return min(maximum, max(minimum, value))
|
|
|
|
|
|
def history_per_repository() -> int:
|
|
return _bounded_setting("REPOPULSE_HISTORY_PER_REPOSITORY", MAX_HISTORY_PER_REPOSITORY, minimum=1, maximum=2_000)
|
|
|
|
|
|
def full_reports_per_repository() -> int:
|
|
return _bounded_setting("REPOPULSE_FULL_REPORTS_PER_REPOSITORY", 30, minimum=1, maximum=500)
|
|
|
|
|
|
def report_retention_days() -> int:
|
|
return _bounded_setting("REPOPULSE_REPORT_RETENTION_DAYS", 365, minimum=1, maximum=3_650)
|
|
|
|
|
|
def job_lease_seconds() -> int:
|
|
return _bounded_setting("REPOPULSE_JOB_LEASE_SECONDS", JOB_LEASE_SECONDS, minimum=30, maximum=3_600)
|
|
|
|
|
|
class ScanStore:
|
|
"""SQLite-backed catalog, durable scan history and recoverable job queue."""
|
|
|
|
def __init__(self, db_path: str | None = None) -> None:
|
|
raw_path = db_path or os.getenv("REPOPULSE_DB", "data/repopulse.db")
|
|
self.path = Path(raw_path)
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self.recovery_metadata: dict[str, str] | None = None
|
|
try:
|
|
self._validate_existing_database()
|
|
self._init_schema()
|
|
except sqlite3.DatabaseError as exc:
|
|
if not self._is_corruption_error(exc):
|
|
raise
|
|
backup = self._quarantine_corrupt_database()
|
|
self._init_schema()
|
|
self.recovery_metadata = {
|
|
"status": "degraded_recovered",
|
|
"detected_at": _now().isoformat(),
|
|
"quarantined_database": backup.name if backup else "unavailable",
|
|
"operator_action": "Verify the quarantined database, restore it if needed, then acknowledge recovery.",
|
|
}
|
|
self.set_state("database_recovery", self.recovery_metadata)
|
|
|
|
@staticmethod
|
|
def _is_corruption_error(error: sqlite3.DatabaseError) -> bool:
|
|
message = str(error).casefold()
|
|
return "not a database" in message or "malformed" in message or "corrupt" in message
|
|
|
|
def _quarantine_corrupt_database(self) -> Path | None:
|
|
if not self.path.exists():
|
|
return None
|
|
stamp = _now().strftime("%Y%m%dT%H%M%S%fZ")
|
|
backup = self.path.with_name(f"{self.path.name}.corrupt-{stamp}-{uuid4().hex[:8]}")
|
|
self.path.replace(backup)
|
|
for suffix in ("-wal", "-shm"):
|
|
sidecar = Path(f"{self.path}{suffix}")
|
|
if sidecar.exists():
|
|
sidecar.replace(Path(f"{backup}{suffix}"))
|
|
return backup
|
|
|
|
def _validate_existing_database(self) -> None:
|
|
if not self.path.exists() or self.path.stat().st_size == 0:
|
|
return
|
|
connection = sqlite3.connect(self.path, timeout=5.0)
|
|
try:
|
|
row = connection.execute("PRAGMA quick_check").fetchone()
|
|
if row is None or str(row[0]).casefold() != "ok":
|
|
raise sqlite3.DatabaseError("database quick_check reported corruption")
|
|
finally:
|
|
connection.close()
|
|
|
|
def _backup_before_migration(self, db: sqlite3.Connection, version: int) -> Path:
|
|
stamp = _now().strftime("%Y%m%dT%H%M%S%fZ")
|
|
backup = self.path.with_name(f"{self.path.name}.pre-v{version}-{stamp}.bak")
|
|
destination = sqlite3.connect(backup)
|
|
try:
|
|
db.backup(destination)
|
|
finally:
|
|
destination.close()
|
|
return backup
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
connection = sqlite3.connect(self.path, timeout=5.0)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA busy_timeout = 5000")
|
|
return connection
|
|
|
|
@contextmanager
|
|
def _connection(self) -> Iterator[sqlite3.Connection]:
|
|
connection = self._connect()
|
|
try:
|
|
yield connection
|
|
connection.commit()
|
|
except BaseException:
|
|
connection.rollback()
|
|
raise
|
|
finally:
|
|
connection.close()
|
|
|
|
@staticmethod
|
|
def _columns(db: sqlite3.Connection, table: str) -> set[str]:
|
|
return {str(row[1]) for row in db.execute(f"PRAGMA table_info({table})").fetchall()}
|
|
|
|
def _init_schema(self) -> None:
|
|
with self._connection() as db:
|
|
current_version = int(db.execute("PRAGMA user_version").fetchone()[0])
|
|
migration_backup: Path | None = None
|
|
if current_version > SCHEMA_VERSION:
|
|
raise RuntimeError(
|
|
f"Database schema version {current_version} is newer than supported version {SCHEMA_VERSION}"
|
|
)
|
|
if current_version and current_version < SCHEMA_VERSION:
|
|
migration_backup = self._backup_before_migration(db, current_version)
|
|
db.execute("PRAGMA journal_mode = WAL")
|
|
db.execute("PRAGMA synchronous = NORMAL")
|
|
db.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS scans (
|
|
id TEXT PRIMARY KEY,
|
|
repository_path TEXT NOT NULL,
|
|
repository_name TEXT NOT NULL,
|
|
scanned_at TEXT NOT NULL,
|
|
health_score INTEGER NOT NULL,
|
|
report_json TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
scan_columns = self._columns(db, "scans")
|
|
additions = {
|
|
"repository_id": "TEXT",
|
|
"history_key": "TEXT",
|
|
"source_origin": "TEXT",
|
|
"provider": "TEXT",
|
|
"branch": "TEXT",
|
|
"requested_revision": "TEXT",
|
|
"actual_revision": "TEXT",
|
|
"trigger": "TEXT",
|
|
"finding_count": "INTEGER",
|
|
"blocker_count": "INTEGER",
|
|
"warning_count": "INTEGER",
|
|
"release_readiness": "TEXT",
|
|
"primary_signal_json": "TEXT",
|
|
"added_findings": "INTEGER",
|
|
"resolved_findings": "INTEGER",
|
|
"triage_priority": "INTEGER",
|
|
"triage_rationale": "TEXT",
|
|
"report_compacted": "INTEGER NOT NULL DEFAULT 0",
|
|
}
|
|
for name, definition in additions.items():
|
|
if name not in scan_columns:
|
|
db.execute(f"ALTER TABLE scans ADD COLUMN {name} {definition}")
|
|
|
|
db.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS repositories (
|
|
id TEXT PRIMARY KEY,
|
|
provider TEXT NOT NULL,
|
|
provider_repository_id TEXT NOT NULL,
|
|
owner TEXT NOT NULL,
|
|
name TEXT NOT NULL,
|
|
full_name TEXT NOT NULL,
|
|
description TEXT,
|
|
html_url TEXT,
|
|
clone_url TEXT,
|
|
default_branch TEXT NOT NULL DEFAULT 'main',
|
|
private INTEGER NOT NULL DEFAULT 0,
|
|
archived INTEGER NOT NULL DEFAULT 0,
|
|
latest_revision TEXT,
|
|
latest_revision_at TEXT,
|
|
last_synced_at TEXT,
|
|
sync_error TEXT,
|
|
policy_json TEXT NOT NULL DEFAULT '{}',
|
|
source_path TEXT,
|
|
UNIQUE(provider, provider_repository_id)
|
|
)
|
|
"""
|
|
)
|
|
repository_columns = self._columns(db, "repositories")
|
|
repository_additions = {
|
|
"description": "TEXT",
|
|
"available": "INTEGER NOT NULL DEFAULT 1",
|
|
}
|
|
for name, definition in repository_additions.items():
|
|
if name not in repository_columns:
|
|
db.execute(f"ALTER TABLE repositories ADD COLUMN {name} {definition}")
|
|
db.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS scan_jobs (
|
|
id TEXT PRIMARY KEY,
|
|
repository_id TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
trigger TEXT NOT NULL,
|
|
requested_revision TEXT,
|
|
actual_revision TEXT,
|
|
phase TEXT NOT NULL,
|
|
progress INTEGER NOT NULL DEFAULT 0,
|
|
message TEXT,
|
|
created_at TEXT NOT NULL,
|
|
started_at TEXT,
|
|
updated_at TEXT NOT NULL,
|
|
finished_at TEXT,
|
|
duration_ms REAL,
|
|
scan_id TEXT,
|
|
error_code TEXT,
|
|
error_detail TEXT,
|
|
attempt INTEGER NOT NULL DEFAULT 0,
|
|
lease_until TEXT,
|
|
deduplicated INTEGER NOT NULL DEFAULT 0
|
|
)
|
|
"""
|
|
)
|
|
job_columns = self._columns(db, "scan_jobs")
|
|
if "duration_ms" not in job_columns:
|
|
db.execute("ALTER TABLE scan_jobs ADD COLUMN duration_ms REAL")
|
|
db.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS app_state (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
if migration_backup:
|
|
db.execute(
|
|
"""
|
|
INSERT INTO app_state (key, value, updated_at) VALUES (?, ?, ?)
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at
|
|
""",
|
|
(
|
|
"last_schema_migration_backup",
|
|
json.dumps(
|
|
{
|
|
"from_version": current_version,
|
|
"to_version": SCHEMA_VERSION,
|
|
"backup_file": migration_backup.name,
|
|
"created_at": _now().isoformat(),
|
|
},
|
|
separators=(",", ":"),
|
|
),
|
|
_now().isoformat(),
|
|
),
|
|
)
|
|
db.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_scans_repository_time ON scans(repository_path, scanned_at DESC)"
|
|
)
|
|
db.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_scans_repository_id_time ON scans(repository_id, scanned_at DESC)"
|
|
)
|
|
db.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_scans_history_time ON scans(history_key, scanned_at DESC)"
|
|
)
|
|
db.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_scans_repository_revision ON scans(repository_id, actual_revision)"
|
|
)
|
|
db.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_jobs_queue ON scan_jobs(status, created_at)"
|
|
)
|
|
db.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_jobs_repository ON scan_jobs(repository_id, created_at DESC)"
|
|
)
|
|
self._backfill_legacy_catalog(db)
|
|
self._backfill_scan_summaries(db)
|
|
db.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
|
|
|
|
def _backfill_legacy_catalog(self, db: sqlite3.Connection) -> None:
|
|
rows = db.execute(
|
|
"""
|
|
SELECT id, repository_path, repository_name, report_json, repository_id, history_key
|
|
FROM scans
|
|
WHERE repository_id IS NULL OR history_key IS NULL
|
|
"""
|
|
).fetchall()
|
|
for row in rows:
|
|
repository_id = row["repository_id"] or local_repository_id(row["repository_path"])
|
|
history_key = row["history_key"] or repository_id
|
|
self._upsert_local_repository_row(
|
|
db,
|
|
repository_id,
|
|
row["repository_name"],
|
|
row["repository_path"],
|
|
)
|
|
db.execute(
|
|
"""
|
|
UPDATE scans
|
|
SET repository_id = ?, history_key = ?, source_origin = COALESCE(source_origin, 'local'),
|
|
provider = COALESCE(provider, 'local'), trigger = COALESCE(trigger, 'local')
|
|
WHERE id = ?
|
|
""",
|
|
(repository_id, history_key, row["id"]),
|
|
)
|
|
|
|
@staticmethod
|
|
def _summary_fields(report: ScanReport, previous: ScanReport | None = None) -> dict[str, object]:
|
|
report = report.model_copy(update={"findings": enrich_findings(report.findings)})
|
|
active = [item for item in report.findings if not item.suppressed]
|
|
severity_rank = {Severity.BLOCKER: 0, Severity.WARNING: 1, Severity.INFO: 2}
|
|
primary = min(
|
|
active,
|
|
key=lambda item: (severity_rank[item.severity], -item.deduction, item.category, item.code),
|
|
default=None,
|
|
)
|
|
primary_signal = (
|
|
RemediationSignal(
|
|
code=primary.code,
|
|
category=primary.category,
|
|
severity=primary.severity,
|
|
message=primary.message,
|
|
count=sum(1 for item in active if item.code == primary.code),
|
|
guidance=primary.guidance,
|
|
)
|
|
if primary
|
|
else None
|
|
)
|
|
previous_active = [item for item in (previous.findings if previous else []) if not item.suppressed]
|
|
active_keys = {finding_fingerprint(item) for item in active}
|
|
previous_keys = {finding_fingerprint(item) for item in previous_active}
|
|
guidance = primary_signal.guidance if primary_signal else None
|
|
return {
|
|
"finding_count": len(active),
|
|
"blocker_count": sum(1 for item in active if item.severity == Severity.BLOCKER),
|
|
"warning_count": sum(1 for item in active if item.severity == Severity.WARNING),
|
|
"release_readiness": report.release_gate.status if report.release_gate else report.release_readiness,
|
|
"primary_signal_json": primary_signal.model_dump_json() if primary_signal else None,
|
|
"added_findings": len(active_keys - previous_keys) if previous else 0,
|
|
"resolved_findings": len(previous_keys - active_keys) if previous else 0,
|
|
"triage_priority": guidance.priority if guidance else 0,
|
|
"triage_rationale": guidance.priority_rationale if guidance else None,
|
|
}
|
|
|
|
def _backfill_scan_summaries(self, db: sqlite3.Connection) -> None:
|
|
rows = db.execute(
|
|
"SELECT rowid AS scan_rowid, * FROM scans "
|
|
"WHERE finding_count IS NULL OR blocker_count IS NULL OR release_readiness IS NULL"
|
|
).fetchall()
|
|
for row in rows:
|
|
report = self._report_from_row(row)
|
|
previous_row = db.execute(
|
|
"""
|
|
SELECT * FROM scans
|
|
WHERE history_key = ? AND (
|
|
scanned_at < ? OR (scanned_at = ? AND rowid < ?)
|
|
)
|
|
ORDER BY scanned_at DESC, rowid DESC LIMIT 1
|
|
""",
|
|
(row["history_key"], row["scanned_at"], row["scanned_at"], row["scan_rowid"]),
|
|
).fetchone()
|
|
previous = self._report_from_row(previous_row) if previous_row else None
|
|
fields = self._summary_fields(report, previous)
|
|
db.execute(
|
|
"""
|
|
UPDATE scans SET finding_count = ?, blocker_count = ?, warning_count = ?, release_readiness = ?,
|
|
primary_signal_json = ?, added_findings = ?, resolved_findings = ?, triage_priority = ?,
|
|
triage_rationale = ? WHERE id = ?
|
|
""",
|
|
(*fields.values(), row["id"]),
|
|
)
|
|
|
|
@staticmethod
|
|
def _upsert_local_repository_row(
|
|
db: sqlite3.Connection,
|
|
repository_id: str,
|
|
name: str,
|
|
source_path: str,
|
|
) -> None:
|
|
db.execute(
|
|
"""
|
|
INSERT INTO repositories (
|
|
id, provider, provider_repository_id, owner, name, full_name,
|
|
default_branch, source_path
|
|
) VALUES (?, 'local', ?, 'local', ?, ?, 'main', ?)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
name = excluded.name,
|
|
full_name = excluded.full_name,
|
|
source_path = COALESCE(excluded.source_path, repositories.source_path)
|
|
""",
|
|
(repository_id, repository_id.removeprefix("local:"), name, f"local/{name}", source_path),
|
|
)
|
|
|
|
@staticmethod
|
|
def _upsert_repository_row(
|
|
db: sqlite3.Connection,
|
|
repository: Repository,
|
|
source_path: str | None = None,
|
|
) -> None:
|
|
db.execute(
|
|
"""
|
|
INSERT INTO repositories (
|
|
id, provider, provider_repository_id, owner, name, full_name,
|
|
description, html_url, clone_url, default_branch, private, archived, available,
|
|
latest_revision, latest_revision_at, last_synced_at, sync_error,
|
|
policy_json, source_path
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
provider = excluded.provider,
|
|
provider_repository_id = excluded.provider_repository_id,
|
|
owner = excluded.owner,
|
|
name = excluded.name,
|
|
full_name = excluded.full_name,
|
|
description = excluded.description,
|
|
html_url = excluded.html_url,
|
|
clone_url = excluded.clone_url,
|
|
default_branch = excluded.default_branch,
|
|
private = excluded.private,
|
|
archived = excluded.archived,
|
|
available = excluded.available,
|
|
latest_revision = COALESCE(excluded.latest_revision, repositories.latest_revision),
|
|
latest_revision_at = COALESCE(excluded.latest_revision_at, repositories.latest_revision_at),
|
|
last_synced_at = COALESCE(excluded.last_synced_at, repositories.last_synced_at),
|
|
sync_error = excluded.sync_error,
|
|
policy_json = excluded.policy_json,
|
|
source_path = COALESCE(excluded.source_path, repositories.source_path)
|
|
""",
|
|
(
|
|
repository.id,
|
|
repository.provider,
|
|
repository.provider_repository_id,
|
|
repository.owner,
|
|
repository.name,
|
|
repository.full_name,
|
|
repository.description,
|
|
repository.html_url,
|
|
repository.clone_url,
|
|
repository.default_branch,
|
|
int(repository.private),
|
|
int(repository.archived),
|
|
int(repository.available),
|
|
repository.latest_revision,
|
|
_iso(repository.latest_revision_at),
|
|
_iso(repository.last_synced_at),
|
|
repository.sync_error,
|
|
repository.policy.model_dump_json(),
|
|
source_path,
|
|
),
|
|
)
|
|
|
|
def upsert_repository(self, repository: Repository, *, source_path: str | None = None) -> Repository:
|
|
with self._connection() as db:
|
|
self._upsert_repository_row(db, repository, source_path)
|
|
row = db.execute("SELECT * FROM repositories WHERE id = ?", (repository.id,)).fetchone()
|
|
return self._repository_from_row(row) if row else repository
|
|
|
|
@staticmethod
|
|
def _repository_from_row(row: sqlite3.Row) -> Repository:
|
|
try:
|
|
policy = RepositoryPolicy.model_validate_json(row["policy_json"] or "{}")
|
|
except (TypeError, ValueError):
|
|
policy = RepositoryPolicy()
|
|
return Repository(
|
|
id=row["id"],
|
|
provider=row["provider"],
|
|
provider_repository_id=row["provider_repository_id"],
|
|
owner=row["owner"],
|
|
name=row["name"],
|
|
full_name=row["full_name"],
|
|
description=row["description"],
|
|
html_url=row["html_url"],
|
|
clone_url=row["clone_url"],
|
|
default_branch=row["default_branch"] or "main",
|
|
private=bool(row["private"]),
|
|
archived=bool(row["archived"]),
|
|
available=bool(row["available"]),
|
|
latest_revision=row["latest_revision"],
|
|
latest_revision_at=_datetime(row["latest_revision_at"]),
|
|
last_synced_at=_datetime(row["last_synced_at"]),
|
|
sync_error=row["sync_error"],
|
|
policy=policy,
|
|
)
|
|
|
|
def get_repository(self, repository_id: str) -> Repository | None:
|
|
with self._connection() as db:
|
|
row = db.execute("SELECT * FROM repositories WHERE id = ?", (repository_id,)).fetchone()
|
|
return self._repository_from_row(row) if row else None
|
|
|
|
def get_repository_source_path(self, repository_id: str) -> str | None:
|
|
with self._connection() as db:
|
|
row = db.execute("SELECT source_path FROM repositories WHERE id = ?", (repository_id,)).fetchone()
|
|
return row["source_path"] if row else None
|
|
|
|
def _repository_summary(self, db: sqlite3.Connection, row: sqlite3.Row) -> RepositorySummary:
|
|
repository = self._repository_from_row(row)
|
|
latest = db.execute(
|
|
"""
|
|
SELECT * FROM scans WHERE repository_id = ?
|
|
ORDER BY scanned_at DESC, rowid DESC LIMIT 1
|
|
""",
|
|
(repository.id,),
|
|
).fetchone()
|
|
active = db.execute(
|
|
"""
|
|
SELECT * FROM scan_jobs WHERE repository_id = ? AND status IN (?, ?, ?, ?, ?)
|
|
ORDER BY created_at DESC LIMIT 1
|
|
""",
|
|
(repository.id, *ACTIVE_JOB_STATUSES),
|
|
).fetchone()
|
|
last_job = db.execute(
|
|
"SELECT * FROM scan_jobs WHERE repository_id = ? ORDER BY created_at DESC LIMIT 1",
|
|
(repository.id,),
|
|
).fetchone()
|
|
last_scan = self._report_from_row(latest) if latest else None
|
|
active_findings = [item for item in (last_scan.findings if last_scan else []) if not item.suppressed]
|
|
severity_rank = {Severity.BLOCKER: 0, Severity.WARNING: 1, Severity.INFO: 2}
|
|
primary_finding = min(
|
|
active_findings,
|
|
key=lambda item: (severity_rank[item.severity], -item.deduction, item.category, item.code),
|
|
default=None,
|
|
)
|
|
primary_signal = (
|
|
RemediationSignal(
|
|
code=primary_finding.code,
|
|
category=primary_finding.category,
|
|
severity=primary_finding.severity,
|
|
message=primary_finding.message,
|
|
count=sum(1 for item in active_findings if item.code == primary_finding.code),
|
|
)
|
|
if primary_finding
|
|
else None
|
|
)
|
|
previous_row = (
|
|
db.execute(
|
|
"""
|
|
SELECT * FROM scans
|
|
WHERE repository_id = ? AND id != ?
|
|
ORDER BY scanned_at DESC, rowid DESC LIMIT 1
|
|
""",
|
|
(repository.id, last_scan.id),
|
|
).fetchone()
|
|
if last_scan
|
|
else None
|
|
)
|
|
previous_scan = self._report_from_row(previous_row) if previous_row else None
|
|
previous_findings = [item for item in (previous_scan.findings if previous_scan else []) if not item.suppressed]
|
|
active_fingerprints = {self._finding_key(item) for item in active_findings}
|
|
previous_fingerprints = {self._finding_key(item) for item in previous_findings}
|
|
last_revision = (
|
|
last_scan.source.actual_revision
|
|
if last_scan and last_scan.source.actual_revision
|
|
else (latest["actual_revision"] if latest else None)
|
|
)
|
|
stale = bool(repository.latest_revision and last_revision and repository.latest_revision != last_revision)
|
|
if repository.archived:
|
|
state = RepositoryState.ARCHIVED
|
|
elif not repository.available:
|
|
state = RepositoryState.UNAVAILABLE
|
|
elif active:
|
|
state = (
|
|
RepositoryState.QUEUED
|
|
if active["status"] == ScanJobStatus.QUEUED.value
|
|
else RepositoryState.SCANNING
|
|
)
|
|
elif last_job and last_job["status"] == ScanJobStatus.FAILED.value:
|
|
last_job_at = _datetime(last_job["created_at"])
|
|
if last_scan is None or (last_job_at and last_job_at > last_scan.scanned_at):
|
|
state = RepositoryState.FAILED
|
|
elif stale:
|
|
state = RepositoryState.STALE
|
|
else:
|
|
state = RepositoryState.CURRENT
|
|
elif last_scan is None:
|
|
state = RepositoryState.NEVER_SCANNED
|
|
elif stale:
|
|
state = RepositoryState.STALE
|
|
else:
|
|
state = RepositoryState.CURRENT
|
|
data = repository.model_dump()
|
|
data.update(
|
|
{
|
|
"state": state,
|
|
"stale": stale,
|
|
"active_job_id": active["id"] if active else None,
|
|
"last_scan_id": last_scan.id if last_scan else None,
|
|
"last_scan_at": last_scan.scanned_at if last_scan else None,
|
|
"last_scan_health": last_scan.health_score if last_scan else None,
|
|
"last_scan_readiness": last_scan.release_gate.status if last_scan and last_scan.release_gate else (last_scan.release_readiness if last_scan else None),
|
|
"last_scan_revision": last_revision,
|
|
"last_scan_blockers": sum(1 for item in active_findings if item.severity == Severity.BLOCKER),
|
|
"last_scan_warnings": sum(1 for item in active_findings if item.severity == Severity.WARNING),
|
|
"last_scan_primary_signal": primary_signal,
|
|
"last_scan_health_delta": (last_scan.health_score - previous_scan.health_score) if previous_scan and last_scan else None,
|
|
"last_scan_added_findings": len(active_fingerprints - previous_fingerprints) if previous_scan else 0,
|
|
"last_scan_resolved_findings": len(previous_fingerprints - active_fingerprints) if previous_scan else 0,
|
|
}
|
|
)
|
|
return RepositorySummary(**data)
|
|
|
|
def _catalog_rows(
|
|
self, db: sqlite3.Connection, *, limit: int, offset: int = 0, repository_id: str | None = None
|
|
) -> list[sqlite3.Row]:
|
|
where = "WHERE r.id = ?" if repository_id else ""
|
|
params: tuple[object, ...] = (repository_id, limit, offset) if repository_id else (limit, offset)
|
|
return db.execute(
|
|
f"""
|
|
WITH ranked_scans AS (
|
|
SELECT s.*, ROW_NUMBER() OVER (
|
|
PARTITION BY s.repository_id ORDER BY s.scanned_at DESC, s.rowid DESC
|
|
) AS scan_rank
|
|
FROM scans AS s WHERE s.repository_id IS NOT NULL
|
|
), active_jobs AS (
|
|
SELECT j.id, j.repository_id, j.status, ROW_NUMBER() OVER (
|
|
PARTITION BY j.repository_id ORDER BY j.created_at DESC, j.rowid DESC
|
|
) AS job_rank
|
|
FROM scan_jobs AS j
|
|
WHERE j.status IN ('queued', 'acquiring_source', 'scanning', 'analyzing', 'persisting')
|
|
), last_jobs AS (
|
|
SELECT j.repository_id, j.status, j.created_at, ROW_NUMBER() OVER (
|
|
PARTITION BY j.repository_id ORDER BY j.created_at DESC, j.rowid DESC
|
|
) AS job_rank
|
|
FROM scan_jobs AS j
|
|
)
|
|
SELECT r.*,
|
|
latest.id AS summary_scan_id, latest.scanned_at AS summary_scanned_at,
|
|
latest.health_score AS summary_health, latest.release_readiness AS summary_readiness,
|
|
latest.actual_revision AS summary_revision, latest.finding_count AS summary_findings,
|
|
latest.blocker_count AS summary_blockers, latest.warning_count AS summary_warnings,
|
|
latest.primary_signal_json AS summary_primary_signal,
|
|
latest.added_findings AS summary_added, latest.resolved_findings AS summary_resolved,
|
|
latest.triage_priority AS summary_priority, latest.triage_rationale AS summary_rationale,
|
|
previous.health_score AS previous_health,
|
|
active.id AS active_job_id, active.status AS active_job_status,
|
|
last_job.status AS last_job_status, last_job.created_at AS last_job_created_at
|
|
FROM repositories AS r
|
|
LEFT JOIN ranked_scans AS latest ON latest.repository_id = r.id AND latest.scan_rank = 1
|
|
LEFT JOIN ranked_scans AS previous ON previous.repository_id = r.id AND previous.scan_rank = 2
|
|
LEFT JOIN active_jobs AS active ON active.repository_id = r.id AND active.job_rank = 1
|
|
LEFT JOIN last_jobs AS last_job ON last_job.repository_id = r.id AND last_job.job_rank = 1
|
|
{where}
|
|
ORDER BY r.archived ASC, r.full_name COLLATE NOCASE ASC
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
params,
|
|
).fetchall()
|
|
|
|
def _compact_repository_summary(self, row: sqlite3.Row) -> RepositorySummary:
|
|
repository = self._repository_from_row(row)
|
|
primary_signal = None
|
|
if row["summary_primary_signal"]:
|
|
try:
|
|
primary_signal = RemediationSignal.model_validate_json(row["summary_primary_signal"])
|
|
except (TypeError, ValueError):
|
|
primary_signal = None
|
|
last_revision = row["summary_revision"]
|
|
stale = bool(repository.latest_revision and last_revision and repository.latest_revision != last_revision)
|
|
if repository.archived:
|
|
state = RepositoryState.ARCHIVED
|
|
elif not repository.available:
|
|
state = RepositoryState.UNAVAILABLE
|
|
elif row["active_job_id"]:
|
|
state = RepositoryState.QUEUED if row["active_job_status"] == ScanJobStatus.QUEUED.value else RepositoryState.SCANNING
|
|
elif row["last_job_status"] == ScanJobStatus.FAILED.value:
|
|
last_job_at = _datetime(row["last_job_created_at"])
|
|
last_scan_at = _datetime(row["summary_scanned_at"])
|
|
if last_scan_at is None or (last_job_at and last_job_at > last_scan_at):
|
|
state = RepositoryState.FAILED
|
|
else:
|
|
state = RepositoryState.STALE if stale else RepositoryState.CURRENT
|
|
elif row["summary_scan_id"] is None:
|
|
state = RepositoryState.NEVER_SCANNED
|
|
else:
|
|
state = RepositoryState.STALE if stale else RepositoryState.CURRENT
|
|
data = repository.model_dump()
|
|
data.update(
|
|
{
|
|
"state": state,
|
|
"stale": stale,
|
|
"active_job_id": row["active_job_id"],
|
|
"last_scan_id": row["summary_scan_id"],
|
|
"last_scan_at": _datetime(row["summary_scanned_at"]),
|
|
"last_scan_health": row["summary_health"],
|
|
"last_scan_readiness": row["summary_readiness"],
|
|
"last_scan_revision": last_revision,
|
|
"last_scan_blockers": int(row["summary_blockers"] or 0),
|
|
"last_scan_warnings": int(row["summary_warnings"] or 0),
|
|
"last_scan_primary_signal": primary_signal,
|
|
"last_scan_health_delta": (
|
|
int(row["summary_health"]) - int(row["previous_health"])
|
|
if row["summary_health"] is not None and row["previous_health"] is not None
|
|
else None
|
|
),
|
|
"last_scan_added_findings": int(row["summary_added"] or 0),
|
|
"last_scan_resolved_findings": int(row["summary_resolved"] or 0),
|
|
"last_scan_priority": int(row["summary_priority"] or 0),
|
|
"last_scan_priority_rationale": row["summary_rationale"],
|
|
}
|
|
)
|
|
return RepositorySummary(**data)
|
|
|
|
def list_repositories(self, limit: int = 100, offset: int = 0) -> list[RepositorySummary]:
|
|
with self._connection() as db:
|
|
return [self._compact_repository_summary(row) for row in self._catalog_rows(db, limit=limit, offset=offset)]
|
|
|
|
def repository_count(self) -> int:
|
|
with self._connection() as db:
|
|
return int(db.execute("SELECT COUNT(*) FROM repositories").fetchone()[0])
|
|
|
|
def repository_state_counts(self) -> dict[str, int]:
|
|
counts: Counter[str] = Counter()
|
|
for repository in self.list_repositories(limit=1000):
|
|
counts[repository.state.value] += 1
|
|
return dict(counts)
|
|
|
|
def get_repository_detail(self, repository_id: str) -> RepositoryDetail | None:
|
|
with self._connection() as db:
|
|
rows = self._catalog_rows(db, repository_id=repository_id, limit=1)
|
|
if not rows:
|
|
return None
|
|
summary = self._compact_repository_summary(rows[0])
|
|
latest = db.execute(
|
|
"SELECT * FROM scans WHERE repository_id = ? ORDER BY scanned_at DESC, rowid DESC LIMIT 1",
|
|
(repository_id,),
|
|
).fetchone()
|
|
return RepositoryDetail(**summary.model_dump(), latest_scan=self._report_from_row(latest) if latest else None)
|
|
|
|
def repository_scans(self, repository_id: str, limit: int = 50) -> list[ScanSummary]:
|
|
with self._connection() as db:
|
|
rows = db.execute(
|
|
"SELECT * FROM scans WHERE repository_id = ? ORDER BY scanned_at DESC, rowid DESC LIMIT ?",
|
|
(repository_id, limit),
|
|
).fetchall()
|
|
result: list[ScanSummary] = []
|
|
for row in rows:
|
|
report = self._report_from_row(row)
|
|
result.append(
|
|
ScanSummary(
|
|
id=report.id,
|
|
repository_id=repository_id,
|
|
scanned_at=report.scanned_at,
|
|
branch=report.source.branch,
|
|
requested_revision=report.source.requested_revision,
|
|
actual_revision=report.source.actual_revision,
|
|
trigger=report.source.trigger,
|
|
health_score=report.health_score,
|
|
release_readiness=report.release_readiness,
|
|
findings=len(report.findings),
|
|
duration_ms=report.duration_ms,
|
|
)
|
|
)
|
|
return result
|
|
|
|
def _report_from_row(self, row: sqlite3.Row | None) -> ScanReport:
|
|
if row is None:
|
|
raise ValueError("Scan row is required")
|
|
report = ScanReport.model_validate_json(row["report_json"])
|
|
source_updates: dict[str, object] = {}
|
|
if row["repository_id"] and not report.source.repository_id:
|
|
source_updates["repository_id"] = row["repository_id"]
|
|
if row["source_origin"] and report.source.origin == SourceOrigin.LOCAL:
|
|
try:
|
|
source_updates["origin"] = SourceOrigin(str(row["source_origin"]))
|
|
except ValueError:
|
|
pass
|
|
if row["provider"] and not report.source.provider:
|
|
source_updates["provider"] = row["provider"]
|
|
for field in ("branch", "requested_revision", "actual_revision", "trigger"):
|
|
value = row[field]
|
|
if value is not None and getattr(report.source, field) in (None, ScanTrigger.MANUAL):
|
|
if field == "trigger":
|
|
try:
|
|
source_updates[field] = ScanTrigger(str(value))
|
|
except ValueError:
|
|
continue
|
|
else:
|
|
source_updates[field] = value
|
|
if source_updates:
|
|
report = report.model_copy(update={"source": report.source.model_copy(update=source_updates)})
|
|
return report.model_copy(
|
|
update={
|
|
"findings": enrich_findings(report.findings),
|
|
"compacted": bool(row["report_compacted"]) if "report_compacted" in row else report.compacted,
|
|
}
|
|
)
|
|
|
|
def _normalise_report(self, report: ScanReport) -> ScanReport:
|
|
if report.source.repository_id:
|
|
return report
|
|
repository_id = local_repository_id(report.repository_path)
|
|
source = report.source.model_copy(
|
|
update={
|
|
"origin": SourceOrigin.LOCAL,
|
|
"provider": "local",
|
|
"repository_id": repository_id,
|
|
"repository_full_name": f"local/{report.repository_name}",
|
|
"trigger": ScanTrigger.LOCAL,
|
|
}
|
|
)
|
|
return report.model_copy(update={"source": source})
|
|
|
|
def save(self, report: ScanReport) -> ScanReport:
|
|
report = self._normalise_report(report)
|
|
report = report.model_copy(update={"findings": enrich_findings(report.findings)})
|
|
source = report.source
|
|
history_key = source.repository_id or report.repository_path
|
|
with self._connection() as db:
|
|
if source.origin == SourceOrigin.LOCAL and source.repository_id:
|
|
self._upsert_local_repository_row(db, source.repository_id, report.repository_name, report.repository_path)
|
|
previous_row = db.execute(
|
|
"""
|
|
SELECT * FROM scans WHERE history_key = ? ORDER BY scanned_at DESC, rowid DESC LIMIT 1
|
|
""",
|
|
(history_key,),
|
|
).fetchone()
|
|
previous = self._report_from_row(previous_row) if previous_row else None
|
|
summary = self._summary_fields(report, previous)
|
|
db.execute(
|
|
"""
|
|
INSERT INTO scans (
|
|
id, repository_path, repository_name, scanned_at, health_score, report_json,
|
|
repository_id, history_key, source_origin, provider, branch,
|
|
requested_revision, actual_revision, trigger, finding_count, blocker_count,
|
|
warning_count, release_readiness, primary_signal_json, added_findings,
|
|
resolved_findings, triage_priority, triage_rationale, report_compacted
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
report.id,
|
|
report.repository_path,
|
|
report.repository_name,
|
|
report.scanned_at.isoformat(),
|
|
report.health_score,
|
|
report.model_dump_json(),
|
|
source.repository_id,
|
|
history_key,
|
|
source.origin.value,
|
|
source.provider,
|
|
source.branch,
|
|
source.requested_revision,
|
|
source.actual_revision,
|
|
source.trigger.value,
|
|
summary["finding_count"],
|
|
summary["blocker_count"],
|
|
summary["warning_count"],
|
|
summary["release_readiness"],
|
|
summary["primary_signal_json"],
|
|
summary["added_findings"],
|
|
summary["resolved_findings"],
|
|
summary["triage_priority"],
|
|
summary["triage_rationale"],
|
|
0,
|
|
),
|
|
)
|
|
self._apply_retention(db, history_key)
|
|
if source.repository_id:
|
|
db.execute(
|
|
"""
|
|
UPDATE repositories
|
|
SET latest_revision = COALESCE(?, latest_revision),
|
|
latest_revision_at = COALESCE(latest_revision_at, ?),
|
|
sync_error = NULL
|
|
WHERE id = ? AND provider = 'local'
|
|
""",
|
|
(source.actual_revision, _iso(report.scanned_at), source.repository_id),
|
|
)
|
|
return report
|
|
|
|
def _apply_retention(self, db: sqlite3.Connection, history_key: str) -> None:
|
|
db.execute(
|
|
"""
|
|
DELETE FROM scans
|
|
WHERE history_key = ?
|
|
AND rowid NOT IN (
|
|
SELECT rowid FROM scans WHERE history_key = ?
|
|
ORDER BY scanned_at DESC, rowid DESC LIMIT ?
|
|
)
|
|
""",
|
|
(history_key, history_key, history_per_repository()),
|
|
)
|
|
cutoff = (_now() - timedelta(days=report_retention_days())).isoformat()
|
|
db.execute("DELETE FROM scans WHERE history_key = ? AND scanned_at < ?", (history_key, cutoff))
|
|
rows = db.execute(
|
|
"""
|
|
SELECT * FROM scans WHERE history_key = ? AND report_compacted = 0
|
|
ORDER BY scanned_at DESC, rowid DESC LIMIT -1 OFFSET ?
|
|
""",
|
|
(history_key, full_reports_per_repository()),
|
|
).fetchall()
|
|
for row in rows:
|
|
report = self._report_from_row(row)
|
|
compact = report.model_copy(
|
|
update={"findings": [], "architecture": ArchitectureGraph(), "hotspots": [], "compacted": True}
|
|
)
|
|
db.execute(
|
|
"UPDATE scans SET report_json = ?, report_compacted = 1 WHERE id = ?",
|
|
(compact.model_dump_json(), row["id"]),
|
|
)
|
|
|
|
def get(self, scan_id: str) -> ScanReport | None:
|
|
with self._connection() as db:
|
|
row = db.execute("SELECT * FROM scans WHERE id = ?", (scan_id,)).fetchone()
|
|
return self._report_from_row(row) if row else None
|
|
|
|
def recent(self, limit: int = 20) -> list[ScanReport]:
|
|
with self._connection() as db:
|
|
rows = db.execute(
|
|
"SELECT * FROM scans ORDER BY scanned_at DESC, rowid DESC LIMIT ?",
|
|
(limit,),
|
|
).fetchall()
|
|
return [self._report_from_row(row) for row in rows]
|
|
|
|
def _history_candidates(self, repository_key: str) -> list[str]:
|
|
if repository_key.startswith(("local:", "gitea:")):
|
|
return [repository_key]
|
|
return [repository_key, local_repository_id(repository_key)]
|
|
|
|
def recent_for_repository(self, repository_key: str, limit: int = 50) -> list[ScanReport]:
|
|
with self._connection() as db:
|
|
for candidate in self._history_candidates(repository_key):
|
|
rows = db.execute(
|
|
"SELECT * FROM scans WHERE history_key = ? ORDER BY scanned_at DESC, rowid DESC LIMIT ?",
|
|
(candidate, limit),
|
|
).fetchall()
|
|
if rows:
|
|
return [self._report_from_row(row) for row in rows]
|
|
return []
|
|
|
|
def previous_for_repository(self, repository_key: str, exclude_id: str) -> ScanReport | None:
|
|
with self._connection() as db:
|
|
for candidate in self._history_candidates(repository_key):
|
|
row = db.execute(
|
|
"""
|
|
SELECT previous.*
|
|
FROM scans AS previous
|
|
JOIN scans AS current ON current.id = ?
|
|
WHERE previous.history_key = ?
|
|
AND previous.id != current.id
|
|
AND (
|
|
previous.scanned_at < current.scanned_at
|
|
OR (previous.scanned_at = current.scanned_at AND previous.rowid < current.rowid)
|
|
)
|
|
ORDER BY previous.scanned_at DESC, previous.rowid DESC
|
|
LIMIT 1
|
|
""",
|
|
(exclude_id, candidate),
|
|
).fetchone()
|
|
if row:
|
|
return self._report_from_row(row)
|
|
return None
|
|
|
|
@staticmethod
|
|
def _finding_key(report_finding) -> str:
|
|
return report_finding.fingerprint or finding_fingerprint(report_finding)
|
|
|
|
def compare(self, report: ScanReport) -> ScanComparison:
|
|
previous = self.previous_for_repository(
|
|
report.source.repository_id or report.repository_path,
|
|
report.id,
|
|
)
|
|
if previous is None:
|
|
return ScanComparison(current_scan_id=report.id)
|
|
|
|
current_metrics = report.metrics.model_dump()
|
|
previous_metrics = previous.metrics.model_dump()
|
|
metric_deltas = {
|
|
key: int(current_metrics[key]) - int(previous_metrics.get(key, 0))
|
|
for key in current_metrics
|
|
}
|
|
category_deltas = {
|
|
category: report.category_scores.get(category, 0) - previous.category_scores.get(category, 0)
|
|
for category in set(report.category_scores) | set(previous.category_scores)
|
|
}
|
|
current_findings = {self._finding_key(item): item for item in report.findings}
|
|
previous_findings = {self._finding_key(item): item for item in previous.findings}
|
|
added = [current_findings[key] for key in current_findings.keys() - previous_findings.keys()]
|
|
resolved = [previous_findings[key] for key in previous_findings.keys() - current_findings.keys()]
|
|
|
|
return ScanComparison(
|
|
current_scan_id=report.id,
|
|
previous_scan_id=previous.id,
|
|
health_score_delta=report.health_score - previous.health_score,
|
|
category_score_deltas=dict(sorted(category_deltas.items())),
|
|
metric_deltas=metric_deltas,
|
|
added_findings=sorted(added, key=lambda item: (item.category, item.path or "", item.line or 0)),
|
|
resolved_findings=sorted(resolved, key=lambda item: (item.category, item.path or "", item.line or 0)),
|
|
)
|
|
|
|
def trend(self, repository_key: str, limit: int = 50) -> list[TrendPoint]:
|
|
reports = list(reversed(self.recent_for_repository(repository_key, limit)))
|
|
points: list[TrendPoint] = []
|
|
for report in reports:
|
|
blockers = sum(1 for item in report.findings if item.severity == Severity.BLOCKER)
|
|
warnings = sum(1 for item in report.findings if item.severity == Severity.WARNING)
|
|
debt = report.metrics.todo + report.metrics.fixme + report.metrics.hack + report.metrics.xxx
|
|
points.append(
|
|
TrendPoint(
|
|
scan_id=report.id,
|
|
scanned_at=report.scanned_at,
|
|
health_score=report.health_score,
|
|
findings=len(report.findings),
|
|
blockers=blockers,
|
|
warnings=warnings,
|
|
debt_markers=debt,
|
|
)
|
|
)
|
|
return points
|
|
|
|
@staticmethod
|
|
def _job_from_row(row: sqlite3.Row) -> ScanJob:
|
|
return ScanJob(
|
|
id=row["id"],
|
|
repository_id=row["repository_id"],
|
|
status=row["status"],
|
|
trigger=row["trigger"],
|
|
requested_revision=row["requested_revision"],
|
|
actual_revision=row["actual_revision"],
|
|
phase=row["phase"],
|
|
progress=int(row["progress"]),
|
|
message=row["message"],
|
|
created_at=_datetime(row["created_at"]) or _now(),
|
|
started_at=_datetime(row["started_at"]),
|
|
updated_at=_datetime(row["updated_at"]) or _now(),
|
|
finished_at=_datetime(row["finished_at"]),
|
|
duration_ms=float(row["duration_ms"]) if row["duration_ms"] is not None else None,
|
|
scan_id=row["scan_id"],
|
|
error_code=row["error_code"],
|
|
error_detail=row["error_detail"],
|
|
attempt=int(row["attempt"]),
|
|
deduplicated=bool(row["deduplicated"]),
|
|
)
|
|
|
|
def _recover_expired_jobs(self, db: sqlite3.Connection, now: str) -> None:
|
|
placeholders = ",".join("?" for _ in ACTIVE_JOB_STATUSES[1:])
|
|
db.execute(
|
|
f"""
|
|
UPDATE scan_jobs
|
|
SET status = 'queued', phase = 'queued', progress = 0,
|
|
message = 'Recovered after worker lease expiry', updated_at = ?, lease_until = NULL
|
|
WHERE status IN ({placeholders}) AND lease_until IS NOT NULL AND lease_until < ?
|
|
""",
|
|
(now, *ACTIVE_JOB_STATUSES[1:], now),
|
|
)
|
|
|
|
def enqueue_job(
|
|
self,
|
|
repository_id: str,
|
|
requested_revision: str | None,
|
|
trigger: ScanTrigger,
|
|
*,
|
|
force: bool = False,
|
|
max_queue: int = 100,
|
|
) -> ScanJob:
|
|
now_dt = _now()
|
|
now = now_dt.isoformat()
|
|
with self._connection() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
if self._maintenance_enabled(db):
|
|
raise MaintenanceModeError("RepoPulse is in maintenance mode")
|
|
self._recover_expired_jobs(db, now)
|
|
repository = db.execute("SELECT id FROM repositories WHERE id = ?", (repository_id,)).fetchone()
|
|
if repository is None:
|
|
raise KeyError(repository_id)
|
|
active_rows = db.execute(
|
|
"""
|
|
SELECT * FROM scan_jobs
|
|
WHERE repository_id = ? AND status IN (?, ?, ?, ?, ?)
|
|
ORDER BY created_at DESC
|
|
""",
|
|
(repository_id, *ACTIVE_JOB_STATUSES),
|
|
).fetchall()
|
|
if not force:
|
|
for row in active_rows:
|
|
if row["requested_revision"] == requested_revision:
|
|
return self._job_from_row(row).model_copy(update={"deduplicated": True})
|
|
if requested_revision and not force:
|
|
completed = db.execute(
|
|
"""
|
|
SELECT * FROM scans
|
|
WHERE repository_id = ? AND actual_revision = ?
|
|
ORDER BY scanned_at DESC, rowid DESC LIMIT 1
|
|
""",
|
|
(repository_id, requested_revision),
|
|
).fetchone()
|
|
if completed:
|
|
existing = db.execute(
|
|
"""
|
|
SELECT * FROM scan_jobs WHERE repository_id = ? AND scan_id = ?
|
|
ORDER BY created_at DESC LIMIT 1
|
|
""",
|
|
(repository_id, completed["id"]),
|
|
).fetchone()
|
|
if existing:
|
|
return self._job_from_row(existing).model_copy(update={"deduplicated": True})
|
|
job_id = str(uuid4())
|
|
db.execute(
|
|
"""
|
|
INSERT INTO scan_jobs (
|
|
id, repository_id, status, trigger, requested_revision, actual_revision,
|
|
phase, progress, message, created_at, updated_at, finished_at, scan_id,
|
|
deduplicated
|
|
) VALUES (?, ?, 'completed', ?, ?, ?, 'completed', 100,
|
|
'Existing revision already analyzed', ?, ?, ?, ?, 1)
|
|
""",
|
|
(
|
|
job_id,
|
|
repository_id,
|
|
trigger.value,
|
|
requested_revision,
|
|
completed["actual_revision"],
|
|
now,
|
|
now,
|
|
completed["scanned_at"],
|
|
completed["id"],
|
|
),
|
|
)
|
|
row = db.execute("SELECT * FROM scan_jobs WHERE id = ?", (job_id,)).fetchone()
|
|
return self._job_from_row(row)
|
|
queued = db.execute(
|
|
"SELECT COUNT(*) FROM scan_jobs WHERE status = 'queued'"
|
|
).fetchone()[0]
|
|
if int(queued) >= max(1, max_queue):
|
|
raise QueueFullError("Scan queue is full; retry after queued work completes")
|
|
if requested_revision:
|
|
for row in active_rows:
|
|
if row["status"] == ScanJobStatus.QUEUED.value and row["requested_revision"] != requested_revision:
|
|
db.execute(
|
|
"""
|
|
UPDATE scan_jobs
|
|
SET status = 'superseded', phase = 'superseded', progress = 100,
|
|
message = 'Superseded by a newer repository revision',
|
|
updated_at = ?, finished_at = ?
|
|
WHERE id = ?
|
|
""",
|
|
(now, now, row["id"]),
|
|
)
|
|
job_id = str(uuid4())
|
|
db.execute(
|
|
"""
|
|
INSERT INTO scan_jobs (
|
|
id, repository_id, status, trigger, requested_revision, phase,
|
|
progress, message, created_at, updated_at
|
|
) VALUES (?, ?, 'queued', ?, ?, 'queued', 0, 'Waiting for an available worker', ?, ?)
|
|
""",
|
|
(job_id, repository_id, trigger.value, requested_revision, now, now),
|
|
)
|
|
row = db.execute("SELECT * FROM scan_jobs WHERE id = ?", (job_id,)).fetchone()
|
|
return self._job_from_row(row)
|
|
|
|
def get_job(self, job_id: str) -> ScanJob | None:
|
|
with self._connection() as db:
|
|
row = db.execute("SELECT * FROM scan_jobs WHERE id = ?", (job_id,)).fetchone()
|
|
return self._job_from_row(row) if row else None
|
|
|
|
def claim_next_job(self) -> ScanJob | None:
|
|
now_dt = _now()
|
|
now = now_dt.isoformat()
|
|
lease_until = (now_dt + timedelta(seconds=job_lease_seconds())).isoformat()
|
|
with self._connection() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
if self._maintenance_enabled(db):
|
|
return None
|
|
self._recover_expired_jobs(db, now)
|
|
row = db.execute(
|
|
"""
|
|
SELECT queued.* FROM scan_jobs AS queued
|
|
WHERE queued.status = 'queued'
|
|
ORDER BY COALESCE((
|
|
SELECT MAX(served.started_at) FROM scan_jobs AS served
|
|
WHERE served.repository_id = queued.repository_id AND served.started_at IS NOT NULL
|
|
), ''), queued.created_at, queued.rowid
|
|
LIMIT 1
|
|
"""
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
db.execute(
|
|
"""
|
|
UPDATE scan_jobs
|
|
SET status = 'acquiring_source', phase = 'acquiring_source', progress = 5,
|
|
message = 'Acquiring exact repository revision', started_at = COALESCE(started_at, ?),
|
|
updated_at = ?, attempt = attempt + 1, lease_until = ?
|
|
WHERE id = ? AND status = 'queued'
|
|
""",
|
|
(now, now, lease_until, row["id"]),
|
|
)
|
|
claimed = db.execute("SELECT * FROM scan_jobs WHERE id = ?", (row["id"],)).fetchone()
|
|
return self._job_from_row(claimed)
|
|
|
|
def renew_job_lease(self, job_id: str) -> ScanJob | None:
|
|
"""Renew only an actively owned lease; terminal jobs are never resurrected."""
|
|
|
|
lease_until = (_now() + timedelta(seconds=job_lease_seconds())).isoformat()
|
|
placeholders = ",".join("?" for _ in ACTIVE_JOB_STATUSES[1:])
|
|
with self._connection() as db:
|
|
db.execute(
|
|
f"""
|
|
UPDATE scan_jobs SET lease_until = ?, updated_at = ?
|
|
WHERE id = ? AND status IN ({placeholders})
|
|
""",
|
|
(lease_until, _now().isoformat(), job_id, *ACTIVE_JOB_STATUSES[1:]),
|
|
)
|
|
row = db.execute("SELECT * FROM scan_jobs WHERE id = ?", (job_id,)).fetchone()
|
|
return self._job_from_row(row) if row else None
|
|
|
|
def update_job(self, job_id: str, **updates: object) -> ScanJob | None:
|
|
allowed = {
|
|
"status",
|
|
"actual_revision",
|
|
"phase",
|
|
"progress",
|
|
"message",
|
|
"started_at",
|
|
"finished_at",
|
|
"duration_ms",
|
|
"scan_id",
|
|
"error_code",
|
|
"error_detail",
|
|
"lease_until",
|
|
}
|
|
invalid = set(updates) - allowed
|
|
if invalid:
|
|
raise ValueError(f"Unsupported job fields: {sorted(invalid)}")
|
|
if not updates:
|
|
return self.get_job(job_id)
|
|
serialised: dict[str, object] = {}
|
|
for key, value in updates.items():
|
|
if isinstance(value, datetime):
|
|
serialised[key] = _iso(value)
|
|
elif isinstance(value, (ScanJobStatus, ScanTrigger)):
|
|
serialised[key] = value.value
|
|
else:
|
|
serialised[key] = value
|
|
serialised["updated_at"] = _now().isoformat()
|
|
if serialised.get("status") in ACTIVE_JOB_STATUSES[1:] and "lease_until" not in serialised:
|
|
serialised["lease_until"] = (_now() + timedelta(seconds=job_lease_seconds())).isoformat()
|
|
if "status" in serialised and serialised["status"] in {
|
|
ScanJobStatus.COMPLETED.value,
|
|
ScanJobStatus.FAILED.value,
|
|
ScanJobStatus.CANCELLED.value,
|
|
ScanJobStatus.SUPERSEDED.value,
|
|
}:
|
|
with self._connection() as db:
|
|
existing = db.execute(
|
|
"SELECT started_at FROM scan_jobs WHERE id = ?", (job_id,)
|
|
).fetchone()
|
|
if "duration_ms" not in serialised and existing and existing["started_at"]:
|
|
started = _datetime(existing["started_at"])
|
|
if started:
|
|
serialised["duration_ms"] = max(0.0, (_now() - started).total_seconds() * 1000)
|
|
serialised.setdefault("finished_at", _now().isoformat())
|
|
serialised.setdefault("lease_until", None)
|
|
columns = ", ".join(f"{key} = ?" for key in serialised)
|
|
with self._connection() as db:
|
|
db.execute(
|
|
f"UPDATE scan_jobs SET {columns} WHERE id = ?",
|
|
(*serialised.values(), job_id),
|
|
)
|
|
row = db.execute("SELECT * FROM scan_jobs WHERE id = ?", (job_id,)).fetchone()
|
|
return self._job_from_row(row) if row else None
|
|
|
|
@staticmethod
|
|
def _maintenance_enabled(db: sqlite3.Connection) -> bool:
|
|
row = db.execute("SELECT value FROM app_state WHERE key = 'maintenance_mode'").fetchone()
|
|
if row is None:
|
|
return False
|
|
try:
|
|
value = json.loads(row["value"])
|
|
except (TypeError, json.JSONDecodeError):
|
|
# Fail closed if persisted control state cannot be interpreted.
|
|
return True
|
|
if isinstance(value, dict) and isinstance(value.get("enabled"), bool):
|
|
return value["enabled"]
|
|
return value is not False
|
|
|
|
def maintenance_status(self) -> dict[str, object]:
|
|
with self._connection() as db:
|
|
row = db.execute(
|
|
"SELECT value FROM app_state WHERE key = 'maintenance_mode'"
|
|
).fetchone()
|
|
if row is None:
|
|
return {"enabled": False, "changed_at": None}
|
|
try:
|
|
value = json.loads(row["value"])
|
|
except (TypeError, json.JSONDecodeError):
|
|
return {"enabled": True, "changed_at": None, "state_error": "invalid_persisted_state"}
|
|
if isinstance(value, dict) and isinstance(value.get("enabled"), bool):
|
|
return {
|
|
"enabled": value["enabled"],
|
|
"changed_at": value.get("changed_at"),
|
|
}
|
|
if isinstance(value, bool):
|
|
return {"enabled": value, "changed_at": None}
|
|
return {"enabled": True, "changed_at": None, "state_error": "invalid_persisted_state"}
|
|
|
|
def set_maintenance_mode(self, enabled: bool) -> dict[str, object]:
|
|
changed_at = _now().isoformat()
|
|
value = json.dumps({"enabled": enabled, "changed_at": changed_at}, separators=(",", ":"))
|
|
with self._connection() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
db.execute(
|
|
"""
|
|
INSERT INTO app_state (key, value, updated_at) VALUES ('maintenance_mode', ?, ?)
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at
|
|
""",
|
|
(value, changed_at),
|
|
)
|
|
return {"enabled": enabled, "changed_at": changed_at}
|
|
|
|
def touch_worker_heartbeat_if_active(self, heartbeat: str) -> bool:
|
|
"""Persist scheduler liveness only when maintenance is not active."""
|
|
|
|
with self._connection() as db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
if self._maintenance_enabled(db):
|
|
return False
|
|
db.execute(
|
|
"""
|
|
INSERT INTO app_state (key, value, updated_at) VALUES ('worker_heartbeat', ?, ?)
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at
|
|
""",
|
|
(heartbeat, _now().isoformat()),
|
|
)
|
|
return True
|
|
|
|
def queue_stats(self) -> dict[str, int | str | None]:
|
|
with self._connection() as db:
|
|
rows = db.execute("SELECT status, COUNT(*) AS count FROM scan_jobs GROUP BY status").fetchall()
|
|
counts = {str(row["status"]): int(row["count"]) for row in rows}
|
|
heartbeat = db.execute("SELECT value FROM app_state WHERE key = 'worker_heartbeat'").fetchone()
|
|
sync = db.execute("SELECT value FROM app_state WHERE key = 'last_gitea_sync'").fetchone()
|
|
return {
|
|
"queued": counts.get(ScanJobStatus.QUEUED.value, 0),
|
|
"active": sum(counts.get(status, 0) for status in ACTIVE_JOB_STATUSES[1:]),
|
|
"completed": counts.get(ScanJobStatus.COMPLETED.value, 0),
|
|
"failed": counts.get(ScanJobStatus.FAILED.value, 0),
|
|
"worker_heartbeat": heartbeat["value"] if heartbeat else None,
|
|
"last_gitea_sync": sync["value"] if sync else None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _tree_stats(root: Path) -> tuple[int, int]:
|
|
"""Return bounded managed-cache file count and bytes without following links."""
|
|
|
|
if not root.exists() or root.is_symlink():
|
|
return 0, 0
|
|
files = 0
|
|
total_bytes = 0
|
|
pending = [root]
|
|
while pending and files < 50_000:
|
|
current = pending.pop()
|
|
try:
|
|
entries = list(current.iterdir())
|
|
except OSError:
|
|
continue
|
|
for entry in entries:
|
|
if entry.is_symlink():
|
|
continue
|
|
try:
|
|
if entry.is_dir():
|
|
pending.append(entry)
|
|
elif entry.is_file():
|
|
files += 1
|
|
total_bytes += entry.stat().st_size
|
|
except OSError:
|
|
continue
|
|
if files >= 50_000:
|
|
break
|
|
return files, total_bytes
|
|
|
|
@staticmethod
|
|
def _managed_entry_count(root: Path) -> int:
|
|
try:
|
|
return sum(1 for entry in root.iterdir() if not entry.is_symlink())
|
|
except OSError:
|
|
return 0
|
|
|
|
def storage_stats(self) -> dict[str, int | str | None]:
|
|
with self._connection() as db:
|
|
row = db.execute(
|
|
"""
|
|
SELECT COUNT(*) AS scan_count,
|
|
SUM(CASE WHEN report_compacted = 1 THEN 1 ELSE 0 END) AS compacted_scan_count,
|
|
MIN(scanned_at) AS oldest_scan_at, MAX(scanned_at) AS newest_scan_at
|
|
FROM scans
|
|
"""
|
|
).fetchone()
|
|
migration_backup = db.execute(
|
|
"SELECT value FROM app_state WHERE key = 'last_schema_migration_backup'"
|
|
).fetchone()
|
|
database_bytes = self.path.stat().st_size if self.path.exists() else 0
|
|
journal = Path(f"{self.path}-wal")
|
|
journal_bytes = journal.stat().st_size if journal.exists() else 0
|
|
cache_root = Path(os.getenv("REPOPULSE_REPOSITORY_CACHE", "data/repos")).expanduser()
|
|
mirrors_root = cache_root / "mirrors"
|
|
worktrees_root = cache_root / "worktrees"
|
|
_mirror_files, mirror_cache_bytes = self._tree_stats(mirrors_root)
|
|
mirror_count = self._managed_entry_count(mirrors_root)
|
|
worktree_count = self._managed_entry_count(worktrees_root)
|
|
return {
|
|
"database_bytes": database_bytes,
|
|
"journal_bytes": journal_bytes,
|
|
"mirror_count": mirror_count,
|
|
"mirror_cache_bytes": mirror_cache_bytes,
|
|
"worktree_count": worktree_count,
|
|
"scan_count": int(row["scan_count"] or 0),
|
|
"compacted_scan_count": int(row["compacted_scan_count"] or 0),
|
|
"oldest_scan_at": row["oldest_scan_at"],
|
|
"newest_scan_at": row["newest_scan_at"],
|
|
"history_per_repository": history_per_repository(),
|
|
"full_reports_per_repository": full_reports_per_repository(),
|
|
"report_retention_days": report_retention_days(),
|
|
"last_schema_migration_backup": migration_backup["value"] if migration_backup else None,
|
|
}
|
|
|
|
def database_recovery_status(self) -> dict[str, object]:
|
|
stored = self.get_state("database_recovery")
|
|
if not stored:
|
|
return {"status": "healthy"}
|
|
try:
|
|
value = json.loads(stored)
|
|
except json.JSONDecodeError:
|
|
value = {"status": "unknown", "detail": "Recovery metadata could not be read"}
|
|
return value if isinstance(value, dict) else {"status": "unknown"}
|
|
|
|
def acknowledge_database_recovery(self) -> None:
|
|
self.recovery_metadata = None
|
|
self.set_state("database_recovery", {"status": "acknowledged", "acknowledged_at": _now().isoformat()})
|
|
|
|
def set_state(self, key: str, value: object) -> None:
|
|
encoded = value if isinstance(value, str) else json.dumps(value, separators=(",", ":"))
|
|
with self._connection() as db:
|
|
db.execute(
|
|
"""
|
|
INSERT INTO app_state (key, value, updated_at) VALUES (?, ?, ?)
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at
|
|
""",
|
|
(key, encoded, _now().isoformat()),
|
|
)
|
|
|
|
def get_state(self, key: str) -> str | None:
|
|
with self._connection() as db:
|
|
row = db.execute("SELECT value FROM app_state WHERE key = ?", (key,)).fetchone()
|
|
return row["value"] if row else None
|
|
|
|
def update_repository_sync(
|
|
self,
|
|
repository: Repository,
|
|
*,
|
|
synced_at: datetime,
|
|
error: str | None = None,
|
|
) -> None:
|
|
updated = repository.model_copy(
|
|
update={"last_synced_at": synced_at, "sync_error": error}
|
|
)
|
|
self.upsert_repository(updated)
|
|
|
|
def record_sync_result(self, result: RepositorySyncResult) -> None:
|
|
self.set_state("last_gitea_sync", result.model_dump_json())
|