// 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 es

import (
	"crypto/tls"
	"fmt"
	"net/http"
	"time"

	"github.com/cenkalti/backoff/v4"
	"github.com/elastic/go-elasticsearch/v7"
	"github.com/elastic/go-elasticsearch/v7/esutil"
	"github.com/opensearch-project/opensearch-go"
	"github.com/opensearch-project/opensearch-go/opensearchutil"
	log2 "github.com/pingcap/log"
)

// RawClient makes a *elasticsearch.Client
func RawClient(host string, port uint16) (*elasticsearch.Client, error) {
	retryBackoff := backoff.NewExponentialBackOff()

	return elasticsearch.NewClient(elasticsearch.Config{
		RetryOnStatus: []int{502, 503, 504, 429},
		Addresses:     []string{fmt.Sprintf("http://%s:%d", host, port)},
		RetryBackoff: func(i int) time.Duration {
			log2.Warn("retry elasticsearch")
			if i == 1 {
				retryBackoff.Reset()
			}
			return retryBackoff.NextBackOff()
		},
		EnableRetryOnTimeout: true,
		MaxRetries:           3,
	})
}

func AwsEsClient(endpoint, user, pw string) (*opensearch.Client, error) {
	retryBackoff := backoff.NewExponentialBackOff()

	cfg := opensearch.Config{
		Addresses: []string{
			endpoint,
		},
		Username: user,
		Password: pw,
		Transport: &http.Transport{
			// MaxIdleConnsPerHost:   10,
			// ResponseHeaderTimeout: 30 * time.Second,
			// DialContext:           (&net.Dialer{Timeout: time.Second}).DialContext,
			TLSClientConfig: &tls.Config{
				MinVersion: tls.VersionTLS11,
			},
		},
		RetryBackoff: func(i int) time.Duration {
			log2.Warn("retry elasticsearch")
			if i == 1 {
				retryBackoff.Reset()
			}
			return retryBackoff.NextBackOff()
		},
		EnableRetryOnTimeout: true,
		MaxRetries:           3,
	}
	cli, _ := opensearch.NewClient(cfg)
	return cli, nil
}

func NewBulkIndexer(cli *elasticsearch.Client, index string) (esutil.BulkIndexer, error) {
	return esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
		Index:         index,            // The default index name
		Client:        cli,              // The Elasticsearch client
		NumWorkers:    10,               // The number of worker goroutines
		FlushBytes:    5000000,          // The flush threshold in bytes
		FlushInterval: 30 * time.Second, // The periodic flush interval
	})
}

func NewBulkIndexerAws(cli *opensearch.Client, index string) (opensearchutil.BulkIndexer, error) {
	return opensearchutil.NewBulkIndexer(opensearchutil.BulkIndexerConfig{
		Index:         index,            // The default index name
		Client:        cli,              // The Elasticsearch client
		NumWorkers:    10,               // The number of worker goroutines
		FlushBytes:    5000000,          // The flush threshold in bytes
		FlushInterval: 30 * time.Second, // The periodic flush interval
	})
}