lingniu-vehicle-ingest/go/vehicle-gateway/internal/feichibridge/service.go

package feichibridge

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sort"
	"strings"
	"sync"
	"time"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
)

type Source interface {
	Vehicles(context.Context) ([]Vehicle, error)
	Snapshot(context.Context, string) (Snapshot, error)
	History(context.Context, string, time.Time, time.Time) ([]Record, error)
}

type FrameTarget interface {
	Connect(context.Context) error
	Send(context.Context, []byte) error
	Close() error
	LastACK() time.Time
}

type ServiceConfig struct {
	PollInterval        time.Duration
	DiscoveryInterval   time.Duration
	BackfillInterval    time.Duration
	BackfillLookback    time.Duration
	BackfillWindow      time.Duration
	BackfillSafetyLag   time.Duration
	SourceStaleAfter    time.Duration
	FetchConcurrency    int
	BackfillEnabled     bool
	StaleReissueEnabled bool
}

type Service struct {
	config  ServiceConfig
	source  Source
	target  FrameTarget
	state   *StateStore
	encoder Encoder
	logger  *slog.Logger
	metrics *metrics.Registry

	mu                sync.RWMutex
	vehicles          []Vehicle
	lastSourceSuccess time.Time
	lastError         error
}

func NewService(config ServiceConfig, source Source, target FrameTarget, state *StateStore, logger *slog.Logger, registry *metrics.Registry) (*Service, error) {
	if source == nil || target == nil || state == nil {
		return nil, errors.New("source, target, and state are required")
	}
	if config.PollInterval <= 0 {
		config.PollInterval = 10 * time.Second
	}
	if config.DiscoveryInterval <= 0 {
		config.DiscoveryInterval = 5 * time.Minute
	}
	if config.BackfillInterval <= 0 {
		config.BackfillInterval = time.Hour
	}
	if config.BackfillLookback <= 0 {
		config.BackfillLookback = time.Hour
	}
	if config.BackfillWindow <= 0 {
		config.BackfillWindow = 20 * time.Minute
	}
	if config.BackfillSafetyLag <= 0 {
		config.BackfillSafetyLag = 30 * time.Second
	}
	if config.SourceStaleAfter <= 0 {
		config.SourceStaleAfter = 2 * time.Minute
	}
	if config.FetchConcurrency <= 0 {
		config.FetchConcurrency = 4
	}
	if logger == nil {
		logger = slog.Default()
	}
	return &Service{
		config: config, source: source, target: target, state: state,
		logger: logger, metrics: registry,
	}, nil
}

func (s *Service) Run(ctx context.Context) error {
	if err := s.target.Connect(ctx); err != nil {
		s.setError(err)
		return fmt.Errorf("connect GB/T 32960 target: %w", err)
	}
	if err := s.discover(ctx); err != nil {
		s.setError(err)
		return fmt.Errorf("initial vehicle discovery: %w", err)
	}

	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		s.discoveryLoop(ctx)
	}()
	go func() {
		defer wg.Done()
		s.realtimeLoop(ctx)
	}()
	if s.config.BackfillEnabled {
		wg.Add(1)
		go func() {
			defer wg.Done()
			s.backfillLoop(ctx)
		}()
	}
	<-ctx.Done()
	_ = s.target.Close()
	wg.Wait()
	return nil
}

func (s *Service) Ready(context.Context) error {
	s.mu.RLock()
	lastSourceSuccess := s.lastSourceSuccess
	lastErr := s.lastError
	s.mu.RUnlock()
	maxSourceAge := max(3*s.config.PollInterval, time.Minute)
	if lastSourceSuccess.IsZero() || time.Since(lastSourceSuccess) > maxSourceAge {
		if lastErr != nil {
			return fmt.Errorf("source unavailable: %w", lastErr)
		}
		return errors.New("source has not completed a successful request")
	}
	lastACK := s.target.LastACK()
	if lastACK.IsZero() || time.Since(lastACK) > max(3*s.config.DiscoveryInterval, 10*time.Minute) {
		return errors.New("GB/T 32960 target has no recent ACK")
	}
	return nil
}

func (s *Service) discoveryLoop(ctx context.Context) {
	ticker := time.NewTicker(s.config.DiscoveryInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if err := s.discover(ctx); err != nil {
				s.recordFailure("discover", err)
			}
		}
	}
}

func (s *Service) realtimeLoop(ctx context.Context) {
	s.pollRealtime(ctx)
	ticker := time.NewTicker(s.config.PollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.pollRealtime(ctx)
		}
	}
}

func (s *Service) backfillLoop(ctx context.Context) {
	s.runBackfill(ctx)
	ticker := time.NewTicker(s.config.BackfillInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.runBackfill(ctx)
		}
	}
}

