lanspread/tools/catalog_source_cache.py

#!/usr/bin/env python3
"""Cheap source metadata cache for complete catalog publication."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import stat
import sys
import tempfile
from pathlib import Path
from typing import Iterable


CACHE_SCHEMA = 1
CATALOG_CONTENT_INDEX = "catalog-content-index-v1.jsonl"
CATALOG_PUBLICATION_MARKER = ".lanspread-catalog-publication-in-progress"


def _absolute(path: Path) -> str:
    return os.path.abspath(os.fspath(path))


def _metadata(path: Path) -> tuple[str, int, int]:
    try:
        info = path.lstat()
    except FileNotFoundError:
        return ("missing", 0, 0)

    mode = stat.S_IFMT(info.st_mode)
    kind = {
        stat.S_IFDIR: "directory",
        stat.S_IFREG: "file",
        stat.S_IFLNK: "symlink",
    }.get(mode, f"special:{mode:o}")
    return (kind, info.st_size, info.st_mtime_ns)


def _tree_entries(root: Path) -> Iterable[tuple[str, str, int, int]]:
    def visit(path: Path, relative: Path) -> Iterable[tuple[str, str, int, int]]:
        kind, size, mtime_ns = _metadata(path)
        yield (relative.as_posix() or ".", kind, size, mtime_ns)
        if kind != "directory":
            return

        with os.scandir(path) as entries:
            children = sorted(entries, key=lambda entry: entry.name)
        for child in children:
            yield from visit(path / child.name, relative / child.name)

    yield from visit(root, Path())


def _tree_fingerprint(root: Path) -> str:
    digest = hashlib.sha256()
    for relative, kind, size, mtime_ns in _tree_entries(root):
        record = json.dumps(
            [relative, kind, size, mtime_ns],
            ensure_ascii=True,
            separators=(",", ":"),
        ).encode("utf-8")
        digest.update(record)
        digest.update(b"\n")
    return digest.hexdigest()


def _path_metadata(path: Path) -> dict[str, int | str]:
    kind, size, mtime_ns = _metadata(path)
    return {"kind": kind, "size": size, "mtime_ns": mtime_ns}


def _cache_payload(
    packages_dir: Path,
    catalog_db: Path,
    manifests_dir: Path,
    unrar: Path,
) -> dict[str, object]:
    return {
        "schema": CACHE_SCHEMA,
        "packages_dir": _absolute(packages_dir),
        "catalog_db": {
            "path": _absolute(catalog_db),
            **_path_metadata(catalog_db),
        },
        "manifests_dir": _absolute(manifests_dir),
        "unrar": {
            "path": _absolute(unrar),
            **_path_metadata(unrar),
        },
        "packages_fingerprint": _tree_fingerprint(packages_dir),
    }


def _output_is_ready(catalog_db: Path, manifests_dir: Path) -> bool:
    db_kind, _, _ = _metadata(catalog_db)
    manifests_kind, _, _ = _metadata(manifests_dir)
    index_kind, _, _ = _metadata(manifests_dir / CATALOG_CONTENT_INDEX)
    marker_kind, _, _ = _metadata(manifests_dir / CATALOG_PUBLICATION_MARKER)
    return (
        db_kind == "file"
        and manifests_kind == "directory"
        and index_kind == "file"
        and marker_kind == "missing"
    )


def cache_is_current(
    stamp: Path,
    packages_dir: Path,
    catalog_db: Path,
    manifests_dir: Path,
    unrar: Path,
) -> bool:
    if not _output_is_ready(catalog_db, manifests_dir):
        return False
    try:
        cached = json.loads(stamp.read_text(encoding="utf-8"))
    except (FileNotFoundError, OSError, UnicodeDecodeError, json.JSONDecodeError):
        return False
    return cached == _cache_payload(packages_dir, catalog_db, manifests_dir, unrar)


def record_cache(
    stamp: Path,
    packages_dir: Path,
    catalog_db: Path,
    manifests_dir: Path,
    unrar: Path,
) -> None:
    if not _output_is_ready(catalog_db, manifests_dir):
        raise RuntimeError("cannot record a catalog cache before publication is complete")
    stamp.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(
        _cache_payload(packages_dir, catalog_db, manifests_dir, unrar),
        ensure_ascii=True,
        indent=2,
        sort_keys=True,
    )
    with tempfile.NamedTemporaryFile(
        "w",
        dir=stamp.parent,
        prefix=f".{stamp.name}.",
        suffix=".tmp",
        encoding="utf-8",
        delete=False,
    ) as temporary:
        temporary.write(payload)
        temporary.write("\n")
        temporary.flush()
        os.fsync(temporary.fileno())
        temporary_path = Path(temporary.name)
    os.replace(temporary_path, stamp)


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)
    for command in ("check", "record"):
        subparser = subparsers.add_parser(command)
        subparser.add_argument("--stamp", type=Path, required=True)
        subparser.add_argument("--packages-dir", type=Path, required=True)
        subparser.add_argument("--catalog-db", type=Path, required=True)
        subparser.add_argument("--manifests-dir", type=Path, required=True)
        subparser.add_argument("--unrar", type=Path, required=True)
    return parser


def main(arguments: list[str] | None = None) -> int:
    options = _parser().parse_args(arguments)
    current = cache_is_current(
        options.stamp,
        options.packages_dir,
        options.catalog_db,
        options.manifests_dir,
        options.unrar,
    )
    if options.command == "check":
        if current:
            print(f"catalog cache hit: {options.packages_dir}")
            return 0
        print(f"catalog cache miss: {options.packages_dir}")
        return 1

    record_cache(
        options.stamp,
        options.packages_dir,
        options.catalog_db,
        options.manifests_dir,
        options.unrar,
    )
    print(f"catalog cache recorded: {options.stamp}")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, RuntimeError) as error:
        print(f"catalog cache error: {error}", file=sys.stderr)
        raise SystemExit(2) from error