north-scape/relayboard

Relayboard: accessible team boards with audited actions, dependency maps, and shared web/MCP use cases.

• dodil-challenge
• dodil-challenge-boards

A remix template · 0 remixes · Remix north-scape/relayboard

app/boards/infrastructure/repository.py

"""DataK3 PostgreSQL-wire adapter. Board mutations and audit rows share one transaction.

FOR UPDATE protects the board aggregate (membership, cards, dependency graph). Connections
are request-local ContextVars, never shared between workers. No out-of-band database exists.
"""
import json
import threading
import uuid
from contextlib import contextmanager
from contextvars import ContextVar

import psycopg
from boards.domain.model import Board, Card, Conflict, Role
from boards.infrastructure.dodil import connect

_connection=ContextVar('relayboard_connection', default=None)
_slots=threading.BoundedSemaphore(8)
_CARD_COLUMNS='id, board_id, title, list_name, assignee, due_date, description, labels, blocked_by, revision, checklist'


def _card(row):
    if row is None: return None
    row=list(row)
    row[7]=json.loads(row[7] or '[]')
    row[8]=json.loads(row[8] or '[]')
    row[10]=json.loads(row[10] or '[]')
    return Card(*row)

class SqlBoards:
    @contextmanager
    def _db(self):
        existing=_connection.get()
        if existing is not None:
            yield existing
        else:
            with _slots, connect() as c: yield c

    @contextmanager
    def atomic(self):
        if _connection.get() is not None:
            yield self
            return
        with _slots, connect() as c:
            token=_connection.set(c)
            try:
                with c.transaction(): yield self
            except psycopg.errors.SerializationFailure:
                raise Conflict('another update happened; reload and retry') from None
            finally: _connection.reset(token)

    def get(self,board_id):
        with self._db() as c:
            suffix=' FOR UPDATE' if _connection.get() is not None else ''
            row=c.execute('SELECT id, title, owner_email FROM boards WHERE id = %s'+suffix,(board_id,)).fetchone()
            if not row: return None
            if _connection.get() is not None:
                # Touch the common aggregate guard, so disjoint dependency edits
                # still conflict on optimistic DataK3 deployments.
                c.execute('UPDATE boards SET revision = revision + 1 WHERE id = %s',(board_id,))
            members=c.execute('SELECT email, role FROM memberships WHERE board_id = %s',(board_id,)).fetchall()
        return Board(*row,members={e:Role(r) for e,r in members})

    def boards_for(self,email):
        with self._db() as 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.get(i) for i in ids) if b]

    def save_board(self,board):
        with self._db() as c:
            exists=c.execute('SELECT id FROM boards WHERE id = %s',(board.id,)).fetchone()
            if exists:
                c.execute('UPDATE boards SET title = %s WHERE id = %s',(board.title,board.id))
            else:
                c.execute('INSERT INTO boards (id,title,owner_email,created_at) VALUES (%s,%s,%s,now())',(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))

    def get_card(self,id):
        with self._db() as c:
            return _card(c.execute('SELECT '+_CARD_COLUMNS+' FROM cards WHERE id = %s',(id,)).fetchone())

    def delete_membership(self,board_id,email):
        with self._db() as c:
            c.execute('DELETE FROM memberships WHERE board_id = %s AND email = %s',(board_id,email))

    def cards_on(self,id):
        with self._db() as c:
            return [_card(r) for r in c.execute('SELECT '+_CARD_COLUMNS+' FROM cards WHERE board_id = %s ORDER BY updated_at, id',(id,))]

    def save_card(self,card):
        values=(card.board_id,card.title,card.list_name,card.assignee,card.due_date,card.description,
                json.dumps(card.labels),json.dumps(card.blocked_by),card.revision,json.dumps(card.checklist))
        with self._db() as c:
            exists=c.execute('SELECT id FROM cards WHERE id = %s',(card.id,)).fetchone()
            if exists:
                c.execute('UPDATE cards SET board_id=%s,title=%s,list_name=%s,assignee=%s,due_date=%s,description=%s,labels=%s,blocked_by=%s,revision=%s,checklist=%s,updated_at=now() WHERE id=%s',values+(card.id,))
            else:
                c.execute('INSERT INTO cards (board_id,title,list_name,assignee,due_date,description,labels,blocked_by,revision,checklist,id,updated_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,now())',values+(card.id,))

    def log(self,board_id,card_id,actor,action,changes=None):
        with self._db() as c:
            c.execute('INSERT INTO activity (id,board_id,card_id,actor,action,changes,event_at) VALUES (%s,%s,%s,%s,%s,%s,now())',
                      (uuid.uuid4().hex,board_id,card_id,actor,action,json.dumps(changes or {})))

    def activity(self,board_id):
        with self._db() as c:
            rows=c.execute('SELECT id,board_id,card_id,actor,action,changes,event_at FROM activity WHERE board_id = %s ORDER BY event_at DESC, id DESC LIMIT 100',(board_id,)).fetchall()
        out=[]
        for row in rows:
            item=dict(zip(('id','board_id','card_id','actor','action','changes','event_at'),row))
            item['changes']=json.loads(item['changes'] or '{}')
            item['event_at']=item['event_at'].isoformat()
            out.append(item)
        return out