package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"sort"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/segmentio/kafka-go"
	_ "github.com/taosdata/driver-go/v3/taosWS"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/eventbus"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/health"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/history"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/topics"
)

func main() {
	logger := observability.NewLogger("vehicle-history-writer")
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	cfg := loadConfig()
	if err := cfg.Validate(); err != nil {
		logger.Error("invalid history writer config", "error", err)
		os.Exit(1)
	}
	db, err := sql.Open(cfg.TDengineDriver, cfg.TDengineDSN)
	if err != nil {
		logger.Error("tdengine open failed", "error", err)
		os.Exit(1)
	}
	defer db.Close()
	if err := db.PingContext(ctx); err != nil {
		logger.Error("tdengine ping failed", "error", err)
		os.Exit(1)
	}
	registry := metrics.NewRegistry()
	metrics.RegisterKafkaConsumerInfo(registry, "vehicle-history-writer", cfg.KafkaGroup, cfg.KafkaTopics)
	registry.SetGauge("vehicle_history_config", metrics.Labels{"setting": "workers"}, float64(cfg.Workers))
	health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-history-writer", []health.Check{
		{Name: "tdengine", Check: db.PingContext},
	}, registry))

	writer := history.NewWriterWithDatabase(db, cfg.TDengineDatabase)
	if cfg.EnsureSchema {
		if err := writer.EnsureSchema(ctx, cfg.TDengineDatabase); err != nil {
			logger.Error("tdengine schema bootstrap failed", "error", err)
			os.Exit(1)
		}
	}
	var appender historyAppender = retryHistoryAppender{
		delegate: writer,
		attempts: cfg.RetryAttempts,
		delay:    cfg.RetryDelay,
		registry: registry,
	}

	logger.Info("history writer started",
		"driver", cfg.TDengineDriver,
		"group", cfg.KafkaGroup,
		"topics", strings.Join(cfg.KafkaTopics, ","),
		"workers", cfg.Workers,
		"batch_size", cfg.BatchSize,
		"batch_wait_ms", cfg.BatchWait,
		"retry_attempts", cfg.RetryAttempts,
		"retry_delay_ms", cfg.RetryDelay.Milliseconds())

	var workers sync.WaitGroup
	for workerID := 1; workerID <= cfg.Workers; workerID++ {
		workers.Add(1)
		go func(id int) {
			defer workers.Done()
			runHistoryConsumer(ctx, logger, registry, appender, cfg, id)
		}(workerID)
	}
	workers.Wait()
}

func runHistoryConsumer(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, appender historyAppender, cfg config, workerID int) {
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:     cfg.KafkaBrokers,
		GroupID:     cfg.KafkaGroup,
		GroupTopics: cfg.KafkaTopics,
		MinBytes:    1,
		MaxBytes:    10e6,
	})
	defer reader.Close()

	workerLabels := metrics.Labels{"worker": strconv.Itoa(workerID)}
	registry.SetGauge("vehicle_history_worker_active", workerLabels, 1)
	defer registry.SetGauge("vehicle_history_worker_active", workerLabels, 0)

	for {
		message, err := reader.FetchMessage(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return
			}
			logger.Error("kafka fetch failed", "error", err)
			continue
		}
		batch := collectHistoryBatch(ctx, reader, message, cfg.BatchSize, time.Duration(cfg.BatchWait)*time.Millisecond)
		processHistoryBatchReliablyForWorker(ctx, logger, registry, appender, reader, batch, cfg.RetryDelay, workerLabels)
	}
}

const kafkaMessageOperationTimeout = 30 * time.Second

type historyAppender interface {
	AppendAll(context.Context, envelope.FrameEnvelope) error
	AppendAllBatch(context.Context, []envelope.FrameEnvelope) error
}

type historyResultAppender interface {
	AppendAllWithResult(context.Context, envelope.FrameEnvelope) (history.AppendResult, error)
	AppendAllBatchWithResult(context.Context, []envelope.FrameEnvelope) (history.AppendResult, error)
}

type kafkaMessageCommitter interface {
	CommitMessages(context.Context, ...kafka.Message) error
}

type kafkaMessageFetcher interface {
	FetchMessage(context.Context) (kafka.Message, error)
}

type historyBatchItem struct {
	message   kafka.Message
	valid     bool
	processed bool
}

type historyBatchOutcome struct {
	commitMessages []kafka.Message
	retryMessages  []kafka.Message
}

func collectHistoryBatch(ctx context.Context, fetcher kafkaMessageFetcher, first kafka.Message, maxSize int, maxWait time.Duration) []kafka.Message {
	if maxSize <= 1 {
		return []kafka.Message{first}
	}
	if maxWait <= 0 {
		maxWait = 100 * time.Millisecond
	}
	batch := []kafka.Message{first}
	deadline := time.Now().Add(maxWait)
	for len(batch) < maxSize {
		remaining := time.Until(deadline)
		if remaining <= 0 {
			break
		}
		fetchCtx, cancel := context.WithTimeout(ctx, remaining)
		message, err := fetcher.FetchMessage(fetchCtx)
		cancel()
		if err != nil {
			break
		}
		batch = append(batch, message)
	}
	return batch
}

func processHistoryMessage(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, appender historyAppender, committer kafkaMessageCommitter, message kafka.Message) {
	messageCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), kafkaMessageOperationTimeout)
	defer cancel()

	addWriterMetric(registry, "vehicle_history_kafka_messages_total", message, "received")
	addWriterLagMetric(registry, message)
	var env envelope.FrameEnvelope
	if err := json.Unmarshal(message.Value, &env); err != nil {
		addWriterMetric(registry, "vehicle_history_kafka_messages_total", message, "invalid_json")
		logger.Warn("skip invalid envelope json", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "error", err)
		if err := committer.CommitMessages(messageCtx, message); err != nil {
			addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "error")
			logger.Error("kafka commit invalid envelope failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "error", err)
			return
		}
		addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "ok")
		return
	}
	if status, err := topics.ValidateRawEnvelope(message.Topic, env); err != nil {
		addWriterMetric(registry, "vehicle_history_kafka_messages_total", message, status)
		logger.Warn("skip mismatched history raw envelope", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "protocol", env.Protocol, "event_id", env.StableEventID(), "event_kind", env.EventKind, "error", err)
		if err := committer.CommitMessages(messageCtx, message); err != nil {
			addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "error")
			logger.Error("kafka commit mismatched raw envelope failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "event_id", env.StableEventID(), "status", status, "error", err)
			return
		}
		addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "ok")
		return
	}
	recordHistoryParsedFieldMetrics(registry, message, env)
	result, err := appendHistoryEnvelope(messageCtx, appender, env)
	if err != nil {
		addWriterMetric(registry, "vehicle_history_writes_total", message, "error")
		logger.Error("tdengine raw append failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "event_id", env.StableEventID(), "error", err)
		return
	}
	recordHistoryDerivedMetrics(registry, []kafka.Message{message}, []envelope.FrameEnvelope{env}, result)
	if result.LocationError != nil {
		logger.Warn("tdengine location append failed after raw append", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "event_id", env.StableEventID(), "error", result.LocationError)
	}
	addWriterMetric(registry, "vehicle_history_writes_total", message, "ok")
	recordHistoryWriteE2EDuration(registry, message, env)
	if err := committer.CommitMessages(messageCtx, message); err != nil {
		addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "error")
		logger.Error("kafka commit failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "error", err)
		return
	}
	addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "ok")
}

func processHistoryBatch(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, appender historyAppender, committer kafkaMessageCommitter, messages []kafka.Message) historyBatchOutcome {
	return processHistoryBatchForWorker(ctx, logger, registry, appender, committer, messages, nil)
}

func processHistoryBatchForWorker(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, appender historyAppender, committer kafkaMessageCommitter, messages []kafka.Message, workerLabels metrics.Labels) historyBatchOutcome {
	if len(messages) == 0 {
		return historyBatchOutcome{}
	}
	messageCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), kafkaMessageOperationTimeout)
	defer cancel()

	envelopes := make([]envelope.FrameEnvelope, 0, len(messages))
	validMessages := make([]kafka.Message, 0, len(messages))
	validItemIndexes := make([]int, 0, len(messages))
	items := make([]historyBatchItem, 0, len(messages))
	for _, message := range messages {
		itemIndex := len(items)
		addWriterMetric(registry, "vehicle_history_kafka_messages_total", message, "received")
		addWriterLagMetric(registry, message)
		var env envelope.FrameEnvelope
		if err := json.Unmarshal(message.Value, &env); err != nil {
			addWriterMetric(registry, "vehicle_history_kafka_messages_total", message, "invalid_json")
			logger.Warn("skip invalid envelope json", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "error", err)
			items = append(items, historyBatchItem{message: message, processed: true})
			continue
		}
		if status, err := topics.ValidateRawEnvelope(message.Topic, env); err != nil {
			addWriterMetric(registry, "vehicle_history_kafka_messages_total", message, status)
			logger.Warn("skip mismatched history raw envelope", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "protocol", env.Protocol, "event_id", env.StableEventID(), "event_kind", env.EventKind, "error", err)
			items = append(items, historyBatchItem{message: message, processed: true})
			continue
		}
		recordHistoryParsedFieldMetrics(registry, message, env)
		envelopes = append(envelopes, env)
		validMessages = append(validMessages, message)
		validItemIndexes = append(validItemIndexes, itemIndex)
		items = append(items, historyBatchItem{message: message, valid: true})
	}
	if len(envelopes) > 0 {
		setBatchPendingForWorker(registry, workerLabels, len(messages), len(envelopes))
		defer setBatchPendingForWorker(registry, workerLabels, 0, 0)
		started := time.Now()
		result, err := appendHistoryBatch(messageCtx, appender, envelopes)
		elapsed := time.Since(started)
		status := "ok"
		if err != nil {
			status = "error"
		}
		addBatchMetric(registry, "vehicle_history_batch_flush_total", status, 1)
		addBatchMetric(registry, "vehicle_history_batch_rows_total", status, float64(len(envelopes)))
		setBatchDuration(registry, status, elapsed)
		if err != nil {
			first := messages[0]
			logger.Error("tdengine raw batch append failed", "topic", first.Topic, "partition", first.Partition, "offset", first.Offset, "rows", len(envelopes), "error", err)
			if isTransientTDengineHistoryError(err) {
				for _, message := range validMessages {
					addWriterMetric(registry, "vehicle_history_writes_total", message, "error")
				}
				addBatchMetric(registry, "vehicle_history_batch_fallback_total", "skipped_transient", 1)
				committed, commitErr := commitHistoryProcessedPrefixAfterFailure(messageCtx, logger, registry, committer, items)
				return historyFailureOutcome(messages, committed, commitErr)
			}
			if fallbackErr := fallbackHistoryBatchToSingles(messageCtx, logger, registry, appender, validMessages, envelopes, validItemIndexes, items); fallbackErr != nil {
				addBatchMetric(registry, "vehicle_history_batch_fallback_total", "error", 1)
				committed, commitErr := commitHistoryProcessedPrefixAfterFailure(messageCtx, logger, registry, committer, items)
				return historyFailureOutcome(messages, committed, commitErr)
			}
			addBatchMetric(registry, "vehicle_history_batch_fallback_total", "ok", 1)
			committed, commitErr := commitHistoryProcessedPrefixAfterFailure(messageCtx, logger, registry, committer, items)
			return historyFailureOutcome(messages, committed, commitErr)
		}
		recordHistoryDerivedMetrics(registry, validMessages, envelopes, result)
		if result.LocationError != nil {
			first := messages[0]
			logger.Warn("tdengine location batch append failed after raw append", "topic", first.Topic, "partition", first.Partition, "offset", first.Offset, "rows", len(envelopes), "location_rows", result.LocationRows, "error", result.LocationError)
		}
		for index, message := range validMessages {
			addWriterMetric(registry, "vehicle_history_writes_total", message, "ok")
			recordHistoryWriteE2EDuration(registry, message, envelopes[index])
		}
	}
	if err := committer.CommitMessages(messageCtx, messages...); err != nil {
		for _, message := range messages {
			addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "error")
		}
		first := messages[0]
		logger.Error("kafka batch commit failed", "topic", first.Topic, "partition", first.Partition, "offset", first.Offset, "messages", len(messages), "error", err)
		return historyBatchOutcome{commitMessages: messages}
	}
	for _, message := range messages {
		addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "ok")
	}
	return historyBatchOutcome{}
}

