openvibers/openvibe.events Commits
Events: pull answers carry latest_cursor (the head as an opaque cursor) (#45)
Changes
@@@ -114,7 +114,7 @@ Each delivery is `POST <endpoint>` with body `{ "event": <envelope>, "seq": n }`Operators: `GET /api/v1/deliveries?status=dead` is the dead-letter queue; `POST /api/v1/deliveries/replay { subscription_id, event_ids: [...] }` or `{ subscription_id, from_seq }` requeues retained events (that is also how a new subscription catches up on history).--Pull consumers: start with `GET /api/v1/events?topic=media.vod.*&limit=100`, then request `GET /api/v1/events?topic=media.vod.*&after=<next_cursor>&limit=100`, using the previous response's opaque `next_cursor` ([ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 7). The response carries `{ events: [{ seq, cursor, event }], next_cursor, next_after_seq, latest_seq }`, plus `gap: { from_seq, to_seq }` when the position is older than both the hot store and the replay tier (below) or from another retention epoch. Use `next_cursor` even when no events match, since it advances past scanned events. Numeric `after_seq=<position>` remains accepted for one release; when both are supplied, `after` takes precedence. The numeric `seq`, `next_after_seq` and `latest_seq` fields also remain for one release. `PUT /api/v1/checkpoints { topic, cursor }` stores a consumer's position here if it has nowhere better (an opaque cursor string is accepted in place of the integer; the answer carries its `epoch`, `carrier` and `next_cursor`).++Pull consumers: start with `GET /api/v1/events?topic=media.vod.*&limit=100`, then request `GET /api/v1/events?topic=media.vod.*&after=<next_cursor>&limit=100`, using the previous response's opaque `next_cursor` ([ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 7). The response carries `{ events: [{ seq, cursor, event }], next_cursor, next_after_seq, latest_seq, latest_cursor }` (`latest_cursor` is the head as an opaque cursor, for a consumer that starts at "now" without history), plus `gap: { from_seq, to_seq }` when the position is older than both the hot store and the replay tier (below) or from another retention epoch. Use `next_cursor` even when no events match, since it advances past scanned events. Numeric `after_seq=<position>` remains accepted for one release; when both are supplied, `after` takes precedence. The numeric `seq`, `next_after_seq` and `latest_seq` fields also remain for one release. `PUT /api/v1/checkpoints { topic, cursor }` stores a consumer's position here if it has nowhere better (an opaque cursor string is accepted in place of the integer; the answer carries its `epoch`, `carrier` and `next_cursor`).### Retention: three tiers@@@ -3,7 +3,9 @@* Pull consumers (service token with events.event.read):** GET /api/v1/events?topic=media.vod.*[,…]&after_seq=0&limit=100-- * -> { events: [{ seq, cursor, event }], next_after_seq, next_cursor, latest_seq, gap? }++ * -> { events: [{ seq, cursor, event }], next_after_seq, next_cursor, latest_seq, latest_cursor, gap? }++ * `latest_cursor` is the head as an opaque cursor: a consumer that starts at "now" (skipping history)++ * stores it, so it never needs the numeric latest_seq.* `after=` takes an opaque cursor beside after_seq (ADR-042 decision 7); a cursor from another* retention epoch answers `gap` — never a silent restart. `gap` ({ from_seq, to_seq }) means* events after the position were already pruned by retention (or the epoch changed). The pull@@@ -105,6 +107,7 @@ function readRouter({ store, auth, worker, limits }) {out.next_after_seq = scanned;out.next_cursor = cursor.encode(scanned, epoch);out.latest_seq = await store.lastSeq();++ out.latest_cursor = cursor.encode(out.latest_seq, epoch);res.json(out);});@@@ -71,6 +71,15 @@ t('pull: cursor, topic filter, internal events included for services', async ()r = await request(h.base, 'GET', `/api/v1/events?topic=live.vod.*&after_seq=${r.body.next_after_seq}`, { token: reader });assert.deepStrictEqual(r.body.events, []);assert.strictEqual(r.body.latest_seq, s3);++ assert.match(r.body.latest_cursor, /^c1\./, 'the head as an opaque cursor');++ {++ // Starting at the head with latest_cursor reads nothing old, and the next event after it arrives.++ const head = r.body.latest_cursor;++ const fromHead = await request(h.base, "GET", `/api/v1/events?topic=live.vod.*&after=${encodeURIComponent(head)}`, { token: reader });++ assert.strictEqual(fromHead.status, 200);++ assert.deepStrictEqual(fromHead.body.events, []);++ assert.strictEqual(fromHead.body.next_cursor, head);++ }const id3 = (await h.db.prepare('SELECT id FROM events WHERE seq = ?').get(s3)).id;r = await request(h.base, 'GET', `/api/v1/events/${id3}`, { token: reader });
@@ -114,7 +114,7 @@ Each delivery is `POST <endpoint>` with body `{ "event": <envelope>, "seq": n }`
@@ -114,7 +114,7 @@ Each delivery is `POST <endpoint>` with body `{ "event": <envelope>, "seq": n }`
Operators: `GET /api/v1/deliveries?status=dead` is the dead-letter queue; `POST /api/v1/deliveries/replay { subscription_id, event_ids: [...] }` or `{ subscription_id, from_seq }` requeues retained events (that is also how a new subscription catches up on history).
Operators: `GET /api/v1/deliveries?status=dead` is the dead-letter queue; `POST /api/v1/deliveries/replay { subscription_id, event_ids: [...] }` or `{ subscription_id, from_seq }` requeues retained events (that is also how a new subscription catches up on history).
-Pull consumers: start with `GET /api/v1/events?topic=media.vod.*&limit=100`, then request `GET /api/v1/events?topic=media.vod.*&after=<next_cursor>&limit=100`, using the previous response's opaque `next_cursor` ([ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 7). The response carries `{ events: [{ seq, cursor, event }], next_cursor, next_after_seq, latest_seq }`, plus `gap: { from_seq, to_seq }` when the position is older than both the hot store and the replay tier (below) or from another retention epoch. Use `next_cursor` even when no events match, since it advances past scanned events. Numeric `after_seq=<position>` remains accepted for one release; when both are supplied, `after` takes precedence. The numeric `seq`, `next_after_seq` and `latest_seq` fields also remain for one release. `PUT /api/v1/checkpoints { topic, cursor }` stores a consumer's position here if it has nowhere better (an opaque cursor string is accepted in place of the integer; the answer carries its `epoch`, `carrier` and `next_cursor`).
+Pull consumers: start with `GET /api/v1/events?topic=media.vod.*&limit=100`, then request `GET /api/v1/events?topic=media.vod.*&after=<next_cursor>&limit=100`, using the previous response's opaque `next_cursor` ([ADR-042](https://github.com/OpenVibers/OpenVibe.Contracts/blob/main/docs/adr/ADR-042-events-fabric.md) decision 7). The response carries `{ events: [{ seq, cursor, event }], next_cursor, next_after_seq, latest_seq, latest_cursor }` (`latest_cursor` is the head as an opaque cursor, for a consumer that starts at "now" without history), plus `gap: { from_seq, to_seq }` when the position is older than both the hot store and the replay tier (below) or from another retention epoch. Use `next_cursor` even when no events match, since it advances past scanned events. Numeric `after_seq=<position>` remains accepted for one release; when both are supplied, `after` takes precedence. The numeric `seq`, `next_after_seq` and `latest_seq` fields also remain for one release. `PUT /api/v1/checkpoints { topic, cursor }` stores a consumer's position here if it has nowhere better (an opaque cursor string is accepted in place of the integer; the answer carries its `epoch`, `carrier` and `next_cursor`).
### Retention: three tiers
### Retention: three tiers
@@ -3,7 +3,9 @@
@@ -3,7 +3,9 @@
* Pull consumers (service token with events.event.read):
* Pull consumers (service token with events.event.read):
*
*
* GET /api/v1/events?topic=media.vod.*[,…]&after_seq=0&limit=100
* GET /api/v1/events?topic=media.vod.*[,…]&after_seq=0&limit=100
- * -> { events: [{ seq, cursor, event }], next_after_seq, next_cursor, latest_seq, gap? }
+ * -> { events: [{ seq, cursor, event }], next_after_seq, next_cursor, latest_seq, latest_cursor, gap? }
+ * `latest_cursor` is the head as an opaque cursor: a consumer that starts at "now" (skipping history)
+ * stores it, so it never needs the numeric latest_seq.
* `after=` takes an opaque cursor beside after_seq (ADR-042 decision 7); a cursor from another
* `after=` takes an opaque cursor beside after_seq (ADR-042 decision 7); a cursor from another
* retention epoch answers `gap` — never a silent restart. `gap` ({ from_seq, to_seq }) means
* retention epoch answers `gap` — never a silent restart. `gap` ({ from_seq, to_seq }) means
* events after the position were already pruned by retention (or the epoch changed). The pull
* events after the position were already pruned by retention (or the epoch changed). The pull
@@ -105,6 +107,7 @@ function readRouter({ store, auth, worker, limits }) {
@@ -105,6 +107,7 @@ function readRouter({ store, auth, worker, limits }) {
out.next_after_seq = scanned;
out.next_after_seq = scanned;
out.next_cursor = cursor.encode(scanned, epoch);
out.next_cursor = cursor.encode(scanned, epoch);
out.latest_seq = await store.lastSeq();
out.latest_seq = await store.lastSeq();
+ out.latest_cursor = cursor.encode(out.latest_seq, epoch);
res.json(out);
res.json(out);
});
});
@@ -71,6 +71,15 @@ t('pull: cursor, topic filter, internal events included for services', async ()
@@ -71,6 +71,15 @@ t('pull: cursor, topic filter, internal events included for services', async ()
r = await request(h.base, 'GET', `/api/v1/events?topic=live.vod.*&after_seq=${r.body.next_after_seq}`, { token: reader });
r = await request(h.base, 'GET', `/api/v1/events?topic=live.vod.*&after_seq=${r.body.next_after_seq}`, { token: reader });
assert.deepStrictEqual(r.body.events, []);
assert.deepStrictEqual(r.body.events, []);
assert.strictEqual(r.body.latest_seq, s3);
assert.strictEqual(r.body.latest_seq, s3);
+ assert.match(r.body.latest_cursor, /^c1\./, 'the head as an opaque cursor');
+ {
+ // Starting at the head with latest_cursor reads nothing old, and the next event after it arrives.
+ const head = r.body.latest_cursor;
+ const fromHead = await request(h.base, "GET", `/api/v1/events?topic=live.vod.*&after=${encodeURIComponent(head)}`, { token: reader });
+ assert.strictEqual(fromHead.status, 200);
+ assert.deepStrictEqual(fromHead.body.events, []);
+ assert.strictEqual(fromHead.body.next_cursor, head);
+ }
const id3 = (await h.db.prepare('SELECT id FROM events WHERE seq = ?').get(s3)).id;
const id3 = (await h.db.prepare('SELECT id FROM events WHERE seq = ?').get(s3)).id;
r = await request(h.base, 'GET', `/api/v1/events/${id3}`, { token: reader });
r = await request(h.base, 'GET', `/api/v1/events/${id3}`, { token: reader });