"""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 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) 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))
    for vid in db.scalars(select(ProductVariant.id).where(ProductVariant.brand_id == alias.id, ProductVariant.merged_into_id.is_(None))):
        b.touch(vid)
    b.notes[row.id] = {"lines_created": created, "lines_pointed": pointed, "line_decisions_moved": moved_n, "line_decisions_kept": kept_n}
    _preferred_name(b, row, brand, "brand")
    b.touched_levels.add("brand")


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


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


# --------------------------------------------------------------------------- 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 ("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)
    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[("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)]