Initial AI manga platform

workers/package.json

@@ -0,0 +1,17 @@
+{
+  "name": "workers",
+  "version": "0.1.0",
+  "private": true,
+  "scripts": {
+    "start": "node dist/main.js",
+    "start:dev": "tsx watch src/main.ts",
+    "build": "tsc -p tsconfig.build.json",
+    "lint": "tsc --noEmit -p tsconfig.json",
+    "typecheck": "tsc --noEmit -p tsconfig.json",
+    "test": "vitest run"
+  },
+  "dependencies": {
+    "bullmq": "^5.77.6",
+    "ioredis": "^5.11.0"
+  }
+}

workers/src/main.spec.ts

@@ -0,0 +1,16 @@
+import { describe, expect, it } from 'vitest';
+import { getWorkerStatus } from './main';
+
+describe('getWorkerStatus', () => {
+  it('defaults to mock mode', () => {
+    expect(getWorkerStatus()).toMatchObject({
+      service: 'queue-worker',
+      mode: 'mock',
+      backend: 'bullmq'
+    });
+    expect(getWorkerStatus().queues).toContainEqual({
+      name: 'image_queue',
+      processor: 'backend_delegate'
+    });
+  });
+});

workers/src/main.ts

@@ -0,0 +1,174 @@
+import { Worker, type ConnectionOptions, type Job } from 'bullmq';
+
+const QUEUE_NAMES = [
+  'novel_queue',
+  'parse_queue',
+  'story_queue',
+  'character_queue',
+  'episode_queue',
+  'script_queue',
+  'storyboard_queue',
+  'image_queue',
+  'audio_queue',
+  'subtitle_queue',
+  'video_queue',
+  'qc_queue',
+  'review_queue',
+  'analytics_queue'
+] as const;
+
+type QueueName = (typeof QUEUE_NAMES)[number];
+
+interface WorkerTaskPayload {
+  task_id?: string;
+  project_id?: string;
+  task_type?: string;
+  retry_count?: number;
+}
+
+interface WorkerRuntime {
+  backendUrl: string;
+  workerSecret: string;
+  redisUrl: string;
+  concurrency: number;
+}
+
+function sanitizeRedisUrl(redisUrl: string) {
+  try {
+    const parsed = new URL(redisUrl);
+
+    if (parsed.username) parsed.username = '***';
+    if (parsed.password) parsed.password = '***';
+
+    return parsed.toString();
+  } catch {
+    return 'redis://127.0.0.1:6379';
+  }
+}
+
+function runtimeFromEnv(): WorkerRuntime {
+  return {
+    backendUrl: trimTrailingSlash(process.env.WORKER_BACKEND_URL || 'http://127.0.0.1:3000/api'),
+    workerSecret: process.env.WORKER_SECRET || 'local_worker_secret_change_me',
+    redisUrl: process.env.REDIS_URL || 'redis://127.0.0.1:6379',
+    concurrency: normalizeConcurrency(process.env.WORKER_CONCURRENCY)
+  };
+}
+
+export function getWorkerStatus() {
+  const runtime = runtimeFromEnv();
+
+  return {
+    service: 'queue-worker',
+    mode: process.env.AI_PROVIDER_MODE ?? 'mock',
+    backend: 'bullmq',
+    backend_url: runtime.backendUrl,
+    redis_url: sanitizeRedisUrl(runtime.redisUrl),
+    concurrency: runtime.concurrency,
+    queues: QUEUE_NAMES.map((name) => ({
+      name,
+      processor: 'backend_delegate'
+    }))
+  };
+}
+
+export function startWorkers(runtime: WorkerRuntime = runtimeFromEnv()) {
+  const connection = createConnectionOptions(runtime.redisUrl);
+  const workers = QUEUE_NAMES.map(
+    (queueName) =>
+      new Worker(
+        queueName,
+        (job) => executeJob(queueName, job as Job<WorkerTaskPayload>, runtime),
+        {
+          connection,
+          concurrency: runtime.concurrency
+        }
+      )
+  );
+
+  return {
+    workers,
+    async close() {
+      await Promise.all(workers.map((worker) => worker.close()));
+    }
+  };
+}
+
+async function executeJob(queueName: QueueName, job: Job<WorkerTaskPayload>, runtime: WorkerRuntime) {
+  const taskId = job.data.task_id;
+
+  if (!taskId) {
+    throw new Error('Worker job missing task_id');
+  }
+
+  const response = await fetch(`${runtime.backendUrl}/internal/worker/tasks/${encodeURIComponent(taskId)}/execute`, {
+    method: 'POST',
+    headers: {
+      'content-type': 'application/json',
+      'x-worker-secret': runtime.workerSecret
+    },
+    body: JSON.stringify({
+      job_id: job.id,
+      queue_name: queueName,
+      retry_count: job.data.retry_count ?? 0
+    })
+  });
+  const payload = (await response.json().catch(() => null)) as
+    | { code?: number; message?: string; data?: unknown }
+    | null;
+
+  if (!response.ok || !payload || payload.code !== 0) {
+    throw new Error(payload?.message || `Backend worker execution failed: HTTP ${response.status}`);
+  }
+
+  return payload.data;
+}
+
+function createConnectionOptions(redisUrl: string): ConnectionOptions {
+  try {
+    const parsed = new URL(redisUrl);
+    const db = parsed.pathname ? Number(parsed.pathname.slice(1)) : 0;
+
+    return {
+      host: parsed.hostname || '127.0.0.1',
+      port: parsed.port ? Number(parsed.port) : 6379,
+      username: parsed.username || undefined,
+      password: parsed.password || undefined,
+      db: Number.isInteger(db) ? db : 0
+    };
+  } catch {
+    return {
+      host: '127.0.0.1',
+      port: 6379
+    };
+  }
+}
+
+function trimTrailingSlash(value: string) {
+  return value.replace(/\/+$/, '');
+}
+
+function normalizeConcurrency(value: string | undefined) {
+  const numberValue = Number(value ?? 2);
+
+  return Number.isInteger(numberValue) && numberValue >= 1 && numberValue <= 20 ? numberValue : 2;
+}
+
+if (require.main === module) {
+  const runner = startWorkers();
+
+  // eslint-disable-next-line no-console
+  console.log(JSON.stringify(getWorkerStatus(), null, 2));
+
+  const shutdown = async () => {
+    await runner.close();
+    process.exit(0);
+  };
+
+  process.on('SIGINT', () => {
+    void shutdown();
+  });
+  process.on('SIGTERM', () => {
+    void shutdown();
+  });
+}

workers/tsconfig.build.json

@@ -0,0 +1,7 @@
+{
+  "extends": "./tsconfig.json",
+  "exclude": [
+    "src/**/*.spec.ts",
+    "src/**/*.test.ts"
+  ]
+}

workers/tsconfig.json

@@ -0,0 +1,14 @@
+{
+  "extends": "../tsconfig.base.json",
+  "compilerOptions": {
+    "module": "CommonJS",
+    "outDir": "dist",
+    "rootDir": "src",
+    "types": [
+      "node"
+    ]
+  },
+  "include": [
+    "src/**/*.ts"
+  ]
+}