openvibers/openvibe.connect Commits
Waiting webhooks go the moment the computer connects, under one event id (CON-15) (#17)
Changes
@@@ -50,7 +50,8 @@ misses nothing (CON-15, ADR-057 decision 7):- Connect keeps the method, the provider's headers (`content-type`, `user-agent`, `x-…`, `stripe-signature`,`svix-…`, `webhook-…`; never cookies, authorization or forwarding headers) and the body. It publishes only apointer, `connect.hook.received`, to OpenVibe.Events, in the same transaction.--- Events delivers that pointer back to Connect's own subscription, retried for about seven days. Connect hands the++- The moment the computer connects, Connect hands it every webhook that waited, oldest first. Events also delivers the++ pointer back to Connect's own subscription, retried for about seven days, which finds it settled. Connect hands thewebhook to your app with `x-openvibe-hook` and `x-openvibe-event-id` (dedupe on it). Your app's answer below 500 settles it, andits body is dropped. While the computer is offline, or the app answers 5xx, Connect says 503 and Events tries again.- Nobody signs in to send a webhook, so check the provider's own signature in your app.nnew file mode 100644@@@ -0,0 +1,4 @@++-- phase: expand++-- Each kept webhook knows its event id (CON-15), so the app sees the same x-openvibe-event-id whether Connect hands it++-- over the moment the computer connects or Events' retry brings it: one id to dedupe on.++ALTER TABLE hook_requests ADD COLUMN IF NOT EXISTS event_id text;@@@ -70,6 +70,9 @@ async function createApp(opts = {}) {const nodeAuth = createNodeAuth({ config, keys, now: s.now });const machines = opts.machines || createMachines({ config, fetchImpl, log, now: s.now });const tunnels = createTunnels({ shares, config, log, nodeAuth, now: s.now, onConnect: async (share) => {++ // Webhooks that waited for this computer (CON-15) go at once, oldest first; Events' later retry finds them++ // settled. One that fails waits for that retry.++ ctx.deliverWaiting(share).catch((err) => log.warn('[OpenVibe.Connect] waiting webhooks not handed over:', err.message));const waiting = await shares.takeWaiters(share.id);for (const subject of waiting) await notifier.shareBack(subject, share);if (waiting.length) log.log(`[OpenVibe.Connect] told ${waiting.length} person(s) ${share.host} is back`);@@@ -94,6 +97,18 @@ async function createApp(opts = {}) {const ctx = { config, s, keys, sso, principal, shares, tunnels, gate, proxy, notifier, nodeAuth, machines, hooks, outbox, accountData, accountSend, log };++ /** A share's computer is connected: its waiting webhooks go to its app, one at a time, oldest first. → delivered count */++ ctx.deliverWaiting = async (share) => {++ let n = 0;++ for (const q of await hooks.waiting(share.id)) {++ const out = await proxy.deliverHook(share, { method: q.method, path: q.path, headers: q.headers, body: q.body, hookId: q.hookId, eventId: q.eventId || q.ref });++ if (out.offline) break;++ if (out.status && out.status < 500 && await hooks.settle(q.ref, out.status)) n++;++ }++ if (n) log.log(`[OpenVibe.Connect] ${n} waiting webhook(s) handed to ${share.host}`);++ return n;++ };++/** Shares whose time is up close themselves (CON-6): their computers are disconnected and their lists forgotten. */ctx.expireSweep = async () => {let due = [];@@@ -11,8 +11,10 @@* create(owner, share, { path, label }) → hook list(shareId) → hooks remove(owner, id)* get(id) receive(hook, { method, headers, body }) → { ref }* request(ref) → the stored request with its hook settle(ref, status) sweep() → rows removed++ * waiting(shareId) → its undelivered requests, oldest first (handed over at once when its computer connects)*/const crypto = require('crypto');++const { ids } = require('openvibe-contracts');const HOOK_ID = /^hk_[0-9A-HJKMNP-TV-Z]{26}$/;const REF_RE = /^hb_[A-Za-z0-9_-]{16,64}$/;@@@ -96,13 +98,16 @@ function createHooks(s, { config, outbox = null, now = () => Date.now() }) {const contentType = (kept.find(([k]) => k === 'content-type') || [null, null])[1];const ref = `hb_${crypto.randomBytes(18).toString('base64url')}`;const at = new Date(now());++ // The event's id is chosen here and kept with the request: the app sees the same x-openvibe-event-id however++ // the webhook reaches it (handed over on connect, or Events' retry).++ const eventId = ids.newId('event');await s.tx(async (t) => {-- await t.exec('INSERT INTO hook_requests (id, hook_id, method, headers, body, received_at, expires_at) VALUES ($1, $2, $3, $4::jsonb, $5, $6, $7)',-- [ref, hook.id, method, JSON.stringify(kept), buf, at.toISOString(), new Date(at.getTime() + KEEP_MS).toISOString()]);++ await t.exec('INSERT INTO hook_requests (id, hook_id, method, headers, body, received_at, expires_at, event_id) VALUES ($1, $2, $3, $4::jsonb, $5, $6, $7, $8)',++ [ref, hook.id, method, JSON.stringify(kept), buf, at.toISOString(), new Date(at.getTime() + KEEP_MS).toISOString(), eventId]);await t.exec('UPDATE share_hooks SET received = received + 1, last_received_at = $2 WHERE id = $1', [hook.id, at.toISOString()]);if (outbox) {await outbox.emitIn(t, {-- event_type: 'connect.hook.received', version: 1, source: 'connect',++ event_id: eventId, event_type: 'connect.hook.received', version: 1, source: 'connect',actor: { type: 'service', id: 'connect' }, subject: { type: 'connect.hook', id: hook.id },visibility: 'internal', priority: 'important',payload: { hook_id: hook.id, share_id: hook.share_id, body_ref: ref, method, size: buf.length, content_type: contentType ? String(contentType).slice(0, 200) : null, received_at: at.toISOString() },@@@ -119,7 +124,17 @@ function createHooks(s, { config, outbox = null, now = () => Date.now() }) {const r = await db.maybe(`SELECT r.*, h.share_id, h.local_path, h.status AS hook_status FROM hook_requests r JOIN share_hooks h ON h.id = r.hook_idWHERE r.id = $1`, [ref]);if (!r || r.hook_status !== 'active' || r.delivered_at || new Date(r.expires_at).getTime() <= now()) return null;-- return { ref: r.id, hookId: r.hook_id, shareId: r.share_id, path: r.local_path, method: r.method, headers: Array.isArray(r.headers) ? r.headers : [], body: r.body };++ return { ref: r.id, hookId: r.hook_id, shareId: r.share_id, path: r.local_path, method: r.method, headers: Array.isArray(r.headers) ? r.headers : [], body: r.body, eventId: r.event_id || null };++ }++++ /** A share's webhooks still waiting for its computer, oldest first (at most `limit`). */++ async function waiting(shareId, limit = 100) {++ const rows = await db.many(`SELECT r.id FROM hook_requests r JOIN share_hooks h ON h.id = r.hook_id++ WHERE h.share_id = $1 AND h.status = 'active' AND r.delivered_at IS NULL AND r.expires_at > $2++ ORDER BY r.received_at LIMIT $3`, [shareId, new Date(now()).toISOString(), limit]);++ const out = [];++ for (const r of rows) { const q = await request(r.id); if (q) out.push(q); }++ return out;}/** The computer's app answered: the request is delivered (its body is dropped) and the hook counts it. */@@@ -136,7 +151,7 @@ function createHooks(s, { config, outbox = null, now = () => Date.now() }) {return r.length;}-- return { get, list, create, remove, receive, request, settle, sweep, KEEP_MS, MAX_BODY };++ return { get, list, create, remove, receive, request, waiting, settle, sweep, KEEP_MS, MAX_BODY };}module.exports = { createHooks, HookError, keepHeaders, HOOK_ID, METHODS, MAX_BODY, KEEP_MS };@@@ -73,15 +73,13 @@ const SECRET = 'hook-delivery-secret-0123456789abcdef';assert.strictEqual(big.status, 413);});-- await check('Events\' delivery: 503 while the computer is offline, then handed to the app at the hook\'s path, once', async () => {++ await check('Events\' delivery is 503 while the computer is offline; connecting hands it to the app at the hook\'s path, once', async () => {const off = await deliver(envelope);assert.strictEqual(off.status, 503, off.text);++ const before = app.seen.length;const c = t.client({ key: made.key, port: app.port });await c.ready();-- const before = app.seen.length;-- const r = await deliver(envelope);-- assert.strictEqual(r.status, 200, r.text);-- assert.deepStrictEqual(r.json(), { delivered: true, status: 200 });++ for (let i = 0; i < 100 && !app.seen.slice(before).some((x) => x.url === '/webhooks/github'); i++) await new Promise((r) => setTimeout(r, 20));const got = app.seen.slice(before).find((x) => x.url === '/webhooks/github');assert.ok(got, 'the app got it');assert.strictEqual(got.method, 'POST');@@@ -89,7 +87,7 @@ const SECRET = 'hook-delivery-secret-0123456789abcdef';assert.strictEqual(got.headers['x-github-event'], 'push');assert.strictEqual(got.headers['x-hub-signature-256'], 'sha256=abc');assert.strictEqual(got.headers['x-openvibe-hook'], hook.id);-- assert.strictEqual(got.headers['x-openvibe-event-id'], envelope.event_id);++ assert.strictEqual(got.headers['x-openvibe-event-id'], envelope.event_id, 'the event\'s own id, whichever way it came');assert.strictEqual(got.headers.cookie, undefined);const again = await deliver(envelope);assert.strictEqual(again.status, 200);@@@ -126,6 +124,39 @@ const SECRET = 'hook-delivery-secret-0123456789abcdef';assert.strictEqual((await t.ctx.s.db.maybe('SELECT id FROM hook_requests WHERE id = $1', [ref])), null);});++ await check('a webhook that waited goes to the app as soon as its computer connects; Events\' retry then finds it settled', async () => {++ const h3 = (await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: '/webhooks/waited' } })).json().hook;++ const w = await t.get(`/h/${h3.id}`, { method: 'POST', body: '{"waited":true}', headers: { 'content-type': 'application/json' } });++ assert.strictEqual(w.status, 202);++ const env = (await t.ctx.s.db.one("SELECT envelope FROM service_outbox WHERE envelope->'payload'->>'body_ref' = $1", [w.json().id])).envelope;++ const before = app.seen.length;++ const c = t.client({ key: made.key, port: app.port });++ await c.ready();++ for (let i = 0; i < 100 && !app.seen.slice(before).some((x) => x.url === '/webhooks/waited'); i++) await new Promise((r) => setTimeout(r, 20));++ const got = app.seen.slice(before).find((x) => x.url === '/webhooks/waited');++ assert.ok(got, 'handed over on connect');++ assert.strictEqual(got.body, '{"waited":true}');++ assert.strictEqual(got.headers['x-openvibe-event-id'], env.event_id, 'the event\'s own id');++ const late = await deliver(env);++ assert.deepStrictEqual(late.json(), { gone: true }, 'Events\' delivery finds it settled');++ assert.strictEqual(app.seen.filter((x) => x.url === '/webhooks/waited').length, 1, 'once');++ c.stop();++ });++++ await check('with the computer connected, Events\' delivery hands the webhook over itself', async () => {++ const h4 = (await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: '/webhooks/live' } })).json().hook;++ const c = t.client({ key: made.key, port: app.port });++ await c.ready();++ const w = await t.get(`/h/${h4.id}`, { method: 'POST', body: '{"live":1}', headers: { 'content-type': 'application/json' } });++ const env = (await t.ctx.s.db.one("SELECT envelope FROM service_outbox WHERE envelope->'payload'->>'body_ref' = $1", [w.json().id])).envelope;++ const r = await deliver(env);++ assert.deepStrictEqual(r.json(), { delivered: true, status: 200 });++ const got = app.seen.filter((x) => x.url === '/webhooks/live');++ assert.strictEqual(got.length, 1);++ assert.strictEqual(got[0].headers['x-openvibe-event-id'], env.event_id);++ c.stop();++ });++await check('the page: make one with a plain form', async () => {const r = await t.get(`/shares/${made.share.id}/hooks`, { as: alex, method: 'POST', form: { path: '/hooks/stripe', label: 'Stripe' } });assert.strictEqual(r.status, 201, r.text);
@@ -50,7 +50,8 @@ misses nothing (CON-15, ADR-057 decision 7):
@@ -50,7 +50,8 @@ misses nothing (CON-15, ADR-057 decision 7):
- Connect keeps the method, the provider's headers (`content-type`, `user-agent`, `x-…`, `stripe-signature`,
- Connect keeps the method, the provider's headers (`content-type`, `user-agent`, `x-…`, `stripe-signature`,
`svix-…`, `webhook-…`; never cookies, authorization or forwarding headers) and the body. It publishes only a
`svix-…`, `webhook-…`; never cookies, authorization or forwarding headers) and the body. It publishes only a
pointer, `connect.hook.received`, to OpenVibe.Events, in the same transaction.
pointer, `connect.hook.received`, to OpenVibe.Events, in the same transaction.
-- Events delivers that pointer back to Connect's own subscription, retried for about seven days. Connect hands the
+- The moment the computer connects, Connect hands it every webhook that waited, oldest first. Events also delivers the
+ pointer back to Connect's own subscription, retried for about seven days, which finds it settled. Connect hands the
webhook to your app with `x-openvibe-hook` and `x-openvibe-event-id` (dedupe on it). Your app's answer below 500 settles it, and
webhook to your app with `x-openvibe-hook` and `x-openvibe-event-id` (dedupe on it). Your app's answer below 500 settles it, and
its body is dropped. While the computer is offline, or the app answers 5xx, Connect says 503 and Events tries again.
its body is dropped. While the computer is offline, or the app answers 5xx, Connect says 503 and Events tries again.
- Nobody signs in to send a webhook, so check the provider's own signature in your app.
- Nobody signs in to send a webhook, so check the provider's own signature in your app.
new file mode 100644
new file mode 100644
@@ -0,0 +1,4 @@
@@ -0,0 +1,4 @@
+-- phase: expand
+-- Each kept webhook knows its event id (CON-15), so the app sees the same x-openvibe-event-id whether Connect hands it
+-- over the moment the computer connects or Events' retry brings it: one id to dedupe on.
+ALTER TABLE hook_requests ADD COLUMN IF NOT EXISTS event_id text;
@@ -70,6 +70,9 @@ async function createApp(opts = {}) {
@@ -70,6 +70,9 @@ async function createApp(opts = {}) {
const nodeAuth = createNodeAuth({ config, keys, now: s.now });
const nodeAuth = createNodeAuth({ config, keys, now: s.now });
const machines = opts.machines || createMachines({ config, fetchImpl, log, now: s.now });
const machines = opts.machines || createMachines({ config, fetchImpl, log, now: s.now });
const tunnels = createTunnels({ shares, config, log, nodeAuth, now: s.now, onConnect: async (share) => {
const tunnels = createTunnels({ shares, config, log, nodeAuth, now: s.now, onConnect: async (share) => {
+ // Webhooks that waited for this computer (CON-15) go at once, oldest first; Events' later retry finds them
+ // settled. One that fails waits for that retry.
+ ctx.deliverWaiting(share).catch((err) => log.warn('[OpenVibe.Connect] waiting webhooks not handed over:', err.message));
const waiting = await shares.takeWaiters(share.id);
const waiting = await shares.takeWaiters(share.id);
for (const subject of waiting) await notifier.shareBack(subject, share);
for (const subject of waiting) await notifier.shareBack(subject, share);
if (waiting.length) log.log(`[OpenVibe.Connect] told ${waiting.length} person(s) ${share.host} is back`);
if (waiting.length) log.log(`[OpenVibe.Connect] told ${waiting.length} person(s) ${share.host} is back`);
@@ -94,6 +97,18 @@ async function createApp(opts = {}) {
@@ -94,6 +97,18 @@ async function createApp(opts = {}) {
const ctx = { config, s, keys, sso, principal, shares, tunnels, gate, proxy, notifier, nodeAuth, machines, hooks, outbox, accountData, accountSend, log };
const ctx = { config, s, keys, sso, principal, shares, tunnels, gate, proxy, notifier, nodeAuth, machines, hooks, outbox, accountData, accountSend, log };
+ /** A share's computer is connected: its waiting webhooks go to its app, one at a time, oldest first. → delivered count */
+ ctx.deliverWaiting = async (share) => {
+ let n = 0;
+ for (const q of await hooks.waiting(share.id)) {
+ const out = await proxy.deliverHook(share, { method: q.method, path: q.path, headers: q.headers, body: q.body, hookId: q.hookId, eventId: q.eventId || q.ref });
+ if (out.offline) break;
+ if (out.status && out.status < 500 && await hooks.settle(q.ref, out.status)) n++;
+ }
+ if (n) log.log(`[OpenVibe.Connect] ${n} waiting webhook(s) handed to ${share.host}`);
+ return n;
+ };
+
/** Shares whose time is up close themselves (CON-6): their computers are disconnected and their lists forgotten. */
/** Shares whose time is up close themselves (CON-6): their computers are disconnected and their lists forgotten. */
ctx.expireSweep = async () => {
ctx.expireSweep = async () => {
let due = [];
let due = [];
@@ -11,8 +11,10 @@
@@ -11,8 +11,10 @@
* create(owner, share, { path, label }) → hook list(shareId) → hooks remove(owner, id)
* create(owner, share, { path, label }) → hook list(shareId) → hooks remove(owner, id)
* get(id) receive(hook, { method, headers, body }) → { ref }
* get(id) receive(hook, { method, headers, body }) → { ref }
* request(ref) → the stored request with its hook settle(ref, status) sweep() → rows removed
* request(ref) → the stored request with its hook settle(ref, status) sweep() → rows removed
+ * waiting(shareId) → its undelivered requests, oldest first (handed over at once when its computer connects)
*/
*/
const crypto = require('crypto');
const crypto = require('crypto');
+const { ids } = require('openvibe-contracts');
const HOOK_ID = /^hk_[0-9A-HJKMNP-TV-Z]{26}$/;
const HOOK_ID = /^hk_[0-9A-HJKMNP-TV-Z]{26}$/;
const REF_RE = /^hb_[A-Za-z0-9_-]{16,64}$/;
const REF_RE = /^hb_[A-Za-z0-9_-]{16,64}$/;
@@ -96,13 +98,16 @@ function createHooks(s, { config, outbox = null, now = () => Date.now() }) {
@@ -96,13 +98,16 @@ function createHooks(s, { config, outbox = null, now = () => Date.now() }) {
const contentType = (kept.find(([k]) => k === 'content-type') || [null, null])[1];
const contentType = (kept.find(([k]) => k === 'content-type') || [null, null])[1];
const ref = `hb_${crypto.randomBytes(18).toString('base64url')}`;
const ref = `hb_${crypto.randomBytes(18).toString('base64url')}`;
const at = new Date(now());
const at = new Date(now());
+ // The event's id is chosen here and kept with the request: the app sees the same x-openvibe-event-id however
+ // the webhook reaches it (handed over on connect, or Events' retry).
+ const eventId = ids.newId('event');
await s.tx(async (t) => {
await s.tx(async (t) => {
- await t.exec('INSERT INTO hook_requests (id, hook_id, method, headers, body, received_at, expires_at) VALUES ($1, $2, $3, $4::jsonb, $5, $6, $7)',
- [ref, hook.id, method, JSON.stringify(kept), buf, at.toISOString(), new Date(at.getTime() + KEEP_MS).toISOString()]);
+ await t.exec('INSERT INTO hook_requests (id, hook_id, method, headers, body, received_at, expires_at, event_id) VALUES ($1, $2, $3, $4::jsonb, $5, $6, $7, $8)',
+ [ref, hook.id, method, JSON.stringify(kept), buf, at.toISOString(), new Date(at.getTime() + KEEP_MS).toISOString(), eventId]);
await t.exec('UPDATE share_hooks SET received = received + 1, last_received_at = $2 WHERE id = $1', [hook.id, at.toISOString()]);
await t.exec('UPDATE share_hooks SET received = received + 1, last_received_at = $2 WHERE id = $1', [hook.id, at.toISOString()]);
if (outbox) {
if (outbox) {
await outbox.emitIn(t, {
await outbox.emitIn(t, {
- event_type: 'connect.hook.received', version: 1, source: 'connect',
+ event_id: eventId, event_type: 'connect.hook.received', version: 1, source: 'connect',
actor: { type: 'service', id: 'connect' }, subject: { type: 'connect.hook', id: hook.id },
actor: { type: 'service', id: 'connect' }, subject: { type: 'connect.hook', id: hook.id },
visibility: 'internal', priority: 'important',
visibility: 'internal', priority: 'important',
payload: { hook_id: hook.id, share_id: hook.share_id, body_ref: ref, method, size: buf.length, content_type: contentType ? String(contentType).slice(0, 200) : null, received_at: at.toISOString() },
payload: { hook_id: hook.id, share_id: hook.share_id, body_ref: ref, method, size: buf.length, content_type: contentType ? String(contentType).slice(0, 200) : null, received_at: at.toISOString() },
@@ -119,7 +124,17 @@ function createHooks(s, { config, outbox = null, now = () => Date.now() }) {
@@ -119,7 +124,17 @@ function createHooks(s, { config, outbox = null, now = () => Date.now() }) {
const r = await db.maybe(`SELECT r.*, h.share_id, h.local_path, h.status AS hook_status FROM hook_requests r JOIN share_hooks h ON h.id = r.hook_id
const r = await db.maybe(`SELECT r.*, h.share_id, h.local_path, h.status AS hook_status FROM hook_requests r JOIN share_hooks h ON h.id = r.hook_id
WHERE r.id = $1`, [ref]);
WHERE r.id = $1`, [ref]);
if (!r || r.hook_status !== 'active' || r.delivered_at || new Date(r.expires_at).getTime() <= now()) return null;
if (!r || r.hook_status !== 'active' || r.delivered_at || new Date(r.expires_at).getTime() <= now()) return null;
- return { ref: r.id, hookId: r.hook_id, shareId: r.share_id, path: r.local_path, method: r.method, headers: Array.isArray(r.headers) ? r.headers : [], body: r.body };
+ return { ref: r.id, hookId: r.hook_id, shareId: r.share_id, path: r.local_path, method: r.method, headers: Array.isArray(r.headers) ? r.headers : [], body: r.body, eventId: r.event_id || null };
+ }
+
+ /** A share's webhooks still waiting for its computer, oldest first (at most `limit`). */
+ async function waiting(shareId, limit = 100) {
+ const rows = await db.many(`SELECT r.id FROM hook_requests r JOIN share_hooks h ON h.id = r.hook_id
+ WHERE h.share_id = $1 AND h.status = 'active' AND r.delivered_at IS NULL AND r.expires_at > $2
+ ORDER BY r.received_at LIMIT $3`, [shareId, new Date(now()).toISOString(), limit]);
+ const out = [];
+ for (const r of rows) { const q = await request(r.id); if (q) out.push(q); }
+ return out;
}
}
/** The computer's app answered: the request is delivered (its body is dropped) and the hook counts it. */
/** The computer's app answered: the request is delivered (its body is dropped) and the hook counts it. */
@@ -136,7 +151,7 @@ function createHooks(s, { config, outbox = null, now = () => Date.now() }) {
@@ -136,7 +151,7 @@ function createHooks(s, { config, outbox = null, now = () => Date.now() }) {
return r.length;
return r.length;
}
}
- return { get, list, create, remove, receive, request, settle, sweep, KEEP_MS, MAX_BODY };
+ return { get, list, create, remove, receive, request, waiting, settle, sweep, KEEP_MS, MAX_BODY };
}
}
module.exports = { createHooks, HookError, keepHeaders, HOOK_ID, METHODS, MAX_BODY, KEEP_MS };
module.exports = { createHooks, HookError, keepHeaders, HOOK_ID, METHODS, MAX_BODY, KEEP_MS };
@@ -73,15 +73,13 @@ const SECRET = 'hook-delivery-secret-0123456789abcdef';
@@ -73,15 +73,13 @@ const SECRET = 'hook-delivery-secret-0123456789abcdef';
assert.strictEqual(big.status, 413);
assert.strictEqual(big.status, 413);
});
});
- await check('Events\' delivery: 503 while the computer is offline, then handed to the app at the hook\'s path, once', async () => {
+ await check('Events\' delivery is 503 while the computer is offline; connecting hands it to the app at the hook\'s path, once', async () => {
const off = await deliver(envelope);
const off = await deliver(envelope);
assert.strictEqual(off.status, 503, off.text);
assert.strictEqual(off.status, 503, off.text);
+ const before = app.seen.length;
const c = t.client({ key: made.key, port: app.port });
const c = t.client({ key: made.key, port: app.port });
await c.ready();
await c.ready();
- const before = app.seen.length;
- const r = await deliver(envelope);
- assert.strictEqual(r.status, 200, r.text);
- assert.deepStrictEqual(r.json(), { delivered: true, status: 200 });
+ for (let i = 0; i < 100 && !app.seen.slice(before).some((x) => x.url === '/webhooks/github'); i++) await new Promise((r) => setTimeout(r, 20));
const got = app.seen.slice(before).find((x) => x.url === '/webhooks/github');
const got = app.seen.slice(before).find((x) => x.url === '/webhooks/github');
assert.ok(got, 'the app got it');
assert.ok(got, 'the app got it');
assert.strictEqual(got.method, 'POST');
assert.strictEqual(got.method, 'POST');
@@ -89,7 +87,7 @@ const SECRET = 'hook-delivery-secret-0123456789abcdef';
@@ -89,7 +87,7 @@ const SECRET = 'hook-delivery-secret-0123456789abcdef';
assert.strictEqual(got.headers['x-github-event'], 'push');
assert.strictEqual(got.headers['x-github-event'], 'push');
assert.strictEqual(got.headers['x-hub-signature-256'], 'sha256=abc');
assert.strictEqual(got.headers['x-hub-signature-256'], 'sha256=abc');
assert.strictEqual(got.headers['x-openvibe-hook'], hook.id);
assert.strictEqual(got.headers['x-openvibe-hook'], hook.id);
- assert.strictEqual(got.headers['x-openvibe-event-id'], envelope.event_id);
+ assert.strictEqual(got.headers['x-openvibe-event-id'], envelope.event_id, 'the event\'s own id, whichever way it came');
assert.strictEqual(got.headers.cookie, undefined);
assert.strictEqual(got.headers.cookie, undefined);
const again = await deliver(envelope);
const again = await deliver(envelope);
assert.strictEqual(again.status, 200);
assert.strictEqual(again.status, 200);
@@ -126,6 +124,39 @@ const SECRET = 'hook-delivery-secret-0123456789abcdef';
@@ -126,6 +124,39 @@ const SECRET = 'hook-delivery-secret-0123456789abcdef';
assert.strictEqual((await t.ctx.s.db.maybe('SELECT id FROM hook_requests WHERE id = $1', [ref])), null);
assert.strictEqual((await t.ctx.s.db.maybe('SELECT id FROM hook_requests WHERE id = $1', [ref])), null);
});
});
+ await check('a webhook that waited goes to the app as soon as its computer connects; Events\' retry then finds it settled', async () => {
+ const h3 = (await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: '/webhooks/waited' } })).json().hook;
+ const w = await t.get(`/h/${h3.id}`, { method: 'POST', body: '{"waited":true}', headers: { 'content-type': 'application/json' } });
+ assert.strictEqual(w.status, 202);
+ const env = (await t.ctx.s.db.one("SELECT envelope FROM service_outbox WHERE envelope->'payload'->>'body_ref' = $1", [w.json().id])).envelope;
+ const before = app.seen.length;
+ const c = t.client({ key: made.key, port: app.port });
+ await c.ready();
+ for (let i = 0; i < 100 && !app.seen.slice(before).some((x) => x.url === '/webhooks/waited'); i++) await new Promise((r) => setTimeout(r, 20));
+ const got = app.seen.slice(before).find((x) => x.url === '/webhooks/waited');
+ assert.ok(got, 'handed over on connect');
+ assert.strictEqual(got.body, '{"waited":true}');
+ assert.strictEqual(got.headers['x-openvibe-event-id'], env.event_id, 'the event\'s own id');
+ const late = await deliver(env);
+ assert.deepStrictEqual(late.json(), { gone: true }, 'Events\' delivery finds it settled');
+ assert.strictEqual(app.seen.filter((x) => x.url === '/webhooks/waited').length, 1, 'once');
+ c.stop();
+ });
+
+ await check('with the computer connected, Events\' delivery hands the webhook over itself', async () => {
+ const h4 = (await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: '/webhooks/live' } })).json().hook;
+ const c = t.client({ key: made.key, port: app.port });
+ await c.ready();
+ const w = await t.get(`/h/${h4.id}`, { method: 'POST', body: '{"live":1}', headers: { 'content-type': 'application/json' } });
+ const env = (await t.ctx.s.db.one("SELECT envelope FROM service_outbox WHERE envelope->'payload'->>'body_ref' = $1", [w.json().id])).envelope;
+ const r = await deliver(env);
+ assert.deepStrictEqual(r.json(), { delivered: true, status: 200 });
+ const got = app.seen.filter((x) => x.url === '/webhooks/live');
+ assert.strictEqual(got.length, 1);
+ assert.strictEqual(got[0].headers['x-openvibe-event-id'], env.event_id);
+ c.stop();
+ });
+
await check('the page: make one with a plain form', async () => {
await check('the page: make one with a plain form', async () => {
const r = await t.get(`/shares/${made.share.id}/hooks`, { as: alex, method: 'POST', form: { path: '/hooks/stripe', label: 'Stripe' } });
const r = await t.get(`/shares/${made.share.id}/hooks`, { as: alex, method: 'POST', form: { path: '/hooks/stripe', label: 'Stripe' } });
assert.strictEqual(r.status, 201, r.text);
assert.strictEqual(r.status, 201, r.text);