IoT Edge Computing & Kiến Trúc Fog

10.000 sensors × 1 reading/giây × 200 bytes = 2 MB/giây = 172 GB/ngày

// Edge data filter — chạy trên gateway device
interface SensorReading {
  deviceId: string;
  timestamp: number;
  temperature: number;
  humidity: number;
  vibration: number;
}
 
class EdgeDataFilter {
  private lastSent: Map<string, SensorReading> = new Map();
 
  // Chỉ forward readings thay đổi đáng kể
  shouldForward(reading: SensorReading): boolean {
    const last = this.lastSent.get(reading.deviceId);
 
    if (!last) {
      this.lastSent.set(reading.deviceId, reading);
      return true; // Luôn forward reading đầu tiên
    }
 
    const tempDelta = Math.abs(reading.temperature - last.temperature);
    const humidDelta = Math.abs(reading.humidity - last.humidity);
    const vibDelta = Math.abs(reading.vibration - last.vibration);
 
    // Forward chỉ khi thay đổi vượt ngưỡng
    if (tempDelta > 0.5 || humidDelta > 2.0 || vibDelta > 10.0) {
      this.lastSent.set(reading.deviceId, reading);
      return true;
    }
 
    return false; // Bỏ qua — không đủ đáng kể
  }
}
 
// Sử dụng trên edge gateway
const filter = new EdgeDataFilter();
const readings = getIncomingSensorReadings(); // Từ MQTT subscription
 
for (const reading of readings) {
  if (filter.shouldForward(reading)) {
    publishToCloud(reading); // Forward lên cloud MQTT broker
  }
  // Âm thầm bỏ qua readings không đáng kể
}

interface AggregatedReading {
  deviceId: string;
  windowStart: number;
  windowEnd: number;
  temperature: {
    min: number;
    max: number;
    avg: number;
    count: number;
  };
  humidity: {
    min: number;
    max: number;
    avg: number;
    count: number;
  };
}
 
class EdgeAggregator {
  private windows: Map<string, SensorReading[]> = new Map();
  private windowDuration = 60_000; // Cửa sổ 1 phút
 
  addReading(reading: SensorReading): void {
    const key = reading.deviceId;
    if (!this.windows.has(key)) {
      this.windows.set(key, []);
    }
    this.windows.get(key)!.push(reading);
  }
 
  flush(deviceId: string): AggregatedReading | null {
    const readings = this.windows.get(deviceId);
    if (!readings || readings.length === 0) return null;
 
    const temps = readings.map(r => r.temperature);
    const humids = readings.map(r => r.humidity);
 
    const result: AggregatedReading = {
      deviceId,
      windowStart: readings[0].timestamp,
      windowEnd: readings[readings.length - 1].timestamp,
      temperature: {
        min: Math.min(...temps),
        max: Math.max(...temps),
        avg: temps.reduce((a, b) => a + b, 0) / temps.length,
        count: temps.length,
      },
      humidity: {
        min: Math.min(...humids),
        max: Math.max(...humids),
        avg: humids.reduce((a, b) => a + b, 0) / humids.length,
        count: humids.length,
      },
    };
 
    this.windows.set(deviceId, []); // Reset cửa sổ
    return result;
  }
}
 
// Mỗi 60 giây, flush aggregates lên cloud
setInterval(() => {
  for (const deviceId of aggregator.getAllDeviceIds()) {
    const summary = aggregator.flush(deviceId);
    if (summary) {
      publishToCloud(summary); // 1 message thay vì 60
    }
  }
}, 60_000);

interface Alert {
  deviceId: string;
  type: 'warning' | 'critical';
  metric: string;
  value: number;
  threshold: number;
  timestamp: number;
  message: string;
}
 
interface ThresholdRule {
  metric: string;
  warningThreshold: number;
  criticalThreshold: number;
  direction: 'above' | 'below';
}
 
class EdgeAlertEngine {
  private rules: ThresholdRule[] = [
    { metric: 'temperature', warningThreshold: 35, criticalThreshold: 45, direction: 'above' },
    { metric: 'temperature', warningThreshold: 5, criticalThreshold: 0, direction: 'below' },
    { metric: 'vibration', warningThreshold: 50, criticalThreshold: 80, direction: 'above' },
    { metric: 'humidity', warningThreshold: 80, criticalThreshold: 95, direction: 'above' },
  ];
 
