openvibers/openvibe.ai Commits
media.analyze reads a recording where it lies instead of downloading it (roadmap WS-O tasks 2 and 5: the ranged fetch Live's adoption waited for).
Changes
@@@ -135,6 +135,9 @@ function load(env = process.env) {sceneThreshold: float(env.AI_MEDIA_SCENE_THRESHOLD, 10),// Speech-to-text in windows: each call stays well inside the whisper provider's timeout (10 min).sttWindowSec: Math.max(10, int(env.AI_MEDIA_STT_WINDOW_SEC, 600)),++ // Read the recording where it lies through the loopback reader (server/media-proxy.js), never downloading++ // it; AI_MEDIA_STREAM=0 downloads it first (then the disk cap applies).++ stream: bool(env.AI_MEDIA_STREAM, true),ffmpeg: env.FFMPEG_BIN || 'ffmpeg',ffprobe: env.FFPROBE_BIN || 'ffprobe',},@@@ -157,6 +157,50 @@ function createFetcher(config, { transport = null } = {}) {throw new AiError(502, 'source.unavailable', 'too many redirects');}++ /**++ * One upstream response for a media reader (server/media-proxy.js): the same URL rules as fetchUrl, redirects++ * judged and followed here (the reader never sees them), a Range header passed through, the body left as a stream.++ * → { res } (an http.IncomingMessage with status 200/206/416), or throws AiError.++ */++ async function openStream(raw, { method = 'GET', range = null, signal, timeoutMs = m.timeoutMs } = {}) {++ let target = judge(raw);++ for (let hop = 0; hop < 4; hop++) {++ const res = await new Promise((resolve, reject) => {++ const { url, internal } = target;++ const mod = url.protocol === 'https:' ? https : http;++ const req = mod.request(url, {++ method: method === 'HEAD' ? 'HEAD' : 'GET',++ headers: { 'User-Agent': 'OpenVibe.AI/0.1 (+https://ai.openvibe.network)', Accept: '*/*', ...(range ? { Range: range } : {}) },++ lookup: internal ? undefined : safeLookup,++ timeout: timeoutMs,++ });++ const onAbort = () => { req.destroy(); reject(new AiError(409, 'run.cancelled', 'cancelled')); };++ if (signal) { if (signal.aborted) return onAbort(); signal.addEventListener('abort', onAbort, { once: true }); }++ req.on('timeout', () => { req.destroy(); reject(new AiError(502, 'source.unavailable', 'fetch timed out')); });++ req.on('error', (err) => reject(err.code === 'EADDRNOTPUBLIC' ? new AiError(422, 'fetch.refused', err.message) : new AiError(502, 'source.unavailable', `fetch failed: ${err.message}`)));++ req.on('response', (r) => { if (signal) signal.removeEventListener('abort', onAbort); resolve(r); });++ req.end();++ });++ if (res.statusCode >= 300 && res.statusCode < 400 && res.headers.location) {++ res.resume();++ if (hop === 3) throw new AiError(502, 'source.unavailable', 'too many redirects');++ target = follow(target, new URL(res.headers.location, target.url).toString());++ continue;++ }++ if (![200, 206, 416].includes(res.statusCode)) { res.resume(); throw new AiError(502, 'source.unavailable', `source answered HTTP ${res.statusCode}`); }++ return { res };++ }++ throw new AiError(502, 'source.unavailable', 'too many redirects');++ }++++ /** A media input's URL (media_url, or a MediaRef). */++ function mediaUrlOf(input) {++ const url = input.media_url || (input.media ? mediaRefUrl(input.media) : null);++ if (!url) throw new AiError(422, 'input.invalid', 'media needs media_url or media (MediaRef)');++ judge(url);++ return url;++ }++/** A MediaRef -> the public URL of that object (legacy kinds only for now). */function mediaRefUrl(ref) {const id = ref && ref.media_id;@@@ -210,7 +254,7 @@ function createFetcher(config, { transport = null } = {}) {return { file, url };}-- return { judge, follow, fetchUrl, mediaRefUrl, loadImage, loadMediaToFile, hostAllowed: (h) => hostAllowed(h, m.allowHosts) };++ return { judge, follow, fetchUrl, openStream, mediaUrlOf, mediaRefUrl, loadImage, loadMediaToFile, hostAllowed: (h) => hostAllowed(h, m.allowHosts) };}/** Any image -> downscaled JPEG (sharp, optional). Falls back to the original bytes. */nnew file mode 100644@@@ -0,0 +1,71 @@++'use strict';++/**++ * A loopback reader for media the analysis streams instead of downloading (roadmap WS-O tasks 2 and 5). ffprobe and++ * ffmpeg read `http://127.0.0.1:<port>/m/<token>`; this server answers each request (GET or HEAD, with any Range) by++ * asking the real URL through the fetcher's rules (fetcher.openStream: the allow-list, public addresses checked at++ * connect time, redirects judged here and never shown to ffmpeg). So a long recording is read where it lies: the++ * signal pass streams it once, and each speech-to-text window seeks to its start with a range request. Nothing is++ * written to disk.++ *++ * const handle = proxy.register(url) // judged now; valid until release() or 6 hours++ * handle.url // what ffmpeg reads++ * handle.release()++ *++ * The server listens on 127.0.0.1 only, on a random port, from the first registration, and serves only registered++ * tokens (128 random bits each).++ */++const http = require('http');++const crypto = require('crypto');++++const TTL_MS = 6 * 3600 * 1000;++++function createMediaProxy({ fetcher, log = console, clock = { now: () => Date.now() } }) {++ const entries = new Map(); // token -> { url, expires }++ let server = null;++ let ready = null;++++ function start() {++ if (ready) return ready;++ server = http.createServer((req, res) => {++ const m = /^\/m\/([0-9a-f]{32})$/.exec(String(req.url || ''));++ const e = m && entries.get(m[1]);++ if (!e || e.expires < clock.now()) { res.writeHead(404).end(); return; }++ if (req.method !== 'GET' && req.method !== 'HEAD') { res.writeHead(405).end(); return; }++ const range = typeof req.headers.range === 'string' && /^bytes=\d*-\d*$/.test(req.headers.range) ? req.headers.range : null;++ const ac = new AbortController();++ req.on('close', () => ac.abort());++ fetcher.openStream(e.url, { method: req.method, range, signal: ac.signal, timeoutMs: 60000 }).then(({ res: up }) => {++ const headers = {};++ for (const k of ['content-type', 'content-length', 'content-range', 'accept-ranges', 'last-modified', 'etag']) if (up.headers[k]) headers[k] = up.headers[k];++ res.writeHead(up.statusCode, headers);++ if (req.method === 'HEAD') { up.resume(); res.end(); return; }++ up.pipe(res);++ up.on('error', () => res.destroy());++ }, (err) => {++ if (ac.signal.aborted) return;++ log.warn && log.warn(`[media-proxy] ${err.code || 'error'}: ${String(err.detail || err.message).slice(0, 200)}`);++ if (!res.headersSent) res.writeHead(err.status === 422 ? 403 : 502).end();++ else res.destroy();++ });++ });++ server.keepAliveTimeout = 5000;++ ready = new Promise((resolve, reject) => { server.once('error', reject); server.listen(0, '127.0.0.1', () => { server.unref(); resolve(server.address().port); }); });++ return ready;++ }++++ /** Judge `url` now and return a local URL for it. */++ async function register(url) {++ fetcher.judge(url);++ const port = await start();++ const token = crypto.randomBytes(16).toString('hex');++ entries.set(token, { url, expires: clock.now() + TTL_MS });++ for (const [k, v] of entries) if (v.expires < clock.now()) entries.delete(k);++ return { url: `http://127.0.0.1:${port}/m/${token}`, release: () => entries.delete(token) };++ }++++ function close() { entries.clear(); if (server) server.close(); server = null; ready = null; }++++ return { register, close, _entries: entries };++}++++module.exports = { createMediaProxy };@@@ -184,6 +184,9 @@ function extractive(facts, highlights) {}function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spawnImpl = spawn, log = console }) {++ // Long recordings are streamed, not downloaded (WS-O task 2): ffmpeg reads them through a loopback reader.++ let proxy = null;++ const reader = () => proxy || (proxy = require('../media-proxy').createMediaProxy({ fetcher, log }));const ma = config.mediaAnalysis || {};const ffmpeg = ma.ffmpeg || 'ffmpeg';const ffprobe = ma.ffprobe || 'ffprobe';@@@ -318,8 +321,15 @@ function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spaasync function analyse(stepDef, input, wf, runRow, ctx) {const gaps = [];-- const dir = work();-- const { file } = await fetcher.loadMediaToFile(input, { signal: ctx.signal, maxBytes: roomFor(dir), timeoutMs: ma.fetchTimeoutMs, dir });++ const streamed = ma.stream !== false;++ let file; let handle = null;++ if (streamed) {++ handle = await reader().register(fetcher.mediaUrlOf(input));++ file = handle.url;++ } else {++ const dir = work();++ ({ file } = await fetcher.loadMediaToFile(input, { signal: ctx.signal, maxBytes: roomFor(dir), timeoutMs: ma.fetchTimeoutMs, dir }));++ }try {const info = await probe(file, ctx.signal);if (!info.video && !info.audio) throw new AiError(422, 'media.unreadable', 'the file has no audio or video stream');@@@ -375,7 +385,8 @@ function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spa// The run's provider: the model that wrote the overview, else the speech-to-text engine that ran.return { output, exec: ov.exec || sttExec || null, route: ov.route || null };} finally {-- try { fs.unlinkSync(file); } catch { /* */ }++ if (handle) handle.release();++ else { try { fs.unlinkSync(file); } catch { /* */ } }}}@@@ -120,9 +120,9 @@ t.test('a real run: signals, scenes, highlights with evidence, and the local modassert.match(prompt, /Loudest moments: 0:1[6-8]/);assert.strictEqual(local.last.model, 'qwen-test');assert.ok(!r.body.run.output.gaps.some((g) => /local model/.test(g)));++ // Streamed through the loopback reader (the default): nothing was written to disk.const work = path.join(h.dir, 'media-tmp');-- assert.ok(fs.existsSync(work), 'the work directory is beside the database (not a tmpfs /tmp)');-- assert.deepStrictEqual(fs.readdirSync(work).filter((f) => f.startsWith('openvibe-ai-media-')), [], 'the downloaded file is removed');++ assert.deepStrictEqual(fs.existsSync(work) ? fs.readdirSync(work).filter((f) => f.startsWith('openvibe-ai-media-')) : [], [], 'nothing downloaded');} finally { await h.stop(); await local.close(); }});@@@ -179,7 +179,7 @@ t.test('too little free disk refuses before downloading', async () => {const counting = nodeHttp.createServer((req, res) => { fetched++; res.writeHead(200, { 'Content-Type': 'video/mp4' }); res.end(FIXTURE); });await new Promise((r) => counting.listen(0, '127.0.0.1', r));const base = `http://127.0.0.1:${counting.address().port}`;-- const h = await boot({ env: { OV_MEDIA_INTERNAL_URL: base, AI_MEDIA_ANALYSIS_DISK_RESERVE_BYTES: String(1024 ** 5) } });++ const h = await boot({ env: { OV_MEDIA_INTERNAL_URL: base, AI_MEDIA_ANALYSIS_DISK_RESERVE_BYTES: String(1024 ** 5), AI_MEDIA_STREAM: '0' } });try {const r = await request(h.base, 'POST', '/api/v1/runs?wait=20000', { tok, body: { workflow: 'media.analyze', input: { media_url: `${base}/fixture.mp4` } } });assert.strictEqual(r.body.run.status, 'failed');nnew file mode 100644@@@ -0,0 +1,58 @@++'use strict';++// The loopback reader media.analyze streams through (server/media-proxy.js; roadmap WS-O task 2): ffmpeg reads a++// registered token and gets the real URL's bytes, Range requests pass through (206 with Content-Range), HEAD answers++// headers only, an unknown or released token is 404, and a redirect the fetcher's rules refuse (a public host++// sending it to the internal Media origin) never reaches the reader: 403.++const assert = require('assert');++const nodeHttp = require('http');++const { suite } = require('./helpers');++const { load } = require('../server/config');++const { createFetcher } = require('../server/fetcher');++const { createMediaProxy } = require('../server/media-proxy');++++const t = suite('media-proxy');++const BODY = Buffer.from('0123456789abcdefghijklmnopqrstuvwxyz');++let upstream; let base; let proxy;++++t.test('boot an internal Media stand-in', async () => {++ upstream = nodeHttp.createServer((req, res) => {++ if (req.url === '/redirect-out') { res.writeHead(302, { Location: 'https://evil.example/x' }); return res.end(); }++ const m = /^bytes=(\d+)-(\d*)$/.exec(req.headers.range || '');++ if (m) {++ const start = Number(m[1]); const end = m[2] ? Number(m[2]) : BODY.length - 1;++ res.writeHead(206, { 'Content-Type': 'video/mp4', 'Content-Length': end - start + 1, 'Content-Range': `bytes ${start}-${end}/${BODY.length}`, 'Accept-Ranges': 'bytes' });++ return res.end(req.method === 'HEAD' ? undefined : BODY.subarray(start, end + 1));++ }++ res.writeHead(200, { 'Content-Type': 'video/mp4', 'Content-Length': BODY.length, 'Accept-Ranges': 'bytes' });++ res.end(req.method === 'HEAD' ? undefined : BODY);++ });++ await new Promise((r) => upstream.listen(0, '127.0.0.1', r));++ base = `http://127.0.0.1:${upstream.address().port}`;++ proxy = createMediaProxy({ fetcher: createFetcher(load({ NODE_ENV: 'test', OV_MEDIA_INTERNAL_URL: base })), log: { warn() {} } });++});++++t.test('whole reads, ranges, HEAD and unknown tokens', async () => {++ const h = await proxy.register(`${base}/v/1`);++ assert.match(h.url, /^http:\/\/127\.0\.0\.1:\d+\/m\/[0-9a-f]{32}$/);++ let r = await fetch(h.url);++ assert.strictEqual(r.status, 200);++ assert.deepStrictEqual(Buffer.from(await r.arrayBuffer()), BODY);++ r = await fetch(h.url, { headers: { Range: 'bytes=10-19' } });++ assert.deepStrictEqual([r.status, r.headers.get('content-range'), Buffer.from(await r.arrayBuffer()).toString()], [206, 'bytes 10-19/36', 'abcdefghij']);++ r = await fetch(h.url, { method: 'HEAD' });++ assert.deepStrictEqual([r.status, r.headers.get('content-length')], [200, '36']);++ assert.strictEqual((await fetch(h.url.replace(/[0-9a-f]{32}$/, '0'.repeat(32)))).status, 404);++ h.release();++ assert.strictEqual((await fetch(h.url)).status, 404, 'released');++});++++t.test('a URL the rules refuse is refused at registration; a refused redirect never reaches the reader', async () => {++ await assert.rejects(proxy.register('http://10.0.0.1/v/1'), (e) => e.code === 'fetch.refused');++ await assert.rejects(proxy.register('https://evil.example/v/1'), (e) => e.code === 'fetch.refused');++ const h = await proxy.register(`${base}/redirect-out`);++ const r = await fetch(h.url);++ assert.ok([403, 502].includes(r.status), `refused: ${r.status}`);++ proxy.close(); upstream.close();++});++++t.run();
@@ -135,6 +135,9 @@ function load(env = process.env) {
@@ -135,6 +135,9 @@ function load(env = process.env) {
sceneThreshold: float(env.AI_MEDIA_SCENE_THRESHOLD, 10),
sceneThreshold: float(env.AI_MEDIA_SCENE_THRESHOLD, 10),
// Speech-to-text in windows: each call stays well inside the whisper provider's timeout (10 min).
// Speech-to-text in windows: each call stays well inside the whisper provider's timeout (10 min).
sttWindowSec: Math.max(10, int(env.AI_MEDIA_STT_WINDOW_SEC, 600)),
sttWindowSec: Math.max(10, int(env.AI_MEDIA_STT_WINDOW_SEC, 600)),
+ // Read the recording where it lies through the loopback reader (server/media-proxy.js), never downloading
+ // it; AI_MEDIA_STREAM=0 downloads it first (then the disk cap applies).
+ stream: bool(env.AI_MEDIA_STREAM, true),
ffmpeg: env.FFMPEG_BIN || 'ffmpeg',
ffmpeg: env.FFMPEG_BIN || 'ffmpeg',
ffprobe: env.FFPROBE_BIN || 'ffprobe',
ffprobe: env.FFPROBE_BIN || 'ffprobe',
},
},
@@ -157,6 +157,50 @@ function createFetcher(config, { transport = null } = {}) {
@@ -157,6 +157,50 @@ function createFetcher(config, { transport = null } = {}) {
throw new AiError(502, 'source.unavailable', 'too many redirects');
throw new AiError(502, 'source.unavailable', 'too many redirects');
}
}
+ /**
+ * One upstream response for a media reader (server/media-proxy.js): the same URL rules as fetchUrl, redirects
+ * judged and followed here (the reader never sees them), a Range header passed through, the body left as a stream.
+ * → { res } (an http.IncomingMessage with status 200/206/416), or throws AiError.
+ */
+ async function openStream(raw, { method = 'GET', range = null, signal, timeoutMs = m.timeoutMs } = {}) {
+ let target = judge(raw);
+ for (let hop = 0; hop < 4; hop++) {
+ const res = await new Promise((resolve, reject) => {
+ const { url, internal } = target;
+ const mod = url.protocol === 'https:' ? https : http;
+ const req = mod.request(url, {
+ method: method === 'HEAD' ? 'HEAD' : 'GET',
+ headers: { 'User-Agent': 'OpenVibe.AI/0.1 (+https://ai.openvibe.network)', Accept: '*/*', ...(range ? { Range: range } : {}) },
+ lookup: internal ? undefined : safeLookup,
+ timeout: timeoutMs,
+ });
+ const onAbort = () => { req.destroy(); reject(new AiError(409, 'run.cancelled', 'cancelled')); };
+ if (signal) { if (signal.aborted) return onAbort(); signal.addEventListener('abort', onAbort, { once: true }); }
+ req.on('timeout', () => { req.destroy(); reject(new AiError(502, 'source.unavailable', 'fetch timed out')); });
+ req.on('error', (err) => reject(err.code === 'EADDRNOTPUBLIC' ? new AiError(422, 'fetch.refused', err.message) : new AiError(502, 'source.unavailable', `fetch failed: ${err.message}`)));
+ req.on('response', (r) => { if (signal) signal.removeEventListener('abort', onAbort); resolve(r); });
+ req.end();
+ });
+ if (res.statusCode >= 300 && res.statusCode < 400 && res.headers.location) {
+ res.resume();
+ if (hop === 3) throw new AiError(502, 'source.unavailable', 'too many redirects');
+ target = follow(target, new URL(res.headers.location, target.url).toString());
+ continue;
+ }
+ if (![200, 206, 416].includes(res.statusCode)) { res.resume(); throw new AiError(502, 'source.unavailable', `source answered HTTP ${res.statusCode}`); }
+ return { res };
+ }
+ throw new AiError(502, 'source.unavailable', 'too many redirects');
+ }
+
+ /** A media input's URL (media_url, or a MediaRef). */
+ function mediaUrlOf(input) {
+ const url = input.media_url || (input.media ? mediaRefUrl(input.media) : null);
+ if (!url) throw new AiError(422, 'input.invalid', 'media needs media_url or media (MediaRef)');
+ judge(url);
+ return url;
+ }
+
/** A MediaRef -> the public URL of that object (legacy kinds only for now). */
/** A MediaRef -> the public URL of that object (legacy kinds only for now). */
function mediaRefUrl(ref) {
function mediaRefUrl(ref) {
const id = ref && ref.media_id;
const id = ref && ref.media_id;
@@ -210,7 +254,7 @@ function createFetcher(config, { transport = null } = {}) {
@@ -210,7 +254,7 @@ function createFetcher(config, { transport = null } = {}) {
return { file, url };
return { file, url };
}
}
- return { judge, follow, fetchUrl, mediaRefUrl, loadImage, loadMediaToFile, hostAllowed: (h) => hostAllowed(h, m.allowHosts) };
+ return { judge, follow, fetchUrl, openStream, mediaUrlOf, mediaRefUrl, loadImage, loadMediaToFile, hostAllowed: (h) => hostAllowed(h, m.allowHosts) };
}
}
/** Any image -> downscaled JPEG (sharp, optional). Falls back to the original bytes. */
/** Any image -> downscaled JPEG (sharp, optional). Falls back to the original bytes. */
new file mode 100644
new file mode 100644
@@ -0,0 +1,71 @@
@@ -0,0 +1,71 @@
+'use strict';
+/**
+ * A loopback reader for media the analysis streams instead of downloading (roadmap WS-O tasks 2 and 5). ffprobe and
+ * ffmpeg read `http://127.0.0.1:<port>/m/<token>`; this server answers each request (GET or HEAD, with any Range) by
+ * asking the real URL through the fetcher's rules (fetcher.openStream: the allow-list, public addresses checked at
+ * connect time, redirects judged here and never shown to ffmpeg). So a long recording is read where it lies: the
+ * signal pass streams it once, and each speech-to-text window seeks to its start with a range request. Nothing is
+ * written to disk.
+ *
+ * const handle = proxy.register(url) // judged now; valid until release() or 6 hours
+ * handle.url // what ffmpeg reads
+ * handle.release()
+ *
+ * The server listens on 127.0.0.1 only, on a random port, from the first registration, and serves only registered
+ * tokens (128 random bits each).
+ */
+const http = require('http');
+const crypto = require('crypto');
+
+const TTL_MS = 6 * 3600 * 1000;
+
+function createMediaProxy({ fetcher, log = console, clock = { now: () => Date.now() } }) {
+ const entries = new Map(); // token -> { url, expires }
+ let server = null;
+ let ready = null;
+
+ function start() {
+ if (ready) return ready;
+ server = http.createServer((req, res) => {
+ const m = /^\/m\/([0-9a-f]{32})$/.exec(String(req.url || ''));
+ const e = m && entries.get(m[1]);
+ if (!e || e.expires < clock.now()) { res.writeHead(404).end(); return; }
+ if (req.method !== 'GET' && req.method !== 'HEAD') { res.writeHead(405).end(); return; }
+ const range = typeof req.headers.range === 'string' && /^bytes=\d*-\d*$/.test(req.headers.range) ? req.headers.range : null;
+ const ac = new AbortController();
+ req.on('close', () => ac.abort());
+ fetcher.openStream(e.url, { method: req.method, range, signal: ac.signal, timeoutMs: 60000 }).then(({ res: up }) => {
+ const headers = {};
+ for (const k of ['content-type', 'content-length', 'content-range', 'accept-ranges', 'last-modified', 'etag']) if (up.headers[k]) headers[k] = up.headers[k];
+ res.writeHead(up.statusCode, headers);
+ if (req.method === 'HEAD') { up.resume(); res.end(); return; }
+ up.pipe(res);
+ up.on('error', () => res.destroy());
+ }, (err) => {
+ if (ac.signal.aborted) return;
+ log.warn && log.warn(`[media-proxy] ${err.code || 'error'}: ${String(err.detail || err.message).slice(0, 200)}`);
+ if (!res.headersSent) res.writeHead(err.status === 422 ? 403 : 502).end();
+ else res.destroy();
+ });
+ });
+ server.keepAliveTimeout = 5000;
+ ready = new Promise((resolve, reject) => { server.once('error', reject); server.listen(0, '127.0.0.1', () => { server.unref(); resolve(server.address().port); }); });
+ return ready;
+ }
+
+ /** Judge `url` now and return a local URL for it. */
+ async function register(url) {
+ fetcher.judge(url);
+ const port = await start();
+ const token = crypto.randomBytes(16).toString('hex');
+ entries.set(token, { url, expires: clock.now() + TTL_MS });
+ for (const [k, v] of entries) if (v.expires < clock.now()) entries.delete(k);
+ return { url: `http://127.0.0.1:${port}/m/${token}`, release: () => entries.delete(token) };
+ }
+
+ function close() { entries.clear(); if (server) server.close(); server = null; ready = null; }
+
+ return { register, close, _entries: entries };
+}
+
+module.exports = { createMediaProxy };
@@ -184,6 +184,9 @@ function extractive(facts, highlights) {
@@ -184,6 +184,9 @@ function extractive(facts, highlights) {
}
}
function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spawnImpl = spawn, log = console }) {
function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spawnImpl = spawn, log = console }) {
+ // Long recordings are streamed, not downloaded (WS-O task 2): ffmpeg reads them through a loopback reader.
+ let proxy = null;
+ const reader = () => proxy || (proxy = require('../media-proxy').createMediaProxy({ fetcher, log }));
const ma = config.mediaAnalysis || {};
const ma = config.mediaAnalysis || {};
const ffmpeg = ma.ffmpeg || 'ffmpeg';
const ffmpeg = ma.ffmpeg || 'ffmpeg';
const ffprobe = ma.ffprobe || 'ffprobe';
const ffprobe = ma.ffprobe || 'ffprobe';
@@ -318,8 +321,15 @@ function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spa
@@ -318,8 +321,15 @@ function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spa
async function analyse(stepDef, input, wf, runRow, ctx) {
async function analyse(stepDef, input, wf, runRow, ctx) {
const gaps = [];
const gaps = [];
- const dir = work();
- const { file } = await fetcher.loadMediaToFile(input, { signal: ctx.signal, maxBytes: roomFor(dir), timeoutMs: ma.fetchTimeoutMs, dir });
+ const streamed = ma.stream !== false;
+ let file; let handle = null;
+ if (streamed) {
+ handle = await reader().register(fetcher.mediaUrlOf(input));
+ file = handle.url;
+ } else {
+ const dir = work();
+ ({ file } = await fetcher.loadMediaToFile(input, { signal: ctx.signal, maxBytes: roomFor(dir), timeoutMs: ma.fetchTimeoutMs, dir }));
+ }
try {
try {
const info = await probe(file, ctx.signal);
const info = await probe(file, ctx.signal);
if (!info.video && !info.audio) throw new AiError(422, 'media.unreadable', 'the file has no audio or video stream');
if (!info.video && !info.audio) throw new AiError(422, 'media.unreadable', 'the file has no audio or video stream');
@@ -375,7 +385,8 @@ function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spa
@@ -375,7 +385,8 @@ function createMediaAnalysis({ registry, pool, fetcher, quotas, config = {}, spa
// The run's provider: the model that wrote the overview, else the speech-to-text engine that ran.
// The run's provider: the model that wrote the overview, else the speech-to-text engine that ran.
return { output, exec: ov.exec || sttExec || null, route: ov.route || null };
return { output, exec: ov.exec || sttExec || null, route: ov.route || null };
} finally {
} finally {
- try { fs.unlinkSync(file); } catch { /* */ }
+ if (handle) handle.release();
+ else { try { fs.unlinkSync(file); } catch { /* */ } }
}
}
}
}
@@ -120,9 +120,9 @@ t.test('a real run: signals, scenes, highlights with evidence, and the local mod
@@ -120,9 +120,9 @@ t.test('a real run: signals, scenes, highlights with evidence, and the local mod
assert.match(prompt, /Loudest moments: 0:1[6-8]/);
assert.match(prompt, /Loudest moments: 0:1[6-8]/);
assert.strictEqual(local.last.model, 'qwen-test');
assert.strictEqual(local.last.model, 'qwen-test');
assert.ok(!r.body.run.output.gaps.some((g) => /local model/.test(g)));
assert.ok(!r.body.run.output.gaps.some((g) => /local model/.test(g)));
+ // Streamed through the loopback reader (the default): nothing was written to disk.
const work = path.join(h.dir, 'media-tmp');
const work = path.join(h.dir, 'media-tmp');
- assert.ok(fs.existsSync(work), 'the work directory is beside the database (not a tmpfs /tmp)');
- assert.deepStrictEqual(fs.readdirSync(work).filter((f) => f.startsWith('openvibe-ai-media-')), [], 'the downloaded file is removed');
+ assert.deepStrictEqual(fs.existsSync(work) ? fs.readdirSync(work).filter((f) => f.startsWith('openvibe-ai-media-')) : [], [], 'nothing downloaded');
} finally { await h.stop(); await local.close(); }
} finally { await h.stop(); await local.close(); }
});
});
@@ -179,7 +179,7 @@ t.test('too little free disk refuses before downloading', async () => {
@@ -179,7 +179,7 @@ t.test('too little free disk refuses before downloading', async () => {
const counting = nodeHttp.createServer((req, res) => { fetched++; res.writeHead(200, { 'Content-Type': 'video/mp4' }); res.end(FIXTURE); });
const counting = nodeHttp.createServer((req, res) => { fetched++; res.writeHead(200, { 'Content-Type': 'video/mp4' }); res.end(FIXTURE); });
await new Promise((r) => counting.listen(0, '127.0.0.1', r));
await new Promise((r) => counting.listen(0, '127.0.0.1', r));
const base = `http://127.0.0.1:${counting.address().port}`;
const base = `http://127.0.0.1:${counting.address().port}`;
- const h = await boot({ env: { OV_MEDIA_INTERNAL_URL: base, AI_MEDIA_ANALYSIS_DISK_RESERVE_BYTES: String(1024 ** 5) } });
+ const h = await boot({ env: { OV_MEDIA_INTERNAL_URL: base, AI_MEDIA_ANALYSIS_DISK_RESERVE_BYTES: String(1024 ** 5), AI_MEDIA_STREAM: '0' } });
try {
try {
const r = await request(h.base, 'POST', '/api/v1/runs?wait=20000', { tok, body: { workflow: 'media.analyze', input: { media_url: `${base}/fixture.mp4` } } });
const r = await request(h.base, 'POST', '/api/v1/runs?wait=20000', { tok, body: { workflow: 'media.analyze', input: { media_url: `${base}/fixture.mp4` } } });
assert.strictEqual(r.body.run.status, 'failed');
assert.strictEqual(r.body.run.status, 'failed');
new file mode 100644
new file mode 100644
@@ -0,0 +1,58 @@
@@ -0,0 +1,58 @@
+'use strict';
+// The loopback reader media.analyze streams through (server/media-proxy.js; roadmap WS-O task 2): ffmpeg reads a
+// registered token and gets the real URL's bytes, Range requests pass through (206 with Content-Range), HEAD answers
+// headers only, an unknown or released token is 404, and a redirect the fetcher's rules refuse (a public host
+// sending it to the internal Media origin) never reaches the reader: 403.
+const assert = require('assert');
+const nodeHttp = require('http');
+const { suite } = require('./helpers');
+const { load } = require('../server/config');
+const { createFetcher } = require('../server/fetcher');
+const { createMediaProxy } = require('../server/media-proxy');
+
+const t = suite('media-proxy');
+const BODY = Buffer.from('0123456789abcdefghijklmnopqrstuvwxyz');
+let upstream; let base; let proxy;
+
+t.test('boot an internal Media stand-in', async () => {
+ upstream = nodeHttp.createServer((req, res) => {
+ if (req.url === '/redirect-out') { res.writeHead(302, { Location: 'https://evil.example/x' }); return res.end(); }
+ const m = /^bytes=(\d+)-(\d*)$/.exec(req.headers.range || '');
+ if (m) {
+ const start = Number(m[1]); const end = m[2] ? Number(m[2]) : BODY.length - 1;
+ res.writeHead(206, { 'Content-Type': 'video/mp4', 'Content-Length': end - start + 1, 'Content-Range': `bytes ${start}-${end}/${BODY.length}`, 'Accept-Ranges': 'bytes' });
+ return res.end(req.method === 'HEAD' ? undefined : BODY.subarray(start, end + 1));
+ }
+ res.writeHead(200, { 'Content-Type': 'video/mp4', 'Content-Length': BODY.length, 'Accept-Ranges': 'bytes' });
+ res.end(req.method === 'HEAD' ? undefined : BODY);
+ });
+ await new Promise((r) => upstream.listen(0, '127.0.0.1', r));
+ base = `http://127.0.0.1:${upstream.address().port}`;
+ proxy = createMediaProxy({ fetcher: createFetcher(load({ NODE_ENV: 'test', OV_MEDIA_INTERNAL_URL: base })), log: { warn() {} } });
+});
+
+t.test('whole reads, ranges, HEAD and unknown tokens', async () => {
+ const h = await proxy.register(`${base}/v/1`);
+ assert.match(h.url, /^http:\/\/127\.0\.0\.1:\d+\/m\/[0-9a-f]{32}$/);
+ let r = await fetch(h.url);
+ assert.strictEqual(r.status, 200);
+ assert.deepStrictEqual(Buffer.from(await r.arrayBuffer()), BODY);
+ r = await fetch(h.url, { headers: { Range: 'bytes=10-19' } });
+ assert.deepStrictEqual([r.status, r.headers.get('content-range'), Buffer.from(await r.arrayBuffer()).toString()], [206, 'bytes 10-19/36', 'abcdefghij']);
+ r = await fetch(h.url, { method: 'HEAD' });
+ assert.deepStrictEqual([r.status, r.headers.get('content-length')], [200, '36']);
+ assert.strictEqual((await fetch(h.url.replace(/[0-9a-f]{32}$/, '0'.repeat(32)))).status, 404);
+ h.release();
+ assert.strictEqual((await fetch(h.url)).status, 404, 'released');
+});
+
+t.test('a URL the rules refuse is refused at registration; a refused redirect never reaches the reader', async () => {
+ await assert.rejects(proxy.register('http://10.0.0.1/v/1'), (e) => e.code === 'fetch.refused');
+ await assert.rejects(proxy.register('https://evil.example/v/1'), (e) => e.code === 'fetch.refused');
+ const h = await proxy.register(`${base}/redirect-out`);
+ const r = await fetch(h.url);
+ assert.ok([403, 502].includes(r.status), `refused: ${r.status}`);
+ proxy.close(); upstream.close();
+});
+
+t.run();