Source code for vibesketch.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-sketch 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 vibesketch.document.io import digest_of, save
from vibesketch.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:`~vibesketch.document.io.digest_of`),
    #: so a save never replaces a change made outside vibe-sketch.
    digest: str | None = None



    @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:
        with self._lock:
            identifier = secrets.token_hex(4)
            while identifier in self._entries:
                identifier = secrets.token_hex(4)
            entry = Entry(identifier, document, path, document.revision if path else None)
            if path is not None:
                entry.digest = digest_of(path)
            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:`~vibesketch.document.io.SaveRefused` (a ``ValueError``)
        when the file changed since it was opened or last saved.
        """
        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")
            save(entry.document, entry.path, expect=entry.digest)
            entry.digest = digest_of(entry.path)
            entry.saved_revision = entry.document.revision
            return entry