package main

import (
	"context"
	cryptorand "crypto/rand"
	"encoding/hex"
	"errors"
	"flag"
	"fmt"
	"net"
	"net/http"
	"os"
	"os/signal"
	"runtime"
	"strconv"
	"syscall"
	"time"

	mqtt "github.com/mochi-mqtt/server/v2"
	mqttauth "github.com/mochi-mqtt/server/v2/hooks/auth"
	"github.com/mochi-mqtt/server/v2/listeners"
	"github.com/mochi-mqtt/server/v2/packets"

	"meshtastic_mqtt_server/internal/ai"
	"meshtastic_mqtt_server/internal/auth"
	"meshtastic_mqtt_server/internal/autoreply"
	blockingpkg "meshtastic_mqtt_server/internal/blocking"
	botpkg "meshtastic_mqtt_server/internal/bot"
	configpkg "meshtastic_mqtt_server/internal/config"
	"meshtastic_mqtt_server/internal/llm"
	"meshtastic_mqtt_server/internal/mqtpp"
	mqttforwardpkg "meshtastic_mqtt_server/internal/mqttforward"
	rspkg "meshtastic_mqtt_server/internal/runtimesettings"
	storepkg "meshtastic_mqtt_server/internal/store"
	webpkg "meshtastic_mqtt_server/internal/web"
)

const (
	ansiGreenBGWhiteText  = "\033[42;37m"
	ansiBlueBGWhiteText   = "\033[44;37m"
	ansiPurpleBGWhiteText = "\033[45;37m"
	ansiCyanBGBlackText   = "\033[46;30m"
	ansiYellowBGBlackText = "\033[43;30m"
	ansiGrayBGWhiteText   = "\033[100;37m"
	ansiRedBGWhiteText    = "\033[41;37m"
	ansiReset             = "\033[0m"
)

type meshtasticFilterHook struct {
	mqtt.HookBase
	server          *mqtt.Server
	key             []byte
	dbQueue         *storepkg.WriteQueue
	stats           *mqttforwardpkg.Stats
	clientStats     *mqttforwardpkg.ClientStats
	blocking        *blockingpkg.Cache
	settings        *rspkg.Cache
	pkiResolver     func(toNodeNum, fromNodeNum uint32) ([]byte, []byte, bool)
	autoAcker       func(record map[string]any)
	consoleLog       bool // 控制台是否打印 MQTT 连接/订阅事件
	packetConsoleLog bool // 控制台是否打印 Meshtastic 数据包
	dedupQueue       *mqttforwardpkg.DedupQueue
}

// ID 返回用于识别 Meshtastic payload 过滤器的 hook 名称。
func (h *meshtasticFilterHook) ID() string {
	return "meshtastic-filter"
}

// Provides 声明该 hook 处理客户端连接、订阅与发布事件。
func (h *meshtasticFilterHook) Provides(b byte) bool {
	return b == mqtt.OnConnect ||
		b == mqtt.OnPublish ||
		b == mqtt.OnSessionEstablished ||
		b == mqtt.OnDisconnect ||
		b == mqtt.OnSubscribed ||
		b == mqtt.OnUnsubscribed ||
		b == mqtt.OnPacketRead ||
		b == mqtt.OnPacketSent
}

// OnConnect 在 MQTT 会话建立前拒绝命中 IP 屏蔽表的客户端。
func (h *meshtasticFilterHook) OnConnect(cl *mqtt.Client, pk packets.Packet) error {
	// 启用 TCP_NODELAY 禁用 Nagle 算法，确保小数据包（包括 TCP ACK）立即发送
	// 这对于 MQTT QoS0 消息特别重要，避免设备因为等待 TCP ACK 而重发
	if cl.Net.Conn != nil {
		if tcpConn, ok := cl.Net.Conn.(*net.TCPConn); ok {
			if err := tcpConn.SetNoDelay(true); err != nil {
				printJSON(map[string]any{"event": "tcp_nodelay_failed", "error": err.Error(), "remote_addr": cl.Net.Remote})
			}
		}
	}

	info := mqttClientInfoFromClient(cl)
	if h.blocking != nil && h.blocking.IsIPBlocked(info.RemoteHost) {
		printJSON(map[string]any{"event": "mqtt_client_rejected", "reason": "blocked_ip", "client_id": info.ClientID, "remote_addr": info.RemoteAddr, "remote_host": info.RemoteHost})
		return packets.ErrNotAuthorized
	}
	// 如果该 client_id 当前已经有活动连接（典型场景：同一 Meshtastic 节点被两台 Android
	// 同时连），broker 默认会按 [MQTT-3.1.4-3] 把旧连接顶下线，导致两边互相踢、日志被刷屏。
	// 这里给后来者随机加个后缀，避免顶号——CONNACK 不会回传新 ID，但客户端只需要能正常
	// 收发消息即可，订阅状态在断开后自然清空。
	if h.server != nil && cl.ID != "" {
		if existing, ok := h.server.Clients.Get(cl.ID); ok && existing != nil && !existing.Closed() && existing != cl {
			original := cl.ID
			cl.ID = original + "-" + randomClientIDSuffix()
			printJSON(map[string]any{
				"event":          "mqtt_client_id_renamed",
				"reason":         "duplicate_client_id",
				"original_id":    original,
				"assigned_id":    cl.ID,
				"remote_addr":    info.RemoteAddr,
				"remote_host":    info.RemoteHost,
				"existing_remote": existing.Net.Remote,
			})
		}
	}
	return nil
}

// randomClientIDSuffix 生成 4 字符的小写 hex 后缀，用于在 client_id 冲突时给后来的连接重命名。
// 用 crypto/rand 是为了避免热路径上 math/rand 的并发锁；2 byte 的熵已经足够区分。
func randomClientIDSuffix() string {
	var b [2]byte
	if _, err := cryptorand.Read(b[:]); err != nil {
		// 极端情况下退到时间戳低 16 位作为兜底
		fallback := uint16(time.Now().UnixNano())
		return fmt.Sprintf("%04x", fallback)
	}
	return hex.EncodeToString(b[:])
}

// OnSessionEstablished 在客户端通过认证、会话建立后打印连接日志。
func (h *meshtasticFilterHook) OnSessionEstablished(cl *mqtt.Client, pk packets.Packet) {
	if !h.consoleLog {
		return
	}
	info := mqttClientInfoFromClient(cl)
	fmt.Fprintf(os.Stderr, "[mqtt] connect    client_id=%s username=%s remote=%s:%s\n",
		info.ClientID, info.Username, info.RemoteHost, info.RemotePort)
}

// OnDisconnect 在客户端断开时打印日志，含触发原因。
func (h *meshtasticFilterHook) OnDisconnect(cl *mqtt.Client, err error, expire bool) {
	if cl != nil {
		h.clientStats.Delete(cl.ID)
	}
	if !h.consoleLog {
		return
	}
	info := mqttClientInfoFromClient(cl)
	reason := "client closed"
	if err != nil {
		reason = err.Error()
	}
	fmt.Fprintf(os.Stderr, "[mqtt] disconnect client_id=%s username=%s remote=%s:%s expire=%t reason=%s\n",
		info.ClientID, info.Username, info.RemoteHost, info.RemotePort, expire, reason)
}

// OnPacketRead 在 broker 收到客户端报文时累计入站计数（客户端 → 服务器）。
// 返回原始 packet 不做修改；该 hook 在 packet 校验前触发。
func (h *meshtasticFilterHook) OnPacketRead(cl *mqtt.Client, pk packets.Packet) (packets.Packet, error) {
	if cl != nil {
		h.clientStats.IncIn(cl.ID)
	}
	return pk, nil
}

// OnPacketSent 在 broker 把报文写出后累计出站计数（服务器 → 客户端）。
func (h *meshtasticFilterHook) OnPacketSent(cl *mqtt.Client, pk packets.Packet, b []byte) {
	if cl != nil {
		h.clientStats.IncOut(cl.ID)
	}
}

// OnSubscribed 客户端订阅成功后打印订阅的 topic filter 列表。
func (h *meshtasticFilterHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte) {
	if !h.consoleLog {
		return
	}
	info := mqttClientInfoFromClient(cl)
	for _, sub := range pk.Filters {
		fmt.Fprintf(os.Stderr, "[mqtt] subscribe  client_id=%s username=%s remote=%s:%s topic=%s\n",
			info.ClientID, info.Username, info.RemoteHost, info.RemotePort, sub.Filter)
	}
}

// OnUnsubscribed 客户端取消订阅后打印日志。
func (h *meshtasticFilterHook) OnUnsubscribed(cl *mqtt.Client, pk packets.Packet) {
	if !h.consoleLog {
		return
	}
	info := mqttClientInfoFromClient(cl)
	for _, sub := range pk.Filters {
		fmt.Fprintf(os.Stderr, "[mqtt] unsubscribe client_id=%s username=%s remote=%s:%s topic=%s\n",
			info.ClientID, info.Username, info.RemoteHost, info.RemotePort, sub.Filter)
	}
}

// OnPublish 在 broker 转发消息前校验 payload；无效消息会被拒绝并丢弃。
func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (packets.Packet, error) {
	valid, _, record := mqtpp.MQTTPP(pk.TopicName, pk.Payload, h.key, mqtpp.Options{
		AllowEncryptedForwarding: h.settings.AllowEncryptedForwarding(),
		PKIKeyResolver:           h.pkiResolver,
	})
	if !valid {
		// 记录拒绝原因，帮助诊断 QoS0 重发问题
		if h.consoleLog {
			info := mqttClientInfoFromClient(cl)
			fmt.Fprintf(os.Stderr, "[mqtt] PUBLISH rejected: client_id=%s topic=%s qos=%d payload_len=%d error=%v\n",
				info.ClientID, pk.TopicName, pk.FixedHeader.Qos, len(pk.Payload), record["error"])
		}
		h.rejectPublish(cl, pk, record)
		return pk, packets.ErrRejectPacket
	}
	if violation := blockingViolationForRecord(h.blocking, record); violation != nil {
		for key, value := range violation {
			record[key] = value
		}
		// 记录屏蔽原因
		if h.consoleLog {
			info := mqttClientInfoFromClient(cl)
			fmt.Fprintf(os.Stderr, "[mqtt] PUBLISH blocked: client_id=%s topic=%s type=%v reason=%v\n",
				info.ClientID, pk.TopicName, violation["blocking_type"], violation["error"])
		}
		h.rejectPublish(cl, pk, record)
		return pk, packets.ErrRejectPacket
	}
	if h.dedupQueue != nil && !h.dedupQueue.TryForward(pk.TopicName, pk.Payload) {
		h.stats.IncDropped()
		return pk, packets.ErrRejectPacket
	}
	h.stats.IncForwarded()

	h.dbQueue.EnqueueRecord(record, mqttClientInfoFromClient(cl))
	if h.autoAcker != nil {
		h.autoAcker(record)
	}
	if h.packetConsoleLog && record["type"] != "empty_packet" {
		printMeshtasticRecord(record)
	}
	return pk, nil
}

func (h *meshtasticFilterHook) rejectPublish(cl *mqtt.Client, pk packets.Packet, record map[string]any) {
	if h.stats != nil {
		h.stats.IncDropped()
	}
	if record == nil {
		record = map[string]any{}
	}
	record["topic"] = pk.TopicName
	h.dbQueue.EnqueueDiscard(record, pk.Payload, mqttClientInfoFromClient(cl))
}

func blockingViolationForRecord(blocking *blockingpkg.Cache, record map[string]any) map[string]any {
	if blocking == nil || record == nil {
		return nil
	}
	if blocking.IsNodeBlocked(record["from"], record["from_num"]) {
		return map[string]any{"error": "blocked node", "blocking_type": "node"}
	}
	var field string
	switch record["type"] {
	case "text_message":
		field = "text"
	case "nodeinfo", "map_report":
		field = "long_name"
	default:
		return nil
	}
	if word, ok := blocking.FindForbiddenWord(record[field]); ok {
		return map[string]any{"error": "forbidden word", "blocking_type": "forbidden_word", "blocking_field": field, "matched_word": word}
	}
	return nil
}

