target-revenue/src/target_revenue/registry.py

"""Hosted Phase Registry and Extension Registry (WP-0006-T03).

Wraps `validation.py`'s already-tested conformance checks with Postgres
persistence, per `docs/adr/ADR-0002-hosted-trust-service-stack.md`. This
module adds multi-tenancy and storage; it must not introduce new validation
logic beyond what `validation.py` already does (TSD §4.1: this task "adds
multi-tenancy and persistence, not new validation logic").

Append-only enforcement for `phase_manifests`, and canonicalization-as-
governance-action for `extensions`, are primarily enforced at the database
level (see `migrations/0001_registries.sql`) — this module's job is to
surface a conformance rejection *before* attempting a write that the
database would reject anyway, so callers get a field-by-field diff instead
of an opaque database error.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

from psycopg import Connection
from psycopg.errors import UniqueViolation
from psycopg.types.json import Jsonb

from . import validation


class RegistrationError(ValueError):
    """Raised with the specific reason a registration was rejected."""


@dataclass(frozen=True)
class Licensor:
    token: str
    licensor_id: str


def authenticate(conn: Connection, token: str) -> Licensor:
    """Resolve a bearer token to a Licensor, or raise RegistrationError.

    Per ADR-0002: write access is scoped per Licensor via this token, kept
    independent of the per-entry Ed25519 signing WP-0002 already
    implements (that signing happens at the Ledger layer, T04 — this
    function only answers "who is calling," not "is this entry authentic").
    """
    row = conn.execute(
        "SELECT licensor_id FROM licensors WHERE token = %s", (token,)
    ).fetchone()
    if row is None:
        raise RegistrationError("unrecognized or revoked token")
    return Licensor(token=token, licensor_id=row[0])


def register_phase_manifest(
    conn: Connection, licensor: Licensor, manifest: dict[str, Any]
) -> None:
    """Validate and persist a Phase Manifest.

    Rejects, with the same field-by-field diff `validation.py` produces
    offline, rather than silently accepting a non-conformant manifest
    (TSD §3.1 validation rule). A manifest is inserted once; there is no
    update path — `phase_manifests` grants no UPDATE/DELETE to the
    application role (migrations/0001_registries.sql), so a second attempt
    to register the same `phase_id` fails at the database's unique
    constraint, surfaced here as RegistrationError rather than a raw
    IntegrityError.
    """
    try:
        validation.validate_phase_manifest(manifest)
    except validation.ConformanceError as exc:
        raise RegistrationError(
            "manifest rejected: " + "; ".join(exc.errors)
        ) from exc

    phase_id = manifest["phase"]["id"]
    try:
        conn.execute(
            """
            INSERT INTO phase_manifests (phase_id, licensor_id, manifest)
            VALUES (%s, %s, %s)
            """,
            (phase_id, licensor.licensor_id, Jsonb(manifest)),
        )
    except UniqueViolation as exc:
        raise RegistrationError(
            f"phase_id {phase_id!r} is already registered; "
            "a Phase Manifest cannot be re-registered or replaced in place"
        ) from exc


def get_phase_manifest(conn: Connection, phase_id: str) -> dict[str, Any] | None:
    row = conn.execute(
        "SELECT manifest FROM phase_manifests WHERE phase_id = %s", (phase_id,)
    ).fetchone()
    return row[0] if row else None


def register_extension(
    conn: Connection, licensor: Licensor, extension: dict[str, Any]
) -> None:
    """Validate and persist a Monetization Extension registration.

    New registrations always start at `status: registered` (assigned here,
    never accepted as an input field from the caller) — `canonical` is a
    separate, governance-gated transition (TS-FR-2; see
    `promote_extension_canonical` below).
    """
    try:
        validation.validate_extension_contract(extension)
    except validation.ConformanceError as exc:
        raise RegistrationError(
            "extension rejected: " + "; ".join(exc.errors)
        ) from exc

    extension_id = extension["id"]
    version = extension["version"]
    try:
        conn.execute(
            """
            INSERT INTO extensions (extension_id, version, licensor_id, contract, status)
            VALUES (%s, %s, %s, %s, 'registered')
            """,
            (extension_id, version, licensor.licensor_id, Jsonb(extension)),
        )
    except UniqueViolation as exc:
        raise RegistrationError(
            f"extension {extension_id!r}@{version!r} is already registered"
        ) from exc


def get_extension(
    conn: Connection, extension_id: str, version: str
) -> dict[str, Any] | None:
    row = conn.execute(
        "SELECT contract, status FROM extensions WHERE extension_id = %s AND version = %s",
        (extension_id, version),
    ).fetchone()
    if row is None:
        return None
    contract, status = row
    return {**contract, "status": status}


def promote_extension_canonical(
    conn: Connection, extension_id: str, version: str, approved_by: str
) -> None:
    """Promote an extension from `registered` to `canonical`.

    Calls the database's `set_extension_status` function rather than an
    UPDATE, because the application role has no UPDATE grant on
    `extensions` at all (migrations/0001_registries.sql) — canonicalization
    is a documented human/governance action recorded with who performed it,
    never a route ordinary application code can take by itself
    (TSD §4.1: "never automated").
    """
    conn.execute(
        "SELECT set_extension_status(%s, %s, %s, %s)",
        (extension_id, version, "canonical", approved_by),
    )