damien-kaiser/taskloft
Taskloft — team task boards (owner / member / viewer), card history, my plate, overdue, blockers, AI checklists and summaries, usable from Claude and ChatGPT over MCP.
A remix template · 0 remixes · Remix damien-kaiser/taskloft
app/boards/infrastructure/repository.py
"""Boards, lists, cards, people, checklists and history in DataK3 tables (schema.sql).
- Writes are keyed upserts (INSERT ... ON CONFLICT (id) DO UPDATE), so a retry is safe.
- A card NUMBER is claimed with a plain INSERT into card_numbers: under concurrency one writer
wins, the other gets 23505 and takes the next number (never an upsert, which lets both win).
- Transient DataK3 faults (40001, 08006, 58030, XX000) are retried; real errors are not.
- At most DB_CONCURRENCY connections at once (a bucket serves 16 concurrent reads).
"""
import json
import os
import threading
import time
import uuid
from typing import Optional
import psycopg
from boards.domain.model import Board, BoardList, Card, Role
from boards.infrastructure.dodil import connect
RETRY = {"40001", "08006", "58030", "XX000"}
_gate = threading.BoundedSemaphore(int(os.environ.get("DB_CONCURRENCY", "8")))
# Additive migrations, run at start-up: a bucket made by an older version of this repo gets the
# new tables and columns, a fresh remix (schema.sql) already has them. Never a rename or a drop.
MIGRATIONS = [
"CREATE TABLE IF NOT EXISTS lists (id TEXT PRIMARY KEY, board_id TEXT NOT NULL, name TEXT NOT NULL, position BIGINT NOT NULL, is_done BOOLEAN)",
"CREATE TABLE IF NOT EXISTS card_numbers (number BIGINT PRIMARY KEY, card_id TEXT NOT NULL)",
"CREATE TABLE IF NOT EXISTS card_blockers (id TEXT PRIMARY KEY, card_id TEXT NOT NULL, blocked_by TEXT NOT NULL)",
"CREATE TABLE IF NOT EXISTS checklist_items (id TEXT PRIMARY KEY, card_id TEXT NOT NULL, text TEXT NOT NULL, done BOOLEAN, position BIGINT)",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS number BIGINT",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS list_id TEXT",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS description TEXT",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS labels TEXT",
"ALTER TABLE cards ADD COLUMN IF NOT EXISTS position DOUBLE",
"ALTER TABLE activity ADD COLUMN IF NOT EXISTS changes TEXT",
"CREATE TABLE IF NOT EXISTS board_summaries (id TEXT PRIMARY KEY, board_id TEXT NOT NULL, text TEXT NOT NULL, made_at TIMESTAMP)",
"CREATE TABLE IF NOT EXISTS card_vectors (card_id TEXT PRIMARY KEY, board_id TEXT NOT NULL, text_hash TEXT NOT NULL, embedding VECTOR(2048))",
]
CARD_COLS = "id, board_id, number, title, list_id, list_name, description, assignee, due_date, labels, position"
def _card(r, blockers=()) -> Card:
return Card(id=r[0], board_id=r[1], number=int(r[2] or 0), title=r[3], list_id=r[4] or "",
list_name=r[5], description=r[6] or "", assignee=r[7], due_date=r[8],
labels=json.loads(r[9]) if r[9] else [], position=float(r[10] or 0.0),
blocked_by=list(blockers))
class SqlBoards:
def _run(self, fn):
"""One connection, retried on DataK3's transient faults (never on a real error)."""
for attempt in range(4):
try:
with _gate, connect() as c:
return fn(c)
except psycopg.Error as e:
if getattr(e, "sqlstate", None) in RETRY and attempt < 3:
time.sleep(0.2 * (attempt + 1))
continue
raise
def migrate(self) -> list:
done = []
def go(c):
for sql in MIGRATIONS:
c.execute(sql)
done.append(sql.split("(")[0][:60])
self._run(go)
return done
def table_counts(self) -> dict:
tables = ("boards", "memberships", "lists", "cards", "card_numbers", "card_blockers",
"checklist_items", "activity", "card_vectors", "board_summaries")
return self._run(lambda c: {t: c.execute(f"SELECT count(*) AS n FROM {t}").fetchone()[0]
for t in tables})
# ------------------------------------------------------------------ boards
def _load_board(self, c, board_id: str) -> Optional[Board]:
row = c.execute("SELECT id, title, owner_email FROM boards WHERE id = %s", (board_id,)).fetchone()
if not row:
return None
members = c.execute("SELECT email, role FROM memberships WHERE board_id = %s", (board_id,)).fetchall()
lists = c.execute("SELECT id, board_id, name, position, is_done FROM lists WHERE board_id = %s "
"ORDER BY position", (board_id,)).fetchall()
return Board(id=row[0], title=row[1], owner_email=row[2],
members={e: Role(r) for e, r in members if r != "owner"},
lists=[BoardList(i, b, n, int(p), bool(d)) for i, b, n, p, d in lists])
def get(self, board_id: str) -> Optional[Board]:
board = self._run(lambda c: self._load_board(c, board_id))
if board is not None and not board.lists: # a board made before lists existed
board.lists = [BoardList(uuid.uuid4().hex[:12], board.id, n, i, n == "Done")
for i, n in enumerate(("To do", "Doing", "Done"))]
self.save_board(board)
return board
def boards_for(self, email: str) -> list:
def go(c):
ids = [r[0] for r in c.execute(
"SELECT id FROM boards WHERE owner_email = %s "
"UNION SELECT board_id FROM memberships WHERE email = %s", (email, email))]
return [b for b in (self._load_board(c, i) for i in ids) if b]
return self._run(go)
def save_board(self, board: Board) -> None:
def go(c):
c.execute("INSERT INTO boards (id, title, owner_email, created_at) VALUES (%s, %s, %s, now()) "
"ON CONFLICT (id) DO UPDATE SET title = EXCLUDED.title", (board.id, board.title, board.owner_email))
for email, role in board.members.items():
c.execute("INSERT INTO memberships (id, board_id, email, role) VALUES (%s, %s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET role = EXCLUDED.role",
(f"{board.id}:{email}", board.id, email, role.value))
for lst in board.lists:
c.execute("INSERT INTO lists (id, board_id, name, position, is_done) VALUES (%s, %s, %s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, position = EXCLUDED.position, "
"is_done = EXCLUDED.is_done", (lst.id, board.id, lst.name, lst.position, lst.is_done))
self._run(go)
def remove_member(self, board_id: str, email: str) -> None:
self._run(lambda c: c.execute("DELETE FROM memberships WHERE id = %s", (f"{board_id}:{email}",)))
# ------------------------------------------------------------------ cards
def claim_number(self, card_id: str) -> int:
for _ in range(25):
def go(c):
n = c.execute("SELECT coalesce(max(number), 0) + 1 AS n FROM card_numbers").fetchone()[0]
c.execute("INSERT INTO card_numbers (number, card_id) VALUES (%s, %s)", (int(n), card_id))
return int(n)
try:
return self._run(go)
except psycopg.errors.UniqueViolation:
continue # someone took that number a moment ago: take the next one
raise RuntimeError("could not claim a card number")
def _blockers(self, c, card_ids) -> dict:
out = {cid: [] for cid in card_ids}
if not out:
return out
marks = ", ".join(["%s"] * len(out))
for cid, by in c.execute(f"SELECT card_id, blocked_by FROM card_blockers WHERE card_id IN ({marks})",
tuple(out)).fetchall():
out.setdefault(cid, []).append(by)
return out
def get_card(self, card_id: str) -> Optional[Card]:
def go(c):
r = c.execute(f"SELECT {CARD_COLS} FROM cards WHERE id = %s", (card_id,)).fetchone()
return _card(r, self._blockers(c, [r[0]])[r[0]]) if r else None
return self._run(go)
def card_by_number(self, number: int) -> Optional[Card]:
def go(c):
r = c.execute(f"SELECT {CARD_COLS} FROM cards WHERE number = %s", (int(number),)).fetchone()
return _card(r, self._blockers(c, [r[0]])[r[0]]) if r else None
return self._run(go)
def cards_on(self, board_id: str) -> list:
def go(c):
rows = c.execute(f"SELECT {CARD_COLS} FROM cards WHERE board_id = %s", (board_id,)).fetchall()
bl = self._blockers(c, [r[0] for r in rows])
return [_card(r, bl.get(r[0], [])) for r in rows]
return self._run(go)
def save_card(self, card: Card) -> None:
self._run(lambda c: c.execute(
"INSERT INTO cards (id, board_id, number, title, list_id, list_name, description, assignee, "
"due_date, labels, position, updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now()) "
"ON CONFLICT (id) DO UPDATE SET title = EXCLUDED.title, list_id = EXCLUDED.list_id, "
"list_name = EXCLUDED.list_name, description = EXCLUDED.description, assignee = EXCLUDED.assignee, "
"due_date = EXCLUDED.due_date, labels = EXCLUDED.labels, position = EXCLUDED.position, "
"updated_at = EXCLUDED.updated_at",
(card.id, card.board_id, card.number, card.title, card.list_id, card.list_name, card.description,
card.assignee, card.due_date, json.dumps(card.labels), card.position)))
# ------------------------------------------------------------------ blockers, checklist
def set_blocker(self, card_id: str, blocked_by: str, on: bool) -> None:
key = f"{card_id}>{blocked_by}"
if on:
self._run(lambda c: c.execute(
"INSERT INTO card_blockers (id, card_id, blocked_by) VALUES (%s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET blocked_by = EXCLUDED.blocked_by", (key, card_id, blocked_by)))
else:
self._run(lambda c: c.execute("DELETE FROM card_blockers WHERE id = %s", (key,)))
def checklist(self, card_id: str) -> list:
rows = self._run(lambda c: c.execute(
"SELECT id, card_id, text, done, position FROM checklist_items WHERE card_id = %s ORDER BY position",
(card_id,)).fetchall())
return [{"id": i, "card_id": cid, "text": t, "done": bool(d), "position": int(p or 0)}
for i, cid, t, d, p in rows]
def save_checklist_item(self, item: dict) -> None:
self._run(lambda c: c.execute(
"INSERT INTO checklist_items (id, card_id, text, done, position) VALUES (%s, %s, %s, %s, %s) "
"ON CONFLICT (id) DO UPDATE SET text = EXCLUDED.text, done = EXCLUDED.done, position = EXCLUDED.position",
(item["id"], item["card_id"], item["text"], bool(item["done"]), int(item["position"]))))
# ------------------------------------------------------------------ history
def log(self, board_id: str, card_id: Optional[str], actor: str, action: str,
changes: Optional[dict] = None) -> None:
self._run(lambda c: c.execute(
"INSERT INTO activity (id, board_id, card_id, actor, action, event_at, changes) "
"VALUES (%s, %s, %s, %s, %s, now(), %s)",
(uuid.uuid4().hex, board_id, card_id, actor, action, json.dumps(changes) if changes else None)))
def history(self, board_id: str, card_id: Optional[str] = None, limit: int = 50) -> list:
def go(c):
if card_id:
return c.execute('SELECT actor, action, event_at, changes, card_id FROM activity '
'WHERE card_id = %s ORDER BY event_at DESC LIMIT %s', (card_id, limit)).fetchall()
return c.execute('SELECT actor, action, event_at, changes, card_id FROM activity '
'WHERE board_id = %s ORDER BY event_at DESC LIMIT %s', (board_id, limit)).fetchall()
return [{"actor": a, "action": act, "at": at.isoformat() if hasattr(at, "isoformat") else str(at),
"changes": json.loads(ch) if ch else None, "card_id": cid}
for a, act, at, ch, cid in self._run(go)]
# ------------------------------------------------------------------ meaning
@staticmethod
def _in(ids) -> str:
return ", ".join(["%s"] * len(ids))
def vector_hashes(self, board_ids: list) -> dict:
if not board_ids:
return {}
rows = self._run(lambda c: c.execute(
f"SELECT card_id, text_hash FROM card_vectors WHERE board_id IN ({self._in(board_ids)})",
tuple(board_ids)).fetchall())
return dict(rows)
def save_vector(self, card_id: str, board_id: str, text_hash: str, vector: list) -> None:
lit = "[" + ",".join(f"{float(x):.7f}" for x in vector) + "]" # pgvector text literal
self._run(lambda c: c.execute(
"INSERT INTO card_vectors (card_id, board_id, text_hash, embedding) VALUES (%s, %s, %s, %s) "
"ON CONFLICT (card_id) DO UPDATE SET board_id = EXCLUDED.board_id, text_hash = EXCLUDED.text_hash, "
"embedding = EXCLUDED.embedding", (card_id, board_id, text_hash, lit)))
def nearest(self, board_ids: list, vector: list, k: int, exclude: Optional[str] = None) -> list:
"""Cosine KNN, filtered to the boards the person can see (DataK3 re-checks the filter exactly)."""
if not board_ids:
return []
lit = "[" + ",".join(f"{float(x):.7f}" for x in vector) + "]"
sql = (f"SELECT card_id, embedding <=> %s AS d FROM card_vectors WHERE board_id IN ({self._in(board_ids)})"
+ (" AND card_id <> %s" if exclude else "") + " ORDER BY d LIMIT %s")
args = (lit, *board_ids, *((exclude,) if exclude else ()), int(k))
return [(cid, float(d)) for cid, d in self._run(lambda c: c.execute(sql, args).fetchall())]
# ------------------------------------------------------------------ summaries
def cached_summary(self, key: str) -> Optional[str]:
r = self._run(lambda c: c.execute("SELECT text FROM board_summaries WHERE id = %s", (key,)).fetchone())
return r[0] if r else None
def save_summary(self, key: str, board_id: str, text: str) -> None:
self._run(lambda c: c.execute(
"INSERT INTO board_summaries (id, board_id, text, made_at) VALUES (%s, %s, %s, now()) "
"ON CONFLICT (id) DO UPDATE SET text = EXCLUDED.text, made_at = EXCLUDED.made_at", (key, board_id, text)))