"""E2E: Users, per-user overrides, triggering runs, and reading the result.

The through-line: a control the owner touches in the browser must change what the ENGINE does
on the next run. A toggle that only repaints the UI is the failure this file exists to catch.
"""

from __future__ import annotations

import re
import time
from collections.abc import Callable

import pytest
from playwright.sync_api import Page, expect

from tests.e2e.conftest import ShortlistApp, build_real_rows
from tests.fakes.fake_plex import DEMO_MOVIES, DEMO_SHOWS

pytestmark = pytest.mark.e2e


def open_rows(page) -> None:
    """Open every collapsed row card on a run page.

    A person's results live inside the row that built them, and a row with siblings starts closed.
    Re-queries each time rather than iterating a snapshot: clicking one row re-renders the list, which
    detaches every handle taken before it.
    """
    for _ in range(8):
        closed = page.get_by_role("button", expanded=False)
        if not closed.count():
            return
        closed.first.click()


LOAD = 20_000
SLOW = 90_000


def _users_by_name(app: ShortlistApp) -> dict[str, dict]:
    return {user["username"]: user for user in app.api("GET", "/api/users").json()}


def _wait_until(condition: Callable[[], bool], message: str, timeout_s: float = 10.0) -> None:
    """Poll the API until a persisted value shows up — robust to the auto-save debounce, which the
    'Saved' indicator can race (it flips before the PUT the debounce fired has fully committed)."""
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        if condition():
            return
        time.sleep(0.25)
    raise AssertionError(message)


class TestTheOwnerIsAUserToo:
    """The person running Shortlist gets a row like anyone else (issue #1).

    plex.tv's `/api/users` returns everyone the server is shared WITH and never its owner, so
    before this the owner had no user row at all — and a one-person server built nothing.
    """

    def test_a_server_where_only_the_owner_is_enabled_still_builds_a_row(
        self, page: Page, app: ShortlistApp, reset_fake_plex
    ):
        """The solo-server case end to end: sync finds the owner, and a run gives them real picks
        from their own watch history — not the popular-titles fallback, which is what an empty
        history would silently produce."""
        state = reset_fake_plex
        app.api("POST", "/api/users/sync")

        owner = _users_by_name(app)[state.owner_username]
        assert owner["user_type"] == "owner"
        assert owner["enabled"] is False  # arrives switched off, like every other new user

        for user in _users_by_name(app).values():
            app.api("PATCH", f"/api/users/{user['id']}", json={"enabled": user["id"] == owner["id"]})
        run = build_real_rows(app)

        rows = [c for c in state.collections.values() if any(lbl.lower() == "shortlist_steve" for lbl in c.labels)]
        assert rows, "the owner is the only enabled user and got no row at all"
        for collection in rows:
            assert collection.item_keys
            assert collection.promoted_own_home, "the owner's row never reached the owner's own Home"
        assert run["stats"]["users_ok"] == 1

        # Their picks come from THEIR history — PMS files the owner's plays under a local account
        # id, so asking by plex.tv id returns nothing and this row would be the cold-start one.
        page.goto(f"/users/{owner['id']}")
        expect(page.get_by_role("heading", name=state.owner_username)).to_be_visible(timeout=LOAD)
        expect(page.get_by_text("New viewer")).to_have_count(0)

    def test_the_owners_row_is_excluded_on_every_share_plex_will_accept_one_for(
        self, app: ShortlistApp, reset_fake_plex
    ):
        """The owner's own filter is never written (Plex cannot restrict them) — that must not be
        mistaken for 'this row needs no hiding'. Everyone else still has to exclude it.

        Everyone Plex ALLOWS, that is. A managed account with a parental profile is the one exception:
        plex.tv answers 422 to any label restriction while the profile is set, so no exclude can exist
        on that share no matter what Shortlist does. Asserting it anyway would be asserting a write
        that cannot succeed. What must hold instead is that the account is REPORTED rather than
        silently passed over — which is the rest of this test (#76)."""
        state = reset_fake_plex
        app.api("POST", "/api/users/sync")
        owner = _users_by_name(app)[state.owner_username]
        app.api("PATCH", f"/api/users/{owner['id']}", json={"enabled": True})

        build_real_rows(app)

        unfilterable = {u.id for u in state.users.values() if u.restriction_profile}
        assert unfilterable, "the roster must carry a profiled account, or this proves only the easy half"
        for account_id, user in state.users.items():
            if account_id in unfilterable:
                continue
            for field_name in ("filterMovies", "filterTelevision"):
                assert "Shortlist_steve" in user.filters[field_name], (
                    f"account {account_id} can see the owner's row in {field_name}"
                )

        # The exception is not a blind spot: the run looked through that account's own eyes and said
        # what it found. A skip that reported nothing is the bug this whole check exists for.
        listed = _users_by_name(app)
        for account_id in unfilterable:
            username = state.users[account_id].username
            assert listed[username]["unhidden_rows"] > 0, (
                f"{username} was skipped by the filter write AND reported as seeing nothing — "
                "the exact silence that hid this for a year"
            )


