"""S3 client and operation helpers for the API plugin.

Delegates the underlying boto3 client construction to
``src.services.seaweedfs_service.SeaweedFSService``. Every S3 operation
exposed here goes through ``s3_breaker`` (see ``src.resilience.breakers``)
so a flaky SeaweedFS endpoint cannot retry-storm DE workers.

NOTE: This module runs inside the Airflow api-server process. Importing
``src.resilience.breakers`` is safe — it has no heavy side effects at
import time.
"""

from src.resilience.breakers import breaker_call, s3_breaker
from src.services.seaweedfs_service import SeaweedFSService, validate_s3_path

_service = None


def get_s3_client():
    """Lazy-initialized boto3 S3 client for SeaweedFS (internal endpoint).

    Returned for backward compatibility — prefer the wrapped helpers below
    so all calls are protected by the circuit breaker.
    """
    global _service
    if _service is None:
        _service = SeaweedFSService()
    return _service.s3_client


def ensure_bucket(bucket: str) -> None:
    """Create bucket if it doesn't exist."""
    global _service
    if _service is None:
        _service = SeaweedFSService()
    breaker_call(s3_breaker, _service._ensure_bucket_exists, bucket)


# ---------------------------------------------------------------------------
# Wrapped S3 operations (internal endpoint)
# ---------------------------------------------------------------------------


def put_object(bucket: str, key: str, body, **kwargs):
    """Upload an object. All extra kwargs forward to boto3 ``put_object``."""
    s3 = get_s3_client()
    return breaker_call(s3_breaker, s3.put_object, Bucket=bucket, Key=key, Body=body, **kwargs)


def head_object(bucket: str, key: str):
    s3 = get_s3_client()
    return breaker_call(s3_breaker, s3.head_object, Bucket=bucket, Key=key)


def delete_object(bucket: str, key: str):
    s3 = get_s3_client()
    return breaker_call(s3_breaker, s3.delete_object, Bucket=bucket, Key=key)


def delete_objects(bucket: str, delete: dict):
    s3 = get_s3_client()
    return breaker_call(s3_breaker, s3.delete_objects, Bucket=bucket, Delete=delete)


def list_objects_v2(bucket: str, prefix: str = ""):
    s3 = get_s3_client()
    return breaker_call(s3_breaker, s3.list_objects_v2, Bucket=bucket, Prefix=prefix)


def download_file(bucket: str, key: str, filename: str):
    s3 = get_s3_client()
    return breaker_call(s3_breaker, s3.download_file, Bucket=bucket, Key=key, Filename=filename)


# ---------------------------------------------------------------------------
# Presigned URLs use a separate boto3 client with the public-facing endpoint
# ---------------------------------------------------------------------------

_presign_client = None


def _get_presign_client():
    """Lazy-init boto3 S3 client for presigned URLs (public endpoint, s3v4)."""
    global _presign_client
    if _presign_client is None:
        import boto3
        from api.config import get_presigned_s3_endpoint
        from botocore.client import Config as BotoConfig

        _presign_client = boto3.client(
            "s3",
            endpoint_url=get_presigned_s3_endpoint(),
            config=BotoConfig(signature_version="s3v4"),
            region_name="us-east-1",
        )
    return _presign_client


def generate_presigned_url(operation: str, params: dict, expires_in: int):
    """Generate a presigned URL through the breaker."""
    client = _get_presign_client()
    return breaker_call(
        s3_breaker,
        client.generate_presigned_url,
        operation,
        Params=params,
        ExpiresIn=expires_in,
    )


__all__ = [
    "get_s3_client",
    "ensure_bucket",
    "validate_s3_path",
    "put_object",
    "head_object",
    "delete_object",
    "delete_objects",
    "list_objects_v2",
    "download_file",
    "generate_presigned_url",
]