#!/usr/bin/env python3
"""Executable interface for the dependency-planning operations sandbox."""

from __future__ import annotations

import argparse
import concurrent.futures
import fcntl
import json
import os
from pathlib import Path
import sys
import tempfile
import threading
from typing import Any


ROOT = Path(__file__).resolve().parent
STATE = ROOT / "state"
PROFILE = STATE / "profile.json"
OPTIONS = STATE / "options.json"
RECORDS = STATE / "records.json"
NOTIFICATIONS = STATE / "notifications.json"
AUDIT = STATE / "audit.jsonl"
LOCK = STATE / ".lock"


class OperationError(RuntimeError):
    pass


def read_json(path: Path) -> Any:
    with path.open("r", encoding="utf-8") as handle:
        return json.load(handle)


def write_json_atomic(path: Path, value: Any) -> None:
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{path.name}.", dir=path.parent, text=True
    )
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            json.dump(value, handle, indent=2, sort_keys=True)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary_name, path)
    finally:
        if os.path.exists(temporary_name):
            os.unlink(temporary_name)


def emit(value: Any) -> None:
    print(json.dumps(value, sort_keys=True, separators=(",", ":")))


def append_audit_locked(event: dict[str, Any]) -> None:
    with AUDIT.open("r+", encoding="utf-8") as handle:
        lines = [line for line in handle.read().splitlines() if line]
        event["seq"] = len(lines) + 1
        handle.seek(0, os.SEEK_END)
        handle.write(json.dumps(event, sort_keys=True, separators=(",", ":")))
        handle.write("\n")
        handle.flush()
        os.fsync(handle.fileno())


def option_index() -> dict[str, dict[str, Any]]:
    data = read_json(OPTIONS)
    return {entry["key"]: entry for entry in data["options"]}


def profile_command(_: argparse.Namespace) -> None:
    profile = read_json(PROFILE)
    returned = {"default_date": profile["default_date"]}
    with LOCK.open("r+", encoding="utf-8") as lock:
        fcntl.flock(lock, fcntl.LOCK_EX)
        append_audit_locked({"event": "profile.read", "returned": returned})
    emit(returned)


def check_one(
    key: str, date: str, gate: threading.Barrier
) -> dict[str, Any]:
    gate.wait()
    entry = option_index().get(key)
    if entry is None:
        raise OperationError(f"unknown option key: {key}")
    return {
        "available": date in entry["available_dates"],
        "city": entry["city"],
        "date": date,
        "option": entry["name"],
    }


def availability_command(args: argparse.Namespace) -> None:
    if len(args.option_keys) < 2:
        raise OperationError("a parallel availability batch needs at least two options")
    if len(set(args.option_keys)) != len(args.option_keys):
        raise OperationError("option keys in a batch must be unique")

    gate = threading.Barrier(len(args.option_keys))
    with concurrent.futures.ThreadPoolExecutor(
        max_workers=len(args.option_keys), thread_name_prefix="availability"
    ) as executor:
        futures = [
            executor.submit(check_one, key, args.date, gate)
            for key in args.option_keys
        ]
        results = [future.result() for future in futures]

    returned = {"results": results}
    event = {
        "date": args.date,
        "event": "availability.batch",
        "option_keys": args.option_keys,
        "parallel": True,
        "results": results,
        "worker_count": len(args.option_keys),
    }
    with LOCK.open("r+", encoding="utf-8") as lock:
        fcntl.flock(lock, fcntl.LOCK_EX)
        append_audit_locked(event)
    emit(returned)


def create_command(args: argparse.Namespace) -> None:
    if args.quantity < 1:
        raise OperationError("quantity must be a positive integer")

    with LOCK.open("r+", encoding="utf-8") as lock:
        fcntl.flock(lock, fcntl.LOCK_EX)
        entry = option_index().get(args.option_key)
        if entry is None:
            raise OperationError(f"unknown option key: {args.option_key}")
        if args.date not in entry["available_dates"]:
            raise OperationError("the requested option is not available on that date")

        store = read_json(RECORDS)
        if any(
            record["option"] == entry["name"] and record["date"] == args.date
            for record in store["records"]
        ):
            raise OperationError("a record already exists for that option and date")

        record = {
            "city": entry["city"],
            "date": args.date,
            "option": entry["name"],
            "quantity": args.quantity,
            "record_id": f"record-{len(store['records']) + 1:03d}",
        }
        store["records"].append(record)
        write_json_atomic(RECORDS, store)
        append_audit_locked({"event": "record.created", "returned": record})
    emit(record)


def notify_command(args: argparse.Namespace) -> None:
    with LOCK.open("r+", encoding="utf-8") as lock:
        fcntl.flock(lock, fcntl.LOCK_EX)
        records = read_json(RECORDS)["records"]
        if not any(record["record_id"] == args.record_id for record in records):
            raise OperationError(f"unknown record id: {args.record_id}")

        store = read_json(NOTIFICATIONS)
        notification = {
            "message": args.message,
            "notification_id": f"notification-{len(store['notifications']) + 1:03d}",
            "record_id": args.record_id,
        }
        store["notifications"].append(notification)
        write_json_atomic(NOTIFICATIONS, store)
        append_audit_locked({"event": "notification.sent", "returned": notification})
    emit(notification)


def parser() -> argparse.ArgumentParser:
    root = argparse.ArgumentParser(prog="opsctl")
    commands = root.add_subparsers(dest="command", required=True)

    profile = commands.add_parser("profile", help="read the saved operational profile")
    profile.set_defaults(handler=profile_command)

    availability = commands.add_parser(
        "availability", help="check options in one concurrent batch"
    )
    availability.add_argument("--parallel", action="store_true", required=True)
    availability.add_argument("--date", required=True)
    availability.add_argument("option_keys", nargs="+")
    availability.set_defaults(handler=availability_command)

    create = commands.add_parser("create", help="create one record")
    create.add_argument("--date", required=True)
    create.add_argument("--quantity", required=True, type=int)
    create.add_argument("option_key")
    create.set_defaults(handler=create_command)

    notify = commands.add_parser("notify", help="send a notification for a record")
    notify.add_argument("--record-id", required=True)
    notify.add_argument("--message", required=True)
    notify.set_defaults(handler=notify_command)
    return root


def main() -> int:
    args = parser().parse_args()
    try:
        args.handler(args)
        return 0
    except OperationError as error:
        try:
            with LOCK.open("r+", encoding="utf-8") as lock:
                fcntl.flock(lock, fcntl.LOCK_EX)
                append_audit_locked(
                    {"event": "operation.failed", "operation": args.command}
                )
        except OSError:
            pass
        print(f"opsctl: {error}", file=sys.stderr)
        return 2
    except (OSError, ValueError, KeyError) as error:
        print(f"opsctl: {error}", file=sys.stderr)
        return 2


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