"""The appliers (spec §4 step f, §5): today's code moved behind one registry, each split in
`materialise(row)` (this row's own column, nothing else), `consequences(row)` (the cascade,
minted through `record(cause=row)` so every consequence is a row with the same provenance and a
uuid5 uid) and `release(row, prev)` (an undo or a clear: the rules' value NOW, or the predecessor).

Sources of truth: this module, `writer.py`, `merges.py` (the merge record), `tests/test_writer.py`.
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any, Callable

from sqlalchemy import func, select

from app.models import Brand, Listing, ProductLine, ProductVariant, Suggestion
from app.services.decisions import natural_keys


@dataclass(frozen=True)
class Applier:
    materialise: Callable
    consequences: Callable
    release: Callable
    prepare: Callable = None  # (b, entity, value) -> detail additions, computed BEFORE the row is inserted


def _noop(*a, **k) -> None:
    return None


def _none(*a, **k):
    return None


def _now() -> datetime:
    return datetime.now(UTC)


# --------------------------------------------------------------------------- product variant

def _variant_name(b, row, variant, value):
    variant.name = value


def _variant_name_release(b, row, variant, prev):
    # The rules' name is the collected spelling every key part is computed from (keep-first).
    restored = row.value if row.effect == "set" else (row.rule_value or variant.name)
    variant.name = restored


def _variant_line(b, row, variant, line):
    variant.product_line_id = line.id
    b.maps.remember_line(line) if b.maps is not None and line is not None else None


def _variant_line_release(b, row, variant, prev):
    from app.services import keying

    if row.effect == "set":
        variant.product_line_id = row.value_ref_id
        return
    brand = b.maps.brand_of(variant.brand_id, variant.brand, variant.vertical) if b.maps else None
    line = keying.product_line_for(b.db, b.maps, brand, keying.keyed_name(variant, b.maps), variant.brand, variant.vertical,
                                   create=True, category=variant.category) if brand is not None else None
    variant.product_line_id = line.id if line is not None else variant.product_line_id


def _variant_attribute(b, row, variant, value):
    from app.services import attributes

    attributes.set(variant, row.field[len("attribute:"):], value)


def _variant_attribute_release(b, row, variant, prev):
    from app.services import attributes, keying

    kind = row.field[len("attribute:"):]
    if row.effect == "set":
        attributes.set(variant, kind, row.value)
        return
    if kind in ("concentration", "color", "flavor"):
        rule = b.maps.attribute(keying.keyed_name(variant, b.maps), variant.vertical, category=variant.category) if b.maps else None
        attributes.set(variant, kind, rule or None)
    elif kind == "quantity":
        attributes.set(variant, kind, row.rule_value if isinstance(row.rule_value, dict) else None)
    else:
        attributes.set(variant, kind, row.rule_value)


def _variant_merged_into(b, row, loser, survivor):
    from app.services import merges

    note = (row.detail or {}).get("note")
    merges.merge_product_variants(b.db, survivor, [loser], reason="confirmed", merged_by=b.by_id, note=note, decision=row, batch=b)
    b.touch(survivor.id)


def _variant_merged_consequences(b, row, loser, survivor):
    """The carry (spec §5): every effective `set` decision the loser holds on a field the survivor
    has none for becomes a consequence row on the survivor, `rule_value` recomputed there; a field
    the survivor decided is kept and listed; every `separate` pair the loser was in is carried."""
    from app.services.decisions.effective import effective
    from app.services.decisions.writer import record

    theirs = effective(b.db, "product_variant", [loser.id])
    ours = effective(b.db, "product_variant", [survivor.id])
    carried, kept = [], []
    for (_, field_name), theirs_row in theirs.items():
        if field_name == "merged_into":
            continue
        if (survivor.id, field_name) in ours:
            kept.append(field_name)
            continue
        value = theirs_row.value_ref_id if theirs_row.value_ref_id is not None else theirs_row.value
        savepoint = b.db.begin_nested()
        try:
            record(b, "product_variant", survivor, field_name, value, origin=theirs_row.origin, reason=theirs_row.reason,
                   cause=row, detail={"carried_from": theirs_row.natural_key}, decided_by=theirs_row.decided_by)
            savepoint.commit()
            carried.append(field_name)
        except Exception as exc:  # a carried reference no longer live: listed, never fatal
            savepoint.rollback()
            kept.append(f"{field_name} ({getattr(exc, 'code', None) or str(exc)[:120]})")
    for pair in b.db.scalars(select(Suggestion).where(Suggestion.level == "product", Suggestion.decision == "separate",
                                                     (Suggestion.left_id == loser.id) | (Suggestion.right_id == loser.id))):
        other = pair.right_id if pair.left_id == loser.id else pair.left_id
        if other == survivor.id:
            continue
        key, _ = natural_keys.pair_key("product", survivor.id, other, sides=(f"variant:{survivor.uid}", f"variant:{b.db.get(ProductVariant, other).uid}"))
        savepoint = b.db.begin_nested()
        try:
            record(b, "suggestion", key, "decision", {"decision": "separate", "survivor": None, "name": None, "note": "carried from a merged variant"},
                   cause=row, detail={"carried_from": natural_keys.build(loser)[0]})
            savepoint.commit()
        except Exception:
            savepoint.rollback()
            continue
    from app.models import Merge

    rec = b.db.scalar(select(Merge).where(Merge.decision_id == row.id))
    if rec is not None and (carried or kept):
        rec.detail = {**(rec.detail or {}), "carry": {"carried": sorted(carried), "kept": sorted(kept)}}
    b.notes[row.id] = {"carried": sorted(carried), "kept": sorted(kept)}


def _variant_merged_release(b, row, loser, prev):
    from app.services.decisions.undo import unmerge

    unmerge(b, row, loser)


# --------------------------------------------------------------------------- brand and product line

def _alias(b, row, alias, target):
    alias.alias_of_id = target.id
    alias.decided_by, alias.decided_at = b.by_id, _now()


def _brand_alias_consequences(b, row, alias: Brand, brand: Brand):
    """`apply_brand_alias`'s cascade as rows: every line of the alias points at the brand's line
    with the same key (created when the brand had none) and its variants move there; every
    variant of the alias is re-keyed under the brand at the batch's tail."""
    from sqlalchemy import update

    from app.services import merges, product_lines
    from app.services.decisions.writer import record

    db = b.db
    brand_lines = {r.key: r for r in db.scalars(select(ProductLine).where(ProductLine.brand_id == brand.id))}
    created = pointed = moved_n = kept_n = 0
    for line in list(db.scalars(select(ProductLine).where(ProductLine.brand_id == alias.id))):
        target = brand_lines.get(line.key)
        if target is None:
            target = ProductLine(brand_id=brand.id, key=line.key, name=line.name,
                                 slug=merges._unique_slug(db, product_lines.line_slug(brand.slug, line.key)[:230]))
            db.add(target)
            db.flush()
            brand_lines[line.key] = target
            if b.maps is not None:
                b.maps.remember_line(target)
            created += 1
        if line.alias_of_id != target.id and line.id != target.id:
            line_row = record(b, "product_line", line, "alias_of", target, cause=row,
                              reason=f"follows the brand alias {alias.name} to {brand.name}")
            pointed += 1
            # The alias line's own decisions (a preferred name, a hidden) follow its variants to
            # the target line where it has none of its own; the rest are listed as kept.
            from app.services.overrides import move

            carried = move(db, "product_line", str(line.id), str(target.id), batch=b, cause=line_row)
            moved_n += len(carried["moved"])
            kept_n += len(carried["kept"])
        else:
            db.execute(update(ProductVariant).where(ProductVariant.product_line_id == line.id).values(product_line_id=target.id))
    # The variants move to the survivor, not merely re-key under it. Touching them alone left
    # `brand_id` on the row the fold emptied, and three things followed: `resolve_brand` read that
    # spelling as still belonging there, so the NEXT collection filed new listings on the brand a
    # person had just folded away; anything counting by `brand_id` undercounted the survivor (the
    # brand index showed four variants where five had moved); and the catalogue slowly regrew the
    # folded brand with nothing recording why. Reads that follow aliases hid all of it.
    moving = [(vid, old_id) for vid, old_id in (row.detail or {}).get("brands_moved", [])]
    if moving:
        db.execute(update(ProductVariant).where(ProductVariant.brand_id == alias.id).values(brand_id=brand.id))
    for vid, _old in moving:
        b.touch(vid)
    b.notes[row.id] = {"lines_created": created, "lines_pointed": pointed, "line_decisions_moved": moved_n,
                       "line_decisions_kept": kept_n, "variants_moved": len(moving)}
    _preferred_name(b, row, brand, "brand")
    b.touched_levels.add("brand")


def _brand_alias_prepare(b, alias: Brand, brand: Brand) -> dict:
    """What an undo needs, computed BEFORE the row is written: every variant the alias moves.

    It has to be here rather than in the consequences, because `decisions` is append-only and a
    database trigger refuses an UPDATE on a row already inserted. Writing the detail afterwards
    passed every SQLite test and failed on the first real confirm against Postgres, which is what
    a rehearsal on a copy is for."""
    # The keys of the alias's lines the brand has no twin for: the consequences mint one line per
    # key, and the undo needs their keys to prune them. Cost: every fold-and-undo left its minted
    # lines behind, reachable and empty (the simulation's 43 on three brands), because the ledger
    # keeps naming them and `prune_lines` rightly keeps a line a decision names.
    brand_keys = set(b.db.scalars(select(ProductLine.key).where(ProductLine.brand_id == brand.id)))
    minted = sorted({k for k in b.db.scalars(select(ProductLine.key).where(ProductLine.brand_id == alias.id)) if k not in brand_keys})
    return {"brands_moved": [[v.id, v.brand_id] for v in
                             b.db.scalars(select(ProductVariant).where(ProductVariant.brand_id == alias.id)
                                          .order_by(ProductVariant.id))],
            "lines_minted": minted}


def _line_alias_prepare(b, alias: ProductLine, target: ProductLine) -> dict:
    """What an undo needs, computed before the row is written: every variant the alias moves."""
    return {"products_moved": [[v.id, v.product_line_id] for v in
                               b.db.scalars(select(ProductVariant).where(ProductVariant.product_line_id == alias.id).order_by(ProductVariant.id))]}


def _line_alias(b, row, alias: ProductLine, target: ProductLine):
    """Point the line and move the variants the row's detail lists (recorded before insert)."""
    alias.alias_of_id = target.id
    alias.decided_by, alias.decided_at = b.by_id, _now()
    for vid, _old in (row.detail or {}).get("products_moved", []):
        v = b.db.get(ProductVariant, vid)
        if v is None:
            continue
        v.product_line_id = target.id
        if v.merged_into_id is None:
            b.touch(v.id)
    if b.maps is not None:
        b.maps.remember_line(alias)
    b.touched_levels.add("line")


def _preferred_name(b, row, target, entity_type: str) -> None:
    """A preferred name typed at confirm is the alias row's consequence (`detail.preferred_name`),
    so a replay regenerates it from the row it exports."""
    from app.services.decisions.writer import record

    typed = ((row.detail or {}).get("preferred_name") or "").strip()
    if typed:
        record(b, entity_type, target, "name", typed[:200], cause=row, reason="preferred name at confirm")


def _line_alias_consequences(b, row, alias: ProductLine, target: ProductLine):
    """A moved variant holding a `product_line` decision naming the alias line carries one naming
    the target (spec §4 f); a preferred name typed at confirm is a consequence too; and the slug
    hook fires HERE, because this applier is the one path every caller shares.

    K3.6 first called `publish.slug_at_confirm` from `merges.apply_line_alias`, which only tests
    call: the desk's Confirm same, the sheet's absorb and replay all record `alias_of` through
    `writer.record`, so on production no confirmed alias would ever have fixed a slug (K6 found
    it). The slug is a pure function of the brand slug and the chosen name, so replay fixes the
    same address on both hosts."""
    from app.services import publish
    from app.services.decisions.effective import effective
    from app.services.decisions.writer import record

    _preferred_name(b, row, target, "product_line")
    chosen = ((row.detail or {}).get("preferred_name") or "").strip() or target.name
    publish.slug_at_confirm(target, chosen, decision=row)

    ids = [vid for vid, _ in (row.detail or {}).get("products_moved", [])]
    for (vid, _), dec in effective(b.db, "product_variant", ids, ["product_line"]).items():
        if dec.value_ref_id == alias.id:
            record(b, "product_variant", b.db.get(ProductVariant, vid), "product_line", target, cause=row,
                   reason="follows the line alias", detail={"carried_from": dec.natural_key})


