"""
Embedding service (FastAPI).

API (simple list-in, list-out; aligned by index):
- POST /embed/text   body: ["text1", "text2", ...] -> [[...], ...]
- POST /embed/image  body: ["url_or_path1", ...]  -> [[...], ...]
"""

import logging
import os
import pathlib
import threading
import time
import uuid
from collections import deque
from dataclasses import dataclass
from logging.handlers import TimedRotatingFileHandler
from typing import Any, Dict, List, Optional

import numpy as np
from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.concurrency import run_in_threadpool

from config.env_config import REDIS_CONFIG
from config.services_config import get_embedding_backend_config
from embeddings.cache_keys import build_image_cache_key, build_text_cache_key
from embeddings.config import CONFIG
from embeddings.protocols import ImageEncoderProtocol
from embeddings.redis_embedding_cache import RedisEmbeddingCache

app = FastAPI(title="saas-search Embedding Service", version="1.0.0")


class _DefaultRequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        if not hasattr(record, "reqid"):
            record.reqid = "-1"
        return True


def configure_embedding_logging() -> None:
    root_logger = logging.getLogger()
    if getattr(root_logger, "_embedding_logging_configured", False):
        return

    log_dir = pathlib.Path("logs")
    verbose_dir = log_dir / "verbose"
    log_dir.mkdir(exist_ok=True)
    verbose_dir.mkdir(parents=True, exist_ok=True)

    log_level = os.getenv("LOG_LEVEL", "INFO").upper()
    numeric_level = getattr(logging, log_level, logging.INFO)
    formatter = logging.Formatter(
        "%(asctime)s | reqid:%(reqid)s | %(name)s | %(levelname)s | %(message)s"
    )
    request_filter = _DefaultRequestIdFilter()

    root_logger.setLevel(numeric_level)

    file_handler = TimedRotatingFileHandler(
        filename=log_dir / "embedding_api.log",
        when="midnight",
        interval=1,
        backupCount=30,
        encoding="utf-8",
    )
    file_handler.setLevel(numeric_level)
    file_handler.setFormatter(formatter)
    file_handler.addFilter(request_filter)
    root_logger.addHandler(file_handler)

    error_handler = TimedRotatingFileHandler(
        filename=log_dir / "embedding_api_error.log",
        when="midnight",
        interval=1,
        backupCount=30,
        encoding="utf-8",
    )
    error_handler.setLevel(logging.ERROR)
    error_handler.setFormatter(formatter)
    error_handler.addFilter(request_filter)
    root_logger.addHandler(error_handler)

    verbose_logger = logging.getLogger("embedding.verbose")
    verbose_logger.setLevel(numeric_level)
    verbose_logger.handlers.clear()
    verbose_logger.propagate = False

    verbose_handler = TimedRotatingFileHandler(
        filename=verbose_dir / "embedding_verbose.log",
        when="midnight",
        interval=1,
        backupCount=30,
        encoding="utf-8",
    )
    verbose_handler.setLevel(numeric_level)
    verbose_handler.setFormatter(formatter)
    verbose_handler.addFilter(request_filter)
    verbose_logger.addHandler(verbose_handler)

    root_logger._embedding_logging_configured = True  # type: ignore[attr-defined]


configure_embedding_logging()
logger = logging.getLogger(__name__)
verbose_logger = logging.getLogger("embedding.verbose")

# Models are loaded at startup, not lazily
_text_model: Optional[Any] = None
_image_model: Optional[ImageEncoderProtocol] = None
_text_backend_name: str = ""
_SERVICE_KIND = (os.getenv("EMBEDDING_SERVICE_KIND", "all") or "all").strip().lower()
if _SERVICE_KIND not in {"all", "text", "image"}:
    raise RuntimeError(
        f"Invalid EMBEDDING_SERVICE_KIND={_SERVICE_KIND!r}; expected all, text, or image"
    )
_TEXT_ENABLED_BY_ENV = os.getenv("EMBEDDING_ENABLE_TEXT_MODEL", "true").lower() in ("1", "true", "yes")
_IMAGE_ENABLED_BY_ENV = os.getenv("EMBEDDING_ENABLE_IMAGE_MODEL", "true").lower() in ("1", "true", "yes")
open_text_model = _TEXT_ENABLED_BY_ENV and _SERVICE_KIND in {"all", "text"}
open_image_model = _IMAGE_ENABLED_BY_ENV and _SERVICE_KIND in {"all", "image"}

_text_encode_lock = threading.Lock()
_image_encode_lock = threading.Lock()

