#!/usr/bin/env python3
"""Executable interface to the sandboxed communications ledger."""

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 / ".messaging"
PROTECTED_DIR = ROOT / ".protected"
AVAILABILITY_PATH = STATE_DIR / "availability.json"
MESSAGES_PATH = STATE_DIR / "messages.json"
PREFERENCES_PATH = STATE_DIR / "preferences.json"
NOTIFICATIONS_PATH = STATE_DIR / "notifications.json"
MUTATIONS_PATH = STATE_DIR / "mutations.json"
ATTEMPTS_PATH = STATE_DIR / "attempts.json"
AUDIT_PATH = STATE_DIR / "audit.jsonl"
LOCK_PATH = STATE_DIR / "lock"
KEY_PATH = PROTECTED_DIR / "audit.key"
READ_DELAY_SECONDS = 0.65

RETRY_NAME = "Evening Shuttle Delay Notice"
RETRY_DATE = "2026-11-18"
RETRY_LOCATION = "Transportation Desk"


class MessagingError(RuntimeError):
    """A user-facing messaging service error."""


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 MessagingError(f"invalid service data: {path.name}")
    return value


def atomic_json_write(path: Path, value: dict[str, Any]) -> None:
    descriptor, temporary_name = tempfile.mkstemp(
        dir=path.parent, prefix=f".{path.name}.", suffix=".tmp", 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, sort_keys=True)
            stream.write("\n")
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
    finally:
        if temporary.exists():
            temporary.unlink()


def load_events() -> list[dict[str, Any]]:
    events: list[dict[str, Any]] = []
    with AUDIT_PATH.open(encoding="utf-8") as stream:
        for line_number, line in enumerate(stream, 1):
            if not line.strip():
                continue
            event = json.loads(line)
            if not isinstance(event, dict):
                raise MessagingError(f"invalid audit event at line {line_number}")
            events.append(event)
    return events


def append_audit_unlocked(event: dict[str, Any]) -> None:
    sealed = dict(event)
    sealed["sequence"] = len(load_events()) + 1
    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_unlocked(event)
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)


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


def base_event(operation: str, started_ns: int, finished_ns: int) -> dict[str, Any]:
    return {
        "finished_ns": finished_ns,
        "operation": operation,
        "parent_pid": os.getppid(),
        "started_ns": started_ns,
    }


def scope(arguments: argparse.Namespace) -> tuple[str, str, str]:
    return arguments.name, arguments.date, arguments.location


