#!/usr/bin/env python3
"""Validate and merge Gitea CI sccache artifacts into one object snapshot."""

from __future__ import annotations

from dataclasses import dataclass
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import shutil
import stat
import tarfile
import tempfile
from typing import BinaryIO, Mapping, Sequence
import zipfile


GIB = 1024 ** 3
DEFAULT_MAX_COMBINED_BYTES = 4 * GIB
MAX_INPUT_OBJECT_BYTES = 4 * GIB
MAX_MANIFEST_BYTES = 16 * 1024 ** 2
MAX_OBJECTS = 100_000
MAX_BASE_OBJECTS = 1_000_000
MAX_TAR_OVERHEAD_BYTES = 128 * 1024 ** 2
OBJECT_PATH = re.compile(r"objects/([0-9a-f])/([0-9a-f])/([0-9a-f]{64})\Z")
SHA256 = re.compile(r"[0-9a-f]{64}\Z")
TEMP_PREFIX = ".gitea-cache-snapshot-"


class SnapshotError(ValueError):
    """The downloaded snapshot does not satisfy the trusted artifact contract."""


@dataclass(frozen=True)
class ArtifactIdentity:
    repository: str
    run_id: int
    run_attempt: int
    job: str
    source_sha: str


@dataclass(frozen=True)
class ArtifactInput:
    archive: Path
    expected: ArtifactIdentity


@dataclass(frozen=True)
class MergeResult:
    object_count: int
    total_bytes: int
    source_sha: str
    rustc: str
    workspace: str
    base_image: str | None
    sccache_version: str


@dataclass(frozen=True)
class _Object:
    path: str
    size: int
    sha256: str | None
    mtime_ns: int
    source_index: int | None


@dataclass(frozen=True)
class _Touch:
    path: str
    mtime_ns: int


@dataclass(frozen=True)
class _Archive:
    input: ArtifactInput
    manifest: Mapping[str, object]
    objects: tuple[_Object, ...]
    touched: tuple[_Touch, ...]
    signature: tuple[int, int, int]


def _is_int(value: object) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)


def _require_string(value: object, field: str) -> str:
    if not isinstance(value, str) or not value or "\0" in value:
        raise SnapshotError(f"manifest {field} must be a non-empty string")
    return value


def _strict_object(pairs: list[tuple[str, object]]) -> dict[str, object]:
    result: dict[str, object] = {}
    for key, value in pairs:
        if key in result:
            raise SnapshotError(f"manifest contains duplicate JSON key: {key}")
        result[key] = value
    return result


def _archive_signature(path: Path) -> tuple[int, int, int]:
    try:
        status = path.stat()
    except OSError as error:
        raise SnapshotError(f"cannot stat artifact archive: {path}") from error
    if path.is_symlink() or not path.is_file():
        raise SnapshotError(f"artifact archive must be a regular file: {path}")
    return status.st_size, status.st_mtime_ns, status.st_ino


def _safe_zip_path(name: str) -> tuple[str, ...]:
    if not name or "\0" in name or "\\" in name or name.startswith("/"):
        raise SnapshotError(f"unsafe ZIP member path: {name!r}")
    parts = PurePosixPath(name.rstrip("/")).parts
    if not parts or any(part in {"", ".", ".."} for part in parts):
        raise SnapshotError(f"unsafe ZIP member path: {name!r}")
    return parts


