package logutil

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

	"github.com/stretchr/testify/assert"
)

func TestParsePingCAPLog(t *testing.T) {
	str := "2025-11-17T02:32:25.521558569Z;ip-172-21-93-194.ec2.internal;tidb10913508581581239630;db-tidb-0;tidb [2025/11/17 02:32:25.521 +00:00] [INFO] [meminfo.go:179] [\"use cgroup memory hook because TiDB is in the container 123\"]"

	r := NewLogReader(strings.NewReader(str), "tidb", 0)
	log, err := r.parseLine(str)
	assert.NoError(t, err)
	assert.Equal(t, str, log.RawLine)
	assert.Equal(t, "INFO", log.Level)
	assert.Equal(t, "meminfo.go:179", log.Position)
	assert.Equal(t, "db-tidb-0", log.Instance)
	assert.Equal(t, "[\"use cgroup memory hook because TiDB is in the container 123\"]", log.Message)
	assert.Equal(t, "[\"use cgroup memory hook because TiDB is in the container ?\"]", log.MsgDigest)
	assert.False(t, log.EventTime.IsZero())
	assert.Equal(t, int64(521000000), int64(log.EventTime.Nanosecond()))
	assert.Equal(t, log.EventTime.Unix(), log.Timestamp)
}

func TestParseTelemetryLog(t *testing.T) {
	str := `2025-11-17T02:32:25.521558569Z;ip-172-21-93-194.ec2.internal;tidb10913508581581239630;db-tidb-0;tidb [2025/12/11 07:59:28.182 +00:00] [INFO] [telemetry.go:66] [category=telemetry] ["telemetry data"="{\"multiSchemaChange\":{\"multi_schema_change_used\":0}}"]`

	r := NewLogReader(strings.NewReader(str), "tidb", 0)
	log, err := r.parseLine(str)
	assert.NoError(t, err)
	assert.Equal(t, str, log.RawLine)
	assert.Equal(t, "INFO", log.Level)
	assert.Equal(t, "telemetry.go:66", log.Position)
	assert.Equal(t, "db-tidb-0", log.Instance)
	assert.Contains(t, log.TelemetryStr, "\"multiSchemaChange\":{\"multi_schema_change_used\":0}")
	assert.True(t, json.Valid([]byte(log.TelemetryStr)))
}

func TestParseUnknownLog(t *testing.T) {
	str := "2025-08-20T02:57:09.744462240Z;ip-172-16-12-189.ec2.internal;tidb10030989866588211287;o11y-vector-agent-f974f544b-8gn9v;o11y-vector-agent 2025-08-20T02:57:09.744327Z  INFO source{component_kind=\"source\" component_id=topsql component_type=topsql}:topsql_schema_manager: vector::sources::topsql::schema_cache: Schema version changed: old=-1, new=57"

	r := NewLogReader(strings.NewReader(str), "o11y-vector-agent", 0)
	log, err := r.parseLine(str)
	assert.NoError(t, err)
	assert.Equal(t, str, log.RawLine)
	assert.Equal(t, "INFO", log.Level)
	assert.Equal(t, "o11y-vector-agent-f974f544b-8gn9v", log.Instance)

	str = "2024-05-31T03:08:00.961027654Z;gke-prod-seed-asia-n-gardener-node-po-6e2bbe4a-z6qq;cluster-service;aws-provider-upstream-7956956848-79gvg;aws-provider-upstream 2024-05-31T03:08:00.960Z\tINFO\tStarting EventSource\t{\"controller\": \"managed/domain.cloudsearch.aws.crossplane.io\", \"controllerGroup\": \"cloudsearch.aws.crossplane.io\", \"controllerKind\": \"Domain\", \"source\": \"kind source: *v1alpha1.Domain\"}"
	r = NewLogReader(strings.NewReader(str), "aws-provider-upstream", 0)
	log, err = r.parseLine(str)
	assert.NoError(t, err)
	assert.Equal(t, int64(1717124880), log.Timestamp)
	assert.Equal(t, "INFO", log.Level)
	assert.Equal(t, "aws-provider-upstream-7956956848-79gvg", log.Instance)
	assert.Equal(t, "Starting EventSource\t{\"controller\": \"managed/domain.cloudsearch.aws.crossplane.io\", \"controllerGroup\": \"cloudsearch.aws.crossplane.io\", \"controllerKind\": \"Domain\", \"source\": \"kind source: *v1alpha1.Domain\"}", log.Message)
}

