import asyncio
import struct

from app.ws_gateway import main as main_module


def _frame(value: int) -> bytes:
    return bytes([value % 256]) * main_module.BYTES_PER_FRAME


def _silent_frame() -> bytes:
    return b"\x00" * main_module.BYTES_PER_FRAME


def _voiced_frame(amp: int = 8000) -> bytes:
    samples = main_module.BYTES_PER_FRAME // 2
    return struct.pack("<%dh" % samples, *([amp] * samples))


def _config(
    *,
    enabled: bool = True,
    energy_shed_enabled: bool = False,
    energy_shed_max_excess_ms: int = 1000,
    silence_dbfs: float = -50.0,
) -> main_module.AudioInputBacklogConfig:
    return main_module.AudioInputBacklogConfig(
        shed_enabled=enabled,
        shed_threshold_ms=100,
        shed_keep_ms=40,
        latency_metrics_enabled=True,
        latency_alert_ms=1000,
        latency_log_interval_s=15.0,
        livekit_source_queue_size_ms=500,
        livekit_source_clear_on_shed=True,
        energy_shed_enabled=energy_shed_enabled,
        energy_shed_max_excess_ms=energy_shed_max_excess_ms,
        silence_dbfs=silence_dbfs,
    )


def test_audio_input_latency_snapshot_separates_raw_and_effective_delay() -> None:
    tracker = main_module.AudioInputLatencyTracker(_config())
    tracker.mark_enabled(100.0)

    fields = tracker.snapshot(frame_count=1064, now=110.872, queue_size_frames=0)

    assert fields["received_audio_ms"] == 21280
    assert fields["elapsed_since_enable_ms"] == 10872
    assert fields["raw_excess_ms"] == 10408
    assert fields["raw_lag_ms"] == 10408
    assert fields["excess_ms"] == 10408
    assert fields["lag_ms"] == 10408

    tracker.total_dropped_frames = 520
    fields = tracker.snapshot(frame_count=1064, now=110.872, queue_size_frames=0)

    assert fields["total_dropped_ms"] == 10400
    assert fields["saved_latency_ms"] == 10400
    assert fields["raw_excess_ms"] == 10408
    assert fields["raw_lag_ms"] == 10408
    assert fields["excess_ms"] == 8
    assert fields["lag_ms"] == 8


def test_shed_audio_queue_backlog_keeps_recent_frames_and_logs_savings(monkeypatch) -> None:
    events = []
    debug_events = []

    def fake_log_flow_event(logger, step, **payload):
        events.append((step, payload))

    monkeypatch.setattr(main_module, "log_flow_event", fake_log_flow_event)
    audio_q: asyncio.Queue[bytes] = asyncio.Queue()
    for value in range(8):
        audio_q.put_nowait(_frame(value))
    tracker = main_module.AudioInputLatencyTracker(
        _config(),
        debug_event_publisher=lambda event, payload: debug_events.append(
            (event, payload)
        ),
    )
    tracker.mark_enabled(0.0)

    dropped = main_module._shed_audio_queue_backlog(
        audio_q,
        config=_config(),
        tracker=tracker,
        source="test",
        frame_count=8,
        now=1.0,
    )

    assert dropped.dropped == 6
    assert audio_q.qsize() == 2
    assert [audio_q.get_nowait(), audio_q.get_nowait()] == [_frame(6), _frame(7)]
    assert events[-1][0] == "audio_in_latency_shed"
    assert events[-1][1]["dropped_ms"] == 120
    assert events[-1][1]["saved_ms"] == 120
    assert events[-1][1]["saved_latency_ms"] == 120
    assert events[-1][1]["queue_before_ms"] == 160
    assert events[-1][1]["queue_after_ms"] == 40
    assert events[-1][1]["total_dropped_ms"] == 120
    assert debug_events[-1][0] == "bridge.audio_in.shed"
    assert debug_events[-1][1]["dropped_ms"] == 120