  // Debounce: không spam alerts
  private lastAlert: Map<string, number> = new Map();
  private cooldownMs = 30_000; // 30 giây cooldown mỗi device+metric
 
  evaluate(reading: SensorReading): Alert[] {
    const alerts: Alert[] = [];
 
    for (const rule of this.rules) {
      const value = reading[rule.metric as keyof SensorReading] as number;
      if (typeof value !== 'number') continue;
 
      const key = `${reading.deviceId}:${rule.metric}:${rule.direction}`;
      const lastTime = this.lastAlert.get(key) || 0;
 
      if (Date.now() - lastTime < this.cooldownMs) continue;
 
      let triggered = false;
      let type: 'warning' | 'critical' = 'warning';
      let threshold = rule.warningThreshold;
 
      if (rule.direction === 'above') {
        if (value >= rule.criticalThreshold) {
          triggered = true;
          type = 'critical';
          threshold = rule.criticalThreshold;
        } else if (value >= rule.warningThreshold) {
          triggered = true;
          type = 'warning';
        }
      } else {
        if (value <= rule.criticalThreshold) {
          triggered = true;
          type = 'critical';
          threshold = rule.criticalThreshold;
        } else if (value <= rule.warningThreshold) {
          triggered = true;
          type = 'warning';
        }
      }
 
      if (triggered) {
        this.lastAlert.set(key, Date.now());
        alerts.push({
          deviceId: reading.deviceId,
          type,
          metric: rule.metric,
          value,
          threshold,
          timestamp: Date.now(),
          message: `${rule.metric} ${rule.direction} ${threshold}: ${value}`,
        });
      }
    }
 
    return alerts;
  }
}
 
// Trên edge gateway
const alertEngine = new EdgeAlertEngine();
 
function processReading(reading: SensorReading) {
  const alerts = alertEngine.evaluate(reading);
 
  for (const alert of alerts) {
    // Phản ứng cục bộ — không cần cloud
    if (alert.type === 'critical') {
      triggerLocalBuzzer();
      activateEmergencyShutdown(alert.deviceId);
    }
 
    // Cũng forward alert lên cloud để logging
    publishToCloud(alert);
  }
}

import * as fs from 'fs';
import * as path from 'path';
 
interface QueuedMessage {
  topic: string;
  payload: string;
  timestamp: number;
  retries: number;
}
 
class StoreAndForward {
  private queue: QueuedMessage[] = [];
  private maxQueueSize = 10_000;
  private persistPath: string;
  private isOnline = true;
 
  constructor(persistDir: string) {
    this.persistPath = path.join(persistDir, 'message-queue.json');
    this.loadFromDisk(); // Khôi phục sau khi restart
  }
 
  async send(topic: string, payload: string): Promise<void> {
    if (this.isOnline) {
      try {
        await this.publishToCloud(topic, payload);
        return; // Gửi thành công
      } catch {
        this.isOnline = false;
        console.log('Mạng sập — chuyển sang store-and-forward');
      }
    }
 
    // Lưu cục bộ
    this.enqueue({ topic, payload, timestamp: Date.now(), retries: 0 });
  }
 
  private enqueue(msg: QueuedMessage): void {
    if (this.queue.length >= this.maxQueueSize) {
      // Bỏ messages cũ nhất không quan trọng (FIFO eviction)
      this.queue.shift();
    }
    this.queue.push(msg);
    this.saveToDisk(); // Sống sót qua device restarts
  }
 
  async flush(): Promise<void> {
    if (this.queue.length === 0) return;
 
    console.log(`Flushing ${this.queue.length} queued messages...`);
    const failed: QueuedMessage[] = [];
 
    for (const msg of this.queue) {
      try {
        await this.publishToCloud(msg.topic, msg.payload);
      } catch {
        if (msg.retries < 3) {
          msg.retries++;
          failed.push(msg);
        } else {
          console.log(`Bỏ message sau 3 lần retry: ${msg.topic}`);
        }
      }
    }
 
    this.queue = failed;
    this.saveToDisk();
 
    if (failed.length === 0) {
      this.isOnline = true;
      console.log('Tất cả queued messages đã flush thành công');
    }
  }
 
