// Background removal for camera tracks with Robust Video Matting (Lin et al., 2021), run on the CPU
// through onnxruntime, or on a remote machine's GPU over ssh when a host is set in the settings. For each take, camera.webm is decoded to 30 fps RGB frames, the recurrent
// model predicts a foreground color and an alpha matte per frame, and the result is written as
// cutout.webm (VP9 with an alpha channel), which Chrome plays with transparency and ffmpeg can
// composite at export. Takes are processed one at a time, since each job already uses most cores.
// The camera is first scaled down to the processing resolution chosen for the video, and a queued
// or running job can be cancelled. On the CPU, the model is not run on repeated frames or on some
// frames with little motion (see SKIP_MOTION).

import { spawn } from 'node:child_process';
import { once } from 'node:events';
import { createReadStream, createWriteStream } from 'node:fs';
import fs from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import { Transform, type Readable } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import ort from 'onnxruntime-node';
import type { Take, Video } from '../shared/types.ts';
import { backgroundOf } from '../shared/types.ts';
import { FPS } from '../shared/layout.ts';
import { probe } from './media.ts';
import { HttpError, exists, getConfig, getVideo, takeDir, updateVideo } from './store.ts';

const MODEL_URL = 'https://github.com/PeterL1n/RobustVideoMatting/releases/download/v1.0.0/rvm_mobilenetv3_fp32.onnx';
const MODEL_PATH = path.join(os.homedir(), '.cache', 'takes', 'models', 'rvm_mobilenetv3_fp32.onnx');

interface Job {
  videoId: string;
  takeId: string;
  stage?: string; // what the running job is doing, for the UI
  progress?: number; // fraction of the stage done, if known
  running: boolean;
  camera?: { width: number; height: number }; // size of the camera track, once running
  height?: number; // lines being processed, once running
  abort: AbortController;
  restart?: boolean; // aborted because the processing resolution changed; run again
  prev: Take['matte']; // persisted state before queueing, restored if a redo is cancelled
}

const jobs = new Map<string, Job>(); // key: videoId/takeId, in queue order
let pumping = false;
let session: Promise<ort.InferenceSession> | null = null;

const key = (videoId: string, takeId: string) => `${videoId}/${takeId}`;

export const cutoutFile = (videoId: string, takeId: string) => path.join(takeDir(videoId, takeId), 'cutout.webm');

// Merge the live queue state into a video before sending it to the client.
export function withMatteStatus(v: Video): Video {
  const takes = v.takes.map((t): Take => {
    const j = jobs.get(key(v.id, t.id));
    return j ? { ...t, matte: { status: j.running ? 'running' : 'queued', stage: j.stage, progress: j.progress, updatedAt: new Date().toISOString() } } : t;
  });
  return { ...v, takes };
}

const setMatte = (videoId: string, takeId: string, matte: Take['matte']) =>
  updateVideo(videoId, (v) => {
    const t = v.takes.find((t) => t.id === takeId);
    if (t) t.matte = matte;
  }).catch(() => {}); // the video or take may have been deleted meanwhile

// The size a camera track is processed at: its own size, scaled down to at most `lines` lines.
function processingSize(width: number, height: number, lines: number) {
  const scale = Math.min(1, lines / height);
  return { W: even(width * scale), H: even(height * scale), capped: scale < 1 };
}

// Queue every ready camera take of the video that has not been processed yet. With a take id, also
// queue that take if it failed, was cancelled, or is done (to process it again at the current
// resolution). A running job whose resolution no longer matches the setting is restarted.
export async function requestMattes(videoId: string, opts: { takeId?: string } = {}) {
  const v = await getVideo(videoId);
  const lines = backgroundOf(v.layout).resolution;
  for (const t of v.takes) {
    if (opts.takeId && t.id !== opts.takeId) continue;
    if (t.status !== 'ready' || !t.tracks.camera) continue;
    const job = jobs.get(key(videoId, t.id));
    if (job) {
      if (job.running && job.camera && !job.restart && processingSize(job.camera.width, job.camera.height, lines).H !== job.height) {
        job.restart = true;
        job.abort.abort(new Error('Restarting at a new resolution'));
      }
      continue;
    }
    if (!opts.takeId) {
      if (t.matte?.status === 'failed' || t.matte?.status === 'cancelled') continue;
      if (t.matte?.status === 'done' && (await exists(cutoutFile(videoId, t.id)))) continue;
    }
    jobs.set(key(videoId, t.id), { videoId, takeId: t.id, running: false, abort: new AbortController(), prev: t.matte });
  }
  pump();
}

