"""Thread-safe Stream lifecycle primitives.

이 모듈은 GStreamer와 FastAPI에 의존하지 않는다. Pipeline worker는 ``RoleRunHandle``로
신호를 전달하고, ``StreamLifecycle``이 요청 상태와 실제 관측 상태를 분리해 소유한다.
"""

from __future__ import annotations

import threading
import time
from uuid import uuid4
from dataclasses import dataclass
from enum import Enum
from typing import Callable, Iterable, Mapping


class DesiredState(str, Enum):
    STOPPED = "stopped"
    RUNNING = "running"


class ObservedState(str, Enum):
    STOPPED = "stopped"
    STARTING = "starting"
    RUNNING = "running"
    STOPPING = "stopping"
    ERROR = "error"


class RoleState(str, Enum):
    INIT = "init"
    FIRST_FRAME = "first_frame"
    ERROR = "error"
    STOPPED = "stopped"


ERROR_MESSAGES = {
    "CONFIG_INVALID": "Stream configuration is invalid.",
    "DEPENDENCY_NOT_READY": "A required runtime dependency is not ready.",
    "SOURCE_CONNECT_FAILED": "The stream source could not be connected.",
    "RTP_PACKET_TIMEOUT": "The stream delivered no RTP packet within the allowed time.",
    "DECODE_FAILED": "The stream video could not be decoded.",
    "PIPELINE_INIT_FAILED": "The inference pipeline could not be initialized.",
    "FIRST_FRAME_TIMEOUT": "A required role did not deliver its first frame in time.",
    "ROLE_RUNTIME_FAILED": "A pipeline role failed while processing frames.",
    "STOP_CLEANUP_FAILED": "Stream resources could not be cleaned up completely.",
}

OBSERVABLE_SOURCE_STAGES = {
    "pipeline_state_changed",
    "decoded_frame_received",
    "preprocess_output",
    "infer_output",
    "consume_entered",
    "native_completed",
    "inference_frame_completed",
}


@dataclass(frozen=True)
class StreamFailure:
    error_code: str
    failure_stage: str
    role: str | None
    message: str

    def as_dict(self) -> dict:
        return {
            "error_code": self.error_code,
            "failure_stage": self.failure_stage,
            "role": self.role,
            "message": self.message,
        }


def make_stream_failure(
    error_code: str,
    failure_stage: str,
    role: str | None = None,
    _unsafe_detail: object | None = None,
) -> StreamFailure:
    """Create an allowlisted error without exposing source URLs or exception text."""
    if error_code not in ERROR_MESSAGES:
        raise ValueError(f"unsupported stream error code: {error_code}")
    return StreamFailure(
        error_code=error_code,
        failure_stage=str(failure_stage),
        role=str(role) if role is not None else None,
        message=ERROR_MESSAGES[error_code],
    )


def classify_pipeline_failure(
    error: BaseException,
    role: str,
    initialized: bool,
) -> StreamFailure:
    """Classify a worker exception while intentionally discarding its raw message."""
    lowered = str(error).lower()
    if any(token in lowered for token in (
        "libcontextra_nvinferserver_rfdetr",
        "custom_process",
        "custom processor",
        "custom_lib",
    )):
        return make_stream_failure(
            "PIPELINE_INIT_FAILED", "custom_processor_load", role, error
        )
    if any(token in lowered for token in (
        "rtsp",
        "connection refused",
        "connection timed out",
        "could not open resource",
        "resource not found",
    )):
        return make_stream_failure("SOURCE_CONNECT_FAILED", "source_connect", role, error)
    if not initialized:
        return make_stream_failure("PIPELINE_INIT_FAILED", "pipeline_start", role, error)
    return make_stream_failure("ROLE_RUNTIME_FAILED", "pipeline_runtime", role, error)