_TEXT_MICROBATCH_WINDOW_SEC = max(
    0.0, float(os.getenv("TEXT_MICROBATCH_WINDOW_MS", "4")) / 1000.0
)
_TEXT_REQUEST_TIMEOUT_SEC = max(
    1.0, float(os.getenv("TEXT_REQUEST_TIMEOUT_SEC", "30"))
)
_TEXT_MAX_INFLIGHT = max(1, int(os.getenv("TEXT_MAX_INFLIGHT", "32")))
_IMAGE_MAX_INFLIGHT = max(1, int(os.getenv("IMAGE_MAX_INFLIGHT", "20")))
_OVERLOAD_STATUS_CODE = int(os.getenv("EMBEDDING_OVERLOAD_STATUS_CODE", "503"))
_LOG_PREVIEW_COUNT = max(1, int(os.getenv("EMBEDDING_LOG_PREVIEW_COUNT", "3")))
_LOG_TEXT_PREVIEW_CHARS = max(32, int(os.getenv("EMBEDDING_LOG_TEXT_PREVIEW_CHARS", "120")))
_LOG_IMAGE_PREVIEW_CHARS = max(32, int(os.getenv("EMBEDDING_LOG_IMAGE_PREVIEW_CHARS", "180")))
_VECTOR_PREVIEW_DIMS = max(1, int(os.getenv("EMBEDDING_VECTOR_PREVIEW_DIMS", "6")))
_CACHE_PREFIX = str(REDIS_CONFIG.get("embedding_cache_prefix", "embedding")).strip() or "embedding"


@dataclass
class _EmbedResult:
    vectors: List[Optional[List[float]]]
    cache_hits: int
    cache_misses: int
    backend_elapsed_ms: float
    mode: str


class _EndpointStats:
    def __init__(self, name: str):
        self.name = name
        self._lock = threading.Lock()
        self.request_total = 0
        self.success_total = 0
        self.failure_total = 0
        self.rejected_total = 0
        self.cache_hits = 0
        self.cache_misses = 0
        self.total_latency_ms = 0.0
        self.total_backend_latency_ms = 0.0

    def record_rejected(self) -> None:
        with self._lock:
            self.request_total += 1
            self.rejected_total += 1

    def record_completed(
        self,
        *,
        success: bool,
        latency_ms: float,
        backend_latency_ms: float,
        cache_hits: int,
        cache_misses: int,
    ) -> None:
        with self._lock:
            self.request_total += 1
            if success:
                self.success_total += 1
            else:
                self.failure_total += 1
            self.cache_hits += max(0, int(cache_hits))
            self.cache_misses += max(0, int(cache_misses))
            self.total_latency_ms += max(0.0, float(latency_ms))
            self.total_backend_latency_ms += max(0.0, float(backend_latency_ms))

    def snapshot(self) -> Dict[str, Any]:
        with self._lock:
            completed = self.success_total + self.failure_total
            return {
                "request_total": self.request_total,
                "success_total": self.success_total,
                "failure_total": self.failure_total,
                "rejected_total": self.rejected_total,
                "cache_hits": self.cache_hits,
                "cache_misses": self.cache_misses,
                "avg_latency_ms": round(self.total_latency_ms / completed, 3) if completed else 0.0,
                "avg_backend_latency_ms": round(self.total_backend_latency_ms / completed, 3)
                if completed
                else 0.0,
            }


class _InflightLimiter:
    def __init__(self, name: str, limit: int):
        self.name = name
        self.limit = max(1, int(limit))
        self._sem = threading.BoundedSemaphore(self.limit)
        self._lock = threading.Lock()
        self._active = 0
        self._rejected = 0
        self._completed = 0
        self._failed = 0
        self._max_active = 0

    def try_acquire(self) -> tuple[bool, int]:
        if not self._sem.acquire(blocking=False):
            with self._lock:
                self._rejected += 1
                active = self._active
            return False, active
        with self._lock:
            self._active += 1
            self._max_active = max(self._max_active, self._active)
            active = self._active
        return True, active

    def release(self, *, success: bool) -> int:
        with self._lock:
            self._active = max(0, self._active - 1)
            if success:
                self._completed += 1
            else:
                self._failed += 1
            active = self._active
        self._sem.release()
        return active

    def snapshot(self) -> Dict[str, int]:
        with self._lock:
            return {
                "limit": self.limit,
                "active": self._active,
                "rejected_total": self._rejected,
                "completed_total": self._completed,
                "failed_total": self._failed,
                "max_active": self._max_active,
            }


_text_request_limiter = _InflightLimiter(name="text", limit=_TEXT_MAX_INFLIGHT)
_image_request_limiter = _InflightLimiter(name="image", limit=_IMAGE_MAX_INFLIGHT)
_text_stats = _EndpointStats(name="text")
_image_stats = _EndpointStats(name="image")
_text_cache = RedisEmbeddingCache(key_prefix=_CACHE_PREFIX, namespace="")
_image_cache = RedisEmbeddingCache(key_prefix=_CACHE_PREFIX, namespace="image")


@dataclass
class _SingleTextTask:
    text: str
    normalize: bool
    created_at: float
    request_id: str
    done: threading.Event
    result: Optional[List[float]] = None
    error: Optional[Exception] = None


_text_single_queue: "deque[_SingleTextTask]" = deque()
_text_single_queue_cv = threading.Condition()
_text_batch_worker: Optional[threading.Thread] = None
_text_batch_worker_stop = False


def _compact_preview(text: str, max_chars: int) -> str:
    compact = " ".join((text or "").split())
    if len(compact) <= max_chars:
        return compact
    return compact[:max_chars] + "..."


def _preview_inputs(items: List[str], max_items: int, max_chars: int) -> List[Dict[str, Any]]:
    previews: List[Dict[str, Any]] = []
    for idx, item in enumerate(items[:max_items]):
        previews.append(
            {
                "idx": idx,
                "len": len(item),
                "preview": _compact_preview(item, max_chars),
            }
        )
    return previews


def _preview_vector(vec: Optional[List[float]], max_dims: int = _VECTOR_PREVIEW_DIMS) -> List[float]:
    if not vec:
        return []
    return [round(float(v), 6) for v in vec[:max_dims]]


def _request_log_extra(request_id: str) -> Dict[str, str]:
    return {"reqid": request_id}


def _resolve_request_id(http_request: Request) -> str:
    header_value = http_request.headers.get("X-Request-ID")
    if header_value and header_value.strip():
        return header_value.strip()[:32]
    return str(uuid.uuid4())[:8]


def _request_client(http_request: Request) -> str:
    client = getattr(http_request, "client", None)
    host = getattr(client, "host", None)
    return str(host or "-")


def _encode_local_st(texts: List[str], normalize_embeddings: bool) -> Any:
    with _text_encode_lock:
        return _text_model.encode(
            texts,
            batch_size=int(CONFIG.TEXT_BATCH_SIZE),
            device=CONFIG.TEXT_DEVICE,
            normalize_embeddings=normalize_embeddings,
        )


def _start_text_batch_worker() -> None:
    global _text_batch_worker, _text_batch_worker_stop
    if _text_batch_worker is not None and _text_batch_worker.is_alive():
        return
    _text_batch_worker_stop = False
    _text_batch_worker = threading.Thread(
        target=_text_batch_worker_loop,
        name="embed-text-microbatch-worker",
        daemon=True,
    )
    _text_batch_worker.start()
    logger.info(
        "Started local_st text micro-batch worker | window_ms=%.1f max_batch=%d",
        _TEXT_MICROBATCH_WINDOW_SEC * 1000.0,
        int(CONFIG.TEXT_BATCH_SIZE),
    )


def _stop_text_batch_worker() -> None:
    global _text_batch_worker_stop
    with _text_single_queue_cv:
        _text_batch_worker_stop = True
        _text_single_queue_cv.notify_all()


def _text_batch_worker_loop() -> None:
    max_batch = max(1, int(CONFIG.TEXT_BATCH_SIZE))
    while True:
        with _text_single_queue_cv:
            while not _text_single_queue and not _text_batch_worker_stop:
                _text_single_queue_cv.wait()
            if _text_batch_worker_stop:
                return

            batch: List[_SingleTextTask] = [_text_single_queue.popleft()]
            deadline = time.perf_counter() + _TEXT_MICROBATCH_WINDOW_SEC

            while len(batch) < max_batch:
                remaining = deadline - time.perf_counter()
                if remaining <= 0:
                    break
                if not _text_single_queue:
                    _text_single_queue_cv.wait(timeout=remaining)
                    continue
                while _text_single_queue and len(batch) < max_batch:
                    batch.append(_text_single_queue.popleft())

        try:
            queue_wait_ms = [(time.perf_counter() - task.created_at) * 1000.0 for task in batch]
            reqids = [task.request_id for task in batch]
            logger.info(
                "text microbatch dispatch | size=%d queue_wait_ms_min=%.2f queue_wait_ms_max=%.2f reqids=%s preview=%s",
                len(batch),
                min(queue_wait_ms) if queue_wait_ms else 0.0,
                max(queue_wait_ms) if queue_wait_ms else 0.0,
                reqids,
                _preview_inputs(
                    [task.text for task in batch],
                    _LOG_PREVIEW_COUNT,
                    _LOG_TEXT_PREVIEW_CHARS,
                ),
            )
            batch_t0 = time.perf_counter()
            embs = _encode_local_st([task.text for task in batch], normalize_embeddings=False)
            if embs is None or len(embs) != len(batch):
                raise RuntimeError(
                    f"Text model response length mismatch in micro-batch: "
                    f"expected {len(batch)}, got {0 if embs is None else len(embs)}"
                )
            for task, emb in zip(batch, embs):
                vec = _as_list(emb, normalize=task.normalize)
                if vec is None:
                    raise RuntimeError("Text model returned empty embedding in micro-batch")
                task.result = vec
            logger.info(
                "text microbatch done | size=%d reqids=%s dim=%d backend_elapsed_ms=%.2f",
                len(batch),
                reqids,
                len(batch[0].result) if batch and batch[0].result is not None else 0,
                (time.perf_counter() - batch_t0) * 1000.0,
            )
        except Exception as exc:
            logger.error(
                "text microbatch failed | size=%d reqids=%s error=%s",
                len(batch),
                [task.request_id for task in batch],
                exc,
                exc_info=True,
            )
            for task in batch:
                task.error = exc
        finally:
            for task in batch:
                task.done.set()


