From 9d08473d5fa64f0cd223ebb1803d55ee32860028 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= Date: Sun, 16 Aug 2026 10:07:51 +0400 Subject: [PATCH] =?UTF-8?q?feat:=20=D0=BC=D0=BE=D0=BD=D0=BE=D0=BB=D0=B8?= =?UTF-8?q?=D1=82=20iot-service=20(=D0=A4=D0=B0=D0=B7=D0=B0=201)=20?= =?UTF-8?q?=E2=80=94=20config/store/api/bridge/consumer=20=D0=B1=D0=B5?= =?UTF-8?q?=D0=B7=20k8s,=20=D1=83=D1=81=D1=82=D1=80=D0=BE=D0=B9=D1=81?= =?UTF-8?q?=D1=82=D0=B2=D0=B0=20=D0=B2=20PG,=20=D0=BD=D0=BE=D0=B2=D1=8B?= =?UTF-8?q?=D0=B5=20Dockerfile/Makefile?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Dockerfile | 38 + Makefile | 27 + cmd/iot-service/main.go | 123 ++ internal/service/api/admin_embed.go | 20 + internal/service/api/console_embed.go | 20 + internal/service/api/handler/admin.go | 58 + internal/service/api/handler/devices.go | 244 +++ internal/service/api/handler/handler.go | 49 + internal/service/api/handler/mqtt_auth.go | 167 +++ internal/service/api/handler/sqsstats.go | 52 + internal/service/api/handler/telemetry.go | 43 + internal/service/api/middleware/auth.go | 101 ++ .../service/api/middleware/auth_helpers.go | 15 + internal/service/api/middleware/logging.go | 35 + internal/service/api/router.go | 70 + internal/service/api/ui/iot-admin.html | 547 +++++++ internal/service/api/ui/iot-console.html | 1330 +++++++++++++++++ internal/service/bridge/bridge.go | 98 ++ internal/service/bridge/handler.go | 71 + internal/service/config/config.go | 150 ++ internal/service/config/defaults.go | 32 + internal/service/consumer/consumer.go | 89 ++ internal/service/consumer/processor.go | 35 + internal/service/sqsclient/sqsclient.go | 38 + internal/service/store/device.go | 29 + internal/service/store/devices.go | 173 +++ internal/service/store/open.go | 62 + 27 files changed, 3716 insertions(+) create mode 100644 Dockerfile create mode 100644 Makefile create mode 100644 cmd/iot-service/main.go create mode 100644 internal/service/api/admin_embed.go create mode 100644 internal/service/api/console_embed.go create mode 100644 internal/service/api/handler/admin.go create mode 100644 internal/service/api/handler/devices.go create mode 100644 internal/service/api/handler/handler.go create mode 100644 internal/service/api/handler/mqtt_auth.go create mode 100644 internal/service/api/handler/sqsstats.go create mode 100644 internal/service/api/handler/telemetry.go create mode 100644 internal/service/api/middleware/auth.go create mode 100644 internal/service/api/middleware/auth_helpers.go create mode 100644 internal/service/api/middleware/logging.go create mode 100644 internal/service/api/router.go create mode 100644 internal/service/api/ui/iot-admin.html create mode 100644 internal/service/api/ui/iot-console.html create mode 100644 internal/service/bridge/bridge.go create mode 100644 internal/service/bridge/handler.go create mode 100644 internal/service/config/config.go create mode 100644 internal/service/config/defaults.go create mode 100644 internal/service/consumer/consumer.go create mode 100644 internal/service/consumer/processor.go create mode 100644 internal/service/sqsclient/sqsclient.go create mode 100644 internal/service/store/device.go create mode 100644 internal/service/store/devices.go create mode 100644 internal/service/store/open.go diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..17b15c5 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,38 @@ +# Dockerfile — монолит iot-service (один бинарник, один контейнер). +# Образ: naeel/iot-service (Docker Hub, публичный — требование платформы). +# +# Env-дефолты (не секреты) вшиты сюда — deck-UI задаёт только секреты. + +FROM golang:1.25 AS builder +WORKDIR /workspace + +COPY go.mod go.sum ./ +RUN go mod download + +COPY cmd/ cmd/ +COPY internal/ internal/ + +ARG VERSION=dev +RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a \ + -ldflags "-X main.version=${VERSION}" -o iot-service ./cmd/iot-service/ + +FROM gcr.io/distroless/static:nonroot +WORKDIR / +COPY --from=builder /workspace/iot-service . + +USER 65532:65532 +EXPOSE 9090 + +ENV API_PORT=9090 \ + SQS_ENDPOINT=http://containerk8s.f1ffb134-7d16-45bd-8bef-69f6ec8ab33c.svc.cluster.local:4100 \ + SQS_QUEUE_NAME=iot-telemetry \ + SQS_REGION=us-east-1 \ + SQS_LONG_POLL_SECONDS=20 \ + SQS_VISIBILITY_TIMEOUT=30 \ + MQTT_CLIENT_ID=iot-bridge \ + MQTT_PORT=8083 \ + MQTT_WS_PATH=/mqtt \ + AUTH_TEST_MODE=false \ + LOG_LEVEL=info + +ENTRYPOINT ["/iot-service"] diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..8a1422a --- /dev/null +++ b/Makefile @@ -0,0 +1,27 @@ +# Makefile — монолит iot-service (образ naeel/iot-service). +# Старые k8s-цели — в legacy/Makefile.old. + +VERSION ?= v0.1.0 +IMAGE ?= naeel/iot-service +LDFLAGS = -X main.version=$(VERSION) + +.PHONY: build vet test tidy docker-build docker-push + +build: + CGO_ENABLED=0 go build -ldflags "$(LDFLAGS)" -o bin/iot-service ./cmd/iot-service/ + +vet: + go vet ./... + +test: + go test ./... + +tidy: + go mod tidy + +docker-build: + docker build --build-arg VERSION=$(VERSION) -t $(IMAGE):$(VERSION) -t $(IMAGE):latest . + +docker-push: + docker push $(IMAGE):$(VERSION) + docker push $(IMAGE):latest diff --git a/cmd/iot-service/main.go b/cmd/iot-service/main.go new file mode 100644 index 0000000..3bf6286 --- /dev/null +++ b/cmd/iot-service/main.go @@ -0,0 +1,123 @@ +// iot-service — монолит IoT на платформе Nubes (контейнер «Простой HTTP»). +// +// Три роли в одном процессе: +// - REST API + MQTT auth/acl (:9090, устройства в PostgreSQL); +// - bridge — MQTT-подписка EMQX → shared-SQS; +// - consumer — shared-SQS → per-tenant PostgreSQL. +// +// Наследие старого IoT (k8s, оператор, CRD) НЕ используется. +package main + +import ( + "context" + "errors" + "log/slog" + "net/http" + "os" + "os/signal" + "sync" + "syscall" + "time" + + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/api" + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/api/handler" + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/bridge" + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/config" + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/consumer" + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/sqsclient" + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/store" + "gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg" +) + +// version — подставляется через ldflags при сборке. +var version = "dev" + +func main() { + cfg, err := config.Load() + if err != nil { + slog.Error("config", "err", err) + os.Exit(1) + } + + log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: cfg.LogLevel})) + slog.SetDefault(log) + + log.Info("iot-service starting", "version", version, "port", cfg.APIPort) + + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT) + defer cancel() + + // PostgreSQL — устройства и телеметрия. + storeCtx, storeCancel := context.WithTimeout(ctx, 15*time.Second) + devices, err := store.Open(storeCtx, cfg.IOTPGDSN) + storeCancel() + if err != nil { + log.Error("devices store", "err", err) + os.Exit(1) + } + defer devices.Close() + log.Info("devices store ready") + + iotStore, err := iotpg.New(cfg.IOTPGDSN, log) + if err != nil { + log.Error("iotpg store", "err", err) + os.Exit(1) + } + defer iotStore.Close() + log.Info("telemetry store ready") + + sqsClient := sqsclient.New(cfg) + + h := &handler.Handler{ + Devices: devices, + IoTPG: iotStore, + SQS: sqsClient, + Cfg: cfg, + Log: log, + Version: version, + StartedAt: time.Now(), + } + + // HTTP API — в основном потоке. + router := api.NewRouter(h, log, cfg.AuthTestMode, version) + server := &http.Server{ + Addr: ":" + cfg.APIPort, + Handler: router, + ReadHeaderTimeout: 10 * time.Second, + } + + // bridge + consumer — в фоне. + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + if err := bridge.Run(ctx, cfg, sqsClient, log); err != nil && !errors.Is(err, context.Canceled) { + log.Error("bridge stopped", "err", err) + } + }() + + go func() { + defer wg.Done() + if err := consumer.Run(ctx, cfg, sqsClient, iotStore, log); err != nil && !errors.Is(err, context.Canceled) { + log.Error("consumer stopped", "err", err) + } + }() + + // Сервер: при завершении ctx — гасим. + go func() { + <-ctx.Done() + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer shutdownCancel() + _ = server.Shutdown(shutdownCtx) + }() + + log.Info("iot-service listening", "addr", server.Addr) + if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Error("http server", "err", err) + os.Exit(1) + } + + wg.Wait() + log.Info("iot-service stopped") +} diff --git a/internal/service/api/admin_embed.go b/internal/service/api/admin_embed.go new file mode 100644 index 0000000..f96036a --- /dev/null +++ b/internal/service/api/admin_embed.go @@ -0,0 +1,20 @@ +// admin_embed.go — встраивает HTML страницы администратора в бинарник. +package api + +import ( + _ "embed" + "net/http" +) + +// iotAdminHTML — страница /iot-admin, встроена при сборке. +// +//go:embed ui/iot-admin.html +var iotAdminHTML []byte + +// ServeIoTAdmin отдаёт HTML страницу администратора. +func ServeIoTAdmin(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.Header().Set("Cache-Control", "no-cache, must-revalidate") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(iotAdminHTML) +} diff --git a/internal/service/api/console_embed.go b/internal/service/api/console_embed.go new file mode 100644 index 0000000..8c999b1 --- /dev/null +++ b/internal/service/api/console_embed.go @@ -0,0 +1,20 @@ +// console_embed.go — встраивает HTML IoT консоли в бинарник. +package api + +import ( + _ "embed" + "net/http" +) + +// iotConsoleHTML — консоль /console, встроена при сборке. +// +//go:embed ui/iot-console.html +var iotConsoleHTML []byte + +// ServeIoTConsole отдаёт HTML консоли. +func ServeIoTConsole(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.Header().Set("Cache-Control", "no-cache, must-revalidate") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(iotConsoleHTML) +} diff --git a/internal/service/api/handler/admin.go b/internal/service/api/handler/admin.go new file mode 100644 index 0000000..455b76a --- /dev/null +++ b/internal/service/api/handler/admin.go @@ -0,0 +1,58 @@ +// admin.go — статистика для страницы администратора. +// +// Endpoint: GET /iot-admin/stats (Bearer ADMIN_STATS_TOKEN). +// Источники: PostgreSQL (iotpg), shared-SQS (атрибуты очереди), рантайм процесса. +// k8s-статусы подов убраны — компоненты теперь в одном процессе. +package handler + +import ( + "context" + "net/http" + "strings" + "time" +) + +// AdminStats — GET /iot-admin/stats. +func (h *Handler) AdminStats(w http.ResponseWriter, r *http.Request) { + if h.Cfg.AdminStatsToken == "" { + writeJSON(w, http.StatusServiceUnavailable, errResp("admin stats not configured: ADMIN_STATS_TOKEN not set")) + return + } + + authHeader := r.Header.Get("Authorization") + if !strings.HasPrefix(authHeader, "Bearer ") || strings.TrimPrefix(authHeader, "Bearer ") != h.Cfg.AdminStatsToken { + writeJSON(w, http.StatusUnauthorized, errResp("unauthorized")) + return + } + + ctx, cancel := context.WithTimeout(r.Context(), 15*time.Second) + defer cancel() + + result := map[string]any{ + "collected_at": time.Now().UTC(), + } + + // PostgreSQL: статистика по всем tenant (как в старом коде). + if h.IoTPG != nil { + pgStats, err := h.IoTPG.GetAdminStats(ctx) + if err != nil { + result["postgres"] = map[string]any{"reachable": false, "error": err.Error()} + } else { + result["postgres"] = pgStats + } + } else { + result["postgres"] = map[string]any{"reachable": false, "error": "IoTPG not configured"} + } + + // SQS: approximate counts очереди телеметрии. + result["sqs"] = h.collectSQSStats(ctx) + + // Рантайм процесса (вместо статусов k8s-подов). + result["runtime"] = map[string]any{ + "version": h.Version, + "uptime_sec": int64(time.Since(h.StartedAt).Seconds()), + "started_at": h.StartedAt.UTC().Format(time.RFC3339), + } + + writeJSON(w, http.StatusOK, result) +} diff --git a/internal/service/api/handler/devices.go b/internal/service/api/handler/devices.go new file mode 100644 index 0000000..b5a9617 --- /dev/null +++ b/internal/service/api/handler/devices.go @@ -0,0 +1,244 @@ +// devices.go — CRUD устройств на PostgreSQL. +// +// Endpoints: +// +// POST /v1/namespaces/{ns}/iot/devices — создать +// GET /v1/namespaces/{ns}/iot/devices — список (без паролей) +// GET /v1/namespaces/{ns}/iot/devices/{name} — одно (с паролем) +// DELETE /v1/namespaces/{ns}/iot/devices/{name} — удалить +// PATCH /v1/namespaces/{ns}/iot/devices/{name} — enabled +package handler + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "net/http" + "time" + + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/store" +) + +// iotDeviceCreateRequest — тело POST. +type iotDeviceCreateRequest struct { + Name string `json:"name"` + DeviceID string `json:"device_id"` + Enabled *bool `json:"enabled"` + Metadata map[string]string `json:"metadata,omitempty"` +} + +// iotDeviceUpdateRequest — тело PATCH. +type iotDeviceUpdateRequest struct { + Enabled *bool `json:"enabled"` +} + +// iotDeviceResponse — ответ API. MQTTPassword — только в GET по имени. +type iotDeviceResponse struct { + Name string `json:"name"` + Namespace string `json:"namespace"` + DeviceID string `json:"device_id"` + Enabled bool `json:"enabled"` + Phase string `json:"phase"` + MQTTUsername string `json:"mqtt_username,omitempty"` + MQTTPassword string `json:"mqtt_password,omitempty"` + TopicPrefix string `json:"topic_prefix,omitempty"` + LastConnected string `json:"last_connected,omitempty"` + Metadata map[string]string `json:"metadata,omitempty"` + CreatedAt string `json:"created_at,omitempty"` +} + +// deviceToResponse конвертирует устройство из БД в ответ API. +// password отдаётся только если непустая (передаётся явно из Get). +func deviceToResponse(d *store.Device, password string) iotDeviceResponse { + resp := iotDeviceResponse{ + Name: d.Name, + Namespace: d.Namespace, + DeviceID: d.DeviceID, + Enabled: d.Enabled, + Phase: d.Phase, + MQTTUsername: d.Namespace + "_" + d.DeviceID, + MQTTPassword: password, + TopicPrefix: d.Namespace + "/", + Metadata: d.Metadata, + } + if d.LastConnected != nil && !d.LastConnected.IsZero() { + resp.LastConnected = d.LastConnected.UTC().Format(time.RFC3339) + } + if !d.CreatedAt.IsZero() { + resp.CreatedAt = d.CreatedAt.UTC().Format("2006-01-02 15:04:05 UTC") + } + return resp +} + +// generateMQTTPassword — 32 случайных байта в hex (как старый контроллер). +func generateMQTTPassword() (string, error) { + buf := make([]byte, 32) + if _, err := rand.Read(buf); err != nil { + return "", err + } + return hex.EncodeToString(buf), nil +} + +// CreateIoTDevice — POST /v1/namespaces/{ns}/iot/devices. +func (h *Handler) CreateIoTDevice(w http.ResponseWriter, r *http.Request) { + ns := pathVar(r, "namespace") + if ns == "" { + ns = "default" + } + + var req iotDeviceCreateRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, errResp("invalid JSON: "+err.Error())) + return + } + if req.Name == "" { + writeJSON(w, http.StatusBadRequest, errResp("name is required")) + return + } + if req.DeviceID == "" { + writeJSON(w, http.StatusBadRequest, errResp("device_id is required")) + return + } + + enabled := true + if req.Enabled != nil { + enabled = *req.Enabled + } + + password, err := generateMQTTPassword() + if err != nil { + h.Log.Error("generate device password", "err", err) + writeJSON(w, http.StatusInternalServerError, errResp("failed to generate credentials")) + return + } + + phase := "Active" + if !enabled { + phase = "Disabled" + } + + device := &store.Device{ + Namespace: ns, + Name: req.Name, + DeviceID: req.DeviceID, + Enabled: enabled, + MQTTPassword: password, + Metadata: req.Metadata, + Phase: phase, + } + + if err := h.Devices.Create(r.Context(), device); err != nil { + if errors.Is(err, store.ErrAlreadyExists) { + writeJSON(w, http.StatusConflict, errResp("iot device already exists")) + return + } + h.Log.Error("create device", "namespace", ns, "name", req.Name, "err", err) + writeJSON(w, http.StatusInternalServerError, errResp("failed to create device")) + return + } + + writeJSON(w, http.StatusCreated, deviceToResponse(device, "")) +} + +// ListIoTDevices — GET /v1/namespaces/{ns}/iot/devices (без паролей). +func (h *Handler) ListIoTDevices(w http.ResponseWriter, r *http.Request) { + ns := pathVar(r, "namespace") + if ns == "" { + ns = "default" + } + + devices, err := h.Devices.List(r.Context(), ns) + if err != nil { + h.Log.Error("list devices", "namespace", ns, "err", err) + writeJSON(w, http.StatusInternalServerError, errResp("failed to list devices")) + return + } + + result := make([]iotDeviceResponse, 0, len(devices)) + for i := range devices { + result = append(result, deviceToResponse(&devices[i], "")) + } + writeJSON(w, http.StatusOK, result) +} + +// GetIoTDevice — GET /v1/namespaces/{ns}/iot/devices/{name} (с паролем). +func (h *Handler) GetIoTDevice(w http.ResponseWriter, r *http.Request) { + ns := pathVar(r, "namespace") + if ns == "" { + ns = "default" + } + name := pathVar(r, "name") + + device, err := h.Devices.GetByName(r.Context(), ns, name) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + writeJSON(w, http.StatusNotFound, errResp("iot device not found")) + return + } + h.Log.Error("get device", "namespace", ns, "name", name, "err", err) + writeJSON(w, http.StatusInternalServerError, errResp("failed to get device")) + return + } + + writeJSON(w, http.StatusOK, deviceToResponse(device, device.MQTTPassword)) +} + +// DeleteIoTDevice — DELETE /v1/namespaces/{ns}/iot/devices/{name}. +func (h *Handler) DeleteIoTDevice(w http.ResponseWriter, r *http.Request) { + ns := pathVar(r, "namespace") + if ns == "" { + ns = "default" + } + name := pathVar(r, "name") + + err := h.Devices.Delete(r.Context(), ns, name) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + writeJSON(w, http.StatusNotFound, errResp("iot device not found")) + return + } + h.Log.Error("delete device", "namespace", ns, "name", name, "err", err) + writeJSON(w, http.StatusInternalServerError, errResp("failed to delete device")) + return + } + + w.WriteHeader(http.StatusNoContent) +} + +// UpdateIoTDevice — PATCH /v1/namespaces/{ns}/iot/devices/{name} (enabled). +func (h *Handler) UpdateIoTDevice(w http.ResponseWriter, r *http.Request) { + ns := pathVar(r, "namespace") + if ns == "" { + ns = "default" + } + name := pathVar(r, "name") + + var req iotDeviceUpdateRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, errResp("invalid JSON: "+err.Error())) + return + } + if req.Enabled == nil { + writeJSON(w, http.StatusBadRequest, errResp("enabled field is required")) + return + } + + err := h.Devices.UpdateEnabled(r.Context(), ns, name, *req.Enabled) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + writeJSON(w, http.StatusNotFound, errResp("iot device not found")) + return + } + h.Log.Error("update device", "namespace", ns, "name", name, "err", err) + writeJSON(w, http.StatusInternalServerError, errResp("failed to update device")) + return + } + + device, err := h.Devices.GetByName(r.Context(), ns, name) + if err != nil { + writeJSON(w, http.StatusInternalServerError, errResp("failed to reload device")) + return + } + writeJSON(w, http.StatusOK, deviceToResponse(device, "")) +} diff --git a/internal/service/api/handler/handler.go b/internal/service/api/handler/handler.go new file mode 100644 index 0000000..22e817c --- /dev/null +++ b/internal/service/api/handler/handler.go @@ -0,0 +1,49 @@ +// Package handler — HTTP-обработчики IoT REST API (новый монолит). +// +// Отличие от старого кода: устройства — в PostgreSQL (таблица iot_devices), +// без k8s-клиента и CRD. Телеметрия — через iotpg, как раньше. +package handler + +import ( + "encoding/json" + "log/slog" + "net/http" + "time" + + "github.com/aws/aws-sdk-go-v2/service/sqs" + "github.com/gorilla/mux" + + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/config" + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/store" + "gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg" +) + +// Handler — зависимости всех HTTP-обработчиков. +type Handler struct { + Devices *store.DeviceStore + IoTPG *iotpg.IoTPostgresStore + SQS *sqs.Client + Cfg *config.Config + Log *slog.Logger + + // Version — версия бинарника (ldflags), StartedAt — для админ-статистики. + Version string + StartedAt time.Time +} + +// writeJSON отправляет JSON-ответ с указанным статусом. +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} + +// errResp возвращает структуру ошибки для JSON. +func errResp(msg string) map[string]string { + return map[string]string{"error": msg} +} + +// pathVar читает переменную пути из gorilla/mux. +func pathVar(r *http.Request, key string) string { + return mux.Vars(r)[key] +} diff --git a/internal/service/api/handler/mqtt_auth.go b/internal/service/api/handler/mqtt_auth.go new file mode 100644 index 0000000..1288d11 --- /dev/null +++ b/internal/service/api/handler/mqtt_auth.go @@ -0,0 +1,167 @@ +// mqtt_auth.go — HTTP-auth и ACL для EMQX (MQTT CONNECT / pub-sub). +// +// Endpoints (без JWT, вызываются EMQX внутри платформы): +// +// POST /internal/mqtt/auth — аутентификация: всегда HTTP 200, {"result":"allow"|"deny"} +// POST /internal/mqtt/acl — авторизация pub/sub +// +// Источник устройств — таблица iot_devices (не k8s Secrets). +package handler + +import ( + "crypto/subtle" + "encoding/json" + "errors" + "net/http" + "strings" + + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/store" +) + +// mqttAuthRequest — тело запроса от EMQX 5.x (username, password, clientid, peerhost). +type mqttAuthRequest struct { + Username string `json:"username"` + Password string `json:"password"` + ClientID string `json:"clientid"` + PeerHost string `json:"peerhost"` +} + +// mqttAuthResponse — ответ для EMQX (всегда HTTP 200). +type mqttAuthResponse struct { + Result string `json:"result"` + ACL []aclRule `json:"acl,omitempty"` +} + +// aclRule — правило ACL для EMQX HTTP auth. +type aclRule struct { + Permission string `json:"permission"` + Action string `json:"action"` + Topic string `json:"topic"` +} + +// mqttAclRequest — тело запроса ACL от EMQX. +type mqttAclRequest struct { + Username string `json:"username"` + ClientID string `json:"clientid"` + Action string `json:"action"` + Topic string `json:"topic"` +} + +// MQTTAuth — POST /internal/mqtt/auth. +// +// Логика: +// 1. bridge-креды (MQTT_USERNAME/MQTT_PASSWORD из env) — проверяются до парсинга; +// 2. username "{ns}_{deviceId}" → поиск устройства в iot_devices; +// 3. constant-time сравнение пароля; проверка enabled; +// 4. best-effort last_connected; ACL по топику устройства. +func (h *Handler) MQTTAuth(w http.ResponseWriter, r *http.Request) { + var req mqttAuthRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"}) + return + } + + // Внутренний bridge-клиент: фиксированные креды, ACL выдаётся ниже. + if h.Cfg.MQTTUsername != "" && req.Username == h.Cfg.MQTTUsername { + if subtle.ConstantTimeCompare([]byte(h.Cfg.MQTTPassword), []byte(req.Password)) == 1 { + writeJSON(w, http.StatusOK, mqttAuthResponse{ + Result: "allow", + ACL: bridgeACLRules(), + }) + return + } + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"}) + return + } + + // Парсим username: "{namespace}_{deviceId}" (первый '_' — разделитель). + idx := strings.Index(req.Username, "_") + if idx <= 0 || idx == len(req.Username)-1 { + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"}) + return + } + ns := req.Username[:idx] + deviceID := req.Username[idx+1:] + + device, err := h.Devices.GetByDeviceID(r.Context(), ns, deviceID) + if err != nil { + if !errors.Is(err, store.ErrNotFound) { + h.Log.Error("mqtt auth: device lookup", "namespace", ns, "device", deviceID, "err", err) + } + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"}) + return + } + if !device.Enabled { + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"}) + return + } + if subtle.ConstantTimeCompare([]byte(device.MQTTPassword), []byte(req.Password)) != 1 { + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"}) + return + } + + // best-effort: обновляем время последнего подключения + h.Devices.TouchLastConnected(r.Context(), ns, deviceID) + + writeJSON(w, http.StatusOK, mqttAuthResponse{ + Result: "allow", + ACL: deviceACLRules(ns, deviceID), + }) +} + +// MQTTAcl — POST /internal/mqtt/acl. +// +// Bridge (clientID из env MQTT_CLIENT_ID): только subscribe. +// Устройство ({ns}_{deviceId}): pub/sub в своём топике. +func (h *Handler) MQTTAcl(w http.ResponseWriter, r *http.Request) { + var req mqttAclRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"}) + return + } + + if req.ClientID == h.Cfg.MQTTClientID { + if req.Action == "subscribe" { + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "allow"}) + return + } + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"}) + return + } + + idx := strings.Index(req.Username, "_") + if idx <= 0 || idx == len(req.Username)-1 { + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"}) + return + } + ns := req.Username[:idx] + deviceID := req.Username[idx+1:] + deviceTopic := ns + "/telemetry/" + deviceID + + switch { + case req.Action == "publish" && req.Topic == deviceTopic: + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "allow"}) + case req.Action == "subscribe" && (req.Topic == deviceTopic || req.Topic == "#"): + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "allow"}) + default: + writeJSON(w, http.StatusOK, mqttAuthResponse{Result: "deny"}) + } +} + +// bridgeACLRules — ACL для внутреннего bridge: subscribe на все telemetry-топики. +func bridgeACLRules() []aclRule { + return []aclRule{ + {Permission: "allow", Action: "subscribe", Topic: "+/telemetry/+"}, + {Permission: "deny", Action: "all", Topic: "#"}, + } +} + +// deviceACLRules — ACL для устройства: только свой топик телеметрии. +func deviceACLRules(ns, deviceID string) []aclRule { + topic := ns + "/telemetry/" + deviceID + return []aclRule{ + {Permission: "allow", Action: "publish", Topic: topic}, + {Permission: "allow", Action: "subscribe", Topic: topic}, + {Permission: "deny", Action: "all", Topic: "#"}, + } +} diff --git a/internal/service/api/handler/sqsstats.go b/internal/service/api/handler/sqsstats.go new file mode 100644 index 0000000..bd5dd32 --- /dev/null +++ b/internal/service/api/handler/sqsstats.go @@ -0,0 +1,52 @@ +// sqsstats.go — сбор статистики очереди shared-SQS для админ-страницы. +package handler + +import ( + "context" + "fmt" + "strconv" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/sqs" + sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types" +) + +// iotAdminSQSStats — информация об очереди для мониторинга. +type iotAdminSQSStats struct { + ApproximateMessages int64 `json:"approximate_messages"` + ApproximateMessagesNotVisible int64 `json:"approximate_messages_not_visible"` + Error string `json:"error,omitempty"` +} + +// collectSQSStats получает ApproximateNumberOfMessages из очереди телеметрии. +func (h *Handler) collectSQSStats(ctx context.Context) iotAdminSQSStats { + if h.SQS == nil { + return iotAdminSQSStats{Error: "SQS client not configured"} + } + + queueUrlOut, err := h.SQS.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{ + QueueName: aws.String(h.Cfg.SQSQueueName), + }) + if err != nil { + return iotAdminSQSStats{Error: fmt.Sprintf("GetQueueUrl: %v", err)} + } + + attrsOut, err := h.SQS.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{ + QueueUrl: queueUrlOut.QueueUrl, + AttributeNames: []sqstypes.QueueAttributeName{ + sqstypes.QueueAttributeNameApproximateNumberOfMessages, + sqstypes.QueueAttributeNameApproximateNumberOfMessagesNotVisible, + }, + }) + if err != nil { + return iotAdminSQSStats{Error: fmt.Sprintf("GetQueueAttributes: %v", err)} + } + + approxMsg, _ := strconv.ParseInt(attrsOut.Attributes["ApproximateNumberOfMessages"], 10, 64) + approxNotVisible, _ := strconv.ParseInt(attrsOut.Attributes["ApproximateNumberOfMessagesNotVisible"], 10, 64) + + return iotAdminSQSStats{ + ApproximateMessages: approxMsg, + ApproximateMessagesNotVisible: approxNotVisible, + } +} diff --git a/internal/service/api/handler/telemetry.go b/internal/service/api/handler/telemetry.go new file mode 100644 index 0000000..f453e5e --- /dev/null +++ b/internal/service/api/handler/telemetry.go @@ -0,0 +1,43 @@ +// telemetry.go — REST-чтение телеметрии из per-tenant PostgreSQL. +// +// Endpoint: GET /v1/namespaces/{ns}/iot/telemetry?device_id={id}&limit={n} +package handler + +import ( + "net/http" + "strconv" + + "gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg" +) + +// ListIoTTelemetry — GET /v1/namespaces/{ns}/iot/telemetry. +func (h *Handler) ListIoTTelemetry(w http.ResponseWriter, r *http.Request) { + if h.IoTPG == nil { + writeJSON(w, http.StatusServiceUnavailable, errResp("IoT telemetry storage not configured")) + return + } + + ns := pathVar(r, "namespace") + deviceID := r.URL.Query().Get("device_id") + limit := 50 + if ls := r.URL.Query().Get("limit"); ls != "" { + if n, err := strconv.Atoi(ls); err == nil && n > 0 { + limit = n + } + } + + rows, err := h.IoTPG.QueryTelemetry(r.Context(), ns, deviceID, limit) + if err != nil { + h.Log.Error("query IoT telemetry", "namespace", ns, "device", deviceID, "err", err) + writeJSON(w, http.StatusInternalServerError, errResp("failed to query telemetry")) + return + } + if rows == nil { + rows = []iotpg.TelemetryRow{} + } + + writeJSON(w, http.StatusOK, map[string]any{ + "items": rows, + "count": len(rows), + }) +} diff --git a/internal/service/api/middleware/auth.go b/internal/service/api/middleware/auth.go new file mode 100644 index 0000000..ab40894 --- /dev/null +++ b/internal/service/api/middleware/auth.go @@ -0,0 +1,101 @@ +// Package middleware — HTTP-промежуточные слои API. +package middleware + +import ( + "encoding/base64" + "log/slog" + "net/http" + "strings" +) + +// Auth — проверка Bearer-токена. +// +// authTestMode (env AUTH_TEST_MODE, дефолт false): +// - true — принимается любая строка без пробелов (для локальных тестов); +// - false — структурная проверка JWT (sub + exp), как в старом коде. +// +// В новой архитектуре подпись JWT не проверяется (как и раньше): внешний +// периметр обеспечивает платформа, полная валидация — на стороне шлюза. +func Auth(authTestMode bool, log *slog.Logger, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + header := r.Header.Get("Authorization") + if header == "" { + log.Warn("auth: missing authorization header", "remote", r.RemoteAddr, "path", r.URL.Path) + http.Error(w, `{"error":"authorization required"}`, http.StatusUnauthorized) + return + } + parts := strings.SplitN(header, " ", 2) + if len(parts) != 2 || !strings.EqualFold(parts[0], "bearer") { + log.Warn("auth: invalid authorization format", "remote", r.RemoteAddr) + http.Error(w, `{"error":"invalid authorization format, use Bearer "}`, http.StatusUnauthorized) + return + } + token := parts[1] + + if authTestMode && isPlainToken(token) { + log.Info("auth: test mode — plain token accepted", "remote", r.RemoteAddr, "path", r.URL.Path) + next.ServeHTTP(w, r) + return + } + + if err := validateJWT(token); 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 + } + next.ServeHTTP(w, r) + }) +} + +// isPlainToken — непустая строка без пробелов, не похожая на JWT (три части через точку). +func isPlainToken(token string) bool { + if token == "" || strings.ContainsAny(token, " \t\n\r") { + return false + } + return len(strings.Split(token, ".")) != 3 +} + +// jwtError — структурная ошибка JWT. +type jwtError struct{ msg string } + +func (e *jwtError) Error() string { return e.msg } + +// validateJWT проверяет структуру JWT: три части, корректный payload, sub и exp. +// Подпись НЕ проверяется — см. комментарий пакета. +func validateJWT(token string) error { + jwtParts := strings.Split(token, ".") + if len(jwtParts) != 3 { + return &jwtError{"not a JWT: expected 3 parts"} + } + payload := jwtParts[1] + switch len(payload) % 4 { + case 2: + payload += "==" + case 3: + payload += "=" + } + decoded, err := base64.URLEncoding.DecodeString(payload) + if err != nil { + decoded, err = base64.StdEncoding.DecodeString(payload) + } + if err != nil { + return &jwtError{"cannot decode JWT payload"} + } + + var claims struct { + Sub string `json:"sub"` + Exp float64 `json:"exp"` + } + if err := jsonUnmarshal(decoded, &claims); err != nil { + return &jwtError{"cannot parse JWT payload"} + } + if claims.Sub == "" { + return &jwtError{"JWT missing sub claim"} + } + if claims.Exp > 0 { + if now := float64(timeNow().Unix()); claims.Exp < now { + return &jwtError{"JWT expired"} + } + } + return nil +} diff --git a/internal/service/api/middleware/auth_helpers.go b/internal/service/api/middleware/auth_helpers.go new file mode 100644 index 0000000..13b846d --- /dev/null +++ b/internal/service/api/middleware/auth_helpers.go @@ -0,0 +1,15 @@ +package middleware + +import ( + "encoding/json" + "time" +) + +// jsonUnmarshal и timeNow вынесены для простоты тестирования. +func jsonUnmarshal(data []byte, v any) error { + return json.Unmarshal(data, v) +} + +func timeNow() time.Time { + return time.Now() +} diff --git a/internal/service/api/middleware/logging.go b/internal/service/api/middleware/logging.go new file mode 100644 index 0000000..5368d21 --- /dev/null +++ b/internal/service/api/middleware/logging.go @@ -0,0 +1,35 @@ +// Package middleware — HTTP-промежуточные слои API. +package middleware + +import ( + "log/slog" + "net/http" + "time" +) + +// responseWriter перехватывает статус ответа для логирования. +type responseWriter struct { + http.ResponseWriter + status int +} + +func (rw *responseWriter) WriteHeader(code int) { + rw.status = code + rw.ResponseWriter.WriteHeader(code) +} + +// Logging логирует каждый HTTP-запрос: метод, путь, статус, длительность. +func Logging(log *slog.Logger, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + start := time.Now() + rw := &responseWriter{ResponseWriter: w, status: http.StatusOK} + next.ServeHTTP(rw, r) + log.Info("http", + "method", r.Method, + "path", r.URL.Path, + "status", rw.status, + "duration_ms", time.Since(start).Milliseconds(), + "remote", r.RemoteAddr, + ) + }) +} diff --git a/internal/service/api/router.go b/internal/service/api/router.go new file mode 100644 index 0000000..ee14c22 --- /dev/null +++ b/internal/service/api/router.go @@ -0,0 +1,70 @@ +// Package api — HTTP-сервер монолита: маршруты, CORS, health. +package api + +import ( + "log/slog" + "net/http" + + "github.com/gorilla/mux" + + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/api/handler" + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/api/middleware" +) + +// corsMiddleware добавляет CORS-заголовки (консоль и API на одном домене платформы). +func corsMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, PATCH, DELETE, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type") + if r.Method == http.MethodOptions { + w.WriteHeader(http.StatusNoContent) + return + } + next.ServeHTTP(w, r) + }) +} + +// NewRouter собирает все маршруты сервиса. +func NewRouter(h *handler.Handler, log *slog.Logger, authTestMode bool, version string) http.Handler { + r := mux.NewRouter() + + // Health — для платформенных проверок контейнера. + r.HandleFunc("/health", healthHandler(version)).Methods(http.MethodGet) + r.HandleFunc("/healthz", healthHandler(version)).Methods(http.MethodGet) + + // UI — публичные страницы (без данных). + r.HandleFunc("/console", ServeIoTConsole).Methods(http.MethodGet) + r.HandleFunc("/iot-admin", ServeIoTAdmin).Methods(http.MethodGet) + + // Админ-статистика — свой 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) + v1.HandleFunc("/namespaces/{namespace}/iot/devices", h.CreateIoTDevice).Methods(http.MethodPost) + v1.HandleFunc("/namespaces/{namespace}/iot/devices/{name}", h.GetIoTDevice).Methods(http.MethodGet) + v1.HandleFunc("/namespaces/{namespace}/iot/devices/{name}", h.DeleteIoTDevice).Methods(http.MethodDelete) + v1.HandleFunc("/namespaces/{namespace}/iot/devices/{name}", h.UpdateIoTDevice).Methods(http.MethodPatch) + v1.HandleFunc("/namespaces/{namespace}/iot/telemetry", h.ListIoTTelemetry).Methods(http.MethodGet) + + v1.Use(func(next http.Handler) http.Handler { + return middleware.Auth(authTestMode, log, next) + }) + + return corsMiddleware(middleware.Logging(log, r)) +} + +// healthHandler отдаёт статус сервиса и версию (для платформы). +func healthHandler(version string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":"ok","version":"` + version + `"}`)) + } +} diff --git a/internal/service/api/ui/iot-admin.html b/internal/service/api/ui/iot-admin.html new file mode 100644 index 0000000..2195a13 --- /dev/null +++ b/internal/service/api/ui/iot-admin.html @@ -0,0 +1,547 @@ + + + + + + + Nubes IoT Admin + + + + + + + + + +
+ +
+

