package main

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

	_ "github.com/go-sql-driver/mysql"
	"github.com/segmentio/kafka-go"

	"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/metrics"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/topics"
)

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

	cfg := loadConfig()
	if err := cfg.Validate(); err != nil {
		logger.Error("invalid stat writer config", "error", err)
		os.Exit(1)
	}
	db, err := sql.Open("mysql", cfg.MySQLDSN)
	if err != nil {
		logger.Error("mysql open failed", "error", err)
		os.Exit(1)
	}
	defer db.Close()
	if err := db.PingContext(ctx); err != nil {
		logger.Error("mysql ping failed", "error", err)
		os.Exit(1)
	}
	registry := metrics.NewRegistry()
	metrics.RegisterKafkaConsumerInfo(registry, "vehicle-stat-writer", cfg.KafkaGroup, cfg.KafkaTopics)
	registry.SetGauge("vehicle_stat_project_interval_seconds", nil, cfg.ProjectInterval.Seconds())
	registry.SetGauge("vehicle_stat_source_touch_interval_seconds", nil, cfg.SourceTouchInterval.Seconds())
	registry.SetGauge("vehicle_stat_cache_retention_seconds", nil, cfg.CacheRetention.Seconds())
	registry.SetGauge("vehicle_stat_cache_cleanup_interval_seconds", nil, cfg.CacheCleanupInterval.Seconds())
	registry.SetGauge("vehicle_stat_baseline_miss_ttl_seconds", nil, cfg.BaselineMissTTL.Seconds())
	registry.SetGauge("vehicle_stat_baseline_hit_ttl_seconds", nil, cfg.BaselineHitTTL.Seconds())
	registry.SetGauge("vehicle_stat_config", metrics.Labels{"setting": "workers"}, float64(cfg.Workers))
	health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-stat-writer", []health.Check{
		{Name: "mysql", Check: db.PingContext},
	}, registry))

	writer := stats.NewWriter(db, cfg.Location)
	writer.SetProjectionInterval(cfg.ProjectInterval)
	writer.SetSourceTouchInterval(cfg.SourceTouchInterval)
	writer.SetCacheRetention(cfg.CacheRetention)
	writer.SetCacheCleanupInterval(cfg.CacheCleanupInterval)
	writer.SetBaselineMissTTL(cfg.BaselineMissTTL)
	writer.SetBaselineHitTTL(cfg.BaselineHitTTL)
	writer.SetMaxCacheEntries(cfg.CacheMaxEntries)
	if cfg.EnsureSchema {
		if err := writer.EnsureSchema(ctx); err != nil {
			logger.Error("mysql schema bootstrap failed", "error", err)
			os.Exit(1)
		}
	}
	if cfg.NormalizePlatformSourcesOnStart {
		statDate := time.Now().In(cfg.Location).Format("2006-01-02")
		normalizeCtx, cancel := context.WithTimeout(ctx, cfg.NormalizePlatformSourcesTimeout)
		normalized, err := stats.NormalizePlatformSourceMileageForDate(normalizeCtx, db, statDate, envelope.ProtocolJT808)
		cancel()
		if err != nil {
			logger.Warn("platform source startup normalization failed", "stat_date", statDate, "protocol", envelope.ProtocolJT808, "normalized", normalized, "error", err)
		} else if normalized > 0 {
			logger.Info("platform source startup normalization finished", "stat_date", statDate, "protocol", envelope.ProtocolJT808, "normalized", normalized)
		}
	}
	var appender statAppender = retryStatAppender{
		delegate: writer,
		attempts: cfg.RetryAttempts,
		delay:    cfg.RetryDelay,
		registry: registry,
	}
	quarantiner := fileStatMessageQuarantiner{dir: cfg.QuarantineDir}

	logger.Info("stat writer started", "group", cfg.KafkaGroup, "topics", strings.Join(cfg.KafkaTopics, ","), "workers", cfg.Workers, "project_interval_seconds", cfg.ProjectInterval.Seconds(), "source_touch_interval_seconds", cfg.SourceTouchInterval.Seconds(), "cache_retention_seconds", cfg.CacheRetention.Seconds(), "cache_cleanup_interval_seconds", cfg.CacheCleanupInterval.Seconds(), "baseline_miss_ttl_seconds", cfg.BaselineMissTTL.Seconds(), "baseline_hit_ttl_seconds", cfg.BaselineHitTTL.Seconds(), "cache_max_entries", cfg.CacheMaxEntries, "batch_size", cfg.BatchSize, "batch_wait_ms", cfg.BatchWait, "retry_attempts", cfg.RetryAttempts, "retry_delay_ms", cfg.RetryDelay.Milliseconds(), "quarantine_dir", cfg.QuarantineDir)
	var workers sync.WaitGroup
	for workerID := 1; workerID <= cfg.Workers; workerID++ {
		workers.Add(1)
		go func(id int) {
			defer workers.Done()
			runStatConsumer(ctx, logger, registry, appender, quarantiner, cfg, id)
		}(workerID)
	}
	workers.Wait()
}

