"""S3 object storage — presigned PUT/GET helpers for user uploads.

Single backend (AWS S3) because prod EC2 already runs in the same
account/region as the bucket (see migration 194) and boto3 picks the
``helix-prod-ec2-role`` instance profile up automatically — no access
key lives in env.

The bucket name + region + key prefix come from ``platform_settings``
(rows ``storage.s3.bucket``, ``storage.s3.region``,
``storage.s3.key_prefix``) so admins can change targets without code
changes. A runtime guard force-overrides the prefix to ``"dev/"``
whenever ``APP_ENV != "production"`` — a local dev environment
pointed at the same bucket can never write into the prod top-level
namespace (the dev/ prefix has a 7-day expiration lifecycle so it
self-cleans).

Endpoints in ``app.storage_router`` consume these helpers; nothing
else in the codebase should be talking to S3 directly for upload —
keep the surface narrow.
"""
from __future__ import annotations

import logging
from typing import Final, cast
from urllib.parse import quote

import boto3
from botocore.config import Config
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings as app_settings
from app.integrations.models import PlatformSetting

logger = logging.getLogger(__name__)

_PROD_ENV: Final[str] = "production"
_FORCED_DEV_PREFIX: Final[str] = "dev/"

# Cache one boto3 client per region. boto3 clients are thread-safe and
# cheap to keep around; rebuilding per request burns ~50ms on metadata
# lookups.
_S3_CLIENT_CACHE: dict[str, boto3.client] = {}  # type: ignore[type-arg]


class StorageNotConfiguredError(RuntimeError):
    """Raised when platform_settings is missing storage.s3.bucket/region."""


def _normalise_prefix(value: str | None) -> str:
    if not value:
        return ""
    stripped = value.strip().lstrip("/")
    if not stripped:
        return ""
    if not stripped.endswith("/"):
        stripped += "/"
    return stripped


async def resolve_storage_config(db: AsyncSession) -> dict[str, str]:
    """Read storage.s3.* from platform_settings + apply dev-prefix guard.

    Returns ``{"bucket": ..., "region": ..., "key_prefix": ...}``.
    Raises :class:`StorageNotConfiguredError` if bucket or region are
    missing — caller should surface this as a 500 to the operator so
    they know to seed the rows.
    """
    bucket_row = await db.get(PlatformSetting, "storage.s3.bucket")
    region_row = await db.get(PlatformSetting, "storage.s3.region")
    prefix_row = await db.get(PlatformSetting, "storage.s3.key_prefix")

    bucket = (bucket_row.value if bucket_row else None) or ""
    region = (region_row.value if region_row else None) or ""
    prefix = (prefix_row.value if prefix_row else None) or ""

    if not bucket or not region:
        raise StorageNotConfiguredError(
            "platform_settings is missing storage.s3.bucket or "
            "storage.s3.region — seed via migration 194 or admin UI."
        )

    prefix = _normalise_prefix(str(prefix))

    # Dev guard: never let a non-production deploy write into the same
    # top-level namespace as prod. Force ``dev/`` regardless of what
    # the DB row says.
    if app_settings.APP_ENV != _PROD_ENV:
        prefix = _FORCED_DEV_PREFIX

    return {
        "bucket": str(bucket),
        "region": str(region),
        "key_prefix": prefix,
    }


def _get_s3_client(region: str):  # noqa: ANN202 — botocore client type is opaque
    client = _S3_CLIENT_CACHE.get(region)
    if client is None:
        # Pin to the regional endpoint + virtual-hosted addressing.
        # Without ``endpoint_url`` boto3 happily returns the legacy
        # global host ``<bucket>.s3.amazonaws.com`` for a region-locked
        # bucket, and S3 answers with a 307 Temporary Redirect to the
        # regional host. A browser-issued PUT against such a presigned
        # URL fails twice over: (a) the redirect drops CORS headers so
        # the browser blocks the second hop, (b) the SigV4 signature is
        # bound to the original host and won't validate against the
        # redirected one. Explicit regional endpoint resolves both —
        # the first PUT lands on the right host with a matching
        # signature, no redirect.
        client = boto3.client(
            "s3",
            region_name=region,
            endpoint_url=f"https://s3.{region}.amazonaws.com",
            config=Config(
                signature_version="s3v4",
                s3={"addressing_style": "virtual"},
                retries={"max_attempts": 3, "mode": "standard"},
            ),
        )
        _S3_CLIENT_CACHE[region] = client
    return client


