gallery_server_client_s190js/api_v1/scanner/activityIngestWorker.js

'use strict';

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

const {
  WEB_ROOT,
  SUPPORTED_ACTIVITY_EXTS,
  SEND_PHOTOS,
  BASE_URL,
} = require('../config');

const scanActivitySingle =
  require('./scanActivitySingle');

const postWithAuth =
  require('./postWithAuth');

const { sha256 } = require('./utils');

const {
  enqueueServerEvent,
} = require('../../db/serverEventOutbox');

const {
  ensureActivityIngestQueue,
  claimNextActivityIngestJob,
  enqueueActivityIngestJob,
  completeActivityIngestJob,
  failActivityIngestJob,
  resetStaleActivityIngestJobs,
} = require('./activityIngestQueue');

function sameVersion(stat, job) {
  return (
    Math.trunc(stat.mtimeMs) ===
      Math.trunc(Number(job.source_mtime_ms)) &&
    stat.size ===
      Number(job.source_size_bytes)
  );
}

function isInside(parent, child) {
  const relative = path.relative(parent, child);

  return (
    relative === '' ||
    (
      !relative.startsWith(`..${path.sep}`) &&
      relative !== '..' &&
      !path.isAbsolute(relative)
    )
  );
}

async function resolveActivityJobFile(job) {
  for (const value of [
    job.user,
    job.cartella,
    job.rel_path,
  ]) {
    if (
      typeof value !== 'string' ||
      !value.trim()
    ) {
      throw new Error(
        'Percorso activity job non valido'
      );
    }
  }

  if (
    path.isAbsolute(job.rel_path) ||
    job.rel_path.includes('\\') ||
    job.rel_path
      .split('/')
      .some(part =>
        !part ||
        part === '.' ||
        part === '..'
      )
  ) {
    throw new Error(
      'Percorso relativo activity non sicuro'
    );
  }

  if (
    job.user === '.' ||
    job.user === '..' ||
    job.cartella === '.' ||
    job.cartella === '..' ||
    job.user.includes('/') ||
    job.user.includes('\\') ||
    job.cartella.includes('/') ||
    job.cartella.includes('\\')
  ) {
    throw new Error(
      'Utente o cartella activity non validi'
    );
  }

  const activitiesRoot = path.resolve(
    __dirname,
    '..',
    '..',
    WEB_ROOT,
    'activities'
  );

  const originalRoot = path.resolve(
    activitiesRoot,
    job.user,
    'original',
    job.cartella
  );

  const absPath = path.resolve(
    originalRoot,
    ...job.rel_path.split('/')
  );

  if (!isInside(originalRoot, absPath)) {
    throw new Error(
      'Activity file esce dalla cartella prevista'
    );
  }

  const realRoot = await fs.realpath(
    originalRoot
  );

  const realFile = await fs.realpath(
    absPath
  );

  if (!isInside(realRoot, realFile)) {
    throw new Error(
      'Activity file risolve fuori dalla cartella prevista'
    );
  }

  const ext = path
    .extname(absPath)
    .toLowerCase();

  if (!SUPPORTED_ACTIVITY_EXTS.has(ext)) {
    throw new Error(
      `Estensione activity non supportata: ${ext}`
    );
  }

  const stat = await fs.stat(realFile);

  if (!stat.isFile()) {
    throw new Error(
      'Activity path non e un file regolare'
    );
  }

  return {
    id: sha256(
      `${job.user}/${job.cartella}/${job.rel_path}`
    ),
    user: job.user,
    cartella: job.cartella,
    file_name: path.basename(absPath),
    name: path.basename(absPath),
    relPath: job.rel_path,
    absPath: realFile,
    ext,
    stat,
    path:
      `/activities/${job.user}/original/` +
      `${job.cartella}/${job.rel_path}`,
  };
}

