package runtimeworker

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

	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/files"
	"github.com/nexuscrm/persistence"
	"github.com/rs/zerolog/log"
)

type fileUploadLifecycleStore interface {
	CheckReady(context.Context) error
	HasExpired(context.Context) (bool, error)
	ProcessExpired(context.Context, int, time.Duration) (int64, error)
}

type fileUploadLifecycleObservationSource interface {
	Observe(context.Context) (files.RuntimeUploadLifecycleState, error)
}

type fileUploadLifecycleObservationFactory func() (fileUploadLifecycleObservationSource, func() error, error)

type fileUploadLifecycleWorker struct {
	store     fileUploadLifecycleStore
	batchSize int
	poll      time.Duration
	lease     time.Duration
	jitter    time.Duration

	observationFactory fileUploadLifecycleObservationFactory
	observationMu      sync.Mutex
	observationSource  fileUploadLifecycleObservationSource
	releaseObservation func() error

	queueDepth  atomic.Int64
	poisonDepth 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
}

func newFileUploadLifecycleWorker(
	store fileUploadLifecycleStore,
	batchSize int,
	poll, lease, jitter time.Duration,
	observationFactory fileUploadLifecycleObservationFactory,
) (*fileUploadLifecycleWorker, error) {
	if store == nil || batchSize <= 0 || poll <= 0 || lease <= 0 || jitter < 0 || observationFactory == nil {
		return nil, errors.New("runtime file upload lifecycle worker: configuration is incomplete")
	}
	worker := &fileUploadLifecycleWorker{
		store: store, batchSize: batchSize, poll: poll, lease: lease, jitter: jitter,
		observationFactory: observationFactory,
	}
	worker.observationStale.Store(true)
	return worker, nil
}

func (w *fileUploadLifecycleWorker) Prepare(ctx context.Context) error {
	if w == nil || w.store == nil {
		return errors.New("runtime file upload lifecycle worker: store is unavailable")
	}
	return w.store.CheckReady(ctx)
}

func (w *fileUploadLifecycleWorker) Run(ctx context.Context) error {
	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 file upload lifecycle initial pass failed")
		waitRetryJitter(ctx, w.jitter)
	}
	stopObservation := w.startObservationLoop(ctx)
	defer stopObservation()
	ticker := time.NewTicker(w.poll)
	defer ticker.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 file upload lifecycle pass failed")
				waitRetryJitter(ctx, w.jitter)
			}
		}
	}
}

func (w *fileUploadLifecycleWorker) runPass(ctx context.Context) error {
	hintStarted := time.Now()
	hasExpired, hintErr := w.store.HasExpired(ctx)
	w.workLoop.observePoll(hintStarted, hasExpired, hintErr)
	if hintErr == nil && !hasExpired {
		return nil
	}
	w.attempts.Add(1)
	w.inFlight.Add(1)
	defer w.inFlight.Add(-1)
	claimStarted := time.Now()
	processed, err := w.store.ProcessExpired(ctx, w.batchSize, w.lease)
	claimed := uint64(0)
	if processed > 0 {
		claimed = uint64(processed)
	}
	w.workLoop.observeClaim(claimStarted, claimed, err == nil && processed == 0)
	if processed > 0 {
		w.processed.Add(uint64(processed))
	}
	if err != nil {
		w.failures.Add(1)
		return err
	}
	return nil
}

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

func (w *fileUploadLifecycleWorker) startObservationLoopWithTiming(ctx context.Context, interval, timeout time.Duration) func() {
	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 *fileUploadLifecycleWorker) refreshObservationWithTimeout(ctx context.Context, timeout time.Duration) {
	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 {
		if ctx.Err() != nil {
			return
		}
		w.observationFailures.Add(1)
		w.observationStale.Store(true)
		log.Warn().Err(err).Msg("runtime file upload lifecycle observation failed")
		return
	}
	completed := time.Now()
	w.queueDepth.Store(state.QueueDepth)
	w.poisonDepth.Store(state.PoisonDepth)
	w.oldestAge.Store(int64(state.OldestAge))
	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 *fileUploadLifecycleWorker) observe(ctx context.Context) (files.RuntimeUploadLifecycleState, error) {
	if w == nil || w.observationFactory == nil {
		return files.RuntimeUploadLifecycleState{}, errors.New("runtime file upload lifecycle: observation source is unavailable")
	}
	w.observationMu.Lock()
	defer w.observationMu.Unlock()
	if w.observationSource == nil {
		source, release, err := w.observationFactory()
		if err != nil {
			return files.RuntimeUploadLifecycleState{}, fmt.Errorf("runtime file upload lifecycle: open observation source: %w", err)
		}
		if source == nil || release == nil {
			if release != nil {
				_ = release()
			}
			return files.RuntimeUploadLifecycleState{}, errors.New("runtime file upload lifecycle: observation source is incomplete")
		}
		w.observationSource = source
		w.releaseObservation = release
	}
	return w.observationSource.Observe(ctx)
}

func (w *fileUploadLifecycleWorker) closeObservation() error {
	if w == nil {
		return nil
	}
	w.observationMu.Lock()
	defer w.observationMu.Unlock()
	release := w.releaseObservation
	w.observationSource = nil
	w.releaseObservation = nil
	if release == nil {
		return nil
	}
	return release()
}

func newFileUploadLifecycleObservationFactory(
	openIndependentDB func(database.PoolOptions) (*sql.DB, error),
	trackers ...*workerAuxiliaryPoolTracker,
) fileUploadLifecycleObservationFactory {
	return func() (fileUploadLifecycleObservationSource, func() error, error) {
		if openIndependentDB == nil {
			return nil, nil, errors.New("independent database opener is unavailable")
		}
		observationDB, release, err := openWorkerObservationPool(
			openIndependentDB, optionalWorkerAuxiliaryPoolTracker(trackers),
		)
		if err != nil {
			return nil, nil, err
		}
		observer, err := files.NewRuntimeUploadLifecycleObserver(observationDB)
		if err != nil {
			_ = release()
			return nil, nil, err
		}
		return observer, release, nil
	}
}

func (w *fileUploadLifecycleWorker) snapshot() workerMetricsSnapshot {
	if w == nil {
		return workerMetricsSnapshot{}
	}
	snapshot := workerMetricsSnapshot{
		QueueDepth: w.queueDepth.Load(), PoisonDepth: w.poisonDepth.Load(),
		OldestAge: time.Duration(w.oldestAge.Load()), InFlight: w.inFlight.Load(),
		Attempts: w.attempts.Load(), Failures: w.failures.Load(), Processed: w.processed.Load(),
		WorkLoop:            w.workLoop.snapshot(),
		ObservationEnabled:  true,
		ObservationDuration: time.Duration(w.observationDurationNanos.Load()),
		ObservationFailures: w.observationFailures.Load(),
		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
}