Build a resumable video migration worker: ledger, bounded concurrency, 429 backoff

mkdir video-migration && cd video-migration
npm init -y
npm pkg set type=module          # everything below is ESM
npm i better-sqlite3 undici

// db.js
import Database from 'better-sqlite3';

export const db = new Database('migration.db');

db.exec(`
  CREATE TABLE IF NOT EXISTS assets (
    id            TEXT PRIMARY KEY,   -- our stable ID, also the idempotency key
    source_url    TEXT NOT NULL,
    state         TEXT NOT NULL DEFAULT 'pending',
      -- pending | inventoried | uploading | processing | ready | failed | rejected
    remote_id     TEXT,
    duration_s    REAL,
    attempts      INTEGER NOT NULL DEFAULT 0,
    retry_after   INTEGER,            -- unix ts; do not re-select before this
    last_error    TEXT,
    updated_at    INTEGER NOT NULL DEFAULT (unixepoch())
  );
  CREATE INDEX IF NOT EXISTS idx_state ON assets(state);
`);

// A real claim: select, then flip to 'uploading' in the same transaction so a second
// worker (or a restart that overlaps a running process) cannot pick up the same rows.
export const claimBatch = db.transaction((limit = 200) => {
  const rows = db.prepare(`
    SELECT * FROM assets
    WHERE state IN ('inventoried','failed')
      AND attempts < 5
      AND (retry_after IS NULL OR retry_after <= unixepoch())
    ORDER BY attempts ASC, rowid ASC
    LIMIT ?
  `).all(limit);

  const claim = db.prepare(
    `UPDATE assets SET state='uploading', updated_at=unixepoch() WHERE id = ?`
  );
  for (const r of rows) claim.run(r.id);
  return rows;
});

// Anything left 'uploading' from a previous run was interrupted, not finished.
export function releaseStale(olderThanSeconds = 900) {
  db.prepare(`
    UPDATE assets SET state='inventoried'
    WHERE state='uploading' AND updated_at < unixepoch() - ?
  `).run(olderThanSeconds);
}

export function setState(id, state, patch = {}) {
  const cols = Object.keys(patch);
  const sets = ['state = ?', 'updated_at = unixepoch()', ...cols.map(c => `${c} = ?`)];
  db.prepare(`UPDATE assets SET ${sets.join(', ')} WHERE id = ?`)
    .run(state, ...cols.map(c => patch[c]), id);
}

// inventory.js
import { execFile } from 'node:child_process';
import { promisify } from 'node:util';
import { db, setState } from './db.js';

const run = promisify(execFile);

async function probe(url) {
  const { stdout } = await run('ffprobe', [
    '-v', 'error',
    '-print_format', 'json',
    '-show_format', '-show_streams',
    url
  ], { maxBuffer: 8 * 1024 * 1024 });
  return JSON.parse(stdout);
}

for (const row of db.prepare(`SELECT * FROM assets WHERE state = 'pending'`).all()) {
  try {
    const info = await probe(row.source_url);
    const v = info.streams.find(s => s.codec_type === 'video');
    if (!v) throw new Error('no video stream');
    setState(row.id, 'inventoried', { duration_s: Number(info.format.duration) });
  } catch (err) {
    setState(row.id, 'rejected', { last_error: String(err.message).slice(0, 500) });
  }
}

$ node inventory.js
$ sqlite3 migration.db "SELECT state, count(*) FROM assets GROUP BY state"
inventoried|9612
rejected|41

// governor.js
export class Governor {
  constructor({ min = 1, max = 16, start = 4 } = {}) {
    this.min = min; this.max = max; this.limit = start;
    this.inFlight = 0; this.queue = [];
    this.pausedUntil = 0;
    this.okStreak = 0;
  }

  async acquire() {
    while (Date.now() < this.pausedUntil || this.inFlight >= this.limit) {
      await new Promise(r => setTimeout(r, 50));
    }
    this.inFlight++;
  }

  release() { this.inFlight--; }

  // Called on every response.
  observe(status, retryAfterHeader) {
    if (status === 429 || status === 503) {
      const secs = parseRetryAfter(retryAfterHeader) ?? 5;
      this.pausedUntil = Date.now() + secs * 1000;
      this.limit = Math.max(this.min, Math.floor(this.limit / 2));
      this.okStreak = 0;
      console.warn(`[governor] backpressure: pausing ${secs}s, concurrency -> ${this.limit}`);
    } else if (status < 400) {
      if (++this.okStreak >= 50 && this.limit < this.max) {
        this.limit++; this.okStreak = 0;
        console.info(`[governor] concurrency -> ${this.limit}`);
      }
    }
  }
}

