rgboards/rg-boards

Lanes, team task boards you run from the web or from Claude/ChatGPT (MCP). FastAPI + DataK3 + appid + Dodil Models.

• dodil-challenge
• dodil-challenge-boards

A remix template · 0 remixes · Remix rgboards/rg-boards

app/boards/infrastructure/memory.py

"""An in-memory BoardRepository: the tests use it, and so does a local run with no Dodil
account (BOARDS_STORE=memory). Same contract as the DataK3 repository."""
import copy
import threading
import uuid
from datetime import datetime, timezone
from typing import Optional

from boards.application.ports import DuplicateKey
from boards.domain.model import Board, Card


class MemoryBoards:
    def __init__(self):
        self.boards, self.cards, self.events = {}, {}, []
        self.links, self.items, self.vectors, self.summaries = set(), {}, {}, {}
        self._lock = threading.Lock()

    def get(self, board_id: str) -> Optional[Board]:
        b = self.boards.get(board_id)
        return copy.deepcopy(b) if b else None

    def boards_for(self, email: str) -> list:
        return [copy.deepcopy(b) for b in self.boards.values()
                if b.owner_email == email or email in b.members]

    def save_board(self, board: Board) -> None:
        self.boards[board.id] = copy.deepcopy(board)

    def get_card(self, card_id: str) -> Optional[Card]:
        c = self.cards.get(card_id)
        return copy.deepcopy(c) if c else None

    def cards_on(self, board_ids: list) -> list:
        ids = set(board_ids)
        return [copy.deepcopy(c) for c in self.cards.values()
                if c.board_id in ids and not c.archived]

    def next_number(self, board_id: str) -> int:
        return max((c.number for c in self.cards.values() if c.board_id == board_id),
                   default=0) + 1

    def insert_card(self, card: Card) -> None:
        with self._lock:
            if card.id in self.cards:
                raise DuplicateKey(card.id)
            self.cards[card.id] = copy.deepcopy(card)

    def update_card(self, card: Card, expected_version: int) -> bool:
        with self._lock:    # compare-and-set, like the UPDATE ... WHERE version = ? in SQL
            cur = self.cards.get(card.id)
            if cur is None or cur.version != expected_version:
                return False
            self.cards[card.id] = copy.deepcopy(card)
            return True

    def log(self, board_id, card_id, actor, action, field=None, old=None, new=None,
            at=None) -> None:
        at = (at or datetime.now(timezone.utc)).astimezone(timezone.utc)
        self.events.append({"id": uuid.uuid4().hex, "board_id": board_id, "card_id": card_id,
                            "actor": actor, "action": action, "field": field, "old": old,
                            "new": new, "at": at.isoformat(timespec="seconds")})

    def _newest_first(self, events) -> list:
        return sorted(events, key=lambda e: e["at"], reverse=True)   # stable: ties keep order

    def history(self, card_id: str) -> list:
        return self._newest_first(e for e in reversed(self.events) if e["card_id"] == card_id)

    def activity(self, board_id: str, limit: int = 50) -> list:
        return self._newest_first(e for e in reversed(self.events)
                                  if e["board_id"] == board_id)[:limit]

    def links_on(self, board_id: str) -> list:
        return sorted((c, b) for (bid, c, b) in self.links if bid == board_id)

    def add_link(self, board_id, card_id, blocked_by, actor) -> None:
        self.links.add((board_id, card_id, blocked_by))

    def remove_link(self, card_id, blocked_by) -> None:
        self.links = {l for l in self.links if not (l[1] == card_id and l[2] == blocked_by)}

    def checklist(self, card_id: str) -> list:
        return sorted((dict(i) for i in self.items.values() if i["card_id"] == card_id),
                      key=lambda i: i["position"])

    def save_check_item(self, board_id: str, item: dict) -> None:
        self.items[item["id"]] = dict(item)

    def vectors_on(self, board_ids: list) -> dict:
        ids = set(board_ids)
        return {cid: (h, v) for cid, (bid, h, v) in self.vectors.items() if bid in ids}

    def save_vector(self, board_id, card_id, text_hash, vector) -> None:
        self.vectors[card_id] = (board_id, text_hash, list(vector))

    def get_summary(self, board_id, day):
        return self.summaries.get((board_id, day))

    def save_summary(self, board_id, day, text) -> None:
        self.summaries[(board_id, day)] = {"summary": text,
                                           "written_at": datetime.now(timezone.utc).isoformat()}