'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, };