527 lines
12 KiB
JavaScript
527 lines
12 KiB
JavaScript
'use strict';
|
|
|
|
const crypto = require('crypto');
|
|
|
|
const {
|
|
withSqliteBusyRetry,
|
|
} = require('../../db/sqliteRetry');
|
|
|
|
const TABLE = 'activity_ingest_jobs';
|
|
|
|
function makeJobId(user, cartella, relPath) {
|
|
return crypto
|
|
.createHash('sha256')
|
|
.update(JSON.stringify([user, cartella, relPath]))
|
|
.digest('hex');
|
|
}
|
|
|
|
async function ensureActivityIngestQueue(db) {
|
|
await withSqliteBusyRetry(
|
|
'activity queue ensure',
|
|
async () => {
|
|
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_activity_ingest_status_created'
|
|
);
|
|
|
|
t.index(
|
|
['user', 'cartella'],
|
|
'idx_activity_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 enqueueActivityIngestJob(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 activity'
|
|
);
|
|
}
|
|
|
|
await ensureActivityIngestQueue(db);
|
|
|
|
const id = makeJobId(user, cartella, relPath);
|
|
const mtime = Math.trunc(mtimeMs);
|
|
const size = Math.trunc(sizeBytes);
|
|
|
|
return withSqliteBusyRetry(
|
|
'activity queue 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) {
|
|
return {
|
|
id,
|
|
status: existing.status,
|
|
unchanged: true,
|
|
};
|
|
}
|
|
|
|
if (existing.status === 'processing') {
|
|
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 enqueueActivityIngestJobs(db, inputs) {
|
|
if (!Array.isArray(inputs)) {
|
|
throw new TypeError('Il batch deve essere un array');
|
|
}
|
|
if (inputs.length === 0) return [];
|
|
|
|
for (const input of inputs) {
|
|
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 nel batch activity');
|
|
}
|
|
}
|
|
|
|
await ensureActivityIngestQueue(db);
|
|
|
|
return withSqliteBusyRetry(
|
|
'activity queue enqueue batch',
|
|
() => db.transaction(async (trx) => {
|
|
const results = [];
|
|
|
|
for (const input of inputs) {
|
|
const { user, cartella, relPath, mtimeMs, sizeBytes } = input;
|
|
const id = makeJobId(user, cartella, relPath);
|
|
const mtime = Math.trunc(mtimeMs);
|
|
const size = Math.trunc(sizeBytes);
|
|
|
|
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) {
|
|
results.push({
|
|
id, status: existing.status, unchanged: true,
|
|
});
|
|
continue;
|
|
}
|
|
|
|
if (existing.status === 'processing') {
|
|
await trx(TABLE)
|
|
.where({ id, status: 'processing' })
|
|
.update({
|
|
source_mtime_ms: mtime,
|
|
source_size_bytes: size,
|
|
rescan_requested: 1,
|
|
updated_at: trx.fn.now(),
|
|
});
|
|
|
|
results.push({
|
|
id, status: 'processing', changedWhileProcessing: true,
|
|
});
|
|
continue;
|
|
}
|
|
|
|
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(),
|
|
});
|
|
|
|
results.push({ id, status: 'pending', updated: true });
|
|
continue;
|
|
}
|
|
|
|
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(),
|
|
});
|
|
|
|
results.push({ id, status: 'pending', inserted: true });
|
|
}
|
|
|
|
return results;
|
|
})
|
|
);
|
|
}
|
|
|
|
async function claimNextActivityIngestJob(db) {
|
|
await ensureActivityIngestQueue(db);
|
|
|
|
return withSqliteBusyRetry(
|
|
'activity queue 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 completeActivityIngestJob(db, id) {
|
|
return withSqliteBusyRetry(
|
|
'activity queue 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 failActivityIngestJob(
|
|
db,
|
|
id,
|
|
error
|
|
) {
|
|
const message = String(
|
|
error?.stack ||
|
|
error?.message ||
|
|
error
|
|
).slice(0, 12000);
|
|
|
|
return withSqliteBusyRetry(
|
|
'activity queue 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 resetStaleActivityIngestJobs(
|
|
db,
|
|
staleMinutes = 30
|
|
) {
|
|
await ensureActivityIngestQueue(db);
|
|
|
|
const cutoff = new Date(
|
|
Date.now() - staleMinutes * 60 * 1000
|
|
)
|
|
.toISOString()
|
|
.replace('T', ' ')
|
|
.replace(/\.\d{3}Z$/, '');
|
|
|
|
return withSqliteBusyRetry(
|
|
'activity queue stale reset',
|
|
() => 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 getActivityIngestQueueCounts(db) {
|
|
await ensureActivityIngestQueue(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 = {
|
|
ensureActivityIngestQueue,
|
|
enqueueActivityIngestJob,
|
|
enqueueActivityIngestJobs,
|
|
claimNextActivityIngestJob,
|
|
completeActivityIngestJob,
|
|
failActivityIngestJob,
|
|
resetStaleActivityIngestJobs,
|
|
getActivityIngestQueueCounts,
|
|
};
|