func runStatConsumer(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, appender statAppender, quarantiner statMessageQuarantiner, 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_stat_worker_active", workerLabels, 1)
	defer registry.SetGauge("vehicle_stat_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 := collectStatBatch(ctx, reader, message, cfg.BatchSize, time.Duration(cfg.BatchWait)*time.Millisecond)
		processStatBatchReliablyForWorker(ctx, logger, registry, appender, reader, quarantiner, batch, cfg.RetryDelay, workerLabels)
	}
}

const kafkaMessageOperationTimeout = 30 * time.Second

type statAppender interface {
	Append(context.Context, envelope.FrameEnvelope) error
}

type statResultAppender interface {
	AppendWithResult(context.Context, envelope.FrameEnvelope) (stats.AppendResult, error)
}

type statCacheReporter interface {
	CacheStats() stats.CacheStats
}

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

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

type statMessageQuarantiner interface {
	Quarantine(context.Context, kafka.Message, error) error
}

type statBatchItem struct {
	message   kafka.Message
	processed bool
}

type statBatchOutcome struct {
	commitMessages []kafka.Message
	retryMessages  []kafka.Message
	failedMessage  *kafka.Message
	writeErr       error
}

var statWriteDurationBucketsMS = []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000}
var statWriteE2EDurationBucketsMS = []float64{10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000}
var statWriteE2ERecent = metrics.NewRecentLatencyByKey(512)

