ulweb/backend/src/device-data.service.ts

import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
import { connect as connectMqtt } from 'mqtt';
import { Client as PgClient } from 'pg';

export interface DeviceRecord {
  id: number;
  topic: string;
  payload: string;
  receivedAt: string;
}

@Injectable()
export class DeviceDataService implements OnModuleInit {
  private readonly logger = new Logger(DeviceDataService.name);
  private readonly listeners = new Set<(record: DeviceRecord) => void>();
  private pgClient: PgClient | null = null;
  private mqttClient: any = null;
  private initialized = false;

  async onModuleInit() {
    await this.initialize();
  }

  async initialize() {
    if (this.initialized) {
      return;
    }

    await this.connectDatabase();
    await this.connectMqtt();
    this.initialized = true;
  }

  async getRecords(limit = 20): Promise<DeviceRecord[]> {
    if (!this.pgClient) {
      await this.initialize();
    }

    const result = await this.pgClient!.query(
      `SELECT id, topic, payload, received_at as "receivedAt"
       FROM device_events
       ORDER BY received_at DESC, id DESC
       LIMIT $1`,
      [limit],
    );

    return result.rows.map((row) => ({
      id: row.id,
      topic: row.topic,
      payload: row.payload,
      receivedAt: row.receivedAt.toISOString(),
    }));
  }

  subscribe(listener: (record: DeviceRecord) => void): () => void {
    this.listeners.add(listener);
    return () => this.listeners.delete(listener);
  }

  private async connectDatabase() {
    try {
      const client = new PgClient({
        host: process.env.DATABASE_HOST || 'postgres',
        port: Number(process.env.DATABASE_PORT || 5432),
        user: process.env.DATABASE_USER || 'ulhub',
        password: process.env.DATABASE_PASSWORD || 'development',
        database: process.env.DATABASE_NAME || 'ulhub',
      });

      await client.connect();
      await client.query(`
        CREATE TABLE IF NOT EXISTS device_events (
          id SERIAL PRIMARY KEY,
          topic TEXT NOT NULL,
          payload TEXT NOT NULL,
          received_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
        )
      `);

      this.pgClient = client;
      this.logger.log('Connected to PostgreSQL and ensured device_events table exists');
    } catch (error) {
      this.logger.error('Failed to connect to PostgreSQL', error);
      throw error;
    }
  }

  private async connectMqtt() {
    try {
      const host = process.env.MQTT_HOST || 'mosquitto';
      const port = Number(process.env.MQTT_PORT || 1883);
      const username = process.env.MQTT_USERNAME || 'testuser';
      const password = process.env.MQTT_PASSWORD || 'testpass';
      this.mqttClient = connectMqtt(`mqtt://${host}:${port}`, {
        username,
        password,
        clientId: `ulhub-backend-${Math.random().toString(16).slice(2)}`,
      });

      this.mqttClient.on('connect', () => {
        this.logger.log(`Connected to MQTT broker at ${host}:${port}`);
        this.mqttClient!.subscribe('devices/#', (err) => {
          if (err) {
            this.logger.error('Failed to subscribe to devices/#', err);
          } else {
            this.logger.log('Subscribed to devices/#');
          }
        });
      });

      this.mqttClient.on('message', async (topic, payload) => {
        await this.persistRecord(topic, payload.toString());
      });

      this.mqttClient.on('error', (error) => {
        this.logger.error('MQTT client error', error);
      });
    } catch (error) {
      this.logger.error('Failed to connect to MQTT broker', error);
    }
  }

  private async persistRecord(topic: string, payload: string): Promise<DeviceRecord> {
    if (!this.pgClient) {
      await this.initialize();
    }

    const result = await this.pgClient!.query(
      `INSERT INTO device_events (topic, payload)
       VALUES ($1, $2)
       RETURNING id, topic, payload, received_at as "receivedAt"`,
      [topic, payload],
    );

    const record: DeviceRecord = {
      id: result.rows[0].id,
      topic: result.rows[0].topic,
      payload: result.rows[0].payload,
      receivedAt: result.rows[0].receivedAt.toISOString(),
    };

    this.notifyListeners(record);
    return record;
  }

  private notifyListeners(record: DeviceRecord) {
    for (const listener of this.listeners) {
      listener(record);
    }
  }
}