Repinoid
e8873fa24d
Раньше ts проставлялся now() в БД при INSERT → при батчинге и параллельных записях порядок телеметрии (ORDER BY ts DESC) ломался. Теперь в ts идёт ReceivedAt из envelope (серверное время приёма, ставит bridge); битый/пустой received_at → zero → стор подставляет now(). Схема не меняется (используется существующая колонка ts). Обновлены callers: consumer/processor.go и legacy cmd/sqs-consumer.
49 lines
2.0 KiB
Go
49 lines
2.0 KiB
Go
// processor.go — разбор envelope телеметрии и запись в per-tenant PostgreSQL.
|
|
package consumer
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/bridge"
|
|
"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg"
|
|
)
|
|
|
|
// errInvalidEnvelope — битый/неразборчивый envelope: повторная обработка
|
|
// бессмысленна, сообщение следует отправить в DLQ, а не ретраить бесконечно.
|
|
var errInvalidEnvelope = errors.New("invalid telemetry envelope")
|
|
|
|
// processTelemetry десериализует envelope и пишет в tenant DB.
|
|
// Битый JSON — возвращает errInvalidEnvelope (consumer шлёт в DLQ),
|
|
// а НЕ молчаливый nil: иначе мусор подтверждается и теряется безвозвратно.
|
|
func processTelemetry(ctx context.Context, body string, store *iotpg.IoTPostgresStore, log *slog.Logger) error {
|
|
var envelope bridge.TelemetryEnvelope
|
|
if err := json.Unmarshal([]byte(body), &envelope); err != nil {
|
|
log.Warn("consumer: invalid envelope, routing to DLQ", "err", err)
|
|
return errInvalidEnvelope
|
|
}
|
|
|
|
// EnsureTenantDB идемпотентен, кэшируется после первого вызова.
|
|
if err := store.EnsureTenantDB(ctx, envelope.Namespace); err != nil {
|
|
return err
|
|
}
|
|
// Время приёма сохраняем в ts (порядок телеметрии). Битый/пустой received_at
|
|
// → zero, стор подставит now().
|
|
receivedAt, perr := time.Parse(time.RFC3339, envelope.ReceivedAt)
|
|
if perr != nil {
|
|
receivedAt = time.Time{}
|
|
}
|
|
if err := store.InsertTelemetry(ctx, envelope.Namespace, envelope.DeviceID, envelope.MessageID, receivedAt, envelope.Payload); err != nil {
|
|
return err
|
|
}
|
|
|
|
log.Info("consumer: telemetry saved",
|
|
"namespace", envelope.Namespace,
|
|
"device", envelope.DeviceID,
|
|
)
|
|
return nil
|
|
}
|