openvibers/openvibe.media Commits
Job events carry a projection, not the job: media.job.<transition> payloads are queue.jobEvent (id, app_id, object_id, type, status, attempts, max_attempts, error_code, cancel_requested, has_result, run_after, decided_at, created_at, updated_at, started_at, finished_at), times in ISO 8601 UTC or null
Changes
@@@ -45,7 +45,8 @@ server/content-type, verify-job (scheduled copy verification), copy-report —see docs/object-model.mdjobs/ media_jobs: queue (states, idempotency, media.job.* events in the state-- change's transaction), worker (light/heavy lanes, leases, retries,++ 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(openvibe-shared: pinned OpenVibe.Shared v1.0.0 release, installed by npm)@@@ -152,7 +153,7 @@ MEDIA_APP_KEYS="live:key1,games:key2"| POST | `/vods/:id/complete` | finalize chunked upload (user JWT ok) || POST | `/vods/:id/finalize` | close recording; remux, probe, thumbnail, webhook || GET | `/vods/:id` | `{ id, title, status, duration, playback_url, thumbnail_url, storage_provider, … }` |--| GET | `/vods?limit&offset&user_id&stream_id&managed_stream_id&include_private&order` | list; `include_private` app-key only; `order` = newest\|oldest\|views |++| GET | `/vods?limit&offset&user_id&stream_id&managed_stream_id&include_private&order&since` | list; `include_private` app-key only; `order` = newest\|oldest\|views; `since` = created at or after (ISO 8601 or `YYYY-MM-DD HH:MM:SS`, UTC) || PUT | `/vods/:id` | `{ title?, description?, visibility? }` || DELETE | `/vods/:id` | deletes local + B2 + R2 objects + row |@@@ -170,10 +171,10 @@ lossless `.master.mkv` recovery archive. A `.seekable` sidecar is remuxed every| method | path | notes ||---|---|---|--| POST | `/clips` | `{ vod_id, start_s, end_s, title?, user_id?, visibility? }` → **202** `{ id, status: 'processing' }`; cut runs in background (from the local file or a presigned B2/R2 URL); duplicate windows are deduplicated; live recordings are clamped to flushed footage. Multipart `video` = direct upload of an already-cut blob → **201** ready |++| POST | `/clips` | `{ vod_id, start_s, end_s, title?, description?, user_id?, visibility?, auto_generated? }` → **202** `{ id, status: 'processing' }`; cut runs in background (from the local file or a presigned B2/R2 URL); duplicate windows are deduplicated; live recordings are clamped to flushed footage. Multipart `video` = direct upload of an already-cut blob → **201** ready || GET | `/clips/:id` | status: `processing | ready | failed` |--| GET | `/clips?limit&offset&vod_id&stream_id&user_id&channel_user_id&hide_self&include_private&order` | list; `channel_user_id` = clipped-channel owner; `include_private` app-key only |--| PUT | `/clips/:id` | `{ title?, visibility? }` |++| GET | `/clips?limit&offset&vod_id&stream_id&user_id&channel_user_id&hide_self&include_private&order&auto_generated&status&since` | list; `channel_user_id` = clipped-channel owner; `include_private` app-key only; `auto_generated=1\|0` = cut by the app's automation (AI auto-clips) or made by a person; `status=ready` = playable clips only; `since` as for VODs |++| PUT | `/clips/:id` | `{ title?, visibility?, auto_generated? }` (`auto_generated` app key only, 403 when acting for a user) || DELETE | `/clips/:id` | local + offloaded objects + row |### Pastes@@@ -356,11 +356,30 @@ Code: `server/jobs/`. Tests: `test/jobs.test.js`, `test/jobs-invariant.test.js`.30 s, 2 min, 8 min, …, at most 1 h) until `max_attempts`; a handler can mark a failure permanent.**Events.** Every state change and its `media.job.<transition>` event (`proposed`, `queued`,--`started`, `retrying`, `succeeded`, `failed`, `cancelled`; subject `job <id>`, payload the job)--commit in one SQLite transaction through Media's outbox, the same rule as the outcome events--(`webhooks.announce()`): no change without its event, no event for a change that rolled back.--Progress events are `low` priority; outcomes and proposals are `important`. Job events go to--OpenVibe.Events only, not to app webhooks.++`started`, `retrying`, `succeeded`, `failed`, `cancelled`; subject `job <id>`) commit in one SQLite++transaction through Media's outbox, the same rule as the outcome events (`webhooks.announce()`): no++change without its event, no event for a change that rolled back. Progress events are `low`++priority; outcomes and proposals are `important`. Job events go to OpenVibe.Events only, not to app++webhooks.++++The payload is the event projection (`queue.jobEvent`, contracts `media.job.<transition>@1`), not++the job as the API answers it:++++```json++{ "app_id": "live", "id": "mjob_…", "object_id": "med_…", "type": "object.split",++ "status": "succeeded", "attempts": 1, "max_attempts": 3, "error_code": null,++ "cancel_requested": false, "has_result": true,++ "run_after": null, "decided_at": null, "created_at": "2026-09-23T21:06:24.000Z",++ "updated_at": "…Z", "started_at": "…Z", "finished_at": "…Z" }++```++++`object_id` is null for a tenant-wide job (`invariant.scan`). `status` is the job's state after the++change (`retrying` carries `queued`, `started` carries `running`); times are ISO 8601 UTC or null;++`error_code` is the stable code only. The tenant's `params`, the `idempotency_key`, the free-text++`error`, the `result` (a thumbnail URL of a private VOD, for one), `created_by`/`decided_by` and++`owner_user_id` (tenant-local user ids) are left out: events travel beyond the tenant. A consumer++that needs them GETs the job with its own tenant-scoped token (`GET /api/v2/:app/jobs/:id`);++`has_result` says whether there is a result to fetch.**Idempotency.** `Idempotency-Key` (or `idempotency_key`) is unique per tenant: the same key with thesame request (type, object, params) answers with the job it made (`Idempotent-Replayed: true`); a@@@ -10,7 +10,8 @@* media.object.uploaded → media.object.uploaded (subject object <id>)* storage.alert | storage.recovered → media.storage.alert | media.storage.recovered* job state changes (server/jobs/queue.js) → media.job.proposed | queued | started | retrying |-- * succeeded | failed | cancelled (subject job <id>; Events only, no webhook)++ * succeeded | failed | cancelled (subject job <id>; Events only, no webhook;++ * payload queue.jobEvent, not the tenant's params/result/error)** The outbox row is written INSIDE the SQLite transaction that makes the state change it describes* (record(), called from webhooks.announce()): the event exists if and only if the change committed,@@@ -99,8 +100,10 @@ function record(webhookEvent, appId, data) {}/**-- * Queue media.job.<transition> for a job row (queue.jobPublic shape). Same rule as record(): call it-- * inside the transaction that changes the job's state; it throws when the insert fails.++ * Queue media.job.<transition> for a job (queue.jobEvent: ids, state, counters, ISO times and++ * has_result; never params, result, error text, idempotency key or who created/decided it; the++ * tenant-scoped GET /api/v2/:app/jobs/:id has those). Same rule as record(): call it inside the++ * transaction that changes the job's state; it throws when the insert fails.*/function recordJob(transition, job) {const priority = JOB_TRANSITIONS[transition];@@@ -28,7 +28,6 @@ const STATUSES = ['proposed', 'queued', 'running', 'succeeded', 'failed', 'canceconst FINISHED = ['succeeded', 'failed', 'cancelled'];const ACTIVE = ['queued', 'running'];const MAX_ACTIVE_PER_TENANT = 50; // queued + running jobs one tenant may have (proposals excluded)--const RESULT_EVENT_MAX = 8192; // a larger result is left out of the event payload (GET the job for it)const bus = new EventEmitter();bus.setMaxListeners(0);@@@ -123,10 +122,41 @@ function jobPublic(row) {};}--function eventPayload(row) {-- const pub = jobPublic(row);-- if (pub.result && JSON.stringify(pub.result).length > RESULT_EVENT_MAX) pub.result = { omitted: true, reason: 'larger than the event limit; GET the job' };-- return pub;++/** A SQLite UTC time ('YYYY-MM-DD HH:MM:SS') as ISO 8601 UTC ('…T…Z'), or null. */++function isoTime(v) {++ if (v == null || v === '') return null;++ const s = String(v);++ const d = new Date(/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(\.\d+)?$/.test(s) ? `${s.replace(' ', 'T')}Z` : s);++ return Number.isNaN(d.getTime()) ? null : d.toISOString();++}++++/**++ * The media.job.<transition> event payload: the job's identity, state, counters and times, and whether++ * it has a result. Deliberately not the tenant's params, the idempotency key, the free-text error, the++ * result (a thumbnail URL of a private VOD, for one), created_by/decided_by or owner_user_id (tenant-local++ * user ids): events travel beyond the tenant, and a consumer that needs those GETs the job with its own++ * tenant-scoped token (GET /api/v2/:app/jobs/:id, queue.jobPublic, unchanged).++ */++function jobEvent(row) {++ if (!row) return null;++ return {++ id: row.id,++ app_id: row.app_id,++ object_id: row.object_id || null,++ type: row.job_type,++ status: row.status,++ attempts: row.attempts,++ max_attempts: row.max_attempts,++ error_code: row.error_code || null,++ cancel_requested: !!row.cancel_requested,++ has_result: parseJson(row.result, null) != null,++ run_after: isoTime(row.run_after),++ decided_at: isoTime(row.decided_at),++ created_at: isoTime(row.created_at),++ updated_at: isoTime(row.updated_at),++ started_at: isoTime(row.started_at),++ finished_at: isoTime(row.finished_at),++ };}/**@@@ -139,7 +169,7 @@ function commit(id, transition, change) {db.getDb().transaction(() => {if (!change()) return;row = get(id);-- if (transition) events.recordJob(transition, eventPayload(row));++ if (transition) events.recordJob(transition, jobEvent(row));})();if (row) {if (transition) events.kick();@@@ -201,7 +231,7 @@ function enqueue({ appId, type, objectId = null, params = {}, status = 'queued',[id, appId, objectId, type, status, key, hash, JSON.stringify(params || {}),Math.max(1, Math.min(10, Number(maxAttempts) || spec.maxAttempts)), createdBy, ownerUserId ?? null]);const row = get(id);-- events.recordJob(status, eventPayload(row));++ events.recordJob(status, jobEvent(row));out = { job: row, created: true, replayed: false, deduped: false };})();if (out.created) { events.kick(); bus.emit('change', out.job); }@@@ -254,7 +284,7 @@ function claim(types, { leaseS = 120, id = null } = {}) {if (!update(next.id, 'queued', `status = 'running', attempts = attempts + 1, started_at = COALESCE(started_at, CURRENT_TIMESTAMP),lease_until = datetime('now', '+${Math.round(leaseS)} seconds'), run_after = NULL`)) return;row = get(next.id);-- events.recordJob('started', eventPayload(row));++ events.recordJob('started', jobEvent(row));})();if (row) { events.kick(); bus.emit('change', row); }return row;@@@ -366,7 +396,7 @@ function prune({ days = 30, types = ['thumbnail.regenerate'] } = {}) {module.exports = {STATUSES, FINISHED, ACTIVE, MAX_ACTIVE_PER_TENANT, JobError, bus,register, typeSpec, typeNames, requestHash, parseJson,-- get, getForApp, jobPublic, list, counts,++ get, getForApp, jobPublic, jobEvent, list, counts,enqueue, approve, cancel,claim, renew, saveCheckpoint, succeed, fail, markCancelled, recoverInterrupted, waitFor, prune,};@@@ -99,6 +99,10 @@ const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;assert.strictEqual(remux.type, 'object.remux', 'no duration: remux first');assert.strictEqual(split.idempotency_key, `invariant:${bigVod}:object.split`);assert.deepStrictEqual(jobEvents(split.id), ['media.job.proposed']);++ const proposedEv = outbox().find((e) => e.subject.id === split.id).payload;++ assert.deepStrictEqual([proposedEv.status, proposedEv.type, proposedEv.object_id, proposedEv.decided_at, proposedEv.started_at, proposedEv.has_result], ['proposed', 'object.split', bigVod, null, null, false]);++ for (const k of ['params', 'idempotency_key', 'created_by', 'decided_by', 'owner_user_id', 'result', 'error']) assert.ok(!(k in proposedEv), `the proposal's event has no ${k}`);++ assert.ok(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.000Z$/.test(proposedEv.created_at), 'ISO 8601 UTC');assert.strictEqual(outbox().find((e) => e.subject.id === split.id).priority, 'important', 'a proposal waiting for its owner is important');worker.start();@@@ -210,6 +214,8 @@ const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;assert.strictEqual(sc.status, 0, sc.stderr);assert.deepStrictEqual([queue.get(manual.job.id).status, queue.get(manual.job.id).decided_by], ['queued', 'owner']);assert.ok(jobEvents(manual.job.id).includes('media.job.queued'), "the script's change queued its event in the outbox");++ const approvedEv = outbox().filter((e) => e.subject.id === manual.job.id && e.event_type === 'media.job.queued').pop().payload;++ assert.ok(/Z$/.test(approvedEv.decided_at) && !('decided_by' in approvedEv) && !JSON.stringify(approvedEv).includes('owner'), 'the approval is dated, its decider is not published');sc = script('cancel', manual.job.id, '--by', 'owner');assert.strictEqual(sc.status, 0, sc.stderr);assert.strictEqual(queue.get(manual.job.id).status, 'cancelled');@@@ -1,6 +1,8 @@'use strict';// Media job system (server/jobs): the schema-only media_jobs table is rebuilt; every state change--// commits with its media.job.* outbox row (and rolls back with it); idempotency keys; retries with++// commits with its media.job.* outbox row (and rolls back with it), whose payload is the event++// projection (ids, state, counters, ISO times, has_result; never params, result, error text, the++// idempotency key or who created/decided it); idempotency keys; retries with// backoff, permanent failures, checkpoints; cancellation by the owner (queued at once, running through// the abort signal) and not by anyone else; interrupted jobs are requeued at start; the v1 thumbnail// route runs as a thumbnail.regenerate job and answers as before.@@@ -78,6 +80,23 @@ const waitFor = async (fn, ms = 5000) => {const conn = db.getDb();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);++ const jobPayloads = (id) => outbox().filter((e) => e.subject.type === 'job' && e.subject.id === id).map((e) => e.payload);++ // The media.job.* payload: exactly these fields. The rest stays behind the tenant-scoped jobs API.++ const EVENT_KEYS = ['app_id', 'attempts', 'cancel_requested', 'created_at', 'decided_at', 'error_code', 'finished_at', 'has_result',++ 'id', 'max_attempts', 'object_id', 'run_after', 'started_at', 'status', 'type', 'updated_at'];++ const LEFT_OUT = ['params', 'idempotency_key', 'error', 'result', 'created_by', 'decided_by', 'owner_user_id'];++ const TIMES = ['run_after', 'decided_at', 'created_at', 'updated_at', 'started_at', 'finished_at'];++ const ISO = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/;++ const TRANSITION_STATUS = { proposed: 'proposed', queued: 'queued', started: 'running', retrying: 'queued', succeeded: 'succeeded', failed: 'failed', cancelled: 'cancelled' };++ const assertJobEvent = (env) => {++ const p = env.payload, what = `${env.event_type} ${p.id}`;++ assert.deepStrictEqual(Object.keys(p).sort(), EVENT_KEYS, `${what}: exactly the event projection`);++ for (const k of LEFT_OUT) assert.ok(!(k in p), `${what}: no ${k}`);++ for (const k of TIMES) assert.ok(p[k] === null || ISO.test(p[k]), `${what}: ${k} is ISO 8601 UTC or null (${p[k]})`);++ assert.ok(ISO.test(p.created_at) && ISO.test(p.updated_at), `${what}: created_at and updated_at are always set`);++ assert.strictEqual(typeof p.has_result, 'boolean');++ assert.strictEqual(p.status, TRANSITION_STATUS[env.event_type.slice('media.job.'.length)], `${what}: status`);++ };const failOutbox = (on) => conn.exec(on? "CREATE TEMP TRIGGER outbox_boom BEFORE INSERT ON event_outbox BEGIN SELECT RAISE(ABORT, 'outbox insert failed'); END": 'DROP TRIGGER IF EXISTS temp.outbox_boom');@@@ -112,6 +131,11 @@ const waitFor = async (fn, ms = 5000) => {assert.deepStrictEqual(jobEvents(a.job.id), ['media.job.queued']);const env = outbox().find((e) => e.subject.id === a.job.id);assert.deepStrictEqual([env.source, env.priority, env.visibility, env.payload.type, env.payload.status, env.payload.app_id], ['media', 'low', 'internal', 'test.flaky', 'queued', 'live']);++ assertJobEvent(env);++ const aRow = queue.get(a.job.id);++ assert.deepStrictEqual([env.payload.created_at, env.payload.updated_at], [aRow.created_at, aRow.updated_at].map((t) => new Date(`${t.replace(' ', 'T')}Z`).toISOString()), 'times are the row\'s, in ISO 8601 UTC');++ assert.deepStrictEqual([env.payload.attempts, env.payload.max_attempts, env.payload.error_code, env.payload.cancel_requested, env.payload.has_result, env.payload.object_id, env.payload.run_after, env.payload.started_at, env.payload.finished_at, env.payload.decided_at],++ [0, 3, null, false, false, null, null, null, null, null]);const q2 = queue.enqueue({ appId: 'live', type: 'test.permanent' });failOutbox(true);assert.throws(() => queue.enqueue({ appId: 'live', type: 'test.flaky', params: { x: 2 } }), /outbox insert failed/);@@@ -126,6 +150,7 @@ const waitFor = async (fn, ms = 5000) => {const k1 = queue.enqueue({ appId: 'live', type: 'test.permanent', params: { n: 1 }, idempotencyKey: 'key-1' });const k2 = queue.enqueue({ appId: 'live', type: 'test.permanent', params: { n: 1 }, idempotencyKey: 'key-1' });assert.ok(k1.created && k2.replayed && k1.job.id === k2.job.id);++ assert.ok(!JSON.stringify(jobPayloads(k1.job.id)).includes('key-1'), 'the idempotency key stays out of the event');assert.throws(() => queue.enqueue({ appId: 'live', type: 'test.permanent', params: { n: 2 }, idempotencyKey: 'key-1' }), (e) => e.code === 'media.job.idempotency_conflict' && e.status === 409);assert.ok(queue.enqueue({ appId: 'games', type: 'test.permanent', params: { n: 1 }, idempotencyKey: 'key-1' }).created, 'keys are per tenant');const d1 = queue.enqueue({ appId: 'live', type: 'test.permanent', params: { same: true }, dedupeActive: true });@@@ -150,7 +175,18 @@ const waitFor = async (fn, ms = 5000) => {assert.strictEqual(queue.get(always.job.id).attempts, 2, 'retried until max_attempts');assert.deepStrictEqual(jobEvents(always.job.id), ['media.job.queued', 'media.job.started', 'media.job.retrying', 'media.job.started', 'media.job.failed']);const failedEv = outbox().filter((e) => e.subject.id === always.job.id).pop();-- assert.deepStrictEqual([failedEv.priority, failedEv.payload.error], ['important', 'nope']);++ assert.deepStrictEqual([failedEv.priority, failedEv.payload.error_code, failedEv.payload.has_result, failedEv.payload.attempts], ['important', null, false, 2]);++ assert.ok(!JSON.stringify(failedEv.payload).includes('nope'), 'the free-text error stays out of the event');++ assert.strictEqual(queue.jobPublic(queue.get(always.job.id)).error, 'nope', 'GET the job for it');++ const permEv = outbox().filter((e) => e.subject.id === perm.job.id).pop();++ assert.deepStrictEqual([permEv.event_type, permEv.payload.error_code, permEv.payload.attempts], ['media.job.failed', 'bad_input', 1], 'the stable code is in the event');++ assert.ok(!JSON.stringify(permEv.payload).includes('will never work'));++ const [aQueued, aStarted, aRetrying, aStarted2, aSucceeded] = jobPayloads(a.job.id);++ assert.deepStrictEqual([aQueued.status, aStarted.status, aRetrying.status, aStarted2.status, aSucceeded.status], ['queued', 'running', 'queued', 'running', 'succeeded']);++ assert.ok(ISO.test(aRetrying.run_after) && ISO.test(aStarted.started_at) && aRetrying.finished_at === null, 'retrying: a backoff time, started, not finished');++ assert.deepStrictEqual([aStarted.attempts, aStarted2.attempts, aStarted2.started_at], [1, 2, aStarted.started_at], 'started_at stays the first attempt\'s start');++ assert.deepStrictEqual([aSucceeded.has_result, ISO.test(aSucceeded.finished_at), aSucceeded.error_code], [true, true, null]);++ assert.ok(!JSON.stringify(aSucceeded).includes('resumed_from') && !JSON.stringify(jobPayloads(a.job.id)).includes('"x"'), 'neither the result nor the params are in the events');console.log('✅ worker: retries with backoff resume from the checkpoint; permanent failures stop; attempts are capped');// ── 5. Cancellation by the owner ──@@@ -181,6 +217,10 @@ const waitFor = async (fn, ms = 5000) => {assert.deepStrictEqual([r.status, r.body.job.cancel_requested], [202, true], 'a running job stops at its next check');assert.ok(await waitFor(() => queue.get(slow.job.id).status === 'cancelled'), 'the abort signal stopped it');assert.deepStrictEqual(jobEvents(slow.job.id), ['media.job.queued', 'media.job.started', 'media.job.cancelled']);++ const slowCancelled = jobPayloads(slow.job.id).pop();++ assert.deepStrictEqual([slowCancelled.cancel_requested, slowCancelled.error_code, ISO.test(slowCancelled.finished_at)], [true, 'cancelled', true]);++ assert.ok(!/user:7|"owner_user_id"|"created_by"|"decided_by"/.test(JSON.stringify(jobPayloads(slow.job.id))), 'who created or cancelled it (tenant-local user ids) stays out of the events');++ assert.deepStrictEqual([queue.jobPublic(queue.get(slow.job.id)).created_by, queue.jobPublic(queue.get(slow.job.id)).owner_user_id], ['app:live:user:7', 7], 'the jobs API still has them');r = await call('DELETE', `/api/v2/live/jobs/${slow.job.id}`);assert.deepStrictEqual([r.status, r.body.code], [409, 'media.job.finished']);const waiting = queue.enqueue({ appId: 'live', type: 'test.permanent', runAfterS: 3600 });@@@ -196,6 +236,8 @@ const waitFor = async (fn, ms = 5000) => {const back = queue.get(orphan.job.id);assert.deepStrictEqual([back.status, back.error_code], ['queued', 'interrupted']);assert.deepStrictEqual(jobEvents(orphan.job.id).slice(-1), ['media.job.retrying']);++ const orphanRetry = jobPayloads(orphan.job.id).pop();++ assert.deepStrictEqual([orphanRetry.status, orphanRetry.error_code, ISO.test(orphanRetry.run_after)], ['queued', 'interrupted', true]);queue.cancel(orphan.job.id, { by: 'test' });console.log('✅ a job left running by a dead process is requeued (or failed when out of attempts)');@@@ -213,6 +255,9 @@ const waitFor = async (fn, ms = 5000) => {assert.deepStrictEqual([r.status, r.body.code], [409, 'media.job.not_proposed']);r = await call('GET', `/api/v2/live/jobs?type=invariant.scan`);assert.deepStrictEqual(r.body.jobs.map((j) => j.id), [scanId]);++ r = await call('GET', `/api/v2/live/jobs/${scanId}`);++ assert.deepStrictEqual([r.status, r.body.job.idempotency_key, r.body.job.params, r.body.job.created_by, r.body.job.created_at.includes('T')], [200, 'scan-1', {}, 'app:live', false],++ 'GET the job answers as before (queue.jobPublic: params, idempotency key, creator, SQLite times)');r = await call('GET', `/api/v2/games/jobs/${scanId}`, { key: 'games-key' });assert.strictEqual(r.status, 404, 'tenants are isolated');r = await call('GET', `/api/v2/live/jobs?status=bogus`);@@@ -235,6 +280,9 @@ const waitFor = async (fn, ms = 5000) => {const tj = queue.get(t1.body.job_id);assert.deepStrictEqual([tj.job_type, tj.status, tj.object_id], ['thumbnail.regenerate', 'succeeded', db.get('SELECT object_id FROM vods WHERE id = 1').object_id]);assert.deepStrictEqual(jobEvents(tj.id), ['media.job.queued', 'media.job.started', 'media.job.succeeded']);++ const thumbDone = jobPayloads(tj.id).pop();++ assert.deepStrictEqual([thumbDone.has_result, thumbDone.object_id], [true, tj.object_id]);++ assert.ok(!JSON.stringify(jobPayloads(tj.id)).includes('/t/vod-1'), 'a thumbnail URL (of a VOD that may be private) stays out of the event');r = await call('POST', '/api/v1/live/thumbnails/vod/2');assert.deepStrictEqual([r.status, r.body.error], [404, 'Media file unavailable'], 'no media is still a 404');assert.strictEqual(queue.get(r.body.job_id).error_code, 'media_unavailable');@@@ -248,6 +296,13 @@ const waitFor = async (fn, ms = 5000) => {assert.ok(await waitFor(() => queue.get(r.body.job.id).status === 'succeeded'));console.log('✅ the v1 thumbnail route runs a thumbnail.regenerate job and answers { url } as before (404 without media)');++ // Every job event this run queued has the projection's shape and its transition's status.++ const all = outbox().filter((e) => e.event_type.startsWith('media.job.'));++ assert.ok(all.length > 30);++ for (const e of all) assertJobEvent(e);++ assert.deepStrictEqual([...new Set(all.map((e) => e.event_type))].sort(), Object.keys(TRANSITION_STATUS).filter((t) => t !== 'proposed').map((t) => `media.job.${t}`).sort());++ console.log(`✅ ${all.length} media.job.* events: the projection only (no params, result, error text, idempotency key, creator/decider or owner user id), ISO 8601 UTC times`);++worker.stop();events._reset();server.close();
@@ -45,7 +45,8 @@ server/
@@ -45,7 +45,8 @@ server/
content-type, verify-job (scheduled copy verification), copy-report —
content-type, verify-job (scheduled copy verification), copy-report —
see docs/object-model.md
see docs/object-model.md
jobs/ media_jobs: queue (states, idempotency, media.job.* events in the state
jobs/ media_jobs: queue (states, idempotency, media.job.* events in the state
- change's transaction), worker (light/heavy lanes, leases, retries,
+ 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,
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
(openvibe-shared: pinned OpenVibe.Shared v1.0.0 release, installed by npm)
(openvibe-shared: pinned OpenVibe.Shared v1.0.0 release, installed by npm)
@@ -152,7 +153,7 @@ MEDIA_APP_KEYS="live:key1,games:key2"
@@ -152,7 +153,7 @@ MEDIA_APP_KEYS="live:key1,games:key2"
| POST | `/vods/:id/complete` | finalize chunked upload (user JWT ok) |
| POST | `/vods/:id/complete` | finalize chunked upload (user JWT ok) |
| POST | `/vods/:id/finalize` | close recording; remux, probe, thumbnail, webhook |
| POST | `/vods/:id/finalize` | close recording; remux, probe, thumbnail, webhook |
| GET | `/vods/:id` | `{ id, title, status, duration, playback_url, thumbnail_url, storage_provider, … }` |
| GET | `/vods/:id` | `{ id, title, status, duration, playback_url, thumbnail_url, storage_provider, … }` |
-| GET | `/vods?limit&offset&user_id&stream_id&managed_stream_id&include_private&order` | list; `include_private` app-key only; `order` = newest\|oldest\|views |
+| GET | `/vods?limit&offset&user_id&stream_id&managed_stream_id&include_private&order&since` | list; `include_private` app-key only; `order` = newest\|oldest\|views; `since` = created at or after (ISO 8601 or `YYYY-MM-DD HH:MM:SS`, UTC) |
| PUT | `/vods/:id` | `{ title?, description?, visibility? }` |
| PUT | `/vods/:id` | `{ title?, description?, visibility? }` |
| DELETE | `/vods/:id` | deletes local + B2 + R2 objects + row |
| DELETE | `/vods/:id` | deletes local + B2 + R2 objects + row |
@@ -170,10 +171,10 @@ lossless `.master.mkv` recovery archive. A `.seekable` sidecar is remuxed every
@@ -170,10 +171,10 @@ lossless `.master.mkv` recovery archive. A `.seekable` sidecar is remuxed every
| method | path | notes |
| method | path | notes |
|---|---|---|
|---|---|---|
-| POST | `/clips` | `{ vod_id, start_s, end_s, title?, user_id?, visibility? }` → **202** `{ id, status: 'processing' }`; cut runs in background (from the local file or a presigned B2/R2 URL); duplicate windows are deduplicated; live recordings are clamped to flushed footage. Multipart `video` = direct upload of an already-cut blob → **201** ready |
+| POST | `/clips` | `{ vod_id, start_s, end_s, title?, description?, user_id?, visibility?, auto_generated? }` → **202** `{ id, status: 'processing' }`; cut runs in background (from the local file or a presigned B2/R2 URL); duplicate windows are deduplicated; live recordings are clamped to flushed footage. Multipart `video` = direct upload of an already-cut blob → **201** ready |
| GET | `/clips/:id` | status: `processing | ready | failed` |
| GET | `/clips/:id` | status: `processing | ready | failed` |
-| GET | `/clips?limit&offset&vod_id&stream_id&user_id&channel_user_id&hide_self&include_private&order` | list; `channel_user_id` = clipped-channel owner; `include_private` app-key only |
-| PUT | `/clips/:id` | `{ title?, visibility? }` |
+| GET | `/clips?limit&offset&vod_id&stream_id&user_id&channel_user_id&hide_self&include_private&order&auto_generated&status&since` | list; `channel_user_id` = clipped-channel owner; `include_private` app-key only; `auto_generated=1\|0` = cut by the app's automation (AI auto-clips) or made by a person; `status=ready` = playable clips only; `since` as for VODs |
+| PUT | `/clips/:id` | `{ title?, visibility?, auto_generated? }` (`auto_generated` app key only, 403 when acting for a user) |
| DELETE | `/clips/:id` | local + offloaded objects + row |
| DELETE | `/clips/:id` | local + offloaded objects + row |
### Pastes
### Pastes
@@ -356,11 +356,30 @@ Code: `server/jobs/`. Tests: `test/jobs.test.js`, `test/jobs-invariant.test.js`.
@@ -356,11 +356,30 @@ Code: `server/jobs/`. Tests: `test/jobs.test.js`, `test/jobs-invariant.test.js`.
30 s, 2 min, 8 min, …, at most 1 h) until `max_attempts`; a handler can mark a failure permanent.
30 s, 2 min, 8 min, …, at most 1 h) until `max_attempts`; a handler can mark a failure permanent.
**Events.** Every state change and its `media.job.<transition>` event (`proposed`, `queued`,
**Events.** Every state change and its `media.job.<transition>` event (`proposed`, `queued`,
-`started`, `retrying`, `succeeded`, `failed`, `cancelled`; subject `job <id>`, payload the job)
-commit in one SQLite transaction through Media's outbox, the same rule as the outcome events
-(`webhooks.announce()`): no change without its event, no event for a change that rolled back.
-Progress events are `low` priority; outcomes and proposals are `important`. Job events go to
-OpenVibe.Events only, not to app webhooks.
+`started`, `retrying`, `succeeded`, `failed`, `cancelled`; subject `job <id>`) commit in one SQLite
+transaction through Media's outbox, the same rule as the outcome events (`webhooks.announce()`): no
+change without its event, no event for a change that rolled back. Progress events are `low`
+priority; outcomes and proposals are `important`. Job events go to OpenVibe.Events only, not to app
+webhooks.
+
+The payload is the event projection (`queue.jobEvent`, contracts `media.job.<transition>@1`), not
+the job as the API answers it:
+
+```json
+{ "app_id": "live", "id": "mjob_…", "object_id": "med_…", "type": "object.split",
+ "status": "succeeded", "attempts": 1, "max_attempts": 3, "error_code": null,
+ "cancel_requested": false, "has_result": true,
+ "run_after": null, "decided_at": null, "created_at": "2026-09-23T21:06:24.000Z",
+ "updated_at": "…Z", "started_at": "…Z", "finished_at": "…Z" }
+```
+
+`object_id` is null for a tenant-wide job (`invariant.scan`). `status` is the job's state after the
+change (`retrying` carries `queued`, `started` carries `running`); times are ISO 8601 UTC or null;
+`error_code` is the stable code only. The tenant's `params`, the `idempotency_key`, the free-text
+`error`, the `result` (a thumbnail URL of a private VOD, for one), `created_by`/`decided_by` and
+`owner_user_id` (tenant-local user ids) are left out: events travel beyond the tenant. A consumer
+that needs them GETs the job with its own tenant-scoped token (`GET /api/v2/:app/jobs/:id`);
+`has_result` says whether there is a result to fetch.
**Idempotency.** `Idempotency-Key` (or `idempotency_key`) is unique per tenant: the same key with the
**Idempotency.** `Idempotency-Key` (or `idempotency_key`) is unique per tenant: the same key with the
same request (type, object, params) answers with the job it made (`Idempotent-Replayed: true`); a
same request (type, object, params) answers with the job it made (`Idempotent-Replayed: true`); a
@@ -10,7 +10,8 @@
@@ -10,7 +10,8 @@
* media.object.uploaded → media.object.uploaded (subject object <id>)
* media.object.uploaded → media.object.uploaded (subject object <id>)
* storage.alert | storage.recovered → media.storage.alert | media.storage.recovered
* storage.alert | storage.recovered → media.storage.alert | media.storage.recovered
* job state changes (server/jobs/queue.js) → media.job.proposed | queued | started | retrying |
* job state changes (server/jobs/queue.js) → media.job.proposed | queued | started | retrying |
- * succeeded | failed | cancelled (subject job <id>; Events only, no webhook)
+ * succeeded | failed | cancelled (subject job <id>; Events only, no webhook;
+ * payload queue.jobEvent, not the tenant's params/result/error)
*
*
* The outbox row is written INSIDE the SQLite transaction that makes the state change it describes
* The outbox row is written INSIDE the SQLite transaction that makes the state change it describes
* (record(), called from webhooks.announce()): the event exists if and only if the change committed,
* (record(), called from webhooks.announce()): the event exists if and only if the change committed,
@@ -99,8 +100,10 @@ function record(webhookEvent, appId, data) {
@@ -99,8 +100,10 @@ function record(webhookEvent, appId, data) {
}
}
/**
/**
- * Queue media.job.<transition> for a job row (queue.jobPublic shape). Same rule as record(): call it
- * inside the transaction that changes the job's state; it throws when the insert fails.
+ * Queue media.job.<transition> for a job (queue.jobEvent: ids, state, counters, ISO times and
+ * has_result; never params, result, error text, idempotency key or who created/decided it; the
+ * tenant-scoped GET /api/v2/:app/jobs/:id has those). Same rule as record(): call it inside the
+ * transaction that changes the job's state; it throws when the insert fails.
*/
*/
function recordJob(transition, job) {
function recordJob(transition, job) {
const priority = JOB_TRANSITIONS[transition];
const priority = JOB_TRANSITIONS[transition];
@@ -28,7 +28,6 @@ const STATUSES = ['proposed', 'queued', 'running', 'succeeded', 'failed', 'cance
@@ -28,7 +28,6 @@ const STATUSES = ['proposed', 'queued', 'running', 'succeeded', 'failed', 'cance
const FINISHED = ['succeeded', 'failed', 'cancelled'];
const FINISHED = ['succeeded', 'failed', 'cancelled'];
const ACTIVE = ['queued', 'running'];
const ACTIVE = ['queued', 'running'];
const MAX_ACTIVE_PER_TENANT = 50; // queued + running jobs one tenant may have (proposals excluded)
const MAX_ACTIVE_PER_TENANT = 50; // queued + running jobs one tenant may have (proposals excluded)
-const RESULT_EVENT_MAX = 8192; // a larger result is left out of the event payload (GET the job for it)
const bus = new EventEmitter();
const bus = new EventEmitter();
bus.setMaxListeners(0);
bus.setMaxListeners(0);
@@ -123,10 +122,41 @@ function jobPublic(row) {
@@ -123,10 +122,41 @@ function jobPublic(row) {
};
};
}
}
-function eventPayload(row) {
- const pub = jobPublic(row);
- if (pub.result && JSON.stringify(pub.result).length > RESULT_EVENT_MAX) pub.result = { omitted: true, reason: 'larger than the event limit; GET the job' };
- return pub;
+/** A SQLite UTC time ('YYYY-MM-DD HH:MM:SS') as ISO 8601 UTC ('…T…Z'), or null. */
+function isoTime(v) {
+ if (v == null || v === '') return null;
+ const s = String(v);
+ const d = new Date(/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(\.\d+)?$/.test(s) ? `${s.replace(' ', 'T')}Z` : s);
+ return Number.isNaN(d.getTime()) ? null : d.toISOString();
+}
+
+/**
+ * The media.job.<transition> event payload: the job's identity, state, counters and times, and whether
+ * it has a result. Deliberately not the tenant's params, the idempotency key, the free-text error, the
+ * result (a thumbnail URL of a private VOD, for one), created_by/decided_by or owner_user_id (tenant-local
+ * user ids): events travel beyond the tenant, and a consumer that needs those GETs the job with its own
+ * tenant-scoped token (GET /api/v2/:app/jobs/:id, queue.jobPublic, unchanged).
+ */
+function jobEvent(row) {
+ if (!row) return null;
+ return {
+ id: row.id,
+ app_id: row.app_id,
+ object_id: row.object_id || null,
+ type: row.job_type,
+ status: row.status,
+ attempts: row.attempts,
+ max_attempts: row.max_attempts,
+ error_code: row.error_code || null,
+ cancel_requested: !!row.cancel_requested,
+ has_result: parseJson(row.result, null) != null,
+ run_after: isoTime(row.run_after),
+ decided_at: isoTime(row.decided_at),
+ created_at: isoTime(row.created_at),
+ updated_at: isoTime(row.updated_at),
+ started_at: isoTime(row.started_at),
+ finished_at: isoTime(row.finished_at),
+ };
}
}
/**
/**
@@ -139,7 +169,7 @@ function commit(id, transition, change) {
@@ -139,7 +169,7 @@ function commit(id, transition, change) {
db.getDb().transaction(() => {
db.getDb().transaction(() => {
if (!change()) return;
if (!change()) return;
row = get(id);
row = get(id);
- if (transition) events.recordJob(transition, eventPayload(row));
+ if (transition) events.recordJob(transition, jobEvent(row));
})();
})();
if (row) {
if (row) {
if (transition) events.kick();
if (transition) events.kick();
@@ -201,7 +231,7 @@ function enqueue({ appId, type, objectId = null, params = {}, status = 'queued',
@@ -201,7 +231,7 @@ function enqueue({ appId, type, objectId = null, params = {}, status = 'queued',
[id, appId, objectId, type, status, key, hash, JSON.stringify(params || {}),
[id, appId, objectId, type, status, key, hash, JSON.stringify(params || {}),
Math.max(1, Math.min(10, Number(maxAttempts) || spec.maxAttempts)), createdBy, ownerUserId ?? null]);
Math.max(1, Math.min(10, Number(maxAttempts) || spec.maxAttempts)), createdBy, ownerUserId ?? null]);
const row = get(id);
const row = get(id);
- events.recordJob(status, eventPayload(row));
+ events.recordJob(status, jobEvent(row));
out = { job: row, created: true, replayed: false, deduped: false };
out = { job: row, created: true, replayed: false, deduped: false };
})();
})();
if (out.created) { events.kick(); bus.emit('change', out.job); }
if (out.created) { events.kick(); bus.emit('change', out.job); }
@@ -254,7 +284,7 @@ function claim(types, { leaseS = 120, id = null } = {}) {
@@ -254,7 +284,7 @@ function claim(types, { leaseS = 120, id = null } = {}) {
if (!update(next.id, 'queued', `status = 'running', attempts = attempts + 1, started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
if (!update(next.id, 'queued', `status = 'running', attempts = attempts + 1, started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
lease_until = datetime('now', '+${Math.round(leaseS)} seconds'), run_after = NULL`)) return;
lease_until = datetime('now', '+${Math.round(leaseS)} seconds'), run_after = NULL`)) return;
row = get(next.id);
row = get(next.id);
- events.recordJob('started', eventPayload(row));
+ events.recordJob('started', jobEvent(row));
})();
})();
if (row) { events.kick(); bus.emit('change', row); }
if (row) { events.kick(); bus.emit('change', row); }
return row;
return row;
@@ -366,7 +396,7 @@ function prune({ days = 30, types = ['thumbnail.regenerate'] } = {}) {
@@ -366,7 +396,7 @@ function prune({ days = 30, types = ['thumbnail.regenerate'] } = {}) {
module.exports = {
module.exports = {
STATUSES, FINISHED, ACTIVE, MAX_ACTIVE_PER_TENANT, JobError, bus,
STATUSES, FINISHED, ACTIVE, MAX_ACTIVE_PER_TENANT, JobError, bus,
register, typeSpec, typeNames, requestHash, parseJson,
register, typeSpec, typeNames, requestHash, parseJson,
- get, getForApp, jobPublic, list, counts,
+ get, getForApp, jobPublic, jobEvent, list, counts,
enqueue, approve, cancel,
enqueue, approve, cancel,
claim, renew, saveCheckpoint, succeed, fail, markCancelled, recoverInterrupted, waitFor, prune,
claim, renew, saveCheckpoint, succeed, fail, markCancelled, recoverInterrupted, waitFor, prune,
};
};
@@ -99,6 +99,10 @@ const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;
@@ -99,6 +99,10 @@ const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;
assert.strictEqual(remux.type, 'object.remux', 'no duration: remux first');
assert.strictEqual(remux.type, 'object.remux', 'no duration: remux first');
assert.strictEqual(split.idempotency_key, `invariant:${bigVod}:object.split`);
assert.strictEqual(split.idempotency_key, `invariant:${bigVod}:object.split`);
assert.deepStrictEqual(jobEvents(split.id), ['media.job.proposed']);
assert.deepStrictEqual(jobEvents(split.id), ['media.job.proposed']);
+ const proposedEv = outbox().find((e) => e.subject.id === split.id).payload;
+ assert.deepStrictEqual([proposedEv.status, proposedEv.type, proposedEv.object_id, proposedEv.decided_at, proposedEv.started_at, proposedEv.has_result], ['proposed', 'object.split', bigVod, null, null, false]);
+ for (const k of ['params', 'idempotency_key', 'created_by', 'decided_by', 'owner_user_id', 'result', 'error']) assert.ok(!(k in proposedEv), `the proposal's event has no ${k}`);
+ assert.ok(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.000Z$/.test(proposedEv.created_at), 'ISO 8601 UTC');
assert.strictEqual(outbox().find((e) => e.subject.id === split.id).priority, 'important', 'a proposal waiting for its owner is important');
assert.strictEqual(outbox().find((e) => e.subject.id === split.id).priority, 'important', 'a proposal waiting for its owner is important');
worker.start();
worker.start();
@@ -210,6 +214,8 @@ const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;
@@ -210,6 +214,8 @@ const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;
assert.strictEqual(sc.status, 0, sc.stderr);
assert.strictEqual(sc.status, 0, sc.stderr);
assert.deepStrictEqual([queue.get(manual.job.id).status, queue.get(manual.job.id).decided_by], ['queued', 'owner']);
assert.deepStrictEqual([queue.get(manual.job.id).status, queue.get(manual.job.id).decided_by], ['queued', 'owner']);
assert.ok(jobEvents(manual.job.id).includes('media.job.queued'), "the script's change queued its event in the outbox");
assert.ok(jobEvents(manual.job.id).includes('media.job.queued'), "the script's change queued its event in the outbox");
+ const approvedEv = outbox().filter((e) => e.subject.id === manual.job.id && e.event_type === 'media.job.queued').pop().payload;
+ assert.ok(/Z$/.test(approvedEv.decided_at) && !('decided_by' in approvedEv) && !JSON.stringify(approvedEv).includes('owner'), 'the approval is dated, its decider is not published');
sc = script('cancel', manual.job.id, '--by', 'owner');
sc = script('cancel', manual.job.id, '--by', 'owner');
assert.strictEqual(sc.status, 0, sc.stderr);
assert.strictEqual(sc.status, 0, sc.stderr);
assert.strictEqual(queue.get(manual.job.id).status, 'cancelled');
assert.strictEqual(queue.get(manual.job.id).status, 'cancelled');
@@ -1,6 +1,8 @@
@@ -1,6 +1,8 @@
'use strict';
'use strict';
// Media job system (server/jobs): the schema-only media_jobs table is rebuilt; every state change
// Media job system (server/jobs): the schema-only media_jobs table is rebuilt; every state change
-// commits with its media.job.* outbox row (and rolls back with it); idempotency keys; retries with
+// commits with its media.job.* outbox row (and rolls back with it), whose payload is the event
+// projection (ids, state, counters, ISO times, has_result; never params, result, error text, the
+// idempotency key or who created/decided it); idempotency keys; retries with
// backoff, permanent failures, checkpoints; cancellation by the owner (queued at once, running through
// backoff, permanent failures, checkpoints; cancellation by the owner (queued at once, running through
// the abort signal) and not by anyone else; interrupted jobs are requeued at start; the v1 thumbnail
// the abort signal) and not by anyone else; interrupted jobs are requeued at start; the v1 thumbnail
// route runs as a thumbnail.regenerate job and answers as before.
// route runs as a thumbnail.regenerate job and answers as before.
@@ -78,6 +80,23 @@ const waitFor = async (fn, ms = 5000) => {
@@ -78,6 +80,23 @@ const waitFor = async (fn, ms = 5000) => {
const conn = db.getDb();
const conn = db.getDb();
const outbox = () => conn.prepare('SELECT envelope FROM event_outbox ORDER BY id').all().map((r) => JSON.parse(r.envelope));
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);
const jobEvents = (id) => outbox().filter((e) => e.subject.type === 'job' && e.subject.id === id).map((e) => e.event_type);
+ const jobPayloads = (id) => outbox().filter((e) => e.subject.type === 'job' && e.subject.id === id).map((e) => e.payload);
+ // The media.job.* payload: exactly these fields. The rest stays behind the tenant-scoped jobs API.
+ const EVENT_KEYS = ['app_id', 'attempts', 'cancel_requested', 'created_at', 'decided_at', 'error_code', 'finished_at', 'has_result',
+ 'id', 'max_attempts', 'object_id', 'run_after', 'started_at', 'status', 'type', 'updated_at'];
+ const LEFT_OUT = ['params', 'idempotency_key', 'error', 'result', 'created_by', 'decided_by', 'owner_user_id'];
+ const TIMES = ['run_after', 'decided_at', 'created_at', 'updated_at', 'started_at', 'finished_at'];
+ const ISO = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/;
+ const TRANSITION_STATUS = { proposed: 'proposed', queued: 'queued', started: 'running', retrying: 'queued', succeeded: 'succeeded', failed: 'failed', cancelled: 'cancelled' };
+ const assertJobEvent = (env) => {
+ const p = env.payload, what = `${env.event_type} ${p.id}`;
+ assert.deepStrictEqual(Object.keys(p).sort(), EVENT_KEYS, `${what}: exactly the event projection`);
+ for (const k of LEFT_OUT) assert.ok(!(k in p), `${what}: no ${k}`);
+ for (const k of TIMES) assert.ok(p[k] === null || ISO.test(p[k]), `${what}: ${k} is ISO 8601 UTC or null (${p[k]})`);
+ assert.ok(ISO.test(p.created_at) && ISO.test(p.updated_at), `${what}: created_at and updated_at are always set`);
+ assert.strictEqual(typeof p.has_result, 'boolean');
+ assert.strictEqual(p.status, TRANSITION_STATUS[env.event_type.slice('media.job.'.length)], `${what}: status`);
+ };
const failOutbox = (on) => conn.exec(on
const failOutbox = (on) => conn.exec(on
? "CREATE TEMP TRIGGER outbox_boom BEFORE INSERT ON event_outbox BEGIN SELECT RAISE(ABORT, 'outbox insert failed'); END"
? "CREATE TEMP TRIGGER outbox_boom BEFORE INSERT ON event_outbox BEGIN SELECT RAISE(ABORT, 'outbox insert failed'); END"
: 'DROP TRIGGER IF EXISTS temp.outbox_boom');
: 'DROP TRIGGER IF EXISTS temp.outbox_boom');
@@ -112,6 +131,11 @@ const waitFor = async (fn, ms = 5000) => {
@@ -112,6 +131,11 @@ const waitFor = async (fn, ms = 5000) => {
assert.deepStrictEqual(jobEvents(a.job.id), ['media.job.queued']);
assert.deepStrictEqual(jobEvents(a.job.id), ['media.job.queued']);
const env = outbox().find((e) => e.subject.id === a.job.id);
const env = outbox().find((e) => e.subject.id === a.job.id);
assert.deepStrictEqual([env.source, env.priority, env.visibility, env.payload.type, env.payload.status, env.payload.app_id], ['media', 'low', 'internal', 'test.flaky', 'queued', 'live']);
assert.deepStrictEqual([env.source, env.priority, env.visibility, env.payload.type, env.payload.status, env.payload.app_id], ['media', 'low', 'internal', 'test.flaky', 'queued', 'live']);
+ assertJobEvent(env);
+ const aRow = queue.get(a.job.id);
+ assert.deepStrictEqual([env.payload.created_at, env.payload.updated_at], [aRow.created_at, aRow.updated_at].map((t) => new Date(`${t.replace(' ', 'T')}Z`).toISOString()), 'times are the row\'s, in ISO 8601 UTC');
+ assert.deepStrictEqual([env.payload.attempts, env.payload.max_attempts, env.payload.error_code, env.payload.cancel_requested, env.payload.has_result, env.payload.object_id, env.payload.run_after, env.payload.started_at, env.payload.finished_at, env.payload.decided_at],
+ [0, 3, null, false, false, null, null, null, null, null]);
const q2 = queue.enqueue({ appId: 'live', type: 'test.permanent' });
const q2 = queue.enqueue({ appId: 'live', type: 'test.permanent' });
failOutbox(true);
failOutbox(true);
assert.throws(() => queue.enqueue({ appId: 'live', type: 'test.flaky', params: { x: 2 } }), /outbox insert failed/);
assert.throws(() => queue.enqueue({ appId: 'live', type: 'test.flaky', params: { x: 2 } }), /outbox insert failed/);
@@ -126,6 +150,7 @@ const waitFor = async (fn, ms = 5000) => {
@@ -126,6 +150,7 @@ const waitFor = async (fn, ms = 5000) => {
const k1 = queue.enqueue({ appId: 'live', type: 'test.permanent', params: { n: 1 }, idempotencyKey: 'key-1' });
const k1 = queue.enqueue({ appId: 'live', type: 'test.permanent', params: { n: 1 }, idempotencyKey: 'key-1' });
const k2 = queue.enqueue({ appId: 'live', type: 'test.permanent', params: { n: 1 }, idempotencyKey: 'key-1' });
const k2 = queue.enqueue({ appId: 'live', type: 'test.permanent', params: { n: 1 }, idempotencyKey: 'key-1' });
assert.ok(k1.created && k2.replayed && k1.job.id === k2.job.id);
assert.ok(k1.created && k2.replayed && k1.job.id === k2.job.id);
+ assert.ok(!JSON.stringify(jobPayloads(k1.job.id)).includes('key-1'), 'the idempotency key stays out of the event');
assert.throws(() => queue.enqueue({ appId: 'live', type: 'test.permanent', params: { n: 2 }, idempotencyKey: 'key-1' }), (e) => e.code === 'media.job.idempotency_conflict' && e.status === 409);
assert.throws(() => queue.enqueue({ appId: 'live', type: 'test.permanent', params: { n: 2 }, idempotencyKey: 'key-1' }), (e) => e.code === 'media.job.idempotency_conflict' && e.status === 409);
assert.ok(queue.enqueue({ appId: 'games', type: 'test.permanent', params: { n: 1 }, idempotencyKey: 'key-1' }).created, 'keys are per tenant');
assert.ok(queue.enqueue({ appId: 'games', type: 'test.permanent', params: { n: 1 }, idempotencyKey: 'key-1' }).created, 'keys are per tenant');
const d1 = queue.enqueue({ appId: 'live', type: 'test.permanent', params: { same: true }, dedupeActive: true });
const d1 = queue.enqueue({ appId: 'live', type: 'test.permanent', params: { same: true }, dedupeActive: true });
@@ -150,7 +175,18 @@ const waitFor = async (fn, ms = 5000) => {
@@ -150,7 +175,18 @@ const waitFor = async (fn, ms = 5000) => {
assert.strictEqual(queue.get(always.job.id).attempts, 2, 'retried until max_attempts');
assert.strictEqual(queue.get(always.job.id).attempts, 2, 'retried until max_attempts');
assert.deepStrictEqual(jobEvents(always.job.id), ['media.job.queued', 'media.job.started', 'media.job.retrying', 'media.job.started', 'media.job.failed']);
assert.deepStrictEqual(jobEvents(always.job.id), ['media.job.queued', 'media.job.started', 'media.job.retrying', 'media.job.started', 'media.job.failed']);
const failedEv = outbox().filter((e) => e.subject.id === always.job.id).pop();
const failedEv = outbox().filter((e) => e.subject.id === always.job.id).pop();
- assert.deepStrictEqual([failedEv.priority, failedEv.payload.error], ['important', 'nope']);
+ assert.deepStrictEqual([failedEv.priority, failedEv.payload.error_code, failedEv.payload.has_result, failedEv.payload.attempts], ['important', null, false, 2]);
+ assert.ok(!JSON.stringify(failedEv.payload).includes('nope'), 'the free-text error stays out of the event');
+ assert.strictEqual(queue.jobPublic(queue.get(always.job.id)).error, 'nope', 'GET the job for it');
+ const permEv = outbox().filter((e) => e.subject.id === perm.job.id).pop();
+ assert.deepStrictEqual([permEv.event_type, permEv.payload.error_code, permEv.payload.attempts], ['media.job.failed', 'bad_input', 1], 'the stable code is in the event');
+ assert.ok(!JSON.stringify(permEv.payload).includes('will never work'));
+ const [aQueued, aStarted, aRetrying, aStarted2, aSucceeded] = jobPayloads(a.job.id);
+ assert.deepStrictEqual([aQueued.status, aStarted.status, aRetrying.status, aStarted2.status, aSucceeded.status], ['queued', 'running', 'queued', 'running', 'succeeded']);
+ assert.ok(ISO.test(aRetrying.run_after) && ISO.test(aStarted.started_at) && aRetrying.finished_at === null, 'retrying: a backoff time, started, not finished');
+ assert.deepStrictEqual([aStarted.attempts, aStarted2.attempts, aStarted2.started_at], [1, 2, aStarted.started_at], 'started_at stays the first attempt\'s start');
+ assert.deepStrictEqual([aSucceeded.has_result, ISO.test(aSucceeded.finished_at), aSucceeded.error_code], [true, true, null]);
+ assert.ok(!JSON.stringify(aSucceeded).includes('resumed_from') && !JSON.stringify(jobPayloads(a.job.id)).includes('"x"'), 'neither the result nor the params are in the events');
console.log('✅ worker: retries with backoff resume from the checkpoint; permanent failures stop; attempts are capped');
console.log('✅ worker: retries with backoff resume from the checkpoint; permanent failures stop; attempts are capped');
// ── 5. Cancellation by the owner ──
// ── 5. Cancellation by the owner ──
@@ -181,6 +217,10 @@ const waitFor = async (fn, ms = 5000) => {
@@ -181,6 +217,10 @@ const waitFor = async (fn, ms = 5000) => {
assert.deepStrictEqual([r.status, r.body.job.cancel_requested], [202, true], 'a running job stops at its next check');
assert.deepStrictEqual([r.status, r.body.job.cancel_requested], [202, true], 'a running job stops at its next check');
assert.ok(await waitFor(() => queue.get(slow.job.id).status === 'cancelled'), 'the abort signal stopped it');
assert.ok(await waitFor(() => queue.get(slow.job.id).status === 'cancelled'), 'the abort signal stopped it');
assert.deepStrictEqual(jobEvents(slow.job.id), ['media.job.queued', 'media.job.started', 'media.job.cancelled']);
assert.deepStrictEqual(jobEvents(slow.job.id), ['media.job.queued', 'media.job.started', 'media.job.cancelled']);
+ const slowCancelled = jobPayloads(slow.job.id).pop();
+ assert.deepStrictEqual([slowCancelled.cancel_requested, slowCancelled.error_code, ISO.test(slowCancelled.finished_at)], [true, 'cancelled', true]);
+ assert.ok(!/user:7|"owner_user_id"|"created_by"|"decided_by"/.test(JSON.stringify(jobPayloads(slow.job.id))), 'who created or cancelled it (tenant-local user ids) stays out of the events');
+ assert.deepStrictEqual([queue.jobPublic(queue.get(slow.job.id)).created_by, queue.jobPublic(queue.get(slow.job.id)).owner_user_id], ['app:live:user:7', 7], 'the jobs API still has them');
r = await call('DELETE', `/api/v2/live/jobs/${slow.job.id}`);
r = await call('DELETE', `/api/v2/live/jobs/${slow.job.id}`);
assert.deepStrictEqual([r.status, r.body.code], [409, 'media.job.finished']);
assert.deepStrictEqual([r.status, r.body.code], [409, 'media.job.finished']);
const waiting = queue.enqueue({ appId: 'live', type: 'test.permanent', runAfterS: 3600 });
const waiting = queue.enqueue({ appId: 'live', type: 'test.permanent', runAfterS: 3600 });
@@ -196,6 +236,8 @@ const waitFor = async (fn, ms = 5000) => {
@@ -196,6 +236,8 @@ const waitFor = async (fn, ms = 5000) => {
const back = queue.get(orphan.job.id);
const back = queue.get(orphan.job.id);
assert.deepStrictEqual([back.status, back.error_code], ['queued', 'interrupted']);
assert.deepStrictEqual([back.status, back.error_code], ['queued', 'interrupted']);
assert.deepStrictEqual(jobEvents(orphan.job.id).slice(-1), ['media.job.retrying']);
assert.deepStrictEqual(jobEvents(orphan.job.id).slice(-1), ['media.job.retrying']);
+ const orphanRetry = jobPayloads(orphan.job.id).pop();
+ assert.deepStrictEqual([orphanRetry.status, orphanRetry.error_code, ISO.test(orphanRetry.run_after)], ['queued', 'interrupted', true]);
queue.cancel(orphan.job.id, { by: 'test' });
queue.cancel(orphan.job.id, { by: 'test' });
console.log('✅ a job left running by a dead process is requeued (or failed when out of attempts)');
console.log('✅ a job left running by a dead process is requeued (or failed when out of attempts)');
@@ -213,6 +255,9 @@ const waitFor = async (fn, ms = 5000) => {
@@ -213,6 +255,9 @@ const waitFor = async (fn, ms = 5000) => {
assert.deepStrictEqual([r.status, r.body.code], [409, 'media.job.not_proposed']);
assert.deepStrictEqual([r.status, r.body.code], [409, 'media.job.not_proposed']);
r = await call('GET', `/api/v2/live/jobs?type=invariant.scan`);
r = await call('GET', `/api/v2/live/jobs?type=invariant.scan`);
assert.deepStrictEqual(r.body.jobs.map((j) => j.id), [scanId]);
assert.deepStrictEqual(r.body.jobs.map((j) => j.id), [scanId]);
+ r = await call('GET', `/api/v2/live/jobs/${scanId}`);
+ assert.deepStrictEqual([r.status, r.body.job.idempotency_key, r.body.job.params, r.body.job.created_by, r.body.job.created_at.includes('T')], [200, 'scan-1', {}, 'app:live', false],
+ 'GET the job answers as before (queue.jobPublic: params, idempotency key, creator, SQLite times)');
r = await call('GET', `/api/v2/games/jobs/${scanId}`, { key: 'games-key' });
r = await call('GET', `/api/v2/games/jobs/${scanId}`, { key: 'games-key' });
assert.strictEqual(r.status, 404, 'tenants are isolated');
assert.strictEqual(r.status, 404, 'tenants are isolated');
r = await call('GET', `/api/v2/live/jobs?status=bogus`);
r = await call('GET', `/api/v2/live/jobs?status=bogus`);
@@ -235,6 +280,9 @@ const waitFor = async (fn, ms = 5000) => {
@@ -235,6 +280,9 @@ const waitFor = async (fn, ms = 5000) => {
const tj = queue.get(t1.body.job_id);
const tj = queue.get(t1.body.job_id);
assert.deepStrictEqual([tj.job_type, tj.status, tj.object_id], ['thumbnail.regenerate', 'succeeded', db.get('SELECT object_id FROM vods WHERE id = 1').object_id]);
assert.deepStrictEqual([tj.job_type, tj.status, tj.object_id], ['thumbnail.regenerate', 'succeeded', db.get('SELECT object_id FROM vods WHERE id = 1').object_id]);
assert.deepStrictEqual(jobEvents(tj.id), ['media.job.queued', 'media.job.started', 'media.job.succeeded']);
assert.deepStrictEqual(jobEvents(tj.id), ['media.job.queued', 'media.job.started', 'media.job.succeeded']);
+ const thumbDone = jobPayloads(tj.id).pop();
+ assert.deepStrictEqual([thumbDone.has_result, thumbDone.object_id], [true, tj.object_id]);
+ assert.ok(!JSON.stringify(jobPayloads(tj.id)).includes('/t/vod-1'), 'a thumbnail URL (of a VOD that may be private) stays out of the event');
r = await call('POST', '/api/v1/live/thumbnails/vod/2');
r = await call('POST', '/api/v1/live/thumbnails/vod/2');
assert.deepStrictEqual([r.status, r.body.error], [404, 'Media file unavailable'], 'no media is still a 404');
assert.deepStrictEqual([r.status, r.body.error], [404, 'Media file unavailable'], 'no media is still a 404');
assert.strictEqual(queue.get(r.body.job_id).error_code, 'media_unavailable');
assert.strictEqual(queue.get(r.body.job_id).error_code, 'media_unavailable');
@@ -248,6 +296,13 @@ const waitFor = async (fn, ms = 5000) => {
@@ -248,6 +296,13 @@ const waitFor = async (fn, ms = 5000) => {
assert.ok(await waitFor(() => queue.get(r.body.job.id).status === 'succeeded'));
assert.ok(await waitFor(() => queue.get(r.body.job.id).status === 'succeeded'));
console.log('✅ the v1 thumbnail route runs a thumbnail.regenerate job and answers { url } as before (404 without media)');
console.log('✅ the v1 thumbnail route runs a thumbnail.regenerate job and answers { url } as before (404 without media)');
+ // Every job event this run queued has the projection's shape and its transition's status.
+ const all = outbox().filter((e) => e.event_type.startsWith('media.job.'));
+ assert.ok(all.length > 30);
+ for (const e of all) assertJobEvent(e);
+ assert.deepStrictEqual([...new Set(all.map((e) => e.event_type))].sort(), Object.keys(TRANSITION_STATUS).filter((t) => t !== 'proposed').map((t) => `media.job.${t}`).sort());
+ console.log(`✅ ${all.length} media.job.* events: the projection only (no params, result, error text, idempotency key, creator/decider or owner user id), ISO 8601 UTC times`);
+
worker.stop();
worker.stop();
events._reset();
events._reset();
server.close();
server.close();