galleria_photo_face_rknn_openCL/api_v1/scanner/faceScanWorker.js

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

const {
  scanPhotoFacesById,
} = require('./faceScanner');

function serializeError(error) {
  const value =
    error?.stack
    || error?.message
    || String(error);

  return String(value).slice(0, 16000);
}

async function claimNextPendingJob() {
  return db.transaction(async (trx) => {
    const job = await trx('face_scan_jobs')
      .where({ status: 'pending' })
      .orderBy('created_at', 'asc')
      .orderBy('photo_id', 'asc')
      .first();

    if (!job) {
      return null;
    }

    const updated = await trx('face_scan_jobs')
      .where({
        photo_id: job.photo_id,
        status: 'pending',
      })
      .update({
        status: 'processing',
        attempts: trx.raw('attempts + 1'),
        faces_detected: null,
        execution_ms: null,
        applied_rotation: null,
        last_error: null,
        started_at: trx.fn.now(),
        completed_at: null,
        updated_at: trx.fn.now(),
      });

    if (updated !== 1) {
      return null;
    }

    return trx('face_scan_jobs')
      .where({ photo_id: job.photo_id })
      .first();
  });
}

async function completeJob(photoId, result) {
  return db.transaction(async (trx) => {
    const job = await trx('face_scan_jobs')
      .where({
        photo_id: photoId,
        status: 'processing',
      })
      .first();

    if (!job) {
      throw new Error(
        `Job processing non trovato: photo_id=${photoId}`,
      );
    }

    if (Number(job.rescan_requested) === 1) {
      const updated = await trx('face_scan_jobs')
        .where({
          photo_id: photoId,
          status: 'processing',
        })
        .update({
          status: 'pending',
          rescan_requested: 0,
          faces_detected: null,
          execution_ms: null,
          applied_rotation: null,
          last_error: null,
          started_at: null,
          completed_at: null,
          updated_at: trx.fn.now(),
        });

      if (updated !== 1) {
        throw new Error(
          `Impossibile riaccodare il job photo_id=${photoId}`,
        );
      }

      return 'pending';
    }

    const updated = await trx('face_scan_jobs')
      .where({
        photo_id: photoId,
        status: 'processing',
      })
      .update({
        status: 'done',
        rescan_requested: 0,
        faces_detected: result.faces_detected,
        execution_ms: result.execution_ms,
        applied_rotation: result.applied_rotation,
        last_error: null,
        completed_at: trx.fn.now(),
        updated_at: trx.fn.now(),
      });

    if (updated !== 1) {
      throw new Error(
        `Impossibile completare il job photo_id=${photoId}`,
      );
    }

    return 'done';
  });
}

async function failJob(photoId, error) {
  return db.transaction(async (trx) => {
    const job = await trx('face_scan_jobs')
      .where({
        photo_id: photoId,
        status: 'processing',
      })
      .first();

    if (!job) {
      console.error(
        '[faceScanWorker] job processing non trovato:',
        photoId,
      );

      return 'missing';
    }

    if (Number(job.rescan_requested) === 1) {
      const updated = await trx('face_scan_jobs')
        .where({
          photo_id: photoId,
          status: 'processing',
        })
        .update({
          status: 'pending',
          rescan_requested: 0,
          faces_detected: null,
          execution_ms: null,
          applied_rotation: null,
          last_error: null,
          started_at: null,
          completed_at: null,
          updated_at: trx.fn.now(),
        });

      if (updated !== 1) {
        throw new Error(
          `Impossibile riaccodare il job photo_id=${photoId}`,
        );
      }

      return 'pending';
    }

    const updated = await trx('face_scan_jobs')
      .where({
        photo_id: photoId,
        status: 'processing',
      })
      .update({
        status: 'error',
        rescan_requested: 0,
        last_error: serializeError(error),
        completed_at: trx.fn.now(),
        updated_at: trx.fn.now(),
      });

    if (updated !== 1) {
      console.error(
        '[faceScanWorker] impossibile registrare errore:',
        photoId,
      );

      return 'missing';
    }

    return 'error';
  });
}

