stack/apps/api/src/knowledge/queues/embedding.processor.ts

import { Processor, WorkerHost } from "@nestjs/bullmq";
import { Logger } from "@nestjs/common";
import { Job } from "bullmq";
import { OllamaEmbeddingService } from "../services/ollama-embedding.service";
import { EmbeddingJobData } from "./embedding-queue.service";

/**
 * Processor for embedding generation jobs
 *
 * This worker processes queued embedding jobs and generates
 * embeddings for knowledge entries using Ollama.
 */
@Processor("embeddings")
export class EmbeddingProcessor extends WorkerHost {
  private readonly logger = new Logger(EmbeddingProcessor.name);

  constructor(private readonly embeddingService: OllamaEmbeddingService) {
    super();
  }

  /**
   * Process an embedding generation job
   *
   * @param job - The embedding job to process
   */
  async process(job: Job<EmbeddingJobData>): Promise<void> {
    const { entryId, content, model } = job.data;

    this.logger.log(`Processing embedding job ${job.id ?? "unknown"} for entry ${entryId}`);

    try {
      const options: { model?: string } = {};
      if (model !== undefined) {
        options.model = model;
      }

      await this.embeddingService.generateAndStoreEmbedding(entryId, content, options);

      this.logger.log(
        `Successfully generated embedding for entry ${entryId} (job: ${job.id ?? "unknown"})`
      );
    } catch (error) {
      this.logger.error(
        `Failed to generate embedding for entry ${entryId} (job: ${job.id ?? "unknown"})`,
        error
      );
      throw error; // Re-throw to trigger retry logic
    }
  }

  /**
   * Handle successful job completion
   *
   * @param job - The completed job
   */
  onCompleted(job: Job<EmbeddingJobData>): void {
    this.logger.log(
      `Successfully generated embedding for entry ${job.data.entryId} (job: ${job.id ?? "unknown"})`
    );
  }

  /**
   * Handle job failure
   *
   * @param job - The failed job
   * @param error - The error that caused the failure
   */
  onFailed(job: Job<EmbeddingJobData>, error: Error): void {
    this.logger.error(
      `Failed to generate embedding for entry ${job.data.entryId} (job: ${job.id ?? "unknown"}) after ${job.attemptsMade.toString()} attempts`,
      error
    );
  }

  /**
   * Alias for process to match test expectations
   */
  async processEmbedding(job: Job<EmbeddingJobData>): Promise<void> {
    return this.process(job);
  }

  /**
   * Alias for onCompleted to match test expectations
   */
  handleCompleted(job: Job<EmbeddingJobData>): void {
    this.onCompleted(job);
  }

  /**
   * Alias for onFailed to match test expectations
   */
  handleFailed(job: Job<EmbeddingJobData>, error: Error): void {
    this.onFailed(job, error);
  }
}