"""Deterministic synthetic repository generator (git fast-import stream).

One ``GenSpec`` describes a repository shape along the git-sizer dimensions

(files, bytes, path depth, history length, refs, big blobs). ``generate()``

writes ONE fast-import stream into ONE ``git fast-import`` process, so the XL

shapes (1M files, 100k commits, 10k refs, multi-GiB blobs) are generated in

minutes, not hours, and without a working tree.

Determinism contract (tested): identical spec -> identical bytes in the

stream -> identical commit ids and ref values on any machine and any Git

that speaks fast-import. Everything that could vary is pinned: identity,

timestamps (fixed epoch + 60 s per commit), the RNG (``random.Random`` seeded

from the spec seed), path layout, content, and ref names. ``refs_sha256``

(sha256 over sorted ``<oid> <ref>`` lines) is the fixture identity recorded

with every row; ``spec_sha256`` identifies the recipe.

Content model:

- text files: unique per (file, revision) so blobs never dedupe by accident;

  a header line plus filler words sliced from a seeded 64 KiB pool;

- big binaries ("sparse where possible"): every 1 MiB segment is a 4 KiB

  seeded random block followed by zeros. Git packs them small, a chunking

  store dedupes the zero runs, and a checkout still writes the full size, so

  the working tree pays the real byte count while the fixture store does not.

Path model: file i lives in leaf directory i // fanout. Leaf directories are

addressed by digits over ``depth - 1`` directory levels (base ``fanout`` when

that fits, otherwise the smallest base that addresses every leaf). Unused

high digits are a zero-padded spine, so ``depth`` 64 yields a long single-child

chain above a branching bottom: every tracked file sits at exactly ``depth``

path components (git-sizer's "maximum path depth").

"""

from __future__ import annotations

import hashlib

import json

import math

import os

import random

import subprocess

import time

from dataclasses import asdict, dataclass, field

from pathlib import Path

from typing import Any, BinaryIO, Dict, List, Optional, Tuple

GENERATOR_VERSION = "scale-gen-2026-09-28.1"

IDENTITY = b"Oak Bench <[email protected]>"

EPOCH = 1_700_000_000

SEGMENT = 1 << 20

SEGMENT_RANDOM = 4096

POOL_BYTES = 64 * 1024

FLUSH_BYTES = 4 << 20

MANIFEST_NAME = "GENERATOR_MANIFEST.json"

PROBE_EVERY = 7  # the probe path is rewritten by every 7th history commit

@dataclass(frozen=True)

