corrosion-admin-panel/backend-nest/src/services/nats.service.ts

import { Injectable, OnModuleInit, OnModuleDestroy, Logger } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { connect, NatsConnection, StringCodec, Subscription } from 'nats';

@Injectable()
export class NatsService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(NatsService.name);
  private nc: NatsConnection | null = null;
  private sc = StringCodec();

  constructor(private config: ConfigService) {}

  async onModuleInit() {
    try {
      const url = this.config.get<string>('nats.url') || 'nats://localhost:4222';
      this.nc = await connect({ servers: url });
      this.logger.log(`Connected to NATS at ${url}`);
    } catch (err) {
      this.logger.warn(`NATS connection failed — running in offline mode: ${(err as Error).message}`);
    }
  }

  async onModuleDestroy() {
    if (this.nc) {
      await this.nc.drain();
    }
  }

  async publish(subject: string, data: Record<string, unknown>): Promise<void> {
    if (!this.nc) {
      this.logger.debug(`[OFFLINE] Would publish to ${subject}: ${JSON.stringify(data)}`);
      return;
    }
    this.nc.publish(subject, this.sc.encode(JSON.stringify(data)));
  }

  async request(subject: string, data: Record<string, unknown>, timeout = 5000): Promise<unknown> {
    if (!this.nc) {
      this.logger.debug(`[OFFLINE] Would request ${subject}: ${JSON.stringify(data)}`);
      return null;
    }
    const msg = await this.nc.request(subject, this.sc.encode(JSON.stringify(data)), { timeout });
    return JSON.parse(this.sc.decode(msg.data));
  }

  subscribe(subject: string, callback: (data: unknown, subject: string) => void): Subscription | null {
    if (!this.nc) {
      this.logger.debug(`[OFFLINE] Would subscribe to ${subject}`);
      return null;
    }
    const sub = this.nc.subscribe(subject);
    (async () => {
      for await (const msg of sub) {
        try {
          const parsed = JSON.parse(this.sc.decode(msg.data));
          callback(parsed, msg.subject);
        } catch {
          callback(this.sc.decode(msg.data), msg.subject);
        }
      }
    })();
    return sub;
  }

  /** Publish a command to a specific license's server */
  async sendServerCommand(licenseId: string, action: string, payload: Record<string, unknown> = {}): Promise<void> {
    await this.publish(`corrosion.${licenseId}.cmd.server`, {
      action,
      ...payload,
      timestamp: new Date().toISOString(),
    });
  }

  /** Publish a deploy command to a specific license's companion agent */
  async sendDeployCommand(licenseId: string, config: Record<string, unknown>): Promise<void> {
    await this.publish(`corrosion.${licenseId}.cmd.deploy`, {
      action: 'deploy',
      config,
      timestamp: new Date().toISOString(),
    });
  }
}