#!/usr/bin/env python3
"""Executable read-only interface to the sandboxed calendar store."""

from __future__ import annotations

import argparse
from concurrent.futures import ThreadPoolExecutor
import hashlib
import hmac
import json
import os
from pathlib import Path
import sqlite3
import sys
import threading
from typing import Callable


ROOT = Path(__file__).resolve().parent
STATE = ROOT / ".calendar"
STORE = STATE / "meetings.sqlite3"
KEY_FILE = STATE / "receipt.key"
TRACE = STATE / "audit.json"
SESSION = STATE / ".search-session"
FORBIDDEN = (
    "list",
    "profile",
    "preferences",
    "availability",
    "create",
    "update",
    "cancel",
    "notify",
)


class CalendarError(Exception):
    pass


def canonical(value: object) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()


def signing_key() -> bytes:
    return bytes.fromhex(KEY_FILE.read_text(encoding="ascii").strip())


def store_digest() -> str:
    return hashlib.sha256(STORE.read_bytes()).hexdigest()


def empty_payload() -> dict[str, object]:
    return {"version": 1, "actions": []}


def load_payload() -> dict[str, object]:
    if not TRACE.exists():
        return empty_payload()
    document = json.loads(TRACE.read_text(encoding="utf-8"))
    if set(document) != {"payload", "signature"}:
        raise CalendarError("calendar audit journal has an invalid envelope")
    payload = document["payload"]
    if not isinstance(payload, dict):
        raise CalendarError("calendar audit journal has an invalid payload")
    expected = hmac.new(signing_key(), canonical(payload), hashlib.sha256).hexdigest()
    if not hmac.compare_digest(str(document["signature"]), expected):
        raise CalendarError("calendar audit journal authentication failed")
    if payload.get("version") != 1 or not isinstance(payload.get("actions"), list):
        raise CalendarError("calendar audit journal has an unsupported format")
    return payload


def save_payload(payload: dict[str, object]) -> None:
    signature = hmac.new(signing_key(), canonical(payload), hashlib.sha256).hexdigest()
    document = {"payload": payload, "signature": signature}
    temporary = TRACE.with_suffix(".tmp")
    temporary.write_text(
        json.dumps(document, sort_keys=True, separators=(",", ":")) + "\n",
        encoding="utf-8",
    )
    temporary.replace(TRACE)


def connection() -> sqlite3.Connection:
    database = sqlite3.connect(f"file:{STORE}?mode=ro&immutable=1", uri=True)
    database.row_factory = sqlite3.Row
    database.execute("PRAGMA query_only = ON")
    return database


def search_one(request: dict[str, str]) -> dict[str, object]:
    with connection() as database:
        rows = database.execute(
            "SELECT id, name, location FROM meetings "
            "WHERE name = ? AND location = ? ORDER BY id",
            (request["name"], request["location"]),
        ).fetchall()
    return {
        "name": request["name"],
        "location": request["location"],
        "matches": [dict(row) for row in rows],
    }


def get_one(request: dict[str, str]) -> dict[str, str]:
    with connection() as database:
        row = database.execute(
            "SELECT id, name, location, date, status FROM meetings WHERE id = ?",
            (request["id"],),
        ).fetchone()
    if row is None:
        raise CalendarError(f"no meeting has stable ID {request['id']!r}")
    return dict(row)


def run_concurrently(
    requests: list[dict[str, str]],
    operation: Callable[[dict[str, str]], object],
) -> tuple[list[object], bool]:
    barrier = threading.Barrier(len(requests))

    def worker(request: dict[str, str]) -> object:
        # Releasing every worker from the same barrier proves that all branches
        # were active before any branch began its operation, without relying on
        # scheduler timing or an artificial delay.
        barrier.wait()
        return operation(request)

    with ThreadPoolExecutor(max_workers=len(requests)) as pool:
        completed = list(pool.map(worker, requests))
    return completed, len(requests) > 1


def process_fingerprint(pid: int) -> str:
    stat = Path(f"/proc/{pid}/stat").read_text(encoding="ascii")
    fields_after_name = stat.rsplit(")", 1)[1].split()
    return f"{pid}:{fields_after_name[19]}"


def terminal_action_fingerprint() -> str:
    return process_fingerprint(os.getppid())


def append_denied(operation: str, requests: object, reason: str) -> int:
    payload = load_payload()
    actions = payload["actions"]
    assert isinstance(actions, list)
    actions.append(
        {
            "sequence": len(actions) + 1,
            "operation": operation,
            "requests": requests,
            "denied": True,
            "reason": reason,
            "store_sha256": store_digest(),
        }
    )
    save_payload(payload)
    print(json.dumps({"error": reason}, sort_keys=True), file=sys.stderr)
    return 2


