The Change Feed

# Coop's engine: the feed triggers and the feed reader (trimmed excerpt)

def _triggers(self, conn, fence):
    # All legacy mutations, including edits without a lifecycle event, enter the feed.
    # A trigger body is a snapshot of the table's columns when it was
    # created; one that no longer lists every column is rebuilt so a
    # migration can never silently drop a field from the change feed.
    conn.execute('BEGIN IMMEDIATE')
    for table in TRACKED:
        columns = [r['name'] for r in conn.execute(f'PRAGMA table_info({table})')]
        for op in ('INSERT', 'UPDATE', 'DELETE'):
            name = f'coop_track_{table}_{op.lower()}'
            ref = 'OLD' if op == 'DELETE' else 'NEW'
            body = 'NULL' if op == 'DELETE' else 'json_object(' + ','.join(f"'{c}',{ref}.{c}" for c in columns) + ')'
            existing = conn.execute("SELECT sql FROM sqlite_master WHERE type='trigger' AND name=?", (name,)).fetchone()
            if existing and op != 'DELETE' and body not in existing['sql']:
                conn.execute(f'DROP TRIGGER {name}')
            conn.execute(f'''CREATE TRIGGER IF NOT EXISTS {name} AFTER {op} ON {table} BEGIN
              INSERT INTO engine_changes(entity,row_id,operation,body,revision,actor,session,causation,created)
              VALUES ('{table}',CAST({ref}.id AS TEXT),'{op.lower()}',{body},COALESCE((SELECT revision FROM engine_revisions WHERE entity='{table}' AND row_id=CAST({ref}.id AS TEXT)),0)+1,coop_context('id'),coop_context('session'),coop_context('causation'),unixepoch('subsec'));
              INSERT INTO engine_revisions VALUES ('{table}',CAST({ref}.id AS TEXT),1)
              ON CONFLICT(entity,row_id) DO UPDATE SET revision=revision+1;
            END''')
            ...  # the coop_fence_* BEFORE trigger from lesson 1
    conn.commit()


def changes(self, after=0, limit=500, epoch=None):
    if after < 0 or not 1 <= limit <= 1000 or epoch not in (None, self.epoch):
        raise Conflict('Invalid cursor or store epoch; obtain a snapshot')
    ...
    if after > maximum:
        raise Conflict('Cursor is beyond this store; obtain a snapshot')
    if minimum and after and after < minimum - 1:
        raise Conflict('Cursor precedes the retained changes; obtain a snapshot')
    return {'epoch': self.epoch, 'changes': rows, 'cursor': rows[-1]['id'] if rows else after, 'watermark': maximum}

"""A minimal sketch of Coop's change feed: AFTER triggers copy every mutation
into a feed with a per-row revision, and a trigger whose column snapshot has
gone stale is rebuilt."""
import json
import sqlite3

conn = sqlite3.connect(":memory:", isolation_level=None)
conn.executescript("""
CREATE TABLE todos (id INTEGER PRIMARY KEY, title TEXT, status TEXT DEFAULT 'open');
CREATE TABLE changes (id INTEGER PRIMARY KEY AUTOINCREMENT, entity TEXT, row_id TEXT,
                      operation TEXT, body TEXT, revision INTEGER);
CREATE TABLE revisions (entity TEXT, row_id TEXT, revision INTEGER, PRIMARY KEY(entity, row_id));
""")


def install_feed(table):
    columns = [row[1] for row in conn.execute(f"PRAGMA table_info({table})")]
    for op in ("INSERT", "UPDATE", "DELETE"):
        name = f"track_{table}_{op.lower()}"
        ref = "OLD" if op == "DELETE" else "NEW"
        body = "NULL" if op == "DELETE" else "json_object(" + ",".join(f"'{c}',{ref}.{c}" for c in columns) + ")"
        existing = conn.execute("SELECT sql FROM sqlite_master WHERE type='trigger' AND name=?", (name,)).fetchone()
        if existing and op != "DELETE" and body not in existing[0]:
            conn.execute(f"DROP TRIGGER {name}")  # its column snapshot is stale
        conn.execute(f"""CREATE TRIGGER IF NOT EXISTS {name} AFTER {op} ON {table} BEGIN
          INSERT INTO changes(entity, row_id, operation, body, revision)
          VALUES ('{table}', CAST({ref}.id AS TEXT), '{op.lower()}', {body},
                  COALESCE((SELECT revision FROM revisions
                            WHERE entity='{table}' AND row_id=CAST({ref}.id AS TEXT)), 0) + 1);
          INSERT INTO revisions VALUES ('{table}', CAST({ref}.id AS TEXT), 1)
          ON CONFLICT(entity, row_id) DO UPDATE SET revision = revision + 1;
        END""")


def revision(row_id):
    row = conn.execute("SELECT revision FROM revisions WHERE entity='todos' AND row_id=?", (str(row_id),)).fetchone()
    return row[0] if row else 0


def last_body():
    return json.loads(conn.execute("SELECT body FROM changes ORDER BY id DESC LIMIT 1").fetchone()[0])


install_feed("todos")
todo = conn.execute("INSERT INTO todos(title) VALUES ('Write the lesson')").lastrowid
accepted = revision(todo)  # a worker accepts the work at this revision
conn.execute("UPDATE todos SET title='Write the lesson in both locales' WHERE id=?", (todo,))
print("accepted at", accepted, "| now", revision(todo), "| report",
      "refused: reconcile first" if revision(todo) != accepted else "accepted")

conn.execute("ALTER TABLE todos ADD COLUMN note TEXT")  # a migration adds a field
conn.execute("UPDATE todos SET note='first' WHERE id=?", (todo,))
print("stale trigger  :", last_body())
install_feed("todos")  # the engine re-runs this at every start
conn.execute("UPDATE todos SET note='second' WHERE id=?", (todo,))
print("rebuilt trigger:", last_body())
print("feed:", conn.execute("SELECT id, operation, revision FROM changes").fetchall())

# Expected output:
# accepted at 1 | now 2 | report refused: reconcile first
# stale trigger  : {'id': 1, 'title': 'Write the lesson in both locales', 'status': 'open'}
# rebuilt trigger: {'id': 1, 'title': 'Write the lesson in both locales', 'status': 'open', 'note': 'second'}
# feed: [(1, 'insert', 1), (2, 'update', 2), (3, 'update', 3), (4, 'update', 4)]