Code-Level Changes Required for Scaling

Context

These are backend code changes needed to support 100-200 concurrent calls. Some are blockers for the ECS migration (must be done before or in parallel), others are improvements that can follow.

Priority Legend:

• 🔴 P0 Blocker. Must be done before going live with ECS.
• 🟡 P1 Important. Should be done in the same sprint.
• 🟢 P2 Nice to have. Can follow in next sprint.

---

🔴 P0: Campaign State → Redis Migration

The Problem:calling.service.ts stores all campaign state in a JavaScript Map:

// client_backend/src/services/calling.service.ts (line ~10)
const campaignStates = new Map<string, CampaignState>();

This means:

• Cannot run 2+ backend instances (each has its own Map)
• State is lost on process restart/crash
• fillCallSlots() runs via setInterval tied to a single process
• activeCalls tracking is per-process, not global

The Fix: Move campaign state to Redis. Replace in-memory Map with Redis hashes and sets.

New dependencies:

npm install ioredis bullmq

Redis key design:

campaign:{campaignUID}:state      → Redis Hash
  - isRunning: "true"/"false"
  - maxConcurrent: "100"
  - phase: "initial"/"retry"
  - userId: "usr_xxx"
  - AgentID: "abc123"
  - isProcessing: "false"
  - lastProcessTime: "1234567890"

campaign:{campaignUID}:active     → Redis Set (phone numbers of active calls)
  Members: ["+919876543210", "+919876543211", ...]

campaign:{campaignUID}:config     → Redis Hash
  - sipTrunkId: "ST_xxx"
  - sipPhoneNumber: "+919876543210"
  - provider: "voicelink"
  - trunkUID: "trk_xxx"

Key functions to refactor:

Function	Current	After
initializeCampaignState()	Creates Map entry	HSET campaign:{uid}:state ...
getCampaignState()	Map.get()	HGETALL campaign:{uid}:state
cleanupCampaignState()	Map.delete()	DEL campaign:{uid}:state, DEL campaign:{uid}:active
addToActiveCalls()	Map.set() on activeCalls	SADD campaign:{uid}:active {phone}
removeFromActiveCalls()	Map.delete() on activeCalls	SREM campaign:{uid}:active {phone}
getActiveCallCount()	Map.size	SCARD campaign:{uid}:active
fillCallSlots()	Direct Map access	Redis ops + distributed lock
startCallingProcess()	setInterval(2000)	BullMQ repeatable job
stopCallingProcess()	clearInterval	Remove BullMQ job + cleanup Redis keys
recoverCampaignStates()	Query DB, check Map	Query DB, check Redis keys

fillCallSlots distributed lock: Only one backend instance should run fillCallSlots for a given campaign at a time. Use a simple Redis SET NX EX lock:

const lockKey = `campaign:${campaignUID}:fill_lock`;
const acquired = await redis.set(lockKey, instanceId, 'NX', 'EX', 5); // 5s TTL
if (!acquired) return; // another instance is already filling

try {
  // ... existing fillCallSlots logic but using Redis for state ...
} finally {
  await redis.del(lockKey);
}

Replace setInterval with BullMQ:

import { Queue, Worker } from 'bullmq';

const campaignQueue = new Queue('campaign-orchestrator', { connection: redisConnection });

// When campaign starts:
await campaignQueue.add(`manage-${campaignUID}`, { campaignUID, userId }, {
  repeat: { every: 2000 },  // every 2 seconds, same as current setInterval
  jobId: `campaign-${campaignUID}`,
});

// Worker (runs on all backend instances, BullMQ handles distribution):
const worker = new Worker('campaign-orchestrator', async (job) => {
  await manageCampaignCalls(job.data.campaignUID, job.data.userId);
}, { connection: redisConnection });

// When campaign stops:
await campaignQueue.removeRepeatableByKey(`campaign-${campaignUID}`);

Files to modify:

• client_backend/src/services/calling.service.ts major refactor
• client_backend/src/config/index.ts add REDIS_URL env var
• client_backend/src/server.ts initialize Redis connection + BullMQ workers

New file:

• client_backend/src/config/redis.ts Redis client singleton

import Redis from 'ioredis';
import { config } from './index.js';

let redisClient: Redis | null = null;

export function getRedis(): Redis {
  if (!redisClient) {
    redisClient = new Redis(config.redisUrl, {
      maxRetriesPerRequest: null, // required for BullMQ
      enableReadyCheck: false,
    });
  }
  return redisClient;
}

---

🔴 P0: Make maxConcurrent Configurable

The Problem: Hardcoded to 2 in two places:

// calling.service.ts line 45
async function initializeCampaignState(campaignUID: string, maxConcurrent = 2)

// calling.service.ts line 637
const state = await initializeCampaignState(campaignUID, 2);

Every campaign can only make 2 calls at a time. For 100 concurrent per org, this must be configurable.

The Fix:

1. Add field to Organization model:

// client_backend/src/models/Organization.ts
maxConcurrentCalls: { type: Number, default: 10, min: 1, max: 500 }

2. Add field to Campaign model:

// client_backend/src/models/Campaign.ts
maxConcurrentCalls: { type: Number, default: null, min: 1, max: 500 }
// null = use org default