Доступ для администратора

+

Введите ADMIN_STATS_TOKEN для просмотра статистики IoT pipeline.

+ + + +
+ + + +
+ + + + diff --git a/internal/service/api/ui/iot-console.html b/internal/service/api/ui/iot-console.html new file mode 100644 index 0000000..01ffe16 --- /dev/null +++ b/internal/service/api/ui/iot-console.html @@ -0,0 +1,1330 @@ + + + + + + + Nubes IoT Console + + + + +
+ + + + + + + diff --git a/internal/service/bridge/bridge.go b/internal/service/bridge/bridge.go new file mode 100644 index 0000000..41bf84e --- /dev/null +++ b/internal/service/bridge/bridge.go @@ -0,0 +1,98 @@ +// Package bridge — MQTT-подписка на телеметрию устройств → shared-SQS. +// +// Роль старого mqtt-bridge, но внутри монолита: подключается к EMQX как +// MQTT-клиент, подписывается на "+/telemetry/+" и шлёт envelope в SQS. +package bridge + +import ( + "context" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go-v2/service/sqs" + mqtt "github.com/eclipse/paho.mqtt.golang" + + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/config" + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/sqsclient" +) + +// telemetryTopicFilter — все namespace, все устройства. +const telemetryTopicFilter = "+/telemetry/+" + +// Run — бесконечный цикл bridge: подключение → подписка → ожидание сигнала завершения. +func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *slog.Logger) error { + log.Info("bridge: starting", "broker", cfg.MQTTBrokerURL, "client_id", cfg.MQTTClientID) + + // URL очереди получаем один раз при старте (как старый consumer). + queueURL, err := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSQueueName) + if err != nil { + return err + } + log.Info("bridge: SQS queue resolved", "queue", cfg.SQSQueueName) + + client, err := connectMQTT(ctx, cfg, log) + if err != nil { + return err + } + defer client.Disconnect(250) + + handler := newMessageHandler(ctx, sqsClient, queueURL, log) + token := client.Subscribe(telemetryTopicFilter, 1, handler) + if !token.WaitTimeout(10 * time.Second) { + return errSubscribeTimeout + } + if token.Error() != nil { + return token.Error() + } + log.Info("bridge: subscribed", "filter", telemetryTopicFilter) + + <-ctx.Done() + log.Info("bridge: shutting down") + return nil +} + +// errSubscribeTimeout — не дождались подтверждения подписки. +var errSubscribeTimeout = errBridge("MQTT subscribe timeout") + +// errBridge — ошибки bridge. +type errBridge string + +func (e errBridge) Error() string { return string(e) } + +// connectMQTT устанавливает подключение к EMQX с автореконнектом. +func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger) (mqtt.Client, error) { + opts := mqtt.NewClientOptions() + opts.AddBroker(cfg.MQTTBrokerURL) + opts.SetClientID(cfg.MQTTClientID) + opts.SetUsername(cfg.MQTTUsername) + opts.SetPassword(cfg.MQTTPassword) + opts.SetAutoReconnect(true) + opts.SetConnectRetry(true) + opts.SetConnectRetryInterval(5 * time.Second) + opts.SetKeepAlive(30 * time.Second) + opts.SetCleanSession(false) + + opts.SetConnectionLostHandler(func(_ mqtt.Client, err error) { + log.Warn("bridge: MQTT connection lost, reconnecting...", "err", err) + }) + opts.SetReconnectingHandler(func(_ mqtt.Client, _ *mqtt.ClientOptions) { + log.Info("bridge: MQTT reconnecting...") + }) + opts.SetOnConnectHandler(func(_ mqtt.Client) { + log.Info("bridge: MQTT connected") + }) + + client := mqtt.NewClient(opts) + token := client.Connect() + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(30 * time.Second): + return nil, errBridge("bridge: MQTT connect timeout") + case <-token.Done(): + } + if token.Error() != nil { + return nil, token.Error() + } + return client, nil +} diff --git a/internal/service/bridge/handler.go b/internal/service/bridge/handler.go new file mode 100644 index 0000000..5af09ca --- /dev/null +++ b/internal/service/bridge/handler.go @@ -0,0 +1,71 @@ +// handler.go — обработчик MQTT-сообщений: envelope → SQS SendMessage. +package bridge + +import ( + "context" + "encoding/json" + "log/slog" + "strings" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/sqs" + mqtt "github.com/eclipse/paho.mqtt.golang" +) + +// TelemetryEnvelope — сообщение, перекладываемое в SQS (разбирается consumer). +type TelemetryEnvelope struct { + Namespace string `json:"namespace"` + DeviceID string `json:"device_id"` + Topic string `json:"topic"` + Payload json.RawMessage `json:"payload"` + ReceivedAt string `json:"received_at"` +} + +// newMessageHandler возвращает обработчик MQTT-сообщений. +func newMessageHandler(ctx context.Context, sqsClient *sqs.Client, queueURL string, log *slog.Logger) mqtt.MessageHandler { + return func(_ mqtt.Client, msg mqtt.Message) { + topic := msg.Topic() + payload := msg.Payload() + + // Топик: "{namespace}/telemetry/{deviceId}". + parts := strings.SplitN(topic, "/", 3) + if len(parts) != 3 { + log.Warn("bridge: unexpected topic format, skipping", "topic", topic) + return + } + ns := parts[0] + deviceID := parts[2] + + // Не-JSON payload оборачиваем в строку. + rawPayload := json.RawMessage(payload) + if !json.Valid(payload) { + quoted, _ := json.Marshal(string(payload)) + rawPayload = json.RawMessage(quoted) + } + + envelope := TelemetryEnvelope{ + Namespace: ns, + DeviceID: deviceID, + Topic: topic, + Payload: rawPayload, + ReceivedAt: time.Now().UTC().Format(time.RFC3339), + } + body, err := json.Marshal(envelope) + if err != nil { + log.Error("bridge: marshal envelope", "topic", topic, "err", err) + return + } + + _, err = sqsClient.SendMessage(ctx, &sqs.SendMessageInput{ + QueueUrl: aws.String(queueURL), + MessageBody: aws.String(string(body)), + }) + if err != nil { + log.Error("bridge: SQS SendMessage failed", "topic", topic, "err", err) + return + } + log.Info("bridge: forwarded telemetry to SQS", + "mqtt_topic", topic, "namespace", ns, "device", deviceID) + } +} diff --git a/internal/service/config/config.go b/internal/service/config/config.go new file mode 100644 index 0000000..51b718b --- /dev/null +++ b/internal/service/config/config.go @@ -0,0 +1,150 @@ +// Package config — конфигурация монолита iot-service из env-переменных. +// +// Принцип: несекретные значения имеют дефолты (продублированы ENV в Dockerfile), +// секреты — обязательны и задаются в deck-UI при создании контейнера. +// Env-переменные после деплоя не меняются — поэтому дефолты вшиты в образ. +package config + +import ( + "fmt" + "log/slog" + "os" + "strconv" + "strings" +) + +// Config — вся конфигурация iot-service. +type Config struct { + // HTTP API + APIPort string + + // PostgreSQL (managed, общий для устройств и телеметрии) + IOTPGDSN string + + // shared-SQS + SQSEndpoint string + SQSAccessKey string + SQSSecretKey string + SQSQueueName string + SQSRegion string + + // SQS-потребитель + SQSLongPollSeconds int + SQSVisibilityTimeout int + + // MQTT (EMQX) + MQTTBrokerURL string + MQTTUsername string + MQTTPassword string + MQTTClientID string + + // Админ-статистика + AdminStatsToken string + + // Безопасность + AuthTestMode bool + + // Логирование + LogLevel slog.Level +} + +// getEnv возвращает значение env или дефолт. +func getEnv(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +// getEnvInt возвращает целое значение env или дефолт (неположительное — дефолт). +func getEnvInt(key string, def int) int { + if v := os.Getenv(key); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + return n + } + } + return def +} + +// getEnvBool разбирает булево значение env. +func getEnvBool(key string, def bool) bool { + v := os.Getenv(key) + if v == "" { + return def + } + switch strings.ToLower(v) { + case "1", "true", "yes", "on": + return true + case "0", "false", "no", "off": + return false + default: + return def + } +} + +// parseLogLevel разбирает уровень логирования. +func parseLogLevel(v string, def slog.Level) slog.Level { + switch strings.ToLower(v) { + case "debug": + return slog.LevelDebug + case "warn", "warning": + return slog.LevelWarn + case "error": + return slog.LevelError + default: + return def + } +} + +// Load читает конфигурацию из env, подставляет дефолты +// и проверяет обязательные (секретные) переменные. +func Load() (*Config, error) { + cfg := &Config{ + APIPort: getEnv("API_PORT", DefaultAPIPort), + IOTPGDSN: os.Getenv("IOT_PG_DSN"), + SQSEndpoint: getEnv("SQS_ENDPOINT", DefaultSQSEndpoint), + SQSAccessKey: os.Getenv("SQS_ACCESS_KEY"), + SQSSecretKey: os.Getenv("SQS_SECRET_KEY"), + SQSQueueName: getEnv("SQS_QUEUE_NAME", DefaultSQSQueueName), + SQSRegion: getEnv("SQS_REGION", DefaultSQSRegion), + SQSLongPollSeconds: getEnvInt("SQS_LONG_POLL_SECONDS", 20), + SQSVisibilityTimeout: getEnvInt("SQS_VISIBILITY_TIMEOUT", 30), + MQTTUsername: os.Getenv("MQTT_USERNAME"), + MQTTPassword: os.Getenv("MQTT_PASSWORD"), + MQTTClientID: getEnv("MQTT_CLIENT_ID", DefaultMQTTClientID), + AdminStatsToken: os.Getenv("ADMIN_STATS_TOKEN"), + AuthTestMode: getEnvBool("AUTH_TEST_MODE", false), + LogLevel: parseLogLevel(os.Getenv("LOG_LEVEL"), slog.LevelInfo), + } + + // MQTT broker URL: либо полный MQTT_BROKER_URL, либо из HOST/PORT/WS_PATH + cfg.MQTTBrokerURL = os.Getenv("MQTT_BROKER_URL") + if cfg.MQTTBrokerURL == "" { + host := getEnv("MQTT_HOST", DefaultMQTTHost) + port := getEnv("MQTT_PORT", DefaultMQTTPort) + path := getEnv("MQTT_WS_PATH", DefaultMQTTWSPath) + cfg.MQTTBrokerURL = "ws://" + host + ":" + port + path + } + + var missing []string + if cfg.IOTPGDSN == "" { + missing = append(missing, "IOT_PG_DSN") + } + if cfg.SQSAccessKey == "" { + missing = append(missing, "SQS_ACCESS_KEY") + } + if cfg.SQSSecretKey == "" { + missing = append(missing, "SQS_SECRET_KEY") + } + if cfg.MQTTUsername == "" { + missing = append(missing, "MQTT_USERNAME") + } + if cfg.MQTTPassword == "" { + missing = append(missing, "MQTT_PASSWORD") + } + if len(missing) > 0 { + return nil, fmt.Errorf("required env vars not set: %s", strings.Join(missing, ", ")) + } + + return cfg, nil +} diff --git a/internal/service/config/defaults.go b/internal/service/config/defaults.go new file mode 100644 index 0000000..0271db0 --- /dev/null +++ b/internal/service/config/defaults.go @@ -0,0 +1,32 @@ +package config + +// Defaults — дефолтные значения env-переменных (не секреты). +// Продублированы в Dockerfile ENV. +const ( + // DefaultAPIPort — единственный порт контейнера: API + /health + /metrics. + DefaultAPIPort = "9090" + + // DefaultSQSEndpoint — внутренний адрес shared-sqs в том же кластере Nubes. + // Обходит внешний шлюз платформы (таймауты 31-33с, MSS 1448, ~150с). + DefaultSQSEndpoint = "http://containerk8s.f1ffb134-7d16-45bd-8bef-69f6ec8ab33c.svc.cluster.local:4100" + + // DefaultSQSQueueName — очередь телеметрии IoT. + DefaultSQSQueueName = "iot-telemetry" + + // DefaultSQSRegion — регион для self-hosted SQS (не важен, оставлен для SDK). + DefaultSQSRegion = "us-east-1" + + // DefaultMQTTHost — внутреннее имя EMQX-контейнера на платформе. + // Реальный UUID namespace EMQX задаётся в deck-UI (MQTT_HOST) при создании. + DefaultMQTTHost = "emqx" + + // DefaultMQTTPort — WebSocket-листенер EMQX (wss внутри платформы). + DefaultMQTTPort = "8083" + + // DefaultMQTTWSPath — путь MQTT-over-WebSocket на EMQX. + DefaultMQTTWSPath = "/mqtt" + + // DefaultMQTTClientID — clientID bridge-подключения к EMQX. + // Используется и в ACL-логике MQTT-auth. + DefaultMQTTClientID = "iot-bridge" +) diff --git a/internal/service/consumer/consumer.go b/internal/service/consumer/consumer.go new file mode 100644 index 0000000..d93f794 --- /dev/null +++ b/internal/service/consumer/consumer.go @@ -0,0 +1,89 @@ +// Package consumer — чтение shared-SQS → per-tenant PostgreSQL. +// +// Роль старого sqs-consumer внутри монолита: long-poll цикл, обработка +// envelope телеметрии, DeleteMessage после успешной записи (at-least-once). +package consumer + +import ( + "context" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/sqs" + + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/config" + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/sqsclient" + "gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg" +) + +// errorBackoff — пауза между попытками при ошибках SQS. +const errorBackoff = 5 * time.Second + +// Run — бесконечный long-poll цикл потребителя. +func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *iotpg.IoTPostgresStore, log *slog.Logger) error { + queueURL, err := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSQueueName) + if err != nil { + return err + } + log.Info("consumer: SQS queue resolved", "queue", cfg.SQSQueueName) + + for { + if ctx.Err() != nil { + break + } + + resp, err := sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{ + QueueUrl: aws.String(queueURL), + MaxNumberOfMessages: 10, + WaitTimeSeconds: int32(cfg.SQSLongPollSeconds), + VisibilityTimeout: int32(cfg.SQSVisibilityTimeout), + }) + if err != nil { + if ctx.Err() != nil { + break + } + log.Error("consumer: ReceiveMessage failed", "err", err) + if !sleepCtx(ctx, errorBackoff) { + break + } + continue + } + + for _, msg := range resp.Messages { + if msg.Body == nil { + continue + } + if err := processTelemetry(ctx, *msg.Body, store, log); err != nil { + log.Error("consumer: process telemetry", "err", err, "message_id", aws.ToString(msg.MessageId)) + // Сообщение вернётся в очередь после visibility timeout. + continue + } + 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.Info("consumer: shutting down") + return nil +} + +// deleteMessage удаляет обработанное сообщение из очереди. +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, + }) + return err +} + +// sleepCtx — ожидание с прерыванием по ctx. false = ctx завершён. +func sleepCtx(ctx context.Context, d time.Duration) bool { + select { + case <-ctx.Done(): + return false + case <-time.After(d): + return true + } +} diff --git a/internal/service/consumer/processor.go b/internal/service/consumer/processor.go new file mode 100644 index 0000000..f1ca974 --- /dev/null +++ b/internal/service/consumer/processor.go @@ -0,0 +1,35 @@ +// processor.go — разбор envelope телеметрии и запись в per-tenant PostgreSQL. +package consumer + +import ( + "context" + "encoding/json" + "log/slog" + + "gitea.services.ngcloud.ru/Nail/IoT/internal/service/bridge" + "gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg" +) + +// processTelemetry десериализует envelope и пишет в tenant DB. +// Битый JSON — пропускаем (не блокируем очередь). +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 + } + + // 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("consumer: telemetry saved", + "namespace", envelope.Namespace, + "device", envelope.DeviceID, + ) + return nil +} diff --git a/internal/service/sqsclient/sqsclient.go b/internal/service/sqsclient/sqsclient.go new file mode 100644 index 0000000..72ff8b8 --- /dev/null +++ b/internal/service/sqsclient/sqsclient.go @@ -0,0 +1,38 @@ +// Package sqsclient — общий фабричный клиент shared-SQS для монолита. +// +// Используют: bridge (SendMessage), consumer (ReceiveMessage/DeleteMessage), +// admin-stats (GetQueueUrl/GetQueueAttributes). +package sqsclient + +import ( + "context" + "fmt" + + "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/service/config" +) + +// New создаёт SQS-клиент с переопределённым endpoint (self-hosted shared-sqs). +func New(cfg *config.Config) *sqs.Client { + return sqs.New(sqs.Options{ + Region: cfg.SQSRegion, + Credentials: credentials.NewStaticCredentialsProvider( + cfg.SQSAccessKey, cfg.SQSSecretKey, "", + ), + BaseEndpoint: aws.String(cfg.SQSEndpoint), + }) +} + +// ResolveQueueURL получает URL очереди по имени (один вызов при старте). +func ResolveQueueURL(ctx context.Context, client *sqs.Client, queueName string) (string, error) { + out, err := client.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{ + QueueName: aws.String(queueName), + }) + if err != nil { + return "", fmt.Errorf("sqs: GetQueueUrl %q: %w", queueName, err) + } + return aws.ToString(out.QueueUrl), nil +} diff --git a/internal/service/store/device.go b/internal/service/store/device.go new file mode 100644 index 0000000..bfd4b34 --- /dev/null +++ b/internal/service/store/device.go @@ -0,0 +1,29 @@ +// Package store — хранилище устройств IoT в PostgreSQL. +// +// Таблица iot_devices заменяет CRD IoTDevice и k8s Secrets старого IoT. +// Устройства: namespace, name, device_id, enabled, mqtt_password, metadata. +package store + +import ( + "errors" + "time" +) + +// ErrNotFound — устройство не найдено. +var ErrNotFound = errors.New("device not found") + +// ErrAlreadyExists — конфликт уникальности (namespace, name) или (namespace, device_id). +var ErrAlreadyExists = errors.New("device already exists") + +// Device — устройство IoT из таблицы iot_devices. +type Device struct { + Namespace string + Name string + DeviceID string + Enabled bool + MQTTPassword string + Metadata map[string]string + Phase string + LastConnected *time.Time + CreatedAt time.Time +} diff --git a/internal/service/store/devices.go b/internal/service/store/devices.go new file mode 100644 index 0000000..f699a75 --- /dev/null +++ b/internal/service/store/devices.go @@ -0,0 +1,173 @@ +package store + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" +) + +// Create вставляет новое устройство. Конфликт (namespace, name) или +// (namespace, device_id) возвращает ErrAlreadyExists. +func (s *DeviceStore) Create(ctx context.Context, d *Device) error { + metadata, err := json.Marshal(d.Metadata) + if err != nil { + return fmt.Errorf("store: marshal metadata: %w", err) + } + _, err = s.db.ExecContext(ctx, ` +INSERT INTO iot_devices (namespace, name, device_id, enabled, mqtt_password, metadata) +VALUES ($1, $2, $3, $4, $5, $6)`, + d.Namespace, d.Name, d.DeviceID, d.Enabled, d.MQTTPassword, metadata, + ) + if err != nil { + if isUniqueViolation(err) { + return ErrAlreadyExists + } + return fmt.Errorf("store: insert device: %w", err) + } + return nil +} + +// List возвращает устройства namespace (без сортировки по паролям — пароли на месте, но API их не отдаёт). +func (s *DeviceStore) List(ctx context.Context, namespace string) ([]Device, error) { + rows, err := s.db.QueryContext(ctx, ` +SELECT namespace, name, device_id, enabled, mqtt_password, metadata, phase, + last_connected, created_at +FROM iot_devices WHERE namespace = $1 ORDER BY name`, namespace) + if err != nil { + return nil, fmt.Errorf("store: list devices: %w", err) + } + defer rows.Close() + return scanDevices(rows) +} + +// GetByName возвращает устройство по (namespace, name). +func (s *DeviceStore) GetByName(ctx context.Context, namespace, name string) (*Device, error) { + return s.get(ctx, `SELECT namespace, name, device_id, enabled, mqtt_password, metadata, phase, + last_connected, created_at +FROM iot_devices WHERE namespace = $1 AND name = $2`, namespace, name) +} + +// GetByDeviceID возвращает устройство по (namespace, device_id) — для MQTT auth. +func (s *DeviceStore) GetByDeviceID(ctx context.Context, namespace, deviceID string) (*Device, error) { + return s.get(ctx, `SELECT namespace, name, device_id, enabled, mqtt_password, metadata, phase, + last_connected, created_at +FROM iot_devices WHERE namespace = $1 AND device_id = $2`, namespace, deviceID) +} + +// get — общая выборка одного устройства. +func (s *DeviceStore) get(ctx context.Context, query string, args ...any) (*Device, error) { + d, err := scanDevice(s.db.QueryRowContext(ctx, query, args...)) + if errors.Is(err, sql.ErrNoRows) { + return nil, ErrNotFound + } + if err != nil { + return nil, fmt.Errorf("store: get device: %w", err) + } + return d, nil +} + +// UpdateEnabled включает/выключает устройство. +func (s *DeviceStore) UpdateEnabled(ctx context.Context, namespace, name string, enabled bool) error { + res, err := s.db.ExecContext(ctx, + `UPDATE iot_devices SET enabled = $3, phase = $4 + WHERE namespace = $1 AND name = $2`, + namespace, name, enabled, phaseOf(enabled)) + if err != nil { + return fmt.Errorf("store: update device: %w", err) + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil +} + +// Delete удаляет устройство. +func (s *DeviceStore) Delete(ctx context.Context, namespace, name string) error { + res, err := s.db.ExecContext(ctx, + `DELETE FROM iot_devices WHERE namespace = $1 AND name = $2`, namespace, name) + if err != nil { + return fmt.Errorf("store: delete device: %w", err) + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil +} + +// TouchLastConnected обновляет last_connected (best-effort при MQTT auth). +func (s *DeviceStore) TouchLastConnected(ctx context.Context, namespace, deviceID string) { + _, _ = s.db.ExecContext(ctx, ` +UPDATE iot_devices SET last_connected = now() +WHERE namespace = $1 AND device_id = $2`, namespace, deviceID) +} + +// Count возвращает количество устройств в namespace (для админ-статистики). +func (s *DeviceStore) Count(ctx context.Context, namespace string) (int64, error) { + var n int64 + err := s.db.QueryRowContext(ctx, + `SELECT COUNT(*) FROM iot_devices WHERE namespace = $1`, namespace).Scan(&n) + if err != nil { + return 0, fmt.Errorf("store: count devices: %w", err) + } + return n, nil +} + +// scanDevices читает все строки результата. +func scanDevices(rows *sql.Rows) ([]Device, error) { + var out []Device + for rows.Next() { + d, err := scanDevice(rows) + if err != nil { + return nil, err + } + out = append(out, *d) + } + return out, rows.Err() +} + +// scanDevice читает одну строку устройства. +type rowScanner interface { + Scan(dest ...any) error +} + +func scanDevice(r rowScanner) (*Device, error) { + var ( + d Device + metadata []byte + lastConn sql.NullTime + ) + err := r.Scan(&d.Namespace, &d.Name, &d.DeviceID, &d.Enabled, &d.MQTTPassword, + &metadata, &d.Phase, &lastConn, &d.CreatedAt) + if err != nil { + return nil, err + } + if len(metadata) > 0 { + _ = json.Unmarshal(metadata, &d.Metadata) + } + if lastConn.Valid { + t := lastConn.Time + d.LastConnected = &t + } + return &d, nil +} + +// phaseOf — статус устройства в зависимости от enabled. +func phaseOf(enabled bool) string { + if enabled { + return "Active" + } + return "Disabled" +} + +// isUniqueViolation определяет нарушение уникальности PostgreSQL. +func isUniqueViolation(err error) bool { + return err != nil && err.Error() != "" && hasSQLState(err, "23505") +} + +func hasSQLState(err error, state string) bool { + type stater interface{ SQLState() string } + var e stater + return errors.As(err, &e) && e.SQLState() == state +} diff --git a/internal/service/store/open.go b/internal/service/store/open.go new file mode 100644 index 0000000..4fd724c --- /dev/null +++ b/internal/service/store/open.go @@ -0,0 +1,62 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "time" +) + +// schemaSQL — схема таблицы устройств. +// PRIMARY KEY (namespace, name) — как в REST-маршруте /devices/{name}. +// Уникальность device_id в рамках namespace — для MQTT username. +const schemaSQL = ` +CREATE TABLE IF NOT EXISTS iot_devices ( + namespace TEXT NOT NULL, + name TEXT NOT NULL, + device_id TEXT NOT NULL, + enabled BOOLEAN NOT NULL DEFAULT true, + mqtt_password TEXT NOT NULL, + metadata JSONB, + phase TEXT NOT NULL DEFAULT 'Active', + last_connected TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (namespace, name) +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_iot_devices_ns_device + ON iot_devices (namespace, device_id); +` + +// DeviceStore — доступ к таблице iot_devices (собственный пул соединений). +type DeviceStore struct { + db *sql.DB +} + +// Open подключается к PostgreSQL по DSN и инициализирует схему. +func Open(ctx context.Context, dsn string) (*DeviceStore, error) { + db, err := sql.Open("postgres", dsn) + if err != nil { + return nil, fmt.Errorf("store: open DB: %w", err) + } + db.SetMaxOpenConns(5) + db.SetMaxIdleConns(2) + db.SetConnMaxLifetime(5 * time.Minute) + + pingCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + if err := db.PingContext(pingCtx); err != nil { + db.Close() + return nil, fmt.Errorf("store: ping DB: %w", err) + } + if _, err := db.ExecContext(ctx, schemaSQL); err != nil { + db.Close() + return nil, fmt.Errorf("store: init schema: %w", err) + } + return &DeviceStore{db: db}, nil +} + +// Close закрывает пул соединений. +func (s *DeviceStore) Close() error { + return s.db.Close() +}