'use strict'; const path = require('path'); const fs = require('fs/promises'); const { WEB_ROOT, SUPPORTED_EXTS } = require('../config'); const scanPhotoSingle = require('./scanPhotoSingle'); const { enqueueServerEvent, } = require('../../db/serverEventOutbox'); const { sha256 } = require('./utils'); const { ensurePhotoIngestQueue, claimNextPhotoIngestJob, enqueuePhotoIngestJob, completePhotoIngestJob, failPhotoIngestJob, resetStalePhotoIngestJobs, } = require('./photoIngestQueue'); 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 resolveJobFile(job) { for (const value of [job.user, job.cartella, job.rel_path]) { if (typeof value !== 'string' || !value.trim()) { throw new Error('Percorso del 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 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 non validi'); } const photosRoot = path.resolve( __dirname, '..', '..', WEB_ROOT, 'photos' ); const originalRoot = path.resolve( photosRoot, job.user, 'original', job.cartella ); const absPath = path.resolve(originalRoot, ...job.rel_path.split('/')); if (!isInside(originalRoot, absPath)) { throw new Error('Il file esce dalla cartella originale prevista'); } const realRoot = await fs.realpath(originalRoot); const realFile = await fs.realpath(absPath); if (!isInside(realRoot, realFile)) { throw new Error('Il file risolve fuori dalla cartella originale'); } const ext = path.extname(absPath).toLowerCase(); if (!SUPPORTED_EXTS.has(ext)) { throw new Error(`Estensione non supportata: ${ext}`); } const stat = await fs.stat(realFile); if (!stat.isFile()) { throw new Error('Il percorso non è un file regolare'); } return { id: sha256(`${job.user}/${job.cartella}/${job.rel_path}`), user: job.user, cartella: job.cartella, name: path.basename(absPath), relPath: job.rel_path, absPath: realFile, ext, stat, path: `/photos/${job.user}/original/${job.cartella}/${job.rel_path}`, }; } async function runOneJob(db) { await ensurePhotoIngestQueue(db); const job = await claimNextPhotoIngestJob(db); if (!job) { return { processed: false }; } console.log(`[PhotoIngest] Avvio job ${job.id}: ${job.rel_path}`); try { const file = await resolveJobFile(job); // Se la versione è già cambiata, aggiorna il job mentre è // processing: completePhotoIngestJob lo riporterà a pending. if (!sameVersion(file.stat, job)) { await enqueuePhotoIngestJob(db, { user: job.user, cartella: job.cartella, relPath: job.rel_path, mtimeMs: file.stat.mtimeMs, sizeBytes: file.stat.size, }); const result = await completePhotoIngestJob(db, job.id); return { processed: true, status: result.status, requeued: Boolean(result.requeued), }; } const previous = await db('photos') .select('id', 'size_bytes', 'mtimeMs', 'fast_hash', 'path') .where({ id: file.id }) .first(); const scannedPhoto = await scanPhotoSingle( db, job.user, job.cartella, file, [], '', previous || null ); // Se il file è scomparso, non eliminare il record fotografico. // Il job va in errore e l'evento di cancellazione resta responsabilità // del watcher. const statAfter = await fs.stat(file.absPath); // Se è cambiato durante la lavorazione, aggiorna la versione in coda // prima di completare il job: rescan_requested garantisce il retry. if ( Math.trunc(statAfter.mtimeMs) !== Math.trunc(file.stat.mtimeMs) || statAfter.size !== file.stat.size ) { await enqueuePhotoIngestJob(db, { user: job.user, cartella: job.cartella, relPath: job.rel_path, mtimeMs: statAfter.mtimeMs, sizeBytes: statAfter.size, }); } const result = await completePhotoIngestJob(db, job.id); console.log( `[PhotoIngest] Job ${job.id} completato; stato: ${result.status}` ); /* * Notifica una foto realmente nuova soltanto dopo che il * job e' stato completato definitivamente. * * IMPORTANTE: * qui NON viene importato ws-server. * Il daemon scrive soltanto nella outbox SQLite. */ if ( !previous && !result.requeued && result.status === 'done' && scannedPhoto && scannedPhoto.id ) { await enqueueServerEvent(db, { channel: 'photo', user: job.user, payload: { type: 'added', id: scannedPhoto.id, }, }); console.log( `[PhotoIngest] Evento gallery accodato photo=${scannedPhoto.id}` ); } return { processed: true, status: result.status, requeued: Boolean(result.requeued), }; } catch (err) { console.error(`[PhotoIngest] Errore job ${job.id}:`, err.message); try { await failPhotoIngestJob(db, job.id, err.message); } catch (queueErr) { console.error( '[PhotoIngest] Errore aggiornando la coda:', queueErr.message ); } return { processed: true, error: err.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 ensurePhotoIngestQueue(db); await resetStalePhotoIngestJobs(db, staleMinutes); console.log(`[PhotoIngest] Worker avviato; polling ${pollMs} ms`); while (true) { try { const result = await runOneJob(db); if (!result.processed) { await new Promise(resolve => setTimeout(resolve, pollMs)); } } catch (err) { console.error('[PhotoIngest] Errore worker:', err); await new Promise(resolve => setTimeout(resolve, pollMs)); } } } module.exports = { runOneJob, runWorker, resolveJobFile, sameVersion, };