func collectStatBatch(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 processStatMessage(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, appender statAppender, committer kafkaMessageCommitter, message kafka.Message) {
	processStatBatch(ctx, logger, registry, appender, committer, []kafka.Message{message})
}

func processStatBatch(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, appender statAppender, committer kafkaMessageCommitter, messages []kafka.Message) statBatchOutcome {
	return processStatBatchForWorker(ctx, logger, registry, appender, committer, messages, nil)
}

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

	setStatBatchPendingForWorker(registry, workerLabels, len(messages))
	defer setStatBatchPendingForWorker(registry, workerLabels, 0)
	defer recordStatCacheMetrics(registry, appender)
	items := make([]statBatchItem, len(messages))
	for index, message := range messages {
		items[index].message = message
	}
	for itemIndex, message := range messages {
		addStatMetric(registry, "vehicle_stat_kafka_messages_total", message, "received")
		addStatLagMetric(registry, message)
		var env envelope.FrameEnvelope
		if err := json.Unmarshal(message.Value, &env); err != nil {
			addStatMetric(registry, "vehicle_stat_kafka_messages_total", message, "invalid_json")
			logger.Warn("skip invalid envelope json", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "error", err)
			items[itemIndex].processed = true
			continue
		}
		if status, err := validateStatFieldsEnvelope(message.Topic, env); err != nil {
			addStatMetric(registry, "vehicle_stat_kafka_messages_total", message, status)
			logger.Warn("skip mismatched stat fields envelope", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "protocol", env.Protocol, "event_id", env.StableEventID(), "event_kind", env.EventKind, "error", err)
			items[itemIndex].processed = true
			continue
		}
		started := time.Now()
		result, err := appendStatEnvelope(messageCtx, appender, env)
		recordStatWriteDuration(registry, message, statusFromError(err), time.Since(started))
		recordStatSampleMetrics(registry, message, env.Protocol, result)
		recordStatSourceMetrics(registry, message, env.Protocol, result)
		recordStatProjectionMetrics(registry, message, env.Protocol, result)
		if err != nil {
			addStatMetric(registry, "vehicle_stat_writes_total", message, "error")
			logger.Error("mysql append failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "event_id", env.StableEventID(), "error", err)
			committed, commitErr := commitStatProcessedPrefixAfterFailure(messageCtx, logger, registry, committer, items)
			failedMessage := message
			outcome := statBatchOutcome{
				retryMessages: eventbus.MessagesAfterCommittedPrefixes(messages, committed),
				failedMessage: &failedMessage,
				writeErr:      err,
			}
			if commitErr != nil {
				outcome.commitMessages = committed
			}
			return outcome
		}
		addStatMetric(registry, "vehicle_stat_writes_total", message, "ok")
		recordStatWriteE2EDuration(registry, message, env)
		items[itemIndex].processed = true
	}
	if err := committer.CommitMessages(messageCtx, messages...); err != nil {
		for _, message := range messages {
			addStatMetric(registry, "vehicle_stat_kafka_commits_total", message, "error")
		}
		first := messages[0]
		logger.Error("kafka commit failed", "topic", first.Topic, "partition", first.Partition, "offset", first.Offset, "messages", len(messages), "error", err)
		return statBatchOutcome{commitMessages: messages}
	}
	for _, message := range messages {
		addStatMetric(registry, "vehicle_stat_kafka_commits_total", message, "ok")
	}
	return statBatchOutcome{}
}

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

func processStatBatchReliablyForWorker(ctx context.Context, logger interface {
	Error(string, ...any)
	Warn(string, ...any)
}, registry *metrics.Registry, appender statAppender, committer kafkaMessageCommitter, quarantiner statMessageQuarantiner, messages []kafka.Message, retryDelay time.Duration, workerLabels metrics.Labels) {
	defer registry.SetGauge("vehicle_stat_retry_pending_messages", workerLabels, 0)
	pending := messages
	for len(pending) > 0 {
		outcome := processStatBatchForWorker(ctx, logger, registry, appender, committer, pending, workerLabels)
		if len(outcome.commitMessages) > 0 {
			registry.IncCounter("vehicle_stat_batch_retries_total", metrics.Labels{"reason": "commit_error"})
			if !retryStatCommit(ctx, logger, registry, committer, outcome.commitMessages, retryDelay) {
				return
			}
		}
		pending = outcome.retryMessages
		if outcome.writeErr != nil && outcome.failedMessage != nil && isQuarantinableStatError(outcome.writeErr) && quarantiner != nil {
			failed := *outcome.failedMessage
			if err := quarantiner.Quarantine(ctx, failed, outcome.writeErr); err != nil {
				logger.Error("stat message quarantine failed", "topic", failed.Topic, "partition", failed.Partition, "offset", failed.Offset, "error", err)
				registry.IncCounter("vehicle_stat_quarantine_total", metrics.Labels{"status": "error", "topic": failed.Topic})
			} else {
				registry.IncCounter("vehicle_stat_quarantine_total", metrics.Labels{"status": "ok", "topic": failed.Topic})
				logger.Warn("quarantined permanent stat write failure", "topic", failed.Topic, "partition", failed.Partition, "offset", failed.Offset, "error", outcome.writeErr)
				if retryStatCommit(ctx, logger, registry, committer, []kafka.Message{failed}, retryDelay) {
					pending = messagesExceptStatMessage(pending, failed)
				}
			}
		}
		registry.SetGauge("vehicle_stat_retry_pending_messages", workerLabels, float64(len(pending)))
		if len(pending) == 0 {
			return
		}
		registry.IncCounter("vehicle_stat_batch_retries_total", metrics.Labels{"reason": "write_error"})
		if !waitForStatRetry(ctx, retryDelay) {
			return
		}
	}
}

type quarantinedStatMessage struct {
	QuarantinedAt string `json:"quarantined_at"`
	Topic         string `json:"topic"`
	Partition     int    `json:"partition"`
	Offset        int64  `json:"offset"`
	Key           []byte `json:"key,omitempty"`
	Value         []byte `json:"value"`
	Error         string `json:"error"`
}

type fileStatMessageQuarantiner struct {
	dir string
}

func (q fileStatMessageQuarantiner) Quarantine(ctx context.Context, message kafka.Message, cause error) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	dir := strings.TrimSpace(q.dir)
	if dir == "" {
		return errors.New("stat quarantine directory is empty")
	}
	if err := os.MkdirAll(dir, 0o750); err != nil {
		return fmt.Errorf("create quarantine directory: %w", err)
	}
	record := quarantinedStatMessage{
		QuarantinedAt: time.Now().UTC().Format(time.RFC3339Nano),
		Topic:         message.Topic,
		Partition:     message.Partition,
		Offset:        message.Offset,
		Key:           message.Key,
		Value:         message.Value,
		Error:         cause.Error(),
	}
	payload, err := json.Marshal(record)
	if err != nil {
		return fmt.Errorf("marshal quarantine record: %w", err)
	}
	name := fmt.Sprintf("%s-%d-%d.json", sanitizeStatQuarantineName(message.Topic), message.Partition, message.Offset)
	temporary, err := os.CreateTemp(dir, "."+name+"-*")
	if err != nil {
		return fmt.Errorf("create quarantine record: %w", err)
	}
	temporaryName := temporary.Name()
	removeTemporary := true
	defer func() {
		_ = temporary.Close()
		if removeTemporary {
			_ = os.Remove(temporaryName)
		}
	}()
	if err := temporary.Chmod(0o600); err != nil {
		return fmt.Errorf("secure quarantine record: %w", err)
	}
	if _, err := temporary.Write(payload); err != nil {
		return fmt.Errorf("write quarantine record: %w", err)
	}
	if err := temporary.Sync(); err != nil {
		return fmt.Errorf("sync quarantine record: %w", err)
	}
	if err := temporary.Close(); err != nil {
		return fmt.Errorf("close quarantine record: %w", err)
	}
	if err := os.Rename(temporaryName, filepath.Join(dir, name)); err != nil {
		return fmt.Errorf("publish quarantine record: %w", err)
	}
	removeTemporary = false
	return nil
}

