cactoandClaude Sonnet 5
938700cf86
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>
86 lines
2.1 KiB
Go
86 lines
2.1 KiB
Go
// 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
|
|
}
|