"""Optional OpenTelemetry-facing observability helpers.

Core keeps this module dependency-light: if ``opentelemetry-api`` is absent,
all span and metric operations are no-ops. Host applications that configure an
OTel SDK/exporter get spans and metrics through the standard API.
"""

from __future__ import annotations

import contextlib
import functools
import threading
from collections.abc import Mapping
from typing import Any, Literal

MetricKind = Literal["counter", "histogram", "observable_gauge"]

INSTRUMENTATION_NAME = "easycat"

SPAN_NAMES = frozenset(
    {
        "easycat.session",
        "easycat.transport.receive",
        "easycat.vad.detect",
        "easycat.stt.stream",
        "easycat.turn.commit",
        "easycat.agent.invoke",
        "easycat.agent.tool",
        "easycat.tts.synthesize",
        "easycat.transport.send",
        "easycat.journal.append",
    }
)

METRIC_DEFINITIONS: Mapping[str, MetricKind] = {
    "easycat.turn.latency": "histogram",
    "easycat.stage.latency": "histogram",
    "easycat.journal.append.latency": "histogram",
    "easycat.sessions.active": "observable_gauge",
    "easycat.turns.total": "counter",
    "easycat.audio.bytes.total": "counter",
    "easycat.audio.frames.total": "counter",
    "easycat.provider.errors.total": "counter",
    "easycat.session.errors.total": "counter",
    "easycat.transport.disconnects.total": "counter",
    "easycat.validation.failures.total": "counter",
    "easycat.queue.depth": "observable_gauge",
    "easycat.queue.dropped.total": "counter",
    "easycat.event_loop.lag": "histogram",
    "easycat.journal.degraded": "observable_gauge",
    "easycat.interruption.total": "counter",
    "easycat.interruption.cutoff_latency": "histogram",
    # ── easycat.server.* (M8) — the VoiceServer process layer ─────────
    # Registered here in the SAME change that first emits them
    # (``easycat.server.metrics``); ``_record_metric`` raises ``ValueError`` on
    # any unregistered name, so registration and emission must land together.
    # ``*.total`` are monotonic counts; ``request.duration`` is a latency
    # distribution (seconds, matching the ``easycat.*.latency`` convention);
    # ``connections.active`` and ``draining`` are point-in-time states read at
    # scrape time (``draining`` is 0/1).
    "easycat.server.requests.total": "counter",
    "easycat.server.request.duration": "histogram",
    "easycat.server.sessions.rejected.total": "counter",
    "easycat.server.connections.active": "observable_gauge",
    "easycat.server.draining": "observable_gauge",
}

LOW_CARDINALITY_ATTRIBUTE_KEYS = frozenset(
    {
        "easycat.provider",
        "easycat.provider_family",
        "easycat.surface",
        "easycat.transport",
        "easycat.debug_mode",
        "easycat.stage",
        "easycat.condition_id",
        "easycat.feature_set",
        "easycat.result",
        "easycat.error_type",
        # ── easycat.server.* labels (M8) ──────────────────────────────
        # ``easycat.route`` is a ROUTE TEMPLATE (never a raw path):
        # ``easycat.server.metrics`` asserts the value is in the enumerated
        # template set before recording. ``easycat.server_state`` is the
        # ``ServerState`` Literal (``serving``/``draining``); ``easycat.auth_result``
        # is the ``AuthReason`` Literal (``allowed``/``missing``/``invalid``).
        # None contain a forbidden substring, so allow-list membership is
        # sufficient. (``easycat.transport`` is already registered above.)
        "easycat.route",
        "easycat.server_state",
        "easycat.auth_result",
    }
)

SPAN_ATTRIBUTE_KEYS = LOW_CARDINALITY_ATTRIBUTE_KEYS | frozenset(
    {
        "gen_ai.operation.name",
        "gen_ai.request.model",
        "gen_ai.system",
    }
)
ALLOWED_ATTRIBUTE_KEYS = LOW_CARDINALITY_ATTRIBUTE_KEYS