  private saveToDisk(): void {
    fs.writeFileSync(this.persistPath, JSON.stringify(this.queue));
  }
 
  private loadFromDisk(): void {
    try {
      const data = fs.readFileSync(this.persistPath, 'utf-8');
      this.queue = JSON.parse(data);
      console.log(`Đã load ${this.queue.length} messages từ disk`);
    } catch {
      this.queue = [];
    }
  }
 
  private async publishToCloud(topic: string, payload: string): Promise<void> {
    // Actual MQTT/HTTP publish lên cloud broker
    throw new Error('Not implemented — dùng MQTT client của bạn ở đây');
  }
}
 
// Sử dụng
const saf = new StoreAndForward('/var/edge/data');
 
// Hoạt động bình thường — gửi hoặc queue tự động
await saf.send('factory/zone-1/temperature', JSON.stringify(reading));
 
// Kiểm tra kết nối định kỳ và flush
setInterval(async () => {
  if (await checkConnectivity()) {
    await saf.flush();
  }
}, 30_000); // Kiểm tra mỗi 30 giây

import mqtt from 'mqtt';
 
interface ZoneData {
  zone: string;
  devices: Map<string, SensorReading[]>;
  lastAggregation: number;
}
 
class FogNode {
  private zones: Map<string, ZoneData> = new Map();
  private edgeBroker: mqtt.MqttClient;
  private cloudBroker: mqtt.MqttClient;
 
  constructor(edgeBrokerUrl: string, cloudBrokerUrl: string) {
    // Kết nối MQTT broker cục bộ (edge devices publish tại đây)
    this.edgeBroker = mqtt.connect(edgeBrokerUrl);
    // Kết nối cloud MQTT broker (fog publishes aggregates lên đây)
    this.cloudBroker = mqtt.connect(cloudBrokerUrl);
 
    this.setupSubscriptions();
    this.startAggregationLoop();
  }
 
  private setupSubscriptions(): void {
    // Subscribe tất cả zone data từ edge devices
    this.edgeBroker.subscribe('factory/+/+/data');
 
    this.edgeBroker.on('message', (topic, message) => {
      // topic: factory/{zone}/{device}/data
      const parts = topic.split('/');
      const zone = parts[1];
      const deviceId = parts[2];
      const reading: SensorReading = JSON.parse(message.toString());
 
      this.ingestReading(zone, deviceId, reading);
    });
  }
 
  private ingestReading(zone: string, deviceId: string, reading: SensorReading): void {
    if (!this.zones.has(zone)) {
      this.zones.set(zone, {
        zone,
        devices: new Map(),
        lastAggregation: Date.now(),
      });
    }
 
    const zoneData = this.zones.get(zone)!;
    if (!zoneData.devices.has(deviceId)) {
      zoneData.devices.set(deviceId, []);
    }
    zoneData.devices.get(deviceId)!.push(reading);
 
    // Chạy cross-device anomaly detection
    this.detectCrossDeviceAnomalies(zone);
  }
 
  private detectCrossDeviceAnomalies(zone: string): void {
    const zoneData = this.zones.get(zone);
    if (!zoneData) return;
 
    // Ví dụ: tương quan nhiệt độ và rung giữa các thiết bị
    const latestReadings: SensorReading[] = [];
    for (const [, readings] of zoneData.devices) {
      if (readings.length > 0) {
        latestReadings.push(readings[readings.length - 1]);
      }
    }
 
    const avgTemp = latestReadings.reduce((sum, r) => sum + r.temperature, 0)
                    / latestReadings.length;
    const maxVibration = Math.max(...latestReadings.map(r => r.vibration));
 
    // Alert cấp zone: nhiệt độ cao + rung cao = có thể hỏng thiết bị
    if (avgTemp > 40 && maxVibration > 60) {
      this.cloudBroker.publish(
        `alerts/factory/${zone}/anomaly`,
        JSON.stringify({
          type: 'cross_device_anomaly',
          zone,
          avgTemp,
          maxVibration,
          timestamp: Date.now(),
          message: `Zone ${zone}: nhiệt độ cao (${avgTemp.toFixed(1)}°C) + rung (${maxVibration})`,
        })
      );
    }
  }
 
