IoT/internal/service/bridge/bridge.go

// Package bridge — MQTT-подписка на телеметрию устройств → shared-SQS.
//
// Роль старого mqtt-bridge, но внутри монолита: подключается к EMQX как
// MQTT-клиент, подписывается на "+/telemetry/+" и шлёт envelope в SQS.
package bridge

import (
	"context"
	"log/slog"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/sqs"
	mqtt "github.com/eclipse/paho.mqtt.golang"

	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/config"
	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/sqsclient"
)

// telemetryTopicFilter — все namespace, все устройства.
const telemetryTopicFilter = "+/telemetry/+"

// subscribeRetryInterval — пауза между попытками подписки при сбое.
// Подписка обязательна (иначе bridge «глохнет»: соединение есть, а топик не слушается).
const subscribeRetryInterval = 2 * time.Second

// Run — бесконечный цикл bridge: подключение → подписка → ожидание сигнала завершения.
func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *slog.Logger) error {
	log.Info("bridge: starting", "broker", cfg.MQTTBrokerURL, "client_id", cfg.MQTTClientID)

	// URL очереди получаем один раз при старте (как старый consumer).
	queueURL, err := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSQueueName)
	if err != nil {
		return err
	}
	log.Info("bridge: SQS queue resolved", "queue", cfg.SQSQueueName)

	// Обработчик сообщений передаём в connectMQTT: подписка выполняется в
	// OnConnectHandler, т.е. повторяется при каждом (ре)подключении.
	// Иначе после потери сессии EMQX бридж оставался бы без подписки
	// (прецедент 2026-08-16: реконнект без resubscribe → телеметрия терялась).
	//
	// Отправка в SQS — через диспатчер (канал + worker-пул): колбэк MQTT
	// не блокируется сетью (фикс CRITICAL из ревью 2026-08-16).
	dispatcher := newSQSDispatcher(sqsClient, queueURL, log)
	dispatcher.start(ctx)

	handler := newMessageHandler(dispatcher, log)

	client, err := connectMQTT(ctx, cfg, log, handler)
	if err != nil {
		dispatcher.closeAndWait()
		return err
	}

	<-ctx.Done()
	log.Info("bridge: shutting down")
	// Порядок: сначала отключить MQTT (источник), затем слить остаток в SQS.
	client.Disconnect(2000)
	dispatcher.closeAndWait()
	return nil
}

// errSubscribeTimeout — не дождались подтверждения подписки.
var errSubscribeTimeout = errBridge("MQTT subscribe timeout")

// errBridge — ошибки bridge.
type errBridge string

func (e errBridge) Error() string { return string(e) }

// connectMQTT устанавливает подключение к EMQX с автореконнектом.
// handler передаётся в OnConnectHandler: подписка повторяется при каждом подключении.
func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger, handler mqtt.MessageHandler) (mqtt.Client, error) {
	opts := mqtt.NewClientOptions()
	opts.AddBroker(cfg.MQTTBrokerURL)
	opts.SetClientID(cfg.MQTTClientID)
	opts.SetUsername(cfg.MQTTUsername)
	opts.SetPassword(cfg.MQTTPassword)
	opts.SetAutoReconnect(true)
	opts.SetConnectRetry(true)
	opts.SetConnectRetryInterval(5 * time.Second)
	opts.SetKeepAlive(30 * time.Second)
	opts.SetCleanSession(false)

	opts.SetConnectionLostHandler(func(_ mqtt.Client, err error) {
		log.Warn("bridge: MQTT connection lost, reconnecting...", "err", err)
	})
	opts.SetReconnectingHandler(func(_ mqtt.Client, _ *mqtt.ClientOptions) {
		log.Info("bridge: MQTT reconnecting...")
	})
	opts.SetOnConnectHandler(func(c mqtt.Client) {
		log.Info("bridge: MQTT connected")
		// Подписка обязательна. Ретраим ограниченно. НЕ вызываем c.Disconnect():
		// в paho это ЯВНОЕ отключение, после которого автореконнект НЕ срабатывает
		// — bridge оставался бы отключённым навсегда (ревью 2026-10-10, 🔴6).
		// При сбое ждём следующего реконнекта (OnConnectHandler вызовется заново).
		for attempt := 1; attempt <= 3; attempt++ {
			token := c.Subscribe(telemetryTopicFilter, 1, handler)
			if token.WaitTimeout(10*time.Second) && token.Error() == nil {
				log.Info("bridge: subscribed", "filter", telemetryTopicFilter, "attempt", attempt)
				return
			}
			log.Error("bridge: subscribe failed", "filter", telemetryTopicFilter, "attempt", attempt, "err", token.Error())
			select {
			case <-ctx.Done():
				return
			case <-time.After(subscribeRetryInterval):
			}
		}
		log.Error("bridge: subscribe failed after retries, will retry on next reconnect", "filter", telemetryTopicFilter)
	})

	client := mqtt.NewClient(opts)
	token := client.Connect()
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case <-time.After(30 * time.Second):
		return nil, errBridge("bridge: MQTT connect timeout")
	case <-token.Done():
	}
	if token.Error() != nil {
		return nil, token.Error()
	}
	return client, nil
}