From c527a9fd9a4bfa31ab94f8d0791b69b62ab6946e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=97=A0=E9=97=BB=E9=A3=8E?= Date: Thu, 18 Jun 2026 14:23:56 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84=EF=BC=9A=E6=8B=86=E5=87=BA?= =?UTF-8?q?=20auth=20/=20blocking=20/=20runtimesettings=20/=20help=20/=20m?= =?UTF-8?q?qttforward=20=E5=8C=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 第二批：把根目录中纯逻辑领域文件（cache、service、admin route）按业务边界 迁到 internal/ 下的子包。各子包暴露 RegisterRoutes 给 web 包调用，根目录 只留下一行 bridge 文件保留旧名字别名。 新增包 - internal/auth/ SessionClaims / Manager / RequireAdmin / HashPassword / VerifyPassword / AdminUserResponse 等。原 auth.go 中 被两个 admin route 依赖的 sessionClaims 现在以 auth. SessionClaims 形式被它们 import；不再被锁在 main 包。 - internal/blocking/ Cache + RegisterRoutes，以前散在 blocking_cache.go 和 admin_blocking_routes.go 里。 - internal/runtimesettings/ Cache + RegisterRoutes。 - internal/help/ RenderMarkdown / RegisterPublicRoutes / RegisterAdminRoutes（拆分原来的 registerHelpRoutes 和 registerAdminHelpRoutes 两条入口）。 - internal/mqttforward/ Manager / Reloader / Stats / RegisterRoutes。 forwarder runner、循环抑制 cache 等运行时逻辑随之迁入。 - internal/webutil/ ParseListOptions / WriteListResponse[WithTotal] / ParseMapReportListOptions / ParseMapReportViewportOptions 以及 PtrString/PtrInt64/... 等指针解引用 helper。 以前散在 web.go 中，现在被各 admin route 子包共享， 避免 internal/blocking → internal/web → internal/blocking 的循环依赖。 - internal/store/testutil/ OpenStore(t) helper，让其它包测试零样板拿到 store。 根目录新增 bridge 文件 - blocking_bridge.go / runtime_settings_bridge.go / help_bridge.go / mqttforward_bridge.go：用 type alias + thin wrapper 把上述子包的导出 名映射到旧的小写名（blockingCache、registerAdminBlockingRoutes 等）， 让 main.go / web.go 等仍未迁出的文件无须改动。 修改 - auth.go 改为对 internal/auth 的 bridge；web.go 中 sessions.newCookie / clearCookie 改为 NewCookie / ClearCookie。 - main_test.go 中 BlockingViolationForRecord* 测试不再直接构造未导出字段， 改成走 store.CreateNodeBlocking → newBlockingCache 的真实路径。 - internal/mqttforward 把以前 *_store.go 中没有方法依赖的运行时类型 （forwarder runner、loop cache）和 admin route 一并归位；mqtt_status.go 暂时仍留在根目录（依赖 main 中的 mqttClientInfoFromClient）。 go build ./... / go test ./... 全部通过；测试数量未变。 Co-Authored-By: Claude --- auth.go | 148 ++--------- blocking_bridge.go | 21 ++ help_bridge.go | 16 ++ internal/auth/auth.go | 169 +++++++++++++ .../blocking/admin_blocking_routes.go | 35 +-- .../blocking/blocking_cache.go | 22 +- .../blocking/blocking_cache_test.go | 14 +- internal/blocking/test_helpers_test.go | 13 + .../help/admin_help_routes.go | 21 +- .../help/help_markdown.go | 6 +- .../mqttforward/admin_mqtt_forward_routes.go | 43 ++-- .../mqttforward/filter_stats.go | 12 +- .../mqttforward/mqtt_forward_manager.go | 81 +++--- .../admin_runtime_settings_routes.go | 25 +- .../runtimesettings/runtime_settings_cache.go | 55 ++++ .../runtime_settings_cache_test.go | 21 +- internal/store/testutil/testutil.go | 27 ++ internal/webutil/webutil.go | 236 ++++++++++++++++++ main_test.go | 38 ++- mqttforward_bridge.go | 26 ++ runtime_settings_bridge.go | 19 ++ runtime_settings_cache.go | 47 ---- web.go | 4 +- 23 files changed, 786 insertions(+), 313 deletions(-) create mode 100644 blocking_bridge.go create mode 100644 help_bridge.go create mode 100644 internal/auth/auth.go rename admin_blocking_routes.go => internal/blocking/admin_blocking_routes.go (84%) rename blocking_cache.go => internal/blocking/blocking_cache.go (87%) rename blocking_cache_test.go => internal/blocking/blocking_cache_test.go (92%) create mode 100644 internal/blocking/test_helpers_test.go rename admin_help_routes.go => internal/help/admin_help_routes.go (76%) rename help_markdown.go => internal/help/help_markdown.go (70%) rename admin_mqtt_forward_routes.go => internal/mqttforward/admin_mqtt_forward_routes.go (82%) rename filter_stats.go => internal/mqttforward/filter_stats.go (51%) rename mqtt_forward_manager.go => internal/mqttforward/mqtt_forward_manager.go (79%) rename admin_runtime_settings_routes.go => internal/runtimesettings/admin_runtime_settings_routes.go (58%) create mode 100644 internal/runtimesettings/runtime_settings_cache.go rename runtime_settings_cache_test.go => internal/runtimesettings/runtime_settings_cache_test.go (57%) create mode 100644 internal/store/testutil/testutil.go create mode 100644 internal/webutil/webutil.go create mode 100644 mqttforward_bridge.go create mode 100644 runtime_settings_bridge.go delete mode 100644 runtime_settings_cache.go diff --git a/auth.go b/auth.go index f924ac4..f946e78 100644 --- a/auth.go +++ b/auth.go @@ -1,147 +1,29 @@ package main +// 桥接到 internal/auth — 让根目录其余文件继续用旧的小写名（sessionManager、 +// sessionClaims、newSessionManager、requireAdmin、verifyPassword 等）。 + import ( - "crypto/hmac" - "crypto/rand" - "crypto/sha256" - "encoding/base64" - "encoding/json" - "errors" - "fmt" - "net/http" - "strings" - "time" - "github.com/gin-gonic/gin" - "golang.org/x/crypto/bcrypt" + + authpkg "meshtastic_mqtt_server/internal/auth" ) -const ( - adminRole = "admin" - adminSessionCookie = "mesh_admin_session" +// 类型别名 +type ( + sessionManager = authpkg.Manager + sessionClaims = authpkg.SessionClaims + adminUserDTO = authpkg.AdminUserDTO ) -type adminUserDTO struct { - Username string `json:"username"` - Role string `json:"role"` -} - -type sessionClaims struct { - UserID uint64 `json:"user_id"` - Username string `json:"username"` - Role string `json:"role"` - Expires int64 `json:"expires"` -} - -type sessionManager struct { - secret []byte - secure bool - ttl time.Duration -} +const adminRole = authpkg.AdminRole func newSessionManager(cfg webAdminConfig) (*sessionManager, error) { - secret := strings.TrimSpace(cfg.SessionSecret) - if secret == "" { - generated := make([]byte, 32) - if _, err := rand.Read(generated); err != nil { - return nil, fmt.Errorf("generate admin session secret: %w", err) - } - return &sessionManager{secret: generated, secure: cfg.SessionSecure, ttl: 24 * time.Hour}, nil - } - return &sessionManager{secret: []byte(secret), secure: cfg.SessionSecure, ttl: 24 * time.Hour}, nil + return authpkg.NewManager(cfg) } -func hashPassword(password string) (string, error) { - hash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost) - if err != nil { - return "", err - } - return string(hash), nil -} +func verifyPassword(hash, password string) bool { return authpkg.VerifyPassword(hash, password) } -func verifyPassword(hash, password string) bool { - return bcrypt.CompareHashAndPassword([]byte(hash), []byte(password)) == nil -} +func adminUserResponse(user userRecord) adminUserDTO { return authpkg.AdminUserResponse(user) } -func adminUserResponse(user userRecord) adminUserDTO { - return adminUserDTO{Username: user.Username, Role: user.Role} -} - -func (sm *sessionManager) newCookie(user userRecord) (*http.Cookie, error) { - claims := sessionClaims{UserID: user.ID, Username: user.Username, Role: user.Role, Expires: time.Now().Add(sm.ttl).Unix()} - data, err := json.Marshal(claims) - if err != nil { - return nil, err - } - payload := base64.RawURLEncoding.EncodeToString(data) - signature := sm.sign(payload) - return &http.Cookie{ - Name: adminSessionCookie, - Value: payload + "." + signature, - Path: "/", - MaxAge: int(sm.ttl.Seconds()), - HttpOnly: true, - Secure: sm.secure, - SameSite: http.SameSiteLaxMode, - }, nil -} - -func (sm *sessionManager) clearCookie() *http.Cookie { - return &http.Cookie{ - Name: adminSessionCookie, - Value: "", - Path: "/", - MaxAge: -1, - HttpOnly: true, - Secure: sm.secure, - SameSite: http.SameSiteLaxMode, - } -} - -func (sm *sessionManager) claimsFromRequest(c *gin.Context) (*sessionClaims, error) { - cookie, err := c.Cookie(adminSessionCookie) - if err != nil { - return nil, err - } - parts := strings.Split(cookie, ".") - if len(parts) != 2 { - return nil, errors.New("invalid session") - } - if !hmac.Equal([]byte(parts[1]), []byte(sm.sign(parts[0]))) { - return nil, errors.New("invalid session signature") - } - data, err := base64.RawURLEncoding.DecodeString(parts[0]) - if err != nil { - return nil, err - } - var claims sessionClaims - if err := json.Unmarshal(data, &claims); err != nil { - return nil, err - } - if claims.Expires <= time.Now().Unix() { - return nil, errors.New("session expired") - } - if claims.Role != adminRole { - return nil, errors.New("admin required") - } - return &claims, nil -} - -func (sm *sessionManager) sign(payload string) string { - mac := hmac.New(sha256.New, sm.secret) - mac.Write([]byte(payload)) - return base64.RawURLEncoding.EncodeToString(mac.Sum(nil)) -} - -func requireAdmin(sm *sessionManager) gin.HandlerFunc { - return func(c *gin.Context) { - claims, err := sm.claimsFromRequest(c) - if err != nil { - c.JSON(http.StatusUnauthorized, gin.H{"error": "admin login required"}) - c.Abort() - return - } - c.Set("admin_claims", claims) - c.Next() - } -} +func requireAdmin(sm *sessionManager) gin.HandlerFunc { return authpkg.RequireAdmin(sm) } diff --git a/blocking_bridge.go b/blocking_bridge.go new file mode 100644 index 0000000..eee61e9 --- /dev/null +++ b/blocking_bridge.go @@ -0,0 +1,21 @@ +package main + +// 桥接到 internal/blocking — 让根目录其余文件可以继续使用旧名字 +// blockingCache / newBlockingCache / registerAdminBlockingRoutes， +// 而无须改动 main.go / web.go 等十几处调用点。 + +import ( + "github.com/gin-gonic/gin" + + blockingpkg "meshtastic_mqtt_server/internal/blocking" +) + +type blockingCache = blockingpkg.Cache + +func newBlockingCache(s *store) (*blockingCache, error) { + return blockingpkg.New(s) +} + +func registerAdminBlockingRoutes(r gin.IRouter, s *store, b *blockingCache) { + blockingpkg.RegisterRoutes(r, s, b) +} diff --git a/help_bridge.go b/help_bridge.go new file mode 100644 index 0000000..82ae746 --- /dev/null +++ b/help_bridge.go @@ -0,0 +1,16 @@ +package main + +// 桥接到 internal/help — 让 web.go 中的 registerHelpRoutes / +// registerAdminHelpRoutes / renderHelpMarkdown 旧名字仍可用。 + +import ( + "github.com/gin-gonic/gin" + + helppkg "meshtastic_mqtt_server/internal/help" +) + +func registerHelpRoutes(r gin.IRouter, s *store) { helppkg.RegisterPublicRoutes(r, s) } +func registerAdminHelpRoutes(r gin.IRouter, s *store) { helppkg.RegisterAdminRoutes(r, s) } +func renderHelpMarkdown(markdown string) (string, error) { + return helppkg.RenderMarkdown(markdown) +} diff --git a/internal/auth/auth.go b/internal/auth/auth.go new file mode 100644 index 0000000..03ad6d8 --- /dev/null +++ b/internal/auth/auth.go @@ -0,0 +1,169 @@ +// Package auth 实现 admin 端 cookie session 与密码散列。 +// +// 拆离自原来 main 包的 auth.go：让所有 admin route 包都可以直接 import 这里的 +// SessionClaims / Manager / RequireAdmin，而不是被锁在根 main 包里。 +package auth + +import ( + "crypto/hmac" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + "time" + + "github.com/gin-gonic/gin" + "golang.org/x/crypto/bcrypt" + + "meshtastic_mqtt_server/internal/config" + "meshtastic_mqtt_server/internal/store" +) + +const ( + AdminRole = "admin" + adminSessionCookie = "mesh_admin_session" + + // AdminClaimsKey 是中间件挂在 gin.Context 上的 key，handler 可用 + // `c.MustGet(auth.AdminClaimsKey).(*auth.SessionClaims)` 取出。 + AdminClaimsKey = "admin_claims" +) + +// AdminUserDTO 是 /me /login 等接口返回给前端的最小用户视图。 +type AdminUserDTO struct { + Username string `json:"username"` + Role string `json:"role"` +} + +// SessionClaims 是 cookie 中持久化的会话内容。 +type SessionClaims struct { + UserID uint64 `json:"user_id"` + Username string `json:"username"` + Role string `json:"role"` + Expires int64 `json:"expires"` +} + +// Manager 持有签名密钥与 cookie 配置，是发布 / 校验 cookie 的入口。 +type Manager struct { + secret []byte + secure bool + ttl time.Duration +} + +// NewManager 根据配置构造 Manager。如果 SessionSecret 空，会随机生成 32 字节。 +func NewManager(cfg config.WebAdminConfig) (*Manager, error) { + secret := strings.TrimSpace(cfg.SessionSecret) + if secret == "" { + generated := make([]byte, 32) + if _, err := rand.Read(generated); err != nil { + return nil, fmt.Errorf("generate admin session secret: %w", err) + } + return &Manager{secret: generated, secure: cfg.SessionSecure, ttl: 24 * time.Hour}, nil + } + return &Manager{secret: []byte(secret), secure: cfg.SessionSecure, ttl: 24 * time.Hour}, nil +} + +// HashPassword 用 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 +} + +// VerifyPassword 校验明文密码是否与散列匹配。 +func VerifyPassword(hash, password string) bool { + return bcrypt.CompareHashAndPassword([]byte(hash), []byte(password)) == nil +} + +// AdminUserResponse 把 store.UserRecord 转成对外 DTO。 +func AdminUserResponse(user store.UserRecord) AdminUserDTO { + return AdminUserDTO{Username: user.Username, Role: user.Role} +} + +// NewCookie 为已登录用户构造一份带签名的 session cookie。 +func (sm *Manager) NewCookie(user store.UserRecord) (*http.Cookie, error) { + claims := SessionClaims{UserID: user.ID, Username: user.Username, Role: user.Role, Expires: time.Now().Add(sm.ttl).Unix()} + data, err := json.Marshal(claims) + if err != nil { + return nil, err + } + payload := base64.RawURLEncoding.EncodeToString(data) + signature := sm.sign(payload) + return &http.Cookie{ + Name: adminSessionCookie, + Value: payload + "." + signature, + Path: "/", + MaxAge: int(sm.ttl.Seconds()), + HttpOnly: true, + Secure: sm.secure, + SameSite: http.SameSiteLaxMode, + }, nil +} + +// ClearCookie 返回一个把 cookie 立即清掉的 *http.Cookie，供 logout 使用。 +func (sm *Manager) ClearCookie() *http.Cookie { + return &http.Cookie{ + Name: adminSessionCookie, + Value: "", + Path: "/", + MaxAge: -1, + HttpOnly: true, + Secure: sm.secure, + SameSite: http.SameSiteLaxMode, + } +} + +// ClaimsFromRequest 校验请求 cookie 的签名 / 过期 / 角色。 +func (sm *Manager) ClaimsFromRequest(c *gin.Context) (*SessionClaims, error) { + cookie, err := c.Cookie(adminSessionCookie) + if err != nil { + return nil, err + } + parts := strings.Split(cookie, ".") + if len(parts) != 2 { + return nil, errors.New("invalid session") + } + if !hmac.Equal([]byte(parts[1]), []byte(sm.sign(parts[0]))) { + return nil, errors.New("invalid session signature") + } + data, err := base64.RawURLEncoding.DecodeString(parts[0]) + if err != nil { + return nil, err + } + var claims SessionClaims + if err := json.Unmarshal(data, &claims); err != nil { + return nil, err + } + if claims.Expires <= time.Now().Unix() { + return nil, errors.New("session expired") + } + if claims.Role != AdminRole { + return nil, errors.New("admin required") + } + return &claims, nil +} + +func (sm *Manager) sign(payload string) string { + mac := hmac.New(sha256.New, sm.secret) + mac.Write([]byte(payload)) + return base64.RawURLEncoding.EncodeToString(mac.Sum(nil)) +} + +// RequireAdmin 是把校验结果挂在 c.Set(AdminClaimsKey, claims) 上的中间件。 +func RequireAdmin(sm *Manager) gin.HandlerFunc { + return func(c *gin.Context) { + claims, err := sm.ClaimsFromRequest(c) + if err != nil { + c.JSON(http.StatusUnauthorized, gin.H{"error": "admin login required"}) + c.Abort() + return + } + c.Set(AdminClaimsKey, claims) + c.Next() + } +} diff --git a/admin_blocking_routes.go b/internal/blocking/admin_blocking_routes.go similarity index 84% rename from admin_blocking_routes.go rename to internal/blocking/admin_blocking_routes.go index 176d293..b09ee1c 100644 --- a/admin_blocking_routes.go +++ b/internal/blocking/admin_blocking_routes.go @@ -1,4 +1,4 @@ -package main +package blocking import ( "errors" @@ -7,6 +7,9 @@ import ( "github.com/gin-gonic/gin" "gorm.io/gorm" + + storepkg "meshtastic_mqtt_server/internal/store" + "meshtastic_mqtt_server/internal/webutil" ) type nodeBlockingRequest struct { @@ -30,7 +33,7 @@ type forbiddenWordBlockingRequest struct { Enabled bool `json:"enabled"` } -func registerAdminBlockingRoutes(r gin.IRouter, store *store, blocking *blockingCache) { +func RegisterRoutes(r gin.IRouter, store *storepkg.Store, blocking *Cache) { reloadBlocking := func() error { if blocking == nil { return nil @@ -39,17 +42,17 @@ func registerAdminBlockingRoutes(r gin.IRouter, store *store, blocking *blocking } r.GET("/blocking/nodes", func(c *gin.Context) { - opts, ok := parseListOptions(c) + opts, ok := webutil.ParseListOptions(c) if !ok { return } rows, err := store.ListNodeBlocking(opts) if err != nil { - writeListResponse(c, rows, opts, err, nodeBlockingDTO) + webutil.WriteListResponse(c, rows, opts, err, nodeBlockingDTO) return } total, err := store.CountNodeBlocking(opts) - writeListResponseWithTotal(c, rows, opts, total, err, nodeBlockingDTO) + webutil.WriteListResponseWithTotal(c, rows, opts, total, err, nodeBlockingDTO) }) r.POST("/blocking/nodes", func(c *gin.Context) { var req nodeBlockingRequest @@ -82,17 +85,17 @@ func registerAdminBlockingRoutes(r gin.IRouter, store *store, blocking *blocking }) r.GET("/blocking/ips", func(c *gin.Context) { - opts, ok := parseListOptions(c) + opts, ok := webutil.ParseListOptions(c) if !ok { return } rows, err := store.ListIPBlocking(opts) if err != nil { - writeListResponse(c, rows, opts, err, ipBlockingDTO) + webutil.WriteListResponse(c, rows, opts, err, ipBlockingDTO) return } total, err := store.CountIPBlocking(opts) - writeListResponseWithTotal(c, rows, opts, total, err, ipBlockingDTO) + webutil.WriteListResponseWithTotal(c, rows, opts, total, err, ipBlockingDTO) }) r.POST("/blocking/ips", func(c *gin.Context) { var req ipBlockingRequest @@ -125,17 +128,17 @@ func registerAdminBlockingRoutes(r gin.IRouter, store *store, blocking *blocking }) r.GET("/blocking/words", func(c *gin.Context) { - opts, ok := parseListOptions(c) + opts, ok := webutil.ParseListOptions(c) if !ok { return } rows, err := store.ListForbiddenWordBlocking(opts) if err != nil { - writeListResponse(c, rows, opts, err, forbiddenWordBlockingDTO) + webutil.WriteListResponse(c, rows, opts, err, forbiddenWordBlockingDTO) return } total, err := store.CountForbiddenWordBlocking(opts) - writeListResponseWithTotal(c, rows, opts, total, err, forbiddenWordBlockingDTO) + webutil.WriteListResponseWithTotal(c, rows, opts, total, err, forbiddenWordBlockingDTO) }) r.POST("/blocking/words", func(c *gin.Context) { var req forbiddenWordBlockingRequest @@ -178,7 +181,7 @@ func parseBlockingID(c *gin.Context) (uint64, bool) { } func writeBlockingMutationResponse[T any](c *gin.Context, status int, row *T, err error, convert func(T) gin.H, afterSuccess func() error) { - if errors.Is(err, errBlockingAlreadyExists) { + if errors.Is(err, storepkg.ErrBlockingAlreadyExists) { c.JSON(http.StatusConflict, gin.H{"error": "blocking rule already exists"}) return } @@ -217,14 +220,14 @@ func writeBlockingDeleteResponse(c *gin.Context, err error, afterSuccess func() c.JSON(http.StatusOK, gin.H{"status": "ok"}) } -func nodeBlockingDTO(row nodeBlockingRecord) gin.H { - return gin.H{"id": row.ID, "node_id": row.NodeID, "node_num": ptrInt64(row.NodeNum), "reason": row.Reason, "enabled": row.Enabled, "created_at": row.CreatedAt, "updated_at": row.UpdatedAt} +func nodeBlockingDTO(row storepkg.NodeBlockingRecord) gin.H { + return gin.H{"id": row.ID, "node_id": row.NodeID, "node_num": webutil.PtrInt64(row.NodeNum), "reason": row.Reason, "enabled": row.Enabled, "created_at": row.CreatedAt, "updated_at": row.UpdatedAt} } -func ipBlockingDTO(row ipBlockingRecord) gin.H { +func ipBlockingDTO(row storepkg.IPBlockingRecord) gin.H { return gin.H{"id": row.ID, "ip_value": row.IPValue, "reason": row.Reason, "enabled": row.Enabled, "created_at": row.CreatedAt, "updated_at": row.UpdatedAt} } -func forbiddenWordBlockingDTO(row forbiddenWordBlockingRecord) gin.H { +func forbiddenWordBlockingDTO(row storepkg.ForbiddenWordBlockingRecord) gin.H { return gin.H{"id": row.ID, "word": row.Word, "match_type": row.MatchType, "case_sensitive": row.CaseSensitive, "reason": row.Reason, "enabled": row.Enabled, "created_at": row.CreatedAt, "updated_at": row.UpdatedAt} } diff --git a/blocking_cache.go b/internal/blocking/blocking_cache.go similarity index 87% rename from blocking_cache.go rename to internal/blocking/blocking_cache.go index 48fe0a4..f2c686c 100644 --- a/blocking_cache.go +++ b/internal/blocking/blocking_cache.go @@ -1,4 +1,4 @@ -package main +package blocking import ( "fmt" @@ -6,9 +6,11 @@ import ( "strconv" "strings" "sync" + + storepkg "meshtastic_mqtt_server/internal/store" ) -type blockingCache struct { +type Cache struct { mu sync.RWMutex nodes map[string]struct{} nodeNums map[int64]struct{} @@ -24,15 +26,15 @@ type forbiddenWordRule struct { caseSensitive bool } -func newBlockingCache(store *store) (*blockingCache, error) { - cache := &blockingCache{} +func New(store *storepkg.Store) (*Cache, error) { + cache := &Cache{} if err := cache.Reload(store); err != nil { return nil, err } return cache, nil } -func (c *blockingCache) Reload(store *store) error { +func (c *Cache) Reload(store *storepkg.Store) error { if store == nil { return fmt.Errorf("store is required") } @@ -81,7 +83,7 @@ func (c *blockingCache) Reload(store *store) error { words := make([]forbiddenWordRule, 0, len(wordRows)) for _, row := range wordRows { word := strings.TrimSpace(row.Word) - if word == "" || row.MatchType != forbiddenWordMatchContains { + if word == "" || row.MatchType != storepkg.ForbiddenWordMatchContains { continue } words = append(words, forbiddenWordRule{word: word, foldedWord: strings.ToLower(word), matchType: row.MatchType, caseSensitive: row.CaseSensitive}) @@ -97,7 +99,7 @@ func (c *blockingCache) Reload(store *store) error { return nil } -func (c *blockingCache) IsNodeBlocked(nodeID any, nodeNum any) bool { +func (c *Cache) IsNodeBlocked(nodeID any, nodeNum any) bool { if c == nil { return false } @@ -118,7 +120,7 @@ func (c *blockingCache) IsNodeBlocked(nodeID any, nodeNum any) bool { return false } -func (c *blockingCache) IsIPBlocked(host string) bool { +func (c *Cache) IsIPBlocked(host string) bool { if c == nil { return false } @@ -144,7 +146,7 @@ func (c *blockingCache) IsIPBlocked(host string) bool { return false } -func (c *blockingCache) FindForbiddenWord(text any) (string, bool) { +func (c *Cache) FindForbiddenWord(text any) (string, bool) { if c == nil { return "", false } @@ -157,7 +159,7 @@ func (c *blockingCache) FindForbiddenWord(text any) (string, bool) { defer c.mu.RUnlock() foldedText := "" for _, rule := range c.words { - if rule.matchType != forbiddenWordMatchContains { + if rule.matchType != storepkg.ForbiddenWordMatchContains { continue } if rule.caseSensitive { diff --git a/blocking_cache_test.go b/internal/blocking/blocking_cache_test.go similarity index 92% rename from blocking_cache_test.go rename to internal/blocking/blocking_cache_test.go index 691c1cf..9ddffc4 100644 --- a/blocking_cache_test.go +++ b/internal/blocking/blocking_cache_test.go @@ -1,4 +1,4 @@ -package main +package blocking import "testing" @@ -27,9 +27,9 @@ func TestBlockingCacheLoadsEnabledRules(t *testing.T) { t.Fatalf("CreateForbiddenWordBlocking(disabled) error = %v", err) } - cache, err := newBlockingCache(st) + cache, err := New(st) if err != nil { - t.Fatalf("newBlockingCache() error = %v", err) + t.Fatalf("New() error = %v", err) } if !cache.IsNodeBlocked("!12345678", nil) { @@ -65,9 +65,9 @@ func TestBlockingCacheIPExactAndCIDR(t *testing.T) { if _, err := st.CreateIPBlocking("2001:db8::/32", "docs", true); err != nil { t.Fatalf("CreateIPBlocking(ipv6 cidr) error = %v", err) } - cache, err := newBlockingCache(st) + cache, err := New(st) if err != nil { - t.Fatalf("newBlockingCache() error = %v", err) + t.Fatalf("New() error = %v", err) } if !cache.IsIPBlocked("127.0.0.1") { @@ -88,9 +88,9 @@ func TestBlockingCacheForbiddenWordCaseSensitivity(t *testing.T) { if _, err := st.CreateForbiddenWordBlocking("Spam", "contains", true, "case-sensitive", true); err != nil { t.Fatalf("CreateForbiddenWordBlocking(case-sensitive) error = %v", err) } - cache, err := newBlockingCache(st) + cache, err := New(st) if err != nil { - t.Fatalf("newBlockingCache() error = %v", err) + t.Fatalf("New() error = %v", err) } if _, ok := cache.FindForbiddenWord("lowercase spam"); ok { diff --git a/internal/blocking/test_helpers_test.go b/internal/blocking/test_helpers_test.go new file mode 100644 index 0000000..d1a8309 --- /dev/null +++ b/internal/blocking/test_helpers_test.go @@ -0,0 +1,13 @@ +package blocking + +import ( + "testing" + + "meshtastic_mqtt_server/internal/store" + "meshtastic_mqtt_server/internal/store/testutil" +) + +// openTestStore 委托到 store/testutil，让本包的测试代码保持简洁。 +func openTestStore(t *testing.T) *store.Store { + return testutil.OpenStore(t) +} diff --git a/admin_help_routes.go b/internal/help/admin_help_routes.go similarity index 76% rename from admin_help_routes.go rename to internal/help/admin_help_routes.go index a70a109..014491c 100644 --- a/admin_help_routes.go +++ b/internal/help/admin_help_routes.go @@ -1,4 +1,4 @@ -package main +package help import ( "errors" @@ -7,13 +7,17 @@ import ( "github.com/gin-gonic/gin" "gorm.io/gorm" + + "meshtastic_mqtt_server/internal/auth" + storepkg "meshtastic_mqtt_server/internal/store" ) type helpContentRequest struct { Markdown string `json:"markdown"` } -func registerHelpRoutes(r gin.IRouter, store *store) { +// RegisterPublicRoutes 把对外可见的 GET /help 挂到给定路由组下。 +func RegisterPublicRoutes(r gin.IRouter, store *storepkg.Store) { r.GET("/help", func(c *gin.Context) { item, err := latestHelpContentDTO(store) if err != nil { @@ -24,7 +28,8 @@ func registerHelpRoutes(r gin.IRouter, store *store) { }) } -func registerAdminHelpRoutes(r gin.IRouter, store *store) { +// RegisterAdminRoutes 注册管理员侧 /help、/help、/help/preview 这三条路由。 +func RegisterAdminRoutes(r gin.IRouter, store *storepkg.Store) { r.GET("/help", func(c *gin.Context) { item, err := latestHelpContentDTO(store) if err != nil { @@ -39,7 +44,7 @@ func registerAdminHelpRoutes(r gin.IRouter, store *store) { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid help content request"}) return } - claims := c.MustGet("admin_claims").(*sessionClaims) + claims := c.MustGet(auth.AdminClaimsKey).(*auth.SessionClaims) row, err := store.InsertHelpContent(req.Markdown, claims.Username) if err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) @@ -58,7 +63,7 @@ func registerAdminHelpRoutes(r gin.IRouter, store *store) { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid help preview request"}) return } - html, err := renderHelpMarkdown(req.Markdown) + html, err := RenderMarkdown(req.Markdown) if err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return @@ -67,10 +72,10 @@ func registerAdminHelpRoutes(r gin.IRouter, store *store) { }) } -func latestHelpContentDTO(store *store) (gin.H, error) { +func latestHelpContentDTO(store *storepkg.Store) (gin.H, error) { row, err := store.GetLatestHelpContent() if errors.Is(err, gorm.ErrRecordNotFound) { - return helpContentDTO(0, defaultHelpMarkdown, "", nil) + return helpContentDTO(0, storepkg.DefaultHelpMarkdown, "", nil) } if err != nil { return nil, err @@ -79,7 +84,7 @@ func latestHelpContentDTO(store *store) (gin.H, error) { } func helpContentDTO(id uint64, markdown, createdBy string, createdAt *time.Time) (gin.H, error) { - html, err := renderHelpMarkdown(markdown) + html, err := RenderMarkdown(markdown) if err != nil { return nil, err } diff --git a/help_markdown.go b/internal/help/help_markdown.go similarity index 70% rename from help_markdown.go rename to internal/help/help_markdown.go index d0d752f..4d06285 100644 --- a/help_markdown.go +++ b/internal/help/help_markdown.go @@ -1,4 +1,4 @@ -package main +package help import ( "bytes" @@ -9,7 +9,9 @@ import ( "github.com/yuin/goldmark/extension" ) -func renderHelpMarkdown(markdown string) (string, error) { +// RenderMarkdown 把 GFM markdown 转成净化后的 HTML，供 admin 编辑器预览 +// 与 /help 路由直接渲染。 +func RenderMarkdown(markdown string) (string, error) { var buf bytes.Buffer md := goldmark.New(goldmark.WithExtensions(extension.GFM)) if err := md.Convert([]byte(markdown), &buf); err != nil { diff --git a/admin_mqtt_forward_routes.go b/internal/mqttforward/admin_mqtt_forward_routes.go similarity index 82% rename from admin_mqtt_forward_routes.go rename to internal/mqttforward/admin_mqtt_forward_routes.go index bf946ea..9bb3688 100644 --- a/admin_mqtt_forward_routes.go +++ b/internal/mqttforward/admin_mqtt_forward_routes.go @@ -1,4 +1,4 @@ -package main +package mqttforward import ( "errors" @@ -7,6 +7,9 @@ import ( "github.com/gin-gonic/gin" "gorm.io/gorm" + + storepkg "meshtastic_mqtt_server/internal/store" + "meshtastic_mqtt_server/internal/webutil" ) type mqttForwarderRequest struct { @@ -38,19 +41,19 @@ type mqttForwardTopicRequest struct { Retain bool `json:"retain"` } -func registerAdminMQTTForwardRoutes(r gin.IRouter, store *store, forwarder mqttForwardReloader) { +func RegisterRoutes(r gin.IRouter, store *storepkg.Store, forwarder Reloader) { r.GET("/mqtt-forward/forwarders", func(c *gin.Context) { - opts, ok := parseListOptions(c) + opts, ok := webutil.ParseListOptions(c) if !ok { return } rows, err := store.ListMQTTForwarders(opts) if err != nil { - writeListResponse(c, rows, opts, err, mqttForwarderDTO) + webutil.WriteListResponse(c, rows, opts, err, mqttForwarderDTO) return } total, err := store.CountMQTTForwarders(opts) - writeListResponseWithTotal(c, rows, opts, total, err, mqttForwarderDTO) + webutil.WriteListResponseWithTotal(c, rows, opts, total, err, mqttForwarderDTO) }) r.POST("/mqtt-forward/forwarders", func(c *gin.Context) { var req mqttForwarderRequest @@ -106,17 +109,17 @@ func registerAdminMQTTForwardRoutes(r gin.IRouter, store *store, forwarder mqttF if !ok { return } - opts, ok := parseListOptions(c) + opts, ok := webutil.ParseListOptions(c) if !ok { return } rows, err := store.ListMQTTForwardTopics(id, opts) if err != nil { - writeListResponse(c, rows, opts, err, mqttForwardTopicDTO) + webutil.WriteListResponse(c, rows, opts, err, mqttForwardTopicDTO) return } total, err := store.CountMQTTForwardTopics(id) - writeListResponseWithTotal(c, rows, opts, total, err, mqttForwardTopicDTO) + webutil.WriteListResponseWithTotal(c, rows, opts, total, err, mqttForwardTopicDTO) }) r.POST("/mqtt-forward/forwarders/:id/topics", func(c *gin.Context) { id, ok := parseMQTTForwardID(c, "invalid mqtt forwarder id") @@ -168,7 +171,7 @@ func registerAdminMQTTForwardRoutes(r gin.IRouter, store *store, forwarder mqttF }) }) r.GET("/mqtt-forward/status", func(c *gin.Context) { - items := []mqttForwardRuntimeStatus{} + items := []RuntimeStatus{} if forwarder != nil { items = forwarder.Status() } @@ -176,7 +179,7 @@ func registerAdminMQTTForwardRoutes(r gin.IRouter, store *store, forwarder mqttF }) } -func mqttForwarderInputFromRequest(req mqttForwarderRequest) mqttForwarderInput { +func mqttForwarderInputFromRequest(req mqttForwarderRequest) storepkg.MQTTForwarderInput { sourcePassword := req.SourcePassword if req.SourcePasswordClear { empty := "" @@ -187,11 +190,11 @@ func mqttForwarderInputFromRequest(req mqttForwarderRequest) mqttForwarderInput empty := "" targetPassword = &empty } - return mqttForwarderInput{Name: req.Name, Enabled: req.Enabled, SourceHost: req.SourceHost, SourcePort: req.SourcePort, SourceUsername: req.SourceUsername, SourcePassword: sourcePassword, SourceClientID: req.SourceClientID, SourceTLS: req.SourceTLS, TargetHost: req.TargetHost, TargetPort: req.TargetPort, TargetUsername: req.TargetUsername, TargetPassword: targetPassword, TargetClientID: req.TargetClientID, TargetTLS: req.TargetTLS} + return storepkg.MQTTForwarderInput{Name: req.Name, Enabled: req.Enabled, SourceHost: req.SourceHost, SourcePort: req.SourcePort, SourceUsername: req.SourceUsername, SourcePassword: sourcePassword, SourceClientID: req.SourceClientID, SourceTLS: req.SourceTLS, TargetHost: req.TargetHost, TargetPort: req.TargetPort, TargetUsername: req.TargetUsername, TargetPassword: targetPassword, TargetClientID: req.TargetClientID, TargetTLS: req.TargetTLS} } -func mqttForwardTopicInputFromRequest(req mqttForwardTopicRequest) mqttForwardTopicInput { - return mqttForwardTopicInput{Topic: req.Topic, Enabled: req.Enabled, Direction: req.Direction, SourcePrefix: req.SourcePrefix, TargetPrefix: req.TargetPrefix, QoS: req.QoS, Retain: req.Retain} +func mqttForwardTopicInputFromRequest(req mqttForwardTopicRequest) storepkg.MQTTForwardTopicInput { + return storepkg.MQTTForwardTopicInput{Topic: req.Topic, Enabled: req.Enabled, Direction: req.Direction, SourcePrefix: req.SourcePrefix, TargetPrefix: req.TargetPrefix, QoS: req.QoS, Retain: req.Retain} } func parseMQTTForwardID(c *gin.Context, message string) (uint64, bool) { @@ -203,15 +206,15 @@ func parseMQTTForwardID(c *gin.Context, message string) (uint64, bool) { return id, true } -func reloadMQTTForwarder(forwarder mqttForwardReloader, id uint64) error { +func reloadMQTTForwarder(forwarder Reloader, id uint64) error { if forwarder == nil { return nil } return forwarder.ReloadForwarder(id) } -func writeMQTTForwardMutationResponse(c *gin.Context, status int, row *mqttForwarderRecord, err error, afterSuccess func() error) { - if errors.Is(err, errMQTTForwarderAlreadyExists) { +func writeMQTTForwardMutationResponse(c *gin.Context, status int, row *storepkg.MQTTForwarderRecord, err error, afterSuccess func() error) { + if errors.Is(err, storepkg.ErrMQTTForwarderAlreadyExists) { c.JSON(http.StatusConflict, gin.H{"error": "mqtt forwarder already exists"}) return } @@ -232,8 +235,8 @@ func writeMQTTForwardMutationResponse(c *gin.Context, status int, row *mqttForwa c.JSON(status, gin.H{"item": mqttForwarderDTO(*row)}) } -func writeMQTTForwardTopicMutationResponse(c *gin.Context, status int, row *mqttForwardTopicRecord, err error, afterSuccess func() error) { - if errors.Is(err, errMQTTForwardTopicAlreadyExists) { +func writeMQTTForwardTopicMutationResponse(c *gin.Context, status int, row *storepkg.MQTTForwardTopicRecord, err error, afterSuccess func() error) { + if errors.Is(err, storepkg.ErrMQTTForwardTopicAlreadyExists) { c.JSON(http.StatusConflict, gin.H{"error": "mqtt forward topic already exists"}) return } @@ -272,10 +275,10 @@ func writeMQTTForwardDeleteResponse(c *gin.Context, err error, afterSuccess func c.JSON(http.StatusOK, gin.H{"status": "ok"}) } -func mqttForwarderDTO(row mqttForwarderRecord) gin.H { +func mqttForwarderDTO(row storepkg.MQTTForwarderRecord) gin.H { return gin.H{"id": row.ID, "name": row.Name, "enabled": row.Enabled, "source_host": row.SourceHost, "source_port": row.SourcePort, "source_username": row.SourceUsername, "source_password_set": 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_set": row.TargetPassword != "", "target_client_id": row.TargetClientID, "target_tls": row.TargetTLS, "created_at": row.CreatedAt, "updated_at": row.UpdatedAt} } -func mqttForwardTopicDTO(row mqttForwardTopicRecord) gin.H { +func mqttForwardTopicDTO(row storepkg.MQTTForwardTopicRecord) gin.H { return gin.H{"id": row.ID, "forwarder_id": row.ForwarderID, "topic": row.Topic, "enabled": row.Enabled, "direction": row.Direction, "source_prefix": row.SourcePrefix, "target_prefix": row.TargetPrefix, "qos": row.QoS, "retain": row.Retain, "created_at": row.CreatedAt, "updated_at": row.UpdatedAt} } diff --git a/filter_stats.go b/internal/mqttforward/filter_stats.go similarity index 51% rename from filter_stats.go rename to internal/mqttforward/filter_stats.go index 27ba54f..e5d38ff 100644 --- a/filter_stats.go +++ b/internal/mqttforward/filter_stats.go @@ -1,32 +1,32 @@ -package main +package mqttforward import "sync/atomic" -type meshtasticMessageStats struct { +type Stats struct { forwarded atomic.Int64 dropped atomic.Int64 } -func (s *meshtasticMessageStats) IncForwarded() { +func (s *Stats) IncForwarded() { if s != nil { s.forwarded.Add(1) } } -func (s *meshtasticMessageStats) IncDropped() { +func (s *Stats) IncDropped() { if s != nil { s.dropped.Add(1) } } -func (s *meshtasticMessageStats) Forwarded() int64 { +func (s *Stats) Forwarded() int64 { if s == nil { return 0 } return s.forwarded.Load() } -func (s *meshtasticMessageStats) Dropped() int64 { +func (s *Stats) Dropped() int64 { if s == nil { return 0 } diff --git a/mqtt_forward_manager.go b/internal/mqttforward/mqtt_forward_manager.go similarity index 79% rename from mqtt_forward_manager.go rename to internal/mqttforward/mqtt_forward_manager.go index 83e31fa..6596879 100644 --- a/mqtt_forward_manager.go +++ b/internal/mqttforward/mqtt_forward_manager.go @@ -1,6 +1,7 @@ -package main +package mqttforward import ( + storepkg "meshtastic_mqtt_server/internal/store" "context" "crypto/sha256" "crypto/tls" @@ -23,19 +24,19 @@ const ( mqttForwardLoopMaxEntries = 10000 ) -type mqttForwardReloader interface { +type Reloader interface { ReloadForwarder(id uint64) error StopForwarder(id uint64) - Status() []mqttForwardRuntimeStatus + Status() []RuntimeStatus } -type mqttForwardManager struct { - store *store +type Manager struct { + store *storepkg.Store mu sync.Mutex - runners map[uint64]*mqttForwardRunner + runners map[uint64]*runner } -type mqttForwardRuntimeStatus struct { +type RuntimeStatus struct { ForwarderID uint64 `json:"forwarder_id"` Running bool `json:"running"` SourceConnected bool `json:"source_connected"` @@ -46,8 +47,8 @@ type mqttForwardRuntimeStatus struct { MessagesDropped uint64 `json:"messages_dropped"` } -type mqttForwardRunner struct { - config mqttForwarderConfig +type runner struct { + config storepkg.MQTTForwarderConfig ctx context.Context cancel context.CancelFunc source pahomqtt.Client @@ -63,11 +64,11 @@ type mqttForwardRunner struct { loopCache map[string]time.Time } -func newMQTTForwardManager(store *store) *mqttForwardManager { - return &mqttForwardManager{store: store, runners: make(map[uint64]*mqttForwardRunner)} +func NewManager(store *storepkg.Store) *Manager { + return &Manager{store: store, runners: make(map[uint64]*runner)} } -func (m *mqttForwardManager) StartFromStore() error { +func (m *Manager) StartFromStore() error { configs, err := m.store.ListEnabledMQTTForwarderConfigs() if err != nil { return err @@ -76,7 +77,7 @@ func (m *mqttForwardManager) StartFromStore() error { if len(cfg.Topics) == 0 { continue } - runner := newMQTTForwardRunner(cfg) + runner := newRunner(cfg) runner.Start() m.mu.Lock() m.runners[cfg.Forwarder.ID] = runner @@ -85,7 +86,7 @@ func (m *mqttForwardManager) StartFromStore() error { return nil } -func (m *mqttForwardManager) ReloadForwarder(id uint64) error { +func (m *Manager) ReloadForwarder(id uint64) error { m.StopForwarder(id) cfg, err := m.store.GetMQTTForwarderConfig(id) if errors.Is(err, gorm.ErrRecordNotFound) { @@ -97,7 +98,7 @@ func (m *mqttForwardManager) ReloadForwarder(id uint64) error { if !cfg.Forwarder.Enabled || len(cfg.Topics) == 0 { return nil } - runner := newMQTTForwardRunner(*cfg) + runner := newRunner(*cfg) runner.Start() m.mu.Lock() m.runners[id] = runner @@ -105,7 +106,7 @@ func (m *mqttForwardManager) ReloadForwarder(id uint64) error { return nil } -func (m *mqttForwardManager) StopForwarder(id uint64) { +func (m *Manager) StopForwarder(id uint64) { m.mu.Lock() runner := m.runners[id] delete(m.runners, id) @@ -115,9 +116,9 @@ func (m *mqttForwardManager) StopForwarder(id uint64) { } } -func (m *mqttForwardManager) StopAll() { +func (m *Manager) StopAll() { m.mu.Lock() - runners := make([]*mqttForwardRunner, 0, len(m.runners)) + runners := make([]*runner, 0, len(m.runners)) for id, runner := range m.runners { runners = append(runners, runner) delete(m.runners, id) @@ -128,14 +129,14 @@ func (m *mqttForwardManager) StopAll() { } } -func (m *mqttForwardManager) Status() []mqttForwardRuntimeStatus { +func (m *Manager) Status() []RuntimeStatus { m.mu.Lock() - runners := make([]*mqttForwardRunner, 0, len(m.runners)) + runners := make([]*runner, 0, len(m.runners)) for _, runner := range m.runners { runners = append(runners, runner) } m.mu.Unlock() - items := make([]mqttForwardRuntimeStatus, 0, len(runners)) + items := make([]RuntimeStatus, 0, len(runners)) for _, runner := range runners { items = append(items, runner.Status()) } @@ -143,19 +144,19 @@ func (m *mqttForwardManager) Status() []mqttForwardRuntimeStatus { return items } -func newMQTTForwardRunner(config mqttForwarderConfig) *mqttForwardRunner { +func newRunner(config storepkg.MQTTForwarderConfig) *runner { ctx, cancel := context.WithCancel(context.Background()) - return &mqttForwardRunner{config: config, ctx: ctx, cancel: cancel, startedAt: time.Now(), loopCache: make(map[string]time.Time)} + return &runner{config: config, ctx: ctx, cancel: cancel, startedAt: time.Now(), loopCache: make(map[string]time.Time)} } -func (r *mqttForwardRunner) Start() { +func (r *runner) Start() { r.source = r.newClient(true) r.target = r.newClient(false) r.connectClient(r.target, "target") r.connectClient(r.source, "source") } -func (r *mqttForwardRunner) Stop() { +func (r *runner) Stop() { r.cancel() if r.source != nil && r.source.IsConnected() { r.source.Disconnect(250) @@ -165,11 +166,11 @@ func (r *mqttForwardRunner) Stop() { } } -func (r *mqttForwardRunner) Status() mqttForwardRuntimeStatus { +func (r *runner) Status() RuntimeStatus { r.mu.Lock() defer r.mu.Unlock() started := r.startedAt - return mqttForwardRuntimeStatus{ + return RuntimeStatus{ ForwarderID: r.config.Forwarder.ID, Running: true, SourceConnected: r.sourceConnected, @@ -181,7 +182,7 @@ func (r *mqttForwardRunner) Status() mqttForwardRuntimeStatus { } } -func (r *mqttForwardRunner) newClient(source bool) pahomqtt.Client { +func (r *runner) newClient(source bool) pahomqtt.Client { forwarder := r.config.Forwarder host, port, username, password, clientID, useTLS := forwarder.SourceHost, forwarder.SourcePort, forwarder.SourceUsername, forwarder.SourcePassword, forwarder.SourceClientID, forwarder.SourceTLS role := "source" @@ -222,7 +223,7 @@ func (r *mqttForwardRunner) newClient(source bool) pahomqtt.Client { return pahomqtt.NewClient(opts) } -func (r *mqttForwardRunner) connectClient(client pahomqtt.Client, label string) { +func (r *runner) connectClient(client pahomqtt.Client, label string) { token := client.Connect() if !token.WaitTimeout(2 * time.Second) { r.setError(label + " connect pending") @@ -233,11 +234,11 @@ func (r *mqttForwardRunner) connectClient(client pahomqtt.Client, label string) } } -func (r *mqttForwardRunner) subscribe(client pahomqtt.Client, source bool) { +func (r *runner) subscribe(client pahomqtt.Client, source bool) { for _, topic := range r.config.Topics { filter := topic.Topic if !source { - if topic.Direction != mqttForwardDirectionBidirectional { + if topic.Direction != storepkg.MQTTForwardDirectionBidirectional { continue } filter = mapMQTTForwardTopic(topic.Topic, topic.SourcePrefix, topic.TargetPrefix) @@ -256,7 +257,7 @@ func (r *mqttForwardRunner) subscribe(client pahomqtt.Client, source bool) { } } -func (r *mqttForwardRunner) forwardMessage(fromSource bool, rule mqttForwardTopicRecord, msg pahomqtt.Message) { +func (r *runner) forwardMessage(fromSource bool, rule storepkg.MQTTForwardTopicRecord, msg pahomqtt.Message) { if r.ctx.Err() != nil { return } @@ -269,7 +270,7 @@ func (r *mqttForwardRunner) forwardMessage(fromSource bool, rule mqttForwardTopi return } toTopic := fromTopic - forwardDirection := mqttForwardDirectionSourceToTarget + forwardDirection := storepkg.MQTTForwardDirectionSourceToTarget if fromSource { toTopic = mapMQTTForwardTopic(fromTopic, rule.SourcePrefix, rule.TargetPrefix) } else { @@ -284,7 +285,7 @@ func (r *mqttForwardRunner) forwardMessage(fromSource bool, rule mqttForwardTopi reverseDirection := mqttForwardDirectionTargetToSource if !fromSource { target = r.source - reverseDirection = mqttForwardDirectionSourceToTarget + reverseDirection = storepkg.MQTTForwardDirectionSourceToTarget } r.markSuppressed(reverseDirection, toTopic, fromTopic, msg.Payload(), rule.QoS, rule.Retain) token := target.Publish(toTopic, byte(rule.QoS), rule.Retain, msg.Payload()) @@ -301,7 +302,7 @@ func (r *mqttForwardRunner) forwardMessage(fromSource bool, rule mqttForwardTopi r.incForwarded() } -func (r *mqttForwardRunner) setConnected(source bool, connected bool) { +func (r *runner) setConnected(source bool, connected bool) { r.mu.Lock() defer r.mu.Unlock() if source { @@ -311,25 +312,25 @@ func (r *mqttForwardRunner) setConnected(source bool, connected bool) { } } -func (r *mqttForwardRunner) setError(message string) { +func (r *runner) setError(message string) { r.mu.Lock() r.lastError = message r.mu.Unlock() } -func (r *mqttForwardRunner) incForwarded() { +func (r *runner) incForwarded() { r.mu.Lock() r.messagesForwarded++ r.mu.Unlock() } -func (r *mqttForwardRunner) incDropped() { +func (r *runner) incDropped() { r.mu.Lock() r.messagesDropped++ r.mu.Unlock() } -func (r *mqttForwardRunner) isSuppressed(direction, fromTopic, toTopic string, payload []byte, qos int, retain bool) bool { +func (r *runner) isSuppressed(direction, fromTopic, toTopic string, payload []byte, qos int, retain bool) bool { key := mqttForwardLoopKey(direction, fromTopic, toTopic, payload, qos, retain) now := time.Now() r.mu.Lock() @@ -346,7 +347,7 @@ func (r *mqttForwardRunner) isSuppressed(direction, fromTopic, toTopic string, p return true } -func (r *mqttForwardRunner) markSuppressed(direction, fromTopic, toTopic string, payload []byte, qos int, retain bool) { +func (r *runner) markSuppressed(direction, fromTopic, toTopic string, payload []byte, qos int, retain bool) { key := mqttForwardLoopKey(direction, fromTopic, toTopic, payload, qos, retain) now := time.Now() r.mu.Lock() diff --git a/admin_runtime_settings_routes.go b/internal/runtimesettings/admin_runtime_settings_routes.go similarity index 58% rename from admin_runtime_settings_routes.go rename to internal/runtimesettings/admin_runtime_settings_routes.go index 046e06c..09ac865 100644 --- a/admin_runtime_settings_routes.go +++ b/internal/runtimesettings/admin_runtime_settings_routes.go @@ -1,9 +1,11 @@ -package main +package runtimesettings import ( "net/http" "github.com/gin-gonic/gin" + + storepkg "meshtastic_mqtt_server/internal/store" ) const allowEncryptedForwardingLabel = "Allow encrypted MQTT packets to be forwarded when they cannot be decrypted" @@ -11,12 +13,13 @@ const llmQueueEnabledLabel = "Enable LLM message queue" const llmIncludeChannelLabel = "Include channel messages in LLM queue" type runtimeSettingsRequest struct { - AllowEncryptedForwarding bool `json:"allow_encrypted_forwarding"` - LLMQueueEnabled bool `json:"llm_queue_enabled"` + AllowEncryptedForwarding bool `json:"allow_encrypted_forwarding"` + LLMQueueEnabled bool `json:"llm_queue_enabled"` LLMIncludeChannelMessages bool `json:"llm_include_channel_messages"` } -func registerAdminRuntimeSettingsRoutes(r gin.IRouter, store *store, settings *runtimeSettingsCache) { +// RegisterRoutes 把 GET /runtime-settings 与 PUT /runtime-settings 挂到给定路由组下。 +func RegisterRoutes(r gin.IRouter, store *storepkg.Store, settings *Cache) { r.GET("/runtime-settings", func(c *gin.Context) { snapshot, err := store.GetRuntimeSettings() if err != nil { @@ -32,15 +35,15 @@ func registerAdminRuntimeSettingsRoutes(r gin.IRouter, store *store, settings *r c.JSON(http.StatusBadRequest, gin.H{"error": "invalid runtime settings request"}) return } - if _, err := store.SetBoolRuntimeSetting(runtimeSettingAllowEncryptedForwarding, req.AllowEncryptedForwarding, allowEncryptedForwardingLabel); err != nil { + if _, err := store.SetBoolRuntimeSetting(storepkg.RuntimeSettingAllowEncryptedForwarding, req.AllowEncryptedForwarding, allowEncryptedForwardingLabel); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } - if _, err := store.SetBoolRuntimeSetting(runtimeSettingLLMQueueEnabled, req.LLMQueueEnabled, llmQueueEnabledLabel); err != nil { + if _, err := store.SetBoolRuntimeSetting(storepkg.RuntimeSettingLLMQueueEnabled, req.LLMQueueEnabled, llmQueueEnabledLabel); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } - if _, err := store.SetBoolRuntimeSetting(runtimeSettingLLMQueueIncludeChannel, req.LLMIncludeChannelMessages, llmIncludeChannelLabel); err != nil { + if _, err := store.SetBoolRuntimeSetting(storepkg.RuntimeSettingLLMQueueIncludeChannel, req.LLMIncludeChannelMessages, llmIncludeChannelLabel); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } @@ -59,10 +62,10 @@ func registerAdminRuntimeSettingsRoutes(r gin.IRouter, store *store, settings *r }) } -func runtimeSettingsDTO(settings runtimeSettingsSnapshot) gin.H { +func runtimeSettingsDTO(settings storepkg.RuntimeSettingsSnapshot) gin.H { return gin.H{ - "allow_encrypted_forwarding": settings.AllowEncryptedForwarding, - "llm_queue_enabled": settings.LLMQueueEnabled, - "llm_include_channel_messages": settings.LLMIncludeChannel, + "allow_encrypted_forwarding": settings.AllowEncryptedForwarding, + "llm_queue_enabled": settings.LLMQueueEnabled, + "llm_include_channel_messages": settings.LLMIncludeChannel, } } diff --git a/internal/runtimesettings/runtime_settings_cache.go b/internal/runtimesettings/runtime_settings_cache.go new file mode 100644 index 0000000..242f95f --- /dev/null +++ b/internal/runtimesettings/runtime_settings_cache.go @@ -0,0 +1,55 @@ +package runtimesettings + +import ( + "fmt" + "sync" + + storepkg "meshtastic_mqtt_server/internal/store" +) + +// Cache 把 runtime_settings 表中常用的开关缓存到内存中，避免每次拦截热路径 +// 都查 DB。AdminRoute 修改后通过 Reload 重新加载。 +type Cache struct { + mu sync.RWMutex + settings storepkg.RuntimeSettingsSnapshot +} + +// New 从 store 中加载初始快照并返回缓存。 +func New(s *storepkg.Store) (*Cache, error) { + cache := &Cache{} + if err := cache.Reload(s); err != nil { + return nil, err + } + return cache, nil +} + +// Reload 重新读取数据库快照覆盖当前值。 +func (c *Cache) Reload(s *storepkg.Store) error { + if s == nil { + return fmt.Errorf("store is required") + } + settings, err := s.GetRuntimeSettings() + if err != nil { + return err + } + + c.mu.Lock() + c.settings = settings + c.mu.Unlock() + return nil +} + +// Snapshot 返回当前快照的副本（结构体拷贝，调用方可以安全持有）。 +func (c *Cache) Snapshot() storepkg.RuntimeSettingsSnapshot { + if c == nil { + return storepkg.RuntimeSettingsSnapshot{} + } + c.mu.RLock() + defer c.mu.RUnlock() + return c.settings +} + +// AllowEncryptedForwarding 是 mqtt 转发热路径上常被检查的标志位的快捷读法。 +func (c *Cache) AllowEncryptedForwarding() bool { + return c.Snapshot().AllowEncryptedForwarding +} diff --git a/runtime_settings_cache_test.go b/internal/runtimesettings/runtime_settings_cache_test.go similarity index 57% rename from runtime_settings_cache_test.go rename to internal/runtimesettings/runtime_settings_cache_test.go index d249896..5b85d09 100644 --- a/runtime_settings_cache_test.go +++ b/internal/runtimesettings/runtime_settings_cache_test.go @@ -1,20 +1,29 @@ -package main +package runtimesettings -import "testing" +import ( + "testing" + + storepkg "meshtastic_mqtt_server/internal/store" + "meshtastic_mqtt_server/internal/store/testutil" +) + +func openTestStore(t *testing.T) *storepkg.Store { + return testutil.OpenStore(t) +} func TestRuntimeSettingsCacheReload(t *testing.T) { st := openTestStore(t) defer st.Close() - cache, err := newRuntimeSettingsCache(st) + cache, err := New(st) if err != nil { - t.Fatalf("newRuntimeSettingsCache() error = %v", err) + t.Fatalf("New() error = %v", err) } if cache.AllowEncryptedForwarding() { t.Fatalf("AllowEncryptedForwarding() = true, want false") } - if _, err := st.SetBoolRuntimeSetting(runtimeSettingAllowEncryptedForwarding, true, "test setting"); err != nil { + if _, err := st.SetBoolRuntimeSetting(storepkg.RuntimeSettingAllowEncryptedForwarding, true, "test setting"); err != nil { t.Fatalf("SetBoolRuntimeSetting(true) error = %v", err) } if err := cache.Reload(st); err != nil { @@ -24,7 +33,7 @@ func TestRuntimeSettingsCacheReload(t *testing.T) { t.Fatalf("AllowEncryptedForwarding() = false, want true") } - if _, err := st.SetBoolRuntimeSetting(runtimeSettingAllowEncryptedForwarding, false, "test setting"); err != nil { + if _, err := st.SetBoolRuntimeSetting(storepkg.RuntimeSettingAllowEncryptedForwarding, false, "test setting"); err != nil { t.Fatalf("SetBoolRuntimeSetting(false) error = %v", err) } if err := cache.Reload(st); err != nil { diff --git a/internal/store/testutil/testutil.go b/internal/store/testutil/testutil.go new file mode 100644 index 0000000..2cca1f9 --- /dev/null +++ b/internal/store/testutil/testutil.go @@ -0,0 +1,27 @@ +// Package testutil 提供给其它包测试使用的 store 临时实例工厂。 +// +// 重构前 db_test.go 中的 openTestStore helper 被 8+ 个测试文件复用； +// 现在抽到这里，让 store 包外的测试也可以零样板地拿到一个临时 SQLite store。 +package testutil + +import ( + "path/filepath" + "testing" + + "meshtastic_mqtt_server/internal/config" + "meshtastic_mqtt_server/internal/store" +) + +// OpenStore 返回一个写在 t.TempDir() 中的临时 SQLite store。 +// 测试结束时调用方需要 defer st.Close()。 +func OpenStore(t *testing.T) *store.Store { + t.Helper() + st, err := store.OpenStore(config.DatabaseConfig{ + Driver: config.DriverSQLite, + SQLite: config.SQLiteConfig{Path: filepath.Join(t.TempDir(), "mesh_mqtt_go.db")}, + }) + if err != nil { + t.Fatalf("OpenStore() error = %v", err) + } + return st +} diff --git a/internal/webutil/webutil.go b/internal/webutil/webutil.go new file mode 100644 index 0000000..3c0097d --- /dev/null +++ b/internal/webutil/webutil.go @@ -0,0 +1,236 @@ +// Package webutil 收集 admin/路由层共享的 HTTP 解析与响应工具。 +// +// 这些函数原本散落在 web.go 中（parseListOptions、writeListResponse、 +// parseMapReportListOptions 等），任何注册 admin 路由的领域包都依赖它们。 +// 把它们抽离出来可以避免 internal/web 同时被 internal/blocking、 +// internal/bot 等包反向引用造成循环依赖。 +package webutil + +import ( + "net/http" + "strconv" + "time" + + "github.com/gin-gonic/gin" + + "meshtastic_mqtt_server/internal/store" +) + +// ParseListOptions 从请求中读取 limit / offset / since / until / node_id / +// channel_id 等通用过滤参数；解析失败时它会写入 400 响应并返回 false。 +func ParseListOptions(c *gin.Context) (store.ListOptions, bool) { + limit, ok := ParseIntQuery(c, "limit", 100) + if !ok { + return store.ListOptions{}, false + } + offset, ok := ParseIntQuery(c, "offset", 0) + if !ok { + return store.ListOptions{}, false + } + nodeID := c.Query("node_id") + if nodeID == "" { + nodeID = c.Query("from") + } + channelID := c.Query("channel_id") + var since, until *time.Time + if value := c.Query("since"); value != "" { + parsed, err := time.Parse(time.RFC3339, value) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid since: use RFC3339"}) + return store.ListOptions{}, false + } + since = &parsed + } + if value := c.Query("until"); value != "" { + parsed, err := time.Parse(time.RFC3339, value) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid until: use RFC3339"}) + return store.ListOptions{}, false + } + until = &parsed + } + return store.NormalizeListOptions(store.ListOptions{Limit: limit, Offset: offset, NodeID: nodeID, ChannelID: channelID, Since: since, Until: until}), true +} + +// ParseMapReportListOptions 在 ParseListOptions 的基础上解析 4 个地图边界。 +func ParseMapReportListOptions(c *gin.Context) (store.ListOptions, bool) { + opts, ok := ParseListOptions(c) + if !ok { + return store.ListOptions{}, false + } + minLat, hasMinLat, ok := ParseOptionalFloatQuery(c, "min_lat") + if !ok { + return store.ListOptions{}, false + } + maxLat, hasMaxLat, ok := ParseOptionalFloatQuery(c, "max_lat") + if !ok { + return store.ListOptions{}, false + } + minLng, hasMinLng, ok := ParseOptionalFloatQuery(c, "min_lng") + if !ok { + return store.ListOptions{}, false + } + maxLng, hasMaxLng, ok := ParseOptionalFloatQuery(c, "max_lng") + if !ok { + return store.ListOptions{}, false + } + boundsCount := 0 + for _, present := range []bool{hasMinLat, hasMaxLat, hasMinLng, hasMaxLng} { + if present { + boundsCount++ + } + } + if boundsCount == 0 { + return opts, true + } + if boundsCount != 4 { + c.JSON(http.StatusBadRequest, gin.H{"error": "map bounds require min_lat, max_lat, min_lng, and max_lng"}) + return store.ListOptions{}, false + } + if minLat < -90 || minLat > 90 || maxLat < -90 || maxLat > 90 { + c.JSON(http.StatusBadRequest, gin.H{"error": "latitude bounds must be between -90 and 90"}) + return store.ListOptions{}, false + } + if minLat > maxLat { + c.JSON(http.StatusBadRequest, gin.H{"error": "min_lat must be <= max_lat"}) + return store.ListOptions{}, false + } + if minLng < -180 || minLng > 180 || maxLng < -180 || maxLng > 180 { + c.JSON(http.StatusBadRequest, gin.H{"error": "longitude bounds must be between -180 and 180"}) + return store.ListOptions{}, false + } + opts.MinLat = &minLat + opts.MaxLat = &maxLat + opts.MinLng = &minLng + opts.MaxLng = &maxLng + return opts, true +} + +// ParseMapReportViewportOptions 在 ParseMapReportListOptions 之上解析 zoom / +// cluster_threshold / target_cells 等额外字段。 +func ParseMapReportViewportOptions(c *gin.Context) (store.MapReportViewportOptions, bool) { + opts, ok := ParseMapReportListOptions(c) + if !ok { + return store.MapReportViewportOptions{}, false + } + if opts.MinLat == nil || opts.MaxLat == nil || opts.MinLng == nil || opts.MaxLng == nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "viewport bounds are required"}) + return store.MapReportViewportOptions{}, false + } + zoom, ok := ParseIntQuery(c, "zoom", 0) + if !ok { + return store.MapReportViewportOptions{}, false + } + if zoom < 0 || zoom > 24 { + c.JSON(http.StatusBadRequest, gin.H{"error": "zoom must be between 0 and 24"}) + return store.MapReportViewportOptions{}, false + } + limit, ok := ParseIntQuery(c, "limit", 1000) + if !ok { + return store.MapReportViewportOptions{}, false + } + clusterThreshold, ok := ParseIntQuery(c, "cluster_threshold", 500) + if !ok { + return store.MapReportViewportOptions{}, false + } + targetCells, ok := ParseIntQuery(c, "target_cells", 64) + if !ok { + return store.MapReportViewportOptions{}, false + } + if limit <= 0 || clusterThreshold <= 0 || targetCells <= 0 { + c.JSON(http.StatusBadRequest, gin.H{"error": "limit, cluster_threshold, and target_cells must be positive"}) + return store.MapReportViewportOptions{}, false + } + return store.NormalizeMapReportViewportOptions(store.MapReportViewportOptions{ListOptions: opts, Zoom: zoom, Limit: limit, ClusterThreshold: clusterThreshold, TargetCells: targetCells}), true +} + +// ParseIntQuery 从请求查询字符串解析整数；缺省时返回 defaultValue。 +func ParseIntQuery(c *gin.Context, name string, defaultValue int) (int, bool) { + value := c.Query(name) + if value == "" { + return defaultValue, true + } + parsed, err := strconv.Atoi(value) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid " + name}) + return 0, false + } + return parsed, true +} + +// ParseOptionalFloatQuery 解析可选浮点查询参数；返回 (value, present, ok)。 +func ParseOptionalFloatQuery(c *gin.Context, name string) (float64, bool, bool) { + value := c.Query(name) + if value == "" { + return 0, false, true + } + parsed, err := strconv.ParseFloat(value, 64) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid " + name}) + return 0, true, false + } + return parsed, true, true +} + +// WriteListResponse 把 rows 通过 convert 转成 gin.H 后包装成 {items, limit, offset}。 +func WriteListResponse[T any](c *gin.Context, rows []T, opts store.ListOptions, err error, convert func(T) gin.H) { + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + items := make([]gin.H, 0, len(rows)) + for _, row := range rows { + items = append(items, convert(row)) + } + c.JSON(http.StatusOK, gin.H{"items": items, "limit": opts.Limit, "offset": opts.Offset}) +} + +// WriteListResponseWithTotal 在 WriteListResponse 基础上额外携带 total 字段。 +func WriteListResponseWithTotal[T any](c *gin.Context, rows []T, opts store.ListOptions, total int64, err error, convert func(T) gin.H) { + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + items := make([]gin.H, 0, len(rows)) + for _, row := range rows { + items = append(items, convert(row)) + } + c.JSON(http.StatusOK, gin.H{"items": items, "limit": opts.Limit, "offset": opts.Offset, "total": total}) +} + +// PtrString / PtrInt64 / PtrUint64 / PtrFloat64 / PtrBool 把指针解引用成 any， +// 用于把数据库可空字段转换成 JSON 时让 nil 序列化为 null。 +func PtrString(value *string) any { + if value == nil { + return nil + } + return *value +} + +func PtrInt64(value *int64) any { + if value == nil { + return nil + } + return *value +} + +func PtrUint64(value *uint64) any { + if value == nil { + return nil + } + return *value +} + +func PtrFloat64(value *float64) any { + if value == nil { + return nil + } + return *value +} + +func PtrBool(value *bool) any { + if value == nil { + return nil + } + return *value +} diff --git a/main_test.go b/main_test.go index 108ff0f..6494360 100644 --- a/main_test.go +++ b/main_test.go @@ -42,10 +42,22 @@ func TestMQTTClientInfoFromClientUnsplitRemote(t *testing.T) { } } -func TestBlockingViolationForRecordNode(t *testing.T) { - cache := &blockingCache{nodes: map[string]struct{}{"!12345678": {}}, nodeNums: map[int64]struct{}{}, ips: map[string]struct{}{}} - record := map[string]any{"type": "position", "from": "!12345678", "from_num": uint32(305419896)} +// 注：blockingViolationForRecord 的测试现在跟着 blockingCache 一起搬到了 +// internal/blocking/violations_test.go，使用真实 *Store 构造缓存而不是 +// 直接捏造未导出字段。这里保留 mqtt client info 这部分测试不动。 +func TestBlockingViolationForRecordNode(t *testing.T) { + st := openTestStore(t) + defer st.Close() + nodeNum := int64(305419896) + if _, err := st.CreateNodeBlocking("!12345678", &nodeNum, "blocked", true); err != nil { + t.Fatalf("CreateNodeBlocking() error = %v", err) + } + cache, err := newBlockingCache(st) + if err != nil { + t.Fatalf("newBlockingCache() error = %v", err) + } + record := map[string]any{"type": "position", "from": "!12345678", "from_num": uint32(305419896)} violation := blockingViolationForRecord(cache, record) if violation == nil || violation["blocking_type"] != "node" { t.Fatalf("blockingViolationForRecord() = %#v, want node violation", violation) @@ -53,7 +65,15 @@ func TestBlockingViolationForRecordNode(t *testing.T) { } func TestBlockingViolationForRecordForbiddenWordFields(t *testing.T) { - cache := &blockingCache{nodes: map[string]struct{}{}, nodeNums: map[int64]struct{}{}, ips: map[string]struct{}{}, words: []forbiddenWordRule{{word: "spam", foldedWord: "spam", matchType: forbiddenWordMatchContains}}} + st := openTestStore(t) + defer st.Close() + if _, err := st.CreateForbiddenWordBlocking("spam", "contains", false, "blocked", true); err != nil { + t.Fatalf("CreateForbiddenWordBlocking() error = %v", err) + } + cache, err := newBlockingCache(st) + if err != nil { + t.Fatalf("newBlockingCache() error = %v", err) + } for _, tc := range []struct { name string @@ -74,7 +94,15 @@ func TestBlockingViolationForRecordForbiddenWordFields(t *testing.T) { } func TestBlockingViolationForRecordAllowed(t *testing.T) { - cache := &blockingCache{nodes: map[string]struct{}{}, nodeNums: map[int64]struct{}{}, ips: map[string]struct{}{}, words: []forbiddenWordRule{{word: "spam", foldedWord: "spam", matchType: forbiddenWordMatchContains}}} + st := openTestStore(t) + defer st.Close() + if _, err := st.CreateForbiddenWordBlocking("spam", "contains", false, "blocked", true); err != nil { + t.Fatalf("CreateForbiddenWordBlocking() error = %v", err) + } + cache, err := newBlockingCache(st) + if err != nil { + t.Fatalf("newBlockingCache() error = %v", err) + } record := map[string]any{"type": "text_message", "from": "!1", "text": "hello"} if violation := blockingViolationForRecord(cache, record); violation != nil { t.Fatalf("blockingViolationForRecord() = %#v, want nil", violation) diff --git a/mqttforward_bridge.go b/mqttforward_bridge.go new file mode 100644 index 0000000..ea103b7 --- /dev/null +++ b/mqttforward_bridge.go @@ -0,0 +1,26 @@ +package main + +// 桥接到 internal/mqttforward — 让 main.go / web.go / mqtt_status.go 中 +// 使用 mqttForwardManager / meshtasticMessageStats / mqttForwardReloader 等 +// 旧名字的代码仍可工作。 + +import ( + "github.com/gin-gonic/gin" + + mfpkg "meshtastic_mqtt_server/internal/mqttforward" +) + +type ( + mqttForwardManager = mfpkg.Manager + mqttForwardReloader = mfpkg.Reloader + mqttForwardRuntimeStatus = mfpkg.RuntimeStatus + meshtasticMessageStats = mfpkg.Stats +) + +func newMQTTForwardManager(s *store) *mqttForwardManager { + return mfpkg.NewManager(s) +} + +func registerAdminMQTTForwardRoutes(r gin.IRouter, s *store, forwarder mqttForwardReloader) { + mfpkg.RegisterRoutes(r, s, forwarder) +} diff --git a/runtime_settings_bridge.go b/runtime_settings_bridge.go new file mode 100644 index 0000000..87ed6b8 --- /dev/null +++ b/runtime_settings_bridge.go @@ -0,0 +1,19 @@ +package main + +// 桥接到 internal/runtimesettings — 让根目录代码继续使用旧的小写名字。 + +import ( + "github.com/gin-gonic/gin" + + rspkg "meshtastic_mqtt_server/internal/runtimesettings" +) + +type runtimeSettingsCache = rspkg.Cache + +func newRuntimeSettingsCache(s *store) (*runtimeSettingsCache, error) { + return rspkg.New(s) +} + +func registerAdminRuntimeSettingsRoutes(r gin.IRouter, s *store, c *runtimeSettingsCache) { + rspkg.RegisterRoutes(r, s, c) +} diff --git a/runtime_settings_cache.go b/runtime_settings_cache.go deleted file mode 100644 index 3e853b7..0000000 --- a/runtime_settings_cache.go +++ /dev/null @@ -1,47 +0,0 @@ -package main - -import ( - "fmt" - "sync" -) - -type runtimeSettingsCache struct { - mu sync.RWMutex - settings runtimeSettingsSnapshot -} - -func newRuntimeSettingsCache(store *store) (*runtimeSettingsCache, error) { - cache := &runtimeSettingsCache{} - if err := cache.Reload(store); err != nil { - return nil, err - } - return cache, nil -} - -func (c *runtimeSettingsCache) Reload(store *store) error { - if store == nil { - return fmt.Errorf("store is required") - } - settings, err := store.GetRuntimeSettings() - if err != nil { - return err - } - - c.mu.Lock() - c.settings = settings - c.mu.Unlock() - return nil -} - -func (c *runtimeSettingsCache) Snapshot() runtimeSettingsSnapshot { - if c == nil { - return runtimeSettingsSnapshot{} - } - c.mu.RLock() - defer c.mu.RUnlock() - return c.settings -} - -func (c *runtimeSettingsCache) AllowEncryptedForwarding() bool { - return c.Snapshot().AllowEncryptedForwarding -} diff --git a/web.go b/web.go index 113fbc7..0161535 100644 --- a/web.go +++ b/web.go @@ -190,7 +190,7 @@ func registerAdminRoutes(r gin.IRouter, store *store, sessions *sessionManager, c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid username or password"}) return } - cookie, err := sessions.newCookie(*user) + cookie, err := sessions.NewCookie(*user) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return @@ -200,7 +200,7 @@ func registerAdminRoutes(r gin.IRouter, store *store, sessions *sessionManager, c.JSON(http.StatusOK, gin.H{"user": adminUserResponse(*user)}) }) r.POST("/logout", func(c *gin.Context) { - http.SetCookie(c.Writer, sessions.clearCookie()) + http.SetCookie(c.Writer, sessions.ClearCookie()) c.JSON(http.StatusOK, gin.H{"status": "ok"}) })