"""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_relationship_update( 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]: """Queue relationship-only changes against one exact existing node.""" if not relationship_changes: raise DocForgeError( "invalid_operation", "Relationship-only updates require at least one change" ) return self.propose_update( changeset_id=changeset_id, expected_changeset_hash=expected_changeset_hash, node_id=node_id, expected_content_hash=expected_content_hash, metadata=None, content=None, relationship_changes=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)