def test_shed_audio_queue_backlog_is_disabled_by_flag(monkeypatch) -> None:
    events = []
    monkeypatch.setattr(
        main_module,
        "log_flow_event",
        lambda logger, step, **payload: events.append((step, payload)),
    )
    audio_q: asyncio.Queue[bytes] = asyncio.Queue()
    for value in range(8):
        audio_q.put_nowait(_frame(value))

    dropped = main_module._shed_audio_queue_backlog(
        audio_q,
        config=_config(enabled=False),
        tracker=main_module.AudioInputLatencyTracker(_config(enabled=False)),
        source="test",
        frame_count=8,
        now=1.0,
    )

    assert dropped.dropped == 0
    assert audio_q.qsize() == 8
    assert events == []


def test_audio_input_backlog_config_defaults_to_enabled(monkeypatch) -> None:
    for name in (
        "AUDIO_IN_BACKLOG_SHED_ENABLED",
        "AUDIO_IN_BACKLOG_SHED_THRESHOLD_MS",
        "AUDIO_IN_BACKLOG_SHED_KEEP_MS",
        "AUDIO_IN_LATENCY_METRICS_ENABLED",
        "AUDIO_IN_LATENCY_ALERT_MS",
        "AUDIO_IN_LATENCY_LOG_INTERVAL_S",
        "LIVEKIT_AUDIO_SOURCE_QUEUE_SIZE_MS",
        "LIVEKIT_AUDIO_SOURCE_CLEAR_ON_SHED",
    ):
        monkeypatch.delenv(name, raising=False)

    config = main_module.audio_input_backlog_config_from_env()

    assert config.shed_enabled is True
    assert config.shed_threshold_ms == 500
    assert config.shed_keep_ms == 300
    assert config.latency_metrics_enabled is True
    assert config.livekit_source_queue_size_ms == 500
    assert config.livekit_source_clear_on_shed is True
    assert config.config_source == "env"
    assert config.call_config_overrides == ()


def test_audio_input_backlog_config_preserves_legacy_source_queue_when_disabled(monkeypatch) -> None:
    monkeypatch.setenv("AUDIO_IN_BACKLOG_SHED_ENABLED", "0")
    monkeypatch.delenv("LIVEKIT_AUDIO_SOURCE_QUEUE_SIZE_MS", raising=False)
    monkeypatch.delenv("LIVEKIT_AUDIO_SOURCE_CLEAR_ON_SHED", raising=False)

    config = main_module.audio_input_backlog_config_from_env()

    assert config.shed_enabled is False
    assert config.livekit_source_queue_size_ms == 5000
    assert config.livekit_source_clear_on_shed is False
    assert config.config_source == "env"


def test_audio_input_backlog_config_prefers_call_config_and_falls_back_to_env(monkeypatch) -> None:
    monkeypatch.setenv("AUDIO_IN_BACKLOG_SHED_ENABLED", "0")
    monkeypatch.setenv("AUDIO_IN_BACKLOG_SHED_THRESHOLD_MS", "900")
    monkeypatch.setenv("AUDIO_IN_BACKLOG_SHED_KEEP_MS", "600")
    monkeypatch.setenv("AUDIO_IN_LATENCY_METRICS_ENABLED", "1")
    monkeypatch.setenv("AUDIO_IN_LATENCY_ALERT_MS", "1200")
    monkeypatch.setenv("AUDIO_IN_LATENCY_LOG_INTERVAL_S", "20")
    monkeypatch.setenv("LIVEKIT_AUDIO_SOURCE_QUEUE_SIZE_MS", "2000")
    monkeypatch.setenv("LIVEKIT_AUDIO_SOURCE_CLEAR_ON_SHED", "0")

    config = main_module.audio_input_backlog_config_from_env(
        {
            "ws": {
                "audioInputBacklogShedEnabled": True,
                "audioInputBacklogShedThresholdMs": 500,
                "audioInputBacklogShedKeepMs": 300,
                "audioInputLatencyAlertMs": 800,
                "livekitAudioSourceQueueSizeMs": 400,
                "livekitAudioSourceClearOnShed": True,
                "audioInputBacklogEnergyShedEnabled": True,
                "audioInputBacklogEnergyShedMaxExcessMs": 600,
                "audioInputBacklogSilenceDbfs": -60,
            }
        }
    )

    assert config.shed_enabled is True
    assert config.shed_threshold_ms == 500
    assert config.shed_keep_ms == 300
    assert config.latency_metrics_enabled is True
    assert config.latency_alert_ms == 800
    assert config.latency_log_interval_s == 20.0
    assert config.livekit_source_queue_size_ms == 400
    assert config.livekit_source_clear_on_shed is True
    assert config.config_source == "call_config"
    assert config.call_config_overrides == (
        "audio_in_backlog_shed_enabled",
        "audio_in_backlog_shed_threshold_ms",
        "audio_in_backlog_shed_keep_ms",
        "audio_in_latency_alert_ms",
        "livekit_audio_source_queue_size_ms",
        "livekit_audio_source_clear_on_shed",
        "audio_in_backlog_energy_shed_enabled",
        "audio_in_backlog_energy_shed_max_excess_ms",
        "audio_in_backlog_silence_dbfs",
    )