def _encode_single_text_with_microbatch(text: str, normalize: bool, request_id: str) -> List[float]:
    task = _SingleTextTask(
        text=text,
        normalize=normalize,
        created_at=time.perf_counter(),
        request_id=request_id,
        done=threading.Event(),
    )
    with _text_single_queue_cv:
        _text_single_queue.append(task)
        _text_single_queue_cv.notify()

    if not task.done.wait(timeout=_TEXT_REQUEST_TIMEOUT_SEC):
        with _text_single_queue_cv:
            try:
                _text_single_queue.remove(task)
            except ValueError:
                pass
        raise RuntimeError(
            f"Timed out waiting for text micro-batch worker ({_TEXT_REQUEST_TIMEOUT_SEC:.1f}s)"
        )
    if task.error is not None:
        raise task.error
    if task.result is None:
        raise RuntimeError("Text micro-batch worker returned empty result")
    return task.result


@app.on_event("startup")
def load_models():
    """Load models at service startup to avoid first-request latency."""
    global _text_model, _image_model, _text_backend_name

    logger.info(
        "Loading embedding models at startup | service_kind=%s text_enabled=%s image_enabled=%s",
        _SERVICE_KIND,
        open_text_model,
        open_image_model,
    )

    if open_text_model:
        try:
            backend_name, backend_cfg = get_embedding_backend_config()
            _text_backend_name = backend_name
            if backend_name == "tei":
                from embeddings.text_embedding_tei import TEITextModel

                base_url = backend_cfg.get("base_url") or CONFIG.TEI_BASE_URL
                timeout_sec = int(backend_cfg.get("timeout_sec") or CONFIG.TEI_TIMEOUT_SEC)
                logger.info("Loading text backend: tei (base_url=%s)", base_url)
                _text_model = TEITextModel(
                    base_url=str(base_url),
                    timeout_sec=timeout_sec,
                )
            elif backend_name == "local_st":
                from embeddings.text_embedding_sentence_transformers import Qwen3TextModel

                model_id = backend_cfg.get("model_id") or CONFIG.TEXT_MODEL_ID
                logger.info("Loading text backend: local_st (model=%s)", model_id)
                _text_model = Qwen3TextModel(model_id=str(model_id))
                _start_text_batch_worker()
            else:
                raise ValueError(
                    f"Unsupported embedding backend: {backend_name}. "
                    "Supported: tei, local_st"
                )
            logger.info("Text backend loaded successfully: %s", _text_backend_name)
        except Exception as e:
            logger.error("Failed to load text model: %s", e, exc_info=True)
            raise

    if open_image_model:
        try:
            if CONFIG.USE_CLIP_AS_SERVICE:
                from embeddings.clip_as_service_encoder import ClipAsServiceImageEncoder

                logger.info(
                    "Loading image encoder via clip-as-service: %s (configured model: %s)",
                    CONFIG.CLIP_AS_SERVICE_SERVER,
                    CONFIG.CLIP_AS_SERVICE_MODEL_NAME,
                )
                _image_model = ClipAsServiceImageEncoder(
                    server=CONFIG.CLIP_AS_SERVICE_SERVER,
                    batch_size=CONFIG.IMAGE_BATCH_SIZE,
                )
                logger.info("Image model (clip-as-service) loaded successfully")
            else:
                from embeddings.clip_model import ClipImageModel

                logger.info(
                    "Loading local image model: %s (device: %s)",
                    CONFIG.IMAGE_MODEL_NAME,
                    CONFIG.IMAGE_DEVICE,
                )
                _image_model = ClipImageModel(
                    model_name=CONFIG.IMAGE_MODEL_NAME,
                    device=CONFIG.IMAGE_DEVICE,
                )
                logger.info("Image model (local CN-CLIP) loaded successfully")
        except Exception as e:
            logger.error("Failed to load image model: %s", e, exc_info=True)
            raise

    logger.info("All embedding models loaded successfully, service ready")


@app.on_event("shutdown")
def stop_workers() -> None:
    _stop_text_batch_worker()


def _normalize_vector(vec: np.ndarray) -> np.ndarray:
    norm = float(np.linalg.norm(vec))
    if not np.isfinite(norm) or norm <= 0.0:
        raise RuntimeError("Embedding vector has invalid norm (must be > 0)")
    return vec / norm


def _as_list(embedding: Optional[np.ndarray], normalize: bool = False) -> Optional[List[float]]:
    if embedding is None:
        return None
    if not isinstance(embedding, np.ndarray):
        embedding = np.array(embedding, dtype=np.float32)
    if embedding.ndim != 1:
        embedding = embedding.reshape(-1)
    embedding = embedding.astype(np.float32, copy=False)
    if normalize:
        embedding = _normalize_vector(embedding).astype(np.float32, copy=False)
    return embedding.tolist()


def _try_full_text_cache_hit(
    normalized: List[str],
    effective_normalize: bool,
) -> Optional[_EmbedResult]:
    out: List[Optional[List[float]]] = []
    for text in normalized:
        cached = _text_cache.get(build_text_cache_key(text, normalize=effective_normalize))
        if cached is None:
            return None
        vec = _as_list(cached, normalize=False)
        if vec is None:
            return None
        out.append(vec)
    return _EmbedResult(
        vectors=out,
        cache_hits=len(out),
        cache_misses=0,
        backend_elapsed_ms=0.0,
        mode="cache-only",
    )


