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