"""The proposals store's thin writer (Stream K2; spec §1.3, §1.4, §6.2's shape): a pass by name,
rows upserted on `(pass, entity_type, natural_key, field)`, each resolved on this host in a
savepoint. K4 owns the load command, the sheet and the approval; K3's arrival proposals write
through this too.

Sources of truth: this module, `models/decisions.py`, `tests/test_proposals_store.py`, the brief's
signature block.
"""

from __future__ import annotations

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

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.models import Proposal, ProposalPass
from app.services.decisions import natural_keys
from app.services.decisions.effective import effective


@dataclass
class ProposalRow:
    entity_type: str
    natural_key: str
    natural_key_detail: dict | None
    field: str
    value: Any
    reason: str | None
    evidence: list
    confidence: float | None
    brand_slug: str
    sheet_line_ref: str | None = None
    position: int = 0


def get_or_create_pass(db: Session, name: str, *, kind: str, generator: str, rules_version: str, process_version: str,
                       scope_brand_slug: str | None = None, note: str | None = None, loaded_by=None, source_file: str | None = None,
                       file_sha256: str | None = None) -> ProposalPass:
    row = db.scalar(select(ProposalPass).where(ProposalPass.name == name))
    if row is None:
        savepoint = db.begin_nested()
        try:
            row = ProposalPass(name=name, kind=kind, generator=generator, rules_version=rules_version, process_version=process_version,
                               scope_brand_slug=scope_brand_slug, note=note, source_file=source_file, file_sha256=file_sha256,
                               loaded_by=getattr(loaded_by, "id", loaded_by), loaded_by_username=getattr(loaded_by, "username", None))
            db.add(row)
            savepoint.commit()
        except IntegrityError:
            savepoint.rollback()
            row = db.scalar(select(ProposalPass).where(ProposalPass.name == name))
    return row


def write(db: Session, pass_name: str, rows: list[ProposalRow], *, kind: str, generator: str, rules_version: str,
          process_version: str, scope_brand_slug: str | None = None, **pass_kw) -> ProposalPass:
    """Get-or-create the pass by name; upsert every row on the idempotency key, resolving its entity
    on this host (`resolution`: resolved | creatable | unresolved), the effective decision on the
    field recorded as `against_decision_id`; an open row takes the new value, a resolved (approved,
    rejected) row is left untouched; each row in its own savepoint."""
    pass_row = get_or_create_pass(db, pass_name, kind=kind, generator=generator, rules_version=rules_version,
                                  process_version=process_version, scope_brand_slug=scope_brand_slug, **pass_kw)
    counts = {"inserted": 0, "updated": 0, "kept": 0, "parked": 0}
    for spec in rows:
        existing = db.scalar(select(Proposal).where(Proposal.pass_id == pass_row.id, Proposal.entity_type == spec.entity_type,
                                                    Proposal.natural_key == spec.natural_key, Proposal.field == spec.field))
        if existing is not None and existing.status not in ("open", "stale", "parked", "withdrawn"):
            counts["kept"] += 1
            continue
        entity_id, resolution, note = _resolve(db, spec)
        against = None
        if entity_id is not None:
            dec = effective(db, spec.entity_type, [entity_id], [spec.field]).get((entity_id, spec.field))
            against = dec.id if dec is not None else None
        savepoint = db.begin_nested()
        try:
            if existing is None:
                existing = Proposal(pass_id=pass_row.id, entity_type=spec.entity_type, natural_key=spec.natural_key, field=spec.field,
                                    brand_slug=spec.brand_slug, generator=generator)
                db.add(existing)
                counts["inserted"] += 1
            else:
                counts["updated"] += 1
            existing.natural_key_detail = spec.natural_key_detail
            existing.value, existing.reason, existing.evidence, existing.confidence = spec.value, spec.reason, spec.evidence or [], spec.confidence
            existing.sheet_line_ref, existing.position, existing.brand_slug = spec.sheet_line_ref, spec.position, spec.brand_slug
            existing.entity_id, existing.resolution, existing.against_decision_id = entity_id, resolution, against
            existing.status = "parked" if resolution == "unresolved" else "open"
            existing.resolution_note = note
            existing.updated_at = datetime.now(UTC)
            savepoint.commit()
        except IntegrityError:
            savepoint.rollback()
            counts["kept"] += 1
        if resolution == "unresolved":
            counts["parked"] += 1
    pass_row.counts = {**(pass_row.counts or {}), **counts, "at": datetime.now(UTC).isoformat()}
    db.flush()
    return pass_row


def _resolve(db: Session, spec: ProposalRow) -> tuple[int | None, str, str | None]:
    if spec.entity_type == "product_line" and spec.natural_key.startswith("line:new:"):
        found = natural_keys.resolve(db, "product_line", spec.natural_key, spec.natural_key_detail)
        if isinstance(found, natural_keys.Parked):
            return None, "creatable", None
        return found.id, "resolved", None
    found = natural_keys.resolve(db, spec.entity_type, spec.natural_key, spec.natural_key_detail)
    if isinstance(found, natural_keys.Parked):
        return None, "unresolved", f"{found.code}: {found.detail or ''}".strip()
    return found.id, "resolved", None