feat(M06-F): 接入订单设备自动联动任务

2026-06-24 15:18:40 +08:00
parent e4941c2ffa
commit 8936ad1c6f
4 changed files with 331 additions and 2 deletions
@@ -0,0 +1,187 @@
+import type { RowDataPacket } from 'mysql2/promise';
+import type { AccessProfile } from '../auth/rbac-repository.js';
+import type { MySqlPool } from '../db/mysql.js';
+import type { AsyncTask } from '../tasks/task-repository.js';
+import { DeviceControlError, type DeviceControlService } from './device-control-service.js';
+
+export const orderDeviceEvents = [
+  'ORDER_PAID',
+  'ORDER_STARTED',
+  'ORDER_RENEWED',
+  'ORDER_CANCELLED',
+  'ORDER_ROOM_CHANGED',
+  'ORDER_FINISHED'
+] as const;
+
+type OrderDeviceEvent = (typeof orderDeviceEvents)[number];
+
+interface OrderRow extends RowDataPacket {
+  id: string;
+  tenantId: string;
+  storeId: string;
+  roomId: string;
+  status: string;
+  startAt: Date;
+  endAt: Date;
+}
+
+export class OrderDeviceAutomationError extends Error {
+  constructor(public readonly code: string) { super(code); }
+}
+
+export class OrderDeviceAutomationService {
+  constructor(
+    private readonly pool: MySqlPool,
+    private readonly deviceControl: Pick<DeviceControlService,
+      'startTask' | 'extendTask' | 'cancelTask' | 'switchSmartSocket'>
+  ) {}
+
+  async handleTask(task: AsyncTask) {
+    if (task.taskType !== 'device.command') {
+      throw new OrderDeviceAutomationError('DEVICE_TASK_TYPE_INVALID');
+    }
+    const payload = parseTaskPayload(task.payload);
+    if (payload.tenantId !== task.tenantId) {
+      throw new OrderDeviceAutomationError('DEVICE_TASK_TENANT_MISMATCH');
+    }
+    const order = await this.loadOrder(payload.tenantId, payload.orderId);
+    if (['ORDER_PAID', 'ORDER_STARTED', 'ORDER_RENEWED'].includes(payload.event)
+      && !['PAID', 'RESERVED', 'IN_PROGRESS'].includes(order.status)) {
+      return { orderId: order.id, skipped: true, reason: 'ORDER_STATUS_TERMINAL' };
+    }
+
+    if (payload.event === 'ORDER_PAID' || payload.event === 'ORDER_STARTED') {
+      await this.startOrderDevices(order, payload.traceId);
+      return { orderId: order.id, action: 'STARTED' };
+    }
+    if (payload.event === 'ORDER_RENEWED') {
+      await this.extendOrderDevices(order, payload.traceId, payload.addMinutes);
+      return { orderId: order.id, action: 'EXTENDED' };
+    }
+    if (payload.event === 'ORDER_ROOM_CHANGED') {
+      if (payload.previousRoomId && payload.previousRoomId !== order.roomId) {
+        await this.cancelRoomDevices(order, payload.traceId, payload.previousRoomId);
+      }
+      await this.startOrderDevices(order, payload.traceId);
+      return { orderId: order.id, action: 'ROOM_CHANGED' };
+    }
+
+    await this.cancelRoomDevices(order, payload.traceId, order.roomId);
+    return { orderId: order.id, action: 'CANCELLED' };
+  }
+
+  private async startOrderDevices(order: OrderRow, traceId: string) {
+    const context = this.context(order, traceId, order.roomId);
+    await this.deviceControl.startTask(context, {
+      minute: remainingMinutes(order.endAt),
+      type: 2
+    });
+    await this.switchSocketIfBound(context, true);
+  }
+
+  private async extendOrderDevices(order: OrderRow, traceId: string, addMinutes?: number) {
+    const context = this.context(order, traceId, order.roomId);
+    await this.deviceControl.extendTask(context, boundedMinutes(addMinutes ?? remainingMinutes(order.endAt)));
+    await this.switchSocketIfBound(context, true);
+  }
+
+  private async cancelRoomDevices(order: OrderRow, traceId: string, roomId: string) {
+    const context = this.context(order, traceId, roomId);
+    await this.deviceControl.cancelTask(context);
+    await this.switchSocketIfBound(context, false);
+  }
+
+  private async switchSocketIfBound(context: ReturnType<OrderDeviceAutomationService['context']>, on: boolean) {
+    try {
+      await this.deviceControl.switchSmartSocket(context, { on, slotNum: 1, orderId: context.orderId });
+    } catch (error) {
+      if (error instanceof DeviceControlError && error.code === 'SMART_SOCKET_NOT_BOUND') return;
+      throw error;
+    }
+  }
+
+  private context(order: OrderRow, traceId: string, roomId: string) {
+    return {
+      tenantId: order.tenantId,
+      storeId: order.storeId,
+      roomId,
+      orderId: order.id,
+      traceId,
+      access: systemDeviceAccess()
+    };
+  }
+
+  private async loadOrder(tenantId: string, orderId: string) {
+    const [rows] = await this.pool.execute<OrderRow[]>(
+      `SELECT id, tenant_id AS tenantId, store_id AS storeId, room_id AS roomId,
+              status, start_at AS startAt, end_at AS endAt
+       FROM qipai_orders
+       WHERE tenant_id = ? AND id = ? AND deleted_at IS NULL`,
+      [tenantId, orderId]
+    );
+    if (!rows[0]) throw new OrderDeviceAutomationError('ORDER_NOT_FOUND');
+    return {
+      ...rows[0],
+      id: String(rows[0].id),
+      tenantId: String(rows[0].tenantId),
+      storeId: String(rows[0].storeId),
+      roomId: String(rows[0].roomId)
+    };
+  }
+}
+
+function parseTaskPayload(payload: unknown): {
+  tenantId: string;
+  orderId: string;
+  event: OrderDeviceEvent;
+  traceId: string;
+  addMinutes?: number;
+  previousRoomId?: string;
+} {
+  if (!payload || typeof payload !== 'object' || Array.isArray(payload)) {
+    throw new OrderDeviceAutomationError('DEVICE_TASK_PAYLOAD_INVALID');
+  }
+  const record = payload as Record<string, unknown>;
+  const tenantId = readId(record.tenantId);
+  const orderId = readId(record.orderId);
+  const event = typeof record.event === 'string'
+    && orderDeviceEvents.includes(record.event as OrderDeviceEvent)
+    ? record.event as OrderDeviceEvent
+    : null;
+  const traceId = typeof record.traceId === 'string' && record.traceId.length > 0
+    ? record.traceId.slice(0, 128)
+    : `device-order-${orderId ?? 'unknown'}`;
+  if (!tenantId || !orderId || !event) {
+    throw new OrderDeviceAutomationError('DEVICE_TASK_PAYLOAD_INVALID');
+  }
+  const addMinutes = readPositiveInt(record.addMinutes);
+  const previousRoomId = readId(record.previousRoomId);
+  return { tenantId, orderId, event, traceId, addMinutes, previousRoomId: previousRoomId ?? undefined };
+}
+
+function readId(value: unknown) {
+  if (typeof value === 'string' && /^[1-9]\d{0,19}$/.test(value)) return value;
+  if (typeof value === 'number' && Number.isSafeInteger(value) && value > 0) return String(value);
+  return null;
+}
+
+function readPositiveInt(value: unknown) {
+  if (typeof value !== 'number' || !Number.isSafeInteger(value) || value <= 0) return undefined;
+  return value;
+}
+
+function remainingMinutes(endAt: Date) {
+  return boundedMinutes(Math.ceil((endAt.getTime() - Date.now()) / 60000));
+}
+
+function boundedMinutes(value: number) {
+  return Math.max(1, Math.min(10080, value));
+}
+
+function systemDeviceAccess(): AccessProfile {
+  return {
+    roles: ['PLATFORM_ADMIN'],
+    capabilities: ['device.write'],
+    storeIds: []
+  };
+}
@@ -6,6 +6,11 @@ import type { AsyncTask, TaskType } from './task-repository.js';
 import { TaskRepository } from './task-repository.js';
 import { closeMySqlPool, createMySqlPool } from '../db/mysql.js';
 import { loadConfig } from '../config.js';