def _snapshot_member(archive: Path) -> tuple[zipfile.ZipFile, zipfile.ZipInfo]:
    try:
        bundle = zipfile.ZipFile(archive)
    except (OSError, zipfile.BadZipFile) as error:
        raise SnapshotError(f"invalid artifact ZIP: {archive}") from error

    try:
        infos = bundle.infolist()
        if len({info.filename for info in infos}) != len(infos):
            raise SnapshotError("ZIP contains duplicate members")
        files: list[zipfile.ZipInfo] = []
        directories: list[tuple[str, ...]] = []
        for info in infos:
            parts = _safe_zip_path(info.filename)
            mode = info.external_attr >> 16
            if info.flag_bits & 1:
                raise SnapshotError("encrypted ZIP members are not accepted")
            if stat.S_ISLNK(mode):
                raise SnapshotError("ZIP links are not accepted")
            if info.is_dir():
                directories.append(parts)
            else:
                files.append(info)
        if len(files) != 1:
            raise SnapshotError("ZIP must contain exactly one snapshot.tar file")
        member = files[0]
        parts = _safe_zip_path(member.filename)
        if not (parts == ("snapshot.tar",) or
                (len(parts) == 2 and parts[1] == "snapshot.tar")):
            raise SnapshotError("ZIP member must be snapshot.tar or <artifact>/snapshot.tar")
        expected_directories = set() if len(parts) == 1 else {(parts[0],)}
        if set(directories) - expected_directories:
            raise SnapshotError("ZIP contains unexpected directory members")
        maximum_tar_size = (
            MAX_INPUT_OBJECT_BYTES + MAX_MANIFEST_BYTES + MAX_TAR_OVERHEAD_BYTES
        )
        if member.file_size > maximum_tar_size:
            raise SnapshotError("snapshot.tar exceeds the per-job input bound")
        if member.compress_type != zipfile.ZIP_STORED:
            raise SnapshotError("snapshot.tar ZIP member must use ZIP_STORED")
        return bundle, member
    except Exception:
        bundle.close()
        raise


def validate_artifact_zip(archive: Path) -> None:
    """Validate the bounded stored ZIP envelope before consuming an artifact."""
    bundle, _ = _snapshot_member(archive)
    with bundle:
        if bundle.testzip() is not None:
            raise SnapshotError("artifact ZIP CRC check failed")


def _tar_stream(archive: Path):
    bundle, member = _snapshot_member(archive)
    try:
        raw = bundle.open(member, "r")
    except Exception:
        bundle.close()
        raise
    try:
        tar = tarfile.open(fileobj=raw, mode="r|")
    except (OSError, tarfile.TarError) as error:
        raw.close()
        bundle.close()
        raise SnapshotError(f"invalid snapshot.tar in {archive}") from error
    return bundle, raw, tar


def _hash_stream(stream: BinaryIO, expected_size: int) -> tuple[int, str]:
    digest = hashlib.sha256()
    total = 0
    while True:
        chunk = stream.read(1024 * 1024)
        if not chunk:
            break
        total += len(chunk)
        if total > expected_size:
            raise SnapshotError("tar member contains more bytes than declared")
        digest.update(chunk)
    if total != expected_size:
        raise SnapshotError("tar member is truncated")
    return total, digest.hexdigest()


def _validate_object_path(path: str) -> None:
    match = OBJECT_PATH.fullmatch(path)
    if not match or match[1] != match[3][0] or match[2] != match[3][1]:
        raise SnapshotError(f"invalid sccache object path: {path!r}")


def _parse_manifest(
    contents: bytes,
    expected: ArtifactIdentity,
    expected_inherited_source_sha: str,
) -> dict[str, object]:
    try:
        manifest = json.loads(contents, object_pairs_hook=_strict_object)
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise SnapshotError("manifest.json is not valid UTF-8 JSON") from error
    if not isinstance(manifest, dict):
        raise SnapshotError("manifest.json must contain an object")
    required = {
        "schema", "repository", "run_id", "run_attempt", "job", "source_sha",
        "rustc", "workspace", "sccache_version", "mode", "inherited_source_sha",
        "objects", "touched",
    }
    allowed = required | {"base_image"}
    if set(manifest) != required and set(manifest) != allowed:
        raise SnapshotError("manifest.json has missing or unexpected fields")
    if not _is_int(manifest["schema"]) or manifest["schema"] != 1:
        raise SnapshotError("manifest schema must be 1")
    for field in ("repository", "job", "source_sha", "rustc", "workspace",
                  "sccache_version", "mode", "inherited_source_sha"):
        _require_string(manifest[field], field)
    if manifest["mode"] != "delta":
        raise SnapshotError("manifest mode must be delta")
    if manifest["inherited_source_sha"] != expected_inherited_source_sha:
        raise SnapshotError("manifest inherited_source_sha does not match expected source image")
    for field in ("run_id", "run_attempt"):
        if not _is_int(manifest[field]) or manifest[field] < 1:
            raise SnapshotError(f"manifest {field} must be a positive integer")
    if "base_image" in manifest:
        _require_string(manifest["base_image"], "base_image")

    for field in ("repository", "run_id", "run_attempt", "job", "source_sha"):
        if manifest[field] != getattr(expected, field):
            raise SnapshotError(f"manifest {field} does not match expected identity")
    if not isinstance(manifest["objects"], list):
        raise SnapshotError("manifest objects must be a list")
    if not isinstance(manifest["touched"], list):
        raise SnapshotError("manifest touched must be a list")
    return manifest


def _manifest_objects(manifest: Mapping[str, object], source_index: int) -> tuple[_Object, ...]:
    rows = manifest["objects"]
    assert isinstance(rows, list)
    if len(rows) > MAX_OBJECTS:
        raise SnapshotError("manifest contains too many objects")
    objects: list[_Object] = []
    seen: set[str] = set()
    for row in rows:
        if not isinstance(row, dict) or set(row) != {"path", "size", "sha256", "mtime_ns"}:
            raise SnapshotError("manifest object has missing or unexpected fields")
        path = row["path"]
        checksum = row["sha256"]
        if not isinstance(path, str):
            raise SnapshotError("manifest object path must be a string")
        _validate_object_path(path)
        if path in seen:
            raise SnapshotError(f"manifest contains duplicate object: {path}")
        seen.add(path)
        if not _is_int(row["size"]) or row["size"] < 0:
            raise SnapshotError(f"manifest object size is invalid: {path}")
        if not isinstance(checksum, str) or not SHA256.fullmatch(checksum):
            raise SnapshotError(f"manifest object sha256 is invalid: {path}")
        if not _is_int(row["mtime_ns"]) or row["mtime_ns"] < 0:
            raise SnapshotError(f"manifest object mtime_ns is invalid: {path}")
        objects.append(_Object(
            path=path,
            size=row["size"],
            sha256=checksum,
            mtime_ns=row["mtime_ns"],
            source_index=source_index,
        ))
    return tuple(objects)


def _manifest_touches(manifest: Mapping[str, object]) -> tuple[_Touch, ...]:
    rows = manifest["touched"]
    assert isinstance(rows, list)
    if len(rows) > MAX_OBJECTS:
        raise SnapshotError("manifest contains too many touched objects")
    touched: list[_Touch] = []
    seen: set[str] = set()
    for row in rows:
        if not isinstance(row, dict) or set(row) != {"path", "mtime_ns"}:
            raise SnapshotError("manifest touched object has missing or unexpected fields")
        path = row["path"]
        if not isinstance(path, str):
            raise SnapshotError("manifest touched path must be a string")
        _validate_object_path(path)
        if path in seen:
            raise SnapshotError(f"manifest contains duplicate touched object: {path}")
        seen.add(path)
        if not _is_int(row["mtime_ns"]) or row["mtime_ns"] < 0:
            raise SnapshotError(f"manifest touched mtime_ns is invalid: {path}")
        touched.append(_Touch(path=path, mtime_ns=row["mtime_ns"]))
    return tuple(touched)