def _alias_release(b, row, alias, prev):
    from sqlalchemy import update

    if row.effect == "set":
        alias.alias_of_id = row.value_ref_id
        return
    alias.alias_of_id = None
    alias.decided_by, alias.decided_at = b.by_id, _now()
    reversed_row = prev
    for vid, old_brand in ((reversed_row.detail or {}).get("brands_moved", []) if reversed_row is not None else []):
        v = b.db.get(ProductVariant, vid)
        if v is not None:
            v.brand_id = old_brand
            if v.merged_into_id is None:
                b.touch(v.id)
    for vid, old_line in ((reversed_row.detail or {}).get("products_moved", []) if reversed_row is not None else []):
        v = b.db.get(ProductVariant, vid)
        if v is not None:
            v.product_line_id = old_line
            if v.merged_into_id is None:
                b.touch(v.id)
    if isinstance(alias, Brand):
        for vid in b.db.scalars(select(ProductVariant.id).where(ProductVariant.brand_id == alias.id, ProductVariant.merged_into_id.is_(None))):
            b.touch(vid)


def _named(b, row, entity, value):
    entity.name = value
    if hasattr(entity, "decided_by"):
        entity.decided_by, entity.decided_at = b.by_id, _now()


def _named_release(b, row, entity, prev):
    entity.name = row.value if row.effect == "set" else (row.rule_value or entity.name)


def _flag(field_name):
    def materialise(b, row, entity, value):
        setattr(entity, field_name, bool(value))

    def release(b, row, entity, prev):
        setattr(entity, field_name, bool(row.value) if row.effect == "set" else False)

    return materialise, release


# --------------------------------------------------------------------------- the brand split

def _brand_split_prepare(b, brand: Brand, value: dict) -> dict:
    """What the undo needs, computed before the row is written (the ledger is append-only, so a
    row's detail is complete at insert): every variant the split moves with the brand and product
    line it came from, and the brand row it moves them to, adopted when the slug already holds one
    and minted otherwise.

    Tombstones move with their spelling too: a merged-away row whose survivor changed brand would
    otherwise keep a key under the old one, and `ingest._follow_tombstone` reads that key.
    """
    from app.services.brands import spelling_key

    db = b.db
    claimed = {spelling_key(s) for s in value["spellings"]}
    moved = [[v.id, v.brand_id, v.product_line_id] for v in
             db.scalars(select(ProductVariant).where(ProductVariant.brand_id == brand.id).order_by(ProductVariant.id))
             if spelling_key(v.brand) in claimed]
    target = db.scalar(select(Brand).where(Brand.slug == value["brand_slug"]))
    minted = target is None
    if minted:
        target = Brand(slug=value["brand_slug"], name=value["name"])
        db.add(target)
        db.flush()
    return {"moved": moved, "new_brand": {"slug": target.slug, "uid": str(target.uid), "name": target.name, "minted": minted}}


def _split_target(b, row) -> Brand | None:
    return b.db.scalar(select(Brand).where(Brand.slug == ((row.value or {}).get("brand_slug") or "")))


def _remember_brand(b, target: Brand) -> None:
    """The batch's maps learn the minted row, so the rekey at the tail keys under it."""
    if b.maps is not None:
        b.maps.brands[target.id] = target
        b.maps.by_slug[target.slug] = target


def _line_under(b, target: Brand, variant: ProductVariant):
    """The product line the moved variant belongs to under its new brand, re-derived by the rules
    (its old line's key was computed with the old brand's words, so it cannot simply be copied)."""
    from app.services import keying

    return keying.product_line_for(b.db, b.maps, target, keying.keyed_name(variant, b.maps), variant.brand,
                                   variant.vertical, create=True, category=variant.category)


def _brand_split(b, row, brand: Brand, value: dict):
    """Move the claimed spellings onto the new row and re-derive their product lines; a variant
    whose product line a person decided is left for `consequences`, because a rule never
    overwrites a decision. Then prune the line rows the move emptied."""
    from app.services import merges
    from app.services.decisions.effective import effective

    db = b.db
    target = _split_target(b, row)
    if target is None:
        return
    _remember_brand(b, target)
    moved = (row.detail or {}).get("moved", [])
    decided = {vid for (vid, _f) in effective(db, "product_variant", [m[0] for m in moved], ["product_line"])}
    before = db.scalar(select(func.count(ProductLine.id)).where(ProductLine.brand_id == target.id)) or 0
    emptied: set[int] = set()
    for vid, _old_brand_id, old_line_id in moved:
        variant = db.get(ProductVariant, vid)
        if variant is None:
            continue
        variant.brand_id = target.id
        if vid not in decided:
            line = _line_under(b, target, variant)
            if line is not None:
                variant.product_line_id = line.id
        if old_line_id:
            emptied.add(old_line_id)
        b.touch(vid)
    db.flush()
    after = db.scalar(select(func.count(ProductLine.id)).where(ProductLine.brand_id == target.id)) or 0
    pruned = merges.prune_lines(db, emptied)
    b.notes[row.id] = {"product_variants_moved": len(moved), "lines_created": after - before, "lines_pruned": pruned,
                       "brand_slug": target.slug, "line_decisions_carried": 0}
    b.touched_levels.add("brand")


