453 lines
14 KiB
TypeScript
453 lines
14 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
|
|
import { CDR_CONSUMER_GROUP, CDR_DEADLETTER_STREAM, CDR_STREAM } from './index.js';
|
|
|
|
export const CDR_EVENT_SCHEMA_VERSION = '1';
|
|
export const CDR_IDEMPOTENCY_KEY_PREFIX = 'lock:cdr:';
|
|
|
|
export type CdrStreamFieldMap = Record<string, string>;
|
|
|
|
export interface CdrRedisCommands {
|
|
xadd(stream: string, id: string, ...fieldValues: string[]): Promise<string>;
|
|
xack(stream: string, group: string, ...ids: string[]): Promise<number>;
|
|
xgroup(...args: string[]): Promise<string>;
|
|
xreadgroup(...args: Array<string | number>): Promise<RedisStreamReadResponse | null>;
|
|
xautoclaim(...args: Array<string | number>): Promise<RedisAutoClaimResponse>;
|
|
set(key: string, value: string, mode: 'NX', expireMode: 'EX', ttlSeconds: number): Promise<'OK' | null>;
|
|
del(...keys: string[]): Promise<number>;
|
|
}
|
|
|
|
export type RedisStreamEntry = [id: string, fields: string[]];
|
|
export type RedisStreamReadResponse = Array<[stream: string, entries: RedisStreamEntry[]]>;
|
|
export type RedisAutoClaimResponse = [nextStartId: string, entries: RedisStreamEntry[], deletedIds?: string[]];
|
|
|
|
export interface CdrStreamEvent {
|
|
schema_version: string;
|
|
event_id: string;
|
|
idempotency_key: string;
|
|
call_id: string;
|
|
node_id: string;
|
|
opensips_instance: string;
|
|
ingress_a_ip: string;
|
|
rtpengine_node: string;
|
|
customer_id: string;
|
|
customer_gateway_id: string;
|
|
customer_gateway_policy_id: string;
|
|
source_ip: string;
|
|
caller: string;
|
|
callee: string;
|
|
raw_callee: string;
|
|
business_prefix_id: string;
|
|
business_prefix: string;
|
|
landing_caller: string;
|
|
landing_callee: string;
|
|
normalized_callee: string;
|
|
callee_city_code: string;
|
|
callee_city_name: string;
|
|
callee_province_name: string;
|
|
callee_operator: string;
|
|
callee_number_type: string;
|
|
vendor_id: string;
|
|
vendor_gateway_id: string;
|
|
line_group_id: string;
|
|
started_at: string;
|
|
answered_at: string;
|
|
ended_at: string;
|
|
duration_sec: string;
|
|
sip_code: string;
|
|
hangup_reason: string;
|
|
recording_key: string;
|
|
config_version: string;
|
|
created_at: string;
|
|
}
|
|
|
|
export interface ParsedCdrStreamEvent extends CdrStreamEvent {
|
|
durationSeconds: number;
|
|
sipCode: number;
|
|
}
|
|
|
|
export interface CdrStreamPublishInput {
|
|
eventId?: string;
|
|
idempotencyKey?: string;
|
|
callId: string;
|
|
nodeId?: string;
|
|
opensipsInstance?: string;
|
|
ingressAIp?: string;
|
|
rtpengineNode?: string;
|
|
customerId?: string;
|
|
customerGatewayId?: string;
|
|
customerGatewayPolicyId?: string;
|
|
sourceIp: string;
|
|
caller?: string;
|
|
callee?: string;
|
|
rawCallee?: string;
|
|
businessPrefixId?: string;
|
|
businessPrefix?: string;
|
|
landingCaller?: string;
|
|
landingCallee?: string;
|
|
normalizedCallee?: string;
|
|
calleeCityCode?: string;
|
|
calleeCityName?: string;
|
|
calleeProvinceName?: string;
|
|
calleeOperator?: string;
|
|
calleeNumberType?: string;
|
|
vendorId?: string;
|
|
vendorGatewayId?: string;
|
|
lineGroupId?: string;
|
|
startedAt?: string;
|
|
answeredAt?: string;
|
|
endedAt?: string;
|
|
durationSec?: number;
|
|
sipCode: number;
|
|
hangupReason: string;
|
|
recordingKey?: string;
|
|
configVersion?: string;
|
|
createdAt?: string;
|
|
}
|
|
|
|
export type CdrHandlerResult = 'processed' | 'duplicate';
|
|
export type CdrStreamHandler = (event: ParsedCdrStreamEvent, redisId: string) => Promise<CdrHandlerResult>;
|
|
|
|
export class CdrRetryableError extends Error {
|
|
constructor(message: string) {
|
|
super(message);
|
|
this.name = 'CdrRetryableError';
|
|
}
|
|
}
|
|
|
|
export interface ProcessCdrBatchOptions {
|
|
stream?: string;
|
|
group?: string;
|
|
consumer: string;
|
|
count?: number;
|
|
blockMs?: number;
|
|
idempotencyTtlSeconds?: number;
|
|
}
|
|
|
|
export interface ProcessPendingOptions {
|
|
stream?: string;
|
|
group?: string;
|
|
consumer: string;
|
|
minIdleMs?: number;
|
|
startId?: string;
|
|
count?: number;
|
|
idempotencyTtlSeconds?: number;
|
|
}
|
|
|
|
export interface CdrProcessSummary {
|
|
processed: number;
|
|
duplicates: number;
|
|
deadlettered: number;
|
|
pendingLeft: number;
|
|
ackedIds: string[];
|
|
}
|
|
|
|
const requiredFields = [
|
|
'schema_version',
|
|
'event_id',
|
|
'idempotency_key',
|
|
'call_id',
|
|
'node_id',
|
|
'opensips_instance',
|
|
'ingress_a_ip',
|
|
'rtpengine_node',
|
|
'source_ip',
|
|
'sip_code',
|
|
'hangup_reason',
|
|
'created_at'
|
|
] as const;
|
|
|
|
export function buildCdrStreamEvent(input: CdrStreamPublishInput): CdrStreamEvent {
|
|
const eventId = input.eventId ?? randomUUID();
|
|
const endedAt = input.endedAt ?? '';
|
|
|
|
return {
|
|
schema_version: CDR_EVENT_SCHEMA_VERSION,
|
|
event_id: eventId,
|
|
idempotency_key: input.idempotencyKey ?? `${input.callId}:${endedAt || eventId}`,
|
|
call_id: input.callId,
|
|
node_id: input.nodeId ?? 'a1',
|
|
opensips_instance: input.opensipsInstance ?? 'opensips-a1',
|
|
ingress_a_ip: input.ingressAIp ?? '',
|
|
rtpengine_node: input.rtpengineNode ?? input.nodeId ?? 'a1',
|
|
customer_id: input.customerId ?? '',
|
|
customer_gateway_id: input.customerGatewayId ?? '',
|
|
customer_gateway_policy_id: input.customerGatewayPolicyId ?? '',
|
|
source_ip: input.sourceIp,
|
|
caller: input.caller ?? '',
|
|
callee: input.callee ?? '',
|
|
raw_callee: input.rawCallee ?? input.callee ?? '',
|
|
business_prefix_id: input.businessPrefixId ?? '',
|
|
business_prefix: input.businessPrefix ?? '',
|
|
landing_caller: input.landingCaller ?? input.caller ?? '',
|
|
landing_callee: input.landingCallee ?? input.callee ?? '',
|
|
normalized_callee: input.normalizedCallee ?? input.callee ?? '',
|
|
callee_city_code: input.calleeCityCode ?? '',
|
|
callee_city_name: input.calleeCityName ?? '',
|
|
callee_province_name: input.calleeProvinceName ?? '',
|
|
callee_operator: input.calleeOperator ?? 'UNKNOWN',
|
|
callee_number_type: input.calleeNumberType ?? 'UNKNOWN',
|
|
vendor_id: input.vendorId ?? '',
|
|
vendor_gateway_id: input.vendorGatewayId ?? '',
|
|
line_group_id: input.lineGroupId ?? '',
|
|
started_at: input.startedAt ?? '',
|
|
answered_at: input.answeredAt ?? '',
|
|
ended_at: endedAt,
|
|
duration_sec: String(input.durationSec ?? 0),
|
|
sip_code: String(input.sipCode),
|
|
hangup_reason: input.hangupReason,
|
|
recording_key: input.recordingKey ?? '',
|
|
config_version: input.configVersion ?? '',
|
|
created_at: input.createdAt ?? new Date().toISOString()
|
|
};
|
|
}
|
|
|
|
export function serializeCdrEvent(event: CdrStreamEvent): string[] {
|
|
return Object.entries(event).flatMap(([field, value]) => [field, value]);
|
|
}
|
|
|
|
export function parseStreamFields(fields: string[]): CdrStreamFieldMap {
|
|
const parsed: CdrStreamFieldMap = {};
|
|
for (let index = 0; index < fields.length; index += 2) {
|
|
const key = fields[index];
|
|
const value = fields[index + 1];
|
|
if (key !== undefined && value !== undefined) {
|
|
parsed[key] = value;
|
|
}
|
|
}
|
|
return parsed;
|
|
}
|
|
|
|
export function parseCdrStreamEvent(fields: string[]): ParsedCdrStreamEvent {
|
|
const parsed = parseStreamFields(fields);
|
|
for (const field of requiredFields) {
|
|
if (!parsed[field]) {
|
|
throw new Error(`CDR stream event missing required field: ${field}`);
|
|
}
|
|
}
|
|
if (parsed.schema_version !== CDR_EVENT_SCHEMA_VERSION) {
|
|
throw new Error(`Unsupported CDR schema version: ${parsed.schema_version}`);
|
|
}
|
|
|
|
const durationSeconds = Number.parseInt(parsed.duration_sec ?? '0', 10);
|
|
const sipCode = Number.parseInt(parsed.sip_code, 10);
|
|
if (!Number.isInteger(durationSeconds) || durationSeconds < 0) {
|
|
throw new Error(`Invalid CDR duration_sec: ${parsed.duration_sec}`);
|
|
}
|
|
if (!Number.isInteger(sipCode) || sipCode < 100 || sipCode > 699) {
|
|
throw new Error(`Invalid CDR sip_code: ${parsed.sip_code}`);
|
|
}
|
|
|
|
return {
|
|
schema_version: parsed.schema_version,
|
|
event_id: parsed.event_id,
|
|
idempotency_key: parsed.idempotency_key,
|
|
call_id: parsed.call_id,
|
|
node_id: parsed.node_id,
|
|
opensips_instance: parsed.opensips_instance,
|
|
ingress_a_ip: parsed.ingress_a_ip,
|
|
rtpengine_node: parsed.rtpengine_node,
|
|
customer_id: parsed.customer_id ?? '',
|
|
customer_gateway_id: parsed.customer_gateway_id ?? '',
|
|
customer_gateway_policy_id: parsed.customer_gateway_policy_id ?? '',
|
|
source_ip: parsed.source_ip,
|
|
caller: parsed.caller ?? '',
|
|
callee: parsed.callee ?? '',
|
|
raw_callee: parsed.raw_callee ?? '',
|
|
business_prefix_id: parsed.business_prefix_id ?? '',
|
|
business_prefix: parsed.business_prefix ?? '',
|
|
landing_caller: parsed.landing_caller ?? '',
|
|
landing_callee: parsed.landing_callee ?? '',
|
|
normalized_callee: parsed.normalized_callee ?? '',
|
|
callee_city_code: parsed.callee_city_code ?? '',
|
|
callee_city_name: parsed.callee_city_name ?? '',
|
|
callee_province_name: parsed.callee_province_name ?? '',
|
|
callee_operator: parsed.callee_operator ?? 'UNKNOWN',
|
|
callee_number_type: parsed.callee_number_type ?? 'UNKNOWN',
|
|
vendor_id: parsed.vendor_id ?? '',
|
|
vendor_gateway_id: parsed.vendor_gateway_id ?? '',
|
|
line_group_id: parsed.line_group_id ?? '',
|
|
started_at: parsed.started_at ?? '',
|
|
answered_at: parsed.answered_at ?? '',
|
|
ended_at: parsed.ended_at ?? '',
|
|
duration_sec: parsed.duration_sec ?? '0',
|
|
sip_code: parsed.sip_code,
|
|
hangup_reason: parsed.hangup_reason,
|
|
recording_key: parsed.recording_key ?? '',
|
|
config_version: parsed.config_version ?? '',
|
|
created_at: parsed.created_at,
|
|
durationSeconds,
|
|
sipCode
|
|
};
|
|
}
|
|
|
|
export async function ensureCdrConsumerGroup(
|
|
redis: CdrRedisCommands,
|
|
stream = CDR_STREAM,
|
|
group = CDR_CONSUMER_GROUP
|
|
): Promise<void> {
|
|
try {
|
|
await redis.xgroup('CREATE', stream, group, '0', 'MKSTREAM');
|
|
} catch (error) {
|
|
if (!(error instanceof Error) || !error.message.includes('BUSYGROUP')) {
|
|
throw error;
|
|
}
|
|
}
|
|
}
|
|
|
|
export async function publishCdrEvent(
|
|
redis: CdrRedisCommands,
|
|
input: CdrStreamPublishInput,
|
|
stream = CDR_STREAM
|
|
): Promise<string> {
|
|
const event = buildCdrStreamEvent(input);
|
|
return redis.xadd(stream, '*', ...serializeCdrEvent(event));
|
|
}
|
|
|
|
export async function processCdrBatch(
|
|
redis: CdrRedisCommands,
|
|
handler: CdrStreamHandler,
|
|
options: ProcessCdrBatchOptions
|
|
): Promise<CdrProcessSummary> {
|
|
const stream = options.stream ?? CDR_STREAM;
|
|
const group = options.group ?? CDR_CONSUMER_GROUP;
|
|
const count = options.count ?? 10;
|
|
const blockMs = options.blockMs ?? 1000;
|
|
const response = await redis.xreadgroup(
|
|
'GROUP',
|
|
group,
|
|
options.consumer,
|
|
'COUNT',
|
|
count,
|
|
'BLOCK',
|
|
blockMs,
|
|
'STREAMS',
|
|
stream,
|
|
'>'
|
|
);
|
|
return processReadResponse(redis, handler, response, {
|
|
stream,
|
|
group,
|
|
idempotencyTtlSeconds: options.idempotencyTtlSeconds ?? 86400
|
|
});
|
|
}
|
|
|
|
export async function processPendingCdrBatch(
|
|
redis: CdrRedisCommands,
|
|
handler: CdrStreamHandler,
|
|
options: ProcessPendingOptions
|
|
): Promise<CdrProcessSummary> {
|
|
const stream = options.stream ?? CDR_STREAM;
|
|
const group = options.group ?? CDR_CONSUMER_GROUP;
|
|
const response = await redis.xautoclaim(
|
|
stream,
|
|
group,
|
|
options.consumer,
|
|
options.minIdleMs ?? 60000,
|
|
options.startId ?? '0-0',
|
|
'COUNT',
|
|
options.count ?? 10
|
|
);
|
|
return processEntries(redis, handler, response[1], {
|
|
stream,
|
|
group,
|
|
idempotencyTtlSeconds: options.idempotencyTtlSeconds ?? 86400
|
|
});
|
|
}
|
|
|
|
async function processReadResponse(
|
|
redis: CdrRedisCommands,
|
|
handler: CdrStreamHandler,
|
|
response: RedisStreamReadResponse | null,
|
|
options: { stream: string; group: string; idempotencyTtlSeconds: number }
|
|
): Promise<CdrProcessSummary> {
|
|
const entries = response?.flatMap(([, streamEntries]) => streamEntries) ?? [];
|
|
return processEntries(redis, handler, entries, options);
|
|
}
|
|
|
|
async function processEntries(
|
|
redis: CdrRedisCommands,
|
|
handler: CdrStreamHandler,
|
|
entries: RedisStreamEntry[],
|
|
options: { stream: string; group: string; idempotencyTtlSeconds: number }
|
|
): Promise<CdrProcessSummary> {
|
|
const summary: CdrProcessSummary = {
|
|
processed: 0,
|
|
duplicates: 0,
|
|
deadlettered: 0,
|
|
pendingLeft: 0,
|
|
ackedIds: []
|
|
};
|
|
|
|
for (const [redisId, fields] of entries) {
|
|
try {
|
|
const event = parseCdrStreamEvent(fields);
|
|
const lock = await redis.set(
|
|
`${CDR_IDEMPOTENCY_KEY_PREFIX}${event.event_id}`,
|
|
redisId,
|
|
'NX',
|
|
'EX',
|
|
options.idempotencyTtlSeconds
|
|
);
|
|
if (lock === null) {
|
|
await ack(redis, options.stream, options.group, redisId, summary);
|
|
summary.duplicates += 1;
|
|
continue;
|
|
}
|
|
|
|
const result = await handler(event, redisId);
|
|
await ack(redis, options.stream, options.group, redisId, summary);
|
|
if (result === 'duplicate') {
|
|
summary.duplicates += 1;
|
|
} else {
|
|
summary.processed += 1;
|
|
}
|
|
} catch (error) {
|
|
if (error instanceof CdrRetryableError) {
|
|
const fieldMap = parseStreamFields(fields);
|
|
if (fieldMap.event_id) {
|
|
await redis.del(`${CDR_IDEMPOTENCY_KEY_PREFIX}${fieldMap.event_id}`);
|
|
}
|
|
summary.pendingLeft += 1;
|
|
continue;
|
|
}
|
|
await moveToDeadletter(redis, redisId, fields, error);
|
|
await ack(redis, options.stream, options.group, redisId, summary);
|
|
summary.deadlettered += 1;
|
|
}
|
|
}
|
|
|
|
return summary;
|
|
}
|
|
|
|
async function ack(
|
|
redis: CdrRedisCommands,
|
|
stream: string,
|
|
group: string,
|
|
redisId: string,
|
|
summary: CdrProcessSummary
|
|
): Promise<void> {
|
|
await redis.xack(stream, group, redisId);
|
|
summary.ackedIds.push(redisId);
|
|
}
|
|
|
|
async function moveToDeadletter(redis: CdrRedisCommands, redisId: string, fields: string[], error: unknown): Promise<void> {
|
|
const fieldMap = parseStreamFields(fields);
|
|
await redis.xadd(
|
|
CDR_DEADLETTER_STREAM,
|
|
'*',
|
|
'original_redis_id',
|
|
redisId,
|
|
'event_id',
|
|
fieldMap.event_id ?? '',
|
|
'call_id',
|
|
fieldMap.call_id ?? '',
|
|
'error',
|
|
error instanceof Error ? error.message : 'Unknown CDR processing error',
|
|
'payload',
|
|
JSON.stringify(fieldMap),
|
|
'deadlettered_at',
|
|
new Date().toISOString()
|
|
);
|
|
}
|