package slowquery

import (
	"bufio"
	"fmt"
	"io"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/errors"
)

const maxLineSize = 1073741824

// SlowLogReader is used to read SlowLog in groups from io.Reader.
// Different slow log entries are grouped into different []string,
// each string in []string represents one line.
type SlowLogReader struct {
	br *bufio.Reader
}

// NewSlowLogReader creates new SlowLogReader from io.Reader.
func NewSlowLogReader(r io.Reader) *SlowLogReader {
	return &SlowLogReader{
		br: bufio.NewReader(r),
	}
}

// ReadNext reads the next slow log entry from the underlying io.Reader.
// Each string in []string represents one line.
func (r *SlowLogReader) ReadNext() ([]string, error) {
	var lines []string
	for {
		line, err := readLine(r.br)
		if err != nil {
			if err == io.EOF {
				return lines, nil
			}
			return nil, err
		}
		lineStr := string(line)
		lines = append(lines, lineStr)
		if strings.HasSuffix(lineStr, ";") {
			if strings.HasPrefix(lineStr, "use") || strings.HasPrefix(lineStr, "# ") {
				continue
			}
			return lines, nil
		}
	}
}

// readLine is a general utility function for reading a complete line
// from bufio.Reader, the single-line size limit is specified by maxLineSize.
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()
		// 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, errors.Errorf("single line length exceeds limit: %v", maxLineSize)
		}
		if err != nil {
			return resByte, err
		}
	}
	return resByte, err
}

// ParseSlowLog parses structured SlowLog from []string.
func ParseSlowLog(lines []string) (*SlowLog, error) {
	sl := &SlowLog{}
	for _, line := range lines {
		if err := parseLine(sl, line); err != nil {
			return nil, err
		}
	}
	return sl, nil
}

var vectorLogPrefixRegex = regexp.MustCompile(`[^;]+;([^;]+);[^;]+;[^;]+;[^;]+`)

