chore: update ignore rules

Не собирался Dockerfile/Makefile (только ./cmd/iot-service/) → в образ/деплой
не попадал; логика поглощена internal/service/consumer/. Был источником пустого
message_id (обход дедупа). message_id оставлен nullable — единственный пишущий
путь (bridge) всегда ставит UUID. LEGACY.md обновлён. Плюс исправлена неточность
в HISTORY про поведение без DLQ.

.gitignore

@@ -54,3 +54,11 @@ Thumbs.db
 *.orig
 *.swp
 .DS_Store
+/HISTORY/2026-08-16-session-log.md
+/deploy/terraform/terraform.tfvars.example
+/deployments/k8s/iot-postgres.yaml
+/doc/api/endpoints-v0.2.3.md
+/doc/deployment-v0.2.3.md
+/legacy/deployments/k8s/iot-postgres.yaml
+/legacy/doc/api/endpoints-v0.2.3.md
+/legacy/doc/deployment-v0.2.3.md

HISTORY/2026-10-07-session-log.md

@@ -0,0 +1,168 @@
+# Журнал сессии — 2026-10-07 (IoT: P0-доработки до production-ready)
+
+**Рабочая папка:** /home/naeel/nubes/SQS_und_IoT/IoT
+**Ветка:** feature/replace-kafka-with-sqs
+
+> Основание: `doc/thinking/remediation-plan-2026-10-05.md` (на ветке master) —
+> план по двум анализам готовности. Работа только по явной команде; после каждой
+> правки — коммит.
+
+---
+
+## Сделано
+
+### P0-1/P0-2 — tenant-граница + обязательная подпись JWT (`5e1cd28`)
+- `internal/service/api/middleware/auth.go`:
+  - `validateJWT` теперь возвращает `sub`;
+  - middleware сверяет `sub` с `{namespace}` из URL (`mux.Vars`) → 403 при
+    несовпадении (закрыт горизонтальный доступ к чужим namespace);
+  - вне `AUTH_TEST_MODE` подпись HS256 ОБЯЗАТЕЛЬНА: без `JWT_HMAC_SECRET`
+    middleware отказывает (fail-closed, 500), а не проверяет «структурно».
+- `internal/service/config/config.go`: `JWT_HMAC_SECRET` добавлен в список
+  обязательных (когда `AUTH_TEST_MODE=false`) — отказ на старте.
+
+### P1-4 — убраны неаутентифицированные `/internal/mqtt/*` (`cd47258`)
+- `internal/service/api/router.go`: удалены маршруты `/internal/mqtt/auth|acl`.
+  EMQX ходит в PostgreSQL напрямую (`backend=postgresql`), HTTP-fallback был
+  не нужен и висел на публичном порту без auth (перебор паролей/ACL).
+- Примечание: те же эндпоинты остаются в ЛЕГАСИ `internal/api/` (не исполняется).
+
+### P0-3 (часть 2) — пароль устройства не отдаётся повторно (`f480be3`)
+- `internal/service/api/handler/devices.go`: `GetIoTDevice` больше не возвращает
+  `mqtt_password` (передаём `""` в `deviceToResponse`). Пароль — только при создании.
+- Часть 1 (хэш паролей в БД + перевод EMQX на `password_hash_algorithm`) НЕ
+  сделана: требует правки `emqx/emqx.conf` и миграции существующих строк.
+
+### P0-7 — подтверждение из SQS только после записи в БД (`c84cbae`)
+- Проблема: `InsertTelemetry` лишь буферизует и возвращает nil до фактической
+  записи; consumer удалял сообщение сразу после `processTelemetry == nil` →
+  краш/flush-ошибка в окне = безвозвратная потеря телеметрии.
+- `internal/storage/iotpg/iot_telemetry_store.go`: добавлен `Flush(ctx)`.
+- `internal/service/consumer/consumer.go`: копим receipt handles, вызываем
+  `store.Flush(ctx)`, и только при успехе — `DeleteMessage`. При ошибке flush
+  сообщения НЕ удаляются (вернутся после visibility timeout).
+
+### P1-9 — `/ready` отражает состояние подсистем
+- `internal/service/api/readiness.go` — тип `Readiness` (atomic-флаги bridge/consumer).
+- `internal/service/api/router.go` — `NewRouter` принимает `*Readiness`; добавлен
+  `/ready` (200 если обе подсистемы в строю, иначе 503). `/health` остался
+  liveness-пробой.
+- `cmd/iot-service/main.go` — bridge/consumer поднимают/опускают свои флаги;
+  если горутина завершилась (ошибка/падение), `/ready` вернёт 503.
+
+### P0-8 — идемпотентность телеметрии (dedup по message_id)
+- `internal/service/bridge/handler.go`: в `TelemetryEnvelope` добавлен `MessageID`
+  (UUID на каждое MQTT-сообщение).
+- `internal/service/consumer/processor.go`: `MessageID` передаётся в `InsertTelemetry`.
+- `internal/storage/iotpg/iot_telemetry_store.go`: в таблицу добавлен столбец
+  `message_id` + UNIQUE INDEX `(device_id, message_id)`; батч-INSERT теперь
+  `ON CONFLICT DO NOTHING`. Пустой message_id → NULL (NULL в unique-индексе не
+  конфликтует → без ложного дедупа). При at-least-once повторе из SQS дубликат
+  не вставляется.
+- Легаси `cmd/sqs-consumer`: вызов обновлён (`""`), файл помечен к удалению (P2-14).
+
+### Ревью-фикс P0-7/P0-8 (найдено при перепроверке кода)
+- ❗ Была реальная потеря данных: `flushAll` и `InsertTelemetry` (при заполнении
+  батча) вынимали строки из буфера и **подменяли буфер ДО записи**; при ошибке
+  `flushTenant` строки исчезали. Хуже — фоновый flusher мог «съесть» строки,
+  которые consumer ждал в `Flush`, тогда `Flush` вернул бы nil (буфер пуст) и
+  consumer **подтвердил бы** SQS-сообщение с незаписанными данными.
+- Фикс: при ошибке `flushTenant` строки ВОЗВРАЩАЮТСЯ в буфер (в обоих путях) —
+  данные не теряются, повтор выполняет flusher/Flush. `ctx` проброшен в
+  `flushAll`/`flushTenant` (раньше `Flush(ctx)` игнорировал ctx).
+- Consumer: receipt handle хранится значением (`string`), а не указателем SDK.
+- Компромисс: при долгом сбое PG буфер растёт (backpressure) — осознанно, вместо
+  тихой потери телеметрии.
+
+## Осталось (не сделано)
+- P0-3 часть 1 (хэш паролей + EMQX), P1-11/P1-12 (тесты/CI), P2-14
+  (вычистить легаси `internal/api/`, `cmd/sqs-consumer`).
+- Платформенные проблемы (wss ~150с, MSS/MTU) — вне кода.
+
+---
+
+## Архитекторский разбор (Дипсик Про, по итогам перепроверки)
+
+Перечитана цепочка `bridge → SQS → consumer → iotpg` целиком. Найдены 5 проблем
+(часть — мой же недочёт при кодинге P0-7/P0-8):
+
+1. 🔴 **Залипание на «ядовитом» батче.** После моего фикса «возвращаем строки в
+   буфер при ошибке» один payload, падающий не по `ON CONFLICT`, зацикливает
+   flush этого namespace навсегда (flusher каждые 200мс). Так «тихая потеря»
+   превратилась в «вечное залипание». ← моя ошибка: фикс без различения типов ошибок.
+2. 🔴 **Битый envelope теряется безвозвратно.** `processTelemetry` на битый JSON
+   возвращает `nil` → consumer подтверждает (DeleteMessage) → мусор исчезает без
+   следа. Если мусор шлёт bridge по багу — данные теряются навсегда, без DLQ.
+3. 🟠 **Нарушение порядка телеметрии.** `ts` ставится `now()` в БД при INSERT, а
+   не в момент приёма (`ReceivedAt` из envelope не идёт в батч). При батчинге и
+   нескольких потоках порядок для device_id не гарантирован.
+4. 🟠 **Дублирование / нет дедупа для `message_id=""`.** `ON CONFLICT` по
+   `(device_id, message_id)` не ловит NULL: legacy `cmd/sqs-consumer` передаёт
+   `""` → дубли вставляются. Плюс гонка InsertTelemetry(full) ↔ flusher на одном
+   namespace = два параллельных INSERT.
+5. 🟡 **Close() может потерять последний батч** при ошибке shutdown-flush
+   (`_ = flushAll(...)` — ошибка игнорируется).
+
+### Решение (как принято)
+- п.1 + п.3 (битые/ядовитые → DLQ) — делаю я (Дипсик Про), это дизайн-развилки.
+- п.2 + п.4 (ReceivedAt; message_id NOT NULL) — передаются Флашу, с критериями.
+- Битое сообщение — **в DLQ** (стандартный паттерн: valid→process, invalid→DLQ).
+- Ядовитый SQL-батч — различать retryable (сеть/PG лежит → ретрай в буфер) от
+  permanent (нарушение constraint → не вешать namespace, выброс в лог/DLQ).
+
+### Статус
+Реализация п.1 (разделение ошибок flush) и п.3 (битые → DLQ) — см. ниже.
+
+### Реализовано (п.3) — битые envelope → DLQ (`42f0045`)
+- `processor.go`: `processTelemetry` возвращает sentinel `errInvalidEnvelope`
+  при битом JSON (вместо молчаливого nil, из-за которого мусор подтверждался и
+  исчезал навсегда).
+- `consumer.go`: при `errInvalidEnvelope` тело кладётся в DLQ
+  (`SQS_DLQ_NAME`, дефолт `iot-telemetry-dlq`) и основное сообщение подтверждается;
+  без DLQ — только лог (НЕ подтверждаем). Реальные ошибки PG — по-прежнему ретрай.
+- `config.go`/`defaults.go`: добавлен `SQSDLQName` + `DefaultSQSDLQName`.
+
+### Реализовано (п.1) — разделение retryable/permanent ошибок flush (`f1a7266`)
+- `iot_telemetry_store.go`: `isRetryableFlushErr` — `'08'` (connection) и
+  57P01/02/03 (shutdown/crash) retryable (возврат в буфер); `'22'` (data exception)
+  и `'23'` (integrity violation) permanent (батч выбрасывается с логом, НЕ в буфер).
+- Закрыто замеченное залипание: один «ядовитый» payload больше не зацикливает
+  flush namespace навсегда.
+
+### Реализовано (п.2) — `ts` = время приёма (`a386d22`)
+- `iot_telemetry_store.go`: `telemetryInsert` получил `ts`; `flushTenant` пишет
+  `ts` явно из `ReceivedAt` (серверное время приёма, ставит bridge), а не `now()`
+  БД. Zero → `now()`. Схема не менялась (существующая колонка `ts`).
+- Callers обновлены: `consumer/processor.go` (парсит `ReceivedAt`), legacy
+  `cmd/sqs-consumer`.
+
+### ОТКРЫТО (п.4) — `message_id NOT NULL` / дедуп legacy
+- Проблема: `UNIQUE(device_id, message_id)` не ловит NULL → legacy
+  `cmd/sqs-consumer` (передаёт `""`) не дедуплицируется; возможны дубли.
+- Развилка, где я не уверен (ОСТАНОВКА, без догадок):
+  1. Сделать `message_id NOT NULL`: нужна миграция (backfill `legacy-'||id` +
+     `SET NOT NULL`). Безопасна, но legacy-консьюмер тогда должен передавать
+     SQS `MessageId` (иначе его строки не пройдут NOT NULL и будут «дропнуты»
+     как permanent). Legacy и так идёт под удаление (P2-14).
+  2. Оставить nullable, задокументировать отсутствие дедупа для legacy и
+     удалить legacy (P2-14). Минимальный риск, но формально «не NOT NULL».
+- Нужно решение владельца.
+
+### Решено (п.4) — удалён legacy `cmd/sqs-consumer`
+- Проверено: `Dockerfile`/`Makefile` собирают ТОЛЬКО `./cmd/iot-service/` →
+  `cmd/sqs-consumer` не попадает в образ/деплой (пустой `message_id` до прода
+  не доходил). Логика поглощена `internal/service/consumer/`.
+- Удалён (`git rm -r cmd/sqs-consumer`); сборка/vet зелёные. `message_id`
+  оставлен nullable — единственный пишущий путь (bridge) всегда ставит UUID.
+- LEGACY.md обновлён: `cmd/sqs-consumer` перенесён в «удалено».
+- Примечание-противоречие: LEGACY.md до этого называл `cmd/sqs-consumer` «НЕ
+  легаси, переиспользуется»; при сверке с Dockerfile/Makefile подтверждено, что
+  standalone-бинарник не используется → удаление корректно.
+
+### Исправление неточности (моей)
+- В коммите п.3 и в этой HISTORY я написал «без DLQ — НЕ подтверждаем». По коду
+  НЕ так: даже без DLQ битый envelope ПОДТВЕРЖДАЕТСЯ (удаляется с логом) — и это
+  правильно (иначе он зациклился бы). Формулировка была неверной, исправлено.
+
+## Проверки
+- `go build ./...` и `go vet ./...` — зелёные на каждом шаге.

