package log

import (
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/util/execdetails"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/slow_query"
	"github.com/pingcap-inc/clinic/pkg/slow_query/variable"

	log2 "github.com/pingcap/log"
)

var logDissectMap = make(map[LogType]logDissect)

func init() {
	RegisterDissect(NewStandardLogDissect())
	RegisterDissect(NewTiflashErrLogDissect())
	RegisterDissect(NewSlowQueryDissector())
}

type logDissect interface {
	name() LogType

	convertToRecord(record LogRecord) (*Record, error)
	convertToSlowQueryRecord(record LogRecord) (*SlowQueryRecord, error)
}

func RegisterDissect(d logDissect, err error) {
	if err != nil {
		log2.Error("init dissector failed", zap.String("name", string(d.name())), zap.Error(err))
		panic(err)
	}

	_, ok := logDissectMap[d.name()]
	if ok {
		log2.Error("the dissector is registered", zap.String("name", string(d.name())))
		panic("the dissector is registered")
	}

	logDissectMap[d.name()] = d
}

type standardLogDissect struct {
	dissector *Dissector
}

func NewStandardLogDissect() (*standardLogDissect, error) {
	dissector, err := New("[%{timestamp}] [%{level}] %{msg}")
	if err != nil {
		return nil, err
	}

	return &standardLogDissect{
		dissector: dissector,
	}, nil
}

func (sld *standardLogDissect) name() LogType {
	return standardType
}

func (sld *standardLogDissect) convertToRecord(record LogRecord) (*Record, error) {
	// if the log is json format, we convert it to record without dissectConvert
	if len(record.Record) >= 2 && record.Record[0] == '{' && record.Record[1] == '"' {
		return jsonToRecord(record)
	}

	c, err := sld.dissector.DissectConvert(record.Record)
	if err != nil {
		log2.Error("convert log failed", zap.Error(err), zap.String("log", record.Record))
		return nil, err
	}

	if len(c) == 0 {
		return nil, nil
	}

	m, ok := c[message].(string)
	if !ok || m == "" {
		return nil, errors.New("empty message")
	}

	l, ok := c[level].(string)
	if !ok || l == "" {
		return nil, errors.New("empty level")
	}

	t, ok := c[timestamp].(string)
	if !ok || t == "" {
		return nil, errors.New("empty timestamp")
	}

	ts, err := time.Parse("2006/01/02 15:04:05.000 -07:00", t)
	if err != nil {
		return nil, err
	}

	return &Record{
		ClusterID:  record.ClusterID,
		PackageID:  record.PackageID,
		Component:  record.Component,
		Host:       record.Host,
		Message:    m,
		Level:      strings.ToLower(l),
		InstanceID: record.InstanceID,
		Timestamp:  ts.UnixNano() / 1000000,
		LogType:    string(record.LogType),
		FileName:   record.FileName,
	}, nil
}

func jsonToRecord(record LogRecord) (*Record, error) {
	var line map[string]any
	err := json.Unmarshal([]byte(record.Record), &line)
	if err != nil {
		return nil, err
	}

	l, ok := line["level"].(string)
	if !ok || l == "" {
		return nil, errors.New("empty level")
	}

	t, ok := line["time"].(string)
	if !ok || t == "" {
		return nil, errors.New("empty timestamp")
	}
	ts, err := time.Parse("2006/01/02 15:04:05.000 -07:00", t)
	if err != nil {
		return nil, err
	}

	line[level] = nil
	line[timestamp] = nil
	m, err := json.Marshal(line)
	if err != nil {
		return nil, err
	}

	return &Record{
		ClusterID:  record.ClusterID,
		PackageID:  record.PackageID,
		Component:  record.Component,
		Host:       record.Host,
		Message:    string(m),
		Level:      strings.ToLower(l),
		InstanceID: record.InstanceID,
		Timestamp:  ts.UnixNano() / 1000000,
		LogType:    string(record.LogType),
		FileName:   record.FileName,
	}, nil
}

func (sld *standardLogDissect) convertToSlowQueryRecord(record LogRecord) (*SlowQueryRecord, error) {
	return nil, errors.New("can not convert to slow query record")
}

type tiflashErrLogDissect struct {
	dissector *Dissector
}

func NewTiflashErrLogDissect() (*tiflashErrLogDissect, error) {
	dissector, err := New("%{timestamp} [%{level}] %{msg}")
	if err != nil {
		return nil, err
	}

	return &tiflashErrLogDissect{
		dissector: dissector,
	}, nil
}

func (sld *tiflashErrLogDissect) name() LogType {
	return tiflashErrType
}