def _try_full_image_cache_hit(
    urls: List[str],
    effective_normalize: bool,
) -> Optional[_EmbedResult]:
    out: List[Optional[List[float]]] = []
    for url in urls:
        cached = _image_cache.get(build_image_cache_key(url, normalize=effective_normalize))
        if cached is None:
            return None
        vec = _as_list(cached, normalize=False)
        if vec is None:
            return None
        out.append(vec)
    return _EmbedResult(
        vectors=out,
        cache_hits=len(out),
        cache_misses=0,
        backend_elapsed_ms=0.0,
        mode="cache-only",
    )


@app.get("/health")
def health() -> Dict[str, Any]:
    """Health check endpoint. Returns status and current throttling stats."""
    ready = (not open_text_model or _text_model is not None) and (not open_image_model or _image_model is not None)
    return {
        "status": "ok" if ready else "degraded",
        "service_kind": _SERVICE_KIND,
        "text_model_loaded": _text_model is not None,
        "text_backend": _text_backend_name,
        "image_model_loaded": _image_model is not None,
        "cache_enabled": {
            "text": _text_cache.redis_client is not None,
            "image": _image_cache.redis_client is not None,
        },
        "limits": {
            "text": _text_request_limiter.snapshot(),
            "image": _image_request_limiter.snapshot(),
        },
        "stats": {
            "text": _text_stats.snapshot(),
            "image": _image_stats.snapshot(),
        },
        "text_microbatch": {
            "window_ms": round(_TEXT_MICROBATCH_WINDOW_SEC * 1000.0, 3),
            "queue_depth": len(_text_single_queue),
            "worker_alive": bool(_text_batch_worker is not None and _text_batch_worker.is_alive()),
            "request_timeout_sec": _TEXT_REQUEST_TIMEOUT_SEC,
        },
    }


@app.get("/ready")
def ready() -> Dict[str, Any]:
    text_ready = (not open_text_model) or (_text_model is not None)
    image_ready = (not open_image_model) or (_image_model is not None)
    if not (text_ready and image_ready):
        raise HTTPException(
            status_code=503,
            detail={
                "service_kind": _SERVICE_KIND,
                "text_ready": text_ready,
                "image_ready": image_ready,
            },
        )
    return {
        "status": "ready",
        "service_kind": _SERVICE_KIND,
        "text_ready": text_ready,
        "image_ready": image_ready,
    }


def _embed_text_impl(
    normalized: List[str],
    effective_normalize: bool,
    request_id: str,
) -> _EmbedResult:
    if _text_model is None:
        raise RuntimeError("Text model not loaded")

    out: List[Optional[List[float]]] = [None] * len(normalized)
    missing_indices: List[int] = []
    missing_texts: List[str] = []
    missing_cache_keys: List[str] = []
    cache_hits = 0
    for idx, text in enumerate(normalized):
        cache_key = build_text_cache_key(text, normalize=effective_normalize)
        cached = _text_cache.get(cache_key)
        if cached is not None:
            vec = _as_list(cached, normalize=False)
            if vec is not None:
                out[idx] = vec
                cache_hits += 1
                continue
        missing_indices.append(idx)
        missing_texts.append(text)
        missing_cache_keys.append(cache_key)

    if not missing_texts:
        logger.info(
            "text backend done | backend=%s mode=cache-only inputs=%d normalize=%s dim=%d cache_hits=%d cache_misses=0 backend_elapsed_ms=0.00",
            _text_backend_name,
            len(normalized),
            effective_normalize,
            len(out[0]) if out and out[0] is not None else 0,
            cache_hits,
            extra=_request_log_extra(request_id),
        )
        return _EmbedResult(
            vectors=out,
            cache_hits=cache_hits,
            cache_misses=0,
            backend_elapsed_ms=0.0,
            mode="cache-only",
        )

    backend_t0 = time.perf_counter()
    try:
        if _text_backend_name == "local_st":
            if len(missing_texts) == 1 and _text_batch_worker is not None:
                computed = [
                    _encode_single_text_with_microbatch(
                        missing_texts[0],
                        normalize=effective_normalize,
                        request_id=request_id,
                    )
                ]
                mode = "microbatch-single"
            else:
                embs = _encode_local_st(missing_texts, normalize_embeddings=False)
                computed = []
                for i, emb in enumerate(embs):
                    vec = _as_list(emb, normalize=effective_normalize)
                    if vec is None:
                        raise RuntimeError(f"Text model returned empty embedding for missing index {i}")
                    computed.append(vec)
                mode = "direct-batch"
        else:
            embs = _text_model.encode(
                missing_texts,
                batch_size=int(CONFIG.TEXT_BATCH_SIZE),
                device=CONFIG.TEXT_DEVICE,
                normalize_embeddings=effective_normalize,
            )
            computed = []
            for i, emb in enumerate(embs):
                vec = _as_list(emb, normalize=False)
                if vec is None:
                    raise RuntimeError(f"Text model returned empty embedding for missing index {i}")
                computed.append(vec)
            mode = "backend-batch"
    except Exception as e:
        logger.error(
            "Text embedding backend failure: %s",
            e,
            exc_info=True,
            extra=_request_log_extra(request_id),
        )
        raise RuntimeError(f"Text embedding backend failure: {e}") from e

    if len(computed) != len(missing_texts):
        raise RuntimeError(
            f"Text model response length mismatch: expected {len(missing_texts)}, "
            f"got {len(computed)}"
        )

    for pos, cache_key, vec in zip(missing_indices, missing_cache_keys, computed):
        out[pos] = vec
        _text_cache.set(cache_key, np.asarray(vec, dtype=np.float32))

    backend_elapsed_ms = (time.perf_counter() - backend_t0) * 1000.0

    logger.info(
        "text backend done | backend=%s mode=%s inputs=%d normalize=%s dim=%d cache_hits=%d cache_misses=%d backend_elapsed_ms=%.2f",
        _text_backend_name,
        mode,
        len(normalized),
        effective_normalize,
        len(out[0]) if out and out[0] is not None else 0,
        cache_hits,
        len(missing_texts),
        backend_elapsed_ms,
        extra=_request_log_extra(request_id),
    )
    return _EmbedResult(
        vectors=out,
        cache_hits=cache_hits,
        cache_misses=len(missing_texts),
        backend_elapsed_ms=backend_elapsed_ms,
        mode=mode,
    )