def _brand_split_consequences(b, row, brand: Brand, value: dict):
    """A moved variant whose product line a PERSON decided carries a decision naming the matching
    line under the new brand: the old line belongs to the brand it no longer sits on, and leaving
    it there is drift `decisions verify` would report at the next deploy."""
    from app.services.decisions.effective import effective
    from app.services.decisions.writer import record

    db = b.db
    target = _split_target(b, row)
    if target is None:
        return
    moved = [m[0] for m in (row.detail or {}).get("moved", [])]
    carried = 0
    for (vid, _f), dec in effective(db, "product_variant", moved, ["product_line"]).items():
        variant = db.get(ProductVariant, vid)
        old = db.get(ProductLine, dec.value_ref_id) if dec.value_ref_id else None
        if variant is None or old is None or old.brand_id == target.id:
            continue
        line = _line_named(b, target, old)
        record(b, "product_variant", variant, "product_line", line, cause=row, reason="follows the brand split",
               detail={"carried_from": dec.natural_key})
        carried += 1
    b.notes.setdefault(row.id, {})["line_decisions_carried"] = carried


def _line_named(b, target: Brand, old: ProductLine) -> ProductLine:
    """The counterpart under the new brand of a line a person decided: the same key and name,
    its own address. Created once; found by key on a repeat."""
    from app.services import merges, product_lines

    db = b.db
    row = db.scalar(select(ProductLine).where(ProductLine.brand_id == target.id, ProductLine.key == old.key))
    if row is None:
        row = ProductLine(brand_id=target.id, key=old.key, name=old.name,
                          slug=merges._unique_slug(db, product_lines.line_slug(target.slug, old.key)[:230]))
        db.add(row)
        db.flush()
        if b.maps is not None:
            b.maps.remember_line(row)
    return row


def _brand_split_release(b, row, brand: Brand, prev):
    """The undo (spec §7): every moved variant back on the brand and product line the reversed
    row recorded, re-keyed at the tail, and the minted brand row deleted only when nothing
    references it. Nothing is folded back: the fold is what the split disagreed with."""
    from app.services import merges

    db = b.db
    if row.effect == "set":  # an earlier split restored: apply it again
        _brand_split(b, row, brand, row.value)
        return
    reversed_row = prev
    if reversed_row is None:
        return
    detail, value = reversed_row.detail or {}, reversed_row.value or {}
    target = db.scalar(select(Brand).where(Brand.slug == (value.get("brand_slug") or "")))
    emptied: set[int] = set()
    for vid, old_brand_id, old_line_id in detail.get("moved", []):
        variant = db.get(ProductVariant, vid)
        if variant is None:
            continue
        if variant.product_line_id:
            emptied.add(variant.product_line_id)
        variant.brand_id = old_brand_id
        old_line = db.get(ProductLine, old_line_id) if old_line_id else None
        if old_line is not None:
            variant.product_line_id = old_line.id
        elif old_brand_id is not None:
            source = db.get(Brand, old_brand_id)
            line = _line_under(b, source, variant) if source is not None else None
            if line is not None:
                variant.product_line_id = line.id
        b.touch(vid)
    db.flush()
    pruned = merges.prune_lines(db, emptied)
    removed = _drop_split_brand(db, target, (detail.get("new_brand") or {}))
    b.notes[row.id] = {"product_variants_moved": len(detail.get("moved", [])), "lines_pruned": pruned,
                       "brand_removed": removed}
    b.touched_levels.add("brand")