// parseLine parses a single line in a slow log entry and assigns the
// corresponding field to the SlowLog.
func parseLine(sl *SlowLog, line string) error {
	if len(line) == 0 {
		return nil
	}
	// Here is a fast path to determine the first line.
	// example: 2022-09-06T08:13:12.987865618Z;ip-... # Time: 2022-09-06T08:13:12.193975183Z
	if (line[0] == '2' && strings.Contains(line, "# Time: ")) || (strings.HasPrefix(line, "# Time:")) {
		arr := strings.SplitN(line, "# Time: ", 2)
		if len(arr) != 2 {
			return fmt.Errorf("invalid first line: %s", line)
		}
		if len(arr[0]) > 0 {
			vecArr := vectorLogPrefixRegex.FindStringSubmatch(arr[0])
			if len(vecArr) == 2 {
				sl.Instance = vecArr[1]
			}
		}
		var err error
		sl.Time, err = time.Parse(time.RFC3339Nano, arr[1])
		if err != nil {
			return err
		}
	} else if strings.HasPrefix(line, "# ") {
		line = line[len("# "):]
		if strings.HasPrefix(line, "User@Host: ") {
			v := line[len("User@Host: "):]
			arr := strings.SplitN(v, " @ ", 2)
			if len(arr) != 2 {
				return fmt.Errorf("invalid user and host line: %v", line)
			}
			userArr := strings.Split(arr[0], "[")
			if len(userArr) < 2 {
				return fmt.Errorf("invalid user format: %v, line: %v", arr[0], line)
			}
			sl.User = strings.TrimSpace(userArr[0])
			hostArr := strings.Split(arr[1], "[")
			if len(hostArr) < 2 {
				return fmt.Errorf("invalid host format: %v, line: %v", arr[1], line)
			}
			sl.Host = strings.TrimSpace(hostArr[0])
		} else if strings.HasPrefix(line, "Session_alias: ") {
			sl.SessionAlias = line[len("Session_alias: "):]
		} else if strings.HasPrefix(line, "Backoff_types: ") {
			sl.BackoffTypes = line[len("Backoff_types: "):]
		} else if strings.HasPrefix(line, "DB: ") {
			sl.DB = line[len("DB: "):]
		} else if strings.HasPrefix(line, "Index_names: ") {
			sl.IndexNames = line[len("Index_names: "):]
		} else if strings.HasPrefix(line, "Digest: ") {
			sl.Digest = line[len("Digest: "):]
		} else if strings.HasPrefix(line, "Stats: ") {
			sl.Stats = line[len("Stats: "):]
		} else if strings.HasPrefix(line, "Cop_proc_addr: ") {
			sl.CopProcAddr = line[len("Cop_proc_addr: "):]
		} else if strings.HasPrefix(line, "Cop_wait_addr: ") {
			sl.CopWaitAddr = line[len("Cop_wait_addr: "):]
		} else if strings.HasPrefix(line, "Warnings: ") {
			sl.Warnings = line[len("Warnings: "):]
		} else if strings.HasPrefix(line, "Cop_backoff_") {
			sl.BackoffDetail = line
		} else if strings.HasPrefix(line, "Plan: ") {
			sl.Plan = line[len("Plan: "):]
		} else if strings.HasPrefix(line, "Resource_group: ") {
			sl.ResourceGroup = line[len("Resource_group: "):]
		} else if strings.HasPrefix(line, "Plan_digest: ") {
			sl.PlanDigest = line[len("Plan_digest: "):]
		} else if strings.HasPrefix(line, "Binary_plan: ") {
			sl.BinaryPlan = line[len("Binary_plan: "):]
		} else if strings.HasPrefix(line, "Prev_stmt: ") {
			sl.PrevStmt = line[len("Prev_stmt: "):]
		} else if strings.HasPrefix(line, "Keyspace_name: ") {
			sl.KeyspaceName = line[len("Keyspace_name: "):]
		} else if strings.HasPrefix(line, "Keyspace_ID: ") {
			sl.KeyspaceID = line[len("Keyspace_ID: "):]
		} else if strings.HasPrefix(line, "Serverless_tenant_ID: ") {
			sl.ServerlessTenantID = line[len("Serverless_tenant_ID: "):]
		} else if strings.HasPrefix(line, "Serverless_project_ID: ") {
			sl.ServerlessProjectID = line[len("Serverless_project_ID: "):]
		} else if strings.HasPrefix(line, "Serverless_cluster_ID: ") {
			sl.ServerlessClusterID = line[len("Serverless_cluster_ID: "):]
		} else {
			ks, vs := splitByColon(line)
			for n := 0; n < len(ks); n++ {
				k, v := ks[n], vs[n]
				switch k {
				case "Txn_start_ts":
					if u, err := strconv.ParseUint(v, 10, 64); err == nil {
						sl.TxnStartTs = u
					}
				case "Conn_ID":
					_, _ = fmt.Sscanf(v, "%d", &sl.ConnID)
				case "Exec_retry_count":
					_, _ = fmt.Sscanf(v, "%d", &sl.ExecRetryCount)
				case "Exec_retry_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.ExecRetryTime)
				case "Query_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.QueryTime)
				case "Parse_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.ParseTime)
				case "Compile_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.CompileTime)
				case "Rewrite_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.RewriteTime)
				case "Preproc_subqueries":
					_, _ = fmt.Sscanf(v, "%d", &sl.PreprocSubqueries)
				case "Preproc_subqueries_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.PreprocSubqueriesTime)
				case "Optimize_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.OptimizeTime)
				case "Wait_TS":
					_, _ = fmt.Sscanf(v, "%f", &sl.WaitTs)
				case "Prewrite_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.PrewriteTime)
				case "Wait_prewrite_binlog_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.WaitPrewriteBinlogTime)
				case "Commit_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.CommitTime)
				case "Get_commit_ts_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.GetCommitTsTime)
				case "Commit_backoff_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.CommitBackoffTime)
				case "Resolve_lock_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.ResolveLockTime)
				case "Local_latch_wait_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.LocalLatchWaitTime)
				case "Write_keys":
					_, _ = fmt.Sscanf(v, "%d", &sl.WriteKeys)
				case "Write_size":
					_, _ = fmt.Sscanf(v, "%d", &sl.WriteSize)
				case "Prewrite_region":
					_, _ = fmt.Sscanf(v, "%d", &sl.PrewriteRegion)
				case "Txn_retry":
					_, _ = fmt.Sscanf(v, "%d", &sl.TxnRetry)
				case "Cop_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.CopTime)
				case "Process_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.ProcessTime)
				case "Wait_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.WaitTime)
				case "Backoff_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.BackoffTime)
				case "LockKeys_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.LockKeysTime)
				case "Request_count":
					_, _ = fmt.Sscanf(v, "%d", &sl.RequestCount)
				case "Total_keys":
					_, _ = fmt.Sscanf(v, "%d", &sl.TotalKeys)
				case "Process_keys":
					_, _ = fmt.Sscanf(v, "%d", &sl.ProcessKeys)
				case "Rocksdb_delete_skipped_count":
					_, _ = fmt.Sscanf(v, "%d", &sl.RocksdbDeleteSkippedCount)
				case "Rocksdb_key_skipped_count":
					_, _ = fmt.Sscanf(v, "%d", &sl.RocksdbKeySkippedCount)
				case "Rocksdb_block_cache_hit_count":
					_, _ = fmt.Sscanf(v, "%d", &sl.RocksdbBlockCacheHitCount)
				case "Rocksdb_block_read_count":
					_, _ = fmt.Sscanf(v, "%d", &sl.RocksdbBlockReadCount)
				case "Rocksdb_block_read_byte":
					_, _ = fmt.Sscanf(v, "%d", &sl.RocksdbBlockReadByte)
				case "Is_internal":
					sl.IsInternal = strToBool(v)
				case "Cop_proc_avg":
					_, _ = fmt.Sscanf(v, "%f", &sl.CopProcAvg)
				case "Cop_proc_p90":
					_, _ = fmt.Sscanf(v, "%f", &sl.CopProcP90)
				case "Cop_proc_max":
					_, _ = fmt.Sscanf(v, "%f", &sl.CopProcMax)
				case "Cop_wait_avg":
					_, _ = fmt.Sscanf(v, "%f", &sl.CopWaitAvg)
				case "Cop_wait_p90":
					_, _ = fmt.Sscanf(v, "%f", &sl.CopWaitP90)
				case "Cop_wait_max":
					_, _ = fmt.Sscanf(v, "%f", &sl.CopWaitMax)
				case "Mem_max":
					_, _ = fmt.Sscanf(v, "%d", &sl.MemMax)
				case "Disk_max":
					_, _ = fmt.Sscanf(v, "%d", &sl.DiskMax)
				case "KV_total":
					_, _ = fmt.Sscanf(v, "%f", &sl.KVTotal)
				case "PD_total":
					_, _ = fmt.Sscanf(v, "%f", &sl.PDTotal)
				case "Backoff_total":
					_, _ = fmt.Sscanf(v, "%f", &sl.BackoffTotal)
				case "Write_sql_response_total":
					_, _ = fmt.Sscanf(v, "%f", &sl.WriteSQLResponseTotal)
				case "Result_rows":
					_, _ = fmt.Sscanf(v, "%d", &sl.ResultRows)
				case "Prepared":
					sl.Prepared = strToBool(v)
				case "Succ":
					sl.Succ = strToBool(v)
				case "IsExplicitTxn":
					sl.IsExplicitTxn = strToBool(v)
				case "IsWriteCacheTable":
					sl.IsWriteCacheTable = strToBool(v)
				case "Plan_from_cache":
					sl.PlanFromCache = strToBool(v)
				case "Plan_from_binding":
					sl.PlanFromBinding = strToBool(v)
				case "Has_more_results":
					sl.HasMoreResults = strToBool(v)
				case "Request_unit_v2":
					_, _ = fmt.Sscanf(v, "%f", &sl.RU)
				case "Is_tiflash":
					sl.IsTiflash = strToBool(v)
				case "Tidb_cpu_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.TidbCPUTime)
				case "Tikv_cpu_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.TikvCPUTime)
				case "IA_remote_read_segment_count":
					_, _ = fmt.Sscanf(v, "%d", &sl.IARemoteReadSegmentCount)
				case "IA_remote_read_segment_size":
					_, _ = fmt.Sscanf(v, "%d", &sl.IARemoteReadSegmentSize)
				case "IA_remote_read_segment_wait_time":
					_, _ = fmt.Sscanf(v, "%f", &sl.IARemoteReadSegmentWaitTime)
				default:
					// omit
				}
			}
		}
	} else if strings.HasSuffix(line, ";") {
		if strings.HasPrefix(line, "use") {
			// `use DB` statements in the slow log is used to keep it be compatible with MySQL,
			// since we already get the current DB from the `# DB` field, we can ignore it here,
			// please see https://github.com/pingcap/tidb/issues/17846 for more details.
			return nil
		} else {
			sl.Query = line
		}
	} else {
		if strings.Contains(line, "following end of new file") {
			// This case:
			// ```
			// 2023-12-19T02...;slowlog tail: /var/log/auditlog/slowlog has been replaced; following end of new file
			// 2023-12-19T02...;slowlog # Time: 2023-12-19T02:35:33.586638997Z
			// # Txn_start_ts: 446418997401618680
			// # Query_time: 4.86387999
			// # Parse_time: 0
			// # Compile_time: 0.00052436
			// ...
			// ```
			return nil
		}
		return fmt.Errorf("invalid line: %s", line)
	}
	return nil
}

// kvSplitRegex is just for split "field: value field: value..."
var kvSplitRegex = regexp.MustCompile(`\w+: `)

// splitByColon split one line like "field: value field: value..."
func splitByColon(line string) (fields, values []string) {
	matches := kvSplitRegex.FindAllStringIndex(line, -1)
	fields = make([]string, 0, len(matches))
	values = make([]string, 0, len(matches))

	beg := 0
	end := 0
	for _, match := range matches {
		// trim ": "
		fields = append(fields, line[match[0]:match[1]-2])

		end = match[0]
		if beg != 0 {
			// trim " "
			values = append(values, line[beg:end-1])
		}
		beg = match[1]
	}

	if end != len(line) {
		// " " does not exist in the end
		values = append(values, line[beg:])
	}
	return fields, values
}

func strToBool(v string) bool {
	return v == "1" || strings.EqualFold(v, "true")
}