3. Update initializeCampaignState and startCallingProcess:

// In initializeCampaignState or startCallingProcess:
const org = await Organization.findOne({ orgUID: campaign.orgUID });
const maxConcurrent = campaign.maxConcurrentCalls
  || org?.maxConcurrentCalls
  || 10; // platform default

4. Add API endpoints for SuperAdmin/Admin to set limits:

• PATCH /superadmin/organizations/:orgUID set org-level limit
• PATCH /api/v2/campaigns/:campaignUID set campaign-level limit

Files to modify:

• client_backend/src/models/Organization.ts
• client_backend/src/models/Campaign.ts
• client_backend/src/services/calling.service.ts
• Relevant controllers for the API endpoints

---

🔴 P0: Add Health Check Endpoint

The Problem: No health check endpoint exists. ALB needs one to know if a backend instance is healthy.

The Fix:

// client_backend/src/app.ts add before other routes
app.get('/health', async (req, res) => {
  try {
    const mongoOk = mongoose.connection.readyState === 1;
    // Once Redis is added:
    // const redisOk = await getRedis().ping() === 'PONG';

    const checks = {
      status: mongoOk ? 'healthy' : 'unhealthy',
      mongodb: mongoOk,
      // redis: redisOk,
      uptime: process.uptime(),
      timestamp: new Date().toISOString(),
    };

    res.status(mongoOk ? 200 : 503).json(checks);
  } catch (err) {
    res.status(503).json({ status: 'unhealthy', error: (err as Error).message });
  }
});

Files to modify:

• client_backend/src/app.ts

---

🔴 P0: MongoDB Connection Pool

The Problem: Default Mongoose connection pool is 5 connections. With 200 concurrent calls, each triggering multiple DB operations (status updates, webhook processing, credit deduction), the pool exhausts and requests queue up.

The Fix:

// client_backend/src/config/database.ts
export const connectDatabase = async (): Promise<void> => {
  await mongoose.connect(config.mongoUri, {
    maxPoolSize: 50,          // up from default 5
    minPoolSize: 5,           // keep some connections warm
    maxIdleTimeMS: 30000,     // close idle connections after 30s
    serverSelectionTimeoutMS: 5000,
    socketTimeoutMS: 45000,
  });
};

Also verify MongoDB Atlas tier:

• M10: 1500 connections max tight for 200 concurrent
• M30: 3000 connections max comfortable
• With 2 backend instances × 50 pool = 100 connections used

Files to modify:

• client_backend/src/config/database.ts

---

🟡 P1: Webhook Processing Queue

The Problem: LiveKit sends webhooks for every call event (participant_joined, participant_left, egress_ended). At 200 concurrent calls, that's 600-1000 webhook events per minute. Currently processed inline in the Express request handler blocks the event loop.

The Fix: Add a BullMQ queue for webhook processing:

// New file: client_backend/src/queues/webhook.queue.ts
import { Queue, Worker } from 'bullmq';
import { getRedis } from '../config/redis.js';

export const webhookQueue = new Queue('livekit-webhooks', {
  connection: getRedis(),
  defaultJobOptions: {
    removeOnComplete: 1000,
    removeOnFail: 5000,
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
  },
});

// Worker processes webhooks
export function startWebhookWorker() {
  const worker = new Worker('livekit-webhooks', async (job) => {
    const { event } = job.data;
    // ... existing webhook handling logic ...
  }, {
    connection: getRedis(),
    concurrency: 10, // process 10 webhooks in parallel
  });
  return worker;
}

// In webhook controller just enqueue, don't process inline:
app.post('/livekit-webhook', async (req, res) => {
  const event = await verifyWebhook(rawBody, authHeader);
  await webhookQueue.add(event.event, { event }, {
    jobId: `${event.event}-${event.id}`, // idempotent
  });
  res.status(200).send('ok'); // respond immediately
});

Files to modify:

• Webhook controller/route (wherever /livekit-webhook is handled)
• client_backend/src/server.ts start webhook worker

New file:

• client_backend/src/queues/webhook.queue.ts

---

🟡 P1: Per-Org Rate Limiting

The Problem: No limit on how many concurrent calls one org can consume. A single org could use all system resources.

The Fix: In fillCallSlots, check org-level concurrent limit using Redis:

// In fillCallSlots, before starting calls:
const orgUID = campaign.orgUID;
const orgActiveKey = `org:${orgUID}:active_calls`;
const orgActiveCalls = await redis.scard(orgActiveKey);
const orgLimit = org.maxConcurrentCalls || 100;

if (orgActiveCalls >= orgLimit) {
  console.log(`Org ${orgUID} at concurrent limit (${orgActiveCalls}/${orgLimit})`);
  return; // don't start more calls for this org
}

// When a call starts:
await redis.sadd(orgActiveKey, contactPhone);

// When a call ends (in webhook handler):
await redis.srem(orgActiveKey, contactPhone);

Also add a global system limit:

const globalActiveKey = 'global:active_calls';
const globalActive = await redis.scard(globalActiveKey);
const GLOBAL_LIMIT = 500; // system-wide safety cap

