package runtimeworker

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

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

type permissionAuthoringCommandLifecycleStore interface {
	CheckReady(context.Context) error
	HasExpired(context.Context) (bool, error)
	CleanupExpired(context.Context, int) (persistence.PermissionAuthoringCommandCleanupResult, error)
}

type permissionAuthoringCommandLifecycleObservationSource interface {
	Observe(context.Context) (persistence.PermissionAuthoringCommandLifecycleState, error)
	ObservePermissionAuthoringConvergence(context.Context) (persistence.PermissionAuthoringConvergenceState, error)
}

type permissionAuthoringConvergenceBatchWorker interface {
	RunBatch(context.Context, string) (permissions.PermissionAuthoringConvergenceBatchResult, error)
}

type permissionAuthoringCommandLifecycleObservationFactory func() (
	permissionAuthoringCommandLifecycleObservationSource,
	func() error,
	error,
)

type permissionAuthoringCommandLifecycleWorker struct {
	store              permissionAuthoringCommandLifecycleStore
	convergence        permissionAuthoringConvergenceBatchWorker
	workerID           string
	batchSize          int
	poll               time.Duration
	jitter             time.Duration
	observationFactory permissionAuthoringCommandLifecycleObservationFactory

	observationMu      sync.Mutex
	observationSource  permissionAuthoringCommandLifecycleObservationSource
	releaseObservation func() error

	queueDepth 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 newPermissionAuthoringCommandLifecycleWorker(
	store permissionAuthoringCommandLifecycleStore,
	convergence permissionAuthoringConvergenceBatchWorker,
	workerID string,
	batchSize int,
	poll, jitter time.Duration,
	observationFactory permissionAuthoringCommandLifecycleObservationFactory,
) (*permissionAuthoringCommandLifecycleWorker, error) {
	if store == nil || convergence == nil || strings.TrimSpace(workerID) == "" ||
		batchSize <= 0 || poll <= 0 || jitter < 0 || observationFactory == nil {
		return nil, errors.New("permission authoring command lifecycle worker: configuration is incomplete")
	}
	worker := &permissionAuthoringCommandLifecycleWorker{
		store: store, convergence: convergence, workerID: workerID,
		batchSize: batchSize, poll: poll, jitter: jitter,
		observationFactory: observationFactory,
	}
	worker.observationStale.Store(true)
	return worker, nil
}

func (worker *permissionAuthoringCommandLifecycleWorker) Prepare(ctx context.Context) error {
	if worker == nil || worker.store == nil {
		return errors.New("permission authoring command lifecycle worker: store is unavailable")
	}
	return worker.store.CheckReady(ctx)
}

func (worker *permissionAuthoringCommandLifecycleWorker) Run(ctx context.Context) error {
	// Convergence recovery is deliberately first: a worker restart must inspect
	// pending committed aggregates before spending its first pass on journal
	// retention. The durable queue, not an API-process wake, drives this pass.
	if err := worker.runConvergencePass(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("permission authoring convergence initial pass failed")
		waitRetryJitter(ctx, worker.jitter)
	}
	if err := worker.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("permission authoring command lifecycle initial pass failed")
		waitRetryJitter(ctx, worker.jitter)
	}
	stopObservation := worker.startObservationLoop(ctx)
	defer stopObservation()
	ticker := time.NewTicker(worker.poll)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			if err := worker.runConvergencePass(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("permission authoring convergence pass failed")
				waitRetryJitter(ctx, worker.jitter)
			}
			if err := worker.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("permission authoring command lifecycle pass failed")
				waitRetryJitter(ctx, worker.jitter)
			}
		}
	}
}

func (worker *permissionAuthoringCommandLifecycleWorker) runConvergencePass(ctx context.Context) error {
	if worker == nil || worker.convergence == nil || strings.TrimSpace(worker.workerID) == "" {
		return errors.New("permission authoring convergence worker is unavailable")
	}
	worker.attempts.Add(1)
	worker.inFlight.Add(1)
	defer worker.inFlight.Add(-1)
	started := time.Now()
	result, err := worker.convergence.RunBatch(ctx, worker.workerID)
	worker.workLoop.observeClaim(started, uint64(max(result.Claimed, 0)), err == nil && result.Claimed == 0)
	if result.Finalized > 0 {
		worker.processed.Add(uint64(result.Finalized))
	}
	if err != nil {
		worker.failures.Add(1)
	}
	return err
}

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

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

func (worker *permissionAuthoringCommandLifecycleWorker) startObservationLoopWithTiming(
	ctx context.Context,
	interval, timeout time.Duration,
) func() {
	observationCtx, cancel := context.WithCancel(ctx)
	done := make(chan struct{})
	go func() {
		defer close(done)
		defer worker.closeObservation()
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-observationCtx.Done():
				return
			case <-ticker.C:
				worker.refreshObservationWithTimeout(observationCtx, timeout)
			}
		}
	}()
	return func() {
		cancel()
		<-done
	}
}

func (worker *permissionAuthoringCommandLifecycleWorker) refreshObservationWithTimeout(
	ctx context.Context,
	timeout time.Duration,
) {
	started := time.Now()
	observationCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	commands, convergence, err := worker.observe(observationCtx)
	worker.observationDurationNanos.Store(int64(time.Since(started)))
	if err != nil {
		if ctx.Err() != nil {
			return
		}
		worker.observationFailures.Add(1)
		worker.observationStale.Store(true)
		log.Warn().Err(err).Msg("permission authoring command lifecycle observation failed")
		return
	}
	completed := time.Now()
	worker.queueDepth.Store(commands.ExpiredCount() + convergence.PendingCount)
	oldestAge := max(commands.OldestAge, convergence.OldestAge)
	worker.oldestAge.Store(int64(oldestAge))
	worker.observationLastSuccessUnixNano.Store(completed.UnixNano())
	worker.observationStale.Store(false)
	if oldestAge > 0 {
		worker.oldestObservedUnixNano.Store(completed.Add(-oldestAge).UnixNano())
	} else {
		worker.oldestObservedUnixNano.Store(0)
	}
}

func (worker *permissionAuthoringCommandLifecycleWorker) observe(
	ctx context.Context,
) (persistence.PermissionAuthoringCommandLifecycleState, persistence.PermissionAuthoringConvergenceState, error) {
	if worker == nil || worker.observationFactory == nil {
		return persistence.PermissionAuthoringCommandLifecycleState{}, persistence.PermissionAuthoringConvergenceState{}, errors.New("permission authoring command lifecycle: observation source is unavailable")
	}
	worker.observationMu.Lock()
	defer worker.observationMu.Unlock()
	if worker.observationSource == nil {
		source, release, err := worker.observationFactory()
		if err != nil {
			return persistence.PermissionAuthoringCommandLifecycleState{}, persistence.PermissionAuthoringConvergenceState{}, fmt.Errorf("permission authoring command lifecycle: open observation source: %w", err)
		}
		if source == nil || release == nil {
			if release != nil {
				_ = release()
			}
			return persistence.PermissionAuthoringCommandLifecycleState{}, persistence.PermissionAuthoringConvergenceState{}, errors.New("permission authoring command lifecycle: observation source is incomplete")
		}
		worker.observationSource = source
		worker.releaseObservation = release
	}
	commands, err := worker.observationSource.Observe(ctx)
	if err != nil {
		return persistence.PermissionAuthoringCommandLifecycleState{}, persistence.PermissionAuthoringConvergenceState{}, err
	}
	convergence, err := worker.observationSource.ObservePermissionAuthoringConvergence(ctx)
	if err != nil {
		return persistence.PermissionAuthoringCommandLifecycleState{}, persistence.PermissionAuthoringConvergenceState{}, err
	}
	return commands, convergence, nil
}

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

func newPermissionAuthoringCommandLifecycleObservationFactory(
	openIndependentDB func(database.PoolOptions) (*sql.DB, error),
	ownership persistence.RuntimeWorkOwnership,
	trackers ...*workerAuxiliaryPoolTracker,
) permissionAuthoringCommandLifecycleObservationFactory {
	return func() (permissionAuthoringCommandLifecycleObservationSource, 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
		}
		store, err := persistence.NewPermissionAuthoringCommandLifecycleStore(observationDB, ownership)
		if err != nil {
			_ = release()
			return nil, nil, err
		}
		convergence, err := persistence.NewPermissionAuthoringConvergenceRepository(observationDB, nil)
		if err != nil {
			_ = release()
			return nil, nil, err
		}
		return permissionAuthoringLifecycleObservationSource{
			PermissionAuthoringCommandLifecycleStore: store,
			PermissionAuthoringConvergenceRepository: convergence,
		}, release, nil
	}
}

type permissionAuthoringLifecycleObservationSource struct {
	*persistence.PermissionAuthoringCommandLifecycleStore
	*persistence.PermissionAuthoringConvergenceRepository
}

func (worker *permissionAuthoringCommandLifecycleWorker) snapshot() workerMetricsSnapshot {
	if worker == nil {
		return workerMetricsSnapshot{}
	}
	snapshot := workerMetricsSnapshot{
		QueueDepth: worker.queueDepth.Load(), OldestAge: time.Duration(worker.oldestAge.Load()),
		InFlight: worker.inFlight.Load(), Attempts: worker.attempts.Load(), Failures: worker.failures.Load(), Processed: worker.processed.Load(),
		WorkLoop:            worker.workLoop.snapshot(),
		ObservationEnabled:  true,
		ObservationDuration: time.Duration(worker.observationDurationNanos.Load()),
		ObservationFailures: worker.observationFailures.Load(),
		ObservationStale:    worker.observationStale.Load(),
	}
	if value := worker.observationLastSuccessUnixNano.Load(); value > 0 {
		snapshot.ObservationLastSuccess = time.Unix(0, value)
	}
	if value := worker.oldestObservedUnixNano.Load(); value > 0 {
		snapshot.OldestObservedAt = time.Unix(0, value)
	}
	return snapshot
}