package config

import (
	"fmt"

	"github.com/thomaspoignant/go-feature-flag/exporter/kafkaexporter"
	"github.com/xitongsys/parquet-go/parquet"
)

// ExporterConf contains all the field to configure an exporter
type ExporterConf struct {
	Kind                    ExporterKind           `mapstructure:"kind" koanf:"kind"`
	OutputDir               string                 `mapstructure:"outputDir" koanf:"outputdir"`
	Format                  string                 `mapstructure:"format" koanf:"format"`
	Filename                string                 `mapstructure:"filename" koanf:"filename"`
	CsvTemplate             string                 `mapstructure:"csvTemplate" koanf:"csvtemplate"`
	Bucket                  string                 `mapstructure:"bucket" koanf:"bucket"`
	Path                    string                 `mapstructure:"path" koanf:"path"`
	EndpointURL             string                 `mapstructure:"endpointUrl" koanf:"endpointurl"`
	Secret                  string                 `mapstructure:"secret" koanf:"secret"`
	Meta                    map[string]string      `mapstructure:"meta" koanf:"meta"`
	LogFormat               string                 `mapstructure:"logFormat" koanf:"logformat"`
	FlushInterval           int64                  `mapstructure:"flushInterval" koanf:"flushinterval"`
	MaxEventInMemory        int64                  `mapstructure:"maxEventInMemory" koanf:"maxeventinmemory"`
	ParquetCompressionCodec string                 `mapstructure:"parquetCompressionCodec" koanf:"parquetcompressioncodec"`
	Headers                 map[string][]string    `mapstructure:"headers" koanf:"headers"`
	QueueURL                string                 `mapstructure:"queueUrl" koanf:"queueurl"`
	Kafka                   kafkaexporter.Settings `mapstructure:"kafka" koanf:"kafka"`
	ProjectID               string                 `mapstructure:"projectID" koanf:"projectid"`
	Topic                   string                 `mapstructure:"topic" koanf:"topic"`
	StreamArn               string                 `mapstructure:"streamArn" koanf:"streamarn"`
	StreamName              string                 `mapstructure:"streamName" koanf:"streamname"`
	AccountName             string                 `mapstructure:"accountName" koanf:"accountname"`
	AccountKey              string                 `mapstructure:"accountKey" koanf:"accountkey"`
	Container               string                 `mapstructure:"container" koanf:"container"`
}

func (c *ExporterConf) IsValid() error {
	if err := c.Kind.IsValid(); err != nil {
		return err
	}
	if c.Kind == FileExporter && c.OutputDir == "" {
		return fmt.Errorf("invalid exporter: no \"outputDir\" property found for kind \"%s\"", c.Kind)
	}
	if (c.Kind == S3Exporter || c.Kind == GoogleStorageExporter) && c.Bucket == "" {
		return fmt.Errorf("invalid exporter: no \"bucket\" property found for kind \"%s\"", c.Kind)
	}
	if (c.Kind == KinesisExporter) && (c.StreamArn == "" && c.StreamName == "") {
		return fmt.Errorf("invalid exporter: no \"streamArn\" or \"streamName\" property found for kind \"%s\"", c.Kind)
	}
	if c.Kind == WebhookExporter && c.EndpointURL == "" {
		return fmt.Errorf("invalid exporter: no \"endpointUrl\" property found for kind \"%s\"", c.Kind)
	}
	if len(c.ParquetCompressionCodec) > 0 {
		if _, err := parquet.CompressionCodecFromString(c.ParquetCompressionCodec); err != nil {
			return fmt.Errorf("invalid exporter: \"parquetCompressionCodec\" err: %v", err)
		}
	}
	if c.Kind == SQSExporter && c.QueueURL == "" {
		return fmt.Errorf("invalid exporter: no \"queueUrl\" property found for kind \"%s\"", c.Kind)
	}

	if c.Kind == KafkaExporter && (c.Kafka.Topic == "" || len(c.Kafka.Addresses) == 0) {
		return fmt.Errorf("invalid exporter: \"kakfa.topic\" and \"kafka.addresses\" are required for kind \"%s\"", c.Kind)
	}

	if c.Kind == PubSubExporter && (c.ProjectID == "" || c.Topic == "") {
		return fmt.Errorf("invalid exporter: \"projectID\" and \"topic\" are required for kind \"%s\"", c.Kind)
	}

	if c.Kind == AzureExporter && c.Container == "" {
		return fmt.Errorf("invalid exporter: no \"container\" property found for kind \"%s\"", c.Kind)
	}
	if c.Kind == AzureExporter && c.AccountName == "" {
		return fmt.Errorf("invalid exporter: no \"accountName\" property found for kind \"%s\"", c.Kind)
	}

	return nil
}

type ExporterKind string

const (
	FileExporter          ExporterKind = "file"
	WebhookExporter       ExporterKind = "webhook"
	LogExporter           ExporterKind = "log"
	S3Exporter            ExporterKind = "s3"
	KinesisExporter       ExporterKind = "kinesis"
	GoogleStorageExporter ExporterKind = "googleStorage"
	SQSExporter           ExporterKind = "sqs"
	KafkaExporter         ExporterKind = "kafka"
	PubSubExporter        ExporterKind = "pubsub"
	AzureExporter         ExporterKind = "azureBlobStorage"
)

// IsValid is checking if the value is part of the enum
func (r ExporterKind) IsValid() error {
	switch r {
	case FileExporter, WebhookExporter, LogExporter, S3Exporter, GoogleStorageExporter, SQSExporter, KafkaExporter,
		PubSubExporter, KinesisExporter, AzureExporter:
		return nil
	}
	return fmt.Errorf("invalid exporter: kind \"%s\" is not supported", r)
}
