重构:拆出 internal/config 与 internal/store 子包

把工程根目录中和"配置加载"、"数据存储"两个领域相关的全部 Go 文件迁到
internal/ 下的子包,按功能分组的第一阶段成果。

internal/config/
- 从 config.go 抽出 Config / MQTTConfig / WebConfig / DatabaseConfig 等
  类型并大写导出,函数改名 Default/Load/Write/Validate/BuildTLS 等。
- 测试随被测代码迁移成 package config 的内部测试。
- 根目录留 config.go 作桥接:用 type alias 让旧的小写名(config /
  mqttConfig / databaseConfig 等)继续可用,避免修改 30+ 处调用方。

internal/store/
- 把 db.go、store_query.go、db_write_queue.go 与 13 个 *_store.go 一并
  迁入;26 个 *Record 类型与 store 同包以避免循环依赖。
- store -> Store;50+ 标识符从小写未导出改为大写导出(包括 record、
  ListOptions、错误变量、bot/llm/runtime 常量、helpers 等)。
- 新增 DB() / Driver() 访问器供 ai 子系统使用,避免直接访问私有字段。
- bot_pki_store.go 独立出来,把 PKI 解密所需的 store 方法集中归类。
- helpers.go 提供 hashPassword / uint32FromRecord / printJSON 等以前在
  其他根目录文件中的辅助;test_helpers_test.go 提供 verifyPassword
  与 publicMapTileSourceDTO 让测试可以本地运行而不依赖 main 包。

根目录新增:
- store_bridge.go:完整 type-alias / 函数包装层,把 internal/store 的
  导出名映射回旧的小写名,让 admin_*_routes.go、web.go、bot_service.go
  等仍未迁出的文件继续编译。后续步骤把它们迁到各自领域包后可逐步删除。
- test_helpers_test.go:根目录测试沿用 openTestStore 的入口。