async function runOneJob(db) {
  await ensureActivityIngestQueue(db);

  const job =
    await claimNextActivityIngestJob(db);

  if (!job) {
    return { processed: false };
  }

  console.log(
    `[ActivityIngest] Avvio job ${job.id}: ` +
    `${job.rel_path}`
  );

  try {
    const file =
      await resolveActivityJobFile(job);

    /*
     * Se il file e' gia cambiato prima di iniziare
     * l'elaborazione, aggiorna la versione in queue.
     */
    if (!sameVersion(file.stat, job)) {
      await enqueueActivityIngestJob(db, {
        user: job.user,
        cartella: job.cartella,
        relPath: job.rel_path,
        mtimeMs: file.stat.mtimeMs,
        sizeBytes: file.stat.size,
      });

      const result =
        await completeActivityIngestJob(
          db,
          job.id
        );

      return {
        processed: true,
        status: result.status,
        requeued: Boolean(result.requeued),
      };
    }

    // Memorizza lo stato precedente per distinguere inserimenti e aggiornamenti.
    const previousActivity = await db('activities')
      .select('id', 'deleted_at')
      .where({ id: file.id })
      .first();

    const newActivities = [];

    const scannedActivity =
      await scanActivitySingle(
        db,
        job.user,
        job.cartella,
        file,
        newActivities
      );

    /*
     * Ricontrolla il file dopo il parsing:
     * se e' cambiato mentre veniva elaborato,
     * richiede una seconda scansione.
     */
    /*
     * Sincronizzazione verso BASE_URL.
     *
     * La semantica resta quella del vecchio scanActivity:
     * un errore remoto viene registrato ma NON rende fallito
     * l'ingest locale.
     */
    if (
      scannedActivity &&
      SEND_PHOTOS &&
      BASE_URL
    ) {
      try {
        scannedActivity.user = job.user;

        await postWithAuth(
          `${BASE_URL}/activities`,
          scannedActivity
        );

        console.log(
          `[ActivityIngest] Activity ${scannedActivity.id} ` +
          `inviata al server remoto`
        );
      } catch (remoteError) {
        console.error(
          `[ActivityIngest] Invio remoto fallito ` +
          `${scannedActivity.id}:`,
          remoteError.message
        );
      }
    }

    const statAfter =
      await fs.stat(file.absPath);

    if (
      Math.trunc(statAfter.mtimeMs) !==
        Math.trunc(file.stat.mtimeMs) ||
      statAfter.size !== file.stat.size
    ) {
      await enqueueActivityIngestJob(db, {
        user: job.user,
        cartella: job.cartella,
        relPath: job.rel_path,
        mtimeMs: statAfter.mtimeMs,
        sizeBytes: statAfter.size,
      });
    }

    const result =
      await completeActivityIngestJob(
        db,
        job.id
      );

    console.log(
      `[ActivityIngest] Job ${job.id} ` +
      `completato; stato: ${result.status}`
    );

    // Notifica la galleria solo dopo il completamento effettivo dell'ingest.
    if (
      scannedActivity &&
      scannedActivity.id &&
      !result.requeued &&
      result.status === 'done'
    ) {
      await enqueueServerEvent(db, {
        channel: 'activity',
        user: job.user,
        payload: {
          type:
            previousActivity && !previousActivity.deleted_at
              ? 'updated'
              : 'added',
          id: scannedActivity.id,
          entity: 'activity',
          mode: 'activities',
        },
      });

      console.log(
        `[ActivityIngest] Evento gallery accodato ` +
        `activity=${scannedActivity.id}`
      );
    }

    return {
      processed: true,
      status: result.status,
      requeued: Boolean(result.requeued),
      activityId:
        scannedActivity?.id || file.id,
    };
  } catch (error) {
    console.error(
      `[ActivityIngest] Errore job ${job.id}:`,
      error.message
    );

    try {
      await failActivityIngestJob(
        db,
        job.id,
        error
      );
    } catch (queueError) {
      console.error(
        '[ActivityIngest] Errore aggiornando queue:',
        queueError.message
      );
    }

    return {
      processed: true,
      error: error.message,
    };
  }
}

async function runWorker(
  db,
  options = {}
) {
  const pollMs = Math.max(
    250,
    Number(options.pollMs) || 2000
  );

  const staleMinutes = Math.max(
    1,
    Number(options.staleMinutes) || 30
  );

  await ensureActivityIngestQueue(db);

  await resetStaleActivityIngestJobs(
    db,
    staleMinutes
  );

  console.log(
    `[ActivityIngest] Worker avviato; ` +
    `polling ${pollMs} ms`
  );

  while (true) {
    try {
      const result = await runOneJob(db);

      if (!result.processed) {
        await new Promise(
          resolve => setTimeout(resolve, pollMs)
        );
      }
    } catch (error) {
      console.error(
        '[ActivityIngest] Errore worker:',
        error
      );

      await new Promise(
        resolve => setTimeout(resolve, pollMs)
      );
    }
  }
}

module.exports = {
  runOneJob,
  runWorker,
  resolveActivityJobFile,
  sameVersion,
};