"""Small privileged filesystem seam for durable runtime configuration publication."""

from __future__ import annotations

import fcntl
import hashlib
import json
import os
import stat
import subprocess
import sys
import tempfile
from pathlib import Path


def fail(msg: str) -> None:
    raise RuntimeError(msg)


def safe_id(v: str) -> bool:
    return bool(__import__("re").fullmatch(r"[a-z][a-z0-9-]{2,62}", v))


def safe_rev(v: str) -> bool:
    return bool(__import__("re").fullmatch(r"[0-9a-f]{40}", v))


def open_dir(parent: int | None, name: str, create: bool = False) -> int:
    """Open one directory component without following a replaced entry.

    The pre-open lstat and post-open fstat identity check is required on Darwin,
    where O_NOFOLLOW has historically been unavailable for directory openat.
    mkdir races are resolved by opening and validating the winner.
    """
    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | os.O_NOFOLLOW
    while True:
        try:
            entry = os.stat(name, dir_fd=parent, follow_symlinks=False)
            if stat.S_ISLNK(entry.st_mode):
                fail("runtime config directory is not trusted")
            fd = os.open(name, flags, dir_fd=parent)
            try:
                opened = os.fstat(fd)
                if (opened.st_dev != entry.st_dev or opened.st_ino != entry.st_ino
                        or not stat.S_ISDIR(opened.st_mode)):
                    fail("runtime config directory changed during open")
                return fd
            except BaseException:
                os.close(fd)
                raise
        except FileNotFoundError:
            if not create:
                raise
            try:
                os.mkdir(name, 0o700, dir_fd=parent)
            except FileExistsError:
                # Another publisher won creation.  Re-enter the identity-checked
                # open path instead of exposing EEXIST to the caller.
                continue
            if parent is not None:
                os.fsync(parent)
            # Re-open through the same no-follow and identity checks.
            continue


def checked_dir(fd: int, expected_mode: int = 0o700) -> None:
    s = os.fstat(fd)
    if (
        not stat.S_ISDIR(s.st_mode)
        or s.st_nlink < 1
        or stat.S_IMODE(s.st_mode) != expected_mode
        or s.st_uid != os.getuid()
    ):
        fail("runtime config directory is not trusted")


