DocForge2/src/docforge/changesets.py

"""Confined atomic storage for project-bound proposal changesets."""

from __future__ import annotations

import fcntl
import json
import os
import tempfile
from collections.abc import Callable, Generator, Mapping
from contextlib import contextmanager
from pathlib import Path
from typing import Any, cast

from .changeset_contract import (
    document_hash,
    normalize_operation,
    validate_document,
    validate_hash,
    validate_id,
)
from .errors import DocForgeError
from .models import Edge, Node, ProjectService, ProjectSnapshot, ProposalWriter
from .project import project_root_fingerprint
from .proposal_projection import ProposalProjector


class ChangesetStore:
    """One project-bound proposal store with an optional immutable writer identity."""

    def __init__(self, project: ProjectService, writer_id: str | None = None) -> None:
        self.project = project
        self.projector = ProposalProjector(project)
        writers = {writer.writer_id: writer for writer in project.descriptor.proposal_writers}
        if writer_id is not None and writer_id not in writers:
            raise DocForgeError(
                "unknown_writer",
                "Proposal writer is not configured for this project",
                writer=writer_id,
            )
        self.writer = writers.get(writer_id) if writer_id is not None else None

    @property
    def writer_id(self) -> str | None:
        return self.writer.writer_id if self.writer is not None else None

    def access(self) -> dict[str, object]:
        if self.writer is None:
            return {"enabled": False, "writer": None, "families": [], "operations": []}
        return {
            "enabled": True,
            "writer": self.writer.writer_id,
            "families": list(self.writer.families),
            "operations": list(self.writer.operations),
        }

    def create(self, changeset_id: str) -> dict[str, object]:
        writer = self._require_writer()
        validate_id(changeset_id, "changeset_id")
        with self._lock():
            path = self._path(changeset_id)
            if path.exists():
                raise DocForgeError(
                    "changeset_exists", "Changeset ID already exists", changeset_id=changeset_id
                )
            existing = tuple(self._root().glob("*.json"))
            if len(existing) >= self.project.descriptor.limits.max_changesets:
                raise DocForgeError("changeset_limit", "Project changeset limit has been reached")
            snapshot = self.project.load()
            document: dict[str, Any] = {
                "schema_version": 1,
                "changeset_id": changeset_id,
                "project_id": snapshot.descriptor.project_id,
                "root_fingerprint": project_root_fingerprint(snapshot.descriptor.root),
                "base_revision": snapshot.revision,
                "base_source_hash": snapshot.source_hash,
                "creator": writer.writer_id,
                "operations": [],
            }
            self._write(path, document)
            return self._result(snapshot, document, valid=True)

    def propose_create(
        self,
        *,
        changeset_id: str,
        expected_changeset_hash: str,
        node_id: str,
        target_source: str,
        metadata: dict[str, Any],
        content: str,
        relationship_changes: list[dict[str, Any]],
        rationale: str,
    ) -> dict[str, object]:
        return self._append(
            changeset_id,
            expected_changeset_hash,
            {
                "operation": "create",
                "node_id": node_id,
                "expected_content_hash": None,
                "target_source": target_source,
                "metadata": metadata,
                "content": content,
                "relationship_changes": relationship_changes,
                "rationale": rationale,
            },
        )

    def propose_update(
        self,
        *,
        changeset_id: str,
        expected_changeset_hash: str,
        node_id: str,
        expected_content_hash: str,
        metadata: dict[str, Any] | None,
        content: str | None,
        relationship_changes: list[dict[str, Any]],
        rationale: str,
    ) -> dict[str, object]:
        return self._append(
            changeset_id,
            expected_changeset_hash,
            {
                "operation": "update",
                "node_id": node_id,
                "expected_content_hash": expected_content_hash,
                "target_source": None,
                "metadata": metadata,
                "content": content,
                "relationship_changes": relationship_changes,
                "rationale": rationale,
            },
        )

    def propose_move(
        self,
        *,
        changeset_id: str,
        expected_changeset_hash: str,
        node_id: str,
        expected_content_hash: str,
        target_source: str,
        rationale: str,
    ) -> dict[str, object]:
        return self._append(
            changeset_id,
            expected_changeset_hash,
            {
                "operation": "move",
                "node_id": node_id,
                "expected_content_hash": expected_content_hash,
                "target_source": target_source,
                "metadata": None,
                "content": None,
                "relationship_changes": [],
                "rationale": rationale,
            },
        )

    def propose_delete(
        self,
        *,
        changeset_id: str,
        expected_changeset_hash: str,
        node_id: str,
        expected_content_hash: str,
        relationship_changes: list[dict[str, Any]],
        rationale: str,
    ) -> dict[str, object]:
        return self._append(
            changeset_id,
            expected_changeset_hash,
            {
                "operation": "delete",
                "node_id": node_id,
                "expected_content_hash": expected_content_hash,
                "target_source": None,
                "metadata": None,
                "content": None,
                "relationship_changes": relationship_changes,
                "rationale": rationale,
            },
        )

    def validate(self, changeset_id: str) -> dict[str, object]:
        validate_id(changeset_id, "changeset_id")
        with self._lock():
            snapshot, document, nodes, edges = self._validate_locked(changeset_id)
            return self._result(
                snapshot,
                document,
                valid=True,
                projected_node_count=len(nodes),
                projected_edge_count=len(edges),
            )

    def list_changesets(self) -> dict[str, object]:
        with self._lock():
            snapshot = self.project.load()
            records: list[dict[str, object]] = []
            for path in sorted(self._root().glob("*.json"), key=lambda item: item.name):
                document = self._read(path)
                records.append(
                    {
                        "changeset_id": document["changeset_id"],
                        "changeset_hash": document_hash(document),
                        "creator": document["creator"],
                        "base_revision": document["base_revision"],
                        "base_source_hash": document["base_source_hash"],
                        "base_state": self._base_state(document, snapshot),
                        "operation_count": len(document["operations"]),
                    }
                )
            return self._base_result(snapshot, count=len(records), changesets=records)

    def inspect(self, changeset_id: str) -> dict[str, object]:
        validate_id(changeset_id, "changeset_id")
        with self._lock():
            document = self._read(self._path(changeset_id))
            snapshot = self.project.load()
            return self._result(
                snapshot,
                document,
                base_state=self._base_state(document, snapshot),
            )

    def diff(self, changeset_id: str) -> dict[str, object]:
        validate_id(changeset_id, "changeset_id")
        with self._lock():
            snapshot, document, _, _ = self._validate_locked(changeset_id)
            nodes = {node.node_id: node for node in snapshot.nodes}
            edges = {(edge.source_id, edge.relation, edge.target_id) for edge in snapshot.edges}
            changes: list[dict[str, object]] = []
            for operation in document["operations"]:
                before = nodes.get(operation["node_id"])
                before_edges = set(edges)
                self.projector.apply_operation(
                    snapshot, nodes, edges, operation, self.projector.writer(document)
                )
                after = nodes.get(operation["node_id"])
                changes.append(
                    self.projector.operation_diff(
                        operation,
                        before,
                        after,
                        sorted(edges - before_edges),
                        sorted(before_edges - edges),
                    )
                )
            return self._result(snapshot, document, valid=True, changes=changes)

    def projected_snapshot(self, changeset_id: str) -> tuple[ProjectSnapshot, str]:
        """Return a validated in-memory proposal projection for derived preview use."""

        validate_id(changeset_id, "changeset_id")
        with self._lock():
            snapshot, document, nodes, edges = self._validate_locked(changeset_id)
            projected = ProjectSnapshot(
                descriptor=snapshot.descriptor,
                nodes=tuple(sorted(nodes.values(), key=lambda node: node.node_id)),
                edges=tuple(Edge(*edge) for edge in sorted(edges)),
                source_hash=snapshot.source_hash,
                revision=snapshot.revision,
            )
            return projected, document_hash(document)

    def apply(
        self,
        *,
        changeset_id: str,
        expected_changeset_hash: str,
        applier_id: str,
        application: Callable[
            [ProjectSnapshot, ProjectSnapshot, tuple[Mapping[str, object], ...]],
            dict[str, object],
        ],
    ) -> dict[str, object]:
        """Apply one exact validated proposal through a project-owned canonical serializer."""

        validate_id(changeset_id, "changeset_id")
        validate_hash(expected_changeset_hash, "expected_changeset_hash")
        if self.writer is None or self.writer.writer_id != applier_id:
            raise DocForgeError(
                "canonical_application_disabled",
                "Canonical applier identity is not configured for this store",
            )
        with self._lock():
            snapshot, document, nodes, edges = self._validate_locked(changeset_id)
            actual_hash = document_hash(document)
            if actual_hash != expected_changeset_hash:
                raise DocForgeError(
                    "changeset_conflict",
                    "Changeset changed after the caller approved it",
                    changeset_id=changeset_id,
                    expected=expected_changeset_hash,
                    actual=actual_hash,
                )
            if document["creator"] != applier_id:
                raise DocForgeError(
                    "changeset_owner_conflict",
                    "Canonical applier does not own this changeset",
                    changeset_id=changeset_id,
                    owner=document["creator"],
                    applier=applier_id,
                )
            projected = ProjectSnapshot(
                descriptor=snapshot.descriptor,
                nodes=tuple(sorted(nodes.values(), key=lambda node: node.node_id)),
                edges=tuple(Edge(*edge) for edge in sorted(edges)),
                source_hash=snapshot.source_hash,
                revision=snapshot.revision,
            )
            payload = application(
                snapshot,
                projected,
                tuple(cast(Mapping[str, object], item) for item in document["operations"]),
            )
            current = self.project.load()
            return self._result(
                current,
                document,
                valid=True,
                applied=True,
                applied_from_revision=snapshot.revision,
                applied_from_source_hash=snapshot.source_hash,
                **payload,
            )

    def _append(
        self,
        changeset_id: str,
        expected_changeset_hash: str,
        operation: dict[str, Any],
    ) -> dict[str, object]:
        writer = self._require_writer()
        validate_id(changeset_id, "changeset_id")
        validate_hash(expected_changeset_hash, "expected_changeset_hash")
        with self._lock():
            path = self._path(changeset_id)
            document = self._read(path)
            actual_hash = document_hash(document)
            if actual_hash != expected_changeset_hash:
                raise DocForgeError(
                    "changeset_conflict",
                    "Changeset changed after the caller read it",
                    changeset_id=changeset_id,
                    expected=expected_changeset_hash,
                    actual=actual_hash,
                )
            if document["creator"] != writer.writer_id:
                raise DocForgeError(
                    "changeset_owner_conflict",
                    "Proposal writer does not own this changeset",
                    changeset_id=changeset_id,
                    owner=document["creator"],
                    writer=writer.writer_id,
                )
            if (
                len(document["operations"])
                >= self.project.descriptor.limits.max_changeset_operations
            ):
                raise DocForgeError(
                    "changeset_operation_limit", "Changeset operation limit has been reached"
                )
            normalized = normalize_operation(operation, sequence=len(document["operations"]) + 1)
            if any(item["node_id"] == normalized["node_id"] for item in document["operations"]):
                raise DocForgeError(
                    "duplicate_operation",
                    "A changeset may touch a node only once",
                    node_id=normalized["node_id"],
                )
            candidate = {**document, "operations": [*document["operations"], normalized]}
            snapshot = self.project.load()
            self._check_base(candidate, snapshot)
            nodes, edges = self.projector.project(snapshot, candidate)
            self._check_proposal_conflicts(candidate, snapshot)
            self._write(path, candidate)
            return self._result(
                snapshot,
                candidate,
                valid=True,
                projected_node_count=len(nodes),
                projected_edge_count=len(edges),
            )

    def _validate_locked(
        self, changeset_id: str
    ) -> tuple[ProjectSnapshot, dict[str, Any], dict[str, Node], set[tuple[str, str, str]]]:
        document = self._read(self._path(changeset_id))
        snapshot = self.project.load()
        self._check_base(document, snapshot)
        nodes, edges = self.projector.project(snapshot, document)
        self._check_proposal_conflicts(document, snapshot)
        return snapshot, document, nodes, edges

    def _check_base(self, document: dict[str, Any], snapshot: ProjectSnapshot) -> None:
        if self._base_state(document, snapshot) != "current":
            raise DocForgeError(
                "base_conflict",
                "Canonical project changed after the changeset was created",
                expected_revision=document["base_revision"],
                actual_revision=snapshot.revision,
                expected_source_hash=document["base_source_hash"],
                actual_source_hash=snapshot.source_hash,
            )

    @staticmethod
    def _base_state(document: dict[str, Any], snapshot: ProjectSnapshot) -> str:
        if (
            document["base_revision"] == snapshot.revision
            and document["base_source_hash"] == snapshot.source_hash
        ):
            return "current"
        return "stale"

    def _check_proposal_conflicts(
        self, document: dict[str, Any], snapshot: ProjectSnapshot
    ) -> None:
        nodes, sources = self.projector.touches(document, snapshot)
        conflicts: list[dict[str, object]] = []
        for path in sorted(self._root().glob("*.json"), key=lambda item: item.name):
            if path.name == f"{document['changeset_id']}.json":
                continue
            other = self._read(path)
            if other["base_source_hash"] != document["base_source_hash"]:
                continue
            other_nodes, other_sources = self.projector.touches(other, snapshot)
            shared_nodes = sorted(nodes & other_nodes)
            shared_sources = sorted(sources & other_sources)
            if shared_nodes or shared_sources:
                conflicts.append(
                    {
                        "changeset_id": other["changeset_id"],
                        "nodes": shared_nodes,
                        "sources": shared_sources,
                    }
                )
        if conflicts:
            raise DocForgeError(
                "proposal_conflict",
                "Changeset overlaps another proposal from the same canonical base",
                conflicts=conflicts,
            )

    def _result(
        self,
        snapshot: ProjectSnapshot,
        document: dict[str, Any],
        **payload: object,
    ) -> dict[str, object]:
        return self._base_result(
            snapshot,
            changeset_id=document["changeset_id"],
            changeset_hash=document_hash(document),
            creator=document["creator"],
            base_revision=document["base_revision"],
            base_source_hash=document["base_source_hash"],
            operation_count=len(document["operations"]),
            operations=document["operations"],
            **payload,
        )

    @staticmethod
    def _base_result(snapshot: ProjectSnapshot, **payload: object) -> dict[str, object]:
        return {
            "status": "ok",
            "project_id": snapshot.descriptor.project_id,
            "project_root_fingerprint": project_root_fingerprint(snapshot.descriptor.root),
            "adapter": snapshot.descriptor.adapter,
            "revision": snapshot.revision,
            "source_hash": snapshot.source_hash,
            **payload,
        }

    def _require_writer(self) -> ProposalWriter:
        if self.writer is None:
            raise DocForgeError(
                "proposal_access_disabled",
                "Server has no configured proposal writer identity",
            )
        return self.writer

    def _read(self, path: Path) -> dict[str, Any]:
        if path.is_symlink():
            raise DocForgeError("path_escape", "Changeset files may not be symbolic links")
        if not path.is_file():
            raise DocForgeError("missing_changeset", "Changeset does not exist", path=path.name)
        if path.stat().st_size > self.project.descriptor.limits.max_changeset_bytes:
            raise DocForgeError("changeset_too_large", "Changeset exceeds configured size limit")
        raw = path.read_bytes()
        if len(raw) > self.project.descriptor.limits.max_changeset_bytes:
            raise DocForgeError("changeset_too_large", "Changeset exceeds configured size limit")
        try:
            document = json.loads(raw)
        except (UnicodeDecodeError, json.JSONDecodeError) as error:
            raise DocForgeError("invalid_changeset", "Changeset is not valid UTF-8 JSON") from error
        return validate_document(self.project, document, path=path)

    def _write(self, path: Path, document: dict[str, Any]) -> None:
        raw = json.dumps(document, sort_keys=True, indent=2).encode("utf-8") + b"\n"
        if len(raw) > self.project.descriptor.limits.max_changeset_bytes:
            raise DocForgeError("changeset_too_large", "Changeset exceeds configured size limit")
        root = self._root()
        descriptor = self.project.descriptor
        previous = path.read_bytes() if path.is_file() and not path.is_symlink() else None
        if path.is_symlink():
            raise DocForgeError("path_escape", "Changeset files may not be symbolic links")
        source_paths = self.project.canonical_source_paths()
        canonical_before = {
            source: source.read_bytes()
            for source in (
                descriptor.descriptor_path,
                *descriptor.authority_files,
                *source_paths,
            )
        }
        file_descriptor, temporary_name = tempfile.mkstemp(prefix=".pending-", dir=root)
        temporary = Path(temporary_name)
        try:
            with os.fdopen(file_descriptor, "wb") as handle:
                handle.write(raw)
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, path)
            directory_descriptor = os.open(root, os.O_RDONLY)
            try:
                os.fsync(directory_descriptor)
            finally:
                os.close(directory_descriptor)
        except Exception:
            temporary.unlink(missing_ok=True)
            raise
        changed_source: str | None = None
        if self.project.canonical_source_paths() != source_paths:
            changed_source = "canonical_source_set"
        else:
            for source, before in canonical_before.items():
                if not source.is_file() or source.read_bytes() != before:
                    changed_source = source.relative_to(descriptor.root).as_posix()
                    break
        if changed_source is not None:
            self._restore(path, previous, root)
            raise DocForgeError(
                "canonical_write_detected",
                "Canonical source changed during changeset storage",
                source=changed_source,
            )

    @staticmethod
    def _restore(path: Path, previous: bytes | None, root: Path) -> None:
        if previous is None:
            path.unlink(missing_ok=True)
            return
        restore_descriptor, restore_name = tempfile.mkstemp(prefix=".rollback-", dir=root)
        restore = Path(restore_name)
        try:
            with os.fdopen(restore_descriptor, "wb") as handle:
                handle.write(previous)
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(restore, path)
        except Exception:
            restore.unlink(missing_ok=True)
            raise

    def _path(self, changeset_id: str) -> Path:
        validate_id(changeset_id, "changeset_id")
        path = self._root() / f"{changeset_id}.json"
        if path.parent.resolve() != self._root():
            raise DocForgeError("path_escape", "Changeset path escaped its configured root")
        return path

    def _root(self) -> Path:
        root = self.project.descriptor.changeset_root
        root.mkdir(parents=True, exist_ok=True)
        if not root.is_dir() or root.resolve() != root:
            raise DocForgeError("path_escape", "Changeset root changed or resolves unexpectedly")
        return root

    @contextmanager
    def _lock(self) -> Generator[None]:
        root = self._root()
        lock_path = root / ".lock"
        try:
            descriptor = os.open(
                lock_path,
                os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW,
                0o600,
            )
        except OSError as error:
            raise DocForgeError("path_escape", "Changeset lock path is not safe") from error
        with os.fdopen(descriptor, "a+b") as handle:
            fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
            try:
                yield
            finally:
                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)