  private startAggregationLoop(): void {
    // Mỗi 5 phút, aggregate và gửi lên cloud
    setInterval(() => {
      for (const [zone, zoneData] of this.zones) {
        const summary = this.aggregateZone(zone, zoneData);
 
        this.cloudBroker.publish(
          `factory/${zone}/summary`,
          JSON.stringify(summary)
        );
 
        // Reset device readings sau aggregation
        for (const [deviceId] of zoneData.devices) {
          zoneData.devices.set(deviceId, []);
        }
        zoneData.lastAggregation = Date.now();
      }
    }, 5 * 60_000);
  }
 
  private aggregateZone(zone: string, data: ZoneData) {
    const deviceSummaries: Record<string, any> = {};
 
    for (const [deviceId, readings] of data.devices) {
      if (readings.length === 0) continue;
 
      const temps = readings.map(r => r.temperature);
      deviceSummaries[deviceId] = {
        readingCount: readings.length,
        temperature: {
          min: Math.min(...temps),
          max: Math.max(...temps),
          avg: temps.reduce((a, b) => a + b, 0) / temps.length,
        },
      };
    }
 
    return {
      zone,
      windowStart: data.lastAggregation,
      windowEnd: Date.now(),
      deviceCount: data.devices.size,
      devices: deviceSummaries,
    };
  }
}
 
// Khởi động fog node
const fog = new FogNode(
  'mqtt://local-broker:1883',   // Edge devices publish tại đây
  'mqtt://cloud-broker:8883'    // Fog publishes aggregates lên đây
);

# greengrass-deployment.yaml
# Greengrass V2 component deployment
RecipeFormatVersion: "2020-01-25"
ComponentName: com.factory.temperature-filter
ComponentVersion: "1.0.0"
Manifests:
  - Platform:
      os: linux
    Lifecycle:
      run: |
        python3 {artifacts:path}/filter.py
    Artifacts:
      - URI: s3://my-bucket/components/filter.py
ComponentDependencies:
  aws.greengrass.Nucleus:
    VersionRequirement: ">=2.0.0"

# filter.py — Greengrass V2 component
import json
import awsiot.greengrasscoreipc as ipc
 
ipc_client = ipc.connect()
 
def on_message(event):
    message = json.loads(event.message.payload.decode())
    temperature = message.get('temperature', 0)
 
    # Chỉ forward readings đáng kể lên cloud
    if temperature > 35 or temperature < 5:
        ipc_client.publish_to_iot_core(
            topic_name=f"factory/alerts/{message['deviceId']}",
            qos='AT_LEAST_ONCE',
            payload=json.dumps({
                'alert': 'temperature_anomaly',
                'value': temperature,
                'deviceId': message['deviceId'],
            })
        )
 
# Subscribe topic MQTT cục bộ
ipc_client.subscribe_to_topic(
    topic='factory/+/temperature',
    on_stream_event=on_message,
)

// deployment.json — Azure IoT Edge deployment manifest
{
  "modulesContent": {
    "$edgeAgent": {
      "properties.desired": {
        "modules": {
          "temperatureFilter": {
            "type": "docker",
            "settings": {
              "image": "myregistry.azurecr.io/temp-filter:1.0",
              "createOptions": "{\"HostConfig\":{\"Privileged\":true}}"
            },
            "status": "running",
            "restartPolicy": "always"
          }
        }
      }
    },
    "$edgeHub": {
      "properties.desired": {
        "routes": {
          "filterToCloud": "FROM /messages/modules/temperatureFilter/outputs/alerts INTO $upstream",
          "sensorToFilter": "FROM /messages/modules/sensor/* INTO BrokeredEndpoint(\"/modules/temperatureFilter/inputs/data\")"
        }
      }
    }
  }
}

# Cài K3s trên edge device (một lệnh duy nhất)
curl -sfL https://get.k3s.io | sh -
 
# Kiểm tra đang chạy
sudo k3s kubectl get nodes
# NAME          STATUS   ROLES                  AGE   VERSION
# edge-gw-01   Ready    control-plane,master   30s   v1.28.4+k3s1

# edge-processor.yaml — Deploy edge processing dưới dạng Kubernetes pod
apiVersion: apps/v1
kind: Deployment
metadata:
  name: edge-data-processor
  namespace: iot
