from __future__ import annotations

from dataclasses import dataclass
import logging
import socket
import threading
import time
from typing import BinaryIO

from services.shared.core import utc_now_iso
from services.shared.db import get_session


LOGGER = logging.getLogger(__name__)


def _bridge_app():
    import services.asterisk_bridge_service.app as bridge_app

    return bridge_app


class AgiHangup(RuntimeError):
    pass


@dataclass(slots=True)
class AgiRequest:
    env: dict[str, str]
    args: list[str]
    peer: str | None = None


def parse_agi_request(reader: BinaryIO, *, peer: str | None = None) -> AgiRequest:
    env: dict[str, str] = {}
    while True:
        raw_line = reader.readline()
        if not raw_line:
            raise EOFError("AGI request ended before headers were fully received")
        line = raw_line.decode("utf-8", "replace").strip()
        if not line:
            break
        if ":" not in line:
            continue
        key, value = line.split(":", 1)
        env[key.strip()] = value.strip()

    args: list[str] = []
    index = 1
    while True:
        key = f"agi_arg_{index}"
        if key not in env:
            break
        args.append(str(env.get(key) or "").strip())
        index += 1
    return AgiRequest(env=env, args=args, peer=peer)


class SimpleAgiChannel:
    def __init__(self, reader: BinaryIO, writer: BinaryIO) -> None:
        self._reader = reader
        self._writer = writer

    def _command(self, command: str) -> str:
        self._writer.write(f"{command}\n".encode("utf-8"))
        self._writer.flush()
        while True:
            raw_line = self._reader.readline()
            if not raw_line:
                raise AgiHangup("AGI channel closed")
            line = raw_line.decode("utf-8", "replace").strip()
            if not line:
                continue
            if line.startswith("200 "):
                return line
            if line.startswith(("510", "511", "520")):
                raise RuntimeError(f"AGI command failed: {line}")

    def _result_code(self, response: str) -> int:
        marker = "result="
        if marker not in response:
            raise RuntimeError(f"Missing AGI result code: {response}")
        value = response.split(marker, 1)[1].split(" ", 1)[0].strip()
        try:
            result = int(value)
        except ValueError as exc:
            raise RuntimeError(f"Invalid AGI result code: {response}") from exc
        if result == -1:
            raise AgiHangup("Asterisk reported hangup")
        return result

    def verbose(self, message: str, *, level: int = 1) -> None:
        safe_message = str(message or "").replace('"', "'")
        self._command(f'VERBOSE "{safe_message}" {max(level, 1)}')

    def play_and_wait_for_digit(self, prompt_audio_key: str, timeout_ms: int) -> str | None:
        key = str(prompt_audio_key or "").strip()
        if not key:
            return None
        # GET OPTION plays the prompt and waits for a matching digit during or immediately after playback.
        result = self._result_code(
            self._command(f'GET OPTION {key} "0123456789#*" {max(timeout_ms, 1)}')
        )
        if result == 0:
            return None
        if 0 <= result <= 255:
            return chr(result)
        return str(result)

    def wait_for_digit(self, timeout_ms: int) -> str | None:
        result = self._result_code(self._command(f"WAIT FOR DIGIT {max(timeout_ms, 1)}"))
        if result == 0:
            return None
        if 0 <= result <= 255:
            return chr(result)
        return str(result)

    def set_variable(self, name: str, value: str | None) -> None:
        variable_name = str(name or "").strip()
        if not variable_name:
            return
        raw_value = str(value or "").replace('"', '\\"')
        self._result_code(self._command(f'SET VARIABLE {variable_name} "{raw_value}"'))


def _prompt_audio_keys(current_node: dict) -> list[str]:
    prompt_sequence = current_node.get("prompt_sequence")
    if isinstance(prompt_sequence, list):
        keys: list[str] = []
        for prompt in prompt_sequence:
            if not isinstance(prompt, dict):
                continue
            key = str(prompt.get("prompt_audio_key") or "").strip()
            if key:
                keys.append(key)
        if keys:
            return keys
    prompt_audio_key = str(current_node.get("prompt_audio_key") or "").strip()
    if prompt_audio_key:
        return [prompt_audio_key]
    return []


