"""The writer's rules the other suites do not pin (Stream K2; spec tests 1, 7, 10, 18, 22).
What it cost: a bulk sheet approval could have written over a person's individual decision;
a human merge lost the loser's decisions; an emptied line named by a decision was pruned; an
approval ran under a live collection. SQLite kit, no network."""
from __future__ import annotations

from datetime import UTC, datetime

import pytest
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app import cli
from app.models import (Account, Award, Base, Brand, CollectionRun, Decision, LEDGER_TABLES, Listing, Merge, ProductLine, ProductVariant,
                        Proposal, Redirect, Retailer, Shop, Source, Suggestion, AttributeAlias)
from app.models.places import Place, ShopPlace
from app.services import keying, proposals_store, publish
from app.services.decisions import effective, replay as replay_service, undo as undo_service, writer

TABLES = [Account.__table__, Source.__table__, CollectionRun.__table__, Brand.__table__, ProductLine.__table__, ProductVariant.__table__,
          AttributeAlias.__table__, Award.__table__, Retailer.__table__, Shop.__table__, Listing.__table__, Merge.__table__,
          Suggestion.__table__, Place.__table__, ShopPlace.__table__, *LEDGER_TABLES]
RIAN = 1


@pytest.fixture
def db():
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    Base.metadata.create_all(engine, tables=TABLES)
    keying.invalidate()
    with sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)() as s:
        s.add(Account(id=RIAN, username="rian", display_name="rian"))
        s.add(Account(id=2, username="adam", display_name="Adam"))
        brand = Brand(id=1, slug="jw", name="Johnnie Walker"); retailer = Retailer(slug="r", name="R"); s.add_all([brand, retailer]); s.flush()
        s.add(ProductLine(id=1, brand_id=1, key="blue label", slug="jw-blue-label", name="Blue Label"))
        s.add(ProductLine(id=2, brand_id=1, key="old rule key", slug="jw-old", name="Old"))
        s.flush()
        for n in (1, 2):
            v = ProductVariant(id=n, name="Blue Label 1L", brand="Johnnie Walker", brand_id=1, vertical="liquor", match_key=f"k{n}", product_line_id=1,
                               quantity_value=1000, quantity_unit="ml", quantity_state="stated", quantity_ml=1000, form="single", attributes={}, abv=40 if n == 1 else None)
            shop = Shop(id=n, retailer_id=retailer.id, code=f"S{n}", iata=f"A0{n}", name=f"Shop {n}", currency="EUR")
            s.add_all([v, shop]); s.flush()
            s.add(Listing(variant_id=n, shop_id=n, source_sku=f"sku{n}"))
        s.commit()
        yield s
    keying.invalidate()


def test_human_first_a_bulk_write_never_overrules_a_person_and_the_check_is_recomputed_inside_the_batch(db):
    with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b:
        writer.record(b, "product_variant", 1, "name", "Rian's name")
    with writer.batch(db, "sheet", "bulk", 2, commit=True, tail=False) as b:
        with pytest.raises(writer.Refused) as refused:
            writer.record(b, "product_variant", 1, "name", "A sheet's name")
        assert refused.value.code == "DECISION_HELD"
        # The same value is not a conflict; an individual write supersedes anything.
        writer.record(b, "product_variant", 1, "name", "Rian's name")
    with writer.batch(db, "route", "individual", 2, commit=True, tail=False) as b:
        writer.record(b, "product_variant", 1, "name", "Adam's name")
    assert db.get(ProductVariant, 1).name == "Adam's name"


