from __future__ import annotations

import asyncio
import hashlib
import logging
import os
import re
import threading
import time
import weakref
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Mapping

from app.utils.logging import log_flow_event

logger = logging.getLogger(__name__)

_QUEUE_SIZE = 64
_MAX_WAV_BYTES = 5 * 1024 * 1024
_MAX_OBJECT_NAME_BYTES = 512
_MAX_PART_CHARS = 64
_SAFE_PART_RE = re.compile(r"[^A-Za-z0-9_.-]+")
_OCI_AUTH_MODE_LOCAL = "local"
_OCI_AUTH_MODE_OKE_WORKLOAD_IDENTITY = "oke_workload_identity"


@dataclass(frozen=True, slots=True)
class STTAudioUploadItem:
    object_name: str
    wav_bytes: bytes
    req_id: str
    message_id: str
    session_id: str
    metadata: Mapping[str, Any]
    logger: Any | None = None
    timeline: Any | None = None


@dataclass(frozen=True, slots=True)
class OCIUploadConfig:
    auth_mode: str
    region: str
    bucket: str
    namespace: str
    user: str = ""
    key_content: str = ""
    fingerprint: str = ""
    tenancy: str = ""


def _safe_path_part(value: Any, *, fallback: str) -> str:
    raw = str(value or "").strip()
    if not raw:
        raw = fallback
    safe = _SAFE_PART_RE.sub("_", raw).strip("._-") or fallback
    if len(safe) <= _MAX_PART_CHARS:
        return safe

    digest = hashlib.sha1(safe.encode("utf-8")).hexdigest()[:12]
    prefix_len = max(1, _MAX_PART_CHARS - len(digest) - 1)
    return f"{safe[:prefix_len].rstrip('._-')}-{digest}"


def _object_name_within_limit(object_name: str) -> str:
    if len(object_name.encode("utf-8")) <= _MAX_OBJECT_NAME_BYTES:
        return object_name

    digest = hashlib.sha1(object_name.encode("utf-8")).hexdigest()
    today = datetime.now().strftime("%Y-%m-%d")
    return f"{today}/oversize/{digest}.wav"


def build_stt_vad_object_name(
    *,
    session_id: Any,
    message_id: Any,
    req_id: Any,
    now: datetime | None = None,
) -> str:
    dt = now or datetime.now()
    session_part = _safe_path_part(session_id, fallback="unknown-session")
    message_part = _safe_path_part(message_id, fallback="unknown-message")
    object_name = (
        f"{dt:%Y-%m-%d}/{session_part}/{message_part}.wav"
    )
    return _object_name_within_limit(object_name)


def _metadata_value(source: Any, *keys: str) -> str:
    if isinstance(source, Mapping):
        lowered = {str(key).lower(): value for key, value in source.items()}
        for key in keys:
            value = source.get(key)
            if value not in (None, ""):
                return str(value).strip()
            value = lowered.get(str(key).lower())
            if value not in (None, ""):
                return str(value).strip()
        return ""

    for key in keys:
        value = getattr(source, key, None)
        if value not in (None, ""):
            return str(value).strip()
    return ""


def _oci_upload_config_from_env() -> OCIUploadConfig | None:
    auth_mode = (
        _OCI_AUTH_MODE_LOCAL
        if os.getenv("OCI_AUTH_MODE", "").strip().lower() == _OCI_AUTH_MODE_LOCAL
        else _OCI_AUTH_MODE_OKE_WORKLOAD_IDENTITY
    )
    shared_values = {
        "region": os.getenv("BUCKET_REGION", "").strip(),
        "bucket": os.getenv("BUCKET_NAME", "").strip(),
        "namespace": os.getenv("BUCKET_NAMESPACE", "").strip(),
    }
    if not all(shared_values.values()):
        return None

    if auth_mode != _OCI_AUTH_MODE_LOCAL:
        return OCIUploadConfig(auth_mode=auth_mode, **shared_values)

    local_values = {
        "user": os.getenv("BUCKET_USER_ID", "").strip(),
        "key_content": os.getenv("BUCKET_PRIVATE_KEY", "").replace("\\n", "\n").strip(),
        "fingerprint": os.getenv("BUCKET_FINGERPRINT", "").strip(),
        "tenancy": os.getenv("BUCKET_TENANCY_ID", "").strip(),
    }
    if not all(local_values.values()):
        return None
    return OCIUploadConfig(auth_mode=auth_mode, **shared_values, **local_values)


