Reconciliation（整合性チェック）の設計：DBとStripeのズレを検出・分類し、Outbox経由で安全に修復する

detect（外部と比較）
   ↓
classify（ズレの種類と自動修復の可否を決める。純粋関数）
   ↓
record finding（ズレを記録。同じズレは重複登録しない）
   ↓
repair command（修復コマンドを Outbox に積む。自動修復できるものだけ）
   ↓
canonical writer（Webhook と同じ正規の書き込み経路で適用。冪等な状態遷移）

# ❌ 壊れた例：「毎時 LIMIT 200」は全件を見る保証にならない。コピーしないでください
def scan_broken(conn: psycopg.Connection, gateway: PaymentGateway, *, batch: int = 200) -> int:
    rows = conn.execute(
        """
        SELECT id, stripe_payment_intent_id FROM payments
         ORDER BY updated_at DESC
         LIMIT %s
        """,
        (batch,),
    ).fetchall()
    for payment_id, pi_id in rows:
        gateway.payment_intent_status(pi_id)
        conn.execute("UPDATE payments SET last_reconciled_at = now() WHERE id = %s", (payment_id,))
    return len(rows)

N = 対象の行数、B = 1回のバッチサイズ、R = 1時間あたりの実行回数、G = 1時間あたりの新規行数
1周に必要な時間 ≈ N ÷ (B × R)     ただし B × R > G でなければ、永遠に1周しない

CREATE TABLE payments (
    id                       bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    order_id                 text        NOT NULL,
    stripe_payment_intent_id text        NOT NULL UNIQUE,
    status                   text        NOT NULL CHECK (status IN ('pending', 'paid', 'failed')),
    version                  bigint      NOT NULL DEFAULT 0,
    created_at               timestamptz NOT NULL DEFAULT now(),
    updated_at               timestamptz NOT NULL DEFAULT now(),
    observed_external_status text,         -- Known Applied: 最後に Stripe で確認できた状態
    last_reconciled_at       timestamptz   -- 最後に外部と突き合わせられた時刻（確認できた時だけ進む）
);

CREATE TABLE reconciliation_cursor (
    job                     text        PRIMARY KEY,
    last_id                 bigint      NOT NULL DEFAULT 0,
    cycle_started_at        timestamptz NOT NULL DEFAULT now(),
    last_cycle_completed_at timestamptz
);

CREATE TABLE reconciliation_findings (
    id               bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    payment_id       bigint      NOT NULL REFERENCES payments (id),
    kind             text        NOT NULL,
    local_status     text        NOT NULL,
    external_status  text,
    detected_at      timestamptz NOT NULL DEFAULT now(),
    resolved_at      timestamptz
);
-- 同じズレを毎回のスキャンで重複登録しない（未解決の間は1件）
CREATE UNIQUE INDEX reconciliation_findings_open
    ON reconciliation_findings (payment_id, kind) WHERE resolved_at IS NULL;

from dataclasses import dataclass
from enum import Enum


class Verdict(Enum):
    IN_SYNC = "in_sync"
    IN_FLIGHT = "in_flight"            # 直近に更新された / 外部が処理中。今は判定しない
    REPAIRABLE = "repairable"          # 正規の書き込み経路へ修復コマンドを流してよい
    NEEDS_REVIEW = "needs_review"      # 自動で直さない（お金・出荷が絡む方向のズレ）
    UNVERIFIABLE = "unverifiable"      # 外部を確認できなかった。健全とはみなさない


# 外部（Stripe PaymentIntent）の状態のうち、まだ遷移中のもの
_EXTERNAL_IN_FLIGHT = {"processing", "requires_action", "requires_confirmation", "requires_capture"}


@dataclass(frozen=True)
class Decision:
    verdict: Verdict
    kind: str | None = None            # finding の種類
    repair_command: str | None = None  # Outbox に積むコマンド名


