Initial LisgloSIPS V2 implementation
This commit is contained in:
hectorzhao
@@ -0,0 +1,408 @@
|
||||
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;
|
||||
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;
|
||||
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 ?? '',
|
||||
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 ?? '',
|
||||
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()
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user