go build ./... 与 go test ./... 全部通过;测试数量与重构前一致。

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
2026-06-18 13:56:36 +08:00
co-authored by Claude
parent 6f6b06e37d
commit eff4972668
32 changed files with 1759 additions and 1462 deletions
+325
View File
@@ -0,0 +1,325 @@
package store
import (
"errors"
"fmt"
"net"
"strings"
"time"
"gorm.io/gorm"
)
const ForbiddenWordMatchContains = "contains"
var ErrBlockingAlreadyExists = errors.New("blocking rule already exists")
func (s *Store) ListNodeBlocking(opts ListOptions) ([]NodeBlockingRecord, error) {
opts = NormalizeListOptions(opts)
var rows []NodeBlockingRecord
q := s.db.Model(&NodeBlockingRecord{}).
Order("updated_at DESC").
Order("id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
func (s *Store) CountNodeBlocking(opts ListOptions) (int64, error) {
var total int64
return total, s.db.Model(&NodeBlockingRecord{}).Count(&total).Error
}
func (s *Store) ListEnabledNodeBlocking() ([]NodeBlockingRecord, error) {
var rows []NodeBlockingRecord
return rows, s.db.Where("enabled = ?", true).Find(&rows).Error
}
func (s *Store) CreateNodeBlocking(nodeID string, nodeNum *int64, reason string, enabled bool) (*NodeBlockingRecord, error) {
nodeID = strings.TrimSpace(nodeID)
if nodeID == "" {
return nil, fmt.Errorf("node id is required")
}
if err := s.ensureNodeBlockingUnique(0, nodeID); err != nil {
return nil, err
}
row := NodeBlockingRecord{NodeID: nodeID, NodeNum: nodeNum, Reason: strings.TrimSpace(reason), Enabled: enabled}
if err := s.db.Create(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) UpdateNodeBlocking(id uint64, nodeID string, nodeNum *int64, reason string, enabled bool) (*NodeBlockingRecord, error) {
if id == 0 {
return nil, fmt.Errorf("blocking rule id is required")
}
nodeID = strings.TrimSpace(nodeID)
if nodeID == "" {
return nil, fmt.Errorf("node id is required")
}
if _, err := s.getNodeBlockingByID(id); err != nil {
return nil, err
}
if err := s.ensureNodeBlockingUnique(id, nodeID); err != nil {
return nil, err
}
updates := map[string]any{"node_id": nodeID, "node_num": nodeNum, "reason": strings.TrimSpace(reason), "enabled": enabled, "updated_at": time.Now()}
if err := s.db.Model(&NodeBlockingRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return nil, err
}
return s.getNodeBlockingByID(id)
}
func (s *Store) DeleteNodeBlocking(id uint64) error {
result := s.db.Where("id = ?", id).Delete(&NodeBlockingRecord{})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
}
func (s *Store) ListIPBlocking(opts ListOptions) ([]IPBlockingRecord, error) {
opts = NormalizeListOptions(opts)
var rows []IPBlockingRecord
q := s.db.Model(&IPBlockingRecord{}).
Order("updated_at DESC").
Order("id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
func (s *Store) CountIPBlocking(opts ListOptions) (int64, error) {
var total int64
return total, s.db.Model(&IPBlockingRecord{}).Count(&total).Error
}
func (s *Store) ListEnabledIPBlocking() ([]IPBlockingRecord, error) {
var rows []IPBlockingRecord
return rows, s.db.Where("enabled = ?", true).Find(&rows).Error
}
func (s *Store) CreateIPBlocking(ipValue string, reason string, enabled bool) (*IPBlockingRecord, error) {
value, err := normalizeIPBlockingValue(ipValue)
if err != nil {
return nil, err
}
if err := s.ensureIPBlockingUnique(0, value); err != nil {
return nil, err
}
row := IPBlockingRecord{IPValue: value, Reason: strings.TrimSpace(reason), Enabled: enabled}
if err := s.db.Create(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) UpdateIPBlocking(id uint64, ipValue string, reason string, enabled bool) (*IPBlockingRecord, error) {
if id == 0 {
return nil, fmt.Errorf("blocking rule id is required")
}
value, err := normalizeIPBlockingValue(ipValue)
if err != nil {
return nil, err
}
if _, err := s.getIPBlockingByID(id); err != nil {
return nil, err
}
if err := s.ensureIPBlockingUnique(id, value); err != nil {
return nil, err
}
updates := map[string]any{"ip_value": value, "reason": strings.TrimSpace(reason), "enabled": enabled, "updated_at": time.Now()}
if err := s.db.Model(&IPBlockingRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return nil, err
}
return s.getIPBlockingByID(id)
}
func (s *Store) DeleteIPBlocking(id uint64) error {
result := s.db.Where("id = ?", id).Delete(&IPBlockingRecord{})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
}
func (s *Store) ListForbiddenWordBlocking(opts ListOptions) ([]ForbiddenWordBlockingRecord, error) {
opts = NormalizeListOptions(opts)
var rows []ForbiddenWordBlockingRecord
q := s.db.Model(&ForbiddenWordBlockingRecord{}).
Order("updated_at DESC").
Order("id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
func (s *Store) CountForbiddenWordBlocking(opts ListOptions) (int64, error) {
var total int64
return total, s.db.Model(&ForbiddenWordBlockingRecord{}).Count(&total).Error
}
func (s *Store) ListEnabledForbiddenWordBlocking() ([]ForbiddenWordBlockingRecord, error) {
var rows []ForbiddenWordBlockingRecord
return rows, s.db.Where("enabled = ?", true).Find(&rows).Error
}
func (s *Store) CreateForbiddenWordBlocking(word, matchType string, caseSensitive bool, reason string, enabled bool) (*ForbiddenWordBlockingRecord, error) {
word = strings.TrimSpace(word)
if word == "" {
return nil, fmt.Errorf("forbidden word is required")
}
matchType, err := normalizeForbiddenWordMatchType(matchType)
if err != nil {
return nil, err
}
if err := s.ensureForbiddenWordBlockingUnique(0, word); err != nil {
return nil, err
}
row := ForbiddenWordBlockingRecord{Word: word, MatchType: matchType, CaseSensitive: caseSensitive, Reason: strings.TrimSpace(reason), Enabled: enabled}
if err := s.db.Create(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) UpdateForbiddenWordBlocking(id uint64, word, matchType string, caseSensitive bool, reason string, enabled bool) (*ForbiddenWordBlockingRecord, error) {
if id == 0 {
return nil, fmt.Errorf("blocking rule id is required")
}
word = strings.TrimSpace(word)
if word == "" {
return nil, fmt.Errorf("forbidden word is required")
}
matchType, err := normalizeForbiddenWordMatchType(matchType)
if err != nil {
return nil, err
}
if _, err := s.getForbiddenWordBlockingByID(id); err != nil {
return nil, err
}
if err := s.ensureForbiddenWordBlockingUnique(id, word); err != nil {
return nil, err
}
updates := map[string]any{"word": word, "match_type": matchType, "case_sensitive": caseSensitive, "reason": strings.TrimSpace(reason), "enabled": enabled, "updated_at": time.Now()}
if err := s.db.Model(&ForbiddenWordBlockingRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return nil, err
}
return s.getForbiddenWordBlockingByID(id)
}
func (s *Store) DeleteForbiddenWordBlocking(id uint64) error {
result := s.db.Where("id = ?", id).Delete(&ForbiddenWordBlockingRecord{})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
}
func (s *Store) getNodeBlockingByID(id uint64) (*NodeBlockingRecord, error) {
var row NodeBlockingRecord
if err := s.db.Where("id = ?", id).Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) getIPBlockingByID(id uint64) (*IPBlockingRecord, error) {
var row IPBlockingRecord
if err := s.db.Where("id = ?", id).Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) getForbiddenWordBlockingByID(id uint64) (*ForbiddenWordBlockingRecord, error) {
var row ForbiddenWordBlockingRecord
if err := s.db.Where("id = ?", id).Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) ensureNodeBlockingUnique(id uint64, nodeID string) error {
var existing NodeBlockingRecord
q := s.db.Where("node_id = ?", nodeID)
if id != 0 {
q = q.Where("id <> ?", id)
}
err := q.Take(&existing).Error
if err == nil {
return ErrBlockingAlreadyExists
}
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
return err
}
func (s *Store) ensureIPBlockingUnique(id uint64, ipValue string) error {
var existing IPBlockingRecord
q := s.db.Where("ip_value = ?", ipValue)
if id != 0 {
q = q.Where("id <> ?", id)
}
err := q.Take(&existing).Error
if err == nil {
return ErrBlockingAlreadyExists
}
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
return err
}
func (s *Store) ensureForbiddenWordBlockingUnique(id uint64, word string) error {
var existing ForbiddenWordBlockingRecord
q := s.db.Where("word = ?", word)
if id != 0 {
q = q.Where("id <> ?", id)
}
err := q.Take(&existing).Error
if err == nil {
return ErrBlockingAlreadyExists
}
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
return err
}
func normalizeIPBlockingValue(value string) (string, error) {
value = strings.TrimSpace(value)
if value == "" {
return "", fmt.Errorf("ip value is required")
}
if ip := net.ParseIP(value); ip != nil {
return ip.String(), nil
}
_, ipNet, err := net.ParseCIDR(value)
if err == nil {
return ipNet.String(), nil
}
return "", fmt.Errorf("ip value must be a valid IP or CIDR")
}
func normalizeForbiddenWordMatchType(matchType string) (string, error) {
matchType = strings.TrimSpace(matchType)
if matchType == "" {
return ForbiddenWordMatchContains, nil
}
if matchType != ForbiddenWordMatchContains {
return "", fmt.Errorf("unsupported forbidden word match type")
}
return matchType, nil
}
+207
View File
@@ -0,0 +1,207 @@
package store
import (
"errors"
"testing"
"gorm.io/gorm"
)
func TestNodeBlockingCRUD(t *testing.T) {
st := openTestStore(t)
defer st.Close()
nodeNum := int64(305419896)
rule, err := st.CreateNodeBlocking(" !12345678 ", &nodeNum, " noisy node ", true)
if err != nil {
t.Fatalf("CreateNodeBlocking() error = %v", err)
}
if rule.NodeID != "!12345678" || rule.NodeNum == nil || *rule.NodeNum != nodeNum || rule.Reason != "noisy node" || !rule.Enabled {
t.Fatalf("created node rule = %+v, want normalized fields", rule)
}
if _, err := st.CreateNodeBlocking("!12345678", nil, "duplicate", true); !errors.Is(err, ErrBlockingAlreadyExists) {
t.Fatalf("duplicate CreateNodeBlocking() error = %v, want ErrBlockingAlreadyExists", err)
}
updatedNum := int64(7)
updated, err := st.UpdateNodeBlocking(rule.ID, "!00000007", &updatedNum, "updated", false)
if err != nil {
t.Fatalf("UpdateNodeBlocking() error = %v", err)
}
if updated.NodeID != "!00000007" || updated.NodeNum == nil || *updated.NodeNum != updatedNum || updated.Reason != "updated" || updated.Enabled {
t.Fatalf("updated node rule = %+v, want updated fields", updated)
}
rows, err := st.ListNodeBlocking(ListOptions{})
if err != nil {
t.Fatalf("ListNodeBlocking() error = %v", err)
}
if len(rows) != 1 || rows[0].ID != rule.ID {
t.Fatalf("ListNodeBlocking() = %+v, want one updated rule", rows)
}
total, err := st.CountNodeBlocking(ListOptions{})
if err != nil || total != 1 {
t.Fatalf("CountNodeBlocking() = %d, %v, want 1, nil", total, err)
}
if err := st.DeleteNodeBlocking(rule.ID); err != nil {
t.Fatalf("DeleteNodeBlocking() error = %v", err)
}
if err := st.DeleteNodeBlocking(rule.ID); !errors.Is(err, gorm.ErrRecordNotFound) {
t.Fatalf("DeleteNodeBlocking(missing) error = %v, want record not found", err)
}
}
func TestNodeBlockingValidation(t *testing.T) {
st := openTestStore(t)
defer st.Close()
if _, err := st.CreateNodeBlocking(" ", nil, "", true); err == nil {
t.Fatal("CreateNodeBlocking(empty) error = nil, want error")
}
if _, err := st.UpdateNodeBlocking(1, "!missing", nil, "", true); !errors.Is(err, gorm.ErrRecordNotFound) {
t.Fatalf("UpdateNodeBlocking(missing) error = %v, want record not found", err)
}
}
func TestIPBlockingCRUDAndValidation(t *testing.T) {
st := openTestStore(t)
defer st.Close()
rule, err := st.CreateIPBlocking(" 127.0.0.1 ", "local", true)
if err != nil {
t.Fatalf("CreateIPBlocking(ip) error = %v", err)
}
if rule.IPValue != "127.0.0.1" || rule.Reason != "local" || !rule.Enabled {
t.Fatalf("created ip rule = %+v, want normalized IP", rule)
}
cidr, err := st.CreateIPBlocking("192.168.1.99/24", "cidr", true)
if err != nil {
t.Fatalf("CreateIPBlocking(cidr) error = %v", err)
}
if cidr.IPValue != "192.168.1.0/24" {
t.Fatalf("cidr IPValue = %q, want 192.168.1.0/24", cidr.IPValue)
}
if _, err := st.CreateIPBlocking("127.0.0.1", "duplicate", true); !errors.Is(err, ErrBlockingAlreadyExists) {
t.Fatalf("duplicate CreateIPBlocking() error = %v, want ErrBlockingAlreadyExists", err)
}
if _, err := st.CreateIPBlocking("not-an-ip", "invalid", true); err == nil {
t.Fatal("CreateIPBlocking(invalid) error = nil, want error")
}
updated, err := st.UpdateIPBlocking(rule.ID, "10.0.0.0/8", "updated", false)
if err != nil {
t.Fatalf("UpdateIPBlocking() error = %v", err)
}
if updated.IPValue != "10.0.0.0/8" || updated.Reason != "updated" || updated.Enabled {
t.Fatalf("updated ip rule = %+v, want updated fields", updated)
}
rows, err := st.ListIPBlocking(ListOptions{})
if err != nil {
t.Fatalf("ListIPBlocking() error = %v", err)
}
if len(rows) != 2 {
t.Fatalf("ListIPBlocking() length = %d, want 2", len(rows))
}
total, err := st.CountIPBlocking(ListOptions{})
if err != nil || total != 2 {
t.Fatalf("CountIPBlocking() = %d, %v, want 2, nil", total, err)
}
if err := st.DeleteIPBlocking(rule.ID); err != nil {
t.Fatalf("DeleteIPBlocking() error = %v", err)
}
if err := st.DeleteIPBlocking(rule.ID); !errors.Is(err, gorm.ErrRecordNotFound) {
t.Fatalf("DeleteIPBlocking(missing) error = %v, want record not found", err)
}
}
func TestListEnabledBlockingRules(t *testing.T) {
st := openTestStore(t)
defer st.Close()
nodeNum := int64(1)
if _, err := st.CreateNodeBlocking("!00000001", &nodeNum, "enabled", true); err != nil {
t.Fatalf("CreateNodeBlocking(enabled) error = %v", err)
}
if _, err := st.CreateNodeBlocking("!00000002", nil, "disabled", false); err != nil {
t.Fatalf("CreateNodeBlocking(disabled) error = %v", err)
}
if rows, err := st.ListEnabledNodeBlocking(); err != nil || len(rows) != 1 || rows[0].NodeID != "!00000001" {
t.Fatalf("ListEnabledNodeBlocking() = %+v, %v, want only enabled node", rows, err)
}
if _, err := st.CreateIPBlocking("127.0.0.1", "enabled", true); err != nil {
t.Fatalf("CreateIPBlocking(enabled) error = %v", err)
}
if _, err := st.CreateIPBlocking("192.168.1.1", "disabled", false); err != nil {
t.Fatalf("CreateIPBlocking(disabled) error = %v", err)
}
if rows, err := st.ListEnabledIPBlocking(); err != nil || len(rows) != 1 || rows[0].IPValue != "127.0.0.1" {
t.Fatalf("ListEnabledIPBlocking() = %+v, %v, want only enabled IP", rows, err)
}
if _, err := st.CreateForbiddenWordBlocking("spam", "contains", false, "enabled", true); err != nil {
t.Fatalf("CreateForbiddenWordBlocking(enabled) error = %v", err)
}
if _, err := st.CreateForbiddenWordBlocking("eggs", "contains", false, "disabled", false); err != nil {
t.Fatalf("CreateForbiddenWordBlocking(disabled) error = %v", err)
}
if rows, err := st.ListEnabledForbiddenWordBlocking(); err != nil || len(rows) != 1 || rows[0].Word != "spam" {
t.Fatalf("ListEnabledForbiddenWordBlocking() = %+v, %v, want only enabled word", rows, err)
}
}
func TestForbiddenWordBlockingCRUDAndValidation(t *testing.T) {
st := openTestStore(t)
defer st.Close()
rule, err := st.CreateForbiddenWordBlocking(" spam ", "", false, "junk", true)
if err != nil {
t.Fatalf("CreateForbiddenWordBlocking() error = %v", err)
}
if rule.Word != "spam" || rule.MatchType != ForbiddenWordMatchContains || rule.CaseSensitive || rule.Reason != "junk" || !rule.Enabled {
t.Fatalf("created word rule = %+v, want normalized fields", rule)
}
if _, err := st.CreateForbiddenWordBlocking("spam", "contains", false, "duplicate", true); !errors.Is(err, ErrBlockingAlreadyExists) {
t.Fatalf("duplicate CreateForbiddenWordBlocking() error = %v, want ErrBlockingAlreadyExists", err)
}
if _, err := st.CreateForbiddenWordBlocking(" ", "contains", false, "empty", true); err == nil {
t.Fatal("CreateForbiddenWordBlocking(empty) error = nil, want error")
}
if _, err := st.CreateForbiddenWordBlocking("regex", "regex", false, "unsupported", true); err == nil {
t.Fatal("CreateForbiddenWordBlocking(unsupported match type) error = nil, want error")
}
updated, err := st.UpdateForbiddenWordBlocking(rule.ID, "Spam", "contains", true, "updated", false)
if err != nil {
t.Fatalf("UpdateForbiddenWordBlocking() error = %v", err)
}
if updated.Word != "Spam" || updated.MatchType != "contains" || !updated.CaseSensitive || updated.Reason != "updated" || updated.Enabled {
t.Fatalf("updated word rule = %+v, want updated fields", updated)
}
rows, err := st.ListForbiddenWordBlocking(ListOptions{})
if err != nil {
t.Fatalf("ListForbiddenWordBlocking() error = %v", err)
}
if len(rows) != 1 || rows[0].ID != rule.ID {
t.Fatalf("ListForbiddenWordBlocking() = %+v, want one updated rule", rows)
}
total, err := st.CountForbiddenWordBlocking(ListOptions{})
if err != nil || total != 1 {
t.Fatalf("CountForbiddenWordBlocking() = %d, %v, want 1, nil", total, err)
}
if err := st.DeleteForbiddenWordBlocking(rule.ID); err != nil {
t.Fatalf("DeleteForbiddenWordBlocking() error = %v", err)
}
if err := st.DeleteForbiddenWordBlocking(rule.ID); !errors.Is(err, gorm.ErrRecordNotFound) {
t.Fatalf("DeleteForbiddenWordBlocking(missing) error = %v, want record not found", err)
}
}
+325
View File
@@ -0,0 +1,325 @@
package store
import (
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"gorm.io/gorm"
)
type BotDirectMessageListOptions struct {
ListOptions
BotID uint64
PeerNodeNum int64
Direction string
}
// InsertBotDirectMessage 把一条机器人 DM(出向或入向)写入 bot_direct_messages 表。
func (s *Store) InsertBotDirectMessage(row *BotDirectMessageRecord) error {
if s == nil || s.db == nil {
return fmt.Errorf("Store is not configured")
}
if row == nil {
return fmt.Errorf("bot direct message is required")
}
if row.Direction == "" {
return fmt.Errorf("bot direct message direction is required")
}
return s.db.Create(row).Error
}
// UpdateBotDirectMessageStatus 更新一条出向 DM 的发送状态(pending → published/failed)。
func (s *Store) UpdateBotDirectMessageStatus(id uint64, status, errText string, publishedAt *time.Time) error {
if s == nil || s.db == nil {
return fmt.Errorf("Store is not configured")
}
if id == 0 {
return fmt.Errorf("bot direct message id is required")
}
updates := map[string]any{
"status": status,
"error": strings.TrimSpace(errText),
"published_at": publishedAt,
}
result := s.db.Model(&BotDirectMessageRecord{}).Where("id = ?", id).Updates(updates)
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
}
// ListBotDirectMessagesByConversation 按 (bot, peer) 反序拉取 DM 历史,给 /admin/bot/direct 页面。
func (s *Store) ListBotDirectMessagesByConversation(opts BotDirectMessageListOptions) ([]BotDirectMessageRecord, error) {
if s == nil || s.db == nil {
return nil, fmt.Errorf("Store is not configured")
}
if opts.BotID == 0 {
return nil, fmt.Errorf("bot id is required")
}
if opts.PeerNodeNum == 0 {
return nil, fmt.Errorf("peer node num is required")
}
opts.ListOptions = NormalizeListOptions(opts.ListOptions)
var rows []BotDirectMessageRecord
q := s.db.Model(&BotDirectMessageRecord{}).
Where("bot_id = ? AND peer_node_num = ?", opts.BotID, opts.PeerNodeNum).
Order("created_at DESC").
Order("id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
if opts.Direction != "" {
q = q.Where("direction = ?", opts.Direction)
}
if opts.Since != nil {
q = q.Where("created_at >= ?", *opts.Since)
}
if opts.Until != nil {
q = q.Where("created_at <= ?", *opts.Until)
}
return rows, q.Find(&rows).Error
}
// CountBotDirectMessagesByConversation 返回会话总条数(前端无限滚动可用,可选)。
func (s *Store) CountBotDirectMessagesByConversation(opts BotDirectMessageListOptions) (int64, error) {
if s == nil || s.db == nil {
return 0, fmt.Errorf("Store is not configured")
}
if opts.BotID == 0 || opts.PeerNodeNum == 0 {
return 0, fmt.Errorf("bot id and peer node num are required")
}
var total int64
q := s.db.Model(&BotDirectMessageRecord{}).
Where("bot_id = ? AND peer_node_num = ?", opts.BotID, opts.PeerNodeNum)
if opts.Direction != "" {
q = q.Where("direction = ?", opts.Direction)
}
if opts.Since != nil {
q = q.Where("created_at >= ?", *opts.Since)
}
if opts.Until != nil {
q = q.Where("created_at <= ?", *opts.Until)
}
return total, q.Count(&total).Error
}
// FindBotForIncomingPKIPacket 在 bot_direct_messages 写入路径上判断接收方是否为受管 bot。
// 返回的 bot 用于填充 BotID/BotNodeID/BotNodeNum;不命中时返回 ErrRecordNotFound。
func (s *Store) FindBotForIncomingPKIPacket(toNodeNum int64) (*BotNodeRecord, error) {
if s == nil || s.db == nil {
return nil, fmt.Errorf("Store is not configured")
}
bot, err := s.GetBotNodeByNodeNum(toNodeNum)
if err != nil {
return nil, err
}
if !bot.Enabled {
return nil, errors.New("bot disabled")
}
return bot, nil
}
// BotDirectConversation 是 /admin/bot/direct 侧边栏需要的会话摘要。
// LastMessageAt / LastText / LastDirection 描述会话最后一条消息,便于按时间排序与预览;
// UnreadCount 仅对 inbound 计数(即未读消息数)。
type BotDirectConversation struct {
BotID uint64 `gorm:"column:bot_id"`
PeerNodeID string `gorm:"column:peer_node_id"`
PeerNodeNum int64 `gorm:"column:peer_node_num"`
LastMessageAt time.Time `gorm:"column:last_message_at"`
LastText string `gorm:"column:last_text"`
LastDirection string `gorm:"column:last_direction"`
UnreadCount int64 `gorm:"column:unread_count"`
TotalCount int64 `gorm:"column:total_count"`
}
// ListBotDirectConversations 聚合给定 bot 下的所有 (peer) 会话,返回最后一条消息及未读数。
// 按最后一条消息时间倒序(最新会话排前面)。limit/offset 走 ListOptions。
func (s *Store) ListBotDirectConversations(botID uint64, opts ListOptions) ([]BotDirectConversation, error) {
if s == nil || s.db == nil {
return nil, fmt.Errorf("Store is not configured")
}
if botID == 0 {
return nil, fmt.Errorf("bot id is required")
}
opts = NormalizeListOptions(opts)
var rows []BotDirectConversation
// 先把每对会话的最后一条消息 ID 取出来,再把这条消息的元数据 join 回去;
// 同时聚合 unread_countinbound 且 read_at IS NULL)和 total_count。
// 这样的两步 join 避免在 GROUP BY 后引用非聚合列(MySQL 严格模式 / SQLite 兼容)。
subLast := s.db.Model(&BotDirectMessageRecord{}).
Select("bot_id, peer_node_id, peer_node_num, MAX(id) AS last_id, COUNT(*) AS total_count, SUM(CASE WHEN direction = ? AND read_at IS NULL THEN 1 ELSE 0 END) AS unread_count", BotDirectMessageDirectionInbound).
Where("bot_id = ?", botID).
Group("bot_id, peer_node_id, peer_node_num")
q := s.db.Table("(?) AS agg", subLast).
Select("agg.bot_id AS bot_id, agg.peer_node_id AS peer_node_id, agg.peer_node_num AS peer_node_num, m.created_at AS last_message_at, m.text AS last_text, m.direction AS last_direction, agg.unread_count AS unread_count, agg.total_count AS total_count").
Joins("JOIN bot_direct_messages m ON m.id = agg.last_id").
Order("m.created_at DESC").
Order("m.id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Scan(&rows).Error
}
// MarkBotDirectMessagesRead 把 (bot, peer) 下未读的 inbound 消息全部标记为已读,返回更新行数。
func (s *Store) MarkBotDirectMessagesRead(botID uint64, peerNodeNum int64) (int64, error) {
if s == nil || s.db == nil {
return 0, fmt.Errorf("Store is not configured")
}
if botID == 0 || peerNodeNum == 0 {
return 0, fmt.Errorf("bot id and peer node num are required")
}
now := time.Now()
result := s.db.Model(&BotDirectMessageRecord{}).
Where("bot_id = ? AND peer_node_num = ? AND direction = ? AND read_at IS NULL", botID, peerNodeNum, BotDirectMessageDirectionInbound).
Update("read_at", &now)
if result.Error != nil {
return 0, result.Error
}
return result.RowsAffected, nil
}
// CountBotDirectUnread 返回某个 bot 全部未读 inbound 消息总数(用于头部小红点)。
func (s *Store) CountBotDirectUnread(botID uint64) (int64, error) {
if s == nil || s.db == nil {
return 0, fmt.Errorf("Store is not configured")
}
if botID == 0 {
return 0, fmt.Errorf("bot id is required")
}
var total int64
err := s.db.Model(&BotDirectMessageRecord{}).
Where("bot_id = ? AND direction = ? AND read_at IS NULL", botID, BotDirectMessageDirectionInbound).
Count(&total).Error
return total, err
}
// isInboundBotDirectMessage 判断 record 是否是“PKI 加密、发往受管 bot”的入向 DM。
// 仅在 type=text_message、pki_encrypted=true、packet_to_num 命中受管 bot 时返回 true。
// 任何步骤失败都返回 false,让记录回落到 text_message 表(与之前行为兼容)。
func isInboundBotDirectMessage(s *Store, record map[string]any) bool {
if s == nil || record == nil {
return false
}
if pki, _ := record["pki_encrypted"].(bool); !pki {
return false
}
toNum, ok := uint32FromRecord(record["packet_to_num"])
if !ok || toNum == 0 {
return false
}
bot, err := s.FindBotForIncomingPKIPacket(int64(toNum))
if err != nil || bot == nil {
return false
}
return true
}
// insertInboundBotDirectMessage 把一条入向 PKI DM 转写入 bot_direct_messages 表。
// 失败时返回错误,由 WriteQueue 统一打印 db_error 事件。
func insertInboundBotDirectMessage(s *Store, record map[string]any, clientInfo MQTTClientInfo) error {
if s == nil {
return fmt.Errorf("Store is not configured")
}
if record == nil {
return fmt.Errorf("record is required")
}
toNum, ok := uint32FromRecord(record["packet_to_num"])
if !ok || toNum == 0 {
return fmt.Errorf("missing packet_to_num")
}
bot, err := s.FindBotForIncomingPKIPacket(int64(toNum))
if err != nil {
return fmt.Errorf("lookup bot for inbound DM: %w", err)
}
peerNum, ok := uint32FromRecord(record["from_num"])
if !ok || peerNum == 0 {
return fmt.Errorf("missing from_num")
}
peerNodeID, _ := record["from"].(string)
if peerNodeID == "" {
return fmt.Errorf("missing from")
}
packetID, _ := uint32FromRecord(record["packet_id"])
topic, _ := record["topic"].(string)
gateway, _ := record["gateway_id"].(string)
var gatewayPtr *string
if gw := strings.TrimSpace(gateway); gw != "" {
gatewayPtr = &gw
}
text, _ := record["text"].(string)
wantAck, _ := record["want_ack"].(bool)
payloadLen, _ := record["payload_len"].(int)
if payloadLen == 0 {
if v, ok := record["payload_len"].(int64); ok {
payloadLen = int(v)
}
}
contentJSON, encodeErr := json.Marshal(record)
var contentPtr *string
if encodeErr == nil {
s := string(contentJSON)
contentPtr = &s
}
now := time.Now()
dm := &BotDirectMessageRecord{
BotID: bot.ID,
BotNodeID: bot.NodeID,
BotNodeNum: bot.NodeNum,
PeerNodeID: peerNodeID,
PeerNodeNum: int64(peerNum),
Direction: BotDirectMessageDirectionInbound,
Topic: topic,
PacketID: int64(packetID),
Text: text,
PayloadLen: int64(payloadLen),
PKIEncrypted: true,
WantAck: wantAck,
GatewayID: gatewayPtr,
Status: BotMessageStatusPublished,
ReceivedAt: &now,
ContentJSON: contentPtr,
}
if err := s.InsertBotDirectMessage(dm); err != nil {
return fmt.Errorf("insert bot direct message from %s: %w", peerNodeID, err)
}
_ = clientInfo // mqtt 元数据已经记录在 content_json 里,这里保留参数以保持队列签名一致
// 同时将消息添加到 LLM 队列(忽略机器人自己发送的消息)
if peerNodeID != bot.NodeID {
longName := NullableString(record["long_name"])
shortName := NullableString(record["short_name"])
channelID := NullableString(record["channel_id"])
_, err = s.EnqueueLLMMessage(LLMMessageQueueInput{
BotID: bot.ID,
BotNodeID: bot.NodeID,
BotNodeNum: bot.NodeNum,
FromNodeID: peerNodeID,
FromNodeNum: int64(peerNum),
LongName: longName,
ShortName: shortName,
Text: text,
PacketID: int64(packetID),
ChannelID: channelID,
Topic: topic,
MessageType: "direct",
ContentJSON: contentPtr,
})
}
if err != nil {
printJSON(map[string]any{
"event": "llm_queue_enqueue_failed",
"bot_id": bot.ID,
"from": peerNodeID,
"text": text,
"error": err.Error(),
})
}
return nil
}
+48
View File
@@ -0,0 +1,48 @@
package store
import (
"encoding/base64"
"encoding/hex"
"errors"
"strings"
"gorm.io/gorm"
)
// GetBotNodeByNodeNum 按节点号查找受管 bot 节点;用于 PKI 解密时把 to 字段映射回本地私钥。
func (s *Store) GetBotNodeByNodeNum(nodeNum int64) (*BotNodeRecord, error) {
if s == nil || s.db == nil {
return nil, errors.New("store not configured")
}
var row BotNodeRecord
if err := s.db.Where("node_num = ?", nodeNum).Take(&row).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, err
}
return nil, err
}
return &row, nil
}
// LookupNodeInfoPublicKey 在 nodeinfo 表中按 node_num 查 X25519 公钥,
// 兼容 hex 与 base64 两种历史存储格式。
func (s *Store) LookupNodeInfoPublicKey(nodeNum uint32) ([]byte, bool) {
var row NodeInfoRecord
if err := s.db.Where("node_num = ?", int64(nodeNum)).Take(&row).Error; err != nil {
return nil, false
}
if row.PublicKey == nil {
return nil, false
}
value := strings.TrimSpace(*row.PublicKey)
if value == "" {
return nil, false
}
if decoded, err := hex.DecodeString(value); err == nil && len(decoded) == 32 {
return decoded, true
}
if decoded, err := base64.StdEncoding.DecodeString(value); err == nil && len(decoded) == 32 {
return decoded, true
}
return nil, false
}
+373
View File
@@ -0,0 +1,373 @@
package store
import (
"crypto/ecdh"
"crypto/rand"
"encoding/base64"
"encoding/binary"
"errors"
"fmt"
"strings"
"time"
"unicode/utf8"
"meshtastic_mqtt_server/mqtpp"
"gorm.io/gorm"
)
const (
BotDefaultTopicPrefix = "msh/CN"
BotDefaultPSK = "AQ=="
BotDefaultNodeInfoBroadcastSeconds = int64(3600)
BotMessageTypeChannel = "channel"
BotMessageTypeDirect = "direct"
BotMessageStatusPending = "pending"
BotMessageStatusPublished = "published"
BotMessageStatusFailed = "failed"
)
var ErrBotNodeAlreadyExists = errors.New("bot node already exists")
type BotNodeInput struct {
NodeNum *int64
LongName string
ShortName string
Enabled bool
DefaultChannelID string
TopicPrefix string
PSK string
NodeInfoBroadcastEnabled bool
NodeInfoBroadcastIntervalSeconds int64
LLMQueueEnabled bool
LLMIncludeChannelMessages bool
}
type BotMessageListOptions struct {
ListOptions
BotID uint64
MessageType string
ChannelID string
}
func (s *Store) ListBotNodes(opts ListOptions) ([]BotNodeRecord, error) {
opts = NormalizeListOptions(opts)
var rows []BotNodeRecord
q := s.db.Model(&BotNodeRecord{}).
Order("updated_at DESC").
Order("id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
func (s *Store) CountBotNodes(opts ListOptions) (int64, error) {
var total int64
return total, s.db.Model(&BotNodeRecord{}).Count(&total).Error
}
func (s *Store) GetBotNode(id uint64) (*BotNodeRecord, error) {
var row BotNodeRecord
if err := s.db.Where("id = ?", id).Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) CreateBotNode(input BotNodeInput) (*BotNodeRecord, error) {
row, err := s.normalizedBotNodeRecord(input)
if err != nil {
return nil, err
}
if err := s.ensureBotNodeUnique(0, row.NodeID, row.NodeNum); err != nil {
return nil, err
}
if err := s.ensureBotNodeDoesNotConflictWithNodeInfo(row.NodeNum, row.NodeID); err != nil {
return nil, err
}
if err := populateBotNodeKeys(row); err != nil {
return nil, err
}
if err := s.db.Create(row).Error; err != nil {
return nil, err
}
return row, nil
}
func (s *Store) UpdateBotNode(id uint64, input BotNodeInput) (*BotNodeRecord, error) {
if id == 0 {
return nil, fmt.Errorf("bot node id is required")
}
existing, err := s.GetBotNode(id)
if err != nil {
return nil, err
}
row, err := s.normalizedBotNodeRecord(input)
if err != nil {
return nil, err
}
if err := s.ensureBotNodeUnique(id, row.NodeID, row.NodeNum); err != nil {
return nil, err
}
// 只有当 node_num 真的发生变化时,才需要校验和 nodeinfo 表的冲突。
// 否则机器人自己广播 NodeInfo 回写到 nodeinfo 表后,UpdateBotNode 会把这条
// 自己的记录当成外部节点冲突,导致 “already exists or conflicts” 报错。
if row.NodeNum != existing.NodeNum {
if err := s.ensureBotNodeDoesNotConflictWithNodeInfo(row.NodeNum, row.NodeID); err != nil {
return nil, err
}
}
updates := map[string]any{
"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,
"nodeinfo_broadcast_enabled": row.NodeInfoBroadcastEnabled,
"nodeinfo_broadcast_interval_seconds": row.NodeInfoBroadcastIntervalSeconds,
"llm_queue_enabled": row.LLMQueueEnabled,
"llm_include_channel_messages": row.LLMIncludeChannelMessages,
"updated_at": time.Now(),
}
if err := s.db.Model(&BotNodeRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return nil, err
}
return s.GetBotNode(id)
}
func (s *Store) DeleteBotNode(id uint64) error {
result := s.db.Where("id = ?", id).Delete(&BotNodeRecord{})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
}
func (s *Store) InsertBotMessage(row *BotMessageRecord) error {
return s.db.Create(row).Error
}
func (s *Store) UpdateBotMessageStatus(id uint64, status, errText string, publishedAt *time.Time) error {
updates := map[string]any{"status": status, "error": strings.TrimSpace(errText), "published_at": publishedAt}
result := s.db.Model(&BotMessageRecord{}).Where("id = ?", id).Updates(updates)
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
}
func (s *Store) UpdateBotNodeInfoBroadcastAt(id uint64, t time.Time) error {
result := s.db.Model(&BotNodeRecord{}).Where("id = ?", id).Updates(map[string]any{"last_nodeinfo_broadcast_at": &t, "updated_at": time.Now()})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
}
func (s *Store) RegenerateBotNodeKeys(id uint64) (*BotNodeRecord, error) {
if id == 0 {
return nil, fmt.Errorf("bot node id is required")
}
row, err := s.GetBotNode(id)
if err != nil {
return nil, err
}
if err := populateBotNodeKeys(row); err != nil {
return nil, err
}
updates := map[string]any{"public_key": row.PublicKey, "private_key": row.PrivateKey, "updated_at": time.Now()}
if err := s.db.Model(&BotNodeRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return nil, err
}
return s.GetBotNode(id)
}
func (s *Store) ListBotMessages(opts BotMessageListOptions) ([]BotMessageRecord, error) {
opts.ListOptions = NormalizeListOptions(opts.ListOptions)
var rows []BotMessageRecord
q := applyBotMessageFilters(s.db.Model(&BotMessageRecord{}), opts).
Order("created_at DESC").
Order("id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
func (s *Store) CountBotMessages(opts BotMessageListOptions) (int64, error) {
var total int64
q := applyBotMessageFilters(s.db.Model(&BotMessageRecord{}), opts)
return total, q.Count(&total).Error
}
func applyBotMessageFilters(q *gorm.DB, opts BotMessageListOptions) *gorm.DB {
if opts.BotID != 0 {
q = q.Where("bot_id = ?", opts.BotID)
}
if opts.MessageType != "" {
q = q.Where("message_type = ?", opts.MessageType)
}
if opts.ChannelID != "" {
q = q.Where("channel_id = ?", opts.ChannelID)
}
if opts.Since != nil {
q = q.Where("created_at >= ?", *opts.Since)
}
if opts.Until != nil {
q = q.Where("created_at <= ?", *opts.Until)
}
return q
}
func (s *Store) normalizedBotNodeRecord(input BotNodeInput) (*BotNodeRecord, error) {
longName := strings.TrimSpace(input.LongName)
shortName := strings.TrimSpace(input.ShortName)
channelID := strings.TrimSpace(input.DefaultChannelID)
psk := strings.TrimSpace(input.PSK)
if psk == "" {
psk = BotDefaultPSK
}
if _, err := mqtpp.ExpandPSK(psk); err != nil {
return nil, err
}
topicPrefix := strings.Trim(strings.TrimSpace(input.TopicPrefix), "/")
if topicPrefix == "" {
topicPrefix = BotDefaultTopicPrefix
}
if longName == "" {
return nil, fmt.Errorf("long name is required")
}
if !utf8.ValidString(longName) {
return nil, fmt.Errorf("long name must be valid utf-8")
}
if shortName == "" {
return nil, fmt.Errorf("short name is required")
}
if !utf8.ValidString(shortName) {
return nil, fmt.Errorf("short name must be valid utf-8")
}
if channelID == "" {
return nil, fmt.Errorf("default channel id is required")
}
interval := input.NodeInfoBroadcastIntervalSeconds
if interval <= 0 {
interval = BotDefaultNodeInfoBroadcastSeconds
}
if interval < 60 {
return nil, fmt.Errorf("nodeinfo broadcast interval must be at least 60 seconds")
}
var nodeNum int64
if input.NodeNum == nil || *input.NodeNum == 0 {
generated, err := s.generateBotNodeNum()
if err != nil {
return nil, err
}
nodeNum = generated
} else {
nodeNum = *input.NodeNum
}
if err := ValidateBotNodeNum(nodeNum); err != nil {
return nil, err
}
return &BotNodeRecord{NodeID: mqtpp.NodeNumToID(uint32(nodeNum)), NodeNum: nodeNum, LongName: longName, ShortName: shortName, Enabled: input.Enabled, DefaultChannelID: channelID, TopicPrefix: topicPrefix, PSK: psk, NodeInfoBroadcastEnabled: input.NodeInfoBroadcastEnabled, NodeInfoBroadcastIntervalSeconds: interval, LLMQueueEnabled: input.LLMQueueEnabled, LLMIncludeChannelMessages: input.LLMIncludeChannelMessages}, nil
}
func populateBotNodeKeys(row *BotNodeRecord) error {
privateKey, err := ecdh.X25519().GenerateKey(rand.Reader)
if err != nil {
return err
}
row.PrivateKey = base64.StdEncoding.EncodeToString(privateKey.Bytes())
row.PublicKey = base64.StdEncoding.EncodeToString(privateKey.PublicKey().Bytes())
return nil
}
func DecodeBotPublicKey(row BotNodeRecord) ([]byte, error) {
if strings.TrimSpace(row.PublicKey) == "" {
return nil, nil
}
key, err := base64.StdEncoding.DecodeString(row.PublicKey)
if err != nil {
return nil, fmt.Errorf("invalid bot public key: %w", err)
}
return key, nil
}
func ValidateBotNodeNum(nodeNum int64) error {
if nodeNum <= 0 || nodeNum >= int64(mqtpp.NodeNumBroadcast) {
return fmt.Errorf("node num must be between 1 and 4294967294")
}
return nil
}
func (s *Store) generateBotNodeNum() (int64, error) {
for i := 0; i < 32; i++ {
var buf [4]byte
if _, err := rand.Read(buf[:]); err != nil {
return 0, err
}
nodeNum := int64(binary.LittleEndian.Uint32(buf[:]) & 0x7fffffff)
if err := ValidateBotNodeNum(nodeNum); err != nil {
continue
}
if err := s.ensureBotNodeUnique(0, mqtpp.NodeNumToID(uint32(nodeNum)), nodeNum); err != nil {
if errors.Is(err, ErrBotNodeAlreadyExists) {
continue
}
return 0, err
}
if err := s.ensureBotNodeDoesNotConflictWithNodeInfo(nodeNum, mqtpp.NodeNumToID(uint32(nodeNum))); err != nil {
if errors.Is(err, ErrBotNodeAlreadyExists) {
continue
}
return 0, err
}
return nodeNum, nil
}
return 0, fmt.Errorf("generate bot node num failed")
}
func (s *Store) ensureBotNodeUnique(id uint64, nodeID string, nodeNum int64) error {
var existing BotNodeRecord
q := s.db.Where("node_id = ? OR node_num = ?", nodeID, nodeNum)
if id != 0 {
q = q.Where("id <> ?", id)
}
err := q.Take(&existing).Error
if err == nil {
return ErrBotNodeAlreadyExists
}
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
return err
}
func (s *Store) ensureBotNodeDoesNotConflictWithNodeInfo(nodeNum int64, selfNodeID string) error {
var existing NodeInfoRecord
q := s.db.Where("node_num = ?", nodeNum)
if selfNodeID != "" {
// 机器人自己广播 NodeInfo 后会以同样的 node_id/node_num 回写 nodeinfo
// 把这条自身记录从冲突检测中排除,避免把自己当成外部节点。
q = q.Where("node_id <> ?", selfNodeID)
}
err := q.Take(&existing).Error
if err == nil {
return ErrBotNodeAlreadyExists
}
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
return err
}
+1436
View File
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+135
View File
@@ -0,0 +1,135 @@
package store
import "sync"
type WriteQueue struct {
store *Store
jobs chan writeJob
wg sync.WaitGroup
}
type writeJob struct {
typeName string
from any
run func() error
errorEvent map[string]any
}
func NewWriteQueue(s *Store) *WriteQueue {
if s == nil {
return nil
}
q := &WriteQueue{
store: s,
jobs: make(chan writeJob, 1024),
}
q.wg.Add(1)
go q.run()
return q
}
func (q *WriteQueue) EnqueueRecord(record map[string]any, clientInfo MQTTClientInfo) {
if q == nil {
return
}
record = cloneDBWriteRecord(record)
switch record["type"] {
case "nodeinfo":
q.enqueue(writeJob{typeName: "nodeinfo", from: record["from"], run: func() error {
return q.store.UpsertNodeInfo(record)
}})
case "map_report":
q.enqueue(writeJob{typeName: "map_report", from: record["from"], run: func() error {
return q.store.UpsertMapReport(record)
}})
case "text_message":
// 私聊(PKI 加密、发往受管 bot)单独走 bot_direct_messages 表,
// 不再写入 text_message 以避免和频道消息混在一起。
if isInboundBotDirectMessage(q.store, record) {
q.enqueue(writeJob{typeName: "bot_direct_message_inbound", from: record["from"], run: func() error {
return insertInboundBotDirectMessage(q.store, record, clientInfo)
}})
return
}
// 频道消息同时也写入 LLM 队列(如果启用的话)
q.enqueue(writeJob{typeName: "llm_channel_message", from: record["from"], run: func() error {
return enqueueChannelMessageToLLM(q.store, record)
}})
q.enqueue(writeJob{typeName: "text_message", from: record["from"], run: func() error {
return q.store.InsertTextMessage(record, clientInfo)
}})
case "position":
q.enqueue(writeJob{typeName: "position", from: record["from"], run: func() error {
return q.store.InsertPosition(record, clientInfo)
}})
case "telemetry":
q.enqueue(writeJob{typeName: "telemetry", from: record["from"], run: func() error {
return q.store.InsertTelemetry(record, clientInfo)
}})
case "routing":
q.enqueue(writeJob{typeName: "routing", from: record["from"], run: func() error {
return q.store.InsertRouting(record, clientInfo)
}})
case "traceroute":
q.enqueue(writeJob{typeName: "traceroute", from: record["from"], run: func() error {
return q.store.InsertTraceroute(record, clientInfo)
}})
}
}
func (q *WriteQueue) EnqueueDiscard(record map[string]any, raw []byte, clientInfo MQTTClientInfo) {
if q == nil {
return
}
record = cloneDBWriteRecord(record)
raw = append([]byte(nil), raw...)
q.enqueue(writeJob{typeName: "discard_details", from: record["from"], errorEvent: map[string]any{"event": "db_error", "type": "discard_details", "topic": record["topic"]}, run: func() error {
return q.store.InsertDiscardDetails(record, raw, clientInfo)
}})
}
func (q *WriteQueue) Close() {
if q == nil {
return
}
close(q.jobs)
q.wg.Wait()
}
func (q *WriteQueue) Len() int {
if q == nil {
return 0
}
return len(q.jobs)
}
func (q *WriteQueue) enqueue(job writeJob) {
q.jobs <- job
}
func (q *WriteQueue) run() {
defer q.wg.Done()
for job := range q.jobs {
if err := job.run(); err != nil {
event := job.errorEvent
if event == nil {
event = map[string]any{"event": "db_error", "type": job.typeName, "from": job.from}
} else {
event = cloneDBWriteRecord(event)
}
event["error"] = err.Error()
printJSON(event)
}
}
}
func cloneDBWriteRecord(record map[string]any) map[string]any {
if record == nil {
return nil
}
cloned := make(map[string]any, len(record))
for key, value := range record {
cloned[key] = value
}
return cloned
}
+104
View File
@@ -0,0 +1,104 @@
package store
import (
"database/sql"
"testing"
)
func TestDBWriteQueueWritesRecordsAsync(t *testing.T) {
st := openTestStore(t)
defer st.Close()
queue := newDBWriteQueue(st)
record := textMessageTestRecord("queued")
queue.EnqueueRecord(record, MQTTClientInfo{ClientID: "client-1"})
record["text"] = "mutated after enqueue"
queue.Close()
var text, clientID string
if err := rawTestDB(t, st).QueryRow("SELECT text, mqtt_client_id FROM text_message WHERE from_id = ?", "!12345678").Scan(&text, &clientID); err != nil {
t.Fatal(err)
}
if text != "queued" || clientID != "client-1" {
t.Fatalf("queued row = text %q client %q, want queued/client-1", text, clientID)
}
}
func TestDBWriteQueueWritesDiscardAsync(t *testing.T) {
st := openTestStore(t)
defer st.Close()
queue := newDBWriteQueue(st)
record := map[string]any{"topic": "msh/test", "error": "bad packet"}
queue.EnqueueDiscard(record, []byte{1, 2, 3}, MQTTClientInfo{RemoteAddr: "127.0.0.1:1883"})
record["error"] = "mutated after enqueue"
queue.Close()
var topic, reason, rawBase64, remoteAddr string
if err := rawTestDB(t, st).QueryRow("SELECT topic, error, raw_base64, mqtt_remote_addr FROM discard_details").Scan(&topic, &reason, &rawBase64, &remoteAddr); err != nil {
t.Fatal(err)
}
if topic != "msh/test" || reason != "bad packet" || rawBase64 != "AQID" || remoteAddr != "127.0.0.1:1883" {
t.Fatalf("discard row = %q/%q/%q/%q, want queued values", topic, reason, rawBase64, remoteAddr)
}
}
func TestDBWriteQueueLen(t *testing.T) {
queue := &WriteQueue{jobs: make(chan writeJob, 1)}
queue.enqueue(writeJob{run: func() error { return nil }})
if queue.Len() != 1 {
t.Fatalf("queue.Len() = %d, want 1", queue.Len())
}
}
func TestDBWriteQueueIgnoresUnsupportedRecordType(t *testing.T) {
st := openTestStore(t)
defer st.Close()
queue := newDBWriteQueue(st)
queue.EnqueueRecord(map[string]any{"type": "empty_packet", "from": "!12345678"}, MQTTClientInfo{})
queue.Close()
var count int
if err := rawTestDB(t, st).QueryRow("SELECT COUNT(*) FROM text_message").Scan(&count); err != nil {
t.Fatal(err)
}
if count != 0 {
t.Fatalf("text_message count = %d, want 0", count)
}
}
func TestDBWriteQueueNilStore(t *testing.T) {
if queue := newDBWriteQueue(nil); queue != nil {
t.Fatalf("newDBWriteQueue(nil) = %#v, want nil", queue)
}
var queue *WriteQueue
queue.EnqueueRecord(textMessageTestRecord("ignored"), MQTTClientInfo{})
queue.EnqueueDiscard(map[string]any{"topic": "ignored"}, []byte{1}, MQTTClientInfo{})
queue.Close()
}
func TestDBWriteQueueRecordValidationErrorDoesNotStopWorker(t *testing.T) {
st := openTestStore(t)
defer st.Close()
queue := newDBWriteQueue(st)
badRecord := textMessageTestRecord("bad")
delete(badRecord, "from")
queue.EnqueueRecord(badRecord, MQTTClientInfo{})
queue.EnqueueRecord(textMessageTestRecord("good"), MQTTClientInfo{})
queue.Close()
var text string
if err := rawTestDB(t, st).QueryRow("SELECT text FROM text_message").Scan(&text); err != nil {
t.Fatal(err)
}
if text != "good" {
t.Fatalf("text = %q, want good", text)
}
var missing sql.NullString
if err := rawTestDB(t, st).QueryRow("SELECT text FROM text_message WHERE text = ?", "bad").Scan(&missing); err != sql.ErrNoRows {
t.Fatalf("bad row error = %v, want sql.ErrNoRows", err)
}
}
+45
View File
@@ -0,0 +1,45 @@
package store
import (
"encoding/base64"
"encoding/json"
"fmt"
)
func (s *Store) InsertDiscardDetails(record map[string]any, raw []byte, clientInfo MQTTClientInfo) error {
details, err := discardDetailsFromRecord(record, raw, clientInfo)
if err != nil {
return err
}
if err := s.db.Create(details).Error; err != nil {
return fmt.Errorf("insert discard_details: %w", err)
}
return nil
}
func discardDetailsFromRecord(record map[string]any, raw []byte, clientInfo MQTTClientInfo) (*DiscardDetailsRecord, error) {
contentJSON, err := json.Marshal(record)
if err != nil {
return nil, fmt.Errorf("encode discard_details content_json: %w", err)
}
return &DiscardDetailsRecord{
Topic: stringValue(record["topic"]),
Error: stringValue(record["error"]),
PayloadLen: int64(len(raw)),
RawBase64: base64.StdEncoding.EncodeToString(raw),
ContentJSON: string(contentJSON),
MQTTClientID: NullableStringValue(clientInfo.ClientID),
MQTTUsername: NullableStringValue(clientInfo.Username),
MQTTListener: NullableStringValue(clientInfo.Listener),
MQTTRemoteAddr: NullableStringValue(clientInfo.RemoteAddr),
MQTTRemoteHost: NullableStringValue(clientInfo.RemoteHost),
MQTTRemotePort: NullableStringValue(clientInfo.RemotePort),
}, nil
}
func stringValue(value any) string {
if s := NullableStringValue(value); s != nil {
return *s
}
return ""
}
+54
View File
@@ -0,0 +1,54 @@
package store
import (
"fmt"
"strings"
)
const maxHelpMarkdownBytes = 200 * 1024
const DefaultHelpMarkdown = `## 连接地址
将 Meshtastic 设备连接到本服务提供的 MQTT broker。
- 默认地址:**mesh.gat-iot.com**
- 默认端口:**1883**
- 用户名称:**meshdev**
- 密码:**large4cats**
## 频道加密要求
为了让服务能够解析 Meshtastic MQTT payload,频道需要满足以下任一条件:
- 频道不加密。
- 使用 Meshtastic 默认 PSK**AQ==**。
如果使用自定义加密密钥,数据可能会被判定为无法解密并丢弃。
## 反馈问题
如果遇到 bug,请在 GitHub [提交 issue](https://github.com/wuwenfengmi1998/meshtastic_mqtt_server),或联系邮箱 [kevin@lmve.net](mailto:kevin@lmve.net)。`
func (s *Store) GetLatestHelpContent() (*HelpContentRecord, error) {
var row HelpContentRecord
if err := s.db.Order("id DESC").Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) InsertHelpContent(markdown, createdBy string) (*HelpContentRecord, error) {
markdown = strings.TrimSpace(markdown)
createdBy = strings.TrimSpace(createdBy)
if markdown == "" {
return nil, fmt.Errorf("markdown is required")
}
if len([]byte(markdown)) > maxHelpMarkdownBytes {
return nil, fmt.Errorf("markdown exceeds %d bytes", maxHelpMarkdownBytes)
}
row := HelpContentRecord{Markdown: markdown, CreatedBy: createdBy}
if err := s.db.Create(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
+45
View File
@@ -0,0 +1,45 @@
package store
import "golang.org/x/crypto/bcrypt"
// AdminRole 是管理员账号在用户表里的角色字符串。其它包通过这个常量与
// `users.role` 字段对齐,避免硬编码。
const AdminRole = "admin"
// printJSON 是 store 包内部的诊断输出钩子。当前实现为 noop——保持与
// 重构前 main.go 的行为一致;如需启用调试,可在调用方替换。
func printJSON(record map[string]any) {
_ = record
}
// hashPassword 与 auth.go 中的散列实现保持一致(bcrypt 默认 cost)。
func hashPassword(password string) (string, error) {
hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
if err != nil {
return "", err
}
return string(hash), nil
}
// uint32FromRecord 把 map[string]any 中的整型字段安全转换为 uint32。
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
}
+522
View File
@@ -0,0 +1,522 @@
package store
import (
"encoding/json"
"errors"
"fmt"
"time"
"gorm.io/gorm"
)
// ============================================
// LLM Provider (llm_providers) - 多 AI API 配置
// ============================================
// ListLLMProviders 列出所有 LLM Provider
func (s *Store) ListLLMProviders(includeInactive bool) ([]LLMProviderRecord, error) {
var rows []LLMProviderRecord
query := s.db.Model(&LLMProviderRecord{})
if !includeInactive {
query = query.Where("active = ?", true)
}
if err := query.Order("created_at DESC").Find(&rows).Error; err != nil {
return nil, fmt.Errorf("list llm providers: %w", err)
}
return rows, nil
}
// GetLLMProvider 获取单个 LLM Provider
func (s *Store) GetLLMProvider(name string) (*LLMProviderRecord, error) {
var record LLMProviderRecord
if err := s.db.Where("name = ?", name).Take(&record).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, err
}
return nil, fmt.Errorf("get llm provider %s: %w", name, err)
}
return &record, nil
}
// CreateLLMProvider 创建 LLM Provider
func (s *Store) CreateLLMProvider(record *LLMProviderRecord) error {
if err := s.db.Create(record).Error; err != nil {
return fmt.Errorf("create llm provider %s: %w", record.Name, err)
}
return nil
}
// UpdateLLMProvider 更新 LLM Provider
func (s *Store) UpdateLLMProvider(name string, updates map[string]any) error {
if err := s.db.Model(&LLMProviderRecord{}).Where("name = ?", name).Updates(updates).Error; err != nil {
return fmt.Errorf("update llm provider %s: %w", name, err)
}
return nil
}
// DeleteLLMProvider 删除 LLM Provider
func (s *Store) DeleteLLMProvider(name string) error {
if err := s.db.Where("name = ?", name).Delete(&LLMProviderRecord{}).Error; err != nil {
return fmt.Errorf("delete llm provider %s: %w", name, err)
}
return nil
}
// EnsureDefaultLLMProvider 确保存在默认 LLM Provider 配置
// 只有当数据库中完全没有任何 provider 配置时,才创建默认配置
func (s *Store) EnsureDefaultLLMProvider() error {
// 先检查是否已经有任何 provider 配置
providers, err := s.ListLLMProviders(true)
if err != nil {
return fmt.Errorf("list llm providers: %w", err)
}
if len(providers) > 0 {
return nil // 已有配置,不创建默认
}
// 创建默认配置
defaultConfig := &LLMProviderRecord{
Name: "default",
Active: true,
APIKey: "",
BaseURL: "https://ark.cn-beijing.volces.com/api/v3",
Model: "",
Timeout: 120,
ContextWindowTokens: 262144,
}
return s.CreateLLMProvider(defaultConfig)
}
// ============================================
// LLM Tool Router (llm_tool_router) - 工具路由配置
// ============================================
// GetLLMToolRouter 获取当前激活的 Tool Router 配置
func (s *Store) GetLLMToolRouter() (*LLMToolRouterRecord, error) {
var record LLMToolRouterRecord
// 默认取第一条记录(ID 最小的),因为通常只需要一个配置
if err := s.db.Order("id ASC").First(&record).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, err
}
return nil, fmt.Errorf("get llm tool router: %w", err)
}
return &record, nil
}
// CreateLLMToolRouter 创建 Tool Router 配置
func (s *Store) CreateLLMToolRouter(record *LLMToolRouterRecord) error {
if err := s.db.Create(record).Error; err != nil {
return fmt.Errorf("create llm tool router: %w", err)
}
return nil
}
// UpdateLLMToolRouter 更新 Tool Router 配置
func (s *Store) UpdateLLMToolRouter(id uint64, updates map[string]any) error {
if err := s.db.Model(&LLMToolRouterRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return fmt.Errorf("update llm tool router %d: %w", id, err)
}
return nil
}
// EnsureDefaultLLMToolRouter 确保存在默认 Tool Router 配置
func (s *Store) EnsureDefaultLLMToolRouter() error {
_, err := s.GetLLMToolRouter()
if err == nil {
return nil // 已存在
}
if !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
// 创建默认配置
defaultConfig := &LLMToolRouterRecord{
Enabled: true,
OpenAIName: "",
Timeout: 30,
MaxTokens: 512,
SystemPrompt: "你可以按需直接调用可用工具来回答用户问题。\n每个工具的 description 描述了它的适用场景和调用条件。\n工具结果优先于模型内置知识;工具失败时必须如实说明,不要编造结果。\n只调用确实必要的工具。",
}
return s.CreateLLMToolRouter(defaultConfig)
}
// ============================================
// LLM Primary Config (llm_primary_config) - 主 AI 回复配置
// ============================================
// GetLLMPrimaryConfig 获取当前激活的主 AI 回复配置
func (s *Store) GetLLMPrimaryConfig() (*LLMPrimaryConfigRecord, error) {
var record LLMPrimaryConfigRecord
// 默认取第一条记录(ID 最小的),因为通常只需要一个配置
if err := s.db.Order("id ASC").First(&record).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, err
}
return nil, fmt.Errorf("get llm primary config: %w", err)
}
return &record, nil
}
// GetLLMPrimaryConfigSystemPrompt 获取主 AI 回复配置中的系统提示词
// 如果没有配置或出错,返回空字符串(autoreply service 会处理这种情况)
func (s *Store) GetLLMPrimaryConfigSystemPrompt() (string, error) {
record, err := s.GetLLMPrimaryConfig()
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
// 没有配置时返回空字符串,使用默认行为
return "", nil
}
return "", err
}
return record.SystemPrompt, nil
}
// GetLLMPrimaryConfigEnableTool 获取是否启用工具调用
func (s *Store) GetLLMPrimaryConfigEnableTool() (bool, error) {
record, err := s.GetLLMPrimaryConfig()
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return false, nil
}
return false, err
}
return record.EnableTool, nil
}
// CreateLLMPrimaryConfig 创建主 AI 回复配置
func (s *Store) CreateLLMPrimaryConfig(record *LLMPrimaryConfigRecord) error {
if err := s.db.Create(record).Error; err != nil {
return fmt.Errorf("create llm primary config: %w", err)
}
return nil
}
// UpdateLLMPrimaryConfig 更新主 AI 回复配置
func (s *Store) UpdateLLMPrimaryConfig(id uint64, updates map[string]any) error {
if err := s.db.Model(&LLMPrimaryConfigRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return fmt.Errorf("update llm primary config %d: %w", id, err)
}
return nil
}
// EnsureDefaultLLMPrimaryConfig 确保存在默认主 AI 回复配置
func (s *Store) EnsureDefaultLLMPrimaryConfig() error {
_, err := s.GetLLMPrimaryConfig()
if err == nil {
return nil // 已存在
}
if !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
// 创建默认配置
defaultConfig := &LLMPrimaryConfigRecord{
Enabled: false,
ProviderName: "",
Timeout: 120,
MaxTokens: 1024,
SystemPrompt: "你是一个 Meshtastic 网络助手。请简洁回答用户问题。\n回答要简短清晰,适合在低带宽无线电环境传输。每次回复限制在200 bytes以内。",
EnableTool: false,
}
return s.CreateLLMPrimaryConfig(defaultConfig)
}
// LLMMessageQueueInput 是添加 LLM 队列消息的输入
type LLMMessageQueueInput struct {
BotID uint64 // 0 表示频道消息
BotNodeID string // 频道消息可为空
BotNodeNum int64 // 频道消息可为 0
FromNodeID string
FromNodeNum int64
LongName *string
ShortName *string
Text string
PacketID int64
ChannelID *string
Topic string
MessageType string // "channel" 或 "direct"
ContentJSON *string
}
// EnqueueLLMMessage 将消息添加到 LLM 队列
func (s *Store) EnqueueLLMMessage(input LLMMessageQueueInput) (*LLMMessageQueueRecord, error) {
var err error
if input.BotID == 0 {
return nil, nil // bot_id 为 0 的消息不再入队
}
// 检查机器人级别的 LLM 队列设置
bot, err := s.GetBotNode(input.BotID)
if err != nil {
return nil, nil // 机器人不存在,静默返回
}
if !bot.LLMQueueEnabled {
return nil, nil // 机器人的 LLM 队列未启用,静默返回
}
// 忽略机器人自己发送的消息,避免自循环
if input.FromNodeID == bot.NodeID {
return nil, nil
}
if input.FromNodeID == "" {
return nil, fmt.Errorf("from_node_id is required")
}
if input.Text == "" {
return nil, fmt.Errorf("text is required")
}
// 检查是否存在重复消息
// packet_id > 0: 用 bot_id + packet_id 去重(频道消息)
// packet_id = 0: 用 bot_id + from_node_id + text 去重(私聊消息,可能没有 packet_id)
// 只排除 pending/processing 状态的消息,允许 error 状态的消息重新入队
var existing LLMMessageQueueRecord
if input.PacketID > 0 {
// 频道消息:用 bot_id + packet_id 去重
err = s.db.Where("bot_id = ? AND packet_id = ? AND deleted_at IS NULL AND status IN (?, ?)",
input.BotID, input.PacketID, LLMMessageStatusPending, LLMMessageStatusProcessing).
Take(&existing).Error
} else {
// 私聊消息:用 bot_id + from_node_id + text 去重(避免同一人连续发相同内容被拒绝)
err = s.db.Where("bot_id = ? AND from_node_id = ? AND text = ? AND deleted_at IS NULL AND status IN (?, ?)",
input.BotID, input.FromNodeID, input.Text, LLMMessageStatusPending, LLMMessageStatusProcessing).
Take(&existing).Error
}
if err == nil {
// 存在正在处理或待处理的相同消息,直接返回
return &existing, nil
}
if !errors.Is(err, gorm.ErrRecordNotFound) {
return nil, fmt.Errorf("check duplicate llm message: %w", err)
}
now := time.Now()
messageType := input.MessageType
if messageType == "" {
messageType = "direct"
}
record := &LLMMessageQueueRecord{
BotID: input.BotID,
BotNodeID: input.BotNodeID,
BotNodeNum: input.BotNodeNum,
FromNodeID: input.FromNodeID,
FromNodeNum: input.FromNodeNum,
LongName: input.LongName,
ShortName: input.ShortName,
Text: input.Text,
PacketID: input.PacketID,
ChannelID: input.ChannelID,
Topic: input.Topic,
MessageType: messageType,
Status: LLMMessageStatusPending,
ReceivedAt: now,
ContentJSON: input.ContentJSON,
}
if err := s.db.Create(record).Error; err != nil {
return nil, fmt.Errorf("enqueue llm message: %w", err)
}
return record, nil
}
// ListLLMMessages 列出 LLM 队列消息
func (s *Store) ListLLMMessages(opts ListOptions, botID uint64, includeDeleted bool) ([]LLMMessageQueueRecord, int64, error) {
var rows []LLMMessageQueueRecord
query := s.db.Model(&LLMMessageQueueRecord{})
if botID > 0 {
query = query.Where("bot_id = ?", botID)
}
if !includeDeleted {
query = query.Where("deleted_at IS NULL")
}
// 先获取总数
var total int64
if err := query.Count(&total).Error; err != nil {
return nil, 0, fmt.Errorf("count llm messages: %w", err)
}
// 排序和分页
query = query.Order("created_at DESC")
if opts.Limit > 0 {
query = query.Limit(opts.Limit)
}
if opts.Offset > 0 {
query = query.Offset(opts.Offset)
}
if err := query.Find(&rows).Error; err != nil {
return nil, 0, fmt.Errorf("list llm messages: %w", err)
}
return rows, total, nil
}
// GetLLMMessage 获取单条 LLM 消息
func (s *Store) GetLLMMessage(id uint64) (*LLMMessageQueueRecord, error) {
var record LLMMessageQueueRecord
if err := s.db.Where("id = ?", id).Take(&record).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, err
}
return nil, fmt.Errorf("get llm message %d: %w", id, err)
}
return &record, nil
}
// UpdateLLMMessageStatus 更新 LLM 消息状态
func (s *Store) UpdateLLMMessageStatus(id uint64, status string, errorMsg string) error {
updates := map[string]any{
"status": status,
"error": errorMsg,
}
if status == LLMMessageStatusProcessed {
now := time.Now()
updates["processed_at"] = &now
}
if err := s.db.Model(&LLMMessageQueueRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return fmt.Errorf("update llm message status %d: %w", id, err)
}
return nil
}
// SoftDeleteLLMMessage 软删除 LLM 消息
func (s *Store) SoftDeleteLLMMessage(id uint64) error {
now := time.Now()
if err := s.db.Model(&LLMMessageQueueRecord{}).Where("id = ?", id).Update("deleted_at", &now).Error; err != nil {
return fmt.Errorf("soft delete llm message %d: %w", id, err)
}
return nil
}
// SoftDeleteLLMMessagesByBot 软删除指定机器人的所有消息
func (s *Store) SoftDeleteLLMMessagesByBot(botID uint64) error {
now := time.Now()
if err := s.db.Model(&LLMMessageQueueRecord{}).Where("bot_id = ? AND deleted_at IS NULL", botID).Update("deleted_at", &now).Error; err != nil {
return fmt.Errorf("soft delete llm messages for bot %d: %w", botID, err)
}
return nil
}
// CleanupDeletedLLMMessages 清理已软删除超过指定时间的消息
func (s *Store) CleanupDeletedLLMMessages(before time.Time) (int64, error) {
result := s.db.Where("deleted_at IS NOT NULL AND deleted_at < ?", before).Delete(&LLMMessageQueueRecord{})
if result.Error != nil {
return 0, fmt.Errorf("cleanup deleted llm messages: %w", result.Error)
}
return result.RowsAffected, nil
}
// LLMMessageDTO 将数据库记录转换为 API 响应格式
func LLMMessageDTO(row LLMMessageQueueRecord) map[string]any {
return map[string]any{
"id": row.ID,
"bot_id": row.BotID,
"bot_node_id": row.BotNodeID,
"bot_node_num": row.BotNodeNum,
"from_node_id": row.FromNodeID,
"from_node_num": row.FromNodeNum,
"long_name": row.LongName,
"short_name": row.ShortName,
"text": row.Text,
"packet_id": row.PacketID,
"channel_id": row.ChannelID,
"topic": row.Topic,
"status": row.Status,
"error": row.Error,
"received_at": row.ReceivedAt,
"processed_at": row.ProcessedAt,
"deleted_at": row.DeletedAt,
"created_at": row.CreatedAt,
}
}
// enqueueChannelMessageToLLM 将频道消息添加到 LLM 队列
// 为每个启用了「包含频道消息」的机器人都创建一条独立的队列记录
func enqueueChannelMessageToLLM(s *Store, record map[string]any) error {
if s == nil {
return nil
}
text, _ := record["text"].(string)
if text == "" {
return nil
}
fromNodeID, _ := record["from"].(string)
if fromNodeID == "" {
return nil
}
fromNodeNum, err := int64FromAny(record["from_num"])
if err != nil {
fromNodeNum = 0
}
var packetID int64
if p, ok := record["packet_id"].(float64); ok {
packetID = int64(p)
}
topic, _ := record["topic"].(string)
var longName, shortName *string
if ln, ok := record["long_name"].(string); ok && ln != "" {
longName = &ln
}
if sn, ok := record["short_name"].(string); ok && sn != "" {
shortName = &sn
}
var channelID *string
if cid, ok := record["channel_id"].(string); ok && cid != "" {
channelID = &cid
}
contentJSON, err := json.Marshal(record)
var contentPtr *string
if err == nil {
s := string(contentJSON)
contentPtr = &s
}
// 查询所有启用了 LLM 队列且包含频道消息的机器人
// SQLite 中 numeric 布尔值用 1/0 存储,必须用整数查询
var bots []BotNodeRecord
err = s.db.Where("llm_queue_enabled = ? AND llm_include_channel_messages = ?", 1, 1).Find(&bots).Error
if err != nil {
return fmt.Errorf("query bots for channel message enqueue: %w", err)
}
// 为每个符合条件的机器人创建一条队列记录(忽略机器人自己发送的消息)
for _, bot := range bots {
if fromNodeID == bot.NodeID {
continue
}
_, err = s.EnqueueLLMMessage(LLMMessageQueueInput{
BotID: bot.ID,
BotNodeID: bot.NodeID,
BotNodeNum: bot.NodeNum,
FromNodeID: fromNodeID,
FromNodeNum: fromNodeNum,
LongName: longName,
ShortName: shortName,
Text: text,
PacketID: packetID,
ChannelID: channelID,
Topic: topic,
MessageType: "channel",
ContentJSON: contentPtr,
})
if err != nil {
printJSON(map[string]any{
"event": "llm_queue_enqueue_failed",
"bot_id": bot.ID,
"from": fromNodeID,
"text": text,
"error": err.Error(),
})
}
}
return nil
}
+18
View File
@@ -0,0 +1,18 @@
package store
func (s *Store) InsertLoginLog(log LoginLogRecord) error {
return s.db.Create(&log).Error
}
func (s *Store) ListLoginLogs(opts ListOptions) ([]LoginLogRecord, error) {
opts = NormalizeListOptions(opts)
var rows []LoginLogRecord
q := s.db.Order("created_at DESC").Order("id DESC").Limit(opts.Limit).Offset(opts.Offset)
if opts.Since != nil {
q = q.Where("created_at >= ?", *opts.Since)
}
if opts.Until != nil {
q = q.Where("created_at <= ?", *opts.Until)
}
return rows, q.Find(&rows).Error
}
+358
View File
@@ -0,0 +1,358 @@
package store
import (
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"net/url"
"strings"
"time"
"unicode"
"gorm.io/gorm"
)
const (
defaultMapTileSourceName = "OpenStreetMap Japan"
defaultMapTileSourceURLTemplate = "https://tile.openstreetmap.jp/{z}/{x}/{y}.png"
defaultMapTileSourceAttribution = "&copy; OpenStreetMap contributors"
defaultMapTileSourceMaxZoom = 19
maxMapTileSourceURLLength = 2048
)
var (
ErrMapTileSourceAlreadyExists = errors.New("map source already exists")
ErrMapTileSourceCannotDeleteDefault = errors.New("default map source cannot be deleted")
ErrMapTileSourceCannotDisableDefault = errors.New("default map source cannot be disabled")
ErrMapTileSourceDefaultMustBeEnabled = errors.New("default map source must be enabled")
)
type MapTileSourceInput struct {
Name string
URLTemplate string
Attribution string
MaxZoom int
Enabled bool
IsDefault bool
ProxyEnabled bool
}
func (s *Store) ListMapTileSources(opts ListOptions) ([]MapTileSourceRecord, error) {
opts = NormalizeListOptions(opts)
var rows []MapTileSourceRecord
q := s.db.Model(&MapTileSourceRecord{}).
Order("is_default DESC").
Order("updated_at DESC").
Order("id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
func (s *Store) CountMapTileSources(opts ListOptions) (int64, error) {
var total int64
return total, s.db.Model(&MapTileSourceRecord{}).Count(&total).Error
}
func (s *Store) ListEnabledMapTileSources() ([]MapTileSourceRecord, error) {
var rows []MapTileSourceRecord
if err := s.db.Model(&MapTileSourceRecord{}).
Where("enabled = ?", true).
Order("is_default DESC").
Order("updated_at DESC").
Order("id DESC").
Find(&rows).Error; err != nil {
return nil, err
}
if len(rows) == 0 {
return []MapTileSourceRecord{defaultMapTileSourceRecord()}, nil
}
return rows, nil
}
func (s *Store) GetDefaultMapTileSource() (*MapTileSourceRecord, error) {
var row MapTileSourceRecord
err := s.db.Where("enabled = ? AND is_default = ?", true, true).Order("id ASC").Take(&row).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
fallback := defaultMapTileSourceRecord()
return &fallback, nil
}
if err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) GetEnabledMapTileSourceByHash(hash string) (*MapTileSourceRecord, error) {
var row MapTileSourceRecord
if err := s.db.Where("enabled = ? AND proxy_enabled = ? AND url_template_hash = ?", true, true, hash).Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) CreateMapTileSource(input MapTileSourceInput) (*MapTileSourceRecord, error) {
row, err := mapTileSourceFromInput(input)
if err != nil {
return nil, err
}
if row.IsDefault && !row.Enabled {
return nil, ErrMapTileSourceDefaultMustBeEnabled
}
if err := s.ensureMapTileSourceUnique(0, row.Name, row.URLTemplate); err != nil {
return nil, err
}
if err := s.db.Transaction(func(tx *gorm.DB) error {
if row.IsDefault {
if err := tx.Model(&MapTileSourceRecord{}).Where("is_default = ?", true).Update("is_default", false).Error; err != nil {
return err
}
}
return tx.Create(row).Error
}); err != nil {
return nil, err
}
return row, nil
}
func (s *Store) UpdateMapTileSource(id uint64, input MapTileSourceInput) (*MapTileSourceRecord, error) {
if id == 0 {
return nil, fmt.Errorf("map source id is required")
}
row, err := mapTileSourceFromInput(input)
if err != nil {
return nil, err
}
var updated MapTileSourceRecord
if err := s.db.Transaction(func(tx *gorm.DB) error {
var existing MapTileSourceRecord
if err := tx.Where("id = ?", id).Take(&existing).Error; err != nil {
return err
}
if existing.IsDefault && !row.Enabled {
return ErrMapTileSourceCannotDisableDefault
}
if row.IsDefault && !row.Enabled {
return ErrMapTileSourceDefaultMustBeEnabled
}
if !row.IsDefault && existing.IsDefault {
row.IsDefault = true
}
if err := ensureMapTileSourceUniqueTx(tx, id, row.Name, row.URLTemplate); err != nil {
return err
}
if row.IsDefault {
if err := tx.Model(&MapTileSourceRecord{}).Where("id <> ? AND is_default = ?", id, true).Update("is_default", false).Error; err != nil {
return err
}
}
updates := map[string]any{
"name": row.Name,
"url_template": row.URLTemplate,
"url_template_hash": row.URLTemplateHash,
"attribution": row.Attribution,
"max_zoom": row.MaxZoom,
"enabled": row.Enabled,
"is_default": row.IsDefault,
"proxy_enabled": row.ProxyEnabled,
"updated_at": time.Now(),
}
if err := tx.Model(&MapTileSourceRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return err
}
return tx.Where("id = ?", id).Take(&updated).Error
}); err != nil {
return nil, err
}
return &updated, nil
}
func (s *Store) DeleteMapTileSource(id uint64) error {
if id == 0 {
return fmt.Errorf("map source id is required")
}
return s.db.Transaction(func(tx *gorm.DB) error {
var row MapTileSourceRecord
if err := tx.Where("id = ?", id).Take(&row).Error; err != nil {
return err
}
if row.IsDefault {
return ErrMapTileSourceCannotDeleteDefault
}
result := tx.Where("id = ?", id).Delete(&MapTileSourceRecord{})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
})
}
func (s *Store) SetDefaultMapTileSource(id uint64) (*MapTileSourceRecord, error) {
if id == 0 {
return nil, fmt.Errorf("map source id is required")
}
var row MapTileSourceRecord
if err := s.db.Transaction(func(tx *gorm.DB) error {
if err := tx.Where("id = ?", id).Take(&row).Error; err != nil {
return err
}
if !row.Enabled {
return ErrMapTileSourceDefaultMustBeEnabled
}
if err := tx.Model(&MapTileSourceRecord{}).Where("is_default = ?", true).Update("is_default", false).Error; err != nil {
return err
}
if err := tx.Model(&MapTileSourceRecord{}).Where("id = ?", id).Updates(map[string]any{"is_default": true, "updated_at": time.Now()}).Error; err != nil {
return err
}
return tx.Where("id = ?", id).Take(&row).Error
}); err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) EnsureDefaultMapTileSource() error {
return s.db.Transaction(func(tx *gorm.DB) error {
var count int64
if err := tx.Model(&MapTileSourceRecord{}).Count(&count).Error; err != nil {
return err
}
if count == 0 {
row := defaultMapTileSourceRecord()
return tx.Create(&row).Error
}
var defaults []MapTileSourceRecord
if err := tx.Where("enabled = ? AND is_default = ?", true, true).Order("id ASC").Find(&defaults).Error; err != nil {
return err
}
if len(defaults) > 0 {
return tx.Model(&MapTileSourceRecord{}).Where("id <> ? AND is_default = ?", defaults[0].ID, true).Update("is_default", false).Error
}
var enabled MapTileSourceRecord
err := tx.Where("enabled = ?", true).Order("id ASC").Take(&enabled).Error
if err == nil {
return tx.Model(&MapTileSourceRecord{}).Where("id = ?", enabled.ID).Updates(map[string]any{"is_default": true, "updated_at": time.Now()}).Error
}
if !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
row := defaultMapTileSourceRecord()
var existing MapTileSourceRecord
err = tx.Where("name = ? OR url_template = ?", row.Name, row.URLTemplate).Order("id ASC").Take(&existing).Error
if err == nil {
return tx.Model(&MapTileSourceRecord{}).Where("id = ?", existing.ID).Updates(map[string]any{"enabled": true, "is_default": true, "updated_at": time.Now()}).Error
}
if !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
return tx.Create(&row).Error
})
}
func MapTileSourceHash(urlTemplate string) string {
h := sha256.Sum256([]byte(urlTemplate))
return hex.EncodeToString(h[:])
}
func defaultMapTileSourceRecord() MapTileSourceRecord {
return MapTileSourceRecord{
Name: defaultMapTileSourceName,
URLTemplate: defaultMapTileSourceURLTemplate,
URLTemplateHash: MapTileSourceHash(defaultMapTileSourceURLTemplate),
Attribution: defaultMapTileSourceAttribution,
MaxZoom: defaultMapTileSourceMaxZoom,
Enabled: true,
IsDefault: true,
ProxyEnabled: true,
}
}
func mapTileSourceFromInput(input MapTileSourceInput) (*MapTileSourceRecord, error) {
name := strings.TrimSpace(input.Name)
if name == "" {
return nil, fmt.Errorf("map source name is required")
}
urlTemplate, err := normalizeMapTileSourceURLTemplate(input.URLTemplate)
if err != nil {
return nil, err
}
maxZoom := input.MaxZoom
if maxZoom == 0 {
maxZoom = defaultMapTileSourceMaxZoom
}
if maxZoom < 1 || maxZoom > 30 {
return nil, fmt.Errorf("max zoom must be between 1 and 30")
}
return &MapTileSourceRecord{
Name: name,
URLTemplate: urlTemplate,
URLTemplateHash: MapTileSourceHash(urlTemplate),
Attribution: strings.TrimSpace(input.Attribution),
MaxZoom: maxZoom,
Enabled: input.Enabled,
IsDefault: input.IsDefault,
ProxyEnabled: input.ProxyEnabled,
}, nil
}
func normalizeMapTileSourceURLTemplate(value string) (string, error) {
value = strings.TrimSpace(value)
if value == "" {
return "", fmt.Errorf("map source url template is required")
}
if len(value) > maxMapTileSourceURLLength {
return "", fmt.Errorf("map source url template is too long")
}
for _, r := range value {
if unicode.IsControl(r) || unicode.IsSpace(r) {
return "", fmt.Errorf("map source url template must not contain whitespace or control characters")
}
}
for _, placeholder := range []string{"{z}", "{x}", "{y}"} {
if strings.Count(value, placeholder) != 1 {
return "", fmt.Errorf("map source url template must contain %s exactly once", placeholder)
}
}
parsed, err := url.Parse(value)
if err != nil {
return "", fmt.Errorf("map source url template is invalid")
}
if parsed.Scheme != "http" && parsed.Scheme != "https" {
return "", fmt.Errorf("map source url template must use http or https")
}
if parsed.Host == "" {
return "", fmt.Errorf("map source url template host is required")
}
if parsed.User != nil {
return "", fmt.Errorf("map source url template must not contain credentials")
}
return value, nil
}
func (s *Store) ensureMapTileSourceUnique(id uint64, name, urlTemplate string) error {
return ensureMapTileSourceUniqueTx(s.db, id, name, urlTemplate)
}
func ensureMapTileSourceUniqueTx(tx *gorm.DB, id uint64, name, urlTemplate string) error {
var existing MapTileSourceRecord
q := tx.Where("name = ? OR url_template = ?", name, urlTemplate)
if id != 0 {
q = q.Where("id <> ?", id)
}
err := q.Take(&existing).Error
if err == nil {
return ErrMapTileSourceAlreadyExists
}
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
return err
}
+258
View File
@@ -0,0 +1,258 @@
package store
import (
"errors"
"strings"
"testing"
"gorm.io/gorm"
)
func TestMapTileSourceDefaultSeeded(t *testing.T) {
st := openTestStore(t)
defer st.Close()
row, err := st.GetDefaultMapTileSource()
if err != nil {
t.Fatalf("GetDefaultMapTileSource() error = %v", err)
}
if row.Name != defaultMapTileSourceName || row.URLTemplate != defaultMapTileSourceURLTemplate || !row.Enabled || !row.IsDefault {
t.Fatalf("default map source = %+v, want built-in default", row)
}
}
func TestCreateMapTileSourceValidation(t *testing.T) {
st := openTestStore(t)
defer st.Close()
if _, err := st.CreateMapTileSource(MapTileSourceInput{Name: "bad", URLTemplate: "https://tiles.example.com/{z}/{x}.png", MaxZoom: 19, Enabled: true, ProxyEnabled: true}); err == nil {
t.Fatal("CreateMapTileSource() missing placeholder error = nil, want error")
}
if _, err := st.CreateMapTileSource(MapTileSourceInput{Name: "bad", URLTemplate: "javascript:alert(1)/{z}/{x}/{y}", MaxZoom: 19, Enabled: true, ProxyEnabled: true}); err == nil {
t.Fatal("CreateMapTileSource() invalid scheme error = nil, want error")
}
if _, err := st.CreateMapTileSource(MapTileSourceInput{Name: "bad", URLTemplate: "https://user:pass@tiles.example.com/{z}/{x}/{y}.png", MaxZoom: 19, Enabled: true, ProxyEnabled: true}); err == nil {
t.Fatal("CreateMapTileSource() credentials error = nil, want error")
}
}
func TestListEnabledMapTileSources(t *testing.T) {
st := openTestStore(t)
defer st.Close()
disabled, err := st.CreateMapTileSource(MapTileSourceInput{Name: "Disabled", URLTemplate: "https://disabled.example.com/{z}/{x}/{y}.png", MaxZoom: 18, Enabled: false})
if err != nil {
t.Fatalf("CreateMapTileSource(disabled) error = %v", err)
}
custom, err := st.CreateMapTileSource(MapTileSourceInput{Name: "Custom", URLTemplate: "https://custom.example.com/{z}/{x}/{y}.png", MaxZoom: 18, Enabled: true, ProxyEnabled: true})
if err != nil {
t.Fatalf("CreateMapTileSource(custom) error = %v", err)
}
if _, err := st.SetDefaultMapTileSource(custom.ID); err != nil {
t.Fatalf("SetDefaultMapTileSource() error = %v", err)
}
rows, err := st.ListEnabledMapTileSources()
if err != nil {
t.Fatalf("ListEnabledMapTileSources() error = %v", err)
}
if len(rows) < 2 {
t.Fatalf("ListEnabledMapTileSources() length = %d, want at least 2", len(rows))
}
if rows[0].ID != custom.ID {
t.Fatalf("first enabled source id = %d, want default %d", rows[0].ID, custom.ID)
}
for _, row := range rows {
if row.ID == disabled.ID {
t.Fatalf("disabled source was returned: %+v", row)
}
if !row.Enabled {
t.Fatalf("disabled row returned: %+v", row)
}
}
}
func TestMapTileSourceDuplicateAndDefaultRules(t *testing.T) {
st := openTestStore(t)
defer st.Close()
first, err := st.CreateMapTileSource(MapTileSourceInput{Name: "Custom", URLTemplate: "https://tiles.example.com/{z}/{x}/{y}.png", MaxZoom: 18, Enabled: true, ProxyEnabled: true})
if err != nil {
t.Fatalf("CreateMapTileSource() error = %v", err)
}
if _, err := st.CreateMapTileSource(MapTileSourceInput{Name: "Custom", URLTemplate: "https://tiles2.example.com/{z}/{x}/{y}.png", MaxZoom: 18, Enabled: true, ProxyEnabled: true}); !errors.Is(err, ErrMapTileSourceAlreadyExists) {
t.Fatalf("duplicate name error = %v, want ErrMapTileSourceAlreadyExists", err)
}
if _, err := st.CreateMapTileSource(MapTileSourceInput{Name: "Custom 2", URLTemplate: first.URLTemplate, MaxZoom: 18, Enabled: true, ProxyEnabled: true}); !errors.Is(err, ErrMapTileSourceAlreadyExists) {
t.Fatalf("duplicate url error = %v, want ErrMapTileSourceAlreadyExists", err)
}
updated, err := st.SetDefaultMapTileSource(first.ID)
if err != nil {
t.Fatalf("SetDefaultMapTileSource() error = %v", err)
}
if !updated.IsDefault {
t.Fatalf("updated default = %+v, want is_default", updated)
}
oldDefault, err := st.GetDefaultMapTileSource()
if err != nil {
t.Fatalf("GetDefaultMapTileSource() error = %v", err)
}
if oldDefault.ID != first.ID {
t.Fatalf("default id = %d, want %d", oldDefault.ID, first.ID)
}
if _, err := st.UpdateMapTileSource(first.ID, MapTileSourceInput{Name: first.Name, URLTemplate: first.URLTemplate, Attribution: first.Attribution, MaxZoom: first.MaxZoom, Enabled: false, IsDefault: true}); !errors.Is(err, ErrMapTileSourceCannotDisableDefault) {
t.Fatalf("disable default error = %v, want ErrMapTileSourceCannotDisableDefault", err)
}
if err := st.DeleteMapTileSource(first.ID); !errors.Is(err, ErrMapTileSourceCannotDeleteDefault) {
t.Fatalf("delete default error = %v, want ErrMapTileSourceCannotDeleteDefault", err)
}
}
func TestMapTileSourceHashIsSetOnCreate(t *testing.T) {
st := openTestStore(t)
defer st.Close()
row, err := st.CreateMapTileSource(MapTileSourceInput{Name: "Hashed", URLTemplate: "https://test.example.com/{z}/{x}/{y}.png", MaxZoom: 18, Enabled: true, ProxyEnabled: true})
if err != nil {
t.Fatalf("CreateMapTileSource() error = %v", err)
}
want := MapTileSourceHash("https://test.example.com/{z}/{x}/{y}.png")
if row.URLTemplateHash != want {
t.Fatalf("URLTemplateHash = %q, want %q", row.URLTemplateHash, want)
}
if !row.ProxyEnabled {
t.Fatal("ProxyEnabled = false, want true")
}
}
func TestMapTileSourceDefaultHasHash(t *testing.T) {
st := openTestStore(t)
defer st.Close()
row, err := st.GetDefaultMapTileSource()
if err != nil {
t.Fatalf("GetDefaultMapTileSource() error = %v", err)
}
want := MapTileSourceHash(defaultMapTileSourceURLTemplate)
if row.URLTemplateHash != want {
t.Fatalf("default URLTemplateHash = %q, want %q", row.URLTemplateHash, want)
}
}
func TestGetEnabledMapTileSourceByHash(t *testing.T) {
st := openTestStore(t)
defer st.Close()
row, err := st.CreateMapTileSource(MapTileSourceInput{Name: "HashLookup", URLTemplate: "https://lookup.example.com/{z}/{x}/{y}.png", MaxZoom: 18, Enabled: true, ProxyEnabled: true})
if err != nil {
t.Fatalf("CreateMapTileSource() error = %v", err)
}
found, err := st.GetEnabledMapTileSourceByHash(row.URLTemplateHash)
if err != nil {
t.Fatalf("GetEnabledMapTileSourceByHash() error = %v", err)
}
if found.ID != row.ID {
t.Fatalf("found ID = %d, want %d", found.ID, row.ID)
}
}
func TestGetEnabledMapTileSourceByHashDisabled(t *testing.T) {
st := openTestStore(t)
defer st.Close()
row, err := st.CreateMapTileSource(MapTileSourceInput{Name: "DisabledHash", URLTemplate: "https://disabled-hash.example.com/{z}/{x}/{y}.png", MaxZoom: 18, Enabled: false})
if err != nil {
t.Fatalf("CreateMapTileSource() error = %v", err)
}
_, err = st.GetEnabledMapTileSourceByHash(row.URLTemplateHash)
if !errors.Is(err, gorm.ErrRecordNotFound) {
t.Fatalf("GetEnabledMapTileSourceByHash(disabled) = %v, want gorm.ErrRecordNotFound", err)
}
}
func TestGetEnabledMapTileSourceByHashProxyDisabled(t *testing.T) {
st := openTestStore(t)
defer st.Close()
row, err := st.CreateMapTileSource(MapTileSourceInput{Name: "ProxyDisabledHash", URLTemplate: "https://proxy-disabled.example.com/{z}/{x}/{y}.png", MaxZoom: 18, Enabled: true, ProxyEnabled: false})
if err != nil {
t.Fatalf("CreateMapTileSource() error = %v", err)
}
_, err = st.GetEnabledMapTileSourceByHash(row.URLTemplateHash)
if !errors.Is(err, gorm.ErrRecordNotFound) {
t.Fatalf("GetEnabledMapTileSourceByHash(proxy disabled) = %v, want gorm.ErrRecordNotFound", err)
}
}
func TestGetEnabledMapTileSourceByHashUnknown(t *testing.T) {
st := openTestStore(t)
defer st.Close()
_, err := st.GetEnabledMapTileSourceByHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
if !errors.Is(err, gorm.ErrRecordNotFound) {
t.Fatalf("GetEnabledMapTileSourceByHash(unknown) = %v, want gorm.ErrRecordNotFound", err)
}
}
func TestPublicMapTileSourceDTOProxyURL(t *testing.T) {
st := openTestStore(t)
defer st.Close()
row, err := st.CreateMapTileSource(MapTileSourceInput{Name: "ProxyTest", URLTemplate: "https://proxy.example.com/{z}/{x}/{y}.png", MaxZoom: 18, Enabled: true, ProxyEnabled: true})
if err != nil {
t.Fatalf("CreateMapTileSource() error = %v", err)
}
dto := publicMapTileSourceDTO(*row)
urlTemplate, ok := dto["url_template"].(string)
if !ok {
t.Fatal("url_template is not a string")
}
wantPrefix := "/api/map/" + row.URLTemplateHash + "?x={x}&y={y}&z={z}"
if urlTemplate != wantPrefix {
t.Fatalf("url_template = %q, want %q", urlTemplate, wantPrefix)
}
if strings.Contains(urlTemplate, "proxy.example.com") {
t.Fatal("url_template should not contain upstream hostname")
}
}
func TestPublicMapTileSourceDTORawURLWhenProxyDisabled(t *testing.T) {
st := openTestStore(t)
defer st.Close()
row, err := st.CreateMapTileSource(MapTileSourceInput{Name: "RawTest", URLTemplate: "https://raw.example.com/{z}/{x}/{y}.png", MaxZoom: 18, Enabled: true, ProxyEnabled: false})
if err != nil {
t.Fatalf("CreateMapTileSource() error = %v", err)
}
dto := publicMapTileSourceDTO(*row)
urlTemplate, ok := dto["url_template"].(string)
if !ok {
t.Fatal("url_template is not a string")
}
if urlTemplate != row.URLTemplate {
t.Fatalf("url_template = %q, want raw %q", urlTemplate, row.URLTemplate)
}
}
func TestMapTileSourceHashFunction(t *testing.T) {
hash1 := MapTileSourceHash("https://tile.openstreetmap.jp/{z}/{x}/{y}.png")
hash2 := MapTileSourceHash("https://tile.openstreetmap.jp/{z}/{x}/{y}.png")
hash3 := MapTileSourceHash("https://other.example.com/{z}/{x}/{y}.png")
if hash1 != hash2 {
t.Fatal("hash should be deterministic")
}
if len(hash1) != 64 {
t.Fatalf("hash length = %d, want 64", len(hash1))
}
if hash1 == hash3 {
t.Fatal("different URLs should produce different hashes")
}
}
+378
View File
@@ -0,0 +1,378 @@
package store
import (
"errors"
"fmt"
"strings"
"time"
"gorm.io/gorm"
)
const (
MQTTForwardDirectionSourceToTarget = "source_to_target"
MQTTForwardDirectionBidirectional = "bidirectional"
)
var (
ErrMQTTForwarderAlreadyExists = errors.New("mqtt forwarder already exists")
ErrMQTTForwardTopicAlreadyExists = errors.New("mqtt forward topic already exists")
)
type MQTTForwarderInput struct {
Name string
Enabled bool
SourceHost string
SourcePort int
SourceUsername string
SourcePassword *string
SourceClientID string
SourceTLS bool
TargetHost string
TargetPort int
TargetUsername string
TargetPassword *string
TargetClientID string
TargetTLS bool
}
type MQTTForwardTopicInput struct {
Topic string
Enabled bool
Direction string
SourcePrefix string
TargetPrefix string
QoS int
Retain bool
}
type MQTTForwarderConfig struct {
Forwarder MQTTForwarderRecord
Topics []MQTTForwardTopicRecord
}
func (s *Store) ListMQTTForwarders(opts ListOptions) ([]MQTTForwarderRecord, error) {
opts = NormalizeListOptions(opts)
var rows []MQTTForwarderRecord
q := s.db.Model(&MQTTForwarderRecord{}).
Order("updated_at DESC").
Order("id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
func (s *Store) CountMQTTForwarders(opts ListOptions) (int64, error) {
var total int64
return total, s.db.Model(&MQTTForwarderRecord{}).Count(&total).Error
}
func (s *Store) GetMQTTForwarder(id uint64) (*MQTTForwarderRecord, error) {
var row MQTTForwarderRecord
if err := s.db.Where("id = ?", id).Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) CreateMQTTForwarder(input MQTTForwarderInput) (*MQTTForwarderRecord, error) {
row, err := mqttForwarderFromInput(input, nil)
if err != nil {
return nil, err
}
if err := s.ensureMQTTForwarderNameUnique(0, row.Name); err != nil {
return nil, err
}
if err := s.db.Create(row).Error; err != nil {
return nil, err
}
return row, nil
}
func (s *Store) UpdateMQTTForwarder(id uint64, input MQTTForwarderInput) (*MQTTForwarderRecord, error) {
if id == 0 {
return nil, fmt.Errorf("mqtt forwarder id is required")
}
existing, err := s.GetMQTTForwarder(id)
if err != nil {
return nil, err
}
row, err := mqttForwarderFromInput(input, existing)
if err != nil {
return nil, err
}
if err := s.ensureMQTTForwarderNameUnique(id, row.Name); err != nil {
return nil, err
}
updates := map[string]any{
"name": row.Name, "enabled": row.Enabled,
"source_host": row.SourceHost, "source_port": row.SourcePort, "source_username": row.SourceUsername,
"source_password": row.SourcePassword, "source_client_id": row.SourceClientID, "source_tls": row.SourceTLS,
"target_host": row.TargetHost, "target_port": row.TargetPort, "target_username": row.TargetUsername,
"target_password": row.TargetPassword, "target_client_id": row.TargetClientID, "target_tls": row.TargetTLS,
"updated_at": time.Now(),
}
if err := s.db.Model(&MQTTForwarderRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return nil, err
}
return s.GetMQTTForwarder(id)
}
func (s *Store) DeleteMQTTForwarder(id uint64) error {
if id == 0 {
return fmt.Errorf("mqtt forwarder id is required")
}
return s.db.Transaction(func(tx *gorm.DB) error {
if err := tx.Where("forwarder_id = ?", id).Delete(&MQTTForwardTopicRecord{}).Error; err != nil {
return err
}
result := tx.Where("id = ?", id).Delete(&MQTTForwarderRecord{})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
})
}
func (s *Store) ListMQTTForwardTopics(forwarderID uint64, opts ListOptions) ([]MQTTForwardTopicRecord, error) {
opts = NormalizeListOptions(opts)
var rows []MQTTForwardTopicRecord
q := s.db.Model(&MQTTForwardTopicRecord{}).
Where("forwarder_id = ?", forwarderID).
Order("updated_at DESC").
Order("id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
func (s *Store) CountMQTTForwardTopics(forwarderID uint64) (int64, error) {
var total int64
return total, s.db.Model(&MQTTForwardTopicRecord{}).Where("forwarder_id = ?", forwarderID).Count(&total).Error
}
func (s *Store) GetMQTTForwardTopic(id uint64) (*MQTTForwardTopicRecord, error) {
var row MQTTForwardTopicRecord
if err := s.db.Where("id = ?", id).Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) CreateMQTTForwardTopic(forwarderID uint64, input MQTTForwardTopicInput) (*MQTTForwardTopicRecord, error) {
if _, err := s.GetMQTTForwarder(forwarderID); err != nil {
return nil, err
}
row, err := mqttForwardTopicFromInput(forwarderID, input)
if err != nil {
return nil, err
}
if err := s.ensureMQTTForwardTopicUnique(0, forwarderID, row.Topic); err != nil {
return nil, err
}
if err := s.db.Create(row).Error; err != nil {
return nil, err
}
return row, nil
}
func (s *Store) UpdateMQTTForwardTopic(id uint64, input MQTTForwardTopicInput) (*MQTTForwardTopicRecord, error) {
if id == 0 {
return nil, fmt.Errorf("mqtt forward topic id is required")
}
existing, err := s.GetMQTTForwardTopic(id)
if err != nil {
return nil, err
}
row, err := mqttForwardTopicFromInput(existing.ForwarderID, input)
if err != nil {
return nil, err
}
if err := s.ensureMQTTForwardTopicUnique(id, existing.ForwarderID, row.Topic); err != nil {
return nil, err
}
updates := map[string]any{
"topic": row.Topic, "enabled": row.Enabled, "direction": row.Direction,
"source_prefix": row.SourcePrefix, "target_prefix": row.TargetPrefix,
"qos": row.QoS, "retain": row.Retain, "updated_at": time.Now(),
}
if err := s.db.Model(&MQTTForwardTopicRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return nil, err
}
return s.GetMQTTForwardTopic(id)
}
func (s *Store) DeleteMQTTForwardTopic(id uint64) error {
result := s.db.Where("id = ?", id).Delete(&MQTTForwardTopicRecord{})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
}
func (s *Store) GetMQTTForwarderConfig(id uint64) (*MQTTForwarderConfig, error) {
forwarder, err := s.GetMQTTForwarder(id)
if err != nil {
return nil, err
}
var topics []MQTTForwardTopicRecord
if err := s.db.Where("forwarder_id = ? AND enabled = ?", id, true).Order("id ASC").Find(&topics).Error; err != nil {
return nil, err
}
return &MQTTForwarderConfig{Forwarder: *forwarder, Topics: topics}, nil
}
func (s *Store) ListEnabledMQTTForwarderConfigs() ([]MQTTForwarderConfig, error) {
var forwarders []MQTTForwarderRecord
if err := s.db.Where("enabled = ?", true).Order("id ASC").Find(&forwarders).Error; err != nil {
return nil, err
}
configs := make([]MQTTForwarderConfig, 0, len(forwarders))
for _, forwarder := range forwarders {
var topics []MQTTForwardTopicRecord
if err := s.db.Where("forwarder_id = ? AND enabled = ?", forwarder.ID, true).Order("id ASC").Find(&topics).Error; err != nil {
return nil, err
}
if len(topics) == 0 {
continue
}
configs = append(configs, MQTTForwarderConfig{Forwarder: forwarder, Topics: topics})
}
return configs, nil
}
func (s *Store) ensureMQTTForwarderNameUnique(id uint64, name string) error {
var existing MQTTForwarderRecord
q := s.db.Where("name = ?", name)
if id != 0 {
q = q.Where("id <> ?", id)
}
err := q.Take(&existing).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
if err != nil {
return err
}
return ErrMQTTForwarderAlreadyExists
}
func (s *Store) ensureMQTTForwardTopicUnique(id, forwarderID uint64, topic string) error {
var existing MQTTForwardTopicRecord
q := s.db.Where("forwarder_id = ? AND topic = ?", forwarderID, topic)
if id != 0 {
q = q.Where("id <> ?", id)
}
err := q.Take(&existing).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
if err != nil {
return err
}
return ErrMQTTForwardTopicAlreadyExists
}
func mqttForwarderFromInput(input MQTTForwarderInput, existing *MQTTForwarderRecord) (*MQTTForwarderRecord, error) {
name := strings.TrimSpace(input.Name)
if name == "" {
return nil, fmt.Errorf("mqtt forwarder name is required")
}
sourceHost := strings.TrimSpace(input.SourceHost)
if sourceHost == "" {
return nil, fmt.Errorf("source host is required")
}
if err := validateMQTTForwardPort(input.SourcePort, "source port"); err != nil {
return nil, err
}
targetHost := strings.TrimSpace(input.TargetHost)
if targetHost == "" {
return nil, fmt.Errorf("target host is required")
}
if err := validateMQTTForwardPort(input.TargetPort, "target port"); err != nil {
return nil, err
}
row := &MQTTForwarderRecord{
Name: name, Enabled: input.Enabled,
SourceHost: sourceHost, SourcePort: input.SourcePort, SourceUsername: strings.TrimSpace(input.SourceUsername), SourceClientID: strings.TrimSpace(input.SourceClientID), SourceTLS: input.SourceTLS,
TargetHost: targetHost, TargetPort: input.TargetPort, TargetUsername: strings.TrimSpace(input.TargetUsername), TargetClientID: strings.TrimSpace(input.TargetClientID), TargetTLS: input.TargetTLS,
}
if input.SourcePassword != nil {
row.SourcePassword = *input.SourcePassword
} else if existing != nil {
row.SourcePassword = existing.SourcePassword
}
if input.TargetPassword != nil {
row.TargetPassword = *input.TargetPassword
} else if existing != nil {
row.TargetPassword = existing.TargetPassword
}
return row, nil
}
func mqttForwardTopicFromInput(forwarderID uint64, input MQTTForwardTopicInput) (*MQTTForwardTopicRecord, error) {
if forwarderID == 0 {
return nil, fmt.Errorf("mqtt forwarder id is required")
}
topic := strings.TrimSpace(input.Topic)
if err := validateMQTTTopicFilter(topic); err != nil {
return nil, err
}
direction, err := normalizeMQTTForwardDirection(input.Direction)
if err != nil {
return nil, err
}
if input.QoS < 0 || input.QoS > 2 {
return nil, fmt.Errorf("qos must be 0, 1, or 2")
}
return &MQTTForwardTopicRecord{
ForwarderID: forwarderID, Topic: topic, Enabled: input.Enabled, Direction: direction,
SourcePrefix: strings.Trim(strings.TrimSpace(input.SourcePrefix), "/"),
TargetPrefix: strings.Trim(strings.TrimSpace(input.TargetPrefix), "/"),
QoS: input.QoS, Retain: input.Retain,
}, nil
}
func validateMQTTForwardPort(port int, label string) error {
if port <= 0 || port > 65535 {
return fmt.Errorf("%s must be between 1 and 65535", label)
}
return nil
}
func normalizeMQTTForwardDirection(direction string) (string, error) {
direction = strings.TrimSpace(direction)
if direction == "" {
direction = MQTTForwardDirectionSourceToTarget
}
switch direction {
case MQTTForwardDirectionSourceToTarget, MQTTForwardDirectionBidirectional:
return direction, nil
default:
return "", fmt.Errorf("invalid mqtt forward direction")
}
}
func validateMQTTTopicFilter(topic string) error {
if topic == "" {
return fmt.Errorf("topic is required")
}
parts := strings.Split(topic, "/")
for i, part := range parts {
if strings.Contains(part, "#") {
if part != "#" || i != len(parts)-1 {
return fmt.Errorf("invalid topic filter: # must be the last level")
}
}
if strings.Contains(part, "+") && part != "+" {
return fmt.Errorf("invalid topic filter: + must occupy an entire level")
}
}
return nil
}
+98
View File
@@ -0,0 +1,98 @@
package store
import (
"errors"
"fmt"
"strconv"
"strings"
"time"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
const (
RuntimeSettingAllowEncryptedForwarding = "mqtt.allow_encrypted_forwarding"
RuntimeSettingLLMQueueEnabled = "llm.queue_enabled"
RuntimeSettingLLMQueueIncludeChannel = "llm.include_channel_messages"
runtimeSettingTypeBool = "bool"
)
type RuntimeSettingsSnapshot struct {
AllowEncryptedForwarding bool
LLMQueueEnabled bool
LLMIncludeChannel bool
}
func (s *Store) GetRuntimeSettings() (RuntimeSettingsSnapshot, error) {
allowEncrypted, err := s.GetBoolRuntimeSetting(RuntimeSettingAllowEncryptedForwarding, false)
if err != nil {
return RuntimeSettingsSnapshot{}, err
}
llmQueueEnabled, err := s.GetBoolRuntimeSetting(RuntimeSettingLLMQueueEnabled, true)
if err != nil {
return RuntimeSettingsSnapshot{}, err
}
llmIncludeChannel, err := s.GetBoolRuntimeSetting(RuntimeSettingLLMQueueIncludeChannel, false)
if err != nil {
return RuntimeSettingsSnapshot{}, err
}
return RuntimeSettingsSnapshot{
AllowEncryptedForwarding: allowEncrypted,
LLMQueueEnabled: llmQueueEnabled,
LLMIncludeChannel: llmIncludeChannel,
}, nil
}
func (s *Store) GetBoolRuntimeSetting(key string, defaultValue bool) (bool, error) {
key = strings.TrimSpace(key)
if key == "" {
return false, fmt.Errorf("runtime setting key is required")
}
var row RuntimeSettingRecord
err := s.db.Where("`key` = ?", key).Take(&row).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
return defaultValue, nil
}
if err != nil {
return false, err
}
if row.ValueType != "" && row.ValueType != runtimeSettingTypeBool {
return false, fmt.Errorf("runtime setting %s has type %s, want %s", key, row.ValueType, runtimeSettingTypeBool)
}
value, err := strconv.ParseBool(strings.TrimSpace(row.Value))
if err != nil {
return false, fmt.Errorf("parse runtime setting %s: %w", key, err)
}
return value, nil
}
func (s *Store) SetBoolRuntimeSetting(key string, value bool, label string) (*RuntimeSettingRecord, error) {
key = strings.TrimSpace(key)
if key == "" {
return nil, fmt.Errorf("runtime setting key is required")
}
row := RuntimeSettingRecord{
Key: key,
Value: strconv.FormatBool(value),
ValueType: runtimeSettingTypeBool,
Label: strings.TrimSpace(label),
}
if err := s.db.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "key"}},
DoUpdates: clause.Assignments(map[string]any{
"value": row.Value,
"value_type": row.ValueType,
"label": row.Label,
"updated_at": time.Now(),
}),
}).Create(&row).Error; err != nil {
return nil, err
}
if err := s.db.Where("`key` = ?", key).Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
@@ -0,0 +1,38 @@
package store
import "testing"
func TestRuntimeSettingsDefaultAndUpdates(t *testing.T) {
st := openTestStore(t)
defer st.Close()
settings, err := st.GetRuntimeSettings()
if err != nil {
t.Fatalf("GetRuntimeSettings() error = %v", err)
}
if settings.AllowEncryptedForwarding {
t.Fatalf("AllowEncryptedForwarding = true, want false")
}
if _, err := st.SetBoolRuntimeSetting(RuntimeSettingAllowEncryptedForwarding, true, "test setting"); err != nil {
t.Fatalf("SetBoolRuntimeSetting(true) error = %v", err)
}
settings, err = st.GetRuntimeSettings()
if err != nil {
t.Fatalf("GetRuntimeSettings() after true error = %v", err)
}
if !settings.AllowEncryptedForwarding {
t.Fatalf("AllowEncryptedForwarding = false, want true")
}
if _, err := st.SetBoolRuntimeSetting(RuntimeSettingAllowEncryptedForwarding, false, "test setting"); err != nil {
t.Fatalf("SetBoolRuntimeSetting(false) error = %v", err)
}
settings, err = st.GetRuntimeSettings()
if err != nil {
t.Fatalf("GetRuntimeSettings() after false error = %v", err)
}
if settings.AllowEncryptedForwarding {
t.Fatalf("AllowEncryptedForwarding = true, want false")
}
}
+133
View File
@@ -0,0 +1,133 @@
package store
import (
"fmt"
"strings"
"time"
"gorm.io/gorm"
"meshtastic_mqtt_server/internal/config"
)
func (s *Store) ListSigns(opts ListOptions) ([]SignRecord, error) {
opts = NormalizeListOptions(opts)
var rows []SignRecord
q := applySignFilters(s.db.Model(&SignRecord{}), opts).
Order("sign_time DESC").
Order("id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
type SignDayCount struct {
Date string `gorm:"column:sign_date"`
Count int64 `gorm:"column:count"`
}
func (s *Store) CountSigns(opts ListOptions) (int64, error) {
var total int64
q := applySignFilters(s.db.Model(&SignRecord{}), opts)
return total, q.Count(&total).Error
}
func (s *Store) CountSignsByDay(opts ListOptions) ([]SignDayCount, error) {
var rows []SignDayCount
dateExpr := "strftime('%Y-%m-%d', sign_time)"
if s.driver == config.DriverMySQL {
dateExpr = "DATE_FORMAT(sign_time, '%Y-%m-%d')"
}
q := applySignFilters(s.db.Model(&SignRecord{}), opts).
Select(dateExpr + " AS sign_date, COUNT(*) AS count").
Group(dateExpr).
Order("sign_date DESC")
return rows, q.Scan(&rows).Error
}
func (s *Store) GetSignByID(id uint64) (*SignRecord, error) {
var row SignRecord
if err := s.db.Where("id = ?", id).Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) CreateSign(nodeID string, longName, shortName *string, signText string, signTime time.Time) (*SignRecord, error) {
nodeID = strings.TrimSpace(nodeID)
signText = strings.TrimSpace(signText)
if nodeID == "" {
return nil, fmt.Errorf("node id is required")
}
if signText == "" {
return nil, fmt.Errorf("sign text is required")
}
if signTime.IsZero() {
signTime = time.Now()
}
row := SignRecord{NodeID: nodeID, LongName: trimNullableString(longName), ShortName: trimNullableString(shortName), SignText: signText, SignTime: signTime}
if err := s.db.Create(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) UpdateSign(id uint64, nodeID string, longName, shortName *string, signText string, signTime time.Time) (*SignRecord, error) {
if id == 0 {
return nil, fmt.Errorf("sign id is required")
}
nodeID = strings.TrimSpace(nodeID)
signText = strings.TrimSpace(signText)
if nodeID == "" {
return nil, fmt.Errorf("node id is required")
}
if signText == "" {
return nil, fmt.Errorf("sign text is required")
}
if signTime.IsZero() {
signTime = time.Now()
}
if _, err := s.GetSignByID(id); err != nil {
return nil, err
}
updates := map[string]any{"node_id": nodeID, "long_name": trimNullableString(longName), "short_name": trimNullableString(shortName), "sign_text": signText, "sign_time": signTime}
if err := s.db.Model(&SignRecord{}).Where("id = ?", id).Updates(updates).Error; err != nil {
return nil, err
}
return s.GetSignByID(id)
}
func (s *Store) DeleteSign(id uint64) error {
result := s.db.Where("id = ?", id).Delete(&SignRecord{})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
}
func applySignFilters(q *gorm.DB, opts ListOptions) *gorm.DB {
if opts.NodeID != "" {
q = q.Where("node_id = ?", opts.NodeID)
}
if opts.Since != nil {
q = q.Where("sign_time >= ?", *opts.Since)
}
if opts.Until != nil {
q = q.Where("sign_time <= ?", *opts.Until)
}
return q
}
func trimNullableString(value *string) *string {
if value == nil {
return nil
}
trimmed := strings.TrimSpace(*value)
if trimmed == "" {
return nil
}
return &trimmed
}
+342
View File
@@ -0,0 +1,342 @@
package store
import (
"fmt"
"math"
"time"
"gorm.io/gorm"
)
type ListOptions struct {
Limit int
Offset int
NodeID string
ChannelID string
Since *time.Time
Until *time.Time
MinLat *float64
MaxLat *float64
MinLng *float64
MaxLng *float64
}
type MapReportViewportOptions struct {
ListOptions ListOptions
Zoom int
Limit int
ClusterThreshold int
TargetCells int
}
type MapReportViewportResult struct {
Mode string
Total int64
Points []MapReportRecord
Clusters []MapReportClusterRecord
Limit int
Zoom int
}
type MapReportClusterRecord struct {
ClusterID string
Latitude float64
Longitude float64
Count int64
}
func (s *Store) Ping() error {
db, err := s.db.DB()
if err != nil {
return err
}
return db.Ping()
}
func NormalizeListOptions(opts ListOptions) ListOptions {
if opts.Limit <= 0 {
opts.Limit = 100
}
if opts.Limit > 500 {
opts.Limit = 500
}
if opts.Offset < 0 {
opts.Offset = 0
}
return opts
}
func (s *Store) ListNodeInfo(opts ListOptions) ([]NodeInfoRecord, error) {
opts = NormalizeListOptions(opts)
var rows []NodeInfoRecord
q := applyNodeFilters(s.db.Model(&NodeInfoRecord{}), opts).
Order("updated_at DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
func (s *Store) CountNodeInfo(opts ListOptions) (int64, error) {
var total int64
q := applyNodeFilters(s.db.Model(&NodeInfoRecord{}), opts)
return total, q.Count(&total).Error
}
func (s *Store) GetNodeInfo(nodeID string) (*NodeInfoRecord, error) {
var row NodeInfoRecord
if err := s.db.Where("node_id = ?", nodeID).Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) ListMapReports(opts ListOptions) ([]MapReportRecord, error) {
opts = NormalizeListOptions(opts)
var rows []MapReportRecord
q := applyMapReportFilters(s.db.Model(&MapReportRecord{}), opts).
Order("updated_at DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
func (s *Store) CountMapReports(opts ListOptions) (int64, error) {
var total int64
q := applyMapReportFilters(s.db.Model(&MapReportRecord{}), opts)
return total, q.Count(&total).Error
}
func (s *Store) GetMapReport(nodeID string) (*MapReportRecord, error) {
var row MapReportRecord
if err := s.db.Where("node_id = ?", nodeID).Take(&row).Error; err != nil {
return nil, err
}
return &row, nil
}
func (s *Store) ListMapReportViewport(opts MapReportViewportOptions) (*MapReportViewportResult, error) {
opts = NormalizeMapReportViewportOptions(opts)
total, err := s.CountMapReports(opts.ListOptions)
if err != nil {
return nil, err
}
result := &MapReportViewportResult{Total: total, Limit: opts.Limit, Zoom: opts.Zoom}
if total <= int64(opts.ClusterThreshold) {
var points []MapReportRecord
q := applyMapReportFilters(s.db.Model(&MapReportRecord{}), opts.ListOptions).
Order("updated_at DESC").
Limit(opts.Limit)
if err := q.Find(&points).Error; err != nil {
return nil, err
}
result.Mode = "points"
result.Points = points
return result, nil
}
clusters, err := s.ListMapReportClusters(opts)
if err != nil {
return nil, err
}
result.Mode = "clusters"
result.Clusters = clusters
return result, nil
}
func (s *Store) ListMapReportClusters(opts MapReportViewportOptions) ([]MapReportClusterRecord, error) {
opts = NormalizeMapReportViewportOptions(opts)
cellSize := mapReportClusterCellSize(opts.ListOptions, opts.TargetCells)
var rows []struct {
LatBucket int64
LngBucket int64
Latitude float64
Longitude float64
Count int64
}
q := applyMapReportFilters(s.db.Model(&MapReportRecord{}), opts.ListOptions).
Select("CAST((latitude + 90.0) / ? AS INTEGER) AS lat_bucket, CAST((longitude + 180.0) / ? AS INTEGER) AS lng_bucket, AVG(latitude) AS latitude, AVG(longitude) AS longitude, COUNT(*) AS count", cellSize, cellSize).
Group("lat_bucket, lng_bucket").
Order("count DESC").
Limit(opts.Limit)
if err := q.Scan(&rows).Error; err != nil {
return nil, err
}
clusters := make([]MapReportClusterRecord, 0, len(rows))
for _, row := range rows {
clusters = append(clusters, MapReportClusterRecord{
ClusterID: fmt.Sprintf("%d:%d", row.LatBucket, row.LngBucket),
Latitude: row.Latitude,
Longitude: row.Longitude,
Count: row.Count,
})
}
return clusters, nil
}
func NormalizeMapReportViewportOptions(opts MapReportViewportOptions) MapReportViewportOptions {
if opts.Limit <= 0 {
opts.Limit = 1000
}
if opts.Limit > 2000 {
opts.Limit = 2000
}
if opts.ClusterThreshold <= 0 {
opts.ClusterThreshold = 500
}
if opts.ClusterThreshold > 5000 {
opts.ClusterThreshold = 5000
}
if opts.TargetCells <= 0 {
opts.TargetCells = 64
}
if opts.TargetCells > 256 {
opts.TargetCells = 256
}
return opts
}
func mapReportClusterCellSize(opts ListOptions, targetCells int) float64 {
latSpan := 180.0
if opts.MinLat != nil && opts.MaxLat != nil {
latSpan = *opts.MaxLat - *opts.MinLat
}
lngSpan := 360.0
if opts.MinLng != nil && opts.MaxLng != nil {
if *opts.MinLng <= *opts.MaxLng {
lngSpan = *opts.MaxLng - *opts.MinLng
} else {
lngSpan = 180 - *opts.MinLng + *opts.MaxLng + 180
}
}
span := math.Max(latSpan, lngSpan)
cellSize := span / float64(targetCells)
if cellSize < 0.0001 {
cellSize = 0.0001
}
return cellSize
}
func (s *Store) DeleteNode(nodeID string) error {
return s.db.Transaction(func(tx *gorm.DB) error {
nodeResult := tx.Where("node_id = ?", nodeID).Delete(&NodeInfoRecord{})
if nodeResult.Error != nil {
return nodeResult.Error
}
reportResult := tx.Where("node_id = ?", nodeID).Delete(&MapReportRecord{})
if reportResult.Error != nil {
return reportResult.Error
}
if nodeResult.RowsAffected+reportResult.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
})
}
func applyNodeFilters(q *gorm.DB, opts ListOptions) *gorm.DB {
if opts.NodeID != "" {
q = q.Where("node_id = ?", opts.NodeID)
}
if opts.Since != nil {
q = q.Where("updated_at >= ?", *opts.Since)
}
if opts.Until != nil {
q = q.Where("updated_at <= ?", *opts.Until)
}
return q
}
func applyMapReportFilters(q *gorm.DB, opts ListOptions) *gorm.DB {
q = applyNodeFilters(q, opts)
if opts.MinLat != nil && opts.MaxLat != nil {
q = q.Where("latitude IS NOT NULL AND latitude >= ? AND latitude <= ?", *opts.MinLat, *opts.MaxLat)
}
if opts.MinLng != nil && opts.MaxLng != nil {
if *opts.MinLng <= *opts.MaxLng {
q = q.Where("longitude IS NOT NULL AND longitude >= ? AND longitude <= ?", *opts.MinLng, *opts.MaxLng)
} else {
q = q.Where("longitude IS NOT NULL AND (longitude >= ? OR longitude <= ?)", *opts.MinLng, *opts.MaxLng)
}
}
return q
}
func (s *Store) ListTextMessages(opts ListOptions) ([]TextMessageRecord, error) {
var rows []TextMessageRecord
return rows, s.listAppendRows(opts, &rows).Error
}
func (s *Store) ListDiscardDetails(opts ListOptions) ([]DiscardDetailsRecord, error) {
opts = NormalizeListOptions(opts)
var rows []DiscardDetailsRecord
q := applyDiscardDetailsFilters(s.db.Model(&DiscardDetailsRecord{}), opts).
Order("created_at DESC").
Order("id DESC").
Limit(opts.Limit).
Offset(opts.Offset)
return rows, q.Find(&rows).Error
}
func (s *Store) CountDiscardDetails(opts ListOptions) (int64, error) {
var total int64
q := applyDiscardDetailsFilters(s.db.Model(&DiscardDetailsRecord{}), opts)
return total, q.Count(&total).Error
}
func applyDiscardDetailsFilters(q *gorm.DB, opts ListOptions) *gorm.DB {
if opts.Since != nil {
q = q.Where("created_at >= ?", *opts.Since)
}
if opts.Until != nil {
q = q.Where("created_at <= ?", *opts.Until)
}
return q
}
func (s *Store) DeleteTextMessage(id uint64) error {
result := s.db.Where("id = ?", id).Delete(&TextMessageRecord{})
if result.Error != nil {
return result.Error
}
if result.RowsAffected == 0 {
return gorm.ErrRecordNotFound
}
return nil
}
func (s *Store) ListPositions(opts ListOptions) ([]PositionRecord, error) {
var rows []PositionRecord
return rows, s.listAppendRows(opts, &rows).Error
}
func (s *Store) ListTelemetry(opts ListOptions) ([]TelemetryRecord, error) {
var rows []TelemetryRecord
return rows, s.listAppendRows(opts, &rows).Error
}
func (s *Store) ListRouting(opts ListOptions) ([]RoutingRecord, error) {
var rows []RoutingRecord
return rows, s.listAppendRows(opts, &rows).Error
}
func (s *Store) ListTraceroute(opts ListOptions) ([]TracerouteRecord, error) {
var rows []TracerouteRecord
return rows, s.listAppendRows(opts, &rows).Error
}
func (s *Store) listAppendRows(opts ListOptions, dest any) *gorm.DB {
opts = NormalizeListOptions(opts)
q := s.db.Order("created_at DESC").Order("id DESC").Limit(opts.Limit).Offset(opts.Offset)
if opts.NodeID != "" {
q = q.Where("from_id = ?", opts.NodeID)
}
if opts.ChannelID != "" {
q = q.Where("channel_id = ?", opts.ChannelID)
}
if opts.Since != nil {
q = q.Where("created_at >= ?", *opts.Since)
}
if opts.Until != nil {
q = q.Where("created_at <= ?", *opts.Until)
}
return q.Find(dest)
}
+45
View File
@@ -0,0 +1,45 @@
package store
import (
"strings"
"golang.org/x/crypto/bcrypt"
)
// 测试 helper —— 为从 main 包搬过来的测试提供它们原本依赖的小写函数。
// 这些 helper 不暴露给生产代码使用;它们的行为应当与 main 包对应实现保持一致。
// verifyPassword 复刻 auth.go 中的 bcrypt 校验,用于 user_store 的测试。
func verifyPassword(hash, password string) bool {
return bcrypt.CompareHashAndPassword([]byte(hash), []byte(password)) == nil
}
// publicMapTileSourceDTO 复刻 admin_map_source_routes.go 中的同名函数,
// 仅供 map_source_store_test.go 验证 ProxyEnabled 时 URL 是否被改写。
// 这里返回 map[string]any 而非 gin.H 以避免引入 gin 依赖。
func publicMapTileSourceDTO(row MapTileSourceRecord) map[string]any {
urlTemplate := row.URLTemplate
if row.ProxyEnabled {
hash := row.URLTemplateHash
if hash == "" {
hash = MapTileSourceHash(row.URLTemplate)
}
urlTemplate = "/api/map/" + hash + "?x={x}&y={y}&z={z}"
}
return map[string]any{
"id": row.ID,
"name": row.Name,
"url_template": urlTemplate,
"attribution": row.Attribution,
"max_zoom": row.MaxZoom,
"enabled": row.Enabled,
"is_default": row.IsDefault,
"proxy_enabled": row.ProxyEnabled,
}
}
// newDBWriteQueue 是 db_write_queue_test.go 期望的旧名字。重新导出供测试使用。
var newDBWriteQueue = NewWriteQueue
// 让 strings 不会被 import-but-not-used(如果上面用不到,就算了——保留以应对将来扩展)
var _ = strings.TrimSpace
+99
View File
@@ -0,0 +1,99 @@
package store
import (
"errors"
"fmt"
"strings"
"time"
"gorm.io/gorm"
)
var ErrUserAlreadyExists = errors.New("user already exists")
func (s *Store) GetUserByUsername(username string) (*UserRecord, error) {
var user UserRecord
if err := s.db.Where("username = ?", username).Take(&user).Error; err != nil {
return nil, err
}
return &user, nil
}
func (s *Store) GetUserByID(id uint64) (*UserRecord, error) {
var user UserRecord
if err := s.db.Where("id = ?", id).Take(&user).Error; err != nil {
return nil, err
}
return &user, nil
}
func (s *Store) ListUsers() ([]UserRecord, error) {
var users []UserRecord
return users, s.db.Order("id ASC").Find(&users).Error
}
func (s *Store) CreateAdminUser(username, password string) (*UserRecord, error) {
username = strings.TrimSpace(username)
if username == "" {
return nil, fmt.Errorf("username is required")
}
if password == "" {
return nil, fmt.Errorf("password is required")
}
if _, err := s.GetUserByUsername(username); err == nil {
return nil, ErrUserAlreadyExists
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
return nil, err
}
hash, err := hashPassword(password)
if err != nil {
return nil, fmt.Errorf("hash user password: %w", err)
}
user := UserRecord{Username: username, PasswordHash: hash, Role: AdminRole}
if err := s.db.Create(&user).Error; err != nil {
return nil, err
}
return &user, nil
}
func (s *Store) UpdateUserPassword(id uint64, password string) (*UserRecord, error) {
if id == 0 {
return nil, fmt.Errorf("user id is required")
}
if password == "" {
return nil, fmt.Errorf("password is required")
}
user, err := s.GetUserByID(id)
if err != nil {
return nil, err
}
hash, err := hashPassword(password)
if err != nil {
return nil, fmt.Errorf("hash user password: %w", err)
}
if err := s.db.Model(&UserRecord{}).Where("id = ?", id).Updates(map[string]any{"password_hash": hash, "updated_at": time.Now()}).Error; err != nil {
return nil, err
}
user.PasswordHash = hash
return s.GetUserByID(id)
}
func (s *Store) EnsureDefaultAdmin(username, password string) error {
var existing UserRecord
err := s.db.Where("username = ?", username).Take(&existing).Error
if err == nil {
return nil
}
if !errors.Is(err, gorm.ErrRecordNotFound) {
return err
}
hash, err := hashPassword(password)
if err != nil {
return fmt.Errorf("hash admin password: %w", err)
}
user := UserRecord{Username: username, PasswordHash: hash, Role: AdminRole}
if err := s.db.Create(&user).Error; err != nil {
return fmt.Errorf("create default admin user: %w", err)
}
return nil
}