package logutil

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"regexp"
	"strconv"
	"strings"
	"time"
	"unicode"

	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/types"
)

type LogReader struct {
	br        *bufio.Reader
	component string
	// fallbackTimestampUnix is usually derived from the raw log object filename.
	// Used only when neither the line nor a prior in-file event time is available.
	fallbackTimestampUnix int64
	// lastEventTime tracks the previous line's event time within this file so
	// lines without a parseable time can advance by 1ns and keep order.
	lastEventTime  time.Time
	parseErrCount  int
	parseErrSample string
}

func NewLogReader(r io.Reader, component string, fallbackTimestampUnix int64) *LogReader {
	return &LogReader{
		br:                    bufio.NewReader(r),
		component:             component,
		fallbackTimestampUnix: fallbackTimestampUnix,
	}
}

// ParseErrorCount returns how many lines failed structured parsing in this reader.
func (r *LogReader) ParseErrorCount() int {
	return r.parseErrCount
}

// ParseErrorSample returns a truncated sample of the first parse failure, if any.
func (r *LogReader) ParseErrorSample() string {
	return r.parseErrSample
}

func (r *LogReader) ReadNext() (*types.Log, error) {
	line, err := readLine(r.br)
	if err != nil {
		return nil, err
	}
	if len(line) == 0 {
		return nil, nil
	}
	return r.parseLine(string(line))
}

func IsPingCAPComponent(component string) bool {
	switch component {
	case "pd", "tidb", "tikv", "tiflash", "ticdc", "backup", "tidb-lightning", "dm-worker", "dm-master":
		return true
	default:
		return false
	}
}

func (r *LogReader) parseLine(line string) (log *types.Log, err error) {
	if IsPingCAPComponent(r.component) {
		log, err = parsePingCAPLog(r.component, line)
	} else {
		log, err = parseUnknownLog(line)
	}
	if err == ErrMalformedLine {
		r.recordParseError(line)
		malformedMsg := truncateMalformedMessage(line, 4*1024)
		// Prefer time from the vector prefix when present; otherwise carry forward
		// the previous in-file event time (+1ns) so ordering is preserved.
		tm := eventTimeFromVectorPrefix(line)
		if tm.IsZero() {
			tm = r.nextFallbackEventTime()
		}
		log = &types.Log{
			RawLine:   line,
			Level:     "UNKNOWN",
			Position:  "unknown",
			Message:   malformedMsg,
			MsgDigest: "unknown",
		}
		log.SetEventTime(tm)
		r.lastEventTime = tm
		return log, nil
	}
	if err != nil {
		return nil, err
	}
	if log.EventTime.IsZero() {
		log.SetEventTime(r.nextFallbackEventTime())
	}
	r.lastEventTime = log.EventTime
	log.BuildDigest()
	return log, nil
}

func (r *LogReader) recordParseError(line string) {
	r.parseErrCount++
	if r.parseErrSample == "" {
		r.parseErrSample = truncateMalformedMessage(line, 512)
	}
}

// nextFallbackEventTime returns lastEventTime+1ns when available, else the file
// timestamp, else now. Used when the current line has no parseable time.
func (r *LogReader) nextFallbackEventTime() time.Time {
	if !r.lastEventTime.IsZero() {
		return r.lastEventTime.Add(time.Nanosecond)
	}
	if r.fallbackTimestampUnix != 0 {
		return time.Unix(r.fallbackTimestampUnix, 0).UTC()
	}
	return time.Now().UTC()
}

var (
	stackReg         = regexp.MustCompile(`\[stack="([^"]*)"\]`)
	ErrMalformedLine = errors.New("malformed line")
)

const maxMalformedMessageSuffix = "..."

func truncateMalformedMessage(line string, maxLen int) string {
	if maxLen <= 0 || len(line) <= maxLen {
		return line
	}
	// Keep suffix small; malformed lines can be arbitrarily long.
	if maxLen <= len(maxMalformedMessageSuffix) {
		return line[:maxLen]
	}
	return line[:maxLen-len(maxMalformedMessageSuffix)] + maxMalformedMessageSuffix
}

// eventTimeFromVectorPrefix extracts the vector prefix timestamp from
// "<ts>;<host>;<tenant>;<pod>;<component> <payload>".
func eventTimeFromVectorPrefix(line string) time.Time {
	// Prefix is before the first space.
	parts := strings.SplitN(line, " ", 2)
	if len(parts) != 2 {
		return time.Time{}
	}
	vectorInfos := strings.SplitN(parts[0], ";", 5)
	if len(vectorInfos) != 5 {
		return time.Time{}
	}
	tm, err := time.Parse(time.RFC3339Nano, vectorInfos[0])
	if err != nil {
		return time.Time{}
	}
	return tm
}

