initial commit
This commit is contained in:
Daniil
@@ -0,0 +1,355 @@
|
||||
import { Queue, Worker, type Job } from "bullmq";
|
||||
import { serverConfig } from "@/srv/config";
|
||||
import {
|
||||
S3Service,
|
||||
type UploadProgress,
|
||||
} from "@/srv/services/s3";
|
||||
import {
|
||||
RENDER_CANCELLED_ERROR_MESSAGE,
|
||||
renderCaptionedVideo,
|
||||
type RenderProgress,
|
||||
} from "@/srv/services/render_video";
|
||||
import { sendWebhook } from "@/srv/services/webhook";
|
||||
import type { DocumentInterface } from "@/srv/types/DocumentSchema";
|
||||
import type { CaptionStyleConfigType } from "@/srv/types/CaptionStyleSchema";
|
||||
|
||||
const QUEUE_NAME = "caption-renders";
|
||||
const PROGRESS_THROTTLE_MS = 3_000;
|
||||
const UPLOAD_PROGRESS_THROTTLE_MS = 1_500;
|
||||
const RENDER_PROGRESS_MIN_PCT = 5;
|
||||
const RENDER_PROGRESS_MAX_PCT = 95;
|
||||
const UPLOAD_PROGRESS_START_PCT = 95;
|
||||
const UPLOAD_PROGRESS_END_PCT = 99;
|
||||
const KILOBYTE = 1024;
|
||||
const MEGABYTE = 1024 * KILOBYTE;
|
||||
const RENDER_CANCELLATION_KEY_PREFIX = "caption-render-cancel:";
|
||||
const RENDER_CANCELLATION_TTL_SECONDS = 60 * 60;
|
||||
const RENDER_CANCELLATION_POLL_MS = 1_000;
|
||||
|
||||
export type RenderJobData = {
|
||||
renderId: string;
|
||||
folder?: string;
|
||||
videoSrc: string;
|
||||
transcription: DocumentInterface;
|
||||
styleConfig?: CaptionStyleConfigType;
|
||||
callbackUrl?: string;
|
||||
};
|
||||
|
||||
export type RenderJobResult = {
|
||||
outputPath: string;
|
||||
callbackDelivered: boolean;
|
||||
};
|
||||
|
||||
export type CancelRenderResult = {
|
||||
renderId: string;
|
||||
status:
|
||||
| "cancelled"
|
||||
| "cancellation_requested"
|
||||
| "completed"
|
||||
| "failed"
|
||||
| "not_found";
|
||||
};
|
||||
|
||||
function formatBytes(bytes: number): string {
|
||||
if (bytes >= MEGABYTE) {
|
||||
return `${(bytes / MEGABYTE).toFixed(1)} MB`;
|
||||
}
|
||||
|
||||
if (bytes >= KILOBYTE) {
|
||||
return `${(bytes / KILOBYTE).toFixed(1)} KB`;
|
||||
}
|
||||
|
||||
return `${bytes} B`;
|
||||
}
|
||||
|
||||
function getRenderCancellationKey(renderId: string): string {
|
||||
return `${RENDER_CANCELLATION_KEY_PREFIX}${renderId}`;
|
||||
}
|
||||
|
||||
async function requestRenderCancellation(renderId: string): Promise<void> {
|
||||
const client = await renderQueue.client;
|
||||
await client.set(
|
||||
getRenderCancellationKey(renderId),
|
||||
"1",
|
||||
"EX",
|
||||
RENDER_CANCELLATION_TTL_SECONDS,
|
||||
);
|
||||
}
|
||||
|
||||
async function isRenderCancellationRequested(renderId: string): Promise<boolean> {
|
||||
const client = await renderQueue.client;
|
||||
const value = await client.get(getRenderCancellationKey(renderId));
|
||||
return value === "1";
|
||||
}
|
||||
|
||||
async function clearRenderCancellation(renderId: string): Promise<void> {
|
||||
const client = await renderQueue.client;
|
||||
await client.del(getRenderCancellationKey(renderId));
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Queue (enqueue side)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const redisConnection = { url: serverConfig.REDIS_URL };
|
||||
|
||||
export const renderQueue = new Queue<RenderJobData, RenderJobResult>(
|
||||
QUEUE_NAME,
|
||||
{ connection: redisConnection },
|
||||
);
|
||||
|
||||
export async function enqueueRender(data: RenderJobData): Promise<string> {
|
||||
const job = await renderQueue.add("render", data, {
|
||||
jobId: data.renderId,
|
||||
attempts: 2,
|
||||
backoff: { type: "exponential", delay: 5_000 },
|
||||
removeOnComplete: { age: 3600 },
|
||||
removeOnFail: { age: 7200 },
|
||||
});
|
||||
return job.id!;
|
||||
}
|
||||
|
||||
export async function cancelRender(renderId: string): Promise<CancelRenderResult> {
|
||||
const job = await renderQueue.getJob(renderId);
|
||||
if (!job) {
|
||||
return { renderId, status: "not_found" };
|
||||
}
|
||||
|
||||
const state = await job.getState();
|
||||
if (state === "completed") {
|
||||
await clearRenderCancellation(renderId);
|
||||
return { renderId, status: "completed" };
|
||||
}
|
||||
|
||||
if (state === "failed") {
|
||||
await clearRenderCancellation(renderId);
|
||||
return { renderId, status: "failed" };
|
||||
}
|
||||
|
||||
if (state === "active") {
|
||||
await requestRenderCancellation(renderId);
|
||||
return { renderId, status: "cancellation_requested" };
|
||||
}
|
||||
|
||||
await clearRenderCancellation(renderId);
|
||||
await job.remove();
|
||||
return { renderId, status: "cancelled" };
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Worker (processing side)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
async function processRenderJob(job: Job<RenderJobData, RenderJobResult>) {
|
||||
const { folder, videoSrc, transcription, styleConfig, callbackUrl } =
|
||||
job.data;
|
||||
const logPrefix = `[render-job:${job.id}]`;
|
||||
const renderId = String(job.id);
|
||||
|
||||
const s3Service = new S3Service("captioned", folder);
|
||||
const filename = s3Service.getFileName(videoSrc);
|
||||
const videoUrl = await s3Service.getFileURL(videoSrc);
|
||||
const uploadAbortController = new AbortController();
|
||||
let cancellationPollTimer: ReturnType<typeof setInterval> | null = null;
|
||||
|
||||
console.log(`${logPrefix} Starting job for ${videoSrc}`);
|
||||
|
||||
const ensureNotCancelled = async () => {
|
||||
if (await isRenderCancellationRequested(renderId)) {
|
||||
throw new Error(RENDER_CANCELLED_ERROR_MESSAGE);
|
||||
}
|
||||
};
|
||||
|
||||
await ensureNotCancelled();
|
||||
|
||||
cancellationPollTimer = setInterval(() => {
|
||||
void isRenderCancellationRequested(renderId).then((cancelRequested) => {
|
||||
if (!cancelRequested) {
|
||||
return;
|
||||
}
|
||||
|
||||
uploadAbortController.abort(RENDER_CANCELLED_ERROR_MESSAGE);
|
||||
});
|
||||
}, RENDER_CANCELLATION_POLL_MS);
|
||||
|
||||
// Send RUNNING webhook
|
||||
if (callbackUrl) {
|
||||
const delivered = await sendWebhook(callbackUrl, {
|
||||
status: "RUNNING",
|
||||
progress_pct: RENDER_PROGRESS_MIN_PCT,
|
||||
current_message: "Рендер субтитров",
|
||||
started_at: new Date().toISOString(),
|
||||
});
|
||||
console.log(`${logPrefix} Start webhook delivered: ${delivered}`);
|
||||
}
|
||||
|
||||
await job.updateProgress(RENDER_PROGRESS_MIN_PCT);
|
||||
|
||||
let lastProgressSend = 0;
|
||||
let lastUploadProgressSend = 0;
|
||||
let lastUploadPct = UPLOAD_PROGRESS_START_PCT;
|
||||
|
||||
const pushRunningProgress = (progressPct: number, currentMessage: string) => {
|
||||
if (callbackUrl) {
|
||||
void sendWebhook(callbackUrl, {
|
||||
status: "RUNNING",
|
||||
progress_pct: progressPct,
|
||||
current_message: currentMessage,
|
||||
});
|
||||
}
|
||||
|
||||
void job.updateProgress(progressPct).catch((error) => {
|
||||
console.error(`${logPrefix} Failed to update BullMQ progress:`, error);
|
||||
});
|
||||
};
|
||||
|
||||
const onProgress = (progress: RenderProgress) => {
|
||||
if (progress.totalFrames <= 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
if (now - lastProgressSend >= PROGRESS_THROTTLE_MS) {
|
||||
lastProgressSend = now;
|
||||
const pct = Math.min(
|
||||
Math.round(
|
||||
(progress.renderedFrames / progress.totalFrames) *
|
||||
(RENDER_PROGRESS_MAX_PCT - RENDER_PROGRESS_MIN_PCT),
|
||||
) + RENDER_PROGRESS_MIN_PCT,
|
||||
RENDER_PROGRESS_MAX_PCT,
|
||||
);
|
||||
pushRunningProgress(
|
||||
pct,
|
||||
`Рендер субтитров: ${progress.renderedFrames}/${progress.totalFrames}`,
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
const onUploadProgress = (progress: UploadProgress) => {
|
||||
const now = Date.now();
|
||||
const uploadPct = Math.min(
|
||||
UPLOAD_PROGRESS_START_PCT +
|
||||
Math.round(
|
||||
progress.percent *
|
||||
(UPLOAD_PROGRESS_END_PCT - UPLOAD_PROGRESS_START_PCT),
|
||||
),
|
||||
UPLOAD_PROGRESS_END_PCT,
|
||||
);
|
||||
const shouldSend =
|
||||
uploadPct > lastUploadPct ||
|
||||
now - lastUploadProgressSend >= UPLOAD_PROGRESS_THROTTLE_MS;
|
||||
|
||||
if (!shouldSend) {
|
||||
return;
|
||||
}
|
||||
|
||||
lastUploadProgressSend = now;
|
||||
lastUploadPct = uploadPct;
|
||||
|
||||
pushRunningProgress(
|
||||
uploadPct,
|
||||
`Загрузка видео: ${Math.round(progress.percent * 100)}% (${formatBytes(progress.uploadedBytes)} / ${formatBytes(progress.totalBytes)})`,
|
||||
);
|
||||
};
|
||||
|
||||
let outputLocalPath: string | undefined;
|
||||
try {
|
||||
const result = await renderCaptionedVideo(
|
||||
transcription,
|
||||
videoUrl,
|
||||
filename,
|
||||
styleConfig,
|
||||
onProgress,
|
||||
() => isRenderCancellationRequested(renderId),
|
||||
);
|
||||
outputLocalPath = result.output;
|
||||
console.log(`${logPrefix} Render finished, starting upload`);
|
||||
|
||||
await ensureNotCancelled();
|
||||
|
||||
pushRunningProgress(
|
||||
UPLOAD_PROGRESS_START_PCT,
|
||||
"Рендер завершён, начинаем загрузку видео",
|
||||
);
|
||||
|
||||
const s3OutPath = await s3Service.uploadFile(
|
||||
result.output,
|
||||
filename,
|
||||
onUploadProgress,
|
||||
uploadAbortController.signal,
|
||||
);
|
||||
console.log(`${logPrefix} Upload completed: ${s3OutPath}`);
|
||||
|
||||
pushRunningProgress(
|
||||
UPLOAD_PROGRESS_END_PCT,
|
||||
"Загрузка завершена, публикуем результат",
|
||||
);
|
||||
|
||||
// Send DONE webhook
|
||||
let callbackDelivered = false;
|
||||
if (callbackUrl) {
|
||||
callbackDelivered = await sendWebhook(callbackUrl, {
|
||||
status: "DONE",
|
||||
progress_pct: 100,
|
||||
current_message: "Готово",
|
||||
output_data: { output_path: s3OutPath },
|
||||
finished_at: new Date().toISOString(),
|
||||
});
|
||||
console.log(`${logPrefix} Done webhook delivered: ${callbackDelivered}`);
|
||||
}
|
||||
|
||||
return { outputPath: s3OutPath, callbackDelivered };
|
||||
} catch (err) {
|
||||
if (
|
||||
err instanceof Error &&
|
||||
err.message === RENDER_CANCELLED_ERROR_MESSAGE
|
||||
) {
|
||||
console.log(`${logPrefix} Cancellation requested`);
|
||||
throw err;
|
||||
}
|
||||
|
||||
console.error(`${logPrefix} Job failed:`, err);
|
||||
// Send FAILED webhook
|
||||
if (callbackUrl) {
|
||||
await sendWebhook(callbackUrl, {
|
||||
status: "FAILED",
|
||||
progress_pct: 0,
|
||||
current_message: "Ошибка рендера",
|
||||
error_message: err instanceof Error ? err.message : String(err),
|
||||
finished_at: new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
throw err;
|
||||
} finally {
|
||||
if (cancellationPollTimer) {
|
||||
clearInterval(cancellationPollTimer);
|
||||
}
|
||||
await clearRenderCancellation(renderId);
|
||||
|
||||
if (outputLocalPath) {
|
||||
await Bun.file(outputLocalPath).delete().catch(() => {});
|
||||
console.log(`${logPrefix} Cleaned up local output ${outputLocalPath}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function startRenderWorker(): Worker<RenderJobData, RenderJobResult> {
|
||||
const worker = new Worker<RenderJobData, RenderJobResult>(
|
||||
QUEUE_NAME,
|
||||
processRenderJob,
|
||||
{
|
||||
connection: redisConnection,
|
||||
concurrency: serverConfig.MAX_CONCURRENT_RENDERS,
|
||||
},
|
||||
);
|
||||
|
||||
worker.on("failed", (job, err) => {
|
||||
console.error(`Render job ${job?.id} failed:`, err.message);
|
||||
});
|
||||
|
||||
worker.on("completed", (job) => {
|
||||
console.log(`Render job ${job.id} completed`);
|
||||
});
|
||||
|
||||
return worker;
|
||||
}
|
||||
Reference in New Issue
Block a user