⚙️ Distributed Worker System¶

// Job Queue Configuration
const Queue = require('bull');
const redis = require('redis');

const renderQueue = new Queue('motion graphics rendering', {
  redis: {
    host: process.env.REDIS_HOST,
    port: process.env.REDIS_PORT,
    password: process.env.REDIS_PASSWORD
  },
  defaultJobOptions: {
    removeOnComplete: 100,
    removeOnFail: 50,
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 2000
    }
  }
});

// Job Types
const JOB_TYPES = {
  MOTION_GRAPHICS_RENDER: 'motion_graphics_render',
  ASSET_PROCESSING: 'asset_processing',
  QUALITY_VALIDATION: 'quality_validation',
  OUTPUT_CONVERSION: 'output_conversion'
};

// Job Priority Levels
const PRIORITY_LEVELS = {
  CRITICAL: 1,    // Real-time requests
  HIGH: 2,        // Premium users
  NORMAL: 3,      // Standard users
  LOW: 4          // Batch processing
};

class JobScheduler {
  constructor() {
    this.queue = renderQueue;
    this.workerRegistry = new Map();
    this.resourceMonitor = new ResourceMonitor();
  }

  async scheduleJob(jobData, priority = PRIORITY_LEVELS.NORMAL) {
    const job = await this.queue.add(jobData.type, jobData, {
      priority: priority,
      delay: this.calculateDelay(jobData),
      jobId: this.generateJobId(jobData)
    });

    return job;
  }

  calculateDelay(jobData) {
    // Calculate delay based on resource availability
    const currentLoad = this.resourceMonitor.getCurrentLoad();
    const estimatedDuration = this.estimateJobDuration(jobData);

    if (currentLoad > 0.8) {
      return estimatedDuration * 0.5; // Delay for resource availability
    }

    return 0; // Immediate execution
  }

  async distributeJob(job) {
    const availableWorkers = await this.getAvailableWorkers();
    const bestWorker = this.selectBestWorker(availableWorkers, job);

    if (bestWorker) {
      await this.assignJobToWorker(job, bestWorker);
    } else {
      await this.queueJobForLater(job);
    }
  }
}

// Worker Implementation
class MotionGraphicsWorker {
  constructor(workerId, config) {
    this.workerId = workerId;
    this.config = config;
    this.status = 'idle';
    this.currentJob = null;
    this.resourceUsage = new ResourceMonitor();
  }

  async start() {
    console.log(`Worker ${this.workerId} starting...`);

    // Register with job scheduler
    await this.registerWithScheduler();

    // Start job processing
    this.processJobs();

    // Start health monitoring
    this.startHealthMonitoring();
  }

  async processJobs() {
    const processor = async (job) => {
      try {
        this.status = 'processing';
        this.currentJob = job;

        console.log(`Worker ${this.workerId} processing job ${job.id}`);

        // Process based on job type
        const result = await this.processJobByType(job.data);

        this.status = 'idle';
        this.currentJob = null;

        return result;
      } catch (error) {
        console.error(`Worker ${this.workerId} job failed:`, error);
        this.status = 'error';
        throw error;
      }
    };

    renderQueue.process(processor);
  }

  async processJobByType(jobData) {
    switch (jobData.type) {
      case JOB_TYPES.MOTION_GRAPHICS_RENDER:
        return await this.renderMotionGraphics(jobData);
      case JOB_TYPES.ASSET_PROCESSING:
        return await this.processAssets(jobData);
      case JOB_TYPES.QUALITY_VALIDATION:
        return await this.validateQuality(jobData);
      default:
        throw new Error(`Unknown job type: ${jobData.type}`);
    }
  }

  async renderMotionGraphics(jobData) {
    const { timeline, outputSpecs } = jobData;

    // Create Natron project
    const natronProject = await this.createNatronProject(timeline);

    // Render via Natron CLI
    const outputPath = await this.renderWithNatron(natronProject, outputSpecs);

    // Post-process output
    const finalOutput = await this.postProcessOutput(outputPath, outputSpecs);

    return {
      success: true,
      outputPath: finalOutput,
      metadata: {
        duration: outputSpecs.duration,
        resolution: outputSpecs.resolution,
        format: outputSpecs.format
      }
    };
  }
}

