eduard/cobra-boards

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

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

app/boards/domain/model.py

"""The board's language and rules. No database, no HTTP: plain Python you can test in a second.

Every rule lives here once. The web page and Claude/ChatGPT reach it through the same use cases,
so a viewer can't edit through one door because the other door forgot to check.
"""
import re
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from enum import Enum
from typing import Optional

DEFAULT_LISTS = (("To do", False), ("Doing", False), ("Done", True))  # (name, is_done)
# Board templates: the lists a new board starts with (name, is_done). "basic" is DEFAULT_LISTS.
TEMPLATES = {
    "basic": DEFAULT_LISTS,
    "sprint": (("Backlog", False), ("This sprint", False), ("In progress", False),
               ("Review", False), ("Done", True)),
    "launch": (("Ideas", False), ("Planned", False), ("Building", False), ("Testing", False),
               ("Launched", True)),
    "onboarding": (("Before day one", False), ("First week", False), ("First month", False),
                   ("Done", True)),
    "support": (("New", False), ("Investigating", False), ("Waiting on customer", False),
                ("Resolved", True)),
}
MAX_TITLE = 200
MAX_TEXT = 5000
MAX_LABELS = 10
MAX_COMMENT = 2000
MAX_ATTACHMENT = 10 * 1024 * 1024   # 10 MB per file
MAX_ATTACHMENTS = 20                # per card


class Role(str, Enum):
    OWNER = "owner"
    MEMBER = "member"
    VIEWER = "viewer"


class Forbidden(Exception):
    """The person may not do this. The doors turn it into 403 (web) or a tool error (MCP)."""


class Invalid(Exception):
    """The request breaks a rule of the board itself. 422 / tool error."""


class Conflict(Exception):
    """The card changed since the caller read it. 409: nothing was written."""


def now() -> datetime:
    return datetime.now(timezone.utc).replace(microsecond=0)


def clean_email(email: str) -> str:
    e = (email or "").strip().lower()
    if "@" not in e or e.startswith("@") or e.endswith("@") or " " in e:
        raise Invalid(f"{email!r} is not an email address")
    return e


def clean_text(value: Optional[str], what: str, limit: int, required: bool = False) -> str:
    v = (value or "").strip()
    if required and not v:
        raise Invalid(f"{what} can't be empty")
    if len(v) > limit:
        raise Invalid(f"{what} is longer than {limit} characters")
    return v


def clean_labels(labels) -> list:
    out = []
    for raw in labels or []:
        label = clean_text(str(raw), "a label", 30).lower()
        if label and label not in out:
            out.append(label)
    if len(out) > MAX_LABELS:
        raise Invalid(f"a card takes at most {MAX_LABELS} labels")
    return out


@dataclass(frozen=True)
class Person:
    """Who is acting: taken from the gateway's sign-in, never from the request body."""
    email: str


@dataclass
class BoardList:
    id: str
    board_id: str
    name: str
    position: float
    is_done: bool = False  # cards here are finished: never overdue, never blocking


@dataclass
class Board:
    id: str
    title: str
    owner_email: str
    members: dict = field(default_factory=dict)  # email -> Role (the owner is implied)
    lists: list = field(default_factory=list)    # BoardList, kept in position order
    created_at: Optional[datetime] = None

    # --- who may do what -------------------------------------------------------------------
    def role_of(self, person: Person) -> Optional[Role]:
        if person.email == self.owner_email:
            return Role.OWNER
        return self.members.get(person.email)

    def require_view(self, person: Person) -> Role:
        role = self.role_of(person)
        if role is None:
            raise Forbidden("you are not on this board")
        return role

    def require_edit(self, person: Person) -> None:
        if self.require_view(person) is Role.VIEWER:
            raise Forbidden("viewers can look but can't change the board")

    def require_owner(self, person: Person) -> None:
        if self.role_of(person) is not Role.OWNER:
            raise Forbidden("only the board's owner can do that")

    def people(self) -> list:
        """Everyone on the board, owner first: who a card may be assigned to."""
        return [self.owner_email] + sorted(self.members)

    # --- members ---------------------------------------------------------------------------
    def set_member(self, by: Person, email: str, role: Role) -> str:
        self.require_owner(by)
        email = clean_email(email)
        if role is Role.OWNER:
            raise Invalid("a board has exactly one owner")
        if email == self.owner_email:
            raise Invalid("that's the owner; the owner's role can't change")
        self.members[email] = role
        return email

    def remove_member(self, by: Person, email: str) -> str:
        self.require_owner(by)
        email = clean_email(email)
        if email == self.owner_email:
            raise Invalid("the owner can't be removed from their own board")
        if email not in self.members:
            raise Invalid(f"{email} is not on this board")
        del self.members[email]
        return email

    # --- lists -----------------------------------------------------------------------------
    def list_named(self, name: str) -> BoardList:
        """Finds a list by name, forgiving case and spacing ('doing' finds 'Doing')."""
        key = " ".join((name or "").split()).lower()
        for lst in self.lists:
            if lst.name.lower() == key:
                return lst
        names = ", ".join(l.name for l in self.lists)
        raise Invalid(f"there's no list called {name!r}; this board has {names}")

    def list_by_id(self, list_id: str) -> BoardList:
        for lst in self.lists:
            if lst.id == list_id:
                return lst
        raise Invalid("that list is not on this board")

    def done_list_ids(self) -> set:
        return {l.id for l in self.lists if l.is_done}

    def check_new_list_name(self, name: str, except_id: Optional[str] = None) -> str:
        name = clean_text(name, "a list name", 60, required=True)
        if any(l.name.lower() == name.lower() and l.id != except_id for l in self.lists):
            raise Invalid(f"this board already has a list called {name!r}")
        return name


@dataclass
class Card:
    id: str                 # "<board_id>-<number>": a natural key, never minted by the database
    board_id: str
    number: int             # short and human: "card 12" in a chat means this
    list_id: str
    title: str
    description: str = ""
    assignee: Optional[str] = None
    due_date: Optional[date] = None
    labels: list = field(default_factory=list)
    position: float = 0.0
    version: int = 1        # every write bumps it; a stale write is refused (Conflict)
    created_by: str = ""
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

    def is_overdue(self, board: Board, today: date) -> bool:
        return bool(self.due_date and self.due_date < today
                    and self.list_id not in board.done_list_ids())

    def is_done(self, board: Board) -> bool:
        return self.list_id in board.done_list_ids()

    def snapshot(self, board: Board) -> dict:
        """What the history compares, with list names a person can read."""
        try:
            list_name = board.list_by_id(self.list_id).name
        except Invalid:
            list_name = self.list_id
        return {"title": self.title, "description": self.description, "assignee": self.assignee,
                "due_date": self.due_date.isoformat() if self.due_date else None,
                "labels": list(self.labels), "list": list_name}


def changes_between(before: dict, after: dict) -> dict:
    """Field -> [old, new], only for what actually changed. This is what the history keeps."""
    return {k: [before.get(k), after.get(k)] for k in after if before.get(k) != after.get(k)}


def describe(changes: dict) -> str:
    """A one-line, human summary of a change, for the activity feed and for Claude."""
    parts = []
    for k, (old, new) in changes.items():
        if k == "list":
            parts.append(f"moved from {old} to {new}")
        elif k == "assignee":
            parts.append(f"assigned to {new}" if new else f"unassigned {old}")
        elif k == "due_date":
            parts.append(f"due {new}" if new else "removed the due date")
        elif k == "labels":
            parts.append("labels: " + (", ".join(new) if new else "none"))
        elif k == "description":
            parts.append("edited the description")
        elif k == "title":
            parts.append(f"renamed to {new!r}")
        else:
            parts.append(f"{k} changed")
    return "; ".join(parts)


@dataclass
class Event:
    """One line of history: who did what to which card, when, and exactly what changed."""
    id: str
    board_id: str
    card_id: Optional[str]
    actor: str
    action: str             # created / updated / moved / commented / member_added ...
    summary: str
    changes: dict = field(default_factory=dict)
    at: Optional[datetime] = None


_MENTION = re.compile(r"(?<![\w.])@([\w.+-]+(?:@[\w-]+(?:\.[\w-]+)+)?)")


def mentions_in(text: str, people: list) -> list:
    """Who a comment @mentions, among the board's people. "@bo" finds bo@cobra.test when it is
    the only person whose address starts with "bo"; a full "@bo@cobra.test" always works. A name
    that matches nobody, or more than one person, mentions nobody: no guessing."""
    out = []
    for raw in _MENTION.findall(text or ""):
        key = raw.lower().rstrip(".")
        hits = [p for p in people if p == key] or \
               [p for p in people if p.split("@")[0] == key] or \
               [p for p in people if p.split("@")[0].startswith(key)]
        if len(hits) == 1 and hits[0] not in out:
            out.append(hits[0])
    return out


SAFE_INLINE = {"image/png", "image/jpeg", "image/gif", "image/webp"}  # shown in the page; others download


def clean_filename(name: str) -> str:
    """A file name that is safe to store and to send back: no folders, no control characters."""
    base = (name or "").replace("\\", "/").split("/")[-1].strip()
    base = re.sub(r"[\x00-\x1f\x7f\"<>|:*?]", "", base)[:120]
    if not base or base in (".", ".."):
        raise Invalid("the file needs a name")
    return base


def would_cycle(blocked: str, blocker: str, edges: dict) -> bool:
    """True if 'blocker blocks blocked' closes a loop. edges: card -> set of its blockers."""
    if blocked == blocker:
        return True
    seen, stack = set(), [blocker]
    while stack:
        cur = stack.pop()
        if cur == blocked:
            return True
        if cur in seen:
            continue
        seen.add(cur)
        stack.extend(edges.get(cur, ()))
    return False