package consuming

import (
	"context"
	"fmt"

	"github.com/centrifugal/centrifugo/v6/internal/api"
	"github.com/centrifugal/centrifugo/v6/internal/configtypes"
	"github.com/centrifugal/centrifugo/v6/internal/metrics"
	"github.com/centrifugal/centrifugo/v6/internal/service"

	"github.com/rs/zerolog"
	"github.com/rs/zerolog/log"
)

type ConsumerConfig = configtypes.Consumer

type Dispatcher interface {
	DispatchCommand(ctx context.Context, method string, data []byte) error
	DispatchPublication(ctx context.Context, channels []string, pub api.ConsumedPublication) error
}

type consumerCommon struct {
	name   string
	log    zerolog.Logger
	nodeID string
}

func New(nodeID string, consumingHandler *api.ConsumingHandler, configs []ConsumerConfig) ([]service.Service, error) {
	dispatcher := api.NewDispatcher(consumingHandler)

	var services []service.Service
	for _, config := range configs {
		if !config.Enabled { // Important to keep this check inside specific type for proper config validation.
			log.Info().
				Str("consumer", config.Name).
				Str("type", config.Type).
				Msg("consumer is not enabled, skip")
			continue
		}
		common := &consumerCommon{
			name:   config.Name,
			log:    log.With().Str("consumer", config.Name).Logger(),
			nodeID: nodeID,
		}
		// Initialize consumer metrics with zero values.
		metrics.InitConsumerMetrics(config.Name)
		var consumer service.Service
		var err error
		switch config.Type {
		case configtypes.ConsumerTypePostgres:
			consumer, err = NewPostgresConsumer(config.Postgres, dispatcher, common)
		case configtypes.ConsumerTypeKafka:
			consumer, err = NewKafkaConsumer(config.Kafka, dispatcher, common)
		case configtypes.ConsumerTypeNatsJetStream:
			consumer, err = NewNatsJetStreamConsumer(config.NatsJetStream, dispatcher, common)
		case configtypes.ConsumerTypeRedisStream:
			consumer, err = NewRedisStreamConsumer(config.RedisStream, dispatcher, common)
		case configtypes.ConsumerTypeGooglePubSub:
			consumer, err = NewGooglePubSubConsumer(config.GooglePubSub, dispatcher, common)
		case configtypes.ConsumerTypeAwsSqs:
			consumer, err = NewAwsSqsConsumer(config.AwsSqs, dispatcher, common)
		case configtypes.ConsumerTypeAzureServiceBus:
			consumer, err = NewAzureServiceBusConsumer(config.AzureServiceBus, dispatcher, common)
		default:
			return nil, fmt.Errorf("unknown consumer type: %s", config.Type)
		}
		if err != nil {
			return nil, fmt.Errorf("error initializing %s consumer (%s): %w", config.Type, config.Name, err)
		}
		services = append(services, consumer)
		metrics.InitConsumerMetrics(config.Name)
		log.Info().
			Str("consumer", config.Name).
			Str("type", config.Type).
			Msg("running consumer")
	}
	return services, nil
}