"""Canonical native-review publication journal registry."""
from __future__ import annotations

import errno
import fcntl
import hashlib
import json
import math
import os
import stat
import tempfile
from pathlib import Path

SCHEMA = "gitea-agent.native-review-binding/v1"
INTENT_SCHEMA = "gitea-agent.native-review-intent/v1"
ROOT_NAME = "native-review-journals"
_ROOT_PRESENCE_SCHEMA = "gitea-agent.native-review-root/v1"
_ENTRY_PRESENCE_SCHEMA = "gitea-agent.native-review-entry/v1"
_MAX_JSON = 16 * 1024 * 1024


def _fail(message):
    raise ValueError(message)


def _canonical(value):
    try:
        return json.dumps(value, sort_keys=True, ensure_ascii=False,
                          allow_nan=False, separators=(",", ":"))
    except (TypeError, ValueError, OverflowError):
        _fail("non-canonical JSON value")


def _strict_load(data):
    def pairs(pairs):
        out = {}
        for key, value in pairs:
            if key in out:
                _fail("duplicate JSON key")
            out[key] = value
        return out
    try:
        return json.loads(data.decode("utf-8"), object_pairs_hook=pairs,
                          parse_constant=lambda value: _fail("non-finite JSON"),
                          parse_float=lambda value: float(value) if math.isfinite(float(value)) else _fail("non-finite JSON"))
    except ValueError:
        raise
    except Exception:
        _fail("malformed journal artifact")


def binding_for(plan, *, origin):
    t = plan["target"]
    return {"schema": SCHEMA, "origin": origin, "owner": t["owner"],
            "repository": t["repository"], "issue": t["issue"],
            "pull": t["pull"], "head": t["head"],
            "contract_sha256": plan["contract_sha256"],
            "reviewer_id": plan["reviewer"]["id"]}


def plan_fingerprint(plan):
    return hashlib.sha256(_canonical(plan).encode("utf-8")).hexdigest()


def root_for(anchor_file):
    harness = Path(anchor_file).resolve().parents[5]
    return harness / ".runtime" / ROOT_NAME


def canonical_path(plan, *, anchor_file, origin, validate):
    validate(plan)
    binding = binding_for(plan, origin=origin)
    key = hashlib.sha256(_canonical(binding).encode("utf-8")).hexdigest()
    return root_for(anchor_file) / "v1" / key


def _raw_path_safe(value):
    if isinstance(value, str):
        if not value.startswith("/"):
            _fail("unsafe journal path")
        parts = value.split("/")
        if any(part in ("", ".", "..") for part in parts[1:]):
            _fail("unsafe journal path")


def _check_ancestor_chain(path, *, allow_missing_leaf=False):
    path = Path(path)
    if not path.is_absolute() or any(part in ("", ".", "..") for part in path.parts):
        _fail("unsafe journal path")
    current = Path(path.anchor)
    for index, part in enumerate(path.parts[1:]):
        current /= part
        try:
            st = current.lstat()
        except FileNotFoundError:
            if allow_missing_leaf and index == len(path.parts) - 2:
                return
            _fail("missing trusted journal ancestor")
        if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode):
            _fail("unsafe journal ancestor")


def _fsync_dir(path):
    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC)
    try:
        os.fsync(fd)
    finally:
        os.close(fd)


def _check_private_file(fd, *, what):
    st = os.fstat(fd)
    if (not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid()
            or stat.S_IMODE(st.st_mode) != 0o600 or st.st_nlink != 1):
        _fail("unsafe journal " + what)
    return st


def _open_bootstrap(parent):
    path = Path(parent) / (ROOT_NAME + ".bootstrap.lock")
    try:
        fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600)
    except OSError:
        _fail("unsafe journal bootstrap lock")
    try:
        _check_private_file(fd, what="bootstrap lock")
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except (BlockingIOError, OSError) as exc:
            if getattr(exc, "errno", None) in (errno.EACCES, errno.EAGAIN):
                _fail("journal bootstrap busy")
            raise
        return fd
    except Exception:
        os.close(fd)
        raise


def _presence_read(path, cap=4096):
    try:
        return _strict_load(_secure_file(path, cap=cap))
    except FileNotFoundError:
        return None


