"""Executable proof for the Cloud inventory reservation primitive.

The production implementation uses SQLite/D1. These tests intentionally exercise
the same single-statement INSERT...SELECT...WHERE shape rather than a read-then-
write approximation, so two concurrent reservations cannot oversell stock.
"""

from __future__ import annotations

import sqlite3
import tempfile
import threading
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
MIGRATION = ROOT.parent / "P2HSCloud" / "migrations" / "0037_cloud_inventory_reservations.sql"


def _connection(path: str) -> sqlite3.Connection:
    db = sqlite3.connect(path, timeout=10, isolation_level=None, check_same_thread=False)
    db.execute("PRAGMA journal_mode=WAL")
    return db


def _reserve(db: sqlite3.Connection, reservation_id: str, key: str, quantity: float) -> bool:
    try:
        row = db.execute(
            """
            INSERT INTO cloud_inventory_reservations (
              id, workspace_id, project_id, record_id, quantity, status,
              idempotency_key, created_at, updated_at
            )
            SELECT ?, 'ws', 'project', 'record', ?, 'active', ?, datetime('now'), datetime('now')
            WHERE (
              SELECT CAST(json_extract(data_json, '$.quantity') AS REAL)
              FROM cloud_records
              WHERE id = 'record' AND project_id = 'project'
                AND entity_key = 'inventory_record' AND deleted_at IS NULL
            ) - COALESCE((
              SELECT SUM(quantity) FROM cloud_inventory_reservations
              WHERE record_id = 'record' AND project_id = 'project' AND status = 'active'
            ), 0) >= ?
            """,
            (reservation_id, quantity, key, quantity),
        )
        return row.rowcount == 1
    except sqlite3.IntegrityError:
        return False


def _database(tmp_path: Path) -> sqlite3.Connection:
    db = _connection(str(tmp_path / "inventory.sqlite"))
    db.executescript(
        """
        CREATE TABLE cloud_records (
          id TEXT PRIMARY KEY, project_id TEXT NOT NULL, entity_key TEXT NOT NULL,
          data_json TEXT NOT NULL, deleted_at TEXT
        );
        CREATE TABLE cloud_inventory_reservations (
          id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, project_id TEXT NOT NULL,
          record_id TEXT NOT NULL, quantity REAL NOT NULL, status TEXT NOT NULL,
          idempotency_key TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL
        );
        CREATE UNIQUE INDEX uq_inventory_reservation_project_idempotency
          ON cloud_inventory_reservations(project_id, idempotency_key)
          WHERE idempotency_key IS NOT NULL;
        INSERT INTO cloud_records VALUES
          ('record', 'project', 'inventory_record', '{"quantity": 10}', NULL);
        """
    )
    return db


def test_inventory_reservation_is_atomic_under_concurrency() -> None:
    with tempfile.TemporaryDirectory(dir=ROOT / ".pytest-tmp") as raw:
        tmp_path = Path(raw)
        seed = _database(tmp_path)
        seed.close()
        results: list[bool] = []
        lock = threading.Lock()

        def attempt(index: int) -> None:
            db = _connection(str(tmp_path / "inventory.sqlite"))
            result = _reserve(db, f"reservation-{index}", f"key-{index}", 6)
            with lock:
                results.append(result)
            db.close()

        threads = [threading.Thread(target=attempt, args=(index,)) for index in range(2)]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()

        assert sorted(results) == [False, True]
        db = _connection(str(tmp_path / "inventory.sqlite"))
        assert db.execute(
            "SELECT COALESCE(SUM(quantity), 0) FROM cloud_inventory_reservations WHERE status = 'active'"
        ).fetchone()[0] == 6
        db.close()


def test_inventory_reservation_idempotency_and_release() -> None:
    with tempfile.TemporaryDirectory(dir=ROOT / ".pytest-tmp") as raw:
        db = _database(Path(raw))
        assert _reserve(db, "reservation-1", "same-request", 6)
        assert not _reserve(db, "reservation-2", "same-request", 1)
        db.execute(
            "UPDATE cloud_inventory_reservations SET status = 'released' WHERE id = 'reservation-1'"
        )
        assert _reserve(db, "reservation-3", "new-request", 6)
        db.close()


def test_production_migration_contains_partial_idempotency_index() -> None:
    migration = MIGRATION.read_text(encoding="utf-8")
    assert "cloud_inventory_reservations" in migration
    assert "UNIQUE INDEX" in migration
    assert "WHERE idempotency_key IS NOT NULL" in migration