Quản lý Thiết bị & Cập nhật OTA trong Hệ thống IoT

{
  "deviceId": "motor-factory-berlin-007",
  "firmwareVersion": "1.2.3",
  "updateInterval": 5000,
  "reportingLevel": "standard",
  "features": {
    "remoteDebug": false,
    "advancedMetrics": false,
    "edgeProcessing": true
  },
  "endpoints": {
    "mqtt": "mqtt.iot.company.com:8883",
    "http": "api.iot.company.com"
  }
}

interface Device {
  // Identity
  deviceId: string;          // Unique ID: "motor-007"
  deviceType: string;        // "temperature-sensor", "gateway"
  serialNumber: string;      // Hardware serial
  manufacturer?: string;
 
  // Credentials
  certificateId?: string;    // For mTLS
  certificateArn?: string;   // AWS ARN, if using AWS
  apiKey?: string;           // Hashed API key
  lastCredentialRotation: Date;
 
  // Location & Groups
  siteId: string;            // "factory-berlin"
  zone?: string;             // "zone-a"
  tags: Record<string, string>; // {"environment": "prod", "model": "v2"}
 
  // State
  status: "provisioning" | "configured" | "active" | "inactive" | "error" | "retired";
  firmwareVersion: string;
  lastHeartbeat: Date;
  uptime?: number;           // seconds
 
  // Configuration
  config: Record<string, any>;
  desiredConfig?: Record<string, any>;
  configVersion: number;
 
  // Metadata
  createdAt: Date;
  updatedAt: Date;
  retiredAt?: Date;
}

// TypeScript queries for device registry
 
interface DeviceRegistry {
  // Basic CRUD
  createDevice(device: Omit<Device, 'createdAt' | 'updatedAt'>): Promise<Device>;
  getDeviceById(deviceId: string): Promise<Device | null>;
  updateDevice(deviceId: string, updates: Partial<Device>): Promise<Device>;
  deleteDevice(deviceId: string): Promise<void>;
 
  // Querying
  listDevicesBySite(siteId: string): Promise<Device[]>;
  listDevicesByType(deviceType: string): Promise<Device[]>;
  findDevicesByTag(key: string, value: string): Promise<Device[]>;
  listInactiveDevices(sinceMinutes: number): Promise<Device[]>;
 
  // Bulk operations
  updateManyDevices(filter: DeviceFilter, updates: Partial<Device>): Promise<number>;
 
  // Caching (for fast lookups)
  getDeviceIdFromSerialNumber(serialNumber: string): Promise<string | null>;
  invalidateDeviceCache(deviceId: string): Promise<void>;
}
 
// PostgreSQL schema
const schema = `
CREATE TABLE devices (
  device_id VARCHAR(255) PRIMARY KEY,
  device_type VARCHAR(100) NOT NULL,
  serial_number VARCHAR(255) UNIQUE NOT NULL,
  manufacturer VARCHAR(255),
 
  site_id VARCHAR(100) NOT NULL,
  zone VARCHAR(100),
  tags JSONB,
 
  status VARCHAR(50) NOT NULL,
  firmware_version VARCHAR(50),
  last_heartbeat TIMESTAMPTZ,
  config JSONB,
  desired_config JSONB,
  config_version INTEGER DEFAULT 0,
 
  created_at TIMESTAMPTZ DEFAULT now(),
  updated_at TIMESTAMPTZ DEFAULT now(),
  retired_at TIMESTAMPTZ,
 
  CONSTRAINT status_valid CHECK (status IN ('provisioning', 'configured', 'active', 'inactive', 'error', 'retired')),
  INDEX idx_site_id (site_id),
  INDEX idx_status (status),
  INDEX idx_last_heartbeat (last_heartbeat),
  INDEX idx_tags (tags)
);
 
CREATE TABLE device_credentials (
  device_id VARCHAR(255) PRIMARY KEY REFERENCES devices(device_id),
  cert_id VARCHAR(255),
  api_key_hash VARCHAR(255),
  last_rotation TIMESTAMPTZ,
  FOREIGN KEY (device_id) REFERENCES devices(device_id) ON DELETE CASCADE
);
`;

