openvibers/openvibe.media Commits
Owner subjects: every media object that names an app-local owner (owner_app + owner_user_id) gets the owner's canonical Network subject in owner_subject, so Live's lineage resolver and other consumers read owner.subject instead of translating Live user ids (roadmap D01/D20; 0 of 3,032 user-owned objects had one). scripts/backfill-owner-subject.js resolves the owners through Network's POST /internal/identity/resolve-batch (500 ids per call). It is a dry run by default and prints per tenant the objects to fill, already set, unresolvable, in an unsupported tenant and with no owner. --apply requires --backup <file.json>: it takes an online backup to <file>.media.db and checks it with integrity_check, writes the rows it will change to <file.json>, then fills them 500 per transaction. It never writes over a non-null owner_subject or a row whose owner changed. --rollback <file.json> [--apply] clears only the rows that still carry the subject the backfill wrote. The script opens SQLite directly, so none of the service's boot work (marking in-progress clip cuts failed) runs. New objects: X-OV-Subject is still stored at creation, and a reconcile job (every MEDIA_OWNER_SUBJECT_INTERVAL_MIN, default 10; MEDIA_OWNER_SUBJECT_SYNC=0 stops it) fills every object that names only an owner id: projections, and v2 uploads that send only user_id. It asks Network only when something is missing, and uploads never wait on it. A re-projection to a different owner drops the subject until the job resolves the new one. Network is asked with a service token (identity.subject.resolve, Network c3b7cf3), and with INTERNAL_API_KEY until Network grants it. Only live ids are resolved (SOURCE_SYSTEMS); no other tenant stores owner ids today
Changes
@@@ -33,6 +33,12 @@ OV_NETWORK_URL=https://openvibe.networkOV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000# Per-service key for Network's /internal/* endpoints (header X-Internal-Key).INTERNAL_API_KEY=++# Owner subjects (docs/object-model.md#owner-subjects): objects that name only an app-local owner++# get the owner's Network subject from POST /internal/identity/resolve-batch, with a service token++# (OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET, capability identity.subject.resolve) or else the key.++# MEDIA_OWNER_SUBJECT_SYNC=0 stops the reconcile job (before rolling a backfill back).++MEDIA_OWNER_SUBJECT_SYNC=++MEDIA_OWNER_SUBJECT_INTERVAL_MIN=10# ── Tenant (app) seeding ─────────────────────────────────────# JSON array, upserted into the `apps` table on every boot (api_key is hashed,@@@ -114,6 +114,57 @@ object `deleted` whenever its row is deleted, which covers all of the inheritednever throw: the legacy write has already happened, so a failed sync only logs a warning. The nextsync or backfill repairs it.++## Owner subjects++++`owner_subject` names an object's owner as a Network subject (`usr_<ULID>`), so consumers such as++Live's lineage resolver (which reads `owner.subject` on the v2 object) never translate an app's own++user ids. An object gets it in one of three ways:++++- **At creation,** when the caller names it: `X-OV-Subject` on `POST /api/v2/:app/objects`. Tools and++ developer-project uploads do this. A derived object (split, remux) copies its source's.++- **From the reconcile job** (`server/objects/owner-subject-job.js`), for every object that names only++ `owner_app` + `owner_user_id`: the projected vods, clips, files, screenshots, avatars and thumbnails,++ and v2 uploads that send only `user_id`. Every `MEDIA_OWNER_SUBJECT_INTERVAL_MIN` minutes (first run++ one minute after boot) it collects the distinct owners that still lack a subject and asks Network's++ `POST /internal/identity/resolve-batch` (`{ system, type: 'user', ids }`, 500 ids per call). It then++ fills their objects. Upload and projection paths never wait on Network. When nothing is missing, a++ run asks Network nothing. An owner Network does not know (a Live account not linked to Network) stays++ `NULL` until a later run finds it.++- **From the one-off backfill** `scripts/backfill-owner-subject.js`, for everything that existed before++ the job.++++Rules for both job and backfill:++++- A non-null `owner_subject` is never overwritten. Every write is guarded by `owner_subject IS NULL`++ and by the owner it was resolved for.++- Only the tenants whose user-id space is known are resolved (`SOURCE_SYSTEMS` in++ `server/objects/owner-subject.js`: `live` -> `live`). Others are reported and left alone.++- A re-projection keeps the subject, unless it changes the owner (`owner_app`/`owner_user_id`). Then++ the subject is dropped until the job resolves the new owner.++- Network is asked with a service token (`OV_OAUTH_CLIENT_ID`/`OV_OAUTH_CLIENT_SECRET`, capability++ `identity.subject.resolve`, audience `openvibe.network`). When Network has not granted it, the++ request uses `INTERNAL_API_KEY`.++++```++node scripts/backfill-owner-subject.js [--db media.db] [--batch 500] [--json] # dry run: counts per tenant++node scripts/backfill-owner-subject.js --apply --backup <file.json> [--batch 500] # fill++node scripts/backfill-owner-subject.js --rollback <file.json> [--apply] # undo (dry without --apply)++```++++- **The dry run** (the default) opens the database read-only and prints per tenant: objects, already++ set, to fill, unresolvable, unsupported tenant and no owner.++- **`--apply --backup <file.json>`** first takes an online backup to `<file>.media.db` and requires++ `PRAGMA integrity_check` = `ok` on it. It then writes `<file.json>` (0600), the rows it is about to++ change, and fills them in `--batch` transactions. Finally it rewrites the file to list exactly the++ rows it changed. It refuses an existing backup name, and a re-run fills only what is still missing.++- **`--rollback <file.json> --apply`** sets `owner_subject` back to `NULL` on each listed row that++ still carries the subject the backfill wrote for the same owner. Rows changed since are counted and++ left alone. Without `--apply` it only reports, which also verifies an applied backfill. Stop the job++ first (`MEDIA_OWNER_SUBJECT_SYNC=0`, restart), or its next run fills the rows again.++++The script opens the database directly, not through `server/db/database.js`, so none of the service's++boot work runs (for example, marking in-progress clip cuts failed).++## Object API v2All routes live under `/api/v2/:app/objects`.@@@ -499,6 +550,8 @@ Not run against production yet: it needs the owner's go-ahead.| `MEDIA_VERIFY_HASH_MAX_MB` | 64 | sha256 local copies up to this size (objects with a `content_hash`) || `MEDIA_VERIFY_MAX_REUPLOADS` | 2 | missing remote copies restored per run || `MEDIA_VERIFY_REPAIR_CORRUPT` | off | `1` also overwrites a corrupt remote copy from a good local copy |++| `MEDIA_OWNER_SUBJECT_SYNC` | on | `0` turns the owner-subject reconcile job off |++| `MEDIA_OWNER_SUBJECT_INTERVAL_MIN` | 10 | minutes between owner-subject reconcile runs |## Not in this passnnew file mode 100644@@@ -0,0 +1,212 @@++#!/usr/bin/env node++'use strict';++/**++ * Give every media_object that names an app-local owner (owner_app + owner_user_id) its owner's++ * canonical Network subject (owner_subject, usr_<ULID>), resolved through Network's++ * POST /internal/identity/resolve-batch (docs/object-model.md#owner-subjects). Once this has run,++ * the service's reconcile job (server/objects/owner-subject-job.js) keeps new objects filled.++ *++ * node scripts/backfill-owner-subject.js [--db media.db] [--batch 500] [--json]++ * Dry run (the default): asks Network, changes nothing, and prints per tenant the objects++ * to fill, already set, unresolvable (Network does not know the owner), with no owner, and++ * those of tenants whose user ids it has no source system for.++ * node scripts/backfill-owner-subject.js --apply --backup <file.json> [--db media.db] [--batch 500]++ * First takes an online backup of the database to <file>.media.db and checks it with++ * PRAGMA integrity_check, then writes <file.json> (every row it is about to change, for++ * --rollback). Only then fills owner_subject, <batch> rows per transaction, never over a++ * non-null value and only while the row still has the owner it was resolved for.++ * Refuses to overwrite either backup file. Re-running fills only what is still missing.++ * node scripts/backfill-owner-subject.js --rollback <file.json> [--apply] [--db media.db]++ * Clears owner_subject again on the rows the file lists that still carry exactly the subject++ * the backfill wrote (rows changed since are left alone and counted). Without --apply it only++ * reports what it would restore, which also verifies an applied backfill.++ * Turn the service's reconcile job off first (MEDIA_OWNER_SUBJECT_SYNC=0 and a restart), or++ * its next run fills the rows again.++ *++ * Network: OV_NETWORK_INTERNAL_URL (default http://127.0.0.1:4000), with a service token++ * (OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET, capability identity.subject.resolve) or else++ * INTERNAL_API_KEY: the names /etc/openvibe/media.env already uses. The database is DB_PATH.++ *++ * The database is opened directly (read-only for a dry run), not through server/db/database.js,++ * whose boot work (migrations, marking in-progress clip cuts failed) must stay with the service.++ * Exit codes: 0 done, 1 error, 2 bad usage.++ */++const fs = require('fs');++const path = require('path');++++const args = process.argv.slice(2);++const has = (name) => args.includes(`--${name}`);++const arg = (name) => { const i = args.indexOf(`--${name}`); return i >= 0 && args[i + 1] && !args[i + 1].startsWith('--') ? args[i + 1] : null; };++const usage = (msg) => { console.error(`backfill-owner-subject: ${msg}`); process.exit(2); };++++for (const flag of ['db', 'backup', 'rollback', 'batch']) if (has(flag) && !arg(flag)) usage(`--${flag} needs a value`);++if (arg('db')) process.env.DB_PATH = path.resolve(arg('db'));++++const Database = require('better-sqlite3');++const config = require('../server/config');++const ownerSubject = require('../server/objects/owner-subject');++++const APPLY = has('apply');++const JSON_OUT = has('json');++const BATCH = arg('batch') ? Number(arg('batch')) : ownerSubject.DEFAULT_BATCH;++if (!Number.isInteger(BATCH) || BATCH < 1 || BATCH > 5000) usage('--batch must be an integer 1-5000');++const DB_PATH = config.db.path;++const ROLLBACK = arg('rollback') ? path.resolve(arg('rollback')) : null;++const BACKUP = arg('backup') ? path.resolve(arg('backup')) : null;++if (ROLLBACK && BACKUP) usage('--rollback and --backup do not go together');++if (APPLY && !ROLLBACK && !BACKUP) usage('--apply needs --backup <file.json> (the rollback file; the database copy goes next to it)');++if (!fs.existsSync(DB_PATH)) usage(`no database at ${DB_PATH} (set DB_PATH or pass --db)`);++++const log = (...a) => { if (!JSON_OUT) console.log(...a); };++++function open({ readonly }) {++ const sqlite = new Database(DB_PATH, { readonly, fileMustExist: true });++ sqlite.pragma('busy_timeout = 15000');++ return sqlite;++}++++function writePrivate(file, text) {++ fs.mkdirSync(path.dirname(file), { recursive: true, mode: 0o700 });++ const tmp = `${file}.tmp-${process.pid}`;++ const fd = fs.openSync(tmp, 'wx', 0o600);++ try { fs.writeSync(fd, text); fs.fsyncSync(fd); } finally { fs.closeSync(fd); }++ fs.renameSync(tmp, file);++}++++function dbCopyPath(file) {++ return file.replace(/\.json$/i, '') + '.media.db';++}++++/** Online backup (better-sqlite3's page-by-page copy; the service keeps writing), then integrity_check. */++async function backupDatabase(sqlite, dest) {++ fs.mkdirSync(path.dirname(dest), { recursive: true, mode: 0o700 });++ await sqlite.backup(dest);++ fs.chmodSync(dest, 0o600);++ const copy = new Database(dest, { readonly: true, fileMustExist: true });++ try {++ const check = copy.pragma('integrity_check', { simple: true });++ const objects = copy.prepare('SELECT count(*) AS n FROM media_objects').get().n;++ return { path: dest, bytes: fs.statSync(dest).size, integrity_check: check, objects };++ } finally { copy.close(); }++}++++function printSummary(before, plan) {++ const apps = [...new Set([...Object.keys(before), ...Object.keys(plan.perApp)])].sort();++ log(`Database: ${DB_PATH}`);++ log(`Network: ${config.network.internalUrl} (asked via ${plan.via || 'nothing: no owner to resolve'})`);++ log('');++ log('app objects already set to fill unresolvable unsupported app no owner');++ for (const app of apps) {++ const b = before[app] || { objects: 0, already_set: 0, no_owner: 0 };++ const p = plan.perApp[app] || { to_fill: 0, unresolvable: 0, unsupported: 0 };++ log(`${app.padEnd(40)} ${String(b.objects).padStart(7)} ${String(b.already_set).padStart(11)} ${String(p.to_fill).padStart(7)} ${String(p.unresolvable).padStart(12)} ${String(p.unsupported).padStart(15)} ${String(b.no_owner).padStart(8)}`);++ }++ const t = plan.totals;++ log(`${'total'.padEnd(40)} ${String(t.objects).padStart(7)} ${String(t.already_set).padStart(11)} ${String(t.to_fill).padStart(7)} ${String(t.unresolvable).padStart(12)} ${String(t.unsupported).padStart(15)} ${String(t.no_owner).padStart(8)}`);++ if (plan.unresolvableOwners.length) {++ log('');++ log(`Unresolvable owners (Network has no subject for them; their objects stay NULL): ${plan.unresolvableOwners.map(o => `${o.owner_app}:${o.owner_user_id} (${o.objects})`).join(', ')}`);++ }++ if (plan.unsupportedOwners.length) {++ log(`Owners in tenants with no known source system (left alone): ${plan.unsupportedOwners.map(o => `${o.owner_app}:${o.owner_user_id} (${o.objects})`).join(', ')}`);++ }++}++++async function makePlan(sqlite) {++ const before = ownerSubject.summary(sqlite);++ const owners = ownerSubject.pendingOwners(sqlite);++ const resolver = ownerSubject.createResolver();++ const r = owners.length ? await ownerSubject.resolveOwners(owners, { resolver }) : { subjects: new Map(), unresolvable: [], unsupported: [] };++ const changes = ownerSubject.planChanges(sqlite, r.subjects, { batch: BATCH });++ const perApp = {};++ const bump = (app, k, n) => { perApp[app] = perApp[app] || { to_fill: 0, unresolvable: 0, unsupported: 0 }; perApp[app][k] += n; };++ for (const c of changes) bump(c.app_id, 'to_fill', 1);++ // Per tenant of the object (owner_app is the same as app_id for every object Media makes today).++ const countBy = (list) => {++ const m = new Map();++ for (const o of list) {++ for (const row of sqlite.prepare(`SELECT app_id, count(*) AS n FROM media_objects WHERE owner_subject IS NULL++ AND COALESCE(owner_app, app_id) = ? AND owner_user_id = ? GROUP BY app_id`).all(o.owner_app, o.owner_user_id)) {++ m.set(row.app_id, (m.get(row.app_id) || 0) + row.n);++ }++ }++ return m;++ };++ for (const [app, n] of countBy(r.unresolvable)) bump(app, 'unresolvable', n);++ for (const [app, n] of countBy(r.unsupported)) bump(app, 'unsupported', n);++ const totals = { objects: 0, already_set: 0, no_owner: 0, to_fill: changes.length, unresolvable: 0, unsupported: 0 };++ for (const b of Object.values(before)) { totals.objects += b.objects; totals.already_set += b.already_set; totals.no_owner += b.no_owner; }++ for (const p of Object.values(perApp)) { totals.unresolvable += p.unresolvable; totals.unsupported += p.unsupported; }++ return { before, perApp, totals, changes, unresolvableOwners: r.unresolvable, unsupportedOwners: r.unsupported, via: resolver.state.via, network_calls: resolver.state.calls, token_error: resolver.state.token_error };++}++++async function backfill() {++ const sqlite = open({ readonly: !APPLY });++ try {++ const plan = await makePlan(sqlite);++ printSummary(plan.before, plan);++ if (plan.token_error && plan.via === 'internal-key') log(`(no service token: ${plan.token_error}; used INTERNAL_API_KEY)`);++ const report = { mode: APPLY ? 'apply' : 'dry-run', db: DB_PATH, network: config.network.internalUrl, via: plan.via, before: plan.before, per_app: plan.perApp, totals: plan.totals,++ unresolvable_owners: plan.unresolvableOwners, unsupported_owners: plan.unsupportedOwners };++ if (!APPLY) {++ log('');++ log('Dry run: nothing was changed. Re-run with --apply --backup <file.json> to fill.');++ if (JSON_OUT) console.log(JSON.stringify(report, null, 2));++ return 0;++ }++ if (!plan.changes.length) {++ log('');++ log('Nothing to fill; no backup taken, nothing changed.');++ if (JSON_OUT) console.log(JSON.stringify({ ...report, applied: 0 }, null, 2));++ return 0;++ }++++ const copyPath = dbCopyPath(BACKUP);++ for (const f of [BACKUP, copyPath]) if (fs.existsSync(f)) usage(`${f} already exists; pick a new --backup name`);++ log('');++ log(`Backing up ${DB_PATH} -> ${copyPath} ...`);++ const copy = await backupDatabase(sqlite, copyPath);++ if (copy.integrity_check !== 'ok') { console.error(`backup failed integrity_check: ${copy.integrity_check}; nothing changed`); return 1; }++ log(` ${copy.bytes} bytes, integrity_check ok, ${copy.objects} objects`);++ const manifest = {++ kind: 'media.owner_subject.backfill', version: 1, created_at: new Date().toISOString(), db: DB_PATH, db_backup: copy.path,++ status: 'planned', rows: plan.changes.map(c => ({ id: c.id, app_id: c.app_id, owner_app: c.owner_app, owner_user_id: c.owner_user_id, owner_subject: c.owner_subject })),++ };++ writePrivate(BACKUP, JSON.stringify(manifest, null, 1));++ log(`Rollback file: ${BACKUP} (${manifest.rows.length} rows)`);++++ const done = ownerSubject.applyChanges(sqlite, plan.changes, {++ batch: BATCH,++ onBatch: (b) => log(` batch ${b.batch}: ${b.rows} rows (filled ${b.applied} so far, ${b.skipped} changed underneath)`),++ });++ // The file now lists exactly the rows this run changed.++ writePrivate(BACKUP, JSON.stringify({ ...manifest, status: 'applied', applied_at: new Date().toISOString(), batches: done.batches,++ skipped: done.skipped.map(c => c.id), rows: done.applied }, null, 1));++ const after = ownerSubject.summary(sqlite);++ log('');++ log(`Filled owner_subject on ${done.applied.length} object(s) in ${done.batches} batch(es); ${done.skipped.length} skipped (changed while running).`);++ log(`Now: ${Object.entries(after).map(([app, s]) => `${app} ${s.already_set}/${s.objects} set, ${s.missing} missing`).join('; ')}`);++ if (JSON_OUT) console.log(JSON.stringify({ ...report, applied: done.applied.length, skipped: done.skipped.length, batches: done.batches, after, rollback_file: BACKUP, db_backup: copy }, null, 2));++ return 0;++ } finally { sqlite.close(); }++}++++function rollback() {++ let manifest;++ try { manifest = JSON.parse(fs.readFileSync(ROLLBACK, 'utf8')); } catch (err) { usage(`cannot read ${ROLLBACK}: ${err.message}`); }++ if (!manifest || manifest.kind !== 'media.owner_subject.backfill' || !Array.isArray(manifest.rows)) usage(`${ROLLBACK} is not an owner_subject backfill file`);++ const sqlite = open({ readonly: !APPLY });++ try {++ const r = ownerSubject.rollbackChanges(sqlite, manifest.rows, { apply: APPLY, batch: BATCH });++ log(`Database: ${DB_PATH}`);++ log(`Rollback file: ${ROLLBACK} (${manifest.status}, ${manifest.rows.length} rows, written ${manifest.applied_at || manifest.created_at})`);++ log(`${APPLY ? 'Restored' : 'Would restore'} ${r.restored} (owner_subject back to NULL); ${r.changed_since} changed since (left alone); ${r.missing} no longer exist.`);++ if (!APPLY) log('Nothing was changed. Re-run with --apply to restore.');++ if (JSON_OUT) console.log(JSON.stringify({ mode: APPLY ? 'rollback' : 'rollback-dry-run', restored: r.restored, changed_since: r.changed_since, missing: r.missing, changed_since_ids: r.rows.changed_since, missing_ids: r.rows.missing }, null, 2));++ return 0;++ } finally { sqlite.close(); }++}++++(ROLLBACK ? Promise.resolve().then(rollback) : backfill())++ .then((code) => process.exit(code))++ .catch((err) => { console.error(`backfill-owner-subject: ${err.message}`); process.exit(1); });@@@ -88,6 +88,13 @@ const config = {repairCorrupt: ['1', 'true', 'on'].includes(String(process.env.MEDIA_VERIFY_REPAIR_CORRUPT || '').toLowerCase()),},++ // Owner subjects (server/objects/owner-subject-job.js): objects that name only an app-local owner get++ // the owner's Network subject (usr_…) on the next run.++ ownerSubject: {++ enabled: !['0', 'false', 'off'].includes(String(process.env.MEDIA_OWNER_SUBJECT_SYNC || '').toLowerCase()),++ intervalMin: intEnv('MEDIA_OWNER_SUBJECT_INTERVAL_MIN', 10),++ },++rtp: {portMin: intEnv('RTP_PORT_MIN', 12000),portMax: intEnv('RTP_PORT_MAX', 12199),@@@ -209,6 +209,7 @@ if (!drill.enabled) {require('./vod/clip-jobs').start(); // failed clip re-cuts (auto-retry, hot-fetch)require('./objects/verify-job').start(); // scheduled copy verification (bounded batches; never deletes)require('./jobs/worker').start(); // media_jobs worker (light + heavy lanes; proposals wait for their owner)++ require('./objects/owner-subject-job').start(); // owner_subject for objects that name only an app-local owner (asks Network)// Descriptor watchdog: a leak here once pinned 80 GB of deleted recordings to the disk.every(5 * 60 * 1000, () => {try {@@@ -273,6 +274,7 @@ function shutdown(signal) {try { healthJob.stop(); } catch { /* */ }try { require('./objects/verify-job').stop(); } catch { /* */ }try { require('./jobs/worker').stop(); } catch { /* */ }++ try { require('./objects/owner-subject-job').stop(); } catch { /* */ }try { auth.stopJwksRefresh(); } catch { /* */ }try { require('./events')._reset(); } catch { /* */ }for (const t of timers) clearInterval(t);@@@ -211,6 +211,11 @@ function project(p) {if (!fields.content_hash) fields.content_hash = found.content_hash;fields.metadata = { ...parseJson(found.metadata, {}), ...fields.metadata };if (found.lifecycle_status === 'deleted') fields.deleted_at = null; // the row exists, so the object does++ // owner_subject (filled by the owner-subject job) is the subject of owner_user_id; a different owner++ // drops it until the job resolves the new one.++ if (String(found.owner_app || found.app_id) !== String(fields.owner_app) || String(found.owner_user_id ?? '') !== String(fields.owner_user_id ?? '')) {++ fields.owner_subject = null;++ }updateObject(id, fields);} else {id = createObject({ ...fields, createdMs: toMs(p.created_at), created_at: p.created_at || null });nnew file mode 100644@@@ -0,0 +1,65 @@++'use strict';++/**++ * OpenVibe.Media — owner-subject reconcile job (docs/object-model.md#owner-subjects).++ *++ * Every MEDIA_OWNER_SUBJECT_INTERVAL_MIN minutes (first run a minute after boot), objects that name++ * an app-local owner (owner_app + owner_user_id) and no owner_subject get their owner's canonical++ * subject from Network (server/objects/owner-subject.js). This is how objects created since the++ * backfill get one: the upload and projection paths never wait on Network, and an object whose owner++ * Network does not know yet (a Live account not linked to Network) is filled on a later run once it is.++ * When there is nothing to fill, a run asks Network nothing.++ *++ * MEDIA_OWNER_SUBJECT_SYNC=0 turns it off (do that before rolling a backfill back, or the next run++ * fills the rows again).++ */++const db = require('../db/database');++const config = require('../config');++const ownerSubject = require('./owner-subject');++++let _timer = null, _startTimer = null, _busy = false, _resolver = null;++const stats = { runs: 0, filled_total: 0, last_run_at: null, last: null, last_error: null };++++async function runOnce({ resolver } = {}) {++ if (_busy) return null;++ _busy = true;++ try {++ if (!resolver) resolver = _resolver || (_resolver = ownerSubject.createResolver());++ const r = await ownerSubject.reconcileOnce(db.getDb(), { resolver });++ stats.runs++;++ stats.filled_total += r.filled;++ stats.last_run_at = new Date().toISOString();++ stats.last = r;++ stats.last_error = null;++ if (r.filled) console.log(`[OwnerSubject] filled owner_subject on ${r.filled} object(s) (${r.unresolvable} still unresolvable, via ${r.via})`);++ return r;++ } catch (err) {++ if (err.message !== stats.last_error) console.warn('[OwnerSubject] run failed (next run retries):', err.message);++ stats.last_error = err.message;++ return null;++ } finally {++ _busy = false;++ }++}++++function start({ initialDelayMs = 60 * 1000 } = {}) {++ if (!config.ownerSubject.enabled) { console.log('[OwnerSubject] reconcile job disabled (MEDIA_OWNER_SUBJECT_SYNC=0)'); return false; }++ if (_timer || _startTimer) return true;++ const tick = () => { runOnce().catch(() => {}); };++ _startTimer = setTimeout(() => {++ _startTimer = null;++ tick();++ _timer = setInterval(tick, Math.max(1, config.ownerSubject.intervalMin) * 60 * 1000);++ if (_timer.unref) _timer.unref();++ }, initialDelayMs);++ if (_startTimer.unref) _startTimer.unref();++ return true;++}++++function stop() {++ if (_startTimer) { clearTimeout(_startTimer); _startTimer = null; }++ if (_timer) { clearInterval(_timer); _timer = null; }++}++++function status() { return { ...stats, running: !!(_timer || _startTimer) }; }++++module.exports = { runOnce, start, stop, status };nnew file mode 100644@@@ -0,0 +1,273 @@++'use strict';++/**++ * OpenVibe.Media — canonical object owners (roadmap D01/D20; docs/object-model.md#owner-subjects).++ *++ * Every media_object should name its owner as a Network subject (owner_subject, usr_<ULID>), so no++ * consumer (Live's lineage resolver, Search, Community) has to translate an app's own user ids.++ * An object created through the v2 API with X-OV-Subject has one from the start. Everything else++ * carries only owner_app + owner_user_id, an id in that app's own user space: the projected vods,++ * clips, files, screenshots, avatars and thumbnails, and v2 uploads that name only user_id.++ *++ * This module fills owner_subject from Network's identity_legacy_map through++ * POST /internal/identity/resolve-batch { system, type: 'user', ids } (at most 500 ids per call).++ * It backs scripts/backfill-owner-subject.js (the one-off backfill, which writes a rollback file)++ * and the reconcile job the service runs every MEDIA_OWNER_SUBJECT_INTERVAL_MIN minutes, which++ * gives objects created since then their subject.++ *++ * Rules:++ * - a non-null owner_subject is never overwritten. Every UPDATE is guarded by owner_subject IS NULL++ * and by the owner it was resolved for (owner_app, owner_user_id), so a concurrent change wins;++ * - an id Network does not know stays NULL and is reported; nothing is guessed;++ * - only the tenants in SOURCE_SYSTEMS are resolved, because the system names the user-id space.++ *++ * Network is asked with a service token (capability identity.subject.resolve, audience++ * openvibe.network; OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET). When no token can be had++ * (a Network that has not granted it yet), it falls back to INTERNAL_API_KEY.++ */++const { ids, serviceAuth } = require('openvibe-contracts');++const config = require('../config');++++// Tenant -> the Network source system its owner_user_id values belong to (identity_legacy_map.source_system).++// Only Live stores app-local user ids here: Tools and developer-project uploads name their owner in++// X-OV-Subject, and Community's files carry no owner. A tenant missing here is reported, never guessed.++const SOURCE_SYSTEMS = Object.freeze({ live: 'live' });++const MAX_IDS_PER_CALL = 500;++const DEFAULT_BATCH = 500;++const TOKEN_RETRY_MS = 10 * 60 * 1000;++++const isUserSubject = (s) => typeof s === 'string' && ids.isSubjectId('user', s);++const ownerKey = (app, userId) => `${app}\u0000${userId}`;++++/**++ * Network resolver. resolve(system, ids) -> Map<String(id), 'usr_…' | null> (null = Network does not++ * know that id). Throws when Network cannot be asked or answers with an error, so a caller can tell++ * "unresolvable" from "unavailable".++ */++function createResolver({++ networkUrl = config.network.internalUrl,++ internalKey = config.network.internalApiKey,++ clientId = process.env.OV_OAUTH_CLIENT_ID || 'media',++ clientSecret = process.env.OV_OAUTH_CLIENT_SECRET || '',++ fetchImpl = globalThis.fetch,++ timeoutMs = 10000,++} = {}) {++ const base = String(networkUrl || '').replace(/\/+$/, '');++ const tokens = clientSecret ? serviceAuth.createTokenClient({++ tokenUrl: `${base}/oauth/token`, clientId, clientSecret, audience: 'openvibe.network',++ scope: 'identity.subject.resolve', fetchImpl, timeoutMs,++ }) : null;++ let tokenRetryAt = 0;++ const state = { via: null, token_error: null, calls: 0 };++++ async function credential() {++ if (tokens && Date.now() >= tokenRetryAt) {++ try { return { via: 'service-token', headers: await tokens.authHeaders() }; } catch (err) {++ state.token_error = err.message;++ tokenRetryAt = Date.now() + TOKEN_RETRY_MS;++ }++ }++ if (internalKey) return { via: 'internal-key', headers: { 'X-Internal-Key': internalKey } };++ throw new Error(tokens++ ? `no service token for identity.subject.resolve (${state.token_error}) and no INTERNAL_API_KEY`++ : 'set OV_OAUTH_CLIENT_SECRET (service token) or INTERNAL_API_KEY to ask Network');++ }++++ async function send(cred, payload) {++ state.calls++;++ let res;++ try {++ res = await fetchImpl(`${base}/internal/identity/resolve-batch`, {++ method: 'POST',++ headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...cred.headers },++ body: JSON.stringify(payload),++ signal: AbortSignal.timeout(timeoutMs),++ });++ } catch (err) {++ throw new Error(`Network unreachable at ${base}: ${err.message}`);++ }++ return res;++ }++++ async function post(payload) {++ let cred = await credential();++ let res = await send(cred, payload);++ if ((res.status === 401 || res.status === 403) && cred.via === 'service-token' && internalKey) {++ // A token Network refuses here (grant withdrawn, key rotated): the key, and no token for a while.++ tokens.invalidate();++ tokenRetryAt = Date.now() + TOKEN_RETRY_MS;++ state.token_error = `resolve-batch refused the token (${res.status})`;++ cred = { via: 'internal-key', headers: { 'X-Internal-Key': internalKey } };++ res = await send(cred, payload);++ }++ const body = await res.json().catch(() => null);++ if (!res.ok || !body || typeof body.results !== 'object' || body.results === null) {++ throw new Error(`Network resolve-batch ${res.status}: ${(body && (body.detail || body.error || body.title)) || 'unexpected answer'}`);++ }++ state.via = cred.via;++ return body.results;++ }++++ async function resolve(system, idList) {++ const out = new Map();++ const list = [...new Set(idList.map(String))];++ for (let i = 0; i < list.length; i += MAX_IDS_PER_CALL) {++ const chunk = list.slice(i, i + MAX_IDS_PER_CALL);++ const results = await post({ system, type: 'user', ids: chunk });++ for (const id of chunk) {++ const p = results[id];++ const sid = p && p.subject && p.subject.type === 'user' ? p.subject.id : null;++ out.set(id, isUserSubject(sid) ? sid : null);++ }++ }++ return out;++ }++++ return { resolve, state };++}++++// ── Reads ────────────────────────────────────────────────────++++/** Per-tenant ownership counts (read-only). */++function summary(sqlite) {++ const rows = sqlite.prepare(`++ SELECT app_id, count(*) AS objects,++ sum(owner_subject IS NOT NULL) AS already_set,++ sum(owner_subject IS NULL AND owner_user_id IS NOT NULL) AS missing,++ sum(owner_subject IS NULL AND owner_user_id IS NULL) AS no_owner++ FROM media_objects GROUP BY app_id ORDER BY app_id`).all();++ const out = {};++ for (const r of rows) out[r.app_id] = { objects: r.objects, already_set: r.already_set || 0, missing: r.missing || 0, no_owner: r.no_owner || 0 };++ return out;++}++++/** Distinct (owner_app, owner_user_id) pairs that still lack a subject, with their object counts. */++function pendingOwners(sqlite) {++ return sqlite.prepare(`++ SELECT COALESCE(owner_app, app_id) AS owner_app, owner_user_id, count(*) AS objects++ FROM media_objects WHERE owner_subject IS NULL AND owner_user_id IS NOT NULL++ GROUP BY COALESCE(owner_app, app_id), owner_user_id ORDER BY 1, 2`).all();++}++++/**++ * Resolve owners through Network, grouped by source system.++ * -> { subjects: Map<ownerKey, 'usr_…'>, unresolvable: [owner], unsupported: [owner] }++ */++async function resolveOwners(owners, { resolver }) {++ const subjects = new Map();++ const unresolvable = [], unsupported = [];++ const bySystem = new Map();++ for (const o of owners) {++ const system = Object.prototype.hasOwnProperty.call(SOURCE_SYSTEMS, o.owner_app) ? SOURCE_SYSTEMS[o.owner_app] : null;++ if (!system) { unsupported.push(o); continue; }++ if (!bySystem.has(system)) bySystem.set(system, []);++ bySystem.get(system).push(o);++ }++ for (const [system, list] of bySystem) {++ const got = await resolver.resolve(system, list.map(o => String(o.owner_user_id)));++ for (const o of list) {++ const sid = got.get(String(o.owner_user_id));++ if (sid) subjects.set(ownerKey(o.owner_app, o.owner_user_id), sid);++ else unresolvable.push(o);++ }++ }++ return { subjects, unresolvable, unsupported };++}++++/**++ * The rows the resolved subjects fill, read in id order `batch` at a time.++ * -> [{ id, app_id, owner_app, owner_user_id, owner_subject }]++ */++function planChanges(sqlite, subjects, { batch = DEFAULT_BATCH } = {}) {++ const page = sqlite.prepare(`++ SELECT id, app_id, COALESCE(owner_app, app_id) AS owner_app, owner_user_id FROM media_objects++ WHERE owner_subject IS NULL AND owner_user_id IS NOT NULL AND id > ? ORDER BY id LIMIT ?`);++ const changes = [];++ let after = '';++ for (;;) {++ const rows = page.all(after, batch);++ if (!rows.length) break;++ for (const r of rows) {++ const sid = subjects.get(ownerKey(r.owner_app, r.owner_user_id));++ if (sid) changes.push({ id: r.id, app_id: r.app_id, owner_app: r.owner_app, owner_user_id: r.owner_user_id, owner_subject: sid });++ }++ after = rows[rows.length - 1].id;++ }++ return changes;++}++++// ── Writes ───────────────────────────────────────────────────++++/**++ * Fill owner_subject on the planned rows, one transaction per `batch` rows. A row is written only if++ * it still has no subject and the same owner. -> { applied: [change], skipped: [change], batches }++ */++function applyChanges(sqlite, changes, { batch = DEFAULT_BATCH, onBatch = null } = {}) {++ const set = sqlite.prepare(`++ UPDATE media_objects SET owner_subject = ?, updated_at = CURRENT_TIMESTAMP++ WHERE id = ? AND owner_subject IS NULL AND COALESCE(owner_app, app_id) = ? AND owner_user_id = ?`);++ const applied = [], skipped = [];++ let batches = 0;++ for (let i = 0; i < changes.length; i += batch) {++ const slice = changes.slice(i, i + batch);++ sqlite.transaction(() => {++ for (const c of slice) (set.run(c.owner_subject, c.id, c.owner_app, c.owner_user_id).changes ? applied : skipped).push(c);++ })();++ batches++;++ if (onBatch) onBatch({ batch: batches, rows: slice.length, applied: applied.length, skipped: skipped.length });++ }++ return { applied, skipped, batches };++}++++/**++ * Undo a backfill: clear owner_subject on each listed row that still carries exactly the subject the++ * backfill wrote for the same owner. A row changed since (another subject, another owner) is left++ * alone and counted. With apply=false nothing is written.++ * -> { restored, changed_since, missing, rows: { restored: [id], changed_since: [id], missing: [id] } }++ */++function rollbackChanges(sqlite, entries, { apply = false, batch = DEFAULT_BATCH } = {}) {++ const get = sqlite.prepare('SELECT id, owner_subject, COALESCE(owner_app, app_id) AS owner_app, owner_user_id FROM media_objects WHERE id = ?');++ const clear = sqlite.prepare(`++ UPDATE media_objects SET owner_subject = NULL, updated_at = CURRENT_TIMESTAMP++ WHERE id = ? AND owner_subject = ? AND COALESCE(owner_app, app_id) = ? AND owner_user_id = ?`);++ const rows = { restored: [], changed_since: [], missing: [] };++ const unchanged = (row, e) => row.owner_subject === e.owner_subject && row.owner_app === e.owner_app && Number(row.owner_user_id) === Number(e.owner_user_id);++ for (let i = 0; i < entries.length; i += batch) {++ const slice = entries.slice(i, i + batch);++ sqlite.transaction(() => {++ for (const e of slice) {++ const row = get.get(e.id);++ if (!row) { rows.missing.push(e.id); continue; }++ if (!unchanged(row, e)) { rows.changed_since.push(e.id); continue; }++ if (apply && !clear.run(e.id, e.owner_subject, e.owner_app, e.owner_user_id).changes) { rows.changed_since.push(e.id); continue; }++ rows.restored.push(e.id);++ }++ })();++ }++ return { restored: rows.restored.length, changed_since: rows.changed_since.length, missing: rows.missing.length, rows };++}++++/**++ * One reconcile pass (the service's job): resolve every owner still lacking a subject and fill their++ * objects. -> { owners, filled, skipped, unresolvable, unsupported, via }++ */++async function reconcileOnce(sqlite, { resolver, batch = DEFAULT_BATCH } = {}) {++ const owners = pendingOwners(sqlite);++ const out = { owners: owners.length, filled: 0, skipped: 0, unresolvable: 0, unsupported: 0, via: null };++ if (!owners.length) return out;++ const callsBefore = resolver.state ? resolver.state.calls : 0;++ const r = await resolveOwners(owners, { resolver });++ out.unresolvable = r.unresolvable.reduce((n, o) => n + o.objects, 0);++ out.unsupported = r.unsupported.reduce((n, o) => n + o.objects, 0);++ out.via = resolver.state && resolver.state.calls > callsBefore ? resolver.state.via : null;++ if (!r.subjects.size) return out;++ const applied = applyChanges(sqlite, planChanges(sqlite, r.subjects, { batch }), { batch });++ out.filled = applied.applied.length;++ out.skipped = applied.skipped.length;++ return out;++}++++module.exports = {++ SOURCE_SYSTEMS, MAX_IDS_PER_CALL, DEFAULT_BATCH,++ createResolver, summary, pendingOwners, resolveOwners, planChanges, applyChanges, rollbackChanges, reconcileOnce,++ isUserSubject, ownerKey,++};nnew file mode 100644@@@ -0,0 +1,294 @@++'use strict';++// Owner subjects (server/objects/owner-subject.js, scripts/backfill-owner-subject.js,++// server/objects/owner-subject-job.js) against an in-process Network stand-in that answers++// /oauth/token and /internal/identity/resolve-batch like Network does: the dry run changes nothing;++// --apply takes a verified database backup and writes a rollback file first, then fills only NULLs in++// batches; an id Network does not know stays NULL; tenants with no source system are left alone;++// --rollback restores only the rows still as the backfill left them; the service token is preferred++// and the internal key is the fallback; and new objects get their subject (X-OV-Subject at creation,++// the reconcile job for everything else), with a re-projection to another owner dropping it.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const { execFile } = require('child_process');++const { ids } = require('openvibe-contracts');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-owner-subject-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++const env = {++ DB_PATH: path.join(tmp, 'media.db'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),++ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'), ASSETS_PATH: dir('assets'),++ MEDIA_PUBLIC_URL: 'https://media.test', OV_NETWORK_URL: 'http://127.0.0.1:9', MEDIA_SIGNING_SECRET: 'test-signing-secret',++ INTERNAL_API_KEY: 'internal-key-0123456789abcdef', OV_OAUTH_CLIENT_ID: 'media', OV_OAUTH_CLIENT_SECRET: 'media-secret',++};++for (const k of ['MEDIA_B2_ENDPOINT', 'MEDIA_B2_BUCKET', 'MEDIA_R2_ENDPOINT', 'MEDIA_R2_BUCKET', 'EVENTS_URL', 'MEDIA_OWNER_SUBJECT_SYNC']) process.env[k] = '';++Object.assign(process.env, env);++++// ── Network stand-in ─────────────────────────────────────────++const net = {++ grant: true, // does 'media' hold identity.subject.resolve?++ map: new Map(), // 'live:<id>' -> usr_…++ calls: [], // { via, system, n }++ tokens: 0,++ down: false,++};++const network = http.createServer((req, res) => {++ let raw = '';++ req.on('data', (c) => { raw += c; });++ req.on('end', () => {++ const send = (status, body) => { res.writeHead(status, { 'content-type': 'application/json' }); res.end(JSON.stringify(body)); };++ if (net.down) return send(503, { error: 'down' });++ if (req.method === 'POST' && req.url === '/oauth/token') {++ const f = new URLSearchParams(raw);++ if (f.get('client_id') !== 'media' || f.get('client_secret') !== 'media-secret') return send(401, { error: 'invalid_client' });++ if (f.get('audience') !== 'openvibe.network' || f.get('scope') !== 'identity.subject.resolve') return send(400, { error: 'invalid_request' });++ if (!net.grant) return send(400, { error: 'invalid_scope', error_description: 'not granted: identity.subject.resolve' });++ net.tokens++;++ return send(200, { access_token: 'svc-media-token', token_type: 'Bearer', expires_in: 300, scope: 'identity.subject.resolve' });++ }++ if (req.method === 'POST' && req.url === '/internal/identity/resolve-batch') {++ const bearer = String(req.headers.authorization || '');++ const via = req.headers['x-internal-key'] === env.INTERNAL_API_KEY ? 'internal-key' : (bearer === 'Bearer svc-media-token' && net.grant ? 'service-token' : null);++ if (!via) return send(403, { error: 'Invalid or missing internal key' });++ const b = JSON.parse(raw || '{}');++ if (!Array.isArray(b.ids) || !b.ids.length || !b.system) return send(400, { code: 'identity.bad_request' });++ if (b.ids.length > 500) return send(413, { code: 'identity.too_many_entries' });++ net.calls.push({ via, system: b.system, type: b.type, n: b.ids.length });++ const results = {};++ for (const id of b.ids) {++ const sid = net.map.get(`${b.system}:${id}`);++ results[String(id)] = sid ? { subject: { type: 'user', id: sid }, network_user_id: 1, username: `u${id}`, legacy_ids: [] } : null;++ }++ return send(200, { results });++ }++ send(404, { error: 'Not found' });++ });++});++++const db = require('../server/db/database');++const model = require('../server/objects/model');++const ownerSubject = require('../server/objects/owner-subject');++const job = require('../server/objects/owner-subject-job');++const config = require('../server/config');++const Database = require('better-sqlite3');++++const SCRIPT = path.join(__dirname, '..', 'scripts', 'backfill-owner-subject.js');++const subjectOf = (uid) => { const s = ids.newId('user', 1700000000000 + uid); net.map.set(`live:${uid}`, s); return s; };++const obj = (o) => model.createObject({ app_id: 'live', owner_app: 'live', kind: 'file', lifecycle_status: 'ready', ...o });++const row = (id) => db.get('SELECT * FROM media_objects WHERE id = ?', [id]);++const snapshot = () => db.all('SELECT id, owner_subject, owner_app, owner_user_id, updated_at FROM media_objects ORDER BY id');++++function runScript(args, extraEnv = {}) {++ return new Promise((resolve) => {++ execFile(process.execPath, [SCRIPT, ...args], { env: { ...process.env, ...extraEnv }, cwd: tmp }, (err, stdout, stderr) => {++ resolve({ code: err ? err.code : 0, stdout, stderr });++ });++ });++}++++(async () => {++ await new Promise((r) => network.listen(0, '127.0.0.1', r));++ const NET = `http://127.0.0.1:${network.address().port}`;++ process.env.OV_NETWORK_INTERNAL_URL = NET;++ config.network.internalUrl = NET;++++ // ── Fixture ──++ const s12 = subjectOf(12), s13 = subjectOf(13);++ const preset = ids.newId('user');++ const o = {++ a: obj({ owner_user_id: 12 }),++ b: obj({ owner_user_id: 12, kind: 'thumbnail' }),++ c: obj({ owner_user_id: 13, kind: 'screenshot' }),++ presetDifferent: obj({ owner_user_id: 12, owner_subject: preset }), // never overwritten, even though Network says s12++ unknown: obj({ owner_user_id: 99 }), // Network has no subject for live user 99++ noOwner: obj({ app_id: 'community', owner_app: 'community' }),++ games: obj({ app_id: 'games', owner_app: 'games', owner_user_id: 12 }), // no source system for games ids++ };++ // 520 more owners: more than one resolve-batch call (500 ids at most per call).++ const many = [];++ for (let uid = 1000; uid < 1520; uid++) { subjectOf(uid); many.push(obj({ owner_user_id: uid, kind: 'clip' })); }++ const TO_FILL = 3 + many.length;++++ // ── Dry run: asks Network, changes nothing ──++ let before = snapshot();++ let r = await runScript(['--json']);++ assert.strictEqual(r.code, 0, r.stderr);++ let rep = JSON.parse(r.stdout);++ assert.strictEqual(rep.mode, 'dry-run');++ assert.strictEqual(rep.via, 'service-token', 'the service token is preferred');++ assert.deepStrictEqual(rep.per_app.live, { to_fill: TO_FILL, unresolvable: 1, unsupported: 0 });++ assert.deepStrictEqual(rep.per_app.games, { to_fill: 0, unresolvable: 0, unsupported: 1 });++ assert.deepStrictEqual([rep.before.live.already_set, rep.before.community.no_owner], [1, 1]);++ assert.deepStrictEqual([rep.totals.to_fill, rep.totals.unresolvable, rep.totals.unsupported, rep.totals.already_set], [TO_FILL, 1, 1, 1]);++ assert.deepStrictEqual(rep.unresolvable_owners, [{ owner_app: 'live', owner_user_id: 99, objects: 1 }]);++ assert.ok(net.calls.length >= 2 && net.calls.every(c => c.n <= 500 && c.system === 'live' && c.type === 'user'), JSON.stringify(net.calls));++ assert.deepStrictEqual(snapshot(), before, 'a dry run changes nothing');++ assert.deepStrictEqual(fs.readdirSync(tmp).filter(f => /json|backup/.test(f)), [], 'and writes no file');++ r = await runScript([]);++ assert.strictEqual(r.code, 0, r.stderr);++ assert.match(r.stdout, /live\s+\d+\s+1\s+523\s+1\s+0\s+0/, r.stdout);++ assert.match(r.stdout, /Unresolvable owners .*live:99 \(1\)/);++ assert.match(r.stdout, /Dry run: nothing was changed/);++ console.log('✅ dry run: per-tenant counts (to fill, already set, unresolvable, unsupported, no owner), nothing written');++++ // ── --apply needs --backup ──++ r = await runScript(['--apply']);++ assert.strictEqual(r.code, 2); assert.match(r.stderr, /--apply needs --backup/);++ assert.deepStrictEqual(snapshot(), before);++++ // ── Apply: backup first, then only NULLs, in batches ──++ const backup = path.join(tmp, 'backups', 'owner-subject-1.json');++ net.calls.length = 0;++ r = await runScript(['--apply', '--backup', backup, '--batch', '200', '--json']);++ assert.strictEqual(r.code, 0, r.stderr);++ const lines = r.stdout.slice(r.stdout.indexOf('{'));++ rep = JSON.parse(lines);++ assert.deepStrictEqual([rep.applied, rep.skipped, rep.batches], [TO_FILL, 0, Math.ceil(TO_FILL / 200)]);++ assert.strictEqual(rep.db_backup.integrity_check, 'ok');++ assert.strictEqual(rep.db_backup.objects, before.length, 'the copy was taken before the fill');++ const copy = new Database(rep.db_backup.path, { readonly: true });++ assert.strictEqual(copy.prepare('SELECT count(*) AS n FROM media_objects WHERE owner_subject IS NOT NULL').get().n, 1, 'the copy is the pre-fill state');++ copy.close();++ assert.strictEqual(fs.statSync(backup).mode & 0o777, 0o600);++ assert.strictEqual(fs.statSync(rep.db_backup.path).mode & 0o777, 0o600);++ assert.strictEqual(rep.db_backup.path, path.join(tmp, 'backups', 'owner-subject-1.media.db'));++ assert.deepStrictEqual([row(o.a).owner_subject, row(o.b).owner_subject, row(o.c).owner_subject], [s12, s12, s13]);++ assert.strictEqual(row(many[0]).owner_subject, net.map.get('live:1000'));++ assert.strictEqual(row(o.presetDifferent).owner_subject, preset, 'a non-null owner_subject is never overwritten');++ assert.strictEqual(row(o.unknown).owner_subject, null, 'an unresolvable id stays NULL');++ assert.strictEqual(row(o.games).owner_subject, null, 'a tenant with no source system is left alone');++ assert.strictEqual(row(o.noOwner).owner_subject, null);++ const manifest = JSON.parse(fs.readFileSync(backup, 'utf8'));++ assert.deepStrictEqual([manifest.kind, manifest.status, manifest.rows.length], ['media.owner_subject.backfill', 'applied', TO_FILL]);++ assert.deepStrictEqual(manifest.rows.find(x => x.id === o.a), { id: o.a, app_id: 'live', owner_app: 'live', owner_user_id: 12, owner_subject: s12 });++ assert.ok(!manifest.rows.some(x => [o.presetDifferent, o.unknown, o.games, o.noOwner].includes(x.id)), 'only the rows it changed');++ console.log('✅ --apply: verified online backup + rollback file first, then fills only NULLs in 200-row batches; unknown ids stay NULL');++++ // ── Idempotent: a second run has nothing to fill (and needs no new backup) ──++ r = await runScript(['--apply', '--backup', backup]);++ assert.strictEqual(r.code, 0, r.stderr);++ assert.match(r.stdout, /Nothing to fill/);++ // A fill with a backup name already used is refused before anything changes.++ const late = obj({ owner_user_id: 13 });++ r = await runScript(['--apply', '--backup', backup]);++ assert.strictEqual(r.code, 2); assert.match(r.stderr, /already exists/);++ assert.strictEqual(row(late).owner_subject, null);++++ // ── Rollback: only rows still as the backfill left them ──++ const other = ids.newId('user');++ db.run('UPDATE media_objects SET owner_subject = ? WHERE id = ?', [other, o.b]); // changed since the backfill++ const beforeRb = snapshot();++ r = await runScript(['--rollback', backup, '--json']);++ assert.strictEqual(r.code, 0, r.stderr);++ rep = JSON.parse(r.stdout.slice(r.stdout.indexOf('{')));++ assert.deepStrictEqual([rep.mode, rep.restored, rep.changed_since, rep.missing], ['rollback-dry-run', TO_FILL - 1, 1, 0]);++ assert.deepStrictEqual(rep.changed_since_ids, [o.b]);++ assert.deepStrictEqual(snapshot(), beforeRb, 'a rollback without --apply changes nothing');++ r = await runScript(['--rollback', backup, '--apply']);++ assert.strictEqual(r.code, 0, r.stderr);++ assert.match(r.stdout, new RegExp(`Restored ${TO_FILL - 1} .*; 1 changed since`));++ assert.deepStrictEqual([row(o.a).owner_subject, row(o.c).owner_subject, row(many[519]).owner_subject], [null, null, null]);++ assert.strictEqual(row(o.b).owner_subject, other, 'a row changed since is left alone');++ assert.strictEqual(row(o.presetDifferent).owner_subject, preset, 'rows the backfill did not change are untouched');++ console.log('✅ --rollback: dry by default, restores only the rows still carrying the backfilled subject');++++ // ── Fallback: no grant yet -> the internal key ──++ net.grant = false; net.calls.length = 0;++ r = await runScript(['--json']);++ assert.strictEqual(r.code, 0, r.stderr);++ rep = JSON.parse(r.stdout);++ assert.strictEqual(rep.via, 'internal-key');++ assert.ok(net.calls.length && net.calls.every(c => c.via === 'internal-key'));++ r = await runScript(['--json'], { INTERNAL_API_KEY: '' });++ assert.strictEqual(r.code, 1, 'neither credential works: an error, not "unresolvable"');++ assert.match(r.stderr, /no service token for identity\.subject\.resolve .* and no INTERNAL_API_KEY/);++ net.down = true;++ r = await runScript([]);++ assert.strictEqual(r.code, 1); assert.match(r.stderr, /Network resolve-batch 503|token endpoint 503/);++ net.down = false; net.grant = true;++ // A token Network refuses at resolve-batch: the key answers instead.++ const resolver = ownerSubject.createResolver({ networkUrl: NET });++ net.grant = true;++ let got = await resolver.resolve('live', ['12', '99', 12]);++ assert.deepStrictEqual([...got.entries()], [['12', s12], ['99', null]]);++ assert.strictEqual(resolver.state.via, 'service-token');++ const tokenOnly = ownerSubject.createResolver({ networkUrl: NET, internalKey: '' });++ net.grant = false;++ await assert.rejects(tokenOnly.resolve('live', ['12']), /no service token/);++ net.grant = true;++ console.log('✅ Network auth: service token first, INTERNAL_API_KEY when Network has not granted it; failures are errors');++++ // ── New objects: X-OV-Subject at creation ──++ const express = require('express');++ db.upsertApp({ app_id: 'live', api_key: 'live-key' });++ const app = express();++ app.use(express.json());++ app.use('/api/v2/:app/objects', require('../server/objects/routes'));++ const server = http.createServer(app);++ await new Promise((res) => server.listen(0, '127.0.0.1', res));++ const base = `http://127.0.0.1:${server.address().port}`;++ const post = (headers, body) => fetch(`${base}/api/v2/live/objects`, { method: 'POST', headers: { authorization: 'Bearer live-key', 'content-type': 'application/json', ...headers }, body: JSON.stringify(body) }).then(async x => ({ status: x.status, body: await x.json() }));++ let p = await post({ 'x-ov-subject': `user:${s13}` }, { kind: 'file', size_bytes: 3, user_id: 13 });++ assert.strictEqual(p.status, 201, JSON.stringify(p.body));++ assert.strictEqual(row(p.body.id).owner_subject, s13, 'X-OV-Subject is stored at creation');++ p = await post({}, { kind: 'file', size_bytes: 3, user_id: 12 });++ const userOnly = p.body.id;++ assert.deepStrictEqual([row(userOnly).owner_subject, row(userOnly).owner_user_id], [null, 12]);++++ // ── New objects: the reconcile job fills the rest (projections included) ──++ const clipId = db.createClip({ app_id: 'live', user_id: 13, channel_user_id: 12, title: 'c', file_path: path.join(env.CLIPS_PATH, 'c.webm'), start_time: 0, end_time: 5 }).lastInsertRowid;++ const clipObj = db.get('SELECT object_id FROM clips WHERE id = ?', [clipId]).object_id;++ assert.ok(clipObj && row(clipObj).owner_subject === null && row(clipObj).owner_user_id === 13);++ net.calls.length = 0;++ const jr = await job.runOnce();++ assert.ok(jr && jr.filled >= 3, JSON.stringify(jr));++ assert.strictEqual(jr.via, 'service-token');++ assert.deepStrictEqual([row(userOnly).owner_subject, row(clipObj).owner_subject, row(late).owner_subject], [s12, s13, s13]);++ assert.strictEqual(row(o.a).owner_subject, s12, 'the job also refills rows a rollback cleared (so it is turned off before one)');++ assert.strictEqual(row(o.unknown).owner_subject, null);++ assert.strictEqual(row(o.b).owner_subject, other, 'never over a non-null value');++ // Network learns live user 99 later: the next run fills it.++ const s99 = subjectOf(99);++ assert.strictEqual((await job.runOnce()).filled, 1);++ assert.strictEqual(row(o.unknown).owner_subject, s99);++ // Nothing left to fill: a run asks Network nothing (games 12 is the only pending owner and has no source system).++ net.calls.length = 0;++ assert.deepStrictEqual(await job.runOnce(), { owners: 1, filled: 0, skipped: 0, unresolvable: 0, unsupported: 1, via: null });++ assert.strictEqual(net.calls.length, 0);++ // Network down: the run fails quietly and the next one retries.++ net.down = true;++ const quiet = console.warn; console.warn = () => {};++ obj({ owner_user_id: 12 });++ assert.strictEqual(await job.runOnce(), null);++ console.warn = quiet;++ assert.match(job.status().last_error, /503/);++ net.down = false;++ assert.strictEqual((await job.runOnce()).filled, 1);++++ // A re-projection keeps the subject; a re-projection to another owner drops it until the job resolves the new one.++ model.sync('clip', clipId);++ assert.strictEqual(row(clipObj).owner_subject, s13);++ db.run('UPDATE clips SET user_id = 12 WHERE id = ?', [clipId]);++ model.sync('clip', clipId);++ assert.deepStrictEqual([row(clipObj).owner_user_id, row(clipObj).owner_subject], [12, null]);++ await job.runOnce();++ assert.strictEqual(row(clipObj).owner_subject, s12);++ // The v2 object's public shape carries it (what Live's lineage resolver reads: owner.subject).++ assert.strictEqual(model.objectPublic(row(clipObj)).owner.subject, s12);++++ // MEDIA_OWNER_SUBJECT_SYNC=0 keeps the job off.++ config.ownerSubject.enabled = false;++ const said = console.log; console.log = () => {};++ assert.strictEqual(job.start(), false);++ console.log = said;++ config.ownerSubject.enabled = true;++ console.log('✅ new objects: X-OV-Subject stored at creation; the reconcile job fills the rest, re-projection keeps or drops it');++++ server.close(); network.close();++ db.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('owner-subject: all checks passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -33,6 +33,12 @@ OV_NETWORK_URL=https://openvibe.network
@@ -33,6 +33,12 @@ OV_NETWORK_URL=https://openvibe.network
OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000
OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000
# Per-service key for Network's /internal/* endpoints (header X-Internal-Key).
# Per-service key for Network's /internal/* endpoints (header X-Internal-Key).
INTERNAL_API_KEY=
INTERNAL_API_KEY=
+# Owner subjects (docs/object-model.md#owner-subjects): objects that name only an app-local owner
+# get the owner's Network subject from POST /internal/identity/resolve-batch, with a service token
+# (OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET, capability identity.subject.resolve) or else the key.
+# MEDIA_OWNER_SUBJECT_SYNC=0 stops the reconcile job (before rolling a backfill back).
+MEDIA_OWNER_SUBJECT_SYNC=
+MEDIA_OWNER_SUBJECT_INTERVAL_MIN=10
# ── Tenant (app) seeding ─────────────────────────────────────
# ── Tenant (app) seeding ─────────────────────────────────────
# JSON array, upserted into the `apps` table on every boot (api_key is hashed,
# JSON array, upserted into the `apps` table on every boot (api_key is hashed,
@@ -114,6 +114,57 @@ object `deleted` whenever its row is deleted, which covers all of the inherited
@@ -114,6 +114,57 @@ object `deleted` whenever its row is deleted, which covers all of the inherited
never throw: the legacy write has already happened, so a failed sync only logs a warning. The next
never throw: the legacy write has already happened, so a failed sync only logs a warning. The next
sync or backfill repairs it.
sync or backfill repairs it.
+## Owner subjects
+
+`owner_subject` names an object's owner as a Network subject (`usr_<ULID>`), so consumers such as
+Live's lineage resolver (which reads `owner.subject` on the v2 object) never translate an app's own
+user ids. An object gets it in one of three ways:
+
+- **At creation,** when the caller names it: `X-OV-Subject` on `POST /api/v2/:app/objects`. Tools and
+ developer-project uploads do this. A derived object (split, remux) copies its source's.
+- **From the reconcile job** (`server/objects/owner-subject-job.js`), for every object that names only
+ `owner_app` + `owner_user_id`: the projected vods, clips, files, screenshots, avatars and thumbnails,
+ and v2 uploads that send only `user_id`. Every `MEDIA_OWNER_SUBJECT_INTERVAL_MIN` minutes (first run
+ one minute after boot) it collects the distinct owners that still lack a subject and asks Network's
+ `POST /internal/identity/resolve-batch` (`{ system, type: 'user', ids }`, 500 ids per call). It then
+ fills their objects. Upload and projection paths never wait on Network. When nothing is missing, a
+ run asks Network nothing. An owner Network does not know (a Live account not linked to Network) stays
+ `NULL` until a later run finds it.
+- **From the one-off backfill** `scripts/backfill-owner-subject.js`, for everything that existed before
+ the job.
+
+Rules for both job and backfill:
+
+- A non-null `owner_subject` is never overwritten. Every write is guarded by `owner_subject IS NULL`
+ and by the owner it was resolved for.
+- Only the tenants whose user-id space is known are resolved (`SOURCE_SYSTEMS` in
+ `server/objects/owner-subject.js`: `live` -> `live`). Others are reported and left alone.
+- A re-projection keeps the subject, unless it changes the owner (`owner_app`/`owner_user_id`). Then
+ the subject is dropped until the job resolves the new owner.
+- Network is asked with a service token (`OV_OAUTH_CLIENT_ID`/`OV_OAUTH_CLIENT_SECRET`, capability
+ `identity.subject.resolve`, audience `openvibe.network`). When Network has not granted it, the
+ request uses `INTERNAL_API_KEY`.
+
+```
+node scripts/backfill-owner-subject.js [--db media.db] [--batch 500] [--json] # dry run: counts per tenant
+node scripts/backfill-owner-subject.js --apply --backup <file.json> [--batch 500] # fill
+node scripts/backfill-owner-subject.js --rollback <file.json> [--apply] # undo (dry without --apply)
+```
+
+- **The dry run** (the default) opens the database read-only and prints per tenant: objects, already
+ set, to fill, unresolvable, unsupported tenant and no owner.
+- **`--apply --backup <file.json>`** first takes an online backup to `<file>.media.db` and requires
+ `PRAGMA integrity_check` = `ok` on it. It then writes `<file.json>` (0600), the rows it is about to
+ change, and fills them in `--batch` transactions. Finally it rewrites the file to list exactly the
+ rows it changed. It refuses an existing backup name, and a re-run fills only what is still missing.
+- **`--rollback <file.json> --apply`** sets `owner_subject` back to `NULL` on each listed row that
+ still carries the subject the backfill wrote for the same owner. Rows changed since are counted and
+ left alone. Without `--apply` it only reports, which also verifies an applied backfill. Stop the job
+ first (`MEDIA_OWNER_SUBJECT_SYNC=0`, restart), or its next run fills the rows again.
+
+The script opens the database directly, not through `server/db/database.js`, so none of the service's
+boot work runs (for example, marking in-progress clip cuts failed).
+
## Object API v2
## Object API v2
All routes live under `/api/v2/:app/objects`.
All routes live under `/api/v2/:app/objects`.
@@ -499,6 +550,8 @@ Not run against production yet: it needs the owner's go-ahead.
@@ -499,6 +550,8 @@ Not run against production yet: it needs the owner's go-ahead.
| `MEDIA_VERIFY_HASH_MAX_MB` | 64 | sha256 local copies up to this size (objects with a `content_hash`) |
| `MEDIA_VERIFY_HASH_MAX_MB` | 64 | sha256 local copies up to this size (objects with a `content_hash`) |
| `MEDIA_VERIFY_MAX_REUPLOADS` | 2 | missing remote copies restored per run |
| `MEDIA_VERIFY_MAX_REUPLOADS` | 2 | missing remote copies restored per run |
| `MEDIA_VERIFY_REPAIR_CORRUPT` | off | `1` also overwrites a corrupt remote copy from a good local copy |
| `MEDIA_VERIFY_REPAIR_CORRUPT` | off | `1` also overwrites a corrupt remote copy from a good local copy |
+| `MEDIA_OWNER_SUBJECT_SYNC` | on | `0` turns the owner-subject reconcile job off |
+| `MEDIA_OWNER_SUBJECT_INTERVAL_MIN` | 10 | minutes between owner-subject reconcile runs |
## Not in this pass
## Not in this pass
new file mode 100644
new file mode 100644
@@ -0,0 +1,212 @@
@@ -0,0 +1,212 @@
+#!/usr/bin/env node
+'use strict';
+/**
+ * Give every media_object that names an app-local owner (owner_app + owner_user_id) its owner's
+ * canonical Network subject (owner_subject, usr_<ULID>), resolved through Network's
+ * POST /internal/identity/resolve-batch (docs/object-model.md#owner-subjects). Once this has run,
+ * the service's reconcile job (server/objects/owner-subject-job.js) keeps new objects filled.
+ *
+ * node scripts/backfill-owner-subject.js [--db media.db] [--batch 500] [--json]
+ * Dry run (the default): asks Network, changes nothing, and prints per tenant the objects
+ * to fill, already set, unresolvable (Network does not know the owner), with no owner, and
+ * those of tenants whose user ids it has no source system for.
+ * node scripts/backfill-owner-subject.js --apply --backup <file.json> [--db media.db] [--batch 500]
+ * First takes an online backup of the database to <file>.media.db and checks it with
+ * PRAGMA integrity_check, then writes <file.json> (every row it is about to change, for
+ * --rollback). Only then fills owner_subject, <batch> rows per transaction, never over a
+ * non-null value and only while the row still has the owner it was resolved for.
+ * Refuses to overwrite either backup file. Re-running fills only what is still missing.
+ * node scripts/backfill-owner-subject.js --rollback <file.json> [--apply] [--db media.db]
+ * Clears owner_subject again on the rows the file lists that still carry exactly the subject
+ * the backfill wrote (rows changed since are left alone and counted). Without --apply it only
+ * reports what it would restore, which also verifies an applied backfill.
+ * Turn the service's reconcile job off first (MEDIA_OWNER_SUBJECT_SYNC=0 and a restart), or
+ * its next run fills the rows again.
+ *
+ * Network: OV_NETWORK_INTERNAL_URL (default http://127.0.0.1:4000), with a service token
+ * (OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET, capability identity.subject.resolve) or else
+ * INTERNAL_API_KEY: the names /etc/openvibe/media.env already uses. The database is DB_PATH.
+ *
+ * The database is opened directly (read-only for a dry run), not through server/db/database.js,
+ * whose boot work (migrations, marking in-progress clip cuts failed) must stay with the service.
+ * Exit codes: 0 done, 1 error, 2 bad usage.
+ */
+const fs = require('fs');
+const path = require('path');
+
+const args = process.argv.slice(2);
+const has = (name) => args.includes(`--${name}`);
+const arg = (name) => { const i = args.indexOf(`--${name}`); return i >= 0 && args[i + 1] && !args[i + 1].startsWith('--') ? args[i + 1] : null; };
+const usage = (msg) => { console.error(`backfill-owner-subject: ${msg}`); process.exit(2); };
+
+for (const flag of ['db', 'backup', 'rollback', 'batch']) if (has(flag) && !arg(flag)) usage(`--${flag} needs a value`);
+if (arg('db')) process.env.DB_PATH = path.resolve(arg('db'));
+
+const Database = require('better-sqlite3');
+const config = require('../server/config');
+const ownerSubject = require('../server/objects/owner-subject');
+
+const APPLY = has('apply');
+const JSON_OUT = has('json');
+const BATCH = arg('batch') ? Number(arg('batch')) : ownerSubject.DEFAULT_BATCH;
+if (!Number.isInteger(BATCH) || BATCH < 1 || BATCH > 5000) usage('--batch must be an integer 1-5000');
+const DB_PATH = config.db.path;
+const ROLLBACK = arg('rollback') ? path.resolve(arg('rollback')) : null;
+const BACKUP = arg('backup') ? path.resolve(arg('backup')) : null;
+if (ROLLBACK && BACKUP) usage('--rollback and --backup do not go together');
+if (APPLY && !ROLLBACK && !BACKUP) usage('--apply needs --backup <file.json> (the rollback file; the database copy goes next to it)');
+if (!fs.existsSync(DB_PATH)) usage(`no database at ${DB_PATH} (set DB_PATH or pass --db)`);
+
+const log = (...a) => { if (!JSON_OUT) console.log(...a); };
+
+function open({ readonly }) {
+ const sqlite = new Database(DB_PATH, { readonly, fileMustExist: true });
+ sqlite.pragma('busy_timeout = 15000');
+ return sqlite;
+}
+
+function writePrivate(file, text) {
+ fs.mkdirSync(path.dirname(file), { recursive: true, mode: 0o700 });
+ const tmp = `${file}.tmp-${process.pid}`;
+ const fd = fs.openSync(tmp, 'wx', 0o600);
+ try { fs.writeSync(fd, text); fs.fsyncSync(fd); } finally { fs.closeSync(fd); }
+ fs.renameSync(tmp, file);
+}
+
+function dbCopyPath(file) {
+ return file.replace(/\.json$/i, '') + '.media.db';
+}
+
+/** Online backup (better-sqlite3's page-by-page copy; the service keeps writing), then integrity_check. */
+async function backupDatabase(sqlite, dest) {
+ fs.mkdirSync(path.dirname(dest), { recursive: true, mode: 0o700 });
+ await sqlite.backup(dest);
+ fs.chmodSync(dest, 0o600);
+ const copy = new Database(dest, { readonly: true, fileMustExist: true });
+ try {
+ const check = copy.pragma('integrity_check', { simple: true });
+ const objects = copy.prepare('SELECT count(*) AS n FROM media_objects').get().n;
+ return { path: dest, bytes: fs.statSync(dest).size, integrity_check: check, objects };
+ } finally { copy.close(); }
+}
+
+function printSummary(before, plan) {
+ const apps = [...new Set([...Object.keys(before), ...Object.keys(plan.perApp)])].sort();
+ log(`Database: ${DB_PATH}`);
+ log(`Network: ${config.network.internalUrl} (asked via ${plan.via || 'nothing: no owner to resolve'})`);
+ log('');
+ log('app objects already set to fill unresolvable unsupported app no owner');
+ for (const app of apps) {
+ const b = before[app] || { objects: 0, already_set: 0, no_owner: 0 };
+ const p = plan.perApp[app] || { to_fill: 0, unresolvable: 0, unsupported: 0 };
+ log(`${app.padEnd(40)} ${String(b.objects).padStart(7)} ${String(b.already_set).padStart(11)} ${String(p.to_fill).padStart(7)} ${String(p.unresolvable).padStart(12)} ${String(p.unsupported).padStart(15)} ${String(b.no_owner).padStart(8)}`);
+ }
+ const t = plan.totals;
+ log(`${'total'.padEnd(40)} ${String(t.objects).padStart(7)} ${String(t.already_set).padStart(11)} ${String(t.to_fill).padStart(7)} ${String(t.unresolvable).padStart(12)} ${String(t.unsupported).padStart(15)} ${String(t.no_owner).padStart(8)}`);
+ if (plan.unresolvableOwners.length) {
+ log('');
+ log(`Unresolvable owners (Network has no subject for them; their objects stay NULL): ${plan.unresolvableOwners.map(o => `${o.owner_app}:${o.owner_user_id} (${o.objects})`).join(', ')}`);
+ }
+ if (plan.unsupportedOwners.length) {
+ log(`Owners in tenants with no known source system (left alone): ${plan.unsupportedOwners.map(o => `${o.owner_app}:${o.owner_user_id} (${o.objects})`).join(', ')}`);
+ }
+}
+
+async function makePlan(sqlite) {
+ const before = ownerSubject.summary(sqlite);
+ const owners = ownerSubject.pendingOwners(sqlite);
+ const resolver = ownerSubject.createResolver();
+ const r = owners.length ? await ownerSubject.resolveOwners(owners, { resolver }) : { subjects: new Map(), unresolvable: [], unsupported: [] };
+ const changes = ownerSubject.planChanges(sqlite, r.subjects, { batch: BATCH });
+ const perApp = {};
+ const bump = (app, k, n) => { perApp[app] = perApp[app] || { to_fill: 0, unresolvable: 0, unsupported: 0 }; perApp[app][k] += n; };
+ for (const c of changes) bump(c.app_id, 'to_fill', 1);
+ // Per tenant of the object (owner_app is the same as app_id for every object Media makes today).
+ const countBy = (list) => {
+ const m = new Map();
+ for (const o of list) {
+ for (const row of sqlite.prepare(`SELECT app_id, count(*) AS n FROM media_objects WHERE owner_subject IS NULL
+ AND COALESCE(owner_app, app_id) = ? AND owner_user_id = ? GROUP BY app_id`).all(o.owner_app, o.owner_user_id)) {
+ m.set(row.app_id, (m.get(row.app_id) || 0) + row.n);
+ }
+ }
+ return m;
+ };
+ for (const [app, n] of countBy(r.unresolvable)) bump(app, 'unresolvable', n);
+ for (const [app, n] of countBy(r.unsupported)) bump(app, 'unsupported', n);
+ const totals = { objects: 0, already_set: 0, no_owner: 0, to_fill: changes.length, unresolvable: 0, unsupported: 0 };
+ for (const b of Object.values(before)) { totals.objects += b.objects; totals.already_set += b.already_set; totals.no_owner += b.no_owner; }
+ for (const p of Object.values(perApp)) { totals.unresolvable += p.unresolvable; totals.unsupported += p.unsupported; }
+ return { before, perApp, totals, changes, unresolvableOwners: r.unresolvable, unsupportedOwners: r.unsupported, via: resolver.state.via, network_calls: resolver.state.calls, token_error: resolver.state.token_error };
+}
+
+async function backfill() {
+ const sqlite = open({ readonly: !APPLY });
+ try {
+ const plan = await makePlan(sqlite);
+ printSummary(plan.before, plan);
+ if (plan.token_error && plan.via === 'internal-key') log(`(no service token: ${plan.token_error}; used INTERNAL_API_KEY)`);
+ const report = { mode: APPLY ? 'apply' : 'dry-run', db: DB_PATH, network: config.network.internalUrl, via: plan.via, before: plan.before, per_app: plan.perApp, totals: plan.totals,
+ unresolvable_owners: plan.unresolvableOwners, unsupported_owners: plan.unsupportedOwners };
+ if (!APPLY) {
+ log('');
+ log('Dry run: nothing was changed. Re-run with --apply --backup <file.json> to fill.');
+ if (JSON_OUT) console.log(JSON.stringify(report, null, 2));
+ return 0;
+ }
+ if (!plan.changes.length) {
+ log('');
+ log('Nothing to fill; no backup taken, nothing changed.');
+ if (JSON_OUT) console.log(JSON.stringify({ ...report, applied: 0 }, null, 2));
+ return 0;
+ }
+
+ const copyPath = dbCopyPath(BACKUP);
+ for (const f of [BACKUP, copyPath]) if (fs.existsSync(f)) usage(`${f} already exists; pick a new --backup name`);
+ log('');
+ log(`Backing up ${DB_PATH} -> ${copyPath} ...`);
+ const copy = await backupDatabase(sqlite, copyPath);
+ if (copy.integrity_check !== 'ok') { console.error(`backup failed integrity_check: ${copy.integrity_check}; nothing changed`); return 1; }
+ log(` ${copy.bytes} bytes, integrity_check ok, ${copy.objects} objects`);
+ const manifest = {
+ kind: 'media.owner_subject.backfill', version: 1, created_at: new Date().toISOString(), db: DB_PATH, db_backup: copy.path,
+ status: 'planned', rows: plan.changes.map(c => ({ id: c.id, app_id: c.app_id, owner_app: c.owner_app, owner_user_id: c.owner_user_id, owner_subject: c.owner_subject })),
+ };
+ writePrivate(BACKUP, JSON.stringify(manifest, null, 1));
+ log(`Rollback file: ${BACKUP} (${manifest.rows.length} rows)`);
+
+ const done = ownerSubject.applyChanges(sqlite, plan.changes, {
+ batch: BATCH,
+ onBatch: (b) => log(` batch ${b.batch}: ${b.rows} rows (filled ${b.applied} so far, ${b.skipped} changed underneath)`),
+ });
+ // The file now lists exactly the rows this run changed.
+ writePrivate(BACKUP, JSON.stringify({ ...manifest, status: 'applied', applied_at: new Date().toISOString(), batches: done.batches,
+ skipped: done.skipped.map(c => c.id), rows: done.applied }, null, 1));
+ const after = ownerSubject.summary(sqlite);
+ log('');
+ log(`Filled owner_subject on ${done.applied.length} object(s) in ${done.batches} batch(es); ${done.skipped.length} skipped (changed while running).`);
+ log(`Now: ${Object.entries(after).map(([app, s]) => `${app} ${s.already_set}/${s.objects} set, ${s.missing} missing`).join('; ')}`);
+ if (JSON_OUT) console.log(JSON.stringify({ ...report, applied: done.applied.length, skipped: done.skipped.length, batches: done.batches, after, rollback_file: BACKUP, db_backup: copy }, null, 2));
+ return 0;
+ } finally { sqlite.close(); }
+}
+
+function rollback() {
+ let manifest;
+ try { manifest = JSON.parse(fs.readFileSync(ROLLBACK, 'utf8')); } catch (err) { usage(`cannot read ${ROLLBACK}: ${err.message}`); }
+ if (!manifest || manifest.kind !== 'media.owner_subject.backfill' || !Array.isArray(manifest.rows)) usage(`${ROLLBACK} is not an owner_subject backfill file`);
+ const sqlite = open({ readonly: !APPLY });
+ try {
+ const r = ownerSubject.rollbackChanges(sqlite, manifest.rows, { apply: APPLY, batch: BATCH });
+ log(`Database: ${DB_PATH}`);
+ log(`Rollback file: ${ROLLBACK} (${manifest.status}, ${manifest.rows.length} rows, written ${manifest.applied_at || manifest.created_at})`);
+ log(`${APPLY ? 'Restored' : 'Would restore'} ${r.restored} (owner_subject back to NULL); ${r.changed_since} changed since (left alone); ${r.missing} no longer exist.`);
+ if (!APPLY) log('Nothing was changed. Re-run with --apply to restore.');
+ if (JSON_OUT) console.log(JSON.stringify({ mode: APPLY ? 'rollback' : 'rollback-dry-run', restored: r.restored, changed_since: r.changed_since, missing: r.missing, changed_since_ids: r.rows.changed_since, missing_ids: r.rows.missing }, null, 2));
+ return 0;
+ } finally { sqlite.close(); }
+}
+
+(ROLLBACK ? Promise.resolve().then(rollback) : backfill())
+ .then((code) => process.exit(code))
+ .catch((err) => { console.error(`backfill-owner-subject: ${err.message}`); process.exit(1); });
@@ -88,6 +88,13 @@ const config = {
@@ -88,6 +88,13 @@ const config = {
repairCorrupt: ['1', 'true', 'on'].includes(String(process.env.MEDIA_VERIFY_REPAIR_CORRUPT || '').toLowerCase()),
repairCorrupt: ['1', 'true', 'on'].includes(String(process.env.MEDIA_VERIFY_REPAIR_CORRUPT || '').toLowerCase()),
},
},
+ // Owner subjects (server/objects/owner-subject-job.js): objects that name only an app-local owner get
+ // the owner's Network subject (usr_…) on the next run.
+ ownerSubject: {
+ enabled: !['0', 'false', 'off'].includes(String(process.env.MEDIA_OWNER_SUBJECT_SYNC || '').toLowerCase()),
+ intervalMin: intEnv('MEDIA_OWNER_SUBJECT_INTERVAL_MIN', 10),
+ },
+
rtp: {
rtp: {
portMin: intEnv('RTP_PORT_MIN', 12000),
portMin: intEnv('RTP_PORT_MIN', 12000),
portMax: intEnv('RTP_PORT_MAX', 12199),
portMax: intEnv('RTP_PORT_MAX', 12199),
@@ -209,6 +209,7 @@ if (!drill.enabled) {
@@ -209,6 +209,7 @@ if (!drill.enabled) {
require('./vod/clip-jobs').start(); // failed clip re-cuts (auto-retry, hot-fetch)
require('./vod/clip-jobs').start(); // failed clip re-cuts (auto-retry, hot-fetch)
require('./objects/verify-job').start(); // scheduled copy verification (bounded batches; never deletes)
require('./objects/verify-job').start(); // scheduled copy verification (bounded batches; never deletes)
require('./jobs/worker').start(); // media_jobs worker (light + heavy lanes; proposals wait for their owner)
require('./jobs/worker').start(); // media_jobs worker (light + heavy lanes; proposals wait for their owner)
+ require('./objects/owner-subject-job').start(); // owner_subject for objects that name only an app-local owner (asks Network)
// Descriptor watchdog: a leak here once pinned 80 GB of deleted recordings to the disk.
// Descriptor watchdog: a leak here once pinned 80 GB of deleted recordings to the disk.
every(5 * 60 * 1000, () => {
every(5 * 60 * 1000, () => {
try {
try {
@@ -273,6 +274,7 @@ function shutdown(signal) {
@@ -273,6 +274,7 @@ function shutdown(signal) {
try { healthJob.stop(); } catch { /* */ }
try { healthJob.stop(); } catch { /* */ }
try { require('./objects/verify-job').stop(); } catch { /* */ }
try { require('./objects/verify-job').stop(); } catch { /* */ }
try { require('./jobs/worker').stop(); } catch { /* */ }
try { require('./jobs/worker').stop(); } catch { /* */ }
+ try { require('./objects/owner-subject-job').stop(); } catch { /* */ }
try { auth.stopJwksRefresh(); } catch { /* */ }
try { auth.stopJwksRefresh(); } catch { /* */ }
try { require('./events')._reset(); } catch { /* */ }
try { require('./events')._reset(); } catch { /* */ }
for (const t of timers) clearInterval(t);
for (const t of timers) clearInterval(t);
@@ -211,6 +211,11 @@ function project(p) {
@@ -211,6 +211,11 @@ function project(p) {
if (!fields.content_hash) fields.content_hash = found.content_hash;
if (!fields.content_hash) fields.content_hash = found.content_hash;
fields.metadata = { ...parseJson(found.metadata, {}), ...fields.metadata };
fields.metadata = { ...parseJson(found.metadata, {}), ...fields.metadata };
if (found.lifecycle_status === 'deleted') fields.deleted_at = null; // the row exists, so the object does
if (found.lifecycle_status === 'deleted') fields.deleted_at = null; // the row exists, so the object does
+ // owner_subject (filled by the owner-subject job) is the subject of owner_user_id; a different owner
+ // drops it until the job resolves the new one.
+ if (String(found.owner_app || found.app_id) !== String(fields.owner_app) || String(found.owner_user_id ?? '') !== String(fields.owner_user_id ?? '')) {
+ fields.owner_subject = null;
+ }
updateObject(id, fields);
updateObject(id, fields);
} else {
} else {
id = createObject({ ...fields, createdMs: toMs(p.created_at), created_at: p.created_at || null });
id = createObject({ ...fields, createdMs: toMs(p.created_at), created_at: p.created_at || null });
new file mode 100644
new file mode 100644
@@ -0,0 +1,65 @@
@@ -0,0 +1,65 @@
+'use strict';
+/**
+ * OpenVibe.Media — owner-subject reconcile job (docs/object-model.md#owner-subjects).
+ *
+ * Every MEDIA_OWNER_SUBJECT_INTERVAL_MIN minutes (first run a minute after boot), objects that name
+ * an app-local owner (owner_app + owner_user_id) and no owner_subject get their owner's canonical
+ * subject from Network (server/objects/owner-subject.js). This is how objects created since the
+ * backfill get one: the upload and projection paths never wait on Network, and an object whose owner
+ * Network does not know yet (a Live account not linked to Network) is filled on a later run once it is.
+ * When there is nothing to fill, a run asks Network nothing.
+ *
+ * MEDIA_OWNER_SUBJECT_SYNC=0 turns it off (do that before rolling a backfill back, or the next run
+ * fills the rows again).
+ */
+const db = require('../db/database');
+const config = require('../config');
+const ownerSubject = require('./owner-subject');
+
+let _timer = null, _startTimer = null, _busy = false, _resolver = null;
+const stats = { runs: 0, filled_total: 0, last_run_at: null, last: null, last_error: null };
+
+async function runOnce({ resolver } = {}) {
+ if (_busy) return null;
+ _busy = true;
+ try {
+ if (!resolver) resolver = _resolver || (_resolver = ownerSubject.createResolver());
+ const r = await ownerSubject.reconcileOnce(db.getDb(), { resolver });
+ stats.runs++;
+ stats.filled_total += r.filled;
+ stats.last_run_at = new Date().toISOString();
+ stats.last = r;
+ stats.last_error = null;
+ if (r.filled) console.log(`[OwnerSubject] filled owner_subject on ${r.filled} object(s) (${r.unresolvable} still unresolvable, via ${r.via})`);
+ return r;
+ } catch (err) {
+ if (err.message !== stats.last_error) console.warn('[OwnerSubject] run failed (next run retries):', err.message);
+ stats.last_error = err.message;
+ return null;
+ } finally {
+ _busy = false;
+ }
+}
+
+function start({ initialDelayMs = 60 * 1000 } = {}) {
+ if (!config.ownerSubject.enabled) { console.log('[OwnerSubject] reconcile job disabled (MEDIA_OWNER_SUBJECT_SYNC=0)'); return false; }
+ if (_timer || _startTimer) return true;
+ const tick = () => { runOnce().catch(() => {}); };
+ _startTimer = setTimeout(() => {
+ _startTimer = null;
+ tick();
+ _timer = setInterval(tick, Math.max(1, config.ownerSubject.intervalMin) * 60 * 1000);
+ if (_timer.unref) _timer.unref();
+ }, initialDelayMs);
+ if (_startTimer.unref) _startTimer.unref();
+ return true;
+}
+
+function stop() {
+ if (_startTimer) { clearTimeout(_startTimer); _startTimer = null; }
+ if (_timer) { clearInterval(_timer); _timer = null; }
+}
+
+function status() { return { ...stats, running: !!(_timer || _startTimer) }; }
+
+module.exports = { runOnce, start, stop, status };
new file mode 100644
new file mode 100644
@@ -0,0 +1,273 @@
@@ -0,0 +1,273 @@
+'use strict';
+/**
+ * OpenVibe.Media — canonical object owners (roadmap D01/D20; docs/object-model.md#owner-subjects).
+ *
+ * Every media_object should name its owner as a Network subject (owner_subject, usr_<ULID>), so no
+ * consumer (Live's lineage resolver, Search, Community) has to translate an app's own user ids.
+ * An object created through the v2 API with X-OV-Subject has one from the start. Everything else
+ * carries only owner_app + owner_user_id, an id in that app's own user space: the projected vods,
+ * clips, files, screenshots, avatars and thumbnails, and v2 uploads that name only user_id.
+ *
+ * This module fills owner_subject from Network's identity_legacy_map through
+ * POST /internal/identity/resolve-batch { system, type: 'user', ids } (at most 500 ids per call).
+ * It backs scripts/backfill-owner-subject.js (the one-off backfill, which writes a rollback file)
+ * and the reconcile job the service runs every MEDIA_OWNER_SUBJECT_INTERVAL_MIN minutes, which
+ * gives objects created since then their subject.
+ *
+ * Rules:
+ * - a non-null owner_subject is never overwritten. Every UPDATE is guarded by owner_subject IS NULL
+ * and by the owner it was resolved for (owner_app, owner_user_id), so a concurrent change wins;
+ * - an id Network does not know stays NULL and is reported; nothing is guessed;
+ * - only the tenants in SOURCE_SYSTEMS are resolved, because the system names the user-id space.
+ *
+ * Network is asked with a service token (capability identity.subject.resolve, audience
+ * openvibe.network; OV_OAUTH_CLIENT_ID / OV_OAUTH_CLIENT_SECRET). When no token can be had
+ * (a Network that has not granted it yet), it falls back to INTERNAL_API_KEY.
+ */
+const { ids, serviceAuth } = require('openvibe-contracts');
+const config = require('../config');
+
+// Tenant -> the Network source system its owner_user_id values belong to (identity_legacy_map.source_system).
+// Only Live stores app-local user ids here: Tools and developer-project uploads name their owner in
+// X-OV-Subject, and Community's files carry no owner. A tenant missing here is reported, never guessed.
+const SOURCE_SYSTEMS = Object.freeze({ live: 'live' });
+const MAX_IDS_PER_CALL = 500;
+const DEFAULT_BATCH = 500;
+const TOKEN_RETRY_MS = 10 * 60 * 1000;
+
+const isUserSubject = (s) => typeof s === 'string' && ids.isSubjectId('user', s);
+const ownerKey = (app, userId) => `${app}\u0000${userId}`;
+
+/**
+ * Network resolver. resolve(system, ids) -> Map<String(id), 'usr_…' | null> (null = Network does not
+ * know that id). Throws when Network cannot be asked or answers with an error, so a caller can tell
+ * "unresolvable" from "unavailable".
+ */
+function createResolver({
+ networkUrl = config.network.internalUrl,
+ internalKey = config.network.internalApiKey,
+ clientId = process.env.OV_OAUTH_CLIENT_ID || 'media',
+ clientSecret = process.env.OV_OAUTH_CLIENT_SECRET || '',
+ fetchImpl = globalThis.fetch,
+ timeoutMs = 10000,
+} = {}) {
+ const base = String(networkUrl || '').replace(/\/+$/, '');
+ const tokens = clientSecret ? serviceAuth.createTokenClient({
+ tokenUrl: `${base}/oauth/token`, clientId, clientSecret, audience: 'openvibe.network',
+ scope: 'identity.subject.resolve', fetchImpl, timeoutMs,
+ }) : null;
+ let tokenRetryAt = 0;
+ const state = { via: null, token_error: null, calls: 0 };
+
+ async function credential() {
+ if (tokens && Date.now() >= tokenRetryAt) {
+ try { return { via: 'service-token', headers: await tokens.authHeaders() }; } catch (err) {
+ state.token_error = err.message;
+ tokenRetryAt = Date.now() + TOKEN_RETRY_MS;
+ }
+ }
+ if (internalKey) return { via: 'internal-key', headers: { 'X-Internal-Key': internalKey } };
+ throw new Error(tokens
+ ? `no service token for identity.subject.resolve (${state.token_error}) and no INTERNAL_API_KEY`
+ : 'set OV_OAUTH_CLIENT_SECRET (service token) or INTERNAL_API_KEY to ask Network');
+ }
+
+ async function send(cred, payload) {
+ state.calls++;
+ let res;
+ try {
+ res = await fetchImpl(`${base}/internal/identity/resolve-batch`, {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json', Accept: 'application/json', ...cred.headers },
+ body: JSON.stringify(payload),
+ signal: AbortSignal.timeout(timeoutMs),
+ });
+ } catch (err) {
+ throw new Error(`Network unreachable at ${base}: ${err.message}`);
+ }
+ return res;
+ }
+
+ async function post(payload) {
+ let cred = await credential();
+ let res = await send(cred, payload);
+ if ((res.status === 401 || res.status === 403) && cred.via === 'service-token' && internalKey) {
+ // A token Network refuses here (grant withdrawn, key rotated): the key, and no token for a while.
+ tokens.invalidate();
+ tokenRetryAt = Date.now() + TOKEN_RETRY_MS;
+ state.token_error = `resolve-batch refused the token (${res.status})`;
+ cred = { via: 'internal-key', headers: { 'X-Internal-Key': internalKey } };
+ res = await send(cred, payload);
+ }
+ const body = await res.json().catch(() => null);
+ if (!res.ok || !body || typeof body.results !== 'object' || body.results === null) {
+ throw new Error(`Network resolve-batch ${res.status}: ${(body && (body.detail || body.error || body.title)) || 'unexpected answer'}`);
+ }
+ state.via = cred.via;
+ return body.results;
+ }
+
+ async function resolve(system, idList) {
+ const out = new Map();
+ const list = [...new Set(idList.map(String))];
+ for (let i = 0; i < list.length; i += MAX_IDS_PER_CALL) {
+ const chunk = list.slice(i, i + MAX_IDS_PER_CALL);
+ const results = await post({ system, type: 'user', ids: chunk });
+ for (const id of chunk) {
+ const p = results[id];
+ const sid = p && p.subject && p.subject.type === 'user' ? p.subject.id : null;
+ out.set(id, isUserSubject(sid) ? sid : null);
+ }
+ }
+ return out;
+ }
+
+ return { resolve, state };
+}
+
+// ── Reads ────────────────────────────────────────────────────
+
+/** Per-tenant ownership counts (read-only). */
+function summary(sqlite) {
+ const rows = sqlite.prepare(`
+ SELECT app_id, count(*) AS objects,
+ sum(owner_subject IS NOT NULL) AS already_set,
+ sum(owner_subject IS NULL AND owner_user_id IS NOT NULL) AS missing,
+ sum(owner_subject IS NULL AND owner_user_id IS NULL) AS no_owner
+ FROM media_objects GROUP BY app_id ORDER BY app_id`).all();
+ const out = {};
+ for (const r of rows) out[r.app_id] = { objects: r.objects, already_set: r.already_set || 0, missing: r.missing || 0, no_owner: r.no_owner || 0 };
+ return out;
+}
+
+/** Distinct (owner_app, owner_user_id) pairs that still lack a subject, with their object counts. */
+function pendingOwners(sqlite) {
+ return sqlite.prepare(`
+ SELECT COALESCE(owner_app, app_id) AS owner_app, owner_user_id, count(*) AS objects
+ FROM media_objects WHERE owner_subject IS NULL AND owner_user_id IS NOT NULL
+ GROUP BY COALESCE(owner_app, app_id), owner_user_id ORDER BY 1, 2`).all();
+}
+
+/**
+ * Resolve owners through Network, grouped by source system.
+ * -> { subjects: Map<ownerKey, 'usr_…'>, unresolvable: [owner], unsupported: [owner] }
+ */
+async function resolveOwners(owners, { resolver }) {
+ const subjects = new Map();
+ const unresolvable = [], unsupported = [];
+ const bySystem = new Map();
+ for (const o of owners) {
+ const system = Object.prototype.hasOwnProperty.call(SOURCE_SYSTEMS, o.owner_app) ? SOURCE_SYSTEMS[o.owner_app] : null;
+ if (!system) { unsupported.push(o); continue; }
+ if (!bySystem.has(system)) bySystem.set(system, []);
+ bySystem.get(system).push(o);
+ }
+ for (const [system, list] of bySystem) {
+ const got = await resolver.resolve(system, list.map(o => String(o.owner_user_id)));
+ for (const o of list) {
+ const sid = got.get(String(o.owner_user_id));
+ if (sid) subjects.set(ownerKey(o.owner_app, o.owner_user_id), sid);
+ else unresolvable.push(o);
+ }
+ }
+ return { subjects, unresolvable, unsupported };
+}
+
+/**
+ * The rows the resolved subjects fill, read in id order `batch` at a time.
+ * -> [{ id, app_id, owner_app, owner_user_id, owner_subject }]
+ */
+function planChanges(sqlite, subjects, { batch = DEFAULT_BATCH } = {}) {
+ const page = sqlite.prepare(`
+ SELECT id, app_id, COALESCE(owner_app, app_id) AS owner_app, owner_user_id FROM media_objects
+ WHERE owner_subject IS NULL AND owner_user_id IS NOT NULL AND id > ? ORDER BY id LIMIT ?`);
+ const changes = [];
+ let after = '';
+ for (;;) {
+ const rows = page.all(after, batch);
+ if (!rows.length) break;
+ for (const r of rows) {
+ const sid = subjects.get(ownerKey(r.owner_app, r.owner_user_id));
+ if (sid) changes.push({ id: r.id, app_id: r.app_id, owner_app: r.owner_app, owner_user_id: r.owner_user_id, owner_subject: sid });
+ }
+ after = rows[rows.length - 1].id;
+ }
+ return changes;
+}
+
+// ── Writes ───────────────────────────────────────────────────
+
+/**
+ * Fill owner_subject on the planned rows, one transaction per `batch` rows. A row is written only if
+ * it still has no subject and the same owner. -> { applied: [change], skipped: [change], batches }
+ */
+function applyChanges(sqlite, changes, { batch = DEFAULT_BATCH, onBatch = null } = {}) {
+ const set = sqlite.prepare(`
+ UPDATE media_objects SET owner_subject = ?, updated_at = CURRENT_TIMESTAMP
+ WHERE id = ? AND owner_subject IS NULL AND COALESCE(owner_app, app_id) = ? AND owner_user_id = ?`);
+ const applied = [], skipped = [];
+ let batches = 0;
+ for (let i = 0; i < changes.length; i += batch) {
+ const slice = changes.slice(i, i + batch);
+ sqlite.transaction(() => {
+ for (const c of slice) (set.run(c.owner_subject, c.id, c.owner_app, c.owner_user_id).changes ? applied : skipped).push(c);
+ })();
+ batches++;
+ if (onBatch) onBatch({ batch: batches, rows: slice.length, applied: applied.length, skipped: skipped.length });
+ }
+ return { applied, skipped, batches };
+}
+
+/**
+ * Undo a backfill: clear owner_subject on each listed row that still carries exactly the subject the
+ * backfill wrote for the same owner. A row changed since (another subject, another owner) is left
+ * alone and counted. With apply=false nothing is written.
+ * -> { restored, changed_since, missing, rows: { restored: [id], changed_since: [id], missing: [id] } }
+ */
+function rollbackChanges(sqlite, entries, { apply = false, batch = DEFAULT_BATCH } = {}) {
+ const get = sqlite.prepare('SELECT id, owner_subject, COALESCE(owner_app, app_id) AS owner_app, owner_user_id FROM media_objects WHERE id = ?');
+ const clear = sqlite.prepare(`
+ UPDATE media_objects SET owner_subject = NULL, updated_at = CURRENT_TIMESTAMP
+ WHERE id = ? AND owner_subject = ? AND COALESCE(owner_app, app_id) = ? AND owner_user_id = ?`);
+ const rows = { restored: [], changed_since: [], missing: [] };
+ const unchanged = (row, e) => row.owner_subject === e.owner_subject && row.owner_app === e.owner_app && Number(row.owner_user_id) === Number(e.owner_user_id);
+ for (let i = 0; i < entries.length; i += batch) {
+ const slice = entries.slice(i, i + batch);
+ sqlite.transaction(() => {
+ for (const e of slice) {
+ const row = get.get(e.id);
+ if (!row) { rows.missing.push(e.id); continue; }
+ if (!unchanged(row, e)) { rows.changed_since.push(e.id); continue; }
+ if (apply && !clear.run(e.id, e.owner_subject, e.owner_app, e.owner_user_id).changes) { rows.changed_since.push(e.id); continue; }
+ rows.restored.push(e.id);
+ }
+ })();
+ }
+ return { restored: rows.restored.length, changed_since: rows.changed_since.length, missing: rows.missing.length, rows };
+}
+
+/**
+ * One reconcile pass (the service's job): resolve every owner still lacking a subject and fill their
+ * objects. -> { owners, filled, skipped, unresolvable, unsupported, via }
+ */
+async function reconcileOnce(sqlite, { resolver, batch = DEFAULT_BATCH } = {}) {
+ const owners = pendingOwners(sqlite);
+ const out = { owners: owners.length, filled: 0, skipped: 0, unresolvable: 0, unsupported: 0, via: null };
+ if (!owners.length) return out;
+ const callsBefore = resolver.state ? resolver.state.calls : 0;
+ const r = await resolveOwners(owners, { resolver });
+ out.unresolvable = r.unresolvable.reduce((n, o) => n + o.objects, 0);
+ out.unsupported = r.unsupported.reduce((n, o) => n + o.objects, 0);
+ out.via = resolver.state && resolver.state.calls > callsBefore ? resolver.state.via : null;
+ if (!r.subjects.size) return out;
+ const applied = applyChanges(sqlite, planChanges(sqlite, r.subjects, { batch }), { batch });
+ out.filled = applied.applied.length;
+ out.skipped = applied.skipped.length;
+ return out;
+}
+
+module.exports = {
+ SOURCE_SYSTEMS, MAX_IDS_PER_CALL, DEFAULT_BATCH,
+ createResolver, summary, pendingOwners, resolveOwners, planChanges, applyChanges, rollbackChanges, reconcileOnce,
+ isUserSubject, ownerKey,
+};
new file mode 100644
new file mode 100644
@@ -0,0 +1,294 @@
@@ -0,0 +1,294 @@
+'use strict';
+// Owner subjects (server/objects/owner-subject.js, scripts/backfill-owner-subject.js,
+// server/objects/owner-subject-job.js) against an in-process Network stand-in that answers
+// /oauth/token and /internal/identity/resolve-batch like Network does: the dry run changes nothing;
+// --apply takes a verified database backup and writes a rollback file first, then fills only NULLs in
+// batches; an id Network does not know stays NULL; tenants with no source system are left alone;
+// --rollback restores only the rows still as the backfill left them; the service token is preferred
+// and the internal key is the fallback; and new objects get their subject (X-OV-Subject at creation,
+// the reconcile job for everything else), with a re-projection to another owner dropping it.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const { execFile } = require('child_process');
+const { ids } = require('openvibe-contracts');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-owner-subject-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+const env = {
+ DB_PATH: path.join(tmp, 'media.db'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),
+ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'), ASSETS_PATH: dir('assets'),
+ MEDIA_PUBLIC_URL: 'https://media.test', OV_NETWORK_URL: 'http://127.0.0.1:9', MEDIA_SIGNING_SECRET: 'test-signing-secret',
+ INTERNAL_API_KEY: 'internal-key-0123456789abcdef', OV_OAUTH_CLIENT_ID: 'media', OV_OAUTH_CLIENT_SECRET: 'media-secret',
+};
+for (const k of ['MEDIA_B2_ENDPOINT', 'MEDIA_B2_BUCKET', 'MEDIA_R2_ENDPOINT', 'MEDIA_R2_BUCKET', 'EVENTS_URL', 'MEDIA_OWNER_SUBJECT_SYNC']) process.env[k] = '';
+Object.assign(process.env, env);
+
+// ── Network stand-in ─────────────────────────────────────────
+const net = {
+ grant: true, // does 'media' hold identity.subject.resolve?
+ map: new Map(), // 'live:<id>' -> usr_…
+ calls: [], // { via, system, n }
+ tokens: 0,
+ down: false,
+};
+const network = http.createServer((req, res) => {
+ let raw = '';
+ req.on('data', (c) => { raw += c; });
+ req.on('end', () => {
+ const send = (status, body) => { res.writeHead(status, { 'content-type': 'application/json' }); res.end(JSON.stringify(body)); };
+ if (net.down) return send(503, { error: 'down' });
+ if (req.method === 'POST' && req.url === '/oauth/token') {
+ const f = new URLSearchParams(raw);
+ if (f.get('client_id') !== 'media' || f.get('client_secret') !== 'media-secret') return send(401, { error: 'invalid_client' });
+ if (f.get('audience') !== 'openvibe.network' || f.get('scope') !== 'identity.subject.resolve') return send(400, { error: 'invalid_request' });
+ if (!net.grant) return send(400, { error: 'invalid_scope', error_description: 'not granted: identity.subject.resolve' });
+ net.tokens++;
+ return send(200, { access_token: 'svc-media-token', token_type: 'Bearer', expires_in: 300, scope: 'identity.subject.resolve' });
+ }
+ if (req.method === 'POST' && req.url === '/internal/identity/resolve-batch') {
+ const bearer = String(req.headers.authorization || '');
+ const via = req.headers['x-internal-key'] === env.INTERNAL_API_KEY ? 'internal-key' : (bearer === 'Bearer svc-media-token' && net.grant ? 'service-token' : null);
+ if (!via) return send(403, { error: 'Invalid or missing internal key' });
+ const b = JSON.parse(raw || '{}');
+ if (!Array.isArray(b.ids) || !b.ids.length || !b.system) return send(400, { code: 'identity.bad_request' });
+ if (b.ids.length > 500) return send(413, { code: 'identity.too_many_entries' });
+ net.calls.push({ via, system: b.system, type: b.type, n: b.ids.length });
+ const results = {};
+ for (const id of b.ids) {
+ const sid = net.map.get(`${b.system}:${id}`);
+ results[String(id)] = sid ? { subject: { type: 'user', id: sid }, network_user_id: 1, username: `u${id}`, legacy_ids: [] } : null;
+ }
+ return send(200, { results });
+ }
+ send(404, { error: 'Not found' });
+ });
+});
+
+const db = require('../server/db/database');
+const model = require('../server/objects/model');
+const ownerSubject = require('../server/objects/owner-subject');
+const job = require('../server/objects/owner-subject-job');
+const config = require('../server/config');
+const Database = require('better-sqlite3');
+
+const SCRIPT = path.join(__dirname, '..', 'scripts', 'backfill-owner-subject.js');
+const subjectOf = (uid) => { const s = ids.newId('user', 1700000000000 + uid); net.map.set(`live:${uid}`, s); return s; };
+const obj = (o) => model.createObject({ app_id: 'live', owner_app: 'live', kind: 'file', lifecycle_status: 'ready', ...o });
+const row = (id) => db.get('SELECT * FROM media_objects WHERE id = ?', [id]);
+const snapshot = () => db.all('SELECT id, owner_subject, owner_app, owner_user_id, updated_at FROM media_objects ORDER BY id');
+
+function runScript(args, extraEnv = {}) {
+ return new Promise((resolve) => {
+ execFile(process.execPath, [SCRIPT, ...args], { env: { ...process.env, ...extraEnv }, cwd: tmp }, (err, stdout, stderr) => {
+ resolve({ code: err ? err.code : 0, stdout, stderr });
+ });
+ });
+}
+
+(async () => {
+ await new Promise((r) => network.listen(0, '127.0.0.1', r));
+ const NET = `http://127.0.0.1:${network.address().port}`;
+ process.env.OV_NETWORK_INTERNAL_URL = NET;
+ config.network.internalUrl = NET;
+
+ // ── Fixture ──
+ const s12 = subjectOf(12), s13 = subjectOf(13);
+ const preset = ids.newId('user');
+ const o = {
+ a: obj({ owner_user_id: 12 }),
+ b: obj({ owner_user_id: 12, kind: 'thumbnail' }),
+ c: obj({ owner_user_id: 13, kind: 'screenshot' }),
+ presetDifferent: obj({ owner_user_id: 12, owner_subject: preset }), // never overwritten, even though Network says s12
+ unknown: obj({ owner_user_id: 99 }), // Network has no subject for live user 99
+ noOwner: obj({ app_id: 'community', owner_app: 'community' }),
+ games: obj({ app_id: 'games', owner_app: 'games', owner_user_id: 12 }), // no source system for games ids
+ };
+ // 520 more owners: more than one resolve-batch call (500 ids at most per call).
+ const many = [];
+ for (let uid = 1000; uid < 1520; uid++) { subjectOf(uid); many.push(obj({ owner_user_id: uid, kind: 'clip' })); }
+ const TO_FILL = 3 + many.length;
+
+ // ── Dry run: asks Network, changes nothing ──
+ let before = snapshot();
+ let r = await runScript(['--json']);
+ assert.strictEqual(r.code, 0, r.stderr);
+ let rep = JSON.parse(r.stdout);
+ assert.strictEqual(rep.mode, 'dry-run');
+ assert.strictEqual(rep.via, 'service-token', 'the service token is preferred');
+ assert.deepStrictEqual(rep.per_app.live, { to_fill: TO_FILL, unresolvable: 1, unsupported: 0 });
+ assert.deepStrictEqual(rep.per_app.games, { to_fill: 0, unresolvable: 0, unsupported: 1 });
+ assert.deepStrictEqual([rep.before.live.already_set, rep.before.community.no_owner], [1, 1]);
+ assert.deepStrictEqual([rep.totals.to_fill, rep.totals.unresolvable, rep.totals.unsupported, rep.totals.already_set], [TO_FILL, 1, 1, 1]);
+ assert.deepStrictEqual(rep.unresolvable_owners, [{ owner_app: 'live', owner_user_id: 99, objects: 1 }]);
+ assert.ok(net.calls.length >= 2 && net.calls.every(c => c.n <= 500 && c.system === 'live' && c.type === 'user'), JSON.stringify(net.calls));
+ assert.deepStrictEqual(snapshot(), before, 'a dry run changes nothing');
+ assert.deepStrictEqual(fs.readdirSync(tmp).filter(f => /json|backup/.test(f)), [], 'and writes no file');
+ r = await runScript([]);
+ assert.strictEqual(r.code, 0, r.stderr);
+ assert.match(r.stdout, /live\s+\d+\s+1\s+523\s+1\s+0\s+0/, r.stdout);
+ assert.match(r.stdout, /Unresolvable owners .*live:99 \(1\)/);
+ assert.match(r.stdout, /Dry run: nothing was changed/);
+ console.log('✅ dry run: per-tenant counts (to fill, already set, unresolvable, unsupported, no owner), nothing written');
+
+ // ── --apply needs --backup ──
+ r = await runScript(['--apply']);
+ assert.strictEqual(r.code, 2); assert.match(r.stderr, /--apply needs --backup/);
+ assert.deepStrictEqual(snapshot(), before);
+
+ // ── Apply: backup first, then only NULLs, in batches ──
+ const backup = path.join(tmp, 'backups', 'owner-subject-1.json');
+ net.calls.length = 0;
+ r = await runScript(['--apply', '--backup', backup, '--batch', '200', '--json']);
+ assert.strictEqual(r.code, 0, r.stderr);
+ const lines = r.stdout.slice(r.stdout.indexOf('{'));
+ rep = JSON.parse(lines);
+ assert.deepStrictEqual([rep.applied, rep.skipped, rep.batches], [TO_FILL, 0, Math.ceil(TO_FILL / 200)]);
+ assert.strictEqual(rep.db_backup.integrity_check, 'ok');
+ assert.strictEqual(rep.db_backup.objects, before.length, 'the copy was taken before the fill');
+ const copy = new Database(rep.db_backup.path, { readonly: true });
+ assert.strictEqual(copy.prepare('SELECT count(*) AS n FROM media_objects WHERE owner_subject IS NOT NULL').get().n, 1, 'the copy is the pre-fill state');
+ copy.close();
+ assert.strictEqual(fs.statSync(backup).mode & 0o777, 0o600);
+ assert.strictEqual(fs.statSync(rep.db_backup.path).mode & 0o777, 0o600);
+ assert.strictEqual(rep.db_backup.path, path.join(tmp, 'backups', 'owner-subject-1.media.db'));
+ assert.deepStrictEqual([row(o.a).owner_subject, row(o.b).owner_subject, row(o.c).owner_subject], [s12, s12, s13]);
+ assert.strictEqual(row(many[0]).owner_subject, net.map.get('live:1000'));
+ assert.strictEqual(row(o.presetDifferent).owner_subject, preset, 'a non-null owner_subject is never overwritten');
+ assert.strictEqual(row(o.unknown).owner_subject, null, 'an unresolvable id stays NULL');
+ assert.strictEqual(row(o.games).owner_subject, null, 'a tenant with no source system is left alone');
+ assert.strictEqual(row(o.noOwner).owner_subject, null);
+ const manifest = JSON.parse(fs.readFileSync(backup, 'utf8'));
+ assert.deepStrictEqual([manifest.kind, manifest.status, manifest.rows.length], ['media.owner_subject.backfill', 'applied', TO_FILL]);
+ assert.deepStrictEqual(manifest.rows.find(x => x.id === o.a), { id: o.a, app_id: 'live', owner_app: 'live', owner_user_id: 12, owner_subject: s12 });
+ assert.ok(!manifest.rows.some(x => [o.presetDifferent, o.unknown, o.games, o.noOwner].includes(x.id)), 'only the rows it changed');
+ console.log('✅ --apply: verified online backup + rollback file first, then fills only NULLs in 200-row batches; unknown ids stay NULL');
+
+ // ── Idempotent: a second run has nothing to fill (and needs no new backup) ──
+ r = await runScript(['--apply', '--backup', backup]);
+ assert.strictEqual(r.code, 0, r.stderr);
+ assert.match(r.stdout, /Nothing to fill/);
+ // A fill with a backup name already used is refused before anything changes.
+ const late = obj({ owner_user_id: 13 });
+ r = await runScript(['--apply', '--backup', backup]);
+ assert.strictEqual(r.code, 2); assert.match(r.stderr, /already exists/);
+ assert.strictEqual(row(late).owner_subject, null);
+
+ // ── Rollback: only rows still as the backfill left them ──
+ const other = ids.newId('user');
+ db.run('UPDATE media_objects SET owner_subject = ? WHERE id = ?', [other, o.b]); // changed since the backfill
+ const beforeRb = snapshot();
+ r = await runScript(['--rollback', backup, '--json']);
+ assert.strictEqual(r.code, 0, r.stderr);
+ rep = JSON.parse(r.stdout.slice(r.stdout.indexOf('{')));
+ assert.deepStrictEqual([rep.mode, rep.restored, rep.changed_since, rep.missing], ['rollback-dry-run', TO_FILL - 1, 1, 0]);
+ assert.deepStrictEqual(rep.changed_since_ids, [o.b]);
+ assert.deepStrictEqual(snapshot(), beforeRb, 'a rollback without --apply changes nothing');
+ r = await runScript(['--rollback', backup, '--apply']);
+ assert.strictEqual(r.code, 0, r.stderr);
+ assert.match(r.stdout, new RegExp(`Restored ${TO_FILL - 1} .*; 1 changed since`));
+ assert.deepStrictEqual([row(o.a).owner_subject, row(o.c).owner_subject, row(many[519]).owner_subject], [null, null, null]);
+ assert.strictEqual(row(o.b).owner_subject, other, 'a row changed since is left alone');
+ assert.strictEqual(row(o.presetDifferent).owner_subject, preset, 'rows the backfill did not change are untouched');
+ console.log('✅ --rollback: dry by default, restores only the rows still carrying the backfilled subject');
+
+ // ── Fallback: no grant yet -> the internal key ──
+ net.grant = false; net.calls.length = 0;
+ r = await runScript(['--json']);
+ assert.strictEqual(r.code, 0, r.stderr);
+ rep = JSON.parse(r.stdout);
+ assert.strictEqual(rep.via, 'internal-key');
+ assert.ok(net.calls.length && net.calls.every(c => c.via === 'internal-key'));
+ r = await runScript(['--json'], { INTERNAL_API_KEY: '' });
+ assert.strictEqual(r.code, 1, 'neither credential works: an error, not "unresolvable"');
+ assert.match(r.stderr, /no service token for identity\.subject\.resolve .* and no INTERNAL_API_KEY/);
+ net.down = true;
+ r = await runScript([]);
+ assert.strictEqual(r.code, 1); assert.match(r.stderr, /Network resolve-batch 503|token endpoint 503/);
+ net.down = false; net.grant = true;
+ // A token Network refuses at resolve-batch: the key answers instead.
+ const resolver = ownerSubject.createResolver({ networkUrl: NET });
+ net.grant = true;
+ let got = await resolver.resolve('live', ['12', '99', 12]);
+ assert.deepStrictEqual([...got.entries()], [['12', s12], ['99', null]]);
+ assert.strictEqual(resolver.state.via, 'service-token');
+ const tokenOnly = ownerSubject.createResolver({ networkUrl: NET, internalKey: '' });
+ net.grant = false;
+ await assert.rejects(tokenOnly.resolve('live', ['12']), /no service token/);
+ net.grant = true;
+ console.log('✅ Network auth: service token first, INTERNAL_API_KEY when Network has not granted it; failures are errors');
+
+ // ── New objects: X-OV-Subject at creation ──
+ const express = require('express');
+ db.upsertApp({ app_id: 'live', api_key: 'live-key' });
+ const app = express();
+ app.use(express.json());
+ app.use('/api/v2/:app/objects', require('../server/objects/routes'));
+ const server = http.createServer(app);
+ await new Promise((res) => server.listen(0, '127.0.0.1', res));
+ const base = `http://127.0.0.1:${server.address().port}`;
+ const post = (headers, body) => fetch(`${base}/api/v2/live/objects`, { method: 'POST', headers: { authorization: 'Bearer live-key', 'content-type': 'application/json', ...headers }, body: JSON.stringify(body) }).then(async x => ({ status: x.status, body: await x.json() }));
+ let p = await post({ 'x-ov-subject': `user:${s13}` }, { kind: 'file', size_bytes: 3, user_id: 13 });
+ assert.strictEqual(p.status, 201, JSON.stringify(p.body));
+ assert.strictEqual(row(p.body.id).owner_subject, s13, 'X-OV-Subject is stored at creation');
+ p = await post({}, { kind: 'file', size_bytes: 3, user_id: 12 });
+ const userOnly = p.body.id;
+ assert.deepStrictEqual([row(userOnly).owner_subject, row(userOnly).owner_user_id], [null, 12]);
+
+ // ── New objects: the reconcile job fills the rest (projections included) ──
+ const clipId = db.createClip({ app_id: 'live', user_id: 13, channel_user_id: 12, title: 'c', file_path: path.join(env.CLIPS_PATH, 'c.webm'), start_time: 0, end_time: 5 }).lastInsertRowid;
+ const clipObj = db.get('SELECT object_id FROM clips WHERE id = ?', [clipId]).object_id;
+ assert.ok(clipObj && row(clipObj).owner_subject === null && row(clipObj).owner_user_id === 13);
+ net.calls.length = 0;
+ const jr = await job.runOnce();
+ assert.ok(jr && jr.filled >= 3, JSON.stringify(jr));
+ assert.strictEqual(jr.via, 'service-token');
+ assert.deepStrictEqual([row(userOnly).owner_subject, row(clipObj).owner_subject, row(late).owner_subject], [s12, s13, s13]);
+ assert.strictEqual(row(o.a).owner_subject, s12, 'the job also refills rows a rollback cleared (so it is turned off before one)');
+ assert.strictEqual(row(o.unknown).owner_subject, null);
+ assert.strictEqual(row(o.b).owner_subject, other, 'never over a non-null value');
+ // Network learns live user 99 later: the next run fills it.
+ const s99 = subjectOf(99);
+ assert.strictEqual((await job.runOnce()).filled, 1);
+ assert.strictEqual(row(o.unknown).owner_subject, s99);
+ // Nothing left to fill: a run asks Network nothing (games 12 is the only pending owner and has no source system).
+ net.calls.length = 0;
+ assert.deepStrictEqual(await job.runOnce(), { owners: 1, filled: 0, skipped: 0, unresolvable: 0, unsupported: 1, via: null });
+ assert.strictEqual(net.calls.length, 0);
+ // Network down: the run fails quietly and the next one retries.
+ net.down = true;
+ const quiet = console.warn; console.warn = () => {};
+ obj({ owner_user_id: 12 });
+ assert.strictEqual(await job.runOnce(), null);
+ console.warn = quiet;
+ assert.match(job.status().last_error, /503/);
+ net.down = false;
+ assert.strictEqual((await job.runOnce()).filled, 1);
+
+ // A re-projection keeps the subject; a re-projection to another owner drops it until the job resolves the new one.
+ model.sync('clip', clipId);
+ assert.strictEqual(row(clipObj).owner_subject, s13);
+ db.run('UPDATE clips SET user_id = 12 WHERE id = ?', [clipId]);
+ model.sync('clip', clipId);
+ assert.deepStrictEqual([row(clipObj).owner_user_id, row(clipObj).owner_subject], [12, null]);
+ await job.runOnce();
+ assert.strictEqual(row(clipObj).owner_subject, s12);
+ // The v2 object's public shape carries it (what Live's lineage resolver reads: owner.subject).
+ assert.strictEqual(model.objectPublic(row(clipObj)).owner.subject, s12);
+
+ // MEDIA_OWNER_SUBJECT_SYNC=0 keeps the job off.
+ config.ownerSubject.enabled = false;
+ const said = console.log; console.log = () => {};
+ assert.strictEqual(job.start(), false);
+ console.log = said;
+ config.ownerSubject.enabled = true;
+ console.log('✅ new objects: X-OV-Subject stored at creation; the reconcile job fills the rest, re-projection keeps or drops it');
+
+ server.close(); network.close();
+ db.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('owner-subject: all checks passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });