package log

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"strings"
	"time"

	"github.com/pingcap/log"
	"github.com/pingcap/tidb/types"
	"go.uber.org/zap"

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

var emptyLogRecord = LogRecord{}

type LogIterator struct {
	LogIteratorOpt
}

type LogIteratorOpt struct {
	reader *bufio.Reader
	//nolint:structcheck
	line           uint64
	host           string
	component      string
	identityPrefix string
	packageID      string
	instanceID     string
	fileName       string
	clusterID      string

	logType LogType
}

func (li *LogIterator) Next() (any, bool) {
	li.line++

	var (
		isPrefix       = true
		err      error = nil
		line, ln []byte
	)

	count := 1
	for isPrefix && err == nil {
		line, isPrefix, err = li.reader.ReadLine()
		ln = append(ln, line...)
		count++
	}

	if count > 100 {
		log.Error("the log line is too big", zap.String("fileUUID", li.packageID), zap.String("file", li.fileName), zap.String("host", li.host), zap.String("instance", li.instanceID), zap.Uint64("line", li.line), zap.Int("count", count))
	}

	lr := func(record string) LogRecord {
		metrics.SinkLogCount.WithLabelValues(li.packageID).Inc()
		return LogRecord{
			Record:     record,
			Host:       li.host,
			RecordID:   fmt.Sprintf("%s-%d", li.identityPrefix, li.line),
			Component:  li.component,
			PackageID:  li.packageID,
			ClusterID:  li.clusterID,
			InstanceID: li.instanceID,
			LogType:    li.logType,
			FileName:   li.fileName,
		}
	}

	if err != nil {
		if err == io.EOF && len(ln) > 0 {
			return lr(string(ln)), true
		}

		if err != io.EOF {
			log.Error("read line failed", zap.String("fileUUID", li.packageID), zap.String("file", li.fileName), zap.String("host", li.host), zap.String("instance", li.instanceID))
		}

		return LogRecord{}, false
	}

	return lr(string(ln)), true
}

func (li *LogIterator) Len() (int, bool) {
	return -1, false
}

type LogRecordOption struct {
	component  string
	host       string
	instanceID string
	logType    LogType
	fileName   string
}

func NewLogIterator(reader io.Reader, identityPrefix, clusterID, packageID string, opt *LogRecordOption) stream.Iterator {
	iteratorOpt := LogIteratorOpt{
		reader:         bufio.NewReader(reader),
		component:      opt.component,
		host:           opt.host,
		identityPrefix: identityPrefix,
		packageID:      packageID,
		clusterID:      clusterID,
		instanceID:     opt.instanceID,
		logType:        opt.logType,
		fileName:       opt.fileName,
	}

	if opt.logType == slowQueryType {
		retriever, err := slow_query.NewSlowQueryRetriever(time.Local, SlowQueryFullCols, slow_query.WithReader(reader))
		if err != nil {
			log.Warn("new slow query retriever error", zap.Error(err))
		}
		return &SlowQueryIterator{
			slowQueryCols:       SlowQueryFullCols,
			slowQueryColsLookUp: SlowQueryFullColsLookup(),
			retriever:           retriever,
			LogIteratorOpt:      iteratorOpt,
			slowLine:            "",
		}
	}
	return &LogIterator{
		LogIteratorOpt: iteratorOpt,
	}
}

type LogRecord struct {
	// content for a single normal log raw, empty when LogType is standardType
	Record string
	// content for a single slow log row, nil when LogType is slowQueryType
	SlowQueryRecord *SlowQueryLogRecord
	Host            string
	Component       string
	RecordID        string
	PackageID       string
	ClusterID       string
	InstanceID      string
	LogType         LogType
	FileName        string
}

func (record *LogRecord) IsSlowQueryLog() bool {
	return record.LogType == slowQueryType
}

type SlowQueryLogRecord struct {
	SlowQueryCol   map[string]int
	SlowQueryDatum []types.Datum
}

func (record *SlowQueryLogRecord) GetDatum(col string) types.Datum {
	idx, ok := record.SlowQueryCol[col]
	if !ok {
		return types.NewDatum(nil)
	}
	if len(record.SlowQueryDatum) < idx {
		return types.NewDatum(nil)
	}
	return record.SlowQueryDatum[idx]
}

// TODO integrate slow query retriever
type SlowQueryIterator struct {
	LogIteratorOpt

	slowLine string

	retriever           *slow_query.SlowQueryRetriever
	slowQueryCols       []string
	slowQueryColsLookUp map[string]int
	slowQueryRowCache   [][]types.Datum
	cacheIdx            int
	strBuilder          strings.Builder
}

func (li *SlowQueryIterator) Next() (any, bool) {
	if li.retriever == nil {
		return nil, false
	}
	// read from row cacheIdx
	logRecord, ok := li.getFromSlowQueryRawCache()
	if ok {
		return logRecord, true
	}
	ctx := context.Background()
	for {
		rows, err := li.retriever.Retrieve(ctx)
		if err != nil {
			log.Warn("retrieve slow log failed", zap.Error(err))
			continue
		}
		// if return rows are empty, this file reaches end.
		if len(rows) == 0 {
			_ = li.retriever.Close()
			break
		}
		// set row cacheIdx and cacheIdx id
		li.setSlowQueryRawCache(rows)
		return li.getFromSlowQueryRawCache()
	}

	return LogRecord{}, false
}

func (li *SlowQueryIterator) Len() (int, bool) {
	return -1, false
}

func (li *SlowQueryIterator) setSlowQueryRawCache(rows [][]types.Datum) {
	li.slowQueryRowCache = rows
	li.cacheIdx = 0
}

func (li *SlowQueryIterator) getFromSlowQueryRawCache() (LogRecord, bool) {
	if len(li.slowQueryRowCache) <= li.cacheIdx {
		return emptyLogRecord, false
	}
	row := li.slowQueryRowCache[li.cacheIdx]
	li.cacheIdx++
	li.strBuilder.Reset()
	// build raw message string.
	for i, col := range li.slowQueryCols {
		if i > 0 {
			li.strBuilder.WriteString(" ")
		}
		li.strBuilder.WriteString(variable.SlowLogRowPrefixStr)
		li.strBuilder.WriteString(col)
		li.strBuilder.WriteString(variable.SlowLogSpaceMarkStr)
		if col == variable.SlowLogTimeStr {
			li.strBuilder.WriteString(row[i].GetString())
		} else {
			// skip plan content.
			li.strBuilder.WriteString(fmt.Sprintf("%v", row[i].GetValue()))
		}
	}
	slowQueryLogRecord := &SlowQueryLogRecord{
		SlowQueryCol:   li.slowQueryColsLookUp,
		SlowQueryDatum: row,
	}
	// row must contain txn start ts
	txnStartTS := row[li.slowQueryColsLookUp[variable.SlowLogTxnStartTSStr]].GetUint64()
	return LogRecord{
		Record:          li.strBuilder.String(),
		SlowQueryRecord: slowQueryLogRecord,
		Host:            li.host,
		RecordID:        fmt.Sprintf("%s-%d", li.identityPrefix, txnStartTS),
		Component:       li.component,
		PackageID:       li.packageID,
		ClusterID:       li.clusterID,
		InstanceID:      li.instanceID,
		LogType:         li.logType,
	}, true
}