akif/arclet-boards
Arclet Boards — private Kanban workspaces with AppID, DataK3, shared web and MCP actions.
A remix template · 1 remixes · Remix akif/arclet-boards
app/application/boards.py
"""Authorized use cases shared by both doors. All mutations and their history
commit in a single DataK3 transaction. No route can bypass these rules."""
from dataclasses import asdict, dataclass
from hashlib import sha256
import json
import re
from uuid import uuid4
from pydantic import ValidationError
from app.application.contracts import READS, WRITES, AI_ACTIONS
from app.domain.dependencies import require_acyclic, blocking_paths
from app.domain import model as m
@dataclass
class Outcome:
data: object
board_id: str | None = None
workspace_id: str | None = None
permission: str = "edit"
class Boards:
def __init__(self, store, assistant=None):
self.store, self.assistant = store, assistant
def call(self, action: str, person: m.Person | None, payload: dict | None = None,
request_id: str | None = None):
if not isinstance(person, m.Person) or not person.sub or not person.email:
raise m.Unauthenticated("Sign in as a person first")
contract = (READS | WRITES).get(action)
if contract is None:
raise m.NotFound("Action not found")
try:
args = contract.model_validate(payload or {}).model_dump()
except ValidationError as error:
raise m.Invalid("Invalid action input: " + "; ".join(
f"{'.'.join(map(str, e['loc']))}: {e['msg']}" for e in error.errors())) from None
write = action in WRITES
if write and (not request_id or not re.fullmatch(r"[A-Za-z0-9_-]{8,128}", request_id)):
raise m.Invalid("Supply an Idempotency-Key of 8–128 letters, digits, _ or -")
canonical = json.dumps([action, args], sort_keys=True, separators=(",", ":"))
digest = sha256(canonical.encode()).hexdigest()
command_id = sha256(f"{person.sub}:{request_id}".encode()).hexdigest()
def prior_result(tx):
if write:
prior = tx.get("commands", command_id, guard=True)
if prior:
if prior["payload_hash"] != digest:
raise m.Conflict("This request ID was already used for a different action")
if prior.get("board_id"):
self._board(tx, person, prior["board_id"], prior["permission"])
elif prior.get("workspace_id"):
self._workspace(tx, person, prior["workspace_id"], prior["permission"])
return json.loads(prior["result_json"])
return None
prepared = None
if action in AI_ACTIONS:
prior = self.store.run(prior_result) if write else None
if prior is not None:
return prior
prepared = self._prepare_ai(action, person, args)
def perform(tx):
prior = prior_result(tx)
if prior is not None:
return prior
extra = {"prepared": prepared} if action in AI_ACTIONS else {}
outcome = getattr(self, "_" + action)(tx, person, **args, **extra)
if write:
tx.insert("commands", {
"id": command_id, "actor_sub": person.sub, "action": action,
"payload_hash": digest, "board_id": outcome.board_id,
"workspace_id": outcome.workspace_id, "permission": outcome.permission,
"result_json": json.dumps(outcome.data), "created_at": m.now(),
})
return outcome.data
return self.store.run(perform, write=write)
@staticmethod
def _content_hash(card):
return sha256(json.dumps(["jina-embeddings-v4:512:SEMANTIC_SIMILARITY",
card["title"],card["description"]],ensure_ascii=False).encode()).hexdigest()
def _prepare_ai(self, action, person, args):
if self.assistant is None:
raise m.Unavailable("Dodil Models is not configured")
def snapshot(tx):
if action == "suggest_checklist":
board,card = self._card(tx,person,args["card_id"])
return {"board_id":board["id"],"card":card,"board_version":board["version"]}
permission = "edit" if action == "refresh_semantic_index" else "view"
board = self._board(tx,person,args["board_id"],permission)
if action == "refresh_semantic_index":
m.check_version(board,args["version"])
cards = tx.find("cards",board_id=board["id"])
data = {"board":board,"cards":cards,"lists":tx.find("board_lists",board_id=board["id"])}
if action == "refresh_semantic_index":
if len(cards) > 100:
raise m.Invalid("Semantic indexing currently supports boards with up to 100 cards")
cached = {row["card_id"]:row for row in tx.find("card_embeddings",board_id=board["id"])}
data["changed"] = [card for card in cards if cached.get(card["id"],{}).get(
"content_hash") != self._content_hash(card)]
else:
data["activity"] = tx.find("card_activity",board_id=board["id"])
return data
data = self.store.run(snapshot)
# Model calls happen outside database transactions. The final transaction
# re-authorizes and checks versions before exposing or persisting output.
if action == "suggest_checklist":
data["items"] = self.assistant.checklist(data["card"]["title"],data["card"]["description"])
elif action == "daily_summary":
done = {row["id"] for row in data["lists"] if row["is_done"]}
eligible = [card for card in data["cards"] if card["list_id"] not in done]
candidates = sorted(eligible,key=lambda card:(not m.overdue(card,False),
card["due_date"] or "9999-12-31",card["id"]))[:100]
choices = self.assistant.focus([{"id":card["id"],"title":card["title"],
"due_date":card["due_date"],"overdue":m.overdue(card,False)} for card in candidates]) if candidates else []
if not set(choices) <= {card["id"] for card in candidates}:
raise m.Unavailable("The model returned an unknown card; summary was discarded")
data["focus_ids"],data["focus_pool_size"] = choices,len(candidates)
else:
data["vectors"] = {}
for start in range(0,len(data["changed"]),32):
batch = data["changed"][start:start+32]
texts = [card["title"]+"\n"+card["description"][:4000] for card in batch]
vectors = self.assistant.embed(texts)
if len(vectors) != len(batch):
raise m.Unavailable("Dodil returned an incomplete embedding batch")
data["vectors"].update({card["id"]:vector for card,vector in zip(batch,vectors)})
return data
def _suggest_checklist(self, tx, person, card_id, prepared):
board,card = self._card(tx,person,card_id)
m.check_version(card,prepared["card"]["version"])
m.check_version(board,prepared["board_version"])
items = [m.text(item,"Suggested item",240) for item in prepared["items"]]
if not 1 <= len(items) <= 8:
raise m.Unavailable("The model did not return a valid draft")
return Outcome({"card_id":card_id,"version":card["version"],
"suggested_checklist":items,"source":"Dodil Models · " +
getattr(self.assistant,"chat_model","kimi-k2.6"),"persisted":False})
def _daily_summary(self, tx, person, board_id, prepared):
board = self._board(tx,person,board_id)
m.check_version(board,prepared["board"]["version"])
done = {row["id"] for row in prepared["lists"] if row["is_done"]}
cards = {card["id"]:card for card in prepared["cards"]}
day = m.now()[:10]
counts = {"total":len(cards),"completed":sum(card["list_id"] in done for card in cards.values()),
"overdue":sum(m.overdue(card,card["list_id"] in done) for card in cards.values()),
"changed_today":len({row["card_id"] for row in prepared["activity"] if row["created_at"][:10] == day})}
focus = [{"id":cards[cid]["id"],"title":cards[cid]["title"],"due_date":cards[cid]["due_date"]}
for cid in prepared["focus_ids"]]
summary = (f"{counts['total']} cards, {counts['completed']} complete, {counts['overdue']} overdue. "
f"{counts['changed_today']} cards touched today (UTC).")
if focus:
summary += " Suggested focus: " + "; ".join(card["title"] for card in focus) + "."
return Outcome({"board_id":board_id,"day":day,"counts":counts,"focus_cards":focus,
"summary":summary,"focus_pool_size":prepared["focus_pool_size"],
"source":"Live board counts; focus selected by Dodil Models from existing card IDs"})
def _refresh_semantic_index(self, tx, person, board_id, version, prepared):
board = self._board(tx,person,board_id,"edit")
m.check_version(board,version)
for card in prepared["changed"]:
values = {"id":card["id"],"board_id":board_id,"card_id":card["id"],
"content_hash":self._content_hash(card),"model":"jina-embeddings-v4","dimensions":512,
"embedding":prepared["vectors"][card["id"]],"updated_at":m.now()}
if tx.get("card_embeddings",card["id"]):
tx.update("card_embeddings",card["id"],{k:v for k,v in values.items() if k != "id"})
else:
tx.insert("card_embeddings",values)
return Outcome({"board_id":board_id,"indexed_count":len(prepared["cards"]),
"updated_count":len(prepared["changed"])},board_id=board_id)
def _find_similar_cards(self, tx, person, card_id):
board,card = self._card(tx,person,card_id)
source = tx.get("card_embeddings",card_id)
if not source or source["content_hash"] != self._content_hash(card):
return Outcome({"card_id":card_id,"board_id":board["id"],"needs_index":True,"cards":[]})
current = {row["id"]:row for row in tx.find("cards",board_id=board["id"])}
results = []
for match in tx.similar_cards(board["id"],card_id,source["embedding"]):
candidate = current.get(match["card_id"])
if candidate and match["content_hash"] == self._content_hash(candidate):
results.append({"id":candidate["id"],"title":candidate["title"],
"description":candidate["description"],"distance":float(match["distance"])})
if len(results) == 5:
break
return Outcome({"card_id":card_id,"board_id":board["id"],"needs_index":False,
"cards":results,"source":"Dodil embeddings · DataK3 cosine vector search"})
@staticmethod
def _role(tx, table, parent, row, person):
if row["owner_sub"] == person.sub:
return "owner"
members = tx.find(table, **{parent: row["id"]})
for membership in members:
if membership.get("user_sub") == person.sub:
return membership["role"]
if not membership.get("user_sub") and membership["email"] == person.email:
# An invitation is addressed by verified AppID email once. Bind
# it to the immutable pool subject; later email changes cannot
# transfer an already accepted membership to a different user.
claimed = tx.get(table, membership["id"], guard=True)
if claimed.get("user_sub") not in (None, person.sub):
continue
tx.update(table, membership["id"], {"user_sub": person.sub, "updated_at": m.now()})
return membership["role"]
return None
def _board(self, tx, person, board_id, permission="view"):
board = tx.get("boards", board_id, guard=True)
if not board:
raise m.NotFound("Resource not found")
role = self._role(tx, "board_memberships", "board_id", board, person)
m.require_role(role, permission)
return {**board, "role": role}
def _workspace(self, tx, person, workspace_id, permission="view"):
workspace = tx.get("workspaces", workspace_id, guard=True)
if not workspace:
raise m.NotFound("Resource not found")
role = self._role(tx, "workspace_memberships", "workspace_id", workspace, person)
m.require_role(role, permission)
return {**workspace, "role": role}
def _card(self, tx, person, card_id, permission="view"):
card = tx.get("cards", card_id, guard=True)
if not card:
raise m.NotFound("Resource not found")
board = self._board(tx, person, card["board_id"], permission)
return board, card
@staticmethod
def _insert(tx, table, entity, **kwargs):
row = asdict(entity(**kwargs))
tx.insert(table, row)
return row
@staticmethod
def _touch(tx, table, row):
changes = {"version": row["version"] + 1, "updated_at": m.now()}
tx.update(table, row["id"], changes)
row.update(changes)
@staticmethod
def _event(tx, person, card, action, changes):
if not changes:
return
tx.insert("card_activity", asdict(m.CardActivity(
id=uuid4().hex, board_id=card["board_id"], card_id=card["id"],
actor_sub=person.sub, actor_email=person.email, action=action,
changes_json=json.dumps(changes), created_at=m.now())))
def _visible_boards(self, tx, person):
# Select candidate ids through identity-owned memberships first. An org's
# shared service account must never be mistaken for row-level security.
owned = tx.find("boards", owner_sub=person.sub)
members = tx.find("board_memberships", user_sub=person.sub)
pending = tx.find("board_memberships", email=person.email)
ids = {b["id"] for b in owned} | {b["board_id"] for b in members + pending
if not b.get("user_sub") or b["user_sub"] == person.sub}
return [self._board(tx, person, bid) for bid in sorted(ids)]
def _list_boards(self, tx, person):
boards = self._visible_boards(tx, person)
for board in boards:
lists = tx.find("board_lists", board_id=board["id"])
cards = tx.find("cards", board_id=board["id"])
done = {row["id"] for row in lists if row["is_done"]}
board.update(card_count=len(cards), completed_count=sum(
card["list_id"] in done for card in cards), list_count=len(lists))
return Outcome(sorted(boards, key=lambda b: b["updated_at"], reverse=True))
def _list_workspaces(self, tx, person):
own = tx.find("workspaces", owner_sub=person.sub)
memberships = tx.find("workspace_memberships", email=person.email)
memberships += tx.find("workspace_memberships", user_sub=person.sub)
ids = {w["id"] for w in own} | {w["workspace_id"] for w in memberships
if not w.get("user_sub") or w["user_sub"] == person.sub}
visible = self._visible_boards(tx, person)
ids |= {b["workspace_id"] for b in visible}
workspaces = []
for wid in sorted(ids):
row = tx.get("workspaces", wid)
if row:
role = self._role(tx, "workspace_memberships", "workspace_id", row, person)
workspaces.append({**row, "role": role or "guest"})
return Outcome(workspaces)
def _get_workspace(self, tx, person, workspace_id):
available = self._list_workspaces(tx, person).data
row = next((w for w in available if w["id"] == workspace_id), None)
if not row:
raise m.NotFound("Resource not found")
row["boards"] = [b for b in self._list_boards(tx, person).data
if b["workspace_id"] == workspace_id]
if row["role"] == "owner":
row["members"] = [{"email": row["owner_email"], "role": "owner"}] + [
{"email": member["email"], "role": member["role"]}
for member in tx.find("workspace_memberships", workspace_id=workspace_id)]
return Outcome(row)
def _create_workspace(self, tx, person, title):
stamp = m.now()
row = self._insert(tx, "workspaces", m.Workspace, id=uuid4().hex,
title=m.text(title, "Workspace name", 80), owner_sub=person.sub,
owner_email=person.email, created_at=stamp, updated_at=stamp)
return Outcome({**row, "role": "owner"}, workspace_id=row["id"], permission="owner")
def _invite_workspace_member(self, tx, person, workspace_id, email, role):
workspace = self._workspace(tx, person, workspace_id, "owner")
email, role = m.email(email), m.member_role(role)
if email == workspace["owner_email"]:
raise m.Invalid("The workspace owner already has access")
key = sha256(f"{workspace_id}:{email}".encode()).hexdigest()
old = tx.get("workspace_memberships", key)
if old:
tx.update("workspace_memberships", key, {"role": role, "updated_at": m.now()})
else:
stamp = m.now()
self._insert(tx, "workspace_memberships", m.WorkspaceMembership,
id=key, workspace_id=workspace_id, email=email, user_sub=None,
role=role, invited_by=person.sub, created_at=stamp, updated_at=stamp)
self._touch(tx, "workspaces", workspace)
return Outcome({"workspace_id": workspace_id, "email": email, "role": role},
workspace_id=workspace_id, permission="owner")
def _create_board(self, tx, person, workspace_id, title, description, color):
self._workspace(tx, person, workspace_id, "edit")
stamp, bid = m.now(), uuid4().hex
row = self._insert(tx, "boards", m.Board, id=bid, workspace_id=workspace_id,
title=m.text(title, "Board name", 100), description=m.text(
description, "Description", 10000, True), color=m.color(color),
owner_sub=person.sub, owner_email=person.email, created_at=stamp, updated_at=stamp)
for index, name in enumerate(("To do", "In progress", "Done")):
self._insert(tx, "board_lists", m.BoardList, id=uuid4().hex, board_id=bid,
title=name, position=index, is_done=index == 2,
created_at=stamp, updated_at=stamp)
return Outcome({**row, "role": "owner"}, board_id=bid, permission="owner")
def _update_board(self, tx, person, board_id, title, description, color, version):
board = self._board(tx, person, board_id, "edit")
m.check_version(board, version)
changes = {"title": m.text(title, "Board name", 100), "description": m.text(
description, "Description", 10000, True), "color": m.color(color)}
tx.update("boards", board_id, changes)
board.update(changes)
self._touch(tx, "boards", board)
return Outcome(board, board_id=board_id)
def _get_board(self, tx, person, board_id):
board = self._board(tx, person, board_id)
lists = sorted(tx.find("board_lists", board_id=board_id), key=lambda r: (r["position"], r["id"]))
cards = sorted(tx.find("cards", board_id=board_id), key=lambda r: (r["position"], r["id"]))
labels = tx.find("card_labels", board_id=board_id)
done = {row["id"] for row in lists if row["is_done"]}
for card in cards:
card["labels"] = [{"name": r["name"], "color": r["color"]} for r in labels
if r["card_id"] == card["id"]]
card["overdue"] = m.overdue(card, card["list_id"] in done)
board.update(lists=lists, cards=cards, members=self._members(tx, board))
return Outcome(board)
@staticmethod
def _members(tx, board):
return [{"email": board["owner_email"], "role": "owner"}] + sorted([
{"email": row["email"], "role": row["role"]}
for row in tx.find("board_memberships", board_id=board["id"])], key=lambda r: r["email"])
def _list_members(self, tx, person, board_id):
board = self._board(tx, person, board_id)
return Outcome(self._members(tx, board))
def _create_list(self, tx, person, board_id, title, is_done):
board = self._board(tx, person, board_id, "edit")
lists = tx.find("board_lists", board_id=board_id)
if len(lists) >= 100:
raise m.Invalid("A board can contain at most 100 lists")
stamp = m.now()
row = self._insert(tx, "board_lists", m.BoardList, id=uuid4().hex,
board_id=board_id, title=m.text(title, "List name", 80), position=len(lists),
is_done=is_done, created_at=stamp, updated_at=stamp)
self._touch(tx, "boards", board)
return Outcome(row, board_id=board_id)
def _update_list(self, tx, person, list_id, title, is_done, version):
row = tx.get("board_lists", list_id, guard=True)
if not row:
raise m.NotFound("Resource not found")
board = self._board(tx, person, row["board_id"], "edit")
m.check_version(row, version)
changes = {"title": m.text(title, "List name", 80), "is_done": is_done}
if is_done and not row["is_done"]:
for card in tx.find("cards",list_id=list_id):
if self._active_blockers(tx,card,completing_list_id=list_id):
raise m.Invalid("This list contains a card with unfinished prerequisites in another list")
if not is_done and row["is_done"]:
for card in tx.find("cards",list_id=list_id):
self._require_can_reopen(tx,card,reopening_list_id=list_id)
# Completion semantics change overdue status for every card on the list.
if row["is_done"] != is_done:
for card in tx.find("cards", list_id=list_id):
self._event(tx, person, card, "list_completion_changed", {
"is_done": {"before": row["is_done"], "after": is_done}})
tx.update("board_lists", list_id, changes)
row.update(changes)
self._touch(tx, "board_lists", row)
self._touch(tx, "boards", board)
return Outcome(row, board_id=board["id"])
def _reorder_lists(self, tx, person, board_id, list_ids, version):
board = self._board(tx, person, board_id, "edit")
m.check_version(board, version)
lists = tx.find("board_lists", board_id=board_id)
if len(list_ids) != len(lists) or len(set(list_ids)) != len(list_ids) or set(
list_ids) != {row["id"] for row in lists}:
raise m.Invalid("Provide each list on this board exactly once")
by_id = {row["id"]: row for row in lists}
for position, lid in enumerate(list_ids):
row = by_id[lid]
if row["position"] != position:
tx.update("board_lists", lid, {"position": position})
self._touch(tx, "board_lists", row)
self._touch(tx, "boards", board)
return Outcome({"board_id": board_id, "list_ids": list_ids, "version": board["version"]},
board_id=board_id)
def _assignee(self, tx, board, value):
if value is None or value == "":
return None
value = m.email(value)
if value not in {member["email"] for member in self._members(tx, board)}:
raise m.Invalid("Assignee must be a member of this board")
return value
def _add_card(self, tx, person, board_id, list_id, title, description, assignee, due_date):
board = self._board(tx, person, board_id, "edit")
target = tx.get("board_lists", list_id)
if not target or target["board_id"] != board_id:
raise m.NotFound("Resource not found")
cards = tx.find("cards", board_id=board_id)
if len(cards) >= 5000:
raise m.Invalid("A board can contain at most 5,000 cards")
stamp = m.now()
row = self._insert(tx, "cards", m.Card, id=uuid4().hex, board_id=board_id,
list_id=list_id, title=m.text(title, "Card title"), description=m.text(
description, "Description", 20000, True), assignee=self._assignee(tx, board, assignee),
due_date=m.due_date(due_date), position=sum(c["list_id"] == list_id for c in cards),
created_by=person.sub, created_at=stamp, updated_at=stamp)
self._event(tx, person, row, "created", {"card": {"before": None, "after": row}})
self._touch(tx, "boards", board)
return Outcome(row, board_id=board_id)
def _get_card(self, tx, person, card_id):
board, card = self._card(tx, person, card_id)
card["labels"] = [{"name": row["name"], "color": row["color"]}
for row in tx.find("card_labels", card_id=card_id)]
card["checklist"] = sorted(tx.find("card_checklist_items", card_id=card_id),
key=lambda r: (r["position"], r["id"]))
card["activity"] = self._get_card_activity(tx, person, card_id).data
target = tx.get("board_lists", card["list_id"])
card.update(role=board["role"], board_title=board["title"],
overdue=m.overdue(card, target["is_done"]))
dependency_data = self._dependency_data(tx,board,card)
card["dependencies"] = dependency_data["dependencies"]
card["blocking_chain"] = dependency_data
return Outcome(card)
def _save_card_fields(self, tx, person, board, card, fields, action="updated"):
changes = {key: {"before": card[key], "after": value}
for key, value in fields.items() if card[key] != value}
if changes:
tx.update("cards", card["id"], fields)
self._touch(tx, "cards", card)
card.update(fields)
self._event(tx, person, card, action, changes)
self._touch(tx, "boards", board)
return card
def _update_card(self, tx, person, card_id, title, description, assignee, due_date, labels, version):
board, card = self._card(tx, person, card_id, "edit")
m.check_version(card, version)
clean_labels = [{"name": m.text(r["name"], "Label", 32), "color": m.color(r["color"])}
for r in labels]
if len({r["name"].casefold() for r in clean_labels}) != len(clean_labels):
raise m.Invalid("Label names must be unique on a card")
old_labels = tx.find("card_labels", card_id=card_id)
old_values = [{"name": r["name"], "color": r["color"]} for r in old_labels]
fields = {"title": m.text(title, "Card title"), "description": m.text(
description, "Description", 20000, True), "assignee": self._assignee(tx, board, assignee),
"due_date": m.due_date(due_date)}
previous_version = card["version"]
self._save_card_fields(tx, person, board, card, fields)
if sorted(old_values, key=lambda r: r["name"]) != sorted(clean_labels, key=lambda r: r["name"]):
for old in old_labels:
tx.delete("card_labels", old["id"])
for label in clean_labels:
self._insert(tx, "card_labels", m.CardLabel, id=uuid4().hex,
board_id=board["id"], card_id=card_id, created_at=m.now(), **label)
if card["version"] == previous_version:
self._touch(tx, "cards", card)
self._touch(tx, "boards", board)
self._event(tx, person, card, "labels_updated", {
"labels": {"before": old_values, "after": clean_labels}})
return Outcome({**card, "labels": clean_labels}, board_id=board["id"])
def _assign_card(self, tx, person, card_id, assignee, version):
board, card = self._card(tx, person, card_id, "edit")
m.check_version(card, version)
self._save_card_fields(tx, person, board, card, {"assignee": self._assignee(tx, board, assignee)},
"assigned")
return Outcome(card, board_id=board["id"])
def _set_due_date(self, tx, person, card_id, due_date, version):
board, card = self._card(tx, person, card_id, "edit")
m.check_version(card, version)
self._save_card_fields(tx, person, board, card, {"due_date": m.due_date(due_date)}, "due_date_set")
return Outcome(card, board_id=board["id"])
def _move_card(self, tx, person, card_id, list_id, index, version, board_version):
board, card = self._card(tx, person, card_id, "edit")
m.check_version(card, version)
m.check_version(board, board_version)
target = tx.get("board_lists", list_id)
if not target or target["board_id"] != board["id"]:
raise m.NotFound("Resource not found")
blockers = self._active_blockers(tx,card) if target["is_done"] else []
if blockers:
raise m.Invalid("Complete prerequisites before moving this card to a completed list: " +
"; ".join(blocker["title"] for blocker in blockers))
if not target["is_done"] and tx.get("board_lists",card["list_id"])["is_done"]:
self._require_can_reopen(tx,card)
affected = sorted(tx.find("cards", board_id=board["id"]),
key=lambda r: (r["position"], r["id"]))
destination = [r for r in affected if r["list_id"] == list_id and r["id"] != card_id]
if index > len(destination):
raise m.Invalid("Drop position is outside the destination list")
destination.insert(index, card)
origin = [r for r in affected if r["list_id"] == card["list_id"] and r["id"] != card_id]
sequences = [(list_id, destination)]
if list_id != card["list_id"]:
sequences.append((card["list_id"], origin))
changed = False
for lid, rows in sequences:
for position, row in enumerate(rows):
if row["list_id"] == lid and row["position"] == position:
continue
delta = {"list_id": {"before": row["list_id"], "after": lid},
"position": {"before": row["position"], "after": position}}
tx.update("cards", row["id"], {"list_id": lid, "position": position})
self._touch(tx, "cards", row)
row.update(list_id=lid, position=position)
self._event(tx, person, row, "moved" if row["id"] == card_id else "reordered", delta)
changed = True
if changed:
self._touch(tx, "boards", board)
return Outcome({**card, "board_version": board["version"]}, board_id=board["id"])
def _invite_member(self, tx, person, board_id, email, role):
board = self._board(tx, person, board_id, "owner")
email, role = m.email(email), m.member_role(role)
if email == board["owner_email"]:
raise m.Invalid("The owner already has access")
key = sha256(f"{board_id}:{email}".encode()).hexdigest()
if tx.get("board_memberships", key):
raise m.Conflict("Already a board member; use Change role")
stamp = m.now()
self._insert(tx, "board_memberships", m.BoardMembership, id=key, board_id=board_id,
email=email, user_sub=None, role=role, invited_by=person.sub,
created_at=stamp, updated_at=stamp)
self._touch(tx, "boards", board)
return Outcome({"board_id": board_id, "email": email, "role": role},
board_id=board_id, permission="owner")
def _change_member_role(self, tx, person, board_id, email, role):
board = self._board(tx, person, board_id, "owner")
email, role = m.email(email), m.member_role(role)
key = sha256(f"{board_id}:{email}".encode()).hexdigest()
if not tx.get("board_memberships", key):
raise m.NotFound("Membership not found")
tx.update("board_memberships", key, {"role": role, "updated_at": m.now()})
self._touch(tx, "boards", board)
return Outcome({"board_id": board_id, "email": email, "role": role},
board_id=board_id, permission="owner")
def _remove_member(self, tx, person, board_id, email):
board = self._board(tx, person, board_id, "owner")
email = m.email(email)
key = sha256(f"{board_id}:{email}".encode()).hexdigest()
if not tx.get("board_memberships", key):
raise m.NotFound("Membership not found")
tx.delete("board_memberships", key)
for card in tx.find("cards", board_id=board_id):
if card["assignee"] == email:
self._save_card_fields(tx, person, board, card, {"assignee": None}, "member_removed")
self._touch(tx, "boards", board)
return Outcome({"board_id": board_id, "email": email, "removed": True},
board_id=board_id, permission="owner")
def _get_card_activity(self, tx, person, card_id):
self._card(tx, person, card_id)
rows = sorted(tx.find("card_activity", card_id=card_id),
key=lambda r: (r["created_at"], r["id"]))
for row in rows:
row["changes"] = json.loads(row.pop("changes_json"))
return Outcome(rows)
def _tasks(self, tx, person, only_overdue=False):
results = []
for board in self._visible_boards(tx, person):
lists = {r["id"]: r for r in tx.find("board_lists", board_id=board["id"])}
for card in tx.find("cards", board_id=board["id"]):
is_overdue = m.overdue(card, lists[card["list_id"]]["is_done"])
if (only_overdue and is_overdue) or (not only_overdue and card["assignee"] == person.email):
results.append({**card, "board_title": board["title"], "role": board["role"],
"list_title": lists[card["list_id"]]["title"], "overdue": is_overdue})
return Outcome(sorted(results, key=lambda r: (r["due_date"] or "9999-12-31", r["id"])))
def _list_my_cards(self, tx, person):
return self._tasks(tx, person)
def _list_overdue(self, tx, person):
return self._tasks(tx, person, only_overdue=True)
def _add_checklist_item(self, tx, person, card_id, title, version):
board, card = self._card(tx, person, card_id, "edit")
m.check_version(card, version)
items = tx.find("card_checklist_items", card_id=card_id)
if len(items) >= 100:
raise m.Invalid("A card can contain at most 100 checklist items")
stamp = m.now()
row = self._insert(tx, "card_checklist_items", m.CardChecklistItem, id=uuid4().hex,
board_id=board["id"], card_id=card_id, title=m.text(title, "Checklist item", 240),
completed=False, position=len(items), created_at=stamp, updated_at=stamp)
self._event(tx, person, card, "checklist_item_added", {"item": {"before": None, "after": row}})
self._touch(tx, "cards", card)
self._touch(tx, "boards", board)
return Outcome(row, board_id=board["id"])
def _complete_checklist_item(self, tx, person, item_id, completed, version):
row = tx.get("card_checklist_items", item_id, guard=True)
if not row:
raise m.NotFound("Resource not found")
board, card = self._card(tx, person, row["card_id"], "edit")
m.check_version(row, version)
if row["completed"] != completed:
self._event(tx, person, card, "checklist_item_changed", {
"item_id": item_id, "title": row["title"],
"completed": {"before": row["completed"], "after": completed}})
tx.update("card_checklist_items", item_id, {"completed": completed})
row["completed"] = completed
self._touch(tx, "card_checklist_items", row)
self._touch(tx, "cards", card)
self._touch(tx, "boards", board)
return Outcome(row, board_id=board["id"])
@staticmethod
def _active_blockers(tx,card,completing_list_id=None):
results = []
for edge in tx.find("card_dependencies",card_id=card["id"]):
dependency = tx.get("cards",edge["depends_on_id"])
if dependency:
target = tx.get("board_lists",dependency["list_id"])
if not target["is_done"] and dependency["list_id"] != completing_list_id:
results.append(dependency)
return results
@staticmethod
def _require_can_reopen(tx,card,reopening_list_id=None):
for edge in tx.find("card_dependencies",depends_on_id=card["id"]):
dependent = tx.get("cards",edge["card_id"])
if dependent and dependent["list_id"] != reopening_list_id:
target = tx.get("board_lists",dependent["list_id"])
if target["is_done"]:
raise m.Invalid("Reopen the completed dependent card first: " + dependent["title"])
@staticmethod
def _dependency_data(tx,board,card):
cards = {row["id"]:row for row in tx.find("cards",board_id=board["id"])}
edges = tx.find("card_dependencies",board_id=board["id"])
done = {row["id"] for row in tx.find("board_lists",board_id=board["id"]) if row["is_done"]}
paths,truncated = blocking_paths(cards,edges,card["id"],done)
dependencies = [{"id":edge["id"],"card_id":edge["depends_on_id"],
"title":cards[edge["depends_on_id"]]["title"],
"completed":cards[edge["depends_on_id"]]["list_id"] in done}
for edge in edges if edge["card_id"] == card["id"] and edge["depends_on_id"] in cards]
chains = [[{"id":cid,"title":cards[cid]["title"]} for cid in path] for path in paths]
explanation = "No unfinished prerequisites."
if chains:
explanation = " ".join(" → waits for → ".join(row["title"] for row in chain) + "."
for chain in chains[:10])
return {"card_id":card["id"],"blocked":bool(paths),"dependencies":dependencies,
"chains":chains,"explanation":explanation,"truncated":truncated}
def _get_blocking_chain(self,tx,person,card_id):
board,card = self._card(tx,person,card_id)
return Outcome(self._dependency_data(tx,board,card))
def _add_dependency(self,tx,person,card_id,depends_on_id,version):
board,card = self._card(tx,person,card_id,"edit")
other_board,prerequisite = self._card(tx,person,depends_on_id)
if other_board["id"] != board["id"]:
raise m.NotFound("Resource not found")
m.check_version(card,version)
key = sha256(f"{card_id}:{depends_on_id}".encode()).hexdigest()
if tx.get("card_dependencies",key):
raise m.Conflict("This prerequisite is already linked")
require_acyclic(tx.find("card_dependencies",board_id=board["id"]),card_id,depends_on_id)
source_list = tx.get("board_lists",card["list_id"])
target_list = tx.get("board_lists",prerequisite["list_id"])
if source_list["is_done"] and not target_list["is_done"]:
raise m.Invalid("Reopen this completed card before linking an unfinished prerequisite")
row = self._insert(tx,"card_dependencies",m.CardDependency,id=key,board_id=board["id"],
card_id=card_id,depends_on_id=depends_on_id,created_by=person.sub,created_at=m.now())
self._event(tx,person,card,"dependency_added",{"prerequisite":{
"before":None,"after":{"id":depends_on_id,"title":prerequisite["title"]}}})
self._touch(tx,"cards",card)
self._touch(tx,"boards",board)
return Outcome(row,board_id=board["id"])
def _remove_dependency(self,tx,person,card_id,depends_on_id,version):
board,card = self._card(tx,person,card_id,"edit")
m.check_version(card,version)
key = sha256(f"{card_id}:{depends_on_id}".encode()).hexdigest()
if not tx.get("card_dependencies",key):
raise m.NotFound("Resource not found")
tx.delete("card_dependencies",key)
self._event(tx,person,card,"dependency_removed",{
"prerequisite": {"before":depends_on_id,"after":None}})
self._touch(tx,"cards",card)
self._touch(tx,"boards",board)
return Outcome({"card_id":card_id,"removed_prerequisite":depends_on_id},board_id=board["id"])
def _apply_checklist(self,tx,person,card_id,items,version):
board,card = self._card(tx,person,card_id,"edit")
m.check_version(card,version)
existing = tx.find("card_checklist_items",card_id=card_id)
clean = [m.text(item,"Checklist item",240) for item in items]
if len(existing) + len(clean) > 100:
raise m.Invalid("A card can contain at most 100 checklist items")
created = []
for index,title in enumerate(clean):
stamp = m.now()
created.append(self._insert(tx,"card_checklist_items",m.CardChecklistItem,id=uuid4().hex,
board_id=board["id"],card_id=card_id,title=title,completed=False,
position=len(existing)+index,created_at=stamp,updated_at=stamp))
self._event(tx,person,card,"checklist_items_added",{"items":{"before":None,"after":created}})
self._touch(tx,"cards",card)
self._touch(tx,"boards",board)
return Outcome({"card_id":card_id,"items":created,"version":card["version"]},board_id=board["id"])