Repinoid
c3106d6e20
bridge пишет, consumer парсит RFC3339Nano (субсекундная точность). Раньше RFC3339 (секунды) + non-FIFO очередь → показания в одну секунду могли встать не в том порядке. Parse RFC3339Nano совместим со старыми RFC3339-сообщениями (доля секунды опциональна).
69 lines
3.5 KiB
Go
69 lines
3.5 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
|
||
}
|
||
|
||
// Семантическая валидация envelope ДО записи в БД: пустые namespace/device_id
|
||
// или пустой payload — битое сообщение, ретраем не исправится. Пустой
|
||
// namespace → EnsureTenantDB создал бы БД «tenant_», пустой payload → NULL
|
||
// (23502) или битый JSONB. Такое шлём в DLQ (errInvalidEnvelope), а не
|
||
// бесконечно ретраим (ревью 2026-10-10, 🔴2).
|
||
if envelope.Namespace == "" || envelope.DeviceID == "" || len(envelope.Payload) == 0 {
|
||
log.Warn("consumer: invalid envelope (empty namespace/device_id/payload), routing to DLQ",
|
||
"namespace", envelope.Namespace, "device", envelope.DeviceID)
|
||
return errInvalidEnvelope
|
||
}
|
||
|
||
// EnsureTenantDB идемпотентен, кэшируется после первого вызова.
|
||
if err := store.EnsureTenantDB(ctx, envelope.Namespace); err != nil {
|
||
if errors.Is(err, iotpg.ErrInvalidNamespace) {
|
||
// Небезопасный namespace — постоянная ошибка → в DLQ (не ретраить).
|
||
log.Warn("consumer: invalid namespace, routing to DLQ", "namespace", envelope.Namespace)
|
||
return errInvalidEnvelope
|
||
}
|
||
return err
|
||
}
|
||
// Время приёма сохраняем в ts (порядок телеметрии). RFC3339Nano — субсекундная
|
||
// точность (очередь non-FIFO: несколько показаний в секунду не должны терять
|
||
// порядок). Parse RFC3339Nano принимает и старые RFC3339 (доля секунды опциональна).
|
||
// Битый/пустой received_at → zero, стор подставит now().
|
||
receivedAt, perr := time.Parse(time.RFC3339Nano, 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
|
||
}
|
||
|
||
// Debug: данные ещё только буферизованы (не записаны), Info на каждое
|
||
// сообщение засорял бы лог (ревью 2026-10-10).
|
||
log.Debug("consumer: telemetry buffered",
|
||
"namespace", envelope.Namespace,
|
||
"device", envelope.DeviceID,
|
||
)
|
||
return nil
|
||
}
|