NestJS SMS Marketing with Plivo: Production-Ready Campaign Tutorial

+-----------------+      +---------------------+      +-------------------+      +----------------+
|  Client (e.g., | ---> | NestJS API Gateway  | ---> | Campaign Service  | ---> | Plivo Service  | ---> | Plivo SMS API  |
|  Admin UI/CLI)  |      | (Controller)        |      | (Business Logic)  |      | (SDK Wrapper)  |      +----------------+
+-----------------+      +---------------------+      +-------------------+      +----------------+
                          |                     ^
                          |                     |
                          v                     |
                      +---------------------+   |
                      | Database (Postgres) | --+
                      | (Campaigns,         |
                      |  Subscribers)       |
                      +---------------------+

npm install -g @nestjs/cli
# or
yarn global add @nestjs/cli

nest new plivo-sms-campaign-app
cd plivo-sms-campaign-app

# Using npm
npm install @nestjs/config @nestjs/typeorm typeorm pg plivo class-validator class-transformer nestjs-pino pino-http @nestjs/schedule
npm install --save-dev @types/cron
# Potentially needed for TypeORM CLI depending on setup:
npm install --save-dev tsconfig-paths
# For Queuing (Optional but Recommended – See Section 9):
# npm install @nestjs/bull bull ioredis
# For Security (Optional – See Section 7):
# npm install helmet nestjs-rate-limiter rate-limiter-flexible @nestjs/jwt @nestjs/passport passport passport-jwt
# npm install --save-dev @types/passport-jwt
# For Monitoring (Optional – See Section 10):
# npm install @nestjs/terminus @nestjs/axios prom-client @willsoto/nestjs-prometheus
# npm install @sentry/node @sentry/tracing

# Using yarn
yarn add @nestjs/config @nestjs/typeorm typeorm pg plivo class-validator class-transformer nestjs-pino pino-http @nestjs/schedule @types/cron
yarn add --dev tsconfig-paths
# yarn add @nestjs/bull bull ioredis
# yarn add helmet nestjs-rate-limiter rate-limiter-flexible @nestjs/jwt @nestjs/passport passport passport-jwt
# yarn add --dev @types/passport-jwt
# yarn add @nestjs/terminus @nestjs/axios prom-client @willsoto/nestjs-prometheus
# yarn add @sentry/node @sentry/tracing

# .env

# Plivo Credentials
PLIVO_AUTH_ID=YOUR_PLIVO_AUTH_ID
PLIVO_AUTH_TOKEN=YOUR_PLIVO_AUTH_TOKEN
PLIVO_SOURCE_NUMBER=+14155551234 # Your Plivo phone number

# Database
DATABASE_URL=postgresql://user:password@localhost:5432/campaign_db

# Application
NODE_ENV=development
PORT=3000
LOG_LEVEL=info # Pino log level (trace, debug, info, warn, error, fatal)

# Security (Optional – See Section 7)
# API_KEY=YOUR_SECRET_API_KEY

# Redis (Optional – For Queues – See Section 9)
# REDIS_HOST=localhost
# REDIS_PORT=6379

// src/config/configuration.ts
import { registerAs } from '@nestjs/config';

export default registerAs('config', () => ({ // Use registerAs for better structure if needed elsewhere
  port: parseInt(process.env.PORT, 10) || 3000,
  nodeEnv: process.env.NODE_ENV || 'development',
  logLevel: process.env.LOG_LEVEL || 'info',
  apiKey: process.env.API_KEY, // Add API_KEY here
  plivo: {
    authId: process.env.PLIVO_AUTH_ID,
    authToken: process.env.PLIVO_AUTH_TOKEN,
    sourceNumber: process.env.PLIVO_SOURCE_NUMBER,
  },
  database: {
    url: process.env.DATABASE_URL,
    // Add SSL config if needed directly from env vars
    // ssl: process.env.DATABASE_SSL === 'true',
    // rejectUnauthorized: process.env.DATABASE_REJECT_UNAUTHORIZED !== 'false', // Example
  },
  redis: { // Add Redis config if using BullMQ
    host: process.env.REDIS_HOST || 'localhost',
    port: parseInt(process.env.REDIS_PORT, 10) || 6379,
  },
}));

// src/main.ts (add after app creation)
const configService = app.get(ConfigService);
console.log('Config loaded:', {
  port: configService.get('config.port'),
  nodeEnv: configService.get('config.nodeEnv'),
  plivoConfigured: !!configService.get('config.plivo.authId'),
});

