fix(go): replay raw before unified events

go/vehicle-gateway/internal/eventbus/durable_sink.go

@@ -32,6 +32,11 @@ type durableRecord struct {
 	Envelope envelope.FrameEnvelope `json:"envelope"`
 }
 
+type durableRecordFile struct {
+	path   string
+	record durableRecord
+}
+
 func NewDurableSink(delegate Sink, cfg DurableConfig) *DurableSink {
 	if delegate == nil {
 		panic("durable delegate sink must not be nil")
@@ -70,24 +75,50 @@ func (s *DurableSink) ReplayOnce(ctx context.Context) error {
 		return err
 	}
 	sort.Strings(files)
+	records := make([]durableRecordFile, 0, len(files))
 	for _, file := range files {
 		record, err := readDurableRecord(file)
 		if err != nil {
 			return err
 		}
-		if err := s.publishRecord(ctx, record); err != nil {
+		records = append(records, durableRecordFile{path: file, record: record})
+	}
+	sortDurableRecords(records)
+	for _, item := range records {
+		if err := s.publishRecord(ctx, item.record); err != nil {
 			return err
 		}
-		if err := os.Remove(file); err != nil {
+		if err := os.Remove(item.path); err != nil {
 			return err
 		}
-		if record.Kind == "raw" {
+		if item.record.Kind == "raw" {
-			s.clearRawPending(record.Envelope)
+			s.clearRawPending(item.record.Envelope)
 		}
 	}
 	return nil
 }
 
+func sortDurableRecords(records []durableRecordFile) {
+	sort.SliceStable(records, func(i, j int) bool {
+		leftEvent := records[i].record.Envelope.StableEventID()
+		rightEvent := records[j].record.Envelope.StableEventID()
+		if leftEvent == rightEvent {
+			return durableKindOrder(records[i].record.Kind) < durableKindOrder(records[j].record.Kind)
+		}
+		return records[i].path < records[j].path
+	})
+}
+
+func durableKindOrder(kind string) int {
+	if kind == "raw" {
+		return 0
+	}
+	if kind == "unified" {
+		return 1
+	}
+	return 2
+}
+
 func (s *DurableSink) ReplayLoop(ctx context.Context, interval time.Duration, onError func(error)) {
 	if interval <= 0 {
 		interval = time.Second

go/vehicle-gateway/internal/eventbus/durable_sink_test.go

@@ -2,6 +2,7 @@ package eventbus
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"os"
 	"path/filepath"
@@ -70,6 +71,30 @@ func TestDurableSinkReplayPublishesInFileOrderAndDeletesFiles(t *testing.T) {
 	}
 }
 
+func TestDurableSinkReplayPublishesRawBeforeUnifiedEvenWhenFilesAreOutOfOrder(t *testing.T) {
+	dir := t.TempDir()
+	env := durableTestEnvelope()
+	writeDurableRecord(t, filepath.Join(dir, "0001-unified.json"), durableRecord{Kind: "unified", Envelope: env})
+	writeDurableRecord(t, filepath.Join(dir, "0002-raw.json"), durableRecord{Kind: "raw", Envelope: env})
+	delegate := &scriptedSink{}
+	sink := NewDurableSink(delegate, DurableConfig{Directory: dir})
+
+	if err := sink.ReplayOnce(context.Background()); err != nil {
+		t.Fatalf("ReplayOnce() error = %v", err)
+	}
+
+	got := delegate.calls
+	want := []string{"raw", "unified"}
+	if len(got) != len(want) {
+		t.Fatalf("calls = %#v, want %#v", got, want)
+	}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Fatalf("calls[%d] = %q, want %q; all=%#v", i, got[i], want, got)
+		}
+	}
+}
+
 func durableTestEnvelope() envelope.FrameEnvelope {
 	return envelope.FrameEnvelope{
 		Protocol:     envelope.ProtocolJT808,
@@ -83,6 +108,17 @@ func durableTestEnvelope() envelope.FrameEnvelope {
 	}
 }
 
+func writeDurableRecord(t *testing.T, path string, record durableRecord) {
+	t.Helper()
+	payload, err := json.Marshal(record)
+	if err != nil {
+		t.Fatalf("marshal durable record: %v", err)
+	}
+	if err := os.WriteFile(path, payload, 0o640); err != nil {
+		t.Fatalf("write durable record: %v", err)
+	}
+}
+
 func spoolFiles(t *testing.T, dir string) []string {
 	t.Helper()
 	matches, err := filepath.Glob(filepath.Join(dir, "*.json"))