function parseRetryAfter(h) {
  if (!h) return null;
  const n = Number(h);
  if (Number.isFinite(n)) return n;               // delta-seconds
  const t = Date.parse(h);                        // HTTP-date
  return Number.isFinite(t) ? Math.max(0, (t - Date.now()) / 1000) : null;
}

// migrate.js
import { request } from 'undici';
import { db, claimBatch, setState, releaseStale } from './db.js';
import { Governor } from './governor.js';

releaseStale();   // recover rows a previous crashed run left mid-flight

const API = process.env.VIDEO_API_BASE;      // e.g. https://api.example.com/v1
const TOKEN = process.env.VIDEO_API_TOKEN;
const gov = new Governor({ start: 4, max: 16 });

async function createAsset(row) {
  const res = await request(`${API}/videos`, {
    method: 'POST',
    headers: {
      'authorization': `Bearer ${TOKEN}`,
      'content-type': 'application/json',
      'idempotency-key': row.id            // <- the whole trick
    },
    body: JSON.stringify({
      input: row.source_url,
      metadata: { migration_ref: row.id }  // fallback lookup key
    })
  });
  const body = await res.body.json().catch(() => ({}));
  gov.observe(res.statusCode, res.headers['retry-after']);
  return { status: res.statusCode, body };
}

async function findExisting(row) {
  const res = await request(`${API}/videos?metadata[migration_ref]=${row.id}`, {
    headers: { authorization: `Bearer ${TOKEN}` }
  });
  const body = await res.body.json().catch(() => ({}));
  gov.observe(res.statusCode, res.headers['retry-after']);
  return body?.data?.[0]?.id ?? null;
}

async function migrateOne(row) {
  await gov.acquire();
  try {
    let { status, body } = await createAsset(row);

    if (status === 429 || status >= 500) {
      // back off this row specifically, so the next claimBatch does not grab it immediately
      const delay = Math.min(300, 2 ** row.attempts) + Math.random() * 5;   // jitter
      setState(row.id, 'failed', {
        attempts: row.attempts + 1,
        retry_after: Math.floor(Date.now() / 1000 + delay),
        last_error: `transient ${status}`
      });
      return;
    }
    if (status === 409) {                       // already created, provider told us
      const existing = await findExisting(row);
      if (existing) return setState(row.id, 'processing', { remote_id: existing });
    }
    if (status >= 400) {
      setState(row.id, 'rejected', {
        attempts: row.attempts + 1,
        last_error: JSON.stringify(body).slice(0, 500)
      });
      return;
    }
    setState(row.id, 'processing', { remote_id: body.id ?? body.data?.id });
  } finally {
    gov.release();
  }
}

for (;;) {
  const batch = claimBatch(200);
  if (batch.length === 0) break;
  await Promise.all(batch.map(migrateOne));
}
console.log('upload pass complete');

$ node migrate.js
[governor] concurrency -> 5
[governor] concurrency -> 6
[governor] backpressure: pausing 12s, concurrency -> 3
[governor] concurrency -> 4
upload pass complete

// webhook.js
import http from 'node:http';
import crypto from 'node:crypto';
import { setState } from './db.js';

const SECRET = process.env.WEBHOOK_SECRET;

function verify(rawBuffer, sig) {
  const expected = crypto.createHmac('sha256', SECRET).update(rawBuffer).digest('hex');
  // timingSafeEqual throws on length mismatch, so guard first
  const a = Buffer.from(expected), b = Buffer.from(sig ?? '', 'utf8');
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

http.createServer((req, res) => {
  const chunks = [];
  req.on('data', c => chunks.push(c));
  req.on('end', () => {
    const raw = Buffer.concat(chunks);   // HMAC the bytes, never a decoded string
    if (!verify(raw, req.headers['x-signature'])) {
      res.writeHead(401).end();
      return;
    }
    const evt = JSON.parse(raw.toString('utf8'));
    const ref = evt.data?.metadata?.migration_ref;
    if (ref && evt.type === 'video.ready') setState(ref, 'ready');
    if (ref && evt.type === 'video.errored') {
      setState(ref, 'rejected', { last_error: evt.data.error ?? 'ingest failed' });
    }
    res.writeHead(204).end();   // ack fast, work later
  });
}).listen(8787);

-- create it once, populate it from the reconcile pass
CREATE TABLE IF NOT EXISTS remote_assets (
  migration_ref     TEXT PRIMARY KEY,
  remote_duration_s REAL
);

-- durations that drifted by more than half a second are the canary
SELECT a.id, a.duration_s, r.remote_duration_s
FROM assets a JOIN remote_assets r ON r.migration_ref = a.id
WHERE abs(a.duration_s - r.remote_duration_s) > 0.5;