jobbi-bewerbung/scripts/jobsuche-runner.js

#!/usr/bin/env node
// Host-side runner for the per-user job search.
//
// Why it lives on the host and not in the app: the app container has neither
// `claude` nor `ollama`. The app therefore only *queues* runs (table suchlaeufe);
// this script — started by cron every few minutes — picks them up and executes
// them. It replaces the two old single-user scripts (jobsuche-cron.sh and
// jobsuche-remote-cron.sh), whose search criteria were hard-coded into the prompt.
//
// Per run it:
//   1. enqueues scheduled runs that are due (per user's Zeitplan),
//   2. claims one queued run at a time (status angefordert -> laeuft),
//   3. builds the prompt from that user's Suchprofil + their Lebenslauf,
//   4. runs the headless agent with THAT user's credentials:
//        - BEWERBUNG_API_KEY = the user's API token (created if missing), so every
//          imported Jobangebot lands in their account and passes the same
//          dedup/blacklist path as the rest of the app,
//        - the model is driven by the user's own Ollama key (see KI_AUTH below),
//   5. parses the agent's final "ERGEBNIS neu=… dubletten=… verworfen=…" line and
//      writes status + counts back, so the UI can show what happened.
//
// Everything is per user; nothing here reads a global .env any more.

const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const { spawn } = require('child_process');
const sqlite3 = require('sqlite3');

const suchprofil = require('../lib/suchprofil');

const DB_PATH = process.env.JOBSUCHE_DB || '/opt/jobbi-bewerbung/data/bewerbungen.db';
const LOG_DIR = process.env.JOBSUCHE_LOG_DIR || '/opt/jobbi-bewerbung/logs';
const API_URL = process.env.JOBSUCHE_API_URL || 'http://localhost:4327/api/v1';
// Swappable so the pipeline can be exercised end-to-end without burning tokens.
const CLAUDE_BIN = process.env.JOBSUCHE_CLAUDE_BIN || 'claude';
// 'benutzer' = each run uses that user's own Ollama key (they pay for their own
// search). 'host' = fall back to the machine's ollama login via `ollama launch`.
const KI_AUTH = process.env.JOBSUCHE_KI_AUTH || 'benutzer';
const RUN_TIMEOUT_MS = (Number(process.env.JOBSUCHE_TIMEOUT) || 2700) * 1000;
// A run that has been 'laeuft' for longer than this lost its process (reboot,
// OOM); it would otherwise block the user's queue forever.
const STALE_MS = RUN_TIMEOUT_MS + 10 * 60 * 1000;
const LOG_RETENTION_DAYS = Number(process.env.JOBSUCHE_LOG_RETENTION_DAYS) || 30;
const CFG = 'cfg:';

const db = new sqlite3.Database(DB_PATH);
const all = (sql, p = []) => new Promise((res, rej) => db.all(sql, p, (e, r) => (e ? rej(e) : res(r))));
const get = (sql, p = []) => new Promise((res, rej) => db.get(sql, p, (e, r) => (e ? rej(e) : res(r))));
const run = (sql, p = []) => new Promise((res, rej) => db.run(sql, p, function (e) { e ? rej(e) : res(this); }));

function log(msg) {
  console.log(`${new Date().toISOString()} ${msg}`);
}

async function cfgWert(userId, key) {
  const row = await get('SELECT value FROM app_state WHERE user_id = ? AND key = ?', [userId, CFG + key]);
  return row && row.value ? row.value : '';
}

// The user's API token, created on first use. Without it the agent could not
// import anything; making the user generate one by hand first would just be a
// trap ("profile active, nothing happens").
async function ensureApiKey(userId) {
  const vorhanden = await cfgWert(userId, 'API_TOKEN');
  if (vorhanden) return vorhanden;
  const token = crypto.randomBytes(32).toString('hex');
  await run(
    'INSERT INTO app_state (user_id, key, value) VALUES (?, ?, ?) ' +
    'ON CONFLICT(user_id, key) DO UPDATE SET value = excluded.value',
    [userId, CFG + 'API_TOKEN', token]
  );
  log(`  API-Key für Benutzer ${userId} automatisch erzeugt.`);
  return token;
}

// The CV text the search profile is derived from. Prefer the Lebenslauf, fall
// back to a Profil/Kurzprofil — the point is to give the model the user's actual
// roles and technologies rather than a hard-coded list.
async function profilText(userId) {
  const rows = await all(
    `SELECT typ, inhalt FROM basis_dokumente
      WHERE user_id = ? AND typ IN ('Lebenslauf', 'Profil/Kurzprofil')
        AND inhalt IS NOT NULL AND inhalt != ''
      ORDER BY CASE typ WHEN 'Lebenslauf' THEN 0 ELSE 1 END, id ASC`,
    [userId]
  );
  return rows.map((r) => (r.inhalt || '').trim()).filter(Boolean).join('\n\n---\n\n');
}

// 1. Queue scheduled runs that are due.
async function faelligeEinreihen() {
  const jetzt = new Date();
  const users = await all('SELECT user_id FROM suchprofil WHERE aktiv = 1');
  for (const { user_id: userId } of users) {
    const profil = suchprofil.fromRow(await get('SELECT * FROM suchprofil WHERE user_id = ?', [userId]));
    if (suchprofil.validate(profil).length) continue; // unbrauchbares Profil -> kein Zeitplan-Lauf

    // Only scheduled runs count here: a manual test run must not swallow the
    // day's automatic run.
    const letzter = await get(
      "SELECT angefordert_at FROM suchlaeufe WHERE user_id = ? AND ausloeser = 'zeitplan' ORDER BY id DESC LIMIT 1",
      [userId]
    );
    const letzterAt = letzter ? `${letzter.angefordert_at}Z`.replace(' ', 'T') : null;
    if (!suchprofil.istFaellig(profil, jetzt, letzterAt)) continue;

    const offen = await get(
      "SELECT id FROM suchlaeufe WHERE user_id = ? AND status IN ('angefordert','laeuft') LIMIT 1",
      [userId]
    );
    if (offen) continue; // läuft schon etwas für diesen Benutzer

    await run("INSERT INTO suchlaeufe (user_id, status, ausloeser) VALUES (?, 'angefordert', 'zeitplan')", [userId]);
    log(`Zeitplan: Lauf für Benutzer ${userId} eingereiht.`);
  }
}

// Runs whose process died: release them so the user is not stuck.
async function verwaisteAufraeumen() {
  const res = await run(
    `UPDATE suchlaeufe SET status = 'fehler', beendet_at = CURRENT_TIMESTAMP,
       fehler = 'Lauf abgebrochen (Prozess nicht mehr vorhanden).'
     WHERE status = 'laeuft'
       AND gestartet_at IS NOT NULL
       AND (julianday('now') - julianday(gestartet_at)) * 86400000 > ?`,
    [STALE_MS]
  );
  if (res.changes) log(`${res.changes} verwaiste(n) Lauf/Läufe als Fehler markiert.`);
}

// Claim the oldest queued run. The conditional UPDATE is the claim: a second
// runner starting concurrently gets changes = 0 and moves on.
async function naechstenLaufClaimen() {
  const kandidat = await get("SELECT * FROM suchlaeufe WHERE status = 'angefordert' ORDER BY id ASC LIMIT 1");
  if (!kandidat) return null;
  const res = await run(
    "UPDATE suchlaeufe SET status = 'laeuft', gestartet_at = CURRENT_TIMESTAMP WHERE id = ? AND status = 'angefordert'",
    [kandidat.id]
  );
  if (!res.changes) return null;
  return get('SELECT * FROM suchlaeufe WHERE id = ?', [kandidat.id]);
}

async function laufAbschliessen(id, status, felder = {}) {
  await run(
    `UPDATE suchlaeufe SET status = ?, beendet_at = CURRENT_TIMESTAMP,
       neu = ?, dubletten = ?, verworfen = ?, fehler = ?, log_datei = ?
     WHERE id = ?`,
    [status, felder.neu ?? null, felder.dubletten ?? null, felder.verworfen ?? null,
     felder.fehler || null, felder.log || null, id]
  );
}

