package outputs

import (
	_ "embed"
	"encoding/json"
	"fmt"
	"reflect"

	"github.com/dolphindb/api-go/multigoroutinetable"
	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/plugins/outputs"
)

// DO NOT REMOVE THE NEXT TWO LINES! This is required to embed the sampleConfig data.
//go:embed sample.conf
var sampleConfig string

type DolphinDBOutput struct {
	Config `toml:",inline"`

	Debug bool `toml:"debug"`

	Log telegraf.Logger `toml:"-"`

	tableSchema *TableSchema                             `toml:"-"`
	serializer  *Serializer                              `toml:"-"`
	writer      *multigoroutinetable.MultiGoroutineTable `toml:"-"`
}

func init() {
	outputs.Add("dolphindb", func() telegraf.Output { return &DolphinDBOutput{} })
}

func (do *DolphinDBOutput) Init() error {
	var err error

	// revise config
	do.Config.Revise()
	if do.Debug {
		do.Log.Debugf("The dolphindb config for telegraf is %+v", do.Config)
	}

	// get table schema from dolphindb
	do.tableSchema, err = NewTableSchema(&do.Config)
	if err != nil {
		return fmt.Errorf("new table schema failed. %v", err)
	}

	b, _ := json.Marshal(do.tableSchema.ColumnMetaSet)
	do.Log.Infof("Schema.ColumnMetaSet is %s", string(b))

	// init the serializer
	do.serializer = NewSerializer(do.tableSchema)
	return nil
}

func (do *DolphinDBOutput) SampleConfig() string {
	return sampleConfig
}

func (do *DolphinDBOutput) Connect() error {
	var err error

	do.writer, err = multigoroutinetable.NewMultiGoroutineTable(&multigoroutinetable.Option{
		Database:       do.Config.Database,
		Address:        do.Config.Address,
		UserID:         do.Config.User,
		Password:       do.Config.Password,
		TableName:      do.Config.TableName,
		GoroutineCount: do.Config.GoroutineCount,
		PartitionCol:   do.Config.PartitionCol,
		BatchSize:      do.Config.BatchSize,
		Throttle:       do.Config.Throttle,
	})
	if err != nil {
		return fmt.Errorf("new dolphindb failed. %v", err)
	}

	do.Log.Info("DolphinDB Connect!")
	return nil
}

func (do *DolphinDBOutput) Close() error {
	do.writer.WaitForGoroutineCompletion()
	status := do.writer.GetStatus()

	do.Log.Infof("DolphinDB Closed! Status is : \n%+v", status)

	if status.ErrMsg != "" {
		return fmt.Errorf("close failed. %v", status.ErrMsg)
	}

	return nil
}

func (do *DolphinDBOutput) Write(metrics []telegraf.Metric) error {
	if len(metrics) == 0 {
		return nil
	}

	for _, metric := range metrics {
		if metric.Name() != do.MetricName {
			continue
		}

		if do.Debug {
			do.printMetricInfo(metric)
		}

		row, err := do.serializer.Serialize(metric)
		if err != nil {
			return fmt.Errorf("serialize metric %+v failed. %v", metric, err)
		}

		err = do.writer.Insert(row...)
		if err != nil {
			return fmt.Errorf("write metric failed. %v", do.writer.GetStatus().ErrMsg)
		}
	}

	return nil
}

func (do *DolphinDBOutput) printMetricInfo(metric telegraf.Metric) {
	do.Log.Debugf("metric name is: %s, type is: %+v, time is: %+v", metric.Name(), metric.Type(), metric.Time())

	for _, t := range metric.TagList() {
		do.Log.Debugf("metric name is: %s, tag is: key: %s, value: %s, kind: %v",
			metric.Name(), t.Key, t.Value, reflect.TypeOf(t.Value).Kind().String())
	}

	for _, f := range metric.FieldList() {
		do.Log.Debugf("metric name is: %s, field is: key: %s, value: %+v, kind: %v",
			metric.Name(), f.Key, f.Value, reflect.TypeOf(f.Value).Kind().String())
	}
}
