dtf-system/app/adapters.py

"""Local-only composition root. No production provider implementations/imports."""
import hashlib
import hmac
import json
import os
from typing import Mapping, NamedTuple, Protocol
from urllib.parse import urlparse
import boto3
from botocore.config import Config
from botocore.exceptions import ClientError

def require_runtime():
    """Validate the supported local and R2-backed deployment modes.

    Real payment, freight, ERP, and WhatsApp providers are deliberately not
    enabled yet. A non-local deployment keeps those adapters fake and disables
    checkout until their audited implementations are added.
    """
    environment = os.environ.get('APP_ENV', 'local')
    for name in ('FREIGHT', 'WHATSAPP'):
        if os.environ.get(f'{name}_ADAPTER') != 'fake':
            raise RuntimeError(f'{name} must use the currently supported fake adapter')
    tiny = os.environ.get('TINY_ADAPTER')
    if tiny == 'tiny':
        from .tiny import required_settings
        for name in required_settings():
            if not os.environ.get(name):
                raise RuntimeError(f'{name} is required for the Tiny adapter')
    elif tiny != 'fake':
        raise RuntimeError('TINY must use the fake or tiny adapter')
    # Mercado Pago is selectable only with its credentials present; it has not
    # yet passed the sandbox flows, so production preflight still blocks it.
    payment = os.environ.get('PAYMENT_ADAPTER')
    if payment == 'mercadopago':
        for name in ('MP_ACCESS_TOKEN', 'MP_WEBHOOK_SECRET'):
            if not os.environ.get(name):
                raise RuntimeError(f'{name} is required for the Mercado Pago adapter')
    elif payment != 'fake':
        raise RuntimeError('PAYMENT must use the fake or mercadopago adapter')
    if environment == 'local':
        if os.environ.get('STORAGE_ADAPTER') != 's3-local':
            raise RuntimeError('Local runtime requires local S3 storage')
        for name in ('S3_ENDPOINT', 'S3_PUBLIC_ENDPOINT'):
            endpoint = urlparse(os.environ[name])
            if endpoint.scheme != 'http' or endpoint.hostname not in ('storage', 'localhost', '127.0.0.1'):
                raise RuntimeError(f'{name} must point to local MinIO')
        return
    if environment != 'production':
        raise RuntimeError('APP_ENV must be local or production')
    if os.environ.get('STORAGE_ADAPTER') != 's3-r2':
        raise RuntimeError('Production runtime requires R2 storage')
    for name in ('S3_ENDPOINT', 'S3_PUBLIC_ENDPOINT'):
        endpoint = urlparse(os.environ[name])
        if endpoint.scheme != 'https' or not (endpoint.hostname or '').endswith('.r2.cloudflarestorage.com'):
            raise RuntimeError(f'{name} must be a Cloudflare R2 S3 API endpoint')
    for name in ('AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY'):
        if not os.environ.get(name):
            raise RuntimeError(f'{name} is required for R2')


# Compatibility alias for local-only callers outside the active runtime.
require_local = require_runtime

class PaymentEvent(NamedTuple):
    """One provider notification, normalised.

    `event_id` identifies the delivery and makes it idempotent. `reference` is
    our quote id, echoed back by the provider. `amount_cents` is what the
    provider says was actually paid, which the service compares against the
    approved total before it will create an order.
    """
    event_id: str
    reference: str
    status: str          # 'approved' | 'rejected' | 'pending' | 'refunded'
    amount_cents: int | None
    raw: dict


class PaymentAdapter(Protocol):
    def create(self, quote_id: str, total_cents: int, customer: dict, method: dict | None = None) -> dict:
        """Start a payment. Must be idempotent on quote_id: a retry after a
        timeout has to return the existing payment, never charge twice."""

    def verify(self, headers: Mapping[str, str], body: bytes, query: Mapping[str, str] | None = None) -> bool:
        """Whether this delivery genuinely came from the provider."""

    def parse(self, body: bytes, query: Mapping[str, str] | None = None) -> PaymentEvent | None:
        """Normalise a verified delivery, or None if it is not about a payment."""


class FakePayment:
    """Local stand-in with a real signature scheme, so the webhook path is
    exercised end to end rather than waiting for a provider account.

    Signs the body with HMAC-SHA256 under PAYMENT_WEBHOOK_SECRET. A real adapter
    replaces verify() and parse() with the provider's own scheme; nothing else in
    the service changes.
    """

    name = 'fake'
    header = 'x-payment-signature'

    def _secret(self) -> bytes | None:
        secret = os.environ.get('PAYMENT_WEBHOOK_SECRET', '')
        return secret.encode() if secret else None

    def create(self, quote_id: str, total_cents: int, customer: dict, method: dict | None = None) -> dict:
        kind = (method or {}).get('type', 'pix')
        return {'provider': 'fake', 'id': f'local-{quote_id}-{kind}',
                'status': 'pending', 'total_cents': total_cents}

    def sign(self, body: bytes) -> str:
        secret = self._secret()
        if secret is None:
            raise RuntimeError('PAYMENT_WEBHOOK_SECRET is not configured')
        return hmac.new(secret, body, hashlib.sha256).hexdigest()

    def verify(self, headers, body: bytes, query=None) -> bool:
        # No configured secret means nothing can be verified, so nothing is
        # accepted. A guessable default would let anyone forge an approval and
        # create an order that was never paid for.
        if self._secret() is None:
            return False
        supplied = headers.get(self.header) or headers.get(self.header.title()) or ''
        return hmac.compare_digest(supplied, self.sign(body))

    def parse(self, body: bytes, query=None):
        try:
            data = json.loads(body)
        except ValueError:
            return None
        if not isinstance(data, dict) or 'event_id' not in data:
            return None
        return PaymentEvent(event_id=str(data['event_id']),
                            reference=str(data.get('reference', '')),
                            status=str(data.get('status', 'pending')),
                            amount_cents=data.get('amount_cents'),
                            raw=data)

    # The local development checkout still needs a direct "it is paid" path.
    def pay(self, quote_id: str, total_cents: int) -> dict:
        return {'provider': 'fake', 'id': f'local-{quote_id}',
                'status': 'paid', 'total_cents': total_cents}

class FreightAdapter(Protocol):
    def quote(self, service: str, postal_code: str) -> dict: ...

class FakeFreight:
    def quote(self, service: str, postal_code: str) -> dict:
        if service == 'pickup':
            return {'provider': 'fake', 'service': 'pickup', 'total_cents': 0, 'postal_code': ''}
        if service != 'mock-standard' or len(postal_code) != 8 or not postal_code.isascii() or not postal_code.isdigit():
            raise ValueError('Select pickup or a mock quote with an eight-digit CEP')
        cents = int(os.environ.get('MOCK_FREIGHT_CENTS', '1500'))
        if cents < 0:
            raise ValueError('Invalid mock freight configuration')
        return {'provider': 'fake', 'service': service, 'postal_code': postal_code,
                'total_cents': cents, 'description': 'Local simulated freight'}

class EventAdapter(Protocol):
    def deliver(self, event_key: str, payload: dict) -> dict: ...

class FakeTiny:
    def deliver(self, event_key: str, payload: dict) -> dict:
        return {'provider': 'fake-tiny', 'reference': f"LOCAL-{payload['number']}",
                'event_key': event_key, 'status': 'recorded-locally'}

class FakeWhatsApp:
    def deliver(self, event_key: str, payload: dict) -> dict:
        return {'provider': 'fake-whatsapp', 'event_key': event_key,
                'event': payload['event'], 'status': 'recorded-locally'}

class ObjectStorage(Protocol):
    def begin(self, key: str) -> str: ...
    def parts(self, key: str, upload_id: str) -> list: ...
    def part_url(self, key: str, upload_id: str, part: int, size: int) -> str: ...
    def complete(self, key: str, upload_id: str, parts: list): ...
    def size(self, key: str) -> int: ...
    def download(self, key: str, name: str) -> str: ...
    def fetch(self, key: str, path: str): ...
    def store(self, key: str, path: str, content_type: str): ...
    def health(self): ...
    def discard(self, key: str, upload_id: str, complete: bool): ...

class LocalS3Storage:
    def __init__(self):
        config = Config(signature_version='s3v4', s3={'addressing_style': 'path'},
                        connect_timeout=3, read_timeout=10, retries={'max_attempts': 2})
        self.client = boto3.client('s3', endpoint_url=os.environ['S3_ENDPOINT'], config=config)
        self.public = boto3.client('s3', endpoint_url=os.environ['S3_PUBLIC_ENDPOINT'], config=config)
        self.bucket = os.environ['S3_BUCKET']

    def initialize(self):
        try:
            self.client.head_bucket(Bucket=self.bucket)
        except ClientError as exc:
            if exc.response['ResponseMetadata']['HTTPStatusCode'] != 404:
                raise
            self.client.create_bucket(Bucket=self.bucket)
        self.client.put_bucket_lifecycle_configuration(Bucket=self.bucket, LifecycleConfiguration={
            'Rules': [{'ID': 'local-artwork-retention', 'Status': 'Enabled', 'Filter': {'Prefix': ''},
                       'Expiration': {'Days': 30}, 'AbortIncompleteMultipartUpload': {'DaysAfterInitiation': 1}}]})

    def health(self):
        self.client.head_bucket(Bucket=self.bucket)

    def begin(self, key):
        return self.client.create_multipart_upload(Bucket=self.bucket, Key=key,
            ContentType='application/octet-stream')['UploadId']

    def parts(self, key, upload_id):
        result = []
        for page in self.client.get_paginator('list_parts').paginate(
                Bucket=self.bucket, Key=key, UploadId=upload_id):
            result.extend(page.get('Parts', []))
        return result

    def part_url(self, key, upload_id, part, size):
        return self.public.generate_presigned_url('upload_part', Params={
            'Bucket': self.bucket, 'Key': key, 'UploadId': upload_id, 'PartNumber': part,
            'ContentLength': size}, ExpiresIn=900)

    def complete(self, key, upload_id, parts):
        self.client.complete_multipart_upload(Bucket=self.bucket, Key=key, UploadId=upload_id,
            MultipartUpload={'Parts': [{'PartNumber': p['PartNumber'], 'ETag': p['ETag']} for p in parts]})

    def size(self, key):
        return self.client.head_object(Bucket=self.bucket, Key=key)['ContentLength']

    def fetch(self, key, path):
        """Copy a stored object to a local file (the worker's scratch space)."""
        self.client.download_file(self.bucket, key, path)

    def store(self, key, path, content_type):
        """Upload a file the service generated itself, such as a print file."""
        self.client.upload_file(path, self.bucket, key, ExtraArgs={'ContentType': content_type})

    def download(self, key, name):
        from urllib.parse import quote
        return self.public.generate_presigned_url('get_object', Params={
            'Bucket': self.bucket, 'Key': key,
            'ResponseContentDisposition': "attachment; filename*=UTF-8''" + quote(name, safe=''),
            'ResponseContentType': 'application/octet-stream'}, ExpiresIn=300)

    def discard(self, key, upload_id, complete):
        if not complete:
            try:
                self.client.abort_multipart_upload(Bucket=self.bucket, Key=key, UploadId=upload_id)
            except ClientError as exc:
                if exc.response['ResponseMetadata']['HTTPStatusCode'] != 404:
                    raise
        self.client.delete_object(Bucket=self.bucket, Key=key)