"""Export and replay by natural key (Stream K2; spec §8, test 20). Two SQLite copies with
shifted ids: every row lands, every column agrees, a second replay applies 0, an older row over
a newer target row is inserted and not materialised, a bulk row over a person's parks, the
export carries no local id, a batch made on A is undone on B by its uid. What it cost before:
`propose --file` carried staging's row ids and could not load on production (plan W13)."""
from __future__ import annotations

import io
import json
import pathlib
import re
from datetime import UTC, datetime, timedelta

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

from app.models import (Account, Award, Base, Brand, CollectionRun, Decision, DecisionBatch, LEDGER_TABLES, Listing, Merge, ProductLine,
                        ProductVariant, Retailer, Shop, Source, Suggestion, AttributeAlias)
from app.models.places import Place, ShopPlace
from app.services import keying
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


def make(offset: int, *, extra_line: bool = False, dash_two: bool = False):
    """A copy: the same rows and uids, ids shifted by `offset`; B holds one extra derived line
    under the alias brand and one line with a -2 slug."""
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    Base.metadata.create_all(engine, tables=TABLES)
    s = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)()
    s.add(Account(id=RIAN, username="rian", display_name="rian"))
    retailer = Retailer(id=1 + offset, slug="r", name="R"); s.add(retailer); s.flush()
    return engine, s


UIDS: dict = {}


def seed(s, offset, *, extra_line=False, dash_two=False):
    import uuid

    def uid(name):
        return UIDS.setdefault(name, uuid.uuid4())
    rabanne = Brand(id=1 + offset, uid=uid("b1"), slug="rabanne", name="Rabanne")
    paco = Brand(id=2 + offset, uid=uid("b2"), slug="paco-rabanne", name="Paco Rabanne")
    s.add_all([rabanne, paco]); s.flush()
    l1 = ProductLine(id=1 + offset, uid=uid("l1"), brand_id=rabanne.id, key="1 million", slug="rabanne-1-million", name="1 Million")
    l2 = ProductLine(id=2 + offset, uid=uid("l2"), brand_id=paco.id, key="1 million", slug="paco-rabanne-1-million", name="1 Million")
    s.add_all([l1, l2])
    if extra_line:
        s.add(ProductLine(id=9 + offset, brand_id=paco.id, key="lady million", slug="paco-rabanne-lady-million", name="Lady Million"))
    if dash_two:
        s.add(ProductLine(id=8 + offset, brand_id=rabanne.id, key="invictus", slug="rabanne-invictus-2", name="Invictus"))
    s.flush()
    for n, (brand, line, name) in enumerate(((rabanne, l1, "1 Million Elixir 100ml"), (paco, l2, "1 Million 100ml"), (rabanne, l1, "1 Million 100ml")), start=1):
        v = ProductVariant(id=n + offset, uid=uid(f"v{n}"), name=name, brand=brand.name, brand_id=brand.id, vertical="beauty", match_key=f"k{n}",
                           product_line_id=line.id, quantity_value=100, quantity_unit="ml", quantity_state="stated", quantity_ml=100, form="single", attributes={})
        shop = Shop(id=n + offset, retailer_id=1 + offset, code=f"S{n}", iata=f"A0{n}", name=f"Shop {n}", currency="EUR")
        s.add_all([v, shop]); s.flush()
        s.add(Listing(id=n + offset, variant_id=v.id, shop_id=shop.id, source_sku=f"sku{n}"))
    s.commit()
    keying.invalidate()
    from app.services import merges
    merges.rekey_product_variants(s, list(s.scalars(select(ProductVariant))), keying.load_maps(s))
    s.commit()


@pytest.fixture
def copies():
    UIDS.clear()
    ea, a = make(0); seed(a, 0)
    eb, b = make(100, extra_line=True, dash_two=True); seed(b, 100, extra_line=True, dash_two=True)
    yield a, b
    a.close(); b.close()


def decide_on_a(a):
    """About a dozen decisions across every entity type: names, a line membership, an attribute, a
    quantity, a hidden, a pin, an ignore, a wording, a Keep separate, a Confirm same that merges,
    a brand alias with a preferred name, and one undo."""
    keying.invalidate()
    with writer.batch(a, "desk", "individual", RIAN, commit=True) as b:
        writer.record(b, "product_variant", 1, "name", "1 Million Elixir", reason="typed")
        writer.record(b, "product_variant", 1, "attribute:concentration", "elixir")
        writer.record(b, "product_variant", 3, "attribute:quantity", {"value": 100, "unit": "ml", "form": "single"})
        writer.record(b, "product_line", 1, "hidden", True, reason="not yet")
        writer.record(b, "listing", 1, "pinned_to", 1)
        writer.record(b, "listing", 2, "ignored", True, reason="a set")
        writer.record(b, "attribute_wording", "wording:beauty|elixir parfum intense", "meaning", "elixir")
    keying.invalidate()
    with writer.batch(a, "desk", "individual", RIAN, commit=True) as b2:
        from app.services.decisions import natural_keys
        pair = natural_keys.pair_key("product", 1, 3, sides=(f"variant:{a.get(ProductVariant, 1).uid}", f"variant:{a.get(ProductVariant, 3).uid}"))[0]
        writer.record(b2, "suggestion", pair, "decision", {"decision": "separate", "survivor": None, "name": None, "note": "different juice"})
        from app.services import merges
        merges.apply_brand_alias(a, a.get(Brand, 2), a.get(Brand, 1), decided_by=RIAN, preferred_name="Rabanne", batch=b2)
    keying.invalidate()
    with writer.batch(a, "route", "individual", RIAN, commit=True, tail=False) as b3:
        hidden = effective(a, "product_line", [1], ["hidden"])[(1, "hidden")]
        undo_service.undo(b3, hidden.id, "show it after all")
    return b, b2, b3


