target-revenue/src/target_revenue/ledger.py

"""Hosted Target Ledger append API (WP-0006-T04).

The highest-risk property this module owns: any party who independently
computes `fold.fold_outstanding_target` over an exported Manifest + Ledger
must get exactly the same Outstanding Target the hosted service itself
would report. This module therefore does nothing clever — it authenticates
the write, validates the entry with the same offline `validation.py`
checks Stage 0 already tested, computes `previous_entry_hash` and
`signature` itself (never trusting client-supplied values for either), and
appends. The fold is never computed or stored here; `fold.py` remains the
single source of truth, run fresh from whatever a caller exports.
"""

from __future__ import annotations

from typing import Any

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from psycopg import Connection
from psycopg.errors import UniqueViolation
from psycopg.types.json import Jsonb

from . import hashing, validation
from .registry import Licensor, RegistrationError, get_phase_manifest


def append_entry(
    conn: Connection,
    licensor: Licensor,
    phase_id: str,
    entry_input: dict[str, Any],
    signing_key: Ed25519PrivateKey,
) -> dict[str, Any]:
    """Validate, chain, sign, and persist one Target Ledger entry.

    `entry_input` must not include `previous_entry_hash` or `signature` —
    those are exclusively server-computed. Supplying them is rejected
    outright rather than silently overwritten, so a caller never mistakes
    a value it sent for one the service actually used.
    """
    if "previous_entry_hash" in entry_input or "signature" in entry_input:
        raise RegistrationError(
            "previous_entry_hash and signature are computed by the Trust "
            "Service instance and must not be supplied by the caller"
        )

    manifest = get_phase_manifest(conn, phase_id)
    if manifest is None:
        raise RegistrationError(f"phase {phase_id!r} is not registered")

    manifest_licensor = conn.execute(
        "SELECT licensor_id FROM phase_manifests WHERE phase_id = %s", (phase_id,)
    ).fetchone()[0]
    if manifest_licensor != licensor.licensor_id:
        raise RegistrationError(
            "this token is not authorized to append to this Phase's ledger"
        )

    if entry_input.get("phase") != phase_id:
        raise RegistrationError(
            f"entry.phase {entry_input.get('phase')!r} does not match "
            f"the target Phase {phase_id!r}"
        )

    # Serialize concurrent appends to this Phase so `previous_entry_hash`
    # always reflects a real, uncontested chain tip — a transaction-scoped
    # advisory lock, released automatically at commit/rollback.
    conn.execute("SELECT pg_advisory_xact_lock(hashtext(%s))", (phase_id,))

    tip = conn.execute(
        """
        SELECT entry FROM ledger_entries
         WHERE phase_id = %s
         ORDER BY sequence DESC
         LIMIT 1
        """,
        (phase_id,),
    ).fetchone()
    previous_entry_hash = hashing.entry_hash(tip[0]) if tip else hashing.GENESIS

    candidate = {**entry_input, "previous_entry_hash": previous_entry_hash}

    try:
        validation.validate_ledger_entry(candidate)
    except validation.ConformanceError as exc:
        raise RegistrationError(
            "ledger entry rejected: " + "; ".join(exc.errors)
        ) from exc

    currency_errors = validation.check_currency_consistency(manifest, [candidate])
    if currency_errors:
        raise RegistrationError(
            "ledger entry rejected: " + "; ".join(currency_errors)
        )

    signed = {**candidate, "signature": hashing.sign_record(candidate, signing_key)}

    try:
        conn.execute(
            """
            INSERT INTO ledger_entries
                (entry_id, phase_id, entry, previous_entry_hash, signature, recognized_at,
                 submitted_by_token)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
            """,
            (
                signed["id"],
                phase_id,
                Jsonb(signed),
                previous_entry_hash,
                signed["signature"],
                signed["recognized_at"],
                licensor.token,
            ),
        )
    except UniqueViolation as exc:
        raise RegistrationError(
            f"entry id {signed['id']!r} already exists"
        ) from exc

    return signed


def get_ledger(conn: Connection, phase_id: str) -> list[dict[str, Any]]:
    """Return a Phase's entries in exact append order (fold input order)."""
    rows = conn.execute(
        """
        SELECT entry FROM ledger_entries
         WHERE phase_id = %s
         ORDER BY sequence ASC
        """,
        (phase_id,),
    ).fetchall()
    return [row[0] for row in rows]


def get_ledger_attribution(conn: Connection, phase_id: str) -> list[dict[str, Any]]:
    """Return, per entry in append order, which credential submitted it
    (WP-0009-T02: `licensors.credential_label`, joined via
    `ledger_entries.submitted_by_token`). This is a hosting-layer-only
    view — the credential label is never part of the signed entry payload
    itself (`get_ledger`'s output), so an export of the raw Ledger remains
    exactly as portable and schema-conformant as before this task."""
    rows = conn.execute(
        """
        SELECT le.entry_id, li.credential_label, li.rights
          FROM ledger_entries le
          LEFT JOIN licensors li ON li.token = le.submitted_by_token
         WHERE le.phase_id = %s
         ORDER BY le.sequence ASC
        """,
        (phase_id,),
    ).fetchall()
    return [
        {"entry_id": entry_id, "credential_label": credential_label, "rights": rights}
        for entry_id, credential_label, rights in rows
    ]