galleria_photo_face_rknn_openCL/tools/runGeocodeDaemon.js

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

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

const {
  processNextJob,
  resetStaleProcessingJobs,
} = require('../api_v1/scanner/geocodeWorker');

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

const DEFAULT_LOCK_PATH =
  path.join(PROJECT_ROOT, 'geocode-daemon.lock');

function parsePositiveInteger(
  value,
  optionName,
) {
  const number = Number(value);

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

  return number;
}

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

  for (
    let index = 0;
    index < argv.length;
    index += 1
  ) {
    const argument = argv[index];
    const value = argv[index + 1];

    if (argument === '--poll-ms') {
      if (value == null) {
        throw new Error(
          'Valore mancante dopo --poll-ms',
        );
      }

      options.pollMs = parsePositiveInteger(
        value,
        '--poll-ms',
      );

      index += 1;
      continue;
    }

    if (
      argument
      === '--reset-stale-minutes'
    ) {
      if (value == null) {
        throw new Error(
          'Valore mancante dopo '
          + '--reset-stale-minutes',
        );
      }

      options.resetStaleMinutes =
        parsePositiveInteger(
          value,
          '--reset-stale-minutes',
        );

      index += 1;
      continue;
    }

    if (argument === '--stale-check-ms') {
      if (value == null) {
        throw new Error(
          'Valore mancante dopo '
          + '--stale-check-ms',
        );
      }

      options.staleCheckMs =
        parsePositiveInteger(
          value,
          '--stale-check-ms',
        );

      index += 1;
      continue;
    }

    if (argument === '--lock-file') {
      if (!value) {
        throw new Error(
          'Valore mancante dopo --lock-file',
        );
      }

      options.lockPath = path.resolve(value);
      index += 1;
      continue;
    }

    throw new Error(
      `Argomento sconosciuto: ${argument}`,
    );
  }

  return options;
}

function wait(milliseconds) {
  return new Promise((resolve) => {
    setTimeout(resolve, milliseconds);
  });
}

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 {
    const text = fs.readFileSync(
      lockPath,
      'utf8',
    );

    return JSON.parse(text);
  } catch {
    return null;
  }
}

function acquireLock(lockPath) {
  const lockDirectory =
    path.dirname(lockPath);

  fs.mkdirSync(
    lockDirectory,
    { recursive: true },
  );

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

    const lockData = {
      pid: process.pid,
      started_at:
        new Date().toISOString(),
      project_root: PROJECT_ROOT,
    };

    fs.writeFileSync(
      descriptor,
      JSON.stringify(lockData, null, 2)
        + '\n',
      'utf8',
    );

    fs.closeSync(descriptor);

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

  const existingLock = readLock(lockPath);
  const existingPid =
    Number(existingLock?.pid);

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

  console.warn(
    '[geocodeDaemon] rimuovo lock obsoleto:',
    {
      lock_path: lockPath,
      previous_pid:
        existingLock?.pid ?? null,
    },
  );

  fs.unlinkSync(lockPath);

  acquireLock(lockPath);
}

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

  if (
    lock
    && Number(lock.pid) !== process.pid
  ) {
    console.warn(
      '[geocodeDaemon] lock appartenente '
      + 'a un altro processo, non rimosso:',
      lockPath,
    );

    return;
  }

  try {
    fs.unlinkSync(lockPath);
  } catch (error) {
    if (error.code !== 'ENOENT') {
      console.warn(
        '[geocodeDaemon] impossibile '
        + 'rimuovere il lock:',
        error.message,
      );
    }
  }
}

async function getQueueStatus() {
  const rows = await db('geocode_jobs')
    .select('status')
    .count({ count: '*' })
    .groupBy('status');

  const status = {
    pending: 0,
    processing: 0,
    done: 0,
    error: 0,
  };

  for (const row of rows) {
    status[row.status] =
      Number(row.count || 0);
  }

  return status;
}

async function main() {
  const options = parseArguments(
    process.argv.slice(2),
  );

  let stopRequested = false;
  let lastStaleCheck = 0;

  acquireLock(options.lockPath);

  const requestStop = (signal) => {
    if (stopRequested) {
      console.log(
        `[geocodeDaemon] ${signal}: `
        + 'arresto già richiesto',
      );

      return;
    }

    stopRequested = true;

    console.log(
      `[geocodeDaemon] ${signal}: `
      + 'arresto dopo il job corrente',
    );
  };

  process.on(
    'SIGINT',
    () => requestStop('SIGINT'),
  );

  process.on(
    'SIGTERM',
    () => requestStop('SIGTERM'),
  );

  const initialReset =
    await resetStaleProcessingJobs(
      options.resetStaleMinutes,
    );

  lastStaleCheck = Date.now();

  console.log(
    '[geocodeDaemon] avviato',
    {
      pid: process.pid,
      project_root: PROJECT_ROOT,
      poll_ms: options.pollMs,
      stale_check_ms:
        options.staleCheckMs,
      reset_stale_minutes:
        options.resetStaleMinutes,
      stale_jobs_reset: initialReset,
      lock_path: options.lockPath,
      queue: await getQueueStatus(),
    },
  );

  let processed = 0;
  let succeeded = 0;
  let failed = 0;
  let rescheduled = 0;

  while (!stopRequested) {
    const now = Date.now();

    if (
      now - lastStaleCheck
      >= options.staleCheckMs
    ) {
      const resetJobs =
        await resetStaleProcessingJobs(
          options.resetStaleMinutes,
        );

      lastStaleCheck = now;

      if (resetJobs > 0) {
        console.log(
          '[geocodeDaemon] job obsoleti '
          + 'ripristinati:',
          resetJobs,
        );
      }
    }

    const result = await processNextJob();

    if (result) {
      processed += 1;

      if (result.status === 'done') {
        succeeded += 1;
      } else if (
        result.status === 'error'
      ) {
        failed += 1;
      } else if (
        result.status === 'pending'
      ) {
        rescheduled += 1;
      }

      continue;
    }

    await wait(options.pollMs);
  }

  console.log(
    '[geocodeDaemon] terminato',
    {
      pid: process.pid,
      jobs_processed: processed,
      jobs_succeeded: succeeded,
      jobs_failed: failed,
      jobs_rescheduled: rescheduled,
      queue: await getQueueStatus(),
    },
  );
}

let lockPath = null;

(async () => {
  const options = parseArguments(
    process.argv.slice(2),
  );

  lockPath = options.lockPath;

  await main();
})()
  .catch((error) => {
    console.error(
      '[geocodeDaemon] errore fatale:',
      error.stack
        || error.message
        || error,
    );

    process.exitCode = 1;
  })
  .finally(async () => {
    if (lockPath) {
      releaseLock(lockPath);
    }

    await db.destroy();
  });