Files
lisglosips/apps/worker-config-publisher/src/publisher.ts
T

580 lines
19 KiB
TypeScript

import crypto from 'node:crypto';
import type { PrismaClient } from '@prisma/client';
import {
CONFIG_ACTIVE_VERSION_KEY,
CONFIG_PREVIOUS_VERSION_KEY,
configVersionManifestKey,
configVersionPrefix,
type RedisClient
} from '@lisglosips/redis';
export interface PublishConfigResult {
published: boolean;
version?: string;
previousVersion?: string | null;
outboxIds: string[];
manifest?: ConfigManifest;
}
export interface ConfigManifest {
schemaVersion: number;
version: string;
generatedAt: string;
customerCount: number;
businessPrefixCount: number;
gatewayCount: number;
customerGatewayIpCount: number;
customerGatewayBusinessPrefixCount: number;
customerGatewayCallerPrefixCount: number;
policyCount: number;
vendorGatewayCount: number;
callerRewriteCount: number;
lineGroupCount: number;
cityCount: number;
phoneSegmentCount: number;
areaCodeCount: number;
carrierPrefixRuleCount: number;
blockedRegionCount: number;
checksum: string;
}
type ConfigSnapshot = {
customers: Array<{
id: string;
status: string;
balance: string;
creditLimit: string;
minBalance: string;
}>;
businessPrefixes: Array<{
id: string;
prefix: string;
name: string;
priority: number;
status: string;
}>;
gateways: Array<{
id: string;
customerId: string;
authMode: string;
sourceIp: string | null;
sourceIps: string[];
sipUsername: string | null;
sipDomain: string | null;
sipHa1: string | null;
lineGroupId: string | null;
billingCycleSec: number;
cycleRate: string;
callerMatchMode: string;
callerPrefixes: string[];
calleeMatchMode: string;
businessPrefixes: Array<{ id: string; prefix: string; name: string; priority: number; status: string }>;
status: string;
}>;
policies: Array<{
id: string;
customerId: string;
gatewayId: string;
lineGroupId: string;
priority: number;
callerMode: string;
callerValue: string | null;
calleeMode: string;
calleeValue: string | null;
status: string;
}>;
vendorGateways: Array<{
id: string;
vendorId: string;
authMode: string;
host: string;
port: number;
transport: string;
sipUsername: string | null;
sipHa1: string | null;
cpsLimit: number;
concurrencyLimit: number;
billingCycleSec: number;
cycleRate: string;
landingCalleePrefix: string | null;
status: string;
forbiddenPeriods: Array<{ weekdayMask: number; startTime: string; endTime: string }>;
codecs: Array<{ codec: string; priority: number }>;
prefixRules: Array<{ direction: string; matchPrefix: string; replacePrefix: string; priority: number }>;
callerRewritePool: Array<{ caller: string; weight: number; status: string }>;
blockedRegions: Array<{
regionScope: string;
provinceCode: string | null;
provinceName: string | null;
cityCode: string | null;
cityName: string | null;
}>;
}>;
lineGroups: Array<{
id: string;
name: string;
status: string;
items: Array<{
id: string;
vendorGatewayId: string;
priority: number;
weight: number;
concurrencyCap: number;
status: string;
}>;
}>;
cities: Array<{
code: string;
provinceCode: string;
provinceName: string;
cityCode: string;
cityName: string;
cityLevel: string;
status: string;
}>;
phoneSegments: Array<{
segment7: string;
cityCode: string;
provinceCode: string;
provinceName: string;
cityName: string;
carrier: string;
numberType: 'MOBILE';
}>;
areaCodes: Array<{
areaCode: string;
cityCode: string;
provinceCode: string;
provinceName: string;
cityName: string;
numberType: 'LANDLINE';
}>;
carrierPrefixRules: Array<{
prefix: string;
carrier: string;
priority: number;
}>;
};
export async function publishPendingConfig(prisma: PrismaClient, redis: RedisClient, now = new Date()): Promise<PublishConfigResult> {
const pending = await prisma.outboxEvent.findMany({
where: {
aggregateType: { in: ['business_prefix_config', 'customer_gateway_config', 'vendor_gateway_config', 'line_group_config', 'number_library_config'] },
status: 'PENDING',
availableAt: { lte: now }
},
orderBy: [{ createdAt: 'asc' }],
take: 100
});
if (pending.length === 0) {
return { published: false, outboxIds: [] };
}
const lockedAt = now;
const locked = await prisma.outboxEvent.updateMany({
where: {
id: { in: pending.map((event) => event.id) },
status: 'PENDING'
},
data: {
status: 'PROCESSING',
lockedAt,
attempts: { increment: 1 }
}
});
if (locked.count === 0) {
return { published: false, outboxIds: [] };
}
const processingIds = pending.map((event) => event.id);
try {
const snapshot = await loadSnapshot(prisma);
const version = `${now.getTime()}`;
const previousVersion = await redis.get(CONFIG_ACTIVE_VERSION_KEY);
const manifest = await writeSnapshot(redis, version, snapshot, previousVersion, now);
await prisma.outboxEvent.updateMany({
where: { id: { in: processingIds }, status: 'PROCESSING' },
data: {
status: 'PUBLISHED',
processedAt: new Date(),
lastError: null
}
});
return {
published: true,
version,
previousVersion,
outboxIds: processingIds,
manifest
};
} catch (error) {
await prisma.outboxEvent.updateMany({
where: { id: { in: processingIds }, status: 'PROCESSING' },
data: {
status: 'FAILED',
lastError: error instanceof Error ? error.message.slice(0, 1000) : 'Unknown publish error'
}
});
throw error;
}
}
export async function rollbackActiveConfig(redis: RedisClient): Promise<{ rolledBack: boolean; activeVersion: string | null; previousVersion: string | null }> {
const activeVersion = await redis.get(CONFIG_ACTIVE_VERSION_KEY);
const previousVersion = await redis.get(CONFIG_PREVIOUS_VERSION_KEY);
if (!previousVersion) {
return { rolledBack: false, activeVersion, previousVersion };
}
await redis
.multi()
.set(CONFIG_ACTIVE_VERSION_KEY, previousVersion)
.set(CONFIG_PREVIOUS_VERSION_KEY, activeVersion ?? '')
.exec();
return { rolledBack: true, activeVersion: previousVersion, previousVersion: activeVersion };
}
async function loadSnapshot(prisma: PrismaClient): Promise<ConfigSnapshot> {
const [customers, businessPrefixes, gateways, policies, vendorGateways, lineGroups, cities, phoneSegments, areaCodes, carrierPrefixRules] = await prisma.$transaction([
prisma.customer.findMany({
where: { deletedAt: null },
orderBy: [{ id: 'asc' }]
}),
prisma.businessPrefix.findMany({
where: { deletedAt: null, status: 'ENABLED' },
orderBy: [{ priority: 'asc' }, { prefix: 'asc' }]
}),
prisma.customerGateway.findMany({
where: { deletedAt: null },
orderBy: [{ id: 'asc' }],
include: {
ips: { where: { deletedAt: null, status: 'ENABLED' }, orderBy: [{ createdAt: 'asc' }] },
callerPrefixes: { orderBy: [{ priority: 'asc' }, { prefix: 'asc' }] },
businessPrefixes: {
orderBy: [{ createdAt: 'asc' }],
include: {
businessPrefix: {
select: { id: true, prefix: true, name: true, priority: true, status: true }
}
}
}
}
}),
prisma.customerGatewayPolicy.findMany({
where: { deletedAt: null },
orderBy: [{ gatewayId: 'asc' }, { priority: 'asc' }]
}),
prisma.vendorGateway.findMany({
where: { deletedAt: null },
orderBy: [{ id: 'asc' }],
include: {
forbiddenPeriods: { orderBy: [{ weekdayMask: 'asc' }, { startTime: 'asc' }] },
codecs: { orderBy: [{ priority: 'asc' }] },
prefixRules: { orderBy: [{ direction: 'asc' }, { priority: 'asc' }] },
callerRewritePool: { where: { deletedAt: null, status: 'ENABLED' }, orderBy: [{ weight: 'desc' }, { caller: 'asc' }] },
blockedRegions: { orderBy: [{ regionScope: 'asc' }, { provinceCode: 'asc' }, { cityCode: 'asc' }] }
}
}),
prisma.landingLineGroup.findMany({
where: { deletedAt: null },
orderBy: [{ id: 'asc' }],
include: {
items: { orderBy: [{ priority: 'asc' }] }
}
}),
prisma.geoCity.findMany({
where: { deletedAt: null, status: 'ENABLED' },
orderBy: [{ provinceCode: 'asc' }, { cityCode: 'asc' }]
}),
prisma.phoneNumberSegment.findMany({
where: { deletedAt: null },
orderBy: [{ segment7: 'asc' }],
include: { city: { select: { provinceCode: true } } }
}),
prisma.phoneAreaCode.findMany({
where: { deletedAt: null },
orderBy: [{ areaCode: 'asc' }],
include: { city: { select: { provinceCode: true } } }
}),
prisma.carrierPrefixRule.findMany({
where: { deletedAt: null },
orderBy: [{ prefix: 'desc' }, { priority: 'asc' }]
})
]);
return {
customers: customers.map((customer) => ({
id: customer.id,
status: customer.status,
balance: customer.balance.toFixed(6),
creditLimit: customer.creditLimit.toFixed(6),
minBalance: customer.minBalance.toFixed(6)
})),
businessPrefixes: businessPrefixes.map((prefixItem) => ({
id: prefixItem.id,
prefix: prefixItem.prefix,
name: prefixItem.name,
priority: prefixItem.priority,
status: prefixItem.status
})),
gateways: gateways.map((gateway) => ({
id: gateway.id,
customerId: gateway.customerId,
authMode: gateway.authMode,
sourceIp: gateway.sourceIp,
sourceIps: gateway.ips.length ? gateway.ips.map((ip) => ip.sourceIp) : gateway.sourceIp ? [gateway.sourceIp] : [],
sipUsername: gateway.sipUsername,
sipDomain: gateway.sipDomain,
sipHa1: gateway.sipHa1,
lineGroupId: gateway.lineGroupId,
billingCycleSec: gateway.billingCycleSec,
cycleRate: gateway.cycleRate.toFixed(6),
callerMatchMode: gateway.callerMatchMode,
callerPrefixes: gateway.callerPrefixes.map((item) => item.prefix),
calleeMatchMode: gateway.calleeMatchMode,
businessPrefixes: gateway.businessPrefixes.map((item) => ({
id: item.businessPrefix.id,
prefix: item.businessPrefix.prefix,
name: item.businessPrefix.name,
priority: item.businessPrefix.priority,
status: item.businessPrefix.status
})),
status: gateway.status
})),
policies: policies.map((policy) => ({
id: policy.id,
customerId: policy.customerId,
gatewayId: policy.gatewayId,
lineGroupId: policy.lineGroupId,
priority: policy.priority,
callerMode: policy.callerMode,
callerValue: policy.callerValue,
calleeMode: policy.calleeMode,
calleeValue: policy.calleeValue,
status: policy.status
})),
vendorGateways: vendorGateways.map((gateway) => ({
id: gateway.id,
vendorId: gateway.vendorId,
authMode: gateway.authMode,
host: gateway.host,
port: gateway.port,
transport: gateway.transport,
sipUsername: gateway.sipUsername,
sipHa1: gateway.sipHa1,
cpsLimit: gateway.cpsLimit,
concurrencyLimit: gateway.concurrencyLimit,
billingCycleSec: gateway.billingCycleSec,
cycleRate: gateway.cycleRate.toFixed(6),
landingCalleePrefix: gateway.landingCalleePrefix,
status: gateway.status,
forbiddenPeriods: gateway.forbiddenPeriods.map((period) => ({
weekdayMask: period.weekdayMask,
startTime: period.startTime,
endTime: period.endTime
})),
codecs: gateway.codecs.map((codec) => ({
codec: codec.codec,
priority: codec.priority
})),
prefixRules: gateway.prefixRules.map((rule) => ({
direction: rule.direction,
matchPrefix: rule.matchPrefix,
replacePrefix: rule.replacePrefix,
priority: rule.priority
})),
callerRewritePool: gateway.callerRewritePool.map((caller) => ({
caller: caller.caller,
weight: caller.weight,
status: caller.status
})),
blockedRegions: gateway.blockedRegions.map((region) => ({
regionScope: region.regionScope,
provinceCode: region.provinceCode,
provinceName: region.provinceName,
cityCode: region.cityCode,
cityName: region.cityName
}))
})),
lineGroups: lineGroups.map((group) => ({
id: group.id,
name: group.name,
status: group.status,
items: group.items.map((item) => ({
id: item.id,
vendorGatewayId: item.vendorGatewayId,
priority: item.priority,
weight: item.weight,
concurrencyCap: item.concurrencyCap,
status: item.status
}))
})),
cities: cities.map((city) => ({
code: city.code,
provinceCode: city.provinceCode,
provinceName: city.provinceName,
cityCode: city.cityCode,
cityName: city.cityName,
cityLevel: city.cityLevel,
status: city.status
})),
phoneSegments: phoneSegments.map((segment) => ({
segment7: segment.segment7,
cityCode: segment.cityCode,
provinceCode: segment.city.provinceCode,
provinceName: segment.provinceName,
cityName: segment.cityName,
carrier: segment.carrier,
numberType: 'MOBILE'
})),
areaCodes: areaCodes.map((areaCode) => ({
areaCode: areaCode.areaCode,
cityCode: areaCode.cityCode,
provinceCode: areaCode.city.provinceCode,
provinceName: areaCode.provinceName,
cityName: areaCode.cityName,
numberType: 'LANDLINE'
})),
carrierPrefixRules: carrierPrefixRules.map((rule) => ({
prefix: rule.prefix,
carrier: rule.carrier,
priority: rule.priority
}))
};
}
async function writeSnapshot(
redis: RedisClient,
version: string,
snapshot: ConfigSnapshot,
previousVersion: string | null,
now: Date
): Promise<ConfigManifest> {
const prefix = configVersionPrefix(version);
const checksum = crypto.createHash('sha256').update(stableJson(snapshot)).digest('hex');
const manifest: ConfigManifest = {
schemaVersion: 2,
version,
generatedAt: now.toISOString(),
customerCount: snapshot.customers.length,
businessPrefixCount: snapshot.businessPrefixes.length,
gatewayCount: snapshot.gateways.length,
customerGatewayIpCount: snapshot.gateways.reduce((sum, gateway) => sum + gateway.sourceIps.length, 0),
customerGatewayBusinessPrefixCount: snapshot.gateways.reduce((sum, gateway) => sum + gateway.businessPrefixes.length, 0),
customerGatewayCallerPrefixCount: snapshot.gateways.reduce((sum, gateway) => sum + gateway.callerPrefixes.length, 0),
policyCount: snapshot.policies.length,
vendorGatewayCount: snapshot.vendorGateways.length,
callerRewriteCount: snapshot.vendorGateways.reduce((sum, gateway) => sum + gateway.callerRewritePool.length, 0),
lineGroupCount: snapshot.lineGroups.length,
cityCount: snapshot.cities.length,
phoneSegmentCount: snapshot.phoneSegments.length,
areaCodeCount: snapshot.areaCodes.length,
carrierPrefixRuleCount: snapshot.carrierPrefixRules.length,
blockedRegionCount: snapshot.vendorGateways.reduce((sum, gateway) => sum + gateway.blockedRegions.length, 0),
checksum
};
const multi = redis.multi();
multi.set(configVersionManifestKey(version), JSON.stringify(manifest));
for (const customer of snapshot.customers) {
multi.set(`${prefix}:customer:${customer.id}`, JSON.stringify(customer));
}
for (const businessPrefix of snapshot.businessPrefixes) {
multi.set(`${prefix}:business_prefix:${businessPrefix.id}`, JSON.stringify(businessPrefix));
multi.set(`${prefix}:business_prefix_value:${businessPrefix.prefix}`, businessPrefix.id);
multi.rpush(`${prefix}:business_prefixes`, JSON.stringify(businessPrefix));
}
for (const gateway of snapshot.gateways) {
multi.set(`${prefix}:customer_gateway:${gateway.id}`, JSON.stringify(gateway));
if (gateway.callerMatchMode === 'PREFIXES') {
for (const callerPrefix of gateway.callerPrefixes) {
multi.rpush(`${prefix}:customer_gateway:${gateway.id}:caller_prefixes`, callerPrefix);
}
}
if (gateway.calleeMatchMode === 'BUSINESS_PREFIXES') {
for (const businessPrefix of gateway.businessPrefixes) {
multi.rpush(`${prefix}:customer_gateway:${gateway.id}:business_prefixes`, JSON.stringify(businessPrefix));
}
}
if (gateway.lineGroupId) {
multi.set(`${prefix}:customer_gateway:${gateway.id}:line_group`, gateway.lineGroupId);
}
if (gateway.authMode === 'IP' || gateway.authMode === 'MIXED') {
for (const sourceIp of gateway.sourceIps) {
multi.set(`${prefix}:auth:ip:${sourceIp}`, gateway.id);
multi.rpush(`${prefix}:auth:ip:${sourceIp}:gateways`, gateway.id);
}
}
if ((gateway.authMode === 'SIP_DIGEST' || gateway.authMode === 'MIXED') && gateway.sipUsername && gateway.sipDomain) {
multi.set(`${prefix}:auth:sip:${gateway.sipUsername}@${gateway.sipDomain}`, gateway.id);
}
}
for (const policy of snapshot.policies) {
multi.rpush(`${prefix}:customer_gateway:${policy.gatewayId}:policies`, JSON.stringify(policy));
}
for (const gateway of snapshot.vendorGateways) {
multi.set(`${prefix}:vendor_gateway:${gateway.id}`, JSON.stringify(gateway));
for (const callerRewrite of gateway.callerRewritePool) {
multi.rpush(`${prefix}:vendor_gateway:${gateway.id}:caller_rewrite_pool`, JSON.stringify(callerRewrite));
}
for (const region of gateway.blockedRegions) {
if (region.regionScope === 'CITY' && region.cityCode) {
multi.sadd(`${prefix}:vendor_gateway:${gateway.id}:blocked_city_codes`, region.cityCode);
}
if (region.regionScope === 'PROVINCE' && region.provinceCode) {
multi.sadd(`${prefix}:vendor_gateway:${gateway.id}:blocked_province_codes`, region.provinceCode);
}
}
}
for (const lineGroup of snapshot.lineGroups) {
multi.set(`${prefix}:line_group:${lineGroup.id}`, JSON.stringify(lineGroup));
for (const item of lineGroup.items) {
multi.rpush(`${prefix}:line_group:${lineGroup.id}:items`, JSON.stringify(item));
}
}
for (const city of snapshot.cities) {
multi.set(`${prefix}:geo_city:${city.cityCode}`, JSON.stringify(city));
}
for (const segment of snapshot.phoneSegments) {
multi.set(`${prefix}:phone_segment:${segment.segment7}`, JSON.stringify(segment));
}
for (const areaCode of snapshot.areaCodes) {
multi.set(`${prefix}:area_code:${areaCode.areaCode}`, JSON.stringify(areaCode));
}
for (const rule of snapshot.carrierPrefixRules) {
multi.set(`${prefix}:carrier_prefix:${rule.prefix}`, JSON.stringify(rule));
}
if (previousVersion) {
multi.set(CONFIG_PREVIOUS_VERSION_KEY, previousVersion);
}
multi.set(CONFIG_ACTIVE_VERSION_KEY, version);
await multi.exec();
return manifest;
}
function stableJson(value: unknown): string {
return JSON.stringify(value, Object.keys(flattenKeys(value)).sort());
}
function flattenKeys(value: unknown, keys: Record<string, true> = {}): Record<string, true> {
if (Array.isArray(value)) {
for (const item of value) {
flattenKeys(item, keys);
}
} else if (value && typeof value === 'object') {
for (const [key, nested] of Object.entries(value as Record<string, unknown>)) {
keys[key] = true;
flattenKeys(nested, keys);
}
}
return keys;
}