┌─────────────────────────────────────────────────────────────┐
│                    Kamailio 核心进程                         │
├─────────────┬─────────────┬─────────────┬───────────────────┤
│  信令层      │  传输层      │  逻辑层      │  扩展接口          │
├─────────────┼─────────────┼─────────────┼───────────────────┤
│ tm (事务)   │ websocket   │ registrar   │ KEMI (Lua/Python) │
│ sl (无状态) │ tls         │ auth_db     │ FFI (C/Rust)      │
│ rr (记录路由)│ sctp        │ permissions │ RPC (JSON-RPC)    │
│ maxfwd      │             │ dialog      │                   │
└─────────────┴─────────────┴─────────────┴───────────────────┘

-- kemi/ws_entry.lua
function ksr_request_route()
    if KSR.is_WS() then
        -- WebSocket 升级握手
        if KSR.is_method("GET") and KSR.hdr_is_present("Upgrade") then
            KSR.ws_handle_handshake()
            return 1
        end
        -- 后续 SIP 消息走标准流程
        KSR.setflag(FLT_WS_CLIENT)
    end
    return 1
end

-- kemi/sdp_rewrite.lua
local function inject_ice_candidates(sdp_body)
    local candidates = get_turn_candidates()  -- 从 Redis/配置中心获取
    local lines = {}
    for line in sdp_body:gmatch("[^rn]+") do
        table.insert(lines, line)
        if line:match("^a=mid:") then
            for _, cand in ipairs(candidates) do
                table.insert(lines, "a=" .. cand)
            end
        end
    end
    return table.concat(lines, "rn")
end

function ksr_onreply_route()
    if KSR.is_method("INVITE") and KSR.has_body("application/sdp") then
        local sdp = KSR.pv.get("$rb")
        local new_sdp = inject_ice_candidates(sdp)
        KSR.sdpops.sdp_rewrite(new_sdp)
    end
    return 1
end

-- 确保后续 ACK/BYE 回到同一 WebSocket 连接
KSR.rr.record_route_preset("websocket")
KSR.rr.loose_route()

[WebRTC Client] ←WSS→ [Kamailio] ←UDP/TCP→ [rtpengine] ←SRTP/RTP→ [SIP UA / PSTN GW]
                              ↑
                        [Redis Cluster]
                        (会话状态/ICE候选/转码策略)

-- kemi/media_control.lua
function handle_media_offer()
    local call_id = KSR.pv.get("$ci")
    local media_ip = get_media_node_ip()  -- 基于 dispatcher 选取
    
    local cmd = string.format(
        "offer %s %s %s %s",
        call_id,
        KSR.pv.get("$rb"),           -- SDP body
        media_ip,                    -- 媒体平面 IP
        "replace-origin=1;ice=remove;rtcp-mux=demux"
    )
    
    local reply = KSR.rtpengine.rtpengine_manage(cmd)
    if reply ~= 1 then
        KSR.log("err", "rtpengine offer failed: " .. reply)
        KSR.sl.send_reply(500, "Media Negotiation Failed")
        return -1
    end
    return 1
end

-- kemi/tenant_router.lua
local TENANT_ROUTES = {
    ["corp-a.example.com"] = { media_pool = "pool-a", billing = "plan_enterprise" },
    ["corp-b.example.com"] = { media_pool = "pool-b", billing = "plan_standard" },
}

function route_by_tenant()
    local domain = KSR.pv.get("$rd")
    local cfg = TENANT_ROUTES[domain] or TENANT_ROUTES["default"]
    
    KSR.pv.sets("$var(media_pool)", cfg.media_pool)
    KSR.pv.sets("$var(billing_plan)", cfg.billing)
    
    -- 分发到对应媒体节点组
    KSR.dispatcher.ds_select_dst(cfg.media_pool, 0)
    return 1
end

function push_cdr_event(event_type)
    local cdr = {
        call_id = KSR.pv.get("$ci"),
        from = KSR.pv.get("$fu"),
        to = KSR.pv.get("$tu"),
        duration = KSR.pv.get("$var(call_duration)"),
        tenant = KSR.pv.get("$var(billing_plan)"),
        event = event_type,
        ts = os.time()
    }
    -- 推送至 Kafka/Redis Stream，消费端异步入库
    KSR.jsonrpc.exec("kafka.produce", "cdr_topic", cjson.encode(cdr))
end

{
  "timestamp": "2025-01-15T10:30:45.123Z",
  "level": "INFO",
  "service": "kamailio-gateway",
  "trace_id": "a1b2c3d4-e5f6-7890",
  "span_id": "s1s2s3s4",
  "call_id": "Call-ID@domain",
  "event": "INVITE_PROCESSED",
  "tenant": "corp-a",
  "media_node": "rtpe-03",
  "latency_ms": 12,
  "sdp_size": 1842
}

-- kemi/tls_manager.lua
local cert_cache = ngx.shared.cert_cache  -- OpenResty/Kong 侧共享内存，或用 Redis

function reload_tls_context()
    local domains = get_tenant_domains()  -- 从 CMDB 获取
    for _, domain in ipairs(domains) do
        local cert, key = fetch_cert_from_vault(domain)  -- HashiCorp Vault / cert-manager
        if cert and key then
            -- 动态更新 Kamailio TLS 上下文（需 5.6+ 支持 tls_reload）
            KSR.tls.reload_context(domain, cert, key)
            cert_cache:set(domain .. ":cert", cert, 86400)
            KSR.log("info", "TLS cert reloaded for " .. domain)
        end
    end
end

-- 定时任务：每 6 小时触发一次，配合 cert-manager 自动续期
KSR.set_timer(21600, "reload_tls_context")

-- 从控制平面拉取最新风控规则（JSON 格式）
function sync_security_rules()
    local rules = http_get("https://sec-api.internal/rules/kamailio")
    for _, rule in ipairs(cjson.decode(rules)) do
        KSR.htable.htable_set("sec_rules", rule.id, cjson.encode(rule))
    end
end

                    ┌──────────────┐
                    │  Global LB   │ (GeoDNS / Anycast)
                    └──────┬───────┘
           ┌──────────────┼──────────────┐
    ┌──────▼──────┐ ┌─────▼─────┐ ┌─────▼─────┐
    │ Region CN1  │ │ Region CN2│ │ Region SG │
    │ (Active)    │ │ (Standby) │ │ (Active)  │
    └──────┬──────┘ └─────┬─────┘ └─────┬─────┘
           │              │             │
    ┌──────▼──────────────▼─────────────▼──────┐
    │         分布式状态存储层                  │
    │  ┌─────────┐ ┌─────────┐ ┌─────────┐    │
    │  │ Redis   │ │ etcd    │ │ Kafka   │    │
    │  │ (Dialog │ │ (Config│ │ (CDR/   │    │
    │  │  State) │ │  Sync)  │ │  Event) │    │
    │  └─────────┘ └─────────┘ └─────────┘    │
    └─────────────────────────────────────────┘

-- kemi/dialog_sync.lua
local dialog_state = require "dialog_state"  -- 封装 Redis 操作

function save_dialog_state()
    local dlg = {
        call_id = KSR.pv.get("$ci"),
        from_tag = KSR.pv.get("$ft"),
        to_tag = KSR.pv.get("$tt"),
        state = KSR.pv.get("$dlg_state"),
        route_set = KSR.pv.get("$dlg_route_set"),
        media_node = KSR.pv.get("$var(media_node)"),
        rtpengine_id = KSR.pv.get("$var(rtpe_id)"),
        updated_at = os.time()
    }
    -- 写入 Redis Cluster，TTL = 通话超时时间 + 缓冲
    dialog_state.set(dlg.call_id .. ":" .. dlg.from_tag, dlg, 7200)
end

