ghpsolutions/quadro

Quadro: team boards with roles, history and an MCP endpoint for Claude and ChatGPT

• dodil-challenge
• dodil-challenge-boards

A remix template · 0 remixes · Remix ghpsolutions/quadro

app/boards/infrastructure/memory.py

"""An in-memory repository with the same contract as the DataK3 one.

Used by the tests and by `STORE=memory` for running the app on a laptop without a bucket.
"""
from __future__ import annotations

import copy
import threading
from typing import Optional, Sequence

from boards.application.ports import NumberTaken
from boards.domain.model import Board, BoardList, Card, Role


class MemoryBoards:
    def __init__(self):
        self._lock = threading.RLock()
        self.boards, self.members, self.lists, self.cards = {}, {}, {}, {}
        self.archived, self.comments, self.events, self.vectors = set(), [], [], {}

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

    def get_board(self, board_id: str) -> Optional[Board]:
        with self._lock:
            b = self.boards.get(board_id)
            if not b:
                return None
            b = copy.deepcopy(b)
            b.members = {e: r for (bid, e), r in self.members.items()
                         if bid == board_id and r is not Role.OWNER}
            return b

    def boards_for(self, email: str) -> list:
        with self._lock:
            ids = {bid for (bid, e) in self.members if e == email}
            return [self.get_board(i) for i in ids if i in self.boards]

    def save_member(self, board_id: str, email: str, role: Role) -> None:
        with self._lock:
            self.members[(board_id, email)] = role

    def remove_member(self, board_id: str, email: str) -> None:
        with self._lock:
            self.members.pop((board_id, email), None)

    def save_list(self, lst: BoardList) -> None:
        with self._lock:
            self.lists[lst.id] = copy.deepcopy(lst)

    def lists_on(self, board_id: str) -> list:
        with self._lock:
            return [copy.deepcopy(l) for l in self.lists.values() if l.board_id == board_id]

    def delete_list(self, list_id: str) -> None:
        with self._lock:
            self.lists.pop(list_id, None)

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

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

    def update_card(self, card: Card, expected_version: int) -> bool:
        with self._lock:
            current = self.cards.get(card.id)
            if current is None or current.version != expected_version or card.id in self.archived:
                return False
            self.cards[card.id] = copy.deepcopy(card)
            return True

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

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

    def delete_card(self, card_id: str) -> None:
        with self._lock:
            self.archived.add(card_id)

    def add_comment(self, comment: dict) -> None:
        with self._lock:
            self.comments.append(dict(comment))

    def comments_for(self, card_id: str) -> list:
        with self._lock:
            return sorted((dict(c) for c in self.comments if c["card_id"] == card_id),
                          key=lambda c: c["created_at"])

    def comment_counts(self, board_id: str) -> dict:
        with self._lock:
            out = {}
            for c in self.comments:
                if c["board_id"] == board_id:
                    out[c["card_id"]] = out.get(c["card_id"], 0) + 1
            return out

    def record(self, event: dict) -> None:
        with self._lock:
            self.events.append(copy.deepcopy(event))

    def activity(self, board_id: str, card_id: Optional[str] = None, limit: int = 50) -> list:
        with self._lock:
            rows = [copy.deepcopy(e) for e in reversed(self.events) if e["board_id"] == board_id
                    and (card_id is None or e["card_id"] == card_id)]
            return rows[:limit]

    def save_embedding(self, card_id: str, board_id: str, digest: str, vector: list) -> None:
        with self._lock:
            self.vectors[card_id] = (board_id, digest, list(vector))

    def embeddings_on(self, board_id: str) -> dict:
        with self._lock:
            return {cid: (d, v) for cid, (bid, d, v) in self.vectors.items() if bid == board_id}