┌─────────────────────────────────────────────────────────────┐
│                     接入与采集层                              │
│  信令网关 | 媒体节点 | 网络设备 | 客户端 SDK | 边缘探针        │
└──────────────────────────┬──────────────────────────────────┘
                           │ gRPC / Kafka / Syslog / SNMP
┌──────────────────────────▼──────────────────────────────────┐
│                     数据处理层                                │
│  流式清洗 (Flink) → 拓扑计算引擎 → 时序/图/日志多模存储        │
└──────────────────────────┬──────────────────────────────────┘
                           │
┌──────────────────────────▼──────────────────────────────────┐
│                     服务编排层                                │
│  拓扑服务 | 诊断服务 | 告警引擎 | API 网关 | 任务调度           │
└──────────────────────────┬──────────────────────────────────┘
                           │
┌──────────────────────────▼──────────────────────────────────┐
│                     应用呈现层                                │
│  拓扑画布 | 会话诊断工作台 | 巡检报表 | 移动端告警推送          │
└─────────────────────────────────────────────────────────────┘

# 伪代码：增量拓扑更新逻辑
def on_signaling_event(event: SignalingEvent):
    # 1. 实体抽取
    session_node = upsert_session(event.session_id, event.timestamp)
    media_nodes = extract_media_nodes(event.candidate_pairs)
    
    # 2. 关系推断（基于 ICE 候选对 + TURN 分配记录）
    for pair in event.candidate_pairs:
        link = infer_media_path(
            local=pair.local, 
            remote=pair.remote, 
            turn_allocation=query_turn(pair.turn_server)
        )
        upsert_link(session_node, link, confidence=calc_confidence(pair))
    
    # 3. 图谱融合：合并物理网络拓扑与逻辑媒体拓扑
    merge_physical_logical_topology(session_node)

Client A ──(UDP:50000)──► TURN ──(UDP:3478)──► SFU-1 ──(UDP:40000)──► Client B
                    ▲                                              │
                    │                                              ▼
               (Relay)                                         (Direct)
                    │                                              ▲
                    └──────────────(TCP:443/TLS)───────────────────┘
                              Fallback Path

# 简化版根因排序
def rank_root_causes(anomaly_path: Path, window: TimeWindow) -> List[RootCause]:
    candidates = []
    for node in anomaly_path.nodes:
        # 节点维度指标
        cpu_sat = query_metric(node, 'cpu_usage', window)
        bw_sat  = query_metric(node, 'bandwidth_util', window)
        err_rate = query_metric(node, 'interface_errors', window)
        
        # 邻居对比
        peer_median = query_peer_median(node, ['cpu_usage', 'bandwidth_util'], window)
        
        score = 0.4 * normalize(cpu_sat - peer_median.cpu) 
              + 0.4 * normalize(bw_sat - peer_median.bw) 
              + 0.2 * normalize(err_rate)
        candidates.append(RootCause(node, score, evidence={...}))
    
    return sorted(candidates, key=lambda x: x.score, reverse=True)[:3]

// 核心节点定义
message TopoNode {
  string node_id = 1;           // 全局唯一 ID：{type}:{region}:{cluster}:{instance}
  NodeType type = 2;            // CLIENT, SFU, MCU, TURN, SBC, ROUTER, SWITCH, GW
  string region = 3;            // 逻辑区域：cn-hangzhou, ap-singapore
  string cluster = 4;           // 归属集群 ID
  map<string, string> labels = 5; // 标签：{role: "primary", version: "v2.3.1", hw: "gpu"}
  google.protobuf.Timestamp create_ts = 6;
  google.protobuf.Timestamp update_ts = 7;
  NodeStatus status = 8;        // ONLINE, DEGRADED, OFFLINE, MAINTENANCE
}