class StreamLifecycle:
    """Request/observation state owner for one Stream.

    STREAMERR-01: required role이 첫 Frame을 전달하기 전에는
    ``observed_state=running``으로 전이하지 않는다.
    """

    def __init__(self):
        self._lock = threading.RLock()
        self.desired_state = DesiredState.STOPPED
        self.observed_state = ObservedState.STOPPED
        self.run_generation = 0
        self.run_id: str | None = None
        self.last_error: StreamFailure | None = None
        self.cleanup_error: StreamFailure | None = None
        self._roles: dict[str, dict] = {}
        self._required_roles: set[str] = set()

    def begin(self, *, assigned_run_id=None, assigned_generation=None) -> int:
        with self._lock:
            if assigned_run_id is not None or assigned_generation is not None:
                if (self.run_generation != 0 or type(assigned_generation) is not int
                        or assigned_generation < 1 or not isinstance(assigned_run_id, str) or not assigned_run_id):
                    raise ValueError('Assigned identity requires a fresh child lifecycle.')
                self.run_generation, self.run_id = assigned_generation, assigned_run_id
            else:
                self.run_generation += 1
                self.run_id = str(uuid4())
            self.desired_state = DesiredState.RUNNING
            self.observed_state = ObservedState.STARTING
            self.last_error = None
            self.cleanup_error = None
            self._roles = {}
            self._required_roles = set()
            return self.run_generation

    def observe_child(self, report):
        with self._lock:
            if (report.get('run_generation'), report.get('run_id')) != (self.run_generation, self.run_id):
                return False
            self._roles = report.get('roles', {})
            if report.get('last_error'):
                self.last_error = StreamFailure(**report['last_error'])
            if self.desired_state == DesiredState.RUNNING:
                state = ObservedState(report['observed_state'])
                if state in (ObservedState.RUNNING, ObservedState.ERROR):
                    self.observed_state = state
            return True

    def configure_roles(
        self,
        generation: int,
        required_roles: Iterable[str],
        optional_roles: Iterable[str] = (),
    ) -> bool:
        with self._lock:
            if generation != self.run_generation:
                return False
            required = {str(role) for role in required_roles}
            optional = {str(role) for role in optional_roles} - required
            self._required_roles = required
            self._roles = {
                role: {
                    "state": RoleState.INIT.value,
                    "required": role in required,
                    "initialized": False,
                    "last_frame_at": None,
                    "source_stage": None,
                    "source_stage_at": None,
                    "source_stages": {},
                    "stage_counts": {},
                    "last_error": None,
                }
                for role in sorted(required | optional)
            }
            return True

    def signal_initialized(self, generation: int, role: str) -> bool:
        with self._lock:
            status = self._current_role(generation, role)
            if status is None or self.desired_state != DesiredState.RUNNING:
                return False
            status["initialized"] = True
            return True

    def signal_frame(self, generation: int, role: str, frame_time: float | None = None) -> bool:
        with self._lock:
            status = self._current_role(generation, role)
            if (status is None or self.desired_state != DesiredState.RUNNING
                    or status["state"] == RoleState.ERROR.value):
                return False
            status["initialized"] = True
            status["state"] = RoleState.FIRST_FRAME.value
            status["last_frame_at"] = float(frame_time if frame_time is not None else time.time())
            if self._required_roles and all(
                self._roles[item]["state"] == RoleState.FIRST_FRAME.value
                for item in self._required_roles
            ):
                self.observed_state = ObservedState.RUNNING
            return True

    def signal_source_stage(
        self,
        generation: int,
        role: str,
        stage: str,
        stage_time: float | None = None,
    ) -> bool:
        """Record only a stage proven by an actual Pipeline callback or Probe."""
        with self._lock:
            if stage not in OBSERVABLE_SOURCE_STAGES:
                return False
            status = self._current_role(generation, role)
            if (status is None or self.desired_state != DesiredState.RUNNING
                    or status["state"] == RoleState.ERROR.value):
                return False
            observed_at = float(
                stage_time if stage_time is not None else time.time()
            )
            status["source_stage"] = str(stage)
            status["source_stage_at"] = observed_at
            status["source_stages"][str(stage)] = observed_at
            status["stage_counts"][stage] = min(status["stage_counts"].get(stage, 0) + 1, 2**63-1)
            status.setdefault("stage_first_at", {}).setdefault(stage, observed_at)
            return True

    def signal_error(self, generation: int, failure: StreamFailure) -> bool:
        with self._lock:
            if generation != self.run_generation:
                return False
            role_status = self._roles.get(failure.role)
            if role_status is not None:
                role_status["state"] = RoleState.ERROR.value
                role_status["last_error"] = failure.as_dict()
            if role_status is None or role_status["required"]:
                self.observed_state = ObservedState.ERROR
            self.last_error = failure
            return True

    def signal_stopped(self, generation: int, role: str) -> bool:
        with self._lock:
            status = self._current_role(generation, role)
            if status is None:
                return False
            if status["state"] != RoleState.ERROR.value:
                status["state"] = RoleState.STOPPED.value
            return True

    def request_stop(self) -> int:
        with self._lock:
            self.desired_state = DesiredState.STOPPED
            if self.observed_state not in (ObservedState.STOPPED, ObservedState.ERROR):
                self.observed_state = ObservedState.STOPPING
            return self.run_generation

    def finish_stop(
        self, generation: int, cleanup_failure: StreamFailure | None = None
    ) -> bool:
        with self._lock:
            if generation != self.run_generation:
                return False
            self.desired_state = DesiredState.STOPPED
            if cleanup_failure is not None:
                self.observed_state = ObservedState.ERROR
                self.cleanup_error = self.cleanup_error or cleanup_failure
                self.last_error = self.last_error or cleanup_failure
            else:
                self.observed_state = ObservedState.STOPPED
            for status in self._roles.values():
                if cleanup_failure is None and status["state"] != RoleState.ERROR.value:
                    status["state"] = RoleState.STOPPED.value
            return True

    def record_cleanup(self, generation: int, failure: StreamFailure | None) -> None:
        with self._lock:
            if generation == self.run_generation and failure is not None:
                self.cleanup_error = self.cleanup_error or failure
                self.last_error = self.last_error or failure
                self.observed_state = ObservedState.ERROR

    def is_active(self, generation: int) -> bool:
        """Check the hot-path generation fence without copying diagnostic history."""
        with self._lock:
            return (generation == self.run_generation
                    and self.desired_state == DesiredState.RUNNING
                    and self.observed_state != ObservedState.ERROR)

    def snapshot(self) -> dict:
        with self._lock:
            return {
                "desired_state": self.desired_state.value,
                "observed_state": self.observed_state.value,
                "state": self.observed_state.value,
                "run_generation": self.run_generation,
                "run_id": self.run_id,
                "roles": {
                    role: {
                        **status,
                        "source_stages": dict(status["source_stages"]),
                        "stage_counts": dict(status["stage_counts"]),
                        "stage_first_at": dict(status.get("stage_first_at", {})),
                        "last_error": dict(status["last_error"])
                        if status["last_error"] is not None else None,
                    }
                    for role, status in self._roles.items()
                },
                "last_error": self.last_error.as_dict() if self.last_error else None,
                "cleanup_error": self.cleanup_error.as_dict() if self.cleanup_error else None,
            }

    def _current_role(self, generation: int, role: str) -> dict | None:
        if generation != self.run_generation:
            return None
        return self._roles.get(role)


