openvibers/openvibe.media Commits

Object-first writes (WS-G task 1, first stage of retiring C-75): every write to a vods/clips/files/pastes row now creates or updates its media_object in the same SQLite transaction (model.withObject; inside announce() for vod/clip outcomes, events still queued after the outcome), so a crash or error between the two can no longer leave drift, and a write whose object cannot be written changes nothing; the recorder start and tier moves fall back to the row alone because the bytes or ffmpeg already moved; a read-only drift report (server/objects/drift.js, scripts/object-drift-report.js) lists rows whose object is missing, unlinked or disagrees on kind, visibility, lifecycle, size, legacy ref or owner; the boot backfill stays until a release shows zero drift

OP OpenVibers committed 25 Sept 2026 · e9a0f3a50c · 20 files, +933 −248


Changes

ddiff --git a/docs/object-model.md b/docs/object-model.md
iindex e3aa6dd..b6d9a17 100644
---- a/docs/object-model.md
++++ b/docs/object-model.md
@@@ -54,8 +54,9 @@ Every v2 route that takes `:id` accepts either form. Native v2 objects have `leg
### How a row maps to its object
--`model.sync(kind, id)` derives the object and its locations from the row as it is at that moment.
--The backfill and every write hook use the same function.
++`model.sync(kind, id)` derives the object and its locations from the row as it is at that moment
++(the per-kind `<kind>Projection(row)` functions). The backfill, every write and the drift report use
++the same functions.
- **vod / clip.**
- Local copy: the file under `VOD_PATH` (vods, resolved by basename as playback does) or the clip's own path.
@@@ -82,37 +83,70 @@ relationships and thumbnail variants.
- **Never moves or deletes bytes, and never calls B2/R2.** A local copy is `present` when the file exists and `missing` when it does not. Remote copies stay `pending` until `reconcile-objects.js --verify` checks them.
- **One transaction,** with one savepoint per row: a failing row is reported and leaves nothing half-written. `--dry-run` rolls everything back and only reports.
- **Report:** counts per kind (seen, created, updated, skipped), location states, skipped rows with a reason, and errors. A real run stores the report in `media_settings` under `objects.backfill.last_report`.
--- **At boot:** the service runs the `--only-missing` form itself 15 s after start, for rows with no `object_id`. The first boot after the upgrade therefore projects everything.
++- **At boot:** the service runs the `--only-missing` form itself 15 s after start, for rows with no `object_id`. The first boot after the upgrade therefore projects everything. With object-first writes it should find nothing; it stays until the [drift report](#drift-report) shows zero drift across a release.
Skip reasons today: `clips-only recording (ephemeral, never published)`, `screenshot paste without a file path`, `external thumbnail url`.
The run holds SQLite's write lock for its duration, typically seconds. Prefer the boot backfill, or
run the script while traffic is quiet.
--### Keeping the model current
++### Keeping the model current: object-first writes
--The old write paths re-project their row after they change it:
++Every write to a projected row writes its object in the same SQLite transaction
++(`model.withObject(kind, ids, write)`; roadmap WS-G task 1, which retires compatibility shim C-75,
++"write the row, then sync the object"). The row change and the re-projection commit together or
++not at all, so a crash or an error between them can no longer leave the object behind its row. A
++write whose object cannot be written fails as a whole and changes nothing. Inside
++`webhooks.announce()` the object joins the outcome's transaction, and its object events are queued
++after the outcome event.
--| path | hook |
++| path | where |
|---|---|
--| vod create | `POST /vods` |
--| recording start | `recorder` sets `file_path` |
--| every finalize outcome | `finalize.finalizeVod` |
--| visibility | `db.setVodVisibility` / `setClipVisibility` |
--| health | `db.updateVodHealth` |
--| title | `PUT /vods/:id`, `PUT /clips/:id` |
++| vod create (visibility and clips-only in the same insert) | `db.createVod` (`POST /vods`) |
++| recording start | `recorder` sets `file_path` (see below) |
++| chunk upload, live progress | `POST /vods/:id/chunks`, recorder progress |
++| every finalize outcome | `finalize.finalizeVod`, inside the vod.ready / vod.failed transaction |
++| title, description, visibility | `PUT /vods/:id`, `PUT /clips/:id`, `db.setVodVisibility` / `setClipVisibility` |
++| health, duration repairs, quarantines | `db.updateVodHealth`, `db.repairVodDuration`, health job, duration reconcile, `vod-storage` quarantines |
| clip create | `db.createClip` (API cuts, uploads and auto-clips) |
--| clip cut / re-cut result | clips routes and `clip-jobs` |
++| clip cut / re-cut result | clips routes and `clip-jobs`, inside the clip.ready / clip.failed transaction |
| file upload | `db.createFile` |
| thumbnail generate / upload | `thumbnail-service`, `POST /thumbnails` |
--| screenshot upload / censor | pastes routes |
++| screenshot upload / update / censor | pastes routes |
| avatar ingest | `avatars/ingest` |
--| tier moves | `moveToCold`, `moveToHot`, `promoteToR2`, `demoteFromR2`, sweep, `migrateLegacy`: the copy the move verified is marked `present` |
++| tier moves | `moveToCold`, `moveToHot`, `promoteToR2`, `demoteFromR2`, sweep, `migrateLegacy`: the copy the move verified is marked `present` in the same transaction |
--**Deletes need no hook.** A SQLite trigger on `vods`, `clips`, `files` and `pastes` marks the
--object `deleted` whenever its row is deleted, which covers all of the inherited delete paths. Hooks
--never throw: the legacy write has already happened, so a failed sync only logs a warning. The next
--sync or backfill repairs it.
++**Deletes need nothing more.** A SQLite trigger on `vods`, `clips`, `files` and `pastes` marks the
++object `deleted` in the same statement whenever its row is deleted, which covers all of the
++inherited delete paths.
++
++**What stays outside the transaction.** The projection stats local files but never reads them:
++content hashes come from the content-hash job, and remote copies stay `pending` until a tier move or
++reconciliation verifies them. Two writes record something that already happened outside the
++database, so they never lose the row to its object: the recorder's start (ffmpeg is running) and the
++tier moves (the bytes have moved). If their combined transaction fails, the row is written alone and
++re-projected after it (`withObjectOrRow`), with a warning. After a move that removed the local file,
++and after a finalize that threw part-way, a follow-up re-projection re-reads the disk. Columns no
++object carries (view counts, `master_file_path`, `cut_attempts`, paste likes and AI fields) are
++written without one. `db.importLegacyRows` (the cutover import) is not converted: the boot backfill
++projects what it inserts.
++
++### Drift report
++
++```
++node scripts/object-drift-report.js [--app live] [--limit 20] [--json] [--out report.json] [--db ./data/media.db]
++```
++
++For every projected row, and each vod/clip thumbnail, the report derives what the object must say
++with the same projection functions and compares it with the object on record: `kind`, `visibility`,
++`lifecycle_status`, `size_bytes`, `legacy_ref`, `app_id`, `owner_app`, `owner_user_id`, and the link
++(the row's `object_id`, or the parent's `thumbnail` variant). A row is `missing` (no object),
++`unlinked` (the link names another object) or `mismatch` (a field differs). It prints counts per
++kind and up to `--limit` examples of each. The database is opened read-only and nothing is written;
++the exit code is always 0.
++
++The boot backfill (rows with no object) and the finalize follow-up stay until a release has run with
++zero drift. They are removed in a later, dated step.
## Owner subjects
ddiff --git a/scripts/object-drift-report.js b/scripts/object-drift-report.js
nnew file mode 100644
iindex 0000000..6716a14
---- /dev/null
++++ b/scripts/object-drift-report.js
@@@ -0,0 +1,40 @@
++#!/usr/bin/env node
++'use strict';
++/**
++ * Object drift report (WS-G task 1, retiring C-75; server/objects/drift.js has the rules): every
++ * projected vods/clips/files/pastes row, and each vod/clip thumbnail, whose media_object is missing,
++ * not linked, or disagrees with the row on kind, visibility, lifecycle status, size, legacy ref or
++ * owner/app. Prints counts per kind and up to --limit examples of each. READ-ONLY: the database is
++ * opened read-only and only local files are stat'ed (paths from the service's environment:
++ * VOD_PATH, CLIPS_PATH, FILES_PATH, THUMBNAILS_PATH, as the projection uses them).
++ *
++ * node scripts/object-drift-report.js [--app live] [--limit 20] [--json] [--out report.json] [--db ./data/media.db]
++ *
++ * Exit code 0 always: it is a report (an error is printed to stderr). Zero drift across a release
++ * is the signal to remove the boot backfill (server/index.js) and the finalize follow-up.
++ */
++const fs = require('fs');
++const path = require('path');
++
++const args = process.argv.slice(2);
++const has = (name) => args.includes(`--${name}`);
++const arg = (name, dflt) => { const i = args.indexOf(`--${name}`); return i >= 0 && args[i + 1] && !args[i + 1].startsWith('--') ? args[i + 1] : dflt; };
++
++try {
++ const dbPath = arg('db') ? path.resolve(arg('db')) : require('../server/config').db.path;
++ const Database = require('better-sqlite3');
++ const h = new Database(dbPath, { readonly: true, fileMustExist: true });
++ const q = {
++ all: (sql, params = []) => h.prepare(sql).all(...params),
++ get: (sql, params = []) => h.prepare(sql).get(...params),
++ };
++ const { buildReport, formatReport } = require('../server/objects/drift');
++ const limit = Math.max(0, parseInt(arg('limit', '20'), 10) || 0);
++ const report = buildReport(q, { appId: arg('app', null), limit });
++ h.close();
++ if (arg('out')) fs.writeFileSync(arg('out'), JSON.stringify(report, null, 2));
++ console.log(has('json') ? JSON.stringify(report, null, 2) : formatReport(report));
++} catch (err) {
++ console.error(`object-drift-report: ${err.message}`);
++}
++process.exit(0);
ddiff --git a/server/avatars/ingest.js b/server/avatars/ingest.js
iindex 59d0069..a45d60e 100644
---- a/server/avatars/ingest.js
++++ b/server/avatars/ingest.js
@@@ -118,11 +118,13 @@ function createIngestHandler({ db, config, screenshotsDir, generateSlug, log = c
const file = path.join(screenshotsDir, `avatar-n${parseInt(user_id, 10) || 0}-${Date.now()}-${crypto.randomBytes(4).toString('hex')}.webp`);
fs.writeFileSync(file, pic.buffer);
const slug = generateSlug();
-- const info = db.run(`INSERT INTO pastes (app_id, slug, user_id, type, title, content, language, visibility, screenshot_path, metadata, ip_address)
++ const insert = () => db.run(`INSERT INTO pastes (app_id, slug, user_id, type, title, content, language, visibility, screenshot_path, metadata, ip_address)
VALUES ('network', ?, ?, 'screenshot', ?, '', 'text', 'unlisted', ?, ?, NULL)`,
[slug, parseInt(user_id, 10) || null, `Avatar${username ? ' of ' + String(username).slice(0, 40) : ''}`, file,
JSON.stringify({ kind: 'avatar', source_host: (() => { try { return new URL(url).hostname; } catch { return null; } })(), size_bytes: pic.buffer.length, mime_type: 'image/webp' })]);
-- if (info && typeof db.syncObject === 'function') db.syncObject('paste', info.lastInsertRowid); // the avatar's media_object
++ // The row and the avatar's media_object in one transaction (a database without the object model: the row alone).
++ if (typeof db.withObject === 'function') db.withObject('paste', (r) => r.lastInsertRowid, insert);
++ else insert();
res.json({ ok: true, slug, url: `${config.publicUrl}/p/${slug}/screenshot`, width: pic.width, height: pic.height, bytes: pic.buffer.length });
} catch (err) {
log.warn('[Avatar ingest]', err.message);
ddiff --git a/server/db/database.js b/server/db/database.js
iindex 1a5b112..0ebe415 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -172,9 +172,13 @@ function _ensureObjectTriggers() {
BEGIN INSERT INTO media_object_changes (object_id, change, previous_visibility) VALUES (NEW.id, 'deleted', OLD.visibility); END;`);
}
--/** Keep a projected row's media_object current. Never throws — the legacy write already happened. */
--function _syncObject(kind, id) {
-- return require('../objects/model').safeSync(kind, id);
++/**
++ * Object-first write (WS-G task 1, retiring C-75): run write() — a change to a projected
++ * vods/clips/files/pastes row — and re-project the row's media_object in the same transaction
++ * (objects/model.js withObject). Throws, with nothing written, when either part fails.
++ */
++function withObject(kind, ids, write) {
++ return require('../objects/model').withObject(kind, ids, write);
}
function seedSettings() {
@@@ -299,14 +303,16 @@ function appAllowedOrigins(app) {
// ── VOD helpers ──────────────────────────────────────────────
--function createVod({ app_id, stream_id, stream_key, managed_stream_id, user_id, title, description, file_path, file_size, duration_seconds, thumbnail_url, master_file_path, meta }) {
-- return run(
-- `INSERT INTO vods (app_id, stream_id, stream_key, managed_stream_id, user_id, title, description, file_path, master_file_path, file_size, duration_seconds, thumbnail_url, meta_json, is_public)
-- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1)`,
++// The row and its media_object in one transaction (a clips-only recording has no object).
++function createVod({ app_id, stream_id, stream_key, managed_stream_id, user_id, title, description, file_path, file_size, duration_seconds, thumbnail_url, master_file_path, meta, visibility, clips_only }) {
++ const vis = visibility ? _normVisibility(visibility) : 'public';
++ return withObject('vod', (r) => r.lastInsertRowid, () => run(
++ `INSERT INTO vods (app_id, stream_id, stream_key, managed_stream_id, user_id, title, description, file_path, master_file_path, file_size, duration_seconds, thumbnail_url, meta_json, is_public, visibility, clips_only)
++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
[app_id, stream_id || null, stream_key || null, managed_stream_id || null, user_id || null, title || 'Recording', description || '',
file_path || null, master_file_path || null, file_size || 0, duration_seconds || 0, thumbnail_url || null,
-- JSON.stringify(meta || {})]
-- );
++ JSON.stringify(meta || {}), vis === 'public' ? 1 : 0, vis, clips_only ? 1 : 0]
++ ));
}
/** Legacy lookup: resolve a row by its file's basename (old /api/vods/file/<name> URLs). */
@@@ -458,12 +464,10 @@ function countVods(appId, filters = {}) {
const VALID_VISIBILITY = new Set(['public', 'unlisted', 'private']);
function _normVisibility(v) { return VALID_VISIBILITY.has(v) ? v : 'public'; }
--// Set VOD/clip visibility; is_public mirrors (1 iff public) so listing filters hold.
++// Set VOD/clip visibility; is_public mirrors (1 iff public) so listing filters hold. Row and object together.
function setVodVisibility(vodId, visibility) {
const vis = _normVisibility(visibility);
-- const r = run('UPDATE vods SET visibility = ?, is_public = ? WHERE id = ?', [vis, vis === 'public' ? 1 : 0, vodId]);
-- _syncObject('vod', vodId);
-- return r;
++ return withObject('vod', vodId, () => run('UPDATE vods SET visibility = ?, is_public = ? WHERE id = ?', [vis, vis === 'public' ? 1 : 0, vodId]));
}
function updateVodHealth(vodId, { status, score, issues = [], probeDuration, probeFormat, quarantine = false, keepPublic = false }) {
@@@ -481,17 +485,15 @@ function updateVodHealth(vodId, { status, score, issues = [], probeDuration, pro
updates.push("last_health_scan_at = datetime('now')");
params.push(vodId);
if (!updates.length) return null;
-- const r = run(`UPDATE vods SET ${updates.join(', ')} WHERE id = ?`, params);
-- _syncObject('vod', vodId);
-- return r;
++ return withObject('vod', vodId, () => run(`UPDATE vods SET ${updates.join(', ')} WHERE id = ?`, params));
}
--/** A measured duration (source probe | remux) replaces the stored one. */
++/** A measured duration (source probe | remux) replaces the stored one (the object's size and duration with it). */
function repairVodDuration(vodId, duration, fileSize, source = 'probe') {
-- return run(
++ return withObject('vod', vodId, () => run(
`UPDATE vods SET duration_seconds = ?, file_size = ?, probe_duration_seconds = ?, duration_source = ?, last_health_scan_at = datetime('now') WHERE id = ?`,
[duration, fileSize, duration, source, vodId]
-- );
++ ));
}
// VODs the periodic health job should scan: finished (not recording), and either never
@@@ -527,17 +529,18 @@ function vodStatus(vod) {
// ── Clip helpers ─────────────────────────────────────────────
--function createClip({ app_id, vod_id, stream_id, user_id, channel_user_id, title, description, file_path, thumbnail_url, start_time, end_time, duration_seconds, is_public, auto_generated, status }) {
++// The row and its media_object in one transaction. `visibility` (public | unlisted | private), when
++// given, sets both columns; otherwise is_public picks public or unlisted as before.
++function createClip({ app_id, vod_id, stream_id, user_id, channel_user_id, title, description, file_path, thumbnail_url, start_time, end_time, duration_seconds, is_public, visibility, auto_generated, status }) {
const pub = (is_public === 0 || is_public === false) ? 0 : 1;
-- const r = run(
++ const vis = visibility ? _normVisibility(visibility) : (pub ? 'public' : 'unlisted');
++ return withObject('clip', (r) => r.lastInsertRowid, () => run(
`INSERT INTO clips (app_id, vod_id, stream_id, user_id, channel_user_id, title, description, file_path, thumbnail_url, start_time, end_time, duration_seconds, is_public, visibility, auto_generated, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
[app_id, vod_id || null, stream_id || null, user_id || null, channel_user_id || null, title || 'Untitled Clip', description || '',
file_path || '', thumbnail_url || null, start_time || 0, end_time || 0, duration_seconds || 0,
-- pub, pub ? 'public' : 'unlisted', auto_generated ? 1 : 0, status || 'ready']
-- );
-- _syncObject('clip', r.lastInsertRowid);
-- return r;
++ vis === 'public' ? 1 : 0, vis, auto_generated ? 1 : 0, status || 'ready']
++ ));
}
function getClipById(id, appId = null) {
@@@ -580,9 +583,7 @@ function countClips(appId, filters = {}) {
function setClipVisibility(clipId, visibility) {
const vis = _normVisibility(visibility);
-- const r = run('UPDATE clips SET visibility = ?, is_public = ? WHERE id = ?', [vis, vis === 'public' ? 1 : 0, clipId]);
-- _syncObject('clip', clipId);
-- return r;
++ return withObject('clip', clipId, () => run('UPDATE clips SET visibility = ?, is_public = ? WHERE id = ?', [vis, vis === 'public' ? 1 : 0, clipId]));
}
function findDuplicateClip({ appId, streamId = null, vodId = null, startTime = 0, endTime = 0, startWindow = 8, endWindow = 10, createdSinceMinutes = 10 }) {
@@@ -722,14 +723,13 @@ function getRecentPasteCommentsByIp(ip, seconds = 10) {
// ── File helpers ─────────────────────────────────────────────
++// The row and its media_object in one transaction (the file is already on disk, sha256 computed by the route).
function createFile({ key, app_id, user_id, original_name, size, mime, sha256 }) {
-- const r = run(
++ return withObject('file', key, () => run(
`INSERT INTO files (key, app_id, user_id, original_name, size, mime, sha256)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
[key, app_id, user_id || null, original_name || null, size || 0, mime || 'application/octet-stream', sha256 || null]
-- );
-- _syncObject('file', key);
-- return r;
++ ));
}
function getFileByKey(key, appId = null) {
@@@ -770,6 +770,7 @@ function listFiles(appId, { limit = 100, offset = 0 } = {}) {
return all('SELECT * FROM files WHERE app_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ?', [appId, limit, offset]);
}
++// The object is marked deleted by trg_files_object_deleted in the same statement.
function deleteFileRow(key) {
return run('DELETE FROM files WHERE key = ?', [key]);
}
@@@ -784,7 +785,8 @@ function appFilesBytes(appId) {
* Bulk-insert rows exported from the predecessor DB. Only columns present in
* the destination table are used; missing ones take schema defaults; app_id is
* backfilled. Row ids are preserved (INSERT OR IGNORE keeps re-runs idempotent).
-- * Returns { inserted, skipped }.
++ * Returns { inserted, skipped }. The one write path that is not object-first (the cutover
++ * import): the boot backfill (objects/backfill.js, rows with no object_id) projects what it inserts.
*/
function importLegacyRows(table, rows, appId = 'live') {
const allowed = new Set(['vods', 'clips', 'pastes', 'paste_likes', 'paste_comments', 'content_views', 'files']);
@@@ -812,7 +814,7 @@ function importLegacyRows(table, rows, appId = 'live') {
}
module.exports = {
-- getDb, run, get, all, close, syncObject: _syncObject, heldSql,
++ getDb, run, get, all, close, withObject, heldSql,
getSetting, setSetting,
// apps
hashApiKey, getApp, listApps, upsertApp, appAllowedOrigins, projectTenantId, ensureProjectTenant, isSandboxTenant,
ddiff --git a/server/index.js b/server/index.js
iindex ef8dd62..86bd3f2 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -234,6 +234,10 @@ if (!drill.enabled) {
try { const r = require('./objects/multipart').purgeExpired(); if (r.expired || r.orphan_dirs) console.log(`[Objects] Multipart: ${r.expired} expired session(s), ${r.orphan_dirs} orphan part dir(s) removed`); } catch (err) { console.warn('[Objects] multipart purge:', err.message); }
});
// Project rows that have no media_object yet (first boot after the upgrade: all of them).
++ // C-75 catch-up: every write now makes its object in the same transaction (objects/model.js
++ // withObject, WS-G task 1), so this should find nothing. It stays until a release has run with
++ // `node scripts/object-drift-report.js` showing zero drift; then it goes, with the finalize
++ // follow-up re-projection (vod/finalize.js), in a dated later step.
setTimeout(() => {
try {
const bf = require('./objects/backfill');
ddiff --git a/server/objects/backfill.js b/server/objects/backfill.js
iindex 3ff0083..9b6d136 100644
---- a/server/objects/backfill.js
++++ b/server/objects/backfill.js
@@@ -3,7 +3,7 @@
*
* One media_object (+ locations, relationships, thumbnail variant) per existing
* vod, clip, file, screenshot/avatar paste and vod/clip thumbnail, derived by the
-- * same sync functions the write hooks use. Idempotent: objects are keyed by their
++ * same sync functions every write uses (model.withObject). Idempotent: objects are keyed by their
* legacy ref, so a re-run updates in place and creates nothing new.
*
* Never touches bytes and never calls B2/R2: local copies are 'present' when the
ddiff --git a/server/objects/drift.js b/server/objects/drift.js
nnew file mode 100644
iindex 0000000..82a3cbe
---- /dev/null
++++ b/server/objects/drift.js
@@@ -0,0 +1,152 @@
++/**
++ * OpenVibe.Media — object drift report (WS-G task 1: retiring compatibility shim C-75)
++ *
++ * Every write to an inherited vods/clips/files/pastes row now makes its media_object in the same
++ * transaction (model.withObject). This report checks that nothing has drifted anyway: for each
++ * projected row it derives what the object must say, with the same <kind>Projection() functions
++ * the writes use, and compares that with the object on record, on the fields a sync writes:
++ *
++ * kind, visibility, lifecycle_status, size_bytes, legacy_ref, app_id, owner_app, owner_user_id
++ *
++ * and the link to it (the row's object_id; for a vod/clip thumbnail, its parent's thumbnail
++ * variant). Each checked row is ok, or has one or more of:
++ * missing no object under its legacy ref, nor at the row's object_id
++ * unlinked the row's object_id (the thumbnail variant) is not the object its legacy ref names
++ * mismatch the object disagrees with the row on one of the fields above
++ * Rows that have no object by design (clips-only recordings, screenshot pastes without a file,
++ * external thumbnail URLs) are counted as skipped; text pastes are not projected and not listed.
++ *
++ * READ-ONLY. Pure SQL over a small `q` interface ({ all(sql, params), get(sql, params) }), so the
++ * script (scripts/object-drift-report.js) runs it on a database opened read-only; beyond that it
++ * only stats local files, as the projection does. Nothing is written, re-projected or moved.
++ */
++'use strict';
++
++const model = require('./model');
++
++const FIELDS = ['kind', 'visibility', 'lifecycle_status', 'size_bytes', 'legacy_ref', 'app_id', 'owner_app', 'owner_user_id'];
++const KINDS = ['vod', 'clip', 'file', 'screenshot', 'avatar', 'thumbnail'];
++const PAGE = 500;
++const OBJECT_COLS = `id, ${FIELDS.join(', ')}`;
++
++/** The compared fields as project() writes them from a projection input. */
++function expectedFields(p) {
++ return {
++ kind: p.kind, visibility: p.visibility, lifecycle_status: p.lifecycle_status, size_bytes: Number(p.size_bytes) || 0,
++ legacy_ref: p.legacy_ref, app_id: p.app_id, owner_app: p.owner_app || p.app_id, owner_user_id: p.owner_user_id ?? null,
++ };
++}
++
++// SQLite hands integers back as numbers and TEXT as strings; an app-local user id may be either.
++const same = (a, b) => (a == null ? '' : String(a)) === (b == null ? '' : String(b));
++
++/** Every row of `table` (optionally one tenant's) in key order, a page at a time. */
++function* eachRow(q, table, keyCol, { appId = null, where = null } = {}) {
++ const conds = [];
++ const params = [];
++ if (appId) { conds.push('app_id = ?'); params.push(appId); }
++ if (where) conds.push(where);
++ let last = null;
++ for (;;) {
++ const c = last == null ? conds : [...conds, `${keyCol} > ?`];
++ const page = q.all(`SELECT * FROM ${table}${c.length ? ` WHERE ${c.join(' AND ')}` : ''} ORDER BY ${keyCol} LIMIT ${PAGE}`,
++ last == null ? params : [...params, last]);
++ for (const r of page) yield r;
++ if (page.length < PAGE) return;
++ last = page[page.length - 1][keyCol];
++ }
++}
++
++function buildReport(q, { appId = null, limit = 20 } = {}) {
++ const report = {
++ generated_at: new Date().toISOString(),
++ app_id: appId,
++ limit,
++ fields: FIELDS,
++ kinds: Object.fromEntries(KINDS.map(k => [k, { rows: 0, ok: 0, skipped: 0, drifted: 0, missing: 0, unlinked: 0, mismatch: 0, examples: [] }])),
++ field_counts: Object.fromEntries(FIELDS.map(f => [f, 0])),
++ total_rows: 0,
++ total_drift: 0,
++ };
++ // Objects by legacy ref, loaded once (the compared columns only); a row's object_id is looked up when its ref finds none.
++ const byRef = new Map(q.all(`SELECT ${OBJECT_COLS} FROM media_objects WHERE legacy_ref IS NOT NULL`).map(o => [o.legacy_ref, o]));
++ const byId = (id) => (id ? q.get(`SELECT ${OBJECT_COLS} FROM media_objects WHERE id = ?`, [id]) || null : null);
++
++ /** Compare one row's projection with its object. Returns the object found (or null). */
++ const check = (kind, where, p, linkedId) => {
++ const k = report.kinds[kind];
++ k.rows++;
++ report.total_rows++;
++ const found = byRef.get(p.legacy_ref) || byId(linkedId);
++ const problems = [];
++ const diff = {};
++ if (!found) {
++ problems.push('missing');
++ } else {
++ if (!same(linkedId, found.id)) { problems.push('unlinked'); diff.object_id = { row: linkedId || null, object: found.id }; }
++ const want = expectedFields(p);
++ for (const f of FIELDS) {
++ if (same(want[f], found[f])) continue;
++ diff[f] = { row: want[f], object: found[f] ?? null };
++ report.field_counts[f]++;
++ }
++ if (FIELDS.some(f => f in diff)) problems.push('mismatch');
++ }
++ if (!problems.length) { k.ok++; return found; }
++ k.drifted++;
++ report.total_drift++;
++ for (const pr of problems) k[pr]++;
++ if (k.examples.length < limit) k.examples.push({ ...where, legacy_ref: p.legacy_ref, object_id: found ? found.id : null, problems, diff });
++ return found;
++ };
++ const skip = (kind) => { report.kinds[kind].skipped++; };
++ const thumbnailOf = (row, table, parentKind, parent) => {
++ const tp = model.thumbnailProjection(row, parentKind);
++ if (!tp) return;
++ if (tp.skipped) return skip('thumbnail');
++ const variant = parent ? q.get("SELECT derived_object_id FROM media_variants WHERE object_id = ? AND variant_name = 'thumbnail'", [parent.id]) : null;
++ check('thumbnail', { table, id: row.id, of: parentKind }, tp, variant ? variant.derived_object_id : null);
++ };
++
++ for (const row of eachRow(q, 'vods', 'id', { appId })) {
++ const p = model.vodProjection(row);
++ if (p.skipped) { skip('vod'); continue; }
++ thumbnailOf(row, 'vods', 'vod', check('vod', { table: 'vods', id: row.id }, p, row.object_id));
++ }
++ for (const row of eachRow(q, 'clips', 'id', { appId })) {
++ thumbnailOf(row, 'clips', 'clip', check('clip', { table: 'clips', id: row.id }, model.clipProjection(row), row.object_id));
++ }
++ for (const row of eachRow(q, 'files', 'key', { appId })) {
++ check('file', { table: 'files', id: row.key }, model.fileProjection(row), row.object_id);
++ }
++ for (const row of eachRow(q, 'pastes', 'id', { appId, where: "type = 'screenshot'" })) {
++ const p = model.pasteProjection(row);
++ if (p.skipped) { skip('screenshot'); continue; }
++ check(p.kind, { table: 'pastes', id: row.id, slug: row.slug }, p, row.object_id);
++ }
++ return report;
++}
++
++const fmt = (v) => (v == null ? 'null' : typeof v === 'string' ? v : JSON.stringify(v));
++
++function formatReport(r) {
++ const lines = [`Object drift report — ${r.generated_at}${r.app_id ? ` (app ${r.app_id})` : ' (all apps)'}`];
++ lines.push(` ${'kind'.padEnd(11)} ${'rows'.padStart(7)} ${'ok'.padStart(7)} ${'drifted'.padStart(8)} ${'missing'.padStart(8)} ${'unlinked'.padStart(9)} ${'mismatch'.padStart(9)} ${'skipped'.padStart(8)}`);
++ for (const [kind, c] of Object.entries(r.kinds)) {
++ lines.push(` ${kind.padEnd(11)} ${String(c.rows).padStart(7)} ${String(c.ok).padStart(7)} ${String(c.drifted).padStart(8)} ${String(c.missing).padStart(8)} ${String(c.unlinked).padStart(9)} ${String(c.mismatch).padStart(9)} ${String(c.skipped).padStart(8)}`);
++ }
++ const fields = Object.entries(r.field_counts).filter(([, n]) => n).map(([f, n]) => `${f} ${n}`);
++ lines.push(` fields that disagree: ${fields.join(', ') || 'none'}`);
++ lines.push(` total drift: ${r.total_drift} of ${r.total_rows} row(s)${r.total_drift ? '' : ' — none'}`);
++ for (const [kind, c] of Object.entries(r.kinds)) {
++ if (!c.examples.length) continue;
++ lines.push(` ${kind} examples (${c.examples.length} of ${c.drifted}):`);
++ for (const e of c.examples) {
++ const what = Object.entries(e.diff).map(([f, d]) => `${f} row=${fmt(d.row)} object=${fmt(d.object)}`).join('; ');
++ lines.push(` ${e.table} ${e.id}${e.of ? ` (${e.of} thumbnail)` : ''} ${e.object_id || '-'} ${e.problems.join('+')}${what ? `: ${what}` : ` (${e.legacy_ref})`}`);
++ }
++ }
++ return lines.join('\n');
++}
++
++module.exports = { buildReport, formatReport, expectedFields, FIELDS, KINDS };
ddiff --git a/server/objects/model.js b/server/objects/model.js
iindex 79035a2..983e655 100644
---- a/server/objects/model.js
++++ b/server/objects/model.js
@@@ -5,8 +5,10 @@
* row per provider copy. The inherited vods/clips/files/pastes rows are typed
* projections over objects: each carries object_id, and the sync*() functions
* below derive the object + locations from the row as it is right now. The same
-- * functions back the one-off backfill and the write hooks on the old APIs, so
-- * the model cannot drift between "imported" and "written since".
++ * functions back the one-off backfill and every write on the old APIs, which
++ * writes the row and its object in one transaction (withObject, WS-G task 1), so
++ * the model cannot drift between "imported" and "written since". The drift
++ * report (./drift.js) checks that it has not.
*
* Location states: present (verified: the local file exists / a HEAD answered),
* missing (verified absent), pending (believed there, not yet verified — every
@@@ -294,6 +296,10 @@ function recordInvariant(objectId) {
}
// ── Per-kind projections ─────────────────────────────────────
++// <kind>Projection(row) derives what the object behind one inherited row must say (project()'s input).
++// Read-only (it stats local files, never writes), and shared by the sync*() writers below and the
++// drift report (./drift.js), so the report checks exactly what a write would have written. A row
++// that has no object of its own answers { skipped: reason }.
function vodLifecycle(row) {
const s = db.vodStatus(row);
@@@ -301,14 +307,12 @@ function vodLifecycle(row) {
return 'uploading';
}
--function syncVod(idOrRow) {
-- const row = typeof idOrRow === 'object' ? idOrRow : db.get('SELECT * FROM vods WHERE id = ?', [idOrRow]);
-- if (!row) return null;
++function vodProjection(row) {
if (row.clips_only) return { skipped: 'clips-only recording (ephemeral, never published)' };
const vs = vodStorage();
const { locs, canonical } = tieredLocations(row, [row.file_path ? vs.localPathForVod(row) : null, row.file_path]);
const local = locs.find(l => l.provider === 'local' && l.state === 'present');
-- const r = project({
++ return {
legacy_ref: legacyRef(row.app_id, 'vod', row.id), existingId: row.object_id,
app_id: row.app_id, kind: 'vod', owner_user_id: row.user_id,
visibility: rowVisibility(row), lifecycle_status: vodLifecycle(row),
@@@ -317,22 +321,16 @@ function syncVod(idOrRow) {
metadata: { title: row.title || null, duration_seconds: Number(row.duration_seconds) || 0, health_status: row.health_status || null,
managed_stream_id: row.managed_stream_id || null, recording: !!row.is_recording },
created_at: row.created_at, locations: locs, canonical,
-- });
-- linkRow('vods', 'id', row.id, r.id);
-- recordInvariant(r.id);
-- const thumbnail = syncThumbnail(row, r.id, 'vod');
-- return { ...r, locations: locs.map(l => l.state), thumbnail };
++ };
}
--function syncClip(idOrRow) {
-- const row = typeof idOrRow === 'object' ? idOrRow : db.get('SELECT * FROM clips WHERE id = ?', [idOrRow]);
-- if (!row) return null;
++function clipProjection(row) {
const clipsDir = path.resolve(config.vod.clipsPath);
const { locs, canonical } = tieredLocations(row, [row.file_path || null, row.file_path ? path.join(clipsDir, path.basename(row.file_path)) : null]);
const status = row.status || 'ready';
const lifecycle = status === 'processing' ? 'uploading' : status === 'failed' ? 'failed' : (row.file_path ? 'ready' : 'failed');
const local = locs.find(l => l.provider === 'local' && l.state === 'present');
-- const r = project({
++ return {
legacy_ref: legacyRef(row.app_id, 'clip', row.id), existingId: row.object_id,
app_id: row.app_id, kind: 'clip', owner_user_id: row.user_id,
visibility: rowVisibility(row), lifecycle_status: lifecycle,
@@@ -341,23 +339,13 @@ function syncClip(idOrRow) {
metadata: { title: row.title || null, duration_seconds: Number(row.duration_seconds) || 0, start_time: row.start_time, end_time: row.end_time,
channel_user_id: row.channel_user_id || null, auto_generated: !!row.auto_generated },
created_at: row.created_at, locations: locs, canonical,
-- });
-- linkRow('clips', 'id', row.id, r.id);
-- if (row.vod_id) {
-- const vod = db.get('SELECT object_id FROM vods WHERE id = ?', [row.vod_id]);
-- if (vod && vod.object_id) setRelationship(r.id, 'clip_of', vod.object_id, { start_time: row.start_time, end_time: row.end_time });
-- }
-- recordInvariant(r.id);
-- const thumbnail = syncThumbnail(row, r.id, 'clip');
-- return { ...r, locations: locs.map(l => l.state), thumbnail };
++ };
}
--function syncFile(keyOrRow) {
-- const row = typeof keyOrRow === 'object' ? keyOrRow : db.get('SELECT * FROM files WHERE key = ?', [keyOrRow]);
-- if (!row) return null;
++function fileProjection(row) {
const local = localLocation([path.join(config.files.path, row.app_id, row.key)]);
if (local.state === 'present' && row.sha256) local.checksum = row.sha256;
-- const r = project({
++ return {
legacy_ref: legacyRef(row.app_id, 'file', row.key), existingId: row.object_id,
app_id: row.app_id, kind: 'file', owner_user_id: row.user_id,
visibility: 'public', // /f/:key serves every file without auth
@@@ -365,21 +353,17 @@ function syncFile(keyOrRow) {
size_bytes: Number(row.size) || 0, content_hash: row.sha256 || null,
metadata: { filename: row.original_name || null },
created_at: row.created_at, locations: [local], canonical: 'local',
-- });
-- linkRow('files', 'key', row.key, r.id);
-- return { ...r, locations: [local.state] };
++ };
}
/** Screenshot pastes (and avatars, which are stored as screenshot pastes) — the bytes only; text is Community's. */
--function syncPaste(idOrRow) {
-- const row = typeof idOrRow === 'object' ? idOrRow : db.get('SELECT * FROM pastes WHERE id = ?', [idOrRow]);
-- if (!row) return null;
++function pasteProjection(row) {
if (row.type !== 'screenshot') return { skipped: 'text paste (no bytes)' };
if (!row.screenshot_path) return { skipped: 'screenshot paste without a file path' };
const meta = parseJson(row.metadata, {}) || {};
const kind = meta.kind === 'avatar' ? 'avatar' : 'screenshot';
const local = localLocation([row.screenshot_path]);
-- const r = project({
++ return {
legacy_ref: legacyRef(row.app_id, kind === 'avatar' ? 'avatar' : 'paste', row.slug), existingId: row.object_id,
app_id: row.app_id, kind, owner_user_id: row.user_id,
visibility: VISIBILITIES.includes(row.visibility) ? row.visibility : 'public', lifecycle_status: 'ready',
@@@ -387,9 +371,53 @@ function syncPaste(idOrRow) {
size_bytes: local.state === 'present' ? local.size_bytes : (Number(meta.size_bytes) || 0),
metadata: { title: row.title || null, slug: row.slug },
created_at: row.created_at, locations: [local], canonical: 'local',
-- });
++ };
++}
++
++function syncVod(idOrRow) {
++ const row = typeof idOrRow === 'object' ? idOrRow : db.get('SELECT * FROM vods WHERE id = ?', [idOrRow]);
++ if (!row) return null;
++ const p = vodProjection(row);
++ if (p.skipped) return p;
++ const r = project(p);
++ linkRow('vods', 'id', row.id, r.id);
++ recordInvariant(r.id);
++ const thumbnail = syncThumbnail(row, r.id, 'vod');
++ return { ...r, locations: p.locations.map(l => l.state), thumbnail };
++}
++
++function syncClip(idOrRow) {
++ const row = typeof idOrRow === 'object' ? idOrRow : db.get('SELECT * FROM clips WHERE id = ?', [idOrRow]);
++ if (!row) return null;
++ const p = clipProjection(row);
++ const r = project(p);
++ linkRow('clips', 'id', row.id, r.id);
++ if (row.vod_id) {
++ const vod = db.get('SELECT object_id FROM vods WHERE id = ?', [row.vod_id]);
++ if (vod && vod.object_id) setRelationship(r.id, 'clip_of', vod.object_id, { start_time: row.start_time, end_time: row.end_time });
++ }
++ recordInvariant(r.id);
++ const thumbnail = syncThumbnail(row, r.id, 'clip');
++ return { ...r, locations: p.locations.map(l => l.state), thumbnail };
++}
++
++function syncFile(keyOrRow) {
++ const row = typeof keyOrRow === 'object' ? keyOrRow : db.get('SELECT * FROM files WHERE key = ?', [keyOrRow]);
++ if (!row) return null;
++ const p = fileProjection(row);
++ const r = project(p);
++ linkRow('files', 'key', row.key, r.id);
++ return { ...r, locations: [p.locations[0].state] };
++}
++
++function syncPaste(idOrRow) {
++ const row = typeof idOrRow === 'object' ? idOrRow : db.get('SELECT * FROM pastes WHERE id = ?', [idOrRow]);
++ if (!row) return null;
++ const p = pasteProjection(row);
++ if (p.skipped) return p;
++ const r = project(p);
linkRow('pastes', 'id', row.id, r.id);
-- return { ...r, kind, locations: [local.state] };
++ return { ...r, kind: p.kind, locations: [p.locations[0].state] };
}
/** Thumbnail file name behind a stored thumbnail_url, or null when it is not one of ours. */
@@@ -407,64 +435,139 @@ function thumbFileFromUrl(u) {
/**
* A vod/clip's thumbnail is one object per parent, updated in place when the
* picture is regenerated (live recordings refresh theirs every couple of minutes).
++ * null when the parent has no thumbnail.
*/
--function syncThumbnail(parentRow, parentObjectId, parentKind) {
++function thumbnailProjection(parentRow, parentKind) {
if (!parentRow.thumbnail_url) return null;
const name = thumbFileFromUrl(parentRow.thumbnail_url);
if (!name) return { skipped: 'external thumbnail url' };
const local = localLocation([path.join(path.resolve(config.thumbnails.path), name)]);
-- const variant = getVariant(parentObjectId, 'thumbnail');
const parentVis = rowVisibility(parentRow);
-- const r = project({
-- legacy_ref: legacyRef(parentRow.app_id, 'thumbnail', name), existingId: variant ? variant.derived_object_id : null,
++ return {
++ legacy_ref: legacyRef(parentRow.app_id, 'thumbnail', name),
app_id: parentRow.app_id, kind: 'thumbnail', owner_user_id: parentRow.user_id,
// /t/<name> is served to anyone holding the name.
visibility: parentVis === 'public' ? 'public' : 'unlisted', lifecycle_status: 'ready',
mime_type: mimeFor(name, 'image/jpeg'), size_bytes: local.state === 'present' ? local.size_bytes : 0,
metadata: { of: parentKind },
created_at: parentRow.created_at, locations: [local], canonical: 'local',
-- });
++ };
++}
++
++function syncThumbnail(parentRow, parentObjectId, parentKind) {
++ const p = thumbnailProjection(parentRow, parentKind);
++ if (!p || p.skipped) return p;
++ const variant = getVariant(parentObjectId, 'thumbnail');
++ const r = project({ ...p, existingId: variant ? variant.derived_object_id : null });
setVariant(parentObjectId, 'thumbnail', r.id);
setRelationship(r.id, 'thumbnail_of', parentObjectId);
-- return { ...r, locations: [local.state] };
++ return { ...r, locations: [p.locations[0].state] };
}
--/** One projection in one transaction, with the object-change events it caused (server/events.js). */
--function _syncTx(fn, arg) {
++/**
++ * Projection work in one transaction, with the object-change events it caused (server/events.js).
++ * Nested in an outer transaction (webhooks.announce, another withObject) it is a savepoint, and the
++ * outer one records the changes at its end, in the same transaction and after its own outcome event,
++ * the order they had when the object was synced after the commit.
++ */
++function _syncTx(fn) {
const events = require('../events');
-- const out = db.getDb().transaction(() => {
-- const r = fn(arg);
-- events.recordObjectChanges();
++ const raw = db.getDb();
++ const outer = raw.inTransaction;
++ const out = raw.transaction(() => {
++ const r = fn();
++ if (!outer) events.recordObjectChanges();
return r;
})();
-- events.kick();
++ if (!outer) events.kick();
return out;
}
++const SYNC_BY_KIND = { vod: syncVod, clip: syncClip, file: syncFile, paste: syncPaste };
++
++/** Row ids named by withObject()'s `ids` (a value, a list, or a function of write()'s result). */
++function _rowIds(ids, out) {
++ return [].concat((typeof ids === 'function' ? ids(out) : ids) ?? []).filter(id => id != null);
++}
++
/** Re-project one inherited row: kind vod|clip|file|paste, id = row id (file key for files). */
function sync(kind, id) {
-- const fn = { vod: syncVod, clip: syncClip, file: syncFile, paste: syncPaste }[kind];
++ const fn = SYNC_BY_KIND[kind];
if (!fn || id == null) return null;
-- return _syncTx(fn, id);
++ return _syncTx(() => fn(id));
}
--/** Never-throwing variant for write hooks outside database.js. */
++/**
++ * Never-throwing re-projection, for follow-ups that re-read what is on disk after a write already
++ * committed with its object (see withObject), and for scripts. Not a write path any more.
++ */
function safeSync(kind, id) {
try { return sync(kind, id); } catch (err) { console.warn(`[Objects] sync ${kind} ${id}:`, err.message); return null; }
}
/**
-- * After a tier move that verified a copy (upload + HEAD, copy + HEAD): re-project,
-- * then mark the verified providers present.
++ * Object-first write (WS-G task 1; retires compatibility shim C-75, "write the row, then sync its
++ * object"): write() changes inherited rows and the objects behind them are re-projected in the SAME
++ * SQLite transaction, so both commit or neither does: no crash or error between the two can leave
++ * an object behind its row. kind vod|clip|file|paste; ids the row id (a file's key), a list of them,
++ * or a function of write()'s result that gives them (an INSERT's lastInsertRowid). write() must be
++ * synchronous database work. Returns write()'s result; throws, with nothing written, when either
++ * part fails. Inside an outer transaction (webhooks.announce) it is a savepoint of that one.
++ *
++ * The projection stats local files (sizes, present/missing copies) but never reads them: content
++ * hashes stay with the content-hash job and remote copies stay 'pending' until reconciliation or a
++ * tier move verifies them (afterTierMove), as before. A row write() deleted needs nothing: the
++ * row-delete trigger marks its object deleted in the same statement (server/db/database.js).
*/
--function afterTierMove(vodId, verifiedProviders = []) {
-- try {
-- const r = sync('vod', vodId);
++function withObject(kind, ids, write) {
++ const fn = SYNC_BY_KIND[kind];
++ if (!fn) throw new Error(`withObject: unknown projection ${kind}`);
++ return _syncTx(() => {
++ const out = write();
++ for (const id of _rowIds(ids, out)) fn(id);
++ return out;
++ });
++}
++
++/**
++ * withObject() for a row write that records something already done outside the database (a
++ * recorder's ffmpeg is running): the row must say so whatever happens to its object. When the
++ * combined transaction fails, the row is written alone and the object re-projected after it (the
++ * pre-WS-G two steps; logged, and the drift report lists the row until a later write catches up).
++ */
++function withObjectOrRow(kind, ids, write) {
++ try { return withObject(kind, ids, write); } catch (err) {
++ console.warn(`[Objects] ${kind}: object not written with its row (${err.message}); writing the row alone`);
++ const out = write();
++ for (const id of _rowIds(ids, out)) safeSync(kind, id);
++ return out;
++ }
++}
++
++/**
++ * A tier move that verified a copy (upload + HEAD, copy + HEAD). write() flips the row's
++ * storage_provider / storage_key; the object is re-projected and the verified providers marked
++ * present in the same transaction. The bytes have already moved, so the flip is never lost to its
++ * object: when the combined transaction fails the row is flipped alone and re-projected after it
++ * (as withObjectOrRow). Without write(), the re-projection alone. The object part never throws.
++ */
++function afterTierMove(vodId, verifiedProviders = [], write = null) {
++ const reproject = () => {
++ const r = syncVod(vodId);
if (!r || !r.id) return;
for (const l of listLocations(r.id)) {
if (verifiedProviders.includes(l.provider)) setLocationState(l.id, { state: 'present' });
}
-- } catch (err) { console.warn(`[Objects] tier sync vod ${vodId}:`, err.message); }
++ };
++ let out;
++ if (write) {
++ try { return _syncTx(() => { const w = write(); reproject(); return w; }); } catch (err) {
++ console.warn(`[Objects] tier move vod ${vodId}: object not written with its row (${err.message}); flipping the row alone`);
++ out = write();
++ }
++ }
++ try { _syncTx(reproject); } catch (err) { console.warn(`[Objects] tier sync vod ${vodId}:`, err.message); }
++ return out;
}
// ── Holds ────────────────────────────────────────────────────
@@@ -611,7 +714,8 @@ module.exports = {
mimeFor, parseJson, parseLegacyRef, thumbFileFromUrl,
getObject, getObjectByLegacyRef, resolveObject, listLocations, listHolds, inheritedHolds, isHeld, isHeldRow,
createObject, updateObject, upsertLocation, setLocationState, setRelationship, setVariant, getVariant,
-- syncVod, syncClip, syncFile, syncPaste, sync, safeSync, afterTierMove,
++ vodProjection, clipProjection, fileProjection, pasteProjection, thumbnailProjection,
++ syncVod, syncClip, syncFile, syncPaste, sync, safeSync, withObject, withObjectOrRow, afterTierMove,
placeHold, releaseHold, holdPublic,
objectFilePath, softDelete, restore, purgeExpired, usedBytes,
legacyPublicUrl, objectPublic,
ddiff --git a/server/pastes/routes.js b/server/pastes/routes.js
iindex 5f042d0..df0903c 100644
---- a/server/pastes/routes.js
++++ b/server/pastes/routes.js
@@@ -372,7 +372,7 @@ router.delete('/admin/forks', tenantAuth(), (req, res) => {
try {
const forks = db.all('SELECT id, screenshot_path FROM pastes WHERE app_id = ? AND forked_from IS NOT NULL', [req.appId]);
for (const f of forks) removePasteScreenshot(f); // unlink screenshots so they don't leak on disk
-- db.run('DELETE FROM pastes WHERE app_id = ? AND forked_from IS NOT NULL', [req.appId]);
++ db.run('DELETE FROM pastes WHERE app_id = ? AND forked_from IS NOT NULL', [req.appId]); // objects: the row-delete trigger
res.json({ success: true, deleted: forks.length });
} catch (err) {
console.error('[Pastes] Delete forks error:', err.message);
@@@ -394,9 +394,9 @@ router.post('/bulk', tenantAuth(), (req, res) => {
// A held screenshot is skipped (its row and bytes stay), not failed halfway.
if (require('../objects/model').isHeldRow(paste)) { skipped++; continue; }
removePasteScreenshot(paste);
-- db.run('DELETE FROM pastes WHERE id = ?', [paste.id]);
++ db.run('DELETE FROM pastes WHERE id = ?', [paste.id]); // its object: the row-delete trigger
} else {
-- db.run('UPDATE pastes SET visibility = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', [action, paste.id]);
++ db.withObject('paste', paste.id, () => db.run('UPDATE pastes SET visibility = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', [action, paste.id]));
}
done++;
}
@@@ -436,9 +436,8 @@ router.post('/:slug/censor', tenantAuth(), (req, res) => {
try {
// Delete the old screenshot file, then swap in the censored one.
try { fs.unlinkSync(paste.screenshot_path); } catch { /* */ }
-- db.run('UPDATE pastes SET screenshot_path = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?',
-- [req.file.path, paste.id]);
-- require('../objects/model').safeSync('paste', paste.id);
++ db.withObject('paste', paste.id, () => db.run('UPDATE pastes SET screenshot_path = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?',
++ [req.file.path, paste.id]));
res.json({ paste: pastePublic(db.getPasteBySlug(paste.slug, req.appId)) });
} catch (err2) {
console.error('[Pastes] Censor error:', err2.message);
@@@ -525,15 +524,15 @@ router.post('/', tenantAuth({ allowUser: true }), screenshotUpload.single('scree
mime_type: req.file.mimetype,
});
-- db.run(
++ // The row and the screenshot's object in one transaction.
++ db.withObject('paste', (r) => r.lastInsertRowid, () => db.run(
`INSERT INTO pastes (app_id, slug, user_id, type, title, content, language, visibility, stream_id, screenshot_path, metadata, burn_after_read, is_nsfw, ip_address)
VALUES (?, ?, ?, 'screenshot', ?, ?, 'text', ?, ?, ?, ?, ?, ?, ?)`,
[req.appId, slug, userId, sanitizeTitle(body.title || 'Screenshot'),
body.description || body.content || '', vis, body.stream_id || null, req.file.path, metadata, burn, nsfw, actor.ip]
-- );
++ ));
const paste = db.getPasteBySlug(slug, req.appId);
-- require('../objects/model').safeSync('paste', paste.id);
return res.status(201).json({ id: paste.id, slug, url: `/p/${slug}`, paste: pastePublic(paste) });
}
@@@ -621,7 +620,8 @@ router.put('/:slug', tenantAuth({ allowUser: true }), (req, res) => {
updates.push('updated_at = CURRENT_TIMESTAMP');
params.push(paste.id);
-- db.run(`UPDATE pastes SET ${updates.join(', ')} WHERE id = ?`, params);
++ // A screenshot's title/visibility are its object's too: row and object in one transaction (text pastes have none).
++ db.withObject('paste', paste.id, () => db.run(`UPDATE pastes SET ${updates.join(', ')} WHERE id = ?`, params));
res.json({ paste: pastePublic(db.getPasteBySlug(paste.slug, req.appId)) });
} catch (err) {
console.error('[Pastes] Update error:', err.message);
@@@ -640,7 +640,7 @@ router.delete('/:slug', tenantAuth({ allowUser: true }), (req, res) => {
if (require('../objects/model').isHeldRow(paste)) return res.status(409).json({ error: 'Screenshot is under a retention hold', code: 'media.object.held' });
removePasteScreenshot(paste);
-- db.run('DELETE FROM pastes WHERE id = ?', [paste.id]);
++ db.run('DELETE FROM pastes WHERE id = ?', [paste.id]); // its object: the row-delete trigger
res.json({ success: true });
} catch (err) {
console.error('[Pastes] Delete error:', err.message);
ddiff --git a/server/thumbnails/routes.js b/server/thumbnails/routes.js
iindex 0758232..96eb8e6 100644
---- a/server/thumbnails/routes.js
++++ b/server/thumbnails/routes.js
@@@ -71,8 +71,7 @@ router.post('/:kind/:id', tenantAuth({ allowUser: true }), upload.single('thumbn
fs.writeFileSync(path.join(thumbService.THUMB_DIR, filename), buffer);
const url = `/t/${filename}`;
const table = kind === 'vod' ? 'vods' : 'clips';
-- db.run(`UPDATE ${table} SET thumbnail_url = ? WHERE id = ?`, [url, numId]);
-- require('../objects/model').safeSync(kind, numId);
++ db.withObject(kind, numId, () => db.run(`UPDATE ${table} SET thumbnail_url = ? WHERE id = ?`, [url, numId])); // row + objects together
return res.json({ url });
}
ddiff --git a/server/thumbnails/thumbnail-service.js b/server/thumbnails/thumbnail-service.js
iindex 4ae82ef..570732a 100644
---- a/server/thumbnails/thumbnail-service.js
++++ b/server/thumbnails/thumbnail-service.js
@@@ -194,10 +194,10 @@ async function generateVodThumbnail(vodId, filePath, opts = {}) {
const thumbUrl = await generateFromVideo(filePath, 'vod', vodId,
isRecording ? { liveEdge: true } : { seekPercent: 10 });
if (thumbUrl) {
-- // Clean up the previous thumbnail for this VOD
-- _removeOldThumb(db.get('SELECT thumbnail_url FROM vods WHERE id = ?', [vodId])?.thumbnail_url, thumbUrl);
-- db.run('UPDATE vods SET thumbnail_url = ? WHERE id = ?', [thumbUrl, vodId]);
-- require('../objects/model').safeSync('vod', vodId); // the thumbnail object follows the picture
++ const old = db.get('SELECT thumbnail_url FROM vods WHERE id = ?', [vodId])?.thumbnail_url;
++ // The row and its objects (the VOD's, and the thumbnail object that follows the picture) together.
++ db.withObject('vod', vodId, () => db.run('UPDATE vods SET thumbnail_url = ? WHERE id = ?', [thumbUrl, vodId]));
++ _removeOldThumb(old, thumbUrl); // the previous picture goes once nothing names it
}
return thumbUrl;
}
@@@ -206,9 +206,9 @@ async function generateVodThumbnail(vodId, filePath, opts = {}) {
async function generateClipThumbnail(clipId, filePath) {
const thumbUrl = await generateFromVideo(filePath, 'clip', clipId, { seekSeconds: 0.5 });
if (thumbUrl) {
-- _removeOldThumb(db.get('SELECT thumbnail_url FROM clips WHERE id = ?', [clipId])?.thumbnail_url, thumbUrl);
-- db.run('UPDATE clips SET thumbnail_url = ? WHERE id = ?', [thumbUrl, clipId]);
-- require('../objects/model').safeSync('clip', clipId);
++ const old = db.get('SELECT thumbnail_url FROM clips WHERE id = ?', [clipId])?.thumbnail_url;
++ db.withObject('clip', clipId, () => db.run('UPDATE clips SET thumbnail_url = ? WHERE id = ?', [thumbUrl, clipId]));
++ _removeOldThumb(old, thumbUrl);
}
return thumbUrl;
}
ddiff --git a/server/vod/clip-jobs.js b/server/vod/clip-jobs.js
iindex 55f53eb..0a1eeeb 100644
---- a/server/vod/clip-jobs.js
++++ b/server/vod/clip-jobs.js
@@@ -20,6 +20,7 @@ const config = require('../config');
const cutter = require('./clip-cutter');
const vodStorage = require('./vod-storage');
const { announce } = require('../webhooks');
++const objects = () => require('../objects/model');
const MAX_ATTEMPTS = 4;
const BACKOFF_MIN = [2, 10, 30, 90]; // minutes before attempt 2, 3, 4, …
@@@ -52,9 +53,13 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {
if (!clip) return { ok: false, error: 'Clip not found' };
if (!clip.vod_id) return { ok: false, error: 'Clip has no source VOD' };
const vod = db.get('SELECT * FROM vods WHERE id = ?', [clip.vod_id]);
-- if (!vod) { db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = NULL WHERE id = ?", ['Source VOD no longer exists', clipId]); return { ok: false, error: 'Source VOD no longer exists' }; }
++ if (!vod) {
++ objects().withObject('clip', clipId, () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = NULL WHERE id = ?", ['Source VOD no longer exists', clipId]));
++ return { ok: false, error: 'Source VOD no longer exists' };
++ }
const attempt = (Number(clip.cut_attempts) || 0) + 1;
-- db.run("UPDATE clips SET status = 'processing', cut_attempts = ?, cut_next_at = NULL WHERE id = ?", [attempt, clipId]);
++ // Row and object (back to uploading) together.
++ objects().withObject('clip', clipId, () => db.run("UPDATE clips SET status = 'processing', cut_attempts = ?, cut_next_at = NULL WHERE id = ?", [attempt, clipId]));
// Attempt 2+: bring a cloud-stored VOD home first when the disk can take it.
let source = await vodStorage.resolveMediaSource(vod);
@@@ -75,15 +80,14 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {
const duration = Math.max(1, (Number(clip.end_time) || 0) - startTime);
const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });
if (!cut.ok) return fail(clipId, clip, attempt, cut.error);
-- // Thumbnail first (the clip.ready payload carries it), then the ready transition and its event
-- // in one transaction (webhooks.announce), then the webhook.
++ // Thumbnail first (the clip.ready payload carries it), then the ready transition, the clip's
++ // object and its event in one transaction (webhooks.announce), then the webhook.
try { await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath); } catch { /* */ }
announce(clip.app_id, 'clip.ready', {
-- change: () => db.run("UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = 'ready', cut_error = NULL, cut_next_at = NULL WHERE id = ?",
-- [cut.filePath, cut.duration, startTime + cut.duration, clipId]),
++ change: () => objects().withObject('clip', clipId, () => db.run("UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = 'ready', cut_error = NULL, cut_next_at = NULL WHERE id = ?",
++ [cut.filePath, cut.duration, startTime + cut.duration, clipId])),
payload: () => _clipPublic(db.getClipById(clipId)),
});
-- require('../objects/model').safeSync('clip', clipId);
console.log(`[Clips] Clip ${clipId} ${reason} OK from vod ${clip.vod_id} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s, attempt ${attempt})`);
return { ok: true };
}
@@@ -97,10 +101,9 @@ function fail(clipId, clip, attempt, error) {
const mins = BACKOFF_MIN[Math.min(attempt - 1, BACKOFF_MIN.length - 1)];
const nextAt = more ? new Date(Date.now() + mins * 60000).toISOString().replace('T', ' ').slice(0, 19) : null;
announce(clip.app_id, 'clip.failed', {
-- change: () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = ? WHERE id = ?", [msg, nextAt, clipId]),
++ change: () => objects().withObject('clip', clipId, () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = ? WHERE id = ?", [msg, nextAt, clipId])),
payload: () => _clipPublic(db.getClipById(clipId)),
});
-- require('../objects/model').safeSync('clip', clipId);
console.warn(`[Clips] Clip ${clipId} attempt ${attempt}/${MAX_ATTEMPTS} failed: ${msg}${more ? ` — retry in ${mins} min` : ' — giving up'}`);
return { ok: false, error: msg, retry_at: nextAt };
}
@@@ -119,8 +122,15 @@ async function sweep() {
function start() {
if (_timer) return;
try { vodStorage.cleanupStaleDownloads(); } catch { /* */ }
-- // Anything left 'processing' by a crash/restart is really failed — queue it for a retry.
-- try { db.run("UPDATE clips SET status = 'failed', cut_error = COALESCE(cut_error, 'interrupted by a restart') WHERE status = 'processing'"); } catch { /* */ }
++ // Anything left 'processing' by a crash/restart is really failed — queue it for a retry. The rows
++ // and their objects (uploading -> failed) in one transaction.
++ try {
++ objects().withObject('clip', (ids) => ids, () => {
++ const ids = db.all("SELECT id FROM clips WHERE status = 'processing'").map(r => r.id);
++ if (ids.length) db.run(`UPDATE clips SET status = 'failed', cut_error = COALESCE(cut_error, 'interrupted by a restart') WHERE id IN (${ids.map(() => '?').join(',')})`, ids);
++ return ids;
++ });
++ } catch (err) { console.warn('[Clips] marking interrupted cuts failed:', err.message); }
setTimeout(() => sweep().catch(() => {}), 40 * 1000);
_timer = setInterval(() => sweep().catch(e => console.warn('[Clips] retry sweep:', e.message)), SWEEP_MS);
if (_timer.unref) _timer.unref();
ddiff --git a/server/vod/clips-routes.js b/server/vod/clips-routes.js
iindex fe90f73..3856999 100644
---- a/server/vod/clips-routes.js
++++ b/server/vod/clips-routes.js
@@@ -207,37 +207,35 @@ router.post('/', tenantAuth({ allowUser: true }), clipUpload.single('video'), as
end_time: endTime,
duration_seconds: duration,
is_public: visibility === 'public' ? 1 : 0,
++ visibility,
status: 'processing',
-- });
++ }); // the row and its object (uploading) in one transaction
const clipId = result.lastInsertRowid;
-- if (visibility !== 'public') db.setClipVisibility(clipId, visibility);
const appId = req.appId;
// Cut in the background; webhook on completion.
(async () => {
const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });
if (cut.ok) {
-- // Thumbnail first so the clip.ready payload carries it; then the ready transition
-- // and its event commit together (webhooks.announce) and the webhook goes out.
++ // Thumbnail first so the clip.ready payload carries it; then the ready transition,
++ // the clip's object and its event commit together (webhooks.announce) and the webhook goes out.
try {
await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath);
} catch { /* */ }
announce(appId, 'clip.ready', {
-- change: () => db.run('UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = ? WHERE id = ?',
-- [cut.filePath, cut.duration, startTime + cut.duration, 'ready', clipId]),
++ change: () => objects.withObject('clip', clipId, () => db.run('UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = ? WHERE id = ?',
++ [cut.filePath, cut.duration, startTime + cut.duration, 'ready', clipId])),
payload: () => clipPublic(db.getClipById(clipId)),
});
-- objects.safeSync('clip', clipId);
console.log(`[Clips] Clip ${clipId} cut from vod ${vodId} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s)`);
} else {
// First attempt failed: record why and let the retry sweeper take it from here
// (attempt 2 pulls a cloud-stored VOD back to local disk first).
const nextAt = new Date(Date.now() + 2 * 60000).toISOString().replace('T', ' ').slice(0, 19);
announce(appId, 'clip.failed', {
-- change: () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_attempts = 1, cut_next_at = ? WHERE id = ?", [String(cut.error || 'cut failed').slice(0, 500), nextAt, clipId]),
++ change: () => objects.withObject('clip', clipId, () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_attempts = 1, cut_next_at = ? WHERE id = ?", [String(cut.error || 'cut failed').slice(0, 500), nextAt, clipId])),
payload: () => clipPublic(db.getClipById(clipId)),
});
-- objects.safeSync('clip', clipId);
console.warn(`[Clips] Clip ${clipId} failed: ${cut.error} — auto-retry in 2 min`);
}
})().catch(err => console.error('[Clips] Background cut error:', err.message));
@@@ -289,10 +287,10 @@ async function _createUploadedClip(req, res) {
end_time: Number.isFinite(endTime) ? endTime : duration,
duration_seconds: duration,
is_public: visibility === 'public' ? 1 : 0,
++ visibility,
status: 'ready',
-- });
++ }); // the row and its object (ready, local copy present) in one transaction
const clipId = result.lastInsertRowid;
-- if (visibility !== 'public') db.setClipVisibility(clipId, visibility);
require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, clipPath)
.catch(err => console.warn(`[Clips] Thumbnail failed for clip ${clipId}:`, err.message));
@@@ -372,7 +370,7 @@ router.post('/:id/recut', tenantAuth(), async (req, res) => {
if (!clip || clip.app_id !== req.appId) return res.status(404).json({ error: 'Clip not found' });
if (clip.status === 'processing') return res.status(409).json({ error: 'Clip is already being cut' });
if (!clip.vod_id) return res.status(422).json({ error: 'Clip has no source VOD to re-cut from' });
-- // A manual retry resets the attempt counter so it gets the full ladder again.
++ // A manual retry resets the attempt counter so it gets the full ladder again (not projected: no object change).
db.run("UPDATE clips SET cut_attempts = 0 WHERE id = ?", [clipId]);
require('./clip-jobs').recutClip(clipId, { reason: 're-cut' }).catch(err => console.error('[Clips] Background re-cut error:', err.message));
res.status(202).json({ id: clipId, status: 'processing' });
@@@ -390,13 +388,12 @@ router.put('/:id', tenantAuth(), (req, res) => {
if (title !== undefined) {
const t = sanitizeClipTitle(title, '');
if (!t) return res.status(400).json({ error: 'Title must be 1-200 characters' });
-- db.run('UPDATE clips SET title = ? WHERE id = ?', [t, clip.id]);
-- if (visibility === undefined) objects.safeSync('clip', clip.id);
++ objects.withObject('clip', clip.id, () => db.run('UPDATE clips SET title = ? WHERE id = ?', [t, clip.id]));
}
// Only the app itself (its automation's own records) may say a clip was machine-made.
if (autoGen !== undefined && flag(autoGen) !== null) {
if (req.authType !== 'app') return res.status(403).json({ error: 'auto_generated is set by the app only' });
-- db.run('UPDATE clips SET auto_generated = ? WHERE id = ?', [flag(autoGen), clip.id]);
++ objects.withObject('clip', clip.id, () => db.run('UPDATE clips SET auto_generated = ? WHERE id = ?', [flag(autoGen), clip.id]));
}
if (visibility !== undefined) db.setClipVisibility(clip.id, visibility);
res.json({ clip: clipPublic(db.getClipById(clip.id, req.appId), { readiness: true }) });
@@@ -422,7 +419,7 @@ router.delete('/:id', tenantAuth(), (req, res) => {
console.warn(`[Clips] Remote object cleanup failed for clip ${clip.id}:`, err.message));
}
-- db.run('DELETE FROM clips WHERE id = ?', [clip.id]);
++ db.run('DELETE FROM clips WHERE id = ?', [clip.id]); // its object is marked deleted by the row-delete trigger, same statement
res.json({ message: 'Clip deleted' });
} catch (err) {
res.status(500).json({ error: 'Failed to delete clip' });
ddiff --git a/server/vod/duration-reconcile.js b/server/vod/duration-reconcile.js
iindex e736437..259e822 100644
---- a/server/vod/duration-reconcile.js
++++ b/server/vod/duration-reconcile.js
@@@ -120,11 +120,10 @@ function writeReport(report, file = null) {
/** Write one repair (only if the row still has the value we read). Returns true when it changed. */
function applyRepair(row) {
-- const changed = db.run(`UPDATE vods SET duration_seconds = ?, probe_duration_seconds = ?, duration_source = ?
++ // The row and its object in one transaction (the object only when the row changed).
++ return db.withObject('vod', (r) => (r.changes > 0 ? row.vod_id : null), () => db.run(`UPDATE vods SET duration_seconds = ?, probe_duration_seconds = ?, duration_source = ?
WHERE id = ? AND COALESCE(duration_seconds, 0) = ? AND COALESCE(duration_source, '') = ? AND COALESCE(is_recording, 0) = 0`,
-- [Math.round(row.measured), row.measured, row.measured_source, row.vod_id, row.stored, row.stored_source || '']).changes > 0;
-- if (changed) require('../objects/model').safeSync('vod', row.vod_id);
-- return changed;
++ [Math.round(row.measured), row.measured, row.measured_source, row.vod_id, row.stored, row.stored_source || ''])).changes > 0;
}
/** Undo repairs listed in a report (only rows that still hold what the repair wrote). */
@@@ -135,9 +134,8 @@ function rollback(rows, { apply = false } = {}) {
const cur = db.get('SELECT duration_seconds, duration_source FROM vods WHERE id = ?', [r.vod_id]);
if (!cur || Number(cur.duration_seconds) !== Math.round(r.measured) || (cur.duration_source || '') !== (r.measured_source || '')) { out.changed_since++; continue; }
if (!apply) { out.would_restore++; continue; }
-- db.run('UPDATE vods SET duration_seconds = ?, probe_duration_seconds = ?, duration_source = ? WHERE id = ?',
-- [r.stored, r.stored_probe ?? r.stored, r.stored_source || null, r.vod_id]);
-- require('../objects/model').safeSync('vod', r.vod_id);
++ db.withObject('vod', r.vod_id, () => db.run('UPDATE vods SET duration_seconds = ?, probe_duration_seconds = ?, duration_source = ? WHERE id = ?',
++ [r.stored, r.stored_probe ?? r.stored, r.stored_source || null, r.vod_id]));
out.restored++;
}
return out;
ddiff --git a/server/vod/finalize.js b/server/vod/finalize.js
iindex 3b2c816..32d1150 100644
---- a/server/vod/finalize.js
++++ b/server/vod/finalize.js
@@@ -20,6 +20,7 @@ const db = require('../db/database');
const config = require('../config');
const tools = require('./media-tools');
const { announce } = require('../webhooks');
++const objects = () => require('../objects/model');
/** Absolute-ize a stored Media-relative URL (/t/…, /api/thumbnails/…). */
function _absUrl(u) {
@@@ -35,8 +36,8 @@ function isFinalizing(vodId) {
}
// Public shape of a vod row for API responses + webhooks. { readiness: true } (API responses) adds the
--// object's readiness levels (server/objects/readiness.js); webhook payloads leave them out, because a
--// finalize announces before it re-projects the object.
++// object's readiness levels (server/objects/readiness.js); webhook payloads leave them out (their shape
++// predates the object model and stays as it is).
function vodPublic(vod, { readiness = false } = {}) {
if (!vod) return null;
const out = {
@@@ -110,13 +111,14 @@ function rebuildWebmFromMaster(masterPath, webmPath) {
}
/**
-- * The state change for a VOD outcome and its event/webhook, committed together (webhooks.announce).
-- * `row()` gives the row the payload describes (read inside the transaction). Returns the payload,
-- * or null when the transaction failed (then nothing changed and nothing was announced).
++ * The state change for a VOD outcome, the VOD's object and the event/webhook, committed together
++ * (webhooks.announce around objects.withObject). `row()` gives the row the payload describes (read
++ * inside the transaction). Returns the payload, or null when the transaction failed (then nothing
++ * changed and nothing was announced). A deleted row's object is marked deleted by its trigger.
*/
function _commitVodOutcome(vod, event, change, row = () => db.getVodById(vod.id)) {
try {
-- return announce(vod.app_id, event, { change, payload: () => vodPublic(row()) }).data;
++ return announce(vod.app_id, event, { change: () => objects().withObject('vod', vod.id, change), payload: () => vodPublic(row()) }).data;
} catch (err) {
console.error(`[VOD] vod ${vod.id}: ${event} not committed:`, err.message);
return null;
@@@ -156,8 +158,11 @@ async function finalizeVod(vodId, opts = {}) {
throw err;
} finally {
_finalizing.delete(vodId);
-- // Every outcome (ready, quarantined, failed) re-projects the object; a deleted row was already marked by trigger.
-- require('../objects/model').safeSync('vod', vodId);
++ // Every outcome (ready, quarantined, failed) now commits with its object (withObject); a deleted
++ // row's object is marked by its trigger. This follow-up only re-reads the disk after a finalize
++ // that threw part-way (a merge or remux rewrote the file, no row write followed). C-75 leftover:
++ // it goes with the boot backfill (server/index.js) once the drift report shows zero drift.
++ objects().safeSync('vod', vodId);
}
}
@@@ -213,7 +218,7 @@ async function _doFinalize(vodId, opts) {
[status, JSON.stringify([status, 'retention_hold']), vodId]));
return null;
};
-- const held = require('../objects/model').isHeldRow(vod);
++ const held = objects().isHeldRow(vod);
if (!filePath || !fs.existsSync(filePath)) {
if (held) return keepHeld('missing_file');
@@@ -334,8 +339,8 @@ async function _doFinalize(vodId, opts) {
// The file vanished between the probe and here. Record it (no wall-clock duration left
// behind by the live updates) and leave it for review.
console.error(`[VOD] Failed to stat finalized VOD ${vodId}:`, err.message);
-- db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review',
-- health_issues_json = ?, last_health_scan_at = datetime('now') WHERE id = ?`, [JSON.stringify(['stat_failed', ...issues]), vodId]);
++ objects().withObject('vod', vodId, () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review',
++ health_issues_json = ?, last_health_scan_at = datetime('now') WHERE id = ?`, [JSON.stringify(['stat_failed', ...issues]), vodId]));
_queueRetry(vodId, 'stat_failed', opts);
return null;
}
@@@ -348,7 +353,7 @@ async function _doFinalize(vodId, opts) {
if (webmComplete) {
fs.unlinkSync(masterPath);
console.log(`[VOD] Removed master archive for vod ${vodId} (${path.basename(masterPath)})`);
-- db.run('UPDATE vods SET master_file_path = NULL WHERE id = ?', [vodId]);
++ db.run('UPDATE vods SET master_file_path = NULL WHERE id = ?', [vodId]); // not projected: no object change
} else {
console.warn(`[VOD] vod ${vodId}: KEEPING master (webm ${durationSeconds}s still < master ${masterDur}s)`);
}
@@@ -373,11 +378,11 @@ async function _doFinalize(vodId, opts) {
// recording is back to the visibility its owner chose.
const lift = _finalizeQuarantined(db.getVodById(vodId) || vod);
-- // The ready transition and the vod.ready event commit together (then the webhook goes out).
-- // A failed commit throws, as the plain UPDATE did before.
++ // The ready transition, the VOD's object and the vod.ready event commit together (then the
++ // webhook goes out). A failed commit throws, as the plain UPDATE did before.
announce(vod.app_id, 'vod.ready', {
-- change: () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = ?, duration_source = ?, file_size = ?, probe_duration_seconds = ?, probe_format_json = ?, health_status = ?, health_issues_json = ?${lift ? ", quarantined_at = NULL, is_public = CASE WHEN COALESCE(visibility, 'public') = 'public' THEN 1 ELSE 0 END" : ''} WHERE id = ?`,
-- [stored, durationSource, stat.size, measured, probeFormatJson, 'ok', JSON.stringify(issues), vodId]),
++ change: () => objects().withObject('vod', vodId, () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = ?, duration_source = ?, file_size = ?, probe_duration_seconds = ?, probe_format_json = ?, health_status = ?, health_issues_json = ?${lift ? ", quarantined_at = NULL, is_public = CASE WHEN COALESCE(visibility, 'public') = 'public' THEN 1 ELSE 0 END" : ''} WHERE id = ?`,
++ [stored, durationSource, stat.size, measured, probeFormatJson, 'ok', JSON.stringify(issues), vodId])),
payload: () => vodPublic(db.getVodById(vodId)),
});
console.log(`[VOD] Finalized: vod ${vodId}, ${stored}s (${durationSource}), ${(stat.size / 1024 / 1024).toFixed(1)}MB`);
ddiff --git a/server/vod/health-job.js b/server/vod/health-job.js
iindex 889ead1..2f7dcf4 100644
---- a/server/vod/health-job.js
++++ b/server/vod/health-job.js
@@@ -62,7 +62,7 @@ async function _scanOne(vod, { deep }) {
if (master && fs.existsSync(master)) {
const freedMb = (fs.statSync(master).size / 1024 / 1024).toFixed(0);
fs.unlinkSync(master);
-- if (vod.master_file_path) { try { db.run('UPDATE vods SET master_file_path = NULL WHERE id = ?', [vod.id]); } catch { /* */ } }
++ if (vod.master_file_path) { try { db.run('UPDATE vods SET master_file_path = NULL WHERE id = ?', [vod.id]); } catch { /* */ } } // not projected
console.log(`[VOD-Health] Reclaimed orphaned master for vod ${vod.id} (${freedMb}MB)`);
}
} catch { /* */ }
@@@ -134,7 +134,7 @@ function _hardDeleteVod(vod) {
try { if (fs.existsSync(p)) fs.unlinkSync(p); } catch { /* */ }
}
}
-- db.run('DELETE FROM vods WHERE id = ?', [vod.id]);
++ db.run('DELETE FROM vods WHERE id = ?', [vod.id]); // its object is marked deleted by the row-delete trigger
return true;
} catch (e) {
console.warn(`[VOD-Health] cleanup delete failed for vod ${vod.id}:`, e.message);
@@@ -155,8 +155,11 @@ async function _cleanupPass(deep) {
const fresh = db.getVodById(vod.id) || vod;
const rescan = await scanner.scanVod(fresh, { decode: true, repairDuration: true, quarantineBad: false });
if (!['corrupt', 'zero_byte', 'missing_file'].includes(rescan.status)) {
-- db.run("UPDATE vods SET quarantined_at = NULL, health_status = 'ok' WHERE id = ?", [vod.id]);
-- console.log(`[VOD-Health] Cleanup recovered vod ${vod.id} from master — un-quarantined`);
++ // Row and object together; a failed write keeps the recovered VOD (never falls through to the delete).
++ try {
++ require('../objects/model').withObject('vod', vod.id, () => db.run("UPDATE vods SET quarantined_at = NULL, health_status = 'ok' WHERE id = ?", [vod.id]));
++ console.log(`[VOD-Health] Cleanup recovered vod ${vod.id} from master — un-quarantined`);
++ } catch (e) { console.warn(`[VOD-Health] vod ${vod.id} recovered from master but not un-quarantined (next pass retries):`, e.message); }
continue;
}
}
@@@ -254,8 +257,8 @@ function sweepJunkVods() {
db.run('DELETE FROM vods WHERE id = ?', [v.id]);
deleted++;
} else {
-- db.run(`UPDATE vods SET health_status = 'needs_review', health_issues_json = ?, quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
-- [JSON.stringify(['short_duration']), v.id]);
++ objects.withObject('vod', v.id, () => db.run(`UPDATE vods SET health_status = 'needs_review', health_issues_json = ?, quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
++ [JSON.stringify(['short_duration']), v.id]));
quarantined++;
}
}
ddiff --git a/server/vod/recorder.js b/server/vod/recorder.js
iindex 08c6586..9fa8e28 100644
---- a/server/vod/recorder.js
++++ b/server/vod/recorder.js
@@@ -186,7 +186,7 @@ class StreamRecorder {
_cleanupFailedVod(vodId, filePath) {
if (!filePath || !fs.existsSync(filePath)) {
try {
-- db.run('DELETE FROM vods WHERE id = ?', [vodId]);
++ db.run('DELETE FROM vods WHERE id = ?', [vodId]); // its object is marked deleted by the row-delete trigger
console.log(`[VOD] Deleted stale failed VOD ${vodId}`);
} catch (err) {
console.warn(`[VOD] Failed to delete stale VOD ${vodId}:`, err.message);
@@@ -194,10 +194,10 @@ class StreamRecorder {
return;
}
try {
-- db.run(
++ require('../objects/model').withObject('vod', vodId, () => db.run(
'UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, quarantined_at = datetime(\'now\'), is_public = 0 WHERE id = ?',
['corrupt', JSON.stringify(['failed_recording_start']), vodId]
-- );
++ ));
console.log(`[VOD] Marked failed VOD ${vodId} as corrupt`);
} catch (err) {
console.warn(`[VOD] Failed to mark VOD ${vodId} as corrupt:`, err.message);
@@@ -238,9 +238,13 @@ class StreamRecorder {
}, 30000);
this.activeRecordings.set(vod.id, recording);
-- db.run('UPDATE vods SET is_recording = 1, file_path = ? WHERE id = ?', [filePath, vod.id]);
-- if (masterPath) db.run('UPDATE vods SET master_file_path = ? WHERE id = ?', [masterPath, vod.id]);
-- require('../objects/model').safeSync('vod', vod.id);
++ // The row and its object in one transaction. ffmpeg is already running, so the row must say
++ // so even if the object cannot be written with it: withObjectOrRow then writes the row alone
++ // and re-projects after it (the finalize that follows re-projects every outcome as well).
++ require('../objects/model').withObjectOrRow('vod', vod.id, () => {
++ db.run('UPDATE vods SET is_recording = 1, file_path = ? WHERE id = ?', [filePath, vod.id]);
++ if (masterPath) db.run('UPDATE vods SET master_file_path = ? WHERE id = ?', [masterPath, vod.id]);
++ });
return recording;
}
@@@ -282,10 +286,12 @@ class StreamRecorder {
require('./finalize').finalizeVod(vodId, { startTimeMs: rec?.startTime }).catch(() => {
// Never bare-mark a failed recording as ready — that published
// 0:00 ghosts. Quarantine it out of listings instead.
-- // The live duration updates were wall-clock estimates: none of it is kept.
-- db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review',
-- health_issues_json = ?, quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
-- [JSON.stringify(['finalize_failed']), vodId]);
++ // The live duration updates were wall-clock estimates: none of it is kept. Row and object together.
++ try {
++ require('../objects/model').withObject('vod', vodId, () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review',
++ health_issues_json = ?, quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
++ [JSON.stringify(['finalize_failed']), vodId]));
++ } catch (e) { console.warn(`[VOD] Failed to quarantine vod ${vodId} after a failed finalize:`, e.message); }
});
if (filePath && !fs.existsSync(filePath)) this._cleanupFailedVod(vodId, filePath);
});
@@@ -312,8 +318,9 @@ class StreamRecorder {
const elapsed = Math.round((Date.now() - rec.startTime) / 1000);
try {
const stat = fs.statSync(rec.filePath);
-- db.run('UPDATE vods SET duration_seconds = ?, file_size = ? WHERE id = ?',
-- [elapsed, stat.size, rec.vodId]);
++ // Progress estimate: the row and its object's size/duration together, or neither (next pass retries).
++ require('../objects/model').withObject('vod', rec.vodId, () => db.run('UPDATE vods SET duration_seconds = ?, file_size = ? WHERE id = ?',
++ [elapsed, stat.size, rec.vodId]));
} catch {}
tools.remuxForLiveSeeking(rec.filePath).catch(() => {});
ddiff --git a/server/vod/routes.js b/server/vod/routes.js
iindex dc2a684..51b4ba9 100644
---- a/server/vod/routes.js
++++ b/server/vod/routes.js
@@@ -103,11 +103,10 @@ router.post('/', tenantAuth(), (req, res) => {
user_id: user_id != null ? user_id : null,
title: (title || 'Recording').toString().slice(0, 300),
meta,
-- });
++ visibility: visibility || null,
++ clips_only: !!clips_only,
++ }); // the row (visibility and clips_only included) and its object in one transaction
const id = result.lastInsertRowid;
-- if (clips_only) db.run('UPDATE vods SET clips_only = 1 WHERE id = ?', [id]);
-- if (visibility) db.setVodVisibility(id, visibility);
-- else if (!clips_only) objects.safeSync('vod', id);
res.status(201).json({ id });
} catch (err) {
console.error('[VOD] Create error:', err.message);
@@@ -188,8 +187,8 @@ router.post('/:id/chunks', tenantAuth({ allowUser: true }), chunkUpload.single('
fs.copyFileSync(req.file.path, filePath);
tools.cleanupTempFile(req.file.path);
-- db.run('UPDATE vods SET file_path = ?, file_size = ?, is_recording = 1 WHERE id = ?',
-- [filePath, fs.statSync(filePath).size, vod.id]);
++ objects.withObject('vod', vod.id, () => db.run('UPDATE vods SET file_path = ?, file_size = ?, is_recording = 1 WHERE id = ?',
++ [filePath, fs.statSync(filePath).size, vod.id]));
rec = { filePath, startTime: Date.now(), chunkCount: 1, currentSegmentId: segmentId, currentSegmentPath: filePath };
activeChunkUploads.set(vod.id, rec);
@@@ -222,7 +221,7 @@ router.post('/:id/chunks', tenantAuth({ allowUser: true }), chunkUpload.single('
const size = tools.getFileSizeSafe(rec.filePath)
+ (rec.currentSegmentPath && rec.currentSegmentPath !== rec.filePath ? tools.getFileSizeSafe(rec.currentSegmentPath) : 0);
const elapsed = Math.round((Date.now() - rec.startTime) / 1000);
-- db.run('UPDATE vods SET file_size = ?, duration_seconds = ? WHERE id = ?', [size, elapsed, vod.id]);
++ objects.withObject('vod', vod.id, () => db.run('UPDATE vods SET file_size = ?, duration_seconds = ? WHERE id = ?', [size, elapsed, vod.id]));
// Seekable sidecar for live DVR (every 2 chunks ≈ ~60s)
if (rec.chunkCount >= 2 && rec.chunkCount % 2 === 0) {
@@@ -327,7 +326,7 @@ router.get('/:id', tenantAuth({ allowUser: true }), async (req, res) => {
const duration = await tools.probeVodDuration(vod.file_path);
if (duration > 0) {
const fileSize = tools.getFileSizeSafe(vod.file_path);
-- db.run("UPDATE vods SET duration_seconds = ?, file_size = ?, duration_source = 'probe' WHERE id = ?", [duration, fileSize, vod.id]);
++ objects.withObject('vod', vod.id, () => db.run("UPDATE vods SET duration_seconds = ?, file_size = ?, duration_source = 'probe' WHERE id = ?", [duration, fileSize, vod.id]));
vod.duration_seconds = duration;
vod.file_size = fileSize;
}
@@@ -352,12 +351,14 @@ router.put('/:id', tenantAuth(), (req, res) => {
const params = [];
if (title !== undefined) { updates.push('title = ?'); params.push(String(title).slice(0, 300)); }
if (description !== undefined) { updates.push('description = ?'); params.push(String(description)); }
-- if (updates.length) {
-- params.push(vod.id);
-- db.run(`UPDATE vods SET ${updates.join(', ')} WHERE id = ?`, params);
-- }
-- if (visibility !== undefined) db.setVodVisibility(vod.id, visibility);
-- else if (updates.length) objects.safeSync('vod', vod.id);
++ // Title, description and visibility, and the object, in one transaction.
++ objects.withObject('vod', vod.id, () => {
++ if (updates.length) {
++ params.push(vod.id);
++ db.run(`UPDATE vods SET ${updates.join(', ')} WHERE id = ?`, params);
++ }
++ if (visibility !== undefined) db.setVodVisibility(vod.id, visibility);
++ });
res.json({ vod: vodPublic(db.getVodById(vod.id, req.appId), { readiness: true }) });
} catch (err) {
console.error('[VOD] Update error:', err.message);
@@@ -384,7 +385,7 @@ router.delete('/:id', tenantAuth(), (req, res) => {
try { if (vod.master_file_path && fs.existsSync(vod.master_file_path)) fs.unlinkSync(vod.master_file_path); } catch { /* */ }
}
-- db.run('DELETE FROM vods WHERE id = ?', [vod.id]);
++ db.run('DELETE FROM vods WHERE id = ?', [vod.id]); // its object is marked deleted by the row-delete trigger, same statement
res.json({ message: 'VOD deleted' });
} catch (err) {
console.error('[VOD] Delete error:', err.message);
ddiff --git a/server/vod/vod-storage.js b/server/vod/vod-storage.js
iindex 9381ffd..186c03c 100644
---- a/server/vod/vod-storage.js
++++ b/server/vod/vod-storage.js
@@@ -232,11 +232,15 @@ function endpointFor(name) {
// Object model (Wave 4): holds freeze an object's placement (moveToCold, moveToHot, promoteToR2 and
// demoteFromR2 refuse a held VOD with { ok: false, held: true }; deleteVodObjects keeps a held object's
--// bytes, a clip's too while its source VOD is held), and every tier move re-projects the VOD's
--// media_locations. Lazy — the model requires this module.
++// bytes, a clip's too while its source VOD is held), and every tier move writes the row's new
++// placement (`write`) and re-projects the VOD's media_locations in one transaction (WS-G task 1;
++// afterTierMove). Lazy — the model requires this module.
const objects = () => require('../objects/model');
function _held(vod) { return objects().isHeldRow(vod); }
--function _moved(vodId, verified) { objects().afterTierMove(vodId, verified); }
++function _moved(vodId, verified, write = null) { return objects().afterTierMove(vodId, verified, write); }
++// After a move removed the local copy (the row already names the remote one): the object's locations
++// follow the disk. File I/O after the commit, so a follow-up rather than part of the write.
++function _localRemoved(vodId) { objects().safeSync('vod', vodId); }
function keyForVod(vod) {
if (vod.storage_key) return vod.storage_key;
@@@ -498,8 +502,7 @@ async function moveToCold(vodId) {
if (!fs.existsSync(local)) {
// Not local — maybe already offloaded
if (await headObject('b2', key)) {
-- db.run("UPDATE vods SET storage_provider = 'b2', storage_key = ? WHERE id = ?", [key, vodId]);
-- _moved(vodId, ['b2']);
++ _moved(vodId, ['b2'], () => db.run("UPDATE vods SET storage_provider = 'b2', storage_key = ? WHERE id = ?", [key, vodId]));
return { ok: true, already: true };
}
// Neither here nor in B2: nothing to offload, ever. Quarantine the row so the
@@@ -514,12 +517,12 @@ async function moveToCold(vodId) {
await uploadFile('b2', key, local);
}
-- db.run("UPDATE vods SET storage_provider = 'b2', storage_key = ? WHERE id = ?", [key, vodId]);
++ _moved(vodId, ['b2'], () => db.run("UPDATE vods SET storage_provider = 'b2', storage_key = ? WHERE id = ?", [key, vodId]));
try { fs.unlinkSync(local); } catch (err) {
console.error(`[VodStorage] Uploaded but failed to remove local file for VOD ${vodId}:`, err.message);
}
cleanupSidecar(local);
-- _moved(vodId, ['b2']);
++ _localRemoved(vodId);
console.log(`[VodStorage] VOD ${vodId} offloaded to B2: ${key} (${(localSize / 1048576).toFixed(1)} MB)`);
return { ok: true, bytes: localSize };
@@@ -562,8 +565,7 @@ async function _moveToHot(vodId) {
const local = localPathForVod(vod);
if (fs.existsSync(local)) {
-- db.run("UPDATE vods SET storage_provider = 'local' WHERE id = ?", [vodId]);
-- _moved(vodId, []);
++ _moved(vodId, [], () => db.run("UPDATE vods SET storage_provider = 'local' WHERE id = ?", [vodId]));
return { ok: true, already: true };
}
@@@ -612,8 +614,7 @@ async function _moveToHot(vodId) {
// Restoring to local keeps the B2 canonical copy; drop any R2 copy.
if (providerOf(vod) === 'r2' && providerConfigured('r2')) await deleteObject('r2', key);
-- db.run("UPDATE vods SET storage_provider = 'local' WHERE id = ?", [vodId]);
-- _moved(vodId, []);
++ _moved(vodId, [], () => db.run("UPDATE vods SET storage_provider = 'local' WHERE id = ?", [vodId]));
console.log(`[VodStorage] VOD ${vodId} restored to local (${(head.size / 1048576).toFixed(1)} MB)`);
return { ok: true, bytes: head.size };
} catch (err) {
@@@ -660,14 +661,14 @@ async function _promoteToR2(vodId) {
else await copyBetweenProviders('b2', 'r2', key);
}
-- db.run("UPDATE vods SET storage_provider = 'r2', storage_key = ? WHERE id = ?", [key, vodId]);
-- _moved(vodId, ['b2', 'r2']);
++ _moved(vodId, ['b2', 'r2'], () => db.run("UPDATE vods SET storage_provider = 'r2', storage_key = ? WHERE id = ?", [key, vodId]));
// Popular VODs live in R2+B2; free the local copy
let freed = 0;
if (fs.existsSync(local)) {
freed = fs.statSync(local).size;
try { fs.unlinkSync(local); cleanupSidecar(local); } catch { freed = 0; }
++ if (freed) _localRemoved(vodId);
}
console.log(`[VodStorage] VOD ${vodId} promoted to R2: ${key}`);
@@@ -701,8 +702,7 @@ async function _demoteFromR2(vodId) {
if (!copied) return { ok: false, error: 'No B2 canonical and copy-back failed' };
}
await deleteObject('r2', key);
-- db.run("UPDATE vods SET storage_provider = 'b2' WHERE id = ?", [vodId]);
-- _moved(vodId, ['b2']);
++ _moved(vodId, ['b2'], () => db.run("UPDATE vods SET storage_provider = 'b2' WHERE id = ?", [vodId]));
console.log(`[VodStorage] VOD ${vodId} demoted from R2 to B2`);
return { ok: true };
} catch (err) {
@@@ -783,13 +783,13 @@ const OFFLOADABLE_WHERE = `
function quarantineMissing(vodId, reason) {
try {
-- db.run(`UPDATE vods
++ objects().withObject('vod', vodId, () => db.run(`UPDATE vods
SET health_status = 'missing_file',
health_issues_json = ?,
last_health_scan_at = datetime('now'),
quarantined_at = COALESCE(quarantined_at, datetime('now')),
is_public = 0
-- WHERE id = ?`, [JSON.stringify(['missing_file', reason]), vodId]);
++ WHERE id = ?`, [JSON.stringify(['missing_file', reason]), vodId]));
} catch (err) {
console.warn(`[VodStorage] Could not quarantine VOD ${vodId}:`, err.message);
}
@@@ -802,19 +802,27 @@ function quarantineMissing(vodId, reason) {
*/
function reconcileGhosts() {
try {
-- const res = db.run(`UPDATE vods
-- SET health_status = 'missing_file',
-- health_issues_json = ?,
-- last_health_scan_at = datetime('now'),
-- quarantined_at = COALESCE(quarantined_at, datetime('now')),
-- is_public = 0
++ // The rows and their objects (-> failed) in one transaction.
++ const ids = objects().withObject('vod', (found) => found, () => {
++ const found = db.all(`SELECT id FROM vods
WHERE COALESCE(storage_provider, 'local') = 'local'
AND COALESCE(is_recording, 0) = 0
AND file_path IS NULL
AND created_at <= datetime('now', '-1 hour')
-- AND COALESCE(health_status, 'ok') NOT IN ('missing_file', 'zero_byte')`,
-- [JSON.stringify(['missing_file', 'recording never produced a file'])]);
-- return (res && typeof res.changes === 'number') ? res.changes : 0;
++ AND COALESCE(health_status, 'ok') NOT IN ('missing_file', 'zero_byte')`).map(r => r.id);
++ if (found.length) {
++ db.run(`UPDATE vods
++ SET health_status = 'missing_file',
++ health_issues_json = ?,
++ last_health_scan_at = datetime('now'),
++ quarantined_at = COALESCE(quarantined_at, datetime('now')),
++ is_public = 0
++ WHERE id IN (${found.map(() => '?').join(',')})`,
++ [JSON.stringify(['missing_file', 'recording never produced a file']), ...found]);
++ }
++ return found;
++ });
++ return ids.length;
} catch (err) {
console.warn('[VodStorage] Ghost reconcile failed:', err.message);
return 0;
@@@ -956,8 +964,7 @@ async function runSweep() {
if (!fs.existsSync(localPathForVod(vod))) {
const key = keyForVod(vod);
if (await headObject('b2', key).catch(() => null)) {
-- db.run("UPDATE vods SET storage_provider = 'b2', storage_key = ? WHERE id = ?", [key, vod.id]);
-- _moved(vod.id, ['b2']);
++ _moved(vod.id, ['b2'], () => db.run("UPDATE vods SET storage_provider = 'b2', storage_key = ? WHERE id = ?", [key, vod.id]));
} else {
quarantineMissing(vod.id, `file not found at ${localPathForVod(vod)}`);
quarantined++;
@@@ -1148,8 +1155,7 @@ async function migrateLegacy() {
try {
const head = await headObject('b2', key);
if (head) {
-- db.run("UPDATE vods SET storage_provider = 'b2', storage_key = ? WHERE id = ?", [key, vod.id]);
-- _moved(vod.id, ['b2']);
++ _moved(vod.id, ['b2'], () => db.run("UPDATE vods SET storage_provider = 'b2', storage_key = ? WHERE id = ?", [key, vod.id]));
flipped++;
} else if (fs.existsSync(localPathForVod(vod))) {
restoredLocal++; // still local, sweep will re-offload
@@@ -1419,6 +1425,8 @@ module.exports = {
deleteLegacyPasteScreenshot,
presignGet,
runSweep,
++ reconcileGhosts,
++ quarantineMissing,
checkProviders,
migrateLegacy,
start,
ddiff --git a/test/object-first-writes.test.js b/test/object-first-writes.test.js
nnew file mode 100644
iindex 0000000..a1fb979
---- /dev/null
++++ b/test/object-first-writes.test.js
@@@ -0,0 +1,319 @@
++'use strict';
++// Object-first writes (WS-G task 1, retiring compatibility shim C-75): every write to an inherited
++// vods/clips/files/pastes row makes or updates its media_object in the SAME SQLite transaction
++// (objects/model.js withObject). For each main write path the object is there, linked and agreeing
++// with the row, as soon as the write returns; when the object cannot be written, the row write
++// rolls back with it (neither row); and the drift report (server/objects/drift.js,
++// scripts/object-drift-report.js) writes nothing and finds drift planted behind the model's back.
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const http = require('http');
++const crypto = require('crypto');
++const { execFileSync } = require('child_process');
++const express = require('express');
++const sharp = require('sharp');
++
++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-object-first-'));
++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
++Object.assign(process.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'), MEDIA_PUBLIC_URL: 'https://media.test',
++});
++
++const db = require('../server/db/database');
++require('../server/views/service').ensureSchema();
++const model = require('../server/objects/model');
++const drift = require('../server/objects/drift');
++const tools = require('../server/vod/media-tools');
++const thumbService = require('../server/thumbnails/thumbnail-service');
++const cutter = require('../server/vod/clip-cutter');
++const vodStorage = require('../server/vod/vod-storage');
++const clipJobs = require('../server/vod/clip-jobs');
++const { finalizeVod } = require('../server/vod/finalize');
++
++db.upsertApp({ app_id: 'live', api_key: 'live-key-object-first' });
++const raw = db.getDb();
++const q = { all: (sql, p = []) => raw.prepare(sql).all(...p), get: (sql, p = []) => raw.prepare(sql).get(...p) };
++const count = (table) => raw.prepare(`SELECT COUNT(*) AS n FROM ${table}`).get().n;
++
++// No ffmpeg in these paths: the thumbnail generators and the probes answer as a real file would.
++thumbService.generateVodThumbnail = async () => null;
++thumbService.generateClipThumbnail = async () => null;
++tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: 30, error: null });
++tools.probeDuration = async () => ({ ok: true, seconds: 30, format: {}, streams: [], error: null });
++tools.remuxForSeeking = async () => true;
++tools.probeVodInfo = async () => ({ duration: 12 });
++
++/** The row, and the object its object_id names; asserts the pair agrees on every field a sync writes. */
++function linked(table, keyCol, key, label) {
++ const row = db.get(`SELECT * FROM ${table} WHERE ${keyCol} = ?`, [key]);
++ assert.ok(row, `${label}: row exists`);
++ assert.ok(row.object_id, `${label}: row names its object`);
++ const obj = model.getObject(row.object_id);
++ assert.ok(obj, `${label}: object exists right after the write`);
++ const p = { vods: model.vodProjection, clips: model.clipProjection, files: model.fileProjection, pastes: model.pasteProjection }[table](row);
++ const want = drift.expectedFields(p);
++ for (const f of drift.FIELDS) assert.strictEqual(String(obj[f] ?? ''), String(want[f] ?? ''), `${label}: object ${f}`);
++ return { row, obj };
++}
++const noDrift = (label) => {
++ const r = drift.buildReport(q);
++ assert.strictEqual(r.total_drift, 0, `${label}:\n${drift.formatReport(r)}`);
++ return r;
++};
++
++const app = express();
++app.use(express.json({ limit: '5mb' }));
++app.use('/api/v1/:app/vods', require('../server/vod/routes'));
++app.use('/api/v1/:app/clips', require('../server/vod/clips-routes'));
++app.use('/api/v1/:app/files', require('../server/files/routes'));
++app.use('/api/v1/:app/thumbnails', require('../server/thumbnails/routes'));
++app.use('/api/v1/:app/pastes', require('../server/pastes/routes'));
++const server = http.createServer(app);
++
++(async () => {
++ await new Promise((r) => server.listen(0, '127.0.0.1', r));
++ const base = `http://127.0.0.1:${server.address().port}/api/v1/live`;
++ const auth = { authorization: 'Bearer live-key-object-first' };
++ const call = async (method, p, body) => {
++ const headers = { ...auth };
++ if (body && !(body instanceof FormData)) headers['content-type'] = 'application/json';
++ const r = await fetch(base + p, { method, headers, body: body && !(body instanceof FormData) ? JSON.stringify(body) : body });
++ return { status: r.status, body: await r.json().catch(() => null) };
++ };
++ const form = (field, bytes, type, name, extra = {}) => {
++ const f = new FormData();
++ for (const [k, v] of Object.entries(extra)) f.append(k, String(v));
++ f.append(field, new Blob([bytes], { type }), name);
++ return f;
++ };
++ const png = await sharp({ create: { width: 24, height: 24, channels: 3, background: '#3366ff' } }).png().toBuffer();
++
++ // ── 1. VOD: create, chunk upload, metadata/visibility, finalize, thumbnail, health, tier move ──
++ let r = await call('POST', '/vods', { title: 'First', user_id: 7, visibility: 'unlisted' });
++ assert.strictEqual(r.status, 201);
++ const vodId = r.body.id;
++ let { obj } = linked('vods', 'id', vodId, 'vod create');
++ assert.deepStrictEqual([obj.kind, obj.visibility, obj.lifecycle_status, obj.legacy_ref], ['vod', 'unlisted', 'uploading', `legacy:live:vod:${vodId}`]);
++
++ r = await call('POST', '/vods', { title: 'Clips only', clips_only: true });
++ assert.strictEqual(r.status, 201);
++ const clipsOnly = db.get('SELECT * FROM vods WHERE id = ?', [r.body.id]);
++ assert.strictEqual(clipsOnly.clips_only, 1, 'clips_only written with the row');
++ assert.strictEqual(clipsOnly.object_id, null, 'a clips-only recording has no object (skipped by design)');
++
++ r = await call('POST', `/vods/${vodId}/chunks`, form('chunk', Buffer.alloc(40000, 1), 'video/webm', 'a.webm'));
++ assert.strictEqual(r.body.status, 'created');
++ ({ obj } = linked('vods', 'id', vodId, 'first chunk'));
++ const loc = () => model.listLocations(obj.id).find(l => l.provider === 'local');
++ assert.strictEqual(loc().state, 'present', 'the recording file is a present local copy at once');
++ assert.strictEqual(obj.size_bytes, 40000);
++ r = await call('POST', `/vods/${vodId}/chunks`, form('chunk', Buffer.alloc(8000, 2), 'video/webm', 'b.webm'));
++ assert.strictEqual(r.body.status, 'appended');
++ ({ obj } = linked('vods', 'id', vodId, 'chunk progress'));
++ assert.strictEqual(obj.size_bytes, 48000, 'progress updates carry the object size with them');
++
++ r = await call('PUT', `/vods/${vodId}`, { title: 'Renamed', visibility: 'private' });
++ assert.strictEqual(r.status, 200);
++ ({ obj } = linked('vods', 'id', vodId, 'vod update'));
++ assert.strictEqual(obj.visibility, 'private');
++ assert.strictEqual(model.parseJson(obj.metadata, {}).title, 'Renamed');
++
++ const done = await finalizeVod(vodId);
++ assert.ok(done && done.health_status === 'ok');
++ ({ obj } = linked('vods', 'id', vodId, 'finalize'));
++ assert.strictEqual(obj.lifecycle_status, 'ready', 'the ready transition commits with its object');
++ assert.strictEqual(model.parseJson(obj.metadata, {}).duration_seconds, 30);
++
++ r = await call('POST', `/thumbnails/vod/${vodId}`, { image: png.toString('base64') });
++ assert.strictEqual(r.status, 200);
++ ({ obj } = linked('vods', 'id', vodId, 'thumbnail'));
++ const thumbName = path.basename(r.body.url);
++ const thumbObj = model.getObjectByLegacyRef(`legacy:live:thumbnail:${thumbName}`);
++ assert.ok(thumbObj, 'the thumbnail object is written with the row');
++ assert.strictEqual(model.getVariant(obj.id, 'thumbnail').derived_object_id, thumbObj.id);
++ assert.deepStrictEqual([thumbObj.visibility, thumbObj.size_bytes], ['unlisted', png.length], 'a private VOD\'s thumbnail is unlisted');
++
++ db.updateVodHealth(vodId, { status: 'corrupt', issues: ['test'], quarantine: true });
++ assert.strictEqual(linked('vods', 'id', vodId, 'health').obj.lifecycle_status, 'failed');
++ db.updateVodHealth(vodId, { status: 'ok', issues: [] });
++
++ model.afterTierMove(vodId, ['b2'], () => db.run("UPDATE vods SET storage_provider = 'b2', storage_key = 'vods/x.webm' WHERE id = ?", [vodId]));
++ ({ obj } = linked('vods', 'id', vodId, 'tier move'));
++ assert.strictEqual(obj.canonical_provider, 'b2');
++ assert.strictEqual(model.listLocations(obj.id).find(l => l.provider === 'b2').state, 'present', 'the verified copy is marked with the flip');
++
++ // A ghost row (a recording that never produced a file) gets its object as it is quarantined.
++ raw.prepare("INSERT INTO vods (id, app_id, title, created_at) VALUES (500, 'live', 'ghost', datetime('now', '-2 hours'))").run();
++ assert.strictEqual(vodStorage.reconcileGhosts(), 1);
++ assert.strictEqual(linked('vods', 'id', 500, 'ghost quarantine').obj.lifecycle_status, 'failed');
++ console.log('✅ VOD writes (create, chunks, update, finalize, thumbnail, health, tier move, ghost quarantine) commit with their object');
++
++ // ── 2. Clips: upload, create + re-cut (processing -> ready | failed), update, delete ──
++ r = await call('POST', '/clips', form('video', Buffer.alloc(5000, 3), 'video/webm', 'c.webm', { title: 'Up', vod_id: vodId, visibility: 'private' }));
++ assert.strictEqual(r.status, 201);
++ const upId = r.body.id;
++ ({ obj } = linked('clips', 'id', upId, 'clip upload'));
++ assert.deepStrictEqual([obj.kind, obj.visibility, obj.lifecycle_status, obj.size_bytes], ['clip', 'private', 'ready', 5000]);
++
++ const cutId = Number(db.createClip({ app_id: 'live', vod_id: vodId, user_id: 7, title: 'Cut', start_time: 1, end_time: 6, duration_seconds: 5, visibility: 'unlisted', status: 'processing' }).lastInsertRowid);
++ ({ obj } = linked('clips', 'id', cutId, 'clip create'));
++ assert.deepStrictEqual([obj.visibility, obj.lifecycle_status], ['unlisted', 'uploading']);
++ const cutFile = path.join(process.env.CLIPS_PATH, 'cut.webm');
++ fs.writeFileSync(cutFile, Buffer.alloc(6000, 4));
++ const realResolve = vodStorage.resolveMediaSource;
++ const realCut = cutter.cutClipFile;
++ vodStorage.resolveMediaSource = async () => ({ kind: 'file', value: '/src.webm' });
++ cutter.cutClipFile = async () => ({ ok: true, filePath: cutFile, duration: 5 });
++ assert.strictEqual((await clipJobs.recutClip(cutId)).ok, true);
++ ({ obj } = linked('clips', 'id', cutId, 'clip ready'));
++ assert.deepStrictEqual([obj.lifecycle_status, obj.size_bytes], ['ready', 6000], 'the ready transition (announce) commits with its object');
++ vodStorage.resolveMediaSource = async () => null;
++ assert.strictEqual((await clipJobs.recutClip(cutId)).ok, false);
++ assert.strictEqual(linked('clips', 'id', cutId, 'clip failed').obj.lifecycle_status, 'failed');
++ vodStorage.resolveMediaSource = realResolve;
++ cutter.cutClipFile = realCut;
++
++ r = await call('PUT', `/clips/${upId}`, { title: 'Better', visibility: 'public', auto_generated: 1 });
++ assert.strictEqual(r.status, 200);
++ ({ obj } = linked('clips', 'id', upId, 'clip update'));
++ assert.strictEqual(obj.visibility, 'public');
++ assert.deepStrictEqual([model.parseJson(obj.metadata, {}).title, model.parseJson(obj.metadata, {}).auto_generated], ['Better', true]);
++
++ const doomedFile = path.join(process.env.CLIPS_PATH, 'doomed.webm');
++ fs.writeFileSync(doomedFile, Buffer.alloc(700, 5));
++ const doomed = Number(db.createClip({ app_id: 'live', vod_id: vodId, title: 'Doomed', file_path: doomedFile, status: 'ready' }).lastInsertRowid);
++ const doomedObj = linked('clips', 'id', doomed, 'clip to delete').obj.id;
++ assert.strictEqual((await call('DELETE', `/clips/${doomed}`)).status, 200);
++ assert.strictEqual(model.getObject(doomedObj).lifecycle_status, 'deleted', 'a deleted row takes its object with it (trigger, same statement)');
++ console.log('✅ clip writes (upload, create, re-cut ready/failed, update, delete) commit with their object');
++
++ // ── 3. Files, screenshot pastes, avatars ──
++ r = await call('POST', '/files', form('file', Buffer.from('hello object model'), 'text/plain', 'notes.txt'));
++ assert.strictEqual(r.status, 201);
++ const fileKey = r.body.key;
++ ({ obj } = linked('files', 'key', fileKey, 'file upload'));
++ assert.strictEqual(obj.content_hash, crypto.createHash('sha256').update('hello object model').digest('hex'));
++ const spare = await call('POST', '/files', form('file', Buffer.from('spare'), 'text/plain', 'spare.txt'));
++ const spareObj = linked('files', 'key', spare.body.key, 'file to delete').obj.id;
++ assert.strictEqual((await call('DELETE', `/files/${encodeURIComponent(spare.body.key)}`)).status, 200);
++ assert.strictEqual(model.getObject(spareObj).lifecycle_status, 'deleted');
++
++ r = await call('POST', '/pastes', form('screenshot', png, 'image/png', 'shot.png', { title: 'Shot', visibility: 'unlisted' }));
++ assert.strictEqual(r.status, 201);
++ const slug = r.body.slug;
++ ({ obj } = linked('pastes', 'slug', slug, 'screenshot paste'));
++ assert.deepStrictEqual([obj.kind, obj.visibility, obj.legacy_ref], ['screenshot', 'unlisted', `legacy:live:paste:${slug}`]);
++ assert.strictEqual((await call('PUT', `/pastes/${slug}`, { title: 'Shot 2', visibility: 'private' })).status, 200);
++ assert.strictEqual(linked('pastes', 'slug', slug, 'paste update').obj.visibility, 'private');
++ assert.strictEqual((await call('POST', '/pastes/bulk', { slugs: [slug], action: 'public' })).status, 200);
++ assert.strictEqual(linked('pastes', 'slug', slug, 'paste bulk').obj.visibility, 'public');
++ const censor = await call('POST', `/pastes/${slug}/censor`, form('screenshot', png, 'image/png', 'censored.png'));
++ assert.strictEqual(censor.status, 200);
++ assert.strictEqual(linked('pastes', 'slug', slug, 'paste censor').obj.size_bytes, png.length);
++ const text = await call('POST', '/pastes', { content: 'just text' });
++ assert.strictEqual(db.get('SELECT object_id FROM pastes WHERE slug = ?', [text.body.slug]).object_id, null, 'a text paste has no bytes and no object');
++
++ // Avatar ingest's write (server/avatars/ingest.js): the avatar row and its object together.
++ const avatarFile = path.join(process.env.PASTES_PATH, 'avatar.webp');
++ fs.writeFileSync(avatarFile, png);
++ const avatarId = Number(db.withObject('paste', (x) => x.lastInsertRowid, () => db.run(`INSERT INTO pastes (app_id, slug, user_id, type, title, content, language, visibility, screenshot_path, metadata)
++ VALUES ('network', 'av1', 42, 'screenshot', 'Avatar', '', 'text', 'unlisted', ?, ?)`, [avatarFile, JSON.stringify({ kind: 'avatar', mime_type: 'image/webp' })])).lastInsertRowid);
++ assert.deepStrictEqual([linked('pastes', 'id', avatarId, 'avatar').obj.kind, linked('pastes', 'id', avatarId, 'avatar').obj.legacy_ref], ['avatar', 'legacy:network:avatar:av1']);
++ console.log('✅ file, screenshot-paste and avatar writes commit with their object; deletes mark it deleted');
++
++ const clean = noDrift('after every write path');
++ assert.ok(clean.total_rows >= 8, 'the report checked the rows written above');
++
++ // ── 4. Atomicity: an object write that fails takes the row write with it ──
++ raw.exec(`CREATE TEMP TRIGGER t_refuse_object_insert BEFORE INSERT ON main.media_objects BEGIN SELECT RAISE(ABORT, 'object write refused (test)'); END;
++ CREATE TEMP TRIGGER t_refuse_object_update BEFORE UPDATE ON main.media_objects BEGIN SELECT RAISE(ABORT, 'object write refused (test)'); END;`);
++ const before = { vods: count('vods'), clips: count('clips'), files: count('files'), pastes: count('pastes'), objects: count('media_objects') };
++ const unchanged = (label) => assert.deepStrictEqual({ vods: count('vods'), clips: count('clips'), files: count('files'), pastes: count('pastes'), objects: count('media_objects') }, before, label);
++
++ assert.throws(() => db.createVod({ app_id: 'live', title: 'No object' }), /object write refused/);
++ unchanged('createVod: neither row');
++ assert.strictEqual((await call('POST', '/vods', { title: 'No object' })).status, 500);
++ unchanged('POST /vods: neither row');
++ assert.throws(() => db.createClip({ app_id: 'live', vod_id: vodId, title: 'No object', status: 'processing' }), /object write refused/);
++ unchanged('createClip: neither row');
++ assert.strictEqual((await call('POST', '/files', form('file', Buffer.from('never stored'), 'text/plain', 'never.txt'))).status, 500);
++ unchanged('POST /files: neither row');
++ assert.strictEqual((await call('POST', '/pastes', form('screenshot', png, 'image/png', 'never.png'))).status, 500);
++ unchanged('POST /pastes screenshot: neither row');
++
++ // Updates: the row keeps its old values when its object cannot follow.
++ assert.throws(() => db.setVodVisibility(vodId, 'public'), /object write refused/);
++ assert.strictEqual(db.get('SELECT visibility FROM vods WHERE id = ?', [vodId]).visibility, 'private', 'visibility change rolled back');
++ assert.strictEqual((await call('PUT', `/clips/${upId}`, { title: 'Lost' })).status, 500);
++ assert.strictEqual(db.get('SELECT title FROM clips WHERE id = ?', [upId]).title, 'Better', 'clip title rolled back');
++ // The announce() path: a finalize whose object cannot be written commits no ready transition (and no event).
++ raw.prepare("UPDATE vods SET is_recording = 1, health_status = 'unknown' WHERE id = ?").run(vodId);
++ await assert.rejects(finalizeVod(vodId, { fromJob: true }), /object write refused/);
++ assert.strictEqual(db.get('SELECT is_recording FROM vods WHERE id = ?', [vodId]).is_recording, 1, 'ready transition rolled back with its object');
++
++ // A write that throws after its projection insert leaves neither row.
++ assert.throws(() => db.withObject('vod', (x) => x.lastInsertRowid, () => {
++ db.run("INSERT INTO vods (app_id, title) VALUES ('live', 'half')");
++ throw new Error('crash after the insert');
++ }), /crash after the insert/);
++ unchanged('thrown after the insert: neither row');
++ raw.exec('DROP TRIGGER temp.t_refuse_object_insert; DROP TRIGGER temp.t_refuse_object_update;');
++ await finalizeVod(vodId, { fromJob: true });
++ noDrift('after the refused writes');
++ console.log('✅ a write whose object cannot be written leaves neither row (inserts, updates, announce path, a throw after the insert)');
++
++ // ── 5. Drift report: read-only, finds planted drift ──
++ raw.prepare("INSERT INTO vods (id, app_id, user_id, title, file_path) VALUES (900, 'live', 3, 'Planted: no object', NULL)").run();
++ const upObj = db.get('SELECT object_id FROM clips WHERE id = ?', [upId]).object_id;
++ raw.prepare("UPDATE media_objects SET visibility = 'private' WHERE id = ?").run(upObj);
++ const fileObj = db.get('SELECT object_id FROM files WHERE key = ?', [fileKey]).object_id;
++ raw.prepare('UPDATE media_objects SET size_bytes = 1, owner_user_id = 99 WHERE id = ?').run(fileObj);
++ raw.prepare('UPDATE pastes SET object_id = NULL WHERE slug = ?').run(slug);
++ const cutObj = db.get('SELECT object_id FROM clips WHERE id = ?', [cutId]).object_id;
++ raw.prepare("UPDATE media_objects SET lifecycle_status = 'deleted' WHERE id = ?").run(cutObj);
++
++ const snapshot = () => ['vods', 'clips', 'files', 'pastes', 'media_objects', 'media_locations', 'media_variants', 'media_relationships']
++ .map(t => JSON.stringify(raw.prepare(`SELECT * FROM ${t} ORDER BY rowid`).all())).join('|');
++ const snap = snapshot();
++ const writesBefore = raw.prepare('SELECT total_changes() AS n').get().n;
++ const rep = drift.buildReport(q, { limit: 5 });
++ assert.strictEqual(raw.prepare('SELECT total_changes() AS n').get().n, writesBefore, 'the report issues no write');
++ assert.strictEqual(snapshot(), snap, 'nothing changed');
++
++ assert.strictEqual(rep.total_drift, 5, drift.formatReport(rep));
++ assert.deepStrictEqual([rep.kinds.vod.missing, rep.kinds.clip.mismatch, rep.kinds.file.mismatch, rep.kinds.screenshot.unlinked], [1, 2, 1, 1]);
++ assert.deepStrictEqual([rep.field_counts.visibility, rep.field_counts.size_bytes, rep.field_counts.owner_user_id, rep.field_counts.lifecycle_status], [1, 1, 1, 1]);
++ const ex = (kind, id) => rep.kinds[kind].examples.find(e => String(e.id) === String(id));
++ assert.deepStrictEqual(ex('vod', 900).problems, ['missing']);
++ assert.deepStrictEqual(ex('clip', upId).diff.visibility, { row: 'public', object: 'private' });
++ assert.deepStrictEqual(ex('clip', cutId).diff.lifecycle_status, { row: 'failed', object: 'deleted' });
++ assert.deepStrictEqual(Object.keys(ex('file', fileKey).diff).sort(), ['owner_user_id', 'size_bytes']);
++ assert.deepStrictEqual(ex('screenshot', db.get('SELECT id FROM pastes WHERE slug = ?', [slug]).id).problems, ['unlinked']);
++ assert.strictEqual(rep.kinds.vod.skipped, 1, 'the clips-only recording is skipped, not drift');
++ assert.strictEqual(drift.buildReport(q, { appId: 'network' }).total_drift, 0, '--app narrows the report');
++ assert.match(drift.formatReport(rep), /total drift: 5 of \d+ row\(s\)/);
++
++ // The script: the database opened read-only, exit 0 whatever it finds (and on an error).
++ const script = path.join(__dirname, '..', 'scripts', 'object-drift-report.js');
++ const out = JSON.parse(execFileSync(process.execPath, [script, '--db', process.env.DB_PATH, '--json', '--limit', '1'], { env: process.env, encoding: 'utf8' }));
++ assert.strictEqual(out.total_drift, 5);
++ assert.ok(Object.values(out.kinds).every(k => k.examples.length <= 1), '--limit caps the examples per kind');
++ assert.match(execFileSync(process.execPath, [script, '--db', process.env.DB_PATH], { env: process.env, encoding: 'utf8' }), /clip examples \(2 of 2\)/);
++ execFileSync(process.execPath, [script, '--db', path.join(tmp, 'missing.db')], { env: process.env, stdio: 'ignore' }); // throws on a non-zero exit
++ assert.strictEqual(snapshot(), snap, 'the script changed nothing either');
++
++ // Re-projecting the planted rows clears the drift the report listed.
++ for (const id of [900]) model.sync('vod', id);
++ model.sync('clip', upId); model.sync('clip', cutId); model.sync('file', fileKey);
++ model.sync('paste', db.get('SELECT id FROM pastes WHERE slug = ?', [slug]).id);
++ noDrift('after re-projecting the planted drift');
++ console.log('✅ drift report: read-only (report and script), counts per kind with examples, finds missing / unlinked / mismatched objects');
++
++ server.close();
++ fs.rmSync(tmp, { recursive: true, force: true });
++ console.log('✅ All object-first write tests passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });