# Edge 集成指南

本文档说明如何实现和集成 Edge 节点，作为设备的代理服务。

## 概述

Edge 节点是位于 Gateway 和设备之间的代理层，负责：
- 管理设备的 WebSocket 连接
- 转发 Gateway 和设备之间的消息
- 维护设备状态和连接信息
- 处理设备的注册和心跳

## 架构设计

```
Gateway (18081)
    ↓
Edge (Dynamic Port)
    ↓
Devices (Multiple Connections)
```

## 实现步骤

### 1. Edge 服务器实现

```typescript
import WebSocket from 'ws';
import { v4 as uuidv4 } from 'uuid';
import { EventEmitter } from 'events';

interface DeviceConnection {
  id: string;
  ws: WebSocket;
  info: any;
  lastHeartbeat: number;
}

export class EdgeServer extends EventEmitter {
  private gatewayClient: WebSocket | null = null;
  private deviceServer: WebSocket.Server;
  private devices: Map<string, DeviceConnection> = new Map();
  private edgeId: string;
  private gatewayUrl: string;
  
  constructor(edgeId: string, gatewayUrl: string, devicePort: number) {
    super();
    this.edgeId = edgeId;
    this.gatewayUrl = gatewayUrl;
    
    // 创建设备服务器
    this.deviceServer = new WebSocket.Server({ port: devicePort });
    this.setupDeviceServer();
    
    // 连接到 Gateway
    this.connectToGateway();
    
    // 启动心跳检查
    this.startHeartbeatCheck();
  }
  
  private connectToGateway(): void {
    console.log(`Connecting to Gateway at ${this.gatewayUrl}`);
    
    this.gatewayClient = new WebSocket(this.gatewayUrl);
    
    this.gatewayClient.on('open', () => {
      console.log('Connected to Gateway');
      this.registerWithGateway();
    });
    
    this.gatewayClient.on('message', (data: WebSocket.Data) => {
      this.handleGatewayMessage(data.toString());
    });
    
    this.gatewayClient.on('close', () => {
      console.log('Disconnected from Gateway, reconnecting...');
      setTimeout(() => this.connectToGateway(), 5000);
    });
    
    this.gatewayClient.on('error', (error) => {
      console.error('Gateway connection error:', error);
    });
  }
  
  private registerWithGateway(): void {
    const registerMessage = {
      type: 'register',
      clientId: this.edgeId,
      clientType: 'EDGE',
      clientInfo: {
        name: 'Edge Node',
        version: '1.0.0',
        capabilities: ['device-proxy', 'batch-command'],
        metadata: {
          maxDevices: 100,
          location: 'tunnel-exit-1'
        }
      },
      timestamp: new Date().toISOString(),
      version: '1.0'
    };
    
    this.sendToGateway(registerMessage);
    
    // 启动心跳
    this.startGatewayHeartbeat();
  }
  
  private setupDeviceServer(): void {
    this.deviceServer.on('connection', (ws: WebSocket, req) => {
      console.log('New device connection');
      
      // 临时存储连接，等待注册
      const tempId = uuidv4();
      let registered = false;
      
      ws.on('message', (data: WebSocket.Data) => {
        const message = JSON.parse(data.toString());
        
        if (message.type === 'register' && !registered) {
          // 处理设备注册
          this.handleDeviceRegistration(ws, message);
          registered = true;
        } else if (registered) {
          // 处理其他消息
          this.handleDeviceMessage(message);
        }
      });
      
      ws.on('close', () => {
        // 查找并移除设备
        for (const [id, device] of this.devices) {
          if (device.ws === ws) {
            console.log(`Device ${id} disconnected`);
            this.devices.delete(id);
            this.notifyGatewayDeviceStatus(id, 'offline');
            break;
          }
        }
      });
      
      ws.on('error', (error) => {
        console.error('Device connection error:', error);
      });
    });
  }
  
  private handleDeviceRegistration(ws: WebSocket, message: any): void {
    const deviceId = message.clientId;
    
    // 检查是否已存在
    if (this.devices.has(deviceId)) {
      console.log(`Device ${deviceId} already registered, closing old connection`);
      this.devices.get(deviceId)!.ws.close();
    }
    
    // 存储设备连接
    this.devices.set(deviceId, {
      id: deviceId,
      ws: ws,
      info: message.clientInfo,
      lastHeartbeat: Date.now()
    });
    
    console.log(`Device ${deviceId} registered`);
    
    // 通知 Gateway 设备上线
    this.notifyGatewayDeviceStatus(deviceId, 'online');
    
    // 回复注册成功
    ws.send(JSON.stringify({
      type: 'register_response',
      success: true,
      timestamp: new Date().toISOString(),
      version: '1.0'
    }));
  }
}
```

