Lars Nolden
70 lines
2.1 KiB
Go
70 lines
2.1 KiB
Go
// Package main — Gebos MQTT → Postgres ingester.
|
||
//
|
||
// Skeleton. Target: < 200 lines.
|
||
//
|
||
// Subscribes to topics of the form `t/<tenant_id>/d/<device_id>/<metric>`,
|
||
// extracts tenant_id and device_id from the topic, writes rows into
|
||
// public.telemetry on db-host as the gebos_ingest role (BYPASSRLS).
|
||
package main
|
||
|
||
import (
|
||
"context"
|
||
"log"
|
||
"os"
|
||
"os/signal"
|
||
"syscall"
|
||
)
|
||
|
||
// Defaults match the local dev stack in nix/dev/process-compose.nix so
|
||
// `go run ./ingester` works without any env setup. Prod always sets these
|
||
// explicitly via the systemd unit's EnvironmentFile.
|
||
const (
|
||
defaultBroker = "tcp://127.0.0.1:1883"
|
||
defaultPgURL = "postgres://gebos_ingest@127.0.0.1:5432/gebos?sslmode=disable"
|
||
defaultPgPassword = "dev"
|
||
)
|
||
|
||
func envOr(key, fallback string) (string, bool) {
|
||
if v := os.Getenv(key); v != "" {
|
||
return v, false
|
||
}
|
||
return fallback, true
|
||
}
|
||
|
||
func main() {
|
||
broker, brokerDefault := envOr("GEBOS_MQTT_BROKER", defaultBroker)
|
||
pgURL, pgURLDefault := envOr("GEBOS_POSTGRES_URL", defaultPgURL)
|
||
pgPassword, pgPasswordDefault := envOr("GEBOS_POSTGRES_PASSWORD", defaultPgPassword)
|
||
_ = pgPassword
|
||
|
||
if brokerDefault || pgURLDefault || pgPasswordDefault {
|
||
log.Printf("gebos-ingester: using dev defaults for: %s%s%s — set env vars in prod",
|
||
ifStr(brokerDefault, "GEBOS_MQTT_BROKER ", ""),
|
||
ifStr(pgURLDefault, "GEBOS_POSTGRES_URL ", ""),
|
||
ifStr(pgPasswordDefault, "GEBOS_POSTGRES_PASSWORD ", ""),
|
||
)
|
||
}
|
||
|
||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||
defer stop()
|
||
|
||
// TODO:
|
||
// 1. pgxpool.New with password injected into the DSN
|
||
// 2. paho MQTT client.Connect, Subscribe("t/+/d/+/+")
|
||
// 3. on message: parse topic → (tenant_id, device_id, metric), parse
|
||
// payload (JSON for now), INSERT INTO public.telemetry
|
||
// 4. batch with a short flush interval (50–100ms) for throughput
|
||
// 5. shutdown on ctx.Done
|
||
|
||
log.Printf("gebos-ingester: broker=%s db=%s (stub)", broker, pgURL)
|
||
<-ctx.Done()
|
||
log.Println("gebos-ingester: shutting down")
|
||
}
|
||
|
||
func ifStr(cond bool, a, b string) string {
|
||
if cond {
|
||
return a
|
||
}
|
||
return b
|
||
}
|