@app.post("/embed/text")
async def embed_text(
    texts: List[str],
    http_request: Request,
    response: Response,
    normalize: Optional[bool] = None,
) -> List[Optional[List[float]]]:
    if _text_model is None:
        raise HTTPException(status_code=503, detail="Text embedding model not loaded in this service")

    request_id = _resolve_request_id(http_request)
    response.headers["X-Request-ID"] = request_id

    effective_normalize = bool(CONFIG.TEXT_NORMALIZE_EMBEDDINGS) if normalize is None else bool(normalize)
    normalized: List[str] = []
    for i, t in enumerate(texts):
        if not isinstance(t, str):
            raise HTTPException(status_code=400, detail=f"Invalid text at index {i}: must be string")
        s = t.strip()
        if not s:
            raise HTTPException(status_code=400, detail=f"Invalid text at index {i}: empty string")
        normalized.append(s)

    cache_check_started = time.perf_counter()
    cache_only = _try_full_text_cache_hit(normalized, effective_normalize)
    if cache_only is not None:
        latency_ms = (time.perf_counter() - cache_check_started) * 1000.0
        _text_stats.record_completed(
            success=True,
            latency_ms=latency_ms,
            backend_latency_ms=0.0,
            cache_hits=cache_only.cache_hits,
            cache_misses=0,
        )
        logger.info(
            "embed_text response | backend=%s mode=cache-only inputs=%d normalize=%s dim=%d cache_hits=%d cache_misses=0 first_vector=%s latency_ms=%.2f",
            _text_backend_name,
            len(normalized),
            effective_normalize,
            len(cache_only.vectors[0]) if cache_only.vectors and cache_only.vectors[0] is not None else 0,
            cache_only.cache_hits,
            _preview_vector(cache_only.vectors[0] if cache_only.vectors else None),
            latency_ms,
            extra=_request_log_extra(request_id),
        )
        return cache_only.vectors

    accepted, active = _text_request_limiter.try_acquire()
    if not accepted:
        _text_stats.record_rejected()
        logger.warning(
            "embed_text rejected | client=%s backend=%s inputs=%d normalize=%s active=%d limit=%d preview=%s",
            _request_client(http_request),
            _text_backend_name,
            len(normalized),
            effective_normalize,
            active,
            _TEXT_MAX_INFLIGHT,
            _preview_inputs(normalized, _LOG_PREVIEW_COUNT, _LOG_TEXT_PREVIEW_CHARS),
            extra=_request_log_extra(request_id),
        )
        raise HTTPException(
            status_code=_OVERLOAD_STATUS_CODE,
            detail=f"Text embedding service busy: active={active}, limit={_TEXT_MAX_INFLIGHT}",
        )

    request_started = time.perf_counter()
    success = False
    backend_elapsed_ms = 0.0
    cache_hits = 0
    cache_misses = 0
    try:
        logger.info(
            "embed_text request | client=%s backend=%s inputs=%d normalize=%s active=%d limit=%d preview=%s",
            _request_client(http_request),
            _text_backend_name,
            len(normalized),
            effective_normalize,
            active,
            _TEXT_MAX_INFLIGHT,
            _preview_inputs(normalized, _LOG_PREVIEW_COUNT, _LOG_TEXT_PREVIEW_CHARS),
            extra=_request_log_extra(request_id),
        )
        verbose_logger.info(
            "embed_text detail | payload=%s normalize=%s backend=%s",
            normalized,
            effective_normalize,
            _text_backend_name,
            extra=_request_log_extra(request_id),
        )
        result = await run_in_threadpool(_embed_text_impl, normalized, effective_normalize, request_id)
        success = True
        backend_elapsed_ms = result.backend_elapsed_ms
        cache_hits = result.cache_hits
        cache_misses = result.cache_misses
        latency_ms = (time.perf_counter() - request_started) * 1000.0
        _text_stats.record_completed(
            success=True,
            latency_ms=latency_ms,
            backend_latency_ms=backend_elapsed_ms,
            cache_hits=cache_hits,
            cache_misses=cache_misses,
        )
        logger.info(
            "embed_text response | backend=%s mode=%s inputs=%d normalize=%s dim=%d cache_hits=%d cache_misses=%d first_vector=%s latency_ms=%.2f",
            _text_backend_name,
            result.mode,
            len(normalized),
            effective_normalize,
            len(result.vectors[0]) if result.vectors and result.vectors[0] is not None else 0,
            cache_hits,
            cache_misses,
            _preview_vector(result.vectors[0] if result.vectors else None),
            latency_ms,
            extra=_request_log_extra(request_id),
        )
        verbose_logger.info(
            "embed_text result detail | count=%d first_vector=%s latency_ms=%.2f",
            len(result.vectors),
            result.vectors[0][: _VECTOR_PREVIEW_DIMS]
            if result.vectors and result.vectors[0] is not None
            else [],
            latency_ms,
            extra=_request_log_extra(request_id),
        )
        return result.vectors
    except HTTPException:
        raise
    except Exception as e:
        latency_ms = (time.perf_counter() - request_started) * 1000.0
        _text_stats.record_completed(
            success=False,
            latency_ms=latency_ms,
            backend_latency_ms=backend_elapsed_ms,
            cache_hits=cache_hits,
            cache_misses=cache_misses,
        )
        logger.error(
            "embed_text failed | backend=%s inputs=%d normalize=%s latency_ms=%.2f error=%s",
            _text_backend_name,
            len(normalized),
            effective_normalize,
            latency_ms,
            e,
            exc_info=True,
            extra=_request_log_extra(request_id),
        )
        raise HTTPException(status_code=502, detail=str(e)) from e
    finally:
        remaining = _text_request_limiter.release(success=success)
        logger.info(
            "embed_text finalize | success=%s active_after=%d",
            success,
            remaining,
            extra=_request_log_extra(request_id),
        )


