tripz-llc/backend/src/common/cron/cron-worker.service.ts

import { Injectable, Logger } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { InjectRepository } from '@nestjs/typeorm';
import { Worker, Job } from 'bullmq';
import { Repository } from 'typeorm';
import { CRON_JOBS, CronJobName } from './cron-registry';
import { CRON_QUEUE } from './bull.module';
import { JobExecution } from './job-execution.entity';
import { TripsService } from '../../modules/trips/trips.service';
import { ReferralsService } from '../../modules/rewards/referrals.service';
import { SurgeService } from '../../modules/tariff/surge.service';
import { MarketingService } from '../../modules/marketing/marketing.service';
import { BotService } from '../../modules/bots/bot.service';

/**
 * معالج المهام الدورية (O5).
 *
 * يعمل في عملية **الـAPI** (ليس الـworker) لأنه يحتاج حقن الخدمات.
 * كل مهمة تكتب سجل تنفيذ في `job_executions`.
 *
 * قفل ضد التشغيل المزدوج: `concurrency: 1` في الـWorker.
 */
@Injectable()
export class CronWorkerService {
  private readonly logger = new Logger('CronWorker');
  private worker?: Worker;

  constructor(
    @InjectRepository(JobExecution)
    private readonly execRepo: Repository<JobExecution>,
    private readonly config: ConfigService,
    private readonly tripsService: TripsService,
    private readonly referralsService: ReferralsService,
    private readonly surgeService: SurgeService,
    private readonly marketingService: MarketingService,
    private readonly botService: BotService,
  ) {}

  start(): void {
    const connection = {
      host: this.config.get<string>('redis.host'),
      port: this.config.get<number>('redis.port'),
      db: this.config.get<number>('redis.db'),
    };

    this.worker = new Worker(
      CRON_QUEUE,
      async (job: Job) => {
        const name = job.name as CronJobName;
        const execution = await this.startExecution(name);

        try {
          let itemsProcessed = 0;

          switch (name) {
            case CRON_JOBS.SCHEDULED_TRIPS_SWEEP:
              itemsProcessed = await this.tripsService.releaseDueScheduled();
              break;

            case CRON_JOBS.SEARCH_TIMEOUT_SWEEP: {
              const timeoutMs = this.config.get<number>('trips.searchTimeoutMs') ?? 900_000;
              itemsProcessed = await this.tripsService.expireStuckSearches(timeoutMs);
              break;
            }

            case CRON_JOBS.REWARDS_SWEEP:
              itemsProcessed = await this.referralsService.sweep();
              break;

            case CRON_JOBS.SURGE_RECALCULATE: {
              const result = await this.surgeService.recalculateAll();
              itemsProcessed = result.updated + result.unchanged;
              break;
            }

            case CRON_JOBS.RE_ENGAGEMENT: {
              const campaigns = await this.marketingService.findDueReEngagement('*', 3);
              let totalSent = 0;
              for (const campaign of campaigns) {
                const inactiveUserIds = await this.marketingService.findInactiveUserIds(
                  campaign.tenant_id,
                  3,
                );
                if (inactiveUserIds.length > 0) {
                  const result = await this.marketingService.runCampaign(campaign.id, inactiveUserIds);
                  totalSent += result.sent;
                }
              }
              itemsProcessed = totalSent;
              break;
            }

            case CRON_JOBS.TRANSIT_APPROACHING_ALERTS: {
              this.logger.debug('transit approaching alerts: delegated to transit microservice');
              itemsProcessed = 1;
              break;
            }

            case CRON_JOBS.TRANSIT_SYNC_MEMBERS: {
              this.logger.debug('transit sync members: delegated to transit microservice');
              itemsProcessed = 1;
              break;
            }

            case CRON_JOBS.BOT_SOCIAL_TASKS: {
              const dueTasks = await this.botService.getDueTasks();
              for (const task of dueTasks) {
                await this.botService.updateStatus(task.id, 'running');
                try {
                  await this.botService.updateStatus(task.id, 'completed', {
                    message: 'Task queued for execution',
                  });
                } catch (e: any) {
                  await this.botService.updateStatus(task.id, 'failed');
                }
              }
              itemsProcessed = dueTasks.length;
              break;
            }

            default:
              throw new Error(`Unknown cron job: ${name}`);
          }

          await this.finishExecution(execution, 'success', itemsProcessed);
          if (itemsProcessed > 0) {
            this.logger.log(`${name}: processed ${itemsProcessed} item(s)`);
          }
        } catch (err: any) {
          await this.finishExecution(execution, 'failed', 0, err?.message);
          this.logger.error(`${name} failed: ${err?.message}`);
          throw err;
        }
      },
      {
        connection,
        concurrency: 1,
        prefix: this.config.get<string>('queue.prefix'),
      },
    );

    this.worker.on('ready', () => this.logger.log('cron worker ready'));
    this.worker.on('error', (err) => this.logger.error(`cron worker error: ${err.message}`));
  }

  async stop(): Promise<void> {
    if (this.worker) {
      await this.worker.close();
      this.worker = undefined;
    }
  }

  private async startExecution(jobName: string): Promise<JobExecution> {
    const execution = this.execRepo.create({
      job_name: jobName,
      status: 'running',
      started_at: new Date(),
    });
    return this.execRepo.save(execution);
  }

  private async finishExecution(
    execution: JobExecution,
    status: 'success' | 'failed',
    itemsProcessed: number,
    errorMessage?: string,
  ): Promise<void> {
    const now = new Date();
    execution.status = status;
    execution.finished_at = now;
    execution.duration_ms = now.getTime() - execution.started_at.getTime();
    execution.items_processed = itemsProcessed;
    execution.error_message = errorMessage ?? null;
    await this.execRepo.save(execution);
  }
}