// src/app.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config'; // Import ConfigService if needed globally
import { ScheduleModule } from '@nestjs/schedule';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import configuration from './config/configuration';
import { DatabaseModule } from './database/database.module';
import { PlivoModule } from './plivo/plivo.module';
import { CampaignModule } from './campaign/campaign.module';
import { LoggerModule } from './shared/logger/logger.module'; // We'll create this next
// Optional Modules:
// import { HealthModule } from './health/health.module';
// import { RateLimiterModule, RateLimiterGuard } from 'nestjs-rate-limiter';
// import { APP_GUARD } from '@nestjs/core';

@Module({
  imports: [
    ConfigModule.forRoot({
      isGlobal: true, // Make config available globally
      load: [configuration],
    }),
    ScheduleModule.forRoot(), // For scheduled tasks / simple retries
    LoggerModule, // Add LoggerModule
    DatabaseModule,
    PlivoModule,
    CampaignModule,
    // HealthModule, // Optional: Uncomment if using Health Checks (Section 10)
    // Optional: Rate Limiting (Section 7)
    /*
    RateLimiterModule.registerAsync({
      imports: [ConfigModule],
      inject: [ConfigService],
      useFactory: (configService: ConfigService) => ({
        // Configure based on configService if needed
        points: 100, // Example: 100 requests
        duration: 60, // Example: per 60 seconds
      }),
    }),
    */
  ],
  controllers: [AppController],
  providers: [
      AppService,
      // Optional: Apply Rate Limiter Globally (Section 7)
      /*
      {
        provide: APP_GUARD,
        useClass: RateLimiterGuard,
      },
      */
  ],
})
export class AppModule {}

// src/plivo/plivo.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { PlivoService } from './plivo.service';

@Module({
  imports: [ConfigModule], // Import ConfigModule if needed directly, or rely on global
  providers: [PlivoService],
  exports: [PlivoService], // Export the service for other modules to use
})
export class PlivoModule {}

// src/plivo/plivo.service.ts
import { Injectable, InternalServerErrorException, Logger } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import * as plivo from 'plivo';

@Injectable()
export class PlivoService {
  private readonly logger = new Logger(PlivoService.name);
  private client: plivo.Client;

  constructor(private configService: ConfigService) {
    const authId = this.configService.get<string>('config.plivo.authId');
    const authToken = this.configService.get<string>('config.plivo.authToken');

    if (!authId || !authToken) {
      this.logger.error('Plivo Auth ID or Auth Token missing in configuration.');
      throw new InternalServerErrorException('Plivo credentials are not configured.');
    }
    // The plivo-node library might internally handle basic retries,
    // but we initialize the client here.
    this.client = new plivo.Client(authId, authToken);
    this.logger.log('Plivo Client Initialized');
  }

  /**
   * Sends an SMS message using Plivo.
   * @param to The destination phone number (E.164 format recommended).
   * @param text The message content.
   * @returns The message UUID from Plivo upon successful queuing.
   * @throws InternalServerErrorException if sending fails.
   */
  async sendSms(to: string, text: string): Promise<string> {
    const sourceNumber = this.configService.get<string>('config.plivo.sourceNumber');
    if (!sourceNumber) {
        this.logger.error('Plivo Source Number missing in configuration.');
        throw new InternalServerErrorException('Plivo source number is not configured.');
    }

    this.logger.debug(`Attempting to send SMS to ${to}`);

    try {
      const response = await this.client.messages.create({
        src: sourceNumber,
        dst: to,
        text: text,
        // You can add a URL for delivery reports if needed (See Section 8 & 11):
        // url: 'https://yourapp.com/plivo/delivery-report',
        // method: 'POST'
      });

      // NOTE: Verify the structure of the response object with your SDK version.
      // Plivo API returns messageUuid as an array. Accessing [0] assumes success and a non-empty array.
      // Error handling might need to inspect the response structure more carefully.
      if (response.messageUuid && Array.isArray(response.messageUuid) && response.messageUuid.length > 0) {
          this.logger.log(`SMS queued successfully to ${to}. Message UUID: ${response.messageUuid[0]}`);
          return response.messageUuid[0];
      } else {
          // Log unexpected response structure
          this.logger.error(`Unexpected response structure from Plivo for ${to}: ${JSON.stringify(response)}`);
          throw new InternalServerErrorException('Unexpected response from Plivo after sending SMS.');
      }
    } catch (error) {
      this.logger.error(`Failed to send SMS to ${to}: ${error.message}`, error.stack);
      // Consider more specific error handling based on Plivo error codes/types if available
      throw new InternalServerErrorException(`Failed to send SMS via Plivo: ${error.message}`);
    }
  }

  // Potential future methods:
  // async getMessageStatus(messageUuid: string) { ... }
  // async handleIncomingSmsWebhook(payload: any) { ... } // For handling incoming messages/opt-outs
  // async handleDeliveryReportWebhook(payload: any) { ... } // For handling DLRs
}

// src/database/entities/campaign.entity.ts
import { Entity, PrimaryGeneratedColumn, Column, CreateDateColumn, UpdateDateColumn, OneToMany } from 'typeorm';
import { Subscriber } from './subscriber.entity';

@Entity('campaigns')
export class Campaign {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Column({ length: 255 })
  name: string;

  @Column('text')
  message: string;

  @Column({ default: false })
  isSent: boolean; // Or could track status: 'draft', 'sending', 'sent', 'failed'

  // Relationship: A campaign can have many subscribers (if tracking sends per campaign)
  // Or subscribers might be independent and targeted during the send operation.
  // For simplicity here, we won't link directly but query subscribers separately.
  // @OneToMany(() => CampaignSubscriber, campaignSubscriber => campaignSubscriber.campaign)
  // campaignSubscribers: CampaignSubscriber[];

  @CreateDateColumn()
  createdAt: Date;

  @UpdateDateColumn()
  updatedAt: Date;
}

// src/database/entities/subscriber.entity.ts
import { Entity, PrimaryGeneratedColumn, Column, CreateDateColumn, UpdateDateColumn, Index } from 'typeorm';

export enum SubscriberStatus {
  ACTIVE = 'active',
  INACTIVE = 'inactive', // e.g., opted-out
  PENDING = 'pending',   // e.g., needs confirmation
}

@Entity('subscribers')
export class Subscriber {
  @PrimaryGeneratedColumn('uuid')
  id: string;

  @Index({ unique: true }) // Ensure phone numbers are unique
  @Column({ length: 20, unique: true }) // E.164 max length is around 15, add buffer
  phoneNumber: string;

  @Column({ length: 100, nullable: true })
  firstName?: string;

  @Column({ length: 100, nullable: true })
  lastName?: string;

  @Column({
    type: 'enum',
    enum: SubscriberStatus,
    default: SubscriberStatus.ACTIVE,
  })
  status: SubscriberStatus;

  // Add tags or list associations if needed for segmentation later
  // @Column('simple-array', { nullable: true })
  // tags?: string[];

  @CreateDateColumn()
  createdAt: Date;

  @UpdateDateColumn()
  updatedAt: Date;
}

// src/database/database.module.ts
import { Module } from '@nestjs/common';
import { TypeOrmModule } from '@nestjs/typeorm';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { Campaign } from './entities/campaign.entity';
import { Subscriber } from './entities/subscriber.entity';

@Module({
  imports: [
    TypeOrmModule.forRootAsync({
      imports: [ConfigModule],
      inject: [ConfigService],
      useFactory: (configService: ConfigService) => {
        const isProduction = configService.get<string>('config.nodeEnv') === 'production';
        return {
            type: 'postgres',
            url: configService.get<string>('config.database.url'),
            entities: [Campaign, Subscriber],
            synchronize: !isProduction, // Auto-create schema in dev, use migrations in prod
            logging: !isProduction ? 'all' : ['error'], // Log SQL in dev
            migrations: [__dirname + '/migrations/*{.ts,.js}'],
            migrationsTableName: 'migrations_history', // Optional: custom migrations table name
            ssl: isProduction
              ? {
                  /**
                   * WARNING: rejectUnauthorized: false is insecure for production environments
                   * as it disables SSL certificate validation.
                   * This should only be used if your database provider requires it AND
                   * you understand the risks (e.g., man-in-the-middle attacks).
                   * Prefer using CA certificates or platform-specific secure connection methods.
                   * Example for Heroku/AWS RDS might need this, but investigate secure options first.
                   */
                  rejectUnauthorized: false, // <--- SECURITY WARNING
                  // Example using CA cert:
                  // ca: configService.get<string>('DATABASE_CA_CERT'),
                }
              : false,
        }
      },
    }),
    // Optionally, re-export TypeOrmModule.forFeature here if needed by services in this module
  ],
  // No providers or exports needed unless you have database-specific services
})
export class DatabaseModule {}