def test_a_human_merge_carries_the_losers_decisions_and_keeps_the_survivors(db):
    with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b:
        writer.record(b, "product_variant", 2, "attribute:abv", 43, reason="the shop says so")
        writer.record(b, "product_variant", 2, "name", "Blue Label Litre")
        writer.record(b, "product_variant", 1, "name", "Blue Label One Litre")
    with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b:
        row = writer.record(b, "product_variant", 2, "merged_into", 1, reason="one bottle")
    assert db.get(ProductVariant, 2).merged_into_id == 1
    ours = effective(db, "product_variant", [1])
    assert ours[(1, "attribute:abv")].value == 43 and ours[(1, "attribute:abv")].caused_by_id == row.id
    assert ours[(1, "attribute:abv")].detail["carried_from"] == row.natural_key and ours[(1, "attribute:abv")].rule_value == 40
    assert ours[(1, "name")].value == "Blue Label One Litre", "a field the survivor decided is kept"
    carry = db.scalar(select(Merge)).detail["carry"]
    assert sorted(carry["carried"]) == ["attribute:abv"] and "name" in carry["kept"]
    assert effective(db, "product_variant", [2])[(2, "name")].value == "Blue Label Litre", "the loser's rows stay as its history"
    assert db.scalar(select(Merge)).decision_id == row.id


def test_prune_keeps_a_line_any_decision_or_redirect_names(db):
    for v in db.scalars(select(ProductVariant)):
        v.product_line_id = 1
    db.commit()
    assert cli.backfill_prune_lines(db) == "prune_lines: 1 empty line row(s) nothing referenced deleted"
    db.add(ProductLine(id=3, brand_id=1, key="gone", slug="jw-gone", name="Gone")); db.add(ProductLine(id=4, brand_id=1, key="redirected", slug="jw-redirected", name="Redirected")); db.commit()
    with writer.batch(db, "route", "individual", RIAN, commit=True, tail=False) as b:
        writer.record(b, "product_variant", 1, "product_line", 3)
        writer.record(b, "product_variant", 1, "product_line", 1)  # supersedes: line 3 is now only in a rule_value / old value
    publish.redirect_write(db, "product_line", "jw-old-address", "jw-redirected")
    db.commit()
    assert cli.backfill_prune_lines(db) == "prune_lines: 0 empty line row(s) nothing referenced deleted"
    assert publish.redirect_for("jw-old-address", "product_line", db) == "jw-redirected"


def test_the_gates_refuse_while_a_collection_runs(db, tmp_path):
    db.add(Source(id=1, slug="s", name="S")); db.add(CollectionRun(source_id=1, started_at=datetime.now(UTC), status="running")); db.commit()
    with pytest.raises(writer.Refused) as refused:
        undo_service.undo_batch(db, "00000000-0000-0000-0000-000000000000", RIAN, "x")
    assert refused.value.code == "COLLECTION_RUNNING"
    f = tmp_path / "x.jsonl"; f.write_text("")
    with pytest.raises(writer.Refused) as refused:
        replay_service.replay(db, f)
    assert refused.value.code == "COLLECTION_RUNNING"
    assert replay_service.replay(db, f, force=True)["applied"] == 0


def test_the_proposals_store_upserts_on_the_idempotency_key_and_resolves_on_this_host(db):
    rows = [proposals_store.ProposalRow("product_variant", f"variant:{db.get(ProductVariant, 1).uid}", None, "attribute:abv", 43,
                                        "the shop says 43", [{"listing": "listing:r/S1/sku1", "source": "listed_name", "span": [0, 2], "text": "43"}],
                                        0.95, "jw", sheet_line_ref="line:new:jw-blue-label", position=1),
            proposals_store.ProposalRow("product_line", "line:new:jw-blue-label", {"brand_slug": "jw", "slug": "jw-blue-label", "name": "Blue Label"},
                                        "name", "Blue Label", "the head words", [], 0.9, "jw", sheet_line_ref="line:new:jw-blue-label", position=0),
            proposals_store.ProposalRow("product_variant", "variant:00000000-0000-0000-0000-000000000000", {"listings": ["listing:r/S9/nope"]},
                                        "name", "Nope", None, [], 0.5, "jw")]
    p = proposals_store.write(db, "claude/2026-09-18/jw-1", rows, kind="session", generator="claude-session", rules_version="5", process_version="1", scope_brand_slug="jw")
    assert p.counts["inserted"] == 3 and p.counts["parked"] == 1
    by_key = {r.natural_key: r for r in db.scalars(select(Proposal))}
    assert by_key[f"variant:{db.get(ProductVariant, 1).uid}"].resolution == "resolved" and by_key["line:new:jw-blue-label"].resolution == "resolved"
    assert by_key["variant:00000000-0000-0000-0000-000000000000"].status == "parked"
    rows[0].value = 41
    p2 = proposals_store.write(db, "claude/2026-09-18/jw-1", rows[:1], kind="session", generator="claude-session", rules_version="5", process_version="1")
    assert p2.id == p.id and p2.counts["updated"] == 1 and by_key[rows[0].natural_key].value == 41