def _atomic_bytes(directory, name, data, *, replace=False):
    directory = Path(directory)
    _check_ancestor_chain(directory)
    target = directory / name
    try:
        st = target.lstat()
        if stat.S_ISLNK(st.st_mode) or not stat.S_ISREG(st.st_mode) or st.st_nlink != 1:
            _fail("unsafe journal artifact")
        if not replace:
            _fail("journal artifact already exists")
    except FileNotFoundError:
        pass
    fd, temp = tempfile.mkstemp(dir=directory, prefix=".native-review-")
    try:
        os.fchmod(fd, 0o600)
        with os.fdopen(fd, "wb") as out:
            out.write(data); out.flush(); os.fsync(out.fileno())
        if replace:
            os.replace(temp, target)
        else:
            try:
                os.link(temp, target)
            except FileExistsError:
                _fail("journal artifact already exists")
            os.unlink(temp)
        _fsync_dir(directory)
    finally:
        try:
            os.unlink(temp)
        except FileNotFoundError:
            pass


def atomic_json(directory, name, value):
    _atomic_bytes(directory, name, _canonical(value).encode("utf-8"))


def ensure_root(root, *, _lock_fd=None):
    root = Path(root)
    parent = root.parent
    _check_ancestor_chain(parent)
    lock_fd = _lock_fd if _lock_fd is not None else _open_bootstrap(parent)
    try:
        root_presence = parent / (root.name + ".presence")
        rp = _presence_read(root_presence)
        root_exists = False
        try:
            rst = root.lstat(); root_exists = True
        except FileNotFoundError:
            rst = None
        if rp is None:
            if root_exists:
                _fail("journal root presence anchor missing")
            root.mkdir(mode=0o700)
            rst = root.lstat()
            atomic_json(parent, root_presence.name, {"schema": _ROOT_PRESENCE_SCHEMA,
                "root": root.name + "/v1", "devino": [rst.st_dev, rst.st_ino]})
        else:
            if set(rp) != {"schema", "root", "devino"} or rp["schema"] != _ROOT_PRESENCE_SCHEMA:
                _fail("malformed journal root presence")
            if not root_exists or stat.S_ISLNK(rst.st_mode) or not stat.S_ISDIR(rst.st_mode):
                _fail("journal root was deleted")
            if [rst.st_dev, rst.st_ino] != rp["devino"]:
                _fail("journal root was replaced")
        rst = root.lstat()
        if (stat.S_ISLNK(rst.st_mode) or not stat.S_ISDIR(rst.st_mode)
                or rst.st_uid != os.getuid() or stat.S_IMODE(rst.st_mode) != 0o700):
            _fail("unsafe journal root")
        version = root / "v1"
        try: vst = version.lstat()
        except FileNotFoundError:
            version.mkdir(mode=0o700); _fsync_dir(root); vst = version.lstat()
        if (stat.S_ISLNK(vst.st_mode) or not stat.S_ISDIR(vst.st_mode)
                or vst.st_uid != os.getuid() or stat.S_IMODE(vst.st_mode) != 0o700):
            _fail("unsafe journal registry")
        return root
    finally:
        if _lock_fd is None:
            os.close(lock_fd)


def _secure_file(path, *, cap=_MAX_JSON):
    _raw_path_safe(os.fspath(path))
    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC)
    try:
        st = _check_private_file(fd, what="artifact")
        if st.st_size > cap:
            _fail("oversize journal artifact")
        chunks = []; total = 0
        while True:
            chunk = os.read(fd, min(65536, cap + 1 - total))
            if not chunk: break
            chunks.append(chunk); total += len(chunk)
            if total > cap: _fail("oversize journal artifact")
        return b"".join(chunks)
    finally:
        os.close(fd)


def _entry_check(entry):
    st = Path(entry).lstat()
    if (stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode)
            or st.st_uid != os.getuid() or stat.S_IMODE(st.st_mode) != 0o700):
        _fail("unsafe journal binding")
    return st


def _presence_path(entry):
    entry = Path(entry)
    return entry.parents[1].parent / (entry.name + ".presence")


