feat(M06-B): 建立设备资产与拓扑管理

2026-06-22 17:40:54 +08:00
parent 07790a7924
commit 74a67ac92d
16 changed files with 1026 additions and 17 deletions
@@ -0,0 +1,366 @@
+import type { PoolConnection, ResultSetHeader, RowDataPacket } from 'mysql2/promise';
+import type { ManagementActor } from '../auth/user-management-repository.js';
+import type { MySqlPool } from '../db/mysql.js';
+
+export type DeviceType = 'CONTROL_BOX' | 'SUB_LOCK' | 'SMART_SOCKET';
+
+export interface DeviceAssetInput {
+  storeId: string;
+  roomId?: string | null;
+  deviceId: string;
+  imei?: string;
+  iccid?: string | null;
+  deviceType: DeviceType;
+  model: string;
+  firmwareVersion: string;
+  signalStrength?: number | null;
+  capabilities: string[];
+}
+
+interface IdRow extends RowDataPacket { id: string }
+interface DeviceRow extends RowDataPacket {
+  id: string; storeId: string; roomId: string | null; deviceId: string; imei: string;
+  iccid: string | null; deviceType: DeviceType; model: string; firmwareVersion: string;
+  status: string; signalStrength: number | null; capabilities: string | string[] | null;
+  stateSnapshot: string | Record<string, unknown> | null; lastSeenAt: Date | null;
+  lastHeartbeatAt: Date | null; maintenanceStatus: string;
+}
+
+export class DeviceError extends Error {
+  constructor(public readonly code: string) { super(code); }
+}
+
+export class DeviceRepository {
+  constructor(private readonly pool: MySqlPool) {}
+
+  async createAsset(actor: ManagementActor, input: DeviceAssetInput) {
+    this.assertStoreScope(actor, input.storeId, true);
+    return this.transaction(async (connection) => {
+      await this.assertStoreAndRoom(connection, actor, input.storeId, input.roomId ?? null);
+      try {
+        const [result] = await connection.execute<ResultSetHeader>(
+          `INSERT INTO qipai_devices
+            (tenant_id, store_id, room_id, device_id, imei, iccid, device_type, model,
+             firmware_version, signal_strength, capabilities)
+           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+          [actor.tenantId, input.storeId, input.roomId ?? null, input.deviceId,
+            input.imei ?? '', input.iccid || null, input.deviceType, input.model,
+            input.firmwareVersion, input.signalStrength ?? null,
+            JSON.stringify([...new Set(input.capabilities)].sort())]
+        );
+        const assetId = String(result.insertId);
+        await this.audit(connection, actor, 'DEVICE_ASSET_CREATED', assetId, {
+          deviceType: input.deviceType, storeId: input.storeId, roomId: input.roomId ?? null
+        });
+        return { assetId };
+      } catch (error) {
+        if (isDuplicate(error)) throw new DeviceError('DEVICE_IDENTITY_CONFLICT');
+        throw error;
+      }
+    });
+  }
+
+  async listAssets(actor: ManagementActor, storeId?: string) {
+    if (storeId) this.assertStoreScope(actor, storeId, false);
+    const scope = this.scope(actor, 'd.store_id');
+    const params: Array<string> = [actor.tenantId, ...scope.params];
+    const storeFilter = storeId ? ' AND d.store_id = ?' : '';
+    if (storeId) params.push(storeId);
+    const [rows] = await this.pool.execute<DeviceRow[]>(
+      `SELECT d.id, d.store_id AS storeId, d.room_id AS roomId, d.device_id AS deviceId,
+              d.imei, d.iccid, d.device_type AS deviceType, d.model,
+              d.firmware_version AS firmwareVersion, d.status, d.signal_strength AS signalStrength,
+              d.capabilities, d.state_snapshot AS stateSnapshot,
+              d.last_seen_at AS lastSeenAt, d.last_heartbeat_at AS lastHeartbeatAt,
+              d.maintenance_status AS maintenanceStatus
+       FROM qipai_devices d
+       WHERE d.tenant_id = ? AND d.deleted_at IS NULL AND ${scope.sql}${storeFilter}
+       ORDER BY d.store_id, d.room_id, d.id`,
+      params
+    );
+    return rows.map((row) => ({
+      ...row,
+      id: String(row.id),
+      storeId: String(row.storeId),
+      roomId: row.roomId === null ? null : String(row.roomId),
+      capabilities: parseJson<string[]>(row.capabilities, []),
+      stateSnapshot: parseJson<Record<string, unknown>>(row.stateSnapshot, {})
+    }));
+  }
+
+  async getTopology(actor: ManagementActor, storeId: string) {
+    this.assertStoreScope(actor, storeId, false);
+    const [assets, channelsResult, linksResult, alertsResult, maintenanceResult] = await Promise.all([
+      this.listAssets(actor, storeId),
+      this.pool.execute<RowDataPacket[]>(
+        `SELECT id, device_id AS assetId, room_id AS roomId, channel_code AS channelCode,
+                purpose, target_key AS targetKey, enabled
+         FROM qipai_device_channels
+         WHERE tenant_id = ? AND store_id = ? ORDER BY device_id, channel_code`,
+        [actor.tenantId, storeId]
+      ),
+      this.pool.execute<RowDataPacket[]>(
+        `SELECT id, parent_device_id AS parentAssetId, child_device_id AS childAssetId,
+                room_id AS roomId, link_type AS linkType, sub_id AS subId, subtype, status
+         FROM qipai_device_links
+         WHERE tenant_id = ? AND store_id = ? ORDER BY parent_device_id, child_device_id`,
+        [actor.tenantId, storeId]
+      ),
+      this.pool.execute<RowDataPacket[]>(
+        `SELECT id, device_id AS assetId, room_id AS roomId, alert_type AS alertType,
+                severity, status, summary, first_seen_at AS firstSeenAt,
+                last_seen_at AS lastSeenAt
+         FROM qipai_device_alerts
+         WHERE tenant_id = ? AND store_id = ? AND status = 'OPEN'
+         ORDER BY severity DESC, last_seen_at DESC`,
+        [actor.tenantId, storeId]
+      ),
+      this.pool.execute<RowDataPacket[]>(
+        `SELECT id, device_id AS assetId, room_id AS roomId, record_type AS recordType,
+                status, description, created_at AS createdAt, resolved_at AS resolvedAt
+         FROM qipai_device_maintenance_records
+         WHERE tenant_id = ? AND store_id = ?
+         ORDER BY created_at DESC LIMIT 200`,
+        [actor.tenantId, storeId]
+      )
+    ]);
+    return {
+      assets,
+      channels: normalizeIds(channelsResult[0]),
+      links: normalizeIds(linksResult[0]),
+      openAlerts: normalizeIds(alertsResult[0]),
+      maintenance: normalizeIds(maintenanceResult[0])
+    };
+  }
+
+  async bindChannel(actor: ManagementActor, input: {
+    assetId: string; storeId: string; roomId: string; channelCode: string; purpose: string;
+  }) {
+    this.assertStoreScope(actor, input.storeId, true);
+    return this.transaction(async (connection) => {
+      const device = await this.lockDevice(connection, actor, input.assetId, input.storeId);
+      if (!['CONTROL_BOX', 'SMART_SOCKET'].includes(device.deviceType)) {
+        throw new DeviceError('DEVICE_CHANNEL_UNSUPPORTED');
+      }
+      await this.assertStoreAndRoom(connection, actor, input.storeId, input.roomId);
+      const targetKey = `room:${input.roomId}:target:${input.purpose}`;
+      try {
+        await connection.execute(
+          `INSERT INTO qipai_device_channels
+            (tenant_id, device_id, store_id, room_id, channel_code, purpose, target_key)
+           VALUES (?, ?, ?, ?, ?, ?, ?)`,
+          [actor.tenantId, input.assetId, input.storeId, input.roomId,
+            input.channelCode, input.purpose, targetKey]
+        );
+      } catch (error) {
+        if (isDuplicate(error)) throw new DeviceError('DEVICE_CONTROL_TARGET_CONFLICT');
+        throw error;
+      }
+      await connection.execute(
+        'UPDATE qipai_devices SET room_id = ? WHERE tenant_id = ? AND id = ?',
+        [input.roomId, actor.tenantId, input.assetId]
+      );
+      await this.audit(connection, actor, 'DEVICE_CHANNEL_BOUND', input.assetId, {
+        roomId: input.roomId, channelCode: input.channelCode, purpose: input.purpose
+      });
+      return { assetId: input.assetId, targetKey };
+    });
+  }
+
+  async bindSubDevice(actor: ManagementActor, input: {
+    parentAssetId: string; childAssetId: string; storeId: string; roomId: string;
+    subId: string; subtype: string;
+  }) {
+    this.assertStoreScope(actor, input.storeId, true);
+    return this.transaction(async (connection) => {
+      const parent = await this.lockDevice(connection, actor, input.parentAssetId, input.storeId);
+      const child = await this.lockDevice(connection, actor, input.childAssetId, input.storeId);
+      if (parent.deviceType !== 'CONTROL_BOX' || child.deviceType !== 'SUB_LOCK') {
+        throw new DeviceError('DEVICE_LINK_TYPE_INVALID');
+      }
+      await this.assertStoreAndRoom(connection, actor, input.storeId, input.roomId);
+      try {
+        await connection.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', ?, ?)`,
+          [actor.tenantId, input.parentAssetId, input.childAssetId, input.storeId,
+            input.roomId, input.subId, input.subtype]
+        );
+      } catch (error) {
+        if (isDuplicate(error)) throw new DeviceError('DEVICE_LINK_CONFLICT');
+        throw error;
+      }
+      await connection.execute(
+        'UPDATE qipai_devices SET room_id = ? WHERE tenant_id = ? AND id = ?',
+        [input.roomId, actor.tenantId, input.childAssetId]
+      );
+      await this.audit(connection, actor, 'SUB_DEVICE_BOUND', input.childAssetId, {
+        parentAssetId: input.parentAssetId, roomId: input.roomId,
+        subId: maskIdentifier(input.subId), subtype: input.subtype
+      });
+      return { childAssetId: input.childAssetId, parentAssetId: input.parentAssetId };
+    });
+  }
+
+  async recordStatus(actor: ManagementActor, input: {
+    assetId: string; storeId: string; onlineStatus: 'ONLINE' | 'OFFLINE' | 'FAULT';
+    signalStrength?: number | null; firmwareVersion?: string; snapshot: Record<string, unknown>;
+  }) {
+    this.assertStoreScope(actor, input.storeId, true);
+    return this.transaction(async (connection) => {
+      await this.lockDevice(connection, actor, input.assetId, input.storeId);
+      const capturedAt = new Date();
+      await connection.execute(
+        `UPDATE qipai_devices SET status = ?, signal_strength = ?,
+          firmware_version = COALESCE(NULLIF(?, ''), firmware_version),
+          state_snapshot = ?, last_seen_at = ?, last_heartbeat_at = ?
+         WHERE tenant_id = ? AND id = ?`,
+        [input.onlineStatus, input.signalStrength ?? null, input.firmwareVersion ?? '',
+          JSON.stringify(input.snapshot), capturedAt, capturedAt, actor.tenantId, input.assetId]
+      );
+      await connection.execute(
+        `INSERT INTO qipai_device_status_snapshots
+          (tenant_id, device_id, online_status, signal_strength, firmware_version, snapshot,
+           captured_at)
+         VALUES (?, ?, ?, ?, ?, ?, ?)`,
+        [actor.tenantId, input.assetId, input.onlineStatus, input.signalStrength ?? null,
+          input.firmwareVersion ?? '', JSON.stringify(input.snapshot), capturedAt]
+      );
+      return { assetId: input.assetId, capturedAt: capturedAt.toISOString() };
+    });
+  }
+
+  async addMaintenance(actor: ManagementActor, input: {
+    assetId: string; storeId: string; roomId?: string | null;
+    recordType: 'INSPECTION' | 'REPAIR' | 'REPLACEMENT';
+    status: 'OPEN' | 'RESOLVED'; description: string;
+  }) {
+    this.assertStoreScope(actor, input.storeId, true);
+    return this.transaction(async (connection) => {
+      await this.lockDevice(connection, actor, input.assetId, input.storeId);
+      const [result] = await connection.execute<ResultSetHeader>(
+        `INSERT INTO qipai_device_maintenance_records
+          (tenant_id, device_id, store_id, room_id, record_type, status, description,
+           created_by, resolved_at)
+         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+        [actor.tenantId, input.assetId, input.storeId, input.roomId ?? null,
+          input.recordType, input.status, input.description, actor.userId,
+          input.status === 'RESOLVED' ? new Date() : null]
+      );
+      await connection.execute(
+        `UPDATE qipai_devices SET maintenance_status = ?
+         WHERE tenant_id = ? AND id = ?`,
+        [input.status === 'OPEN' ? 'MAINTENANCE' : 'NORMAL', actor.tenantId, input.assetId]
+      );
+      await this.audit(connection, actor, 'DEVICE_MAINTENANCE_RECORDED', input.assetId, {
+        recordType: input.recordType, status: input.status
+      });
+      return { maintenanceId: String(result.insertId) };
+    });
+  }
+
+  private assertStoreScope(actor: ManagementActor, storeId: string, write: boolean) {
+    if (actor.access.capabilities.includes('tenant.manage')
+        || actor.access.roles.includes('PLATFORM_ADMIN')) return;
+    const capability = write ? 'device.write' : 'device.read';
+    if (!actor.access.capabilities.includes(capability)
+        || !actor.access.storeIds.includes(storeId)) {
+      throw new DeviceError('DEVICE_SCOPE_FORBIDDEN');
+    }
+  }
+
+  private scope(actor: ManagementActor, expression: string) {
+    if (actor.access.capabilities.includes('tenant.manage')
+        || actor.access.roles.includes('PLATFORM_ADMIN')) return { sql: '1 = 1', params: [] as string[] };
+    if (actor.access.storeIds.length === 0) return { sql: '1 = 0', params: [] as string[] };
+    return {
+      sql: `${expression} IN (${actor.access.storeIds.map(() => '?').join(',')})`,
+      params: actor.access.storeIds
+    };
+  }
+
+  private async assertStoreAndRoom(
+    connection: PoolConnection, actor: ManagementActor, storeId: string, roomId: string | null
+  ) {
+    const [stores] = await connection.execute<IdRow[]>(
+      'SELECT id FROM qipai_stores WHERE tenant_id = ? AND id = ? AND deleted_at IS NULL',
+      [actor.tenantId, storeId]
+    );
+    if (!stores[0]) throw new DeviceError('DEVICE_STORE_NOT_FOUND');
+    if (!roomId) return;
+    const [rooms] = await connection.execute<IdRow[]>(
+      `SELECT id FROM qipai_rooms
+       WHERE tenant_id = ? AND store_id = ? AND id = ? AND deleted_at IS NULL`,
+      [actor.tenantId, storeId, roomId]
+    );
+    if (!rooms[0]) throw new DeviceError('DEVICE_ROOM_NOT_FOUND');
+  }
+
+  private async lockDevice(
+    connection: PoolConnection, actor: ManagementActor, assetId: string, storeId: string
+  ) {
+    const [rows] = await connection.execute<Array<RowDataPacket & { id: string; deviceType: DeviceType }>>(
+      `SELECT id, device_type AS deviceType FROM qipai_devices
+       WHERE tenant_id = ? AND store_id = ? AND id = ? AND deleted_at IS NULL FOR UPDATE`,
+      [actor.tenantId, storeId, assetId]
+    );
+    if (!rows[0]) throw new DeviceError('DEVICE_NOT_FOUND');
+    return rows[0];
+  }
+
+  private async audit(
+    connection: PoolConnection, actor: ManagementActor,
+    action: string, resourceId: string, metadata: Record<string, unknown>
+  ) {
+    await connection.execute(
+      `INSERT INTO qipai_audit_logs
+        (tenant_id, actor_type, actor_id, action, resource_type, resource_id,
+         trace_id, ip, user_agent, metadata)
+       VALUES (?, 'USER', ?, ?, 'DEVICE', ?, ?, ?, ?, ?)`,
+      [actor.tenantId, actor.userId, action, resourceId, actor.traceId,
+        actor.ip, actor.userAgent.slice(0, 255), JSON.stringify(metadata)]
+    );
+  }
+
+  private async transaction<T>(work: (connection: PoolConnection) => Promise<T>) {
+    const connection = await this.pool.getConnection();
+    try {
+      await connection.beginTransaction();
+      const result = await work(connection);
+      await connection.commit();
+      return result;
+    } catch (error) {
+      await connection.rollback();
+      throw error;
+    } finally {
+      connection.release();
+    }
+  }
+}
+
+function isDuplicate(error: unknown): boolean {
+  return typeof error === 'object' && error !== null
+    && 'code' in error && error.code === 'ER_DUP_ENTRY';
+}
+
+function parseJson<T>(value: unknown, fallback: T): T {
+  if (value && typeof value === 'object') return value as T;
+  if (typeof value !== 'string' || value.length === 0) return fallback;
+  try { return JSON.parse(value) as T; } catch { return fallback; }
+}
+
+function maskIdentifier(value: string): string {
+  if (value.length <= 4) return '*'.repeat(value.length);
+  return `${value.slice(0, 2)}${'*'.repeat(Math.min(8, value.length - 4))}${value.slice(-2)}`;
+}
+
+function normalizeIds(rows: RowDataPacket[]) {
+  return rows.map((row) => Object.fromEntries(
+    Object.entries(row).map(([key, value]) => [
+      key,
+      (key === 'id' || key.endsWith('Id')) && value !== null ? String(value) : value
+    ])
+  ));
+}