From d144070198e71a86564447b5308a4325dfafad0c Mon Sep 17 00:00:00 2001 From: Darren Date: Thu, 9 Jul 2026 11:54:40 +0000 Subject: [PATCH 01/14] Add livestream service --- .devcontainer.json | 8 +- .env.example | 2 + .gitignore | 1 + package.json | 1 + pnpm-lock.yaml | 37 ++++ src/app.module.ts | 8 +- src/livestream/index.ts | 3 + src/livestream/livestream.channels.ts | 27 +++ src/livestream/livestream.controller.ts | 18 ++ src/livestream/livestream.module.ts | 10 ++ src/livestream/livestream.service.ts | 228 ++++++++++++++++++++++++ 11 files changed, 338 insertions(+), 5 deletions(-) create mode 100644 .env.example create mode 100644 src/livestream/index.ts create mode 100644 src/livestream/livestream.channels.ts create mode 100644 src/livestream/livestream.controller.ts create mode 100644 src/livestream/livestream.module.ts create mode 100644 src/livestream/livestream.service.ts diff --git a/.devcontainer.json b/.devcontainer.json index af9fd3c..c827180 100644 --- a/.devcontainer.json +++ b/.devcontainer.json @@ -4,11 +4,11 @@ "name": "Open Home Foundation - API", "image": "mcr.microsoft.com/devcontainers/typescript-node:24-bookworm", "forwardPorts": [ - 443 + 3000 ], "portsAttributes": { - "443": { - "label": "HTTPS server" + "3000": { + "label": "web-api (HTTP)" } }, "customizations": { @@ -31,4 +31,4 @@ }, "postCreateCommand": "pnpm install", "remoteUser": "node" -} +} \ No newline at end of file diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..f90ed8e --- /dev/null +++ b/.env.example @@ -0,0 +1,2 @@ +# YouTube Data API v3 key (https://console.cloud.google.com/apis/credentials) +YOUTUBE_API_KEY= diff --git a/.gitignore b/.gitignore index dd6e803..7f5ed9b 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ node_modules/ dist/ *.log .DS_Store +.env diff --git a/package.json b/package.json index 35cda97..bad3664 100644 --- a/package.json +++ b/package.json @@ -10,6 +10,7 @@ }, "dependencies": { "@nestjs/common": "^11.1.19", + "@nestjs/config": "^4.0.4", "@nestjs/core": "^11.1.19", "@nestjs/platform-express": "^11.1.19", "@nestjs/platform-socket.io": "^11.1.19", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 59ced72..4deffa8 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -11,6 +11,9 @@ importers: '@nestjs/common': specifier: ^11.1.19 version: 11.1.27(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/config': + specifier: ^4.0.4 + version: 4.0.4(@nestjs/common@11.1.27(reflect-metadata@0.2.2)(rxjs@7.8.2))(rxjs@7.8.2) '@nestjs/core': specifier: ^11.1.19 version: 11.1.27(@nestjs/common@11.1.27(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.1.27)(@nestjs/websockets@11.1.27)(reflect-metadata@0.2.2)(rxjs@7.8.2) @@ -302,6 +305,12 @@ packages: class-validator: optional: true + '@nestjs/config@4.0.4': + resolution: {integrity: sha512-CJPjNitr0bAufSEnRe2N+JbnVmMmDoo6hvKCPzXgZoGwJSmp/dZPk9f/RMbuD/+Q1ZJPjwsRpq0vxna++Knwow==} + peerDependencies: + '@nestjs/common': ^10.0.0 || ^11.0.0 + rxjs: ^7.1.0 + '@nestjs/core@11.1.27': resolution: {integrity: sha512-K6DX7hcqmZdeXkv7tsPakKBRCgqL19a4mtbX4FluY0hWtFdtPKp6lbe+lb8gWPfvLdbOWr/CPScn7BSjBX+Ecg==} engines: {node: '>= 20'} @@ -721,6 +730,18 @@ packages: resolution: {integrity: sha512-X07nttJQkwkfKfvTPG/KSnE2OMdcUCao6+eXF3wmnIQRn2aPAHH3VxDbDOdegkd6JbPsXqShpvEOHfAT+nCNwQ==} engines: {node: '>=0.3.1'} + dotenv-expand@12.0.3: + resolution: {integrity: sha512-uc47g4b+4k/M/SeaW1y4OApx+mtLWl92l5LMPP0GNXctZqELk+YGgOPIIC5elYmUH4OuoK3JLhuRUYegeySiFA==} + engines: {node: '>=12'} + + dotenv@16.6.1: + resolution: {integrity: sha512-uBq4egWHTcTt33a72vpSG0z3HnPuIl6NqYcTrKEg2azoEyl2hpW0zqlxysq2pK9HlDIHyHyakeYaYnSAwd8bow==} + engines: {node: '>=12'} + + dotenv@17.4.1: + resolution: {integrity: sha512-k8DaKGP6r1G30Lx8V4+pCsLzKr8vLmV2paqEj1Y55GdAgJuIqpRp5FfajGF8KtwMxCz9qJc6wUIJnm053d/WCw==} + engines: {node: '>=12'} + dunder-proto@1.0.1: resolution: {integrity: sha512-KIN/nDJBQRcXw0MLVhZE9iQHmG68qAVIBg9CqmUYjmQIhgij9U5MFvrqkUL5FbtyyzZuOeOt0zdeRe4UY7ct+A==} engines: {node: '>= 0.4'} @@ -1815,6 +1836,14 @@ snapshots: transitivePeerDependencies: - supports-color + '@nestjs/config@4.0.4(@nestjs/common@11.1.27(reflect-metadata@0.2.2)(rxjs@7.8.2))(rxjs@7.8.2)': + dependencies: + '@nestjs/common': 11.1.27(reflect-metadata@0.2.2)(rxjs@7.8.2) + dotenv: 17.4.1 + dotenv-expand: 12.0.3 + lodash: 4.18.1 + rxjs: 7.8.2 + '@nestjs/core@11.1.27(@nestjs/common@11.1.27(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.1.27)(@nestjs/websockets@11.1.27)(reflect-metadata@0.2.2)(rxjs@7.8.2)': dependencies: '@nestjs/common': 11.1.27(reflect-metadata@0.2.2)(rxjs@7.8.2) @@ -2244,6 +2273,14 @@ snapshots: diff@4.0.4: {} + dotenv-expand@12.0.3: + dependencies: + dotenv: 16.6.1 + + dotenv@16.6.1: {} + + dotenv@17.4.1: {} + dunder-proto@1.0.1: dependencies: call-bind-apply-helpers: 1.0.2 diff --git a/src/app.module.ts b/src/app.module.ts index 1f51aca..464ad93 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -1,11 +1,17 @@ import { Module } from '@nestjs/common'; +import { ConfigModule } from '@nestjs/config'; import { AppController } from './app.controller'; import { AppGateway } from './app.gateway'; import { HealthModule } from './health'; import { getVersionInfo } from './health/version'; +import { LivestreamModule } from './livestream'; @Module({ - imports: [HealthModule.register({ version: getVersionInfo() })], + imports: [ + ConfigModule.forRoot({ isGlobal: true }), + HealthModule.register({ version: getVersionInfo() }), + LivestreamModule, + ], controllers: [AppController], providers: [AppGateway], }) diff --git a/src/livestream/index.ts b/src/livestream/index.ts new file mode 100644 index 0000000..b6c63be --- /dev/null +++ b/src/livestream/index.ts @@ -0,0 +1,3 @@ +export * from './livestream.channels'; +export * from './livestream.module'; +export * from './livestream.service'; diff --git a/src/livestream/livestream.channels.ts b/src/livestream/livestream.channels.ts new file mode 100644 index 0000000..e18eff9 --- /dev/null +++ b/src/livestream/livestream.channels.ts @@ -0,0 +1,27 @@ +export interface Channel { + /** URL-safe identifier used in the API path, e.g. "home-assistant". */ + slug: string; + /** Human-friendly display name. */ + name: string; + /** YouTube handle without the leading "@". */ + handle: string; +} + +/** + * Open Home Foundation project channels tracked for livestreams. + * Add a new project by appending an entry here — no other code changes needed. + */ +export const CHANNELS: readonly Channel[] = [ + { slug: 'home-assistant', name: 'Home Assistant', handle: 'home_assistant' }, + { slug: 'esphome', name: 'ESPHome', handle: 'esphomeio' }, + { + slug: 'open-home-foundation', + name: 'Open Home Foundation', + handle: 'OpenHomeFndn', + }, + { + slug: 'music-assistant', + name: 'Music Assistant', + handle: 'musicassistantio', + }, +]; diff --git a/src/livestream/livestream.controller.ts b/src/livestream/livestream.controller.ts new file mode 100644 index 0000000..0155ca9 --- /dev/null +++ b/src/livestream/livestream.controller.ts @@ -0,0 +1,18 @@ +import { Controller, Get, Param } from '@nestjs/common'; + +import { LivestreamInfo, LivestreamService } from './livestream.service'; + +@Controller('livestream') +export class LivestreamController { + constructor(private readonly livestream: LivestreamService) {} + + @Get() + getAll(): Promise { + return this.livestream.getAll(); + } + + @Get(':slug') + getStatus(@Param('slug') slug: string): Promise { + return this.livestream.getStatus(slug); + } +} diff --git a/src/livestream/livestream.module.ts b/src/livestream/livestream.module.ts new file mode 100644 index 0000000..e67af08 --- /dev/null +++ b/src/livestream/livestream.module.ts @@ -0,0 +1,10 @@ +import { Module } from '@nestjs/common'; + +import { LivestreamController } from './livestream.controller'; +import { LivestreamService } from './livestream.service'; + +@Module({ + controllers: [LivestreamController], + providers: [LivestreamService], +}) +export class LivestreamModule {} diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts new file mode 100644 index 0000000..0475ace --- /dev/null +++ b/src/livestream/livestream.service.ts @@ -0,0 +1,228 @@ +import { + Injectable, + Logger, + NotFoundException, + ServiceUnavailableException, +} from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; + +import { Channel, CHANNELS } from './livestream.channels'; + +export type LivestreamStatus = 'live' | 'upcoming' | 'past' | 'none'; + +export interface LivestreamInfo { + /** Channel slug, e.g. "home-assistant". */ + channel: string; + /** Human-friendly channel name. */ + channelName: string; + status: LivestreamStatus; + title?: string; + url?: string; + /** ISO 8601 scheduled start time; present when status is "upcoming". */ + startTime?: string; + /** ISO 8601 timestamp of when the data was last fetched from YouTube. */ + fetchedAt: string; +} + +interface SearchResult { + videoId: string; + title: string; +} + +interface CacheEntry { + info: LivestreamInfo; + cachedAt: number; +} + +const API_BASE = 'https://www.googleapis.com/youtube/v3'; +const CACHE_TTL_MS = 5 * 60 * 1000; +const PAST_WINDOW_MS = 24 * 60 * 60 * 1000; + +const watchUrl = (videoId: string) => + `https://www.youtube.com/watch?v=${videoId}`; + +@Injectable() +export class LivestreamService { + private readonly logger = new Logger(LivestreamService.name); + private readonly channelsBySlug = new Map( + CHANNELS.map((c) => [c.slug, c] as const), + ); + private readonly cache = new Map(); + private readonly channelIds = new Map(); + private readonly refreshing = new Map>(); + + constructor(private readonly config: ConfigService) {} + + async getAll(): Promise { + return Promise.all(CHANNELS.map((channel) => this.getStatus(channel.slug))); + } + + async getStatus(slug: string): Promise { + const channel = this.channelsBySlug.get(slug); + if (!channel) { + throw new NotFoundException(`Unknown channel "${slug}"`); + } + + const cached = this.cache.get(slug); + if (cached && Date.now() - cached.cachedAt < CACHE_TTL_MS) { + return cached.info; + } + + let pending = this.refreshing.get(slug); + if (!pending) { + pending = this.refresh(channel).finally(() => + this.refreshing.delete(slug), + ); + this.refreshing.set(slug, pending); + } + + try { + return await pending; + } catch (err) { + if (cached) { + this.logger.warn( + `Refresh failed for ${slug}, serving stale data: ${err}`, + ); + return cached.info; + } + this.logger.error( + `Failed to fetch livestream status for ${slug}: ${err}`, + ); + throw new ServiceUnavailableException( + 'Unable to fetch livestream status from YouTube', + ); + } + } + + private async refresh(channel: Channel): Promise { + const channelId = await this.resolveChannelId(channel); + const base = { + channel: channel.slug, + channelName: channel.name, + fetchedAt: new Date().toISOString(), + }; + + const [live] = await this.search(channelId, 'live', 1); + if (live) { + return this.setCache(channel.slug, { + ...base, + status: 'live', + title: live.title, + url: watchUrl(live.videoId), + }); + } + + const upcoming = await this.search(channelId, 'upcoming'); + if (upcoming.length > 0) { + const details = await this.videoDetails(upcoming.map((v) => v.videoId)); + const next = details + .map((d: any) => ({ + videoId: d.id as string, + title: d.snippet?.title as string, + start: d.liveStreamingDetails?.scheduledStartTime as + | string + | undefined, + })) + .filter((d) => d.start) + .sort((a, b) => Date.parse(a.start!) - Date.parse(b.start!))[0]; + if (next) { + return this.setCache(channel.slug, { + ...base, + status: 'upcoming', + title: next.title, + url: watchUrl(next.videoId), + startTime: next.start, + }); + } + } + + const [completed] = await this.search(channelId, 'completed', 1); + if (completed) { + const [details] = await this.videoDetails([completed.videoId]); + const endedAt: string | undefined = + details?.liveStreamingDetails?.actualEndTime; + if (endedAt && Date.now() - Date.parse(endedAt) <= PAST_WINDOW_MS) { + return this.setCache(channel.slug, { + ...base, + status: 'past', + title: details.snippet?.title ?? completed.title, + url: watchUrl(completed.videoId), + }); + } + } + + return this.setCache(channel.slug, { ...base, status: 'none' }); + } + + private setCache(slug: string, info: LivestreamInfo): LivestreamInfo { + this.cache.set(slug, { info, cachedAt: Date.now() }); + return info; + } + + private async resolveChannelId(channel: Channel): Promise { + const cached = this.channelIds.get(channel.slug); + if (cached) { + return cached; + } + const data = await this.apiGet('channels', { + part: 'id', + forHandle: channel.handle, + }); + const id: string | undefined = data.items?.[0]?.id; + if (!id) { + throw new Error(`Channel not found for handle @${channel.handle}`); + } + this.channelIds.set(channel.slug, id); + return id; + } + + private async search( + channelId: string, + eventType: 'live' | 'upcoming' | 'completed', + maxResults = 5, + ): Promise { + const data = await this.apiGet('search', { + part: 'snippet', + channelId, + eventType, + type: 'video', + order: 'date', + maxResults: String(maxResults), + }); + return (data.items ?? []).map((item: any) => ({ + videoId: item.id.videoId, + title: item.snippet.title, + })); + } + + private async videoDetails(videoIds: string[]): Promise { + if (videoIds.length === 0) { + return []; + } + const data = await this.apiGet('videos', { + part: 'snippet,liveStreamingDetails', + id: videoIds.join(','), + }); + return data.items ?? []; + } + + private async apiGet( + path: string, + params: Record, + ): Promise { + const key = this.config.get('YOUTUBE_API_KEY'); + if (!key) { + throw new Error('YOUTUBE_API_KEY is not set'); + } + const url = new URL(`${API_BASE}/${path}`); + for (const [name, value] of Object.entries(params)) { + url.searchParams.set(name, value); + } + url.searchParams.set('key', key); + const res = await fetch(url); + if (!res.ok) { + throw new Error(`YouTube API ${path} request failed: ${res.status}`); + } + return res.json(); + } +} From 5ff8c2711dcc555f4c48431963caf8d8fad99248 Mon Sep 17 00:00:00 2001 From: Darren Griffin Date: Thu, 9 Jul 2026 13:11:35 +0100 Subject: [PATCH 02/14] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/livestream/livestream.service.ts | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts index 0475ace..11614de 100644 --- a/src/livestream/livestream.service.ts +++ b/src/livestream/livestream.service.ts @@ -54,7 +54,23 @@ export class LivestreamService { constructor(private readonly config: ConfigService) {} async getAll(): Promise { - return Promise.all(CHANNELS.map((channel) => this.getStatus(channel.slug))); + return Promise.all( + CHANNELS.map(async (channel) => { + try { + return await this.getStatus(channel.slug); + } catch (err) { + this.logger.warn( + `Failed to fetch livestream status for ${channel.slug}, returning status "none": ${err}`, + ); + return { + channel: channel.slug, + channelName: channel.name, + status: 'none', + fetchedAt: new Date().toISOString(), + }; + } + }), + ); } async getStatus(slug: string): Promise { From dfb64ca1bdd812e25d53db18b334c18576d9c0ff Mon Sep 17 00:00:00 2001 From: Darren Griffin Date: Thu, 9 Jul 2026 13:11:45 +0100 Subject: [PATCH 03/14] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/livestream/livestream.service.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts index 11614de..2cfaf54 100644 --- a/src/livestream/livestream.service.ts +++ b/src/livestream/livestream.service.ts @@ -235,9 +235,12 @@ export class LivestreamService { url.searchParams.set(name, value); } url.searchParams.set('key', key); - const res = await fetch(url); + const res = await fetch(url, { signal: AbortSignal.timeout(10_000) }); if (!res.ok) { - throw new Error(`YouTube API ${path} request failed: ${res.status}`); + const body = await res.text().catch(() => ''); + throw new Error( + `YouTube API ${path} request failed: ${res.status} ${res.statusText}${body ? ` - ${body}` : ''}`, + ); } return res.json(); } From 519a22022fc94dfec09f807c4da168a2555663d8 Mon Sep 17 00:00:00 2001 From: Darren Griffin Date: Thu, 9 Jul 2026 13:20:15 +0100 Subject: [PATCH 04/14] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/livestream/livestream.service.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts index 2cfaf54..94bae6c 100644 --- a/src/livestream/livestream.service.ts +++ b/src/livestream/livestream.service.ts @@ -97,7 +97,7 @@ export class LivestreamService { } catch (err) { if (cached) { this.logger.warn( - `Refresh failed for ${slug}, serving stale data: ${err}`, + `Refresh failed for ${slug}, serving stale data: ${err instanceof Error ? err.stack ?? err.message : String(err)}`, ); return cached.info; } From 290fc9ac4426372515e67a21adfc62457703c8e1 Mon Sep 17 00:00:00 2001 From: Darren Griffin Date: Thu, 9 Jul 2026 13:20:21 +0100 Subject: [PATCH 05/14] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/livestream/livestream.service.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts index 94bae6c..8a89f6d 100644 --- a/src/livestream/livestream.service.ts +++ b/src/livestream/livestream.service.ts @@ -102,7 +102,7 @@ export class LivestreamService { return cached.info; } this.logger.error( - `Failed to fetch livestream status for ${slug}: ${err}`, + `Failed to fetch livestream status for ${slug}: ${err instanceof Error ? err.stack ?? err.message : String(err)}`, ); throw new ServiceUnavailableException( 'Unable to fetch livestream status from YouTube', From 5c9b8d4d940016cf0baac55d2bbae4e8de7c4196 Mon Sep 17 00:00:00 2001 From: Darren Griffin Date: Thu, 9 Jul 2026 13:20:43 +0100 Subject: [PATCH 06/14] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/livestream/livestream.service.ts | 52 +++++++++++++++++++--------- 1 file changed, 36 insertions(+), 16 deletions(-) diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts index 8a89f6d..17f1257 100644 --- a/src/livestream/livestream.service.ts +++ b/src/livestream/livestream.service.ts @@ -54,23 +54,43 @@ export class LivestreamService { constructor(private readonly config: ConfigService) {} async getAll(): Promise { - return Promise.all( - CHANNELS.map(async (channel) => { - try { - return await this.getStatus(channel.slug); - } catch (err) { - this.logger.warn( - `Failed to fetch livestream status for ${channel.slug}, returning status "none": ${err}`, - ); - return { - channel: channel.slug, - channelName: channel.name, - status: 'none', - fetchedAt: new Date().toISOString(), - }; - } - }), + const results = await Promise.allSettled( + CHANNELS.map((channel) => this.getStatus(channel.slug)), ); + + const infos: LivestreamInfo[] = []; + let failures = 0; + + for (let i = 0; i < results.length; i++) { + const channel = CHANNELS[i]; + const res = results[i]; + + if (res.status === 'fulfilled') { + infos.push(res.value); + continue; + } + + failures++; + this.logger.warn( + `Failed to fetch livestream status for ${channel.slug}, returning status "none": ${res.reason instanceof Error ? res.reason.stack ?? res.reason.message : String(res.reason)}`, + ); + infos.push({ + channel: channel.slug, + channelName: channel.name, + status: 'none', + fetchedAt: new Date().toISOString(), + }); + } + + // If nothing could be fetched (and thus nothing could be served from cache), + // return an error instead of silently reporting "none" for every channel. + if (failures === CHANNELS.length) { + throw new ServiceUnavailableException( + 'Unable to fetch livestream status from YouTube', + ); + } + + return infos; } async getStatus(slug: string): Promise { From e4bbc26953f839c065661b4bd9f09e0880aeec47 Mon Sep 17 00:00:00 2001 From: Darren Date: Thu, 9 Jul 2026 14:06:09 +0000 Subject: [PATCH 07/14] Switch implementation --- .env.example | 9 + src/livestream/index.ts | 1 + src/livestream/livestream.controller.ts | 4 +- src/livestream/livestream.module.ts | 6 +- src/livestream/livestream.service.ts | 451 ++++++++++++++++-------- src/livestream/pubsub.controller.ts | 90 +++++ src/livestream/pubsub.service.ts | 97 +++++ src/main.ts | 2 +- 8 files changed, 501 insertions(+), 159 deletions(-) create mode 100644 src/livestream/pubsub.controller.ts create mode 100644 src/livestream/pubsub.service.ts diff --git a/.env.example b/.env.example index f90ed8e..9bc9d70 100644 --- a/.env.example +++ b/.env.example @@ -1,2 +1,11 @@ # YouTube Data API v3 key (https://console.cloud.google.com/apis/credentials) YOUTUBE_API_KEY= + +# Publicly reachable base URL of this service, used as the WebSub callback +# target for YouTube push notifications (e.g. https://api.openhomefoundation.org). +# Leave blank in local dev without a public tunnel; status will still update via +# the periodic reconcile. +PUBLIC_BASE_URL= + +# Optional shared secret for verifying WebSub push payloads (HMAC signature). +PUBSUB_SECRET= diff --git a/src/livestream/index.ts b/src/livestream/index.ts index b6c63be..dadc7b9 100644 --- a/src/livestream/index.ts +++ b/src/livestream/index.ts @@ -1,3 +1,4 @@ export * from './livestream.channels'; export * from './livestream.module'; export * from './livestream.service'; +export * from './pubsub.service'; diff --git a/src/livestream/livestream.controller.ts b/src/livestream/livestream.controller.ts index 0155ca9..bd7524f 100644 --- a/src/livestream/livestream.controller.ts +++ b/src/livestream/livestream.controller.ts @@ -7,12 +7,12 @@ export class LivestreamController { constructor(private readonly livestream: LivestreamService) {} @Get() - getAll(): Promise { + getAll(): LivestreamInfo[] { return this.livestream.getAll(); } @Get(':slug') - getStatus(@Param('slug') slug: string): Promise { + getStatus(@Param('slug') slug: string): LivestreamInfo { return this.livestream.getStatus(slug); } } diff --git a/src/livestream/livestream.module.ts b/src/livestream/livestream.module.ts index e67af08..7e18081 100644 --- a/src/livestream/livestream.module.ts +++ b/src/livestream/livestream.module.ts @@ -2,9 +2,11 @@ import { Module } from '@nestjs/common'; import { LivestreamController } from './livestream.controller'; import { LivestreamService } from './livestream.service'; +import { PubSubController } from './pubsub.controller'; +import { PubSubService } from './pubsub.service'; @Module({ - controllers: [LivestreamController], - providers: [LivestreamService], + controllers: [LivestreamController, PubSubController], + providers: [LivestreamService, PubSubService], }) export class LivestreamModule {} diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts index 17f1257..b94b717 100644 --- a/src/livestream/livestream.service.ts +++ b/src/livestream/livestream.service.ts @@ -2,7 +2,8 @@ import { Injectable, Logger, NotFoundException, - ServiceUnavailableException, + OnModuleDestroy, + OnModuleInit, } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; @@ -20,226 +21,367 @@ export interface LivestreamInfo { url?: string; /** ISO 8601 scheduled start time; present when status is "upcoming". */ startTime?: string; - /** ISO 8601 timestamp of when the data was last fetched from YouTube. */ - fetchedAt: string; + /** ISO 8601 timestamp of when this channel's state was last updated. */ + updatedAt: string; } -interface SearchResult { +/** A livestream video we are tracking the state of. */ +interface TrackedVideo { videoId: string; title: string; -} - -interface CacheEntry { - info: LivestreamInfo; - cachedAt: number; + status: LivestreamStatus; + scheduledStartTime?: string; + actualEndTime?: string; + updatedAt: number; } const API_BASE = 'https://www.googleapis.com/youtube/v3'; -const CACHE_TTL_MS = 5 * 60 * 1000; const PAST_WINDOW_MS = 24 * 60 * 60 * 1000; +/** Re-scan channels' RSS feeds to discover newly scheduled/published videos. */ +const DISCOVERY_INTERVAL_MS = 5 * 60 * 1000; +/** Base cadence for transition polling (used when a stream is live/imminent). */ +const RECONCILE_TICK_MS = 10 * 1000; +/** When idle, only reconcile every Nth tick (10s * 6 = 60s). */ +const IDLE_TICKS = 6; +/** Treat an upcoming stream as "imminent" within this window of its start. */ +const SOON_WINDOW_MS = 15 * 60 * 1000; const watchUrl = (videoId: string) => `https://www.youtube.com/watch?v=${videoId}`; +const videoIdPattern = /([^<]+)<\/yt:videoId>/g; + @Injectable() -export class LivestreamService { +export class LivestreamService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(LivestreamService.name); private readonly channelsBySlug = new Map( CHANNELS.map((c) => [c.slug, c] as const), ); - private readonly cache = new Map(); + /** Derived, ready-to-serve status per channel slug. */ + private readonly state = new Map(); + /** Tracked livestream videos per channel slug. */ + private readonly tracked = new Map>(); private readonly channelIds = new Map(); - private readonly refreshing = new Map>(); + private readonly slugByChannelId = new Map(); + private reconcileTimer?: ReturnType; + private discoveryTimer?: ReturnType; + private tickCount = 0; constructor(private readonly config: ConfigService) {} - async getAll(): Promise { - const results = await Promise.allSettled( - CHANNELS.map((channel) => this.getStatus(channel.slug)), + async onModuleInit(): Promise { + // Seed initial state from each channel's (free) RSS feed so we are not + // blank until the first push arrives. + await this.discovery(); + this.discoveryTimer = setInterval( + () => void this.discovery(), + DISCOVERY_INTERVAL_MS, ); + this.reconcileTimer = setInterval(() => void this.tick(), RECONCILE_TICK_MS); + } - const infos: LivestreamInfo[] = []; - let failures = 0; + onModuleDestroy(): void { + if (this.discoveryTimer) { + clearInterval(this.discoveryTimer); + } + if (this.reconcileTimer) { + clearInterval(this.reconcileTimer); + } + } - for (let i = 0; i < results.length; i++) { - const channel = CHANNELS[i]; - const res = results[i]; + getAll(): LivestreamInfo[] { + return CHANNELS.map((channel) => this.readState(channel)); + } - if (res.status === 'fulfilled') { - infos.push(res.value); - continue; - } + getStatus(slug: string): LivestreamInfo { + const channel = this.channelsBySlug.get(slug); + if (!channel) { + throw new NotFoundException(`Unknown channel "${slug}"`); + } + return this.readState(channel); + } - failures++; - this.logger.warn( - `Failed to fetch livestream status for ${channel.slug}, returning status "none": ${res.reason instanceof Error ? res.reason.stack ?? res.reason.message : String(res.reason)}`, - ); - infos.push({ - channel: channel.slug, - channelName: channel.name, - status: 'none', - fetchedAt: new Date().toISOString(), - }); + /** Resolve a channel's YouTube channel ID (UC…), caching the result. */ + async resolveChannelId(channel: Channel): Promise { + const cached = this.channelIds.get(channel.slug); + if (cached) { + return cached; } + const data = await this.apiGet('channels', { + part: 'id', + forHandle: channel.handle, + }); + const id: string | undefined = data.items?.[0]?.id; + if (!id) { + throw new Error(`Channel not found for handle @${channel.handle}`); + } + this.channelIds.set(channel.slug, id); + this.slugByChannelId.set(id, channel.slug); + return id; + } - // If nothing could be fetched (and thus nothing could be served from cache), - // return an error instead of silently reporting "none" for every channel. - if (failures === CHANNELS.length) { - throw new ServiceUnavailableException( - 'Unable to fetch livestream status from YouTube', - ); + /** + * Handle a WebSub push: fetch the referenced videos' live details and update + * state. Costs 1 quota unit per (batched) lookup. + */ + async handleNotification( + channelId: string, + videoIds: string[], + ): Promise { + const slug = this.slugByChannelId.get(channelId); + if (!slug) { + this.logger.warn(`Notification for unknown channel ${channelId}`); + return; + } + const items = await this.videoDetails(videoIds); + for (const item of items) { + this.track(slug, item); } + this.recompute(slug); + } - return infos; + private readState(channel: Channel): LivestreamInfo { + return this.state.get(channel.slug) ?? this.defaultInfo(channel); } - async getStatus(slug: string): Promise { - const channel = this.channelsBySlug.get(slug); - if (!channel) { - throw new NotFoundException(`Unknown channel "${slug}"`); + private defaultInfo(channel: Channel): LivestreamInfo { + return { + channel: channel.slug, + channelName: channel.name, + status: 'none', + updatedAt: new Date().toISOString(), + }; + } + + /** + * Discover videos from every channel's (free) RSS feed and classify them + * with a cheap videos.list lookup. Runs on startup and on a timer to catch + * newly scheduled streams that WebSub may not push. + */ + private async discovery(): Promise { + await Promise.all( + CHANNELS.map((channel) => + this.discoverChannel(channel).catch((err) => + this.logger.error(`Discovery failed for ${channel.slug}: ${err}`), + ), + ), + ); + } + + private async discoverChannel(channel: Channel): Promise { + const channelId = await this.resolveChannelId(channel); + const videoIds = await this.fetchFeedVideoIds(channelId); + if (videoIds.length > 0) { + const items = await this.videoDetails(videoIds); + for (const item of items) { + this.track(channel.slug, item); + } } + this.recompute(channel.slug); + } - const cached = this.cache.get(slug); - if (cached && Date.now() - cached.cachedAt < CACHE_TTL_MS) { - return cached.info; + /** + * Adaptive transition poller: checks every tick while a stream is live or + * imminent, and only every IDLE_TICKS-th tick otherwise, so idle channels + * cost effectively no quota. + */ + private async tick(): Promise { + this.tickCount = (this.tickCount + 1) % IDLE_TICKS; + if (!this.hasActiveStream() && this.tickCount !== 0) { + return; } + await this.reconcile(); + } - let pending = this.refreshing.get(slug); - if (!pending) { - pending = this.refresh(channel).finally(() => - this.refreshing.delete(slug), - ); - this.refreshing.set(slug, pending); + private hasActiveStream(): boolean { + const now = Date.now(); + for (const videos of this.tracked.values()) { + for (const v of videos.values()) { + if (v.status === 'live') { + return true; + } + if ( + v.status === 'upcoming' && + v.scheduledStartTime && + Date.parse(v.scheduledStartTime) - now <= SOON_WINDOW_MS + ) { + return true; + } + } } + return false; + } + /** Re-check tracked upcoming/live videos to catch live/ended transitions. */ + private async reconcile(): Promise { + const active: string[] = []; + for (const videos of this.tracked.values()) { + for (const v of videos.values()) { + if (v.status === 'live' || v.status === 'upcoming') { + active.push(v.videoId); + } + } + } + if (active.length === 0) { + return; + } try { - return await pending; - } catch (err) { - if (cached) { - this.logger.warn( - `Refresh failed for ${slug}, serving stale data: ${err instanceof Error ? err.stack ?? err.message : String(err)}`, - ); - return cached.info; + const items = await this.videoDetails(active); + const touched = new Set(); + for (const item of items) { + const slug = this.slugByChannelId.get(item.snippet?.channelId); + if (slug) { + this.track(slug, item); + touched.add(slug); + } } - this.logger.error( - `Failed to fetch livestream status for ${slug}: ${err instanceof Error ? err.stack ?? err.message : String(err)}`, - ); - throw new ServiceUnavailableException( - 'Unable to fetch livestream status from YouTube', - ); + for (const slug of touched) { + this.recompute(slug); + } + } catch (err) { + this.logger.warn(`Reconcile failed: ${err}`); } } - private async refresh(channel: Channel): Promise { - const channelId = await this.resolveChannelId(channel); + /** Record (or drop) a video's livestream state from a YouTube API item. */ + private track(slug: string, item: any): void { + const videoId: string = item.id; + const details = item.liveStreamingDetails; + const videos = this.tracked.get(slug) ?? new Map(); + this.tracked.set(slug, videos); + + // Not a livestream (regular upload) — ignore. + if (!details) { + videos.delete(videoId); + return; + } + + let status: LivestreamStatus; + if (details.actualEndTime) { + status = 'past'; + } else if (details.actualStartTime) { + status = 'live'; + } else if (details.scheduledStartTime) { + status = 'upcoming'; + } else { + status = 'none'; + } + + // Drop streams that ended more than the past-window ago. + if ( + status === 'past' && + Date.now() - Date.parse(details.actualEndTime) > PAST_WINDOW_MS + ) { + videos.delete(videoId); + return; + } + if (status === 'none') { + videos.delete(videoId); + return; + } + + videos.set(videoId, { + videoId, + title: item.snippet?.title ?? '', + status, + scheduledStartTime: details.scheduledStartTime, + actualEndTime: details.actualEndTime, + updatedAt: Date.now(), + }); + } + + /** Recompute the derived channel status from its tracked videos. */ + private recompute(slug: string): void { + const channel = this.channelsBySlug.get(slug); + if (!channel) { + return; + } const base = { channel: channel.slug, channelName: channel.name, - fetchedAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), }; + const videos = [...(this.tracked.get(slug)?.values() ?? [])]; + const now = Date.now(); - const [live] = await this.search(channelId, 'live', 1); + const live = videos.find((v) => v.status === 'live'); if (live) { - return this.setCache(channel.slug, { + this.state.set(slug, { ...base, status: 'live', title: live.title, url: watchUrl(live.videoId), }); + return; } - const upcoming = await this.search(channelId, 'upcoming'); - if (upcoming.length > 0) { - const details = await this.videoDetails(upcoming.map((v) => v.videoId)); - const next = details - .map((d: any) => ({ - videoId: d.id as string, - title: d.snippet?.title as string, - start: d.liveStreamingDetails?.scheduledStartTime as - | string - | undefined, - })) - .filter((d) => d.start) - .sort((a, b) => Date.parse(a.start!) - Date.parse(b.start!))[0]; - if (next) { - return this.setCache(channel.slug, { - ...base, - status: 'upcoming', - title: next.title, - url: watchUrl(next.videoId), - startTime: next.start, - }); - } + const upcoming = videos + .filter((v) => v.status === 'upcoming' && v.scheduledStartTime) + .sort( + (a, b) => + Date.parse(a.scheduledStartTime!) - Date.parse(b.scheduledStartTime!), + )[0]; + if (upcoming) { + this.state.set(slug, { + ...base, + status: 'upcoming', + title: upcoming.title, + url: watchUrl(upcoming.videoId), + startTime: upcoming.scheduledStartTime, + }); + return; } - const [completed] = await this.search(channelId, 'completed', 1); - if (completed) { - const [details] = await this.videoDetails([completed.videoId]); - const endedAt: string | undefined = - details?.liveStreamingDetails?.actualEndTime; - if (endedAt && Date.now() - Date.parse(endedAt) <= PAST_WINDOW_MS) { - return this.setCache(channel.slug, { - ...base, - status: 'past', - title: details.snippet?.title ?? completed.title, - url: watchUrl(completed.videoId), - }); - } + const past = videos + .filter( + (v) => + v.status === 'past' && + v.actualEndTime && + now - Date.parse(v.actualEndTime) <= PAST_WINDOW_MS, + ) + .sort( + (a, b) => Date.parse(b.actualEndTime!) - Date.parse(a.actualEndTime!), + )[0]; + if (past) { + this.state.set(slug, { + ...base, + status: 'past', + title: past.title, + url: watchUrl(past.videoId), + }); + return; } - return this.setCache(channel.slug, { ...base, status: 'none' }); - } - - private setCache(slug: string, info: LivestreamInfo): LivestreamInfo { - this.cache.set(slug, { info, cachedAt: Date.now() }); - return info; + this.state.set(slug, { ...base, status: 'none' }); } - private async resolveChannelId(channel: Channel): Promise { - const cached = this.channelIds.get(channel.slug); - if (cached) { - return cached; - } - const data = await this.apiGet('channels', { - part: 'id', - forHandle: channel.handle, - }); - const id: string | undefined = data.items?.[0]?.id; - if (!id) { - throw new Error(`Channel not found for handle @${channel.handle}`); + /** Fetch recent video IDs from a channel's RSS feed (free, no quota). */ + private async fetchFeedVideoIds(channelId: string): Promise { + const res = await fetch( + `https://www.youtube.com/feeds/videos.xml?channel_id=${channelId}`, + { signal: AbortSignal.timeout(10_000) }, + ); + if (!res.ok) { + throw new Error(`Feed request failed: ${res.status}`); } - this.channelIds.set(channel.slug, id); - return id; - } - - private async search( - channelId: string, - eventType: 'live' | 'upcoming' | 'completed', - maxResults = 5, - ): Promise { - const data = await this.apiGet('search', { - part: 'snippet', - channelId, - eventType, - type: 'video', - order: 'date', - maxResults: String(maxResults), - }); - return (data.items ?? []).map((item: any) => ({ - videoId: item.id.videoId, - title: item.snippet.title, - })); + const xml = await res.text(); + return [...xml.matchAll(videoIdPattern)].map((m) => m[1]); } private async videoDetails(videoIds: string[]): Promise { - if (videoIds.length === 0) { + const unique = [...new Set(videoIds)]; + if (unique.length === 0) { return []; } - const data = await this.apiGet('videos', { - part: 'snippet,liveStreamingDetails', - id: videoIds.join(','), - }); - return data.items ?? []; + const items: any[] = []; + for (let i = 0; i < unique.length; i += 50) { + const data = await this.apiGet('videos', { + part: 'snippet,liveStreamingDetails', + id: unique.slice(i, i + 50).join(','), + }); + items.push(...(data.items ?? [])); + } + return items; } private async apiGet( @@ -259,7 +401,8 @@ export class LivestreamService { if (!res.ok) { const body = await res.text().catch(() => ''); throw new Error( - `YouTube API ${path} request failed: ${res.status} ${res.statusText}${body ? ` - ${body}` : ''}`, + `YouTube API ${path} request failed: ${res.status} ${res.statusText}` + + (body ? ` - ${body}` : ''), ); } return res.json(); diff --git a/src/livestream/pubsub.controller.ts b/src/livestream/pubsub.controller.ts new file mode 100644 index 0000000..43dec57 --- /dev/null +++ b/src/livestream/pubsub.controller.ts @@ -0,0 +1,90 @@ +import { createHmac, timingSafeEqual } from 'node:crypto'; +import { IncomingMessage } from 'node:http'; + +import { + Controller, + ForbiddenException, + Get, + Header, + HttpCode, + Logger, + Post, + Query, + RawBodyRequest, + Req, +} from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; + +import { LivestreamService } from './livestream.service'; + +const channelIdPattern = /([^<]+)<\/yt:channelId>/; +const videoIdPattern = /([^<]+)<\/yt:videoId>/g; + +/** + * WebSub (PubSubHubbub) callback for YouTube push notifications. + * https://developers.google.com/youtube/v3/guides/push_notifications + */ +@Controller('pubsub') +export class PubSubController { + private readonly logger = new Logger(PubSubController.name); + + constructor( + private readonly config: ConfigService, + private readonly livestream: LivestreamService, + ) {} + + /** Hub subscription verification — echo back the challenge. */ + @Get() + @Header('Content-Type', 'text/plain') + verify(@Query() query: Record): string { + const challenge = query['hub.challenge']; + if (!challenge) { + throw new ForbiddenException('Missing hub.challenge'); + } + this.logger.log( + `Verified ${query['hub.mode']} for ${query['hub.topic']}`, + ); + return challenge; + } + + /** Content-distribution notification with an Atom feed payload. */ + @Post() + @HttpCode(204) + notify(@Req() req: RawBodyRequest): void { + const raw = req.rawBody?.toString('utf8') ?? ''; + if (!this.verifySignature(req, raw)) { + throw new ForbiddenException('Invalid signature'); + } + + const channelId = channelIdPattern.exec(raw)?.[1]; + const videoIds = [...raw.matchAll(videoIdPattern)].map((m) => m[1]); + if (!channelId || videoIds.length === 0) { + return; + } + + // Fire-and-forget: acknowledge the hub quickly. + void this.livestream + .handleNotification(channelId, videoIds) + .catch((err) => this.logger.error(`Failed to handle push: ${err}`)); + } + + private verifySignature( + req: RawBodyRequest, + raw: string, + ): boolean { + const secret = this.config.get('PUBSUB_SECRET'); + if (!secret) { + return true; // Verification disabled when no secret configured. + } + const header = req.headers['x-hub-signature']; + const value = Array.isArray(header) ? header[0] : header; + const [algo, signature] = value?.split('=') ?? []; + if (!algo || !signature) { + return false; + } + const expected = createHmac(algo, secret).update(raw).digest('hex'); + const a = Buffer.from(signature); + const b = Buffer.from(expected); + return a.length === b.length && timingSafeEqual(a, b); + } +} diff --git a/src/livestream/pubsub.service.ts b/src/livestream/pubsub.service.ts new file mode 100644 index 0000000..a624012 --- /dev/null +++ b/src/livestream/pubsub.service.ts @@ -0,0 +1,97 @@ +import { + Injectable, + Logger, + OnModuleDestroy, + OnModuleInit, +} from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; + +import { CHANNELS } from './livestream.channels'; +import { LivestreamService } from './livestream.service'; + +const HUB_URL = 'https://pubsubhubbub.appspot.com/subscribe'; +/** Requested subscription lease (seconds). The hub caps this (~5–10 days). */ +const LEASE_SECONDS = 432000; +/** Re-subscribe before the lease expires. */ +const RENEW_INTERVAL_MS = 4 * 24 * 60 * 60 * 1000; + +const topicUrl = (channelId: string) => + `https://www.youtube.com/xml/feeds/videos.xml?channel_id=${channelId}`; + +/** + * Manages WebSub (PubSubHubbub) subscriptions to each channel's YouTube feed so + * we receive push notifications instead of polling. + */ +@Injectable() +export class PubSubService implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(PubSubService.name); + private renewTimer?: ReturnType; + + constructor( + private readonly config: ConfigService, + private readonly livestream: LivestreamService, + ) {} + + onModuleInit(): void { + const baseUrl = this.config.get('PUBLIC_BASE_URL'); + if (!baseUrl) { + this.logger.warn( + 'PUBLIC_BASE_URL not set — skipping push subscriptions. ' + + 'Status will only update from the periodic reconcile.', + ); + return; + } + void this.subscribeAll(baseUrl); + this.renewTimer = setInterval( + () => void this.subscribeAll(baseUrl), + RENEW_INTERVAL_MS, + ); + } + + onModuleDestroy(): void { + if (this.renewTimer) { + clearInterval(this.renewTimer); + } + } + + private async subscribeAll(baseUrl: string): Promise { + const callback = new URL('/pubsub', baseUrl).toString(); + const secret = this.config.get('PUBSUB_SECRET'); + await Promise.all( + CHANNELS.map(async (channel) => { + try { + const channelId = await this.livestream.resolveChannelId(channel); + await this.subscribe(topicUrl(channelId), callback, secret); + this.logger.log(`Subscribed to ${channel.slug}`); + } catch (err) { + this.logger.error(`Subscribe failed for ${channel.slug}: ${err}`); + } + }), + ); + } + + private async subscribe( + topic: string, + callback: string, + secret?: string, + ): Promise { + const params = new URLSearchParams({ + 'hub.mode': 'subscribe', + 'hub.topic': topic, + 'hub.callback': callback, + 'hub.verify': 'async', + 'hub.lease_seconds': String(LEASE_SECONDS), + }); + if (secret) { + params.set('hub.secret', secret); + } + const res = await fetch(HUB_URL, { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: params, + }); + if (!res.ok) { + throw new Error(`Hub responded ${res.status}`); + } + } +} diff --git a/src/main.ts b/src/main.ts index a09900f..0863e27 100644 --- a/src/main.ts +++ b/src/main.ts @@ -2,7 +2,7 @@ import { NestFactory } from '@nestjs/core'; import { AppModule } from './app.module'; async function bootstrap() { - const app = await NestFactory.create(AppModule); + const app = await NestFactory.create(AppModule, { rawBody: true }); const port = Number(process.env.PORT ?? 3000); await app.listen(port, '0.0.0.0'); } From d12e059ee226328ed77c5f53d79c156fb0678a90 Mon Sep 17 00:00:00 2001 From: Darren Griffin Date: Thu, 9 Jul 2026 15:14:07 +0100 Subject: [PATCH 08/14] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/livestream/pubsub.service.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/src/livestream/pubsub.service.ts b/src/livestream/pubsub.service.ts index a624012..7c94560 100644 --- a/src/livestream/pubsub.service.ts +++ b/src/livestream/pubsub.service.ts @@ -89,6 +89,7 @@ export class PubSubService implements OnModuleInit, OnModuleDestroy { method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, body: params, + signal: AbortSignal.timeout(10_000), }); if (!res.ok) { throw new Error(`Hub responded ${res.status}`); From 3bee330b0c3572c8614ce479347a9c0da46f0d90 Mon Sep 17 00:00:00 2001 From: Darren Griffin Date: Thu, 9 Jul 2026 15:14:12 +0100 Subject: [PATCH 09/14] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/livestream/pubsub.controller.ts | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/src/livestream/pubsub.controller.ts b/src/livestream/pubsub.controller.ts index 43dec57..d89fb80 100644 --- a/src/livestream/pubsub.controller.ts +++ b/src/livestream/pubsub.controller.ts @@ -79,12 +79,18 @@ export class PubSubController { const header = req.headers['x-hub-signature']; const value = Array.isArray(header) ? header[0] : header; const [algo, signature] = value?.split('=') ?? []; - if (!algo || !signature) { + if (algo !== 'sha1' || !signature) { + return false; + } + try { + const expected = createHmac('sha1', secret).update(raw).digest(); + const provided = Buffer.from(signature, 'hex'); + return ( + provided.length === expected.length && + timingSafeEqual(provided, expected) + ); + } catch { return false; } - const expected = createHmac(algo, secret).update(raw).digest('hex'); - const a = Buffer.from(signature); - const b = Buffer.from(expected); - return a.length === b.length && timingSafeEqual(a, b); } } From 9cfc4f4128ce1d4ba6b225981265622db5223e31 Mon Sep 17 00:00:00 2001 From: Darren Date: Thu, 9 Jul 2026 14:24:37 +0000 Subject: [PATCH 10/14] Refactor YouTube feed URL usage to utilize a centralized feedUrl function --- src/livestream/livestream.channels.ts | 9 +++++++++ src/livestream/livestream.service.ts | 9 ++++----- src/livestream/pubsub.service.ts | 7 ++----- 3 files changed, 15 insertions(+), 10 deletions(-) diff --git a/src/livestream/livestream.channels.ts b/src/livestream/livestream.channels.ts index e18eff9..53df9d8 100644 --- a/src/livestream/livestream.channels.ts +++ b/src/livestream/livestream.channels.ts @@ -25,3 +25,12 @@ export const CHANNELS: readonly Channel[] = [ handle: 'musicassistantio', }, ]; + +/** + * Canonical YouTube channel RSS feed URL, used both as the discovery source and + * as the WebSub (PubSubHubbub) subscription topic — keep a single form so the + * two code paths cannot drift. This is the URL documented for the hub topic: + * https://developers.google.com/youtube/v3/guides/push_notifications + */ +export const feedUrl = (channelId: string): string => + `https://www.youtube.com/feeds/videos.xml?channel_id=${channelId}`; diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts index b94b717..6015c00 100644 --- a/src/livestream/livestream.service.ts +++ b/src/livestream/livestream.service.ts @@ -7,7 +7,7 @@ import { } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; -import { Channel, CHANNELS } from './livestream.channels'; +import { Channel, CHANNELS, feedUrl } from './livestream.channels'; export type LivestreamStatus = 'live' | 'upcoming' | 'past' | 'none'; @@ -357,10 +357,9 @@ export class LivestreamService implements OnModuleInit, OnModuleDestroy { /** Fetch recent video IDs from a channel's RSS feed (free, no quota). */ private async fetchFeedVideoIds(channelId: string): Promise { - const res = await fetch( - `https://www.youtube.com/feeds/videos.xml?channel_id=${channelId}`, - { signal: AbortSignal.timeout(10_000) }, - ); + const res = await fetch(feedUrl(channelId), { + signal: AbortSignal.timeout(10_000), + }); if (!res.ok) { throw new Error(`Feed request failed: ${res.status}`); } diff --git a/src/livestream/pubsub.service.ts b/src/livestream/pubsub.service.ts index 7c94560..825a46b 100644 --- a/src/livestream/pubsub.service.ts +++ b/src/livestream/pubsub.service.ts @@ -6,7 +6,7 @@ import { } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; -import { CHANNELS } from './livestream.channels'; +import { CHANNELS, feedUrl } from './livestream.channels'; import { LivestreamService } from './livestream.service'; const HUB_URL = 'https://pubsubhubbub.appspot.com/subscribe'; @@ -15,9 +15,6 @@ const LEASE_SECONDS = 432000; /** Re-subscribe before the lease expires. */ const RENEW_INTERVAL_MS = 4 * 24 * 60 * 60 * 1000; -const topicUrl = (channelId: string) => - `https://www.youtube.com/xml/feeds/videos.xml?channel_id=${channelId}`; - /** * Manages WebSub (PubSubHubbub) subscriptions to each channel's YouTube feed so * we receive push notifications instead of polling. @@ -61,7 +58,7 @@ export class PubSubService implements OnModuleInit, OnModuleDestroy { CHANNELS.map(async (channel) => { try { const channelId = await this.livestream.resolveChannelId(channel); - await this.subscribe(topicUrl(channelId), callback, secret); + await this.subscribe(feedUrl(channelId), callback, secret); this.logger.log(`Subscribed to ${channel.slug}`); } catch (err) { this.logger.error(`Subscribe failed for ${channel.slug}: ${err}`); From 92b2eb2cfe4b0f905e2252c1cb68aec3c7338db8 Mon Sep 17 00:00:00 2001 From: Darren Date: Thu, 9 Jul 2026 15:00:45 +0000 Subject: [PATCH 11/14] Enhance LivestreamService with discovery and reconcile state management; add pruning of expired streams. Update PubSubController to limit processed video IDs and improve signature verification. Refactor PubSubService to handle subscription errors gracefully. --- src/livestream/livestream.service.ts | 55 ++++++++++++++++++++++++---- src/livestream/pubsub.controller.ts | 21 ++++++++--- src/livestream/pubsub.service.ts | 18 ++++++++- 3 files changed, 79 insertions(+), 15 deletions(-) diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts index 6015c00..0ab27b4 100644 --- a/src/livestream/livestream.service.ts +++ b/src/livestream/livestream.service.ts @@ -66,6 +66,8 @@ export class LivestreamService implements OnModuleInit, OnModuleDestroy { private reconcileTimer?: ReturnType; private discoveryTimer?: ReturnType; private tickCount = 0; + private discoveryRunning = false; + private reconcileRunning = false; constructor(private readonly config: ConfigService) {} @@ -159,13 +161,21 @@ export class LivestreamService implements OnModuleInit, OnModuleDestroy { * newly scheduled streams that WebSub may not push. */ private async discovery(): Promise { - await Promise.all( - CHANNELS.map((channel) => - this.discoverChannel(channel).catch((err) => - this.logger.error(`Discovery failed for ${channel.slug}: ${err}`), + if (this.discoveryRunning) { + return; + } + this.discoveryRunning = true; + try { + await Promise.all( + CHANNELS.map((channel) => + this.discoverChannel(channel).catch((err) => + this.logger.error(`Discovery failed for ${channel.slug}: ${err}`), + ), ), - ), - ); + ); + } finally { + this.discoveryRunning = false; + } } private async discoverChannel(channel: Channel): Promise { @@ -186,11 +196,42 @@ export class LivestreamService implements OnModuleInit, OnModuleDestroy { * cost effectively no quota. */ private async tick(): Promise { + // Evict streams whose past-window has elapsed, even when nothing is active. + this.pruneExpired(); this.tickCount = (this.tickCount + 1) % IDLE_TICKS; if (!this.hasActiveStream() && this.tickCount !== 0) { return; } - await this.reconcile(); + if (this.reconcileRunning) { + return; + } + this.reconcileRunning = true; + try { + await this.reconcile(); + } finally { + this.reconcileRunning = false; + } + } + + /** Drop tracked "past" streams older than the window and recompute state. */ + private pruneExpired(): void { + const now = Date.now(); + for (const [slug, videos] of this.tracked) { + let changed = false; + for (const [videoId, v] of videos) { + if ( + v.status === 'past' && + v.actualEndTime && + now - Date.parse(v.actualEndTime) > PAST_WINDOW_MS + ) { + videos.delete(videoId); + changed = true; + } + } + if (changed) { + this.recompute(slug); + } + } } private hasActiveStream(): boolean { diff --git a/src/livestream/pubsub.controller.ts b/src/livestream/pubsub.controller.ts index d89fb80..fa41203 100644 --- a/src/livestream/pubsub.controller.ts +++ b/src/livestream/pubsub.controller.ts @@ -19,6 +19,8 @@ import { LivestreamService } from './livestream.service'; const channelIdPattern = /([^<]+)<\/yt:channelId>/; const videoIdPattern = /([^<]+)<\/yt:videoId>/g; +/** Cap videos processed per push to bound quota/CPU on oversized payloads. */ +const MAX_VIDEO_IDS = 50; /** * WebSub (PubSubHubbub) callback for YouTube push notifications. @@ -51,13 +53,16 @@ export class PubSubController { @Post() @HttpCode(204) notify(@Req() req: RawBodyRequest): void { - const raw = req.rawBody?.toString('utf8') ?? ''; - if (!this.verifySignature(req, raw)) { + const rawBody = req.rawBody; + if (!this.verifySignature(req, rawBody)) { throw new ForbiddenException('Invalid signature'); } - const channelId = channelIdPattern.exec(raw)?.[1]; - const videoIds = [...raw.matchAll(videoIdPattern)].map((m) => m[1]); + const xml = rawBody?.toString('utf8') ?? ''; + const channelId = channelIdPattern.exec(xml)?.[1]; + const videoIds = [...xml.matchAll(videoIdPattern)] + .map((m) => m[1]) + .slice(0, MAX_VIDEO_IDS); if (!channelId || videoIds.length === 0) { return; } @@ -70,12 +75,16 @@ export class PubSubController { private verifySignature( req: RawBodyRequest, - raw: string, + rawBody: Buffer | undefined, ): boolean { const secret = this.config.get('PUBSUB_SECRET'); if (!secret) { return true; // Verification disabled when no secret configured. } + // A secret is configured but we have no raw bytes to verify — fail closed. + if (!rawBody) { + return false; + } const header = req.headers['x-hub-signature']; const value = Array.isArray(header) ? header[0] : header; const [algo, signature] = value?.split('=') ?? []; @@ -83,7 +92,7 @@ export class PubSubController { return false; } try { - const expected = createHmac('sha1', secret).update(raw).digest(); + const expected = createHmac('sha1', secret).update(rawBody).digest(); const provided = Buffer.from(signature, 'hex'); return ( provided.length === expected.length && diff --git a/src/livestream/pubsub.service.ts b/src/livestream/pubsub.service.ts index 825a46b..59d7977 100644 --- a/src/livestream/pubsub.service.ts +++ b/src/livestream/pubsub.service.ts @@ -38,9 +38,14 @@ export class PubSubService implements OnModuleInit, OnModuleDestroy { ); return; } - void this.subscribeAll(baseUrl); + if (!this.config.get('PUBSUB_SECRET')) { + this.logger.warn( + 'PUBSUB_SECRET not set — /pubsub will accept unsigned notifications.', + ); + } + void this.runSubscribeAll(baseUrl); this.renewTimer = setInterval( - () => void this.subscribeAll(baseUrl), + () => void this.runSubscribeAll(baseUrl), RENEW_INTERVAL_MS, ); } @@ -51,6 +56,15 @@ export class PubSubService implements OnModuleInit, OnModuleDestroy { } } + /** subscribeAll wrapper that never rejects (e.g. invalid PUBLIC_BASE_URL). */ + private async runSubscribeAll(baseUrl: string): Promise { + try { + await this.subscribeAll(baseUrl); + } catch (err) { + this.logger.error(`Subscription refresh failed: ${err}`); + } + } + private async subscribeAll(baseUrl: string): Promise { const callback = new URL('/pubsub', baseUrl).toString(); const secret = this.config.get('PUBSUB_SECRET'); From 0284ebb6e319f623c12ff1820ead9b872c40bf78 Mon Sep 17 00:00:00 2001 From: Darren Date: Thu, 9 Jul 2026 15:07:28 +0000 Subject: [PATCH 12/14] Refactor error handling in LivestreamService, PubSubController, and PubSubService to use stackOf function for improved logging of error details. --- src/livestream/livestream.service.ts | 29 +++++++++++++++++++++++++--- src/livestream/pubsub.controller.ts | 5 ++++- src/livestream/pubsub.service.ts | 7 +++++-- 3 files changed, 35 insertions(+), 6 deletions(-) diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts index 0ab27b4..ac750b1 100644 --- a/src/livestream/livestream.service.ts +++ b/src/livestream/livestream.service.ts @@ -51,6 +51,9 @@ const watchUrl = (videoId: string) => const videoIdPattern = /([^<]+)<\/yt:videoId>/g; +const stackOf = (err: unknown): string => + err instanceof Error ? (err.stack ?? err.message) : String(err); + @Injectable() export class LivestreamService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(LivestreamService.name); @@ -63,6 +66,10 @@ export class LivestreamService implements OnModuleInit, OnModuleDestroy { private readonly tracked = new Map>(); private readonly channelIds = new Map(); private readonly slugByChannelId = new Map(); + /** In-flight channel-ID resolutions, to de-duplicate concurrent lookups. */ + private readonly channelIdPromises = new Map>(); + /** Stable timestamp used for channels that have no state yet. */ + private readonly startedAt = new Date().toISOString(); private reconcileTimer?: ReturnType; private discoveryTimer?: ReturnType; private tickCount = 0; @@ -109,6 +116,19 @@ export class LivestreamService implements OnModuleInit, OnModuleDestroy { if (cached) { return cached; } + // De-duplicate concurrent resolutions (e.g. discovery + subscription on + // startup) so they share a single channels.list request. + let pending = this.channelIdPromises.get(channel.slug); + if (!pending) { + pending = this.fetchChannelId(channel).finally(() => + this.channelIdPromises.delete(channel.slug), + ); + this.channelIdPromises.set(channel.slug, pending); + } + return pending; + } + + private async fetchChannelId(channel: Channel): Promise { const data = await this.apiGet('channels', { part: 'id', forHandle: channel.handle, @@ -151,7 +171,7 @@ export class LivestreamService implements OnModuleInit, OnModuleDestroy { channel: channel.slug, channelName: channel.name, status: 'none', - updatedAt: new Date().toISOString(), + updatedAt: this.startedAt, }; } @@ -169,7 +189,10 @@ export class LivestreamService implements OnModuleInit, OnModuleDestroy { await Promise.all( CHANNELS.map((channel) => this.discoverChannel(channel).catch((err) => - this.logger.error(`Discovery failed for ${channel.slug}: ${err}`), + this.logger.error( + `Discovery failed for ${channel.slug}`, + stackOf(err), + ), ), ), ); @@ -280,7 +303,7 @@ export class LivestreamService implements OnModuleInit, OnModuleDestroy { this.recompute(slug); } } catch (err) { - this.logger.warn(`Reconcile failed: ${err}`); + this.logger.warn(`Reconcile failed: ${stackOf(err)}`); } } diff --git a/src/livestream/pubsub.controller.ts b/src/livestream/pubsub.controller.ts index fa41203..43ade71 100644 --- a/src/livestream/pubsub.controller.ts +++ b/src/livestream/pubsub.controller.ts @@ -22,6 +22,9 @@ const videoIdPattern = /([^<]+)<\/yt:videoId>/g; /** Cap videos processed per push to bound quota/CPU on oversized payloads. */ const MAX_VIDEO_IDS = 50; +const stackOf = (err: unknown): string => + err instanceof Error ? (err.stack ?? err.message) : String(err); + /** * WebSub (PubSubHubbub) callback for YouTube push notifications. * https://developers.google.com/youtube/v3/guides/push_notifications @@ -70,7 +73,7 @@ export class PubSubController { // Fire-and-forget: acknowledge the hub quickly. void this.livestream .handleNotification(channelId, videoIds) - .catch((err) => this.logger.error(`Failed to handle push: ${err}`)); + .catch((err) => this.logger.error('Failed to handle push', stackOf(err))); } private verifySignature( diff --git a/src/livestream/pubsub.service.ts b/src/livestream/pubsub.service.ts index 59d7977..97503ea 100644 --- a/src/livestream/pubsub.service.ts +++ b/src/livestream/pubsub.service.ts @@ -15,6 +15,9 @@ const LEASE_SECONDS = 432000; /** Re-subscribe before the lease expires. */ const RENEW_INTERVAL_MS = 4 * 24 * 60 * 60 * 1000; +const stackOf = (err: unknown): string => + err instanceof Error ? (err.stack ?? err.message) : String(err); + /** * Manages WebSub (PubSubHubbub) subscriptions to each channel's YouTube feed so * we receive push notifications instead of polling. @@ -61,7 +64,7 @@ export class PubSubService implements OnModuleInit, OnModuleDestroy { try { await this.subscribeAll(baseUrl); } catch (err) { - this.logger.error(`Subscription refresh failed: ${err}`); + this.logger.error('Subscription refresh failed', stackOf(err)); } } @@ -75,7 +78,7 @@ export class PubSubService implements OnModuleInit, OnModuleDestroy { await this.subscribe(feedUrl(channelId), callback, secret); this.logger.log(`Subscribed to ${channel.slug}`); } catch (err) { - this.logger.error(`Subscribe failed for ${channel.slug}: ${err}`); + this.logger.error(`Subscribe failed for ${channel.slug}`, stackOf(err)); } }), ); From 0b463b5fe5e720c202781bc2cf8ba1ada03ebf2d Mon Sep 17 00:00:00 2001 From: Darren Date: Thu, 9 Jul 2026 15:11:35 +0000 Subject: [PATCH 13/14] Enhance LivestreamService error handling and ensure deterministic state for channels; update PubSubController to enforce PUBSUB_SECRET validation and improve signature verification. Refactor main.ts to ensure proper raw body parsing for YouTube WebSub pushes. --- src/livestream/livestream.service.ts | 11 ++++++++--- src/livestream/pubsub.controller.ts | 14 +++++++++----- src/main.ts | 10 +++++++++- 3 files changed, 26 insertions(+), 9 deletions(-) diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts index ac750b1..effd9f2 100644 --- a/src/livestream/livestream.service.ts +++ b/src/livestream/livestream.service.ts @@ -188,12 +188,17 @@ export class LivestreamService implements OnModuleInit, OnModuleDestroy { try { await Promise.all( CHANNELS.map((channel) => - this.discoverChannel(channel).catch((err) => + this.discoverChannel(channel).catch((err) => { this.logger.error( `Discovery failed for ${channel.slug}`, stackOf(err), - ), - ), + ); + // Ensure the channel still has a deterministic state entry so the + // API doesn't fall back to a fresh defaultInfo on every request. + if (!this.state.has(channel.slug)) { + this.state.set(channel.slug, this.defaultInfo(channel)); + } + }), ), ); } finally { diff --git a/src/livestream/pubsub.controller.ts b/src/livestream/pubsub.controller.ts index 43ade71..2e15381 100644 --- a/src/livestream/pubsub.controller.ts +++ b/src/livestream/pubsub.controller.ts @@ -56,8 +56,15 @@ export class PubSubController { @Post() @HttpCode(204) notify(@Req() req: RawBodyRequest): void { + const secret = this.config.get('PUBSUB_SECRET'); + // Fail closed: without a shared secret we can't authenticate the sender, + // so refuse to do any (quota-costing) work for unauthenticated callers. + if (!secret) { + throw new ForbiddenException('PUBSUB_SECRET not configured'); + } + const rawBody = req.rawBody; - if (!this.verifySignature(req, rawBody)) { + if (!this.verifySignature(secret, req, rawBody)) { throw new ForbiddenException('Invalid signature'); } @@ -77,13 +84,10 @@ export class PubSubController { } private verifySignature( + secret: string, req: RawBodyRequest, rawBody: Buffer | undefined, ): boolean { - const secret = this.config.get('PUBSUB_SECRET'); - if (!secret) { - return true; // Verification disabled when no secret configured. - } // A secret is configured but we have no raw bytes to verify — fail closed. if (!rawBody) { return false; diff --git a/src/main.ts b/src/main.ts index 0863e27..f939c3b 100644 --- a/src/main.ts +++ b/src/main.ts @@ -1,8 +1,16 @@ import { NestFactory } from '@nestjs/core'; +import { NestExpressApplication } from '@nestjs/platform-express'; import { AppModule } from './app.module'; async function bootstrap() { - const app = await NestFactory.create(AppModule, { rawBody: true }); + const app = await NestFactory.create(AppModule, { + rawBody: true, + }); + // YouTube WebSub pushes arrive as Atom XML; register a raw parser for those + // content types so `req.rawBody` is populated for signature verification. + app.useBodyParser('raw', { + type: ['application/atom+xml', 'application/xml', 'text/xml'], + }); const port = Number(process.env.PORT ?? 3000); await app.listen(port, '0.0.0.0'); } From 6219cb1711a50445c95a4537b5cc0f7e844e7e94 Mon Sep 17 00:00:00 2001 From: Darren Date: Thu, 9 Jul 2026 16:07:48 +0000 Subject: [PATCH 14/14] Update environment configuration and enhance PubSub handling; require PUBSUB_SECRET for push notifications, improve error handling in PubSubController, and adjust raw body parsing in main.ts for HMAC verification. --- .env.example | 5 ++++- src/livestream/pubsub.controller.ts | 33 ++++++++++++++--------------- src/livestream/pubsub.service.ts | 5 ++++- src/main.ts | 7 +++--- 4 files changed, 27 insertions(+), 23 deletions(-) diff --git a/.env.example b/.env.example index 9bc9d70..83cbd25 100644 --- a/.env.example +++ b/.env.example @@ -7,5 +7,8 @@ YOUTUBE_API_KEY= # the periodic reconcile. PUBLIC_BASE_URL= -# Optional shared secret for verifying WebSub push payloads (HMAC signature). +# Shared secret for verifying WebSub push payloads (HMAC signature). Required +# for push notifications: without it the service skips subscriptions and the +# /pubsub webhook rejects all incoming notifications (403). Generate one with +# `openssl rand -hex 32`. PUBSUB_SECRET= diff --git a/src/livestream/pubsub.controller.ts b/src/livestream/pubsub.controller.ts index 2e15381..4dbb243 100644 --- a/src/livestream/pubsub.controller.ts +++ b/src/livestream/pubsub.controller.ts @@ -1,17 +1,17 @@ import { createHmac, timingSafeEqual } from 'node:crypto'; -import { IncomingMessage } from 'node:http'; import { + BadRequestException, + Body, Controller, ForbiddenException, Get, Header, + Headers, HttpCode, Logger, Post, Query, - RawBodyRequest, - Req, } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; @@ -44,7 +44,7 @@ export class PubSubController { verify(@Query() query: Record): string { const challenge = query['hub.challenge']; if (!challenge) { - throw new ForbiddenException('Missing hub.challenge'); + throw new BadRequestException('Missing hub.challenge'); } this.logger.log( `Verified ${query['hub.mode']} for ${query['hub.topic']}`, @@ -55,20 +55,21 @@ export class PubSubController { /** Content-distribution notification with an Atom feed payload. */ @Post() @HttpCode(204) - notify(@Req() req: RawBodyRequest): void { + notify( + @Body() body: Buffer, + @Headers('x-hub-signature') signature?: string, + ): void { const secret = this.config.get('PUBSUB_SECRET'); // Fail closed: without a shared secret we can't authenticate the sender, // so refuse to do any (quota-costing) work for unauthenticated callers. if (!secret) { throw new ForbiddenException('PUBSUB_SECRET not configured'); } - - const rawBody = req.rawBody; - if (!this.verifySignature(secret, req, rawBody)) { + if (!this.verifySignature(secret, signature, body)) { throw new ForbiddenException('Invalid signature'); } - const xml = rawBody?.toString('utf8') ?? ''; + const xml = Buffer.isBuffer(body) ? body.toString('utf8') : ''; const channelId = channelIdPattern.exec(xml)?.[1]; const videoIds = [...xml.matchAll(videoIdPattern)] .map((m) => m[1]) @@ -85,21 +86,19 @@ export class PubSubController { private verifySignature( secret: string, - req: RawBodyRequest, - rawBody: Buffer | undefined, + signatureHeader: string | undefined, + body: Buffer, ): boolean { - // A secret is configured but we have no raw bytes to verify — fail closed. - if (!rawBody) { + // No raw bytes to verify (wrong/missing parser) — fail closed. + if (!Buffer.isBuffer(body)) { return false; } - const header = req.headers['x-hub-signature']; - const value = Array.isArray(header) ? header[0] : header; - const [algo, signature] = value?.split('=') ?? []; + const [algo, signature] = signatureHeader?.split('=') ?? []; if (algo !== 'sha1' || !signature) { return false; } try { - const expected = createHmac('sha1', secret).update(rawBody).digest(); + const expected = createHmac('sha1', secret).update(body).digest(); const provided = Buffer.from(signature, 'hex'); return ( provided.length === expected.length && diff --git a/src/livestream/pubsub.service.ts b/src/livestream/pubsub.service.ts index 97503ea..5a2297a 100644 --- a/src/livestream/pubsub.service.ts +++ b/src/livestream/pubsub.service.ts @@ -43,8 +43,11 @@ export class PubSubService implements OnModuleInit, OnModuleDestroy { } if (!this.config.get('PUBSUB_SECRET')) { this.logger.warn( - 'PUBSUB_SECRET not set — /pubsub will accept unsigned notifications.', + 'PUBSUB_SECRET not set — skipping push subscriptions ' + + '(the /pubsub webhook rejects unsigned pushes). ' + + 'Status will only update from the periodic reconcile.', ); + return; } void this.runSubscribeAll(baseUrl); this.renewTimer = setInterval( diff --git a/src/main.ts b/src/main.ts index f939c3b..2d1c16f 100644 --- a/src/main.ts +++ b/src/main.ts @@ -3,11 +3,10 @@ import { NestExpressApplication } from '@nestjs/platform-express'; import { AppModule } from './app.module'; async function bootstrap() { - const app = await NestFactory.create(AppModule, { - rawBody: true, - }); + const app = await NestFactory.create(AppModule); // YouTube WebSub pushes arrive as Atom XML; register a raw parser for those - // content types so `req.rawBody` is populated for signature verification. + // content types so the /pubsub handler receives the exact bytes (needed for + // HMAC signature verification). app.useBodyParser('raw', { type: ['application/atom+xml', 'application/xml', 'text/xml'], });