// src/business/queue/business-provisioning.processor.ts import { EntityManager } from "@mikro-orm/postgresql"; import { Processor } from "@nestjs/bullmq"; import { Logger } from "@nestjs/common"; import { Job } from "bullmq"; import { WorkerProcessor } from "../../../common/queues/worker.processor"; import { SUBSCRIPTIONS } from "../constant"; import { Business } from "../entities/business.entity"; import { IBusinessProvisioningJob } from "../interfaces/IBusiness"; @Processor(SUBSCRIPTIONS.PROVISIONING_QUEUE_NAME) export class BusinessProvisioningProcessor extends WorkerProcessor { protected readonly logger = new Logger(BusinessProvisioningProcessor.name); constructor(private readonly em: EntityManager) { super(); } async process(job: Job) { switch (job.name) { case SUBSCRIPTIONS.PROVISIONING_JOB_NAME: await this.handleBusinessCreated(job); break; default: this.logger.warn(`Unknown job name: ${job.name}`); } } private async handleBusinessCreated(job: Job) { const em = this.em.fork(); const { subscriptionId, serviceId, serviceName, businessName } = job.data; // Only act if the event is for this service if (serviceId !== SUBSCRIPTIONS.SERVICE_ID) { this.logger.debug(`Skipping job for service: ${serviceName}`); return; } // Check if already exists let business = await em.findOne(Business, { danakSubscriptionId: subscriptionId }); if (!business) { business = em.create(Business, { danakSubscriptionId: subscriptionId, name: businessName, // logoUrl: job.data.logoUrl, // Uncomment if logoUrl is provided in job data }); await em.persistAndFlush(business); this.logger.log(`Created business for subscription ${subscriptionId}`); } else { this.logger.debug(`Business already exists for subscription ${subscriptionId}`); } } }