spirekeeper/cassette_transport.py

"""
Cassette-config Nostr transport — operator ↔ ATM kind-30078 publish + consume.

Per the locked design at aiolabs/satmachineadmin#29 (paired with
lamassu-next#56) and the dcd0874 privacy-by-default pivot, the operator
publishes position-keyed cassette config to a target ATM via:

  kind = 30078 (NIP-78, replaceable)
  tags = [
    ["d", "bitspire-cassettes:<atm_pubkey_hex>"],
    ["p", "<atm_pubkey_hex>"]
  ]
  content = NIP-44 v2 encrypted JSON of PublishCassettesPayload.to_wire_dict()
  pubkey = operator pubkey
  sig    = operator signature

The ATM-side consumer (lamassu-next#56) subscribes by the d-tag + its own
npub, decrypts, validates, applies, hot-reloads HAL.

The operator → ATM direction carries OPERATIONS as of v2 (bitspire ADR-004):
refill, empty, recount, set_denomination, each with an id the machine dedups
on. It used to carry absolute counts, which meant the operator and the machine
both wrote the same value over a transport that never tells a writer it lost —
so a form loaded before a dispense discarded that dispense when published.

Reverse direction (ATM → operator, continuous: the machine publishes on
startup, after every change to its bays, and on a heartbeat):

  kind = 30078
  tags = [
    ["d", "bitspire-cassettes-state:<atm_pubkey_hex>"],
    ["p", "<operator_pubkey_hex>"]
  ]
  content = NIP-44 v2 encrypted JSON, same PublishCassettesPayload shape
  pubkey = ATM pubkey

This module owns the wire-format side of both directions. The consumer
task (tasks.py) calls `decrypt_and_parse_state_event` per incoming event;
the API endpoint (views_api.py) calls `publish_ops_to_atm` per operation.

The `<m>` placeholder semantics (load-bearing per the 2026-05-30T11:50Z
coord-log entry): always the ATM's hex pubkey, NEVER spirekeeper's
internal dca_machines.id UUID. Helper `_atm_hex_pubkey(machine)`
centralises the canonicalisation via lnbits.utils.nostr.normalize_public_key.
"""

from __future__ import annotations

import json

from lnbits.core.services.nip46_bunker_client import (
    NsecBunkerRpcError,
    NsecBunkerTimeoutError,
)
from lnbits.core.signers.base import (
    NostrSigner,
    SignerUnavailableError,
)
from lnbits.utils.nostr import normalize_public_key

from .models import (
    CassetteOp,
    Machine,
    PublishCassetteOpsPayload,
    PublishCassettesPayload,
)
from .nip44 import Nip44Error
from .nostr_publish import (
    NostrPublishError,
    OperatorIdentityMissing,  # re-export for callers that catch this
    RelayUnavailable,  # re-export
    SignerUnavailable,  # re-export
    nip44_decrypt_via_signer,
    publish_encrypted_kind_30078,
)

# Re-exported so external callers (views_api etc.) can keep importing
# from cassette_transport without breakage. Same for the public
# constants below.
__all__ = [
    "CassetteTransportError",
    "CassetteEventDecodeError",
    "CassetteEventTransientError",
    "OperatorIdentityMissing",
    "SignerUnavailable",
    "RelayUnavailable",
    "build_state_d_tags_for_machines",
    "decrypt_and_parse_state_event",
    "publish_ops_to_atm",
]

_D_TAG_CONFIG_PREFIX = "bitspire-cassettes:"  # operator → ATM
_D_TAG_STATE_PREFIX = "bitspire-cassettes-state:"  # ATM → operator


# =============================================================================
# Errors — cassette-specific subclasses of the generic NostrPublishError
# =============================================================================


class CassetteTransportError(NostrPublishError):
    """Generic cassette-transport error. Subclasses distinguish failure
    modes so the API can surface meaningful HTTP statuses + the consumer
    task can log + skip without crashing.

    Bridges back-compat with pre-extraction callers that catch this
    class — now equivalent to NostrPublishError plus the two consumer-
    side decode/transient distinctions below.
    """


class CassetteEventDecodeError(CassetteTransportError):
    """Inbound state event failed validation: bad signature, NIP-44 v2
    decrypt failure, or payload didn't conform to PublishCassettesPayload.
    Terminal — caller should log + skip, advancing past the event."""


class CassetteEventTransientError(CassetteTransportError):
    """Inbound state event couldn't be decrypted because the signer
    component (typically the bunker) is transiently unavailable. Caller
    should NOT advance past the event; retry on next tick.

    Distinct from CassetteEventDecodeError so the consumer task can
    differentiate "MAC failed, give up" from "bunker is partitioned, try
    again in a few seconds" — surfaced by lnbits at coord-log
    2026-05-31T07:10Z as the load-bearing distinction post-PR-#38."""


# =============================================================================
# Helpers — canonical pubkey + d-tag construction
# =============================================================================


def _atm_hex_pubkey(machine: Machine) -> str:
    """Canonicalise machine.machine_npub (hex OR npub bech32 — operator
    enters either in the UI) to lowercase hex. ALL d-tag substitutions
    use this value; using the internal machine.id UUID would silently
    no-op the wire-level filter (per coord-log 11:50Z load-bearing nudge).
    """
    return normalize_public_key(machine.machine_npub).lower()


def _config_d_tag(atm_pubkey_hex: str) -> str:
    """d-tag for operator → ATM publish. ATM subscribes by this tag."""
    return f"{_D_TAG_CONFIG_PREFIX}{atm_pubkey_hex}"


