"""Threads on the Interaction Standard (R2 task T1): the subject resolver, the backfill, the
get-or-create under a race, the one read per surface, and the routes.

What the pieces cost when they were wrong elsewhere: a select-then-insert on a shared row lost
whole collection runs to the barcode constraint (agents.md), so `thread_for` inserts inside a
savepoint and re-reads on IntegrityError; a key renamed under a comment orphans it, so the
legacy keys map both ways and the backfill is idempotent (a second run reports zero). The
route tests run in-process on the accounts suite's SQLite: the new comment POST makes the
thread and keeps the legacy key beside it, a delete is soft and leaves every read, a resolve
belongs to the opener or a curator, and the running list's threads answer 404 to a client.
"""

from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from app.models import Account, AccountLevel, DiscussionComment, DiscussionItem, Thread
from app.services import accounts
from app.services import discussion as disc
from app.services.identity import Actor
from tests import _accounts as T
from tests.kit import _env

T0 = datetime(2026, 9, 11, 1, 0, tzinfo=UTC)


class TestSubjects:
    @pytest.mark.parametrize("item_id, key, expected", [
        (3, None, ("decision", "3")),
        (None, "general", ("page", "discuss")),
        (None, "my-airports", ("feature", "my-airports")),
        (None, "structure:urls", ("structure", "urls")),
        (None, "quote:later-price-alerts", ("quote", "later-price-alerts")),
        (None, "todo:9", ("todo", "9")),
        (None, "item:decide-page-types-18-sep", ("item", "decide-page-types-18-sep")),
    ])
    def test_legacy_keys_map_to_subjects_and_back(self, item_id, key, expected):
        assert disc.subject_of_legacy(item_id, key) == expected
        assert disc.legacy_of(*expected) == (item_id, key)

    def test_an_unknown_type_before_the_colon_is_a_board_key(self):
        # A board key may carry a colon only if its head is not a subject type.
        assert disc.subject_of_legacy(None, "beta:thing") == ("feature", "beta:thing")

    def test_permission_and_anchor_and_link(self, monkeypatch):
        assert disc.permission_of("item") == accounts.PERM_PLAN_VIEW
        assert disc.permission_of("structure") == accounts.PERM_CLIENT_VIEW
        assert disc.anchor("quote", "later-Price Alerts") == "t-quote-later-price-alerts"
        monkeypatch.setattr(disc.settings, "public_base_url", "https://dfp.test/")
        assert disc.deep_link("structure", "urls", 88) == "https://dfp.test/discuss?tab=structure#t-structure-urls#c-88"
        assert disc.deep_link("item", "decide-x") == "https://dfp.test/issues#t-item-decide-x"
        assert disc.deep_link("feature", "my-airports", 5) == "https://dfp.test/discuss#t-feature-my-airports#c-5"
        assert not disc.valid_subject("stage", "1") and not disc.valid_subject("todo", "a:b")

    def test_mentions_are_case_folded_deduped_and_not_emails(self):
        assert disc.mentions_in("ask @Adam and @adam, then @mark. mail rian@example.com") == ["adam", "mark"]


@pytest.fixture
def db(monkeypatch):
    T.fresh(monkeypatch)
    T.person("rian", display_name="Rian")
    with _env.TestSessionLocal() as s:
        s.add(AccountLevel(name="admin", permissions=list(accounts.SEED_LEVELS["admin"]["permissions"]), assignable=[]))
        s.commit()
    T.person("adam", level="admin", display_name="Adam")
    T.person("mark", level="admin", display_name="Mark")
    with _env.TestSessionLocal() as s:
        yield s


def ids() -> dict[str, int]:
    with _env.TestSessionLocal() as s:
        return {a.username: a.id for a in s.scalars(select(Account))}


class TestBackfill:
    def test_threads_backfill_links_every_legacy_key_once(self, db):
        db.add(DiscussionItem(id=3, title="Awards rules"))
        db.add_all([
            DiscussionComment(item_id=3, author="Adam", body="on the card", created_at=T0, updated_at=T0),
            DiscussionComment(feature_key="general", author="Adam", body="anything", created_at=T0, updated_at=T0),
            DiscussionComment(feature_key="structure:urls", author="Mark", body="one", created_at=T0, updated_at=T0),
            DiscussionComment(feature_key="structure:urls", author="Mark", body="two, edited", created_at=T0,
                              updated_at=T0 + timedelta(minutes=5)),
            DiscussionComment(feature_key="my-airports", author="Adam", body="board", created_at=T0, updated_at=T0),
        ])
        db.commit()
        assert disc.backfill_threads(db) == "threads: 4 thread(s) created, 5 comment(s) linked"
        threads = {(t.subject_type, t.subject_id) for t in db.scalars(select(Thread))}
        assert threads == {("decision", "3"), ("page", "discuss"), ("structure", "urls"), ("feature", "my-airports")}
        db.expire_all()
        moved = db.scalar(select(DiscussionComment).where(DiscussionComment.body == "two, edited"))
        # A moved updated_at is not an edit signal (the who-column backfill moved it too): nothing
        # is marked, and the link did not move updated_at again.
        assert moved.edited_at is None and moved.updated_at.replace(tzinfo=UTC) == T0 + timedelta(minutes=5)
        assert all(c.thread_id for c in db.scalars(select(DiscussionComment)))
        assert disc.backfill_threads(db) == "threads: 0 thread(s) created, 0 comment(s) linked"


