// 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

	// bridgeFlushTimeout — сколько ждать слива остатка буфера в SQS при shutdown.
	// Если SQS лежит и ретраи не завершаются, по этому таймауту ctx отменяется,
	// воркеры обрывают отправку (остаток логируется как дропнутый).
	bridgeFlushTimeout = 10 * 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

	// mu защищает закрытие канала: не даёт enqueue писать в уже закрытый ch
	// (паника «send on closed channel» от позднего колбэка paho — ревью 🔴3).
	mu     sync.Mutex
	closed bool

	// ctx для отправки НЕ связан с сигнальным ctx bridge: иначе при shutdown
	// воркеры сразу вернули бы «context canceled» и дропнули остаток буфера
	// (ревью 🔴3). Отменяется в closeAndWait по bridgeFlushTimeout, если воркеры
	// не успели слить остаток.
	ctx    context.Context
	cancel context.CancelFunc
}

// newSQSDispatcher создаёт диспатчер. start() запускает worker-пул.
func newSQSDispatcher(client *sqs.Client, queueURL string, log *slog.Logger) *sqsDispatcher {
	// ctx независимый от сигнала bridge: см. поле ctx. Отменяется только в
	// closeAndWait (по таймауту), чтобы при shutdown воркеры успели слить буфер.
	ctx, cancel := context.WithCancel(context.Background())
	return &sqsDispatcher{
		ch:       make(chan TelemetryEnvelope, dispatcherQueueSize),
		client:   client,
		queueURL: queueURL,
		log:      log,
		ctx:      ctx,
		cancel:   cancel,
	}
}

// start запускает worker-пул. Воркеры используют d.ctx (независимый от сигнала).
func (d *sqsDispatcher) start() {
	for i := 0; i < dispatcherWorkers; i++ {
		d.wg.Add(1)
		go func(wid int) {
			defer d.wg.Done()
			d.worker(wid)
		}(i)
	}
}

// enqueue кладёт envelope в канал НЕблокирующе. false = буфер полон (дроп) или
// канал уже закрыт (shutdown). Под mu, чтобы не писать в закрытый ch.
func (d *sqsDispatcher) enqueue(e TelemetryEnvelope) bool {
	d.mu.Lock()
	defer d.mu.Unlock()
	if d.closed {
		return false
	}
	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 закрывает канал и ждёт, пока воркеры сольют остаток буфера в SQS.
// Закрытие — под mu (см. поле mu): после него поздний колбэк paho не упадёт на
// «send on closed channel». Флаш ограничен bridgeFlushTimeout: если SQS лежит,
// отменяем d.ctx, чтобы sendWithRetry оборвался и wg.Wait() вернулся.
func (d *sqsDispatcher) closeAndWait() {
	d.mu.Lock()
	d.closed = true
	close(d.ch)
	d.mu.Unlock()

	t := time.AfterFunc(bridgeFlushTimeout, d.cancel)
	d.wg.Wait()
	t.Stop()
	d.log.Info("bridge: SQS dispatcher stopped", "dropped_total", d.dropped.Load())
}

// worker — цикл отправки из канала с ретраями. Использует d.ctx (не сигнальный),
// чтобы при shutdown остаток буфера успел сойти в SQS.
func (d *sqsDispatcher) worker(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(d.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
		}
	}
}