def export_text(a) -> str:
    out = io.StringIO()
    replay_service.export(a, all_=True, out=out)
    return out.getvalue()


def test_the_export_carries_no_local_id_and_replays_onto_shifted_ids_with_every_column_equal(copies, tmp_path):
    a, b = copies
    ba, bb, bundo = decide_on_a(a)
    text = export_text(a)
    for line in text.splitlines():
        row = json.loads(line)
        for key, value in row.items():
            assert not (key == "id" or key.endswith("_id")) or value is None, f"a local id leaked: {key}"
    file = tmp_path / "a.jsonl"; file.write_text(text)
    counts = replay_service.replay(b, file, label="A")
    assert counts["parked"] == 0 and counts["held"] == 0, counts
    assert counts["applied"] == len([1 for l in text.splitlines() if json.loads(l)["kind"] == "decision"])
    # Every materialised column on B equals A's, alias-followed and by uid.
    for va in a.scalars(select(ProductVariant)):
        vb = b.scalar(select(ProductVariant).where(ProductVariant.uid == va.uid))
        assert (vb.name, vb.attributes.get("attribute"), vb.quantity_ml) == (va.name, va.attributes.get("attribute"), va.quantity_ml)
        assert (vb.merged_into_id is None) == (va.merged_into_id is None)
    assert b.get(Brand, 102).alias_of_id == 101 and b.get(Brand, 101).name == "Rabanne"
    assert b.get(ProductLine, 102).alias_of_id == 101 and b.get(ProductLine, 101).hidden is False
    assert b.get(Listing, 101).pinned_variant_id == 101 and b.get(Listing, 102).ignored_at is not None
    assert b.scalar(select(Suggestion).where(Suggestion.decision == "separate")) is not None
    assert b.scalar(select(AttributeAlias).where(AttributeAlias.raw == "elixir parfum intense")).canonical == "elixir"
    a_uids = {str(u) for u in a.scalars(select(Decision.uid))}
    b_uids = {str(u) for u in b.scalars(select(Decision.uid))}
    assert a_uids <= b_uids and b.scalar(select(DecisionBatch).where(DecisionBatch.uid == ba.uid)) is not None
    # B's surplus is consequences only: the alias cascade over its extra derived line.
    surplus = list(b.scalars(select(Decision).where(Decision.uid.in_([__import__("uuid").UUID(x) for x in b_uids - a_uids]))))
    assert surplus and all(d.caused_by_id is not None for d in surplus) and {d.entity_id for d in surplus} == {109}
    # A second replay applies nothing.
    again = replay_service.replay(b, file, label="A")
    assert again["applied"] == 0 and again["already"] == counts["applied"] and again["parked"] == 0
    # A batch made on A is undone on B by its uid.
    undone = undo_service.undo_batch(b, bb.uid, RIAN, "undo the alias on B")
    assert undone.reverses_batch_id == b.scalar(select(DecisionBatch.id).where(DecisionBatch.uid == bb.uid))
    assert b.get(Brand, 102).alias_of_id is None


def test_an_older_row_never_overrides_a_newer_target_and_a_bulk_row_parks_over_a_person(copies, tmp_path):
    a, b = copies
    with writer.batch(a, "route", "individual", RIAN, commit=True, tail=False) as ba:
        writer.record(ba, "product_variant", 2, "name", "Older on A", decided_at=datetime.now(UTC) - timedelta(hours=2))
    with writer.batch(a, "sheet", "bulk", RIAN, commit=True, tail=False) as ba2:
        writer.record(ba2, "product_variant", 3, "name", "Sheet name three", decided_at=datetime.now(UTC) - timedelta(hours=2))
    with writer.batch(b, "route", "individual", RIAN, commit=True, tail=False) as bb:
        writer.record(bb, "product_variant", 102, "name", "Newer on B")            # newer than the replayed row
        writer.record(bb, "product_variant", 103, "name", "A person on B", decided_at=datetime.now(UTC) - timedelta(hours=3))  # older but a person
    file = tmp_path / "a.jsonl"; file.write_text(export_text(a))
    counts = replay_service.replay(b, file, label="A")
    assert counts["superseded"] == 1 and counts["held"] == 1 and counts["applied"] == 0
    assert b.get(ProductVariant, 102).name == "Newer on B" and b.get(ProductVariant, 103).name == "A person on B"
    inserted = b.scalar(select(Decision).where(Decision.entity_id == 102, Decision.replayed_from.isnot(None)))
    assert inserted is not None and inserted.detail["conflict"]["kept"] == "target"
    remainder = pathlib.Path(str(file) + ".remainder.jsonl")
    assert remainder.exists() and json.loads(remainder.read_text().splitlines()[0])["park_code"] == "DECISION_HELD"


def test_check_writes_nothing_and_an_undone_batch_refuses_a_batch_export(copies, tmp_path):
    a, b = copies
    ba, bb, bundo = decide_on_a(a)
    file = tmp_path / "a.jsonl"; file.write_text(export_text(a))
    before = b.scalar(select(Decision.id).order_by(Decision.id.desc()))
    counts = replay_service.replay(b, file, check=True, label="A")
    assert counts["check"] and counts["applied"] > 0
    assert b.scalar(select(Decision.id).order_by(Decision.id.desc())) == before
    with pytest.raises(writer.Refused) as refused:
        undone = undo_service.undo_batch(a, ba.uid, RIAN, "undo A")
        replay_service.export(a, batch_uid=str(ba.uid), out=io.StringIO())
    assert refused.value.code == "BATCH_UNDONE"