package worker

import (
	"context"
	"errors"
	"fmt"
	"sync/atomic"
	"time"
)

var (
	ErrAlreadyRunning     = errors.New("worker: controller is already running")
	ErrShutdownTimeout    = errors.New("worker: attempt shutdown timed out")
	ErrControllerUnusable = errors.New("worker: controller is unusable after an unknown durable outcome or incomplete shutdown")
	// ErrSourceOperationNotCommitted may be wrapped by a mutating Source method
	// only when the source can prove that no durable transition was committed.
	// Every other mutating-source error is outcome-unknown and disables reuse.
	ErrSourceOperationNotCommitted = errors.New("worker: source operation did not commit a durable transition")
)

// Attempt is one already-claimed durable unit. Run owns the complete domain
// effect lifecycle, including any heartbeat and durable disposition. A nil
// return means the source has durably handled the attempt outcome; a non-nil
// return is process-fatal because the controller cannot safely infer retry.
type Attempt struct {
	ID  string
	Run func(context.Context) error
}

// ClaimBatch contains at most the capacity requested from Source. More asks
// the controller for an immediate bounded continuation after capacity becomes
// available. A source must never report More with an empty batch because that
// would create an unproductive hot loop.
type ClaimBatch struct {
	Attempts []Attempt
	More     bool
}

// Source owns durable work identity, claim and lease transitions, effect
// execution, disposition, and exact scheduling facts. Every method must honor
// context cancellation; the controller cannot safely detach a blocked durable
// state transition during shutdown. Reconcile and Claim may mutate durable
// state, so their errors are outcome-unknown unless they wrap
// ErrSourceOperationNotCommitted after proving that no durable transition
// committed. CheckReady and NextWorkAt are read-only.
type Source interface {
	CheckReady(context.Context) error
	Reconcile(context.Context) error
	Claim(context.Context, int) (ClaimBatch, error)
	NextWorkAt(context.Context) (time.Time, bool, error)
}

// GenerationProbe is an optional cheap cross-process change detector. A
// changed generation requests source reconciliation; the generation is only a
// wake cursor and never work authority.
type GenerationProbe interface {
	ReadGeneration(context.Context, uint64) (uint64, error)
}

// Config declares process-local scheduling and shutdown bounds. Domain retry,
// lease, heartbeat, and fairness policy do not belong here.
type Config struct {
	Parallelism        int
	ClaimBatchSize     int
	ReconcileInterval  time.Duration
	GenerationInterval time.Duration
	ShutdownTimeout    time.Duration
}

func (c Config) validate(probe GenerationProbe) error {
	if c.Parallelism <= 0 || c.ClaimBatchSize <= 0 || c.ReconcileInterval <= 0 || c.ShutdownTimeout <= 0 {
		return errors.New("worker: controller configuration is incomplete")
	}
	if probe == nil && c.GenerationInterval != 0 {
		return errors.New("worker: generation interval requires a generation probe")
	}
	if probe != nil && c.GenerationInterval <= 0 {
		return errors.New("worker: generation probe requires a positive interval")
	}
	return nil
}

// Snapshot is process-local operational evidence. Domain queue depth, oldest
// age, poison, lease expiry, and uncertain outcomes remain domain observations.
type Snapshot struct {
	Ready                       bool
	Unusable                    bool
	InFlight                    int64
	ReadinessChecks             uint64
	WakeRequests                uint64
	WakeCoalesced               uint64
	ReconcileRequests           uint64
	ReconcileCoalesced          uint64
	Reconciliations             uint64
	GenerationProbes            uint64
	GenerationChanges           uint64
	ClaimCalls                  uint64
	EmptyClaims                 uint64
	NextWorkLookups             uint64
	OverdueNextWorkSuppressions uint64
	Claimed                     uint64
	Completed                   uint64
	AttemptFailures             uint64
	NextWorkScheduled           bool
	NextWorkUnixNano            int64
}

