package eventbus

import (
	"context"
	"time"

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

type RetryConfig struct {
	Attempts       int
	Backoff        time.Duration
	AttemptTimeout time.Duration
}

type RetryingSink struct {
	delegate Sink
	cfg      RetryConfig
}

func NewRetryingSink(delegate Sink, cfg RetryConfig) *RetryingSink {
	if delegate == nil {
		panic("retry delegate sink must not be nil")
	}
	if cfg.Attempts <= 0 {
		cfg.Attempts = 1
	}
	return &RetryingSink{delegate: delegate, cfg: cfg}
}

func (s *RetryingSink) PublishRaw(ctx context.Context, env envelope.FrameEnvelope) error {
	return s.withRetry(ctx, func(attemptCtx context.Context) error {
		return s.delegate.PublishRaw(attemptCtx, env)
	})
}

func (s *RetryingSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelope) error {
	return s.withRetry(ctx, func(attemptCtx context.Context) error {
		return s.delegate.PublishUnified(attemptCtx, env)
	})
}

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

func (s *RetryingSink) withRetry(ctx context.Context, publish func(context.Context) error) error {
	var lastErr error
	for attempt := 1; attempt <= s.cfg.Attempts; attempt++ {
		attemptCtx := ctx
		cancel := func() {}
		if s.cfg.AttemptTimeout > 0 {
			attemptCtx, cancel = context.WithTimeout(ctx, s.cfg.AttemptTimeout)
		}
		err := publish(attemptCtx)
		cancel()
		if err == nil {
			return nil
		}
		lastErr = err
		if ctx.Err() != nil {
			return ctx.Err()
		}
		if attempt == s.cfg.Attempts {
			break
		}
		if s.cfg.Backoff > 0 {
			timer := time.NewTimer(s.cfg.Backoff)
			select {
			case <-ctx.Done():
				if !timer.Stop() {
					<-timer.C
				}
				return ctx.Err()
			case <-timer.C:
			}
		}
	}
	return lastErr
}