重构：拆出 bot / sign / mapsource / llmadmin / web 包

第三批（最终批）：把 root 中剩下所有领域文件按功能搬到 internal/ 下。完成后根目录从 46 个 .go 文件降到 14 个，其中 11 个是 bridge（type alias + thin wrapper），仅供尚未改造的 main.go 引用。新增包 - internal/bot/ Service / TextSender / NewPKIKeyResolver / RegisterRoutes，含 PKI 直连发送、节点信息广播、 outbound DM 持久化等业务逻辑。 - internal/sign/ SignDTO / SignDayCountDTO / RegisterAdminRoutes，把原来分散在 admin_sign_routes.go 与 web.go 中的 sign DTO/路由收拢到一处。 - internal/mapsource/ AdminDTO / PublicDTO / RegisterAdminRoutes / RegisterPublicRoutes。 - internal/llmadmin/ LLM 消息队列、Provider、ToolRouter、PrimaryConfig 的 admin 路由。 - internal/web/ 路由总入口（NewRouter/NewHTTPServer/ServeUnixSocket）、各资源的 GET API、admin 用户/登录/MQTT 状态、所有 DTO 函数。把 auth.go 的 sessionClaims 升级为 auth.SessionClaims；mqtt_status.go 重写成 MQTTRuntimeStatus / AdminMQTTStatus 结构体并把 client info 解析在 web 包内自带，不再依赖 main 包。 map_tile_proxy_routes 与测试一起搬入。修改 - web.go 中 parseListOptions / writeListResponse / ptrString 等本地 helper 改为对 internal/webutil 的 thin wrapper，避免重复实现。 - internal/auth 在 step 4 已创建，本批中 web 包正式开始引用其 Manager / RequireAdmin / SessionClaims。根目录新增 bridge：bot_bridge.go / sign_bridge.go / mapsource_bridge.go / llmadmin_bridge.go / web_bridge.go。后者把 mqttRuntimeStatus 包成 webpkg.MQTTStatusProvider 适配器，使 main.go 中旧字段名保持可用。 go build ./... / go test ./... 全部通过；测试数量未变。 Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-18 18:24:14 +08:00
parent c527a9fd9a
commit 9394aa0f4a
18 changed files with 565 additions and 530 deletions
@@ -0,0 +1,353 @@
+package bot
+
+import (
+	"errors"
+	"net/http"
+	"strconv"
+
+	"github.com/gin-gonic/gin"
+	"gorm.io/gorm"
+
+	"meshtastic_mqtt_server/internal/auth"
+	storepkg "meshtastic_mqtt_server/internal/store"
+	"meshtastic_mqtt_server/internal/webutil"
+)
+
+type botNodeRequest struct {
+	NodeNum                          *int64 `json:"node_num"`
+	LongName                         string `json:"long_name"`
+	ShortName                        string `json:"short_name"`
+	Enabled                          bool   `json:"enabled"`
+	DefaultChannelID                 string `json:"default_channel_id"`
+	TopicPrefix                      string `json:"topic_prefix"`
+	PSK                              string `json:"psk"`
+	NodeInfoBroadcastEnabled         bool   `json:"nodeinfo_broadcast_enabled"`
+	NodeInfoBroadcastIntervalSeconds int64  `json:"nodeinfo_broadcast_interval_seconds"`
+	LLMQueueEnabled                  bool   `json:"llm_queue_enabled"`
+	LLMIncludeChannelMessages        bool   `json:"llm_include_channel_messages"`
+}
+
+type botSendMessageRequest struct {
+	BotID       uint64 `json:"bot_id"`
+	MessageType string `json:"message_type"`
+	ChannelID   string `json:"channel_id"`
+	ToNodeID    string `json:"to_node_id"`
+	ToNodeNum   *int64 `json:"to_node_num"`
+	Text        string `json:"text"`
+}
+
+func RegisterRoutes(r gin.IRouter, store *storepkg.Store, sender TextSender) {
+	r.GET("/bot/nodes", func(c *gin.Context) {
+		opts, ok := webutil.ParseListOptions(c)
+		if !ok {
+			return
+		}
+		rows, err := store.ListBotNodes(opts)
+		if err != nil {
+			webutil.WriteListResponse(c, rows, opts, err, botNodeDTO)
+			return
+		}
+		total, err := store.CountBotNodes(opts)
+		webutil.WriteListResponseWithTotal(c, rows, opts, total, err, botNodeDTO)
+	})
+	r.POST("/bot/nodes", func(c *gin.Context) {
+		var req botNodeRequest
+		if err := c.ShouldBindJSON(&req); err != nil {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid bot node request"})
+			return
+		}
+		row, err := store.CreateBotNode(botNodeInputFromRequest(req))
+		writeBotNodeMutationResponse(c, http.StatusCreated, row, err)
+	})
+	r.PUT("/bot/nodes/:id", func(c *gin.Context) {
+		id, ok := parseBotID(c, "invalid bot node id")
+		if !ok {
+			return
+		}
+		var req botNodeRequest
+		if err := c.ShouldBindJSON(&req); err != nil {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid bot node request"})
+			return
+		}
+		row, err := store.UpdateBotNode(id, botNodeInputFromRequest(req))
+		writeBotNodeMutationResponse(c, http.StatusOK, row, err)
+	})
+	r.DELETE("/bot/nodes/:id", func(c *gin.Context) {
+		id, ok := parseBotID(c, "invalid bot node id")
+		if !ok {
+			return
+		}
+		if err := store.DeleteBotNode(id); errors.Is(err, gorm.ErrRecordNotFound) {
+			c.JSON(http.StatusNotFound, gin.H{"error": "bot node not found"})
+			return
+		} else if err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+			return
+		}
+		c.JSON(http.StatusOK, gin.H{"status": "ok"})
+	})
+	r.POST("/bot/nodes/:id/keys/regenerate", func(c *gin.Context) {
+		id, ok := parseBotID(c, "invalid bot node id")
+		if !ok {
+			return
+		}
+		row, err := store.RegenerateBotNodeKeys(id)
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			c.JSON(http.StatusNotFound, gin.H{"error": "bot node not found"})
+			return
+		}
+		if err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+			return
+		}
+		c.JSON(http.StatusOK, gin.H{"item": botNodeDTO(*row)})
+	})
+	r.POST("/bot/nodes/:id/nodeinfo", func(c *gin.Context) {
+		if sender == nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": "bot sender is not configured"})
+			return
+		}
+		id, ok := parseBotID(c, "invalid bot node id")
+		if !ok {
+			return
+		}
+		row, err := sender.PublishNodeInfoByID(c.Request.Context(), id)
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			c.JSON(http.StatusNotFound, gin.H{"error": "bot node not found"})
+			return
+		}
+		if err != nil {
+			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+			return
+		}
+		c.JSON(http.StatusOK, gin.H{"item": botNodeDTO(*row)})
+	})
+	r.GET("/bot/messages", func(c *gin.Context) {
+		opts, ok := parseBotMessageListOptions(c)
+		if !ok {
+			return
+		}
+		rows, err := store.ListBotMessages(opts)
+		if err != nil {
+			webutil.WriteListResponse(c, rows, opts.ListOptions, err, botMessageDTO)
+			return
+		}
+		total, err := store.CountBotMessages(opts)
+		webutil.WriteListResponseWithTotal(c, rows, opts.ListOptions, total, err, botMessageDTO)
+	})
+	r.GET("/bot/direct-messages", func(c *gin.Context) {
+		opts, ok := webutil.ParseListOptions(c)
+		if !ok {
+			return
+		}
+		botID, err := strconv.ParseUint(c.Query("bot_id"), 10, 64)
+		if err != nil || botID == 0 {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid bot id"})
+			return
+		}
+		if _, err := store.GetBotNode(botID); errors.Is(err, gorm.ErrRecordNotFound) {
+			c.JSON(http.StatusNotFound, gin.H{"error": "bot node not found"})
+			return
+		} else if err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+			return
+		}
+		target, err := strconv.ParseInt(c.Query("target_node_num"), 10, 64)
+		if err != nil || target <= 0 {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid target node num"})
+			return
+		}
+		dmOpts := storepkg.BotDirectMessageListOptions{ListOptions: opts, BotID: botID, PeerNodeNum: target, Direction: c.Query("direction")}
+		rows, err := store.ListBotDirectMessagesByConversation(dmOpts)
+		if err != nil {
+			webutil.WriteListResponse(c, rows, opts, err, botDirectMessageDTO)
+			return
+		}
+		total, err := store.CountBotDirectMessagesByConversation(dmOpts)
+		webutil.WriteListResponseWithTotal(c, rows, opts, total, err, botDirectMessageDTO)
+	})
+	// /bot/conversations 返回某个 bot 下所有会话的概要（最后一条消息 + 未读数），
+	// 给前端侧边栏渲染会话列表使用。
+	r.GET("/bot/conversations", func(c *gin.Context) {
+		opts, ok := webutil.ParseListOptions(c)
+		if !ok {
+			return
+		}
+		botID, err := strconv.ParseUint(c.Query("bot_id"), 10, 64)
+		if err != nil || botID == 0 {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid bot id"})
+			return
+		}
+		if _, err := store.GetBotNode(botID); errors.Is(err, gorm.ErrRecordNotFound) {
+			c.JSON(http.StatusNotFound, gin.H{"error": "bot node not found"})
+			return
+		} else if err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+			return
+		}
+		rows, err := store.ListBotDirectConversations(botID, opts)
+		if err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+			return
+		}
+		unread, err := store.CountBotDirectUnread(botID)
+		if err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+			return
+		}
+		items := make([]gin.H, 0, len(rows))
+		for _, row := range rows {
+			items = append(items, botDirectConversationDTO(row))
+		}
+		c.JSON(http.StatusOK, gin.H{"items": items, "limit": opts.Limit, "offset": opts.Offset, "unread_total": unread})
+	})
+	// /bot/direct-messages/read 把指定 (bot, peer) 下所有未读 inbound 消息标记为已读。
+	r.POST("/bot/direct-messages/read", func(c *gin.Context) {
+		var req struct {
+			BotID       uint64 `json:"bot_id"`
+			PeerNodeNum int64  `json:"peer_node_num"`
+		}
+		if err := c.ShouldBindJSON(&req); err != nil {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid mark-read request"})
+			return
+		}
+		if req.BotID == 0 || req.PeerNodeNum == 0 {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "bot_id and peer_node_num are required"})
+			return
+		}
+		if _, err := store.GetBotNode(req.BotID); errors.Is(err, gorm.ErrRecordNotFound) {
+			c.JSON(http.StatusNotFound, gin.H{"error": "bot node not found"})
+			return
+		} else if err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+			return
+		}
+		updated, err := store.MarkBotDirectMessagesRead(req.BotID, req.PeerNodeNum)
+		if err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+			return
+		}
+		c.JSON(http.StatusOK, gin.H{"updated": updated})
+	})
+	r.POST("/bot/messages", func(c *gin.Context) {
+		if sender == nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": "bot sender is not configured"})
+			return
+		}
+		var req botSendMessageRequest
+		if err := c.ShouldBindJSON(&req); err != nil {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid bot message request"})
+			return
+		}
+		claims := c.MustGet("admin_claims").(*auth.SessionClaims)
+		row, err := sender.SendText(c.Request.Context(), SendTextRequest{BotID: req.BotID, MessageType: req.MessageType, ChannelID: req.ChannelID, ToNodeID: req.ToNodeID, ToNodeNum: req.ToNodeNum, Text: req.Text, CreatedBy: claims.Username})
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			c.JSON(http.StatusNotFound, gin.H{"error": "bot node not found"})
+			return
+		}
+		if err != nil {
+			status := http.StatusBadRequest
+			if row != nil && row.ID != 0 {
+				c.JSON(http.StatusAccepted, gin.H{"item": botMessageDTO(*row), "error": err.Error()})
+				return
+			}
+			c.JSON(status, gin.H{"error": err.Error()})
+			return
+		}
+		c.JSON(http.StatusCreated, gin.H{"item": botMessageDTO(*row)})
+	})
+}
+
+func botNodeInputFromRequest(req botNodeRequest) storepkg.BotNodeInput {
+	return storepkg.BotNodeInput{NodeNum: req.NodeNum, LongName: req.LongName, ShortName: req.ShortName, Enabled: req.Enabled, DefaultChannelID: req.DefaultChannelID, TopicPrefix: req.TopicPrefix, PSK: req.PSK, NodeInfoBroadcastEnabled: req.NodeInfoBroadcastEnabled, NodeInfoBroadcastIntervalSeconds: req.NodeInfoBroadcastIntervalSeconds, LLMQueueEnabled: req.LLMQueueEnabled, LLMIncludeChannelMessages: req.LLMIncludeChannelMessages}
+}
+
+func parseBotID(c *gin.Context, message string) (uint64, bool) {
+	id, err := strconv.ParseUint(c.Param("id"), 10, 64)
+	if err != nil || id == 0 {
+		c.JSON(http.StatusBadRequest, gin.H{"error": message})
+		return 0, false
+	}
+	return id, true
+}
+
+func parseBotMessageListOptions(c *gin.Context) (storepkg.BotMessageListOptions, bool) {
+	listOpts, ok := webutil.ParseListOptions(c)
+	if !ok {
+		return storepkg.BotMessageListOptions{}, false
+	}
+	opts := storepkg.BotMessageListOptions{ListOptions: listOpts, MessageType: c.Query("message_type"), ChannelID: c.Query("channel_id")}
+	if value := c.Query("bot_id"); value != "" {
+		id, err := strconv.ParseUint(value, 10, 64)
+		if err != nil {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "invalid bot id"})
+			return storepkg.BotMessageListOptions{}, false
+		}
+		opts.BotID = id
+	}
+	return opts, true
+}
+
+func writeBotNodeMutationResponse(c *gin.Context, status int, row *storepkg.BotNodeRecord, err error) {
+	if errors.Is(err, storepkg.ErrBotNodeAlreadyExists) {
+		c.JSON(http.StatusConflict, gin.H{"error": "bot node already exists or conflicts with existing node"})
+		return
+	}
+	if errors.Is(err, gorm.ErrRecordNotFound) {
+		c.JSON(http.StatusNotFound, gin.H{"error": "bot node not found"})
+		return
+	}
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(status, gin.H{"item": botNodeDTO(*row)})
+}
+
+func botNodeDTO(row storepkg.BotNodeRecord) gin.H {
+	return gin.H{"id": row.ID, "node_id": row.NodeID, "node_num": row.NodeNum, "long_name": row.LongName, "short_name": row.ShortName, "enabled": row.Enabled, "default_channel_id": row.DefaultChannelID, "topic_prefix": row.TopicPrefix, "psk": row.PSK, "public_key": row.PublicKey, "private_key_set": row.PrivateKey != "", "nodeinfo_broadcast_enabled": row.NodeInfoBroadcastEnabled, "nodeinfo_broadcast_interval_seconds": row.NodeInfoBroadcastIntervalSeconds, "last_nodeinfo_broadcast_at": row.LastNodeInfoBroadcastAt, "llm_queue_enabled": row.LLMQueueEnabled, "llm_include_channel_messages": row.LLMIncludeChannelMessages, "created_at": row.CreatedAt, "updated_at": row.UpdatedAt}
+}
+
+func botMessageDTO(row storepkg.BotMessageRecord) gin.H {
+	return gin.H{"id": row.ID, "bot_id": row.BotID, "bot_node_id": row.BotNodeID, "bot_node_num": row.BotNodeNum, "message_type": row.MessageType, "channel_id": row.ChannelID, "to_node_id": row.ToNodeID, "to_node_num": row.ToNodeNum, "topic": row.Topic, "packet_id": row.PacketID, "text": row.Text, "payload_len": row.PayloadLen, "encrypted": row.Encrypted, "status": row.Status, "error": row.Error, "published_at": row.PublishedAt, "created_by": row.CreatedBy, "created_at": row.CreatedAt}
+}
+
+func botDirectMessageDTO(row storepkg.BotDirectMessageRecord) gin.H {
+	return gin.H{
+		"id":             row.ID,
+		"bot_id":         row.BotID,
+		"bot_node_id":    row.BotNodeID,
+		"bot_node_num":   row.BotNodeNum,
+		"peer_node_id":   row.PeerNodeID,
+		"peer_node_num":  row.PeerNodeNum,
+		"direction":      row.Direction,
+		"topic":          row.Topic,
+		"packet_id":      row.PacketID,
+		"text":           row.Text,
+		"payload_len":    row.PayloadLen,
+		"pki_encrypted":  row.PKIEncrypted,
+		"want_ack":       row.WantAck,
+		"gateway_id":     row.GatewayID,
+		"status":         row.Status,
+		"error":          row.Error,
+		"bot_message_id": row.BotMessageID,
+		"created_by":     row.CreatedBy,
+		"published_at":   row.PublishedAt,
+		"received_at":    row.ReceivedAt,
+		"read_at":        row.ReadAt,
+		"created_at":     row.CreatedAt,
+	}
+}
+
+func botDirectConversationDTO(row storepkg.BotDirectConversation) gin.H {
+	return gin.H{
+		"bot_id":          row.BotID,
+		"peer_node_id":    row.PeerNodeID,
+		"peer_node_num":   row.PeerNodeNum,
+		"last_message_at": row.LastMessageAt,
+		"last_text":       row.LastText,
+		"last_direction":  row.LastDirection,
+		"unread_count":    row.UnreadCount,
+		"total_count":     row.TotalCount,
+	}
+}
@@ -0,0 +1,36 @@
+package bot
+
+import (
+	"encoding/base64"
+	"strings"
+
+	storepkg "meshtastic_mqtt_server/internal/store"
+)
+
+// NewPKIKeyResolver 返回 mqtpp 在解密 PKI 加密包时使用的回调：根据接收者
+// 节点号查找受管 bot 的私钥，并根据发送者节点号在 nodeinfo 表中查找其公钥。
+// 返回 ok=false 时调用方会跳过 PKI 路径并回落到 channel PSK 解密。
+func NewPKIKeyResolver(s *storepkg.Store) func(toNodeNum, fromNodeNum uint32) ([]byte, []byte, bool) {
+	if s == nil {
+		return nil
+	}
+	return func(toNodeNum, fromNodeNum uint32) ([]byte, []byte, bool) {
+		bot, err := s.GetBotNodeByNodeNum(int64(toNodeNum))
+		if err != nil {
+			return nil, nil, false
+		}
+		privateKeyB64 := strings.TrimSpace(bot.PrivateKey)
+		if privateKeyB64 == "" {
+			return nil, nil, false
+		}
+		privateKey, err := base64.StdEncoding.DecodeString(privateKeyB64)
+		if err != nil || len(privateKey) != 32 {
+			return nil, nil, false
+		}
+		fromPublic, ok := s.LookupNodeInfoPublicKey(fromNodeNum)
+		if !ok {
+			return nil, nil, false
+		}
+		return privateKey, fromPublic, true
+	}
+}
@@ -0,0 +1,681 @@
+package bot
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/base64"
+	"encoding/binary"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+	"unicode/utf8"
+
+	mqtt "github.com/mochi-mqtt/server/v2"
+	"gorm.io/gorm"
+
+	storepkg "meshtastic_mqtt_server/internal/store"
+	"meshtastic_mqtt_server/mqtpp"
+)
+
+const botMaxTextBytes = 200
+
+type SendTextRequest struct {
+	BotID       uint64
+	MessageType string
+	ChannelID   string
+	ToNodeID    string
+	ToNodeNum   *int64
+	Text        string
+	CreatedBy   string
+}
+
+type TextSender interface {
+	SendText(ctx context.Context, req SendTextRequest) (*storepkg.BotMessageRecord, error)
+	PublishNodeInfoByID(ctx context.Context, id uint64) (*storepkg.BotNodeRecord, error)
+}
+
+type Service struct {
+	store  *storepkg.Store
+	server *mqtt.Server
+	key    []byte
+}
+
+func NewService(store *storepkg.Store, server *mqtt.Server, key []byte) *Service {
+	return &Service{store: store, server: server, key: key}
+}
+
+func (s *Service) StartNodeInfoBroadcaster(ctx context.Context) {
+	if s == nil || s.store == nil || s.server == nil {
+		return
+	}
+	go s.runNodeInfoBroadcaster(ctx)
+}
+
+// MaybeAutoAck 在收到一个发往本地 bot 的 want_ack 包时，回送一个 Routing-NONE ACK。
+//
+// 与固件 ReliableRouter::sniffReceived 行为对齐：
+//   - PKI 加密包（pki_encrypted=true）用 X25519+AES-CCM 加密 ACK，channel_id="PKI"
+//   - 其它情况用原 channel + bot PSK 加密
+//
+// 解析失败、目标不是受管 bot、或缺少必要的密钥时，安静返回不报错——这条路径只是“尽力”。
+func (s *Service) MaybeAutoAck(record map[string]any) {
+	if s == nil || s.store == nil || s.server == nil || record == nil {
+		return
+	}
+	wantAck, _ := record["want_ack"].(bool)
+	if !wantAck {
+		return
+	}
+	toNum, ok := uint32FromRecord(record["packet_to_num"])
+	if !ok || toNum == 0 || toNum == mqtpp.NodeNumBroadcast {
+		return
+	}
+	fromNum, ok := uint32FromRecord(record["packet_from_num"])
+	if !ok || fromNum == 0 {
+		return
+	}
+	requestID, ok := uint32FromRecord(record["packet_id"])
+	if !ok || requestID == 0 {
+		return
+	}
+	bot, err := s.store.GetBotNodeByNodeNum(int64(toNum))
+	if err != nil || bot == nil || !bot.Enabled {
+		return
+	}
+	pkiEncrypted, _ := record["pki_encrypted"].(bool)
+	channelID, _ := record["channel_id"].(string)
+
+	ackPacketID, err := randomPacketID()
+	if err != nil {
+		return
+	}
+	topic := botMQTTTopic(bot.TopicPrefix, fallbackChannelID(channelID, pkiEncrypted, bot.DefaultChannelID), bot.NodeID)
+
+	var raw []byte
+	if pkiEncrypted {
+		raw, err = s.buildPKIAck(bot, fromNum, ackPacketID, requestID)
+		if err == nil {
+			topic = botMQTTTopic(bot.TopicPrefix, mqtpp.PKIChannelID, bot.NodeID)
+		}
+	} else {
+		raw, err = s.buildPSKAck(bot, fromNum, ackPacketID, requestID, channelID)
+	}
+	if err != nil || raw == nil {
+		printJSON(map[string]any{"event": "bot_auto_ack_skipped", "bot_node_id": bot.NodeID, "to": fromNum, "request_id": requestID, "error": errString(err)})
+		return
+	}
+	if err := s.server.Publish(topic, raw, false, 0); err != nil {
+		printJSON(map[string]any{"event": "bot_auto_ack_publish_failed", "bot_node_id": bot.NodeID, "topic": topic, "error": err.Error()})
+	}
+}
+
+func (s *Service) buildPKIAck(bot *storepkg.BotNodeRecord, toNum, ackPacketID, requestID uint32) ([]byte, error) {
+	privateKeyB64 := strings.TrimSpace(bot.PrivateKey)
+	if privateKeyB64 == "" {
+		return nil, fmt.Errorf("bot has no private key")
+	}
+	privateKey, err := base64.StdEncoding.DecodeString(privateKeyB64)
+	if err != nil {
+		return nil, err
+	}
+	senderPublic, err := storepkg.DecodeBotPublicKey(*bot)
+	if err != nil {
+		return nil, err
+	}
+	recipientPublic, ok := s.store.LookupNodeInfoPublicKey(toNum)
+	if !ok {
+		return nil, fmt.Errorf("recipient %s has no public key on file", mqtpp.NodeNumToID(toNum))
+	}
+	return mqtpp.BuildPKIAckServiceEnvelope(mqtpp.PKIAckBuildOptions{
+		FromNodeNum:   uint32(bot.NodeNum),
+		ToNodeNum:     toNum,
+		PacketID:      ackPacketID,
+		RequestID:     requestID,
+		GatewayID:     bot.NodeID,
+		ViaMQTT:       true,
+		SenderPrivate: privateKey,
+		RecipientPub:  recipientPublic,
+		SenderPublic:  senderPublic,
+	})
+}
+
+func (s *Service) buildPSKAck(bot *storepkg.BotNodeRecord, toNum, ackPacketID, requestID uint32, channelID string) ([]byte, error) {
+	channel := fallbackChannelID(channelID, false, bot.DefaultChannelID)
+	if channel == "" || channel == mqtpp.PKIChannelID {
+		return nil, fmt.Errorf("no channel id available for psk ack")
+	}
+	psk := strings.TrimSpace(bot.PSK)
+	if psk == "" {
+		psk = storepkg.BotDefaultPSK
+	}
+	key, err := mqtpp.ExpandPSK(psk)
+	if err != nil {
+		return nil, err
+	}
+	return mqtpp.BuildAckServiceEnvelope(mqtpp.AckBuildOptions{
+		PacketBuildOptions: mqtpp.PacketBuildOptions{
+			FromNodeNum: uint32(bot.NodeNum),
+			ToNodeNum:   toNum,
+			PacketID:    ackPacketID,
+			ChannelID:   channel,
+			GatewayID:   bot.NodeID,
+			PSK:         key,
+			Encrypt:     true,
+			ViaMQTT:     true,
+		},
+		RequestID: requestID,
+	})
+}
+
+func fallbackChannelID(channelID string, pkiEncrypted bool, defaultChannelID string) string {
+	channelID = strings.TrimSpace(channelID)
+	if pkiEncrypted {
+		return mqtpp.PKIChannelID
+	}
+	if channelID != "" && channelID != mqtpp.PKIChannelID {
+		return channelID
+	}
+	return defaultChannelID
+}
+
+func uint32FromRecord(value any) (uint32, bool) {
+	switch v := value.(type) {
+	case uint32:
+		return v, true
+	case int:
+		if v >= 0 {
+			return uint32(v), true
+		}
+	case int64:
+		if v >= 0 {
+			return uint32(v), true
+		}
+	case uint64:
+		return uint32(v), true
+	case float64:
+		if v >= 0 {
+			return uint32(v), true
+		}
+	}
+	return 0, false
+}
+
+func errString(err error) string {
+	if err == nil {
+		return ""
+	}
+	return err.Error()
+}
+
+func (s *Service) SendText(_ context.Context, req SendTextRequest) (*storepkg.BotMessageRecord, error) {
+	if s == nil || s.store == nil {
+		return nil, fmt.Errorf("bot service is not configured")
+	}
+	bot, err := s.store.GetBotNode(req.BotID)
+	if err != nil {
+		return nil, err
+	}
+	if !bot.Enabled {
+		return nil, fmt.Errorf("bot node is disabled")
+	}
+	messageType, err := normalizeBotMessageType(req.MessageType)
+	if err != nil {
+		return nil, err
+	}
+	text := strings.TrimSpace(req.Text)
+	if text == "" {
+		return nil, fmt.Errorf("text is required")
+	}
+	if !utf8.ValidString(text) {
+		return nil, fmt.Errorf("text must be valid utf-8")
+	}
+	if len([]byte(text)) > botMaxTextBytes {
+		return nil, fmt.Errorf("text is too long, max %d bytes", botMaxTextBytes)
+	}
+	toNodeNum, toNodeID, err := botMessageTarget(messageType, req)
+	if err != nil {
+		return nil, err
+	}
+	packetID, err := randomPacketID()
+	if err != nil {
+		return nil, err
+	}
+	fromNodeNum := uint32(bot.NodeNum)
+
+	// direct 私聊走 PKI；channel 群聊保留旧的 AES-CTR + PSK 路径
+	if messageType == storepkg.BotMessageTypeDirect {
+		return s.sendPKIDirect(bot, fromNodeNum, uint32(toNodeNum), toNodeID, packetID, text, req.CreatedBy)
+	}
+
+	channelID := strings.TrimSpace(req.ChannelID)
+	if channelID == "" {
+		channelID = bot.DefaultChannelID
+	}
+	if channelID == "" {
+		return nil, fmt.Errorf("channel id is required")
+	}
+	psk := strings.TrimSpace(bot.PSK)
+	if psk == "" {
+		psk = storepkg.BotDefaultPSK
+	}
+	key, err := mqtpp.ExpandPSK(psk)
+	if err != nil {
+		return nil, err
+	}
+	raw, err := mqtpp.BuildTextMessageServiceEnvelope(mqtpp.TextMessageBuildOptions{
+		PacketBuildOptions: mqtpp.PacketBuildOptions{
+			FromNodeNum: fromNodeNum,
+			ToNodeNum:   uint32(toNodeNum),
+			PacketID:    packetID,
+			ChannelID:   channelID,
+			GatewayID:   bot.NodeID,
+			PSK:         key,
+			Encrypt:     true,
+			ViaMQTT:     true,
+		},
+		Text: text,
+	})
+	if err != nil {
+		return nil, err
+	}
+	topic := botMQTTTopic(bot.TopicPrefix, channelID, bot.NodeID)
+	row := &storepkg.BotMessageRecord{
+		BotID:       bot.ID,
+		BotNodeID:   bot.NodeID,
+		BotNodeNum:  bot.NodeNum,
+		MessageType: messageType,
+		ChannelID:   channelID,
+		ToNodeID:    toNodeID,
+		ToNodeNum:   int64PtrOrNil(toNodeNum, false),
+		Topic:       topic,
+		PacketID:    int64(packetID),
+		Text:        text,
+		PayloadLen:  int64(len(raw)),
+		Encrypted:   true,
+		Status:      storepkg.BotMessageStatusPending,
+		CreatedBy:   strings.TrimSpace(req.CreatedBy),
+	}
+	return s.persistAndPublish(row, topic, raw)
+}
+
+// sendPKIDirect 按固件 PKI 流程发送私聊：
+//   - 从 nodeinfo 中查目标节点的 X25519 公钥
+//   - 用 bot 自身私钥与对端公钥派生共享密钥，AES-CCM(M=8,L=2) 加密
+//   - ServiceEnvelope.channel_id = "PKI"，topic 也用 "PKI"
+func (s *Service) sendPKIDirect(bot *storepkg.BotNodeRecord, fromNodeNum, toNodeNum uint32, toNodeID *string, packetID uint32, text, createdBy string) (*storepkg.BotMessageRecord, error) {
+	if toNodeID == nil {
+		return nil, fmt.Errorf("target node id is required for pki direct message")
+	}
+	privateKeyB64 := strings.TrimSpace(bot.PrivateKey)
+	if privateKeyB64 == "" {
+		return nil, fmt.Errorf("bot has no private key, regenerate keys first")
+	}
+	privateKey, err := base64.StdEncoding.DecodeString(privateKeyB64)
+	if err != nil {
+		return nil, fmt.Errorf("invalid bot private key: %w", err)
+	}
+	senderPublic, err := storepkg.DecodeBotPublicKey(*bot)
+	if err != nil {
+		return nil, err
+	}
+	recipientPublic, err := s.lookupRecipientPublicKey(*toNodeID)
+	if err != nil {
+		return nil, err
+	}
+
+	raw, err := mqtpp.BuildPKITextMessageServiceEnvelope(mqtpp.PKITextMessageBuildOptions{
+		FromNodeNum:   fromNodeNum,
+		ToNodeNum:     toNodeNum,
+		PacketID:      packetID,
+		GatewayID:     bot.NodeID,
+		ViaMQTT:       true,
+		SenderPrivate: privateKey,
+		RecipientPub:  recipientPublic,
+		SenderPublic:  senderPublic,
+		Text:          text,
+	})
+	if err != nil {
+		return nil, err
+	}
+	topic := botMQTTTopic(bot.TopicPrefix, mqtpp.PKIChannelID, bot.NodeID)
+	row := &storepkg.BotMessageRecord{
+		BotID:       bot.ID,
+		BotNodeID:   bot.NodeID,
+		BotNodeNum:  bot.NodeNum,
+		MessageType: storepkg.BotMessageTypeDirect,
+		ChannelID:   mqtpp.PKIChannelID,
+		ToNodeID:    toNodeID,
+		ToNodeNum:   int64PtrOrNil(int64(toNodeNum), true),
+		Topic:       topic,
+		PacketID:    int64(packetID),
+		Text:        text,
+		PayloadLen:  int64(len(raw)),
+		Encrypted:   true,
+		Status:      storepkg.BotMessageStatusPending,
+		CreatedBy:   strings.TrimSpace(createdBy),
+	}
+	result, err := s.persistAndPublish(row, topic, raw)
+	// 不论发送结果如何，都把 DM 镜像写入 bot_direct_messages 以驱动 /admin/bot/direct 渲染。
+	// 这里把发送结果（status/error/published_at）同步过去——成功时 status=published，
+	// 失败时 status=failed，前端就能看到本地视图与发送日志一致。
+	s.recordOutboundDirectMessage(bot, row, *toNodeID, toNodeNum, text, len(raw), err)
+	return result, err
+}
+
+// recordOutboundDirectMessage 把出向 PKI DM 写入 bot_direct_messages。失败仅打日志。
+func (s *Service) recordOutboundDirectMessage(bot *storepkg.BotNodeRecord, msg *storepkg.BotMessageRecord, peerNodeID string, peerNodeNum uint32, text string, payloadLen int, sendErr error) {
+	if s == nil || s.store == nil || msg == nil || bot == nil {
+		return
+	}
+	status := msg.Status
+	if status == "" {
+		if sendErr != nil {
+			status = storepkg.BotMessageStatusFailed
+		} else {
+			status = storepkg.BotMessageStatusPublished
+		}
+	}
+	errText := msg.Error
+	if errText == "" && sendErr != nil {
+		errText = sendErr.Error()
+	}
+	createdBy := strings.TrimSpace(msg.CreatedBy)
+	var createdByPtr *string
+	if createdBy != "" {
+		createdByPtr = &createdBy
+	}
+	gateway := strings.TrimSpace(bot.NodeID)
+	var gatewayPtr *string
+	if gateway != "" {
+		gatewayPtr = &gateway
+	}
+	var botMessageID *uint64
+	if msg.ID != 0 {
+		id := msg.ID
+		botMessageID = &id
+	}
+	now := time.Now()
+	dm := &storepkg.BotDirectMessageRecord{
+		BotID:        bot.ID,
+		BotNodeID:    bot.NodeID,
+		BotNodeNum:   bot.NodeNum,
+		PeerNodeID:   peerNodeID,
+		PeerNodeNum:  int64(peerNodeNum),
+		Direction:    storepkg.BotDirectMessageDirectionOutbound,
+		Topic:        msg.Topic,
+		PacketID:     msg.PacketID,
+		Text:         text,
+		PayloadLen:   int64(payloadLen),
+		PKIEncrypted: true,
+		WantAck:      false, // 我们当前发送的 DM 默认不显式请求 ack
+		GatewayID:    gatewayPtr,
+		Status:       status,
+		Error:        strings.TrimSpace(errText),
+		BotMessageID: botMessageID,
+		CreatedBy:    createdByPtr,
+		PublishedAt:  msg.PublishedAt,
+		// 出向消息从产生那一刻起就视为“已读”，未读计数只关心 inbound。
+		ReadAt: &now,
+	}
+	if err := s.store.InsertBotDirectMessage(dm); err != nil {
+		printJSON(map[string]any{
+			"event":           "bot_direct_message_outbound_persist_failed",
+			"bot_node_id":     bot.NodeID,
+			"peer_node_id":    peerNodeID,
+			"bot_message_id":  msg.ID,
+			"error":           err.Error(),
+		})
+	}
+}
+
+// lookupRecipientPublicKey 从 nodeinfo 表中按 node_id 查询目标节点的 X25519 公钥（hex 编码）。
+func (s *Service) lookupRecipientPublicKey(nodeID string) ([]byte, error) {
+	node, err := s.store.GetNodeInfo(nodeID)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, fmt.Errorf("recipient node %s not found in nodeinfo, cannot send PKI message", nodeID)
+		}
+		return nil, err
+	}
+	if node.PublicKey == nil || strings.TrimSpace(*node.PublicKey) == "" {
+		return nil, fmt.Errorf("recipient node %s has no public key on file", nodeID)
+	}
+	keyHex := strings.TrimSpace(*node.PublicKey)
+	keyBytes, err := hex.DecodeString(keyHex)
+	if err != nil {
+		// 兼容历史上可能存储为 base64 的情况
+		if alt, altErr := base64.StdEncoding.DecodeString(keyHex); altErr == nil {
+			keyBytes = alt
+		} else {
+			return nil, fmt.Errorf("invalid recipient public key for %s: %w", nodeID, err)
+		}
+	}
+	if len(keyBytes) != 32 {
+		return nil, fmt.Errorf("recipient public key for %s has unexpected length %d", nodeID, len(keyBytes))
+	}
+	return keyBytes, nil
+}
+
+// persistAndPublish 把消息记录入库后发布到 MQTT，统一处理失败状态写回。
+func (s *Service) persistAndPublish(row *storepkg.BotMessageRecord, topic string, raw []byte) (*storepkg.BotMessageRecord, error) {
+	if err := s.store.InsertBotMessage(row); err != nil {
+		return nil, err
+	}
+	if s.server == nil {
+		_ = s.store.UpdateBotMessageStatus(row.ID, storepkg.BotMessageStatusFailed, "mqtt server is not configured", nil)
+		row.Status = storepkg.BotMessageStatusFailed
+		row.Error = "mqtt server is not configured"
+		return row, fmt.Errorf("mqtt server is not configured")
+	}
+	if err := s.server.Publish(topic, raw, false, 0); err != nil {
+		_ = s.store.UpdateBotMessageStatus(row.ID, storepkg.BotMessageStatusFailed, err.Error(), nil)
+		row.Status = storepkg.BotMessageStatusFailed
+		row.Error = err.Error()
+		return row, err
+	}
+	now := time.Now()
+	if err := s.store.UpdateBotMessageStatus(row.ID, storepkg.BotMessageStatusPublished, "", &now); err != nil {
+		return nil, err
+	}
+	row.Status = storepkg.BotMessageStatusPublished
+	row.Error = ""
+	row.PublishedAt = &now
+	return row, nil
+}
+
+func (s *Service) runNodeInfoBroadcaster(ctx context.Context) {
+	ticker := time.NewTicker(time.Minute)
+	defer ticker.Stop()
+	s.broadcastDueNodeInfo(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			s.broadcastDueNodeInfo(ctx)
+		}
+	}
+}
+
+func (s *Service) broadcastDueNodeInfo(ctx context.Context) {
+	rows, err := s.store.ListBotNodes(storepkg.ListOptions{Limit: 500})
+	if err != nil {
+		printJSON(map[string]any{"event": "bot_nodeinfo_broadcast_failed", "error": err.Error()})
+		return
+	}
+	now := time.Now()
+	for _, bot := range rows {
+		if ctx.Err() != nil {
+			return
+		}
+		if !bot.Enabled || !bot.NodeInfoBroadcastEnabled {
+			continue
+		}
+		interval := time.Duration(bot.NodeInfoBroadcastIntervalSeconds) * time.Second
+		if interval <= 0 {
+			interval = time.Duration(storepkg.BotDefaultNodeInfoBroadcastSeconds) * time.Second
+		}
+		if bot.LastNodeInfoBroadcastAt != nil && now.Sub(*bot.LastNodeInfoBroadcastAt) < interval {
+			continue
+		}
+		if err := s.PublishNodeInfo(ctx, bot); err != nil {
+			printJSON(map[string]any{"event": "bot_nodeinfo_broadcast_failed", "bot_id": bot.ID, "node_id": bot.NodeID, "error": err.Error()})
+		}
+	}
+}
+
+func (s *Service) PublishNodeInfoByID(ctx context.Context, id uint64) (*storepkg.BotNodeRecord, error) {
+	if s == nil || s.store == nil {
+		return nil, fmt.Errorf("bot service is not configured")
+	}
+	bot, err := s.store.GetBotNode(id)
+	if err != nil {
+		return nil, err
+	}
+	if !bot.Enabled {
+		return nil, fmt.Errorf("bot node is disabled")
+	}
+	if err := s.PublishNodeInfo(ctx, *bot); err != nil {
+		return nil, err
+	}
+	updated, err := s.store.GetBotNode(id)
+	if err != nil {
+		return bot, nil
+	}
+	return updated, nil
+}
+
+func (s *Service) PublishNodeInfo(_ context.Context, bot storepkg.BotNodeRecord) error {
+	if s == nil || s.server == nil {
+		return fmt.Errorf("mqtt server is not configured")
+	}
+	if strings.TrimSpace(bot.PublicKey) == "" && s.store != nil {
+		updated, err := s.store.RegenerateBotNodeKeys(bot.ID)
+		if err != nil {
+			return err
+		}
+		bot = *updated
+	}
+	psk := strings.TrimSpace(bot.PSK)
+	if psk == "" {
+		psk = storepkg.BotDefaultPSK
+	}
+	key, err := mqtpp.ExpandPSK(psk)
+	if err != nil {
+		return err
+	}
+	packetID, err := randomPacketID()
+	if err != nil {
+		return err
+	}
+	publicKey, err := storepkg.DecodeBotPublicKey(bot)
+	if err != nil {
+		return err
+	}
+	raw, err := mqtpp.BuildNodeInfoServiceEnvelope(mqtpp.NodeInfoBuildOptions{
+		PacketBuildOptions: mqtpp.PacketBuildOptions{
+			FromNodeNum: uint32(bot.NodeNum),
+			ToNodeNum:   mqtpp.NodeNumBroadcast,
+			PacketID:    packetID,
+			ChannelID:   bot.DefaultChannelID,
+			GatewayID:   bot.NodeID,
+			PSK:         key,
+			Encrypt:     true,
+			ViaMQTT:     true,
+		},
+		NodeID:     bot.NodeID,
+		LongName:   bot.LongName,
+		ShortName:  bot.ShortName,
+		HWModel:    255,
+		Role:       0,
+		IsLicensed: false,
+		PublicKey:  publicKey,
+	})
+	if err != nil {
+		return err
+	}
+	topic := botMQTTTopic(bot.TopicPrefix, bot.DefaultChannelID, bot.NodeID)
+	if err := s.server.Publish(topic, raw, false, 0); err != nil {
+		return err
+	}
+	if s.store != nil {
+		valid, _, record := mqtpp.MQTTPP(topic, raw, key, mqtpp.Options{AllowEncryptedForwarding: true})
+		if valid && record["type"] == "nodeinfo" {
+			if err := s.store.UpsertNodeInfo(record); err != nil {
+				return err
+			}
+		}
+	}
+	return s.store.UpdateBotNodeInfoBroadcastAt(bot.ID, time.Now())
+}
+
+func normalizeBotMessageType(value string) (string, error) {
+	switch strings.TrimSpace(value) {
+	case "", storepkg.BotMessageTypeChannel:
+		return storepkg.BotMessageTypeChannel, nil
+	case storepkg.BotMessageTypeDirect:
+		return storepkg.BotMessageTypeDirect, nil
+	default:
+		return "", fmt.Errorf("message type must be channel or direct")
+	}
+}
+
+func botMessageTarget(messageType string, req SendTextRequest) (int64, *string, error) {
+	if messageType == storepkg.BotMessageTypeChannel {
+		return int64(mqtpp.NodeNumBroadcast), nil, nil
+	}
+	if req.ToNodeNum != nil && *req.ToNodeNum > 0 {
+		if err := storepkg.ValidateBotNodeNum(*req.ToNodeNum); err != nil {
+			return 0, nil, err
+		}
+		nodeID := mqtpp.NodeNumToID(uint32(*req.ToNodeNum))
+		return *req.ToNodeNum, &nodeID, nil
+	}
+	toNodeID := strings.TrimSpace(req.ToNodeID)
+	if toNodeID == "" {
+		return 0, nil, fmt.Errorf("target node is required for direct message")
+	}
+	nodeNum, err := mqtpp.ParseNodeID(toNodeID)
+	if err != nil {
+		return 0, nil, err
+	}
+	if err := storepkg.ValidateBotNodeNum(int64(nodeNum)); err != nil {
+		return 0, nil, err
+	}
+	normalized := mqtpp.NodeNumToID(nodeNum)
+	return int64(nodeNum), &normalized, nil
+}
+
+func botMQTTTopic(topicPrefix, channelID, nodeID string) string {
+	prefix := strings.Trim(strings.TrimSpace(topicPrefix), "/")
+	if prefix == "" {
+		prefix = storepkg.BotDefaultTopicPrefix
+	}
+	if strings.HasSuffix(prefix, "/2/e") {
+		return prefix + "/" + channelID + "/" + nodeID
+	}
+	return prefix + "/2/e/" + channelID + "/" + nodeID
+}
+
+func randomPacketID() (uint32, error) {
+	for i := 0; i < 8; i++ {
+		var buf [4]byte
+		if _, err := rand.Read(buf[:]); err != nil {
+			return 0, err
+		}
+		id := binary.LittleEndian.Uint32(buf[:])
+		if id != 0 {
+			return id, nil
+		}
+	}
+	return 0, fmt.Errorf("generate packet id failed")
+}
+
+func int64PtrOrNil(value int64, ok bool) *int64 {
+	if !ok {
+		return nil
+	}
+	return &value
+}
@@ -0,0 +1,8 @@
+package bot
+
+// printJSON 是 bot service 的内部诊断输出钩子；当前为 noop，与重构前
+// main.go 中的 printJSON 行为一致（注释掉了实际写出）。
+// 如需调试可直接替换实现。
+func printJSON(record map[string]any) {
+	_ = record
+}