researchChat/lib/queue.js

const { EventEmitter } = require('node:events');

const events = new EventEmitter();
events.setMaxListeners(100);

let queue = [];
let running = false;

// Abort-Controller für laufende Jobs (für manuellen Abbruch)
const controllers = new Map();

/** Abort-Controller für eine Session registrieren (einmal pro Job). */
function registerController(sessionId) {
  const controller = new AbortController();
  controllers.set(sessionId, controller);
  return controller;
}

/** Laufenden Job einer Session abbrechen. */
function abortRunning(sessionId) {
  const controller = controllers.get(sessionId);
  if (controller) controller.abort();
  controllers.delete(sessionId);
}

/** Läuft für diese Session gerade ein Job im Speicher? */
function hasController(sessionId) {
  return controllers.has(sessionId);
}

/** Controller nach Job-Ende entfernen. */
function clearController(sessionId) {
  controllers.delete(sessionId);
}

/** Einen Job anmelden. Läuft sofort, wenn nichts anderes läuft, sonst wartet er. */
function enqueue(job) {
  queue.push(job);
  events.emit('job-queued', job.sessionId);
  processNext();
}

/** Job aus der Warteschlange holen und ausführen. */
async function processNext() {
  if (running || queue.length === 0) return;
  const job = queue.shift();
  running = true;
  try {
    await job.run();
  } catch (e) {
    job.onError?.(e);
  } finally {
    running = false;
    events.emit('job-finished', job.sessionId);
    processNext();
  }
}

function queuedBefore(sessionId) {
  return queue.findIndex((j) => j.sessionId === sessionId);
}

function isRunning(sessionId) {
  return running;
}

function cancelQueued(sessionId) {
  queue = queue.filter((j) => j.sessionId !== sessionId);
}

function state() {
  return { running, queued: queue.map((j) => j.sessionId) };
}

module.exports = { events, enqueue, queuedBefore, isRunning, cancelQueued, state, registerController, abortRunning, hasController, clearController };