// package.json (add to "scripts")
{
  "scripts": {
    // ... other scripts
    "build": "nest build",
    "start": "node dist/main",
    "start:dev": "nest start --watch",
    "start:debug": "nest start --debug --watch",
    "start:prod": "node dist/main",
    "lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
    "test": "jest",
    "test:watch": "jest --watch",
    "test:cov": "jest --coverage",
    "test:debug": "node --inspect-brk -r tsconfig-paths/register -r ts-node/register node_modules/.bin/jest --runInBand",
    "test:e2e": "jest --config ./test/jest-e2e.json",
    "typeorm": "ts-node -r tsconfig-paths/register ./node_modules/typeorm/cli.js --dataSource src/database/data-source.ts",
    "migration:generate": "npm run typeorm -- migration:generate src/database/migrations/%npm_config_name%",
    "migration:run": "npm run typeorm -- migration:run",
    "migration:revert": "npm run typeorm -- migration:revert"
  }
}

// src/database/data-source.ts
import { DataSource, DataSourceOptions } from 'typeorm';
import * as dotenv from 'dotenv';
import * as path from 'path';

// Load environment variables from .env file relative to project root
dotenv.config({ path: path.resolve(__dirname, '../../.env') });

export const dataSourceOptions: DataSourceOptions = {
  type: 'postgres',
  url: process.env.DATABASE_URL,
  entities: [path.join(__dirname, '/entities/*.entity{.ts,.js}')], // Use path.join for robustness
  migrations: [path.join(__dirname, '/migrations/*{.ts,.js}')],
  logging: process.env.NODE_ENV !== 'production' ? ['query', 'error'] : ['error'],
  synchronize: false, // Never use synchronize true for migrations generation/running
  migrationsTableName: 'migrations_history',
  ssl: process.env.NODE_ENV === 'production'
    ? {
        /**
         * WARNING: rejectUnauthorized: false is insecure for production environments.
         * See warning in database.module.ts. Match the configuration there.
         */
        rejectUnauthorized: false, // <--- SECURITY WARNING (Match app module config)
        // ca: process.env.DATABASE_CA_CERT, // Example
      }
    : false,
};

const dataSource = new DataSource(dataSourceOptions);
export default dataSource;

version: '3.8'
services:
  postgres:
    image: postgres:14-alpine
    container_name: plivo-campaign-db
    restart: always
    environment:
      POSTGRES_DB: campaign_db
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

# Ensure your .env file has the correct DATABASE_URL
# The --name argument (-n) specifies the migration file name
npm run migration:generate --name=InitialSchema
# or
# yarn migration:generate -n InitialSchema

npm run migration:run
# or
# yarn migration:run

# Using Docker:
docker exec -it plivo-campaign-db psql -U user -d campaign_db -c "\dt"

# Should show: campaigns, subscribers, migrations_history tables

// src/campaign/dto/create-campaign.dto.ts
import { IsString, MinLength, MaxLength, IsNotEmpty } from 'class-validator';

export class CreateCampaignDto {
  @IsString()
  @IsNotEmpty()
  @MinLength(3)
  @MaxLength(255)
  name: string;

  @IsString()
  @IsNotEmpty()
  @MinLength(5)
  @MaxLength(1600) // Max length for concatenated SMS (10 segments × 160 GSM-7 chars)
  message: string;
}

// src/campaign/dto/add-subscriber.dto.ts
import { IsString, IsNotEmpty, IsPhoneNumber, IsOptional, MaxLength } from 'class-validator';

export class AddSubscriberDto {
  @IsPhoneNumber(null) // Use null for region-agnostic phone number validation (basic format check)
  @IsNotEmpty()
  phoneNumber: string; // Should be E.164 format

  @IsString()
  @IsOptional()
  @MaxLength(100)
  firstName?: string;

  @IsString()
  @IsOptional()
  @MaxLength(100)
  lastName?: string;
}

// src/common/utils/phone.util.ts
import { parsePhoneNumber, CountryCode } from 'libphonenumber-js';

export function normalizeToE164(phoneNumber: string, defaultCountry: CountryCode = 'US'): string {
  try {
    const parsed = parsePhoneNumber(phoneNumber, defaultCountry);
    if (!parsed || !parsed.isValid()) {
      throw new Error('Invalid phone number');
    }
    return parsed.format('E.164');
  } catch (error) {
    throw new Error(`Failed to normalize phone number: ${error.message}`);
  }
}