// 核心边定义
message TopoEdge {
  string edge_id = 1;           // hash(src_id, dst_id, protocol, vlan)
  string src_node_id = 2;
  string dst_node_id = 3;
  LinkType link_type = 4;       // MEDIA_UDP, MEDIA_TCP_TLS, SIGNALING_GRPC, PHYSICAL_FIBER, VXLAN_OVERLAY
  LinkLayer layer = 5;          // L3_LOGICAL, L4_TRANSPORT, L1_PHYSICAL
  map<string, string> attrs = 6; // {src_port, dst_port, vlan_id, mtu, fec_enabled, bandwidth_cap_mbps}
  double confidence = 7;        // 0.0~1.0 多源融合置信度
  google.protobuf.Timestamp first_seen = 8;
  google.protobuf.Timestamp last_seen = 9;
  EdgeStatus status = 10;       // ACTIVE, STANDBY, BROKEN, EXPIRED
}

# 媒体质量指标标准化命名
metrics:
  - name: rtc_media_rtt_ms
    type: Gauge
    labels: [session_id, media_node_id, direction, codec, candidate_type] # direction: send/recv
    desc: "端到端 RTT 毫秒数"
  - name: rtc_media_jitter_ms
    type: Gauge
    labels: [session_id, media_node_id, direction]
  - name: rtc_media_packet_loss_ratio
    type: Gauge
    labels: [session_id, media_node_id, direction]
  - name: rtc_media_nack_count_total
    type: Counter
    labels: [session_id, media_node_id, media_type] # audio/video
  - name: rtc_media_pli_count_total
    type: Counter
    labels: [session_id, media_node_id]
  - name: rtc_media_bandwidth_estimate_kbps
    type: Gauge
    labels: [session_id, media_node_id, direction]
  - name: rtc_ice_state
    type: Gauge # 0:new 1:checking 2:connected 3:completed 4:failed 5:disconnected
    labels: [session_id, media_node_id, candidate_pair_id]
  - name: rtc_turn_allocate_bytes_total
    type: Counter
    labels: [turn_server_id, realm, transport] # udp/tcp/tls

{
  "session_context": {
    "session_id": "sess_abc123",
    "tenant_id": "tnt_001",
    "start_time": "2024-01-15T08:30:00.123Z",
    "end_time": "2024-01-15T09:15:22.456Z",
    "participants": [
      {"user_id": "u_001", "client_type": "web_chrome_120", "ip": "203.0.113.10", "isp": "China Telecom"},
      {"user_id": "u_002", "client_type": "mac_native_2.3", "ip": "198.51.100.5", "isp": "China Unicom"}
    ],
    "media_topology_snapshots": [
      {"ts": "08:30:05", "path": "u_001 -> TURN:hz-01 -> SFU:hz-02 -> u_002", "relay_type": "TURN_UDP"},
      {"ts": "08:45:12", "path": "u_001 -> SFU:hz-02 -> u_002", "relay_type": "DIRECT_UDP", "trigger": "ICE_NOMINATION_SUCCEEDED"}
    ],
    "quality_timeline": [
      {"ts": "08:40:00", "mos": 4.2, "rtt": 45, "jitter": 12, "plr": 0.001},
      {"ts": "08:42:00", "mos": 2.1, "rtt": 280, "jitter": 85, "plr": 0.08}
    ],
    "root_cause_analysis": {
      "primary_cause": "SFU:hz-02 CPU saturation",
      "evidence": ["cpu_usage>95% for 5min", "packet_queue_drop>1000/s"],
      "confidence": 0.92
    }
  }
}

# 训练目标：节点级异常得分 + 链路级异常得分
class DyGATAnomalyDetector(nn.Module):
    def __init__(self, feat_dim, hidden_dim, num_layers, heads):
        super().__init__()
        self.temporal_encoder = nn.GRU(feat_dim, hidden_dim, batch_first=True) # 时序编码
        self.gat_layers = nn.ModuleList([
            GATConv(hidden_dim, hidden_dim, heads=heads, concat=False) 
            for _ in range(num_layers)
        ])
        self.node_head = nn.Linear(hidden_dim, 1)  # 节点异常概率
        self.edge_head = nn.Linear(hidden_dim * 2, 1) # 链路异常概率

    def forward(self, snapshots: List[Data]) -> Tuple[Tensor, Tensor]:
        # snapshots: 连续 T 个时刻的图快照 [x, edge_index, edge_attr]
        h_seq = []
        for g in snapshots:
            h = self.temporal_encoder(g.x.unsqueeze(0))[0][:, -1] # [N, hidden]
            for gat in self.gat_layers:
                h = F.elu(gat(h, g.edge_index, g.edge_attr))
            h_seq.append(h)
        h_final = h_seq[-1] # 取最新时刻表征
        
        node_score = torch.sigmoid(self.node_head(h_final))
        # 链路得分：拼接首尾节点表征
        src, dst = snapshots[-1].edge_index
        edge_feat = torch.cat([h_final[src], h_final[dst]], dim=-1)
        edge_score = torch.sigmoid(self.edge_head(edge_feat))
        return node_score, edge_score

