diff --git a/backend/package.json b/backend/package.json index dc9957e..9a1ba85 100644 --- a/backend/package.json +++ b/backend/package.json @@ -17,7 +17,7 @@ "db:migrate:verify": "npm run build && node dist/db/migrate-cli.js verify", "db:migrate:down": "npm run build && node dist/db/migrate-cli.js down", "test:mysql:migration": "npm run build && node tests/mysql-migration-roundtrip.test.mjs", - "test": "npm run build && node tests/backend-contract.test.mjs && node tests/mqtt-service.test.mjs && node tests/migration-contract.test.mjs && node tests/mysql-pool-contract.test.mjs && node tests/migration-runner.test.mjs && node tests/legacy-money.test.mjs && node tests/legacy-read-repository.test.mjs && node tests/task-repository.test.mjs && node tests/platform-config-repository.test.mjs && node tests/auth.test.mjs && node tests/rbac.test.mjs && node tests/user-management.test.mjs && node tests/store-room.test.mjs && node tests/content-management.test.mjs && node tests/store-discovery.test.mjs && node tests/store-access.test.mjs && node tests/pricing.test.mjs && node tests/order-state.test.mjs && node tests/order-management.test.mjs && node tests/order-share.test.mjs && node tests/payment.test.mjs && node tests/wechat-pay.test.mjs && node tests/third-party.test.mjs && node tests/profit-sharing.test.mjs && node tests/device.test.mjs && node tests/iot-protocol.test.mjs && node tests/device-control.test.mjs" + "test": "npm run build && node tests/backend-contract.test.mjs && node tests/mqtt-service.test.mjs && node tests/migration-contract.test.mjs && node tests/mysql-pool-contract.test.mjs && node tests/migration-runner.test.mjs && node tests/legacy-money.test.mjs && node tests/legacy-read-repository.test.mjs && node tests/task-repository.test.mjs && node tests/platform-config-repository.test.mjs && node tests/auth.test.mjs && node tests/rbac.test.mjs && node tests/user-management.test.mjs && node tests/store-room.test.mjs && node tests/content-management.test.mjs && node tests/store-discovery.test.mjs && node tests/store-access.test.mjs && node tests/pricing.test.mjs && node tests/order-state.test.mjs && node tests/order-management.test.mjs && node tests/order-share.test.mjs && node tests/payment.test.mjs && node tests/wechat-pay.test.mjs && node tests/third-party.test.mjs && node tests/profit-sharing.test.mjs && node tests/device.test.mjs && node tests/iot-protocol.test.mjs && node tests/device-control.test.mjs && node tests/order-device-automation.test.mjs" }, "dependencies": { "@fastify/cors": "^11.2.0", diff --git a/backend/src/devices/order-device-automation-service.ts b/backend/src/devices/order-device-automation-service.ts new file mode 100644 index 0000000..a3256c3 --- /dev/null +++ b/backend/src/devices/order-device-automation-service.ts @@ -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 + ) {} + + 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, 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( + `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; + 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: [] + }; +} diff --git a/backend/src/tasks/worker.ts b/backend/src/tasks/worker.ts index 6e3144e..3fe139a 100644 --- a/backend/src/tasks/worker.ts +++ b/backend/src/tasks/worker.ts @@ -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; @@ -60,17 +65,30 @@ export function sleep(ms: number): Promise { 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); } } diff --git a/backend/tests/order-device-automation.test.mjs b/backend/tests/order-device-automation.test.mjs new file mode 100644 index 0000000..c3a9f1b --- /dev/null +++ b/backend/tests/order-device-automation.test.mjs @@ -0,0 +1,124 @@ +import assert from 'node:assert/strict'; +import { + OrderDeviceAutomationError, + OrderDeviceAutomationService +} from '../dist/devices/order-device-automation-service.js'; +import { DeviceControlError } from '../dist/devices/device-control-service.js'; + +const calls = []; +let currentOrder = { + id: 31, + tenantId: 7, + storeId: 11, + roomId: 51, + status: 'IN_PROGRESS', + startAt: new Date(Date.now() - 30 * 60_000), + endAt: new Date(Date.now() + 90 * 60_000) +}; +const service = new OrderDeviceAutomationService({ + async execute(sql, params) { + if (sql.includes('FROM qipai_orders')) { + assert.equal(params[0], '7'); + assert.equal(params[1], '31'); + return [[currentOrder], []]; + } + return [[], []]; + } +}, { + async startTask(context, input) { + calls.push(['startTask', context, input]); + return { status: 'PUBLISHED' }; + }, + async extendTask(context, addminute) { + calls.push(['extendTask', context, addminute]); + return { status: 'PUBLISHED' }; + }, + async cancelTask(context) { + calls.push(['cancelTask', context]); + return { status: 'PUBLISHED' }; + }, + async switchSmartSocket(context, input) { + calls.push(['socket', context, input]); + if (context.roomId === '52') { + throw new DeviceControlError('SMART_SOCKET_NOT_BOUND'); + } + return { status: 'PUBLISHED' }; + } +}); + +function task(payload) { + return { + id: '1', + tenantId: '7', + taskType: 'device.command', + idempotencyKey: `device:${payload.orderId}:${payload.event}`, + payload, + status: 'RUNNING', + attempts: 1, + maxAttempts: 8 + }; +} + +await service.handleTask(task({ + tenantId: '7', + orderId: '31', + event: 'ORDER_STARTED', + traceId: 'm06f-start' +})); +assert.equal(calls[0][0], 'startTask'); +assert.equal(calls[0][2].type, 2); +assert.equal(calls[0][1].access.roles.includes('PLATFORM_ADMIN'), true); +assert.equal(calls[1][0], 'socket'); +assert.equal(calls[1][2].on, true); + +await service.handleTask(task({ + tenantId: '7', + orderId: '31', + event: 'ORDER_RENEWED', + addMinutes: 45, + traceId: 'm06f-renew' +})); +assert.equal(calls.at(-2)[0], 'extendTask'); +assert.equal(calls.at(-2)[2], 45); +assert.equal(calls.at(-1)[2].on, true); + +await service.handleTask(task({ + tenantId: '7', + orderId: '31', + event: 'ORDER_CANCELLED', + traceId: 'm06f-cancel' +})); +assert.equal(calls.at(-2)[0], 'cancelTask'); +assert.equal(calls.at(-1)[2].on, false); + +currentOrder = { ...currentOrder, roomId: 52, status: 'IN_PROGRESS' }; +const changed = await service.handleTask(task({ + tenantId: '7', + orderId: '31', + event: 'ORDER_ROOM_CHANGED', + previousRoomId: '51', + traceId: 'm06f-room' +})); +assert.equal(changed.action, 'ROOM_CHANGED'); +assert.equal(calls.at(-3)[0], 'socket'); +assert.equal(calls.at(-3)[1].roomId, '51'); +assert.equal(calls.at(-3)[2].on, false); +assert.equal(calls.at(-2)[0], 'startTask'); +assert.equal(calls.at(-2)[1].roomId, '52'); + +currentOrder = { ...currentOrder, status: 'CLOSED' }; +const skipped = await service.handleTask(task({ + tenantId: '7', + orderId: '31', + event: 'ORDER_STARTED', + traceId: 'm06f-stale' +})); +assert.equal(skipped.skipped, true); + +await assert.rejects( + () => service.handleTask({ ...task({ tenantId: '8', orderId: '31', event: 'ORDER_STARTED' }), tenantId: '7' }), + (error) => error instanceof OrderDeviceAutomationError + && error.code === 'DEVICE_TASK_TENANT_MISMATCH' +); + +console.log('PASS: M06-F order device automation dispatches start, renew, cancel and room-change tasks safely.');