Klaus
2dd5ed5877
check / flake-check (pull_request) Has been cancelled
- sops-nix flake input, age + ssh-to-age in devShell, .sops.yaml with per-host recipients (real pubkeys still TODO until lars bootstraps). - nix/modules/gebos-secrets.nix centralises per-host secret subsets and renders /run/secrets/gebos-env via sops.templates. Service modules now read EnvironmentFile= from services.gebos.secrets.envFile instead of the placeholder /etc/gebos/secrets.env. - Falls back to a writeText env file when nix/secrets/secrets.yaml is absent, so `nix flake check` and first-boot eval work pre-bootstrap. - nix/secrets/README.md walks through age key + sops bootstrap. - Dev defaults: ingester binary falls back to local tcp/postgres URLs, process-compose adds mosquitto and wires GEBOS_MQTT_BROKER / GEBOS_POSTGRES_URL / GEBOS_POSTGRES_PASSWORD + VITE_SUPABASE_URL / VITE_SUPABASE_ANON_KEY, frontend defaults to 127.0.0.1:8000 in DEV. Refs #23 (comment #110).
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
|
||
}
|