#!/usr/bin/env python3
"""Revoke a Fleet Ops demo bearer token by jti before its natural expiry."""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
import tempfile
from pathlib import Path
from typing import cast

_JTI_PATTERN = re.compile(r"^[0-9a-f]{32}$")


def _validate_jti(jti: str) -> str:
    if not _JTI_PATTERN.fullmatch(jti):
        raise ValueError(
            "jti must be the 32-character lowercase hexadecimal value emitted at mint time"
        )
    return jti


def _load_document(path: Path) -> dict[str, object]:
    if not path.exists():
        return {"schemaVersion": "1.0", "revokedJtis": []}
    if not path.is_file():
        raise ValueError("revocations file is not a regular file")
    try:
        document = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        raise ValueError("revocations file is invalid JSON") from exc
    revoked = document.get("revokedJtis") if isinstance(document, dict) else None
    if (
        not isinstance(document, dict)
        or document.get("schemaVersion") != "1.0"
        or not isinstance(revoked, list)
        or not all(isinstance(item, str) and _JTI_PATTERN.fullmatch(item) for item in revoked)
    ):
        raise ValueError("revocations file does not match schemaVersion 1.0")
    return cast(dict[str, object], document)


def _atomic_write(path: Path, document: dict[str, object]) -> None:
    if not path.parent.is_dir():
        raise ValueError("revocations file parent directory does not exist")
    existing_metadata = path.stat() if path.exists() else None
    temporary_name: str | None = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False
        ) as temporary:
            temporary_name = temporary.name
            os.fchmod(temporary.fileno(), 0o600)
            if existing_metadata is not None and os.name != "nt":
                fchown = getattr(os, "fchown", None)
                if not callable(fchown):
                    raise ValueError("platform cannot preserve revocation file ownership")
                fchown(temporary.fileno(), existing_metadata.st_uid, existing_metadata.st_gid)
            json.dump(document, temporary, indent=2)
            temporary.write("\n")
            temporary.flush()
            os.fsync(temporary.fileno())
        os.replace(temporary_name, path)
    finally:
        if temporary_name is not None and Path(temporary_name).exists():
            Path(temporary_name).unlink()


def revoke(path: Path, jti: str) -> list[str]:
    document = _load_document(path)
    current = cast(list[str], document["revokedJtis"])
    revoked = {_validate_jti(item) for item in current}
    revoked.add(_validate_jti(jti))
    updated: dict[str, object] = {"schemaVersion": "1.0", "revokedJtis": sorted(revoked)}
    _atomic_write(path, updated)
    return cast(list[str], updated["revokedJtis"])


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--jti", required=True)
    parser.add_argument(
        "--revocations-file",
        type=Path,
        required=True,
        help="Path to the HOST-side revocations file -- MUST be the exact path set as "
        "DEMO_TOKEN_REVOCATIONS_FILE_HOST in the operator env file. "
        "A mismatch here means revocation silently does nothing.",
    )
    args = parser.parse_args()
    revoked = revoke(args.revocations_file, args.jti)
    print(f"revocation recorded: count={len(revoked)} file={args.revocations_file}")
    return 0


if __name__ == "__main__":
    sys.exit(main())