def test_shed_energy_guided_drops_only_silence_under_small_excess(monkeypatch) -> None:
    events = []
    monkeypatch.setattr(
        main_module,
        "log_flow_event",
        lambda logger, step, **payload: events.append((step, payload)),
    )
    config = _config(energy_shed_enabled=True, energy_shed_max_excess_ms=1000)
    # [V, S, V, S, V, S, V, S] -> 4 de voz, 4 de silencio
    frames = []
    for _ in range(4):
        frames.append(_voiced_frame())
        frames.append(_silent_frame())
    audio_q: asyncio.Queue[bytes] = asyncio.Queue()
    for frame in frames:
        audio_q.put_nowait(frame)
    tracker = main_module.AudioInputLatencyTracker(config)
    tracker.mark_enabled(0.0)

    dropped = main_module._shed_audio_queue_backlog(
        audio_q,
        config=config,
        tracker=tracker,
        source="test",
        frame_count=8,
        now=1.0,
    )

    # excesso pequeno (120 ms < 1000): descarta so os 4 silencios, preserva a voz.
    assert dropped.dropped == 4
    assert audio_q.qsize() == 4
    remaining = [audio_q.get_nowait() for _ in range(4)]
    assert all(f == _voiced_frame() for f in remaining)
    assert events[-1][0] == "audio_in_latency_shed"
    assert events[-1][1]["mode"] == "energy"
    assert events[-1][1]["dropped_silent_frames"] == 4
    assert events[-1][1]["dropped_voiced_frames"] == 0
    assert events[-1][1]["content_policy"] == "silence_only"
    assert events[-1][1]["sync_action"] == "compress_silence"


def test_shed_falls_back_to_blind_under_large_excess(monkeypatch) -> None:
    events = []
    monkeypatch.setattr(
        main_module,
        "log_flow_event",
        lambda logger, step, **payload: events.append((step, payload)),
    )
    # max_excess minusculo: qualquer excesso cai no regime de rajada (cego)
    config = _config(energy_shed_enabled=True, energy_shed_max_excess_ms=40)
    audio_q: asyncio.Queue[bytes] = asyncio.Queue()
    for value in range(8):
        audio_q.put_nowait(_frame(value))
    tracker = main_module.AudioInputLatencyTracker(config)
    tracker.mark_enabled(0.0)

    dropped = main_module._shed_audio_queue_backlog(
        audio_q,
        config=config,
        tracker=tracker,
        source="test",
        frame_count=8,
        now=1.0,
    )

    assert dropped.dropped == 6
    assert audio_q.qsize() == 2
    assert [audio_q.get_nowait(), audio_q.get_nowait()] == [_frame(6), _frame(7)]
    assert events[-1][1]["mode"] == "blind"
    assert events[-1][1]["content_policy"] == "oldest_frames"
    assert events[-1][1]["sync_action"] == "drop_audio_to_catch_up"


def test_audio_input_backlog_config_energy_defaults(monkeypatch) -> None:
    for name in (
        "AUDIO_IN_BACKLOG_ENERGY_SHED_ENABLED",
        "AUDIO_IN_BACKLOG_ENERGY_SHED_MAX_EXCESS_MS",
        "AUDIO_IN_BACKLOG_SILENCE_DBFS",
    ):
        monkeypatch.delenv(name, raising=False)

    config = main_module.audio_input_backlog_config_from_env()

    assert config.energy_shed_enabled is True
    assert config.energy_shed_max_excess_ms == 1000
    assert config.silence_dbfs == -60.0
    assert config.silence_rms_threshold == 32