### 2. 消息路由实现

```typescript
class MessageRouter {
  
  // Gateway 消息处理
  private handleGatewayMessage(data: string): void {
    try {
      const message = JSON.parse(data);
      
      switch (message.type) {
        case 'command':
          this.routeCommandToDevice(message);
          break;
          
        case 'program':
          this.routeProgramToDevice(message);
          break;
          
        case 'heartbeat':
          // Gateway 心跳响应
          break;
          
        default:
          console.log(`Unknown message type from Gateway: ${message.type}`);
      }
    } catch (error) {
      console.error('Error handling Gateway message:', error);
    }
  }
  
  // 路由命令到设备
  private routeCommandToDevice(message: any): void {
    // targetClientId 直接就是设备ID
    const deviceId = message.targetClientId;
    
    // 查找设备
    const device = this.devices.get(deviceId);
    if (!device) {
      this.sendErrorToGateway(message.requestRef, 'DEVICE_NOT_FOUND', 
        `Device ${deviceId} not connected`);
      return;
    }
    
    // 修改 targetClientId 为设备ID
    message.targetClientId = deviceId;
    
    // 转发到设备
    device.ws.send(JSON.stringify(message));
    console.log(`Forwarded command to device ${deviceId}`);
  }
  
  // 设备消息处理
  private handleDeviceMessage(message: any): void {
    switch (message.type) {
      case 'command_response':
      case 'progress_update':
      case 'program_response':
        // 转发响应到 Gateway
        this.sendToGateway(message);
        break;
        
      case 'heartbeat':
        // 更新设备心跳时间
        const device = this.devices.get(message.clientId);
        if (device) {
          device.lastHeartbeat = Date.now();
        }
        break;
        
      case 'error':
        // 转发错误到 Gateway
        this.sendToGateway(message);
        break;
        
      default:
        console.log(`Unknown message type from device: ${message.type}`);
    }
  }
  
  // 发送消息到 Gateway
  private sendToGateway(message: any): void {
    if (this.gatewayClient && this.gatewayClient.readyState === WebSocket.OPEN) {
      this.gatewayClient.send(JSON.stringify(message));
    } else {
      console.error('Gateway connection not available');
      // 可以考虑缓存消息，等待重连
    }
  }
  
  // 发送错误到 Gateway
  private sendErrorToGateway(requestRef: string, code: string, message: string): void {
    const errorMessage = {
      type: 'error',
      requestRef,
      code,
      message,
      timestamp: new Date().toISOString(),
      version: '1.0'
    };
    
    this.sendToGateway(errorMessage);
  }
}
```

### 3. 设备状态管理

```typescript
class DeviceManager {
  private devices: Map<string, DeviceConnection> = new Map();
  private deviceStatus: Map<string, DeviceStatus> = new Map();
  
  interface DeviceStatus {
    online: boolean;
    lastSeen: Date;
    metrics: {
      commandsReceived: number;
      commandsSucceeded: number;
      commandsFailed: number;
      averageResponseTime: number;
    };
  }
  
  // 通知 Gateway 设备状态变化
  private notifyGatewayDeviceStatus(deviceId: string, status: 'online' | 'offline'): void {
    const statusMessage = {
      type: 'device_status',
      edgeId: this.edgeId,
      deviceId,
      status,
      timestamp: new Date().toISOString(),
      version: '1.0'
    };
    
    this.sendToGateway(statusMessage);
  }
  
  // 获取所有设备状态
  getDeviceStatuses(): DeviceStatus[] {
    const statuses = [];
    
    for (const [id, device] of this.devices) {
      const status = this.deviceStatus.get(id) || this.createDefaultStatus();
      statuses.push({
        deviceId: id,
        ...status,
        info: device.info
      });
    }
    
    return statuses;
  }
  
  // 更新设备指标
  updateDeviceMetrics(deviceId: string, success: boolean, responseTime: number): void {
    const status = this.deviceStatus.get(deviceId) || this.createDefaultStatus();
    
    status.metrics.commandsReceived++;
    if (success) {
      status.metrics.commandsSucceeded++;
    } else {
      status.metrics.commandsFailed++;
    }
    
    // 计算平均响应时间
    const total = status.metrics.commandsSucceeded;
    status.metrics.averageResponseTime = 
      (status.metrics.averageResponseTime * (total - 1) + responseTime) / total;
    
    this.deviceStatus.set(deviceId, status);
  }
}
```

### 4. 心跳和健康检查

```typescript
class HealthChecker {
  private heartbeatInterval = 30000; // 30秒
  private heartbeatTimeout = 90000;  // 90秒
  
  // Gateway 心跳
  private startGatewayHeartbeat(): void {
    setInterval(() => {
      if (this.gatewayClient && this.gatewayClient.readyState === WebSocket.OPEN) {
        const heartbeat = {
          type: 'heartbeat',
          clientId: this.edgeId,
          timestamp: new Date().toISOString(),
          version: '1.0'
        };
        
        this.gatewayClient.send(JSON.stringify(heartbeat));
      }
    }, this.heartbeatInterval);
  }
  
  // 设备心跳检查
  private startHeartbeatCheck(): void {
    setInterval(() => {
      const now = Date.now();
      
      for (const [id, device] of this.devices) {
        if (now - device.lastHeartbeat > this.heartbeatTimeout) {
          console.log(`Device ${id} heartbeat timeout, removing`);
          device.ws.close();
          this.devices.delete(id);
          this.notifyGatewayDeviceStatus(id, 'offline');
        }
      }
    }, this.heartbeatInterval);
  }
  
  // 健康检查端点
  getHealthStatus(): HealthStatus {
    return {
      status: 'healthy',
      edgeId: this.edgeId,
      gatewayConnected: this.gatewayClient?.readyState === WebSocket.OPEN,
      deviceCount: this.devices.size,
      uptime: process.uptime(),
      memory: process.memoryUsage(),
      timestamp: new Date().toISOString()
    };
  }
}
```

### 5. 批量命令处理

```typescript
class BatchCommandHandler {
  
  // 处理批量命令
  async handleBatchCommand(message: any): Promise<void> {
    const { command } = message;
    // targetClientId 直接就是设备ID
    const targetDeviceId = message.targetClientId;
    
    // 批量命令直接转发给设备
    const device = this.devices.get(targetDeviceId);
    if (!device) {
      this.sendErrorToGateway(message.requestRef, 'DEVICE_NOT_FOUND', 
        `Device ${targetDeviceId} not connected`);
      return;
    }
    
    // 设备会处理批量逻辑并发送进度更新
    device.ws.send(JSON.stringify(message));
    
    console.log(`Forwarded batch command to device ${targetDeviceId}`);
  }
  
  // 聚合批量命令的进度更新
  private aggregateBatchProgress(updates: ProgressUpdate[]): any {
    const total = updates.length;
    const successful = updates.filter(u => u.status === 'COMPLETED').length;
    const failed = updates.filter(u => u.status === 'FAILED').length;
    const inProgress = updates.filter(u => u.status === 'IN_PROGRESS').length;
    
    return {
      total,
      successful,
      failed,
      inProgress,
      progress: (successful + failed) / total * 100
    };
  }
}
```

### 6. 配置和启动

```typescript
// config.ts
export interface EdgeConfig {
  edgeId: string;
  gatewayUrl: string;
  devicePort: number;
  maxDevices: number;
  heartbeatInterval: number;
  logLevel: 'debug' | 'info' | 'warn' | 'error';
}

// index.ts
import { EdgeServer } from './edge-server';
import { EdgeConfig } from './config';

const config: EdgeConfig = {
  edgeId: process.env.EDGE_ID || 'edge-001',
  gatewayUrl: process.env.GATEWAY_URL || 'ws://localhost:18081',
  devicePort: parseInt(process.env.DEVICE_PORT || '18090'),
  maxDevices: 100,
  heartbeatInterval: 30000,
  logLevel: 'info'
};

// 启动 Edge 服务
const edge = new EdgeServer(config.edgeId, config.gatewayUrl, config.devicePort);

// 健康检查 HTTP 端点
import express from 'express';
const app = express();

app.get('/health', (req, res) => {
  res.json(edge.getHealthStatus());
});

app.get('/devices', (req, res) => {
  res.json(edge.getDeviceStatuses());
});

app.listen(18091, () => {
  console.log('Edge HTTP server listening on port 18091');
});

// 优雅关闭
process.on('SIGTERM', () => {
  console.log('Shutting down Edge server...');
  edge.shutdown();
  process.exit(0);
});
```

