rgboards/rg-boards
Lanes, team task boards you run from the web or from Claude/ChatGPT (MCP). FastAPI + DataK3 + appid + Dodil Models.
A remix template · 0 remixes · Remix rgboards/rg-boards
app/boards/infrastructure/repository.py
"""Boards, cards, history, links, checklists and the embedding cache in DataK3 tables
(schema.sql), over the Postgres wire.
- Card numbers are claimed with a plain INSERT (one writer wins, the rest get 23505).
- Card edits are compare-and-set on `version`, so two people can't silently overwrite each other.
- Every query that reads boards is filtered by the caller's email in the use cases.
"""
import json
import threading
import time
import uuid
from datetime import datetime, timezone
from typing import Optional
import psycopg
from boards.application.ports import DuplicateKey
from boards.domain.model import DEFAULT_LISTS, Board, Card, Role
from boards.infrastructure.dodil import connect
CARD_COLS = ("id, board_id, number, title, list_name, description, assignee, due_date, labels, "
"position, version, archived, created_by")
RETRY = ("40001", "08006", "58030", "XX000")
_sem = threading.BoundedSemaphore(8) # DataK3 allows 16 concurrent reads per bucket
MAX_AGE = 240 # seconds; the connection's password is a short-lived token
class _Pool:
"""A small shared pool of DataK3 connections. Opening one costs a TLS handshake plus a
token, so the first page load (several parallel requests) reuses warm ones instead of
every worker thread opening its own."""
def __init__(self):
self._idle, self._lock = [], threading.Lock()
def get(self) -> tuple:
with self._lock:
while self._idle:
conn, born = self._idle.pop()
if not conn.closed and born > time.time() - MAX_AGE:
return conn, born
conn.close()
return connect(), time.time()
def put(self, conn, born) -> None:
if conn.closed or born < time.time() - MAX_AGE:
conn.close()
return
with self._lock:
self._idle.append((conn, born))
def warm(self, n: int = 2) -> None:
"""Open n connections ahead of the first request (called off the request path)."""
got = [self.get() for _ in range(n)]
for conn, born in got:
self.put(conn, born)
POOL = _Pool()
def _card(r) -> Card:
return Card(id=r[0], board_id=r[1], number=int(r[2] or 0), title=r[3], list_name=r[4],
description=r[5] or "", assignee=r[6], due_date=r[7],
labels=[x for x in (r[8] or "").split(",") if x], position=float(r[9] or 0),
version=int(r[10] or 1), archived=bool(r[11]), created_by=r[12] or "")
def _utc(ts) -> str:
"""Timestamps go out as ISO 8601 with an explicit UTC offset, so every page and every
assistant can show them in the reader's own time zone."""
if not ts:
return ""
if ts.tzinfo is None:
ts = ts.replace(tzinfo=timezone.utc)
return ts.astimezone(timezone.utc).isoformat(timespec="seconds")
def _in(values: list) -> tuple:
"""DataK3 (DuckDB underneath) has no ANY(array) binding: expand IN (%s, %s, ...)."""
return ", ".join(["%s"] * len(values)), tuple(values)
class SqlBoards:
def warm(self, n: int = 2) -> None:
POOL.warm(n)
def _run(self, sql: str, params: tuple = (), fetch: bool = True):
last = None
for attempt in range(3):
with _sem:
conn = born = None
try:
conn, born = POOL.get()
cur = conn.execute(sql, params)
out = (cur.fetchall() if cur.description else []) if fetch else cur.rowcount
POOL.put(conn, born)
return out
except psycopg.errors.UniqueViolation as e:
POOL.put(conn, born)
raise DuplicateKey(str(e))
except (psycopg.OperationalError, psycopg.InterfaceError) as e:
last = e
except psycopg.Error as e:
if getattr(e, "sqlstate", None) not in RETRY:
POOL.put(conn, born)
raise
last = e
if conn is not None:
conn.close()
time.sleep(0.2 * (attempt + 1))
raise last
# ------------------------------------------------------------------ boards
def _boards(self, ids: list) -> list:
if not ids:
return []
ph, p = _in(ids)
rows = self._run(f"SELECT id, title, owner_email, lists FROM boards WHERE id IN ({ph})", p)
members = self._run(f"SELECT board_id, email, role FROM memberships "
f"WHERE board_id IN ({ph})", p)
out = []
for bid, title, owner, lists in rows:
m = {e: Role(r) for b, e, r in members if b == bid and r != "owner"}
out.append(Board(id=bid, title=title, owner_email=owner, members=m,
lists=json.loads(lists) if lists else list(DEFAULT_LISTS)))
return out
def get(self, board_id: str) -> Optional[Board]:
found = self._boards([board_id])
return found[0] if found else None
def boards_for(self, email: str) -> list:
ids = [r[0] for r in self._run(
"SELECT id FROM boards WHERE owner_email = %s "
"UNION SELECT board_id FROM memberships WHERE email = %s", (email, email))]
return self._boards(ids)
def save_board(self, board: Board) -> None:
self._run("INSERT INTO boards (id, title, owner_email, created_at, lists) "
"VALUES (%s, %s, %s, now(), %s) ON CONFLICT (id) DO UPDATE SET "
"title = EXCLUDED.title, owner_email = EXCLUDED.owner_email, "
"lists = EXCLUDED.lists",
(board.id, board.title, board.owner_email, json.dumps(board.lists)), False)
have = {r[0] for r in self._run("SELECT email FROM memberships WHERE board_id = %s",
(board.id,))}
for email, role in board.members.items():
self._run("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), False)
for gone in have - set(board.members):
self._run("DELETE FROM memberships WHERE id = %s", (f"{board.id}:{gone}",), False)
# ------------------------------------------------------------------ cards
def get_card(self, card_id: str) -> Optional[Card]:
rows = self._run(f"SELECT {CARD_COLS} FROM cards WHERE id = %s", (card_id,))
return _card(rows[0]) if rows else None
def cards_on(self, board_ids: list) -> list:
if not board_ids:
return []
ph, p = _in(board_ids)
rows = self._run(f"SELECT {CARD_COLS} FROM cards WHERE board_id IN ({ph}) "
f"AND (archived IS NULL OR archived = FALSE)", p)
return [_card(r) for r in rows]
def next_number(self, board_id: str) -> int:
rows = self._run("SELECT max(number) FROM cards WHERE board_id = %s", (board_id,))
return int(rows[0][0] or 0) + 1 if rows else 1
def insert_card(self, card: Card) -> None:
self._run("INSERT INTO cards (id, board_id, number, title, list_name, description, "
"assignee, due_date, labels, position, version, archived, created_by, "
"created_at, updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, "
"%s, %s, now(), now())",
(card.id, card.board_id, card.number, card.title, card.list_name,
card.description, card.assignee, card.due_date, ",".join(card.labels),
card.position, card.version, card.archived, card.created_by), False)
def update_card(self, card: Card, expected_version: int) -> bool:
n = self._run("UPDATE cards SET title = %s, list_name = %s, description = %s, "
"assignee = %s, due_date = %s, labels = %s, position = %s, version = %s, "
"archived = %s, updated_at = now() WHERE id = %s AND "
"coalesce(version, 1) = %s",
(card.title, card.list_name, card.description, card.assignee,
card.due_date, ",".join(card.labels), card.position, card.version,
card.archived, card.id, expected_version), False)
if n is None or n < 0: # adapter didn't report a count: read back to be sure
cur = self.get_card(card.id)
return bool(cur and cur.version == card.version)
return n > 0
# ------------------------------------------------------------------ history
def log(self, board_id, card_id, actor, action, field=None, old=None, new=None,
at=None) -> None:
# The use case passes the time (UTC), so a seed or an import can write a real timeline.
# (DataK3 takes a literal or now() in VALUES, not an expression around a parameter.)
at = (at or datetime.now(timezone.utc)).astimezone(timezone.utc).replace(tzinfo=None)
self._run("INSERT INTO activity (id, board_id, card_id, actor, action, event_at, field, "
"old_value, new_value) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
(uuid.uuid4().hex, board_id, card_id, actor, action, at, field, old, new),
False)
def _events(self, where: str, params: tuple, limit: int) -> list:
rows = self._run(f"SELECT id, board_id, card_id, actor, action, field, old_value, "
f"new_value, event_at FROM activity WHERE {where} "
f"ORDER BY event_at DESC LIMIT {int(limit)}", params)
return [{"id": r[0], "board_id": r[1], "card_id": r[2], "actor": r[3], "action": r[4],
"field": r[5], "old": r[6], "new": r[7],
"at": _utc(r[8])} for r in rows]
def history(self, card_id: str) -> list:
return self._events("card_id = %s", (card_id,), 200)
def activity(self, board_id: str, limit: int = 50) -> list:
return self._events("board_id = %s", (board_id,), limit)
# ------------------------------------------------------------------ links
def links_on(self, board_id: str) -> list:
return [(r[0], r[1]) for r in self._run(
"SELECT card_id, blocked_by FROM card_links WHERE board_id = %s", (board_id,))]
def add_link(self, board_id, card_id, blocked_by, actor) -> None:
try:
self._run("INSERT INTO card_links (id, board_id, card_id, blocked_by, created_by, "
"created_at) VALUES (%s, %s, %s, %s, %s, now())",
(f"{card_id}|{blocked_by}", board_id, card_id, blocked_by, actor), False)
except DuplicateKey:
pass # already linked: adding it twice is the same as once
def remove_link(self, card_id, blocked_by) -> None:
self._run("DELETE FROM card_links WHERE id = %s", (f"{card_id}|{blocked_by}",), False)
# ------------------------------------------------------------------ checklists
def checklist(self, card_id: str) -> list:
rows = self._run("SELECT id, card_id, text, done, position FROM checklist_items "
"WHERE card_id = %s ORDER BY position", (card_id,))
return [{"id": r[0], "card_id": r[1], "text": r[2], "done": bool(r[3]),
"position": int(r[4] or 0)} for r in rows]
def save_check_item(self, board_id: str, item: dict) -> None:
self._run("INSERT INTO checklist_items (id, board_id, card_id, text, done, position) "
"VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (id) DO UPDATE SET "
"text = EXCLUDED.text, done = EXCLUDED.done, position = EXCLUDED.position",
(item["id"], board_id, item["card_id"], item["text"], bool(item["done"]),
int(item["position"])), False)
# ------------------------------------------------------------------ embeddings cache
def vectors_on(self, board_ids: list) -> dict:
if not board_ids:
return {}
ph, p = _in(board_ids)
rows = self._run(f"SELECT card_id, text_hash, vector FROM card_vectors "
f"WHERE board_id IN ({ph})", p)
return {r[0]: (r[1], json.loads(r[2])) for r in rows}
def save_vector(self, board_id, card_id, text_hash, vector) -> None:
self._run("INSERT INTO card_vectors (card_id, board_id, text_hash, vector) "
"VALUES (%s, %s, %s, %s) ON CONFLICT (card_id) DO UPDATE SET "
"board_id = EXCLUDED.board_id, text_hash = EXCLUDED.text_hash, "
"vector = EXCLUDED.vector",
(card_id, board_id, text_hash, json.dumps([round(x, 6) for x in vector])),
False)
# ------------------------------------------------------------------ daily summaries
def get_summary(self, board_id: str, day: str) -> Optional[dict]:
rows = self._run("SELECT text, written_at FROM board_summaries WHERE id = %s",
(f"{board_id}:{day}",))
if not rows:
return None
return {"summary": rows[0][0], "written_at": _utc(rows[0][1])}
def save_summary(self, board_id: str, day: str, text: str) -> None:
self._run("INSERT INTO board_summaries (id, board_id, day, text, written_at) "
"VALUES (%s, %s, %s, %s, now()) ON CONFLICT (id) DO UPDATE SET "
"text = EXCLUDED.text, written_at = EXCLUDED.written_at",
(f"{board_id}:{day}", board_id, day, text), False)
def self_check(self) -> dict:
"""Row counts per table, read with the app's OWN credential (POST / for ignite invoke)."""
out = {}
for t in ("boards", "memberships", "cards", "activity", "card_links",
"checklist_items", "card_vectors", "board_summaries"):
out[t] = int(self._run(f"SELECT count(*) FROM {t}")[0][0])
return out