campaign

src/modules/notifications/events/campaign-sms-dispatch.events.ts

@@ -0,0 +1,7 @@
+export class CampaignSmsDispatchResultEvent {
+  constructor(
+    public readonly campaignId: string,
+    public readonly success: boolean,
+    public readonly quantity: number = 1,
+  ) {}
+}

src/modules/notifications/interfaces/jobs-queue.interface.ts

@@ -12,6 +12,7 @@ export interface DirectSmsQueueJob {
   templateId: string;
   params?: Record<string, unknown>;
   restaurantId?: string;
+  campaignId?: string;
   /** Number of SMS units billed (1 for short templates, 3–4 for long ones). Defaults to 1. */
   quantity?: number;
 }

src/modules/notifications/processors/direct-sms.processor.ts

@@ -6,6 +6,7 @@ import { DirectSmsQueueJob } from '../interfaces/jobs-queue.interface';
 import { NotificationQueueNameEnum } from '../constants/queue';
 import { SmsService } from '../services/sms.service';
 import { SmsSentEvent } from '../events/sms.events';
+import { CampaignSmsDispatchResultEvent } from '../events/campaign-sms-dispatch.events';
 
 @Processor(NotificationQueueNameEnum.DIRECT_SMS)
 export class DirectSmsProcessor extends WorkerHost {
@@ -19,7 +20,7 @@ export class DirectSmsProcessor extends WorkerHost {
   }
 
   async process(job: Job<DirectSmsQueueJob>) {
-    const { phone, templateId, params, restaurantId, quantity = 1 } = job.data;
+    const { phone, templateId, params, restaurantId, campaignId, quantity = 1 } = job.data;
 
     this.logger.log(`Processing direct SMS - phone: ${phone}, templateId: ${templateId}`);
 
@@ -39,6 +40,13 @@ export class DirectSmsProcessor extends WorkerHost {
         );
       }
 
+      if (campaignId) {
+        this.eventEmitter.emit(
+          CampaignSmsDispatchResultEvent.name,
+          new CampaignSmsDispatchResultEvent(campaignId, true, quantity),
+        );
+      }
+
       return { success: true };
     } catch (error) {
       this.logger.error(`Error processing direct SMS job ${job.id}:`, error);
@@ -52,7 +60,18 @@ export class DirectSmsProcessor extends WorkerHost {
   }
 
   @OnWorkerEvent('failed')
-  onFailed(job: Job, error: Error) {
+  onFailed(job: Job<DirectSmsQueueJob>, error: Error) {
     this.logger.error(`Direct SMS job ${job.id} failed:`, error);
+
+    const totalAttempts = job.opts.attempts ?? 1;
+    const isFinalFailure = job.attemptsMade >= totalAttempts;
+    const campaignId = job.data?.campaignId;
+
+    if (isFinalFailure && campaignId) {
+      this.eventEmitter.emit(
+        CampaignSmsDispatchResultEvent.name,
+        new CampaignSmsDispatchResultEvent(campaignId, false, job.data.quantity ?? 1),
+      );
+    }
   }
 }

src/modules/restaurants/providers/restaurant-wallet.service.ts

@@ -0,0 +1,49 @@
+import { BadRequestException, Injectable } from '@nestjs/common';
+import { EntityManager } from '@mikro-orm/postgresql';
+import { Restaurant } from '../entities/restaurant.entity';
+import { RestaurantCreditTransaction } from '../entities/restaurant-credit-transaction.entity';
+import {
+  RestaurantCreditTransactionReason,
+  RestaurantCreditTransactionType,
+} from '../interface/restaurant-credit-transaction.interface';
+
+@Injectable()
+export class RestaurantWalletService {
+  constructor(private readonly defaultEm: EntityManager) {}
+
+  async createTransaction(params: {
+    restaurant: Restaurant;
+    amount: number;
+    type: RestaurantCreditTransactionType;
+    reason: RestaurantCreditTransactionReason;
+    insufficientBalanceMessage?: string;
+    em?: EntityManager;
+  }): Promise<void> {
+    const { restaurant, amount, type, reason, insufficientBalanceMessage, em = this.defaultEm } = params;
+
+    if (amount <= 0) {
+      return;
+    }
+
+    const currentCredit = Number(restaurant.credit ?? 0);
+    if (type === RestaurantCreditTransactionType.DEBIT && currentCredit < amount) {
+      throw new BadRequestException(insufficientBalanceMessage ?? 'اعتبار کیف پول رستوران کافی نیست');
+    }
+
+    const newBalance =
+      type === RestaurantCreditTransactionType.DEBIT ? currentCredit - amount : currentCredit + amount;
+
+    restaurant.credit = newBalance;
+
+    const creditTransaction = em.create(RestaurantCreditTransaction, {
+      restaurant,
+      amount,
+      balance: newBalance,
+      type,
+      reason,
+    });
+
+    await em.persistAndFlush([restaurant, creditTransaction]);
+  }
+
+}

src/modules/restaurants/restaurants.module.ts

@@ -15,12 +15,21 @@ import { UtilsModule } from '../utils/utils.module';
 import { Background } from './entities/background.entity';
 import { BackgroundRepository } from './repositories/background.repository';
 import { RestaurantCreditTransaction } from './entities/restaurant-credit-transaction.entity';
+import { RestaurantWalletService } from './providers/restaurant-wallet.service';
 
 @Module({
   controllers: [RestaurantsController, ScheduleController],
-  providers: [RestaurantsService, RestRepository, ScheduleRepository, ScheduleService, RestaurantCrone, BackgroundRepository],
+  providers: [
+    RestaurantsService,
+    RestRepository,
+    ScheduleRepository,
+    ScheduleService,
+    RestaurantCrone,
+    BackgroundRepository,
+    RestaurantWalletService,
+  ],
   imports: [MikroOrmModule.forFeature([Restaurant, Schedule, Background, RestaurantCreditTransaction]),
     JwtModule, forwardRef(() => AuthModule), UtilsModule],
-  exports: [RestRepository, ScheduleRepository, ScheduleService, RestaurantsService],
+  exports: [RestRepository, ScheduleRepository, ScheduleService, RestaurantsService, RestaurantWalletService],
 })
 export class RestaurantsModule { }

src/modules/users/entities/campaign.entity.ts

@@ -1,6 +1,6 @@
 import { Entity, Enum, Index, ManyToOne, Property } from '@mikro-orm/core';
 import { BaseEntity } from '../../../common/entities/base.entity';
-import { CampaignSentType } from '../interface/campaign';
+import { CampaignSentType, CampaignStatus } from '../interface/campaign';
 import { Restaurant } from 'src/modules/restaurants/entities/restaurant.entity';
 
 @Entity({ tableName: 'campaigns' })
@@ -20,4 +20,16 @@ export class Campaign extends BaseEntity {
 
   @Property()
   discountCode!: string;
+
+  @Property({ default: 0 })
+  totalCount: number = 0;
+
+  @Property({ default: 0 })
+  sentCount: number = 0;
+
+  @Property({ default: 0 })
+  failedCount: number = 0;
+
+  @Enum(() => CampaignStatus)
+  status: CampaignStatus = CampaignStatus.SENDING;
 }

src/modules/users/interface/campaign.ts

@@ -2,3 +2,8 @@ export enum CampaignSentType {
   PUSH = 'push',
   SMS = 'sms',
 }
+
+export enum CampaignStatus {
+  SENDING = 'sending',
+  COMPLETED = 'completed',
+}

src/modules/users/providers/campaign-dispatch.listener.ts

@@ -0,0 +1,85 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { EnsureRequestContext, LockMode } from '@mikro-orm/core';
+import { EntityManager } from '@mikro-orm/postgresql';
+import { OnEvent } from '@nestjs/event-emitter';
+import { Campaign } from '../entities/campaign.entity';
+import { CampaignStatus } from '../interface/campaign';
+import { CampaignSmsDispatchResultEvent } from 'src/modules/notifications/events/campaign-sms-dispatch.events';
+import { ConfigService } from '@nestjs/config';
+import { RestaurantWalletService } from 'src/modules/restaurants/providers/restaurant-wallet.service';
+import {
+  RestaurantCreditTransactionReason,
+  RestaurantCreditTransactionType,
+} from 'src/modules/restaurants/interface/restaurant-credit-transaction.interface';
+
+@Injectable()
+export class CampaignDispatchListener {
+  private readonly logger = new Logger(CampaignDispatchListener.name);
+
+  constructor(
+    private readonly em: EntityManager,
+    private readonly configService: ConfigService,
+    private readonly restaurantWalletService: RestaurantWalletService,
+  ) {}
+
+  @OnEvent(CampaignSmsDispatchResultEvent.name)
+  @EnsureRequestContext()
+  async handleCampaignSmsDispatchResult(event: CampaignSmsDispatchResultEvent): Promise<void> {
+    try {
+      await this.em.transactional(async em => {
+        const campaign = await em.findOne(
+          Campaign,
+          { id: event.campaignId },
+          { lockMode: LockMode.PESSIMISTIC_WRITE, populate: ['restaurant'] },
+        );
+
+        if (!campaign) {
+          this.logger.warn(`Campaign not found for dispatch result: ${event.campaignId}`);
+          return;
+        }
+
+        if (campaign.status === CampaignStatus.COMPLETED) {
+          return;
+        }
+
+        if (event.success) {
+          campaign.sentCount += event.quantity;
+        } else {
+          campaign.failedCount += event.quantity;
+        }
+
+        const processedCount = campaign.sentCount + campaign.failedCount;
+        if (campaign.totalCount > 0 && processedCount >= campaign.totalCount) {
+          campaign.status = CampaignStatus.COMPLETED;
+
+          if (campaign.failedCount > 0) {
+            const refundAmount = campaign.failedCount * this.getSmsUnitPrice();
+            await this.restaurantWalletService.createTransaction({
+              restaurant: campaign.restaurant,
+              amount: refundAmount,
+              type: RestaurantCreditTransactionType.CREDIT,
+              reason: RestaurantCreditTransactionReason.ADJUSTMENT,
+              em,
+            });
+          }
+        }
+
+        em.persist(campaign);
+      });
+    } catch (error) {
+      this.logger.error(
+        `Failed to update campaign dispatch counters: ${event.campaignId}`,
+        error instanceof Error ? error.stack : String(error),
+      );
+    }
+  }
+
+  private getSmsUnitPrice(): number {
+    const configuredPrice = this.configService.get<number>('SMS_UNIT_PRICE');
+    if (typeof configuredPrice === 'number' && Number.isFinite(configuredPrice) && configuredPrice >= 0) {
+      return configuredPrice;
+    }
+
+    return 1;
+  }
+}

src/modules/users/providers/campaign.service.ts

@@ -8,7 +8,13 @@ import { UserGroupRepository } from '../repositories/user-group.repository';
 import { UserGroupUserRepository } from '../repositories/user-group-user.repository';
 import { SmsQueueService } from 'src/modules/notifications/services/sms-queue.service';
 import { ConfigService } from '@nestjs/config';
-import { CampaignSentType } from '../interface/campaign';
+import { CampaignSentType, CampaignStatus } from '../interface/campaign';
+import { DirectSmsQueueJob } from 'src/modules/notifications/interfaces/jobs-queue.interface';
+import { RestaurantWalletService } from 'src/modules/restaurants/providers/restaurant-wallet.service';
+import {
+  RestaurantCreditTransactionReason,
+  RestaurantCreditTransactionType,
+} from 'src/modules/restaurants/interface/restaurant-credit-transaction.interface';
 
 @Injectable()
 export class CampaignService {
@@ -17,6 +23,7 @@ export class CampaignService {
     private readonly userGroupRepository: UserGroupRepository,
     private readonly userGroupUserRepository: UserGroupUserRepository,
     private readonly restaurantsService: RestaurantsService,
+    private readonly restaurantWalletService: RestaurantWalletService,
     private readonly smsQueueService: SmsQueueService,
     private readonly configService: ConfigService,
     private readonly em: EntityManager,
@@ -45,13 +52,39 @@ export class CampaignService {
       groups: uniqueGroups,
       sentType: dto.sentType,
       discountCode: dto.discountCode,
+      status: dto.sentType === CampaignSentType.SMS ? CampaignStatus.SENDING : CampaignStatus.COMPLETED,
+      totalCount: 0,
+      sentCount: 0,
+      failedCount: 0,
     };
 
     const campaign = this.campaignRepository.create(data);
     await this.em.persistAndFlush(campaign);
 
     if (dto.sentType === CampaignSentType.SMS) {
-      await this.runSmsDiscountCampaignFlow(restId, uniqueGroups, dto, restaurant.name);
+      const smsDispatchPlan = await this.buildSmsDiscountDispatchPlan(
+        restId,
+        uniqueGroups,
+        dto,
+        restaurant.name,
+        campaign.id,
+      );
+
+      campaign.totalCount = smsDispatchPlan.totalCount;
+      if (smsDispatchPlan.totalCount === 0) {
+        campaign.status = CampaignStatus.COMPLETED;
+      } else {
+        await this.restaurantWalletService.createTransaction({
+          restaurant,
+          amount: smsDispatchPlan.totalCost,
+          type: RestaurantCreditTransactionType.DEBIT,
+          reason: RestaurantCreditTransactionReason.SMS_SEND,
+          insufficientBalanceMessage: 'اعتبار کیف پول رستوران برای ارسال کمپین پیامکی کافی نیست',
+        });
+        await this.smsQueueService.enqueueBulk(smsDispatchPlan.jobs);
+      }
+
+      await this.em.persistAndFlush(campaign);
     } else if (dto.sentType === CampaignSentType.PUSH) {
       await this.runPushDiscountCampaignFlow(restId, uniqueGroups, dto);
     }
@@ -59,12 +92,13 @@ export class CampaignService {
     return campaign;
   }
 
-  private async runSmsDiscountCampaignFlow(
+  private async buildSmsDiscountDispatchPlan(
     restId: string,
     uniqueGroups: string[],
     dto: CreateCampaignDto,
     restaurantTitle: string,
-  ): Promise<void> {
+    campaignId: string,
+  ): Promise<{ jobs: DirectSmsQueueJob[]; totalCount: number; totalCost: number }> {
     const discountTemplateId = this.configService.get<string>('SMS_PATTERN_DISCOUNT_CODE');
     if (!discountTemplateId) {
       throw new BadRequestException('الگوی پیامک تخفیف تنظیم نشده است');
@@ -93,11 +127,14 @@ export class CampaignService {
     }
 
     if (recipientsByPhone.size === 0) {
-      return;
+      return {
+        jobs: [],
+        totalCount: 0,
+        totalCost: 0,
+      };
     }
 
-    await this.smsQueueService.enqueueBulk(
+    const jobs: DirectSmsQueueJob[] = [...recipientsByPhone.entries()].map(([phone, firstName]) => ({
-      [...recipientsByPhone.entries()].map(([phone, firstName]) => ({
         phone,
         templateId: discountTemplateId,
         restaurantId: restId,
@@ -108,8 +145,27 @@ export class CampaignService {
           oca: dto.ocasion,
           rest: restaurantTitle,
         },
-      })),
+        campaignId,
-    );
+      }));
+
+    const totalCount = jobs.length;
+    const smsUnitPrice = this.getSmsUnitPrice();
+    const totalCost = totalCount * smsUnitPrice;
+
+    return {
+      jobs,
+      totalCount,
+      totalCost,
+    };
+  }
+
+  private getSmsUnitPrice(): number {
+    const configuredPrice = this.configService.get<number>('SMS_UNIT_PRICE');
+    if (typeof configuredPrice === 'number' && Number.isFinite(configuredPrice) && configuredPrice >= 0) {
+      return configuredPrice;
+    }
+
+    return 1;
   }
 
   private async runPushDiscountCampaignFlow(

src/modules/users/user.module.ts

@@ -26,9 +26,10 @@ import { CampaignRepository } from './repositories/campaign.repository';
 import { CampaignService } from './providers/campaign.service';
 import { CampaignsController } from './controllers/campaigns.controller';
 import { NotificationsModule } from '../notifications/notifications.module';
+import { CampaignDispatchListener } from './providers/campaign-dispatch.listener';
 
 @Module({
-  providers: [UserService, WalletService, UserGroupService, CampaignService,
+  providers: [UserService, WalletService, UserGroupService, CampaignService, CampaignDispatchListener,
     UserRepository, WalletTransactionRepository,
     PointTransactionRepository, UserRestaurantRepository, UserGroupRepository, UserGroupUserRepository, CampaignRepository, UserRestaurantCrone],
   controllers: [UsersController, UserGroupsController, CampaignsController],