platform-ci/platform_ci/delivery/completion.py

"""Maven Central publication and its terminal completion receipt, from the verified tested bundle, with the existing frozen primitives.

UNTESTED: written while test execution was suspended; nothing here has run, signed, uploaded or read a credential.

Frozen primitives (root-copied byte-identically into `publisher_frozen/`, source `references/central-producer`, live deployment not revalidated), reached only through the
`legacy` bridge: `qrun_central` (`publish_central`, `verified_purls`, `CentralPending`, `CentralDuplicate`) and `qrun_central_transport` (`CentralClient`, `GpgSigner`).
Mount names come from that source and are NOT proof that any grant exists live.

What this adds around them (the generic part of the captured `promote_reactor`):
- The Central evidence is the VERIFIED retained files of the tested bundle (`adapters.maven_modules`), so Central receives exactly the bytes the private registry received;
  nothing is rebuilt or retested.
- The receipt's two hashes keep their legacy meaning: `archive_sha256` is the hash of the reactor archive and `manifest_sha256` of its `manifest.json`, built deterministically from the
  retained files by `maven_reactor` (see its note for the one honest difference: the zip container is deterministic, not the old forge upload). The legacy consumers
  (`qrun-source-version.mjs`, `qbit_rc.py`) therefore read the same schema and the same fields, in the same namespace; no schema or storage version change was needed.
- The original deployment id is saved durably BEFORE any wait, bound to (build commit, archive hash, version), so a retry never uploads again; a pending deployment is
  `pending`, never complete; only a terminal receipt that validates against the bundle completes, and an existing one is re-checked against Central first.
- The destination is authorization data: the enrollment's `central` object names the registry owner of the receipts and whether a private source may publish publicly. The
  receipt module is `completion_receipt` (the frozen receipt primitive with owner as a parameter; its provenance is recorded there).
"""

import hashlib
import json
import urllib.error
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from types import SimpleNamespace
from typing import Any, Final, Protocol, cast

from platform_ci.delivery import completion_receipt, legacy
from platform_ci.delivery.adapters import REACTOR_METADATA_FILES, maven_modules, reactor_manifest_sha256
from platform_ci.delivery.enrollment import CentralAuthorization
from platform_ci.delivery.errors import DeliveryError
from platform_ci.delivery.manifest import ManifestFile
from platform_ci.delivery.maven_reactor import ReactorArchive, build_reactor, from_retained
from platform_ci.delivery.reuse import ReleaseArtifact

_NOT_FOUND: Final = 404


class RepositoryReader(Protocol):
   """The one forge read Central needs: the repository document, to prove its identity and visibility."""

   def get(self, path: str) -> object:
      """Return the decoded JSON of an enrolled `/repos/<repository>` path."""
      ...


@dataclass(frozen=True)
class CentralServices:
   """What a Central publication needs, created only when a channel asks for it: the Central client, the GPG signer and the generic-registry writer and reader."""

   client: legacy.CentralClientPort
   signer: object
   completion_writer: Callable[..., tuple[int, bytes]]
   completion_reader: Callable[..., tuple[int, bytes]]
   close: Callable[[], None] | None = None


@dataclass(frozen=True)
class CentralOutcome:
   """`complete` with the receipt's fields, or `pending` with the deployment state; never complete on an accepted upload."""

   state: str
   detail: str
   completion: dict[str, Any] | None = None


def verified_modules(artifact: ReleaseArtifact, files: tuple[ManifestFile, ...], version: str) -> list[dict[str, Any]]:
   """Return the verified modules of the retained reactor, all of the version the build recorded, parents (pom packaging) first."""
   modules = maven_modules(artifact, files)
   if {cast(str, module["version"]) for module in modules} != {version}:
      raise DeliveryError("the verified modules do not all declare the version the build recorded")
   return sorted(modules, key=lambda module: (module["packaging"] != "pom", module["artifactId"]))


def central_evidence(modules: list[dict[str, Any]], version: str) -> dict[str, Any]:
   """Return the frozen primitives' evidence: the version and the modules (coordinates, packaging, retained file bytes)."""
   return {"version": version, "modules": [{key: value for key, value in module.items() if key != "version"} for module in modules]}


