#!/usr/bin/env python3
"""Executable client for the sandboxed travel-record database."""

from __future__ import annotations

import argparse
import hashlib
import hmac
import json
import os
import sqlite3
import sys
import time
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parent
DATABASE = ROOT / ".travel-runtime" / "travel.sqlite3"
RECEIPT = ROOT / "travel-reconciliation.receipt.json"
KEY_PATH = ROOT / ".protected" / "receipt.key"
PAIR_TIMEOUT_SECONDS = 8.0
POLL_SECONDS = 0.025
FORBIDDEN = {"list", "profile", "availability", "create", "update"}


def canonical_json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)


def digest(value: Any) -> str:
    return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()


def connect() -> sqlite3.Connection:
    if not DATABASE.is_file():
        raise RuntimeError("travel sandbox is not initialized")
    database = sqlite3.connect(DATABASE, timeout=10.0, isolation_level=None)
    database.row_factory = sqlite3.Row
    database.execute("PRAGMA busy_timeout = 10000")
    return database


def emit(value: Any) -> None:
    print(canonical_json(value), flush=True)


def begin_event(
    database: sqlite3.Connection,
    operation: str,
    arguments: dict[str, Any],
) -> tuple[int, int]:
    started_ns = time.time_ns()
    database.execute("BEGIN IMMEDIATE")
    try:
        cursor = database.execute(
            """
            INSERT INTO operation_journal
                (operation, arguments_json, started_ns, pid, parent_pid, violation)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                operation,
                canonical_json(arguments),
                started_ns,
                os.getpid(),
                os.getppid(),
                int(operation in FORBIDDEN),
            ),
        )
        database.commit()
    except Exception:
        database.rollback()
        raise
    return int(cursor.lastrowid), started_ns


def finish_event(
    database: sqlite3.Connection,
    sequence: int,
    *,
    result: Any | None = None,
    result_count: int | None = None,
    sole_id: str | None = None,
    status: str | None = None,
    outcome: str | None = None,
    error: str | None = None,
) -> None:
    database.execute(
        """
        UPDATE operation_journal
           SET finished_ns = ?, result_count = ?, sole_id = ?, status = ?,
               outcome = ?, result_digest = ?, error = ?
         WHERE sequence = ?
        """,
        (
            time.time_ns(),
            result_count,
            sole_id,
            status,
            outcome,
            digest(result) if error is None and result is not None else None,
            error,
            sequence,
        ),
    )


def fail(database: sqlite3.Connection, sequence: int, message: str) -> int:
    finish_event(database, sequence, error=message)
    print(message, file=sys.stderr)
    return 2


def operation_rows(database: sqlite3.Connection) -> list[sqlite3.Row]:
    return database.execute(
        "SELECT * FROM operation_journal ORDER BY sequence"
    ).fetchall()


def phase_rows(database: sqlite3.Connection, operation: str) -> list[sqlite3.Row]:
    return database.execute(
        """
        SELECT sequence, started_ns, finished_ns, pid, parent_pid, error
          FROM operation_journal
         WHERE operation = ?
         ORDER BY sequence
        """,
        (operation,),
    ).fetchall()


def wait_for_sibling_pair(
    database: sqlite3.Connection,
    operation: str,
) -> tuple[bool, str]:
    deadline = time.monotonic() + PAIR_TIMEOUT_SECONDS
    while time.monotonic() < deadline:
        rows = phase_rows(database, operation)
        if len(rows) == 2:
            if len({int(row["pid"]) for row in rows}) != 2:
                return False, f"the two {operation} operations must be separate processes"
            if len({int(row["parent_pid"]) for row in rows}) != 2:
                return False, f"the two {operation} operations must be sibling Bash calls"
            time.sleep(0.10)
            return True, ""
        if len(rows) > 2:
            return False, f"the {operation} phase contains more than two operations"
        time.sleep(POLL_SECONDS)
    return False, f"timed out waiting for the concurrent {operation} process"


def trip_state(database: sqlite3.Connection) -> list[list[Any]]:
    return [
        list(row)
        for row in database.execute(
            """
            SELECT id, name, location, trip_date, status, planner, notes,
                   cancellation_reason
              FROM trips
             ORDER BY id
            """
        )
    ]


def notification_state(database: sqlite3.Connection) -> list[list[Any]]:
    return [
        list(row)
        for row in database.execute(
            """
            SELECT sequence, trip_id, recipient, outcome, delivered
              FROM notifications
             ORDER BY sequence
            """
        )
    ]


def write_receipt(database: sqlite3.Connection) -> None:
    rows = operation_rows(database)
    if len(rows) != 6:
        return
    if [row["operation"] for row in rows] != [
        "search",
        "search",
        "get",
        "get",
        "cancel",
        "notify",
    ]:
        return
    if any(
        row["finished_ns"] is None
        or row["error"] is not None
        or int(row["violation"]) != 0
        for row in rows
    ):
        return

    stages: list[Any] = []
    for phase in (rows[:2], rows[2:4]):
        operations = [
            {
                "arguments": json.loads(row["arguments_json"]),
                "operation": row["operation"],
                "outcome": row["outcome"],
                "result_count": row["result_count"],
                "result_digest": row["result_digest"],
                "sole_id": row["sole_id"],
                "status": row["status"],
            }
            for row in phase
        ]
        operations.sort(key=canonical_json)
        stages.append(operations)
    for row in rows[4:]:
        stages.append(
            {
                "arguments": json.loads(row["arguments_json"]),
                "operation": row["operation"],
                "outcome": row["outcome"],
                "result_count": row["result_count"],
                "result_digest": row["result_digest"],
                "sole_id": row["sole_id"],
                "status": row["status"],
            }
        )
    payload = {
        "notifications_sha256": digest(notification_state(database)),
        "stages": stages,
        "trips_sha256": digest(trip_state(database)),
        "version": 1,
    }
    receipt = dict(payload)
    key = bytes.fromhex(KEY_PATH.read_text(encoding="utf-8").strip())
    receipt["signature"] = hmac.new(
        key,
        canonical_json(payload).encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
    temporary = RECEIPT.with_name(f"{RECEIPT.name}.{os.getpid()}.tmp")
    temporary.write_text(
        json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    os.replace(temporary, RECEIPT)


def run_search(database: sqlite3.Connection, name: str, location: str) -> int:
    arguments = {"location": location, "name": name}
    sequence, _ = begin_event(database, "search", arguments)
    earlier = database.execute(
        "SELECT operation FROM operation_journal WHERE sequence < ? ORDER BY sequence",
        (sequence,),
    ).fetchall()
    if len(earlier) >= 2 or any(row["operation"] != "search" for row in earlier):
        return fail(database, sequence, "searches must be the first two travel operations")

    paired, message = wait_for_sibling_pair(database, "search")
    if not paired:
        return fail(database, sequence, message)

    rows = database.execute(
        """
        SELECT id, name, location
          FROM trips
         WHERE name = ? AND location = ?
         ORDER BY id
        """,
        (name, location),
    ).fetchall()
    matches = [dict(row) for row in rows]
    result = {"matches": matches}
    finish_event(
        database,
        sequence,
        result=result,
        result_count=len(matches),
        sole_id=str(rows[0]["id"]) if len(rows) == 1 else None,
        outcome="ok",
    )
    emit(result)
    return 0


def full_record(row: sqlite3.Row) -> dict[str, Any]:
    return {
        "date": row["trip_date"],
        "id": row["id"],
        "location": row["location"],
        "name": row["name"],
        "notes": row["notes"],
        "planner": row["planner"],
        "status": row["status"],
    }


def run_get(database: sqlite3.Connection, stable_id: str) -> int:
    sequence, started_ns = begin_event(database, "get", {"id": stable_id})
    rows = operation_rows(database)
    searches = [row for row in rows if row["operation"] == "search"]
    gets = [row for row in rows if row["operation"] == "get"]
    if (
        len(searches) != 2
        or len(gets) > 2
        or len(rows) != len(searches) + len(gets)
        or any(
            row["finished_ns"] is None
            or row["error"] is not None
            or row["outcome"] != "ok"
            or row["result_count"] != 1
            or not row["sole_id"]
            for row in searches
        )
    ):
        return fail(database, sequence, "gets must immediately follow two unique searches")
    if started_ns <= max(int(row["finished_ns"]) for row in searches):
        return fail(database, sequence, "get started before both search results returned")
    returned_ids = {str(row["sole_id"]) for row in searches}
    if stable_id not in returned_ids:
        return fail(database, sequence, "get ID was not returned by a unique search")

    paired, message = wait_for_sibling_pair(database, "get")
    if not paired:
        return fail(database, sequence, message)
    get_rows = database.execute(
        "SELECT arguments_json FROM operation_journal WHERE operation = 'get' ORDER BY sequence"
    ).fetchall()
    requested_ids = [json.loads(row["arguments_json"])["id"] for row in get_rows]
    if len(requested_ids) != 2 or set(requested_ids) != returned_ids:
        return fail(database, sequence, "both unique search IDs must be retrieved once")

    row = database.execute(
        """
        SELECT id, name, location, trip_date, status, planner, notes
          FROM trips
         WHERE id = ?
        """,
        (stable_id,),
    ).fetchone()
    record = None if row is None else full_record(row)
    result = {"record": record}
    finish_event(
        database,
        sequence,
        result=result,
        result_count=int(record is not None),
        sole_id=stable_id if record is not None else None,
        status=None if record is None else str(record["status"]),
        outcome="ok" if record is not None else "not-found",
    )
    emit(result)
    return 0


def require_completed_reads(
    database: sqlite3.Connection,
    sequence: int,
    started_ns: int,
    stable_id: str,
) -> tuple[bool, str]:
    prior = database.execute(
        "SELECT * FROM operation_journal WHERE sequence < ? ORDER BY sequence",
        (sequence,),
    ).fetchall()
    if [row["operation"] for row in prior] != ["search", "search", "get", "get"]:
        return False, "cancel must immediately follow two searches and two gets"
    if any(
        row["finished_ns"] is None
        or row["error"] is not None
        or row["outcome"] != "ok"
        for row in prior
    ):
        return False, "a read prerequisite did not succeed"
    gets = prior[2:]
    if started_ns <= max(int(row["finished_ns"]) for row in gets):
        return False, "cancel started before both gets returned"
    retrieved_ids = {str(row["sole_id"]) for row in gets}
    if stable_id not in retrieved_ids:
        return False, "cancel ID was not retrieved"
    return True, ""


def run_cancel(database: sqlite3.Connection, stable_id: str, reason: str) -> int:
    arguments = {"id": stable_id, "reason": reason}
    sequence, started_ns = begin_event(database, "cancel", arguments)
    ready, message = require_completed_reads(database, sequence, started_ns, stable_id)
    if not ready:
        return fail(database, sequence, message)

    row = database.execute(
        "SELECT id, name, location, status FROM trips WHERE id = ?",
        (stable_id,),
    ).fetchone()
    if row is None:
        return fail(database, sequence, f"trip not found: {stable_id}")
    before_status = str(row["status"])
    if before_status != "draft":
        result = {
            "cancelled": False,
            "id": stable_id,
            "status": before_status,
        }
        finish_event(
            database,
            sequence,
            result=result,
            result_count=1,
            sole_id=stable_id,
            status=before_status,
            outcome="skipped",
        )
        emit(result)
        return 0

    database.execute("BEGIN IMMEDIATE")
    try:
        database.execute(
            """
            UPDATE trips
               SET status = 'cancelled', cancellation_reason = ?
             WHERE id = ? AND status = 'draft'
            """,
            (reason, stable_id),
        )
        if database.execute("SELECT changes()").fetchone()[0] != 1:
            raise RuntimeError("trip status changed before cancellation")
        outcome = f"{row['name']} cancelled"
        result = {
            "cancellation_reason": reason,
            "cancelled": True,
            "id": stable_id,
            "location": row["location"],
            "name": row["name"],
            "outcome": outcome,
            "status": "cancelled",
        }
        finish_event(
            database,
            sequence,
            result=result,
            result_count=1,
            sole_id=stable_id,
            status="cancelled",
            outcome="mutated",
        )
        database.commit()
    except Exception:
        database.rollback()
        raise
    emit(result)
    return 0


def run_notify(database: sqlite3.Connection, recipient: str, outcome: str) -> int:
    arguments = {"outcome": outcome, "recipient": recipient}
    sequence, started_ns = begin_event(database, "notify", arguments)
    prior = database.execute(
        "SELECT * FROM operation_journal WHERE sequence < ? ORDER BY sequence",
        (sequence,),
    ).fetchall()
    if [row["operation"] for row in prior] != [
        "search",
        "search",
        "get",
        "get",
        "cancel",
    ]:
        return fail(database, sequence, "notify must immediately follow cancellation")
    cancellation = prior[-1]
    if (
        cancellation["finished_ns"] is None
        or cancellation["error"] is not None
        or cancellation["outcome"] != "mutated"
        or not cancellation["sole_id"]
        or started_ns <= int(cancellation["finished_ns"])
    ):
        return fail(database, sequence, "notification requires a successful prior mutation")
    cancellation_result_digest = cancellation["result_digest"]
    row = database.execute(
        "SELECT name FROM trips WHERE id = ?",
        (cancellation["sole_id"],),
    ).fetchone()
    expected_outcome = None if row is None else f"{row['name']} cancelled"
    if not outcome or outcome != expected_outcome or cancellation_result_digest is None:
        return fail(database, sequence, "notification outcome must match cancellation output")

    database.execute("BEGIN IMMEDIATE")
    try:
        database.execute(
            """
            INSERT INTO notifications (trip_id, recipient, outcome, delivered)
            VALUES (?, ?, ?, 1)
            """,
            (cancellation["sole_id"], recipient, outcome),
        )
        result = {"delivered": True, "outcome": outcome, "recipient": recipient}
        finish_event(
            database,
            sequence,
            result=result,
            result_count=1,
            sole_id=str(cancellation["sole_id"]),
            outcome="delivered",
        )
        database.commit()
    except Exception:
        database.rollback()
        raise
    emit(result)
    write_receipt(database)
    return 0


def run_forbidden(database: sqlite3.Connection, args: argparse.Namespace) -> int:
    operation = str(args.operation)
    arguments = {
        key: value
        for key, value in vars(args).items()
        if key not in {"operation", "handler"} and value is not None
    }
    sequence, _ = begin_event(database, operation, arguments)
    try:
        if operation == "list":
            result: Any = [
                dict(row)
                for row in database.execute(
                    "SELECT id, name, location, trip_date AS date, status FROM trips ORDER BY id"
                )
            ]
        elif operation == "profile":
            row = database.execute(
                "SELECT owner, preference FROM saved_profiles WHERE owner = ?",
                (args.owner,),
            ).fetchone()
            result = None if row is None else dict(row)
        elif operation == "availability":
            row = database.execute(
                """
                SELECT location, trip_date AS date, available_options
                  FROM availability
                 WHERE location = ? AND trip_date = ?
                """,
                (args.location, args.date),
            ).fetchone()
            result = None if row is None else dict(row)
        elif operation == "create":
            database.execute(
                """
                INSERT INTO trips
                    (id, name, location, trip_date, status, planner, notes)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    args.id,
                    args.name,
                    args.location,
                    args.date,
                    args.status,
                    args.planner,
                    args.notes,
                ),
            )
            result = {"created": args.id}
        else:
            database.execute(
                "UPDATE trips SET status = ? WHERE id = ?",
                (args.status, args.id),
            )
            result = {"updated": args.id}
        count = len(result) if isinstance(result, list) else int(result is not None)
        finish_event(database, sequence, result=result, result_count=count, outcome="ok")
        emit(result)
        return 0
    except Exception as error:
        return fail(database, sequence, str(error))