def _play_current_node_prompt(
    channel: SimpleAgiChannel,
    current_node: dict,
    *,
    timeout_ms: int,
) -> str | None:
    prompt_keys = _prompt_audio_keys(current_node)
    if not prompt_keys:
        return None
    # Keep inter-prompt gaps minimal: capture digits during each playback, then
    # perform a single normal wait after the full prompt sequence.
    for prompt_audio_key in prompt_keys:
        digit = channel.play_and_wait_for_digit(prompt_audio_key, 1)
        if digit is not None:
            return digit
    return None


@dataclass(slots=True)
class _CallContext:
    call_id: str
    linked_id: str | None
    queue_code: str
    fallback_queue_code: str | None
    entry_extension: str | None


def _parse_call_context(request: AgiRequest) -> _CallContext:
    call_id = str(request.args[0] if len(request.args) >= 1 else request.env.get("agi_uniqueid") or "").strip()
    linked_id = str(request.args[1] if len(request.args) >= 2 else "").strip() or None
    queue_code = str(request.args[2] if len(request.args) >= 3 else "").strip()
    fallback_queue_code = str(request.args[3] if len(request.args) >= 4 else "").strip() or None
    entry_extension = str(request.args[4] if len(request.args) >= 5 else "").strip() or None
    if not call_id:
        raise RuntimeError("Missing CallID for IVR FastAGI request")
    if not queue_code:
        raise RuntimeError("Missing queue_code for IVR FastAGI request")
    return _CallContext(
        call_id=call_id,
        linked_id=linked_id,
        queue_code=queue_code,
        fallback_queue_code=fallback_queue_code,
        entry_extension=entry_extension,
    )


def _queue_code_for_queue_id(queue_id: str | None) -> str | None:
    bridge = _bridge_app()
    target = str(queue_id or "").strip()
    if not target:
        return None
    for queue_code, mapped_queue_id in bridge._queue_map().items():
        if str(mapped_queue_id).strip() == target:
            return str(queue_code).strip() or None
    return None


def _resolve_transfer_extension(queue_code: str | None) -> str | None:
    bridge = _bridge_app()
    target_code = str(queue_code or "").strip()
    if not target_code:
        return None
    extension = str(bridge._transfer_target_map().get(target_code) or "").strip()
    return extension or None


def _set_transfer_target(
    channel: SimpleAgiChannel,
    *,
    queue_code: str | None,
    result: str,
    reuse_existing_call: bool = False,
) -> bool:
    target_extension = _resolve_transfer_extension(queue_code)
    if not target_extension:
        return False
    channel.set_variable("MVPCC_IVR_RESULT", result)
    channel.set_variable("MVPCC_IVR_TARGET_QUEUE_CODE", queue_code)
    channel.set_variable("MVPCC_IVR_TARGET_EXTENSION", target_extension)
    if reuse_existing_call and target_extension == "7100":
        channel.set_variable("MVPCC_AI_QUEUE_OVERRIDE", queue_code)
        channel.set_variable("MVPCC_AI_REUSE_CALL", "1")
    else:
        channel.set_variable("MVPCC_AI_QUEUE_OVERRIDE", "")
        channel.set_variable("MVPCC_AI_REUSE_CALL", "")
    return True


def _fallback_to_human(channel: SimpleAgiChannel, fallback_queue_code: str | None, *, reason: str) -> None:
    if _set_transfer_target(channel, queue_code=fallback_queue_code, result=reason):
        return
    channel.set_variable("MVPCC_IVR_RESULT", reason)


def _wait_for_call_link_context(call_id: str, *, stop_event: threading.Event | None = None) -> dict[str, str | None] | None:
    bridge = _bridge_app()
    deadline = time.monotonic() + bridge._ivr_call_link_wait_seconds()
    while True:
        session = get_session()
        try:
            link = bridge._find_call_link(session, call_id)
            if link is not None:
                return {
                    "interaction_id": str(link.interaction_id or "").strip() or None,
                    "queue_id": str(link.queue_id or "").strip() or None,
                    "queue_code": str(link.queue_code or "").strip() or None,
                    "caller_number": str(link.caller_number or "").strip() or None,
                    "caller_name": str(link.caller_name or "").strip() or None,
                }
        finally:
            session.close()
        if stop_event is not None and stop_event.is_set():
            return None
        if time.monotonic() >= deadline:
            return None
        time.sleep(0.25)


