hivemq setup

ingester/cmd/gebos-ingester/main.go

@@ -0,0 +1,69 @@
+// 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
+}