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

package feichibridge

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"time"
)

type VehicleState struct {
	LastRealtimeTime         time.Time `json:"last_realtime_time,omitempty"`
	LastRealtimeHash         string    `json:"last_realtime_hash,omitempty"`
	LastRealtimeACKAt        time.Time `json:"last_realtime_ack_at,omitempty"`
	LastSnapshotReissueTime  time.Time `json:"last_snapshot_reissue_time,omitempty"`
	LastSnapshotReissueHash  string    `json:"last_snapshot_reissue_hash,omitempty"`
	LastSnapshotReissueACKAt time.Time `json:"last_snapshot_reissue_ack_at,omitempty"`
	BackfillCursor           time.Time `json:"backfill_cursor,omitempty"`
	LastBackfillACKAt        time.Time `json:"last_backfill_ack_at,omitempty"`
}

func (s *StateStore) CommitSnapshotReissue(vin string, at time.Time, hash string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	current := s.state.Vehicles[vin]
	current.LastSnapshotReissueTime = at
	current.LastSnapshotReissueHash = hash
	current.LastSnapshotReissueACKAt = time.Now()
	s.state.Vehicles[vin] = current
	return s.saveLocked()
}

type persistentState struct {
	Version        int                     `json:"version"`
	PlatformSerial uint16                  `json:"platform_serial"`
	Vehicles       map[string]VehicleState `json:"vehicles"`
}

type StateStore struct {
	mu    sync.Mutex
	path  string
	state persistentState
}

func OpenStateStore(path string) (*StateStore, error) {
	store := &StateStore{
		path: path,
		state: persistentState{
			Version:  1,
			Vehicles: map[string]VehicleState{},
		},
	}
	encoded, err := os.ReadFile(path)
	if errors.Is(err, os.ErrNotExist) {
		return store, nil
	}
	if err != nil {
		return nil, fmt.Errorf("read bridge state: %w", err)
	}
	if err := json.Unmarshal(encoded, &store.state); err != nil {
		return nil, fmt.Errorf("decode bridge state: %w", err)
	}
	if store.state.Vehicles == nil {
		store.state.Vehicles = map[string]VehicleState{}
	}
	return store, nil
}

func (s *StateStore) Vehicle(vin string) VehicleState {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.state.Vehicles[vin]
}

func (s *StateStore) CommitRealtime(vin string, at time.Time, hash string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	current := s.state.Vehicles[vin]
	current.LastRealtimeTime = at
	current.LastRealtimeHash = hash
	current.LastRealtimeACKAt = time.Now()
	s.state.Vehicles[vin] = current
	return s.saveLocked()
}

func (s *StateStore) CommitBackfill(vin string, at time.Time) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	current := s.state.Vehicles[vin]
	current.BackfillCursor = at
	current.LastBackfillACKAt = time.Now()
	s.state.Vehicles[vin] = current
	return s.saveLocked()
}

func (s *StateStore) AdvanceBackfillCursor(vin string, at time.Time) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	current := s.state.Vehicles[vin]
	current.BackfillCursor = at
	s.state.Vehicles[vin] = current
	return s.saveLocked()
}

func (s *StateStore) NextPlatformSerial() (uint16, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.state.PlatformSerial++
	if s.state.PlatformSerial == 0 {
		s.state.PlatformSerial = 1
	}
	if err := s.saveLocked(); err != nil {
		return 0, err
	}
	return s.state.PlatformSerial, nil
}

func (s *StateStore) saveLocked() error {
	if err := os.MkdirAll(filepath.Dir(s.path), 0o750); err != nil {
		return fmt.Errorf("create bridge state directory: %w", err)
	}
	encoded, err := json.MarshalIndent(s.state, "", "  ")
	if err != nil {
		return err
	}
	temp, err := os.CreateTemp(filepath.Dir(s.path), ".state-*.json")
	if err != nil {
		return err
	}
	tempName := temp.Name()
	defer os.Remove(tempName)
	if err := temp.Chmod(0o600); err != nil {
		temp.Close()
		return err
	}
	if _, err := temp.Write(encoded); err != nil {
		temp.Close()
		return err
	}
	if err := temp.Sync(); err != nil {
		temp.Close()
		return err
	}
	if err := temp.Close(); err != nil {
		return err
	}
	if err := os.Rename(tempName, s.path); err != nil {
		return fmt.Errorf("replace bridge state: %w", err)
	}
	return nil
}

func hashBytes(value []byte) string {
	sum := sha256.Sum256(value)
	return hex.EncodeToString(sum[:])
}