QIWEI-skill/runtime/callback-service/src/portrait-queue-worker.mjs

import path from 'node:path';
import { createRequire } from 'node:module';
import { PACKAGE_ROOT } from './config-loader.mjs';

const require = createRequire(import.meta.url);
const { processPortraitQueue } = require(path.join(
  PACKAGE_ROOT,
  'mcp',
  'src',
  'tools',
  'qiwei-portrait-tags-run.js',
));

export class PortraitQueueWorker {
  constructor({ intervalMs = 60_000, limit = 5, onState = () => {}, processor = processPortraitQueue } = {}) {
    this.intervalMs = intervalMs;
    this.limit = limit;
    this.onState = onState;
    this.processor = processor;
    this.running = false;
    this.timer = null;
    this.inFlight = null;
  }

  start() {
    if (this.running) return;
    this.running = true;
    this.onState({ portraitQueue: { status: 'running', lastError: '' } });
    void this.runOnce();
    this.timer = setInterval(() => void this.runOnce(), this.intervalMs);
    this.timer.unref?.();
  }

  async runOnce() {
    if (!this.running) return { skipped: true, reason: 'stopped' };
    if (this.inFlight) return { skipped: true, reason: 'in-flight' };

    this.inFlight = (async () => {
      try {
        const result = await this.processor(this.limit);
        this.onState({
          portraitQueue: {
            status: 'running',
            lastRunAt: new Date().toISOString(),
            processed: Number(result?.processed || 0),
            remaining: Number(result?.remaining || 0),
            lastError: '',
          },
        });
        return result;
      } catch (error) {
        this.onState({
          portraitQueue: {
            status: 'error',
            lastRunAt: new Date().toISOString(),
            lastError: error.message,
          },
        });
        return { processed: 0, error: error.message };
      } finally {
        this.inFlight = null;
      }
    })();

    return this.inFlight;
  }

  async stop() {
    this.running = false;
    if (this.timer) clearInterval(this.timer);
    this.timer = null;
    if (this.inFlight) await this.inFlight;
    this.onState({ portraitQueue: { status: 'stopped', stoppedAt: new Date().toISOString() } });
  }
}