109 lines
3.6 KiB
Go
109 lines
3.6 KiB
Go
// Package consumer — чтение shared-SQS → per-tenant PostgreSQL.
|
||
//
|
||
// Роль старого sqs-consumer внутри монолита: long-poll цикл, обработка
|
||
// envelope телеметрии, DeleteMessage после успешной записи (at-least-once).
|
||
package consumer
|
||
|
||
import (
|
||
"context"
|
||
"log/slog"
|
||
"time"
|
||
|
||
"github.com/aws/aws-sdk-go-v2/aws"
|
||
"github.com/aws/aws-sdk-go-v2/service/sqs"
|
||
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/config"
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/sqsclient"
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg"
|
||
)
|
||
|
||
// errorBackoff — пауза между попытками при ошибках SQS.
|
||
const errorBackoff = 5 * time.Second
|
||
|
||
// maxProcessBackoff — потолок backoff после ошибок обработки (PG).
|
||
const maxProcessBackoff = 60 * time.Second
|
||
|
||
// Run — бесконечный long-poll цикл потребителя.
|
||
func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *iotpg.IoTPostgresStore, log *slog.Logger) error {
|
||
queueURL, err := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSQueueName)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
log.Info("consumer: SQS queue resolved", "queue", cfg.SQSQueueName)
|
||
|
||
for {
|
||
if ctx.Err() != nil {
|
||
break
|
||
}
|
||
|
||
resp, err := sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
|
||
QueueUrl: aws.String(queueURL),
|
||
MaxNumberOfMessages: 10,
|
||
WaitTimeSeconds: int32(cfg.SQSLongPollSeconds),
|
||
VisibilityTimeout: int32(cfg.SQSVisibilityTimeout),
|
||
})
|
||
if err != nil {
|
||
if ctx.Err() != nil {
|
||
break
|
||
}
|
||
log.Error("consumer: ReceiveMessage failed", "err", err)
|
||
if !sleepCtx(ctx, errorBackoff) {
|
||
break
|
||
}
|
||
continue
|
||
}
|
||
|
||
processBackoff := time.Second
|
||
for _, msg := range resp.Messages {
|
||
if msg.Body == nil || *msg.Body == "" {
|
||
log.Warn("consumer: empty message body, deleting",
|
||
"message_id", aws.ToString(msg.MessageId))
|
||
if err := deleteMessage(ctx, sqsClient, queueURL, msg.ReceiptHandle, log); err != nil {
|
||
log.Error("consumer: DeleteMessage failed", "err", err, "message_id", aws.ToString(msg.MessageId))
|
||
}
|
||
continue
|
||
}
|
||
if err := processTelemetry(ctx, *msg.Body, store, log); err != nil {
|
||
log.Error("consumer: process telemetry", "err", err, "message_id", aws.ToString(msg.MessageId))
|
||
// Сообщение вернётся в очередь после visibility timeout.
|
||
// Backoff с jitter — не долбить упавший PG в цикле
|
||
// (фикс HIGH из ревью 2026-08-16: лавина ретраев).
|
||
if !sleepCtx(ctx, processBackoff) {
|
||
break
|
||
}
|
||
processBackoff *= 2
|
||
if processBackoff > maxProcessBackoff {
|
||
processBackoff = maxProcessBackoff
|
||
}
|
||
continue
|
||
}
|
||
processBackoff = time.Second
|
||
if err := deleteMessage(ctx, sqsClient, queueURL, msg.ReceiptHandle, log); err != nil {
|
||
log.Error("consumer: DeleteMessage failed", "err", err, "message_id", aws.ToString(msg.MessageId))
|
||
}
|
||
}
|
||
}
|
||
|
||
log.Info("consumer: shutting down")
|
||
return nil
|
||
}
|
||
|
||
// deleteMessage удаляет обработанное сообщение из очереди.
|
||
func deleteMessage(ctx context.Context, client *sqs.Client, queueURL string, receiptHandle *string, log *slog.Logger) error {
|
||
_, err := client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
|
||
QueueUrl: aws.String(queueURL),
|
||
ReceiptHandle: receiptHandle,
|
||
})
|
||
return err
|
||
}
|
||
|
||
// sleepCtx — ожидание с прерыванием по ctx. false = ctx завершён.
|
||
func sleepCtx(ctx context.Context, d time.Duration) bool {
|
||
select {
|
||
case <-ctx.Done():
|
||
return false
|
||
case <-time.After(d):
|
||
return true
|
||
}
|
||
}
|