gallery_server_client_s190js/tools/runPhotoIngestDaemon.js

'use strict';

const fs = require('fs');
const path = require('path');

const db = require('../db/knex');

const {
  ensurePhotoIngestQueue,
  resetStalePhotoIngestJobs,
  getPhotoIngestQueueCounts,
} = require('../api_v1/scanner/photoIngestQueue');

const { runOneJob } = require('../api_v1/scanner/photoIngestWorker');

const PROJECT_ROOT = path.resolve(__dirname, '..');
const DEFAULT_LOCK_PATH = path.join(
  PROJECT_ROOT,
  'photo-ingest-daemon.lock'
);

function positiveInteger(value, option) {
  const number = Number(value);

  if (!Number.isInteger(number) || number <= 0) {
    throw new Error(`${option} deve essere un intero positivo`);
  }

  return number;
}

function parseArguments(argv) {
  const options = {
    pollMs: 2000,
    resetStaleMinutes: 30,
    staleCheckMs: 60000,
    lockPath:
      process.env.PHOTO_INGEST_DAEMON_LOCK || DEFAULT_LOCK_PATH,
    help: false,
    dryRun: false,
  };

  for (let i = 0; i < argv.length; i += 1) {
    const arg = argv[i];

    if (arg === '--help' || arg === '-h') {
      options.help = true;
      continue;
    }

    if (arg === '--dry-run') {
      options.dryRun = true;
      continue;
    }

    const names = {
      '--poll-ms': 'pollMs',
      '--reset-stale-minutes': 'resetStaleMinutes',
      '--stale-check-ms': 'staleCheckMs',
      '--lock-file': 'lockPath',
    };

    const key = names[arg];

    if (!key) {
      throw new Error(`Argomento sconosciuto: ${arg}`);
    }

    const value = argv[i + 1];

    if (value == null || value.startsWith('--')) {
      throw new Error(`Valore mancante dopo ${arg}`);
    }

    options[key] = key === 'lockPath'
      ? path.resolve(value)
      : positiveInteger(value, arg);

    i += 1;
  }

  return options;
}

function isProcessAlive(pid) {
  if (!Number.isInteger(pid) || pid <= 0) return false;

  try {
    process.kill(pid, 0);
    return true;
  } catch (error) {
    return error.code === 'EPERM';
  }
}

function readLock(lockPath) {
  try {
    return JSON.parse(fs.readFileSync(lockPath, 'utf8'));
  } catch {
    return null;
  }
}

function acquireLock(lockPath) {
  fs.mkdirSync(path.dirname(lockPath), { recursive: true });

  try {
    const fd = fs.openSync(lockPath, 'wx');

    try {
      fs.writeFileSync(fd, JSON.stringify({
        pid: process.pid,
        started_at: new Date().toISOString(),
        project_root: PROJECT_ROOT,
      }, null, 2) + '\n', 'utf8');
    } finally {
      fs.closeSync(fd);
    }

    return;
  } catch (error) {
    if (error.code !== 'EEXIST') throw error;
  }

  const existing = readLock(lockPath);
  const pid = Number(existing?.pid);

  if (isProcessAlive(pid)) {
    throw new Error(
      `Daemon già attivo con PID ${pid}. Lock: ${lockPath}`
    );
  }

  // Non cancellare automaticamente un lock illeggibile:
  // potrebbe appartenere a un processo in fase di avvio.
  if (!existing || !Number.isInteger(pid) || pid <= 0) {
    throw new Error(
      `Lock esistente non verificabile: ${lockPath}. `
      + 'Verificarlo manualmente prima di rimuoverlo.'
    );
  }

  fs.unlinkSync(lockPath);
  acquireLock(lockPath);
}

function releaseLock(lockPath) {
  const lock = readLock(lockPath);

  if (Number(lock?.pid) !== process.pid) return;

  try {
    fs.unlinkSync(lockPath);
  } catch (error) {
    if (error.code !== 'ENOENT') {
      console.error('[PhotoIngestDaemon] Errore rimozione lock:', error.message);
    }
  }
}

function sleep(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

async function main(options) {
  if (options.help) {
    console.log(
      'Uso: node tools/runPhotoIngestDaemon.js [opzioni]\n'
      + '  --dry-run                  Test senza elaborare o recuperare job\n'
      + '  --poll-ms N                 Attesa tra controlli (default 2000)\n'
      + '  --reset-stale-minutes N     Età job bloccati (default 30)\n'
      + '  --stale-check-ms N          Intervallo recupero (default 60000)\n'
      + '  --lock-file PERCORSO        Percorso lock personalizzato'
    );
    return;
  }

  let stopping = false;
  let lockAcquired = false;
  let lastStaleCheck = 0;

  const requestStop = signal => {
    if (!stopping) {
      console.log(`[PhotoIngestDaemon] Ricevuto ${signal}; arresto richiesto.`);
      stopping = true;
    }
  };

  const onSigint = () => requestStop('SIGINT');
  const onSigterm = () => requestStop('SIGTERM');

  process.on('SIGINT', onSigint);
  process.on('SIGTERM', onSigterm);

  try {
    acquireLock(options.lockPath);
    lockAcquired = true;

    await ensurePhotoIngestQueue(db);

    // Recupero iniziale eseguito soltanto dopo l'acquisizione del lock.
    const recovered = options.dryRun
      ? 0
      : await resetStalePhotoIngestJobs(
          db,
          options.resetStaleMinutes
        );

    lastStaleCheck = Date.now();

    console.log('[PhotoIngestDaemon] Avviato.', {
      pid: process.pid,
      pollMs: options.pollMs,
      staleCheckMs: options.staleCheckMs,
      dryRun: options.dryRun,
      recovered,
      queue: await getPhotoIngestQueueCounts(db),
    });

    while (!stopping) {
      try {
        // Durante un job non eseguiamo un recupero concorrente.
        if (options.dryRun) {
          await sleep(options.pollMs);
          continue;
        }

        const result = await runOneJob(db);

        if (result.error) {
          console.error('[PhotoIngestDaemon] Job fallito:', result.error);
        }

        if (!result.processed && !stopping) {
          await sleep(options.pollMs);
        }

        if (
          !stopping &&
          Date.now() - lastStaleCheck >= options.staleCheckMs
        ) {
          const count = await resetStalePhotoIngestJobs(
            db,
            options.resetStaleMinutes
          );

          lastStaleCheck = Date.now();

          if (count) {
            console.warn(
              `[PhotoIngestDaemon] Recuperati ${count} job bloccati.`
            );
          }
        }
      } catch (error) {
        console.error('[PhotoIngestDaemon] Errore ciclo:', error);
        if (!stopping) await sleep(options.pollMs);
      }
    }
  } finally {
    process.removeListener('SIGINT', onSigint);
    process.removeListener('SIGTERM', onSigterm);

    if (lockAcquired) releaseLock(options.lockPath);
  }
}

(async () => {
  let options;

  try {
    options = parseArguments(process.argv.slice(2));
    await main(options);
  } catch (error) {
    console.error('[PhotoIngestDaemon] Errore fatale:', error.message);
    process.exitCode = 1;
  } finally {
    await db.destroy();
  }
})();