// Controller supplies process-local execution opportunities while Source and
// Attempt retain all durable authority.
type Controller struct {
	source Source
	probe  GenerationProbe
	config Config

	wakeCh      chan struct{}
	reconcileCh chan struct{}
	running     atomic.Bool
	unusable    atomic.Bool
	ready       atomic.Bool
	inFlight    atomic.Int64

	readinessChecks             atomic.Uint64
	wakeRequests                atomic.Uint64
	wakeCoalesced               atomic.Uint64
	reconcileRequests           atomic.Uint64
	reconcileCoalesced          atomic.Uint64
	reconciliations             atomic.Uint64
	generationProbes            atomic.Uint64
	generationChanges           atomic.Uint64
	claimCalls                  atomic.Uint64
	emptyClaims                 atomic.Uint64
	nextWorkLookups             atomic.Uint64
	overdueNextWorkSuppressions atomic.Uint64
	claimed                     atomic.Uint64
	completed                   atomic.Uint64
	attemptFailures             atomic.Uint64
	nextWorkScheduled           atomic.Bool
	nextWorkUnixNano            atomic.Int64
}

// NewController constructs one process-local controller for a domain Source.
func NewController(source Source, probe GenerationProbe, config Config) (*Controller, error) {
	if source == nil {
		return nil, errors.New("worker: source is required")
	}
	if err := config.validate(probe); err != nil {
		return nil, err
	}
	return &Controller{
		source: source, probe: probe, config: config,
		wakeCh: make(chan struct{}, 1), reconcileCh: make(chan struct{}, 1),
	}, nil
}

// Wake coalesces a process-local hint that durable work may now be claimable.
// Claim remains the authority, and the periodic reconciliation backstop means
// loss of a wake cannot permanently strand work.
func (c *Controller) Wake() {
	if c == nil {
		return
	}
	c.wakeRequests.Add(1)
	select {
	case c.wakeCh <- struct{}{}:
	default:
		c.wakeCoalesced.Add(1)
	}
}

// RequestReconcile coalesces a hint that derived state or durable acceptance
// materialization should be reconciled before another claim pass.
func (c *Controller) RequestReconcile() {
	if c == nil {
		return
	}
	c.reconcileRequests.Add(1)
	select {
	case c.reconcileCh <- struct{}{}:
	default:
		c.reconcileCoalesced.Add(1)
	}
}

// Ready reports whether startup readiness and reconciliation have completed.
func (c *Controller) Ready() bool { return c != nil && c.ready.Load() }

// Snapshot returns process-local operational evidence without querying domain
// state.
func (c *Controller) Snapshot() Snapshot {
	if c == nil {
		return Snapshot{}
	}
	return Snapshot{
		Ready: c.ready.Load(), Unusable: c.unusable.Load(), InFlight: c.inFlight.Load(),
		ReadinessChecks: c.readinessChecks.Load(),
		WakeRequests:    c.wakeRequests.Load(), WakeCoalesced: c.wakeCoalesced.Load(),
		ReconcileRequests: c.reconcileRequests.Load(), ReconcileCoalesced: c.reconcileCoalesced.Load(),
		Reconciliations: c.reconciliations.Load(), GenerationProbes: c.generationProbes.Load(),
		GenerationChanges: c.generationChanges.Load(), ClaimCalls: c.claimCalls.Load(),
		EmptyClaims: c.emptyClaims.Load(), NextWorkLookups: c.nextWorkLookups.Load(),
		OverdueNextWorkSuppressions: c.overdueNextWorkSuppressions.Load(),
		Claimed:                     c.claimed.Load(), Completed: c.completed.Load(),
		AttemptFailures: c.attemptFailures.Load(), NextWorkScheduled: c.nextWorkScheduled.Load(),
		NextWorkUnixNano: c.nextWorkUnixNano.Load(),
	}
}

type attemptResult struct{ err error }