// Health Check System
class HealthCheckSystem {
  constructor() {
    this.checks = new Map();
    this.setupDefaultChecks();
  }

  setupDefaultChecks() {
    // System resource check
    this.addCheck('system_resources', async () => {
      const cpuUsage = await this.getCpuUsage();
      const memoryUsage = await this.getMemoryUsage();
      const diskUsage = await this.getDiskUsage();

      return {
        status: cpuUsage < 90 && memoryUsage < 90 && diskUsage < 90 ? 'healthy' : 'unhealthy',
        details: { cpuUsage, memoryUsage, diskUsage }
      };
    });

    // Natron availability check
    this.addCheck('natron_availability', async () => {
      try {
        const result = await this.checkNatronInstallation();
        return {
          status: result ? 'healthy' : 'unhealthy',
          details: { natronVersion: result }
        };
      } catch (error) {
        return {
          status: 'unhealthy',
          details: { error: error.message }
        };
      }
    });

    // Redis connection check
    this.addCheck('redis_connection', async () => {
      try {
        await redis.ping();
        return { status: 'healthy', details: {} };
      } catch (error) {
        return {
          status: 'unhealthy',
          details: { error: error.message }
        };
      }
    });
  }

  async runAllChecks() {
    const results = {};

    for (const [name, check] of this.checks) {
      try {
        results[name] = await check();
      } catch (error) {
        results[name] = {
          status: 'unhealthy',
          details: { error: error.message }
        };
      }
    }

    return results;
  }
}

// Metrics Collection
class MetricsCollector {
  constructor() {
    this.metrics = new Map();
    this.setupPrometheusMetrics();
  }

  setupPrometheusMetrics() {
    const prometheus = require('prom-client');

    // Job metrics
    this.jobCounter = new prometheus.Counter({
      name: 'jobs_processed_total',
      help: 'Total number of jobs processed',
      labelNames: ['worker_id', 'job_type', 'status']
    });

    this.jobDuration = new prometheus.Histogram({
      name: 'job_duration_seconds',
      help: 'Duration of job processing',
      labelNames: ['worker_id', 'job_type'],
      buckets: [1, 5, 15, 30, 60, 120, 300, 600]
    });

    // Resource metrics
    this.cpuUsage = new prometheus.Gauge({
      name: 'worker_cpu_usage_percent',
      help: 'CPU usage percentage',
      labelNames: ['worker_id']
    });

    this.memoryUsage = new prometheus.Gauge({
      name: 'worker_memory_usage_bytes',
      help: 'Memory usage in bytes',
      labelNames: ['worker_id']
    });
  }

  recordJobCompletion(workerId, jobType, duration, success) {
    this.jobCounter.inc({
      worker_id: workerId,
      job_type: jobType,
      status: success ? 'success' : 'failure'
    });

    this.jobDuration.observe(
      { worker_id: workerId, job_type: jobType },
      duration
    );
  }

  updateResourceMetrics(workerId, cpuUsage, memoryUsage) {
    this.cpuUsage.set({ worker_id: workerId }, cpuUsage);
    this.memoryUsage.set({ worker_id: workerId }, memoryUsage);
  }
}

// Custom Scaling Logic
class CustomScaler {
  constructor() {
    this.queueMonitor = new QueueMonitor();
    this.resourceMonitor = new ResourceMonitor();
  }

  async shouldScaleUp() {
    const queueLength = await this.queueMonitor.getQueueLength();
    const averageJobDuration = await this.queueMonitor.getAverageJobDuration();
    const currentWorkers = await this.getCurrentWorkerCount();

    // Scale up if queue is getting long and jobs are taking too long
    const estimatedWaitTime = (queueLength * averageJobDuration) / currentWorkers;

    return estimatedWaitTime > 300; // 5 minutes
  }

  async shouldScaleDown() {
    const queueLength = await this.queueMonitor.getQueueLength();
    const currentWorkers = await this.getCurrentWorkerCount();
    const averageUtilization = await this.resourceMonitor.getAverageUtilization();

    // Scale down if queue is short and utilization is low
    return queueLength < 5 && averageUtilization < 30 && currentWorkers > 2;
  }
}