FORBIDDEN_ATTRIBUTE_KEYS = frozenset(
    {
        "api_key",
        "file_path",
        "generated_text",
        "ip_address",
        "model_output",
        "phone_number",
        "prompt",
        "provider_request_id",
        "session_id",
        "span_id",
        "tool_arguments",
        "trace_id",
        "transcript",
        "turn_id",
        "url",
        "user_id",
    }
)

# Defense-in-depth against new PII-bearing keys: any attribute whose name
# *contains* one of these substrings is rejected even when it is not on the
# explicit forbidden list above. The explicit allow-list bypasses only this
# substring heuristic; the exact forbidden keys always take precedence.
_FORBIDDEN_SUBSTRINGS = (
    "transcript",
    "prompt",
    "content",
    "text",
    "body",
    "secret",
    "token",
)

_COUNTERS: dict[str, Any] = {}
_HISTOGRAMS: dict[str, Any] = {}
_GAUGES: dict[str, Any] = {}
_GAUGE_VALUES: dict[
    str,
    dict[tuple[tuple[str, str | int | float | bool], ...], tuple[int | float, dict[str, Any]]],
] = {}
_GAUGE_LOCK = threading.Lock()
_ACTIVE_SESSIONS = 0
_ACTIVE_SESSIONS_LOCK = threading.Lock()


_NOOP_SPAN = contextlib.nullcontext()


def span(
    name: str,
    attributes: Mapping[str, Any] | None = None,
) -> contextlib.AbstractContextManager[Any]:
    """Start an OTel span when tracing is configured; otherwise no-op."""
    if name not in SPAN_NAMES:
        raise ValueError(f"unknown EasyCat span name: {name}")
    tracer = _get_tracer()
    if tracer is None:
        _validate_attribute_keys(attributes, allowed_keys=SPAN_ATTRIBUTE_KEYS)
        return _NOOP_SPAN
    safe_attributes = sanitize_attributes(attributes, allowed_keys=SPAN_ATTRIBUTE_KEYS)
    return tracer.start_as_current_span(name, attributes=safe_attributes)


def record_histogram(
    name: str,
    value: float,
    attributes: Mapping[str, Any] | None = None,
) -> None:
    _record_metric(name, "histogram", value, attributes)


def increment_counter(
    name: str,
    value: float = 1,
    attributes: Mapping[str, Any] | None = None,
) -> None:
    _record_metric(name, "counter", value, attributes)


def observe_gauge(
    name: str,
    value: float,
    attributes: Mapping[str, Any] | None = None,
) -> None:
    _record_metric(name, "observable_gauge", value, attributes)


def tracing_available() -> bool:
    """Whether the optional OpenTelemetry tracing API is importable."""
    return _get_tracer() is not None


def metrics_available() -> bool:
    """Whether the optional OpenTelemetry metrics API is importable."""
    return _get_meter() is not None


def session_started() -> None:
    global _ACTIVE_SESSIONS
    with _ACTIVE_SESSIONS_LOCK:
        _ACTIVE_SESSIONS += 1
        active = _ACTIVE_SESSIONS
    observe_gauge("easycat.sessions.active", active)


def session_ended() -> None:
    global _ACTIVE_SESSIONS
    with _ACTIVE_SESSIONS_LOCK:
        _ACTIVE_SESSIONS = max(0, _ACTIVE_SESSIONS - 1)
        active = _ACTIVE_SESSIONS
    observe_gauge("easycat.sessions.active", active)


def sanitize_attributes(
    attributes: Mapping[str, Any] | None = None,
    *,
    allowed_keys: frozenset[str] = ALLOWED_ATTRIBUTE_KEYS,
) -> dict[str, Any]:
    if not attributes:
        return {}

    sanitized: dict[str, Any] = {}
    for key, value in attributes.items():
        normalized = str(key)
        _validate_attribute_key(normalized, allowed_keys)
        sanitized[normalized] = _safe_attribute_value(value)
    return sanitized


def _validate_attribute_key(normalized: str, allowed_keys: frozenset[str]) -> None:
    if normalized in FORBIDDEN_ATTRIBUTE_KEYS:
        raise ValueError(f"forbidden observability attribute: {normalized}")
    # Allow-list membership bypasses only the substring guard below.
    if normalized in allowed_keys:
        return
    lowered = normalized.lower()
    if any(substring in lowered for substring in _FORBIDDEN_SUBSTRINGS):
        raise ValueError(f"forbidden observability attribute: {normalized}")
    raise ValueError(f"unsupported observability attribute: {normalized}")


