#!/usr/bin/env python3
"""Local fleet registry executable for a Pi dependency-planning task."""

from __future__ import annotations

import argparse
import fcntl
import hashlib
import hmac
import json
import os
from pathlib import Path
import sys
import tempfile
import time
from typing import Any, Callable


ROOT = Path(__file__).resolve().parent
STATE_DIR = ROOT / ".fleet"
RECORDS_PATH = STATE_DIR / "records.json"
AUDIT_PATH = STATE_DIR / "audit.jsonl"
NOTIFICATIONS_PATH = STATE_DIR / "notifications.json"
LOCK_PATH = STATE_DIR / "lock"
KEY_PATH = ROOT / ".protected" / "audit.key"
OPERATION_DELAY_SECONDS = 0.55


def canonical(value: Any) -> bytes:
    return json.dumps(
        value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
    ).encode("utf-8")


def load_object(path: Path) -> dict[str, Any]:
    with path.open(encoding="utf-8") as stream:
        value = json.load(stream)
    if not isinstance(value, dict):
        raise RuntimeError(f"invalid fleet file: {path.name}")
    return value


def load_records() -> list[dict[str, Any]]:
    document = load_object(RECORDS_PATH)
    records = document.get("records")
    if document.get("version") != 1 or not isinstance(records, list):
        raise RuntimeError("invalid fleet record store")
    if not all(isinstance(record, dict) for record in records):
        raise RuntimeError("invalid fleet record entry")
    return records


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


def digest(value: Any) -> str:
    return hashlib.sha256(canonical(value)).hexdigest()


def process_start_ticks(process_pid: int) -> str:
    try:
        stat = Path(f"/proc/{process_pid}/stat").read_text(encoding="utf-8")
        return stat.rsplit(") ", 1)[1].split()[19]
    except (OSError, IndexError):
        return "unavailable"


def append_audit_locked(event: dict[str, Any]) -> None:
    AUDIT_PATH.touch(exist_ok=True)
    with AUDIT_PATH.open(encoding="utf-8") as stream:
        sequence = sum(1 for line in stream if line.strip()) + 1
    sealed = dict(event)
    sealed["sequence"] = sequence
    key = KEY_PATH.read_bytes().strip()
    sealed["seal"] = hmac.new(key, canonical(sealed), hashlib.sha256).hexdigest()
    with AUDIT_PATH.open("a", encoding="utf-8") as stream:
        stream.write(json.dumps(sealed, ensure_ascii=False, sort_keys=True))
        stream.write("\n")
        stream.flush()
        os.fsync(stream.fileno())


def append_audit(event: dict[str, Any]) -> None:
    LOCK_PATH.touch(exist_ok=True)
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        append_audit_locked(event)
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)


def event(operation: str, started: int, **details: Any) -> dict[str, Any]:
    parent_pid = os.getppid()
    return {
        "operation": operation,
        "started_ns": started,
        "finished_ns": time.monotonic_ns(),
        "process_pid": os.getpid(),
        "process_start_ticks": process_start_ticks(os.getpid()),
        "process_group_id": os.getpgrp(),
        "session_id": os.getsid(0),
        "parent_pid": parent_pid,
        "parent_start_ticks": process_start_ticks(parent_pid),
        **details,
    }


def emit(value: dict[str, Any]) -> None:
    json.dump(value, sys.stdout, ensure_ascii=False, sort_keys=True)
    sys.stdout.write("\n")