def test_a_collection_that_died_does_not_block_the_review_for_ever(db):
    """A run still `running` long after it started belongs to a dead process, and must not hold
    approvals.

    Cost: found in the review simulation. Nothing clears a crashed run but the next run of that
    same source or `backfill stuck_runs` by hand, and the approval gate counted any `running` row
    at all. A source that died on a Saturday sweep would have refused every decision on Sunday
    with "wait", pointing at a collection that was never going to finish.
    """
    from datetime import UTC, datetime, timedelta

    from app.models import CollectionRun, Source
    from app.services.decisions.writer import Refused, refuse_if_collecting
    from app.services.ingest import STUCK_AFTER

    db.add(Source(id=1, slug="s", name="S"))
    db.flush()
    now = datetime.now(UTC)
    live = CollectionRun(source_id=1, started_at=now - timedelta(minutes=5), status="running")
    db.add(live)
    db.flush()
    with pytest.raises(Refused) as refused:
        refuse_if_collecting(db)
    assert refused.value.code == "COLLECTION_RUNNING", "a collection really running still holds the review"

    live.started_at = now - STUCK_AFTER - timedelta(minutes=1)
    db.flush()
    refuse_if_collecting(db)  # the same row, now older than a run can credibly be: no refusal


def test_a_run_that_predates_the_container_is_dead_however_young_it_is(db, monkeypatch):
    """A reboot or a deploy kills every collector in the container, so a run that began before the
    container did is dead -- even three hours into a twenty-four hour timer.

    Cost: mosiah ran out of memory on 18 Sep (someone else's editor server, five global OOM kills,
    no container ever over its cap) and was rebooted mid-sweep. Two runs, avolta-ath and
    extime-paris, stayed `running`. They were three hours old, so `backfill stuck_runs` declined
    them and the approval gate counted them as live collections: every approval on the review
    would have been refused with "wait" until 19:08 the following evening, the weekend before the
    soft launch, pointing at two processes the reboot had already taken.
    """
    from datetime import UTC, datetime, timedelta

    from app.models import CollectionRun, Source
    from app.services import ingest
    from app.services.decisions.writer import Refused, refuse_if_collecting

    now = datetime.now(UTC)
    booted = now - timedelta(hours=1)          # the container came up an hour ago
    monkeypatch.setattr(ingest, "process_boot", lambda *a, **k: booted)

    db.add(Source(id=1, slug="s", name="S"))
    db.flush()
    orphan = CollectionRun(source_id=1, started_at=now - timedelta(hours=3), status="running")
    db.add(orphan)
    db.flush()

    assert ingest.is_stuck(orphan, now, boot=booted), "it began before the container it ran in"
    refuse_if_collecting(db)  # and so it does not hold the review

    orphan.started_at = now - timedelta(minutes=5)   # started after the boot: really collecting
    db.flush()
    assert not ingest.is_stuck(orphan, now, boot=booted)
    with pytest.raises(Refused) as refused:
        refuse_if_collecting(db)
    assert refused.value.code == "COLLECTION_RUNNING"


