"""E22.5 review round 1: a failure is reported once, with its code, and a repeating one is not a
flood.

Each case here is a place where the first sweep got the line wrong in one of two directions: a
failure that left no line at all (a download the watchdog killed), or one that left too many (the
same edit failure twice, a warning per segment request, per settings read).
"""

import json
import logging
from pathlib import Path
from types import SimpleNamespace

import pytest
import yt_dlp

from app import logs, sysconfig, worker
from app.downloads import pipeline
from app.downloads.stated import StatedFailure
from app.models import DownloadJob, MediaAsset
from app.plex import stream as plex_stream


class _Sink(logging.Handler):
    def __init__(self):
        super().__init__(level=logging.NOTSET)
        self.lines: list[str] = []
        self.setFormatter(logs.JsonFormatter())
        self.addFilter(logs.ContextFilter())

    def emit(self, record):
        self.lines.append(self.format(record))

    def objects(self) -> list[dict]:
        return [json.loads(line) for line in self.lines]

    def events(self, code: str) -> list[dict]:
        return [o for o in self.objects() if o.get("event") == code]


@pytest.fixture
def sink():
    logger = logging.getLogger("siftlode")
    handler = _Sink()
    previous = logger.level
    logger.addHandler(handler)
    logger.setLevel(logs.TRACE)
    try:
        yield handler
    finally:
        logger.removeHandler(handler)
        logger.setLevel(previous)


class _NoDb:
    """A session that finds nothing and commits nothing — these cases are about the line."""

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

    def get(self, *a, **k):
        return None

    def commit(self):
        pass


@pytest.fixture
def finalized(monkeypatch):
    """Whether `finalize_if_running` fires: True unless a case says a route got there first."""
    state = {"fires": True}
    monkeypatch.setattr(worker, "SessionLocal", _NoDb)
    monkeypatch.setattr(pipeline, "SessionLocal", _NoDb)
    fire = lambda db, job_id, **fields: state["fires"]  # noqa: E731
    monkeypatch.setattr(worker.service, "finalize_if_running", fire)
    return state


# --- the worker's own failures ---------------------------------------------------------------
@pytest.mark.parametrize("cause", ["stalled", "timeout"])
def test_a_download_the_watchdog_kills_leaves_a_coded_line(sink, finalized, cause):
    """The child is killed, so it never reports; before this the worker did not either, and a
    stalled download failed with no line anywhere."""
    worker._fail_running(42, error="stopped", error_code="network", cause=cause)

    (line,) = sink.events("download.job.failed")
    assert line["level"] == "WARNING"
    assert line["attrs"] == {"job_id": 42, "error_code": "network", "cause": cause}


def test_a_job_a_route_already_stopped_is_not_reported_as_failed(sink, finalized):
    finalized["fires"] = False
    worker._fail_running(42, error="stopped", error_code="network", cause="stalled")

    assert sink.events("download.job.failed") == []


def test_a_caller_with_its_own_line_is_not_reported_twice(sink, finalized):
    worker._fail_running(42, error="Worker error", error_code=None)

    assert sink.events("download.job.failed") == []


def test_a_child_that_died_is_reported_with_its_exit_code(sink, finalized):
    worker._fail_running(
        42, error="exited", error_code="unknown", cause="child_exit", exit_code=137
    )

    (line,) = sink.events("download.job.failed")
    assert line["attrs"]["exit_code"] == 137


@pytest.mark.parametrize(
    ("fires", "code"), [(True, "download.job.failed"), (False, "download.worker.error")]
)
def test_a_worker_error_is_one_line_that_says_what_it_did_to_the_job(sink, fires, code):
    """A route's cancel that landed first is not a failure — the error is still reported, with
    its stack, but not as the job failing."""
    try:
        raise KeyError("a worker bug")
    except KeyError as exc:
        worker._report_worker_error(42, exc, "crash", lambda: fires)

    (line,) = [o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]
    assert line["event"] == code
    assert "KeyError" in line["exc"]


def test_a_worker_error_is_reported_even_when_failing_the_job_raises(sink):
    def db_down():
        raise ConnectionError("the database went away")

    with pytest.raises(ConnectionError):
        worker._report_worker_error(42, KeyError("a worker bug"), "crash", db_down)

    (line,) = sink.events("download.worker.error")
    assert line["attrs"]["cause"] == "crash"