def _validate_attribute_keys(
    attributes: Mapping[str, Any] | None,
    *,
    allowed_keys: frozenset[str] = ALLOWED_ATTRIBUTE_KEYS,
) -> None:
    """Validate attribute names without materializing an exporter payload."""
    if not attributes:
        return
    for key in attributes:
        _validate_attribute_key(str(key), allowed_keys)


def _record_metric(
    name: str,
    expected_kind: MetricKind,
    value: float,
    attributes: Mapping[str, Any] | None,
) -> None:
    if METRIC_DEFINITIONS.get(name) != expected_kind:
        raise ValueError(f"metric {name!r} is not a {expected_kind}")
    meter = _get_meter()
    if meter is None:
        _validate_attribute_keys(attributes)
        return
    safe_attributes = sanitize_attributes(attributes)
    if expected_kind == "observable_gauge":
        _update_gauge_value(name, value, safe_attributes)
        instrument = _GAUGES.get(name)
        if instrument is None and hasattr(meter, "create_observable_gauge"):
            instrument = _create_observable_gauge(meter, name)
            _GAUGES[name] = instrument
        # Compatibility for test fakes or alternate metric APIs. Real OTel
        # observable gauges read values from the registered callback above.
        if instrument is not None and hasattr(instrument, "observe"):
            instrument.observe(value, attributes=safe_attributes)
        return

    if expected_kind == "counter":
        instrument = _COUNTERS.get(name)
        if instrument is None:
            instrument = meter.create_counter(name)
            _COUNTERS[name] = instrument
        instrument.add(value, attributes=safe_attributes)
        return
    if expected_kind == "histogram":
        instrument = _HISTOGRAMS.get(name)
        if instrument is None:
            instrument = meter.create_histogram(name)
            _HISTOGRAMS[name] = instrument
        instrument.record(value, attributes=safe_attributes)
        return


def _safe_attribute_value(value: Any) -> str | int | float | bool:
    if isinstance(value, bool | int | float):
        return value
    return str(value)


def _update_gauge_value(name: str, value: float, attributes: dict[str, Any]) -> None:
    key = tuple(sorted(attributes.items()))
    with _GAUGE_LOCK:
        values = _GAUGE_VALUES.setdefault(name, {})
        values[key] = (value, attributes)


def _create_observable_gauge(meter: Any, name: str) -> Any:
    callback = _gauge_callback(name)
    try:
        return meter.create_observable_gauge(name, callbacks=[callback])
    except TypeError:
        return meter.create_observable_gauge(name)


def _gauge_callback(name: str) -> Any:
    def _callback(_options: Any = None) -> list[Any]:
        with _GAUGE_LOCK:
            observations = list(_GAUGE_VALUES.get(name, {}).values())
        return [_make_observation(value, attributes) for value, attributes in observations]

    return _callback


def _make_observation(value: float, attributes: dict[str, Any]) -> Any:
    from opentelemetry.metrics import Observation

    return Observation(value, attributes=attributes)


@functools.cache
def _get_tracer() -> Any | None:
    """Return the process-wide tracer without repeating import discovery.

    OpenTelemetry returns a proxy tracer before an SDK provider is configured,
    so retaining this handle preserves its supported late-configuration
    behavior while keeping import machinery off every span's hot path.
    """
    try:
        from opentelemetry import trace
    except ImportError:
        return None
    return trace.get_tracer(INSTRUMENTATION_NAME)


@functools.cache
def _get_meter() -> Any | None:
    """Return the process-wide meter without repeating import discovery.

    Like tracers, OpenTelemetry meters are proxy objects that pick up a
    provider configured later in process startup.  Caching the proxy avoids a
    costly package lookup for every audio frame and journal append.
    """
    try:
        from opentelemetry import metrics
    except ImportError:
        return None
    return metrics.get_meter(INSTRUMENTATION_NAME)
