quickstart.py

"""Turnkey local quickstart runtime."""

from __future__ import annotations

import datetime as dt
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Protocol

from security_scanner.catalog.scan_target import ScanTarget
from security_scanner.runtime.incremental_discovery import (
    DEFAULT_JOB_PRIORITY,
    DEFAULT_MAX_ATTEMPTS,
    DEFAULT_REF_PATTERNS,
    DISCOVERY_MODE_ENQUEUE,
    DISCOVERY_MODE_INITIALIZE,
    DiscoveryScannerConfig,
    GitDiscovery,
    IncrementalDiscoveryRequest,
    IncrementalDiscoverySummary,
    run_incremental_discovery,
)
from security_scanner.runtime.queue_status import QueueStatusRequest, read_queue_status
from security_scanner.runtime.scan_worker import (
    ScanWorkerRequest,
    ScanWorkerSummary,
)
from security_scanner.storage.adapters.nosql_db.items import (
    SCAN_JOB_STATUS_PENDING,
    normalize_scan_target_url,
    repo_id_for_scan_target_url,
    scan_job_id_for,
)
from security_scanner.storage.base import QueueStatus, ScanJob


class QuickstartStore(Protocol):
    """Storage capabilities needed by quickstart."""

    def bootstrap(self) -> None:
        """Create storage resources."""

    def put_scan_target(self, target: ScanTarget) -> None:
        """Persist a target."""

    def enqueue_commit_scan_job(self, job: ScanJob) -> bool:
        """Persist a job if absent."""


@dataclass(frozen=True)
class QuickstartRequest:
    """Inputs for a turnkey local scan bootstrap."""

    target_url: str
    target_name: str | None
    store: QuickstartStore
    fetch_repo: Callable[[str], Path]
    git: GitDiscovery
    scanner_config: DiscoveryScannerConfig
    worker_request_factory: Callable[[int], ScanWorkerRequest] | None = None
    run_worker: bool = True
    max_jobs: int = 10
    storage_wait_seconds: int = 30
    now_factory: Callable[[], dt.datetime] = lambda: dt.datetime.now(dt.UTC).replace(
        microsecond=0
    )


@dataclass(frozen=True)
class QuickstartSummary:
    """Operator-facing quickstart summary."""

    target_url: str
    initialized: IncrementalDiscoverySummary
    current_jobs_enqueued: int
    enqueued: IncrementalDiscoverySummary
    worker: ScanWorkerSummary | None
    status: QueueStatus


def run_quickstart(request: QuickstartRequest) -> QuickstartSummary:
    """Bootstrap storage, register a target, enqueue current tip, and process jobs."""
    _bootstrap_with_wait(request.store, request.storage_wait_seconds)

    target = ScanTarget(
        url=normalize_scan_target_url(request.target_url),
        name=request.target_name or _default_target_name(request.target_url),
        enabled=True,
    )
    request.store.put_scan_target(target)
    scoped_store = _SingleTargetStore(request.store, target)

    initialized = run_incremental_discovery(
        IncrementalDiscoveryRequest(
            mode=DISCOVERY_MODE_INITIALIZE,
            store=scoped_store,
            fetch_repo=request.fetch_repo,
            git=request.git,
            scanner=request.scanner_config,
            ref_patterns=DEFAULT_REF_PATTERNS,
            now_factory=request.now_factory,
        )
    )
    current_jobs_enqueued = _enqueue_current_tip_jobs(request, target)
    enqueued = run_incremental_discovery(
        IncrementalDiscoveryRequest(
            mode=DISCOVERY_MODE_ENQUEUE,
            store=scoped_store,
            fetch_repo=request.fetch_repo,
            git=request.git,
            scanner=request.scanner_config,
            ref_patterns=DEFAULT_REF_PATTERNS,
            now_factory=request.now_factory,
        )
    )

    worker_summary = None
    if request.run_worker:
        if request.worker_request_factory is None:
            raise ValueError("worker_request_factory is required when run_worker is true")
        worker_summary = run_worker_once(request.worker_request_factory(request.max_jobs))

    status = read_queue_status(
        QueueStatusRequest(store=request.store, now_factory=request.now_factory)
    )
    return QuickstartSummary(
        target_url=target.url,
        initialized=initialized,
        current_jobs_enqueued=current_jobs_enqueued,
        enqueued=enqueued,
        worker=worker_summary,
        status=status,
    )