def test_a_crash_the_database_will_not_record_does_not_end_the_worker_thread(monkeypatch, sink):
    """The crash handler writes to the database; if that raises too, the thread must go on to
    the next job — a dead thread is a download slot gone until the worker restarts."""
    claims = iter([42, 43])
    seen = []

    def claim():
        job = next(claims, None)
        if job is None:
            worker._stop.set()
        return job

    def crash(job_id):
        seen.append(job_id)
        raise RuntimeError("boom")

    def db_down(*a, **k):
        raise ConnectionError("the database went away")

    monkeypatch.setattr(worker, "_claim_job", claim)
    monkeypatch.setattr(worker, "_process_job", crash)
    monkeypatch.setattr(worker, "_fail_running", db_down)
    monkeypatch.setattr(worker.settings, "download_poll_seconds", 0)
    worker._stop.clear()
    try:
        worker._worker_loop(0)
    finally:
        worker._stop.clear()
    assert seen == [42, 43]


class _StoppedProc:
    """A child that is still tearing down when the supervisor polls, after a route paused it."""

    pid = 4242
    returncode = None

    def wait(self, timeout=None):
        import subprocess

        raise subprocess.TimeoutExpired("child", timeout or 0)


def test_the_worker_stopping_a_paused_child_is_not_a_second_abort_line(monkeypatch, sink):
    """The child logs `download.job.aborted` when it notices the pause itself; the worker's line
    is its own fact — it had to stop the child — so one pause is never two aborts."""
    monkeypatch.setattr(worker, "_spawn_download_child", lambda job_id: _StoppedProc())
    monkeypatch.setattr(worker, "_job_liveness", lambda job_id: ("paused", 1.0))
    monkeypatch.setattr(worker, "_stop_and_free", lambda *a: None)

    worker._run_supervised(42, 7)

    assert sink.events("download.job.aborted") == []
    assert len(sink.events("download.child.stopped")) == 1


# --- an edit failure, once ---------------------------------------------------------------------
def test_an_edit_failure_is_one_line_with_its_code_and_our_own_stack(sink, finalized):
    try:
        raise TypeError("a spec shape nobody expected")
    except TypeError as exc:
        pipeline._fail_edit(9, 3, str(exc), exc)

    (line,) = sink.events("edit.job.failed")
    assert line["attrs"] == {"job_id": 9, "asset_id": 3, "error_type": "TypeError"}
    assert "TypeError" in line["exc"]
    assert len([o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]) == 1


def test_an_edit_that_ffmpeg_refused_names_no_file_and_carries_no_stack(sink, finalized):
    """ffmpeg's stderr tail names its input, whose path is the channel and the title. The message
    is scrubbed; the stack is left off, because the message is the whole story."""
    msg = "ffmpeg failed (1): /downloads/Some_Channel/Season_2024/A_Private_Title_a3.mp4: Invalid data"
    pipeline._fail_edit(9, 3, msg, StatedFailure(msg))

    (line,) = sink.events("edit.job.failed")
    assert "A_Private_Title" not in json.dumps(line)
    assert "exc" not in line


def test_an_edit_failure_is_reported_even_when_its_write_fails(sink, monkeypatch):
    """The database is often what failed; then the write fails too, and the line must not go
    down with it."""

    class _Down:
        def __enter__(self):
            raise ConnectionError("the database went away")

        def __exit__(self, *exc):
            return False

    monkeypatch.setattr(pipeline, "SessionLocal", _Down)
    with pytest.raises(ConnectionError):
        pipeline._fail_edit(9, 3, "boom", KeyError("x"))

    assert len(sink.events("edit.job.failed")) == 1


class _EditDb(_NoDb):
    """An edit job whose source is ready, for `_process_edit` to get as far as its encode."""

    def get(self, model, _id):
        if model is DownloadJob:
            return SimpleNamespace(source_asset_id=7, edit_spec={}, user_id=1)
        if model is MediaAsset:
            return SimpleNamespace(
                status="ready", rel_path="src.mp4", width=640, height=360, duration_s=10
            )
        return None