class TestUsers:
    def test_selected_pause_survives_reload_without_changing_access_or_other_people(
        self, page: Page, app: ShortlistApp
    ):
        before = _users_by_name(app)
        page.goto("/users")
        page.get_by_label("Search users").fill("sarah")
        # The checkboxes appear once selecting is switched on.
        page.get_by_role("button", name="Select people", exact=True).click()
        page.get_by_role("checkbox", name="Select sarah", exact=True).check()
        page.get_by_role("button", name="Pause rebuilding", exact=True).click()
        _wait_until(
            lambda: _users_by_name(app)["sarah"]["prefs"].get("paused") is True,
            "selected pause must reach the stored preferences",
        )
        page.reload()
        sarah = page.get_by_role("row").filter(has=page.get_by_role("link", name="sarah", exact=True))
        expect(sarah.get_by_text("Paused", exact=True)).to_be_visible(timeout=LOAD)
        paused = _users_by_name(app)
        assert {name: user["enabled"] for name, user in paused.items()} == {
            name: user["enabled"] for name, user in before.items()
        }
        for name in before.keys() - {"sarah"}:
            assert paused[name]["prefs"] == before[name]["prefs"]

        page.get_by_role("button", name="Select people", exact=True).click()  # a reload leaves selecting off
        page.get_by_role("checkbox", name="Select sarah", exact=True).check()
        page.get_by_role("button", name="Resume rebuilding", exact=True).click()
        _wait_until(
            lambda: _users_by_name(app)["sarah"]["prefs"].get("paused") is False,
            "selected resume must reach the stored preferences",
        )
        assert _users_by_name(app)["sarah"]["enabled"] == before["sarah"]["enabled"]

    def test_disabling_a_user_persists_and_leaves_the_others_alone(self, page: Page, app: ShortlistApp):
        page.goto("/users")
        jess = page.get_by_role("switch", name="Shortlist row for jess")
        expect(jess).to_be_checked(timeout=LOAD)

        jess.click()
        expect(jess).not_to_be_checked(timeout=LOAD)

        page.reload()
        expect(page.get_by_role("switch", name="Shortlist row for jess")).not_to_be_checked(timeout=LOAD)

        # Exactly one user changed — a broadcast PATCH would pass a "it persisted" test too.
        users = _users_by_name(app)
        assert users["jess"]["enabled"] is False
        assert users["sarah"]["enabled"] is True
        assert users["mike"]["enabled"] is True

    def test_per_user_overrides_persist(self, page: Page, app: ShortlistApp):
        sarah_id = _users_by_name(app)["sarah"]["id"]
        page.goto(f"/users/{sarah_id}")
        expect(page.get_by_role("heading", name="sarah")).to_be_visible(timeout=LOAD)

        # Pause is a user-level control in the header (switch is "on" when active).
        paused = page.get_by_role("switch", name="Pause or resume sarah")
        expect(paused).to_be_checked(timeout=LOAD)
        paused.click()
        expect(paused).not_to_be_checked(timeout=LOAD)

        # Size is a PER-ROW override now: open a row's drawer, switch on a custom size and type 10.
        # The drawer AUTO-SAVES (no Save button — the whole app is one save paradigm).
        page.get_by_role("button", name="Customize for this person").first.click()
        page.get_by_role("switch", name="Custom row size for this person").click()
        titles = page.get_by_label("Titles for this person")
        titles.fill("10")
        titles.blur()  # the free number field commits on blur
        expect(page.get_by_text("Saved").first).to_be_visible(timeout=LOAD)

        # Pause reaches the user's prefs; the size override lives on the row, not user prefs. Poll the
        # API for the persisted value rather than trusting the indicator, which races the debounce.
        assert _users_by_name(app)["sarah"]["prefs"]["paused"] is True
        expect_row_size = 10

        def _size_persisted() -> bool:
            rows = app.api("GET", f"/api/users/{sarah_id}/rows").json()
            return any(r["override"].get("row_size") == expect_row_size for r in rows)

        _wait_until(_size_persisted, "the per-row size override must persist")
        # Exactly this user changed — mike's prefs stay empty.
        assert _users_by_name(app)["mike"]["prefs"] == {}

    def test_a_paused_user_is_skipped_by_the_next_run(self, page: Page, app: ShortlistApp, reset_fake_plex):
        """The override has to reach the ENGINE, not just the users table."""
        state = reset_fake_plex
        sarah_id = _users_by_name(app)["sarah"]["id"]

        page.goto(f"/users/{sarah_id}")
        expect(page.get_by_role("heading", name="sarah")).to_be_visible(timeout=LOAD)
        # The header pause switch is "on" when active; clicking it pauses this user.
        paused = page.get_by_role("switch", name="Pause or resume sarah")
        expect(paused).to_be_checked(timeout=LOAD)
        paused.click()
        expect(paused).not_to_be_checked(timeout=LOAD)

        run = build_real_rows(app)
        built = {result["slug"] for result in app.api("GET", f"/api/runs/{run['id']}").json()["users"]}
        assert built == {"mike", "jess"}, "a paused user must not be rebuilt"

        labels = {label.lower() for c in state.collections.values() for label in c.labels}
        # sarah's OWNER label is the one that must be absent — that is what "not rebuilt" looks like
        # on the server. The constant `shortlist` label is on every row we write and names nobody, so
        # it is expected here and says nothing about who was built.
        assert labels == {"shortlist", "shortlist_mike", "shortlist_jess"}, (
            "sarah's OWNER label must be absent — that is what 'not rebuilt' looks like on the server"
        )
        assert "shortlist_sarah" not in labels


