// Copyright 2021 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

// This file only contains version related variables and consts, all
// type definitions and functions shall not be implemented here.
// This file is excluded from CI tests.

package configuration

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"

	"github.com/opensearch-project/opensearch-go/opensearchutil"
	log2 "github.com/pingcap/log"
	"go.uber.org/multierr"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/config"
	"github.com/pingcap-inc/clinic/pkg/importer/utils"
)

type ConfigurationSink struct {
	index opensearchutil.BulkIndexer
}

// var (
// 	configurationMapping = `{
//   "mappings": {
//     "properties": {
// 	  "package_id": {
// 	    "type": "keyword"
// 	  },
//       "host": {
//         "type": "keyword"
//       },
//       "instance": {
//         "type": "keyword"
//       },
//       "component": {
//         "type": "keyword"
//       },
//       "path": {
//         "type": "keyword"
//       },
// 	  "content": {
//         "type": "text",
//         "fields": {
//           "keyword": {"type": "keyword", "ignore_above": 256}
//         }
//       }
//     }
//   }
// }`
// )

func NewConfigurationSink(cfg config.LogConfig, clusterID uint64) (*ConfigurationSink, error) {
	// TODO
	return nil, nil
	//cli, err := es.AwsEsClient(cfg.Endpoint, cfg.User, cfg.Password)
	//if err != nil {
	//	return nil, err
	//}
	//
	//index := fmt.Sprintf("%d-configuration", clusterID)
	//
	//if err := es.CreateIndexIfAbsentAws(cli, index, configurationMapping); err != nil {
	//	return nil, err
	//}
	//es.CreateIndexPatternIfAbsent(cfg, index)
	//
	//bi, err := es.NewBulkIndexerAws(cli, index)
	//if err != nil {
	//	return nil, err
	//}
	//
	//return &ConfigurationSink{
	//	index: bi,
	//}, nil
}

func (es *ConfigurationSink) Sink(cfgs []ConfigurationRecord) error {
	var errs error
	for _, cfg := range cfgs {
		if cfg.PackageID == "" {
			continue
		}

		body, err := json.Marshal(cfg)
		if err != nil {
			log2.Error(
				"marshal configuration failed",
				zap.String("package", cfg.PackageID),
				zap.String("path", cfg.Path),
				zap.String("host", cfg.Host),
				zap.Error(err))
			continue
		}

		err = es.index.Add(context.TODO(), opensearchutil.BulkIndexerItem{
			Action:     "index",
			Body:       bytes.NewReader(body),
			DocumentID: fmt.Sprintf("%s-%s-%s", utils.FnvHash(cfg.PackageID), utils.FnvHash(cfg.Host), utils.FnvHash(cfg.Path)),
			// OnFailure is called for each failed operation
			OnFailure: func(ctx context.Context, item opensearchutil.BulkIndexerItem, res opensearchutil.BulkIndexerResponseItem, err error) {
				if err != nil {
					log2.Error("insert configuration to es failed", zap.Error(err))
				} else {
					log2.Error("insert configuration to es failed", zap.String("type", res.Error.Type), zap.String("reason", res.Error.Reason))
				}
			},
		})

		errs = multierr.Append(errs, err)
	}

	return errs
}

func (es *ConfigurationSink) Close() error {
	return es.index.Close(context.TODO())
}