// Run starts the controller and blocks until cancellation or a fatal error.
// An unknown mutating-source outcome, attempt error, or shutdown timeout makes
// this Controller permanently unusable; callers must fail the process before
// unresolved durable work can overlap a replacement.
func (c *Controller) Run(ctx context.Context) error {
	if c == nil || c.source == nil {
		return errors.New("worker: controller is unavailable")
	}
	if ctx == nil {
		return errors.New("worker: context is required")
	}
	if c.unusable.Load() {
		return ErrControllerUnusable
	}
	if !c.running.CompareAndSwap(false, true) {
		return ErrAlreadyRunning
	}
	defer c.running.Store(false)
	if ctx.Err() != nil {
		return nil
	}

	err := c.source.CheckReady(ctx)
	c.readinessChecks.Add(1)
	if err != nil {
		if ctx.Err() != nil {
			return nil
		}
		return fmt.Errorf("worker: source readiness: %w", err)
	}
	generation, err := c.readInitialGeneration(ctx)
	if err != nil {
		if ctx.Err() != nil {
			return nil
		}
		return err
	}
	if err := c.reconcile(ctx); err != nil {
		if canceledWithoutCommit(ctx, err) {
			return nil
		}
		return err
	}

	workCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	results := make(chan attemptResult, c.config.Parallelism)
	active := 0
	more := true

	reconcileTicker := time.NewTicker(c.config.ReconcileInterval)
	defer reconcileTicker.Stop()
	var generationTicker *time.Ticker
	var generationTick <-chan time.Time
	if c.probe != nil {
		generationTicker = time.NewTicker(c.config.GenerationInterval)
		generationTick = generationTicker.C
		defer generationTicker.Stop()
	}
	var dueTimer *time.Timer
	var due <-chan time.Time
	defer func() {
		if dueTimer != nil {
			dueTimer.Stop()
		}
		c.nextWorkScheduled.Store(false)
		c.nextWorkUnixNano.Store(0)
	}()

	c.ready.Store(true)
	defer c.ready.Store(false)

	for {
		for more && active < c.config.Parallelism {
			if ctx.Err() != nil {
				return c.stopAttempts(cancel, results, active, nil)
			}
			// A queued reconciliation request takes priority over every new
			// claim turn, including a More continuation and a wake selected
			// concurrently with reconciliation. A request arriving after this
			// check belongs to the following claim turn.
			select {
			case <-c.reconcileCh:
				if err := c.reconcile(workCtx); err != nil {
					return c.stopAfterReconcileError(ctx, cancel, results, active, err)
				}
				more = true
				continue
			default:
			}
			select {
			case <-reconcileTicker.C:
				if err := c.reconcile(workCtx); err != nil {
					return c.stopAfterReconcileError(ctx, cancel, results, active, err)
				}
				continue
			default:
			}
			select {
			case <-generationTick:
				current, probeErr := c.probe.ReadGeneration(workCtx, generation)
				c.generationProbes.Add(1)
				if probeErr != nil {
					if ctx.Err() != nil {
						return c.stopAttempts(cancel, results, active, nil)
					}
					return c.stopAttempts(cancel, results, active, fmt.Errorf("worker: generation probe: %w", probeErr))
				}
				if current != generation {
					generation = current
					c.generationChanges.Add(1)
					if err := c.reconcile(workCtx); err != nil {
						return c.stopAfterReconcileError(ctx, cancel, results, active, err)
					}
				}
				continue
			default:
			}
			free := c.config.Parallelism - active
			limit := c.config.ClaimBatchSize
			if limit > free {
				limit = free
			}
			batch, claimErr := c.source.Claim(workCtx, limit)
			c.claimCalls.Add(1)
			if claimErr != nil {
				knownUncommitted := c.classifyMutatingSourceError(claimErr)
				if knownUncommitted && len(batch.Attempts) == 0 {
					if ctx.Err() != nil {
						return c.stopAttempts(cancel, results, active, nil)
					}
					return c.stopAttempts(cancel, results, active, fmt.Errorf("worker: claim: %w", claimErr))
				}
				// A no-commit marker together with attempts is contradictory.
				// Unknown outcomes were already marked unusable by classification.
				if knownUncommitted {
					c.unusable.Store(true)
				}
				if ctx.Err() != nil {
					return c.stopAttempts(cancel, results, active, fmt.Errorf("worker: claim during shutdown: %w", claimErr))
				}
				return c.stopAttempts(cancel, results, active, fmt.Errorf("worker: claim: %w", claimErr))
			}
			if len(batch.Attempts) > limit {
				c.unusable.Store(true)
				return c.stopAttempts(cancel, results, active, fmt.Errorf("worker: source returned %d attempts for capacity %d", len(batch.Attempts), limit))
			}
			if len(batch.Attempts) == 0 {
				c.emptyClaims.Add(1)
				if batch.More {
					c.unusable.Store(true)
					return c.stopAttempts(cancel, results, active, errors.New("worker: source reported continuation without attempts"))
				}
				more = false
				if err := c.resetDueTimer(workCtx, &dueTimer, &due); err != nil {
					if ctx.Err() != nil {
						return c.stopAttempts(cancel, results, active, nil)
					}
					return c.stopAttempts(cancel, results, active, err)
				}
				break
			}
			for _, attempt := range batch.Attempts {
				if attempt.ID == "" || attempt.Run == nil {
					c.unusable.Store(true)
					return c.stopAttempts(cancel, results, active, errors.New("worker: source returned an incomplete attempt"))
				}
			}
			more = batch.More
			c.claimed.Add(uint64(len(batch.Attempts)))
			for _, attempt := range batch.Attempts {
				attempt := attempt
				active++
				c.inFlight.Add(1)
				go func() {
					results <- attemptResult{err: runAttempt(workCtx, attempt)}
				}()
			}
			if !more {
				if err := c.resetDueTimer(workCtx, &dueTimer, &due); err != nil {
					if ctx.Err() != nil {
						return c.stopAttempts(cancel, results, active, nil)
					}
					return c.stopAttempts(cancel, results, active, err)
				}
			}
			if active >= c.config.Parallelism {
				break
			}
		}

		select {
		case <-ctx.Done():
			return c.stopAttempts(cancel, results, active, nil)
		case result := <-results:
			active--
			c.inFlight.Add(-1)
			if result.err != nil {
				c.attemptFailures.Add(1)
				c.unusable.Store(true)
				if ctx.Err() != nil {
					return c.stopAttempts(cancel, results, active, fmt.Errorf("worker: attempt during shutdown: %w", result.err))
				}
				return c.stopAttempts(cancel, results, active, fmt.Errorf("worker: attempt: %w", result.err))
			}
			c.completed.Add(1)
		case <-c.wakeCh:
			more = true
		case <-c.reconcileCh:
			if err := c.reconcile(workCtx); err != nil {
				return c.stopAfterReconcileError(ctx, cancel, results, active, err)
			}
			more = true
		case <-due:
			due = nil
			dueTimer = nil
			c.nextWorkScheduled.Store(false)
			c.nextWorkUnixNano.Store(0)
			more = true
		case <-reconcileTicker.C:
			if err := c.reconcile(workCtx); err != nil {
				return c.stopAfterReconcileError(ctx, cancel, results, active, err)
			}
			more = true
		case <-generationTick:
			current, probeErr := c.probe.ReadGeneration(workCtx, generation)
			c.generationProbes.Add(1)
			if probeErr != nil {
				if ctx.Err() != nil {
					return c.stopAttempts(cancel, results, active, nil)
				}
				return c.stopAttempts(cancel, results, active, fmt.Errorf("worker: generation probe: %w", probeErr))
			}
			if current != generation {
				generation = current
				c.generationChanges.Add(1)
				if err := c.reconcile(workCtx); err != nil {
					return c.stopAfterReconcileError(ctx, cancel, results, active, err)
				}
				more = true
			}
		}
	}
}