func TestParseUnknownLogfmtLog(t *testing.T) {
	payload := `level=info ts=2022-10-16T21:17:44.085055757Z caller=operator.go:200 component=alertmanageroperator msg="connection established" cluster-version=v1.21.14-gke.2700`
	str := "2022-10-16T21:17:44.085464273Z;gke-prod-seed-asia-n-gardener-node-po-31add738-37yl;monitoring;prometheus-operator-b4bf9fb96-s6hd4;prometheus-operator " + payload

	r := NewLogReader(strings.NewReader(str), "prometheus-operator", 0)
	log, err := r.parseLine(str)
	assert.NoError(t, err)
	assert.Equal(t, str, log.RawLine)
	assert.Equal(t, int64(1665955064), log.Timestamp)
	assert.Equal(t, "INFO", log.Level)
	assert.Equal(t, "operator.go:200", log.Position)
	assert.Equal(t, payload, log.Message)
	assert.Equal(t, `level=info ts=?-?-?T?:?:?.?Z caller=operator.go:? component=alertmanageroperator msg="connection established" cluster-version=v?.?.?-gke.?`, log.MsgDigest)
	assert.Equal(t, "prometheus-operator-b4bf9fb96-s6hd4", log.Instance)
	assert.Equal(t, 85055757, log.EventTime.Nanosecond())
	assert.Equal(t, log.EventTime.Unix(), log.Timestamp)
}

func TestParseUnknownLogfmtLogWithoutTS(t *testing.T) {
	// Loki "request timings" logs may omit `ts=` and only carry `level=` + `msg=`.
	// The parser should fall back to the header timestamp (vector prefix), keeping nanos.
	payload := `level=info msg="request timings" insight=true source=loki_ruler`
	str := `2026-07-29T04:39:18.521894561Z;ip-10-1-98-170.us-west-2.compute.internal;observability;loki-ha-backend-0;loki ` + payload
	r := NewLogReader(strings.NewReader(str), "loki", 0)
	log, err := r.parseLine(str)
	assert.NoError(t, err)
	assert.Equal(t, "INFO", log.Level)
	assert.Equal(t, payload, log.Message)
	assert.Contains(t, log.Message, `msg="request timings"`)
	assert.Contains(t, log.Message, "source=loki_ruler")
	assert.NotZero(t, log.Timestamp)
	assert.Equal(t, 521894561, log.EventTime.Nanosecond())
	// `caller=` is optional; but missing ts must not trigger the UNKNOWN placeholder.
	assert.NotEqual(t, "unknown", log.Position)
	assert.NotEqual(t, "unknown", log.MsgDigest)
}

func TestParseUnknownLogfmtLogKeepsQueryPayload(t *testing.T) {
	payload := `level=info ts=2026-07-31T01:37:31.853856502Z caller=engine.go:274 component=ruler evaluation_mode=local org_id=fake msg="executing query" query="(max by (cluster_id)(count_over_time({container=\"tiflash\"}[1m]))) > 0)" query_hash=2603839996 type=instant`
	str := `2026-07-31T01:37:31.853856502Z;ip-10-1-98-170.us-west-2.compute.internal;infra-prod;loki-ha-backend-0;loki ` + payload
	r := NewLogReader(strings.NewReader(str), "loki", 0)
	log, err := r.parseLine(str)
	assert.NoError(t, err)
	assert.Equal(t, "INFO", log.Level)
	assert.Equal(t, "engine.go:274", log.Position)
	assert.Equal(t, payload, log.Message)
	assert.Contains(t, log.Message, "query_hash=2603839996")
	assert.Contains(t, log.Message, `msg="executing query"`)
	assert.Equal(t, 853856502, log.EventTime.Nanosecond())
}