func sanitizeStatQuarantineName(value string) string {
	value = strings.TrimSpace(value)
	if value == "" {
		return "unknown-topic"
	}
	var out strings.Builder
	for _, r := range value {
		if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_' || r == '.' {
			out.WriteRune(r)
		} else {
			out.WriteByte('_')
		}
	}
	return out.String()
}

func messagesExceptStatMessage(messages []kafka.Message, excluded kafka.Message) []kafka.Message {
	out := make([]kafka.Message, 0, len(messages))
	removed := false
	for _, message := range messages {
		if !removed && message.Topic == excluded.Topic && message.Partition == excluded.Partition && message.Offset == excluded.Offset {
			removed = true
			continue
		}
		out = append(out, message)
	}
	return out
}

func retryStatCommit(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 {
				addStatMetric(registry, "vehicle_stat_kafka_commits_total", message, "ok")
			}
			return true
		}
		for _, message := range messages {
			addStatMetric(registry, "vehicle_stat_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_stat_batch_retries_total", metrics.Labels{"reason": "commit_error"})
		if !waitForStatRetry(ctx, retryDelay) {
			return false
		}
	}
	return true
}

func waitForStatRetry(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 validateStatFieldsEnvelope(topic string, env envelope.FrameEnvelope) (string, error) {
	return topics.ValidateFieldsEnvelope(topic, env)
}

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

func processedPrefixMessagesByPartition(items []statBatchItem) []kafka.Message {
	type partitionKey struct {
		topic     string
		partition int
	}
	groups := map[partitionKey][]statBatchItem{}
	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 appendStatEnvelope(ctx context.Context, appender statAppender, env envelope.FrameEnvelope) (stats.AppendResult, error) {
	if resultAppender, ok := appender.(statResultAppender); ok {
		return resultAppender.AppendWithResult(ctx, env)
	}
	return stats.AppendResult{}, appender.Append(ctx, env)
}

type retryStatAppender struct {
	delegate statAppender
	attempts int
	delay    time.Duration
	registry *metrics.Registry
}

func (a retryStatAppender) Append(ctx context.Context, env envelope.FrameEnvelope) error {
	_, err := a.AppendWithResult(ctx, env)
	return err
}

func (a retryStatAppender) AppendWithResult(ctx context.Context, env envelope.FrameEnvelope) (stats.AppendResult, error) {
	if a.delegate == nil {
		return stats.AppendResult{}, nil
	}
	attempts := a.attempts
	if attempts <= 0 {
		attempts = 1
	}
	var result stats.AppendResult
	var err error
	for attempt := 1; attempt <= attempts; attempt++ {
		result, err = appendStatEnvelope(ctx, a.delegate, env)
		if err == nil || (!isTransientMySQLStatError(err) && !isQuarantinableStatError(err)) {
			return result, err
		}
		if attempt == attempts {
			a.recordRetry("single", "exhausted")
			return result, err
		}
		a.recordRetry("single", "retry")
		if a.delay <= 0 {
			continue
		}
		timer := time.NewTimer(a.delay)
		select {
		case <-ctx.Done():
			timer.Stop()
			return result, ctx.Err()
		case <-timer.C:
		}
	}
	return result, err
}

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

func (a retryStatAppender) CacheStats() stats.CacheStats {
	if reporter, ok := a.delegate.(statCacheReporter); ok {
		return reporter.CacheStats()
	}
	return stats.CacheStats{}
}

func isTransientMySQLStatError(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, "deadlock") ||
		strings.Contains(text, "error 1213") ||
		strings.Contains(text, "40001") ||
		strings.Contains(text, "lock wait timeout") ||
		strings.Contains(text, "error 1205") ||
		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, "invalid 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 isQuarantinableStatError(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()))
	// Only deterministic row-data failures may advance past a Kafka message
	// after the configured finite retries. Schema, permission and unknown
	// application errors remain failure-closed so a systemic outage cannot
	// silently quarantine an entire stream.
	return strings.Contains(text, "error 1048") ||
		strings.Contains(text, "cannot be null") ||
		strings.Contains(text, "error 1264") ||
		strings.Contains(text, "out of range value") ||
		strings.Contains(text, "error 1265") ||
		strings.Contains(text, "data truncated") ||
		strings.Contains(text, "error 1292") ||
		strings.Contains(text, "incorrect datetime value") ||
		strings.Contains(text, "error 1366") ||
		strings.Contains(text, "incorrect decimal value") ||
		strings.Contains(text, "incorrect integer value") ||
		strings.Contains(text, "error 1406") ||
		strings.Contains(text, "data too long for column") ||
		strings.Contains(text, "error 3819") ||
		strings.Contains(text, "check constraint")
}

func addStatMetric(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_stat_kafka_messages_total":
		metrics.RecordLastActivity(registry, "vehicle_stat_last_message_unix_seconds", labels)
	case "vehicle_stat_writes_total":
		metrics.RecordLastActivity(registry, "vehicle_stat_last_write_unix_seconds", labels)
	case "vehicle_stat_kafka_commits_total":
		metrics.RecordLastActivity(registry, "vehicle_stat_last_commit_unix_seconds", labels)
	}
}

