feat: expand vehicle data platform capabilities

go/vehicle-gateway/cmd/stat-writer/main.go

@@ -75,6 +75,13 @@ func main() {
 			os.Exit(1)
 		}
 	}
+	if cfg.HydrogenCapacitySyncEnabled {
+		syncHydrogenCapacities(ctx, logger, registry, db, writer, cfg)
+	} else if loaded, err := writer.ReloadHydrogenTankCapacities(ctx); err != nil {
+		logger.Warn("hydrogen tank capacity cache load failed", "error", err)
+	} else {
+		registry.SetGauge("vehicle_stat_hydrogen_capacity_cache_entries", nil, float64(loaded))
+	}
 	if cfg.NormalizePlatformSourcesOnStart {
 		statDate := time.Now().In(cfg.Location).Format("2006-01-02")
 		normalizeCtx, cancel := context.WithTimeout(ctx, cfg.NormalizePlatformSourcesTimeout)
@@ -108,6 +115,13 @@ func main() {
 
 	logger.Info("stat writer started", "group", cfg.KafkaGroup, "topics", strings.Join(cfg.KafkaTopics, ","), "workers", cfg.Workers, "project_interval_seconds", cfg.ProjectInterval.Seconds(), "source_touch_interval_seconds", cfg.SourceTouchInterval.Seconds(), "cache_retention_seconds", cfg.CacheRetention.Seconds(), "cache_cleanup_interval_seconds", cfg.CacheCleanupInterval.Seconds(), "baseline_miss_ttl_seconds", cfg.BaselineMissTTL.Seconds(), "baseline_hit_ttl_seconds", cfg.BaselineHitTTL.Seconds(), "cache_max_entries", cfg.CacheMaxEntries, "batch_size", cfg.BatchSize, "batch_wait_ms", cfg.BatchWait, "retry_attempts", cfg.RetryAttempts, "retry_delay_ms", cfg.RetryDelay.Milliseconds(), "quarantine_dir", cfg.QuarantineDir)
 	var background sync.WaitGroup
+	if cfg.HydrogenCapacitySyncEnabled && cfg.HydrogenCapacitySyncInterval > 0 {
+		background.Add(1)
+		go func() {
+			defer background.Done()
+			runHydrogenCapacitySync(ctx, logger, registry, db, writer, cfg)
+		}()
+	}
 	if cfg.ProjectInterval > 0 {
 		background.Add(1)
 		go func() {
@@ -132,6 +146,46 @@ func main() {
 	}
 }
 
+func runHydrogenCapacitySync(ctx context.Context, logger interface {
+	Info(string, ...any)
+	Warn(string, ...any)
+}, registry *metrics.Registry, db *sql.DB, writer *stats.Writer, cfg config) {
+	ticker := time.NewTicker(cfg.HydrogenCapacitySyncInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			syncHydrogenCapacities(ctx, logger, registry, db, writer, cfg)
+		}
+	}
+}
+
+func syncHydrogenCapacities(ctx context.Context, logger interface {
+	Info(string, ...any)
+	Warn(string, ...any)
+}, registry *metrics.Registry, db *sql.DB, writer *stats.Writer, cfg config) {
+	syncCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), cfg.HydrogenCapacitySyncTimeout)
+	defer cancel()
+	result, err := stats.SyncHydrogenTankCapacities(syncCtx, db, cfg.HydrogenCapacitySourceSchema)
+	if err != nil {
+		registry.IncCounter("vehicle_stat_hydrogen_capacity_sync_total", metrics.Labels{"status": "error"})
+		logger.Warn("hydrogen tank capacity sync failed; retaining current cache", "source_schema", cfg.HydrogenCapacitySourceSchema, "error", err)
+		return
+	}
+	loaded, err := writer.ReloadHydrogenTankCapacities(syncCtx)
+	if err != nil {
+		registry.IncCounter("vehicle_stat_hydrogen_capacity_sync_total", metrics.Labels{"status": "cache_error"})
+		logger.Warn("hydrogen tank capacity cache reload failed", "error", err)
+		return
+	}
+	registry.IncCounter("vehicle_stat_hydrogen_capacity_sync_total", metrics.Labels{"status": "ok"})
+	registry.SetGauge("vehicle_stat_hydrogen_capacity_cache_entries", nil, float64(loaded))
+	metrics.RecordLastActivity(registry, "vehicle_stat_hydrogen_capacity_last_sync_unix_seconds", nil)
+	logger.Info("hydrogen tank capacities synchronized", "source_schema", cfg.HydrogenCapacitySourceSchema, "read", result.Read, "written", result.Written, "deactivated", result.Deactivated, "cache_entries", loaded)
+}
+
 type pendingProjectionFlusher interface {
 	FlushPendingProjections(context.Context, time.Time) (stats.ProjectionFlushResult, error)
 }
