Bound topic event history queries

src/msg/plugins/content.py

@@ -46,6 +46,8 @@ async def topic_governance_event(tx,ctx,request,topic,action,target, *, reason=N
                 data={'topic_id':topic,'action':action,'target_subject':target,
                       **({'reason':reason} if reason else {})})
     await tx.append_event(event)
+    tx.execute('''INSERT INTO topic_event_projection (seq,topic,body)
+        SELECT seq,?,body FROM events WHERE id=?''',(topic,event.id),write=True)
     if target is not None:
         notice={'id':new_id('message'),'sender':ctx.principal.subject,'actor':ctx.principal.actor,
                 'recipient':target,'resource':{'id':target,'revision':None},'time':wire(ctx.now),
@@ -451,13 +453,11 @@ async def topic_events(ctx,request,tx):
         position=app.cursors.decode(request.arguments['cursor'],'topic_events',binding) if request.arguments.get('cursor') else 9223372036854775807
         limit=request.arguments.get('limit',10)
         found=[]
-        for seq,raw in tx.execute('SELECT seq,body FROM events WHERE seq<? ORDER BY seq DESC',(position,)):
+        rows=tx.execute('''SELECT seq,body FROM topic_event_projection
+            WHERE topic=? AND seq<? ORDER BY seq DESC LIMIT ?''',(topic,position,limit+1))
+        for seq,raw in rows:
             event=loads(raw)
-            if event['type'] not in TOPIC_EVENT_CODES or event['data'].get('topic_id')!=topic:
-                continue
             found.append((seq,event))
-            if len(found)>limit:
-                break
         items=[]
         for seq,event in found[:limit]:
             data=dict(event['data'])

src/msg/storage/postgres.py

@@ -77,6 +77,18 @@
 CREATE TABLE IF NOT EXISTS transfers (id TEXT PRIMARY KEY, subject TEXT NOT NULL, generation INTEGER NOT NULL, body TEXT NOT NULL, limits TEXT NOT NULL DEFAULT '{}');
 CREATE TABLE IF NOT EXISTS chunks (transfer_id TEXT NOT NULL, "offset" BIGINT NOT NULL, length BIGINT NOT NULL, body TEXT NOT NULL, PRIMARY KEY(transfer_id,"offset"));
 CREATE TABLE IF NOT EXISTS events (seq BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, id TEXT UNIQUE NOT NULL, body TEXT NOT NULL);
+CREATE TABLE IF NOT EXISTS topic_event_projection (
+ seq BIGINT PRIMARY KEY REFERENCES events(seq) ON DELETE CASCADE,
+ topic TEXT NOT NULL, body TEXT NOT NULL);
+CREATE INDEX IF NOT EXISTS topic_event_projection_topic ON topic_event_projection(topic,seq DESC);
+INSERT INTO topic_event_projection (seq,topic,body)
+ SELECT seq,body::jsonb->'data'->>'topic_id',body FROM events
+ WHERE body::jsonb->>'type' IN (
+  'topic.create','topic.member.join','topic.member.request','topic.member.leave',
+  'topic.member.invite','topic.member.approve','topic.member.remove','topic.member.promote',
+  'topic.member.demote','topic.member.ban','topic.member.unban','topic.policy.change',
+  'topic.archive','topic.restore','topic.move','topic.chmod','topic.chgrp','topic.chown',
+  'topic.configure') ON CONFLICT DO NOTHING;
 CREATE TABLE IF NOT EXISTS audit (seq BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, digest TEXT UNIQUE NOT NULL, previous TEXT, body TEXT NOT NULL);
 CREATE OR REPLACE FUNCTION msg_audit_append_only() RETURNS trigger LANGUAGE plpgsql AS $$
 BEGIN RAISE EXCEPTION 'append_only_audit' USING ERRCODE = '23514'; END;

src/msg/storage/sqlite.py

@@ -74,6 +74,18 @@
 CREATE TABLE IF NOT EXISTS transfers (id TEXT PRIMARY KEY, subject TEXT NOT NULL, generation INTEGER NOT NULL, body TEXT NOT NULL, limits TEXT NOT NULL DEFAULT '{}');
 CREATE TABLE IF NOT EXISTS chunks (transfer_id TEXT NOT NULL, offset INTEGER NOT NULL, length INTEGER NOT NULL, body TEXT NOT NULL, PRIMARY KEY(transfer_id,offset));
 CREATE TABLE IF NOT EXISTS events (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT UNIQUE NOT NULL, body TEXT NOT NULL);
+CREATE TABLE IF NOT EXISTS topic_event_projection (
+ seq INTEGER PRIMARY KEY REFERENCES events(seq) ON DELETE CASCADE,
+ topic TEXT NOT NULL, body TEXT NOT NULL);
+CREATE INDEX IF NOT EXISTS topic_event_projection_topic ON topic_event_projection(topic,seq DESC);
+INSERT OR IGNORE INTO topic_event_projection (seq,topic,body)
+ SELECT seq,json_extract(body,'$.data.topic_id'),body FROM events
+ WHERE json_extract(body,'$.type') IN (
+  'topic.create','topic.member.join','topic.member.request','topic.member.leave',
+  'topic.member.invite','topic.member.approve','topic.member.remove','topic.member.promote',
+  'topic.member.demote','topic.member.ban','topic.member.unban','topic.policy.change',
+  'topic.archive','topic.restore','topic.move','topic.chmod','topic.chgrp','topic.chown',
+  'topic.configure');
 CREATE TABLE IF NOT EXISTS audit (seq INTEGER PRIMARY KEY AUTOINCREMENT, digest TEXT UNIQUE NOT NULL, previous TEXT, body TEXT NOT NULL);
 CREATE TRIGGER IF NOT EXISTS audit_no_update BEFORE UPDATE ON audit BEGIN SELECT RAISE(ABORT,'append_only_audit'); END;
 CREATE TRIGGER IF NOT EXISTS audit_no_delete BEFORE DELETE ON audit BEGIN SELECT RAISE(ABORT,'append_only_audit'); END;