feat: persistent photo job queue, worker, history with rollback

- Replace in-memory photoAiJobs Map with DB-backed photo_jobs table
- Background photo worker (worker.js) with retry, backoff, stale reset
  - Handles both AI (Real-ESRGAN) and server-side (sharp) enhancement
  - Controlled via photo_worker_enabled setting
- Photo job history in enhance modal with before/after thumbnails + rollback
- Worker dashboard: photo jobs section with status counts, recent table,
  compare slider for before/after, rollback from worker UI
- New endpoints: /api/photo-jobs/status|wake|enabled|requeue-failed,
  /api/entries/:id/photo/jobs (history), .../rollback
- swapEntryPhotoFiles logs every mutation to photo_jobs table
This commit is contained in:
dev
2026-09-17 17:36:56 +03:00
parent 27a2238082
commit 1145be02f3
8 changed files with 876 additions and 55 deletions
+229 -52
View File
@@ -5,7 +5,7 @@ const rateLimit = require('express-rate-limit');
const helmet = require('helmet');
const bcrypt = require('bcrypt');
const heicConvert = require('heic-convert');
const { createEntryAutoChecker } = require('./worker');
const { createEntryAutoChecker, createPhotoEnhanceWorker } = require('./worker');
const https = require('https');
const path = require('path');
@@ -300,6 +300,14 @@ app.get('/uploads/thumb/:name', fileLimiter, async (req, res) => {
return sendImageThumb(res, fp);
});
app.get('/uploads/.originals/:name', fileLimiter, async (req, res) => {
const name = req.params.name;
if (!/^[A-Za-z0-9._-]+$/.test(name)) return res.status(400).end();
const fp = path.join(ORIGINALS_DIR, name);
if (!fs.existsSync(fp) || !fs.statSync(fp).isFile()) return res.status(404).end();
return sendImageThumb(res, fp);
});
app.use((req, res, next) => {
const p = req.path;
if (!p.startsWith('/uploads') && !p.startsWith('/vendor')) {
@@ -702,6 +710,25 @@ async function ensureEntryAiColumns() {
}
}
async function ensurePhotoJobsTable() {
await pool.query(`CREATE TABLE IF NOT EXISTS photo_jobs (
id SERIAL PRIMARY KEY,
entry_id INT NOT NULL REFERENCES entries(id) ON DELETE CASCADE,
action VARCHAR(20) NOT NULL DEFAULT 'ai',
params JSONB,
before_path VARCHAR(255),
after_path VARCHAR(255),
status VARCHAR(20) NOT NULL DEFAULT 'pending',
attempts INT NOT NULL DEFAULT 0,
error TEXT,
created_at TIMESTAMPTZ DEFAULT now(),
finished_at TIMESTAMPTZ
)`);
await pool.query(`CREATE INDEX IF NOT EXISTS idx_photo_jobs_pending ON photo_jobs(id) WHERE status = 'pending'`);
await pool.query(`CREATE INDEX IF NOT EXISTS idx_photo_jobs_entry_id ON photo_jobs(entry_id)`);
await pool.query(`INSERT INTO settings (key, value) VALUES ('photo_worker_enabled', 'true') ON CONFLICT (key) DO NOTHING`);
}
async function ensureUserTables() {
await pool.query(`CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
@@ -1071,6 +1098,7 @@ const PHOTO_AI_URL = process.env.PHOTO_AI_URL || '';
const AI_MODEL = process.env.AI_MODEL || 'qwen2.5-1.5b-instruct-q4_k_m.gguf';
const AI_DEFAULT_PROMPT = process.env.AI_PROMPT || 'Ты — редактор текстов. Исправь ТОЛЬКО грамматические, орфографические и пунктуационные ошибки в тексте. Приведи к правильному регистру буквы. НЕ меняй слова, структуру предложений, стиль или смысл текста. Верни ТОЛЬКО исправленный текст без пояснений.';
let entryAutoChecker = null;
let photoWorker = null;
async function getAiPrompt() {
const prompt = await getSetting('ai_prompt', AI_DEFAULT_PROMPT);
@@ -3678,7 +3706,7 @@ function clampEnhanceParam(v, min, max, def) {
return Number.isFinite(n) ? Math.min(max, Math.max(min, n)) : def;
}
async function swapEntryPhotoFiles(req, entryId, oldPath, newPath, { keepOriginal = true } = {}) {
async function swapEntryPhotoFiles(req, entryId, oldPath, newPath, { keepOriginal = true, action = 'enhance' } = {}) {
let originalPath = null;
const { rows: prevRows } = await pool.query('SELECT photo_original_path FROM entries WHERE id = $1', [entryId]);
const prevOriginal = prevRows.length ? prevRows[0].photo_original_path : null;
@@ -3702,6 +3730,10 @@ async function swapEntryPhotoFiles(req, entryId, oldPath, newPath, { keepOrigina
await pool.query('UPDATE entry_photos SET photo_path = $1 WHERE entry_id = $2 AND photo_path = $3', [newPath, entryId, oldPath]);
const oldThumb = path.join('uploads', '.thumbs', path.basename(oldPath).replace(/\.[^.]+$/, '') + '.webp');
safeUnlink(oldThumb);
await pool.query(
'INSERT INTO photo_jobs (entry_id, action, status, before_path, after_path, finished_at) VALUES ($1, $2, $3, $4, $5, now())',
[entryId, action, 'done', originalPath, newPath]
);
await logAudit(req, 'entry.photo.enhance', { entry_id: entryId, old_path: oldPath, new_path: newPath, original_path: originalPath });
invalidateEntries();
}
@@ -3757,7 +3789,7 @@ app.put('/api/entries/:id/photo/enhance', requireAuth, (req, res, next) => {
await convertPhoto(req.file);
newPath = `/uploads/${req.file.filename}`;
}
await swapEntryPhotoFiles(req, req.params.id, oldPath, newPath);
await swapEntryPhotoFiles(req, req.params.id, oldPath, newPath, { action: 'enhance' });
res.json({ ok: true, photo_path: newPath, engine });
} catch (e) {
if (req.file) safeUnlink(req.file.path);
@@ -3766,37 +3798,6 @@ app.put('/api/entries/:id/photo/enhance', requireAuth, (req, res, next) => {
}
});
const photoAiJobs = new Map();
let photoAiChain = Promise.resolve();
async function runPhotoAiJob(job) {
job.status = 'processing';
try {
const { rows } = await pool.query('SELECT photo_path FROM entries WHERE id = $1', [job.entryId]);
if (!rows.length || !rows[0].photo_path) throw new Error('У записи нет фото');
const photoPath = rows[0].photo_path;
const srcAbs = path.join(UPLOADS_DIR, photoPath.replace(/^\/+/, '').replace(/^uploads\//, ''));
const buf = fs.readFileSync(srcAbs);
const fd = new FormData();
fd.append('image', new Blob([buf], { type: 'image/jpeg' }), 'photo.jpg');
fd.append('scale', '2');
const resp = await fetch(PHOTO_AI_URL.replace(/\/+$/, '') + '/enhance', {
method: 'POST',
body: fd,
signal: AbortSignal.timeout(300000),
});
if (!resp.ok) throw new Error('AI service error: ' + resp.status);
const out = Buffer.from(await resp.arrayBuffer());
const newName = crypto.randomBytes(12).toString('hex') + '.jpg';
fs.writeFileSync(path.join(UPLOADS_DIR, newName), out);
job.photo_path = `/uploads/${newName}`;
job.status = 'done';
} catch (e) {
job.status = 'error';
job.error = e.message || String(e);
}
}
app.post('/api/entries/:id/photo/enhance-ai', requireAuth, async (req, res) => {
if (!PHOTO_AI_URL) return res.status(503).json({ error: 'ИИ-обработка фото не настроена' });
if (req.user.role !== 'admin') {
@@ -3807,32 +3808,106 @@ app.post('/api/entries/:id/photo/enhance-ai', requireAuth, async (req, res) => {
const { rows: cur } = await pool.query('SELECT photo_path FROM entries WHERE id = $1', [req.params.id]);
if (!cur.length) return res.status(404).json({ error: 'Запись не найдена' });
if (!cur[0].photo_path) return res.status(400).json({ error: 'У записи нет фото' });
const job = {
id: crypto.randomBytes(8).toString('hex'),
entryId: parseInt(req.params.id, 10),
req: { user: req.user, ip: req.ip },
status: 'pending',
created_at: new Date().toISOString(),
};
photoAiJobs.set(job.id, job);
for (const [id, j] of photoAiJobs) {
if (j.status === 'done' || j.status === 'error') {
if (Date.now() - new Date(j.created_at).getTime() > 3600000) photoAiJobs.delete(id);
}
try {
const { rows } = await pool.query(
`INSERT INTO photo_jobs (entry_id, action, status, params) VALUES ($1, 'ai', 'pending', NULL) RETURNING id`,
[req.params.id]
);
if (photoWorker) photoWorker.notify();
await logAudit(req, 'entry.photo.enhance-ai.queue', { entry_id: req.params.id, job_id: rows[0].id });
res.json({ jobId: rows[0].id });
} catch (e) {
console.error('POST /api/entries/:id/photo/enhance-ai:', e);
res.status(500).json({ error: 'Ошибка постановки задания в очередь' });
}
photoAiChain = photoAiChain.then(() => runPhotoAiJob(job)).catch(() => {});
res.json({ jobId: job.id });
});
app.get('/api/entries/:id/photo/enhance-ai/:jobId', requireAuth, async (req, res) => {
const job = photoAiJobs.get(req.params.jobId);
if (!job || job.entryId !== parseInt(req.params.id, 10)) return res.status(404).json({ error: 'Задание не найдено' });
const jobId = parseInt(req.params.jobId, 10);
if (!Number.isFinite(jobId)) return res.status(404).json({ error: 'Задание не найдено' });
const { rows } = await pool.query(
'SELECT status, error, after_path FROM photo_jobs WHERE id = $1 AND entry_id = $2',
[jobId, req.params.id]
);
if (!rows.length) return res.status(404).json({ error: 'Задание не найдено' });
const job = rows[0];
const out = { status: job.status };
if (job.status === 'error') out.error = job.error;
if (job.status === 'done') out.photo_path = job.photo_path;
if (job.status === 'done') out.photo_path = job.after_path;
res.json(out);
});
app.get('/api/entries/:id/photo/jobs', requireAuth, async (req, res) => {
if (req.user.role !== 'admin') {
const acc = await entryAccessible(req.user, req.params.id);
if (!acc.found) return res.status(404).json({ error: 'Not found' });
if (!acc.allowed) return res.status(403).json({ error: 'Нет доступа к этой записи' });
}
const { rows } = await pool.query(
`SELECT id, action, status, params, before_path, after_path, error, created_at, finished_at
FROM photo_jobs WHERE entry_id = $1 ORDER BY id DESC LIMIT 50`,
[req.params.id]
);
res.json(rows);
});
app.post('/api/entries/:id/photo/jobs/:jobId/rollback', requireAuth, async (req, res) => {
if (req.user.role !== 'admin') {
const acc = await entryAccessible(req.user, req.params.id);
if (!acc.found) return res.status(404).json({ error: 'Not found' });
if (!acc.allowed) return res.status(403).json({ error: 'Нет доступа к этой записи' });
}
try {
const jobId = parseInt(req.params.jobId, 10);
if (!Number.isFinite(jobId)) return res.status(404).json({ error: 'Задание не найдено' });
const { rows: jobRows } = await pool.query(
`SELECT id, before_path FROM photo_jobs WHERE id = $1 AND entry_id = $2 AND status = 'done'`,
[jobId, req.params.id]
);
if (!jobRows.length) return res.status(404).json({ error: 'Задание не найдено' });
const job = jobRows[0];
if (!job.before_path || !job.before_path.startsWith('/uploads/.originals/')) {
return res.status(400).json({ error: 'Нет сохранённой версии для отката' });
}
const beforeName = path.basename(job.before_path);
if (!/^[A-Za-z0-9._-]+$/.test(beforeName)) return res.status(400).json({ error: 'Некорректный путь' });
const beforeAbs = path.join(ORIGINALS_DIR, beforeName);
if (!fs.existsSync(beforeAbs)) return res.status(400).json({ error: 'Файл версии не найден' });
const { rows: cur } = await pool.query('SELECT photo_path FROM entries WHERE id = $1', [req.params.id]);
if (!cur.length) return res.status(404).json({ error: 'Запись не найдена' });
const oldPath = cur[0].photo_path;
if (!oldPath) return res.status(400).json({ error: 'У записи нет фото' });
const ext = path.extname(beforeName) || '.jpg';
const newPath = `/uploads/${crypto.randomBytes(12).toString('hex')}${ext}`;
fs.copyFileSync(beforeAbs, path.join(UPLOADS_DIR, path.basename(newPath)));
const oldAbs = path.join(UPLOADS_DIR, String(oldPath).replace(/^\/+/, '').replace(/^uploads\//, ''));
let rollbackBefore = null;
if (fs.existsSync(oldAbs)) {
try {
const backupName = crypto.randomBytes(12).toString('hex') + (path.extname(oldAbs) || '.jpg');
fs.renameSync(oldAbs, path.join(ORIGINALS_DIR, backupName));
rollbackBefore = `/uploads/.originals/${backupName}`;
} catch (e) {
console.error('photo rollback backup failed:', e);
}
}
const oldThumb = path.join('uploads', '.thumbs', path.basename(oldPath).replace(/\.[^.]+$/, '') + '.webp');
safeUnlink(oldThumb);
await pool.query('UPDATE entries SET photo_path = $1 WHERE id = $2', [newPath, req.params.id]);
await pool.query('UPDATE entry_photos SET photo_path = $1 WHERE entry_id = $2 AND photo_path = $3', [newPath, req.params.id, oldPath]);
await pool.query(
'INSERT INTO photo_jobs (entry_id, action, status, before_path, after_path, error, finished_at) VALUES ($1, $2, $3, $4, $5, NULL, now())',
[req.params.id, 'rollback', 'done', rollbackBefore, newPath]
);
await logAudit(req, 'entry.photo.rollback', { entry_id: req.params.id, job_id: jobId, restored_path: newPath });
invalidateEntries();
res.json({ ok: true, photo_path: newPath });
} catch (e) {
console.error('POST /api/entries/:id/photo/jobs/:jobId/rollback:', e);
res.status(500).json({ error: 'Ошибка отката фотографии' });
}
});
app.delete('/api/entries/:id/photo/enhance-ai/preview', requireAuth, async (req, res) => {
const { path: p } = req.body || {};
if (!isSafeUploadPath(p)) return res.status(400).json({ error: 'Некорректный путь' });
@@ -3861,9 +3936,22 @@ app.post('/api/entries/:id/photo/restore-original', requireAuth, async (req, res
await pool.query('UPDATE entries SET photo_path = $1, photo_original_path = NULL WHERE id = $2', [newPath, req.params.id]);
await pool.query('UPDATE entry_photos SET photo_path = $1 WHERE entry_id = $2 AND photo_path = $3', [newPath, req.params.id, oldPath]);
const oldAbs = path.join(UPLOADS_DIR, String(oldPath).replace(/^\/+/, '').replace(/^uploads\//, ''));
try { if (fs.existsSync(oldAbs)) fs.unlinkSync(oldAbs); } catch {}
let beforePath = null;
if (fs.existsSync(oldAbs)) {
try {
const backupName = crypto.randomBytes(12).toString('hex') + (path.extname(oldAbs) || '.jpg');
fs.renameSync(oldAbs, path.join(ORIGINALS_DIR, backupName));
beforePath = `/uploads/.originals/${backupName}`;
} catch (e) {
console.error('photo original backup failed:', e);
}
}
const oldThumb = path.join('uploads', '.thumbs', path.basename(oldPath).replace(/\.[^.]+$/, '') + '.webp');
safeUnlink(oldThumb);
await pool.query(
'INSERT INTO photo_jobs (entry_id, action, status, before_path, after_path, error, finished_at) VALUES ($1, $2, $3, $4, $5, NULL, now())',
[req.params.id, 'restore', 'done', beforePath, newPath]
);
await logAudit(req, 'entry.photo.restore_original', { entry_id: req.params.id, restored_path: newPath });
invalidateEntries();
res.json({ ok: true, photo_path: newPath });
@@ -4232,6 +4320,82 @@ app.post('/api/entries/:id/ai/revert', requireAuth, async (req, res) => {
res.json({ ok: true });
});
app.get('/api/photo-jobs/status', requireAdmin, async (_, res) => {
const { rows } = await pool.query(
`SELECT status, count(*)::int AS n FROM photo_jobs GROUP BY status`
);
const counts = { pending: 0, processing: 0, done: 0, error: 0 };
rows.forEach(r => { counts[r.status] = r.n; });
const [pending, recent, errors] = await Promise.all([
pool.query(
`SELECT j.id, j.entry_id, j.action, j.created_at, e.student_name, g.name AS group_name
FROM photo_jobs j
JOIN entries e ON e.id = j.entry_id
JOIN groups g ON g.id = e.group_id
WHERE j.status IN ('pending', 'processing')
ORDER BY j.id ASC LIMIT 20`
),
pool.query(
`SELECT j.id, j.entry_id, j.action, j.status, j.params, j.before_path, j.after_path, j.error, j.created_at, j.finished_at,
e.student_name, g.name AS group_name
FROM photo_jobs j
JOIN entries e ON e.id = j.entry_id
JOIN groups g ON g.id = e.group_id
WHERE j.finished_at IS NOT NULL
ORDER BY j.finished_at DESC LIMIT 30`
),
pool.query(
`SELECT j.id, j.entry_id, j.action, j.error, j.finished_at, e.student_name, g.name AS group_name
FROM photo_jobs j
JOIN entries e ON e.id = j.entry_id
JOIN groups g ON g.id = e.group_id
WHERE j.status = 'error'
ORDER BY j.finished_at DESC NULLS LAST, j.id DESC LIMIT 20`
),
]);
const enabled = String(await getSetting('photo_worker_enabled', 'true')) !== 'false';
const service = await aiHealthCheck();
res.json({
enabled,
ai_configured: !!PHOTO_AI_URL,
ai_url: PHOTO_AI_URL,
worker: photoWorker ? photoWorker.getInfo() : null,
counts,
pending: pending.rows,
recent: recent.rows,
errors: errors.rows,
});
});
app.post('/api/photo-jobs/wake', requireAdmin, async (req, res) => {
if (photoWorker) photoWorker.notify();
await logAudit(req, 'photo-jobs.wake', {});
res.json({ ok: true });
});
app.post('/api/photo-jobs/enabled', requireAdmin, async (req, res) => {
const enabled = !!req.body?.enabled;
await pool.query(
`INSERT INTO settings (key, value) VALUES ('photo_worker_enabled', $1)
ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value`,
[enabled ? 'true' : 'false']
);
if (enabled && photoWorker) photoWorker.notify();
await logAudit(req, 'photo-jobs.enabled', { enabled });
res.json({ ok: true, enabled });
});
app.post('/api/photo-jobs/requeue-failed', requireAdmin, async (req, res) => {
const { rowCount } = await pool.query(
`UPDATE photo_jobs SET status = 'pending', error = NULL, finished_at = NULL
WHERE status = 'error'`
);
if (photoWorker) photoWorker.notify();
await logAudit(req, 'photo-jobs.requeue-failed', { count: rowCount });
invalidateEntries();
res.json({ ok: true, count: rowCount });
});
app.get('/api/trash', requireAuth, async (req, res) => {
const { limit, offset } = req.query;
const bw = branchScope(req.user);
@@ -4351,8 +4515,21 @@ if (fs.existsSync(certPath) && fs.existsSync(keyPath)) {
try { await ensureEntryPhotosTable(); } catch (err) { console.error('Entry photos table:', err); }
try { await ensurePhotoOriginalColumn(); } catch (err) { console.error('Entry original photo column:', err); }
try { await ensureEntryAiColumns(); } catch (err) { console.error('Entry AI columns:', err); }
try { await ensurePhotoJobsTable(); } catch (err) { console.error('Photo jobs table:', err); }
try { await sweepOrphanedUploads(); } catch (err) { console.error('Upload sweep:', err); }
entryAutoChecker = createEntryAutoChecker({ pool, getSetting, logAudit, aiUrl: AI_URL, defaultPrompt: AI_DEFAULT_PROMPT });
entryAutoChecker.start();
console.log('AI auto-check worker started');
photoWorker = createPhotoEnhanceWorker({
pool,
getSetting,
logAudit,
invalidateEntries,
sharp,
photoAiUrl: PHOTO_AI_URL,
uploadsDir: UPLOADS_DIR,
originalsDir: ORIGINALS_DIR,
});
photoWorker.start();
console.log('Photo enhance worker started');
})();