favro-sync/favro_sync/favro_fuse.py

import dataclasses
import errno
import re
import stat
from collections.abc import Iterator
from logging import getLogger

import fuse

from .favro_client import FavroClient
from .favro_data_model import Card, SeqId
from .favro_markdown import format_card, CardContents, parse_card_contents

logger = getLogger(__name__)

fuse.fuse_python_api = (0, 2)

class FavroStat(fuse.Stat):
    def __init__(self):
        self.st_mode = 0
        self.st_ino = 0
        self.st_dev = 0
        self.st_nlink = 0
        self.st_uid = 0
        self.st_gid = 0
        self.st_size = 0
        self.st_atime = 0
        self.st_mtime = 0
        self.st_ctime = 0


CARD_FILENAME_FORMAT = 'PAR-{seq_id}.md'
CARD_FILENAME_REGEX = r'^\/PAR\-(\d+)\.md$'


@dataclasses.dataclass(frozen=True)
class Thing:
    @staticmethod
    def from_path(path_str: str) -> 'Thing | None':
        if path_str == '/':
            return RootThing()
        if m := re.match(CARD_FILENAME_REGEX, path_str):
            return CardThing(SeqId(int(m.group(1))))
        return None


@dataclasses.dataclass(frozen=True)
class RootThing(Thing):
    pass


@dataclasses.dataclass(frozen=True)
class CardThing(Thing):
    seq_id: SeqId

class FavroFuse(fuse.Fuse):
    """Favro Filesystem in Userspace."""

    def __init__(self, favro_client: FavroClient, **kwargs):
        self.favro_client = favro_client
        super().__init__(**kwargs)

    def getattr(self, path: str) -> FavroStat | int:
        thing = Thing.from_path(path)

        st = FavroStat()
        if isinstance(thing, RootThing):
            st.st_mode = stat.S_IFDIR | 0o755
            st.st_nlink = 2
        elif isinstance(thing, CardThing):
            card = self.favro_client.get_card(thing.seq_id)

            st.st_mode = stat.S_IFREG | 0o666
            st.st_nlink = 1
            st.st_size = len(format_card(card))
            st.st_ctime = int(card.creation_date.timestamp())
            st.st_mtime = st.st_ctime  # TODO
        else:
            return -errno.ENOENT
        return st

    def readdir(self, path: str, offset: int) -> Iterator[fuse.Direntry]:
        logger.warning('readdir(path=%s, offset=%s)',path, offset)
        yield fuse.Direntry('.')
        yield fuse.Direntry('..')

        for card in self.favro_client.get_todo_list_cards():
            yield fuse.Direntry(CARD_FILENAME_FORMAT.format(seq_id=card.seq_id.raw_id))

    def open(self, path: str, flags) -> int | None:
        thing = Thing.from_path(path)
        if not isinstance(thing, CardThing):
            return -errno.ENOENT
        return None

    def read(self, path: str, size: int, offset: int) -> bytes | int:
        # Check that this is a card thing.
        thing = Thing.from_path(path)
        if not isinstance(thing, CardThing):
            return -errno.ENOENT

        card = self.favro_client.get_card(thing.seq_id)

        contents_str = format_card(card)
        contents = bytes(contents_str, 'utf8')

        slen = len(contents)
        if offset < slen:
            if offset + size > slen:
                size = slen - offset
            buf = contents[offset : offset + size]
        else:
            buf = b''
        return buf

    def write(self, path: str, written_buffer: bytes, offset: int) -> int:
        # Check that this is a card thing.
        thing = Thing.from_path(path)
        if not isinstance(thing, CardThing):
            return -errno.ENOENT

        card = self.favro_client.get_card(thing.seq_id)

        # Splice contents
        contents_str = format_card(card)
        contents = bytes(contents_str, 'utf8')
        contents = splice(contents, written_buffer, offset)
        contents_str = contents.decode('utf8')

        # Write to favro
        card_updated = parse_card_contents(contents_str)
        self.favro_client.update_card_contents(card.card_id, card_updated)

        # Return amount written
        return len(written_buffer)

    def truncate(self, path: str, new_size: int):
        # Check that this is a card thing.
        thing = Thing.from_path(path)
        if not isinstance(thing, CardThing):
            return -errno.ENOENT

        card = self.favro_client.get_card(thing.seq_id)

        # Splice contents
        contents_str = format_card(card)
        contents = bytes(contents_str, 'utf8')
        old_size = len(contents)
        contents = contents[0:new_size] + b' ' * (old_size - new_size)
        assert len(contents) == old_size
        contents_str = contents.decode('utf8')

        # Write to favro
        card_updated = parse_card_contents(contents_str)
        self.favro_client.update_card_contents_locally(card.card_id, card_updated)

        # Return amount written
        return 0


def splice(original_buffer: bytes, input_buffer: bytes, offset: int) -> bytes:
    return (
        original_buffer[0 : offset - 1]
        + input_buffer
        + original_buffer[offset + len(input_buffer) + 1 : len(original_buffer)]
    )


HELP = (
    """
Userspace hello example

"""
    + fuse.Fuse.fusage
)


def start_favro_fuse(favro_client: FavroClient):
    # TODO:
    server = FavroFuse(
        favro_client=favro_client,
        version='%prog ' + fuse.__version__,
        usage=HELP,
        dash_s_do='setsingle',
    )
    server.parse(errex=1)
    server.main()