vibe-sketcher/Documentation

Source code for vibesketcher.server.store

"""The server's documents: in memory, revisioned, optionally backed by a file.

Human and agent edit the same document through the revision counter
(roadmap, section 3.2): every write names the revision it read, and a stale
write gets a conflict and must re-read. A document opened from a file on the
command line (``vibe-sketcher serve scheme.vsk``) can be saved back to that
file, and only to it: the HTTP API never takes a filesystem path.
"""

from __future__ import annotations

import secrets
import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from vibesketcher.document.io import (
    ABSENT,
    SaveRefused,
    _DocumentSnapshot,
    _read_snapshot,
    _save_with_digest,
)
from vibesketcher.document.model import Document

__all__ = ["Conflict", "Entry", "NotFound", "Store"]




class NotFound(KeyError):
    """No document with that id."""





class Conflict(RuntimeError):
    """The write named a revision that is no longer current."""

    def __init__(self, current: int, given: int) -> None:
        super().__init__(f"revision {given} is stale; the document is at revision {current}")
        self.current = current
        self.given = given





@dataclass
class Entry:
    id: str
    document: Document
    #: The file it was opened from and saves to, if any.
    path: Path | None = None
    #: The revision last written to ``path``.
    saved_revision: int | None = None
    warnings: list[str] = field(default_factory=list)
    #: What the QVF job container it was opened from says about its calculation.
    result: dict[str, Any] | None = None
    #: The file's bytes as last read or written (:func:`~vibesketcher.document.io.digest_of`),
    #: so a save never replaces a change made outside vibe-sketcher.
    digest: str | None = None
    #: Physical destination selected when the file binding was created. The
    #: displayed path may be an alias, but a later retarget must not rebind Save.
    _destination: Path | None = field(default=None, init=False, repr=False)



    @property
    def dirty(self) -> bool:
        return self.path is not None and self.saved_revision != self.document.revision






class Store:
    def __init__(self) -> None:
        self._entries: dict[str, Entry] = {}
        self._lock = threading.Lock()
        #: The document the browser opens first, when the server was started with one.
        self.default_id: str | None = None



    def add(self, document: Document, *, path: Path | None = None) -> Entry:
        """Add a document; a supplied file binding must describe that same document.

        Use :meth:`open` to parse and bind a file in one read. For compatibility,
        callers may still supply both a document and path, but stale or edited
        contents cannot be paired with that file's current digest. A missing
        destination starts unsaved and may be created by its first save.
        """
        snapshot = _read_snapshot(path, allow_missing=True) if path is not None else None
        if snapshot is not None and snapshot.digest != ABSENT and snapshot.document != document:
            raise SaveRefused(
                "The supplied document does not match the file being bound. "
                "Reopen it with Store.open, or save a copy to a new path."
            )
        return self._insert(document, snapshot)




    def open(self, path: Path) -> Entry:
        """Read and bind one file snapshot, or a new unsaved document if absent."""
        snapshot = _read_snapshot(path, allow_missing=True)
        return self._insert(snapshot.document, snapshot)


    def _insert(self, document: Document, snapshot: _DocumentSnapshot | None) -> Entry:
        with self._lock:
            identifier = secrets.token_hex(4)
            while identifier in self._entries:
                identifier = secrets.token_hex(4)
            entry = Entry(identifier, document)
            if snapshot is not None:
                entry.path = snapshot.path
                entry._destination = snapshot.destination
                entry.digest = snapshot.digest
                entry.result = snapshot.result
                if snapshot.digest != ABSENT:
                    entry.saved_revision = document.revision
            self._entries[identifier] = entry
            return entry



    def get(self, identifier: str) -> Entry:
        with self._lock:
            try:
                return self._entries[identifier]
            except KeyError:
                raise NotFound(identifier) from None




    def entries(self) -> list[Entry]:
        with self._lock:
            return list(self._entries.values())




    def replace(
        self,
        identifier: str,
        document: Document,
        *,
        expected: int,
        warnings: list[str] | None = None,
    ) -> Entry:
        """Store ``document`` as the next revision, if ``expected`` is current."""
        with self._lock:
            try:
                entry = self._entries[identifier]
            except KeyError:
                raise NotFound(identifier) from None
            current = entry.document.revision
            if expected != current:
                raise Conflict(current, expected)
            entry.document = document.model_copy(update={"revision": current + 1})
            entry.warnings = list(warnings or [])
            return entry




    def remove(self, identifier: str) -> None:
        with self._lock:
            if self._entries.pop(identifier, None) is None:
                raise NotFound(identifier)
            if self.default_id == identifier:
                self.default_id = None




    def save(self, identifier: str) -> Entry:
        """Write the document back to the file it came from.

        Refused with :class:`~vibesketcher.document.io.SaveRefused` (a ``ValueError``)
        when the file changed since it was opened or last saved, or its path
        no longer resolves to the destination selected by :meth:`add`.
        """
        with self._lock:
            try:
                entry = self._entries[identifier]
            except KeyError:
                raise NotFound(identifier) from None
            if entry.path is None:
                raise ValueError("this document has no file; download it instead")
            _, written_digest = _save_with_digest(
                entry.document, entry.path, expect=entry.digest, destination=entry._destination
            )
            entry.digest = written_digest
            entry.saved_revision = entry.document.revision
            return entry