Build Your Own Ledger

"""toy_ledger.py: Coop's core ideas in miniature. One writer, a write fence,
a guarded take, idempotent commands, an invitation nonce, one-transaction
accept with an outcome obligation, reports, and a sweep. Stdlib only."""
import fcntl, hashlib, json, os, re, secrets, sqlite3, tempfile, time

OUTCOME_DUE = 2 * 3600  # Coop's backstop after a managed accept is two hours
SCHEMA = """
CREATE TABLE IF NOT EXISTS todos(id INTEGER PRIMARY KEY, title TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'open', taken_by TEXT, revision INTEGER NOT NULL DEFAULT 1);
CREATE TABLE IF NOT EXISTS commands(key TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, result TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS invitations(id TEXT PRIMARY KEY, todo_id INTEGER NOT NULL, session TEXT NOT NULL,
  nonce TEXT NOT NULL, revision INTEGER NOT NULL, status TEXT NOT NULL DEFAULT 'invited');
CREATE TABLE IF NOT EXISTS obligations(id INTEGER PRIMARY KEY, todo_id INTEGER NOT NULL, session TEXT NOT NULL,
  due REAL NOT NULL, status TEXT NOT NULL DEFAULT 'pending', note TEXT);
"""
TABLES = ('todos', 'commands', 'invitations', 'obligations')


class Conflict(Exception):
    pass