class GenSpec:

    """A repository recipe. Field names follow git-sizer's dimensions."""

    name: str

    seed: str

    files: int = 100

    file_bytes: int = 256

    depth: int = 3

    fanout: int = 32

    commits: int = 1

    touches_per_commit: int = 4

    branches: int = 0

    branch_commit_files: int = 2

    tags: int = 0

    binaries: int = 0

    binary_bytes: int = 0

    binary_revisions: int = 0

    ignore_patterns: Tuple[str, ...] = ()

    conflict_files: int = 0

    notes: str = ""

    def validate(self) -> None:

        if self.files < 1:

            raise ValueError("files must be >= 1")

        if self.depth < 1:

            raise ValueError("depth must be >= 1 (1 = files at the root)")

        if self.fanout < 2:

            raise ValueError("fanout must be >= 2")

        if self.commits < 1:

            raise ValueError("commits must be >= 1")

        if self.file_bytes < 32:

            raise ValueError("file_bytes must be >= 32 (room for the unique header line)")

        if self.binary_revisions and not self.binaries:

            raise ValueError("binary_revisions needs binaries >= 1")

        if self.binary_revisions > self.commits:

            raise ValueError("binary_revisions cannot exceed commits")

        if self.conflict_files > self.files:

            raise ValueError("conflict_files cannot exceed files")

        if self.branch_commit_files > self.files:

            raise ValueError("branch_commit_files cannot exceed files")

    def to_json(self) -> Dict[str, Any]:

        data = asdict(self)

        data["ignore_patterns"] = list(self.ignore_patterns)

        return data

    def spec_sha256(self) -> str:

        payload = json.dumps({"generator_version": GENERATOR_VERSION, "spec": self.to_json()},

                             sort_keys=True, separators=(",", ":"))

        return hashlib.sha256(payload.encode()).hexdigest()

    def estimated_bytes(self) -> Dict[str, int]:

        """Rough working-tree and store sizes, for disk budgeting before generation."""

        text = self.files * self.file_bytes

        binary = self.binaries * self.binary_bytes

        # APFS/ext4 allocate at least one 4 KiB block per small file.

        worktree = self.files * max(4096, self.file_bytes) + binary

        store = text + self.commits * self.touches_per_commit * self.file_bytes \

            + max(1, self.binary_revisions) * self.binaries * (self.binary_bytes // SEGMENT + 1) * SEGMENT_RANDOM

        return {"worktree_bytes": worktree, "store_bytes_estimate": store}

# ---------------------------------------------------------------------------

# Layout and content (pure functions: also used by oracles)

# ---------------------------------------------------------------------------

def layout(spec: GenSpec) -> Tuple[int, int]:

    """(directory levels, digit base) for the spec's path layout."""

    dir_levels = spec.depth - 1

    if dir_levels == 0:

        return 0, spec.fanout

    leaves = max(1, math.ceil(spec.files / spec.fanout))

    needed = 1 if leaves <= 1 else math.ceil(math.log(leaves, spec.fanout) - 1e-12)

    if needed <= dir_levels:

        return dir_levels, spec.fanout

    base = 2

    while base ** dir_levels < leaves:

        base += 1

    return dir_levels, base

def file_path(spec: GenSpec, index: int, _cache: Optional[Tuple[int, int]] = None) -> str:

    dir_levels, base = _cache or layout(spec)

    name = f"f{index}.txt"

    if dir_levels == 0:

        return name

    leaf = index // spec.fanout

    digits = []

    for _ in range(dir_levels):

        digits.append(leaf % base)

        leaf //= base

    parts = [f"d{digit}" for digit in reversed(digits)]

    parts.append(name)

    return "/".join(parts)

def branch_name(number: int) -> str:

    return f"b{number:05d}"

def branch_changes(spec: GenSpec, number: int, _cache: Optional[Tuple[int, int]] = None) -> List[Tuple[int, str]]:

    """(file index, path) rewritten by side branch ``number`` (content revision = branch name)."""

    cache = _cache or layout(spec)

    out: List[Tuple[int, str]] = []

    seen = set()

    for j in range(spec.branch_commit_files):

        index = (number * 7919 + j * 104729 + 1) % spec.files

        if index in seen:

            continue

        seen.add(index)

        out.append((index, file_path(spec, index, cache)))

    return out

def binary_path(index: int) -> str:

    return f"bin/blob{index}.bin"

def _pool(seed: str) -> bytes:

    rng = random.Random(f"{seed}:pool")

    words = [b"alpha", b"branch", b"commit", b"delta", b"engine", b"field", b"graph", b"hash",

             b"index", b"journal", b"kernel", b"ledger", b"merge", b"node", b"oracle", b"patch",

             b"query", b"record", b"stream", b"tree", b"update", b"vector", b"worker", b"zone"]

    out = bytearray()

    column = 0

    while len(out) < POOL_BYTES + 8192:

        word = words[rng.randrange(len(words))]

        out += word

        column += len(word) + 1

        if column > 60:

            out += b"\n"

            column = 0

        else:

            out += b" "

    return bytes(out)

class ContentFactory:

    """Byte-exact content for any (file, revision); oracles call the same code."""

    def __init__(self, spec: GenSpec) -> None:

        self.spec = spec

        self.pool = _pool(spec.seed)

    def text(self, index: int, revision: str) -> bytes:

        header = f"{self.spec.name}:{index}:{revision}\n".encode()

        size = max(self.spec.file_bytes, len(header) + 1)

        filler = size - len(header)

        # Cheap deterministic offset mixing (no per-file hashing).

        offset = ((index * 2654435761) ^ (len(revision) * 40503) ^ sum(revision.encode())) % POOL_BYTES

        body = self.pool[offset:offset + filler]

        if len(body) < filler:

            body = (body + self.pool)[:filler]

        if body and not body.endswith(b"\n"):

            body = body[:-1] + b"\n"

        return header + body

    def binary_segments(self, index: int, revision: int):

        """Yield the binary's bytes in <= 1 MiB pieces (never materialized whole)."""

        total = self.spec.binary_bytes

        segment_no = 0

        written = 0

        while written < total:

            rng = random.Random(f"{self.spec.seed}:bin:{index}:{revision}:{segment_no}")

            head = rng.getrandbits(SEGMENT_RANDOM * 8).to_bytes(SEGMENT_RANDOM, "little")

            size = min(SEGMENT, total - written)

            if size <= SEGMENT_RANDOM:

                piece = head[:size]

                yield piece

            else:

                yield head

                zeros = size - SEGMENT_RANDOM

                while zeros > 0:

                    step = min(zeros, 1 << 18)

                    yield bytes(step)

                    zeros -= step

            written += size

            segment_no += 1

def binary_sha1_and_sha256(factory: ContentFactory, index: int, revision: int) -> Tuple[str, str]:

    """(git blob id, sha256) of a generated binary, streamed."""

    total = factory.spec.binary_bytes

    git = hashlib.sha1(f"blob {total}\0".encode())

    plain = hashlib.sha256()

    for piece in factory.binary_segments(index, revision):

        git.update(piece)

        plain.update(piece)

    return git.hexdigest(), plain.hexdigest()

def git_blob_id(data: bytes) -> str:

    return hashlib.sha1(b"blob %d\0" % len(data) + data).hexdigest()

# ---------------------------------------------------------------------------

# Stream writer

# ---------------------------------------------------------------------------

class _Stream:

    def __init__(self, sink: BinaryIO) -> None:

        self.sink = sink

        self.buffer: List[bytes] = []

        self.size = 0

        self.total = 0

    def write(self, data: bytes) -> None:

        self.buffer.append(data)

        self.size += len(data)

        if self.size >= FLUSH_BYTES:

            self.flush()

    def flush(self) -> None:

        if self.buffer:

            blob = b"".join(self.buffer)

            self.sink.write(blob)

            self.total += len(blob)

            self.buffer = []

            self.size = 0

    def data(self, payload: bytes) -> None:

        self.write(b"data %d\n" % len(payload))

        self.write(payload)

        self.write(b"\n")

def _commit_header(stream: _Stream, ref: bytes, mark: int, when: int, message: bytes,

                   parent: Optional[int]) -> None:

    stream.write(b"commit " + ref + b"\n")

    stream.write(b"mark :%d\n" % mark)

    stamp = b"%d +0000" % when

    stream.write(b"author " + IDENTITY + b" " + stamp + b"\n")

    stream.write(b"committer " + IDENTITY + b" " + stamp + b"\n")

    stream.data(message)

    if parent is not None:

        stream.write(b"from :%d\n" % parent)

def write_stream(spec: GenSpec, sink: BinaryIO) -> Dict[str, Any]:

    """Write the full fast-import stream for ``spec``; return the manifest facts."""

    spec.validate()

    rng = random.Random(f"{spec.seed}:history")

    factory = ContentFactory(spec)

    cache = layout(spec)

    stream = _Stream(sink)

    stream.write(b"feature done\n")

    main = b"refs/heads/main"

    probe_index = 0

    probe_path = file_path(spec, probe_index, cache)

    probe_commits = 1  # the initial commit adds it

    touched_paths_last: List[str] = []

    mark = 0

    tick = 0

    # Commit 1: the full tree.

    mark += 1

    tick += 1

    _commit_header(stream, main, mark, EPOCH + 60 * tick, b"c1 initial tree\n", None)

    if spec.ignore_patterns:

        stream.write(b"M 100644 inline .gitignore\n")

        stream.data(("\n".join(spec.ignore_patterns) + "\n").encode())

    for index in range(spec.files):

        stream.write(b"M 100644 inline " + file_path(spec, index, cache).encode() + b"\n")

        stream.data(factory.text(index, "r1"))

    for index in range(spec.binaries):

        stream.write(b"M 100644 inline " + binary_path(index).encode() + b"\n")

        stream.write(b"data %d\n" % spec.binary_bytes)

        for piece in factory.binary_segments(index, 1):

            stream.write(piece)

        stream.write(b"\n")

    commit_marks = [mark]

    # Commits 2..N: seeded touch sets; every PROBE_EVERY-th also rewrites the probe path.

    for number in range(2, spec.commits + 1):

        mark += 1

        tick += 1

        _commit_header(stream, main, mark, EPOCH + 60 * tick, b"c%d\n" % number, mark - 1)

        chosen = set()

        for _ in range(min(spec.touches_per_commit, spec.files)):

            chosen.add(rng.randrange(spec.files))

        if number % PROBE_EVERY == 0:

            chosen.add(probe_index)

        elif probe_index in chosen:

            chosen.discard(probe_index)

        if probe_index in chosen:

            probe_commits += 1

        touched_paths_last = []

        for index in sorted(chosen):

            path = file_path(spec, index, cache)

            touched_paths_last.append(path)

            stream.write(b"M 100644 inline " + path.encode() + b"\n")

            stream.data(factory.text(index, f"r{number}"))

        if number <= spec.binary_revisions:

            stream.write(b"M 100644 inline " + binary_path(0).encode() + b"\n")

            stream.write(b"data %d\n" % spec.binary_bytes)

            for piece in factory.binary_segments(0, number):

                stream.write(piece)

            stream.write(b"\n")

        commit_marks.append(mark)

    tip = mark

    # Branches: each forks from the tip with one commit of branch_commit_files rewrites.

    branch_sample: Dict[str, Any] = {}

    for number in range(spec.branches):

        name = branch_name(number)

        mark += 1

        tick += 1

        _commit_header(stream, b"refs/heads/" + name.encode(), mark, EPOCH + 60 * tick,

                       b"branch " + name.encode() + b"\n", tip)

        changed = []

        for index, path in branch_changes(spec, number, cache):

            changed.append(path)

            stream.write(b"M 100644 inline " + path.encode() + b"\n")

            stream.data(factory.text(index, name))

        if number == 0:

            branch_sample = {"name": name, "changed_paths": sorted(changed)}

    # Conflict shape: two branches rewrite the same files differently.

    conflict: Dict[str, Any] = {}

    if spec.conflict_files:

        conflict_paths = [file_path(spec, i, cache) for i in range(spec.conflict_files)]

        for side in ("bench-ours", "bench-theirs"):

            mark += 1

            tick += 1

            _commit_header(stream, b"refs/heads/" + side.encode(), mark, EPOCH + 60 * tick,

                           side.encode() + b"\n", tip)

            for i in range(spec.conflict_files):

                stream.write(b"M 100644 inline " + conflict_paths[i].encode() + b"\n")

                stream.data(factory.text(i, side))

        conflict = {"ours": "bench-ours", "theirs": "bench-theirs", "paths": len(conflict_paths),

                    "paths_sample": conflict_paths[:20]}

    # Lightweight tags at seeded history commits.

    for number in range(spec.tags):

        target = commit_marks[rng.randrange(len(commit_marks))]

        stream.write(b"reset refs/tags/t%05d\nfrom :%d\n\n" % (number, target))

    stream.write(b"done\n")

    stream.flush()

    return {

        "main_commits": spec.commits,

        "total_commits": mark,

        "probe_path": probe_path,

        "probe_path_commit_count": probe_commits,

        "last_commit_paths": sorted(touched_paths_last),

        "branch_sample": branch_sample,

        "conflict": conflict,

        "binary_paths": [binary_path(i) for i in range(spec.binaries)],

        "stream_bytes": stream.total,

    }

# ---------------------------------------------------------------------------

# Generation

# ---------------------------------------------------------------------------

def git_env() -> Dict[str, str]:

    env = {key: value for key, value in os.environ.items() if not key.startswith("GIT_")}

    env.update(GIT_CONFIG_NOSYSTEM="1", GIT_CONFIG_GLOBAL=os.devnull, GIT_TERMINAL_PROMPT="0",

               LC_ALL="C", TZ="UTC")

    return env

def refs_digest(git_bin: str, repo: Path) -> Tuple[str, int]:

    out = subprocess.run([git_bin, "-C", str(repo), "for-each-ref", "--format=%(objectname) %(refname)"],

                         check=True, capture_output=True, env=git_env()).stdout

    lines = sorted(line for line in out.splitlines() if line)

    return hashlib.sha256(b"\n".join(lines) + b"\n").hexdigest(), len(lines)

def generate(spec: GenSpec, out_dir: Path, git_bin: str = "git") -> Dict[str, Any]:

    """Generate ``out_dir/src.git`` (bare) plus ``GENERATOR_MANIFEST.json``.

    Idempotent: an existing fixture whose manifest names the same spec_sha256

    and still resolves to the recorded refs digest is reused, not rebuilt.

    """

    spec.validate()

    out_dir = Path(out_dir)

    repo = out_dir / "src.git"

    manifest_path = out_dir / MANIFEST_NAME

    if manifest_path.exists() and repo.exists():

        existing = json.loads(manifest_path.read_text())

        if existing.get("spec_sha256") == spec.spec_sha256():

            digest, _ = refs_digest(git_bin, repo)

            if digest == existing.get("refs_sha256"):

                existing["reused"] = True

                return existing

    if repo.exists():

        import shutil

        shutil.rmtree(repo)

    out_dir.mkdir(parents=True, exist_ok=True)

    env = git_env()

    subprocess.run([git_bin, "init", "-q", "--bare", "-b", "main", str(repo)], check=True, env=env)

    for key, value in (("gc.auto", "0"), ("core.bigFileThreshold", "512m")):

        subprocess.run([git_bin, "-C", str(repo), "config", key, value], check=True, env=env)

    started = time.perf_counter()

    proc = subprocess.Popen([git_bin, "-C", str(repo), "fast-import", "--quiet", "--done"],

                            stdin=subprocess.PIPE, env=env)

    assert proc.stdin is not None

    try:

        facts = write_stream(spec, proc.stdin)

    finally:

        proc.stdin.close()

    returncode = proc.wait()

    if returncode != 0:

        raise RuntimeError(f"git fast-import failed with exit {returncode}")

    elapsed = time.perf_counter() - started

    subprocess.run([git_bin, "-C", str(repo), "symbolic-ref", "HEAD", "refs/heads/main"], check=True, env=env)

    head = subprocess.run([git_bin, "-C", str(repo), "rev-parse", "refs/heads/main", "refs/heads/main^{tree}"],

                          check=True, capture_output=True, text=True, env=env).stdout.split()

    digest, ref_count = refs_digest(git_bin, repo)

    version = subprocess.run([git_bin, "--version"], capture_output=True, text=True, env=env).stdout.strip()

    manifest = {

        "schema_version": 1,

        "generator": "suites/scale/scalesuite/generator.py",

        "generator_version": GENERATOR_VERSION,

        "spec": spec.to_json(),

        "spec_sha256": spec.spec_sha256(),

        "refs_sha256": digest,

        "ref_count": ref_count,

        "head_commit": head[0],

        "head_tree": head[1],

        "generation_elapsed_s": round(elapsed, 3),

        "generation_git_version": version,

        "store_bytes": dir_bytes(repo),

        **facts,

    }

    manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")

    manifest["reused"] = False

    return manifest

def dir_bytes(root: Path) -> int:

    """Allocated bytes under root (st_blocks; sparse-aware), no symlink following."""

    total = 0

    for base, dirs, files in os.walk(root):

        for name in files:

            try:

                total += os.lstat(os.path.join(base, name)).st_blocks * 512

            except OSError:

                pass

    return total