package runtimeworker

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

	"github.com/nexuscrm/persistence"
	"github.com/rs/zerolog/log"
)

type metadataWorkState struct {
	queueDepth int64
	poison     int64
	oldestAge  time.Duration
}

// metadataWorkProcessor is the role-specific queue/effect owner. ProcessOne
// returns the number of durable rows finalized and whether another immediately
// eligible unit was observed.
type metadataWorkProcessor interface {
	CheckReady(context.Context) error
	Observe(context.Context) (metadataWorkState, error)
	Maintain(context.Context, int) error
	ProcessOne(context.Context, int) (processed int, found bool, err error)
}

// metadataWorkHinter separates the high-frequency negative admission hint
// from the lower-frequency queue-health aggregate. Implementations must treat
// the hint as advisory; ProcessOne remains the fenced authority.
type metadataWorkHinter interface {
	HasWork(context.Context) (bool, error)
}

// metadataClaimedWork is an opaque durable lease whose database claim has
// already committed. The coordinator starts it only in an immediately free
// effect slot; the role-specific processor retains ownership of effect and
// retry semantics.
type metadataClaimedWork struct {
	run func(context.Context) error
}

// metadataWorkBatchClaimer opts a parallel work family into single-owner,
// capacity-aware admission. ClaimBatch must return no more than limit durable
// leases and must not start their effects before returning.
type metadataWorkBatchClaimer interface {
	ClaimBatch(context.Context, int) ([]metadataClaimedWork, error)
}

type metadataWorkDiagnostics struct {
	admissionFenceAttempts uint64
	admissionFenceDuration time.Duration
}

type metadataWorkDiagnosticsProvider interface {
	WorkDiagnostics() metadataWorkDiagnostics
}

type metadataObservationCloser interface {
	closeObservation() error
}

type metadataWakeSource uint8

const (
	metadataWakeSourceNone metadataWakeSource = iota
	metadataWakeSourcePoll
	metadataWakeSourceSuccessor
)

const (
	metadataObservationInterval = time.Hour
	metadataObservationTimeout  = 30 * time.Second
)

type metadataWorker struct {
	processor   metadataWorkProcessor
	batchSize   int
	poll        time.Duration
	jitter      time.Duration
	maintenance time.Duration
	parallelism int

	observationEnabled bool

	queueDepth atomic.Int64
	poison     atomic.Int64
	oldestAge  atomic.Int64
	inFlight   atomic.Int64
	attempts   atomic.Uint64
	failures   atomic.Uint64
	processed  atomic.Uint64
	workLoop   workLoopObservation

	observationLastSuccessUnixNano atomic.Int64
	observationDurationNanos       atomic.Int64
	observationFailures            atomic.Uint64
	observationStale               atomic.Bool
	oldestObservedUnixNano         atomic.Int64

	pollActivations             atomic.Uint64
	successorActivations        atomic.Uint64
	pollWakeCoalesced           atomic.Uint64
	successorWakeCoalesced      atomic.Uint64
	pollFirstEmptyAttempts      atomic.Uint64
	successorFirstEmptyAttempts atomic.Uint64
	continuationAttempts        atomic.Uint64
	continuationEmptyAttempts   atomic.Uint64
}

func newMetadataWorker(processor metadataWorkProcessor, batchSize int, poll, jitter, maintenance time.Duration, parallelism ...int) (*metadataWorker, error) {
	if processor == nil || batchSize <= 0 || poll <= 0 || jitter < 0 || maintenance < 0 {
		return nil, errors.New("runtime metadata worker: configuration is incomplete")
	}
	lanes := 1
	if len(parallelism) > 0 {
		lanes = parallelism[0]
	}
	if lanes <= 0 {
		return nil, errors.New("runtime metadata worker: parallelism must be positive")
	}
	worker := &metadataWorker{
		processor: processor, batchSize: batchSize, poll: poll, jitter: jitter,
		maintenance: maintenance, parallelism: lanes,
	}
	if _, ok := processor.(metadataWorkHinter); ok {
		worker.observationEnabled = true
		worker.observationStale.Store(true)
	}
	return worker, nil
}

func (w *metadataWorker) Prepare(ctx context.Context) error {
	if w == nil || w.processor == nil {
		return errors.New("runtime metadata worker: processor is unavailable")
	}
	return w.processor.CheckReady(ctx)
}

