Low Level Design-(LLD)---fluxx hqx classifier

// src/features/classifications/v1/routes/classifications.routes.ts
import { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify';
import { ClassificationService } from '../services/classification.service';
import { JobStatusService } from '../services/job-status.service';
import { ClassifyRequestDTO } from '../models/dto/classify-request.dto';
import { Logger } from 'framework-nodejs-extensions';
interface ClassificationsRouteHandlerDeps {
  classificationService: ClassificationService;
  jobStatusService: JobStatusService;
  logger: Logger;
}
export async function classificationsRoutes(
  fastify: FastifyInstance,
  opts: ClassificationsRouteHandlerDeps
) {
  const { classificationService, jobStatusService, logger } = opts;
  // POST /api/v1/Classifications
  fastify.post<{
    Body: ClassifyRequestDTO;
  }>(
    '/api/v1/Classifications',
    {
      schema: {
        description: 'Submit PDF for asynchronous classification',
        tags: ['Classification'],
        consumes: ['multipart/form-data'],
        body: {
          type: 'object',
          required: ['file', 'classificationPrompt'],
          properties: {
            file: {
              type: 'string',
              format: 'binary',
              description: 'PDF file to classify',
            },
            classificationPrompt: {
              type: 'string',
              description: 'LLM prompt for classification',
              minLength: 10,
              maxLength: 2000,
            },
          },
        },
        response: {
          201: {
            description: 'Classification job created',
            type: 'object',
            properties: {
              jobId: { type: 'string', format: 'uuid' },
              status: { type: 'string', enum: ['in-progress'] },
              createdAt: { type: 'string', format: 'date-time' },
              statusUrl: { type: 'string' },
            },
          },
          400: { $ref: '#/components/schemas/Error' },
          413: { $ref: '#/components/schemas/Error' },
          422: { $ref: '#/components/schemas/Error' },
        },
      },
      config: {
        authRequired: true,
        allowedScopes: ['classifier.write'],
      },
    },
    async (request: FastifyRequest, reply: FastifyReply) => {
      logger.info('Received classification request', {
        correlationId: request.id,
      });
      // Extract file from multipart request
      const data = await request.file();
      
      if (!data) {
        return reply.status(400).send({
          error: 'MISSING_FILE',
          message: 'PDF file is required',
        });
      }
      // Validate file type
      if (data.mimetype !== 'application/pdf') {
        return reply.status(400).send({
          error: 'INVALID_FILE_TYPE',
          message: 'Only PDF files are supported',
        });
      }
      // Extract classification prompt from fields
      const fields = data.fields as any;
      const classificationPrompt = fields?.classificationPrompt?.value;
      if (!classificationPrompt) {
        return reply.status(400).send({
          error: 'MISSING_PROMPT',
          message: 'classificationPrompt is required',
        });
      }
      // Create classification job
      const result = await classificationService.createJob({
        fileBuffer: await data.toBuffer(),
        fileName: data.filename,
        classificationPrompt,
        userId: request.user?.id,
      });
      reply.status(201)
        .header('Location', `/api/v1/Classifications/${result.jobId}`)
        .send(result);
    }
  );
  // GET /api/v1/Classifications/:jobId
  fastify.get<{
    Params: { jobId: string };
  }>(
    '/api/v1/Classifications/:jobId',
    {
      schema: {
        description: 'Get classification job status',
        tags: ['Status'],
        params: {
          type: 'object',
          required: ['jobId'],
          properties: {
            jobId: {
              type: 'string',
              format: 'uuid',
              description: 'Classification job ID',
            },
          },
        },
        response: {
          200: {
            description: 'Job status retrieved',
            type: 'object',
            properties: {
              jobId: { type: 'string', format: 'uuid' },
              status: {
                type: 'string',
                enum: ['in-progress', 'success', 'failed'],
              },
              createdAt: { type: 'string', format: 'date-time' },
              updatedAt: { type: 'string', format: 'date-time' },
              completedAt: { type: 'string', format: 'date-time' },
              result: { type: 'object' },
              error: { type: 'object' },
            },
          },
          404: { $ref: '#/components/schemas/Error' },
        },
      },
      config: {
        authRequired: true,
        allowedScopes: ['classifier.read'],
      },
    },
    async (request: FastifyRequest<{ Params: { jobId: string } }>, reply: FastifyReply) => {
      const { jobId } = request.params;
      logger.info('Fetching job status', { jobId });
      const status = await jobStatusService.getJobStatus(jobId);
      if (!status) {
        return reply.status(404).send({
          error: 'JOB_NOT_FOUND',
          message: `Job with ID ${jobId} not found`,
        });
      }
      reply.send(status);
    }
  );
}

// src/features/classifications/v1/services/classification.service.ts
import { v4 as uuidv4 } from 'uuid';
import { Logger, ConfigClient } from 'framework-nodejs-extensions';
import { BlobStorageClient } from 'framework-nodejs-storage';
import { ServiceBusClient } from 'framework-nodejs-pubsub';
import { JobRepository } from '../repositories/job.repository';
import { ClassifyRequestDTO } from '../models/dto/classify-request.dto';
import { ClassifyResponseDTO } from '../models/dto/classify-response.dto';
import { JobStatus } from '../models/job.model';
export class ClassificationService {
  constructor(
    private logger: Logger,
    private blobStorageClient: BlobStorageClient,
    private serviceBusClient: ServiceBusClient,
    private jobRepository: JobRepository,
    private configClient: ConfigClient
  ) {}
  async createJob(request: ClassifyRequestDTO): Promise<ClassifyResponseDTO> {
    const jobId = uuidv4();
    const timestamp = new Date();
    this.logger.info('Creating classification job', { jobId });
    try {
      // 1. Validate file size
      const maxFileSizeMB = await this.configClient.get<number>('MAX_PDF_SIZE_MB') || 50;
      const fileSizeBytes = request.fileBuffer.length;
      const fileSizeMB = fileSizeBytes / (1024 * 1024);
      if (fileSizeMB > maxFileSizeMB) {
        throw new Error(`File size ${fileSizeMB.toFixed(2)}MB exceeds maximum ${maxFileSizeMB}MB`);
      }
      // 2. Upload PDF to blob storage
      const blobPath = `uploaded-pdfs/${jobId}/${request.fileName}`;
      await this.blobStorageClient.upload(
        'hqx-classification-storage',
        blobPath,
        request.fileBuffer
      );
      this.logger.info('PDF uploaded to blob storage', { jobId, blobPath });
      // 3. Create job record in database
      await this.jobRepository.create({
        jobId,
        userId: request.userId,
        fileName: request.fileName,
        fileSizeBytes,
        pdfBlobPath: blobPath,
        classificationPrompt: request.classificationPrompt,
        status: JobStatus.InProgress,
        createdAt: timestamp,
        updatedAt: timestamp,
      });
      this.logger.info('Job record created in database', { jobId });
      // 4. Send message to Service Bus queue
      const queueMessage = {
        jobId,
        pdfBlobPath: blobPath,
        classificationPrompt: request.classificationPrompt,
        createdAt: timestamp.toISOString(),
      };
      await this.serviceBusClient.sendMessage('hqx-classification-jobs', queueMessage);
      this.logger.info('Job message sent to queue', { jobId });
      // 5. Return response
      return {
        jobId,
        status: 'in-progress',
        createdAt: timestamp.toISOString(),
        statusUrl: `/api/v1/Classifications/${jobId}`,
      };
    } catch (error) {
      this.logger.error('Failed to create classification job', { jobId, error });
      
      // Attempt cleanup
      await this.jobRepository.updateStatus(jobId, JobStatus.Failed, {
        errorCode: 'JOB_CREATION_FAILED',
        errorMessage: error.message,
      });
      throw error;
    }
  }
}

// src/features/classifications/v1/services/job-status.service.ts
import { Logger } from 'framework-nodejs-extensions';
import { BlobStorageClient } from 'framework-nodejs-storage';
import { JobRepository } from '../repositories/job.repository';
import { ClassificationResultRepository } from '../repositories/classification-result.repository';
import { StatusResponseDTO } from '../models/dto/status-response.dto';
import { JobStatus } from '../models/job.model';
export class JobStatusService {
  constructor(
    private logger: Logger,
    private jobRepository: JobRepository,
    private classificationResultRepository: ClassificationResultRepository,
    private blobStorageClient: BlobStorageClient
  ) {}
  async getJobStatus(jobId: string): Promise<StatusResponseDTO | null> {
    this.logger.info('Retrieving job status', { jobId });
    // Fetch job from database
    const job = await this.jobRepository.findById(jobId);
    if (!job) {
      this.logger.warn('Job not found', { jobId });
      return null;
    }
    // Build base response
    const response: StatusResponseDTO = {
      jobId: job.jobId,
      status: job.status,
      createdAt: job.createdAt.toISOString(),
      updatedAt: job.updatedAt.toISOString(),
    };
    // Add status-specific data
    switch (job.status) {
      case JobStatus.InProgress:
        response.progress = {
          totalPages: job.totalPages || 0,
          processedPages: job.processedPages || 0,
          percentComplete: job.totalPages 
            ? Math.round((job.processedPages / job.totalPages) * 100)
            : 0,
        };
        break;
      case JobStatus.Success:
        response.completedAt = job.completedAt?.toISOString();
        
        // Fetch classification result
        const result = await this.classificationResultRepository.findByJobId(jobId);
        if (result) {
          response.result = result.classificationData;
        }
        break;
      case JobStatus.Failed:
        response.failedAt = job.failedAt?.toISOString();
        response.error = {
          code: job.errorCode,
          message: job.errorMessage,
          details: job.errorDetails,
        };
        break;
    }
    return response;
  }
}

// src/features/classifications/v1/services/pdf-splitter.service.ts
import { PDFDocument } from 'pdf-lib';
import { Logger } from 'framework-nodejs-extensions';
export interface PDFPage {
  pageNumber: number;
  pdfBytes: Buffer;
}
export class PDFSplitterService {
  constructor(private logger: Logger) {}
  async splitPDF(pdfBuffer: Buffer): Promise<PDFPage[]> {
    this.logger.info('Splitting PDF into pages');
    try {
      const pdfDoc = await PDFDocument.load(pdfBuffer);
      const pageCount = pdfDoc.getPageCount();
      this.logger.info('PDF loaded', { pageCount });
      const pages: PDFPage[] = [];
      for (let i = 0; i < pageCount; i++) {
        // Create new PDF with single page
        const singlePageDoc = await PDFDocument.create();
        const [copiedPage] = await singlePageDoc.copyPages(pdfDoc, [i]);
        singlePageDoc.addPage(copiedPage);
        // Save as bytes
        const pdfBytes = await singlePageDoc.save();
        pages.push({
          pageNumber: i + 1,
          pdfBytes: Buffer.from(pdfBytes),
        });
      }
      this.logger.info('PDF split complete', { pageCount: pages.length });
      return pages;
    } catch (error) {
      this.logger.error('Failed to split PDF', { error });
      throw new Error(`PDF splitting failed: ${error.message}`);
    }
  }
}

// src/features/classifications/v1/services/image-converter.service.ts
import { fromBuffer } from 'pdf2pic';
import sharp from 'sharp';
import { Logger } from 'framework-nodejs-extensions';
export interface ConvertedImage {
  pageNumber: number;
  imageBuffer: Buffer;
  format: 'png' | 'jpeg';
  width: number;
  height: number;
}
export class ImageConverterService {
  constructor(private logger: Logger) {}
  async convertToImage(
    pdfBuffer: Buffer,
    pageNumber: number,
    options?: { format?: 'png' | 'jpeg'; quality?: number; dpi?: number }
  ): Promise<ConvertedImage> {
    const { format = 'png', quality = 90, dpi = 150 } = options || {};
    this.logger.debug('Converting PDF page to image', { pageNumber, format, dpi });
    try {
      // Convert PDF to image using pdf2pic
      const converter = fromBuffer(pdfBuffer, {
        density: dpi,
        format: format,
        quality: quality,
        width: 2048,
        height: 2048,
      });
      const result = await converter(1, { responseType: 'buffer' });
      if (!result || !result.buffer) {
        throw new Error('Image conversion produced no output');
      }
      // Optimize with sharp
      const optimized = await sharp(result.buffer)
        .resize(1024, 1024, { fit: 'inside', withoutEnlargement: true })
        .toBuffer();
      const metadata = await sharp(optimized).metadata();
      this.logger.debug('Image conversion complete', {
        pageNumber,
        width: metadata.width,
        height: metadata.height,
        size: optimized.length,
      });
      return {
        pageNumber,
        imageBuffer: optimized,
        format,
        width: metadata.width || 0,
        height: metadata.height || 0,
      };
    } catch (error) {
      this.logger.error('Failed to convert PDF page to image', { pageNumber, error });
      throw new Error(`Image conversion failed for page ${pageNumber}: ${error.message}`);
    }
  }
  async convertPages(pdfPages: Buffer[]): Promise<ConvertedImage[]> {
    const images: ConvertedImage[] = [];
    for (let i = 0; i < pdfPages.length; i++) {
      const image = await this.convertToImage(pdfPages[i], i + 1);
      images.push(image);
    }
    return images;
  }
}

// src/features/classifications/v1/services/batch-processor.service.ts
import { Logger, ConfigClient } from 'framework-nodejs-extensions';
import { LLMInvocationService } from './llm-invocation.service';
import { ConvertedImage } from './image-converter.service';
export interface BatchResult {
  batchNumber: number;
  pageNumbers: number[];
  classification: any;
  success: boolean;
  error?: string;
}
export class BatchProcessorService {
  private readonly BATCH_SIZE = 20;
  constructor(
    private logger: Logger,
    private configClient: ConfigClient,
    private llmInvocationService: LLMInvocationService
  ) {}
  async processBatches(
    images: ConvertedImage[],
    classificationPrompt: string
  ): Promise<BatchResult[]> {
    this.logger.info('Processing image batches', {
      totalImages: images.length,
      batchSize: this.BATCH_SIZE,
    });
    const batches = this.createBatches(images);
    const results: BatchResult[] = [];
    // Process batches sequentially (can be parallelized with concurrency limit)
    for (let i = 0; i < batches.length; i++) {
      const batch = batches[i];
      const batchNumber = i + 1;
      this.logger.info('Processing batch', {
        batchNumber,
        totalBatches: batches.length,
        imageCount: batch.length,
      });
      try {
        const classification = await this.llmInvocationService.classifyImages(
          batch,
          classificationPrompt
        );
        results.push({
          batchNumber,
          pageNumbers: batch.map((img) => img.pageNumber),
          classification,
          success: true,
        });
        this.logger.info('Batch processed successfully', { batchNumber });
      } catch (error) {
        this.logger.error('Batch processing failed', { batchNumber, error });
        results.push({
          batchNumber,
          pageNumbers: batch.map((img) => img.pageNumber),
          classification: null,
          success: false,
          error: error.message,
        });
        // Optionally retry failed batch
        // await this.retryBatch(batch, classificationPrompt);
      }
    }
    return results;
  }
  private createBatches(images: ConvertedImage[]): ConvertedImage[][] {
    const batches: ConvertedImage[][] = [];
    for (let i = 0; i < images.length; i += this.BATCH_SIZE) {
      batches.push(images.slice(i, i + this.BATCH_SIZE));
    }
    this.logger.info('Created batches', {
      totalImages: images.length,
      batchCount: batches.length,
    });
    return batches;
  }
}

// src/features/classifications/v1/services/llm-invocation.service.ts
import { Logger, ConfigClient } from 'framework-nodejs-extensions';
import { ServiceInvocationClient } from 'framework-nodejs-service-invocation';
import { ConvertedImage } from './image-converter.service';
export class LLMInvocationService {
  constructor(
    private logger: Logger,
    private configClient: ConfigClient,
    private httpClient: ServiceInvocationClient
  ) {}
  async classifyImages(
    images: ConvertedImage[],
    classificationPrompt: string
  ): Promise<any> {
    this.logger.info('Invoking LLM for classification', {
      imageCount: images.length,
    });
    const apiKey = await this.configClient.get<string>('OPENAI_API_KEY');
    const apiUrl = await this.configClient.get<string>('OPENAI_API_URL');
    const model = await this.configClient.get<string>('OPENAI_MODEL') || 'gpt-4-vision-preview';
    // Build LLM request
    const messages = [
      {
        role: 'system',
        content: `You are an expert in HQ Exchange financial document classification. You specialize in identifying and categorizing financial and accounting documents used in audit engagements.
Classify the provided document pages into one of these HQ Exchange document types:
**Financial Statements:**
- Trial Balance
- General Ledger
- General Ledger Detail
**Banking:**
- Bank Statements
- Bank Reconciliations
- Cash
- Check Register
**Accounts:**
- Accounts Payable
- Accounts Receivable
- Confirmations
**Asset Management:**
- Fixed Assets
- Fixed Asset Additions
- Fixed Asset Disposals
- Prepaid Expenses
- Inventory
**Liabilities:**
- Debt Agreements
- Accrued Expenses
- Accrued Payroll
- Lease Agreements
- Leases
**Investments:**
- Investments
- Investment Statements
**Operational:**
- Legal Invoices
- Board Minutes
- Revenue
- Related Party Transactions
- Personnel and Policy Manuals
**Other:**
- Other Information Not Previously Requested by CLA
Return a JSON object with a 'forms' array. Each form object should have:
- page_start: starting page number (integer)
- page_end: ending page number (integer)
- form_type: one of the document types listed above (exact match)
- confidence: confidence score between 0 and 1 (number)
For forms that appear on non-contiguous pages, omit page_start/page_end and instead use:
- page_ranges: array of objects with 'start' and 'end' properties
Example response:
{
  "forms": [
    {"page_start": 1, "page_end": 15, "form_type": "Trial Balance", "confidence": 0.96},
    {"page_start": 16, "page_end": 28, "form_type": "General Ledger Detail", "confidence": 0.94},
    {"form_type": "Other Information Not Previously Requested by CLA", "confidence": 0.55, "page_ranges": [{"start": 29, "end": 35}]}
  ]
}`,
      },
      {
        role: 'user',
        content: [
          {
            type: 'text',
            text: classificationPrompt,
          },
          ...images.map((img) => ({
            type: 'image_url',
            image_url: {
              url: `data:image/${img.format};base64,${img.imageBuffer.toString('base64')}`,
            },
          })),
        ],
      },
    ];
    const requestBody = {
      model,
      messages,
      max_tokens: 4096,
      response_format: { type: 'json_object' },
    };
    try {
      const response = await this.httpClient.post(
        `${apiUrl}/chat/completions`,
        requestBody,
        {
          headers: {
            'Content-Type': 'application/json',
            Authorization: `Bearer ${apiKey}`,
          },
          timeout: 60000, // 60 seconds
        }
      );
      const classification = JSON.parse(response.data.choices[0].message.content);
      this.logger.info('LLM classification success', {
        imageCount: images.length,
      });
      return classification;
    } catch (error) {
      this.logger.error('LLM invocation failed', { error });
      throw new Error(`LLM classification failed: ${error.message}`);
    }
  }
}

// src/features/classifications/v1/services/result-collator.service.ts
import { Logger } from 'framework-nodejs-extensions';
import { BatchResult } from './batch-processor.service';
export interface FormClassification {
  page_start?: number;
  page_end?: number;
  form_type: string;
  confidence: number;
  page_ranges?: Array<{ start: number; end: number }>;
}
export interface CollatedResult {
  is_multi_form: boolean;
  num_pages: number;
  forms: FormClassification[];
  classified_in_chunks: boolean;
  num_chunks: number;
}
export class ResultCollatorService {
  constructor(private logger: Logger) {}
  async collateResults(batchResults: BatchResult[]): Promise<CollatedResult> {
    this.logger.info('Collating batch results', {
      batchCount: batchResults.length,
    });
    const successfulBatches = batchResults.filter((b) => b.success);
    // Extract all classifications
    const classifications = successfulBatches.map((b) => b.classification);
    // Merge and normalize forms from all batches
    const allForms = this.mergeForms(classifications);
    // Determine if document contains multiple forms
    const isMultiForm = allForms.length > 1;
    // Count total pages
    const totalPages = this.countTotalPages(batchResults);
    const collatedResult: CollatedResult = {
      is_multi_form: isMultiForm,
      num_pages: totalPages,
      forms: allForms,
      classified_in_chunks: true,
      num_chunks: batchResults.length,
    };
    this.logger.info('Result collation complete', {
      isMultiForm,
      totalPages,
      formsCount: allForms.length,
      numChunks: batchResults.length,
    });
    return collatedResult;
  }
  private mergeForms(classifications: any[]): FormClassification[] {
    const allForms: FormClassification[] = [];
    for (const classification of classifications) {
      if (classification.forms && Array.isArray(classification.forms)) {
        for (const form of classification.forms) {
          allForms.push({
            page_start: form.page_start,
            page_end: form.page_end,
            form_type: form.form_type,
            confidence: form.confidence,
            page_ranges: form.page_ranges,
          });
        }
      }
    }
    // Sort forms by page_start (or first range start)
    allForms.sort((a, b) => {
      const aStart = a.page_start || (a.page_ranges?.[0]?.start ?? Infinity);
      const bStart = b.page_start || (b.page_ranges?.[0]?.start ?? Infinity);
      return aStart - bStart;
    });
    // Optionally merge contiguous forms of the same type
    return this.mergeContiguousForms(allForms);
  }
  private mergeContiguousForms(forms: FormClassification[]): FormClassification[] {
    if (forms.length === 0) return [];
    const merged: FormClassification[] = [];
    let current = { ...forms[0] };
    for (let i = 1; i < forms.length; i++) {
      const next = forms[i];
      // Only merge if same form type and contiguous pages
      if (
        current.form_type === next.form_type &&
        current.page_end !== undefined &&
        next.page_start !== undefined &&
        current.page_end + 1 === next.page_start
      ) {
        // Merge into current
        current.page_end = next.page_end;
        // Average confidence
        current.confidence = (current.confidence + next.confidence) / 2;
      } else {
        // Push current and start new
        merged.push(current);
        current = { ...next };
      }
    }
    // Don't forget the last one
    merged.push(current);
    return merged;
  }
  private countTotalPages(batchResults: BatchResult[]): number {
    const allPages = batchResults.flatMap((b) => b.pageNumbers);
    return Math.max(...allPages, 0);
  }
}

// src/features/classifications/v1/workers/classification-worker.ts
import { Logger } from 'framework-nodejs-extensions';
import { ServiceBusClient } from 'framework-nodejs-pubsub';
import { BlobStorageClient } from 'framework-nodejs-storage';
import { JobRepository } from '../repositories/job.repository';
import { ClassificationResultRepository } from '../repositories/classification-result.repository';
import { PDFSplitterService } from '../services/pdf-splitter.service';
import { ImageConverterService } from '../services/image-converter.service';
import { BatchProcessorService } from '../services/batch-processor.service';
import { ResultCollatorService } from '../services/result-collator.service';
import { JobStatus } from '../models/job.model';
export class ClassificationWorker {
  constructor(
    private logger: Logger,
    private serviceBusClient: ServiceBusClient,
    private blobStorageClient: BlobStorageClient,
    private jobRepository: JobRepository,
    private classificationResultRepository: ClassificationResultRepository,
    private pdfSplitterService: PDFSplitterService,
    private imageConverterService: ImageConverterService,
    private batchProcessorService: BatchProcessorService,
    private resultCollatorService: ResultCollatorService
  ) {}
  async start(): Promise<void> {
    this.logger.info('Starting classification worker');
    // Subscribe to Service Bus queue
    await this.serviceBusClient.subscribeToQueue(
      'hqx-classification-jobs',
      async (message) => {
        await this.processJob(message);
      }
    );
    this.logger.info('Worker subscribed to hqx-classification-jobs queue');
  }
  private async processJob(message: any): Promise<void> {
    const { jobId, pdfBlobPath, classificationPrompt } = message;
    this.logger.info('Processing classification job', { jobId });
    try {
      // 1. Download PDF from blob storage
      const pdfBuffer = await this.blobStorageClient.download(
        'hqx-classification-storage',
        pdfBlobPath
      );
      this.logger.info('PDF downloaded from storage', { jobId });
      // 2. Split PDF into pages
      const pages = await this.pdfSplitterService.splitPDF(pdfBuffer);
      // Update job with total pages
      await this.jobRepository.updateProgress(jobId, {
        totalPages: pages.length,
        processedPages: 0,
      });
      this.logger.info('PDF split into pages', { jobId, pageCount: pages.length });
      // 3. Convert pages to images
      const images = await this.imageConverterService.convertPages(
        pages.map((p) => p.pdfBytes)
      );
      this.logger.info('Pages converted to images', { jobId, imageCount: images.length });
      // 4. Process images in batches
      const batchResults = await this.batchProcessorService.processBatches(
        images,
        classificationPrompt
      );
      this.logger.info('Batch processing complete', {
        jobId,
        batchCount: batchResults.length,
      });
      // 5. Collate results
      const collatedResult = await this.resultCollatorService.collateResults(batchResults);
      this.logger.info('Results collated', { jobId });
      // 6. Save result to database
      await this.classificationResultRepository.create({
        jobId,
        classificationData: collatedResult,
        createdAt: new Date(),
      });
      // 7. Upload result to blob storage
      const resultBlobPath = `classification-results/${jobId}/result.json`;
      await this.blobStorageClient.upload(
        'hqx-classification-storage',
        resultBlobPath,
        Buffer.from(JSON.stringify(collatedResult, null, 2))
      );
      this.logger.info('Result saved to storage', { jobId, resultBlobPath });
      // 8. Update job status to success
      await this.jobRepository.updateStatus(jobId, JobStatus.Success, {
        completedAt: new Date(),
        processedPages: collatedResult.processedPages,
        resultBlobPath,
      });
      this.logger.info('Classification job completed successfully', { jobId });
    } catch (error) {
      this.logger.error('Classification job failed', { jobId, error });
      // Update job status to failed
      await this.jobRepository.updateStatus(jobId, JobStatus.Failed, {
        failedAt: new Date(),
        errorCode: 'PROCESSING_ERROR',
        errorMessage: error.message,
        errorDetails: error.stack,
      });
    }
  }
  async stop(): Promise<void> {
    this.logger.info('Stopping classification worker');
    await this.serviceBusClient.close();
  }
}