LEGACY.md

@@ -43,5 +43,9 @@
 - `HISTORY/2026-08-16-session-log.md`
 - `doc/sqs-integration.md`
 - `ws-probe/`
-- `cmd/mqtt-bridge/`, `cmd/sqs-consumer/`, `internal/` — Go-код, переиспользуется
-  в новом монолите (после адаптации под платформу)
+- `cmd/mqtt-bridge/`, `internal/` — Go-код, переиспользуется в новом монолите
+  (после адаптации под платформу)
+- ⛔ `cmd/sqs-consumer/` — **УДАЛЁН 2026-10-07**: standalone-консьюмер, который
+  НЕ собирался Dockerfile/Makefile (те билдят только `./cmd/iot-service/`); его
+  логика поглощена `internal/service/consumer/`. Был источником пустого
+  `message_id` (обход дедупа).

cmd/iot-service/main.go

@@ -89,7 +89,9 @@ func main() {
 	}
 
 	// HTTP API — в основном потоке.
-	router := api.NewRouter(h, log, cfg.AuthTestMode, cfg.JwtHMACSecret, version)
+	// Readiness: /ready отдаёт 503, пока bridge/consumer не подняты.
+	ready := &api.Readiness{}
+	router := api.NewRouter(h, log, cfg.AuthTestMode, cfg.JwtHMACSecret, version, ready)
 	server := &http.Server{
 		Addr:              ":" + cfg.APIPort,
 		Handler:           router,
@@ -102,6 +104,8 @@ func main() {
 
 	go func() {
 		defer wg.Done()
+		ready.SetBridge(true)
+		defer ready.SetBridge(false)
 		if err := bridge.Run(ctx, cfg, sqsClient, log); err != nil && !errors.Is(err, context.Canceled) {
 			log.Error("bridge stopped", "err", err)
 		}
@@ -109,6 +113,8 @@ func main() {
 
 	go func() {
 		defer wg.Done()
+		ready.SetConsumer(true)
+		defer ready.SetConsumer(false)
 		if err := consumer.Run(ctx, cfg, sqsClient, iotStore, log); err != nil && !errors.Is(err, context.Canceled) {
 			log.Error("consumer stopped", "err", err)
 		}

cmd/sqs-consumer/main.go

@@ -1,207 +0,0 @@
-// Создано: 2026-04-12 (замена kafka-consumer → sqs-consumer)
-// sqs-consumer/main.go — iot-sqs-consumer: читает IoT телеметрию из shared-SQS → пишет в Postgres.
-//
-// Роль в архитектуре:
-//   SQS очередь "iot-telemetry" → iot-sqs-consumer → IoT Postgres (per-tenant DB)
-//
-// Long polling (WaitTimeSeconds=20) — минимизирует пустые запросы к SQS.
-// При временной недоступности Postgres — сообщения остаются в очереди (visibility timeout).
-// DeleteMessage вызывается только после успешной записи в Postgres.
-//
-// Конфигурация через env vars:
-//   SQS_ENDPOINT   — https://qu.kube5s.ru (или внутрикластерный endpoint)
-//   SQS_ACCESS_KEY — Access Key для shared-SQS tenant
-//   SQS_SECRET_KEY — Secret Key для shared-SQS tenant
-//   SQS_QUEUE_NAME — имя очереди (default: iot-telemetry)
-//   SQS_REGION     — регион (default: us-east-1)
-//   IOT_PG_DSN     — postgres://user:pass@host:5432/iotdb (master DSN для IoT Postgres)
-
-package main
-
-import (
-	"context"
-	"encoding/json"
-	"log/slog"
-	"os"
-	"os/signal"
-	"syscall"
-	"time"
-
-	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/credentials"
-	"github.com/aws/aws-sdk-go-v2/service/sqs"
-
-	"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg"
-)
-
-// sqsConsumerConfig — конфигурация из env vars.
-type sqsConsumerConfig struct {
-	SQSEndpoint  string
-	SQSAccessKey string
-	SQSSecretKey string
-	SQSQueueName string
-	SQSRegion    string
-}
-
-// iotTelemetryMessage — envelope из SQS (идентичен bridge).
-type iotTelemetryMessage struct {
-	Namespace  string          `json:"namespace"`
-	DeviceID   string          `json:"device_id"`
-	Topic      string          `json:"topic"`
-	Payload    json.RawMessage `json:"payload"`
-	ReceivedAt string          `json:"received_at"`
-}
-
-func main() {
-	log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
-
-	cfg := loadConsumerConfig()
-
-	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
-	defer cancel()
-
-	log.Info("starting iot-sqs-consumer",
-		"sqs_endpoint", cfg.SQSEndpoint,
-		"sqs_queue", cfg.SQSQueueName,
-	)
-
-	// IoT Postgres — обязательный компонент для этого сервиса
-	iotStore, err := iotpg.NewFromEnv(log)
-	if err != nil || iotStore == nil {
-		log.Error("failed to connect to IoT Postgres — IOT_PG_DSN required", "err", err)
-		os.Exit(1)
-	}
-	defer iotStore.Close()
-	log.Info("connected to IoT Postgres")
-
-	// SQS клиент
-	sqsClient := sqs.New(sqs.Options{
-		Region: cfg.SQSRegion,
-		Credentials: credentials.NewStaticCredentialsProvider(
-			cfg.SQSAccessKey, cfg.SQSSecretKey, "",
-		),
-		BaseEndpoint: aws.String(cfg.SQSEndpoint),
-	})
-
-	// Получаем QueueUrl по имени один раз при старте
-	queueUrlOut, err := sqsClient.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{
-		QueueName: aws.String(cfg.SQSQueueName),
-	})
-	if err != nil {
-		log.Error("failed to get SQS queue URL — очередь должна существовать", "queue", cfg.SQSQueueName, "err", err)
-		os.Exit(1)
-	}
-	queueURL := *queueUrlOut.QueueUrl
-	log.Info("SQS queue resolved", "queue_url", queueURL)
-
-	log.Info("sqs consumer ready, starting polling loop...")
-
-	// Polling loop — long polling (WaitTimeSeconds=20) минимизирует пустые запросы.
-	// При ошибке SQS — backoff 5 сек и продолжаем.
-	for {
-		if ctx.Err() != nil {
-			break
-		}
-
-		resp, err := sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
-			QueueUrl:            aws.String(queueURL),
-			MaxNumberOfMessages: 10,
-			WaitTimeSeconds:     20, // long polling — SQS держит соединение до 20 сек
-		})
-		if err != nil {
-			if ctx.Err() != nil {
-				break // штатное завершение
-			}
-			log.Error("SQS ReceiveMessage failed", "err", err)
-			// Backoff при ошибках SQS — не спамим запросами
-			select {
-			case <-ctx.Done():
-			case <-time.After(5 * time.Second):
-			}
-			continue
-		}
-
-		for _, sqsMsg := range resp.Messages {
-			if sqsMsg.Body == nil {
-				continue
-			}
-
-			if err := processSQSTelemetry(ctx, *sqsMsg.Body, iotStore, log); err != nil {
-				log.Error("process telemetry message", "err", err, "message_id", derefStr(sqsMsg.MessageId))
-				// НЕ удаляем сообщение — оно вернётся в очередь после visibility timeout
-				continue
-			}
-
-			// Удаляем сообщение из SQS только после успешной записи в Postgres
-			_, err := sqsClient.DeleteMessage(ctx, &sqs.DeleteMessageInput{
-				QueueUrl:      aws.String(queueURL),
-				ReceiptHandle: sqsMsg.ReceiptHandle,
-			})
-			if err != nil {
-				log.Error("SQS DeleteMessage failed", "err", err, "message_id", derefStr(sqsMsg.MessageId))
-			}
-		}
-	}
-
-	log.Info("shutting down iot-sqs-consumer")
-}
-
-// processSQSTelemetry десериализует envelope из SQS и записывает в Postgres.
-func processSQSTelemetry(ctx context.Context, body string, store *iotpg.IoTPostgresStore, log *slog.Logger) error {
-	var envelope iotTelemetryMessage
-	if err := json.Unmarshal([]byte(body), &envelope); err != nil {
-		// Битое сообщение — логируем и пропускаем (не блокируем очередь)
-		log.Warn("failed to unmarshal telemetry envelope, skipping", "err", err, "raw", body)
-		return nil
-	}
-
-	// EnsureTenantDB идемпотентен — кэшируется после первого вызова
-	if err := store.EnsureTenantDB(ctx, envelope.Namespace); err != nil {
-		return err
-	}
-
-	if err := store.InsertTelemetry(ctx, envelope.Namespace, envelope.DeviceID, envelope.Payload); err != nil {
-		return err
-	}
-
-	log.Info("telemetry saved to Postgres",
-		"namespace", envelope.Namespace,
-		"device", envelope.DeviceID,
-	)
-	return nil
-}
-
-// loadConsumerConfig читает конфигурацию из env vars.
-func loadConsumerConfig() sqsConsumerConfig {
-	required := func(key string) string {
-		v := os.Getenv(key)
-		if v == "" {
-			slog.Error("required env var not set", "key", key)
-			os.Exit(1)
-		}
-		return v
-	}
-
-	return sqsConsumerConfig{
-		SQSEndpoint:  required("SQS_ENDPOINT"),
-		SQSAccessKey: required("SQS_ACCESS_KEY"),
-		SQSSecretKey: required("SQS_SECRET_KEY"),
-		SQSQueueName: getEnvOrDefault("SQS_QUEUE_NAME", "iot-telemetry"),
-		SQSRegion:    getEnvOrDefault("SQS_REGION", "us-east-1"),
-	}
-}
-
-func getEnvOrDefault(key, defaultVal string) string {
-	if v := os.Getenv(key); v != "" {
-		return v
-	}
-	return defaultVal
-}
-
-// derefStr — безопасная разыменовка строкового указателя.
-func derefStr(s *string) string {
-	if s == nil {
-		return ""
-	}
-	return *s
-}

internal/service/api/handler/devices.go

@@ -34,7 +34,9 @@ type iotDeviceUpdateRequest struct {
 	Enabled *bool `json:"enabled"`
 }
 
-// iotDeviceResponse — ответ API. MQTTPassword — только в GET по имени.
+// iotDeviceResponse — ответ API. MQTTPassword заполняется ТОЛЬКО при создании
+// устройства (отдаётся один раз) и НИКОГДА при чтении — чтобы пароль не утекал
+// через GET и не хранился в логах/истории.
 type iotDeviceResponse struct {
 	Name          string            `json:"name"`
 	Namespace     string            `json:"namespace"`
@@ -50,7 +52,8 @@ type iotDeviceResponse struct {
 }
 
 // deviceToResponse конвертирует устройство из БД в ответ API.
-// password отдаётся только если непустая (передаётся явно из Get).
+// password отдаётся только если непустая — только на создании, при чтении
+// передаётся пустая строка (пароль повторно не показывается).
 func deviceToResponse(d *store.Device, password string) iotDeviceResponse {
 	resp := iotDeviceResponse{
 		Name:         d.Name,
@@ -193,7 +196,7 @@ func (h *Handler) GetIoTDevice(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	writeJSON(w, http.StatusOK, deviceToResponse(device, device.MQTTPassword))
+	writeJSON(w, http.StatusOK, deviceToResponse(device, ""))
 }
 
 // DeleteIoTDevice — DELETE /v1/namespaces/{ns}/iot/devices/{name}.

internal/service/api/middleware/auth.go

@@ -8,6 +8,8 @@ import (
 	"log/slog"
 	"net/http"
 	"strings"
+
+	"github.com/gorilla/mux"
 )
 
 // Auth — проверка Bearer-токена.
@@ -39,11 +41,29 @@ func Auth(authTestMode bool, hmacSecret string, log *slog.Logger, next http.Hand
 			return
 		}
 
-		if err := validateJWT(token, hmacSecret); err != nil {
+		// Fail-closed: вне тестового режима подпись ОБЯЗАТЕЛЬНА. Без секрета
+		// токен можно подделать — отказываем в обслуживании, а не проверяем «структурно».
+		if !authTestMode && hmacSecret == "" {
+			log.Error("auth: JWT_HMAC_SECRET не задан при AUTH_TEST_MODE=false — отказ (fail-closed)")
+			http.Error(w, `{"error":"auth misconfigured"}`, http.StatusInternalServerError)
+			return
+		}
+
+		sub, err := validateJWT(token, hmacSecret)
+		if err != nil {
 			log.Warn("auth: invalid token", "remote", r.RemoteAddr, "path", r.URL.Path, "reason", err.Error())
 			http.Error(w, `{"error":"invalid token"}`, http.StatusForbidden)
 			return
 		}
+
+		// Tenant-граница: `sub` из токена ОБЯЗАН совпадать с {namespace} из URL.
+		// Иначе любой валидный токен читал/писал бы любой namespace (горизонтальный доступ).
+		if ns := mux.Vars(r)["namespace"]; ns != "" && ns != sub {
+			log.Warn("auth: namespace mismatch — forbidden", "remote", r.RemoteAddr, "sub", sub, "namespace", ns, "path", r.URL.Path)
+			http.Error(w, `{"error":"forbidden"}`, http.StatusForbidden)
+			return
+		}
+
 		next.ServeHTTP(w, r)
 	})
 }
@@ -62,11 +82,11 @@ type jwtError struct{ msg string }
 func (e *jwtError) Error() string { return e.msg }
 
 // validateJWT проверяет структуру JWT (sub + exp) и, если задан secret,
-// HS256-подпись. Без secret подпись не проверяется — см. комментарий Auth.
-func validateJWT(token, secret string) error {
+// HS256-подпись. Возвращает `sub` (идентичность пользователя) для сверки с namespace.
+func validateJWT(token, secret string) (string, error) {
 	jwtParts := strings.Split(token, ".")
 	if len(jwtParts) != 3 {
-		return &jwtError{"not a JWT: expected 3 parts"}
+		return "", &jwtError{"not a JWT: expected 3 parts"}
 	}
 
 	if secret != "" {
@@ -78,17 +98,17 @@ func validateJWT(token, secret string) error {
 			headerBytes, err = base64.StdEncoding.DecodeString(jwtParts[0])
 		}
 		if err != nil || jsonUnmarshal(headerBytes, &header) != nil || header.Alg != "HS256" {
-			return &jwtError{"JWT must be HS256 when JWT_HMAC_SECRET is set"}
+			return "", &jwtError{"JWT must be HS256 when JWT_HMAC_SECRET is set"}
 		}
 		mac := hmac.New(sha256.New, []byte(secret))
 		mac.Write([]byte(jwtParts[0] + "." + jwtParts[1]))
 		expected := mac.Sum(nil)
 		sig, err := base64.RawURLEncoding.DecodeString(jwtParts[2])
 		if err != nil {
-			return &jwtError{"cannot decode JWT signature"}
+			return "", &jwtError{"cannot decode JWT signature"}
 		}
 		if !hmac.Equal(expected, sig) {
-			return &jwtError{"JWT signature mismatch"}
+			return "", &jwtError{"JWT signature mismatch"}
 		}
 	}
 
@@ -104,7 +124,7 @@ func validateJWT(token, secret string) error {
 		decoded, err = base64.StdEncoding.DecodeString(payload)
 	}
 	if err != nil {
-		return &jwtError{"cannot decode JWT payload"}
+		return "", &jwtError{"cannot decode JWT payload"}
 	}
 
 	var claims struct {
@@ -112,15 +132,15 @@ func validateJWT(token, secret string) error {
 		Exp float64 `json:"exp"`
 	}
 	if err := jsonUnmarshal(decoded, &claims); err != nil {
-		return &jwtError{"cannot parse JWT payload"}
+		return "", &jwtError{"cannot parse JWT payload"}
 	}
 	if claims.Sub == "" {
-		return &jwtError{"JWT missing sub claim"}
+		return "", &jwtError{"JWT missing sub claim"}
 	}
 	if claims.Exp > 0 {
 		if now := float64(timeNow().Unix()); claims.Exp < now {
-			return &jwtError{"JWT expired"}
+			return "", &jwtError{"JWT expired"}
 		}
 	}
-	return nil
+	return claims.Sub, nil
 }

internal/service/api/readiness.go

@@ -0,0 +1,22 @@
+// readiness.go — состояние подсистем для readiness-пробы /ready.
+package api
+
+import "sync/atomic"
+
+// Readiness отслеживает, работают ли фоновые подсистемы (bridge, consumer).
+// /health — liveness (процесс жив), /ready — readiness (подсистемы в строю).
+type Readiness struct {
+	bridgeUp   atomic.Bool
+	consumerUp atomic.Bool
+}
+
+// SetBridge — отметить состояние bridge.
+func (r *Readiness) SetBridge(up bool) { r.bridgeUp.Store(up) }
+
+// SetConsumer — отметить состояние consumer.
+func (r *Readiness) SetConsumer(up bool) { r.consumerUp.Store(up) }
+
+// Ready — true, когда обе подсистемы работают.
+func (r *Readiness) Ready() bool {
+	return r.bridgeUp.Load() && r.consumerUp.Load()
+}

internal/service/api/router.go

@@ -26,13 +26,16 @@ func corsMiddleware(next http.Handler) http.Handler {
 }
 
 // NewRouter собирает все маршруты сервиса.
-func NewRouter(h *handler.Handler, log *slog.Logger, authTestMode bool, jwtHMACSecret string, version string) http.Handler {
+func NewRouter(h *handler.Handler, log *slog.Logger, authTestMode bool, jwtHMACSecret string, version string, ready *Readiness) http.Handler {
 	r := mux.NewRouter()
 
-	// Health — для платформенных проверок контейнера.
+	// Health — для платформенных проверок контейнера (liveness).
 	r.HandleFunc("/health", healthHandler(version)).Methods(http.MethodGet)
 	r.HandleFunc("/healthz", healthHandler(version)).Methods(http.MethodGet)
 
+	// Ready — readiness: 503, если фоновые подсистемы (bridge/consumer) не в строю.
+	r.HandleFunc("/ready", readyHandler(ready)).Methods(http.MethodGet)
+
 	// Лендинг "/" — инструкция (дизайн Nubes, как у shared-sqs) + статика.
 	r.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
 		ServeLanding(w, req, version)
@@ -46,10 +49,6 @@ func NewRouter(h *handler.Handler, log *slog.Logger, authTestMode bool, jwtHMACS
 	// Админ-статистика — свой Bearer-токен.
 	r.HandleFunc("/iot-admin/stats", h.AdminStats).Methods(http.MethodGet)
 
-	// MQTT auth/acl — для EMQX, без JWT.
-	r.HandleFunc("/internal/mqtt/auth", h.MQTTAuth).Methods(http.MethodPost)
-	r.HandleFunc("/internal/mqtt/acl", h.MQTTAcl).Methods(http.MethodPost)
-
 	// /v1 — под JWT.
 	v1 := r.PathPrefix("/v1").Subrouter()
 	v1.HandleFunc("/namespaces/{namespace}/iot/devices", h.ListIoTDevices).Methods(http.MethodGet)
@@ -74,3 +73,18 @@ func healthHandler(version string) http.HandlerFunc {
 		_, _ = w.Write([]byte(`{"status":"ok","version":"` + version + `"}`))
 	}
 }
+
+// readyHandler отдаёт 200 если все подсистемы в строю, иначе 503.
+// Отделяет readiness от liveness: процесс может быть жив, но pipeline стоять.
+func readyHandler(ready *Readiness) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		if ready != nil && !ready.Ready() {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			_, _ = w.Write([]byte(`{"status":"not ready"}`))
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte(`{"status":"ready"}`))
+	}
+}

internal/service/bridge/handler.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/google/uuid"
 )
 
 // TelemetryEnvelope — сообщение, перекладываемое в SQS (разбирается consumer).
@@ -20,6 +21,9 @@ type TelemetryEnvelope struct {
 	Topic      string          `json:"topic"`
 	Payload    json.RawMessage `json:"payload"`
 	ReceivedAt string          `json:"received_at"`
+	// MessageID — уникальный id сообщения. Нужен для идемпотентности: SQS
+	// at-least-once может доставить дубликат — consumer дедуплицирует по нему.
+	MessageID string `json:"message_id"`
 }
 
 // newMessageHandler возвращает обработчик MQTT-сообщений:
@@ -51,6 +55,7 @@ func newMessageHandler(dispatcher *sqsDispatcher, log *slog.Logger) mqtt.Message
 			Topic:      topic,
 			Payload:    rawPayload,
 			ReceivedAt: time.Now().UTC().Format(time.RFC3339),
+			MessageID:  uuid.NewString(),
 		})
 	}
 }

