"""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