def reactor_for(artifact: ReleaseArtifact, files: tuple[ManifestFile, ...], version: str) -> ReactorArchive:
   """Return the deterministic reactor archive of the tested bundle: its hashes are the completion receipt's, and its members are what the legacy consumers read.

   When the bundle already retains a reactor capture (`manifest.json`, `SOURCE_COMMIT`, `SHA256SUMS`, as an Android or other captured reactor does), those three files are used
   byte for byte and the archive is the deterministic zip of the retained entries under their retained relative paths: `manifest_sha256` is the hash of the existing manifest bytes,
   whatever their formatting, and nothing is regenerated. It is cross-checked against the hash the verified bundle manifest records for those bytes, and every module file the
   adapter grouped (a Gradle `.module` included) must be one of the retained entries. For ordinary Maven files with no captured metadata the archive is generated deterministically.
   """
   modules = verified_modules(artifact, files, version)
   metadata = tuple(entry for entry in files if PurePosixPath(entry.path).name in REACTOR_METADATA_FILES)
   if not metadata:
      return build_reactor(artifact.build_sha, version, modules)
   roots = {PurePosixPath(entry.path).parent for entry in metadata}
   if len(metadata) != len(REACTOR_METADATA_FILES) or len(roots) != 1:
      raise DeliveryError("the retained reactor metadata is incomplete or ambiguous")
   root = next(iter(roots))
   members: dict[str, bytes] = {}
   for entry in files:
      try:
         name = PurePosixPath(entry.path).relative_to(root).as_posix()
      except ValueError as error:
         raise DeliveryError(f"{entry.path} lies outside the retained reactor root") from error
      members[name] = artifact.bundle.read_bytes(entry)
   reactor = from_retained(artifact.build_sha, version, members)
   if reactor.manifest_sha256 != reactor_manifest_sha256(files):
      raise DeliveryError("the retained manifest.json bytes differ from the hash the verified bundle records for them")
   retained_payload = {hashlib.sha256(data).hexdigest() for name, data in members.items() if name.startswith("files/")}
   if {hashlib.sha256(item["data"]).hexdigest() for module in modules for item in module["files"]} != retained_payload:
      raise DeliveryError("the modules grouped for publication and the files of the retained reactor differ")
   return reactor


def _write_once(target: Path, document: dict[str, Any]) -> None:
   """Write a durable JSON file once; a replay must match exactly, and a conflict is an error rather than an overwrite."""
   if target.exists():
      if json.loads(target.read_text(encoding="utf-8")) != json.loads(json.dumps(document)):
         raise DeliveryError("an existing Central upload record conflicts with these tested bytes")
      return
   target.parent.mkdir(parents=True, exist_ok=True)
   temporary = target.with_suffix(".tmp")
   temporary.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8", newline="\n")
   temporary.replace(target)


