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

from __future__ import annotations

import argparse
from contextlib import contextmanager
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 / ".inventory"
RECORDS_PATH = STATE_DIR / "records.json"
AUDIT_PATH = STATE_DIR / "audit.jsonl"
NOTIFICATIONS_PATH = STATE_DIR / "notifications.json"
LOCK_PATH = STATE_DIR / "lock"
PARALLEL_PATH = STATE_DIR / "parallel.json"
KEY_PATH = ROOT / ".protected" / "audit.key"


class InventoryError(RuntimeError):
    """A user-facing inventory 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 InventoryError(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 InventoryError("invalid inventory store")
    if not all(isinstance(record, dict) for record in records):
        raise InventoryError("invalid inventory record")
    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 InventoryError(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 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 load_parallel_document() -> dict[str, Any]:
    document = load_object(PARALLEL_PATH)
    layers = document.get("layers")
    if document.get("version") != 1 or not isinstance(layers, dict):
        raise InventoryError("invalid parallel-operation store")
    if not all(
        isinstance(layer, str)
        and isinstance(participants, list)
        and all(
            isinstance(participant, dict)
            and isinstance(participant.get("token"), str)
            and isinstance(participant.get("pid"), int)
            and not isinstance(participant["pid"], bool)
            and isinstance(participant.get("done"), bool)
            for participant in participants
        )
        for layer, participants in layers.items()
    ):
        raise InventoryError("invalid parallel-operation state")
    return document


def join_parallel_layer(layer: str) -> str:
    token = f"{os.getpid()}-{time.monotonic_ns()}"
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        document = load_parallel_document()
        participants = document["layers"].setdefault(layer, [])
        if len(participants) >= 2:
            raise InventoryError(f"parallel {layer} layer has too many participants")
        participants.append({"done": False, "pid": os.getpid(), "token": token})
        atomic_json_write(PARALLEL_PATH, document)
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)

    while True:
        with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
            fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
            document = load_parallel_document()
            participants = document["layers"].get(layer, [])
            if len(participants) == 2 and any(
                participant.get("token") == token for participant in participants
            ):
                fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
                return token
            fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
        time.sleep(0.01)


def leave_parallel_layer(layer: str, token: str) -> None:
    with LOCK_PATH.open("a+", encoding="utf-8") as lock_stream:
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_EX)
        document = load_parallel_document()
        participants = document["layers"].get(layer)
        if not isinstance(participants, list):
            raise InventoryError(f"parallel {layer} layer is incomplete")
        matching = [
            participant
            for participant in participants
            if participant.get("token") == token
        ]
        if len(matching) != 1:
            raise InventoryError(f"parallel {layer} participant is missing")
        matching[0]["done"] = True
        if len(participants) == 2 and all(
            participant.get("done") is True for participant in participants
        ):
            del document["layers"][layer]
        atomic_json_write(PARALLEL_PATH, document)
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)


@contextmanager
def parallel_layer(layer: str):
    token = join_parallel_layer(layer)
    try:
        yield
    finally:
        leave_parallel_layer(layer, token)


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


def require_completed_search_layer(record_id: str) -> None:
    events = load_events()
    if len(events) != 2 or [event.get("operation") for event in events] != [
        "search",
        "search",
    ]:
        raise InventoryError("get requires exactly two completed searches")
    if any(event.get("outcome") != "ok" for event in events):
        raise InventoryError("a search prerequisite did not succeed")
    if any(
        not isinstance(event.get("result_ids"), list)
        or len(event["result_ids"]) != 1
        for event in events
    ):
        raise InventoryError("every search must resolve to exactly one record")
    returned_ids = {event["result_ids"][0] for event in events}
    if record_id not in returned_ids:
        raise InventoryError("record ID was not returned by the completed searches")


def require_completed_get_layer(
    events: list[dict[str, Any]], record_id: str
) -> dict[str, Any]:
    if len(events) != 4:
        raise InventoryError("update requires two searches and two retrievals")
    if [event.get("operation") for event in events[:2]] != ["search", "search"]:
        raise InventoryError("search layer is incomplete")
    if [event.get("operation") for event in events[2:]] != ["get", "get"]:
        raise InventoryError("retrieval layer is incomplete")
    if any(event.get("outcome") != "ok" for event in events):
        raise InventoryError("a prerequisite operation did not succeed")
    if any(len(event.get("result_ids", [])) != 1 for event in events[:2]):
        raise InventoryError("a search was not uniquely resolved")
    retrieved = [
        event for event in events[2:] if event.get("record_id") == record_id
    ]
    if len(retrieved) != 1:
        raise InventoryError("record ID was not retrieved exactly once")
    return retrieved[0]


def require_successful_update(
    events: list[dict[str, Any]], record_id: str
) -> dict[str, Any]:
    if len(events) != 5:
        raise InventoryError("notice requires the completed reconciliation and 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 InventoryError("notice requires a successful status-changing update")
    return update


def search_inventory(arguments: argparse.Namespace) -> int:
    started = time.monotonic_ns()
    with parallel_layer("search"):
        matches = [
            {
                "id": record["id"],
                "location": record["location"],
                "name": record["name"],
            }
            for record in read_records()
            if record.get("name") == arguments.name
            and record.get("location") == arguments.location
        ]
        matches.sort(key=lambda record: record["id"])
        finished = time.monotonic_ns()
        append_audit(
            {
                "finished_ns": finished,
                "location": arguments.location,
                "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_inventory(arguments: argparse.Namespace) -> int:
    require_completed_search_layer(arguments.id)
    started = time.monotonic_ns()
    with parallel_layer("get"):
        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 InventoryError(f"inventory 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_inventory(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()
        retrieved = require_completed_get_layer(events, arguments.id)
        document = load_object(RECORDS_PATH)
        records = document.get("records")
        if not isinstance(records, list):
            raise InventoryError("invalid inventory store")
        record = next(
            (record for record in records if record.get("id") == arguments.id),
            None,
        )
        if record is None:
            raise InventoryError(f"inventory record not found: {arguments.id}")
        before = record.get("status")
        if retrieved.get("status") != arguments.if_status:
            raise InventoryError("conditional status does not match retrieved record")
        if before != arguments.if_status:
            finished = time.monotonic_ns()
            append_audit_unlocked(
                {
                    "actual_status": before,
                    "expected_status": arguments.if_status,
                    "finished_ns": finished,
                    "operation": "update",
                    "outcome": "guard-not-met",
                    "record_id": arguments.id,
                    "started_ns": started,
                    "updated": False,
                }
            )
            print(
                f"inventoryctl: update skipped; status is {before!r}, not "
                f"{arguments.if_status!r}",
                file=sys.stderr,
            )
            return 3
        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": True,
            }
        )
        fcntl.flock(lock_stream.fileno(), fcntl.LOCK_UN)
    emit(
        {
            "after_status": arguments.status,
            "before_status": before,
            "record_id": arguments.id,
            "updated": True,
        }
    )
    return 0


def notify_coordinator(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)
        record = next(
            (record for record in read_records() if record.get("id") == arguments.id),
            None,
        )
        if record is None:
            raise InventoryError(f"inventory record not found: {arguments.id}")
        if record.get("status") != update.get("after_status"):
            raise InventoryError("record 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 InventoryError("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="inventoryctl",
        description=(
            "Search, retrieve, conditionally update, and send outcome notices "
            "for local inventory records."
        ),
        epilog=(
            "Command forms:\n"
            "  inventoryctl search --name NAME --location LOCATION\n"
            "  inventoryctl get --id ID\n"
            "  inventoryctl update --id ID --if-status STATUS --status STATUS\n"
            "  inventoryctl notify --id ID --recipient RECIPIENT"
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    commands = parser.add_subparsers(dest="command", required=True)

    search_parser = commands.add_parser(
        "search", help="search by exact item name and exact location"
    )
    search_parser.add_argument("--name", required=True)
    search_parser.add_argument("--location", required=True)
    search_parser.set_defaults(handler=search_inventory)

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

    update_parser = commands.add_parser(
        "update", help="conditionally change one retrieved record's status"
    )
    update_parser.add_argument("--id", required=True)
    update_parser.add_argument("--if-status", required=True)
    update_parser.add_argument(
        "--status", required=True, choices=("reorder-placed",)
    )
    update_parser.set_defaults(handler=update_inventory)

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


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


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