#!/usr/bin/env python3
"""Native rsync payload accounting, not a performance benchmark. MIT license."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import platform
import re
import shutil
import subprocess
import tempfile
from datetime import datetime, timezone

SIZE = 262144
BLOCKS = (512, 4096)
MODES = {"local-default": [], "whole-file": ["--whole-file"],
         "delta": ["--no-whole-file"]}

def digest(data):
    return hashlib.sha256(data).hexdigest()

def fixtures():
    basis = hashlib.shake_256(b"hesela-rsync-delta-v1").digest(SIZE)
    p = SIZE // 2 + 17
    return basis, {
        "unchanged": basis,
        "prefix-insert": b"x" + basis,
        "middle-insert": basis[:p] + b"x" + basis[p:],
        "middle-delete": basis[:p] + basis[p + 1:],
        "middle-replace": basis[:p] + bytes([basis[p] ^ 255]) + basis[p + 1:],
        "all-replaced": hashlib.shake_256(b"hesela-rsync-delta-different-v1").digest(SIZE),
    }

def aligned_reuse(basis, source, size):
    blocks = {basis[i:i + size] for i in range(0, len(basis), size)}
    # Complete blocks only; this control is not rsync's sliding search.
    return sum(size for i in range(0, len(source) - size + 1, size)
               if source[i:i + size] in blocks)

def parse_stats(stdout):
    fields = {"literalBytes": "Literal data", "matchedBytes": "Matched data",
              "transferredFileBytes": "Total transferred file size",
              "sentBytes": "Total bytes sent", "receivedBytes": "Total bytes received"}
    result = {}
    for key, label in fields.items():
        match = re.search(r"^" + re.escape(label) + r": ([0-9,]+)(?: bytes)?$", stdout, re.M)
        if not match:
            raise ValueError("Missing statistic: " + label)
        result[key] = int(match.group(1).replace(",", ""))
    return result

def observe(scratch):
    binary = shutil.which("rsync")
    if not binary:
        raise RuntimeError("rsync must already be installed")
    basis, inputs = fixtures()
    version = subprocess.run([binary, "--version"], check=True, capture_output=True, text=True).stdout
    rows = []
    with tempfile.TemporaryDirectory(prefix="hesela-rsync-delta-", dir=scratch) as work:
        root = Path(work)
        env = {k: v for k, v in os.environ.items() if not k.startswith("RSYNC_")}
        env.update(LC_ALL="C", HOME=work, TZ="UTC")
        for name, source in inputs.items():
            for size in BLOCKS:
                for mode, flags in MODES.items():
                    src, dst = root / "source.bin", root / "destination.bin"
                    src.write_bytes(source)
                    dst.write_bytes(basis)
                    os.utime(src, ns=(1700000000000000000,) * 2)
                    os.utime(dst, ns=(1700000000000000000,) * 2)
                    args = ["-t", "--ignore-times", "--stats", "--checksum-choice=xxh128",
                            "--block-size=" + str(size), *flags, "--", "source.bin", "destination.bin"]
                    run = subprocess.run([binary, *args], cwd=root, env=env,
                                         capture_output=True, text=True, timeout=20)
                    if run.returncode:
                        raise RuntimeError(run.stderr)
                    stats = parse_stats(run.stdout)
                    output = dst.read_bytes()
                    assert src.read_bytes() == source, "source changed"
                    assert output == source, "complete byte comparison failed"
                    assert stats["literalBytes"] + stats["matchedBytes"] == len(source)
                    assert stats["transferredFileBytes"] == len(source)
                    rows.append({"fixture": name, "blockBytes": size, "mode": mode,
                                 "sourceBytes": len(source), "sourceSha256": digest(source),
                                 "destinationSha256": digest(output), "byteEqual": True,
                                 "alignedFullBlockReuseBytes": aligned_reuse(basis, source, size),
                                 "command": ["rsync", *args], "exitCode": run.returncode,
                                 **stats, "stdout": run.stdout, "stderr": run.stderr})
        fs_type = subprocess.run(["stat", "-f", "-c", "%T", work], check=True,
                                 capture_output=True, text=True).stdout.strip()
    return {
        "schemaVersion": 1, "id": "rsync-delta", "generatedAt": datetime.now(timezone.utc).isoformat(),
        "evidenceType": "Native software observations on deterministic synthetic fixtures",
        "scriptSha256": digest(Path(__file__).read_bytes()),
        "environment": {"python": platform.python_version(), "os": platform.platform(),
                        "rsyncVersion": version, "binarySha256": digest(Path(binary).read_bytes()),
                        "filesystem": fs_type},
        "method": {"basisBytes": SIZE, "basisSha256": digest(basis),
                   "basisRecipe": "SHAKE-256 of ASCII hesela-rsync-delta-v1, 262144 bytes",
                   "differentRecipe": "SHAKE-256 of ASCII hesela-rsync-delta-different-v1, 262144 bytes",
                   "middleEditOffset": SIZE // 2 + 17, "blockSizesBytes": list(BLOCKS),
                   "caseCount": len(rows), "transport": "local paths; delta mode explicitly selected",
                   "selection": "--ignore-times forces all fixtures into transfer, including unchanged",
                   "checksum": "xxh128 explicitly selected, not an authentication claim",
                   "oracle": "complete destination/source byte equality and source-unchanged check",
                   "alignedControl": "complete source blocks from offset zero compared to the basis block set",
                   "repetitions": 1, "randomSeed": None},
        "limits": ["Not a network throughput or disk-write benchmark; no timing claim.",
                   "Literal bytes exclude signatures, references, framing and transport overhead.",
                   "One installed rsync build is identified, not recommended.",
                   "No concurrent modification, crash, cold-media verification or restore test.",
                   "Constructed cases do not estimate workload-wide savings or failure probabilities.",
                   "Not a replication of SkySync and no adversarial checksum-collision test."],
        "cases": rows,
    }

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--scratch", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    result = observe(args.scratch)
    Path(args.output).write_text(json.dumps(result, indent=2) + "\n")
    print(json.dumps({"cases": len(result["cases"]), "byteEqual": all(r["byteEqual"] for r in result["cases"]),
                      "rsync": result["environment"]["rsyncVersion"].splitlines()[0]}))

if __name__ == "__main__":
    main()