// Install: npm install libphonenumber-js

// src/campaign/campaign.service.ts
import { Injectable, Logger, NotFoundException, BadRequestException, InternalServerErrorException } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { Campaign } from '../database/entities/campaign.entity';
import { Subscriber, SubscriberStatus } from '../database/entities/subscriber.entity';
import { PlivoService } from '../plivo/plivo.service';
import { CreateCampaignDto } from './dto/create-campaign.dto';
import { AddSubscriberDto } from './dto/add-subscriber.dto';
// Import Queue if using advanced queuing (See Section 9)
// import { Queue } from 'bull';
// import { InjectQueue } from '@nestjs/bull';

@Injectable()
export class CampaignService {
  private readonly logger = new Logger(CampaignService.name);

  constructor(
    @InjectRepository(Campaign)
    private campaignRepository: Repository<Campaign>,
    @InjectRepository(Subscriber)
    private subscriberRepository: Repository<Subscriber>,
    private plivoService: PlivoService,
    // Inject Queue if using BullMQ (See Section 9)
    // @InjectQueue('sms-queue') private smsQueue: Queue,
  ) {}

  async createCampaign(createCampaignDto: CreateCampaignDto): Promise<Campaign> {
    const campaign = this.campaignRepository.create(createCampaignDto);
    const savedCampaign = await this.campaignRepository.save(campaign);
    this.logger.log(`Campaign created with ID: ${savedCampaign.id}`);
    return savedCampaign;
  }

  async findAllCampaigns(page: number = 1, limit: number = 20): Promise<{ campaigns: Campaign[]; total: number; page: number; totalPages: number }> {
    const [campaigns, total] = await this.campaignRepository.findAndCount({
      order: { createdAt: 'DESC' },
      skip: (page - 1) * limit,
      take: limit,
    });

    return {
      campaigns,
      total,
      page,
      totalPages: Math.ceil(total / limit),
    };
  }

  async findCampaignById(id: string): Promise<Campaign> {
    const campaign = await this.campaignRepository.findOneBy({ id });
    if (!campaign) {
      throw new NotFoundException(`Campaign with ID ${id} not found`);
    }
    return campaign;
  }

  async addSubscriber(addSubscriberDto: AddSubscriberDto): Promise<Subscriber> {
    // Normalize phone number to E.164 format
    // In production, use normalizeToE164() helper here
    const subscriber = this.subscriberRepository.create({
      ...addSubscriberDto,
      status: SubscriberStatus.ACTIVE,
    });

    try {
      const savedSubscriber = await this.subscriberRepository.save(subscriber);
      this.logger.log(`Subscriber added: ${savedSubscriber.phoneNumber}`);
      return savedSubscriber;
    } catch (error) {
      if (error.code === '23505') { // PostgreSQL unique violation
        throw new BadRequestException(`Phone number ${addSubscriberDto.phoneNumber} already exists`);
      }
      throw error;
    }
  }

  async findAllSubscribers(page: number = 1, limit: number = 50): Promise<{ subscribers: Subscriber[]; total: number; page: number; totalPages: number }> {
    const [subscribers, total] = await this.subscriberRepository.findAndCount({
      order: { createdAt: 'DESC' },
      skip: (page - 1) * limit,
      take: limit,
    });

    return {
      subscribers,
      total,
      page,
      totalPages: Math.ceil(total / limit),
    };
  }

  async updateSubscriberStatus(phoneNumber: string, status: SubscriberStatus): Promise<Subscriber> {
    const subscriber = await this.subscriberRepository.findOne({ where: { phoneNumber } });
    if (!subscriber) {
      throw new NotFoundException(`Subscriber ${phoneNumber} not found`);
    }

    subscriber.status = status;
    return this.subscriberRepository.save(subscriber);
  }

  async handleOptOut(phoneNumber: string): Promise<void> {
    this.logger.log(`Processing opt-out for ${phoneNumber}`);
    await this.updateSubscriberStatus(phoneNumber, SubscriberStatus.INACTIVE);
  }

