package kafka

import (
	"github.com/Clymene-project/Clymene/cmd/promtail/app/api"
	"github.com/Clymene-project/Clymene/cmd/promtail/app/scrapeconfig"
	"github.com/Clymene-project/Clymene/cmd/promtail/app/targets/target"
	"github.com/prometheus/client_golang/prometheus"
	"go.uber.org/zap"
)

// TargetManager manages a series of kafka targets.
type TargetManager struct {
	logger        *zap.Logger
	targetSyncers map[string]*TargetSyncer
}

// NewTargetManager creates a new Kafka managers.
func NewTargetManager(
	reg prometheus.Registerer,
	logger *zap.Logger,
	pushClient api.EntryHandler,
	scrapeConfigs []scrapeconfig.Config,
) (*TargetManager, error) {
	tm := &TargetManager{
		logger:        logger,
		targetSyncers: make(map[string]*TargetSyncer),
	}
	for _, cfg := range scrapeConfigs {
		t, err := NewSyncer(reg, logger, cfg, pushClient)
		if err != nil {
			return nil, err
		}
		tm.targetSyncers[cfg.JobName] = t
	}

	return tm, nil
}

// Ready returns true if at least one Kafka target is active.
func (tm *TargetManager) Ready() bool {
	for _, t := range tm.targetSyncers {
		if len(t.getActiveTargets()) > 0 {
			return true
		}
	}
	return false
}

func (tm *TargetManager) Stop() {
	for _, t := range tm.targetSyncers {
		if err := t.Stop(); err != nil {
			t.logger.Error("error stopping kafka target", zap.Error(err))
		}
	}
}

func (tm *TargetManager) ActiveTargets() map[string][]target.Target {
	result := make(map[string][]target.Target, len(tm.targetSyncers))
	for k, v := range tm.targetSyncers {
		result[k] = v.getActiveTargets()
	}
	return result
}

func (tm *TargetManager) AllTargets() map[string][]target.Target {
	result := make(map[string][]target.Target, len(tm.targetSyncers))
	for k, v := range tm.targetSyncers {
		result[k] = append(v.getActiveTargets(), v.getDroppedTargets()...)
	}
	return result
}