// Cancel the queued and running jobs of a video (or of one take). A cancelled redo keeps its
// previous cutout; otherwise the take is marked cancelled, so that it is not queued again until
// asked for.
export async function cancelMattes(videoId: string, takeId?: string) {
  for (const job of [...jobs.values()]) {
    if (job.videoId !== videoId || (takeId && job.takeId !== takeId)) continue;
    job.restart = false;
    if (job.running) job.abort.abort(new Error('Cancelled'));
    else {
      jobs.delete(key(job.videoId, job.takeId)); // before the await, so that the pump does not start it
      await setMatte(job.videoId, job.takeId, cancelledState(job));
    }
  }
}

const cancelledState = (job: Job): Take['matte'] => (job.prev?.status === 'done' ? job.prev : { status: 'cancelled', updatedAt: new Date().toISOString() });

async function pump() {
  if (pumping) return;
  pumping = true;
  try {
    for (;;) {
      const job = jobs.values().next().value as Job | undefined;
      if (!job) break;
      job.running = true;
      const k = key(job.videoId, job.takeId);
      try {
        const t0 = Date.now();
        const { matteHost, matteSetup } = await getConfig();
        const { frames, H, capped } = await processTake(job, { host: matteHost, setup: matteSetup });
        console.log(`matte: ${k}: ${frames} frames at ${H} lines in ${((Date.now() - t0) / 1000).toFixed(1)} s${matteHost ? ` on ${matteHost}` : ''}`);
        await setMatte(job.videoId, job.takeId, { status: 'done', height: H, capped, updatedAt: new Date().toISOString() });
        jobs.delete(k);
      } catch (e) {
        if (job.restart) {
          // Run it again next, with a fresh controller; it is still first in the queue.
          console.log(`matte: ${k}: restarting at a new resolution`);
          Object.assign(job, { running: false, stage: undefined, progress: undefined, camera: undefined, height: undefined, restart: false, abort: new AbortController() });
        } else if (job.abort.signal.aborted) {
          console.log(`matte: ${k}: cancelled`);
          await setMatte(job.videoId, job.takeId, cancelledState(job));
          jobs.delete(k);
        } else {
          console.warn(`matte failed for ${k}:`, (e as Error).message);
          await setMatte(job.videoId, job.takeId, { status: 'failed', error: (e as Error).message, updatedAt: new Date().toISOString() });
          jobs.delete(k);
        }
      }
    }
  } finally {
    pumping = false;
  }
}

async function ensureModel() {
  if (await exists(MODEL_PATH)) return;
  console.log(`Downloading background removal model to ${MODEL_PATH}`);
  const res = await fetch(MODEL_URL);
  if (!res.ok) throw new Error(`Could not download the model (${res.status}) from ${MODEL_URL}`);
  await fs.mkdir(path.dirname(MODEL_PATH), { recursive: true });
  const tmp = `${MODEL_PATH}.${process.pid}.tmp`;
  await fs.writeFile(tmp, Buffer.from(await res.arrayBuffer()));
  await fs.rename(tmp, MODEL_PATH);
}

function getSession() {
  if (!session) {
    session = ensureModel().then(() =>
      ort.InferenceSession.create(MODEL_PATH, {
        executionProviders: ['cpu'],
        graphOptimizationLevel: 'all',
        intraOpNumThreads: Math.max(1, os.cpus().length - 2),
      }),
    );
    session.catch(() => (session = null)); // allow a retry after e.g. a failed download
  }
  return session;
}

