package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"

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

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/authentication"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/eventbus"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/gateway"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/health"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/identity"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/gb32960"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/jt808"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/topics"
)

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

	registry := metrics.NewRegistry()
	sink, err := buildSink(ctx, logger, registry)
	if err != nil {
		logger.Error("build sink failed", "error", err)
		os.Exit(1)
	}
	defer sink.Close()
	resolver, jt808DeviceTokens, closeResolver, err := buildIdentityResolver(ctx, logger, registry)
	if err != nil {
		logger.Error("build identity resolver failed", "error", err)
		os.Exit(1)
	}
	defer closeResolver()
	health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-gateway", nil, registry))
	publishUnified := envBool("PUBLISH_UNIFIED_ENABLED", false)
	delegateFields := envBool("FIELDS_DERIVE_FROM_RAW_ENABLED", strings.TrimSpace(os.Getenv("NATS_URL")) != "")
	gb32960Authenticator, gb32960AuthMode, gb32960CredentialCount, err := buildGB32960Authenticator()
	if err != nil {
		logger.Error("build gb32960 authenticator failed", "error", err)
		os.Exit(1)
	}
	jt808AuthCode := env("JT808_REGISTER_AUTH_CODE", "g7gps")
	jt808Authenticator, jt808AuthMode, err := buildJT808Authenticator(jt808AuthCode, jt808DeviceTokens)
	if err != nil {
		logger.Error("build jt808 authenticator failed", "error", err)
		os.Exit(1)
	}
	recordAuthenticationConfig(registry, envelope.ProtocolGB32960, gb32960AuthMode, gb32960CredentialCount)
	recordAuthenticationConfig(registry, envelope.ProtocolJT808, jt808AuthMode, int(boolMetric(strings.TrimSpace(jt808AuthCode) != "")))
	logger.Info("protocol authentication configured", "gb32960_mode", gb32960AuthMode, "gb32960_accounts", gb32960CredentialCount, "jt808_mode", jt808AuthMode)

	protocols := []gateway.TCPProtocol{
		{
			Protocol:     envelope.ProtocolGB32960,
			Addr:         env("GB32960_TCP_ADDR", ":32960"),
			Extract:      gb32960.ExtractFrames,
			Parse:        gb32960.ParseFrame,
			Authenticate: gb32960Authenticator,
			Respond:      gb32960.AutoResponse,
		},
		{
			Protocol:     envelope.ProtocolJT808,
			Addr:         env("JT808_TCP_ADDR", ":808"),
			Extract:      jt808.ExtractFrames,
			Parse:        jt808.ParseFrame,
			Authenticate: jt808Authenticator,
			Respond:      jt808.NewAutoResponder(jt808AuthCode).Respond,
		},
	}

	var wg sync.WaitGroup
	errs := make(chan error, len(protocols))
	for _, protocol := range protocols {
		server, err := gateway.NewTCPServer(gateway.TCPServerConfig{
			Protocol:       protocol,
			Sink:           sink,
			Resolver:       resolver,
			Logger:         logger,
			Metrics:        registry,
			ReadBufferSize: envInt("TCP_READ_BUFFER_BYTES", 64*1024),
			IdleTimeout:    time.Duration(envInt("TCP_IDLE_TIMEOUT_SECONDS", 180)) * time.Second,
			MaxConnections: envInt("TCP_MAX_CONNECTIONS", 120_000),
			PublishUnified: publishUnified,
			DelegateFields: delegateFields,
		})
		if err != nil {
			logger.Error("build tcp server failed", "protocol", protocol.Protocol, "error", err)
			os.Exit(1)
		}
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := server.ListenAndServe(ctx); err != nil {
				errs <- err
			}
		}()
	}
	if envBool("YUTONG_MQTT_ENABLED", false) {
		client, err := gateway.NewMQTTClient(gateway.MQTTClientConfig{
			EndpointName:            env("YUTONG_MQTT_ENDPOINT", env("YUTONG_MQTT_ENDPOINT_NAME", "yutong")),
			Broker:                  env("YUTONG_MQTT_URI", ""),
			ClientID:                env("YUTONG_MQTT_CLIENT_ID", "lingniu-go-yutong-mqtt"),
			Username:                env("YUTONG_MQTT_USERNAME", ""),
			Password:                env("YUTONG_MQTT_PASSWORD", ""),
			Topics:                  splitCSV(env("YUTONG_MQTT_TOPICS", env("YUTONG_MQTT_TOPIC", "/ytforward/shln/+"))),
			QoS:                     byte(envInt("YUTONG_MQTT_QOS", 2)),
			CleanSession:            envBool("YUTONG_MQTT_CLEAN_SESSION", false),
			KeepAlive:               time.Duration(envInt("YUTONG_MQTT_KEEP_ALIVE_SECONDS", 20)) * time.Second,
			ConnectTimeout:          time.Duration(envInt("YUTONG_MQTT_CONNECTION_TIMEOUT_SECONDS", 10)) * time.Second,
			TLSCACertPath:           env("YUTONG_MQTT_TLS_CA_PEM", ""),
			TLSClientCertPath:       env("YUTONG_MQTT_TLS_CLIENT_PEM", ""),
			TLSClientKeyPath:        env("YUTONG_MQTT_TLS_CLIENT_KEY", ""),
			TLSHostnameVerification: envBool("YUTONG_MQTT_TLS_HOSTNAME_VERIFICATION_ENABLED", true),
			Sink:                    sink,
			Resolver:                resolver,
			Logger:                  logger,
			Metrics:                 registry,
			PublishUnified:          publishUnified,
			DelegateFields:          delegateFields,
		})
		if err != nil {
			logger.Error("build yutong mqtt client failed", "error", err)
			os.Exit(1)
		}
		if err := client.Start(ctx); err != nil {
			logger.Error("start yutong mqtt client failed", "error", err)
			os.Exit(1)
		}
		logger.Info("yutong mqtt client started")
	}

	logger.Info("vehicle gateway started", "fields_derive_from_raw_enabled", delegateFields)
	select {
	case <-ctx.Done():
	case err := <-errs:
		logger.Error("vehicle gateway listener failed", "error", err)
		stop()
	}
	wg.Wait()
}

func buildGB32960Authenticator() (authentication.Authenticator, authentication.Mode, int, error) {
	mode, err := authentication.ParseMode(os.Getenv("GB32960_AUTH_MODE"), authentication.ModeObserve)
	if err != nil {
		return nil, "", 0, err
	}
	credentials, err := loadGB32960Credentials()
	if err != nil {
		return nil, "", 0, err
	}
	if mode == authentication.ModeEnforce && len(credentials) == 0 {
		return nil, "", 0, fmt.Errorf("GB32960_AUTH_MODE=enforce requires configured platform credentials")
	}
	return authentication.NewGB32960PlatformAuthenticator(mode, credentials), mode, len(credentials), nil
}

func buildJT808Authenticator(authCode string, deviceTokens authentication.JT808DeviceTokenProvider) (authentication.Authenticator, authentication.Mode, error) {
	mode, err := authentication.ParseMode(os.Getenv("JT808_AUTH_MODE"), authentication.ModeObserve)
	if err != nil {
		return nil, "", err
	}
	if mode == authentication.ModeEnforce && strings.TrimSpace(authCode) == "" && deviceTokens == nil {
		return nil, "", fmt.Errorf("JT808_AUTH_MODE=enforce requires a configured or device token provider")
	}
	return authentication.NewJT808Authenticator(mode, authCode, deviceTokens), mode, nil
}

func loadGB32960Credentials() (map[string][]string, error) {
	payload := strings.TrimSpace(os.Getenv("GB32960_PLATFORM_CREDENTIALS_JSON"))
	if path := strings.TrimSpace(os.Getenv("GB32960_PLATFORM_CREDENTIALS_FILE")); path != "" {
		contents, err := os.ReadFile(path)
		if err != nil {
			return nil, fmt.Errorf("read gb32960 credentials file: %w", err)
		}
		payload = strings.TrimSpace(string(contents))
	}
	if payload == "" {
		return map[string][]string{}, nil
	}
	rawCredentials := map[string]json.RawMessage{}
	if err := json.Unmarshal([]byte(payload), &rawCredentials); err != nil {
		return nil, fmt.Errorf("parse gb32960 platform credentials: %w", err)
	}
	credentials := make(map[string][]string, len(rawCredentials))
	for username, rawPassword := range rawCredentials {
		trimmed := strings.TrimSpace(username)
		if trimmed == "" {
			continue
		}
		var passwords []string
		var single string
		if err := json.Unmarshal(rawPassword, &single); err == nil {
			passwords = []string{single}
		} else if err := json.Unmarshal(rawPassword, &passwords); err != nil {
			return nil, fmt.Errorf("parse gb32960 credentials for account %q: expected string or string array", trimmed)
		}
		for _, password := range passwords {
			if password != "" {
				credentials[trimmed] = append(credentials[trimmed], password)
			}
		}
		if len(credentials[trimmed]) == 0 {
			delete(credentials, trimmed)
		}
	}
	return credentials, nil
}

func recordAuthenticationConfig(registry *metrics.Registry, protocol envelope.Protocol, mode authentication.Mode, credentialCount int) {
	if registry == nil {
		return
	}
	registry.SetGauge("vehicle_gateway_authentication_mode", metrics.Labels{
		"protocol": string(protocol),
		"mode":     string(mode),
	}, 1)
	registry.SetGauge("vehicle_gateway_authentication_credentials", metrics.Labels{
		"protocol": string(protocol),
	}, float64(credentialCount))
}

func buildIdentityResolver(ctx context.Context, logger *slog.Logger, registry *metrics.Registry) (identity.Resolver, authentication.JT808DeviceTokenProvider, func(), error) {
	dsn := env("IDENTITY_MYSQL_DSN", strings.TrimSpace(os.Getenv("MYSQL_DSN")))
	if strings.TrimSpace(dsn) == "" {
		logger.Warn("identity mysql dsn is empty; using noop identity resolver")
		return identity.NoopResolver{}, nil, func() {}, nil
	}
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		logger.Error("identity mysql configuration failed; continuing with unresolved identities", "error", err)
		return identity.NoopResolver{}, nil, func() {}, nil
	}
	db.SetMaxOpenConns(envInt("IDENTITY_MYSQL_MAX_OPEN_CONNS", 16))
	db.SetMaxIdleConns(envInt("IDENTITY_MYSQL_MAX_IDLE_CONNS", 8))
	db.SetConnMaxLifetime(time.Duration(envInt("IDENTITY_MYSQL_CONN_MAX_LIFETIME_SECONDS", 300)) * time.Second)
	db.SetConnMaxIdleTime(time.Duration(envInt("IDENTITY_MYSQL_CONN_MAX_IDLE_SECONDS", 60)) * time.Second)
	table := env("VEHICLE_IDENTITY_TABLE", "vehicle_identity_binding")
	cacheMaxEntries := envInt("IDENTITY_LOOKUP_CACHE_MAX_ENTRIES", 300000)
	cacheCleanupInterval := time.Duration(envInt("IDENTITY_LOOKUP_CACHE_CLEANUP_INTERVAL_SECONDS", 60)) * time.Second
	staleLookupTTLSeconds := envInt("IDENTITY_STALE_LOOKUP_TTL_SECONDS", 3600)
	registrationGatewayWritesEnabled := envBool("JT808_REGISTRATION_GATEWAY_WRITES_ENABLED", false)
	resolver := identity.NewMySQLResolverWithOptions(db, table, identity.MySQLResolverOptions{
		SnapshotOnlyLookups:         envBool("IDENTITY_SNAPSHOT_ONLY_ENABLED", true),
		LocationTouchInterval:       time.Duration(envInt("JT808_REGISTRATION_LOCATION_TOUCH_INTERVAL_SECONDS", 600)) * time.Second,
		LocationTouchRetryInterval:  time.Duration(envInt("JT808_REGISTRATION_LOCATION_TOUCH_RETRY_INTERVAL_SECONDS", 5)) * time.Second,
		RegistrationWriteAttempts:   envInt("JT808_REGISTRATION_WRITE_RETRY_ATTEMPTS", 2),
		RegistrationWriteRetryDelay: time.Duration(envInt("JT808_REGISTRATION_WRITE_RETRY_DELAY_MS", 20)) * time.Millisecond,
		LookupCacheTTL:              time.Duration(envInt("IDENTITY_LOOKUP_CACHE_TTL_SECONDS", 600)) * time.Second,
		StaleLookupTTL:              time.Duration(staleLookupTTLSeconds) * time.Second,
		CacheCleanupInterval:        cacheCleanupInterval,
		MaxCacheEntries:             cacheMaxEntries,
		SourceCodeLookup:            envBool("IDENTITY_SOURCE_CODE_LOOKUP_ENABLED", true),
		AsyncRegistrationWrites:     envBool("JT808_REGISTRATION_ASYNC_WRITE_ENABLED", true),
		RegistrationWriteQueueSize:  envInt("JT808_REGISTRATION_WRITE_QUEUE_SIZE", 100000),
		RegistrationWriteWorkers:    envInt("JT808_REGISTRATION_WRITE_WORKERS", 4),
		RegistrationWriteTimeout:    time.Duration(envInt("JT808_REGISTRATION_WRITE_TIMEOUT_MS", 5000)) * time.Millisecond,
		RegistrationEnqueueTimeout:  time.Duration(envInt("JT808_REGISTRATION_WRITE_ENQUEUE_TIMEOUT_MS", 50)) * time.Millisecond,
		DisableRegistrationWrites:   !registrationGatewayWritesEnabled,
		OnRegistrationWriteResult: func(result identity.RegistrationWriteResult) {
			recordJT808RegistrationWriteResult(registry, result)
		},
		OnRegistrationWriteError: func(err error) {
			logger.Warn("jt808 registration async write failed", "error", err)
		},
	})
	startIdentityDatabaseMaintenance(
		ctx,
		logger,
		db,
		resolver,
		envBool("IDENTITY_MYSQL_ENSURE_SCHEMA", false),
		time.Duration(envInt("IDENTITY_MYSQL_PING_TIMEOUT_MS", 3000))*time.Millisecond,
		time.Duration(envInt("IDENTITY_MYSQL_SCHEMA_TIMEOUT_SECONDS", 10))*time.Second,
	)
	startIdentitySnapshotRefresh(
		ctx,
		logger,
		registry,
		resolver,
		time.Duration(envInt("IDENTITY_SNAPSHOT_REFRESH_INTERVAL_SECONDS", 60))*time.Second,
		time.Duration(envInt("IDENTITY_SNAPSHOT_REFRESH_TIMEOUT_SECONDS", 10))*time.Second,
	)
	startIdentityCacheMetrics(ctx, registry, resolver, time.Duration(envInt("IDENTITY_CACHE_METRICS_INTERVAL_SECONDS", 30))*time.Second)
	resolveTimeout := time.Duration(envInt("IDENTITY_RESOLVE_TIMEOUT_MS", 50)) * time.Millisecond
	registry.SetGauge("vehicle_gateway_jt808_registration_gateway_writes_enabled", nil, boolMetric(registrationGatewayWritesEnabled))
	logger.Info("identity mysql resolver enabled", "table", table, "snapshot_only_enabled", envBool("IDENTITY_SNAPSHOT_ONLY_ENABLED", true), "snapshot_refresh_interval_seconds", envInt("IDENTITY_SNAPSHOT_REFRESH_INTERVAL_SECONDS", 60), "lookup_cache_ttl_seconds", envInt("IDENTITY_LOOKUP_CACHE_TTL_SECONDS", 600), "stale_lookup_ttl_seconds", staleLookupTTLSeconds, "lookup_cache_max_entries", cacheMaxEntries, "lookup_cache_cleanup_interval_seconds", cacheCleanupInterval.Seconds(), "source_code_lookup_enabled", envBool("IDENTITY_SOURCE_CODE_LOOKUP_ENABLED", true), "resolve_timeout_ms", resolveTimeout.Milliseconds(), "jt808_registration_gateway_writes_enabled", registrationGatewayWritesEnabled, "jt808_registration_location_touch_retry_interval_seconds", envInt("JT808_REGISTRATION_LOCATION_TOUCH_RETRY_INTERVAL_SECONDS", 5), "jt808_registration_write_retry_attempts", envInt("JT808_REGISTRATION_WRITE_RETRY_ATTEMPTS", 2), "jt808_registration_write_retry_delay_ms", envInt("JT808_REGISTRATION_WRITE_RETRY_DELAY_MS", 20), "jt808_registration_async_write_enabled", envBool("JT808_REGISTRATION_ASYNC_WRITE_ENABLED", true), "jt808_registration_write_queue_size", envInt("JT808_REGISTRATION_WRITE_QUEUE_SIZE", 100000), "jt808_registration_write_workers", envInt("JT808_REGISTRATION_WRITE_WORKERS", 4), "jt808_registration_write_timeout_ms", envInt("JT808_REGISTRATION_WRITE_TIMEOUT_MS", 5000), "jt808_registration_write_enqueue_timeout_ms", envInt("JT808_REGISTRATION_WRITE_ENQUEUE_TIMEOUT_MS", 50))
	return identity.TimeoutResolver{Delegate: resolver, Timeout: resolveTimeout}, resolver, func() {
		_ = resolver.Close()
		_ = db.Close()
	}, nil
}

type identitySnapshotRefresher interface {
	RefreshSnapshot(context.Context) (identity.SnapshotRefreshResult, error)
}

type identityDatabasePinger interface {
	PingContext(context.Context) error
}

type identitySchemaEnsurer interface {
	EnsureSchema(context.Context) error
}

func startIdentityDatabaseMaintenance(ctx context.Context, logger *slog.Logger, db identityDatabasePinger, schema identitySchemaEnsurer, ensureSchema bool, pingTimeout time.Duration, schemaTimeout time.Duration) {
	if db == nil {
		return
	}
	if pingTimeout <= 0 {
		pingTimeout = 3 * time.Second
	}
	if schemaTimeout <= 0 {
		schemaTimeout = 10 * time.Second
	}
	go func() {
		pingCtx, cancelPing := context.WithTimeout(ctx, pingTimeout)
		err := db.PingContext(pingCtx)
		cancelPing()
		if err != nil {
			logger.Warn("identity mysql unavailable; ingress remains enabled and snapshot refresh will retry", "error", err)
			return
		}
		if !ensureSchema || schema == nil {
			return
		}
		schemaCtx, cancelSchema := context.WithTimeout(ctx, schemaTimeout)
		err = schema.EnsureSchema(schemaCtx)
		cancelSchema()
		if err != nil {
			logger.Warn("identity mysql schema check failed; ingress remains enabled", "error", err)
		}
	}()
}

func startIdentitySnapshotRefresh(ctx context.Context, logger *slog.Logger, registry *metrics.Registry, refresher identitySnapshotRefresher, interval time.Duration, timeout time.Duration) {
	if refresher == nil {
		return
	}
	if interval <= 0 {
		interval = time.Minute
	}
	if timeout <= 0 {
		timeout = 10 * time.Second
	}
	refresh := func() {
		refreshCtx, cancel := context.WithTimeout(ctx, timeout)
		result, err := refresher.RefreshSnapshot(refreshCtx)
		cancel()
		if err != nil {
			recordIdentitySnapshotRefresh(registry, identity.SnapshotRefreshResult{}, "error")
			logger.Warn("identity snapshot refresh failed; keeping last known good snapshot", "error", err)
			return
		}
		recordIdentitySnapshotRefresh(registry, result, "ok")
		logger.Info("identity snapshot refreshed", "binding_entries", result.BindingEntries, "identifier_entries", result.IdentifierEntries, "registration_entries", result.RegistrationEntries, "source_entries", result.SourceEntries)
	}
	go func() {
		refresh()
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				refresh()
			}
		}
	}()
}

func recordIdentitySnapshotRefresh(registry *metrics.Registry, result identity.SnapshotRefreshResult, status string) {
	if registry == nil {
		return
	}
	registry.IncCounter("vehicle_gateway_identity_snapshot_refresh_total", metrics.Labels{"status": status})
	if status != "ok" {
		return
	}
	for _, item := range []struct {
		kind  string
		value int
	}{
		{kind: "binding", value: result.BindingEntries},
		{kind: "identifier", value: result.IdentifierEntries},
		{kind: "registration", value: result.RegistrationEntries},
		{kind: "source", value: result.SourceEntries},
	} {
		registry.SetGauge("vehicle_gateway_identity_snapshot_entries", metrics.Labels{"kind": item.kind}, float64(item.value))
	}
	registry.SetGauge("vehicle_gateway_identity_snapshot_last_success_unix_seconds", nil, float64(result.RefreshedAt.Unix()))
}

type identityCacheStatsReporter interface {
	CacheStats() identity.CacheStats
}

func startIdentityCacheMetrics(ctx context.Context, registry *metrics.Registry, reporter identityCacheStatsReporter, interval time.Duration) {
	if registry == nil || reporter == nil {
		return
	}
	if interval <= 0 {
		interval = 30 * time.Second
	}
	recordIdentityCacheStats(registry, reporter.CacheStats())
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				recordIdentityCacheStats(registry, reporter.CacheStats())
			}
		}
	}()
}

func recordIdentityCacheStats(registry *metrics.Registry, stats identity.CacheStats) {
	if registry == nil {
		return
	}
	for _, item := range []struct {
		name  string
		value int
		max   int
	}{
		{name: "lookup", value: stats.LookupEntries, max: stats.MaxEntries},
		{name: "registration", value: stats.RegistrationEntries, max: stats.MaxEntries},
		{name: "source_code", value: stats.SourceCodeEntries, max: stats.MaxEntries},
		{name: "location_touch", value: stats.LocationTouchEntries, max: stats.MaxEntries},
		{name: "location_touch_failure", value: stats.LocationTouchFailureEntries, max: stats.MaxEntries},
		{name: "registration_write_queue", value: stats.RegistrationWriteQueueDepth, max: stats.RegistrationWriteQueueCap},
		{name: "snapshot_binding", value: stats.SnapshotBindingEntries, max: stats.MaxEntries},
		{name: "snapshot_identifier", value: stats.SnapshotIdentifierEntries, max: stats.MaxEntries},
		{name: "snapshot_registration", value: stats.SnapshotRegistrationEntries, max: stats.MaxEntries},
		{name: "snapshot_source", value: stats.SnapshotSourceEntries, max: stats.MaxEntries},
	} {
		registry.SetGauge("vehicle_gateway_identity_cache_entries", metrics.Labels{"cache": item.name}, float64(item.value))
		registry.SetGauge("vehicle_gateway_identity_cache_max_entries", metrics.Labels{"cache": item.name}, float64(item.max))
	}
	ready := 0.0
	if stats.SnapshotReady {
		ready = 1
	}
	registry.SetGauge("vehicle_gateway_identity_snapshot_ready", nil, ready)
	if !stats.SnapshotRefreshedAt.IsZero() {
		registry.SetGauge("vehicle_gateway_identity_snapshot_last_success_unix_seconds", nil, float64(stats.SnapshotRefreshedAt.Unix()))
	}
}

func recordJT808RegistrationWriteResult(registry *metrics.Registry, result identity.RegistrationWriteResult) {
	if registry == nil {
		return
	}
	mode := strings.TrimSpace(result.Mode)
	if mode == "" {
		mode = "unknown"
	}
	status := strings.TrimSpace(result.Status)
	if status == "" {
		status = "unknown"
	}
	registry.IncCounter("vehicle_gateway_jt808_registration_write_total", metrics.Labels{
		"mode":   mode,
		"status": status,
	})
	metrics.RecordLastActivity(registry, "vehicle_gateway_last_jt808_registration_write_unix_seconds", metrics.Labels{
		"mode":   mode,
		"status": status,
	})
}

func buildSink(ctx context.Context, logger *slog.Logger, registry *metrics.Registry) (eventbus.Sink, error) {
	if strings.TrimSpace(os.Getenv("NATS_URL")) != "" {
		natsConfig := natsSinkConfigFromEnv()
		sink, err := eventbus.NewNATSSink(natsConfig)
		if err != nil {
			return nil, err
		}
		outboxRuntime := natsOutboxConfigFromEnv(registry, func(err error) {
			logger.Warn("nats durable outbox publish failed", "error", err)
		})
		if outboxRuntime.Enabled {
			outbox, err := eventbus.NewDurableOutboxSink(sink, outboxRuntime.Config)
			if err != nil {
				_ = sink.Close()
				return nil, err
			}
			go outbox.ReplayLoop(ctx, outboxRuntime.ReplayInterval)
			logger.Info("nats durable outbox enabled",
				"dir", outboxRuntime.Config.Directory,
				"fsync", outboxRuntime.Config.SyncWrites,
				"replay_interval_ms", outboxRuntime.ReplayInterval.Milliseconds(),
				"replay_batch_size", outboxRuntime.Config.ReplayBatchSize,
				"close_timeout_ms", outboxRuntime.Config.CloseTimeout.Milliseconds(),
				"wal_segment_bytes", outboxRuntime.Config.WALSegmentBytes,
				"wal_segment_age_ms", outboxRuntime.Config.WALSegmentAge.Milliseconds(),
				"wal_append_queue_size", outboxRuntime.Config.WALAppendQueue,
				"wal_commit_batch_size", outboxRuntime.Config.WALCommitBatch,
				"wal_commit_interval_ms", outboxRuntime.Config.WALCommitWait.Milliseconds(),
				"max_inflight", natsConfig.AsyncMaxPending,
				"ack_timeout_ms", natsConfig.AsyncAckTimeout.Milliseconds(),
			)
			return outbox, nil
		}
		var out eventbus.Sink = eventbus.NewRetryingSink(sink, eventbus.RetryConfig{
			Attempts:       envInt("NATS_PUBLISH_ATTEMPTS", 3),
			Backoff:        time.Duration(envInt("NATS_PUBLISH_BACKOFF_MS", 100)) * time.Millisecond,
			AttemptTimeout: time.Duration(envInt("NATS_PUBLISH_TIMEOUT_MS", 3000)) * time.Millisecond,
		})
		spoolDir := strings.TrimSpace(os.Getenv("NATS_SPOOL_DIR"))
		if spoolDir != "" {
			replayBatchSize := envInt("NATS_SPOOL_REPLAY_BATCH_SIZE", 200)
			durable := eventbus.NewDurableSink(out, eventbus.DurableConfig{
				Directory:       spoolDir,
				ReplayBatchSize: replayBatchSize,
				Metrics:         registry,
				Name:            "nats",
			})
			interval := time.Duration(envInt("NATS_SPOOL_REPLAY_INTERVAL_MS", 1000)) * time.Millisecond
			go durable.ReplayLoop(ctx, interval, func(err error) {
				logger.Warn("nats spool replay failed", "error", err)
			})
			logger.Info("nats durable spool enabled", "dir", spoolDir, "replay_interval_ms", interval.Milliseconds(), "replay_batch_size", replayBatchSize)
			out = durable
		}
		if envBool("NATS_ASYNC_ENABLED", true) {
			queueSize := envInt("NATS_ASYNC_QUEUE_SIZE", envInt("KAFKA_ASYNC_QUEUE_SIZE", 100000))
			workers := envInt("NATS_ASYNC_WORKERS", envInt("KAFKA_ASYNC_WORKERS", 8))
			enqueueTimeout := time.Duration(envInt("NATS_ASYNC_ENQUEUE_TIMEOUT_MS", envInt("KAFKA_ASYNC_ENQUEUE_TIMEOUT_MS", 1000))) * time.Millisecond
			timeout := time.Duration(envInt("NATS_ASYNC_PUBLISH_TIMEOUT_MS", envInt("KAFKA_ASYNC_PUBLISH_TIMEOUT_MS", 30000))) * time.Millisecond
			if envBool("NATS_PARTITIONED_ASYNC_ENABLED", true) {
				rawQueueSize, derivedQueueSize, rawWorkers, derivedWorkers := partitionedAsyncConfigFromEnv("NATS", queueSize, workers)
				rawEnqueueTimeout, derivedEnqueueTimeout := partitionedAsyncEnqueueTimeoutsFromEnv("NATS", enqueueTimeout)
				out = eventbus.NewPartitionedAsyncSink(out, eventbus.PartitionedAsyncConfig{
					RawQueueSize:          rawQueueSize,
					DerivedQueueSize:      derivedQueueSize,
					RawWorkers:            rawWorkers,
					DerivedWorkers:        derivedWorkers,
					RawEnqueueTimeout:     rawEnqueueTimeout,
					DerivedEnqueueTimeout: derivedEnqueueTimeout,
					EnqueueTimeout:        enqueueTimeout,
					OperationTimeout:      timeout,
					Metrics:               registry,
					Name:                  "nats",
					OnError: func(err error) {
						logger.Warn("nats partitioned async publish failed", "error", err)
					},
				})
				logger.Info("nats partitioned async publish enabled", "raw_queue_size", rawQueueSize, "derived_queue_size", derivedQueueSize, "raw_workers", rawWorkers, "derived_workers", derivedWorkers, "raw_enqueue_timeout_ms", rawEnqueueTimeout.Milliseconds(), "derived_enqueue_timeout_ms", derivedEnqueueTimeout.Milliseconds(), "publish_timeout_ms", timeout.Milliseconds())
			} else {
				out = eventbus.NewAsyncSink(out, eventbus.AsyncConfig{
					QueueSize:        queueSize,
					Workers:          workers,
					EnqueueTimeout:   enqueueTimeout,
					OperationTimeout: timeout,
					Metrics:          registry,
					Name:             "nats",
					OnError: func(err error) {
						logger.Warn("nats async publish failed", "error", err)
					},
				})
				logger.Info("nats async publish enabled", "queue_size", queueSize, "workers", workers, "enqueue_timeout_ms", enqueueTimeout.Milliseconds(), "publish_timeout_ms", timeout.Milliseconds())
			}
		}
		logger.Info("nats jetstream sink enabled", "url", env("NATS_URL", ""), "unified_subject", natsSinkConfigFromEnv().UnifiedSubject, "publish_unified_enabled", envBool("PUBLISH_UNIFIED_ENABLED", false), "publish_fields_enabled", !envBool("FIELDS_DERIVE_FROM_RAW_ENABLED", true))
		return out, nil
	}
	brokers := splitCSV(os.Getenv("KAFKA_BROKERS"))
	if len(brokers) == 0 {
		logger.Warn("KAFKA_BROKERS is empty; using log sink")
		return eventbus.NewLogSink(logger), nil
	}
	sink, err := eventbus.NewKafkaSink(eventbus.KafkaConfig{
		Brokers: brokers,
		RawTopics: map[envelope.Protocol]string{
			envelope.ProtocolGB32960:    env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960),
			envelope.ProtocolJT808:      env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808),
			envelope.ProtocolYutongMQTT: env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT),
		},
		FieldsTopics: map[envelope.Protocol]string{
			envelope.ProtocolGB32960:    env("KAFKA_TOPIC_GB32960_FIELDS", topics.FieldsGB32960),
			envelope.ProtocolJT808:      env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808),
			envelope.ProtocolYutongMQTT: env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT),
		},
		UnifiedTopic: env("KAFKA_TOPIC_UNIFIED", topics.Unified),
	})
	if err != nil {
		return nil, err
	}
	var out eventbus.Sink = eventbus.NewRetryingSink(sink, eventbus.RetryConfig{
		Attempts:       envInt("KAFKA_PUBLISH_ATTEMPTS", 3),
		Backoff:        time.Duration(envInt("KAFKA_PUBLISH_BACKOFF_MS", 100)) * time.Millisecond,
		AttemptTimeout: time.Duration(envInt("KAFKA_PUBLISH_TIMEOUT_MS", 3000)) * time.Millisecond,
	})
	spoolDir := strings.TrimSpace(os.Getenv("KAFKA_SPOOL_DIR"))
	if spoolDir != "" {
		replayBatchSize := envInt("KAFKA_SPOOL_REPLAY_BATCH_SIZE", 200)
		durable := eventbus.NewDurableSink(out, eventbus.DurableConfig{
			Directory:       spoolDir,
			ReplayBatchSize: replayBatchSize,
			Metrics:         registry,
			Name:            "kafka",
		})
		interval := time.Duration(envInt("KAFKA_SPOOL_REPLAY_INTERVAL_MS", 1000)) * time.Millisecond
		go durable.ReplayLoop(ctx, interval, func(err error) {
			logger.Warn("kafka spool replay failed", "error", err)
		})
		logger.Info("kafka durable spool enabled", "dir", spoolDir, "replay_interval_ms", interval.Milliseconds(), "replay_batch_size", replayBatchSize)
		out = durable
	}
	if envBool("KAFKA_ASYNC_ENABLED", true) {
		queueSize := envInt("KAFKA_ASYNC_QUEUE_SIZE", 100000)
		workers := envInt("KAFKA_ASYNC_WORKERS", 8)
		enqueueTimeout := time.Duration(envInt("KAFKA_ASYNC_ENQUEUE_TIMEOUT_MS", 1000)) * time.Millisecond
		timeout := time.Duration(envInt("KAFKA_ASYNC_PUBLISH_TIMEOUT_MS", 30000)) * time.Millisecond
		if envBool("KAFKA_PARTITIONED_ASYNC_ENABLED", true) {
			rawQueueSize, derivedQueueSize, rawWorkers, derivedWorkers := partitionedAsyncConfigFromEnv("KAFKA", queueSize, workers)
			rawEnqueueTimeout, derivedEnqueueTimeout := partitionedAsyncEnqueueTimeoutsFromEnv("KAFKA", enqueueTimeout)
			out = eventbus.NewPartitionedAsyncSink(out, eventbus.PartitionedAsyncConfig{
				RawQueueSize:          rawQueueSize,
				DerivedQueueSize:      derivedQueueSize,
				RawWorkers:            rawWorkers,
				DerivedWorkers:        derivedWorkers,
				RawEnqueueTimeout:     rawEnqueueTimeout,
				DerivedEnqueueTimeout: derivedEnqueueTimeout,
				EnqueueTimeout:        enqueueTimeout,
				OperationTimeout:      timeout,
				Metrics:               registry,
				Name:                  "kafka",
				OnError: func(err error) {
					logger.Warn("kafka partitioned async publish failed", "error", err)
				},
			})
			logger.Info("kafka partitioned async publish enabled", "raw_queue_size", rawQueueSize, "derived_queue_size", derivedQueueSize, "raw_workers", rawWorkers, "derived_workers", derivedWorkers, "raw_enqueue_timeout_ms", rawEnqueueTimeout.Milliseconds(), "derived_enqueue_timeout_ms", derivedEnqueueTimeout.Milliseconds(), "publish_timeout_ms", timeout.Milliseconds())
		} else {
			out = eventbus.NewAsyncSink(out, eventbus.AsyncConfig{
				QueueSize:        queueSize,
				Workers:          workers,
				EnqueueTimeout:   enqueueTimeout,
				OperationTimeout: timeout,
				Metrics:          registry,
				Name:             "kafka",
				OnError: func(err error) {
					logger.Warn("kafka async publish failed", "error", err)
				},
			})
			logger.Info("kafka async publish enabled", "queue_size", queueSize, "workers", workers, "enqueue_timeout_ms", enqueueTimeout.Milliseconds(), "publish_timeout_ms", timeout.Milliseconds())
		}
	}
	return out, nil
}