+import { MqttService } from '../mqtt/mqtt-service.js';
+import { DeviceCommandService } from '../devices/device-command-service.js';
+import { DeviceControlService } from '../devices/device-control-service.js';
+import { IotMessageService } from '../devices/iot-message-service.js';
+import { OrderDeviceAutomationService } from '../devices/order-device-automation-service.js';

 type TaskHandler = (task: AsyncTask) => Promise<void>;

@@ -60,17 +65,30 @@ export function sleep(ms: number): Promise<void> {
 if (process.argv[1] && fileURLToPath(import.meta.url) === resolve(process.argv[1])) {
  const config = loadConfig();
  const pool = createMySqlPool(config);
+  const iotMessages = new IotMessageService(pool);
+  const mqtt = new MqttService(config.mqtt, undefined, (topic, payload) =>
+    iotMessages.handle(topic, payload)
+  );
+  const deviceControl = new DeviceControlService(
+    pool,
+    new DeviceCommandService(iotMessages, mqtt)
+  );
+  const orderDevices = new OrderDeviceAutomationService(pool, deviceControl);
  const worker = new TaskWorker({
    repository: new TaskRepository(pool),
-    handlers: new Map(),
+    handlers: new Map([
+      ['device.command', async (task) => { await orderDevices.handleTask(task); }]
+    ]),
    workerId: `${hostname()}:${process.pid}:${randomUUID()}`
  });
  const shutdown = () => worker.stop();
  process.on('SIGINT', shutdown);
  process.on('SIGTERM', shutdown);
  try {
+    mqtt.start();
    await worker.run();
  } finally {
+    await mqtt.stop();
    await closeMySqlPool(pool);
  }
 }