# IoT ↔ SQS: сводка всех находок и мыслей (handoff для нового чата/репы) > Дата: 2026-10-05. > Назначение: полностью самодостаточный документ — чтобы, открыв его в НОВОЙ > репе или новом чате, можно было продолжить без чтения истории. > Вся информация ниже взята из реального кода/доков репозитория IoT > (`/home/naeel/nubes/IoT`), а НЕ из памяти и НЕ из чужих заметок. --- ## 0. КЛЮЧЕВОЙ РАЗВОРОТ: что в какой репе **Есть ДВА разных проекта, в ДВУХ разных репах:** | Проект | Репа | Что делает | Состояние | |---|---|---|---| | **shared-sqs** | отдельная репа (SQS-service) | SQS-совместимая очередь (AWS API, Go, форк GoAWS), multi-tenant | **ГОТОВ, работает на проде v0.1.35** (все тесты зелёные, нагрузка без ошибок) | | **IoT** | эта репа (`/home/naeel/nubes/IoT`) | телеметрия устройств: MQTT → SQS → PostgreSQL → REST | **MVP, не production-ready** (есть подтверждённые блокеры) | **Зависимость:** IoT **потребляет** shared-sqs как внешний сервис (очередь телеметрии `iot-telemetry`). shared-sqs от IoT **не зависит**. **Вывод о порядке (отвечая на вопрос «что сначала — SQS или IoT»):** - shared-sqs фактически **уже готов и работает** раньше, чем IoT дотянет до production. Его «доделывать» не надо для старта работ по IoT. - Порядок работ по сути такой: **SQS уже есть (работает) → теперь доработки IoT**, который на него опирается. - Если планируется создавать оба «с нуля» — логический порядок архитектурно: сначала SQS, потом IoT (т.к. IoT от него зависит). **ВНИМАНИЕ про ветки IoT-репы (важно, легко запутаться):** - Актуальный код IoT (монолит `iot-service`, `internal/service/`, `internal/storage/iotpg/`) лежит на ветке **`feature/replace-kafka-with-sqs`**. Имя ветки — ПЕРЕЖИТОК старой миграции Kafka→SQS (апрель 2026), к текущей задаче отношения не имеет. Не путать имя с сутью: это текущая ветка IoT. - Ветка **`master`** — СТАРАЯ, без монолита (там легаси: `internal/api/`, CRD, k8s-манифесты, `deployments/`, `examples/`, `config/crd/`). НЕ актуальна. - На `master` ошибочно закоммичены 3 файла документации (см. раздел 5). --- ## 1. Что такое текущий IoT (архитектура) **Монолит `iot-service`** — один Go-бинарник с тремя ролями в одном процессе: 1. **REST API** + MQTT auth/acl на порту `:9090` (устройства хранятся в PostgreSQL); 2. **bridge** — MQTT-подписка на EMQX → отправка в shared-SQS; 3. **consumer** — чтение из shared-SQS (long-poll) → запись в per-tenant PostgreSQL. **Конвейер телеметрии:** ``` устройство → wss → EMQX (authn+authz напрямую в PG) → бридж → shared-SQS → consumer → per-tenant PostgreSQL → REST API (/v1/namespaces/{ns}/iot/telemetry) ``` **Стек:** - Go 1.25, gorilla/mux, paho.mqtt.golang, aws-sdk-go-v2 (SQS), lib/pq. - EMQX 5.5.1 (образ `naeel/iot-emqx`), shared-SQS (SQS-совместимая очередь), PostgreSQL managed (per-tenant БД). - Модуль: `gitea.services.ngcloud.ru/Nail/IoT`. **Ключевые файлы актуального кода (на ветке feature):** - `cmd/iot-service/main.go` — точка входа: config → PG → SQS → API + горутины bridge/consumer. - `internal/service/config/config.go`, `defaults.go` — env-конфигурация. - `internal/service/api/router.go`, `middleware/auth.go` — маршруты + JWT. - `internal/service/api/handler/*.go` — devices, mqtt_auth, telemetry, admin, sqsstats. - `internal/service/store/*.go` — устройства в PG (`iot_devices`). - `internal/storage/iotpg/iot_telemetry_store.go` — запись телеметрии (batch). - `internal/service/bridge/*.go` — bridge → SQS. - `internal/service/consumer/*.go` — SQS → PG. - `internal/service/sqsclient/sqsclient.go` — SQS-клиент (endpoint override). - `emqx/emqx.conf` — конфиг EMQX (auth/ACL — напрямую в PostgreSQL). --- ## 2. Как IoT использует shared-sqs (факты) - Внутренний endpoint shared-sqs (тот же кластер Nubes, realm `iot-naeel`): `http://containerk8s.f1ffb134-7d16-45bd-8bef-69f6ec8ab33c.svc.cluster.local:4100` (вшит дефолтом в `config/defaults.go`). - Очередь: `iot-telemetry`, регион `us-east-1`. - Тенант IoT: `t-fec713a719ad0b33`; креды детерминированы из email: tenantID `t-`+sha256(email)[:8], AccessKey `SSAK-`+sha256(email)[:12], SecretKey sha256(email+":shared-sqs:secret-key:v1"). - Лимиты shared-sqs: 50 очередей на тенанта, сообщение до 256 КБ, Receive 1–10. - В бридже: неблокирующий SQS-диспетчер (канал 10000 + 4 worker'а, ретраи 1/2/4с), дроп при переполнении буфера и при oversize >250 КБ. - В consumer: long-poll 20с, VisibilityTimeout 120с (в коде), backoff при сбое PG (1с→60с), DeleteMessage после обработки. --- ## 3. ПОДТВЕРЖДЁННЫЕ БЛОКЕРЫ IoT (для доработок) > Источник: два независимых анализа + ревью Opus + моя перепроверка построчно. > Все — по КОДУ, с точными ссылками. Сгруппировано по приоритету. ### P0 — компрометация/потеря данных или доступ к чужим данным **P0-1. Tenant-изоляция не работает.** - `internal/service/api/middleware/auth.go` (`validateJWT`, ~стр. 62, 120): проверяет только наличие `sub` и (необязательно) `exp`, а `sub` НИКАК не сверяется с `{namespace}` из URL. - Обработчики берут namespace из пути (`handler/devices.go:91`, `handler/telemetry.go:19`) без сверки с токеном. - Итог: **любой валидный токен читает/пишет ЛЮБОЙ namespace** — горизонтальный доступ к чужим устройствам и телеметрии. **P0-2. Проверка подписи JWT необязательна.** - `auth.go` (~стр. 70): при пустом `JWT_HMAC_SECRET` подпись НЕ проверяется — только структура токена. Токен подделывается вручную (`{"sub":"x","exp":...}`). - `config/config.go` (~стр. 118): дефолт секрета пустой, секрет не в списке required env. **P0-3. Пароли устройств в открытом виде.** - `store/open.go:21`: `mqtt_password TEXT` — хранится открытым текстом. - `handler/devices.go:189`: пароль отдаётся повторно через `GET /devices/{name}`. - `emqx/emqx.conf:46`: `password_hash_algorithm name=plain`. - Компрометация БД = все пароли устройств + возможность выдать их повторно. **P0-7. Потеря телеметрии: SQS-удаление до фактической записи.** - `consumer/consumer.go:78`: удаляет сообщение из SQS после `processTelemetry`==nil. - `iotpg/iot_telemetry_store.go:356-376`: `InsertTelemetry` при неполном батче кладёт строку В ПАМЯТЬ и возвращает nil (запись ещё не сделана). - `iotpg/iot_telemetry_store.go:129-135` (`flushAll`): подменяет буфер на пустой ДО записи в PG, при ошибке — только логирует. - Итог: краш между «кладём в буфер» и «flush» → сообщение уже удалено из SQS, в PG не попало → **безвозвратная потеря**. **P0-8. Нет идемпотентности (дубли).** - `iot_telemetry` таблица: `id BIGSERIAL`, без ключа дедупликации. - `bridge/handler.go:19`: envelope без message-id. - At-least-once SQS → **молчаливые дубли** при повторе сообщения. ### P1 — важные риски эксплуатации **P1-4. Лишние HTTP-auth/ACL-эндпоинты.** - `router.go:50-51`: `/internal/mqtt/auth` и `/internal/mqtt/acl` висят на публичном порту 9090 БЕЗ аутентификации. - Но EMQX ходит по PG-auth напрямую (`emqx/emqx.conf`, `backend=postgresql`), эти эндпоинты — НЕиспользуемый fallback. Рекомендация: **удалить**, а не защищать. - Риск: кто достучался до порта — перебирает пароли устройств и получает ACL. **P1-5. `AUTH_TEST_MODE=true` на проде.** - Был включён для диагностики DevOps (HISTORY §38); любой Bearer-токен проходит. - Нужно выставить `false` + задать `JWT_HMAC_SECRET`. **P1-9. `/health` не отражает состояние.** - `router.go:72`: `healthHandler` всегда `{"status":"ok"}`, не проверяет bridge/consumer/PG/SQS. - `main.go`: ошибки `bridge.Run`/`consumer.Run` только логируются; горутина завершается, а liveness остаётся зелёным. **P1-10. Дропы в бридже без метрик.** - `bridge/sender.go:73`: дроп при переполнении RAM-очереди (10000). - `bridge/sender.go:112`: дроп после исчерпания ретраев. ### P2 — обвязка/гигиена **P2-6. TLS.** `sslmode=require` (без verify-full) к PG; EMQX `verify=verify_none` (`emqx.conf:40`). **P2-11/P2-12. Нет тестов и CI.** `*_test.go` отсутствует; `.github/workflows` нет (в `.github/` только правила). **P2-13. `/metrics` заявлен, но не зарегистрирован.** - `config/defaults.go:6`: в комментарии упомянут `/metrics`, но роут в `router.go` НЕ создан. prometheus — только indirect dep. Rate-limiting нет. **P2-14. Легаси не вычищено.** - Остались `cmd/iot-operator`, `controllers/`, `internal/api/` (старый k8s-код), `cmd/mqtt-bridge`, `cmd/sqs-consumer`; k8s/controller-runtime в `go.mod` (~12 строк); бинарники в `bin/`, `ws-probe/ws-probe`. **P2-15. Дрейф доков.** - `doc/deployment-nubes-production.md` — образы `v0.1.2`; Makefile — `v0.1.13`. - `loadtests/README.md` — visibility «30с» (в коде 120с), описывает уже исправленную блокировку MQTT-обработчика. **P2-16. `emqx.conf`: `cookie = "iot-emqx-cookie-mvp"` захардкожен.** ### ПЛАТФОРМА (вне нашего кода, но блокирует прод) 1. **wss-обрывы ~150с** на внешнем шлюзе Nubes (детерминированно, доказано с 3 независимых точек: станция, Vultr; внутри кластера и с ВМ — чисто). Тикет Nubes подготовлен (`doc/thinking/nubes-devops-report-2026-08-16.md`). 2. **MSS/MTU-баг** — плавающий обрыв ответов >~15–64 КБ (баг нестабилен по маршруту/времени). Обход в коде: offset-пагинация API, внутренний endpoint SQS. --- ## 4. Что уже ХОРОШО (не трогать без нужды) - Конвейер end-to-end работает и проверен на проде (HISTORY: baseline 2390/2390 без потерь/дублей, ACL-нарушения отсекаются, reconnect-storm чистый). - Уже исправлены (по итогам ревью 2026-08-16): асинхронный SQS-диспатчер, VisibilityTimeout 120с, backoff, graceful shutdown, batch-INSERT) — НЕ регрессировать это. - EMQX auth/ACL — напрямую в PostgreSQL, без зависимости от монолита. - MQTT-reconnect с повторной подпиской (OnConnectHandler). - Документация: деплой без kubectl, интеграция с SQS, 10 нагрузочных сценариев. --- ## 5. Что уже записано в этой репе (файлы документации) - `doc/thinking/readiness-assessment-2026-10-05.md` — два анализа (5/10 и 60–65%) + моё мнение. - `doc/thinking/remediation-plan-2026-10-05.md` — план доработок (4 трека, P0/P1/P2). - `doc/thinking/opus-review-prompt.md` — промпт для ревью (allowlist файлов). - **ВНИМАНИЕ:** эти 3 файла лежат в ветке `master` (коммит `a4e8fd5`, запушен), а НЕ на актуальной `feature/replace-kafka-with-sqs`, где живёт код. `doc/sqs-integration.md`, `doc/deployment-nubes-production.md`, `HISTORY/` — на feature-ветке. **Backup-ветки (сохранение состояний):** - `backup/2026-08-17-feature-sqs` (старая точка). - `backup/2026-10-05-feature-sqs` (сделана 2026-10-05, на `12db7d3`). --- ## 6. Рекомендуемый порядок работ (по плану) 1. P0-1 + P0-2 — tenant-граница (связать `sub` с namespace) + обязательная подпись JWT. 2. P0-7 + P0-8 — доставка без потерь (ack только после успешного flush) + dedup (message-id). 3. P0-5 — выключить `AUTH_TEST_MODE`. 4. P0-3 + P1-4 — пароли (не отдавать повторно/хешировать) + удалить `/internal/mqtt/*`. 5. P1-9 + P1-10 — честный `/health` + метрики дропов. 6. P1-11 + P1-12 — Go-тесты + CI. 7. P2 (гигиена: легаси, доки, cookie, TLS).