openvibers/openvibe.watch Commits
A watch can follow a YouTube channel (WebSub, agent plan A5) (#201)
Changes
nnew file mode 100644@@@ -0,0 +1,6 @@++-- phase: expand++-- OpenVibe.Watch YouTube (agent plan A5): a webhook watch can follow a YouTube channel through YouTube's WebSub hub++-- (provider 'youtube'). `topic` is the channel's feed the hub was asked for, and `lease_until` (ms) when the hub's++-- subscription ends, so Watch renews it before then. Other providers leave both NULL.++ALTER TABLE webhook_secrets ADD COLUMN IF NOT EXISTS topic text;++ALTER TABLE webhook_secrets ADD COLUMN IF NOT EXISTS lease_until bigint;@@@ -102,7 +102,7 @@ function watchesRouter({ registry, auth, limits, webhookSources = null }) {}// { provider, secret }: the sender's own signing secret (Slack, Stripe) instead of a new one of Watch's.const b = req.body && typeof req.body === 'object' ? req.body : {};-- const hook = await webhookSources.issue(row.id, { provider: b.provider || null, secret: b.secret || null });++ const hook = await webhookSources.issue(row.id, { provider: b.provider || null, secret: b.secret || null, channel_id: b.channel_id || null });if (hook.error) return res.status(422).type('application/problem+json').send(JSON.stringify({ type: 'about:blank', title: 'watch.invalid', status: 422, code: 'watch.invalid', detail: hook.error }));res.json({ watch_id: row.id, webhook: hook });} catch (err) { fail(res, req, err); }@@@ -112,6 +112,8 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,// (the signature covers the exact bytes), so no body parser runs before it.// Email in (A3): the Cloudflare mail Worker posts each message here, signed with WATCH_MAIL_INGRESS_SECRET.if (webhookSources) app.post('/hooks/email', (req, res) => { webhookSources.receiveMail(req, res).catch((err) => { log.error(`[webhook-sources] mail: ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });++ // YouTube's hub checks a subscription with a GET before it sends anything (server/webhook-sources.js verifyIntent).++ if (webhookSources) app.get('/hooks/:id', (req, res) => { webhookSources.verifyIntent(req, res).catch((err) => { log.error(`[webhook-sources] verify: ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });if (webhookSources) app.post('/hooks/:id', (req, res) => { webhookSources.receive(req, res).catch((err) => { log.error(`[webhook-sources] ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });app.use('/api', express.json({ limit: '256kb', type: ['application/json', 'application/*+json'] }));@@@ -99,6 +99,10 @@ function load(env = process.env) {domain: (env.WATCH_MAIL_DOMAIN || 'openvibe.watch').trim().toLowerCase(),ingressSecret: env.WATCH_MAIL_INGRESS_SECRET || null,},++ // YouTube (agent plan A5, server/webhook-sources.js): the WebSub hub a YouTube watch subscribes through.++ youtube: {++ hub: (env.WATCH_YOUTUBE_HUB || 'https://pubsubhubbub.appspot.com/subscribe').trim(),++ },// Actions (server/actions.js): every WATCH_ACTIONS_TICK_MS deliver up to WATCH_ACTIONS_BATCH queued actions of fired// watches. Notifications need the Network OAuth client secret (a token for network.notifications.push).@@@ -827,6 +827,7 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: CREATE_Sconst acts = actions ? await actions.recent(row.id, 10) : [];const hookProvider = v.source.kind === 'webhook' && ctx.webhookSources && ctx.webhookSources.providerOf ? await ctx.webhookSources.providerOf(row.id) : null;const hookAddress = hookProvider === 'email' && ctx.webhookSources.addressOf ? await ctx.webhookSources.addressOf(row.id) : null;++ const hookYoutube = hookProvider === 'youtube' && ctx.webhookSources.youtubeOf ? await ctx.webhookSources.youtubeOf(row.id) : null;const mailIn = !!(config.mailIn && config.mailIn.ingressSecret);const configuredActions = (v.action || []).map((a) => a.kind === 'actor-agent'? html`<div>Run agent <a href="https://openvibe.actor/me/agents/${a.agent_id}"><code>${a.agent_id}</code></a>${a.note ? html` — ${a.note}` : ''}</div>`@@@ -893,14 +894,21 @@ ${ran ? notice(html`The check ran: <strong>${W.stateWords(ran)}</strong>.`, ran${testRun ? testCheckNotice(testRun, testObs, v) : ''}<dl class="facts"><dt>Status</dt><dd>${badge(unarmed ? 'Not armed yet' : W.statusWords(v.status), unarmed ? 'warn' : W.statusKind(v.status))}</dd>--${v.source.kind === 'webhook' && hookAddress++${v.source.kind === 'webhook' && hookYoutube++ ? html`<dt>YouTube channel</dt><dd><a href="https://www.youtube.com/channel/${hookYoutube.channel_id}" rel="noopener">${hookYoutube.channel_id}</a><br><span class="muted small">${hookYoutube.lease_until && hookYoutube.lease_until > Date.now()++ ? `A new or changed video wakes the watch. YouTube's hub confirmed it until ${new Date(hookYoutube.lease_until).toISOString().slice(0, 10)}; Watch renews it before then.`++ : 'Waiting for YouTube\'s hub to confirm (usually within a minute; Watch asks again each hour).'}</span>++<details class="hook-provider"><summary>Follow another channel?</summary>${youtubeForm(v)}</details>++<details class="hook-provider"><summary>Use a webhook instead?</summary><p class="muted small">A webhook address with a secret of Watch's own replaces the channel: YouTube's hub is told to stop.</p><form method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a webhook secret</button></form></details></dd>`++ : v.source.kind === 'webhook' && hookAddress? html`<dt>Email address</dt><dd><code class="copyable">${hookAddress}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="email"><button class="btn" type="submit">Make a new address</button></form><br><span class="muted small">Mail sent here wakes the watch. Anyone with the address can, so give it only to senders you mean; a new address retires this one.</span><details class="hook-provider"><summary>Use a webhook instead?</summary><p class="muted small">A webhook address with a secret of Watch's own replaces the email address, which stops working at once.</p><form method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a webhook secret</button></form></details></dd>`: v.source.kind === 'webhook'? html`<dt>Webhook address</dt><dd><code>${config.baseUrl}/hooks/${v.id}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a new secret</button></form><br><span class="muted small">${hookProvider ? `Signed by ${({ slack: 'Slack', stripe: 'Stripe', discord: 'Discord' })[hookProvider] || hookProvider} with the key you gave. A new secret of Watch's own replaces it.` : 'The secret was shown once, when the watch was made. A new one replaces it at once.'}</span><details class="hook-provider"><summary>Signed by Slack, Stripe or Discord?</summary><p class="muted small">They sign with a key of their own: paste it here and Watch checks their signature instead of its own. It is kept to check each delivery and never shown again.</p><form method="post" action="/watches/${v.id}/webhook-secret"><label>Sender <select name="provider"><option value="slack">Slack (signing secret)</option><option value="stripe">Stripe (endpoint secret, whsec_…)</option><option value="discord">Discord (application public key)</option></select></label> <label>Its signing secret <input name="secret" type="password" autocomplete="off" required maxlength="220"></label> <button class="btn" type="submit">Use it</button></form></details>${mailIn ? html`--<details class="hook-provider"><summary>Wake it by email?</summary><p class="muted small">The watch gets an address of its own at ${config.mailIn.domain}: mail sent there runs the check with its sender, subject and text. It replaces the webhook secret.</p><form method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="email"><button class="btn" type="submit">Give it an email address</button></form></details>` : ''}</dd>`++<details class="hook-provider"><summary>Wake it by email?</summary><p class="muted small">The watch gets an address of its own at ${config.mailIn.domain}: mail sent there runs the check with its sender, subject and text. It replaces the webhook secret.</p><form method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="email"><button class="btn" type="submit">Give it an email address</button></form></details>` : ''}++<details class="hook-provider"><summary>Follow a YouTube channel?</summary><p class="muted small">YouTube tells the watch about every new or changed video on the channel, through its public hub. It replaces the webhook secret.</p>${youtubeForm(v)}</details></dd>`: html`<dt>Source</dt><dd><code>${v.source.kind}</code> · <code>${v.source.url || (v.source.pattern ? v.source.pattern : '(none)')}</code></dd>`}<dt>Cadence</dt><dd>${v.cadence ? W.everyWords(v.cadence.every_sec) : (v.source.kind === 'webhook' ? 'when a delivery arrives' : (v.source.kind === 'event' ? 'when its event arrives' : 'not polled'))}</dd><dt>Actions</dt><dd>${configuredActions}</dd>@@@ -964,6 +972,8 @@ ${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty:const row = await owned(req, res);if (!row) return notFoundPage(req, res);await registry.remove(row.id, principalOf(req));++ // As the API's delete does: its webhook secret or address goes, and a YouTube hub is told to stop.++ if (ctx.webhookSources) await ctx.webhookSources.forget(row.id);return res.redirect(303, '/watches');});@@@ -994,18 +1004,35 @@ ${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty:if (v.source.kind !== 'webhook' || !ctx.webhookSources) return detailNotice(req, res, row, 409, 'Only a webhook watch has a secret.');// A sender's own signing secret (Slack, Stripe), or a new secret of Watch's own.const b = req.body || {};-- const hook = await ctx.webhookSources.issue(row.id, { provider: b.provider || null, secret: b.secret || null });++ const hook = await ctx.webhookSources.issue(row.id, { provider: b.provider || null, secret: b.secret || null, channel_id: b.channel_id || null });if (hook.error) return detailNotice(req, res, row, 422, `Not changed: ${hook.error}.`);return secretPage(req, res, v, hook, 200);});++ /** The form that makes a webhook watch follow a YouTube channel (provider youtube). */++ function youtubeForm(v) {++ return html`<form method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="youtube"><label>Channel id <input name="channel_id" required pattern="UC[A-Za-z0-9_\\-]{22}" maxlength="24" placeholder="UC…" autocomplete="off" spellcheck="false"></label> <button class="btn" type="submit">Follow it</button><br><span class="muted small">The id in the channel's address, youtube.com/channel/UC…</span></form>`;++ }++/** The address and secret of a webhook watch, shown once: no cache, no index, and never again after this page. */function secretPage(req, res, v, hook, status) {res.set('Cache-Control', 'no-store');res.set('Referrer-Policy', 'no-referrer');return page(req, res, {title: `${v.name}: webhook address`, private: true, crumbs: [{ label: 'Your watches', href: '/watches' }, { label: v.name, href: `/watches/${v.id}` }],-- body: hook.provider === 'email' ? html`<h1>Your watch has an email address</h1>++ body: hook.provider === 'youtube' ? html`<h1>Your watch follows a YouTube channel</h1>++${notice(hook.subscribed ? 'YouTube\'s hub took the subscription and checks it next, usually within a minute.' : 'Saved, but YouTube\'s hub did not answer: Watch asks it again within the hour.', hook.subscribed ? 'ok' : 'warn')}++<dl class="facts">++<dt>Channel</dt><dd><a href="https://www.youtube.com/channel/${hook.channel_id}" rel="noopener">${hook.channel_id}</a></dd>++</dl>++<h2>What the watch reads</h2>++<ul class="facts">++<li>Each new or changed video runs the check with <code>channel_id</code>, <code>videos</code> (each with <code>id</code>, <code>title</code>, <code>url</code>, <code>author</code>, <code>published</code>, <code>updated</code>) and <code>deleted</code>. For example <code>value_path: "videos.0.title"</code> with the condition "contains: live".</li>++<li>YouTube's hub signs each notification with a secret Watch gave it; one that is not signed so is ignored.</li>++<li>The subscription lasts ten days and Watch renews it before then, while the watch is active.</li>++</ul>++<p>${W.inWords(v)}</p>++<p><a class="btn" href="/watches/${v.id}">Back to the watch</a></p>` : hook.provider === 'email' ? html`<h1>Your watch has an email address</h1>${notice('Mail sent to this address wakes the watch. The address is the key: anyone who has it can wake the watch, so give it only to senders you mean.', 'ok')}<dl class="facts"><dt>Address</dt><dd><code class="copyable">${hook.address}</code></dd>@@@ -97,7 +97,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.const eventSources = createEventSources({ db, check, config, fetchImpl, log, now });registry.onEventSourceChange(() => eventSources.kick());-- const webhookSources = createWebhookSources({ db, registry, check, config, now, log });++ const webhookSources = createWebhookSources({ db, registry, check, config, now, log, fetchImpl });const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });const auth = createAuth({ config, log });// The public site: sessions in Watch's own database, the Network key to verify a sign-in token@@@ -124,6 +124,8 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchtry { await observations.prune(); } catch (err) { log.error(`[observations] prune: ${err.message}`); }try { await eventSources.prune(); } catch (err) { log.error(`[event-sources] prune: ${err.message}`); }try { await sessions.prune(); } catch (err) { log.error(`[sessions] prune: ${err.message}`); }++ // YouTube's hub leases end after ten days: renew the ones ending within two (server/webhook-sources.js).++ try { await webhookSources.renewLeases(); } catch (err) { log.error(`[webhook-sources] renew: ${err.message}`); }}, 3600 * 1000);pruneTimer.unref?.();const outboxPruneTimer = setInterval(async () => { try { await outbox.outbox.prune(); } catch (err) { log.error(`[outbox] prune: ${err.message}`); } }, 6 * 3600 * 1000);@@@ -33,6 +33,15 @@* text, text_truncated, attachments } as its JSON input. An unknown address answers 404, and the* Worker refuses the mail, so the sender gets a bounce. The address is the key: anyone who has it* can wake the watch, and making a new one retires the old.++ * YouTube (agent plan A5) POST …/webhook-secret { provider: 'youtube', channel_id: 'UC…' } subscribes the watch to++ * the channel through YouTube's WebSub hub (WATCH_YOUTUBE_HUB, pubsubhubbub.appspot.com): the callback is++ * the watch's own /hooks/<id>, signed with a new whk_ secret of Watch's (X-Hub-Signature: sha1=<hex++ * HMAC>, as the hub sends it). The hub first checks the callback with a GET (hub.mode, hub.topic,++ * hub.challenge): only this watch's own topic is echoed. Each notification (an Atom entry for a new or++ * changed video, or a deleted one) runs the check with { provider, channel_id, videos: [{ id, channel_id,++ * title, url, author, published, updated }], deleted: [{ id, when }] } as its JSON input. A lease runs++ * ten days and is renewed two days before it ends (renewLeases, hourly). A bad signature is ignored++ * with a 2xx, as WebSub asks, so the hub does not retry it.* Answers 200 { state, triggered } after the check; 404 for an unknown id or a watch that is not a webhook* watch; 401 for a missing or wrong signature (the same answer either way); 409 for a paused watch;* 429 over the watch's hourly budget (WATCH_WEBHOOK_MAX_PER_HOUR, default 60) with Retry-After;@@@ -56,6 +65,56 @@ const B32 = '0123456789abcdefghjkmnpqrstvwxyz';const newMailLocal = () => `in-${Array.from(crypto.randomBytes(26), (b) => B32[b & 31]).join('')}`;const newSecret = () => `whk_${crypto.randomBytes(32).toString('base64url')}`;++++// YouTube through its WebSub hub (agent plan A5): a channel id, its feed (the hub's topic), and the lease asked for.++const YT_CHANNEL_RE = /^UC[A-Za-z0-9_-]{22}$/;++const ytTopic = (channelId) => `https://www.youtube.com/xml/feeds/videos.xml?channel_id=${channelId}`;++const YT_LEASE_SEC = 10 * 86400;++const YT_RENEW_BEFORE_MS = 2 * 86400 * 1000;++++const XML_ENTITIES = { amp: '&', lt: '<', gt: '>', quot: '"', apos: "'" };++const unxml = (s) => String(s == null ? '' : s)++ .replace(/<!\[CDATA\[([\s\S]*?)\]\]>/g, '$1')++ .replace(/&(#x[0-9a-f]+|#[0-9]+|[a-z]+);/gi, (m, e) => (e[0] === '#'++ ? String.fromCodePoint(e[1] === 'x' || e[1] === 'X' ? parseInt(e.slice(2), 16) : parseInt(e.slice(1), 10))++ : (XML_ENTITIES[e.toLowerCase()] != null ? XML_ENTITIES[e.toLowerCase()] : m)))++ .trim();++const tag = (xml, name) => { const m = xml.match(new RegExp(`<${name}(?:\\s[^>]*)?>([\\s\\S]*?)</${name}>`)); return m ? unxml(m[1]) : null; };++++/**++ * A YouTube WebSub notification (Atom) as the check's input: the videos it names and the ones it says were deleted.++ * Only what the feed says: no title or channel is invented when the feed leaves it out.++ */++function parseYoutubeFeed(xml) {++ const s = String(xml || '');++ const videos = [];++ for (const m of s.matchAll(/<entry(?:\s[^>]*)?>([\s\S]*?)<\/entry>/g)) {++ const e = m[1];++ const id = tag(e, 'yt:videoId');++ if (!id) continue;++ const link = e.match(/<link\b[^>]*\brel="alternate"[^>]*\bhref="([^"]+)"/) || e.match(/<link\b[^>]*\bhref="([^"]+)"[^>]*\brel="alternate"/);++ const author = e.match(/<author>[\s\S]*?<name>([\s\S]*?)<\/name>/);++ videos.push({++ id, channel_id: tag(e, 'yt:channelId'), title: tag(e, 'title'),++ url: link ? unxml(link[1]) : `https://www.youtube.com/watch?v=${id}`,++ author: author ? unxml(author[1]) : null, published: tag(e, 'published'), updated: tag(e, 'updated'),++ });++ }++ const deleted = [...s.matchAll(/<at:deleted-entry\b[^>]*>/g)].map((m) => {++ const ref = (m[0].match(/\bref="([^"]*)"/) || [])[1] || '';++ const when = (m[0].match(/\bwhen="([^"]*)"/) || [])[1] || null;++ return { id: unxml(ref).replace(/^yt:video:/, ''), when: when ? unxml(when) : null };++ }).filter((d) => d.id);++ const channel = videos.find((v) => v.channel_id) || null;++ return { provider: 'youtube', channel_id: channel ? channel.channel_id : null, videos, deleted };++}++++/** WebSub's signature: X-Hub-Signature: <sha1|sha256|sha384|sha512>=<hex HMAC of the body with the secret>. */++function websubVerified(secret, raw, headers) {++ const m = String(headers['x-hub-signature'] || '').match(/^(sha1|sha256|sha384|sha512)=([0-9a-f]+)$/i);++ if (!m) return false;++ return safeEqual(m[2].toLowerCase(), crypto.createHmac(m[1].toLowerCase(), secret).update(raw).digest('hex'));++}const endpointFor = (watchId) => `/internal/hooks/${watchId}`;function safeEqual(a, b) {@@@ -107,6 +166,7 @@ function verified(secret, raw, headers, { provider = null, now = Date.now() } =if (provider === 'slack') return slackVerified(secret, raw, headers, now);if (provider === 'stripe') return stripeVerified(secret, raw, headers, now);if (provider === 'discord') return discordVerified(secret, raw, headers, now);++ if (provider === 'youtube') return websubVerified(secret, raw, headers);if (provider) return false;const tg = headers['x-telegram-bot-api-secret-token'];if (tg != null && safeEqual(String(tg), secret)) return true;@@@ -120,7 +180,7 @@ function verified(secret, raw, headers, { provider = null, now = Date.now() } =return false;}--function createWebhookSources({ db, registry, check, config, now = () => Date.now(), log = console }) {++function createWebhookSources({ db, registry, check, config, now = () => Date.now(), log = console, fetchImpl = globalThis.fetch }) {const maxPerHour = Number.isInteger(config.webhookSources && config.webhookSources.maxPerWatchPerHour) && config.webhookSources.maxPerWatchPerHour > 0? config.webhookSources.maxPerWatchPerHour : 60;const counts = new Map(); // watch id → { hour, n }@@@ -128,10 +188,17 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.nodomain: String((config.mailIn && config.mailIn.domain) || 'openvibe.watch').toLowerCase(),ingressSecret: (config.mailIn && config.mailIn.ingressSecret) || null,};++ const youtubeHub = String((config.youtube && config.youtube.hub) || 'https://pubsubhubbub.appspot.com/subscribe');const st = {set: db.prepare(`INSERT INTO webhook_secrets (watch_id, secret, created_at, provider) VALUES (?, ?, ?, ?)-- ON CONFLICT (watch_id) DO UPDATE SET secret = excluded.secret, created_at = excluded.created_at, provider = excluded.provider`),-- get: db.prepare('SELECT secret, provider FROM webhook_secrets WHERE watch_id = ?'),++ ON CONFLICT (watch_id) DO UPDATE SET secret = excluded.secret, created_at = excluded.created_at, provider = excluded.provider,++ topic = NULL, lease_until = NULL`),++ setYoutube: db.prepare(`INSERT INTO webhook_secrets (watch_id, secret, created_at, provider, topic, lease_until) VALUES (?, ?, ?, 'youtube', ?, NULL)++ ON CONFLICT (watch_id) DO UPDATE SET secret = excluded.secret, created_at = excluded.created_at, provider = 'youtube',++ topic = excluded.topic, lease_until = NULL`),++ lease: db.prepare("UPDATE webhook_secrets SET lease_until = ? WHERE watch_id = ? AND provider = 'youtube' AND topic = ?"),++ due: db.prepare("SELECT watch_id, secret, topic FROM webhook_secrets WHERE provider = 'youtube' AND (lease_until IS NULL OR lease_until < ?) ORDER BY watch_id LIMIT 200"),++ get: db.prepare('SELECT secret, provider, topic, lease_until FROM webhook_secrets WHERE watch_id = ?'),drop: db.prepare('DELETE FROM webhook_secrets WHERE watch_id = ?'),watch: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),byMail: db.prepare("SELECT watch_id FROM webhook_secrets WHERE provider = 'email' AND secret = ?"),@@@ -142,8 +209,36 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no* secret ({ provider: 'slack' | 'stripe', secret }), that secret is kept instead and never echoed back. A bad* provider or secret answers { error } and changes nothing.*/-- async function issue(watchId, { provider = null, secret: theirs = null } = {}) {++ /** Ask YouTube's hub to (un)subscribe a watch's callback. True when the hub took the request (it checks it next). */++ async function hub(mode, watchId, secret, topic) {++ const form = new URLSearchParams({ 'hub.callback': `${config.baseUrl}/hooks/${watchId}`, 'hub.topic': topic, 'hub.mode': mode, 'hub.verify': 'async' });++ if (mode === 'subscribe') { form.set('hub.secret', secret); form.set('hub.lease_seconds', String(YT_LEASE_SEC)); }++ try {++ const r = await fetchImpl(youtubeHub, { method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, body: form.toString(), signal: AbortSignal.timeout(15000) });++ if (r.status === 202 || r.status === 204) return true;++ log.warn(`[webhook-sources] youtube ${mode} ${watchId}: the hub answered ${r.status}`);++ } catch (err) { log.warn(`[webhook-sources] youtube ${mode} ${watchId}: ${err.message}`); }++ return false;++ }++ /** A watch leaving YouTube: the hub is told, best effort (a lease left alone ends in ten days anyway). */++ async function leaveYoutube(watchId) {++ const kept = await st.get.get(watchId);++ if (kept && kept.provider === 'youtube' && kept.topic) await hub('unsubscribe', watchId, kept.secret, kept.topic);++ }++++ async function issue(watchId, { provider = null, secret: theirs = null, channel_id: channelId = null } = {}) {const url = `${config.baseUrl}/hooks/${watchId}`;++ if (provider === 'youtube') {++ const ch = String(channelId || '').trim();++ if (!YT_CHANNEL_RE.test(ch)) return { error: 'a YouTube channel id starts with UC and has 24 characters (youtube.com/channel/UC…)' };++ const before = await st.get.get(watchId);++ if (before && before.provider === 'youtube' && before.topic && before.topic !== ytTopic(ch)) await hub('unsubscribe', watchId, before.secret, before.topic);++ const secret = newSecret();++ await st.setYoutube.run(watchId, secret, now(), ytTopic(ch));++ const asked = await hub('subscribe', watchId, secret, ytTopic(ch));++ return { url, provider: 'youtube', channel_id: ch, subscribed: asked, signature: 'YouTube\'s hub signs each notification with a secret Watch gave it (X-Hub-Signature)' };++ }++ await leaveYoutube(watchId);if (provider === 'email') {if (!mail.ingressSecret) return { error: 'email in is not set up on this server' };const local = newMailLocal();@@@ -162,7 +257,48 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.noawait st.set.run(watchId, secret, now(), null);return { url, secret, signature: 'X-Hub-Signature-256: sha256=<hex HMAC-SHA256 of the body with the secret>, Authorization: Bearer <secret>, or Telegram\'s X-Telegram-Bot-Api-Secret-Token' };}-- async function forget(watchId) { await st.drop.run(watchId); }++ async function forget(watchId) { await leaveYoutube(watchId); await st.drop.run(watchId); }++ /** The channel a YouTube watch follows and when its lease ends (shown on the watch's page), or null. */++ async function youtubeOf(watchId) {++ const row = await st.get.get(watchId);++ if (!row || row.provider !== 'youtube' || !row.topic) return null;++ const m = row.topic.match(/channel_id=(UC[A-Za-z0-9_-]{22})$/);++ return { channel_id: m ? m[1] : null, lease_until: row.lease_until == null ? null : Number(row.lease_until) };++ }++ /** Renew every YouTube lease that ends within two days, or that the hub never confirmed (hourly, server/index.js). */++ async function renewLeases() {++ let asked = 0;++ for (const row of await st.due.all(now() + YT_RENEW_BEFORE_MS)) {++ const w = await st.watch.get(row.watch_id);++ if (!w || w.status !== 'active') continue;++ if (await hub('subscribe', row.watch_id, row.secret, row.topic)) asked++;++ }++ return asked;++ }++++ /**++ * GET /hooks/:id — the hub's check of a (un)subscription: its challenge is echoed only for the topic this watch++ * follows (subscribe), or one it no longer follows (unsubscribe). A subscribe confirmed sets the lease's end.++ */++ async function verifyIntent(req, res) {++ const id = String(req.params.id || '');++ const q = new URL(req.originalUrl || req.url, 'http://watch.invalid').searchParams;++ const mode = q.get('hub.mode');++ const topic = q.get('hub.topic');++ const challenge = q.get('hub.challenge');++ res.setHeader('Cache-Control', 'no-store');++ const kept = /^wch_[0-9A-HJKMNP-TV-Z]{26}$/.test(id) ? await st.get.get(id) : null;++ const follows = !!(kept && kept.provider === 'youtube' && kept.topic && kept.topic === topic);++ if (mode === 'denied') { log.warn(`[webhook-sources] youtube ${id}: the hub denied the subscription (${String(q.get('hub.reason') || '').slice(0, 200)})`); return res.status(200).end(); }++ if (!challenge || challenge.length > 512 || !topic) return problem(res, 404, 'watch.not_found', 'no subscription to confirm here');++ if (mode === 'subscribe' && follows) {++ const lease = Number(q.get('hub.lease_seconds'));++ await st.lease.run(now() + (Number.isFinite(lease) && lease > 0 ? Math.min(lease, 30 * 86400) : YT_LEASE_SEC) * 1000, id, topic);++ return res.status(200).type('text/plain').send(challenge);++ }++ if (mode === 'unsubscribe' && !follows) return res.status(200).type('text/plain').send(challenge);++ return problem(res, 404, 'watch.not_found', 'this watch does not follow that topic');++ }/** The watch's email address, when it has one (shown to its owner on the watch's page). */async function addressOf(watchId) {const row = await st.get.get(watchId);@@@ -207,7 +343,11 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.noif (!row || !source || source.kind !== 'webhook') return problem(res, 404, 'watch.not_found', 'no webhook watch at this address');const kept = await st.get.get(id) || {};const provider = kept.provider || null;-- if (!verified(kept.secret || null, raw, req.headers, { provider, now: now() })) return problem(res, 401, 'watch.bad_signature', 'the delivery is not signed with this watch\'s secret');++ if (!verified(kept.secret || null, raw, req.headers, { provider, now: now() })) {++ // WebSub: a notification whose signature does not match is ignored but acknowledged, so the hub does not resend it.++ if (provider === 'youtube') { log.warn(`[webhook-sources] youtube ${id}: a notification with a bad signature was ignored`); res.setHeader('Cache-Control', 'no-store'); return res.status(202).json({ watch_id: id, ignored: true }); }++ return problem(res, 401, 'watch.bad_signature', 'the delivery is not signed with this watch\'s secret');++ }if (provider === 'slack') {// Slack checks a new address with a challenge it wants back, and resends an event it did not see answered// within three seconds: the first delivery is the one that runs the check.@@@ -233,9 +373,13 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no}if (row.status !== 'active') return problem(res, 409, 'watch.disabled', `the watch is ${row.status}`);if (!underBudget(id)) return problem(res, 429, 'watch.rate_limited', `at most ${maxPerHour} deliveries an hour wake one watch`, { 'Retry-After': String(retryAfter()) });++ // A YouTube notification is Atom: the check reads it as JSON (parseYoutubeFeed).++ const input = provider === 'youtube'++ ? { body: Buffer.from(JSON.stringify(parseYoutubeFeed(raw.toString('utf8')))), content_type: 'application/json' }++ : { body: raw, content_type: req.headers['content-type'] || null };let out;try {-- out = await check.run(id, { trigger: 'webhook', input: { body: raw, content_type: req.headers['content-type'] || null } });++ out = await check.run(id, { trigger: 'webhook', input });} catch (err) {log.error(`[webhook-sources] ${id}: ${err.message}`);return problem(res, 500, 'watch.internal', 'the check failed; try again');@@@ -280,7 +424,7 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.noreturn res.json({ watch_id: row.id, state: out && out.run ? out.run.state : null, triggered: !!(out && out.met) });}-- return { issue, forget, providerOf, addressOf, receive, receiveMail, endpointFor, verified, SECRET_RE, PROVIDERS };++ return { issue, forget, providerOf, addressOf, youtubeOf, renewLeases, verifyIntent, receive, receiveMail, endpointFor, verified, SECRET_RE, PROVIDERS };}/** The mail Worker's signature: X-OpenVibe-Mail-Signature: t=<time>,v1=<hex HMAC-SHA256(secret, time + "." + body)>. */@@@ -292,4 +436,4 @@ function mailSigned(secret, raw, headers, nowMs = Date.now()) {return parts.filter(([k]) => k === 'v1').some(([, v]) => safeEqual(v || '', want));}--module.exports = { createWebhookSources, verified, mailSigned, endpointFor, newSecret, newMailLocal, MAIL_LOCAL_RE, PROVIDERS };++module.exports = { createWebhookSources, verified, mailSigned, endpointFor, newSecret, newMailLocal, MAIL_LOCAL_RE, PROVIDERS, parseYoutubeFeed, websubVerified, YT_CHANNEL_RE };nnew file mode 100644@@@ -0,0 +1,149 @@++'use strict';++/**++ * A webhook watch that follows a YouTube channel (agent plan A5) through YouTube's WebSub hub: the watch subscribes,++ * the hub checks the callback with a GET (only this watch's own topic is echoed), each notification is signed with++ * the secret Watch gave the hub (X-Hub-Signature) and becomes JSON for the check, a bad signature is acknowledged and++ * ignored, leases are renewed before they end, and leaving YouTube tells the hub. The hub here is a local fake.++ */++const assert = require('assert');++const crypto = require('crypto');++const http = require('http');++const { boot, request, userToken, suite, newUser } = require('./helpers');++const { parseYoutubeFeed, websubVerified } = require('../server/webhook-sources');++++const t = suite('youtube-websub');++let svc, hub, watch;++const me = newUser();++const ME = userToken(me);++const OTHER = userToken(newUser());++const CHANNEL = 'UC' + 'AbCdEfGhIjKlMnOpQrStUv';++const TOPIC = `https://www.youtube.com/xml/feeds/videos.xml?channel_id=${CHANNEL}`;++const api = (method, p, opts = {}) => request(svc.base, method, p, opts);++const runs = async (id) => await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid').all(id);++const sign = (secret, body, algo = 'sha1') => `${algo}=${crypto.createHmac(algo, secret).update(body).digest('hex')}`;++const feed = (title) => `<?xml version='1.0' encoding='UTF-8'?>++<feed xmlns:yt="http://www.youtube.com/xml/schemas/2015" xmlns="http://www.w3.org/2005/Atom">++ <link rel="hub" href="https://pubsubhubbub.appspot.com"/>++ <title>YouTube video feed</title>++ <entry>++ <id>yt:video:VIDEO_ID_01</id>++ <yt:videoId>VIDEO_ID_01</yt:videoId>++ <yt:channelId>${CHANNEL}</yt:channelId>++ <title>${title}</title>++ <link rel="alternate" href="https://www.youtube.com/watch?v=VIDEO_ID_01"/>++ <author><name>Channel & Friends</name><uri>https://www.youtube.com/channel/${CHANNEL}</uri></author>++ <published>2026-11-03T08:00:00+00:00</published>++ <updated>2026-11-03T08:00:05+00:00</updated>++ </entry>++</feed>`;++++function startHub() {++ const calls = [];++ const server = http.createServer((req, res) => {++ let body = '';++ req.on('data', (c) => { body += c; });++ req.on('end', () => { calls.push(Object.fromEntries(new URLSearchParams(body))); res.writeHead(202); res.end(); });++ });++ return new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve({ calls, url: `http://127.0.0.1:${server.address().port}/subscribe`, close: () => new Promise((r) => server.close(r)) })));++}++++t('boot', async () => {++ hub = await startHub();++ svc = await boot({ env: { WATCH_YOUTUBE_HUB: hub.url } });++});++++t('the feed: videos with their fields, entities decoded, deleted entries; the signature in every WebSub algorithm', () => {++ const f = parseYoutubeFeed(feed('Going LIVE <now>'));++ assert.deepStrictEqual(f.videos, [{ id: 'VIDEO_ID_01', channel_id: CHANNEL, title: 'Going LIVE <now>', url: 'https://www.youtube.com/watch?v=VIDEO_ID_01', author: 'Channel & Friends', published: '2026-11-03T08:00:00+00:00', updated: '2026-11-03T08:00:05+00:00' }]);++ assert.strictEqual(f.channel_id, CHANNEL);++ const gone = parseYoutubeFeed(`<feed xmlns:at="http://purl.org/atompub/tombstones/1.0"><at:deleted-entry ref="yt:video:OLD_VIDEO_1" when="2026-11-03T09:00:00+00:00"><link href="https://www.youtube.com/watch?v=OLD_VIDEO_1"/></at:deleted-entry></feed>`);++ assert.deepStrictEqual([gone.videos, gone.deleted], [[], [{ id: 'OLD_VIDEO_1', when: '2026-11-03T09:00:00+00:00' }]]);++ assert.deepStrictEqual(parseYoutubeFeed('not xml at all'), { provider: 'youtube', channel_id: null, videos: [], deleted: [] });++ const body = Buffer.from('x');++ for (const algo of ['sha1', 'sha256', 'sha512']) assert.strictEqual(websubVerified('whk_s', body, { 'x-hub-signature': sign('whk_s', body, algo) }), true, algo);++ assert.strictEqual(websubVerified('whk_s', body, { 'x-hub-signature': sign('whk_t', body) }), false);++ assert.strictEqual(websubVerified('whk_s', body, {}), false);++});++++t('the owner points a webhook watch at a channel: Watch subscribes at the hub with its callback, topic and a secret', async () => {++ const made = await api('POST', '/api/v1/watches', { token: ME, body: { name: 'channel goes live', source: { kind: 'webhook', endpoint: '/internal/hooks/x' }, cadence: null, extraction: { kind: 'json', value_path: 'videos.0.title' }, condition: { op: 'contains', value: 'LIVE' }, action: [{ kind: 'notification' }] } });++ assert.strictEqual(made.status, 201, JSON.stringify(made.body));++ watch = made.body.watch;++ assert.strictEqual((await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: { provider: 'youtube', channel_id: 'not-a-channel' } })).status, 422);++ assert.strictEqual((await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: OTHER, body: { provider: 'youtube', channel_id: CHANNEL } })).status, 404);++ const set = await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: { provider: 'youtube', channel_id: CHANNEL } });++ assert.strictEqual(set.status, 200, JSON.stringify(set.body));++ assert.deepStrictEqual([set.body.webhook.provider, set.body.webhook.channel_id, set.body.webhook.subscribed], ['youtube', CHANNEL, true]);++ assert.ok(!('secret' in set.body.webhook), 'the hub\'s secret is never shown');++ const asked = hub.calls[hub.calls.length - 1];++ assert.deepStrictEqual([asked['hub.mode'], asked['hub.topic'], asked['hub.callback'], asked['hub.verify'], asked['hub.lease_seconds']], ['subscribe', TOPIC, `${svc.config.baseUrl}/hooks/${watch.id}`, 'async', '864000']);++ const kept = await svc.db.prepare('SELECT secret, provider, topic, lease_until FROM webhook_secrets WHERE watch_id = ?').get(watch.id);++ assert.deepStrictEqual([kept.secret, kept.provider, kept.topic, kept.lease_until], [asked['hub.secret'], 'youtube', TOPIC, null]);++ assert.match(kept.secret, /^whk_/);++});++++t('the hub checks the callback: its challenge comes back only for this watch\'s topic, and the lease is kept', async () => {++ const verify = (q) => fetch(`${svc.base}/hooks/${watch.id}?${new URLSearchParams(q)}`);++ const wrong = await verify({ 'hub.mode': 'subscribe', 'hub.topic': TOPIC.replace(CHANNEL, 'UC' + 'z'.repeat(22)), 'hub.challenge': 'c1', 'hub.lease_seconds': '864000' });++ assert.strictEqual(wrong.status, 404, 'another channel is not confirmed');++ const unsub = await verify({ 'hub.mode': 'unsubscribe', 'hub.topic': TOPIC, 'hub.challenge': 'c2' });++ assert.strictEqual(unsub.status, 404, 'an unsubscribe of the topic it still follows is refused');++ const ok = await verify({ 'hub.mode': 'subscribe', 'hub.topic': TOPIC, 'hub.challenge': 'abc123XYZ', 'hub.lease_seconds': '864000' });++ assert.strictEqual(ok.status, 200);++ assert.strictEqual(await ok.text(), 'abc123XYZ');++ const lease = Number((await svc.db.prepare('SELECT lease_until FROM webhook_secrets WHERE watch_id = ?').get(watch.id)).lease_until);++ assert.ok(Math.abs(lease - (Date.now() + 864000 * 1000)) < 60_000, 'the lease ends ten days from now');++ assert.strictEqual((await fetch(`${svc.base}/hooks/wch_00000000000000000000000000?hub.mode=subscribe&hub.topic=x&hub.challenge=y`)).status, 404);++ const page = await svc.get(`/watches/${watch.id}`, { as: { subject: me, username: 'yt-owner' } });++ assert.ok(page.text.includes(`youtube.com/channel/${CHANNEL}`) && page.text.includes('confirmed it until'), 'the watch\'s page shows the channel and the lease');++});++++t('a signed notification runs the check with the feed as JSON; a badly signed one is acknowledged and ignored', async () => {++ const secret = (await svc.db.prepare('SELECT secret FROM webhook_secrets WHERE watch_id = ?').get(watch.id)).secret;++ const before = (await runs(watch.id)).length;++ const bad = await fetch(`${svc.base}/hooks/${watch.id}`, { method: 'POST', body: feed('Going LIVE'), headers: { 'Content-Type': 'application/atom+xml', 'X-Hub-Signature': sign('whk_wrong', feed('Going LIVE')) } });++ assert.strictEqual(bad.status, 202, 'WebSub: acknowledged, so the hub does not resend it');++ assert.strictEqual((await bad.json()).ignored, true);++ assert.strictEqual((await runs(watch.id)).length, before, 'and nothing ran');++ const body = feed('Going LIVE: the big build');++ const r = await fetch(`${svc.base}/hooks/${watch.id}`, { method: 'POST', body, headers: { 'Content-Type': 'application/atom+xml', 'X-Hub-Signature': sign(secret, body) } });++ assert.strictEqual(r.status, 200);++ const out = await r.json();++ assert.strictEqual(out.triggered, true, JSON.stringify(out));++ const quiet = feed('A cooking video');++ const r2 = await fetch(`${svc.base}/hooks/${watch.id}`, { method: 'POST', body: quiet, headers: { 'Content-Type': 'application/atom+xml', 'X-Hub-Signature': sign(secret, quiet) } });++ assert.strictEqual((await r2.json()).triggered, false, 'a title without LIVE does not meet the condition');++});++++t('leases ending within two days are renewed; leaving YouTube tells the hub, which may then confirm the unsubscribe', async () => {++ await svc.db.prepare('UPDATE webhook_secrets SET lease_until = ? WHERE watch_id = ?').run(Date.now() + 86400 * 1000, watch.id);++ const n = hub.calls.length;++ assert.strictEqual(await svc.webhookSources.renewLeases(), 1);++ assert.deepStrictEqual([hub.calls.length, hub.calls[n]['hub.mode'], hub.calls[n]['hub.topic']], [n + 1, 'subscribe', TOPIC]);++ assert.strictEqual(await svc.webhookSources.renewLeases(), 1, 'until the hub confirms again, it is still due');++ await svc.db.prepare('UPDATE webhook_secrets SET lease_until = ? WHERE watch_id = ?').run(Date.now() + 9 * 86400 * 1000, watch.id);++ assert.strictEqual(await svc.webhookSources.renewLeases(), 0, 'a lease with days left is left alone');++ const back = await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: {} });++ assert.strictEqual(back.status, 200);++ const last = hub.calls[hub.calls.length - 1];++ assert.deepStrictEqual([last['hub.mode'], last['hub.topic'], last['hub.secret']], ['unsubscribe', TOPIC, undefined]);++ const kept = await svc.db.prepare('SELECT provider, topic, lease_until FROM webhook_secrets WHERE watch_id = ?').get(watch.id);++ assert.deepStrictEqual([kept.provider, kept.topic, kept.lease_until], [null, null, null]);++ const confirm = await fetch(`${svc.base}/hooks/${watch.id}?${new URLSearchParams({ 'hub.mode': 'unsubscribe', 'hub.topic': TOPIC, 'hub.challenge': 'bye' })}`);++ assert.deepStrictEqual([confirm.status, await confirm.text()], [200, 'bye']);++});++++t('deleting a YouTube watch on its page tells the hub and drops its secret', async () => {++ const made = await api('POST', '/api/v1/watches', { token: ME, body: { name: 'to delete', source: { kind: 'webhook', endpoint: '/internal/hooks/x' }, cadence: null, extraction: { kind: 'json', value_path: 'videos.0.title' }, condition: { op: 'contains', value: 'x' }, action: [{ kind: 'notification' }] } });++ const id = made.body.watch.id;++ assert.strictEqual((await api('POST', `/api/v1/watches/${id}/webhook-secret`, { token: ME, body: { provider: 'youtube', channel_id: CHANNEL } })).status, 200);++ const del = await svc.get(`/watches/${id}/delete`, { as: { subject: me, username: 'yt-owner' }, form: {} });++ assert.strictEqual(del.status, 303);++ const last = hub.calls[hub.calls.length - 1];++ assert.deepStrictEqual([last['hub.mode'], last['hub.callback']], ['unsubscribe', `${svc.config.baseUrl}/hooks/${id}`]);++ assert.strictEqual(await svc.db.prepare('SELECT 1 FROM webhook_secrets WHERE watch_id = ?').get(id), undefined);++});++++t('stop', async () => { await svc.stop(); await hub.close(); });++++t.run();
new file mode 100644
new file mode 100644
@@ -0,0 +1,6 @@
@@ -0,0 +1,6 @@
+-- phase: expand
+-- OpenVibe.Watch YouTube (agent plan A5): a webhook watch can follow a YouTube channel through YouTube's WebSub hub
+-- (provider 'youtube'). `topic` is the channel's feed the hub was asked for, and `lease_until` (ms) when the hub's
+-- subscription ends, so Watch renews it before then. Other providers leave both NULL.
+ALTER TABLE webhook_secrets ADD COLUMN IF NOT EXISTS topic text;
+ALTER TABLE webhook_secrets ADD COLUMN IF NOT EXISTS lease_until bigint;
@@ -102,7 +102,7 @@ function watchesRouter({ registry, auth, limits, webhookSources = null }) {
@@ -102,7 +102,7 @@ function watchesRouter({ registry, auth, limits, webhookSources = null }) {
}
}
// { provider, secret }: the sender's own signing secret (Slack, Stripe) instead of a new one of Watch's.
// { provider, secret }: the sender's own signing secret (Slack, Stripe) instead of a new one of Watch's.
const b = req.body && typeof req.body === 'object' ? req.body : {};
const b = req.body && typeof req.body === 'object' ? req.body : {};
- const hook = await webhookSources.issue(row.id, { provider: b.provider || null, secret: b.secret || null });
+ const hook = await webhookSources.issue(row.id, { provider: b.provider || null, secret: b.secret || null, channel_id: b.channel_id || null });
if (hook.error) return res.status(422).type('application/problem+json').send(JSON.stringify({ type: 'about:blank', title: 'watch.invalid', status: 422, code: 'watch.invalid', detail: hook.error }));
if (hook.error) return res.status(422).type('application/problem+json').send(JSON.stringify({ type: 'about:blank', title: 'watch.invalid', status: 422, code: 'watch.invalid', detail: hook.error }));
res.json({ watch_id: row.id, webhook: hook });
res.json({ watch_id: row.id, webhook: hook });
} catch (err) { fail(res, req, err); }
} catch (err) { fail(res, req, err); }
@@ -112,6 +112,8 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
@@ -112,6 +112,8 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
// (the signature covers the exact bytes), so no body parser runs before it.
// (the signature covers the exact bytes), so no body parser runs before it.
// Email in (A3): the Cloudflare mail Worker posts each message here, signed with WATCH_MAIL_INGRESS_SECRET.
// Email in (A3): the Cloudflare mail Worker posts each message here, signed with WATCH_MAIL_INGRESS_SECRET.
if (webhookSources) app.post('/hooks/email', (req, res) => { webhookSources.receiveMail(req, res).catch((err) => { log.error(`[webhook-sources] mail: ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });
if (webhookSources) app.post('/hooks/email', (req, res) => { webhookSources.receiveMail(req, res).catch((err) => { log.error(`[webhook-sources] mail: ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });
+ // YouTube's hub checks a subscription with a GET before it sends anything (server/webhook-sources.js verifyIntent).
+ if (webhookSources) app.get('/hooks/:id', (req, res) => { webhookSources.verifyIntent(req, res).catch((err) => { log.error(`[webhook-sources] verify: ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });
if (webhookSources) app.post('/hooks/:id', (req, res) => { webhookSources.receive(req, res).catch((err) => { log.error(`[webhook-sources] ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });
if (webhookSources) app.post('/hooks/:id', (req, res) => { webhookSources.receive(req, res).catch((err) => { log.error(`[webhook-sources] ${err.message}`); if (!res.headersSent) res.status(500).end(); }); });
app.use('/api', express.json({ limit: '256kb', type: ['application/json', 'application/*+json'] }));
app.use('/api', express.json({ limit: '256kb', type: ['application/json', 'application/*+json'] }));
@@ -99,6 +99,10 @@ function load(env = process.env) {
@@ -99,6 +99,10 @@ function load(env = process.env) {
domain: (env.WATCH_MAIL_DOMAIN || 'openvibe.watch').trim().toLowerCase(),
domain: (env.WATCH_MAIL_DOMAIN || 'openvibe.watch').trim().toLowerCase(),
ingressSecret: env.WATCH_MAIL_INGRESS_SECRET || null,
ingressSecret: env.WATCH_MAIL_INGRESS_SECRET || null,
},
},
+ // YouTube (agent plan A5, server/webhook-sources.js): the WebSub hub a YouTube watch subscribes through.
+ youtube: {
+ hub: (env.WATCH_YOUTUBE_HUB || 'https://pubsubhubbub.appspot.com/subscribe').trim(),
+ },
// Actions (server/actions.js): every WATCH_ACTIONS_TICK_MS deliver up to WATCH_ACTIONS_BATCH queued actions of fired
// Actions (server/actions.js): every WATCH_ACTIONS_TICK_MS deliver up to WATCH_ACTIONS_BATCH queued actions of fired
// watches. Notifications need the Network OAuth client secret (a token for network.notifications.push).
// watches. Notifications need the Network OAuth client secret (a token for network.notifications.push).
@@ -827,6 +827,7 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: CREATE_S
@@ -827,6 +827,7 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: CREATE_S
const acts = actions ? await actions.recent(row.id, 10) : [];
const acts = actions ? await actions.recent(row.id, 10) : [];
const hookProvider = v.source.kind === 'webhook' && ctx.webhookSources && ctx.webhookSources.providerOf ? await ctx.webhookSources.providerOf(row.id) : null;
const hookProvider = v.source.kind === 'webhook' && ctx.webhookSources && ctx.webhookSources.providerOf ? await ctx.webhookSources.providerOf(row.id) : null;
const hookAddress = hookProvider === 'email' && ctx.webhookSources.addressOf ? await ctx.webhookSources.addressOf(row.id) : null;
const hookAddress = hookProvider === 'email' && ctx.webhookSources.addressOf ? await ctx.webhookSources.addressOf(row.id) : null;
+ const hookYoutube = hookProvider === 'youtube' && ctx.webhookSources.youtubeOf ? await ctx.webhookSources.youtubeOf(row.id) : null;
const mailIn = !!(config.mailIn && config.mailIn.ingressSecret);
const mailIn = !!(config.mailIn && config.mailIn.ingressSecret);
const configuredActions = (v.action || []).map((a) => a.kind === 'actor-agent'
const configuredActions = (v.action || []).map((a) => a.kind === 'actor-agent'
? html`<div>Run agent <a href="https://openvibe.actor/me/agents/${a.agent_id}"><code>${a.agent_id}</code></a>${a.note ? html` — ${a.note}` : ''}</div>`
? html`<div>Run agent <a href="https://openvibe.actor/me/agents/${a.agent_id}"><code>${a.agent_id}</code></a>${a.note ? html` — ${a.note}` : ''}</div>`
@@ -893,14 +894,21 @@ ${ran ? notice(html`The check ran: <strong>${W.stateWords(ran)}</strong>.`, ran
@@ -893,14 +894,21 @@ ${ran ? notice(html`The check ran: <strong>${W.stateWords(ran)}</strong>.`, ran
${testRun ? testCheckNotice(testRun, testObs, v) : ''}
${testRun ? testCheckNotice(testRun, testObs, v) : ''}
<dl class="facts">
<dl class="facts">
<dt>Status</dt><dd>${badge(unarmed ? 'Not armed yet' : W.statusWords(v.status), unarmed ? 'warn' : W.statusKind(v.status))}</dd>
<dt>Status</dt><dd>${badge(unarmed ? 'Not armed yet' : W.statusWords(v.status), unarmed ? 'warn' : W.statusKind(v.status))}</dd>
-${v.source.kind === 'webhook' && hookAddress
+${v.source.kind === 'webhook' && hookYoutube
+ ? html`<dt>YouTube channel</dt><dd><a href="https://www.youtube.com/channel/${hookYoutube.channel_id}" rel="noopener">${hookYoutube.channel_id}</a><br><span class="muted small">${hookYoutube.lease_until && hookYoutube.lease_until > Date.now()
+ ? `A new or changed video wakes the watch. YouTube's hub confirmed it until ${new Date(hookYoutube.lease_until).toISOString().slice(0, 10)}; Watch renews it before then.`
+ : 'Waiting for YouTube\'s hub to confirm (usually within a minute; Watch asks again each hour).'}</span>
+<details class="hook-provider"><summary>Follow another channel?</summary>${youtubeForm(v)}</details>
+<details class="hook-provider"><summary>Use a webhook instead?</summary><p class="muted small">A webhook address with a secret of Watch's own replaces the channel: YouTube's hub is told to stop.</p><form method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a webhook secret</button></form></details></dd>`
+ : v.source.kind === 'webhook' && hookAddress
? html`<dt>Email address</dt><dd><code class="copyable">${hookAddress}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="email"><button class="btn" type="submit">Make a new address</button></form><br><span class="muted small">Mail sent here wakes the watch. Anyone with the address can, so give it only to senders you mean; a new address retires this one.</span>
? html`<dt>Email address</dt><dd><code class="copyable">${hookAddress}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="email"><button class="btn" type="submit">Make a new address</button></form><br><span class="muted small">Mail sent here wakes the watch. Anyone with the address can, so give it only to senders you mean; a new address retires this one.</span>
<details class="hook-provider"><summary>Use a webhook instead?</summary><p class="muted small">A webhook address with a secret of Watch's own replaces the email address, which stops working at once.</p><form method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a webhook secret</button></form></details></dd>`
<details class="hook-provider"><summary>Use a webhook instead?</summary><p class="muted small">A webhook address with a secret of Watch's own replaces the email address, which stops working at once.</p><form method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a webhook secret</button></form></details></dd>`
: v.source.kind === 'webhook'
: v.source.kind === 'webhook'
? html`<dt>Webhook address</dt><dd><code>${config.baseUrl}/hooks/${v.id}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a new secret</button></form><br><span class="muted small">${hookProvider ? `Signed by ${({ slack: 'Slack', stripe: 'Stripe', discord: 'Discord' })[hookProvider] || hookProvider} with the key you gave. A new secret of Watch's own replaces it.` : 'The secret was shown once, when the watch was made. A new one replaces it at once.'}</span>
? html`<dt>Webhook address</dt><dd><code>${config.baseUrl}/hooks/${v.id}</code> <form class="inline" method="post" action="/watches/${v.id}/webhook-secret"><button class="btn" type="submit">Make a new secret</button></form><br><span class="muted small">${hookProvider ? `Signed by ${({ slack: 'Slack', stripe: 'Stripe', discord: 'Discord' })[hookProvider] || hookProvider} with the key you gave. A new secret of Watch's own replaces it.` : 'The secret was shown once, when the watch was made. A new one replaces it at once.'}</span>
<details class="hook-provider"><summary>Signed by Slack, Stripe or Discord?</summary><p class="muted small">They sign with a key of their own: paste it here and Watch checks their signature instead of its own. It is kept to check each delivery and never shown again.</p>
<details class="hook-provider"><summary>Signed by Slack, Stripe or Discord?</summary><p class="muted small">They sign with a key of their own: paste it here and Watch checks their signature instead of its own. It is kept to check each delivery and never shown again.</p>
<form method="post" action="/watches/${v.id}/webhook-secret"><label>Sender <select name="provider"><option value="slack">Slack (signing secret)</option><option value="stripe">Stripe (endpoint secret, whsec_…)</option><option value="discord">Discord (application public key)</option></select></label> <label>Its signing secret <input name="secret" type="password" autocomplete="off" required maxlength="220"></label> <button class="btn" type="submit">Use it</button></form></details>${mailIn ? html`
<form method="post" action="/watches/${v.id}/webhook-secret"><label>Sender <select name="provider"><option value="slack">Slack (signing secret)</option><option value="stripe">Stripe (endpoint secret, whsec_…)</option><option value="discord">Discord (application public key)</option></select></label> <label>Its signing secret <input name="secret" type="password" autocomplete="off" required maxlength="220"></label> <button class="btn" type="submit">Use it</button></form></details>${mailIn ? html`
-<details class="hook-provider"><summary>Wake it by email?</summary><p class="muted small">The watch gets an address of its own at ${config.mailIn.domain}: mail sent there runs the check with its sender, subject and text. It replaces the webhook secret.</p><form method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="email"><button class="btn" type="submit">Give it an email address</button></form></details>` : ''}</dd>`
+<details class="hook-provider"><summary>Wake it by email?</summary><p class="muted small">The watch gets an address of its own at ${config.mailIn.domain}: mail sent there runs the check with its sender, subject and text. It replaces the webhook secret.</p><form method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="email"><button class="btn" type="submit">Give it an email address</button></form></details>` : ''}
+<details class="hook-provider"><summary>Follow a YouTube channel?</summary><p class="muted small">YouTube tells the watch about every new or changed video on the channel, through its public hub. It replaces the webhook secret.</p>${youtubeForm(v)}</details></dd>`
: html`<dt>Source</dt><dd><code>${v.source.kind}</code> · <code>${v.source.url || (v.source.pattern ? v.source.pattern : '(none)')}</code></dd>`}
: html`<dt>Source</dt><dd><code>${v.source.kind}</code> · <code>${v.source.url || (v.source.pattern ? v.source.pattern : '(none)')}</code></dd>`}
<dt>Cadence</dt><dd>${v.cadence ? W.everyWords(v.cadence.every_sec) : (v.source.kind === 'webhook' ? 'when a delivery arrives' : (v.source.kind === 'event' ? 'when its event arrives' : 'not polled'))}</dd>
<dt>Cadence</dt><dd>${v.cadence ? W.everyWords(v.cadence.every_sec) : (v.source.kind === 'webhook' ? 'when a delivery arrives' : (v.source.kind === 'event' ? 'when its event arrives' : 'not polled'))}</dd>
<dt>Actions</dt><dd>${configuredActions}</dd>
<dt>Actions</dt><dd>${configuredActions}</dd>
@@ -964,6 +972,8 @@ ${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty:
@@ -964,6 +972,8 @@ ${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty:
const row = await owned(req, res);
const row = await owned(req, res);
if (!row) return notFoundPage(req, res);
if (!row) return notFoundPage(req, res);
await registry.remove(row.id, principalOf(req));
await registry.remove(row.id, principalOf(req));
+ // As the API's delete does: its webhook secret or address goes, and a YouTube hub is told to stop.
+ if (ctx.webhookSources) await ctx.webhookSources.forget(row.id);
return res.redirect(303, '/watches');
return res.redirect(303, '/watches');
});
});
@@ -994,18 +1004,35 @@ ${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty:
@@ -994,18 +1004,35 @@ ${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty:
if (v.source.kind !== 'webhook' || !ctx.webhookSources) return detailNotice(req, res, row, 409, 'Only a webhook watch has a secret.');
if (v.source.kind !== 'webhook' || !ctx.webhookSources) return detailNotice(req, res, row, 409, 'Only a webhook watch has a secret.');
// A sender's own signing secret (Slack, Stripe), or a new secret of Watch's own.
// A sender's own signing secret (Slack, Stripe), or a new secret of Watch's own.
const b = req.body || {};
const b = req.body || {};
- const hook = await ctx.webhookSources.issue(row.id, { provider: b.provider || null, secret: b.secret || null });
+ const hook = await ctx.webhookSources.issue(row.id, { provider: b.provider || null, secret: b.secret || null, channel_id: b.channel_id || null });
if (hook.error) return detailNotice(req, res, row, 422, `Not changed: ${hook.error}.`);
if (hook.error) return detailNotice(req, res, row, 422, `Not changed: ${hook.error}.`);
return secretPage(req, res, v, hook, 200);
return secretPage(req, res, v, hook, 200);
});
});
+ /** The form that makes a webhook watch follow a YouTube channel (provider youtube). */
+ function youtubeForm(v) {
+ return html`<form method="post" action="/watches/${v.id}/webhook-secret"><input type="hidden" name="provider" value="youtube"><label>Channel id <input name="channel_id" required pattern="UC[A-Za-z0-9_\\-]{22}" maxlength="24" placeholder="UC…" autocomplete="off" spellcheck="false"></label> <button class="btn" type="submit">Follow it</button><br><span class="muted small">The id in the channel's address, youtube.com/channel/UC…</span></form>`;
+ }
+
/** The address and secret of a webhook watch, shown once: no cache, no index, and never again after this page. */
/** The address and secret of a webhook watch, shown once: no cache, no index, and never again after this page. */
function secretPage(req, res, v, hook, status) {
function secretPage(req, res, v, hook, status) {
res.set('Cache-Control', 'no-store');
res.set('Cache-Control', 'no-store');
res.set('Referrer-Policy', 'no-referrer');
res.set('Referrer-Policy', 'no-referrer');
return page(req, res, {
return page(req, res, {
title: `${v.name}: webhook address`, private: true, crumbs: [{ label: 'Your watches', href: '/watches' }, { label: v.name, href: `/watches/${v.id}` }],
title: `${v.name}: webhook address`, private: true, crumbs: [{ label: 'Your watches', href: '/watches' }, { label: v.name, href: `/watches/${v.id}` }],
- body: hook.provider === 'email' ? html`<h1>Your watch has an email address</h1>
+ body: hook.provider === 'youtube' ? html`<h1>Your watch follows a YouTube channel</h1>
+${notice(hook.subscribed ? 'YouTube\'s hub took the subscription and checks it next, usually within a minute.' : 'Saved, but YouTube\'s hub did not answer: Watch asks it again within the hour.', hook.subscribed ? 'ok' : 'warn')}
+<dl class="facts">
+<dt>Channel</dt><dd><a href="https://www.youtube.com/channel/${hook.channel_id}" rel="noopener">${hook.channel_id}</a></dd>
+</dl>
+<h2>What the watch reads</h2>
+<ul class="facts">
+<li>Each new or changed video runs the check with <code>channel_id</code>, <code>videos</code> (each with <code>id</code>, <code>title</code>, <code>url</code>, <code>author</code>, <code>published</code>, <code>updated</code>) and <code>deleted</code>. For example <code>value_path: "videos.0.title"</code> with the condition "contains: live".</li>
+<li>YouTube's hub signs each notification with a secret Watch gave it; one that is not signed so is ignored.</li>
+<li>The subscription lasts ten days and Watch renews it before then, while the watch is active.</li>
+</ul>
+<p>${W.inWords(v)}</p>
+<p><a class="btn" href="/watches/${v.id}">Back to the watch</a></p>` : hook.provider === 'email' ? html`<h1>Your watch has an email address</h1>
${notice('Mail sent to this address wakes the watch. The address is the key: anyone who has it can wake the watch, so give it only to senders you mean.', 'ok')}
${notice('Mail sent to this address wakes the watch. The address is the key: anyone who has it can wake the watch, so give it only to senders you mean.', 'ok')}
<dl class="facts">
<dl class="facts">
<dt>Address</dt><dd><code class="copyable">${hook.address}</code></dd>
<dt>Address</dt><dd><code class="copyable">${hook.address}</code></dd>
@@ -97,7 +97,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -97,7 +97,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.
// Event sources (plan T18 step 5): watches woken by OpenVibe.Events events, with their subscriptions kept in step.
const eventSources = createEventSources({ db, check, config, fetchImpl, log, now });
const eventSources = createEventSources({ db, check, config, fetchImpl, log, now });
registry.onEventSourceChange(() => eventSources.kick());
registry.onEventSourceChange(() => eventSources.kick());
- const webhookSources = createWebhookSources({ db, registry, check, config, now, log });
+ const webhookSources = createWebhookSources({ db, registry, check, config, now, log, fetchImpl });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
const auth = createAuth({ config, log });
const auth = createAuth({ config, log });
// The public site: sessions in Watch's own database, the Network key to verify a sign-in token
// The public site: sessions in Watch's own database, the Network key to verify a sign-in token
@@ -124,6 +124,8 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -124,6 +124,8 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
try { await observations.prune(); } catch (err) { log.error(`[observations] prune: ${err.message}`); }
try { await observations.prune(); } catch (err) { log.error(`[observations] prune: ${err.message}`); }
try { await eventSources.prune(); } catch (err) { log.error(`[event-sources] prune: ${err.message}`); }
try { await eventSources.prune(); } catch (err) { log.error(`[event-sources] prune: ${err.message}`); }
try { await sessions.prune(); } catch (err) { log.error(`[sessions] prune: ${err.message}`); }
try { await sessions.prune(); } catch (err) { log.error(`[sessions] prune: ${err.message}`); }
+ // YouTube's hub leases end after ten days: renew the ones ending within two (server/webhook-sources.js).
+ try { await webhookSources.renewLeases(); } catch (err) { log.error(`[webhook-sources] renew: ${err.message}`); }
}, 3600 * 1000);
}, 3600 * 1000);
pruneTimer.unref?.();
pruneTimer.unref?.();
const outboxPruneTimer = setInterval(async () => { try { await outbox.outbox.prune(); } catch (err) { log.error(`[outbox] prune: ${err.message}`); } }, 6 * 3600 * 1000);
const outboxPruneTimer = setInterval(async () => { try { await outbox.outbox.prune(); } catch (err) { log.error(`[outbox] prune: ${err.message}`); } }, 6 * 3600 * 1000);
@@ -33,6 +33,15 @@
@@ -33,6 +33,15 @@
* text, text_truncated, attachments } as its JSON input. An unknown address answers 404, and the
* text, text_truncated, attachments } as its JSON input. An unknown address answers 404, and the
* Worker refuses the mail, so the sender gets a bounce. The address is the key: anyone who has it
* Worker refuses the mail, so the sender gets a bounce. The address is the key: anyone who has it
* can wake the watch, and making a new one retires the old.
* can wake the watch, and making a new one retires the old.
+ * YouTube (agent plan A5) POST …/webhook-secret { provider: 'youtube', channel_id: 'UC…' } subscribes the watch to
+ * the channel through YouTube's WebSub hub (WATCH_YOUTUBE_HUB, pubsubhubbub.appspot.com): the callback is
+ * the watch's own /hooks/<id>, signed with a new whk_ secret of Watch's (X-Hub-Signature: sha1=<hex
+ * HMAC>, as the hub sends it). The hub first checks the callback with a GET (hub.mode, hub.topic,
+ * hub.challenge): only this watch's own topic is echoed. Each notification (an Atom entry for a new or
+ * changed video, or a deleted one) runs the check with { provider, channel_id, videos: [{ id, channel_id,
+ * title, url, author, published, updated }], deleted: [{ id, when }] } as its JSON input. A lease runs
+ * ten days and is renewed two days before it ends (renewLeases, hourly). A bad signature is ignored
+ * with a 2xx, as WebSub asks, so the hub does not retry it.
* Answers 200 { state, triggered } after the check; 404 for an unknown id or a watch that is not a webhook
* Answers 200 { state, triggered } after the check; 404 for an unknown id or a watch that is not a webhook
* watch; 401 for a missing or wrong signature (the same answer either way); 409 for a paused watch;
* watch; 401 for a missing or wrong signature (the same answer either way); 409 for a paused watch;
* 429 over the watch's hourly budget (WATCH_WEBHOOK_MAX_PER_HOUR, default 60) with Retry-After;
* 429 over the watch's hourly budget (WATCH_WEBHOOK_MAX_PER_HOUR, default 60) with Retry-After;
@@ -56,6 +65,56 @@ const B32 = '0123456789abcdefghjkmnpqrstvwxyz';
@@ -56,6 +65,56 @@ const B32 = '0123456789abcdefghjkmnpqrstvwxyz';
const newMailLocal = () => `in-${Array.from(crypto.randomBytes(26), (b) => B32[b & 31]).join('')}`;
const newMailLocal = () => `in-${Array.from(crypto.randomBytes(26), (b) => B32[b & 31]).join('')}`;
const newSecret = () => `whk_${crypto.randomBytes(32).toString('base64url')}`;
const newSecret = () => `whk_${crypto.randomBytes(32).toString('base64url')}`;
+
+// YouTube through its WebSub hub (agent plan A5): a channel id, its feed (the hub's topic), and the lease asked for.
+const YT_CHANNEL_RE = /^UC[A-Za-z0-9_-]{22}$/;
+const ytTopic = (channelId) => `https://www.youtube.com/xml/feeds/videos.xml?channel_id=${channelId}`;
+const YT_LEASE_SEC = 10 * 86400;
+const YT_RENEW_BEFORE_MS = 2 * 86400 * 1000;
+
+const XML_ENTITIES = { amp: '&', lt: '<', gt: '>', quot: '"', apos: "'" };
+const unxml = (s) => String(s == null ? '' : s)
+ .replace(/<!\[CDATA\[([\s\S]*?)\]\]>/g, '$1')
+ .replace(/&(#x[0-9a-f]+|#[0-9]+|[a-z]+);/gi, (m, e) => (e[0] === '#'
+ ? String.fromCodePoint(e[1] === 'x' || e[1] === 'X' ? parseInt(e.slice(2), 16) : parseInt(e.slice(1), 10))
+ : (XML_ENTITIES[e.toLowerCase()] != null ? XML_ENTITIES[e.toLowerCase()] : m)))
+ .trim();
+const tag = (xml, name) => { const m = xml.match(new RegExp(`<${name}(?:\\s[^>]*)?>([\\s\\S]*?)</${name}>`)); return m ? unxml(m[1]) : null; };
+
+/**
+ * A YouTube WebSub notification (Atom) as the check's input: the videos it names and the ones it says were deleted.
+ * Only what the feed says: no title or channel is invented when the feed leaves it out.
+ */
+function parseYoutubeFeed(xml) {
+ const s = String(xml || '');
+ const videos = [];
+ for (const m of s.matchAll(/<entry(?:\s[^>]*)?>([\s\S]*?)<\/entry>/g)) {
+ const e = m[1];
+ const id = tag(e, 'yt:videoId');
+ if (!id) continue;
+ const link = e.match(/<link\b[^>]*\brel="alternate"[^>]*\bhref="([^"]+)"/) || e.match(/<link\b[^>]*\bhref="([^"]+)"[^>]*\brel="alternate"/);
+ const author = e.match(/<author>[\s\S]*?<name>([\s\S]*?)<\/name>/);
+ videos.push({
+ id, channel_id: tag(e, 'yt:channelId'), title: tag(e, 'title'),
+ url: link ? unxml(link[1]) : `https://www.youtube.com/watch?v=${id}`,
+ author: author ? unxml(author[1]) : null, published: tag(e, 'published'), updated: tag(e, 'updated'),
+ });
+ }
+ const deleted = [...s.matchAll(/<at:deleted-entry\b[^>]*>/g)].map((m) => {
+ const ref = (m[0].match(/\bref="([^"]*)"/) || [])[1] || '';
+ const when = (m[0].match(/\bwhen="([^"]*)"/) || [])[1] || null;
+ return { id: unxml(ref).replace(/^yt:video:/, ''), when: when ? unxml(when) : null };
+ }).filter((d) => d.id);
+ const channel = videos.find((v) => v.channel_id) || null;
+ return { provider: 'youtube', channel_id: channel ? channel.channel_id : null, videos, deleted };
+}
+
+/** WebSub's signature: X-Hub-Signature: <sha1|sha256|sha384|sha512>=<hex HMAC of the body with the secret>. */
+function websubVerified(secret, raw, headers) {
+ const m = String(headers['x-hub-signature'] || '').match(/^(sha1|sha256|sha384|sha512)=([0-9a-f]+)$/i);
+ if (!m) return false;
+ return safeEqual(m[2].toLowerCase(), crypto.createHmac(m[1].toLowerCase(), secret).update(raw).digest('hex'));
+}
const endpointFor = (watchId) => `/internal/hooks/${watchId}`;
const endpointFor = (watchId) => `/internal/hooks/${watchId}`;
function safeEqual(a, b) {
function safeEqual(a, b) {
@@ -107,6 +166,7 @@ function verified(secret, raw, headers, { provider = null, now = Date.now() } =
@@ -107,6 +166,7 @@ function verified(secret, raw, headers, { provider = null, now = Date.now() } =
if (provider === 'slack') return slackVerified(secret, raw, headers, now);
if (provider === 'slack') return slackVerified(secret, raw, headers, now);
if (provider === 'stripe') return stripeVerified(secret, raw, headers, now);
if (provider === 'stripe') return stripeVerified(secret, raw, headers, now);
if (provider === 'discord') return discordVerified(secret, raw, headers, now);
if (provider === 'discord') return discordVerified(secret, raw, headers, now);
+ if (provider === 'youtube') return websubVerified(secret, raw, headers);
if (provider) return false;
if (provider) return false;
const tg = headers['x-telegram-bot-api-secret-token'];
const tg = headers['x-telegram-bot-api-secret-token'];
if (tg != null && safeEqual(String(tg), secret)) return true;
if (tg != null && safeEqual(String(tg), secret)) return true;
@@ -120,7 +180,7 @@ function verified(secret, raw, headers, { provider = null, now = Date.now() } =
@@ -120,7 +180,7 @@ function verified(secret, raw, headers, { provider = null, now = Date.now() } =
return false;
return false;
}
}
-function createWebhookSources({ db, registry, check, config, now = () => Date.now(), log = console }) {
+function createWebhookSources({ db, registry, check, config, now = () => Date.now(), log = console, fetchImpl = globalThis.fetch }) {
const maxPerHour = Number.isInteger(config.webhookSources && config.webhookSources.maxPerWatchPerHour) && config.webhookSources.maxPerWatchPerHour > 0
const maxPerHour = Number.isInteger(config.webhookSources && config.webhookSources.maxPerWatchPerHour) && config.webhookSources.maxPerWatchPerHour > 0
? config.webhookSources.maxPerWatchPerHour : 60;
? config.webhookSources.maxPerWatchPerHour : 60;
const counts = new Map(); // watch id → { hour, n }
const counts = new Map(); // watch id → { hour, n }
@@ -128,10 +188,17 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
@@ -128,10 +188,17 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
domain: String((config.mailIn && config.mailIn.domain) || 'openvibe.watch').toLowerCase(),
domain: String((config.mailIn && config.mailIn.domain) || 'openvibe.watch').toLowerCase(),
ingressSecret: (config.mailIn && config.mailIn.ingressSecret) || null,
ingressSecret: (config.mailIn && config.mailIn.ingressSecret) || null,
};
};
+ const youtubeHub = String((config.youtube && config.youtube.hub) || 'https://pubsubhubbub.appspot.com/subscribe');
const st = {
const st = {
set: db.prepare(`INSERT INTO webhook_secrets (watch_id, secret, created_at, provider) VALUES (?, ?, ?, ?)
set: db.prepare(`INSERT INTO webhook_secrets (watch_id, secret, created_at, provider) VALUES (?, ?, ?, ?)
- ON CONFLICT (watch_id) DO UPDATE SET secret = excluded.secret, created_at = excluded.created_at, provider = excluded.provider`),
- get: db.prepare('SELECT secret, provider FROM webhook_secrets WHERE watch_id = ?'),
+ ON CONFLICT (watch_id) DO UPDATE SET secret = excluded.secret, created_at = excluded.created_at, provider = excluded.provider,
+ topic = NULL, lease_until = NULL`),
+ setYoutube: db.prepare(`INSERT INTO webhook_secrets (watch_id, secret, created_at, provider, topic, lease_until) VALUES (?, ?, ?, 'youtube', ?, NULL)
+ ON CONFLICT (watch_id) DO UPDATE SET secret = excluded.secret, created_at = excluded.created_at, provider = 'youtube',
+ topic = excluded.topic, lease_until = NULL`),
+ lease: db.prepare("UPDATE webhook_secrets SET lease_until = ? WHERE watch_id = ? AND provider = 'youtube' AND topic = ?"),
+ due: db.prepare("SELECT watch_id, secret, topic FROM webhook_secrets WHERE provider = 'youtube' AND (lease_until IS NULL OR lease_until < ?) ORDER BY watch_id LIMIT 200"),
+ get: db.prepare('SELECT secret, provider, topic, lease_until FROM webhook_secrets WHERE watch_id = ?'),
drop: db.prepare('DELETE FROM webhook_secrets WHERE watch_id = ?'),
drop: db.prepare('DELETE FROM webhook_secrets WHERE watch_id = ?'),
watch: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),
watch: db.prepare('SELECT * FROM watches WHERE id = ? AND deleted_at IS NULL'),
byMail: db.prepare("SELECT watch_id FROM webhook_secrets WHERE provider = 'email' AND secret = ?"),
byMail: db.prepare("SELECT watch_id FROM webhook_secrets WHERE provider = 'email' AND secret = ?"),
@@ -142,8 +209,36 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
@@ -142,8 +209,36 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
* secret ({ provider: 'slack' | 'stripe', secret }), that secret is kept instead and never echoed back. A bad
* secret ({ provider: 'slack' | 'stripe', secret }), that secret is kept instead and never echoed back. A bad
* provider or secret answers { error } and changes nothing.
* provider or secret answers { error } and changes nothing.
*/
*/
- async function issue(watchId, { provider = null, secret: theirs = null } = {}) {
+ /** Ask YouTube's hub to (un)subscribe a watch's callback. True when the hub took the request (it checks it next). */
+ async function hub(mode, watchId, secret, topic) {
+ const form = new URLSearchParams({ 'hub.callback': `${config.baseUrl}/hooks/${watchId}`, 'hub.topic': topic, 'hub.mode': mode, 'hub.verify': 'async' });
+ if (mode === 'subscribe') { form.set('hub.secret', secret); form.set('hub.lease_seconds', String(YT_LEASE_SEC)); }
+ try {
+ const r = await fetchImpl(youtubeHub, { method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, body: form.toString(), signal: AbortSignal.timeout(15000) });
+ if (r.status === 202 || r.status === 204) return true;
+ log.warn(`[webhook-sources] youtube ${mode} ${watchId}: the hub answered ${r.status}`);
+ } catch (err) { log.warn(`[webhook-sources] youtube ${mode} ${watchId}: ${err.message}`); }
+ return false;
+ }
+ /** A watch leaving YouTube: the hub is told, best effort (a lease left alone ends in ten days anyway). */
+ async function leaveYoutube(watchId) {
+ const kept = await st.get.get(watchId);
+ if (kept && kept.provider === 'youtube' && kept.topic) await hub('unsubscribe', watchId, kept.secret, kept.topic);
+ }
+
+ async function issue(watchId, { provider = null, secret: theirs = null, channel_id: channelId = null } = {}) {
const url = `${config.baseUrl}/hooks/${watchId}`;
const url = `${config.baseUrl}/hooks/${watchId}`;
+ if (provider === 'youtube') {
+ const ch = String(channelId || '').trim();
+ if (!YT_CHANNEL_RE.test(ch)) return { error: 'a YouTube channel id starts with UC and has 24 characters (youtube.com/channel/UC…)' };
+ const before = await st.get.get(watchId);
+ if (before && before.provider === 'youtube' && before.topic && before.topic !== ytTopic(ch)) await hub('unsubscribe', watchId, before.secret, before.topic);
+ const secret = newSecret();
+ await st.setYoutube.run(watchId, secret, now(), ytTopic(ch));
+ const asked = await hub('subscribe', watchId, secret, ytTopic(ch));
+ return { url, provider: 'youtube', channel_id: ch, subscribed: asked, signature: 'YouTube\'s hub signs each notification with a secret Watch gave it (X-Hub-Signature)' };
+ }
+ await leaveYoutube(watchId);
if (provider === 'email') {
if (provider === 'email') {
if (!mail.ingressSecret) return { error: 'email in is not set up on this server' };
if (!mail.ingressSecret) return { error: 'email in is not set up on this server' };
const local = newMailLocal();
const local = newMailLocal();
@@ -162,7 +257,48 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
@@ -162,7 +257,48 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
await st.set.run(watchId, secret, now(), null);
await st.set.run(watchId, secret, now(), null);
return { url, secret, signature: 'X-Hub-Signature-256: sha256=<hex HMAC-SHA256 of the body with the secret>, Authorization: Bearer <secret>, or Telegram\'s X-Telegram-Bot-Api-Secret-Token' };
return { url, secret, signature: 'X-Hub-Signature-256: sha256=<hex HMAC-SHA256 of the body with the secret>, Authorization: Bearer <secret>, or Telegram\'s X-Telegram-Bot-Api-Secret-Token' };
}
}
- async function forget(watchId) { await st.drop.run(watchId); }
+ async function forget(watchId) { await leaveYoutube(watchId); await st.drop.run(watchId); }
+ /** The channel a YouTube watch follows and when its lease ends (shown on the watch's page), or null. */
+ async function youtubeOf(watchId) {
+ const row = await st.get.get(watchId);
+ if (!row || row.provider !== 'youtube' || !row.topic) return null;
+ const m = row.topic.match(/channel_id=(UC[A-Za-z0-9_-]{22})$/);
+ return { channel_id: m ? m[1] : null, lease_until: row.lease_until == null ? null : Number(row.lease_until) };
+ }
+ /** Renew every YouTube lease that ends within two days, or that the hub never confirmed (hourly, server/index.js). */
+ async function renewLeases() {
+ let asked = 0;
+ for (const row of await st.due.all(now() + YT_RENEW_BEFORE_MS)) {
+ const w = await st.watch.get(row.watch_id);
+ if (!w || w.status !== 'active') continue;
+ if (await hub('subscribe', row.watch_id, row.secret, row.topic)) asked++;
+ }
+ return asked;
+ }
+
+ /**
+ * GET /hooks/:id — the hub's check of a (un)subscription: its challenge is echoed only for the topic this watch
+ * follows (subscribe), or one it no longer follows (unsubscribe). A subscribe confirmed sets the lease's end.
+ */
+ async function verifyIntent(req, res) {
+ const id = String(req.params.id || '');
+ const q = new URL(req.originalUrl || req.url, 'http://watch.invalid').searchParams;
+ const mode = q.get('hub.mode');
+ const topic = q.get('hub.topic');
+ const challenge = q.get('hub.challenge');
+ res.setHeader('Cache-Control', 'no-store');
+ const kept = /^wch_[0-9A-HJKMNP-TV-Z]{26}$/.test(id) ? await st.get.get(id) : null;
+ const follows = !!(kept && kept.provider === 'youtube' && kept.topic && kept.topic === topic);
+ if (mode === 'denied') { log.warn(`[webhook-sources] youtube ${id}: the hub denied the subscription (${String(q.get('hub.reason') || '').slice(0, 200)})`); return res.status(200).end(); }
+ if (!challenge || challenge.length > 512 || !topic) return problem(res, 404, 'watch.not_found', 'no subscription to confirm here');
+ if (mode === 'subscribe' && follows) {
+ const lease = Number(q.get('hub.lease_seconds'));
+ await st.lease.run(now() + (Number.isFinite(lease) && lease > 0 ? Math.min(lease, 30 * 86400) : YT_LEASE_SEC) * 1000, id, topic);
+ return res.status(200).type('text/plain').send(challenge);
+ }
+ if (mode === 'unsubscribe' && !follows) return res.status(200).type('text/plain').send(challenge);
+ return problem(res, 404, 'watch.not_found', 'this watch does not follow that topic');
+ }
/** The watch's email address, when it has one (shown to its owner on the watch's page). */
/** The watch's email address, when it has one (shown to its owner on the watch's page). */
async function addressOf(watchId) {
async function addressOf(watchId) {
const row = await st.get.get(watchId);
const row = await st.get.get(watchId);
@@ -207,7 +343,11 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
@@ -207,7 +343,11 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
if (!row || !source || source.kind !== 'webhook') return problem(res, 404, 'watch.not_found', 'no webhook watch at this address');
if (!row || !source || source.kind !== 'webhook') return problem(res, 404, 'watch.not_found', 'no webhook watch at this address');
const kept = await st.get.get(id) || {};
const kept = await st.get.get(id) || {};
const provider = kept.provider || null;
const provider = kept.provider || null;
- if (!verified(kept.secret || null, raw, req.headers, { provider, now: now() })) return problem(res, 401, 'watch.bad_signature', 'the delivery is not signed with this watch\'s secret');
+ if (!verified(kept.secret || null, raw, req.headers, { provider, now: now() })) {
+ // WebSub: a notification whose signature does not match is ignored but acknowledged, so the hub does not resend it.
+ if (provider === 'youtube') { log.warn(`[webhook-sources] youtube ${id}: a notification with a bad signature was ignored`); res.setHeader('Cache-Control', 'no-store'); return res.status(202).json({ watch_id: id, ignored: true }); }
+ return problem(res, 401, 'watch.bad_signature', 'the delivery is not signed with this watch\'s secret');
+ }
if (provider === 'slack') {
if (provider === 'slack') {
// Slack checks a new address with a challenge it wants back, and resends an event it did not see answered
// Slack checks a new address with a challenge it wants back, and resends an event it did not see answered
// within three seconds: the first delivery is the one that runs the check.
// within three seconds: the first delivery is the one that runs the check.
@@ -233,9 +373,13 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
@@ -233,9 +373,13 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
}
}
if (row.status !== 'active') return problem(res, 409, 'watch.disabled', `the watch is ${row.status}`);
if (row.status !== 'active') return problem(res, 409, 'watch.disabled', `the watch is ${row.status}`);
if (!underBudget(id)) return problem(res, 429, 'watch.rate_limited', `at most ${maxPerHour} deliveries an hour wake one watch`, { 'Retry-After': String(retryAfter()) });
if (!underBudget(id)) return problem(res, 429, 'watch.rate_limited', `at most ${maxPerHour} deliveries an hour wake one watch`, { 'Retry-After': String(retryAfter()) });
+ // A YouTube notification is Atom: the check reads it as JSON (parseYoutubeFeed).
+ const input = provider === 'youtube'
+ ? { body: Buffer.from(JSON.stringify(parseYoutubeFeed(raw.toString('utf8')))), content_type: 'application/json' }
+ : { body: raw, content_type: req.headers['content-type'] || null };
let out;
let out;
try {
try {
- out = await check.run(id, { trigger: 'webhook', input: { body: raw, content_type: req.headers['content-type'] || null } });
+ out = await check.run(id, { trigger: 'webhook', input });
} catch (err) {
} catch (err) {
log.error(`[webhook-sources] ${id}: ${err.message}`);
log.error(`[webhook-sources] ${id}: ${err.message}`);
return problem(res, 500, 'watch.internal', 'the check failed; try again');
return problem(res, 500, 'watch.internal', 'the check failed; try again');
@@ -280,7 +424,7 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
@@ -280,7 +424,7 @@ function createWebhookSources({ db, registry, check, config, now = () => Date.no
return res.json({ watch_id: row.id, state: out && out.run ? out.run.state : null, triggered: !!(out && out.met) });
return res.json({ watch_id: row.id, state: out && out.run ? out.run.state : null, triggered: !!(out && out.met) });
}
}
- return { issue, forget, providerOf, addressOf, receive, receiveMail, endpointFor, verified, SECRET_RE, PROVIDERS };
+ return { issue, forget, providerOf, addressOf, youtubeOf, renewLeases, verifyIntent, receive, receiveMail, endpointFor, verified, SECRET_RE, PROVIDERS };
}
}
/** The mail Worker's signature: X-OpenVibe-Mail-Signature: t=<time>,v1=<hex HMAC-SHA256(secret, time + "." + body)>. */
/** The mail Worker's signature: X-OpenVibe-Mail-Signature: t=<time>,v1=<hex HMAC-SHA256(secret, time + "." + body)>. */
@@ -292,4 +436,4 @@ function mailSigned(secret, raw, headers, nowMs = Date.now()) {
@@ -292,4 +436,4 @@ function mailSigned(secret, raw, headers, nowMs = Date.now()) {
return parts.filter(([k]) => k === 'v1').some(([, v]) => safeEqual(v || '', want));
return parts.filter(([k]) => k === 'v1').some(([, v]) => safeEqual(v || '', want));
}
}
-module.exports = { createWebhookSources, verified, mailSigned, endpointFor, newSecret, newMailLocal, MAIL_LOCAL_RE, PROVIDERS };
+module.exports = { createWebhookSources, verified, mailSigned, endpointFor, newSecret, newMailLocal, MAIL_LOCAL_RE, PROVIDERS, parseYoutubeFeed, websubVerified, YT_CHANNEL_RE };
new file mode 100644
new file mode 100644
@@ -0,0 +1,149 @@
@@ -0,0 +1,149 @@
+'use strict';
+/**
+ * A webhook watch that follows a YouTube channel (agent plan A5) through YouTube's WebSub hub: the watch subscribes,
+ * the hub checks the callback with a GET (only this watch's own topic is echoed), each notification is signed with
+ * the secret Watch gave the hub (X-Hub-Signature) and becomes JSON for the check, a bad signature is acknowledged and
+ * ignored, leases are renewed before they end, and leaving YouTube tells the hub. The hub here is a local fake.
+ */
+const assert = require('assert');
+const crypto = require('crypto');
+const http = require('http');
+const { boot, request, userToken, suite, newUser } = require('./helpers');
+const { parseYoutubeFeed, websubVerified } = require('../server/webhook-sources');
+
+const t = suite('youtube-websub');
+let svc, hub, watch;
+const me = newUser();
+const ME = userToken(me);
+const OTHER = userToken(newUser());
+const CHANNEL = 'UC' + 'AbCdEfGhIjKlMnOpQrStUv';
+const TOPIC = `https://www.youtube.com/xml/feeds/videos.xml?channel_id=${CHANNEL}`;
+const api = (method, p, opts = {}) => request(svc.base, method, p, opts);
+const runs = async (id) => await svc.db.prepare('SELECT * FROM check_runs WHERE watch_id = ? ORDER BY rid').all(id);
+const sign = (secret, body, algo = 'sha1') => `${algo}=${crypto.createHmac(algo, secret).update(body).digest('hex')}`;
+const feed = (title) => `<?xml version='1.0' encoding='UTF-8'?>
+<feed xmlns:yt="http://www.youtube.com/xml/schemas/2015" xmlns="http://www.w3.org/2005/Atom">
+ <link rel="hub" href="https://pubsubhubbub.appspot.com"/>
+ <title>YouTube video feed</title>
+ <entry>
+ <id>yt:video:VIDEO_ID_01</id>
+ <yt:videoId>VIDEO_ID_01</yt:videoId>
+ <yt:channelId>${CHANNEL}</yt:channelId>
+ <title>${title}</title>
+ <link rel="alternate" href="https://www.youtube.com/watch?v=VIDEO_ID_01"/>
+ <author><name>Channel & Friends</name><uri>https://www.youtube.com/channel/${CHANNEL}</uri></author>
+ <published>2026-11-03T08:00:00+00:00</published>
+ <updated>2026-11-03T08:00:05+00:00</updated>
+ </entry>
+</feed>`;
+
+function startHub() {
+ const calls = [];
+ const server = http.createServer((req, res) => {
+ let body = '';
+ req.on('data', (c) => { body += c; });
+ req.on('end', () => { calls.push(Object.fromEntries(new URLSearchParams(body))); res.writeHead(202); res.end(); });
+ });
+ return new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve({ calls, url: `http://127.0.0.1:${server.address().port}/subscribe`, close: () => new Promise((r) => server.close(r)) })));
+}
+
+t('boot', async () => {
+ hub = await startHub();
+ svc = await boot({ env: { WATCH_YOUTUBE_HUB: hub.url } });
+});
+
+t('the feed: videos with their fields, entities decoded, deleted entries; the signature in every WebSub algorithm', () => {
+ const f = parseYoutubeFeed(feed('Going LIVE <now>'));
+ assert.deepStrictEqual(f.videos, [{ id: 'VIDEO_ID_01', channel_id: CHANNEL, title: 'Going LIVE <now>', url: 'https://www.youtube.com/watch?v=VIDEO_ID_01', author: 'Channel & Friends', published: '2026-11-03T08:00:00+00:00', updated: '2026-11-03T08:00:05+00:00' }]);
+ assert.strictEqual(f.channel_id, CHANNEL);
+ const gone = parseYoutubeFeed(`<feed xmlns:at="http://purl.org/atompub/tombstones/1.0"><at:deleted-entry ref="yt:video:OLD_VIDEO_1" when="2026-11-03T09:00:00+00:00"><link href="https://www.youtube.com/watch?v=OLD_VIDEO_1"/></at:deleted-entry></feed>`);
+ assert.deepStrictEqual([gone.videos, gone.deleted], [[], [{ id: 'OLD_VIDEO_1', when: '2026-11-03T09:00:00+00:00' }]]);
+ assert.deepStrictEqual(parseYoutubeFeed('not xml at all'), { provider: 'youtube', channel_id: null, videos: [], deleted: [] });
+ const body = Buffer.from('x');
+ for (const algo of ['sha1', 'sha256', 'sha512']) assert.strictEqual(websubVerified('whk_s', body, { 'x-hub-signature': sign('whk_s', body, algo) }), true, algo);
+ assert.strictEqual(websubVerified('whk_s', body, { 'x-hub-signature': sign('whk_t', body) }), false);
+ assert.strictEqual(websubVerified('whk_s', body, {}), false);
+});
+
+t('the owner points a webhook watch at a channel: Watch subscribes at the hub with its callback, topic and a secret', async () => {
+ const made = await api('POST', '/api/v1/watches', { token: ME, body: { name: 'channel goes live', source: { kind: 'webhook', endpoint: '/internal/hooks/x' }, cadence: null, extraction: { kind: 'json', value_path: 'videos.0.title' }, condition: { op: 'contains', value: 'LIVE' }, action: [{ kind: 'notification' }] } });
+ assert.strictEqual(made.status, 201, JSON.stringify(made.body));
+ watch = made.body.watch;
+ assert.strictEqual((await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: { provider: 'youtube', channel_id: 'not-a-channel' } })).status, 422);
+ assert.strictEqual((await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: OTHER, body: { provider: 'youtube', channel_id: CHANNEL } })).status, 404);
+ const set = await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: { provider: 'youtube', channel_id: CHANNEL } });
+ assert.strictEqual(set.status, 200, JSON.stringify(set.body));
+ assert.deepStrictEqual([set.body.webhook.provider, set.body.webhook.channel_id, set.body.webhook.subscribed], ['youtube', CHANNEL, true]);
+ assert.ok(!('secret' in set.body.webhook), 'the hub\'s secret is never shown');
+ const asked = hub.calls[hub.calls.length - 1];
+ assert.deepStrictEqual([asked['hub.mode'], asked['hub.topic'], asked['hub.callback'], asked['hub.verify'], asked['hub.lease_seconds']], ['subscribe', TOPIC, `${svc.config.baseUrl}/hooks/${watch.id}`, 'async', '864000']);
+ const kept = await svc.db.prepare('SELECT secret, provider, topic, lease_until FROM webhook_secrets WHERE watch_id = ?').get(watch.id);
+ assert.deepStrictEqual([kept.secret, kept.provider, kept.topic, kept.lease_until], [asked['hub.secret'], 'youtube', TOPIC, null]);
+ assert.match(kept.secret, /^whk_/);
+});
+
+t('the hub checks the callback: its challenge comes back only for this watch\'s topic, and the lease is kept', async () => {
+ const verify = (q) => fetch(`${svc.base}/hooks/${watch.id}?${new URLSearchParams(q)}`);
+ const wrong = await verify({ 'hub.mode': 'subscribe', 'hub.topic': TOPIC.replace(CHANNEL, 'UC' + 'z'.repeat(22)), 'hub.challenge': 'c1', 'hub.lease_seconds': '864000' });
+ assert.strictEqual(wrong.status, 404, 'another channel is not confirmed');
+ const unsub = await verify({ 'hub.mode': 'unsubscribe', 'hub.topic': TOPIC, 'hub.challenge': 'c2' });
+ assert.strictEqual(unsub.status, 404, 'an unsubscribe of the topic it still follows is refused');
+ const ok = await verify({ 'hub.mode': 'subscribe', 'hub.topic': TOPIC, 'hub.challenge': 'abc123XYZ', 'hub.lease_seconds': '864000' });
+ assert.strictEqual(ok.status, 200);
+ assert.strictEqual(await ok.text(), 'abc123XYZ');
+ const lease = Number((await svc.db.prepare('SELECT lease_until FROM webhook_secrets WHERE watch_id = ?').get(watch.id)).lease_until);
+ assert.ok(Math.abs(lease - (Date.now() + 864000 * 1000)) < 60_000, 'the lease ends ten days from now');
+ assert.strictEqual((await fetch(`${svc.base}/hooks/wch_00000000000000000000000000?hub.mode=subscribe&hub.topic=x&hub.challenge=y`)).status, 404);
+ const page = await svc.get(`/watches/${watch.id}`, { as: { subject: me, username: 'yt-owner' } });
+ assert.ok(page.text.includes(`youtube.com/channel/${CHANNEL}`) && page.text.includes('confirmed it until'), 'the watch\'s page shows the channel and the lease');
+});
+
+t('a signed notification runs the check with the feed as JSON; a badly signed one is acknowledged and ignored', async () => {
+ const secret = (await svc.db.prepare('SELECT secret FROM webhook_secrets WHERE watch_id = ?').get(watch.id)).secret;
+ const before = (await runs(watch.id)).length;
+ const bad = await fetch(`${svc.base}/hooks/${watch.id}`, { method: 'POST', body: feed('Going LIVE'), headers: { 'Content-Type': 'application/atom+xml', 'X-Hub-Signature': sign('whk_wrong', feed('Going LIVE')) } });
+ assert.strictEqual(bad.status, 202, 'WebSub: acknowledged, so the hub does not resend it');
+ assert.strictEqual((await bad.json()).ignored, true);
+ assert.strictEqual((await runs(watch.id)).length, before, 'and nothing ran');
+ const body = feed('Going LIVE: the big build');
+ const r = await fetch(`${svc.base}/hooks/${watch.id}`, { method: 'POST', body, headers: { 'Content-Type': 'application/atom+xml', 'X-Hub-Signature': sign(secret, body) } });
+ assert.strictEqual(r.status, 200);
+ const out = await r.json();
+ assert.strictEqual(out.triggered, true, JSON.stringify(out));
+ const quiet = feed('A cooking video');
+ const r2 = await fetch(`${svc.base}/hooks/${watch.id}`, { method: 'POST', body: quiet, headers: { 'Content-Type': 'application/atom+xml', 'X-Hub-Signature': sign(secret, quiet) } });
+ assert.strictEqual((await r2.json()).triggered, false, 'a title without LIVE does not meet the condition');
+});
+
+t('leases ending within two days are renewed; leaving YouTube tells the hub, which may then confirm the unsubscribe', async () => {
+ await svc.db.prepare('UPDATE webhook_secrets SET lease_until = ? WHERE watch_id = ?').run(Date.now() + 86400 * 1000, watch.id);
+ const n = hub.calls.length;
+ assert.strictEqual(await svc.webhookSources.renewLeases(), 1);
+ assert.deepStrictEqual([hub.calls.length, hub.calls[n]['hub.mode'], hub.calls[n]['hub.topic']], [n + 1, 'subscribe', TOPIC]);
+ assert.strictEqual(await svc.webhookSources.renewLeases(), 1, 'until the hub confirms again, it is still due');
+ await svc.db.prepare('UPDATE webhook_secrets SET lease_until = ? WHERE watch_id = ?').run(Date.now() + 9 * 86400 * 1000, watch.id);
+ assert.strictEqual(await svc.webhookSources.renewLeases(), 0, 'a lease with days left is left alone');
+ const back = await api('POST', `/api/v1/watches/${watch.id}/webhook-secret`, { token: ME, body: {} });
+ assert.strictEqual(back.status, 200);
+ const last = hub.calls[hub.calls.length - 1];
+ assert.deepStrictEqual([last['hub.mode'], last['hub.topic'], last['hub.secret']], ['unsubscribe', TOPIC, undefined]);
+ const kept = await svc.db.prepare('SELECT provider, topic, lease_until FROM webhook_secrets WHERE watch_id = ?').get(watch.id);
+ assert.deepStrictEqual([kept.provider, kept.topic, kept.lease_until], [null, null, null]);
+ const confirm = await fetch(`${svc.base}/hooks/${watch.id}?${new URLSearchParams({ 'hub.mode': 'unsubscribe', 'hub.topic': TOPIC, 'hub.challenge': 'bye' })}`);
+ assert.deepStrictEqual([confirm.status, await confirm.text()], [200, 'bye']);
+});
+
+t('deleting a YouTube watch on its page tells the hub and drops its secret', async () => {
+ const made = await api('POST', '/api/v1/watches', { token: ME, body: { name: 'to delete', source: { kind: 'webhook', endpoint: '/internal/hooks/x' }, cadence: null, extraction: { kind: 'json', value_path: 'videos.0.title' }, condition: { op: 'contains', value: 'x' }, action: [{ kind: 'notification' }] } });
+ const id = made.body.watch.id;
+ assert.strictEqual((await api('POST', `/api/v1/watches/${id}/webhook-secret`, { token: ME, body: { provider: 'youtube', channel_id: CHANNEL } })).status, 200);
+ const del = await svc.get(`/watches/${id}/delete`, { as: { subject: me, username: 'yt-owner' }, form: {} });
+ assert.strictEqual(del.status, 303);
+ const last = hub.calls[hub.calls.length - 1];
+ assert.deepStrictEqual([last['hub.mode'], last['hub.callback']], ['unsubscribe', `${svc.config.baseUrl}/hooks/${id}`]);
+ assert.strictEqual(await svc.db.prepare('SELECT 1 FROM webhook_secrets WHERE watch_id = ?').get(id), undefined);
+});
+
+t('stop', async () => { await svc.stop(); await hub.close(); });
+
+t.run();