func historyFailureOutcome(messages []kafka.Message, committed []kafka.Message, commitErr error) historyBatchOutcome {
	outcome := historyBatchOutcome{
		retryMessages: eventbus.MessagesAfterCommittedPrefixes(messages, committed),
	}
	if commitErr != nil {
		outcome.commitMessages = committed
	}
	return outcome
}

func processHistoryBatchReliably(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, appender historyAppender, committer kafkaMessageCommitter, messages []kafka.Message, retryDelay time.Duration) {
	processHistoryBatchReliablyForWorker(ctx, logger, registry, appender, committer, messages, retryDelay, nil)
}

func processHistoryBatchReliablyForWorker(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, appender historyAppender, committer kafkaMessageCommitter, messages []kafka.Message, retryDelay time.Duration, workerLabels metrics.Labels) {
	defer registry.SetGauge("vehicle_history_retry_pending_messages", workerLabels, 0)
	pending := messages
	for len(pending) > 0 {
		outcome := processHistoryBatchForWorker(ctx, logger, registry, appender, committer, pending, workerLabels)
		if len(outcome.commitMessages) > 0 {
			registry.IncCounter("vehicle_history_batch_retries_total", metrics.Labels{"reason": "commit_error"})
			if !retryHistoryCommit(ctx, logger, registry, committer, outcome.commitMessages, retryDelay) {
				return
			}
		}
		pending = outcome.retryMessages
		registry.SetGauge("vehicle_history_retry_pending_messages", workerLabels, float64(len(pending)))
		if len(pending) == 0 {
			return
		}
		registry.IncCounter("vehicle_history_batch_retries_total", metrics.Labels{"reason": "write_error"})
		if !waitForHistoryRetry(ctx, retryDelay) {
			return
		}
	}
}

func retryHistoryCommit(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, committer kafkaMessageCommitter, messages []kafka.Message, retryDelay time.Duration) bool {
	for len(messages) > 0 {
		operationCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), kafkaMessageOperationTimeout)
		err := committer.CommitMessages(operationCtx, messages...)
		cancel()
		if err == nil {
			for _, message := range messages {
				addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "ok")
			}
			return true
		}
		for _, message := range messages {
			addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "error")
		}
		first := messages[0]
		logger.Error("kafka commit retry failed", "topic", first.Topic, "partition", first.Partition, "offset", first.Offset, "messages", len(messages), "error", err)
		registry.IncCounter("vehicle_history_batch_retries_total", metrics.Labels{"reason": "commit_error"})
		if !waitForHistoryRetry(ctx, retryDelay) {
			return false
		}
	}
	return true
}

func waitForHistoryRetry(ctx context.Context, retryDelay time.Duration) bool {
	if retryDelay <= 0 {
		retryDelay = 100 * time.Millisecond
	}
	timer := time.NewTimer(retryDelay)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-timer.C:
		return true
	}
}

func fallbackHistoryBatchToSingles(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, appender historyAppender, messages []kafka.Message, envelopes []envelope.FrameEnvelope, itemIndexes []int, items []historyBatchItem) error {
	for index, env := range envelopes {
		message := messages[index]
		started := time.Now()
		result, err := appendHistoryEnvelope(ctx, appender, env)
		setFallbackDuration(registry, statusFromError(err), time.Since(started))
		if err != nil {
			addWriterMetric(registry, "vehicle_history_writes_total", message, "error")
			logger.Error("tdengine raw single append failed after batch fallback", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "event_id", env.StableEventID(), "error", err)
			return err
		}
		recordHistoryDerivedMetrics(registry, []kafka.Message{message}, []envelope.FrameEnvelope{env}, result)
		if result.LocationError != nil {
			logger.Warn("tdengine location append failed after raw single fallback", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "event_id", env.StableEventID(), "error", result.LocationError)
		}
		addWriterMetric(registry, "vehicle_history_writes_total", message, "ok")
		recordHistoryWriteE2EDuration(registry, message, env)
		items[itemIndexes[index]].processed = true
	}
	return nil
}

func commitHistoryProcessedPrefixAfterFailure(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, committer kafkaMessageCommitter, items []historyBatchItem) ([]kafka.Message, error) {
	committable := processedHistoryPrefixMessagesByPartition(items)
	if len(committable) == 0 {
		return nil, nil
	}
	if err := committer.CommitMessages(ctx, committable...); err != nil {
		for _, message := range committable {
			addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "error")
		}
		first := committable[0]
		logger.Error("kafka processed-prefix commit failed after raw batch append failure", "topic", first.Topic, "partition", first.Partition, "offset", first.Offset, "messages", len(committable), "error", err)
		return committable, err
	}
	for _, message := range committable {
		addWriterMetric(registry, "vehicle_history_kafka_commits_total", message, "ok")
	}
	return committable, nil
}

func processedHistoryPrefixMessagesByPartition(items []historyBatchItem) []kafka.Message {
	type partitionKey struct {
		topic     string
		partition int
	}
	groups := map[partitionKey][]historyBatchItem{}
	for _, item := range items {
		key := partitionKey{topic: item.message.Topic, partition: item.message.Partition}
		groups[key] = append(groups[key], item)
	}
	var keys []partitionKey
	for key := range groups {
		keys = append(keys, key)
	}
	sort.Slice(keys, func(i, j int) bool {
		if keys[i].topic != keys[j].topic {
			return keys[i].topic < keys[j].topic
		}
		return keys[i].partition < keys[j].partition
	})
	var out []kafka.Message
	for _, key := range keys {
		group := groups[key]
		sort.Slice(group, func(i, j int) bool {
			return group[i].message.Offset < group[j].message.Offset
		})
		var previousOffset int64
		for index, item := range group {
			if index > 0 && item.message.Offset != previousOffset+1 {
				break
			}
			if !item.processed {
				break
			}
			out = append(out, item.message)
			previousOffset = item.message.Offset
		}
	}
	return out
}

func appendHistoryEnvelope(ctx context.Context, appender historyAppender, env envelope.FrameEnvelope) (history.AppendResult, error) {
	if resultAppender, ok := appender.(historyResultAppender); ok {
		return resultAppender.AppendAllWithResult(ctx, env)
	}
	if err := appender.AppendAll(ctx, env); err != nil {
		return history.AppendResult{}, err
	}
	return history.AppendResult{RawRows: 1}, nil
}

func appendHistoryBatch(ctx context.Context, appender historyAppender, envelopes []envelope.FrameEnvelope) (history.AppendResult, error) {
	if resultAppender, ok := appender.(historyResultAppender); ok {
		return resultAppender.AppendAllBatchWithResult(ctx, envelopes)
	}
	if err := appender.AppendAllBatch(ctx, envelopes); err != nil {
		return history.AppendResult{}, err
	}
	return history.AppendResult{RawRows: len(envelopes)}, nil
}

type retryHistoryAppender struct {
	delegate historyAppender
	attempts int
	delay    time.Duration
	registry *metrics.Registry
}

func (a retryHistoryAppender) AppendAll(ctx context.Context, env envelope.FrameEnvelope) error {
	_, err := a.AppendAllWithResult(ctx, env)
	return err
}

