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.

WP-0009-T02 extended the `licensors` auth model so one Licensor tenant
(e.g. `binky`) can hold multiple, individually-issued, individually-
revocable credentials (`issue_sub_credential`, `revoke_sub_credential`),
each carrying its own `credential_label` and rights tier
(`RIGHTS_TIERS`), while still resolving to the same `licensor_id` for
every phase-ownership check in this module and `ledger.py` — unchanged
from the single-token model WP-0006 originally shipped.
"""

from __future__ import annotations

import secrets
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."""


# Rights tiers accepted 2026-07-30
# (workplans/TREV-WP-0009-target-revenue-control-plane.md T01,
# specs/TargetRevenueControlPlaneConcept.md §2), ordered least to most
# privileged. Enforcing what each tier may actually do is the Control
# Plane's own job (WP-0009-T03/T04) — this module only carries the tier
# label through authentication and offers `has_right` as a shared ordinal
# comparison, so that enforcement logic doesn't have to reinvent the
# ordering itself.
RIGHTS_TIERS = ("viewer", "contributor", "operator", "admin")


def has_right(rights: str, minimum: str) -> bool:
    """True if `rights` is at least as privileged as `minimum` (RIGHTS_TIERS order)."""
    return RIGHTS_TIERS.index(rights) >= RIGHTS_TIERS.index(minimum)


@dataclass(frozen=True)
class Licensor:
    """An authenticated caller: a Licensor tenant, via one of possibly
    several individually-issued, individually-revocable credentials
    (WP-0009-T02). `credential_label` and `rights` distinguish *which*
    credential authenticated, for attribution and authorization purposes,
    even though every credential for the same `licensor_id` has identical
    phase-ownership rights in `registry.py`/`ledger.py`'s existing checks.
    """

    token: str
    licensor_id: str
    credential_label: str | None = None
    rights: str = "operator"


def generate_credential_token() -> str:
    """A new random credential token, suitable for `issue_sub_credential`."""
    return secrets.token_hex(32)


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").

    A revoked credential (`revoked_at` set) is rejected exactly like an
    unrecognized one — from the caller's perspective there is no
    observable difference between "never existed" and "no longer valid,"
    which is the correct behavior for a revoked credential.
    """
    row = conn.execute(
        """
        SELECT licensor_id, credential_label, rights
          FROM licensors
         WHERE token = %s AND revoked_at IS NULL
        """,
        (token,),
    ).fetchone()
    if row is None:
        raise RegistrationError("unrecognized or revoked token")
    licensor_id, credential_label, rights = row
    return Licensor(
        token=token, licensor_id=licensor_id, credential_label=credential_label, rights=rights
    )


def create_licensor_identity(conn: Connection, licensor_id: str) -> None:
    """Onboard a brand-new Licensor tenant identity (`licensor_identities`),
    a prerequisite before any credential can be issued for it
    (`issue_sub_credential`'s foreign key requires the identity to already
    exist). Deliberately a separate, explicit step rather than an implicit
    side effect of issuing the first credential — matching this
    framework's existing pattern of governance actions being explicit
    (`promote_extension_canonical`, `revoke_sub_credential`), not
    automated. There is exactly one identity today (`binky`,
    `specs/TRSL-Governance.md` §1); this function exists so a future
    additional tenant does not require a schema change to onboard.
    """
    try:
        conn.execute(
            "INSERT INTO licensor_identities (licensor_id) VALUES (%s)", (licensor_id,)
        )
    except UniqueViolation as exc:
        raise RegistrationError(f"licensor identity {licensor_id!r} already exists") from exc


def issue_sub_credential(
    conn: Connection,
    licensor_id: str,
    credential_label: str,
    rights: str,
    issued_by: str,
) -> Licensor:
    """Issue a new, individually-revocable credential for a Licensor
    tenant (e.g. a second credential for `binky`, labeled for a specific
    human user). This is itself a governance action — `issued_by` records
    who issued it, the same attributable-action pattern already used by
    `promote_extension_canonical`. The Licensor identity is auto-created
    on first use if it doesn't already exist
    (`migrations/0005_licensor_credentials.sql`'s `ensure_licensor_identity`
    trigger) — `create_licensor_identity` remains available for callers
    that want tenant onboarding as its own explicit, auditable step rather
    than an implicit side effect of the first credential.
    """
    if rights not in RIGHTS_TIERS:
        raise RegistrationError(f"rights must be one of {RIGHTS_TIERS}, got {rights!r}")
    token = generate_credential_token()
    try:
        conn.execute(
            """
            INSERT INTO licensors (token, licensor_id, credential_label, rights, issued_by)
            VALUES (%s, %s, %s, %s, %s)
            """,
            (token, licensor_id, credential_label, rights, issued_by),
        )
    except UniqueViolation as exc:
        raise RegistrationError(
            f"an active credential labeled {credential_label!r} already exists for "
            f"licensor {licensor_id!r}"
        ) from exc
    return Licensor(
        token=token, licensor_id=licensor_id, credential_label=credential_label, rights=rights
    )


def revoke_sub_credential(conn: Connection, token: str) -> None:
    """Revoke a credential via the database's `revoke_credential` function
    (not a direct UPDATE — the application role has no UPDATE grant on
    `licensors` at all, matching `promote_extension_canonical`'s pattern).
    Idempotent: revoking an already-revoked or unknown token is not an
    error — the caller's intent (this token must not work) is satisfied
    either way."""
    conn.execute("SELECT revoke_credential(%s)", (token,))


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 list_phase_manifests_for_licensor(conn: Connection, licensor_id: str) -> list[dict[str, Any]]:
    """All Phases registered by one Licensor tenant, most recently
    registered first — needed by any UI that wants to show "my Phases"
    (WP-0009-T04's Control Plane dashboard) rather than requiring a
    caller to already know every `phase_id` in advance."""
    rows = conn.execute(
        """
        SELECT manifest FROM phase_manifests
         WHERE licensor_id = %s
         ORDER BY registered_at DESC
        """,
        (licensor_id,),
    ).fetchall()
    return [row[0] for row in rows]


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),
    )