mirror of
https://github.com/yusufipk/YTChatHub.git
synced 2026-09-12 03:16:10 +00:00
refactor: migrate from pnpm workspaces to single-package structure with shared types
This commit is contained in:
Yusuf İpek
@@ -0,0 +1,187 @@
|
||||
import Fastify from 'fastify';
|
||||
import cors from '@fastify/cors';
|
||||
import EventEmitter from 'eventemitter3';
|
||||
import type { ChatMessage } from '@shared/chat';
|
||||
import { bootstrapInnertube, type IngestionContext } from './ingestion/youtubei';
|
||||
|
||||
const MAX_MESSAGES = 500;
|
||||
|
||||
export async function startBackend() {
|
||||
const fastify = Fastify({
|
||||
logger: true
|
||||
});
|
||||
|
||||
await fastify.register(cors, { origin: true });
|
||||
|
||||
const store: ChatMessage[] = [];
|
||||
let currentSelection: ChatMessage | null = null;
|
||||
const overlayEmitter = new EventEmitter<{ update: (message: ChatMessage | null) => void }>();
|
||||
|
||||
const rawLiveId = process.env.YOUTUBE_LIVE_ID ?? '';
|
||||
const parsedLiveId = extractLiveId(rawLiveId);
|
||||
const shouldMock = !parsedLiveId;
|
||||
|
||||
let ingestion: IngestionContext | null = null;
|
||||
|
||||
if (!shouldMock) {
|
||||
try {
|
||||
console.log(`[Backend] Connecting to YouTube Live ID: ${parsedLiveId}`);
|
||||
ingestion = await bootstrapInnertube(parsedLiveId);
|
||||
console.log(`[Backend] ✓ YouTube chat connected successfully`);
|
||||
ingestion.emitter.on('message', (message) => {
|
||||
console.log(`[Chat] ${message.author}: ${message.text}`);
|
||||
store.push(message);
|
||||
if (store.length > MAX_MESSAGES) {
|
||||
store.splice(0, store.length - MAX_MESSAGES);
|
||||
}
|
||||
});
|
||||
ingestion.emitter.on('error', (error) => {
|
||||
console.error('[Backend] Innertube ingestion error:', error);
|
||||
});
|
||||
} catch (error) {
|
||||
console.error('[Backend] Failed to bootstrap Innertube; falling back to mock data:', error);
|
||||
}
|
||||
} else {
|
||||
console.log('[Backend] No YOUTUBE_LIVE_ID found, running in mock mode');
|
||||
}
|
||||
|
||||
if (!ingestion) {
|
||||
seedMockMessages(store, overlayEmitter);
|
||||
}
|
||||
|
||||
fastify.get('/health', async () => ({
|
||||
status: 'ok',
|
||||
messages: store.length,
|
||||
selection: currentSelection?.id ?? null,
|
||||
mode: ingestion ? 'live' : 'mock'
|
||||
}));
|
||||
|
||||
fastify.get('/chat/messages', async () => ({
|
||||
messages: store
|
||||
}));
|
||||
|
||||
fastify.post<{ Body: { id?: string } }>('/overlay/selection', async (request, reply) => {
|
||||
const { id } = request.body ?? {};
|
||||
if (!id) {
|
||||
reply.status(400);
|
||||
return { error: 'id is required' };
|
||||
}
|
||||
|
||||
const message = store.find((item) => item.id === id);
|
||||
if (!message) {
|
||||
reply.status(404);
|
||||
return { error: 'message not found' };
|
||||
}
|
||||
|
||||
currentSelection = message;
|
||||
overlayEmitter.emit('update', currentSelection);
|
||||
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
fastify.delete('/overlay/selection', async () => {
|
||||
currentSelection = null;
|
||||
overlayEmitter.emit('update', null);
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
fastify.options('/overlay/stream', async (request, reply) => {
|
||||
reply.headers({
|
||||
'Access-Control-Allow-Origin': '*',
|
||||
'Access-Control-Allow-Methods': 'GET, OPTIONS',
|
||||
'Access-Control-Allow-Headers': 'Content-Type'
|
||||
});
|
||||
reply.status(204).send();
|
||||
});
|
||||
|
||||
fastify.get('/overlay/stream', async (request, reply) => {
|
||||
reply.hijack();
|
||||
|
||||
const res = reply.raw;
|
||||
res.setHeader('Content-Type', 'text/event-stream');
|
||||
res.setHeader('Cache-Control', 'no-cache');
|
||||
res.setHeader('Connection', 'keep-alive');
|
||||
res.setHeader('Access-Control-Allow-Origin', '*');
|
||||
res.setHeader('Access-Control-Allow-Methods', 'GET, OPTIONS');
|
||||
res.setHeader('Access-Control-Allow-Headers', 'Content-Type');
|
||||
res.writeHead(200);
|
||||
res.write(': connected\n\n');
|
||||
|
||||
const send = (message: ChatMessage | null) => {
|
||||
res.write(`event: selection\ndata: ${JSON.stringify({ message })}\n\n`);
|
||||
};
|
||||
|
||||
const heartbeat = setInterval(() => {
|
||||
res.write('event: heartbeat\ndata: {}\n\n');
|
||||
}, 15000);
|
||||
|
||||
overlayEmitter.on('update', send);
|
||||
|
||||
if (currentSelection) {
|
||||
send(currentSelection);
|
||||
}
|
||||
|
||||
request.raw.on('close', () => {
|
||||
clearInterval(heartbeat);
|
||||
overlayEmitter.off('update', send);
|
||||
});
|
||||
});
|
||||
|
||||
const port = Number(process.env.PORT ?? 4100);
|
||||
|
||||
await fastify.listen({ port, host: '0.0.0.0' });
|
||||
|
||||
console.log(`[Backend] Server listening on http://localhost:${port}`);
|
||||
}
|
||||
|
||||
function extractLiveId(input: string): string {
|
||||
if (!input) return '';
|
||||
|
||||
const trimmed = input.trim();
|
||||
if (/^[a-zA-Z0-9_-]{10,}$/.test(trimmed)) {
|
||||
return trimmed;
|
||||
}
|
||||
|
||||
try {
|
||||
const url = new URL(trimmed);
|
||||
if (url.searchParams.has('v')) {
|
||||
return url.searchParams.get('v') ?? '';
|
||||
}
|
||||
const pathname = url.pathname.split('/').filter(Boolean).pop();
|
||||
return pathname ?? '';
|
||||
} catch (error) {
|
||||
console.warn('Invalid YOUTUBE_LIVE_ID provided', error);
|
||||
return '';
|
||||
}
|
||||
}
|
||||
|
||||
function seedMockMessages(
|
||||
store: ChatMessage[],
|
||||
overlayEmitter: EventEmitter<{ update: (message: ChatMessage | null) => void }>
|
||||
) {
|
||||
let counter = 0;
|
||||
const authors = ['Ada', 'Linus', 'Grace', 'Marge'];
|
||||
setInterval(() => {
|
||||
const message: ChatMessage = {
|
||||
id: `mock-${Date.now()}`,
|
||||
author: authors[counter % authors.length],
|
||||
text: `Mock message #${counter}`,
|
||||
publishedAt: new Date().toISOString()
|
||||
};
|
||||
store.push(message);
|
||||
if (store.length > MAX_MESSAGES) {
|
||||
store.splice(0, store.length - MAX_MESSAGES);
|
||||
}
|
||||
if (counter % 5 === 0) {
|
||||
overlayEmitter.emit('update', message);
|
||||
}
|
||||
counter += 1;
|
||||
}, 2000);
|
||||
}
|
||||
|
||||
if (import.meta.url === `file://${process.argv[1]}`) {
|
||||
startBackend().catch((error) => {
|
||||
console.error('Failed to start backend', error);
|
||||
process.exit(1);
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
import type { ChatMessage } from '@shared/chat';
|
||||
import EventEmitter from 'eventemitter3';
|
||||
import Innertube, { UniversalCache } from 'youtubei.js';
|
||||
|
||||
export type IngestionContext = {
|
||||
client: Innertube;
|
||||
liveChat: any;
|
||||
videoId: string;
|
||||
emitter: ChatEventEmitter;
|
||||
};
|
||||
|
||||
export type ContinuationState = {
|
||||
messages: ChatMessage[];
|
||||
nextToken: string | null;
|
||||
timeoutMs: number;
|
||||
};
|
||||
|
||||
export type ChatEventEmitter = EventEmitter<{
|
||||
message: (message: ChatMessage) => void;
|
||||
error: (error: unknown) => void;
|
||||
}>;
|
||||
|
||||
const defaultTimeout = 1500;
|
||||
|
||||
export async function bootstrapInnertube(videoId: string): Promise<IngestionContext> {
|
||||
if (!videoId) {
|
||||
throw new Error('YOUTUBE_LIVE_ID is required to bootstrap Innertube');
|
||||
}
|
||||
|
||||
console.log('[Ingestion] Creating Innertube client...');
|
||||
const client = await Innertube.create({
|
||||
cache: new UniversalCache(false)
|
||||
});
|
||||
|
||||
console.log('[Ingestion] Fetching video info...');
|
||||
const info = await client.getInfo(videoId);
|
||||
|
||||
console.log('[Ingestion] Getting live chat...');
|
||||
const liveChat = info.getLiveChat();
|
||||
|
||||
if (!liveChat) {
|
||||
throw new Error('This video does not have an active live chat');
|
||||
}
|
||||
|
||||
const emitter: ChatEventEmitter = new EventEmitter();
|
||||
|
||||
liveChat.on('chat-update', (action: any) => {
|
||||
const normalized = normalizeAction(action);
|
||||
if (normalized) {
|
||||
emitter.emit('message', normalized);
|
||||
}
|
||||
});
|
||||
|
||||
liveChat.on('error', (err: unknown) => {
|
||||
console.error('[Ingestion] Live chat error:', err);
|
||||
emitter.emit('error', err);
|
||||
});
|
||||
|
||||
console.log('[Ingestion] Starting live chat listener...');
|
||||
liveChat.start();
|
||||
console.log('[Ingestion] Live chat listener started');
|
||||
|
||||
return {
|
||||
client,
|
||||
liveChat,
|
||||
videoId,
|
||||
emitter
|
||||
};
|
||||
}
|
||||
|
||||
export async function fetchChatBatch(
|
||||
ctx: IngestionContext,
|
||||
options?: { windowMs?: number }
|
||||
): Promise<ContinuationState> {
|
||||
const windowMs = options?.windowMs ?? defaultTimeout;
|
||||
const collected: ChatMessage[] = [];
|
||||
|
||||
const listener = (message: ChatMessage) => {
|
||||
collected.push(message);
|
||||
};
|
||||
|
||||
ctx.emitter.on('message', listener);
|
||||
|
||||
await delay(windowMs);
|
||||
|
||||
ctx.emitter.off('message', listener);
|
||||
|
||||
return {
|
||||
messages: collected,
|
||||
nextToken: ctx.liveChat?.continuation?.token ?? null,
|
||||
timeoutMs: ctx.liveChat?.continuation?.timeout_ms ?? defaultTimeout
|
||||
};
|
||||
}
|
||||
|
||||
function resolveMessageText(item: any): string {
|
||||
if (!item?.message) return '';
|
||||
|
||||
if (typeof item.message === 'string') {
|
||||
return item.message;
|
||||
}
|
||||
|
||||
if (typeof item.message?.toString === 'function') {
|
||||
return item.message.toString();
|
||||
}
|
||||
|
||||
if (Array.isArray(item.message?.runs)) {
|
||||
return item.message.runs.map((run: any) => run.text ?? '').join('');
|
||||
}
|
||||
|
||||
return '';
|
||||
}
|
||||
|
||||
function resolveTimestamp(timestamp: number | string | undefined): string {
|
||||
if (!timestamp) {
|
||||
return new Date().toISOString();
|
||||
}
|
||||
|
||||
const numeric = typeof timestamp === 'string' ? Number(timestamp) : timestamp;
|
||||
if (Number.isFinite(numeric)) {
|
||||
const millis = numeric > 1e12 ? numeric / 1000 : numeric;
|
||||
return new Date(millis).toISOString();
|
||||
}
|
||||
|
||||
return new Date().toISOString();
|
||||
}
|
||||
|
||||
function normalizeAction(action: any): ChatMessage | null {
|
||||
if (!action || action.type !== 'AddChatItemAction') {
|
||||
return null;
|
||||
}
|
||||
|
||||
const item = action.item;
|
||||
if (!item) return null;
|
||||
|
||||
if (
|
||||
item.type === 'LiveChatTextMessage' ||
|
||||
item.type === 'LiveChatPaidMessage' ||
|
||||
item.type === 'LiveChatMembershipItem'
|
||||
) {
|
||||
return {
|
||||
id: String(item.id ?? item.timestamp_usec ?? Date.now()),
|
||||
author: String(item.author?.name ?? 'Unknown'),
|
||||
text: resolveMessageText(item),
|
||||
publishedAt: resolveTimestamp(item.timestamp ?? item.timestamp_usec)
|
||||
};
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
function delay(ms: number) {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
Reference in New Issue
Block a user