// Cache device metadata in Redis for fast lookups
// Key: device:{device_id}
// TTL: 5 minutes (or on-demand invalidation)
 
async function getDeviceWithCache(deviceId: string): Promise<Device> {
  // Try cache first
  const cached = await redis.get(`device:${deviceId}`);
  if (cached) {
    return JSON.parse(cached);
  }
 
  // Query database
  const device = await db.getDeviceById(deviceId);
  if (!device) throw new DeviceNotFoundError();
 
  // Cache for 5 minutes
  await redis.setex(`device:${deviceId}`, 300, JSON.stringify(device));
 
  return device;
}
 
// Invalidate cache when device is updated
async function updateDeviceAndInvalidateCache(deviceId: string, updates: Partial<Device>) {
  const updated = await db.updateDevice(deviceId, updates);
  await redis.del(`device:${deviceId}`); // Force refresh on next read
  return updated;
}

{
  "state": {
    "desired": {
      "color": "red",
      "temperature": 22,
      "power": "on"
    },
    "reported": {
      "color": "red",
      "temperature": 20,
      "power": "off"
    }
  },
  "metadata": {
    "desired": {
      "color": { "timestamp": 1673456789 },
      "temperature": { "timestamp": 1673456700 }
    },
    "reported": {
      "color": { "timestamp": 1673456500 },
      "temperature": { "timestamp": 1673456600 },
      "power": { "timestamp": 1673456400 }
    }
  },
  "version": 42,
  "lastUpdateTime": 1673456789
}

// Device (TypeScript/Node.js)
const mqtt = require('mqtt');
 
const client = mqtt.connect('mqtt://mqtt.iot.company.com:8883', {
  clientId: 'motor-007',
  cert: fs.readFileSync('device-cert.pem'),
  key: fs.readFileSync('device-key.pem'),
  ca: fs.readFileSync('ca-cert.pem'),
  rejectUnauthorized: true,
});
 
// Subscribe to shadow/update để nhận desired state changes
client.subscribe('$aws/things/motor-007/shadow/update/delta', (err) => {
  if (err) console.error('Subscribe failed:', err);
});
 
// Khi cloud update desired state
client.on('message', (topic, message) => {
  const delta = JSON.parse(message.toString());
 
  // Reconcile: apply desired state khác với reported
  if (delta.state.power !== reportedState.power) {
    await setPower(delta.state.power);
    reportedState.power = delta.state.power;
 
    // Report back to cloud
    publishShadowUpdate({
      state: {
        reported: reportedState,
      },
    });
  }
});
 
async function publishShadowUpdate(update: any) {
  client.publish(
    '$aws/things/motor-007/shadow/update',
    JSON.stringify(update),
  );
}

// Cloud (TypeScript/Node.js)
async function setDevicePower(deviceId: string, power: 'on' | 'off') {
  // 1. Update shadow's desired state
  await iotService.updateShadow(deviceId, {
    state: {
      desired: { power },
    },
  });
 
  // 2. Chờ thiết bị report lại
  const maxWaitMs = 30000;
  const startTime = Date.now();
 
  while (Date.now() - startTime < maxWaitMs) {
    const shadow = await iotService.getShadow(deviceId);
 
    if (shadow.state.reported.power === power) {
      console.log(`✅ Device confirmed power=${power}`);
      return { success: true, updated: shadow };
    }
 
    // Device offline hoặc vẫn syncing
    await new Promise((r) => setTimeout(r, 1000));
  }
 
  // Timeout: device không sync
  throw new Error(
    `Device ${deviceId} failed to sync power state within ${maxWaitMs}ms`,
  );
}