func (a retryHistoryAppender) AppendAllWithResult(ctx context.Context, env envelope.FrameEnvelope) (history.AppendResult, error) {
	if a.delegate == nil {
		return history.AppendResult{}, nil
	}
	attempts := a.attempts
	if attempts <= 0 {
		attempts = 1
	}
	var result history.AppendResult
	var err error
	for attempt := 1; attempt <= attempts; attempt++ {
		result, err = appendHistoryEnvelope(ctx, a.delegate, env)
		if err == nil || !isTransientTDengineHistoryError(err) {
			return result, err
		}
		if attempt == attempts {
			a.recordRetry("single", "exhausted")
			return result, err
		}
		a.recordRetry("single", "retry")
		if waitErr := sleepBeforeRetry(ctx, a.delay); waitErr != nil {
			return result, waitErr
		}
	}
	return result, err
}

func (a retryHistoryAppender) AppendAllBatch(ctx context.Context, envelopes []envelope.FrameEnvelope) error {
	_, err := a.AppendAllBatchWithResult(ctx, envelopes)
	return err
}

func (a retryHistoryAppender) AppendAllBatchWithResult(ctx context.Context, envelopes []envelope.FrameEnvelope) (history.AppendResult, error) {
	if a.delegate == nil {
		return history.AppendResult{}, nil
	}
	attempts := a.attempts
	if attempts <= 0 {
		attempts = 1
	}
	var result history.AppendResult
	var err error
	for attempt := 1; attempt <= attempts; attempt++ {
		result, err = appendHistoryBatch(ctx, a.delegate, envelopes)
		if err == nil || !isTransientTDengineHistoryError(err) {
			return result, err
		}
		if attempt == attempts {
			a.recordRetry("batch", "exhausted")
			return result, err
		}
		a.recordRetry("batch", "retry")
		if waitErr := sleepBeforeRetry(ctx, a.delay); waitErr != nil {
			return result, waitErr
		}
	}
	return result, err
}

func (a retryHistoryAppender) recordRetry(operation string, status string) {
	if a.registry == nil {
		return
	}
	labels := metrics.Labels{"operation": operation, "status": status}
	a.registry.IncCounter("vehicle_history_write_retries_total", labels)
	metrics.RecordLastActivity(a.registry, "vehicle_history_last_write_retry_unix_seconds", labels)
}

func sleepBeforeRetry(ctx context.Context, delay time.Duration) error {
	if delay <= 0 {
		return nil
	}
	timer := time.NewTimer(delay)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-timer.C:
		return nil
	}
}

func isTransientTDengineHistoryError(err error) bool {
	if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		return false
	}
	text := strings.ToLower(strings.TrimSpace(err.Error()))
	return strings.Contains(text, "timeout") ||
		strings.Contains(text, "temporary") ||
		strings.Contains(text, "temporarily") ||
		strings.Contains(text, "connection refused") ||
		strings.Contains(text, "connection reset") ||
		strings.Contains(text, "connection closed") ||
		strings.Contains(text, "broken pipe") ||
		strings.Contains(text, "bad connection") ||
		strings.Contains(text, "i/o timeout") ||
		text == "eof" ||
		strings.Contains(text, "unexpected eof") ||
		strings.Contains(text, "server is down") ||
		strings.Contains(text, "network is unreachable") ||
		strings.Contains(text, "no route to host")
}

func addWriterMetric(registry *metrics.Registry, name string, message kafka.Message, status string) {
	if registry == nil {
		return
	}
	labels := metrics.Labels{"topic": message.Topic, "status": status}
	registry.IncCounter(name, labels)
	switch name {
	case "vehicle_history_kafka_messages_total":
		metrics.RecordLastActivity(registry, "vehicle_history_last_message_unix_seconds", labels)
	case "vehicle_history_writes_total":
		metrics.RecordLastActivity(registry, "vehicle_history_last_write_unix_seconds", labels)
	case "vehicle_history_kafka_commits_total":
		metrics.RecordLastActivity(registry, "vehicle_history_last_commit_unix_seconds", labels)
	case "vehicle_history_location_writes_total":
		metrics.RecordLastActivity(registry, "vehicle_history_last_location_write_unix_seconds", labels)
	}
}

