diff --git a/docs/groupmind-local.md b/docs/groupmind-local.md index 5617a59..0508411 100644 --- a/docs/groupmind-local.md +++ b/docs/groupmind-local.md @@ -31,7 +31,13 @@ If `IAK_RELAY_TOKEN` is set, every request must present it via `X-API-Key` or | `GET` | `/health` | — | always open; reports message count | | `POST` | `/api/v1/messages` | `{ room, body, from?, metadata? }` | create a message | | `POST` | `/api/v1/rooms/:room/messages` | `{ body, from?, metadata? }` | create (room in path) | -| `GET` | `/api/v1/rooms/:room/messages` | `?limit=&since=` | newest-first, GroupMind shape | +| `GET` | `/api/v1/rooms/:room/messages` | `?limit=&since=&after=` | see below | + +`GET` returns the recent view (**newest-first**) by default. Pass `after=` +for an incremental **cursor feed**: messages with `seq > after`, **oldest-first**, +capped by `limit` — the client advances its cursor and re-fetches, so nothing is +skipped even when messages share a millisecond timestamp. Every message carries a +monotonic `seq` (the exact cursor) alongside its millisecond `created_at`. An agent points its room client at `http://:18790/api/v1` when cloud GroupMind is unreachable — same paths, just a different base URL. diff --git a/scripts/local-relay.mjs b/scripts/local-relay.mjs index ef73265..307338d 100644 --- a/scripts/local-relay.mjs +++ b/scripts/local-relay.mjs @@ -35,7 +35,7 @@ import http from 'node:http'; import { appendFileSync, mkdirSync, readFileSync, existsSync } from 'node:fs'; -import { randomUUID } from 'node:crypto'; +import { randomUUID, timingSafeEqual, createHash } from 'node:crypto'; import { homedir } from 'node:os'; import { dirname, join } from 'node:path'; @@ -52,6 +52,10 @@ function readMessages(storePath) { if (!line.trim()) continue; try { out.push(JSON.parse(line)); } catch { /* skip a corrupt line, never crash a read */ } } + // `seq` is the 1-based append position, assigned HERE on read so it is present + // and correct even for rows written before seq existed (e.g. a store from the + // #37 MVP). This is the authoritative cursor value; any stored seq is ignored. + out.forEach((m, i) => { m.seq = i + 1; }); return out; } @@ -60,6 +64,14 @@ function appendMessage(storePath, msg) { appendFileSync(storePath, JSON.stringify(msg) + '\n'); } +// Constant-time token compare. Hash both sides to a fixed 32 bytes first so +// timingSafeEqual never sees a length mismatch (and length isn't leaked). +function safeTokenEqual(provided, expected) { + const a = createHash('sha256').update(String(provided ?? '')).digest(); + const b = createHash('sha256').update(String(expected ?? '')).digest(); + return timingSafeEqual(a, b); +} + function send(res, status, obj) { const payload = JSON.stringify(obj); res.writeHead(status, { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(payload) }); @@ -115,10 +127,18 @@ export function startRelay(opts = {}) { if (!token) log('WARNING: no IAK_RELAY_TOKEN set — relay is LAN-open (fine on a trusted home LAN only).'); + // `seq` = 1-based append position (see readMessages). `created_at` is + // millisecond-resolution so it can't cursor exactly (a same-ms checkpoint + // would skip messages); `seq` is the precise cursor the failover + sync + // layers rely on. `count` is seeded from the (possibly pre-seq) store so old + // rows are counted, and it tracks the next position across appends in-process. + let count = readMessages(storePath).length; + const commit = (msg) => { msg.seq = ++count; appendMessage(storePath, msg); return msg; }; + function authorized(req) { if (!token) return true; const bearer = (req.headers['authorization'] || '').replace(/^Bearer\s+/i, ''); - return req.headers['x-api-key'] === token || bearer === token; + return safeTokenEqual(req.headers['x-api-key'], token) || safeTokenEqual(bearer, token); } const server = http.createServer(async (req, res) => { @@ -138,12 +158,24 @@ export function startRelay(opts = {}) { const room = decodeURIComponent(getMatch[1]); const limit = Math.min(Math.max(1, Number(url.searchParams.get('limit')) || DEFAULT_LIMIT), MAX_LIMIT); const since = url.searchParams.get('since'); + const after = url.searchParams.get('after'); // exact seq cursor (authoritative) + const cursoring = after !== null && after !== ''; + // readMessages returns JSONL append order = chronological, each with its + // positional seq. let msgs = readMessages(storePath).filter((m) => m.room === room); - if (since) msgs = msgs.filter((m) => m.created_at > since); - // JSONL append order IS chronological, so it orders even messages that - // share a millisecond timestamp. Take the last `limit` (newest) and - // reverse to newest-first, matching GroupMind's shape. - const messages = msgs.slice(-limit).reverse(); + if (cursoring) { + // `after` (seq) is authoritative — never ALSO narrow by the + // millisecond `since`, which would re-drop the same-ms messages the + // cursor is meant to return (PR #38 review). + msgs = msgs.filter((m) => m.seq > Number(after)); + } else if (since) { + msgs = msgs.filter((m) => m.created_at > since); + } + // Cursor feed (`after`): chronological forward from the cursor, capped — + // the client advances its cursor and re-fetches, so nothing is skipped + // even for same-millisecond messages. Recent view (no cursor): + // newest-first, matching GroupMind's shape. + const messages = cursoring ? msgs.slice(0, limit) : msgs.slice(-limit).reverse(); return send(res, 200, { messages, count: messages.length }); } @@ -152,8 +184,7 @@ export function startRelay(opts = {}) { const b = await readJsonBody(req); const { msg, error } = buildMessage({ room: b.room, body: b.body, from: b.from, metadata: b.metadata }); if (error) return send(res, 400, { error }); - appendMessage(storePath, msg); - return send(res, 201, msg); + return send(res, 201, commit(msg)); } // POST /api/v1/rooms/:room/messages { body, ... } @@ -163,8 +194,7 @@ export function startRelay(opts = {}) { const b = await readJsonBody(req); const { msg, error } = buildMessage({ room, body: b.body, from: b.from, metadata: b.metadata }); if (error) return send(res, 400, { error }); - appendMessage(storePath, msg); - return send(res, 201, msg); + return send(res, 201, commit(msg)); } return send(res, 404, { error: 'not found' }); diff --git a/test/local-relay.test.mjs b/test/local-relay.test.mjs index 9dc29ea..b8ec4da 100644 --- a/test/local-relay.test.mjs +++ b/test/local-relay.test.mjs @@ -3,7 +3,7 @@ import test from 'node:test'; import { once } from 'node:events'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { rmSync } from 'node:fs'; +import { rmSync, writeFileSync } from 'node:fs'; import { randomUUID } from 'node:crypto'; import { startRelay } from '../scripts/local-relay.mjs'; @@ -126,3 +126,84 @@ test('unknown route is 404', async () => { assert.equal(r.status, 404); }); }); + +test('messages get a monotonic seq; after-cursor is a no-skip chronological feed', async () => { + await withRelay({}, async ({ base, headers }) => { + const posted = []; + for (const n of [1, 2, 3, 4]) { + const r = await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'c', body: `m${n}` }), + }); + posted.push(await r.json()); + } + // seq strictly increasing even when created_at collides in the same ms + for (let i = 1; i < posted.length; i++) assert.ok(posted[i].seq > posted[i - 1].seq, 'seq increases'); + + // cursor from 0: all four, oldest-first + const all = await (await fetch(`${base}/api/v1/rooms/c/messages?after=0`, { headers })).json(); + assert.deepEqual(all.messages.map((m) => m.body), ['m1', 'm2', 'm3', 'm4']); + + // cursor from m2's seq: exactly m3, m4 — the same-ms skip that `since` alone would cause + const after2 = await (await fetch(`${base}/api/v1/rooms/c/messages?after=${posted[1].seq}`, { headers })).json(); + assert.deepEqual(after2.messages.map((m) => m.body), ['m3', 'm4']); + + // cursor from the last seq: caught up, empty + const afterLast = await (await fetch(`${base}/api/v1/rooms/c/messages?after=${posted[3].seq}`, { headers })).json(); + assert.equal(afterLast.count, 0); + }); +}); + +test('after-cursor honors limit and stays chronological (forward feed pages)', async () => { + await withRelay({}, async ({ base, headers }) => { + for (const n of [1, 2, 3, 4, 5]) { + await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'c', body: `m${n}` }), + }); + } + const page = await (await fetch(`${base}/api/v1/rooms/c/messages?after=0&limit=2`, { headers })).json(); + assert.deepEqual(page.messages.map((m) => m.body), ['m1', 'm2'], 'oldest-first page from the cursor'); + }); +}); + +test('after is authoritative over since (since must not re-drop same-ms messages)', async () => { + await withRelay({}, async ({ base, headers }) => { + for (const n of [1, 2, 3]) { + await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'c', body: `m${n}` }), + }); + } + // A future `since` would exclude everything on its own; with `after=0` + // present it must be ignored, so all three still come back. + const future = new Date(Date.now() + 60_000).toISOString(); + const res = await (await fetch( + `${base}/api/v1/rooms/c/messages?after=0&since=${encodeURIComponent(future)}`, { headers })).json(); + assert.deepEqual(res.messages.map((m) => m.body), ['m1', 'm2', 'm3'], 'after wins; since ignored'); + }); +}); + +test('rows written before seq existed still get a positional seq (upgrade compat)', async () => { + const storePath = join(tmpdir(), `iak-relay-seed-${randomUUID()}.jsonl`); + // Simulate a #37-era store: rows with NO seq field. + const row = (id, body) => JSON.stringify({ + id, room: 'r', from: '@x', body, created_at: '2026-01-01T00:00:00.000Z', origin: 'local-relay', + }); + writeFileSync(storePath, `${row('a', 'old1')}\n${row('b', 'old2')}\n`); + const server = startRelay({ port: 0, storePath, logger: () => {} }); + await once(server, 'listening'); + const base = `http://127.0.0.1:${server.address().port}`; + try { + // after=0 must include BOTH pre-seq rows (not silently skip them) + const feed = await (await fetch(`${base}/api/v1/rooms/r/messages?after=0`)).json(); + assert.deepEqual(feed.messages.map((m) => m.body), ['old1', 'old2']); + assert.deepEqual(feed.messages.map((m) => m.seq), [1, 2], 'positional seq assigned to legacy rows'); + // a new POST continues the sequence past the seeded rows + const posted = await (await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ room: 'r', body: 'new' }), + })).json(); + assert.equal(posted.seq, 3, 'new message continues after the seeded rows'); + } finally { + server.close(); + await once(server, 'close'); + rmSync(storePath, { force: true }); + } +});