func (w *metadataWorker) Run(ctx context.Context) error {
	if w != nil && w.parallelism > 1 {
		if claimer, ok := w.processor.(metadataWorkBatchClaimer); ok {
			return w.runCoordinatedParallel(ctx, claimer)
		}
		return w.runParallel(ctx)
	}
	return w.runSerial(ctx)
}

func (w *metadataWorker) runSerial(ctx context.Context) error {
	if err := w.runMaintenance(ctx); errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		return err
	} else if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
		log.Warn().Err(err).Msg("runtime metadata worker initial maintenance failed")
		waitRetryJitter(ctx, w.jitter)
	}
	if err := w.runPass(ctx); errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		return err
	} else if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
		log.Warn().Err(err).Msg("runtime metadata worker initial pass failed")
		waitRetryJitter(ctx, w.jitter)
	}
	stopObservation := w.startObservationLoop(ctx)
	defer stopObservation()
	ticker := time.NewTicker(w.poll)
	defer ticker.Stop()
	var maintenanceTicker *time.Ticker
	var maintenance <-chan time.Time
	if w.maintenance > 0 {
		maintenanceTicker = time.NewTicker(w.maintenance)
		maintenance = maintenanceTicker.C
		defer maintenanceTicker.Stop()
	}
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			if err := w.runPass(ctx); errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
				return err
			} else if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
				log.Warn().Err(err).Msg("runtime metadata worker pass failed")
				waitRetryJitter(ctx, w.jitter)
			}
		case <-maintenance:
			if err := w.runMaintenance(ctx); errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
				return err
			} else if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
				log.Warn().Err(err).Msg("runtime metadata worker maintenance failed")
				waitRetryJitter(ctx, w.jitter)
			}
		}
	}
}

// runCoordinatedParallel gives one goroutine exclusive ownership of durable
// admission while effect goroutines consume already committed leases. It never
// claims beyond immediately free effect capacity. Completion signals are
// coalesced into one refill, so a successful effect no longer both retains a
// claimant and clones another independent claimant at a burst boundary.
func (w *metadataWorker) runCoordinatedParallel(ctx context.Context, claimer metadataWorkBatchClaimer) error {
	if err := w.runMaintenance(ctx); errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		return err
	} else if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
		log.Warn().Err(err).Msg("runtime metadata worker initial maintenance failed")
	}

	workCtx, cancel := context.WithCancel(ctx)
	completionCh := make(chan error, w.parallelism)
	var effects sync.WaitGroup
	active := 0
	stop := func() {
		cancel()
		effects.Wait()
		for active > 0 {
			<-completionCh
			active--
			w.inFlight.Add(-1)
		}
	}
	defer stop()

	wakeCh := make(chan metadataWakeSource, 1)
	wakeOne := func(source metadataWakeSource) {
		select {
		case wakeCh <- source:
		default:
			switch source {
			case metadataWakeSourcePoll:
				w.pollWakeCoalesced.Add(1)
			case metadataWakeSourceSuccessor:
				w.successorWakeCoalesced.Add(1)
			}
		}
	}

	refill := func(source metadataWakeSource) error {
		if active >= w.parallelism {
			return nil
		}
		pending, hintErr := w.hasWork(workCtx)
		if hintErr == nil && !pending {
			return nil
		}
		switch source {
		case metadataWakeSourcePoll:
			w.pollActivations.Add(1)
		case metadataWakeSourceSuccessor:
			w.successorActivations.Add(1)
		}

		free := w.parallelism - active
		claimStarted := time.Now()
		claimed, err := claimer.ClaimBatch(workCtx, free)
		w.workLoop.observeClaim(claimStarted, uint64(len(claimed)), err == nil && len(claimed) == 0)
		if err != nil {
			return err
		}
		if len(claimed) > free {
			return fmt.Errorf("runtime metadata worker: batch claimer returned %d leases for %d free slots", len(claimed), free)
		}
		if len(claimed) == 0 {
			switch source {
			case metadataWakeSourcePoll:
				w.pollFirstEmptyAttempts.Add(1)
			case metadataWakeSourceSuccessor:
				w.successorFirstEmptyAttempts.Add(1)
			}
			return nil
		}
		for _, work := range claimed {
			if work.run == nil {
				return errors.New("runtime metadata worker: batch claimer returned an incomplete durable lease")
			}
		}
		for _, work := range claimed {
			active++
			w.inFlight.Add(1)
			w.attempts.Add(1)
			effects.Add(1)
			go func(work metadataClaimedWork) {
				defer effects.Done()
				completionCh <- work.run(workCtx)
			}(work)
		}
		return nil
	}

	stopObservation := w.startObservationLoop(workCtx)
	defer stopObservation()
	pollTicker := time.NewTicker(w.poll)
	defer pollTicker.Stop()
	var maintenanceTicker *time.Ticker
	var maintenance <-chan time.Time
	if w.maintenance > 0 {
		maintenanceTicker = time.NewTicker(w.maintenance)
		maintenance = maintenanceTicker.C
		defer maintenanceTicker.Stop()
	}
	wakeOne(metadataWakeSourcePoll)

	complete := func(err error) error {
		active--
		w.inFlight.Add(-1)
		if err != nil {
			if workCtx.Err() != nil && (errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)) {
				return nil
			}
			w.failures.Add(1)
			return err
		}
		w.processed.Add(1)
		return nil
	}

	for {
		select {
		case <-ctx.Done():
			return nil
		case source := <-wakeCh:
			if err := refill(source); err != nil {
				w.failures.Add(1)
				return err
			}
		case err := <-completionCh:
			if err := complete(err); err != nil {
				return err
			}
		drainCompletions:
			for {
				select {
				case err := <-completionCh:
					if err := complete(err); err != nil {
						return err
					}
				default:
					break drainCompletions
				}
			}
			wakeOne(metadataWakeSourceSuccessor)
		case <-pollTicker.C:
			wakeOne(metadataWakeSourcePoll)
		case <-maintenance:
			if err := w.runMaintenance(workCtx); errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
				return err
			} else if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
				log.Warn().Err(err).Msg("runtime metadata worker maintenance failed")
			}
		}
	}
}