func (c *Controller) readInitialGeneration(ctx context.Context) (uint64, error) {
	if c.probe == nil {
		return 0, nil
	}
	generation, err := c.probe.ReadGeneration(ctx, 0)
	c.generationProbes.Add(1)
	if err != nil {
		return 0, fmt.Errorf("worker: initial generation probe: %w", err)
	}
	return generation, nil
}

func (c *Controller) reconcile(ctx context.Context) error {
	c.reconciliations.Add(1)
	if err := c.source.Reconcile(ctx); err != nil {
		c.classifyMutatingSourceError(err)
		return fmt.Errorf("worker: reconcile: %w", err)
	}
	return nil
}

func sourceOperationDidNotCommit(err error) bool {
	return errors.Is(err, ErrSourceOperationNotCommitted)
}

func (c *Controller) classifyMutatingSourceError(err error) bool {
	if sourceOperationDidNotCommit(err) {
		return true
	}
	c.unusable.Store(true)
	return false
}

func canceledWithoutCommit(ctx context.Context, err error) bool {
	return ctx.Err() != nil && sourceOperationDidNotCommit(err)
}

func (c *Controller) stopAfterReconcileError(
	ctx context.Context,
	cancel context.CancelFunc,
	results <-chan attemptResult,
	active int,
	err error,
) error {
	if canceledWithoutCommit(ctx, err) {
		return c.stopAttempts(cancel, results, active, nil)
	}
	return c.stopAttempts(cancel, results, active, err)
}