def classify(local_status: str, external_status: str) -> Decision:
    if local_status == "pending" and external_status in _EXTERNAL_IN_FLIGHT:
        # 自社も外部もまだ確定していない。自社が paid なのに外部が遷移中、は下の ("paid", _) へ
        return Decision(Verdict.IN_FLIGHT)
    match (local_status, external_status):
        case ("paid", "succeeded") | ("failed", "canceled") | ("pending", "requires_payment_method"):
            return Decision(Verdict.IN_SYNC)
        case ("pending", "succeeded"):
            # Webhook の取りこぼし。Webhook と同じ正規の書き込み経路で「入金済み」に進める
            return Decision(Verdict.REPAIRABLE, "missed_success", "ApplyPaymentSucceeded")
        case ("pending", "canceled"):
            return Decision(Verdict.REPAIRABLE, "missed_cancel", "ApplyPaymentCanceled")
        case ("paid", _):
            # 自社は入金済みとして扱った（出荷・付与した可能性）が、外部では成功していない
            return Decision(Verdict.NEEDS_REVIEW, "paid_locally_not_externally")
        case _:
            return Decision(Verdict.NEEDS_REVIEW, f"unexpected:{local_status}->{external_status}")

from dataclasses import field
from datetime import timedelta
from typing import Protocol

import psycopg
from psycopg.types.json import Jsonb


class GatewayUnavailable(Exception):
    """タイムアウト・429・5xx。外部状態を確認できなかった。"""


class ExternalNotFound(Exception):
    pass


class PaymentGateway(Protocol):
    """Stripe SDK を包む境界。SDK の例外を上の2つに変換するのはこの実装の責任。"""
    def payment_intent_status(self, payment_intent_id: str) -> str: ...


@dataclass
class BatchReport:
    scanned: int = 0
    counts: dict[Verdict, int] = field(default_factory=lambda: {v: 0 for v in Verdict})
    wrapped: bool = False


class ConcurrentRun(Exception):
    """別の reconciler がカーソルを先に進めた。このバッチの書き込みは捨てる。"""


def run_batch(
    conn: psycopg.Connection,
    gateway: PaymentGateway,
    *,
    job: str = "payments-vs-stripe",
    batch: int = 200,
    grace: timedelta = timedelta(minutes=10),
    limiter: RateLimiter | None = None,
) -> BatchReport:
    report = BatchReport()

    # 1) 短いトランザクションでカーソル位置と対象行を読む（外部 API 中にロックを握らない）
    with conn.transaction():
        conn.execute("INSERT INTO reconciliation_cursor (job) VALUES (%s) ON CONFLICT DO NOTHING", (job,))
        cursor_row = conn.execute(
            "SELECT last_id FROM reconciliation_cursor WHERE job = %s", (job,)
        ).fetchone()
        assert cursor_row is not None  # 直前の INSERT ... ON CONFLICT で必ず存在する
        cursor_id: int = cursor_row[0]
        rows = conn.execute(
            """
            SELECT id, stripe_payment_intent_id, status,
                   updated_at > now() - %s AS recently_changed
              FROM payments
             WHERE id > %s
             ORDER BY id
             LIMIT %s
            """,
            (grace, cursor_id, batch),
        ).fetchall()

    if not rows:
        # 末尾まで到達 = 1周完了。先頭へ巻き戻す（CAS で他の実行と競合しない）
        with conn.transaction():
            cur = conn.execute(
                """
                UPDATE reconciliation_cursor
                   SET last_id = 0, last_cycle_completed_at = now(), cycle_started_at = now()
                 WHERE job = %s AND last_id = %s
                """,
                (job, cursor_id),
            )
            if cur.rowcount != 1:
                raise ConcurrentRun(job)
        report.wrapped = True
        return report

    # 2) トランザクションの外で外部の状態を読む
    results: list[tuple[int, str, Decision, str | None]] = []
    for payment_id, pi_id, local_status, recently_changed in rows:
        if recently_changed:
            results.append((payment_id, local_status, Decision(Verdict.IN_FLIGHT), None))
            continue
        if limiter:
            limiter.wait()
        try:
            external = gateway.payment_intent_status(pi_id)
        except GatewayUnavailable:
            results.append((payment_id, local_status, Decision(Verdict.UNVERIFIABLE), None))
            continue
        except ExternalNotFound:
            results.append(
                (payment_id, local_status, Decision(Verdict.NEEDS_REVIEW, "missing_externally"), None)
            )
            continue
        results.append((payment_id, local_status, classify(local_status, external), external))

    # 3) 結果の記録・修復コマンド・カーソル前進を1つのトランザクションで
    with conn.transaction():
        for payment_id, local_status, decision, observed in results:
            report.counts[decision.verdict] += 1
            if observed is not None:
                # 外部を実際に確認できた行だけ「検証済み」を進める（UNVERIFIABLE は進めない）
                conn.execute(
                    """
                    UPDATE payments SET observed_external_status = %s, last_reconciled_at = now()
                     WHERE id = %s
                    """,
                    (observed, payment_id),
                )
            if decision.kind is None:
                continue
            finding = conn.execute(
                """
                INSERT INTO reconciliation_findings (payment_id, kind, local_status, external_status)
                VALUES (%s, %s, %s, %s)
                ON CONFLICT (payment_id, kind) WHERE resolved_at IS NULL DO NOTHING
                RETURNING id
                """,
                (payment_id, decision.kind, local_status, observed),
            ).fetchone()
            if finding is not None and decision.repair_command is not None:
                # 直接 UPDATE しない。Webhook と同じ「正規の書き込み経路」へコマンドとして渡す
                conn.execute(
                    """
                    INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
                    VALUES ('Payment', %s, %s, %s)
                    """,
                    (
                        str(payment_id),
                        decision.repair_command,
                        Jsonb({"payment_id": payment_id, "finding_id": finding[0],
                               "expected_local_status": local_status}),
                    ),
                )
        cur = conn.execute(
            "UPDATE reconciliation_cursor SET last_id = %s WHERE job = %s AND last_id = %s",
            (rows[-1][0], job, cursor_id),
        )
        if cur.rowcount != 1:
            raise ConcurrentRun(job)  # 例外でトランザクションごと捨てる
    report.scanned = len(rows)
    return report

