优化结构

cmd/wire.go

@@ -17,6 +17,7 @@ import (
 	"kra/internal/integration/cache"
 	"kra/internal/server"
 	"kra/internal/server/handler"
+	"kra/internal/server/middleware"
 	"kra/internal/server/router"
 	"kra/internal/service"
 	"kra/internal/worker"
@@ -42,6 +43,7 @@ func wireApp(*conf.Server, *conf.Runtime, *slog.Logger, *logging.ReloadableLogge
 		initialize.ProviderSet,
 		wire.Bind(new(initialize.Backend), new(*data.Data)),
 		wire.Bind(new(cache.RedisProvider), new(*data.Data)),
+		wire.Bind(new(middleware.TokenAuthenticator), new(*service.AuthService)),
 		wire.Bind(new(biz.TaskMethodRegistry), new(*platformtask.Registry)),
 		biz.ProviderSet,
 		service.ProviderSet,

cmd/wire_gen.go

@@ -156,25 +156,20 @@ func wireApp(confServer *conf.Server, runtime *conf.Runtime, logger *slog.Logger
 	v := handler.NewSet(authority, menu, api, permission, organization, announcement, handlerEmail, handlerPayment, task, media, audit, export, version, dictionary, parameter, apiToken, systemConfig, public, user, navigation, session, integrationConfig)
 	routes := router.NewRoutes(v)
 	taskMethods := worker.NewTaskMethods(taskUsecase, mediaUsecase, runtime)
-	mqReloadable, cleanup2, err := mq.New(store, logger)
+	moduleRuntime := app.Runtime(routes, taskMethods, registry)
+	websocketServer, cleanup2, err := websocket.New(store)
 	if err != nil {
 		cleanup()
 		return nil, nil, err
 	}
-	moduleRuntime, err := app.Runtime(routes, taskMethods, registry, catalog, mqReloadable)
-	if err != nil {
-		cleanup2()
-		cleanup()
-		return nil, nil, err
-	}
-	websocketServer, cleanup3, err := websocket.New(store)
-	if err != nil {
-		cleanup2()
-		cleanup()
-		return nil, nil, err
-	}
 	engine := server.NewGinEngineWithRuntime(runtime, accessControlService, authService, securityService, auditRecorder, logger, string2, moduleRuntime, websocketServer)
 	httpServer := server.NewGinServer(confServer, engine)
+	mqReloadable, cleanup3, err := mq.New(store, logger)
+	if err != nil {
+		cleanup2()
+		cleanup()
+		return nil, nil, err
+	}
 	kratosApp := newApp(logger, httpServer, taskScheduler, auditRecorder, reloadableLogger, mqReloadable)
 	return kratosApp, func() {
 		cleanup3()

internal/app/catalog.go

@@ -7,7 +7,6 @@ import (
 	systemrouter "kra/internal/server/router"
 	systemworker "kra/internal/worker"
 	"kra/pkg/module"
-	"kra/pkg/mq"
 	platformtask "kra/pkg/task"
 )
 
@@ -28,10 +27,7 @@ func TaskRegistry(catalog module.Catalog) *platformtask.Registry {
 }
 
 // Runtime composes HTTP route contributors from the enabled modules.
-func Runtime(systemRoutes *systemrouter.Routes, systemTasks *systemworker.TaskMethods, registry *platformtask.Registry, catalog module.Catalog, subscriptions mq.SubscriptionRegistrar) (*module.Runtime, error) {
+func Runtime(systemRoutes *systemrouter.Routes, systemTasks *systemworker.TaskMethods, registry *platformtask.Registry) *module.Runtime {
 	platformtask.Apply(registry, systemTasks)
-	if err := mq.ApplySubscriptions(subscriptions, catalog.SubscriptionContributors()...); err != nil {
-		return nil, err
-	}
-	return module.NewRuntime(systemRoutes), nil
+	return module.NewRuntime(systemRoutes)
 }

internal/integration/mq/emqx_test.go

@@ -12,10 +12,10 @@ import (
 )
 
 type fakeClient struct {
-	subscribed       []string
-	unsubscribed     []string
-	handlers         map[string]platformmq.Handler
-	subscribeErrors  map[string]error
+	subscribed        []string
+	unsubscribed      []string
+	handlers          map[string]platformmq.Handler
+	subscribeErrors   map[string]error
 	unsubscribeErrors map[string]error
 }
 

internal/server/middleware/access_log_test.go

@@ -122,3 +122,25 @@ func TestAccessLogOmitsPaymentIntegrationConfigBody(t *testing.T) {
 		t.Fatalf("payment configuration summary missing: %s", logText)
 	}
 }
+
+func TestAccessLogRedactsWebSocketQueryToken(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	var logs bytes.Buffer
+	logger := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelInfo}))
+	runtime := conf.NewRuntime(nil, &conf.AdminBackend{})
+	engine := gin.New()
+	engine.Use(AccessLog(runtime, logger, "test"))
+	engine.GET("/ws", func(c *gin.Context) { c.Status(http.StatusSwitchingProtocols) })
+
+	request := httptest.NewRequest(http.MethodGet, "/ws?token=jwt-secret&room=orders", nil)
+	response := httptest.NewRecorder()
+	engine.ServeHTTP(response, request)
+
+	logText := logs.String()
+	if strings.Contains(logText, "jwt-secret") {
+		t.Fatalf("websocket token leaked into access log: %s", logText)
+	}
+	if !strings.Contains(logText, "token=%2A%2A%2A") || !strings.Contains(logText, "room=orders") {
+		t.Fatalf("redacted query missing expected fields: %s", logText)
+	}
+}

internal/server/providers_system.go

@@ -1,8 +1,6 @@
 package server
 
-import (
-	"github.com/google/wire"
-)
+import "github.com/google/wire"
 
 // ProviderSet is server providers.
 var ProviderSet = wire.NewSet(NewGinEngineWithRuntime, NewGinServer)

pkg/module/module.go

@@ -6,7 +6,6 @@ package module
 import (
 	"github.com/gin-gonic/gin"
 	"kra/pkg/database/migration"
-	"kra/pkg/mq"
 	"kra/pkg/task"
 )
 
@@ -30,12 +29,11 @@ type TimedTask struct {
 }
 
 type Definition struct {
-	Name                     string
-	Migrations               []migration.Step
-	Surface                  Surface
-	TimedTasks               []TimedTask
-	Tasks                    []task.Method
-	SubscriptionContributors []mq.SubscriptionContributor
+	Name       string
+	Migrations []migration.Step
+	Surface    Surface
+	TimedTasks []TimedTask
+	Tasks      []task.Method
 }
 
 type Catalog struct {
@@ -75,16 +73,6 @@ func (c Catalog) TaskMethods() []task.Method {
 	return methods
 }
 
-// SubscriptionContributors returns the module declarations that are applied
-// to the process-wide MQ runtime during application startup.
-func (c Catalog) SubscriptionContributors() []mq.SubscriptionContributor {
-	var contributors []mq.SubscriptionContributor
-	for _, item := range c.Definitions {
-		contributors = append(contributors, item.SubscriptionContributors...)
-	}
-	return contributors
-}
-
 type RouteRegistrar interface {
 	RegisterRoutes(public, private *gin.RouterGroup, engine *gin.Engine)
 }

pkg/mq/rabbitmq.go

@@ -191,6 +191,9 @@ func (c *RabbitMQ) Publish(ctx context.Context, topic string, payload []byte, qo
 	if c == nil {
 		return ErrUnavailable
 	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	if c != nil && strings.TrimSpace(topic) == "" {
 		topic = c.config.RoutingKey
 	}
@@ -231,6 +234,9 @@ func (c *RabbitMQ) Subscribe(ctx context.Context, topic string, qos byte, handle
 	if c == nil {
 		return ErrUnavailable
 	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	if c != nil && strings.TrimSpace(topic) == "" {
 		topic = c.config.RoutingKey
 	}
@@ -331,6 +337,9 @@ func (c *RabbitMQ) Unsubscribe(ctx context.Context, topics ...string) error {
 	if c == nil {
 		return ErrUnavailable
 	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	if len(topics) == 0 {
 		return errors.New("rabbitmq routing keys are empty")
 	}