def availability(arguments: argparse.Namespace) -> int:
    started_ns = time.monotonic_ns()
    parent_pid = os.getppid()
    time.sleep(READ_DELAY_SECONDS)
    requested = scope(arguments)

    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        attempts_document = load_object(ATTEMPTS_PATH)
        attempts = attempts_document.get("attempts")
        if attempts_document.get("version") != 1 or not isinstance(attempts, list):
            raise MessagingError("invalid attempt ledger")

        attempt_entry = next(
            (
                entry
                for entry in attempts
                if isinstance(entry, dict)
                and (entry.get("name"), entry.get("date"), entry.get("location"))
                == requested
            ),
            None,
        )
        if attempt_entry is None:
            attempt_entry = {
                "count": 0,
                "date": arguments.date,
                "location": arguments.location,
                "name": arguments.name,
            }
            attempts.append(attempt_entry)
        count = attempt_entry.get("count")
        if not isinstance(count, int) or isinstance(count, bool) or count < 0:
            raise MessagingError("invalid attempt count")
        attempt = count + 1
        attempt_entry["count"] = attempt
        attempts.sort(key=lambda item: (item["name"], item["date"], item["location"]))
        atomic_json_write(ATTEMPTS_PATH, attempts_document)

        availability_document = load_object(AVAILABILITY_PATH)
        slots = availability_document.get("slots")
        if availability_document.get("version") != 1 or not isinstance(slots, list):
            raise MessagingError("invalid availability ledger")
        slot = next(
            (
                entry
                for entry in slots
                if isinstance(entry, dict)
                and (entry.get("name"), entry.get("date"), entry.get("location"))
                == requested
            ),
            None,
        )

        finished_ns = time.monotonic_ns()
        event = {
            "attempt": attempt,
            "date": arguments.date,
            "finished_ns": finished_ns,
            "location": arguments.location,
            "name": arguments.name,
            "operation": "availability",
            "parent_pid": parent_pid,
            "started_ns": started_ns,
        }
        if requested == (RETRY_NAME, RETRY_DATE, RETRY_LOCATION) and attempt == 1:
            event.update(
                {
                    "error_code": "SERVICE_BUSY",
                    "may_have_committed": False,
                    "outcome": "temporary_error",
                    "retryable": True,
                }
            )
            append_audit_unlocked(event)
            fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
            emit(
                {
                    "attempt": attempt,
                    "error": "SERVICE_BUSY",
                    "may_have_committed": False,
                    "message": "availability service is temporarily busy",
                    "retryable": True,
                },
                error=True,
            )
            return 75

        if slot is None:
            event.update(
                {
                    "may_have_committed": False,
                    "outcome": "not_found",
                    "retryable": False,
                }
            )
            append_audit_unlocked(event)
            fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
            emit(
                {
                    "attempt": attempt,
                    "error": "NOT_FOUND",
                    "may_have_committed": False,
                    "retryable": False,
                },
                error=True,
            )
            return 4

        available = slot.get("available")
        if not isinstance(available, bool):
            raise MessagingError("invalid availability value")
        event.update(
            {
                "available": available,
                "may_have_committed": False,
                "outcome": "ok",
                "retryable": False,
            }
        )
        append_audit_unlocked(event)
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)

    emit(
        {
            "attempt": attempt,
            "available": available,
            "date": arguments.date,
            "location": arguments.location,
            "name": arguments.name,
        }
    )
    return 0


def read_operation(
    operation: str,
    producer: Callable[[], dict[str, Any]],
    details: dict[str, Any] | None = None,
) -> int:
    started_ns = time.monotonic_ns()
    time.sleep(0.1)
    result = producer()
    finished_ns = time.monotonic_ns()
    event = base_event(operation, started_ns, finished_ns)
    event.update(details or {})
    event["outcome"] = "ok"
    append_audit(event)
    emit(result)
    return 0


def search(arguments: argparse.Namespace) -> int:
    def produce() -> dict[str, Any]:
        records = load_object(MESSAGES_PATH).get("records", [])
        matches = [
            record
            for record in records
            if isinstance(record, dict)
            and record.get("name") == arguments.name
            and record.get("location") == arguments.location
        ]
        return {"count": len(matches), "matches": matches}

    return read_operation(
        "search",
        produce,
        {"location": arguments.location, "name": arguments.name},
    )


def get_record(arguments: argparse.Namespace) -> int:
    def produce() -> dict[str, Any]:
        records = load_object(MESSAGES_PATH).get("records", [])
        record = next(
            (
                item
                for item in records
                if isinstance(item, dict) and item.get("id") == arguments.id
            ),
            None,
        )
        return {"record": record}

    return read_operation("get", produce, {"record_id": arguments.id})


def list_records(_arguments: argparse.Namespace) -> int:
    return read_operation("list", lambda: load_object(MESSAGES_PATH))


def preferences(_arguments: argparse.Namespace) -> int:
    return read_operation("preferences", lambda: load_object(PREFERENCES_PATH))


