fix: close caller analytics SIP integration gaps

apps/api/src/modules/caller-analytics/caller-analytics.service.ts

@@ -90,7 +90,7 @@ export class CallerAnalyticsService {
       dataThrough: h?.payload.dataThrough ?? null, lagMs: h?.payload.lagMs ?? null, durationPrecision: 'milliseconds',
       health: h?.payload ?? null, summary: analyticsRates(sum), summaryScope: '时间及业务筛选；样本/比例/异常筛选仅影响号码列表',
       numbers: filtered.slice(q.skip, q.skip + q.take), total: filtered.length, skip: q.skip, take: q.take,
-      trends: trends.map(normalize), historyNotice: '仅统计新采集链路的数据；旧固定6秒话单不参与应答统计或校准。' };
+      trends: trends.map(r => ({ ...normalize(r), time: Number(r.time) })), historyNotice: '仅统计新采集链路的数据；旧固定6秒话单不参与应答统计或校准。' };
   }
   async detail(raw: Query, user: CurrentUser) {
     const q = parseAnalyticsQuery(raw); if (!q.caller || !q.customerId) throw new BadRequestException('请选择客户和主叫号码');

apps/worker-cdr/src/analytics-main.ts

@@ -46,7 +46,7 @@ async function collect() {
     }
     // Persist only after durable queue writes. Crash before checkpoint replays harmlessly.
     cursor.offset += last + 1;
-    await writeFile(`${checkpoint}.tmp`, JSON.stringify(cursor), { mode: 0o600 }); await rename(`${checkpoint}.tmp`, checkpoint);
+    await writeFile(`${checkpoint}.tmp`, JSON.stringify({ ...cursor, gap }), { mode: 0o600 }); await rename(`${checkpoint}.tmp`, checkpoint);
   } finally { await file.close(); }
 }
 async function entries(entries: Array<[string, string[]]>) {
@@ -73,7 +73,7 @@ async function observeActive() {
   }
 }
 async function main() {
-  try { cursor = JSON.parse(await readFile(checkpoint, 'utf8')); } catch { /* First start reads the retained spool. */ }
+  try { const saved = JSON.parse(await readFile(checkpoint, 'utf8')); cursor = saved; gap = saved.gap === true; } catch { /* First start reads the retained spool. */ }
   await redis.connect(); await db.$connect();
   try { await redis.xgroup('CREATE', ANALYTICS_STREAM, ANALYTICS_GROUP, '0', 'MKSTREAM'); } catch (e) { if (!(e instanceof Error) || !e.message.includes('BUSYGROUP')) throw e; }
   let pendingCursor = '0-0';

infra/server-b/caller-analytics/deploy.sh

@@ -21,13 +21,19 @@ for file in /etc/lisglosips/caller-analytics.env /etc/rsyslog.d/35-caller-analyt
   if test -f "$file"; then cp -a "$file" "$backup/$(basename "$file")"; fi
 done
 changed=0
+was_running=$(systemctl is-active lisglosips@caller-analytics || true)
 rollback() {
   rc=$?
   if test "$changed" = 1; then
     systemctl stop lisglosips@caller-analytics || true
     cp -a "$backup/opensips.cfg" /etc/opensips/opensips.cfg
+    for file in /etc/lisglosips/caller-analytics.env /etc/rsyslog.d/35-caller-analytics.conf; do
+      if test -f "$backup/$(basename "$file")"; then cp -a "$backup/$(basename "$file")" "$file"; else rm -f "$file"; fi
+    done
     ln -sfn "$old" /opt/lisglosips/current
+    systemctl restart rsyslog || true
     systemctl restart opensips lisglosips@api || true
+    if test "$was_running" = active; then systemctl start lisglosips@caller-analytics || true; fi
   fi
   echo "FAILED rc=$rc backup=$backup (additive analytics tables retained)"
   exit "$rc"
@@ -44,6 +50,8 @@ if grep -q 'CRA1|' "$source_cfg"; then
 else
   /usr/bin/node "$new/scripts/instrument-caller-analytics.mjs" "$source_cfg" "$new/opensips-candidate.cfg"
 fi
+python3 "$new/infra/server-b/caller-analytics/verify-hotpath.py" "$new/opensips-candidate.cfg"
+systemctl start lisglosips-redis-hotpath-load.service
 opensips -C -f "$new/opensips-candidate.cfg"
 set -a
 source /etc/lisglosips/secrets/mysql-migrate.env

infra/server-b/caller-analytics/rsyslog.conf

@@ -1,5 +1,5 @@
 # OpenSIPS emits only analytics observations here. No network destination.
-if ($programname == 'opensips' and $msg contains 'CRA1|') then {
+if (($programname == 'opensips' or $syslogtag startswith '/usr/sbin/opensips[') and $msg contains 'CRA1|') then {
   action(type="omfile" file="/var/log/lisglosips/caller-analytics.log"
          fileOwner="syslog" fileGroup="lisglosips" fileCreateMode="0640"
          queue.type="LinkedList" queue.filename="caller-analytics"

infra/server-b/caller-analytics/verify-hotpath.py

@@ -0,0 +1,14 @@
+#!/usr/bin/env python3
+"""Align candidate EVALSHA with the installed Lua bytes actually loaded by the loader."""
+import hashlib
+from pathlib import Path
+import re
+import sys
+config = Path(sys.argv[1])
+script = Path('/etc/opensips/lisglosips_hotpath.lua').read_text()
+sha = hashlib.sha1(script.encode()).hexdigest()
+text = config.read_text()
+matches = re.findall(r'EVALSHA ([0-9a-f]{40}) 1 cfg:active_version', text)
+assert len(matches) == 1, 'Unexpected hotpath layout'
+config.write_text(text.replace('EVALSHA ' + matches[0], 'EVALSHA ' + sha))
+print('Installed hotpath SHA:', sha, 'previous:', matches[0])

tests/api/caller-analytics-real.mjs

@@ -6,7 +6,7 @@ import { writeFile } from 'node:fs/promises';
 import assert from 'node:assert/strict';
 const db = new PrismaClient();
 const fixture = 'cra_20260831';
-const user = await db.user.findFirst({ where:{ status:'ENABLED', roles:{some:{roleId:'ROLE_SUPER_ADMIN'}} } });
+const user = await db.user.findFirst({ where:{ status:'ENABLED', userRoles:{some:{roleId:'ROLE_SUPER_ADMIN'}} } });
 assert(user, 'An existing test administrator is required');
 const token = signAccessToken({sub:user.id,username:user.username,roles:['ROLE_SUPER_ADMIN'],typ:'access'}, {secret:process.env.AUTH_ACCESS_TOKEN_SECRET,issuer:process.env.AUTH_TOKEN_ISSUER||'lisglosips-api',audience:process.env.AUTH_TOKEN_AUDIENCE||'lisglosips-web',ttlSeconds:900});
 const get = async (path, authenticated = true) => {