def open_binding(path, plan, *, anchor_file, origin, validate, require_new=False):
    _raw_path_safe(os.fspath(path))
    validate(plan)
    expected = canonical_path(plan, anchor_file=anchor_file, origin=origin, validate=validate)
    if Path(path) != expected or (isinstance(path, str) and path != str(expected)):
        _fail("non-canonical journal path")
    root = expected.parents[1]
    _check_ancestor_chain(root.parent, allow_missing_leaf=True)
    try:
        root.parent.mkdir(mode=0o700)
        _fsync_dir(root.parent.parent)
    except FileExistsError:
        pass
    _entry_check(root.parent)
    bootstrap = _open_bootstrap(root.parent)
    try:
        ensure_root(root, _lock_fd=bootstrap)
        entry = expected
        presence = _presence_path(entry)
        ep = _presence_read(presence)
        exists = entry.exists()
        if require_new and (exists or ep is not None):
            _fail("canonical journal entry already exists")
        if ep is not None:
            if set(ep) != {"schema", "entry", "devino", "fingerprint", "state"} or ep["schema"] != _ENTRY_PRESENCE_SCHEMA:
                _fail("malformed journal entry presence")
            if ep["state"] != "complete" or not exists:
                _fail("interrupted or deleted journal entry")
            st = _entry_check(entry)
            if [st.st_dev, st.st_ino] != ep["devino"]:
                _fail("journal entry was replaced")
        elif exists:
            _fail("journal entry presence anchor missing")
        else:
            entry.mkdir(mode=0o700); _fsync_dir(root)
            if require_new:
                atomic_json(entry, "import.pending.json", {
                    "schema": "gitea-agent.native-review-import/v1",
                    "state": "pending"})
        st = _entry_check(entry)
        pending = entry / "import.pending.json"
        if not require_new and pending.exists():
            _fail("interrupted legacy registration")
        binding = binding_for(plan, origin=origin)
        expected_record = {"schema": SCHEMA, "binding": binding, "fingerprint": plan_fingerprint(plan)}
        record_path = entry / "binding.json"
        try:
            existing = _strict_load(_secure_file(record_path, cap=1024 * 1024))
        except FileNotFoundError:
            existing = None
        if existing is not None and _canonical(existing) != _canonical(expected_record):
            _fail("binding record mismatch")
        if existing is None:
            if ep is not None:
                _fail("known journal binding record missing")
            if require_new is False and record_path.exists() and not record_path.is_file():
                _fail("unsafe journal binding record")
            atomic_json(entry, "binding.json", expected_record)
        if ep is None:
            _atomic_bytes(presence.parent, presence.name, _canonical({
                "schema": _ENTRY_PRESENCE_SCHEMA, "entry": entry.name,
                "devino": [st.st_dev, st.st_ino], "fingerprint": expected_record["fingerprint"],
                "state": "complete"}).encode())
        else:
            if ep["fingerprint"] != expected_record["fingerprint"]:
                _fail("entry binding fingerprint mismatch")
        return entry
    finally:
        os.close(bootstrap)


def assert_bound(entry, plan, *, anchor_file, origin, validate):
    validate(plan)
    entry = Path(entry)
    expected = canonical_path(plan, anchor_file=anchor_file, origin=origin, validate=validate)
    if entry != expected:
        _fail("non-canonical journal path")
    root = expected.parents[1]
    _check_ancestor_chain(entry)
    _entry_check(root.parent); rst = _entry_check(root)
    _entry_check(root / 'v1'); st = _entry_check(entry)
    rp = _presence_read(root.parent / (root.name + ".presence"))
    expected_rp = {"schema": _ROOT_PRESENCE_SCHEMA, "root": root.name + "/v1",
                   "devino": [rst.st_dev, rst.st_ino]}
    if _canonical(rp) != _canonical(expected_rp):
        _fail("journal root presence mismatch")
    expected_record = {"schema": SCHEMA, "binding": binding_for(plan, origin=origin), "fingerprint": plan_fingerprint(plan)}
    ep = _presence_read(_presence_path(entry))
    expected_ep = {"schema": _ENTRY_PRESENCE_SCHEMA, "entry": entry.name,
                   "devino": [st.st_dev, st.st_ino], "fingerprint": expected_record["fingerprint"],
                   "state": "complete"}
    if _canonical(ep) != _canonical(expected_ep):
        _fail("journal entry presence mismatch")
    record = _strict_load(_secure_file(entry / "binding.json", cap=1024 * 1024))
    if _canonical(record) != _canonical(expected_record):
        _fail("binding record mismatch")
    return entry


def acquire_lock(entry):
    path = Path(entry) / "publication.lock"
    fd = None
    try:
        fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC, 0o600)
        _check_private_file(fd, what="lock")
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError as exc:
            if exc.errno in (errno.EACCES, errno.EAGAIN):
                _fail("journal lock busy")
            raise
        return fd
    except BaseException:
        if fd is not None:
            os.close(fd)
        raise


def read_json(entry, name, cap=_MAX_JSON):
    try:
        return _strict_load(_secure_file(Path(entry) / name, cap=cap))
    except FileNotFoundError:
        return None