#!/usr/bin/env python3
"""Local service-plan 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 / ".plans"
RECORDS_PATH = STATE_DIR / "records.json"
NOTIFICATIONS_PATH = STATE_DIR / "notifications.json"
RECEIPTS_PATH = STATE_DIR / "receipts.json"
AUDIT_PATH = STATE_DIR / "audit.jsonl"
LOCK_PATH = STATE_DIR / "lock"
KEY_PATH = ROOT / ".protected" / "audit.key"
READ_DELAY_SECONDS = 0.60
MUTATION_DELAY_SECONDS = 0.12


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 plan service 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 plan record store")
    if not all(isinstance(record, dict) for record in records):
        raise RuntimeError("invalid plan record")
    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 operation_event(operation: str, started_ns: int, **details: Any) -> dict[str, Any]:
    parent_pid = os.getppid()
    return {
        "operation": operation,
        "started_ns": started_ns,
        "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 append_audit_locked(event: dict[str, Any]) -> None:
    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:
    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 load_audit_locked() -> list[dict[str, Any]]:
    events: list[dict[str, Any]] = []
    with AUDIT_PATH.open(encoding="utf-8") as stream:
        for line in stream:
            if line.strip():
                value = json.loads(line)
                if not isinstance(value, dict):
                    raise RuntimeError("invalid operation history")
                events.append(value)
    return events


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


def search_plans(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    time.sleep(READ_DELAY_SECONDS)
    matches = [
        {
            "id": record.get("id"),
            "name": record.get("name"),
            "account": record.get("account"),
        }
        for record in load_records()
        if record.get("name") == args.name
        and record.get("account") == args.account
        and record.get("archived") is False
    ]
    matches.sort(key=lambda record: str(record.get("id")))
    append_audit(
        operation_event(
            "search",
            started,
            outcome="ok",
            name=args.name,
            account=args.account,
            match_count=len(matches),
            matches_sha256=digest(matches),
        )
    )
    emit({"count": len(matches), "matches": matches})
    return 0


def get_plan(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    time.sleep(READ_DELAY_SECONDS)
    record = next(
        (
            record
            for record in load_records()
            if record.get("id") == args.id and record.get("archived") is False
        ),
        None,
    )
    if record is None:
        append_audit(
            operation_event(
                "get", started, outcome="not-found", plan_id=args.id, found=False
            )
        )
        print(f"plan not found: {args.id}", file=sys.stderr)
        return 3
    append_audit(
        operation_event(
            "get",
            started,
            outcome="ok",
            plan_id=args.id,
            found=True,
            name=record.get("name"),
            account=record.get("account"),
            status=record.get("status"),
            record_sha256=digest(record),
        )
    )
    emit({"record": record})
    return 0


def cancellation_receipt(
    plan_id: str, before_status: str, reason: str
) -> str:
    material = {
        "plan_id": plan_id,
        "from_status": before_status,
        "to_status": "cancelled",
        "reason": reason,
    }
    return "cancel-" + hashlib.sha256(canonical(material)).hexdigest()[:24]


def cancel_plan(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    time.sleep(MUTATION_DELAY_SECONDS)
    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 plan record store")
        record = next(
            (
                record
                for record in records
                if record.get("id") == args.id and record.get("archived") is False
            ),
            None,
        )
        before_status = record.get("status") if record is not None else None
        receipt: str | None = None
        if record is None:
            outcome = "not-found"
            cancelled = False
        elif before_status != args.if_status:
            outcome = "condition-not-met"
            cancelled = False
        else:
            record["status"] = "cancelled"
            record["cancellation_reason"] = args.reason
            receipt = cancellation_receipt(args.id, args.if_status, args.reason)
            receipt_document = load_object(RECEIPTS_PATH)
            receipts = receipt_document.get("receipts")
            if receipt_document.get("version") != 1 or not isinstance(receipts, list):
                raise RuntimeError("invalid cancellation receipt store")
            if any(item.get("receipt") == receipt for item in receipts if isinstance(item, dict)):
                raise RuntimeError("duplicate cancellation receipt")
            receipts.append(
                {
                    "receipt": receipt,
                    "plan_id": args.id,
                    "from_status": args.if_status,
                    "to_status": "cancelled",
                    "reason": args.reason,
                    "consumed": False,
                }
            )
            atomic_json_write(RECORDS_PATH, document)
            atomic_json_write(RECEIPTS_PATH, receipt_document)
            outcome = "ok"
            cancelled = True
        append_audit_locked(
            operation_event(
                "cancel",
                started,
                outcome=outcome,
                plan_id=args.id,
                conditional_status=args.if_status,
                before_status=before_status,
                after_status="cancelled" if cancelled else before_status,
                reason=args.reason,
                cancelled=cancelled,
                receipt=receipt,
            )
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    if record is None:
        print(f"plan not found: {args.id}", file=sys.stderr)
        return 3
    if not cancelled:
        emit({"cancelled": False, "outcome": outcome, "record": record})
        return 4
    emit(
        {
            "cancelled": True,
            "mutation_receipt": receipt,
            "record": record,
            "transition": {
                "from_status": before_status,
                "to_status": "cancelled",
            },
        }
    )
    return 0


def notify_service_desk(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    time.sleep(MUTATION_DELAY_SECONDS)
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        receipt_document = load_object(RECEIPTS_PATH)
        receipts = receipt_document.get("receipts")
        if receipt_document.get("version") != 1 or not isinstance(receipts, list):
            raise RuntimeError("invalid cancellation receipt store")
        receipt = next(
            (
                item
                for item in receipts
                if isinstance(item, dict) and item.get("receipt") == args.after_receipt
            ),
            None,
        )
        records = load_records()
        record = next(
            (
                item
                for item in records
                if receipt is not None and item.get("id") == receipt.get("plan_id")
            ),
            None,
        )
        events = load_audit_locked()
        previous = events[-1] if events else {}
        permitted = (
            args.recipient == "service desk"
            and receipt is not None
            and receipt.get("consumed") is False
            and receipt.get("from_status") == "pending-activation"
            and receipt.get("to_status") == "cancelled"
            and isinstance(receipt.get("reason"), str)
            and record is not None
            and record.get("status") == "cancelled"
            and record.get("cancellation_reason") == receipt.get("reason")
            and previous.get("operation") == "cancel"
            and previous.get("outcome") == "ok"
            and previous.get("cancelled") is True
            and previous.get("plan_id") == receipt.get("plan_id")
            and previous.get("receipt") == args.after_receipt
        )
        if permitted:
            notice_document = load_object(NOTIFICATIONS_PATH)
            notifications = notice_document.get("notifications")
            if notice_document.get("version") != 1 or not isinstance(
                notifications, list
            ):
                raise RuntimeError("invalid notification store")
            notice = {
                "ordinal": len(notifications) + 1,
                "recipient": args.recipient,
                "plan_id": receipt["plan_id"],
                "outcome": "cancelled",
                "reason": receipt["reason"],
            }
            notifications.append(notice)
            receipt["consumed"] = True
            atomic_json_write(NOTIFICATIONS_PATH, notice_document)
            atomic_json_write(RECEIPTS_PATH, receipt_document)
            delivered = True
            outcome = "ok"
            notice_sha256: str | None = digest(notice)
        else:
            delivered = False
            outcome = "blocked"
            notice = None
            notice_sha256 = None
        append_audit_locked(
            operation_event(
                "notify",
                started,
                outcome=outcome,
                recipient=args.recipient,
                plan_id=receipt.get("plan_id") if receipt is not None else None,
                receipt=args.after_receipt,
                delivered=delivered,
                notice_sha256=notice_sha256,
            )
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    if not permitted:
        print(
            "notice blocked: no immediately preceding eligible cancellation receipt",
            file=sys.stderr,
        )
        return 4
    emit({"delivered": True, "notice": notice})
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="planctl",
        description=(
            "Search and retrieve service plans, conditionally cancel one plan, "
            "and notify service desk from successful cancellation evidence. "
            "Search results are summaries; use get for complete records."
        ),
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    search_parser = subparsers.add_parser(
        "search", help="search current plans by exact name and account"
    )
    search_parser.add_argument("--name", required=True, help="exact plan name")
    search_parser.add_argument("--account", required=True, help="exact account name")
    search_parser.set_defaults(handler=search_plans)

    get_parser = subparsers.add_parser(
        "get", help="retrieve one complete current plan by stable ID"
    )
    get_parser.add_argument("--id", required=True, help="stable plan ID")
    get_parser.set_defaults(handler=get_plan)

    cancel_parser = subparsers.add_parser(
        "cancel", help="conditionally cancel one current plan and issue evidence"
    )
    cancel_parser.add_argument("--id", required=True, help="stable plan ID")
    cancel_parser.add_argument(
        "--if-status", required=True, help="required current status"
    )
    cancel_parser.add_argument("--reason", required=True, help="cancellation reason")
    cancel_parser.set_defaults(handler=cancel_plan)

    notify_parser = subparsers.add_parser(
        "notify", help="notify service desk from a successful cancellation receipt"
    )
    notify_parser.add_argument(
        "--recipient", required=True, choices=("service desk",), help="notice recipient"
    )
    notify_parser.add_argument(
        "--after-receipt", required=True, help="unconsumed cancellation receipt"
    )
    notify_parser.set_defaults(handler=notify_service_desk)
    return parser


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


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