spec:
  replicas: 1
  selector:
    matchLabels:
      app: edge-processor
  template:
    metadata:
      labels:
        app: edge-processor
    spec:
      containers:
      - name: processor
        image: myregistry/edge-processor:1.0
        resources:
          limits:
            cpu: "500m"
            memory: "256Mi"
          requests:
            cpu: "100m"
            memory: "128Mi"
        env:
        - name: MQTT_BROKER
          value: "mqtt://mosquitto:1883"
        - name: CLOUD_BROKER
          value: "mqtts://cloud-broker:8883"
        volumeMounts:
        - name: edge-data
          mountPath: /var/edge/data
      volumes:
      - name: edge-data
        hostPath:
          path: /var/edge/data
          type: DirectoryOrCreate
---
# Local MQTT broker dưới dạng Kubernetes service
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mosquitto
  namespace: iot
spec:
  replicas: 1
  selector:
    matchLabels:
      app: mosquitto
  template:
    metadata:
      labels:
        app: mosquitto
    spec:
      containers:
      - name: mosquitto
        image: eclipse-mosquitto:2
        ports:
        - containerPort: 1883
        volumeMounts:
        - name: config
          mountPath: /mosquitto/config
      volumes:
      - name: config
        configMap:
          name: mosquitto-config
---
apiVersion: v1
kind: Service
metadata:
  name: mosquitto
  namespace: iot
spec:
  selector:
    app: mosquitto
  ports:
  - port: 1883
    targetPort: 1883

# docker-compose.edge.yml
version: "3.8"
services:
  mosquitto:
    image: eclipse-mosquitto:2
    ports:
      - "1883:1883"
    volumes:
      - ./mosquitto/config:/mosquitto/config
      - mosquitto-data:/mosquitto/data
    restart: unless-stopped
 
  edge-processor:
    image: myregistry/edge-processor:1.0
    environment:
      - MQTT_BROKER=mqtt://mosquitto:1883
      - CLOUD_BROKER=mqtts://cloud.example.com:8883
      - STORE_PATH=/data/queue
    volumes:
      - edge-data:/data
    depends_on:
      - mosquitto
    restart: unless-stopped
 
  node-red:
    image: nodered/node-red:latest
    ports:
      - "1880:1880"
    volumes:
      - nodered-data:/data
    restart: unless-stopped
 
volumes:
  mosquitto-data:
  edge-data:
  nodered-data:

# Deploy trên edge gateway
docker compose -f docker-compose.edge.yml up -d
 
# Cập nhật một service
docker compose -f docker-compose.edge.yml pull edge-processor
docker compose -f docker-compose.edge.yml up -d edge-processor
 
# Kiểm tra trạng thái
docker compose -f docker-compose.edge.yml ps

type Priority = 'critical' | 'normal' | 'low';
 
interface SyncConfig {
  critical: { mode: 'stream' };      // Real-time
  normal: { mode: 'batch'; intervalMs: number };  // Mỗi 5 phút
  low: { mode: 'batch'; intervalMs: number };     // Mỗi 30 phút
}
 
class HybridSync {
  private config: SyncConfig = {
    critical: { mode: 'stream' },
    normal: { mode: 'batch', intervalMs: 5 * 60_000 },
    low: { mode: 'batch', intervalMs: 30 * 60_000 },
  };
 
  private normalBatch: QueuedMessage[] = [];
  private lowBatch: QueuedMessage[] = [];
  private cloudClient: mqtt.MqttClient;
 
  constructor(cloudBrokerUrl: string) {
    this.cloudClient = mqtt.connect(cloudBrokerUrl);
 
    // Flush batch định kỳ
    setInterval(() => this.flushBatch('normal'), this.config.normal.intervalMs);
    setInterval(() => this.flushBatch('low'), this.config.low.intervalMs);
  }
 
  async sync(topic: string, payload: string, priority: Priority): Promise<void> {
    const message: QueuedMessage = { topic, payload, timestamp: Date.now(), retries: 0 };
 
    switch (priority) {
      case 'critical':
        // Stream ngay lập tức
        await this.publishNow(message);
        break;
      case 'normal':
        this.normalBatch.push(message);
        break;
      case 'low':
        this.lowBatch.push(message);
        break;
    }
  }
 
