package operator

import (
	"context"
	"sync"

	"github.com/prometheus/client_golang/prometheus"

	"github.com/grafana/grafana-app-sdk/health"
	"github.com/grafana/grafana-app-sdk/metrics"
)

var (
	_ Informer         = &ConcurrentInformer{}
	_ health.Checker   = &ConcurrentInformer{}
	_ metrics.Provider = &ConcurrentInformer{}
)

// ConcurrentInformer implements the Informer interface, wrapping another Informer implementation
// to provide concurrent handling of events.
// Events will still be emitted sequentially, but the event handler methods on added ResourceWatchers
// (ie the business logic) will be processed by concurrent workers. Events for an object will be assigned
// to the same worker to preserve the per-object in-order guarantee provided by K8s client tooling.
type ConcurrentInformer struct {
	errorHandler func(context.Context, error)

	informer             Informer
	watchers             []*concurrentWatcher
	maxConcurrentWorkers uint64
	queueDepth           prometheus.GaugeFunc

	mtx sync.RWMutex
}

// ConcurrentInformerOptions are options for the ConcurrentInformer.
type ConcurrentInformerOptions struct {
	// ErrorHandler is a user-specified error handling function. If left nil, DefaultErrorHandler will be used.
	ErrorHandler func(context.Context, error)
	// MaxConcurrentWorkers is a limit on the number of workers to run concurrently for each ResourceWatcher. Each
	// worker maintains a queue of events which are processed sequentially. Events for a particular object are assigned
	// to the same worker, as to maintain the guarantee of in-order delivery of events per object.
	// By default, a single worker is run to process all events sequentially.
	MaxConcurrentWorkers uint64
	// MetricsConfig controls the namespace for Prometheus metrics exposed by the ConcurrentInformer.
	MetricsConfig metrics.Config
	// ResourceKind is used as the "kind" label on queue depth metrics.
	// When empty, the label value defaults to an empty string.
	ResourceKind string
}

// NewConcurrentInformer creates a new ConcurrentInformer wrapping the provided Informer.
//
// Deprecated: Use NewConcurrentInformerFromOptions instead, which accepts InformerOptions.
func NewConcurrentInformer(inf Informer, opts ConcurrentInformerOptions) (
	*ConcurrentInformer, error) {
	ci := &ConcurrentInformer{
		errorHandler:         DefaultErrorHandler,
		informer:             inf,
		watchers:             make([]*concurrentWatcher, 0),
		maxConcurrentWorkers: 10,
	}
	if opts.ErrorHandler != nil {
		ci.errorHandler = opts.ErrorHandler
	}
	if opts.MaxConcurrentWorkers > 0 {
		ci.maxConcurrentWorkers = opts.MaxConcurrentWorkers
	}
	ci.queueDepth = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
		Namespace:   opts.MetricsConfig.Namespace,
		Subsystem:   "concurrent_watcher",
		Name:        "queue_depth",
		Help:        "Current number of pending events across all concurrent watcher worker queues.",
		ConstLabels: prometheus.Labels{"kind": opts.ResourceKind},
	}, ci.sumQueueDepth)

	return ci, nil
}

// NewConcurrentInformerFromOptions creates a new ConcurrentInformer wrapping the provided Informer,
// using InformerOptions for configuration.
func NewConcurrentInformerFromOptions(inf Informer, opts InformerOptions) (
	*ConcurrentInformer, error) {
	ci := &ConcurrentInformer{
		errorHandler:         DefaultErrorHandler,
		informer:             inf,
		watchers:             make([]*concurrentWatcher, 0),
		maxConcurrentWorkers: 10,
	}
	if opts.ErrorHandler != nil {
		ci.errorHandler = opts.ErrorHandler
	}
	if opts.MaxConcurrentWorkers > 0 {
		ci.maxConcurrentWorkers = opts.MaxConcurrentWorkers
	}
	ci.queueDepth = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
		Namespace:   opts.MetricsConfig.Namespace,
		Subsystem:   "concurrent_watcher",
		Name:        "queue_depth",
		Help:        "Current number of pending events across all concurrent watcher worker queues.",
		ConstLabels: prometheus.Labels{"kind": opts.ResourceKind},
	}, ci.sumQueueDepth)

	return ci, nil
}

// AddEventHandler adds a ResourceWatcher as an event handler for watch events from the informer.
// The ResourceWatcher is wrapped before adding it to the underlying Informer, to allow concurrent
// handling of the events.
// Event handlers are not guaranteed to be executed in parallel or in any particular order by the underlying
// Informer. If you want to coordinate between ResourceWatchers, use an InformerController.
// nolint:dupl
func (ci *ConcurrentInformer) AddEventHandler(handler ResourceWatcher) error {
	cw, err := newConcurrentWatcher(handler, ci.maxConcurrentWorkers, ci.errorHandler)
	if err != nil {
		return err
	}

	{
		ci.mtx.Lock()
		ci.watchers = append(ci.watchers, cw)
		ci.mtx.Unlock()
	}

	return ci.informer.AddEventHandler(cw)
}

// Run starts the informer and blocks until stopCh receives a message
func (ci *ConcurrentInformer) Run(ctx context.Context) error {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	ci.mtx.RLock()
	for _, cw := range ci.watchers {
		go cw.Run(ctx)
	}
	ci.mtx.RUnlock()

	return ci.informer.Run(ctx)
}

func (ci *ConcurrentInformer) HealthChecks() []health.Check {
	checks := make([]health.Check, 0, 2)
	if cast, ok := ci.informer.(health.Check); ok {
		checks = append(checks, cast)
	}
	if cast, ok := ci.informer.(health.Checker); ok {
		checks = append(checks, cast.HealthChecks()...)
	}
	return checks
}

// WaitForSync waits for the informer to sync.
// If the sync is not complete within the context deadline, it will return a timeout error.
func (ci *ConcurrentInformer) WaitForSync(ctx context.Context) error {
	return ci.informer.WaitForSync(ctx)
}

// PrometheusCollectors implements metrics.Provider.
func (ci *ConcurrentInformer) PrometheusCollectors() []prometheus.Collector {
	return []prometheus.Collector{ci.queueDepth}
}

func (ci *ConcurrentInformer) sumQueueDepth() float64 {
	ci.mtx.RLock()
	defer ci.mtx.RUnlock()
	var total int64
	for _, cw := range ci.watchers {
		for _, q := range cw.workers {
			total += q.Len()
		}
	}
	return float64(total)
}
