package main

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

	"github.com/segmentio/kafka-go"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/history"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
)

func TestProcessHistoryMessageUsesUncancelledContextForAppendAndCommit(t *testing.T) {
	parent, cancel := context.WithCancel(context.Background())
	cancel()

	env := envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolJT808,
		MessageID:    "0x0200",
		Phone:        "13307795425",
		EventTimeMS:  1782918600000,
		ReceivedAtMS: 1782918600000,
		ParseStatus:  envelope.ParseOK,
	}
	payload, err := json.Marshal(env)
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{}
	committer := &contextCheckingHistoryCommitter{}

	processHistoryMessage(parent, discardHistoryLogger{}, nil, appender, committer, kafka.Message{Value: payload})

	if appender.ctxErr != nil {
		t.Fatalf("appender saw cancelled context: %v", appender.ctxErr)
	}
	if committer.ctxErr != nil {
		t.Fatalf("committer saw cancelled context: %v", committer.ctxErr)
	}
	if appender.count != 1 || committer.count != 1 {
		t.Fatalf("appends=%d commits=%d", appender.count, committer.count)
	}
}

func TestProcessHistoryMessageRecordsMetrics(t *testing.T) {
	env := envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		VIN:          "VIN001",
		MessageID:    "0x02",
		ReceivedAtMS: time.Now().Add(-time.Second).UnixMilli(),
		ParseStatus:  envelope.ParseOK,
		ParsedFields: map[string]any{
			"gb32960.vehicle.speed_kmh": "42",
		},
	}
	payload, err := json.Marshal(env)
	if err != nil {
		t.Fatal(err)
	}
	registry := metrics.NewRegistry()

	processHistoryMessage(
		context.Background(),
		discardHistoryLogger{},
		registry,
		&contextCheckingHistoryAppender{},
		&contextCheckingHistoryCommitter{},
		kafka.Message{Topic: "vehicle.raw.gb32960.v1", Partition: 1, Offset: 20, HighWaterMark: 23, Value: payload},
	)

	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_kafka_messages_total{status="received",topic="vehicle.raw.gb32960.v1"} 1`,
		`vehicle_history_last_message_unix_seconds{status="received",topic="vehicle.raw.gb32960.v1"} `,
		`vehicle_history_writes_total{status="ok",topic="vehicle.raw.gb32960.v1"} 1`,
		`vehicle_history_last_write_unix_seconds{status="ok",topic="vehicle.raw.gb32960.v1"} `,
		`vehicle_history_write_e2e_duration_ms_histogram_count{topic="vehicle.raw.gb32960.v1"} 1`,
		`vehicle_history_write_e2e_recent_p99_ms{topic="vehicle.raw.gb32960.v1"} `,
		`vehicle_history_write_e2e_recent_samples{topic="vehicle.raw.gb32960.v1"} `,
		`vehicle_history_last_write_e2e_unix_seconds{topic="vehicle.raw.gb32960.v1"} `,
		`vehicle_history_kafka_commits_total{status="ok",topic="vehicle.raw.gb32960.v1"} 1`,
		`vehicle_history_last_commit_unix_seconds{status="ok",topic="vehicle.raw.gb32960.v1"} `,
		`vehicle_history_kafka_lag{partition="1",topic="vehicle.raw.gb32960.v1"} 2`,
		`vehicle_history_parsed_fields_total{protocol="GB32960",status="present",topic="vehicle.raw.gb32960.v1"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("metrics missing %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryMessageSkipsExplicitNonRawEventKind(t *testing.T) {
	env := envelope.FrameEnvelope{
		Protocol:  envelope.ProtocolJT808,
		EventKind: envelope.EventKindFields,
		MessageID: "0x0200",
		VIN:       "VIN001",
	}
	payload, err := json.Marshal(env)
	if err != nil {
		t.Fatal(err)
	}
	registry := metrics.NewRegistry()
	appender := &contextCheckingHistoryAppender{}
	committer := &contextCheckingHistoryCommitter{}

	processHistoryMessage(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		kafka.Message{Topic: "vehicle.raw.go.jt808.v1", Partition: 1, Offset: 20, HighWaterMark: 21, Value: payload},
	)

	if appender.count != 0 || appender.batchCount != 0 {
		t.Fatalf("appender count=%d batch=%d, want no TDengine writes", appender.count, appender.batchCount)
	}
	if committer.count != 1 || committer.messageCount != 1 {
		t.Fatalf("commits=%d messages=%d, want 1/1", committer.count, committer.messageCount)
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_kafka_messages_total{status="event_kind_mismatch",topic="vehicle.raw.go.jt808.v1"} 1`,
		`vehicle_history_kafka_commits_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("mismatch metric missing %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryMessageRecordsMissingParsedFieldsForRealtimeFrame(t *testing.T) {
	env := envelope.FrameEnvelope{
		Protocol:    envelope.ProtocolJT808,
		MessageID:   "0x0200",
		Phone:       "13307795425",
		ParseStatus: envelope.ParseOK,
		Parsed:      map[string]any{"location": map[string]any{"speed_kmh": 30}},
	}
	payload, err := json.Marshal(env)
	if err != nil {
		t.Fatal(err)
	}
	registry := metrics.NewRegistry()

	processHistoryMessage(
		context.Background(),
		discardHistoryLogger{},
		registry,
		&contextCheckingHistoryAppender{},
		&contextCheckingHistoryCommitter{},
		kafka.Message{Topic: "vehicle.raw.go.jt808.v1", Value: payload},
	)

	text := registry.Render()
	if !strings.Contains(text, `vehicle_history_parsed_fields_total{protocol="JT808",status="missing",topic="vehicle.raw.go.jt808.v1"} 1`) {
		t.Fatalf("missing parsed field metric not recorded:\n%s", text)
	}
}

func TestRecordHistoryWriteE2EDurationSkipsMissingReceivedAt(t *testing.T) {
	registry := metrics.NewRegistry()

	recordHistoryWriteE2EDuration(registry, kafka.Message{Topic: "vehicle.raw.go.jt808.v1"}, envelope.FrameEnvelope{})

	text := registry.Render()
	if strings.Contains(text, "vehicle_history_write_e2e_duration_ms_histogram") || strings.Contains(text, "vehicle_history_write_e2e_recent") {
		t.Fatalf("unexpected e2e metric without received_at:\n%s", text)
	}
}

func TestProcessHistoryMessageSkipsParsedFieldMetricForNonRealtimeFrame(t *testing.T) {
	env := envelope.FrameEnvelope{
		Protocol:    envelope.ProtocolJT808,
		MessageID:   "0x0102",
		Phone:       "13307795425",
		ParseStatus: envelope.ParseOK,
	}
	payload, err := json.Marshal(env)
	if err != nil {
		t.Fatal(err)
	}
	registry := metrics.NewRegistry()

	processHistoryMessage(
		context.Background(),
		discardHistoryLogger{},
		registry,
		&contextCheckingHistoryAppender{},
		&contextCheckingHistoryCommitter{},
		kafka.Message{Topic: "vehicle.raw.go.jt808.v1", Value: payload},
	)

	text := registry.Render()
	if strings.Contains(text, "vehicle_history_parsed_fields_total") {
		t.Fatalf("non-realtime frame should not record parsed field quality metric:\n%s", text)
	}
}

func TestProcessHistoryBatchAppendsAllBeforeCommit(t *testing.T) {
	first := envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolGB32960,
		VIN:          "VIN001",
		MessageID:    "0x02",
		ReceivedAtMS: time.Now().Add(-time.Second).UnixMilli(),
		ParseStatus:  envelope.ParseOK,
		ParsedFields: map[string]any{
			"gb32960.vehicle.speed_kmh": "42",
		},
	}
	second := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200", ReceivedAtMS: time.Now().Add(-time.Second).UnixMilli(), ParseStatus: envelope.ParseOK, Parsed: map[string]any{"location": map[string]any{"speed_kmh": 30}}}
	firstPayload, err := json.Marshal(first)
	if err != nil {
		t.Fatal(err)
	}
	secondPayload, err := json.Marshal(second)
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{}
	committer := &contextCheckingHistoryCommitter{}
	registry := metrics.NewRegistry()

	processHistoryBatch(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		[]kafka.Message{
			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, HighWaterMark: 13, Value: firstPayload},
			{Topic: "vehicle.raw.go.jt808.v1", Partition: 2, Offset: 20, HighWaterMark: 21, Value: secondPayload},
		},
	)

	if appender.batchCount != 1 {
		t.Fatalf("batch appends = %d, want 1", appender.batchCount)
	}
	if got := len(appender.batch); got != 2 {
		t.Fatalf("batch size = %d, want 2", got)
	}
	if committer.count != 1 {
		t.Fatalf("commit calls = %d, want 1", committer.count)
	}
	if committer.messageCount != 2 {
		t.Fatalf("committed messages = %d, want 2", committer.messageCount)
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_batch_rows_total{status="ok"} 2`,
		`vehicle_history_batch_flush_total{status="ok"} 1`,
		`vehicle_history_batch_flush_duration_ms_histogram_bucket{le="+Inf",status="ok"} 1`,
		`vehicle_history_batch_flush_duration_ms_histogram_count{status="ok"} 1`,
		`vehicle_history_batch_flush_duration_ms_histogram_sum{status="ok"}`,
		`vehicle_history_write_e2e_duration_ms_histogram_count{topic="vehicle.raw.go.gb32960.v1"} 1`,
		`vehicle_history_write_e2e_duration_ms_histogram_count{topic="vehicle.raw.go.jt808.v1"} 1`,
		`vehicle_history_write_e2e_recent_p99_ms{topic="vehicle.raw.go.gb32960.v1"} `,
		`vehicle_history_write_e2e_recent_p99_ms{topic="vehicle.raw.go.jt808.v1"} `,
		`vehicle_history_parsed_fields_total{protocol="GB32960",status="present",topic="vehicle.raw.go.gb32960.v1"} 1`,
		`vehicle_history_parsed_fields_total{protocol="JT808",status="missing",topic="vehicle.raw.go.jt808.v1"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("batch metric missing %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryBatchRecordsDerivedLocationStatusesPerEnvelope(t *testing.T) {
	nonRealtime := envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolJT808,
		VIN:          "VIN001",
		MessageID:    "0x0100",
		ReceivedAtMS: time.Now().Add(-time.Second).UnixMilli(),
		ParseStatus:  envelope.ParseOK,
		ParsedFields: map[string]any{
			"jt808.location.longitude": 121.1,
			"jt808.location.latitude":  30.2,
		},
	}
	realtimeLocation := envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolJT808,
		VIN:          "VIN002",
		MessageID:    "0x0200",
		ReceivedAtMS: time.Now().Add(-time.Second).UnixMilli(),
		ParseStatus:  envelope.ParseOK,
		ParsedFields: map[string]any{
			"jt808.location.longitude": 121.2,
			"jt808.location.latitude":  30.3,
		},
	}
	nonRealtimePayload, err := json.Marshal(nonRealtime)
	if err != nil {
		t.Fatal(err)
	}
	realtimePayload, err := json.Marshal(realtimeLocation)
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{
		result: history.AppendResult{RawRows: 2, LocationRows: 1},
	}
	committer := &contextCheckingHistoryCommitter{}
	registry := metrics.NewRegistry()

	processHistoryBatch(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		[]kafka.Message{
			{Topic: "vehicle.raw.go.jt808.v1", Partition: 1, Offset: 10, HighWaterMark: 12, Value: nonRealtimePayload},
			{Topic: "vehicle.raw.go.jt808.v1", Partition: 1, Offset: 11, HighWaterMark: 12, Value: realtimePayload},
		},
	)

	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_location_writes_total{status="skipped_non_realtime",topic="vehicle.raw.go.jt808.v1"} 1`,
		`vehicle_history_location_writes_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 1`,
		`vehicle_history_location_rows_total{status="ok"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("derived location status metric missing %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryBatchSkipsExplicitNonRawEventKindAndWritesValidRows(t *testing.T) {
	mismatched := envelope.FrameEnvelope{
		Protocol:  envelope.ProtocolJT808,
		EventKind: envelope.EventKindFields,
		MessageID: "0x0200",
		VIN:       "VIN_BAD",
	}
	valid := envelope.FrameEnvelope{
		Protocol:     envelope.ProtocolJT808,
		EventKind:    envelope.EventKindRaw,
		MessageID:    "0x0200",
		VIN:          "VIN001",
		ReceivedAtMS: time.Now().Add(-time.Second).UnixMilli(),
		ParseStatus:  envelope.ParseOK,
	}
	mismatchedPayload, err := mismatched.MarshalJSONBytes()
	if err != nil {
		t.Fatal(err)
	}
	validPayload, err := valid.MarshalJSONBytes()
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{}
	committer := &contextCheckingHistoryCommitter{}
	registry := metrics.NewRegistry()

	processHistoryBatch(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		[]kafka.Message{
			{Topic: "vehicle.raw.go.jt808.v1", Partition: 1, Offset: 10, HighWaterMark: 12, Value: mismatchedPayload},
			{Topic: "vehicle.raw.go.jt808.v1", Partition: 1, Offset: 11, HighWaterMark: 12, Value: validPayload},
		},
	)

	if appender.batchCount != 1 || len(appender.batch) != 1 {
		t.Fatalf("batch appends=%d rows=%d, want only valid row", appender.batchCount, len(appender.batch))
	}
	if committer.count != 1 || committer.messageCount != 2 {
		t.Fatalf("commit calls=%d messages=%d, want 1/2", committer.count, committer.messageCount)
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_kafka_messages_total{status="event_kind_mismatch",topic="vehicle.raw.go.jt808.v1"} 1`,
		`vehicle_history_writes_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 1`,
		`vehicle_history_kafka_commits_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 2`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("batch mismatch metric missing %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryBatchExposesPendingBatchDepthDuringAppend(t *testing.T) {
	first := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"}
	second := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"}
	firstPayload, err := json.Marshal(first)
	if err != nil {
		t.Fatal(err)
	}
	secondPayload, err := json.Marshal(second)
	if err != nil {
		t.Fatal(err)
	}
	registry := metrics.NewRegistry()
	appender := &contextCheckingHistoryAppender{
		onAppendBatch: func() {
			text := registry.Render()
			for _, want := range []string{
				`vehicle_history_batch_pending_messages 2`,
				`vehicle_history_batch_pending_rows 2`,
			} {
				if !strings.Contains(text, want) {
					t.Fatalf("pending batch metric missing %s while appending:\n%s", want, text)
				}
			}
		},
	}

	processHistoryBatch(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		&contextCheckingHistoryCommitter{},
		[]kafka.Message{
			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, HighWaterMark: 13, Value: firstPayload},
			{Topic: "vehicle.raw.go.jt808.v1", Partition: 2, Offset: 20, HighWaterMark: 21, Value: secondPayload},
		},
	)

	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_batch_pending_messages 0`,
		`vehicle_history_batch_pending_rows 0`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("pending batch metric should reset after append, missing %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryBatchKeepsPendingGaugePerWorker(t *testing.T) {
	env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"}
	payload, err := json.Marshal(env)
	if err != nil {
		t.Fatal(err)
	}
	registry := metrics.NewRegistry()

	processHistoryBatchForWorker(
		context.Background(),
		discardHistoryLogger{},
		registry,
		&contextCheckingHistoryAppender{},
		&contextCheckingHistoryCommitter{},
		[]kafka.Message{{Topic: "vehicle.raw.go.jt808.v1", Partition: 1, Offset: 10, Value: payload}},
		metrics.Labels{"worker": "2"},
	)

	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_batch_pending_messages{worker="2"} 0`,
		`vehicle_history_batch_pending_rows{worker="2"} 0`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("worker pending gauge missing or not reset, want %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryBatchSkipsInvalidJSONWithoutCountingItAsWriteOK(t *testing.T) {
	valid := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200", ParseStatus: envelope.ParseOK}
	validPayload, err := json.Marshal(valid)
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{}
	committer := &contextCheckingHistoryCommitter{}
	registry := metrics.NewRegistry()

	processHistoryBatch(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		[]kafka.Message{
			{Topic: "vehicle.raw.go.jt808.v1", Partition: 1, Offset: 10, HighWaterMark: 13, Value: []byte(`{bad json`)},
			{Topic: "vehicle.raw.go.jt808.v1", Partition: 1, Offset: 11, HighWaterMark: 13, Value: validPayload},
		},
	)

	if appender.batchCount != 1 {
		t.Fatalf("batch appends = %d, want 1", appender.batchCount)
	}
	if got := len(appender.batch); got != 1 {
		t.Fatalf("batch size = %d, want only valid envelope", got)
	}
	if committer.messageCount != 2 {
		t.Fatalf("committed messages = %d, want valid and invalid committed", committer.messageCount)
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_kafka_messages_total{status="invalid_json",topic="vehicle.raw.go.jt808.v1"} 1`,
		`vehicle_history_writes_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 1`,
		`vehicle_history_kafka_commits_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 2`,
		`vehicle_history_batch_rows_total{status="ok"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("invalid json batch metric missing %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryBatchDoesNotCountInvalidJSONAsWriteError(t *testing.T) {
	valid := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"}
	payload, err := json.Marshal(valid)
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{err: errTestHistoryAppend}
	registry := metrics.NewRegistry()
	committer := &contextCheckingHistoryCommitter{}

	processHistoryBatch(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		[]kafka.Message{
			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, HighWaterMark: 12, Value: []byte(`{bad json`)},
			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 11, HighWaterMark: 12, Value: payload},
		},
	)

	if committer.messageCount != 1 {
		t.Fatalf("committed messages = %d, want invalid prefix only", committer.messageCount)
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_kafka_messages_total{status="invalid_json",topic="vehicle.raw.go.gb32960.v1"} 1`,
		`vehicle_history_writes_total{status="error",topic="vehicle.raw.go.gb32960.v1"} 1`,
		`vehicle_history_kafka_commits_total{status="ok",topic="vehicle.raw.go.gb32960.v1"} 1`,
		`vehicle_history_batch_rows_total{status="error"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("invalid json write error metric missing %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryBatchDoesNotCommitInvalidJSONAfterFailedValidOffset(t *testing.T) {
	valid := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"}
	payload, err := json.Marshal(valid)
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{err: errTestHistoryAppend}
	committer := &contextCheckingHistoryCommitter{}

	processHistoryBatch(
		context.Background(),
		discardHistoryLogger{},
		metrics.NewRegistry(),
		appender,
		committer,
		[]kafka.Message{
			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, HighWaterMark: 12, Value: payload},
			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 11, HighWaterMark: 12, Value: []byte(`{bad json`)},
		},
	)

	if committer.count != 0 {
		t.Fatalf("commit calls = %d, want 0 because valid offset failed before invalid json", committer.count)
	}
}

func TestProcessedHistoryPrefixMessagesByPartitionStopsAtOffsetGap(t *testing.T) {
	items := []historyBatchItem{
		{message: kafka.Message{Topic: "vehicle.raw.go.jt808.v1", Partition: 1, Offset: 10}, processed: true},
		{message: kafka.Message{Topic: "vehicle.raw.go.jt808.v1", Partition: 1, Offset: 12}, processed: true},
	}

	got := processedHistoryPrefixMessagesByPartition(items)
	if len(got) != 1 || got[0].Offset != 10 {
		t.Fatalf("processed prefix = %#v, want only offset 10", got)
	}
}

func TestProcessHistoryBatchFallsBackToSinglesAfterNonTransientBatchFailure(t *testing.T) {
	first := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"}
	second := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"}
	firstPayload, err := json.Marshal(first)
	if err != nil {
		t.Fatal(err)
	}
	secondPayload, err := json.Marshal(second)
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{
		batchErr: errors.New("syntax error in batch insert"),
		result:   history.AppendResult{RawRows: 1, LocationRows: 1},
	}
	committer := &contextCheckingHistoryCommitter{}
	registry := metrics.NewRegistry()

	processHistoryBatch(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		[]kafka.Message{
			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, Value: []byte(`{bad json`)},
			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 11, Value: firstPayload},
			{Topic: "vehicle.raw.go.jt808.v1", Partition: 2, Offset: 20, Value: secondPayload},
		},
	)

	if appender.batchCount != 1 {
		t.Fatalf("batch attempts = %d, want 1", appender.batchCount)
	}
	if appender.count != 2 {
		t.Fatalf("single fallback attempts = %d, want 2", appender.count)
	}
	if committer.messageCount != 3 {
		t.Fatalf("committed messages = %d, want invalid and fallback-success messages", committer.messageCount)
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_batch_flush_total{status="error"} 1`,
		`vehicle_history_batch_fallback_total{status="ok"} 1`,
		`vehicle_history_writes_total{status="ok",topic="vehicle.raw.go.gb32960.v1"} 1`,
		`vehicle_history_writes_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 1`,
		`vehicle_history_kafka_commits_total{status="ok",topic="vehicle.raw.go.gb32960.v1"} 2`,
		`vehicle_history_kafka_commits_total{status="ok",topic="vehicle.raw.go.jt808.v1"} 1`,
		`vehicle_history_batch_fallback_duration_ms_histogram_count{status="ok"} 2`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("fallback metric missing %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryBatchFallbackCommitsOnlyProcessedPrefixWhenSingleFails(t *testing.T) {
	success := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"}
	failed := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN002", MessageID: "0x02"}
	successPayload, err := json.Marshal(success)
	if err != nil {
		t.Fatal(err)
	}
	failedPayload, err := json.Marshal(failed)
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{
		batchErr:          errors.New("syntax error in batch insert"),
		singleErr:         errTestHistoryAppend,
		failSingleOnCount: 2,
	}
	committer := &contextCheckingHistoryCommitter{}
	registry := metrics.NewRegistry()

	outcome := processHistoryBatch(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		[]kafka.Message{
			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, Value: []byte(`{bad json`)},
			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 11, Value: successPayload},
			{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 12, Value: failedPayload},
		},
	)

	if appender.batchCount != 1 {
		t.Fatalf("batch attempts = %d, want 1", appender.batchCount)
	}
	if appender.count != 2 {
		t.Fatalf("single fallback attempts = %d, want stop after failed second single", appender.count)
	}
	if committer.messageCount != 2 {
		t.Fatalf("committed messages = %d, want invalid and first successful single only", committer.messageCount)
	}
	if got := committedOffsets(outcome.retryMessages); len(got) != 1 || got[0] != 12 {
		t.Fatalf("retry offsets = %#v, want [12]", got)
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_batch_fallback_total{status="error"} 1`,
		`vehicle_history_writes_total{status="ok",topic="vehicle.raw.go.gb32960.v1"} 1`,
		`vehicle_history_writes_total{status="error",topic="vehicle.raw.go.gb32960.v1"} 1`,
		`vehicle_history_kafka_commits_total{status="ok",topic="vehicle.raw.go.gb32960.v1"} 2`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("fallback failure metric missing %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryBatchReliablyRetriesFailedSuffix(t *testing.T) {
	first := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"}
	second := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN002", MessageID: "0x02"}
	third := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN003", MessageID: "0x02"}
	firstPayload, err := json.Marshal(first)
	if err != nil {
		t.Fatal(err)
	}
	secondPayload, err := json.Marshal(second)
	if err != nil {
		t.Fatal(err)
	}
	thirdPayload, err := json.Marshal(third)
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{
		batchErr:          errors.New("syntax error in batch insert"),
		singleErr:         errTestHistoryAppend,
		failSingleOnCount: 2,
	}
	committer := &contextCheckingHistoryCommitter{}
	registry := metrics.NewRegistry()
	messages := []kafka.Message{
		{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, Value: firstPayload},
		{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 11, Value: secondPayload},
		{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 12, Value: thirdPayload},
	}

	processHistoryBatchReliably(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		messages,
		time.Nanosecond,
	)

	if appender.batchCount != 2 {
		t.Fatalf("batch attempts = %d, want initial batch and failed suffix retry", appender.batchCount)
	}
	if appender.count != 4 {
		t.Fatalf("single fallback attempts = %d, want 4", appender.count)
	}
	if got := committedOffsets(committer.messages); len(got) != 3 || got[0] != 10 || got[1] != 11 || got[2] != 12 {
		t.Fatalf("committed offsets = %#v, want [10 11 12]", got)
	}
	text := registry.Render()
	if !strings.Contains(text, `vehicle_history_batch_retries_total{reason="write_error"} 1`) {
		t.Fatalf("write retry metric missing:\n%s", text)
	}
}

func TestProcessHistoryBatchReliablyRetriesCommitWithoutReappending(t *testing.T) {
	env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"}
	payload, err := json.Marshal(env)
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{}
	committer := &contextCheckingHistoryCommitter{err: errors.New("commit failed"), failOnCount: 1}
	registry := metrics.NewRegistry()
	messages := []kafka.Message{
		{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, Value: payload},
		{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 11, Value: payload},
	}

	processHistoryBatchReliably(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		messages,
		time.Nanosecond,
	)

	if appender.batchCount != 1 {
		t.Fatalf("batch attempts = %d, want 1 without replay after commit failure", appender.batchCount)
	}
	if committer.count != 2 {
		t.Fatalf("commit attempts = %d, want initial failure and one retry", committer.count)
	}
	text := registry.Render()
	if !strings.Contains(text, `vehicle_history_batch_retries_total{reason="commit_error"} 1`) {
		t.Fatalf("commit retry metric missing:\n%s", text)
	}
}

func TestProcessHistoryBatchDoesNotCommitWhenAppendFails(t *testing.T) {
	env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"}
	payload, err := json.Marshal(env)
	if err != nil {
		t.Fatal(err)
	}
	appender := &contextCheckingHistoryAppender{err: errTestHistoryAppend}
	committer := &contextCheckingHistoryCommitter{}
	registry := metrics.NewRegistry()

	processHistoryBatch(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		[]kafka.Message{{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, HighWaterMark: 11, Value: payload}},
	)

	if committer.count != 0 {
		t.Fatalf("commit calls = %d, want 0", committer.count)
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_batch_rows_total{status="error"} 1`,
		`vehicle_history_batch_flush_total{status="error"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("batch error metric missing %s:\n%s", want, text)
		}
	}
}

func TestProcessHistoryBatchCommitsWhenLocationDerivedWriteFails(t *testing.T) {
	env := envelope.FrameEnvelope{
		Protocol:  envelope.ProtocolGB32960,
		VIN:       "VIN001",
		MessageID: "0x02",
		ParsedFields: map[string]any{
			"gb32960.position.longitude": 121.1,
			"gb32960.position.latitude":  30.2,
		},
	}
	payload, err := json.Marshal(env)
	if err != nil {
		t.Fatal(err)
	}
	locationErr := errors.New("location child table failed")
	appender := &contextCheckingHistoryAppender{
		result: history.AppendResult{
			RawRows:       1,
			LocationError: locationErr,
		},
	}
	committer := &contextCheckingHistoryCommitter{}
	registry := metrics.NewRegistry()

	processHistoryBatch(
		context.Background(),
		discardHistoryLogger{},
		registry,
		appender,
		committer,
		[]kafka.Message{{Topic: "vehicle.raw.go.gb32960.v1", Partition: 1, Offset: 10, HighWaterMark: 11, Value: payload}},
	)

	if committer.messageCount != 1 {
		t.Fatalf("committed messages = %d, want raw-success message committed", committer.messageCount)
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_writes_total{status="ok",topic="vehicle.raw.go.gb32960.v1"} 1`,
		`vehicle_history_location_writes_total{status="error",topic="vehicle.raw.go.gb32960.v1"} 1`,
		`vehicle_history_location_batch_flush_total{status="error"} 1`,
		`vehicle_history_kafka_commits_total{status="ok",topic="vehicle.raw.go.gb32960.v1"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("location derived failure metric missing %s:\n%s", want, text)
		}
	}
}

func TestRetryHistoryAppenderRetriesTransientTDengineError(t *testing.T) {
	transientErr := errors.New("taosws i/o timeout")
	registry := metrics.NewRegistry()
	appender := &contextCheckingHistoryAppender{
		result: history.AppendResult{RawRows: 1, LocationRows: 1},
	}
	appender.onAppendBatch = func() {
		if appender.batchCount == 1 {
			appender.err = transientErr
			return
		}
		appender.err = nil
	}
	retryAppender := retryHistoryAppender{
		delegate: appender,
		attempts: 3,
		registry: registry,
	}

	result, err := retryAppender.AppendAllBatchWithResult(context.Background(), []envelope.FrameEnvelope{
		{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"},
	})
	if err != nil {
		t.Fatalf("AppendAllBatchWithResult returned error: %v", err)
	}
	if appender.batchCount != 2 {
		t.Fatalf("batch attempts = %d, want 2", appender.batchCount)
	}
	if result.RawRows != 1 || result.LocationRows != 1 {
		t.Fatalf("result = %+v, want raw and location rows preserved", result)
	}
	text := registry.Render()
	if !strings.Contains(text, `vehicle_history_write_retries_total{operation="batch",status="retry"} 1`) {
		t.Fatalf("retry metric missing:\n%s", text)
	}
}

func TestRetryHistoryAppenderRecordsExhaustedTransientError(t *testing.T) {
	registry := metrics.NewRegistry()
	appender := &contextCheckingHistoryAppender{err: errors.New("connection reset by peer")}
	retryAppender := retryHistoryAppender{
		delegate: appender,
		attempts: 2,
		registry: registry,
	}

	_, err := retryAppender.AppendAllBatchWithResult(context.Background(), []envelope.FrameEnvelope{
		{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"},
	})
	if err == nil {
		t.Fatal("AppendAllBatchWithResult error = nil, want exhausted transient error")
	}
	if appender.batchCount != 2 {
		t.Fatalf("batch attempts = %d, want 2", appender.batchCount)
	}
	text := registry.Render()
	for _, want := range []string{
		`vehicle_history_write_retries_total{operation="batch",status="retry"} 1`,
		`vehicle_history_write_retries_total{operation="batch",status="exhausted"} 1`,
	} {
		if !strings.Contains(text, want) {
			t.Fatalf("retry metric missing %s:\n%s", want, text)
		}
	}
}

func TestRetryHistoryAppenderDoesNotRetryNonTransientError(t *testing.T) {
	appender := &contextCheckingHistoryAppender{err: errors.New("syntax error near table name")}
	retryAppender := retryHistoryAppender{
		delegate: appender,
		attempts: 3,
	}

	_, err := retryAppender.AppendAllBatchWithResult(context.Background(), []envelope.FrameEnvelope{
		{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"},
	})
	if err == nil {
		t.Fatal("AppendAllBatchWithResult error = nil, want non-transient error")
	}
	if appender.batchCount != 1 {
		t.Fatalf("batch attempts = %d, want 1", appender.batchCount)
	}
}

func TestRetryHistoryAppenderDoesNotRetryLocationDerivedError(t *testing.T) {
	locationErr := errors.New("location child table failed")
	appender := &contextCheckingHistoryAppender{
		result: history.AppendResult{
			RawRows:       1,
			LocationError: locationErr,
		},
	}
	retryAppender := retryHistoryAppender{
		delegate: appender,
		attempts: 3,
	}

	result, err := retryAppender.AppendAllBatchWithResult(context.Background(), []envelope.FrameEnvelope{
		{Protocol: envelope.ProtocolGB32960, VIN: "VIN001", MessageID: "0x02"},
	})
	if err != nil {
		t.Fatalf("AppendAllBatchWithResult returned error: %v", err)
	}
	if appender.batchCount != 1 {
		t.Fatalf("batch attempts = %d, want 1", appender.batchCount)
	}
	if !errors.Is(result.LocationError, locationErr) {
		t.Fatalf("LocationError = %v, want %v", result.LocationError, locationErr)
	}
}

func TestLoadConfigDefaultsToGoRawTopics(t *testing.T) {
	cfg := loadConfig()

	got := strings.Join(cfg.KafkaTopics, ",")
	want := "vehicle.raw.go.gb32960.v1,vehicle.raw.go.jt808.v1,vehicle.raw.go.yutong-mqtt.v1"
	if got != want {
		t.Fatalf("KafkaTopics = %q, want %q", got, want)
	}
	if cfg.BatchSize != 200 {
		t.Fatalf("BatchSize = %d, want 200", cfg.BatchSize)
	}
	if cfg.Workers != 3 {
		t.Fatalf("Workers = %d, want 3", cfg.Workers)
	}
	if cfg.BatchWait != 20 {
		t.Fatalf("BatchWait = %d, want 20", cfg.BatchWait)
	}
	if cfg.RetryAttempts != 3 {
		t.Fatalf("RetryAttempts = %d, want 3", cfg.RetryAttempts)
	}
	if cfg.RetryDelay != 20*time.Millisecond {
		t.Fatalf("RetryDelay = %s, want 20ms", cfg.RetryDelay)
	}
}

func TestLoadConfigReadsBatchSettings(t *testing.T) {
	t.Setenv("HISTORY_WORKERS", "5")
	t.Setenv("HISTORY_BATCH_SIZE", "500")
	t.Setenv("HISTORY_BATCH_WAIT_MS", "250")
	t.Setenv("HISTORY_RETRY_ATTEMPTS", "5")
	t.Setenv("HISTORY_RETRY_DELAY_MS", "33")

	cfg := loadConfig()

	if cfg.Workers != 5 {
		t.Fatalf("Workers = %d, want 5", cfg.Workers)
	}
	if cfg.BatchSize != 500 {
		t.Fatalf("BatchSize = %d, want 500", cfg.BatchSize)
	}
	if cfg.BatchWait != 250 {
		t.Fatalf("BatchWait = %d, want 250", cfg.BatchWait)
	}
	if cfg.RetryAttempts != 5 {
		t.Fatalf("RetryAttempts = %d, want 5", cfg.RetryAttempts)
	}
	if cfg.RetryDelay != 33*time.Millisecond {
		t.Fatalf("RetryDelay = %s, want 33ms", cfg.RetryDelay)
	}
}

func TestConfigValidateRejectsFieldsTopics(t *testing.T) {
	cfg := config{KafkaTopics: []string{"vehicle.fields.go.jt808.v1"}, Workers: 1}

	err := cfg.Validate()

	if err == nil || !strings.Contains(err.Error(), "raw topics only") {
		t.Fatalf("Validate() error = %v, want raw topic rejection", err)
	}
}

func TestConfigValidateRequiresPositiveWorkers(t *testing.T) {
	cfg := config{KafkaTopics: []string{"vehicle.raw.go.jt808.v1"}, Workers: 0}

	err := cfg.Validate()

	if err == nil || !strings.Contains(err.Error(), "HISTORY_WORKERS") {
		t.Fatalf("Validate() error = %v, want HISTORY_WORKERS hint", err)
	}
}

type contextCheckingHistoryAppender struct {
	ctxErr            error
	count             int
	batchCount        int
	batch             []envelope.FrameEnvelope
	err               error
	batchErr          error
	singleErr         error
	failSingleOnCount int
	result            history.AppendResult
	onAppendBatch     func()
}

func (a *contextCheckingHistoryAppender) AppendAll(ctx context.Context, _ envelope.FrameEnvelope) error {
	a.ctxErr = ctx.Err()
	a.count++
	if a.ctxErr != nil {
		return a.ctxErr
	}
	return a.currentSingleErr()
}

func (a *contextCheckingHistoryAppender) AppendAllWithResult(ctx context.Context, _ envelope.FrameEnvelope) (history.AppendResult, error) {
	a.ctxErr = ctx.Err()
	a.count++
	if a.ctxErr != nil {
		return history.AppendResult{}, a.ctxErr
	}
	if err := a.currentSingleErr(); err != nil {
		return history.AppendResult{}, err
	}
	result := a.result
	if result.RawRows == 0 {
		result.RawRows = 1
	}
	return result, nil
}

func (a *contextCheckingHistoryAppender) AppendAllBatch(ctx context.Context, envs []envelope.FrameEnvelope) error {
	a.ctxErr = ctx.Err()
	a.batchCount++
	a.batch = append([]envelope.FrameEnvelope(nil), envs...)
	if a.onAppendBatch != nil {
		a.onAppendBatch()
	}
	if a.ctxErr != nil {
		return a.ctxErr
	}
	return a.currentBatchErr()
}

func (a *contextCheckingHistoryAppender) AppendAllBatchWithResult(ctx context.Context, envs []envelope.FrameEnvelope) (history.AppendResult, error) {
	a.ctxErr = ctx.Err()
	a.batchCount++
	a.batch = append([]envelope.FrameEnvelope(nil), envs...)
	if a.onAppendBatch != nil {
		a.onAppendBatch()
	}
	if a.ctxErr != nil {
		return history.AppendResult{}, a.ctxErr
	}
	if err := a.currentBatchErr(); err != nil {
		return history.AppendResult{}, err
	}
	result := a.result
	if result.RawRows == 0 {
		result.RawRows = len(envs)
	}
	return result, nil
}

func (a *contextCheckingHistoryAppender) currentSingleErr() error {
	if a.failSingleOnCount > 0 && a.count == a.failSingleOnCount {
		if a.singleErr != nil {
			return a.singleErr
		}
		return a.err
	}
	if a.failSingleOnCount > 0 {
		return nil
	}
	if a.singleErr != nil {
		return a.singleErr
	}
	return a.err
}

func (a *contextCheckingHistoryAppender) currentBatchErr() error {
	if a.batchErr != nil {
		return a.batchErr
	}
	return a.err
}

type contextCheckingHistoryCommitter struct {
	ctxErr       error
	count        int
	messageCount int
	messages     []kafka.Message
	err          error
	failOnCount  int
}

func (c *contextCheckingHistoryCommitter) CommitMessages(ctx context.Context, messages ...kafka.Message) error {
	c.ctxErr = ctx.Err()
	c.count++
	c.messageCount += len(messages)
	c.messages = append(c.messages, messages...)
	if c.ctxErr != nil {
		return c.ctxErr
	}
	if c.err != nil && (c.failOnCount == 0 || c.count == c.failOnCount) {
		return c.err
	}
	return nil
}

type discardHistoryLogger struct{}

func (discardHistoryLogger) Error(string, ...any) {}
func (discardHistoryLogger) Warn(string, ...any)  {}

var errTestHistoryAppend = errors.New("test history append failed")

func committedOffsets(messages []kafka.Message) []int64 {
	offsets := make([]int64, len(messages))
	for index, message := range messages {
		offsets[index] = message.Offset
	}
	return offsets
}