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..83cbd25 --- /dev/null +++ b/.env.example @@ -0,0 +1,14 @@ +# 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= + +# 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/.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..dadc7b9 --- /dev/null +++ b/src/livestream/index.ts @@ -0,0 +1,4 @@ +export * from './livestream.channels'; +export * from './livestream.module'; +export * from './livestream.service'; +export * from './pubsub.service'; diff --git a/src/livestream/livestream.channels.ts b/src/livestream/livestream.channels.ts new file mode 100644 index 0000000..53df9d8 --- /dev/null +++ b/src/livestream/livestream.channels.ts @@ -0,0 +1,36 @@ +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', + }, +]; + +/** + * 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.controller.ts b/src/livestream/livestream.controller.ts new file mode 100644 index 0000000..bd7524f --- /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(): LivestreamInfo[] { + return this.livestream.getAll(); + } + + @Get(':slug') + 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 new file mode 100644 index 0000000..7e18081 --- /dev/null +++ b/src/livestream/livestream.module.ts @@ -0,0 +1,12 @@ +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, PubSubController], + providers: [LivestreamService, PubSubService], +}) +export class LivestreamModule {} diff --git a/src/livestream/livestream.service.ts b/src/livestream/livestream.service.ts new file mode 100644 index 0000000..effd9f2 --- /dev/null +++ b/src/livestream/livestream.service.ts @@ -0,0 +1,478 @@ +import { + Injectable, + Logger, + NotFoundException, + OnModuleDestroy, + OnModuleInit, +} from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; + +import { Channel, CHANNELS, feedUrl } 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 this channel's state was last updated. */ + updatedAt: string; +} + +/** A livestream video we are tracking the state of. */ +interface TrackedVideo { + videoId: string; + title: string; + status: LivestreamStatus; + scheduledStartTime?: string; + actualEndTime?: string; + updatedAt: number; +} + +const API_BASE = 'https://www.googleapis.com/youtube/v3'; +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; + +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); + private readonly channelsBySlug = new Map( + CHANNELS.map((c) => [c.slug, c] as const), + ); + /** 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 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; + private discoveryRunning = false; + private reconcileRunning = false; + + constructor(private readonly config: ConfigService) {} + + 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); + } + + onModuleDestroy(): void { + if (this.discoveryTimer) { + clearInterval(this.discoveryTimer); + } + if (this.reconcileTimer) { + clearInterval(this.reconcileTimer); + } + } + + getAll(): LivestreamInfo[] { + return CHANNELS.map((channel) => this.readState(channel)); + } + + getStatus(slug: string): LivestreamInfo { + const channel = this.channelsBySlug.get(slug); + if (!channel) { + throw new NotFoundException(`Unknown channel "${slug}"`); + } + return this.readState(channel); + } + + /** 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; + } + // 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, + }); + 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; + } + + /** + * 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); + } + + private readState(channel: Channel): LivestreamInfo { + return this.state.get(channel.slug) ?? this.defaultInfo(channel); + } + + private defaultInfo(channel: Channel): LivestreamInfo { + return { + channel: channel.slug, + channelName: channel.name, + status: 'none', + updatedAt: this.startedAt, + }; + } + + /** + * 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 { + 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}`, + 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 { + this.discoveryRunning = false; + } + } + + 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); + } + + /** + * 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 { + // 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; + } + 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 { + 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 { + 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); + } + } + for (const slug of touched) { + this.recompute(slug); + } + } catch (err) { + this.logger.warn(`Reconcile failed: ${stackOf(err)}`); + } + } + + /** 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, + updatedAt: new Date().toISOString(), + }; + const videos = [...(this.tracked.get(slug)?.values() ?? [])]; + const now = Date.now(); + + const live = videos.find((v) => v.status === 'live'); + if (live) { + this.state.set(slug, { + ...base, + status: 'live', + title: live.title, + url: watchUrl(live.videoId), + }); + return; + } + + 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 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; + } + + this.state.set(slug, { ...base, status: 'none' }); + } + + /** Fetch recent video IDs from a channel's RSS feed (free, no quota). */ + private async fetchFeedVideoIds(channelId: string): Promise { + const res = await fetch(feedUrl(channelId), { + signal: AbortSignal.timeout(10_000), + }); + if (!res.ok) { + throw new Error(`Feed request failed: ${res.status}`); + } + const xml = await res.text(); + return [...xml.matchAll(videoIdPattern)].map((m) => m[1]); + } + + private async videoDetails(videoIds: string[]): Promise { + const unique = [...new Set(videoIds)]; + if (unique.length === 0) { + return []; + } + 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( + 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, { signal: AbortSignal.timeout(10_000) }); + if (!res.ok) { + const body = await res.text().catch(() => ''); + throw new Error( + `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..4dbb243 --- /dev/null +++ b/src/livestream/pubsub.controller.ts @@ -0,0 +1,111 @@ +import { createHmac, timingSafeEqual } from 'node:crypto'; + +import { + BadRequestException, + Body, + Controller, + ForbiddenException, + Get, + Header, + Headers, + HttpCode, + Logger, + Post, + Query, +} from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; + +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; + +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 + */ +@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 BadRequestException('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( + @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'); + } + if (!this.verifySignature(secret, signature, body)) { + throw new ForbiddenException('Invalid signature'); + } + + const xml = Buffer.isBuffer(body) ? body.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; + } + + // Fire-and-forget: acknowledge the hub quickly. + void this.livestream + .handleNotification(channelId, videoIds) + .catch((err) => this.logger.error('Failed to handle push', stackOf(err))); + } + + private verifySignature( + secret: string, + signatureHeader: string | undefined, + body: Buffer, + ): boolean { + // No raw bytes to verify (wrong/missing parser) — fail closed. + if (!Buffer.isBuffer(body)) { + return false; + } + const [algo, signature] = signatureHeader?.split('=') ?? []; + if (algo !== 'sha1' || !signature) { + return false; + } + try { + const expected = createHmac('sha1', secret).update(body).digest(); + const provided = Buffer.from(signature, 'hex'); + return ( + provided.length === expected.length && + timingSafeEqual(provided, expected) + ); + } catch { + return false; + } + } +} diff --git a/src/livestream/pubsub.service.ts b/src/livestream/pubsub.service.ts new file mode 100644 index 0000000..5a2297a --- /dev/null +++ b/src/livestream/pubsub.service.ts @@ -0,0 +1,115 @@ +import { + Injectable, + Logger, + OnModuleDestroy, + OnModuleInit, +} from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; + +import { CHANNELS, feedUrl } 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 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. + */ +@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; + } + if (!this.config.get('PUBSUB_SECRET')) { + this.logger.warn( + '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( + () => void this.runSubscribeAll(baseUrl), + RENEW_INTERVAL_MS, + ); + } + + onModuleDestroy(): void { + if (this.renewTimer) { + clearInterval(this.renewTimer); + } + } + + /** 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', stackOf(err)); + } + } + + 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(feedUrl(channelId), callback, secret); + this.logger.log(`Subscribed to ${channel.slug}`); + } catch (err) { + this.logger.error(`Subscribe failed for ${channel.slug}`, stackOf(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, + signal: AbortSignal.timeout(10_000), + }); + if (!res.ok) { + throw new Error(`Hub responded ${res.status}`); + } + } +} diff --git a/src/main.ts b/src/main.ts index a09900f..2d1c16f 100644 --- a/src/main.ts +++ b/src/main.ts @@ -1,8 +1,15 @@ 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); + const app = await NestFactory.create(AppModule); + // YouTube WebSub pushes arrive as Atom XML; register a raw parser for those + // 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'], + }); const port = Number(process.env.PORT ?? 3000); await app.listen(port, '0.0.0.0'); }