import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import { Kafka, Producer } from 'kafkajs';
import { ITelemetryProducerService } from '../../domain/ports/telemetry-producer.service.interface';

@Injectable()
export class TelemetryProducerService implements ITelemetryProducerService, OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(TelemetryProducerService.name);
  private readonly kafka: Kafka;
  private producer: Producer;

  constructor() {
    this.kafka = new Kafka({
      clientId: 'telemetry-service',
      brokers: (process.env.KAFKA_BROKERS || 'localhost:9092').split(','),
    });
    this.producer = this.kafka.producer();
  }

  async onModuleInit() {
    try {
      await this.producer.connect();
      this.logger.log('Kafka telemetry producer connected');
    } catch (err) {
      this.logger.warn(`Kafka producer connection failed (non-fatal): ${err.message}`);
    }
  }

  async onModuleDestroy() {
    try {
      await this.producer.disconnect();
    } catch (_) {}
  }

  async publishSessionStarted(data: {
    userId?: string;
    installId: string;
    timestamp: number;
    properties?: Record<string, any>;
  }) {
    try {
      await this.producer.send({
        topic: 'telemetry.session.started',
        messages: [{
          key: data.userId || data.installId,
          value: JSON.stringify({
            userId: data.userId,
            installId: data.installId,
            timestamp: data.timestamp,
            properties: data.properties || {},
            source: 'telemetry-service',
          }),
        }],
      });
    } catch (err) {
      this.logger.warn(`Failed to publish session event: ${err.message}`);
    }
  }

  async publishHeartbeat(userId: string, installId: string, appVersion: string) {
    try {
      await this.producer.send({
        topic: 'telemetry.heartbeat',
        messages: [{
          key: userId,
          value: JSON.stringify({ userId, installId, appVersion, ts: Date.now() }),
        }],
      });
    } catch (err) {
      // Heartbeat kafka publish is optional, just log
      this.logger.debug(`Heartbeat publish skipped: ${err.message}`);
    }
  }
}