func (s *Service) discover(ctx context.Context) error {
	start := time.Now()
	vehicles, err := s.source.Vehicles(ctx)
	s.observeAPI("vehicles", start, err)
	if err != nil {
		return err
	}
	filtered := make([]Vehicle, 0, len(vehicles))
	for _, vehicle := range vehicles {
		vehicle.VIN = strings.TrimSpace(vehicle.VIN)
		vehicle.VehicleID = strings.TrimSpace(vehicle.VehicleID)
		if len(vehicle.VIN) != 17 || vehicle.VehicleID == "" {
			continue
		}
		if vehicle.RuleTypeName != "" && !strings.Contains(strings.ToUpper(vehicle.RuleTypeName), "32960") {
			continue
		}
		filtered = append(filtered, vehicle)
	}
	sort.Slice(filtered, func(i, j int) bool { return filtered[i].VIN < filtered[j].VIN })
	s.mu.Lock()
	s.vehicles = filtered
	s.lastSourceSuccess = time.Now()
	s.lastError = nil
	s.mu.Unlock()
	if s.metrics != nil {
		s.metrics.SetGauge("vehicle_feichi_bridge_vehicles", nil, float64(len(filtered)))
	}
	s.logger.Info("feichi vehicles discovered", "count", len(filtered))
	return nil
}

type fetchedSnapshot struct {
	vehicle  Vehicle
	record   Record
	at       time.Time
	hash     string
	err      error
	duration time.Duration
}

func (s *Service) pollRealtime(ctx context.Context) {
	vehicles := s.vehicleSnapshot()
	jobs := make(chan Vehicle)
	results := make(chan fetchedSnapshot, len(vehicles))
	var wg sync.WaitGroup
	for worker := 0; worker < min(s.config.FetchConcurrency, len(vehicles)); worker++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for vehicle := range jobs {
				start := time.Now()
				snapshot, err := s.source.Snapshot(ctx, vehicle.VehicleID)
				result := fetchedSnapshot{vehicle: vehicle, duration: time.Since(start), err: err}
				if err == nil {
					result.record = Record(snapshot.DataItems)
					result.at, result.err = recordTime(result.record)
					result.hash = canonicalHash(result.record)
				}
				results <- result
			}
		}()
	}
	for _, vehicle := range vehicles {
		jobs <- vehicle
	}
	close(jobs)
	wg.Wait()
	close(results)

	var snapshots []fetchedSnapshot
	for result := range results {
		s.observeAPIWithDuration("snapshot", result.duration, result.err)
		if result.err != nil {
			s.recordFailure("snapshot", fmt.Errorf("VIN %s: %w", result.vehicle.VIN, result.err))
			continue
		}
		snapshots = append(snapshots, result)
		s.markSourceSuccess()
	}
	sort.Slice(snapshots, func(i, j int) bool {
		if snapshots[i].at.Equal(snapshots[j].at) {
			return snapshots[i].vehicle.VIN < snapshots[j].vehicle.VIN
		}
		return snapshots[i].at.Before(snapshots[j].at)
	})
	for _, snapshot := range snapshots {
		if time.Since(snapshot.at) > s.config.SourceStaleAfter {
			if s.config.StaleReissueEnabled {
				s.reissueStaleSnapshot(ctx, snapshot)
			}
			continue
		}
		current := s.state.Vehicle(snapshot.vehicle.VIN)
		if snapshot.at.Before(current.LastRealtimeTime) ||
			(snapshot.at.Equal(current.LastRealtimeTime) && snapshot.hash == current.LastRealtimeHash) {
			continue
		}
		frame, err := s.encoder.DataFrame(CommandRealtime, snapshot.vehicle.VIN, snapshot.at, snapshot.record)
		if err != nil {
			s.recordFailure("encode", fmt.Errorf("VIN %s: %w", snapshot.vehicle.VIN, err))
			continue
		}
		if err := s.target.Send(ctx, frame); err != nil {
			s.recordFrame(CommandRealtime, "error", snapshot.vehicle.VIN, snapshot.at)
			s.recordFailure("send", fmt.Errorf("VIN %s: %w", snapshot.vehicle.VIN, err))
			continue
		}
		if err := s.state.CommitRealtime(snapshot.vehicle.VIN, snapshot.at, snapshot.hash); err != nil {
			s.recordFailure("state", err)
			continue
		}
		s.recordFrame(CommandRealtime, "acked", snapshot.vehicle.VIN, snapshot.at)
	}
}

func (s *Service) reissueStaleSnapshot(ctx context.Context, snapshot fetchedSnapshot) {
	current := s.state.Vehicle(snapshot.vehicle.VIN)
	if snapshot.at.Before(current.LastSnapshotReissueTime) ||
		(snapshot.at.Equal(current.LastSnapshotReissueTime) && snapshot.hash == current.LastSnapshotReissueHash) {
		return
	}
	frame, err := s.encoder.DataFrame(CommandReissue, snapshot.vehicle.VIN, snapshot.at, snapshot.record)
	if err != nil {
		s.recordFailure("encode_stale_reissue", fmt.Errorf("VIN %s: %w", snapshot.vehicle.VIN, err))
		return
	}
	if err := s.target.Send(ctx, frame); err != nil {
		s.recordFrame(CommandReissue, "error", snapshot.vehicle.VIN, snapshot.at)
		s.recordFailure("send_stale_reissue", fmt.Errorf("VIN %s: %w", snapshot.vehicle.VIN, err))
		return
	}
	if err := s.state.CommitSnapshotReissue(snapshot.vehicle.VIN, snapshot.at, snapshot.hash); err != nil {
		s.recordFailure("state", err)
		return
	}
	s.recordFrame(CommandReissue, "acked", snapshot.vehicle.VIN, snapshot.at)
}

