#!/usr/bin/env python3
"""Executable interface to the sandboxed support-case 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


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


class SupportError(RuntimeError):
    """A user-facing support 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 SupportError(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 read_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 SupportError("invalid record store")
    if not all(isinstance(record, dict) for record in records):
        raise SupportError("invalid record entry")
    return records


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


def record_digest(record: dict[str, Any]) -> str:
    return hashlib.sha256(canonical(record)).hexdigest()


def caller_parent_start_ticks(parent_pid: int) -> int:
    stat = Path(f"/proc/{parent_pid}/stat").read_text(encoding="utf-8")
    fields_after_name = stat.rsplit(")", 1)[1].split()
    return int(fields_after_name[19])


def append_audit_unlocked(event: dict[str, Any]) -> None:
    sealed = dict(event)
    parent_pid = os.getppid()
    sealed["caller_parent_pid"] = parent_pid
    sealed["caller_parent_start_ticks"] = caller_parent_start_ticks(parent_pid)
    sealed["caller_pid_namespace"] = os.readlink("/proc/self/ns/pid")
    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 require_help_before_search() -> None:
    events = load_events()
    if not events or events[0].get("operation") != "help":
        raise SupportError("search requires a completed top-level --help call")
    if events[0].get("outcome") != "ok":
        raise SupportError("the top-level --help call did not succeed")
    if len(events) > 2 or any(
        event.get("operation") != "search" for event in events[1:]
    ):
        raise SupportError("the search layer is already complete or out of order")


def require_completed_search_layer(record_id: str) -> None:
    events = load_events()
    if len(events) not in (3, 4) or [
        event.get("operation") for event in events[:3]
    ] != ["help", "search", "search"]:
        raise SupportError("get requires exactly two completed searches")
    if any(event.get("operation") != "get" for event in events[3:]):
        raise SupportError("the retrieval layer is already complete or out of order")
    if any(event.get("outcome") != "ok" for event in events):
        raise SupportError("a search prerequisite did not succeed")
    returned_ids = {
        value
        for event in events[1:3]
        for value in event.get("result_ids", [])
        if isinstance(value, str)
    }
    if record_id not in returned_ids:
        raise SupportError("record ID was not returned by the completed searches")
    if any(event.get("record_id") == record_id for event in events[3:]):
        raise SupportError("record ID was already retrieved")


def require_completed_get_layer(events: list[dict[str, Any]], record_id: str) -> None:
    if len(events) != 5:
        raise SupportError(
            "update requires help, exactly two searches, and two retrievals"
        )
    if events[0].get("operation") != "help":
        raise SupportError("top-level help call is missing")
    if [event.get("operation") for event in events[1:3]] != ["search", "search"]:
        raise SupportError("search layer is incomplete")
    if [event.get("operation") for event in events[3:]] != ["get", "get"]:
        raise SupportError("retrieval layer is incomplete")
    if any(event.get("outcome") != "ok" for event in events):
        raise SupportError("a prerequisite operation did not succeed")
    retrieved_ids = {event.get("record_id") for event in events[3:]}
    if record_id not in retrieved_ids:
        raise SupportError("record ID was not retrieved")


def require_successful_update(events: list[dict[str, Any]], record_id: str) -> dict[str, Any]:
    if len(events) != 6:
        raise SupportError(
            "notice requires help, the completed reconciliation, and one update"
        )
    update = events[-1]
    if (
        update.get("operation") != "update"
        or update.get("outcome") != "ok"
        or update.get("record_id") != record_id
        or update.get("updated") is not True
    ):
        raise SupportError("notice requires a successful status-changing update")
    return update


def record_help() -> None:
    started = time.monotonic_ns()
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        if load_events():
            raise SupportError("top-level --help must be the first service invocation")
        finished = time.monotonic_ns()
        append_audit_unlocked(
            {
                "finished_ns": finished,
                "operation": "help",
                "outcome": "ok",
                "started_ns": started,
            }
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)


def search_records(arguments: argparse.Namespace) -> int:
    require_help_before_search()
    started = time.monotonic_ns()
    time.sleep(READ_DELAY_SECONDS)
    matches = [
        {
            "account": record["account"],
            "id": record["id"],
            "name": record["name"],
        }
        for record in read_records()
        if record.get("name") == arguments.name
        and record.get("account") == arguments.account
    ]
    matches.sort(key=lambda record: record["id"])
    finished = time.monotonic_ns()
    append_audit(
        {
            "account": arguments.account,
            "finished_ns": finished,
            "name": arguments.name,
            "operation": "search",
            "outcome": "ok",
            "result_ids": [record["id"] for record in matches],
            "started_ns": started,
        }
    )
    emit({"count": len(matches), "matches": matches})
    return 0


def get_record(arguments: argparse.Namespace) -> int:
    require_completed_search_layer(arguments.id)
    started = time.monotonic_ns()
    time.sleep(READ_DELAY_SECONDS)
    record = next(
        (record for record in read_records() if record.get("id") == arguments.id), None
    )
    finished = time.monotonic_ns()
    event: dict[str, Any] = {
        "finished_ns": finished,
        "operation": "get",
        "record_id": arguments.id,
        "started_ns": started,
    }
    if record is None:
        event.update({"found": False, "outcome": "not-found"})
        append_audit(event)
        raise SupportError(f"record not found: {arguments.id}")
    event.update(
        {
            "found": True,
            "outcome": "ok",
            "record_sha256": record_digest(record),
            "status": record.get("status"),
        }
    )
    append_audit(event)
    emit({"record": record})
    return 0


def update_record(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_completed_get_layer(events, arguments.id)
        document = load_object(RECORDS_PATH)
        records = document.get("records")
        if not isinstance(records, list):
            raise SupportError("invalid record store")
        record = next(
            (record for record in records if record.get("id") == arguments.id), None
        )
        if record is None:
            raise SupportError(f"record not found: {arguments.id}")
        before = record.get("status")
        record["status"] = arguments.status
        atomic_json_write(RECORDS_PATH, document)
        finished = time.monotonic_ns()
        append_audit_unlocked(
            {
                "after_status": arguments.status,
                "before_status": before,
                "finished_ns": finished,
                "operation": "update",
                "outcome": "ok",
                "record_id": arguments.id,
                "started_ns": started,
                "updated": before != arguments.status,
            }
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    emit(
        {
            "after_status": arguments.status,
            "before_status": before,
            "record_id": arguments.id,
            "updated": before != arguments.status,
        }
    )
    return 0


def notify_lead(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()
        update = require_successful_update(events, arguments.id)
        records = read_records()
        record = next(
            (record for record in records if record.get("id") == arguments.id), None
        )
        if record is None:
            raise SupportError(f"record not found: {arguments.id}")
        if record.get("status") != update.get("after_status"):
            raise SupportError("case status no longer matches the successful update")
        subject = f"Record update: {record.get('name')}"
        body = (
            f"{record.get('name')} was updated from {update.get('before_status')} "
            f"to {update.get('after_status')} after status verification."
        )
        document = load_object(NOTIFICATIONS_PATH)
        notifications = document.get("notifications")
        if not isinstance(notifications, list):
            raise SupportError("invalid notification store")
        notification = {
            "body": body,
            "ordinal": len(notifications) + 1,
            "recipient": arguments.recipient,
            "record_id": arguments.id,
            "subject": subject,
        }
        notifications.append(notification)
        atomic_json_write(NOTIFICATIONS_PATH, document)
        finished = time.monotonic_ns()
        append_audit_unlocked(
            {
                "finished_ns": finished,
                "notification_sha256": hashlib.sha256(
                    canonical(notification)
                ).hexdigest(),
                "operation": "notify",
                "outcome": "ok",
                "recipient": arguments.recipient,
                "record_id": arguments.id,
                "started_ns": started,
            }
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    emit(
        {
            "body": body,
            "delivered": True,
            "recipient": arguments.recipient,
            "record_id": arguments.id,
            "subject": subject,
        }
    )
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="supportctl",
        description=(
            "Search, retrieve, update, and send outcome notices for local support cases."
        ),
    )
    commands = parser.add_subparsers(dest="command", required=True)

    search_parser = commands.add_parser(
        "search", help="search by exact case name and account"
    )
    search_parser.add_argument("--name", required=True)
    search_parser.add_argument("--account", required=True)
    search_parser.set_defaults(handler=search_records)

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

    update_parser = commands.add_parser("update", help="change one case status")
    update_parser.add_argument("--id", required=True)
    update_parser.add_argument("--status", required=True)
    update_parser.set_defaults(handler=update_record)

    notify_parser = commands.add_parser(
        "notify", help="send the outcome of the immediately preceding successful update"
    )
    notify_parser.add_argument("--id", required=True)
    notify_parser.add_argument(
        "--recipient", required=True, choices=("support lead",)
    )
    notify_parser.set_defaults(handler=notify_lead)
    return parser


def main() -> int:
    if sys.argv[1:] == ["--help"]:
        try:
            record_help()
        except (SupportError, OSError, ValueError, json.JSONDecodeError) as error:
            print(f"supportctl: {error}", file=sys.stderr)
            return 2
        build_parser().print_help()
        return 0
    arguments = build_parser().parse_args()
    try:
        return int(arguments.handler(arguments))
    except (SupportError, OSError, ValueError, json.JSONDecodeError) as error:
        print(f"supportctl: {error}", file=sys.stderr)
        return 2


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