  async sendCampaign(campaignId: string): Promise<{ message: string; totalSent: number; failedNumbers: string[] }> {
    this.logger.log(`Initiating send for campaign ID: ${campaignId}`);
    const campaign = await this.findCampaignById(campaignId);

    if (campaign.isSent) {
      throw new BadRequestException(`Campaign ${campaignId} has already been sent.`);
    }

    // Fetch active subscribers with pagination for large lists
    const subscribers = await this.subscriberRepository.find({
      where: { status: SubscriberStatus.ACTIVE },
      select: ['phoneNumber'],
    });

    if (subscribers.length === 0) {
      this.logger.warn(`No active subscribers found for campaign ${campaignId}.`);
      return { message: 'No active subscribers to send to.', totalSent: 0, failedNumbers: [] };
    }

    this.logger.log(`Found ${subscribers.length} active subscribers for campaign ${campaignId}.`);

    let sentCount = 0;
    const failedNumbers: string[] = [];

    // --- Approach 1: Parallel Sending with Promise.all ---
    // WARNING: This approach sends all messages concurrently. It WILL likely hit Plivo's rate limits
    // for any reasonably sized list. It can overwhelm the Plivo API and lead to failures.
    // USE WITH EXTREME CAUTION and only for very small lists or low rate limits.
    // The QUEUE approach below is STRONGLY RECOMMENDED for production.
    this.logger.warn(`Using Promise.all for sending – This is NOT recommended for production due to rate limits!`);
    const sendPromises = subscribers.map(async (subscriber) => {
      try {
        await this.plivoService.sendSms(subscriber.phoneNumber, campaign.message);
        return { success: true, number: subscriber.phoneNumber };
      } catch (error) {
        this.logger.error(`Failed sending to ${subscriber.phoneNumber}: ${error.message}`);
        return { success: false, number: subscriber.phoneNumber };
      }
    });

    const results = await Promise.all(sendPromises);

    results.forEach(result => {
      if (result.success) {
        sentCount++;
      } else {
        failedNumbers.push(result.number);
      }
    });
    // --- End Approach 1 ---

    // --- Approach 2: Basic Sequential Sending (Better than Promise.all for rate limits, but SLOW) ---
    /*
    this.logger.log(`Using sequential sending loop.`);
    for (const subscriber of subscribers) {
      try {
        await this.plivoService.sendSms(subscriber.phoneNumber, campaign.message);
        sentCount++;
        // Basic rate limiting delay – adjust based on your Plivo MPS limit
        // Example: If limit is 1 MPS, delay by 1000 ms. If 10 MPS, delay by 100 ms.
        await new Promise(resolve => setTimeout(resolve, 100)); // e.g., 100 ms delay for ~10 MPS
      } catch (error) {
        this.logger.error(`Failed sending to ${subscriber.phoneNumber}: ${error.message}`);
        failedNumbers.push(subscriber.phoneNumber);
      }
    }
    */
    // --- End Approach 2 ---

    // --- Approach 3: Recommended Production Approach – Using a Queue (See Section 9) ---
    /*
    this.logger.log(`Adding ${subscribers.length} SMS jobs to the queue.`);
    for (const subscriber of subscribers) {
      try {
        await this.smsQueue.add('send-sms-job', {
          to: subscriber.phoneNumber,
          text: campaign.message,
          campaignId: campaignId,
        }, {
          attempts: 3,
          backoff: { type: 'exponential', delay: 1000 },
          removeOnComplete: true,
          removeOnFail: 5000,
        });
      } catch (error) {
        this.logger.error(`Failed to add job for ${subscriber.phoneNumber} to queue: ${error.message}`);
        failedNumbers.push(subscriber.phoneNumber);
      }
    }
    sentCount = subscribers.length - failedNumbers.length;
    this.logger.log(`Added ${sentCount} SMS jobs to the queue for campaign ${campaignId}. Failures adding: ${failedNumbers.length}`);
    */
    // --- End Approach 3 ---

    // Mark campaign as sent (use transaction in production)
    campaign.isSent = true;
    await this.campaignRepository.save(campaign);

    return { message: 'Campaign send initiated.', totalSent: sentCount, failedNumbers };
  }
}

// Example using TypeORM transactions
async sendCampaignWithTransaction(campaignId: string): Promise<any> {
  return await this.campaignRepository.manager.transaction(async (transactionalEntityManager) => {
    const campaign = await transactionalEntityManager.findOne(Campaign, { where: { id: campaignId } });

    if (!campaign || campaign.isSent) {
      throw new BadRequestException('Campaign already sent or not found');
    }

    // Send messages...

    campaign.isSent = true;
    await transactionalEntityManager.save(campaign);

    return { message: 'Campaign sent successfully' };
  });
}