cardano-checkout-py/cardano_checkout/ipfs.py

"""Minimal IPFS client — upload + pin via kubo's HTTP API.

Designed for the ``chromaticcraft`` shape: a small local kubo daemon
runs alongside the web app, accepts uploads from end users (e.g. Abby
uploading a photo of a finished custom order), pins locally for fast
serving, and optionally mirrors pins to a second remote node
(Lucy-on-LAN) for archival redundancy.

No IPFS libraries are imported — just httpx against the kubo REST API
(v0). Keeps the SDK surface minimal.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Optional

import httpx

logger = logging.getLogger(__name__)


@dataclass
class IPFSClient:
    """Kubo-compatible IPFS client.

    Attributes:
        api_url: Base URL of the kubo HTTP API (default ``http://127.0.0.1:5001``).
        timeout: Per-request timeout in seconds (default 60 — uploads can be slow).
        mirror_api_urls: Optional list of additional kubo endpoints to
            ``pin add`` the CID on after a successful primary pin. Use this
            to mirror to Lucy or any other archival node.
    """

    api_url: str = "http://127.0.0.1:5001"
    timeout: float = 60.0
    mirror_api_urls: list[str] = None  # type: ignore[assignment]

    def __post_init__(self) -> None:
        if self.mirror_api_urls is None:
            self.mirror_api_urls = []

    async def add(self, data: bytes, filename: str = "upload") -> str:
        """Upload bytes and pin them locally.

        Args:
            data: Raw bytes to add.
            filename: Logical name used by clients browsing the DAG
                (doesn't affect the CID).

        Returns:
            CID (base58, v0 or base32 v1 depending on kubo defaults).

        Raises:
            RuntimeError: If the daemon is unreachable or returns a non-2xx.
        """
        url = f"{self.api_url.rstrip('/')}/api/v0/add"
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            resp = await client.post(
                url,
                files={"file": (filename, data, "application/octet-stream")},
                params={"pin": "true", "cid-version": "1"},
            )
        if resp.status_code >= 400:
            raise RuntimeError(f"ipfs add {resp.status_code}: {resp.text[:200]}")
        # kubo's /add streams NDJSON; each line is one {Name, Hash, Size}.
        # For a single file upload the last line carries the wrapping CID.
        last_cid: Optional[str] = None
        for line in resp.text.strip().splitlines():
            if '"Hash"' in line:
                import json
                obj = json.loads(line)
                last_cid = obj.get("Hash")
        if not last_cid:
            raise RuntimeError(f"ipfs add: no CID in response: {resp.text[:200]}")

        # Mirror pins (best effort — a mirror failure should not poison the primary upload).
        for mirror in self.mirror_api_urls:
            try:
                await self._pin_on(mirror, last_cid)
            except Exception as exc:
                logger.warning("[ipfs] mirror pin to %s failed for %s: %s", mirror, last_cid, exc)

        return last_cid

    async def _pin_on(self, api_url: str, cid: str) -> None:
        """Pin an existing CID on a remote kubo node."""
        url = f"{api_url.rstrip('/')}/api/v0/pin/add"
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            resp = await client.post(url, params={"arg": cid})
        if resp.status_code >= 400:
            raise RuntimeError(f"pin/add {resp.status_code}: {resp.text[:200]}")


async def pin_bytes(
    data: bytes,
    api_url: str = "http://127.0.0.1:5001",
    mirror_api_urls: Optional[list[str]] = None,
    filename: str = "upload",
) -> str:
    """Convenience wrapper: one-shot upload + pin (+ optional mirror).

    Returns the CID.
    """
    client = IPFSClient(api_url=api_url, mirror_api_urls=mirror_api_urls or [])
    return await client.add(data, filename=filename)