def mutation(arguments: argparse.Namespace) -> int:
    started_ns = time.monotonic_ns()
    operation = arguments.command
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        result: dict[str, Any]
        record_id = getattr(arguments, "id", None)

        if operation == "notify":
            document = load_object(NOTIFICATIONS_PATH)
            notifications = document.get("notifications")
            if not isinstance(notifications, list):
                raise MessagingError("invalid notification ledger")
            notifications.append(
                {
                    "message": arguments.message,
                    "ordinal": len(notifications) + 1,
                    "record_id": record_id,
                }
            )
            atomic_json_write(NOTIFICATIONS_PATH, document)
            result = {"notified": True, "record_id": record_id}
        else:
            document = load_object(MESSAGES_PATH)
            records = document.get("records")
            if not isinstance(records, list):
                raise MessagingError("invalid message ledger")
            if operation == "create":
                record_id = f"mes-{900 + len(records)}"
                records.append(
                    {
                        "date": arguments.date,
                        "id": record_id,
                        "location": arguments.location,
                        "name": arguments.name,
                        "status": arguments.status,
                    }
                )
                result = {"created": True, "record_id": record_id}
            else:
                record = next(
                    (
                        item
                        for item in records
                        if isinstance(item, dict) and item.get("id") == record_id
                    ),
                    None,
                )
                if record is None:
                    raise MessagingError(f"record not found: {record_id}")
                before = record.get("status")
                record["status"] = (
                    arguments.status if operation == "update" else "cancelled"
                )
                result = {
                    "after_status": record["status"],
                    "before_status": before,
                    "record_id": record_id,
                    "updated": before != record["status"],
                }
            atomic_json_write(MESSAGES_PATH, document)

        mutation_document = load_object(MUTATIONS_PATH)
        mutations = mutation_document.get("mutations")
        if not isinstance(mutations, list):
            raise MessagingError("invalid mutation ledger")
        mutations.append(
            {
                "operation": operation,
                "ordinal": len(mutations) + 1,
                "record_id": record_id,
            }
        )
        atomic_json_write(MUTATIONS_PATH, mutation_document)

        finished_ns = time.monotonic_ns()
        event = base_event(operation, started_ns, finished_ns)
        event.update({"outcome": "ok", "record_id": record_id})
        append_audit_unlocked(event)
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)

    emit(result)
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="messagingctl",
        description="Operate the local communications scheduling ledger.",
    )
    commands = parser.add_subparsers(dest="command", required=True)

    availability_parser = commands.add_parser(
        "availability", help="check one exact publication slot"
    )
    availability_parser.add_argument("--name", required=True)
    availability_parser.add_argument("--date", required=True)
    availability_parser.add_argument("--location", required=True)
    availability_parser.set_defaults(handler=availability)

    search_parser = commands.add_parser("search", help="search message records")
    search_parser.add_argument("--name", required=True)
    search_parser.add_argument("--location", required=True)
    search_parser.set_defaults(handler=search)

    get_parser = commands.add_parser("get", help="retrieve a message record")
    get_parser.add_argument("--id", required=True)
    get_parser.set_defaults(handler=get_record)

    list_parser = commands.add_parser("list", help="list message records")
    list_parser.set_defaults(handler=list_records)

    preferences_parser = commands.add_parser(
        "preferences", help="read saved messaging preferences"
    )
    preferences_parser.set_defaults(handler=preferences)

    create_parser = commands.add_parser("create", help="create a message record")
    create_parser.add_argument("--name", required=True)
    create_parser.add_argument("--date", required=True)
    create_parser.add_argument("--location", required=True)
    create_parser.add_argument("--status", required=True)
    create_parser.set_defaults(handler=mutation)

    update_parser = commands.add_parser("update", help="update a message status")
    update_parser.add_argument("--id", required=True)
    update_parser.add_argument("--status", required=True)
    update_parser.set_defaults(handler=mutation)

    cancel_parser = commands.add_parser("cancel", help="cancel a message record")
    cancel_parser.add_argument("--id", required=True)
    cancel_parser.set_defaults(handler=mutation)

    notify_parser = commands.add_parser("notify", help="send a message notification")
    notify_parser.add_argument("--id", required=True)
    notify_parser.add_argument("--message", required=True)
    notify_parser.set_defaults(handler=mutation)
    return parser


def main() -> int:
    arguments = build_parser().parse_args()
    try:
        return int(arguments.handler(arguments))
    except (MessagingError, OSError, ValueError, json.JSONDecodeError) as error:
        print(f"messagingctl: {error}", file=sys.stderr)
        return 2


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