  private async publishNow(msg: QueuedMessage): Promise<void> {
    this.cloudClient.publish(msg.topic, msg.payload, { qos: 1 });
  }
 
  private flushBatch(priority: 'normal' | 'low'): void {
    const batch = priority === 'normal' ? this.normalBatch : this.lowBatch;
    if (batch.length === 0) return;
 
    // Gửi dưới dạng batch nén duy nhất
    const batchPayload = JSON.stringify({
      batchId: crypto.randomUUID(),
      priority,
      count: batch.length,
      messages: batch,
    });
 
    this.cloudClient.publish(`batch/${priority}`, batchPayload, { qos: 1 });
 
    // Xóa batch
    if (priority === 'normal') {
      this.normalBatch = [];
    } else {
      this.lowBatch = [];
    }
  }
}
 
// Sử dụng
const sync = new HybridSync('mqtts://cloud-broker:8883');
 
// Alert quan trọng — stream ngay lập tức
await sync.sync('alerts/factory/zone-1', alertPayload, 'critical');
 
// Telemetry bình thường — batch mỗi 5 phút
await sync.sync('telemetry/zone-1', telemetryPayload, 'normal');
 
// Environmental logs — batch mỗi 30 phút
await sync.sync('logs/zone-1', logPayload, 'low');

# edge_inference.py — Chạy trên Raspberry Pi hoặc Jetson Nano
import numpy as np
import json
import paho.mqtt.client as mqtt
 
# TensorFlow Lite — tối ưu cho edge devices
import tflite_runtime.interpreter as tflite
 
class EdgeAnomalyDetector:
    def __init__(self, model_path: str):
        # Load model đã quantize (thường 1-10 MB)
        self.interpreter = tflite.Interpreter(model_path=model_path)
        self.interpreter.allocate_tensors()
 
        self.input_details = self.interpreter.get_input_details()
        self.output_details = self.interpreter.get_output_details()
 
        # Sliding window các readings gần đây
        self.window_size = 60  # 60 giây dữ liệu
        self.buffer = []
 
    def add_reading(self, reading: dict) -> dict | None:
        """Thêm reading và trả về dự đoán anomaly nếu window đầy."""
        self.buffer.append([
            reading['temperature'],
            reading['humidity'],
            reading['vibration'],
            reading['pressure'],
        ])
 
        if len(self.buffer) < self.window_size:
            return None  # Chưa đủ dữ liệu
 
        if len(self.buffer) > self.window_size:
            self.buffer.pop(0)  # Sliding window
 
        return self.predict()
 
    def predict(self) -> dict:
        """Chạy inference trên window hiện tại."""
        # Chuẩn bị input tensor
        input_data = np.array([self.buffer], dtype=np.float32)
        self.interpreter.set_tensor(self.input_details[0]['index'], input_data)
 
        # Chạy inference
        self.interpreter.invoke()
 
        # Lấy output
        output = self.interpreter.get_tensor(self.output_details[0]['index'])
        anomaly_score = float(output[0][0])
 
        return {
            'anomaly_score': anomaly_score,
            'is_anomaly': anomaly_score > 0.8,  # Ngưỡng
            'confidence': min(anomaly_score, 1.0 - anomaly_score) * 2,
        }
 
 
# Tích hợp MQTT
detector = EdgeAnomalyDetector('models/anomaly_detector.tflite')
client = mqtt.Client()
 
def on_message(client, userdata, msg):
    reading = json.loads(msg.payload.decode())
    result = detector.add_reading(reading)
 
    if result and result['is_anomaly']:
        # Publish anomaly alert cục bộ và lên cloud
        alert = {
            'type': 'ml_anomaly',
            'deviceId': reading['deviceId'],
            'score': result['anomaly_score'],
            'confidence': result['confidence'],
            'timestamp': reading['timestamp'],
        }
        client.publish('alerts/anomaly', json.dumps(alert))
        print(f"⚠️ Phát hiện bất thường: score={result['anomaly_score']:.3f}")
 
client.on_message = on_message
client.connect('localhost', 1883)
client.subscribe('sensors/+/data')
client.loop_forever()

type ConnectivityState = 'online' | 'degraded' | 'offline';
 