def _validate_archive(
    item: ArtifactInput,
    source_index: int,
    expected_inherited_source_sha: str,
) -> _Archive:
    path = Path(item.archive)
    signature = _archive_signature(path)
    bundle, raw, tar = _tar_stream(path)
    actual: dict[str, tuple[int, str]] = {}
    manifest_contents: bytes | None = None
    total_bytes = 0
    member_count = 0
    try:
        try:
            for member in tar:
                member_count += 1
                if member_count > MAX_OBJECTS + 1:
                    raise SnapshotError("snapshot.tar contains too many members")
                if manifest_contents is not None:
                    raise SnapshotError("manifest.json must be the last tar member")
                if not member.isreg():
                    raise SnapshotError(f"tar member must be a regular file: {member.name!r}")
                if member.name == "manifest.json":
                    if member.size < 0 or member.size > MAX_MANIFEST_BYTES:
                        raise SnapshotError("manifest.json exceeds its size bound")
                    stream = tar.extractfile(member)
                    if stream is None:
                        raise SnapshotError("cannot read manifest.json")
                    manifest_contents = stream.read(MAX_MANIFEST_BYTES + 1)
                    if len(manifest_contents) != member.size:
                        raise SnapshotError("manifest.json is truncated")
                    continue
                _validate_object_path(member.name)
                if member.size < 0:
                    raise SnapshotError(f"tar member has a negative size: {member.name}")
                if member.name in actual:
                    raise SnapshotError(f"snapshot.tar contains duplicate member: {member.name}")
                total_bytes += member.size
                if total_bytes > MAX_INPUT_OBJECT_BYTES:
                    raise SnapshotError("snapshot exceeds the per-job object byte bound")
                stream = tar.extractfile(member)
                if stream is None:
                    raise SnapshotError(f"cannot read tar member: {member.name}")
                actual[member.name] = _hash_stream(stream, member.size)
        except (OSError, tarfile.TarError, zipfile.BadZipFile) as error:
            raise SnapshotError(f"cannot stream snapshot archive: {path}") from error
    finally:
        tar.close()
        raw.close()
        bundle.close()

    if manifest_contents is None:
        raise SnapshotError("snapshot.tar is missing final manifest.json")
    manifest = _parse_manifest(
        manifest_contents, item.expected, expected_inherited_source_sha,
    )
    objects = _manifest_objects(manifest, source_index)
    touched = _manifest_touches(manifest)
    if {obj.path for obj in objects} & {touch.path for touch in touched}:
        raise SnapshotError("an object cannot be both uploaded and touched")
    described = {obj.path: (obj.size, obj.sha256) for obj in objects}
    if actual != described:
        raise SnapshotError("manifest object list, sizes, or checksums do not match snapshot.tar")
    return _Archive(
        input=item, manifest=manifest, objects=objects, touched=touched,
        signature=signature,
    )


def _consistent_metadata(archives: Sequence[_Archive]) -> tuple[str, str, str, str, str | None]:
    first = archives[0].manifest
    fields = (
        "source_sha", "rustc", "workspace", "sccache_version", "base_image",
        "inherited_source_sha",
    )
    expected = tuple(first.get(field) for field in fields)
    for archive in archives[1:]:
        if tuple(archive.manifest.get(field) for field in fields) != expected:
            raise SnapshotError("artifact compiler, workspace, source, or base-image metadata differs")
    source_sha, rustc, workspace, sccache_version, base_image, _ = expected
    assert isinstance(source_sha, str)
    assert isinstance(rustc, str)
    assert isinstance(workspace, str)
    assert isinstance(sccache_version, str)
    assert base_image is None or isinstance(base_image, str)
    return source_sha, rustc, workspace, sccache_version, base_image


def _validated_base_objects(base_objects: os.PathLike[str] | str) -> tuple[Path, dict[str, _Object]]:
    requested = Path(base_objects)
    if not requested.is_absolute():
        raise SnapshotError("base_objects must be absolute")
    try:
        root = requested.resolve(strict=True)
    except OSError as error:
        raise SnapshotError("base_objects must exist") from error
    if requested != root or requested.is_symlink() or not root.is_dir():
        raise SnapshotError("base_objects must be a real canonical directory")

    objects: dict[str, _Object] = {}
    for first in root.iterdir():
        if first.is_symlink() or not first.is_dir() or not re.fullmatch(r"[0-9a-f]", first.name):
            raise SnapshotError(f"noncanonical entry in base objects: {first}")
        for second in first.iterdir():
            if second.is_symlink() or not second.is_dir() or not re.fullmatch(r"[0-9a-f]", second.name):
                raise SnapshotError(f"noncanonical entry in base objects: {second}")
            for entry in second.iterdir():
                relative = f"objects/{first.name}/{second.name}/{entry.name}"
                _validate_object_path(relative)
                if entry.is_symlink() or not entry.is_file():
                    raise SnapshotError(f"base object must be a regular file: {entry}")
                status = entry.stat()
                objects[relative] = _Object(
                    path=relative,
                    size=status.st_size,
                    sha256=None,
                    mtime_ns=status.st_mtime_ns,
                    source_index=None,
                )
                if len(objects) > MAX_BASE_OBJECTS:
                    raise SnapshotError("base_objects contains too many files")
    return root, objects


def _file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as source:
        while True:
            chunk = source.read(1024 * 1024)
            if not chunk:
                return digest.hexdigest()
            digest.update(chunk)


