watch-finished-turbo/apps/service/src/task-queue.service.ts

import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
import * as fs from 'fs';
import { ConfigService } from './config.service';
import { DbService } from './db.service';
import { EventsGateway } from './events.gateway';
import { HandbrakeService } from './handbrake.service';

interface Task {
  id: number;
  type: string;
  status: string;
  progress: number;
  dataset?: string;
  input?: string;
  output?: string;
  preset?: string;
  priority: number;
  retry_count?: number;
  max_retries?: number;
  error_message?: string;
  created_at: string;
  updated_at: string;
}

interface QueueSettings {
  batchSize: number;
  concurrency: number;
  retryEnabled: boolean;
  maxRetries: number;
  retryDelay: number; // in milliseconds
  processingInterval: number; // in milliseconds
}

export type { QueueSettings };

@Injectable()
export class TaskQueueService implements OnModuleInit {
  private logger = new Logger('TaskQueueService');
  private isProcessing = false;
  private processingInterval: NodeJS.Timeout | null = null;
  private activeTasks = new Set<number>();
  private isStopping = false;
  private gracefulStopResolver: (() => void) | null = null;
  private queueSettings: QueueSettings;

  constructor(
    private readonly db: DbService,
    private readonly handbrake: HandbrakeService,
    private readonly eventsGateway: EventsGateway,
    private readonly config: ConfigService,
  ) {
    this.loadQueueSettings();
  }

  private loadQueueSettings() {
    const settings = this.config.getSettings('queue', {});
    this.queueSettings = {
      batchSize: settings.batchSize || 10,
      concurrency: settings.concurrency || 1,
      retryEnabled: settings.retryEnabled !== false, // default true
      maxRetries: settings.maxRetries || 3,
      retryDelay: settings.retryDelay || 30000, // 30 seconds default
      processingInterval: settings.processingInterval || 5000, // 5 seconds default
    };
  }

  updateQueueSettings(settings: Partial<QueueSettings>) {
    this.queueSettings = { ...this.queueSettings, ...settings };
    // Save to config
    const currentSettings = this.config.getSettings('queue', {});
    this.config.setSettings({
      queue: { ...currentSettings, ...settings },
    });

    // Restart processing with new interval if changed
    if (settings.processingInterval && this.processingInterval) {
      this.stopProcessing();
      this.startProcessing();
    }
  }

  getQueueSettings(): QueueSettings {
    return { ...this.queueSettings };
  }

  onModuleInit() {
    // Don't start processing automatically - wait for explicit start command
  }

  startProcessing() {
    // Reset any pending stop request
    this.isStopping = false;
    this.gracefulStopResolver = null;

    if (this.processingInterval) {
      return;
    }

    this.processingInterval = setInterval(() => {
      this.processPendingTasks();
    }, this.queueSettings.processingInterval);
  }

  stopProcessing() {
    if (this.processingInterval) {
      clearInterval(this.processingInterval);
      this.processingInterval = null;
    }
  }

  private stopImmediate() {
    this.isStopping = false;
    this.gracefulStopResolver = null;
    this.stopProcessing();
    return { stopped: true, activeTasks: this.activeTasks.size };
  }

  private resolveGracefulStopIfDrained() {
    if (this.isStopping && this.activeTasks.size === 0 && !this.isProcessing) {
      const resolver = this.gracefulStopResolver;
      this.isStopping = false;
      this.gracefulStopResolver = null;
      if (resolver) {
        resolver();
      }
    }
  }

  private async stopGracefully() {
    this.isStopping = true;
    this.stopProcessing();
    this.logger.log(
      `Graceful stop requested; waiting for ${this.activeTasks.size} active task(s)`,
    );

    if (this.activeTasks.size === 0 && !this.isProcessing) {
      this.isStopping = false;
      return { stopped: true, drained: true, activeTasks: 0 };
    }

    return new Promise((resolve) => {
      this.gracefulStopResolver = () => {
        this.logger.log('All active tasks completed; queue stopped.');
        resolve({ stopped: true, drained: true, activeTasks: 0 });
      };
    });
  }

  async stop(graceful = true) {
    if (!graceful) {
      return this.stopImmediate();
    }

    return this.stopGracefully();
  }

  // Public API methods
  start() {
    this.startProcessing();
    return { started: true };
  }

