192 lines
7.9 KiB
JavaScript
192 lines
7.9 KiB
JavaScript
const express = require('express');
|
|
const { v4: uuidv4 } = require('uuid');
|
|
|
|
function createTaskRoutes(db) {
|
|
const router = express.Router();
|
|
|
|
function parseJsonArray(value) {
|
|
try {
|
|
const parsed = JSON.parse(value || '[]');
|
|
return Array.isArray(parsed) ? parsed : [];
|
|
} catch (_) {
|
|
return [];
|
|
}
|
|
}
|
|
|
|
// List publish tasks
|
|
router.get('/', (req, res) => {
|
|
const { shop_id, status } = req.query;
|
|
let sql = 'SELECT * FROM publish_tasks WHERE 1=1';
|
|
const params = [];
|
|
if (shop_id) { sql += ' AND shop_id = ?'; params.push(shop_id); }
|
|
if (status) { sql += ' AND status = ?'; params.push(status); }
|
|
sql += ' ORDER BY created_at DESC';
|
|
const tasks = db.prepare(sql).all(...params);
|
|
res.json(tasks);
|
|
});
|
|
|
|
// Claim a task (extension calls this)
|
|
router.post('/claim', (req, res) => {
|
|
const { shop_id, claimed_by } = req.body;
|
|
if (!shop_id || !claimed_by) {
|
|
return res.status(400).json({ error: 'shop_id and claimed_by are required' });
|
|
}
|
|
const shop = db.prepare('SELECT id FROM shops WHERE id = ?').get(shop_id);
|
|
if (!shop) {
|
|
return res.status(404).json({ error: 'Shop not found' });
|
|
}
|
|
|
|
// Check daily limit
|
|
const limit = db.prepare("SELECT value FROM configs WHERE key = 'daily_publish_limit'").get();
|
|
const dailyLimit = parseInt(limit?.value || '20', 10);
|
|
const todayCount = db.prepare(
|
|
"SELECT COUNT(*) as cnt FROM publish_tasks WHERE shop_id = ? AND status = 'completed' AND date(created_at) = date('now')"
|
|
).get(shop_id);
|
|
if (todayCount.cnt >= dailyLimit) {
|
|
return res.status(429).json({ error: 'Daily publish limit reached', count: todayCount.cnt, limit: dailyLimit });
|
|
}
|
|
|
|
// Check consecutive failures
|
|
const failLimit = db.prepare("SELECT value FROM configs WHERE key = 'max_consecutive_failures'").get();
|
|
const maxFails = parseInt(failLimit?.value || '3', 10);
|
|
const recentFails = db.prepare(
|
|
"SELECT status FROM publish_tasks WHERE shop_id = ? ORDER BY created_at DESC LIMIT ?"
|
|
).all(shop_id, maxFails);
|
|
if (recentFails.length >= maxFails && recentFails.every(t => t.status === 'failed')) {
|
|
return res.status(423).json({ error: 'Too many consecutive failures, please check manually' });
|
|
}
|
|
|
|
db.prepare(
|
|
"UPDATE publish_tasks SET status = 'pending', claimed_by = NULL, claimed_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE shop_id = ? AND status IN ('claimed', 'executing') AND claimed_at < datetime('now', '-10 minutes')"
|
|
).run(shop_id);
|
|
db.prepare(
|
|
"UPDATE notes SET status = 'approved', worker_id = NULL, claimed_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE shop_id = ? AND status = 'publishing' AND claimed_at < datetime('now', '-10 minutes')"
|
|
).run(shop_id);
|
|
|
|
const claim = db.transaction(() => {
|
|
const task = db.prepare(
|
|
"SELECT * FROM publish_tasks WHERE shop_id = ? AND status = 'pending' ORDER BY created_at ASC LIMIT 1"
|
|
).get(shop_id);
|
|
if (!task) return null;
|
|
const result = db.prepare(
|
|
"UPDATE publish_tasks SET status = 'claimed', claimed_by = ?, claimed_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = 'pending'"
|
|
).run(claimed_by, task.id);
|
|
if (result.changes === 0) return null;
|
|
db.prepare("UPDATE notes SET status = 'publishing', worker_id = ?, claimed_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ?")
|
|
.run(claimed_by, task.note_id);
|
|
return task;
|
|
});
|
|
|
|
const task = claim();
|
|
if (!task) return res.status(404).json({ error: 'No pending tasks' });
|
|
|
|
const note = db.prepare(
|
|
'SELECT n.*, p.title AS product_title, s.shop_id AS external_shop_id, s.shop_name FROM notes n LEFT JOIN products p ON p.id = n.product_id LEFT JOIN shops s ON s.id = n.shop_id WHERE n.id = ?'
|
|
).get(task.note_id);
|
|
|
|
res.json({
|
|
task_id: task.id,
|
|
note: note ? {
|
|
...note,
|
|
topics: parseJsonArray(note.topics),
|
|
image_paths: parseJsonArray(note.image_paths),
|
|
} : null,
|
|
});
|
|
});
|
|
|
|
// Complete a task
|
|
router.post('/:id/complete', (req, res) => {
|
|
const task = db.prepare('SELECT * FROM publish_tasks WHERE id = ?').get(req.params.id);
|
|
if (!task) return res.status(404).json({ error: 'Task not found' });
|
|
if (task.status === 'completed') return res.json({ success: true, alreadyCompleted: true });
|
|
if (task.status !== 'claimed' && task.status !== 'executing') {
|
|
return res.status(409).json({ error: `Task is ${task.status}, cannot complete` });
|
|
}
|
|
if (req.body.claimed_by && task.claimed_by && req.body.claimed_by !== task.claimed_by) {
|
|
return res.status(403).json({ error: 'Task claimed by another worker' });
|
|
}
|
|
|
|
const complete = db.transaction(() => {
|
|
const result = db.prepare(
|
|
"UPDATE publish_tasks SET status = 'completed', updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status IN ('claimed', 'executing')"
|
|
).run(req.params.id);
|
|
if (result.changes === 0) return false;
|
|
db.prepare(
|
|
"UPDATE notes SET status = 'published', publish_result = ?, published_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ?"
|
|
).run(JSON.stringify(req.body.result || {}), task.note_id);
|
|
return true;
|
|
});
|
|
|
|
if (!complete()) return res.status(409).json({ error: 'Task state changed, cannot complete' });
|
|
res.json({ success: true });
|
|
});
|
|
|
|
// Fail a task
|
|
router.post('/:id/fail', (req, res) => {
|
|
const task = db.prepare('SELECT * FROM publish_tasks WHERE id = ?').get(req.params.id);
|
|
if (!task) return res.status(404).json({ error: 'Task not found' });
|
|
if (task.status === 'completed') {
|
|
return res.status(409).json({ error: 'Completed task cannot be marked failed' });
|
|
}
|
|
if (task.status !== 'claimed' && task.status !== 'executing') {
|
|
return res.status(409).json({ error: `Task is ${task.status}, cannot fail` });
|
|
}
|
|
if (req.body.claimed_by && task.claimed_by && req.body.claimed_by !== task.claimed_by) {
|
|
return res.status(403).json({ error: 'Task claimed by another worker' });
|
|
}
|
|
|
|
const errorMessage = String(req.body.error_message || 'Unknown error').slice(0, 1000);
|
|
const fail = db.transaction(() => {
|
|
const result = db.prepare(
|
|
"UPDATE publish_tasks SET status = 'failed', error_message = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status IN ('claimed', 'executing')"
|
|
).run(errorMessage, req.params.id);
|
|
if (result.changes === 0) return false;
|
|
db.prepare(
|
|
"UPDATE notes SET status = 'failed', error_message = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?"
|
|
).run(errorMessage, task.note_id);
|
|
return true;
|
|
});
|
|
|
|
if (!fail()) return res.status(409).json({ error: 'Task state changed, cannot fail' });
|
|
res.json({ success: true });
|
|
});
|
|
|
|
// Create tasks for approved notes
|
|
router.post('/create-batch', (req, res) => {
|
|
const { shop_id } = req.body;
|
|
if (!shop_id) return res.status(400).json({ error: 'shop_id required' });
|
|
|
|
const approvedNotes = db.prepare(
|
|
"SELECT id, shop_id FROM notes WHERE shop_id = ? AND status = 'approved'"
|
|
).all(shop_id);
|
|
|
|
if (approvedNotes.length === 0) {
|
|
return res.json({ count: 0, message: 'No approved notes found' });
|
|
}
|
|
|
|
const insert = db.prepare(
|
|
'INSERT INTO publish_tasks (id, note_id, shop_id) VALUES (?, ?, ?)'
|
|
);
|
|
const existingTaskNoteIds = new Set(
|
|
db.prepare("SELECT note_id FROM publish_tasks WHERE shop_id = ? AND status IN ('pending','claimed','executing')").all(shop_id).map(t => t.note_id)
|
|
);
|
|
|
|
let count = 0;
|
|
const batch = db.transaction(() => {
|
|
for (const note of approvedNotes) {
|
|
if (!existingTaskNoteIds.has(note.id)) {
|
|
insert.run(uuidv4(), note.id, note.shop_id);
|
|
count++;
|
|
}
|
|
}
|
|
});
|
|
batch();
|
|
|
|
res.json({ count, message: `Created ${count} new tasks` });
|
|
});
|
|
|
|
return router;
|
|
}
|
|
|
|
module.exports = createTaskRoutes;
|