Слияние main: перевод README на русский

Implement ingest-service: MQTT -> ClickHouse (batched) + RabbitMQ event
Subscribes to devices/+/telemetry, validates payloads, buffers writes into ClickHouse (flush on size or interval), and emits a per-reading event on RabbitMQ for rule-engine-service to consume later. Adds the initial Postgres and ClickHouse schema migrations, a Dockerfile, and wires the service into docker-compose. Verified end-to-end via docker compose + mosquitto_pub: row lands in ClickHouse and the event reaches the telemetry.new_reading queue. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-26 17:11:35 +05:00 · 2026-07-26 16:52:45 +05:00
19 changed files with 991 additions and 2 deletions
@@ -32,6 +32,13 @@ MQTT_PORT=1883
 # --- device-control-service (gRPC) ---
 DEVICE_CONTROL_GRPC_PORT=50051

+# --- ingest-service ---
+INGEST_MQTT_CLIENT_ID=ingest-service
+INGEST_MQTT_TOPIC=devices/+/telemetry
+INGEST_BATCH_MAX_SIZE=500
+INGEST_BATCH_FLUSH_INTERVAL=5s
+INGEST_BATCH_FLUSH_TIMEOUT=10s
+
 # --- Laravel app (stage 2) ---
 APP_KEY=
 APP_URL=http://localhost:8000
@@ -55,6 +55,11 @@ services:
      - ./migrations/clickhouse:/docker-entrypoint-initdb.d:ro
    networks:
      - home-automation
+    healthcheck:
+      test: ["CMD", "clickhouse-client", "--query", "SELECT 1"]
+      interval: 5s
+      timeout: 5s
+      retries: 10
    restart: unless-stopped

  redis:
@@ -86,6 +91,31 @@ services:
      retries: 10
    restart: unless-stopped

-  # App services (ingest-service, device-control-service, rule-engine-service,
+  ingest-service:
+    build: ./services/ingest-service
+    environment:
+      MQTT_HOST: mosquitto
+      MQTT_PORT: 1883
+      CLICKHOUSE_HOST: clickhouse
+      CLICKHOUSE_NATIVE_PORT: 9000
+      CLICKHOUSE_DB: ${CLICKHOUSE_DB}
+      CLICKHOUSE_USER: ${CLICKHOUSE_USER}
+      CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD}
+      RABBITMQ_HOST: rabbitmq
+      RABBITMQ_PORT: 5672
+      RABBITMQ_USER: ${RABBITMQ_USER}
+      RABBITMQ_PASSWORD: ${RABBITMQ_PASSWORD}
+    depends_on:
+      mosquitto:
+        condition: service_started
+      clickhouse:
+        condition: service_healthy
+      rabbitmq:
+        condition: service_healthy
+    networks:
+      - home-automation
+    restart: unless-stopped
+
+  # Remaining app services (device-control-service, rule-engine-service,
  # health-check-service, esp32-emulator, laravel-app) are added here as they
  # get implemented — see services/*/README.md for the plan for each one.