async function processClaimedJob(job) {
  if (!job || !job.photo_id) {
    throw new Error('Job non valido');
  }

  const workerStartedAt = Date.now();

  console.log(
    '[faceScanWorker] processing',
    {
      photo_id: job.photo_id,
      attempt: job.attempts,
    },
  );

  try {
    const result = await scanPhotoFacesById(
      job.photo_id,
    );

    const finalStatus = await completeJob(
      job.photo_id,
      result,
    );

    const summary = {
      status: finalStatus,
      photo_id: job.photo_id,
      attempt: job.attempts,
      faces_detected:
        finalStatus === 'done'
          ? result.faces_detected
          : null,
      applied_rotation:
        finalStatus === 'done'
          ? result.applied_rotation
          : null,
      execution_ms:
        finalStatus === 'done'
          ? result.execution_ms
          : null,
      rescan_required:
        finalStatus === 'pending',
      worker_ms: Date.now() - workerStartedAt,
    };

    console.log(
      '[faceScanWorker] done',
      summary,
    );

    return summary;
  } catch (error) {
    const finalStatus = await failJob(
      job.photo_id,
      error,
    );

    const summary = {
      status: finalStatus,
      photo_id: job.photo_id,
      attempt: job.attempts,
      error:
        finalStatus === 'error'
          ? serializeError(error)
          : null,
      rescan_required:
        finalStatus === 'pending',
      worker_ms: Date.now() - workerStartedAt,
    };

    console.error(
      '[faceScanWorker] error',
      summary,
    );

    return summary;
  }
}

async function processNextJob() {
  const job = await claimNextPendingJob();

  if (!job) {
    return null;
  }

  return processClaimedJob(job);
}

async function runWorker(options = {}) {
  const maxJobs =
    options.maxJobs == null
      ? null
      : Number(options.maxJobs);

  if (
    maxJobs !== null
    && (
      !Number.isInteger(maxJobs)
      || maxJobs <= 0
    )
  ) {
    throw new Error(
      'maxJobs deve essere un intero positivo',
    );
  }

  const shouldStop =
    typeof options.shouldStop === 'function'
      ? options.shouldStop
      : () => false;

  const results = [];

  while (
    !shouldStop()
    && (
      maxJobs === null
      || results.length < maxJobs
    )
  ) {
    const result = await processNextJob();

    if (!result) {
      break;
    }

    results.push(result);
  }

  const succeeded = results.filter(
    (result) => result.status === 'done',
  ).length;

  const failed = results.filter(
    (result) => result.status === 'error',
  ).length;

  const rescheduled = results.filter(
    (result) => result.status === 'pending',
  ).length;

  return {
    jobs_processed: results.length,
    jobs_succeeded: succeeded,
    jobs_failed: failed,
    jobs_rescheduled: rescheduled,
    stopped: shouldStop(),
    results,
  };
}

async function resetStaleProcessingJobs(
  staleMinutes = 30,
) {
  const minutes = Number(staleMinutes);

  if (
    !Number.isFinite(minutes)
    || minutes <= 0
  ) {
    throw new Error(
      'staleMinutes deve essere positivo',
    );
  }

  const modifier =
    `-${Math.floor(minutes)} minutes`;

  const updated = await db('face_scan_jobs')
    .where({ status: 'processing' })
    .where(
      'started_at',
      '<',
      db.raw(
        "datetime('now', ?)",
        [modifier],
      ),
    )
    .update({
      status: 'pending',
      last_error:
        'Job recuperato dopo arresto del worker',
      started_at: null,
      completed_at: null,
      updated_at: db.fn.now(),
    });

  return updated;
}

module.exports = {
  serializeError,
  claimNextPendingJob,
  completeJob,
  failJob,
  processClaimedJob,
  processNextJob,
  runWorker,
  resetStaleProcessingJobs,
};