`services/claims/sqlite_claim_repo.py`

134 lines
"""Concrete SQLite ClaimRepository — makes the claim subsystem runnable end-to-end.

Implements the ClaimRepository protocol from claim_workflow against the schema's
claim_case / claim_event / outbox_event tables, sharing the same connection as
SqliteRepository (one system-of-record DB). Production is the PostgreSQL equivalent with
real SELECT ... FOR UPDATE row locks; here the prototype is single-threaded so the
optimistic version field + the UNIQUE(claim_id, idempotency_key) constraint carry integrity.

Crucially the DB — not a mock — enforces idempotency: a duplicate (claim_id, idempotency_key)
raises sqlite3.IntegrityError, so the C3 fix is proven against real constraints.
"""
from __future__ import annotations

import json
import sqlite3
import uuid
from datetime import datetime, timezone

from services.claims.claim_workflow import Claim, ClaimStatus


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


class SqliteClaimRepository:
    def __init__(self, conn: sqlite3.Connection) -> None:
        self.conn = conn
        self.conn.row_factory = sqlite3.Row

    def create_claim(self, jurisdiction: str, public_property_reference: str,
                     claimant_id: uuid.UUID) -> Claim:
        claim = Claim(
            claim_id=uuid.uuid4(), jurisdiction=jurisdiction,
            public_property_reference=public_property_reference,
            claimant_id=claimant_id, status=ClaimStatus.DRAFT, version=1,
        )
        self.conn.execute(
            """INSERT INTO claim_case
               (claim_id, jurisdiction_id, public_property_reference, claimant_id,
                status, version, state_case_id)
               VALUES (?,?,?,?,?,?,?)""",
            (str(claim.claim_id), claim.jurisdiction, claim.public_property_reference,
             str(claim.claimant_id), claim.status.value, claim.version, None),
        )
        self.conn.commit()
        return claim

    # --- ClaimRepository protocol -----------------------------------------
    def get_for_update(self, claim_id: uuid.UUID) -> Claim:
        row = self.conn.execute(
            "SELECT * FROM claim_case WHERE claim_id=?", (str(claim_id),)
        ).fetchone()
        if row is None:
            raise KeyError(f"claim {claim_id} not found")
        return Claim(
            claim_id=uuid.UUID(row["claim_id"]),
            jurisdiction=row["jurisdiction_id"],
            public_property_reference=row["public_property_reference"],
            claimant_id=uuid.UUID(row["claimant_id"]),
            status=ClaimStatus(row["status"]),
            version=row["version"],
            state_case_id=row["state_case_id"],
        )

    def save(self, claim: Claim) -> None:
        self.conn.execute(
            "UPDATE claim_case SET status=?, version=?, state_case_id=? WHERE claim_id=?",
            (claim.status.value, claim.version, claim.state_case_id, str(claim.claim_id)),
        )
        self.conn.commit()

    def append_event(self, claim_id: uuid.UUID, event_type: str, payload: dict,
                     idempotency_key: str) -> None:
        # UNIQUE(claim_id, idempotency_key) is enforced by the DB — a collision raises
        # sqlite3.IntegrityError (the real C3 guard, not a mock's imitation).
        self.conn.execute(
            """INSERT INTO claim_event
               (claim_event_id, claim_id, event_type, payload, idempotency_key, created_at)
               VALUES (?,?,?,?,?,?)""",
            (str(uuid.uuid4()), str(claim_id), event_type, json.dumps(payload),
             idempotency_key, _now()),
        )
        self.conn.commit()

    def add_outbox_event(self, event_type: str, aggregate_id: uuid.UUID, payload: dict) -> None:
        self.conn.execute(
            """INSERT INTO outbox_event
               (outbox_event_id, aggregate_id, event_type, payload, delivery_state)
               VALUES (?,?,?,?, 'pending')""",
            (str(uuid.uuid4()), str(aggregate_id), event_type, json.dumps(payload)),
        )
        self.conn.commit()

    # --- reads for tests / workers ----------------------------------------
    def events(self, claim_id: uuid.UUID) -> list[dict]:
        rows = self.conn.execute(
            "SELECT event_type, idempotency_key FROM claim_event WHERE claim_id=? ORDER BY created_at",
            (str(claim_id),),
        ).fetchall()
        return [dict(r) for r in rows]

    def pending_outbox(self, aggregate_id: uuid.UUID) -> list[dict]:
        rows = self.conn.execute(
            "SELECT event_type, delivery_state FROM outbox_event WHERE aggregate_id=?",
            (str(aggregate_id),),
        ).fetchall()
        return [dict(r) for r in rows]

    # --- outbox worker surface --------------------------------------------
    def fetch_pending_outbox(self, limit: int = 100) -> list[dict]:
        rows = self.conn.execute(
            """SELECT outbox_event_id, aggregate_id, event_type, payload, attempts
               FROM outbox_event WHERE delivery_state='pending'
               ORDER BY rowid LIMIT ?""",
            (limit,),
        ).fetchall()
        return [dict(r) for r in rows]

    def mark_outbox(self, outbox_event_id: str, delivery_state: str,
                    attempts: int, last_error: str | None = None) -> None:
        self.conn.execute(
            "UPDATE outbox_event SET delivery_state=?, attempts=?, last_error=? WHERE outbox_event_id=?",
            (delivery_state, attempts, last_error, outbox_event_id),
        )
        self.conn.commit()

    def outbox_row(self, outbox_event_id: str) -> dict | None:
        r = self.conn.execute(
            "SELECT delivery_state, attempts, last_error FROM outbox_event WHERE outbox_event_id=?",
            (outbox_event_id,),
        ).fetchone()
        return dict(r) if r else None