qipai/backend/src/devices/iot-message-service.ts

import { createHash } from 'node:crypto';
import type { ResultSetHeader, RowDataPacket } from 'mysql2/promise';
import type { MySqlPool } from '../db/mysql.js';
import {
  JilianControlBoxAdapter,
  JilianSmartSocketAdapter,
  JilianSub1GLockAdapter,
  type NormalizedVendorMessage,
  type ProtocolAdapter
} from './jilian-adapters.js';

interface DeviceRow extends RowDataPacket {
  id: string; tenantId: string; storeId: string; roomId: string | null;
  deviceId: string; deviceType: 'CONTROL_BOX' | 'SUB_LOCK' | 'SMART_SOCKET';
}
interface IdRow extends RowDataPacket { id: string }

export class IotMessageService {
  private readonly adapters: Record<DeviceRow['deviceType'], ProtocolAdapter> = {
    CONTROL_BOX: new JilianControlBoxAdapter(),
    SUB_LOCK: new JilianSub1GLockAdapter(),
    SMART_SOCKET: new JilianSmartSocketAdapter()
  };

  constructor(private readonly pool: MySqlPool) {}

  async handle(topic: string, payload: Buffer): Promise<void> {
    const payloadText = payload.toString('utf8');
    const payloadHash = createHash('sha256').update(payload).digest('hex');
    const topicMatch = /^\/(devicesend|devicewill)\/([A-Za-z0-9_-]{1,64})$/.exec(topic);
    if (!topicMatch) {
      await this.deadLetter(null, null, topic, payloadHash, payloadText, 'MQTT_TOPIC_INVALID');
      return;
    }

    const deviceCode = topicMatch[2];
    const [devices] = await this.pool.execute<DeviceRow[]>(
      `SELECT id, tenant_id AS tenantId, store_id AS storeId, room_id AS roomId,
              device_id AS deviceId, device_type AS deviceType
       FROM qipai_devices
       WHERE device_id = ? AND deleted_at IS NULL`,
      [deviceCode]
    );
    const device = devices[0];
    if (!device) {
      await this.deadLetter(null, null, topic, payloadHash, payloadText, 'MQTT_DEVICE_UNKNOWN');
      return;
    }

    let rawPayload: unknown;
    let normalized: NormalizedVendorMessage;
    try {
      rawPayload = JSON.parse(payloadText);
      normalized = topicMatch[1] === 'devicewill'
        ? normalizeWill(rawPayload)
        : this.adapters[device.deviceType].parseUplink(rawPayload);
      if (normalized.deviceId && normalized.deviceId !== device.deviceId) {
        throw new Error('Payload DeviceID does not match MQTT topic.');
      }
    } catch (error) {
      await this.deadLetter(
        device.tenantId, device.id, topic, payloadHash, payloadText,
        'MQTT_PAYLOAD_INVALID', error instanceof Error ? error.message : 'Invalid payload'
      );
      return;
    }

    const [result] = await this.pool.execute<ResultSetHeader>(
      `INSERT INTO qipai_iot_device_events
        (tenant_id, device_id, store_id, room_id, command_id, topic, event_type,
         payload_hash, raw_payload, normalized_payload, event_at, processing_status)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'PROCESSED')
       ON DUPLICATE KEY UPDATE receive_count = receive_count + 1,
         received_at = UTC_TIMESTAMP(3)`,
      [device.tenantId, device.id, device.storeId, device.roomId,
        normalized.commandId, topic, normalized.eventType, payloadHash,
        JSON.stringify(rawPayload), JSON.stringify(normalized),
        normalized.eventAt]
    );
    if (result.affectedRows !== 1) return;

    await this.updateDevice(device, normalized, rawPayload);
    if (normalized.kind === 'ACK' && normalized.commandId) {
      await this.applyAcknowledgement(device, normalized);
    }
  }