  private async processPendingTasks() {
    if (this.isProcessing) {
      return; // Already processing
    }

    if (this.isStopping) {
      this.resolveGracefulStopIfDrained();
      return;
    }

    try {
      this.isProcessing = true;

      // Check for tasks that need retry
      await this.processRetryTasks();

      // Get pending tasks up to batch size
      const pendingTasks = this.db.getPendingTasks(
        this.queueSettings.batchSize,
      ) as Task[];

      if (pendingTasks.length === 0) {
        return; // No tasks to process
      }

      // Process tasks up to batch size, respecting concurrency limit
      const processingPromises: Promise<void>[] = [];

      for (const task of pendingTasks) {
        if (this.activeTasks.size >= this.queueSettings.concurrency) {
          break; // Respect concurrency limit
        }

        if (!task.input || !task.output || !task.preset) {
          this.logger.error(`Task ${task.id} is missing required fields`);
          this.db.updateTask(task.id, {
            status: 'failed',
            error_message: 'Missing required fields: input, output, or preset',
          });
          continue;
        }

        // Mark task as processing
        this.db.updateTask(task.id, { status: 'processing' });
        this.activeTasks.add(task.id);

        // Emit task update
        this.eventsGateway.emitTaskUpdate({
          type: 'started',
          taskId: task.id,
          task: 'handbrake',
          input: task.input,
          output: task.output,
          preset: task.preset,
        });

        // Process task asynchronously
        const processPromise = this.processTask(task);
        processingPromises.push(processPromise);
      }

      // Wait for all concurrent tasks to complete
      await Promise.allSettled(processingPromises);
    } catch (error) {
      this.logger.error(`Error in processPendingTasks: ${error.message}`);
    } finally {
      this.isProcessing = false;
      this.resolveGracefulStopIfDrained();
    }
  }

  private async processRetryTasks() {
    if (!this.queueSettings.retryEnabled) {
      return;
    }

    try {
      // Get failed tasks that haven't exceeded max retries
      const failedTasks = (this.db.getAllTasks() as Task[]).filter(
        (task) =>
          task.status === 'failed' &&
          (task.retry_count || 0) < this.queueSettings.maxRetries,
      );

      for (const task of failedTasks) {
        const retryCount = (task.retry_count || 0) + 1;
        const lastUpdate = new Date(task.updated_at);
        const timeSinceFailure = Date.now() - lastUpdate.getTime();

        // Check if enough time has passed for retry
        if (timeSinceFailure >= this.queueSettings.retryDelay) {
          // Reset task for retry
          this.db.updateTask(task.id, {
            status: 'pending',
            progress: 0,
            retry_count: retryCount,
            error_message: undefined,
          });

          // Emit retry event
          this.eventsGateway.emitTaskUpdate({
            type: 'retry',
            taskId: task.id,
            task: 'handbrake',
            retryCount,
          });
        }
      }
    } catch (error) {
      this.logger.error(`Error in processRetryTasks: ${error.message}`);
    }
  }