def _embed_image_impl(
    urls: List[str],
    effective_normalize: bool,
    request_id: str,
) -> _EmbedResult:
    if _image_model is None:
        raise RuntimeError("Image model not loaded")

    out: List[Optional[List[float]]] = [None] * len(urls)
    missing_indices: List[int] = []
    missing_urls: List[str] = []
    missing_cache_keys: List[str] = []
    cache_hits = 0
    for idx, url in enumerate(urls):
        cache_key = build_image_cache_key(url, normalize=effective_normalize)
        cached = _image_cache.get(cache_key)
        if cached is not None:
            vec = _as_list(cached, normalize=False)
            if vec is not None:
                out[idx] = vec
                cache_hits += 1
                continue
        missing_indices.append(idx)
        missing_urls.append(url)
        missing_cache_keys.append(cache_key)

    if not missing_urls:
        logger.info(
            "image backend done | mode=cache-only inputs=%d normalize=%s dim=%d cache_hits=%d cache_misses=0 backend_elapsed_ms=0.00",
            len(urls),
            effective_normalize,
            len(out[0]) if out and out[0] is not None else 0,
            cache_hits,
            extra=_request_log_extra(request_id),
        )
        return _EmbedResult(
            vectors=out,
            cache_hits=cache_hits,
            cache_misses=0,
            backend_elapsed_ms=0.0,
            mode="cache-only",
        )

    backend_t0 = time.perf_counter()
    with _image_encode_lock:
        vectors = _image_model.encode_image_urls(
            missing_urls,
            batch_size=CONFIG.IMAGE_BATCH_SIZE,
            normalize_embeddings=effective_normalize,
        )
    if vectors is None or len(vectors) != len(missing_urls):
        raise RuntimeError(
            f"Image model response length mismatch: expected {len(missing_urls)}, "
            f"got {0 if vectors is None else len(vectors)}"
        )

    for pos, cache_key, vec in zip(missing_indices, missing_cache_keys, vectors):
        out_vec = _as_list(vec, normalize=effective_normalize)
        if out_vec is None:
            raise RuntimeError(f"Image model returned empty embedding for position {pos}")
        out[pos] = out_vec
        _image_cache.set(cache_key, np.asarray(out_vec, dtype=np.float32))

    backend_elapsed_ms = (time.perf_counter() - backend_t0) * 1000.0

    logger.info(
        "image backend done | mode=backend-batch inputs=%d normalize=%s dim=%d cache_hits=%d cache_misses=%d backend_elapsed_ms=%.2f",
        len(urls),
        effective_normalize,
        len(out[0]) if out and out[0] is not None else 0,
        cache_hits,
        len(missing_urls),
        backend_elapsed_ms,
        extra=_request_log_extra(request_id),
    )
    return _EmbedResult(
        vectors=out,
        cache_hits=cache_hits,
        cache_misses=len(missing_urls),
        backend_elapsed_ms=backend_elapsed_ms,
        mode="backend-batch",
    )