// Yield fixed-size frames from a raw video stream. The same buffer is reused, so each frame must be
// consumed before asking for the next.
async function* readFrames(stream: Readable, size: number) {
  const frame = Buffer.alloc(size);
  let off = 0;
  for await (const chunk of stream as AsyncIterable<Buffer>) {
    let p = 0;
    while (p < chunk.length) {
      const n = Math.min(size - off, chunk.length - p);
      chunk.copy(frame, off, p, p + n);
      off += n;
      p += n;
      if (off === size) {
        yield frame;
        off = 0;
      }
    }
  }
}

const even = (n: number) => Math.max(2, Math.round(n / 2) * 2);

// ffmpeg arguments shared by the local and the remote path, so that both produce the same cutout.webm.
const decoderArgs = (src: string, W: number, H: number) => ['-v', 'error', '-i', src, '-an', '-vf', `fps=${FPS}:start_time=0,scale=${W}:${H}`, '-f', 'rawvideo', '-pix_fmt', 'rgb24', '-'];
const encoderArgs = (out: string, W: number, H: number) =>
  ['-y', '-v', 'error', '-f', 'rawvideo', '-pix_fmt', 'rgba', '-s', `${W}x${H}`, '-r', String(FPS), '-i', '-',
    '-c:v', 'libvpx-vp9', '-pix_fmt', 'yuva420p', '-deadline', 'realtime', '-cpu-used', '8', '-row-mt', '1', '-b:v', '3M', '-auto-alt-ref', '0', out];

interface Plan {
  src: string;
  out: string;
  W: number;
  H: number;
  ds: number;
  total: number;
}

async function processTake(job: Job, remote: { host: string; setup: string }) {
  const src = path.join(takeDir(job.videoId, job.takeId), 'camera.webm');
  const info = await probe(src);
  if (!info.width || !info.height) throw new HttpError(400, 'Camera track has no video');
  const lines = backgroundOf((await getVideo(job.videoId)).layout).resolution;
  const { W, H, capped } = processingSize(info.width, info.height, lines);
  job.camera = { width: info.width, height: info.height };
  job.height = H;
  // Downsample ratio for the model's coarse pass, which takes most of the time. Following the RVM
  // guidance it is at most 480 px wide (0.25 at 1080p, 0.375 at 720p); at low processing resolutions
  // it is also capped at 0.6, so that a lower setting shrinks the coarse pass too (on a 12-thread
  // laptop, 76 ms per frame at 426x240 instead of 250 ms at 1280x720).
  const ds = Math.min(0.6, 480 / Math.max(W, H));
  const total = Math.max(1, Math.round(info.duration * FPS));

  const out = cutoutFile(job.videoId, job.takeId);
  const tmp = `${out}.partial.webm`;
  const plan: Plan = { src, out: tmp, W, H, ds, total };
  try {
    job.abort.signal.throwIfAborted();
    const n = remote.host ? await runRemote(job, plan, remote.host, remote.setup) : await runLocal(job, plan);
    job.abort.signal.throwIfAborted();
    if (n === 0) throw new Error('No camera frames were decoded');
    await fs.rename(tmp, out);
    return { frames: n, H, capped };
  } catch (e) {
    await fs.rm(tmp, { force: true });
    throw e;
  }
}

// Skipping frames on the CPU. Much of a camera track repeats: a webcam that delivers 15 frames per
// second in dim light has each frame decoded twice at 30 fps, and when the speaker is still, one
// frame differs little from the next. So a frame identical to the previous one gets the same output
// without running the model, and when the frames on either side of a frame differ by less than
// SKIP_MOTION (mean absolute difference, 0-255 per channel) near the speaker's outline, the model is
// not run on it either. Its matte is then the mean of its neighbours' mattes, and its colour is its
// own camera image plus the mean of the neighbours' corrections (the model's foreground colour minus
// the camera image), which are nonzero only near the outline; so the face stays exactly as recorded.
// At most every other frame is skipped this way. On a 13-minute take at 240 lines, with the
// threshold of 2, the matte of a skipped frame had at most 3% of its outline pixels off by more than
// 0.2 from what the model gives for that frame.
const SKIP_MOTION = 2;

interface Matted {
  raw: Buffer; // RGB, interleaved
  pha: Float32Array;
  fgr: Float32Array; // planar RGB
  band?: Uint8Array; // pixels near the outline, computed when needed
}