def walk(root: str, comps: list[str], create: bool = True) -> int:
    """Open an absolute path component-by-component without following symlinks.

    In particular, never use os.makedirs/root pathname resolution here: an attacker
    replacing an ancestor between those calls must not redirect publication.
    """
    if not os.path.isabs(root):
        fail("data root must be absolute")
    # macOS exposes temporary directories through the conventional /var and
    # /tmp symlinks. Resolve only these OS-owned aliases; workspace-owned
    # ancestors remain component checked and are never realpath-followed.
    if root == "/var" or root == "/tmp" or root.startswith(("/var/", "/tmp/")):
        root = "/private" + root
    parts = [part for part in Path(root).parts if part not in ("", "/")]
    if any(part in (".", "..") or "/" in part for part in parts + comps):
        fail("unsafe path component")
    fd = os.open("/", os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        all_components = [*parts, *comps]
        for index, component in enumerate(all_components):
            nxt = open_dir(fd, component, create)
            # Ancestors such as /var/folders are installation-owned and commonly
            # 0755; the trusted runtime root and every workspace child are private.
            info = os.fstat(nxt)
            if (not stat.S_ISDIR(info.st_mode) or info.st_nlink < 1
                    or (index >= len(parts) - 1 and (info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700))):
                os.close(nxt)
                fail("runtime config directory is not trusted")
            os.close(fd)
            fd = nxt
        return fd
    except BaseException:
        os.close(fd)
        raise


def read_regular(fd: int, mode: int, expected: bytes | None = None) -> os.stat_result:
    s = os.fstat(fd)
    if (
        not stat.S_ISREG(s.st_mode)
        or s.st_nlink != 1
        or stat.S_IMODE(s.st_mode) != mode
        or s.st_uid != os.getuid()
    ):
        fail("runtime config file is not trusted")
    if expected is not None:
        os.lseek(fd, 0, os.SEEK_SET)
        chunks = []
        while True:
            x = os.read(fd, 1024 * 1024)
            if not x:
                break
            chunks.append(x)
        if b"".join(chunks) != expected:
            fail("same-revision runtime configuration changed")
    return s


def write_all(fd: int, data: bytes) -> None:
    pos = 0
    while pos < len(data):
        n = os.write(fd, data[pos:])
        if n <= 0:
            fail("short runtime config write")
        pos += n


def read_all(fd: int, limit: int = 16 * 1024 * 1024) -> bytes:
    os.lseek(fd, 0, os.SEEK_SET)
    chunks: list[bytes] = []
    total = 0
    while True:
        chunk = os.read(fd, min(1024 * 1024, limit - total))
        if not chunk:
            return b"".join(chunks)
        chunks.append(chunk)
        total += len(chunk)
        if total > limit:
            fail("runtime config file is too large")


def strict_manifest(value: object) -> dict:
    if not isinstance(value, dict):
        fail("runtime config manifest is invalid")
    required = {
        "version", "workspace_id", "workspace_revision", "descriptor_git_blob",
        "descriptor_sha256", "descriptor_dev", "descriptor_ino", "config_sha256",
        "config_dwh_binding", "config_dev", "config_ino", "config_size", "config_mode",
        "config_uid", "config_nlink",
    }
    if set(value) != required or value.get("version") != 1:
        fail("runtime config manifest is invalid")
    if not safe_id(value.get("workspace_id")) or not safe_rev(value.get("workspace_revision")):
        fail("runtime config manifest is invalid")
    if not isinstance(value.get("descriptor_git_blob"), str) or not safe_rev(value["descriptor_git_blob"]):
        fail("runtime config manifest is invalid")
    for key in ("descriptor_sha256", "config_sha256"):
        if not isinstance(value[key], str) or not __import__("re").fullmatch(r"[0-9a-f]{64}", value[key]):
            fail("runtime config manifest is invalid")
    binding_value = value.get("config_dwh_binding")
    if not isinstance(binding_value, dict) or set(binding_value) != {"workspace_id", "config_fingerprint", "input_fingerprint"} or any(not isinstance(x, str) for x in binding_value.values()):
        fail("runtime config manifest is invalid")
    for key in ("descriptor_dev", "descriptor_ino", "config_dev", "config_ino", "config_size", "config_uid", "config_nlink"):
        if not isinstance(value[key], str) or not value[key].isdigit():
            fail("runtime config manifest is invalid")
    if value["config_mode"] != "400":
        fail("runtime config manifest is invalid")
    return value


def publish(inp: dict) -> dict:
    root = inp.get("data_root")
    wid = inp.get("workspace_id")
    rev = inp.get("workspace_revision")
    if (
        not isinstance(root, str)
        or not os.path.isabs(root)
        or not safe_id(wid)
        or not safe_rev(rev)
    ):
        fail("invalid publication identity")
    try:
        content = bytes.fromhex(inp["config_hex"])
    except (TypeError, ValueError):
        fail("invalid config bytes")
    base = inp.get("manifest_base")
    if not isinstance(base, dict):
        fail("invalid manifest")
    if base.get("workspace_id") != wid or base.get("workspace_revision") != rev:
        fail("manifest identity mismatch")
    sessions = walk(root, ["sessions"], True)
    ws = open_dir(sessions, wid, True)
    checked_dir(ws)
    prep = open_dir(ws, "preprocessing", True)
    checked_dir(prep)
    cfgdir = open_dir(prep, "runtime-config", True)
    checked_dir(cfgdir)
    mandir = open_dir(prep, "runtime-config-manifests", True)
    checked_dir(mandir)
    # The retained preprocessing directory is the single cross-process lock seam.
    # No pathname lock file is created in the workspace layout.
    fcntl.flock(prep, fcntl.LOCK_EX)
    try:
        name = f"{rev}.yaml"
        mname = f"{rev}.json"

        def current(dfd, n, mode):
            try:
                fd = os.open(n, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=dfd)
            except FileNotFoundError:
                return None
            try:
                return (fd, read_regular(fd, mode))
            except:
                os.close(fd)
                raise

        got = current(cfgdir, name, 0o400)
        if got:
            fd, s = got
            os.lseek(fd, 0, os.SEEK_SET)
            old = read_all(fd)
            os.close(fd)
            if old != content:
                fail("same-revision runtime configuration changed")
        else:
            stage = f".{name}.staging-{os.getpid()}-{os.urandom(8).hex()}"
            fd = os.open(
                stage, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=cfgdir
            )
            try:
                write_all(fd, content)
                os.fchmod(fd, 0o400)
                os.fsync(fd)
                try:
                    os.link(
                        stage, name, src_dir_fd=cfgdir, dst_dir_fd=cfgdir, follow_symlinks=False
                    )
                except FileExistsError:
                    pass
                # Keep metadata ordering explicit even on filesystems where a
                # hardlink publication does not retain fchmod as expected.
                os.fchmod(fd, 0o400)
                os.fsync(fd)
            finally:
                os.close(fd)
                try:
                    os.unlink(stage, dir_fd=cfgdir)
                except FileNotFoundError:
                    pass
            got = current(cfgdir, name, 0o400)
            if not got:
                fail("runtime config publication failed")
            fd, s = got
            try:
                os.lseek(fd, 0, os.SEEK_SET)
                if read_all(fd) != content:
                    fail("same-revision runtime configuration changed")
            finally:
                os.close(fd)
            os.fsync(cfgdir)
        # Identity is deliberately recorded after final no-replace publication.
        got = current(cfgdir, name, 0o400)
        assert got
        fd, s = got
        os.close(fd)
        manifest = {"version": 1, **dict(base)}
        manifest.update(
            {
                "config_sha256": hashlib.sha256(content).hexdigest(),
                "config_dev": str(s.st_dev),
                "config_ino": str(s.st_ino),
                "config_size": str(s.st_size),
                "config_mode": format(stat.S_IMODE(s.st_mode), "o"),
                "config_uid": str(s.st_uid),
                "config_nlink": str(s.st_nlink),
            }
        )
        strict_manifest(manifest)
        mb = (json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n").encode()
        oldm = current(mandir, mname, 0o600)
        if oldm:
            mfd, _ = oldm
            os.lseek(mfd, 0, os.SEEK_SET)
            existing = read_all(mfd)
            os.close(mfd)
            try: strict_manifest(json.loads(existing.decode()))
            except (ValueError, TypeError, UnicodeError, RuntimeError): fail("runtime config manifest is invalid")
            if existing != mb:
                fail("same-revision runtime configuration changed")
        else:
            stage = f".{mname}.staging-{os.getpid()}-{os.urandom(8).hex()}"
            fd = os.open(
                stage, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=mandir
            )
            try:
                write_all(fd, mb)
                os.fchmod(fd, 0o600)
                os.fsync(fd)
                try:
                    os.link(stage, mname, src_dir_fd=mandir, dst_dir_fd=mandir, follow_symlinks=False)
                except FileExistsError:
                    # A no-replace loser is successful only after validating the
                    # durable winner byte-for-byte and against the strict schema.
                    winner = current(mandir, mname, 0o600)
                    if winner is None:
                        fail("runtime config manifest publication raced")
                    wfd, _ = winner
                    try:
                        existing = read_all(wfd)
                    finally:
                        os.close(wfd)
                    try:
                        strict_manifest(json.loads(existing.decode()))
                    except (ValueError, TypeError, UnicodeError, RuntimeError):
                        fail("runtime config manifest is invalid")
                    if existing != mb:
                        fail("same-revision runtime configuration changed")
            finally:
                os.close(fd)
                try:
                    os.unlink(stage, dir_fd=mandir)
                except FileNotFoundError:
                    pass
            os.fsync(mandir)
        return {
            "path": f"{root}/sessions/{wid}/preprocessing/runtime-config/{name}",
            "manifestPath": f"{root}/sessions/{wid}/preprocessing/runtime-config-manifests/{mname}",
            "manifest": mb.decode(),
            "manifest_sha256": hashlib.sha256(mb).hexdigest(),
            "dev": s.st_dev,
            "ino": s.st_ino,
        }
    finally:
        os.close(cfgdir)
        os.close(mandir)
        os.close(prep)
        os.close(ws)
        os.close(sessions)


def verified_snapshot(inp: dict) -> dict:
    root = inp.get("snapshots_root")
    rev = inp.get("workspace_revision")
    wid = inp.get("workspace_id")
    if (
        not isinstance(root, str)
        or not os.path.isabs(root)
        or not safe_rev(rev)
        or not safe_id(wid)
    ):
        fail("invalid snapshot identity")
    # Component-relative no-follow traversal all the way to the retained descriptor.
    sroot = walk(root, [], False)
    rdir = open_dir(sroot, rev, False)
    checked_dir(rdir)
    fd = os.open(f"{wid}.yaml", os.O_RDONLY | os.O_NOFOLLOW, dir_fd=rdir)
    try:
        descriptor_info = read_regular(fd, 0o400)
        chunks = []
        while True:
            x = os.read(fd, 1024 * 1024)
            if not x:
                break
            chunks.append(x)
        source = b"".join(chunks)
    finally:
        os.close(fd)
    mf = os.open("snapshot.json", os.O_RDONLY | os.O_NOFOLLOW, dir_fd=rdir)
    try:
        read_regular(mf, 0o400)
        payload = b""
        while True:
            x = os.read(mf, 1024 * 1024)
            if not x:
                break
            payload += x
    finally:
        os.close(mf)
    try:
        manifest = json.loads(payload.decode())
    except (UnicodeDecodeError, json.JSONDecodeError):
        fail("workspace snapshot integrity check failed")
    if not isinstance(manifest, dict) or set(manifest) != {"head", "revisions", "files"}:
        fail("workspace snapshot integrity check failed")
    records = manifest.get("revisions")
    files = manifest.get("files")
    if not isinstance(records, list) or not isinstance(files, dict) or not records:
        fail("workspace snapshot integrity check failed")
    record_by_id: dict[str, dict] = {}
    for item in records:
        if not isinstance(item, dict) or set(item) != {"id", "commit", "blob", "snapshotPath"}:
            fail("workspace snapshot integrity check failed")
        item_id = item.get("id")
        if not isinstance(item_id, str) or not safe_id(item_id) or item_id in record_by_id:
            fail("workspace snapshot integrity check failed")
        if item.get("commit") != rev or item.get("snapshotPath") != f"{root}/{rev}/{item_id}.yaml":
            fail("workspace snapshot integrity check failed")
        if not isinstance(item.get("blob"), str) or not safe_rev(item["blob"]):
            fail("workspace snapshot integrity check failed")
        record_by_id[item_id] = item
    expected_names = {name for item_id in record_by_id for name in (f"{item_id}.yaml", f"{item_id}.env.example", f"{item_id}.md")}
    if set(files) != expected_names or any(not isinstance(v, str) or not __import__("re").fullmatch(r"[0-9a-f]{64}", v) for v in files.values()):
        fail("workspace snapshot integrity check failed")
    # Verify every immutable file declared by snapshot.json, not just the selected
    # descriptor.  This prevents extra records/files from smuggling a second state.
    for filename in sorted(expected_names):
        f = os.open(filename, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=rdir)
        try:
            read_regular(f, 0o400)
            actual = hashlib.sha256(read_all(f)).hexdigest()
        finally:
            os.close(f)
        if actual != files[filename]:
            fail("workspace snapshot integrity check failed")
    record = record_by_id.get(wid)
    expected_path = f"{root}/{rev}/{wid}.yaml"
    if record is None or manifest.get("head") != rev or record.get("snapshotPath") != expected_path:
        fail("workspace snapshot integrity check failed")
    if files.get(f"{wid}.yaml") != hashlib.sha256(source).hexdigest():
        fail("workspace snapshot integrity check failed")
    repo = inp.get("repository_root")
    if not isinstance(repo, str) or not os.path.isabs(repo):
        fail("invalid repository root")
    try:
        blob = subprocess.check_output(
            ["git", "-C", repo, "rev-parse", f"{rev}:workspaces/{wid}.yaml"],
            stderr=subprocess.DEVNULL, text=True, timeout=5,
        ).strip()
        git_source = subprocess.check_output(
            ["git", "-C", repo, "show", f"{rev}:workspaces/{wid}.yaml"],
            stderr=subprocess.DEVNULL, timeout=5,
        )
    except (OSError, subprocess.SubprocessError):
        fail("workspace Git revision is unavailable")
    try:
        git_text = git_source.decode("utf-8")
    except UnicodeDecodeError:
        fail("workspace Git descriptor identity mismatch")
    if blob != record.get("blob"):
        fail("workspace Git descriptor identity mismatch")
    return {
        "workspace_id": wid,
        "workspace_revision": rev,
        "source": source.decode(),
        "git_source": git_text,
        "sha256": hashlib.sha256(source).hexdigest(),
        "descriptor_git_blob": record.get("blob"),
        "descriptor_dev": descriptor_info.st_dev,
        "descriptor_ino": descriptor_info.st_ino,
        "snapshot_path": f"{root}/{rev}/{wid}.yaml",
    }


def binding(inp: dict) -> dict:
    try:
        raw = bytes.fromhex(inp["config_hex"])
    except (TypeError, ValueError):
        fail("invalid config bytes")
    # Use the harness' own Pydantic loader and config_dwh_binding; this is intentionally
    # not a TypeScript reimplementation of its normalization/fingerprinting rules.
    from tht.config import load_config
    from tht.jobs.dwh_pipeline import config_dwh_binding

    with tempfile.NamedTemporaryFile(
        prefix="runtime-binding-", suffix=".yaml", delete=False
    ) as stream:
        stream.write(raw)
        path = Path(stream.name)
    try:
        return config_dwh_binding(load_config(path))
    finally:
        try:
            path.unlink()
        except OSError:
            pass


def main() -> None:
    try:
        inp = json.load(sys.stdin)
        action = inp.get("action")
        if action == "publish":
            result = publish(inp)
        elif action == "verified-snapshot":
            result = verified_snapshot(inp)
        elif action == "binding":
            result = binding(inp)
        else:
            fail("unsupported runtime config action")
        print(json.dumps(result))
    except Exception as e:  # noqa: BLE001
        print(json.dumps({"error": str(e)}))
        raise SystemExit(1)


if __name__ == "__main__":
    main()