openvibers/openvibe.services Commits
Services: the project archive pages Events by opaque cursor (#24)
Changes
@@@ -15,7 +15,7 @@* media/<env>/objects.jsonl every object Media holds for the project (soft-deleted ones too),* with `download`: its public URL, or a signed URL valid urlTtlS* events/<env>.jsonl the project's app events (app.<project_key>.*) Events still keeps,-- * one { seq, event } per line++ * one { seq, cursor, event } per line, read by Events' opaque cursor** Objects are listed with download URLs rather than copied into the zip: a project may hold a GiB* or more, and Services stays out of the byte path. Media signs private and sandbox URLs for at most@@@ -162,19 +162,21 @@ function createArchiver({ config, fetchImpl, log = console }) {async function eventsPart(project, env, token) {const topic = `app.${projectKey(project.id)}.*`;const events = [];-- let after = 0;-- let end = null;++ let after = null; // Events' opaque cursor (ADR-042 decision 7): none = from the oldest retained event++ let end = null; // the head when the export started: events published during it are left outlet gap = null;let pages = 0;let stopped = false;for (;;) {-- const page = await get('events', 'events', env, token, '/api/v1/events', { topic, after_seq: after, limit: EVENT_PAGE });-- if (end === null) end = Number(page.latest_seq) || 0;++ const page = await get('events', 'events', env, token, '/api/v1/events', { topic, ...(after ? { after } : {}), limit: EVENT_PAGE });++ if (end === null) end = page.latest_cursor || null;if (page.gap && !gap) gap = page.gap;-- events.push(...(page.events || []));-- const next = Number(page.next_after_seq);++ const got = page.events || [];++ events.push(...got);if (events.length > opts.maxEvents) { stopped = true; break; }-- if (!(next > after) || next >= end) break;++ const next = page.next_cursor || null;++ // A short page, a cursor that did not move, or the head the export started at: done.++ if (!next || next === after || next === end || got.length < EVENT_PAGE) break;if (++pages >= MAX_EVENT_PAGES) { stopped = true; break; }after = next;}@@@ -184,7 +186,7 @@ function createArchiver({ config, fetchImpl, log = console }) {summary: {count: events.length, complete: !stopped, limit: opts.maxEvents, topic,first_seq: events.length ? events[0].seq : null, last_seq: events.length ? events[events.length - 1].seq : null,-- ...(stopped ? { next_after_seq: events.length ? events[events.length - 1].seq : after, note: `stopped at ${opts.maxEvents} events (SERVICES_EXPORT_MAX_EVENTS); pull the rest from Events with after_seq=next_after_seq` } : {}),++ ...(stopped ? { next_cursor: events.length ? events[events.length - 1].cursor || after : after, note: `stopped at ${opts.maxEvents} events (SERVICES_EXPORT_MAX_EVENTS); pull the rest from Events with after=next_cursor` } : {}),retention: `Events keeps app events for a limited time (production 30 days, sandbox 7 by default); older ones are gone${gap ? ` (the oldest kept is seq ${gap.to_seq + 1})` : ''}.`,},};@@@ -15,7 +15,7 @@* sub app:app_<project ULID>, read-only caps, purpose export), recorded in exportTokens.* Events POST /api/v1/events: verifies the app token (openvibe-contracts), needs the capability;* GET /api/v1/events (pull, events.app.read): the token's project and env only, paged-- * like Events (next_after_seq, latest_seq); addAppEvent() seeds it++ * like Events (opaque cursors: after=, next_cursor, latest_cursor); addAppEvent() seeds it* Media POST /api/v1/:app/files: verifies the token, capability and namespace;* GET /api/v2/:project/(objects|objects/:id/download|namespaces) for app-shaped tokens* of that project, the token's env picking the tenant; addObject() seeds it@@@ -347,7 +347,7 @@ const problemJson = (json, status, code, detail) => json(status, { type: `https:/*** Events: POST /api/v1/events with an app or service token carrying events.event.publish; GET* /api/v1/events pulls an app token's own project and environment (events.app.read, topic-- * app.<project_key>.* only), with Events' paging: next_after_seq moves to latest_seq at the end.++ * app.<project_key>.* only), with Events' paging: next_cursor moves to the head (latest_cursor) at the end.*/async function startEvents(network) {const published = [];@@@ -368,10 +368,15 @@ async function startEvents(network) {if (!v.claims.cap.includes('events.app.read')) return problemJson(json, 403, 'capability.denied', 'events.app.read not granted');const key = `p${v.claims.project_id.replace(/^prj_/, '').toLowerCase()}`;if (url.searchParams.get('topic') !== `app.${key}.*`) return problemJson(json, 403, 'events.topic_not_allowed', `app.* patterns must name your project: app.${key}.*`);-- const after = Number(url.searchParams.get('after_seq') || 0);++ // Opaque cursors as Events makes them (c1.<epoch>.<base64url(seq)>); the archive never parses one.++ const enc = (n) => `c1.0.${Buffer.from(String(n), 'utf8').toString('base64url')}`;++ const given = url.searchParams.get('after');++ const after = given ? Number(Buffer.from(String(given).replace(/^c1\.0\./, ''), 'base64url').toString('utf8')) : Number(url.searchParams.get('after_seq') || 0);++ if (!Number.isFinite(after)) return problemJson(json, 400, 'events.bad_request', 'after must be an opaque cursor');const limit = Math.min(1000, Number(url.searchParams.get('limit') || 100));const rows = stored.filter((r) => r.seq > after && r.project_id === v.claims.project_id && r.env === v.claims.env).slice(0, limit);-- return json(200, { events: rows.map((r) => ({ seq: r.seq, event: r.event })), next_after_seq: rows.length === limit ? rows[rows.length - 1].seq : seq, latest_seq: seq });++ const next = rows.length === limit ? rows[rows.length - 1].seq : seq;++ return json(200, { events: rows.map((r) => ({ seq: r.seq, cursor: enc(r.seq), event: r.event })), next_after_seq: next, next_cursor: enc(next), latest_seq: seq, latest_cursor: enc(seq) });}if (req.url === '/api/v1/events' && req.method === 'POST') {const v = verify(req);@@@ -133,6 +133,8 @@ const { boot, check, done } = require('./helpers/boot');assert.deepStrictEqual(prod[5].event, E.production[5].event, 'the envelope as Events keeps it');const pulls = t.events.requests.filter((q) => q.method === 'GET' && q.path.startsWith('/api/v1/events'));assert.ok(pulls.length >= 3, 'production took more than one page');++ assert.ok(pulls.every((q) => !/[?&]after_seq=/.test(q.path)), 'paged by opaque cursor, never a numeric position');++ assert.ok(pulls.some((q) => /[?&]after=c1\./.test(q.path)), 'the second page asked after= the first page\'s next_cursor');});await check('Network minted the tokens for the owner, one per service and environment; Media and Events saw only those', async () => {@@@ -244,7 +246,9 @@ const { boot, check, done } = require('./helpers/boot');assert.strictEqual(o.next_cursor, listed[3].id, 'continue from the last one listed');assert.strictEqual(m.parts.media.sandbox.objects.complete, true, 'the sandbox has 3: under the limit');const ev = m.parts.events.production;-- assert.deepStrictEqual([ev.count, ev.complete, ev.next_after_seq], [1000, false, E.production[999].seq]);++ assert.deepStrictEqual([ev.count, ev.complete], [1000, false]);++ assert.strictEqual(ev.next_cursor, lines(z.get('events/production.jsonl'))[999].cursor, 'continue from the last event exported, by cursor');++ assert.ok(!('next_after_seq' in ev));assert.strictEqual(lines(z.get('events/production.jsonl')).length, 1000);assert.strictEqual(m.parts.events.sandbox.complete, true);assert.match(z.get('README.txt').toString(), /NOT COMPLETE/);
@@ -15,7 +15,7 @@
@@ -15,7 +15,7 @@
* media/<env>/objects.jsonl every object Media holds for the project (soft-deleted ones too),
* media/<env>/objects.jsonl every object Media holds for the project (soft-deleted ones too),
* with `download`: its public URL, or a signed URL valid urlTtlS
* with `download`: its public URL, or a signed URL valid urlTtlS
* events/<env>.jsonl the project's app events (app.<project_key>.*) Events still keeps,
* events/<env>.jsonl the project's app events (app.<project_key>.*) Events still keeps,
- * one { seq, event } per line
+ * one { seq, cursor, event } per line, read by Events' opaque cursor
*
*
* Objects are listed with download URLs rather than copied into the zip: a project may hold a GiB
* Objects are listed with download URLs rather than copied into the zip: a project may hold a GiB
* or more, and Services stays out of the byte path. Media signs private and sandbox URLs for at most
* or more, and Services stays out of the byte path. Media signs private and sandbox URLs for at most
@@ -162,19 +162,21 @@ function createArchiver({ config, fetchImpl, log = console }) {
@@ -162,19 +162,21 @@ function createArchiver({ config, fetchImpl, log = console }) {
async function eventsPart(project, env, token) {
async function eventsPart(project, env, token) {
const topic = `app.${projectKey(project.id)}.*`;
const topic = `app.${projectKey(project.id)}.*`;
const events = [];
const events = [];
- let after = 0;
- let end = null;
+ let after = null; // Events' opaque cursor (ADR-042 decision 7): none = from the oldest retained event
+ let end = null; // the head when the export started: events published during it are left out
let gap = null;
let gap = null;
let pages = 0;
let pages = 0;
let stopped = false;
let stopped = false;
for (;;) {
for (;;) {
- const page = await get('events', 'events', env, token, '/api/v1/events', { topic, after_seq: after, limit: EVENT_PAGE });
- if (end === null) end = Number(page.latest_seq) || 0;
+ const page = await get('events', 'events', env, token, '/api/v1/events', { topic, ...(after ? { after } : {}), limit: EVENT_PAGE });
+ if (end === null) end = page.latest_cursor || null;
if (page.gap && !gap) gap = page.gap;
if (page.gap && !gap) gap = page.gap;
- events.push(...(page.events || []));
- const next = Number(page.next_after_seq);
+ const got = page.events || [];
+ events.push(...got);
if (events.length > opts.maxEvents) { stopped = true; break; }
if (events.length > opts.maxEvents) { stopped = true; break; }
- if (!(next > after) || next >= end) break;
+ const next = page.next_cursor || null;
+ // A short page, a cursor that did not move, or the head the export started at: done.
+ if (!next || next === after || next === end || got.length < EVENT_PAGE) break;
if (++pages >= MAX_EVENT_PAGES) { stopped = true; break; }
if (++pages >= MAX_EVENT_PAGES) { stopped = true; break; }
after = next;
after = next;
}
}
@@ -184,7 +186,7 @@ function createArchiver({ config, fetchImpl, log = console }) {
@@ -184,7 +186,7 @@ function createArchiver({ config, fetchImpl, log = console }) {
summary: {
summary: {
count: events.length, complete: !stopped, limit: opts.maxEvents, topic,
count: events.length, complete: !stopped, limit: opts.maxEvents, topic,
first_seq: events.length ? events[0].seq : null, last_seq: events.length ? events[events.length - 1].seq : null,
first_seq: events.length ? events[0].seq : null, last_seq: events.length ? events[events.length - 1].seq : null,
- ...(stopped ? { next_after_seq: events.length ? events[events.length - 1].seq : after, note: `stopped at ${opts.maxEvents} events (SERVICES_EXPORT_MAX_EVENTS); pull the rest from Events with after_seq=next_after_seq` } : {}),
+ ...(stopped ? { next_cursor: events.length ? events[events.length - 1].cursor || after : after, note: `stopped at ${opts.maxEvents} events (SERVICES_EXPORT_MAX_EVENTS); pull the rest from Events with after=next_cursor` } : {}),
retention: `Events keeps app events for a limited time (production 30 days, sandbox 7 by default); older ones are gone${gap ? ` (the oldest kept is seq ${gap.to_seq + 1})` : ''}.`,
retention: `Events keeps app events for a limited time (production 30 days, sandbox 7 by default); older ones are gone${gap ? ` (the oldest kept is seq ${gap.to_seq + 1})` : ''}.`,
},
},
};
};
@@ -15,7 +15,7 @@
@@ -15,7 +15,7 @@
* sub app:app_<project ULID>, read-only caps, purpose export), recorded in exportTokens.
* sub app:app_<project ULID>, read-only caps, purpose export), recorded in exportTokens.
* Events POST /api/v1/events: verifies the app token (openvibe-contracts), needs the capability;
* Events POST /api/v1/events: verifies the app token (openvibe-contracts), needs the capability;
* GET /api/v1/events (pull, events.app.read): the token's project and env only, paged
* GET /api/v1/events (pull, events.app.read): the token's project and env only, paged
- * like Events (next_after_seq, latest_seq); addAppEvent() seeds it
+ * like Events (opaque cursors: after=, next_cursor, latest_cursor); addAppEvent() seeds it
* Media POST /api/v1/:app/files: verifies the token, capability and namespace;
* Media POST /api/v1/:app/files: verifies the token, capability and namespace;
* GET /api/v2/:project/(objects|objects/:id/download|namespaces) for app-shaped tokens
* GET /api/v2/:project/(objects|objects/:id/download|namespaces) for app-shaped tokens
* of that project, the token's env picking the tenant; addObject() seeds it
* of that project, the token's env picking the tenant; addObject() seeds it
@@ -347,7 +347,7 @@ const problemJson = (json, status, code, detail) => json(status, { type: `https:
@@ -347,7 +347,7 @@ const problemJson = (json, status, code, detail) => json(status, { type: `https:
/**
/**
* Events: POST /api/v1/events with an app or service token carrying events.event.publish; GET
* Events: POST /api/v1/events with an app or service token carrying events.event.publish; GET
* /api/v1/events pulls an app token's own project and environment (events.app.read, topic
* /api/v1/events pulls an app token's own project and environment (events.app.read, topic
- * app.<project_key>.* only), with Events' paging: next_after_seq moves to latest_seq at the end.
+ * app.<project_key>.* only), with Events' paging: next_cursor moves to the head (latest_cursor) at the end.
*/
*/
async function startEvents(network) {
async function startEvents(network) {
const published = [];
const published = [];
@@ -368,10 +368,15 @@ async function startEvents(network) {
@@ -368,10 +368,15 @@ async function startEvents(network) {
if (!v.claims.cap.includes('events.app.read')) return problemJson(json, 403, 'capability.denied', 'events.app.read not granted');
if (!v.claims.cap.includes('events.app.read')) return problemJson(json, 403, 'capability.denied', 'events.app.read not granted');
const key = `p${v.claims.project_id.replace(/^prj_/, '').toLowerCase()}`;
const key = `p${v.claims.project_id.replace(/^prj_/, '').toLowerCase()}`;
if (url.searchParams.get('topic') !== `app.${key}.*`) return problemJson(json, 403, 'events.topic_not_allowed', `app.* patterns must name your project: app.${key}.*`);
if (url.searchParams.get('topic') !== `app.${key}.*`) return problemJson(json, 403, 'events.topic_not_allowed', `app.* patterns must name your project: app.${key}.*`);
- const after = Number(url.searchParams.get('after_seq') || 0);
+ // Opaque cursors as Events makes them (c1.<epoch>.<base64url(seq)>); the archive never parses one.
+ const enc = (n) => `c1.0.${Buffer.from(String(n), 'utf8').toString('base64url')}`;
+ const given = url.searchParams.get('after');
+ const after = given ? Number(Buffer.from(String(given).replace(/^c1\.0\./, ''), 'base64url').toString('utf8')) : Number(url.searchParams.get('after_seq') || 0);
+ if (!Number.isFinite(after)) return problemJson(json, 400, 'events.bad_request', 'after must be an opaque cursor');
const limit = Math.min(1000, Number(url.searchParams.get('limit') || 100));
const limit = Math.min(1000, Number(url.searchParams.get('limit') || 100));
const rows = stored.filter((r) => r.seq > after && r.project_id === v.claims.project_id && r.env === v.claims.env).slice(0, limit);
const rows = stored.filter((r) => r.seq > after && r.project_id === v.claims.project_id && r.env === v.claims.env).slice(0, limit);
- return json(200, { events: rows.map((r) => ({ seq: r.seq, event: r.event })), next_after_seq: rows.length === limit ? rows[rows.length - 1].seq : seq, latest_seq: seq });
+ const next = rows.length === limit ? rows[rows.length - 1].seq : seq;
+ return json(200, { events: rows.map((r) => ({ seq: r.seq, cursor: enc(r.seq), event: r.event })), next_after_seq: next, next_cursor: enc(next), latest_seq: seq, latest_cursor: enc(seq) });
}
}
if (req.url === '/api/v1/events' && req.method === 'POST') {
if (req.url === '/api/v1/events' && req.method === 'POST') {
const v = verify(req);
const v = verify(req);
@@ -133,6 +133,8 @@ const { boot, check, done } = require('./helpers/boot');
@@ -133,6 +133,8 @@ const { boot, check, done } = require('./helpers/boot');
assert.deepStrictEqual(prod[5].event, E.production[5].event, 'the envelope as Events keeps it');
assert.deepStrictEqual(prod[5].event, E.production[5].event, 'the envelope as Events keeps it');
const pulls = t.events.requests.filter((q) => q.method === 'GET' && q.path.startsWith('/api/v1/events'));
const pulls = t.events.requests.filter((q) => q.method === 'GET' && q.path.startsWith('/api/v1/events'));
assert.ok(pulls.length >= 3, 'production took more than one page');
assert.ok(pulls.length >= 3, 'production took more than one page');
+ assert.ok(pulls.every((q) => !/[?&]after_seq=/.test(q.path)), 'paged by opaque cursor, never a numeric position');
+ assert.ok(pulls.some((q) => /[?&]after=c1\./.test(q.path)), 'the second page asked after= the first page\'s next_cursor');
});
});
await check('Network minted the tokens for the owner, one per service and environment; Media and Events saw only those', async () => {
await check('Network minted the tokens for the owner, one per service and environment; Media and Events saw only those', async () => {
@@ -244,7 +246,9 @@ const { boot, check, done } = require('./helpers/boot');
@@ -244,7 +246,9 @@ const { boot, check, done } = require('./helpers/boot');
assert.strictEqual(o.next_cursor, listed[3].id, 'continue from the last one listed');
assert.strictEqual(o.next_cursor, listed[3].id, 'continue from the last one listed');
assert.strictEqual(m.parts.media.sandbox.objects.complete, true, 'the sandbox has 3: under the limit');
assert.strictEqual(m.parts.media.sandbox.objects.complete, true, 'the sandbox has 3: under the limit');
const ev = m.parts.events.production;
const ev = m.parts.events.production;
- assert.deepStrictEqual([ev.count, ev.complete, ev.next_after_seq], [1000, false, E.production[999].seq]);
+ assert.deepStrictEqual([ev.count, ev.complete], [1000, false]);
+ assert.strictEqual(ev.next_cursor, lines(z.get('events/production.jsonl'))[999].cursor, 'continue from the last event exported, by cursor');
+ assert.ok(!('next_after_seq' in ev));
assert.strictEqual(lines(z.get('events/production.jsonl')).length, 1000);
assert.strictEqual(lines(z.get('events/production.jsonl')).length, 1000);
assert.strictEqual(m.parts.events.sandbox.complete, true);
assert.strictEqual(m.parts.events.sandbox.complete, true);
assert.match(z.get('README.txt').toString(), /NOT COMPLETE/);
assert.match(z.get('README.txt').toString(), /NOT COMPLETE/);