"""Manual evidence records, administrator approval and audit-safe reversal.

Manual records are immutable submitted facts. Only an approved record becomes
a canonical ledger event (``approve_new``) or joins one (``approve_link``);
returned/exception/pending records never affect a balance and never leak to
the counterparty. Approved facts change only through a ``reverse`` decision
that creates an opposite new event (or detaches a linked claim) — the original
is never edited. Idempotency keys and UNIQUE claims prevent double counting.
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
import json
import sqlite3

from .db import utc_now
from .ledger_events import (
    LedgerConflictError,
    LedgerInputError,
    create_event,
    create_reversal,
    current_revision,
    manual_source_claim,
)
from .subjects import SUBJECTS

MANUAL_STATES = ("pending", "approved", "returned", "exception", "reversed")
FUNDING_SOURCES = ("approved_bank_account", "personal_transit", "other")
DATE_KEYS = ("occurred_at",)


class ManualConflictError(ValueError):
    """A claim/idempotency/revision conflict (mapped to HTTP 409)."""


class ManualInputError(ValueError):
    """Invalid manual record input (mapped to HTTP 400/422)."""


def _parse_amount(amount: object) -> Decimal:
    try:
        value = Decimal(str(amount))
    except InvalidOperation:
        raise ManualInputError("金额不是有效的十进制数。") from None
    if not value.is_finite() or value <= 0:
        raise ManualInputError("金额必须大于零。")
    return value


def _validate_date(value: object, field: str) -> str:
    text = str(value or "").strip()
    if len(text) < 10:
        raise ManualInputError(f"{field}必须是 YYYY-MM-DD 或完整时间。")
    try:
        datetime.fromisoformat(text[:10])
    except ValueError:
        raise ManualInputError(f"{field}必须是 YYYY-MM-DD 或完整时间。") from None
    return text


def _business_today() -> str:
    """Shanghai business date (the default reversal effective date)."""
    return datetime.now(timezone(timedelta(hours=8))).date().isoformat()


# ---------------------------------------------------------------------------
# Submit
# ---------------------------------------------------------------------------


def submit(
    connection: sqlite3.Connection,
    *,
    company_id: int,
    counterparty_company_id: int,
    occurred_at: str,
    direction: str,
    amount: str,
    currency: str,
    funding_source: str,
    requested_subject: str,
    request_key: str,
    actor: sqlite3.Row,
    bank_account_id: object = None,
    personal_transit_mapping_id: object = None,
    related_source_row_id: object = None,
    summary: object = None,
    reason: object = None,
    evidence: object = None,
    supersedes_record_id: object = None,
) -> dict[str, object]:
    """Submit one manual record for review. Idempotent on ``(company_id, request_key)``."""
    request_key = str(request_key or "").strip()
    if not request_key:
        raise ManualInputError("必须提供提交幂等键 request_key。")
    if int(company_id) == int(counterparty_company_id):
        raise ManualInputError("对方公司不能与本公司相同。")
    if direction not in ("outgoing", "incoming"):
        raise ManualInputError("方向必须是 outgoing 或 incoming。")
    if funding_source not in FUNDING_SOURCES:
        raise ManualInputError(f"资金来源必须是：{'、'.join(FUNDING_SOURCES)}。")
    if requested_subject not in SUBJECTS:
        raise ManualInputError("科目必须是应收/应付/其他应收/其他应付之一。")
    _validate_date(occurred_at, "业务日期")
    currency = str(currency or "").strip()
    if not currency:
        raise ManualInputError("币种不能为空。")
    amount = str(_parse_amount(amount))

    for label, raw in (
        ("company_id", company_id), ("counterparty_company_id", counterparty_company_id),
    ):
        row = connection.execute("SELECT id FROM companies WHERE id = ?", (int(raw),)).fetchone()
        if row is None:
            raise ManualInputError(f"{label} 指向的公司不存在。")

    bank_account_id = _resolve_account_ref(
        connection, bank_account_id, company_id, "银行账户"
    )
    mapping_id = _resolve_account_ref(
        connection, personal_transit_mapping_id, company_id, "个人过账映射"
    )
    related_row = None
    if related_source_row_id not in (None, ""):
        related_row = connection.execute(
            "SELECT r.id, b.company_id FROM source_rows r "
            "JOIN sheet_batches s ON s.id = r.sheet_batch_id "
            "JOIN import_batches b ON b.id = s.import_batch_id "
            "WHERE r.id = ?",
            (int(related_source_row_id),),
        ).fetchone()
        if related_row is None:
            raise ManualInputError("关联银行源行不存在。")

    if funding_source == "approved_bank_account" and bank_account_id is None:
        raise ManualInputError("资金来源为已批准账户时必须指定银行账户。")
    if funding_source == "personal_transit" and mapping_id is None:
        raise ManualInputError("资金来源为个人过账时必须指定个人过账映射。")

    supersedes_id = None
    if supersedes_record_id not in (None, ""):
        parent = connection.execute(
            "SELECT id, company_id FROM manual_records WHERE id = ?",
            (int(supersedes_record_id),),
        ).fetchone()
        if parent is None or parent["company_id"] != int(company_id):
            raise ManualInputError("supersedes_record_id 无效。")
        supersedes_id = int(supersedes_record_id)

    now = utc_now()
    began = False
    if not connection.in_transaction:
        connection.execute("BEGIN IMMEDIATE")
        began = True
    try:
        # Re-check inside the write transaction: concurrent identical submits
        # serialize here, so a replay is found before any INSERT.
        existing = connection.execute(
            "SELECT id FROM manual_records WHERE company_id = ? AND request_key = ?",
            (int(company_id), request_key),
        ).fetchone()
        if existing is not None:
            if began:
                connection.commit()
            return _record_payload(connection, existing["id"], idempotent_replay=True)
        try:
            cursor = connection.execute(
                """
                INSERT INTO manual_records (
                    company_id, counterparty_company_id, occurred_at, direction,
                    amount, amount_scale, currency, funding_source, bank_account_id,
                    personal_transit_mapping_id, related_source_row_id,
                    requested_subject, summary, reason, evidence_json, request_key,
                    supersedes_record_id, submitted_by, created_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    int(company_id), int(counterparty_company_id), occurred_at, direction,
                    amount, _scale_of(amount), currency, funding_source, bank_account_id,
                    mapping_id, related_row["id"] if related_row is not None else None,
                    requested_subject, str(summary or "") or None,
                    str(reason or "") or None,
                    json.dumps(evidence, ensure_ascii=False) if evidence else None,
                    request_key, supersedes_id,
                    actor["id"], now,
                ),
            )
        except sqlite3.IntegrityError:
            # A concurrent identical submit won the race and committed first;
            # surface the existing record idempotently instead of a UNIQUE 500.
            if began:
                connection.rollback()
            existing = connection.execute(
                "SELECT id FROM manual_records WHERE company_id = ? AND request_key = ?",
                (int(company_id), request_key),
            ).fetchone()
            if existing is not None:
                return _record_payload(connection, existing["id"], idempotent_replay=True)
            raise
        record_id = int(cursor.lastrowid)
        _append_decision(
            connection, record_id, state="pending", action="submit",
            reason=str(reason or "") or None, actor=actor,
        )
    except Exception:
        if began:
            connection.rollback()
        raise
    else:
        if began:
            connection.commit()
        return _record_payload(connection, record_id)


