package log

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

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

	"github.com/pingcap-inc/clinic/pkg/config"
	"github.com/pingcap-inc/clinic/pkg/es"

	log2 "github.com/pingcap/log"
)

type LogSink struct {
	previousLogRecord previousLogRecord
	// index             opensearchutil.BulkIndexer
	awsIndex          opensearchutil.BulkIndexer
	awsSlowQueryIndex opensearchutil.BulkIndexer
	beginTS           int64 // timestamp(ms) of earliest log that need to rebuild
	endTS             int64 // timestamp(ms) of latest log that need to rebuild
	// dissect *log.Dissector
	// slowQueryIndex opensearchutil.BulkIndexer // a new index for store slow query log
}

type previousLogRecord struct {
	Host       string
	Component  string
	PackageID  string
	InstanceID string
	LogType    LogType
	TimeStamp  int64
	ClusterID  string
}

func NewLogSink(cfg config.LogConfig, orgID uint, beginTS, endTS int64) (*LogSink, error) {
	cli, err := es.AwsEsClient(cfg.Endpoint, cfg.User, cfg.Password)
	if err != nil {
		return nil, err
	}

	index := es.GetLogIndexName(orgID)
	slowQueryIndex := es.GetSlowLogIndexName(orgID)

	// for normal index
	//kibana, err := model.GetByOrgID(orgID)
	//if err != nil {
	//	log2.Error("get org failed", zap.Uint("org", orgID), zap.Error(err))
	//	return nil, err
	//}

	if err := es.CreateIndexIfAbsentAws(cli, index, logmapping); err != nil {
		return nil, err
	}
	es.CreateIndexPatternIfAbsent(orgID, cfg, index)

	// for slow query log index
	if err := es.CreateIndexIfAbsentAws(cli, slowQueryIndex, SlowQueryLogMapping); err != nil {
		return nil, err
	}
	es.CreateIndexPatternIfAbsent(orgID, cfg, slowQueryIndex)

	bi, err := es.NewBulkIndexerAws(cli, index)
	if err != nil {
		return nil, err
	}

	slowBI, err := es.NewBulkIndexerAws(cli, slowQueryIndex)
	if err != nil {
		return nil, err
	}

	return &LogSink{
		awsIndex: bi,
		// index:             bi,
		// slowQueryIndex:    slowBI,
		awsSlowQueryIndex: slowBI,
		beginTS:           beginTS,
		endTS:             endTS,
	}, nil
}

func (es *LogSink) Sink(logs []LogRecord) error {
	var errs error
	for _, olog := range logs {
		if olog.RecordID == "" {
			continue
		}
		if olog.IsSlowQueryLog() {
			if err := es.sinkSlowQueryLog(olog); err != nil {
				log2.Error("sink slow query log failed", zap.Error(err))
				errs = multierr.Append(errs, err)
			}
		} else {
			if err := es.sinkLog(olog); err != nil {
				log2.Error("sink log failed", zap.Error(err))
				errs = multierr.Append(errs, err)
			}
		}
	}

	return errs
}

func (es *LogSink) sinkLog(olog LogRecord) error {
	record, err := es.parseNormalLog(olog)
	if err != nil {
		log2.Error("convert message to record struct failed", zap.String("fileUUID", olog.PackageID), zap.String("component", olog.Component), zap.String("host", olog.Host), zap.String("instance", olog.InstanceID), zap.Error(err))
		return nil
	}
	// ignore log which is out of range
	if record.Timestamp < es.beginTS ||
		record.Timestamp > es.endTS {
		return nil
	}

	body, err := json.Marshal(record)
	if err != nil {
		log2.Error("marshal log failed", zap.String("log", olog.Record), zap.Error(err))
		return nil
	}

	return es.awsIndex.Add(context.Background(), opensearchutil.BulkIndexerItem{
		Action:     "index",
		Body:       bytes.NewReader(body),
		DocumentID: olog.RecordID,
		OnSuccess: func(ctx context.Context, item opensearchutil.BulkIndexerItem, res opensearchutil.BulkIndexerResponseItem) {
			// atomic.AddUint64(&countSuccessful, 1)
		},
		// 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 log to es failed", zap.Error(err))
			} else {
				log2.Error("insert log to es failed", zap.String("type", res.Error.Type), zap.String("reason", res.Error.Reason))
			}
		},
	})
}

func (es *LogSink) parseNormalLog(olog LogRecord) (*Record, error) {
	if olog.Record == "" {
		log2.Warn("empty log record", zap.String("fileUUID", olog.PackageID), zap.String("host", olog.Host), zap.String("instance", olog.InstanceID), zap.String("component", olog.Component))
		return nil, nil
	}

	dissector, ok := logDissectMap[olog.LogType]
	if !ok {
		log2.Warn("get log dissector failed", zap.String("fileUUID", olog.PackageID), zap.String("host", olog.Host), zap.String("instance", olog.InstanceID), zap.String("component", olog.Component))
		return nil, nil
	}
	record, err := dissector.convertToRecord(olog)
	if record != nil && err == nil {
		es.previousLogRecord = previousLogRecord{
			Host:       olog.Host,
			Component:  olog.Component,
			PackageID:  olog.PackageID,
			InstanceID: olog.InstanceID,
			LogType:    olog.LogType,
			TimeStamp:  record.Timestamp,
			ClusterID:  olog.ClusterID,
		}
		return record, nil
	}

	if isSameComponent(es.previousLogRecord, olog) {
		log2.Warn("use time of the previous record", zap.String("fileUUID", olog.PackageID), zap.String("host", olog.Host), zap.String("instance", olog.InstanceID), zap.String("component", olog.Component), zap.Error(err))
		return &Record{
			Timestamp:  es.previousLogRecord.TimeStamp,
			Level:      "error",
			Message:    olog.Record,
			Host:       olog.Host,
			Component:  olog.Component,
			PackageID:  olog.PackageID,
			ClusterID:  olog.ClusterID,
			InstanceID: olog.InstanceID,
			LogType:    string(olog.LogType),
		}, nil
	}

	return nil, err
}

func isSameComponent(previousRecord previousLogRecord, currentRecord LogRecord) bool {
	if previousRecord.LogType != currentRecord.LogType {
		return false
	}

	if previousRecord.Component != currentRecord.Component {
		return false
	}

	if previousRecord.PackageID != currentRecord.PackageID {
		return false
	}

	if previousRecord.Host != currentRecord.Host {
		return false
	}

	return previousRecord.InstanceID == currentRecord.InstanceID
}

func (es *LogSink) sinkSlowQueryLog(olog LogRecord) error {
	if olog.LogType != slowQueryType || olog.SlowQueryRecord == nil {
		return errors.New("log type not match")
	}
	dissector, ok := logDissectMap[olog.LogType]
	if !ok {
		return fmt.Errorf("get log dissector failed with logType: %v", olog.LogType)
	}
	slowRecord, err := dissector.convertToSlowQueryRecord(olog)
	if err != nil {
		return err
	}
	body, err := json.Marshal(slowRecord)
	if err != nil {
		return err
	}
	return es.awsSlowQueryIndex.Add(context.Background(), opensearchutil.BulkIndexerItem{
		Action:     "index",
		Body:       bytes.NewReader(body),
		DocumentID: olog.RecordID,
		OnSuccess: func(ctx context.Context, item opensearchutil.BulkIndexerItem, res opensearchutil.BulkIndexerResponseItem) {
			// atomic.AddUint64(&countSuccessful, 1)
		},
		// 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 log to es failed", zap.Error(err))
			} else {
				log2.Error("insert log to es failed", zap.String("type", res.Error.Type), zap.String("reason", res.Error.Reason))
			}
		},
	})
}

func (es *LogSink) Close() error {
	ctx := context.Background()
	if err := es.awsIndex.Close(ctx); err != nil {
		return err
	}
	if err := es.awsSlowQueryIndex.Close(ctx); err != nil {
		return err
	}
	return nil
}

type Record struct {
	Timestamp int64 `json:"timestamp"`

	Level string `json:"level"`

	Message string `json:"message"`

	Host string `json:"host"`

	Component string `json:"component"`

	PackageID string `json:"packageid"` //nolint:misspell

	InstanceID string `json:"instanceid"`

	LogType string `json:"type"`

	FileName string `json:"filename"`

	ClusterID string `json:"cluster_id"`
}