// Device firmware update sequence
async function performOtaUpdate(firmwareUrl: string, expectedHash: string) {
  try {
    // 1. Download đến temporary location
    console.log('Downloading firmware...');
    const tempFile = '/tmp/firmware.bin';
    await downloadFile(firmwareUrl, tempFile);
 
    // 2. Verify hash (prevent corruption)
    const actualHash = await calculateHash(tempFile);
    if (actualHash !== expectedHash) {
      throw new Error(`Hash mismatch: ${actualHash} !== ${expectedHash}`);
    }
 
    // 3. Verify signature (prevent tampering)
    const publicKey = fs.readFileSync('/secure/public-key.pem');
    if (!verifySig(tempFile, publicKey)) {
      throw new Error('Firmware signature verification failed');
    }
 
    // 4. Install
    console.log('Installing firmware...');
    await installFirmware(tempFile);
 
    // 5. Reboot
    await reboot();
 
    // Device reboot ở đây
    // On next boot, report success
  } catch (error) {
    console.error('OTA failed:', error);
    // Keep running old firmware
    // Report error to cloud
    await reportOtaFailure(error.message);
  }
}

Device Storage:
[Bootloader] [Partition A: Firmware v1.2.3] [Partition B: Empty]

// Bootloader code (C)
#define PARTITION_A 0x0000
#define PARTITION_B 0x8000
#define BOOT_FLAG_ADDRESS 0xF000
 
typedef struct {
  uint32_t boot_count;
  uint32_t crash_count;
  uint32_t is_stable;
} PartitionState;
 
void bootloader_main() {
  PartitionState* boot_state = (PartitionState*)BOOT_FLAG_ADDRESS;
 
  // Nếu partition B được mark cho boot
  if (boot_state->is_stable == 0xB) {
    // Cho nó một cơ hội
    boot_state->boot_count++;
 
    // Boot partition B
    jump_to_partition(PARTITION_B);
 
    // Nếu quay lại đây, partition B crash während boot
    // Revert sang A
    boot_state->is_stable = 0xA;
  }
 
  // Boot partition A (default)
  jump_to_partition(PARTITION_A);
}
 
// Sau khi boot thành công, firmware gọi hàm này
void confirm_partition() {
  PartitionState* boot_state = (PartitionState*)BOOT_FLAG_ADDRESS;
  boot_state->boot_count = 0; // Reset
  boot_state->is_stable = 1;  // Mark as good
}

// Device firmware (after booting successfully)
async function confirmPartitionAfterBoot() {
  // Chờ 30 giây để ensure system stable
  // (nếu crash/reboot, bootloader rollback)
  await sleep(30000);
 
  // Confirm partition này good
  await markCurrentPartitionAsStable();
 
  // Report to cloud
  await reportOtaSuccess({
    version: FIRMWARE_VERSION,
    partition: 'B',
  });
}

// Cloud: Generate delta
const oldFirmware = fs.readFileSync('firmware-v1.2.3.bin');
const newFirmware = fs.readFileSync('firmware-v1.3.0.bin');
const delta = generateDelta(oldFirmware, newFirmware);
 
console.log(`Old: ${oldFirmware.length} bytes, New: ${newFirmware.length} bytes`);
console.log(`Delta: ${delta.size} bytes (${((delta.size / newFirmware.length) * 100).toFixed(1)}% of full)`);
 
// Device: Download delta và reconstruct
async function applyDeltaUpdate(deltaUrl: string) {
  const currentFirmware = await readCurrentFirmware();
  const delta = await downloadFile(deltaUrl);
 
  // Patch current firmware với delta
  const newFirmware = applyBsdiffPatch(currentFirmware, delta);
 
  // Verify
  if (hashOf(newFirmware) !== expectedNewHash) {
    throw new Error('Patched firmware hash mismatch');
  }
 
  // Flash nó
  await flashFirmware(newFirmware);
}

