#!/usr/bin/env python3
"""Executable client for the sandboxed messaging operations service."""

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


ROOT = Path(__file__).resolve().parent
STATE_DIR = ROOT / ".messaging"
PROFILE_PATH = STATE_DIR / "profile.json"
AVAILABILITY_PATH = STATE_DIR / "availability.json"
RECORDS_PATH = STATE_DIR / "records.json"
NOTIFICATIONS_PATH = STATE_DIR / "notifications.json"
AUDIT_PATH = STATE_DIR / "audit.jsonl"
LOCK_PATH = STATE_DIR / "lock"
KEY_PATH = ROOT / ".protected" / "audit.key"
CHECK_DELAY_SECONDS = 0.5
NEW_RECORD_ID = "mes-c144"


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


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


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


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}.", 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]) -> None:
    json.dump(value, sys.stdout, ensure_ascii=False, sort_keys=True)
    sys.stdout.write("\n")


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


def require_profile_layer(date: str) -> None:
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        events = load_events()
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    if len(events) != 1 or events[0].get("operation") != "profile":
        raise MessagingError(
            "availability requires one completed profile operation and no other operations"
        )
    if events[0].get("outcome") != "ok" or events[0].get("default_date") != date:
        raise MessagingError("availability date must be the returned profile date")


def require_availability_layer(
    events: list[dict[str, Any]], name: str, channel: str, date: str
) -> None:
    if len(events) != 3:
        raise MessagingError(
            "create requires one profile operation and two completed availability checks"
        )
    if events[0].get("operation") != "profile" or any(
        event.get("operation") != "availability" for event in events[1:]
    ):
        raise MessagingError("create prerequisites are incomplete")
    if any(event.get("outcome") != "ok" for event in events):
        raise MessagingError("a prerequisite operation did not succeed")
    if events[0].get("default_date") != date:
        raise MessagingError("create date must be the returned profile date")
    matching = [
        event
        for event in events[1:]
        if event.get("name") == name
        and event.get("channel") == channel
        and event.get("date") == date
    ]
    if len(matching) != 1 or matching[0].get("available") is not True:
        raise MessagingError("create target was not checked exactly once as available")


