till
d680b3f67a
- pairing (admin code -> chat link), bundling of forwarded images via drafts+quiet-window sweeper - recipe pick via inline keyboard, job (origin=telegram) creation, one summary reply with result files + links - compressed-photo detection/flag; /start /help /rezepte /status - webhook route (secret token), admin pairing-code + links endpoints, AdminApp telegram section - worker notifies chat on completion; migration 002 adds jobs.telegram_chat_id; setupTelegram on init
70 lines
3.3 KiB
TypeScript
70 lines
3.3 KiB
TypeScript
import { one, query } from './lib/db';
|
|
import { startWorker, type GenerateJob } from './lib/queue';
|
|
import { processItem } from './lib/process';
|
|
import { deliverPendingForJob } from './lib/delivery';
|
|
|
|
/** Startet den Warteschlangen-Arbeiter (im selben Node-Prozess wie die App). */
|
|
export async function startImageWorker(): Promise<void> {
|
|
const s = await one<{ concurrency: number }>('SELECT concurrency FROM settings WHERE id=1');
|
|
const concurrency = s?.concurrency ?? 2;
|
|
await startWorker(concurrency, handle);
|
|
}
|
|
|
|
async function handle(job: GenerateJob): Promise<void> {
|
|
// Angehaltene/abgebrochene Aufträge nicht verarbeiten.
|
|
const j = await one<{ status: string }>('SELECT status FROM jobs WHERE id=$1', [job.jobId]);
|
|
if (!j || j.status === 'paused' || j.status === 'cancelled') {
|
|
await query(`UPDATE items SET status='queued' WHERE id=$1 AND status='running'`, [job.itemId]);
|
|
return;
|
|
}
|
|
await query(`UPDATE jobs SET status='running' WHERE id=$1 AND status='queued'`, [job.jobId]);
|
|
|
|
try {
|
|
const res = await processItem(job.itemId);
|
|
if (res.ok) {
|
|
await query(`UPDATE jobs SET done_count=done_count+1 WHERE id=$1`, [job.jobId]);
|
|
} else {
|
|
await query(`UPDATE jobs SET failed_count=failed_count+1 WHERE id=$1`, [job.jobId]);
|
|
}
|
|
} catch (e: any) {
|
|
if (e?.status === 402) {
|
|
// Guthaben erschöpft → Auftrag anhalten, Position zurück in die Schlange.
|
|
await query(`UPDATE jobs SET status='paused' WHERE id=$1`, [job.jobId]);
|
|
await query(`UPDATE items SET status='queued', error_message='Guthaben erschöpft' WHERE id=$1`,
|
|
[job.itemId]);
|
|
console.error('[worker] 402 — Auftrag angehalten', job.jobId);
|
|
return;
|
|
}
|
|
await query(`UPDATE items SET status='failed', error_message=$2 WHERE id=$1`,
|
|
[job.itemId, String(e?.message || e).slice(0, 500)]);
|
|
await query(`UPDATE jobs SET failed_count=failed_count+1 WHERE id=$1`, [job.jobId]);
|
|
}
|
|
|
|
await maybeComplete(job.jobId);
|
|
}
|
|
|
|
/** Auftrag abschließen, wenn alle Positionen fertig/fehlgeschlagen sind. */
|
|
async function maybeComplete(jobId: string): Promise<void> {
|
|
const row = await one<{ total: number; done: number; failed: number; open: number }>(
|
|
`SELECT total,
|
|
(SELECT count(*) FROM items WHERE job_id=$1 AND status='done')::int AS done,
|
|
(SELECT count(*) FROM items WHERE job_id=$1 AND status='failed')::int AS failed,
|
|
(SELECT count(*) FROM items WHERE job_id=$1 AND status IN ('queued','running'))::int AS open
|
|
FROM jobs WHERE id=$1`, [jobId]);
|
|
if (row && row.open === 0) {
|
|
await query(`UPDATE jobs SET status='done', finished_at=now() WHERE id=$1 AND status<>'cancelled'`,
|
|
[jobId]);
|
|
// Picdrop-Auslieferung für alle offenen Positionen anstoßen.
|
|
try { await deliverPendingForJob(jobId); } catch (e) { console.error('[worker] Auslieferung:', e); }
|
|
// Telegram-Rückmeldung, wenn der Auftrag aus einem Chat kam.
|
|
const jt = await one<{ origin: string; telegram_chat_id: string | null }>(
|
|
'SELECT origin, telegram_chat_id FROM jobs WHERE id=$1', [jobId]);
|
|
if (jt?.origin === 'telegram' && jt.telegram_chat_id) {
|
|
try {
|
|
const { notifyJobDone } = await import('./lib/telegram');
|
|
await notifyJobDone(Number(jt.telegram_chat_id), jobId);
|
|
} catch (e) { console.error('[worker] Telegram-Notify:', e); }
|
|
}
|
|
}
|
|
}
|