def test_process_boot_reads_the_container_start_from_proc(tmp_path):
    """`process_boot` parses /proc the way the kernel writes it, comm-with-spaces and all."""
    import os

    from datetime import UTC, datetime

    from app.services import ingest

    hz = os.sysconf("SC_CLK_TCK")
    (tmp_path / "1").mkdir()
    # field 2 is the comm, in parentheses, and may itself contain spaces and brackets.
    (tmp_path / "1" / "stat").write_text(
        "1 (my app :)) S " + " ".join(str(i) for i in range(4, 22)) + f" {90 * hz} " + "0 " * 30)
    (tmp_path / "stat").write_text("cpu  1 2 3\nbtime 1758000000\nprocesses 99\n")

    ingest.process_boot.cache_clear()
    assert ingest.process_boot(str(tmp_path)) == datetime.fromtimestamp(1758000000 + 90, UTC)

    # Where /proc is not Linux's, the rule simply does not fire and the timer is all there is.
    assert ingest.process_boot(str(tmp_path / "nothing-here")) is None


def test_the_live_page_calls_an_orphaned_run_dead_like_the_gate_does(db, monkeypatch):
    """What `/collectors#live` shows and what the approval gate believes must be one answer.

    Cost: the live tab computed `dead` from the 24-hour timer on its own, so the two runs the
    18 Sep reboot orphaned would have been drawn as collections in progress -- with elapsed time
    ticking up and a prices-per-minute rate -- for the 21 hours left on that timer, while the
    review refused every approval because of them. Since Stream AW4 the word is
    `LiveCollector.state == "dead"`, from the same `is_stuck` the gate reads.
    """
    from datetime import UTC, datetime, timedelta

    from app.models import CollectionRun, Source
    from app.services import collector_view, ingest
    from app.services.collectors.registry import COLLECTORS

    now = datetime.now(UTC)
    monkeypatch.setattr(ingest, "process_boot", lambda *a, **k: now - timedelta(hours=1))
    slug = sorted(COLLECTORS)[0]
    db.add(Source(id=1, slug=slug, name="S"))
    db.add(CollectionRun(source_id=1, started_at=now - timedelta(hours=3), status="running",
                         products_seen=200, prices_written=200))
    db.flush()

    out = collector_view.live(db, now=now)
    row = next(c for c in out.collectors if c.slug == slug)
    assert row.state == "dead" and row.state_detail == "dead: began before this container"
    assert row.run is not None and row.run.id is not None
    assert out.collectors[0].slug == slug, "a live state sorts before every idle collector"
    assert out.totals.dead == 1


def test_a_run_whose_heartbeat_went_silent_for_an_hour_is_dead_however_young_it_is(db):
    """The hook writes `heartbeat_at` at every request boundary and every poll of a pause, so a
    running row with no beat for `STUCK_HEARTBEAT` belongs to a process that is gone: OOM-killed,
    or shed by the guard before the hook existed.

    Cost: the 19 Sep sweep's guard shed four collectors with SIGTERM and their four `running` rows
    sat for the 24 hours the timer had left, holding the review's approval gate; a page that read
    them as live collections would have offered Pause and Stop on processes that did not exist.
    A NULL heartbeat falls through to the timer and the boot rule: a run from before the hook is
    judged as before.
    """
    from datetime import UTC, datetime, timedelta

    from app.models import CollectionRun, Source
    from app.services import ingest

    now = datetime.now(UTC)
    db.add(Source(id=1, slug="s", name="S"))
    run = CollectionRun(source_id=1, started_at=now - timedelta(hours=2), status="running",
                        heartbeat_at=now - ingest.STUCK_HEARTBEAT - timedelta(minutes=1))
    db.add(run)
    db.flush()
    assert ingest.is_stuck(run, now, boot=None)
    assert not ingest.is_stuck(run, now, boot=None, heartbeat=False), "the gate's own rule ignores the beat (an issue is filed)"
    run.heartbeat_at = now - timedelta(seconds=30)
    assert not ingest.is_stuck(run, now, boot=None)
    run.heartbeat_at = None
    assert not ingest.is_stuck(run, now, boot=None), "no beat at all is judged by time alone"
    run.started_at = now - ingest.STUCK_AFTER - timedelta(minutes=1)
    assert ingest.is_stuck(run, now, boot=None)