def search_records(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    time.sleep(OPERATION_DELAY_SECONDS)
    matches = [
        {
            "id": record.get("id"),
            "name": record.get("name"),
            "depot": record.get("depot"),
        }
        for record in load_records()
        if record.get("name") == args.name
        and record.get("depot") == args.depot
    ]
    append_audit(
        event(
            "search",
            started,
            outcome="ok",
            name=args.name,
            depot=args.depot,
            match_count=len(matches),
            matches_sha256=digest(matches),
        )
    )
    emit({"count": len(matches), "matches": matches})
    return 0


def get_record(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    time.sleep(OPERATION_DELAY_SECONDS)
    record = next(
        (record for record in load_records() if record.get("id") == args.id), None
    )
    if record is None:
        append_audit(
            event(
                "get", started, outcome="not-found", record_id=args.id, found=False
            )
        )
        print(f"record not found: {args.id}", file=sys.stderr)
        return 3
    append_audit(
        event(
            "get",
            started,
            outcome="ok",
            record_id=args.id,
            found=True,
            name=record.get("name"),
            depot=record.get("depot"),
            status=record.get("status"),
            record_sha256=digest(record),
        )
    )
    emit({"record": record})
    return 0


def update_record(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    time.sleep(OPERATION_DELAY_SECONDS)
    LOCK_PATH.touch(exist_ok=True)
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        document = load_object(RECORDS_PATH)
        records = document.get("records")
        if not isinstance(records, list):
            raise RuntimeError("invalid fleet record store")
        record = next(
            (record for record in records if record.get("id") == args.id), None
        )
        before = record.get("status") if record is not None else None
        if record is None:
            outcome = "not-found"
            updated = False
        elif before != args.if_status:
            outcome = "condition-not-met"
            updated = False
        else:
            record["status"] = args.status
            atomic_json_write(RECORDS_PATH, document)
            outcome = "ok"
            updated = True
        append_audit_locked(
            event(
                "update",
                started,
                outcome=outcome,
                record_id=args.id,
                conditional_status=args.if_status,
                before_status=before,
                after_status=args.status if updated else before,
                updated=updated,
            )
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    if record is None:
        print(f"record not found: {args.id}", file=sys.stderr)
        return 3
    if not updated:
        emit({"outcome": outcome, "record": record, "updated": False})
        return 4
    emit({"before_status": before, "record": record, "updated": True})
    return 0


def notify(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    time.sleep(OPERATION_DELAY_SECONDS)
    LOCK_PATH.touch(exist_ok=True)
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        document = load_object(NOTIFICATIONS_PATH)
        notifications = document.get("notifications")
        if not isinstance(notifications, list):
            raise RuntimeError("invalid fleet notification store")
        notice = {
            "ordinal": len(notifications) + 1,
            "recipient": args.recipient,
            "record_id": args.record_id,
            "subject": args.subject,
            "message": args.message,
        }
        notifications.append(notice)
        atomic_json_write(NOTIFICATIONS_PATH, document)
        append_audit_locked(
            event(
                "notify",
                started,
                outcome="ok",
                recipient=args.recipient,
                record_id=args.record_id,
                notice_sha256=digest(notice),
                delivered=True,
            )
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    emit({"delivered": True, "notice": notice})
    return 0


def auxiliary_read(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    time.sleep(OPERATION_DELAY_SECONDS)
    records = load_records()
    if args.command == "list":
        result: dict[str, Any] = {"count": len(records), "records": records}
    elif args.command == "profile":
        record = next(
            (record for record in records if record.get("id") == args.id), None
        )
        result = {"profile": record}
    else:
        record = next(
            (record for record in records if record.get("id") == args.id), None
        )
        result = {
            "available": bool(record and record.get("status") == "available"),
            "record_id": args.id,
        }
    append_audit(event(args.command, started, outcome="ok"))
    emit(result)
    return 0


def create_record(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    time.sleep(OPERATION_DELAY_SECONDS)
    LOCK_PATH.touch(exist_ok=True)
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        document = load_object(RECORDS_PATH)
        records = document.get("records")
        if not isinstance(records, list):
            raise RuntimeError("invalid fleet record store")
        record = {
            "id": f"fle-created-{time.monotonic_ns()}",
            "name": args.name,
            "depot": args.depot,
            "status": args.status,
        }
        records.append(record)
        atomic_json_write(RECORDS_PATH, document)
        append_audit_locked(
            event(
                "create", started, outcome="ok", record_id=record["id"], created=True
            )
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    emit({"created": True, "record": record})
    return 0


def cancel_record(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    time.sleep(OPERATION_DELAY_SECONDS)
    LOCK_PATH.touch(exist_ok=True)
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        document = load_object(RECORDS_PATH)
        records = document.get("records")
        if not isinstance(records, list):
            raise RuntimeError("invalid fleet record store")
        record = next(
            (record for record in records if record.get("id") == args.id), None
        )
        if record is not None:
            record["status"] = "cancelled"
            atomic_json_write(RECORDS_PATH, document)
        append_audit_locked(
            event(
                "cancel",
                started,
                outcome="ok" if record is not None else "not-found",
                record_id=args.id,
                cancelled=record is not None,
            )
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    emit({"cancelled": record is not None, "record_id": args.id})
    return 0 if record is not None else 3


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="fleetctl",
        description="Search, retrieve, and manage the local fleet registry.",
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    search_parser = subparsers.add_parser(
        "search", help="search by exact fleet unit name and depot"
    )
    search_parser.add_argument("--name", required=True)
    search_parser.add_argument("--depot", required=True)
    search_parser.set_defaults(handler=search_records)

    get_parser = subparsers.add_parser("get", help="retrieve one complete record by ID")
    get_parser.add_argument("--id", required=True)
    get_parser.set_defaults(handler=get_record)

    update_parser = subparsers.add_parser(
        "update", help="conditionally change one record's status"
    )
    update_parser.add_argument("--id", required=True)
    update_parser.add_argument("--if-status", required=True)
    update_parser.add_argument("--status", required=True)
    update_parser.set_defaults(handler=update_record)

    notify_parser = subparsers.add_parser("notify", help="send one fleet notice")
    notify_parser.add_argument("--recipient", required=True)
    notify_parser.add_argument("--record-id", required=True)
    notify_parser.add_argument("--subject", required=True)
    notify_parser.add_argument("--message", required=True)
    notify_parser.set_defaults(handler=notify)

    list_parser = subparsers.add_parser("list", help="list all fleet records")
    list_parser.set_defaults(handler=auxiliary_read)

    profile_parser = subparsers.add_parser("profile", help="read a saved profile")
    profile_parser.add_argument("--id", required=True)
    profile_parser.set_defaults(handler=auxiliary_read)

    availability_parser = subparsers.add_parser(
        "availability", help="check fleet availability"
    )
    availability_parser.add_argument("--id", required=True)
    availability_parser.set_defaults(handler=auxiliary_read)

    create_parser = subparsers.add_parser("create", help="create a fleet unit record")
    create_parser.add_argument("--name", required=True)
    create_parser.add_argument("--depot", required=True)
    create_parser.add_argument("--status", required=True)
    create_parser.set_defaults(handler=create_record)

    cancel_parser = subparsers.add_parser("cancel", help="cancel a fleet unit record")
    cancel_parser.add_argument("--id", required=True)
    cancel_parser.set_defaults(handler=cancel_record)
    return parser


def main() -> int:
    try:
        parser = build_parser()
        if sys.argv[1:] == ["--help"]:
            started = time.monotonic_ns()
            time.sleep(0.02)
            append_audit(
                event("help", started, outcome="ok", displayed=True)
            )
            parser.print_help()
            return 0
        args = parser.parse_args()
        handler: Callable[[argparse.Namespace], int] = args.handler
        return handler(args)
    except (OSError, RuntimeError, json.JSONDecodeError) as error:
        print(f"fleetctl error: {error}", file=sys.stderr)
        return 2


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