def test_an_edit_that_crashes_is_said_once(sink, finalized, monkeypatch, tmp_path):
    """The line `_process_edit`'s handler used to log before `_fail_edit` logged it again."""
    (tmp_path / "src.mp4").write_bytes(b"x")
    monkeypatch.setattr(pipeline, "SessionLocal", _EditDb)
    monkeypatch.setattr(pipeline.settings, "download_root", str(tmp_path))
    monkeypatch.setattr(pipeline, "_edit_staging_dir", lambda asset_id: tmp_path / "staging")

    def odd_spec(*a, **k):
        raise TypeError("a spec shape nobody expected")

    monkeypatch.setattr(pipeline, "_set_job", odd_spec)

    pipeline._process_edit(9, 3)

    (line,) = [o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]
    assert line["event"] == "edit.job.failed"
    assert "TypeError" in line["exc"]


def test_an_edit_whose_source_is_gone_is_found_by_the_same_code(sink, finalized):
    pipeline._fail_edit(9, 3, "The source file is missing.")

    (line,) = sink.events("edit.job.failed")
    assert line["attrs"] == {"job_id": 9, "asset_id": 3}


# --- a download failure, scrubbed ---------------------------------------------------------------
def test_a_failed_download_line_does_not_put_back_what_the_adapter_took_out(sink):
    """yt-dlp logs its error through the adapter (scrubbed) and then raises the same text, which
    the pipeline logs again. Unscrubbed, that second line is the URL the first one hid."""
    msg = "ERROR: Unsupported URL: https://example.org/private/path?id=123"
    pipeline._log_download_failed(5, msg, "unknown", yt_dlp.utils.DownloadError(msg))

    (line,) = sink.events("download.job.failed")
    assert "private/path" not in json.dumps(line)
    assert "example.org/…" in line["msg"]
    # Unclassified, but yt-dlp's own error: its stack is the `except`, its foot the raw text.
    assert "exc" not in line


@pytest.mark.parametrize(
    "exc",
    [
        RuntimeError("dictionary changed size during iteration"),
        RecursionError("maximum recursion depth exceeded"),
        NotImplementedError(),
        FileNotFoundError(2, "No such file", "/downloads/Some_Channel/Season_2024/A_Private_Title"),
    ],
    ids=lambda e: type(e).__name__,
)
def test_a_bug_that_python_raises_keeps_its_stack_and_names_no_file(sink, exc):
    """Only the failures WE raise with the whole story in the message go without a stack. The
    same types are what Python raises for a bug of ours, and then the stack is the only lead."""
    try:
        raise exc
    except (RuntimeError, OSError) as caught:
        pipeline._log_download_failed(5, str(caught), "unknown", caught)

    (line,) = sink.events("download.job.failed")
    assert type(exc).__name__ in line["exc"]
    assert "A_Private_Title" not in json.dumps(line)


def test_a_stack_does_not_carry_the_download_error_it_was_raised_over(sink):
    """A database error raised inside `except DownloadError` has yt-dlp's text as its context,
    and a plain `exc_info` prints that link unscrubbed."""
    # A variable, as in the pipeline: a frame prints its SOURCE line, and that is code, not data.
    text = "ERROR: [youtube] dQw4w9WgXcQ: Sign in to confirm"
    try:
        try:
            raise yt_dlp.utils.DownloadError(text)
        except yt_dlp.utils.DownloadError:
            raise ConnectionError("server closed the connection unexpectedly")  # noqa: B904
    except ConnectionError as exc:
        pipeline._log_download_failed(5, str(exc), "unknown", exc)

    (line,) = sink.events("download.job.failed")
    assert "During handling of the above exception" in line["exc"]
    assert "Sign in to confirm" in line["exc"]
    assert "dQw4w9WgXcQ" not in json.dumps(line)


def test_an_unclassified_failure_of_our_own_keeps_its_stack(sink):
    try:
        raise KeyError("requested_formats")
    except KeyError as exc:
        pipeline._log_download_failed(5, str(exc), "unknown", exc)

    (line,) = sink.events("download.job.failed")
    assert "KeyError" in line["exc"]
    assert line["attrs"]["error_type"] == "KeyError"


