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 16:52:45 +05:00
parent 6ce487735f
commit 938700cf86
19 changed files with 990 additions and 2 deletions
@@ -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")
+			}
+		})
+	}
+}