ICON-CRM/server/src/scanner/pipeline.js

import { pool } from '../db.js';
import { SOURCES, sourceReady } from './sources.js';
import { agentTerms, findCompany, normalizeCompany, scoreJob, stripQuotes } from './matching.js';
import { loadScannerSecrets, setLastScanAt } from './settings.js';

/*
 * Ein Suchlauf.
 *
 *   1. aktive Agenten laden
 *   2. je Boerse die Schlagwoerter aller Agenten zusammenfuehren - dasselbe
 *      Wort wird nur einmal abgefragt, egal wie viele Agenten es tragen
 *   3. Anzeigen abrufen - jeden Alters (das Dashboard filtert nach Alter)
 *   4. Anzeigen speichern, Firma gegen das CRM abgleichen
 *   5. jede Anzeige gegen jeden Agenten bewerten, der diese Boerse durchsucht
 *
 * Zwei Aufrufer (Zeitplan und Knopf "Jetzt scannen") teilen sich die Sperre
 * unten. Lehre aus dem perinco-Scanner (26.08.2026): ohne sie stapeln sich die
 * Laeufe. Das Freigeben steht zwingend im finally, sonst bliebe die Sperre
 * nach einer Ausnahme fuer immer stehen.
 */

let running = false;
let runningSince = null;
// Notventil: haengt ein Lauf, gibt der naechste Aufruf die Sperre frei.
const MAX_RUNTIME_MS = 30 * 60 * 1000;
// Pause zwischen zwei Abrufen derselben Boerse - freundlich bleiben.
const REQUEST_GAP_MS = 1500;
// Anzeigen, die so lange nicht mehr gesehen wurden, fliegen samt Treffern raus.
const KEEP_DAYS = 60;

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

export function scanStatus() {
  return { running, since: runningSince ? new Date(runningSince).toISOString() : null };
}

export async function runScan(trigger = 'manual') {
  if (running && Date.now() - runningSince < MAX_RUNTIME_MS) {
    return { skipped: true, reason: 'Es läuft bereits ein Suchlauf.' };
  }
  running = true;
  runningSince = Date.now();
  try {
    return await execute(trigger);
  } finally {
    running = false;
    runningSince = null;
  }
}

async function execute(trigger) {
  const started = new Date();
  // Beginn des Laufs, nicht sein Ende: das Intervall ist ein Takt.
  await setLastScanAt(started);

  const [runResult] = await pool.query('INSERT INTO icon_scanner_runs (trigger_kind) VALUES (?)', [trigger]);
  const runId = runResult.insertId;

  const secrets = await loadScannerSecrets();
  const agents = await loadActiveAgents();

  // Boerse -> Schlagwoerter inkl. Alarm-Schlagwoerter (ohne Doppelte)
  const plan = new Map();
  for (const agent of agents) {
    for (const source of agent.sources) {
      if (!SOURCES[source] || !sourceReady(source, secrets)) continue;
      const words = plan.get(source) ?? new Map();
      for (const keyword of agentTerms(agent)) words.set(keyword.toLowerCase(), keyword);
      plan.set(source, words);
    }
  }

  const perSource = {};
  const jobs = new Map();

  for (const [source, words] of plan) {
    const stats = { ok: true, count: 0, errors: [] };
    let first = true;
    for (const keyword of words.values()) {
      if (!first) await sleep(REQUEST_GAP_MS);
      first = false;
      try {
        // Seit 29.09.2026 ohne Altersgrenze: auch aeltere Anzeigen kommen
        // herein, gefiltert wird im Dashboard ("Anzeigenalter").
        const found = await SOURCES[source].fetch(keyword, { secrets });
        for (const job of found) {
          jobs.set(`${job.source}:${job.externalId}`, job);
          stats.count += 1;
        }
      } catch (err) {
        stats.ok = false;
        stats.errors.push(`${keyword}: ${err.message}`);
      }
    }
    // Teilweise Fehler sind kein Totalausfall - nur, wenn gar nichts kam.
    if (stats.count > 0) stats.ok = true;
    stats.errors = stats.errors.slice(0, 5);
    perSource[source] = stats;
  }

  const companies = await loadCompanyKeys();
  let matchesNew = 0;

  for (const job of jobs.values()) {
    const jobId = await saveJob(job, findCompany(companies, job.company));
    for (const agent of agents) {
      if (!agent.sources.includes(job.source)) continue;
      const result = scoreJob(agent, job);
      if (!result) continue;
      const [insert] = await pool.query(
        `INSERT IGNORE INTO icon_scanner_matches (agent_id, job_id, score, hits, alarm) VALUES (?, ?, ?, ?, ?)`,
        [agent.id, jobId, result.score, JSON.stringify(result.hits), result.alarm ? 1 : 0]
      );
      matchesNew += insert.affectedRows;
    }
  }

  // Seit dem letzten Lauf angelegte Unternehmen auch bei aelteren Anzeigen erkennen.
  await relinkCompanies();
  await cleanup();

  await pool.query(
    `UPDATE icon_scanner_runs
        SET finished_at = CURRENT_TIMESTAMP, jobs_found = ?, matches_new = ?, per_source = ?
      WHERE id = ?`,
    [jobs.size, matchesNew, JSON.stringify(perSource), runId]
  );

  console.log(
    `[Scanner] Lauf ${runId} (${trigger}): ${jobs.size} Anzeigen, ${matchesNew} neue Treffer, ` +
      Object.entries(perSource)
        .map(([source, s]) => `${source}=${s.ok ? s.count : 'Fehler'}`)
        .join(' ')
  );

  return { skipped: false, runId, jobs: jobs.size, matchesNew, perSource };
}

