348 lines
8.6 KiB
JavaScript
348 lines
8.6 KiB
JavaScript
'use strict';
|
|
|
|
const crypto = require('crypto');
|
|
|
|
const TABLE = 'photo_ingest_jobs';
|
|
|
|
const SQLITE_BUSY_RETRY_DELAYS_MS = [
|
|
25,
|
|
50,
|
|
100,
|
|
200,
|
|
400,
|
|
800,
|
|
1600,
|
|
];
|
|
|
|
function sleep(ms) {
|
|
return new Promise(resolve => setTimeout(resolve, ms));
|
|
}
|
|
|
|
function isSqliteBusy(error) {
|
|
if (!error) return false;
|
|
|
|
if (
|
|
error.code === 'SQLITE_BUSY' ||
|
|
error.code === 'SQLITE_LOCKED'
|
|
) {
|
|
return true;
|
|
}
|
|
|
|
const message = String(error.message || error).toLowerCase();
|
|
|
|
return (
|
|
message.includes('database is locked') ||
|
|
message.includes('database table is locked')
|
|
);
|
|
}
|
|
|
|
async function withSqliteBusyRetry(label, operation) {
|
|
let retry = 0;
|
|
|
|
while (true) {
|
|
try {
|
|
return await operation();
|
|
} catch (error) {
|
|
if (
|
|
!isSqliteBusy(error) ||
|
|
retry >= SQLITE_BUSY_RETRY_DELAYS_MS.length
|
|
) {
|
|
throw error;
|
|
}
|
|
|
|
const delay = SQLITE_BUSY_RETRY_DELAYS_MS[retry];
|
|
retry += 1;
|
|
|
|
console.warn(
|
|
`[PhotoIngestQueue] SQLITE_BUSY ${label}; ` +
|
|
`retry=${retry}/${SQLITE_BUSY_RETRY_DELAYS_MS.length} ` +
|
|
`delay=${delay}ms`
|
|
);
|
|
|
|
await sleep(delay);
|
|
}
|
|
}
|
|
}
|
|
|
|
function makeJobId(user, cartella, relPath) {
|
|
return crypto
|
|
.createHash('sha256')
|
|
.update(JSON.stringify([user, cartella, relPath]))
|
|
.digest('hex');
|
|
}
|
|
|
|
async function ensurePhotoIngestQueue(db) {
|
|
if (!(await db.schema.hasTable(TABLE))) {
|
|
await db.schema.createTable(TABLE, (t) => {
|
|
t.string('id').primary();
|
|
t.string('user').notNullable();
|
|
t.string('cartella').notNullable();
|
|
t.string('rel_path').notNullable();
|
|
t.integer('source_mtime_ms').notNullable();
|
|
t.integer('source_size_bytes').notNullable();
|
|
t.string('status').notNullable().defaultTo('pending');
|
|
t.integer('attempts').notNullable().defaultTo(0);
|
|
t.integer('rescan_requested').notNullable().defaultTo(0);
|
|
t.text('last_error').nullable();
|
|
t.datetime('started_at').nullable();
|
|
t.datetime('completed_at').nullable();
|
|
t.datetime('created_at').notNullable().defaultTo(db.fn.now());
|
|
t.datetime('updated_at').notNullable().defaultTo(db.fn.now());
|
|
t.index(['status', 'created_at'], 'idx_photo_ingest_status_created');
|
|
t.index(['user', 'cartella'], 'idx_photo_ingest_folder');
|
|
});
|
|
}
|
|
|
|
if (!(await db.schema.hasColumn(TABLE, 'rescan_requested'))) {
|
|
await db.schema.alterTable(TABLE, (t) => {
|
|
t.integer('rescan_requested').notNullable().defaultTo(0);
|
|
});
|
|
}
|
|
}
|
|
|
|
async function enqueuePhotoIngestJob(db, input) {
|
|
const { user, cartella, relPath, mtimeMs, sizeBytes } = input;
|
|
|
|
if (
|
|
typeof user !== 'string' || !user ||
|
|
typeof cartella !== 'string' || !cartella ||
|
|
typeof relPath !== 'string' || !relPath ||
|
|
!Number.isFinite(mtimeMs) || mtimeMs < 0 ||
|
|
!Number.isFinite(sizeBytes) || sizeBytes < 0
|
|
) {
|
|
throw new TypeError('Dati non validi per accodare la foto');
|
|
}
|
|
|
|
await ensurePhotoIngestQueue(db);
|
|
|
|
const id = makeJobId(user, cartella, relPath);
|
|
const mtime = Math.trunc(mtimeMs);
|
|
const size = Math.trunc(sizeBytes);
|
|
|
|
return withSqliteBusyRetry('enqueue', () =>
|
|
db.transaction(async (trx) => {
|
|
const existing = await trx(TABLE).where({ id }).first();
|
|
|
|
if (existing) {
|
|
const unchanged =
|
|
Number(existing.source_mtime_ms) === mtime &&
|
|
Number(existing.source_size_bytes) === size;
|
|
|
|
if (unchanged) {
|
|
// Un errore non viene riavviato da eventi duplicati.
|
|
return { id, status: existing.status, unchanged: true };
|
|
}
|
|
|
|
if (existing.status === 'processing') {
|
|
// Conserva la versione più recente richiesta.
|
|
// Il worker finirà il lavoro corrente e poi riaccoderà.
|
|
await trx(TABLE).where({ id, status: 'processing' }).update({
|
|
source_mtime_ms: mtime,
|
|
source_size_bytes: size,
|
|
rescan_requested: 1,
|
|
updated_at: trx.fn.now(),
|
|
});
|
|
|
|
return {
|
|
id,
|
|
status: 'processing',
|
|
changedWhileProcessing: true,
|
|
};
|
|
}
|
|
|
|
await trx(TABLE).where({ id }).update({
|
|
source_mtime_ms: mtime,
|
|
source_size_bytes: size,
|
|
status: 'pending',
|
|
attempts: 0,
|
|
rescan_requested: 0,
|
|
last_error: null,
|
|
started_at: null,
|
|
completed_at: null,
|
|
updated_at: trx.fn.now(),
|
|
});
|
|
|
|
return { id, status: 'pending', updated: true };
|
|
}
|
|
|
|
await trx(TABLE).insert({
|
|
id,
|
|
user,
|
|
cartella,
|
|
rel_path: relPath,
|
|
source_mtime_ms: mtime,
|
|
source_size_bytes: size,
|
|
status: 'pending',
|
|
attempts: 0,
|
|
rescan_requested: 0,
|
|
created_at: trx.fn.now(),
|
|
updated_at: trx.fn.now(),
|
|
});
|
|
|
|
return { id, status: 'pending', inserted: true };
|
|
})
|
|
);
|
|
}
|
|
|
|
async function claimNextPhotoIngestJob(db) {
|
|
await ensurePhotoIngestQueue(db);
|
|
|
|
return withSqliteBusyRetry('claim', () =>
|
|
db.transaction(async (trx) => {
|
|
const job = await trx(TABLE)
|
|
.where({ status: 'pending' })
|
|
.orderBy('created_at', 'asc')
|
|
.first();
|
|
|
|
if (!job) return null;
|
|
|
|
const changed = await trx(TABLE)
|
|
.where({ id: job.id, status: 'pending' })
|
|
.update({
|
|
status: 'processing',
|
|
attempts: trx.raw('attempts + 1'),
|
|
rescan_requested: 0,
|
|
started_at: trx.fn.now(),
|
|
completed_at: null,
|
|
last_error: null,
|
|
updated_at: trx.fn.now(),
|
|
});
|
|
|
|
if (!changed) return null;
|
|
|
|
return {
|
|
...job,
|
|
status: 'processing',
|
|
attempts: Number(job.attempts || 0) + 1,
|
|
rescan_requested: 0,
|
|
};
|
|
})
|
|
);
|
|
}
|
|
|
|
async function completePhotoIngestJob(db, id) {
|
|
return withSqliteBusyRetry('complete', () =>
|
|
db.transaction(async (trx) => {
|
|
const job = await trx(TABLE)
|
|
.where({ id, status: 'processing' })
|
|
.first();
|
|
|
|
if (!job) return { updated: false };
|
|
|
|
if (Number(job.rescan_requested) === 1) {
|
|
await trx(TABLE).where({ id, status: 'processing' }).update({
|
|
status: 'pending',
|
|
rescan_requested: 0,
|
|
started_at: null,
|
|
completed_at: null,
|
|
last_error: null,
|
|
updated_at: trx.fn.now(),
|
|
});
|
|
|
|
return { updated: true, status: 'pending', requeued: true };
|
|
}
|
|
|
|
await trx(TABLE).where({ id, status: 'processing' }).update({
|
|
status: 'done',
|
|
completed_at: trx.fn.now(),
|
|
updated_at: trx.fn.now(),
|
|
last_error: null,
|
|
});
|
|
|
|
return { updated: true, status: 'done', requeued: false };
|
|
})
|
|
);
|
|
}
|
|
|
|
async function failPhotoIngestJob(db, id, error) {
|
|
const message = String(error?.stack || error?.message || error)
|
|
.slice(0, 12000);
|
|
|
|
return withSqliteBusyRetry('fail', () =>
|
|
db.transaction(async (trx) => {
|
|
const job = await trx(TABLE)
|
|
.where({ id, status: 'processing' })
|
|
.first();
|
|
|
|
if (!job) return { updated: false };
|
|
|
|
if (Number(job.rescan_requested) === 1) {
|
|
await trx(TABLE).where({ id, status: 'processing' }).update({
|
|
status: 'pending',
|
|
rescan_requested: 0,
|
|
started_at: null,
|
|
completed_at: null,
|
|
last_error: message,
|
|
updated_at: trx.fn.now(),
|
|
});
|
|
|
|
return { updated: true, status: 'pending', requeued: true };
|
|
}
|
|
|
|
await trx(TABLE).where({ id, status: 'processing' }).update({
|
|
status: 'error',
|
|
last_error: message,
|
|
completed_at: trx.fn.now(),
|
|
updated_at: trx.fn.now(),
|
|
});
|
|
|
|
return { updated: true, status: 'error', requeued: false };
|
|
})
|
|
);
|
|
}
|
|
|
|
async function resetStalePhotoIngestJobs(db, staleMinutes = 30) {
|
|
await ensurePhotoIngestQueue(db);
|
|
|
|
const cutoff = new Date(Date.now() - staleMinutes * 60 * 1000)
|
|
.toISOString()
|
|
.replace('T', ' ')
|
|
.replace(/\.\d{3}Z$/, '');
|
|
|
|
return db(TABLE)
|
|
.where({ status: 'processing' })
|
|
.whereNotNull('started_at')
|
|
.where('started_at', '<', cutoff)
|
|
.update({
|
|
status: 'pending',
|
|
rescan_requested: 0,
|
|
started_at: null,
|
|
completed_at: null,
|
|
last_error: 'Recuperato dopo interruzione del worker',
|
|
updated_at: db.fn.now(),
|
|
});
|
|
}
|
|
|
|
async function getPhotoIngestQueueCounts(db) {
|
|
await ensurePhotoIngestQueue(db);
|
|
|
|
const rows = await db(TABLE)
|
|
.select('status')
|
|
.count({ total: '*' })
|
|
.groupBy('status');
|
|
|
|
const counts = {
|
|
pending: 0,
|
|
processing: 0,
|
|
done: 0,
|
|
error: 0,
|
|
};
|
|
|
|
for (const row of rows) {
|
|
if (Object.hasOwn(counts, row.status)) {
|
|
counts[row.status] = Number(row.total);
|
|
}
|
|
}
|
|
|
|
return counts;
|
|
}
|
|
|
|
module.exports = {
|
|
ensurePhotoIngestQueue,
|
|
enqueuePhotoIngestJob,
|
|
claimNextPhotoIngestJob,
|
|
completePhotoIngestJob,
|
|
failPhotoIngestJob,
|
|
resetStalePhotoIngestJobs,
|
|
getPhotoIngestQueueCounts,
|
|
};
|