// SlowQueryRecord is sent to slow query index
type SlowQueryRecord struct {
	Timestamp int64 `json:"timestamp"`

	// Message string `json:"message"`

	// Digest string `json:"digest"`

	PlanDigest string `json:"plan_digest"`

	// ProcessKeys int64 `json:"process_keys"`

	// ProcessTime float64 `json:"process_time"`

	// TotalKeys int64 `json:"total_keys"`

	// RocksdbDeleteSkippedCount int64 `json:"rocksdb_delete_skipped_count"`

	// Host string `json:"host"`

	Component string `json:"component"`

	PackageID string `json:"packageid"` //nolint:misspell

	ClusterID string `json:"cluster_id"`

	InstanceID string `json:"instanceid"`

	LogType string `json:"type"`

	BackoffTime float64 `json:"backoff_time"`

	BackoffTypes string `json:"backoff_types"`

	CommitBackoffTime float64 `json:"commit_backoff_time"`

	CommitTime float64 `json:"commit_time"`

	CompileTime float64 `json:"compile_time"`

	ConnetionID uint64 `json:"connection_id"`

	CopProcAddr string `json:"cop_proc_addr"`

	CopProcAvg float64 `json:"cop_proc_avg"`

	CopProcMax float64 `json:"cop_proc_max"`

	CopProcP90 float64 `json:"cop_proc_p90"`

	CopTime float64 `json:"cop_time"`

	CopWaitAddr string `json:"cop_wait_addr"`

	CopWaitAvg float64 `json:"cop_wait_avg"`

	CopWaitMax float64 `json:"cop_wait_max"`

	CopWaitP90 float64 `json:"cop_wait_p90"`

	Digest string `json:"digest"`

	DiskMax int64 `json:"disk_max"`

	ExecRetryTime float64 `json:"exec_retry_time"`

	GetCommitTsTime float64 `json:"get_commit_ts_time"`

	Host string `json:"host"`

	IndexNames string `json:"index_names"`

	// Instance string `json:"instance"`

	IsInternal int64 `json:"is_internal"`

	LocalLatchWaitTime float64 `json:"local_latch_wait_time"`

	LockKeysTime float64 `json:"lock_keys_time"`

	MemoryMax int64 `json:"memory_max"`

	OptimizeTime float64 `json:"optimize_time"`

	ParseTime float64 `json:"parse_time"`

	Plan string `json:"plan"`

	BinaryPlan string `json:"binary_plan"`

	PreprocSubqueriesTime float64 `json:"preproc_subqueries_time"`

	PrevStmt string `json:"prev_stmt"`

	PrewriteRegion uint64 `json:"prewrite_region"`

	PrewriteTime float64 `json:"prewrite_time"`

	ProcessKeys uint64 `json:"process_keys"`

	ProcessTime float64 `json:"process_time"`

	Query string `json:"query"`

	QueryTime float64 `json:"query_time"`

	RequestCount uint64 `json:"request_count"`

	ResolveLockTime float64 `json:"resolve_lock_time"`

	RewriteTime float64 `json:"rewrite_time"`

	RocksdbBlockCacheHitCount uint64 `json:"rocksdb_block_cache_hit_count"`

	RocksdbBlockReadByte uint64 `json:"rocksdb_block_read_byte"`

	RocksdbBlockReadCount uint64 `json:"rocksdb_block_read_count"`

	RocksdbDeleteSkippedCount uint64 `json:"rocksdb_delete_skipped_count"`

	RocksdbKeySkippedCount uint64 `json:"rocksdb_key_skipped_count"`

	Stats string `json:"stats"`

	Success int64 `json:"success"`

	TotalKeys uint64 `json:"total_keys"`

	TxnRetry uint64 `json:"txn_retry"`

	TxnStartTs uint64 `json:"txn_start_ts"`

	User string `json:"user"`

	WaitPrewriteBinlogTime float64 `json:"wait_prewrite_binlog_time"`

	WaitTime float64 `json:"wait_time"`

	WaitTs float64 `json:"wait_ts"`

	WriteKeys uint64 `json:"write_keys"`

	WriteSize uint64 `json:"write_size"`

	WriteSqlResponseTotal float64 `json:"write_sql_response_total"`

	DB string `json:"db"`
}