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

package eventbus

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

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

type DurableConfig struct {
	Directory       string
	ReplayBatchSize int
}

type DurableSink struct {
	delegate Sink
	dir      string

	mu              sync.Mutex
	seq             uint64
	rawPending      map[string]struct{}
	replayBatchSize int
}

type durableRecord struct {
	Kind     string                 `json:"kind"`
	Envelope envelope.FrameEnvelope `json:"envelope"`
}

type durableRecordFile struct {
	path   string
	record durableRecord
}

type recordPublishingSink interface {
	PublishRecords(context.Context, []durableRecord) error
}

const durableReplayOperationTimeout = 30 * time.Second

func NewDurableSink(delegate Sink, cfg DurableConfig) *DurableSink {
	if delegate == nil {
		panic("durable delegate sink must not be nil")
	}
	return &DurableSink{
		delegate:        delegate,
		dir:             strings.TrimSpace(cfg.Directory),
		rawPending:      map[string]struct{}{},
		replayBatchSize: cfg.ReplayBatchSize,
	}
}

func (s *DurableSink) PublishRaw(ctx context.Context, env envelope.FrameEnvelope) error {
	if err := s.delegate.PublishRaw(ctx, env); err == nil {
		return nil
	}
	if err := s.spool("raw", env); err != nil {
		return err
	}
	s.markRawPending(env)
	return nil
}

func (s *DurableSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelope) error {
	if s.isRawPending(env) {
		return s.spool("unified", env)
	}
	if err := s.delegate.PublishUnified(ctx, env); err == nil {
		return nil
	}
	return s.spool("unified", env)
}

func (s *DurableSink) PublishFields(ctx context.Context, env envelope.FrameEnvelope) error {
	if s.isRawPending(env) {
		return s.spool("fields", env)
	}
	if err := s.delegate.PublishFields(ctx, env); err == nil {
		return nil
	}
	return s.spool("fields", env)
}

func (s *DurableSink) ReplayOnce(ctx context.Context) error {
	return s.replay(ctx, 0)
}

func (s *DurableSink) replay(ctx context.Context, limit int) error {
	files, err := durableFiles(s.dir, limit)
	if err != nil {
		return err
	}
	records := make([]durableRecordFile, 0, len(files))
	for _, file := range files {
		record, err := readDurableRecord(file)
		if err != nil {
			return err
		}
		records = append(records, durableRecordFile{path: file, record: record})
	}
	sortDurableRecords(records)
	if publisher, ok := s.delegate.(recordPublishingSink); ok {
		durableRecords := make([]durableRecord, 0, len(records))
		for _, item := range records {
			durableRecords = append(durableRecords, item.record)
		}
		if err := publisher.PublishRecords(ctx, durableRecords); err != nil {
			return err
		}
		for _, item := range records {
			if err := os.Remove(item.path); err != nil {
				return err
			}
			if item.record.Kind == "raw" {
				s.clearRawPending(item.record.Envelope)
			}
		}
		return nil
	}
	for _, item := range records {
		if err := s.publishRecord(ctx, item.record); err != nil {
			return err
		}
		if err := os.Remove(item.path); err != nil {
			return err
		}
		if item.record.Kind == "raw" {
			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
	}
	if kind == "fields" {
		return 2
	}
	return 3
}

func (s *DurableSink) ReplayLoop(ctx context.Context, interval time.Duration, onError func(error)) {
	if interval <= 0 {
		interval = time.Second
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if err := s.replayOnceFromLoop(ctx); err != nil && onError != nil {
				onError(err)
			}
		}
	}
}

func (s *DurableSink) replayOnceFromLoop(ctx context.Context) error {
	replayCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), durableReplayOperationTimeout)
	defer cancel()
	return s.replay(replayCtx, s.replayBatchSize)
}

func (s *DurableSink) Close() error {
	return s.delegate.Close()
}

func (s *DurableSink) publishRecord(ctx context.Context, record durableRecord) error {
	switch record.Kind {
	case "raw":
		return s.delegate.PublishRaw(ctx, record.Envelope)
	case "unified":
		return s.delegate.PublishUnified(ctx, record.Envelope)
	case "fields":
		return s.delegate.PublishFields(ctx, record.Envelope)
	default:
		return errUnknownRecordKind(record.Kind)
	}
}

func errUnknownRecordKind(kind string) error {
	return fmt.Errorf("unknown durable record kind %q", kind)
}

func (s *DurableSink) spool(kind string, env envelope.FrameEnvelope) error {
	if s.dir == "" {
		return fmt.Errorf("durable spool directory is empty")
	}
	if env.EventID == "" {
		env.EventID = env.StableEventID()
	}
	if env.ParseStatus == "" {
		env.ParseStatus = envelope.ParseOK
	}
	if err := os.MkdirAll(s.dir, 0o750); err != nil {
		return err
	}
	payload, err := json.Marshal(durableRecord{Kind: kind, Envelope: env})
	if err != nil {
		return err
	}
	name := s.nextFileName(env, kind)
	path := filepath.Join(s.dir, name)
	tmp := path + ".tmp"
	if err := os.WriteFile(tmp, payload, 0o640); err != nil {
		return err
	}
	return os.Rename(tmp, path)
}

func (s *DurableSink) nextFileName(env envelope.FrameEnvelope, kind string) string {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.seq++
	eventID := env.StableEventID()
	if len(eventID) > 12 {
		eventID = eventID[:12]
	}
	return fmt.Sprintf("%020d-%06d-%s-%s.json", time.Now().UnixNano(), s.seq, kind, eventID)
}

func (s *DurableSink) markRawPending(env envelope.FrameEnvelope) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.rawPending[env.StableEventID()] = struct{}{}
}

func (s *DurableSink) clearRawPending(env envelope.FrameEnvelope) {
	s.mu.Lock()
	defer s.mu.Unlock()
	delete(s.rawPending, env.StableEventID())
}

func (s *DurableSink) isRawPending(env envelope.FrameEnvelope) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	_, ok := s.rawPending[env.StableEventID()]
	return ok
}

func readDurableRecord(path string) (durableRecord, error) {
	var record durableRecord
	payload, err := os.ReadFile(path)
	if err != nil {
		return record, err
	}
	return record, json.Unmarshal(payload, &record)
}

func durableFiles(dir string, limit int) ([]string, error) {
	if limit > 0 {
		handle, err := os.Open(dir)
		if err != nil {
			if os.IsNotExist(err) {
				return nil, nil
			}
			return nil, err
		}
		defer handle.Close()
		return durableFilesFromReader(dir, limit, handle)
	}
	files, err := filepath.Glob(filepath.Join(dir, "*.json"))
	if err != nil {
		return nil, err
	}
	sort.Strings(files)
	return files, nil
}

type durableNameReader interface {
	Readdirnames(int) ([]string, error)
}

func durableFilesFromReader(dir string, limit int, reader durableNameReader) ([]string, error) {
	if limit <= 0 {
		return nil, nil
	}
	files := make([]string, 0, limit)
	for len(files) < limit {
		names, err := reader.Readdirnames(limit - len(files))
		for _, name := range names {
			if !strings.HasSuffix(name, ".json") {
				continue
			}
			files = append(files, filepath.Join(dir, name))
			if len(files) >= limit {
				break
			}
		}
		if err != nil {
			if errorsIsEOF(err) {
				break
			}
			return nil, err
		}
		if len(names) == 0 {
			break
		}
	}
	sort.Strings(files)
	return files, nil
}

func errorsIsEOF(err error) bool {
	return err == io.EOF
}