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

package eventbus

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

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

type DurableOutboxConfig struct {
	Directory       string
	ReplayBatchSize int
	SyncWrites      bool
	CloseTimeout    time.Duration
	WALSegmentBytes int64
	WALSegmentAge   time.Duration
	WALAppendQueue  int
	WALCommitBatch  int
	WALCommitWait   time.Duration
	Metrics         *metrics.Registry
	Name            string
	OnError         func(error)
}

type asyncRecordPublishingSink interface {
	Sink
	ValidateRecord(durableRecord) error
	PublishRecordAsync(durableRecord, func(error)) error
}

// DurableOutboxSink accepts a record only after its WAL commit is durable.
// Publishing is asynchronous; the WAL record remains replayable until the
// broker returns PubAck. Stable event IDs make crash-window replays idempotent.
type DurableOutboxSink struct {
	delegate        asyncRecordPublishingSink
	store           *durableOutboxWAL
	replayBatchSize int
	closeTimeout    time.Duration
	metrics         *metrics.Registry
	name            string
	onError         func(error)

	acceptMu sync.RWMutex
	mu       sync.Mutex
	closed   bool
	inflight map[outboxWALRecordRef]struct{}
	pending  sync.WaitGroup
	closeOne sync.Once
	closeErr error
}

var ErrDurableOutboxClosed = errors.New("durable outbox is closed")

func NewDurableOutboxSink(delegate Sink, cfg DurableOutboxConfig) (*DurableOutboxSink, error) {
	publisher, ok := delegate.(asyncRecordPublishingSink)
	if !ok || publisher == nil {
		return nil, errors.New("durable outbox delegate must support async record publishing")
	}
	dir := strings.TrimSpace(cfg.Directory)
	if dir == "" {
		return nil, errors.New("durable outbox directory is required")
	}
	if cfg.ReplayBatchSize <= 0 {
		cfg.ReplayBatchSize = 1000
	}
	if cfg.CloseTimeout <= 0 {
		cfg.CloseTimeout = 5 * time.Second
	}
	store, err := newDurableOutboxWAL(durableOutboxWALConfig{
		Directory:      dir,
		SyncWrites:     cfg.SyncWrites,
		SegmentBytes:   cfg.WALSegmentBytes,
		SegmentAge:     cfg.WALSegmentAge,
		AppendQueue:    cfg.WALAppendQueue,
		CommitBatch:    cfg.WALCommitBatch,
		CommitInterval: cfg.WALCommitWait,
	})
	if err != nil {
		return nil, err
	}
	s := &DurableOutboxSink{
		delegate:        publisher,
		store:           store,
		replayBatchSize: cfg.ReplayBatchSize,
		closeTimeout:    cfg.CloseTimeout,
		metrics:         cfg.Metrics,
		name:            durableMetricName(cfg.Name),
		onError:         cfg.OnError,
		inflight:        map[outboxWALRecordRef]struct{}{},
	}
	s.recordBacklog()
	return s, nil
}

func (s *DurableOutboxSink) PublishRaw(ctx context.Context, env envelope.FrameEnvelope) error {
	return s.persistAndSubmit(ctx, "raw", env)
}

func (s *DurableOutboxSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelope) error {
	return s.persistAndSubmit(ctx, "unified", env)
}

func (s *DurableOutboxSink) PublishFields(ctx context.Context, env envelope.FrameEnvelope) error {
	return s.persistAndSubmit(ctx, "fields", env)
}

func (s *DurableOutboxSink) Close() error {
	s.closeOne.Do(func() {
		s.acceptMu.Lock()
		s.mu.Lock()
		s.closed = true
		s.mu.Unlock()
		s.acceptMu.Unlock()

		walErr := s.store.Close()
		done := make(chan struct{})
		go func() {
			s.pending.Wait()
			close(done)
		}()
		var waitErr error
		select {
		case <-done:
		case <-time.After(s.closeTimeout):
			waitErr = fmt.Errorf("durable outbox close timed out after %s", s.closeTimeout)
			s.recordPublish("all", "close_timeout")
		}
		s.closeErr = errors.Join(walErr, waitErr, s.delegate.Close())
	})
	return s.closeErr
}

// ReplayOnce claims at most one configured batch. A bounded claim prevents a
// large recovered backlog from creating an unbounded number of PubAck futures.
func (s *DurableOutboxSink) ReplayOnce(ctx context.Context) error {
	s.acceptMu.RLock()
	defer s.acceptMu.RUnlock()
	if s.isClosed() {
		return ErrDurableOutboxClosed
	}
	return s.replay(ctx, s.replayBatchSize)
}

func (s *DurableOutboxSink) ReplayLoop(ctx context.Context, interval time.Duration) {
	if interval <= 0 {
		interval = time.Second
	}
	if err := s.ReplayOnce(ctx); err != nil && !errors.Is(err, ErrDurableOutboxClosed) {
		s.reportError(err)
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if err := s.ReplayOnce(ctx); err != nil && !errors.Is(err, ErrDurableOutboxClosed) {
				s.reportError(err)
			}
		}
	}
}