func mqttClientInfoFromClient(cl *mqtt.Client) storepkg.MQTTClientInfo {
	if cl == nil {
		return storepkg.MQTTClientInfo{}
	}

	info := storepkg.MQTTClientInfo{
		ClientID:   cl.ID,
		Username:   string(cl.Properties.Username),
		Listener:   cl.Net.Listener,
		RemoteAddr: cl.Net.Remote,
	}
	if info.RemoteAddr == "" && cl.Net.Conn != nil && cl.Net.Conn.RemoteAddr() != nil {
		info.RemoteAddr = cl.Net.Conn.RemoteAddr().String()
	}
	if host, port, err := net.SplitHostPort(info.RemoteAddr); err == nil {
		info.RemoteHost = host
		info.RemotePort = port
	} else {
		info.RemoteHost = info.RemoteAddr
	}
	return info
}

// main 是程序入口，负责解析参数并启动 MQTT broker。
func main() {
	cfg, err := parseArgs()
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(2)
	}
	if err := run(cfg); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

// parseArgs 加载配置文件、解析命令行覆盖项，并展开 Meshtastic channel PSK。
func parseArgs() (*configpkg.Config, error) {
	cfg, err := configpkg.Load(configpkg.DefaultPath())
	if err != nil {
		return nil, err
	}

	flag.StringVar(&cfg.MQTT.Host, "host", cfg.MQTT.Host, "MQTT broker listen host")
	flag.IntVar(&cfg.MQTT.Port, "port", cfg.MQTT.Port, "MQTT broker listen port")
	flag.StringVar(&cfg.Meshtastic.PSK, "psk", cfg.Meshtastic.PSK, "Base64 channel PSK used to try decrypting encrypted packets")
	flag.BoolVar(&cfg.MQTT.TLS.Enabled, "tls", cfg.MQTT.TLS.Enabled, "Enable MQTT TLS listener")
	flag.StringVar(&cfg.MQTT.TLS.CertFile, "tls-cert", cfg.MQTT.TLS.CertFile, "MQTT TLS certificate file")
	flag.StringVar(&cfg.MQTT.TLS.KeyFile, "tls-key", cfg.MQTT.TLS.KeyFile, "MQTT TLS private key file")
	flag.StringVar(&cfg.Database.Driver, "db-driver", cfg.Database.Driver, "Database driver: sqlite or mysql")
	flag.StringVar(&cfg.Database.SQLite.Path, "sqlite-path", cfg.Database.SQLite.Path, "SQLite database file path")
	flag.StringVar(&cfg.Database.MySQL.DSN, "mysql-dsn", cfg.Database.MySQL.DSN, "MySQL database DSN")
	flag.BoolVar(&cfg.Web.Enabled, "web", cfg.Web.Enabled, "Enable Gin web server")
	flag.BoolVar(&cfg.Web.PortEnabled, "web-port-enabled", cfg.Web.PortEnabled, "Enable web server on TCP host and port")
	flag.BoolVar(&cfg.Web.SocketEnabled, "web-socket-enabled", cfg.Web.SocketEnabled, "Enable web server on Unix socket; unsupported on Windows")
	flag.StringVar(&cfg.Web.Host, "web-host", cfg.Web.Host, "Web server listen host")
	flag.IntVar(&cfg.Web.Port, "web-port", cfg.Web.Port, "Web server listen port")
	flag.StringVar(&cfg.Web.SocketPath, "web-socket-path", cfg.Web.SocketPath, "Web server Unix socket path; unsupported on Windows")
	flag.StringVar(&cfg.Web.StaticDir, "web-static-dir", cfg.Web.StaticDir, "Web frontend static files directory")
	flag.StringVar(&cfg.Web.MapTileCacheDir, "web-map-tile-cache-dir", cfg.Web.MapTileCacheDir, "Map tile disk cache root directory")
	flag.StringVar(&cfg.Web.Admin.Username, "admin-username", cfg.Web.Admin.Username, "Web admin username")
	flag.Parse()

	if value := os.Getenv("MESH_ADMIN_PASSWORD"); value != "" {
		cfg.Web.Admin.Password = value
	}
	if value := os.Getenv("MESH_ADMIN_SESSION_SECRET"); value != "" {
		cfg.Web.Admin.SessionSecret = value
	}
	configpkg.ClearWebSocketPathOnUnsupportedGOOS(cfg, runtime.GOOS)

	if err := configpkg.Validate(cfg); err != nil {
		return nil, err
	}
	key, err := mqtpp.ExpandPSK(cfg.Meshtastic.PSK)
	if err != nil {
		return nil, err
	}
	cfg.Key = key
	return cfg, nil
}

