IoT/cmd/iot-service/main.go

// iot-service — монолит IoT на платформе Nubes (контейнер «Простой HTTP»).
//
// Три роли в одном процессе:
//   - REST API + MQTT auth/acl (:9090, устройства в PostgreSQL);
//   - bridge — MQTT-подписка EMQX → shared-SQS;
//   - consumer — shared-SQS → per-tenant PostgreSQL.
//
// Наследие старого IoT (k8s, оператор, CRD) НЕ используется.
package main

import (
	"context"
	"errors"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/api"
	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/api/handler"
	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/bridge"
	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/config"
	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/consumer"
	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/sqsclient"
	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/store"
	"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg"
)

// version — подставляется через ldflags при сборке.
var version = "dev"

// main — тонкая обёртка: os.Exit обходит defer-ы, поэтому вся логика в run(),
// которая возвращает код выхода и выполняет свои defer-ы (devices/iotStore
// Close, финальный flush) ДО выхода (ревью 🔴5: «os.Exit обходит defer-ы»).
func main() {
	os.Exit(run())
}

func run() int {
	cfg, err := config.Load()
	if err != nil {
		slog.Error("config", "err", err)
		return 1
	}

	log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: cfg.LogLevel}))
	slog.SetDefault(log)

	log.Info("iot-service starting", "version", version, "port", cfg.APIPort)

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

	// PostgreSQL — устройства и телеметрия.
	storeCtx, storeCancel := context.WithTimeout(ctx, 15*time.Second)
	devices, err := store.Open(storeCtx, cfg.IOTPGDSN)
	storeCancel()
	if err != nil {
		log.Error("devices store", "err", err)
		return 1
	}
	defer devices.Close()
	log.Info("devices store ready")

	// Служебная строка бриджа в iot_devices — для PG-auth EMQX (хвост 3).
	ensureCtx, ensureCancel := context.WithTimeout(ctx, 10*time.Second)
	err = devices.EnsureBridgeDevice(ensureCtx, cfg.MQTTUsername, cfg.MQTTPassword)
	ensureCancel()
	if err != nil {
		log.Error("ensure bridge device", "err", err)
		return 1
	}
	log.Info("bridge device ensured", "username", cfg.MQTTUsername)

	iotStore, err := iotpg.New(cfg.IOTPGDSN, log)
	if err != nil {
		log.Error("iotpg store", "err", err)
		return 1
	}
	defer iotStore.Close()
	log.Info("telemetry store ready")

	sqsClient := sqsclient.New(cfg)

	h := &handler.Handler{
		Devices:   devices,
		IoTPG:     iotStore,
		SQS:       sqsClient,
		Cfg:       cfg,
		Log:       log,
		Version:   version,
		StartedAt: time.Now(),
	}

	// HTTP API — в основном потоке.
	// Readiness: /ready отдаёт 503, пока bridge/consumer не подняты.
	ready := &api.Readiness{}
	router := api.NewRouter(h, log, cfg.AuthTestMode, cfg.JwtHMACSecret, version, ready)
	server := &http.Server{
		Addr:              ":" + cfg.APIPort,
		Handler:           router,
		ReadHeaderTimeout: 10 * time.Second,
		ReadTimeout:       15 * time.Second,
		// WriteTimeout с запасом: admin-stats/telemetry отвечают через SQS/PG
		// (внутренние таймауты до 15с). IdleTimeout — для keep-alive платформы.
		WriteTimeout: 60 * time.Second,
		IdleTimeout:  120 * time.Second,
	}

	// bridge + consumer — в фоне.
	var wg sync.WaitGroup
	wg.Add(2)

	// fatal — подсистема (bridge/consumer) завершилась с ПОСТОЯННОЙ ошибкой.
	// Тогда после штатной остановки выходим с кодом ≠ 0, чтобы платформа
	// перезапустила контейнер (ревью 🔴5: раньше процесс оставался жив, а
	// /health отдавал 200).
	var fatal atomic.Bool

	go func() {
		defer wg.Done()
		// ready.SetBridge управляется самим bridge: true только после успешной
		// подписки (находка #3). Иначе «соединён, но не подписан» давал /ready=200
		// и молча терял телеметрию.
		defer ready.SetBridge(false)
		if err := bridge.Run(ctx, cfg, sqsClient, log, ready.SetBridge); err != nil && !errors.Is(err, context.Canceled) {
			log.Error("bridge stopped", "err", err)
			fatal.Store(true)
			cancel() // штатный shutdown, затем выход ≠ 0
		}
	}()

	go func() {
		defer wg.Done()
		ready.SetConsumer(true)
		defer ready.SetConsumer(false)
		if err := consumer.Run(ctx, cfg, sqsClient, iotStore, log); err != nil && !errors.Is(err, context.Canceled) {
			log.Error("consumer stopped", "err", err)
			fatal.Store(true)
			cancel()
		}
	}()

	// Сервер гаснет ПОСЛЕДНИМ: сначала останавливаются bridge+consumer
	// (ctx → disconnect MQTT → flush SQS), потом HTTP. Иначе платформа
	// может убить контейнер по liveness, пока bridge/consumer ещё работают
	// (фикс MEDIUM из ревью 2026-08-16).
	go func() {
		<-ctx.Done()
		wg.Wait()
		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer shutdownCancel()
		_ = server.Shutdown(shutdownCtx)
	}()

	log.Info("iot-service listening", "addr", server.Addr)
	if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
		log.Error("http server", "err", err)
		cancel() // останавливаем bridge/consumer
		wg.Wait()
		return 1
	}

	wg.Wait()
	if fatal.Load() {
		// Не os.Exit: возвращаем код, чтобы отработали defer-ы (iotStore.Close →
		// финальный flush последнего батча). os.Exit их бы обошёл.
		log.Error("iot-service: подсистема завершилась с ошибкой — выход ≠ 0")
		return 1
	}
	log.Info("iot-service stopped")
	return 0
}