def _object_zip_signature(stream: BinaryIO, expected: _Object) -> tuple:
    """只允许 sccache 对象的 ZIP 成员排列不同，不放宽内容或元数据校验。"""
    with tempfile.TemporaryFile() as temporary:
        digest = hashlib.sha256()
        total = 0
        while chunk := stream.read(1024 * 1024):
            total += len(chunk)
            if total > expected.size:
                raise SnapshotError("cache object grew while comparing ZIP contents")
            digest.update(chunk)
            temporary.write(chunk)
        if (total, digest.hexdigest()) != (expected.size, expected.sha256):
            raise SnapshotError("cache object checksum changed while comparing ZIP contents")
        temporary.seek(0)
        with zipfile.ZipFile(temporary) as bundle:
            infos = bundle.infolist()
            if (not infos or len(infos) > MAX_OBJECTS
                    or len({info.filename for info in infos}) != len(infos)
                    or sum(info.file_size for info in infos) > expected.size):
                raise SnapshotError("invalid sccache ZIP member set")
            members = []
            for info in sorted(infos, key=lambda entry: entry.filename):
                _safe_zip_path(info.filename)
                mode = info.external_attr >> 16
                if (info.is_dir() or info.flag_bits & 1 or stat.S_ISLNK(mode)
                        or stat.S_IFMT(mode) not in (0, stat.S_IFREG)
                        or info.compress_type != zipfile.ZIP_STORED):
                    raise SnapshotError("unsupported sccache ZIP member")
                with bundle.open(info) as contents:
                    size, checksum = _hash_stream(contents, info.file_size)
                members.append((
                    info.filename, size, checksum, info.CRC, info.compress_size,
                    info.compress_type, info.date_time, info.flag_bits,
                    info.external_attr, info.internal_attr, info.create_system,
                    info.create_version, info.extract_version, info.reserved,
                    info.extra, info.comment,
                ))
            return bundle.comment, tuple(members)


def _equivalent_delta_paths(archives: Sequence[_Archive]) -> set[str]:
    variants: dict[str, set[tuple[int, str | None]]] = {}
    for archive in archives:
        for obj in archive.objects:
            variants.setdefault(obj.path, set()).add((obj.size, obj.sha256))
    conflicts = {path for path, versions in variants.items() if len(versions) > 1}
    if not conflicts:
        return conflicts
    for path in conflicts:
        if len({size for size, _ in variants[path]}) != 1:
            raise SnapshotError(f"conflicting content for duplicate object: {path}")
    signatures = {}
    # 每份归档最多额外顺序读取一次，仅将冲突对象暂存到磁盘供 ZIP 随机读取。
    for archive in archives:
        wanted = {obj.path: obj for obj in archive.objects if obj.path in conflicts}
        if not wanted:
            continue
        if _archive_signature(archive.input.archive) != archive.signature:
            raise SnapshotError("artifact archive changed while comparing cache objects")
        bundle, raw, tar = _tar_stream(archive.input.archive)
        try:
            for member in tar:
                expected = wanted.get(member.name)
                if expected is None:
                    continue
                if not member.isreg() or member.size != expected.size:
                    raise SnapshotError("cache object changed while comparing ZIP contents")
                stream = tar.extractfile(member)
                if stream is None:
                    raise SnapshotError("cache object is missing while comparing ZIP contents")
                try:
                    with stream:
                        signature = _object_zip_signature(stream, expected)
                except (SnapshotError, zipfile.BadZipFile, NotImplementedError) as error:
                    raise SnapshotError(f"conflicting content for duplicate object: {member.name}") from error
                if member.name in signatures and signatures[member.name] != signature:
                    raise SnapshotError(f"conflicting content for duplicate object: {member.name}")
                signatures[member.name] = signature
                del wanted[member.name]
            if wanted:
                raise SnapshotError("cache objects disappeared while comparing ZIP contents")
        finally:
            tar.close()
            raw.close()
            bundle.close()
    return conflicts