func (c *Controller) resetDueTimer(ctx context.Context, timer **time.Timer, due *<-chan time.Time) error {
	if *timer != nil {
		if !(*timer).Stop() {
			select {
			case <-(*timer).C:
			default:
			}
		}
		*timer = nil
		*due = nil
	}
	next, found, err := c.source.NextWorkAt(ctx)
	c.nextWorkLookups.Add(1)
	if err != nil {
		return fmt.Errorf("worker: next work: %w", err)
	}
	if !found {
		c.nextWorkScheduled.Store(false)
		c.nextWorkUnixNano.Store(0)
		return nil
	}
	delay := time.Until(next)
	if delay <= 0 {
		// Claim is the immediate-work authority. If it just returned no
		// continuation, an overdue scheduling fact is stale, concurrently
		// consumed, or otherwise not actionable. Re-arming it at zero delay
		// would turn one stale fact into an unbounded empty-query loop. Leave
		// the timer disarmed; Wake, generation change, or reconciliation will
		// offer the next bounded claim turn.
		c.overdueNextWorkSuppressions.Add(1)
		c.nextWorkScheduled.Store(false)
		c.nextWorkUnixNano.Store(0)
		return nil
	}
	*timer = time.NewTimer(delay)
	*due = (*timer).C
	c.nextWorkUnixNano.Store(next.UnixNano())
	c.nextWorkScheduled.Store(true)
	return nil
}

func (c *Controller) stopAttempts(cancel context.CancelFunc, results <-chan attemptResult, active int, cause error) error {
	cancel()
	if active == 0 {
		return cause
	}
	var drainErr error
	timer := time.NewTimer(c.config.ShutdownTimeout)
	defer timer.Stop()
	for active > 0 {
		select {
		case result := <-results:
			active--
			c.inFlight.Add(-1)
			if result.err != nil {
				c.attemptFailures.Add(1)
				c.unusable.Store(true)
				drainErr = errors.Join(drainErr, fmt.Errorf("worker: attempt during shutdown: %w", result.err))
			} else {
				c.completed.Add(1)
			}
		case <-timer.C:
			c.unusable.Store(true)
			return errors.Join(cause, drainErr, ErrShutdownTimeout)
		}
	}
	return errors.Join(cause, drainErr)
}

func runAttempt(ctx context.Context, attempt Attempt) (err error) {
	defer func() {
		if recovered := recover(); recovered != nil {
			err = fmt.Errorf("attempt %q panicked: %v", attempt.ID, recovered)
		}
	}()
	return attempt.Run(ctx)
}