feat(M06-D): 接入控制箱与Sub-1G门锁业务控制

This commit is contained in:
Codex
2026-06-22 18:38:28 +08:00
parent f71ac09a0b
commit d15fd3f0ff
10 changed files with 685 additions and 9 deletions
+1 -1
View File
@@ -17,7 +17,7 @@
"db:migrate:verify": "npm run build && node dist/db/migrate-cli.js verify", "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", "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: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" "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"
}, },
"dependencies": { "dependencies": {
"@fastify/cors": "^11.2.0", "@fastify/cors": "^11.2.0",
+7
View File
@@ -42,6 +42,9 @@ import {
registerThirdPartyRoutes, type ThirdPartyRouteOptions registerThirdPartyRoutes, type ThirdPartyRouteOptions
} from './routes/third-party.js'; } from './routes/third-party.js';
import { registerDeviceRoutes, type DeviceRouteOptions } from './routes/devices.js'; import { registerDeviceRoutes, type DeviceRouteOptions } from './routes/devices.js';
import {
registerDeviceControlRoutes, type DeviceControlRouteOptions
} from './routes/device-control.js';
export interface BuildAppOptions { export interface BuildAppOptions {
config?: AppConfig; config?: AppConfig;
@@ -60,6 +63,7 @@ export interface BuildAppOptions {
thirdParty?: ThirdPartyRouteOptions; thirdParty?: ThirdPartyRouteOptions;
mqtt?: MqttHealthProvider; mqtt?: MqttHealthProvider;
devices?: DeviceRouteOptions; devices?: DeviceRouteOptions;
deviceControl?: DeviceControlRouteOptions;
} }
declare module 'fastify' { declare module 'fastify' {
@@ -146,6 +150,9 @@ export async function buildApp(options: BuildAppOptions = {}): Promise<FastifyIn
if (options.devices) { if (options.devices) {
await registerDeviceRoutes(app, options.devices); await registerDeviceRoutes(app, options.devices);
} }
if (options.deviceControl) {
await registerDeviceControlRoutes(app, options.deviceControl);
}
return app; return app;
} }
@@ -0,0 +1,170 @@
import type { RowDataPacket } from 'mysql2/promise';
import type { AccessProfile } from '../auth/rbac-repository.js';
import type { MySqlPool } from '../db/mysql.js';
import type { DeviceCommandService } from './device-command-service.js';
import {
JilianControlBoxAdapter,
JilianSub1GLockAdapter
} from './jilian-adapters.js';
interface DeviceRow extends RowDataPacket {
id: string; deviceId: string; storeId: string; roomId: string | null;
deviceType: 'CONTROL_BOX' | 'SUB_LOCK' | 'SMART_SOCKET';
status: string;
}
export class DeviceControlError extends Error {
constructor(public readonly code: string) { super(code); }
}
export class DeviceControlService {
private readonly controlBox = new JilianControlBoxAdapter();
private readonly subLock = new JilianSub1GLockAdapter();
constructor(
private readonly pool: MySqlPool,
private readonly commands: Pick<DeviceCommandService, 'issue'>
) {}
async controlPower(context: CommandContext, input: {
slot1?: 'on' | 'off'; slot2?: 'on' | 'off';
slot3?: 'on' | 'off'; slotall?: 'on' | 'off';
}) {
return this.issueControlBox(context, 'ConctolPower',
(id) => this.controlBox.controlPower({ id, ...input }));
}
async controlDoor(context: CommandContext, input: {
order: 'open' | 'close'; holdopen?: 0 | 1; delayTime?: number;
}) {
if (input.holdopen === 1 && !this.isManager(context.access)) {
throw new DeviceControlError('DEVICE_HOLD_OPEN_FORBIDDEN');
}
return this.issueControlBox(context, 'Crldoor',
(id) => this.controlBox.controlDoor({ id, ...input }));
}
async playTts(context: CommandContext, input: {
content: string; volume?: number; playCount?: number; priority?: number;
}) {
return this.issueControlBox(context, 'PlayTTS',
(id) => this.controlBox.playTts({ id, ...input }));
}
async stopTts(context: CommandContext) {
return this.issueControlBox(context, 'stopTTS', (id) => this.controlBox.stopTts(id));
}
async controlLed(context: CommandContext, minute: number) {
return this.issueControlBox(context, 'CrlLED',
(id) => this.controlBox.controlLed({ id, minute }));
}
async startTask(context: CommandContext, input: {
minute: number; type: 1 | 2 | 3; subID?: string;
holdopen?: 0 | 1; delayTime?: number;
}) {
return this.issueControlBox(context, 'task',
(id) => this.controlBox.startTask({ id, ...input }), context.orderId);
}
async extendTask(context: CommandContext, addminute: number) {
return this.issueControlBox(context, 'addtask',
(id) => this.controlBox.extendTask({ id, addminute }), context.orderId);
}
async cancelTask(context: CommandContext) {
return this.issueControlBox(context, 'canceltask',
(id) => this.controlBox.cancelTask(id), context.orderId);
}
async pairSubLock(context: CommandContext, timeout = 60) {
return this.issueControlBox(context, 'AddDevice',
(id) => this.subLock.pair({ id, timeout }));
}
async controlSubLock(context: CommandContext, input: {
subID: string;
order: 'open' | 'close' | 'setkey' | 'delkey' | 'setcard' | 'delcard' | 'factoryreset';
holdopen?: 0 | 1; delayTime?: number; content?: string;
dangerConfirmation?: string;
}) {
if (['factoryreset'].includes(input.order)) {
if (!context.access.roles.includes('PLATFORM_ADMIN')
|| input.dangerConfirmation !== 'CONFIRM_FACTORY_RESET') {
throw new DeviceControlError('DEVICE_DANGEROUS_ACTION_FORBIDDEN');
}
}
if (['delkey', 'delcard'].includes(input.order) && !input.content) {
if (!context.access.roles.includes('PLATFORM_ADMIN')
|| input.dangerConfirmation !== 'CONFIRM_CLEAR_CREDENTIALS') {
throw new DeviceControlError('DEVICE_DANGEROUS_ACTION_FORBIDDEN');
}
}
const { dangerConfirmation: _, ...vendorInput } = input;
return this.issueControlBox(context, 'CtrlDevice',
(id) => this.subLock.control({ id, ...vendorInput }));
}
private async issueControlBox(
context: CommandContext,
commandType: string,
payloadFactory: (id: string) => Record<string, unknown>,
orderId?: string | null
) {
this.assertWriteScope(context.access, context.storeId);
const device = await this.resolveControlBox(context);
if (device.status === 'OFFLINE') throw new DeviceControlError('DEVICE_OFFLINE');
return this.commands.issue({
tenantId: context.tenantId,
assetId: String(device.id),
deviceId: device.deviceId,
storeId: context.storeId,
roomId: context.roomId,
orderId,
commandType,
payloadFactory,
traceId: context.traceId,
expiresAt: context.expiresAt
});
}
private async resolveControlBox(context: CommandContext) {
const [rows] = await this.pool.execute<DeviceRow[]>(
`SELECT id, device_id AS deviceId, store_id AS storeId, room_id AS roomId,
device_type AS deviceType, status
FROM qipai_devices
WHERE tenant_id = ? AND store_id = ? AND room_id = ?
AND device_type = 'CONTROL_BOX' AND deleted_at IS NULL
ORDER BY id LIMIT 1`,
[context.tenantId, context.storeId, context.roomId]
);
if (!rows[0]) throw new DeviceControlError('CONTROL_BOX_NOT_BOUND');
return rows[0];
}
private assertWriteScope(access: AccessProfile, storeId: string) {
if (access.roles.includes('PLATFORM_ADMIN') || access.capabilities.includes('tenant.manage')) {
return;
}
if (!access.capabilities.includes('device.write') || !access.storeIds.includes(storeId)) {
throw new DeviceControlError('DEVICE_SCOPE_FORBIDDEN');
}
}
private isManager(access: AccessProfile) {
return access.roles.some((role) =>
['STORE_ADMIN', 'TENANT_ADMIN', 'PLATFORM_ADMIN'].includes(role)
);
}
}
export interface CommandContext {
tenantId: string;
storeId: string;
roomId: string;
orderId?: string | null;
traceId: string;
access: AccessProfile;
expiresAt?: Date | null;
}
+115 -2
View File
@@ -14,6 +14,9 @@ interface DeviceRow extends RowDataPacket {
deviceId: string; deviceType: 'CONTROL_BOX' | 'SUB_LOCK' | 'SMART_SOCKET'; deviceId: string; deviceType: 'CONTROL_BOX' | 'SUB_LOCK' | 'SMART_SOCKET';
} }
interface IdRow extends RowDataPacket { id: string } interface IdRow extends RowDataPacket { id: string }
interface CommandRow extends RowDataPacket {
id: string; commandType: string; storeId: string; roomId: string | null;
}
export class IotMessageService { export class IotMessageService {
private readonly adapters: Record<DeviceRow['deviceType'], ProtocolAdapter> = { private readonly adapters: Record<DeviceRow['deviceType'], ProtocolAdapter> = {
@@ -65,6 +68,8 @@ export class IotMessageService {
return; return;
} }
const safePayload = sanitizeSensitivePayload(rawPayload, normalized);
normalized = { ...normalized, payload: safePayload };
const [result] = await this.pool.execute<ResultSetHeader>( const [result] = await this.pool.execute<ResultSetHeader>(
`INSERT INTO qipai_iot_device_events `INSERT INTO qipai_iot_device_events
(tenant_id, device_id, store_id, room_id, command_id, topic, event_type, (tenant_id, device_id, store_id, room_id, command_id, topic, event_type,
@@ -74,12 +79,13 @@ export class IotMessageService {
received_at = UTC_TIMESTAMP(3)`, received_at = UTC_TIMESTAMP(3)`,
[device.tenantId, device.id, device.storeId, device.roomId, [device.tenantId, device.id, device.storeId, device.roomId,
normalized.commandId, topic, normalized.eventType, payloadHash, normalized.commandId, topic, normalized.eventType, payloadHash,
JSON.stringify(rawPayload), JSON.stringify(normalized), JSON.stringify(safePayload), JSON.stringify(normalized),
normalized.eventAt] normalized.eventAt]
); );
if (result.affectedRows !== 1) return; if (result.affectedRows !== 1) return;
await this.updateDevice(device, normalized, rawPayload); await this.updateDevice(device, normalized, safePayload);
await this.applyAlerts(device, normalized);
if (normalized.kind === 'ACK' && normalized.commandId) { if (normalized.kind === 'ACK' && normalized.commandId) {
await this.applyAcknowledgement(device, normalized); await this.applyAcknowledgement(device, normalized);
} }
@@ -145,6 +151,80 @@ export class IotMessageService {
[status, JSON.stringify(message.payload), failureCode, [status, JSON.stringify(message.payload), failureCode,
device.tenantId, device.id, message.commandId] device.tenantId, device.id, message.commandId]
); );
if (successful && message.eventType === 'AddDevice') {
await this.persistPairedSubLock(device, message);
}
}
private async persistPairedSubLock(device: DeviceRow, message: NormalizedVendorMessage) {
const subId = readString(message.payload.subID ?? message.payload.subId);
const subtype = readString(message.payload.subtype);
if (!subId || !subtype || !message.commandId) return;
const [commands] = await this.pool.execute<CommandRow[]>(
`SELECT id, command_type AS commandType, store_id AS storeId, room_id AS roomId
FROM qipai_iot_commands
WHERE tenant_id = ? AND device_id = ? AND command_id = ?`,
[device.tenantId, device.id, message.commandId]
);
const command = commands[0];
if (!command?.roomId || command.commandType !== 'AddDevice') return;
const childDeviceId = `${device.deviceId}_SUB_${subId}`.slice(0, 64);
await this.pool.execute(
`INSERT INTO qipai_devices
(tenant_id, store_id, room_id, device_id, device_type, model, capabilities, status)
VALUES (?, ?, ?, ?, 'SUB_LOCK', ?, JSON_ARRAY('LOCK'), 'ONLINE')
ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id), room_id = VALUES(room_id),
model = VALUES(model), status = 'ONLINE'`,
[device.tenantId, command.storeId, command.roomId, childDeviceId,
subtype === '14' ? '701C' : subtype === '15' ? '701G' : `SUBTYPE_${subtype}`]
);
const [childRows] = await this.pool.execute<IdRow[]>(
`SELECT id FROM qipai_devices
WHERE tenant_id = ? AND device_id = ? AND deleted_at IS NULL`,
[device.tenantId, childDeviceId]
);
if (!childRows[0]) return;
await this.pool.execute(
`INSERT INTO qipai_device_links
(tenant_id, parent_device_id, child_device_id, store_id, room_id,
link_type, sub_id, subtype)
VALUES (?, ?, ?, ?, ?, 'SUB_1G', ?, ?)
ON DUPLICATE KEY UPDATE parent_device_id = VALUES(parent_device_id),
child_device_id = VALUES(child_device_id), room_id = VALUES(room_id),
sub_id = VALUES(sub_id), subtype = VALUES(subtype), status = 'BOUND'`,
[device.tenantId, device.id, childRows[0].id, command.storeId,
command.roomId, subId, subtype]
);
}
private async applyAlerts(device: DeviceRow, message: NormalizedVendorMessage) {
const alertResult = ['timeout', 'full', 'unconfirm'].includes(message.result ?? '')
? `DEVICE_${message.result?.toUpperCase()}`
: null;
if (alertResult) {
await this.upsertAlert(device, alertResult,
message.result === 'unconfirm' ? 'HIGH' : 'MEDIUM',
`Device command returned ${message.result}.`);
}
const battery = readNumber(message.payload.battery);
if (battery !== null && battery <= 20) {
await this.upsertAlert(device, 'LOW_BATTERY', battery <= 10 ? 'HIGH' : 'MEDIUM',
`Device battery is ${battery}%.`);
}
}
private async upsertAlert(
device: DeviceRow, alertType: string, severity: string, summary: string
) {
await this.pool.execute(
`INSERT INTO qipai_device_alerts
(tenant_id, device_id, store_id, room_id, alert_type, severity, summary)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE severity = VALUES(severity), summary = VALUES(summary),
last_seen_at = UTC_TIMESTAMP(3)`,
[device.tenantId, device.id, device.storeId, device.roomId,
alertType, severity, summary]
);
} }
private async updateDevice( private async updateDevice(
@@ -207,3 +287,36 @@ function readDeviceId(record: Record<string, unknown>): string | null {
const value = record.DeviceID ?? record.deviceID; const value = record.DeviceID ?? record.deviceID;
return typeof value === 'string' ? value : null; return typeof value === 'string' ? value : null;
} }
function sanitizeSensitivePayload(
payload: unknown, normalized: NormalizedVendorMessage
): Record<string, unknown> {
const record = { ...zRecord(payload) };
if (normalized.eventType === 'record' && typeof record.content === 'string') {
const content = record.content;
record.contentHash = createHash('sha256').update(content).digest('hex');
record.contentMasked = content.length <= 4
? '*'.repeat(content.length)
: `${content.slice(0, 2)}${'*'.repeat(Math.min(8, content.length - 4))}${content.slice(-2)}`;
delete record.content;
}
if (typeof record.password === 'string') {
record.password = '<redacted>';
}
if (typeof record.card === 'string') {
record.card = '<redacted>';
}
return record;
}
function readString(value: unknown): string | null {
return typeof value === 'string' && value.length > 0 ? value : null;
}
function readNumber(value: unknown): number | null {
if (typeof value === 'number' && Number.isFinite(value)) return value;
if (typeof value === 'string' && value.trim() !== '' && Number.isFinite(Number(value))) {
return Number(value);
}
return null;
}
+153
View File
@@ -0,0 +1,153 @@
import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { z } from 'zod';
import type { AuthRepository } from '../auth/auth-repository.js';
import { authenticateAccessToken } from '../auth/authenticate.js';
import type { AccessProfile } from '../auth/rbac-repository.js';
import {
DeviceControlError,
type CommandContext,
type DeviceControlService
} from '../devices/device-control-service.js';
const id = z.string().regex(/^[1-9]\d{0,19}$/);
const contextSchema = z.object({
storeId: id,
roomId: id,
orderId: id.nullable().optional()
});
const powerSchema = contextSchema.extend({
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);
const doorSchema = contextSchema.extend({
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)
});
const ttsSchema = contextSchema.extend({
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)
});
const minuteSchema = contextSchema.extend({
minute: z.number().int().min(0).max(10080)
});
const taskSchema = contextSchema.extend({
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)
});
const extendSchema = contextSchema.extend({
addminute: z.number().int().min(1).max(10080)
});
const pairSchema = contextSchema.extend({
timeout: z.number().int().min(10).max(300).default(60)
});
const subLockSchema = contextSchema.extend({
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(),
dangerConfirmation: z.string().max(64).optional()
});
export interface DeviceControlRouteOptions {
service: Pick<DeviceControlService,
'controlPower' | 'controlDoor' | 'playTts' | 'stopTts' | 'controlLed'
| 'startTask' | 'extendTask' | 'cancelTask' | 'pairSubLock' | 'controlSubLock'>;
authRepository: Pick<AuthRepository, 'validateSession'>;
accessControl: { getAccessProfile(tenantId: string, userId: string): Promise<AccessProfile> };
jwtSecret: string;
}
export async function registerDeviceControlRoutes(
app: FastifyInstance, options: DeviceControlRouteOptions
) {
register('/admin-api/device-control/power', powerSchema,
(context, body) => options.service.controlPower(context, body));
register('/admin-api/device-control/door', doorSchema,
(context, body) => options.service.controlDoor(context, body));
register('/admin-api/device-control/tts', ttsSchema,
(context, body) => options.service.playTts(context, body));
register('/admin-api/device-control/tts/stop', contextSchema,
(context) => options.service.stopTts(context));
register('/admin-api/device-control/led', minuteSchema,
(context, body) => options.service.controlLed(context, body.minute));
register('/admin-api/device-control/task/start', taskSchema,
(context, body) => options.service.startTask(context, body));
register('/admin-api/device-control/task/extend', extendSchema,
(context, body) => options.service.extendTask(context, body.addminute));
register('/admin-api/device-control/task/cancel', contextSchema,
(context) => options.service.cancelTask(context));
register('/admin-api/device-control/sub-lock/pair', pairSchema,
(context, body) => options.service.pairSubLock(context, body.timeout));
register('/admin-api/device-control/sub-lock/action', subLockSchema,
(context, body) => options.service.controlSubLock(context, body));
function register<T extends z.ZodTypeAny>(
url: string,
schema: T,
handler: (context: CommandContext, body: z.infer<T>) => Promise<unknown>
) {
app.post(url, async (request, reply) => {
const parsed = schema.safeParse(request.body);
if (!parsed.success) return invalid(reply, request.traceId);
const context = await requireContext(request, reply, options, parsed.data);
if (!context) return;
try {
return {
code: 0, data: await handler(context, parsed.data), traceId: request.traceId
};
} catch (error) {
if (!(error instanceof DeviceControlError)) throw error;
const status = error.code.endsWith('_FORBIDDEN') ? 403
: error.code === 'DEVICE_OFFLINE' ? 409 : 400;
return reply.status(status).send({
code: error.code, message: 'Device control was rejected.', traceId: request.traceId
});
}
});
}
}
async function requireContext(
request: FastifyRequest, reply: FastifyReply, options: DeviceControlRouteOptions,
input: { storeId: string; roomId: string; orderId?: string | null }
) {
const auth = await authenticateAccessToken(
request.headers.authorization, options.authRepository, options.jwtSecret
);
if (!auth) {
reply.status(401).send({
code: 'AUTH_SESSION_INVALID', message: 'Authentication required.',
traceId: request.traceId
});
return null;
}
const access = await options.accessControl.getAccessProfile(
auth.session.tenantId, auth.session.user.id
);
return {
tenantId: auth.session.tenantId,
storeId: input.storeId,
roomId: input.roomId,
orderId: input.orderId,
traceId: request.traceId,
access
};
}
function invalid(reply: FastifyReply, traceId: string) {
return reply.status(400).send({
code: 'INVALID_DEVICE_CONTROL_REQUEST',
message: 'The device control request is invalid.', traceId
});
}
+9
View File
@@ -29,6 +29,8 @@ import { ThirdPartyService } from './third-party/third-party-service.js';
import { MqttService } from './mqtt/mqtt-service.js'; import { MqttService } from './mqtt/mqtt-service.js';
import { DeviceRepository } from './devices/device-repository.js'; import { DeviceRepository } from './devices/device-repository.js';
import { IotMessageService } from './devices/iot-message-service.js'; import { IotMessageService } from './devices/iot-message-service.js';
import { DeviceCommandService } from './devices/device-command-service.js';
import { DeviceControlService } from './devices/device-control-service.js';
const config = loadConfig(); const config = loadConfig();
const pool = createMySqlPool(config); const pool = createMySqlPool(config);
@@ -43,6 +45,7 @@ const iotMessages = new IotMessageService(pool);
const mqtt = new MqttService(config.mqtt, undefined, (topic, payload) => const mqtt = new MqttService(config.mqtt, undefined, (topic, payload) =>
iotMessages.handle(topic, payload) iotMessages.handle(topic, payload)
); );
const deviceCommands = new DeviceCommandService(iotMessages, mqtt);
const app = await buildApp({ const app = await buildApp({
config, config,
mqtt, mqtt,
@@ -144,6 +147,12 @@ const app = await buildApp({
authRepository, authRepository,
accessControl, accessControl,
jwtSecret: config.auth.jwtSecret jwtSecret: config.auth.jwtSecret
},
deviceControl: {
service: new DeviceControlService(pool, deviceCommands),
authRepository,
accessControl,
jwtSecret: config.auth.jwtSecret
} }
}); });
app.addHook('onClose', async () => { app.addHook('onClose', async () => {
+138
View File
@@ -0,0 +1,138 @@
import assert from 'node:assert/strict';
import { buildApp } from '../dist/app.js';
import { signAccessToken } from '../dist/auth/jwt.js';
import {
DeviceControlError,
DeviceControlService
} from '../dist/devices/device-control-service.js';
const managerAccess = {
roles: ['STORE_ADMIN'],
capabilities: ['device.read', 'device.write'],
storeIds: ['11']
};
const issued = [];
const service = new DeviceControlService({
async execute(sql) {
if (sql.includes('FROM qipai_devices')) {
return [[{
id: 51, deviceId: 'BOX_001', storeId: 11, roomId: 31,
deviceType: 'CONTROL_BOX', status: 'ONLINE'
}], []];
}
return [[], []];
}
}, {
async issue(input) {
const payload = input.payloadFactory('1782120000001');
issued.push({ ...input, payload });
return { commandId: '1782120000001', status: 'PUBLISHED' };
}
});
const context = {
tenantId: '7', storeId: '11', roomId: '31', traceId: 'trace',
access: managerAccess
};
await service.controlPower(context, { slot1: 'on', slot3: 'off' });
assert.equal(issued.at(-1).payload.action, 'ConctolPower');
await service.controlDoor(context, { order: 'open', holdopen: 1, delayTime: 4 });
assert.equal(issued.at(-1).payload.action, 'Crldoor');
await service.playTts(context, { content: '欢迎光临' });
assert.equal(issued.at(-1).payload.action, 'PlayTTS');
await service.stopTts(context);
assert.equal(issued.at(-1).payload.action, 'stopTTS');
await service.controlLed(context, 30);
assert.equal(issued.at(-1).payload.action, 'CrlLED');
await service.startTask(context, { minute: 120, type: 2, subID: 'SUB001' });
assert.equal(issued.at(-1).payload.action, 'task');
await service.extendTask(context, 30);
assert.equal(issued.at(-1).payload.action, 'addtask');
await service.cancelTask(context);
assert.equal(issued.at(-1).payload.action, 'canceltask');
await service.pairSubLock(context, 60);
assert.equal(issued.at(-1).payload.action, 'AddDevice');
await service.controlSubLock(context, {
subID: 'SUB001', order: 'open', delayTime: 4
});
assert.equal(issued.at(-1).payload.action, 'CtrlDevice');
await assert.rejects(
() => service.controlSubLock(context, {
subID: 'SUB001', order: 'factoryreset', dangerConfirmation: 'CONFIRM_FACTORY_RESET'
}),
(error) => error instanceof DeviceControlError
&& error.code === 'DEVICE_DANGEROUS_ACTION_FORBIDDEN'
);
const platformContext = {
...context,
access: { roles: ['PLATFORM_ADMIN'], capabilities: [], storeIds: [] }
};
await service.controlSubLock(platformContext, {
subID: 'SUB001', order: 'factoryreset', dangerConfirmation: 'CONFIRM_FACTORY_RESET'
});
assert.equal(issued.at(-1).payload.order, 'factoryreset');
assert.equal('dangerConfirmation' in issued.at(-1).payload, false);
await assert.rejects(
() => service.controlPower({
...context,
access: { roles: ['STAFF'], capabilities: ['device.write'], storeIds: ['12'] }
}, { slot1: 'on' }),
(error) => error.code === 'DEVICE_SCOPE_FORBIDDEN'
);
const secret = 'test-only-jwt-secret-with-at-least-32-characters';
const token = signAccessToken({
sub: '22', sid: '5c4d3af8-c63c-4edb-bf95-b84127bb3f6e',
tid: '7', aid: '9', rv: 1
}, secret, 900);
let routed;
const app = await buildApp({
deviceControl: {
jwtSecret: secret,
authRepository: {
async validateSession() {
return {
id: '5c4d3af8-c63c-4edb-bf95-b84127bb3f6e',
tenantId: '7', platformAppId: '9', expiresAt: new Date(Date.now() + 60000),
user: {
id: '22', tenantId: '7', userType: 'STAFF', status: 'ACTIVE',
roleVersion: 1, nickname: '', avatarUrl: '', phone: ''
}
};
}
},
accessControl: { async getAccessProfile() { return managerAccess; } },
service: {
async controlPower(_context, input) { routed = input; return { status: 'PUBLISHED' }; },
async controlDoor() { return {}; },
async playTts() { return {}; },
async stopTts() { return {}; },
async controlLed() { return {}; },
async startTask() { return {}; },
async extendTask() { return {}; },
async cancelTask() { return {}; },
async pairSubLock() { return {}; },
async controlSubLock() { return {}; }
}
}
});
const response = await app.inject({
method: 'POST',
url: '/admin-api/device-control/power',
headers: { authorization: `Bearer ${token}` },
payload: { storeId: '11', roomId: '31', slot1: 'on' }
});
assert.equal(response.statusCode, 200);
assert.equal(routed.slot1, 'on');
const invalid = await app.inject({
method: 'POST',
url: '/admin-api/device-control/door',
headers: { authorization: `Bearer ${token}` },
payload: { storeId: '11', roomId: '31', order: 'open', delayTime: 15 }
});
assert.equal(invalid.statusCode, 400);
await app.close();
console.log('PASS: M06-D control-box and Sub-1G command gates and routes work.');
+29 -3
View File
@@ -86,7 +86,7 @@ assert.equal(generateCommandId(1782120000000, 7), '1782120000007');
assert.match(generateCommandId(), /^\d{13}$/); assert.match(generateCommandId(), /^\d{13}$/);
const calls = []; const calls = [];
let eventInsertCount = 0; const eventHashes = new Set();
const service = new IotMessageService({ const service = new IotMessageService({
async execute(sql, params) { async execute(sql, params) {
calls.push({ sql, params }); calls.push({ sql, params });
@@ -97,8 +97,10 @@ const service = new IotMessageService({
}], []]; }], []];
} }
if (sql.includes('INSERT INTO qipai_iot_device_events')) { if (sql.includes('INSERT INTO qipai_iot_device_events')) {
eventInsertCount += 1; const hash = params[7];
return [{ affectedRows: eventInsertCount === 1 ? 1 : 2 }, []]; const duplicate = eventHashes.has(hash);
eventHashes.add(hash);
return [{ affectedRows: duplicate ? 2 : 1 }, []];
} }
return [{ affectedRows: 1, insertId: 81 }, []]; return [{ affectedRows: 1, insertId: 81 }, []];
} }
@@ -125,6 +127,30 @@ assert.equal(
true true
); );
const recordPayload = Buffer.from(JSON.stringify({
DeviceID: 'BOX_001', event: 'record', type: 'card',
state: 'open', content: 'CARD-PRIVATE-1234', battery: 9
}));
await service.handle('/devicesend/BOX_001', recordPayload);
const recordInsert = calls.findLast((item) =>
item.sql.includes('INSERT INTO qipai_iot_device_events')
);
assert.equal(recordInsert.params.some((value) =>
typeof value === 'string' && value.includes('CARD-PRIVATE-1234')
), false);
assert.equal(calls.some((item) =>
item.sql.includes('INSERT INTO qipai_device_alerts')
&& item.params.includes('LOW_BATTERY')
), true);
await service.handle('/devicesend/BOX_001', Buffer.from(JSON.stringify({
DeviceID: 'BOX_001', id: '124', action: 'task', result: 'unconfirm'
})));
assert.equal(calls.some((item) =>
item.sql.includes('INSERT INTO qipai_device_alerts')
&& item.params.includes('DEVICE_UNCONFIRM')
), true);
await assert.rejects( await assert.rejects(
() => service.createCommand({ () => service.createCommand({
tenantId: '7', assetId: '51', storeId: '11', tenantId: '7', assetId: '51', storeId: '11',
@@ -1727,6 +1727,62 @@ async function assertIotMessages(pool, context) {
FROM qipai_iot_dead_letters WHERE topic = '/invalid/topic'` FROM qipai_iot_dead_letters WHERE topic = '/invalid/topic'`
); );
assert.deepEqual(deadRows, [{ errorCode: 'MQTT_TOPIC_INVALID', receiveCount: 1 }]); assert.deepEqual(deadRows, [{ errorCode: 'MQTT_TOPIC_INVALID', receiveCount: 1 }]);
await service.handle(`/devicesend/${device.deviceId}`, Buffer.from(JSON.stringify({
DeviceID: device.deviceId,
event: 'record',
type: 'card',
state: 'open',
content: 'PRIVATE-CARD-001',
battery: 9,
timestamp: 1782120001
})));
const [recordRows] = await pool.query(
`SELECT JSON_UNQUOTE(JSON_EXTRACT(raw_payload, '$.content')) AS content,
JSON_UNQUOTE(JSON_EXTRACT(raw_payload, '$.contentHash')) AS contentHash
FROM qipai_iot_device_events
WHERE tenant_id = ? AND event_type = 'record'`,
[context.tenantId]
);
assert.equal(recordRows[0].content, null);
assert.match(recordRows[0].contentHash, /^[a-f0-9]{64}$/);
const [alertRows] = await pool.query(
`SELECT alert_type AS alertType, severity
FROM qipai_device_alerts
WHERE tenant_id = ? AND device_id = ? AND status = 'OPEN'`,
[context.tenantId, device.id]
);
assert.deepEqual(alertRows, [{ alertType: 'LOW_BATTERY', severity: 'HIGH' }]);
await service.createCommand({
tenantId: context.tenantId,
assetId: String(device.id),
storeId: String(device.storeId),
roomId: String(device.roomId),
commandId: '1782120000002',
commandType: 'AddDevice',
payload: { action: 'AddDevice', id: '1782120000002', timeout: 60 },
traceId: 'm06d-pair-test'
});
await service.markPublished(context.tenantId, '1782120000002');
await service.handle(`/devicesend/${device.deviceId}`, Buffer.from(JSON.stringify({
DeviceID: device.deviceId,
id: '1782120000002',
action: 'AddDevice',
result: 'ok',
subID: 'SUB-AUTO-001',
subtype: '14',
timestamp: 1782120002
})));
const [linkRows] = await pool.query(
`SELECT l.sub_id AS subId, l.subtype, d.model
FROM qipai_device_links l
INNER JOIN qipai_devices d
ON d.tenant_id = l.tenant_id AND d.id = l.child_device_id
WHERE l.tenant_id = ? AND l.parent_device_id = ? AND l.sub_id = 'SUB-AUTO-001'`,
[context.tenantId, device.id]
);
assert.deepEqual(linkRows, [{ subId: 'SUB-AUTO-001', subtype: '14', model: '701C' }]);
} }
const config = loadConfig(); const config = loadConfig();
@@ -1797,7 +1853,7 @@ try {
await executeMigrationPlan(pool, plans.down); await executeMigrationPlan(pool, plans.down);
assert.deepEqual(await readCoreTables(pool), []); assert.deepEqual(await readCoreTables(pool), []);
await assertLegacyCompatibility(pool); await assertLegacyCompatibility(pool);
console.log('PASS: down removed all M01-B through M06-C tables.'); console.log('PASS: down removed all M01-B through M06-C migration tables.');
await executeMigrationPlan(pool, plans.up); await executeMigrationPlan(pool, plans.up);
await executeMigrationPlan(pool, plans.verify); await executeMigrationPlan(pool, plans.verify);
@@ -1925,6 +1981,10 @@ try {
'QoS 1 duplicate event receive count', 'QoS 1 duplicate event receive count',
'ACK correlation without duplicate side effects', 'ACK correlation without duplicate side effects',
'invalid Topic dead-letter persistence' 'invalid Topic dead-letter persistence'
,
'door record credential hashing and masking',
'low-battery alert upsert',
'AddDevice ACK creates 701C parent-child topology'
] ]
}, null, 2)); }, null, 2));
} finally { } finally {
+2 -2
View File
@@ -93,6 +93,6 @@ export QIPAI_MYSQL_USER="${username}"
export QIPAI_MYSQL_PASSWORD="${password}" export QIPAI_MYSQL_PASSWORD="${password}"
export QIPAI_MYSQL_CONNECTION_LIMIT=2 export QIPAI_MYSQL_CONNECTION_LIMIT=2
echo "INFO: MySQL ${mysql_version}; running M01-B through M06-C migration roundtrip in a temporary database." echo "INFO: MySQL ${mysql_version}; running M01-B through M06-D integration roundtrip in a temporary database."
npm --prefix backend run test:mysql:migration npm --prefix backend run test:mysql:migration
echo "PASS: M01-B through M06-C live MySQL migration roundtrip completed." echo "PASS: M01-B through M06-D live MySQL integration roundtrip completed."