#!/usr/bin/env python3
"""Executable communications registry for the Pi harness 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


ROOT = Path(__file__).resolve().parent
STATE_DIR = ROOT / ".messages"
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.28


class MessageError(RuntimeError):
    """A user-facing communications registry error."""


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


def load_document(path: Path, collection: str) -> dict[str, Any]:
    with path.open(encoding="utf-8") as stream:
        value = json.load(stream)
    if (
        not isinstance(value, dict)
        or value.get("version") != 1
        or not isinstance(value.get(collection), list)
        or not all(isinstance(item, dict) for item in value[collection])
    ):
        raise MessageError(f"invalid communications store: {path.name}")
    return value


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 MessageError(f"invalid audit event at line {line_number}")
            events.append(event)
    return events


def atomic_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 record_digest(record: dict[str, Any]) -> str:
    return hashlib.sha256(canonical(record)).hexdigest()


def append_event_locked(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_event(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_event_locked(event)
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)


def provenance(started: int, finished: int) -> dict[str, int]:
    return {
        "finished_ns": finished,
        "parent_process_id": os.getppid(),
        "process_id": os.getpid(),
        "started_ns": started,
    }


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


def require_search_layer(record_id: str) -> None:
    events = load_events()
    if len(events) != 2 or any(
        event.get("operation") != "search" or event.get("outcome") != "ok"
        for event in events
    ):
        raise MessageError("get requires exactly two completed searches")
    returned_ids = {
        value
        for event in events
        for value in event.get("result_ids", [])
        if isinstance(value, str)
    }
    if record_id not in returned_ids:
        raise MessageError("record ID was not returned by the completed searches")


def require_retrieval_layer(events: list[dict[str, Any]], record_id: str) -> None:
    if len(events) != 4:
        raise MessageError("schedule requires two searches and two retrievals")
    if [event.get("operation") for event in events[:2]] != ["search", "search"]:
        raise MessageError("search dependency layer is incomplete")
    if [event.get("operation") for event in events[2:]] != ["get", "get"]:
        raise MessageError("retrieval dependency layer is incomplete")
    if any(event.get("outcome") != "ok" for event in events):
        raise MessageError("a prerequisite operation did not succeed")
    retrieved_ids = {event.get("record_id") for event in events[2:]}
    if record_id not in retrieved_ids:
        raise MessageError("record ID was not retrieved")


def require_successful_schedule(
    events: list[dict[str, Any]], receipt_value: str
) -> dict[str, Any]:
    if len(events) != 5:
        raise MessageError("notify requires a completed scheduling mutation")
    schedule_event = events[-1]
    if (
        [event.get("operation") for event in events[:2]] != ["search", "search"]
        or [event.get("operation") for event in events[2:4]] != ["get", "get"]
        or schedule_event.get("operation") != "schedule"
        or schedule_event.get("outcome") != "ok"
        or schedule_event.get("scheduled") is not True
        or schedule_event.get("receipt") != receipt_value
    ):
        raise MessageError("notification prerequisite did not succeed")
    return schedule_event


def search(args: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    records = load_document(RECORDS_PATH, "records")["records"]
    matches = [
        {"id": record["id"], "team": record["team"], "title": record["title"]}
        for record in records
        if record.get("title") == args.title and record.get("team") == args.team
    ]
    matches.sort(key=lambda record: record["id"])
    time.sleep(READ_DELAY_SECONDS)
    finished = time.monotonic_ns()
    append_event(
        {
            "operation": "search",
            "outcome": "ok",
            "result_ids": [record["id"] for record in matches],
            "team": args.team,
            "title": args.title,
            **provenance(started, finished),
        }
    )
    emit(
        {
            "count": len(matches),
            "query": {"team": args.team, "title": args.title},
            "matches": matches,
        }
    )
    return 0


def get_record(args: argparse.Namespace) -> int:
    require_search_layer(args.id)
    started = time.monotonic_ns()
    records = load_document(RECORDS_PATH, "records")["records"]
    record = next((item for item in records if item.get("id") == args.id), None)
    time.sleep(READ_DELAY_SECONDS)
    finished = time.monotonic_ns()
    event: dict[str, Any] = {
        "operation": "get",
        "record_id": args.id,
        **provenance(started, finished),
    }
    if record is None:
        event.update({"found": False, "outcome": "not-found"})
        append_event(event)
        raise MessageError(f"record not found: {args.id}")
    event.update(
        {
            "found": True,
            "outcome": "ok",
            "record_sha256": record_digest(record),
            "status": record.get("status"),
        }
    )
    append_event(event)
    emit({"record": record})
    return 0


def schedule(args: 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_retrieval_layer(events, args.id)
        records_document = load_document(RECORDS_PATH, "records")
        records = records_document["records"]
        record = next((item for item in records if item.get("id") == args.id), None)
        before = record.get("status") if record is not None else None
        if record is None or before != "draft" or args.if_status != "draft":
            finished = time.monotonic_ns()
            append_event_locked(
                {
                    "after_status": before,
                    "before_status": before,
                    "conditional_status": args.if_status,
                    "operation": "schedule",
                    "outcome": (
                        "not-found" if record is None else "condition-not-met"
                    ),
                    "receipt": None,
                    "record_id": args.id,
                    "scheduled": False,
                    **provenance(started, finished),
                }
            )
            fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
            emit(
                {
                    "record_id": args.id,
                    "scheduled": False,
                    "status": before,
                }
            )
            return 4

        receipt_value = "sch-" + hashlib.sha256(
            f"{args.id}\0{started}\0{os.getpid()}".encode("utf-8")
        ).hexdigest()[:20]
        record["status"] = "scheduled"
        receipts_document = load_document(RECEIPTS_PATH, "receipts")
        receipts_document["receipts"].append(
            {
                "consumed": False,
                "from_status": "draft",
                "message_id": args.id,
                "receipt": receipt_value,
                "to_status": "scheduled",
            }
        )
        atomic_write(RECORDS_PATH, records_document)
        atomic_write(RECEIPTS_PATH, receipts_document)
        finished = time.monotonic_ns()
        append_event_locked(
            {
                "after_status": "scheduled",
                "before_status": before,
                "conditional_status": args.if_status,
                "operation": "schedule",
                "outcome": "ok",
                "receipt": receipt_value,
                "record_id": args.id,
                "scheduled": True,
                **provenance(started, finished),
            }
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    emit(
        {
            "mutation_receipt": receipt_value,
            "record": record,
            "scheduled": True,
            "transition": {"from_status": "draft", "to_status": "scheduled"},
        }
    )
    return 0


def notify(args: 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)
        schedule_event = require_successful_schedule(
            load_events(), args.after_receipt
        )
        receipts_document = load_document(RECEIPTS_PATH, "receipts")
        receipt = next(
            (
                item
                for item in receipts_document["receipts"]
                if item.get("receipt") == args.after_receipt
            ),
            None,
        )
        records = load_document(RECORDS_PATH, "records")["records"]
        record = (
            next(
                (
                    item
                    for item in records
                    if item.get("id") == receipt.get("message_id")
                ),
                None,
            )
            if receipt is not None
            else None
        )
        valid = (
            receipt is not None
            and receipt.get("consumed") is False
            and receipt.get("message_id") == schedule_event.get("record_id")
            and receipt.get("from_status") == "draft"
            and receipt.get("to_status") == "scheduled"
            and record is not None
            and record.get("status") == "scheduled"
        )
        if not valid:
            raise MessageError(
                "notification rejected: no unconsumed successful schedule receipt"
            )

        subject = f"Scheduling outcome: {record['title']}"
        message = (
            f"{record['title']} in {record['team']} changed from draft to scheduled."
        )
        notifications_document = load_document(
            NOTIFICATIONS_PATH, "notifications"
        )
        notification = {
            "message": message,
            "message_id": receipt["message_id"],
            "mutation_receipt": receipt["receipt"],
            "ordinal": len(notifications_document["notifications"]) + 1,
            "recipient": args.recipient,
            "subject": subject,
        }
        notifications_document["notifications"].append(notification)
        receipt["consumed"] = True
        atomic_write(NOTIFICATIONS_PATH, notifications_document)
        atomic_write(RECEIPTS_PATH, receipts_document)
        finished = time.monotonic_ns()
        append_event_locked(
            {
                "delivered": True,
                "message_sha256": hashlib.sha256(
                    message.encode("utf-8")
                ).hexdigest(),
                "operation": "notify",
                "outcome": "ok",
                "receipt": receipt["receipt"],
                "recipient": args.recipient,
                "record_id": receipt["message_id"],
                **provenance(started, finished),
            }
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    emit(
        {
            "delivered": True,
            "message": message,
            "mutation_receipt": receipt["receipt"],
            "recipient": args.recipient,
            "record_id": receipt["message_id"],
            "subject": subject,
        }
    )
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="messagectl",
        description=(
            "Search, retrieve, conditionally schedule, and notify for local "
            "communications records."
        ),
    )
    commands = parser.add_subparsers(dest="command", required=True)

    search_parser = commands.add_parser(
        "search", help="find exact title-and-team matches"
    )
    search_parser.add_argument("--title", required=True, help="exact message title")
    search_parser.add_argument("--team", required=True, help="exact owning team")
    search_parser.set_defaults(handler=search)

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

    schedule_parser = commands.add_parser(
        "schedule",
        help=(
            "conditionally change one retrieved draft to scheduled and return "
            "mutation evidence"
        ),
    )
    schedule_parser.add_argument("--id", required=True, help="retrieved stable ID")
    schedule_parser.add_argument(
        "--if-status", required=True, help="status observed in the complete record"
    )
    schedule_parser.set_defaults(handler=schedule)

    notify_parser = commands.add_parser(
        "notify",
        help="notify after a successful schedule using its mutation evidence",
    )
    notify_parser.add_argument("--recipient", required=True)
    notify_parser.add_argument(
        "--after-receipt", required=True, help="receipt returned by schedule"
    )
    notify_parser.set_defaults(handler=notify)
    return parser


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


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