Oktivo Platform Campaign & Call Flow Architecture

Production architecture for high-throughput AI voice calling: thousands of calls/minute, distributed orchestration, real-time credit billing, and campaign lifecycle management.

---

Platform Overview

Oktivo is a white-label AI voice calling platform with a 3-tier hierarchy:

SuperAdmin (platform owner)
  └── Organization (white-label tenant, owns credit pool + concurrent call limits)
        └── Customer (end user who runs campaigns, has own credit balance)

A customer uploads a CSV of contacts, creates a campaign, and starts it. The system calls each contact via SIP, connects them to an AI agent, records the conversation, analyzes it, deducts credits, and stores everything on the Contact document.

---

High-Level Production Architecture

---

How It Handles Thousands of Calls/Minute

Distributed Orchestration via Redis + BullMQ

No single backend instance owns a campaign. All state lives in Redis, and BullMQ distributes work across instances.

Key design decisions:

• Each campaign gets a repeatable BullMQ job that fires every 2 seconds
• fillCallSlots() acquires a distributed lock (5s TTL) so only one instance fills slots per campaign at a time
• Worker concurrency = 5 jobs in parallel, rate limited to 50 jobs/second
• Active calls tracked in Redis hashes (not in-memory Maps) survives instance restarts

Concurrent Call Rate Limiting (3 levels)

Resolution priority for maxConcurrent:

1. campaign.maxConcurrentCalls (set per campaign, 1-500)
2. org.plan.maxConcurrentCalls (set by SuperAdmin when creating org)
3. Platform default: 10

Example: Org has maxConcurrentCalls=50, customer runs 3 campaigns each with maxConcurrent=20. The org-level cap ensures total across all 3 campaigns never exceeds 50.

---

Complete Campaign Lifecycle

Step 1: CSV Upload → Campaign + Contacts Created

CSV limits:

• Max file size: 30MB (enforced on both frontend and backend via multer)
• Deduplication by phone number within the CSV
• Invalid phone numbers are skipped during parsing

Step 2: Campaign Start (with Duplicate Protection)

Restart protection: When a completed campaign is started again, the backend returns a 409 requiring explicit confirmation. On confirm, all contacts are reset to pending with call data cleared. This prevents accidental duplicate billing.

