'use strict';
// Both WinSW and interactive validation use this supervisor. Only official IPC
// drains the runner; loss of admission never kills an in-flight video job.
const fs = require('node:fs');
const path = require('node:path');
const { spawn, execFile } = require('node:child_process');
const root = process.env.PT_ROOT || 'C:\\pt-worker';
const state = path.join(root, 'state');
const cfg = JSON.parse(fs.readFileSync(path.join(state, 'worker.json'), 'utf8').replace(/^\uFEFF/, ''));
const env = { ...process.env, PT_ROOT: root, PT_STATE_DIR: state,
  PT_WORKER_CONFIG: path.join(state, 'worker.json'),
  APPDATA: path.join(root, 'profile', 'Roaming'), LOCALAPPDATA: path.join(root, 'profile', 'Local'),
  TMP: path.join(root, 'tmp'), TEMP: path.join(root, 'tmp'),
  FFMPEG_PATH: path.join(root, 'bin', 'ffmpeg-nvenc.exe'),
  FFPROBE_PATH: path.join(root, 'bin', 'ffprobe.exe'), PT_NVENC_DEVICE: String(cfg.nvenc_device || 0) };
let child = null, draining = false, asking = false, stopping = false, restartAfter = 0;
function read(name) { try { return JSON.parse(fs.readFileSync(path.join(state, name), 'utf8').replace(/^\uFEFF/, '')); } catch { return {}; } }
function emit(data) { fs.appendFileSync(path.join(state, 'events.jsonl'), JSON.stringify({ time: Date.now()/1000, ...data }) + '\n'); }
function atomic(name, data) { const file = path.join(state, name), tmp = file+'.'+process.pid+'.tmp'; fs.writeFileSync(tmp, JSON.stringify(data)); fs.renameSync(tmp, file); }
function drain() {
  if (!child || asking || draining) return;
  const target = child;
  asking = true;
  execFile(cfg.node_binary, [path.join(root, 'bin', 'ipc-nvenc.cjs'), 'graceful-shutdown'],
    { env, timeout: 17000, windowsHide: true }, err => { asking = false; if (!err && child === target) draining = true; });
}
function tick() {
  if (fs.existsSync(path.join(state, 'supervisor-stop'))) stopping = true;
  const now = Date.now()/1000, admission = read('status.json'), control = read('control.json');
  const allow = !stopping && control.paused === false && admission.admission_allowed === true &&
    now - admission.updated_at < 30 && ['AVAILABLE', 'BUSY'].includes(admission.status) &&
    !fs.existsSync(path.join(state, 'gpu-unavailable'));
  if (child && !allow) drain();
  if (!child && allow && now >= restartAfter) {
    draining = false;
    child = spawn(cfg.node_binary, ['--require', path.join(root, 'bin', 'observe.cjs'),
      path.join(root, 'runner', 'node_modules', '@peertube', 'peertube-runner', 'dist', 'peertube-runner.js'),
      'server', '--enable-job', 'vod-web-video-transcoding', '--enable-job', 'vod-hls-transcoding',
      '--enable-job', 'vod-audio-merge-transcoding'], { env, windowsHide: true, stdio: ['ignore', 'pipe', 'pipe'] });
    // Upstream logs can contain tokens. Retain only known safe milestones.
    let buffer = '';
    for (const stream of [child.stdout, child.stderr]) stream.on('data', chunk => {
      buffer += chunk.toString();
      const lines = buffer.split(/\r?\n/); buffer = lines.pop().slice(-8192);
      for (const line of lines) {
        const match = line.match(/(Finished processing|Processing) job of type ([\w-]+): ([a-f\d-]{36})/);
        if (match) emit({ kind: 'runner_milestone', phase: match[1].startsWith('Finished') ? 'completed' : 'processing', job_type: match[2], uuid: match[3] });
        if (line.includes('Connected to ') && line.includes('/runners')) emit({kind: 'runner_connected'});
        if (line.includes('Cannot process job')) emit({kind: 'runner_error', reason: 'process_job_failed'});
      }
    });
    child.once('error', () => { emit({kind: 'runner_error', reason: 'spawn_failed'}); });
    child.once('exit', code => {
      emit({kind: 'runner_exit', code, graceful_requested: draining || asking});
      child = null; draining = false; asking = false; restartAfter = Date.now()/1000 + 15;
    });
    emit({kind: 'runner_start', concurrency: cfg.concurrency});
  }
  atomic('runner.json', {running: !!child, draining, updated_at: now, pid: child?.pid || null});
  if (stopping && !child) process.exit(0);
}
fs.rmSync(path.join(state, 'supervisor-stop'), {force: true});
process.on('SIGINT', () => { stopping = true; drain(); });
process.on('SIGTERM', () => { stopping = true; drain(); });
setInterval(tick, 1000); tick();