RoleUpdate = Callable[[int, str, str, object | None], bool]


class RoleRunHandle:
    """Thread-safe signals and stop hooks for one Pipeline role."""

    def __init__(self, role: str, generation: int, on_update: RoleUpdate | None = None):
        from app.start_diagnostics import RoleDiagnostics
        self.diagnostic = RoleDiagnostics()
        self.role = str(role)
        self.generation = int(generation)
        self._on_update = on_update
        self._lock = threading.RLock()
        self._thread: threading.Thread | None = None
        self._pipeline = None
        self._started_pipeline = None
        self._initialized = False
        self._first_frame = False
        self._failure: StreamFailure | None = None
        self._stop_requested = False
        self._transition_lock = threading.Lock()
        self._stop_finished = False
        self._stop_failure: StreamFailure | None = None

    @property
    def initialized(self) -> bool:
        with self._lock:
            return self._initialized

    @property
    def first_frame(self) -> bool:
        with self._lock:
            return self._first_frame

    @property
    def failure(self) -> StreamFailure | None:
        with self._lock:
            return self._failure

    @property
    def stop_requested(self) -> bool:
        with self._lock:
            return self._stop_requested

    def bind_thread(self, thread: threading.Thread) -> None:
        with self._lock:
            self._thread = thread

    def bind_pipeline(self, pipeline) -> None:
        with self._lock:
            self._pipeline = pipeline
        self.diagnostic.mark("bound")

    def diagnostic_snapshot(self):
        snapshot = self.diagnostic.snapshot()
        if snapshot is not None:
            snapshot.update(role=self.role, generation=self.generation,
                            worker_alive=self.is_alive(), sampled_ns=time.monotonic_ns())
        return snapshot

    def mark_initialized(self) -> None:
        with self._lock:
            if self._failure is not None:
                return
            self._initialized = True
        self._emit("initialized")

    def mark_frame(self, frame_time: float | None = None, detection_count: int = 0) -> None:
        del detection_count
        with self._lock:
            if self._failure is not None or self._stop_requested:
                return
            self._initialized = True
            self._first_frame = True
        self._emit("frame", frame_time if frame_time is not None else time.time())

    def mark_source_stage(self, stage: str, stage_time: float | None = None) -> None:
        with self._lock:
            if self._failure is not None or self._stop_requested:
                return
        self._emit(
            "source_stage",
            (str(stage), stage_time if stage_time is not None else time.time()),
        )

    def mark_failure(self, failure: StreamFailure) -> None:
        with self._lock:
            if self._failure is not None or self._stop_requested:
                return
            self._failure = failure
        self._emit("error", failure)

    def mark_stopped(self) -> None:
        self._emit("stopped")

    def start_pipeline(self, on_message=None):
        # SDK 호출 중 callback이 사용하는 _lock은 잡지 않는다.
        with self._transition_lock:
            with self._lock:
                if self._stop_requested:
                    return None
                pipeline = self._pipeline
                self._started_pipeline = pipeline
            self.diagnostic.mark("start_enter")
            result = pipeline.start(on_message) if on_message is not None else pipeline.start()
            self.diagnostic.mark("start_returned")
            return result

    def stop(self) -> StreamFailure | None:
        with self._lock:
            self._stop_requested = True
        self.diagnostic.mark("stop_requested")
        with self._transition_lock:
            with self._lock:
                if self._stop_finished:
                    self.diagnostic.mark("stop_already_finished")
                    return self._stop_failure
                pipeline = self._pipeline
                matches_started = pipeline is self._started_pipeline and pipeline is not None
            try:
                if pipeline is not None:
                    self.diagnostic.mark("stop_enter", matches_started=matches_started)
                    pipeline.stop()
                    self.diagnostic.mark("stop_returned")
                else:
                    self.diagnostic.mark("stop_no_pipeline")
            except Exception as error:
                self.diagnostic.mark("stop_raised")
                self._stop_failure = make_stream_failure(
                    "STOP_CLEANUP_FAILED", "pipeline_stop", self.role, error)
            with self._lock:
                self._stop_finished = True
                return self._stop_failure

    def join(self, timeout: float | None = None) -> None:
        with self._lock:
            thread = self._thread
        if thread is not None and thread is not threading.current_thread():
            thread.join(timeout=timeout)

    def is_alive(self) -> bool:
        with self._lock:
            thread = self._thread
        return bool(thread and thread.is_alive())

    def _emit(self, event: str, payload: object | None = None) -> bool:
        if self._on_update is None:
            return True
        return self._on_update(self.generation, self.role, event, payload)


def stop_role_handles(handles: Mapping[str, RoleRunHandle], *, deadline: float | None = None) -> StreamFailure | None:
    """Stop and join every role; return the first bounded cleanup failure."""
    first_failure = None
    failed = set()
    for handle in list(handles.values()):
        failure = handle.stop()
        if failure is not None:
            failed.add(handle.role)
        if first_failure is None and failure is not None:
            first_failure = failure
    for handle in list(handles.values()):
        handle.join(timeout=5 if deadline is None else max(0, deadline - time.monotonic()))
        if handle.is_alive():
            first_failure = first_failure or make_stream_failure(
                "STOP_CLEANUP_FAILED", "worker_join", handle.role)
        elif handle.role not in failed:
            handle.mark_stopped()
    return first_failure


def lifecycle_command_payload(stream_id: str, lifecycle_status: Mapping[str, object]) -> dict:
    """Compatibility-preserving payload for Stream lifecycle commands."""
    return {
        "stream_id": stream_id,
        "state": lifecycle_status["state"],
        "desired_state": lifecycle_status["desired_state"],
        "observed_state": lifecycle_status["observed_state"],
        "run_generation": lifecycle_status["run_generation"],
        "run_id": lifecycle_status.get("run_id"),
        "last_error": lifecycle_status.get("last_error"),
        "cleanup_error": lifecycle_status.get("cleanup_error"),
    }