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>
70 lines
2.1 KiB
Go
70 lines
2.1 KiB
Go
// 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
|
|
}
|