export function parseList(value) {
  try {
    const list = JSON.parse(value ?? '[]');
    return Array.isArray(list) ? list.map(stripQuotes).filter(Boolean) : [];
  } catch {
    return [];
  }
}

async function loadActiveAgents() {
  const [rows] = await pool.query('SELECT * FROM icon_scanner_agents WHERE is_active = 1');
  return rows.map((row) => ({
    id: row.id,
    keywords: parseList(row.keywords),
    alarmKeywords: parseList(row.alarm_keywords),
    alarmMinScore: Number(row.alarm_min_score ?? 0),
    sources: parseList(row.sources),
    excludes: parseList(row.excludes),
    excludeCompanies: parseList(row.exclude_companies),
  }));
}

async function loadCompanyKeys() {
  const [rows] = await pool.query('SELECT id, name FROM icon_companies');
  return rows
    .map((row) => ({ id: row.id, key: normalizeCompany(row.name) }))
    .filter((row) => row.key)
    .sort((a, b) => b.key.length - a.key.length);
}

/** Speichert oder aktualisiert eine Anzeige und liefert ihre Kennung. */
async function saveJob(job, companyId) {
  const cut = (value, max) => (value == null ? null : String(value).slice(0, max));
  // LAST_INSERT_ID(id) im Update-Zweig: so kommt auch bei einer vorhandenen
  // Anzeige deren Kennung als insertId zurueck - ohne zweite Abfrage.
  const [result] = await pool.query(
    `INSERT INTO icon_scanner_jobs
       (source, external_id, title, company, company_id, location, country,
        employment_type, url, description, published_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
     ON DUPLICATE KEY UPDATE
       id = LAST_INSERT_ID(id),
       title = VALUES(title),
       company = VALUES(company),
       company_id = VALUES(company_id),
       location = VALUES(location),
       employment_type = COALESCE(VALUES(employment_type), employment_type),
       description = COALESCE(VALUES(description), description),
       published_at = COALESCE(published_at, VALUES(published_at)),
       last_seen_at = CURRENT_TIMESTAMP`,
    [
      job.source,
      cut(job.externalId, 190),
      cut(job.title, 500),
      cut(job.company, 300),
      companyId,
      cut(job.location, 300),
      job.country,
      cut(job.employmentType, 60),
      cut(job.url, 1000),
      cut(job.description, 60_000),
      job.publishedAt,
    ]
  );
  return result.insertId;
}

/*
 * Alte Anzeigen raeumen. Was jemand "in Bearbeitung" hat, bleibt stehen.
 */
async function cleanup() {
  await pool.query(
    `DELETE j FROM icon_scanner_jobs j
      WHERE j.last_seen_at < NOW() - INTERVAL ? DAY
        AND NOT EXISTS (SELECT 1 FROM icon_scanner_matches m
                         WHERE m.job_id = j.id AND m.status = 'in_bearbeitung')`,
    [KEEP_DAYS]
  );
  await pool.query('DELETE FROM icon_scanner_runs WHERE started_at < NOW() - INTERVAL 30 DAY');
}

/**
 * Nach dem Anlegen oder Umbenennen von Unternehmen: vorhandene Anzeigen neu
 * zuordnen. Billig - reiner Namensvergleich.
 */
export async function relinkCompanies() {
  const companies = await loadCompanyKeys();
  const [jobs] = await pool.query('SELECT id, company, company_id FROM icon_scanner_jobs WHERE company IS NOT NULL');
  let changed = 0;
  for (const job of jobs) {
    const id = findCompany(companies, job.company);
    if (id === job.company_id) continue;
    await pool.query('UPDATE icon_scanner_jobs SET company_id = ? WHERE id = ?', [id, job.id]);
    changed += 1;
  }
  return changed;
}