ulweb/backend/src/sim/sim-mqtts.service.ts

import { BadRequestException, Injectable, Logger, NotFoundException, OnModuleDestroy } from '@nestjs/common';
import { connect, MqttClient } from 'mqtt';
import { PkiService } from '../certificates/pki.service';
import { PrismaService } from '../prisma/prisma.service';

// Publishes as a real field device would: connects to the 8883 TLS listener
// and authenticates with the device's own issued client certificate (CN =
// serial number), rather than going through the backend's own privileged
// broker connection (see MqttClientService). This is what actually exercises
// the mTLS handshake + devices/%u/# ACL scoping end-to-end, so a simulated
// device is only ever as trusted as a real one.
@Injectable()
export class SimMqttsService implements OnModuleDestroy {
  private readonly logger = new Logger(SimMqttsService.name);
  private readonly clients = new Map<string, Promise<MqttClient>>();

  constructor(
    private readonly prisma: PrismaService,
    private readonly pki: PkiService,
  ) {}

  async publish(orgId: string, serial: string, topic: string, payload: object): Promise<void> {
    const client = await this.clientFor(orgId, serial);
    await new Promise<void>((resolve, reject) => {
      client.publish(topic, JSON.stringify(payload), { qos: 1 }, (err) => (err ? reject(err) : resolve()));
    });
  }

  private clientFor(orgId: string, serial: string): Promise<MqttClient> {
    const existing = this.clients.get(serial);
    if (existing) {
      return existing;
    }
    const created = this.connectAsDevice(orgId, serial).catch((err) => {
      this.clients.delete(serial);
      throw err;
    });
    this.clients.set(serial, created);
    return created;
  }

  private async connectAsDevice(orgId: string, serial: string): Promise<MqttClient> {
    const device = await this.prisma.device.findFirst({ where: { orgId, serialNumber: serial } });
    if (!device) {
      throw new NotFoundException(
        `No device with serial "${serial}" in this organization — create one on the Devices page first`,
      );
    }
    const cert = await this.prisma.deviceCertificate.findUnique({ where: { deviceId: device.id } });
    if (!cert) {
      throw new BadRequestException(
        `Device "${serial}" has no certificate issued — issue one from the Devices page, then retry`,
      );
    }
    const ca = await this.pki.caCertPem();

    const host = process.env.MQTT_HOST || 'mosquitto';
    const port = Number(process.env.MQTT_TLS_PORT || 8883);
    // The broker's server cert is issued for whatever hostname an admin chose
    // when provisioning it (PkiService.provisionServerCert), which may not
    // match this container's docker-network hostname. Overriding just the TLS
    // servername lets us verify the chain + name against what it was actually
    // issued for, rather than disabling verification.
    const servername = process.env.MQTT_TLS_SERVERNAME || 'localhost';

    return new Promise<MqttClient>((resolve, reject) => {
      const client = connect({
        host,
        port,
        protocol: 'mqtts',
        servername,
        ca,
        cert: cert.certificatePem,
        key: cert.privateKeyPem,
        rejectUnauthorized: true,
        connectTimeout: 8000,
        clientId: `ulhub-sim-${serial}-${Math.random().toString(16).slice(2)}`,
      });
      const onError = (err: Error) => {
        client.end(true);
        reject(new BadRequestException(`MQTTS connection failed for device "${serial}": ${err.message}`));
      };
      client.once('error', onError);
      client.once('connect', () => {
        client.removeListener('error', onError);
        client.on('error', (err) => this.logger.warn(`Simulated device ${serial} MQTTS error: ${err.message}`));
        resolve(client);
      });
    });
  }

  async onModuleDestroy(): Promise<void> {
    for (const pending of this.clients.values()) {
      await pending.then((client) => client.endAsync()).catch(() => undefined);
    }
  }
}