From e795cdb6935568611a5d27d78991bf9e7ee2f7df Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 22 Jun 2026 18:29:07 +0800 Subject: [PATCH] =?UTF-8?q?feat(M06-C):=20=E5=AE=8C=E6=88=90=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E5=8D=8F=E8=AE=AE=E9=80=82=E9=85=8D=E4=B8=8E=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E5=B9=82=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/package.json | 2 +- backend/src/db/migration-runner.ts | 10 +- backend/src/devices/device-command-service.ts | 42 ++++ backend/src/devices/iot-message-service.ts | 209 +++++++++++++++++ backend/src/devices/jilian-adapters.ts | 218 ++++++++++++++++++ backend/src/mqtt/mqtt-service.ts | 20 +- backend/src/server.ts | 6 +- backend/tests/iot-protocol.test.mjs | 161 +++++++++++++ backend/tests/migration-contract.test.mjs | 16 +- backend/tests/migration-runner.test.mjs | 3 +- .../tests/mysql-migration-roundtrip.test.mjs | 77 ++++++- .../2026062220_m06c_iot_messages.down.sql | 4 + .../2026062220_m06c_iot_messages.up.sql | 89 +++++++ .../2026062220_m06c_iot_messages.verify.sql | 23 ++ scripts/dev/wsl/mysql-migration-roundtrip.sh | 4 +- 15 files changed, 867 insertions(+), 17 deletions(-) create mode 100644 backend/src/devices/device-command-service.ts create mode 100644 backend/src/devices/iot-message-service.ts create mode 100644 backend/src/devices/jilian-adapters.ts create mode 100644 backend/tests/iot-protocol.test.mjs create mode 100644 database/migrations/2026062220_m06c_iot_messages.down.sql create mode 100644 database/migrations/2026062220_m06c_iot_messages.up.sql create mode 100644 database/migrations/2026062220_m06c_iot_messages.verify.sql diff --git a/backend/package.json b/backend/package.json index 3bec866..8bd5e95 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" + "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" }, "dependencies": { "@fastify/cors": "^11.2.0", diff --git a/backend/src/db/migration-runner.ts b/backend/src/db/migration-runner.ts index 3311c27..48605f3 100644 --- a/backend/src/db/migration-runner.ts +++ b/backend/src/db/migration-runner.ts @@ -39,7 +39,8 @@ const migrationFiles: Record = { 'database/migrations/2026062216_m05b_wechat_refunds.up.sql', 'database/migrations/2026062217_m05c_third_party.up.sql', 'database/migrations/2026062218_m05d_profit_sharing.up.sql', - 'database/migrations/2026062219_m06b_device_topology.up.sql' + 'database/migrations/2026062219_m06b_device_topology.up.sql', + 'database/migrations/2026062220_m06c_iot_messages.up.sql' ], verify: [ 'database/migrations/2026061601_m01b_core_schema.verify.sql', @@ -60,9 +61,11 @@ const migrationFiles: Record = { 'database/migrations/2026062216_m05b_wechat_refunds.verify.sql', 'database/migrations/2026062217_m05c_third_party.verify.sql', 'database/migrations/2026062218_m05d_profit_sharing.verify.sql', - 'database/migrations/2026062219_m06b_device_topology.verify.sql' + 'database/migrations/2026062219_m06b_device_topology.verify.sql', + 'database/migrations/2026062220_m06c_iot_messages.verify.sql' ], down: [ + 'database/migrations/2026062220_m06c_iot_messages.down.sql', 'database/migrations/2026062219_m06b_device_topology.down.sql', 'database/migrations/2026062218_m05d_profit_sharing.down.sql', 'database/migrations/2026062217_m05c_third_party.down.sql', @@ -216,7 +219,8 @@ export async function executeMigrationPlan( 1, 5, 3, 1, 5, 6, 1, 3, 7, 5, 1, - 5, 8, 5, 2, 1 + 5, 8, 5, 2, 1, + 3, 3, 9, 1 ][index] ?? 1; if (!Array.isArray(result) || result.length < minimumRows) { throw new Error( diff --git a/backend/src/devices/device-command-service.ts b/backend/src/devices/device-command-service.ts new file mode 100644 index 0000000..ec8810f --- /dev/null +++ b/backend/src/devices/device-command-service.ts @@ -0,0 +1,42 @@ +import type { MqttTransport } from '../mqtt/mqtt-service.js'; +import { generateCommandId, type IotMessageService } from './iot-message-service.js'; + +export class DeviceCommandService { + private sequence = 0; + + constructor( + private readonly messages: Pick, + private readonly transport: Pick + ) {} + + async issue(input: { + tenantId: string; assetId: string; deviceId: string; storeId: string; + roomId?: string | null; orderId?: string | null; commandType: string; + payloadFactory: (commandId: string) => Record; + traceId: string; expiresAt?: Date | null; + }) { + const commandId = generateCommandId(Date.now(), this.sequence++); + const payload = input.payloadFactory(commandId); + await this.messages.createCommand({ + tenantId: input.tenantId, + assetId: input.assetId, + storeId: input.storeId, + roomId: input.roomId, + orderId: input.orderId, + commandId, + commandType: input.commandType, + payload, + traceId: input.traceId, + expiresAt: input.expiresAt + }); + try { + await this.transport.publishDeviceCommand(input.deviceId, JSON.stringify(payload)); + await this.messages.markPublished(input.tenantId, commandId); + return { commandId, status: 'PUBLISHED' as const }; + } catch (error) { + await this.messages.markPublishFailed(input.tenantId, commandId, 'MQTT_PUBLISH_FAILED'); + throw error; + } + } +} diff --git a/backend/src/devices/iot-message-service.ts b/backend/src/devices/iot-message-service.ts new file mode 100644 index 0000000..25b1306 --- /dev/null +++ b/backend/src/devices/iot-message-service.ts @@ -0,0 +1,209 @@ +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 = { + CONTROL_BOX: new JilianControlBoxAdapter(), + SUB_LOCK: new JilianSub1GLockAdapter(), + SMART_SOCKET: new JilianSmartSocketAdapter() + }; + + constructor(private readonly pool: MySqlPool) {} + + async handle(topic: string, payload: Buffer): Promise { + 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( + `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( + `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; 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( + `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( + `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( + `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( + `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 { + if (!payload || typeof payload !== 'object' || Array.isArray(payload)) { + throw new Error('MQTT payload must be a JSON object.'); + } + return payload as Record; +} + +function readDeviceId(record: Record): string | null { + const value = record.DeviceID ?? record.deviceID; + return typeof value === 'string' ? value : null; +} diff --git a/backend/src/devices/jilian-adapters.ts b/backend/src/devices/jilian-adapters.ts new file mode 100644 index 0000000..710e8ee --- /dev/null +++ b/backend/src/devices/jilian-adapters.ts @@ -0,0 +1,218 @@ +import { z } from 'zod'; + +const commandId = z.string().regex(/^\d{1,13}$/); +const resultCode = z.enum([ + 'ok', 'fail', 'busy', 'unconfirm', 'timeout', 'full', 'unknown' +]); +const vendorMessage = z.object({ + id: commandId.optional(), + DeviceID: z.string().min(1).max(64).optional(), + deviceID: z.string().min(1).max(64).optional(), + IMEI: z.string().max(64).optional(), + result: resultCode.optional(), + event: z.string().max(64).optional(), + action: z.string().max(64).optional(), + read: z.string().max(64).optional(), + timestamp: z.union([z.string(), z.number()]).optional() +}).passthrough(); + +export type NormalizedVendorMessage = { + kind: 'ACK' | 'EVENT' | 'SNAPSHOT'; + commandId: string | null; + eventType: string; + result: z.infer | null; + deviceId: string | null; + eventAt: Date | null; + payload: Record; +}; + +export interface ProtocolAdapter { + parseUplink(payload: unknown): NormalizedVendorMessage; +} + +export class JilianControlBoxAdapter implements ProtocolAdapter { + read(target: 'basicInfo' | 'mqttConfig' | 'startVoice' | 'task' | 'taskconfig') { + return z.object({ read: z.literal(target) }).parse({ read: target }); + } + + controlPower(input: { + id: string; slot1?: 'on' | 'off'; slot2?: 'on' | 'off'; + slot3?: 'on' | 'off'; slotall?: 'on' | 'off'; + }) { + return z.object({ + action: z.literal('ConctolPower'), id: commandId, + slot1: z.enum(['on', 'off']).optional(), + slot2: z.enum(['on', 'off']).optional(), + slot3: z.enum(['on', 'off']).optional(), + slotall: z.enum(['on', 'off']).optional() + }).refine((value) => value.slot1 || value.slot2 || value.slot3 || value.slotall) + .parse({ action: 'ConctolPower', ...input }); + } + + controlDoor(input: { + id: string; order: 'open' | 'close'; holdopen?: 0 | 1; delayTime?: number; + }) { + return z.object({ + action: z.literal('Crldoor'), id: commandId, + order: z.enum(['open', 'close']), holdopen: z.union([z.literal(0), z.literal(1)]).default(0), + delayTime: z.number().int().min(1).max(14).default(4) + }).parse({ action: 'Crldoor', ...input }); + } + + playTts(input: { + id: string; content: string; volume?: number; playCount?: number; + priority?: number; speaker?: number; style?: number; speed?: number; pitch?: number; + }) { + return z.object({ + action: z.literal('PlayTTS'), id: commandId, + content: z.string().trim().min(1).max(500), + volume: z.number().int().min(0).max(100).default(80), + playCount: z.number().int().min(1).max(10).default(1), + priority: z.number().int().min(0).max(10).default(0), + speaker: z.number().int().min(0).max(20).default(0), + style: z.number().int().min(0).max(20).default(0), + speed: z.number().int().min(-500).max(500).default(0), + pitch: z.number().int().min(-500).max(500).default(0) + }).parse({ action: 'PlayTTS', ...input }); + } + + stopTts(id: string) { + return z.object({ action: z.literal('stopTTS'), id: commandId }) + .parse({ action: 'stopTTS', id }); + } + + controlLed(input: { id: string; minute: number }) { + return z.object({ + action: z.literal('CrlLED'), id: commandId, + minute: z.number().int().min(0).max(10080) + }).parse({ action: 'CrlLED', ...input }); + } + + startTask(input: { + id: string; minute: number; type: 1 | 2 | 3; + subID?: string; holdopen?: 0 | 1; delayTime?: number; + }) { + return z.object({ + action: z.literal('task'), id: commandId, + minute: z.number().int().min(1).max(10080), + type: z.union([z.literal(1), z.literal(2), z.literal(3)]), + subID: z.string().min(1).max(64).optional(), + holdopen: z.union([z.literal(0), z.literal(1)]).default(0), + delayTime: z.number().int().min(1).max(14).default(4) + }).parse({ action: 'task', ...input }); + } + + extendTask(input: { id: string; addminute: number }) { + return z.object({ + action: z.literal('addtask'), id: commandId, + addminute: z.number().int().min(1).max(10080) + }).parse({ action: 'addtask', ...input }); + } + + cancelTask(id: string) { + return z.object({ action: z.literal('canceltask'), id: commandId }) + .parse({ action: 'canceltask', id }); + } + + parseUplink(payload: unknown) { + return normalizeVendorMessage(payload); + } +} + +export class JilianSub1GLockAdapter implements ProtocolAdapter { + pair(input: { id: string; timeout?: number }) { + return z.object({ + action: z.literal('AddDevice'), id: commandId, + timeout: z.number().int().min(10).max(300).default(60) + }).parse({ action: 'AddDevice', ...input }); + } + + control(input: { + id: string; subID: string; + order: 'open' | 'close' | 'setkey' | 'delkey' | 'setcard' | 'delcard' | 'factoryreset'; + holdopen?: 0 | 1; delayTime?: number; content?: string; + }) { + return z.object({ + action: z.literal('CtrlDevice'), id: commandId, + subID: z.string().min(1).max(64), + order: z.enum([ + 'open', 'close', 'setkey', 'delkey', 'setcard', 'delcard', 'factoryreset' + ]), + holdopen: z.union([z.literal(0), z.literal(1)]).optional(), + delayTime: z.number().int().min(1).max(14).optional(), + content: z.string().min(1).max(128).optional() + }).superRefine((value, context) => { + if (['setkey', 'setcard'].includes(value.order) && !value.content) { + context.addIssue({ code: z.ZodIssueCode.custom, message: 'content is required' }); + } + }).parse({ action: 'CtrlDevice', ...input }); + } + + parseUplink(payload: unknown) { + return normalizeVendorMessage(payload); + } +} + +export class JilianSmartSocketAdapter implements ProtocolAdapter { + read(target: 'basicInfo' | 'workInfo') { + return z.object({ read: z.literal(target) }).parse({ read: target }); + } + + switch(input: { id: string; on: boolean; slotNum?: number }) { + return z.object({ + action: z.enum(['on', 'off']), id: commandId, + slotNum: z.number().int().min(1).max(20).default(1) + }).parse({ action: input.on ? 'on' : 'off', id: input.id, slotNum: input.slotNum }); + } + + localTask(input: { + id: string; taskNum: number; action: 'on' | 'off'; + mode: 'once' | 'daily' | 'weekly'; time: string; weekdays?: number[]; + }) { + return z.object({ + action: z.literal('localtask'), id: commandId, + taskNum: z.number().int().min(1).max(20), + switch: z.enum(['on', 'off']), + mode: z.enum(['once', 'daily', 'weekly']), + time: z.string().regex(/^\d{2}:\d{2}$/), + weekdays: z.array(z.number().int().min(1).max(7)).max(7).optional() + }).parse({ + action: 'localtask', id: input.id, taskNum: input.taskNum, + switch: input.action, mode: input.mode, time: input.time, weekdays: input.weekdays + }); + } + + clearTask(input: { id: string; taskNum: number }) { + return z.object({ + action: z.literal('clearTask'), id: commandId, + taskNum: z.number().int().min(0).max(20) + }).parse({ action: 'clearTask', ...input }); + } + + parseUplink(payload: unknown) { + return normalizeVendorMessage(payload); + } +} + +function normalizeVendorMessage(payload: unknown): NormalizedVendorMessage { + const parsed = vendorMessage.parse(payload); + const record = parsed as Record; + const eventName = parsed.event ?? parsed.action ?? parsed.read ?? 'snapshot'; + return { + kind: parsed.result ? 'ACK' : parsed.event ? 'EVENT' : 'SNAPSHOT', + commandId: parsed.id ?? null, + eventType: eventName, + result: parsed.result ?? null, + deviceId: parsed.DeviceID ?? parsed.deviceID ?? null, + eventAt: parseEventAt(parsed.timestamp), + payload: record + }; +} + +function parseEventAt(value: string | number | undefined): Date | null { + if (value === undefined) return null; + const date = typeof value === 'number' + ? new Date(value < 10_000_000_000 ? value * 1000 : value) + : new Date(value); + return Number.isNaN(date.getTime()) ? null : date; +} diff --git a/backend/src/mqtt/mqtt-service.ts b/backend/src/mqtt/mqtt-service.ts index 8a97b6d..c24d84e 100644 --- a/backend/src/mqtt/mqtt-service.ts +++ b/backend/src/mqtt/mqtt-service.ts @@ -26,6 +26,7 @@ export interface MqttClientLike { } export type MqttClientFactory = (url: string, options: IClientOptions) => MqttClientLike; +export type MqttMessageHandler = (topic: string, payload: Buffer) => Promise; export interface MqttHealthSnapshot { configured: boolean; @@ -39,7 +40,14 @@ export interface MqttHealthSnapshot { lastError: string | null; } -export class MqttService { +export interface MqttTransport { + start(): void; + stop(): Promise; + publishDeviceCommand(deviceId: string, payload: Buffer | string): Promise; + health(): MqttHealthSnapshot; +} + +export class MqttService implements MqttTransport { private client: MqttClientLike | null = null; private subscriptionsReady = false; private reconnectCount = 0; @@ -51,7 +59,8 @@ export class MqttService { constructor( private readonly config: AppConfig['mqtt'], - private readonly clientFactory: MqttClientFactory = connect as MqttClientFactory + private readonly clientFactory: MqttClientFactory = connect as MqttClientFactory, + private readonly messageHandler?: MqttMessageHandler ) {} start(): void { @@ -89,7 +98,7 @@ export class MqttService { this.client.on('error', (error: Error) => { this.lastError = sanitizeError(error); }); - this.client.on('message', (_topic: string, payload: Buffer) => { + this.client.on('message', (topic: string, payload: Buffer) => { if (payload.length > this.config.maxMessageBytes) { this.rejectedOversizeMessages += 1; this.lastError = `MQTT message exceeded ${this.config.maxMessageBytes} bytes`; @@ -97,6 +106,11 @@ export class MqttService { } this.receivedMessages += 1; this.lastMessageAt = new Date().toISOString(); + void this.messageHandler?.(topic, payload).catch((error: unknown) => { + this.lastError = sanitizeError( + error instanceof Error ? error : new Error('MQTT message handler failed') + ); + }); }); } diff --git a/backend/src/server.ts b/backend/src/server.ts index abb1abc..0e2f284 100644 --- a/backend/src/server.ts +++ b/backend/src/server.ts @@ -28,6 +28,7 @@ import { import { ThirdPartyService } from './third-party/third-party-service.js'; import { MqttService } from './mqtt/mqtt-service.js'; import { DeviceRepository } from './devices/device-repository.js'; +import { IotMessageService } from './devices/iot-message-service.js'; const config = loadConfig(); const pool = createMySqlPool(config); @@ -38,7 +39,10 @@ const paymentRepository = new PaymentRepository(pool); const wechatCredentials = parseWechatPayCredentials(config.payment.wechatCredentialsJson); const wechatPayClient = new WechatPayClient(new FetchWechatPayTransport()); const thirdPartyCredentials = parseThirdPartyCredentials(config.thirdParty.credentialsJson); -const mqtt = new MqttService(config.mqtt); +const iotMessages = new IotMessageService(pool); +const mqtt = new MqttService(config.mqtt, undefined, (topic, payload) => + iotMessages.handle(topic, payload) +); const app = await buildApp({ config, mqtt, diff --git a/backend/tests/iot-protocol.test.mjs b/backend/tests/iot-protocol.test.mjs new file mode 100644 index 0000000..834a1d1 --- /dev/null +++ b/backend/tests/iot-protocol.test.mjs @@ -0,0 +1,161 @@ +import assert from 'node:assert/strict'; +import { + JilianControlBoxAdapter, + JilianSmartSocketAdapter, + JilianSub1GLockAdapter +} from '../dist/devices/jilian-adapters.js'; +import { + generateCommandId, + IotMessageService +} from '../dist/devices/iot-message-service.js'; +import { DeviceCommandService } from '../dist/devices/device-command-service.js'; + +const control = new JilianControlBoxAdapter(); +assert.deepEqual(control.read('basicInfo'), { read: 'basicInfo' }); +assert.deepEqual(control.controlPower({ + id: '1234567890123', slot1: 'on', slot3: 'off' +}), { + action: 'ConctolPower', id: '1234567890123', slot1: 'on', slot3: 'off' +}); +assert.equal(control.controlDoor({ + id: '1234567890123', order: 'open' +}).action, 'Crldoor'); +assert.equal(control.playTts({ + id: '1234567890123', content: '欢迎光临' +}).action, 'PlayTTS'); +assert.deepEqual(control.stopTts('1234567890123'), { + action: 'stopTTS', id: '1234567890123' +}); +assert.equal(control.controlLed({ id: '1234567890123', minute: 30 }).action, 'CrlLED'); +assert.equal(control.startTask({ + id: '1234567890123', minute: 120, type: 2, subID: 'SUB001' +}).action, 'task'); +assert.equal(control.extendTask({ + id: '1234567890123', addminute: 30 +}).action, 'addtask'); +assert.equal(control.cancelTask('1234567890123').action, 'canceltask'); +assert.throws( + () => control.controlDoor({ id: '12345678901234', order: 'open' }), + /Invalid/ +); +assert.throws( + () => control.controlDoor({ id: '123', order: 'open', delayTime: 15 }), + /less than or equal to 14/ +); + +const lock = new JilianSub1GLockAdapter(); +assert.deepEqual(lock.pair({ id: '123', timeout: 60 }), { + action: 'AddDevice', id: '123', timeout: 60 +}); +assert.equal(lock.control({ + id: '123', subID: 'SUB001', order: 'open', delayTime: 4 +}).action, 'CtrlDevice'); +assert.throws( + () => lock.control({ id: '123', subID: 'SUB001', order: 'setkey' }), + /content is required/ +); + +const socket = new JilianSmartSocketAdapter(); +assert.deepEqual(socket.read('workInfo'), { read: 'workInfo' }); +assert.deepEqual(socket.switch({ id: '123', on: true }), { + action: 'on', id: '123', slotNum: 1 +}); +assert.equal(socket.localTask({ + id: '123', taskNum: 20, action: 'off', mode: 'weekly', + time: '23:00', weekdays: [1, 5] +}).action, 'localtask'); +assert.deepEqual(socket.clearTask({ id: '123', taskNum: 0 }), { + action: 'clearTask', id: '123', taskNum: 0 +}); + +const ack = control.parseUplink({ + DeviceID: 'BOX_001', id: '123', result: 'unconfirm', action: 'task' +}); +assert.equal(ack.kind, 'ACK'); +assert.equal(ack.result, 'unconfirm'); +assert.equal(ack.eventType, 'task'); +const event = lock.parseUplink({ + deviceID: 'BOX_001', event: 'record', type: 'card', state: 'open', + timestamp: 1782120000 +}); +assert.equal(event.kind, 'EVENT'); +assert.equal(event.eventType, 'record'); +assert.equal(event.eventAt?.getUTCFullYear(), 2026); + +assert.equal(generateCommandId(1782120000000, 7), '1782120000007'); +assert.match(generateCommandId(), /^\d{13}$/); + +const calls = []; +let eventInsertCount = 0; +const service = new IotMessageService({ + async execute(sql, params) { + calls.push({ sql, params }); + if (sql.includes('FROM qipai_devices')) { + return [[{ + id: 51, tenantId: 7, storeId: 11, roomId: 31, + deviceId: 'BOX_001', deviceType: 'CONTROL_BOX' + }], []]; + } + if (sql.includes('INSERT INTO qipai_iot_device_events')) { + eventInsertCount += 1; + return [{ affectedRows: eventInsertCount === 1 ? 1 : 2 }, []]; + } + return [{ affectedRows: 1, insertId: 81 }, []]; + } +}); +const payload = Buffer.from(JSON.stringify({ + DeviceID: 'BOX_001', id: '123', result: 'ok', action: 'ConctolPower', + slot1: 'on' +})); +await service.handle('/devicesend/BOX_001', payload); +await service.handle('/devicesend/BOX_001', payload); +assert.equal( + calls.filter((item) => item.sql.includes('UPDATE qipai_iot_commands')).length, + 1, + 'duplicate QoS 1 payload must not repeat command/device side effects' +); +assert.equal( + calls.some((item) => item.sql.includes('qipai_iot_dead_letters')), + false +); + +await service.handle('/bad-topic/BOX_001', Buffer.from('{}')); +assert.equal( + calls.some((item) => item.sql.includes('qipai_iot_dead_letters')), + true +); + +await assert.rejects( + () => service.createCommand({ + tenantId: '7', assetId: '51', storeId: '11', + commandId: '12345678901234', commandType: 'ConctolPower', + payload: {}, traceId: 'trace' + }), + /IOT_COMMAND_ID_INVALID/ +); + +const commandCalls = []; +const commandService = new DeviceCommandService({ + async createCommand(input) { commandCalls.push(['create', input]); }, + async markPublished(tenantId, commandId) { + commandCalls.push(['published', tenantId, commandId]); + }, + async markPublishFailed() { throw new Error('unexpected publish failure'); } +}, { + async publishDeviceCommand(deviceId, payloadText) { + commandCalls.push(['mqtt', deviceId, JSON.parse(payloadText)]); + } +}); +const issued = await commandService.issue({ + tenantId: '7', assetId: '51', deviceId: 'BOX_001', storeId: '11', + commandType: 'ConctolPower', + payloadFactory: (id) => control.controlPower({ id, slot1: 'on' }), + traceId: 'trace' +}); +assert.equal(issued.status, 'PUBLISHED'); +assert.match(issued.commandId, /^\d{13}$/); +assert.equal(commandCalls[0][0], 'create'); +assert.equal(commandCalls[1][0], 'mqtt'); +assert.equal(commandCalls[2][0], 'published'); + +console.log('PASS: M06-C adapters, vendor spelling, command IDs, ACKs and QoS 1 dedup work.'); diff --git a/backend/tests/migration-contract.test.mjs b/backend/tests/migration-contract.test.mjs index 372ecea..3fef286 100644 --- a/backend/tests/migration-contract.test.mjs +++ b/backend/tests/migration-contract.test.mjs @@ -66,6 +66,9 @@ const profitSharingVerifySql = read('database/migrations/2026062218_m05d_profit_ const deviceTopologyUpSql = read('database/migrations/2026062219_m06b_device_topology.up.sql'); const deviceTopologyDownSql = read('database/migrations/2026062219_m06b_device_topology.down.sql'); const deviceTopologyVerifySql = read('database/migrations/2026062219_m06b_device_topology.verify.sql'); +const iotMessagesUpSql = read('database/migrations/2026062220_m06c_iot_messages.up.sql'); +const iotMessagesDownSql = read('database/migrations/2026062220_m06c_iot_messages.down.sql'); +const iotMessagesVerifySql = read('database/migrations/2026062220_m06c_iot_messages.verify.sql'); const coreTables = [ 'qipai_schema_migrations', @@ -301,5 +304,16 @@ assert.match(deviceTopologyUpSql, /uq_qipai_device_link_child/); assert.match(deviceTopologyUpSql, /uq_qipai_device_link_room_type/); assert.match(deviceTopologyUpSql, /device\.read/); assert.match(deviceTopologyUpSql, /device\.write/); +for (const table of [ + 'qipai_iot_commands', 'qipai_iot_device_events', 'qipai_iot_dead_letters' +]) { + assert.match(iotMessagesUpSql, new RegExp(`CREATE TABLE IF NOT EXISTS ${table}`)); + assert.match(iotMessagesDownSql, new RegExp(`DROP TABLE IF EXISTS ${table}`)); + assert.match(iotMessagesVerifySql, new RegExp(`'${table}'`)); +} +assert.match(iotMessagesUpSql, /command_id VARCHAR\(13\)/); +assert.match(iotMessagesUpSql, /uq_qipai_iot_event_dedup/); +assert.match(iotMessagesUpSql, /receive_count INT UNSIGNED/); +assert.match(iotMessagesUpSql, /PENDING/); -console.log('PASS: M01-B through M06-B migration contracts are present.'); +console.log('PASS: M01-B through M06-C migration contracts are present.'); diff --git a/backend/tests/migration-runner.test.mjs b/backend/tests/migration-runner.test.mjs index c0301b6..8062ff8 100644 --- a/backend/tests/migration-runner.test.mjs +++ b/backend/tests/migration-runner.test.mjs @@ -30,7 +30,8 @@ assert.match(plan.file, /2026062015_m05a_payment_domain\.up\.sql/); assert.match(plan.file, /2026062216_m05b_wechat_refunds\.up\.sql/); assert.match(plan.file, /2026062217_m05c_third_party\.up\.sql/); assert.match(plan.file, /2026062218_m05d_profit_sharing\.up\.sql/); -assert.match(plan.file, /2026062219_m06b_device_topology\.up\.sql$/); +assert.match(plan.file, /2026062219_m06b_device_topology\.up\.sql/); +assert.match(plan.file, /2026062220_m06c_iot_messages\.up\.sql$/); assert.match(plan.checksum, /^[a-f0-9]{64}$/); assert.ok(plan.statements.length >= 11); diff --git a/backend/tests/mysql-migration-roundtrip.test.mjs b/backend/tests/mysql-migration-roundtrip.test.mjs index 1285a96..9d0e313 100644 --- a/backend/tests/mysql-migration-roundtrip.test.mjs +++ b/backend/tests/mysql-migration-roundtrip.test.mjs @@ -36,6 +36,7 @@ import { WechatPayClient } from '../dist/payments/wechat-pay-client.js'; import { ThirdPartyClient } from '../dist/third-party/third-party-client.js'; import { ThirdPartyService } from '../dist/third-party/third-party-service.js'; import { DeviceRepository } from '../dist/devices/device-repository.js'; +import { IotMessageService } from '../dist/devices/iot-message-service.js'; import { executeMigrationPlan, loadMigrationPlan, @@ -58,6 +59,9 @@ const expectedTables = [ 'qipai_group_redemptions', 'qipai_group_vouchers', 'qipai_holiday_calendar', + 'qipai_iot_commands', + 'qipai_iot_dead_letters', + 'qipai_iot_device_events', 'qipai_legacy_table_mappings', 'qipai_media_assets', 'qipai_members', @@ -121,12 +125,13 @@ async function readMigrationVersions(pool) { const [rows] = await pool.query( `SELECT version, name FROM qipai_schema_migrations - WHERE version IN (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + WHERE version IN (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ORDER BY version`, ['2026061601', '2026061802', '2026061803', '2026061804', '2026061805', '2026061806', '2026061807', '2026061808', '2026061809', '2026061810', '2026061811', '2026062012', '2026062013', '2026062014', - '2026062015', '2026062216', '2026062217', '2026062218', '2026062219'] + '2026062015', '2026062216', '2026062217', '2026062218', '2026062219', + '2026062220'] ); return rows; } @@ -1670,6 +1675,60 @@ async function assertDeviceTopology(pool, context) { }]); } +async function assertIotMessages(pool, context) { + const [deviceRows] = await pool.query( + `SELECT id, store_id AS storeId, room_id AS roomId, device_id AS deviceId + FROM qipai_devices + WHERE tenant_id = ? AND device_id = 'M06B_BOX_001'`, + [context.tenantId] + ); + const device = deviceRows[0]; + const service = new IotMessageService(pool); + await service.createCommand({ + tenantId: context.tenantId, + assetId: String(device.id), + storeId: String(device.storeId), + roomId: String(device.roomId), + commandId: '1782120000001', + commandType: 'ConctolPower', + payload: { + action: 'ConctolPower', id: '1782120000001', slot1: 'on' + }, + traceId: 'm06c-live-test' + }); + assert.equal(await service.markPublished(context.tenantId, '1782120000001'), true); + const payload = Buffer.from(JSON.stringify({ + DeviceID: device.deviceId, + id: '1782120000001', + action: 'ConctolPower', + result: 'ok', + slot1: 'on', + timestamp: 1782120000 + })); + await service.handle(`/devicesend/${device.deviceId}`, payload); + await service.handle(`/devicesend/${device.deviceId}`, payload); + const [commandRows] = await pool.query( + `SELECT status, failure_code AS failureCode + FROM qipai_iot_commands + WHERE tenant_id = ? AND command_id = '1782120000001'`, + [context.tenantId] + ); + assert.deepEqual(commandRows, [{ status: 'ACKED', failureCode: '' }]); + const [eventRows] = await pool.query( + `SELECT receive_count AS receiveCount, processing_status AS processingStatus + FROM qipai_iot_device_events + WHERE tenant_id = ? AND command_id = '1782120000001'`, + [context.tenantId] + ); + assert.deepEqual(eventRows, [{ receiveCount: 2, processingStatus: 'PROCESSED' }]); + await service.handle('/invalid/topic', Buffer.from('{bad-json')); + const [deadRows] = await pool.query( + `SELECT error_code AS errorCode, receive_count AS receiveCount + FROM qipai_iot_dead_letters WHERE topic = '/invalid/topic'` + ); + assert.deepEqual(deadRows, [{ errorCode: 'MQTT_TOPIC_INVALID', receiveCount: 1 }]); +} + const config = loadConfig(); assert.equal(config.mysql.passwordConfigured, true, 'Live migration test requires a temporary password.'); assert.match( @@ -1712,7 +1771,8 @@ try { { version: '2026062216', name: 'm05b_wechat_refunds' }, { version: '2026062217', name: 'm05c_third_party' }, { version: '2026062218', name: 'm05d_profit_sharing' }, - { version: '2026062219', name: 'm06b_device_topology' } + { version: '2026062219', name: 'm06b_device_topology' }, + { version: '2026062220', name: 'm06c_iot_messages' } ]); await assertTaskDurability(pool); const loginContext = await assertPlatformTenantIsolation(pool); @@ -1730,13 +1790,14 @@ try { await assertThirdPartyDomain(pool, loginContext); await assertProfitSharingDomain(pool, loginContext); await assertDeviceTopology(pool, loginContext); + await assertIotMessages(pool, loginContext); await assertLegacyCompatibility(pool); console.log('PASS: first up, verify, tenant isolation and revocable auth checks completed.'); await executeMigrationPlan(pool, plans.down); assert.deepEqual(await readCoreTables(pool), []); await assertLegacyCompatibility(pool); - console.log('PASS: down removed all M01-B through M06-B tables.'); + console.log('PASS: down removed all M01-B through M06-C tables.'); await executeMigrationPlan(pool, plans.up); await executeMigrationPlan(pool, plans.verify); @@ -1760,7 +1821,8 @@ try { { version: '2026062216', name: 'm05b_wechat_refunds' }, { version: '2026062217', name: 'm05c_third_party' }, { version: '2026062218', name: 'm05d_profit_sharing' }, - { version: '2026062219', name: 'm06b_device_topology' } + { version: '2026062219', name: 'm06b_device_topology' }, + { version: '2026062220', name: 'm06c_iot_messages' } ]); await assertLegacyCompatibility(pool); console.log('PASS: second up and verify restored the schema.'); @@ -1858,6 +1920,11 @@ try { 'control target conflict across control box and smart socket', 'Sub-1G parent-child topology', 'device status snapshots and maintenance state' + , + '13-digit IoT command state transition', + 'QoS 1 duplicate event receive count', + 'ACK correlation without duplicate side effects', + 'invalid Topic dead-letter persistence' ] }, null, 2)); } finally { diff --git a/database/migrations/2026062220_m06c_iot_messages.down.sql b/database/migrations/2026062220_m06c_iot_messages.down.sql new file mode 100644 index 0000000..4cb3a36 --- /dev/null +++ b/database/migrations/2026062220_m06c_iot_messages.down.sql @@ -0,0 +1,4 @@ +DELETE FROM qipai_schema_migrations WHERE version = '2026062220'; +DROP TABLE IF EXISTS qipai_iot_dead_letters; +DROP TABLE IF EXISTS qipai_iot_device_events; +DROP TABLE IF EXISTS qipai_iot_commands; diff --git a/database/migrations/2026062220_m06c_iot_messages.up.sql b/database/migrations/2026062220_m06c_iot_messages.up.sql new file mode 100644 index 0000000..85e4a3f --- /dev/null +++ b/database/migrations/2026062220_m06c_iot_messages.up.sql @@ -0,0 +1,89 @@ +CREATE TABLE IF NOT EXISTS qipai_iot_commands ( + id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, + tenant_id BIGINT UNSIGNED NOT NULL, + device_id BIGINT UNSIGNED NOT NULL, + store_id BIGINT UNSIGNED NOT NULL, + room_id BIGINT UNSIGNED NULL, + order_id BIGINT UNSIGNED NULL, + command_id VARCHAR(13) NOT NULL, + command_type VARCHAR(64) NOT NULL, + status VARCHAR(32) NOT NULL DEFAULT 'PENDING', + request_payload JSON NOT NULL, + response_payload JSON NULL, + trace_id VARCHAR(128) NOT NULL, + retry_count SMALLINT UNSIGNED NOT NULL DEFAULT 0, + published_at DATETIME(3) NULL, + acknowledged_at DATETIME(3) NULL, + expires_at DATETIME(3) NULL, + failure_code VARCHAR(64) NOT NULL DEFAULT '', + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) + ON UPDATE CURRENT_TIMESTAMP(3), + CONSTRAINT fk_qipai_iot_command_tenant + FOREIGN KEY (tenant_id) REFERENCES qipai_tenants(id), + CONSTRAINT fk_qipai_iot_command_device + FOREIGN KEY (device_id) REFERENCES qipai_devices(id), + CONSTRAINT fk_qipai_iot_command_store + FOREIGN KEY (store_id) REFERENCES qipai_stores(id), + CONSTRAINT fk_qipai_iot_command_room + FOREIGN KEY (room_id) REFERENCES qipai_rooms(id), + CONSTRAINT fk_qipai_iot_command_order + FOREIGN KEY (order_id) REFERENCES qipai_orders(id), + UNIQUE KEY uq_qipai_iot_command_id (tenant_id, command_id), + KEY idx_qipai_iot_command_status (tenant_id, status, expires_at), + KEY idx_qipai_iot_command_device (tenant_id, device_id, created_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS qipai_iot_device_events ( + id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, + tenant_id BIGINT UNSIGNED NOT NULL, + device_id BIGINT UNSIGNED NOT NULL, + store_id BIGINT UNSIGNED NOT NULL, + room_id BIGINT UNSIGNED NULL, + command_id VARCHAR(13) NULL, + topic VARCHAR(255) NOT NULL, + event_type VARCHAR(64) NOT NULL, + payload_hash CHAR(64) NOT NULL, + raw_payload JSON NOT NULL, + normalized_payload JSON NULL, + event_at DATETIME(3) NULL, + received_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + receive_count INT UNSIGNED NOT NULL DEFAULT 1, + processing_status VARCHAR(32) NOT NULL DEFAULT 'RECEIVED', + CONSTRAINT fk_qipai_iot_event_tenant + FOREIGN KEY (tenant_id) REFERENCES qipai_tenants(id), + CONSTRAINT fk_qipai_iot_event_device + FOREIGN KEY (device_id) REFERENCES qipai_devices(id), + CONSTRAINT fk_qipai_iot_event_store + FOREIGN KEY (store_id) REFERENCES qipai_stores(id), + CONSTRAINT fk_qipai_iot_event_room + FOREIGN KEY (room_id) REFERENCES qipai_rooms(id), + UNIQUE KEY uq_qipai_iot_event_dedup + (tenant_id, device_id, topic, payload_hash), + KEY idx_qipai_iot_event_command (tenant_id, command_id), + KEY idx_qipai_iot_event_status (tenant_id, processing_status, received_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS qipai_iot_dead_letters ( + id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, + tenant_id BIGINT UNSIGNED NULL, + device_id BIGINT UNSIGNED NULL, + topic VARCHAR(255) NOT NULL, + payload_hash CHAR(64) NOT NULL, + raw_payload MEDIUMTEXT NOT NULL, + error_code VARCHAR(64) NOT NULL, + error_message VARCHAR(500) NOT NULL, + receive_count INT UNSIGNED NOT NULL DEFAULT 1, + first_received_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + last_received_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + resolved_at DATETIME(3) NULL, + CONSTRAINT fk_qipai_iot_dead_tenant + FOREIGN KEY (tenant_id) REFERENCES qipai_tenants(id), + CONSTRAINT fk_qipai_iot_dead_device + FOREIGN KEY (device_id) REFERENCES qipai_devices(id), + UNIQUE KEY uq_qipai_iot_dead_dedup (topic, payload_hash), + KEY idx_qipai_iot_dead_open (resolved_at, last_received_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +INSERT IGNORE INTO qipai_schema_migrations (version, name) +VALUES ('2026062220', 'm06c_iot_messages'); diff --git a/database/migrations/2026062220_m06c_iot_messages.verify.sql b/database/migrations/2026062220_m06c_iot_messages.verify.sql new file mode 100644 index 0000000..a925ae6 --- /dev/null +++ b/database/migrations/2026062220_m06c_iot_messages.verify.sql @@ -0,0 +1,23 @@ +SELECT table_name FROM information_schema.tables +WHERE table_schema = DATABASE() AND table_name IN ( + 'qipai_iot_commands', 'qipai_iot_device_events', 'qipai_iot_dead_letters' +) ORDER BY table_name; + +SELECT table_name, index_name FROM information_schema.statistics +WHERE table_schema = DATABASE() + AND ((table_name = 'qipai_iot_commands' + AND index_name = 'uq_qipai_iot_command_id') + OR (table_name = 'qipai_iot_device_events' + AND index_name = 'uq_qipai_iot_event_dedup') + OR (table_name = 'qipai_iot_dead_letters' + AND index_name = 'uq_qipai_iot_dead_dedup')) +GROUP BY table_name, index_name ORDER BY table_name; + +SELECT column_name FROM information_schema.columns +WHERE table_schema = DATABASE() AND table_name = 'qipai_iot_commands' + AND column_name IN ( + 'command_id', 'status', 'request_payload', 'response_payload', + 'retry_count', 'published_at', 'acknowledged_at', 'expires_at', 'failure_code' + ) ORDER BY column_name; + +SELECT version, name FROM qipai_schema_migrations WHERE version = '2026062220'; diff --git a/scripts/dev/wsl/mysql-migration-roundtrip.sh b/scripts/dev/wsl/mysql-migration-roundtrip.sh index b88d40d..e5f2d4a 100644 --- a/scripts/dev/wsl/mysql-migration-roundtrip.sh +++ b/scripts/dev/wsl/mysql-migration-roundtrip.sh @@ -93,6 +93,6 @@ export QIPAI_MYSQL_USER="${username}" export QIPAI_MYSQL_PASSWORD="${password}" export QIPAI_MYSQL_CONNECTION_LIMIT=2 -echo "INFO: MySQL ${mysql_version}; running M01-B through M06-B migration roundtrip in a temporary database." +echo "INFO: MySQL ${mysql_version}; running M01-B through M06-C migration roundtrip in a temporary database." npm --prefix backend run test:mysql:migration -echo "PASS: M01-B through M06-B live MySQL migration roundtrip completed." +echo "PASS: M01-B through M06-C live MySQL migration roundtrip completed."