// Spawn the headless agent and collect its output.
function agentStarten(prompt, env, logStream) {
  return new Promise((resolve) => {
    let argv;
    if (KI_AUTH === 'host') {
      // Legacy path: the machine's ollama login pays for the run.
      argv = ['launch', 'claude', '--model', env.JOBSUCHE_MODEL, '--',
        '--dangerously-skip-permissions', '-p', prompt];
      argv = { cmd: 'ollama', args: argv };
    } else {
      // Per-user path: Claude Code talks to Ollama's Anthropic-compatible API with
      // *this user's* key, so the search is billed to whoever requested it.
      argv = { cmd: CLAUDE_BIN, args: ['--dangerously-skip-permissions', '-p', prompt] };
    }

    const kind = spawn(argv.cmd, argv.args, {
      env,
      cwd: '/opt/jobbi-bewerbung',
      stdio: ['ignore', 'pipe', 'pipe'],
    });

    let ausgabe = '';
    const sammeln = (buf) => {
      const s = buf.toString();
      ausgabe += s;
      logStream.write(s);
    };
    kind.stdout.on('data', sammeln);
    kind.stderr.on('data', sammeln);

    const timer = setTimeout(() => {
      kind.kill('SIGTERM');
      setTimeout(() => kind.kill('SIGKILL'), 60000);
    }, RUN_TIMEOUT_MS);

    kind.on('error', (err) => {
      clearTimeout(timer);
      resolve({ code: -1, ausgabe, fehler: `Agent nicht startbar: ${err.message}` });
    });
    kind.on('close', (code, signal) => {
      clearTimeout(timer);
      const abgebrochen = signal === 'SIGTERM' || signal === 'SIGKILL';
      resolve({
        code,
        ausgabe,
        fehler: abgebrochen ? `Zeitlimit von ${RUN_TIMEOUT_MS / 1000}s überschritten.` : null,
      });
    });
  });
}

