dtf-system/app/adapters.py

"""Local-only composition root. No production provider implementations/imports."""
import os
from typing import 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 ('PAYMENT', 'FREIGHT', 'TINY', 'WHATSAPP'):
        if os.environ.get(f'{name}_ADAPTER') != 'fake':
            raise RuntimeError(f'{name} must use the currently supported fake 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 PaymentAdapter(Protocol):
    def pay(self, quote_id: str, total_cents: int) -> dict: ...

class FakePayment:
    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 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 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)