package main import ( "context" "flag" "fmt" "log" "net/http" "os" "os/signal" "path/filepath" "strconv" "syscall" "time" "gitea.dooplex.hu/admin/felhom-hub/internal/api" "gitea.dooplex.hu/admin/felhom-hub/internal/assets" "gitea.dooplex.hu/admin/felhom-hub/internal/claim" "gitea.dooplex.hu/admin/felhom-hub/internal/gitea" "gitea.dooplex.hu/admin/felhom-hub/internal/hetznerapi" "gitea.dooplex.hu/admin/felhom-hub/internal/intent" "gitea.dooplex.hu/admin/felhom-hub/internal/mailrelay" "gitea.dooplex.hu/admin/felhom-hub/internal/monitor" "gitea.dooplex.hu/admin/felhom-hub/internal/notify" "gitea.dooplex.hu/admin/felhom-hub/internal/offsite" "gitea.dooplex.hu/admin/felhom-hub/internal/offsiteheal" "gitea.dooplex.hu/admin/felhom-hub/internal/pbsdrheal" "gitea.dooplex.hu/admin/felhom-hub/internal/poke" "gitea.dooplex.hu/admin/felhom-hub/internal/store" "gitea.dooplex.hu/admin/felhom-hub/internal/tenantsync" "gitea.dooplex.hu/admin/felhom-hub/internal/web" "gitea.dooplex.hu/admin/felhom-hub/internal/wgsync" "gopkg.in/yaml.v3" ) var ( Version = "dev" BuildTime = "unknown" ) // Config is the hub configuration loaded from hub.yaml. type Config struct { Auth struct { PasswordHash string `yaml:"password_hash"` } `yaml:"auth"` API struct { ReportAPIKey string `yaml:"report_api_key"` } `yaml:"api"` Notifications struct { ResendAPIKey string `yaml:"resend_api_key"` FromEmail string `yaml:"from_email"` OperatorEmail string `yaml:"operator_email"` OperatorEnabled bool `yaml:"operator_enabled"` } `yaml:"notifications"` Retention struct { MaxDays int `yaml:"max_days"` PruneSchedule string `yaml:"prune_schedule"` } `yaml:"retention"` Alerting struct { StaleThreshold string `yaml:"stale_threshold"` // Host root-fs disk-pressure thresholds (percent). Empty/0/invalid → defaults 90/95 // (normalizeDiskThresholds). Seed-only config, like stale_threshold. HostDiskWarnPercent float64 `yaml:"host_disk_warn_percent"` HostDiskCritPercent float64 `yaml:"host_disk_crit_percent"` // Per-storage-target fill thresholds (percent). Empty/0/invalid → defaults 90/95. Independent of // the host-root thresholds above so a single storage's policy can be tuned separately. StorageFillWarnPercent float64 `yaml:"storage_fill_warn_percent"` StorageFillCritPercent float64 `yaml:"storage_fill_crit_percent"` // Offsite POOL-BOX aggregate thresholds (v0.64.0, R-5). Fill = box used vs capacity (percent); // oversub = Σ(shared soft quotas)/capacity (ratio). Empty/0/invalid → defaults 80/90/2.0 (the // checker normalizes). Threshold VALUES are Claude's encoding of the starter suggestion — Viktor's // ruling pending; these keys are the one-line flip when ruled. OffsiteBoxFillWarnPercent float64 `yaml:"offsite_box_fill_warn_percent"` OffsiteBoxFillCritPercent float64 `yaml:"offsite_box_fill_crit_percent"` OffsiteOversubWarnRatio float64 `yaml:"offsite_oversub_warn_ratio"` // PBS-DR datastore fill thresholds (v0.65.0). Separate keys defaulting to the SAME 80/90 as the // restic pool box, so PBS can be tuned independently later without touching the restic policy. // No oversubscription concept for PBS (namespaces, not quotas) — fill only. PBSDRBoxFillWarnPercent float64 `yaml:"pbsdr_box_fill_warn_percent"` PBSDRBoxFillCritPercent float64 `yaml:"pbsdr_box_fill_crit_percent"` // Consecutive failed 15-minute box reads before the hub reports the store unreachable // (v0.106.0, R-339). 0/invalid → 3 (≈30–45 min). Applies to BOTH the restic pool box and the // PBS-DR datastore — one knob, because the two are the same operator question ("can I still // see the off-site tier") and splitting it would invite them to drift apart. BoxUnreachableWindows int `yaml:"box_unreachable_windows"` } `yaml:"alerting"` Registry struct { Image string `yaml:"image"` Username string `yaml:"username"` Token string `yaml:"token"` CheckInterval string `yaml:"check_interval"` TemplateInterval string `yaml:"template_interval"` } `yaml:"registry"` Server struct { Listen string `yaml:"listen"` DataDir string `yaml:"data_dir"` } `yaml:"server"` ControllerUpdates struct { // DefaultMinVersion is the global controller-version FLOOR fallback (Phase 2 managed updates): // the minimum controller version any customer's box auto-updates to, unless overridden // per-customer or via the operator UI (hub_settings). Empty = no global floor. Env-overridable // with DEFAULT_MIN_CONTROLLER_VERSION. DefaultMinVersion string `yaml:"default_min_version"` } `yaml:"controller_updates"` Mail MailConfig `yaml:"mail"` } // MailConfig tunes the app-email passthrough (POST /api/v1/mail). type MailConfig struct { // PerCustomerPerMinute caps a single customer's forwarded messages per minute (abuse // containment — one box can't drain the shared Resend quota). Default 30. PerCustomerPerMinute int `yaml:"per_customer_per_minute"` // FromDomains is the From-header allowlist (backstop; Resend is the final backstop). // Default ["felhom.eu"]. FromDomains []string `yaml:"from_domains"` } func (m MailConfig) effectivePerMinute() int { if m.PerCustomerPerMinute <= 0 { return 30 } return m.PerCustomerPerMinute } func (m MailConfig) effectiveFromDomains() []string { if len(m.FromDomains) == 0 { return []string{"felhom.eu"} } return m.FromDomains } func main() { configPath := flag.String("config", "/etc/felhom-hub/hub.yaml", "Path to configuration file") showVersion := flag.Bool("version", false, "Show version and exit") flag.Parse() if *showVersion { fmt.Printf("felhom-hub %s (built %s)\n", Version, BuildTime) os.Exit(0) } logger := log.New(os.Stdout, "", log.LstdFlags) logger.Printf("[INFO] felhom-hub %s starting", Version) // Load config cfg := loadConfig(*configPath, logger) // Environment variable overrides (for k8s Secrets) if v := os.Getenv("REGISTRY_USERNAME"); v != "" { cfg.Registry.Username = v } if v := os.Getenv("REGISTRY_TOKEN"); v != "" { cfg.Registry.Token = v } if v := os.Getenv("DEFAULT_MIN_CONTROLLER_VERSION"); v != "" { cfg.ControllerUpdates.DefaultMinVersion = v } // Resend API key is sourced from Secret/resend-api (env RESEND_API_KEY), never from the // committed ConfigMap — see documentation/runbooks/secrets.md. The ConfigMap field stays // empty as a placeholder; this override is the live source. if v := os.Getenv("RESEND_API_KEY"); v != "" { cfg.Notifications.ResendAPIKey = v } // The operator/global bearer key (api.report_api_key) is sourced from Secret/report-api // (env REPORT_API_KEY) since v0.53.0 — the ConfigMap field is an empty placeholder (the // previously-committed literal is dead once rotated; see the publish-runbook ROTATION // notes). Same pattern as RESEND_API_KEY above. if v := os.Getenv("REPORT_API_KEY"); v != "" { cfg.API.ReportAPIKey = v } // Ensure data dir exists os.MkdirAll(cfg.Server.DataDir, 0755) // Initialize store dbPath := filepath.Join(cfg.Server.DataDir, "hub.db") dataStore, err := store.New(dbPath, logger) if err != nil { logger.Fatalf("[FATAL] Failed to initialize store: %v", err) } defer dataStore.Close() logger.Printf("[INFO] Database opened at %s", dbPath) // Phase 2 managed updates: seed the global controller-version floor fallback (config/env). A // hub_settings row set via the operator UI overrides this at runtime. dataStore.SetDefaultMinControllerVersion(cfg.ControllerUpdates.DefaultMinVersion) if cfg.ControllerUpdates.DefaultMinVersion != "" { logger.Printf("[INFO] Default controller-version floor: %s", cfg.ControllerUpdates.DefaultMinVersion) } // BUNDLE slice: seed the Day-0 artifact manifest from env (ARTIFACT_AGENT_VERSION / // ARTIFACT_AGENT_SHA256 / ARTIFACT_GOLDEN_VERSION / ARTIFACT_GOLDEN_SHA256). The operator UI is the // primary editor, but the UI is password-gated, so env-seeding is the same escape hatch the floor // uses. Seed ONLY fields the DB doesn't already have, so a UI edit sticks across restarts and a // fresh install still gets values from the deployment env. { m := dataStore.GetArtifactManifest() changed := false if m.AgentVersion == "" { if v := os.Getenv("ARTIFACT_AGENT_VERSION"); v != "" { m.AgentVersion = v changed = true } } if m.AgentSHA256 == "" { if v := os.Getenv("ARTIFACT_AGENT_SHA256"); v != "" { m.AgentSHA256 = v changed = true } } if m.GoldenVersion == "" { if v := os.Getenv("ARTIFACT_GOLDEN_VERSION"); v != "" { m.GoldenVersion = v changed = true } } if m.GoldenSHA256 == "" { if v := os.Getenv("ARTIFACT_GOLDEN_SHA256"); v != "" { m.GoldenSHA256 = v changed = true } } if changed { if err := dataStore.SetArtifactManifest(m); err != nil { logger.Printf("[WARN] Failed to seed artifact manifest from env: %v", err) } else { logger.Printf("[INFO] Artifact manifest seeded from env: agent=%s golden=%s", m.AgentVersion, m.GoldenVersion) } } } // Parse stale threshold staleThreshold, err := time.ParseDuration(cfg.Alerting.StaleThreshold) if err != nil { staleThreshold = 30 * time.Minute } // Background context for all goroutines ctx, cancel := context.WithCancel(context.Background()) defer cancel() // Initialize template fetcher for customer config generation var templateFetcher *web.TemplateFetcher if cfg.Registry.Username != "" && cfg.Registry.Token != "" { templateInterval, err := time.ParseDuration(cfg.Registry.TemplateInterval) if err != nil { templateInterval = 1 * time.Hour } templateFetcher = web.NewTemplateFetcher(cfg.Registry.Username, cfg.Registry.Token, templateInterval, logger) go templateFetcher.Run(ctx) logger.Printf("[INFO] Template fetcher started (every %s)", cfg.Registry.TemplateInterval) } // Initialize asset manager (PVC storage with image seed) assetsDir := filepath.Join(cfg.Server.DataDir, "assets") assetsMgr := assets.New(assetsDir, "/usr/share/felhom/assets-seed", logger) // Initialize handlers — pass templateFetcher as interface (nil-safe) var templateProvider api.ConfigTemplateProvider if templateFetcher != nil { templateProvider = templateFetcher } apiHandler := api.New(dataStore, cfg.API.ReportAPIKey, cfg.Notifications.ResendAPIKey, cfg.Notifications.FromEmail, templateProvider, logger) apiHandler.SetAssetManager(assetsMgr) // Initialize notification dispatcher dispatcher := notify.NewDispatcher( dataStore, cfg.Notifications.ResendAPIKey, cfg.Notifications.FromEmail, cfg.Notifications.OperatorEmail, cfg.Notifications.OperatorEnabled, logger, ) apiHandler.SetDispatcher(dispatcher) // Customer-claim password arc (v0.50.0, F-4): the code engine — the dispatcher delivers the // Hungarian emails, the store holds bcrypt(code) only. Wired into the API (Day-0 issue at // config retrieve, live-box issue + claimed ingest + ACK on report, reset endpoint) and the // web UI (status chip + resend). claimEngine := &claim.Engine{Store: dataStore, Mailer: dispatcher, Logger: logger} apiHandler.SetClaimEngine(claimEngine) // App-email passthrough (POST /api/v1/mail): a customer box's shim forwards a raw message // here and the hub re-emits it to Resend over SMTP, UNCHANGED (raw passthrough — NOT the // dispatcher's structured HTTP-API path, which drops inline CID images). The Resend key stays // hub-side. Wired only when a key is present; otherwise the endpoint returns 503. if cfg.Notifications.ResendAPIKey != "" { mailSender := mailrelay.NewResendSMTP(cfg.Notifications.ResendAPIKey) apiHandler.SetMailRelay(mailSender, cfg.Mail.PerCustomerPerMinute, cfg.Mail.FromDomains) logger.Printf("[INFO] App-email relay enabled (limit %d/min/customer, From domains %v)", cfg.Mail.effectivePerMinute(), cfg.Mail.effectiveFromDomains()) } else { logger.Printf("[INFO] App-email relay disabled (no Resend key)") } webServer := web.New(dataStore, cfg.Auth.PasswordHash, cfg.API.ReportAPIKey, Version, staleThreshold, logger) webServer.SetTemplateFetcher(templateFetcher) webServer.SetAssetManager(assetsMgr) webServer.SetClaimEngine(claimEngine) // v0.50.0 — Setup-tab claim chip + resend button webServer.SetSelfBindMailer(dispatcher) // v0.66.0 (R-27) — customer self-bind link button (sibling of claim mailer) // Day-0 artifact version dropdowns: let the operator pick a version and have the hub derive the // sha256 from Gitea (no hand-copied checksums). Reuses the registry creds; degrades to manual text // entry when they're absent. if cfg.Registry.Username != "" && cfg.Registry.Token != "" { // IN-CLUSTER, not the public ingress (v0.100.0). The hub and Gitea share this cluster, and // every one of these is a small metadata call made dozens of times per Configuration page // load. Measured on DooPlex: ~0.11 s in-cluster against 0.26–1.16 s out through the public // name, which adds a DNS lookup, the ingress hop and a TLS handshake to each one for nothing. // // Overridable so a hub running outside the cluster still works; the plain-HTTP default is // safe only because it never leaves the cluster network — do not point this at a public // host without https, the registry token rides the Authorization header. // // If the service is unreachable, artifactChoices logs a WARN and returns nil, and the form // degrades to manual text entry. That is the pre-existing behaviour and it stays. giteaAPI := os.Getenv("GITEA_API_URL") if giteaAPI == "" { giteaAPI = "http://gitea.gitea-system.svc.cluster.local:3000" } webServer.SetGiteaClient(gitea.New(giteaAPI, "admin", cfg.Registry.Username, cfg.Registry.Token)) logger.Printf("[INFO] Gitea artifact browser enabled (Day-0 version dropdowns) via %s", giteaAPI) } // Offsite provisioning (SLICE 1): enabled when a Hetzner storage-box token is provided out-of-band. // PREREQUISITE: this MUST be a token scoped to a DEDICATED Hetzner project (the shared project token can // delete ep0 — SPIKE §6). Base is api.hetzner.com (NOT api.hetzner.cloud). Absent token → offsite UI // still renders, but saving with offsite enabled returns "not configured". var offsiteBoxChecker *monitor.OffsiteBoxChecker var pbsdrBoxChecker *monitor.PBSDRBoxChecker // R-5 v0.65.0: PBS-DR datastore fill (constructed with the tenantsync client below) // R-71c: the delivery checker's self-heal invokes the SAME Re-issue path the operator button // uses (webServer satisfies monitor.OffsiteReissuer). Wired ONLY when the provisioner exists — // a heal that cannot actually restage must never run (it would emit a restaged event for a // silent no-op: ReissueOffsiteForCustomer returns nil when offsite is unconfigured). var offsiteHealReissuer monitor.OffsiteReissuer if tok := os.Getenv("HETZNER_TOKEN"); tok != "" { poolBoxID, _ := strconv.ParseInt(os.Getenv("HETZNER_POOL_BOX_ID"), 10, 64) location := os.Getenv("HETZNER_LOCATION") if location == "" { location = "fsn1" } client := hetznerapi.NewClient(func() string { return os.Getenv("HETZNER_TOKEN") }) webServer.SetOffsiteProvisioner(&offsite.Provisioner{ API: client, Store: dataStore, Scanner: offsite.SSHHostKeyScanner{}, PoolBoxID: poolBoxID, Location: location, Logger: logger, }) logger.Printf("[INFO] Offsite provisioning enabled (pool_box=%d, location=%s)", poolBoxID, location) offsiteHealReissuer = webServer // R-71c heal armed (provisioner present) // R-5 (v0.64.0): the pool-box aggregate checker shares the SAME client + pool box id (GET-only). // It needs a valid box id to poll; without one, the aggregate stays unconfigured. if poolBoxID != 0 { offsiteBoxChecker = monitor.NewOffsiteBoxChecker(client, poolBoxID, dataStore, cfg.Alerting.OffsiteBoxFillWarnPercent, cfg.Alerting.OffsiteBoxFillCritPercent, cfg.Alerting.OffsiteOversubWarnRatio, cfg.Alerting.BoxUnreachableWindows, dispatcher.ProcessEvent, logger) webServer.SetOffsiteBox(offsiteBoxChecker.Snapshot) } else { logger.Printf("[INFO] Offsite pool-box aggregate disabled (HETZNER_POOL_BOX_ID unset)") } } // v0.57.0 (F3) — clean-slate re-enrollment auto re-issues offsite credentials to the fresh box. // The API host-enroll path calls this seam; the web server owns the offsite provisioner + config // bump (and, via ReissueCredentials, the escrow invalidation + events). No-op when offsite is // unconfigured or the customer has no offsite tier. apiHandler.SetOffsiteReissuer(webServer.ReissueOffsiteForCustomer) // Direction-2 immediate-sync (v0.58.0): one in-memory operator-intent notifier, shared by the // web handlers (which Bump it after every intent write) and the API handler (which long-polls it // at GET /api/v1/wait). In-memory BY DESIGN — a restart resets generations to zero; the box // compares with != , so a restart costs exactly one harmless full-state report. Closed before // server.Shutdown (below) so held waits complete instantly instead of eating the grace window. intentHub := intent.New() apiHandler.SetIntentHub(intentHub) webServer.SetIntentHub(intentHub) // Build HTTP mux mux := http.NewServeMux() // Health check endpoint — bypasses auth (for k8s probes) mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { if err := dataStore.Ping(); err != nil { w.WriteHeader(http.StatusServiceUnavailable) w.Write([]byte("db unhealthy")) return } w.WriteHeader(http.StatusOK) w.Write([]byte("ok")) }) // API routes (no dashboard auth for report ingest) mux.Handle("/api/v1/", apiHandler) // Web routes (auth required) if cfg.Auth.PasswordHash != "" { mux.Handle("/", webServer.RequireAuth(http.HandlerFunc(webServer.ServeHTTP))) } else { mux.Handle("/", http.HandlerFunc(webServer.ServeHTTP)) } // Start HTTP server server := &http.Server{ Addr: cfg.Server.Listen, Handler: mux, ReadTimeout: 30 * time.Second, WriteTimeout: 60 * time.Second, IdleTimeout: 120 * time.Second, } // Initialize version checker for controller image registry var versionChecker *web.VersionChecker if cfg.Registry.Username != "" && cfg.Registry.Token != "" { checkInterval, err := time.ParseDuration(cfg.Registry.CheckInterval) if err != nil { checkInterval = 6 * time.Hour } versionChecker = web.NewVersionChecker(cfg.Registry.Image, cfg.Registry.Username, cfg.Registry.Token, checkInterval, logger) go versionChecker.Run(ctx) logger.Printf("[INFO] Registry version checker started (every %s)", cfg.Registry.CheckInterval) } else { logger.Printf("[INFO] Registry version checker disabled (no credentials configured)") } webServer.SetVersionChecker(versionChecker) // Advertise latest on the controller report ACK (Phase 2). Guard the nil case so we don't store a // non-nil interface wrapping a nil *VersionChecker. if versionChecker != nil { apiHandler.SetLatestVersionProvider(versionChecker) } // S1 offsite connectivity: the WG peer-sync reconciler (internal/wgsync). Config from env // (mirror of the Resend/registry pattern): the SSH private key from the mounted Secret file, // the (non-secret) pinned host key from plain env. Any piece missing → disabled with an INFO // log; the /admin/wg mutations still work against the DB and report sync:"disabled". { wgAddr := os.Getenv("WG_ENDPOINT_SSH_ADDR") wgUser := os.Getenv("WG_ENDPOINT_SSH_USER") if wgUser == "" { wgUser = "felhom-peersync" } wgKeyFile := os.Getenv("WG_ENDPOINT_SSH_KEY_FILE") wgHostKey := os.Getenv("WG_ENDPOINT_SSH_HOSTKEY") if wgAddr != "" && wgKeyFile != "" && wgHostKey != "" { keyPEM, err := os.ReadFile(wgKeyFile) if err != nil { logger.Printf("[ERROR] WG peer-sync disabled: read key file %s: %v", wgKeyFile, err) } else if wgClient, err := wgsync.New(wgsync.Config{ Addr: wgAddr, User: wgUser, PrivateKey: keyPEM, HostKeyLine: wgHostKey, }, logger); err != nil { logger.Printf("[ERROR] WG peer-sync disabled: %v", err) } else { wgReconciler := wgsync.NewReconciler(dataStore, wgClient, logger) go wgReconciler.Run(ctx) apiHandler.SetWGSyncer(wgReconciler) logger.Printf("[INFO] WG peer-sync enabled (endpoint %s, user %s)", wgAddr, wgUser) } } else { logger.Printf("[INFO] WG peer-sync disabled (endpoint not configured)") } // PBS DR tier (SLICE 1): the tenantsync client — same endpoint address + pinned host key // as peersync, its OWN private key (the authorized_keys line selects the forced command). tsKeyFile := os.Getenv("TENANTSYNC_SSH_KEY_FILE") if wgAddr != "" && wgHostKey != "" && tsKeyFile != "" { keyPEM, err := os.ReadFile(tsKeyFile) if os.IsNotExist(err) { // The env names the mount path unconditionally; the optional Secret may not exist yet. logger.Printf("[INFO] PBS DR tenantsync disabled (key %s not present)", tsKeyFile) } else if err != nil { logger.Printf("[ERROR] PBS DR tenantsync disabled: read key file %s: %v", tsKeyFile, err) } else if tsClient, err := tenantsync.New(tenantsync.Config{ Addr: wgAddr, User: wgUser, PrivateKey: keyPEM, HostKeyLine: wgHostKey, }, logger); err != nil { logger.Printf("[ERROR] PBS DR tenantsync disabled: %v", err) } else { webServer.SetTenantSync(tsClient) // v0.51.0 DR-tier cascade (scenario A): a host's first WG registration may be // the descriptor's last unmet precondition — auto-provision hands-free. apiHandler.SetWGRegisteredHook(webServer.PBSDRAutoProvision) logger.Printf("[INFO] PBS DR tenantsync enabled (endpoint %s, user %s; WG-registration auto-provision hook armed)", wgAddr, wgUser) // R-5 (v0.65.0): the PBS-DR datastore fill checker shares the SAME tenantsync client // (read-only usage op). Graceful vs an ep0 still on script ≤ v1.1.0 (unavailable state). pbsdrBoxChecker = monitor.NewPBSDRBoxChecker(tsClient, cfg.Alerting.PBSDRBoxFillWarnPercent, cfg.Alerting.PBSDRBoxFillCritPercent, cfg.Alerting.BoxUnreachableWindows, dispatcher.ProcessEvent, logger) webServer.SetPBSDRBox(pbsdrBoxChecker.Snapshot) } } else { logger.Printf("[INFO] PBS DR tenantsync disabled (key or endpoint not configured)") } // Agent-plane immediate-sync SENDER (Direction-2a, v0.59.0; SPIKE-immediate-sync-transport): // same endpoint + pinned host key + peersync user as wgsync/tenantsync, its OWN forced-command // key (POKE_SSH_KEY_FILE, optional Secret). A poke is contentless + fire-and-forget; if this // is unconfigured, an agent-plane save just bumps the generation and the box picks it up on its // next ≤15-min report (no error to the operator). pokeKeyFile := os.Getenv("POKE_SSH_KEY_FILE") if wgAddr != "" && wgHostKey != "" && pokeKeyFile != "" { keyPEM, err := os.ReadFile(pokeKeyFile) if os.IsNotExist(err) { logger.Printf("[INFO] agent-plane poke disabled (key %s not present)", pokeKeyFile) } else if err != nil { logger.Printf("[ERROR] agent-plane poke disabled: read key file %s: %v", pokeKeyFile, err) } else if pokeClient, err := poke.New(poke.Config{ Addr: wgAddr, User: wgUser, PrivateKey: keyPEM, HostKeyLine: wgHostKey, }, logger); err != nil { logger.Printf("[ERROR] agent-plane poke disabled: %v", err) } else { // One notifier, both planes' system-initiated sites (v0.63.0): the web server's // pbsdr writes AND the api handler's admin desired-state / operator-peer writes. n := poke.NewNotifier(dataStore, pokeClient, logger) webServer.SetPoke(n) apiHandler.SetPoker(n) logger.Printf("[INFO] agent-plane poke enabled (endpoint %s, user %s; web + api admin seams armed)", wgAddr, wgUser) } } else { logger.Printf("[INFO] agent-plane poke disabled (key or endpoint not configured)") } } // PBS-DR self-heal reconciler (internal/pbsdrheal, from SPIKE-pbsdr-selfheal-2026-07-15). Re-arms // a consumable secret for boxes stuck in waiting_secret/consumed_failed after losing their // converged marker (re-install / restore / snapshot rollback onto a stable host_id — the case no // existing path recovers). The primary heal (re-stage the stored secret) needs no endpoint; the // escalation reuses webServer.ReissuePBSDR (a no-op-error if tenantsync is unconfigured). Reads // the hub DB only; a converged/disabled host is a pure no-op. PBSDRHEAL_ONLY_HOST scopes a // supervised first rollout to one host (empty = whole fleet — the steady state). { pbsdrReconciler := pbsdrheal.NewReconciler(dataStore, pbsdrheal.NewActions(dataStore, webServer), logger) if only := os.Getenv("PBSDRHEAL_ONLY_HOST"); only != "" { pbsdrReconciler.RestrictToHost(only) logger.Printf("[INFO] PBS-DR self-heal reconciler RESTRICTED to host %s (supervised rollout scope)", only) } go pbsdrReconciler.Run(ctx) logger.Printf("[INFO] PBS-DR self-heal reconciler started (interval 5m)") } // Off-site credential self-heal reconciler (internal/offsiteheal, R-193 / R-204 item 4). The // sibling of the block above, for the last of the four manual interventions the 2026-08-04 drill // needed: a REBUILT box has no off-site credential because its predecessor spent the one-time // password. It acts ONLY on a state the box DECLARES (an absence has four meanings and the hub // cannot tell them apart; the box can), sustained across two distinct reports, re-staging the // stored secret before ever minting. A healthy box is a pure no-op. OFFSITEHEAL_ONLY_CUSTOMER // scopes a supervised first rollout (empty = whole fleet — the steady state). { offsiteReconciler := offsiteheal.NewReconciler(dataStore, offsiteheal.NewActions(dataStore, webServer), logger) if only := os.Getenv("OFFSITEHEAL_ONLY_CUSTOMER"); only != "" { offsiteReconciler.RestrictToCustomer(only) logger.Printf("[INFO] offsite credential self-heal RESTRICTED to customer %s (supervised rollout scope)", only) } go offsiteReconciler.Run(ctx) logger.Printf("[INFO] offsite credential self-heal reconciler started (interval 5m, debounce 2 reports)") } // Session cleanup — removes expired sessions every hour go webServer.CleanupSessions(ctx) // Prune on startup, then daily at configured time (default 04:30) if cfg.Retention.MaxDays > 0 { pruneAll(dataStore, cfg.Retention.MaxDays, logger) go scheduleDaily(ctx, "prune", cfg.Retention.PruneSchedule, func() { pruneAll(dataStore, cfg.Retention.MaxDays, logger) }, logger) } // Staleness checker — runs every 60s stalenessChecker := monitor.NewStalenessChecker(dataStore, staleThreshold, dispatcher.ProcessEvent, logger) // v0.7.0: host-domain dead-man's-switch (sibling; the controller checker above is // unchanged and keeps running until the slice-10 cutover). Same 60s cadence. hostStalenessChecker := monitor.NewHostStalenessChecker(dataStore, staleThreshold, dispatcher.ProcessEvent, logger) // v0.44.0-agent: operator alert when an agent reports a degraded privileged capability (a missing // `sudo -n` grant — the 2026-06-28 cutover class). Same 60s sweep + transition/cooldown plumbing. hostCapabilityChecker := monitor.NewHostCapabilityChecker(dataStore, dispatcher.ProcessEvent, logger) // agent-v0.48.0: proactive fleet-wide agent-re-key detection — alert when a host's reported // local-API leaf fingerprint changes (independent of the controller channel-check). Same 60s sweep. hostLeafChecker := monitor.NewHostLeafChecker(dataStore, dispatcher.ProcessEvent, logger) // v0.23.0: HOST root-fs disk-pressure alert (the silent vzdump-on-root-fills class). Reads the host // root disk_percent the agent reports; born/persistent (a disk already full at hub restart alerts on // cycle 1). Distinct event types from the controller's GUEST disk_warning/disk_critical. Same 60s sweep. hostDiskChecker := monitor.NewHostDiskChecker(dataStore, cfg.Alerting.HostDiskWarnPercent, cfg.Alerting.HostDiskCritPercent, dispatcher.ProcessEvent, logger) // v0.25.0: PER-STORAGE worst-fill alert — generalizes the host-root check to any reported storage // target (dump/backup volume, data drive, lvmthin pool, PBS datastore). Born/persistent; excludes the // root-backed builtin (hostDiskChecker owns root, no double-alert); emits natural `critical`. Same sweep. storageFillChecker := monitor.NewStorageFillChecker(dataStore, cfg.Alerting.StorageFillWarnPercent, cfg.Alerting.StorageFillCritPercent, dispatcher.ProcessEvent, logger) // TASK G1: warn when a host's agent-independent watchdog auto-healed a missing /run/sshd privsep // dir — a recurring clobber that can lead to an SSH lockout (complements host_staleness). Same sweep. hostMgmtPlaneChecker := monitor.NewHostMgmtPlaneChecker(dataStore, dispatcher.ProcessEvent, logger) // TASK H1: alert when a host's OPERATOR ACCESS is degraded — felhom-sshd down (with the operator // peer configured) or its config invalid. Transition-based, same 60s sweep. hostOOBChecker := monitor.NewHostOOBChecker(dataStore, dispatcher.ProcessEvent, logger) // SLICE 4: offsite backup health from the controller report's `offsite` object — soft-quota fill // (90/95% of quota_gb) + staleness (enabled+escrowed but no run >48h — the silently-stuck detector; // run FAILURES already alert via backup_failed). Nil-safe on pre-v0.109 reports. Same sweep. offsiteChecker := monitor.NewOffsiteChecker(dataStore, 0, dispatcher.ProcessEvent, logger) // R-85: the restore-test result becomes a signal instead of a log line. Two distinct events // (failure vs staleness), operator-tier only — neither has a customerMessages entry. restoreTestChecker := monitor.NewRestoreTestChecker(dataStore, dispatcher.ProcessEvent, logger) // R-70 + R-71c: the delivery-state checker — surfaces the burned-credential shape as // offsite_delivery_stuck (warning, 24h/customer) and self-heals it via the Re-issue path // (offsite_credential_restaged, one restage/customer/24h, R-39(a)-guarded). Cooldowns are // durable (events table), so a hub restart neither floods nor silently re-heals. // R-523: the agent's in-guest controller supervisor → controller_restarted_by_agent / controller_crashloop. controllerSupervisorChecker := monitor.NewControllerSupervisorChecker(dataStore, dispatcher.ProcessEvent, logger) offsiteDeliveryChecker := monitor.NewOffsiteDeliveryChecker(dataStore, offsiteHealReissuer, dispatcher.ProcessEvent, logger) go func() { ticker := time.NewTicker(60 * time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: stalenessChecker.Check() hostStalenessChecker.Check() hostCapabilityChecker.Check() hostLeafChecker.Check() hostDiskChecker.Check() storageFillChecker.Check() hostMgmtPlaneChecker.Check() hostOOBChecker.Check() offsiteChecker.Check() restoreTestChecker.Check() // R-85: restore-test failure + per-tier staleness offsiteDeliveryChecker.Check() controllerSupervisorChecker.Check() if offsiteBoxChecker != nil { offsiteBoxChecker.Check() // R-5: restic pool-box aggregate (fetch-throttled internally) } if pbsdrBoxChecker != nil { pbsdrBoxChecker.Check() // R-5 v0.65.0: PBS-DR datastore fill (ep0 usage op, throttled) } // v0.46.0: pulled log bundles are transient diagnostics — 72 h TTL. if n, perr := dataStore.PurgeExpiredLogBundles(time.Now()); perr != nil { logger.Printf("[WARN] log-bundle TTL purge failed: %v", perr) } else if n > 0 { logger.Printf("[INFO] log-bundle TTL purge: %d expired bundle(s) dropped", n) } } } }() // Backup deadline checker — runs daily at 05:00 Budapest go scheduleDaily(ctx, "deadline-check", "05:00", func() { monitor.CheckBackupDeadlines(dataStore, stalenessChecker, dispatcher.ProcessEvent, logger) }, logger) // Signal handling sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) go func() { sig := <-sigCh logger.Printf("[INFO] Received signal %v, shutting down...", sig) cancel() // Complete every held long-poll before Shutdown so they don't consume the grace window. intentHub.Close() shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 15*time.Second) defer shutdownCancel() if err := server.Shutdown(shutdownCtx); err != nil { logger.Printf("[ERROR] HTTP server shutdown error: %v", err) } }() logger.Printf("[INFO] Listening on %s", cfg.Server.Listen) if err := server.ListenAndServe(); err != http.ErrServerClosed { logger.Fatalf("[FATAL] HTTP server error: %v", err) } logger.Println("[INFO] felhom-hub stopped") } func loadConfig(path string, logger *log.Logger) *Config { cfg := &Config{} // Defaults cfg.Server.Listen = ":8080" cfg.Server.DataDir = "/data" cfg.Retention.MaxDays = 90 cfg.Retention.PruneSchedule = "04:30" cfg.Alerting.StaleThreshold = "30m" data, err := os.ReadFile(path) if err != nil { logger.Printf("[WARN] Config file not found at %s, using defaults", path) return cfg } if err := yaml.Unmarshal(data, cfg); err != nil { logger.Printf("[WARN] Failed to parse config: %v, using defaults", err) return cfg } // Apply defaults for zero values if cfg.Server.Listen == "" { cfg.Server.Listen = ":8080" } if cfg.Server.DataDir == "" { cfg.Server.DataDir = "/data" } if cfg.Retention.MaxDays == 0 { cfg.Retention.MaxDays = 90 } if cfg.Alerting.StaleThreshold == "" { cfg.Alerting.StaleThreshold = "30m" } if cfg.Notifications.FromEmail == "" { cfg.Notifications.FromEmail = "monitoring@felhom.eu" } if cfg.Registry.Image == "" { cfg.Registry.Image = "gitea.dooplex.hu/admin/felhom-controller" } if cfg.Registry.CheckInterval == "" { cfg.Registry.CheckInterval = "6h" } if cfg.Registry.TemplateInterval == "" { cfg.Registry.TemplateInterval = "1h" } return cfg } // scheduleDaily runs fn once daily at the given "HH:MM" time in Europe/Budapest. // It blocks until ctx is cancelled. func scheduleDaily(ctx context.Context, name, timeStr string, fn func(), logger *log.Logger) { budapest, err := time.LoadLocation("Europe/Budapest") if err != nil { budapest = time.FixedZone("CET", 3600) } hour, min := parseHM(timeStr) for { now := time.Now().In(budapest) next := time.Date(now.Year(), now.Month(), now.Day(), hour, min, 0, 0, budapest) if !next.After(now) { next = next.Add(24 * time.Hour) } delay := time.Until(next) logger.Printf("[INFO] %s: next run at %s (in %s)", name, next.Format("2006-01-02 15:04 MST"), delay.Round(time.Second)) select { case <-ctx.Done(): return case <-time.After(delay): fn() } } } // parseHM parses "HH:MM" into hour and minute. Returns 0, 0 on invalid input. func parseHM(s string) (int, int) { var h, m int if _, err := fmt.Sscanf(s, "%d:%d", &h, &m); err != nil { return 0, 0 } return h, m } func pruneAll(s *store.Store, maxDays int, logger *log.Logger) { if deleted, err := s.Prune(maxDays); err != nil { logger.Printf("[WARN] Prune reports failed: %v", err) } else if deleted > 0 { logger.Printf("[INFO] Pruned %d old report rows", deleted) } if deleted, err := s.PruneEvents(maxDays); err != nil { logger.Printf("[WARN] Prune events failed: %v", err) } else if deleted > 0 { logger.Printf("[INFO] Pruned %d old event rows", deleted) } if n, err := s.PruneAppTelemetry(time.Now().Add(-90 * 24 * time.Hour)); err != nil { logger.Printf("[ERROR] Prune app telemetry: %v", err) } else if n > 0 { logger.Printf("[INFO] Pruned %d old app telemetry rows", n) } if n, err := s.PruneStaleIssues(time.Now().Add(-30 * 24 * time.Hour)); err != nil { logger.Printf("[ERROR] Prune stale issues: %v", err) } else if n > 0 { logger.Printf("[INFO] Pruned %d stale app issues", n) } }