async def presign_put(
    db: AsyncSession,
    *,
    object_key: str,
    content_type: str,
    size_bytes: int,
    checksum_sha256: str,
    expires_in: int = 600,
) -> dict[str, object]:
    """Generate a single-create, size-enforcing presigned PUT.

    ``object_key`` is the already-resolved FULL key. SigV4 binds the exact
    length, type, checksum and ``If-None-Match: *`` headers. S3 therefore
    rejects both differently-sized bodies and every replay after the first
    successful create. Never store or log the returned URL or headers.
    """
    import asyncio

    cfg = await resolve_storage_config(db)
    # Both the client build and the signing call are blocking boto3 work: on
    # first use with an EC2 instance-profile role, botocore resolves creds
    # lazily via IMDS HTTP round-trips (with retries), which would otherwise
    # block the whole event loop — stalling every concurrent request — on the
    # first presign per region after boot. Run it in a worker thread, same as
    # ``get_object_bytes`` / ``stat_object``.
    def _sign() -> dict[str, object]:
        client = _get_s3_client(cfg["region"])
        url = client.generate_presigned_url(
            "put_object",
            Params={
                "Bucket": cfg["bucket"],
                "Key": object_key,
                "ContentLength": size_bytes,
                "ContentType": content_type,
                "ChecksumSHA256": checksum_sha256,
                "IfNoneMatch": "*",
            },
            ExpiresIn=expires_in,
        )
        return {
            "url": url,
            "headers": {
                "Content-Type": content_type,
                "If-None-Match": "*",
                "x-amz-checksum-sha256": checksum_sha256,
            },
        }

    signed = await asyncio.to_thread(_sign)
    return {
        **signed,
        "object_key": object_key,
        "bucket": cfg["bucket"],
        "expires_in": expires_in,
    }


async def delete_object(db: AsyncSession, *, object_key: str) -> None:
    """Idempotently delete one exact object key from the configured bucket."""
    import asyncio

    cfg = await resolve_storage_config(db)

    def _delete() -> None:
        client = _get_s3_client(cfg["region"])
        client.delete_object(Bucket=cfg["bucket"], Key=object_key)

    await asyncio.to_thread(_delete)


async def delete_object_if_unchanged(
    *,
    bucket: str,
    region: str,
    object_key: str,
    expected_size_bytes: int,
    expected_content_type: str,
    expected_checksum_sha256: str,
) -> str:
    """Conditionally remove the exact uploaded object without a DB session.

    Returns ``"deleted"``, ``"missing"``, or ``"changed"``. The HEAD,
    conditional delete, and verification HEAD all run outside any database
    transaction. ``If-Match`` closes the replacement window between the first
    HEAD and DELETE; the second HEAD detects a new current object appearing
    after the delete. Callers must retry ``"changed"`` rather than marking the
    durable grant deleted.
    """
    import asyncio

    from botocore.exceptions import ClientError

    def _error_code(exc: ClientError) -> str:
        return str(exc.response.get("Error", {}).get("Code", ""))

    def _head(client) -> dict[str, object] | None:  # noqa: ANN001
        try:
            return cast(
                dict[str, object],
                client.head_object(
                    Bucket=bucket,
                    Key=object_key,
                    ChecksumMode="ENABLED",
                ),
            )
        except ClientError as exc:
            if _error_code(exc) in ("NoSuchKey", "404", "NotFound"):
                return None
            raise

    def _delete() -> str:
        client = _get_s3_client(region)
        current = _head(client)
        if current is None:
            return "missing"
        actual_size = current.get("ContentLength")
        if (
            not isinstance(actual_size, int)
            or actual_size != expected_size_bytes
            or str(current.get("ContentType") or "") != expected_content_type
            or str(current.get("ChecksumSHA256") or "")
            != expected_checksum_sha256
        ):
            return "changed"
        etag = str(current.get("ETag") or "")
        if not etag:
            return "changed"
        try:
            client.delete_object(
                Bucket=bucket,
                Key=object_key,
                IfMatch=etag,
            )
        except ClientError as exc:
            if _error_code(exc) in ("NoSuchKey", "404", "NotFound"):
                return "missing"
            if _error_code(exc) in ("PreconditionFailed", "412"):
                return "changed"
            raise
        return "deleted" if _head(client) is None else "changed"

    return await asyncio.to_thread(_delete)


async def presign_get(
    db: AsyncSession,
    *,
    object_key: str,
    expires_in: int = 600,
    download_filename: str | None = None,
) -> str:
    """Generate a presigned GET URL for an object already in S3.

    ``object_key`` is the FULL key as stored in the DB (includes any
    prefix that was active at upload time). Callers must enforce
    per-resource ACL before issuing this — the URL bypasses Helix
    auth once minted.

    ``download_filename`` signs ``ResponseContentDisposition: attachment``
    so the browser saves the file under that name instead of rendering it
    in a tab. Pass it for a button labelled "Download" — without it an
    image or a video opens inline, which is not what that word means, and
    inside an in-app webview the new tab may not open at all.
    """
    import asyncio

    cfg = await resolve_storage_config(db)

    params: dict[str, str] = {"Bucket": cfg["bucket"], "Key": object_key}
    if download_filename:
        # RFC 6266: an ASCII fallback plus a UTF-8 form, because the name
        # can be CJK and a bare ``filename=`` must stay ASCII.
        safe = _ascii_disposition_name(download_filename)
        quoted = quote(download_filename, safe="")
        params["ResponseContentDisposition"] = (
            f'attachment; filename="{safe}"; filename*=UTF-8\'\'{quoted}'
        )

    # Blocking boto3 (client build + signing) — off the event loop, same as
    # ``presign_put`` / ``get_object_bytes``.
    def _sign() -> str:
        client = _get_s3_client(cfg["region"])
        return client.generate_presigned_url(
            "get_object",
            Params=params,
            ExpiresIn=expires_in,
        )

    return await asyncio.to_thread(_sign)


