From 4e78986e39c969f38e3864bb0159884a234de570 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Tue, 28 Jul 2026 18:09:37 +0000 Subject: [PATCH] jobs: video/audio download as a two-phase job (backend) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Turns the downloader into a server-side job on the existing jobs spine (Postgres persistence, live WS viewers + replay, abort, /jobs UI) — but with its own executor and its own lane, since it's deterministic scripting, not an agent, and a multi-hour playlist mustn't block agentic jobs. - reclip-client.ts (new, shared): reclipInfo / reclipPlaylist / reclipDownloadOne (single download → streams the file to a dir, abort-aware). Extracted so both the file-browser endpoints and the job executor use one client. - execute-download.ts (new): the two-phase executor — phase 1 metadata (expand playlist, fetch each info, keep survivors, skip errors), phase 2 download (each survivor in the chosen format; skip download errors). Emits a compact `download:progress` snapshot (counters, not per-item events — playlists are thousands of items). Throws on abort / fatal. - job manager: `download` mode dispatch → executeDownload; persists download:progress; adds execution LANES (download vs default) so the two run independently and each serializes on its own; promoteNext fills both lanes. - POST /api/tasks/jobs/download { url, format, dir, root?, label? } — enqueues a download job (own lane, no capability task needed; traversal-guarded target). - schema: `download` added to the mode enum (drizzle text-enum — no DB migration); getPendingJobs() query for lane filling. Verified the executor with a mocked ReClip client: two-phase filtering, skip-on- error counts, and abort-throws all correct. Co-Authored-By: Claude Opus 4.8 --- src/databases/officer_db/src/index.ts | 1 + .../officer_db/src/queries/pipeline-jobs.ts | 5 + .../officer_db/src/schema/pipeline-jobs.ts | 57 ++++---- src/servers/api/file-browser/router.ts | 4 +- src/servers/api/tasks/execute-download.ts | 110 +++++++++++++++ src/servers/api/tasks/pipeline-job-manager.ts | 126 ++++++++++++------ src/servers/api/tasks/pipeline-jobs-routes.ts | 34 +++++ src/servers/reclip-client.ts | 119 +++++++++++++++++ 8 files changed, 387 insertions(+), 69 deletions(-) create mode 100644 src/servers/api/tasks/execute-download.ts create mode 100644 src/servers/reclip-client.ts diff --git a/src/databases/officer_db/src/index.ts b/src/databases/officer_db/src/index.ts index 4c79f9af..1e3b8e97 100644 --- a/src/databases/officer_db/src/index.ts +++ b/src/databases/officer_db/src/index.ts @@ -70,6 +70,7 @@ export { updatePipelineJob, getPipelineJobsForUser, getOldestPendingJob, + getPendingJobs, countPendingJobs, deletePipelineJob, deleteTerminalJobsForUser, diff --git a/src/databases/officer_db/src/queries/pipeline-jobs.ts b/src/databases/officer_db/src/queries/pipeline-jobs.ts index 91a9a36b..c7ee2b30 100644 --- a/src/databases/officer_db/src/queries/pipeline-jobs.ts +++ b/src/databases/officer_db/src/queries/pipeline-jobs.ts @@ -62,6 +62,11 @@ export async function getOldestPendingJob() { return rows[0] ?? null; } +// All queued jobs, oldest first — used to fill each execution lane (download vs agentic) independently. +export async function getPendingJobs() { + return db.select().from(pipelineJobs).where(eq(pipelineJobs.status, 'pending')).orderBy(asc(pipelineJobs.createdAt)); +} + // On restart, only RUNNING jobs are orphaned (their process died) → interrupted. PENDING jobs are the // queue backlog and must survive to be promoted after startup. export async function markInterruptedJobs() { diff --git a/src/databases/officer_db/src/schema/pipeline-jobs.ts b/src/databases/officer_db/src/schema/pipeline-jobs.ts index a26a2430..1999a032 100644 --- a/src/databases/officer_db/src/schema/pipeline-jobs.ts +++ b/src/databases/officer_db/src/schema/pipeline-jobs.ts @@ -1,26 +1,37 @@ import { pgTable, text, integer, timestamp, jsonb, index } from 'drizzle-orm/pg-core'; import { users } from './auth'; -export const pipelineJobs = pgTable('pipeline_jobs', { - id: text('id').primaryKey(), - userId: integer('user_id').notNull().references(() => users.id, { onDelete: 'cascade' }), - taskDirName: text('task_dir_name').notNull(), - taskName: text('task_name').notNull(), - // Job kind — pipeline (multi-step agentic), script (single bash/py/ts task), later agentic. - mode: text('mode', { enum: ['pipeline', 'script', 'agentic'] }).notNull().default('pipeline'), - status: text('status', { enum: ['pending', 'running', 'completed', 'failed', 'stopped', 'interrupted'] }).notNull().default('pending'), - inputs: jsonb('inputs').notNull().default({}), - cwd: text('cwd'), - config: jsonb('config').notNull(), - progress: jsonb('progress'), - totalCost: jsonb('total_cost'), - error: text('error'), - // Script jobs: the process exit code (null until finished / for non-script jobs). - exitCode: integer('exit_code'), - createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(), - startedAt: timestamp('started_at', { withTimezone: true }), - completedAt: timestamp('completed_at', { withTimezone: true }), -}, (table) => [ - index('idx_pipeline_jobs_user_created').on(table.userId, table.createdAt), - index('idx_pipeline_jobs_status').on(table.status), -]); +export const pipelineJobs = pgTable( + 'pipeline_jobs', + { + id: text('id').primaryKey(), + userId: integer('user_id') + .notNull() + .references(() => users.id, { onDelete: 'cascade' }), + taskDirName: text('task_dir_name').notNull(), + taskName: text('task_name').notNull(), + // Job kind — pipeline (multi-step agentic), script (single bash/py/ts task), download (video/audio via + // ReClip — its own lane so a long download doesn't block agentic jobs), later agentic. + mode: text('mode', { enum: ['pipeline', 'script', 'agentic', 'download'] }) + .notNull() + .default('pipeline'), + status: text('status', { enum: ['pending', 'running', 'completed', 'failed', 'stopped', 'interrupted'] }) + .notNull() + .default('pending'), + inputs: jsonb('inputs').notNull().default({}), + cwd: text('cwd'), + config: jsonb('config').notNull(), + progress: jsonb('progress'), + totalCost: jsonb('total_cost'), + error: text('error'), + // Script jobs: the process exit code (null until finished / for non-script jobs). + exitCode: integer('exit_code'), + createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(), + startedAt: timestamp('started_at', { withTimezone: true }), + completedAt: timestamp('completed_at', { withTimezone: true }), + }, + (table) => [ + index('idx_pipeline_jobs_user_created').on(table.userId, table.createdAt), + index('idx_pipeline_jobs_status').on(table.status), + ], +); diff --git a/src/servers/api/file-browser/router.ts b/src/servers/api/file-browser/router.ts index 9ef3b1aa..c54b4778 100644 --- a/src/servers/api/file-browser/router.ts +++ b/src/servers/api/file-browser/router.ts @@ -43,7 +43,7 @@ function getUserDataDir(email: string): string { return join(DATA_PATH, email); } -function getRootDir(user: UserCtx, root?: string): string { +export function getRootDir(user: UserCtx, root?: string): string { if (!root || root === 'home') return getOwnerHomeDir(user.email); if (root === 'user-data') return getUserDataDir(user.email); throw errors.BAD_REQUEST(`Invalid root: ${root}`); @@ -54,7 +54,7 @@ function getRootDir(user: UserCtx, root?: string): string { // passes a `/home/br` check — which is how `..` segments escaped. const isInside = (root: string, target: string): boolean => target === root || target.startsWith(root + sep); -function resolveUserPath(rootDir: string, relPath: string): string { +export function resolveUserPath(rootDir: string, relPath: string): string { const resolved = resolve(rootDir, relPath.replace(/^\/+/, '')); if (!isInside(rootDir, resolved)) throw errors.FORBIDDEN('Path outside root directory'); return resolved; diff --git a/src/servers/api/tasks/execute-download.ts b/src/servers/api/tasks/execute-download.ts new file mode 100644 index 00000000..3deb8ff8 --- /dev/null +++ b/src/servers/api/tasks/execute-download.ts @@ -0,0 +1,110 @@ +import { reclipInfo, reclipPlaylist, reclipDownloadOne, type ReclipInfo } from '../../reclip-client'; + +// The download-job executor — pure scripting, no agent. Two phases: +// 1. metadata — expand the playlist, fetch each item's info sequentially, keep the ones that resolve +// (skip the errors: private/deleted/unavailable). +// 2. download — download every survivor in the chosen format (audio/video); skip anything that fails. +// Emits a compact `download:progress` snapshot (counters, not per-item events — a playlist can be +// thousands of items). Throws on abort or a fatal error (playlist expansion) so the job manager marks it +// stopped/failed; per-item errors are counted and skipped, never fatal. + +type Counts = { done: number; failed: number; total: number }; +export type DownloadProgress = { + phase: 'expanding' | 'metadata' | 'download' | 'done'; + meta: Counts; + dl: Counts; + current?: string; // title of the item currently downloading +}; + +export type DownloadEvent = { type: 'download:progress'; progress: DownloadProgress }; + +export type ExecuteDownloadParams = { + jobId: string; + userId: number; + email: string; + username?: string | null; + inputs: Record; // { url, format: 'audio'|'video', absDir } + cwd?: string; + abortSignal: { aborted: boolean }; + emit: (event: DownloadEvent) => void; +}; + +const EMIT_THROTTLE_MS = 750; + +export async function executeDownload(params: ExecuteDownloadParams): Promise { + const { url, format, absDir } = params.inputs; + if (!url || !absDir) throw new Error('download job missing url or target directory'); + const audioOnly = format !== 'video'; // default to audio + + const progress: DownloadProgress = { + phase: 'expanding', + meta: { done: 0, failed: 0, total: 0 }, + dl: { done: 0, failed: 0, total: 0 }, + }; + let lastEmit = 0; + const emit = (force = false) => { + const now = Date.now(); + if (!force && now - lastEmit < EMIT_THROTTLE_MS) return; + lastEmit = now; + params.emit({ + type: 'download:progress', + progress: { ...progress, meta: { ...progress.meta }, dl: { ...progress.dl } }, + }); + }; + const checkAbort = () => { + if (params.abortSignal.aborted) throw new Error('aborted'); + }; + + emit(true); + + // ── Expand ── + let urls = [url]; + if (url.includes('list=')) { + const pl = await reclipPlaylist(url); + if (pl.error) throw new Error(pl.error); + if (pl.urls?.length) urls = pl.urls; + } + + // ── Phase 1: metadata (keep survivors) ── + progress.phase = 'metadata'; + progress.meta.total = urls.length; + emit(true); + const valid: Array<{ url: string; title: string }> = []; + for (const u of urls) { + checkAbort(); + const info = await reclipInfo(u).catch((): ReclipInfo => ({ error: 'fetch failed' })); + if (info && !info.error) valid.push({ url: u, title: info.title ?? '' }); + else progress.meta.failed++; + progress.meta.done++; + emit(); + } + emit(true); + + // ── Phase 2: download survivors ── + progress.phase = 'download'; + progress.dl.total = valid.length; + emit(true); + for (const item of valid) { + checkAbort(); + progress.current = item.title || item.url; + emit(true); + try { + await reclipDownloadOne({ + url: item.url, + destDir: absDir, + audioOnly, + title: item.title, + signal: params.abortSignal, + }); + progress.dl.done++; + } catch { + checkAbort(); // an abort surfaces as a throw here — re-check so it stops instead of counting as a skip + progress.dl.failed++; + } + emit(); + } + + progress.phase = 'done'; + progress.current = undefined; + emit(true); +} diff --git a/src/servers/api/tasks/pipeline-job-manager.ts b/src/servers/api/tasks/pipeline-job-manager.ts index 57991cff..c3097b37 100644 --- a/src/servers/api/tasks/pipeline-job-manager.ts +++ b/src/servers/api/tasks/pipeline-job-manager.ts @@ -6,7 +6,7 @@ import { getPipelineJob, updatePipelineJob, getPipelineJobsForUser, - getOldestPendingJob, + getPendingJobs, countPendingJobs, deletePipelineJob, deleteTerminalJobsForUser, @@ -18,10 +18,16 @@ import { toShellUsername } from '../../data-path'; import { executePipeline } from './pipeline-executor'; import type { OutMessage } from './pipeline-executor'; import { executeScript, jobLogPath, type ScriptEvent } from './execute-script'; +import { executeDownload, type DownloadEvent } from './execute-download'; -// Everything a job can stream — pipeline structural events plus script stdout/stderr/exit. -type JobEvent = OutMessage | ScriptEvent; -type JobMode = 'pipeline' | 'script' | 'agentic'; +// Everything a job can stream — pipeline structural events, script stdout/stderr/exit, download progress. +type JobEvent = OutMessage | ScriptEvent | DownloadEvent; +type JobMode = 'pipeline' | 'script' | 'agentic' | 'download'; + +// Execution lanes: downloads run independently of agentic/script jobs so a multi-hour playlist doesn't +// block capability jobs. Each lane runs one job at a time (single user). +type Lane = 'download' | 'default'; +const laneOf = (mode: JobMode): Lane => (mode === 'download' ? 'download' : 'default'); type WSData = { userId: number; @@ -32,6 +38,7 @@ type WSData = { type LiveJob = { jobId: string; userId: number; + lane: Lane; abortSignal: { aborted: boolean }; emitter: EventEmitter; eventBuffer: JobEvent[]; @@ -46,6 +53,12 @@ const PROGRESS_FLUSH_MS = 3000; const liveJobs = new Map(); +const runningInLane = (lane: Lane): number => { + let n = 0; + for (const job of liveJobs.values()) if (job.lane === lane) n++; + return n; +}; + function sendToViewer(ws: ServerWebSocket, jobId: string, event: JobEvent) { if (ws.readyState === 1) { ws.send(JSON.stringify({ jobId, ...event })); @@ -94,7 +107,8 @@ export async function enqueueJob( ): Promise<{ jobId: string; status: 'running' | 'pending' }> { const jobId = randomUUID(); const mode: JobMode = params.mode ?? 'pipeline'; - const run = action === 'start' || runningCount() === 0; + // Queue within the job's lane: run now if forced, or if that lane is idle. + const run = action === 'start' || runningInLane(laneOf(mode)) === 0; await createPipelineJob({ id: jobId, userId: params.userId, @@ -122,6 +136,7 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) { const job: LiveJob = { jobId, userId: params.userId, + lane: laneOf(mode), abortSignal: { aborted: false }, emitter: new EventEmitter(), eventBuffer: [], @@ -141,6 +156,10 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) { job.progressDirty = true; job.lastProgress = extractProgress(event, job.lastProgress); } + if (event.type === 'download:progress') { + job.progressDirty = true; + job.lastProgress = event.progress; + } if (event.type === 'step:complete' || event.type === 'iteration:complete') { const cost = 'cost' in event ? event.cost : undefined; if (cost) { @@ -176,30 +195,41 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) { // Run the job in the background — dispatch by mode. const runner: Promise = - mode === 'script' - ? executeScript({ + mode === 'download' + ? executeDownload({ jobId, userId: params.userId, email: params.email, username: params.username, - taskDirName: params.taskDirName, inputs: params.inputs, cwd: params.cwd, abortSignal: job.abortSignal, emit, }) - : executePipeline({ - userId: params.userId, - email: params.email, - username: params.username, - taskDirName: params.taskDirName, - inputs: params.inputs, - cwd: params.cwd, - model: params.model, - startAt: params.startAt, - abortSignal: job.abortSignal, - emit, - }); + : mode === 'script' + ? executeScript({ + jobId, + userId: params.userId, + email: params.email, + username: params.username, + taskDirName: params.taskDirName, + inputs: params.inputs, + cwd: params.cwd, + abortSignal: job.abortSignal, + emit, + }) + : executePipeline({ + userId: params.userId, + email: params.email, + username: params.username, + taskDirName: params.taskDirName, + inputs: params.inputs, + cwd: params.cwd, + model: params.model, + startAt: params.startAt, + abortSignal: job.abortSignal, + emit, + }); runner .then(async (result) => { @@ -235,32 +265,40 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) { }); } -// When a job finishes (and nothing else is running), promote the oldest queued job. Also called on -// startup to resume a backlog. Rebuilds the user context from userId since there's no request here. +// When a job finishes (or on startup), fill each idle lane with its oldest queued job. Rebuilds the user +// context from userId since there's no request here. Lanes are independent — a running download doesn't +// hold back an agentic job and vice-versa. async function promoteNext(): Promise { - if (runningCount() > 0) return; - const next = await getOldestPendingJob(); - if (!next) return; - const user = await getUserById(next.userId); - if (!user) { - await updatePipelineJob(next.id, { status: 'failed', error: 'user not found', completedAt: new Date() }).catch( - () => {}, - ); - return promoteNext(); + const pending = await getPendingJobs(); // oldest first + const launchedNow = new Set(); + for (const lane of ['default', 'download'] as Lane[]) { + if (runningInLane(lane) > 0) continue; + for (const next of pending) { + const nextMode = (next.mode as JobMode) ?? 'pipeline'; + if (laneOf(nextMode) !== lane || launchedNow.has(next.id)) continue; + const user = await getUserById(next.userId); + if (!user) { + await updatePipelineJob(next.id, { status: 'failed', error: 'user not found', completedAt: new Date() }).catch( + () => {}, + ); + continue; // skip to the next candidate in this lane + } + await updatePipelineJob(next.id, { status: 'running', startedAt: new Date() }); + launchedNow.add(next.id); + launch(next.id, nextMode, { + userId: next.userId, + email: user.email, + username: toShellUsername(user.username ?? '', user.email), + mode: nextMode, + taskDirName: next.taskDirName, + taskName: next.taskName, + inputs: next.inputs as Record, + cwd: next.cwd ?? undefined, + config: next.config, + }); + break; // lane filled + } } - await updatePipelineJob(next.id, { status: 'running', startedAt: new Date() }); - const nextMode = (next.mode as JobMode) ?? 'pipeline'; - launch(next.id, nextMode, { - userId: next.userId, - email: user.email, - username: toShellUsername(user.username ?? '', user.email), - mode: nextMode, - taskDirName: next.taskDirName, - taskName: next.taskName, - inputs: next.inputs as Record, - cwd: next.cwd ?? undefined, - config: next.config, - }); } export function attachViewer(jobId: string, ws: ServerWebSocket) { diff --git a/src/servers/api/tasks/pipeline-jobs-routes.ts b/src/servers/api/tasks/pipeline-jobs-routes.ts index 7fea9c05..4ad751e6 100644 --- a/src/servers/api/tasks/pipeline-jobs-routes.ts +++ b/src/servers/api/tasks/pipeline-jobs-routes.ts @@ -5,6 +5,7 @@ import { getPipelineJobsForUser, getPipelineJob } from 'officerdb'; import { getTaskByDirName } from './task-files'; import * as jobManager from './pipeline-job-manager'; import { jobLogPath } from './execute-script'; +import { getRootDir, resolveUserPath } from '../file-browser/router'; export const pipelineJobsRouter = createRouter(); @@ -70,6 +71,39 @@ pipelineJobsRouter.post('/', async (c) => { return c.json({ jobId, status }); }); +// POST /download — enqueue a video/audio download job (ReClip). Runs in its own lane, needs no capability +// task. Body: { url (video or playlist), format:'audio'|'video', dir (target folder, home-relative), +// root?, label? }. The job expands + fetches metadata (phase 1) then downloads survivors (phase 2). +pipelineJobsRouter.post('/download', async (c) => { + const user = c.get('user'); + const body = await c.req.json<{ + url: string; + format?: 'audio' | 'video'; + dir?: string; + root?: string; + label?: string; + }>(); + if (!body.url) throw errors.BAD_REQUEST('url is required'); + const rootDir = getRootDir(user, body.root); + const absDir = resolveUserPath(rootDir, body.dir ?? '/'); // traversal-guarded + const format = body.format === 'video' ? 'video' : 'audio'; + const { jobId, status } = await jobManager.enqueueJob( + { + userId: user.id, + email: user.email, + username: user.username ?? '', + mode: 'download', + taskDirName: 'video-download', + taskName: body.label || (format === 'audio' ? 'Audio download' : 'Video download'), + inputs: { url: body.url, format, absDir }, + cwd: body.dir ?? '/', + config: {}, + }, + 'queue', + ); + return c.json({ jobId, status }); +}); + // GET /counts — header-badge summary { running, runningJobId, queued }. Before /:id so it isn't // captured as an id. pipelineJobsRouter.get('/counts', async (c) => { diff --git a/src/servers/reclip-client.ts b/src/servers/reclip-client.ts new file mode 100644 index 00000000..a3e61f75 --- /dev/null +++ b/src/servers/reclip-client.ts @@ -0,0 +1,119 @@ +import { join } from 'node:path'; +import { mkdir } from 'node:fs/promises'; + +// Shared client for the ReClip download service (its own yt-dlp). The platform is a pure proxy: it never +// runs yt-dlp itself. Used by the file-browser download endpoints AND the download-job executor. + +export const RECLIP_BASE = process.env.RECLIP_URL ?? 'http://localhost:8899'; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +export type ReclipInfo = { + title?: string; + thumbnail?: string; + duration?: number; + uploader?: string; + formats?: Array<{ height?: number; id?: string; label?: string }>; + error?: string; +}; + +/** Single-video metadata (title/thumbnail/duration/uploader/formats). Returns { error } inline. */ +export async function reclipInfo(url: string): Promise { + const res = await fetch(`${RECLIP_BASE}/api/info`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ url }), + signal: AbortSignal.timeout(90_000), + }).catch(() => null); + if (!res) return { error: `Could not reach ReClip at ${RECLIP_BASE}` }; + return (await res.json().catch(() => ({}))) as ReclipInfo; +} + +/** Expand a playlist URL into its individual video URLs. */ +export async function reclipPlaylist(url: string): Promise<{ urls?: string[]; error?: string }> { + const res = await fetch(`${RECLIP_BASE}/api/playlist`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ url }), + signal: AbortSignal.timeout(120_000), + }).catch(() => null); + if (!res) return { error: `Could not reach ReClip at ${RECLIP_BASE}` }; + return (await res.json().catch(() => ({}))) as { urls?: string[]; error?: string }; +} + +type DownloadOpts = { + url: string; + destDir: string; // absolute directory to write the finished file into + audioOnly: boolean; + title?: string; // for a title-based filename (ReClip names by job id otherwise) + onPhase?: (phase: 'transferring') => void; + signal?: { aborted: boolean }; +}; + +/** + * Download ONE video/audio via ReClip and stream the finished file into `destDir`. Resolves with the + * saved filename; throws on any failure (unreachable / rejected / job error / timeout / abort). Respects + * a cooperative `signal.aborted` between polls and while streaming. + */ +export async function reclipDownloadOne(opts: DownloadOpts): Promise { + const { url, destDir, audioOnly, signal } = opts; + const aborted = () => signal?.aborted === true; + + let title = opts.title ?? ''; + if (!title) { + const info = await reclipInfo(url).catch(() => null); + title = info?.title ?? ''; + } + + const dlRes = await fetch(`${RECLIP_BASE}/api/download`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ url, format: audioOnly ? 'audio' : 'video', title }), + signal: AbortSignal.timeout(30_000), + }).catch(() => { + throw new Error(`Could not reach ReClip at ${RECLIP_BASE}`); + }); + if (!dlRes.ok) throw new Error('ReClip rejected the download request'); + const reclipJob = ((await dlRes.json()) as { job_id?: string }).job_id; + if (!reclipJob) throw new Error('ReClip did not return a job id'); + + // Poll ReClip until done/error (generous deadline; ReClip enforces its own per-download cap). + const deadline = Date.now() + 60 * 60_000; + let filename = ''; + for (;;) { + if (aborted()) throw new Error('aborted'); + if (Date.now() > deadline) throw new Error('Download timed out'); + await sleep(2000); + const stRes = await fetch(`${RECLIP_BASE}/api/status/${reclipJob}`, { signal: AbortSignal.timeout(15_000) }).catch( + () => null, + ); + if (!stRes?.ok) continue; + const st = (await stRes.json()) as { status: string; error?: string | null; filename?: string | null }; + if (st.status === 'error') throw new Error(st.error || 'ReClip download failed'); + if (st.status === 'done') { + filename = st.filename || `${reclipJob}.${audioOnly ? 'mp3' : 'mp4'}`; + break; + } + } + + opts.onPhase?.('transferring'); + + // Stream the finished file into the destination folder (filename is title-sanitized by ReClip). + const fileRes = await fetch(`${RECLIP_BASE}/api/file/${reclipJob}`, { signal: AbortSignal.timeout(600_000) }); + const body = fileRes.body; + if (!fileRes.ok || !body) throw new Error('Failed to fetch the downloaded file from ReClip'); + await mkdir(destDir, { recursive: true }); + const sink = Bun.file(join(destDir, filename)).writer(); + const reader = body.getReader(); + try { + for (;;) { + if (aborted()) throw new Error('aborted'); + const { done, value } = await reader.read(); + if (done) break; + sink.write(value); + } + } finally { + await sink.end(); + } + return filename; +}