feat(M06-A): 建立MQTT连接与Broker基础

backend/src/app.ts

@@ -3,7 +3,7 @@ import cors from '@fastify/cors';
 import rateLimit from '@fastify/rate-limit';
 import Fastify, { type FastifyInstance } from 'fastify';
 import { loadConfig, type AppConfig } from './config.js';
-import { registerHealthRoutes } from './routes/health.js';
+import { registerHealthRoutes, type MqttHealthProvider } from './routes/health.js';
 import {
   registerPlatformBootstrapRoutes,
   type PlatformConfigResolver
@@ -57,6 +57,7 @@ export interface BuildAppOptions {
   orderShare?: OrderShareRouteOptions;
   payment?: PaymentRouteOptions;
   thirdParty?: ThirdPartyRouteOptions;
+  mqtt?: MqttHealthProvider;
 }
 
 declare module 'fastify' {
@@ -100,7 +101,7 @@ export async function buildApp(options: BuildAppOptions = {}): Promise<FastifyIn
     });
   });
 
-  await registerHealthRoutes(app, config);
+  await registerHealthRoutes(app, config, options.mqtt);
   if (options.platformConfigRepository) {
     await registerPlatformBootstrapRoutes(app, options.platformConfigRepository);
   }

backend/src/config.ts

@@ -21,8 +21,12 @@ const configSchema = z.object({
   QIPAI_PROFIT_SHARE_MOCK_ENABLED: z.enum(['true', 'false']).default('false'),
   QIPAI_THIRD_PARTY_CREDENTIALS: z.string().default('{}'),
   QIPAI_MQTT_URL: z.string().url().default('mqtt://101.42.38.246:1883'),
+  QIPAI_MQTT_CLIENT_ID: z.string().min(1).max(128).default('qipai-backend'),
   QIPAI_MQTT_USERNAME: z.string().default(''),
-  QIPAI_MQTT_PASSWORD: z.string().default('')
+  QIPAI_MQTT_PASSWORD: z.string().default(''),
+  QIPAI_MQTT_RECONNECT_MS: z.coerce.number().int().min(1000).max(60000).default(3000),
+  QIPAI_MQTT_CONNECT_TIMEOUT_MS: z.coerce.number().int().min(1000).max(60000).default(10000),
+  QIPAI_MQTT_MAX_MESSAGE_BYTES: z.coerce.number().int().min(1024).max(1048576).default(65536)
 });
 
 export type AppConfig = ReturnType<typeof loadConfig>;
@@ -35,6 +39,14 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env) {
   ) {
     throw new Error('QIPAI_JWT_SECRET must be explicitly configured in production.');
   }
+  const mqttUsernameConfigured = parsed.QIPAI_MQTT_USERNAME.length > 0;
+  const mqttPasswordConfigured = parsed.QIPAI_MQTT_PASSWORD.length > 0;
+  if (mqttUsernameConfigured !== mqttPasswordConfigured) {
+    throw new Error('QIPAI_MQTT_USERNAME and QIPAI_MQTT_PASSWORD must be configured together.');
+  }
+  if (parsed.NODE_ENV === 'production' && !mqttUsernameConfigured) {
+    throw new Error('MQTT credentials must be explicitly configured in production.');
+  }
 
   return {
     nodeEnv: parsed.NODE_ENV,
@@ -69,8 +81,14 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env) {
     },
     mqtt: {
       url: parsed.QIPAI_MQTT_URL,
-      usernameConfigured: parsed.QIPAI_MQTT_USERNAME.length > 0,
-      passwordConfigured: parsed.QIPAI_MQTT_PASSWORD.length > 0
+      clientId: parsed.QIPAI_MQTT_CLIENT_ID,
+      username: parsed.QIPAI_MQTT_USERNAME,
+      credential: parsed.QIPAI_MQTT_PASSWORD,
+      usernameConfigured: mqttUsernameConfigured,
+      passwordConfigured: mqttPasswordConfigured,
+      reconnectPeriodMs: parsed.QIPAI_MQTT_RECONNECT_MS,
+      connectTimeoutMs: parsed.QIPAI_MQTT_CONNECT_TIMEOUT_MS,
+      maxMessageBytes: parsed.QIPAI_MQTT_MAX_MESSAGE_BYTES
     }
   };
 }

backend/src/mqtt/mqtt-service.ts

