Fix SCUM companion trajectory storage

2026-08-31 15:42:16 +08:00
parent bc927a5144
commit ea4e780562
26 changed files with 1028 additions and 112 deletions
@@ -0,0 +1,166 @@
+package companion
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+)
+
+type TrajectorySource interface {
+	ReadPositionRows(context.Context, int) ([]map[string]any, error)
+	ReadVehicleRows(context.Context, int) ([]map[string]any, error)
+}
+
+type TrajectoryStore interface {
+	EnsureSchema(context.Context) error
+	StorePositionRows(context.Context, string, []map[string]any, time.Time) (int, error)
+	StoreVehicleRows(context.Context, string, []map[string]any, time.Time) (int, error)
+}
+
+type TrajectoryCollectionReport struct {
+	CollectedAt   time.Time
+	PositionRows  int
+	VehicleRows   int
+	StoredSamples int
+	Status        string
+	Reason        string
+}
+
+type TrajectoryCollector struct {
+	Source           TrajectorySource
+	Store            TrajectoryStore
+	ServerInstanceID string
+	Interval         time.Duration
+	MaxRows          int
+	Now              func() time.Time
+	schemaOnce       sync.Once
+	schemaErr        error
+}
+
+func NewTrajectoryCollector(config Config, source TrajectorySource, store TrajectoryStore) *TrajectoryCollector {
+	return &TrajectoryCollector{
+		Source:           source,
+		Store:            store,
+		ServerInstanceID: config.Component.ServerInstanceID,
+		Interval:         time.Duration(config.Trajectory.IntervalSeconds) * time.Second,
+		MaxRows:          config.Trajectory.MaxRows,
+	}
+}
+
+func (collector *TrajectoryCollector) CollectOnce(ctx context.Context) (TrajectoryCollectionReport, error) {
+	if collector == nil || collector.Source == nil || collector.Store == nil || strings.TrimSpace(collector.ServerInstanceID) == "" {
+		return TrajectoryCollectionReport{}, fmt.Errorf("SCUM trajectory collector is not configured")
+	}
+	collector.schemaOnce.Do(func() { collector.schemaErr = collector.Store.EnsureSchema(ctx) })
+	if collector.schemaErr != nil {
+		return TrajectoryCollectionReport{}, collector.schemaErr
+	}
+	sampledAt := collector.clock()().UTC()
+	report := TrajectoryCollectionReport{CollectedAt: sampledAt, Status: "healthy"}
+	positions, err := collector.Source.ReadPositionRows(ctx, collector.MaxRows)
+	if err != nil {
+		report.Status, report.Reason = "degraded", "position collection failed"
+		return report, err
+	}
+	report.PositionRows = len(positions)
+	written, err := collector.Store.StorePositionRows(ctx, collector.ServerInstanceID, positions, sampledAt)
+	if err != nil {
+		report.Status, report.Reason = "degraded", "position storage failed"
+		return report, err
+	}
+	report.StoredSamples += written
+	vehicles, err := collector.Source.ReadVehicleRows(ctx, collector.MaxRows)
+	if err != nil {
+		report.Status, report.Reason = "degraded", "vehicle collection failed"
+		return report, err
+	}
+	report.VehicleRows = len(vehicles)
+	written, err = collector.Store.StoreVehicleRows(ctx, collector.ServerInstanceID, vehicles, sampledAt)
+	if err != nil {
+		report.Status, report.Reason = "degraded", "vehicle storage failed"
+		return report, err
+	}
+	report.StoredSamples += written
+	report.Reason = "raw world coordinates stored"
+	return report, nil
+}
+
+func (collector *TrajectoryCollector) Run(ctx context.Context, status *TrajectoryCollectionStatus) error {
+	interval := collector.Interval
+	if interval < time.Second {
+		interval = time.Duration(DefaultTrajectoryCollectionIntervalSecs) * time.Second
+	}
+	if report, err := collector.CollectOnce(ctx); status != nil {
+		status.Record(report, err)
+	} else if err != nil {
+		return err
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+			report, err := collector.CollectOnce(ctx)
+			if status != nil {
+				status.Record(report, err)
+				continue
+			}
+			if err != nil {
+				return err
+			}
+		}
+	}
+}
+
+func (collector *TrajectoryCollector) clock() func() time.Time {
+	if collector.Now != nil {
+		return collector.Now
+	}
+	return time.Now
+}
+
+type TrajectoryCollectionStatus struct {
+	mu     sync.Mutex
+	latest TrajectoryCollectionReport
+	err    error
+}
+
+func (status *TrajectoryCollectionStatus) Record(report TrajectoryCollectionReport, err error) {
+	if status == nil {
+		return
+	}
+	status.mu.Lock()
+	defer status.mu.Unlock()
+	status.latest = report
+	status.err = err
+}
+
+func (status *TrajectoryCollectionStatus) HealthReport() HealthReport {
+	if status == nil {
+		return HealthReport{Status: "healthy", Reason: "typed companion dispatcher ready"}
+	}
+	status.mu.Lock()
+	defer status.mu.Unlock()
+	if status.latest.Status == "healthy" && status.err == nil {
+		return HealthReport{Status: "healthy", Reason: safeHealthReason(status.latest.Reason, "typed companion dispatcher ready")}
+	}
+	if !status.latest.CollectedAt.IsZero() && status.err == nil {
+		return HealthReport{Status: "healthy", Reason: "raw world coordinate collection ready"}
+	}
+	if status.err != nil {
+		return HealthReport{Status: "degraded", Reason: safeHealthReason(status.latest.Reason, "trajectory collection waiting for source data")}
+	}
+	return HealthReport{Status: "degraded", Reason: "trajectory collection waiting for first sample"}
+}
+
+func safeHealthReason(value string, fallback string) string {
+	value = strings.TrimSpace(value)
+	if value == "" {
+		return fallback
+	}
+	return value
+}