@@ -0,0 +1,9 @@
+CREATE TABLE IF NOT EXISTS telemetry.telemetry (
+    device_id   String,
+    zone_id     String,
+    sensor_type String,
+    value       Float64,
+    recorded_at DateTime,
+    received_at DateTime
+) ENGINE = MergeTree()
+ORDER BY (device_id, recorded_at);
@@ -0,0 +1,59 @@
+-- Core platform schema: users, zones, device types, devices, automation rules.
+-- Deliberately generic — no growbox-specific tables or columns. A "growbox"
+-- is just a row in zones with device_types like light/pump/fan attached to it.
+
+CREATE TABLE users (
+    id            BIGSERIAL PRIMARY KEY,
+    email         TEXT NOT NULL UNIQUE,
+    password_hash TEXT NOT NULL,
+    role          TEXT NOT NULL DEFAULT 'owner' CHECK (role IN ('owner', 'viewer')),
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE zones (
+    id          BIGSERIAL PRIMARY KEY,
+    user_id     BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    name        TEXT NOT NULL,
+    description TEXT,
+    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE device_types (
+    id           BIGSERIAL PRIMARY KEY,
+    code         TEXT NOT NULL UNIQUE, -- light, pump, fan, sensor_temp_humidity, vacuum, ...
+    category     TEXT NOT NULL CHECK (category IN ('actuator', 'sensor')),
+    capabilities JSONB NOT NULL DEFAULT '[]' -- e.g. ["turn_on","turn_off","set_level"]
+);
+
+CREATE TABLE devices (
+    id             BIGSERIAL PRIMARY KEY,
+    user_id        BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    zone_id        BIGINT NOT NULL REFERENCES zones(id) ON DELETE CASCADE,
+    device_type_id BIGINT NOT NULL REFERENCES device_types(id),
+    name           TEXT NOT NULL,
+    external_id    TEXT NOT NULL UNIQUE, -- physical device id (ESP32 chip id, etc.)
+    protocol       TEXT NOT NULL DEFAULT 'mqtt',
+    status         TEXT NOT NULL DEFAULT 'offline' CHECK (status IN ('online', 'offline')),
+    created_at     TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX devices_zone_id_idx ON devices(zone_id);
+CREATE INDEX devices_external_id_idx ON devices(external_id);
+
+CREATE TABLE automation_rules (
+    id                          BIGSERIAL PRIMARY KEY,
+    user_id                     BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    zone_id                     BIGINT NOT NULL REFERENCES zones(id) ON DELETE CASCADE,
+    target_device_id            BIGINT NOT NULL REFERENCES devices(id) ON DELETE CASCADE,
+    condition_source_device_id  BIGINT NOT NULL REFERENCES devices(id) ON DELETE CASCADE,
+    condition_sensor_type       TEXT NOT NULL,
+    condition_operator          TEXT NOT NULL CHECK (condition_operator IN ('>', '<', '>=', '<=', '=', '!=')),
+    condition_value             DOUBLE PRECISION NOT NULL,
+    action_type                 TEXT NOT NULL, -- e.g. turn_on, turn_off, set_level
+    action_params               JSONB NOT NULL DEFAULT '{}',
+    is_active                   BOOLEAN NOT NULL DEFAULT true,
+    created_at                  TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX automation_rules_zone_id_idx ON automation_rules(zone_id);
+CREATE INDEX automation_rules_active_idx ON automation_rules(is_active) WHERE is_active;
@@ -0,0 +1,12 @@
+FROM golang:1.25-alpine AS build
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN go mod download
+COPY . .
+RUN CGO_ENABLED=0 go build -o /out/ingest-service ./cmd/ingest-service
+
+FROM alpine:3.20
+RUN adduser -D -u 10001 app
+COPY --from=build /out/ingest-service /usr/local/bin/ingest-service
+USER app
+ENTRYPOINT ["/usr/local/bin/ingest-service"]
@@ -1,6 +1,6 @@
 # ingest-service (Go)

-Статус: пока не реализован.
+Статус: реализован (MVP).

 Зона ответственности:
 - Подписка на MQTT-топики `devices/{device_id}/telemetry`.
@@ -11,3 +11,25 @@
 Не входит в зону ответственности: интерпретация того, что *означает*
 показание датчика (пороги, действия) — это задача rule-engine-service. Этот
 сервис только принимает и сохраняет данные.
+
+Про `zone_id`: payload телеметрии несёт `zone_id` напрямую (публикующее
+устройство/эмулятор само знает свою зону), а не ingest-service ищет его в
+PostgreSQL. Это удерживает зависимости ingest-service в рамках
+MQTT/ClickHouse/RabbitMQ, как и на диаграмме архитектуры в корневом README.
+Для реального парка устройств, которые не будут знать свою зону,
+понадобится lookup через реестр устройств (с кэшированием, как rule-engine
+кэширует правила) — это естественное развитие, пока не реализовано.
+
+## Запуск
+
+```bash
+cd services/ingest-service
+go test ./...
+go build ./cmd/ingest-service
+```
+
+Конфигурация читается из переменных окружения (см. секцию ingest-service в
+корневом `.env.example`). Через `docker compose up ingest-service` сервис
+обращается напрямую к контейнерам `mosquitto`/`clickhouse`/`rabbitmq`; для
+локального запуска через `go run` выставь `MQTT_HOST=localhost`,
+`CLICKHOUSE_HOST=localhost` и т.д.
@@ -0,0 +1,118 @@
+// Command ingest-service subscribes to device telemetry over MQTT, batches
+// it into ClickHouse, and publishes a "new reading" event per message for
+// rule-engine-service to consume.
+package main
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+	"time"
+
+	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/batch"
+	chstore "git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/clickhouse"
+	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/config"
+	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/mqttclient"
+	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/rabbitmq"
+	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/telemetry"
+)
+
+// incomingBufferSize bounds how many parsed readings can wait for the worker
+// goroutine before the MQTT callback starts dropping messages. It exists so
+// a slow ClickHouse/RabbitMQ round-trip can't stall MQTT message delivery.
+const incomingBufferSize = 1000
+
+func main() {
+	logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
+
+	if err := run(logger); err != nil {
+		logger.Error("ingest-service exited with error", "error", err)
+		os.Exit(1)
+	}
+}
+
+func run(logger *slog.Logger) error {
+	cfg, err := config.Load()
+	if err != nil {
+		return err
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	store, err := chstore.Connect(ctx, chstore.Config{
+		Addr:     cfg.ClickHouseAddr,
+		Database: cfg.ClickHouseDatabase,
+		Username: cfg.ClickHouseUsername,
+		Password: cfg.ClickHousePassword,
+	})
+	cancel()
+	if err != nil {
+		return err
+	}
+	defer store.Close()
+
+	publisher, err := rabbitmq.Connect(cfg.RabbitMQURL)
+	if err != nil {
+		return err
+	}
+	defer publisher.Close()
+
+	batcher := batch.New(cfg.BatchMaxSize, cfg.BatchFlushInterval, cfg.BatchFlushTimeout, store.InsertBatch, logger)
+	batcher.Start()
+	defer batcher.Stop()
+
+	incoming := make(chan telemetry.Reading, incomingBufferSize)
+
+	var workerWG sync.WaitGroup
+	workerWG.Add(1)
+	go func() {
+		defer workerWG.Done()
+		for reading := range incoming {
+			batcher.Add(reading)
+
+			pubCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			err := publisher.PublishReading(pubCtx, reading)
+			cancel()
+			if err != nil {
+				logger.Error("publish reading event failed", "device_id", reading.DeviceID, "error", err)
+			}
+		}
+	}()
+
+	subscriber, err := mqttclient.Connect(mqttclient.Config{
+		BrokerURL: cfg.MQTTBrokerURL,
+		ClientID:  cfg.MQTTClientID,
+		Topic:     cfg.MQTTTopic,
+		QoS:       1,
+	}, func(payload []byte, receivedAt time.Time) {
+		reading, err := telemetry.ParseReading(payload, receivedAt)
+		if err != nil {
+			logger.Warn("dropping invalid telemetry payload", "error", err)
+			return
+		}
+
+		select {
+		case incoming <- reading:
+		default:
+			logger.Error("dropping reading: worker backlog full", "device_id", reading.DeviceID)
+		}
+	}, logger)
+	if err != nil {
+		return err
+	}
+
+	logger.Info("ingest-service started", "mqtt_topic", cfg.MQTTTopic)
+
+	stop := make(chan os.Signal, 1)
+	signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
+	<-stop
+
+	logger.Info("shutting down")
+	subscriber.Close()
+	close(incoming)
+	workerWG.Wait()
+
+	return nil
+}
@@ -0,0 +1,30 @@
+module git.cactoz.su/cacto/home_automatization/services/ingest-service
+
+go 1.25.0
+
+require (
+	github.com/ClickHouse/clickhouse-go/v2 v2.47.0
+	github.com/eclipse/paho.mqtt.golang v1.5.1
+	github.com/rabbitmq/amqp091-go v1.13.0
+)
+
+require (
+	github.com/ClickHouse/ch-go v0.73.0 // indirect
+	github.com/andybalholm/brotli v1.2.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/go-faster/city v1.0.1 // indirect
+	github.com/go-faster/errors v0.7.1 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/gorilla/websocket v1.5.3 // indirect
+	github.com/klauspost/compress v1.18.6 // indirect
+	github.com/paulmach/orb v0.13.0 // indirect
+	github.com/pierrec/lz4/v4 v4.1.27 // indirect
+	github.com/segmentio/asm v1.2.1 // indirect
+	github.com/shopspring/decimal v1.4.0 // indirect
+	go.opentelemetry.io/otel v1.44.0 // indirect
+	go.opentelemetry.io/otel/trace v1.44.0 // indirect
+	go.yaml.in/yaml/v3 v3.0.4 // indirect
+	golang.org/x/net v0.56.0 // indirect
+	golang.org/x/sync v0.20.0 // indirect
+	golang.org/x/sys v0.46.0 // indirect
+)
@@ -0,0 +1,58 @@
+github.com/ClickHouse/ch-go v0.73.0 h1:jsHiGRbQ3sz+gekvDFJF29LWDo5dzbJm5s1h8TWVP2M=
+github.com/ClickHouse/ch-go v0.73.0/go.mod h1:wkFIxrqlXeRJ9cn3r5Fz5Qen9jl5aTMPuGZeuJpANNY=
+github.com/ClickHouse/clickhouse-go/v2 v2.47.0 h1:ZDAzrnKSOPTIsm4tdUNfrii2yc8dk4SVRLC77BR7Z5Q=
+github.com/ClickHouse/clickhouse-go/v2 v2.47.0/go.mod h1:sPj7C7UYQ2MWHcfX+4eGN6nwnCqwUKfgO6PcwKpd6K8=
+github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro=
+github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2ISgCl2W7nE=
+github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU=
+github.com/go-faster/city v1.0.1 h1:4WAxSZ3V2Ws4QRDrscLEDcibJY8uf41H6AhXDrNDcGw=
+github.com/go-faster/city v1.0.1/go.mod h1:jKcUJId49qdW3L1qKHH/3wPeUstCVpVSXTM6vO3VcTw=
+github.com/go-faster/errors v0.7.1 h1:MkJTnDoEdi9pDabt1dpWf7AA8/BaSYZqibYyhZ20AYg=
+github.com/go-faster/errors v0.7.1/go.mod h1:5ySTjWFiphBs07IKuiL69nxdfd5+fzh1u7FPGZP2quo=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao=
+github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/paulmach/orb v0.13.0 h1:r7n7mQGGF+cj/CbcivEj9J3HGK+XR+yXnvzRdq9saIw=
+github.com/paulmach/orb v0.13.0/go.mod h1:6scRWINywA2Jf05dcjOfLfxrUIMECvTSG2MVbRLxu/k=
+github.com/pierrec/lz4/v4 v4.1.27 h1:+PhzhWDrjRj89TH2sw43nE3+4+W8lSxIuQadEHZyjUk=
+github.com/pierrec/lz4/v4 v4.1.27/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA=
+github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
+github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0=
+github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs=
+github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k=
+github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
+github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
+go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
+go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
+go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk=
+go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
+go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
+golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o=
+golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
+golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw=
+golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
@@ -0,0 +1,105 @@
+// Package batch accumulates telemetry readings and flushes them in bulk,
+// either once a size threshold is hit or on a fixed interval — so ClickHouse
+// gets bulk inserts instead of one round-trip per reading.
+package batch
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+
+	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/telemetry"
+)
+
+// FlushFunc persists one batch. It is called with a context bounded by
+// FlushTimeout, both for interval-triggered and shutdown-triggered flushes.
+type FlushFunc func(ctx context.Context, readings []telemetry.Reading) error
+
+// Batcher buffers readings in memory and flushes them via FlushFunc when
+// MaxSize is reached or Interval elapses, whichever comes first.
+type Batcher struct {
+	maxSize      int
+	interval     time.Duration
+	flushTimeout time.Duration
+	flush        FlushFunc
+	logger       *slog.Logger
+
+	mu  sync.Mutex
+	buf []telemetry.Reading
+
+	stop chan struct{}
+	done chan struct{}
+}
+
+func New(maxSize int, interval, flushTimeout time.Duration, flush FlushFunc, logger *slog.Logger) *Batcher {
+	return &Batcher{
+		maxSize:      maxSize,
+		interval:     interval,
+		flushTimeout: flushTimeout,
+		flush:        flush,
+		logger:       logger,
+	}
+}
+
+// Start begins the interval-flush loop. Call once before the first Add.
+func (b *Batcher) Start() {
+	b.stop = make(chan struct{})
+	b.done = make(chan struct{})
+	go b.loop()
+}
+
+func (b *Batcher) loop() {
+	defer close(b.done)
+	ticker := time.NewTicker(b.interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			b.flushNow("interval")
+		case <-b.stop:
+			b.flushNow("shutdown")
+			return
+		}
+	}
+}
+
+// Add appends a reading, flushing immediately if the buffer just reached
+// maxSize rather than waiting for the next tick.
+func (b *Batcher) Add(r telemetry.Reading) {
+	b.mu.Lock()
+	b.buf = append(b.buf, r)
+	full := len(b.buf) >= b.maxSize
+	b.mu.Unlock()
+
+	if full {
+		b.flushNow("size")
+	}
+}
+
+// Stop flushes any remaining buffered readings and waits for the flush loop
+// to exit. Safe to call once, after Start.
+func (b *Batcher) Stop() {
+	close(b.stop)
+	<-b.done
+}
+
+func (b *Batcher) flushNow(reason string) {
+	b.mu.Lock()
+	if len(b.buf) == 0 {
+		b.mu.Unlock()
+		return
+	}
+	readings := b.buf
+	b.buf = nil
+	b.mu.Unlock()
+
+	ctx, cancel := context.WithTimeout(context.Background(), b.flushTimeout)
+	defer cancel()
+
+	if err := b.flush(ctx, readings); err != nil {
+		b.logger.Error("batch flush failed", "reason", reason, "size", len(readings), "error", err)
+		return
+	}
+	b.logger.Debug("batch flushed", "reason", reason, "size", len(readings))
+}
@@ -0,0 +1,84 @@
+package batch
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"testing"
+	"time"
+
+	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/telemetry"
+)
+
+func discardLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+func reading(deviceID string) telemetry.Reading {
+	return telemetry.Reading{DeviceID: deviceID, ZoneID: "z1", SensorType: "temperature", Value: 1}
+}
+
+func awaitBatch(t *testing.T, flushed chan []telemetry.Reading, timeout time.Duration) []telemetry.Reading {
+	t.Helper()
+	select {
+	case batch := <-flushed:
+		return batch
+	case <-time.After(timeout):
+		t.Fatal("timed out waiting for flush")
+		return nil
+	}
+}
+
+func TestBatcher_FlushesOnSize(t *testing.T) {
+	flushed := make(chan []telemetry.Reading, 1)
+	b := New(3, time.Hour, time.Second, func(_ context.Context, r []telemetry.Reading) error {
+		flushed <- r
+		return nil
+	}, discardLogger())
+	b.Start()
+	defer b.Stop()
+
+	b.Add(reading("a"))
+	b.Add(reading("b"))
+	b.Add(reading("c")) // hits maxSize, should trigger an immediate flush
+
+	batch := awaitBatch(t, flushed, time.Second)
+	if len(batch) != 3 {
+		t.Fatalf("got %d readings, want 3", len(batch))
+	}
+}
+
+func TestBatcher_FlushesOnInterval(t *testing.T) {
+	flushed := make(chan []telemetry.Reading, 1)
+	b := New(100, 20*time.Millisecond, time.Second, func(_ context.Context, r []telemetry.Reading) error {
+		flushed <- r
+		return nil
+	}, discardLogger())
+	b.Start()
+	defer b.Stop()
+
+	b.Add(reading("a"))
+
+	batch := awaitBatch(t, flushed, time.Second)
+	if len(batch) != 1 {
+		t.Fatalf("got %d readings, want 1", len(batch))
+	}
+}
+
+func TestBatcher_StopFlushesRemainder(t *testing.T) {
+	flushed := make(chan []telemetry.Reading, 1)
+	b := New(100, time.Hour, time.Second, func(_ context.Context, r []telemetry.Reading) error {
+		flushed <- r
+		return nil
+	}, discardLogger())
+	b.Start()
+
+	b.Add(reading("a"))
+	b.Add(reading("b"))
+	b.Stop()
+
+	batch := awaitBatch(t, flushed, time.Second)
+	if len(batch) != 2 {
+		t.Fatalf("got %d readings, want 2", len(batch))
+	}
+}
@@ -0,0 +1,76 @@
+// Package clickhouse writes batches of telemetry readings to ClickHouse.
+package clickhouse
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+
+	ch "github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
+
+	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/telemetry"
+)
+
+type Config struct {
+	Addr     string
+	Database string
+	Username string
+	Password string
+	UseTLS   bool
+}
+
+type Store struct {
+	conn driver.Conn
+}
+
+func Connect(ctx context.Context, cfg Config) (*Store, error) {
+	opts := &ch.Options{
+		Addr: []string{cfg.Addr},
+		Auth: ch.Auth{
+			Database: cfg.Database,
+			Username: cfg.Username,
+			Password: cfg.Password,
+		},
+	}
+	if cfg.UseTLS {
+		opts.TLS = &tls.Config{}
+	}
+
+	conn, err := ch.Open(opts)
+	if err != nil {
+		return nil, fmt.Errorf("open clickhouse connection: %w", err)
+	}
+	if err := conn.Ping(ctx); err != nil {
+		return nil, fmt.Errorf("ping clickhouse: %w", err)
+	}
+	return &Store{conn: conn}, nil
+}
+
+func (s *Store) Close() error {
+	return s.conn.Close()
+}
+
+// InsertBatch writes readings to the telemetry table in a single INSERT.
+func (s *Store) InsertBatch(ctx context.Context, readings []telemetry.Reading) error {
+	if len(readings) == 0 {
+		return nil
+	}
+
+	batch, err := s.conn.PrepareBatch(ctx, "INSERT INTO telemetry (device_id, zone_id, sensor_type, value, recorded_at, received_at)")
+	if err != nil {
+		return fmt.Errorf("prepare batch: %w", err)
+	}
+	defer batch.Close()
+
+	for _, r := range readings {
+		if err := batch.Append(r.DeviceID, r.ZoneID, r.SensorType, r.Value, r.RecordedAt, r.ReceivedAt); err != nil {
+			return fmt.Errorf("append reading for device %s: %w", r.DeviceID, err)
+		}
+	}
+
+	if err := batch.Send(); err != nil {
+		return fmt.Errorf("send batch: %w", err)
+	}
+	return nil
+}
@@ -0,0 +1,91 @@
+// Package config loads ingest-service settings from environment variables,
+// matching the names used in the repo-root .env.example.
+package config
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"time"
+)
+
+type Config struct {
+	MQTTBrokerURL string
+	MQTTClientID  string
+	MQTTTopic     string
+
+	ClickHouseAddr     string
+	ClickHouseDatabase string
+	ClickHouseUsername string
+	ClickHousePassword string
+
+	RabbitMQURL string
+
+	BatchMaxSize       int
+	BatchFlushInterval time.Duration
+	BatchFlushTimeout  time.Duration
+}
+
+func Load() (Config, error) {
+	cfg := Config{
+		MQTTBrokerURL: fmt.Sprintf("tcp://%s:%s", getEnv("MQTT_HOST", "localhost"), getEnv("MQTT_PORT", "1883")),
+		MQTTClientID:  getEnv("INGEST_MQTT_CLIENT_ID", "ingest-service"),
+		MQTTTopic:     getEnv("INGEST_MQTT_TOPIC", "devices/+/telemetry"),
+
+		ClickHouseAddr:     fmt.Sprintf("%s:%s", getEnv("CLICKHOUSE_HOST", "localhost"), getEnv("CLICKHOUSE_NATIVE_PORT", "9000")),
+		ClickHouseDatabase: getEnv("CLICKHOUSE_DB", "telemetry"),
+		ClickHouseUsername: getEnv("CLICKHOUSE_USER", "default"),
+		ClickHousePassword: getEnv("CLICKHOUSE_PASSWORD", ""),
+
+		RabbitMQURL: fmt.Sprintf("amqp://%s:%s@%s:%s/",
+			getEnv("RABBITMQ_USER", "guest"),
+			getEnv("RABBITMQ_PASSWORD", "guest"),
+			getEnv("RABBITMQ_HOST", "localhost"),
+			getEnv("RABBITMQ_PORT", "5672"),
+		),
+	}
+
+	var err error
+	if cfg.BatchMaxSize, err = getEnvInt("INGEST_BATCH_MAX_SIZE", 500); err != nil {
+		return Config{}, err
+	}
+	if cfg.BatchFlushInterval, err = getEnvDuration("INGEST_BATCH_FLUSH_INTERVAL", 5*time.Second); err != nil {
+		return Config{}, err
+	}
+	if cfg.BatchFlushTimeout, err = getEnvDuration("INGEST_BATCH_FLUSH_TIMEOUT", 10*time.Second); err != nil {
+		return Config{}, err
+	}
+
+	return cfg, nil
+}
+
+func getEnv(key, fallback string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return fallback
+}
+
+func getEnvInt(key string, fallback int) (int, error) {
+	v := os.Getenv(key)
+	if v == "" {
+		return fallback, nil
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil {
+		return 0, fmt.Errorf("%s: %w", key, err)
+	}
+	return n, nil
+}
+
+func getEnvDuration(key string, fallback time.Duration) (time.Duration, error) {
+	v := os.Getenv(key)
+	if v == "" {
+		return fallback, nil
+	}
+	d, err := time.ParseDuration(v)
+	if err != nil {
+		return 0, fmt.Errorf("%s: %w", key, err)
+	}
+	return d, nil
+}
@@ -0,0 +1,60 @@
+// Package mqttclient subscribes to device telemetry topics over MQTT.
+package mqttclient
+
+import (
+	"fmt"
+	"log/slog"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+// Handler receives one message's raw payload and the time it arrived.
+type Handler func(payload []byte, receivedAt time.Time)
+
+type Config struct {
+	BrokerURL string // e.g. tcp://mosquitto:1883
+	ClientID  string
+	Topic     string // e.g. devices/+/telemetry
+	QoS       byte
+}
+
+type Subscriber struct {
+	client mqtt.Client
+}
+
+// Connect dials the broker and (re-)subscribes to cfg.Topic on every
+// successful connection, so a dropped connection resubscribes automatically
+// once auto-reconnect succeeds.
+func Connect(cfg Config, handler Handler, logger *slog.Logger) (*Subscriber, error) {
+	opts := mqtt.NewClientOptions().
+		AddBroker(cfg.BrokerURL).
+		SetClientID(cfg.ClientID).
+		SetAutoReconnect(true).
+		SetConnectRetry(true).
+		SetOnConnectHandler(func(c mqtt.Client) {
+			token := c.Subscribe(cfg.Topic, cfg.QoS, func(_ mqtt.Client, msg mqtt.Message) {
+				handler(msg.Payload(), time.Now())
+			})
+			token.Wait()
+			if err := token.Error(); err != nil {
+				logger.Error("mqtt subscribe failed", "topic", cfg.Topic, "error", err)
+			}
+		}).
+		SetConnectionLostHandler(func(_ mqtt.Client, err error) {
+			logger.Warn("mqtt connection lost", "error", err)
+		})
+
+	client := mqtt.NewClient(opts)
+	token := client.Connect()
+	token.Wait()
+	if err := token.Error(); err != nil {
+		return nil, fmt.Errorf("connect to mqtt broker %s: %w", cfg.BrokerURL, err)
+	}
+	return &Subscriber{client: client}, nil
+}
+
+// Close disconnects, waiting up to quiesceMS for in-flight work to settle.
+func (s *Subscriber) Close() {
+	s.client.Disconnect(250)
+}
@@ -0,0 +1,85 @@
+// Package rabbitmq publishes "new reading" events for rule-engine-service to
+// consume. Delivery reliability matters more than latency here, so messages
+// are persistent and the queue is durable.
+package rabbitmq
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+
+	"git.cactoz.su/cacto/home_automatization/services/ingest-service/internal/telemetry"
+)
+
+const NewReadingQueue = "telemetry.new_reading"
+
+type Publisher struct {
+	conn *amqp.Connection
+	ch   *amqp.Channel
+}
+
+func Connect(url string) (*Publisher, error) {
+	conn, err := amqp.Dial(url)
+	if err != nil {
+		return nil, fmt.Errorf("dial rabbitmq: %w", err)
+	}
+
+	ch, err := conn.Channel()
+	if err != nil {
+		conn.Close()
+		return nil, fmt.Errorf("open channel: %w", err)
+	}
+
+	if _, err := ch.QueueDeclare(NewReadingQueue, true, false, false, false, nil); err != nil {
+		ch.Close()
+		conn.Close()
+		return nil, fmt.Errorf("declare queue %s: %w", NewReadingQueue, err)
+	}
+
+	return &Publisher{conn: conn, ch: ch}, nil
+}
+
+func (p *Publisher) Close() error {
+	if err := p.ch.Close(); err != nil {
+		p.conn.Close()
+		return err
+	}
+	return p.conn.Close()
+}
+
+// newReadingEvent is the wire format published to rule-engine-service.
+type newReadingEvent struct {
+	DeviceID   string    `json:"device_id"`
+	ZoneID     string    `json:"zone_id"`
+	SensorType string    `json:"sensor_type"`
+	Value      float64   `json:"value"`
+	RecordedAt time.Time `json:"recorded_at"`
+}
+
+// PublishReading emits one event per reading to NewReadingQueue.
+func (p *Publisher) PublishReading(ctx context.Context, r telemetry.Reading) error {
+	body, err := json.Marshal(newReadingEvent{
+		DeviceID:   r.DeviceID,
+		ZoneID:     r.ZoneID,
+		SensorType: r.SensorType,
+		Value:      r.Value,
+		RecordedAt: r.RecordedAt,
+	})
+	if err != nil {
+		return fmt.Errorf("marshal event: %w", err)
+	}
+
+	err = p.ch.PublishWithContext(ctx, "", NewReadingQueue, false, false, amqp.Publishing{
+		ContentType:  "application/json",
+		DeliveryMode: amqp.Persistent,
+		Timestamp:    time.Now(),
+		Body:         body,
+	})
+	if err != nil {
+		return fmt.Errorf("publish reading for device %s: %w", r.DeviceID, err)
+	}
+	return nil
+}
@@ -0,0 +1,69 @@
+// Package telemetry parses and validates raw device telemetry payloads.
+package telemetry
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+// Reading is a single validated sensor reading, ready to be stored in
+// ClickHouse and published as an event for rule-engine-service.
+type Reading struct {
+	DeviceID   string
+	ZoneID     string
+	SensorType string
+	Value      float64
+	RecordedAt time.Time
+	ReceivedAt time.Time
+}
+
+// payload mirrors the wire format devices publish to devices/{device_id}/telemetry.
+//
+// ZoneID travels with the payload (rather than being looked up from
+// PostgreSQL) so ingest-service stays decoupled from the device registry,
+// matching the ingest→ClickHouse/RabbitMQ-only data flow in the platform
+// architecture — the publishing client is responsible for knowing its own zone.
+type payload struct {
+	DeviceID   string  `json:"device_id"`
+	ZoneID     string  `json:"zone_id"`
+	SensorType string  `json:"sensor_type"`
+	Value      float64 `json:"value"`
+	Timestamp  string  `json:"timestamp"`
+}
+
+// ParseReading decodes and validates a raw MQTT message body. receivedAt is
+// the time the server received the message, stamped independently of
+// whatever the device claims, so ingest/processing delay can be measured later.
+func ParseReading(raw []byte, receivedAt time.Time) (Reading, error) {
+	var p payload
+	if err := json.Unmarshal(raw, &p); err != nil {
+		return Reading{}, fmt.Errorf("invalid json: %w", err)
+	}
+
+	if p.DeviceID == "" {
+		return Reading{}, fmt.Errorf("device_id is required")
+	}
+	if p.ZoneID == "" {
+		return Reading{}, fmt.Errorf("zone_id is required")
+	}
+	if p.SensorType == "" {
+		return Reading{}, fmt.Errorf("sensor_type is required")
+	}
+	if p.Timestamp == "" {
+		return Reading{}, fmt.Errorf("timestamp is required")
+	}
+	recordedAt, err := time.Parse(time.RFC3339, p.Timestamp)
+	if err != nil {
+		return Reading{}, fmt.Errorf("invalid timestamp %q: %w", p.Timestamp, err)
+	}
+
+	return Reading{
+		DeviceID:   p.DeviceID,
+		ZoneID:     p.ZoneID,
+		SensorType: p.SensorType,
+		Value:      p.Value,
+		RecordedAt: recordedAt,
+		ReceivedAt: receivedAt,
+	}, nil
+}
@@ -0,0 +1,74 @@
+package telemetry
+
+import (
+	"testing"
+	"time"
+)
+
+func TestParseReading_Valid(t *testing.T) {
+	receivedAt := time.Date(2026, 7, 26, 12, 0, 5, 0, time.UTC)
+	raw := []byte(`{
+		"device_id": "esp32-abc123",
+		"zone_id": "growbox-1",
+		"sensor_type": "temperature",
+		"value": 24.5,
+		"timestamp": "2026-07-26T12:00:00Z"
+	}`)
+
+	got, err := ParseReading(raw, receivedAt)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	want := Reading{
+		DeviceID:   "esp32-abc123",
+		ZoneID:     "growbox-1",
+		SensorType: "temperature",
+		Value:      24.5,
+		RecordedAt: time.Date(2026, 7, 26, 12, 0, 0, 0, time.UTC),
+		ReceivedAt: receivedAt,
+	}
+	if got != want {
+		t.Fatalf("got %+v, want %+v", got, want)
+	}
+}
+
+func TestParseReading_Invalid(t *testing.T) {
+	receivedAt := time.Now()
+
+	cases := map[string]string{
+		"malformed json": `{not json`,
+		"missing device_id": `{
+			"zone_id": "growbox-1", "sensor_type": "temperature",
+			"value": 1, "timestamp": "2026-07-26T12:00:00Z"
+		}`,
+		"missing zone_id": `{
+			"device_id": "esp32-1", "sensor_type": "temperature",
+			"value": 1, "timestamp": "2026-07-26T12:00:00Z"
+		}`,
+		"missing sensor_type": `{
+			"device_id": "esp32-1", "zone_id": "growbox-1",
+			"value": 1, "timestamp": "2026-07-26T12:00:00Z"
+		}`,
+		"missing timestamp": `{
+			"device_id": "esp32-1", "zone_id": "growbox-1",
+			"sensor_type": "temperature", "value": 1
+		}`,
+		"invalid timestamp format": `{
+			"device_id": "esp32-1", "zone_id": "growbox-1",
+			"sensor_type": "temperature", "value": 1, "timestamp": "not-a-date"
+		}`,
+		"value wrong type": `{
+			"device_id": "esp32-1", "zone_id": "growbox-1",
+			"sensor_type": "temperature", "value": "not-a-number", "timestamp": "2026-07-26T12:00:00Z"
+		}`,
+	}
+
+	for name, raw := range cases {
+		t.Run(name, func(t *testing.T) {
+			if _, err := ParseReading([]byte(raw), receivedAt); err == nil {
+				t.Fatalf("expected error, got nil")
+			}
+		})
+	}
+}