// Pixels within 3 px of where the matte is neither 0 nor 1 (a dilation, done as sliding-window
// counts along the rows and then the columns).
function outlineBand(pha: Float32Array, W: number, H: number, R = 3) {
  const rows = new Uint8Array(W * H);
  for (let y = 0; y < H; y++) {
    const o = y * W;
    const edge = (x: number) => (x >= 0 && x < W && pha[o + x] > 0.02 && pha[o + x] < 0.98 ? 1 : 0);
    let count = 0;
    for (let x = 0; x < R; x++) count += edge(x);
    for (let x = 0; x < W; x++) {
      count += edge(x + R) - edge(x - R - 1);
      rows[o + x] = count > 0 ? 1 : 0;
    }
  }
  const band = new Uint8Array(W * H);
  for (let x = 0; x < W; x++) {
    const near = (y: number) => (y >= 0 && y < H ? rows[y * W + x] : 0);
    let count = 0;
    for (let y = 0; y < R; y++) count += near(y);
    for (let y = 0; y < H; y++) {
      count += near(y + R) - near(y - R - 1);
      band[y * W + x] = count > 0 ? 1 : 0;
    }
  }
  return band;
}

// Mean absolute difference of two RGB frames over a band (0 when the band is empty).
function bandMotion(a: Buffer, b: Buffer, band: Uint8Array) {
  let sum = 0;
  let n = 0;
  for (let i = 0; i < band.length; i++) {
    if (!band[i]) continue;
    const j = i * 3;
    sum += Math.abs(a[j] - b[j]) + Math.abs(a[j + 1] - b[j + 1]) + Math.abs(a[j + 2] - b[j + 2]);
    n += 3;
  }
  return n ? sum / n : 0;
}

async function runLocal(job: Job, { src, out, W, H, ds, total }: Plan): Promise<number> {
  job.stage = (await exists(MODEL_PATH)) ? 'Loading model' : 'Downloading model';
  const sess = await getSession();
  job.stage = 'Removing background';
  job.progress = 0;
  const { signal } = job.abort;
  signal.throwIfAborted();
  const decoder = spawn('ffmpeg', decoderArgs(src, W, H), { stdio: ['ignore', 'pipe', 'pipe'] });
  const encoder = spawn('ffmpeg', encoderArgs(out, W, H), { stdio: ['pipe', 'ignore', 'pipe'] });
  // SIGKILL, since ffmpeg does not act on SIGTERM while it is blocked reading its input.
  const kill = () => {
    encoder.stdin.destroy();
    decoder.kill('SIGKILL');
    encoder.kill('SIGKILL');
  };
  signal.addEventListener('abort', kill, { once: true });
  encoder.stdin.on('error', () => {}); // e.g. EPIPE after a kill; the exit code reports failures
  let stderr = '';
  decoder.stderr.on('data', (d) => (stderr += d));
  encoder.stderr.on('data', (d) => (stderr += d));
  const decoderDone = once(decoder, 'close');
  const encoderDone = once(encoder, 'close');

  const plane = W * H;
  const input = new Float32Array(3 * plane);
  let rec: ort.Tensor[] = [0, 1, 2, 3].map(() => new ort.Tensor('float32', new Float32Array([0]), [1, 1, 1, 1]));
  const dsr = new ort.Tensor('float32', new Float32Array([ds]), [1]);

  const matte = async (raw: Buffer): Promise<Matted> => {
    for (let i = 0, j = 0; i < plane; i++, j += 3) {
      input[i] = raw[j] / 255;
      input[i + plane] = raw[j + 1] / 255;
      input[i + 2 * plane] = raw[j + 2] / 255;
    }
    const r = await sess.run({
      src: new ort.Tensor('float32', input, [1, 3, H, W]),
      r1i: rec[0],
      r2i: rec[1],
      r3i: rec[2],
      r4i: rec[3],
      downsample_ratio: dsr,
    });
    rec = [r.r1o, r.r2o, r.r3o, r.r4o];
    return { raw, pha: r.pha.data as Float32Array, fgr: r.fgr.data as Float32Array };
  };
  const rgbaOf = (m: Matted) => {
    const rgba = new Uint8ClampedArray(plane * 4);
    for (let i = 0, j = 0; i < plane; i++, j += 4) {
      rgba[j] = m.fgr[i] * 255;
      rgba[j + 1] = m.fgr[i + plane] * 255;
      rgba[j + 2] = m.fgr[i + 2 * plane] * 255;
      rgba[j + 3] = m.pha[i] * 255;
    }
    return Buffer.from(rgba.buffer);
  };
  // A skipped frame between two matted ones (see SKIP_MOTION).
  const rgbaBetween = (raw: Buffer, p: Matted, q: Matted) => {
    const rgba = new Uint8ClampedArray(plane * 4);
    for (let i = 0, j = 0, k = 0; i < plane; i++, j += 4, k += 3) {
      for (let c = 0; c < 3; c++) {
        const corr = p.fgr[i + c * plane] - p.raw[k + c] / 255 + q.fgr[i + c * plane] - q.raw[k + c] / 255;
        rgba[j + c] = raw[k + c] + corr * 127.5;
      }
      rgba[j + 3] = (p.pha[i] + q.pha[i]) * 127.5;
    }
    return Buffer.from(rgba.buffer);
  };

  let n = 0; // frames written
  const write = async (buf: Buffer, count = 1) => {
    for (let c = 0; c < count; c++) {
      if (!encoder.stdin.write(buf)) await once(encoder.stdin, 'drain', { signal });
      n++;
    }
    job.progress = Math.min(0.99, n / total);
  };

  let prevRaw: Buffer | null = null; // the previous decoded frame
  let last: Matted | null = null; // the last frame the model ran on
  let lastOut: Buffer | null = null; // the output of the previous decoded frame
  let pending: { raw: Buffer; count: number } | null = null; // a frame that may be skipped, and its repeats
  let processed = 0;
  let skipped = 0;
  try {
    for await (const frame of readFrames(decoder.stdout, plane * 3)) {
      signal.throwIfAborted();
      if (prevRaw && frame.equals(prevRaw)) {
        if (pending) pending.count++;
        else await write(lastOut!);
        continue;
      }
      const raw = Buffer.from(frame); // readFrames reuses its buffer
      prevRaw = raw;
      if (!last) {
        last = await matte(raw);
        processed++;
        await write((lastOut = rgbaOf(last)));
      } else if (!pending) {
        pending = { raw, count: 1 };
      } else {
        last.band ??= outlineBand(last.pha, W, H);
        if (bandMotion(last.raw, raw, last.band) < SKIP_MOTION) {
          const next = await matte(raw);
          processed++;
          skipped++;
          await write(rgbaBetween(pending.raw, last, next), pending.count);
          last = next;
          pending = null;
          await write((lastOut = rgbaOf(last)));
        } else {
          last = await matte(pending.raw);
          processed++;
          await write((lastOut = rgbaOf(last)), pending.count);
          pending = { raw, count: 1 };
        }
      }
    }
    signal.throwIfAborted();
    if (pending) {
      last = await matte(pending.raw);
      processed++;
      await write(rgbaOf(last), pending.count);
    }
    encoder.stdin.end();
    const [[dcode], [ecode]] = await Promise.all([decoderDone, encoderDone]);
    if (dcode !== 0 || ecode !== 0) throw new Error(`ffmpeg failed: ${stderr.trim().split('\n').slice(-4).join('\n')}`);
    console.log(`matte: model ran on ${processed} of ${n} frames (${skipped} skipped for low motion, the rest repeated)`);
    return n;
  } catch (e) {
    kill();
    throw e;
  } finally {
    signal.removeEventListener('abort', kill);
  }
}

// ---- remote processing over ssh ----
//
// A single ssh connection carries the whole job: the camera track goes to the remote machine on stdin
// (scaled down here first to the processing size, when that is smaller, so that far less has to be
// uploaded),
// the remote shell script below runs matte-worker.py on it in a temporary directory, and the
// finished cutout.webm comes back on stdout. On first use the script creates a Python environment
// with onnxruntime-gpu in ~/.cache/takes/venv on the remote machine; pip installs the CUDA and cuDNN
// libraries as well, so the remote machine needs only an NVIDIA driver, Python 3.10 or later, and
// ffmpeg (possibly made available by the setup command, e.g. "module load python ffmpeg").
// The remote side reports what it is doing on stderr with "takes:" lines (stage changes, frames
// processed, the start of the upload, and the size of the result), which become the job's stage
// and progress; other lines are logged. Cancelling a job closes the ssh connection; the remote worker then fails on its next progress
// line (a broken pipe), and the script's trap removes the temporary directory.