class ConnectivityManager {
  private state: ConnectivityState = 'online';
  private failedPings = 0;
  private maxFailedPings = 3;
  private listeners: ((state: ConnectivityState) => void)[] = [];
 
  async checkConnectivity(): Promise<ConnectivityState> {
    try {
      const start = Date.now();
      await fetch('https://cloud-broker.example.com/health', {
        signal: AbortSignal.timeout(5000),
      });
      const latency = Date.now() - start;
 
      this.failedPings = 0;
 
      if (latency > 2000) {
        this.setState('degraded');
      } else {
        this.setState('online');
      }
    } catch {
      this.failedPings++;
 
      if (this.failedPings >= this.maxFailedPings) {
        this.setState('offline');
      } else {
        this.setState('degraded');
      }
    }
 
    return this.state;
  }
 
  private setState(newState: ConnectivityState): void {
    if (this.state !== newState) {
      console.log(`Kết nối: ${this.state} → ${newState}`);
      this.state = newState;
      this.listeners.forEach(fn => fn(newState));
    }
  }
 
  onStateChange(listener: (state: ConnectivityState) => void): void {
    this.listeners.push(listener);
  }
 
  getState(): ConnectivityState {
    return this.state;
  }
}
 
// Sử dụng
const connectivity = new ConnectivityManager();
 
connectivity.onStateChange((state) => {
  switch (state) {
    case 'online':
      console.log('Kết nối đầy đủ — streaming lên cloud');
      sync.flushAllQueues();
      break;
    case 'degraded':
      console.log('Suy giảm — batch mode, chỉ critical');
      sync.switchToBatchMode();
      break;
    case 'offline':
      console.log('Offline — xử lý cục bộ only');
      sync.pauseCloudSync();
      break;
  }
});
 
// Kiểm tra mỗi 30 giây
setInterval(() => connectivity.checkConnectivity(), 30_000);

interface VersionedData<T> {
  data: T;
  version: number;
  lastModified: number;
  source: 'edge' | 'cloud';
}
 
type ConflictStrategy = 'cloud-wins' | 'edge-wins' | 'latest-wins' | 'merge';
 
class ConflictResolver {
  resolve<T>(
    local: VersionedData<T>,
    remote: VersionedData<T>,
    strategy: ConflictStrategy
  ): VersionedData<T> {
    // Không có conflict — versions khớp
    if (local.version === remote.version) return local;
 
    switch (strategy) {
      case 'cloud-wins':
        return remote;
 
      case 'edge-wins':
        return local;
 
      case 'latest-wins':
        return local.lastModified > remote.lastModified ? local : remote;
 
      case 'merge':
        // Logic merge tùy chỉnh — phụ thuộc vào kiểu dữ liệu
        return this.mergeData(local, remote);
 
      default:
        return remote; // Default an toàn
    }
  }
 
  private mergeData<T>(
    local: VersionedData<T>,
    remote: VersionedData<T>
  ): VersionedData<T> {
    // Cho sensor configs: dùng cloud cho thresholds, edge cho calibration
    const merged = {
      ...remote.data,   // Cloud settings được ưu tiên
      ...local.data,    // Edge overrides (ví dụ: calibration cục bộ)
    };
 
    return {
      data: merged as T,
      version: Math.max(local.version, remote.version) + 1,
      lastModified: Date.now(),
      source: 'edge',
    };
  }
}

❌ 10.000 sensors → 172 GB/ngày raw data → cloud → $$$
 
✅ 10.000 sensors → edge filter → 5 GB/ngày summaries → cloud → $

❌ Mạng sập → edge device báo lỗi → mất dữ liệu
 
✅ Mạng sập → store-and-forward → auto-flush khi kết nối lại

❌ Chạy Kubernetes + Kafka + PostgreSQL trên Raspberry Pi
 
✅ Chạy Node.js process đơn giản + SQLite + MQTT trên Raspberry Pi

❌ Edge device chạy root, không TLS, password mặc định, ports mở
 
✅ Least privilege, mTLS, credentials riêng mỗi thiết bị, firewall rules

❌ Push ML model mới → thay thế cái cũ ngay → predictions sai
 
✅ Push ML model mới → chạy shadow cùng cái cũ → validate → swap