def _resolve_account_ref(connection, raw, company_id: int, label: str) -> int | None:
    if raw in (None, ""):
        return None
    row = connection.execute(
        "SELECT id, company_id FROM bank_accounts WHERE id = ?", (int(raw),)
    ).fetchone()
    if row is None:
        raise ManualInputError(f"{label}不存在。")
    if row["company_id"] != int(company_id):
        raise ManualInputError(f"{label}必须属于提交公司。")
    return int(raw)


def _scale_of(amount: str) -> int:
    exponent = Decimal(amount).as_tuple().exponent
    return max(0, -int(exponent))


# ---------------------------------------------------------------------------
# Decisions
# ---------------------------------------------------------------------------


def _append_decision(
    connection: sqlite3.Connection,
    record_id: int,
    *,
    state: str,
    action: str,
    reason: str | None,
    actor: sqlite3.Row,
    idempotency_key: str | None = None,
    supersedes_decision_id: int | None = None,
) -> int:
    row = connection.execute(
        "SELECT COALESCE(MAX(revision), 0) AS m FROM manual_record_decisions WHERE record_id = ?",
        (record_id,),
    ).fetchone()
    revision = int(row["m"]) + 1
    now = utc_now()
    cursor = connection.execute(
        """
        INSERT INTO manual_record_decisions (
            record_id, revision, state, action, reason, actor_user_id,
            actor_username, idempotency_key, supersedes_decision_id, created_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            record_id, revision, state, action, reason,
            actor["id"], actor["username"], idempotency_key,
            supersedes_decision_id, now,
        ),
    )
    decision_id = int(cursor.lastrowid)
    connection.execute(
        """
        INSERT OR REPLACE INTO current_manual_record_decisions (record_id, decision_id)
        VALUES (?, ?)
        """,
        (record_id, decision_id),
    )
    return decision_id


def _current_decision(connection: sqlite3.Connection, record_id: int) -> sqlite3.Row | None:
    return connection.execute(
        """
        SELECT d.* FROM current_manual_record_decisions c
        JOIN manual_record_decisions d ON d.id = c.decision_id
        WHERE c.record_id = ?
        """,
        (record_id,),
    ).fetchone()


def decide(
    connection: sqlite3.Connection,
    record_id: int,
    action: str,
    *,
    reason: str,
    expected_decision_id: int | None,
    request_key: str | None,
    actor: sqlite3.Row,
    subject_code: object = None,
    target_ledger_event_id: object = None,
    effective_at: object = None,
) -> dict[str, object]:
    """Apply an administrator decision to a manual record.

    ``approve_new`` creates a confirmed ledger event; ``approve_link`` joins an
    existing ledger event without adding a second economic impact; ``return``
    and ``exception`` never produce a balance; ``reverse`` creates an opposite
    reversal event (or detaches a linked claim) with an independent business
    effective date — explicit ``effective_at`` or the approval business day.
    Replays return the earlier outcome via ``idempotency_key``.
    """
    reason = (reason or "").strip()
    if not reason:
        raise ManualInputError("必须填写审核原因。")
    if action not in ("approve_new", "approve_link", "return", "exception", "reverse"):
        raise ManualInputError("未知的审核决定类型。")

    began = False
    if not connection.in_transaction:
        connection.execute("BEGIN IMMEDIATE")
        began = True
    try:
        record = connection.execute(
            "SELECT * FROM manual_records WHERE id = ?", (record_id,)
        ).fetchone()
        if record is None:
            raise ManualConflictError("手工记录不存在。")
        if request_key:
            existing = connection.execute(
                "SELECT * FROM manual_record_decisions WHERE record_id = ? AND idempotency_key = ?",
                (record_id, request_key),
            ).fetchone()
            if existing is not None:
                if began:
                    connection.commit()
                return _decision_payload(connection, record_id, existing["id"])

        current = _current_decision(connection, record_id)
        if current is None:
            raise ManualConflictError("该记录没有当前状态。")
        if expected_decision_id is not None and int(expected_decision_id) != current["id"]:
            raise ManualConflictError("记录已发生变更，请刷新后重试。")

        if action == "approve_new":
            outcome = _approve_new(
                connection, record, current, actor, subject_code, reason, request_key
            )
        elif action == "approve_link":
            outcome = _approve_link(
                connection, record, current, actor, target_ledger_event_id, reason,
                request_key,
            )
        elif action == "return":
            if current["state"] != "pending":
                raise ManualConflictError("只有待复核的记录可以退回。")
            decision_id = _append_decision(
                connection, record_id, state="returned", action=action,
                reason=reason, actor=actor, idempotency_key=request_key,
                supersedes_decision_id=current["id"],
            )
            outcome = {"decision_id": decision_id, "ledger_event_id": None}
        elif action == "exception":
            if current["state"] != "pending":
                raise ManualConflictError("只有待复核的记录可以转为异常。")
            decision_id = _append_decision(
                connection, record_id, state="exception", action=action,
                reason=reason, actor=actor, idempotency_key=request_key,
                supersedes_decision_id=current["id"],
            )
            outcome = {"decision_id": decision_id, "ledger_event_id": None}
        else:  # reverse
            if current["state"] != "approved":
                raise ManualConflictError("只有已批准记录可以冲销。")
            outcome = _reverse(
                connection, record, current, actor, request_key, reason,
                effective_at=effective_at,
            )

        _store_audit(connection, record, current, action, outcome, reason, actor)
    except Exception:
        if began:
            connection.rollback()
        raise
    else:
        if began:
            connection.commit()
        return _decision_payload(connection, record_id, outcome["decision_id"])


def _approve_new(
    connection: sqlite3.Connection,
    record: sqlite3.Row,
    current: sqlite3.Row,
    actor: sqlite3.Row,
    subject_code: object,
    reason: str,
    idempotency_key: str | None,
) -> dict[str, object]:
    subject = str(subject_code or record["requested_subject"] or "")
    if subject not in SUBJECTS:
        raise ManualInputError("科目必须是应收/应付/其他应收/其他应付之一。")
    if record["direction"] == "outgoing":
        payer, payee = record["company_id"], record["counterparty_company_id"]
    else:
        payer, payee = record["counterparty_company_id"], record["company_id"]
    event_id, _revision_id = create_event(
        connection,
        state="confirmed",
        effective_at=record["occurred_at"],
        amount=record["amount"],
        currency=record["currency"],
        payer_company_id=payer,
        payee_company_id=payee,
        perspective_company_id=record["company_id"],
        subject_code=subject,
        source_kind="manual",
        source_revision_token=None,
        posting_kind="normal",
        rule_version="manual-record-v1",
        evidence_json=json.dumps({"manual_record_id": record["id"]}, ensure_ascii=False),
        actor=actor,
        reason=reason,
    )
    decision_id = _append_decision(
        connection, record["id"], state="approved", action="approve_new",
        reason=reason, actor=actor,
        idempotency_key=idempotency_key,
        supersedes_decision_id=current["id"],
    )
    connection.execute(
        """
        INSERT INTO ledger_event_manual_sources (manual_record_id, ledger_event_id)
        VALUES (?, ?)
        """,
        (record["id"], event_id),
    )
    return {"decision_id": decision_id, "ledger_event_id": event_id}


def _approve_link(
    connection: sqlite3.Connection,
    record: sqlite3.Row,
    current: sqlite3.Row,
    actor: sqlite3.Row,
    target_ledger_event_id: object,
    reason: str,
    idempotency_key: str | None,
) -> dict[str, object]:
    if target_ledger_event_id in (None, ""):
        raise ManualInputError("approve_link 必须指定目标往来事件。")
    target = connection.execute(
        "SELECT id, lifecycle FROM ledger_events WHERE id = ?",
        (int(target_ledger_event_id),),
    ).fetchone()
    if target is None or target["lifecycle"] != "active":
        raise ManualConflictError("目标往来事件不存在。")
    if manual_source_claim(connection, record["id"]) is not None:
        raise ManualConflictError("该手工记录已关联往来事件。")
    decision_id = _append_decision(
        connection, record["id"], state="approved", action="approve_link",
        reason=reason, actor=actor,
        idempotency_key=idempotency_key,
        supersedes_decision_id=current["id"],
    )
    connection.execute(
        """
        INSERT INTO ledger_event_manual_sources (manual_record_id, ledger_event_id)
        VALUES (?, ?)
        """,
        (record["id"], int(target_ledger_event_id)),
    )
    return {"decision_id": decision_id, "ledger_event_id": int(target_ledger_event_id)}


def _reverse(
    connection: sqlite3.Connection,
    record: sqlite3.Row,
    current: sqlite3.Row,
    actor: sqlite3.Row,
    request_key: str | None,
    reason: str,
    effective_at: object = None,
) -> dict[str, object]:
    claim = manual_source_claim(connection, record["id"])
    if claim is None:
        raise ManualConflictError("该记录尚未关联往来事件，无法冲销。")
    event_id = claim["ledger_event_id"]
    revision = current_revision(connection, event_id)
    if revision is None:
        raise ManualConflictError("关联的往来事件没有当前修订。")

    if _is_manual_creation_source(revision, record["id"]):
        # This record created the event (approve_new): it added the economic
        # impact, so reversing it must always produce an equal-amount reversal
        # event — even when other manual evidence was later linked onto the
        # same event. The original impact must not survive in balances.
        if effective_at is not None and str(effective_at).strip():
            effective_at = _validate_date(effective_at, "冲销生效日")
        else:
            effective_at = _business_today()
        create_reversal(
            connection,
            event_id,
            source_kind="manual",
            source_revision_token=str(record["id"]),
            effective_at=effective_at,
            reason=reason,
            actor=actor,
            idempotency_key=request_key,
            rule_version="manual-record-v1",
        )
    else:
        # The manual was linked evidence (approve_link) on an event it never
        # created: it added no second impact, so reversing detaches the claim
        # and the underlying economic impact stays.
        connection.execute(
            "DELETE FROM ledger_event_manual_sources WHERE manual_record_id = ?",
            (record["id"],),
        )
    decision_id = _append_decision(
        connection, record["id"], state="reversed", action="reverse",
        reason=reason, actor=actor, idempotency_key=request_key,
        supersedes_decision_id=current["id"],
    )
    return {"decision_id": decision_id, "ledger_event_id": None}


def _is_manual_creation_source(revision: sqlite3.Row, manual_record_id: int) -> bool:
    """True when ``manual_record_id`` created the event via ``approve_new``.

    The creation source is recorded in the event's immutable revision
    ``evidence_json``; a linked evidence record is never the creation source
    and carries no second economic impact.
    """
    if revision["source_kind"] != "manual":
        return False
    try:
        evidence = json.loads(revision["evidence_json"] or "{}")
    except (TypeError, ValueError):
        return False
    return evidence.get("manual_record_id") == manual_record_id


def _store_audit(
    connection, record, current, action, outcome, reason, actor
) -> None:
    from .auth import audit

    audit(
        connection,
        f"manual_{action}",
        actor=actor,
        target=f"manual_record:{record['id']}",
        detail=(
            f"decision:{outcome['decision_id']};"
            f"ledger_event:{outcome.get('ledger_event_id')};reason:{reason}"
        ),
    )


# ---------------------------------------------------------------------------
# Candidates and queries
# ---------------------------------------------------------------------------


def find_candidates(connection: sqlite3.Connection, record_id: int) -> list[dict[str, object]]:
    """Deterministic hints shown before approval; never auto-merged."""
    record = connection.execute(
        "SELECT * FROM manual_records WHERE id = ?", (record_id,)
    ).fetchone()
    if record is None:
        return []
    wanted_direction = "incoming" if record["direction"] == "outgoing" else "outgoing"
    date_prefix = str(record["occurred_at"])[:10]
    candidates: list[dict[str, object]] = []

    bank_rows = connection.execute(
        """
        SELECT e.event_id, e.amount, e.currency, e.effective_at, e.pairing,
               e.payer_company_id, e.payee_company_id, e.decision_id
        FROM eligible_intercompany_events e
        WHERE (e.payer_company_id = ? AND e.payee_company_id = ?)
           OR (e.payer_company_id = ? AND e.payee_company_id = ?)
        ORDER BY e.event_id
        """,
        (
            record["company_id"], record["counterparty_company_id"],
            record["counterparty_company_id"], record["company_id"],
        ),
    ).fetchall()
    for row in bank_rows:
        if row["amount"] != record["amount"] or row["currency"] != record["currency"]:
            continue
        event_direction = (
            "outgoing" if row["payer_company_id"] == record["company_id"] else "incoming"
        )
        if event_direction != wanted_direction:
            continue
        candidates.append(
            {
                "kind": "bank_event",
                "ledger_event_id": _ledger_event_of_bank(connection, row["event_id"]),
                "bank_event_id": row["event_id"],
                "decision_id": row["decision_id"],
                "amount": row["amount"],
                "currency": row["currency"],
                "effective_at": row["effective_at"],
                "pairing": row["pairing"],
                "hint": "已存在匹配的银行规范事件，建议关联",
            }
        )

    manual_rows = connection.execute(
        """
        SELECT m.id, m.company_id, m.counterparty_company_id, m.direction,
               m.amount, m.currency, m.occurred_at, d.state
        FROM manual_records m
        JOIN current_manual_record_decisions c ON c.record_id = m.id
        JOIN manual_record_decisions d ON d.id = c.decision_id
        WHERE m.id != ? AND m.amount = ? AND m.currency = ?
          AND (
            (m.company_id = ? AND m.counterparty_company_id = ?)
            OR (m.company_id = ? AND m.counterparty_company_id = ?)
          )
        ORDER BY m.id
        """,
        (
            record["id"], record["amount"], record["currency"],
            record["company_id"], record["counterparty_company_id"],
            record["counterparty_company_id"], record["company_id"],
        ),
    ).fetchall()
    for row in manual_rows:
        if row["direction"] != wanted_direction:
            continue
        if row["state"] not in ("approved", "pending"):
            continue
        candidates.append(
            {
                "kind": "manual_record",
                "ledger_event_id": None,
                "manual_record_id": row["id"],
                "amount": row["amount"],
                "currency": row["currency"],
                "occurred_at": row["occurred_at"],
                "state": row["state"],
                "hint": "存在方向相反的同额手工记录，建议核对后关联",
            }
        )
    return candidates


def _ledger_event_of_bank(connection, bank_event_id: int) -> int | None:
    claim = connection.execute(
        "SELECT ledger_event_id FROM ledger_event_bank_sources WHERE bank_event_id = ?",
        (bank_event_id,),
    ).fetchone()
    return claim["ledger_event_id"] if claim is not None else None


def list_records(
    connection: sqlite3.Connection,
    *,
    company_id: int | None = None,
    state: str | None = None,
    limit: int = 100,
) -> list[sqlite3.Row]:
    conditions: list[str] = []
    params: list[object] = []
    if company_id is not None:
        conditions.append("(m.company_id = ? OR m.counterparty_company_id = ?)")
        params.extend([company_id, company_id])
    if state is not None:
        if state not in MANUAL_STATES:
            raise ManualInputError("无效的记录状态。")
        conditions.append("d.state = ?")
        params.append(state)
    where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
    return connection.execute(
        f"""
        SELECT m.*, d.id AS decision_id, d.state AS state, d.revision AS decision_revision,
               d.action AS action, d.reason AS decision_reason,
               d.actor_username AS decision_actor, d.created_at AS decision_at,
               c.name AS company_name, cc.name AS counterparty_company_name,
               u.username AS submitted_by_username
        FROM manual_records m
        JOIN current_manual_record_decisions c ON c.record_id = m.id
        JOIN manual_record_decisions d ON d.id = c.decision_id
        LEFT JOIN companies c ON c.id = m.company_id
        LEFT JOIN companies cc ON cc.id = m.counterparty_company_id
        LEFT JOIN users u ON u.id = m.submitted_by
        {where}
        ORDER BY m.id DESC
        LIMIT ?
        """,
        (*params, max(1, int(limit))),
    ).fetchall()


def rebuild_current_manual_projection(connection: sqlite3.Connection) -> int:
    began = False
    if not connection.in_transaction:
        connection.execute("BEGIN IMMEDIATE")
        began = True
    try:
        connection.execute("DELETE FROM current_manual_record_decisions")
        rows = connection.execute(
            """
            SELECT m.id AS record_id,
                   (SELECT d2.id FROM manual_record_decisions d2
                     WHERE d2.record_id = m.id
                     ORDER BY d2.revision DESC LIMIT 1) AS latest_id
            FROM manual_records m
            """
        ).fetchall()
        rebuilt = 0
        for row in rows:
            if row["latest_id"] is None:
                continue
            connection.execute(
                """
                INSERT OR REPLACE INTO current_manual_record_decisions (record_id, decision_id)
                VALUES (?, ?)
                """,
                (row["record_id"], row["latest_id"]),
            )
            rebuilt += 1
    except Exception:
        if began:
            connection.rollback()
        raise
    else:
        if began:
            connection.commit()
        return rebuilt


# ---------------------------------------------------------------------------
# Payloads
# ---------------------------------------------------------------------------


def _record_payload(connection: sqlite3.Connection, record_id: int, *, idempotent_replay: bool = False) -> dict[str, object]:
    rows = list_records(connection, limit=1000)
    row = next((item for item in rows if item["id"] == record_id), None)
    if row is None:
        raise ManualInputError("手工记录不存在。")
    payload = _row_payload(connection, row)
    if idempotent_replay:
        payload["idempotent_replay"] = True
    return payload


def _row_payload(connection: sqlite3.Connection, row: sqlite3.Row) -> dict[str, object]:
    evidence = json.loads(row["evidence_json"] or "{}") if row["evidence_json"] else {}
    return {
        "id": row["id"],
        "company_id": row["company_id"],
        "company_name": row["company_name"],
        "counterparty_company_id": row["counterparty_company_id"],
        "counterparty_company_name": row["counterparty_company_name"],
        "occurred_at": row["occurred_at"],
        "direction": row["direction"],
        "amount": row["amount"],
        "currency": row["currency"],
        "funding_source": row["funding_source"],
        "bank_account_id": row["bank_account_id"],
        "personal_transit_mapping_id": row["personal_transit_mapping_id"],
        "related_source_row_id": row["related_source_row_id"],
        "requested_subject": row["requested_subject"],
        "summary": row["summary"],
        "reason": row["reason"],
        "request_key": row["request_key"],
        "supersedes_record_id": row["supersedes_record_id"],
        "submitted_by": row["submitted_by"],
        "submitted_by_username": row["submitted_by_username"] if "submitted_by_username" in row.keys() else None,
        "attachment_name": evidence.get("attachment_name"),
        "created_at": row["created_at"],
        "state": row["state"],
        "decision_id": row["decision_id"],
        "decision_revision": row["decision_revision"],
        "decision_action": row["action"],
        "decision_reason": row["decision_reason"],
        "decision_actor": row["decision_actor"],
        "decision_at": row["decision_at"],
        "candidates": find_candidates(connection, row["id"]),
    }


def _decision_payload(connection: sqlite3.Connection, record_id: int, decision_id: int) -> dict[str, object]:
    row = connection.execute(
        """
        SELECT d.* FROM manual_record_decisions d
        WHERE d.id = ?
        """,
        (decision_id,),
    ).fetchone()
    record = connection.execute(
        "SELECT * FROM manual_records WHERE id = ?", (record_id,)
    ).fetchone()
    claim = manual_source_claim(connection, record_id)
    return {
        "record_id": record_id,
        "decision_id": decision_id,
        "revision": row["revision"],
        "state": row["state"],
        "action": row["action"],
        "reason": row["reason"],
        "actor_username": row["actor_username"],
        "created_at": row["created_at"],
        "ledger_event_id": claim["ledger_event_id"] if claim is not None else None,
        "requested_subject": record["requested_subject"],
        "amount": record["amount"],
        "currency": record["currency"],
        "occurred_at": record["occurred_at"],
    }