class TestThreadFor:
    def test_get_or_create_survives_a_race(self, db, monkeypatch):
        first = disc.thread_for(db, "structure", "urls", label="Structure › URL patterns")
        assert disc.thread_for(db, "structure", "urls") is first and first.label == "Structure › URL patterns"
        # Two writers: the second's insert hits the unique constraint and re-reads the winner.
        calls = {"n": 0}
        real_thread_of = disc.thread_of

        def racing_thread_of(session, subject_type, subject_id):
            calls["n"] += 1
            if calls["n"] == 1:
                return None  # the row is not there yet when we look ...
            return real_thread_of(session, subject_type, subject_id)

        monkeypatch.setattr(disc, "thread_of", racing_thread_of)
        again = disc.thread_for(db, "structure", "urls")  # ... but it is by the time we insert
        assert again.id == first.id
        assert db.scalar(select(Thread).where(Thread.subject_type == "structure")) is not None
        assert len(list(db.scalars(select(Thread)))) == 1

    def test_discussion_of_lists_starters_and_counts(self, db):
        actor = Actor(ids()["adam"], "adam", "Adam")
        # One commit per write, as a request does: the permission lookups in `people_for` open
        # the kit's own session, and the suite's single connection rolls back pending rows.
        c = disc.add_comment(db, subject_type="structure", subject_id="urls", actor=actor, body="hello", label="URLs")
        db.commit()
        gone = disc.add_comment(db, subject_type="structure", subject_id="map", actor=actor, body="bye")
        db.commit()
        disc.delete_comment(db, gone)
        db.commit()
        out = disc.discussion_of(db, [
            {"subject_type": "structure", "subject_id": "urls"},
            {"subject_type": "structure", "subject_id": "map"},
            {"subject_type": "structure", "subject_id": "schema"},
        ])
        assert out["open"] == 1 and out["total"] == 1
        assert out["subjects"][0]["thread"]["comments"][0]["id"] == c.id
        assert out["subjects"][0]["thread"]["label"] == "URLs"
        assert out["subjects"][1]["thread"] is None  # every comment deleted: a starter again
        assert out["subjects"][2]["thread"] is None
        with pytest.raises(ValueError):
            disc.add_comment(db, subject_type="structure", subject_id="urls", actor=actor, body="  ")


