trove/trovedb/fs.py

import os
import datetime as dt
from pathlib import Path
from typing import Optional, Iterable, Iterator, override

from .trove import Note, Trove, BadNoteType, TreeEntry, NoteNotFound, ObjectId
from . import trove as tr

class FSNote(Note):
    def __init__(self, trove: 'FSTrove', path: Path):
        self._trove: FSTrove = trove
        self._fs_path: Path = path.resolve()
        if not self._fs_path.is_relative_to(trove.root):
            raise ValueError("Path must be relative to the root directory")
        self._object_id: str = path.relative_to(trove.root).as_posix()

    @property
    def object_id(self) -> tr.ObjectId:
        return self._object_id

    @property
    def fs_path(self) -> Path:
        return self._fs_path

    @property
    def mtime(self):
        """Return modification time as datetime."""
        stat = self._fs_path.stat()
        return dt.datetime.fromtimestamp(stat.st_mtime, tz=dt.timezone.utc)

    @property
    def readonly(self) -> bool:
        """Check if the note is readonly based on file permissions."""
        return not os.access(self._fs_path, os.W_OK)

    @property
    def mime(self) -> str:
        """Return MIME type, defaulting to generic binary stream."""
        if self._fs_path.is_dir():
            return "inode/directory"
        return "application/octet-stream"

    def get_raw_metadata(self, key: str) -> Optional[bytes]:
        # TODO: FIXME
        return None

    def set_raw_metadata(self, key: str, value: bytes) -> None:
        # TODO: FIXME
        pass

    def read_content(self) -> bytes:
        """Read the raw content of the note."""
        if self._fs_path.is_file():
            return self._fs_path.read_bytes()
        return b""

    def write_content(self, data:bytes) -> None:
        """Write the raw content of the note."""
        self._fs_path.write_bytes(data)

    def _new_child_subdir(self, name: str, exist_ok: bool = True) -> Path:
        ex_path = self._fs_path / name
        try:
            ex_path.mkdir(exist_ok=exist_ok)
            return ex_path
        except FileExistsError:
            raise tr.ErrorExists(str(ex_path)) from None
        except OSError as e:
            raise tr.ErrorWithErrno(e.errno, str(e)) from None

    def new_child(self, name: str, mime: str, content: bytes | None, executable: bool, hidden: bool) -> Note:
        """Create a new child note."""

        if content is None:
            content = b""

        if mime == 'inode/directory':
            if content:
                raise NotImplementedError("FSNote does not support children")
            return FSNote(self._trove, self._new_child_subdir(name, False))

        ex_path = self._fs_path / name
        ex_path.write_bytes(content)
        return FSNote(self._trove, ex_path)

    def child(self, name: str) -> Note:
        """Retrieve a child not by name."""
        target_path = self._fs_path / name
        return self._trove.get_raw_note_by_path(target_path)

    def rm_child(self, name: str, recurse: bool):
        target_path = self._fs_path / name
        if not target_path.exists():
            raise tr.ErrorNotFound(name)
        if target_path.is_dir():
            if recurse:
                raise NotImplementedError("Recursive deletion not supported")
            else:
                target_path.rmdir()
        else:
            target_path.unlink()
            # TODO: remove meta directory!

    def unlink_(self, name: str):
        target_path = self._fs_path / name
        if not target_path.exists():
            return
        if target_path.is_dir():
            target_path.rmdir()
        else:
            target_path.unlink()

    def link_(self, name: str, note: Note):
        if not isinstance(note, FSNote):
            raise BadNoteType("Only blob notes can be linked")

        target_path = self._fs_path / name
        if target_path.exists():
            self.unlink_(name)

        note_path = note._fs_path

        # If the note is in .working, move it to the new location.
        if self._trove.working in note_path.parents:
            os.rename(note_path, target_path)
        else:
            # If it's already linked somewhere, create a hard link if it's a file.
            if note_path.is_file():
                try:
                    os.link(note_path, target_path)
                except OSError:
                    # Fallback to rename if link fails (e.g. cross-device, though we assume single FS)
                    os.rename(note_path, target_path)
            else:
                # Directories cannot be hardlinked.
                # We move it to the new location.
                os.rename(note_path, target_path)

    def children(self) -> Iterator[TreeEntry]:
        """Get all children of this note."""
        if not self._fs_path.is_dir():
            return
        for item in self._fs_path.iterdir():
            if item.name == ".trove":
                continue
            yield TreeEntry(name=item.name, object_id=str(self._fs_path / item.name))


class FSTrove(Trove):
    def __init__(self, root: str | Path):
        self.root = Path(root).absolute()
        self._root_inode = self.root.stat().st_ino
        self.dot_trove = self.root / ".trove"
        self.working = self.dot_trove / ".working"
        self.dot_trove.mkdir(exist_ok=True)
        self.working.mkdir(exist_ok=True)
        
    @classmethod
    def open(cls, path: str | Path, create: bool = False) -> 'FSTrove':
        p = Path(path)
        if not p.exists():
            if not create:
                raise FileNotFoundError(f"Root path not found: {p}")
            p.mkdir(parents=True)
        return cls(p)

    def get_raw_note_by_path(self, path: Path) -> Note:
        if not path.exists():
            raise tr.ErrorNotFound(str(path))
        return FSNote(self, path=path)

    def get_raw_note(self, note_id: ObjectId) -> Note:
        p = self.root / str(note_id)
        if not p.exists():
            raise NoteNotFound(note_id)
        return self.get_raw_note_by_path(p)

    @override
    def move(self, src_parent: Note, src_name: str, dst_parent: Note, dst_name: str, overwrite: bool):
        """Move a child note to a new location."""
        src_note = src_parent.child(src_name)
        if not isinstance(src_note, FSNote):
            raise tr.ErrorBadType("not a valid DB note")
        if not isinstance(dst_parent, FSNote):
            raise tr.ErrorBadType("not a valid DB note")
        if not dst_parent._fs_path.is_dir():
            raise NotImplementedError("FSTrove promoting during move")

        # Remove existing target
        if overwrite:
            dst_parent.unlink_(dst_name)

        # Link to new parent, unlink from old
        dst_parent.link_(dst_name, src_note)
        src_parent.rm_child(src_name, True)

    def create_blob(self, data: bytes | None = None) -> Note:
        raise NotImplementedError("FSTrove does not support blobs")

    def get_root(self) -> Note:
        return FSNote(self, path=self.root)

    def _get_metadata(self, inode: int, key: str) -> Optional[bytes]:
        raise NotImplementedError("FSTrove does not support metadata")

    def _set_metadata(self, inode: int, key: str, value: bytes):
        raise NotImplementedError("FSTrove does not support metadata")

    def close(self):
        pass