function recover_dialog_state()
    -- 重启/扩容时从 Redis 恢复 dialog 记忆，避免 BYE/RE-INVITE 404
    local key = KSR.pv.get("$ci") .. ":" .. KSR.pv.get("$ft")
    local dlg = dialog_state.get(key)
    if dlg then
        KSR.dialog.dlg_manage("restore", dlg)
        KSR.log("info", "Dialog restored from Redis: " .. key)
    end
end

# 连接队列与内存
net.core.somaxconn = 65535
net.core.netdev_max_backlog = 65535
net.ipv4.tcp_max_syn_backlog = 65535
net.ipv4.tcp_mem = 786432 1048576 1572864
net.ipv4.tcp_rmem = 4096 87380 16777216
net.ipv4.tcp_wmem = 4096 65536 16777216

# 端口复用与 TIME_WAIT 优化
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 15
net.ipv4.ip_local_port_range = 10000 65535

# 中断亲和性（RPS/RFS）
net.core.rps_sock_flow_entries = 32768

# 假设 16 核 CPU，网卡 8 队列
for i in {0..7}; do
    echo $((1 << i)) > /sys/class/eth0/queues/rx-$i/rps_cpus
    ethtool -X eth0 equal 8
done

# kamailio-config/configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: kamailio-config
data:
  kamailio.cfg: |
    # 由 CI/CD 渲染注入
    #!KAMAILIO
    {{ .Values.global.cfg_header | indent 4 }}
    {{- range .Values.modules }}
    loadmodule "{{ .name }}"
    {{- if .params }}
    {{- range .params }}
    modparam("{{ $.name }}", "{{ .key }}", "{{ .value }}")
    {{- end }}
    {{- end }}
    {{- end }}
    {{ .Values.routing_logic | indent 4 }}
  routing.lua: |
    {{ .Files.Get "kemi/routing.lua" | indent 4 }}

# 预热新 Pod：注入配置、预热 Lua、建立 TCP 连接池
kubectl set env deployment/kamailio WARMUP=true
sleep 30

# 逐个替换：maxSurge=25%, maxUnavailable=0
kubectl rollout restart deployment/kamailio --pod-running-timeout=5m

# 升级后校验：自动化冒烟测试
kubectl run smoke-test --image=my/sipp -- 
  -sf uac_register.xml -r 100 -m 192.168.1.100

-- kemi/recording.lua
function start_compliance_recording()
    local call_id = KSR.pv.get("$ci")
    local tenant = KSR.pv.get("$var(tenant_id)")
    local bucket = string.format("s3://compliance-recordings/%s/%s/", tenant, os.date("%Y/%m/%d"))
    
    -- rtpengine 支持 S3 直接上传（需 12.0+）
    local cmd = string.format(
        "record %s %s %s",
        call_id,
        bucket .. call_id .. ".wav",
        "format=wav,stereo=yes,metadata=tenant:" .. tenant
    )
    KSR.rtpengine.rtpengine_manage(cmd)
end

[WebRTC Client] ←SRTP→ [rtpengine] ←RTP→ [Media Proxy (gRPC/RTSP)] ←PCM→ [ASR Engine]
                                                      ↑
                                            [Kamailio RPC 指令控制]

-- kemi/ai_routing.lua
function ai_pre_routing()
    -- 仅对 VIP 租户启用
    if KSR.pv.get("$var(tenant_plan)") ~= "vip" then return 1 end
    
    -- 异步调用 AI 意图识别服务（非阻塞）
    local payload = {
        caller = KSR.pv.get("$fu"),
        callee = KSR.pv.get("$tu"),
        history = get_recent_calls(KSR.pv.get("$fu"), 5)
    }
    KSR.jsonrpc.exec_async("ai.intent.predict", cjson.encode(payload), "on_ai_result")
    return 1  -- 继续走默认路由，AI 结果回调中可能重新路由
end

function on_ai_result(result)
    local intent = cjson.decode(result).intent
    if intent == "complaint" then
        -- 转接专席队列
        KSR.dispatcher.ds_select_dst("vip_complaint_pool", 0)
    elseif intent == "sales" then
        KSR.dispatcher.ds_select_dst("sales_pool", 0)
    end
end