func recordStatSampleMetrics(registry *metrics.Registry, message kafka.Message, protocol envelope.Protocol, result stats.AppendResult) {
	addStatSampleMetric(registry, message, protocol, "found", result.SamplesFound)
	addStatSampleMetric(registry, message, protocol, "written", result.SamplesWritten)
	addStatSampleMetric(registry, message, protocol, "skipped_missing_fields", result.SamplesSkippedMissingFields)
	addStatSampleMetric(registry, message, protocol, "skipped_missing_vin", result.SamplesSkippedMissingVIN)
	addStatSampleMetric(registry, message, protocol, "skipped_missing_mileage", result.SamplesSkippedMissingMileage)
	addStatSampleMetric(registry, message, protocol, "recovered_stationary", result.SamplesRecoveredStationary)
	addStatSampleMetric(registry, message, protocol, "skipped_non_mileage_frame", result.SamplesSkippedNonMileageFrame)
	addStatSampleMetric(registry, message, protocol, "skipped_non_positive_mileage", result.SamplesSkippedNonPositiveMileage)
	addStatSampleMetric(registry, message, protocol, "skipped_missing_time", result.SamplesSkippedMissingTime)
	addStatSampleMetric(registry, message, protocol, "skipped_same_mileage", result.SamplesSkippedSameMileage)
	addStatSampleMetric(registry, message, protocol, "skipped_missing_source", result.SamplesSkippedMissingSource)
	addStatSampleMetric(registry, message, protocol, "event_time_future_adjusted", result.SamplesAdjustedFutureEventTime)
}