def _ivr_runtime_url(path: str) -> str:
    bridge = _bridge_app()
    return f"{bridge._ivr_service_url()}{path}"


def _start_session(call_context: _CallContext, *, interaction_id: str | None, queue_id: str) -> dict:
    bridge = _bridge_app()
    return bridge._post_json(
        _ivr_runtime_url("/ivr/sessions/start"),
        {
            "call_id": call_context.call_id,
            "queue_id": queue_id,
            "interaction_id": interaction_id,
        },
        timeout_seconds=bridge._forward_timeout_seconds(10.0),
        max_attempts=1,
        retry_backoff_seconds=0.0,
    )


def _send_dtmf(session_id: str, digit: str) -> dict:
    bridge = _bridge_app()
    return bridge._post_json(
        _ivr_runtime_url(f"/ivr/sessions/{session_id}/dtmf"),
        {"digit": digit},
        timeout_seconds=bridge._forward_timeout_seconds(10.0),
        max_attempts=1,
        retry_backoff_seconds=0.0,
    )


def _send_no_input(session_id: str) -> dict:
    bridge = _bridge_app()
    return bridge._post_json(
        _ivr_runtime_url(f"/ivr/sessions/{session_id}/no-input"),
        {},
        timeout_seconds=bridge._forward_timeout_seconds(10.0),
        max_attempts=1,
        retry_backoff_seconds=0.0,
    )


def _abandon_session(session_id: str | None) -> None:
    bridge = _bridge_app()
    normalized = str(session_id or "").strip()
    if not normalized:
        return
    try:
        bridge._post_json(
            _ivr_runtime_url(f"/ivr/sessions/{normalized}/abandon"),
            {},
            timeout_seconds=bridge._forward_timeout_seconds(5.0),
            max_attempts=1,
            retry_backoff_seconds=0.0,
        )
    except Exception:
        LOGGER.debug("Unable to abandon IVR session %s", normalized, exc_info=True)


def _resolve_completed_queue_code(session_payload: dict, fallback_queue_code: str | None) -> str | None:
    resolved_queue_code = str(session_payload.get("resolved_queue_code") or "").strip()
    if resolved_queue_code:
        return resolved_queue_code
    resolved_queue_id = str(session_payload.get("resolved_queue_id") or "").strip()
    return _queue_code_for_queue_id(resolved_queue_id) or fallback_queue_code


def _start_ai_session_for_ivr_transfer(
    *,
    call_context: _CallContext,
    interaction_id: str | None,
    queue_code: str | None,
    caller_number: str | None,
    caller_name: str | None,
) -> bool:
    bridge = _bridge_app()
    target_queue_code = str(queue_code or "").strip()
    if not target_queue_code:
        return False
    resolve_ai_config = getattr(bridge, "_ai_voice_config_for_queue", None)
    start_voice_ai_session = getattr(bridge, "_start_voice_ai_session", None)
    find_call_link = getattr(bridge, "_find_call_link", None)
    append_timeline = getattr(bridge, "_append_interaction_timeline", None)
    if not callable(resolve_ai_config) or not callable(start_voice_ai_session) or not callable(find_call_link):
        return False
    ai_config = resolve_ai_config(target_queue_code)
    if not ai_config:
        return False
    queue_id = str(bridge._queue_map().get(target_queue_code) or "").strip()
    active_interaction_id = str(interaction_id or "").strip()
    if not queue_id or not active_interaction_id:
        return False
    if _resolve_transfer_extension(target_queue_code) != "7100":
        return False

    started = start_voice_ai_session(
        call_id=call_context.call_id,
        linked_id=call_context.linked_id,
        interaction_id=active_interaction_id,
        queue_id=queue_id,
        queue_code=target_queue_code,
        caller_number=caller_number,
        caller_name=caller_name,
        ai_config=ai_config,
    )

    now = utc_now_iso()
    session = get_session()
    try:
        link = find_call_link(session, call_context.call_id)
        if link is not None:
            link.queue_code = target_queue_code
            link.queue_id = queue_id
            link.interaction_id = active_interaction_id
            link.voice_session_id = str((started or {}).get("voice_session_id") or "").strip() or link.voice_session_id
            link.ai_session_id = str((started or {}).get("ai_session_id") or "").strip() or link.ai_session_id
            link.ai_state = str((started or {}).get("status") or "queued").strip() or "queued"
            link.ai_handoff_reason = None
            link.ai_last_model_at = now
            link.updated_at = now
            session.commit()
        else:
            session.rollback()
    finally:
        session.close()

    try:
        if callable(append_timeline):
            append_timeline(
                interaction_id=active_interaction_id,
                action="ai.session_started",
                metadata={
                    "call_id": call_context.call_id,
                    "queue_code": target_queue_code,
                    "voice_session_id": str((started or {}).get("voice_session_id") or "").strip() or None,
                    "ai_session_id": str((started or {}).get("ai_session_id") or "").strip() or None,
                    "source": "ivr-transfer",
                },
            )
    except Exception:
        LOGGER.debug(
            "Unable to append IVR AI session timeline for %s",
            call_context.call_id,
            exc_info=True,
        )
    return True


def _handle_completed_session(
    channel: SimpleAgiChannel,
    call_context: _CallContext,
    session_payload: dict,
    *,
    interaction_id: str | None,
    caller_number: str | None,
    caller_name: str | None,
    fallback_queue_code: str | None,
) -> None:
    queue_code = _resolve_completed_queue_code(session_payload, fallback_queue_code)
    if _start_ai_session_for_ivr_transfer(
        call_context=call_context,
        interaction_id=interaction_id,
        queue_code=queue_code,
        caller_number=caller_number,
        caller_name=caller_name,
    ):
        if _set_transfer_target(
            channel,
            queue_code=queue_code,
            result="completed",
            reuse_existing_call=True,
        ):
            return
        _fallback_to_human(channel, fallback_queue_code, reason="fallback_missing_ai_transfer_target")
        return
    if _set_transfer_target(channel, queue_code=queue_code, result="completed"):
        return
    _fallback_to_human(channel, fallback_queue_code, reason="fallback_missing_transfer_target")


def run_ivr_call(
    channel: SimpleAgiChannel,
    request: AgiRequest,
    *,
    stop_event: threading.Event | None = None,
) -> None:
    bridge = _bridge_app()
    call_context: _CallContext | None = None
    session_id: str | None = None
    no_input_retries = 0
    invalid_retries = 0

    try:
        call_context = _parse_call_context(request)
        link_context = _wait_for_call_link_context(call_context.call_id, stop_event=stop_event) or {}
        queue_id = (
            str(link_context.get("queue_id") or "").strip()
            or str(bridge._queue_map().get(call_context.queue_code) or "").strip()
        )
        interaction_id = str(link_context.get("interaction_id") or "").strip() or None
        caller_number = str(link_context.get("caller_number") or "").strip() or None
        caller_name = str(link_context.get("caller_name") or "").strip() or None
        if not queue_id:
            raise RuntimeError(f"Unknown IVR queue_code: {call_context.queue_code}")

        started = _start_session(call_context, interaction_id=interaction_id, queue_id=queue_id)
        session_payload = started.get("session") or {}
        session_id = str(session_payload.get("session_id") or "").strip() or None
        current_node = started.get("current_node") if isinstance(started.get("current_node"), dict) else {}
        if session_id:
            channel.set_variable("MVPCC_IVR_SESSION_ID", session_id)
        channel.verbose(f"IVR session started for call {call_context.call_id}", level=1)

        while stop_event is None or not stop_event.is_set():
            digit: str | None = None
            digit = _play_current_node_prompt(
                channel,
                current_node,
                timeout_ms=bridge._ivr_dtmf_timeout_seconds() * 1000,
            )

            if digit is None:
                digit = channel.wait_for_digit(bridge._ivr_dtmf_timeout_seconds() * 1000)
            if digit is None:
                no_input_retries += 1
                invalid_retries = 0
                response = _send_no_input(session_id or "")
                session_payload = response.get("session") or {}
                current_node = response.get("current_node") if isinstance(response.get("current_node"), dict) else {}
                if response.get("completed"):
                    _handle_completed_session(
                        channel,
                        call_context,
                        session_payload,
                        interaction_id=interaction_id,
                        caller_number=caller_number,
                        caller_name=caller_name,
                        fallback_queue_code=call_context.fallback_queue_code,
                    )
                    return
                if no_input_retries > bridge._ivr_max_no_input_retries():
                    _abandon_session(session_id)
                    _fallback_to_human(channel, call_context.fallback_queue_code, reason="fallback_no_input")
                    return
                continue

            no_input_retries = 0
            current_options = {
                str(option.get("digit") or "").strip()
                for option in current_node.get("options", [])
                if isinstance(option, dict) and str(option.get("digit") or "").strip()
            }
            if digit not in current_options:
                invalid_retries += 1
            else:
                invalid_retries = 0

            response = _send_dtmf(session_id or "", digit)
            session_payload = response.get("session") or {}
            current_node = response.get("current_node") if isinstance(response.get("current_node"), dict) else {}
            if response.get("completed"):
                _handle_completed_session(
                    channel,
                    call_context,
                    session_payload,
                    interaction_id=interaction_id,
                    caller_number=caller_number,
                    caller_name=caller_name,
                    fallback_queue_code=call_context.fallback_queue_code,
                )
                return
            if digit not in current_options and invalid_retries > bridge._ivr_max_invalid_retries():
                _abandon_session(session_id)
                _fallback_to_human(channel, call_context.fallback_queue_code, reason="fallback_invalid")
                return

        _abandon_session(session_id)
        _fallback_to_human(channel, call_context.fallback_queue_code, reason="fallback_shutdown")
    except AgiHangup:
        _abandon_session(session_id)
    except Exception:
        LOGGER.exception(
            "IVR FastAGI call failed for %s",
            call_context.call_id if call_context is not None else "<unknown>",
        )
        _abandon_session(session_id)
        try:
            _fallback_to_human(
                channel,
                call_context.fallback_queue_code if call_context is not None else None,
                reason="fallback_error",
            )
        except Exception:
            LOGGER.debug(
                "Unable to set IVR fallback target for %s",
                call_context.call_id if call_context is not None else "<unknown>",
                exc_info=True,
            )


def _handle_client_connection(conn: socket.socket, addr: tuple[str, int], *, stop_event: threading.Event | None) -> None:
    peer = f"{addr[0]}:{addr[1]}"
    with conn:
        reader = conn.makefile("rb")
        writer = conn.makefile("wb")
        try:
            request = parse_agi_request(reader, peer=peer)
            run_ivr_call(SimpleAgiChannel(reader, writer), request, stop_event=stop_event)
        finally:
            try:
                reader.close()
            finally:
                writer.close()


def fastagi_loop(stop_event: threading.Event | None = None) -> None:
    bridge = _bridge_app()
    active_stop_event = stop_event or bridge._background_stop_event()
    if not bridge._bridge_enabled() or not bridge._ivr_fastagi_enabled():
        while not active_stop_event.is_set():
            if active_stop_event.wait(0.5):
                break
        return

    bind_host = bridge._ivr_fastagi_host()
    bind_port = bridge._ivr_fastagi_port()
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((bind_host, bind_port))
    server.listen()
    server.settimeout(1.0)
    LOGGER.info("IVR FastAGI listener started on %s:%s", bind_host, bind_port)
    try:
        while not active_stop_event.is_set():
            try:
                conn, addr = server.accept()
            except socket.timeout:
                continue
            except OSError:
                if active_stop_event.is_set():
                    break
                raise
            thread = threading.Thread(
                target=_handle_client_connection,
                args=(conn, addr),
                kwargs={"stop_event": active_stop_event},
                daemon=True,
                name=f"asterisk-ivr-fastagi-client-{addr[0]}:{addr[1]}",
            )
            thread.start()
    finally:
        try:
            server.close()
        except OSError:
            pass