const WORKER = fs.readFile(new URL('./matte-worker.py', import.meta.url), 'utf8');

const shq = (s: string) => `'${s.replaceAll("'", `'\\''`)}'`;

function remoteScript(setup: string, worker: string) {
  return `{ ${setup || ':'}
} </dev/null
set -e
echo "takes:stage Checking the Python environment on $(hostname)" >&2
base=~/.cache/takes
venv=$base/venv
if [ ! -f "$venv/.ready" ]; then
  python3 -c 'import sys; sys.exit(sys.version_info < (3, 10))' </dev/null || {
    echo "Python 3.10 or later is needed on $(hostname), found: $(python3 --version 2>&1)" >&2; exit 1; }
  echo "takes:stage Installing onnxruntime-gpu on $(hostname) (first use only; about 2.5 GB)" >&2
  rm -rf "$venv"
  mkdir -p "$base"
  python3 -m venv "$venv" </dev/null
  "$venv/bin/pip" install --quiet --disable-pip-version-check 'onnxruntime-gpu[cuda,cudnn]' numpy </dev/null >&2
  touch "$venv/.ready"
fi
command -v ffmpeg >/dev/null || { echo "ffmpeg was not found on $(hostname)" >&2; exit 1; }
work=$(mktemp -d)
trap 'rm -rf "$work"' EXIT
cat > "$work/worker.py" <<'TAKES_WORKER'
${worker}
TAKES_WORKER
echo "takes:upload" >&2
cat > "$work/in.webm"
cd "$work"
"$venv/bin/python" worker.py "$1" </dev/null >&2
echo "takes:result $(wc -c < out.webm)" >&2
cat out.webm
`;
}

async function runRemote(job: Job, { src, out, W, H, ds, total }: Plan, host: string, setup: string): Promise<number> {
  const spec = { modelUrl: MODEL_URL, width: W, height: H, downsampleRatio: ds, decode: decoderArgs('in.webm', W, H), encode: encoderArgs('out.webm', W, H) };
  const command = `bash -lc ${shq(remoteScript(setup, await WORKER))} takes-matte ${shq(JSON.stringify(spec))}`;
  const upload = await uploadSource(job, src, W, H, total / FPS, (f) => {
    if (job.stage === 'Uploading camera track') job.progress = Math.min(1, f);
  });
  // BatchMode makes ssh fail instead of prompting for a password, since there is no terminal.
  const ssh = spawn('ssh', ['-T', '-o', 'BatchMode=yes', '--', host, command], { stdio: ['pipe', 'pipe', 'pipe'] });
  const kill = () => {
    ssh.kill();
    upload.kill();
  };
  job.abort.signal.addEventListener('abort', kill, { once: true });
  job.stage = `Connecting to ${host}`;
  job.progress = undefined;
  let received = 0;
  let resultSize = 0;
  ssh.stdout.on('data', (c: Buffer) => {
    received += c.length;
    if (resultSize) job.progress = Math.min(1, received / resultSize);
  });
  const closed = new Promise<number | null>((resolve, reject) => {
    ssh.on('error', reject);
    ssh.on('close', resolve);
  });
  const lines: string[] = []; // recent diagnostics, for the error message
  let n = 0;
  let partial = '';
  ssh.stderr.setEncoding('utf8');
  ssh.stderr.on('data', (d: string) => {
    const parts = (partial + d).split('\n');
    partial = parts.pop()!;
    for (const line of parts.map((l) => l.trimEnd())) {
      const m = /^takes:(progress|stage|upload|result) ?(.*)$/.exec(line);
      if (m?.[1] === 'progress') {
        n = parseInt(m[2]);
        job.progress = Math.min(0.99, n / total);
      } else if (m?.[1] === 'stage') {
        console.log(`matte (${host}): ${m[2]}`);
        job.stage = m[2];
        job.progress = m[2] === 'Removing background' ? 0 : undefined;
      } else if (m?.[1] === 'upload') {
        job.stage = 'Uploading camera track';
        job.progress = Math.min(1, upload.fraction());
      } else if (m?.[1] === 'result') {
        resultSize = parseInt(m[2]) || 0;
        job.stage = 'Receiving result';
        job.progress = resultSize ? Math.min(1, received / resultSize) : undefined;
      } else if (line.trim()) {
        console.log(`matte (${host}): ${line}`);
        lines.push(line);
        if (lines.length > 20) lines.shift();
      }
    }
  });
  try {
    const [code] = await Promise.all([
      closed,
      pipeline(upload.stream, upload.counter, ssh.stdin).catch(() => {}), // an early exit shows up in the exit code
      pipeline(ssh.stdout, createWriteStream(out)),
      // A failed downscale would otherwise go unnoticed and give a cutout of only part of the take.
      upload.done.catch((e) => {
        ssh.kill();
        throw e;
      }),
    ]);
    if (code !== 0) throw new Error(`Background removal on ${host} failed: ${lines.slice(-6).join('\n') || `ssh exited with code ${code}`}`);
    console.log(`matte (${host}): uploaded ${(upload.bytes() / 1e6).toFixed(1)} MB${upload.scaled ? ` (scaled to ${W}x${H})` : ''}`);
    return n;
  } catch (e) {
    kill();
    throw e;
  } finally {
    job.abort.signal.removeEventListener('abort', kill);
  }
}

// What is uploaded for a remote job: the camera track scaled down and re-encoded to the processing
// size (H.264 at CRF 18, visually lossless: on real camera footage, 0.2 MB per 10 s at 240 lines
// against 3.4 MB for the original 720p track), or the original
// file when no scaling is needed. The progress is the fraction of the take sent so far; ffmpeg can
// only run as far ahead of the upload as the pipe buffers allow.
async function uploadSource(job: Job, src: string, W: number, H: number, duration: number, onProgress: (fraction: number) => void) {
  let bytes = 0;
  let fraction: () => number;
  let stream: Readable;
  let kill = () => {};
  let done: Promise<void> = Promise.resolve();
  const scaled = !!job.camera && H < job.camera.height;
  if (scaled) {
    const ff = spawn(
      'ffmpeg',
      ['-v', 'error', '-nostats', '-progress', 'pipe:2', '-i', src, '-an', '-vf', `fps=${FPS}:start_time=0,scale=${W}:${H}`,
        '-c:v', 'libx264', '-preset', 'veryfast', '-crf', '18', '-pix_fmt', 'yuv420p', '-f', 'matroska', '-'],
      { stdio: ['ignore', 'pipe', 'pipe'] },
    );
    let outTime = 0;
    let errors = '';
    ff.stderr.setEncoding('utf8');
    ff.stderr.on('data', (d: string) => {
      for (const line of d.split('\n')) {
        const m = /^out_time_us=(\d+)/.exec(line);
        if (m) outTime = parseInt(m[1]) / 1e6;
        else if (line && !/^\w+=/.test(line)) errors += line + '\n';
      }
      onProgress(fraction());
    });
    fraction = () => (duration > 0 ? outTime / duration : 0);
    stream = ff.stdout;
    kill = () => ff.kill('SIGKILL');
    done = once(ff, 'close').then(([code]) => {
      if (code !== 0 && !job.abort.signal.aborted) throw new Error(`Could not scale the camera track for upload: ${errors.trim().split('\n').slice(-3).join('\n') || `ffmpeg exited with code ${code}`}`);
    });
  } else {
    const size = (await fs.stat(src)).size;
    fraction = () => bytes / size;
    stream = createReadStream(src);
  }
  const counter = new Transform({
    transform(chunk: Buffer, _enc, cb) {
      bytes += chunk.length;
      if (!scaled) onProgress(fraction());
      cb(null, chunk);
    },
  });
  return { stream, counter, scaled, done, kill, fraction: () => fraction(), bytes: () => bytes };
}