func addStatSampleMetric(registry *metrics.Registry, message kafka.Message, protocol envelope.Protocol, status string, count int) {
	if registry == nil || count == 0 {
		return
	}
	registry.AddCounter("vehicle_stat_samples_total", metrics.Labels{
		"topic":    message.Topic,
		"protocol": statProtocolLabel(protocol),
		"status":   status,
	}, float64(count))
}

func recordStatSourceMetrics(registry *metrics.Registry, message kafka.Message, protocol envelope.Protocol, result stats.AppendResult) {
	addStatSourceMetric(registry, message, protocol, "attempted", result.SourceTouchesAttempted)
	addStatSourceMetric(registry, message, protocol, "written", result.SourceTouchesWritten)
	addStatSourceMetric(registry, message, protocol, "skipped_throttled", result.SourceTouchesSkippedThrottled)
	addStatSourceMetric(registry, message, protocol, "skipped_missing_endpoint", result.SourceTouchesSkippedMissing)
	addStatSourceMetric(registry, message, protocol, "skipped_unmanaged", result.SourceTouchesSkippedUnmanaged)
}

func addStatSourceMetric(registry *metrics.Registry, message kafka.Message, protocol envelope.Protocol, status string, count int) {
	if registry == nil || count == 0 {
		return
	}
	registry.AddCounter("vehicle_stat_sources_total", metrics.Labels{
		"topic":    message.Topic,
		"protocol": statProtocolLabel(protocol),
		"status":   status,
	}, float64(count))
}

func recordStatProjectionMetrics(registry *metrics.Registry, message kafka.Message, protocol envelope.Protocol, result stats.AppendResult) {
	addStatProjectionMetric(registry, message, protocol, "attempted", result.ProjectionsAttempted)
	addStatProjectionMetric(registry, message, protocol, "written", result.ProjectionsWritten)
	addStatProjectionMetric(registry, message, protocol, "skipped_throttled", result.ProjectionsSkippedThrottled)
}

