from __future__ import annotations

import asyncio
import json
import logging
import time
from typing import Any, Awaitable, Callable

from app.utils.call_timeline import CallTimeline
from app.ws_gateway.readiness import build_completed_stop_message
from app.ws_gateway.session_lifecycle import RoomLifecycleState, close_websocket


async def enable_client_audio(
    *,
    logger: logging.Logger,
    timeline: CallTimeline,
    client_audio_enabled: asyncio.Event,
    room_name: str,
    protocol: str,
) -> None:
    if client_audio_enabled.is_set():
        return

    client_audio_enabled.set()
    logger.info(
        "CLIENT_AUDIO_ENABLED | room=%s | protocol=%s",
        room_name,
        protocol,
    )
    timeline.emit("client_audio_enabled")


async def notify_client_audio_enabled(
    *,
    room: Any,
    logger: logging.Logger,
    timeline: CallTimeline,
    room_name: str,
    protocol: str,
    track_published: asyncio.Event,
    agent_ready: asyncio.Event,
    lifecycle: RoomLifecycleState,
    repeat: bool = False,
) -> None:
    await track_published.wait()
    await agent_ready.wait()

    last_generation = -1
    while True:
        await lifecycle.agent_connected.wait()
        generation = lifecycle.agent_generation
        if repeat and generation == last_generation:
            await asyncio.sleep(0.05)
            continue

        if lifecycle.agent_participant is None:
            if not repeat:
                return
            await asyncio.sleep(0.05)
            continue

        agent_ident = str(getattr(lifecycle.agent_participant, "identity", "") or "")
        if not agent_ident:
            if not repeat:
                return
            await asyncio.sleep(0.05)
            continue

        payload = {
            "type": "client_audio_enabled",
            "room": room_name,
            "protocol": protocol,
        }
        try:
            await room.local_participant.publish_data(
                json.dumps(payload, ensure_ascii=False),
                reliable=True,
                destination_identities=[agent_ident],
                topic="bridge.control",
            )
            logger.info("BRIDGE_CONTROL | sent client_audio_enabled | agent_identity=%s", agent_ident)
            timeline.emit(
                "bridge_control_sent",
                control_type="client_audio_enabled",
                agent_identity=agent_ident,
            )
            last_generation = generation
        except Exception:
            logger.debug("BRIDGE_CONTROL send failed", exc_info=True)

        if not repeat:
            return

        while lifecycle.agent_generation == generation and lifecycle.agent_connected.is_set():
            await asyncio.sleep(0.05)


async def stream_agent_audio_when_ready(
    *,
    agent_ready: asyncio.Event,
    lifecycle: RoomLifecycleState,
    agent_q: asyncio.Queue[bytes],
    activity: "AudioActivity",
    timeline: CallTimeline | None,
    stream_agent_audio: Callable[..., Awaitable[None]],
    repeat: bool = False,
) -> None:
    await agent_ready.wait()

    while True:
        await lifecycle.agent_connected.wait()
        participant = lifecycle.agent_participant
        generation = lifecycle.agent_generation
        if participant is None:
            if not repeat:
                return
            await asyncio.sleep(0.05)
            continue

        await stream_agent_audio(
            agent_q,
            participant,
            activity,
            timeline=timeline,
        )

        if not repeat:
            return

        while (
            lifecycle.agent_generation == generation
            and lifecycle.agent_participant is participant
            and lifecycle.agent_connected.is_set()
        ):
            await asyncio.sleep(0.05)


async def watch_mock_stop_after_first_audio(
    *,
    ws: Any,
    logger: logging.Logger,
    timeline: CallTimeline | None,
    room_name: str,
    protocol: str,
    first_agent_audio_sent: asyncio.Event,
    activity: "AudioActivity",
    agent_q: asyncio.Queue[bytes],
    silence_s: float,
    reason: str,
) -> None:
    await first_agent_audio_sent.wait()

    silence_s = max(0.0, float(silence_s))
    reason = str(reason or "").strip() or "DONE"
    logger.info(
        "MOCK_STOP_AFTER_FIRST_AUDIO_ARMED | room=%s | protocol=%s | silence_s=%.3f | reason=%s",
        room_name,
        protocol,
        silence_s,
        reason,
    )
    if timeline is not None:
        timeline.emit(
            "mock_stop_after_first_audio_armed",
            silence_ms=round(silence_s * 1000),
            reason=reason,
        )

    poll_interval_s = min(0.1, max(0.02, silence_s / 4 if silence_s > 0 else 0.02))
    while True:
        idle_for = time.monotonic() - activity.last_agent_out
        if agent_q.empty() and idle_for >= silence_s:
            break
        await asyncio.sleep(poll_interval_s)

    stop_message = build_completed_stop_message(reason)
    try:
        await ws.send_text(json.dumps(stop_message, ensure_ascii=False))
        logger.info(
            "MOCK_STOP_AFTER_FIRST_AUDIO_SENT | room=%s | protocol=%s | reason=%s",
            room_name,
            protocol,
            reason,
        )
        if timeline is not None:
            timeline.emit("mock_stop_after_first_audio_sent", reason=reason)
    except Exception:
        logger.debug("MOCK_STOP_AFTER_FIRST_AUDIO send failed", exc_info=True)
        return

    try:
        await close_websocket(ws, code=1000, reason=f"mock_stop:{reason}")
    except Exception:
        logger.debug("MOCK_STOP_AFTER_FIRST_AUDIO close failed", exc_info=True)