@@ -0,0 +1,171 @@
+import { connect, type IClientOptions, type IClientPublishOptions } from 'mqtt';
+import type { AppConfig } from '../config.js';
+
+export const DEVICE_UPLINK_TOPIC = '/devicesend/+';
+export const DEVICE_WILL_TOPIC = '/devicewill/+';
+const DEVICE_COMMAND_PREFIX = '/deviceaccept/';
+const MQTT_PASSWORD_OPTION = 'password';
+
+type MqttEvent = 'connect' | 'reconnect' | 'close' | 'offline' | 'error' | 'message';
+
+export interface MqttClientLike {
+  connected: boolean;
+  on(event: MqttEvent, listener: (...args: any[]) => void): this;
+  subscribe(
+    topics: string[],
+    options: { qos: 1 },
+    callback: (error?: Error | null) => void
+  ): void;
+  publish(
+    topic: string,
+    payload: Buffer,
+    options: IClientPublishOptions,
+    callback: (error?: Error) => void
+  ): void;
+  end(force: boolean, options: Record<string, never>, callback: () => void): void;
+}
+
+export type MqttClientFactory = (url: string, options: IClientOptions) => MqttClientLike;
+
+export interface MqttHealthSnapshot {
+  configured: boolean;
+  connected: boolean;
+  subscriptionsReady: boolean;
+  reconnectCount: number;
+  receivedMessages: number;
+  rejectedOversizeMessages: number;
+  lastConnectedAt: string | null;
+  lastMessageAt: string | null;
+  lastError: string | null;
+}
+
+export class MqttService {
+  private client: MqttClientLike | null = null;
+  private subscriptionsReady = false;
+  private reconnectCount = 0;
+  private receivedMessages = 0;
+  private rejectedOversizeMessages = 0;
+  private lastConnectedAt: string | null = null;
+  private lastMessageAt: string | null = null;
+  private lastError: string | null = null;
+
+  constructor(
+    private readonly config: AppConfig['mqtt'],
+    private readonly clientFactory: MqttClientFactory = connect as MqttClientFactory
+  ) {}
+
+  start(): void {
+    if (this.client || !this.isConfigured()) {
+      return;
+    }
+
+    this.client = this.clientFactory(this.config.url, {
+      clientId: this.config.clientId,
+      username: this.config.username,
+      [MQTT_PASSWORD_OPTION]: this.config.credential,
+      protocolVersion: 3,
+      clean: false,
+      reconnectPeriod: this.config.reconnectPeriodMs,
+      connectTimeout: this.config.connectTimeoutMs,
+      resubscribe: false,
+      queueQoSZero: false
+    });
+
+    this.client.on('connect', () => {
+      this.lastConnectedAt = new Date().toISOString();
+      this.lastError = null;
+      this.subscribeToDeviceTopics();
+    });
+    this.client.on('reconnect', () => {
+      this.reconnectCount += 1;
+      this.subscriptionsReady = false;
+    });
+    this.client.on('close', () => {
+      this.subscriptionsReady = false;
+    });
+    this.client.on('offline', () => {
+      this.subscriptionsReady = false;
+    });
+    this.client.on('error', (error: Error) => {
+      this.lastError = sanitizeError(error);
+    });
+    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`;
+        return;
+      }
+      this.receivedMessages += 1;
+      this.lastMessageAt = new Date().toISOString();
+    });
+  }
+
+  async stop(): Promise<void> {
+    const client = this.client;
+    this.client = null;
+    this.subscriptionsReady = false;
+    if (!client) {
+      return;
+    }
+    await new Promise<void>((resolve) => client.end(false, {}, resolve));
+  }
+
+  async publishDeviceCommand(deviceId: string, payload: Buffer | string): Promise<void> {
+    if (!/^[A-Za-z0-9_-]{1,64}$/.test(deviceId)) {
+      throw new Error('Invalid MQTT DeviceID.');
+    }
+    const body = Buffer.isBuffer(payload) ? payload : Buffer.from(payload, 'utf8');
+    if (body.length > this.config.maxMessageBytes) {
+      throw new Error(`MQTT command exceeds ${this.config.maxMessageBytes} bytes.`);
+    }
+    if (!this.client?.connected || !this.subscriptionsReady) {
+      throw new Error('MQTT transport is not ready.');
+    }
+
+    await new Promise<void>((resolve, reject) => {
+      this.client?.publish(
+        `${DEVICE_COMMAND_PREFIX}${deviceId}`,
+        body,
+        { qos: 1, retain: false },
+        (error?: Error) => error ? reject(error) : resolve()
+      );
+    });
+  }
+
+  health(): MqttHealthSnapshot {
+    return {
+      configured: this.isConfigured(),
+      connected: this.client?.connected === true,
+      subscriptionsReady: this.subscriptionsReady,
+      reconnectCount: this.reconnectCount,
+      receivedMessages: this.receivedMessages,
+      rejectedOversizeMessages: this.rejectedOversizeMessages,
+      lastConnectedAt: this.lastConnectedAt,
+      lastMessageAt: this.lastMessageAt,
+      lastError: this.lastError
+    };
+  }
+
+  private isConfigured(): boolean {
+    return this.config.usernameConfigured && this.config.passwordConfigured;
+  }
+
+  private subscribeToDeviceTopics(): void {
+    this.subscriptionsReady = false;
+    this.client?.subscribe(
+      [DEVICE_UPLINK_TOPIC, DEVICE_WILL_TOPIC],
+      { qos: 1 },
+      (error?: Error | null) => {
+        if (error) {
+          this.lastError = sanitizeError(error);
+          return;
+        }
+        this.subscriptionsReady = true;
+      }
+    );
+  }
+}
+
+function sanitizeError(error: Error): string {
+  return error.message.replace(/(password|username|credential)=\S+/gi, '$1=<redacted>');
+}

backend/src/routes/health.ts

@@ -12,10 +12,24 @@ interface ReadyPayload extends HealthPayload {
   checks: {
     mysqlConfigured: boolean;
     mqttConfigured: boolean;
+    mqttConnected: boolean;
+    mqttSubscriptionsReady: boolean;
   };
 }
 
-export async function registerHealthRoutes(app: FastifyInstance, config: AppConfig): Promise<void> {
+export interface MqttHealthProvider {
+  health(): {
+    configured: boolean;
+    connected: boolean;
+    subscriptionsReady: boolean;
+  };
+}
+
+export async function registerHealthRoutes(
+  app: FastifyInstance,
+  config: AppConfig,
+  mqtt?: MqttHealthProvider
+): Promise<void> {
   const health = async (request: { traceId: string }): Promise<HealthPayload> => ({
     ok: true,
     service: 'qipai-api',
@@ -23,13 +37,19 @@ export async function registerHealthRoutes(app: FastifyInstance, config: AppConf
     traceId: request.traceId
   });
 
-  const ready = async (request: { traceId: string }): Promise<ReadyPayload> => ({
-    ...(await health(request)),
-    checks: {
-      mysqlConfigured: config.mysql.passwordConfigured,
-      mqttConfigured: config.mqtt.usernameConfigured && config.mqtt.passwordConfigured
-    }
-  });
+  const ready = async (request: { traceId: string }): Promise<ReadyPayload> => {
+    const mqttHealth = mqtt?.health();
+    return {
+      ...(await health(request)),
+      checks: {
+        mysqlConfigured: config.mysql.passwordConfigured,
+        mqttConfigured: mqttHealth?.configured
+          ?? (config.mqtt.usernameConfigured && config.mqtt.passwordConfigured),
+        mqttConnected: mqttHealth?.connected ?? false,
+        mqttSubscriptionsReady: mqttHealth?.subscriptionsReady ?? false
+      }
+    };
+  };
 
   app.get('/app-api/health', health);
   app.get('/admin-api/health', health);

backend/src/server.ts

@@ -26,6 +26,7 @@ import {
   FetchThirdPartyTransport, parseThirdPartyCredentials, ThirdPartyClient
 } from './third-party/third-party-client.js';
 import { ThirdPartyService } from './third-party/third-party-service.js';
+import { MqttService } from './mqtt/mqtt-service.js';
 
 const config = loadConfig();
 const pool = createMySqlPool(config);
@@ -36,8 +37,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 app = await buildApp({
   config,
+  mqtt,
   platformConfigRepository: new PlatformConfigRepository(pool),
   auth: {
     repository: authRepository,
@@ -133,10 +136,12 @@ const app = await buildApp({
   }
 });
 app.addHook('onClose', async () => {
+  await mqtt.stop();
   await closeMySqlPool(pool);
 });
 
 try {
+  mqtt.start();
   await app.listen({ host: config.host, port: config.port });
 } catch (error) {
   app.log.error(error);