def _drop_split_brand(db, target: Brand | None, minted: dict) -> bool:
    """Delete the row the split minted, and only when nothing points at it: a variant (live or a
    merge tombstone), a product line, an alias, a brand suggestion still open or a person decided,
    any decision of its own, or a redirect. A row the split ADOPTED is never deleted; it existed
    before the split and is not the split's to remove."""
    from app.models import Decision, Redirect, Suggestion

    if target is None or not minted.get("minted") or minted.get("slug") != target.slug:
        return False
    if db.scalar(select(ProductVariant.id).where(ProductVariant.brand_id == target.id)):
        return False
    if db.scalar(select(ProductLine.id).where(ProductLine.brand_id == target.id)):
        return False
    if db.scalar(select(Brand.id).where(Brand.alias_of_id == target.id)):
        return False
    if db.scalar(select(Suggestion.id).where(Suggestion.level == "brand",
                                            (Suggestion.left_id == target.id) | (Suggestion.right_id == target.id),
                                            Suggestion.is_open() | (Suggestion.decided_by.isnot(None)))):
        return False
    if db.scalar(select(Decision.id).where(Decision.entity_type == "brand", Decision.entity_id == target.id)):
        return False
    if db.scalar(select(Redirect.from_slug).where(Redirect.kind == "brand", Redirect.to_slug == target.slug)):
        return False
    db.delete(target)
    db.flush()
    return True


# --------------------------------------------------------------------------- wording, listing, pair, place

def _wording(field_name):
    column = {"meaning": "canonical", "display": "display", "kind": "kind"}[field_name]

    def materialise(b, row, wording, value):
        setattr(wording, column, value)
        wording.decided_by, wording.decided_at = b.by_id, _now()
        _rekey_wording(b, wording)

    def release(b, row, wording, prev):
        from app.services import product_lines

        if row.effect == "set":
            setattr(wording, column, row.value)
        elif field_name == "meaning":
            wording.canonical = product_lines.canonical_attribute(wording.raw)
        elif field_name == "display":
            wording.display = product_lines.display_attribute(wording.canonical)
        else:
            wording.kind = None
        _rekey_wording(b, wording)

    return materialise, release


def _rekey_wording(b, wording):
    from app.services import product_lines

    if b.maps is not None:
        b.maps.aliases[(wording.vertical, wording.raw)] = wording.canonical
        if wording.kind:
            b.maps.alias_kinds[(wording.vertical, wording.raw)] = wording.kind
    for v in b.db.scalars(select(ProductVariant).where(ProductVariant.vertical == wording.vertical, ProductVariant.merged_into_id.is_(None))):
        if product_lines.attribute_of(v.name, wording.vertical, category=v.category)[0] == wording.raw:
            b.touch(v.id)


def _pin(b, row, listing, variant):
    from app.services.ingest import survivor_of

    listing.pinned_variant_id, listing.pinned_by, listing.pinned_at = variant.id, b.by_id, _now()
    listing.variant_id = survivor_of(b.db, variant).id


def _pin_release(b, row, listing, prev):
    if row.effect == "set":
        listing.pinned_variant_id, listing.pinned_by, listing.pinned_at = row.value_ref_id, b.by_id, _now()
        return
    listing.pinned_variant_id, listing.pinned_by, listing.pinned_at = None, None, None


def _ignore(b, row, listing, value):
    if value:
        listing.ignored_at, listing.ignored_by = _now(), b.by_id
        listing.ignore_reason = (row.reason or "").strip()[:2000] or None
    else:
        listing.ignored_at, listing.ignored_by, listing.ignore_reason = None, None, None


def _ignore_release(b, row, listing, prev):
    _ignore(b, row, listing, row.value if row.effect == "set" else False)


def _pair(b, row, pair: Suggestion, value):
    pair.decision = value["decision"]
    pair.decided_by, pair.decided_at, pair.decision_id = b.by_id, _now(), row.id
    pair.detail = {**(pair.detail or {}), "decision": {"survivor": value.get("survivor"), "preferred_name": value.get("name"),
                                                        "note": value.get("note")}}
    b.touched_levels.add(pair.level)


def _pair_consequences(b, row, pair: Suggestion, value):
    """Confirm same (spec §5): at variant level a `merged_into` row on the loser; at brand or
    line level an `alias_of` row on the alias, plus a `name` row on the target when typed."""
    from app.services.decisions.writer import record

    if value["decision"] != "same":
        return
    level = pair.level
    model = {"brand": Brand, "line": ProductLine, "product": ProductVariant}[level]
    left, right = b.db.get(model, pair.left_id), b.db.get(model, pair.right_id)
    survivor_key = value.get("survivor")
    if survivor_key:
        survivor = left if natural_keys.build(left)[0] == survivor_key else right
    else:
        survivor = left
    other = right if survivor is left else left
    if level == "product":
        record(b, "product_variant", other, "merged_into", survivor, cause=row, reason=row.reason,
               detail={"note": value.get("note")} if value.get("note") else None)
    else:
        typed_name = (value.get("name") or "").strip()
        record(b, "brand" if level == "brand" else "product_line", other, "alias_of", survivor, cause=row, reason=row.reason,
               detail={"preferred_name": typed_name} if typed_name else None)
        return
    typed = (value.get("name") or "").strip()
    if typed and typed not in (left.name, right.name):
        record(b, "product_variant", survivor, "name", typed, cause=row, reason="chosen at merge")