if (globalActive >= GLOBAL_LIMIT) {
  console.log(`Global concurrent limit reached (${globalActive}/${GLOBAL_LIMIT})`);
  return;
}

Files to modify:

• client_backend/src/services/calling.service.ts

---

🟡 P1: Switch Egress to Track-based Recording

The Problem: Current code uses startRoomCompositeEgress which costs 1.5 CPU per recording (it renders a full browser-based composite). For audio-only calls, this is massive overkill.

Current code:

// client_backend/src/services/livekit.service.ts
const opts: RoomCompositeOptions = {
  audioOnly: true,
  audioMixing: AudioMixing.DUAL_CHANNEL_AGENT,
};
const info = await client.startRoomCompositeEgress(roomName, output, opts);

The Fix: Switch to Track Composite Egress (~0.5 CPU) or Track Egress (~0.3 CPU):

// Option A: Track Composite (still mixes tracks, but no browser rendering)
const info = await client.startTrackCompositeEgress(
  roomName,
  output,
  {
    audioTrackId: '', // empty = mix all audio tracks
  }
);

// Option B: Participant Egress (records specific participant)
// Requires knowing participant SID get from webhook

⚠️ Testing required: Dual-channel agent mode (AudioMixing.DUAL_CHANNEL_AGENT) may only work with Room Composite. If so, we either:

1. Accept single-channel mixed audio (simpler, lower CPU)
2. Stay with Room Composite but provision more egress instances
3. Record at SIP provider level instead (VoiceLink/Telnyx may offer this)

Files to modify:

• client_backend/src/services/livekit.service.ts startRoomRecording()

---

🟢 P2: Structured Logging

The Problem: Current logging is console.log with emoji prefixes. Not parseable by CloudWatch Logs Insights or any log aggregation tool.

The Fix: Use a structured logger (pino or winston):

// client_backend/src/utils/logger.ts
import pino from 'pino';

export const logger = pino({
  level: process.env.LOG_LEVEL || 'info',
  formatters: {
    level: (label) => ({ level: label }),
  },
  timestamp: pino.stdTimeFunctions.isoTime,
});

// Usage:
logger.info({ campaignUID, activeCount, slotsAvailable }, 'fillCallSlots running');
// Instead of:
console.log(`🔍 fillCallSlots: Active: ${activeCount}, Max: ${state.maxConcurrent}, Available: ${slotsAvailable}`);

Files to modify:

• Create client_backend/src/utils/logger.ts
• Replace console.log/error/warn across all service files (can be done incrementally)

---

🟢 P2: Graceful Shutdown

The Problem: When the backend process is killed (ECS task replacement, deploy), active campaigns lose state and calls may not be cleaned up properly.

The Fix:

// client_backend/src/server.ts
process.on('SIGTERM', async () => {
  logger.info('SIGTERM received. Starting graceful shutdown...');

  // 1. Stop accepting new requests
  server.close();

  // 2. Stop all campaign intervals (or BullMQ workers)
  await stopAllCampaigns();

  // 3. Wait for in-flight requests to complete (max 30s)
  await new Promise(resolve => setTimeout(resolve, 30000));

  // 4. Close DB connections
  await mongoose.disconnect();
  await getRedis().quit();

  logger.info('Graceful shutdown complete');
  process.exit(0);
});

Files to modify:

• client_backend/src/server.ts

---

🟢 P2: Index Review for High-Frequency Queries

fillCallSlots runs every 2 seconds per campaign and queries:

Contact.find({ campaignUID, userId, status: "pending" }).sort({ createdAt: 1 }).limit(N)
Contact.countDocuments({ campaignUID, userId, status: "in_progress" })

Ensure compound indexes exist:

// client_backend/src/models/Contact.ts
contactSchema.index({ campaignUID: 1, userId: 1, status: 1, createdAt: 1 });

Also check:

campaignSchema.index({ orgUID: 1, isRunning: 1 }); // already exists ✅
contactSchema.index({ campaignUID: 1, status: 1 });  // verify this exists

Files to modify:

• client_backend/src/models/Contact.ts verify/add indexes

---

Summary Execution Order

#	Task	Priority	Effort	Depends On
1	Health check endpoint	🔴 P0	30 min	Nothing
2	MongoDB connection pool	🔴 P0	15 min	Nothing
3	maxConcurrent configurable	🔴 P0	2-3 hours	Nothing
4	Redis client setup	🔴 P0	1 hour	ElastiCache ready (DevOps)
5	Campaign state → Redis	🔴 P0	1-2 days	#4
6	Webhook queue (BullMQ)	🟡 P1	3-4 hours	#4
7	Per-org rate limiting	🟡 P1	2-3 hours	#5
8	Egress type switch	🟡 P1	3-4 hours + testing	Nothing
9	Structured logging	🟢 P2	4-6 hours	Nothing
10	Graceful shutdown	🟢 P2	1-2 hours	#5
11	Index review	🟢 P2	1 hour	Nothing

Items 1-3 can be done immediately, in parallel with DevOps work. Item 5 (campaign state → Redis) is the biggest piece and the most critical it unblocks running multiple backend instances.