func (sld *tiflashErrLogDissect) convertToRecord(record LogRecord) (*Record, error) {
	c, err := sld.dissector.DissectConvert(record.Record)
	if err != nil {
		log2.Error("convert log failed", zap.Error(err), zap.String("log", record.Record))
		return nil, err
	}

	if len(c) == 0 {
		return nil, nil
	}

	m, ok := c[message].(string)
	if !ok || m == "" {
		return nil, errors.New("empty message")
	}

	t, ok := c[timestamp].(string)
	if !ok || t == "" {
		return nil, errors.New("empty timestamp")
	}

	loc, err := time.LoadLocation("Asia/Chongqing")
	if err != nil {
		log2.Error("create location failed", zap.Error(err))
		loc = time.Local
	}
	ts, err := time.ParseInLocation("2006.01.02 15:04:05.000000", t, loc)
	if err != nil {
		return nil, err
	}

	return &Record{
		ClusterID:  record.ClusterID,
		PackageID:  record.PackageID,
		Component:  record.Component,
		Host:       record.Host,
		Message:    fmt.Sprintf("[%s] %s", c[level], m),
		Level:      "error",
		InstanceID: record.InstanceID,
		Timestamp:  ts.UnixNano() / 1000000,
		LogType:    string(record.LogType),
		FileName:   record.FileName,
	}, nil
}

func (sld *tiflashErrLogDissect) convertToSlowQueryRecord(record LogRecord) (*SlowQueryRecord, error) {
	return nil, errors.New("can not convert to slow query record")
}

type slowqueryDissector struct {
	dissector *Dissector
}

func NewSlowQueryDissector() (*slowqueryDissector, error) {
	dissector, err := New("%{} %{} %{timestamp} %{msg}")
	if err != nil {
		return nil, err
	}
	return &slowqueryDissector{dissector: dissector}, nil
}

func (sd *slowqueryDissector) name() LogType {
	return slowQueryType
}

func (sd *slowqueryDissector) convertToRecord(record LogRecord) (*Record, error) {
	return nil, errors.New("can not convert to record")
}

func (sd *slowqueryDissector) convertToSlowQueryRecord(record LogRecord) (*SlowQueryRecord, error) {
	if record.SlowQueryRecord == nil {
		return nil, errors.New("record has empty slow query record")
	}

	// TODO querytemplate id
	// queryTemplate := record.SlowQueryRecord.GetDatum(variable)
	// TODO finish time, latency, generatePlanTime

	backOffTime := record.SlowQueryRecord.GetDatum(execdetails.BackoffTimeStr)
	backOffTypes := record.SlowQueryRecord.GetDatum(execdetails.BackoffTypesStr)

	commitBackoffTime := record.SlowQueryRecord.GetDatum(execdetails.CommitBackoffTimeStr)
	commitTimeStr := record.SlowQueryRecord.GetDatum(execdetails.CommitTimeStr)

	compileTimeStr := record.SlowQueryRecord.GetDatum(variable.SlowLogCompileTimeStr)

	connectionID := record.SlowQueryRecord.GetDatum(variable.SlowLogConnIDStr)

	copProcAddr := record.SlowQueryRecord.GetDatum(variable.SlowLogCopProcAddr)
	copProcAvgStr := record.SlowQueryRecord.GetDatum(variable.SlowLogCopProcAvg)
	copProcMaxStr := record.SlowQueryRecord.GetDatum(variable.SlowLogCopProcMax)
	copProcP90Str := record.SlowQueryRecord.GetDatum(variable.SlowLogCopProcP90)
	copTimeStr := record.SlowQueryRecord.GetDatum(execdetails.CopTimeStr)
	copWaitAddr := record.SlowQueryRecord.GetDatum(variable.SlowLogCopWaitAddr)
	copWaitAvgStr := record.SlowQueryRecord.GetDatum(variable.SlowLogCopWaitAvg)
	copWaitMaxStr := record.SlowQueryRecord.GetDatum(variable.SlowLogCopWaitMax)
	copWaitP90Str := record.SlowQueryRecord.GetDatum(variable.SlowLogCopWaitP90)

	db := record.SlowQueryRecord.GetDatum(variable.SlowLogDBStr)
	digest := record.SlowQueryRecord.GetDatum(variable.SlowLogDigestStr)

	diskMax := record.SlowQueryRecord.GetDatum(variable.SlowLogDiskMax)

	execRetryTimeStr := record.SlowQueryRecord.GetDatum(variable.SlowLogExecRetryTime)

	getCommitTsTime := record.SlowQueryRecord.GetDatum(execdetails.GetCommitTSTimeStr)

	// host := record.SlowQueryRecord.GetDatum(variable.SlowLogHostStr)

	// instance := record.SlowQueryRecord.GetDatum(execdetails.)
	indexNames := record.SlowQueryRecord.GetDatum(variable.SlowLogIndexNamesStr)

	isInternal := record.SlowQueryRecord.GetDatum(variable.SlowLogIsInternalStr)

	localLatchWaitTime := record.SlowQueryRecord.GetDatum(execdetails.LocalLatchWaitTimeStr)

	lockKeysTime := record.SlowQueryRecord.GetDatum(execdetails.LockKeysTimeStr)

	memoryMax := record.SlowQueryRecord.GetDatum(variable.SlowLogMemMax)

	optimizeTimeStr := record.SlowQueryRecord.GetDatum(variable.SlowLogOptimizeTimeStr)

	parseTimeStr := record.SlowQueryRecord.GetDatum(variable.SlowLogParseTimeStr)

	plan := record.SlowQueryRecord.GetDatum(variable.SlowLogPlan)
	binaryPlan := record.SlowQueryRecord.GetDatum(variable.SlowLogBinaryPlan)

	preProcSubQueryTimeStr := record.SlowQueryRecord.GetDatum(variable.SlowLogPreProcSubQueryTimeStr)
	prevStmt := record.SlowQueryRecord.GetDatum(variable.SlowLogPrevStmt)

	prewriteRegion := record.SlowQueryRecord.GetDatum(execdetails.PrewriteRegionStr)
	prewriteTime := record.SlowQueryRecord.GetDatum(execdetails.PreWriteTimeStr)
	processKeys := record.SlowQueryRecord.GetDatum(execdetails.ProcessKeysStr)
	processTime := record.SlowQueryRecord.GetDatum(execdetails.ProcessTimeStr)

	queryStr := record.SlowQueryRecord.GetDatum(variable.SlowLogQuerySQLStr)
	queryTimeStr := record.SlowQueryRecord.GetDatum(variable.SlowLogQueryTimeStr)

	requestCount := record.SlowQueryRecord.GetDatum(execdetails.RequestCountStr)
	resolveLockTime := record.SlowQueryRecord.GetDatum(execdetails.ResolveLockTimeStr)

	rewriteTimeStr := record.SlowQueryRecord.GetDatum(variable.SlowLogRewriteTimeStr)
	rocksdbBlockCacheHitCountStr := record.SlowQueryRecord.GetDatum(execdetails.RocksdbBlockCacheHitCountStr)
	rocksdbBlockReadByteStr := record.SlowQueryRecord.GetDatum(execdetails.RocksdbBlockReadByteStr)
	rocksdbBlockReadCount := record.SlowQueryRecord.GetDatum(execdetails.RocksdbBlockReadCountStr)
	rocksdbDeleteSkippedCount := record.SlowQueryRecord.GetDatum(execdetails.RocksdbDeleteSkippedCountStr)
	rocksdbKeySkippedCount := record.SlowQueryRecord.GetDatum(execdetails.RocksdbKeySkippedCountStr)

	stats := record.SlowQueryRecord.GetDatum(variable.SlowLogStatsInfoStr)
	success := record.SlowQueryRecord.GetDatum(variable.SlowLogSucc)

	totalKeys := record.SlowQueryRecord.GetDatum(execdetails.TotalKeysStr)
	txnRetry := record.SlowQueryRecord.GetDatum(execdetails.TxnRetryStr)

	txnStartTs := record.SlowQueryRecord.GetDatum(variable.SlowLogTxnStartTSStr)

	user := record.SlowQueryRecord.GetDatum(variable.SlowLogUserStr)

	waitPreWriteBinlogTimeStr := record.SlowQueryRecord.GetDatum(execdetails.WaitPrewriteBinlogTimeStr)

	waitTime := record.SlowQueryRecord.GetDatum(execdetails.WaitTimeStr)

	waitTs := record.SlowQueryRecord.GetDatum(variable.SlowLogWaitTSTimeStr)

	writeKeys := record.SlowQueryRecord.GetDatum(execdetails.WriteKeysStr)
	writeSize := record.SlowQueryRecord.GetDatum(execdetails.WriteSizeStr)
	writeSqlResponseTotal := record.SlowQueryRecord.GetDatum(variable.SlowLogWriteSQLRespTotal)

	timeStr := record.SlowQueryRecord.GetDatum(variable.SlowLogTimeStr)

	planDigest := record.SlowQueryRecord.GetDatum(variable.SlowLogPlanDigest)
	// processTime := record.SlowQueryRecord.GetDatum(execdetails.ProcessTimeStr)
	// processKeys := record.SlowQueryRecord.GetDatum(execdetails.ProcessKeysStr)

	// rocksdbDeleteSkippedCnt := record.SlowQueryRecord.GetDatum(execdetails.RocksdbDeleteSkippedCountStr)

	goTime, err := slow_query.ParseTime(timeStr.GetString())
	if err != nil {
		return nil, err
	}
	slowRecord := &SlowQueryRecord{
		Timestamp:         goTime.UnixNano() / 1000000,
		BackoffTime:       backOffTime.GetFloat64(),
		BackoffTypes:      backOffTypes.GetString(),
		CommitBackoffTime: commitBackoffTime.GetFloat64(),
		CommitTime:        commitTimeStr.GetFloat64(),
		CompileTime:       compileTimeStr.GetFloat64(),
		ConnetionID:       connectionID.GetUint64(),
		CopProcAddr:       copProcAddr.GetString(),
		CopProcAvg:        copProcAvgStr.GetFloat64(),
		CopProcMax:        copProcMaxStr.GetFloat64(),
		CopProcP90:        copProcP90Str.GetFloat64(),
		CopTime:           copTimeStr.GetFloat64(),
		CopWaitAddr:       copWaitAddr.GetString(),
		CopWaitAvg:        copWaitAvgStr.GetFloat64(),
		CopWaitMax:        copWaitMaxStr.GetFloat64(),
		CopWaitP90:        copWaitP90Str.GetFloat64(),
		DB:                db.GetString(),
		// Message:                   record.Record,
		Digest:          digest.GetString(),
		DiskMax:         diskMax.GetInt64(),
		ExecRetryTime:   execRetryTimeStr.GetFloat64(),
		GetCommitTsTime: getCommitTsTime.GetFloat64(),
		Host:            record.Host,

		IndexNames: indexNames.GetString(),
		// Instance: ,

		IsInternal:            isInternal.GetInt64(),
		LocalLatchWaitTime:    localLatchWaitTime.GetFloat64(),
		LockKeysTime:          lockKeysTime.GetFloat64(),
		MemoryMax:             memoryMax.GetInt64(),
		OptimizeTime:          optimizeTimeStr.GetFloat64(),
		ParseTime:             parseTimeStr.GetFloat64(),
		Plan:                  plan.GetString(),
		BinaryPlan:            binaryPlan.GetString(),
		PreprocSubqueriesTime: preProcSubQueryTimeStr.GetFloat64(),
		PrevStmt:              prevStmt.GetString(),

		PrewriteRegion: prewriteRegion.GetUint64(),
		PrewriteTime:   prewriteTime.GetFloat64(),

		ProcessKeys:               processKeys.GetUint64(),
		ProcessTime:               processTime.GetFloat64(),
		Query:                     queryStr.GetString(),
		QueryTime:                 queryTimeStr.GetFloat64(),
		RequestCount:              requestCount.GetUint64(),
		ResolveLockTime:           resolveLockTime.GetFloat64(),
		RewriteTime:               rewriteTimeStr.GetFloat64(),
		RocksdbBlockCacheHitCount: rocksdbBlockCacheHitCountStr.GetUint64(),
		RocksdbBlockReadByte:      rocksdbBlockReadByteStr.GetUint64(),
		RocksdbBlockReadCount:     rocksdbBlockReadCount.GetUint64(),
		RocksdbDeleteSkippedCount: rocksdbDeleteSkippedCount.GetUint64(),
		RocksdbKeySkippedCount:    rocksdbKeySkippedCount.GetUint64(),
		Stats:                     stats.GetString(),
		Success:                   success.GetInt64(),
		TotalKeys:                 totalKeys.GetUint64(),

		TxnRetry:               txnRetry.GetUint64(),
		TxnStartTs:             txnStartTs.GetUint64(),
		User:                   user.GetString(),
		WaitPrewriteBinlogTime: waitPreWriteBinlogTimeStr.GetFloat64(),
		WaitTime:               waitTime.GetFloat64(),
		WaitTs:                 waitTs.GetFloat64(),
		WriteKeys:              writeKeys.GetUint64(),
		WriteSize:              writeSize.GetUint64(),
		WriteSqlResponseTotal:  writeSqlResponseTotal.GetFloat64(),

		PlanDigest: planDigest.GetString(),
		// ProcessTime:               processTime.GetFloat64(),
		// ProcessKeys:               processKeys.GetInt64(),

		ClusterID:  record.ClusterID,
		Component:  record.Component,
		PackageID:  record.PackageID,
		InstanceID: record.InstanceID,
		LogType:    string(record.LogType),
	}
	return slowRecord, nil
}