diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index 784f309fce..28d5894d0b 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -18,6 +18,7 @@ import ( _ "github.com/Wei-Shaw/sub2api/ent/runtime" "github.com/Wei-Shaw/sub2api/internal/config" "github.com/Wei-Shaw/sub2api/internal/handler" + _ "github.com/Wei-Shaw/sub2api/internal/modules/standard" // 插件模块插装清单（唯一插装点） "github.com/Wei-Shaw/sub2api/internal/pkg/logger" "github.com/Wei-Shaw/sub2api/internal/server/middleware" "github.com/Wei-Shaw/sub2api/internal/setup" @@ -154,6 +155,21 @@ func runMainServer() { } defer app.Cleanup() + // 启动插件模块：Build（实例化 → Provision → Validate）+ Start（按稳定序）。 + // 任一失败立即中止启动（fail-fast）。注意：log.Fatalf 经 os.Exit 退出， + // defer 的 app.Cleanup() 不会执行；而部分后台服务在 Wire 构造期已自启动并 + // 持有 Redis leader 锁，必须先显式 Cleanup 再退出，避免锁滞留到 TTL 过期。 + // ⚠ 不要把下面的 log.Fatalf 改成 return：那会让上面 defer 的 Cleanup 与 + // 此处显式 Cleanup 双触发，而 Cleanup 各步骤并未统一保证幂等（审计 B-2）。 + if err := app.Runtime.Build(); err != nil { + app.Cleanup() + log.Fatalf("Failed to build plugin modules: %v", err) + } + if err := app.Runtime.Start(context.Background()); err != nil { + app.Cleanup() + log.Fatalf("Failed to start plugin modules: %v", err) + } + // 启动服务器 go func() { if err := app.Server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { diff --git a/backend/cmd/server/wire.go b/backend/cmd/server/wire.go index da119aaaca..4d92f36ebe 100644 --- a/backend/cmd/server/wire.go +++ b/backend/cmd/server/wire.go @@ -14,6 +14,7 @@ import ( "github.com/Wei-Shaw/sub2api/internal/config" "github.com/Wei-Shaw/sub2api/internal/handler" "github.com/Wei-Shaw/sub2api/internal/payment" + "github.com/Wei-Shaw/sub2api/internal/plugin" "github.com/Wei-Shaw/sub2api/internal/repository" "github.com/Wei-Shaw/sub2api/internal/server" "github.com/Wei-Shaw/sub2api/internal/server/middleware" @@ -24,7 +25,10 @@ import ( ) type Application struct { - Server *http.Server + Server *http.Server + // Runtime 是插件模块生命周期驱动器：main 在 HTTP server 启动前 + // 调用 Build + Start，关闭时由 Cleanup 序列执行 Stop。 + Runtime *plugin.Runtime Cleanup func() } @@ -43,6 +47,9 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { // Server layer ProviderSet server.ProviderSet, + // Plugin kernel ProviderSet (Host + module Runtime) + plugin.ProviderSet, + // Privacy client factory for OpenAI training opt-out providePrivacyClientFactory, @@ -53,7 +60,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { provideCleanup, // Application struct - wire.Struct(new(Application), "Server", "Cleanup"), + wire.Struct(new(Application), "Server", "Runtime", "Cleanup"), ) return nil, nil } @@ -100,6 +107,7 @@ func provideCleanup( paymentOrderExpiry *service.PaymentOrderExpiryService, channelMonitorRunner *service.ChannelMonitorRunner, quotaFlusher *service.UserPlatformQuotaUsageFlusher, + moduleRuntime *plugin.Runtime, ) func() { return func() { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) @@ -112,6 +120,14 @@ func provideCleanup( // 应用层清理步骤可并行执行，基础设施资源（Redis/Ent）最后按顺序关闭。 parallelSteps := []cleanupStep{ + // 插件模块 Stop 属于业务服务关闭组：模块停止时可能仍需写 DB/Redis， + // 必须先于基础设施（Redis/Ent）关闭执行。 + {"PluginModuleRuntime", func() error { + if moduleRuntime == nil { + return nil + } + return moduleRuntime.Stop(ctx) + }}, {"OpsScheduledReportService", func() error { if opsScheduledReport != nil { opsScheduledReport.Stop() diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index 879d56d22f..3eedd8559f 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -13,6 +13,7 @@ import ( "github.com/Wei-Shaw/sub2api/internal/handler" "github.com/Wei-Shaw/sub2api/internal/handler/admin" "github.com/Wei-Shaw/sub2api/internal/payment" + "github.com/Wei-Shaw/sub2api/internal/plugin" "github.com/Wei-Shaw/sub2api/internal/repository" "github.com/Wei-Shaw/sub2api/internal/server" "github.com/Wei-Shaw/sub2api/internal/server/middleware" @@ -27,6 +28,7 @@ import ( import ( _ "embed" _ "github.com/Wei-Shaw/sub2api/ent/runtime" + _ "github.com/Wei-Shaw/sub2api/internal/modules/standard" ) // Injectors from wire.go: @@ -241,12 +243,20 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { paymentHandler := admin.NewPaymentHandler(paymentService, paymentConfigService) affiliateHandler := admin.NewAffiliateHandler(affiliateService, adminService) complianceHandler := admin.NewComplianceHandler(settingService) - adminHandlers := handler.ProvideAdminHandlers(dashboardHandler, adminUserHandler, groupHandler, accountHandler, adminAnnouncementHandler, dataManagementHandler, backupHandler, oAuthHandler, openAIOAuthHandler, geminiOAuthHandler, antigravityOAuthHandler, proxyHandler, adminRedeemHandler, promoHandler, settingHandler, opsHandler, systemHandler, adminSubscriptionHandler, adminUsageHandler, userAttributeHandler, errorPassthroughHandler, tlsFingerprintProfileHandler, adminAPIKeyHandler, scheduledTestHandler, channelHandler, channelMonitorHandler, channelMonitorRequestTemplateHandler, contentModerationHandler, paymentHandler, affiliateHandler, complianceHandler) + pluginConfig, err := plugin.ProvideModuleConfig(configConfig) + if err != nil { + return nil, err + } + host := plugin.ProvideHost(pluginConfig, client, redisClient) + runtime := plugin.ProvideModuleRuntime(host, pluginConfig) + moduleHandler := admin.NewModuleHandler(runtime) + adminHandlers := handler.ProvideAdminHandlers(dashboardHandler, adminUserHandler, groupHandler, accountHandler, adminAnnouncementHandler, dataManagementHandler, backupHandler, oAuthHandler, openAIOAuthHandler, geminiOAuthHandler, antigravityOAuthHandler, proxyHandler, adminRedeemHandler, promoHandler, settingHandler, opsHandler, systemHandler, adminSubscriptionHandler, adminUsageHandler, userAttributeHandler, errorPassthroughHandler, tlsFingerprintProfileHandler, adminAPIKeyHandler, scheduledTestHandler, channelHandler, channelMonitorHandler, channelMonitorRequestTemplateHandler, contentModerationHandler, paymentHandler, affiliateHandler, complianceHandler, moduleHandler) usageRecordWorkerPool := service.NewUsageRecordWorkerPool(configConfig) + chain := handler.ProvideGatewayHookChain(contentModerationService) userMsgQueueCache := repository.NewUserMsgQueueCache(redisClient) userMessageQueueService := service.ProvideUserMessageQueueService(userMsgQueueCache, rpmCache, configConfig) - gatewayHandler := handler.NewGatewayHandler(gatewayService, geminiMessagesCompatService, antigravityGatewayService, userService, concurrencyService, billingCacheService, usageService, apiKeyService, usageRecordWorkerPool, errorPassthroughService, contentModerationService, userMessageQueueService, configConfig, settingService) - openAIGatewayHandler := handler.NewOpenAIGatewayHandler(openAIGatewayService, concurrencyService, billingCacheService, apiKeyService, usageRecordWorkerPool, errorPassthroughService, contentModerationService, configConfig) + gatewayHandler := handler.NewGatewayHandler(gatewayService, geminiMessagesCompatService, antigravityGatewayService, userService, concurrencyService, billingCacheService, usageService, apiKeyService, usageRecordWorkerPool, errorPassthroughService, chain, userMessageQueueService, configConfig, settingService) + openAIGatewayHandler := handler.NewOpenAIGatewayHandler(openAIGatewayService, concurrencyService, billingCacheService, apiKeyService, usageRecordWorkerPool, errorPassthroughService, contentModerationService, chain, configConfig) handlerSettingHandler := handler.ProvideSettingHandler(settingService, buildInfo, notificationEmailService) totpHandler := handler.NewTotpHandler(totpService) handlerPaymentHandler := handler.NewPaymentHandler(paymentService, paymentConfigService, channelService) @@ -273,9 +283,10 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { paymentOrderExpiryService := service.ProvidePaymentOrderExpiryService(paymentService, leaderLockCache, db) channelMonitorRunner := service.ProvideChannelMonitorRunner(channelMonitorService, settingService) userPlatformQuotaUsageFlusher := service.ProvideUserPlatformQuotaUsageFlusher(configConfig, billingCache, serviceUserPlatformQuotaRepository, timingWheelService) - v := provideCleanup(client, redisClient, opsMetricsCollector, opsAggregationService, opsAlertEvaluatorService, opsCleanupService, opsScheduledReportService, opsSystemLogSink, schedulerSnapshotService, tokenRefreshService, accountExpiryService, proxyExpiryService, subscriptionExpiryService, usageCleanupService, idempotencyCleanupService, pricingService, emailQueueService, billingCacheService, usageRecordWorkerPool, subscriptionService, oAuthService, openAIOAuthService, geminiOAuthService, antigravityOAuthService, openAIGatewayService, scheduledTestRunnerService, backupService, paymentOrderExpiryService, channelMonitorRunner, userPlatformQuotaUsageFlusher) + v := provideCleanup(client, redisClient, opsMetricsCollector, opsAggregationService, opsAlertEvaluatorService, opsCleanupService, opsScheduledReportService, opsSystemLogSink, schedulerSnapshotService, tokenRefreshService, accountExpiryService, proxyExpiryService, subscriptionExpiryService, usageCleanupService, idempotencyCleanupService, pricingService, emailQueueService, billingCacheService, usageRecordWorkerPool, subscriptionService, oAuthService, openAIOAuthService, geminiOAuthService, antigravityOAuthService, openAIGatewayService, scheduledTestRunnerService, backupService, paymentOrderExpiryService, channelMonitorRunner, userPlatformQuotaUsageFlusher, runtime) application := &Application{ Server: httpServer, + Runtime: runtime, Cleanup: v, } return application, nil @@ -284,7 +295,10 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { // wire.go: type Application struct { - Server *http.Server + Server *http.Server + // Runtime 是插件模块生命周期驱动器：main 在 HTTP server 启动前 + // 调用 Build + Start，关闭时由 Cleanup 序列执行 Stop。 + Runtime *plugin.Runtime Cleanup func() } @@ -330,6 +344,7 @@ func provideCleanup( paymentOrderExpiry *service.PaymentOrderExpiryService, channelMonitorRunner *service.ChannelMonitorRunner, quotaFlusher *service.UserPlatformQuotaUsageFlusher, + moduleRuntime *plugin.Runtime, ) func() { return func() { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) @@ -341,6 +356,13 @@ func provideCleanup( } parallelSteps := []cleanupStep{ + + {"PluginModuleRuntime", func() error { + if moduleRuntime == nil { + return nil + } + return moduleRuntime.Stop(ctx) + }}, {"OpsScheduledReportService", func() error { if opsScheduledReport != nil { opsScheduledReport.Stop() diff --git a/backend/cmd/server/wire_gen_test.go b/backend/cmd/server/wire_gen_test.go index 4a00c1cceb..8abe8dcb3c 100644 --- a/backend/cmd/server/wire_gen_test.go +++ b/backend/cmd/server/wire_gen_test.go @@ -80,6 +80,7 @@ func TestProvideCleanup_WithMinimalDependencies_NoPanic(t *testing.T) { nil, // paymentOrderExpiry nil, // channelMonitorRunner nil, // quotaFlusher + nil, // moduleRuntime ) require.NotPanics(t, func() { diff --git a/backend/go.mod b/backend/go.mod index 62be56c86e..a2629782a4 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -119,7 +119,7 @@ require ( github.com/mattn/go-isatty v0.0.20 // indirect github.com/mdelapenya/tlscert v0.2.0 // indirect github.com/mitchellh/go-wordwrap v1.0.1 // indirect - github.com/mitchellh/mapstructure v1.5.0 // indirect + github.com/mitchellh/mapstructure v1.5.0 github.com/moby/docker-image-spec v1.3.1 // indirect github.com/moby/go-archive v0.1.0 // indirect github.com/moby/patternmatcher v0.6.0 // indirect diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 2422b2cab0..03ee8b48ec 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -93,6 +93,15 @@ type Config struct { Gemini GeminiConfig `mapstructure:"gemini"` Update UpdateConfig `mapstructure:"update"` Idempotency IdempotencyConfig `mapstructure:"idempotency"` + + // Modules 是插件模块配置子树（modules:）的原始内容。 + // 顶层 key 为模块 ID 字面量（如 "job.hello"），value 为该模块的原始配置 + // （含可选 enabled 键），语义解析由 internal/plugin 包负责。 + // + // 注意：该字段不走 viper.Unmarshal（mapstructure:"-"），而是在 load() 中通过 + // viper.Get 手工提取——Unmarshal 基于 AllSettings 重建嵌套结构，会把含点的 + // 模块 ID key 错误拆分为多级嵌套。缺省（未配置）时为空 map，行为与现状完全等价。 + Modules map[string]map[string]any `mapstructure:"-" yaml:"-"` } type LogConfig struct { @@ -1397,6 +1406,11 @@ func load(allowMissingJWTSecret bool) (*Config, error) { if err := viper.Unmarshal(&cfg); err != nil { return nil, fmt.Errorf("unmarshal config error: %w", err) } + modules, err := normalizeModulesSubtree(viper.Get("modules")) + if err != nil { + return nil, fmt.Errorf("parse modules config error: %w", err) + } + cfg.Modules = modules if cfg.Gateway.OpenAIScheduler.StickyEscapeTTFTMs == 0 { cfg.Gateway.OpenAIScheduler.StickyEscapeTTFTMs = 15000 } @@ -1530,6 +1544,33 @@ func load(allowMissingJWTSecret bool) (*Config, error) { return &cfg, nil } +// normalizeModulesSubtree 将 viper 读取的 modules: 原始子树规整为 +// map[模块ID]map[配置键]值。raw 为 nil（未配置）时返回空 map（零行为影响）； +// 子树或某个模块项不是映射时返回错误。空模块项（如 "job.hello:" 后无内容） +// 规整为空 map，表示"提及但无私有配置"。 +func normalizeModulesSubtree(raw any) (map[string]map[string]any, error) { + result := make(map[string]map[string]any) + if raw == nil { + return result, nil + } + top, ok := raw.(map[string]any) + if !ok { + return nil, fmt.Errorf("modules: expected a mapping, got %T", raw) + } + for key, value := range top { + if value == nil { + result[key] = map[string]any{} + continue + } + entry, ok := value.(map[string]any) + if !ok { + return nil, fmt.Errorf("modules.%s: expected a mapping, got %T", key, value) + } + result[key] = entry + } + return result, nil +} + func setDefaults() { viper.SetDefault("run_mode", RunModeStandard) diff --git a/backend/internal/config/modules_config_test.go b/backend/internal/config/modules_config_test.go new file mode 100644 index 0000000000..915eca3ad6 --- /dev/null +++ b/backend/internal/config/modules_config_test.go @@ -0,0 +1,89 @@ +package config + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestLoadModulesSubtreeDefaultEmpty 验证不配置 modules: 子树时， +// cfg.Modules 为空（非 nil）map，解析结果与现状完全等价。 +func TestLoadModulesSubtreeDefaultEmpty(t *testing.T) { + resetViperWithJWTSecret(t) + + cfg, err := Load() + require.NoError(t, err) + require.NotNil(t, cfg.Modules) + require.Empty(t, cfg.Modules) +} + +// TestLoadModulesSubtreeParsing 验证 modules: 子树解析： +// 含点的模块 ID 必须作为完整 key 保留（不被 viper 按 "." 拆分为多级嵌套）。 +func TestLoadModulesSubtreeParsing(t *testing.T) { + resetViperWithJWTSecret(t) + + tempDir := t.TempDir() + configPath := filepath.Join(tempDir, "config.yaml") + yaml := ` +modules: + job.hello: + enabled: true + greeting: "hi" + payment.provider.stripe: + enabled: false + job.bare: +` + require.NoError(t, os.WriteFile(configPath, []byte(yaml), 0o644)) + t.Setenv("DATA_DIR", tempDir) + + cfg, err := Load() + require.NoError(t, err) + require.Len(t, cfg.Modules, 3) + + require.Equal(t, map[string]any{"enabled": true, "greeting": "hi"}, cfg.Modules["job.hello"]) + require.Equal(t, map[string]any{"enabled": false}, cfg.Modules["payment.provider.stripe"]) + // 空模块项规整为空 map（提及但无私有配置）。 + require.NotNil(t, cfg.Modules["job.bare"]) + require.Empty(t, cfg.Modules["job.bare"]) +} + +// TestLoadModulesSubtreeInvalidShape 验证 modules 子树形状非法时报错。 +func TestLoadModulesSubtreeInvalidShape(t *testing.T) { + resetViperWithJWTSecret(t) + + tempDir := t.TempDir() + configPath := filepath.Join(tempDir, "config.yaml") + require.NoError(t, os.WriteFile(configPath, []byte("modules:\n job.hello: true\n"), 0o644)) + t.Setenv("DATA_DIR", tempDir) + + _, err := Load() + require.Error(t, err) + require.Contains(t, err.Error(), "modules") +} + +// TestNormalizeModulesSubtree 直接覆盖规整函数的边界分支。 +func TestNormalizeModulesSubtree(t *testing.T) { + t.Run("nil input", func(t *testing.T) { + out, err := normalizeModulesSubtree(nil) + require.NoError(t, err) + require.NotNil(t, out) + require.Empty(t, out) + }) + t.Run("non-map top level", func(t *testing.T) { + _, err := normalizeModulesSubtree([]any{"x"}) + require.Error(t, err) + }) + t.Run("non-map entry", func(t *testing.T) { + _, err := normalizeModulesSubtree(map[string]any{"job.hello": 42}) + require.Error(t, err) + require.Contains(t, err.Error(), "job.hello") + }) + t.Run("nil entry becomes empty map", func(t *testing.T) { + out, err := normalizeModulesSubtree(map[string]any{"job.hello": nil}) + require.NoError(t, err) + require.NotNil(t, out["job.hello"]) + require.Empty(t, out["job.hello"]) + }) +} diff --git a/backend/internal/handler/admin/module_handler.go b/backend/internal/handler/admin/module_handler.go new file mode 100644 index 0000000000..efaf7b3da8 --- /dev/null +++ b/backend/internal/handler/admin/module_handler.go @@ -0,0 +1,60 @@ +package admin + +import ( + "sort" + + "github.com/Wei-Shaw/sub2api/internal/pkg/response" + "github.com/Wei-Shaw/sub2api/internal/plugin" + "github.com/Wei-Shaw/sub2api/internal/util/logredact" + + "github.com/gin-gonic/gin" +) + +// moduleStatusSource 提供插件模块状态快照能力（*plugin.Runtime 实现该接口，测试可注入桩）。 +type moduleStatusSource interface { + Snapshot() []plugin.ModuleStatus +} + +// ModuleHandler handles plugin module observability operations (read-only) +type ModuleHandler struct { + runtime moduleStatusSource +} + +// NewModuleHandler creates a new ModuleHandler +func NewModuleHandler(runtime *plugin.Runtime) *ModuleHandler { + return &ModuleHandler{runtime: runtime} +} + +// ModuleStatusItem 是单个插件模块状态的响应条目 +type ModuleStatusItem struct { + ID string `json:"id"` + Namespace string `json:"namespace"` + Name string `json:"name"` + Enabled bool `json:"enabled"` + State string `json:"state"` + Error string `json:"error"` +} + +// List 返回全部已注册插件模块的状态清单（含未启用模块），按模块 ID 字典序排序 +// GET /api/v1/admin/modules +func (h *ModuleHandler) List(c *gin.Context) { + statuses := h.runtime.Snapshot() + modules := make([]ModuleStatusItem, 0, len(statuses)) + for _, status := range statuses { + errText := status.Err + if errText != "" { + // 错误文本可能包含敏感信息（如上游返回的凭据片段），输出前统一脱敏。 + errText = logredact.RedactText(errText) + } + modules = append(modules, ModuleStatusItem{ + ID: string(status.ID), + Namespace: status.ID.Namespace(), + Name: status.ID.Name(), + Enabled: status.Enabled, + State: string(status.State), + Error: errText, + }) + } + sort.Slice(modules, func(i, j int) bool { return modules[i].ID < modules[j].ID }) + response.Success(c, gin.H{"modules": modules}) +} diff --git a/backend/internal/handler/admin/module_handler_test.go b/backend/internal/handler/admin/module_handler_test.go new file mode 100644 index 0000000000..ed0725f600 --- /dev/null +++ b/backend/internal/handler/admin/module_handler_test.go @@ -0,0 +1,177 @@ +package admin + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/plugin" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +type moduleStatusSourceStub struct { + statuses []plugin.ModuleStatus +} + +func (s *moduleStatusSourceStub) Snapshot() []plugin.ModuleStatus { + return s.statuses +} + +type moduleListEnvelope struct { + Code int `json:"code"` + Message string `json:"message"` + Data struct { + Modules []struct { + ID string `json:"id"` + Namespace string `json:"namespace"` + Name string `json:"name"` + Enabled bool `json:"enabled"` + State string `json:"state"` + Error string `json:"error"` + } `json:"modules"` + } `json:"data"` +} + +func performModuleList(t *testing.T, source moduleStatusSource) (*httptest.ResponseRecorder, moduleListEnvelope) { + t.Helper() + gin.SetMode(gin.TestMode) + router := gin.New() + handler := &ModuleHandler{runtime: source} + router.GET("/admin/modules", handler.List) + + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/admin/modules", nil) + router.ServeHTTP(w, req) + + var envelope moduleListEnvelope + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &envelope)) + return w, envelope +} + +func TestModuleHandlerListSerializesSnapshot(t *testing.T) { + stub := &moduleStatusSourceStub{statuses: []plugin.ModuleStatus{ + {ID: "gateway.platform.anthropic", Enabled: true, State: plugin.StateRunning}, + {ID: "job.hello", Enabled: false, State: plugin.StateRegistered}, + }} + + w, envelope := performModuleList(t, stub) + + require.Equal(t, http.StatusOK, w.Code) + require.Equal(t, 0, envelope.Code) + require.Equal(t, "success", envelope.Message) + require.Len(t, envelope.Data.Modules, 2) + + first := envelope.Data.Modules[0] + require.Equal(t, "gateway.platform.anthropic", first.ID) + require.Equal(t, "gateway.platform", first.Namespace) + require.Equal(t, "anthropic", first.Name) + require.True(t, first.Enabled) + require.Equal(t, "running", first.State) + require.Empty(t, first.Error) + + // 未启用模块也在列：state=registered、enabled=false + second := envelope.Data.Modules[1] + require.Equal(t, "job.hello", second.ID) + require.Equal(t, "job", second.Namespace) + require.Equal(t, "hello", second.Name) + require.False(t, second.Enabled) + require.Equal(t, "registered", second.State) + require.Empty(t, second.Error) + + // snake_case 字段恒定输出（含空 error） + require.Contains(t, w.Body.String(), `"error":""`) + require.Contains(t, w.Body.String(), `"state":"registered"`) +} + +func TestModuleHandlerListSortsByID(t *testing.T) { + stub := &moduleStatusSourceStub{statuses: []plugin.ModuleStatus{ + {ID: "job.zeta", Enabled: true, State: plugin.StateRunning}, + {ID: "gateway.alpha", Enabled: true, State: plugin.StateRunning}, + {ID: "job.alpha", Enabled: false, State: plugin.StateRegistered}, + }} + + _, envelope := performModuleList(t, stub) + + ids := make([]string, 0, len(envelope.Data.Modules)) + for _, m := range envelope.Data.Modules { + ids = append(ids, m.ID) + } + require.Equal(t, []string{"gateway.alpha", "job.alpha", "job.zeta"}, ids) +} + +func TestModuleHandlerListRedactsErrorText(t *testing.T) { + stub := &moduleStatusSourceStub{statuses: []plugin.ModuleStatus{ + {ID: "job.bad", Enabled: true, State: plugin.StateErrored, Err: "provision failed: access_token=ya29.secret-value"}, + }} + + _, envelope := performModuleList(t, stub) + + require.Len(t, envelope.Data.Modules, 1) + errText := envelope.Data.Modules[0].Error + require.Contains(t, errText, "provision failed") + require.NotContains(t, errText, "ya29.secret-value") +} + +func TestModuleHandlerListEmptySnapshot(t *testing.T) { + w, envelope := performModuleList(t, &moduleStatusSourceStub{}) + + require.Equal(t, http.StatusOK, w.Code) + require.NotNil(t, envelope.Data.Modules) + require.Empty(t, envelope.Data.Modules) + require.Contains(t, w.Body.String(), `"modules":[]`) +} + +type moduleHandlerFakeModule struct { + id plugin.ModuleID + enabledByDefault bool +} + +func (m *moduleHandlerFakeModule) ModuleInfo() plugin.ModuleInfo { + id := m.id + enabled := m.enabledByDefault + return plugin.ModuleInfo{ + ID: id, + New: func() plugin.Module { return &moduleHandlerFakeModule{id: id, enabledByDefault: enabled} }, + EnabledByDefault: enabled, + } +} + +// TestModuleHandlerListWithRuntimeSnapshot 用真实 Runtime（隔离注册表）验证 +// 接口返回与 Runtime.Snapshot 一致：enabled 模块为 running， +// disabled 模块也在列且 state=registered、enabled=false。 +func TestModuleHandlerListWithRuntimeSnapshot(t *testing.T) { + registry := plugin.NewRegistry() + registry.RegisterModule(&moduleHandlerFakeModule{id: "job.enabled", enabledByDefault: true}) + registry.RegisterModule(&moduleHandlerFakeModule{id: "job.disabled", enabledByDefault: false}) + + runtime := plugin.NewRuntimeWithRegistry(nil, nil, registry) + require.NoError(t, runtime.Build()) + require.NoError(t, runtime.Start(context.Background())) + t.Cleanup(func() { _ = runtime.Stop(context.Background()) }) + + gin.SetMode(gin.TestMode) + router := gin.New() + router.GET("/admin/modules", NewModuleHandler(runtime).List) + + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/admin/modules", nil) + router.ServeHTTP(w, req) + + require.Equal(t, http.StatusOK, w.Code) + var envelope moduleListEnvelope + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &envelope)) + require.Len(t, envelope.Data.Modules, 2) + + disabled := envelope.Data.Modules[0] + require.Equal(t, "job.disabled", disabled.ID) + require.False(t, disabled.Enabled) + require.Equal(t, "registered", disabled.State) + + enabled := envelope.Data.Modules[1] + require.Equal(t, "job.enabled", enabled.ID) + require.True(t, enabled.Enabled) + require.Equal(t, "running", enabled.State) +} diff --git a/backend/internal/handler/handler.go b/backend/internal/handler/handler.go index 95cac6274c..182b625be0 100644 --- a/backend/internal/handler/handler.go +++ b/backend/internal/handler/handler.go @@ -37,6 +37,7 @@ type AdminHandlers struct { Payment *admin.PaymentHandler Affiliate *admin.AffiliateHandler Compliance *admin.ComplianceHandler + Module *admin.ModuleHandler } // Handlers contains all HTTP handlers diff --git a/backend/internal/handler/wire.go b/backend/internal/handler/wire.go index 7d7f4dc622..97ed8db2c1 100644 --- a/backend/internal/handler/wire.go +++ b/backend/internal/handler/wire.go @@ -40,6 +40,7 @@ func ProvideAdminHandlers( paymentHandler *admin.PaymentHandler, affiliateHandler *admin.AffiliateHandler, complianceHandler *admin.ComplianceHandler, + moduleHandler *admin.ModuleHandler, ) *AdminHandlers { return &AdminHandlers{ Dashboard: dashboardHandler, @@ -73,6 +74,7 @@ func ProvideAdminHandlers( Payment: paymentHandler, Affiliate: affiliateHandler, Compliance: complianceHandler, + Module: moduleHandler, } } @@ -147,6 +149,7 @@ var ProviderSet = wire.NewSet( NewSubscriptionHandler, NewAnnouncementHandler, NewChannelMonitorUserHandler, + ProvideGatewayHookChain, NewGatewayHandler, NewOpenAIGatewayHandler, NewTotpHandler, @@ -187,6 +190,7 @@ var ProviderSet = wire.NewSet( admin.NewPaymentHandler, admin.NewAffiliateHandler, admin.NewComplianceHandler, + admin.NewModuleHandler, // AdminHandlers and Handlers constructors ProvideAdminHandlers, diff --git a/backend/internal/modules/hello/audit_regression_test.go b/backend/internal/modules/hello/audit_regression_test.go new file mode 100644 index 0000000000..ff81ca70b5 --- /dev/null +++ b/backend/internal/modules/hello/audit_regression_test.go @@ -0,0 +1,34 @@ +package hello + +// Phase-1.5 对抗式审计（2026-06-11）B-1 回归测试： +// Stop 传入已取消的 ctx 时，若 worker 实际已干净退出，必须返回 nil +// （修复前 select 双臂就绪的随机选取可能把成功停止误报为失败）。 + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/Wei-Shaw/sub2api/internal/plugin/plugintest" +) + +func TestStopWithPreCancelledContextAfterWorkerExit(t *testing.T) { + m := new(Module) + require.NoError(t, m.Provision(plugintest.NewHost(t))) + require.NoError(t, m.Validate()) + require.NoError(t, m.Start(context.Background())) + + // 让 worker 先行退出并确认退出完成（直接驱动内部 cancel，规避计时竞态）。 + m.cancel() + select { + case <-m.done: + case <-time.After(5 * time.Second): + t.Fatal("worker 未在期限内退出") + } + + cancelled, cancel := context.WithCancel(context.Background()) + cancel() + require.NoError(t, m.Stop(cancelled), "worker 已退出时，即便 ctx 已取消 Stop 也应返回 nil") +} diff --git a/backend/internal/modules/hello/hello.go b/backend/internal/modules/hello/hello.go new file mode 100644 index 0000000000..97884d7a31 --- /dev/null +++ b/backend/internal/modules/hello/hello.go @@ -0,0 +1,154 @@ +// Package hello 实现示例插件模块 job.hello。 +// +// 该模块是插件内核的最小可运行示例：实现全部四个可选生命周期接口 +// （Provision / Validate / Start / Stop），启动后按配置间隔周期性输出 +// 一条 debug 日志。默认 disabled——仅在 `modules:` 配置中显式 enabled +// 后才会运行，用于验证插件链路连通性，并作为编写新模块的参考实现。 +package hello + +import ( + "context" + "errors" + "fmt" + "time" + + "go.uber.org/zap" + + "github.com/Wei-Shaw/sub2api/internal/plugin" +) + +// ID 是 hello 模块的模块 ID。 +const ID plugin.ModuleID = "job.hello" + +func init() { + plugin.RegisterModule(&Module{}) +} + +// Config 是 hello 模块的私有配置 +// （`modules:` 子树中 "job.hello" 项除 enabled 外的部分）。 +type Config struct { + // Interval 是周期日志的输出间隔，必须大于 0。 + Interval time.Duration `mapstructure:"interval"` + + // Greeting 是周期日志输出的内容，不能为空。 + Greeting string `mapstructure:"greeting"` +} + +// defaultConfig 返回 hello 模块的默认配置（用户未配置的字段保持默认值）。 +func defaultConfig() Config { + return Config{ + Interval: 30 * time.Second, + Greeting: "hello from job.hello", + } +} + +// Module 是 job.hello 模块实例。零值即可用，依赖在 Provision 中装配； +// 每次 Runtime.Build 都会通过 ModuleInfo.New 创建全新实例。 +type Module struct { + cfg Config + logger *zap.Logger + + cancel context.CancelFunc + done chan struct{} +} + +// 编译期断言：Module 实现 plugin.Module 与全部四个可选生命周期接口。 +var ( + _ plugin.Module = (*Module)(nil) + _ plugin.Provisioner = (*Module)(nil) + _ plugin.Validator = (*Module)(nil) + _ plugin.Starter = (*Module)(nil) + _ plugin.Stopper = (*Module)(nil) +) + +// ModuleInfo 返回模块注册信息。EnabledByDefault 为 false： +// 未在 `modules:` 中显式 enabled 时模块不实例化、无任何副作用。 +func (*Module) ModuleInfo() plugin.ModuleInfo { + return plugin.ModuleInfo{ + ID: ID, + New: func() plugin.Module { return new(Module) }, + EnabledByDefault: false, + } +} + +// Provision 从宿主获取 logger，并在默认配置之上解码模块私有配置。 +func (m *Module) Provision(host *plugin.Host) error { + m.cfg = defaultConfig() + if host != nil { + m.logger = host.Logger + if host.ConfigOf != nil { + if err := host.ConfigOf(ID, &m.cfg); err != nil { + return err + } + } + } + if m.logger == nil { + m.logger = zap.NewNop() + } + return nil +} + +// Validate 校验模块配置：interval 必须大于 0，greeting 不能为空。 +func (m *Module) Validate() error { + if m.cfg.Interval <= 0 { + return fmt.Errorf("hello: interval must be positive, got %s", m.cfg.Interval) + } + if m.cfg.Greeting == "" { + return errors.New("hello: greeting must not be empty") + } + return nil +} + +// Start 启动周期日志 goroutine。 +// +// 模块持有独立于启动 ctx 的生命周期上下文，由 Stop 负责取消， +// 因此模块的运行寿命不受启动流程上下文的影响。 +func (m *Module) Start(_ context.Context) error { + ctx, cancel := context.WithCancel(context.Background()) + m.cancel = cancel + m.done = make(chan struct{}) + go m.run(ctx) + m.logger.Info("hello module started", + zap.String("module", string(ID)), + zap.Duration("interval", m.cfg.Interval)) + return nil +} + +// run 按配置间隔输出周期 debug 日志，直到 ctx 被取消。 +func (m *Module) run(ctx context.Context) { + defer close(m.done) + ticker := time.NewTicker(m.cfg.Interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + m.logger.Debug(m.cfg.Greeting, zap.String("module", string(ID))) + } + } +} + +// Stop 取消周期 goroutine 并等待其退出，尊重 ctx 的 deadline。 +// 未 Start 过（或 Start 之前失败）时调用为 no-op。 +func (m *Module) Stop(ctx context.Context) error { + if m.cancel == nil { + return nil + } + m.cancel() + // 先非阻塞探测 worker 是否已退出：当传入的 ctx 已取消/超时而 worker 其实 + // 已干净退出时，select 双臂就绪的随机选取可能把成功停止误报为失败（审计 B-1）。 + select { + case <-m.done: + m.logger.Info("hello module stopped", zap.String("module", string(ID))) + return nil + default: + } + select { + case <-m.done: + m.logger.Info("hello module stopped", zap.String("module", string(ID))) + return nil + case <-ctx.Done(): + return fmt.Errorf("hello: stop interrupted while waiting for worker exit: %w", ctx.Err()) + } +} diff --git a/backend/internal/modules/hello/hello_test.go b/backend/internal/modules/hello/hello_test.go new file mode 100644 index 0000000000..75362d5c73 --- /dev/null +++ b/backend/internal/modules/hello/hello_test.go @@ -0,0 +1,178 @@ +package hello + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/Wei-Shaw/sub2api/internal/plugin" + "github.com/Wei-Shaw/sub2api/internal/plugin/plugintest" +) + +// TestModuleRegisteredInDefaultRegistry 验证 init() 已向包级默认注册表 +// 注册 job.hello，且默认 disabled（EnabledByDefault=false）。 +func TestModuleRegisteredInDefaultRegistry(t *testing.T) { + info, ok := plugin.GetModule(ID) + require.True(t, ok, "job.hello should self-register via init()") + require.Equal(t, ID, info.ID) + require.False(t, info.EnabledByDefault, "hello module must be disabled by default") + require.NotNil(t, info.New) + require.IsType(t, &Module{}, info.New()) +} + +// TestProvisionDefaults 验证未提供私有配置时 Provision 使用默认配置且可通过校验。 +func TestProvisionDefaults(t *testing.T) { + m := new(Module) + require.NoError(t, m.Provision(plugintest.NewHost(t))) + require.Equal(t, defaultConfig(), m.cfg) + require.NoError(t, m.Validate()) + + // nil host 防御路径：不 panic、使用默认配置。 + m2 := new(Module) + require.NoError(t, m2.Provision(nil)) + require.Equal(t, defaultConfig(), m2.cfg) + require.NoError(t, m2.Validate()) +} + +// TestProvisionDecodesPrivateConfig 验证 Provision 通过 host.ConfigOf +// 解码模块私有配置（interval 时长字符串、greeting 覆盖）。 +func TestProvisionDecodesPrivateConfig(t *testing.T) { + host := plugintest.NewHost(t, plugintest.WithConfig(map[string]map[string]any{ + "job.hello": {"enabled": true, "interval": "15ms", "greeting": "hi"}, + })) + + m := new(Module) + require.NoError(t, m.Provision(host)) + require.Equal(t, 15*time.Millisecond, m.cfg.Interval) + require.Equal(t, "hi", m.cfg.Greeting) + require.NoError(t, m.Validate()) +} + +// TestProvisionConfigDecodeError 验证私有配置类型错误时 Provision 返回错误。 +func TestProvisionConfigDecodeError(t *testing.T) { + host := plugintest.NewHost(t, plugintest.WithConfig(map[string]map[string]any{ + "job.hello": {"interval": "not-a-duration"}, + })) + + m := new(Module) + err := m.Provision(host) + require.Error(t, err) + require.Contains(t, err.Error(), "job.hello") +} + +// TestValidateRejectsInvalidConfig 验证非法配置被 Validate 拒绝。 +func TestValidateRejectsInvalidConfig(t *testing.T) { + tests := []struct { + name string + cfg Config + wantErr string + }{ + {"zero interval", Config{Interval: 0, Greeting: "hi"}, "interval must be positive"}, + {"negative interval", Config{Interval: -time.Second, Greeting: "hi"}, "interval must be positive"}, + {"empty greeting", Config{Interval: time.Second, Greeting: ""}, "greeting must not be empty"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + m := &Module{cfg: tt.cfg} + err := m.Validate() + require.Error(t, err) + require.Contains(t, err.Error(), tt.wantErr) + }) + } +} + +// TestStartPeriodicLogAndStop 验证 Start 后周期 debug 日志持续出现， +// Stop 后 goroutine 优雅退出。 +func TestStartPeriodicLogAndStop(t *testing.T) { + logOpt, logs := plugintest.WithObservedLogger() + host := plugintest.NewHost(t, logOpt, plugintest.WithConfig(map[string]map[string]any{ + "job.hello": {"interval": "5ms"}, + })) + + m := new(Module) + require.NoError(t, m.Provision(host)) + require.NoError(t, m.Validate()) + require.NoError(t, m.Start(context.Background())) + + require.Eventually(t, func() bool { + return logs.FilterMessage(m.cfg.Greeting).Len() >= 2 + }, 2*time.Second, 5*time.Millisecond, "periodic debug log should appear repeatedly") + + require.NoError(t, m.Stop(context.Background())) + select { + case <-m.done: + // goroutine 已退出。 + default: + t.Fatal("worker goroutine should have exited after Stop") + } + require.Equal(t, 1, logs.FilterMessage("hello module stopped").Len()) +} + +// TestStopWithoutStartIsNoop 验证未 Start 时 Stop 为 no-op。 +func TestStopWithoutStartIsNoop(t *testing.T) { + m := new(Module) + require.NoError(t, m.Provision(plugintest.NewHost(t))) + require.NoError(t, m.Stop(context.Background())) +} + +// TestRuntimeDefaultConfigIsNoop 验证缺省配置（modules: 子树为空）下， +// hello 模块保持 disabled：Build/Start/Stop 全程 no-op，无任何副作用。 +func TestRuntimeDefaultConfigIsNoop(t *testing.T) { + logOpt, logs := plugintest.WithObservedLogger() + rt := plugintest.RunLifecycle(t, &Module{}, plugintest.NewHost(t, logOpt), nil) + + require.NoError(t, rt.Stop(context.Background())) + + snap := rt.Snapshot() + require.Len(t, snap, 1) + require.Equal(t, ID, snap[0].ID) + require.False(t, snap[0].Enabled) + require.Equal(t, plugin.StateRegistered, snap[0].State) + require.Zero(t, logs.Len(), "disabled module must not produce any module logs") +} + +// TestRuntimeEnabledHelloLifecycle 验证启用 job.hello 后完整生命周期： +// Build（Provision/Validate）→ Start（周期日志出现）→ Stop（优雅退出）。 +func TestRuntimeEnabledHelloLifecycle(t *testing.T) { + logOpt, logs := plugintest.WithObservedLogger() + rt := plugintest.RunLifecycle(t, &Module{}, plugintest.NewHost(t, logOpt), map[string]map[string]any{ + "job.hello": {"enabled": true, "interval": "5ms"}, + }) + + require.Eventually(t, func() bool { + return logs.FilterMessage(defaultConfig().Greeting).Len() >= 1 + }, 2*time.Second, 5*time.Millisecond, "periodic debug log should appear after Start") + + require.NoError(t, rt.Stop(context.Background())) + snap := rt.Snapshot() + require.Len(t, snap, 1) + require.True(t, snap[0].Enabled) + require.Equal(t, plugin.StateStopped, snap[0].State) +} + +// TestRuntimeInvalidHelloConfigAbortsBuild 验证 hello 配置非法时 +// Validate 失败导致 Build 返回错误（启动中止，fail-fast）。 +func TestRuntimeInvalidHelloConfigAbortsBuild(t *testing.T) { + err := plugintest.BuildExpectingError(t, &Module{}, plugintest.NewHost(t), map[string]map[string]any{ + "job.hello": {"enabled": true, "interval": "-1s"}, + }) + require.Contains(t, err.Error(), `validate module "job.hello"`) + require.Contains(t, err.Error(), "interval must be positive") +} + +// TestRuntimeIgnoresUnknownConfiguredModule 验证配置中出现格式合法但 +// 未注册的模块 ID 时，Runtime 按内核语义忽略该项，不影响其他模块。 +func TestRuntimeIgnoresUnknownConfiguredModule(t *testing.T) { + rt := plugintest.RunLifecycle(t, &Module{}, plugintest.NewHost(t), map[string]map[string]any{ + "job.hello": {"enabled": true, "interval": "5ms"}, + "job.unknown": {"enabled": true}, + }) + + require.NoError(t, rt.Stop(context.Background())) + + snap := rt.Snapshot() + require.Len(t, snap, 1, "snapshot should only contain registered modules") + require.Equal(t, ID, snap[0].ID) +} diff --git a/backend/internal/modules/standard/imports.go b/backend/internal/modules/standard/imports.go new file mode 100644 index 0000000000..e2fda89b3b --- /dev/null +++ b/backend/internal/modules/standard/imports.go @@ -0,0 +1,12 @@ +// Package standard 是插件模块的唯一插装清单（instrumentation list）。 +// +// 所有需要编译进二进制的内置模块必须且只能在本文件中匿名 import： +// 模块包在 init() 期向包级默认注册表自注册，cmd/server 匿名 import +// 本包即完成全部内置模块的插装。新增/移除模块只需修改本文件， +// 使插装变更集中、可 review。 +package standard + +import ( + // job.hello：示例模块（默认 disabled），验证插件链路连通性。 + _ "github.com/Wei-Shaw/sub2api/internal/modules/hello" +) diff --git a/backend/internal/plugin/audit_regression_test.go b/backend/internal/plugin/audit_regression_test.go new file mode 100644 index 0000000000..3755c6f02e --- /dev/null +++ b/backend/internal/plugin/audit_regression_test.go @@ -0,0 +1,59 @@ +package plugin + +// Phase-1.5 对抗式审计（2026-06-11）发现的 P2 问题回归测试： +// A-1: 失败的 Build 不得被静默重跑（会对已 Provision 实例不 Stop 即覆盖） +// C-1: enabled 空值（YAML null）按"未显式配置"处理，走模块默认值 + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestRuntimeBuildFailureBlocksRetry 固化审计 A-1 修复： +// Build 失败后，二次 Build 与 Start 均被拒绝，且不重新驱动任何生命周期。 +func TestRuntimeBuildFailureBlocksRetry(t *testing.T) { + log := &lifecycleLog{} + okMod := &fakeModule{id: "job.a", enabledByDefault: true, log: log} + badMod := &fakeModule{id: "job.b", enabledByDefault: true, log: log, provisionErr: errors.New("boom")} + rt := newTestRuntime(t, Config{}, okMod, badMod) + + require.Error(t, rt.Build(), "首次 Build 应因 job.b Provision 失败而报错") + firstLog := log.snapshot() + + err := rt.Build() + require.Error(t, err) + require.Contains(t, err.Error(), "cannot be rebuilt", "失败后的二次 Build 应被明确拒绝") + require.Equal(t, firstLog, log.snapshot(), "二次 Build 不得重新 Provision 任何模块") + + require.Error(t, rt.Start(context.Background()), "Build 未成功时 Start 应被拒绝") +} + +// TestRuntimeBuildSuccessThenRebuildStillRejected 确认成功路径的原有 guard 语义不变。 +func TestRuntimeBuildSuccessThenRebuildStillRejected(t *testing.T) { + rt := newTestRuntime(t, Config{}, &plainModule{id: "job.ok"}) + require.NoError(t, rt.Build()) + err := rt.Build() + require.Error(t, err) + require.Contains(t, err.Error(), "already built") +} + +// TestParseConfigEnabledNilTreatedAsUnset 固化审计 C-1 修复： +// `enabled:`（YAML 空值 → nil）等价于未配置，三态语义回落到模块默认值， +// 且不影响同条目下私有配置的解析。 +func TestParseConfigEnabledNilTreatedAsUnset(t *testing.T) { + cfg, err := ParseConfig(map[string]map[string]any{ + "job.hello": {"enabled": nil, "greeting": "hi"}, + }) + require.NoError(t, err) + + mc, ok := cfg["job.hello"] + require.True(t, ok) + require.Nil(t, mc.Enabled, "enabled 空值应视为未显式配置") + require.Equal(t, "hi", mc.Raw["greeting"], "私有配置不受 enabled 空值影响") + + require.True(t, cfg.enabledFor(ModuleInfo{ID: "job.hello", EnabledByDefault: true})) + require.False(t, cfg.enabledFor(ModuleInfo{ID: "job.hello", EnabledByDefault: false})) +} diff --git a/backend/internal/plugin/config.go b/backend/internal/plugin/config.go new file mode 100644 index 0000000000..054ccd5db9 --- /dev/null +++ b/backend/internal/plugin/config.go @@ -0,0 +1,122 @@ +package plugin + +import ( + "fmt" + "strconv" + + "github.com/mitchellh/mapstructure" +) + +// enabledKey 是每个模块配置项中由内核保留解释的键，其余键全部属于模块私有配置。 +const enabledKey = "enabled" + +// ModuleConfig 是单个模块在 `modules:` 配置子树中的配置项。 +type ModuleConfig struct { + // Enabled 是模块的显式启停开关；nil 表示未显式配置， + // 由模块注册时声明的 ModuleInfo.EnabledByDefault 决定。 + Enabled *bool + + // Raw 是模块的私有配置（去掉 enabled 键之后的剩余部分）， + // 由 Config.Of 解码到模块自定义的配置 struct，内核不解释其内容。 + Raw map[string]any +} + +// Config 是 `modules:` 配置子树的解析结果：模块 ID → 模块配置。 +// +// 缺省（子树不存在）时为空 map，所有模块按各自默认 enabled 值运行， +// 行为与未引入插件内核前完全一致。 +type Config map[ModuleID]ModuleConfig + +// ParseConfig 将全局配置中 `modules:` 子树的原始映射解析为 Config。 +// +// raw 的顶层 key 必须是合法的模块 ID 字面量（如 "job.hello"，YAML 中带点的 +// 完整 key），value 为该模块的配置映射；其中 enabled 键被内核解释为启停开关 +// （接受 bool 或可解析为 bool 的字符串），其余键原样保留为模块私有配置。 +// +// 允许出现"格式合法但未注册"的模块 ID（例如不同编译变体的配置共用）， +// 这类配置项会被 Runtime 忽略；但 ID 格式非法或 enabled 类型错误会返回错误， +// 以便尽早暴露配置笔误。 +// +// 注意：全局配置经 Viper 加载时，所有 key（含模块 ID 与私有配置键）会被 +// 静默小写化——配置文件里写 "Job.Hello" 实际到达这里时已是 "job.hello"。 +// 因此模块 ID 与私有配置 struct 的 mapstructure 标签必须全小写。 +func ParseConfig(raw map[string]map[string]any) (Config, error) { + cfg := make(Config, len(raw)) + for key, entry := range raw { + id := ModuleID(key) + if err := id.validate(); err != nil { + return nil, fmt.Errorf("plugin config: invalid module ID in modules config: %w", err) + } + mc := ModuleConfig{Raw: make(map[string]any, len(entry))} + for k, v := range entry { + if k == enabledKey { + // YAML 中写空的 `enabled:`（值为 null）按"未显式配置"处理， + // 走模块默认值——与 Enabled=nil 的三态语义一致（审计 C-1）。 + if v == nil { + continue + } + b, err := toBool(v) + if err != nil { + return nil, fmt.Errorf("plugin config: module %q: invalid enabled value: %w", id, err) + } + mc.Enabled = &b + continue + } + mc.Raw[k] = v + } + cfg[id] = mc + } + return cfg, nil +} + +// Of 将模块 id 的 raw 私有配置解码到 out（必须是指向配置 struct 的指针）。 +// +// 解码行为与项目全局配置（Viper 默认解码器）保持一致： +// mapstructure 标签、弱类型转换、字符串到 time.Duration / 逗号分隔切片的钩子。 +// 模块未配置或私有配置为空时不修改 out（模块应自带默认值并返回 nil）； +// 类型不匹配等解码失败时返回包含模块 ID 的错误。 +func (c Config) Of(id ModuleID, out any) error { + mc, ok := c[id] + if !ok || len(mc.Raw) == 0 { + return nil + } + decoder, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{ + Result: out, + WeaklyTypedInput: true, + DecodeHook: mapstructure.ComposeDecodeHookFunc( + mapstructure.StringToTimeDurationHookFunc(), + mapstructure.StringToSliceHookFunc(","), + ), + }) + if err != nil { + return fmt.Errorf("plugin config: build decoder for module %q: %w", id, err) + } + if err := decoder.Decode(mc.Raw); err != nil { + return fmt.Errorf("plugin config: decode config for module %q: %w", id, err) + } + return nil +} + +// enabledFor 解析模块的最终启停状态：显式配置优先，否则用注册时声明的默认值。 +func (c Config) enabledFor(info ModuleInfo) bool { + if mc, ok := c[info.ID]; ok && mc.Enabled != nil { + return *mc.Enabled + } + return info.EnabledByDefault +} + +// toBool 将 enabled 配置值弱类型转换为 bool（接受 bool 与 strconv.ParseBool 可解析的字符串）。 +func toBool(v any) (bool, error) { + switch val := v.(type) { + case bool: + return val, nil + case string: + b, err := strconv.ParseBool(val) + if err != nil { + return false, fmt.Errorf("cannot parse %q as bool", val) + } + return b, nil + default: + return false, fmt.Errorf("expected bool, got %T", v) + } +} diff --git a/backend/internal/plugin/config_test.go b/backend/internal/plugin/config_test.go new file mode 100644 index 0000000000..0b15131463 --- /dev/null +++ b/backend/internal/plugin/config_test.go @@ -0,0 +1,135 @@ +package plugin + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestParseConfigEnabledTriState(t *testing.T) { + cfg, err := ParseConfig(map[string]map[string]any{ + "job.on": {"enabled": true}, + "job.off": {"enabled": false}, + "job.unset": {"greeting": "hi"}, + "job.strbool": {"enabled": "true"}, // 弱类型：字符串 bool（如来自环境变量） + }) + require.NoError(t, err) + + require.NotNil(t, cfg["job.on"].Enabled) + require.True(t, *cfg["job.on"].Enabled) + + require.NotNil(t, cfg["job.off"].Enabled) + require.False(t, *cfg["job.off"].Enabled) + + require.Nil(t, cfg["job.unset"].Enabled, "未显式配置 enabled 时必须为 nil（用模块默认值）") + + require.NotNil(t, cfg["job.strbool"].Enabled) + require.True(t, *cfg["job.strbool"].Enabled) +} + +func TestParseConfigSeparatesEnabledFromRaw(t *testing.T) { + cfg, err := ParseConfig(map[string]map[string]any{ + "job.hello": {"enabled": true, "greeting": "hi", "interval": "5s"}, + }) + require.NoError(t, err) + + mc := cfg["job.hello"] + require.Equal(t, map[string]any{"greeting": "hi", "interval": "5s"}, mc.Raw, + "enabled 键属于内核，不得混入模块私有配置") +} + +func TestParseConfigErrors(t *testing.T) { + t.Run("invalid module ID", func(t *testing.T) { + _, err := ParseConfig(map[string]map[string]any{"Job.Hello": {}}) + require.Error(t, err) + require.Contains(t, err.Error(), "Job.Hello") + }) + t.Run("invalid enabled type", func(t *testing.T) { + _, err := ParseConfig(map[string]map[string]any{"job.hello": {"enabled": 123}}) + require.Error(t, err) + require.Contains(t, err.Error(), "job.hello") + }) + t.Run("invalid enabled string", func(t *testing.T) { + _, err := ParseConfig(map[string]map[string]any{"job.hello": {"enabled": "not-a-bool"}}) + require.Error(t, err) + }) + t.Run("unregistered but valid ID is allowed", func(t *testing.T) { + cfg, err := ParseConfig(map[string]map[string]any{"future.module": {"enabled": true}}) + require.NoError(t, err) + require.Contains(t, cfg, ModuleID("future.module")) + }) +} + +type helloModuleConfig struct { + Greeting string `mapstructure:"greeting"` + Interval time.Duration `mapstructure:"interval"` + Targets []string `mapstructure:"targets"` + Limit int `mapstructure:"limit"` +} + +func TestConfigOfDecodesRawIntoStruct(t *testing.T) { + cfg, err := ParseConfig(map[string]map[string]any{ + "job.hello": { + "enabled": true, + "greeting": "hi", + "interval": "5s", // 字符串 → time.Duration（与 viper 默认钩子一致） + "targets": "a,b,c", // 逗号分隔 → 切片（与 viper 默认钩子一致） + "limit": "42", // 弱类型：字符串 → int + }, + }) + require.NoError(t, err) + + var out helloModuleConfig + require.NoError(t, cfg.Of("job.hello", &out)) + require.Equal(t, "hi", out.Greeting) + require.Equal(t, 5*time.Second, out.Interval) + require.Equal(t, []string{"a", "b", "c"}, out.Targets) + require.Equal(t, 42, out.Limit) +} + +func TestConfigOfMissingModuleLeavesOutUntouched(t *testing.T) { + cfg := Config{} + out := helloModuleConfig{Greeting: "default-value"} + require.NoError(t, cfg.Of("job.absent", &out)) + require.Equal(t, "default-value", out.Greeting, "模块未配置时不得修改 out（保留模块默认值）") +} + +func TestConfigOfTypeErrorContainsModuleID(t *testing.T) { + cfg, err := ParseConfig(map[string]map[string]any{ + "job.hello": {"limit": "not-an-int"}, + }) + require.NoError(t, err) + + var out helloModuleConfig + err = cfg.Of("job.hello", &out) + require.Error(t, err) + require.Contains(t, err.Error(), "job.hello", "解码错误必须包含模块 ID") +} + +func TestConfigEnabledFor(t *testing.T) { + on, off := true, false + cfg := Config{ + "job.explicit-on": {Enabled: &on}, + "job.explicit-off": {Enabled: &off}, + "job.mentioned": {}, + } + + tests := []struct { + name string + info ModuleInfo + want bool + }{ + {"explicit on overrides default off", ModuleInfo{ID: "job.explicit-on", EnabledByDefault: false}, true}, + {"explicit off overrides default on", ModuleInfo{ID: "job.explicit-off", EnabledByDefault: true}, false}, + {"mentioned without enabled uses default true", ModuleInfo{ID: "job.mentioned", EnabledByDefault: true}, true}, + {"mentioned without enabled uses default false", ModuleInfo{ID: "job.mentioned", EnabledByDefault: false}, false}, + {"absent uses default true", ModuleInfo{ID: "job.absent", EnabledByDefault: true}, true}, + {"absent uses default false", ModuleInfo{ID: "job.absent", EnabledByDefault: false}, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.want, cfg.enabledFor(tt.info)) + }) + } +} diff --git a/backend/internal/plugin/host.go b/backend/internal/plugin/host.go new file mode 100644 index 0000000000..fa22daee35 --- /dev/null +++ b/backend/internal/plugin/host.go @@ -0,0 +1,34 @@ +package plugin + +import ( + "github.com/redis/go-redis/v9" + "go.uber.org/zap" + + ent "github.com/Wei-Shaw/sub2api/ent" +) + +// Host 是模块可使用的宿主能力面（纯数据结构，由宿主装配后注入 Provision）。 +// +// 边界纪律：Host 只暴露 ports 级基础能力，禁止暴露任何具体 Service； +// 本阶段仅 Logger / ConfigOf / DB / Redis 四项，后续每新增一项 +// 必须在插件化 ROADMAP 决策记录中登记，防止 Host 膨胀为上帝对象。 +type Host struct { + // Logger 是宿主提供的结构化日志器。模块应通过它输出日志， + // 而不是自建全局 logger。 + Logger *zap.Logger + + // ConfigOf 将 `modules:` 配置子树中 id 对应的 raw 私有配置解码到 out + // （out 必须是指向模块自定义配置 struct 的指针）。 + // 模块未配置时不修改 out（模块应自带默认值）；解码类型错误时返回错误。 + // + // 该字段通常由宿主装配为 Config.Of；若装配时留空， + // NewRuntime 会默认绑定到 Runtime 自身持有的 Config。 + ConfigOf func(id ModuleID, out any) error + + // DB 是 Ent ORM 客户端（数据库访问能力）。 + DB *ent.Client + + // Redis 是 go-redis 客户端（缓存/队列能力）。 + // 使用 UniversalClient 接口以兼容单机与集群部署形态。 + Redis redis.UniversalClient +} diff --git a/backend/internal/plugin/module.go b/backend/internal/plugin/module.go new file mode 100644 index 0000000000..b080b60547 --- /dev/null +++ b/backend/internal/plugin/module.go @@ -0,0 +1,133 @@ +// Package plugin 实现 Caddy 式进程内插件内核。 +// +// 内核由四部分组成： +// - 模块注册表（registry.go）：命名空间化的模块目录，模块包在 init() 期通过 +// RegisterModule 自注册，编译期插装清单（internal/modules/standard）决定哪些模块被编译进来； +// - 配置子树（config.go）：来自全局配置 `modules:` 的每模块配置（enabled + raw 私有配置）； +// - 宿主能力面（host.go）：模块可使用的全部宿主能力，只暴露 ports 级接口； +// - 生命周期驱动器（runtime.go）：按注册表稳定序实例化 enabled 模块并驱动 +// Provision → Validate → Start → Stop（逆序）生命周期。 +// +// 注册表只是"目录"，模块的实例化与生命周期统一由 Runtime 驱动。 +package plugin + +import ( + "context" + "fmt" + "strings" +) + +// ModuleID 是模块的全局唯一标识，采用点分层级命名， +// 形如 "job.hello"、"gateway.platform.anthropic"。 +// +// 最后一段为模块名（Name），之前的部分为命名空间（Namespace）。 +// 每一段只允许 [a-z0-9_-] 字符且不能为空。 +type ModuleID string + +// Namespace 返回模块 ID 的命名空间部分（去掉最后一段）。 +// +// 例如 "gateway.platform.anthropic" 的命名空间为 "gateway.platform"； +// 单段 ID（如 "hello"）的命名空间为空字符串。 +func (id ModuleID) Namespace() string { + lastDot := strings.LastIndex(string(id), ".") + if lastDot < 0 { + return "" + } + return string(id)[:lastDot] +} + +// Name 返回模块 ID 的最后一段，即模块名。 +// +// 例如 "gateway.platform.anthropic" 的模块名为 "anthropic"。 +func (id ModuleID) Name() string { + lastDot := strings.LastIndex(string(id), ".") + if lastDot < 0 { + return string(id) + } + return string(id)[lastDot+1:] +} + +// validate 校验模块 ID 格式：非空、点分、各段为非空的 [a-z0-9_-]+。 +// 校验规则有意保持宽松，不做长度等额外限制。 +func (id ModuleID) validate() error { + if id == "" { + return fmt.Errorf("module ID is empty") + } + for _, segment := range strings.Split(string(id), ".") { + if segment == "" { + return fmt.Errorf("module ID %q contains an empty segment", id) + } + for _, r := range segment { + if (r < 'a' || r > 'z') && (r < '0' || r > '9') && r != '_' && r != '-' { + return fmt.Errorf("module ID %q contains invalid character %q (allowed: [a-z0-9_-] per segment)", id, r) + } + } + } + return nil +} + +// ModuleInfo 描述一个模块的注册信息（模块目录中的一条记录）。 +type ModuleInfo struct { + // ID 是模块的全局唯一标识，见 ModuleID。 + ID ModuleID + + // New 返回该模块的一个新实例。 + // 每次 Runtime.Build 都会通过 New 创建全新实例，模块不应依赖包级可变状态。 + New func() Module + + // EnabledByDefault 声明该模块在 `modules:` 配置未显式设置 enabled 时的默认启停状态。 + // + // 设计说明（默认 enabled 的声明方式）：默认值由模块在注册时通过本字段自行声明， + // 而不是由内核统一约定。这样"缺省配置 = 内置模块全部按各自默认 enabled 值运行" + // 的语义完全由各模块的注册信息决定；零值 false 意味着新模块默认不启用， + // 保证引入新模块不会在未配置时改变现有行为（与 Phase-1 零行为变更要求一致）。 + EnabledByDefault bool +} + +// Module 是所有插件模块必须实现的最小接口。 +// +// 模块通常是一个 struct，在所属包的 init() 中调用 RegisterModule 完成自注册。 +// 除 Module 外，模块可按需实现以下可选生命周期接口： +// Provisioner、Validator、Starter、Stopper。 +type Module interface { + // ModuleInfo 返回模块的注册信息。该方法必须无副作用且可在零值实例上调用。 + ModuleInfo() ModuleInfo +} + +// Provisioner 是可选生命周期接口：在模块实例化后、校验前装配依赖。 +// +// Provision 在 Runtime.Build 阶段被调用，模块应在此从 Host 获取所需能力 +// （Logger / ConfigOf / DB / Redis）并完成自身初始化。失败将中止启动。 +type Provisioner interface { + Provision(host *Host) error +} + +// Validator 是可选生命周期接口：在 Provision 之后校验模块配置与状态。 +// +// Validate 在 Runtime.Build 阶段被调用，失败将中止启动。 +type Validator interface { + Validate() error +} + +// Starter 是可选生命周期接口：启动模块的后台工作（goroutine、监听等）。 +// +// Start 在 Runtime.Start 阶段按注册表稳定序被调用，失败将中止启动并 +// 触发对已启动模块的逆序 Stop 回滚。 +// +// 契约：实现必须快速返回——长任务应自行启动 goroutine（参照 hello 模块）。 +// Runtime 在持锁状态下调用 Start/Stop，阻塞的实现会同时阻塞 Snapshot +// （即 admin 模块可观测接口）与其他生命周期操作。 +type Starter interface { + Start(ctx context.Context) error +} + +// Stopper 是可选生命周期接口：停止模块并释放资源。 +// +// Stop 在 Runtime.Stop 阶段按注册表稳定序的逆序被调用， +// 实现必须尊重 ctx 的 deadline（与进程优雅关闭超时协同）。 +// +// 契约：与 Start 相同，实现必须快速返回（等待自身 goroutine 退出时 +// 用 select 配合 ctx.Done()，参照 hello 模块），原因见 Starter 注释。 +type Stopper interface { + Stop(ctx context.Context) error +} diff --git a/backend/internal/plugin/module_test.go b/backend/internal/plugin/module_test.go new file mode 100644 index 0000000000..948a56212d --- /dev/null +++ b/backend/internal/plugin/module_test.go @@ -0,0 +1,55 @@ +package plugin + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestModuleIDNamespaceAndName(t *testing.T) { + tests := []struct { + id ModuleID + namespace string + name string + }{ + {id: "job.hello", namespace: "job", name: "hello"}, + {id: "gateway.platform.anthropic", namespace: "gateway.platform", name: "anthropic"}, + {id: "hello", namespace: "", name: "hello"}, + {id: "payment.provider.stripe-checkout", namespace: "payment.provider", name: "stripe-checkout"}, + } + for _, tt := range tests { + t.Run(string(tt.id), func(t *testing.T) { + require.Equal(t, tt.namespace, tt.id.Namespace()) + require.Equal(t, tt.name, tt.id.Name()) + }) + } +} + +func TestModuleIDValidate(t *testing.T) { + valid := []ModuleID{ + "hello", + "job.hello", + "gateway.platform.anthropic", + "a.b-c.d_e", + "ns.mod123", + } + for _, id := range valid { + require.NoError(t, id.validate(), "id %q should be valid", id) + } + + invalid := []ModuleID{ + "", + ".", + ".hello", + "hello.", + "job..hello", + "Job.hello", // 大写 + "job.hello!", // 非法字符 + "job. hello", // 空格 + "job.héllo", // 非 ASCII + "job.hello.World", // 末段大写 + } + for _, id := range invalid { + require.Error(t, id.validate(), "id %q should be invalid", id) + } +} diff --git a/backend/internal/plugin/plugintest/host.go b/backend/internal/plugin/plugintest/host.go new file mode 100644 index 0000000000..ea22054d42 --- /dev/null +++ b/backend/internal/plugin/plugintest/host.go @@ -0,0 +1,103 @@ +// Package plugintest 提供插件模块的测试夹具（test fixtures）。 +// +// 本包是模块作者的测试入口：NewHost 一行构造可按需附带 miniredis、 +// observed logger、配置子树的宿主能力面；RunLifecycle 一行驱动 +// "隔离注册 → 解析配置 → Build → Start → t.Cleanup 自动 Stop" 的完整 +// 生命周期。命名与用法风格对齐标准库 net/http/httptest。 +// +// 使用约束：本包 import "testing"，仅供 _test.go 文件 import， +// 禁止进入任何生产代码的依赖面。该约束以本声明 + 代码评审人工约定保证 +// （未配置 depguard 规则；若被非 _test 文件 import，testing 包会被 +// 连带编入生产二进制，评审时一票否决）。 +// +// 夹具纪律：夹具自身绝不吞错——Build/Start/Stop 或配置解析失败一律经 +// tb.Fatalf / tb.Errorf 报告，且错误信息包含被测模块 ID。 +package plugintest + +import ( + "testing" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + + "github.com/Wei-Shaw/sub2api/internal/plugin" +) + +// Option 定制 NewHost 构造的 Host。Option 在 NewHost 内部被依次应用， +// tb 即传给 NewHost 的测试句柄；Option 内部的失败一律经 tb 报告（不吞错）。 +type Option func(tb testing.TB, h *plugin.Host) + +// NewHost 构造模块测试用的宿主能力面，默认零依赖即可使用： +// nop logger、空 ConfigOf（对任何模块 ID 都不报错、不修改 out）、 +// nil DB、nil Redis。按需叠加 WithRedis / WithConfig / WithObservedLogger。 +// +// Host 是纯数据结构：本包未覆盖的少见场景（如注入 enttest 客户端、 +// 自定义 logger）可在 NewHost 返回后直接对字段赋值，无需新增 Option。 +func NewHost(tb testing.TB, opts ...Option) *plugin.Host { + tb.Helper() + host := &plugin.Host{ + Logger: zap.NewNop(), + ConfigOf: plugin.Config{}.Of, + } + for _, opt := range opts { + opt(tb, host) + } + return host +} + +// WithRedis 为 Host 附带一个由 miniredis 支撑的 go-redis 客户端。 +// +// miniredis 实例与 redis 客户端的生命周期均挂在 tb.Cleanup 上自动清理； +// tb 通常就是传给 NewHost 的同一个测试句柄。 +func WithRedis(tb testing.TB) Option { + return func(_ testing.TB, h *plugin.Host) { + tb.Helper() + mr := miniredis.RunT(tb) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + tb.Cleanup(func() { + if err := rdb.Close(); err != nil { + tb.Errorf("plugintest: close redis client: %v", err) + } + }) + h.Redis = rdb + } +} + +// WithConfig 将 Host.ConfigOf 绑定为 raw（与全局配置 `modules:` 子树同形状） +// 的解析结果，供直接调用 Provision 等不经 Runtime 的测试读取模块私有配置。 +// raw 解析失败（模块 ID 格式非法、enabled 类型错误）立即 tb.Fatal。 +// +// 注意：经 RunLifecycle / BuildOnly 驱动生命周期时，模块配置以其 raw 参数 +// 为唯一来源（会覆盖本 Option 设置的 ConfigOf），两者不要混用。 +func WithConfig(raw map[string]map[string]any) Option { + return func(tb testing.TB, h *plugin.Host) { + tb.Helper() + cfg, err := plugin.ParseConfig(raw) + if err != nil { + tb.Fatalf("plugintest: parse modules config: %v", err) + } + h.ConfigOf = cfg.Of + } +} + +// WithObservedLogger 将 Host.Logger 替换为 debug 级别的 zap observer logger， +// 并把可断言日志记录的句柄作为第二返回值一并返回： +// +// logOpt, logs := plugintest.WithObservedLogger() +// host := plugintest.NewHost(t, logOpt) +// // ... 驱动模块 ... +// require.Equal(t, 1, logs.FilterMessage("hello").Len()) +// +// 设计说明：observer 句柄随 Option 直接返回，而非提供 Logs(host) 式的查表 +// 函数——调用方一行解构即可拿到句柄，包内也无需维护 Host → observer 的 +// 映射状态（与 PHASE_PLAN 草案中 Logs(t) 形态的差异已在任务交付说明登记）。 +func WithObservedLogger() (Option, *observer.ObservedLogs) { + core, logs := observer.New(zapcore.DebugLevel) + logger := zap.New(core) + return func(_ testing.TB, h *plugin.Host) { + h.Logger = logger + }, logs +} diff --git a/backend/internal/plugin/plugintest/plugintest_test.go b/backend/internal/plugin/plugintest/plugintest_test.go new file mode 100644 index 0000000000..6e45464542 --- /dev/null +++ b/backend/internal/plugin/plugintest/plugintest_test.go @@ -0,0 +1,318 @@ +package plugintest_test + +import ( + "context" + "errors" + "fmt" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/Wei-Shaw/sub2api/internal/plugin" + "github.com/Wei-Shaw/sub2api/internal/plugin/plugintest" +) + +// fixtureConfig 是 fakeModule 的私有配置，用于验证 raw 配置经 host.ConfigOf 抵达模块。 +type fixtureConfig struct { + Greeting string `mapstructure:"greeting"` +} + +// fakeModule 是自测用模块：New 返回自身以便直接断言生命周期调用计数， +// 可按需注入各阶段错误。生命周期由 Runtime 在测试 goroutine 上同步驱动 +// （cleanup 与 t.Run 完成均有 happens-before 保证），计数无需加锁。 +type fakeModule struct { + id plugin.ModuleID + enabledByDefault bool + + provisionErr error + validateErr error + startErr error + stopErr error + + provisions int + validates int + starts int + stops int + host *plugin.Host + cfg fixtureConfig +} + +func (m *fakeModule) ModuleInfo() plugin.ModuleInfo { + return plugin.ModuleInfo{ + ID: m.id, + EnabledByDefault: m.enabledByDefault, + New: func() plugin.Module { return m }, + } +} + +func (m *fakeModule) Provision(host *plugin.Host) error { + m.provisions++ + m.host = host + if host != nil && host.ConfigOf != nil { + if err := host.ConfigOf(m.id, &m.cfg); err != nil { + return err + } + } + return m.provisionErr +} + +func (m *fakeModule) Validate() error { m.validates++; return m.validateErr } + +func (m *fakeModule) Start(context.Context) error { m.starts++; return m.startErr } + +func (m *fakeModule) Stop(context.Context) error { m.stops++; return m.stopErr } + +// errFatalCalled 是 recordingTB.Fatalf 用于模拟"立即中止"语义的 panic 哨兵。 +var errFatalCalled = errors.New("recordingTB: Fatalf called") + +// recordingTB 包装真实 *testing.T，捕获夹具的 Fatalf / Errorf / Logf / Cleanup +// 调用，用于验证夹具在错误路径上不吞错。Fatalf 记录信息后 panic(errFatalCalled) +// 模拟真实 t.Fatalf 的中止语义，由 expectFatal 捕获恢复。 +type recordingTB struct { + testing.TB + fatals []string + errs []string + logLines []string + cleanups []func() +} + +func (r *recordingTB) Helper() {} + +func (r *recordingTB) Logf(format string, args ...any) { + r.logLines = append(r.logLines, fmt.Sprintf(format, args...)) +} + +func (r *recordingTB) Errorf(format string, args ...any) { + r.errs = append(r.errs, fmt.Sprintf(format, args...)) +} + +func (r *recordingTB) Fatalf(format string, args ...any) { + r.fatals = append(r.fatals, fmt.Sprintf(format, args...)) + panic(errFatalCalled) +} + +func (r *recordingTB) Cleanup(f func()) { r.cleanups = append(r.cleanups, f) } + +// runCleanups 按注册逆序（与真实 testing 一致）执行捕获的 cleanup。 +func (r *recordingTB) runCleanups() { + for i := len(r.cleanups) - 1; i >= 0; i-- { + r.cleanups[i]() + } +} + +// expectFatal 执行 fn 并断言它经 recordingTB.Fatalf 中止，返回最后一条 fatal 信息。 +func expectFatal(t *testing.T, rec *recordingTB, fn func()) (msg string) { + t.Helper() + defer func() { + if r := recover(); r != nil { + if r != errFatalCalled { + panic(r) // 非哨兵 panic 原样上抛 + } + require.NotEmpty(t, rec.fatals) + msg = rec.fatals[len(rec.fatals)-1] + } + }() + fn() + t.Fatal("expected the fixture to call Fatalf, but it returned normally") + return "" +} + +func TestNewHostDefaults(t *testing.T) { + host := plugintest.NewHost(t) + require.NotNil(t, host.Logger, "默认应提供 nop logger 而非 nil") + require.NotNil(t, host.ConfigOf, "默认应提供空 ConfigOf 而非 nil") + require.Nil(t, host.DB) + require.Nil(t, host.Redis) + + // 空 ConfigOf：对任意模块 ID 不报错、不修改 out（模块默认值得以保留）。 + cfg := fixtureConfig{Greeting: "default"} + require.NoError(t, host.ConfigOf("job.anything", &cfg)) + require.Equal(t, "default", cfg.Greeting) +} + +func TestWithRedis(t *testing.T) { + host := plugintest.NewHost(t, plugintest.WithRedis(t)) + require.NotNil(t, host.Redis) + + ctx := context.Background() + require.NoError(t, host.Redis.Set(ctx, "k", "v", 0).Err()) + got, err := host.Redis.Get(ctx, "k").Result() + require.NoError(t, err) + require.Equal(t, "v", got) +} + +func TestWithConfig(t *testing.T) { + host := plugintest.NewHost(t, plugintest.WithConfig(map[string]map[string]any{ + "job.fixture": {"enabled": true, "greeting": "hi"}, + })) + + var cfg fixtureConfig + require.NoError(t, host.ConfigOf("job.fixture", &cfg)) + require.Equal(t, "hi", cfg.Greeting, "私有配置应可经 ConfigOf 解码（enabled 键被内核剥离）") +} + +func TestWithConfigParseFailureFatals(t *testing.T) { + rec := &recordingTB{TB: t} + msg := expectFatal(t, rec, func() { + plugintest.NewHost(rec, plugintest.WithConfig(map[string]map[string]any{ + "INVALID ID": {}, + })) + }) + require.Contains(t, msg, "parse modules config") +} + +func TestWithObservedLogger(t *testing.T) { + logOpt, logs := plugintest.WithObservedLogger() + host := plugintest.NewHost(t, logOpt) + + host.Logger.Debug("observed-ping") + require.Equal(t, 1, logs.FilterMessage("observed-ping").Len()) +} + +func TestRunLifecycleSuccessAndAutoStop(t *testing.T) { + m := &fakeModule{id: "job.fixture"} + t.Run("lifecycle", func(t *testing.T) { + rt := plugintest.RunLifecycle(t, m, plugintest.NewHost(t), map[string]map[string]any{ + "job.fixture": {"enabled": true, "greeting": "hi"}, + }) + + require.Equal(t, 1, m.provisions) + require.Equal(t, 1, m.validates) + require.Equal(t, 1, m.starts) + require.Zero(t, m.stops) + require.Equal(t, "hi", m.cfg.Greeting, "raw 配置应经 host.ConfigOf 抵达模块") + require.NotNil(t, m.host) + + snap := rt.Snapshot() + require.Len(t, snap, 1, "隔离注册表应只含被测模块") + require.Equal(t, plugin.StateRunning, snap[0].State) + }) + require.Equal(t, 1, m.stops, "t.Cleanup 应自动 Stop 模块") +} + +func TestRunLifecycleRawOverridesHostConfig(t *testing.T) { + m := &fakeModule{id: "job.fixture"} + host := plugintest.NewHost(t, plugintest.WithConfig(map[string]map[string]any{ + "job.fixture": {"greeting": "from-with-config"}, + })) + plugintest.RunLifecycle(t, m, host, map[string]map[string]any{ + "job.fixture": {"enabled": true, "greeting": "from-raw"}, + }) + require.Equal(t, "from-raw", m.cfg.Greeting, + "RunLifecycle 的 raw 是配置唯一来源，应覆盖 WithConfig 的旧绑定") +} + +func TestRunLifecycleNilHostUsesDefaultHost(t *testing.T) { + m := &fakeModule{id: "job.fixture"} + plugintest.RunLifecycle(t, m, nil, map[string]map[string]any{ + "job.fixture": {"enabled": true}, + }) + require.NotNil(t, m.host, "nil host 应被替换为默认 Host") + require.NotNil(t, m.host.Logger) + require.NotNil(t, m.host.ConfigOf) +} + +func TestRunLifecycleExplicitStopThenCleanupIsIdempotent(t *testing.T) { + m := &fakeModule{id: "job.fixture"} + t.Run("lifecycle", func(t *testing.T) { + rt := plugintest.RunLifecycle(t, m, nil, map[string]map[string]any{ + "job.fixture": {"enabled": true}, + }) + require.NoError(t, rt.Stop(context.Background())) + require.Equal(t, 1, m.stops) + }) + require.Equal(t, 1, m.stops, "测试已显式 Stop 时，cleanup 的二次 Stop 应为 no-op") +} + +func TestRunLifecycleDisabledModuleIsNoop(t *testing.T) { + m := &fakeModule{id: "job.fixture"} // EnabledByDefault=false 且 raw 未显式 enabled + rt := plugintest.RunLifecycle(t, m, nil, nil) + + require.Zero(t, m.provisions, "disabled 模块不应经历任何生命周期（语义与生产一致，不强制 enable）") + snap := rt.Snapshot() + require.Len(t, snap, 1) + require.False(t, snap[0].Enabled) + require.Equal(t, plugin.StateRegistered, snap[0].State) +} + +func TestBuildOnlyDoesNotStart(t *testing.T) { + m := &fakeModule{id: "job.fixture"} + t.Run("build", func(t *testing.T) { + rt := plugintest.BuildOnly(t, m, nil, map[string]map[string]any{ + "job.fixture": {"enabled": true}, + }) + require.Equal(t, 1, m.provisions) + require.Equal(t, 1, m.validates) + require.Zero(t, m.starts) + require.Equal(t, plugin.StateProvisioned, rt.Snapshot()[0].State) + }) + require.Zero(t, m.stops, "未 Start 的模块在 cleanup 兜底 Stop 中不应被调用 Stop") +} + +func TestBuildOnlyLogsDisabledHint(t *testing.T) { + rec := &recordingTB{TB: t} + m := &fakeModule{id: "job.fixture"} + plugintest.BuildOnly(rec, m, nil, nil) + + require.NotEmpty(t, rec.logLines, "disabled 模块应输出提示日志") + require.Contains(t, rec.logLines[0], "disabled") + require.Contains(t, rec.logLines[0], `"job.fixture"`) + rec.runCleanups() + require.Empty(t, rec.errs) +} + +func TestRunLifecycleBuildFailureFatalsWithModuleID(t *testing.T) { + rec := &recordingTB{TB: t} + m := &fakeModule{id: "job.fixture", enabledByDefault: true, provisionErr: errors.New("provision boom")} + msg := expectFatal(t, rec, func() { + plugintest.RunLifecycle(rec, m, nil, nil) + }) + require.Contains(t, msg, `"job.fixture"`, "Build 失败信息必须包含模块 ID") + require.Contains(t, msg, "provision boom") +} + +func TestRunLifecycleStartFailureFatalsWithModuleID(t *testing.T) { + rec := &recordingTB{TB: t} + m := &fakeModule{id: "job.fixture", enabledByDefault: true, startErr: errors.New("start boom")} + msg := expectFatal(t, rec, func() { + plugintest.RunLifecycle(rec, m, nil, nil) + }) + require.Contains(t, msg, `"job.fixture"`, "Start 失败信息必须包含模块 ID") + require.Contains(t, msg, "start boom") + + // Start 失败的模块未进入 running，cleanup 兜底 Stop 应为无错误的 no-op。 + rec.runCleanups() + require.Empty(t, rec.errs) + require.Zero(t, m.stops) +} + +func TestCleanupStopFailureReportedWithModuleID(t *testing.T) { + rec := &recordingTB{TB: t} + m := &fakeModule{id: "job.fixture", enabledByDefault: true, stopErr: errors.New("stop boom")} + plugintest.RunLifecycle(rec, m, nil, nil) + require.Empty(t, rec.fatals) + require.NotEmpty(t, rec.cleanups) + + rec.runCleanups() + require.Len(t, rec.errs, 1, "cleanup Stop 失败必须经 Errorf 报告（不吞错）") + require.Contains(t, rec.errs[0], `"job.fixture"`) + require.Contains(t, rec.errs[0], "stop boom") +} + +func TestBuildExpectingErrorReturnsBuildError(t *testing.T) { + m := &fakeModule{id: "job.fixture", enabledByDefault: true, validateErr: errors.New("validate boom")} + err := plugintest.BuildExpectingError(t, m, nil, nil) + require.Error(t, err) + require.Contains(t, err.Error(), `"job.fixture"`) + require.Contains(t, err.Error(), "validate boom") +} + +func TestBuildExpectingErrorFatalsOnUnexpectedSuccess(t *testing.T) { + rec := &recordingTB{TB: t} + m := &fakeModule{id: "job.fixture", enabledByDefault: true} + msg := expectFatal(t, rec, func() { + _ = plugintest.BuildExpectingError(rec, m, nil, nil) + }) + require.Contains(t, msg, "expected Build") + require.Contains(t, msg, `"job.fixture"`) +} diff --git a/backend/internal/plugin/plugintest/runtime.go b/backend/internal/plugin/plugintest/runtime.go new file mode 100644 index 0000000000..4757dc1180 --- /dev/null +++ b/backend/internal/plugin/plugintest/runtime.go @@ -0,0 +1,120 @@ +package plugintest + +import ( + "context" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/plugin" +) + +// stopTimeout 是 t.Cleanup 自动 Stop 的超时上限：模块 Stop 悬挂时， +// 测试以包含模块 ID 的明确错误失败，而不是吊死整个 go test 进程。 +const stopTimeout = 10 * time.Second + +// RunLifecycle 驱动单个模块的完整生命周期：在隔离注册表中注册 m → +// 解析 raw 配置 → Build（Provision / Validate）→ Start。任一步失败立即 +// tb.Fatal（错误信息含模块 ID）。测试结束时 t.Cleanup 自动 Stop 并断言无 +// 错误；测试中途已显式 rt.Stop 时，清理阶段的二次 Stop 为 no-op +// （Runtime.Stop 只停 running 模块，幂等）。 +// +// 语义说明： +// - raw 与全局配置 `modules:` 子树同形状，是模块配置的唯一来源 +// （host 上已有的 ConfigOf——包括 WithConfig 设置的——会被重绑定为 +// raw 的解析结果）；nil 表示空配置； +// - enabled 语义与生产完全一致、不做任何强制：raw 未显式 enabled 且模块 +// 默认 disabled 时，Build / Start 全程 no-op（此时会 tb.Logf 提示， +// 便于排查"忘了写 enabled: true"）； +// - host 为 nil 时等价于 NewHost(tb) 的默认 Host； +// - Runtime 经 ModuleInfo().New 创建实例驱动生命周期——除非模块的 New +// 返回自身（fake 模块常如此），传入的 m 仅用于注册，生命周期效果应 +// 通过日志（WithObservedLogger）与返回 Runtime 的 Snapshot 断言， +// 而不是 m 的内部字段。 +// +// 需要断言"Build 应失败"用 BuildExpectingError；更复杂的场景（Start 失败 +// 回滚、多模块编排等）直接使用内核 API（plugin.NewRegistry + +// plugin.NewRuntimeWithRegistry），内核 API 保持完全公开。 +func RunLifecycle(tb testing.TB, m plugin.Module, host *plugin.Host, raw map[string]map[string]any) *plugin.Runtime { + tb.Helper() + rt, id := buildOnly(tb, m, host, raw) + if err := rt.Start(context.Background()); err != nil { + tb.Fatalf("plugintest: start module %q: %v", id, err) + } + return rt +} + +// BuildOnly 与 RunLifecycle 相同但停在 Build 之后（不 Start），用于只验证 +// Provision / Validate 行为的测试。返回的 Runtime 可由测试自行 Start； +// 无论是否 Start，t.Cleanup 都会执行兜底 Stop（未启动时为 no-op）并断言无错误。 +func BuildOnly(tb testing.TB, m plugin.Module, host *plugin.Host, raw map[string]map[string]any) *plugin.Runtime { + tb.Helper() + rt, _ := buildOnly(tb, m, host, raw) + return rt +} + +// BuildExpectingError 与 BuildOnly 同样装配 Runtime，但期望 Build 失败： +// Build 意外成功时 tb.Fatal；失败时返回该错误供测试断言内容 +// （内核保证 Build 错误信息包含失败模块 ID）。 +// +// 设计说明：单独提供本变体而不是给 RunLifecycle 加 expectError 参数， +// 是为了让成功 / 失败两种期望在调用处一目了然，且失败路径不注册多余的 +// cleanup（Build 失败后无 running 模块，无需 Stop）。raw 解析失败仍然 +// tb.Fatal——配置字面量笔误属于测试自身缺陷，不属于"期望中的 Build 失败"。 +// "Start 应失败"的场景本包不提供夹具（hello 与内核测试均无此样板复用需求， +// 避免为假设需求预铺 API），请直接使用内核 API。 +func BuildExpectingError(tb testing.TB, m plugin.Module, host *plugin.Host, raw map[string]map[string]any) error { + tb.Helper() + rt, id := newRuntime(tb, m, host, raw) + err := rt.Build() + if err == nil { + tb.Fatalf("plugintest: expected Build of module %q to fail, but it succeeded", id) + } + return err +} + +// buildOnly 是 RunLifecycle / BuildOnly 的共享实现：装配 Runtime、执行 Build、 +// 输出 disabled 提示并注册 t.Cleanup 兜底 Stop。 +func buildOnly(tb testing.TB, m plugin.Module, host *plugin.Host, raw map[string]map[string]any) (*plugin.Runtime, plugin.ModuleID) { + tb.Helper() + rt, id := newRuntime(tb, m, host, raw) + if err := rt.Build(); err != nil { + tb.Fatalf("plugintest: build module %q: %v", id, err) + } + for _, status := range rt.Snapshot() { + if status.ID == id && !status.Enabled { + tb.Logf("plugintest: module %q resolved to disabled under the given config; its lifecycle is a no-op (set \"enabled\": true in raw to run it)", id) + } + } + tb.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), stopTimeout) + defer cancel() + if err := rt.Stop(ctx); err != nil { + tb.Errorf("plugintest: stop module %q during cleanup: %v", id, err) + } + }) + return rt, id +} + +// newRuntime 完成各夹具入口共享的装配：隔离注册表注册 m、解析 raw 配置、 +// 将 host.ConfigOf 重绑定为解析结果、创建 Runtime。 +func newRuntime(tb testing.TB, m plugin.Module, host *plugin.Host, raw map[string]map[string]any) (*plugin.Runtime, plugin.ModuleID) { + tb.Helper() + if m == nil { + tb.Fatalf("plugintest: module under test is nil") + } + id := m.ModuleInfo().ID + cfg, err := plugin.ParseConfig(raw) + if err != nil { + tb.Fatalf("plugintest: parse modules config for module %q: %v", id, err) + } + if host == nil { + host = NewHost(tb) + } + // raw 是生命周期夹具的唯一配置来源：统一重绑定 ConfigOf，避免 host 上的 + // 旧绑定（NewHost 默认空配置 / WithConfig）与 Runtime 实际持有的 cfg 不一致。 + host.ConfigOf = cfg.Of + + registry := plugin.NewRegistry() + registry.RegisterModule(m) // 注册非法（ID 格式等）时按内核语义 panic，测试直接失败 + return plugin.NewRuntimeWithRegistry(host, cfg, registry), id +} diff --git a/backend/internal/plugin/registry.go b/backend/internal/plugin/registry.go new file mode 100644 index 0000000000..69824897b4 --- /dev/null +++ b/backend/internal/plugin/registry.go @@ -0,0 +1,113 @@ +package plugin + +import ( + "fmt" + "sort" + "sync" +) + +// Registry 是模块注册表：从 ModuleID 到 ModuleInfo 的并发安全目录。 +// +// 注册表只负责"有哪些模块"，不负责实例化与生命周期（那是 Runtime 的职责）。 +// 常规代码使用包级默认注册表（RegisterModule 等包级函数）； +// 测试可通过 NewRegistry 创建相互隔离的实例。 +type Registry struct { + mu sync.RWMutex + modules map[ModuleID]ModuleInfo +} + +// NewRegistry 创建一个空的、与包级默认注册表相互隔离的注册表实例。 +func NewRegistry() *Registry { + return &Registry{modules: make(map[ModuleID]ModuleInfo)} +} + +// RegisterModule 将模块注册进本注册表。 +// +// 与包级 RegisterModule 相同的 panic 语义：模块为 nil、ID 为空、ID 格式非法、 +// New 为 nil 或 ID 重复时 panic。注册错误属于编译期插装错误，应尽早暴露。 +func (r *Registry) RegisterModule(m Module) { + if m == nil { + panic("plugin: RegisterModule called with nil module") + } + info := m.ModuleInfo() + if err := info.ID.validate(); err != nil { + panic(fmt.Sprintf("plugin: RegisterModule: invalid module ID: %v", err)) + } + if info.New == nil { + panic(fmt.Sprintf("plugin: RegisterModule: module %q has a nil New function", info.ID)) + } + + r.mu.Lock() + defer r.mu.Unlock() + if _, dup := r.modules[info.ID]; dup { + panic(fmt.Sprintf("plugin: RegisterModule: module %q is already registered (duplicate registration; check the instrumentation list in internal/modules/standard)", info.ID)) + } + r.modules[info.ID] = info +} + +// GetModule 按 ID 查询模块注册信息，第二个返回值表示是否存在。 +func (r *Registry) GetModule(id ModuleID) (ModuleInfo, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + info, ok := r.modules[id] + return info, ok +} + +// GetModulesInNamespace 返回命名空间恰好为 ns 的全部模块，按 ID 字典序稳定排序。 +// +// 匹配为精确匹配（非前缀匹配）：ns 为 "gateway.platform" 时返回 +// "gateway.platform.anthropic" 等直接子模块，但不包含 "gateway.platform.x.y"。 +// ns 为空字符串时返回所有单段 ID 的顶层模块。 +func (r *Registry) GetModulesInNamespace(ns string) []ModuleInfo { + r.mu.RLock() + defer r.mu.RUnlock() + var result []ModuleInfo + for id, info := range r.modules { + if id.Namespace() == ns { + result = append(result, info) + } + } + sortModuleInfos(result) + return result +} + +// Modules 返回注册表中的全部模块，按 ID 字典序稳定排序。 +// Runtime 以该顺序作为模块生命周期的确定性顺序。 +func (r *Registry) Modules() []ModuleInfo { + r.mu.RLock() + defer r.mu.RUnlock() + result := make([]ModuleInfo, 0, len(r.modules)) + for _, info := range r.modules { + result = append(result, info) + } + sortModuleInfos(result) + return result +} + +// sortModuleInfos 按 ID 字典序原地排序，保证遍历顺序的确定性。 +func sortModuleInfos(infos []ModuleInfo) { + sort.Slice(infos, func(i, j int) bool { return infos[i].ID < infos[j].ID }) +} + +// defaultRegistry 是包级默认注册表，模块包在 init() 期向其自注册。 +var defaultRegistry = NewRegistry() + +// RegisterModule 将模块注册进包级默认注册表，必须在 init() 期调用。 +// +// 模块为 nil、ID 为空、ID 格式非法、New 为 nil 或 ID 重复时 panic： +// 注册错误属于编译期插装错误（插装清单 import 错误、模块 ID 冲突等）， +// 必须在进程启动最早期暴露，而不是延迟到运行时。 +func RegisterModule(m Module) { + defaultRegistry.RegisterModule(m) +} + +// GetModule 按 ID 在包级默认注册表中查询模块注册信息。 +func GetModule(id ModuleID) (ModuleInfo, bool) { + return defaultRegistry.GetModule(id) +} + +// GetModulesInNamespace 在包级默认注册表中查询命名空间恰好为 ns 的全部模块， +// 按 ID 字典序稳定排序。匹配语义见 (*Registry).GetModulesInNamespace。 +func GetModulesInNamespace(ns string) []ModuleInfo { + return defaultRegistry.GetModulesInNamespace(ns) +} diff --git a/backend/internal/plugin/registry_test.go b/backend/internal/plugin/registry_test.go new file mode 100644 index 0000000000..7f2c20b96a --- /dev/null +++ b/backend/internal/plugin/registry_test.go @@ -0,0 +1,176 @@ +package plugin + +import ( + "fmt" + "sync" + "testing" + + "github.com/stretchr/testify/require" +) + +// stubModule 是注册表测试用的最小模块实现。 +type stubModule struct { + info ModuleInfo +} + +func (m *stubModule) ModuleInfo() ModuleInfo { return m.info } + +// newStubModule 构造一个 ID 为 id 的最小可注册模块。 +func newStubModule(id ModuleID) *stubModule { + m := &stubModule{} + m.info = ModuleInfo{ + ID: id, + New: func() Module { return &stubModule{info: m.info} }, + } + return m +} + +func TestRegistryRegisterAndGet(t *testing.T) { + reg := NewRegistry() + reg.RegisterModule(newStubModule("job.hello")) + + info, ok := reg.GetModule("job.hello") + require.True(t, ok) + require.Equal(t, ModuleID("job.hello"), info.ID) + require.NotNil(t, info.New) + + _, ok = reg.GetModule("job.missing") + require.False(t, ok) +} + +func TestRegistryDuplicateIDPanics(t *testing.T) { + reg := NewRegistry() + reg.RegisterModule(newStubModule("job.hello")) + + require.PanicsWithValue(t, + `plugin: RegisterModule: module "job.hello" is already registered (duplicate registration; check the instrumentation list in internal/modules/standard)`, + func() { reg.RegisterModule(newStubModule("job.hello")) }, + ) +} + +func TestRegistryInvalidRegistrationPanics(t *testing.T) { + reg := NewRegistry() + + t.Run("nil module", func(t *testing.T) { + require.Panics(t, func() { reg.RegisterModule(nil) }) + }) + t.Run("empty ID", func(t *testing.T) { + require.Panics(t, func() { reg.RegisterModule(newStubModule("")) }) + }) + t.Run("invalid ID format", func(t *testing.T) { + // panic 信息必须包含非法 ID，便于排查插装清单。 + defer func() { + r := recover() + require.NotNil(t, r) + require.Contains(t, fmt.Sprint(r), "Job.Hello") + }() + reg.RegisterModule(newStubModule("Job.Hello")) + }) + t.Run("nil New func", func(t *testing.T) { + require.Panics(t, func() { + reg.RegisterModule(&stubModule{info: ModuleInfo{ID: "job.no-new"}}) + }) + }) +} + +func TestRegistryGetModulesInNamespace(t *testing.T) { + reg := NewRegistry() + // 故意乱序注册，验证返回结果按 ID 字典序稳定排序。 + for _, id := range []ModuleID{ + "job.zeta", + "job.alpha", + "job.middle", + "gateway.platform.anthropic", + "gateway.platform.openai", + "toplevel", + } { + reg.RegisterModule(newStubModule(id)) + } + + ids := func(infos []ModuleInfo) []ModuleID { + out := make([]ModuleID, 0, len(infos)) + for _, info := range infos { + out = append(out, info.ID) + } + return out + } + + require.Equal(t, []ModuleID{"job.alpha", "job.middle", "job.zeta"}, ids(reg.GetModulesInNamespace("job"))) + // 精确匹配：ns "gateway" 不包含 "gateway.platform.*"。 + require.Empty(t, reg.GetModulesInNamespace("gateway")) + require.Equal(t, []ModuleID{"gateway.platform.anthropic", "gateway.platform.openai"}, + ids(reg.GetModulesInNamespace("gateway.platform"))) + // 空命名空间返回单段顶层模块。 + require.Equal(t, []ModuleID{"toplevel"}, ids(reg.GetModulesInNamespace(""))) + + // 多次调用结果稳定。 + require.Equal(t, ids(reg.GetModulesInNamespace("job")), ids(reg.GetModulesInNamespace("job"))) +} + +func TestRegistryModulesSorted(t *testing.T) { + reg := NewRegistry() + for _, id := range []ModuleID{"c.mod", "a.mod", "b.mod"} { + reg.RegisterModule(newStubModule(id)) + } + all := reg.Modules() + require.Len(t, all, 3) + require.Equal(t, ModuleID("a.mod"), all[0].ID) + require.Equal(t, ModuleID("b.mod"), all[1].ID) + require.Equal(t, ModuleID("c.mod"), all[2].ID) +} + +func TestRegistryIsolation(t *testing.T) { + regA := NewRegistry() + regB := NewRegistry() + regA.RegisterModule(newStubModule("job.only-in-a")) + + _, ok := regA.GetModule("job.only-in-a") + require.True(t, ok) + _, ok = regB.GetModule("job.only-in-a") + require.False(t, ok, "isolated registry must not see other registry's modules") + // 隔离实例也不影响包级默认注册表。 + _, ok = GetModule("job.only-in-a") + require.False(t, ok) +} + +func TestPackageLevelFunctionsDelegateToDefaultRegistry(t *testing.T) { + // 使用唯一 ID，避免与同进程其他测试冲突（包级默认注册表是进程级状态）。 + const id = ModuleID("test_default_registry.probe") + RegisterModule(newStubModule(id)) + + info, ok := GetModule(id) + require.True(t, ok) + require.Equal(t, id, info.ID) + + infos := GetModulesInNamespace("test_default_registry") + require.Len(t, infos, 1) + require.Equal(t, id, infos[0].ID) +} + +func TestRegistryConcurrentRegistration(t *testing.T) { + reg := NewRegistry() + const n = 64 + + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func(i int) { + defer wg.Done() + reg.RegisterModule(newStubModule(ModuleID(fmt.Sprintf("concurrent.mod-%03d", i)))) + }(i) + } + // 并发读不应与写竞争（-race 验证）。 + for i := 0; i < 8; i++ { + go func() { + _ = reg.GetModulesInNamespace("concurrent") + _, _ = reg.GetModule("concurrent.mod-000") + }() + } + wg.Wait() + + all := reg.Modules() + require.Len(t, all, n) + for i := 1; i < len(all); i++ { + require.Less(t, all[i-1].ID, all[i].ID, "Modules() must be sorted") + } +} diff --git a/backend/internal/plugin/runtime.go b/backend/internal/plugin/runtime.go new file mode 100644 index 0000000000..450121f441 --- /dev/null +++ b/backend/internal/plugin/runtime.go @@ -0,0 +1,291 @@ +package plugin + +import ( + "context" + "errors" + "fmt" + "sync" + + "go.uber.org/zap" +) + +// State 是模块在 Runtime 生命周期中的状态。 +type State string + +// 模块生命周期状态。状态机： +// +// registered ──Build──▶ provisioned ──Start──▶ running ──Stop──▶ stopped +// │（disabled 模块停留在 registered） 任一阶段失败 ──▶ errored +const ( + // StateRegistered 模块已注册但未实例化（disabled 模块始终处于该状态）。 + StateRegistered State = "registered" + // StateProvisioned 模块已实例化并通过 Provision/Validate，等待启动。 + StateProvisioned State = "provisioned" + // StateRunning 模块已启动（未实现 Starter 的 enabled 模块在 Start 阶段也进入该状态）。 + StateRunning State = "running" + // StateStopped 模块已正常停止。 + StateStopped State = "stopped" + // StateErrored 模块在某个生命周期阶段失败，错误见 ModuleStatus.Err。 + StateErrored State = "errored" +) + +// ModuleStatus 是单个模块的状态快照（admin 可观测用）。 +type ModuleStatus struct { + // ID 模块 ID。 + ID ModuleID + // Enabled 模块最终的启停状态（显式配置优先，否则为注册声明的默认值）。 + Enabled bool + // State 模块当前生命周期状态。 + State State + // Err 模块最近一次生命周期失败的错误信息，无错误时为空字符串。 + Err string +} + +// moduleRecord 是 Runtime 内部维护的单模块运行时记录。 +type moduleRecord struct { + info ModuleInfo + enabled bool + instance Module + state State + err error +} + +// Runtime 是模块生命周期驱动器：按注册表稳定序（ID 字典序）实例化 enabled +// 模块并驱动 Provision → Validate → Start → Stop（逆序）生命周期。 +// +// 使用方式（单次使用，不支持 Stop 后重新 Start）： +// +// rt := plugin.NewRuntime(host, cfg) +// if err := rt.Build(); err != nil { ... } // 启动中止 +// if err := rt.Start(ctx); err != nil { ... } // 已自动逆序回滚 +// defer rt.Stop(shutdownCtx) // 接入既有 cleanup 序列 +// +// 所有方法并发安全。 +type Runtime struct { + mu sync.Mutex + host *Host + config Config + registry *Registry + logger *zap.Logger + + built bool + buildAttempted bool + started bool + records []*moduleRecord +} + +// NewRuntime 创建基于包级默认注册表的 Runtime。 +// +// host 为 nil 时使用空 Host；若 host.ConfigOf 未装配，则默认绑定为 cfg.Of， +// 使模块在 Provision 中总能读取到自己的私有配置。 +func NewRuntime(host *Host, cfg Config) *Runtime { + return NewRuntimeWithRegistry(host, cfg, defaultRegistry) +} + +// NewRuntimeWithRegistry 创建基于指定注册表的 Runtime（测试隔离用）。 +// 其余语义与 NewRuntime 一致。 +func NewRuntimeWithRegistry(host *Host, cfg Config, registry *Registry) *Runtime { + if host == nil { + host = &Host{} + } + if cfg == nil { + cfg = Config{} + } + if registry == nil { + registry = defaultRegistry + } + if host.ConfigOf == nil { + host.ConfigOf = cfg.Of + } + logger := host.Logger + if logger == nil { + logger = zap.NewNop() + } + return &Runtime{ + host: host, + config: cfg, + registry: registry, + logger: logger, + } +} + +// Build 按注册表稳定序实例化全部 enabled 模块并依次执行 Provision、Validate。 +// +// 任一模块失败立即返回包含模块 ID 的错误（启动中止），不会继续处理后续模块； +// 已成功 Provision 的模块保持 provisioned 状态，由进程启动失败路径整体退出。 +// Build 只能调用一次：无论成败，第二次调用一律拒绝——失败的 Build 不可重试 +// （重试会对已 Provision 的旧实例不 Stop 即覆盖，造成资源泄漏），应整体重启进程。 +func (r *Runtime) Build() error { + r.mu.Lock() + defer r.mu.Unlock() + if r.buildAttempted { + if r.built { + return errors.New("plugin: runtime already built") + } + return errors.New("plugin: previous build failed; runtime cannot be rebuilt, restart the process") + } + r.buildAttempted = true + + // 先为全部已注册模块建立记录（含 disabled），保证失败后 Snapshot 仍是完整视图。 + infos := r.registry.Modules() // 已按 ID 字典序稳定排序 + r.records = make([]*moduleRecord, 0, len(infos)) + for _, info := range infos { + r.records = append(r.records, &moduleRecord{ + info: info, + enabled: r.config.enabledFor(info), + state: StateRegistered, + }) + } + + for _, rec := range r.records { + if !rec.enabled { + continue + } + info := rec.info + rec.instance = info.New() + if rec.instance == nil { + rec.err = errors.New("module New func returned nil") + rec.state = StateErrored + return fmt.Errorf("plugin: build module %q: New returned nil instance", info.ID) + } + if p, ok := rec.instance.(Provisioner); ok { + if err := p.Provision(r.host); err != nil { + rec.err = err + rec.state = StateErrored + return fmt.Errorf("plugin: provision module %q: %w", info.ID, err) + } + } + if v, ok := rec.instance.(Validator); ok { + if err := v.Validate(); err != nil { + rec.err = err + rec.state = StateErrored + return fmt.Errorf("plugin: validate module %q: %w", info.ID, err) + } + } + rec.state = StateProvisioned + r.logger.Debug("plugin module provisioned", zap.String("module", string(info.ID))) + } + r.built = true + return nil +} + +// Start 按注册表稳定序启动全部 enabled 模块（对实现 Starter 的模块调用 Start， +// 未实现 Starter 的模块直接视为 running）。 +// +// 任一模块启动失败时：对已进入 running 的模块按逆序执行 Stop 回滚 +// （回滚沿用传入的 ctx）。返回的错误包含失败模块 ID；若回滚中某些模块的 +// Stop 也失败，这些错误会一并 errors.Join 进返回值（同时记日志）， +// 让调用方能看到哪些模块的资源未释放干净。 +func (r *Runtime) Start(ctx context.Context) error { + r.mu.Lock() + defer r.mu.Unlock() + if !r.built { + return errors.New("plugin: runtime must be built before start") + } + if r.started { + return errors.New("plugin: runtime already started") + } + + startedRecs := make([]*moduleRecord, 0, len(r.records)) + for _, rec := range r.records { + if !rec.enabled || rec.state != StateProvisioned { + continue + } + if s, ok := rec.instance.(Starter); ok { + if err := s.Start(ctx); err != nil { + rec.err = err + rec.state = StateErrored + // 逆序回滚已启动模块；回滚失败并入返回错误（见方法注释）。 + rollbackErrs := r.stopRecordsLocked(ctx, startedRecs) + startErr := fmt.Errorf("plugin: start module %q: %w", rec.info.ID, err) + if len(rollbackErrs) > 0 { + return errors.Join(append([]error{startErr}, rollbackErrs...)...) + } + return startErr + } + } + rec.state = StateRunning + startedRecs = append(startedRecs, rec) + r.logger.Info("plugin module started", zap.String("module", string(rec.info.ID))) + } + r.started = true + return nil +} + +// Stop 按注册表稳定序的逆序停止全部 running 模块。 +// +// 单个模块 Stop 失败只记日志并继续关闭其余模块（与既有 cleanup 序列的容错 +// 纪律一致），最终返回 errors.Join 聚合的失败集合（全部成功时为 nil）。 +// ctx 原样传递给各模块 Stop，模块必须自行尊重其 deadline。 +func (r *Runtime) Stop(ctx context.Context) error { + r.mu.Lock() + defer r.mu.Unlock() + + running := make([]*moduleRecord, 0, len(r.records)) + for _, rec := range r.records { + if rec.state == StateRunning { + running = append(running, rec) + } + } + errs := r.stopRecordsLocked(ctx, running) + r.started = false + return errors.Join(errs...) +} + +// stopRecordsLocked 逆序停止 recs 中的模块（recs 为正向顺序），返回失败集合。 +// 调用方必须已持有 r.mu。 +func (r *Runtime) stopRecordsLocked(ctx context.Context, recs []*moduleRecord) []error { + var errs []error + for i := len(recs) - 1; i >= 0; i-- { + rec := recs[i] + if s, ok := rec.instance.(Stopper); ok { + if err := s.Stop(ctx); err != nil { + rec.err = err + rec.state = StateErrored + r.logger.Error("plugin module stop failed", + zap.String("module", string(rec.info.ID)), zap.Error(err)) + errs = append(errs, fmt.Errorf("plugin: stop module %q: %w", rec.info.ID, err)) + continue + } + } + rec.state = StateStopped + r.logger.Info("plugin module stopped", zap.String("module", string(rec.info.ID))) + } + return errs +} + +// Snapshot 返回全部已注册模块的状态快照，按 ID 字典序稳定排序。 +// +// Build 之前调用时，基于注册表与配置即时计算（状态均为 registered）； +// Build 之后返回 Runtime 维护的实际生命周期状态。 +func (r *Runtime) Snapshot() []ModuleStatus { + r.mu.Lock() + defer r.mu.Unlock() + + if r.records == nil { + infos := r.registry.Modules() + statuses := make([]ModuleStatus, 0, len(infos)) + for _, info := range infos { + statuses = append(statuses, ModuleStatus{ + ID: info.ID, + Enabled: r.config.enabledFor(info), + State: StateRegistered, + }) + } + return statuses + } + + statuses := make([]ModuleStatus, 0, len(r.records)) + for _, rec := range r.records { + status := ModuleStatus{ + ID: rec.info.ID, + Enabled: rec.enabled, + State: rec.state, + } + if rec.err != nil { + status.Err = rec.err.Error() + } + statuses = append(statuses, status) + } + return statuses +} diff --git a/backend/internal/plugin/runtime_test.go b/backend/internal/plugin/runtime_test.go new file mode 100644 index 0000000000..a2404c09c4 --- /dev/null +++ b/backend/internal/plugin/runtime_test.go @@ -0,0 +1,389 @@ +package plugin + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// lifecycleLog 记录全部 fake 模块的生命周期调用序列（"模块ID:阶段"）。 +type lifecycleLog struct { + mu sync.Mutex + events []string +} + +func (l *lifecycleLog) add(event string) { + l.mu.Lock() + defer l.mu.Unlock() + l.events = append(l.events, event) +} + +func (l *lifecycleLog) snapshot() []string { + l.mu.Lock() + defer l.mu.Unlock() + return append([]string(nil), l.events...) +} + +// fakeModule 是实现全部可选生命周期接口的测试模块，按配置注入各阶段错误。 +type fakeModule struct { + id ModuleID + enabledByDefault bool + log *lifecycleLog + + provisionErr error + validateErr error + startErr error + stopErr error + stopFn func(ctx context.Context) error // 非 nil 时覆盖默认 Stop 行为 + + host *Host // Provision 时记录，供断言 +} + +func (m *fakeModule) ModuleInfo() ModuleInfo { + return ModuleInfo{ + ID: m.id, + EnabledByDefault: m.enabledByDefault, + New: func() Module { return m }, + } +} + +func (m *fakeModule) Provision(host *Host) error { + m.host = host + m.log.add(string(m.id) + ":provision") + return m.provisionErr +} + +func (m *fakeModule) Validate() error { + m.log.add(string(m.id) + ":validate") + return m.validateErr +} + +func (m *fakeModule) Start(ctx context.Context) error { + m.log.add(string(m.id) + ":start") + return m.startErr +} + +func (m *fakeModule) Stop(ctx context.Context) error { + m.log.add(string(m.id) + ":stop") + if m.stopFn != nil { + return m.stopFn(ctx) + } + return m.stopErr +} + +// plainModule 只实现 Module（无任何生命周期接口）。 +type plainModule struct { + id ModuleID +} + +func (m *plainModule) ModuleInfo() ModuleInfo { + return ModuleInfo{ + ID: m.id, + EnabledByDefault: true, + New: func() Module { return m }, + } +} + +// newTestRuntime 构造隔离注册表 + Runtime。 +func newTestRuntime(t *testing.T, cfg Config, mods ...Module) *Runtime { + t.Helper() + reg := NewRegistry() + for _, m := range mods { + reg.RegisterModule(m) + } + return NewRuntimeWithRegistry(&Host{}, cfg, reg) +} + +func statusByID(t *testing.T, statuses []ModuleStatus, id ModuleID) ModuleStatus { + t.Helper() + for _, s := range statuses { + if s.ID == id { + return s + } + } + t.Fatalf("module %q not found in snapshot", id) + return ModuleStatus{} +} + +func TestRuntimeBuildLifecycleOrderAndEnabledTriState(t *testing.T) { + log := &lifecycleLog{} + on, off := true, false + cfg := Config{ + "job.c-default-off-enabled": {Enabled: &on}, + "job.d-default-on-disabled": {Enabled: &off}, + } + rt := newTestRuntime(t, cfg, + // 故意乱序注册，验证 Build 按 ID 字典序执行。 + &fakeModule{id: "job.b-default-on", enabledByDefault: true, log: log}, + &fakeModule{id: "job.a-default-on", enabledByDefault: true, log: log}, + &fakeModule{id: "job.c-default-off-enabled", enabledByDefault: false, log: log}, + &fakeModule{id: "job.d-default-on-disabled", enabledByDefault: true, log: log}, + &fakeModule{id: "job.e-default-off", enabledByDefault: false, log: log}, + ) + + require.NoError(t, rt.Build()) + + // enabled 三态：默认 on（a、b）、显式 enable 默认 off 的 c、显式 disable 默认 on 的 d、默认 off 的 e。 + require.Equal(t, []string{ + "job.a-default-on:provision", "job.a-default-on:validate", + "job.b-default-on:provision", "job.b-default-on:validate", + "job.c-default-off-enabled:provision", "job.c-default-off-enabled:validate", + }, log.snapshot()) + + statuses := rt.Snapshot() + require.Equal(t, StateProvisioned, statusByID(t, statuses, "job.a-default-on").State) + require.Equal(t, StateProvisioned, statusByID(t, statuses, "job.c-default-off-enabled").State) + require.False(t, statusByID(t, statuses, "job.d-default-on-disabled").Enabled) + require.Equal(t, StateRegistered, statusByID(t, statuses, "job.d-default-on-disabled").State) + require.False(t, statusByID(t, statuses, "job.e-default-off").Enabled) + require.Equal(t, StateRegistered, statusByID(t, statuses, "job.e-default-off").State) +} + +func TestRuntimeBuildProvisionFailureAborts(t *testing.T) { + log := &lifecycleLog{} + provisionErr := errors.New("provision boom") + rt := newTestRuntime(t, Config{}, + &fakeModule{id: "job.a", enabledByDefault: true, log: log}, + &fakeModule{id: "job.b", enabledByDefault: true, log: log, provisionErr: provisionErr}, + &fakeModule{id: "job.c", enabledByDefault: true, log: log}, + ) + + err := rt.Build() + require.Error(t, err) + require.Contains(t, err.Error(), "job.b", "Build 错误必须包含失败模块 ID") + require.ErrorIs(t, err, provisionErr) + + // 失败即中止：job.c 不会被 Provision；job.b 的 Validate 不会执行。 + require.Equal(t, []string{ + "job.a:provision", "job.a:validate", + "job.b:provision", + }, log.snapshot()) + + statuses := rt.Snapshot() + require.Equal(t, StateProvisioned, statusByID(t, statuses, "job.a").State) + bStatus := statusByID(t, statuses, "job.b") + require.Equal(t, StateErrored, bStatus.State) + require.Contains(t, bStatus.Err, "provision boom") + require.Equal(t, StateRegistered, statusByID(t, statuses, "job.c").State) + + // Build 失败后 Start 必须被拒绝。 + require.Error(t, rt.Start(context.Background())) +} + +func TestRuntimeBuildValidateFailureAborts(t *testing.T) { + log := &lifecycleLog{} + rt := newTestRuntime(t, Config{}, + &fakeModule{id: "job.a", enabledByDefault: true, log: log, validateErr: errors.New("validate boom")}, + &fakeModule{id: "job.b", enabledByDefault: true, log: log}, + ) + + err := rt.Build() + require.Error(t, err) + require.Contains(t, err.Error(), "job.a") + require.Equal(t, []string{"job.a:provision", "job.a:validate"}, log.snapshot()) + require.Equal(t, StateErrored, statusByID(t, rt.Snapshot(), "job.a").State) +} + +func TestRuntimeStartOrderAndStopReverseOrder(t *testing.T) { + log := &lifecycleLog{} + rt := newTestRuntime(t, Config{}, + &fakeModule{id: "job.b", enabledByDefault: true, log: log}, + &fakeModule{id: "job.a", enabledByDefault: true, log: log}, + &fakeModule{id: "job.c", enabledByDefault: true, log: log}, + ) + require.NoError(t, rt.Build()) + require.NoError(t, rt.Start(context.Background())) + + for _, s := range rt.Snapshot() { + require.Equal(t, StateRunning, s.State) + } + + require.NoError(t, rt.Stop(context.Background())) + + require.Equal(t, []string{ + "job.a:provision", "job.a:validate", + "job.b:provision", "job.b:validate", + "job.c:provision", "job.c:validate", + "job.a:start", "job.b:start", "job.c:start", + "job.c:stop", "job.b:stop", "job.a:stop", // 逆序关闭 + }, log.snapshot()) + + for _, s := range rt.Snapshot() { + require.Equal(t, StateStopped, s.State) + } +} + +func TestRuntimeStartHalfwayFailureRollsBackInReverse(t *testing.T) { + log := &lifecycleLog{} + startErr := errors.New("start boom") + rt := newTestRuntime(t, Config{}, + &fakeModule{id: "job.a", enabledByDefault: true, log: log}, + &fakeModule{id: "job.b", enabledByDefault: true, log: log}, + &fakeModule{id: "job.c", enabledByDefault: true, log: log, startErr: startErr}, + &fakeModule{id: "job.d", enabledByDefault: true, log: log}, + ) + require.NoError(t, rt.Build()) + + err := rt.Start(context.Background()) + require.Error(t, err) + require.Contains(t, err.Error(), "job.c", "Start 错误必须包含失败模块 ID") + require.ErrorIs(t, err, startErr) + + // a、b 已启动；c 启动失败；回滚必须按 b → a 逆序 Stop；d 不得被 Start。 + require.Equal(t, []string{ + "job.a:provision", "job.a:validate", + "job.b:provision", "job.b:validate", + "job.c:provision", "job.c:validate", + "job.d:provision", "job.d:validate", + "job.a:start", "job.b:start", "job.c:start", + "job.b:stop", "job.a:stop", + }, log.snapshot()) + + statuses := rt.Snapshot() + require.Equal(t, StateStopped, statusByID(t, statuses, "job.a").State) + require.Equal(t, StateStopped, statusByID(t, statuses, "job.b").State) + cStatus := statusByID(t, statuses, "job.c") + require.Equal(t, StateErrored, cStatus.State) + require.Contains(t, cStatus.Err, "start boom") + require.Equal(t, StateProvisioned, statusByID(t, statuses, "job.d").State) +} + +func TestRuntimeStopToleratesSingleFailure(t *testing.T) { + log := &lifecycleLog{} + stopErr := errors.New("stop boom") + rt := newTestRuntime(t, Config{}, + &fakeModule{id: "job.a", enabledByDefault: true, log: log}, + &fakeModule{id: "job.b", enabledByDefault: true, log: log, stopErr: stopErr}, + &fakeModule{id: "job.c", enabledByDefault: true, log: log}, + ) + require.NoError(t, rt.Build()) + require.NoError(t, rt.Start(context.Background())) + + err := rt.Stop(context.Background()) + require.Error(t, err) + require.ErrorIs(t, err, stopErr) + require.Contains(t, err.Error(), "job.b") + + // 单个失败不中断其余模块关闭：c → b（失败）→ a 全部被调用。 + stops := log.snapshot()[9:] + require.Equal(t, []string{"job.c:stop", "job.b:stop", "job.a:stop"}, stops) + + statuses := rt.Snapshot() + require.Equal(t, StateStopped, statusByID(t, statuses, "job.a").State) + require.Equal(t, StateErrored, statusByID(t, statuses, "job.b").State) + require.Contains(t, statusByID(t, statuses, "job.b").Err, "stop boom") + require.Equal(t, StateStopped, statusByID(t, statuses, "job.c").State) +} + +func TestRuntimeStopRespectsContextDeadline(t *testing.T) { + log := &lifecycleLog{} + blocking := &fakeModule{ + id: "job.blocking", enabledByDefault: true, log: log, + stopFn: func(ctx context.Context) error { + <-ctx.Done() // 模拟尊重 ctx deadline 的慢关闭 + return ctx.Err() + }, + } + rt := newTestRuntime(t, Config{}, blocking, + &fakeModule{id: "job.a", enabledByDefault: true, log: log}, + ) + require.NoError(t, rt.Build()) + require.NoError(t, rt.Start(context.Background())) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + begin := time.Now() + err := rt.Stop(ctx) + require.Less(t, time.Since(begin), 2*time.Second, "Stop 必须随 ctx deadline 返回，不得无限阻塞") + require.Error(t, err) + require.ErrorIs(t, err, context.DeadlineExceeded) + + // 即使 blocking 超时失败，job.a 仍被关闭（容错继续）。 + require.Equal(t, StateStopped, statusByID(t, rt.Snapshot(), "job.a").State) +} + +func TestRuntimeModuleWithoutLifecycleInterfaces(t *testing.T) { + rt := newTestRuntime(t, Config{}, &plainModule{id: "job.plain"}) + require.NoError(t, rt.Build()) + require.NoError(t, rt.Start(context.Background())) + require.Equal(t, StateRunning, statusByID(t, rt.Snapshot(), "job.plain").State, + "未实现 Starter 的 enabled 模块在 Start 后视为 running") + require.NoError(t, rt.Stop(context.Background())) + require.Equal(t, StateStopped, statusByID(t, rt.Snapshot(), "job.plain").State) +} + +func TestRuntimeGuards(t *testing.T) { + rt := newTestRuntime(t, Config{}) + require.Error(t, rt.Start(context.Background()), "未 Build 不得 Start") + require.NoError(t, rt.Build()) + require.Error(t, rt.Build(), "Build 只能调用一次") + require.NoError(t, rt.Start(context.Background())) + require.Error(t, rt.Start(context.Background()), "Start 只能调用一次") +} + +func TestRuntimeSnapshotBeforeBuild(t *testing.T) { + log := &lifecycleLog{} + off := false + rt := newTestRuntime(t, Config{"job.b": {Enabled: &off}}, + &fakeModule{id: "job.b", enabledByDefault: true, log: log}, + &fakeModule{id: "job.a", enabledByDefault: true, log: log}, + ) + + statuses := rt.Snapshot() + require.Len(t, statuses, 2) + require.Equal(t, ModuleID("job.a"), statuses[0].ID, "Snapshot 按 ID 字典序") + require.Equal(t, ModuleID("job.b"), statuses[1].ID) + for _, s := range statuses { + require.Equal(t, StateRegistered, s.State) + require.Empty(t, s.Err) + } + require.True(t, statuses[0].Enabled) + require.False(t, statuses[1].Enabled, "Build 前 Snapshot 也要正确解析 enabled 三态") +} + +func TestRuntimeHostConfigOfDefaultBinding(t *testing.T) { + log := &lifecycleLog{} + cfg, err := ParseConfig(map[string]map[string]any{ + "job.hello": {"enabled": true, "greeting": "from-config"}, + }) + require.NoError(t, err) + + mod := &fakeModule{id: "job.hello", log: log} + rt := newTestRuntime(t, cfg, mod) + require.NoError(t, rt.Build()) + + // NewRuntime 未显式装配 ConfigOf 时默认绑定到自身 Config.Of。 + require.NotNil(t, mod.host) + require.NotNil(t, mod.host.ConfigOf) + var out helloModuleConfig + require.NoError(t, mod.host.ConfigOf("job.hello", &out)) + require.Equal(t, "from-config", out.Greeting) +} + +// 审计 B2 修复的回归测试：Start 半途失败触发回滚时，若已启动模块的回滚 Stop +// 也失败，该失败必须并入 Start 的返回错误（而不是只记日志被吞掉）。 +func TestRuntimeStartRollbackStopFailureJoinedIntoError(t *testing.T) { + log := &lifecycleLog{} + stopBoom := errors.New("stop-a-boom") + rt := newTestRuntime(t, Config{}, + &fakeModule{id: "job.a", enabledByDefault: true, log: log, stopErr: stopBoom}, + &fakeModule{id: "job.b", enabledByDefault: true, log: log, startErr: errors.New("start-b-boom")}, + ) + require.NoError(t, rt.Build()) + + err := rt.Start(context.Background()) + require.Error(t, err) + // 启动失败与回滚失败都必须在返回错误中可见（errors.Join 可穿透）。 + require.ErrorIs(t, err, stopBoom) + require.Contains(t, err.Error(), `start module "job.b"`) + require.Contains(t, err.Error(), `stop module "job.a"`) + + statuses := rt.Snapshot() + require.Equal(t, StateErrored, statusByID(t, statuses, "job.a").State, "回滚失败的模块应为 errored") + require.Equal(t, StateErrored, statusByID(t, statuses, "job.b").State) +} diff --git a/backend/internal/plugin/wire.go b/backend/internal/plugin/wire.go new file mode 100644 index 0000000000..ff7faacdbc --- /dev/null +++ b/backend/internal/plugin/wire.go @@ -0,0 +1,52 @@ +package plugin + +import ( + "github.com/google/wire" + "github.com/redis/go-redis/v9" + + ent "github.com/Wei-Shaw/sub2api/ent" + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/pkg/logger" +) + +// ProviderSet 提供插件内核的依赖：modules: 配置子树、宿主能力面 Host +// 与模块生命周期驱动器 Runtime。 +// +// 边界约定（见 Phase-1 计划）：Wire 只负责装配 Host 与 Runtime， +// 模块目录由注册表（插装清单 internal/modules/standard）负责， +// 模块内部依赖一律经 Host 获取，不进入 Wire graph。 +var ProviderSet = wire.NewSet( + ProvideModuleConfig, + ProvideHost, + ProvideModuleRuntime, +) + +// ProvideModuleConfig 解析全局配置中 `modules:` 子树为插件模块配置。 +// +// 解析失败（模块 ID 格式非法、enabled 类型错误）返回错误， +// 使应用启动 fail-fast，配置笔误在启动时即暴露。 +func ProvideModuleConfig(cfg *config.Config) (Config, error) { + return ParseConfig(cfg.Modules) +} + +// ProvideHost 装配模块宿主能力面：聚合全局结构化日志器、模块私有配置 +// 解码能力（ConfigOf 绑定为 modules: 配置子树的 Config.Of）、 +// Ent 客户端与 Redis 客户端。 +// +// 日志器复用项目全局 logger（main 在依赖装配前已完成 logger.Init）。 +func ProvideHost(moduleCfg Config, entClient *ent.Client, rdb *redis.Client) *Host { + return &Host{ + Logger: logger.L(), + ConfigOf: moduleCfg.Of, + DB: entClient, + Redis: rdb, + } +} + +// ProvideModuleRuntime 创建基于包级默认注册表的模块生命周期驱动器。 +// +// 生命周期驱动由 main 负责：HTTP server 启动前 Build + Start， +// 关闭时 Stop 接入 cleanup 序列（先于 Redis/Ent 等基础设施关闭）。 +func ProvideModuleRuntime(host *Host, moduleCfg Config) *Runtime { + return NewRuntime(host, moduleCfg) +} diff --git a/backend/internal/server/routes/admin.go b/backend/internal/server/routes/admin.go index e6afa1436c..1eb407ab4f 100644 --- a/backend/internal/server/routes/admin.go +++ b/backend/internal/server/routes/admin.go @@ -103,9 +103,17 @@ func RegisterAdminRoutes( // 邀请返利（专属用户管理） registerAffiliateRoutes(admin, h) + + // 插件模块可观测 + registerModuleRoutes(admin, h) } } +// registerModuleRoutes 注册插件模块可观测的管理端路由（只读） +func registerModuleRoutes(admin *gin.RouterGroup, h *handler.Handlers) { + admin.GET("/modules", h.Admin.Module.List) +} + func registerAdminComplianceRoutes(admin *gin.RouterGroup, h *handler.Handlers) { compliance := admin.Group("/compliance") {