api-server/src/workers/document-import.worker.ts

import { Processor, WorkerHost } from '@nestjs/bullmq';
import { Logger } from '@nestjs/common';
import { Job } from 'bullmq';
import { QUEUE_DOCUMENT_IMPORT } from '../infrastructure/queue/queue.service';
import { DocumentImportRepository } from '../modules/document-import/document-import.repository';
import { KnowledgeItemsRepository } from '../modules/knowledge-items/knowledge-items.repository';
import { KnowledgeImportWorkflow } from '../modules/ai/workflows/knowledge-import.workflow';
import { RedisService } from '../infrastructure/redis/redis.service';

@Processor(QUEUE_DOCUMENT_IMPORT)
export class DocumentImportWorker extends WorkerHost {
  private readonly logger = new Logger(DocumentImportWorker.name);

  constructor(
    private readonly repository: DocumentImportRepository,
    private readonly knowledgeItemsRepo: KnowledgeItemsRepository,
    private readonly workflow: KnowledgeImportWorkflow,
    private readonly redis: RedisService,
  ) {
    super();
  }

  async process(job: Job<{
    importId: string;
    userId: string;
    knowledgeBaseId?: string;
    sourceId?: string;
    rawText?: string;
    fileName?: string;
  }>) {
    const { importId, userId, knowledgeBaseId, rawText, fileName } = job.data;
    // Resolve sourceId: from job data first, then fallback to DocumentImport record
    let sourceId = job.data.sourceId;
    if (!sourceId) {
      const importRecord = await this.repository.findById(importId);
      sourceId = importRecord?.sourceId ?? undefined;
    }
    this.logger.log(`Processing document import job ${job.id}, importId=${importId}, sourceId=${sourceId ?? 'none'}`);

    try {
      if (!rawText) {
        await this.repository.updateStatus(importId, 'completed');
        await this.redis.set(`job:document-import:${importId}:status`, 'completed', 86400);
        await this.redis.set(`job:document-import:${importId}:progress`, '100', 86400);
        await this.redis.set(`job:document-import:${importId}:message`, '无需解析的空文件', 86400);
        return;
      }

      await this.repository.updateStatus(importId, 'processing');
      await this.redis.set(`job:document-import:${importId}:status`, 'parsing', 86400);
      await this.redis.set(`job:document-import:${importId}:progress`, '25', 86400);
      await this.redis.set(`job:document-import:${importId}:message`, 'AI 正在分析文本，提取知识点...', 86400);

      const result = await this.workflow.execute({
        userId,
        rawText,
        sourceName: fileName,
      });

      await this.redis.set(`job:document-import:${importId}:status`, 'saving', 86400);
      await this.redis.set(`job:document-import:${importId}:progress`, '80', 86400);
      await this.redis.set(`job:document-import:${importId}:message`, `正在保存 ${result.knowledgePoints.length} 个知识点...`, 86400);

      if (knowledgeBaseId && result.knowledgePoints.length > 0) {
        for (let i = 0; i < result.knowledgePoints.length; i++) {
          const kp = result.knowledgePoints[i];
          await this.knowledgeItemsRepo.create(userId, knowledgeBaseId, {
            title: kp.title,
            content: kp.content,
            itemType: 'lesson',
            orderIndex: kp.suggestedOrder ?? i + 1,
            sourceId: sourceId,
            sourceRef: sourceId,
          });
        }
      }

      await this.repository.updateStatus(importId, 'completed');
      await this.redis.set(`job:document-import:${importId}:status`, 'completed', 86400);
      await this.redis.set(`job:document-import:${importId}:progress`, '100', 86400);
      await this.redis.set(`job:document-import:${importId}:message`, `成功提取 ${result.knowledgePoints.length} 个知识点`, 86400);
      this.logger.log(`Document import job ${job.id} completed: ${result.knowledgePoints.length} knowledge points`);
    } catch (err: any) {
      this.logger.error(`Document import job ${job.id} failed: ${err.message}`);
      await this.repository.updateStatus(importId, 'failed');
      await this.redis.set(`job:document-import:${importId}:status`, 'failed', 86400);
      await this.redis.set(`job:document-import:${importId}:message`, `导入失败: ${err.message}`, 86400);
      throw err;
    }
  }
}