// run 创建 MQTT broker 和 Web 服务，并阻塞等待退出信号。
func run(cfg *configpkg.Config) error {
	store, err := storepkg.OpenStore(cfg.Database, cfg.ConsoleLog.SQL)
	if err != nil {
		return err
	}
	defer store.Close()
	dbQueue := storepkg.NewWriteQueue(store)
	defer dbQueue.Close()
	if err := store.EnsureDefaultAdmin(cfg.Web.Admin.Username, cfg.Web.Admin.Password); err != nil {
		return err
	}

	blocking, err := blockingpkg.New(store)
	if err != nil {
		return err
	}
	settings, err := rspkg.New(store)
	if err != nil {
		return err
	}

	messageStats := &mqttforwardpkg.Stats{}
	clientStats := mqttforwardpkg.NewClientStats()
	server, mqttHook, mqttAddr, err := startMQTTServer(cfg, store, dbQueue, messageStats, clientStats, blocking, settings)
	if err != nil {
		return err
	}
	botSender := botpkg.NewService(store, server, cfg.Key)
	mqttHook.autoAcker = botSender.MaybeAutoAck
	botCtx, stopBotBroadcaster := context.WithCancel(context.Background())
	defer stopBotBroadcaster()
	botSender.StartNodeInfoBroadcaster(botCtx)
	forwardManager := mqttforwardpkg.NewManager(store)
	if err := forwardManager.StartFromStore(); err != nil {
		server.Close()
		return err
	}
	defer forwardManager.StopAll()

	// Initialize AI Service
	// Create bot sender adapter - 支持频道消息和私聊消息两种发送方式
	botSenderAdapter := autoreply.NewBotServiceAdapter(
		// SendDirectText: 发送私聊消息
		func(ctx context.Context, botID uint64, toNodeNum int64, text string) error {
			_, err := botSender.SendText(ctx, botpkg.SendTextRequest{
				BotID:       botID,
				MessageType: "direct",
				ToNodeNum:   &toNodeNum,
				Text:        text,
			})
			return err
		},
		// SendChannelText: 发送频道消息
		func(ctx context.Context, botID uint64, channelID string, text string) error {
			_, err := botSender.SendText(ctx, botpkg.SendTextRequest{
				BotID:       botID,
				MessageType: "channel",
				ChannelID:   channelID,
				Text:        text,
			})
			return err
		},
	)

	aiManager := ai.NewAIManager(ai.Config{
		DataDir:          cfg.AI.DataDir,
		Enabled:          cfg.AI.Enabled,
		ConsoleLog:       cfg.ConsoleLog.LLM,
		ToolConfigStore:  store,
		ToolRouterStore:  store,
		TopicRouterStore: store,
		Store:            store,
	}, store.DB(), botSenderAdapter, botCtx, store)

	if cfg.AI.Enabled {
		aiManager.SetConfigEnabled(true)
		providers, err := store.ListLLMProviders(true)
		if err != nil {
			fmt.Fprintf(os.Stderr, "Warning: failed to load LLM providers: %v\n", err)
		} else if len(providers) > 0 {
			providerConfigs := make([]llm.ProviderConfig, 0, len(providers))
			for _, p := range providers {
				providerConfigs = append(providerConfigs, llm.ProviderConfig{
					Name:                p.Name,
					Active:              p.Active,
					APIKey:              p.APIKey,
					BaseURL:             p.BaseURL,
					Model:               p.Model,
					Timeout:             p.Timeout,
					ContextWindowTokens: p.ContextWindowTokens,
				})
			}
			aiManager.SetProviderConfigs(providerConfigs)
			if err := aiManager.Init(); err != nil {
				fmt.Fprintf(os.Stderr, "Warning: failed to initialize AI service: %v\n", err)
			} else {
				defer aiManager.Stop()
				printJSON(map[string]any{"event": "ai_service_started", "providers": len(providerConfigs)})
			}
		} else {
			fmt.Fprintf(os.Stderr, "Warning: AI service is enabled but no LLM providers configured\n")
		}
	}

	var httpServers []*http.Server
	errCh := make(chan error, 2)
	if cfg.Web.Enabled {
		sessions, err := auth.NewManager(cfg.Web.Admin)
		if err != nil {
			return err
		}
		mqttStatus := webpkg.MQTTRuntimeStatus{Server: server, Address: mqttAddr, TLS: cfg.MQTT.TLS.Enabled, Stats: messageStats, ClientStats: clientStats, DBQueue: dbQueue, DedupQueue: mqttHook.dedupQueue}
		handler := webpkg.NewRouter(cfg.Web, cfg.ConsoleLog.Web, store, sessions, mqttStatus, blocking, forwardManager, settings, botSender, aiManager)
		webAddresses := []string{}
		if cfg.Web.PortEnabled {
			httpServer := &http.Server{
				Addr:    net.JoinHostPort(cfg.Web.Host, strconv.Itoa(cfg.Web.Port)),
				Handler: handler,
			}
			httpServers = append(httpServers, httpServer)
			webAddresses = append(webAddresses, httpServer.Addr)
			go func() {
				if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
					errCh <- err
				}
			}()
		}
		if cfg.Web.SocketEnabled {
			httpServer := &http.Server{Handler: handler}
			httpServers = append(httpServers, httpServer)
			webAddresses = append(webAddresses, cfg.Web.SocketPath)
			go func() {
				if err := webpkg.ServeUnixSocket(httpServer, cfg.Web.SocketPath); err != nil && !errors.Is(err, http.ErrServerClosed) {
					errCh <- err
				}
			}()
		}
		webStarted := map[string]any{"event": "web_started", "addresses": webAddresses, "static_dir": cfg.Web.StaticDir}
		if len(webAddresses) > 0 {
			webStarted["address"] = webAddresses[0]
		}
		printJSON(webStarted)
	}

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)

	var runErr error
	select {
	case <-sigCh:
	case runErr = <-errCh:
	}

	for _, httpServer := range httpServers {
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		if err := httpServer.Shutdown(ctx); err != nil && runErr == nil {
			runErr = err
		}
		cancel()
	}
	if err := server.Close(); err != nil && runErr == nil {
		runErr = err
	}
	if mqttHook.dedupQueue != nil {
		mqttHook.dedupQueue.Stop()
	}
	return runErr
}

func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *storepkg.WriteQueue, stats *mqttforwardpkg.Stats, clientStats *mqttforwardpkg.ClientStats, blocking *blockingpkg.Cache, settings *rspkg.Cache) (*mqtt.Server, *meshtasticFilterHook, string, error) {
	server := mqtt.New(&mqtt.Options{InlineClient: true})
	if err := server.AddHook(new(mqttauth.AllowHook), nil); err != nil {
		return nil, nil, "", err
	}
	dedupQueue := mqttforwardpkg.NewDedupQueue()
	dedupQueue.Start()
	hook := &meshtasticFilterHook{
		server:           server,
		key:              cfg.Key,
		dbQueue:          dbQueue,
		stats:            stats,
		clientStats:      clientStats,
		blocking:         blocking,
		settings:         settings,
		pkiResolver:      botpkg.NewPKIKeyResolver(store),
		consoleLog:       cfg.ConsoleLog.MQTT,
		packetConsoleLog: cfg.ConsoleLog.Meshtastic,
		dedupQueue:       dedupQueue,
	}
	if err := server.AddHook(hook, nil); err != nil {
		return nil, nil, "", err
	}

	addr := net.JoinHostPort(cfg.MQTT.Host, strconv.Itoa(cfg.MQTT.Port))
	tlsConfig, err := configpkg.BuildTLS(cfg.MQTT.TLS)
	if err != nil {
		return nil, nil, "", err
	}
	listener := listeners.NewTCP(listeners.Config{ID: "tcp", Address: addr, TLSConfig: tlsConfig})
	if err := server.AddListener(listener); err != nil {
		return nil, nil, "", err
	}
	if err := server.Serve(); err != nil {
		return nil, nil, "", err
	}
	printJSON(map[string]any{"event": "broker_started", "address": addr, "tls": cfg.MQTT.TLS.Enabled})
	return server, hook, addr, nil
}