// runParallel owns long-lived independent pull lanes. One bounded scout wakes
// per poll while the queue is idle; the first successful effect fans out the
// drain lanes. Parallel capacity therefore scales with real backlog without
// multiplying empty remote-database claims and aggregate observations by the
// configured lane count. There is no batch-wide join barrier: a lane blocked
// inside one effect does not stop peers from draining later eligible keys. The
// only join is lifecycle shutdown.
func (w *metadataWorker) runParallel(ctx context.Context) error {
	if err := w.runMaintenance(ctx); errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		return err
	} else if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
		log.Warn().Err(err).Msg("runtime metadata worker initial maintenance failed")
	}
	lanesCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	errCh := make(chan error, w.parallelism)
	// A single buffered successor token makes parallelism grow from completed
	// work instead of broadcasting speculative claims to every configured lane.
	// Waiting lanes consume tokens immediately; when every lane is busy, at most
	// one future wake remains queued.
	wakeCh := make(chan metadataWakeSource, 1)
	wakeOne := func(source metadataWakeSource) {
		select {
		case wakeCh <- source:
		default:
			switch source {
			case metadataWakeSourcePoll:
				w.pollWakeCoalesced.Add(1)
			case metadataWakeSourceSuccessor:
				w.successorWakeCoalesced.Add(1)
			}
		}
	}
	var lanes sync.WaitGroup
	for lane := 0; lane < w.parallelism; lane++ {
		lanes.Add(1)
		go func() {
			defer lanes.Done()
			for {
				select {
				case <-lanesCtx.Done():
					return
				case source := <-wakeCh:
					switch source {
					case metadataWakeSourcePoll:
						w.pollActivations.Add(1)
					case metadataWakeSourceSuccessor:
						w.successorActivations.Add(1)
					}
					if err := w.runDrainPassFrom(lanesCtx, func() {
						wakeOne(metadataWakeSourceSuccessor)
					}, source); err != nil {
						if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
							errCh <- err
						}
						return
					}
				}
			}
		}()
	}
	waitLanes := func() { cancel(); lanes.Wait() }
	pollTicker := time.NewTicker(w.poll)
	defer pollTicker.Stop()
	pollPending := func() {
		pending, err := w.hasWork(lanesCtx)
		// Observation is a negative hint, never claim authority. A failed
		// observation must retain the existing fenced claim path. A successful
		// zero may only defer a concurrently arriving row until the next bounded
		// poll; every positive hint is rechecked by ProcessOne under its existing
		// ownership and row fences.
		if err != nil || pending {
			wakeOne(metadataWakeSourcePoll)
		}
	}
	stopObservation := w.startObservationLoop(lanesCtx)
	defer stopObservation()
	pollPending()

	var maintenanceTicker *time.Ticker
	var maintenance <-chan time.Time
	if w.maintenance > 0 {
		maintenanceTicker = time.NewTicker(w.maintenance)
		maintenance = maintenanceTicker.C
		defer maintenanceTicker.Stop()
	}
	for {
		select {
		case <-ctx.Done():
			waitLanes()
			return nil
		case err := <-errCh:
			waitLanes()
			return err
		case <-pollTicker.C:
			pollPending()
		case <-maintenance:
			if err := w.runMaintenance(lanesCtx); errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
				waitLanes()
				return err
			} else if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
				log.Warn().Err(err).Msg("runtime metadata worker maintenance failed")
			}
		}
	}
}

func (w *metadataWorker) runMaintenance(ctx context.Context) error {
	if w == nil || w.maintenance == 0 {
		return nil
	}
	w.inFlight.Add(1)
	defer w.inFlight.Add(-1)
	if err := w.processor.Maintain(ctx, w.batchSize); err != nil {
		w.failures.Add(1)
		return err
	}
	return nil
}

func (w *metadataWorker) runPass(ctx context.Context) error {
	pending, err := w.hasWork(ctx)
	if err == nil && !pending {
		return nil
	}
	return w.runDrainPass(ctx, nil)
}

func (w *metadataWorker) hasWork(ctx context.Context) (bool, error) {
	started := time.Now()
	var pending bool
	var err error
	if hinter, ok := w.processor.(metadataWorkHinter); ok {
		pending, err = hinter.HasWork(ctx)
	} else {
		var state metadataWorkState
		state, err = w.observe(ctx)
		pending = state.queueDepth > 0
	}
	w.workLoop.observePoll(started, pending, err)
	return pending, err
}

func (w *metadataWorker) startObservationLoop(ctx context.Context) func() {
	return w.startObservationLoopWithTiming(ctx, metadataObservationInterval, metadataObservationTimeout)
}

func (w *metadataWorker) startObservationLoopWithTiming(ctx context.Context, interval, timeout time.Duration) func() {
	if w == nil || !w.observationEnabled {
		return func() {}
	}
	// The exact queue-health aggregate is diagnostics only. In particular, do
	// not sample during startup: a deployment or restart must not put the RU
	// spike back onto the serving cutover window.
	observationCtx, cancel := context.WithCancel(ctx)
	done := make(chan struct{})
	go func() {
		defer close(done)
		defer w.closeObservation()
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-observationCtx.Done():
				return
			case <-ticker.C:
				w.refreshObservationWithTimeout(observationCtx, timeout)
			}
		}
	}()
	return func() {
		cancel()
		<-done
	}
}

func (w *metadataWorker) closeObservation() {
	closer, ok := w.processor.(metadataObservationCloser)
	if !ok {
		return
	}
	if err := closer.closeObservation(); err != nil {
		log.Warn().Err(err).Msg("runtime metadata worker observation close failed")
	}
}

func (w *metadataWorker) refreshObservationWithTimeout(ctx context.Context, timeout time.Duration) {
	if w == nil || !w.observationEnabled {
		return
	}
	started := time.Now()
	observationCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	state, err := w.observe(observationCtx)
	w.observationDurationNanos.Store(int64(time.Since(started)))
	if err != nil {
		// Process shutdown is not a sampling failure. A sampler-owned timeout or
		// any other observation error is fail-open: retain the last good queue
		// snapshot and mark only its diagnostic freshness stale.
		if ctx.Err() != nil {
			return
		}
		w.observationFailures.Add(1)
		w.observationStale.Store(true)
		log.Warn().Err(err).Msg("runtime metadata worker observation failed")
		return
	}
	completed := time.Now()
	w.observationLastSuccessUnixNano.Store(completed.UnixNano())
	w.observationStale.Store(false)
	if state.oldestAge > 0 {
		w.oldestObservedUnixNano.Store(completed.Add(-state.oldestAge).UnixNano())
	} else {
		w.oldestObservedUnixNano.Store(0)
	}
}

func (w *metadataWorker) observe(ctx context.Context) (metadataWorkState, error) {
	state, err := w.processor.Observe(ctx)
	if err != nil {
		return metadataWorkState{}, err
	}
	w.queueDepth.Store(state.queueDepth)
	w.poison.Store(state.poison)
	w.oldestAge.Store(int64(state.oldestAge))
	return state, nil
}

func (w *metadataWorker) runDrainPass(ctx context.Context, backlogObserved func()) error {
	return w.runDrainPassFrom(ctx, backlogObserved, metadataWakeSourceNone)
}

func (w *metadataWorker) runDrainPassFrom(ctx context.Context, backlogObserved func(), source metadataWakeSource) error {
	w.inFlight.Add(1)
	defer w.inFlight.Add(-1)
	attemptInPass := 0
	for remaining := w.batchSize; remaining > 0; {
		attemptInPass++
		if attemptInPass > 1 {
			w.continuationAttempts.Add(1)
		}
		w.attempts.Add(1)
		claimStarted := time.Now()
		processed, found, err := w.processor.ProcessOne(ctx, remaining)
		claimed := uint64(0)
		if found {
			claimed = 1
		}
		w.workLoop.observeClaim(claimStarted, claimed, err == nil && !found)
		if processed > 0 {
			w.processed.Add(uint64(processed))
			remaining -= processed
			if found && backlogObserved != nil {
				backlogObserved()
			}
		}
		if err != nil {
			w.failures.Add(1)
			return err
		}
		if !found || processed <= 0 {
			if attemptInPass > 1 {
				w.continuationEmptyAttempts.Add(1)
			} else {
				switch source {
				case metadataWakeSourcePoll:
					w.pollFirstEmptyAttempts.Add(1)
				case metadataWakeSourceSuccessor:
					w.successorFirstEmptyAttempts.Add(1)
				}
			}
			return nil
		}
	}
	return nil
}

func (w *metadataWorker) snapshot() workerMetricsSnapshot {
	if w == nil {
		return workerMetricsSnapshot{}
	}
	snapshot := workerMetricsSnapshot{
		QueueDepth: w.queueDepth.Load(), PoisonDepth: w.poison.Load(), OldestAge: time.Duration(w.oldestAge.Load()),
		InFlight: w.inFlight.Load(), Attempts: w.attempts.Load(), Failures: w.failures.Load(), Processed: w.processed.Load(),
		PollActivations: w.pollActivations.Load(), SuccessorActivations: w.successorActivations.Load(),
		PollWakeCoalesced: w.pollWakeCoalesced.Load(), SuccessorWakeCoalesced: w.successorWakeCoalesced.Load(),
		PollFirstEmptyAttempts:      w.pollFirstEmptyAttempts.Load(),
		SuccessorFirstEmptyAttempts: w.successorFirstEmptyAttempts.Load(),
		ContinuationAttempts:        w.continuationAttempts.Load(),
		ContinuationEmptyAttempts:   w.continuationEmptyAttempts.Load(),
		WorkLoop:                    w.workLoop.snapshot(),
	}
	if provider, ok := w.processor.(metadataWorkDiagnosticsProvider); ok {
		diagnostics := provider.WorkDiagnostics()
		snapshot.AdmissionFenceAttempts = diagnostics.admissionFenceAttempts
		snapshot.AdmissionFenceDuration = diagnostics.admissionFenceDuration
	}
	if w.observationEnabled {
		snapshot.ObservationEnabled = true
		snapshot.ObservationDuration = time.Duration(w.observationDurationNanos.Load())
		snapshot.ObservationFailures = w.observationFailures.Load()
		snapshot.ObservationStale = w.observationStale.Load()
		if value := w.observationLastSuccessUnixNano.Load(); value > 0 {
			snapshot.ObservationLastSuccess = time.Unix(0, value)
		}
		if value := w.oldestObservedUnixNano.Load(); value > 0 {
			snapshot.OldestObservedAt = time.Unix(0, value)
		}
	}
	return snapshot
}