klarbild/src/worker.ts

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);
}

/** Monatliches Budget erreicht? (Summe usage.cost im laufenden Kalendermonat.) */
async function budgetExceeded(): Promise<boolean> {
  const s = await one<{ monthly_budget: number | null }>('SELECT monthly_budget FROM settings WHERE id=1');
  const cap = s?.monthly_budget ? Number(s.monthly_budget) : 0;
  if (!cap || cap <= 0) return false;
  const row = await one<{ spent: number }>(
    `SELECT COALESCE(sum(cost),0)::float AS spent FROM items
       WHERE cost IS NOT NULL AND created_at >= date_trunc('month', now())`);
  return (row?.spent || 0) >= cap;
}

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;
  }

  // Kostendeckel: bei Erreichen Auftrag anhalten, Position zurück in die Schlange.
  if (await budgetExceeded()) {
    await query(`UPDATE jobs SET status='paused' WHERE id=$1`, [job.jobId]);
    await query(`UPDATE items SET status='queued', error_message='Monatsbudget erreicht' WHERE id=$1`, [job.itemId]);
    console.error('[worker] Monatsbudget erreicht — Auftrag angehalten', job.jobId);
    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) {
    // Idempotent: nur der erste Übergang nach 'done' löst Auslieferung + Rückmeldung aus.
    const done = await query(
      `UPDATE jobs SET status='done', finished_at=now()
        WHERE id=$1 AND status NOT IN ('done','cancelled') RETURNING id`, [jobId]);
    if (done.length === 0) return;
    // 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); }
    }
  }
}