"""The effective read: newest wins by (decided_at, id), the kit does not invert it, a release
empties the field, and a kept-separate pair follows its sides through a merge (Stream K2,
spec §3, tests 2 and 8's read half). What it cost: a `DISTINCT ON` read would compile to a
plain DISTINCT on SQLite and pick the oldest row; the replay of an older staging decision
would then silently override a newer production one. SQLite kit."""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

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

from app.models import Account, Base, Brand, Decision, DecisionBatch, LEDGER_TABLES, ProductVariant, Suggestion
from app.services import keying
from app.services.decisions import effective, separated

TABLES = [Account.__table__, Brand.__table__, ProductVariant.__table__, Suggestion.__table__, *LEDGER_TABLES]
T0 = datetime(2026, 9, 17, 10, 0, tzinfo=UTC)


@pytest.fixture
def db():
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    Base.metadata.create_all(engine, tables=TABLES)
    with sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)() as s:
        s.add(DecisionBatch(id=1, origin_host="kit", kind="cli", mode="individual"))
        s.flush()
        yield s


def row(db, entity_id, field, value, at, effect="set", **kw):
    d = Decision(uid=uuid.uuid4(), decided_at=at, origin_host="kit", entity_type="product_variant", entity_id=entity_id,
                 natural_key=f"variant:{entity_id}", field=field, effect=effect, value=value, rules_version="5",
                 origin="person", batch_id=1, mode="individual", **kw)
    db.add(d); db.flush()
    return d


def test_newest_wins_by_decided_at_then_id_whichever_order_they_load(db):
    later = row(db, 7, "name", "Two", T0 + timedelta(minutes=5))      # lower id, newer time
    earlier = row(db, 7, "name", "One", T0)                          # higher id, older time
    assert effective(db, "product_variant")[(7, "name")] is later
    third = row(db, 7, "name", "Three", T0 + timedelta(minutes=5))   # same time, higher id
    assert effective(db, "product_variant", [7], ["name"])[(7, "name")] is third
    assert effective(db, "product_variant", [], ["name"]) == {}


def test_a_release_empties_the_field_and_a_later_set_fills_it_again(db):
    row(db, 7, "name", "One", T0)
    row(db, 7, "name", None, T0 + timedelta(minutes=1), effect="release")
    assert (7, "name") not in effective(db, "product_variant")
    row(db, 7, "name", "Two", T0 + timedelta(minutes=2))
    assert effective(db, "product_variant")[(7, "name")].value == "Two"


def test_the_maps_read_the_ledger_with_the_legacy_names_and_the_separated_pairs(db):
    brand = Brand(slug="b", name="B"); db.add(brand); db.flush()
    a = ProductVariant(id=1, name="A 1L", brand="B", brand_id=brand.id, match_key="k", vertical="liquor")
    b = ProductVariant(id=2, name="B 1L", brand="B", brand_id=brand.id, match_key="k", vertical="liquor")
    c = ProductVariant(id=3, name="C 1L", brand="B", brand_id=brand.id, match_key="k", vertical="liquor")
    db.add_all([a, b, c]); db.flush()
    row(db, 1, "product_line", "line:x", T0, value_ref_id=42, rule_value=41)
    row(db, 1, "attribute:quantity", {"value": 100, "unit": "ml"}, T0)
    row(db, 1, "attribute:concentration", "edp", T0, rule_value="edt")
    db.add(Suggestion(level="product", left_id=1, right_id=2, reason="key", decision="separate")); db.flush()
    keying.invalidate()
    maps = keying.load_maps(db)
    slot = maps.decided[1]
    assert slot["product_line_id"].value == 42 and slot["product_line_id"].collected_value == 41
    assert slot["quantity"].value == {"value": 100, "unit": "ml"}
    assert slot["attribute"].value == "edp" and slot["attribute"].collected_value == "edt"
    assert maps.separated == {frozenset((1, 2))}
    # After 2 merges into 3 the veto follows: {1, 3} is still separate.
    b.merged_into_id = 3; db.flush()
    assert separated(db) == {frozenset((1, 3))}