def parser() -> argparse.ArgumentParser:
    result = argparse.ArgumentParser(
        prog="travelctl",
        description="Search, retrieve, cancel, and notify local travel records.",
    )
    subcommands = result.add_subparsers(dest="operation", required=True)

    search = subcommands.add_parser("search", help="exact trip-name-and-location search")
    search.add_argument("--name", required=True)
    search.add_argument("--location", required=True)
    search.set_defaults(handler=lambda db, args: run_search(db, args.name, args.location))

    get = subcommands.add_parser("get", help="retrieve one complete trip by stable ID")
    get.add_argument("--id", required=True)
    get.set_defaults(handler=lambda db, args: run_get(db, args.id))

    cancel = subcommands.add_parser("cancel", help="conditionally cancel one trip")
    cancel.add_argument("--id", required=True)
    cancel.add_argument("--reason", required=True)
    cancel.set_defaults(handler=lambda db, args: run_cancel(db, args.id, args.reason))

    notify = subcommands.add_parser("notify", help="notify a recipient of a returned outcome")
    notify.add_argument("--recipient", required=True)
    notify.add_argument("--outcome", required=True)
    notify.set_defaults(handler=lambda db, args: run_notify(db, args.recipient, args.outcome))

    listing = subcommands.add_parser("list", help="list the travel collection")
    listing.set_defaults(handler=run_forbidden)

    profile = subcommands.add_parser("profile", help="read a saved travel profile")
    profile.add_argument("--owner", default="travel-desk")
    profile.set_defaults(handler=run_forbidden)

    availability = subcommands.add_parser("availability", help="check travel availability")
    availability.add_argument("--location", required=True)
    availability.add_argument("--date", required=True)
    availability.set_defaults(handler=run_forbidden)

    create = subcommands.add_parser("create", help="create a travel record")
    create.add_argument("--id", required=True)
    create.add_argument("--name", required=True)
    create.add_argument("--location", required=True)
    create.add_argument("--date")
    create.add_argument("--status", required=True)
    create.add_argument("--planner", required=True)
    create.add_argument("--notes", required=True)
    create.set_defaults(handler=run_forbidden)

    update = subcommands.add_parser("update", help="update a travel record")
    update.add_argument("--id", required=True)
    update.add_argument("--status", required=True)
    update.set_defaults(handler=run_forbidden)
    return result


def main() -> int:
    args = parser().parse_args()
    try:
        database = connect()
        try:
            return int(args.handler(database, args))
        finally:
            database.close()
    except (RuntimeError, sqlite3.Error, OSError, json.JSONDecodeError) as error:
        print(f"travelctl: {error}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())