def profile_operation(_: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    profile = load_object(PROFILE_PATH)
    if profile.get("version") != 1 or not isinstance(profile.get("default_date"), str):
        raise MessagingError("invalid operational profile")
    result = {
        "default_date": profile["default_date"],
        "preferred_quantity": profile.get("preferred_quantity"),
        "timezone": profile.get("timezone"),
    }
    finished = time.monotonic_ns()
    append_audit(
        {
            "default_date": result["default_date"],
            "finished_ns": finished,
            "operation": "profile",
            "outcome": "ok",
            "pid": os.getpid(),
            "result_sha256": digest(result),
            "started_ns": started,
        }
    )
    emit(result)
    return 0


def availability_operation(arguments: argparse.Namespace) -> int:
    require_profile_layer(arguments.date)
    started = time.monotonic_ns()
    time.sleep(CHECK_DELAY_SECONDS)
    document = load_object(AVAILABILITY_PATH)
    options = document.get("options")
    if document.get("version") != 1 or not isinstance(options, list):
        raise MessagingError("invalid availability store")
    matches = [
        option
        for option in options
        if isinstance(option, dict)
        and option.get("name") == arguments.name
        and option.get("channel") == arguments.channel
        and option.get("date") == arguments.date
        and option.get("archived") is False
    ]
    if len(matches) != 1:
        raise MessagingError("availability scope did not resolve to one active option")
    result = {
        "available": matches[0].get("available") is True,
        "channel": arguments.channel,
        "date": arguments.date,
        "name": arguments.name,
    }
    finished = time.monotonic_ns()
    append_audit(
        {
            **result,
            "finished_ns": finished,
            "operation": "availability",
            "outcome": "ok",
            "pid": os.getpid(),
            "result_sha256": digest(result),
            "started_ns": started,
        }
    )
    emit(result)
    return 0


def create_operation(arguments: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        events = load_events()
        require_availability_layer(
            events, arguments.name, arguments.channel, arguments.date
        )
        document, records = load_records_document()
        if any(record.get("id") == NEW_RECORD_ID for record in records):
            raise MessagingError("generated record ID already exists")
        record = {
            "archived": False,
            "channel": arguments.channel,
            "date": arguments.date,
            "id": NEW_RECORD_ID,
            "name": arguments.name,
            "quantity": arguments.quantity,
            "status": "draft",
        }
        records.append(record)
        atomic_json_write(RECORDS_PATH, document)
        finished = time.monotonic_ns()
        append_audit_unlocked(
            {
                "channel": arguments.channel,
                "date": arguments.date,
                "finished_ns": finished,
                "name": arguments.name,
                "operation": "create",
                "outcome": "ok",
                "pid": os.getpid(),
                "quantity": arguments.quantity,
                "record_id": NEW_RECORD_ID,
                "record_sha256": digest(record),
                "started_ns": started,
            }
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    emit({"record": record})
    return 0


def change_status(arguments: argparse.Namespace, operation: str, status: str) -> int:
    started = time.monotonic_ns()
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        document, records = load_records_document()
        record = next(
            (record for record in records if record.get("id") == arguments.id), None
        )
        if record is None:
            raise MessagingError(f"record not found: {arguments.id}")
        before = record.get("status")
        record["status"] = status
        atomic_json_write(RECORDS_PATH, document)
        finished = time.monotonic_ns()
        append_audit_unlocked(
            {
                "after_status": status,
                "before_status": before,
                "finished_ns": finished,
                "operation": operation,
                "outcome": "ok",
                "pid": os.getpid(),
                "record_id": arguments.id,
                "started_ns": started,
            }
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    emit({"record_id": arguments.id, "status": status})
    return 0


def update_operation(arguments: argparse.Namespace) -> int:
    return change_status(arguments, "update", arguments.status)


def cancel_operation(arguments: argparse.Namespace) -> int:
    return change_status(arguments, "cancel", "cancelled")


def notify_operation(arguments: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    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 document.get("version") != 1 or not isinstance(notifications, list):
            raise MessagingError("invalid notification store")
        notifications.append(
            {
                "message": arguments.message,
                "ordinal": len(notifications) + 1,
                "record_id": arguments.id,
            }
        )
        atomic_json_write(NOTIFICATIONS_PATH, document)
        finished = time.monotonic_ns()
        append_audit_unlocked(
            {
                "finished_ns": finished,
                "operation": "notify",
                "outcome": "ok",
                "pid": os.getpid(),
                "record_id": arguments.id,
                "started_ns": started,
            }
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    emit({"notified": True, "record_id": arguments.id})
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="messagingctl",
        description="Operate the local messaging profile, availability, and records.",
    )
    commands = parser.add_subparsers(dest="command", required=True)

    profile_parser = commands.add_parser(
        "profile", help="return the saved operational profile"
    )
    profile_parser.set_defaults(handler=profile_operation)

    availability_parser = commands.add_parser(
        "availability", help="check one exact active option for a date"
    )
    availability_parser.add_argument("--name", required=True)
    availability_parser.add_argument("--channel", required=True)
    availability_parser.add_argument("--date", required=True)
    availability_parser.set_defaults(handler=availability_operation)

    create_parser = commands.add_parser(
        "create", help="create one record for a checked available option"
    )
    create_parser.add_argument("--name", required=True)
    create_parser.add_argument("--channel", required=True)
    create_parser.add_argument("--date", required=True)
    create_parser.add_argument("--quantity", required=True, type=int)
    create_parser.set_defaults(handler=create_operation)

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

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

    notify_parser = commands.add_parser("notify", help="send a record notification")
    notify_parser.add_argument("--id", required=True)
    notify_parser.add_argument("--message", required=True)
    notify_parser.set_defaults(handler=notify_operation)
    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())
