openvibers/openvibe.media Commits
Waveform and preview-sprite variants as jobs (WS-G task 13): object.waveform (PNG, showwavespic) and object.sprite (JPEG sheet from one fast seek per frame, layout in its metadata) make private asset objects recorded as the source's waveform / sprite variant; no_audio / no_video are permanent; a remote-only source over MEDIA_PREVIEW_REMOTE_MAX_MB gets no waveform
Changes
@@@ -50,7 +50,7 @@ server/change's transaction; payload queue.jobEvent: ids, state, ISO times,has_result, never params/result/error text/creator), worker (light/heavy lanes, leases, retries,cancellation), routes (/api/v2/:app/jobs), types: thumbnail.regenerate,-- invariant.scan (proposes split/remux), object.split, object.remux,++ invariant.scan (proposes split/remux), object.split, object.remux, object.waveform, object.sprite,vod.finalize (orphans and failed finalizes, with backoff),vod.duration.reconcile (stored vs measured durations, local and B2/R2),storage.orphans.scan (monthly storage orphan report, service-wide, report only)@@@ -599,6 +599,8 @@ 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 |++| `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) || `vod.duration.reconcile` | heavy | One bounded batch of the tenant's finished VODs: stored duration vs a measurement of the real file, local or the B2/R2 copy (ffprobe over a presigned URL, ranged reads). `params { limit (1-200, 25), after_id?, apply (false), confirm_remote (false) }`; without `after_id` it walks the library from where the last run stopped. Repairs only confirmed, clearly wrong values when `apply`. Result `{ counts, range, next_after_id, report }`. See [Duration reconciliation](#duration-reconciliation) || `storage.orphans.scan` | light | The [storage orphan report](#storage-orphan-report), service-wide, under the system app `_media`; report only. `params { limit (1-100000, 2000) }`. Result `{ totals, providers, summary, report }` (the report file under `<data>/reports/`). Scheduled every `MEDIA_ORPHAN_SCAN_DAYS`; tenants cannot enqueue it (403) |@@@ -50,7 +50,7 @@ function extFor(obj, sourcePath) {}function mimeFor(ext, obj) {-- return { '.webm': 'video/webm', '.mp4': 'video/mp4', '.mkv': 'video/x-matroska', '.m4a': 'audio/mp4', '.ogg': 'video/ogg', '.mp3': 'audio/mpeg' }[ext]++ return { '.webm': 'video/webm', '.mp4': 'video/mp4', '.mkv': 'video/x-matroska', '.m4a': 'audio/mp4', '.ogg': 'video/ogg', '.mp3': 'audio/mpeg', '.png': 'image/png', '.jpg': 'image/jpeg' }[ext]|| obj.mime_type || 'application/octet-stream';}@@@ -168,7 +168,8 @@ function cleanupWork(jobId) {* move the bytes under OBJECTS_PATH/<app>/<id>, verified local location, relationship; `alsoInTx`* (the job checkpoint) commits with it. Returns the new object id.*/--async function adopt({ src, file, ext, job, relation, metadata, variant, saveCheckpoint, checkpointFor }) {++// kind/suffix: a derived image (waveform, sprite) is an `asset` named <source>-<suffix>.<ext>; parts and remuxes keep the source's kind.++async function adopt({ src, file, ext, job, relation, metadata, variant, saveCheckpoint, checkpointFor, kind = null, suffix = null }) {const size = fs.statSync(file).size;const hash = await sha256File(file);const md = model.parseJson(src.metadata, {});@@@ -181,9 +182,9 @@ async function adopt({ src, file, ext, job, relation, metadata, variant, saveChedb.run(`INSERT INTO media_objects (id, app_id, namespace, kind, owner_subject, owner_app, owner_user_id, visibility, lifecycle_status,mime_type, size_bytes, content_hash, canonical_provider, canonical_key, legacy_ref, metadata)VALUES (?, ?, ?, ?, ?, ?, ?, 'private', 'ready', ?, ?, ?, 'local', ?, NULL, ?)`,-- [id, src.app_id, src.namespace || src.app_id, src.kind, src.owner_subject || null, src.owner_app || src.app_id, src.owner_user_id ?? null,++ [id, src.app_id, src.namespace || src.app_id, kind || src.kind, src.owner_subject || null, src.owner_app || src.app_id, src.owner_user_id ?? null,mimeFor(ext, src), size, hash, dest,-- JSON.stringify({ title: md.title || null, derived_from: src.id, job_id: job.id, filename: `${src.id}${metadata.part ? `-part${metadata.part}` : '-remux'}${ext}`, ...metadata })]);++ JSON.stringify({ title: md.title || null, derived_from: src.id, job_id: job.id, filename: `${src.id}${suffix ? `-${suffix}` : metadata.part ? `-part${metadata.part}` : '-remux'}${ext}`, ...metadata })]);model.upsertLocation(id, { provider: 'local', key: dest, state: 'present', size_bytes: size, checksum: hash, verified: true });model.setRelationship(id, 'derived_from', src.id, { job_id: job.id, ...relation });if (variant) model.setVariant(src.id, variant, id, `${job.type}@1`);@@@ -307,4 +308,6 @@ module.exports = {split: { lane: 'heavy', maxAttempts: 3, timeoutMs: 6 * 3600 * 1000, needsObject: true, validate: validateSplit, run: runSplit },remux: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateRemux, run: runRemux },resolveSource, isMediaObject, extFor,++ // for the preview jobs (previews.js)++ loadSource, adopt, ffmpeg, inputArgs, probeDuration, workDir, cleanupWork, existing, checkRoom,};nnew file mode 100644@@@ -0,0 +1,155 @@++/**++ * OpenVibe.Media — job types object.waveform and object.sprite (heavy lane; roadmap WS-G task 13)++ *++ * Preview variants of a ready vod/clip (or a native video/audio object). Like object.remux they never++ * touch the source: each output is a new private `asset` object in the same tenant, owned like the++ * source, with a verified local copy, a `derived_from` relationship and the source's variant of that name++ * (GET /api/v2/objects/:id lists variants).++ *++ * object.waveform params { width?, height? } a PNG of the audio's waveform (ffmpeg showwavespic), 1800×140++ * by default. It decodes the whole audio track, so a remote source over++ * MEDIA_PREVIEW_REMOTE_MAX_MB (2048) is refused: the egress would cost more than the preview.++ * result { source_id, object_id, size_bytes, width, height, method: 'waveform' }++ * object.sprite params { frames?, columns?, tile_width? } a JPEG sheet of evenly spaced frames for++ * seek previews: 100 frames by default (at most one per 2 s), 10 columns, 160 px wide++ * tiles. Each frame is its own fast seek, so a long remote VOD is never decoded end to end.++ * The layout is in the object's metadata (sprite: { interval_seconds, count, columns,++ * rows, tile_width, tile_height }): frame i covers [i × interval, (i + 1) × interval) and++ * sits at column i % columns, row ⌊i / columns⌋.++ * result { source_id, object_id, size_bytes, sprite, method: 'sprite' }++ *++ * Refused before any work: a source that is not ready or not a video/audio object, a waveform of a source++ * with no audio or a sprite of one with no video (permanent), not enough disk or quota (as the other++ * derive jobs).++ */++'use strict';++const fs = require('fs');++const path = require('path');++const { JobError } = require('./queue');++const d = require('./derive');++++const MB = 1024 * 1024;++const REMOTE_MAX_BYTES = (Number(process.env.MEDIA_PREVIEW_REMOTE_MAX_MB) || 2048) * MB;++const int = (v, dflt, min, max, name) => {++ if (v == null) return dflt;++ const n = Number(v);++ if (!Number.isInteger(n) || n < min || n > max) throw new JobError('media.job.invalid', `params.${name} must be an integer from ${min} to ${max}`, { permanent: true });++ return n;++};++++/** Which streams the source has: { audio, video, duration }. */++async function streamsOf(input) {++ const info = await require('../vod/media-tools').probeVodInfo(input).catch(() => null);++ const streams = (info && info.streams) || [];++ // An unreadable probe says nothing: ffmpeg itself decides then.++ if (!streams.length) return { audio: null, video: null, duration: Number(info && info.duration) || 0 };++ return { audio: streams.some((s) => s.codec_type === 'audio'), video: streams.some((s) => s.codec_type === 'video'), duration: Number(info.duration) || 0 };++}++++// ── object.waveform ─────────────────────────────────────────++++function validateWaveform({ obj, params }) {++ if (!obj) throw new JobError('media.job.invalid', 'object.waveform needs object_id', { permanent: true });++ if (!d.isMediaObject(obj)) throw new JobError('media.job.invalid', 'Only vod/clip objects (or video/audio objects) have a waveform', { permanent: true });++ const p = params || {};++ return { width: int(p.width, 1800, 200, 4000, 'width'), height: int(p.height, 140, 40, 600, 'height') };++}++++async function runWaveform(job, ctx) {++ const src = d.loadSource(job);++ const p = validateWaveform({ obj: src, params: job.params });++ if (ctx.checkpoint && ctx.checkpoint.object_id) { d.cleanupWork(job.id); return { source_id: src.id, method: 'waveform', ...ctx.checkpoint }; }++ const source = await d.resolveSource(src);++ if (!source) throw new JobError('media_unavailable', 'The source bytes are unavailable (no local file and no cloud copy)', { permanent: true });++ if (source.remote && Number(src.size_bytes) > REMOTE_MAX_BYTES) {++ throw new JobError('too_large', `A waveform decodes the whole audio track; this source is only in cloud storage and larger than ${REMOTE_MAX_BYTES / MB} MB`, { permanent: true });++ }++ d.checkRoom(src, 8 * MB);++ const out = path.join(d.workDir(job.id), 'waveform.png');++ const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', ...d.inputArgs(source), '-i', source.input,++ '-filter_complex', `[0:a:0]aformat=channel_layouts=mono,showwavespic=s=${p.width}x${p.height}:colors=#60a5fa:scale=sqrt[w]`, '-map', '[w]', '-frames:v', '1', out],++ { signal: ctx.signal, timeoutMs: 2 * 3600 * 1000 });++ if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted');++ if (!r.ok || !d.existing(out) || fs.statSync(out).size === 0) {++ d.cleanupWork(job.id);++ const noAudio = /matches no streams|0:a:0/i.test(r.stderr);++ throw new JobError(noAudio ? 'no_audio' : 'ffmpeg_failed', noAudio ? 'The source has no audio track' : `ffmpeg exited ${r.code}: ${String(r.stderr).trim().split('\n').slice(-2).join(' ').slice(0, 300)}`, { permanent: noAudio });++ }++ const made = await d.adopt({++ src, file: out, ext: '.png', job, variant: 'waveform', kind: 'asset', suffix: 'waveform',++ relation: { method: 'waveform', width: p.width, height: p.height },++ metadata: { waveform: { width: p.width, height: p.height } },++ saveCheckpoint: ctx.saveCheckpoint,++ checkpointFor: (id, size) => ({ object_id: id, size_bytes: size, width: p.width, height: p.height }),++ });++ d.cleanupWork(job.id);++ return { source_id: src.id, method: 'waveform', object_id: made.id, size_bytes: made.size, width: p.width, height: p.height };++}++++// ── object.sprite ───────────────────────────────────────────++++function validateSprite({ obj, params }) {++ if (!obj) throw new JobError('media.job.invalid', 'object.sprite needs object_id', { permanent: true });++ if (!d.isMediaObject(obj)) throw new JobError('media.job.invalid', 'Only vod/clip objects (or video objects) have a preview sprite', { permanent: true });++ const p = params || {};++ return { frames: int(p.frames, 100, 4, 400, 'frames'), columns: int(p.columns, 10, 1, 40, 'columns'), tile_width: int(p.tile_width, 160, 64, 480, 'tile_width') };++}++++/** The layout for a duration: frames at most one per 2 s. */++function spriteLayout(duration, { frames, columns, tile_width }) {++ const count = Math.max(1, Math.min(frames, Math.floor(duration / 2) || 1));++ const interval = duration > 0 ? duration / count : 0;++ const cols = Math.min(columns, count);++ return { count, interval_seconds: Math.round(interval * 1000) / 1000, columns: cols, rows: Math.ceil(count / cols), tile_width, tile_height: Math.round(tile_width * 9 / 16 / 2) * 2 };++}++++async function runSprite(job, ctx) {++ const src = d.loadSource(job);++ const p = validateSprite({ obj: src, params: job.params });++ if (ctx.checkpoint && ctx.checkpoint.object_id) { d.cleanupWork(job.id); return { source_id: src.id, method: 'sprite', ...ctx.checkpoint }; }++ const source = await d.resolveSource(src);++ if (!source) throw new JobError('media_unavailable', 'The source bytes are unavailable (no local file and no cloud copy)', { permanent: true });++ d.checkRoom(src, 16 * MB);++ const meta = await streamsOf(source.input);++ const md = require('../objects/model').parseJson(src.metadata, {});++ const duration = meta.duration || Number(md.duration_seconds) || 0;++ if (!(duration > 0)) throw new JobError('no_duration', 'The source has no known duration to spread frames over', { permanent: true });++ if (meta.video === false) throw new JobError('no_video', 'The source has no video track', { permanent: true });++ const layout = spriteLayout(duration, p);++ const dir = d.workDir(job.id);++ const tile = `scale=${layout.tile_width}:${layout.tile_height}:force_original_aspect_ratio=decrease,pad=${layout.tile_width}:${layout.tile_height}:(ow-iw)/2:(oh-ih)/2`;++ for (let i = 0; i < layout.count; i++) {++ if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted');++ const frame = path.join(dir, `f${String(i).padStart(4, '0')}.jpg`);++ if (d.existing(frame)) continue;++ const at = Math.min(duration - 0.1, i * layout.interval_seconds + layout.interval_seconds / 2);++ const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', ...d.inputArgs(source), '-ss', String(Math.max(0, at).toFixed(3)), '-i', source.input,++ '-frames:v', '1', '-vf', tile, '-q:v', '5', frame], { signal: ctx.signal, timeoutMs: 5 * 60 * 1000 });++ // A frame that cannot be read (a gap in a recording) is black rather than failing the sheet.++ if (!r.ok || !d.existing(frame)) {++ await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-f', 'lavfi', '-i', `color=c=black:s=${layout.tile_width}x${layout.tile_height}`, '-frames:v', '1', frame], { signal: ctx.signal, timeoutMs: 60000 });++ }++ }++ const out = path.join(dir, 'sprite.jpg');++ const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-framerate', '1', '-i', path.join(dir, 'f%04d.jpg'),++ '-vf', `tile=${layout.columns}x${layout.rows}`, '-frames:v', '1', '-q:v', '4', out], { signal: ctx.signal, timeoutMs: 10 * 60 * 1000 });++ if (!r.ok || !d.existing(out) || fs.statSync(out).size === 0) {++ d.cleanupWork(job.id);++ throw new JobError('ffmpeg_failed', `ffmpeg exited ${r.code}: ${String(r.stderr).trim().split('\n').slice(-2).join(' ').slice(0, 300)}`);++ }++ const made = await d.adopt({++ src, file: out, ext: '.jpg', job, variant: 'sprite', kind: 'asset', suffix: 'sprite',++ relation: { method: 'sprite', ...layout },++ metadata: { sprite: layout, source_duration_seconds: duration },++ saveCheckpoint: ctx.saveCheckpoint,++ checkpointFor: (id, size) => ({ object_id: id, size_bytes: size, sprite: layout }),++ });++ d.cleanupWork(job.id);++ return { source_id: src.id, method: 'sprite', object_id: made.id, size_bytes: made.size, sprite: layout };++}++++module.exports = {++ waveform: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateWaveform, run: runWaveform },++ sprite: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateSprite, run: runSprite },++ spriteLayout,++};@@@ -5,6 +5,8 @@* invariant.scan light the size-invariant validator: proposes object.split / object.remux* object.split heavy stream-copy parts of a large vod/clip as new private objects* object.remux heavy a stream-copy remux of a vod/clip as a new private object++ * 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* 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)@@@ -18,6 +20,8 @@ queue.register('thumbnail.regenerate', require('./thumbnail').spec);queue.register('invariant.scan', require('./invariant-scan').spec);queue.register('object.split', require('./derive').split);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('vod.duration.reconcile', require('./duration-reconcile').spec);queue.register('object.hash', require('./content-hash').spec);nnew file mode 100644@@@ -0,0 +1,141 @@++'use strict';++// object.waveform and object.sprite (WS-G task 13): preview variants made as jobs through the jobs API.++// Each makes a private `asset` object (PNG / JPEG) derived from the source and recorded as its variant;++// the sprite's layout (frames, interval, grid, tile size) is in its metadata and matches the image; a++// source with no audio has no waveform and one with no video no sprite (permanent failures); the source++// is untouched.++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-previews-'));++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; };++const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;++const size = (file) => { const r = spawnSync('ffprobe', ['-v', 'error', '-select_streams', 'v:0', '-show_entries', 'stream=width,height', '-of', 'csv=p=0', file], { encoding: 'utf8' }); return r.stdout.trim(); };++++(async () => {++ // The layout alone: at most one frame per 2 s, a full grid, 16:9 tiles.++ const { spriteLayout } = require('../server/jobs/previews');++ assert.deepStrictEqual(spriteLayout(6, { frames: 100, columns: 10, tile_width: 160 }), { count: 3, interval_seconds: 2, columns: 3, rows: 1, tile_width: 160, tile_height: 90 });++ assert.deepStrictEqual(spriteLayout(43083, { frames: 100, columns: 10, tile_width: 160 }), { count: 100, interval_seconds: 430.83, columns: 10, rows: 10, tile_width: 160, tile_height: 90 });++ assert.strictEqual(spriteLayout(1, { frames: 100, columns: 10, tile_width: 160 }).count, 1);++ if (!hasFfmpeg) { console.log('⚠️ ffmpeg not found: preview jobs 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 model = require('../server/objects/model');++ const queue = require('../server/jobs/queue');++ const worker = require('../server/jobs/worker');++ const express = require('express');++ const app = express();++ app.use(express.json());++ 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) };++ };++ worker.start();++++ // Sources: a 20 s VOD with sound, a silent clip file, and an audio-only file.++ const mk = (name, args) => { const f = path.join(process.env.VOD_PATH, name); const r = spawnSync('ffmpeg', ['-v', 'error', '-y', ...args, f]); assert.strictEqual(r.status, 0, String(r.stderr)); return f; };++ const av = mk('rec-20.mp4', ['-f', 'lavfi', '-i', 'testsrc=duration=20:size=320x180:rate=10', '-f', 'lavfi', '-i', 'sine=frequency=440:duration=20', '-c:v', 'mpeg4', '-g', '10', '-c:a', 'aac', '-shortest']);++ conn.prepare("INSERT INTO vods (id, app_id, user_id, title, file_path, file_size, is_public, duration_seconds) VALUES (20, 'live', 5, 'Twenty seconds', ?, ?, 1, 20)").run(av, fs.statSync(av).size);++ const vodObj = model.sync('vod', 20).id;++ const before = JSON.stringify([model.getObject(vodObj), model.listLocations(vodObj)]);++ const native = (file, mime) => {++ const id = model.createObject({ app_id: 'live', owner_user_id: 5, lifecycle_status: 'ready', kind: 'file', mime_type: mime, size_bytes: fs.statSync(file).size, metadata: { duration_seconds: 6 } });++ model.upsertLocation(id, { provider: 'local', key: file, state: 'present', size_bytes: fs.statSync(file).size, verified: true });++ return id;++ };++ const silent = native(mk('silent.mp4', ['-f', 'lavfi', '-i', 'testsrc=duration=6:size=320x180:rate=10', '-c:v', 'mpeg4']), 'video/mp4');++ const audio = native(mk('tone.m4a', ['-f', 'lavfi', '-i', 'sine=frequency=220:duration=6', '-c:a', 'aac']), 'audio/mp4');++++ const runJob = async (type, objectId, params) => {++ const r = await call('POST', '/api/v2/live/jobs', { body: { type, object_id: objectId, ...(params && { params }) } });++ assert.strictEqual(r.status, 202, JSON.stringify(r.body));++ const fin = await waitFor(() => ["succeeded", "failed"].includes(queue.get(r.body.job.id).status));++ assert.ok(fin, `${type} finishes: ${JSON.stringify(queue.jobPublic(queue.get(r.body.job.id)))}`);++ return queue.jobPublic(queue.get(r.body.job.id));++ };++++ // ── Waveform ──++ const w = await runJob('object.waveform', 'legacy:live:vod:20');++ assert.strictEqual(w.status, 'succeeded', JSON.stringify(w.error));++ const wo = model.getObject(w.result.object_id);++ assert.deepStrictEqual([wo.kind, wo.mime_type, wo.visibility, wo.lifecycle_status, wo.owner_user_id], ['asset', 'image/png', 'private', 'ready', 5]);++ const wloc = model.listLocations(wo.id)[0];++ assert.strictEqual(size(wloc.key), '1800,140', 'the default waveform size');++ assert.strictEqual(model.getVariant(vodObj, 'waveform').derived_object_id, wo.id);++ assert.ok(/-waveform\.png$/.test(model.parseJson(wo.metadata, {}).filename));++++ const w2 = await runJob('object.waveform', vodObj, { width: 400, height: 60 });++ assert.strictEqual(size(model.listLocations(w2.result.object_id)[0].key), '400,60');++ assert.strictEqual(model.getVariant(vodObj, 'waveform').derived_object_id, w2.result.object_id, 'a new waveform replaces the variant');++++ const ws = await runJob('object.waveform', silent);++ assert.strictEqual(ws.status, 'failed');++ assert.strictEqual(ws.error_code, 'no_audio', ws.error);++ assert.strictEqual(ws.attempts, 1, 'no retry: it is permanent');++++ // ── Sprite ──++ const s = await runJob('object.sprite', vodObj);++ assert.strictEqual(s.status, 'succeeded', JSON.stringify(s.error));++ assert.deepStrictEqual(s.result.sprite, { count: 10, interval_seconds: 2, columns: 10, rows: 1, tile_width: 160, tile_height: 90 });++ const so = model.getObject(s.result.object_id);++ assert.deepStrictEqual([so.kind, so.mime_type, so.visibility], ['asset', 'image/jpeg', 'private']);++ assert.strictEqual(size(model.listLocations(so.id)[0].key), '1600,90', 'the image is the layout: 10 × 160 by 1 × 90');++ assert.deepStrictEqual(model.parseJson(so.metadata, {}).sprite, s.result.sprite, 'the layout is in the metadata');++ assert.strictEqual(model.getVariant(vodObj, 'sprite').derived_object_id, so.id);++++ const s2 = await runJob('object.sprite', silent, { frames: 4, columns: 2, tile_width: 96 });++ assert.strictEqual(s2.status, 'succeeded', JSON.stringify(s2.error));++ assert.deepStrictEqual([s2.result.sprite.count, s2.result.sprite.rows], [3, 2], '6 s: 3 frames on a 2-column grid');++ assert.strictEqual(size(model.listLocations(s2.result.object_id)[0].key), '192,108');++++ const sa = await runJob('object.sprite', audio);++ assert.strictEqual(sa.status, 'failed');++ assert.strictEqual(sa.error_code, 'no_video', sa.error);++++ const bad = await call('POST', '/api/v2/live/jobs', { body: { type: 'object.sprite', object_id: vodObj, params: { frames: 1 } } });++ assert.strictEqual(bad.status, 400, 'params are checked when the job is asked for');++++ assert.strictEqual(JSON.stringify([model.getObject(vodObj), model.listLocations(vodObj)]), before, 'the source is untouched');++ assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'work directories are removed');++++ worker.stop();++ events._reset();++ server.close();++ stub.close();++ console.log('jobs-previews: all checks passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -50,7 +50,7 @@ server/
@@ -50,7 +50,7 @@ server/
change's transaction; payload queue.jobEvent: ids, state, ISO times,
change's transaction; payload queue.jobEvent: ids, state, ISO times,
has_result, never params/result/error text/creator), worker (light/heavy lanes, leases, retries,
has_result, never params/result/error text/creator), worker (light/heavy lanes, leases, retries,
cancellation), routes (/api/v2/:app/jobs), types: thumbnail.regenerate,
cancellation), routes (/api/v2/:app/jobs), types: thumbnail.regenerate,
- invariant.scan (proposes split/remux), object.split, object.remux,
+ invariant.scan (proposes split/remux), object.split, object.remux, object.waveform, object.sprite,
vod.finalize (orphans and failed finalizes, with backoff),
vod.finalize (orphans and failed finalizes, with backoff),
vod.duration.reconcile (stored vs measured durations, local and B2/R2),
vod.duration.reconcile (stored vs measured durations, local and B2/R2),
storage.orphans.scan (monthly storage orphan report, service-wide, report only)
storage.orphans.scan (monthly storage orphan report, service-wide, report only)
@@ -599,6 +599,8 @@ requeued (or failed when out of attempts). Handlers checkpoint progress and resu
@@ -599,6 +599,8 @@ 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 |
+| `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) |
| `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) |
| `vod.duration.reconcile` | heavy | One bounded batch of the tenant's finished VODs: stored duration vs a measurement of the real file, local or the B2/R2 copy (ffprobe over a presigned URL, ranged reads). `params { limit (1-200, 25), after_id?, apply (false), confirm_remote (false) }`; without `after_id` it walks the library from where the last run stopped. Repairs only confirmed, clearly wrong values when `apply`. Result `{ counts, range, next_after_id, report }`. See [Duration reconciliation](#duration-reconciliation) |
| `vod.duration.reconcile` | heavy | One bounded batch of the tenant's finished VODs: stored duration vs a measurement of the real file, local or the B2/R2 copy (ffprobe over a presigned URL, ranged reads). `params { limit (1-200, 25), after_id?, apply (false), confirm_remote (false) }`; without `after_id` it walks the library from where the last run stopped. Repairs only confirmed, clearly wrong values when `apply`. Result `{ counts, range, next_after_id, report }`. See [Duration reconciliation](#duration-reconciliation) |
| `storage.orphans.scan` | light | The [storage orphan report](#storage-orphan-report), service-wide, under the system app `_media`; report only. `params { limit (1-100000, 2000) }`. Result `{ totals, providers, summary, report }` (the report file under `<data>/reports/`). Scheduled every `MEDIA_ORPHAN_SCAN_DAYS`; tenants cannot enqueue it (403) |
| `storage.orphans.scan` | light | The [storage orphan report](#storage-orphan-report), service-wide, under the system app `_media`; report only. `params { limit (1-100000, 2000) }`. Result `{ totals, providers, summary, report }` (the report file under `<data>/reports/`). Scheduled every `MEDIA_ORPHAN_SCAN_DAYS`; tenants cannot enqueue it (403) |
@@ -50,7 +50,7 @@ function extFor(obj, sourcePath) {
@@ -50,7 +50,7 @@ function extFor(obj, sourcePath) {
}
}
function mimeFor(ext, obj) {
function mimeFor(ext, obj) {
- return { '.webm': 'video/webm', '.mp4': 'video/mp4', '.mkv': 'video/x-matroska', '.m4a': 'audio/mp4', '.ogg': 'video/ogg', '.mp3': 'audio/mpeg' }[ext]
+ return { '.webm': 'video/webm', '.mp4': 'video/mp4', '.mkv': 'video/x-matroska', '.m4a': 'audio/mp4', '.ogg': 'video/ogg', '.mp3': 'audio/mpeg', '.png': 'image/png', '.jpg': 'image/jpeg' }[ext]
|| obj.mime_type || 'application/octet-stream';
|| obj.mime_type || 'application/octet-stream';
}
}
@@ -168,7 +168,8 @@ function cleanupWork(jobId) {
@@ -168,7 +168,8 @@ function cleanupWork(jobId) {
* move the bytes under OBJECTS_PATH/<app>/<id>, verified local location, relationship; `alsoInTx`
* move the bytes under OBJECTS_PATH/<app>/<id>, verified local location, relationship; `alsoInTx`
* (the job checkpoint) commits with it. Returns the new object id.
* (the job checkpoint) commits with it. Returns the new object id.
*/
*/
-async function adopt({ src, file, ext, job, relation, metadata, variant, saveCheckpoint, checkpointFor }) {
+// kind/suffix: a derived image (waveform, sprite) is an `asset` named <source>-<suffix>.<ext>; parts and remuxes keep the source's kind.
+async function adopt({ src, file, ext, job, relation, metadata, variant, saveCheckpoint, checkpointFor, kind = null, suffix = null }) {
const size = fs.statSync(file).size;
const size = fs.statSync(file).size;
const hash = await sha256File(file);
const hash = await sha256File(file);
const md = model.parseJson(src.metadata, {});
const md = model.parseJson(src.metadata, {});
@@ -181,9 +182,9 @@ async function adopt({ src, file, ext, job, relation, metadata, variant, saveChe
@@ -181,9 +182,9 @@ async function adopt({ src, file, ext, job, relation, metadata, variant, saveChe
db.run(`INSERT INTO media_objects (id, app_id, namespace, kind, owner_subject, owner_app, owner_user_id, visibility, lifecycle_status,
db.run(`INSERT INTO media_objects (id, app_id, namespace, kind, owner_subject, owner_app, owner_user_id, visibility, lifecycle_status,
mime_type, size_bytes, content_hash, canonical_provider, canonical_key, legacy_ref, metadata)
mime_type, size_bytes, content_hash, canonical_provider, canonical_key, legacy_ref, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, 'private', 'ready', ?, ?, ?, 'local', ?, NULL, ?)`,
VALUES (?, ?, ?, ?, ?, ?, ?, 'private', 'ready', ?, ?, ?, 'local', ?, NULL, ?)`,
- [id, src.app_id, src.namespace || src.app_id, src.kind, src.owner_subject || null, src.owner_app || src.app_id, src.owner_user_id ?? null,
+ [id, src.app_id, src.namespace || src.app_id, kind || src.kind, src.owner_subject || null, src.owner_app || src.app_id, src.owner_user_id ?? null,
mimeFor(ext, src), size, hash, dest,
mimeFor(ext, src), size, hash, dest,
- JSON.stringify({ title: md.title || null, derived_from: src.id, job_id: job.id, filename: `${src.id}${metadata.part ? `-part${metadata.part}` : '-remux'}${ext}`, ...metadata })]);
+ JSON.stringify({ title: md.title || null, derived_from: src.id, job_id: job.id, filename: `${src.id}${suffix ? `-${suffix}` : metadata.part ? `-part${metadata.part}` : '-remux'}${ext}`, ...metadata })]);
model.upsertLocation(id, { provider: 'local', key: dest, state: 'present', size_bytes: size, checksum: hash, verified: true });
model.upsertLocation(id, { provider: 'local', key: dest, state: 'present', size_bytes: size, checksum: hash, verified: true });
model.setRelationship(id, 'derived_from', src.id, { job_id: job.id, ...relation });
model.setRelationship(id, 'derived_from', src.id, { job_id: job.id, ...relation });
if (variant) model.setVariant(src.id, variant, id, `${job.type}@1`);
if (variant) model.setVariant(src.id, variant, id, `${job.type}@1`);
@@ -307,4 +308,6 @@ module.exports = {
@@ -307,4 +308,6 @@ module.exports = {
split: { lane: 'heavy', maxAttempts: 3, timeoutMs: 6 * 3600 * 1000, needsObject: true, validate: validateSplit, run: runSplit },
split: { lane: 'heavy', maxAttempts: 3, timeoutMs: 6 * 3600 * 1000, needsObject: true, validate: validateSplit, run: runSplit },
remux: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateRemux, run: runRemux },
remux: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateRemux, run: runRemux },
resolveSource, isMediaObject, extFor,
resolveSource, isMediaObject, extFor,
+ // for the preview jobs (previews.js)
+ loadSource, adopt, ffmpeg, inputArgs, probeDuration, workDir, cleanupWork, existing, checkRoom,
};
};
new file mode 100644
new file mode 100644
@@ -0,0 +1,155 @@
@@ -0,0 +1,155 @@
+/**
+ * OpenVibe.Media — job types object.waveform and object.sprite (heavy lane; roadmap WS-G task 13)
+ *
+ * Preview variants of a ready vod/clip (or a native video/audio object). Like object.remux they never
+ * touch the source: each output is a new private `asset` object in the same tenant, owned like the
+ * source, with a verified local copy, a `derived_from` relationship and the source's variant of that name
+ * (GET /api/v2/objects/:id lists variants).
+ *
+ * object.waveform params { width?, height? } a PNG of the audio's waveform (ffmpeg showwavespic), 1800×140
+ * by default. It decodes the whole audio track, so a remote source over
+ * MEDIA_PREVIEW_REMOTE_MAX_MB (2048) is refused: the egress would cost more than the preview.
+ * result { source_id, object_id, size_bytes, width, height, method: 'waveform' }
+ * object.sprite params { frames?, columns?, tile_width? } a JPEG sheet of evenly spaced frames for
+ * seek previews: 100 frames by default (at most one per 2 s), 10 columns, 160 px wide
+ * tiles. Each frame is its own fast seek, so a long remote VOD is never decoded end to end.
+ * The layout is in the object's metadata (sprite: { interval_seconds, count, columns,
+ * rows, tile_width, tile_height }): frame i covers [i × interval, (i + 1) × interval) and
+ * sits at column i % columns, row ⌊i / columns⌋.
+ * result { source_id, object_id, size_bytes, sprite, method: 'sprite' }
+ *
+ * Refused before any work: a source that is not ready or not a video/audio object, a waveform of a source
+ * with no audio or a sprite of one with no video (permanent), not enough disk or quota (as the other
+ * derive jobs).
+ */
+'use strict';
+const fs = require('fs');
+const path = require('path');
+const { JobError } = require('./queue');
+const d = require('./derive');
+
+const MB = 1024 * 1024;
+const REMOTE_MAX_BYTES = (Number(process.env.MEDIA_PREVIEW_REMOTE_MAX_MB) || 2048) * MB;
+const int = (v, dflt, min, max, name) => {
+ if (v == null) return dflt;
+ const n = Number(v);
+ if (!Number.isInteger(n) || n < min || n > max) throw new JobError('media.job.invalid', `params.${name} must be an integer from ${min} to ${max}`, { permanent: true });
+ return n;
+};
+
+/** Which streams the source has: { audio, video, duration }. */
+async function streamsOf(input) {
+ const info = await require('../vod/media-tools').probeVodInfo(input).catch(() => null);
+ const streams = (info && info.streams) || [];
+ // An unreadable probe says nothing: ffmpeg itself decides then.
+ if (!streams.length) return { audio: null, video: null, duration: Number(info && info.duration) || 0 };
+ return { audio: streams.some((s) => s.codec_type === 'audio'), video: streams.some((s) => s.codec_type === 'video'), duration: Number(info.duration) || 0 };
+}
+
+// ── object.waveform ─────────────────────────────────────────
+
+function validateWaveform({ obj, params }) {
+ if (!obj) throw new JobError('media.job.invalid', 'object.waveform needs object_id', { permanent: true });
+ if (!d.isMediaObject(obj)) throw new JobError('media.job.invalid', 'Only vod/clip objects (or video/audio objects) have a waveform', { permanent: true });
+ const p = params || {};
+ return { width: int(p.width, 1800, 200, 4000, 'width'), height: int(p.height, 140, 40, 600, 'height') };
+}
+
+async function runWaveform(job, ctx) {
+ const src = d.loadSource(job);
+ const p = validateWaveform({ obj: src, params: job.params });
+ if (ctx.checkpoint && ctx.checkpoint.object_id) { d.cleanupWork(job.id); return { source_id: src.id, method: 'waveform', ...ctx.checkpoint }; }
+ const source = await d.resolveSource(src);
+ if (!source) throw new JobError('media_unavailable', 'The source bytes are unavailable (no local file and no cloud copy)', { permanent: true });
+ if (source.remote && Number(src.size_bytes) > REMOTE_MAX_BYTES) {
+ throw new JobError('too_large', `A waveform decodes the whole audio track; this source is only in cloud storage and larger than ${REMOTE_MAX_BYTES / MB} MB`, { permanent: true });
+ }
+ d.checkRoom(src, 8 * MB);
+ const out = path.join(d.workDir(job.id), 'waveform.png');
+ const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', ...d.inputArgs(source), '-i', source.input,
+ '-filter_complex', `[0:a:0]aformat=channel_layouts=mono,showwavespic=s=${p.width}x${p.height}:colors=#60a5fa:scale=sqrt[w]`, '-map', '[w]', '-frames:v', '1', out],
+ { signal: ctx.signal, timeoutMs: 2 * 3600 * 1000 });
+ if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted');
+ if (!r.ok || !d.existing(out) || fs.statSync(out).size === 0) {
+ d.cleanupWork(job.id);
+ const noAudio = /matches no streams|0:a:0/i.test(r.stderr);
+ throw new JobError(noAudio ? 'no_audio' : 'ffmpeg_failed', noAudio ? 'The source has no audio track' : `ffmpeg exited ${r.code}: ${String(r.stderr).trim().split('\n').slice(-2).join(' ').slice(0, 300)}`, { permanent: noAudio });
+ }
+ const made = await d.adopt({
+ src, file: out, ext: '.png', job, variant: 'waveform', kind: 'asset', suffix: 'waveform',
+ relation: { method: 'waveform', width: p.width, height: p.height },
+ metadata: { waveform: { width: p.width, height: p.height } },
+ saveCheckpoint: ctx.saveCheckpoint,
+ checkpointFor: (id, size) => ({ object_id: id, size_bytes: size, width: p.width, height: p.height }),
+ });
+ d.cleanupWork(job.id);
+ return { source_id: src.id, method: 'waveform', object_id: made.id, size_bytes: made.size, width: p.width, height: p.height };
+}
+
+// ── object.sprite ───────────────────────────────────────────
+
+function validateSprite({ obj, params }) {
+ if (!obj) throw new JobError('media.job.invalid', 'object.sprite needs object_id', { permanent: true });
+ if (!d.isMediaObject(obj)) throw new JobError('media.job.invalid', 'Only vod/clip objects (or video objects) have a preview sprite', { permanent: true });
+ const p = params || {};
+ return { frames: int(p.frames, 100, 4, 400, 'frames'), columns: int(p.columns, 10, 1, 40, 'columns'), tile_width: int(p.tile_width, 160, 64, 480, 'tile_width') };
+}
+
+/** The layout for a duration: frames at most one per 2 s. */
+function spriteLayout(duration, { frames, columns, tile_width }) {
+ const count = Math.max(1, Math.min(frames, Math.floor(duration / 2) || 1));
+ const interval = duration > 0 ? duration / count : 0;
+ const cols = Math.min(columns, count);
+ return { count, interval_seconds: Math.round(interval * 1000) / 1000, columns: cols, rows: Math.ceil(count / cols), tile_width, tile_height: Math.round(tile_width * 9 / 16 / 2) * 2 };
+}
+
+async function runSprite(job, ctx) {
+ const src = d.loadSource(job);
+ const p = validateSprite({ obj: src, params: job.params });
+ if (ctx.checkpoint && ctx.checkpoint.object_id) { d.cleanupWork(job.id); return { source_id: src.id, method: 'sprite', ...ctx.checkpoint }; }
+ const source = await d.resolveSource(src);
+ if (!source) throw new JobError('media_unavailable', 'The source bytes are unavailable (no local file and no cloud copy)', { permanent: true });
+ d.checkRoom(src, 16 * MB);
+ const meta = await streamsOf(source.input);
+ const md = require('../objects/model').parseJson(src.metadata, {});
+ const duration = meta.duration || Number(md.duration_seconds) || 0;
+ if (!(duration > 0)) throw new JobError('no_duration', 'The source has no known duration to spread frames over', { permanent: true });
+ if (meta.video === false) throw new JobError('no_video', 'The source has no video track', { permanent: true });
+ const layout = spriteLayout(duration, p);
+ const dir = d.workDir(job.id);
+ const tile = `scale=${layout.tile_width}:${layout.tile_height}:force_original_aspect_ratio=decrease,pad=${layout.tile_width}:${layout.tile_height}:(ow-iw)/2:(oh-ih)/2`;
+ for (let i = 0; i < layout.count; i++) {
+ if (ctx.signal.aborted) throw ctx.signal.reason || new Error('aborted');
+ const frame = path.join(dir, `f${String(i).padStart(4, '0')}.jpg`);
+ if (d.existing(frame)) continue;
+ const at = Math.min(duration - 0.1, i * layout.interval_seconds + layout.interval_seconds / 2);
+ const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', ...d.inputArgs(source), '-ss', String(Math.max(0, at).toFixed(3)), '-i', source.input,
+ '-frames:v', '1', '-vf', tile, '-q:v', '5', frame], { signal: ctx.signal, timeoutMs: 5 * 60 * 1000 });
+ // A frame that cannot be read (a gap in a recording) is black rather than failing the sheet.
+ if (!r.ok || !d.existing(frame)) {
+ await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-f', 'lavfi', '-i', `color=c=black:s=${layout.tile_width}x${layout.tile_height}`, '-frames:v', '1', frame], { signal: ctx.signal, timeoutMs: 60000 });
+ }
+ }
+ const out = path.join(dir, 'sprite.jpg');
+ const r = await d.ffmpeg(['-y', '-nostdin', '-v', 'error', '-framerate', '1', '-i', path.join(dir, 'f%04d.jpg'),
+ '-vf', `tile=${layout.columns}x${layout.rows}`, '-frames:v', '1', '-q:v', '4', out], { signal: ctx.signal, timeoutMs: 10 * 60 * 1000 });
+ if (!r.ok || !d.existing(out) || fs.statSync(out).size === 0) {
+ d.cleanupWork(job.id);
+ throw new JobError('ffmpeg_failed', `ffmpeg exited ${r.code}: ${String(r.stderr).trim().split('\n').slice(-2).join(' ').slice(0, 300)}`);
+ }
+ const made = await d.adopt({
+ src, file: out, ext: '.jpg', job, variant: 'sprite', kind: 'asset', suffix: 'sprite',
+ relation: { method: 'sprite', ...layout },
+ metadata: { sprite: layout, source_duration_seconds: duration },
+ saveCheckpoint: ctx.saveCheckpoint,
+ checkpointFor: (id, size) => ({ object_id: id, size_bytes: size, sprite: layout }),
+ });
+ d.cleanupWork(job.id);
+ return { source_id: src.id, method: 'sprite', object_id: made.id, size_bytes: made.size, sprite: layout };
+}
+
+module.exports = {
+ waveform: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateWaveform, run: runWaveform },
+ sprite: { lane: 'heavy', maxAttempts: 3, timeoutMs: 3 * 3600 * 1000, needsObject: true, validate: validateSprite, run: runSprite },
+ spriteLayout,
+};
@@ -5,6 +5,8 @@
@@ -5,6 +5,8 @@
* invariant.scan light the size-invariant validator: proposes object.split / object.remux
* invariant.scan light the size-invariant validator: proposes object.split / object.remux
* object.split heavy stream-copy parts of a large vod/clip as new private objects
* object.split heavy stream-copy parts of a large vod/clip as new private objects
* object.remux heavy a stream-copy remux of a vod/clip as a new private object
* object.remux heavy a stream-copy remux of a vod/clip as a new private object
+ * 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
* vod.finalize finalize finalize a recording whose finalize failed or never ran (orphans), with backoff
* 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)
@@ -18,6 +20,8 @@ queue.register('thumbnail.regenerate', require('./thumbnail').spec);
@@ -18,6 +20,8 @@ queue.register('thumbnail.regenerate', require('./thumbnail').spec);
queue.register('invariant.scan', require('./invariant-scan').spec);
queue.register('invariant.scan', require('./invariant-scan').spec);
queue.register('object.split', require('./derive').split);
queue.register('object.split', require('./derive').split);
queue.register('object.remux', require('./derive').remux);
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('vod.finalize', require('./vod-finalize').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);
new file mode 100644
new file mode 100644
@@ -0,0 +1,141 @@
@@ -0,0 +1,141 @@
+'use strict';
+// object.waveform and object.sprite (WS-G task 13): preview variants made as jobs through the jobs API.
+// Each makes a private `asset` object (PNG / JPEG) derived from the source and recorded as its variant;
+// the sprite's layout (frames, interval, grid, tile size) is in its metadata and matches the image; a
+// source with no audio has no waveform and one with no video no sprite (permanent failures); the source
+// is untouched.
+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-previews-'));
+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; };
+const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;
+const size = (file) => { const r = spawnSync('ffprobe', ['-v', 'error', '-select_streams', 'v:0', '-show_entries', 'stream=width,height', '-of', 'csv=p=0', file], { encoding: 'utf8' }); return r.stdout.trim(); };
+
+(async () => {
+ // The layout alone: at most one frame per 2 s, a full grid, 16:9 tiles.
+ const { spriteLayout } = require('../server/jobs/previews');
+ assert.deepStrictEqual(spriteLayout(6, { frames: 100, columns: 10, tile_width: 160 }), { count: 3, interval_seconds: 2, columns: 3, rows: 1, tile_width: 160, tile_height: 90 });
+ assert.deepStrictEqual(spriteLayout(43083, { frames: 100, columns: 10, tile_width: 160 }), { count: 100, interval_seconds: 430.83, columns: 10, rows: 10, tile_width: 160, tile_height: 90 });
+ assert.strictEqual(spriteLayout(1, { frames: 100, columns: 10, tile_width: 160 }).count, 1);
+ if (!hasFfmpeg) { console.log('⚠️ ffmpeg not found: preview jobs 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 model = require('../server/objects/model');
+ const queue = require('../server/jobs/queue');
+ const worker = require('../server/jobs/worker');
+ const express = require('express');
+ const app = express();
+ app.use(express.json());
+ 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) };
+ };
+ worker.start();
+
+ // Sources: a 20 s VOD with sound, a silent clip file, and an audio-only file.
+ const mk = (name, args) => { const f = path.join(process.env.VOD_PATH, name); const r = spawnSync('ffmpeg', ['-v', 'error', '-y', ...args, f]); assert.strictEqual(r.status, 0, String(r.stderr)); return f; };
+ const av = mk('rec-20.mp4', ['-f', 'lavfi', '-i', 'testsrc=duration=20:size=320x180:rate=10', '-f', 'lavfi', '-i', 'sine=frequency=440:duration=20', '-c:v', 'mpeg4', '-g', '10', '-c:a', 'aac', '-shortest']);
+ conn.prepare("INSERT INTO vods (id, app_id, user_id, title, file_path, file_size, is_public, duration_seconds) VALUES (20, 'live', 5, 'Twenty seconds', ?, ?, 1, 20)").run(av, fs.statSync(av).size);
+ const vodObj = model.sync('vod', 20).id;
+ const before = JSON.stringify([model.getObject(vodObj), model.listLocations(vodObj)]);
+ const native = (file, mime) => {
+ const id = model.createObject({ app_id: 'live', owner_user_id: 5, lifecycle_status: 'ready', kind: 'file', mime_type: mime, size_bytes: fs.statSync(file).size, metadata: { duration_seconds: 6 } });
+ model.upsertLocation(id, { provider: 'local', key: file, state: 'present', size_bytes: fs.statSync(file).size, verified: true });
+ return id;
+ };
+ const silent = native(mk('silent.mp4', ['-f', 'lavfi', '-i', 'testsrc=duration=6:size=320x180:rate=10', '-c:v', 'mpeg4']), 'video/mp4');
+ const audio = native(mk('tone.m4a', ['-f', 'lavfi', '-i', 'sine=frequency=220:duration=6', '-c:a', 'aac']), 'audio/mp4');
+
+ const runJob = async (type, objectId, params) => {
+ const r = await call('POST', '/api/v2/live/jobs', { body: { type, object_id: objectId, ...(params && { params }) } });
+ assert.strictEqual(r.status, 202, JSON.stringify(r.body));
+ const fin = await waitFor(() => ["succeeded", "failed"].includes(queue.get(r.body.job.id).status));
+ assert.ok(fin, `${type} finishes: ${JSON.stringify(queue.jobPublic(queue.get(r.body.job.id)))}`);
+ return queue.jobPublic(queue.get(r.body.job.id));
+ };
+
+ // ── Waveform ──
+ const w = await runJob('object.waveform', 'legacy:live:vod:20');
+ assert.strictEqual(w.status, 'succeeded', JSON.stringify(w.error));
+ const wo = model.getObject(w.result.object_id);
+ assert.deepStrictEqual([wo.kind, wo.mime_type, wo.visibility, wo.lifecycle_status, wo.owner_user_id], ['asset', 'image/png', 'private', 'ready', 5]);
+ const wloc = model.listLocations(wo.id)[0];
+ assert.strictEqual(size(wloc.key), '1800,140', 'the default waveform size');
+ assert.strictEqual(model.getVariant(vodObj, 'waveform').derived_object_id, wo.id);
+ assert.ok(/-waveform\.png$/.test(model.parseJson(wo.metadata, {}).filename));
+
+ const w2 = await runJob('object.waveform', vodObj, { width: 400, height: 60 });
+ assert.strictEqual(size(model.listLocations(w2.result.object_id)[0].key), '400,60');
+ assert.strictEqual(model.getVariant(vodObj, 'waveform').derived_object_id, w2.result.object_id, 'a new waveform replaces the variant');
+
+ const ws = await runJob('object.waveform', silent);
+ assert.strictEqual(ws.status, 'failed');
+ assert.strictEqual(ws.error_code, 'no_audio', ws.error);
+ assert.strictEqual(ws.attempts, 1, 'no retry: it is permanent');
+
+ // ── Sprite ──
+ const s = await runJob('object.sprite', vodObj);
+ assert.strictEqual(s.status, 'succeeded', JSON.stringify(s.error));
+ assert.deepStrictEqual(s.result.sprite, { count: 10, interval_seconds: 2, columns: 10, rows: 1, tile_width: 160, tile_height: 90 });
+ const so = model.getObject(s.result.object_id);
+ assert.deepStrictEqual([so.kind, so.mime_type, so.visibility], ['asset', 'image/jpeg', 'private']);
+ assert.strictEqual(size(model.listLocations(so.id)[0].key), '1600,90', 'the image is the layout: 10 × 160 by 1 × 90');
+ assert.deepStrictEqual(model.parseJson(so.metadata, {}).sprite, s.result.sprite, 'the layout is in the metadata');
+ assert.strictEqual(model.getVariant(vodObj, 'sprite').derived_object_id, so.id);
+
+ const s2 = await runJob('object.sprite', silent, { frames: 4, columns: 2, tile_width: 96 });
+ assert.strictEqual(s2.status, 'succeeded', JSON.stringify(s2.error));
+ assert.deepStrictEqual([s2.result.sprite.count, s2.result.sprite.rows], [3, 2], '6 s: 3 frames on a 2-column grid');
+ assert.strictEqual(size(model.listLocations(s2.result.object_id)[0].key), '192,108');
+
+ const sa = await runJob('object.sprite', audio);
+ assert.strictEqual(sa.status, 'failed');
+ assert.strictEqual(sa.error_code, 'no_video', sa.error);
+
+ const bad = await call('POST', '/api/v2/live/jobs', { body: { type: 'object.sprite', object_id: vodObj, params: { frames: 1 } } });
+ assert.strictEqual(bad.status, 400, 'params are checked when the job is asked for');
+
+ assert.strictEqual(JSON.stringify([model.getObject(vodObj), model.listLocations(vodObj)]), before, 'the source is untouched');
+ assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'work directories are removed');
+
+ worker.stop();
+ events._reset();
+ server.close();
+ stub.close();
+ console.log('jobs-previews: all checks passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });