// 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
	// sendMaxAttempts — попыток SendMessage на одно сообщение.
	sendMaxAttempts = 3

	// 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 с backoff 1с/2с/4с.
func (d *sqsDispatcher) sendWithRetry(ctx context.Context, body string) error {
	var lastErr error
	backoff := time.Second
	for attempt := 1; attempt <= sendMaxAttempts; attempt++ {
		_, err := d.client.SendMessage(ctx, &sqs.SendMessageInput{
			QueueUrl:    aws.String(d.queueURL),
			MessageBody: aws.String(body),
		})
		if err == nil {
			return nil
		}
		lastErr = err
		if ctx.Err() != nil {
			return lastErr
		}
		if attempt < sendMaxAttempts {
			select {
			case <-ctx.Done():
				return lastErr
			case <-time.After(backoff):
			}
			backoff *= 2
		}
	}
	return lastErr
}