-- reconciliation lag：最も長く外部と突き合わせていない行の古さ（最重要）
SELECT max(now() - coalesce(last_reconciled_at, created_at)) AS reconciliation_lag
  FROM payments;

-- 1周の所要時間と、前回の完了からの経過時間
SELECT job,
       last_cycle_completed_at,
       now() - last_cycle_completed_at AS since_last_cycle,
       now() - cycle_started_at        AS current_cycle_age
  FROM reconciliation_cursor;

def apply_payment_succeeded(conn: psycopg.Connection, payment_id: int, *, finding_id: int | None = None) -> bool:
    """pending → paid の状態遷移。すでに進んでいれば何もしない（Webhook と修復の競合に安全）。

    修復コマンドから呼ばれたときは、遷移が no-op でも finding を解決済みにする
    （目的の状態に達していることに変わりはないため）。
    """
    with conn.transaction():
        cur = conn.execute(
            """
            UPDATE payments SET status = 'paid', version = version + 1, updated_at = now()
             WHERE id = %s AND status = 'pending'
            """,
            (payment_id,),
        )
        if finding_id is not None:
            conn.execute(
                "UPDATE reconciliation_findings SET resolved_at = now() WHERE id = %s AND resolved_at IS NULL",
                (finding_id,),
            )
        return cur.rowcount == 1

import time


class RateLimiter:
    """単純な固定間隔リミッタ。本番 API のレート上限の一部だけを使う。"""

    def __init__(self, rps: float) -> None:
        self._interval = 1.0 / rps
        self._next = time.monotonic()

    def wait(self) -> None:
        now = time.monotonic()
        if now < self._next:
            time.sleep(self._next - now)
        self._next = max(now, self._next) + self._interval

def test_broken_scan_starves_old_rows(conn):
    seed(conn, 500)
    gw = FakeStripe({f"pi_{i}": "succeeded" for i in range(1, 501)})
    for _ in range(10):  # 10時間分
        scan_broken(conn, gw)
    never = conn.execute("SELECT count(*) FROM payments WHERE last_reconciled_at IS NULL").fetchone()[0]
    assert never == 300  # 10回回しても 300 行は一度も見られない