@@ -790,6 +844,23 @@ func recordStatSampleMetrics(registry *metrics.Registry, message kafka.Message,
 	addStatSampleMetric(registry, message, protocol, "skipped_same_mileage", result.SamplesSkippedSameMileage)
 	addStatSampleMetric(registry, message, protocol, "skipped_missing_source", result.SamplesSkippedMissingSource)
 	addStatSampleMetric(registry, message, protocol, "event_time_future_adjusted", result.SamplesAdjustedFutureEventTime)
+	addHydrogenStreamMetric(registry, message, "found", result.HydrogenSamplesFound)
+	addHydrogenStreamMetric(registry, message, "written", result.HydrogenSamplesWritten)
+	addHydrogenStreamMetric(registry, message, "duplicate_or_late", result.HydrogenSamplesDuplicate)
+	addHydrogenStreamMetric(registry, message, "invalid", result.HydrogenSamplesInvalid)
+	addHydrogenStreamMetric(registry, message, "not_current_date", result.HydrogenSamplesNotCurrent)
+}
+
+func addHydrogenStreamMetric(registry *metrics.Registry, message kafka.Message, status string, count int) {
+	if registry == nil || count == 0 {
+		return
+	}
+	registry.AddCounter("vehicle_stat_hydrogen_stream_total", metrics.Labels{
+		"topic": message.Topic, "status": status,
+	}, float64(count))
+	if status == "written" {
+		metrics.RecordLastActivity(registry, "vehicle_stat_last_hydrogen_stream_write_unix_seconds", metrics.Labels{"status": status})
+	}
 }
 
 func addStatSampleMetric(registry *metrics.Registry, message kafka.Message, protocol envelope.Protocol, status string, count int) {
@@ -950,6 +1021,10 @@ type config struct {
 	ReconcileProjectionsOnStart     bool
 	ReconcileProjectionsTimeout     time.Duration
 	QuarantineDir                   string
+	HydrogenCapacitySyncEnabled     bool
+	HydrogenCapacitySourceSchema    string
+	HydrogenCapacitySyncInterval    time.Duration
+	HydrogenCapacitySyncTimeout     time.Duration
 }
 
 func (c config) Validate() error {
@@ -996,6 +1071,10 @@ func loadConfig() config {
 		ReconcileProjectionsOnStart:     env("STATS_RECONCILE_PROJECTIONS_ON_START", "true") != "false",
 		ReconcileProjectionsTimeout:     time.Duration(envInt("STATS_RECONCILE_PROJECTIONS_TIMEOUT_SECONDS", 30)) * time.Second,
 		QuarantineDir:                   env("STATS_QUARANTINE_DIR", "/var/lib/lingniu-go-native/stat-writer-quarantine"),
+		HydrogenCapacitySyncEnabled:     env("HYDROGEN_CAPACITY_SYNC_ENABLED", "true") != "false",
+		HydrogenCapacitySourceSchema:    env("HYDROGEN_CAPACITY_SOURCE_SCHEMA", "ln_asset_management"),
+		HydrogenCapacitySyncInterval:    time.Duration(envInt("HYDROGEN_CAPACITY_SYNC_INTERVAL_SECONDS", 21600)) * time.Second,
+		HydrogenCapacitySyncTimeout:     time.Duration(envInt("HYDROGEN_CAPACITY_SYNC_TIMEOUT_SECONDS", 60)) * time.Second,
 	}
 }