TDarr-Auto-Requeue/tdarr_requeue.mjs

#!/usr/bin/env node
/**
 * tdarr_requeue.mjs – v17 (2025-07-13)
 *
 * • Runs continuously; wakes up every TDARR_INTERVAL_MIN minutes.
 * • If staging count < TDARR_STAGING_LIMIT:
 *      – Pull up to TDARR_BATCH_SIZE items from Status ▸ “Processed” (table2)
 *        where newSize == '-'.
 *      – Re-queue them via a single /bulk-update-files call
 *        (same payload as the Tdarr GUI).
 *
 * All behaviour is configured via .env variables – see README.
 */

import 'dotenv/config';
import axios    from 'axios';
import pino     from 'pino';
import minimist from 'minimist';

/* ────────────────────────────────────────────────────────────
 *  Environment / CLI
 * ─────────────────────────────────────────────────────────── */
const cli            = minimist(process.argv.slice(2));
const API_BASE       = (cli.url || process.env.TDARR_URL || 'http://localhost:8265/api/v2').replace(/\/?$/, '/');
const API_KEY        = (cli['api-key'] || process.env.TDARR_API_KEY || '').trim();

const STAGING_LIMIT  = +process.env.TDARR_STAGING_LIMIT  || 50;
const BATCH_SIZE     = +process.env.TDARR_BATCH_SIZE     || 50;
const INTERVAL_MIN   = +process.env.TDARR_INTERVAL_MIN   || 60;     // scheduler interval
const RETRIES        = +process.env.TDARR_RETRIES        || 4;
const BACKOFF_MS     = +process.env.TDARR_BACKOFF_MS     || 2_000;
const BULK_TIMEOUT_MS= +process.env.BULK_TIMEOUT_MS      || 120_000; // generous timeout

if (!API_KEY) {
  console.error('❌  TDARR_API_KEY is missing'); process.exit(20);
}

/* ─── Logger ----------------------------------------------------------- */
const log = pino({
  level: process.env.LOG_LEVEL ?? 'info',
  transport: process.env.LOG_PRETTY === '1'
    ? { target: 'pino-pretty', options: { translateTime: 'SYS:standard', colorize: true } }
    : undefined
});

/* ─── Axios client ----------------------------------------------------- */
const http = axios.create({
  baseURL: API_BASE,
  timeout: 30_000,
  headers: { 'content-type': 'application/json', 'x-api-key': API_KEY }
});
http.interceptors.response.use(r => r, e => {
  if ([401, 403].includes(e.response?.status)) {
    log.error('🔑  Authentication failed – check TDARR_API_KEY');
    process.exit(20);
  }
  return Promise.reject(e);
});

/* ─── API helper with retry & back-off ------------------------------- */
async function api(endpoint, payload, tries = RETRIES) {
  for (let attempt = 1; attempt <= tries; attempt++) {
    try {
      const { data } = await http.post(endpoint, { data: payload });
      return data.data ?? data;        // Tdarr wraps its result in {data: …}
    } catch (err) {
      log.warn({ endpoint, attempt, status: err.response?.status, body: err.response?.data },
               '⏳  retrying…');
      if (attempt === tries) throw err;
      await new Promise(r => setTimeout(r, BACKOFF_MS * 2 ** attempt));
    }
  }
}

/* ─── Tdarr helpers ---------------------------------------------------- */
const stagingCount = () =>
  api('cruddb', { collection: 'StagedJSONDB', mode: 'getAll' })
    .then(arr => arr.length || 0);

const candidatePaths = () =>
  api('client/status-tables', {
    start: 0,
    pageSize: BATCH_SIZE,
    filters: [{ id: 'newSize', value: '-' }],
    sorts:   [],
    opts:    { table: 'table2' }     // “Processed” table
  }).then(res => Array.isArray(res.array) ? res.array.map(f => f._id) : []);

/* ─── Robust re-queue -------------------------------------------------- */
async function requeue(paths) {
  if (!paths.length) {
    log.warn('⚠️  No candidates found');
    return 0;
  }

  const payload = {
    fileIds:   paths,
    updatedObj:{ TranscodeDecisionMaker: 'Queued' }
  };
  log.debug({ payload }, '📨  bulk-update-files payload');

  const before = await stagingCount();

  try {
    await http.post('bulk-update-files', { data: payload }, { timeout: BULK_TIMEOUT_MS });
  } catch (err) {
    if (err.code === 'ECONNABORTED') {
      log.warn('⌛ bulk-update-files timed out – verifying outcome …');
      const after = await stagingCount();
      if (after > before) {
        log.info({ before, after }, '✅  bulk-update-files succeeded despite timeout');
        return paths.length;
      }
    }
    throw err;  // propagate any other failure or verification miss
  }
  return paths.length;
}

/* ─── One processing cycle -------------------------------------------- */
async function runCycle() {
  log.info('🚀  Requeue cycle started');

  const before = await stagingCount();
  log.info({ before }, '📦  items currently in staging');

  if (before >= STAGING_LIMIT) {
    log.info('🟢  staging limit reached – nothing to do');
    return;
  }

  const paths = await candidatePaths();
  log.debug({ sample: paths.slice(0, 3) }, '🔍  example candidates');

  const queued = await requeue(paths);
  log.info({ queued }, '✅  files requeued');

  const after = await stagingCount();
  log.info({ after }, '📈  staging count after requeue');
}

/* ─── Built-in scheduler ---------------------------------------------- */
(async () => {
  log.info(`📅  Scheduler running every ${INTERVAL_MIN} minute(s)`);
  while (true) {
    try { await runCycle(); }
    catch (err) { log.error({ err }, '💥  cycle failed'); }

    await new Promise(r => setTimeout(r, INTERVAL_MIN * 60_000));
  }
})();