package scraper

import (
	"context"
	"fmt"
	"sync"
	"time"

	"sigs.k8s.io/controller-runtime/pkg/log"
)

type manager struct {
	interval time.Duration
	workers  []*scrapeWorker
}

type scrapeWorker struct {
	index   int
	scraper Scraper
	wake    chan struct{}
}

var _ Manager = (*manager)(nil)

// NewManager constructs a Manager for a fixed set of Scrapers.
func NewManager(cfg *ManagerConfig, scrapers ...Scraper) (Manager, error) {
	if cfg == nil {
		return nil, fmt.Errorf("scraper manager: config is required")
	}
	if cfg.Interval <= 0 {
		return nil, fmt.Errorf("scraper manager: interval must be greater than zero")
	}

	workers := make([]*scrapeWorker, 0, len(scrapers))
	for i, configured := range scrapers {
		if configured == nil {
			return nil, fmt.Errorf("scraper manager: scraper[%d] is required", i)
		}
		workers = append(workers, &scrapeWorker{
			index:   i,
			scraper: configured,
			wake:    make(chan struct{}, 1),
		})
	}
	return &manager{interval: cfg.Interval, workers: workers}, nil
}

// Start schedules every Scraper immediately and then at the configured
// interval. Each Scraper has an independent serial worker, so slow calls are
// coalesced without delaying other Scrapers.
func (m *manager) Start(ctx context.Context) error {
	var workers sync.WaitGroup
	for _, worker := range m.workers {
		workers.Add(1)
		go func() {
			defer workers.Done()
			worker.run(ctx)
		}()
	}
	m.schedule()

	ticker := time.NewTicker(m.interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			workers.Wait()
			return nil
		case <-ticker.C:
			m.schedule()
		}
	}
}

func (m *manager) schedule() {
	for _, worker := range m.workers {
		select {
		case worker.wake <- struct{}{}:
		default:
		}
	}
}

func (w *scrapeWorker) run(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case <-w.wake:
			if err := w.scraper.Scrape(ctx); err != nil && ctx.Err() == nil {
				log.FromContext(ctx).Error(err, "unable to scrape metrics", "scraper", w.index)
			}
		}
	}
}

// NeedLeaderElection ensures only the active controller manager runs the
// configured Scrapers.
func (*manager) NeedLeaderElection() bool {
	return true
}