// Cloud: Deploy theo stage
async function stageFirmwareUpdate(
  deviceFilter: DeviceFilter,
  newFirmwareUrl: string,
) {
  // Stage 1: 5% thiết bị
  const stage1Devices = await selectDevicesForUpdate(deviceFilter, 0.05);
  await pushUpdateToDevices(stage1Devices, newFirmwareUrl);
  await sleep(4 * 60 * 60 * 1000); // Chờ 4 giờ
 
  // Monitor metric
  const stage1Metrics = await getMetricsForDevices(stage1Devices);
  if (stage1Metrics.errorRate > 0.1) {
    // Hơn 10% error rate, abort!
    console.error('Stage 1 failed, aborting rollout');
    return;
  }
 
  // Stage 2: 25%
  const stage2Devices = await selectDevicesForUpdate(deviceFilter, 0.25);
  await pushUpdateToDevices(stage2Devices, newFirmwareUrl);
  await sleep(4 * 60 * 60 * 1000);
 
  // Stage 3: 100%
  const remaining = await selectDevicesForUpdate(deviceFilter, 1.0);
  await pushUpdateToDevices(remaining, newFirmwareUrl);
}

// Device: Heartbeat loop
async function heartbeatLoop() {
  setInterval(async () => {
    const heartbeat = {
      deviceId: DEVICE_ID,
      timestamp: Date.now(),
      uptime: getUptimeSeconds(),
      freeMemory: os.freemem(),
      cpuTemperature: await readCPUTemp(),
      firmwareVersion: FIRMWARE_VERSION,
      networkSignalStrength: getSignalStrength(),
      errorCount: getErrorsSinceLastHeartbeat(),
    };
 
    await publishToMqtt(`devices/${DEVICE_ID}/heartbeat`, heartbeat);
  }, 5 * 60 * 1000); // Mỗi 5 phút
}

// Cloud: Heartbeat listener
const lastHeartbeat = new Map<string, number>();
 
mqttClient.on('message', async (topic, message) => {
  const match = topic.match(/devices\/(.+)\/heartbeat/);
  if (!match) return;
 
  const deviceId = match[1];
  const heartbeat = JSON.parse(message.toString());
 
  // Update last heartbeat timestamp
  lastHeartbeat.set(deviceId, Date.now());
 
  // Kiểm anomaly
  if (heartbeat.cpuTemperature > 80) {
    await alertDeviceOverheating(deviceId, heartbeat.cpuTemperature);
  }
 
  if (heartbeat.errorCount > 100) {
    await alertHighErrorRate(deviceId, heartbeat.errorCount);
  }
 
  // Store metric cho dashboard
  await metrics.record({
    deviceId,
    ...heartbeat,
  });
});
 
// Background job: Check offline device
setInterval(async () => {
  const now = Date.now();
  const timeout = 10 * 60 * 1000; // 10 phút
 
  for (const [deviceId, lastTime] of lastHeartbeat.entries()) {
    if (now - lastTime > timeout) {
      await markDeviceOffline(deviceId);
      await alertDeviceOffline(deviceId);
 
      lastHeartbeat.delete(deviceId);
    }
  }
}, 60 * 1000); // Check mỗi phút

// Device support diagnostic request
mqttClient.subscribe(`devices/${DEVICE_ID}/diagnostics/request`, (err) => {
  if (err) console.error('Subscribe failed:', err);
});
 
mqttClient.on('message', async (topic, message) => {
  if (topic === `devices/${DEVICE_ID}/diagnostics/request`) {
    const req = JSON.parse(message.toString());
 
    const response = await handleDiagnosticRequest(req);
 
    // Gửi response về
    mqttClient.publish(
      `devices/${DEVICE_ID}/diagnostics/response`,
      JSON.stringify(response),
    );
  }
});
 
async function handleDiagnosticRequest(req: DiagnosticRequest) {
  switch (req.type) {
    case 'system-info':
      return {
        uptime: getUptimeSeconds(),
        freeMemory: os.freemem(),
        totalMemory: os.totalmem(),
        cpuTemperature: await readCPUTemp(),
        diskUsage: await getDiskUsage(),
        networkStatus: getNetworkStatus(),
      };
 
    case 'log-tail': // N dòng cuối logs
      return {
        logs: await getTailLogs(req.lines || 50),
      };
 
    case 'config-dump':
      return {
        config: getCurrentConfig(),
        configVersion: CONFIG_VERSION,
      };
 
    case 'ping':
      return { pong: true, timestamp: Date.now() };
 
    case 'reboot':
      // Nguy hiểm! Cần xác nhận
      if (req.confirmationToken === await getRebootToken()) {
        await scheduleReboot(req.delaySeconds || 10);
        return { scheduled: true };
      }
      return { error: 'Invalid confirmation token' };
 
    default:
      return { error: `Unknown request type: ${req.type}` };
  }
}