class Ledger:
    def __init__(self, path):
        self.path, self.closed = path, False
        self.lock = open(path + '.lock', 'a+')
        try:  # One writer: an exclusive, non-blocking flock beside the store.
            fcntl.flock(self.lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            self.lock.close()
            raise Conflict('another writer owns this ledger')
        conn = self.connect()
        conn.executescript(SCHEMA)
        for table in TABLES:  # The fence: shared triggers, private permission.
            for op in ('INSERT', 'UPDATE', 'DELETE'):
                conn.execute(f"""CREATE TRIGGER IF NOT EXISTS fence_{table}_{op.lower()}
                    BEFORE {op} ON {table} BEGIN
                    SELECT CASE WHEN write_permit() != 1 THEN RAISE(ABORT, 'writer fenced') END; END""")
        conn.close()

    def connect(self):
        if self.closed:
            raise Conflict('this writer is stopped')
        conn = sqlite3.connect(self.path, isolation_level=None, timeout=5)
        conn.create_function('write_permit', 0, lambda: int(not self.closed))
        return conn

    def close(self):
        self.closed = True
        fcntl.flock(self.lock.fileno(), fcntl.LOCK_UN)
        self.lock.close()

    def command(self, key, name, args, run):
        """Run `run(conn)` once per key; a replay returns the stored answer."""
        fingerprint = hashlib.sha256(json.dumps([name, args]).encode()).hexdigest()
        conn = self.connect()
        try:
            conn.execute('BEGIN IMMEDIATE')
            row = conn.execute('SELECT fingerprint, result FROM commands WHERE key=?', (key,)).fetchone()
            if row:
                conn.execute('ROLLBACK')
                if row[0] != fingerprint:
                    raise Conflict(f'key {key!r} already names a different command')
                return json.loads(row[1])
            result = run(conn)
            conn.execute('INSERT INTO commands VALUES (?,?,?)', (key, fingerprint, json.dumps(result)))
            conn.execute('COMMIT')
            return result
        except BaseException:
            if conn.in_transaction:
                conn.execute('ROLLBACK')
            raise
        finally:
            conn.close()

    def add(self, key, title):
        return self.command(key, 'add', [title], lambda c: c.execute(
            'INSERT INTO todos(title) VALUES (?)', (title,)).lastrowid)

    def edit(self, key, todo_id, title):
        return self.command(key, 'edit', [todo_id, title], lambda c: c.execute(
            'UPDATE todos SET title=?, revision=revision+1 WHERE id=?', (title, todo_id)).rowcount)

    def invite(self, key, todo_id, session):
        """Offer work without its text: the invitation carries only a nonce."""
        def run(c):
            row = c.execute('SELECT status, revision FROM todos WHERE id=?', (todo_id,)).fetchone()
            if not row or row[0] != 'open':
                raise Conflict(f'todo #{todo_id} is not open')
            if c.execute("SELECT 1 FROM invitations WHERE todo_id=? AND status='invited'", (todo_id,)).fetchone():
                raise Conflict(f'todo #{todo_id} already has an active invitation')
            iid, nonce = secrets.token_hex(4), secrets.token_hex(8)
            c.execute('INSERT INTO invitations(id, todo_id, session, nonce, revision) VALUES (?,?,?,?,?)',
                      (iid, todo_id, session, nonce, row[1]))
            return {'invitation': iid, 'nonce': nonce}
        return self.command(key, 'invite', [todo_id, session], run)

    def accept(self, key, invitation, session, answer):
        """Nonce check, guarded take, obligation: one transaction, then the task."""
        def run(c):
            row = c.execute("SELECT todo_id, session, nonce, revision FROM invitations WHERE id=? AND status='invited'",
                            (invitation,)).fetchone()
            if not row or row[1] != session:
                raise Conflict('invitation is stale or belongs to another receiver')
            if not re.search(r'(?<![\w-])ACCEPT\s+' + re.escape(row[2]) + r'(?![\w-])', answer):
                raise Conflict('the answer does not echo this invitation\'s nonce')
            taken = c.execute("UPDATE todos SET status='taken', taken_by=?, revision=revision+1 "
                              "WHERE id=? AND status='open' AND revision=?", (session, row[0], row[3])).rowcount
            if taken != 1:
                c.execute("UPDATE invitations SET status='stale' WHERE id=?", (invitation,))
                return {'refused': 'work changed before acceptance; invite again'}
            c.execute('INSERT INTO obligations(todo_id, session, due) VALUES (?,?,?)',
                      (row[0], session, time.time() + OUTCOME_DUE))
            c.execute("UPDATE invitations SET status='accepted' WHERE id=?", (invitation,))
            title = c.execute('SELECT title FROM todos WHERE id=?', (row[0],)).fetchone()[0]
            return {'todo': row[0], 'task': title}
        return self.command(key, 'accept', [invitation, session, answer], run)

    def report(self, key, todo_id, session, outcome, body):
        """done closes the todo and pays the debt; blocked keeps both, owed now."""
        if outcome not in ('done', 'blocked') or not body.strip():
            raise ValueError('an outcome needs done|blocked and a body')
        def run(c):
            owed = c.execute("SELECT id FROM obligations WHERE todo_id=? AND session=? AND status='pending'",
                             (todo_id, session)).fetchone()
            if not owed:
                raise Conflict(f'{session} owes nothing on todo #{todo_id}')
            if outcome == 'blocked':
                c.execute('UPDATE obligations SET due=?, note=? WHERE id=?', (time.time(), body, owed[0]))
                return {'todo': todo_id, 'status': 'still taken, review owed now'}
            c.execute("UPDATE todos SET status='done', revision=revision+1 WHERE id=?", (todo_id,))
            c.execute("UPDATE obligations SET status='resolved', note=? WHERE id=?", (body, owed[0]))
            return {'todo': todo_id, 'status': 'done'}
        return self.command(key, 'report', [todo_id, session, outcome, body], run)

    def sweep(self):
        """Every outcome still owed, soonest first. A read: no key, no receipt."""
        conn = self.connect()
        rows = conn.execute("SELECT todo_id, session, due, note FROM obligations "
                            "WHERE status='pending' ORDER BY due").fetchall()
        conn.close()
        now = time.time()
        return [f"#{t} {s}: " + ('owed now' if due <= now else f'due in {round((due - now) / 60)} min')
                + (f' ({note})' if note else '') for t, s, due, note in rows]


def main():
    path = os.path.join(tempfile.mkdtemp(prefix='toy-ledger-'), 'ledger.db')
    ledger = Ledger(path)
    try:
        Ledger(path)
    except Conflict as exc:
        print('second writer :', exc)
    outsider = sqlite3.connect(path)
    try:
        outsider.execute("INSERT INTO todos(title) VALUES ('sneaky')")
    except sqlite3.OperationalError as exc:
        print('outside write :', exc)
    outsider.close()
    a = ledger.add('add-a', 'Write the fleet lesson')
    b = ledger.add('add-b', 'Fix the flaky test')
    c = ledger.add('add-c', 'Rotate the logs')
    print('replayed add  :', ledger.add('add-a', 'Write the fleet lesson') == a)
    try:
        ledger.add('add-a', 'Something else entirely')
    except Conflict as exc:
        print('reused key    :', exc)
    inv = ledger.invite('inv-a', a, 'laptop/chat-7')
    try:
        ledger.accept('acc-a0', inv['invitation'], 'laptop/chat-7', 'Sure, on it!')
    except Conflict as exc:
        print('no nonce      :', exc)
    print('accepted      :', ledger.accept('acc-a', inv['invitation'], 'laptop/chat-7', f"ACCEPT {inv['nonce']}"))
    inv = ledger.invite('inv-b', b, 'office/chat-3')
    ledger.edit('edit-b', b, 'Fix the flaky test in both suites')
    print('edited first  :', ledger.accept('acc-b', inv['invitation'], 'office/chat-3', f"ok, ACCEPT {inv['nonce']}"))
    inv = ledger.invite('inv-b2', b, 'office/chat-3')
    ledger.accept('acc-b2', inv['invitation'], 'office/chat-3', f"ACCEPT {inv['nonce']}")
    inv = ledger.invite('inv-c', c, 'laptop/chat-9')
    ledger.accept('acc-c', inv['invitation'], 'laptop/chat-9', f"ACCEPT {inv['nonce']}")
    print('blocked       :', ledger.report('rep-b', b, 'office/chat-3', 'blocked', 'needs Dad: which suite first?'))
    print('done          :', ledger.report('rep-a', a, 'laptop/chat-7', 'done', 'pushed; remote SHA verified'))
    for line in ledger.sweep():
        print('owed          :', line)
    ledger.close()


if __name__ == '__main__':
    main()

second writer : another writer owns this ledger
outside write : no such function: write_permit
replayed add  : True
reused key    : key 'add-a' already names a different command
no nonce      : the answer does not echo this invitation's nonce
accepted      : {'todo': 1, 'task': 'Write the fleet lesson'}
edited first  : {'refused': 'work changed before acceptance; invite again'}
blocked       : {'todo': 2, 'status': 'still taken, review owed now'}
done          : {'todo': 1, 'status': 'done'}
owed          : #2 office/chat-3: owed now (needs Dad: which suite first?)
owed          : #3 laptop/chat-9: due in 120 min