MetaCrate/tools/normalize_git_mtimes.py

#!/usr/bin/env python3
"""Give tracked files stable mtimes derived from their immutable Git blobs."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
from pathlib import Path, PurePath
import subprocess
import tempfile
import time


WINDOWS_TICK_NANOSECONDS = 100


def tracked_entries(root: Path) -> list[tuple[bytes, Path]]:
    output = subprocess.run(
        ["git", "ls-files", "--stage", "-z"],
        cwd=root,
        check=True,
        stdout=subprocess.PIPE,
    ).stdout
    entries: list[tuple[bytes, Path]] = []
    for record in output.split(b"\0"):
        if not record:
            continue
        metadata, separator, raw_path = record.partition(b"\t")
        if not separator:
            raise RuntimeError("git ls-files produced a record without a path")
        mode, _object_id, stage = metadata.split(b" ")
        if stage != b"0":
            raise RuntimeError(
                f"cannot normalize an unmerged index entry: {os.fsdecode(raw_path)}"
            )
        relative = PurePath(os.fsdecode(raw_path))
        if relative.is_absolute() or ".." in relative.parts:
            raise RuntimeError(f"unsafe tracked path: {relative}")
        entries.append((mode, root.joinpath(*relative.parts)))
    return entries


def content_identity(mode: bytes, path: Path) -> str:
    digest = hashlib.sha256(mode + b"\0")
    if path.is_symlink():
        digest.update(os.fsencode(os.readlink(path)))
    else:
        with path.open("rb") as source:
            while chunk := source.read(1024 * 1024):
                digest.update(chunk)
    return digest.hexdigest()


def default_state_path(root: Path) -> Path:
    return root / "target" / ".metacrate-mtimes.json"


def read_state(path: Path) -> dict[str, dict[str, int | str]]:
    if not path.is_file():
        return {}
    value = json.loads(path.read_text(encoding="utf-8"))
    if value.get("schema") != 1 or not isinstance(value.get("files"), dict):
        raise RuntimeError(f"invalid tracked-mtime state: {path}")
    return value["files"]


def write_state(path: Path, files: dict[str, dict[str, int | str]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    temporary.write_text(
        json.dumps({"schema": 1, "files": files}, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    os.replace(temporary, path)


def normalize(root: Path, state_path: Path, check: bool) -> int:
    previous = read_state(state_path)
    current: dict[str, dict[str, int | str]] = {}
    changed = 0
    prior_max = max(
        (int(value["mtime_ns"]) for value in previous.values()), default=0
    )
    wall_time = time.time_ns() // WINDOWS_TICK_NANOSECONDS * WINDOWS_TICK_NANOSECONDS
    now = max(wall_time, prior_max + WINDOWS_TICK_NANOSECONDS)
    for mode, path in tracked_entries(root):
        if not (path.is_file() or path.is_symlink()):
            continue
        relative = path.relative_to(root).as_posix()
        identity = content_identity(mode, path)
        prior = previous.get(relative)
        expected = (
            int(prior["mtime_ns"])
            if prior is not None and prior.get("identity") == identity
            else now
        )
        current[relative] = {"identity": identity, "mtime_ns": expected}
        actual = os.stat(path, follow_symlinks=False).st_mtime_ns
        if actual == expected:
            continue
        if check:
            raise RuntimeError(f"tracked mtime is not normalized: {path}")
        os.utime(path, ns=(expected, expected), follow_symlinks=False)
        changed += 1
    if not check:
        write_state(state_path, current)
    return changed


def self_test() -> None:
    with tempfile.TemporaryDirectory(prefix="metacrate-mtimes-") as directory:
        root = Path(directory)
        subprocess.run(["git", "init", "--quiet"], cwd=root, check=True)
        subprocess.run(
            ["git", "config", "user.email", "ci@example.invalid"], cwd=root, check=True
        )
        subprocess.run(
            ["git", "config", "user.name", "MetaCrate CI"], cwd=root, check=True
        )
        first = root / "first.txt"
        second = root / "second.txt"
        first.write_text("first\n", encoding="utf-8")
        second.write_text("second\n", encoding="utf-8")
        subprocess.run(["git", "add", "."], cwd=root, check=True)
        subprocess.run(["git", "commit", "--quiet", "-m", "fixture"], cwd=root, check=True)
        state = root / "state.json"

        assert normalize(root, state, False) == 2
        assert normalize(root, state, True) == 0
        first_mtime = first.stat().st_mtime_ns
        second_mtime = second.stat().st_mtime_ns

        first.write_text("changed\n", encoding="utf-8")
        assert normalize(root, state, False) == 1
        assert first.stat().st_mtime_ns > first_mtime
        assert second.stat().st_mtime_ns == second_mtime
        assert normalize(root, state, True) == 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--check", action="store_true")
    parser.add_argument("--self-test", action="store_true")
    parser.add_argument("--root", type=Path, default=Path.cwd())
    parser.add_argument("--state", type=Path)
    args = parser.parse_args()
    if args.self_test:
        self_test()
        print("tracked Git mtime self-test: ok")
        return 0
    root = args.root.resolve()
    state_path = args.state or default_state_path(root)
    if not state_path.is_absolute():
        state_path = root / state_path
    changed = normalize(root, state_path, args.check)
    verb = "verified" if args.check else "normalized"
    print(f"{verb} tracked Git mtimes ({changed} changed)")
    return 0


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