func TestParseMalformedKeepsTimestampFromPrefixAndTruncatesMessage(t *testing.T) {
	// Missing `level=` should make this line malformed.
	// Fallback timestamp should come from the vector prefix, not time.Now().
	prefix := "2026-07-29T04:39:18.521894561Z"
	// Use a long message to ensure truncation.
	msg := "x" + strings.Repeat("y", 5000)
	str := `2026-07-29T04:39:18.521894561Z;ip-10-1-98-170.us-west-2.compute.internal;observability;loki-ha-backend-0;loki ts=2026-07-29T04:39:18.521894561Z msg="` + msg + `"`

	r := NewLogReader(strings.NewReader(str), "loki", 0)
	log, err := r.parseLine(str)
	assert.NoError(t, err)
	assert.Equal(t, "UNKNOWN", log.Level)
	assert.Equal(t, "unknown", log.Position)
	assert.Equal(t, "unknown", log.MsgDigest)
	assert.NotZero(t, log.Timestamp)
	assert.Equal(t, 1, r.ParseErrorCount())
	assert.NotEmpty(t, r.ParseErrorSample())

	// Ensure truncation (~4KB). We use a loose bound to avoid depending on exact suffix length.
	assert.LessOrEqual(t, len(log.Message), 4096+10)
	// Message should not be the literal "unknown".
	assert.NotEqual(t, "unknown", log.Message)

	// Timestamp should match prefix, not time.Now() at runtime, and keep nanos.
	tm, parseErr := time.Parse(time.RFC3339Nano, prefix)
	assert.NoError(t, parseErr)
	assert.Equal(t, tm.Unix(), log.Timestamp)
	assert.Equal(t, tm.Nanosecond(), log.EventTime.Nanosecond())
}

func TestMalformedWithoutTimeUsesPreviousPlusOneNs(t *testing.T) {
	good := `2026-07-31T01:37:31.853856502Z;ip-10-1-98-170.us-west-2.compute.internal;infra-prod;loki-ha-backend-0;loki level=info ts=2026-07-31T01:37:31.853856502Z caller=engine.go:274 msg="executing query"`
	// Continuation / broken line with no vector prefix and no parseable time.
	bad := `stack frame without timestamp or vector prefix`
	fileTs := int64(1785495600)

	r := NewLogReader(strings.NewReader(good+"\n"+bad+"\n"+bad), "loki", fileTs)

	first, err := r.ReadNext()
	assert.NoError(t, err)
	assert.Equal(t, "INFO", first.Level)
	assert.Equal(t, 853856502, first.EventTime.Nanosecond())

	second, err := r.ReadNext()
	assert.NoError(t, err)
	assert.Equal(t, "UNKNOWN", second.Level)
	assert.Equal(t, first.EventTime.Add(time.Nanosecond), second.EventTime)

	third, err := r.ReadNext()
	assert.NoError(t, err)
	assert.Equal(t, "UNKNOWN", third.Level)
	assert.Equal(t, second.EventTime.Add(time.Nanosecond), third.EventTime)

	assert.Equal(t, 2, r.ParseErrorCount())
	assert.Equal(t, bad, r.ParseErrorSample())
}

func TestMalformedFirstLineUsesFileTimestamp(t *testing.T) {
	bad := `not a parseable log line`
	fileTs := int64(1785495600)
	r := NewLogReader(strings.NewReader(bad), "loki", fileTs)
	log, err := r.ReadNext()
	assert.NoError(t, err)
	assert.Equal(t, "UNKNOWN", log.Level)
	assert.Equal(t, fileTs, log.Timestamp)
	assert.Equal(t, 1, r.ParseErrorCount())
}

func TestParseUnknownWrappedPingCAPLog(t *testing.T) {
	str := "2025-06-25T17:03:05.069224469Z;gke-prod-13728130891-backup-fa401ebb--b586e8a6-brqv;tidb10123293495242577103;backup-ab-10123293495242577103-2915737-wzpm2;backup I0625 17:03:05.069092       9 backup.go:334] [2025/06/25 17:03:05.068 +00:00] [INFO] [meminfo.go:179] [\"use cgroup memory hook because TiDB is in the container\"]"

	r := NewLogReader(strings.NewReader(str), "backup", 0)
	log, err := r.parseLine(str)
	assert.NoError(t, err)
	assert.Equal(t, "INFO", log.Level)
	assert.Equal(t, "meminfo.go:179", log.Position)
	assert.Equal(t, "[\"use cgroup memory hook because TiDB is in the container\"]", log.Message)
	assert.Equal(t, "backup-ab-10123293495242577103-2915737-wzpm2", log.Instance)
}

func BenchmarkParsePingCAPLog(b *testing.B) {
	str := "2025-11-17T02:32:25.521558569Z;ip-172-21-93-194.ec2.internal;tidb10913508581581239630;db-tidb-0;tidb [2025/11/17 02:32:25.521 +00:00] [INFO] [meminfo.go:179] [\"use cgroup memory hook because TiDB is in the container\"]"

	r := NewLogReader(strings.NewReader(str), "tidb", 0)
	for b.Loop() {
		_, err := r.parseLine(str)
		if err != nil {
			b.Fatal(err)
		}
	}
}