def ensure_central_completion(  # noqa: PLR0913 - keyword-only evidence of one publication; the call site names every field
   *,
   repository: str,
   repository_id: int,
   authorization: CentralAuthorization | None,
   forge: RepositoryReader,
   artifact: ReleaseArtifact,
   files: tuple[ManifestFile, ...],
   version: str,
   request_name: str,
   channel: str,
   receipts: Path,
   services: Callable[[], CentralServices] | None,
   primitives: SimpleNamespace | None = None,
) -> CentralOutcome:
   """Publish the verified bundle to Central if needed and return its completion: `complete` only with a terminal, validated, bound receipt.

   Idempotent: an existing valid receipt is re-verified against Central and returned without any upload; an existing durable deployment id is resumed, never replaced.
   """
   if authorization is None:
      raise DeliveryError("the enrollment has no `central` authorization, so Central publication is refused")
   info = forge.get(f"/repos/{repository}")
   fields = cast("dict[str, Any]", info) if isinstance(info, dict) else {}
   if fields.get("id") != repository_id or not isinstance(fields.get("private"), bool):
      raise DeliveryError("the repository identity or visibility could not be proven for Central")
   if fields["private"] and not authorization.allow_private:
      raise DeliveryError("Central is a public route and the enrollment does not allow this private repository")
   if services is None:
      raise DeliveryError("the Central signing and publication services were not supplied (the existing /run/publisher-central mounts and a generic-registry writer and reader)")
   primitive = primitives or legacy.central_primitives()
   owner = authorization.completion_owner
   run_id, build_sha = int(artifact.run.run_id), artifact.build_sha
   modules = verified_modules(artifact, files, version)
   reactor = reactor_for(artifact, files, version)
   evidence = central_evidence(modules, version)
   gav = sorted(f"{module['groupId']}:{module['artifactId']}:{version}" for module in modules)
   snapshot = legacy.snapshot_stamp(artifact.run.created_at, run_id) if version.endswith("-SNAPSHOT") else None
   target = "central-snapshots" if snapshot else "central-publisher"
   expected = {"repository": repository, "repository_id": repository_id, "run_id": run_id, "source_commit": build_sha, "version": version, "archive_sha256": reactor.archive_sha256, "manifest_sha256": reactor.manifest_sha256, "gav": gav, "target": target}
   binding = {"source_commit": build_sha, "archive_sha256": reactor.archive_sha256, "version": version}
   service = services()
   try:
      with legacy.translate_refusals():
         existing = _read_existing(owner, service, expected)
         if existing is not None:
            _recheck_central(primitive, service, evidence, existing, snapshot is not None)
            return CentralOutcome("complete", "an existing terminal receipt, re-verified against Central", existing)
         upload_record = receipts / repository / f"{build_sha}.{request_name}.{channel}.central-upload.json"
         deployment_id = None
         if upload_record.exists():
            prior = json.loads(upload_record.read_text(encoding="utf-8"))
            if any(prior.get(key) != value for key, value in binding.items()):
               raise DeliveryError("a saved Central upload belongs to different tested bytes")
            deployment_id = prior["deployment_id"]

         def save_deployment(value: str) -> None:
            _write_once(upload_record, binding | {"deployment_id": value})

         try:
            external = primitive.central.publish_central(evidence, service.signer, service.client, snapshot, deployment_id=deployment_id, save_deployment=save_deployment, wait=False)
         except primitive.central.CentralPending as pending:
            return CentralOutcome("pending", str(pending))
         except primitive.central.CentralDuplicate as duplicate:
            raise DeliveryError("Central already holds this deployment; the operator decides, nothing is retried automatically") from duplicate
         completion = {
            "schema": completion_receipt.SCHEMA,
            "repository": repository,
            "repository_id": repository_id,
            "run_id": run_id,
            "source_commit": build_sha,
            "version": version,
            "archive_sha256": reactor.archive_sha256,
            "manifest_sha256": reactor.manifest_sha256,
            "gav": gav,
            "target": target,
            "state": external["central_status"],
            "deployment_id": external["central_deployment"],
            "files": external["central_files"],
            "purls": external["central_purls"],
         }
         completion_receipt.publish(owner, completion, service.completion_writer, service.completion_reader)
         return CentralOutcome("complete", "published and recorded", completion)
   finally:
      if service.close is not None:
         service.close()


def _read_existing(owner: str, service: CentralServices, expected: dict[str, Any]) -> dict[str, Any] | None:
   """Read the immutable receipt, validated against the tested bundle; an absent one (404) is None, any mismatch raises."""
   try:
      return completion_receipt.read(owner, service.completion_reader, expected["repository_id"], expected["run_id"], expected["source_commit"], expected)
   except urllib.error.HTTPError as error:
      error.close()
      if error.code == _NOT_FOUND:
         return None
      raise


def _recheck_central(primitive: SimpleNamespace, service: CentralServices, evidence: dict[str, Any], receipt: dict[str, Any], snapshot: bool) -> None:
   """Re-check that what the receipt records is still true at Central: a release still PUBLISHED with the same purls, or snapshot files byte-identical."""
   if snapshot:
      for entry in receipt["files"]:
         if hashlib.sha256(service.client.read_snapshot(entry["path"])).hexdigest() != entry["sha256"]:
            raise DeliveryError("a previously completed Central snapshot file changed")
      return
   status = service.client.status(receipt["deployment_id"])
   if status.get("deploymentState") != "PUBLISHED" or primitive.central.verified_purls(evidence, status.get("purls")) != receipt["purls"]:
      raise DeliveryError("a previously completed Central deployment is no longer published with the same coordinates")