package outpost

import (
	"context"
	"errors"
	"fmt"
	"math/rand/v2"
	"sync"
	"time"
)

const (
	defaultBatchSize       = 16
	defaultWorkers         = 4
	defaultPollInterval    = time.Second
	defaultDeliveryTimeout = 30 * time.Second
	defaultMaxAttempts     = 8

	// storeOpTimeout bounds the Complete and Fail writes. They deliberately
	// ignore the caller's cancellation so a delivery that already succeeded is
	// still recorded during shutdown.
	storeOpTimeout = 5 * time.Second

	// maxDerivedLease rejects configurations whose derived lease would be
	// unreasonable, nearly always a batch size or worker mistake, and bounds
	// the multiplication in NewRelay.
	maxDerivedLease = 30 * time.Minute
)

// Relay claims messages from a Store and hands them to a Producer.
type Relay struct {
	store    Store
	producer Producer
	hooks    Hooks

	batch       int
	workers     int
	timeout     time.Duration
	maxAttempts int
	backoff     Backoff
	lease       time.Duration
}

// RelayOption configures a Relay.
type RelayOption func(*Relay)

// WithBatchSize sets how many messages one claim asks for. Default 16.
func WithBatchSize(n int) RelayOption { return func(r *Relay) { r.batch = n } }

// WithWorkers sets how many deliveries run concurrently. Default 4.
func WithWorkers(n int) RelayOption { return func(r *Relay) { r.workers = n } }

// WithDeliveryTimeout bounds a single delivery. Default 30s.
func WithDeliveryTimeout(d time.Duration) RelayOption { return func(r *Relay) { r.timeout = d } }

// WithMaxAttempts sets how many deliveries a message gets before it is
// abandoned. Default 8.
func WithMaxAttempts(n int) RelayOption { return func(r *Relay) { r.maxAttempts = n } }

// WithBackoff sets the retry delay policy. It is a lower bound, not an exact
// schedule: the relay retries no earlier than the computed delay and spreads it
// by up to 50% so independent workers do not retry in lockstep. Pass a
// deterministic backoff; the relay applies the spread. The default is
// ExponentialBackoff(1s, 10m).
func WithBackoff(b Backoff) RelayOption { return func(r *Relay) { r.backoff = b } }

// WithHooks sets the observability hooks.
func WithHooks(h Hooks) RelayOption { return func(r *Relay) { r.hooks = h } }

// NewRelay validates the configuration and derives the lease duration:
// (ceil(batch/workers)+1) * deliveryTimeout, so a claimed batch cannot outlive
// its lease even when every delivery runs to its deadline. The arithmetic is
// done in int64 and bounded by maxDerivedLease before it is multiplied, so an
// absurd batch size fails loudly instead of wrapping into a tiny lease.
func NewRelay(store Store, producer Producer, opts ...RelayOption) (*Relay, error) {
	if store == nil {
		return nil, errors.New("outpost: nil store")
	}
	if producer == nil {
		return nil, errors.New("outpost: nil producer")
	}
	r := &Relay{
		store:       store,
		producer:    producer,
		batch:       defaultBatchSize,
		workers:     defaultWorkers,
		timeout:     defaultDeliveryTimeout,
		maxAttempts: defaultMaxAttempts,
		backoff:     ExponentialBackoff(time.Second, 10*time.Minute),
	}
	for _, opt := range opts {
		opt(r)
	}
	switch {
	case r.batch < 1:
		return nil, fmt.Errorf("outpost: batch size must be >= 1, got %d", r.batch)
	case r.workers < 1:
		return nil, fmt.Errorf("outpost: workers must be >= 1, got %d", r.workers)
	case r.timeout <= 0:
		return nil, fmt.Errorf("outpost: delivery timeout must be > 0, got %s", r.timeout)
	case r.maxAttempts < 1:
		return nil, fmt.Errorf("outpost: max attempts must be >= 1, got %d", r.maxAttempts)
	case r.backoff == nil:
		return nil, errors.New("outpost: nil backoff")
	}
	batches := int64(r.batch) / int64(r.workers)
	if int64(r.batch)%int64(r.workers) != 0 {
		batches++
	}
	if limit := int64(maxDerivedLease / r.timeout); batches >= limit {
		return nil, fmt.Errorf("outpost: derived lease would exceed %s; lower the batch size or raise the worker count", maxDerivedLease)
	}
	r.lease = time.Duration(batches+1) * r.timeout
	return r, nil
}

// Run polls until ctx is done, one batch at a time. Store errors are reported
// through Hooks.OnError and retried after a pause; Run does not return until
// ctx is cancelled.
func (r *Relay) Run(ctx context.Context) {
	for ctx.Err() == nil {
		n, err := r.cycle(ctx)
		if err != nil {
			r.hooks.err(err)
			if !sleep(ctx, defaultPollInterval) {
				return
			}
			continue
		}
		if n == 0 && !sleep(ctx, defaultPollInterval) {
			return
		}
	}
}

// RunOnce performs a single claim-and-process cycle and reports how many
// messages were claimed. Loop it to drain, or call it from a cron job.
func (r *Relay) RunOnce(ctx context.Context) (int, error) {
	return r.cycle(ctx)
}

// cycle claims one batch, delivers it, then records the outcomes.
func (r *Relay) cycle(ctx context.Context) (int, error) {
	claims, err := r.store.Claim(ctx, r.batch, r.lease)
	if err != nil {
		return 0, err
	}
	if len(claims) == 0 {
		return 0, nil
	}

	var (
		mu       sync.Mutex
		done     []ClaimRef
		failures []failure
		wg       sync.WaitGroup
	)
	sem := make(chan struct{}, r.workers)
	for _, c := range claims {
		if ctx.Err() != nil {
			// Shutting down: remaining leases expire and are redelivered.
			break
		}
		sem <- struct{}{}
		wg.Add(1)
		go func() {
			defer wg.Done()
			defer func() { <-sem }()
			ref, f := r.deliver(ctx, c)
			switch {
			case ref != nil:
				mu.Lock()
				done = append(done, *ref)
				mu.Unlock()
			case f != nil:
				mu.Lock()
				failures = append(failures, *f)
				mu.Unlock()
			}
		}()
	}
	wg.Wait()

	if len(done) > 0 {
		r.record(ctx, func(ctx context.Context) error { return r.store.Complete(ctx, done...) })
	}
	for _, f := range failures {
		r.record(ctx, func(ctx context.Context) error { return r.store.Fail(ctx, f.ref, f.cause, f.retryAt) })
	}
	return len(claims), nil
}

type failure struct {
	ref     ClaimRef
	cause   error
	retryAt *time.Time
}

// deliver runs one delivery. It returns the ref on success, a failure when one
// must be recorded, and neither while the relay is shutting down.
func (r *Relay) deliver(ctx context.Context, c Claim) (*ClaimRef, *failure) {
	info := Info{ID: c.ID, Name: c.Name, Key: c.Key, Attempt: c.Attempts + 1, Headers: c.Headers}

	dctx, cancel := context.WithTimeout(ctx, r.timeout)
	defer cancel()
	dctx = context.WithValue(dctx, infoKey{}, info)

	err := r.producer.Produce(dctx, c.Message)
	switch {
	case err == nil:
		ref := ClaimRef{ID: c.ID, Token: c.Token}
		return &ref, nil
	case ctx.Err() != nil:
		return nil, nil
	}

	retryAt, abandon := r.verdict(c, err)
	if abandon {
		r.hooks.deadLetter(info, err)
	} else {
		r.hooks.retry(info, err, *retryAt)
	}
	ref := ClaimRef{ID: c.ID, Token: c.Token}
	return nil, &failure{ref: ref, cause: err, retryAt: retryAt}
}

// verdict maps a producer error to a retry time, or abandons the message.
// Precedence: Permanent, then exhausted attempts, then RetryAfter, then the
// configured backoff.
func (r *Relay) verdict(c Claim, err error) (*time.Time, bool) {
	if isPermanent(err) || c.Attempts+1 >= r.maxAttempts {
		return nil, true
	}
	delay := r.backoff(c.Attempts + 1)
	if d, ok := retryDelay(err); ok {
		delay = d
	}
	t := time.Now().Add(spread(delay))
	return &t, false
}

// spread returns d plus up to half of it, so independent workers do not retry
// at the same instant. It never returns less than d: a configured backoff and a
// provider's Retry-After are both floors.
//
// A half is the largest fraction that keeps attempt windows disjoint under a
// doubling backoff, which is why it is a constant and not a setting.
//
// math/rand/v2's top-level functions are safe for concurrent use, so this needs
// no state or mutex even though every delivery goroutine calls it.
func spread(d time.Duration) time.Duration {
	if d <= 0 {
		return d
	}
	const maxDuration = time.Duration(1<<63 - 1)
	if d > maxDuration-d/2 {
		// Guard the addition: an absurd delay would wrap into the past.
		return d
	}
	return d + time.Duration(rand.Int64N(int64(d/2)+1))
}

// record writes an outcome to the store, ignoring cancellation and stale
// claims.
func (r *Relay) record(ctx context.Context, fn func(context.Context) error) {
	wctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), storeOpTimeout)
	defer cancel()
	if err := fn(wctx); err != nil && !errors.Is(err, ErrStaleClaim) {
		r.hooks.err(err)
	}
}

func sleep(ctx context.Context, d time.Duration) bool {
	if d <= 0 {
		return ctx.Err() == nil
	}
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}