From eb9250911dd80cba41fb3a5ee7a7428261dc7c17 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 20 Jul 2026 12:20:26 +0200 Subject: [PATCH] feat: report driver inventory to Nova --- .changeset/driver-fleet-inventory.md | 5 + docs/nova-integration.md | 17 +- go/cmd/ftw/main.go | 58 +++- go/internal/driverinventory/inventory.go | 303 ++++++++++++++++++ go/internal/driverinventory/inventory_test.go | 143 +++++++++ go/internal/drivers/catalog.go | 15 +- go/internal/nova/publisher.go | 119 ++++++- go/internal/nova/publisher_test.go | 34 ++ 8 files changed, 681 insertions(+), 13 deletions(-) create mode 100644 .changeset/driver-fleet-inventory.md create mode 100644 go/internal/driverinventory/inventory.go create mode 100644 go/internal/driverinventory/inventory_test.go diff --git a/.changeset/driver-fleet-inventory.md b/.changeset/driver-fleet-inventory.md new file mode 100644 index 00000000..4a7775a3 --- /dev/null +++ b/.changeset/driver-fleet-inventory.md @@ -0,0 +1,5 @@ +--- +"ftw": minor +--- + +Report a bounded driver inventory to Nova when federation is enabled. Reports include loaded code hashes and package provenance, but no site config, device IDs, endpoints or credentials. diff --git a/docs/nova-integration.md b/docs/nova-integration.md index b7338733..fed50235 100644 --- a/docs/nova-integration.md +++ b/docs/nova-integration.md @@ -17,8 +17,21 @@ FTW keeps its clean snake_case site convention internally. The default vocabulary and battery sign at this one boundary. `schema_mode: unified` publishes the clean schema when the target Nova deployment supports it. -Only DER telemetry and registered device identity are published. Operator -credentials and FTW configuration are not telemetry payloads. +Only DER telemetry, registered device identity and the bounded driver inventory +are published. Operator credentials and FTW configuration are not payloads. + +## Driver inventory + +When Nova federation is enabled, FTW also publishes +`sourceful.driver-inventory/v1` on connect, after a loaded-driver change and at +least every 15 minutes. It reports driver ID, version, loaded source or package +hash, package channel, declared control class, instance counts and health. + +The report does not contain instance or site names, driver config, connection +details, device IDs, tokens, logs, command inputs or vendor responses. Nova +uses the authenticated MQTT identity for gateway and organization ownership. +Fleet reports must show how many FTW gateways sent a fresh inventory; the first +beta counts do not cover FTW sites that have not enabled Nova federation. ## Claim and provision diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go index da6d4050..38bfb2cb 100644 --- a/go/cmd/ftw/main.go +++ b/go/cmd/ftw/main.go @@ -31,6 +31,7 @@ import ( "github.com/srcfl/ftw/go/internal/control" "github.com/srcfl/ftw/go/internal/currency" "github.com/srcfl/ftw/go/internal/devtools" + "github.com/srcfl/ftw/go/internal/driverinventory" "github.com/srcfl/ftw/go/internal/driverrepo" "github.com/srcfl/ftw/go/internal/drivers" "github.com/srcfl/ftw/go/internal/events" @@ -2063,7 +2064,23 @@ func main() { siteIdentity, err := nova.LoadOrCreateIdentity(identityKeyPath) if err != nil { slog.Warn("nova federation disabled — gateway identity unavailable", "err", err, "path", identityKeyPath) - } else if pub, err := nova.Start(cfg.Nova, siteIdentity, st, tel); err != nil { + } else if pub, err := nova.Start(cfg.Nova, siteIdentity, st, tel, + nova.WithDriverInventory(func(now time.Time) (driverinventory.Snapshot, error) { + cfgMu.RLock() + driverCfg := append([]config.Driver(nil), cfg.Drivers...) + cfgMu.RUnlock() + return driverinventory.Build(now, driverinventory.Input{ + HostVersion: Version, + Drivers: driverCfg, + RunningNames: reg.Names(), + Health: tel.AllHealth(), + UserDriverDir: *userDriversDirFlag, + ManagedDriverDir: driverRepository.ActiveDir(), + BundledDriverDir: resolveDriverDir(), + RepositoryDrivers: inventoryRepositoryArtifacts(driverRepository), + }) + }), + ); err != nil { slog.Warn("nova publisher failed to start", "err", err) } else if pub != nil { defer pub.Stop() @@ -2539,6 +2556,45 @@ func main() { } } +func inventoryRepositoryArtifacts(manager *driverrepo.Manager) []driverinventory.RepositoryArtifact { + if manager == nil { + return nil + } + active := manager.Status().Active + out := make([]driverinventory.RepositoryArtifact, 0, len(active)) + for _, installed := range active { + item := driverinventory.RepositoryArtifact{ + LogicalPath: installed.LogicalPath, + InstalledPath: installed.InstalledPath, + DriverID: installed.DriverID, + Version: installed.Version, + SHA256: installed.SHA256, + RepositoryID: installed.RepoID, + } + versions, err := manager.AvailableVersions(installed.DriverID) + if err == nil { + for _, candidate := range versions { + driver := candidate.Driver + if candidate.RepositoryID != installed.RepoID || driver.Version != installed.Version || !strings.EqualFold(driver.SHA256, installed.SHA256) { + continue + } + item.PackageID = driver.PackageID + item.PackageChannel = driver.Channel + if driver.PackageID != "" { + if driver.Metadata.ReadOnly { + item.ControlClass = "read_only" + } else { + item.ControlClass = "control" + } + } + break + } + } + out = append(out, item) + } + return out +} + // snapshotLoop writes a recovery snapshot of state.db daily and once on // shutdown. The snapshot is the restore source if state.db corrupts (see // state.openChecked). cache.db needs none — it's re-fetchable. Daily (not diff --git a/go/internal/driverinventory/inventory.go b/go/internal/driverinventory/inventory.go new file mode 100644 index 00000000..e2211086 --- /dev/null +++ b/go/internal/driverinventory/inventory.go @@ -0,0 +1,303 @@ +// Package driverinventory builds a bounded, secret-free snapshot of the Lua +// code that FTW has configured and loaded. +package driverinventory + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "os" + "path/filepath" + "regexp" + "sort" + "strings" + "time" + + "github.com/srcfl/ftw/go/internal/config" + "github.com/srcfl/ftw/go/internal/drivers" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +const ( + SchemaVersion = "sourceful.driver-inventory/v1" + RuntimeABI = "gopher-lua-source-v1" + HostAPI = 1 + MaxDrivers = 128 +) + +var driverIDRE = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]{0,127}$`) + +type Host struct { + Product string `json:"product"` + Version string `json:"version"` + UpdateChannel string `json:"update_channel"` + Target string `json:"target"` + RuntimeABI string `json:"runtime_abi"` + HostAPI int `json:"host_api"` +} + +type Health struct { + OK int `json:"ok"` + Degraded int `json:"degraded"` + Offline int `json:"offline"` + Unknown int `json:"unknown"` +} + +type Driver struct { + DriverID string `json:"driver_id"` + Version string `json:"version"` + Source string `json:"source"` + PackageID string `json:"package_id,omitempty"` + RepositoryID string `json:"repository_id,omitempty"` + PackageChannel string `json:"package_channel,omitempty"` + ArtifactSHA256 string `json:"artifact_sha256,omitempty"` + SourceSHA256 string `json:"source_sha256,omitempty"` + ControlClass string `json:"control_class"` + ConfiguredInstances int `json:"configured_instances"` + RunningInstances int `json:"running_instances"` + Health Health `json:"health"` +} + +type Snapshot struct { + SchemaVersion string `json:"schema_version"` + GeneratedAt time.Time `json:"generated_at"` + Host Host `json:"host"` + Drivers []Driver `json:"drivers"` +} + +// RepositoryArtifact describes one activated repository file. PackageID and +// Channel are set only when a signed canonical package supplied them. +type RepositoryArtifact struct { + LogicalPath string + InstalledPath string + DriverID string + Version string + SHA256 string + RepositoryID string + PackageID string + PackageChannel string + ControlClass string +} + +type Input struct { + HostVersion string + Drivers []config.Driver + RunningNames []string + Health map[string]telemetry.DriverHealth + UserDriverDir string + ManagedDriverDir string + BundledDriverDir string + RepositoryDrivers []RepositoryArtifact +} + +func Build(now time.Time, in Input) (Snapshot, error) { + snapshot := Snapshot{ + SchemaVersion: SchemaVersion, + GeneratedAt: now.UTC(), + Host: Host{ + Product: "ftw", + Version: hostVersion(in.HostVersion), + UpdateChannel: updateChannel(in.HostVersion), + Target: "ftw-core", + RuntimeABI: RuntimeABI, + HostAPI: HostAPI, + }, + Drivers: []Driver{}, + } + + running := make(map[string]bool, len(in.RunningNames)) + for _, name := range in.RunningNames { + running[name] = true + } + artifacts := artifactByPath(in.RepositoryDrivers, in.ManagedDriverDir) + grouped := make(map[string]*Driver) + configured := 0 + for _, cfg := range in.Drivers { + if cfg.Disabled { + continue + } + configured++ + if configured > MaxDrivers { + return Snapshot{}, fmt.Errorf("driver inventory exceeds %d configured instances", MaxDrivers) + } + row, err := inspectDriver(cfg.Lua, in, artifacts) + if err != nil { + return Snapshot{}, fmt.Errorf("inspect driver %q: %w", cfg.Name, err) + } + key := identityKey(row) + item := grouped[key] + if item == nil { + copyRow := row + grouped[key] = ©Row + item = ©Row + } + item.ConfiguredInstances++ + if running[cfg.Name] { + item.RunningInstances++ + } + addHealth(&item.Health, in.Health[cfg.Name], running[cfg.Name], hasHealth(in.Health, cfg.Name)) + } + + for _, row := range grouped { + snapshot.Drivers = append(snapshot.Drivers, *row) + } + sort.Slice(snapshot.Drivers, func(i, j int) bool { + return identityKey(snapshot.Drivers[i]) < identityKey(snapshot.Drivers[j]) + }) + return snapshot, nil +} + +func inspectDriver(path string, in Input, artifacts map[string]RepositoryArtifact) (Driver, error) { + if strings.TrimSpace(path) == "" { + return Driver{}, errors.New("Lua path is empty") + } + data, err := os.ReadFile(path) + if err != nil { + return Driver{}, err + } + sum := sha256.Sum256(data) + loadedSHA := hex.EncodeToString(sum[:]) + metadata, _ := drivers.ParseCatalogFile(path) + id := metadata.ID + if !driverIDRE.MatchString(id) { + // File and instance names may contain customer or site details. Keep a + // fixed fallback and let the source hash distinguish unknown drivers. + id = "unknown" + } + version := metadata.Version + if version == "" { + version = "unknown" + } + row := Driver{DriverID: id, Version: version, ControlClass: "unknown"} + if metadata.ReadOnlyDeclared { + if metadata.ReadOnly { + row.ControlClass = "read_only" + } else { + row.ControlClass = "control" + } + } + + if within(path, in.ManagedDriverDir) { + rel, _ := filepath.Rel(in.ManagedDriverDir, path) + artifact, ok := artifacts[filepath.ToSlash(rel)] + if ok && strings.EqualFold(artifact.SHA256, loadedSHA) { + row.DriverID = artifact.DriverID + row.Version = artifact.Version + row.RepositoryID = artifact.RepositoryID + row.ArtifactSHA256 = strings.ToLower(artifact.SHA256) + if artifact.ControlClass == "read_only" || artifact.ControlClass == "control" { + row.ControlClass = artifact.ControlClass + } + if artifact.PackageID != "" && (artifact.PackageChannel == "beta" || artifact.PackageChannel == "stable") { + row.Source = "managed" + row.PackageID = artifact.PackageID + row.PackageChannel = artifact.PackageChannel + } else { + row.Source = "legacy_repository" + } + return row, nil + } + // A missing or changed activation must not claim signed provenance. + row.Source = "local_override" + row.SourceSHA256 = loadedSHA + return row, nil + } + if within(path, in.UserDriverDir) { + row.Source = "local_override" + } else if within(path, in.BundledDriverDir) { + row.Source = "bundled" + } else { + row.Source = "local_override" + } + row.SourceSHA256 = loadedSHA + return row, nil +} + +func artifactByPath(items []RepositoryArtifact, managedDir string) map[string]RepositoryArtifact { + out := make(map[string]RepositoryArtifact, len(items)*2) + for _, item := range items { + logical := filepath.ToSlash(strings.TrimPrefix(item.LogicalPath, "drivers/")) + if logical != "" { + out[logical] = item + } + if item.InstalledPath != "" && within(item.InstalledPath, managedDir) { + rel, err := filepath.Rel(managedDir, item.InstalledPath) + if err == nil { + out[filepath.ToSlash(rel)] = item + } + } + } + return out +} + +func addHealth(out *Health, health telemetry.DriverHealth, running, exists bool) { + if !exists || !running { + out.Unknown++ + return + } + if health.DeviceFault { + out.Degraded++ + return + } + switch health.Status { + case telemetry.StatusOk: + out.OK++ + case telemetry.StatusDegraded: + out.Degraded++ + case telemetry.StatusOffline: + out.Offline++ + default: + out.Unknown++ + } +} + +func hasHealth(health map[string]telemetry.DriverHealth, name string) bool { + _, ok := health[name] + return ok +} + +func identityKey(row Driver) string { + return strings.Join([]string{ + row.DriverID, row.Version, row.Source, row.PackageID, row.RepositoryID, + row.PackageChannel, row.ArtifactSHA256, row.SourceSHA256, row.ControlClass, + }, "\x00") +} + +func within(path, root string) bool { + if path == "" || root == "" { + return false + } + absPath, err := filepath.Abs(path) + if err != nil { + return false + } + absRoot, err := filepath.Abs(root) + if err != nil { + return false + } + rel, err := filepath.Rel(absRoot, absPath) + return err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) +} + +func hostVersion(version string) string { + version = strings.TrimSpace(strings.TrimPrefix(version, "v")) + if version == "" { + return "dev" + } + if len(version) > 64 { + return version[:64] + } + return version +} + +func updateChannel(version string) string { + version = strings.ToLower(version) + if strings.Contains(version, "-beta") { + return "beta" + } + if version != "" && version != "dev" && !strings.Contains(version, "-") { + return "stable" + } + return "unknown" +} diff --git a/go/internal/driverinventory/inventory_test.go b/go/internal/driverinventory/inventory_test.go new file mode 100644 index 00000000..410461c0 --- /dev/null +++ b/go/internal/driverinventory/inventory_test.go @@ -0,0 +1,143 @@ +package driverinventory + +import ( + "crypto/sha256" + "encoding/hex" + "os" + "path/filepath" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/config" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +func writeDriver(t *testing.T, dir, name, id, version string, readOnly bool) (string, string) { + t.Helper() + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + readOnlyLua := "false" + if readOnly { + readOnlyLua = "true" + } + body := []byte("DRIVER = {\n id = \"" + id + "\",\n name = \"Test\",\n version = \"" + version + "\",\n read_only = " + readOnlyLua + "\n}\n") + path := filepath.Join(dir, name) + if err := os.WriteFile(path, body, 0o600); err != nil { + t.Fatal(err) + } + sum := sha256.Sum256(body) + return path, hex.EncodeToString(sum[:]) +} + +func TestBuildGroupsBundledInstancesWithoutLeakingNames(t *testing.T) { + dir := t.TempDir() + path, sourceSHA := writeDriver(t, dir, "sdm630.lua", "sdm630", "1.1.1", true) + last := time.Now() + snapshot, err := Build(time.Date(2026, 7, 20, 10, 0, 0, 0, time.UTC), Input{ + HostVersion: "v1.5.0-beta.2", + BundledDriverDir: dir, + Drivers: []config.Driver{ + {Name: "customer-meter-one", Lua: path}, + {Name: "customer-meter-two", Lua: path}, + }, + RunningNames: []string{"customer-meter-one"}, + Health: map[string]telemetry.DriverHealth{ + "customer-meter-one": {Status: telemetry.StatusOk, LastSuccess: &last}, + }, + }) + if err != nil { + t.Fatal(err) + } + if snapshot.Host.UpdateChannel != "beta" || len(snapshot.Drivers) != 1 { + t.Fatalf("snapshot = %+v", snapshot) + } + row := snapshot.Drivers[0] + if row.DriverID != "sdm630" || row.Source != "bundled" || row.SourceSHA256 != sourceSHA || row.ConfiguredInstances != 2 || row.RunningInstances != 1 || row.Health.OK != 1 || row.Health.Unknown != 1 { + t.Fatalf("row = %+v", row) + } +} + +func TestBuildBindsMatchingCanonicalArtifact(t *testing.T) { + dir := t.TempDir() + path, artifactSHA := writeDriver(t, dir, "sdm630.lua", "sdm630", "1.1.1", true) + snapshot, err := Build(time.Now(), Input{ + HostVersion: "1.5.0", + ManagedDriverDir: dir, + Drivers: []config.Driver{{Name: "meter", Lua: path}}, + RunningNames: []string{"meter"}, + RepositoryDrivers: []RepositoryArtifact{{ + LogicalPath: "drivers/sdm630.lua", DriverID: "sdm630", Version: "1.1.1", + SHA256: artifactSHA, RepositoryID: "sourceful", PackageID: "com.sourceful.driver.sdm630", + PackageChannel: "beta", ControlClass: "read_only", + }}, + }) + if err != nil { + t.Fatal(err) + } + row := snapshot.Drivers[0] + if row.Source != "managed" || row.PackageID != "com.sourceful.driver.sdm630" || row.ArtifactSHA256 != artifactSHA || row.SourceSHA256 != "" || row.ControlClass != "read_only" { + t.Fatalf("row = %+v", row) + } +} + +func TestBuildDoesNotGuessControlClassFromMissingMetadata(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "old.lua") + if err := os.WriteFile(path, []byte("DRIVER = {\n id = \"old\",\n version = \"1.0.0\"\n}\n"), 0o600); err != nil { + t.Fatal(err) + } + snapshot, err := Build(time.Now(), Input{ + HostVersion: "1.5.0", BundledDriverDir: dir, + Drivers: []config.Driver{{Name: "old-instance", Lua: path}}, + }) + if err != nil { + t.Fatal(err) + } + if got := snapshot.Drivers[0].ControlClass; got != "unknown" { + t.Fatalf("control class = %q, want unknown", got) + } +} + +func TestBuildDowngradesChangedManagedFileToLocalOverride(t *testing.T) { + dir := t.TempDir() + path, _ := writeDriver(t, dir, "meter.lua", "meter", "1.0.0", true) + snapshot, err := Build(time.Now(), Input{ + HostVersion: "1.5.0", + ManagedDriverDir: dir, + Drivers: []config.Driver{{Name: "meter", Lua: path}}, + RepositoryDrivers: []RepositoryArtifact{{ + LogicalPath: "drivers/meter.lua", DriverID: "meter", Version: "1.0.0", + SHA256: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + RepositoryID: "sourceful", PackageID: "com.sourceful.driver.meter", PackageChannel: "beta", + }}, + }) + if err != nil { + t.Fatal(err) + } + row := snapshot.Drivers[0] + if row.Source != "local_override" || row.SourceSHA256 == "" || row.PackageID != "" || row.ArtifactSHA256 != "" { + t.Fatalf("row = %+v", row) + } +} + +func TestBuildMarksOldRepositoryArtifactAsLegacy(t *testing.T) { + dir := t.TempDir() + path, artifactSHA := writeDriver(t, dir, "old.lua", "old", "0.9.0", false) + snapshot, err := Build(time.Now(), Input{ + HostVersion: "1.5.0", + ManagedDriverDir: dir, + Drivers: []config.Driver{{Name: "old-instance", Lua: path}}, + RepositoryDrivers: []RepositoryArtifact{{ + LogicalPath: "drivers/old.lua", DriverID: "old", Version: "0.9.0", + SHA256: artifactSHA, RepositoryID: "legacy-ftw", + }}, + }) + if err != nil { + t.Fatal(err) + } + row := snapshot.Drivers[0] + if row.Source != "legacy_repository" || row.RepositoryID != "legacy-ftw" || row.PackageID != "" || row.ArtifactSHA256 != artifactSHA { + t.Fatalf("row = %+v", row) + } +} diff --git a/go/internal/drivers/catalog.go b/go/internal/drivers/catalog.go index 4d764d5f..10727b5e 100644 --- a/go/internal/drivers/catalog.go +++ b/go/internal/drivers/catalog.go @@ -37,6 +37,9 @@ type CatalogEntry struct { // UI uses it to avoid presenting battery capacity as a control opt-in and // to enable battery_telemetry_only for gateway-style drivers. ReadOnly bool `json:"read_only,omitempty"` + // ReadOnlyDeclared distinguishes an explicit false value from old metadata + // that did not state whether the driver can control hardware. + ReadOnlyDeclared bool `json:"-"` // Verification: who's actually run this driver against real // hardware and how long. Populated from the DRIVER block's optional @@ -179,7 +182,7 @@ func parseCatalogEntry(path string) (CatalogEntry, error) { e.Capabilities = pickList(block, "capabilities") e.HTTPHosts = pickList(block, "http_hosts") e.ConnectionDefaults = pickKVBlock(block, "connection_defaults") - e.ReadOnly = pickBool(block, "read_only") + e.ReadOnly, e.ReadOnlyDeclared = pickOptionalBool(block, "read_only") e.VerificationStatus = normalizeVerificationStatus(pickString(block, "verification_status")) e.VerifiedBy = pickList(block, "verified_by") e.VerifiedAt = pickString(block, "verified_at") @@ -292,9 +295,17 @@ func pickInt(block, name string) int { } func pickBool(block, name string) bool { + value, _ := pickOptionalBool(block, name) + return value +} + +func pickOptionalBool(block, name string) (bool, bool) { re := regexp.MustCompile(`(?mi)^\s*` + regexp.QuoteMeta(name) + `\s*=\s*(true|false)`) m := re.FindStringSubmatch(block) - return len(m) >= 2 && strings.EqualFold(m[1], "true") + if len(m) < 2 { + return false, false + } + return strings.EqualFold(m[1], "true"), true } // pickList matches `name = { "a", "b", "c" }` inside the block. diff --git a/go/internal/nova/publisher.go b/go/internal/nova/publisher.go index b98d39df..8bbd77e1 100644 --- a/go/internal/nova/publisher.go +++ b/go/internal/nova/publisher.go @@ -1,6 +1,7 @@ package nova import ( + "crypto/sha256" "encoding/json" "fmt" "log/slog" @@ -10,10 +11,27 @@ import ( paho "github.com/eclipse/paho.mqtt.golang" "github.com/srcfl/ftw/go/internal/config" + "github.com/srcfl/ftw/go/internal/driverinventory" "github.com/srcfl/ftw/go/internal/state" "github.com/srcfl/ftw/go/internal/telemetry" ) +const ( + driverInventoryCheckInterval = 30 * time.Second + driverInventoryMaxInterval = 15 * time.Minute + driverInventoryMaxBytes = 128 * 1024 +) + +type DriverInventoryProvider func(now time.Time) (driverinventory.Snapshot, error) + +type Option func(*Publisher) + +// WithDriverInventory adds the canonical FTW driver inventory heartbeat. +// It does not change driver install, activation or control state. +func WithDriverInventory(provider DriverInventoryProvider) Option { + return func(p *Publisher) { p.driverInventory = provider } +} + // Publisher owns one MQTT connection to Nova's broker plus a periodic // publish loop. It mirrors the shape of internal/ha.Bridge so the // wiring in main.go reads consistently for operators who know HA. @@ -32,13 +50,19 @@ type Publisher struct { lastPublishMs int64 publishedCount int64 missingDERs map[string]bool // one-shot WARN dedupe per (device,type) + + driverInventory DriverInventoryProvider + driverInventoryWake chan struct{} + lastInventorySHA [sha256.Size]byte + lastInventoryPublish time.Time + inventoryPublishCount int64 } // Start connects to Nova's MQTT broker (JWT-as-password) and begins the // publish loop. Returns immediately; the goroutine runs until Stop. // Safe to pass a nil cfg — returns (nil, nil) so callers can gate on // `cfg.Nova != nil && cfg.Nova.Enabled`. -func Start(cfg *config.Nova, id *Identity, store *state.Store, tel *telemetry.Store) (*Publisher, error) { +func Start(cfg *config.Nova, id *Identity, store *state.Store, tel *telemetry.Store, options ...Option) (*Publisher, error) { if cfg == nil || !cfg.Enabled { return nil, nil } @@ -46,13 +70,17 @@ func Start(cfg *config.Nova, id *Identity, store *state.Store, tel *telemetry.St return nil, fmt.Errorf("nova.Start: identity is required") } p := &Publisher{ - cfg: cfg, - id: id, - store: store, - tel: tel, - stop: make(chan struct{}), - done: make(chan struct{}), - missingDERs: make(map[string]bool), + cfg: cfg, + id: id, + store: store, + tel: tel, + stop: make(chan struct{}), + done: make(chan struct{}), + missingDERs: make(map[string]bool), + driverInventoryWake: make(chan struct{}, 1), + } + for _, option := range options { + option(p) } scheme := "tcp" @@ -80,6 +108,7 @@ func Start(cfg *config.Nova, id *Identity, store *state.Store, tel *telemetry.St slog.Info("nova MQTT connected", "broker", fmt.Sprintf("%s:%d", cfg.MQTTHost, cfg.MQTTPort), "gateway_serial", cfg.GatewaySerial) + p.requestDriverInventory() }). SetConnectionLostHandler(func(_ paho.Client, err error) { slog.Warn("nova MQTT connection lost", "err", err) @@ -143,16 +172,90 @@ func (p *Publisher) run() { defer close(p.done) tick := time.NewTicker(time.Duration(p.cfg.PublishIntervalS) * time.Second) defer tick.Stop() + inventoryTick := time.NewTicker(driverInventoryCheckInterval) + defer inventoryTick.Stop() for { select { case <-p.stop: return case <-tick.C: p.publishOnce() + case <-inventoryTick.C: + p.publishDriverInventory(false) + case <-p.driverInventoryWake: + p.publishDriverInventory(true) } } } +func (p *Publisher) requestDriverInventory() { + if p.driverInventory == nil { + return + } + select { + case p.driverInventoryWake <- struct{}{}: + default: + } +} + +func (p *Publisher) publishDriverInventory(force bool) { + if p.driverInventory == nil || p.client == nil || !p.client.IsConnected() { + return + } + now := time.Now().UTC() + snapshot, err := p.driverInventory(now) + if err != nil { + slog.Warn("nova: build driver inventory", "err", err) + return + } + contentSHA, err := driverInventoryContentSHA(snapshot) + if err != nil { + slog.Warn("nova: hash driver inventory", "err", err) + return + } + p.mu.Lock() + unchanged := contentSHA == p.lastInventorySHA + recent := now.Sub(p.lastInventoryPublish) < driverInventoryMaxInterval + p.mu.Unlock() + if !force && unchanged && recent { + return + } + snapshot.GeneratedAt = now + wire, err := json.Marshal(snapshot) + if err != nil { + slog.Warn("nova: encode driver inventory", "err", err) + return + } + if len(wire) > driverInventoryMaxBytes { + slog.Warn("nova: driver inventory exceeds payload limit", "bytes", len(wire)) + return + } + topic := driverInventoryTopic(p.cfg.GatewaySerial) + tok := p.client.Publish(topic, 1, true, wire) + if !tok.WaitTimeout(2*time.Second) || tok.Error() != nil { + slog.Warn("nova: publish driver inventory", "topic", topic, "err", tok.Error()) + return + } + p.mu.Lock() + p.lastInventorySHA = contentSHA + p.lastInventoryPublish = now + p.inventoryPublishCount++ + p.mu.Unlock() +} + +func driverInventoryContentSHA(snapshot driverinventory.Snapshot) ([sha256.Size]byte, error) { + snapshot.GeneratedAt = time.Time{} + wire, err := json.Marshal(snapshot) + if err != nil { + return [sha256.Size]byte{}, err + } + return sha256.Sum256(wire), nil +} + +func driverInventoryTopic(gatewaySerial string) string { + return fmt.Sprintf("gateways/%s/inventory/drivers/json/v1", sanitizeTopicSegment(gatewaySerial)) +} + // publishOnce snapshots every registered device × der_type, assembles // a clean DerTelemetry, translates per SchemaMode, and publishes. // Skips devices/DERs that have not been provisioned in Nova — those diff --git a/go/internal/nova/publisher_test.go b/go/internal/nova/publisher_test.go index 9574f8c2..e511f619 100644 --- a/go/internal/nova/publisher_test.go +++ b/go/internal/nova/publisher_test.go @@ -5,10 +5,44 @@ import ( "testing" "time" + "github.com/srcfl/ftw/go/internal/driverinventory" "github.com/srcfl/ftw/go/internal/state" "github.com/srcfl/ftw/go/internal/telemetry" ) +func TestDriverInventoryContentSHAIgnoresGeneratedAt(t *testing.T) { + first := driverinventory.Snapshot{ + SchemaVersion: driverinventory.SchemaVersion, + GeneratedAt: time.Unix(1, 0), + Host: driverinventory.Host{ + Product: "ftw", Version: "1.5.0-beta.1", UpdateChannel: "beta", + Target: "ftw-core", RuntimeABI: driverinventory.RuntimeABI, HostAPI: driverinventory.HostAPI, + }, + Drivers: []driverinventory.Driver{{ + DriverID: "sdm630", Version: "1.1.1", Source: "bundled", + SourceSHA256: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + ControlClass: "read_only", ConfiguredInstances: 1, RunningInstances: 1, + Health: driverinventory.Health{OK: 1}, + }}, + } + second := first + second.GeneratedAt = time.Unix(2, 0) + firstSHA, err := driverInventoryContentSHA(first) + if err != nil { + t.Fatal(err) + } + secondSHA, err := driverInventoryContentSHA(second) + if err != nil { + t.Fatal(err) + } + if firstSHA != secondSHA { + t.Fatal("generated_at changed content identity") + } + if got := driverInventoryTopic("f42w-gw-1"); got != "gateways/f42w-gw-1/inventory/drivers/json/v1" { + t.Fatalf("topic = %q", got) + } +} + // TestAssemble_PicksUpClean Snake CaseFromLuaEmit confirms that // arbitrary fields a Lua driver emits inside host.emit() flow into // the clean payload unmodified — because the emit convention and