internal/service/config/config.go

@@ -26,6 +26,7 @@ type Config struct {
 	SQSAccessKey string
 	SQSSecretKey string
 	SQSQueueName string
+	SQSDLQName   string
 	SQSRegion    string
 
 	// SQS-потребитель
@@ -109,6 +110,7 @@ func Load() (*Config, error) {
 		SQSAccessKey:         os.Getenv("SQS_ACCESS_KEY"),
 		SQSSecretKey:         os.Getenv("SQS_SECRET_KEY"),
 		SQSQueueName:         getEnv("SQS_QUEUE_NAME", DefaultSQSQueueName),
+		SQSDLQName:           getEnv("SQS_DLQ_NAME", DefaultSQSDLQName),
 		SQSRegion:            getEnv("SQS_REGION", DefaultSQSRegion),
 		SQSLongPollSeconds:   getEnvInt("SQS_LONG_POLL_SECONDS", 20),
 		SQSVisibilityTimeout: getEnvInt("SQS_VISIBILITY_TIMEOUT", 120),
@@ -149,6 +151,11 @@ func Load() (*Config, error) {
 	if cfg.MQTTPassword == "" {
 		missing = append(missing, "MQTT_PASSWORD")
 	}
+	// Вне тестового режима JWT_HMAC_SECRET ОБЯЗАТЕЛЕН: без него подпись не
+	// проверяется и токен можно подделать (fail-closed на старте).
+	if !cfg.AuthTestMode && cfg.JwtHMACSecret == "" {
+		missing = append(missing, "JWT_HMAC_SECRET (или включите AUTH_TEST_MODE для тестов)")
+	}
 	if len(missing) > 0 {
 		return nil, fmt.Errorf("required env vars not set: %s", strings.Join(missing, ", "))
 	}

internal/service/config/defaults.go

@@ -13,6 +13,11 @@ const (
 	// DefaultSQSQueueName — очередь телеметрии IoT.
 	DefaultSQSQueueName = "iot-telemetry"
 
+	// DefaultSQSDLQName — DLQ для «мёртвых» сообщений: битые envelope, которые
+	// нельзя разобрать (повторная обработка бессмысленна). Валидные данные
+	// идут в основную очередь; неразборчивые — сюда, а не отбрасываются.
+	DefaultSQSDLQName = "iot-telemetry-dlq"
+
 	// DefaultSQSRegion — регион для self-hosted SQS (не важен, оставлен для SDK).
 	DefaultSQSRegion = "us-east-1"
 

internal/service/consumer/consumer.go

@@ -6,6 +6,7 @@ package consumer
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"time"
 
@@ -29,7 +30,14 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
 	if err != nil {
 		return err
 	}
-	log.Info("consumer: SQS queue resolved", "queue", cfg.SQSQueueName)
+	// DLQ присутствует не всегда: резолвим лениво, отказ не фатален (битые
+	// сообщения тогда просто НЕ подтверждаются и оседают после ретраев).
+	dlqURL, dlqErr := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSDLQName)
+	if dlqErr != nil {
+		log.Warn("consumer: DLQ не найдена, битые сообщения будут только логироваться", "dlq", cfg.SQSDLQName, "err", dlqErr)
+		dlqURL = ""
+	}
+	log.Info("consumer: SQS queue resolved", "queue", cfg.SQSQueueName, "dlq", cfg.SQSDLQName)
 
 	for {
 		if ctx.Err() != nil {
@@ -54,18 +62,38 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
 		}
 
 		processBackoff := time.Second
+		// Копим подтверждения, но НЕ удаляем из SQS сразу: InsertTelemetry лишь
+		// буферизует и возвращает nil до фактической записи. Удаляем только после
+		// успешного Flush — иначе краш/flush-ошибка = потеря подтверждённых
+		// сообщений (P0-7).
+		type pendingAck struct {
+			handle string
+			id     string
+		}
+		var pending []pendingAck
 		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))
-				}
+				log.Warn("consumer: empty message body", "message_id", aws.ToString(msg.MessageId))
+				pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)})
 				continue
 			}
 			if err := processTelemetry(ctx, *msg.Body, store, log); err != nil {
+				// Битый envelope → DLQ (а не бесконечный ретрай и не потеря):
+				// кладём тело в DLQ, затем подтверждаем основное сообщение.
+				if errors.Is(err, errInvalidEnvelope) {
+					if dlqURL != "" {
+						if derr := sendToDLQ(ctx, sqsClient, dlqURL, *msg.Body, log); derr != nil {
+							log.Error("consumer: send to DLQ failed", "err", derr, "message_id", aws.ToString(msg.MessageId))
+						}
+					} else {
+						log.Warn("consumer: invalid envelope dropped (DLQ отсутствует)", "message_id", aws.ToString(msg.MessageId))
+					}
+					// Подтверждаем: DLQ-отправка (или логирование) — терминальная обработка.
+					pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)})
+					continue
+				}
 				log.Error("consumer: process telemetry", "err", err, "message_id", aws.ToString(msg.MessageId))