class TestRoutes:
    def test_a_new_shape_post_makes_the_thread_and_keeps_the_legacy_key(self, db):
        c = T.client()
        T.as_user(c, "adam")
        r = c.post("/api/discussion/comments", json={"subject_type": "todo", "subject_id": "9", "body": "done",
                                                     "label": "To-do › Airport write-ups"})
        assert r.status_code == 201, r.text
        row = db.get(DiscussionComment, r.json()["id"])
        assert row.feature_key == "todo:9" and row.thread_id is not None and row.author == "Adam"
        thread = db.get(Thread, row.thread_id)
        assert (thread.subject_type, thread.subject_id, thread.label) == ("todo", "9", "To-do › Airport write-ups")
        # The legacy read still lists it; the new read carries the thread.
        assert [x["feature_key"] for x in c.get("/api/discussion/feature-comments").json()] == ["todo:9"]
        threads = c.get("/api/discussion/threads").json()
        assert len(threads) == 1 and threads[0]["anchor"] == "t-todo-9" and threads[0]["comments"][0]["body"] == "done"
        assert c.post("/api/discussion/comments", json={"subject_type": "stage", "subject_id": "1", "body": "x"}).status_code == 422

    def test_delete_is_soft_and_leaves_every_read(self, db):
        c = T.client()
        T.as_user(c, "mark")
        cid = c.post("/api/discussion/comments", json={"subject_type": "structure", "subject_id": "urls", "body": "marks"}).json()["id"]
        T.as_user(c, "adam")
        r = c.post(f"/api/discussion/comments/{cid}/delete")
        assert r.status_code == 403 and r.json()["detail"]["error_code"] == "FORBIDDEN"
        T.as_user(c, "mark")
        assert c.post(f"/api/discussion/comments/{cid}/delete").status_code == 200
        assert db.get(DiscussionComment, cid).deleted_at is not None  # the row stays
        assert c.get("/api/discussion/feature-comments").json() == []
        assert c.get("/api/discussion/threads").json() == []
        assert c.post(f"/api/discussion/feature-comments/{cid}", json={"body": "late"}).status_code == 404
        # The owner (discussion.curate by absence) removes anyone's.
        cid2 = c.post("/api/discussion/comments", json={"subject_type": "structure", "subject_id": "urls", "body": "again"}).json()["id"]
        T.as_user(c, "rian")
        assert c.post(f"/api/discussion/comments/{cid2}/delete").status_code == 200

    def test_resolve_belongs_to_the_opener_or_a_curator(self, db):
        c = T.client()
        T.as_user(c, "adam")
        c.post("/api/discussion/comments", json={"subject_type": "quote", "subject_id": "guides", "body": "opened by adam"})
        tid = c.get("/api/discussion/threads").json()[0]["id"]
        T.as_user(c, "mark")
        assert c.post(f"/api/discussion/threads/{tid}/resolve", json={"resolved": True}).status_code == 403
        T.as_user(c, "adam")
        r = c.post(f"/api/discussion/threads/{tid}/resolve", json={"resolved": True})
        assert r.status_code == 200 and r.json()["resolved"] is True and r.json()["resolved_by_id"] == ids()["adam"]
        T.as_user(c, "rian")
        assert c.post(f"/api/discussion/threads/{tid}/resolve", json={"resolved": False}).json()["resolved"] is False

    def test_the_running_lists_threads_are_the_owners(self, db):
        c = T.client()
        T.as_user(c, "rian")
        r = c.post("/api/discussion/comments", json={"subject_type": "item", "subject_id": "decide-x", "body": "the plan"})
        assert r.status_code == 201
        cid, tid = r.json()["id"], r.json()["thread_id"]
        assert r.json()["mine"] is True
        assert [t["subject_type"] for t in c.get("/api/discussion/threads").json()] == ["item"]
        T.as_user(c, "adam")
        assert c.get("/api/discussion/threads").json() == []
        assert c.post("/api/discussion/comments", json={"subject_type": "item", "subject_id": "decide-x", "body": "peek"}).status_code == 404
        assert c.post(f"/api/discussion/comments/{cid}/delete").status_code == 404
        assert c.post(f"/api/discussion/threads/{tid}/resolve", json={"resolved": True}).status_code == 404

    def test_the_new_edit_route_reaches_a_cards_comment_and_the_read_says_mine(self, db):
        c = T.client()
        T.as_user(c, "adam")
        db.add(DiscussionItem(id=3, title="Awards rules"))
        db.commit()
        cid = c.post("/api/discussion/3/comments", json={"body": "on the card"}).json()["id"]
        assert c.post(f"/api/discussion/feature-comments/{cid}", json={"body": "x"}).status_code == 404  # the legacy route's shape
        r = c.post(f"/api/discussion/comments/{cid}/edit", json={"body": "on the card, reworded"})
        assert r.status_code == 200 and r.json()["edited_at"] is not None
        thread = c.get("/api/discussion/threads").json()[0]
        assert (thread["subject_type"], thread["subject_id"], thread["label"]) == ("decision", "3", "Awards rules")
        assert thread["comments"][0]["mine"] is True and thread["comments"][0]["body"] == "on the card, reworded"
        T.as_user(c, "mark")
        assert c.get("/api/discussion/threads").json()[0]["comments"][0]["mine"] is False
        assert c.post(f"/api/discussion/comments/{cid}/edit", json={"body": "not mine"}).status_code == 403

    def test_an_edit_stamps_edited_at(self, db):
        c = T.client()
        T.as_user(c, "adam")
        cid = c.post("/api/discussion/feature-comments", json={"feature_key": "structure:urls", "body": "first"}).json()["id"]
        r = c.post(f"/api/discussion/feature-comments/{cid}", json={"body": "second"})
        assert r.status_code == 200 and r.json()["edited_at"] is not None and r.json()["body"] == "second"
        assert db.get(DiscussionComment, cid).edited_at is not None


LATER_KEYS_FROZEN = [
    "automated-collection", "traveller-accounts-and-logins", "price-alerts", "the-full-newsletter-system",
    "administration-area", "price-history-tracker", "interactive-map", "store-partner-portal",
    "consumer-price-submissions", "paid-producer-listings", "full-category-expansion",
]


def test_the_quote_pages_later_keys_are_frozen():
    """`quote:later-<key>` was derived from the line's name at render time; comments attach by
    exact key, so the keys are explicit in quote.ts now and pinned here: rewording a name must
    never move its conversation. Add a key for a new line; never rename one."""
    import pathlib as _p
    import re as _re

    src = (_p.Path(__file__).resolve().parents[1] / "web" / "src" / "lib" / "quote.ts").read_text()
    later = src[src.index("export const LATER"):]
    later = later[:later.index("];")]
    assert _re.findall(r'key: "([^"]+)"', later)[:len(LATER_KEYS_FROZEN)] == LATER_KEYS_FROZEN


def test_integrity_error_path_is_the_kits_savepoint(db):
    """A direct check that the savepoint really isolates the failed insert: after a
    constraint violation inside `thread_for`, the session is still usable and commits."""
    disc.thread_for(db, "todo", "9")
    db.commit()
    with pytest.raises(IntegrityError):
        with db.begin_nested():
            db.add(Thread(subject_type="todo", subject_id="9"))
            db.flush()
    db.add(Thread(subject_type="todo", subject_id="10"))
    db.commit()
    assert {t.subject_id for t in db.scalars(select(Thread))} == {"9", "10"}