openvibers/openvibe.connect Commits
Webhooks that wait for your computer (CON-15) (#16)
Changes
@@@ -39,6 +39,28 @@ seconds, one use). It sends a fresh node token before the old one runs out, so aserving within about five minutes. While a computer serves a share, the share's key does not work; "Use a connect keyinstead" lets the computer go and shows a new key once (ADR-057 decision 4, CON-4).++### Webhooks that wait for your computer++++On a share's page, "Make a webhook address" gives you `https://connect.openvibe.website/h/hk_…` and a path on your app++(`/webhooks/github`). Give that address to GitHub, Stripe or anything else that sends webhooks. Each webhook is kept for++up to 7 days and handed to your app, at that path, as soon as your computer is connected, so a laptop that sleeps++misses nothing (CON-15, ADR-057 decision 7):++++- The address takes POST, PUT and PATCH, up to 1 MB and 120 a minute, and answers `202` at once.++- 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++ 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++ 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.++- Nobody signs in to send a webhook, so check the provider's own signature in your app.++++| Call | What |++|---|---|++| `GET /api/v1/shares/:id/hooks` | its webhook addresses, with `received`, `delivered` and `waiting` |++| `POST /api/v1/shares/:id/hooks` `{ path, label? }` | a new address → `201 { hook }` (at most 10 a share) |++| `DELETE /api/v1/shares/:id/hooks/:hook` | remove it; webhooks still waiting are dropped |++## What your app sees- `Host: localhost:<port>`; the share's address in `X-Forwarded-Host`; `X-Forwarded-Proto: https`.@@@ -104,6 +126,7 @@ browser ──https──▶ Cloudflare ──▶ nginx ──▶ Connect (5240)no bodies; a restart forgets them.- `server/auth/node-auth.js` — a computer as OpenVibe Node: its node token, and the single-use tunnel tokens (CON-4).- `server/machines.js` — the owner's computers, read from Network with the owner's own token.++- `server/hooks.js`, `server/http/hooks.js` — webhook addresses, the kept webhooks, and Events' deliveries (CON-15).- `public/ov-connect.js` — the client.The v0 transport is a WebSocket on 443 through Cloudflare (ADR-057's fallback path), so nothing needs a UDP port or annew file mode 100644@@@ -0,0 +1,55 @@++-- phase: expand++-- Hooks that wait for your computer (CON-15, ADR-057 decision 7). A hook is one address,++-- https://connect.openvibe.website/h/<id>, that a provider (GitHub, Stripe, …) sends webhooks to. Connect keeps each++-- request here for 7 days and publishes only a pointer to it (connect.hook.received, contracts 0.282.0) through the++-- outbox below. Events retries Connect's own subscription for up to seven days, and Connect hands the request to the++-- share's computer, at the hook's local path, once it is connected.++CREATE TABLE IF NOT EXISTS share_hooks (++ id text PRIMARY KEY, -- hk_<ulid>++ share_id text NOT NULL REFERENCES shares(id) ON DELETE CASCADE,++ owner_subject text NOT NULL, -- usr_… (the share's owner)++ label text,++ local_path text NOT NULL, -- where on the computer's app it goes: /webhooks/github++ status text NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'removed')),++ created_at timestamptz NOT NULL DEFAULT now(),++ removed_at timestamptz,++ received bigint NOT NULL DEFAULT 0,++ delivered bigint NOT NULL DEFAULT 0,++ last_received_at timestamptz,++ last_delivered_at timestamptz,++ last_status integer -- what the computer's app answered last++);++CREATE INDEX IF NOT EXISTS share_hooks_share ON share_hooks (share_id, created_at DESC);++++-- One webhook as it arrived: its method, the provider's headers (an allowlist; never cookies or forwarding headers),++-- and its body, until it is delivered or 7 days pass.++CREATE TABLE IF NOT EXISTS hook_requests (++ id text PRIMARY KEY, -- hb_<random>: the body_ref in the event++ hook_id text NOT NULL REFERENCES share_hooks(id) ON DELETE CASCADE,++ method text NOT NULL,++ headers jsonb NOT NULL DEFAULT '[]'::jsonb,++ body bytea NOT NULL,++ received_at timestamptz NOT NULL DEFAULT now(),++ expires_at timestamptz NOT NULL,++ delivered_at timestamptz,++ status integer++);++CREATE INDEX IF NOT EXISTS hook_requests_expiry ON hook_requests (expires_at);++CREATE INDEX IF NOT EXISTS hook_requests_hook ON hook_requests (hook_id, received_at DESC);++++-- The events outbox (openvibe-sdk/events createServiceOutbox): a request and its event commit together.++CREATE TABLE IF NOT EXISTS service_outbox (++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,++ event_id text NOT NULL UNIQUE,++ envelope jsonb NOT NULL,++ traceparent text,++ created_at bigint NOT NULL,++ attempts integer NOT NULL DEFAULT 0,++ next_attempt_at bigint NOT NULL DEFAULT 0,++ sent_at bigint,++ seq bigint,++ rejected_at bigint,++ last_error text++);++CREATE INDEX IF NOT EXISTS service_outbox_due ON service_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;++CREATE INDEX IF NOT EXISTS service_outbox_sent ON service_outbox (sent_at) WHERE sent_at IS NOT NULL;@@@ -132,6 +132,18 @@.cn-run { display: grid; gap: 10px; }.cn-run h2 { margin: 0; }.cn-run a { color: var(--room-strong); }++/* Webhooks that wait for your computer (CON-15): one row per address, its copy box and counts; the add form. */++.cn-hooks-sec { display: grid; gap: 10px; }++.cn-hooks-sec h2 { margin: 0; }++.cn-hooks-sec > p { margin: 0; color: var(--ovk-muted); }++.cn-hooks { list-style: none; margin: 0; padding: 0; display: grid; gap: 6px; }++.cn-hooks li { display: flex; flex-wrap: wrap; align-items: center; justify-content: space-between; gap: 8px 12px; padding: 10px 14px; border-radius: 10px; background: var(--ovk-panel); border: 1px solid var(--ovk-line); }++.cn-hook-main { display: grid; gap: 4px; min-width: 0; flex: 1 1 260px; }++.cn-hook-main code { overflow-wrap: anywhere; }++.cn-hook-main small { color: var(--ovk-muted); font-size: 13px; overflow-wrap: anywhere; }++.cn-hook-main small b { color: var(--ovk-text); }++.cn-hook-add { display: flex; flex-wrap: wrap; align-items: end; gap: 10px 14px; }++.cn-hook-add .cn-field { flex: 1 1 200px; }/* Which computer serves it (CON-4): the owner's machines, one row each with its button; the bound one's commands. */.cn-node { display: grid; gap: 10px; }.cn-node h2 { margin: 0; }@@@ -31,6 +31,9 @@ const { createApi } = require('./http/api');const { createNodeApi } = require('./http/node-api');const { createNodeAuth } = require('./auth/node-auth');const { createMachines } = require('./machines');++const { createHooks } = require('./hooks');++const { createHookRoutes } = require('./http/hooks');++const { createServiceOutbox } = require('openvibe-sdk/events');const { createPageRoutes } = require('./http/pages');const { createServiceReadiness } = require('./observability');const { assetVersion, setRelease } = require('./render/layout');@@@ -79,7 +82,17 @@ async function createApp(opts = {}) {? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }): async () => { throw new Error('OV_OAUTH_CLIENT_SECRET is not set: Connect cannot answer account events'); });-- const ctx = { config, s, keys, sso, principal, shares, tunnels, gate, proxy, notifier, nodeAuth, machines, accountData, accountSend, log };++ // Hooks that wait for your computer (CON-15): a webhook and its pointer commit together; the relay (started in++ // index.js, never in drill mode) publishes the pointer to OpenVibe.Events with Connect's own token.++ const outbox = opts.outbox !== undefined ? opts.outbox : createServiceOutbox({++ db: s.db, source: 'connect', eventsUrl: config.events.url, networkInternalUrl: config.networkInternalUrl,++ clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl, log, now: s.now,++ table: 'service_outbox', eventTypes: ['connect.hook.received'],++ validate: (env) => contracts.validate('events.event-envelope@1', env),++ });++ const hooks = createHooks(s, { config, outbox, now: s.now });++++ const ctx = { config, s, keys, sso, principal, shares, tunnels, gate, proxy, notifier, nodeAuth, machines, hooks, outbox, accountData, accountSend, log };/** Shares whose time is up close themselves (CON-6): their computers are disconnected and their lists forgotten. */ctx.expireSweep = async () => {@@@ -87,6 +100,8 @@ async function createApp(opts = {}) {try { due = await shares.expireDue(); } catch (err) { log.warn('[OpenVibe.Connect] expiry sweep failed:', err.message); return []; }due.forEach((r) => { tunnels.close(r.id, 4003, 'the share ended'); proxy.forget(r.host); proxy.inspector.forget(r.id); });if (due.length) log.log(`[OpenVibe.Connect] ${due.length} share(s) ended on time`);++ // Webhooks kept past their 7 days go, delivered or not (CON-15).++ try { const gone = await hooks.sweep(); if (gone) log.log(`[OpenVibe.Connect] ${gone} webhook(s) past 7 days removed`); } catch (err) { log.warn('[OpenVibe.Connect] webhook sweep failed:', err.message); }return due;};if (opts.expirySweep !== false) {@@@ -168,6 +183,9 @@ async function createApp(opts = {}) {app.use('/auth', sso.routes());{ const legal = require('openvibe-shared/legal'); app.get(legal.PATHS, legal.handler({ id: 'connect', service: 'connect', host: 'connect.openvibe.website', name: 'OpenVibe Connect', profile: 'ugc' })); }++ // ── Webhooks to a share's hooks, and Events' deliveries of them back (CON-15; /internal/ is loopback only) ──++ app.use(createHookRoutes(ctx));++// ── OpenVibe.Events → this service (loopback only; nginx answers 404 for /internal/) ──app.use('/internal/events', accountData.consumer({secrets: config.events.secrets,@@@ -382,7 +382,43 @@ function createProxy({ config, shares, tunnels, gate, log = console, now = () =>/** The grant refused someone: counted on the share page as "turned away". */function denied(share, who) { visit(share, who, { denied: true }); }-- return { handle, upgrade, isShareHost, flush, forget, denied, stop, responseHeaders, forwardHeaders, inspector };++ /**++ * One stored webhook to the share's computer (CON-15): the request as it arrived, at the hook's path, with++ * x-openvibe-hook and x-openvibe-event-id so the app can tell and dedupe. The answer's body is read and dropped.++ * → { status } when the app answered, { offline: true } with no computer, { error } when it broke on the way.++ */++ function deliverHook(share, { method, path, headers = [], body, hookId, eventId }) {++ const tunnel = tunnels.get(share.id);++ if (!tunnel) return Promise.resolve({ offline: true });++ const port = (tunnel.clientInfo && tunnel.clientInfo.port) || share.local_port;++ const rec = inspector.begin(share.id, { method, path, who: { username: 'webhook' } });++ return new Promise((resolve) => {++ let done = false;++ const finish = (out) => { if (done) return; done = true; clearTimeout(timer); inspector.end(rec, out.status ? { status: out.status } : { error: out.error || 'offline' }); resolve(out); };++ let status = 0;++ const id = tunnel.open({++ onRes(m) { status = Number(m.status) || 0; },++ onData() {},++ onEnd() { tunnel.release(id); finish(status >= 100 && status <= 599 ? { status } : { error: 'bad_status' }); },++ onAbort(m) { finish({ error: (m && m.code) || 'local_error' }); },++ onPause() {}, onResume() {},++ });++ const timer = setTimeout(() => { tunnel.abort(id, 'timeout'); finish({ error: 'timeout' }); }, config.shares.requestTimeoutMs);++ timer.unref();++ const out = (Array.isArray(headers) ? headers : []).filter((h) => Array.isArray(h) && h.length === 2 && !/^(host|content-length|x-openvibe-)/i.test(String(h[0])));++ out.push(['host', `localhost:${port}`], ['x-forwarded-host', share.host], ['x-forwarded-proto', config.shares.scheme],++ ['x-openvibe-hook', String(hookId || '')], ['x-openvibe-event-id', String(eventId || '')], ['content-length', String(body ? body.length : 0)]);++ tunnel.send({ t: 'req', id, method, path, headers: out });++ if (body && body.length) {++ rec.in += body.length;++ const CHUNK = 32 * 1024;++ for (let i = 0; i < body.length; i += CHUNK) tunnel.data(id, body.subarray(i, i + CHUNK));++ }++ tunnel.end(id);++ });++ }++++ return { handle, upgrade, isShareHost, flush, forget, denied, stop, responseHeaders, forwardHeaders, deliverHook, inspector };}/** Write an HTTP answer straight to an upgrade's socket (the gate's 302/401/404/502 before any tunnel opens). */nnew file mode 100644@@@ -0,0 +1,142 @@++'use strict';++++/**++ * Hooks that wait for your computer (CON-15, ADR-057 decision 7). A hook is an address a provider sends webhooks to,++ * https://connect.openvibe.website/h/<hk_…>, bound to one share and a path on the computer's app. A webhook is kept++ * (7 days) and a pointer to it is published as connect.hook.received through the outbox, in one transaction. Events++ * delivers it to Connect's own subscription (retried for up to seven days), and Connect hands it to the share's++ * computer when it is connected. The computer may be asleep for days and still get every webhook, in order of arrival++ * or not, each once it answers.++ *++ * 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++ */++const crypto = require('crypto');++++const HOOK_ID = /^hk_[0-9A-HJKMNP-TV-Z]{26}$/;++const REF_RE = /^hb_[A-Za-z0-9_-]{16,64}$/;++const PATH_RE = /^\/[\x21-\x7e]{0,499}$/;++const METHODS = ['POST', 'PUT', 'PATCH'];++const KEEP_MS = 7 * 86400_000;++const MAX_BODY = 1024 * 1024;++const PER_SHARE = 10;++// The provider headers a webhook keeps: what apps check signatures and events with. Never cookies, authorization,++// forwarding or Cloudflare headers, and never Connect's own.++const DROP = /^(x-forwarded-|x-real-ip$|x-openvibe-|cf-|x-amz-cf-)/;++const KEEP = /^(content-type|content-encoding|user-agent|stripe-signature|svix-.+|webhook-.+|x-.+)$/;++++class HookError extends Error {++ constructor(status, code, message) { super(message); this.status = status; this.code = code; }++}++++/** The headers a stored webhook keeps: at most 40, each at most 2 KB. → [[name, value]] */++function keepHeaders(headers) {++ const out = [];++ for (const [k, v] of Object.entries(headers || {})) {++ const name = String(k).toLowerCase();++ if (DROP.test(name) || !KEEP.test(name)) continue;++ const value = Array.isArray(v) ? v.join(', ') : String(v);++ if (value.length > 2048) continue;++ out.push([name, value]);++ if (out.length >= 40) break;++ }++ return out;++}++++function createHooks(s, { config, outbox = null, now = () => Date.now() }) {++ const { db } = s;++ const base = config.baseUrl;++ const iso = (v) => (v ? new Date(v).toISOString() : null);++ const view = (r) => r && ({++ id: r.id, share_id: r.share_id, url: `${base}/h/${r.id}`, local_path: r.local_path, label: r.label || null, status: r.status,++ created_at: iso(r.created_at), received: Number(r.received || 0), delivered: Number(r.delivered || 0),++ last_received_at: iso(r.last_received_at), last_delivered_at: iso(r.last_delivered_at), last_status: r.last_status == null ? null : Number(r.last_status),++ waiting: r.waiting == null ? undefined : Number(r.waiting),++ });++++ async function get(id) {++ if (!HOOK_ID.test(String(id || ''))) return null;++ return view(await db.maybe('SELECT * FROM share_hooks WHERE id = $1', [id]));++ }++++ async function list(shareId) {++ const rows = await db.many(`SELECT h.*, (SELECT count(*)::int FROM hook_requests r WHERE r.hook_id = h.id AND r.delivered_at IS NULL AND r.expires_at > now()) AS waiting++ FROM share_hooks h WHERE h.share_id = $1 AND h.status = 'active' ORDER BY h.created_at`, [shareId]);++ return rows.map(view);++ }++++ async function create(owner, share, input = {}) {++ if (!share || share.owner_subject !== owner.subject) throw new HookError(404, 'share.not_found', 'No such share.');++ if (share.status !== 'active') throw new HookError(410, 'share.revoked', 'That share is closed.');++ const localPath = String(input.path || '').trim() || '/';++ if (!PATH_RE.test(localPath)) throw new HookError(422, 'hook.invalid_path', 'The path on your app starts with "/" and has no spaces, like /webhooks/github.');++ const label = input.label ? String(input.label).trim().slice(0, 80) || null : null;++ const n = await db.one("SELECT count(*)::int AS n FROM share_hooks WHERE share_id = $1 AND status = 'active'", [share.id]);++ if (n.n >= PER_SHARE) throw new HookError(409, 'hook.limit', `A share can have ${PER_SHARE} webhook addresses. Remove one first.`);++ const id = s.newId('hk');++ await db.exec('INSERT INTO share_hooks (id, share_id, owner_subject, label, local_path) VALUES ($1, $2, $3, $4, $5)', [id, share.id, owner.subject, label, localPath]);++ return get(id);++ }++++ async function remove(owner, id) {++ const hook = await db.maybe('SELECT * FROM share_hooks WHERE id = $1', [String(id || '')]);++ if (!hook || hook.owner_subject !== owner.subject) throw new HookError(404, 'hook.not_found', 'No such webhook address.');++ await db.exec("UPDATE share_hooks SET status = 'removed', removed_at = now() WHERE id = $1", [hook.id]);++ await db.exec('DELETE FROM hook_requests WHERE hook_id = $1 AND delivered_at IS NULL', [hook.id]);++ return view({ ...hook, status: 'removed' });++ }++++ /** A webhook arrived: kept, and its pointer published, together. → { ref } */++ async function receive(hook, { method, headers, body }) {++ if (!METHODS.includes(method)) throw new HookError(405, 'hook.method', 'A webhook address takes POST, PUT or PATCH.');++ const buf = Buffer.isBuffer(body) ? body : Buffer.from(body == null ? '' : String(body));++ if (buf.length > MAX_BODY) throw new HookError(413, 'hook.too_large', 'A webhook is at most 1 MB.');++ const kept = keepHeaders(headers);++ const contentType = (kept.find(([k]) => k === 'content-type') || [null, null])[1];++ const ref = `hb_${crypto.randomBytes(18).toString('base64url')}`;++ const at = new Date(now());++ 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('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',++ 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() },++ });++ }++ });++ if (outbox && outbox.kick) outbox.kick().catch(() => {});++ return { ref };++ }++++ /** The stored request a delivery points at, with its hook; null when it was delivered, removed or expired. */++ async function request(ref) {++ if (!REF_RE.test(String(ref || ''))) return null;++ 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]);++ 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 };++ }++++ /** The computer's app answered: the request is delivered (its body is dropped) and the hook counts it. */++ async function settle(ref, status) {++ const at = new Date(now()).toISOString();++ const r = await db.maybe("UPDATE hook_requests SET delivered_at = $2, status = $3, body = ''::bytea WHERE id = $1 AND delivered_at IS NULL RETURNING hook_id", [ref, at, status]);++ if (r) await db.exec('UPDATE share_hooks SET delivered = delivered + 1, last_delivered_at = $2, last_status = $3 WHERE id = $1', [r.hook_id, at, status]);++ return Boolean(r);++ }++++ /** Requests older than 7 days go, delivered or not. */++ async function sweep() {++ const r = await db.many('DELETE FROM hook_requests WHERE expires_at <= $1 RETURNING id', [new Date(now()).toISOString()]);++ return r.length;++ }++++ return { get, list, create, remove, receive, request, settle, sweep, KEEP_MS, MAX_BODY };++}++++module.exports = { createHooks, HookError, keepHeaders, HOOK_ID, METHODS, MAX_BODY, KEEP_MS };@@@ -14,6 +14,9 @@* PUT /shares/:id/node { node: 'nod_…' } one of your computers serves it, as OpenVibe Node (CON-4): the key* stops working → { share }; { node: null } lets the computer go → { share, key } (a new key)* GET /machines your computers that can serve a share (Network's list of your machines)++ * GET /shares/:id/hooks its webhook addresses (CON-15): each kept webhook waits up to 7 days for the computer++ * POST /shares/:id/hooks { path, label? } → 201 { hook }: https://connect.openvibe.website/h/<id> → that path++ * DELETE /shares/:id/hooks/:hook removes it; webhooks still waiting are dropped* GET /shares/:id/visits who opened it in the last 7 days* GET /shares/:id/requests the last 100 requests through it, newest first (?after=<seq>: only newer ones);* memory only, query values shown as "…" (edge/inspector.js)@@@ -24,9 +27,10 @@ const express = require('express');const { http } = require('openvibe-contracts');const { asyncRouter } = require('./router');const { ShareError } = require('../shares');++const { HookError } = require('../hooks');function createApi(ctx) {-- const { shares, tunnels, principal, proxy, machines } = ctx;++ const { shares, tunnels, principal, proxy, machines, hooks } = ctx;const r = asyncRouter();r.use(express.json({ limit: '16kb' }));r.use((req, res, next) => { res.set('Cache-Control', 'no-store'); next(); });@@@ -38,7 +42,7 @@ function createApi(ctx) {const shape = (s) => ({ ...s, connected: tunnels.isConnected(s.id) });const guard = (fn) => async (req, res) => {try { return await fn(req, res); } catch (err) {-- if (err instanceof ShareError) return fail(res, req, err.status, err.code, err.message);++ if (err instanceof ShareError || err instanceof HookError) return fail(res, req, err.status, err.code, err.message);throw err;}};@@@ -114,6 +118,24 @@ function createApi(ctx) {return res.json({ share: shape(updated) });}));++ r.get('/shares/:id/hooks', guard(async (req, res) => {++ const share = await mine(req, res);++ return share ? res.json({ hooks: await hooks.list(share.id) }) : undefined;++ }));++ r.post('/shares/:id/hooks', guard(async (req, res) => {++ const share = await mine(req, res);++ if (!share) return undefined;++ const b = req.body || {};++ return res.status(201).json({ hook: await hooks.create(owner(req), share, { path: b.path, label: b.label }) });++ }));++ r.delete('/shares/:id/hooks/:hook', guard(async (req, res) => {++ const share = await mine(req, res);++ if (!share) return undefined;++ const hook = await hooks.get(req.params.hook);++ if (!hook || hook.share_id !== share.id) return fail(res, req, 404, 'hook.not_found', 'No such webhook address.');++ return res.json({ hook: await hooks.remove(owner(req), hook.id) });++ }));++r.get('/shares/:id/requests', guard(async (req, res) => {const share = await mine(req, res);return share ? res.json({ requests: proxy.inspector.list(share.id, { after: req.query.after }), keep: proxy.inspector.KEEP }) : undefined;nnew file mode 100644@@@ -0,0 +1,74 @@++'use strict';++++/**++ * Hooks that wait for your computer (CON-15): the address providers send to, and Events' deliveries back.++ *++ * POST|PUT|PATCH /h/:id a provider's webhook (1 MB): kept 7 days, its pointer published → 202++ * POST /internal/hook-deliveries Events, signed v2, loopback only (nginx answers 404 for /internal/): the++ * webhook goes to the share's computer. 200 once its app answered (any status++ * below 500) or when the webhook is gone; 503 while the computer is offline or++ * its app fails, so Events tries again (up to seven days, server/index.js).++ */++const express = require('express');++const { parseDelivery } = require('openvibe-sdk/events');++const { HookError, METHODS, MAX_BODY } = require('../hooks');++++const LOOPBACK = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']);++const PER_HOOK_MINUTE = 120;++++function createHookRoutes(ctx) {++ const { config, hooks, shares, proxy, log } = ctx;++ const r = express.Router();++ const counts = new Map(); // hook id → { minute, n }++++ const tooMany = (id) => {++ const minute = Math.floor(Date.now() / 60_000);++ const c = counts.get(id);++ if (!c || c.minute !== minute) { counts.set(id, { minute, n: 1 }); if (counts.size > 5000) counts.delete(counts.keys().next().value); return false; }++ c.n += 1;++ return c.n > PER_HOOK_MINUTE;++ };++ const answer = (res, status, body) => res.status(status).set('Cache-Control', 'no-store').json(body);++++ r.all('/h/:id', express.raw({ type: () => true, limit: MAX_BODY }), async (req, res) => {++ if (!METHODS.includes(req.method)) return answer(res.set('Allow', METHODS.join(', ')), 405, { error: 'method', detail: 'A webhook address takes POST, PUT or PATCH.' });++ const hook = await hooks.get(req.params.id);++ if (!hook || hook.status !== 'active') return answer(res, 404, { error: 'not_found', detail: 'No webhook address here.' });++ const share = await shares.get(hook.share_id);++ if (!share || share.status !== 'active') return answer(res, 410, { error: 'gone', detail: 'The share this address served is closed.' });++ if (tooMany(hook.id)) return answer(res.set('Retry-After', '60'), 429, { error: 'rate_limited', detail: `At most ${PER_HOOK_MINUTE} webhooks a minute to one address.` });++ try {++ const { ref } = await hooks.receive(hook, { method: req.method, headers: req.headers, body: Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0) });++ return answer(res, 202, { received: true, id: ref, delivered_when: 'the computer is connected (kept up to 7 days)' });++ } catch (err) {++ if (err instanceof HookError) return answer(res, err.status, { error: err.code, detail: err.message });++ log.warn('[OpenVibe.Connect] webhook not kept:', err && err.message);++ return answer(res.set('Retry-After', '30'), 503, { error: 'unavailable', detail: 'Try again shortly.' });++ }++ });++++ r.post('/internal/hook-deliveries', express.raw({ type: () => true, limit: '64kb' }), async (req, res) => {++ if (!LOOPBACK.has(String(req.socket.remoteAddress || ''))) return res.status(404).end();++ let d = null;++ for (const secret of config.events.secrets || []) {++ if (secret) d = parseDelivery(req.body, req.headers, secret);++ if (d) break;++ }++ if (!d) return res.status(401).json({ error: 'bad_signature' });++ const e = d.event || {};++ if (e.event_type !== 'connect.hook.received' || !e.payload) return res.status(200).json({ ignored: true });++ const stored = await hooks.request(e.payload.body_ref);++ if (!stored) return res.status(200).json({ gone: true });++ const share = await shares.get(stored.shareId);++ if (!share || share.status !== 'active') return res.status(200).json({ gone: true });++ const out = await proxy.deliverHook(share, { method: stored.method, path: stored.path, headers: stored.headers, body: stored.body, hookId: stored.hookId, eventId: e.event_id });++ if (out.offline) return res.status(503).set('Retry-After', '300').json({ waiting: 'the computer is not connected' });++ if (!out.status || out.status >= 500) return res.status(503).json({ retry: out.error || `the app answered ${out.status}` });++ await hooks.settle(stored.ref, out.status);++ return res.status(200).json({ delivered: true, status: out.status });++ });++++ return r;++}++++module.exports = { createHookRoutes };@@@ -10,6 +10,8 @@* POST /shares/:id/key a new key (shown once); the connected computer is disconnected* POST /shares/:id/node node=nod_… one of your computers serves it (CON-4); node= (empty) lets it go and shows a* new key once++ * POST /shares/:id/hooks path=&label= a webhook address for it (CON-15)++ * POST /shares/:id/hooks/:hook/remove* POST /shares/:id/close close it: the address answers "closed" from then on* /grant?share=&next= the gate's sign-in step (server/edge/gate.js): signed in and allowed → back to the* share host with a one-time code@@@ -21,9 +23,10 @@ const { sameOrigin } = require('./principal');const { send } = require('../render/layout');const views = require('../views');const { ShareError } = require('../shares');++const { HookError } = require('../hooks');function createPageRoutes(ctx) {-- const { config, shares, tunnels, gate, proxy, machines } = ctx;++ const { config, shares, tunnels, gate, proxy, machines, hooks } = ctx;const r = asyncRouter();r.use(express.urlencoded({ extended: false, limit: '16kb' }));const page = (req, res, o, status = 200) => send(res, status, { viewer: req.viewer, config, path: req.originalUrl, ...o });@@@ -82,11 +85,12 @@ function createPageRoutes(ctx) {async function showShare(req, res, share, extra = {}, status = 200) {const tunnel = tunnels.get(share.id);-- const [visits, mine] = await Promise.all([++ const [visits, mine, hookList] = await Promise.all([shares.visits(share.id),share.status === 'active' ? machines.list(req.viewer.token) : Promise.resolve({ ok: true, machines: [] }),++ share.status === 'active' ? hooks.list(share.id) : Promise.resolve([]),]);-- return page(req, res, { title: share.host, body: views.sharePage({ share, connected: Boolean(tunnel), tunnel, visits, requests: ctx.proxy ? ctx.proxy.inspector.list(share.id) : [], baseUrl: config.baseUrl, staff: req.viewer.role === 'admin', machines: mine.ok ? mine.machines : null, ...extra }) }, status);++ return page(req, res, { title: share.host, body: views.sharePage({ share, connected: Boolean(tunnel), tunnel, visits, requests: ctx.proxy ? ctx.proxy.inspector.list(share.id) : [], baseUrl: config.baseUrl, staff: req.viewer.role === 'admin', machines: mine.ok ? mine.machines : null, hooks: hookList, ...extra }) }, status);}r.get('/shares/:id', async (req, res) => {@@@ -147,6 +151,29 @@ function createPageRoutes(ctx) {}});++ r.post('/shares/:id/hooks', async (req, res) => {++ if (!guard(req, res)) return undefined;++ const share = await mine(req, res);++ if (!share) return undefined;++ try {++ const hook = await hooks.create(owner(req), share, { path: req.body.path, label: req.body.label });++ return showShare(req, res, share, { notice: `Webhook address made. Give ${hook.url} to the service that sends them.` }, 201);++ } catch (err) {++ if (!(err instanceof HookError)) throw err;++ return showShare(req, res, share, { error: err.message }, err.status || 422);++ }++ });++++ r.post('/shares/:id/hooks/:hook/remove', async (req, res) => {++ if (!guard(req, res)) return undefined;++ const share = await mine(req, res);++ if (!share) return undefined;++ const hook = await hooks.get(req.params.hook);++ if (!hook || hook.share_id !== share.id) return showShare(req, res, share, { error: 'No such webhook address.' }, 404);++ await hooks.remove(owner(req), hook.id);++ return showShare(req, res, share, { notice: 'Removed. That address answers "not found" from now on.' });++ });++r.post('/shares/:id/close', async (req, res) => {if (!guard(req, res)) return undefined;const share = await mine(req, res);@@@ -18,10 +18,14 @@ const TABLES = [// the account is taken off it.{ table: 'share_reports', subject: 'reporter_subject', value: (usr) => usr, file: 'reports.json', erase: { anonymize: {} },columns: ['id', 'host', 'reason', 'details', 'status', 'created_at'] },++ // Webhook addresses on a person's shares (CON-15). Their waiting webhooks (hook_requests) go with them, and with++ // the shares (ON DELETE CASCADE); the export names the addresses and their counts, never a webhook's body.++ { table: 'share_hooks', subject: 'owner_subject', value: (usr) => usr, file: 'webhooks.json', erase: 'delete',++ columns: ['id', 'share_id', 'label', 'local_path', 'status', 'created_at', 'removed_at', 'received', 'delivered', 'last_received_at', 'last_delivered_at', 'last_status'] },{table: 'shares', subject: 'owner_subject', value: (usr) => usr, file: 'shares.json', erase: 'delete',columns: ['id', 'owner_username', 'label', 'host', 'local_port', 'mode', 'named', 'status', 'policy_version', 'key_prefix', 'note',-- 'created_at', 'updated_at', 'revoked_at', 'last_connected_at', 'last_visit_at', 'requests', 'bytes_in', 'bytes_out'],++ 'node_id', 'node_name', 'created_at', 'updated_at', 'revoked_at', 'last_connected_at', 'last_visit_at', 'requests', 'bytes_in', 'bytes_out'],},];@@@ -30,7 +34,7 @@ function create({ db, log = console } = {}) {db,service: 'connect',tables: TABLES,-- note: 'Your shares on OpenVibe Connect and your visits to other people\'s shares. Connect never stores what passed through a share.',++ note: 'Your shares on OpenVibe Connect, their webhook addresses, and your visits to other people\'s shares. Connect never stores what passed through a share; a webhook to one of your addresses is kept only until your computer takes it, at most 7 days.',log,});}@@@ -60,13 +60,47 @@ async function start() {clientSecret: config.oauth.clientSecret,log: ctx.log || console,});-- const extra = [() => { if (subs) subs.stop(); }];++ // Hooks (CON-15): the outbox relay publishes webhook pointers, and Connect's own subscription brings them back to++ // /internal/hook-deliveries, retried for about seven days while a computer is offline. Off in drill mode and++ // without the events URL, the delivery secret and the client secret.++ const hookSubs = config.drill ? null : ensureHookSubscription({ config, log: ctx.log || console });++ if (!config.drill && ctx.outbox) ctx.outbox.start();++ const extra = [() => { if (subs) subs.stop(); }, () => { if (hookSubs) hookSubs.stop(); }, () => { if (ctx.outbox) ctx.outbox.stop(); }];createLifecycle({ server, ctx, extra });return { server, ctx };}++// About seven days of retries (Events allows 20 attempts, each wait at most a day): 5 s, 30 s, 2 min, 10 min, 30 min,++// 1, 2, 3, 4, 6, 8 hours, four of 12 hours and four of a day.++const HOOK_BACKOFF_MS = [5e3, 30e3, 120e3, 600e3, 1800e3, 3600e3, 7200e3, 10800e3, 14400e3, 21600e3, 28800e3,++ 43200e3, 43200e3, 43200e3, 43200e3, 86400e3, 86400e3, 86400e3, 86400e3];++++/** Connect's subscription to its own connect.hook.received, made once (409 means it is there); tried again every 10 minutes until it is. */++function ensureHookSubscription({ config, log = console, fetchImpl = globalThis.fetch }) {++ const secret = config.events.secrets[0] || '';++ if (!config.events.url || !secret || !config.oauth.clientSecret) { log.warn('[OpenVibe.Connect] webhook deliveries off: CONNECT_EVENTS_URL, CONNECT_EVENTS_SECRET or the client secret is unset'); return null; }++ const { serviceAuth } = require('openvibe-contracts');++ const tokens = serviceAuth.createTokenClient({ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, audience: 'openvibe.events', scope: 'events.subscription.manage', fetchImpl });++ let timer = null;++ let stopped = false;++ async function once() {++ try {++ const res = await fetchImpl(`${config.events.url}/api/v1/subscriptions`, {++ method: 'POST', headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...(await tokens.authHeaders()) },++ body: JSON.stringify({ topic_pattern: 'connect.hook.received', endpoint: `http://127.0.0.1:${config.port}/internal/hook-deliveries`, secret, retry_policy: { max_attempts: 20, backoff_ms: HOOK_BACKOFF_MS } }),++ signal: AbortSignal.timeout(10_000),++ });++ if (res.status === 201 || res.status === 409) { log.log(`[OpenVibe.Connect] webhook deliveries subscribed (${res.status})`); return; }++ log.warn(`[OpenVibe.Connect] webhook subscription not made: HTTP ${res.status}`);++ } catch (err) { log.warn(`[OpenVibe.Connect] webhook subscription not made: ${err.message}`); }++ if (!stopped) { timer = setTimeout(once, 600_000); timer.unref(); }++ }++ once();++ return { stop() { stopped = true; clearTimeout(timer); } };++}++if (require.main === module) {start().catch((err) => { console.error('[OpenVibe.Connect] failed to start:', err); process.exit(1); });}--module.exports = { start, createLifecycle };++module.exports = { start, createLifecycle, ensureHookSubscription, HOOK_BACKOFF_MS };@@@ -259,7 +259,26 @@ function nodeBlock(share, machines, now) {return `<section class="cn-node" aria-labelledby="cn-node-h"><h2 id="cn-node-h">Or serve it from one of your computers</h2>${intro}${list}</section>`;}--function sharePage({ share, connected, tunnel = null, visits = [], requests = [], key = null, baseUrl, notice = null, error = null, staff = false, machines = [], now = Date.now() }) {++/**++ * Webhooks that wait for your computer (CON-15): each address, where it goes on the computer's app, what came and what++ * was handed over, and how many wait; the form for another. Plain forms, no script.++ */++function hooksBlock(share, hooks, now) {++ const rows = hooks.length ? `<ul class="cn-hooks">${hooks.map((h, i) => `<li><div class="cn-hook-main"><div class="cn-copy"><code id="cn-hook-${i}">${esc(h.url)}</code><button type="button" class="cn-btn cn-btn-quiet" data-ovk-copy="#cn-hook-${i}" hidden>${kit.icon('copy')} Copy</button></div>++<small>→ localhost:${esc(share.local_port)}${esc(h.local_path)}${h.label ? ` · ${esc(h.label)}` : ''} · ${esc(h.received)} received, ${esc(h.delivered)} handed over${h.waiting ? `, <b>${esc(h.waiting)} waiting for your computer</b>` : ''}${h.last_received_at ? ` · last ${esc(ago(h.last_received_at, now))}` : ''}${h.last_status ? ` · your app answered ${esc(h.last_status)}` : ''}</small></div>++<form method="post" action="/shares/${esc(share.id)}/hooks/${esc(h.id)}/remove"><button class="cn-btn cn-btn-quiet" type="submit">${kit.icon('close')} Remove</button></form></li>`).join('')}</ul>`++ : '<p class="cn-empty">No webhook addresses yet.</p>';++ return `<section class="cn-hooks-sec" aria-labelledby="cn-hooks-h"><h2 id="cn-hooks-h">Webhooks that wait for your computer</h2>++<p>Give a service (GitHub, Stripe, anything that sends webhooks) an address here instead of your share's. Each webhook is kept for up to 7 days and handed to your app, at the path you pick, as soon as your computer is connected: a laptop that sleeps misses nothing. Nobody has to sign in to send one, so check the provider's signature in your app.</p>++${rows}++<form class="cn-form cn-hook-add" method="post" action="/shares/${esc(share.id)}/hooks">++<label class="cn-field"><span>Path on your app</span><input name="path" maxlength="500" value="/webhooks" placeholder="/webhooks/github" autocapitalize="none" autocomplete="off" spellcheck="false"></label>++<label class="cn-field"><span>Name (optional)</span><input name="label" maxlength="80" placeholder="GitHub pushes"></label>++<button class="cn-btn" type="submit">${kit.icon('plus')} Make a webhook address</button>++</form></section>`;++}++++function sharePage({ share, connected, tunnel = null, visits = [], requests = [], key = null, baseUrl, notice = null, error = null, staff = false, machines = [], hooks = [], now = Date.now() }) {const closed = share.status !== 'active';const state = closed ? 'Closed' : connected ? 'Connected' : 'Not connected';const stateText = closed ? `Closed ${ago(share.revoked_at, now)}. The address answers "closed" and stays yours.`@@@ -302,6 +321,7 @@ ${share.expires_at ? `<div><dt>${closed ? 'Ended' : 'Closes'}</dt><dd><time date</dl>${!key && !closed && !share.node ? `<section class="cn-run"><h2>Connect it</h2><div class="cn-copy"><code id="cn-run">node ov-connect.js --key ovc_… --port ${esc(share.local_port)}</code></div><p class="cn-fine">The key was shown once, when it was made. <a href="${esc(CLIENT_URL(baseUrl))}">Download ov-connect.js</a> (Node 22 or newer).</p></section>` : ''}${closed ? '' : nodeBlock(share, machines, now)}++${closed ? '' : hooksBlock(share, hooks, now)}<section aria-labelledby="cn-visits-h"><h2 id="cn-visits-h">Who opened it (7 days)</h2>${visitRows}</section>${closed ? '' : `<section class="cn-inspector" aria-labelledby="cn-reqs-h" data-cn-reqs="${esc(share.id)}"><h2 id="cn-reqs-h">Requests</h2><p class="cn-fine">The last 100 through this share, newest first, added as they happen. Kept in memory only, never written down: no headers, no bodies, and query values show as "…".</p>nnew file mode 100644@@@ -0,0 +1,139 @@++'use strict';++/**++ * Hooks that wait for your computer (CON-15): a webhook address on a share, a provider's webhook kept with only the++ * provider's headers and its pointer in the outbox (one transaction), Events' signed delivery handed to the computer++ * once it is connected (503 while it is not, so Events retries), once only, and the sweep after 7 days.++ * node test/hooks.test.js++ */++const assert = require('assert');++const contracts = require('openvibe-contracts');++const { signDeliveryHeaders } = require('openvibe-sdk/events');++const { boot, startLocalApp } = require('./helpers/boot');++const { check, done } = require('./helpers/check');++++const SECRET = 'hook-delivery-secret-0123456789abcdef';++++(async () => {++ const t = await boot({ env: { CONNECT_EVENTS_SECRET: SECRET } });++ const app = await startLocalApp();++ const alex = t.network.addUser('alex');++ const bob = t.network.addUser('bob');++ const made = await t.makeShare(alex, { label: 'bot', port: app.port });++ let hook;++ let envelope;++++ const deliver = async (env, secret = SECRET) => {++ const raw = Buffer.from(JSON.stringify({ event: env }));++ const headers = { 'content-type': 'application/json', ...signDeliveryHeaders(raw, secret), 'x-openvibe-subscription-id': 'sub_x', 'x-openvibe-delivery-attempt': '1' };++ return t.get('/internal/hook-deliveries', { method: 'POST', body: raw, headers });++ };++++ await check('the owner makes a webhook address: a path on the app, at most ten, never on another person\'s share', async () => {++ const r = await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: '/webhooks/github', label: 'GitHub pushes' } });++ assert.strictEqual(r.status, 201, r.text);++ hook = r.json().hook;++ assert.match(hook.url, /^https:\/\/connect\.openvibe\.website\/h\/hk_[0-9A-HJKMNP-TV-Z]{26}$/);++ assert.strictEqual(hook.local_path, '/webhooks/github');++ assert.strictEqual((await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: 'no-slash' } })).status, 422);++ assert.strictEqual((await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: '/a b' } })).status, 422);++ assert.strictEqual((await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: bob, json: { path: '/x' } })).status, 404);++ const list = await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex });++ assert.deepStrictEqual(list.json().hooks.map((h) => h.id), [hook.id]);++ const page = await t.get(`/shares/${made.share.id}`, { as: alex });++ assert.match(page.text, /Webhooks that wait for your computer/);++ assert.ok(page.text.includes(hook.url));++ });++++ await check('a provider\'s webhook: 202, kept with its headers (never cookies or forwarding), its pointer in the outbox', async () => {++ const body = JSON.stringify({ ref: 'refs/heads/main', commits: 3 });++ const r = await t.get(`/h/${hook.id}`, { method: 'POST', body, headers: { 'content-type': 'application/json', 'x-github-event': 'push', 'x-hub-signature-256': 'sha256=abc', cookie: 'session=secret', 'x-forwarded-for': '203.0.113.9', authorization: 'Bearer nope' } });++ assert.strictEqual(r.status, 202, r.text);++ const ref = r.json().id;++ assert.match(ref, /^hb_/);++ const row = await t.ctx.s.db.one('SELECT method, headers, body FROM hook_requests WHERE id = $1', [ref]);++ assert.strictEqual(row.method, 'POST');++ assert.strictEqual(Buffer.from(row.body).toString(), body);++ const names = row.headers.map(([k]) => k);++ assert.ok(names.includes('x-github-event') && names.includes('x-hub-signature-256') && names.includes('content-type'), names.join(','));++ assert.ok(!names.includes('cookie') && !names.includes('x-forwarded-for') && !names.includes('authorization'), names.join(','));++ const out = await t.ctx.s.db.many("SELECT envelope FROM service_outbox WHERE envelope->>'event_type' = 'connect.hook.received'");++ assert.strictEqual(out.length, 1);++ envelope = out[0].envelope;++ assert.deepStrictEqual(Object.keys(envelope.payload).sort(), ['body_ref', 'content_type', 'hook_id', 'method', 'received_at', 'share_id', 'size']);++ assert.strictEqual(envelope.payload.body_ref, ref);++ assert.ok(!JSON.stringify(envelope).includes('refs/heads/main'), 'no body in the event');++ const v = contracts.validate('events.event-envelope@1', envelope);++ assert.ok(v.valid !== false, JSON.stringify(v.errors || v));++ });++++ await check('the address takes POST, PUT and PATCH only; an unknown address is 404; more than 1 MB is refused', async () => {++ assert.strictEqual((await t.get(`/h/${hook.id}`)).status, 405);++ assert.strictEqual((await t.get('/h/hk_01K9ZZZZZZZZZZZZZZZZZZZZZZ', { method: 'POST', body: '{}' })).status, 404);++ const big = await t.get(`/h/${hook.id}`, { method: 'POST', body: Buffer.alloc(1024 * 1024 + 10, 97), headers: { 'content-type': 'application/octet-stream' } });++ 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 () => {++ const off = await deliver(envelope);++ assert.strictEqual(off.status, 503, off.text);++ 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 });++ const got = app.seen.slice(before).find((x) => x.url === '/webhooks/github');++ assert.ok(got, 'the app got it');++ assert.strictEqual(got.method, 'POST');++ assert.strictEqual(got.body, JSON.stringify({ ref: 'refs/heads/main', commits: 3 }));++ 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.cookie, undefined);++ const again = await deliver(envelope);++ assert.strictEqual(again.status, 200);++ assert.deepStrictEqual(again.json(), { gone: true }, 'delivered once');++ assert.strictEqual(app.seen.filter((x) => x.url === '/webhooks/github').length, 1);++ const kept = await t.ctx.s.db.one('SELECT length(body)::int AS n, status FROM hook_requests WHERE id = $1', [envelope.payload.body_ref]);++ assert.deepStrictEqual([kept.n, kept.status], [0, 200], 'its body is dropped once delivered');++ const h = (await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex })).json().hooks[0];++ assert.deepStrictEqual([h.received, h.delivered, h.last_status], [1, 1, 200]);++ c.stop();++ });++++ await check('a delivery must be signed, from this machine; another event type is acknowledged and ignored', async () => {++ assert.strictEqual((await deliver(envelope, 'not-the-secret-0123456789abcdefgh')).status, 401);++ const other = await deliver({ ...envelope, event_type: 'connect.something.else' });++ assert.deepStrictEqual(other.json(), { ignored: true });++ });++++ await check('removing the address: it answers 404 and its waiting webhooks are dropped', async () => {++ const w = await t.get(`/h/${hook.id}`, { method: 'POST', body: '{"n":2}', headers: { 'content-type': 'application/json' } });++ assert.strictEqual(w.status, 202);++ const r = await t.get(`/api/v1/shares/${made.share.id}/hooks/${hook.id}`, { as: alex, method: 'DELETE' });++ assert.strictEqual(r.status, 200, r.text);++ assert.strictEqual((await t.get(`/h/${hook.id}`, { method: 'POST', body: '{}' })).status, 404);++ const left = await t.ctx.s.db.one('SELECT count(*)::int AS n FROM hook_requests WHERE hook_id = $1 AND delivered_at IS NULL', [hook.id]);++ assert.strictEqual(left.n, 0);++ });++++ await check('the sweep removes webhooks past their 7 days', async () => {++ const h2 = (await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: '/x' } })).json().hook;++ const ref = (await t.get(`/h/${h2.id}`, { method: 'POST', body: 'old' })).json().id;++ await t.ctx.s.db.exec("UPDATE hook_requests SET expires_at = now() - interval '1 minute' WHERE id = $1", [ref]);++ await t.ctx.expireSweep();++ assert.strictEqual((await t.ctx.s.db.maybe('SELECT id FROM hook_requests WHERE id = $1', [ref])), null);++ });++++ 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);++ assert.match(r.text, /Webhook address made/);++ assert.match(r.text, /localhost:\d+\/hooks\/stripe · Stripe/);++ });++++ await t.close();++ await app.close();++ done();++})().catch((e) => { console.error(e); process.exit(1); });
@@ -39,6 +39,28 @@ seconds, one use). It sends a fresh node token before the old one runs out, so a
@@ -39,6 +39,28 @@ seconds, one use). It sends a fresh node token before the old one runs out, so a
serving within about five minutes. While a computer serves a share, the share's key does not work; "Use a connect key
serving within about five minutes. While a computer serves a share, the share's key does not work; "Use a connect key
instead" lets the computer go and shows a new key once (ADR-057 decision 4, CON-4).
instead" lets the computer go and shows a new key once (ADR-057 decision 4, CON-4).
+### Webhooks that wait for your computer
+
+On a share's page, "Make a webhook address" gives you `https://connect.openvibe.website/h/hk_…` and a path on your app
+(`/webhooks/github`). Give that address to GitHub, Stripe or anything else that sends webhooks. Each webhook is kept for
+up to 7 days and handed to your app, at that path, as soon as your computer is connected, so a laptop that sleeps
+misses nothing (CON-15, ADR-057 decision 7):
+
+- The address takes POST, PUT and PATCH, up to 1 MB and 120 a minute, and answers `202` at once.
+- 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
+ 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
+ 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.
+- Nobody signs in to send a webhook, so check the provider's own signature in your app.
+
+| Call | What |
+|---|---|
+| `GET /api/v1/shares/:id/hooks` | its webhook addresses, with `received`, `delivered` and `waiting` |
+| `POST /api/v1/shares/:id/hooks` `{ path, label? }` | a new address → `201 { hook }` (at most 10 a share) |
+| `DELETE /api/v1/shares/:id/hooks/:hook` | remove it; webhooks still waiting are dropped |
+
## What your app sees
## What your app sees
- `Host: localhost:<port>`; the share's address in `X-Forwarded-Host`; `X-Forwarded-Proto: https`.
- `Host: localhost:<port>`; the share's address in `X-Forwarded-Host`; `X-Forwarded-Proto: https`.
@@ -104,6 +126,7 @@ browser ──https──▶ Cloudflare ──▶ nginx ──▶ Connect (5240)
@@ -104,6 +126,7 @@ browser ──https──▶ Cloudflare ──▶ nginx ──▶ Connect (5240)
no bodies; a restart forgets them.
no bodies; a restart forgets them.
- `server/auth/node-auth.js` — a computer as OpenVibe Node: its node token, and the single-use tunnel tokens (CON-4).
- `server/auth/node-auth.js` — a computer as OpenVibe Node: its node token, and the single-use tunnel tokens (CON-4).
- `server/machines.js` — the owner's computers, read from Network with the owner's own token.
- `server/machines.js` — the owner's computers, read from Network with the owner's own token.
+- `server/hooks.js`, `server/http/hooks.js` — webhook addresses, the kept webhooks, and Events' deliveries (CON-15).
- `public/ov-connect.js` — the client.
- `public/ov-connect.js` — the client.
The v0 transport is a WebSocket on 443 through Cloudflare (ADR-057's fallback path), so nothing needs a UDP port or a
The v0 transport is a WebSocket on 443 through Cloudflare (ADR-057's fallback path), so nothing needs a UDP port or a
new file mode 100644
new file mode 100644
@@ -0,0 +1,55 @@
@@ -0,0 +1,55 @@
+-- phase: expand
+-- Hooks that wait for your computer (CON-15, ADR-057 decision 7). A hook is one address,
+-- https://connect.openvibe.website/h/<id>, that a provider (GitHub, Stripe, …) sends webhooks to. Connect keeps each
+-- request here for 7 days and publishes only a pointer to it (connect.hook.received, contracts 0.282.0) through the
+-- outbox below. Events retries Connect's own subscription for up to seven days, and Connect hands the request to the
+-- share's computer, at the hook's local path, once it is connected.
+CREATE TABLE IF NOT EXISTS share_hooks (
+ id text PRIMARY KEY, -- hk_<ulid>
+ share_id text NOT NULL REFERENCES shares(id) ON DELETE CASCADE,
+ owner_subject text NOT NULL, -- usr_… (the share's owner)
+ label text,
+ local_path text NOT NULL, -- where on the computer's app it goes: /webhooks/github
+ status text NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'removed')),
+ created_at timestamptz NOT NULL DEFAULT now(),
+ removed_at timestamptz,
+ received bigint NOT NULL DEFAULT 0,
+ delivered bigint NOT NULL DEFAULT 0,
+ last_received_at timestamptz,
+ last_delivered_at timestamptz,
+ last_status integer -- what the computer's app answered last
+);
+CREATE INDEX IF NOT EXISTS share_hooks_share ON share_hooks (share_id, created_at DESC);
+
+-- One webhook as it arrived: its method, the provider's headers (an allowlist; never cookies or forwarding headers),
+-- and its body, until it is delivered or 7 days pass.
+CREATE TABLE IF NOT EXISTS hook_requests (
+ id text PRIMARY KEY, -- hb_<random>: the body_ref in the event
+ hook_id text NOT NULL REFERENCES share_hooks(id) ON DELETE CASCADE,
+ method text NOT NULL,
+ headers jsonb NOT NULL DEFAULT '[]'::jsonb,
+ body bytea NOT NULL,
+ received_at timestamptz NOT NULL DEFAULT now(),
+ expires_at timestamptz NOT NULL,
+ delivered_at timestamptz,
+ status integer
+);
+CREATE INDEX IF NOT EXISTS hook_requests_expiry ON hook_requests (expires_at);
+CREATE INDEX IF NOT EXISTS hook_requests_hook ON hook_requests (hook_id, received_at DESC);
+
+-- The events outbox (openvibe-sdk/events createServiceOutbox): a request and its event commit together.
+CREATE TABLE IF NOT EXISTS service_outbox (
+ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+ event_id text NOT NULL UNIQUE,
+ envelope jsonb NOT NULL,
+ traceparent text,
+ created_at bigint NOT NULL,
+ attempts integer NOT NULL DEFAULT 0,
+ next_attempt_at bigint NOT NULL DEFAULT 0,
+ sent_at bigint,
+ seq bigint,
+ rejected_at bigint,
+ last_error text
+);
+CREATE INDEX IF NOT EXISTS service_outbox_due ON service_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
+CREATE INDEX IF NOT EXISTS service_outbox_sent ON service_outbox (sent_at) WHERE sent_at IS NOT NULL;
@@ -132,6 +132,18 @@
@@ -132,6 +132,18 @@
.cn-run { display: grid; gap: 10px; }
.cn-run { display: grid; gap: 10px; }
.cn-run h2 { margin: 0; }
.cn-run h2 { margin: 0; }
.cn-run a { color: var(--room-strong); }
.cn-run a { color: var(--room-strong); }
+/* Webhooks that wait for your computer (CON-15): one row per address, its copy box and counts; the add form. */
+.cn-hooks-sec { display: grid; gap: 10px; }
+.cn-hooks-sec h2 { margin: 0; }
+.cn-hooks-sec > p { margin: 0; color: var(--ovk-muted); }
+.cn-hooks { list-style: none; margin: 0; padding: 0; display: grid; gap: 6px; }
+.cn-hooks li { display: flex; flex-wrap: wrap; align-items: center; justify-content: space-between; gap: 8px 12px; padding: 10px 14px; border-radius: 10px; background: var(--ovk-panel); border: 1px solid var(--ovk-line); }
+.cn-hook-main { display: grid; gap: 4px; min-width: 0; flex: 1 1 260px; }
+.cn-hook-main code { overflow-wrap: anywhere; }
+.cn-hook-main small { color: var(--ovk-muted); font-size: 13px; overflow-wrap: anywhere; }
+.cn-hook-main small b { color: var(--ovk-text); }
+.cn-hook-add { display: flex; flex-wrap: wrap; align-items: end; gap: 10px 14px; }
+.cn-hook-add .cn-field { flex: 1 1 200px; }
/* Which computer serves it (CON-4): the owner's machines, one row each with its button; the bound one's commands. */
/* Which computer serves it (CON-4): the owner's machines, one row each with its button; the bound one's commands. */
.cn-node { display: grid; gap: 10px; }
.cn-node { display: grid; gap: 10px; }
.cn-node h2 { margin: 0; }
.cn-node h2 { margin: 0; }
@@ -31,6 +31,9 @@ const { createApi } = require('./http/api');
@@ -31,6 +31,9 @@ const { createApi } = require('./http/api');
const { createNodeApi } = require('./http/node-api');
const { createNodeApi } = require('./http/node-api');
const { createNodeAuth } = require('./auth/node-auth');
const { createNodeAuth } = require('./auth/node-auth');
const { createMachines } = require('./machines');
const { createMachines } = require('./machines');
+const { createHooks } = require('./hooks');
+const { createHookRoutes } = require('./http/hooks');
+const { createServiceOutbox } = require('openvibe-sdk/events');
const { createPageRoutes } = require('./http/pages');
const { createPageRoutes } = require('./http/pages');
const { createServiceReadiness } = require('./observability');
const { createServiceReadiness } = require('./observability');
const { assetVersion, setRelease } = require('./render/layout');
const { assetVersion, setRelease } = require('./render/layout');
@@ -79,7 +82,17 @@ async function createApp(opts = {}) {
@@ -79,7 +82,17 @@ async function createApp(opts = {}) {
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
: async () => { throw new Error('OV_OAUTH_CLIENT_SECRET is not set: Connect cannot answer account events'); });
: async () => { throw new Error('OV_OAUTH_CLIENT_SECRET is not set: Connect cannot answer account events'); });
- const ctx = { config, s, keys, sso, principal, shares, tunnels, gate, proxy, notifier, nodeAuth, machines, accountData, accountSend, log };
+ // Hooks that wait for your computer (CON-15): a webhook and its pointer commit together; the relay (started in
+ // index.js, never in drill mode) publishes the pointer to OpenVibe.Events with Connect's own token.
+ const outbox = opts.outbox !== undefined ? opts.outbox : createServiceOutbox({
+ db: s.db, source: 'connect', eventsUrl: config.events.url, networkInternalUrl: config.networkInternalUrl,
+ clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl, log, now: s.now,
+ table: 'service_outbox', eventTypes: ['connect.hook.received'],
+ validate: (env) => contracts.validate('events.event-envelope@1', env),
+ });
+ const hooks = createHooks(s, { config, outbox, now: s.now });
+
+ const ctx = { config, s, keys, sso, principal, shares, tunnels, gate, proxy, notifier, nodeAuth, machines, hooks, outbox, accountData, accountSend, log };
/** 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 () => {
@@ -87,6 +100,8 @@ async function createApp(opts = {}) {
@@ -87,6 +100,8 @@ async function createApp(opts = {}) {
try { due = await shares.expireDue(); } catch (err) { log.warn('[OpenVibe.Connect] expiry sweep failed:', err.message); return []; }
try { due = await shares.expireDue(); } catch (err) { log.warn('[OpenVibe.Connect] expiry sweep failed:', err.message); return []; }
due.forEach((r) => { tunnels.close(r.id, 4003, 'the share ended'); proxy.forget(r.host); proxy.inspector.forget(r.id); });
due.forEach((r) => { tunnels.close(r.id, 4003, 'the share ended'); proxy.forget(r.host); proxy.inspector.forget(r.id); });
if (due.length) log.log(`[OpenVibe.Connect] ${due.length} share(s) ended on time`);
if (due.length) log.log(`[OpenVibe.Connect] ${due.length} share(s) ended on time`);
+ // Webhooks kept past their 7 days go, delivered or not (CON-15).
+ try { const gone = await hooks.sweep(); if (gone) log.log(`[OpenVibe.Connect] ${gone} webhook(s) past 7 days removed`); } catch (err) { log.warn('[OpenVibe.Connect] webhook sweep failed:', err.message); }
return due;
return due;
};
};
if (opts.expirySweep !== false) {
if (opts.expirySweep !== false) {
@@ -168,6 +183,9 @@ async function createApp(opts = {}) {
@@ -168,6 +183,9 @@ async function createApp(opts = {}) {
app.use('/auth', sso.routes());
app.use('/auth', sso.routes());
{ const legal = require('openvibe-shared/legal'); app.get(legal.PATHS, legal.handler({ id: 'connect', service: 'connect', host: 'connect.openvibe.website', name: 'OpenVibe Connect', profile: 'ugc' })); }
{ const legal = require('openvibe-shared/legal'); app.get(legal.PATHS, legal.handler({ id: 'connect', service: 'connect', host: 'connect.openvibe.website', name: 'OpenVibe Connect', profile: 'ugc' })); }
+ // ── Webhooks to a share's hooks, and Events' deliveries of them back (CON-15; /internal/ is loopback only) ──
+ app.use(createHookRoutes(ctx));
+
// ── OpenVibe.Events → this service (loopback only; nginx answers 404 for /internal/) ──
// ── OpenVibe.Events → this service (loopback only; nginx answers 404 for /internal/) ──
app.use('/internal/events', accountData.consumer({
app.use('/internal/events', accountData.consumer({
secrets: config.events.secrets,
secrets: config.events.secrets,
@@ -382,7 +382,43 @@ function createProxy({ config, shares, tunnels, gate, log = console, now = () =>
@@ -382,7 +382,43 @@ function createProxy({ config, shares, tunnels, gate, log = console, now = () =>
/** The grant refused someone: counted on the share page as "turned away". */
/** The grant refused someone: counted on the share page as "turned away". */
function denied(share, who) { visit(share, who, { denied: true }); }
function denied(share, who) { visit(share, who, { denied: true }); }
- return { handle, upgrade, isShareHost, flush, forget, denied, stop, responseHeaders, forwardHeaders, inspector };
+ /**
+ * One stored webhook to the share's computer (CON-15): the request as it arrived, at the hook's path, with
+ * x-openvibe-hook and x-openvibe-event-id so the app can tell and dedupe. The answer's body is read and dropped.
+ * → { status } when the app answered, { offline: true } with no computer, { error } when it broke on the way.
+ */
+ function deliverHook(share, { method, path, headers = [], body, hookId, eventId }) {
+ const tunnel = tunnels.get(share.id);
+ if (!tunnel) return Promise.resolve({ offline: true });
+ const port = (tunnel.clientInfo && tunnel.clientInfo.port) || share.local_port;
+ const rec = inspector.begin(share.id, { method, path, who: { username: 'webhook' } });
+ return new Promise((resolve) => {
+ let done = false;
+ const finish = (out) => { if (done) return; done = true; clearTimeout(timer); inspector.end(rec, out.status ? { status: out.status } : { error: out.error || 'offline' }); resolve(out); };
+ let status = 0;
+ const id = tunnel.open({
+ onRes(m) { status = Number(m.status) || 0; },
+ onData() {},
+ onEnd() { tunnel.release(id); finish(status >= 100 && status <= 599 ? { status } : { error: 'bad_status' }); },
+ onAbort(m) { finish({ error: (m && m.code) || 'local_error' }); },
+ onPause() {}, onResume() {},
+ });
+ const timer = setTimeout(() => { tunnel.abort(id, 'timeout'); finish({ error: 'timeout' }); }, config.shares.requestTimeoutMs);
+ timer.unref();
+ const out = (Array.isArray(headers) ? headers : []).filter((h) => Array.isArray(h) && h.length === 2 && !/^(host|content-length|x-openvibe-)/i.test(String(h[0])));
+ out.push(['host', `localhost:${port}`], ['x-forwarded-host', share.host], ['x-forwarded-proto', config.shares.scheme],
+ ['x-openvibe-hook', String(hookId || '')], ['x-openvibe-event-id', String(eventId || '')], ['content-length', String(body ? body.length : 0)]);
+ tunnel.send({ t: 'req', id, method, path, headers: out });
+ if (body && body.length) {
+ rec.in += body.length;
+ const CHUNK = 32 * 1024;
+ for (let i = 0; i < body.length; i += CHUNK) tunnel.data(id, body.subarray(i, i + CHUNK));
+ }
+ tunnel.end(id);
+ });
+ }
+
+ return { handle, upgrade, isShareHost, flush, forget, denied, stop, responseHeaders, forwardHeaders, deliverHook, inspector };
}
}
/** Write an HTTP answer straight to an upgrade's socket (the gate's 302/401/404/502 before any tunnel opens). */
/** Write an HTTP answer straight to an upgrade's socket (the gate's 302/401/404/502 before any tunnel opens). */
new file mode 100644
new file mode 100644
@@ -0,0 +1,142 @@
@@ -0,0 +1,142 @@
+'use strict';
+
+/**
+ * Hooks that wait for your computer (CON-15, ADR-057 decision 7). A hook is an address a provider sends webhooks to,
+ * https://connect.openvibe.website/h/<hk_…>, bound to one share and a path on the computer's app. A webhook is kept
+ * (7 days) and a pointer to it is published as connect.hook.received through the outbox, in one transaction. Events
+ * delivers it to Connect's own subscription (retried for up to seven days), and Connect hands it to the share's
+ * computer when it is connected. The computer may be asleep for days and still get every webhook, in order of arrival
+ * or not, each once it answers.
+ *
+ * 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
+ */
+const crypto = require('crypto');
+
+const HOOK_ID = /^hk_[0-9A-HJKMNP-TV-Z]{26}$/;
+const REF_RE = /^hb_[A-Za-z0-9_-]{16,64}$/;
+const PATH_RE = /^\/[\x21-\x7e]{0,499}$/;
+const METHODS = ['POST', 'PUT', 'PATCH'];
+const KEEP_MS = 7 * 86400_000;
+const MAX_BODY = 1024 * 1024;
+const PER_SHARE = 10;
+// The provider headers a webhook keeps: what apps check signatures and events with. Never cookies, authorization,
+// forwarding or Cloudflare headers, and never Connect's own.
+const DROP = /^(x-forwarded-|x-real-ip$|x-openvibe-|cf-|x-amz-cf-)/;
+const KEEP = /^(content-type|content-encoding|user-agent|stripe-signature|svix-.+|webhook-.+|x-.+)$/;
+
+class HookError extends Error {
+ constructor(status, code, message) { super(message); this.status = status; this.code = code; }
+}
+
+/** The headers a stored webhook keeps: at most 40, each at most 2 KB. → [[name, value]] */
+function keepHeaders(headers) {
+ const out = [];
+ for (const [k, v] of Object.entries(headers || {})) {
+ const name = String(k).toLowerCase();
+ if (DROP.test(name) || !KEEP.test(name)) continue;
+ const value = Array.isArray(v) ? v.join(', ') : String(v);
+ if (value.length > 2048) continue;
+ out.push([name, value]);
+ if (out.length >= 40) break;
+ }
+ return out;
+}
+
+function createHooks(s, { config, outbox = null, now = () => Date.now() }) {
+ const { db } = s;
+ const base = config.baseUrl;
+ const iso = (v) => (v ? new Date(v).toISOString() : null);
+ const view = (r) => r && ({
+ id: r.id, share_id: r.share_id, url: `${base}/h/${r.id}`, local_path: r.local_path, label: r.label || null, status: r.status,
+ created_at: iso(r.created_at), received: Number(r.received || 0), delivered: Number(r.delivered || 0),
+ last_received_at: iso(r.last_received_at), last_delivered_at: iso(r.last_delivered_at), last_status: r.last_status == null ? null : Number(r.last_status),
+ waiting: r.waiting == null ? undefined : Number(r.waiting),
+ });
+
+ async function get(id) {
+ if (!HOOK_ID.test(String(id || ''))) return null;
+ return view(await db.maybe('SELECT * FROM share_hooks WHERE id = $1', [id]));
+ }
+
+ async function list(shareId) {
+ const rows = await db.many(`SELECT h.*, (SELECT count(*)::int FROM hook_requests r WHERE r.hook_id = h.id AND r.delivered_at IS NULL AND r.expires_at > now()) AS waiting
+ FROM share_hooks h WHERE h.share_id = $1 AND h.status = 'active' ORDER BY h.created_at`, [shareId]);
+ return rows.map(view);
+ }
+
+ async function create(owner, share, input = {}) {
+ if (!share || share.owner_subject !== owner.subject) throw new HookError(404, 'share.not_found', 'No such share.');
+ if (share.status !== 'active') throw new HookError(410, 'share.revoked', 'That share is closed.');
+ const localPath = String(input.path || '').trim() || '/';
+ if (!PATH_RE.test(localPath)) throw new HookError(422, 'hook.invalid_path', 'The path on your app starts with "/" and has no spaces, like /webhooks/github.');
+ const label = input.label ? String(input.label).trim().slice(0, 80) || null : null;
+ const n = await db.one("SELECT count(*)::int AS n FROM share_hooks WHERE share_id = $1 AND status = 'active'", [share.id]);
+ if (n.n >= PER_SHARE) throw new HookError(409, 'hook.limit', `A share can have ${PER_SHARE} webhook addresses. Remove one first.`);
+ const id = s.newId('hk');
+ await db.exec('INSERT INTO share_hooks (id, share_id, owner_subject, label, local_path) VALUES ($1, $2, $3, $4, $5)', [id, share.id, owner.subject, label, localPath]);
+ return get(id);
+ }
+
+ async function remove(owner, id) {
+ const hook = await db.maybe('SELECT * FROM share_hooks WHERE id = $1', [String(id || '')]);
+ if (!hook || hook.owner_subject !== owner.subject) throw new HookError(404, 'hook.not_found', 'No such webhook address.');
+ await db.exec("UPDATE share_hooks SET status = 'removed', removed_at = now() WHERE id = $1", [hook.id]);
+ await db.exec('DELETE FROM hook_requests WHERE hook_id = $1 AND delivered_at IS NULL', [hook.id]);
+ return view({ ...hook, status: 'removed' });
+ }
+
+ /** A webhook arrived: kept, and its pointer published, together. → { ref } */
+ async function receive(hook, { method, headers, body }) {
+ if (!METHODS.includes(method)) throw new HookError(405, 'hook.method', 'A webhook address takes POST, PUT or PATCH.');
+ const buf = Buffer.isBuffer(body) ? body : Buffer.from(body == null ? '' : String(body));
+ if (buf.length > MAX_BODY) throw new HookError(413, 'hook.too_large', 'A webhook is at most 1 MB.');
+ const kept = keepHeaders(headers);
+ const contentType = (kept.find(([k]) => k === 'content-type') || [null, null])[1];
+ const ref = `hb_${crypto.randomBytes(18).toString('base64url')}`;
+ const at = new Date(now());
+ 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('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',
+ 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() },
+ });
+ }
+ });
+ if (outbox && outbox.kick) outbox.kick().catch(() => {});
+ return { ref };
+ }
+
+ /** The stored request a delivery points at, with its hook; null when it was delivered, removed or expired. */
+ async function request(ref) {
+ if (!REF_RE.test(String(ref || ''))) return null;
+ 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]);
+ 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 };
+ }
+
+ /** The computer's app answered: the request is delivered (its body is dropped) and the hook counts it. */
+ async function settle(ref, status) {
+ const at = new Date(now()).toISOString();
+ const r = await db.maybe("UPDATE hook_requests SET delivered_at = $2, status = $3, body = ''::bytea WHERE id = $1 AND delivered_at IS NULL RETURNING hook_id", [ref, at, status]);
+ if (r) await db.exec('UPDATE share_hooks SET delivered = delivered + 1, last_delivered_at = $2, last_status = $3 WHERE id = $1', [r.hook_id, at, status]);
+ return Boolean(r);
+ }
+
+ /** Requests older than 7 days go, delivered or not. */
+ async function sweep() {
+ const r = await db.many('DELETE FROM hook_requests WHERE expires_at <= $1 RETURNING id', [new Date(now()).toISOString()]);
+ return r.length;
+ }
+
+ return { get, list, create, remove, receive, request, settle, sweep, KEEP_MS, MAX_BODY };
+}
+
+module.exports = { createHooks, HookError, keepHeaders, HOOK_ID, METHODS, MAX_BODY, KEEP_MS };
@@ -14,6 +14,9 @@
@@ -14,6 +14,9 @@
* PUT /shares/:id/node { node: 'nod_…' } one of your computers serves it, as OpenVibe Node (CON-4): the key
* PUT /shares/:id/node { node: 'nod_…' } one of your computers serves it, as OpenVibe Node (CON-4): the key
* stops working → { share }; { node: null } lets the computer go → { share, key } (a new key)
* stops working → { share }; { node: null } lets the computer go → { share, key } (a new key)
* GET /machines your computers that can serve a share (Network's list of your machines)
* GET /machines your computers that can serve a share (Network's list of your machines)
+ * GET /shares/:id/hooks its webhook addresses (CON-15): each kept webhook waits up to 7 days for the computer
+ * POST /shares/:id/hooks { path, label? } → 201 { hook }: https://connect.openvibe.website/h/<id> → that path
+ * DELETE /shares/:id/hooks/:hook removes it; webhooks still waiting are dropped
* GET /shares/:id/visits who opened it in the last 7 days
* GET /shares/:id/visits who opened it in the last 7 days
* GET /shares/:id/requests the last 100 requests through it, newest first (?after=<seq>: only newer ones);
* GET /shares/:id/requests the last 100 requests through it, newest first (?after=<seq>: only newer ones);
* memory only, query values shown as "…" (edge/inspector.js)
* memory only, query values shown as "…" (edge/inspector.js)
@@ -24,9 +27,10 @@ const express = require('express');
@@ -24,9 +27,10 @@ const express = require('express');
const { http } = require('openvibe-contracts');
const { http } = require('openvibe-contracts');
const { asyncRouter } = require('./router');
const { asyncRouter } = require('./router');
const { ShareError } = require('../shares');
const { ShareError } = require('../shares');
+const { HookError } = require('../hooks');
function createApi(ctx) {
function createApi(ctx) {
- const { shares, tunnels, principal, proxy, machines } = ctx;
+ const { shares, tunnels, principal, proxy, machines, hooks } = ctx;
const r = asyncRouter();
const r = asyncRouter();
r.use(express.json({ limit: '16kb' }));
r.use(express.json({ limit: '16kb' }));
r.use((req, res, next) => { res.set('Cache-Control', 'no-store'); next(); });
r.use((req, res, next) => { res.set('Cache-Control', 'no-store'); next(); });
@@ -38,7 +42,7 @@ function createApi(ctx) {
@@ -38,7 +42,7 @@ function createApi(ctx) {
const shape = (s) => ({ ...s, connected: tunnels.isConnected(s.id) });
const shape = (s) => ({ ...s, connected: tunnels.isConnected(s.id) });
const guard = (fn) => async (req, res) => {
const guard = (fn) => async (req, res) => {
try { return await fn(req, res); } catch (err) {
try { return await fn(req, res); } catch (err) {
- if (err instanceof ShareError) return fail(res, req, err.status, err.code, err.message);
+ if (err instanceof ShareError || err instanceof HookError) return fail(res, req, err.status, err.code, err.message);
throw err;
throw err;
}
}
};
};
@@ -114,6 +118,24 @@ function createApi(ctx) {
@@ -114,6 +118,24 @@ function createApi(ctx) {
return res.json({ share: shape(updated) });
return res.json({ share: shape(updated) });
}));
}));
+ r.get('/shares/:id/hooks', guard(async (req, res) => {
+ const share = await mine(req, res);
+ return share ? res.json({ hooks: await hooks.list(share.id) }) : undefined;
+ }));
+ r.post('/shares/:id/hooks', guard(async (req, res) => {
+ const share = await mine(req, res);
+ if (!share) return undefined;
+ const b = req.body || {};
+ return res.status(201).json({ hook: await hooks.create(owner(req), share, { path: b.path, label: b.label }) });
+ }));
+ r.delete('/shares/:id/hooks/:hook', guard(async (req, res) => {
+ const share = await mine(req, res);
+ if (!share) return undefined;
+ const hook = await hooks.get(req.params.hook);
+ if (!hook || hook.share_id !== share.id) return fail(res, req, 404, 'hook.not_found', 'No such webhook address.');
+ return res.json({ hook: await hooks.remove(owner(req), hook.id) });
+ }));
+
r.get('/shares/:id/requests', guard(async (req, res) => {
r.get('/shares/:id/requests', guard(async (req, res) => {
const share = await mine(req, res);
const share = await mine(req, res);
return share ? res.json({ requests: proxy.inspector.list(share.id, { after: req.query.after }), keep: proxy.inspector.KEEP }) : undefined;
return share ? res.json({ requests: proxy.inspector.list(share.id, { after: req.query.after }), keep: proxy.inspector.KEEP }) : undefined;
new file mode 100644
new file mode 100644
@@ -0,0 +1,74 @@
@@ -0,0 +1,74 @@
+'use strict';
+
+/**
+ * Hooks that wait for your computer (CON-15): the address providers send to, and Events' deliveries back.
+ *
+ * POST|PUT|PATCH /h/:id a provider's webhook (1 MB): kept 7 days, its pointer published → 202
+ * POST /internal/hook-deliveries Events, signed v2, loopback only (nginx answers 404 for /internal/): the
+ * webhook goes to the share's computer. 200 once its app answered (any status
+ * below 500) or when the webhook is gone; 503 while the computer is offline or
+ * its app fails, so Events tries again (up to seven days, server/index.js).
+ */
+const express = require('express');
+const { parseDelivery } = require('openvibe-sdk/events');
+const { HookError, METHODS, MAX_BODY } = require('../hooks');
+
+const LOOPBACK = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']);
+const PER_HOOK_MINUTE = 120;
+
+function createHookRoutes(ctx) {
+ const { config, hooks, shares, proxy, log } = ctx;
+ const r = express.Router();
+ const counts = new Map(); // hook id → { minute, n }
+
+ const tooMany = (id) => {
+ const minute = Math.floor(Date.now() / 60_000);
+ const c = counts.get(id);
+ if (!c || c.minute !== minute) { counts.set(id, { minute, n: 1 }); if (counts.size > 5000) counts.delete(counts.keys().next().value); return false; }
+ c.n += 1;
+ return c.n > PER_HOOK_MINUTE;
+ };
+ const answer = (res, status, body) => res.status(status).set('Cache-Control', 'no-store').json(body);
+
+ r.all('/h/:id', express.raw({ type: () => true, limit: MAX_BODY }), async (req, res) => {
+ if (!METHODS.includes(req.method)) return answer(res.set('Allow', METHODS.join(', ')), 405, { error: 'method', detail: 'A webhook address takes POST, PUT or PATCH.' });
+ const hook = await hooks.get(req.params.id);
+ if (!hook || hook.status !== 'active') return answer(res, 404, { error: 'not_found', detail: 'No webhook address here.' });
+ const share = await shares.get(hook.share_id);
+ if (!share || share.status !== 'active') return answer(res, 410, { error: 'gone', detail: 'The share this address served is closed.' });
+ if (tooMany(hook.id)) return answer(res.set('Retry-After', '60'), 429, { error: 'rate_limited', detail: `At most ${PER_HOOK_MINUTE} webhooks a minute to one address.` });
+ try {
+ const { ref } = await hooks.receive(hook, { method: req.method, headers: req.headers, body: Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0) });
+ return answer(res, 202, { received: true, id: ref, delivered_when: 'the computer is connected (kept up to 7 days)' });
+ } catch (err) {
+ if (err instanceof HookError) return answer(res, err.status, { error: err.code, detail: err.message });
+ log.warn('[OpenVibe.Connect] webhook not kept:', err && err.message);
+ return answer(res.set('Retry-After', '30'), 503, { error: 'unavailable', detail: 'Try again shortly.' });
+ }
+ });
+
+ r.post('/internal/hook-deliveries', express.raw({ type: () => true, limit: '64kb' }), async (req, res) => {
+ if (!LOOPBACK.has(String(req.socket.remoteAddress || ''))) return res.status(404).end();
+ let d = null;
+ for (const secret of config.events.secrets || []) {
+ if (secret) d = parseDelivery(req.body, req.headers, secret);
+ if (d) break;
+ }
+ if (!d) return res.status(401).json({ error: 'bad_signature' });
+ const e = d.event || {};
+ if (e.event_type !== 'connect.hook.received' || !e.payload) return res.status(200).json({ ignored: true });
+ const stored = await hooks.request(e.payload.body_ref);
+ if (!stored) return res.status(200).json({ gone: true });
+ const share = await shares.get(stored.shareId);
+ if (!share || share.status !== 'active') return res.status(200).json({ gone: true });
+ const out = await proxy.deliverHook(share, { method: stored.method, path: stored.path, headers: stored.headers, body: stored.body, hookId: stored.hookId, eventId: e.event_id });
+ if (out.offline) return res.status(503).set('Retry-After', '300').json({ waiting: 'the computer is not connected' });
+ if (!out.status || out.status >= 500) return res.status(503).json({ retry: out.error || `the app answered ${out.status}` });
+ await hooks.settle(stored.ref, out.status);
+ return res.status(200).json({ delivered: true, status: out.status });
+ });
+
+ return r;
+}
+
+module.exports = { createHookRoutes };
@@ -10,6 +10,8 @@
@@ -10,6 +10,8 @@
* POST /shares/:id/key a new key (shown once); the connected computer is disconnected
* POST /shares/:id/key a new key (shown once); the connected computer is disconnected
* POST /shares/:id/node node=nod_… one of your computers serves it (CON-4); node= (empty) lets it go and shows a
* POST /shares/:id/node node=nod_… one of your computers serves it (CON-4); node= (empty) lets it go and shows a
* new key once
* new key once
+ * POST /shares/:id/hooks path=&label= a webhook address for it (CON-15)
+ * POST /shares/:id/hooks/:hook/remove
* POST /shares/:id/close close it: the address answers "closed" from then on
* POST /shares/:id/close close it: the address answers "closed" from then on
* /grant?share=&next= the gate's sign-in step (server/edge/gate.js): signed in and allowed → back to the
* /grant?share=&next= the gate's sign-in step (server/edge/gate.js): signed in and allowed → back to the
* share host with a one-time code
* share host with a one-time code
@@ -21,9 +23,10 @@ const { sameOrigin } = require('./principal');
@@ -21,9 +23,10 @@ const { sameOrigin } = require('./principal');
const { send } = require('../render/layout');
const { send } = require('../render/layout');
const views = require('../views');
const views = require('../views');
const { ShareError } = require('../shares');
const { ShareError } = require('../shares');
+const { HookError } = require('../hooks');
function createPageRoutes(ctx) {
function createPageRoutes(ctx) {
- const { config, shares, tunnels, gate, proxy, machines } = ctx;
+ const { config, shares, tunnels, gate, proxy, machines, hooks } = ctx;
const r = asyncRouter();
const r = asyncRouter();
r.use(express.urlencoded({ extended: false, limit: '16kb' }));
r.use(express.urlencoded({ extended: false, limit: '16kb' }));
const page = (req, res, o, status = 200) => send(res, status, { viewer: req.viewer, config, path: req.originalUrl, ...o });
const page = (req, res, o, status = 200) => send(res, status, { viewer: req.viewer, config, path: req.originalUrl, ...o });
@@ -82,11 +85,12 @@ function createPageRoutes(ctx) {
@@ -82,11 +85,12 @@ function createPageRoutes(ctx) {
async function showShare(req, res, share, extra = {}, status = 200) {
async function showShare(req, res, share, extra = {}, status = 200) {
const tunnel = tunnels.get(share.id);
const tunnel = tunnels.get(share.id);
- const [visits, mine] = await Promise.all([
+ const [visits, mine, hookList] = await Promise.all([
shares.visits(share.id),
shares.visits(share.id),
share.status === 'active' ? machines.list(req.viewer.token) : Promise.resolve({ ok: true, machines: [] }),
share.status === 'active' ? machines.list(req.viewer.token) : Promise.resolve({ ok: true, machines: [] }),
+ share.status === 'active' ? hooks.list(share.id) : Promise.resolve([]),
]);
]);
- return page(req, res, { title: share.host, body: views.sharePage({ share, connected: Boolean(tunnel), tunnel, visits, requests: ctx.proxy ? ctx.proxy.inspector.list(share.id) : [], baseUrl: config.baseUrl, staff: req.viewer.role === 'admin', machines: mine.ok ? mine.machines : null, ...extra }) }, status);
+ return page(req, res, { title: share.host, body: views.sharePage({ share, connected: Boolean(tunnel), tunnel, visits, requests: ctx.proxy ? ctx.proxy.inspector.list(share.id) : [], baseUrl: config.baseUrl, staff: req.viewer.role === 'admin', machines: mine.ok ? mine.machines : null, hooks: hookList, ...extra }) }, status);
}
}
r.get('/shares/:id', async (req, res) => {
r.get('/shares/:id', async (req, res) => {
@@ -147,6 +151,29 @@ function createPageRoutes(ctx) {
@@ -147,6 +151,29 @@ function createPageRoutes(ctx) {
}
}
});
});
+ r.post('/shares/:id/hooks', async (req, res) => {
+ if (!guard(req, res)) return undefined;
+ const share = await mine(req, res);
+ if (!share) return undefined;
+ try {
+ const hook = await hooks.create(owner(req), share, { path: req.body.path, label: req.body.label });
+ return showShare(req, res, share, { notice: `Webhook address made. Give ${hook.url} to the service that sends them.` }, 201);
+ } catch (err) {
+ if (!(err instanceof HookError)) throw err;
+ return showShare(req, res, share, { error: err.message }, err.status || 422);
+ }
+ });
+
+ r.post('/shares/:id/hooks/:hook/remove', async (req, res) => {
+ if (!guard(req, res)) return undefined;
+ const share = await mine(req, res);
+ if (!share) return undefined;
+ const hook = await hooks.get(req.params.hook);
+ if (!hook || hook.share_id !== share.id) return showShare(req, res, share, { error: 'No such webhook address.' }, 404);
+ await hooks.remove(owner(req), hook.id);
+ return showShare(req, res, share, { notice: 'Removed. That address answers "not found" from now on.' });
+ });
+
r.post('/shares/:id/close', async (req, res) => {
r.post('/shares/:id/close', async (req, res) => {
if (!guard(req, res)) return undefined;
if (!guard(req, res)) return undefined;
const share = await mine(req, res);
const share = await mine(req, res);
@@ -18,10 +18,14 @@ const TABLES = [
@@ -18,10 +18,14 @@ const TABLES = [
// the account is taken off it.
// the account is taken off it.
{ table: 'share_reports', subject: 'reporter_subject', value: (usr) => usr, file: 'reports.json', erase: { anonymize: {} },
{ table: 'share_reports', subject: 'reporter_subject', value: (usr) => usr, file: 'reports.json', erase: { anonymize: {} },
columns: ['id', 'host', 'reason', 'details', 'status', 'created_at'] },
columns: ['id', 'host', 'reason', 'details', 'status', 'created_at'] },
+ // Webhook addresses on a person's shares (CON-15). Their waiting webhooks (hook_requests) go with them, and with
+ // the shares (ON DELETE CASCADE); the export names the addresses and their counts, never a webhook's body.
+ { table: 'share_hooks', subject: 'owner_subject', value: (usr) => usr, file: 'webhooks.json', erase: 'delete',
+ columns: ['id', 'share_id', 'label', 'local_path', 'status', 'created_at', 'removed_at', 'received', 'delivered', 'last_received_at', 'last_delivered_at', 'last_status'] },
{
{
table: 'shares', subject: 'owner_subject', value: (usr) => usr, file: 'shares.json', erase: 'delete',
table: 'shares', subject: 'owner_subject', value: (usr) => usr, file: 'shares.json', erase: 'delete',
columns: ['id', 'owner_username', 'label', 'host', 'local_port', 'mode', 'named', 'status', 'policy_version', 'key_prefix', 'note',
columns: ['id', 'owner_username', 'label', 'host', 'local_port', 'mode', 'named', 'status', 'policy_version', 'key_prefix', 'note',
- 'created_at', 'updated_at', 'revoked_at', 'last_connected_at', 'last_visit_at', 'requests', 'bytes_in', 'bytes_out'],
+ 'node_id', 'node_name', 'created_at', 'updated_at', 'revoked_at', 'last_connected_at', 'last_visit_at', 'requests', 'bytes_in', 'bytes_out'],
},
},
];
];
@@ -30,7 +34,7 @@ function create({ db, log = console } = {}) {
@@ -30,7 +34,7 @@ function create({ db, log = console } = {}) {
db,
db,
service: 'connect',
service: 'connect',
tables: TABLES,
tables: TABLES,
- note: 'Your shares on OpenVibe Connect and your visits to other people\'s shares. Connect never stores what passed through a share.',
+ note: 'Your shares on OpenVibe Connect, their webhook addresses, and your visits to other people\'s shares. Connect never stores what passed through a share; a webhook to one of your addresses is kept only until your computer takes it, at most 7 days.',
log,
log,
});
});
}
}
@@ -60,13 +60,47 @@ async function start() {
@@ -60,13 +60,47 @@ async function start() {
clientSecret: config.oauth.clientSecret,
clientSecret: config.oauth.clientSecret,
log: ctx.log || console,
log: ctx.log || console,
});
});
- const extra = [() => { if (subs) subs.stop(); }];
+ // Hooks (CON-15): the outbox relay publishes webhook pointers, and Connect's own subscription brings them back to
+ // /internal/hook-deliveries, retried for about seven days while a computer is offline. Off in drill mode and
+ // without the events URL, the delivery secret and the client secret.
+ const hookSubs = config.drill ? null : ensureHookSubscription({ config, log: ctx.log || console });
+ if (!config.drill && ctx.outbox) ctx.outbox.start();
+ const extra = [() => { if (subs) subs.stop(); }, () => { if (hookSubs) hookSubs.stop(); }, () => { if (ctx.outbox) ctx.outbox.stop(); }];
createLifecycle({ server, ctx, extra });
createLifecycle({ server, ctx, extra });
return { server, ctx };
return { server, ctx };
}
}
+// About seven days of retries (Events allows 20 attempts, each wait at most a day): 5 s, 30 s, 2 min, 10 min, 30 min,
+// 1, 2, 3, 4, 6, 8 hours, four of 12 hours and four of a day.
+const HOOK_BACKOFF_MS = [5e3, 30e3, 120e3, 600e3, 1800e3, 3600e3, 7200e3, 10800e3, 14400e3, 21600e3, 28800e3,
+ 43200e3, 43200e3, 43200e3, 43200e3, 86400e3, 86400e3, 86400e3, 86400e3];
+
+/** Connect's subscription to its own connect.hook.received, made once (409 means it is there); tried again every 10 minutes until it is. */
+function ensureHookSubscription({ config, log = console, fetchImpl = globalThis.fetch }) {
+ const secret = config.events.secrets[0] || '';
+ if (!config.events.url || !secret || !config.oauth.clientSecret) { log.warn('[OpenVibe.Connect] webhook deliveries off: CONNECT_EVENTS_URL, CONNECT_EVENTS_SECRET or the client secret is unset'); return null; }
+ const { serviceAuth } = require('openvibe-contracts');
+ const tokens = serviceAuth.createTokenClient({ tokenUrl: `${config.networkInternalUrl}/oauth/token`, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, audience: 'openvibe.events', scope: 'events.subscription.manage', fetchImpl });
+ let timer = null;
+ let stopped = false;
+ async function once() {
+ try {
+ const res = await fetchImpl(`${config.events.url}/api/v1/subscriptions`, {
+ method: 'POST', headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...(await tokens.authHeaders()) },
+ body: JSON.stringify({ topic_pattern: 'connect.hook.received', endpoint: `http://127.0.0.1:${config.port}/internal/hook-deliveries`, secret, retry_policy: { max_attempts: 20, backoff_ms: HOOK_BACKOFF_MS } }),
+ signal: AbortSignal.timeout(10_000),
+ });
+ if (res.status === 201 || res.status === 409) { log.log(`[OpenVibe.Connect] webhook deliveries subscribed (${res.status})`); return; }
+ log.warn(`[OpenVibe.Connect] webhook subscription not made: HTTP ${res.status}`);
+ } catch (err) { log.warn(`[OpenVibe.Connect] webhook subscription not made: ${err.message}`); }
+ if (!stopped) { timer = setTimeout(once, 600_000); timer.unref(); }
+ }
+ once();
+ return { stop() { stopped = true; clearTimeout(timer); } };
+}
+
if (require.main === module) {
if (require.main === module) {
start().catch((err) => { console.error('[OpenVibe.Connect] failed to start:', err); process.exit(1); });
start().catch((err) => { console.error('[OpenVibe.Connect] failed to start:', err); process.exit(1); });
}
}
-module.exports = { start, createLifecycle };
+module.exports = { start, createLifecycle, ensureHookSubscription, HOOK_BACKOFF_MS };
@@ -259,7 +259,26 @@ function nodeBlock(share, machines, now) {
@@ -259,7 +259,26 @@ function nodeBlock(share, machines, now) {
return `<section class="cn-node" aria-labelledby="cn-node-h"><h2 id="cn-node-h">Or serve it from one of your computers</h2>${intro}${list}</section>`;
return `<section class="cn-node" aria-labelledby="cn-node-h"><h2 id="cn-node-h">Or serve it from one of your computers</h2>${intro}${list}</section>`;
}
}
-function sharePage({ share, connected, tunnel = null, visits = [], requests = [], key = null, baseUrl, notice = null, error = null, staff = false, machines = [], now = Date.now() }) {
+/**
+ * Webhooks that wait for your computer (CON-15): each address, where it goes on the computer's app, what came and what
+ * was handed over, and how many wait; the form for another. Plain forms, no script.
+ */
+function hooksBlock(share, hooks, now) {
+ const rows = hooks.length ? `<ul class="cn-hooks">${hooks.map((h, i) => `<li><div class="cn-hook-main"><div class="cn-copy"><code id="cn-hook-${i}">${esc(h.url)}</code><button type="button" class="cn-btn cn-btn-quiet" data-ovk-copy="#cn-hook-${i}" hidden>${kit.icon('copy')} Copy</button></div>
+<small>→ localhost:${esc(share.local_port)}${esc(h.local_path)}${h.label ? ` · ${esc(h.label)}` : ''} · ${esc(h.received)} received, ${esc(h.delivered)} handed over${h.waiting ? `, <b>${esc(h.waiting)} waiting for your computer</b>` : ''}${h.last_received_at ? ` · last ${esc(ago(h.last_received_at, now))}` : ''}${h.last_status ? ` · your app answered ${esc(h.last_status)}` : ''}</small></div>
+<form method="post" action="/shares/${esc(share.id)}/hooks/${esc(h.id)}/remove"><button class="cn-btn cn-btn-quiet" type="submit">${kit.icon('close')} Remove</button></form></li>`).join('')}</ul>`
+ : '<p class="cn-empty">No webhook addresses yet.</p>';
+ return `<section class="cn-hooks-sec" aria-labelledby="cn-hooks-h"><h2 id="cn-hooks-h">Webhooks that wait for your computer</h2>
+<p>Give a service (GitHub, Stripe, anything that sends webhooks) an address here instead of your share's. Each webhook is kept for up to 7 days and handed to your app, at the path you pick, as soon as your computer is connected: a laptop that sleeps misses nothing. Nobody has to sign in to send one, so check the provider's signature in your app.</p>
+${rows}
+<form class="cn-form cn-hook-add" method="post" action="/shares/${esc(share.id)}/hooks">
+<label class="cn-field"><span>Path on your app</span><input name="path" maxlength="500" value="/webhooks" placeholder="/webhooks/github" autocapitalize="none" autocomplete="off" spellcheck="false"></label>
+<label class="cn-field"><span>Name (optional)</span><input name="label" maxlength="80" placeholder="GitHub pushes"></label>
+<button class="cn-btn" type="submit">${kit.icon('plus')} Make a webhook address</button>
+</form></section>`;
+}
+
+function sharePage({ share, connected, tunnel = null, visits = [], requests = [], key = null, baseUrl, notice = null, error = null, staff = false, machines = [], hooks = [], now = Date.now() }) {
const closed = share.status !== 'active';
const closed = share.status !== 'active';
const state = closed ? 'Closed' : connected ? 'Connected' : 'Not connected';
const state = closed ? 'Closed' : connected ? 'Connected' : 'Not connected';
const stateText = closed ? `Closed ${ago(share.revoked_at, now)}. The address answers "closed" and stays yours.`
const stateText = closed ? `Closed ${ago(share.revoked_at, now)}. The address answers "closed" and stays yours.`
@@ -302,6 +321,7 @@ ${share.expires_at ? `<div><dt>${closed ? 'Ended' : 'Closes'}</dt><dd><time date
@@ -302,6 +321,7 @@ ${share.expires_at ? `<div><dt>${closed ? 'Ended' : 'Closes'}</dt><dd><time date
</dl>
</dl>
${!key && !closed && !share.node ? `<section class="cn-run"><h2>Connect it</h2><div class="cn-copy"><code id="cn-run">node ov-connect.js --key ovc_… --port ${esc(share.local_port)}</code></div><p class="cn-fine">The key was shown once, when it was made. <a href="${esc(CLIENT_URL(baseUrl))}">Download ov-connect.js</a> (Node 22 or newer).</p></section>` : ''}
${!key && !closed && !share.node ? `<section class="cn-run"><h2>Connect it</h2><div class="cn-copy"><code id="cn-run">node ov-connect.js --key ovc_… --port ${esc(share.local_port)}</code></div><p class="cn-fine">The key was shown once, when it was made. <a href="${esc(CLIENT_URL(baseUrl))}">Download ov-connect.js</a> (Node 22 or newer).</p></section>` : ''}
${closed ? '' : nodeBlock(share, machines, now)}
${closed ? '' : nodeBlock(share, machines, now)}
+${closed ? '' : hooksBlock(share, hooks, now)}
<section aria-labelledby="cn-visits-h"><h2 id="cn-visits-h">Who opened it (7 days)</h2>${visitRows}</section>
<section aria-labelledby="cn-visits-h"><h2 id="cn-visits-h">Who opened it (7 days)</h2>${visitRows}</section>
${closed ? '' : `<section class="cn-inspector" aria-labelledby="cn-reqs-h" data-cn-reqs="${esc(share.id)}"><h2 id="cn-reqs-h">Requests</h2>
${closed ? '' : `<section class="cn-inspector" aria-labelledby="cn-reqs-h" data-cn-reqs="${esc(share.id)}"><h2 id="cn-reqs-h">Requests</h2>
<p class="cn-fine">The last 100 through this share, newest first, added as they happen. Kept in memory only, never written down: no headers, no bodies, and query values show as "…".</p>
<p class="cn-fine">The last 100 through this share, newest first, added as they happen. Kept in memory only, never written down: no headers, no bodies, and query values show as "…".</p>
new file mode 100644
new file mode 100644
@@ -0,0 +1,139 @@
@@ -0,0 +1,139 @@
+'use strict';
+/**
+ * Hooks that wait for your computer (CON-15): a webhook address on a share, a provider's webhook kept with only the
+ * provider's headers and its pointer in the outbox (one transaction), Events' signed delivery handed to the computer
+ * once it is connected (503 while it is not, so Events retries), once only, and the sweep after 7 days.
+ * node test/hooks.test.js
+ */
+const assert = require('assert');
+const contracts = require('openvibe-contracts');
+const { signDeliveryHeaders } = require('openvibe-sdk/events');
+const { boot, startLocalApp } = require('./helpers/boot');
+const { check, done } = require('./helpers/check');
+
+const SECRET = 'hook-delivery-secret-0123456789abcdef';
+
+(async () => {
+ const t = await boot({ env: { CONNECT_EVENTS_SECRET: SECRET } });
+ const app = await startLocalApp();
+ const alex = t.network.addUser('alex');
+ const bob = t.network.addUser('bob');
+ const made = await t.makeShare(alex, { label: 'bot', port: app.port });
+ let hook;
+ let envelope;
+
+ const deliver = async (env, secret = SECRET) => {
+ const raw = Buffer.from(JSON.stringify({ event: env }));
+ const headers = { 'content-type': 'application/json', ...signDeliveryHeaders(raw, secret), 'x-openvibe-subscription-id': 'sub_x', 'x-openvibe-delivery-attempt': '1' };
+ return t.get('/internal/hook-deliveries', { method: 'POST', body: raw, headers });
+ };
+
+ await check('the owner makes a webhook address: a path on the app, at most ten, never on another person\'s share', async () => {
+ const r = await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: '/webhooks/github', label: 'GitHub pushes' } });
+ assert.strictEqual(r.status, 201, r.text);
+ hook = r.json().hook;
+ assert.match(hook.url, /^https:\/\/connect\.openvibe\.website\/h\/hk_[0-9A-HJKMNP-TV-Z]{26}$/);
+ assert.strictEqual(hook.local_path, '/webhooks/github');
+ assert.strictEqual((await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: 'no-slash' } })).status, 422);
+ assert.strictEqual((await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: '/a b' } })).status, 422);
+ assert.strictEqual((await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: bob, json: { path: '/x' } })).status, 404);
+ const list = await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex });
+ assert.deepStrictEqual(list.json().hooks.map((h) => h.id), [hook.id]);
+ const page = await t.get(`/shares/${made.share.id}`, { as: alex });
+ assert.match(page.text, /Webhooks that wait for your computer/);
+ assert.ok(page.text.includes(hook.url));
+ });
+
+ await check('a provider\'s webhook: 202, kept with its headers (never cookies or forwarding), its pointer in the outbox', async () => {
+ const body = JSON.stringify({ ref: 'refs/heads/main', commits: 3 });
+ const r = await t.get(`/h/${hook.id}`, { method: 'POST', body, headers: { 'content-type': 'application/json', 'x-github-event': 'push', 'x-hub-signature-256': 'sha256=abc', cookie: 'session=secret', 'x-forwarded-for': '203.0.113.9', authorization: 'Bearer nope' } });
+ assert.strictEqual(r.status, 202, r.text);
+ const ref = r.json().id;
+ assert.match(ref, /^hb_/);
+ const row = await t.ctx.s.db.one('SELECT method, headers, body FROM hook_requests WHERE id = $1', [ref]);
+ assert.strictEqual(row.method, 'POST');
+ assert.strictEqual(Buffer.from(row.body).toString(), body);
+ const names = row.headers.map(([k]) => k);
+ assert.ok(names.includes('x-github-event') && names.includes('x-hub-signature-256') && names.includes('content-type'), names.join(','));
+ assert.ok(!names.includes('cookie') && !names.includes('x-forwarded-for') && !names.includes('authorization'), names.join(','));
+ const out = await t.ctx.s.db.many("SELECT envelope FROM service_outbox WHERE envelope->>'event_type' = 'connect.hook.received'");
+ assert.strictEqual(out.length, 1);
+ envelope = out[0].envelope;
+ assert.deepStrictEqual(Object.keys(envelope.payload).sort(), ['body_ref', 'content_type', 'hook_id', 'method', 'received_at', 'share_id', 'size']);
+ assert.strictEqual(envelope.payload.body_ref, ref);
+ assert.ok(!JSON.stringify(envelope).includes('refs/heads/main'), 'no body in the event');
+ const v = contracts.validate('events.event-envelope@1', envelope);
+ assert.ok(v.valid !== false, JSON.stringify(v.errors || v));
+ });
+
+ await check('the address takes POST, PUT and PATCH only; an unknown address is 404; more than 1 MB is refused', async () => {
+ assert.strictEqual((await t.get(`/h/${hook.id}`)).status, 405);
+ assert.strictEqual((await t.get('/h/hk_01K9ZZZZZZZZZZZZZZZZZZZZZZ', { method: 'POST', body: '{}' })).status, 404);
+ const big = await t.get(`/h/${hook.id}`, { method: 'POST', body: Buffer.alloc(1024 * 1024 + 10, 97), headers: { 'content-type': 'application/octet-stream' } });
+ 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 () => {
+ const off = await deliver(envelope);
+ assert.strictEqual(off.status, 503, off.text);
+ 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 });
+ const got = app.seen.slice(before).find((x) => x.url === '/webhooks/github');
+ assert.ok(got, 'the app got it');
+ assert.strictEqual(got.method, 'POST');
+ assert.strictEqual(got.body, JSON.stringify({ ref: 'refs/heads/main', commits: 3 }));
+ 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.cookie, undefined);
+ const again = await deliver(envelope);
+ assert.strictEqual(again.status, 200);
+ assert.deepStrictEqual(again.json(), { gone: true }, 'delivered once');
+ assert.strictEqual(app.seen.filter((x) => x.url === '/webhooks/github').length, 1);
+ const kept = await t.ctx.s.db.one('SELECT length(body)::int AS n, status FROM hook_requests WHERE id = $1', [envelope.payload.body_ref]);
+ assert.deepStrictEqual([kept.n, kept.status], [0, 200], 'its body is dropped once delivered');
+ const h = (await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex })).json().hooks[0];
+ assert.deepStrictEqual([h.received, h.delivered, h.last_status], [1, 1, 200]);
+ c.stop();
+ });
+
+ await check('a delivery must be signed, from this machine; another event type is acknowledged and ignored', async () => {
+ assert.strictEqual((await deliver(envelope, 'not-the-secret-0123456789abcdefgh')).status, 401);
+ const other = await deliver({ ...envelope, event_type: 'connect.something.else' });
+ assert.deepStrictEqual(other.json(), { ignored: true });
+ });
+
+ await check('removing the address: it answers 404 and its waiting webhooks are dropped', async () => {
+ const w = await t.get(`/h/${hook.id}`, { method: 'POST', body: '{"n":2}', headers: { 'content-type': 'application/json' } });
+ assert.strictEqual(w.status, 202);
+ const r = await t.get(`/api/v1/shares/${made.share.id}/hooks/${hook.id}`, { as: alex, method: 'DELETE' });
+ assert.strictEqual(r.status, 200, r.text);
+ assert.strictEqual((await t.get(`/h/${hook.id}`, { method: 'POST', body: '{}' })).status, 404);
+ const left = await t.ctx.s.db.one('SELECT count(*)::int AS n FROM hook_requests WHERE hook_id = $1 AND delivered_at IS NULL', [hook.id]);
+ assert.strictEqual(left.n, 0);
+ });
+
+ await check('the sweep removes webhooks past their 7 days', async () => {
+ const h2 = (await t.get(`/api/v1/shares/${made.share.id}/hooks`, { as: alex, json: { path: '/x' } })).json().hook;
+ const ref = (await t.get(`/h/${h2.id}`, { method: 'POST', body: 'old' })).json().id;
+ await t.ctx.s.db.exec("UPDATE hook_requests SET expires_at = now() - interval '1 minute' WHERE id = $1", [ref]);
+ await t.ctx.expireSweep();
+ assert.strictEqual((await t.ctx.s.db.maybe('SELECT id FROM hook_requests WHERE id = $1', [ref])), null);
+ });
+
+ 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);
+ assert.match(r.text, /Webhook address made/);
+ assert.match(r.text, /localhost:\d+\/hooks\/stripe · Stripe/);
+ });
+
+ await t.close();
+ await app.close();
+ done();
+})().catch((e) => { console.error(e); process.exit(1); });