package case_driver

import (
	"context"
	"strings"

	. "github.com/tidbcloud/dataflow-service-e2e/internal/config"
	"github.com/tidbcloud/dataflow-service-e2e/internal/log"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
)

type kafkaDependency struct {
	lgr log.SimpleLogger
}

func NewKafkaDependency(lgr log.SimpleLogger) Dependency {
	return &kafkaDependency{
		lgr: lgr,
	}
}

func (k kafkaDependency) Type() string {
	return KAFKA
}

func (k kafkaDependency) GetExistingUsableEntities() ([]DependencyEntity, error) {
	var entities []DependencyEntity
	for _, it := range GlobalConfig.Kafka {
		it.CreatedByDriver = false
		projectIDs := strings.Split(it.ProjectID, ",")
		if len(projectIDs) <= 1 {
			entities = append(entities, it)
			continue
		}
		// expand one kafka entry into multiple entities, one per project ID
		for _, projectID := range projectIDs {
			projectID = strings.TrimSpace(projectID)
			if projectID == "" {
				continue
			}
			copied := it.Copy()
			copied.ProjectID = projectID
			entities = append(entities, copied)
		}
	}
	return entities, nil
}

func (k kafkaDependency) ProvisionEntities(_ context.Context, requirementStats EntitiesStats) (<-chan DependencyEntity, error) {
	// TODO implement me
	panic("implement me")
}

func (k kafkaDependency) Delete(entities ...DependencyEntity) error {
	// TODO implement me
	panic("implement me")
}