Real-Time Call Detail Record (CDR) Processor

export interface CallDetailRecord {
  // Call Identification
  callId: string;                    // Unique call identifier
  recordId: string;                  // CDR record ID
  recordType: 'START' | 'STOP' | 'INTERIM' | 'EVENT';
  
  // Timestamp Information
  startTime: string;                 // ISO 8601 format
  answerTime?: string;               // When call was answered
  endTime?: string;                  // When call ended
  duration: number;                  // Total call duration in seconds
  billableSeconds: number;           // Rounded duration for billing
  
  // Party Information
  callingParty: {
    number: string;                  // E.164 format (+1234567890)
    numberType: 'MOBILE' | 'LANDLINE' | 'VOIP';
    subscriberId: string;            // Internal subscriber ID
    location: {
      country: string;
      city?: string;
      coordinates?: { lat: number; lon: number };
    };
  };
  
  calledParty: {
    number: string;
    numberType: 'MOBILE' | 'LANDLINE' | 'VOIP' | 'INTERNATIONAL';
    destination: string;             // Country/region
  };
  
  // Call Characteristics
  callType: 'VOICE' | 'VIDEO' | 'CONFERENCE';
  callDirection: 'MOBILE_ORIGINATED' | 'MOBILE_TERMINATED';
  terminationReason: 'NORMAL' | 'BUSY' | 'NO_ANSWER' | 'FAILED' | 'CANCELLED';
  
  // Network Information
  sipData: {
    fromUri: string;                 // SIP From header
    toUri: string;                   // SIP To header
    callIdHeader: string;            // SIP Call-ID
    userAgent: string;               // SIP User-Agent
    codec: string;                   // Audio codec (G.711, Opus, etc.)
    ipAddress: string;               // Caller IP address
  };
  
  // Quality Metrics
  qos?: {
    jitter: number;                  // Milliseconds
    packetLoss: number;              // Percentage
    latency: number;                 // Milliseconds
    mos: number;                     // Mean Opinion Score (1-5)
  };
  
  // Billing Information
  cost: {
    currency: string;                // ISO 4217 (USD, EUR, etc.)
    amount: number;                  // Total call cost
    perMinuteRate: number;          // Rate charged
    taxAmount: number;               // Tax portion
  };
  
  // Additional Metadata
  metadata: {
    serverNode: string;              // Which PBX handled the call
    recordingId?: string;            // If call was recorded
    tags: string[];                  // Custom tags (fraud-alert, premium, etc.)
  };
}

import { KinesisStreamEvent, KinesisStreamRecord } from 'aws-lambda';
import { DynamoDBClient } from '@aws-sdk/client-dynamodb';
import { DynamoDBDocumentClient, UpdateCommand } from '@aws-sdk/lib-dynamodb';
import { CallDetailRecord } from '../types/cdr';

export class CDRAggregator {
  private docClient: DynamoDBDocumentClient;
  private tableName: string;

  constructor() {
    const client = new DynamoDBClient({});
    this.docClient = DynamoDBDocumentClient.from(client);
    this.tableName = process.env.AGGREGATES_TABLE!;
  }

  async processBatch(records: KinesisStreamRecord[]): Promise<void> {
    const aggregates = new Map<string, AggregateMetrics>();

    // Process all records in batch
    for (const record of records) {
      const cdr = this.decodeCDR(record);
      
      if (cdr.recordType === 'STOP' || cdr.recordType === 'INTERIM') {
        this.aggregateCDR(cdr, aggregates);
      }
    }

    // Write aggregates to DynamoDB
    await this.writeAggregates(aggregates);
  }

  private decodeCDR(record: KinesisStreamRecord): CallDetailRecord {
    const payload = Buffer.from(record.kinesis.data, 'base64').toString('utf-8');
    return JSON.parse(payload) as CallDetailRecord;
  }

  private aggregateCDR(
    cdr: CallDetailRecord,
    aggregates: Map<string, AggregateMetrics>
  ): void {
    const subscriberId = cdr.callingParty.subscriberId;
    const dateKey = cdr.startTime.substring(0, 10); // YYYY-MM-DD
    const hourKey = cdr.startTime.substring(11, 13); // HH
    
    const key = `${subscriberId}#${dateKey}#${hourKey}`;
    
    if (!aggregates.has(key)) {
      aggregates.set(key, {
        subscriberId,
        date: dateKey,
        hour: parseInt(hourKey),
        callCount: 0,
        totalDuration: 0,
        totalCost: 0,
        callsByDirection: { MO: 0, MT: 0 },
        callsByType: { VOICE: 0, VIDEO: 0, CONFERENCE: 0 }
      });
    }

    const agg = aggregates.get(key)!;
    agg.callCount++;
    agg.totalDuration += cdr.duration;
    agg.totalCost += cdr.cost.amount;
    
    if (cdr.callDirection === 'MOBILE_ORIGINATED') {
      agg.callsByDirection.MO++;
    } else {
      agg.callsByDirection.MT++;
    }
    
    agg.callsByType[cdr.callType]++;
  }

  private async writeAggregates(
    aggregates: Map<string, AggregateMetrics>
  ): Promise<void> {
    const writePromises = Array.from(aggregates.entries()).map(
      async ([key, metrics]) => {
        const [subscriberId, date, hour] = key.split('#');
        
        await this.docClient.send(new UpdateCommand({
          TableName: this.tableName,
          Key: {
            subscriberId,
            dateHour: `${date}#${hour}`
          },
          UpdateExpression: `
            ADD callCount :count,
                totalDuration :duration,
                totalCost :cost,
                callsByDirection.MO :mo,
                callsByDirection.MT :mt,
                callsByType.VOICE :voice,
                callsByType.VIDEO :video
          `,
          ExpressionAttributeValues: {
            ':count': metrics.callCount,
            ':duration': metrics.totalDuration,
            ':cost': metrics.totalCost,
            ':mo': metrics.callsByDirection.MO,
            ':mt': metrics.callsByDirection.MT,
            ':voice': metrics.callsByType.VOICE || 0,
            ':video': metrics.callsByType.VIDEO || 0
          }
        }));
      }
    );

    await Promise.all(writePromises);
  }
}

interface AggregateMetrics {
  subscriberId: string;
  date: string;
  hour: number;
  callCount: number;
  totalDuration: number;
  totalCost: number;
  callsByDirection: { MO: number; MT: number };
  callsByType: { VOICE: number; VIDEO: number; CONFERENCE: number };
}

export const handler = async (event: KinesisStreamEvent): Promise<void> => {
  const aggregator = new CDRAggregator();
  
  try {
    await aggregator.processBatch(event.Records);
    console.log(`Processed ${event.Records.length} CDR records`);
  } catch (error) {
    console.error('Error processing CDR batch:', error);
    throw error; // Trigger Lambda retry
  }
};