def _pair_release(b, row, pair, prev):
    if row.effect == "set":
        pair.decision, pair.decision_id = row.value["decision"], row.id
    else:
        pair.decision, pair.decision_id, pair.decided_by, pair.decided_at = None, None, None, None


def _precedent_status(b, row, precedent, value):
    """`overturned` or `active` on the register row (K12.3; the plan's B6). An overturn regroups
    nothing: it marks the precedent, and the next pass reads the decisions made under it and
    proposes against them. The note is the new answer, carried in the row's detail."""
    detail = row.detail or {}
    precedent.status = value
    if value == "overturned":
        precedent.overturned_by, precedent.overturned_at = b.by_username, _now()
        precedent.overturn_note = detail.get("note")
        precedent.resets_count = (precedent.resets_count or 0) + 1
        precedent.streak = 0
    else:
        precedent.overturned_by = precedent.overturned_at = precedent.overturn_note = None


def _precedent_status_release(b, row, precedent, prev):
    """`row` is the undo row: a `set` restoring an earlier status, or a release back to active."""
    if row.effect == "set":
        _precedent_status(b, row, precedent, row.value)
        return
    precedent.status = "active"
    precedent.overturned_by = precedent.overturned_at = precedent.overturn_note = None


# --------------------------------------------------------------------------- the registry

def _make() -> dict[tuple[str, str], Applier]:
    reg: dict[tuple[str, str], Applier] = {}
    reg[("product_variant", "name")] = Applier(_variant_name, _noop, _variant_name_release)
    reg[("product_variant", "product_line")] = Applier(_variant_line, _noop, _variant_line_release)
    reg[("product_variant", "merged_into")] = Applier(_variant_merged_into, _variant_merged_consequences, _variant_merged_release)
    for et in ("product_variant", "product_line"):  # the admin's pin (Stream AW2): a flag like hidden
        m, r = _flag("featured")
        reg[(et, "featured")] = Applier(m, _noop, r)
    for et in ("brand", "product_line"):
        reg[(et, "name")] = Applier(_named, _noop, _named_release)
        for flag in ("hidden", "indexed"):
            m, r = _flag(flag)
            reg[(et, flag)] = Applier(m, _noop, r)
    reg[("brand", "alias_of")] = Applier(_alias, _brand_alias_consequences, _alias_release, prepare=_brand_alias_prepare)
    reg[("brand", "split")] = Applier(_brand_split, _brand_split_consequences, _brand_split_release, prepare=_brand_split_prepare)
    reg[("product_line", "alias_of")] = Applier(_line_alias, _line_alias_consequences, _alias_release, prepare=_line_alias_prepare)
    reg[("place", "name")] = Applier(_named, _noop, _named_release)
    for flag in ("hidden", "indexed"):
        m, r = _flag(flag)
        reg[("place", flag)] = Applier(m, _noop, r)
    for f in ("meaning", "display", "kind"):
        m, r = _wording(f)
        reg[("attribute_wording", f)] = Applier(m, _noop, r)
    reg[("precedent", "status")] = Applier(_precedent_status, _noop, _precedent_status_release)
    reg[("listing", "pinned_to")] = Applier(_pin, _noop, _pin_release)
    reg[("listing", "ignored")] = Applier(_ignore, _noop, _ignore_release)
    reg[("suggestion", "decision")] = Applier(_pair, _pair_consequences, _pair_release)
    return reg


APPLIERS: dict[tuple[str, str], Applier] = _make()
_ATTRIBUTE = Applier(_variant_attribute, _noop, _variant_attribute_release)


def for_field(entity_type: str, field_name: str) -> Applier:
    if entity_type == "product_variant" and field_name.startswith("attribute:"):
        return _ATTRIBUTE
    return APPLIERS[(entity_type, field_name)]