async function laufAusfuehren(lauf) {
  const userId = lauf.user_id;
  const user = await get('SELECT username FROM users WHERE id = ?', [userId]);
  const name = user ? user.username : `#${userId}`;
  log(`Lauf ${lauf.id} (Benutzer ${name}, ${lauf.ausloeser}) startet.`);

  const profil = suchprofil.fromRow(await get('SELECT * FROM suchprofil WHERE user_id = ?', [userId]));
  const fehler = suchprofil.validate(profil);
  if (fehler.length) {
    await laufAbschliessen(lauf.id, 'fehler', { fehler: fehler.join(' ') });
    log(`  abgebrochen: ${fehler.join(' ')}`);
    return;
  }

  const ollamaKey = await cfgWert(userId, 'OLLAMA_API_KEY');
  if (KI_AUTH !== 'host' && !ollamaKey) {
    const msg = 'Kein eigener Ollama-API-Schlüssel hinterlegt — der Suchlauf läuft über das KI-Kontingent des Benutzers. Bitte in den Einstellungen eintragen.';
    await laufAbschliessen(lauf.id, 'fehler', { fehler: msg });
    log(`  abgebrochen: kein Ollama-Key für ${name}.`);
    return;
  }

  const apiKey = await ensureApiKey(userId);
  const modell = (await cfgWert(userId, 'OLLAMA_MODEL')) || 'glm-5.2:cloud';
  const prompt = suchprofil.buildPrompt(profil, await profilText(userId));

  fs.mkdirSync(LOG_DIR, { recursive: true });
  const stamp = new Date().toISOString().replace(/[:.]/g, '-');
  const logDatei = path.join(LOG_DIR, `jobsuche_${name}_${stamp}.log`);
  const logStream = fs.createWriteStream(logDatei, { flags: 'a' });
  logStream.write(`=== Suchlauf ${lauf.id} — Benutzer ${name} (${lauf.ausloeser}) ===\n`);
  logStream.write(`Modus: ${profil.modus} | Städte: ${profil.staedte.join(', ') || '–'}\n`);
  logStream.write(`Modell: ${modell} | KI-Auth: ${KI_AUTH}\n${'-'.repeat(50)}\n`);

  const env = {
    ...process.env,
    HOME: process.env.HOME || '/root',
    PATH: `/root/.local/bin:/usr/local/bin:/usr/bin:/bin:${process.env.PATH || ''}`,
    IS_SANDBOX: '1',
    // The tracker API — the agent imports Jobangebote as THIS user.
    BEWERBUNG_API_URL: API_URL,
    BEWERBUNG_API_KEY: apiKey,
    JOBSUCHE_MODEL: modell,
  };
  if (KI_AUTH !== 'host') {
    // Point Claude Code at Ollama's Anthropic-compatible endpoint with the user's key.
    env.ANTHROPIC_BASE_URL = (await cfgWert(userId, 'OLLAMA_HOST')) || 'https://ollama.com';
    env.ANTHROPIC_AUTH_TOKEN = ollamaKey;
    env.ANTHROPIC_MODEL = modell;
    env.ANTHROPIC_SMALL_FAST_MODEL = modell;
  }

  const { code, ausgabe, fehler: laufFehler } = await agentStarten(prompt, env, logStream);
  const ergebnis = suchprofil.parseErgebnis(ausgabe);

  logStream.write(`\n${'-'.repeat(50)}\nExit-Code: ${code}\n`);
  logStream.end();

  if (laufFehler) {
    await laufAbschliessen(lauf.id, 'fehler', { fehler: laufFehler, log: logDatei });
    log(`  Lauf ${lauf.id} fehlgeschlagen: ${laufFehler}`);
    return;
  }
  if (code !== 0 && !ergebnis) {
    await laufAbschliessen(lauf.id, 'fehler', {
      fehler: `Suchlauf mit Exit-Code ${code} beendet (Details im Log).`,
      log: logDatei,
    });
    log(`  Lauf ${lauf.id} fehlgeschlagen (Exit ${code}).`);
    return;
  }
  if (!ergebnis) {
    // The agent finished but did not report its tally — treat as done with unknown
    // numbers rather than as a failure; the import itself already happened.
    await laufAbschliessen(lauf.id, 'fertig', { neu: 0, dubletten: 0, verworfen: 0, log: logDatei });
    log(`  Lauf ${lauf.id} beendet, aber ohne ERGEBNIS-Zeile.`);
    return;
  }

  await laufAbschliessen(lauf.id, 'fertig', { ...ergebnis, log: logDatei });
  log(`  Lauf ${lauf.id} fertig: ${ergebnis.neu} neu, ${ergebnis.dubletten} Dubletten, ${ergebnis.verworfen} verworfen.`);
}

function alteLogsAufraeumen() {
  try {
    const grenze = Date.now() - LOG_RETENTION_DAYS * 86400000;
    for (const f of fs.readdirSync(LOG_DIR)) {
      if (!f.startsWith('jobsuche_') || !f.endsWith('.log')) continue;
      const p = path.join(LOG_DIR, f);
      if (fs.statSync(p).mtimeMs < grenze) fs.unlinkSync(p);
    }
  } catch (e) { /* Aufräumen ist best effort */ }
}

async function main() {
  if (!fs.existsSync(DB_PATH)) {
    console.error(`Datenbank nicht gefunden: ${DB_PATH}`);
    process.exit(1);
  }
  await run('PRAGMA busy_timeout = 10000'); // die App schreibt parallel

  await verwaisteAufraeumen();
  await faelligeEinreihen();

  // Work the queue until it is empty. Runs are serialised on purpose: a search is
  // web- and token-heavy, and several in parallel would trip rate limits.
  let lauf;
  let anzahl = 0;
  while ((lauf = await naechstenLaufClaimen())) {
    await laufAusfuehren(lauf);
    anzahl += 1;
  }
  if (!anzahl) log('Nichts zu tun.');

  alteLogsAufraeumen();
  db.close();
}

main().catch((e) => {
  console.error('Runner-Fehler:', e);
  process.exit(1);
});