feat(media): HLS adaptive bitrate streaming with MP4 fallback
Replaces single-MP4 + range-request streaming with HLS multi-bitrate
segments to fix video stutter through the Newt tunnel. Range-request
bursts were the root cause; HLS chunks are small and tunnel-friendly,
plus the player adapts bitrate to bandwidth.
Backend
- New BullMQ `hls-transcode` queue (in-process worker, concurrency 1)
- FFmpeg single-pass transcode → 360p/720p/1080p variants with aligned
keyframes; output at /media/local/hls/{id}/master.m3u8
- New /api/{videos|public}/{id}/hls/* routes serving signed manifests
and segments (URLs emitted as /media/* so nginx rewrites to media-api)
- Prisma: HlsStatus enum + 6 fields on Video + index, migration
- Upload + yt-dlp fetch paths enqueue transcode jobs
- ENABLE_HLS_TRANSCODE flag (default off; gates enqueue only)
- Backfill script: `npm run backfill:hls`
- media-api bumped to 4 CPU / 2G for FFmpeg headroom
Frontend
- New useHls hook: lazy-imports hls.js (kept out of main bundle),
native HLS on Safari/iOS, gives up after 2 NETWORK_ERRORs so MP4
fallback engages cleanly
- VideoPlayer, VideoViewerModal, ShortsPage, ProductDetailPage now
prefer HLS when ready; MP4 fallback is automatic
- ShortsPage prefetches next-3 master manifests via <link rel="prefetch">
- PublicVideoCard hover preview stays MP4 (avoids hls.js init latency)
Bunker Admin
This commit is contained in:
@@ -13,7 +13,8 @@
|
||||
"prisma:migrate": "prisma migrate dev",
|
||||
"prisma:migrate:deploy": "prisma migrate deploy",
|
||||
"prisma:seed": "tsx prisma/seed.ts",
|
||||
"prisma:studio": "prisma studio"
|
||||
"prisma:studio": "prisma studio",
|
||||
"backfill:hls": "tsx scripts/backfill-hls.ts"
|
||||
},
|
||||
"dependencies": {
|
||||
"@fastify/cors": "^11.2.0",
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
-- CreateEnum
|
||||
CREATE TYPE "HlsStatus" AS ENUM ('PENDING', 'PROCESSING', 'READY', 'FAILED', 'SKIPPED');
|
||||
|
||||
-- AlterTable
|
||||
ALTER TABLE "videos" ADD COLUMN "hls_job_id" TEXT,
|
||||
ADD COLUMN "hls_manifest_path" TEXT,
|
||||
ADD COLUMN "hls_status" "HlsStatus",
|
||||
ADD COLUMN "hls_transcode_error" TEXT,
|
||||
ADD COLUMN "hls_transcoded_at" TIMESTAMP(3),
|
||||
ADD COLUMN "hls_variants" JSONB;
|
||||
|
||||
-- CreateIndex
|
||||
CREATE INDEX "idx_videos_hls_status" ON "videos"("hls_status");
|
||||
@@ -1485,6 +1485,15 @@ enum DirectoryType {
|
||||
highlights
|
||||
}
|
||||
|
||||
// HLS adaptive bitrate transcoding state for the Video model.
|
||||
enum HlsStatus {
|
||||
PENDING
|
||||
PROCESSING
|
||||
READY
|
||||
FAILED
|
||||
SKIPPED
|
||||
}
|
||||
|
||||
enum ResourceCategory {
|
||||
gpu_ai
|
||||
gpu_encode
|
||||
@@ -1800,6 +1809,17 @@ model Video {
|
||||
// Uploader tracking
|
||||
uploaderId String? @map("uploader_id")
|
||||
|
||||
// HLS adaptive bitrate transcoding state.
|
||||
// null = never queued; PENDING after upload; PROCESSING when worker picks up;
|
||||
// READY when master.m3u8 + variants exist on disk; FAILED on transcode error;
|
||||
// SKIPPED when ENABLE_HLS_TRANSCODE was off at enqueue time.
|
||||
hlsStatus HlsStatus? @map("hls_status")
|
||||
hlsManifestPath String? @map("hls_manifest_path") // /media/local/hls/{id}/master.m3u8
|
||||
hlsTranscodedAt DateTime? @map("hls_transcoded_at")
|
||||
hlsTranscodeError String? @map("hls_transcode_error")
|
||||
hlsVariants Json? @map("hls_variants") // [{height, bitrate, path}, ...]
|
||||
hlsJobId String? @map("hls_job_id")
|
||||
|
||||
// Relations
|
||||
uploader User? @relation("VideoUploader", fields: [uploaderId], references: [id])
|
||||
locker User? @relation("VideoLocker", fields: [lockedById], references: [id])
|
||||
@@ -1840,6 +1860,7 @@ model Video {
|
||||
@@index([category, isPublished], map: "idx_videos_category_published")
|
||||
@@index([isShort, isPublished, isLocked], map: "idx_videos_short_published")
|
||||
@@index([uploaderId], map: "idx_videos_uploader")
|
||||
@@index([hlsStatus], map: "idx_videos_hls_status")
|
||||
@@map("videos")
|
||||
}
|
||||
|
||||
|
||||
67
api/scripts/backfill-hls.ts
Normal file
67
api/scripts/backfill-hls.ts
Normal file
@@ -0,0 +1,67 @@
|
||||
/**
|
||||
* Backfill HLS transcoding for existing videos.
|
||||
*
|
||||
* Finds every Video record that has never been queued for HLS (hlsStatus is
|
||||
* NULL) and is otherwise transcodable, then enqueues a transcode job for
|
||||
* each. Idempotent — re-runs only pick up still-NULL rows. Skips invalid or
|
||||
* zero-duration videos.
|
||||
*
|
||||
* Bypasses the ENABLE_HLS_TRANSCODE flag (calls forceSubmitTranscode)
|
||||
* because the flag is meant to gate the *upload-time* enqueue; once an
|
||||
* operator runs this script they're explicitly asking for transcoding.
|
||||
*
|
||||
* Usage:
|
||||
* docker compose exec api tsx scripts/backfill-hls.ts
|
||||
* # or after building:
|
||||
* docker compose exec api node dist/scripts/backfill-hls.js
|
||||
*/
|
||||
|
||||
import { prisma } from '../src/config/database';
|
||||
import { hlsTranscodeQueueService } from '../src/services/hls-transcode-queue.service';
|
||||
import { logger } from '../src/utils/logger';
|
||||
|
||||
async function main() {
|
||||
const candidates = await prisma.video.findMany({
|
||||
where: {
|
||||
hlsStatus: null,
|
||||
isValid: true,
|
||||
durationSeconds: { gt: 0 },
|
||||
width: { gt: 0 },
|
||||
height: { gt: 0 },
|
||||
},
|
||||
select: { id: true, filename: true, durationSeconds: true },
|
||||
orderBy: { id: 'asc' },
|
||||
});
|
||||
|
||||
if (candidates.length === 0) {
|
||||
logger.info('[backfill-hls] No videos require HLS transcoding.');
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
logger.info(`[backfill-hls] Enqueueing ${candidates.length} video(s) for HLS transcoding`);
|
||||
|
||||
let enqueued = 0;
|
||||
for (const video of candidates) {
|
||||
try {
|
||||
const jobId = await hlsTranscodeQueueService.forceSubmitTranscode(video.id);
|
||||
enqueued++;
|
||||
logger.info(`[backfill-hls] Enqueued video ${video.id} (${video.filename}) → job ${jobId}`);
|
||||
} catch (err) {
|
||||
logger.error(`[backfill-hls] Failed to enqueue video ${video.id}: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
}
|
||||
|
||||
logger.info(`[backfill-hls] Done. ${enqueued}/${candidates.length} jobs enqueued. Worker concurrency is 1, so total wall time depends on per-video transcode duration (~2 min per 1080p video).`);
|
||||
|
||||
// Give BullMQ a moment to flush, then exit cleanly.
|
||||
await new Promise((r) => setTimeout(r, 500));
|
||||
await hlsTranscodeQueueService.close();
|
||||
await prisma.$disconnect();
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
main().catch(async (err) => {
|
||||
logger.error(`[backfill-hls] Fatal: ${err instanceof Error ? err.stack : String(err)}`);
|
||||
await prisma.$disconnect().catch(() => {});
|
||||
process.exit(1);
|
||||
});
|
||||
@@ -188,6 +188,12 @@ const envSchema = z.object({
|
||||
MEDIA_ROOT: z.string().default('/media/library'),
|
||||
MEDIA_UPLOADS: z.string().default('/media/uploads'),
|
||||
MAX_UPLOAD_SIZE_GB: z.coerce.number().default(10),
|
||||
// HLS adaptive bitrate transcoding. When false, uploads are not enqueued
|
||||
// for transcoding (the worker stays registered so PENDING jobs from a
|
||||
// previous run still process if the flag is flipped back on). MP4 range-
|
||||
// request streaming continues to work as a fallback for un-transcoded
|
||||
// videos regardless of this flag.
|
||||
ENABLE_HLS_TRANSCODE: z.string().default('false'),
|
||||
|
||||
// Container Registry (remote — gitea.bnkops.com)
|
||||
GITEA_REGISTRY: z.string().default('gitea.bnkops.com/admin'),
|
||||
|
||||
@@ -21,7 +21,9 @@ import { shortsRoutes } from './modules/media/routes/shorts.routes';
|
||||
import { upvoteRoutes } from './modules/media/routes/upvote.routes';
|
||||
import { videoScheduleQueueService } from './services/video-schedule-queue.service';
|
||||
import { videoFetchQueueService } from './services/video-fetch-queue.service';
|
||||
import { hlsTranscodeQueueService } from './services/hls-transcode-queue.service';
|
||||
import { fetchRoutes } from './modules/media/routes/fetch.routes';
|
||||
import { hlsRoutes } from './modules/media/routes/hls.routes';
|
||||
import { playlistsPublicRoutes } from './modules/media/routes/playlists-public.routes';
|
||||
import { playlistsUserRoutes } from './modules/media/routes/playlists-user.routes';
|
||||
import { playlistsAdminRoutes } from './modules/media/routes/playlists-admin.routes';
|
||||
@@ -55,6 +57,7 @@ process.on('SIGTERM', async () => {
|
||||
logger.info('SIGTERM received, shutting down gracefully...');
|
||||
await videoScheduleQueueService.close();
|
||||
await videoFetchQueueService.close();
|
||||
await hlsTranscodeQueueService.close();
|
||||
fastify.close(() => {
|
||||
logger.info('Media API server closed');
|
||||
process.exit(0);
|
||||
@@ -135,6 +138,7 @@ const start = async () => {
|
||||
await fastify.register(uploadRoutes, { prefix: '/api/videos' });
|
||||
await fastify.register(videoActionsRoutes, { prefix: '/api/videos' });
|
||||
await fastify.register(videoScheduleRoutes, { prefix: '/api/videos' });
|
||||
await fastify.register(hlsRoutes, { prefix: '/api' });
|
||||
await fastify.register(videoTrackingRoutes, { prefix: '/api/track' });
|
||||
await fastify.register(reactionsRoutes, { prefix: '/api/reactions' });
|
||||
await fastify.register(publicRoutes, { prefix: '/api' });
|
||||
@@ -184,6 +188,12 @@ const start = async () => {
|
||||
videoFetchQueueService.startWorker();
|
||||
logger.info('Video fetch queue worker initialized');
|
||||
|
||||
// Start HLS transcode worker (always on; the ENABLE_HLS_TRANSCODE flag
|
||||
// gates enqueue, not worker registration, so existing PENDING jobs from
|
||||
// a prior run still process if the flag was flipped back on).
|
||||
hlsTranscodeQueueService.startWorker();
|
||||
logger.info('HLS transcode queue worker initialized');
|
||||
|
||||
if (env.ENABLE_MEDIA_FEATURES !== 'true') {
|
||||
logger.warn('Media features are disabled (ENABLE_MEDIA_FEATURES=false)');
|
||||
}
|
||||
|
||||
383
api/src/modules/media/routes/hls.routes.ts
Normal file
383
api/src/modules/media/routes/hls.routes.ts
Normal file
@@ -0,0 +1,383 @@
|
||||
import { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify';
|
||||
import { createReadStream } from 'fs';
|
||||
import { readFile, access, stat } from 'fs/promises';
|
||||
import path from 'path';
|
||||
import jwt from 'jsonwebtoken';
|
||||
import { UserRole, UserStatus } from '@prisma/client';
|
||||
import { prisma } from '../../../config/database';
|
||||
import { env } from '../../../config/env';
|
||||
import { logger } from '../../../utils/logger';
|
||||
import { hasAnyRole, MEDIA_ROLES, getUserRoles } from '../../../utils/roles';
|
||||
import { signMediaPath, verifyMediaSignature } from '../../../utils/signed-url';
|
||||
|
||||
const HLS_ROOT = '/media/local/hls';
|
||||
// 2-hour TTL on segment URLs — longer than typical viewing session, so a
|
||||
// player that has the manifest cached doesn't have to re-sign mid-playback.
|
||||
const SEGMENT_TTL_SECONDS = 2 * 60 * 60;
|
||||
// Master/variant playlists share the same TTL for consistency.
|
||||
const MANIFEST_TTL_SECONDS = SEGMENT_TTL_SECONDS;
|
||||
|
||||
// Whitelist sanitizer for path components in URLs (variant + filename).
|
||||
const SAFE_PATH_RE = /^[a-zA-Z0-9._-]+$/;
|
||||
|
||||
/**
|
||||
* Identify whether a request is from an authenticated admin/media-role user.
|
||||
* Mirrors the logic in video-streaming.routes.ts so admin HLS access works
|
||||
* the same way (Bearer JWT or signed URL params).
|
||||
*/
|
||||
async function isAdminRequest(request: FastifyRequest): Promise<boolean> {
|
||||
try {
|
||||
let userId: string | undefined;
|
||||
|
||||
const authHeader = request.headers.authorization;
|
||||
const query = request.query as Record<string, string | undefined>;
|
||||
|
||||
if (authHeader?.startsWith('Bearer ')) {
|
||||
const payload = jwt.verify(authHeader.substring(7), env.JWT_ACCESS_SECRET, { algorithms: ['HS256'] }) as {
|
||||
id: string;
|
||||
role: UserRole;
|
||||
roles?: UserRole[];
|
||||
};
|
||||
if (!hasAnyRole(payload, MEDIA_ROLES)) return false;
|
||||
userId = payload.id;
|
||||
} else if (query.sig && query.exp && query.uid) {
|
||||
const result = verifyMediaSignature(request.url, query);
|
||||
if (!result.valid) return false;
|
||||
userId = result.userId;
|
||||
}
|
||||
|
||||
if (!userId) return false;
|
||||
|
||||
const user = await prisma.user.findUnique({
|
||||
where: { id: userId },
|
||||
select: { status: true, role: true, roles: true },
|
||||
});
|
||||
if (!user || user.status !== UserStatus.ACTIVE) return false;
|
||||
return hasAnyRole({ role: user.role as UserRole, roles: getUserRoles(user) }, MEDIA_ROLES);
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Sanitize a manifest line: rewrite a relative URI (`360p/index.m3u8` or
|
||||
* `seg_00001.ts`) into an absolute path beneath the given basePath, with a
|
||||
* fresh signed-URL query string. Lines starting with `#` or empty are
|
||||
* passed through untouched.
|
||||
*
|
||||
* The path emitted to the player is the *client-side* `/media/*` path that
|
||||
* nginx rewrites to `/api/*` before reaching this server. Signatures are
|
||||
* computed against the *server-side* `/api/*` path because that's what
|
||||
* `verifyMediaSignature(request.url, ...)` sees on the inbound request.
|
||||
*/
|
||||
function rewriteManifestLines(
|
||||
manifestText: string,
|
||||
clientBasePath: string, // e.g. `/media/videos/123/hls` (browser-facing)
|
||||
serverBasePath: string, // e.g. `/api/videos/123/hls` (post-nginx-rewrite, used for signing)
|
||||
prefixSegmentsWith: string, // e.g. `360p/` for variant playlists; '' for master
|
||||
uid: string,
|
||||
ttlSeconds: number,
|
||||
): string {
|
||||
return manifestText
|
||||
.split('\n')
|
||||
.map((line) => {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed || trimmed.startsWith('#')) return line;
|
||||
const clientUrl = `${clientBasePath}/${prefixSegmentsWith}${trimmed}`;
|
||||
const serverPath = `${serverBasePath}/${prefixSegmentsWith}${trimmed}`;
|
||||
const signed = signMediaPath(serverPath, uid, ttlSeconds);
|
||||
const sep = clientUrl.includes('?') ? '&' : '?';
|
||||
return `${clientUrl}${sep}sig=${signed.sig}&exp=${signed.exp}&uid=${signed.uid}`;
|
||||
})
|
||||
.join('\n');
|
||||
}
|
||||
|
||||
/** Lookup video + HLS state from DB, returning a lightweight record. */
|
||||
async function loadVideoForHls(videoId: number, scope: 'admin' | 'public') {
|
||||
if (scope === 'public') {
|
||||
return prisma.video.findFirst({
|
||||
where: { id: videoId, isPublished: true, isLocked: false },
|
||||
select: {
|
||||
id: true,
|
||||
accessLevel: true,
|
||||
hlsStatus: true,
|
||||
hlsManifestPath: true,
|
||||
},
|
||||
});
|
||||
}
|
||||
return prisma.video.findUnique({
|
||||
where: { id: videoId },
|
||||
select: {
|
||||
id: true,
|
||||
accessLevel: true,
|
||||
hlsStatus: true,
|
||||
hlsManifestPath: true,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscription check for non-free videos on public endpoints. Mirrors
|
||||
* public.routes.ts content-gating logic.
|
||||
*/
|
||||
async function checkPublicAccess(
|
||||
request: FastifyRequest,
|
||||
reply: FastifyReply,
|
||||
accessLevel: string | null | undefined,
|
||||
): Promise<boolean> {
|
||||
if (!accessLevel || accessLevel === 'free') return true;
|
||||
|
||||
const userId = (request as any).user?.id;
|
||||
if (!userId) {
|
||||
reply.code(403).send({
|
||||
message: 'This content requires a subscription',
|
||||
accessLevel,
|
||||
requiresAuth: true,
|
||||
});
|
||||
return false;
|
||||
}
|
||||
const subscription = await prisma.userSubscription.findFirst({
|
||||
where: { userId, status: 'active' },
|
||||
include: { plan: true },
|
||||
});
|
||||
if (!subscription) {
|
||||
reply.code(403).send({
|
||||
message: 'This content requires an active subscription',
|
||||
accessLevel,
|
||||
requiresSubscription: true,
|
||||
});
|
||||
return false;
|
||||
}
|
||||
if (accessLevel === 'premium' && (subscription.plan?.tier ?? 0) < 2) {
|
||||
reply.code(403).send({
|
||||
message: 'This content requires a premium subscription',
|
||||
accessLevel,
|
||||
requiresUpgrade: true,
|
||||
});
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Resolve a path inside the HLS root, blocking traversal. */
|
||||
function resolveHlsPath(...parts: string[]): string {
|
||||
const candidate = path.resolve(path.join(HLS_ROOT, ...parts));
|
||||
if (!candidate.startsWith(path.resolve(HLS_ROOT))) {
|
||||
throw new Error('Path traversal blocked');
|
||||
}
|
||||
return candidate;
|
||||
}
|
||||
|
||||
export async function hlsRoutes(fastify: FastifyInstance) {
|
||||
// ───────────────────────────────────────────────────────────────────
|
||||
// Admin: master, variant, segment
|
||||
// ───────────────────────────────────────────────────────────────────
|
||||
|
||||
fastify.get<{ Params: { id: string } }>(
|
||||
'/videos/:id/hls/master.m3u8',
|
||||
async (request, reply) => {
|
||||
if (!(await isAdminRequest(request))) {
|
||||
return reply.code(401).send({ message: 'Authentication required' });
|
||||
}
|
||||
return serveMaster(request, reply, 'admin');
|
||||
},
|
||||
);
|
||||
|
||||
fastify.get<{ Params: { id: string; variant: string } }>(
|
||||
'/videos/:id/hls/:variant/index.m3u8',
|
||||
async (request, reply) => {
|
||||
if (!(await isAdminRequest(request))) {
|
||||
return reply.code(401).send({ message: 'Authentication required' });
|
||||
}
|
||||
return serveVariant(request, reply, 'admin');
|
||||
},
|
||||
);
|
||||
|
||||
fastify.get<{ Params: { id: string; variant: string; filename: string } }>(
|
||||
'/videos/:id/hls/:variant/:filename',
|
||||
async (request, reply) => {
|
||||
// Segments are only authorized via the signed URL embedded in the
|
||||
// variant playlist — we do NOT also accept Bearer auth here (keeps the
|
||||
// hot path tiny and avoids a DB lookup per segment).
|
||||
const query = request.query as Record<string, string | undefined>;
|
||||
const result = verifyMediaSignature(request.url, query);
|
||||
if (!result.valid) {
|
||||
return reply.code(403).send({ message: 'Invalid or expired signature' });
|
||||
}
|
||||
return serveSegment(request, reply);
|
||||
},
|
||||
);
|
||||
|
||||
// ───────────────────────────────────────────────────────────────────
|
||||
// Public: master, variant, segment (gated by publish + access level)
|
||||
// ───────────────────────────────────────────────────────────────────
|
||||
|
||||
fastify.get<{ Params: { id: string } }>(
|
||||
'/public/:id/hls/master.m3u8',
|
||||
async (request, reply) => {
|
||||
return serveMaster(request, reply, 'public');
|
||||
},
|
||||
);
|
||||
|
||||
fastify.get<{ Params: { id: string; variant: string } }>(
|
||||
'/public/:id/hls/:variant/index.m3u8',
|
||||
async (request, reply) => {
|
||||
return serveVariant(request, reply, 'public');
|
||||
},
|
||||
);
|
||||
|
||||
fastify.get<{ Params: { id: string; variant: string; filename: string } }>(
|
||||
'/public/:id/hls/:variant/:filename',
|
||||
async (request, reply) => {
|
||||
const query = request.query as Record<string, string | undefined>;
|
||||
const result = verifyMediaSignature(request.url, query);
|
||||
if (!result.valid) {
|
||||
return reply.code(403).send({ message: 'Invalid or expired signature' });
|
||||
}
|
||||
return serveSegment(request, reply);
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────
|
||||
// Handlers
|
||||
// ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
async function serveMaster(
|
||||
request: FastifyRequest<{ Params: { id: string } }>,
|
||||
reply: FastifyReply,
|
||||
scope: 'admin' | 'public',
|
||||
) {
|
||||
const videoId = parseInt(request.params.id);
|
||||
if (isNaN(videoId)) {
|
||||
return reply.code(400).send({ message: 'Invalid video ID' });
|
||||
}
|
||||
|
||||
const video = await loadVideoForHls(videoId, scope);
|
||||
if (!video || video.hlsStatus !== 'READY' || !video.hlsManifestPath) {
|
||||
return reply.code(404).send({ message: 'HLS manifest not available for this video' });
|
||||
}
|
||||
|
||||
if (scope === 'public') {
|
||||
if (!(await checkPublicAccess(request, reply, video.accessLevel))) return; // reply already sent
|
||||
}
|
||||
|
||||
// Read master.m3u8 from disk.
|
||||
let masterPath: string;
|
||||
try {
|
||||
masterPath = resolveHlsPath(String(videoId), 'master.m3u8');
|
||||
await access(masterPath);
|
||||
} catch (err) {
|
||||
logger.warn(`HLS master.m3u8 missing on disk for video ${videoId}: ${err}`);
|
||||
return reply.code(404).send({ message: 'HLS manifest file missing' });
|
||||
}
|
||||
const masterText = await readFile(masterPath, 'utf-8');
|
||||
|
||||
// The master's variant URIs look like "360p/index.m3u8". Rewrite each to
|
||||
// an absolute, server-signed URL pointing at our variant endpoint. We emit
|
||||
// browser-facing `/media/*` paths (rewritten to `/api/*` by nginx) but
|
||||
// sign against the server-side `/api/*` path the verifier will see.
|
||||
const clientBase = scope === 'admin'
|
||||
? `/media/videos/${videoId}/hls`
|
||||
: `/media/public/${videoId}/hls`;
|
||||
const serverBase = scope === 'admin'
|
||||
? `/api/videos/${videoId}/hls`
|
||||
: `/api/public/${videoId}/hls`;
|
||||
const uid = scope === 'admin'
|
||||
? ((request.query as Record<string, string | undefined>).uid ?? 'admin')
|
||||
: `public-${videoId}`;
|
||||
|
||||
const rewritten = rewriteManifestLines(masterText, clientBase, serverBase, '', uid, MANIFEST_TTL_SECONDS);
|
||||
|
||||
reply
|
||||
.header('Content-Type', 'application/vnd.apple.mpegurl')
|
||||
.header('Cache-Control', 'no-store')
|
||||
.send(rewritten);
|
||||
}
|
||||
|
||||
async function serveVariant(
|
||||
request: FastifyRequest<{ Params: { id: string; variant: string } }>,
|
||||
reply: FastifyReply,
|
||||
scope: 'admin' | 'public',
|
||||
) {
|
||||
const videoId = parseInt(request.params.id);
|
||||
const variant = request.params.variant;
|
||||
if (isNaN(videoId) || !SAFE_PATH_RE.test(variant)) {
|
||||
return reply.code(400).send({ message: 'Invalid params' });
|
||||
}
|
||||
|
||||
const video = await loadVideoForHls(videoId, scope);
|
||||
if (!video || video.hlsStatus !== 'READY') {
|
||||
return reply.code(404).send({ message: 'HLS manifest not available' });
|
||||
}
|
||||
if (scope === 'public') {
|
||||
if (!(await checkPublicAccess(request, reply, video.accessLevel))) return;
|
||||
}
|
||||
|
||||
let variantPath: string;
|
||||
try {
|
||||
variantPath = resolveHlsPath(String(videoId), variant, 'index.m3u8');
|
||||
await access(variantPath);
|
||||
} catch {
|
||||
return reply.code(404).send({ message: 'Variant playlist not found' });
|
||||
}
|
||||
const variantText = await readFile(variantPath, 'utf-8');
|
||||
|
||||
const clientBase = scope === 'admin'
|
||||
? `/media/videos/${videoId}/hls`
|
||||
: `/media/public/${videoId}/hls`;
|
||||
const serverBase = scope === 'admin'
|
||||
? `/api/videos/${videoId}/hls`
|
||||
: `/api/public/${videoId}/hls`;
|
||||
const uid = scope === 'admin'
|
||||
? ((request.query as Record<string, string | undefined>).uid ?? 'admin')
|
||||
: `public-${videoId}`;
|
||||
|
||||
const rewritten = rewriteManifestLines(
|
||||
variantText,
|
||||
clientBase,
|
||||
serverBase,
|
||||
`${variant}/`,
|
||||
uid,
|
||||
SEGMENT_TTL_SECONDS,
|
||||
);
|
||||
|
||||
reply
|
||||
.header('Content-Type', 'application/vnd.apple.mpegurl')
|
||||
.header('Cache-Control', 'no-store')
|
||||
.send(rewritten);
|
||||
}
|
||||
|
||||
async function serveSegment(
|
||||
request: FastifyRequest<{ Params: { id: string; variant: string; filename: string } }>,
|
||||
reply: FastifyReply,
|
||||
) {
|
||||
const videoId = parseInt(request.params.id);
|
||||
const { variant, filename } = request.params;
|
||||
|
||||
if (isNaN(videoId) || !SAFE_PATH_RE.test(variant) || !SAFE_PATH_RE.test(filename)) {
|
||||
return reply.code(400).send({ message: 'Invalid params' });
|
||||
}
|
||||
|
||||
let segmentPath: string;
|
||||
try {
|
||||
segmentPath = resolveHlsPath(String(videoId), variant, filename);
|
||||
await access(segmentPath);
|
||||
} catch {
|
||||
return reply.code(404).send({ message: 'Segment not found' });
|
||||
}
|
||||
|
||||
const stats = await stat(segmentPath);
|
||||
const isPlaylist = filename.endsWith('.m3u8');
|
||||
const contentType = isPlaylist ? 'application/vnd.apple.mpegurl' : 'video/mp2t';
|
||||
|
||||
reply
|
||||
.header('Content-Type', contentType)
|
||||
.header('Content-Length', stats.size)
|
||||
// Segments are content-addressed in the sense that {videoId}/{variant}/{name}
|
||||
// never changes content; safe to cache aggressively at the browser. The
|
||||
// signature in the URL keeps cache keys per-session.
|
||||
.header('Cache-Control', 'private, max-age=3600');
|
||||
|
||||
return reply.send(createReadStream(segmentPath));
|
||||
}
|
||||
@@ -7,6 +7,7 @@ import { randomUUID } from 'crypto';
|
||||
import { prisma } from '../../../config/database';
|
||||
import { extractVideoMetadata, validateVideoFile } from '../services/ffprobe.service';
|
||||
import { ThumbnailService } from '../services/thumbnail.service';
|
||||
import { hlsTranscodeQueueService } from '../../../services/hls-transcode-queue.service';
|
||||
import { logger } from '../../../utils/logger';
|
||||
import { z } from 'zod';
|
||||
import { requireAdminRole } from '../middleware/auth';
|
||||
@@ -126,6 +127,13 @@ async function uploadVideo(request: FastifyRequest, reply: FastifyReply) {
|
||||
logger.error(`Failed to generate thumbnail for video ${video.id}:`, thumbnailError);
|
||||
}
|
||||
|
||||
// Enqueue HLS transcode (no-op when ENABLE_HLS_TRANSCODE=false; sets SKIPPED).
|
||||
try {
|
||||
await hlsTranscodeQueueService.submitTranscode(video.id);
|
||||
} catch (hlsErr) {
|
||||
logger.error(`Failed to enqueue HLS transcode for video ${video.id}:`, hlsErr);
|
||||
}
|
||||
|
||||
return reply.code(201).send({
|
||||
message: 'Video uploaded successfully',
|
||||
video,
|
||||
@@ -247,6 +255,13 @@ async function uploadBatch(request: FastifyRequest, reply: FastifyReply) {
|
||||
logger.error(`Failed to generate thumbnail for video ${video.id}:`, thumbnailError);
|
||||
}
|
||||
|
||||
// Enqueue HLS transcode (no-op when flag off).
|
||||
try {
|
||||
await hlsTranscodeQueueService.submitTranscode(video.id);
|
||||
} catch (hlsErr) {
|
||||
logger.error(`Failed to enqueue HLS transcode for video ${video.id}:`, hlsErr);
|
||||
}
|
||||
|
||||
results.push({
|
||||
filename: file.filename,
|
||||
success: true,
|
||||
|
||||
@@ -9,7 +9,7 @@ import { prisma } from '../../../config/database';
|
||||
import { env } from '../../../config/env';
|
||||
import { logger } from '../../../utils/logger';
|
||||
import { hasAnyRole, MEDIA_ROLES, getUserRoles } from '../../../utils/roles';
|
||||
import { verifyMediaSignature } from '../../../utils/signed-url';
|
||||
import { signMediaPath, verifyMediaSignature } from '../../../utils/signed-url';
|
||||
|
||||
/**
|
||||
* Check if the request is from an authenticated admin user.
|
||||
@@ -284,6 +284,43 @@ export async function videoStreamingRoutes(fastify: FastifyInstance) {
|
||||
? `/media/videos/${video.id}/thumbnail`
|
||||
: null;
|
||||
|
||||
// HLS manifest URL — only present when transcoding has completed.
|
||||
// We emit browser-facing `/media/*` paths (rewritten by nginx to
|
||||
// `/api/*` and proxied to media-api). For admin previews we sign
|
||||
// against the post-rewrite server-side path so the verifier matches.
|
||||
let hlsManifestUrl: string | null = null;
|
||||
if (video.hlsStatus === 'READY' && video.hlsManifestPath) {
|
||||
const clientPath = admin
|
||||
? `/media/videos/${video.id}/hls/master.m3u8`
|
||||
: `/media/public/${video.id}/hls/master.m3u8`;
|
||||
const serverPath = admin
|
||||
? `/api/videos/${video.id}/hls/master.m3u8`
|
||||
: `/api/public/${video.id}/hls/master.m3u8`;
|
||||
if (admin) {
|
||||
let uid = 'admin';
|
||||
try {
|
||||
const authHeader = request.headers.authorization;
|
||||
if (authHeader?.startsWith('Bearer ')) {
|
||||
const payload = jwt.verify(
|
||||
authHeader.substring(7),
|
||||
env.JWT_ACCESS_SECRET,
|
||||
{ algorithms: ['HS256'] },
|
||||
) as { id?: string };
|
||||
if (payload.id) uid = payload.id;
|
||||
} else {
|
||||
const q = request.query as Record<string, string | undefined>;
|
||||
if (q.uid) uid = q.uid;
|
||||
}
|
||||
} catch { /* keep default */ }
|
||||
const signed = signMediaPath(serverPath, uid, 2 * 60 * 60);
|
||||
hlsManifestUrl = `${clientPath}?sig=${signed.sig}&exp=${signed.exp}&uid=${signed.uid}`;
|
||||
} else {
|
||||
// Public manifest: nginx-rewritten path; the public master route
|
||||
// is unsigned (gated by isPublished + access level on the server).
|
||||
hlsManifestUrl = clientPath;
|
||||
}
|
||||
}
|
||||
|
||||
// Return public metadata
|
||||
return {
|
||||
id: video.id,
|
||||
@@ -296,6 +333,8 @@ export async function videoStreamingRoutes(fastify: FastifyInstance) {
|
||||
quality: video.quality,
|
||||
streamUrl,
|
||||
thumbnailUrl,
|
||||
hlsStatus: video.hlsStatus,
|
||||
hlsManifestUrl,
|
||||
createdAt: video.createdAt,
|
||||
};
|
||||
} catch (error) {
|
||||
|
||||
211
api/src/modules/media/services/hls-transcode.service.ts
Normal file
211
api/src/modules/media/services/hls-transcode.service.ts
Normal file
@@ -0,0 +1,211 @@
|
||||
import { spawn } from 'child_process';
|
||||
import fs from 'fs/promises';
|
||||
import path from 'path';
|
||||
import { logger } from '../../../utils/logger';
|
||||
|
||||
const HLS_ROOT = '/media/local/hls';
|
||||
const FFMPEG_TIMEOUT_MS = 60 * 60 * 1000; // 1 hour cap; long-form video can exceed default 30s
|
||||
|
||||
export interface HlsVariant {
|
||||
name: string; // '360p' | '720p' | '1080p'
|
||||
height: number; // short-side height in pixels
|
||||
bitrate: number; // target video bitrate in kbps
|
||||
path: string; // relative path under {videoId}/, e.g. '360p/index.m3u8'
|
||||
}
|
||||
|
||||
export interface HlsTranscodeOptions {
|
||||
videoId: number;
|
||||
sourcePath: string;
|
||||
durationSeconds: number;
|
||||
sourceWidth: number;
|
||||
sourceHeight: number;
|
||||
/** Optional progress callback, called with 0-100 as transcoding advances. */
|
||||
onProgress?: (percent: number) => void | Promise<void>;
|
||||
}
|
||||
|
||||
export interface HlsTranscodeResult {
|
||||
manifestPath: string; // absolute path to master.m3u8 on disk
|
||||
manifestRelativePath: string; // e.g. {videoId}/master.m3u8 (relative to HLS_ROOT)
|
||||
variants: HlsVariant[];
|
||||
}
|
||||
|
||||
// Bitrate ladder. Each rung is { name, height, videoKbps, maxKbps, bufKbps, audioKbps }.
|
||||
// Ladder is keyed off short-side height so a 1080×1920 vertical short still
|
||||
// gets 360p/720p/1080p *short-side* renditions (width follows via scale=-2:H).
|
||||
const LADDER = [
|
||||
{ name: '360p', height: 360, videoKbps: 800, maxKbps: 856, bufKbps: 1200, audioKbps: 96 },
|
||||
{ name: '720p', height: 720, videoKbps: 2800, maxKbps: 2996, bufKbps: 4200, audioKbps: 128 },
|
||||
{ name: '1080p', height: 1080, videoKbps: 5000, maxKbps: 5350, bufKbps: 7500, audioKbps: 128 },
|
||||
];
|
||||
|
||||
/**
|
||||
* Transcode a source video to HLS adaptive bitrate using a single FFmpeg
|
||||
* invocation. One decode pass produces all variants in parallel, with
|
||||
* keyframes aligned every 2s so hls.js can switch renditions cleanly.
|
||||
*
|
||||
* Output layout:
|
||||
* /media/local/hls/{videoId}/master.m3u8
|
||||
* /media/local/hls/{videoId}/360p/index.m3u8
|
||||
* /media/local/hls/{videoId}/360p/seg_00000.ts
|
||||
* ...
|
||||
*/
|
||||
export async function transcodeToHls(opts: HlsTranscodeOptions): Promise<HlsTranscodeResult> {
|
||||
const { videoId, sourcePath, durationSeconds, sourceWidth, sourceHeight, onProgress } = opts;
|
||||
|
||||
// Pick variants up to the source's short-side resolution. Always include 360p.
|
||||
const shortSide = Math.min(sourceWidth, sourceHeight);
|
||||
const variants = LADDER.filter((v, i) => i === 0 || v.height <= shortSide);
|
||||
|
||||
const outDir = path.join(HLS_ROOT, String(videoId));
|
||||
await fs.rm(outDir, { recursive: true, force: true });
|
||||
await fs.mkdir(outDir, { recursive: true });
|
||||
|
||||
// Build the filter_complex graph: split video N ways, scale each.
|
||||
const splitTargets = variants.map((_, i) => `[v${i}]`).join('');
|
||||
const scaleFilters = variants
|
||||
.map((v, i) => `[v${i}]scale=-2:${v.height}[v${i}o]`)
|
||||
.join('; ');
|
||||
const filterComplex = `[0:v]split=${variants.length}${splitTargets}; ${scaleFilters}`;
|
||||
|
||||
// Per-stream encode args.
|
||||
const streamArgs: string[] = [];
|
||||
variants.forEach((v, i) => {
|
||||
streamArgs.push(
|
||||
'-map', `[v${i}o]`,
|
||||
`-c:v:${i}`, 'libx264',
|
||||
`-b:v:${i}`, `${v.videoKbps}k`,
|
||||
`-maxrate:v:${i}`, `${v.maxKbps}k`,
|
||||
`-bufsize:v:${i}`, `${v.bufKbps}k`,
|
||||
);
|
||||
});
|
||||
|
||||
// Audio: map source audio once per variant so each rendition has its own audio track.
|
||||
variants.forEach((v, i) => {
|
||||
streamArgs.push('-map', 'a:0?', `-c:a:${i}`, 'aac', `-b:a:${i}`, `${v.audioKbps}k`, '-ac', '2');
|
||||
});
|
||||
|
||||
// var_stream_map associates video+audio streams per variant.
|
||||
const varStreamMap = variants
|
||||
.map((v, i) => `v:${i},a:${i},name:${v.name}`)
|
||||
.join(' ');
|
||||
|
||||
const ffmpegArgs = [
|
||||
'-hide_banner',
|
||||
'-y',
|
||||
'-i', sourcePath,
|
||||
'-filter_complex', filterComplex,
|
||||
...streamArgs,
|
||||
'-preset', 'veryfast',
|
||||
'-profile:v', 'main',
|
||||
'-sc_threshold', '0',
|
||||
'-g', '48',
|
||||
'-keyint_min', '48',
|
||||
'-force_key_frames', 'expr:gte(t,n_forced*2)',
|
||||
'-hls_time', '4',
|
||||
'-hls_playlist_type', 'vod',
|
||||
'-hls_flags', 'independent_segments',
|
||||
'-hls_segment_filename', path.join(outDir, '%v', 'seg_%05d.ts'),
|
||||
'-master_pl_name', 'master.m3u8',
|
||||
'-var_stream_map', varStreamMap,
|
||||
path.join(outDir, '%v', 'index.m3u8'),
|
||||
];
|
||||
|
||||
// Pre-create per-variant subdirs so FFmpeg can write segments.
|
||||
for (const v of variants) {
|
||||
await fs.mkdir(path.join(outDir, v.name), { recursive: true });
|
||||
}
|
||||
|
||||
logger.info(`[hls] Starting transcode for video ${videoId} (${variants.length} variants: ${variants.map(v => v.name).join(', ')})`);
|
||||
|
||||
await runFfmpeg(ffmpegArgs, durationSeconds, onProgress);
|
||||
|
||||
// FFmpeg's HLS muxer writes per-variant playlists into directories named
|
||||
// by the var_stream_map's name. Confirm master.m3u8 exists.
|
||||
const masterPath = path.join(outDir, 'master.m3u8');
|
||||
try {
|
||||
await fs.access(masterPath);
|
||||
} catch {
|
||||
// Cleanup partial output.
|
||||
await fs.rm(outDir, { recursive: true, force: true });
|
||||
throw new Error('FFmpeg completed but master.m3u8 was not produced');
|
||||
}
|
||||
|
||||
const result: HlsTranscodeResult = {
|
||||
manifestPath: masterPath,
|
||||
manifestRelativePath: path.join(String(videoId), 'master.m3u8'),
|
||||
variants: variants.map(v => ({
|
||||
name: v.name,
|
||||
height: v.height,
|
||||
bitrate: v.videoKbps,
|
||||
path: path.join(v.name, 'index.m3u8'),
|
||||
})),
|
||||
};
|
||||
|
||||
logger.info(`[hls] Transcode complete for video ${videoId}`);
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Spawn FFmpeg with a hard timeout and stderr-based progress parsing.
|
||||
* FFmpeg writes "time=HH:MM:SS.cc" lines to stderr periodically; we parse
|
||||
* those and call onProgress with 0-100.
|
||||
*/
|
||||
function runFfmpeg(
|
||||
args: string[],
|
||||
durationSeconds: number,
|
||||
onProgress?: (percent: number) => void | Promise<void>,
|
||||
): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const child = spawn('ffmpeg', args);
|
||||
let stderrTail = '';
|
||||
let lastReportedPercent = -1;
|
||||
|
||||
const timeout = setTimeout(() => {
|
||||
child.kill('SIGKILL');
|
||||
reject(new Error(`FFmpeg timeout after ${FFMPEG_TIMEOUT_MS / 1000}s`));
|
||||
}, FFMPEG_TIMEOUT_MS);
|
||||
|
||||
child.stderr.on('data', (data: Buffer) => {
|
||||
const text = data.toString();
|
||||
// Keep a tail for error reporting on non-zero exit.
|
||||
stderrTail = (stderrTail + text).slice(-4000);
|
||||
|
||||
if (!onProgress || durationSeconds <= 0) return;
|
||||
// Parse "time=HH:MM:SS.cc" — newest occurrence in this chunk.
|
||||
const matches = text.match(/time=(\d+):(\d+):(\d+(?:\.\d+)?)/g);
|
||||
if (!matches || matches.length === 0) return;
|
||||
const last = matches[matches.length - 1];
|
||||
const m = last.match(/time=(\d+):(\d+):(\d+(?:\.\d+)?)/);
|
||||
if (!m) return;
|
||||
const elapsed = Number(m[1]) * 3600 + Number(m[2]) * 60 + Number(m[3]);
|
||||
const percent = Math.min(99, Math.floor((elapsed / durationSeconds) * 100));
|
||||
if (percent > lastReportedPercent) {
|
||||
lastReportedPercent = percent;
|
||||
// Fire-and-forget; queue can swallow errors here.
|
||||
Promise.resolve(onProgress(percent)).catch(() => {});
|
||||
}
|
||||
});
|
||||
|
||||
child.on('close', (code) => {
|
||||
clearTimeout(timeout);
|
||||
if (code === 0) {
|
||||
resolve();
|
||||
} else {
|
||||
reject(new Error(`FFmpeg exited with code ${code}: ${stderrTail.slice(-1000)}`));
|
||||
}
|
||||
});
|
||||
|
||||
child.on('error', (err) => {
|
||||
clearTimeout(timeout);
|
||||
reject(err);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove the HLS output directory for a video (e.g. on transcode failure or video deletion).
|
||||
*/
|
||||
export async function cleanupHlsOutput(videoId: number): Promise<void> {
|
||||
const dir = path.join(HLS_ROOT, String(videoId));
|
||||
await fs.rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
238
api/src/services/hls-transcode-queue.service.ts
Normal file
238
api/src/services/hls-transcode-queue.service.ts
Normal file
@@ -0,0 +1,238 @@
|
||||
import { Queue, Worker, type Job } from 'bullmq';
|
||||
import Redis from 'ioredis';
|
||||
import { env } from '../config/env';
|
||||
import { prisma } from '../config/database';
|
||||
import {
|
||||
transcodeToHls,
|
||||
cleanupHlsOutput,
|
||||
type HlsTranscodeResult,
|
||||
} from '../modules/media/services/hls-transcode.service';
|
||||
import { logger } from '../utils/logger';
|
||||
|
||||
interface HlsTranscodeJobData {
|
||||
videoId: number;
|
||||
}
|
||||
|
||||
interface HlsTranscodeJobResult {
|
||||
videoId: number;
|
||||
variants: Array<{ name: string; height: number; bitrate: number; path: string }>;
|
||||
}
|
||||
|
||||
const QUEUE_NAME = 'hls-transcode';
|
||||
|
||||
class HlsTranscodeQueueService {
|
||||
private queue: Queue<HlsTranscodeJobData, HlsTranscodeJobResult>;
|
||||
private worker: Worker<HlsTranscodeJobData, HlsTranscodeJobResult> | null = null;
|
||||
private redis: Redis | null = null;
|
||||
|
||||
constructor() {
|
||||
this.queue = new Queue(QUEUE_NAME, {
|
||||
connection: { url: env.REDIS_URL },
|
||||
defaultJobOptions: {
|
||||
attempts: 2,
|
||||
backoff: { type: 'exponential', delay: 60_000 },
|
||||
removeOnComplete: { age: 7 * 24 * 60 * 60, count: 200 },
|
||||
removeOnFail: { age: 30 * 24 * 60 * 60 },
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private getRedis(): Redis {
|
||||
if (!this.redis) {
|
||||
this.redis = new Redis(env.REDIS_URL);
|
||||
}
|
||||
return this.redis;
|
||||
}
|
||||
|
||||
/** Append a log line for a job (Redis list with 24h TTL + pubsub for SSE). */
|
||||
private async appendJobLog(jobId: string, line: string): Promise<void> {
|
||||
const key = `hls-log:${jobId}`;
|
||||
const redis = this.getRedis();
|
||||
await redis.rpush(key, line);
|
||||
await redis.expire(key, 86400);
|
||||
await redis.publish(`hls-log-stream:${jobId}`, line);
|
||||
}
|
||||
|
||||
/** Get accumulated log lines for a job. */
|
||||
async getJobLog(jobId: string): Promise<string[]> {
|
||||
const key = `hls-log:${jobId}`;
|
||||
return this.getRedis().lrange(key, 0, -1);
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the in-process worker. Concurrency is 1 because FFmpeg saturates
|
||||
* the available cores; running two transcodes in parallel just slows both
|
||||
* down and risks OOM.
|
||||
*/
|
||||
startWorker(): void {
|
||||
this.worker = new Worker<HlsTranscodeJobData, HlsTranscodeJobResult>(
|
||||
QUEUE_NAME,
|
||||
async (job) => this.processJob(job),
|
||||
{
|
||||
connection: { url: env.REDIS_URL },
|
||||
concurrency: 1,
|
||||
},
|
||||
);
|
||||
|
||||
this.worker.on('completed', (job) => {
|
||||
logger.info(`[hls] job ${job.id} completed for video ${job.data.videoId}`);
|
||||
});
|
||||
|
||||
this.worker.on('failed', (job, err) => {
|
||||
logger.error(`[hls] job ${job?.id} failed: ${err.message}`);
|
||||
});
|
||||
|
||||
logger.info('[hls] HLS transcode queue worker started');
|
||||
}
|
||||
|
||||
private async processJob(job: Job<HlsTranscodeJobData>): Promise<HlsTranscodeJobResult> {
|
||||
const { videoId } = job.data;
|
||||
const jobId = job.id!;
|
||||
|
||||
await this.appendJobLog(jobId, `Starting HLS transcode for video ${videoId}`);
|
||||
|
||||
const video = await prisma.video.findUnique({ where: { id: videoId } });
|
||||
if (!video) {
|
||||
throw new Error(`Video ${videoId} not found`);
|
||||
}
|
||||
if (!video.path) {
|
||||
throw new Error(`Video ${videoId} has no source path`);
|
||||
}
|
||||
if (!video.durationSeconds || video.durationSeconds <= 0) {
|
||||
throw new Error(`Video ${videoId} has no duration`);
|
||||
}
|
||||
if (!video.width || !video.height) {
|
||||
throw new Error(`Video ${videoId} has no dimensions`);
|
||||
}
|
||||
|
||||
await prisma.video.update({
|
||||
where: { id: videoId },
|
||||
data: {
|
||||
hlsStatus: 'PROCESSING',
|
||||
hlsJobId: jobId,
|
||||
hlsTranscodeError: null,
|
||||
},
|
||||
});
|
||||
|
||||
let result: HlsTranscodeResult;
|
||||
try {
|
||||
result = await transcodeToHls({
|
||||
videoId,
|
||||
sourcePath: video.path,
|
||||
durationSeconds: video.durationSeconds,
|
||||
sourceWidth: video.width,
|
||||
sourceHeight: video.height,
|
||||
onProgress: async (percent) => {
|
||||
await job.updateProgress(percent);
|
||||
// Coarse log lines so the UI can stream progress without flooding.
|
||||
if (percent % 10 === 0) {
|
||||
await this.appendJobLog(jobId, `Transcoding: ${percent}%`);
|
||||
}
|
||||
},
|
||||
});
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
await this.appendJobLog(jobId, `FAILED: ${message}`);
|
||||
// Best-effort cleanup; transcodeToHls already rms partial output on failure.
|
||||
await cleanupHlsOutput(videoId).catch(() => {});
|
||||
await prisma.video.update({
|
||||
where: { id: videoId },
|
||||
data: {
|
||||
hlsStatus: 'FAILED',
|
||||
hlsTranscodeError: message.slice(0, 1000),
|
||||
},
|
||||
});
|
||||
throw err; // BullMQ will retry per attempts/backoff config.
|
||||
}
|
||||
|
||||
await prisma.video.update({
|
||||
where: { id: videoId },
|
||||
data: {
|
||||
hlsStatus: 'READY',
|
||||
hlsManifestPath: result.manifestRelativePath,
|
||||
hlsTranscodedAt: new Date(),
|
||||
hlsTranscodeError: null,
|
||||
hlsVariants: result.variants as unknown as object,
|
||||
},
|
||||
});
|
||||
|
||||
await this.appendJobLog(jobId, `Transcode complete: ${result.variants.map(v => v.name).join(', ')}`);
|
||||
await this.getRedis().publish(`hls-log-stream:${jobId}`, '__DONE__');
|
||||
|
||||
return {
|
||||
videoId,
|
||||
variants: result.variants,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Submit a transcode job. When ENABLE_HLS_TRANSCODE=false, this is a no-op
|
||||
* that marks the video as SKIPPED (so the upload flow stays synchronous and
|
||||
* the operator can opt in later via the backfill script).
|
||||
*/
|
||||
async submitTranscode(videoId: number): Promise<{ jobId: string | null; skipped: boolean }> {
|
||||
if (env.ENABLE_HLS_TRANSCODE !== 'true') {
|
||||
await prisma.video.update({
|
||||
where: { id: videoId },
|
||||
data: { hlsStatus: 'SKIPPED' },
|
||||
});
|
||||
return { jobId: null, skipped: true };
|
||||
}
|
||||
|
||||
const job = await this.queue.add('transcode', { videoId });
|
||||
await prisma.video.update({
|
||||
where: { id: videoId },
|
||||
data: { hlsStatus: 'PENDING', hlsJobId: job.id ?? null },
|
||||
});
|
||||
|
||||
logger.info(`[hls] enqueued transcode job ${job.id} for video ${videoId}`);
|
||||
return { jobId: job.id ?? null, skipped: false };
|
||||
}
|
||||
|
||||
/**
|
||||
* Force-enqueue a transcode job, bypassing the ENABLE_HLS_TRANSCODE flag.
|
||||
* Used by the backfill script so an admin can run the backfill against
|
||||
* existing videos without flipping the flag for new uploads.
|
||||
*/
|
||||
async forceSubmitTranscode(videoId: number): Promise<string> {
|
||||
const job = await this.queue.add('transcode', { videoId });
|
||||
await prisma.video.update({
|
||||
where: { id: videoId },
|
||||
data: { hlsStatus: 'PENDING', hlsJobId: job.id ?? null },
|
||||
});
|
||||
logger.info(`[hls] (forced) enqueued transcode job ${job.id} for video ${videoId}`);
|
||||
return job.id!;
|
||||
}
|
||||
|
||||
/** Get a single job by ID. */
|
||||
async getJob(jobId: string) {
|
||||
const job = await this.queue.getJob(jobId);
|
||||
if (!job) return null;
|
||||
|
||||
const state = await job.getState();
|
||||
return {
|
||||
id: job.id,
|
||||
data: job.data,
|
||||
state,
|
||||
progress: job.progress as number,
|
||||
returnvalue: job.returnvalue,
|
||||
failedReason: job.failedReason,
|
||||
timestamp: job.timestamp,
|
||||
finishedOn: job.finishedOn,
|
||||
processedOn: job.processedOn,
|
||||
};
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
if (this.worker) {
|
||||
await this.worker.close();
|
||||
}
|
||||
await this.queue.close();
|
||||
if (this.redis) {
|
||||
await this.redis.quit();
|
||||
}
|
||||
logger.info('[hls] HLS transcode queue closed');
|
||||
}
|
||||
}
|
||||
|
||||
export const hlsTranscodeQueueService = new HlsTranscodeQueueService();
|
||||
@@ -372,6 +372,19 @@ class VideoFetchQueueService {
|
||||
await this.appendJobLog(jobId, `Thumbnail generation failed (non-fatal): ${thumbnailErr instanceof Error ? thumbnailErr.message : 'Unknown error'}`);
|
||||
}
|
||||
|
||||
// Enqueue HLS transcode (lazy import to avoid module-cycle with media-server bootstrap).
|
||||
try {
|
||||
const { hlsTranscodeQueueService } = await import('./hls-transcode-queue.service');
|
||||
const result = await hlsTranscodeQueueService.submitTranscode(video.id);
|
||||
if (result.skipped) {
|
||||
await this.appendJobLog(jobId, 'HLS transcode skipped (flag off)');
|
||||
} else {
|
||||
await this.appendJobLog(jobId, `HLS transcode enqueued (job ${result.jobId})`);
|
||||
}
|
||||
} catch (hlsErr) {
|
||||
await this.appendJobLog(jobId, `HLS enqueue failed (non-fatal): ${hlsErr instanceof Error ? hlsErr.message : 'Unknown error'}`);
|
||||
}
|
||||
|
||||
resolve({ videoId: video.id, title });
|
||||
} catch (err) {
|
||||
reject(err);
|
||||
|
||||
Reference in New Issue
Block a user