IoT/internal/service/bridge/sender.go

// sender.go — асинхронная отправка envelope в SQS из бриджа.
//
// Мотивация (Sonnet-ревью 2026-08-16, находка CRITICAL): SendMessage был
// синхронным внутри MQTT-колбэка paho. При недоступности SQS (таймаут ~30с)
// блокировался приём ВСЕХ MQTT-сообщений → потери телеметрии.
//
// Теперь: колбэк кладёт envelope в буферизованный канал НЕблокирующе
// (переполнение — дроп со счётчиком), worker-пул отправляет в SQS с
// ограниченными ретраями и backoff.
package bridge

import (
	"context"
	"encoding/json"
	"log/slog"
	"sync"
	"sync/atomic"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
)

// Константы диспатчера.
const (
	// dispatcherQueueSize — буфер envelope перед SQS (~100 msg/s * 100с буфера).
	dispatcherQueueSize = 10000
	// dispatcherWorkers — горутин, отправляющих в SQS.
	dispatcherWorkers = 4

	// sendAttemptTimeout — таймаут ОДНОЙ попытки SendMessage. Без него зависшее
	// соединение (SQS висит, но не отвечает) блокировало бы воркера навсегда и
	// забило бы весь пул (ревью 🔴4).
	sendAttemptTimeout = 10 * time.Second
	// sendMaxBackoff — потолок паузы между попытками (экспоненциальный backoff).
	sendMaxBackoff = 8 * time.Second

	// maxSQSBodyBytes — потолок размера envelope для SQS.
	// SQS (и shared-sqs) режут сообщения >256KB — запас на оверхед.
	maxSQSBodyBytes = 250 * 1024
)

// sqsDispatcher — канал + worker-пул для отправки в SQS.
type sqsDispatcher struct {
	ch       chan TelemetryEnvelope
	client   *sqs.Client
	queueURL string
	log      *slog.Logger
	wg       sync.WaitGroup
	dropped  atomic.Int64
}

// newSQSDispatcher создаёт диспатчер. start() запускает worker-пул.
func newSQSDispatcher(client *sqs.Client, queueURL string, log *slog.Logger) *sqsDispatcher {
	return &sqsDispatcher{
		ch:       make(chan TelemetryEnvelope, dispatcherQueueSize),
		client:   client,
		queueURL: queueURL,
		log:      log,
	}
}

// start запускает worker-пул.
func (d *sqsDispatcher) start(ctx context.Context) {
	for i := 0; i < dispatcherWorkers; i++ {
		d.wg.Add(1)
		go func(wid int) {
			defer d.wg.Done()
			d.worker(ctx, wid)
		}(i)
	}
}

// enqueue кладёт envelope в канал НЕблокирующе. false = буфер полон (дроп).
func (d *sqsDispatcher) enqueue(e TelemetryEnvelope) bool {
	select {
	case d.ch <- e:
		return true
	default:
		n := d.dropped.Add(1)
		if n%100 == 1 {
			d.log.Error("bridge: SQS dispatch queue full, dropping",
				"namespace", e.Namespace, "device", e.DeviceID,
				"dropped_total", n)
		}
		return false
	}
}

// closeAndWait закрывает канал и ждёт завершения worker'ов (flush).
func (d *sqsDispatcher) closeAndWait() {
	close(d.ch)
	d.wg.Wait()
	d.log.Info("bridge: SQS dispatcher stopped", "dropped_total", d.dropped.Load())
}

// worker — цикл отправки из канала с ограниченными ретраями.
func (d *sqsDispatcher) worker(ctx context.Context, wid int) {
	for e := range d.ch {
		body, err := json.Marshal(e)
		if err != nil {
			d.log.Error("bridge: marshal envelope", "worker", wid,
				"namespace", e.Namespace, "err", err)
			continue
		}
		if len(body) > maxSQSBodyBytes {
			// Лимит SQS 256KB — такие сообщения не пройдут в принципе
			// (проверено 2026-08-16: InvalidParameterValue message size
			// exceeds the limit). Дроп с явным логом.
			d.log.Error("bridge: payload too large for SQS, dropping",
				"worker", wid, "namespace", e.Namespace,
				"device", e.DeviceID, "bytes", len(body),
				"limit_bytes", maxSQSBodyBytes)
			continue
		}
		if err := d.sendWithRetry(ctx, string(body)); err != nil {
			d.log.Error("bridge: SQS send failed after retries, dropping",
				"worker", wid, "namespace", e.Namespace,
				"device", e.DeviceID, "err", err)
		}
	}
}

// sendWithRetry — SendMessage с ретраями до остановки (ctx) и таймаутом на
// попытку. НЕ дропаем после фиксированного числа попыток (было 3 ≈ 7с): буфер
// на dispatcherQueueSize рассчитан на поглощение временной недоступности SQS,
// поэтому ретраим до ctx.Done() (shutdown) или успеха — иначе потеря была бы
// безвозвратной (ревью 🔴4).
func (d *sqsDispatcher) sendWithRetry(ctx context.Context, body string) error {
	backoff := time.Second
	for {
		attemptCtx, cancel := context.WithTimeout(ctx, sendAttemptTimeout)
		_, err := d.client.SendMessage(attemptCtx, &sqs.SendMessageInput{
			QueueUrl:    aws.String(d.queueURL),
			MessageBody: aws.String(body),
		})
		cancel()
		if err == nil {
			return nil
		}
		if ctx.Err() != nil {
			// shutdown: ctx отменён, ретраить бессмысленно.
			return err
		}
		select {
		case <-ctx.Done():
			return err
		case <-time.After(backoff):
		}
		if backoff < sendMaxBackoff {
			backoff *= 2
		}
	}
}