From d29c3a87b48b48edba73ddc3eaec7a89ffefe32f Mon Sep 17 00:00:00 2001 From: neutrino2211 Date: Sat, 4 Apr 2026 18:01:24 +0100 Subject: [PATCH] feat(vod): add background jobs for video sync and processing Two scheduled jobs: - syncVideos (hourly): Fetches videos from streamplace, creates cache folders in /tmp/vod-cache/{hash}/, stores master.m3u8 and video.json - processVideos (every 30 min): Scans cache folders, generates thumbnail.jpg, sprites.jpg, and sprites.vtt for unprocessed videos Uses marker file pattern (sprites.jpg existence) for state tracking. Processes max 5 videos per run to avoid timeouts. Co-Authored-By: Claude Opus 4.5 --- apps/vod/src/jobs.ts | 337 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 337 insertions(+) create mode 100644 apps/vod/src/jobs.ts diff --git a/apps/vod/src/jobs.ts b/apps/vod/src/jobs.ts new file mode 100644 index 0000000..fcbf7ec --- /dev/null +++ b/apps/vod/src/jobs.ts @@ -0,0 +1,337 @@ +/** + * VOD Background Jobs + * + * Two scheduled jobs for video processing: + * 1. syncVideos - Fetches videos from streamplace, creates cache folders with m3u8 files (hourly) + * 2. processVideos - Generates thumbnails and VTTs for unprocessed videos (every 30 min) + */ + +import { manifest, job } from "@at-run/runtime" +import { AtpAgent } from "@atproto/api" +import { Parser } from "m3u8-parser" +import { generateThumbnail } from "./thumbnail" +import { generateSpriteWithVtt } from "./vtt" + +// Re-export for bundle +export { v } from "@at-run/runtime" + +declare const Deno: { + env: { get(key: string): string | undefined } + readTextFile(path: string): Promise + writeTextFile(path: string, content: string): Promise + readDir(path: string): AsyncIterable<{ name: string; isDirectory: boolean }> + mkdir(path: string, options?: { recursive?: boolean }): Promise + stat(path: string): Promise<{ isFile: boolean; isDirectory: boolean }> +} + +const STREAM_PLACE_DID = "did:plc:rbvrr34edl5ddpuwcubjiost" +const STREAM_PLACE_PDS = "https://iameli.com" +const VOD_XRPC_BASE = "https://vod-beta.stream.place/xrpc/" +const CACHE_DIR = "/tmp/vod-cache" + +// Max videos to process per job run (avoid timeout) +const MAX_PROCESS_PER_RUN = 5 + +// ============================================================================= +// Bundle Manifest +// ============================================================================= + +export const bundle = manifest({ + name: "vod-jobs", + description: "VOD background jobs - sync videos and generate thumbnails/VTTs", + permissions: { + net: [ + "vod-beta.stream.place", + "iameli.com", + "plc.directory", + ], + read: [CACHE_DIR], + write: [CACHE_DIR], + run: ["ffmpeg", "ffprobe"], + }, +}) + +// ============================================================================= +// Utilities +// ============================================================================= + +const streamPlaceAgent = new AtpAgent({ service: STREAM_PLACE_PDS }) + +async function sha256(data: string): Promise { + const bytes = new TextEncoder().encode(data) + const hash = await crypto.subtle.digest("SHA-256", bytes.buffer as ArrayBuffer) + return [...new Uint8Array(hash)].map(b => b.toString(16).padStart(2, "0")).join("") +} + +async function exists(path: string): Promise { + try { + await Deno.stat(path) + return true + } catch { + return false + } +} + +function rewritePlaylistUrls(content: string, baseUrl: string): string { + return content + .split("\n") + .map((line) => { + const trimmed = line.trim() + if (!trimmed || trimmed.startsWith("#")) { + if (trimmed.includes('URI="') && !trimmed.includes('URI="http')) { + return trimmed.replace(/URI="([^"]+)"/, `URI="${baseUrl}$1"`) + } + return line + } + if (!trimmed.startsWith("http")) { + return baseUrl + trimmed + } + return line + }) + .join("\n") +} + +interface VideoRecord { + uri: string + title: string + creator: string + duration: number + createdAt: string +} + +interface CachedVideo { + uri: string + title: string + creator: string + duration: number + createdAt: string + cachedAt: string + hash: string +} + +// ============================================================================= +// Job 1: Sync Videos (hourly) +// ============================================================================= + +export const syncVideos = job({ + description: "Fetch videos from streamplace and cache m3u8 playlists", + schedule: { + type: "cron", + cron: "0 * * * *", // Every hour + }, + permissions: { + net: [ + "vod-beta.stream.place", + "iameli.com", + "plc.directory", + ], + read: [CACHE_DIR], + write: [CACHE_DIR], + }, + limits: { + timeoutMs: 300_000, // 5 minutes + }, + async handler() { + console.log("[syncVideos] Starting video sync...") + + // Ensure cache directory exists + await Deno.mkdir(CACHE_DIR, { recursive: true }) + + // Fetch all videos from streamplace + let cursor: string | undefined + const allVideos: VideoRecord[] = [] + + do { + const response = await streamPlaceAgent.com.atproto.repo.listRecords({ + repo: STREAM_PLACE_DID, + collection: "place.stream.video", + limit: 100, + cursor, + }) + + for (const record of response.data.records) { + const value = record.value as { + title: string + creator: string + duration: number + createdAt: string + } + allVideos.push({ + uri: record.uri, + title: value.title, + creator: value.creator, + duration: value.duration, + createdAt: value.createdAt, + }) + } + + cursor = response.data.cursor + } while (cursor) + + console.log(`[syncVideos] Found ${allVideos.length} videos`) + + let synced = 0 + let skipped = 0 + + for (const video of allVideos) { + const hash = await sha256(video.uri) + const basePath = `${CACHE_DIR}/${hash}` + const metaPath = `${basePath}/video.json` + const playlistPath = `${basePath}/master.m3u8` + + // Skip if already cached + if (await exists(metaPath)) { + skipped++ + continue + } + + // Create directory + await Deno.mkdir(basePath, { recursive: true }) + + // Fetch and cache the master playlist + const playlistUrl = `${VOD_XRPC_BASE}place.stream.playback.getVideoPlaylist?uri=${encodeURIComponent(video.uri)}` + + try { + const res = await fetch(playlistUrl) + if (!res.ok) { + console.error(`[syncVideos] Failed to fetch playlist for ${video.title}: ${res.status}`) + continue + } + + const rawPlaylist = await res.text() + const rewrittenPlaylist = rewritePlaylistUrls(rawPlaylist, VOD_XRPC_BASE) + await Deno.writeTextFile(playlistPath, rewrittenPlaylist) + + // Save video metadata + const cachedVideo: CachedVideo = { + ...video, + cachedAt: new Date().toISOString(), + hash, + } + await Deno.writeTextFile(metaPath, JSON.stringify(cachedVideo, null, 2)) + + synced++ + console.log(`[syncVideos] Cached: ${video.title}`) + } catch (err) { + console.error(`[syncVideos] Error caching ${video.title}:`, err) + } + } + + console.log(`[syncVideos] Complete: ${synced} synced, ${skipped} already cached`) + + return { + success: true, + total: allVideos.length, + synced, + skipped, + } + }, +}) + +// ============================================================================= +// Job 2: Process Videos (every 30 minutes) +// ============================================================================= + +export const processVideos = job({ + description: "Generate thumbnails and VTTs for unprocessed videos", + schedule: { + type: "cron", + cron: "*/30 * * * *", // Every 30 minutes + }, + permissions: { + read: [CACHE_DIR], + write: [CACHE_DIR], + run: ["ffmpeg", "ffprobe"], + }, + limits: { + timeoutMs: 600_000, // 10 minutes + }, + async handler() { + console.log("[processVideos] Starting video processing...") + + // Check if cache directory exists + if (!(await exists(CACHE_DIR))) { + console.log("[processVideos] Cache directory doesn't exist, nothing to process") + return { success: true, processed: 0, skipped: 0, errors: 0 } + } + + const toProcess: Array<{ path: string; video: CachedVideo }> = [] + + // Scan cache directory for unprocessed videos + for await (const entry of Deno.readDir(CACHE_DIR)) { + if (!entry.isDirectory) continue + + const basePath = `${CACHE_DIR}/${entry.name}` + const metaPath = `${basePath}/video.json` + const spritePath = `${basePath}/sprites.jpg` + + // Skip if no metadata (not a valid video folder) + if (!(await exists(metaPath))) continue + + // Skip if already processed (sprite exists) + if (await exists(spritePath)) continue + + // Read video metadata + try { + const metaContent = await Deno.readTextFile(metaPath) + const video = JSON.parse(metaContent) as CachedVideo + toProcess.push({ path: basePath, video }) + } catch (err) { + console.error(`[processVideos] Error reading metadata for ${entry.name}:`, err) + } + } + + console.log(`[processVideos] Found ${toProcess.length} videos to process`) + + // Limit processing per run to avoid timeout + const batch = toProcess.slice(0, MAX_PROCESS_PER_RUN) + + let processed = 0 + let errors = 0 + + for (const { path, video } of batch) { + const playlistPath = `${path}/master.m3u8` + const thumbnailPath = `${path}/thumbnail.jpg` + + console.log(`[processVideos] Processing: ${video.title}`) + + try { + // Generate thumbnail (3 seconds in) + console.log(`[processVideos] Generating thumbnail...`) + await generateThumbnail(playlistPath, thumbnailPath, "00:00:03") + + // Generate sprite sheet and VTT + console.log(`[processVideos] Generating sprites and VTT...`) + await generateSpriteWithVtt(playlistPath, path) + + processed++ + console.log(`[processVideos] Done: ${video.title}`) + } catch (err) { + errors++ + console.error(`[processVideos] Error processing ${video.title}:`, err) + + // Write error marker so we can debug later + try { + await Deno.writeTextFile( + `${path}/error.txt`, + `${new Date().toISOString()}\n${err instanceof Error ? err.message : String(err)}` + ) + } catch { + // Ignore error writing error file + } + } + } + + const skipped = toProcess.length - batch.length + + console.log(`[processVideos] Complete: ${processed} processed, ${skipped} queued for next run, ${errors} errors`) + + return { + success: true, + processed, + skipped, + errors, + remaining: toProcess.length - batch.length, + } + }, +}) -- 2.54.0