func partitionedAsyncConfigFromEnv(prefix string, queueSize int, workers int) (rawQueueSize int, derivedQueueSize int, rawWorkers int, derivedWorkers int) {
	derivedQueueDefault := queueSize / 2
	if derivedQueueDefault <= 0 {
		derivedQueueDefault = 1
	}
	derivedWorkersDefault := workers / 2
	if derivedWorkersDefault <= 0 {
		derivedWorkersDefault = 1
	}
	rawQueueSize = envInt(prefix+"_ASYNC_RAW_QUEUE_SIZE", queueSize)
	derivedQueueSize = envInt(prefix+"_ASYNC_DERIVED_QUEUE_SIZE", derivedQueueDefault)
	rawWorkers = envInt(prefix+"_ASYNC_RAW_WORKERS", workers)
	derivedWorkers = envInt(prefix+"_ASYNC_DERIVED_WORKERS", derivedWorkersDefault)
	return rawQueueSize, derivedQueueSize, rawWorkers, derivedWorkers
}

func partitionedAsyncEnqueueTimeoutsFromEnv(prefix string, enqueueTimeout time.Duration) (rawEnqueueTimeout time.Duration, derivedEnqueueTimeout time.Duration) {
	rawEnqueueTimeout = time.Duration(envInt(prefix+"_ASYNC_RAW_ENQUEUE_TIMEOUT_MS", int(enqueueTimeout/time.Millisecond))) * time.Millisecond
	derivedEnqueueTimeout = time.Duration(envInt(prefix+"_ASYNC_DERIVED_ENQUEUE_TIMEOUT_MS", 50)) * time.Millisecond
	return rawEnqueueTimeout, derivedEnqueueTimeout
}

func natsSinkConfigFromEnv() eventbus.NATSConfig {
	return eventbus.NATSConfig{
		URL:             env("NATS_URL", ""),
		Name:            env("NATS_CLIENT_NAME", "lingniu-vehicle-gateway"),
		AsyncMaxPending: envInt("NATS_OUTBOX_MAX_INFLIGHT", 10_000),
		AsyncAckTimeout: time.Duration(envInt("NATS_OUTBOX_ACK_TIMEOUT_MS", 3000)) * time.Millisecond,
		RawSubjects: map[envelope.Protocol]string{
			envelope.ProtocolGB32960:    env("NATS_SUBJECT_GB32960_RAW", env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960)),
			envelope.ProtocolJT808:      env("NATS_SUBJECT_JT808_RAW", env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808)),
			envelope.ProtocolYutongMQTT: env("NATS_SUBJECT_YUTONG_MQTT_RAW", env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT)),
		},
		FieldsSubjects: map[envelope.Protocol]string{
			envelope.ProtocolGB32960:    env("NATS_SUBJECT_GB32960_FIELDS", env("KAFKA_TOPIC_GB32960_FIELDS", topics.FieldsGB32960)),
			envelope.ProtocolJT808:      env("NATS_SUBJECT_JT808_FIELDS", env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808)),
			envelope.ProtocolYutongMQTT: env("NATS_SUBJECT_YUTONG_MQTT_FIELDS", env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT)),
		},
		UnifiedSubject: env("NATS_SUBJECT_UNIFIED", env("KAFKA_TOPIC_UNIFIED", topics.Unified)),
	}
}

type natsOutboxRuntimeConfig struct {
	Enabled        bool
	Config         eventbus.DurableOutboxConfig
	ReplayInterval time.Duration
}

func natsOutboxConfigFromEnv(registry *metrics.Registry, onError func(error)) natsOutboxRuntimeConfig {
	directory := strings.TrimSpace(os.Getenv("NATS_OUTBOX_DIR"))
	return natsOutboxRuntimeConfig{
		Enabled: envBool("NATS_DURABLE_OUTBOX_ENABLED", false),
		Config: eventbus.DurableOutboxConfig{
			Directory:       directory,
			ReplayBatchSize: envInt("NATS_OUTBOX_REPLAY_BATCH_SIZE", 1000),
			SyncWrites:      envBool("NATS_OUTBOX_FSYNC", true),
			CloseTimeout:    time.Duration(envInt("NATS_OUTBOX_CLOSE_TIMEOUT_MS", 5000)) * time.Millisecond,
			WALSegmentBytes: int64(envInt("NATS_OUTBOX_WAL_SEGMENT_BYTES", 16<<20)),
			WALSegmentAge:   time.Duration(envInt("NATS_OUTBOX_WAL_SEGMENT_AGE_MS", 5000)) * time.Millisecond,
			WALAppendQueue:  envInt("NATS_OUTBOX_WAL_APPEND_QUEUE_SIZE", 100_000),
			WALCommitBatch:  envInt("NATS_OUTBOX_WAL_COMMIT_BATCH_SIZE", 256),
			WALCommitWait:   time.Duration(envInt("NATS_OUTBOX_WAL_COMMIT_INTERVAL_MS", 1)) * time.Millisecond,
			Metrics:         registry,
			Name:            "nats-outbox",
			OnError:         onError,
		},
		ReplayInterval: time.Duration(envInt("NATS_OUTBOX_REPLAY_INTERVAL_MS", 1000)) * 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
	}
	var out int
	for _, r := range value {
		if r < '0' || r > '9' {
			return fallback
		}
		out = out*10 + int(r-'0')
	}
	if out <= 0 {
		return fallback
	}
	return out
}

func envBool(key string, fallback bool) bool {
	value := strings.ToLower(strings.TrimSpace(os.Getenv(key)))
	if value == "" {
		return fallback
	}
	return value == "1" || value == "true" || value == "yes" || value == "on"
}

func boolMetric(value bool) float64 {
	if value {
		return 1
	}
	return 0
}

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
}