-				// Сообщение вернётся в очередь после visibility timeout.
+				// Сообщение НЕ подтверждаем — вернётся после visibility timeout.
 				// Backoff с jitter — не долбить упавший PG в цикле
 				// (фикс HIGH из ревью 2026-08-16: лавина ретраев).
 				if !sleepCtx(ctx, processBackoff) {
@@ -78,8 +106,20 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
 				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))
+			pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)})
+		}
+
+		if len(pending) == 0 {
+			continue
+		}
+		// Сначала гарантируем запись в БД, только потом удаляем из SQS.
+		if err := store.Flush(ctx); err != nil {
+			log.Error("consumer: flush failed — сообщения НЕ удалены, вернутся после visibility timeout", "err", err)
+			continue
+		}
+		for _, p := range pending {
+			if err := deleteMessage(ctx, sqsClient, queueURL, p.handle, log); err != nil {
+				log.Error("consumer: DeleteMessage failed", "err", err, "message_id", p.id)
 			}
 		}
 	}
@@ -89,10 +129,19 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
 }
 
 // deleteMessage удаляет обработанное сообщение из очереди.
-func deleteMessage(ctx context.Context, client *sqs.Client, queueURL string, receiptHandle *string, log *slog.Logger) error {
+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,
+		ReceiptHandle: aws.String(receiptHandle),
+	})
+	return err
+}
+
+// sendToDLQ — кладёт битое неразборчивое сообщение в DLQ для диагностики.
+func sendToDLQ(ctx context.Context, client *sqs.Client, dlqURL, body string, log *slog.Logger) error {
+	_, err := client.SendMessage(ctx, &sqs.SendMessageInput{
+		QueueUrl:    aws.String(dlqURL),
+		MessageBody: aws.String(body),
 	})
 	return err
 }

