officerdev/platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

replace hardcoded download job with download-media script capability

Browse Source 
move video/audio downloads off the dedicated ReClip download lane onto the
generic script-job path:

- delete execute-download.ts, reclip-client.ts and the POST /jobs/download
  endpoint; drop the 'download' mode from the pipeline_jobs enum (legacy rows
  tolerated)
- execute-script.ts: strip the @@officer:progress@@ sentinel from the log,
  emit progress events, and isolate viewer/log writes (safeEmit/safeLog) so a
  broadcast or log throw can't wedge the stdout pump
- pipeline-job-manager.ts: persist latest progress; guard sendToViewer sends
- ScriptJobDetail: render the two progress bars; DownloadJobDetail kept for
  legacy history rows
- TaskRunnerModal: ScriptRunner descends into the triggered directory
- VideoDownloadPanel: rewire startJob to the script-job path

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
pastilhas
2026-07-29 01:10:32 +00:00
co-authored by Claude Opus 4.8
parent b523c7d408
commit b2fb6f148c
10 changed files with 212 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/servers/api/tasks/execute-download.ts
-163
View File
@@ -1,163 +0,0 @@
import { writeFile } from 'node:fs/promises';
import { join } from 'node:path';
import { reclipInfo, reclipPlaylist, reclipDownloadOne, type ReclipInfo } from '../../reclip-client';
// Write a per-item description sidecar next to its media file — same base name, `.txt` (e.g.
// "Song Name.mp3" → "Song Name.txt"). Best-effort; skipped when the description is empty.
async function writeDescriptionSidecar(dir: string, mediaFilename: string, description: string): Promise<void> {
const desc = description.trim();
if (!desc) return;
const base = mediaFilename.replace(/\.[^./\\]+$/, ''); // strip the media extension only
try {
await writeFile(join(dir, `${base}.txt`), desc, 'utf8');
} catch {
/* best-effort — a missing sidecar shouldn't fail the download */
}
}
const FS_ILLEGAL = /[/\\:*?"<>|\u0000-\u001f]/g;
const safeName = (s: string) => s.replace(FS_ILLEGAL, '_').replace(/\s+/g, ' ').trim().slice(0, 180);
// On a FAILED download there's no media file to pair with — still drop a reference so the missed video is
// recoverable: a `<title>.txt` (or `<videoId>.txt` when untitled) holding the URL + the description. Always
// written (even with an empty description — the URL is the reference).
async function writeMissedSidecar(
dir: string,
item: { url: string; title: string; description: string },
): Promise<void> {
const vid = item.url.match(/[?&]v=([\w-]+)/)?.[1];
const name = safeName(item.title) || vid || 'missed';
const content = [item.url, item.description.trim()].filter(Boolean).join('\n\n');
try {
await writeFile(join(dir, `${name}.txt`), content, 'utf8');
} catch {
/* best-effort */
}
}
// The download-job executor — pure scripting, no agent. Two phases:
// 1. metadata — fetch each item's info (title + validity); keep the ones that resolve, skip the errors
// (private / deleted / unavailable). The title is required: ReClip names the output file
// from it (no title → a hash filename), so we cannot skip this pass.
// 2. download — download every survivor in the chosen format (audio/video), passing its title; skip
// anything that fails.
// The item list comes from the caller (inputs.urls — the exact list the panel expanded, so a Mix/radio
// playlist that returns a different set each call can't drift); falls back to expanding inputs.url. 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; per-item errors are counted + skipped.
type Counts = { done: number; failed: number; total: number };
export type DownloadProgress = {
phase: 'expanding' | 'metadata' | 'download' | 'done';
meta: Counts; // done = kept (title fetched), failed = skipped
dl: Counts; // done = saved, failed = failed
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<string, string>; // { urls (JSON) | 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<void> {
const { url, urls: urlsJson, format, absDir } = params.inputs;
if (!absDir) throw new Error('download job missing 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);
// ── Resolve the item list: the caller's exact list, else expand the url server-side ──
let urls: string[];
if (urlsJson) {
urls = JSON.parse(urlsJson) as string[];
} else if (url) {
if (url.includes('list=')) {
const pl = await reclipPlaylist(url);
if (pl.error) throw new Error(pl.error);
urls = pl.urls?.length ? pl.urls : [url];
} else {
urls = [url];
}
} else {
throw new Error('download job missing url(s)');
}
// ── Phase 1: metadata (title + validity) ──
progress.phase = 'metadata';
progress.meta.total = urls.length;
emit(true);
const valid: Array<{ url: string; title: string; description: 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 ?? '', description: info.description ?? '' });
progress.meta.done++;
} else {
progress.meta.failed++;
}
emit();
}
emit(true);
// ── Phase 2: download survivors (title → real filename) ──
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,
// Filename is known before the file transfers — write the description sidecar alongside it, in
// parallel with the download stream. Same base name guarantees they pair up.
onFilename: (filename) => void writeDescriptionSidecar(absDir, filename, item.description),
});
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++;
void writeMissedSidecar(absDir, item); // no media file — leave a reference to what we missed
}
emit();
}
progress.phase = 'done';
progress.current = undefined;
emit(true);
}
src/servers/api/tasks/execute-script.ts
+69 -7
View File
@@ -12,9 +12,15 @@ export type ScriptEvent =
| { type: 'started'; taskName: string }
| { type: 'stdout'; data: string }
| { type: 'stderr'; data: string }
| { type: 'progress'; progress: unknown }
| { type: 'exit'; code: number }
| { type: 'error'; message: string };
// A script can emit machine-readable progress on stdout: a line `@@officer:progress@@ {json}`. Those
// lines are plucked out (parsed into a `progress` event, kept OUT of the human log); everything else
// streams verbatim.
const PROGRESS_SENTINEL = '@@officer:progress@@';
export type ExecuteScriptParams = {
jobId: string;
userId: number;
@@ -92,6 +98,23 @@ export async function executeScript(params: ExecuteScriptParams): Promise<{ exit
mkdirSync(join(DATA_PATH, 'jobs'), { recursive: true });
const log = createWriteStream(jobLogPath(jobId), { flags: 'w' });
// The pumps must keep draining the child's stdout no matter what: a throw from a viewer broadcast or
// a log-stream error must not stop consumption, or the child's stdout pipe backs up and the whole job
// silently wedges (log + progress freeze while it keeps running). So both side-effects are isolated.
const safeEmit = (event: ScriptEvent) => {
try {
emit(event);
} catch {
/* a dead viewer / broadcast error must never kill the pump */
}
};
const safeLog = (data: string) => {
try {
log.write(data);
} catch {
/* a log-stream error must never kill the pump */
}
};
const cleanup = () => {
try {
rmSync(join(scriptPath, '..'), { recursive: true, force: true });
@@ -116,24 +139,63 @@ export async function executeScript(params: ExecuteScriptParams): Promise<{ exit
}
}, 500);
const decoder = new TextDecoder();
const pump = async (reader: ReadableStreamDefaultReader<Uint8Array>, type: 'stdout' | 'stderr') => {
// stderr: raw passthrough. Each pump owns its decoder (stateful across chunks; must not be shared).
const pumpRaw = async (reader: ReadableStreamDefaultReader<Uint8Array>, type: 'stdout' | 'stderr') => {
const decoder = new TextDecoder();
try {
for (;;) {
const { done, value } = await reader.read();
if (done) break;
const data = decoder.decode(value);
emit({ type, data });
log.write(data);
const data = decoder.decode(value, { stream: true });
if (!data) continue;
safeEmit({ type, data });
safeLog(data);
}
} catch {
// stream closed
}
};
// stdout: line-buffered so progress-sentinel lines can be extracted from the human log + stream.
const pumpStdout = async (reader: ReadableStreamDefaultReader<Uint8Array>) => {
const decoder = new TextDecoder();
let buf = '';
const handleLine = (line: string, hadNewline: boolean) => {
if (line.startsWith(PROGRESS_SENTINEL)) {
let parsed: unknown;
try {
parsed = JSON.parse(line.slice(PROGRESS_SENTINEL.length).trim());
} catch {
return; // malformed progress payload — drop it, don't leak the sentinel into the log
}
safeEmit({ type: 'progress', progress: parsed });
return;
}
const out = hadNewline ? `${line}\n` : line;
if (!out) return;
safeEmit({ type: 'stdout', data: out });
safeLog(out);
};
try {
for (;;) {
const { done, value } = await reader.read();
if (done) break;
buf += decoder.decode(value, { stream: true });
let nl: number;
while ((nl = buf.indexOf('\n')) !== -1) {
handleLine(buf.slice(0, nl), true);
buf = buf.slice(nl + 1);
}
}
} catch {
// stream closed
}
if (buf) handleLine(buf, false); // trailing partial line (no newline)
};
const [, , exitCode] = await Promise.all([
pump(proc.stdout.getReader(), 'stdout'),
pump(proc.stderr.getReader(), 'stderr'),
pumpStdout(proc.stdout.getReader()),
pumpRaw(proc.stderr.getReader(), 'stderr'),
proc.exited,
]);
src/servers/api/tasks/pipeline-job-manager.ts
+52 -77
View File
@@ -18,16 +18,10 @@ 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, 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');
// Everything a job can stream — pipeline structural events, script stdout/stderr/exit/progress.
type JobEvent = OutMessage | ScriptEvent;
type JobMode = 'pipeline' | 'script' | 'agentic';
type WSData = {
userId: number;
@@ -38,7 +32,6 @@ type WSData = {
type LiveJob = {
jobId: string;
userId: number;
lane: Lane;
abortSignal: { aborted: boolean };
emitter: EventEmitter;
eventBuffer: JobEvent[];
@@ -53,15 +46,15 @@ const PROGRESS_FLUSH_MS = 3000;
const liveJobs = new Map<string, LiveJob>();
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<WSData>, jobId: string, event: JobEvent) {
// readyState can flip to closing between this check and send (e.g. a page refresh), so send() may
// still throw. A dead viewer must never propagate into the job's output pump, so swallow it here.
if (ws.readyState === 1) {
ws.send(JSON.stringify({ jobId, ...event }));
try {
ws.send(JSON.stringify({ jobId, ...event }));
} catch {
/* viewer went away mid-send */
}
}
}
@@ -107,8 +100,8 @@ export async function enqueueJob(
): Promise<{ jobId: string; status: 'running' | 'pending' }> {
const jobId = randomUUID();
const mode: JobMode = params.mode ?? 'pipeline';
// Queue within the job's lane: run now if forced, or if that lane is idle.
const run = action === 'start' || runningInLane(laneOf(mode)) === 0;
// One global queue (single user): run now if forced, or if nothing is currently running.
const run = action === 'start' || liveJobs.size === 0;
await createPipelineJob({
id: jobId,
userId: params.userId,
@@ -136,7 +129,6 @@ 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: [],
@@ -156,7 +148,7 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) {
job.progressDirty = true;
job.lastProgress = extractProgress(event, job.lastProgress);
}
if (event.type === 'download:progress') {
if (event.type === 'progress') {
job.progressDirty = true;
job.lastProgress = event.progress;
}
@@ -195,41 +187,30 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) {
// Run the job in the background — dispatch by mode.
const runner: Promise<void | { exitCode: number }> =
mode === 'download'
? executeDownload({
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,
})
: 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,
});
: 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) => {
@@ -265,39 +246,33 @@ function launch(jobId: string, mode: JobMode, params: StartJobParams) {
});
}
// 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.
// When a job finishes (or on startup), launch the oldest queued job — one global queue, one job at a
// time (single user). Rebuilds the user context from userId since there's no request here.
async function promoteNext(): Promise<void> {
if (liveJobs.size > 0) return; // something is already running
const pending = await getPendingJobs(); // oldest first
const launchedNow = new Set<string>();
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<string, string>,
cwd: next.cwd ?? undefined,
config: next.config,
});
break; // lane filled
for (const next of pending) {
const nextMode = (next.mode as JobMode) ?? 'pipeline';
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
}
await updatePipelineJob(next.id, { status: 'running', startedAt: new Date() });
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<string, string>,
cwd: next.cwd ?? undefined,
config: next.config,
});
return; // one at a time
}
}
src/servers/api/tasks/pipeline-jobs-routes.ts
-39
View File
@@ -5,7 +5,6 @@ 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();
@@ -71,44 +70,6 @@ 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: { urls[] (the exact list to download — preferred), OR url (expanded server-side),
// format:'audio'|'video', dir (target folder, home-relative), root?, label? }. Passing `urls` locks the
// job to the list the panel already expanded, so a Mix/radio playlist can't drift between fetch + job.
pipelineJobsRouter.post('/download', async (c) => {
const user = c.get('user');
const body = await c.req.json<{
url?: string;
urls?: string[];
format?: 'audio' | 'video';
dir?: string;
root?: string;
label?: string;
}>();
if (!body.url && !body.urls?.length) throw errors.BAD_REQUEST('url or urls is required');
const rootDir = getRootDir(user, body.root);
const absDir = resolveUserPath(rootDir, body.dir ?? '/'); // traversal-guarded
const format = body.format === 'video' ? 'video' : 'audio';
const inputs: Record<string, string> = body.urls?.length
? { urls: JSON.stringify(body.urls), format, absDir }
: { url: body.url!, format, absDir };
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,
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) => {