cardano-checkout-py/cardano_checkout/mint.py

"""CIP-25 v2 NFT certificate-of-authenticity minting.

This module produces the NFT cert attached to a confirmed merchant
order. One NFT per order, pinned-once metadata (image CID from IPFS
via :mod:`cardano_checkout.ipfs`), sent directly to the customer's
wallet in the same transaction.

Design decisions:

- **CIP-25 v2** (not CIP-68). CIP-25 is universally supported by
  every Cardano wallet (Eternl, Lace, Yoroi, Vespr, Typhon). CIP-68
  adds reference-NFT mutability we do not need for a static cert.
- **Single policy per merchant studio.** All of a studio's certs share
  one policy_id so wallets group them cleanly. The policy key is a
  native script under the studio's custody — Sulkta pattern is a
  multi-sig native script stored on Lucy.
- **Policy has a time-lock** (invalid-after slot) so the "no more
  editions can be minted after X" claim is cryptographically enforceable.
  Recommended: generous lock (100 years) so policy_id stays stable,
  but revokable in-contract via ``mint policy revoke`` flow.
- **No reference script, no Plutus.** Pure native scripts + standard
  CIP-25 metadata keeps the tx cheap (~0.18 ADA fee + min-utxo for the
  NFT output).

Cold-signing workflow
---------------------

The mint function does *not* sign. It builds the transaction body + the
auxiliary data, computes the tx id, and returns an :class:`UnsignedMint`
carrying the CBOR-encoded body plus a human-readable summary so the
operator can sanity-check before signing. The operator then:

1. Transfers the unsigned CBOR to the cold host (Lucy, via `scp`, USB,
   QR code, whatever the threat model tolerates).
2. Signs offline with the policy-required skey(s) — for Sulkta's
   chromatic policy that's ``Cobb.skey`` + ``Kayos.skey``.
3. Transfers the signed CBOR back to the hot host.
4. Calls :func:`submit_signed_tx` to hand it to Ogmios.

See ``docs/minting-workflow.md`` for the full operator runbook.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Optional

logger = logging.getLogger(__name__)

if TYPE_CHECKING:  # pragma: no cover — hints only
    from pycardano import ChainContext


# ---------------------------------------------------------------------------
# Policy model
# ---------------------------------------------------------------------------


@dataclass
class MintPolicy:
    """A native-script minting policy under the SDK's custody model.

    Attributes:
        policy_id: Hex blake2b-224 hash of the native script CBOR. Stable
            for the life of the policy — shipped with every cert minted
            under it. Becomes the Cardano ``policy_id`` of the NFT asset.
        script_cbor_hex: Hex-encoded CBOR of the native script itself.
            Submitted alongside the mint tx witness.
        required_signer_hashes: Payment-key hashes (hex) of every skey
            that must sign the mint tx. For Sulkta's chromatic policy
            this is 2 entries: Cobb + Kayos.
        locked_after_slot: Optional slot beyond which the policy rejects
            further mints. None = no time lock (not recommended for
            certificates — a lock makes the "no more editions" claim
            mathematically verifiable).
    """

    policy_id: str
    script_cbor_hex: str
    required_signer_hashes: list[str] = field(default_factory=list)
    locked_after_slot: Optional[int] = None


@dataclass
class UnsignedMint:
    """An unsigned mint transaction, ready to be handed to a cold signer.

    Attributes:
        tx_id: Transaction hash computed from the body alone (stable across
            signing — the same id the explorer will show once submitted).
        tx_body_cbor_hex: Hex-encoded CBOR of the transaction *body*.
            This is what gets moved to the cold host.
        auxiliary_data_cbor_hex: Hex-encoded CBOR of the auxiliary data
            (metadata + native script). Required to reconstruct the full
            transaction before submission.
        native_script_cbor_hex: Hex-encoded CBOR of the minting policy's
            native script. Needed by the cold signer to construct the
            correct witness set.
        required_signer_hashes: List of payment-key hashes (hex) the cold
            signer must provide. Mirrors ``MintPolicy.required_signer_hashes``.
        summary: Human-readable description of the tx — operator should
            eyeball this before signing to confirm they're signing what
            they think they're signing.
    """

    tx_id: str
    tx_body_cbor_hex: str
    auxiliary_data_cbor_hex: str
    native_script_cbor_hex: str
    required_signer_hashes: list[str]
    summary: str


# ---------------------------------------------------------------------------
# Metadata builder (pure, no pycardano dep)
# ---------------------------------------------------------------------------


def build_cip25_metadata(
    policy_id: str,
    asset_name: str,
    name: str,
    image_cid: str,
    description: str = "",
    media_type: str = "image/jpeg",
    properties: Optional[dict] = None,
) -> dict:
    """Assemble the ``{721: {...}}`` metadatum envelope for a single NFT.

    CIP-25 v2 image field takes an ``ipfs://<CID>`` URI. Description, if
    longer than 64 characters, is split into an array of ≤64-char chunks
    (CIP-25 constraint from the Cardano metadata schema — strings larger
    than 64 chars are encoded as a list of chunks).

    Args:
        policy_id: Hex policy id (same as on the asset).
        asset_name: UTF-8 asset name — used as the dict key under policy_id.
        name: Human-readable NFT title (shown in wallets).
        image_cid: IPFS CID — the function prepends ``ipfs://``.
        description: Optional longer text. Will be chunked if > 64 chars.
        media_type: MIME type of the image. Default ``image/jpeg``.
        properties: Additional key/value pairs merged into the metadata blob.

    Returns:
        Dict ready to submit as tx metadatum label 721.
    """

    def chunk64(s: str) -> list[str]:
        if len(s) <= 64:
            return [s]
        return [s[i : i + 64] for i in range(0, len(s), 64)]

    desc: object = description
    if isinstance(description, str) and len(description) > 64:
        desc = chunk64(description)

    body: dict = {
        "name": name,
        "image": f"ipfs://{image_cid}",
        "mediaType": media_type,
    }
    if desc:
        body["description"] = desc
    if properties:
        body.update(properties)

    return {
        "721": {
            policy_id: {
                asset_name: body,
            },
            "version": "2.0",
        }
    }


# ---------------------------------------------------------------------------
# Mint transaction builder (cold-signer flow)
# ---------------------------------------------------------------------------


def _require_pycardano():
    try:
        import pycardano  # noqa: F401
    except ImportError as exc:  # pragma: no cover — env sanity
        raise RuntimeError(
            "pycardano is required for mint transaction construction. "
            "Add pycardano>=0.11.0 to requirements.txt and reinstall."
        ) from exc


def _metadata_dict_with_int_keys(metadata: dict) -> dict:
    """Convert string top-level metadata labels to ints for pycardano Metadata.

    CIP-25 v2 nests everything under label ``721``. We accept both ``{"721": ...}``
    (builder output) and ``{721: ...}`` (raw) for ergonomics.
    """
    converted: dict = {}
    for key, val in metadata.items():
        try:
            converted[int(key)] = val
        except (TypeError, ValueError):
            converted[key] = val
    return converted


async def mint_nft_cert(
    policy: MintPolicy,
    asset_name: str,
    metadata: dict,
    recipient_address: str,
    funding_address: str,
    context: Optional["ChainContext"] = None,
    ogmios_host: str = "127.0.0.1",
    ogmios_port: int = 1337,
    network: str = "mainnet",
    min_lovelace_for_nft_utxo: int = 1_500_000,
) -> UnsignedMint:
    """Build an unsigned mint+send transaction for a CIP-25 v2 NFT cert.

    Constructs a transaction that:

    1. Mints exactly 1 of ``{policy.policy_id}.{asset_name}``.
    2. Sends that single token to ``recipient_address`` in its own UTxO
       with the minimum-ADA padding (default 1.5 ADA).
    3. Attaches the CIP-25 v2 metadata (label 721) + the policy's
       native script as tx auxiliary data.
    4. Returns the unsigned body for the cold signer to sign — does NOT
       sign, does NOT submit.

    UTxOs for fees + min-ADA are sourced from ``funding_address`` (the
    merchant's hot wallet on Rackham, which does not hold any policy keys).

    Args:
        policy: Merchant's minting policy.
        asset_name: UTF-8 asset name (will be hex-encoded per CIP-25). Max 32 bytes.
        metadata: CIP-25 metadata dict — typically the output of
            :func:`build_cip25_metadata`. Accepts ``{"721": ...}`` or ``{721: ...}``.
        recipient_address: Bech32 address of the wallet that receives the NFT.
        funding_address: Bech32 address that pays the tx fee + NFT min-ADA.
        context: Optional chain context. If omitted a fresh
            :class:`pycardano.OgmiosChainContext` is built from
            ``ogmios_host``/``ogmios_port``.
        ogmios_host: Host of the local Ogmios HTTP+WS endpoint.
        ogmios_port: Port of the local Ogmios endpoint.
        network: ``"mainnet"`` or ``"testnet"`` (preprod / preview).
        min_lovelace_for_nft_utxo: ADA (in lovelace) to attach to the NFT
            output so it satisfies the ledger's min-UTxO floor. Default 1.5 ADA.

    Returns:
        :class:`UnsignedMint` bundle ready for the cold-signer hand-off.

    Raises:
        RuntimeError: If pycardano is unavailable, or tx construction fails.
        ValueError: If ``asset_name`` is empty or > 32 bytes.
    """
    _require_pycardano()

    if not asset_name or len(asset_name.encode("utf-8")) > 32:
        raise ValueError(
            "asset_name must be a non-empty UTF-8 string <= 32 bytes "
            f"(got {len(asset_name.encode('utf-8'))} bytes)"
        )

    from pycardano import (
        Address,
        Asset,
        AssetName,
        AuxiliaryData,
        Metadata,
        MultiAsset,
        NativeScript,
        Network,
        ScriptHash,
        TransactionBuilder,
        TransactionOutput,
        Value,
    )

    if context is None:
        from cardano_checkout.txbuild import make_ogmios_context

        context = make_ogmios_context(
            host=ogmios_host, port=ogmios_port, network=network
        )

    net = Network.MAINNET if network == "mainnet" else Network.TESTNET

    # ------------------------------------------------------------------
    # Assemble the mint MultiAsset
    # ------------------------------------------------------------------
    policy_hash = ScriptHash.from_primitive(bytes.fromhex(policy.policy_id))
    asset_name_obj = AssetName(asset_name.encode("utf-8"))
    asset = Asset()
    asset[asset_name_obj] = 1
    mint_bundle = MultiAsset()
    mint_bundle[policy_hash] = asset

    # ------------------------------------------------------------------
    # Native script + auxiliary data (metadata + script witness)
    # ------------------------------------------------------------------
    native_script = NativeScript.from_cbor(bytes.fromhex(policy.script_cbor_hex))

    metadata_obj = Metadata(_metadata_dict_with_int_keys(metadata))
    aux = AuxiliaryData(metadata_obj)
    # AuxiliaryData in pycardano also carries native_scripts attached to the tx body;
    # the builder below handles native scripts separately via add_minting_script.

    # ------------------------------------------------------------------
    # Addresses
    # ------------------------------------------------------------------
    sender = Address.from_primitive(funding_address)
    recipient = Address.from_primitive(recipient_address)
    if sender.network != net or recipient.network != net:
        raise ValueError(
            f"Address network mismatch: requested {network}, "
            f"sender={sender.network.name}, recipient={recipient.network.name}"
        )

    # ------------------------------------------------------------------
    # Build the transaction
    # ------------------------------------------------------------------
    builder = TransactionBuilder(context)
    builder.add_input_address(sender)

    # Attach mint bundle + policy as a minting script.
    builder.mint = mint_bundle
    builder.native_scripts = [native_script]
    builder.auxiliary_data = aux

    # Output: the newly minted NFT in its own UTxO at the recipient, padded
    # with min-ADA so the ledger accepts it.
    nft_value = Value(min_lovelace_for_nft_utxo, mint_bundle)
    builder.add_output(TransactionOutput(recipient, nft_value))

    # If the policy has a time lock, the mint tx MUST set ttl <= locked_after_slot
    # or the node will reject the witness. Let pycardano pick validity normally,
    # but clamp ttl when a lock slot is set.
    ttl_offset = None
    if policy.locked_after_slot is not None:
        try:
            chain_tip = context.last_block_slot  # type: ignore[attr-defined]
            # Cap at 2 hours or (locked_after_slot - chain_tip), whichever is smaller.
            two_hours_in_slots = 2 * 60 * 60  # ~1 slot/s on mainnet
            ttl_offset = max(
                60, min(two_hours_in_slots, policy.locked_after_slot - chain_tip)
            )
        except Exception:  # pragma: no cover — context without chain tip
            ttl_offset = None

    try:
        tx_body = builder.build(
            change_address=sender,
            auto_ttl_offset=ttl_offset,
            auto_validity_start_offset=-30,
        )
    except Exception as exc:
        raise RuntimeError(f"Failed to build mint tx body: {exc}") from exc

    tx_id = str(tx_body.id)

    summary_lines = [
        f"Mint 1 x {policy.policy_id}.{asset_name}",
        f"  -> recipient: {recipient_address}",
        f"  fees paid by: {funding_address}",
        f"  tx_id (pre-sign): {tx_id}",
        f"  network: {network}",
        f"  required signers: {len(policy.required_signer_hashes)} "
        f"({', '.join(h[:16] + '...' for h in policy.required_signer_hashes) or 'NONE — check policy'})",
    ]
    if policy.locked_after_slot is not None:
        summary_lines.append(
            f"  policy time-lock: slot <= {policy.locked_after_slot}"
        )

    return UnsignedMint(
        tx_id=tx_id,
        tx_body_cbor_hex=tx_body.to_cbor_hex(),
        auxiliary_data_cbor_hex=aux.to_cbor_hex(),
        native_script_cbor_hex=policy.script_cbor_hex,
        required_signer_hashes=list(policy.required_signer_hashes),
        summary="\n".join(summary_lines),
    )


# ---------------------------------------------------------------------------
# Signed-tx submission
# ---------------------------------------------------------------------------


def submit_signed_tx(
    signed_tx_cbor_hex: str,
    context: Optional["ChainContext"] = None,
    ogmios_host: str = "127.0.0.1",
    ogmios_port: int = 1337,
    network: str = "mainnet",
) -> str:
    """Submit a cold-signed transaction to the network via Ogmios.

    The cold signer produces a fully-assembled :class:`pycardano.Transaction`
    — body + witness set + auxiliary data — serialised as CBOR. This
    function deserialises that blob, hands it to Ogmios, and returns the
    tx hash.

    Args:
        signed_tx_cbor_hex: Hex-encoded CBOR of the signed transaction.
        context: Optional chain context; built from ``ogmios_host/port`` if omitted.
        ogmios_host: Host of the Ogmios endpoint.
        ogmios_port: Port of the Ogmios endpoint.
        network: ``"mainnet"`` or ``"testnet"``.

    Returns:
        Transaction hash (hex) — stable identifier for the submitted tx.

    Raises:
        RuntimeError: If pycardano is unavailable, or submission fails.
    """
    _require_pycardano()

    from pycardano import Transaction

    if context is None:
        from cardano_checkout.txbuild import make_ogmios_context

        context = make_ogmios_context(
            host=ogmios_host, port=ogmios_port, network=network
        )

    try:
        tx = Transaction.from_cbor(bytes.fromhex(signed_tx_cbor_hex))
    except Exception as exc:
        raise RuntimeError(f"signed_tx_cbor_hex is not valid transaction CBOR: {exc}") from exc

    try:
        context.submit_tx(tx)  # type: ignore[attr-defined]
    except Exception as exc:
        raise RuntimeError(f"Ogmios rejected the signed tx: {exc}") from exc

    tx_hash = str(tx.id)
    logger.info("[mint] submitted signed tx %s", tx_hash)
    return tx_hash