  private async processTask(task: Task): Promise<void> {
    try {
      // Check if output file already exists and skip if it does (unless requeued)
      if (task.output && fs.existsSync(task.output)) {
        // Check if this task was requeued (has retry_count > 0 or was manually requeued)
        const wasRequeued =
          (task.retry_count || 0) > 0 || (task.priority || 0) > 0;

        if (!wasRequeued) {
          // Skip processing - file already exists
          this.db.updateTask(task.id, { status: 'skipped', progress: 100 });

          // Update file status if it exists
          if (task.dataset) {
            this.db.setFile(task.dataset, task.input!, {
              status: 'success',
              date: new Date().toISOString(),
            });
          }

          // Emit skipped event
          this.eventsGateway.emitTaskUpdate({
            type: 'skipped',
            taskId: task.id,
            task: 'handbrake',
            input: task.input,
            output: task.output,
            preset: task.preset,
            success: true,
          });
          return;
        }
      }

      // Process the file
      const success = await this.handbrake.processWithHandbrake(
        task.input!,
        task.output!,
        task.preset!,
        task.id,
      );

      if (success) {
        // Update task status
        this.db.updateTask(task.id, { status: 'completed', progress: 100 });

        // Update file status if it exists
        if (task.dataset) {
          this.db.setFile(task.dataset, task.input!, {
            status: 'success',
            date: new Date().toISOString(),
          });
        }

        // Emit completion event
        this.eventsGateway.emitTaskUpdate({
          type: 'completed',
          taskId: task.id,
          task: 'handbrake',
          input: task.input,
          output: task.output,
          preset: task.preset,
          success: true,
        });
      } else {
        throw new Error('Handbrake processing failed');
      }
    } catch (error) {
      const retryCount = task.retry_count || 0;

      if (
        this.queueSettings.retryEnabled &&
        retryCount < this.queueSettings.maxRetries
      ) {
        // Mark for retry
        this.db.updateTask(task.id, {
          status: 'failed',
          error_message: error.message,
          retry_count: retryCount + 1,
        });

        this.logger.warn(
          `Task ${task.id} failed, will retry (attempt ${retryCount + 1}): ${error.message}`,
        );
      } else {
        // Final failure
        this.db.updateTask(task.id, {
          status: 'failed',
          error_message: error.message,
        });

        // Update file status if it exists
        if (task.dataset) {
          this.db.setFile(task.dataset, task.input!, {
            status: 'error',
            date: new Date().toISOString(),
          });
        }

        this.logger.error(
          `Task ${task.id} failed permanently: ${error.message}`,
        );
      }

      // Emit failure event
      this.eventsGateway.emitTaskUpdate({
        type: 'failed',
        taskId: task.id,
        task: 'handbrake',
        input: task.input,
        output: task.output,
        preset: task.preset,
        success: false,
        error: error.message,
        retryCount: retryCount + 1,
        maxRetries: this.queueSettings.maxRetries,
      });
    } finally {
      // Remove from active tasks
      this.activeTasks.delete(task.id);
      this.resolveGracefulStopIfDrained();
    }
  }

  // Manual task creation (for requeueing from web interface)
  createTask(taskData: {
    dataset: string;
    input: string;
    output: string;
    preset: string;
    priority?: number;
    status?: string;
  }) {
    // Check if output file already exists (case-insensitive matching)
    if (this.handbrake.outputFileExists(taskData.output)) {
      this.logger.warn(
        `Output file already exists: ${taskData.output}, skipping task creation`,
      );
      throw new Error(`Output file already exists: ${taskData.output}`);
    }

    // Check if file already exists in database
    const existingFile = this.db.findFile(taskData.dataset, taskData.input);
    if (!existingFile) {
      // Create file record
      this.db.setFile(taskData.dataset, taskData.input, {
        output: taskData.output,
        status: 'pending',
        date: new Date().toISOString(),
      });
    }

    // Create task
    const task = this.db.createTask({
      type: 'handbrake',
      status: taskData.status || 'pending',
      dataset: taskData.dataset,
      input: taskData.input,
      output: taskData.output,
      preset: taskData.preset,
      priority: taskData.priority || 0,
    });

    this.logger.log(`Created task ${task.id} for file: ${taskData.input}`);
    return task;
  }

  // Get queue status
  getQueueStatus() {
    const allTasks = this.db.getAllTasks() as Task[];
    const pending = allTasks.filter((t) => t.status === 'pending').length;
    const processing = allTasks.filter((t) => t.status === 'processing').length;
    const completed = allTasks.filter((t) => t.status === 'completed').length;
    const failed = allTasks.filter((t) => t.status === 'failed').length;

    return {
      isProcessing: !!this.processingInterval, // Whether task processing is enabled/running
      isProcessingCycle: this.isProcessing, // Whether currently in a processing cycle
      isStopping: this.isStopping, // Whether a graceful stop has been requested
      activeTasks: this.activeTasks.size,
      pending,
      processing,
      completed,
      failed,
      total: allTasks.length,
      settings: this.getQueueSettings(),
    };
  }

  // Get task by input file
  getTaskByInput(input: string): Task | undefined {
    return this.db.getTaskByInput(input) as Task | undefined;
  }

  // Update task status
  updateTaskStatus(taskId: number, status: string, errorMessage?: string) {
    const updateData: any = { status };
    if (errorMessage) {
      updateData.error_message = errorMessage;
    }
    return this.db.updateTask(taskId, updateData);
  }
}