_OCI_CLIENT: Any | None = None
_OCI_CLIENT_CONFIG: OCIUploadConfig | None = None
_OCI_CLIENT_LOCK = threading.Lock()
# ObjectStorageClient owns a requests Session and must not be used concurrently
# by the STT and entire-call upload workers.
_OCI_UPLOAD_LOCK = threading.Lock()


def _oci_client(config: OCIUploadConfig) -> Any:
    global _OCI_CLIENT, _OCI_CLIENT_CONFIG

    with _OCI_CLIENT_LOCK:
        if _OCI_CLIENT is not None and _OCI_CLIENT_CONFIG == config:
            return _OCI_CLIENT

        from oci.object_storage import ObjectStorageClient

        if config.auth_mode == _OCI_AUTH_MODE_LOCAL:
            client_config = {
                "user": config.user,
                "key_content": config.key_content,
                "fingerprint": config.fingerprint,
                "tenancy": config.tenancy,
                "region": config.region,
            }
            _OCI_CLIENT = ObjectStorageClient(config=client_config)
        else:
            from oci.auth.signers import get_oke_workload_identity_resource_principal_signer

            signer = get_oke_workload_identity_resource_principal_signer()
            _OCI_CLIENT = ObjectStorageClient(
                config={"region": config.region},
                signer=signer,
            )
        _OCI_CLIENT_CONFIG = config
        session = getattr(getattr(_OCI_CLIENT, "base_client", None), "session", None)
        if session is not None:
            session.trust_env = False
        return _OCI_CLIENT


def _reset_oci_client() -> None:
    """Discard a client whose HTTP connection pool may be in an invalid state."""
    global _OCI_CLIENT, _OCI_CLIENT_CONFIG

    with _OCI_CLIENT_LOCK:
        client = _OCI_CLIENT
        _OCI_CLIENT = None
        _OCI_CLIENT_CONFIG = None

    session = getattr(getattr(client, "base_client", None), "session", None)
    close = getattr(session, "close", None)
    if callable(close):
        try:
            close()
        except Exception:
            logger.debug("failed closing discarded OCI client session", exc_info=True)


def _put_object_sync(item: STTAudioUploadItem, config: OCIUploadConfig) -> Any:
    with _OCI_UPLOAD_LOCK:
        client = _oci_client(config)
        return client.put_object(
            namespace_name=config.namespace,
            bucket_name=config.bucket,
            object_name=item.object_name,
            put_object_body=item.wav_bytes,
        )


class _STTAudioUploadWorker:
    def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
        self._loop = loop
        self._queue: asyncio.Queue[STTAudioUploadItem] = asyncio.Queue(maxsize=_QUEUE_SIZE)
        self._task: asyncio.Task[None] | None = None

    def enqueue(self, item: STTAudioUploadItem) -> bool:
        if self._task is None or self._task.done():
            self._task = self._loop.create_task(
                self._run(),
                name="stt-audio-upload-worker",
            )
        try:
            self._queue.put_nowait(item)
            return True
        except asyncio.QueueFull:
            return False

    async def _run(self) -> None:
        while True:
            item = await self._queue.get()
            try:
                await _upload_item(item)
            except Exception:
                log_flow_event(
                    item.logger or logger,
                    "stt_audio_upload_failed",
                    request_id=item.req_id,
                    message_id=item.message_id,
                    session_id=item.session_id,
                    object_name=item.object_name,
                    bytes=len(item.wav_bytes),
                    **_log_metadata_fields(item),
                )
                _emit_timeline(
                    item,
                    "stt_audio_upload_failed",
                    message_id=item.message_id,
                    object_name=item.object_name,
                    bytes=len(item.wav_bytes),
                    **_log_metadata_fields(item),
                )
                (item.logger or logger).exception(
                    "[stt][audio_upload_failed] req_id=%s object_name=%s",
                    item.req_id,
                    item.object_name,
                )
            finally:
                self._queue.task_done()


_WORKERS: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, _STTAudioUploadWorker] = (
    weakref.WeakKeyDictionary()
)
_WORKERS_LOCK = threading.Lock()


def _worker_for_loop(loop: asyncio.AbstractEventLoop) -> _STTAudioUploadWorker:
    with _WORKERS_LOCK:
        worker = _WORKERS.get(loop)
        if worker is None:
            worker = _STTAudioUploadWorker(loop)
            _WORKERS[loop] = worker
        return worker


def _emit_timeline(item: STTAudioUploadItem, event: str, **fields: Any) -> None:
    if item.timeline is None:
        return
    try:
        item.timeline.emit(event, request_id=item.req_id, **fields)
    except Exception:
        pass


def _log_metadata_fields(item: STTAudioUploadItem) -> dict[str, Any]:
    return {
        str(key): value
        for key, value in item.metadata.items()
        if value not in (None, "")
    }


async def _upload_item(item: STTAudioUploadItem) -> None:
    started = time.perf_counter()
    config = _oci_upload_config_from_env()
    if config is None:
        raise RuntimeError("OCI Object Storage upload config is missing")

    response = await asyncio.to_thread(_put_object_sync, item, config)

    duration_ms = round((time.perf_counter() - started) * 1000)
    log_flow_event(
        item.logger or logger,
        "stt_audio_upload_done",
        request_id=item.req_id,
        message_id=item.message_id,
        session_id=item.session_id,
        object_name=item.object_name,
        bytes=len(item.wav_bytes),
        duration_ms=duration_ms,
        bucket=config.bucket,
        namespace=config.namespace,
        oci_status=getattr(response, "status", None),
        **_log_metadata_fields(item),
    )
    _emit_timeline(
        item,
        "stt_audio_upload_done",
        message_id=item.message_id,
        object_name=item.object_name,
        bytes=len(item.wav_bytes),
        duration_ms=duration_ms,
        bucket=config.bucket,
        namespace=config.namespace,
        oci_status=getattr(response, "status", None),
        **_log_metadata_fields(item),
    )


def enqueue_stt_vad_audio_upload(
    *,
    wav_bytes: bytes,
    req_id: str,
    message_id: str,
    structured_log_context: Any = None,
    metadata: Mapping[str, Any] | None = None,
    logger_override: Any | None = None,
    timeline: Any | None = None,
) -> str | None:
    if _oci_upload_config_from_env() is None:
        return None

    active_logger = logger_override or logger
    session_id = _metadata_value(structured_log_context, "session_id", "sessionId")
    object_name = build_stt_vad_object_name(
        session_id=session_id,
        message_id=message_id,
        req_id=req_id,
    )

    if len(wav_bytes) > _MAX_WAV_BYTES:
        log_flow_event(
            active_logger,
            "stt_audio_upload_drop",
            request_id=req_id,
            message_id=message_id,
            session_id=session_id,
            object_name=object_name,
            reason="too_large",
            bytes=len(wav_bytes),
        )
        return object_name

    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        log_flow_event(
            active_logger,
            "stt_audio_upload_drop",
            request_id=req_id,
            message_id=message_id,
            session_id=session_id,
            object_name=object_name,
            reason="no_running_loop",
        )
        return object_name

    item = STTAudioUploadItem(
        object_name=object_name,
        wav_bytes=wav_bytes,
        req_id=req_id,
        message_id=message_id,
        session_id=session_id,
        metadata=metadata or {},
        logger=active_logger,
        timeline=timeline,
    )
    queued = _worker_for_loop(loop).enqueue(item)
    if not queued:
        log_flow_event(
            active_logger,
            "stt_audio_upload_drop",
            request_id=req_id,
            message_id=message_id,
            session_id=session_id,
            object_name=object_name,
            reason="queue_full",
            bytes=len(wav_bytes),
        )
        return object_name

    log_flow_event(
        active_logger,
        "stt_audio_upload_enqueued",
        request_id=req_id,
        message_id=message_id,
        session_id=session_id,
        object_name=object_name,
        bytes=len(wav_bytes),
    )
    return object_name