def execute_search(names: list[str], locations: list[str]) -> int:
    requests = [
        {"name": name, "location": location}
        for name, location in zip(names, locations)
    ]
    payload = load_payload()
    actions = payload["actions"]
    assert isinstance(actions, list)
    if len(names) != 2 or len(locations) != 2:
        return append_denied("search", requests, "search requires exactly two branches")
    if actions:
        return append_denied("search", requests, "search phase has already run")

    searches, overlap = run_concurrently(requests, search_one)
    response = {"searches": searches}
    actions.append(
        {
            "sequence": 1,
            "operation": "search",
            "requests": requests,
            "response": response,
            "worker_count": len(requests),
            "overlap_observed": overlap,
            "store_sha256": store_digest(),
        }
    )
    save_payload(payload)
    SESSION.write_text(terminal_action_fingerprint() + "\n", encoding="ascii")
    print(json.dumps(response, sort_keys=True))
    return 0


def unique_search_ids(search_action: dict[str, object]) -> list[str]:
    response = search_action.get("response")
    if not isinstance(response, dict) or not isinstance(response.get("searches"), list):
        raise CalendarError("search journal is malformed")
    stable_ids: list[str] = []
    for branch in response["searches"]:
        if not isinstance(branch, dict) or not isinstance(branch.get("matches"), list):
            raise CalendarError("search journal is malformed")
        matches = branch["matches"]
        if len(matches) == 1 and isinstance(matches[0], dict) and isinstance(matches[0].get("id"), str):
            stable_ids.append(matches[0]["id"])
    return stable_ids


def execute_get(stable_ids: list[str]) -> int:
    requests = [{"id": stable_id} for stable_id in stable_ids]
    payload = load_payload()
    actions = payload["actions"]
    assert isinstance(actions, list)
    if len(actions) != 1 or not isinstance(actions[0], dict) or actions[0].get("operation") != "search":
        return append_denied("get", requests, "get requires one completed search phase")
    resolved = unique_search_ids(actions[0])
    if not stable_ids or len(stable_ids) != len(set(stable_ids)):
        return append_denied("get", requests, "get requires distinct resolved stable IDs")
    if any(stable_id not in resolved for stable_id in stable_ids):
        return append_denied("get", requests, "get may use only uniquely resolved search IDs")
    if not SESSION.is_file():
        return append_denied("get", requests, "search action provenance is missing")

    previous_action = SESSION.read_text(encoding="ascii").strip()
    separate_action = previous_action != terminal_action_fingerprint()
    records, overlap = run_concurrently(requests, get_one)
    response = {"records": records}
    actions.append(
        {
            "sequence": 2,
            "operation": "get",
            "requests": requests,
            "response": response,
            "worker_count": len(requests),
            "overlap_observed": overlap,
            "after_search_completion": True,
            "separate_terminal_action": separate_action,
            "dependency_source": "unique-search-results",
            "store_sha256": store_digest(),
        }
    )
    save_payload(payload)
    SESSION.unlink()
    print(json.dumps(response, sort_keys=True))
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="calendarctl",
        description=(
            "Read-only access to the sandboxed calendar service. The search and get "
            "commands accept repeated branch arguments and execute those branches concurrently."
        ),
    )
    subcommands = parser.add_subparsers(dest="operation", required=True)

    search = subcommands.add_parser("search", help="run exact name-and-location branches concurrently")
    search.add_argument("--name", action="append", required=True, help="exact meeting name; repeat per branch")
    search.add_argument("--location", action="append", required=True, help="exact location; repeat per branch")

    get = subcommands.add_parser("get", help="retrieve uniquely resolved stable IDs concurrently")
    get.add_argument("--id", action="append", required=True, help="stable ID from search; repeat per branch")

    for operation in FORBIDDEN:
        blocked = subcommands.add_parser(operation, help="disabled for this read-only audit")
        blocked.add_argument("arguments", nargs="*")
    return parser


def main() -> int:
    args = build_parser().parse_args()
    try:
        if args.operation == "search":
            return execute_search(args.name, args.location)
        if args.operation == "get":
            return execute_get(args.id)
        return append_denied(args.operation, args.arguments, f"{args.operation} is forbidden in this audit")
    except (CalendarError, OSError, ValueError, KeyError, TypeError, json.JSONDecodeError, sqlite3.Error) as error:
        print(json.dumps({"error": str(error)}, sort_keys=True), file=sys.stderr)
        return 2


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