  async createCommand(input: {
    tenantId: string; assetId: string; storeId: string; roomId?: string | null;
    orderId?: string | null; commandId: string; commandType: string;
    payload: Record<string, unknown>; traceId: string; expiresAt?: Date | null;
  }) {
    if (!/^\d{1,13}$/.test(input.commandId)) {
      throw new Error('IOT_COMMAND_ID_INVALID');
    }
    const [result] = await this.pool.execute<ResultSetHeader>(
      `INSERT INTO qipai_iot_commands
        (tenant_id, device_id, store_id, room_id, order_id, command_id, command_type,
         request_payload, trace_id, expires_at)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
      [input.tenantId, input.assetId, input.storeId, input.roomId ?? null,
        input.orderId ?? null, input.commandId, input.commandType,
        JSON.stringify(input.payload), input.traceId, input.expiresAt ?? null]
    );
    return { recordId: String(result.insertId), commandId: input.commandId };
  }

  async markPublished(tenantId: string, commandId: string) {
    const [result] = await this.pool.execute<ResultSetHeader>(
      `UPDATE qipai_iot_commands SET status = 'PUBLISHED', published_at = UTC_TIMESTAMP(3)
       WHERE tenant_id = ? AND command_id = ? AND status = 'PENDING'
         AND (expires_at IS NULL OR expires_at > UTC_TIMESTAMP(3))`,
      [tenantId, commandId]
    );
    return result.affectedRows === 1;
  }

  async markTimedOut(tenantId: string, commandId: string) {
    const [result] = await this.pool.execute<ResultSetHeader>(
      `UPDATE qipai_iot_commands SET status = 'TIMEOUT', failure_code = 'ACK_TIMEOUT'
       WHERE tenant_id = ? AND command_id = ? AND status = 'PUBLISHED'`,
      [tenantId, commandId]
    );
    return result.affectedRows === 1;
  }

  async markPublishFailed(tenantId: string, commandId: string, failureCode: string) {
    const [result] = await this.pool.execute<ResultSetHeader>(
      `UPDATE qipai_iot_commands SET status = 'FAILED', failure_code = ?
       WHERE tenant_id = ? AND command_id = ? AND status = 'PENDING'`,
      [failureCode.slice(0, 64), tenantId, commandId]
    );
    return result.affectedRows === 1;
  }

  private async applyAcknowledgement(device: DeviceRow, message: NormalizedVendorMessage) {
    const successful = message.result === 'ok';
    const status = successful ? 'ACKED' : 'FAILED';
    const failureCode = successful ? '' : `DEVICE_${(message.result ?? 'unknown').toUpperCase()}`;
    await this.pool.execute(
      `UPDATE qipai_iot_commands SET status = ?, response_payload = ?,
        acknowledged_at = UTC_TIMESTAMP(3), failure_code = ?
       WHERE tenant_id = ? AND device_id = ? AND command_id = ?
         AND status IN ('PENDING', 'PUBLISHED', 'TIMEOUT')`,
      [status, JSON.stringify(message.payload), failureCode,
        device.tenantId, device.id, message.commandId]
    );
  }

  private async updateDevice(
    device: DeviceRow, message: NormalizedVendorMessage, rawPayload: unknown
  ) {
    const offline = message.eventType === 'will';
    await this.pool.execute(
      `UPDATE qipai_devices SET status = ?, state_snapshot = ?,
        last_seen_at = UTC_TIMESTAMP(3),
        last_heartbeat_at = CASE WHEN ? = 0 THEN UTC_TIMESTAMP(3) ELSE last_heartbeat_at END
       WHERE tenant_id = ? AND id = ?`,
      [offline ? 'OFFLINE' : 'ONLINE', JSON.stringify(rawPayload), offline ? 1 : 0,
        device.tenantId, device.id]
    );
  }

  private async deadLetter(
    tenantId: string | null, deviceId: string | null, topic: string,
    payloadHash: string, rawPayload: string, code: string, message = code
  ) {
    await this.pool.execute(
      `INSERT INTO qipai_iot_dead_letters
        (tenant_id, device_id, topic, payload_hash, raw_payload, error_code, error_message)
       VALUES (?, ?, ?, ?, ?, ?, ?)
       ON DUPLICATE KEY UPDATE receive_count = receive_count + 1,
         last_received_at = UTC_TIMESTAMP(3), error_code = VALUES(error_code),
         error_message = VALUES(error_message)`,
      [tenantId, deviceId, topic, payloadHash, rawPayload.slice(0, 1_000_000),
        code, message.slice(0, 500)]
    );
  }
}

export function generateCommandId(now = Date.now(), sequence = 0): string {
  const seconds = Math.floor(now / 1000) % 10_000_000_000;
  return `${seconds.toString().padStart(10, '0')}${(sequence % 1000).toString().padStart(3, '0')}`;
}

function normalizeWill(payload: unknown): NormalizedVendorMessage {
  const record = zRecord(payload);
  return {
    kind: 'EVENT',
    commandId: null,
    eventType: 'will',
    result: null,
    deviceId: readDeviceId(record),
    eventAt: null,
    payload: record
  };
}

function zRecord(payload: unknown): Record<string, unknown> {
  if (!payload || typeof payload !== 'object' || Array.isArray(payload)) {
    throw new Error('MQTT payload must be a JSON object.');
  }
  return payload as Record<string, unknown>;
}

function readDeviceId(record: Record<string, unknown>): string | null {
  const value = record.DeviceID ?? record.deviceID;
  return typeof value === 'string' ? value : null;
}