func (s *DurableOutboxSink) persistAndSubmit(ctx context.Context, kind string, env envelope.FrameEnvelope) error {
	s.acceptMu.RLock()
	defer s.acceptMu.RUnlock()
	if s.isClosed() {
		return ErrDurableOutboxClosed
	}
	record := durableRecord{Kind: kind, Envelope: normalizeDurableEnvelope(env)}
	if err := s.delegate.ValidateRecord(record); err != nil {
		s.recordPublish(kind, "validation_error")
		return err
	}
	stored, err := s.store.Append(ctx, record)
	if err != nil {
		s.recordSpool(kind, "error")
		return err
	}
	s.recordSpool(kind, "ok")
	s.recordBacklog()
	if err := s.submit(stored); err != nil {
		// The durable commit is the device-facing acceptance boundary. Broker
		// submission errors are surfaced operationally and recovered by replay.
		s.reportError(err)
	}
	return nil
}

func normalizeDurableEnvelope(env envelope.FrameEnvelope) envelope.FrameEnvelope {
	if env.EventID == "" {
		env.EventID = env.StableEventID()
	}
	if env.ParseStatus == "" {
		env.ParseStatus = envelope.ParseOK
	}
	return env
}

func (s *DurableOutboxSink) submit(stored storedOutboxRecord) error {
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		s.store.Release(stored.Ref)
		return ErrDurableOutboxClosed
	}
	if _, exists := s.inflight[stored.Ref]; exists {
		s.mu.Unlock()
		return nil
	}
	s.inflight[stored.Ref] = struct{}{}
	s.pending.Add(1)
	inflight := len(s.inflight)
	s.mu.Unlock()
	s.recordInflight(inflight)

	err := s.delegate.PublishRecordAsync(stored.Record, func(publishErr error) {
		s.complete(stored, publishErr)
	})
	if err == nil {
		s.recordPublish(stored.Record.Kind, "submitted")
		return nil
	}
	s.mu.Lock()
	delete(s.inflight, stored.Ref)
	inflight = len(s.inflight)
	s.mu.Unlock()
	s.pending.Done()
	s.store.Release(stored.Ref)
	s.recordInflight(inflight)
	s.recordPublish(stored.Record.Kind, "submit_error")
	return fmt.Errorf("submit durable outbox record: %w", err)
}

func (s *DurableOutboxSink) complete(stored storedOutboxRecord, publishErr error) {
	defer s.pending.Done()
	status := "acked"
	if publishErr != nil {
		status = "ack_error"
		s.store.Release(stored.Ref)
	} else if err := s.store.Ack(stored.Ref); err != nil {
		publishErr = err
		status = "remove_error"
	}
	s.mu.Lock()
	delete(s.inflight, stored.Ref)
	inflight := len(s.inflight)
	s.mu.Unlock()
	s.recordInflight(inflight)
	s.recordBacklog()
	s.recordPublish(stored.Record.Kind, status)
	if publishErr != nil {
		s.reportError(publishErr)
	}
}

func (s *DurableOutboxSink) replay(ctx context.Context, limit int) error {
	if ctx != nil && ctx.Err() != nil {
		return ctx.Err()
	}
	records, err := s.store.ClaimPending(limit)
	if err != nil {
		return fmt.Errorf("claim durable outbox records: %w", err)
	}
	for index, stored := range records {
		if ctx != nil && ctx.Err() != nil {
			for _, remaining := range records[index:] {
				s.store.Release(remaining.Ref)
			}
			return ctx.Err()
		}
		if err := s.delegate.ValidateRecord(stored.Record); err != nil {
			s.store.Release(stored.Ref)
			for _, remaining := range records[index+1:] {
				s.store.Release(remaining.Ref)
			}
			s.recordPublish(stored.Record.Kind, "validation_error")
			return fmt.Errorf("validate durable outbox replay record: %w", err)
		}
		if err := s.submit(stored); err != nil && !errors.Is(err, ErrDurableOutboxClosed) {
			s.reportError(err)
		}
	}
	return nil
}

func (s *DurableOutboxSink) isClosed() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.closed
}

func (s *DurableOutboxSink) reportError(err error) {
	if err != nil && s.onError != nil {
		s.onError(err)
	}
}

func (s *DurableOutboxSink) recordSpool(kind string, status string) {
	if s.metrics == nil {
		return
	}
	s.metrics.IncCounter("vehicle_durable_spool_records_total", metrics.Labels{
		"name": s.name, "kind": kind, "status": status,
	})
}

func (s *DurableOutboxSink) recordPublish(kind string, status string) {
	if s.metrics == nil {
		return
	}
	s.metrics.IncCounter("vehicle_durable_outbox_publish_total", metrics.Labels{
		"name": s.name, "kind": kind, "status": status,
	})
}

func (s *DurableOutboxSink) recordInflight(value int) {
	if s.metrics == nil {
		return
	}
	s.metrics.SetGauge("vehicle_durable_outbox_inflight", metrics.Labels{"name": s.name}, float64(value))
}

func (s *DurableOutboxSink) recordBacklog() {
	if s.metrics == nil {
		return
	}
	count, oldest := s.store.Stats()
	labels := metrics.Labels{"name": s.name}
	// Keep the legacy gauge during migration because capacity gates already
	// consume it; its value now represents durable WAL records, not files.
	s.metrics.SetGauge("vehicle_durable_spool_backlog_files", labels, float64(count))
	s.metrics.SetGauge("vehicle_durable_outbox_backlog_records", labels, float64(count))
	ageSeconds := 0.0
	if count > 0 && !oldest.IsZero() {
		ageSeconds = time.Since(oldest).Seconds()
		if ageSeconds < 0 {
			ageSeconds = 0
		}
	}
	s.metrics.SetGauge("vehicle_durable_spool_oldest_age_seconds", labels, ageSeconds)
}