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

# Журнал сессии — 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 ./...` — зелёные на каждом шаге.