def _select_objects(
    archives: Sequence[_Archive],
    base_root: Path,
    base: Mapping[str, _Object],
    maximum: int,
) -> tuple[_Object, ...]:
    merged = dict(base)
    equivalent_paths = _equivalent_delta_paths(archives)
    delta_paths = {obj.path for archive in archives for obj in archive.objects}
    touched_paths = {touch.path for archive in archives for touch in archive.touched}
    overlap = delta_paths & touched_paths
    if overlap:
        raise SnapshotError(f"object appears as both delta and touched: {min(overlap)}")
    for archive in archives:
        for touch in archive.touched:
            current = merged.get(touch.path)
            if current is None or current.source_index is not None:
                raise SnapshotError(f"touched object is absent from inherited base: {touch.path}")
            if touch.mtime_ns > current.mtime_ns:
                merged[touch.path] = _Object(
                    path=current.path,
                    size=current.size,
                    sha256=current.sha256,
                    mtime_ns=touch.mtime_ns,
                    source_index=None,
                )

    base_hashes: dict[str, str] = {}
    for archive in archives:
        for candidate in archive.objects:
            current = merged.get(candidate.path)
            if current is None:
                merged[candidate.path] = candidate
            elif current.source_index is None:
                if current.size != candidate.size:
                    raise SnapshotError(f"conflicting content for duplicate object: {candidate.path}")
                checksum = base_hashes.get(candidate.path)
                if checksum is None:
                    checksum = _file_sha256(
                        base_root.joinpath(*PurePosixPath(candidate.path).parts[1:])
                    )
                    base_hashes[candidate.path] = checksum
                if checksum != candidate.sha256:
                    raise SnapshotError(f"conflicting content for duplicate object: {candidate.path}")
                merged[candidate.path] = _Object(
                    path=current.path,
                    size=current.size,
                    sha256=checksum,
                    mtime_ns=max(current.mtime_ns, candidate.mtime_ns),
                    source_index=None,
                )
            elif ((current.size, current.sha256) != (candidate.size, candidate.sha256)
                  and candidate.path not in equivalent_paths):
                raise SnapshotError(f"conflicting content for duplicate object: {candidate.path}")
            elif candidate.mtime_ns > current.mtime_ns:
                merged[candidate.path] = _Object(
                    path=current.path,
                    size=current.size,
                    sha256=current.sha256,
                    mtime_ns=candidate.mtime_ns,
                    source_index=current.source_index,
                )

    selected: list[_Object] = []
    total = 0
    for candidate in sorted(merged.values(), key=lambda obj: (-obj.mtime_ns, obj.path)):
        if candidate.size <= maximum - total:
            selected.append(candidate)
            total += candidate.size
    return tuple(selected)


def _copy_selected(archive: _Archive, selected: Mapping[str, _Object], root: Path) -> None:
    if _archive_signature(Path(archive.input.archive)) != archive.signature:
        raise SnapshotError(f"artifact archive changed while merging: {archive.input.archive}")
    bundle, raw, tar = _tar_stream(Path(archive.input.archive))
    remaining = set(selected)
    try:
        try:
            for member in tar:
                target_object = selected.get(member.name)
                if target_object is None:
                    continue
                if not member.isreg() or member.size != target_object.size:
                    raise SnapshotError(f"selected object changed while merging: {member.name}")
                source = tar.extractfile(member)
                if source is None:
                    raise SnapshotError(f"cannot read selected object: {member.name}")
                destination = root.joinpath(*PurePosixPath(member.name).parts)
                destination.parent.mkdir(parents=True, exist_ok=True)
                digest = hashlib.sha256()
                written = 0
                with destination.open("xb") as output:
                    while True:
                        chunk = source.read(1024 * 1024)
                        if not chunk:
                            break
                        written += len(chunk)
                        if written > target_object.size:
                            raise SnapshotError(f"selected object grew while merging: {member.name}")
                        digest.update(chunk)
                        output.write(chunk)
                if written != target_object.size or digest.hexdigest() != target_object.sha256:
                    raise SnapshotError(f"selected object checksum changed while merging: {member.name}")
                os.utime(destination, ns=(target_object.mtime_ns, target_object.mtime_ns))
                remaining.remove(member.name)
        except (OSError, tarfile.TarError, zipfile.BadZipFile) as error:
            raise SnapshotError(f"cannot copy selected objects from {archive.input.archive}") from error
    finally:
        tar.close()
        raw.close()
        bundle.close()
    if remaining:
        raise SnapshotError(f"selected objects disappeared from {archive.input.archive}")