graph LR
    A[网络拥塞] --> B[SFU 出口队列丢包]
    B --> C[客户端 NACK 激增]
    C --> D[编码器降码率]
    D --> E[用户感知 MOS 下降]
    F[SFU CPU 高] --> B
    G[上游骨干网故障] --> A
    H[新版本部署] --> F

┌─────────────┐     ┌─────────────┐
│  Region A   │◄───►│  Region B   │  (同城双活，网络延迟 < 2ms)
│  (Primary)  │     │ (Standby)   │
└──────┬──────┘     └──────┬──────┘
       │                   │
       ▼                   ▼
┌─────────────────────────────────────┐
│      Global Meta Store (etcd/raft)  │  拓扑元数据、任务调度状态、模型版本
│         3/5 副本跨 AZ 部署          │
└─────────────────────────────────────┘

// SDK 端自适应采样伪代码
func (c *Collector) adaptiveSample(session *Session) bool {
    // 基础采样率
    baseRate := 0.1 
    
    // 质量触发增强
    if session.CurrentMOS() < 3.0 || session.PacketLoss() > 0.05 {
        return true // 100% 采样
    }
    
    // 关键用户/会议室全采
    if session.UserTier() == "VIP" || session.IsMeetingRoom() {
        return true
    }
    
    // 网络环境异常增强
    if session.NetworkType() == "4G/5G" || session.ISP() != "Enterprise" {
        return rand.Float64() < 0.5 // 50% 采样
    }
    
    return rand.Float64() < baseRate
}

paths:
  /api/v1/topology/subgraph:
    post:
      summary: 查询会话/区域/集群子图
      parameters:
        - name: X-Tenant-ID
          in: header
          required: true
          schema: {type: string}
      requestBody:
        content:
          application/json:
            schema:
              type: object
              properties:
                query_type: {type: string, enum: [by_session, by_region, by_failure_domain]}
                session_ids: {type: array, items: {type: string}}
                region: {type: string}
                depth: {type: integer, default: 2} # 拓扑跳数
                include_metrics: {type: boolean, default: true}
                time_range: {$ref: '#/components/schemas/TimeRange'}
      responses:
        '200':
          description: GraphSON 格式子图
          content:
            application/json:
              schema: {$ref: '#/components/schemas/TopologySubGraph'}

  /api/v1/diagnosis/session/{session_id}/root-cause:
    get:
      summary: 获取会话根因分析报告
      responses:
        '200':
          content:
            application/json:
              schema: {$ref: '#/components/schemas/RootCauseReport'}

  /api/v1/inspection/tasks:
    post:
      summary: 创建周期性巡检任务
      requestBody:
        content:
          application/json:
            schema:
              properties:
                name: {type: string}
                cron: {type: string} # "0 2 * * *"
                scope: {$ref: '#/components/schemas/InspectionScope'}
                ruleset: {type: string} # 规则集版本
                notification: {$ref: '#/components/schemas/NotificationConfig'}

# 插件接口规范
class DiagnosisPlugin(ABC):
    @property
    @abstractmethod
    def plugin_id(self) -> str: ...
    
    @property
    @abstractmethod
    def version(self) -> str: ...
    
    @abstractmethod
    def on_session_start(self, ctx: SessionContext) -> Optional[List[ProbeTask]]: ...
    # 返回主动探测任务（如：发起 iperf3、traceroute、WEBRTC 模拟通话）
    
    @abstractmethod
    def evaluate(self, ctx: SessionContext, metrics: MetricBundle) -> List[Finding]: ...
    # 返回发现项，自动汇聚至诊断报告
    
    @abstractmethod
    def render_panel(self, ctx: SessionContext) -> ReactComponent: ...
    # 返回前端自定义面板组件（JSX/JSON Schema）