137 lines
4.0 KiB
JavaScript
137 lines
4.0 KiB
JavaScript
const Database = require('better-sqlite3');
|
|
const path = require('path');
|
|
const log = require('./logger');
|
|
|
|
const DB_PATH = path.join('/mnt/webdata/storage', 'jobs.db');
|
|
const db = new Database(DB_PATH);
|
|
|
|
// WAL mode for better concurrency
|
|
db.pragma('journal_mode = WAL');
|
|
db.pragma('busy_timeout = 5000');
|
|
|
|
// Create jobs table
|
|
db.exec(`
|
|
CREATE TABLE IF NOT EXISTS jobs (
|
|
id TEXT PRIMARY KEY,
|
|
type TEXT NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'pending',
|
|
payload TEXT NOT NULL DEFAULT '{}',
|
|
result TEXT,
|
|
progress INTEGER DEFAULT 0,
|
|
error TEXT,
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL,
|
|
started_at INTEGER,
|
|
finished_at INTEGER
|
|
)
|
|
`);
|
|
|
|
db.exec(`CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status)`);
|
|
db.exec(`CREATE INDEX IF NOT EXISTS idx_jobs_type ON jobs(type)`);
|
|
|
|
// Prepared statements
|
|
const stmts = {
|
|
insert: db.prepare(`INSERT INTO jobs (id, type, status, payload, progress, created_at, updated_at) VALUES (?, ?, 'pending', ?, 0, ?, ?)`),
|
|
get: db.prepare(`SELECT * FROM jobs WHERE id = ?`),
|
|
updateStatus: db.prepare(`UPDATE jobs SET status = ?, updated_at = ? WHERE id = ?`),
|
|
updateProgress: db.prepare(`UPDATE jobs SET progress = ?, updated_at = ? WHERE id = ?`),
|
|
finish: db.prepare(`UPDATE jobs SET status = 'done', progress = 100, result = ?, finished_at = ?, updated_at = ? WHERE id = ?`),
|
|
fail: db.prepare(`UPDATE jobs SET status = 'error', error = ?, finished_at = ?, updated_at = ? WHERE id = ?`),
|
|
pending: db.prepare(`SELECT * FROM jobs WHERE type = ? AND status = 'pending' ORDER BY created_at LIMIT ?`),
|
|
cleanup: db.prepare(`DELETE FROM jobs WHERE finished_at < ? AND status IN ('done', 'error')`),
|
|
stats: db.prepare(`SELECT status, COUNT(*) as count FROM jobs GROUP BY status`),
|
|
activeCount: db.prepare(`SELECT COUNT(*) as count FROM jobs WHERE status = 'processing'`),
|
|
queueDepth: db.prepare(`SELECT COUNT(*) as count FROM jobs WHERE status = 'pending'`),
|
|
};
|
|
|
|
function generateId(prefix) {
|
|
return `${prefix}_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
|
|
}
|
|
|
|
function addJob(type, payload) {
|
|
const id = generateId(type);
|
|
const now = Date.now();
|
|
stmts.insert.run(id, type, JSON.stringify(payload), now, now);
|
|
log.info(`Job added: ${id}`, { type });
|
|
return id;
|
|
}
|
|
|
|
function getJob(id) {
|
|
const row = stmts.get.get(id);
|
|
if (!row) return null;
|
|
row.payload = JSON.parse(row.payload);
|
|
if (row.result) row.result = JSON.parse(row.result);
|
|
return row;
|
|
}
|
|
|
|
function startJob(id) {
|
|
const now = Date.now();
|
|
db.prepare(`UPDATE jobs SET status = 'processing', started_at = ?, updated_at = ? WHERE id = ?`).run(now, now, id);
|
|
}
|
|
|
|
function updateProgress(id, progress) {
|
|
stmts.updateProgress.run(progress, Date.now(), id);
|
|
}
|
|
|
|
function finishJob(id, result) {
|
|
const now = Date.now();
|
|
stmts.finish.run(JSON.stringify(result), now, now, id);
|
|
log.info(`Job done: ${id}`);
|
|
}
|
|
|
|
function failJob(id, error) {
|
|
const now = Date.now();
|
|
stmts.fail.run(String(error).slice(0, 500), now, now, id);
|
|
log.error(`Job failed: ${id}`, { error: String(error).slice(0, 200) });
|
|
}
|
|
|
|
function getPending(type, limit = 5) {
|
|
return stmts.pending.all(type, limit).map(row => {
|
|
row.payload = JSON.parse(row.payload);
|
|
return row;
|
|
});
|
|
}
|
|
|
|
function getStats() {
|
|
const rows = stmts.stats.all();
|
|
const stats = {};
|
|
for (const r of rows) stats[r.status] = r.count;
|
|
return stats;
|
|
}
|
|
|
|
function getActiveCount() {
|
|
return stmts.activeCount.get().count;
|
|
}
|
|
|
|
function getQueueDepth() {
|
|
return stmts.queueDepth.get().count;
|
|
}
|
|
|
|
// Cleanup old finished jobs (older than 1 hour)
|
|
function cleanup() {
|
|
const cutoff = Date.now() - 3600000;
|
|
const result = stmts.cleanup.run(cutoff);
|
|
if (result.changes > 0) log.info(`Cleaned up ${result.changes} old jobs`);
|
|
}
|
|
|
|
// Run cleanup every 10 min
|
|
setInterval(cleanup, 600000);
|
|
|
|
// Graceful close
|
|
process.on('exit', () => { try { db.close(); } catch {} });
|
|
|
|
module.exports = {
|
|
addJob,
|
|
getJob,
|
|
startJob,
|
|
updateProgress,
|
|
finishJob,
|
|
failJob,
|
|
getPending,
|
|
getStats,
|
|
getActiveCount,
|
|
getQueueDepth,
|
|
cleanup,
|
|
db,
|
|
};
|