def _copy_base(base_root: Path, selected: Mapping[str, _Object], root: Path) -> None:
    for relative, obj in selected.items():
        source = base_root.joinpath(*PurePosixPath(relative).parts[1:])
        if source.is_symlink() or not source.is_file() or source.stat().st_size != obj.size:
            raise SnapshotError(f"inherited base object changed while merging: {relative}")
        destination = root.joinpath(*PurePosixPath(relative).parts)
        destination.parent.mkdir(parents=True, exist_ok=True)
        with source.open("rb") as input_stream, destination.open("xb") as output_stream:
            shutil.copyfileobj(input_stream, output_stream, length=1024 * 1024)
        if destination.stat().st_size != obj.size:
            raise SnapshotError(f"inherited base object was truncated while merging: {relative}")
        os.utime(destination, ns=(obj.mtime_ns, obj.mtime_ns))


def _write_metadata(path: Path, value: str) -> None:
    path.write_text(value if value.endswith("\n") else value + "\n", encoding="utf-8")


def _validated_output(output_dir: os.PathLike[str] | str) -> tuple[Path, Path]:
    output = Path(output_dir)
    if not output.is_absolute():
        raise SnapshotError("output_dir must be absolute")
    if output.exists() or output.is_symlink():
        raise SnapshotError("output_dir must not already exist")
    try:
        parent = output.parent.resolve(strict=True)
    except OSError as error:
        raise SnapshotError("output_dir parent must already exist") from error
    if output.parent.is_symlink() or not parent.is_dir():
        raise SnapshotError("output_dir parent must be a real directory")
    if output.parent != parent:
        raise SnapshotError("output_dir parent must use its canonical absolute path")
    return output, parent


def merge_snapshots(
    inputs: Sequence[ArtifactInput],
    output_dir: os.PathLike[str] | str,
    *,
    base_objects: os.PathLike[str] | str,
    expected_inherited_source_sha: str,
    max_combined_bytes: int = DEFAULT_MAX_COMBINED_BYTES,
) -> MergeResult:
    """Validate artifacts and atomically create a bounded sccache object snapshot."""
    if not inputs:
        raise SnapshotError("at least one artifact is required")
    if not _is_int(max_combined_bytes) or max_combined_bytes < 0:
        raise SnapshotError("max_combined_bytes must be a non-negative integer")
    _require_string(expected_inherited_source_sha, "expected_inherited_source_sha")
    output, parent = _validated_output(output_dir)

    base_root, base = _validated_base_objects(base_objects)
    archives = tuple(
        _validate_archive(item, index, expected_inherited_source_sha)
        for index, item in enumerate(inputs)
    )
    source_sha, rustc, workspace, sccache_version, base_image = _consistent_metadata(archives)
    selected = _select_objects(archives, base_root, base, max_combined_bytes)
    by_source: dict[int, dict[str, _Object]] = {}
    from_base: dict[str, _Object] = {}
    for obj in selected:
        if obj.source_index is None:
            from_base[obj.path] = obj
        else:
            by_source.setdefault(obj.source_index, {})[obj.path] = obj

    temporary = Path(tempfile.mkdtemp(prefix=TEMP_PREFIX, dir=parent))
    try:
        snapshot = temporary / "snapshot"
        (snapshot / "objects").mkdir(parents=True)
        _copy_base(base_root, from_base, snapshot)
        for source_index, objects in by_source.items():
            _copy_selected(archives[source_index], objects, snapshot)
        _write_metadata(snapshot / "rustc.txt", rustc)
        _write_metadata(snapshot / "workspace.txt", workspace)
        _write_metadata(snapshot / "source-commit.txt", source_sha)
        if base_image is not None:
            _write_metadata(snapshot / "base-image.txt", base_image)
        snapshot.replace(output)
    finally:
        if temporary.parent != parent or not temporary.name.startswith(TEMP_PREFIX):
            raise RuntimeError("refusing to clean an unexpected temporary directory")
        shutil.rmtree(temporary, ignore_errors=False)

    return MergeResult(
        object_count=len(selected),
        total_bytes=sum(obj.size for obj in selected),
        source_sha=source_sha,
        rustc=rustc,
        workspace=workspace,
        base_image=base_image,
        sccache_version=sccache_version,
    )