func GetInstanceInfoFromReader(line string) (string, error) {
	line = strings.TrimSpace(line)
	lines := strings.SplitN(line, " ", 2)
	if len(lines) != 2 {
		return "", ErrMalformedLine
	}
	vectorInfos := strings.SplitN(lines[0], ";", 5)
	if len(vectorInfos) != 5 {
		return "", ErrMalformedLine
	}
	return vectorInfos[3], nil
}

func parsePingCAPLog(component, line string) (*types.Log, error) {
	lines := strings.SplitN(line, " ", 2)
	if len(lines) != 2 {
		return nil, ErrMalformedLine
	}
	vectorInfos := strings.SplitN(lines[0], ";", 5)
	if len(vectorInfos) != 5 {
		return nil, ErrMalformedLine
	}

	pingcapLogLine := lines[1]
	if !strings.HasPrefix(pingcapLogLine, "[") {
		if idx := strings.Index(pingcapLogLine, " ["); idx != -1 && idx+1 < len(pingcapLogLine) && pingcapLogLine[idx+1] == '[' {
			pingcapLogLine = pingcapLogLine[idx+1:]
		}
	}
	log, err := parsePingCAPPayload(line, vectorInfos[3], pingcapLogLine)
	if err != nil {
		return nil, err
	}

	if processor, ok := logMessageProcessors[log.Position]; ok {
		log.Message = processor(line, log.Message)
	}

	if component == "tidb" && strings.Contains(log.Message, "telemetry data") {
		start := strings.IndexByte(log.Message, '{')
		end := strings.LastIndexByte(log.Message, '}')

		if start == -1 || end == -1 || end < start {
			log.TelemetryStr = ""
		} else {
			unescaped, err := strconv.Unquote(log.Message[start-1 : end+2])
			if err != nil {
				log.TelemetryStr = ""
			} else {
				log.TelemetryStr = unescaped
			}
		}
	}
	return log, nil
}

type jsonLine struct {
	Ts     string `json:"ts"`
	Time   string `json:"time"`
	Level  string `json:"level"`
	Logger string `json:"logger"`
}

func parseUnknownLog(line string) (*types.Log, error) {
	lines := strings.SplitN(line, " ", 2)
	if len(lines) != 2 {
		return nil, ErrMalformedLine
	}
	vectorInfos := strings.SplitN(lines[0], ";", 5)
	if len(vectorInfos) != 5 {
		return nil, ErrMalformedLine
	}
	// Some components (e.g. Loki) may omit `ts=` in the payload; in that case
	// we can still use the header timestamp (vectorInfos[0]) as a fallback.
	var fallbackTime time.Time
	if tm, err := time.Parse(time.RFC3339Nano, vectorInfos[0]); err == nil {
		fallbackTime = tm
	}
	lines[1] = strings.TrimSpace(lines[1])
	if len(lines[1]) == 0 {
		return nil, ErrMalformedLine
	}
	switch lines[1][0] {
	case '[':
		return parsePingCAPPayload(line, vectorInfos[3], lines[1])
	case '{':
		// JSON type log
		l := jsonLine{}
		if err := json.Unmarshal([]byte(lines[1]), &l); err != nil {
			return nil, ErrMalformedLine
		}
		rawTS := l.Time
		if rawTS == "" {
			rawTS = l.Ts
		}
		if rawTS == "" {
			return nil, ErrMalformedLine
		}
		tm, err := parseJSONLogTime(rawTS)
		if err != nil {
			return nil, ErrMalformedLine
		}
		log := &types.Log{
			RawLine:  line,
			Level:    strings.ToUpper(l.Level),
			Instance: vectorInfos[3],
			Position: l.Logger,
		}
		log.SetEventTime(tm)
		return log, nil
	case '2':
		log, err := parseTimePrefixedLog(line, vectorInfos[3], lines[1])
		if err == nil {
			return log, nil
		}
		return nil, ErrMalformedLine
	default:
		return parseLogfmtLog(line, vectorInfos[3], lines[1], fallbackTime)
	}
}

func parseJSONLogTime(raw string) (time.Time, error) {
	layouts := []string{
		time.RFC3339Nano,
		time.RFC3339,
		"2006-01-02T15:04:05Z",
		"2006-01-02T15:04:05",
	}
	for _, layout := range layouts {
		if tm, err := time.Parse(layout, raw); err == nil {
			return tm, nil
		}
	}
	return time.Time{}, ErrMalformedLine
}