def test_the_metadata_backfill_names_the_asset_and_not_the_download(sink, monkeypatch):
    from app.downloads import adapters, backfill

    url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"

    class _Refusing:
        def __init__(self, opts):
            pass

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

        def extract_info(self, target, download):
            raise yt_dlp.utils.DownloadError(
                f"ERROR: [youtube] dQw4w9WgXcQ: Video unavailable ({target})"
            )

    monkeypatch.setattr(backfill.yt_dlp, "YoutubeDL", _Refusing)
    assert backfill._extract_meta(url, adapters.YouTubeAdapter(), 11) is None

    lines = [o for o in sink.objects() if "meta extract failed" in o["msg"]]
    assert lines and all("dQw4w9WgXcQ" not in json.dumps(o) for o in lines)


# --- a repeating failure, once ------------------------------------------------------------------
def test_a_stale_setting_is_reported_once_not_on_every_read(sink, monkeypatch):
    spec = next(s for s in sysconfig.SPECS if s.type == "int")
    monkeypatch.setattr(sysconfig, "_raw_override", lambda db, key: "not-a-number")
    monkeypatch.setattr(sysconfig, "_coerce_warned", set())

    for _ in range(3):
        sysconfig.get(None, spec.key)

    assert len(sink.events("sysconfig.override_uncoercible")) == 1


class _RefusingDir:
    """A session directory on a disk that has gone away: every probe raises."""

    def __truediv__(self, name):
        return SimpleNamespace(exists=self._refuse)

    @staticmethod
    def _refuse():
        raise PermissionError("the disk went read-only")


def test_a_refusing_session_directory_is_reported_once_per_session(sink):
    """Every segment request probes, and hls.js retries each 404 — a line per probe is a flood."""
    session = plex_stream.HlsSession("k", Path("."), None, 0.0, ())
    session.dir = _RefusingDir()

    for seq in (3, 4, 5):
        assert not plex_stream.artifact_may_arrive(session, plex_stream.segment_name(seq, None))

    (line,) = sink.events("plex.stream.segment_probe_failed")
    assert line["attrs"]["reason"] == "PermissionError"


def test_a_retried_attempt_names_the_host_and_not_the_video(sink, monkeypatch, tmp_path):
    """The first attempt's DownloadError is logged when the fallback set takes over — one line
    before the adapter's scrubbed copy of the same text would have hidden it."""
    url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
    attempts = iter(
        [yt_dlp.utils.DownloadError(f"ERROR: [youtube] dQw4w9WgXcQ: Sign in ({url})"), {"id": "x"}]
    )

    class _YDL:
        def __init__(self, opts):
            pass

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

        def extract_info(self, target, download):
            outcome = next(attempts)
            if isinstance(outcome, Exception):
                raise outcome
            return outcome

    adapter = SimpleNamespace(
        client_sets=lambda authed: [["web"], ["tv"]],
        resolve_url=lambda target, opts: target,
        ydl_opts_extra=lambda clients: {},
        retriable=lambda text: True,
    )
    monkeypatch.setattr(pipeline.yt_dlp, "YoutubeDL", _YDL)
    monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: None)
    monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **fields: None)

    assert pipeline._extract_with_fallback(1, 1, url, {}, tmp_path, adapter) == {"id": "x"}

    (line,) = [o for o in sink.objects() if "retrying with fallback" in o["msg"]]
    assert "dQw4w9WgXcQ" not in json.dumps(line)


def test_a_failed_poster_or_storyboard_names_no_file(sink, tmp_path):
    """ffmpeg's CalledProcessError quotes its argv, and the source path in it is built from the
    channel and the title."""
    from app.downloads import edit as editmod

    rel = "Some_Channel/Season_2024/Some_Channel_-_20240101_-_A_Private_Title_a3.mp4"
    (tmp_path / rel).parent.mkdir(parents=True)
    (tmp_path / rel).write_bytes(b"not a video")
    asset = SimpleNamespace(id=3, rel_path=rel, duration_s=10.0, storyboard_path=None)

    assert editmod.ensure_poster(asset, str(tmp_path)) is None
    assert editmod.ensure_storyboard(asset, str(tmp_path)) is None

    lines = [o for o in sink.objects() if "gen skipped" in o["msg"]]
    assert len(lines) == 2
    assert all("Some_Channel" not in json.dumps(o) for o in lines)
    assert all("CalledProcessError" in o["msg"] for o in lines)