internal/service/consumer/processor.go

@@ -4,26 +4,39 @@ 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 — пропускаем (не блокируем очередь).
+// Битый 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: failed to unmarshal envelope, skipping", "err", err)
-		return 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
 	}
-	if err := store.InsertTelemetry(ctx, envelope.Namespace, envelope.DeviceID, envelope.Payload); err != nil {
+	// Время приёма сохраняем в 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
 	}
 

internal/storage/iotpg/iot_telemetry_store.go

@@ -50,8 +50,10 @@ type IoTPostgresStore struct {
 
 // telemetryInsert — строка для батч-вставки.
 type telemetryInsert struct {
-	deviceID string
-	payload  []byte
+	deviceID  string
+	messageID string
+	ts        time.Time
+	payload   []byte
 }
 
 // telemetryBatchSize — размер батча перед синхронным flush.
@@ -112,14 +114,17 @@ func (s *IoTPostgresStore) startFlusher() {
 			case <-s.stopFlush:
 				return
 			case <-t.C:
-				s.flushAll(false)
+				s.flushAll(context.Background())
 			}
 		}
 	}()
 }
 
-// flushAll — flush всех накопленных батчей. reportErr=false: только лог.
-func (s *IoTPostgresStore) flushAll(reportErr bool) error {
+// flushAll — flush всех накопленных батчей под переданным ctx.
+// ВАЖНО: при ошибке записи тенанта строки НЕ теряются — возвращаются обратно
+// в буфер и будут повторены. Иначе фоновый flusher мог бы молча потерять
+// телеметрию, а consumer, увидев пустой буфер в Flush, подтвердил бы SQS-сообщение.
+func (s *IoTPostgresStore) flushAll(ctx context.Context) error {
 	s.mu.Lock()
 	if len(s.batches) == 0 {
 		s.mu.Unlock()
@@ -131,35 +136,67 @@ func (s *IoTPostgresStore) flushAll(reportErr bool) error {
 
 	var firstErr error
 	for ns, rows := range batches {
-		if err := s.flushTenant(ns, rows); err != nil {
+		if err := s.flushTenant(ctx, ns, rows); err != nil {
 			s.log.Error("iotpg: flush batch", "namespace", ns, "rows", len(rows), "err", err)
 			if firstErr == nil {
 				firstErr = err
 			}
+			if isRetryableFlushErr(err) {
+				// Сеть/PG временно недоступен — возвращаем строки в буфер, повторим.
+				s.mu.Lock()
+				s.batches[ns] = append(s.batches[ns], rows...)
+				s.mu.Unlock()
+			} else {
+				// Постоянная ошибка (неверные данные/constraint) — повтор бессмыслен:
+				// ВЫБРАСЫВАЕМ батч, иначе он зависнет в буфере навсегда и заблокирует
+				// весь namespace (вечные 200мс-ретраи). Логируем подробно.
+				s.log.Error("iotpg: permanent flush error, batch dropped", "namespace", ns, "rows", len(rows), "err", err)
+			}
 		}
 	}
 	return firstErr
 }
 
+// Flush — принудительно синхронно пишет все накопленные батчи телеметрии в БД и
+// возвращает первую ошибку. Нужен consumer'у, чтобы подтверждать (удалять из SQS)
+// ТОЛЬКО фактически записанные данные: без этого краш/flush-ошибка после
+// InsertTelemetry (которая лишь буферизует и возвращает nil) = потеря сообщений.
+func (s *IoTPostgresStore) Flush(ctx context.Context) error {
+	return s.flushAll(ctx)
+}
+
 // flushTenant — INSERT батча строк одного тенанта (multi-VALUES).
-func (s *IoTPostgresStore) flushTenant(ns string, rows []telemetryInsert) error {
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+func (s *IoTPostgresStore) flushTenant(ctx context.Context, ns string, rows []telemetryInsert) error {
+	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
 	defer cancel()
 	tenantDB, err := s.getTenantDB(ctx, ns)
 	if err != nil {
 		return err
 	}
-	// INSERT INTO iot_telemetry (device_id, payload) VALUES ($1,$2),($3,$4),...
+	// INSERT INTO iot_telemetry (device_id, payload, message_id, ts) VALUES ($1,$2,$3,$4),...
+	// ts — ВРЕМЯ ПРИЁМА (bridge ReceivedAt), а не момент INSERT: иначе батчинг и
+	// параллельные записи ломают порядок телеметрии (ORDER BY ts DESC).
+	// Zero → now() (fallback для legacy-сообщений без received_at).
+	// ON CONFLICT DO NOTHING — дедуп по (device_id, message_id) (P0-8).
 	var sb strings.Builder
-	sb.WriteString("INSERT INTO iot_telemetry (device_id, payload) VALUES ")
-	args := make([]any, 0, len(rows)*2)
+	sb.WriteString("INSERT INTO iot_telemetry (device_id, payload, message_id, ts) VALUES ")
+	args := make([]any, 0, len(rows)*4)
 	for i, r := range rows {
 		if i > 0 {
 			sb.WriteString(",")
 		}
-		sb.WriteString(fmt.Sprintf("($%d,$%d)", i*2+1, i*2+2))
-		args = append(args, r.deviceID, r.payload)
+		sb.WriteString(fmt.Sprintf("($%d,$%d,$%d,$%d)", i*4+1, i*4+2, i*4+3, i*4+4))
+		var mid any
+		if r.messageID != "" {
+			mid = r.messageID
+		}
+		ts := r.ts
+		if ts.IsZero() {
+			ts = time.Now().UTC()
+		}
+		args = append(args, r.deviceID, r.payload, mid, ts)
 	}
+	sb.WriteString(" ON CONFLICT DO NOTHING")
 	_, err = tenantDB.ExecContext(ctx, sb.String(), args...)
 	return err
 }
@@ -262,13 +299,18 @@ func (s *IoTPostgresStore) EnsureTenantDB(ctx context.Context, namespace string)
 	}
 	_, err = tenantDB.ExecContext(ctx, `
 CREATE TABLE IF NOT EXISTS iot_telemetry (
-id        BIGSERIAL PRIMARY KEY,
-device_id TEXT        NOT NULL,
-ts        TIMESTAMPTZ NOT NULL DEFAULT now(),
-payload   JSONB       NOT NULL
+id         BIGSERIAL PRIMARY KEY,
+device_id  TEXT        NOT NULL,
+message_id TEXT,
+ts         TIMESTAMPTZ NOT NULL DEFAULT now(),
+payload    JSONB       NOT NULL
 );
 CREATE INDEX IF NOT EXISTS idx_iot_telemetry_device_ts
 ON iot_telemetry (device_id, ts DESC);
+-- Миграция существующих таблиц + уникальный ключ дедупликации (P0-8).
+ALTER TABLE iot_telemetry ADD COLUMN IF NOT EXISTS message_id TEXT;
+CREATE UNIQUE INDEX IF NOT EXISTS uniq_iot_telemetry_device_msg
+ON iot_telemetry (device_id, message_id);
 `)
 	if err != nil {
 		return err
@@ -280,11 +322,16 @@ ON iot_telemetry (device_id, ts DESC);
 // InsertTelemetry ставит строку в батч-буфер тенанта.
 // При накоплении telemetryBatchSize строк батч пишется синхронно (ошибка
 // возвращается вызывающему); неполные батчи дописывает фоновый flusher.
-func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, deviceID string, payload json.RawMessage) error {
+// messageID — уникальный id сообщения для дедупликации (может быть пустым →
+// тогда строка не участвует в дедупе).
+// receivedAt — время приёма сообщения (для колонки ts); zero → стор подставит now().
+func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, deviceID, messageID string, receivedAt time.Time, payload json.RawMessage) error {
 	s.mu.Lock()
 	s.batches[namespace] = append(s.batches[namespace], telemetryInsert{
-		deviceID: deviceID,
-		payload:  append([]byte(nil), payload...),
+		deviceID:  deviceID,
+		messageID: messageID,
+		ts:        receivedAt,
+		payload:   append([]byte(nil), payload...),
 	})
 	full := len(s.batches[namespace]) >= telemetryBatchSize
 	var rows []telemetryInsert
@@ -295,11 +342,44 @@ func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, devic
 	s.mu.Unlock()
 
 	if full {
-		return s.flushTenant(namespace, rows)
+		if err := s.flushTenant(ctx, namespace, rows); err != nil {
+			if isRetryableFlushErr(err) {
+				// Не теряем строки: возвращаем в буфер (сеть/PG временно лежит).
+				s.mu.Lock()
+				s.batches[namespace] = append(s.batches[namespace], rows...)
+				s.mu.Unlock()
+			} else {
+				s.log.Error("iotpg: permanent error, batch dropped", "namespace", namespace, "rows", len(rows), "err", err)
+			}
+			return err
+		}
 	}
 	return nil
 }
 
+// isRetryableFlushErr — классифицирует ошибку INSERT: retryable (сеть, падение
+// PG, таймаут) или permanent (неверные данные/нарушение constraint). Permanent
+// НЕ должно возвращаться в буфер — иначе один «ядовитый» payload зациклит flush
+// namespace навсегда.
+func isRetryableFlushErr(err error) bool {
+	var pqErr *pq.Error
+	if errors.As(err, &pqErr) {
+		switch {
+		case strings.HasPrefix(string(pqErr.Code), "08"): // connection exception
+			return true
+		case pqErr.Code == "57P01" || pqErr.Code == "57P02" || pqErr.Code == "57P03": // admin shutdown / crash / cannot connect now
+			return true
+		case strings.HasPrefix(string(pqErr.Code), "22"): // data exception (invalid data)
+			return false
+		case strings.HasPrefix(string(pqErr.Code), "23"): // integrity constraint violation
+			return false
+		}
+	}
+	// Не-pq ошибка (например, context.DeadlineExceeded, net timeout) — считаем
+	// retryable: это среда, а не данные.
+	return true
+}
+
 // QueryTelemetry читает телеметрию из tenant DB (ts DESC).
 // deviceID — фильтр (пустая строка = все устройства). limit — max записей
 // (50..1000), offset — смещение для пагинации.
@@ -463,7 +543,7 @@ func isDBNotExistErr(err error) bool {
 func (s *IoTPostgresStore) Close() error {
 	close(s.stopFlush)
 	s.flushWG.Wait()
-	_ = s.flushAll(true)
+	_ = s.flushAll(context.Background())
 	s.tenants.Range(func(_, value any) bool {
 		if db, ok := value.(*sql.DB); ok {
 			db.Close()