func addBatchMetric(registry *metrics.Registry, name string, status string, value float64) {
	if registry == nil {
		return
	}
	registry.AddCounter(name, metrics.Labels{"status": status}, value)
}

func recordHistoryParsedFieldMetrics(registry *metrics.Registry, message kafka.Message, env envelope.FrameEnvelope) {
	if registry == nil || !envelope.IsRealtimeTelemetryFrame(env) {
		return
	}
	status := "present"
	if len(env.ParsedFields) == 0 {
		status = "missing"
	}
	registry.IncCounter("vehicle_history_parsed_fields_total", metrics.Labels{
		"topic":    message.Topic,
		"protocol": historyProtocolLabel(env.Protocol),
		"status":   status,
	})
}

func recordHistoryDerivedMetrics(registry *metrics.Registry, messages []kafka.Message, envelopes []envelope.FrameEnvelope, result history.AppendResult) {
	if registry == nil || len(messages) == 0 {
		return
	}
	batchStatus := "skipped"
	if result.LocationError != nil {
		batchStatus = "error"
	} else if result.LocationRows > 0 {
		batchStatus = "ok"
	}
	for index, message := range messages {
		status := batchStatus
		if index < len(envelopes) {
			status = history.LocationStatus(envelopes[index])
			if status == history.LocationStatusOK && result.LocationError != nil {
				status = "error"
			}
		}
		addWriterMetric(registry, "vehicle_history_location_writes_total", message, status)
	}
	addBatchMetric(registry, "vehicle_history_location_batch_flush_total", batchStatus, 1)
	addBatchMetric(registry, "vehicle_history_location_rows_total", batchStatus, float64(result.LocationRows))
}

func historyProtocolLabel(protocol envelope.Protocol) string {
	protocolLabel := strings.TrimSpace(string(protocol))
	if protocolLabel == "" {
		return "UNKNOWN"
	}
	return protocolLabel
}

func setBatchDuration(registry *metrics.Registry, status string, elapsed time.Duration) {
	if registry == nil {
		return
	}
	elapsedMS := float64(elapsed.Milliseconds())
	labels := metrics.Labels{"status": status}
	registry.SetGauge("vehicle_history_batch_flush_duration_ms", labels, elapsedMS)
	registry.ObserveHistogram("vehicle_history_batch_flush_duration_ms_histogram", labels, historyBatchFlushDurationBucketsMS, elapsedMS)
}

var historyBatchFlushDurationBucketsMS = []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000}

func setFallbackDuration(registry *metrics.Registry, status string, elapsed time.Duration) {
	if registry == nil {
		return
	}
	elapsedMS := float64(elapsed.Milliseconds())
	labels := metrics.Labels{"status": status}
	registry.SetGauge("vehicle_history_batch_fallback_duration_ms", labels, elapsedMS)
	registry.ObserveHistogram("vehicle_history_batch_fallback_duration_ms_histogram", labels, historyBatchFallbackDurationBucketsMS, elapsedMS)
}

var historyBatchFallbackDurationBucketsMS = []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000}

var historyWriteE2EDurationBucketsMS = []float64{10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000}
var historyWriteE2ERecent = metrics.NewRecentLatencyByKey(512)

func recordHistoryWriteE2EDuration(registry *metrics.Registry, message kafka.Message, env envelope.FrameEnvelope) {
	if registry == nil || env.ReceivedAtMS <= 0 {
		return
	}
	elapsed := time.Since(time.UnixMilli(env.ReceivedAtMS)).Milliseconds()
	if elapsed < 0 {
		elapsed = 0
	}
	labels := metrics.Labels{"topic": message.Topic}
	registry.ObserveHistogram("vehicle_history_write_e2e_duration_ms_histogram", labels, historyWriteE2EDurationBucketsMS, float64(elapsed))
	p99, samples := historyWriteE2ERecent.Observe(message.Topic, float64(elapsed))
	registry.SetGauge("vehicle_history_write_e2e_recent_p99_ms", labels, p99)
	registry.SetGauge("vehicle_history_write_e2e_recent_samples", labels, float64(samples))
	metrics.RecordLastActivity(registry, "vehicle_history_last_write_e2e_unix_seconds", labels)
}

