package runtimeworker

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

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

type auditOutboxController interface {
	Prepare(context.Context) error
	Run(context.Context) error
	Snapshot() audit.OutboxWorkerSnapshot
}

type auditOutboxObservationSource interface {
	ObserveQueue(context.Context) (audit.OutboxQueueState, error)
}

type auditOutboxObservationFactory func() (auditOutboxObservationSource, func() error, error)

// auditOutboxRuntimeWorker keeps durable Audit processing and exact queue
// diagnostics on separate control and connection paths. Observation state is
// exported only as metrics and can never authorize or suppress business work.
type auditOutboxRuntimeWorker struct {
	controller         auditOutboxController
	observationFactory auditOutboxObservationFactory

	observationMu      sync.Mutex
	observationSource  auditOutboxObservationSource
	releaseObservation func() error

	queueDepth atomic.Int64
	poison     atomic.Int64
	oldestAge  atomic.Int64

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

func newAuditOutboxRuntimeWorker(
	controller auditOutboxController,
	observationFactory auditOutboxObservationFactory,
) (*auditOutboxRuntimeWorker, error) {
	if controller == nil || observationFactory == nil {
		return nil, errors.New("runtime audit outbox worker: controller and observation source are required")
	}
	worker := &auditOutboxRuntimeWorker{controller: controller, observationFactory: observationFactory}
	worker.observationStale.Store(true)
	return worker, nil
}

func (w *auditOutboxRuntimeWorker) Prepare(ctx context.Context) error {
	if w == nil || w.controller == nil {
		return errors.New("runtime audit outbox worker: controller is unavailable")
	}
	return w.controller.Prepare(ctx)
}

func (w *auditOutboxRuntimeWorker) Run(ctx context.Context) error {
	if w == nil || w.controller == nil {
		return errors.New("runtime audit outbox worker: controller is unavailable")
	}
	return w.runWithObservationTiming(ctx, metadataObservationInterval, metadataObservationTimeout)
}

func (w *auditOutboxRuntimeWorker) runWithObservationTiming(ctx context.Context, interval, timeout time.Duration) error {
	stopObservation := w.startObservationLoopWithTiming(ctx, interval, timeout)
	defer stopObservation()
	return w.controller.Run(ctx)
}

func (w *auditOutboxRuntimeWorker) 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 *auditOutboxRuntimeWorker) 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 audit outbox observation failed")
		return
	}
	completed := time.Now()
	w.queueDepth.Store(state.Depth)
	w.poison.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 *auditOutboxRuntimeWorker) observe(ctx context.Context) (audit.OutboxQueueState, error) {
	if w == nil || w.observationFactory == nil {
		return audit.OutboxQueueState{}, errors.New("runtime audit outbox worker: observation source is unavailable")
	}
	w.observationMu.Lock()
	defer w.observationMu.Unlock()
	if w.observationSource == nil {
		source, release, err := w.observationFactory()
		if err != nil {
			return audit.OutboxQueueState{}, fmt.Errorf("runtime audit outbox worker: open observation source: %w", err)
		}
		if source == nil || release == nil {
			if release != nil {
				_ = release()
			}
			return audit.OutboxQueueState{}, errors.New("runtime audit outbox worker: observation source is incomplete")
		}
		w.observationSource = source
		w.releaseObservation = release
	}
	return w.observationSource.ObserveQueue(ctx)
}

func (w *auditOutboxRuntimeWorker) 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 newAuditOutboxObservationFactory(
	openIndependentDB func(database.PoolOptions) (*sql.DB, error),
	ownership persistence.RuntimeWorkOwnership,
	trackers ...*workerAuxiliaryPoolTracker,
) auditOutboxObservationFactory {
	return func() (auditOutboxObservationSource, 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 := audit.ResolveFencedOutboxQueueObserver(observationDB, ownership)
		if err != nil {
			_ = release()
			return nil, nil, err
		}
		return observer, release, nil
	}
}

func (w *auditOutboxRuntimeWorker) snapshot() workerMetricsSnapshot {
	if w == nil || w.controller == nil {
		return workerMetricsSnapshot{}
	}
	business := w.controller.Snapshot()
	snapshot := workerMetricsSnapshot{
		QueueDepth: w.queueDepth.Load(), PoisonDepth: w.poison.Load(), OldestAge: time.Duration(w.oldestAge.Load()),
		InFlight: business.InFlight, Attempts: business.Attempts, Failures: business.Failures, Processed: business.Processed,
		ObservationEnabled:  true,
		ObservationDuration: time.Duration(w.observationDurationNanos.Load()),
		ObservationFailures: w.observationFailures.Load(),
		ObservationStale:    w.observationStale.Load(),
		WorkLoop: workLoopSnapshot{
			PollChecks: business.PollChecks, ConfirmedIdle: business.ConfirmedIdle,
			ClaimAttempts: business.ClaimAttempts, EmptyClaims: business.EmptyClaims,
			Claimed: business.Claimed, PollDuration: business.PollDuration,
			ClaimDuration: business.ClaimDuration,
		},
	}
	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
}