import { PrismaClient, ProcessingStage, ProcessingStatus } from '@prisma/client'; const prisma = new PrismaClient(); /** * Centralized processing status management */ export class ProcessingStatusManager { /** * Initialize processing status for a session with all stages set to PENDING */ static async initializeSession(sessionId: string): Promise { const stages = [ ProcessingStage.CSV_IMPORT, ProcessingStage.TRANSCRIPT_FETCH, ProcessingStage.SESSION_CREATION, ProcessingStage.AI_ANALYSIS, ProcessingStage.QUESTION_EXTRACTION, ]; // Create all processing status records for this session await prisma.sessionProcessingStatus.createMany({ data: stages.map(stage => ({ sessionId, stage, status: ProcessingStatus.PENDING, })), skipDuplicates: true, // In case some already exist }); } /** * Start a processing stage */ static async startStage( sessionId: string, stage: ProcessingStage, metadata?: any ): Promise { await prisma.sessionProcessingStatus.upsert({ where: { sessionId_stage: { sessionId, stage } }, update: { status: ProcessingStatus.IN_PROGRESS, startedAt: new Date(), errorMessage: null, metadata: metadata || null, }, create: { sessionId, stage, status: ProcessingStatus.IN_PROGRESS, startedAt: new Date(), metadata: metadata || null, }, }); } /** * Complete a processing stage successfully */ static async completeStage( sessionId: string, stage: ProcessingStage, metadata?: any ): Promise { await prisma.sessionProcessingStatus.upsert({ where: { sessionId_stage: { sessionId, stage } }, update: { status: ProcessingStatus.COMPLETED, completedAt: new Date(), errorMessage: null, metadata: metadata || null, }, create: { sessionId, stage, status: ProcessingStatus.COMPLETED, startedAt: new Date(), completedAt: new Date(), metadata: metadata || null, }, }); } /** * Mark a processing stage as failed */ static async failStage( sessionId: string, stage: ProcessingStage, errorMessage: string, metadata?: any ): Promise { await prisma.sessionProcessingStatus.upsert({ where: { sessionId_stage: { sessionId, stage } }, update: { status: ProcessingStatus.FAILED, completedAt: new Date(), errorMessage, retryCount: { increment: 1 }, metadata: metadata || null, }, create: { sessionId, stage, status: ProcessingStatus.FAILED, startedAt: new Date(), completedAt: new Date(), errorMessage, retryCount: 1, metadata: metadata || null, }, }); } /** * Skip a processing stage (e.g., no transcript URL available) */ static async skipStage( sessionId: string, stage: ProcessingStage, reason: string ): Promise { await prisma.sessionProcessingStatus.upsert({ where: { sessionId_stage: { sessionId, stage } }, update: { status: ProcessingStatus.SKIPPED, completedAt: new Date(), errorMessage: reason, }, create: { sessionId, stage, status: ProcessingStatus.SKIPPED, startedAt: new Date(), completedAt: new Date(), errorMessage: reason, }, }); } /** * Get processing status for a specific session */ static async getSessionStatus(sessionId: string) { return await prisma.sessionProcessingStatus.findMany({ where: { sessionId }, orderBy: { stage: 'asc' }, }); } /** * Get sessions that need processing for a specific stage */ static async getSessionsNeedingProcessing( stage: ProcessingStage, limit: number = 50 ) { return await prisma.sessionProcessingStatus.findMany({ where: { stage, status: ProcessingStatus.PENDING, }, include: { session: { include: { import: true, company: true, }, }, }, take: limit, orderBy: { session: { createdAt: 'asc' } }, }); } /** * Get pipeline status overview */ static async getPipelineStatus() { // Get counts by stage and status const statusCounts = await prisma.sessionProcessingStatus.groupBy({ by: ['stage', 'status'], _count: { id: true }, }); // Get total sessions const totalSessions = await prisma.session.count(); // Organize the data const pipeline: Record> = {}; for (const { stage, status, _count } of statusCounts) { if (!pipeline[stage]) { pipeline[stage] = {}; } pipeline[stage][status] = _count.id; } return { totalSessions, pipeline, }; } /** * Get sessions with failed processing */ static async getFailedSessions(stage?: ProcessingStage) { const where: any = { status: ProcessingStatus.FAILED, }; if (stage) { where.stage = stage; } return await prisma.sessionProcessingStatus.findMany({ where, include: { session: { include: { import: true, }, }, }, orderBy: { completedAt: 'desc' }, }); } /** * Reset a failed stage for retry */ static async resetStageForRetry(sessionId: string, stage: ProcessingStage): Promise { await prisma.sessionProcessingStatus.update({ where: { sessionId_stage: { sessionId, stage } }, data: { status: ProcessingStatus.PENDING, startedAt: null, completedAt: null, errorMessage: null, }, }); } /** * Check if a session has completed a specific stage */ static async hasCompletedStage(sessionId: string, stage: ProcessingStage): Promise { const status = await prisma.sessionProcessingStatus.findUnique({ where: { sessionId_stage: { sessionId, stage } }, }); return status?.status === ProcessingStatus.COMPLETED; } /** * Check if a session is ready for a specific stage (previous stages completed) */ static async isReadyForStage(sessionId: string, stage: ProcessingStage): Promise { const stageOrder = [ ProcessingStage.CSV_IMPORT, ProcessingStage.TRANSCRIPT_FETCH, ProcessingStage.SESSION_CREATION, ProcessingStage.AI_ANALYSIS, ProcessingStage.QUESTION_EXTRACTION, ]; const currentStageIndex = stageOrder.indexOf(stage); if (currentStageIndex === 0) return true; // First stage is always ready // Check if all previous stages are completed const previousStages = stageOrder.slice(0, currentStageIndex); for (const prevStage of previousStages) { const isCompleted = await this.hasCompletedStage(sessionId, prevStage); if (!isCompleted) return false; } return true; } }