func (s *Service) runBackfill(ctx context.Context) {
	for _, vehicle := range s.vehicleSnapshot() {
		if ctx.Err() != nil {
			return
		}
		if err := s.backfillVehicle(ctx, vehicle); err != nil {
			s.recordFailure("backfill", fmt.Errorf("VIN %s: %w", vehicle.VIN, err))
		}
	}
}

func (s *Service) backfillVehicle(ctx context.Context, vehicle Vehicle) error {
	end := time.Now().Add(-s.config.BackfillSafetyLag)
	cursor := s.state.Vehicle(vehicle.VIN).BackfillCursor
	if cursor.IsZero() {
		cursor = end.Add(-s.config.BackfillLookback)
	}
	for cursor.Before(end) {
		windowEnd := cursor.Add(s.config.BackfillWindow)
		if windowEnd.After(end) {
			windowEnd = end
		}
		start := time.Now()
		records, err := s.source.History(ctx, vehicle.VIN, cursor, windowEnd)
		s.observeAPI("history", start, err)
		if err != nil {
			return err
		}
		sort.Slice(records, func(i, j int) bool {
			left, _ := recordTime(records[i])
			right, _ := recordTime(records[j])
			return left.Before(right)
		})
		committed := cursor
		for _, record := range records {
			at, err := recordTime(record)
			if err != nil || !at.After(cursor) || at.After(windowEnd) {
				continue
			}
			frame, err := s.encoder.DataFrame(CommandReissue, vehicle.VIN, at, record)
			if err != nil {
				return err
			}
			if err := s.target.Send(ctx, frame); err != nil {
				s.recordFrame(CommandReissue, "error", vehicle.VIN, at)
				return err
			}
			if err := s.state.CommitBackfill(vehicle.VIN, at); err != nil {
				return err
			}
			s.recordFrame(CommandReissue, "acked", vehicle.VIN, at)
			committed = at
		}
		if !committed.After(cursor) || committed.Before(windowEnd) {
			if err := s.state.AdvanceBackfillCursor(vehicle.VIN, windowEnd); err != nil {
				return err
			}
		}
		cursor = windowEnd
		s.markSourceSuccess()
	}
	return nil
}

func (s *Service) vehicleSnapshot() []Vehicle {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return append([]Vehicle(nil), s.vehicles...)
}

func (s *Service) markSourceSuccess() {
	s.mu.Lock()
	s.lastSourceSuccess = time.Now()
	s.lastError = nil
	s.mu.Unlock()
	if s.metrics != nil {
		metrics.RecordLastActivity(s.metrics, "vehicle_feichi_bridge_source_last_success_unix_seconds", nil)
	}
}

func (s *Service) setError(err error) {
	s.mu.Lock()
	s.lastError = err
	s.mu.Unlock()
}

func (s *Service) recordFailure(operation string, err error) {
	if err == nil {
		return
	}
	s.setError(err)
	if s.metrics != nil {
		s.metrics.IncCounter("vehicle_feichi_bridge_errors_total", metrics.Labels{"operation": operation})
	}
	s.logger.Error("feichi bridge operation failed", "operation", operation, "error", err)
}

func (s *Service) observeAPI(operation string, start time.Time, err error) {
	s.observeAPIWithDuration(operation, time.Since(start), err)
}

func (s *Service) observeAPIWithDuration(operation string, duration time.Duration, err error) {
	if s.metrics == nil {
		return
	}
	status := "ok"
	if err != nil {
		status = "error"
	}
	s.metrics.IncCounter("vehicle_feichi_bridge_api_requests_total", metrics.Labels{"operation": operation, "status": status})
	s.metrics.ObserveHistogram(
		"vehicle_feichi_bridge_api_request_duration_seconds",
		metrics.Labels{"operation": operation},
		[]float64{0.1, 0.25, 0.5, 1, 2, 5},
		duration.Seconds(),
	)
}

func (s *Service) recordFrame(command byte, status, vin string, sourceTime time.Time) {
	if s.metrics != nil {
		s.metrics.IncCounter("vehicle_feichi_bridge_frames_total", metrics.Labels{
			"command": fmt.Sprintf("0x%02X", command),
			"status":  status,
		})
	}
	if status == "acked" {
		if s.metrics != nil {
			metrics.RecordLastActivity(s.metrics, "vehicle_feichi_bridge_target_last_ack_unix_seconds", nil)
			s.metrics.SetGauge(
				"vehicle_feichi_bridge_vehicle_last_ack_unix_seconds",
				metrics.Labels{"vin": vin, "command": fmt.Sprintf("0x%02X", command)},
				float64(time.Now().Unix()),
			)
		}
		s.logger.Info(
			"GB/T 32960 vehicle frame acknowledged",
			"vin", vin,
			"command", fmt.Sprintf("0x%02X", command),
			"source_time", sourceTime,
		)
	}
}