@app.post("/embed/image")
async def embed_image(
    images: List[str],
    http_request: Request,
    response: Response,
    normalize: Optional[bool] = None,
) -> List[Optional[List[float]]]:
    if _image_model is None:
        raise HTTPException(status_code=503, detail="Image embedding model not loaded in this service")

    request_id = _resolve_request_id(http_request)
    response.headers["X-Request-ID"] = request_id

    effective_normalize = bool(CONFIG.IMAGE_NORMALIZE_EMBEDDINGS) if normalize is None else bool(normalize)
    urls: List[str] = []
    for i, url_or_path in enumerate(images):
        if not isinstance(url_or_path, str):
            raise HTTPException(status_code=400, detail=f"Invalid image at index {i}: must be string URL/path")
        s = url_or_path.strip()
        if not s:
            raise HTTPException(status_code=400, detail=f"Invalid image at index {i}: empty URL/path")
        urls.append(s)

    cache_check_started = time.perf_counter()
    cache_only = _try_full_image_cache_hit(urls, effective_normalize)
    if cache_only is not None:
        latency_ms = (time.perf_counter() - cache_check_started) * 1000.0
        _image_stats.record_completed(
            success=True,
            latency_ms=latency_ms,
            backend_latency_ms=0.0,
            cache_hits=cache_only.cache_hits,
            cache_misses=0,
        )
        logger.info(
            "embed_image response | mode=cache-only inputs=%d normalize=%s dim=%d cache_hits=%d cache_misses=0 first_vector=%s latency_ms=%.2f",
            len(urls),
            effective_normalize,
            len(cache_only.vectors[0]) if cache_only.vectors and cache_only.vectors[0] is not None else 0,
            cache_only.cache_hits,
            _preview_vector(cache_only.vectors[0] if cache_only.vectors else None),
            latency_ms,
            extra=_request_log_extra(request_id),
        )
        return cache_only.vectors

    accepted, active = _image_request_limiter.try_acquire()
    if not accepted:
        _image_stats.record_rejected()
        logger.warning(
            "embed_image rejected | client=%s inputs=%d normalize=%s active=%d limit=%d preview=%s",
            _request_client(http_request),
            len(urls),
            effective_normalize,
            active,
            _IMAGE_MAX_INFLIGHT,
            _preview_inputs(urls, _LOG_PREVIEW_COUNT, _LOG_IMAGE_PREVIEW_CHARS),
            extra=_request_log_extra(request_id),
        )
        raise HTTPException(
            status_code=_OVERLOAD_STATUS_CODE,
            detail=f"Image embedding service busy: active={active}, limit={_IMAGE_MAX_INFLIGHT}",
        )

    request_started = time.perf_counter()
    success = False
    backend_elapsed_ms = 0.0
    cache_hits = 0
    cache_misses = 0
    try:
        logger.info(
            "embed_image request | client=%s inputs=%d normalize=%s active=%d limit=%d preview=%s",
            _request_client(http_request),
            len(urls),
            effective_normalize,
            active,
            _IMAGE_MAX_INFLIGHT,
            _preview_inputs(urls, _LOG_PREVIEW_COUNT, _LOG_IMAGE_PREVIEW_CHARS),
            extra=_request_log_extra(request_id),
        )
        verbose_logger.info(
            "embed_image detail | payload=%s normalize=%s",
            urls,
            effective_normalize,
            extra=_request_log_extra(request_id),
        )
        result = await run_in_threadpool(_embed_image_impl, urls, effective_normalize, request_id)
        success = True
        backend_elapsed_ms = result.backend_elapsed_ms
        cache_hits = result.cache_hits
        cache_misses = result.cache_misses
        latency_ms = (time.perf_counter() - request_started) * 1000.0
        _image_stats.record_completed(
            success=True,
            latency_ms=latency_ms,
            backend_latency_ms=backend_elapsed_ms,
            cache_hits=cache_hits,
            cache_misses=cache_misses,
        )
        logger.info(
            "embed_image response | mode=%s inputs=%d normalize=%s dim=%d cache_hits=%d cache_misses=%d first_vector=%s latency_ms=%.2f",
            result.mode,
            len(urls),
            effective_normalize,
            len(result.vectors[0]) if result.vectors and result.vectors[0] is not None else 0,
            cache_hits,
            cache_misses,
            _preview_vector(result.vectors[0] if result.vectors else None),
            latency_ms,
            extra=_request_log_extra(request_id),
        )
        verbose_logger.info(
            "embed_image result detail | count=%d first_vector=%s latency_ms=%.2f",
            len(result.vectors),
            result.vectors[0][: _VECTOR_PREVIEW_DIMS]
            if result.vectors and result.vectors[0] is not None
            else [],
            latency_ms,
            extra=_request_log_extra(request_id),
        )
        return result.vectors
    except HTTPException:
        raise
    except Exception as e:
        latency_ms = (time.perf_counter() - request_started) * 1000.0
        _image_stats.record_completed(
            success=False,
            latency_ms=latency_ms,
            backend_latency_ms=backend_elapsed_ms,
            cache_hits=cache_hits,
            cache_misses=cache_misses,
        )
        logger.error(
            "embed_image failed | inputs=%d normalize=%s latency_ms=%.2f error=%s",
            len(urls),
            effective_normalize,
            latency_ms,
            e,
            exc_info=True,
            extra=_request_log_extra(request_id),
        )
        raise HTTPException(status_code=502, detail=f"Image embedding backend failure: {e}") from e
    finally:
        remaining = _image_request_limiter.release(success=success)
        logger.info(
            "embed_image finalize | success=%s active_after=%d",
            success,
            remaining,
            extra=_request_log_extra(request_id),
        )