diff --git a/src/modules/document-import/document-import.service.ts b/src/modules/document-import/document-import.service.ts index af72c16..e75bdc2 100644 --- a/src/modules/document-import/document-import.service.ts +++ b/src/modules/document-import/document-import.service.ts @@ -1,10 +1,12 @@ -import { Injectable } from '@nestjs/common'; +import { Injectable, Logger } from '@nestjs/common'; import { DocumentImportRepository } from './document-import.repository'; import { RedisService } from '../../infrastructure/redis/redis.service'; import { QueueService } from '../../infrastructure/queue/queue.service'; @Injectable() export class DocumentImportService { + private readonly logger = new Logger(DocumentImportService.name); + constructor( private readonly repository: DocumentImportRepository, private readonly redis: RedisService, @@ -18,30 +20,40 @@ export class DocumentImportService { sourceType?: string; rawText?: string; }) { - const lockKey = `lock:document-import:${dto.fileName || Date.now()}`; - const lockToken = await this.redis.lock(lockKey, 1800); - if (!lockToken) { - throw new Error('相同文件正在导入中,请稍候'); + this.logger.log(`createImport called: userId=${dto.userId}, kbId=${dto.knowledgeBaseId}, fileName=${dto.fileName}, sourceType=${dto.sourceType}`); + try { + const lockKey = `lock:document-import:${dto.fileName || Date.now()}`; + this.logger.log(`Attempting Redis lock: ${lockKey}`); + const lockToken = await this.redis.lock(lockKey, 1800); + if (!lockToken) { + throw new Error('相同文件正在导入中,请稍候'); + } + + this.logger.log(`Creating DB record...`); + const job = await this.repository.create(dto); + this.logger.log(`DB record created: ${job.id}`); + + await this.redis.set(`job:document-import:${job.id}:status`, 'pending', 86400); + await this.redis.set(`job:document-import:${job.id}:progress`, '0', 86400); + await this.redis.set(`job:document-import:${job.id}:message`, '任务已加入队列', 86400); + + this.logger.log(`Enqueuing job...`); + await this.queue.add('document-import', { + importId: job.id, + userId: dto.userId || 'anonymous', + knowledgeBaseId: dto.knowledgeBaseId, + rawText: dto.rawText, + fileName: dto.fileName, + }); + this.logger.log(`Job enqueued: ${job.id}`); + + await this.redis.unlock(lockKey, lockToken); + + return { jobId: job.id, status: 'queued' }; + } catch (err: any) { + this.logger.error(`createImport failed: ${err.message}`, err.stack); + throw err; } - - const job = await this.repository.create(dto); - - await this.redis.set(`job:document-import:${job.id}:status`, 'pending', 86400); - await this.redis.set(`job:document-import:${job.id}:progress`, '0', 86400); - await this.redis.set(`job:document-import:${job.id}:message`, '任务已加入队列', 86400); - - await this.queue.add('document-import', { - importId: job.id, - userId: dto.userId || 'anonymous', - knowledgeBaseId: dto.knowledgeBaseId, - rawText: dto.rawText, - fileName: dto.fileName, - }); - - // Release the lock — the worker will re-lock if needed - await this.redis.unlock(lockKey, lockToken); - - return { jobId: job.id, status: 'queued' }; } async getStatus(id: string) {