func statusFromError(err error) string {
	if err != nil {
		return "error"
	}
	return "ok"
}

func setBatchPending(registry *metrics.Registry, messages int, rows int) {
	setBatchPendingForWorker(registry, nil, messages, rows)
}

func setBatchPendingForWorker(registry *metrics.Registry, workerLabels metrics.Labels, messages int, rows int) {
	if registry == nil {
		return
	}
	registry.SetGauge("vehicle_history_batch_pending_messages", workerLabels, float64(messages))
	registry.SetGauge("vehicle_history_batch_pending_rows", workerLabels, float64(rows))
}

func addWriterLagMetric(registry *metrics.Registry, message kafka.Message) {
	if registry == nil {
		return
	}
	registry.SetKafkaLag("vehicle_history_kafka_lag", message.Topic, message.Partition, message.Offset, message.HighWaterMark)
}

type config struct {
	KafkaBrokers     []string
	KafkaTopics      []string
	KafkaGroup       string
	TDengineDriver   string
	TDengineDSN      string
	TDengineDatabase string
	EnsureSchema     bool
	Workers          int
	BatchSize        int
	BatchWait        int
	RetryAttempts    int
	RetryDelay       time.Duration
}

func (c config) Validate() error {
	if len(c.KafkaTopics) == 0 {
		return errors.New("KAFKA_TOPICS must include raw topics")
	}
	for _, topic := range c.KafkaTopics {
		if _, ok := topics.ProtocolForKnownRawTopic(topic); !ok {
			return fmt.Errorf("history-writer consumes known raw topics only, got %q", topic)
		}
	}
	if c.Workers <= 0 {
		return errors.New("HISTORY_WORKERS must be greater than zero")
	}
	return nil
}

func loadConfig() config {
	return config{
		KafkaBrokers:     splitCSV(env("KAFKA_BROKERS", "127.0.0.1:9092")),
		KafkaTopics:      splitCSV(env("KAFKA_TOPICS", strings.Join([]string{topics.RawGB32960, topics.RawJT808, topics.RawYutongMQTT}, ","))),
		KafkaGroup:       env("KAFKA_GROUP", "go-history-writer"),
		TDengineDriver:   env("TDENGINE_DRIVER", "taosWS"),
		TDengineDSN:      env("TDENGINE_DSN", ""),
		TDengineDatabase: env("TDENGINE_DATABASE", history.DefaultDatabase),
		EnsureSchema:     env("TDENGINE_ENSURE_SCHEMA", "true") != "false",
		Workers:          envInt("HISTORY_WORKERS", 3),
		BatchSize:        envInt("HISTORY_BATCH_SIZE", 200),
		BatchWait:        envInt("HISTORY_BATCH_WAIT_MS", 20),
		RetryAttempts:    envInt("HISTORY_RETRY_ATTEMPTS", 3),
		RetryDelay:       time.Duration(envInt("HISTORY_RETRY_DELAY_MS", 20)) * time.Millisecond,
	}
}

func env(key string, fallback string) string {
	value := strings.TrimSpace(os.Getenv(key))
	if value == "" {
		return fallback
	}
	return value
}

func envInt(key string, fallback int) int {
	value := strings.TrimSpace(os.Getenv(key))
	if value == "" {
		return fallback
	}
	parsed, err := strconv.Atoi(value)
	if err != nil {
		return fallback
	}
	return parsed
}

func splitCSV(value string) []string {
	var out []string
	for _, item := range strings.Split(value, ",") {
		item = strings.TrimSpace(item)
		if item != "" {
			out = append(out, item)
		}
	}
	return out
}