def render_quickstart_summary(summary: QuickstartSummary) -> str:
    """Render a stable quickstart report."""
    worker = summary.worker or ScanWorkerSummary()
    return (
        f"target: {summary.target_url}\n"
        f"initialized refs: {summary.initialized.refs_observed}\n"
        f"current jobs enqueued: {summary.current_jobs_enqueued}\n"
        f"discovery jobs enqueued: {summary.enqueued.jobs_enqueued}\n"
        f"worker leased: {worker.leased}\n"
        f"worker completed: {worker.completed}\n"
        f"pending: {summary.status.job_counts_by_status.get('pending', 0)}\n"
        f"completed: {summary.status.job_counts_by_status.get('completed', 0)}\n"
    )


def run_worker_once(request: ScanWorkerRequest) -> ScanWorkerSummary:
    """Small indirection so tests can monkeypatch worker execution."""
    from security_scanner.runtime.scan_worker import run_scan_worker_once

    return run_scan_worker_once(request)


class _SingleTargetStore:
    """Adapter that scopes discovery to the quickstart target only."""

    def __init__(self, store: QuickstartStore, target: ScanTarget) -> None:
        self._store = store
        self._target = target

    def list_scan_targets(self) -> list[ScanTarget]:
        return [self._target]

    def __getattr__(self, name: str):
        return getattr(self._store, name)


def _enqueue_current_tip_jobs(
    request: QuickstartRequest,
    target: ScanTarget,
) -> int:
    repo_path = request.fetch_repo(target.url)
    refs = request.git.list_remote_refs(repo_path, DEFAULT_REF_PATTERNS)
    repo_id = repo_id_for_scan_target_url(target.url)
    repo_url = normalize_scan_target_url(target.url)
    enqueued = 0
    for git_ref in refs:
        job = _current_tip_job(
            repo_id=repo_id,
            repo_url=repo_url,
            ref_name=git_ref.ref_name,
            commit_sha=git_ref.commit_sha,
            scanner=request.scanner_config,
            now=_now(request),
        )
        if request.store.enqueue_commit_scan_job(job):
            enqueued += 1
    return enqueued


def _current_tip_job(
    *,
    repo_id: str,
    repo_url: str,
    ref_name: str,
    commit_sha: str,
    scanner: DiscoveryScannerConfig,
    now: dt.datetime,
) -> ScanJob:
    job_id = scan_job_id_for(
        repo_id=repo_id,
        commit_sha=commit_sha,
        scanner_name=scanner.scanner_name,
        scanner_version=scanner.scanner_version,
        rule_pack_version=scanner.rule_pack_version,
        scanner_config_hash=scanner.scanner_config_hash,
    )
    return ScanJob(
        job_id=job_id,
        repo_id=repo_id,
        repo_url=repo_url,
        ref_name=ref_name,
        old_sha=None,
        new_sha=commit_sha,
        commit_sha=commit_sha,
        commit_range=None,
        scanner_name=scanner.scanner_name,
        scanner_version=scanner.scanner_version,
        rule_pack_version=scanner.rule_pack_version,
        scanner_config_hash=scanner.scanner_config_hash,
        priority=DEFAULT_JOB_PRIORITY,
        status=SCAN_JOB_STATUS_PENDING,
        attempts=0,
        max_attempts=DEFAULT_MAX_ATTEMPTS,
        worker_id=None,
        lease_until=None,
        next_attempt_at=now,
        created_at=now,
        updated_at=now,
    )


def _bootstrap_with_wait(store: QuickstartStore, wait_seconds: int) -> None:
    deadline = time.monotonic() + max(wait_seconds, 0)
    last_error: Exception | None = None
    while True:
        try:
            store.bootstrap()
            return
        except Exception as exc:  # noqa: BLE001 - backend readiness is time-bound.
            last_error = exc
            if time.monotonic() >= deadline:
                break
            time.sleep(1)
    raise RuntimeError(
        "storage bootstrap failed; start DynamoDB Local and retry"
    ) from last_error


def _default_target_name(url: str) -> str:
    path = normalize_scan_target_url(url).rsplit("://", 1)[-1]
    segments = [s for s in path.split("/") if s]
    if len(segments) >= 3:
        return "/".join(segments[-2:])
    if len(segments) == 2:
        return segments[-1]
    return url


def _now(request: QuickstartRequest) -> dt.datetime:
    value = request.now_factory()
    if value.tzinfo is None:
        return value.replace(tzinfo=dt.UTC)
    return value.astimezone(dt.UTC).replace(microsecond=0)