"""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
import re
import shutil
import subprocess
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.downloads.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_the_database_would_not_fail_is_not_reported_as_failed(sink, monkeypatch):
    """The job is still `running`: the runner reports this failure with the write's, and the
    worker fails the job (`child_exit`). A line here as well would be a second, and a false one."""

    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 sink.events("edit.job.failed") == []


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, "files_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", "/files/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)


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

    url = "https://example.org/v/Our%20Trip%3A%20Day%201.mp4"

    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("ERROR: [generic] Our Trip: Day 1: Unable to download")

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

    lines = [o for o in sink.objects() if "meta extract failed" in o["msg"]]
    assert lines and all("Day 1" 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)


@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available")
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["msg"] for o in sink.objects() if "gen skipped" in o["msg"]]
    # The whole message: a long tmp root would push a leaked argv's path past any length cut.
    assert len(lines) == 2
    assert all(
        re.fullmatch(r"\w+ gen skipped for asset 3: CalledProcessError \d+", m) for m in lines
    )


@pytest.mark.parametrize(
    "err",
    [
        subprocess.TimeoutExpired(["ffmpeg", "-i", "/files/Some_Channel/A_Private_Title.mp4"], 30),
        FileNotFoundError(2, "No such file", "/files/Some_Channel/A_Private_Title.jpg"),
        subprocess.CalledProcessError(1, ["ffmpeg", "-i", "/files/Some_Channel/x.mp4"]),
    ],
    ids=lambda e: type(e).__name__,
)
def test_an_image_step_failure_is_its_kind_and_nothing_it_quotes(err):
    from app.downloads import edit as editmod

    code = getattr(err, "returncode", None)
    assert editmod._step_failure(err) == " ".join(
        [type(err).__name__, *([str(code)] if code is not None else [])]
    )


def test_a_thumbnail_that_cannot_be_copied_names_no_file(sink, monkeypatch, tmp_path):
    from app.downloads import edit as editmod

    src = tmp_path / "thumb.jpg"
    src.write_bytes(b"\xff\xd8\xff rest")

    def refuse(a, b):
        raise PermissionError(13, "Permission denied", "/files/Some_Channel/A_Private_Title.jpg")

    monkeypatch.setattr(editmod.shutil, "copyfile", refuse)
    assert editmod.thumbnail_to_jpg(src, tmp_path / "out.jpg") is None

    (line,) = [o for o in sink.objects() if "thumbnail copy failed" in o["msg"]]
    assert "Some_Channel" not in json.dumps(line)


# --- the child's last stop ------------------------------------------------------------------------
def test_a_failure_the_child_cannot_record_is_one_scrubbed_line_and_a_non_zero_exit(
    sink, monkeypatch
):
    """The pipeline's failure write was refused. Python's own hook would print the chain raw —
    and its context is the download's failure, the URL in it. The worker fails the job on the
    exit code (`child_exit`), so this line does not claim the job failed."""
    from app.downloads import runner

    url = "https://example.org/v/Our%20Trip%3A%20Day%201.mp4"
    text = f"ERROR: [generic] Our Trip: Day 1: Unable to download webpage ({url})"

    def download(*a):
        try:
            raise yt_dlp.utils.DownloadError(text)
        except yt_dlp.utils.DownloadError:
            raise ConnectionError("the database went away")  # noqa: B904

    monkeypatch.setattr(pipeline, "_download", download)
    with pytest.raises(SystemExit) as exited:
        runner._run(5, 7, "download", "url", url, {}, 1)

    assert exited.value.code == 1
    (line,) = [o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]
    assert line["event"] == "download.child.failed"
    assert "Unable to download webpage" in line["exc"]
    # "Day 1" too: the fallback pattern alone stops at the first `: ` and leaves it.
    assert "Trip" not in json.dumps(line) and "Day 1" not in json.dumps(line)


def test_a_crash_the_database_will_not_record_carries_its_stack_once(monkeypatch, sink):
    claims = iter([42])

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

    def crash(job_id):
        raise RuntimeError("the one crash")

    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()

    stacks = [o["exc"] for o in sink.objects() if "exc" in o]
    assert sum("the one crash" in s for s in stacks) == 1
    (write,) = sink.events("download.worker.fail_write_failed")
    assert "the database went away" in write["exc"]


def test_a_watchdog_error_on_a_job_it_failed_is_that_jobs_failure_line(
    sink, finalized, monkeypatch
):
    """Through the real `_fail_after_kill` → `_fail_running`: the verdict it returns picks the
    event, so a lost `return` would log every such failure as a mere worker error."""

    class _Alive:
        pid = 4242
        returncode = None

        def wait(self, timeout=None):
            raise subprocess.TimeoutExpired("child", timeout or 0)

    def broken(job_id):
        raise KeyError("a watchdog bug")

    monkeypatch.setattr(worker, "_spawn_download_child", lambda job_id: _Alive())
    monkeypatch.setattr(worker, "_job_liveness", broken)
    monkeypatch.setattr(worker, "_stop_and_free", lambda *a: None)

    worker._run_supervised(42, 7)

    (line,) = [o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]
    assert (line["event"], line["attrs"]["cause"]) == ("download.job.failed", "supervisor_error")
    assert "a watchdog bug" in line["exc"]