diff --git a/.env.example b/.env.example index 0a341c6..66628e9 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/docker-compose.yml b/docker-compose.yml index bc42a55..bb4c826 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -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. diff --git a/migrations/clickhouse/.gitkeep b/migrations/clickhouse/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/migrations/clickhouse/0001_init.sql b/migrations/clickhouse/0001_init.sql new file mode 100644 index 0000000..2baffb6 --- /dev/null +++ b/migrations/clickhouse/0001_init.sql @@ -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); diff --git a/migrations/postgres/.gitkeep b/migrations/postgres/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/migrations/postgres/0001_init.sql b/migrations/postgres/0001_init.sql new file mode 100644 index 0000000..c364af3 --- /dev/null +++ b/migrations/postgres/0001_init.sql @@ -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; diff --git a/services/ingest-service/Dockerfile b/services/ingest-service/Dockerfile new file mode 100644 index 0000000..2172cca --- /dev/null +++ b/services/ingest-service/Dockerfile @@ -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"] diff --git a/services/ingest-service/README.md b/services/ingest-service/README.md index b2341da..7c61a1f 100644 --- a/services/ingest-service/README.md +++ b/services/ingest-service/README.md @@ -1,6 +1,6 @@ # ingest-service (Go) -Status: not implemented yet. +Status: implemented (MVP). Responsibility: - Subscribe to MQTT topics `devices/{device_id}/telemetry`. @@ -11,3 +11,24 @@ Responsibility: Not this service's job: interpreting what a sensor reading *means* (thresholds, actions) — that belongs to rule-engine-service. This service only ingests and stores. + +Note on `zone_id`: the telemetry payload carries `zone_id` directly (the +publishing device/emulator is configured with its own zone), rather than +ingest-service looking it up from PostgreSQL. This keeps ingest-service's +dependencies limited to MQTT/ClickHouse/RabbitMQ, matching the architecture +diagram in the root README. A real fleet where devices don't know their own +zone would need a device-registry lookup (cached, like rule-engine caches +rules) — that's a natural follow-up, not implemented yet. + +## Running + +```bash +cd services/ingest-service +go test ./... +go build ./cmd/ingest-service +``` + +Configuration is env-var driven (see the ingest-service section of the +repo-root `.env.example`). Via `docker compose up ingest-service` it talks to +the `mosquitto`/`clickhouse`/`rabbitmq` containers directly; connect from a +local `go run` by exporting `MQTT_HOST=localhost`, `CLICKHOUSE_HOST=localhost`, etc. diff --git a/services/ingest-service/cmd/ingest-service/main.go b/services/ingest-service/cmd/ingest-service/main.go new file mode 100644 index 0000000..ac198cb --- /dev/null +++ b/services/ingest-service/cmd/ingest-service/main.go @@ -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 +} diff --git a/services/ingest-service/go.mod b/services/ingest-service/go.mod new file mode 100644 index 0000000..15f9002 --- /dev/null +++ b/services/ingest-service/go.mod @@ -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 +) diff --git a/services/ingest-service/go.sum b/services/ingest-service/go.sum new file mode 100644 index 0000000..ad9d49d --- /dev/null +++ b/services/ingest-service/go.sum @@ -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= diff --git a/services/ingest-service/internal/batch/batcher.go b/services/ingest-service/internal/batch/batcher.go new file mode 100644 index 0000000..5668249 --- /dev/null +++ b/services/ingest-service/internal/batch/batcher.go @@ -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)) +} diff --git a/services/ingest-service/internal/batch/batcher_test.go b/services/ingest-service/internal/batch/batcher_test.go new file mode 100644 index 0000000..cd6db8d --- /dev/null +++ b/services/ingest-service/internal/batch/batcher_test.go @@ -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)) + } +} diff --git a/services/ingest-service/internal/clickhouse/store.go b/services/ingest-service/internal/clickhouse/store.go new file mode 100644 index 0000000..6d9e4bd --- /dev/null +++ b/services/ingest-service/internal/clickhouse/store.go @@ -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 +} diff --git a/services/ingest-service/internal/config/config.go b/services/ingest-service/internal/config/config.go new file mode 100644 index 0000000..1b4e913 --- /dev/null +++ b/services/ingest-service/internal/config/config.go @@ -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 +} diff --git a/services/ingest-service/internal/mqttclient/subscriber.go b/services/ingest-service/internal/mqttclient/subscriber.go new file mode 100644 index 0000000..c369d28 --- /dev/null +++ b/services/ingest-service/internal/mqttclient/subscriber.go @@ -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) +} diff --git a/services/ingest-service/internal/rabbitmq/publisher.go b/services/ingest-service/internal/rabbitmq/publisher.go new file mode 100644 index 0000000..2441c4f --- /dev/null +++ b/services/ingest-service/internal/rabbitmq/publisher.go @@ -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 +} diff --git a/services/ingest-service/internal/telemetry/reading.go b/services/ingest-service/internal/telemetry/reading.go new file mode 100644 index 0000000..e7219fb --- /dev/null +++ b/services/ingest-service/internal/telemetry/reading.go @@ -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 +} diff --git a/services/ingest-service/internal/telemetry/reading_test.go b/services/ingest-service/internal/telemetry/reading_test.go new file mode 100644 index 0000000..114db08 --- /dev/null +++ b/services/ingest-service/internal/telemetry/reading_test.go @@ -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") + } + }) + } +}