## 部署建议

### 1. Docker 部署

```dockerfile
FROM node:18-alpine

WORKDIR /app

COPY package*.json ./
RUN npm ci --only=production

COPY . .

EXPOSE 18090 18091

CMD ["node", "dist/index.js"]
```

### 2. 环境变量

```bash
# .env
EDGE_ID=edge-001
GATEWAY_URL=ws://gateway:18081
DEVICE_PORT=18090
LOG_LEVEL=info
NODE_ENV=production
```

### 3. 监控和日志

```typescript
import winston from 'winston';

const logger = winston.createLogger({
  level: config.logLevel,
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.json()
  ),
  transports: [
    new winston.transports.File({ filename: 'edge.log' }),
    new winston.transports.Console()
  ]
});

// 使用结构化日志
logger.info('Device connected', {
  deviceId: device.id,
  edgeId: this.edgeId,
  timestamp: new Date().toISOString()
});
```

## 高可用设计

### 1. 自动重连

```typescript
class ReconnectManager {
  private reconnectDelay = 5000;
  private maxReconnectDelay = 60000;
  private reconnectAttempts = 0;
  
  async reconnectToGateway(): Promise<void> {
    const delay = Math.min(
      this.reconnectDelay * Math.pow(2, this.reconnectAttempts),
      this.maxReconnectDelay
    );
    
    this.reconnectAttempts++;
    
    await new Promise(resolve => setTimeout(resolve, delay));
    
    try {
      await this.connectToGateway();
      this.reconnectAttempts = 0;
    } catch (error) {
      console.error('Reconnection failed:', error);
      this.reconnectToGateway();
    }
  }
}
```

### 2. 消息缓存

```typescript
class MessageBuffer {
  private buffer: any[] = [];
  private maxSize = 1000;
  
  add(message: any): void {
    if (this.buffer.length >= this.maxSize) {
      this.buffer.shift(); // 移除最旧的消息
    }
    this.buffer.push(message);
  }
  
  flush(): any[] {
    const messages = [...this.buffer];
    this.buffer = [];
    return messages;
  }
}
```

## 性能优化

### 1. 连接池

```typescript
class ConnectionPool {
  private pool: WebSocket[] = [];
  private maxConnections = 5;
  
  async getConnection(): Promise<WebSocket> {
    // 实现连接池逻辑
  }
}
```

### 2. 消息批处理

```typescript
class MessageBatcher {
  private batch: any[] = [];
  private batchSize = 100;
  private batchInterval = 100; // ms
  
  add(message: any): void {
    this.batch.push(message);
    
    if (this.batch.length >= this.batchSize) {
      this.flush();
    }
  }
  
  private flush(): void {
    if (this.batch.length > 0) {
      this.sendBatch(this.batch);
      this.batch = [];
    }
  }
}
```

## 故障排查

### 常见问题

1. **设备连接不上**
   - 检查防火墙设置
   - 验证设备端口是否正确
   - 查看 Edge 日志

2. **消息转发失败**
   - 确认设备已注册
   - 检查消息格式
   - 验证 Gateway 连接状态

3. **性能问题**
   - 监控设备数量
   - 检查消息队列长度
   - 优化消息处理逻辑

## 总结

Edge 节点实现的关键点：

1. **双向代理** - 正确处理 Gateway 和设备之间的消息转发
2. **连接管理** - 可靠的连接维护和自动重连
3. **状态跟踪** - 准确的设备状态管理
4. **性能优化** - 批处理、连接池等优化手段
5. **高可用性** - 容错、缓存、重试机制

遵循这些指南，可以构建一个稳定、高效的 Edge 代理服务。