// Cloud: Request diagnostics
async function getDiagnosticsForDevice(deviceId: string) {
  const response = await sendDiagnosticRequest(deviceId, {
    type: 'system-info',
  });
 
  console.log(`Device ${deviceId}:`);
  console.log(`  Uptime: ${response.uptime} seconds`);
  console.log(`  Memory: ${response.freeMemory} / ${response.totalMemory} bytes`);
  console.log(`  CPU Temp: ${response.cpuTemperature}°C`);
}
 
async function sendDiagnosticRequest(deviceId: string, req: DiagnosticRequest) {
  const requestId = uuid();
 
  // Gửi request
  await mqttClient.publish(
    `devices/${deviceId}/diagnostics/request`,
    JSON.stringify({ ...req, requestId }),
  );
 
  // Chờ response
  return new Promise((resolve, reject) => {
    const timeout = setTimeout(() => {
      reject(new Error(`Diagnostic request timeout for device ${deviceId}`));
    }, 30000);
 
    mqttClient.on('message', (topic, message) => {
      if (topic === `devices/${deviceId}/diagnostics/response`) {
        const response = JSON.parse(message.toString());
        if (response.requestId === requestId) {
          clearTimeout(timeout);
          resolve(response);
        }
      }
    });
  });
}

// Tag device để dễ targeting
await updateDevice('motor-007', {
  tags: {
    environment: 'production',
    location: 'Berlin',
    model: 'v2-heavy',
    firmwareVersion: '1.2.3',
    'maintenance-due': '2026-04-01',
  },
});
 
// Query device theo tag
const berlinMotors = await findDevicesByTag('location', 'Berlin');
const outOfDate = await findDevicesByTag('maintenance-due', '<2026-03-12');

// Define batch job
interface FleetJob {
  id: string;
  type: 'firmware-update' | 'config-sync' | 'restart' | 'diagnostic';
  deviceFilter: {
    tags?: Record<string, string>;
    statuses?: string[];
    regions?: string[];
  };
  payload: any;
  schedule: {
    startTime: Date;
    maxParallel: number;
    retryPolicy: { maxAttempts: number; backoffMultiplier: number };
  };
  status: 'pending' | 'running' | 'completed' | 'failed';
  progress: {
    total: number;
    succeeded: number;
    failed: number;
    inProgress: number;
  };
}
 
// Tạo và execute job
async function createFleetJob(job: FleetJob) {
  // Tìm tất cả matching device
  const devices = await findDevicesByFilter(job.deviceFilter);
  job.progress.total = devices.length;
 
  // Persist job
  await db.saveFleetJob(job);
 
  // Execute in background
  executeFleetJobAsync(job, devices);
}
 
async function executeFleetJobAsync(job: FleetJob, devices: Device[]) {
  const jobId = job.id;
  const queue = new PQueue({ concurrency: job.schedule.maxParallel });
 
  for (const device of devices) {
    queue.add(async () => {
      try {
        await executeJobOnDevice(jobId, job, device);
 
        // Update progress
        await db.updateFleetJobProgress(jobId, { succeeded: 1 });
      } catch (error) {
        console.error(`Job ${jobId} failed on device ${device.deviceId}:`, error);
        await db.updateFleetJobProgress(jobId, { failed: 1 });
 
        // Record failure để retry
        await db.recordFleetJobFailure(jobId, device.deviceId, error);
      }
    });
  }
 
  await queue.onIdle();
 
  // Tất cả xong
  await db.updateFleetJob(jobId, { status: 'completed' });
}