// printJSON 将记录编码为 JSON 后按数据包类型着色输出。
func printJSON(record map[string]any) {
	printJSONBytes(record, mqtpp.MustJSON(record))
}

// printMeshtasticRecord 把 Meshtastic 解码后的 record 按 type 拼成可读的彩色单行，
// 不输出原始 JSON。保留与 printJSONBytes 一致的色码方案。
func printMeshtasticRecord(record map[string]any) {
	if record == nil {
		return
	}
	typ, _ := record["type"].(string)
	from := stringField(record, "from")
	channel := stringField(record, "channel_id")
	gateway := stringField(record, "gateway_id")

	var color string
	var body string
	switch typ {
	case "nodeinfo":
		color = ansiGreenBGWhiteText
		body = fmt.Sprintf("nodeinfo   from=%s long=%q short=%q hw=%s role=%s",
			from, stringField(record, "long_name"), stringField(record, "short_name"),
			stringField(record, "hw_model"), stringField(record, "role"))
	case "map_report":
		color = ansiBlueBGWhiteText
		body = fmt.Sprintf("map_report from=%s long=%q lat=%v lon=%v alt=%v fw=%s region=%s",
			from, stringField(record, "long_name"),
			record["latitude"], record["longitude"], record["altitude"],
			stringField(record, "firmware_version"), stringField(record, "region"))
	case "text_message":
		color = ansiPurpleBGWhiteText
		body = fmt.Sprintf("text       from=%s channel=%s text=%q",
			from, channel, stringField(record, "text"))
	case "position":
		color = ansiCyanBGBlackText
		body = fmt.Sprintf("position   from=%s lat=%v lon=%v alt=%v",
			from, record["latitude"], record["longitude"], record["altitude"])
	case "telemetry":
		color = ansiYellowBGBlackText
		body = fmt.Sprintf("telemetry  from=%s tt=%v metrics=%v",
			from, record["telemetry_type"], record["metrics"])
	case "routing":
		color = ansiGrayBGWhiteText
		body = fmt.Sprintf("routing    from=%s pkt_id=%v", from, record["packet_id"])
	case "traceroute":
		color = ansiGrayBGWhiteText
		body = fmt.Sprintf("traceroute from=%s pkt_id=%v", from, record["packet_id"])
	default:
		if record["error"] != nil {
			color = ansiRedBGWhiteText
			body = fmt.Sprintf("%-10s from=%s error=%v topic=%s", typ, from,
				record["error"], stringField(record, "topic"))
		} else {
			body = fmt.Sprintf("%-10s from=%s", typ, from)
		}
	}
	if gateway != "" {
		body += " gw=" + gateway
	}
	if color != "" {
		fmt.Printf("%s%s%s\n", color, body, ansiReset)
		return
	}
	fmt.Println(body)
}

func stringField(record map[string]any, key string) string {
	if v, ok := record[key].(string); ok {
		return v
	}
	return ""
}

// printJSONBytes 使用已编码好的 JSON 文本，并根据记录 type 选择控制台颜色。
func printJSONBytes(record map[string]any, text []byte) {
	switch record["type"] {
	case "nodeinfo":
		fmt.Printf("%s%s%s\n", ansiGreenBGWhiteText, text, ansiReset)
	case "map_report":
		fmt.Printf("%s%s%s\n", ansiBlueBGWhiteText, text, ansiReset)
	case "text_message":
		fmt.Printf("%s%s%s\n", ansiPurpleBGWhiteText, text, ansiReset)
	case "position":
		fmt.Printf("%s%s%s\n", ansiCyanBGBlackText, text, ansiReset)
	case "telemetry":
		fmt.Printf("%s%s%s\n", ansiYellowBGBlackText, text, ansiReset)
	case "routing", "traceroute":
		fmt.Printf("%s%s%s\n", ansiGrayBGWhiteText, text, ansiReset)
	default:
		if record["error"] != nil {
			fmt.Printf("%s%s%s\n", ansiRedBGWhiteText, text, ansiReset)
			return
		}
		fmt.Println(string(text))
	}
}