eduard/cobra-boards

Cobra Boards, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Dodil Models + MCP.

• dodil-challenge
• dodil-challenge-boards

A remix template · 1 remixes · Remix eduard/cobra-boards

app/boards/application/ports.py

"""What the use cases need from the outside world. infrastructure/ provides it; tests fake it."""
from typing import Optional, Protocol

from boards.domain.model import Board, BoardList, Card, Event


class BoardRepository(Protocol):
    # boards, members, lists
    def get(self, board_id: str) -> Optional[Board]: ...
    def boards_for(self, email: str) -> list: ...
    def create_board(self, board: Board) -> None: ...
    def rename_board(self, board_id: str, title: str) -> None: ...
    def save_member(self, board_id: str, email: str, role: str) -> None: ...
    def delete_member(self, board_id: str, email: str) -> None: ...
    def save_list(self, lst: BoardList) -> None: ...

    # cards
    def get_card(self, card_id: str) -> Optional[Card]: ...
    def card_by_number(self, board_id: str, number: int) -> Optional[Card]: ...
    def cards_on(self, board_id: str) -> list: ...
    def cards_assigned_to(self, email: str) -> list: ...
    def next_number(self, board_id: str) -> int: ...
    def insert_card(self, card: Card, event: Event) -> bool: ...
    def update_card(self, card: Card, expected_version: int, event: Event) -> bool: ...
    def unassign_all(self, board_id: str, email: str, event_for) -> int: ...

    # history
    def history(self, card_id: str, limit: int = 50) -> list: ...
    def board_activity(self, board_id: str, limit: int = 50) -> list: ...
    def log(self, event: Event) -> None: ...

    # dependencies: card_id is blocked by blocker_id
    def blockers_on(self, board_id: str) -> dict: ...
    def add_dependency(self, board_id: str, card_id: str, blocker_id: str) -> None: ...
    def remove_dependency(self, card_id: str, blocker_id: str) -> None: ...

    # checklists
    def checklist(self, card_id: str) -> list: ...
    def save_checklist_item(self, card_id: str, item_id: str, text: str, done: bool,
                            position: float) -> None: ...
    def delete_checklist_item(self, card_id: str, item_id: str) -> None: ...

    # meaning search
    def save_embedding(self, card_id: str, board_id: str, vector: list, text_hash: str) -> None: ...
    def similar_cards(self, board_ids: list, vector: list, limit: int) -> list: ...
    def embedding_hash(self, card_id: str) -> Optional[str]: ...

    # live: the newest history line on a board (the page polls it to know something changed)
    def latest_event(self, board_id: str) -> Optional[Event]: ...

    # comments and the people they notify
    def add_comment(self, comment: dict, event: Event) -> None: ...
    def comments(self, card_id: str) -> list: ...
    def add_notifications(self, items: list) -> None: ...
    def notifications_for(self, email: str, limit: int = 50) -> list: ...
    def mark_read(self, email: str, ids: Optional[list], at) -> int: ...

    # attachments (the bytes live in the object store, the record here)
    def add_attachment(self, att: dict, event: Event) -> None: ...
    def attachments(self, card_id: str) -> list: ...
    def get_attachment(self, att_id: str) -> Optional[dict]: ...
    def delete_attachment(self, att_id: str, event: Event) -> None: ...

    # archive: an archived card leaves the board but keeps its number and history
    def archive_card(self, card_id: str, board_id: str, by: str, at, event: Event) -> None: ...
    def restore_card(self, card_id: str, event: Event) -> None: ...
    def archived_on(self, board_id: str) -> dict: ...
    def card_counts(self, board_id: str) -> dict: ...  # card_id -> {comments, files}


class ObjectStore(Protocol):
    """Where attachment bytes live: DataK3 object storage in production, a dict in tests."""
    def put(self, key: str, data: bytes, content_type: str) -> None: ...
    def get(self, key: str) -> bytes: ...
    def delete(self, key: str) -> None: ...


class AssistantUnavailable(Exception):
    """The model did not answer (timeout, 5xx). The board itself keeps working."""


class Assistant(Protocol):
    def checklist(self, title: str, description: str) -> list: ...
    def summary(self, board_title: str, facts: str) -> str: ...
    def embed(self, text: str) -> list: ...