func addStatProjectionMetric(registry *metrics.Registry, message kafka.Message, protocol envelope.Protocol, status string, count int) {
	if registry == nil || count == 0 {
		return
	}
	registry.AddCounter("vehicle_stat_projections_total", metrics.Labels{
		"topic":    message.Topic,
		"protocol": statProtocolLabel(protocol),
		"status":   status,
	}, float64(count))
}

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

func recordStatCacheMetrics(registry *metrics.Registry, appender statAppender) {
	if registry == nil {
		return
	}
	reporter, ok := appender.(statCacheReporter)
	if !ok {
		return
	}
	stats := reporter.CacheStats()
	setStatCacheGauge(registry, "last_total_mileage", stats.LastTotalMileageEntries, stats.MaxEntries, stats.LastCleanupTotalMileage, stats.TotalMileageEvictions)
	setStatCacheGauge(registry, "source_seen", stats.LastSourceSeenEntries, stats.MaxEntries, stats.LastCleanupSourceSeen, stats.SourceSeenEvictions)
	setStatCacheGauge(registry, "projection", stats.LastProjectionEntries, stats.MaxEntries, stats.LastCleanupProjection, stats.ProjectionEvictions)
	setStatCacheGauge(registry, "baseline", stats.BaselineEntries, stats.MaxEntries, stats.LastCleanupBaseline, stats.BaselineEvictions)
	if !stats.LastCleanupAt.IsZero() {
		registry.SetGauge("vehicle_stat_cache_last_cleanup_unix_seconds", nil, float64(stats.LastCleanupAt.Unix()))
	}
}

func setStatCacheGauge(registry *metrics.Registry, cache string, entries int, maxEntries int, lastCleanupDeleted int, evictions int) {
	labels := metrics.Labels{"cache": cache}
	registry.SetGauge("vehicle_stat_cache_entries", labels, float64(entries))
	registry.SetGauge("vehicle_stat_cache_max_entries", labels, float64(maxEntries))
	registry.SetGauge("vehicle_stat_cache_last_cleanup_deleted", labels, float64(lastCleanupDeleted))
	registry.SetGauge("vehicle_stat_cache_evictions_total", labels, float64(evictions))
}

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

func recordStatWriteDuration(registry *metrics.Registry, message kafka.Message, status string, elapsed time.Duration) {
	if registry == nil {
		return
	}
	registry.ObserveHistogram("vehicle_stat_write_duration_ms_histogram", metrics.Labels{
		"topic":  message.Topic,
		"status": status,
	}, statWriteDurationBucketsMS, float64(elapsed.Milliseconds()))
}

func recordStatWriteE2EDuration(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_stat_write_e2e_duration_ms_histogram", labels, statWriteE2EDurationBucketsMS, float64(elapsed))
	p99, samples := statWriteE2ERecent.Observe(message.Topic, float64(elapsed))
	registry.SetGauge("vehicle_stat_write_e2e_recent_p99_ms", labels, p99)
	registry.SetGauge("vehicle_stat_write_e2e_recent_samples", labels, float64(samples))
	metrics.RecordLastActivity(registry, "vehicle_stat_last_write_e2e_unix_seconds", labels)
}

func setStatBatchPending(registry *metrics.Registry, messages int) {
	setStatBatchPendingForWorker(registry, nil, messages)
}

func setStatBatchPendingForWorker(registry *metrics.Registry, workerLabels metrics.Labels, messages int) {
	if registry == nil {
		return
	}
	registry.SetGauge("vehicle_stat_batch_pending_messages", workerLabels, float64(messages))
}

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