def _state_d_tag(atm_pubkey_hex: str) -> str:
    """d-tag for ATM → operator publish (bootstrap in v1, continuous v2)."""
    return f"{_D_TAG_STATE_PREFIX}{atm_pubkey_hex}"


def build_state_d_tags_for_machines(machines: list[Machine]) -> list[str]:
    """Bootstrap-consumer subscription filter helper: returns the full
    `#d=[...]` list for all known PAIRED ATMs an operator subscribes to.
    Unpaired machines (machine_npub is None — nullable since #29/m011) have no
    state-beacon d-tag yet, so skip them rather than crash `_atm_hex_pubkey`."""
    return [_state_d_tag(_atm_hex_pubkey(m)) for m in machines if m.machine_npub]


# =============================================================================
# Publish — operator → ATM (the spirekeeper API path)
# =============================================================================


async def publish_ops_to_atm(
    machine: Machine,
    ops: list[CassetteOp],
    operator_user_id: str,
) -> dict:
    """Publish the operator's recent cassette OPERATIONS to the target ATM.

    The v2 wire (bitspire ADR-004). Replaces sending absolute counts, which
    let a dashboard form loaded before a dispense silently discard that
    dispense — the operator and the machine were both writing the same value
    over a transport that never tells a writer it lost.

    `ops` is a WINDOW, oldest-first, not just the newest change. The event is
    addressable, so each publish replaces the last, and a machine that was
    offline for one of them would otherwise never see that operation again.
    Carrying the recent history means the channel heals itself without anyone
    noticing it broke. Re-delivery is harmless because each op carries an id
    the machine dedups on.
    """
    atm_pubkey_hex = _atm_hex_pubkey(machine)
    payload = PublishCassetteOpsPayload(ops=ops)
    signed = await publish_encrypted_kind_30078(
        operator_user_id=operator_user_id,
        recipient_pubkey_hex=atm_pubkey_hex,
        d_tag=_config_d_tag(atm_pubkey_hex),
        payload=payload.to_wire_dict(),
        log_context=(
            f"cassette ops (machine={machine.id}, ops={[o.op_type for o in ops]})"
        ),
    )
    return signed


# =============================================================================
# Consume — ATM → operator (the machine's state reports)
# =============================================================================


async def decrypt_and_parse_state_event(
    event: dict, account, signer: NostrSigner
) -> PublishCassettesPayload:
    """Decrypt + parse an inbound `bitspire-cassettes-state:<atm_pubkey_hex>`
    event the ATM published toward the operator.

    Caller is responsible for:
      - filtering on `kind=30078` and the expected `#d` tag list
      - verifying the event signature (lnbits.utils.nostr.verify_event)
      - confirming `event["pubkey"]` matches a known ATM (= machine.machine_npub
        canonicalised) — the consumer task does this before calling here
      - resolving the operator's account + signer via
        `_resolve_operator_signer(...)` and passing them in

    This function does:
      - NIP-44 v2 decrypt of event["content"] via `signer.nip44_decrypt`
        (bunker round-trip on RemoteBunkerSigner; direct prvkey on the
        transitional LocalSigner path)
      - JSON parse + PublishCassettesPayload validation

    Error mapping:
      - CassetteEventTransientError on NsecBunkerTimeoutError → caller
        should NOT advance state_event_id; retry on next consumer tick
      - CassetteEventDecodeError on anything else (bunker RPC reject,
        signer unavailable, MAC failure, JSON parse, payload shape) →
        terminal; caller logs + skips
    """
    sender_pubkey = event.get("pubkey")
    content = event.get("content")
    if not isinstance(sender_pubkey, str) or not isinstance(content, str):
        raise CassetteEventDecodeError(
            "event missing required pubkey or content fields"
        )

    try:
        plaintext = await nip44_decrypt_via_signer(
            account, signer, content, sender_pubkey
        )
    except NsecBunkerTimeoutError as exc:
        raise CassetteEventTransientError(
            f"bunker unreachable while decrypting cassette state event: {exc}"
        ) from exc
    except NsecBunkerRpcError as exc:
        raise CassetteEventDecodeError(
            f"bunker rejected nip44_decrypt (policy / MAC / config): {exc}"
        ) from exc
    except SignerUnavailableError as exc:
        raise CassetteEventDecodeError(f"signer cannot nip44-decrypt: {exc}") from exc
    except Nip44Error as exc:
        # Hand-rolled LocalSigner fallback path (transitional) — MAC fail
        # / version mismatch / length issue.
        raise CassetteEventDecodeError(
            f"NIP-44 v2 decrypt failed (LocalSigner fallback path): {exc}"
        ) from exc
    except ValueError as exc:
        # coincurve raises ValueError on a malformed pubkey hex (only
        # reachable via the LocalSigner fallback path; the bunker handles
        # pubkey validation server-side).
        raise CassetteEventDecodeError(f"sender pubkey is malformed: {exc}") from exc

    try:
        raw = json.loads(plaintext)
    except json.JSONDecodeError as exc:
        raise CassetteEventDecodeError(
            f"decrypted content isn't valid JSON: {exc}"
        ) from exc

    try:
        return PublishCassettesPayload(**raw)
    except Exception as exc:
        raise CassetteEventDecodeError(
            f"payload didn't validate as PublishCassettesPayload: {exc}"
        ) from exc