Phone number resolution: The Campaign model does NOT store customerUID. Instead, campaign.userId (the customer's userUID) is passed to telephonyService.resolveCallConfig() as the customerUID parameter. This resolves the phone number assigned to that customer within the org. The same userId is used as customerUID fallback when dispatching individual calls in fillCallSlots().

Step 3: Call Dispatch Loop (every 2 seconds)

Step 4: Single Call From Dispatch to Completion

Step 5: Credit Deduction Flow

Example: 3-minute call, customerRate=3 cr/min, platformRate=1 cr/min

• Customer charged: 9 credits
• Org charged: 3 credits (from org pool)
• Org profit retained: 6 credits

Step 6: Campaign Completion + Retry Mechanism

Contact Status Lifecycle

Campaign Stop Flow

Campaign Restart Flow (with Stopped Contacts)

When restarting a stopped campaign, the backend returns a 409 with callStatusBreakdown that includes a "stopped" entry for contacts that were stopped by the user. The frontend shows these as a selectable option in the restart dialog.

Restart filter handling:

• "stopped" is a contact status (not a callStatus), so it's filtered by status: "stopped"
• Other options (ended, rejected, etc.) are filtered by callStatus: { $in: [...] }
• When both are selected, an $or query combines them

---

Real-Time Frontend Updates

The SSE endpoint streams both campaign status (isRunning, counts) and contact updates every 3 seconds. The campaigns list page additionally polls the campaigns API every 5 seconds while any campaign is running, so the user sees real-time progress without manual refresh.

---

Contact Single Source of Truth

The Contact document holds ALL call data. No need to query separate collections.

Group	Fields	Set By
Identity	name, phone, contactUID, campaignUID, AgentID, userId, customerUID, orgUID	CSV upload (customerUID = userId)
Flags	isTestCall	Test call creation (true) / CSV upload (false, default)
Status	status, callStatus, attempts, lastError	Webhook + calling.service
Timing	participantJoinedAt, participantLeftAt, callAnsweredAt	Webhook + /start-recording
LiveKit	roomId, sipCallId, sipTrunkId, disconnectReason, livekitData	Webhook
Recording	s3RecordingKey, egressId, recordingUrl, billingDuration, callDuration, AnsDuration	Webhook + Egress
Transcription	transcription (full conversation)	Agent → POST /transcription-sync
Analysis	analysis (summary, sentiment, interests)	Analysis → POST /analysis-sync

---

Call Status Mapping

SIP Status → callStatus

sip.callStatus	disconnectReason	→ callStatus
active	ROOM_DELETED	ended_by_agent
active	MEDIA_FAILURE	media_failure
active	other	ended
hangup	CLIENT_INITIATED	ended_by_user
ringing	USER_REJECTED / USER_UNAVAILABLE	rejected
ringing	CONNECTION_TIMEOUT	not picked

callStatus → Contact.status

callStatus	→ Contact.status
ended, ended_by_agent, ended_by_user, active, hangup, completed	done
failed, rejected, not_answered, not picked, media_failure, unknown, dialing, ringing	failed
force_stopped	stopped (set by force-stop or drain timeout, not by webhook)
null (no call made)	stopped (set when campaign is stopped before contact was dispatched)

Contact Status Values

Status	Meaning	Set By
pending	Waiting to be called	CSV upload, campaign restart
in_progress	Currently being called	fillCallSlots → startSingleCall
done	Call completed successfully	Webhook (handleCallStatusUpdate)
failed	Call failed (rejected, no answer, media failure, etc.)	Webhook or stale call timeout
skipped	Invalid phone number	fillCallSlots validation
stopped	Campaign was stopped before/during call	stopCallingProcess / finalizeCampaignStop / force-stop

---

Recording Pipeline

Recording starts only after the phone user picks up (not during ringing):

Timeline:
  T+0.0s  SIP participant joins room → participant_joined fires
  T+0.0s  Phone starts ringing (sip.callStatus = "dialing")
  T+7.4s  User picks up → create_sip_participant() returns
  T+7.4s  Agent calls POST /start-recording → egress starts 🎙️
  T+7.4s  callAnsweredAt stored on Contact
  T+30.0s User hangs up → participant_left fires → egress stopped ⏹️
  T+30.0s billingDuration = ceil(30.0 - 7.4) = 23s

S3 Folder Structure

s3://oktivo-recordings/
  └── {orgUID}/
        └── {customerUID}/
              └── {campaignUID}/
                    └── {contactUID}/
                          └── {phone}_{YYYYMMDD_HHmmss}.mp4

Duration Computation (computeBillingDuration())

1. Call must have been answered (sipCallStatus = "active" or "hangup")
2. startTime = callAnsweredAt ?? participantJoinedAt
3. durationSec = ceil((participantLeftAt - startTime) / 1000)
4. Validate: > 0 and ≤ 14400 (4 hours)
5. Write billingDuration = callDuration = AnsDuration atomically

---

Safety Nets & Failure Recovery

Stale Call Cleanup

Every 2-second management loop checks for calls active longer than 5 minutes. These are removed from Redis and the contact is marked failed (retryable).

Campaign Drain Timeout

When a campaign is stopped, active calls are allowed to finish naturally (draining). If draining exceeds 2 minutes, the system force-finalizes: remaining in-progress contacts are marked as stopped with callStatus: "force_stopped", Redis state is cleaned up, and the campaign is fully stopped.

Force-Stop (User-Initiated)

If a user clicks Stop a second time while the campaign is draining, a force-stop is triggered:

• All in-progress contacts are immediately marked as stopped (callStatus: "force_stopped")
• Redis active call tracking is cleaned up
• Campaign is finalized immediately without waiting for webhooks

Stale stopRequested Detection

When a campaign is restarted, there's a race condition where finalizeCampaignStop from the old run might set stopRequested: true after the restart clears it. fillCallSlots detects this by checking if the Redis state has no stopRequestedAt (freshly initialized) if so, it clears the stale stopRequested and continues dispatching.

room_finished Safety Net

When a LiveKit room closes, the handler checks for:

• Stuck in_progress contacts → marks as failed or ended
• Missing recordingUrl → attempts S3 presigned URL generation
• Missing participantLeftAt → sets it to now, computes billing duration

completeCampaign() Safety Net

Before marking a campaign as done, any remaining in_progress contacts are force-marked as failed. This prevents contacts from being permanently stuck.

finalizeCampaignStop() Completion Check

Before marking a campaign as stopped, checks if all contacts are actually processed (done/failed/skipped with no stopped contacts). If so, treats it as a natural completion instead of a stop. This prevents campaigns from being incorrectly marked as "stopped" when all calls actually completed during the drain period.

Billing Duration Race Condition

On short calls, participant_left can fire before POST /start-recording commits callAnsweredAt. The handler re-fetches the Contact right before computing duration to get the latest data.

Credit Deduction Idempotency

Each credit deduction uses serviceRef = call_{contactUID}_{roomName} as an idempotency key. If a webhook fires twice (retry), the second deduction is skipped. The consumeCredits() function checks for existing transactions with the same serviceRef before processing.

Call History Archive Deduplication

When restarting a campaign, call data is archived to callHistory before resetting contacts. The archive function checks if the last callHistory entry already matches the current data (same callStatus, s3RecordingKey, callStartedAt) to prevent duplicate entries from repeated restarts.

Server Restart Recovery

On startup, recoverCampaignStates() scans for campaigns with isRunning=true in MongoDB:

• If Redis state exists and is running → skip (another instance is handling it)
• If no pending/in_progress contacts → mark campaign as complete
• Otherwise → reset stale in_progress contacts to pending and restart the campaign

---

Campaign Restart (Duplicate Protection)

This prevents accidental duplicate billing when a user clicks Start on a campaign that already ran. The confirmation dialog shows how many contacts were done/failed so the user understands the impact.

---

Test Calls

Test calls use the same webhook pipeline as campaign calls. When a customer triggers a test call:

1. A real Contact document is created with isTestCall: true, status: "in_progress", and roomId set
2. campaignUID is set to test_{agentId}_{timestamp} (won't collide with real campaigns)
3. customerUID is set to req.userId (same as campaign flow)
4. Phone number is resolved via telephonyService.resolveCallConfig(orgUID, userId) identical to campaign start
5. Call is dispatched to LiveKit via createDispatch() with the same metadata shape

Because the Contact document exists with roomId, the entire webhook pipeline works identically:

• participant_joined → stores sipCallId, participantJoinedAt
• POST /start-recording → starts egress, stores callAnsweredAt
• participant_left → computes billingDuration, deducts credits, updates status
• egress_ended → stores recordingUrl
• POST /transcription-sync → stores transcription
• POST /analysis-sync → stores analysis

Credits ARE deducted for test calls (same as campaign calls). A legacy TestCall record is also created for backward compatibility.

Update (April 2026): Test calls are now free the webhook controller skips credit deduction when contact.isTestCall === true. Test calls still go through the full pipeline (recording, transcription, analysis) but no credits are consumed.

---

Webhook Endpoints

Endpoint	Called By	Purpose
POST /livekit-webhook	LiveKit Server	participant_joined, participant_left, egress_ended, room_finished
POST /start-recording	Python Agent	Start egress after call answered, store callAnsweredAt
POST /transcription-sync	Python Agent	Store conversation transcription on Contact
POST /analysis-sync	Analysis Service	Store sentiment/interest analysis on Contact
POST /update-call-status	Legacy	Backward-compatible call status update

Customer-Facing API Endpoints

Endpoint	Auth	Purpose
GET /contacts	JWT	List contacts with full filter support: date range, campaign, agent, search, direction, duration range, bookmarked, and dynamic analysis.* filters. See "Contact List Filters" section below.
GET /recording/:contactUID	Query token	Stream/redirect recording audio for <audio> elements. Uses presigned S3 URL.
GET /campaigns	JWT	List campaigns with stats (includes stopped count). Returns scheduledStartTime, scheduledEndTime, scheduled status.
POST /upload-campaign	JWT	Upload CSV + create campaign. Accepts SelectedAgentId, scheduledStartTime.
POST /start-campaign/:uid	JWT	Start campaign. Returns 409 with requiresConfirmation + callStatusBreakdown (includes stopped count) for completed/stopped campaigns.
POST /stop-campaign/:uid	JWT	Stop campaign. Accepts { forceStop: true } to immediately terminate all active calls.
GET /dashboard-stats	JWT	Dashboard analytics. Supports range=custom&from=&to= for date filtering.

---

Key Files

Backend (client_backend/src/)

File	Purpose
services/calling.service.ts	Campaign orchestration BullMQ worker, fillCallSlots, startSingleCall, retry logic, Redis state management
services/credit.service.ts	3-tier credit system consumeCredits (atomic), checkBalance, distributeToCustomer
controllers/webhook.controller.ts	LiveKit webhooks, recording start, transcription/analysis sync, credit deduction on call end
controllers/campaign.controller.ts	Campaign CRUD, start/stop with duplicate protection, SSE real-time updates, CSV upload with 30MB limit
controllers/agent.controller.ts	Agent CRUD, test call dispatch with Contact creation (isTestCall: true)
services/livekit.service.ts	LiveKit SDK createDispatch, DispatchMetadata interface, egress, webhook verification
services/telephony.service.ts	Resolves SIP trunk + phone number for outbound calls (via PhoneNumberAssignment)
services/s3.service.ts	S3 recording key builder, presigned URLs, existence checks
utils/dispatch.utils.ts	DRY dispatch helpers buildAgentConfig, mergeDynamicValues, fetchDynamicTools, fetchMcpServers, extractIvrActions, extractVoiceBehaviorFields
utils/duration.utils.ts	computeBillingDuration() pure function, provider-agnostic
utils/csv.utils.ts	parseCSVBuffer() CSV parsing, deduplication, sets customerUID = userId on contacts
repositories/campaign.repository.ts	Campaign DB queries, getCampaignStats (explicit status counts)
config/redis.ts	Redis connection (ElastiCache in production)

Frontend (app/src/)

File	Purpose
app/src/pages/customer/CampaignsPage.tsx	Campaign list with auto-polling (5s while running), restart confirmation dialog, agent change warnings
app/src/pages/customer/UploadPage.tsx	CSV upload with 50K row limit, agent selection, credit estimation, CSV preview, schedule campaign
app/src/pages/customer/CampaignDetailPage.tsx	Campaign detail with auto-polling (8s while running)
app/src/pages/customer/HistoryPage.tsx	Call history with date range, campaign/agent filters, direction, duration, bookmarked, analysis filters, per-page selector, CSV export, auto-polling for in-progress contacts
app/src/pages/customer/AnalyticsPage.tsx	Analytics with date range, campaign/agent filters, per-campaign daily chart
app/src/services/campaign.service.ts	API client start(uid, agentId, confirmRestart?), upload(formData), listContacts(params)

Contact List Filters (GET /contacts)

The History page uses the following query parameters to filter contacts:

Parameter	Type	Description
from	ISO date string	Start of date range (inclusive)
to	ISO date string	End of date range (inclusive, end of day)
status	string	"stopped" filters by contact status; all others filter by callStatus
campaignUID	string	Filter by campaign
agentId	string	Filter by agent
search	string	Regex search across name, phone, originalPhone
direction	"inbound" | "outbound" | "all"	Call direction (outbound includes legacy null values)
minDuration	number (seconds)	Minimum billing duration
maxDuration	number (seconds)	Maximum billing duration
bookmarked	"true" | "false"	Filter by bookmark status
analysis.*	string	Dynamic analysis field filters (e.g., analysis.sentiment=positive). Boolean strings are coerced. Multiple analysis filters use AND logic.
page	number	Page number (default: 1)
limit	number	Results per page (default: 50, max: 500)

Analysis Filters are dynamic they depend on the campaign's agent analysis schema. The frontend fetches the schema via GET /contacts/analysis-schema/:campaignUID and renders appropriate filter controls (dropdowns for enums, toggles for booleans).

AnalysisFilters component (app/src/components/shared/AnalysisFilters.tsx): Renders dynamic filter UI based on the agent's analysis schema. Only visible when a campaign is selected.

Agent (Agent/)

File	Purpose
agent.py	LiveKit worker entry point
src/call_handler.py	SIP call + recording trigger via aiohttp
src/services/post_call.py	Transcription/analysis sync

---

Environment Variables

client_backend

Variable	Example	Purpose
PORT	3000	Express server port
MONGO_URI	mongodb+srv://...	MongoDB Atlas (M30, 50 conn pool)
REDIS_URL	redis://elasticache:6379	Redis (ElastiCache Multi-AZ)
LIVEKIT_URL	wss://voice.oktivo.com:7880	LiveKit WebSocket
LIVEKIT_API_KEY	LiveKitProdKey	LiveKit API key
LIVEKIT_API_SECRET	...	LiveKit API secret
AWS_S3_BUCKET	oktivo-recordings	S3 bucket for recordings
AWS_S3_REGION	ap-south-1	S3 region

---

Production Scaling Targets

Component	Count	Purpose
Backend instances	2-5	Behind ALB, BullMQ distributes work
LiveKit servers	2-4	Behind NLB, handle WebRTC rooms
Egress workers	3-5	Auto-scaling, 1.5 CPU per recording
Agent workers	3-5	Python LiveKit workers, one agent per call
Redis	ElastiCache Multi-AZ	Campaign state, locks, active call tracking
MongoDB	Atlas M30	50 connection pool, compound indexes on campaignUID+status

Throughput: 100-200 concurrent calls at steady state, with BullMQ processing up to 50 campaign management jobs/second across all instances.