gallery_server_client_s190js/tools/runActivityIngestDaemon.js

'use strict';

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

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

const {
  ensureActivityIngestQueue,
  resetStaleActivityIngestJobs,
  getActivityIngestQueueCounts,
} = require(
  '../api_v1/scanner/activityIngestQueue'
);

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

const PROJECT_ROOT = path.resolve(
  __dirname,
  '..'
);

const DEFAULT_LOCK_PATH = path.join(
  PROJECT_ROOT,
  'activity-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.ACTIVITY_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 gia attivo con PID ${pid}. ` +
      `Lock: ${lockPath}`
    );
  }

  if (
    !existing ||
    !Number.isInteger(pid) ||
    pid <= 0
  ) {
    throw new Error(
      `Lock esistente non verificabile: ` +
      `${lockPath}`
    );
  }

  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(
        '[ActivityIngestDaemon] ' +
        '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/runActivityIngestDaemon.js [opzioni]\n' +
      '  --dry-run\n' +
      '  --poll-ms N\n' +
      '  --reset-stale-minutes N\n' +
      '  --stale-check-ms N\n' +
      '  --lock-file PERCORSO'
    );
    return;
  }

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

  const requestStop = signal => {
    if (!stopping) {
      console.log(
        `[ActivityIngestDaemon] ` +
        `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 ensureActivityIngestQueue(db);

    const recovered =
      options.dryRun
        ? 0
        : await resetStaleActivityIngestJobs(
            db,
            options.resetStaleMinutes
          );

    lastStaleCheck = Date.now();

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

    while (!stopping) {
      try {
        if (options.dryRun) {
          await sleep(options.pollMs);
          continue;
        }

        const result =
          await runOneJob(db);

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

        /*
         * Come Photo daemon:
         * se ha processato un job, prova subito
         * il successivo. Polling solo a coda vuota.
         */
        if (
          !result.processed &&
          !stopping
        ) {
          await sleep(options.pollMs);
        }

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

          lastStaleCheck = Date.now();

          if (count) {
            console.warn(
              `[ActivityIngestDaemon] ` +
              `Recuperati ${count} job bloccati.`
            );
          }
        }
      } catch (error) {
        console.error(
          '[ActivityIngestDaemon] 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(
      '[ActivityIngestDaemon] Errore fatale:',
      error.message
    );

    process.exitCode = 1;
  } finally {
    await db.destroy();
  }
})();