Community: the relay's Events worker reads by opaque cursor (latest_cursor, next_cursor) (#45)

ddiff --git a/migrations/0006_relay_cursor_position.sql b/migrations/0006_relay_cursor_position.sql
nnew file mode 100644
iindex 0000000..c3339e4
---- /dev/null
++++ b/migrations/0006_relay_cursor_position.sql
@@@ -0,0 +1,6 @@
++-- phase: expand
++-- The Discord relay's Events worker keeps its place as Events' opaque cursor (ADR-042 decision 7; plan T7 retires the
++-- numeric seq): `position` holds the page's next_cursor, or the head's latest_cursor on a first start. `cursor` (the
++-- numeric seq) is still written while Events sends it, so a worker of the previous release keeps its place too; a
++-- later contract migration drops it once Events stops sending numeric positions.
++ALTER TABLE relay_cursors ADD COLUMN IF NOT EXISTS position text;
ddiff --git a/server/relay/events-worker.js b/server/relay/events-worker.js
iindex b6f82e2..d2acb9b 100644
---- a/server/relay/events-worker.js
++++ b/server/relay/events-worker.js
@@@ -8,8 +8,12 @@
   *
   *   - Cursor: the relay_cursors row 'discord-relay:events', moved in the same transaction as the
   *     deliveries its page queued. A crash replays at most the page in progress, and the dedupe keys
-- *     make that replay queue nothing twice.
-- *   - First start (no cursor yet): it begins at Events' head. Threads from before are never announced.
++ *     make that replay queue nothing twice. `position` is Events' opaque cursor (the page's next_cursor),
++ *     the place the worker reads on from; `cursor` is the numeric seq, still written while Events sends it
++ *     (a worker of the previous release, and the lag shown to staff). A row with no position yet (written
++ *     by the previous release) is read on from its numeric cursor once, and has a position from then on.
++ *   - First start (no cursor yet): it begins at Events' head (latest_cursor). Threads from before are never
++ *     announced.
   *   - Events carry ids, never content: the worker reads the thread or post as it is now and the relay's
   *     rules decide (public spaces only, never members-only, never origin 'discord'). Redacted events
   *     and events from any other source are ignored.
@@@ -51,11 +55,13 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, c
      let stopped = false;
  
      const iso = () => new Date(now()).toISOString();
--    const readCursor = async () => await db.prepare('SELECT cursor, latest_seq FROM relay_cursors WHERE name = ?').get(CURSOR) || null;
--    async function writeCursor(cursor, latest) {
--        await db.prepare(`INSERT INTO relay_cursors (name, cursor, latest_seq, updated_at) VALUES (?, ?, ?, ov_now())
--                    ON CONFLICT(name) DO UPDATE SET cursor = excluded.cursor, latest_seq = excluded.latest_seq, updated_at = ov_now()`).run(CURSOR, cursor, latest == null ? null : latest);
++    const readCursor = async () => await db.prepare('SELECT cursor, position, latest_seq FROM relay_cursors WHERE name = ?').get(CURSOR) || null;
++    async function writeCursor(cursor, latest, position) {
++        await db.prepare(`INSERT INTO relay_cursors (name, cursor, latest_seq, position, updated_at) VALUES (?, ?, ?, ?, ov_now())
++                    ON CONFLICT(name) DO UPDATE SET cursor = excluded.cursor, latest_seq = excluded.latest_seq, position = excluded.position, updated_at = ov_now()`)
++            .run(CURSOR, cursor, latest == null ? null : latest, position || null);
      }
++    const num = (v) => (Number.isFinite(Number(v)) && v !== null && v !== undefined ? Number(v) : null);
  
      /** One event → deliveries queued (0 when it is not for the relay). */
      async function handle(ev) {
@@@ -86,33 +92,43 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, c
              try {
                  let cur = await readCursor();
                  if (!cur) {
--                    const head = await client.pull({ topic: TOPICS, afterSeq: 0, limit: 1 });
--                    const latest = Number(head.latest_seq) || 0;
--                    await writeCursor(latest, latest);
++                    // First start: Events' head, as an opaque cursor (the numeric seq beside it while Events sends one).
++                    const head = await client.pull({ topic: TOPICS, limit: 1 });
++                    const latest = num(head.latest_seq) || 0;
++                    await writeCursor(latest, latest, head.latest_cursor || null);
                      stats.started_at_seq = latest;
--                    console.log(`[Relay] Events worker starts at seq ${latest} (earlier threads are not announced)`);
--                    cur = { cursor: latest, latest_seq: latest };
++                    console.log(`[Relay] Events worker starts at the head (seq ${latest}; earlier threads are not announced)`);
++                    cur = { cursor: latest, position: head.latest_cursor || null, latest_seq: latest };
                  }
--                let cursor = Number(cur.cursor) || 0;
++                let cursor = num(cur.cursor) || 0;
++                let position = cur.position || null;
                  for (let pages = 0; pages < MAX_PAGES_PER_TICK && !stopped; pages++) {
--                    const page = await client.pull({ topic: TOPICS, afterSeq: cursor, limit: pageLimit });
++                    // Read on from the opaque position; a row from the previous release has only the numeric cursor.
++                    const page = position
++                        ? await client.pull({ topic: TOPICS, after: position, limit: pageLimit })
++                        : await client.pull({ topic: TOPICS, afterSeq: cursor, limit: pageLimit });
                      if (stopped) break;
                      if (page.gap) {
                          stats.last_gap = { from_seq: page.gap.from_seq, to_seq: page.gap.to_seq, at: iso() };
                          console.warn(`[Relay] Events gap: seq ${page.gap.from_seq}..${page.gap.to_seq} were pruned before the relay read them; threads and replies in it are not relayed`);
                      }
--                    const next = Number(page.next_after_seq);
--                    const latest = Number(page.latest_seq);
--                    const moved = Number.isFinite(next) && next > cursor ? next : cursor;
++                    const nextPosition = page.next_cursor || position;
++                    const next = num(page.next_after_seq);
++                    const nextCursor = next != null && next > cursor ? next : cursor;
++                    const latest = num(page.latest_seq);
++                    const items = page.events || [];
                      queuedNow += await db.tx(async () => {
                          let n = 0;
--                        for (const item of page.events || []) n += await handle(item && item.event);
--                        await writeCursor(moved, Number.isFinite(latest) ? latest : null);
++                        for (const item of items) n += await handle(item && item.event);
++                        await writeCursor(nextCursor, latest, nextPosition);
                          return n;
                      });
                      stats.pages++;
--                    if (moved === cursor || !Number.isFinite(latest) || moved >= latest) break;
--                    cursor = moved;
++                    const moved = nextPosition !== position || nextCursor !== cursor;
++                    position = nextPosition;
++                    cursor = nextCursor;
++                    // A short page means the read reached the head (next_cursor already moved past what did not match).
++                    if (!moved || items.length < pageLimit) break;
                  }
                  stats.queued += queuedNow;
                  stats.last_ok_at = iso();
@@@ -151,7 +167,7 @@ function createRelayEventsWorker({ db, relay, events = null, eventsUrl = null, c
          const cur = await readCursor();
          return {
              enabled: true, running: !!timer, topics: TOPICS,
--            cursor: cur ? cur.cursor : null, latest_seq: cur ? cur.latest_seq : null,
++            cursor: cur ? cur.cursor : null, position: cur ? cur.position || null : null, latest_seq: cur ? cur.latest_seq : null,
              lag: cur && cur.latest_seq != null ? Math.max(cur.latest_seq - cur.cursor, 0) : null,
              events_seen: stats.seen, deliveries_queued: stats.queued, started_at_seq: stats.started_at_seq,
              last_ok_at: stats.last_ok_at, last_error: stats.last_error, last_error_at: stats.last_error_at, last_gap: stats.last_gap,
ddiff --git a/test/helpers/fake-events.js b/test/helpers/fake-events.js
iindex c607d73..db92617 100644
---- a/test/helpers/fake-events.js
++++ b/test/helpers/fake-events.js
@@@ -4,7 +4,9 @@
   * the pull route the relay's Events worker reads.
   *
   *   POST /api/v1/events   one envelope or { events: [...] } → { event_id, seq, duplicate } | { results }
-- *   GET  /api/v1/events?topic=a.*,b.*&after_seq=&limit=  → { events: [{ seq, event }], next_after_seq, latest_seq, gap? }
++ *   GET  /api/v1/events?topic=a.*,b.*&after=<cursor>|after_seq=&limit=
++ *        → { events: [{ seq, cursor, event }], next_after_seq, next_cursor, latest_seq, latest_cursor, gap? }
++ *   Cursors are opaque to the worker; here, as in Events, `c1.<epoch>.<base64url(seq)>` (one epoch, 0).
   *
   * Topic patterns follow Events (`*` is one or more whole segments). Every call needs a Bearer token.
   * `failNext` answers that many calls with 503; `prunedThrough` makes seq ≤ it look pruned (a gap).
@@@ -15,6 +17,9 @@ function matcher(pattern) {
      return new RegExp(`^${pattern.split('.').map((s) => (s === '*' ? '[a-z0-9_]+(?:\\.[a-z0-9_]+)*' : s)).join('\\.')}$`);
  }
  
++const encode = (seq) => `c1.0.${Buffer.from(String(seq), 'utf8').toString('base64url')}`;
++const decode = (c) => { const m = /^c1\.0\.([A-Za-z0-9_-]+)$/.exec(String(c)); return m ? Number(Buffer.from(m[1], 'base64url').toString('utf8')) : NaN; };
++
  function startEvents() {
      const log = [];      // { seq, event }
      const calls = [];
@@@ -42,7 +47,9 @@ function startEvents() {
              }
              if (u.pathname === '/api/v1/events' && req.method === 'GET') {
                  const patterns = String(u.searchParams.get('topic') || '*').split(',').map(matcher);
--                const after = Number(u.searchParams.get('after_seq') || 0);
++                const cur = u.searchParams.get('after');
++                const after = cur ? decode(cur) : Number(u.searchParams.get('after_seq') || 0);
++                if (!Number.isFinite(after)) return json(400, { type: 'about:blank', code: 'events.bad_request', status: 400, detail: 'after must be an opaque cursor' });
                  const limit = Number(u.searchParams.get('limit') || 100);
                  const out = {};
                  let from = after;
@@@ -55,9 +62,11 @@ function startEvents() {
                      if (patterns.some((re) => re.test(item.event.event_type))) rows.push(item);
                      if (rows.length >= limit) break;
                  }
--                out.events = rows;
++                out.events = rows.map((r) => ({ ...r, cursor: encode(r.seq) }));
                  out.next_after_seq = cursor;
++                out.next_cursor = encode(cursor);
                  out.latest_seq = log.length;
++                out.latest_cursor = encode(log.length);
                  return json(200, out);
              }
              return json(404, { type: 'about:blank', code: 'route.not_found', status: 404 });
@@@ -66,7 +75,7 @@ function startEvents() {
      /** Put an event straight into the log (as if some producer had published it). → seq */
      function add(event) { log.push({ seq: log.length + 1, event }); return log.length; }
      return new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve({
--        url: `http://127.0.0.1:${server.address().port}`, log, calls, state, add,
++        url: `http://127.0.0.1:${server.address().port}`, log, calls, state, add, encode,
          close: () => new Promise((r) => server.close(r)),
      })));
  }
ddiff --git a/test/relay-events.test.js b/test/relay-events.test.js
iindex 826b839..a185de4 100644
---- a/test/relay-events.test.js
++++ b/test/relay-events.test.js
@@@ -72,7 +72,8 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
          const st = await worker.status();
          assert.deepStrictEqual([st.enabled, st.cursor, st.latest_seq, st.started_at_seq, st.lag], [true, 3, 3, 3, 0]);
          assert.strictEqual((await deliveries()).length, 0);
--        assert.deepStrictEqual(await db.prepare('SELECT name, cursor FROM relay_cursors').all(), [{ name: CURSOR, cursor: 3 }]);
++        assert.deepStrictEqual(await db.prepare('SELECT name, cursor, position FROM relay_cursors').all(), [{ name: CURSOR, cursor: 3, position: events.encode(3) }], 'the head, stored as the opaque latest_cursor');
++        assert.ok(events.calls.every((c) => !/[?&]after_seq=[1-9]/.test(c.path)), 'no numeric position is ever sent once a cursor is stored');
          before = events.calls.length;
      });
  
@@@ -99,7 +100,7 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
      });
  
      await check('a replay (cursor moved back) queues nothing twice and posts nothing twice', async () => {
--        await db.prepare('UPDATE relay_cursors SET cursor = 3 WHERE name = ?').run(CURSOR);
++        await db.prepare('UPDATE relay_cursors SET position = ? WHERE name = ?').run(events.encode(3), CURSOR);
          assert.deepStrictEqual(await worker.tick(), { queued: 0 });
          await relay.drain();
          assert.strictEqual(hook.hits.length, 2);
@@@ -143,8 +144,21 @@ const postEvent = (p, t, visibility = 'public') => envelope('community.post.crea
          assert.strictEqual((await worker.status()).last_error, null);
      });
  
++    await check('a row from the previous release (numeric cursor, no position) reads on from it once, then by cursor', async () => {
++        await db.prepare('UPDATE relay_cursors SET cursor = 3, position = NULL WHERE name = ?').run(CURSOR);
++        const from = events.calls.length;
++        assert.deepStrictEqual(await worker.tick(), { queued: 0 }, 'what it read again was already queued');
++        const sent = events.calls.slice(from).map((c) => c.path);
++        assert.ok(sent.some((p) => /[?&]after_seq=3\b/.test(p)), 'the first read uses the numeric cursor it had');
++        const row = await db.prepare('SELECT cursor, position FROM relay_cursors WHERE name = ?').get(CURSOR);
++        assert.strictEqual(row.position, events.encode(events.log.length), 'and from then on the opaque cursor');
++        const again = events.calls.length;
++        await worker.tick();
++        assert.ok(events.calls.slice(again).every((c) => /[?&]after=c1\./.test(c.path)), 'later reads send the cursor');
++    });
++
      await check('a retention gap is logged and shown to staff; reading carries on after it', async () => {
--        await db.prepare('UPDATE relay_cursors SET cursor = 1 WHERE name = ?').run(CURSOR);
++        await db.prepare('UPDATE relay_cursors SET position = ? WHERE name = ?').run(events.encode(1), CURSOR);
          events.state.prunedThrough = 5;
          await worker.tick();
          assert.deepStrictEqual([(await worker.status()).last_gap.from_seq, (await worker.status()).last_gap.to_seq], [2, 5]);