#!/usr/bin/env python3
"""Measure inotify hints on fresh temporary files; no external dependencies."""
import argparse
import ctypes
from datetime import datetime, timezone
import hashlib
import json
import mmap
import os
from pathlib import Path
import platform
import select
import struct
import subprocess
import tempfile

IN_MODIFY = 0x2
IN_CLOSE_WRITE = 0x8
IN_Q_OVERFLOW = 0x4000
HEADER = struct.Struct("=iIII")
CASES = ("write-one", "write-burst", "write-drained", "mapped-shared",
         "mapped-private", "no-write", "parent-watch", "child-watch")


def parse_events(data):
    events = []
    offset = 0
    while offset < len(data):
        if len(data) - offset < HEADER.size:
            raise ValueError("Truncated inotify header")
        wd, mask, cookie, length = HEADER.unpack_from(data, offset)
        offset += HEADER.size
        if length > len(data) - offset:
            raise ValueError("Truncated inotify name")
        name = data[offset:offset + length].split(b"\0", 1)[0]
        offset += length
        if mask & IN_Q_OVERFLOW:
            raise RuntimeError("Queue overflow invalidates this observation")
        events.append({"wd": wd, "mask": mask, "cookie": cookie,
                       "name": os.fsdecode(name)})
    return events


class Watch:
    def __init__(self, path):
        libc = ctypes.CDLL(None, use_errno=True)
        libc.inotify_init1.argtypes = [ctypes.c_int]
        libc.inotify_init1.restype = ctypes.c_int
        libc.inotify_add_watch.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_uint32]
        libc.inotify_add_watch.restype = ctypes.c_int
        self.fd = libc.inotify_init1(os.O_NONBLOCK | os.O_CLOEXEC)
        if self.fd < 0:
            raise OSError(ctypes.get_errno(), "inotify_init1")
        if libc.inotify_add_watch(self.fd, os.fsencode(path),
                                  IN_MODIFY | IN_CLOSE_WRITE) < 0:
            error = ctypes.get_errno()
            os.close(self.fd)
            raise OSError(error, "inotify_add_watch")

    def drain(self, wait=False):
        if wait:
            select.select([self.fd], [], [], 0.05)
        events = []
        while True:
            try:
                data = os.read(self.fd, 65536)
            except BlockingIOError:
                return events
            if not data:
                raise RuntimeError("Unexpected inotify EOF")
            events.extend(parse_events(data))

    def close(self):
        os.close(self.fd)


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


def one_case(directory, case):
    with tempfile.TemporaryDirectory(prefix="hesela-inotify-", dir=directory) as tmp:
        root = Path(tmp)
        child = root / "child"
        child.mkdir()
        path = child / "payload"
        original = b"A" * mmap.PAGESIZE
        path.write_bytes(original)
        fd = os.open(path, os.O_RDWR)
        watch = None
        try:
            target = root if case == "parent-watch" else child if case == "child-watch" else path
            watch = Watch(target)
            before_close = []
            expected = bytearray(original)
            calls = 0
            if case in ("write-one", "parent-watch", "child-watch", "write-burst", "write-drained"):
                calls = 32 if case in ("write-burst", "write-drained") else 1
                for offset in range(calls):
                    if os.pwrite(fd, b"B", offset) != 1:
                        raise RuntimeError("Short pwrite")
                    expected[offset] = ord("B")
                    if case == "write-drained":
                        before_close.extend(watch.drain())
            elif case in ("mapped-shared", "mapped-private"):
                flags = mmap.MAP_SHARED if case == "mapped-shared" else mmap.MAP_PRIVATE
                with mmap.mmap(fd, len(original), flags=flags,
                               prot=mmap.PROT_READ | mmap.PROT_WRITE) as mapping:
                    mapping[:32] = b"B" * 32
                    mapping.flush()
                    if mapping[:32] != b"B" * 32:
                        raise RuntimeError("Mapped store was not observed")
                if case == "mapped-shared":
                    expected[:32] = b"B" * 32
            elif case != "no-write":
                raise ValueError(case)
            actual = os.pread(fd, len(original) + 1, 0)
            if actual != bytes(expected):
                raise RuntimeError("Full payload comparison failed: " + case)
            before_close.extend(watch.drain(wait=True))
            os.close(fd)
            fd = -1
            after_close = watch.drain(wait=True)
            return {
                "case": case, "pwriteCalls": calls, "fileBytes": len(actual),
                "changedBytes": sum(a != b for a, b in zip(original, actual)),
                "originalSha256": digest(original), "finalSha256": digest(actual),
                "modifyBeforeClose": sum(bool(e["mask"] & IN_MODIFY) for e in before_close),
                "closeWriteBeforeClose": sum(bool(e["mask"] & IN_CLOSE_WRITE) for e in before_close),
                "closeWriteAfterClose": sum(bool(e["mask"] & IN_CLOSE_WRITE) for e in after_close),
                "eventsBeforeClose": before_close, "eventsAfterClose": after_close,
            }
        finally:
            if fd >= 0:
                os.close(fd)
            if watch is not None:
                watch.close()


def run(directories, repetitions):
    if repetitions < 1:
        raise ValueError("repetitions must be positive")
    results = []
    for directory in directories:
        directory = Path(directory).resolve(strict=True)
        fs = subprocess.check_output(
            ["findmnt", "-n", "-o", "FSTYPE", "--target", str(directory)], text=True).strip()
        results.append({"filesystem": fs, "directory": str(directory),
                        "observations": [[one_case(directory, case) for case in CASES]
                                         for _ in range(repetitions)]})
    return {
        "schemaVersion": 1, "generatedAt": datetime.now(timezone.utc).isoformat(),
        "repetitions": repetitions, "casesPerRepetition": len(CASES),
        "environment": {"kernel": platform.release(), "python": platform.python_version(),
                        "libc": list(platform.libc_ver()), "pageBytes": mmap.PAGESIZE},
        "watchMask": IN_MODIFY | IN_CLOSE_WRITE, "quietWaitSeconds": 0.05,
        "sourceSha256": digest(Path(__file__).read_bytes()),
        "scope": "Sequential fresh-file API observations on one host; not power-loss or backup-product tests.",
        "runs": results,
    }


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--directory", action="append", required=True)
    parser.add_argument("--repetitions", type=int, default=5)
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()
    payload = json.dumps(run(args.directory, args.repetitions), indent=2) + "\n"
    if args.output:
        args.output.write_text(payload)
    else:
        print(payload, end="")