// Создано: 2026-04-06
// kafka-consumer/main.go — iot-kafka-consumer: читает IoT телеметрию из Kafka → пишет в Postgres.
//
// Роль в архитектуре:
//   Kafka топик "iot.telemetry" → iot-kafka-consumer → IoT Postgres (per-tenant DB)
//
// Consumer group "iot-pg-consumer" — позволяет запускать несколько реплик без дублирования.
// При временной недоступности Postgres — Kafka хранит сообщения (retention 7 дней).
//
// Конфигурация через env vars:
//   KAFKA_BROKERS  — kafka.sless.svc.cluster.local:9092 (или managed Kafka в prod)
//   IOT_PG_DSN     — postgres://user:pass@host:5432/iotdb (master DSN для IoT Postgres)

package main

import (
	"context"
	"encoding/json"
	"log/slog"
	"os"
	"os/signal"
	"strings"
	"syscall"

	kafka "github.com/segmentio/kafka-go"

	"gitea-naeel.giteak8s.services.ngcloud.ru/naeel/sless/internal/storage/iotpg"
)

// kafkaConsumerConfig — конфигурация из env vars.
type kafkaConsumerConfig struct {
	KafkaBrokers string
}

// iotTelemetryMessage — envelope из Kafka (идентичен bridge).
type iotTelemetryMessage struct {
	Namespace  string          `json:"namespace"`
	DeviceID   string          `json:"device_id"`
	Topic      string          `json:"topic"`
	Payload    json.RawMessage `json:"payload"`
	ReceivedAt string          `json:"received_at"`
}

// iotTelemetryTopic — Kafka топик (должен совпадать с bridge).
const iotTelemetryTopic = "iot.telemetry"

// iotConsumerGroup — идентификатор consumer group.
// При нескольких репликах Kafka распределяет партиции между ними.
const iotConsumerGroup = "iot-pg-consumer"

func main() {
	log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))

	cfg := loadConsumerConfig()

	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
	defer cancel()

	log.Info("starting iot-kafka-consumer",
		"kafka_brokers", cfg.KafkaBrokers,
		"topic", iotTelemetryTopic,
		"group", iotConsumerGroup,
	)

	// IoT Postgres — обязательный компонент для этого сервиса
	iotStore, err := iotpg.NewFromEnv(log)
	if err != nil || iotStore == nil {
		log.Error("failed to connect to IoT Postgres — IOT_PG_DSN required", "err", err)
		os.Exit(1)
	}
	defer iotStore.Close()
	log.Info("connected to IoT Postgres")

	// Kafka reader с consumer group — автоматически коммитит offsets после обработки
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  strings.Split(cfg.KafkaBrokers, ","),
		Topic:    iotTelemetryTopic,
		GroupID:  iotConsumerGroup,
		MinBytes: 1,
		MaxBytes: 1 << 20, // 1MB
	})
	defer reader.Close()

	log.Info("kafka reader ready, waiting for messages...")

	for {
		// FetchMessage — блокирует до следующего сообщения
		kafkaMsg, err := reader.FetchMessage(ctx)
		if err != nil {
			if ctx.Err() != nil {
				break // штатное завершение
			}
			log.Error("fetch from Kafka", "err", err)
			continue
		}

		if err := processKafkaTelemetry(ctx, kafkaMsg, iotStore, log); err != nil {
			log.Error("process telemetry message", "err", err)
			// НЕ коммитим offset — сообщение будет перечитано при следующем старте
			continue
		}

		// Коммитим offset только после успешной записи в Postgres
		if err := reader.CommitMessages(ctx, kafkaMsg); err != nil {
			log.Error("commit Kafka offset", "err", err)
		}
	}

	log.Info("shutting down iot-kafka-consumer")
}

// processKafkaTelemetry десериализует сообщение из Kafka и записывает в Postgres.
func processKafkaTelemetry(ctx context.Context, msg kafka.Message, store *iotpg.IoTPostgresStore, log *slog.Logger) error {
	var envelope iotTelemetryMessage
	if err := json.Unmarshal(msg.Value, &envelope); err != nil {
		// Битое сообщение — логируем и пропускаем (не блокируем очередь)
		log.Warn("failed to unmarshal telemetry envelope, skipping", "err", err, "raw", string(msg.Value))
		return nil
	}

	// EnsureTenantDB идемпотентен — кэшируется после первого вызова
	if err := store.EnsureTenantDB(ctx, envelope.Namespace); err != nil {
		return err
	}

	if err := store.InsertTelemetry(ctx, envelope.Namespace, envelope.DeviceID, envelope.Payload); err != nil {
		return err
	}

	log.Info("telemetry saved to Postgres",
		"namespace", envelope.Namespace,
		"device", envelope.DeviceID,
		"kafka_offset", msg.Offset,
	)
	return nil
}

// loadConsumerConfig читает конфигурацию из env vars.
func loadConsumerConfig() kafkaConsumerConfig {
	return kafkaConsumerConfig{
		KafkaBrokers: getEnvOrDefault("KAFKA_BROKERS", "kafka.sless.svc.cluster.local:9092"),
	}
}

func getEnvOrDefault(key, defaultVal string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return defaultVal
}