func parseLogfmtLog(line, instance, payload string, fallbackTime time.Time) (*types.Log, error) {
	level, ok := extractLogfmtValue(payload, "level")
	if !ok {
		return nil, ErrMalformedLine
	}
	var tm time.Time
	if tsValue, ok := extractLogfmtValue(payload, "ts"); ok {
		parsed, err := time.Parse(time.RFC3339Nano, tsValue)
		if err != nil {
			return nil, ErrMalformedLine
		}
		tm = parsed
	} else if !fallbackTime.IsZero() {
		tm = fallbackTime
	} else {
		return nil, ErrMalformedLine
	}

	log := &types.Log{
		RawLine:  line,
		Instance: instance,
		Level:    strings.ToUpper(level),
		// Keep the full logfmt payload so variable labels (query, query_hash, etc.)
		// are preserved under the fixed schema's message column.
		Message: payload,
	}
	log.SetEventTime(tm)
	if caller, ok := extractLogfmtValue(payload, "caller"); ok {
		log.Position = caller
	}
	return log, nil
}

func parsePingCAPPayload(line, instance, payload string) (*types.Log, error) {
	m := strings.SplitN(payload, "] [", 4)
	if len(m) != 4 {
		return nil, ErrMalformedLine
	}
	m[0] = strings.TrimPrefix(m[0], "[")

	tm, err := time.Parse("2006/01/02 15:04:05.000 -07:00", m[0])
	if err != nil {
		return nil, ErrMalformedLine
	}

	log := &types.Log{
		RawLine:  line,
		Level:    strings.ToUpper(m[1]),
		Position: m[2],
		Message:  "[" + m[3],
		Instance: instance,
	}
	log.SetEventTime(tm)
	return log, nil
}

func parseTimePrefixedLog(line, instance, payload string) (*types.Log, error) {
	parts := strings.Fields(payload)
	if len(parts) < 2 {
		return nil, ErrMalformedLine
	}

	tm, err := time.Parse(time.RFC3339Nano, parts[0])
	if err != nil {
		return nil, ErrMalformedLine
	}

	log := &types.Log{
		RawLine:  line,
		Instance: instance,
		Level:    strings.ToUpper(parts[1]),
	}
	log.SetEventTime(tm)
	if len(parts) > 2 {
		message := strings.TrimLeftFunc(payload, unicode.IsSpace)
		message = strings.TrimPrefix(message, parts[0])
		message = strings.TrimLeftFunc(message, unicode.IsSpace)
		message = strings.TrimPrefix(message, parts[1])
		log.Message = strings.TrimSpace(message)
	}
	return log, nil
}

func extractLogfmtValue(payload, key string) (string, bool) {
	pattern := key + "="
	searchFrom := 0
	for {
		idx := strings.Index(payload[searchFrom:], pattern)
		if idx == -1 {
			return "", false
		}
		idx += searchFrom
		if idx > 0 && !unicode.IsSpace(rune(payload[idx-1])) {
			searchFrom = idx + len(pattern)
			continue
		}

		valueStart := idx + len(pattern)
		if valueStart >= len(payload) {
			return "", false
		}
		if payload[valueStart] == '"' {
			valueStart++
			valueEnd := strings.IndexByte(payload[valueStart:], '"')
			if valueEnd == -1 {
				return "", false
			}
			return payload[valueStart : valueStart+valueEnd], true
		}

		valueEnd := strings.IndexFunc(payload[valueStart:], unicode.IsSpace)
		if valueEnd == -1 {
			return payload[valueStart:], true
		}
		return payload[valueStart : valueStart+valueEnd], true
	}
}

type LogMessageProcessor func(fullLine string, originalMessage string) string

var logMessageProcessors = make(map[string]LogMessageProcessor)

func RegisterLogMessageProcessor(position string, processor LogMessageProcessor) {
	logMessageProcessors[position] = processor
}

func InitLogProcessors() {
	RegisterLogMessageProcessor("terror.go:315", func(fullLine string, originalMessage string) string {
		match := stackReg.FindStringSubmatch(fullLine)
		if len(match) > 1 {
			stackContent := match[1]
			return originalMessage + "[stack=\"" + stackContent + "\"]"
		}
		return originalMessage
	})
}

const maxLineSize = 1073741824

func readLine(reader *bufio.Reader) ([]byte, error) {
	var resByte []byte
	lineByte, isPrefix, err := reader.ReadLine()
	if isPrefix {
		// Need to read more data.
		resByte = make([]byte, len(lineByte), len(lineByte)*2)
	} else {
		resByte = make([]byte, len(lineByte))
	}
	// Use copy here to avoid shallow copy problem.
	copy(resByte, lineByte)
	if err != nil {
		return resByte, err
	}
	var tempLine []byte
	for isPrefix {
		tempLine, isPrefix, err = reader.ReadLine()
		if len(tempLine) > 0 {
			// nolint:makezero
			resByte = append(resByte, tempLine...)
		}
		// Use the max value of max_allowed_packet to check the single line length.
		if len(resByte) > maxLineSize {
			return resByte, fmt.Errorf("single line length exceeds limit: %v", maxLineSize)
		}
		if err != nil {
			return resByte, err
		}
	}
	return resByte, err
}