class TestRuns:
    def test_a_run_started_from_the_ui_lands_in_runs_with_per_user_rows(
        self, page: Page, app: ShortlistApp, reset_fake_plex
    ):
        state = reset_fake_plex

        page.goto("/runs")
        expect(page.get_by_text("No run history")).to_be_visible(timeout=LOAD)
        page.get_by_role("button", name="Run all rows now").click()

        # The row appears the moment the run is queued — the owner is never left guessing.
        run_link = page.get_by_role("link", name="#1")
        expect(run_link).to_be_visible(timeout=LOAD)
        expect(page.get_by_role("cell", name="manual")).to_be_visible()

        app.wait_for_run(1)
        page.reload()
        # The result cell leads with the title-cased status ("OK · 1 warning") — scoped to the table,
        # since the page's stats bar also shows "OK" as the last-run status. The users cell keeps "3 ok".
        expect(page.get_by_role("table").get_by_role("cell", name=re.compile(r"^OK\b"))).to_be_visible(timeout=LOAD)
        expect(page.get_by_role("cell", name="3 ok")).to_be_visible()
        # 5 rows for 3 users: sarah and the cold-start jess each get one per library; mike watches only TV.
        assert len(state.collections) == 5

        page.get_by_role("link", name="#1").click()
        expect(page.get_by_role("heading", name="Run #1")).to_be_visible(timeout=LOAD)
        # Run detail is tabbed, and Rows lands first. A person's results live inside the row that
        # built them, so open every row before reading them.
        open_rows(page)
        # Every user is a clickable tab in the run's nav; the selected one's rows show below it.
        for username in ("sarah", "mike", "jess"):
            expect(page.get_by_role("tab", name=re.compile(username, re.IGNORECASE))).to_be_visible()

    def test_run_detail_shows_what_changed(self, page: Page, app: ShortlistApp, reset_fake_plex):
        """The diff is the audit trail (rule 10): everything Added on the first run, and on the
        next run the staleness guard rotates fresh titles in — without ever shrinking a row."""
        state = reset_fake_plex
        first = build_real_rows(app)
        # The engine must record a per-(row, library) breakdown — the whole run page is built from it.
        run = app.api("GET", f"/api/runs/{first['id']}").json()
        assert any(b["added"] for u in run["users"] for b in u["breakdown"]), "no per-library breakdown recorded"

        page.goto(f"/runs/{first['id']}")
        expect(page.get_by_role("heading", name=f"Run #{first['id']}")).to_be_visible(timeout=LOAD)
        open_rows(page)
        # First run: every pick is new, so the selected user's row shows a "+N new" badge and nothing
        # removed. Picks render as a ranked list, and their titles are answerable from here.
        expect(page.get_by_text(re.compile(r"\+\d+ new")).first).to_be_visible()
        expect(page.get_by_text(re.compile(r"\d+ removed"))).to_have_count(0)
        # Asserted against a title the API says was actually delivered, rather than a pattern the
        # demo library's names happen to match — those are real film and show titles now, chosen so
        # the docs screenshots look like a real library, and they will change again.
        delivered = next(p["title"] for u in run["users"] for b in u["breakdown"] for p in b["picks"])
        expect(page.locator("body")).to_contain_text(delivered)

        def row_sizes() -> dict[str, int]:
            sizes: dict[str, int] = {}
            for collection in state.collections.values():
                for label in collection.labels:
                    if label.lower().startswith("shortlist_"):
                        sizes[label.lower()] = sizes.get(label.lower(), 0) + len(collection.item_keys)
            return sizes

        before = row_sizes()
        second = build_real_rows(app)
        page.goto(f"/runs/{second['id']}")
        expect(page.get_by_role("heading", name=f"Run #{second['id']}")).to_be_visible(timeout=LOAD)
        open_rows(page)

        # "Don't repeat the last 3 runs' picks" rotates the row; it must never leave it short.
        # When fresh candidates run out, held-back titles backfill — a row shrinking (or being
        # pruned) because the pool was thin for one night would be a regression, not variety.
        assert row_sizes() == before, "a re-run must keep every row exactly as full as it was"

        for user in app.api("GET", f"/api/runs/{second['id']}").json()["users"]:
            diff = user["diff"]
            assert len(diff.get("added", [])) == len(diff.get("removed", [])), (
                f"{user['slug']}: a rotated-out title must be replaced, not simply dropped"
            )
        # The second run's page renders its users (the per-user nav), so its results are reachable.
        expect(page.get_by_role("tab").first).to_be_visible(timeout=LOAD)

    def test_a_users_picks_explain_themselves(self, page: Page, app: ShortlistApp):
        """Every pick carries its "Because you watched …" reason — the product's whole promise."""
        build_real_rows(app)
        sarah_id = _users_by_name(app)["sarah"]["id"]
        run_id = app.api("GET", "/api/runs").json()[0]["id"]
        sarah_picks = next(
            u["picks"] for u in app.api("GET", f"/api/runs/{run_id}").json()["users"] if u["slug"] == "sarah"
        )

        page.goto(f"/users/{sarah_id}")
        expect(page.get_by_role("heading", name="sarah")).to_be_visible(timeout=LOAD)

        # Every poster carries its own reason under the title, so the first such caption is what
        # shows the page has loaded.
        reason_re = re.compile(r"Because you watched .+|Worth another watch|They asked for these")
        expect(page.get_by_text(re.compile(r"^Because you watched .+")).first).to_be_visible(timeout=LOAD)
        # Rows collapse to the first 5 picks (collapseAfter=5), so expand every row before counting:
        # this test asserts EVERY pick carries a reason, not just the first few.
        # Re-query each iteration: clicking one toggle re-renders the DOM and stales the locator list.
        while (toggle := page.get_by_role("button", name=re.compile(r"Show all \d+")).first).is_visible():
            toggle.click()
        # Every pick is a list item, and every one of them names why it is there. sarah watches movies
        # AND TV, so both libraries' rows show a reason.
        picks = page.get_by_role("listitem").filter(has_text=re.compile(r"suggested by"))
        expect(picks).to_have_count(len(sarah_picks))
        expect(picks.filter(has_text=reason_re)).to_have_count(len(sarah_picks))
        titles = {p["title"] for p in sarah_picks}
        assert titles & {t for t, _ in DEMO_MOVIES} and titles & {t for t, _ in DEMO_SHOWS}, (
            "sarah's row should mix both libraries — otherwise this test proves nothing about them"
        )

        # Ranked, and the rank the engine chose is the rank shown.
        expect(page.get_by_text(sarah_picks[0]["title"], exact=True).first).to_be_visible()

    def test_the_jess_gets_the_cold_start_row_and_says_so(self, page: Page, app: ShortlistApp):
        """No history -> the popular-titles fallback, labelled honestly rather than faked."""
        build_real_rows(app)
        jess_id = _users_by_name(app)["jess"]["id"]

        page.goto(f"/users/{jess_id}")
        expect(page.get_by_role("heading", name="jess")).to_be_visible(timeout=LOAD)
        expect(page.get_by_text("New viewer").first).to_be_visible()
        expect(page.get_by_text("Popular on this server").first).to_be_visible()