def _ascii_disposition_name(raw: str) -> str:
    """An ASCII-safe ``filename=`` value, never empty.

    Quotes and backslashes would terminate the header value early, and
    non-ASCII bytes are illegal there; both are replaced rather than
    stripped so the name keeps its shape.
    """
    cleaned = "".join(
        char if 32 <= ord(char) < 127 and char not in '"\\' else "_"
        for char in raw
    )
    return cleaned.strip() or "download"


async def get_object_bytes(
    db: AsyncSession,
    *,
    object_key: str,
) -> bytes | None:
    """Server-side read of an object's bytes from the storage bucket.

    Read-counterpart to :func:`presign_get`: ``object_key`` is the FULL
    key (NO prefix is prepended), so callers that store release/manifest
    artifacts under a fixed top-level namespace (e.g. ``cli-dist/``) read
    the exact key they wrote, independent of the per-env ``dev/`` upload
    prefix. Returns ``None`` when the object is absent (S3 ``NoSuchKey``
    / 404) so callers can render a graceful "unavailable" state rather
    than 500. Other boto/transport errors propagate.

    ``get_object`` is a blocking boto3 call — runs in a worker thread.
    """
    import asyncio

    from botocore.exceptions import ClientError

    cfg = await resolve_storage_config(db)
    client = _get_s3_client(cfg["region"])

    def _get() -> bytes | None:
        try:
            resp = client.get_object(Bucket=cfg["bucket"], Key=object_key)
        except ClientError as exc:
            code = exc.response.get("Error", {}).get("Code", "")
            if code in ("NoSuchKey", "404", "NotFound"):
                return None
            raise
        return resp["Body"].read()

    return await asyncio.to_thread(_get)


async def stat_object(
    db: AsyncSession,
    *,
    object_key: str,
) -> dict[str, object] | None:
    """HEAD an object — returns verified size, type, and SHA-256 or ``None``.

    Register-after-presign flows must not trust the client's claimed
    object existence/size: the browser POSTs directly to S3, so the only
    authoritative record of what actually landed is S3 itself. ``object_key``
    is the FULL key as stored in the DB. Returns ``None`` when the object
    does not exist; other boto/transport errors propagate.
    """
    import asyncio

    from botocore.exceptions import ClientError

    cfg = await resolve_storage_config(db)
    client = _get_s3_client(cfg["region"])

    def _head() -> dict[str, object] | None:
        try:
            resp = client.head_object(
                Bucket=cfg["bucket"], Key=object_key, ChecksumMode="ENABLED",
            )
        except ClientError as exc:
            code = exc.response.get("Error", {}).get("Code", "")
            if code in ("NoSuchKey", "404", "NotFound"):
                return None
            raise
        return {
            "size_bytes": int(resp.get("ContentLength") or 0),
            "content_type": str(resp.get("ContentType") or ""),
            "checksum_sha256": str(resp.get("ChecksumSHA256") or ""),
        }

    return await asyncio.to_thread(_head)


async def put_object_bytes(
    db: AsyncSession,
    *,
    relative_key: str,
    content: bytes,
    content_type: str,
) -> str:
    """Server-side upload of an in-memory payload to the storage bucket.

    Built for flows where the file already transits the backend (e.g.
    approval supplement uploads, which must ALSO go to Feishu) — the
    browser-direct presigned-PUT path in ``app.storage_router`` doesn't
    fit because the backend needs the bytes anyway and a second browser
    upload would double the user's bandwidth cost.

    ``relative_key`` is prefix-less (same contract as
    :func:`presign_put`); returns the FULL object key (with prefix) to
    store in the DB. ``put_object`` is a blocking boto3 call — runs in
    a worker thread so a 50 MB body doesn't stall the event loop.
    """
    import asyncio

    cfg = await resolve_storage_config(db)
    full_key = cfg["key_prefix"] + relative_key.lstrip("/")
    client = _get_s3_client(cfg["region"])

    def _put() -> None:
        client.put_object(
            Bucket=cfg["bucket"],
            Key=full_key,
            Body=content,
            ContentType=content_type,
        )

    await asyncio.to_thread(_put)
    return full_key