type config struct {
	KafkaBrokers                    []string
	KafkaTopics                     []string
	KafkaGroup                      string
	MySQLDSN                        string
	EnsureSchema                    bool
	Location                        *time.Location
	ProjectInterval                 time.Duration
	SourceTouchInterval             time.Duration
	CacheRetention                  time.Duration
	CacheCleanupInterval            time.Duration
	BaselineMissTTL                 time.Duration
	BaselineHitTTL                  time.Duration
	CacheMaxEntries                 int
	Workers                         int
	BatchSize                       int
	BatchWait                       int
	RetryAttempts                   int
	RetryDelay                      time.Duration
	NormalizePlatformSourcesOnStart bool
	NormalizePlatformSourcesTimeout time.Duration
	QuarantineDir                   string
}

func (c config) Validate() error {
	if len(c.KafkaTopics) == 0 {
		return fmt.Errorf("KAFKA_TOPICS must include fields topics")
	}
	for _, topic := range c.KafkaTopics {
		if !strings.HasPrefix(topic, "vehicle.fields.") {
			return fmt.Errorf("stat-writer consumes fields topics only, got %q", topic)
		}
	}
	if c.Workers <= 0 {
		return fmt.Errorf("STATS_WORKERS must be greater than zero")
	}
	return nil
}

func loadConfig() config {
	loc, err := time.LoadLocation(env("LOCAL_TZ", "Asia/Shanghai"))
	if err != nil {
		loc = time.FixedZone("Asia/Shanghai", 8*3600)
	}
	return config{
		KafkaBrokers:                    splitCSV(env("KAFKA_BROKERS", "127.0.0.1:9092")),
		KafkaTopics:                     splitCSV(env("KAFKA_TOPICS", strings.Join([]string{topics.FieldsGB32960, topics.FieldsJT808, topics.FieldsYutongMQTT}, ","))),
		KafkaGroup:                      env("KAFKA_GROUP", "go-stat-writer"),
		MySQLDSN:                        env("MYSQL_DSN", ""),
		EnsureSchema:                    env("MYSQL_ENSURE_SCHEMA", "true") != "false",
		Location:                        loc,
		ProjectInterval:                 time.Duration(envInt("STATS_PROJECT_INTERVAL_SECONDS", 15)) * time.Second,
		SourceTouchInterval:             time.Duration(envInt("STATS_SOURCE_TOUCH_INTERVAL_SECONDS", 60)) * time.Second,
		CacheRetention:                  time.Duration(envInt("STATS_CACHE_RETENTION_HOURS", 72)) * time.Hour,
		CacheCleanupInterval:            time.Duration(envInt("STATS_CACHE_CLEANUP_INTERVAL_SECONDS", 600)) * time.Second,
		BaselineMissTTL:                 time.Duration(envInt("STATS_BASELINE_MISS_TTL_SECONDS", 60)) * time.Second,
		BaselineHitTTL:                  time.Duration(envInt("STATS_BASELINE_HIT_TTL_SECONDS", 300)) * time.Second,
		CacheMaxEntries:                 envInt("STATS_CACHE_MAX_ENTRIES", 1000000),
		Workers:                         envInt("STATS_WORKERS", 3),
		BatchSize:                       envInt("STATS_BATCH_SIZE", 200),
		BatchWait:                       envInt("STATS_BATCH_WAIT_MS", 20),
		RetryAttempts:                   envInt("STATS_RETRY_ATTEMPTS", 3),
		RetryDelay:                      time.Duration(envInt("STATS_RETRY_DELAY_MS", 20)) * time.Millisecond,
		NormalizePlatformSourcesOnStart: env("STATS_NORMALIZE_PLATFORM_SOURCES_ON_START", "true") != "false",
		NormalizePlatformSourcesTimeout: time.Duration(envInt("STATS_NORMALIZE_PLATFORM_SOURCES_TIMEOUT_SECONDS", 30)) * time.Second,
		QuarantineDir:                   env("STATS_QUARANTINE_DIR", "/var/lib/lingniu-go-native/stat-writer-quarantine"),
	}
}

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
}