openvibers/openvibe.media Commits
Clip cut on media_jobs (WS-G task 3): a new clip's cut and every re-cut run as the job clip.cut (lane clips, 2 at a time, runs while recording): POST /clips and /:id/recut answer 202 with job_id for the UI to follow and reattach to, media.job.* events, queue-owned retries (2/10/30 min, a window with no footage fails at once), clip.ready/clip.failed and the webhook as before; the old sweeper only retries clips without a job
Changes
@@@ -599,6 +599,7 @@ requeued (or failed when out of attempts). Handlers checkpoint progress and resu| `invariant.scan` | light | The [size-invariant validator](#public-object-size-invariant): records violations, proposes `object.split` / `object.remux`, withdraws moot proposals. Tenant-wide (no object) || `object.split` | heavy | Stream-copies a vod/clip (or a video/audio object) into parts (`params.parts` 2-1000 or `segment_seconds`), each a new **private** object with a `derived_from` relationship (`job_id`, `part`, `start_seconds`, `duration_seconds`). Cuts land on keyframes, so neighbouring parts can overlap slightly. The source is never changed. Checkpointed per part || `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant |++| `clip.cut` | clips | Cut (or re-cut) a clip row from its VOD: `POST /api/v1/:app/clips` and `/:id/recut` answer 202 with `job_id`, which a UI polls and reattaches to (`GET /api/v2/:app/jobs/:id`). `clip.ready`/`clip.failed` and the webhook come from the cut as before. It retries after 2, 10 and 30 min (from attempt 2 a cloud VOD is fetched home first); a window with no footage fails at once. The `clips` lane (`MEDIA_JOBS_CLIPS_CONCURRENCY`, 2) runs while recording || `object.waveform` | heavy | The audio waveform as a PNG (`showwavespic`, 1800×140 by default; `width`, `height`). It is a private `asset` object and the source's `waveform` variant. A remote-only source over `MEDIA_PREVIEW_REMOTE_MAX_MB` (2048) is refused, because it decodes the whole audio track. `no_audio` is permanent || `object.sprite` | heavy | A seek-preview sprite sheet as a JPEG, made from one fast seek per frame: 100 frames by default, at most one per 2 s, 10 columns and 160 px tiles (`frames`, `columns`, `tile_width`). It is a private `asset` object and the source's `sprite` variant. The layout goes in its metadata as `sprite: { count, interval_seconds, columns, rows, tile_width, tile_height }`: frame *i* covers [*i*·interval, (*i*+1)·interval) at column *i* mod columns, row ⌊*i*/columns⌋. `no_video` is permanent || `object.hash` | light | sha256 of ready objects' present local copies that have no hash yet, onto the object (`content_hash`, `metadata.hash_basis`) and the local location (`checksum`). Tenant-wide batch (`params { limit (1-500, 50), budget_mb }`) or one object. Scheduled per tenant every `MEDIA_HASH_INTERVAL_MIN`. Result `{ hashed, bytes, skipped, remaining }`. See [Scheduled verification](#scheduled-verification) |@@@ -71,6 +71,7 @@ const config = {lightConcurrency: Math.max(1, intEnv('MEDIA_JOBS_LIGHT_CONCURRENCY', 2)), // thumbnails, scansheavyConcurrency: Math.max(1, intEnv('MEDIA_JOBS_HEAVY_CONCURRENCY', 1)), // split, remuxfinalizeConcurrency: Math.max(1, intEnv('MEDIA_JOBS_FINALIZE_CONCURRENCY', 1)), // vod.finalize retries++ clipsConcurrency: Math.max(1, intEnv('MEDIA_JOBS_CLIPS_CONCURRENCY', 2)), // clip.cut (runs while recording)// Heavy jobs wait while a recording is running (as the health job does), unless this is on.heavyWhileRecording: ['1', 'true', 'on'].includes(String(process.env.MEDIA_JOBS_HEAVY_WHILE_RECORDING || '').toLowerCase()),leaseS: Math.max(30, intEnv('MEDIA_JOBS_LEASE_S', 120)),@@@ -8,6 +8,7 @@* object.waveform heavy the audio waveform (PNG) of a vod/clip as the source's `waveform` variant* object.sprite heavy a seek-preview sprite sheet (JPEG, layout in its metadata) as the `sprite` variant* vod.finalize finalize finalize a recording whose finalize failed or never ran (orphans), with backoff++ * clip.cut clips cut (or re-cut) a clip row from its VOD; retries with backoff (server/vod/clip-jobs.js)* vod.duration.reconcile heavy stored VOD durations vs a measurement of the real file (local or B2/R2), one batch* object.hash light sha256 of local copies with no content hash yet, one bounded batch (or one object)* storage.orphans.scan light the storage orphan report, service-wide (report only; never deletes)@@@ -23,6 +24,7 @@ queue.register('object.remux', require('./derive').remux);queue.register('object.waveform', require('./previews').waveform);queue.register('object.sprite', require('./previews').sprite);queue.register('vod.finalize', require('./vod-finalize').spec);++queue.register('clip.cut', require('../vod/clip-jobs').spec);queue.register('vod.duration.reconcile', require('./duration-reconcile').spec);queue.register('object.hash', require('./content-hash').spec);queue.register('storage.orphans.scan', require('./storage-orphans').spec);@@@ -7,6 +7,8 @@* recording runs unless MEDIA_JOBS_HEAVY_WHILE_RECORDING=1* finalize vod.finalize MEDIA_JOBS_FINALIZE_CONCURRENCY (1); runs while recording* (it is the work the recorder does when a stream ends)++ * clips clip.cut MEDIA_JOBS_CLIPS_CONCURRENCY (2); runs while recording (people++ * clip live streams and wait for the result)** A running job holds a lease (MEDIA_JOBS_LEASE_S) that a heartbeat renews; the heartbeat also notices* an owner's cancel request. At start every job left `running` by the previous process is retried (or@@@ -41,6 +43,7 @@ function lanes() {{ name: 'light', types: by('light'), max: cfg().lightConcurrency },{ name: 'heavy', types: by('heavy'), max: cfg().heavyConcurrency },{ name: 'finalize', types: by('finalize'), max: cfg().finalizeConcurrency },++ { name: 'clips', types: by('clips'), max: cfg().clipsConcurrency || 2 },];}@@@ -5,6 +5,10 @@* resolves the source, cuts, updates the row, thumbnails,* commits the outcome with its event, fires the webhook. Records cut_error / cut_attempts and* schedules an automatic retry on failure.++ * spec the media job clip.cut (lane clips; roadmap WS-G task 3): a new clip's++ * cut and every re-cut run as a job, so they carry media.job.* events and a++ * job id the UI can reattach to (GET /api/v2/:app/jobs/:id). The queue++ * owns the retries (backoff 2, 10, 30 min; a hopeless source fails at once).* start() sweeper: every 3 min, failed clips that still have* attempts left are re-cut. From the second attempt on, a* VOD that lives in cloud storage is first pulled back to@@@ -48,7 +52,7 @@ async function recutClip(clipId, opts = {}) {return result;}--async function _recutClip(clipId, { reason = 'recut' } = {}) {++async function _recutClip(clipId, { reason = 'recut', viaJob = false } = {}) {const clip = db.getClipById(clipId);if (!clip) return { ok: false, error: 'Clip not found' };if (!clip.vod_id) return { ok: false, error: 'Clip has no source VOD' };@@@ -74,12 +78,12 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {console.log(`[Clips] clip ${clipId}: not enough free disk to pull vod ${vod.id} home — cutting from the cloud copy`);}}-- if (!source) return fail(clipId, clip, attempt, 'VOD media unavailable (not on disk and no cloud copy)');++ if (!source) return fail(clipId, clip, attempt, 'VOD media unavailable (not on disk and no cloud copy)', { viaJob });const startTime = Number(clip.start_time) || 0;const duration = Math.max(1, (Number(clip.end_time) || 0) - startTime);const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });-- if (!cut.ok) return fail(clipId, clip, attempt, cut.error);++ if (!cut.ok) return fail(clipId, clip, attempt, cut.error, { viaJob });// Thumbnail first (the clip.ready payload carries it), then the ready transition, the clip's// object and its event in one transaction (webhooks.announce), then the webhook.try { await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath); } catch { /* */ }@@@ -89,23 +93,24 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {payload: () => _clipPublic(db.getClipById(clipId)),});console.log(`[Clips] Clip ${clipId} ${reason} OK from vod ${clip.vod_id} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s, attempt ${attempt})`);-- return { ok: true };++ return { ok: true, duration_seconds: cut.duration };}--function fail(clipId, clip, attempt, error) {++/** A failed attempt. Under a job the queue schedules the retry, so nothing is left for the sweeper. */++function fail(clipId, clip, attempt, error, { viaJob = false } = {}) {const msg = String(error || 'cut failed').slice(0, 500);// Nothing to retry when the source itself has no footage there (empty recording, window// past the end of what was captured): give up now instead of burning the whole ladder.const hopeless = /No decodable footage|Error opening input: End of file|Source VOD no longer exists|Invalid data found when processing input/i.test(msg);-- const more = attempt < MAX_ATTEMPTS && !hopeless;++ const more = !viaJob && attempt < MAX_ATTEMPTS && !hopeless;const mins = BACKOFF_MIN[Math.min(attempt - 1, BACKOFF_MIN.length - 1)];const nextAt = more ? new Date(Date.now() + mins * 60000).toISOString().replace('T', ' ').slice(0, 19) : null;announce(clip.app_id, 'clip.failed', {change: () => objects().withObject('clip', clipId, () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = ? WHERE id = ?", [msg, nextAt, clipId])),payload: () => _clipPublic(db.getClipById(clipId)),});-- console.warn(`[Clips] Clip ${clipId} attempt ${attempt}/${MAX_ATTEMPTS} failed: ${msg}${more ? ` — retry in ${mins} min` : ' — giving up'}`);-- return { ok: false, error: msg, retry_at: nextAt };++ console.warn(`[Clips] Clip ${clipId} attempt ${attempt}/${MAX_ATTEMPTS} failed: ${msg}${viaJob ? (hopeless ? ' — giving up' : ' — the job retries') : more ? ` — retry in ${mins} min` : ' — giving up'}`);++ return { ok: false, error: msg, retry_at: nextAt, hopeless };}async function sweep() {@@@ -114,7 +119,9 @@ async function sweep() {try {const due = db.all(`SELECT id FROM clips WHERE status = 'failed' AND vod_id IS NOT NULLAND COALESCE(cut_attempts, 0) < ? AND (cut_next_at IS NULL OR cut_next_at <= CURRENT_TIMESTAMP)-- AND created_at >= datetime('now', '-30 days') ORDER BY created_at DESC LIMIT 4`, [MAX_ATTEMPTS]) || [];++ AND created_at >= datetime('now', '-30 days')++ AND NOT EXISTS (SELECT 1 FROM media_jobs j WHERE j.job_type = 'clip.cut' AND j.app_id = clips.app_id AND CAST(json_extract(j.params, '$.clip_id') AS INTEGER) = clips.id)++ ORDER BY created_at DESC LIMIT 4`, [MAX_ATTEMPTS]) || [];for (const row of due) { try { await recutClip(row.id, { reason: 'auto-retry' }); } catch (e) { console.warn(`[Clips] auto-retry ${row.id}:`, e.message); } }} finally { _busy = false; }}@@@ -137,4 +144,41 @@ function start() {console.log('[Clips] cut retry sweeper started (every 3 min, up to 4 attempts, hot-fetch from attempt 2)');}--module.exports = { recutClip, sweep, start, MAX_ATTEMPTS };++// ── clip.cut as a media job ──────────────────────────────────++++/** The clip's media object (legacy:<app>:clip:<id>), the job's object. */++function clipObjectId(appId, clipId) {++ const row = db.get('SELECT id FROM media_objects WHERE legacy_ref = ?', [`legacy:${appId}:clip:${clipId}`]);++ return row ? row.id : null;++}++++/** Queue the cut of a clip row → { job, created }. A repeated request for the same clip reuses its active job. */++function enqueueCut(appId, clipId, { reason = 'cut', createdBy = null, ownerUserId = null } = {}) {++ const queue = require('../jobs/queue');++ const out = queue.enqueue({ appId, type: 'clip.cut', objectId: clipObjectId(appId, clipId), params: { clip_id: Number(clipId), reason }, dedupeActive: true, createdBy, ownerUserId });++ try { require('../jobs/worker').kick(); } catch { /* the poll picks it up */ }++ return out;++}++++const spec = {++ lane: 'clips',++ maxAttempts: MAX_ATTEMPTS,++ timeoutMs: RECUT_CAP_MS + 5 * 60 * 1000,++ needsObject: false,++ backoffS: (attempt) => BACKOFF_MIN[Math.min(Math.max(0, attempt - 1), BACKOFF_MIN.length - 1)] * 60, // after attempt 1: 2 min, then 10, 30++ validate({ params }) {++ const id = Number(params && params.clip_id);++ if (!Number.isInteger(id) || id < 1) throw new (require('../jobs/queue').JobError)('media.job.invalid', 'clip.cut needs params.clip_id', { permanent: true });++ return { clip_id: id, reason: String((params && params.reason) || 'cut').slice(0, 40) };++ },++ async run(job) {++ const { JobError } = require('../jobs/queue');++ const clip = db.getClipById(job.params.clip_id);++ if (!clip || clip.app_id !== job.app_id) throw new JobError('not_found', 'The clip no longer exists', { permanent: true });++ const r = await recutClip(clip.id, { reason: job.params.reason || 'cut', viaJob: true });++ if (!r.ok) throw new JobError('clip_cut_failed', r.error || 'cut failed', { permanent: !!r.hopeless });++ return { clip_id: clip.id, status: 'ready', duration_seconds: r.duration_seconds ?? null };++ },++};++++module.exports = { recutClip, sweep, start, enqueueCut, clipObjectId, spec, MAX_ATTEMPTS };@@@ -20,9 +20,7 @@ const multer = require('multer');const config = require('../config');const db = require('../db/database');const tools = require('./media-tools');--const cutter = require('./clip-cutter');const { tenantAuth, tenantCors } = require('../auth');--const { announce } = require('../webhooks');const objects = require('../objects/model');const router = express.Router({ mergeParams: true });@@@ -157,10 +155,6 @@ router.post('/', tenantAuth({ allowUser: true }), clipUpload.single('video'), asreturn res.status(200).json({ id: duplicate.id, status: duplicate.status, deduplicated: true });}-- if (cutter.isBusy()) {-- return res.status(503).json({ error: 'Server is busy processing other clips. Please try again in a few seconds.' });-- }--// Resolve a source ffmpeg can read: local file, or presigned B2/R2 URL.const vodStorage = require('./vod-storage');const source = await vodStorage.resolveMediaSource(vod);@@@ -213,34 +207,14 @@ router.post('/', tenantAuth({ allowUser: true }), clipUpload.single('video'), asconst clipId = result.lastInsertRowid;const appId = req.appId;-- // Cut in the background; webhook on completion.-- (async () => {-- const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });-- if (cut.ok) {-- // Thumbnail first so the clip.ready payload carries it; then the ready transition,-- // the clip's object and its event commit together (webhooks.announce) and the webhook goes out.-- try {-- await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath);-- } catch { /* */ }-- announce(appId, 'clip.ready', {-- change: () => objects.withObject('clip', clipId, () => db.run('UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = ? WHERE id = ?',-- [cut.filePath, cut.duration, startTime + cut.duration, 'ready', clipId])),-- payload: () => clipPublic(db.getClipById(clipId)),-- });-- console.log(`[Clips] Clip ${clipId} cut from vod ${vodId} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s)`);-- } else {-- // First attempt failed: record why and let the retry sweeper take it from here-- // (attempt 2 pulls a cloud-stored VOD back to local disk first).-- const nextAt = new Date(Date.now() + 2 * 60000).toISOString().replace('T', ' ').slice(0, 19);-- announce(appId, 'clip.failed', {-- change: () => objects.withObject('clip', clipId, () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_attempts = 1, cut_next_at = ? WHERE id = ?", [String(cut.error || 'cut failed').slice(0, 500), nextAt, clipId])),-- payload: () => clipPublic(db.getClipById(clipId)),-- });-- console.warn(`[Clips] Clip ${clipId} failed: ${cut.error} — auto-retry in 2 min`);-- }-- })().catch(err => console.error('[Clips] Background cut error:', err.message));++ // The cut is a media job (clip.cut, WS-G task 3): media.job.* events, retries with backoff, and a++ // job id the caller's UI follows (GET /api/v2/:app/jobs/:id) and reattaches to after a reload.++ // clip.ready / clip.failed and the webhook come from the cut itself, as before.++ const { job } = require('./clip-jobs').enqueueCut(appId, clipId, {++ reason: 'cut', createdBy: req.authType === 'user' ? `app:${appId}:user:${req.userId}` : `app:${appId}`, ownerUserId: userId ?? null,++ });-- res.status(202).json({ id: clipId, status: 'processing' });++ res.status(202).json({ id: clipId, status: 'processing', job_id: job.id });} catch (err) {console.error('[Clips] Create error:', err.message);if (req.file) tools.cleanupTempFile(req.file.path);@@@ -372,8 +346,8 @@ router.post('/:id/recut', tenantAuth(), async (req, res) => {if (!clip.vod_id) return res.status(422).json({ error: 'Clip has no source VOD to re-cut from' });// A manual retry resets the attempt counter so it gets the full ladder again (not projected: no object change).db.run("UPDATE clips SET cut_attempts = 0 WHERE id = ?", [clipId]);-- require('./clip-jobs').recutClip(clipId, { reason: 're-cut' }).catch(err => console.error('[Clips] Background re-cut error:', err.message));-- res.status(202).json({ id: clipId, status: 'processing' });++ const { job } = require('./clip-jobs').enqueueCut(req.appId, clipId, { reason: 're-cut', createdBy: `app:${req.appId}` });++ res.status(202).json({ id: clipId, status: 'processing', job_id: job.id });} catch (err) {console.error('[Clips] Re-cut error:', err.message);res.status(500).json({ error: 'Failed to re-cut clip' });nnew file mode 100644@@@ -0,0 +1,120 @@++'use strict';++// clip.cut as a media job (WS-G task 3): POST /clips answers 202 with the clip and its job id; the job++// cuts the clip (clip.ready as before) with media.job.* events, and GET /api/v2/:app/jobs/:id is what a++// UI polls and reattaches to. A window with no footage fails permanently at once; the old sweeper leaves++// clips that have a job alone; a re-cut is a new job, and asking again while it runs gets the same one.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const { spawnSync } = require('child_process');++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-clipjob-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++process.env.DB_PATH = path.join(tmp, 'media.db');++process.env.VOD_PATH = dir('vods');++process.env.CLIPS_PATH = dir('clips');++process.env.FILES_PATH = dir('files');++process.env.THUMBNAILS_PATH = dir('thumbnails');++process.env.PASTES_PATH = dir('pastes');++process.env.OBJECTS_PATH = dir('objects');++process.env.MEDIA_JOBS_POLL_MS = '50';++process.env.MEDIA_INVARIANT_SCAN_HOURS = '0';++process.env.MEDIA_UPLOAD_MIN_FREE_MB = '0';++const stub = http.createServer((req, res) => {++ req.resume();++ req.on('end', () => {++ res.setHeader('Content-Type', 'application/json');++ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));++ res.statusCode = 404; res.end('{}');++ });++});++const sleep = (ms) => new Promise((r) => setTimeout(r, ms));++const waitFor = async (fn, ms = 30000) => { const end = Date.now() + ms; while (Date.now() < end) { if (await fn()) return true; await sleep(25); } return false; };++++(async () => {++ if (spawnSync('ffmpeg', ['-version']).status !== 0) { console.log('⚠️ ffmpeg not found: clip.cut skipped'); process.exit(0); }++ await new Promise((r) => stub.listen(0, '127.0.0.1', r));++ const base = `http://127.0.0.1:${stub.address().port}`;++ const db = require('../server/db/database');++ const conn = db.getDb();++ db.upsertApp({ app_id: 'live', api_key: 'live-key' });++ const events = require('../server/events');++ events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 60000 });++ const queue = require('../server/jobs/queue');++ const worker = require('../server/jobs/worker');++ const clipJobs = require('../server/vod/clip-jobs');++ const express = require('express');++ const app = express();++ app.use(express.json());++ app.use('/api/v1/:app/clips', require('../server/vod/clips-routes'));++ app.use('/api/v2/:app/jobs', require('../server/jobs/routes'));++ const server = http.createServer(app);++ await new Promise((r) => server.listen(0, '127.0.0.1', r));++ const api = `http://127.0.0.1:${server.address().port}`;++ const call = async (method, p, body) => {++ const res = await fetch(api + p, { method, headers: { authorization: 'Bearer live-key', 'content-type': 'application/json' }, body: body === undefined ? undefined : JSON.stringify(body) });++ return { status: res.status, body: await res.json().catch(() => null) };++ };++ const outbox = () => conn.prepare('SELECT envelope FROM event_outbox ORDER BY id').all().map((r) => JSON.parse(r.envelope));++ const jobEvents = (id) => outbox().filter((e) => e.subject.type === 'job' && e.subject.id === id).map((e) => e.event_type);++ worker.start();++++ const src = path.join(process.env.VOD_PATH, 'rec-30.webm');++ const mk = spawnSync('ffmpeg', ['-v', 'error', '-y', '-f', 'lavfi', '-i', 'testsrc=duration=30:size=320x180:rate=10', '-f', 'lavfi', '-i', 'sine=frequency=440:duration=30',++ '-c:v', 'libvpx', '-g', '10', '-b:v', '200k', '-c:a', 'libopus', '-shortest', src]);++ assert.strictEqual(mk.status, 0, String(mk.stderr));++ conn.prepare("INSERT INTO vods (id, app_id, user_id, title, file_path, file_size, is_public, duration_seconds) VALUES (30, 'live', 5, 'Thirty seconds', ?, ?, 1, 30)").run(src, fs.statSync(src).size);++++ // ── A clip: 202 with its job; the job cuts it ──++ let r = await call('POST', '/api/v1/live/clips', { vod_id: 30, start_s: 5, end_s: 12, title: 'Five to twelve', user_id: 5 });++ assert.strictEqual(r.status, 202, JSON.stringify(r.body));++ assert.ok(/^mjob_/.test(r.body.job_id), 'the answer names the job');++ const clipId = r.body.id, jobId = r.body.job_id;++ assert.ok(await waitFor(() => ['succeeded', 'failed'].includes(queue.get(jobId).status)), 'the cut finishes');++ const j = await call('GET', `/api/v2/live/jobs/${jobId}`);++ assert.strictEqual(j.status, 200);++ assert.deepStrictEqual([j.body.job.type, j.body.job.status, j.body.job.result.clip_id], ['clip.cut', 'succeeded', clipId], JSON.stringify(j.body.job));++ const clip = db.getClipById(clipId);++ assert.strictEqual(clip.status, 'ready');++ assert.ok(fs.existsSync(clip.file_path) && Math.abs(clip.duration_seconds - 7) < 1.5, `a 7 s clip (${clip.duration_seconds})`);++ assert.deepStrictEqual(jobEvents(jobId), ['media.job.queued', 'media.job.started', 'media.job.succeeded']);++ assert.ok(outbox().some((e) => e.event_type === 'media.clip.ready' && e.subject.type === 'clip' && String(e.subject.id) === String(clipId)), 'media.clip.ready as before');++++ // ── A window with no footage: permanent at once, and the sweeper leaves it alone ──++ r = await call('POST', '/api/v1/live/clips', { vod_id: 30, start_s: 28, end_s: 35, title: 'Past the end', user_id: 5 });++ assert.strictEqual(r.status, 202, JSON.stringify(r.body));++ const badJob = r.body.job_id, badClip = r.body.id;++ assert.ok(await waitFor(() => ['succeeded', 'failed'].includes(queue.get(badJob).status)));++ const bad = queue.jobPublic(queue.get(badJob));++ if (bad.status === 'failed') {++ assert.strictEqual(db.getClipById(badClip).status, 'failed');++ assert.strictEqual(db.getClipById(badClip).cut_next_at, null, 'no sweeper retry is scheduled under a job');++ const before = db.getClipById(badClip).cut_attempts;++ await clipJobs.sweep();++ assert.strictEqual(db.getClipById(badClip).cut_attempts, before, 'the sweeper leaves a clip that has a job');++ } else {++ assert.strictEqual(db.getClipById(badClip).status, 'ready', 'ffmpeg cut what there was');++ }++++ // ── A re-cut is a new job; asking again while it is active gets the same one ──++ const r1 = await call('POST', `/api/v1/live/clips/${clipId}/recut`);++ assert.strictEqual(r1.status, 202, JSON.stringify(r1.body));++ assert.ok(r1.body.job_id && r1.body.job_id !== jobId, 'a new job');++ const same = clipJobs.enqueueCut('live', clipId, { reason: 're-cut' });++ assert.ok(same.job.id === r1.body.job_id || queue.get(r1.body.job_id).status === 'succeeded', 'an active re-cut is reused');++ assert.ok(await waitFor(() => queue.get(r1.body.job_id).status === 'succeeded'));++ assert.strictEqual(db.getClipById(clipId).status, 'ready');++++ // ── Another tenant cannot see the job ──++ db.upsertApp({ app_id: 'games', api_key: 'games-key' });++ const other = await fetch(`${api}/api/v2/games/jobs/${jobId}`, { headers: { authorization: 'Bearer games-key' } });++ assert.strictEqual(other.status, 404);++++ worker.stop();++ events._reset();++ server.close();++ stub.close();++ console.log('clip-cut-job: all checks passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -599,6 +599,7 @@ requeued (or failed when out of attempts). Handlers checkpoint progress and resu
@@ -599,6 +599,7 @@ requeued (or failed when out of attempts). Handlers checkpoint progress and resu
| `invariant.scan` | light | The [size-invariant validator](#public-object-size-invariant): records violations, proposes `object.split` / `object.remux`, withdraws moot proposals. Tenant-wide (no object) |
| `invariant.scan` | light | The [size-invariant validator](#public-object-size-invariant): records violations, proposes `object.split` / `object.remux`, withdraws moot proposals. Tenant-wide (no object) |
| `object.split` | heavy | Stream-copies a vod/clip (or a video/audio object) into parts (`params.parts` 2-1000 or `segment_seconds`), each a new **private** object with a `derived_from` relationship (`job_id`, `part`, `start_seconds`, `duration_seconds`). Cuts land on keyframes, so neighbouring parts can overlap slightly. The source is never changed. Checkpointed per part |
| `object.split` | heavy | Stream-copies a vod/clip (or a video/audio object) into parts (`params.parts` 2-1000 or `segment_seconds`), each a new **private** object with a `derived_from` relationship (`job_id`, `part`, `start_seconds`, `duration_seconds`). Cuts land on keyframes, so neighbouring parts can overlap slightly. The source is never changed. Checkpointed per part |
| `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant |
| `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant |
+| `clip.cut` | clips | Cut (or re-cut) a clip row from its VOD: `POST /api/v1/:app/clips` and `/:id/recut` answer 202 with `job_id`, which a UI polls and reattaches to (`GET /api/v2/:app/jobs/:id`). `clip.ready`/`clip.failed` and the webhook come from the cut as before. It retries after 2, 10 and 30 min (from attempt 2 a cloud VOD is fetched home first); a window with no footage fails at once. The `clips` lane (`MEDIA_JOBS_CLIPS_CONCURRENCY`, 2) runs while recording |
| `object.waveform` | heavy | The audio waveform as a PNG (`showwavespic`, 1800×140 by default; `width`, `height`). It is a private `asset` object and the source's `waveform` variant. A remote-only source over `MEDIA_PREVIEW_REMOTE_MAX_MB` (2048) is refused, because it decodes the whole audio track. `no_audio` is permanent |
| `object.waveform` | heavy | The audio waveform as a PNG (`showwavespic`, 1800×140 by default; `width`, `height`). It is a private `asset` object and the source's `waveform` variant. A remote-only source over `MEDIA_PREVIEW_REMOTE_MAX_MB` (2048) is refused, because it decodes the whole audio track. `no_audio` is permanent |
| `object.sprite` | heavy | A seek-preview sprite sheet as a JPEG, made from one fast seek per frame: 100 frames by default, at most one per 2 s, 10 columns and 160 px tiles (`frames`, `columns`, `tile_width`). It is a private `asset` object and the source's `sprite` variant. The layout goes in its metadata as `sprite: { count, interval_seconds, columns, rows, tile_width, tile_height }`: frame *i* covers [*i*·interval, (*i*+1)·interval) at column *i* mod columns, row ⌊*i*/columns⌋. `no_video` is permanent |
| `object.sprite` | heavy | A seek-preview sprite sheet as a JPEG, made from one fast seek per frame: 100 frames by default, at most one per 2 s, 10 columns and 160 px tiles (`frames`, `columns`, `tile_width`). It is a private `asset` object and the source's `sprite` variant. The layout goes in its metadata as `sprite: { count, interval_seconds, columns, rows, tile_width, tile_height }`: frame *i* covers [*i*·interval, (*i*+1)·interval) at column *i* mod columns, row ⌊*i*/columns⌋. `no_video` is permanent |
| `object.hash` | light | sha256 of ready objects' present local copies that have no hash yet, onto the object (`content_hash`, `metadata.hash_basis`) and the local location (`checksum`). Tenant-wide batch (`params { limit (1-500, 50), budget_mb }`) or one object. Scheduled per tenant every `MEDIA_HASH_INTERVAL_MIN`. Result `{ hashed, bytes, skipped, remaining }`. See [Scheduled verification](#scheduled-verification) |
| `object.hash` | light | sha256 of ready objects' present local copies that have no hash yet, onto the object (`content_hash`, `metadata.hash_basis`) and the local location (`checksum`). Tenant-wide batch (`params { limit (1-500, 50), budget_mb }`) or one object. Scheduled per tenant every `MEDIA_HASH_INTERVAL_MIN`. Result `{ hashed, bytes, skipped, remaining }`. See [Scheduled verification](#scheduled-verification) |
@@ -71,6 +71,7 @@ const config = {
@@ -71,6 +71,7 @@ const config = {
lightConcurrency: Math.max(1, intEnv('MEDIA_JOBS_LIGHT_CONCURRENCY', 2)), // thumbnails, scans
lightConcurrency: Math.max(1, intEnv('MEDIA_JOBS_LIGHT_CONCURRENCY', 2)), // thumbnails, scans
heavyConcurrency: Math.max(1, intEnv('MEDIA_JOBS_HEAVY_CONCURRENCY', 1)), // split, remux
heavyConcurrency: Math.max(1, intEnv('MEDIA_JOBS_HEAVY_CONCURRENCY', 1)), // split, remux
finalizeConcurrency: Math.max(1, intEnv('MEDIA_JOBS_FINALIZE_CONCURRENCY', 1)), // vod.finalize retries
finalizeConcurrency: Math.max(1, intEnv('MEDIA_JOBS_FINALIZE_CONCURRENCY', 1)), // vod.finalize retries
+ clipsConcurrency: Math.max(1, intEnv('MEDIA_JOBS_CLIPS_CONCURRENCY', 2)), // clip.cut (runs while recording)
// Heavy jobs wait while a recording is running (as the health job does), unless this is on.
// Heavy jobs wait while a recording is running (as the health job does), unless this is on.
heavyWhileRecording: ['1', 'true', 'on'].includes(String(process.env.MEDIA_JOBS_HEAVY_WHILE_RECORDING || '').toLowerCase()),
heavyWhileRecording: ['1', 'true', 'on'].includes(String(process.env.MEDIA_JOBS_HEAVY_WHILE_RECORDING || '').toLowerCase()),
leaseS: Math.max(30, intEnv('MEDIA_JOBS_LEASE_S', 120)),
leaseS: Math.max(30, intEnv('MEDIA_JOBS_LEASE_S', 120)),
@@ -8,6 +8,7 @@
@@ -8,6 +8,7 @@
* object.waveform heavy the audio waveform (PNG) of a vod/clip as the source's `waveform` variant
* object.waveform heavy the audio waveform (PNG) of a vod/clip as the source's `waveform` variant
* object.sprite heavy a seek-preview sprite sheet (JPEG, layout in its metadata) as the `sprite` variant
* object.sprite heavy a seek-preview sprite sheet (JPEG, layout in its metadata) as the `sprite` variant
* vod.finalize finalize finalize a recording whose finalize failed or never ran (orphans), with backoff
* vod.finalize finalize finalize a recording whose finalize failed or never ran (orphans), with backoff
+ * clip.cut clips cut (or re-cut) a clip row from its VOD; retries with backoff (server/vod/clip-jobs.js)
* vod.duration.reconcile heavy stored VOD durations vs a measurement of the real file (local or B2/R2), one batch
* vod.duration.reconcile heavy stored VOD durations vs a measurement of the real file (local or B2/R2), one batch
* object.hash light sha256 of local copies with no content hash yet, one bounded batch (or one object)
* object.hash light sha256 of local copies with no content hash yet, one bounded batch (or one object)
* storage.orphans.scan light the storage orphan report, service-wide (report only; never deletes)
* storage.orphans.scan light the storage orphan report, service-wide (report only; never deletes)
@@ -23,6 +24,7 @@ queue.register('object.remux', require('./derive').remux);
@@ -23,6 +24,7 @@ queue.register('object.remux', require('./derive').remux);
queue.register('object.waveform', require('./previews').waveform);
queue.register('object.waveform', require('./previews').waveform);
queue.register('object.sprite', require('./previews').sprite);
queue.register('object.sprite', require('./previews').sprite);
queue.register('vod.finalize', require('./vod-finalize').spec);
queue.register('vod.finalize', require('./vod-finalize').spec);
+queue.register('clip.cut', require('../vod/clip-jobs').spec);
queue.register('vod.duration.reconcile', require('./duration-reconcile').spec);
queue.register('vod.duration.reconcile', require('./duration-reconcile').spec);
queue.register('object.hash', require('./content-hash').spec);
queue.register('object.hash', require('./content-hash').spec);
queue.register('storage.orphans.scan', require('./storage-orphans').spec);
queue.register('storage.orphans.scan', require('./storage-orphans').spec);
@@ -7,6 +7,8 @@
@@ -7,6 +7,8 @@
* recording runs unless MEDIA_JOBS_HEAVY_WHILE_RECORDING=1
* recording runs unless MEDIA_JOBS_HEAVY_WHILE_RECORDING=1
* finalize vod.finalize MEDIA_JOBS_FINALIZE_CONCURRENCY (1); runs while recording
* finalize vod.finalize MEDIA_JOBS_FINALIZE_CONCURRENCY (1); runs while recording
* (it is the work the recorder does when a stream ends)
* (it is the work the recorder does when a stream ends)
+ * clips clip.cut MEDIA_JOBS_CLIPS_CONCURRENCY (2); runs while recording (people
+ * clip live streams and wait for the result)
*
*
* A running job holds a lease (MEDIA_JOBS_LEASE_S) that a heartbeat renews; the heartbeat also notices
* A running job holds a lease (MEDIA_JOBS_LEASE_S) that a heartbeat renews; the heartbeat also notices
* an owner's cancel request. At start every job left `running` by the previous process is retried (or
* an owner's cancel request. At start every job left `running` by the previous process is retried (or
@@ -41,6 +43,7 @@ function lanes() {
@@ -41,6 +43,7 @@ function lanes() {
{ name: 'light', types: by('light'), max: cfg().lightConcurrency },
{ name: 'light', types: by('light'), max: cfg().lightConcurrency },
{ name: 'heavy', types: by('heavy'), max: cfg().heavyConcurrency },
{ name: 'heavy', types: by('heavy'), max: cfg().heavyConcurrency },
{ name: 'finalize', types: by('finalize'), max: cfg().finalizeConcurrency },
{ name: 'finalize', types: by('finalize'), max: cfg().finalizeConcurrency },
+ { name: 'clips', types: by('clips'), max: cfg().clipsConcurrency || 2 },
];
];
}
}
@@ -5,6 +5,10 @@
@@ -5,6 +5,10 @@
* resolves the source, cuts, updates the row, thumbnails,
* resolves the source, cuts, updates the row, thumbnails,
* commits the outcome with its event, fires the webhook. Records cut_error / cut_attempts and
* commits the outcome with its event, fires the webhook. Records cut_error / cut_attempts and
* schedules an automatic retry on failure.
* schedules an automatic retry on failure.
+ * spec the media job clip.cut (lane clips; roadmap WS-G task 3): a new clip's
+ * cut and every re-cut run as a job, so they carry media.job.* events and a
+ * job id the UI can reattach to (GET /api/v2/:app/jobs/:id). The queue
+ * owns the retries (backoff 2, 10, 30 min; a hopeless source fails at once).
* start() sweeper: every 3 min, failed clips that still have
* start() sweeper: every 3 min, failed clips that still have
* attempts left are re-cut. From the second attempt on, a
* attempts left are re-cut. From the second attempt on, a
* VOD that lives in cloud storage is first pulled back to
* VOD that lives in cloud storage is first pulled back to
@@ -48,7 +52,7 @@ async function recutClip(clipId, opts = {}) {
@@ -48,7 +52,7 @@ async function recutClip(clipId, opts = {}) {
return result;
return result;
}
}
-async function _recutClip(clipId, { reason = 'recut' } = {}) {
+async function _recutClip(clipId, { reason = 'recut', viaJob = false } = {}) {
const clip = db.getClipById(clipId);
const clip = db.getClipById(clipId);
if (!clip) return { ok: false, error: 'Clip not found' };
if (!clip) return { ok: false, error: 'Clip not found' };
if (!clip.vod_id) return { ok: false, error: 'Clip has no source VOD' };
if (!clip.vod_id) return { ok: false, error: 'Clip has no source VOD' };
@@ -74,12 +78,12 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {
@@ -74,12 +78,12 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {
console.log(`[Clips] clip ${clipId}: not enough free disk to pull vod ${vod.id} home — cutting from the cloud copy`);
console.log(`[Clips] clip ${clipId}: not enough free disk to pull vod ${vod.id} home — cutting from the cloud copy`);
}
}
}
}
- if (!source) return fail(clipId, clip, attempt, 'VOD media unavailable (not on disk and no cloud copy)');
+ if (!source) return fail(clipId, clip, attempt, 'VOD media unavailable (not on disk and no cloud copy)', { viaJob });
const startTime = Number(clip.start_time) || 0;
const startTime = Number(clip.start_time) || 0;
const duration = Math.max(1, (Number(clip.end_time) || 0) - startTime);
const duration = Math.max(1, (Number(clip.end_time) || 0) - startTime);
const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });
const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });
- if (!cut.ok) return fail(clipId, clip, attempt, cut.error);
+ if (!cut.ok) return fail(clipId, clip, attempt, cut.error, { viaJob });
// Thumbnail first (the clip.ready payload carries it), then the ready transition, the clip's
// Thumbnail first (the clip.ready payload carries it), then the ready transition, the clip's
// object and its event in one transaction (webhooks.announce), then the webhook.
// object and its event in one transaction (webhooks.announce), then the webhook.
try { await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath); } catch { /* */ }
try { await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath); } catch { /* */ }
@@ -89,23 +93,24 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {
@@ -89,23 +93,24 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {
payload: () => _clipPublic(db.getClipById(clipId)),
payload: () => _clipPublic(db.getClipById(clipId)),
});
});
console.log(`[Clips] Clip ${clipId} ${reason} OK from vod ${clip.vod_id} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s, attempt ${attempt})`);
console.log(`[Clips] Clip ${clipId} ${reason} OK from vod ${clip.vod_id} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s, attempt ${attempt})`);
- return { ok: true };
+ return { ok: true, duration_seconds: cut.duration };
}
}
-function fail(clipId, clip, attempt, error) {
+/** A failed attempt. Under a job the queue schedules the retry, so nothing is left for the sweeper. */
+function fail(clipId, clip, attempt, error, { viaJob = false } = {}) {
const msg = String(error || 'cut failed').slice(0, 500);
const msg = String(error || 'cut failed').slice(0, 500);
// Nothing to retry when the source itself has no footage there (empty recording, window
// Nothing to retry when the source itself has no footage there (empty recording, window
// past the end of what was captured): give up now instead of burning the whole ladder.
// past the end of what was captured): give up now instead of burning the whole ladder.
const hopeless = /No decodable footage|Error opening input: End of file|Source VOD no longer exists|Invalid data found when processing input/i.test(msg);
const hopeless = /No decodable footage|Error opening input: End of file|Source VOD no longer exists|Invalid data found when processing input/i.test(msg);
- const more = attempt < MAX_ATTEMPTS && !hopeless;
+ const more = !viaJob && attempt < MAX_ATTEMPTS && !hopeless;
const mins = BACKOFF_MIN[Math.min(attempt - 1, BACKOFF_MIN.length - 1)];
const mins = BACKOFF_MIN[Math.min(attempt - 1, BACKOFF_MIN.length - 1)];
const nextAt = more ? new Date(Date.now() + mins * 60000).toISOString().replace('T', ' ').slice(0, 19) : null;
const nextAt = more ? new Date(Date.now() + mins * 60000).toISOString().replace('T', ' ').slice(0, 19) : null;
announce(clip.app_id, 'clip.failed', {
announce(clip.app_id, 'clip.failed', {
change: () => objects().withObject('clip', clipId, () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = ? WHERE id = ?", [msg, nextAt, clipId])),
change: () => objects().withObject('clip', clipId, () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = ? WHERE id = ?", [msg, nextAt, clipId])),
payload: () => _clipPublic(db.getClipById(clipId)),
payload: () => _clipPublic(db.getClipById(clipId)),
});
});
- console.warn(`[Clips] Clip ${clipId} attempt ${attempt}/${MAX_ATTEMPTS} failed: ${msg}${more ? ` — retry in ${mins} min` : ' — giving up'}`);
- return { ok: false, error: msg, retry_at: nextAt };
+ console.warn(`[Clips] Clip ${clipId} attempt ${attempt}/${MAX_ATTEMPTS} failed: ${msg}${viaJob ? (hopeless ? ' — giving up' : ' — the job retries') : more ? ` — retry in ${mins} min` : ' — giving up'}`);
+ return { ok: false, error: msg, retry_at: nextAt, hopeless };
}
}
async function sweep() {
async function sweep() {
@@ -114,7 +119,9 @@ async function sweep() {
@@ -114,7 +119,9 @@ async function sweep() {
try {
try {
const due = db.all(`SELECT id FROM clips WHERE status = 'failed' AND vod_id IS NOT NULL
const due = db.all(`SELECT id FROM clips WHERE status = 'failed' AND vod_id IS NOT NULL
AND COALESCE(cut_attempts, 0) < ? AND (cut_next_at IS NULL OR cut_next_at <= CURRENT_TIMESTAMP)
AND COALESCE(cut_attempts, 0) < ? AND (cut_next_at IS NULL OR cut_next_at <= CURRENT_TIMESTAMP)
- AND created_at >= datetime('now', '-30 days') ORDER BY created_at DESC LIMIT 4`, [MAX_ATTEMPTS]) || [];
+ AND created_at >= datetime('now', '-30 days')
+ AND NOT EXISTS (SELECT 1 FROM media_jobs j WHERE j.job_type = 'clip.cut' AND j.app_id = clips.app_id AND CAST(json_extract(j.params, '$.clip_id') AS INTEGER) = clips.id)
+ ORDER BY created_at DESC LIMIT 4`, [MAX_ATTEMPTS]) || [];
for (const row of due) { try { await recutClip(row.id, { reason: 'auto-retry' }); } catch (e) { console.warn(`[Clips] auto-retry ${row.id}:`, e.message); } }
for (const row of due) { try { await recutClip(row.id, { reason: 'auto-retry' }); } catch (e) { console.warn(`[Clips] auto-retry ${row.id}:`, e.message); } }
} finally { _busy = false; }
} finally { _busy = false; }
}
}
@@ -137,4 +144,41 @@ function start() {
@@ -137,4 +144,41 @@ function start() {
console.log('[Clips] cut retry sweeper started (every 3 min, up to 4 attempts, hot-fetch from attempt 2)');
console.log('[Clips] cut retry sweeper started (every 3 min, up to 4 attempts, hot-fetch from attempt 2)');
}
}
-module.exports = { recutClip, sweep, start, MAX_ATTEMPTS };
+// ── clip.cut as a media job ──────────────────────────────────
+
+/** The clip's media object (legacy:<app>:clip:<id>), the job's object. */
+function clipObjectId(appId, clipId) {
+ const row = db.get('SELECT id FROM media_objects WHERE legacy_ref = ?', [`legacy:${appId}:clip:${clipId}`]);
+ return row ? row.id : null;
+}
+
+/** Queue the cut of a clip row → { job, created }. A repeated request for the same clip reuses its active job. */
+function enqueueCut(appId, clipId, { reason = 'cut', createdBy = null, ownerUserId = null } = {}) {
+ const queue = require('../jobs/queue');
+ const out = queue.enqueue({ appId, type: 'clip.cut', objectId: clipObjectId(appId, clipId), params: { clip_id: Number(clipId), reason }, dedupeActive: true, createdBy, ownerUserId });
+ try { require('../jobs/worker').kick(); } catch { /* the poll picks it up */ }
+ return out;
+}
+
+const spec = {
+ lane: 'clips',
+ maxAttempts: MAX_ATTEMPTS,
+ timeoutMs: RECUT_CAP_MS + 5 * 60 * 1000,
+ needsObject: false,
+ backoffS: (attempt) => BACKOFF_MIN[Math.min(Math.max(0, attempt - 1), BACKOFF_MIN.length - 1)] * 60, // after attempt 1: 2 min, then 10, 30
+ validate({ params }) {
+ const id = Number(params && params.clip_id);
+ if (!Number.isInteger(id) || id < 1) throw new (require('../jobs/queue').JobError)('media.job.invalid', 'clip.cut needs params.clip_id', { permanent: true });
+ return { clip_id: id, reason: String((params && params.reason) || 'cut').slice(0, 40) };
+ },
+ async run(job) {
+ const { JobError } = require('../jobs/queue');
+ const clip = db.getClipById(job.params.clip_id);
+ if (!clip || clip.app_id !== job.app_id) throw new JobError('not_found', 'The clip no longer exists', { permanent: true });
+ const r = await recutClip(clip.id, { reason: job.params.reason || 'cut', viaJob: true });
+ if (!r.ok) throw new JobError('clip_cut_failed', r.error || 'cut failed', { permanent: !!r.hopeless });
+ return { clip_id: clip.id, status: 'ready', duration_seconds: r.duration_seconds ?? null };
+ },
+};
+
+module.exports = { recutClip, sweep, start, enqueueCut, clipObjectId, spec, MAX_ATTEMPTS };
@@ -20,9 +20,7 @@ const multer = require('multer');
@@ -20,9 +20,7 @@ const multer = require('multer');
const config = require('../config');
const config = require('../config');
const db = require('../db/database');
const db = require('../db/database');
const tools = require('./media-tools');
const tools = require('./media-tools');
-const cutter = require('./clip-cutter');
const { tenantAuth, tenantCors } = require('../auth');
const { tenantAuth, tenantCors } = require('../auth');
-const { announce } = require('../webhooks');
const objects = require('../objects/model');
const objects = require('../objects/model');
const router = express.Router({ mergeParams: true });
const router = express.Router({ mergeParams: true });
@@ -157,10 +155,6 @@ router.post('/', tenantAuth({ allowUser: true }), clipUpload.single('video'), as
@@ -157,10 +155,6 @@ router.post('/', tenantAuth({ allowUser: true }), clipUpload.single('video'), as
return res.status(200).json({ id: duplicate.id, status: duplicate.status, deduplicated: true });
return res.status(200).json({ id: duplicate.id, status: duplicate.status, deduplicated: true });
}
}
- if (cutter.isBusy()) {
- return res.status(503).json({ error: 'Server is busy processing other clips. Please try again in a few seconds.' });
- }
-
// Resolve a source ffmpeg can read: local file, or presigned B2/R2 URL.
// Resolve a source ffmpeg can read: local file, or presigned B2/R2 URL.
const vodStorage = require('./vod-storage');
const vodStorage = require('./vod-storage');
const source = await vodStorage.resolveMediaSource(vod);
const source = await vodStorage.resolveMediaSource(vod);
@@ -213,34 +207,14 @@ router.post('/', tenantAuth({ allowUser: true }), clipUpload.single('video'), as
@@ -213,34 +207,14 @@ router.post('/', tenantAuth({ allowUser: true }), clipUpload.single('video'), as
const clipId = result.lastInsertRowid;
const clipId = result.lastInsertRowid;
const appId = req.appId;
const appId = req.appId;
- // Cut in the background; webhook on completion.
- (async () => {
- const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });
- if (cut.ok) {
- // Thumbnail first so the clip.ready payload carries it; then the ready transition,
- // the clip's object and its event commit together (webhooks.announce) and the webhook goes out.
- try {
- await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath);
- } catch { /* */ }
- announce(appId, 'clip.ready', {
- change: () => objects.withObject('clip', clipId, () => db.run('UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = ? WHERE id = ?',
- [cut.filePath, cut.duration, startTime + cut.duration, 'ready', clipId])),
- payload: () => clipPublic(db.getClipById(clipId)),
- });
- console.log(`[Clips] Clip ${clipId} cut from vod ${vodId} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s)`);
- } else {
- // First attempt failed: record why and let the retry sweeper take it from here
- // (attempt 2 pulls a cloud-stored VOD back to local disk first).
- const nextAt = new Date(Date.now() + 2 * 60000).toISOString().replace('T', ' ').slice(0, 19);
- announce(appId, 'clip.failed', {
- change: () => objects.withObject('clip', clipId, () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_attempts = 1, cut_next_at = ? WHERE id = ?", [String(cut.error || 'cut failed').slice(0, 500), nextAt, clipId])),
- payload: () => clipPublic(db.getClipById(clipId)),
- });
- console.warn(`[Clips] Clip ${clipId} failed: ${cut.error} — auto-retry in 2 min`);
- }
- })().catch(err => console.error('[Clips] Background cut error:', err.message));
+ // The cut is a media job (clip.cut, WS-G task 3): media.job.* events, retries with backoff, and a
+ // job id the caller's UI follows (GET /api/v2/:app/jobs/:id) and reattaches to after a reload.
+ // clip.ready / clip.failed and the webhook come from the cut itself, as before.
+ const { job } = require('./clip-jobs').enqueueCut(appId, clipId, {
+ reason: 'cut', createdBy: req.authType === 'user' ? `app:${appId}:user:${req.userId}` : `app:${appId}`, ownerUserId: userId ?? null,
+ });
- res.status(202).json({ id: clipId, status: 'processing' });
+ res.status(202).json({ id: clipId, status: 'processing', job_id: job.id });
} catch (err) {
} catch (err) {
console.error('[Clips] Create error:', err.message);
console.error('[Clips] Create error:', err.message);
if (req.file) tools.cleanupTempFile(req.file.path);
if (req.file) tools.cleanupTempFile(req.file.path);
@@ -372,8 +346,8 @@ router.post('/:id/recut', tenantAuth(), async (req, res) => {
@@ -372,8 +346,8 @@ router.post('/:id/recut', tenantAuth(), async (req, res) => {
if (!clip.vod_id) return res.status(422).json({ error: 'Clip has no source VOD to re-cut from' });
if (!clip.vod_id) return res.status(422).json({ error: 'Clip has no source VOD to re-cut from' });
// A manual retry resets the attempt counter so it gets the full ladder again (not projected: no object change).
// A manual retry resets the attempt counter so it gets the full ladder again (not projected: no object change).
db.run("UPDATE clips SET cut_attempts = 0 WHERE id = ?", [clipId]);
db.run("UPDATE clips SET cut_attempts = 0 WHERE id = ?", [clipId]);
- require('./clip-jobs').recutClip(clipId, { reason: 're-cut' }).catch(err => console.error('[Clips] Background re-cut error:', err.message));
- res.status(202).json({ id: clipId, status: 'processing' });
+ const { job } = require('./clip-jobs').enqueueCut(req.appId, clipId, { reason: 're-cut', createdBy: `app:${req.appId}` });
+ res.status(202).json({ id: clipId, status: 'processing', job_id: job.id });
} catch (err) {
} catch (err) {
console.error('[Clips] Re-cut error:', err.message);
console.error('[Clips] Re-cut error:', err.message);
res.status(500).json({ error: 'Failed to re-cut clip' });
res.status(500).json({ error: 'Failed to re-cut clip' });
new file mode 100644
new file mode 100644
@@ -0,0 +1,120 @@
@@ -0,0 +1,120 @@
+'use strict';
+// clip.cut as a media job (WS-G task 3): POST /clips answers 202 with the clip and its job id; the job
+// cuts the clip (clip.ready as before) with media.job.* events, and GET /api/v2/:app/jobs/:id is what a
+// UI polls and reattaches to. A window with no footage fails permanently at once; the old sweeper leaves
+// clips that have a job alone; a re-cut is a new job, and asking again while it runs gets the same one.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const { spawnSync } = require('child_process');
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-clipjob-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+process.env.DB_PATH = path.join(tmp, 'media.db');
+process.env.VOD_PATH = dir('vods');
+process.env.CLIPS_PATH = dir('clips');
+process.env.FILES_PATH = dir('files');
+process.env.THUMBNAILS_PATH = dir('thumbnails');
+process.env.PASTES_PATH = dir('pastes');
+process.env.OBJECTS_PATH = dir('objects');
+process.env.MEDIA_JOBS_POLL_MS = '50';
+process.env.MEDIA_INVARIANT_SCAN_HOURS = '0';
+process.env.MEDIA_UPLOAD_MIN_FREE_MB = '0';
+const stub = http.createServer((req, res) => {
+ req.resume();
+ req.on('end', () => {
+ res.setHeader('Content-Type', 'application/json');
+ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));
+ res.statusCode = 404; res.end('{}');
+ });
+});
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+const waitFor = async (fn, ms = 30000) => { const end = Date.now() + ms; while (Date.now() < end) { if (await fn()) return true; await sleep(25); } return false; };
+
+(async () => {
+ if (spawnSync('ffmpeg', ['-version']).status !== 0) { console.log('⚠️ ffmpeg not found: clip.cut skipped'); process.exit(0); }
+ await new Promise((r) => stub.listen(0, '127.0.0.1', r));
+ const base = `http://127.0.0.1:${stub.address().port}`;
+ const db = require('../server/db/database');
+ const conn = db.getDb();
+ db.upsertApp({ app_id: 'live', api_key: 'live-key' });
+ const events = require('../server/events');
+ events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 60000 });
+ const queue = require('../server/jobs/queue');
+ const worker = require('../server/jobs/worker');
+ const clipJobs = require('../server/vod/clip-jobs');
+ const express = require('express');
+ const app = express();
+ app.use(express.json());
+ app.use('/api/v1/:app/clips', require('../server/vod/clips-routes'));
+ app.use('/api/v2/:app/jobs', require('../server/jobs/routes'));
+ const server = http.createServer(app);
+ await new Promise((r) => server.listen(0, '127.0.0.1', r));
+ const api = `http://127.0.0.1:${server.address().port}`;
+ const call = async (method, p, body) => {
+ const res = await fetch(api + p, { method, headers: { authorization: 'Bearer live-key', 'content-type': 'application/json' }, body: body === undefined ? undefined : JSON.stringify(body) });
+ return { status: res.status, body: await res.json().catch(() => null) };
+ };
+ const outbox = () => conn.prepare('SELECT envelope FROM event_outbox ORDER BY id').all().map((r) => JSON.parse(r.envelope));
+ const jobEvents = (id) => outbox().filter((e) => e.subject.type === 'job' && e.subject.id === id).map((e) => e.event_type);
+ worker.start();
+
+ const src = path.join(process.env.VOD_PATH, 'rec-30.webm');
+ const mk = spawnSync('ffmpeg', ['-v', 'error', '-y', '-f', 'lavfi', '-i', 'testsrc=duration=30:size=320x180:rate=10', '-f', 'lavfi', '-i', 'sine=frequency=440:duration=30',
+ '-c:v', 'libvpx', '-g', '10', '-b:v', '200k', '-c:a', 'libopus', '-shortest', src]);
+ assert.strictEqual(mk.status, 0, String(mk.stderr));
+ conn.prepare("INSERT INTO vods (id, app_id, user_id, title, file_path, file_size, is_public, duration_seconds) VALUES (30, 'live', 5, 'Thirty seconds', ?, ?, 1, 30)").run(src, fs.statSync(src).size);
+
+ // ── A clip: 202 with its job; the job cuts it ──
+ let r = await call('POST', '/api/v1/live/clips', { vod_id: 30, start_s: 5, end_s: 12, title: 'Five to twelve', user_id: 5 });
+ assert.strictEqual(r.status, 202, JSON.stringify(r.body));
+ assert.ok(/^mjob_/.test(r.body.job_id), 'the answer names the job');
+ const clipId = r.body.id, jobId = r.body.job_id;
+ assert.ok(await waitFor(() => ['succeeded', 'failed'].includes(queue.get(jobId).status)), 'the cut finishes');
+ const j = await call('GET', `/api/v2/live/jobs/${jobId}`);
+ assert.strictEqual(j.status, 200);
+ assert.deepStrictEqual([j.body.job.type, j.body.job.status, j.body.job.result.clip_id], ['clip.cut', 'succeeded', clipId], JSON.stringify(j.body.job));
+ const clip = db.getClipById(clipId);
+ assert.strictEqual(clip.status, 'ready');
+ assert.ok(fs.existsSync(clip.file_path) && Math.abs(clip.duration_seconds - 7) < 1.5, `a 7 s clip (${clip.duration_seconds})`);
+ assert.deepStrictEqual(jobEvents(jobId), ['media.job.queued', 'media.job.started', 'media.job.succeeded']);
+ assert.ok(outbox().some((e) => e.event_type === 'media.clip.ready' && e.subject.type === 'clip' && String(e.subject.id) === String(clipId)), 'media.clip.ready as before');
+
+ // ── A window with no footage: permanent at once, and the sweeper leaves it alone ──
+ r = await call('POST', '/api/v1/live/clips', { vod_id: 30, start_s: 28, end_s: 35, title: 'Past the end', user_id: 5 });
+ assert.strictEqual(r.status, 202, JSON.stringify(r.body));
+ const badJob = r.body.job_id, badClip = r.body.id;
+ assert.ok(await waitFor(() => ['succeeded', 'failed'].includes(queue.get(badJob).status)));
+ const bad = queue.jobPublic(queue.get(badJob));
+ if (bad.status === 'failed') {
+ assert.strictEqual(db.getClipById(badClip).status, 'failed');
+ assert.strictEqual(db.getClipById(badClip).cut_next_at, null, 'no sweeper retry is scheduled under a job');
+ const before = db.getClipById(badClip).cut_attempts;
+ await clipJobs.sweep();
+ assert.strictEqual(db.getClipById(badClip).cut_attempts, before, 'the sweeper leaves a clip that has a job');
+ } else {
+ assert.strictEqual(db.getClipById(badClip).status, 'ready', 'ffmpeg cut what there was');
+ }
+
+ // ── A re-cut is a new job; asking again while it is active gets the same one ──
+ const r1 = await call('POST', `/api/v1/live/clips/${clipId}/recut`);
+ assert.strictEqual(r1.status, 202, JSON.stringify(r1.body));
+ assert.ok(r1.body.job_id && r1.body.job_id !== jobId, 'a new job');
+ const same = clipJobs.enqueueCut('live', clipId, { reason: 're-cut' });
+ assert.ok(same.job.id === r1.body.job_id || queue.get(r1.body.job_id).status === 'succeeded', 'an active re-cut is reused');
+ assert.ok(await waitFor(() => queue.get(r1.body.job_id).status === 'succeeded'));
+ assert.strictEqual(db.getClipById(clipId).status, 'ready');
+
+ // ── Another tenant cannot see the job ──
+ db.upsertApp({ app_id: 'games', api_key: 'games-key' });
+ const other = await fetch(`${api}/api/v2/games/jobs/${jobId}`, { headers: { authorization: 'Bearer games-key' } });
+ assert.strictEqual(other.status, 404);
+
+ worker.stop();
+ events._reset();
+ server.close();
+ stub.close();
+ console.log('clip-cut-job: all checks passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });