From 001e39da50d798c89641fd31eb4bb915841ae703 Mon Sep 17 00:00:00 2001 From: Yvan <8574526@qq,com> Date: Tue, 25 Aug 2026 16:38:51 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E7=BB=93=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CLAUDE.md | 2 +- cmd/main.go | 45 +++---- cmd/wire.go | 1 - cmd/wire_gen.go | 97 +++++++------- internal/data/config_store.go | 1 - internal/data/data.go | 27 ---- internal/data/initialization_backend.go | 2 - internal/global/resources.go | 163 ------------------------ internal/global/resources_test.go | 147 --------------------- 9 files changed, 67 insertions(+), 418 deletions(-) delete mode 100644 internal/global/resources.go delete mode 100644 internal/global/resources_test.go diff --git a/CLAUDE.md b/CLAUDE.md index ce46cda..469f8c7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -10,7 +10,7 @@ api/// Proto sources and generated stubs. Public contract. cmd// Entrypoint, Wire injector, main.go. configs/ Runtime config (config.yaml). No secrets. internal/config/ Viper config models, loading, snapshots, and reloads. -internal/global/ Process-wide shared resource registry. +internal/global/ Process-wide shared runtime resources. internal/server/ HTTP/gRPC server wiring. internal/service/ Transport adapters; one file per resource. internal/biz/ Domain models, usecases, repo interfaces, errors. diff --git a/cmd/main.go b/cmd/main.go index c6641aa..c096979 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -10,11 +10,6 @@ import ( "kra/internal/app" "kra/internal/config" - "kra/internal/data" - "kra/internal/global" - mqintegration "kra/internal/integration/mq" - "kra/internal/integration/storage" - websocketintegration "kra/internal/integration/websocket" "kra/internal/server/router" "kra/internal/service" "kra/internal/service/dto" @@ -43,6 +38,14 @@ var ( id, _ = os.Hostname() ) +// Service identity reported to logs, health output, and the API documentation. +// A plain `go run` leaves the ldflags empty, so these are the fallbacks. +const ( + appName = "Kra Admin" + appDescription = "Kratos 管理后台服务" + defaultVersion = "v0.0.0-dev" +) + func init() { flag.StringVar(&flagconf, "conf", "./configs", "config path, eg: -conf config.yaml") } @@ -56,27 +59,7 @@ func runtimeContributions(systemRoutes *router.Routes, systemTasks *worker.TaskM } } -func installGlobalResources(logger *slog.Logger, dataStore *data.Data, fileStorage *storage.Reloadable, broker *mqintegration.Reloadable, websocket *websocketintegration.Server, scheduler *worker.TaskScheduler) *global.ResourceRegistry { - install := func() { - global.Install(global.Resources{ - Logger: logger, - DB: dataStore.DB(), - NamedDBs: dataStore.NamedDatabases(), - Redis: dataStore.RedisClient(), - NamedRedis: dataStore.NamedRedisClients(), - Mongo: dataStore.MongoClient(), - Storage: fileStorage, - MQ: broker, - WebSocket: websocket, - Scheduler: scheduler, - }) - } - dataStore.SetResourceHook(install) - install() - return global.DefaultRegistry() -} - -func newApp(logger *slog.Logger, hs *kratoshttp.Server, scheduler *worker.TaskScheduler, audit *service.AuditRecorder, loggerControl *logging.ReloadableLogger, _ mq.Client, _ *global.ResourceRegistry) *kratos.App { +func newApp(logger *slog.Logger, hs *kratoshttp.Server, scheduler *worker.TaskScheduler, audit *service.AuditRecorder, loggerControl *logging.ReloadableLogger, _ mq.Client) *kratos.App { if audit != nil && loggerControl != nil { loggerControl.SetErrorSink(logging.ErrorSinkFunc(func(ctx context.Context, entry logging.ErrorEntry) error { return audit.CreateErrorRequest(ctx, &dto.ErrorRecordRequest{Form: entry.Form, Info: entry.Info, Level: entry.Level, RequestID: entry.RequestID, TraceID: entry.TraceID}) @@ -86,7 +69,7 @@ func newApp(logger *slog.Logger, hs *kratoshttp.Server, scheduler *worker.TaskSc kratos.ID(id), kratos.Name(Name), kratos.Version(Version), - kratos.Metadata(map[string]string{}), + kratos.Metadata(map[string]string{"description": appDescription}), kratos.Logger(logger), kratos.Server( hs, @@ -135,6 +118,14 @@ func swaggerAddress(server *config.Server, admin *config.Admin) string { func main() { flag.Parse() + // A plain `go run` leaves the ldflags empty, so fall back to the declared + // identity rather than reporting a blank service. + if Name == "" { + Name = appName + } + if Version == "" { + Version = defaultVersion + } runtime, err := config.LoadStore(flagconf) if err != nil { panic(err) diff --git a/cmd/wire.go b/cmd/wire.go index 12f468f..5172fca 100644 --- a/cmd/wire.go +++ b/cmd/wire.go @@ -50,7 +50,6 @@ func wireApp(*config.Server, *config.Store, *slog.Logger, *logging.ReloadableLog wire.Bind(new(taskbiz.TaskMethodRegistry), new(*platformtask.Registry)), biz.ProviderSet, service.ProviderSet, - installGlobalResources, newApp, )) } diff --git a/cmd/wire_gen.go b/cmd/wire_gen.go index c2dc84a..348b320 100644 --- a/cmd/wire_gen.go +++ b/cmd/wire_gen.go @@ -59,7 +59,7 @@ func wireApp(configServer *config.Server, store *config.Store, logger *slog.Logg authorityAccessRepo := system.NewAuthorityAccessRepo(dataData) apiRepo := system.NewAPIRepo(dataData) accessControlUsecase := system2.NewAccessControlUsecase(authorityAccessRepo, apiRepo) - accessControlService := system3.NewAccessControlService(accessControlUsecase) + v := system3.NewAccessControlService(accessControlUsecase) userRepo := system.NewUserRepo(dataData) userUsecase := system2.NewUserUsecase(userRepo) securityRepo := system.NewSecurityRepo(dataData) @@ -72,44 +72,44 @@ func wireApp(configServer *config.Server, store *config.Store, logger *slog.Logg auditRecordRepo := system.NewAuditRecorderRepo(dataData) authenticationUsecase := system2.NewAuthenticationUsecase(userUsecase, securityUsecase, tokenIssuer, auditRecordRepo) authService := system3.NewAuthService(authenticationUsecase) - securityService := system3.NewSecurityService(securityUsecase) + v2 := system3.NewSecurityService(securityUsecase) auditRecorderUsecase := system2.NewAuditRecorderUsecase(auditRecordRepo) - auditRecorder := system3.NewAuditRecorder(auditRecorderUsecase) + v3 := system3.NewAuditRecorder(auditRecorderUsecase) authorityUsecase := system2.NewAuthorityUsecase(authorityAccessRepo) - authorityService := system3.NewAuthorityService(authorityUsecase) - authority := handler.NewAuthority(authorityService) + v4 := system3.NewAuthorityService(authorityUsecase) + authority := handler.NewAuthority(v4) menuRepo := system.NewMenuRepo(dataData) menuUsecase := system2.NewMenuUsecase(menuRepo) - menuService := system3.NewMenuService(menuUsecase) - menu := handler.NewMenu(menuService) + v5 := system3.NewMenuService(menuUsecase) + menu := handler.NewMenu(v5) apiUsecase := system2.NewAPIUsecase(apiRepo) - apiService := system3.NewAPIService(apiUsecase, runtimeSettings) - api := handler.NewAPI(apiService) + v6 := system3.NewAPIService(apiUsecase, runtimeSettings) + api := handler.NewAPI(v6) permissionRepo := system.NewPermissionRepo(dataData) permissionUsecase := system2.NewPermissionUsecase(permissionRepo) - permissionService := system3.NewPermissionService(permissionUsecase) - permission := handler.NewPermission(permissionService) + v7 := system3.NewPermissionService(permissionUsecase) + permission := handler.NewPermission(v7) departmentRepo := system.NewDepartmentRepo(dataData) departmentUsecase := system2.NewDepartmentUsecase(departmentRepo) - departmentService := system3.NewDepartmentService(departmentUsecase) + v8 := system3.NewDepartmentService(departmentUsecase) positionRepo := system.NewPositionRepo(dataData) positionUsecase := system2.NewPositionUsecase(positionRepo) - positionService := system3.NewPositionService(positionUsecase) - organization := handler.NewOrganization(departmentService, positionService) + v9 := system3.NewPositionService(positionUsecase) + organization := handler.NewOrganization(v8, v9) announcementRepo := system.NewAnnouncementRepo(dataData) announcementUsecase := system2.NewAnnouncementUsecase(announcementRepo) - announcementService := system3.NewAnnouncementService(announcementUsecase) - announcement := handler.NewAnnouncement(announcementService) + v10 := system3.NewAnnouncementService(announcementUsecase) + announcement := handler.NewAnnouncement(v10) emailRepo := email.NewEmailRepo(store) emailUsecase := system2.NewEmailUsecase(emailRepo) - emailService := system3.NewEmailService(emailUsecase) - handlerEmail := handler.NewEmail(emailService) + v11 := system3.NewEmailService(emailUsecase) + handlerEmail := handler.NewEmail(v11) paymentConfigReader := integration.NewPaymentConfigReader(dataData) paymentRepo := payment.NewPaymentRepo(dataData, paymentConfigReader) paymentOrderRepo := payment.NewPaymentOrderRepo(dataData) paymentUsecase := payment2.NewPaymentUsecase(paymentRepo, paymentOrderRepo, logger) - paymentService := payment3.NewPaymentService(paymentUsecase) - handlerPayment := handler.NewPayment(paymentService) + v12 := payment3.NewPaymentService(paymentUsecase) + handlerPayment := handler.NewPayment(v12) taskRepo := task.NewTaskRepo(dataData) registry := app.TaskRegistry(catalog) taskUsecase := task2.NewTaskUsecaseWithRegistry(taskRepo, registry) @@ -117,53 +117,53 @@ func wireApp(configServer *config.Server, store *config.Store, logger *slog.Logg taskScheduler := worker.NewTaskScheduler(taskUsecase, authorityUsecase, taskExecutor, logger) taskRuntime := worker.NewTaskRuntime(taskScheduler) taskApplicationUsecase := task2.NewTaskApplicationUsecase(taskUsecase, taskRuntime) - taskService := task3.NewTaskService(taskApplicationUsecase) - handlerTask := handler.NewTask(taskService) + v13 := task3.NewTaskService(taskApplicationUsecase) + handlerTask := handler.NewTask(v13) mediaRepo := system.NewMediaRepo(dataData) mediaUsecase := system2.NewMediaUsecase(mediaRepo, reloadable, runtimeSettings) - mediaService := system3.NewMediaService(mediaUsecase, runtimeSettings) - media := handler.NewMedia(mediaService) + v14 := system3.NewMediaService(mediaUsecase, runtimeSettings) + media := handler.NewMedia(v14) auditQueryRepo := system.NewAuditRepo(dataData) auditUsecase := system2.NewAuditUsecase(auditQueryRepo) - auditService := system3.NewAuditService(auditUsecase) + v15 := system3.NewAuditService(auditUsecase) logFileRepo := system.NewLogFileRepo(dataData) logViewerUsecase := system2.NewLogViewerUsecase(logFileRepo) - logViewerService := system3.NewLogViewerService(logViewerUsecase) - audit := handler.NewAudit(auditService, auditRecorder, logViewerService, logger) + v16 := system3.NewLogViewerService(logViewerUsecase) + audit := handler.NewAudit(v15, v3, v16, logger) exportRepo := system.NewExportRepo(dataData) exportUsecase := system2.NewExportUsecase(exportRepo) - exportService := system3.NewExportService(exportUsecase, systemCache) - export := handler.NewExport(exportService) + v17 := system3.NewExportService(exportUsecase, systemCache) + export := handler.NewExport(v17) versionRepo := system.NewVersionRepo(dataData) versionUsecase := system2.NewVersionUsecase(versionRepo) - versionService := system3.NewVersionService(versionUsecase) - version := handler.NewVersion(versionService) + v18 := system3.NewVersionService(versionUsecase) + version := handler.NewVersion(v18) dictionaryRepo := system.NewDictionaryRepo(dataData) dictionaryUsecase := system2.NewDictionaryUsecase(dictionaryRepo) - dictionaryService := system3.NewDictionaryService(dictionaryUsecase) - dictionary := handler.NewDictionary(dictionaryService) + v19 := system3.NewDictionaryService(dictionaryUsecase) + dictionary := handler.NewDictionary(v19) parameterRepo := system.NewParameterRepo(dataData) parameterUsecase := system2.NewParameterUsecase(parameterRepo) - parameterService := system3.NewParameterService(parameterUsecase) - parameter := handler.NewParameter(parameterService) - tokenService := system3.NewTokenService(tokenUsecase, tokenIssuer) - apiToken := handler.NewAPIToken(tokenService) + v20 := system3.NewParameterService(parameterUsecase) + parameter := handler.NewParameter(v20) + v21 := system3.NewTokenService(tokenUsecase, tokenIssuer) + apiToken := handler.NewAPIToken(v21) initializationRepo := initialize.NewRepo(dataData, catalog) taskReloader := worker.NewTaskReloader(taskScheduler) systemConfigUsecase := system2.NewSystemConfigUsecase(initializationRepo, taskReloader) - systemConfigService := system3.NewSystemConfigService(systemConfigUsecase, runtimeSettings) - systemConfig := handler.NewSystemConfig(systemConfigService, securityService) - public := handler.NewPublic(authService, systemConfigService, securityService) - userService := system3.NewUserService(userUsecase, securityService) - user := handler.NewUser(userService, authService) - navigation := handler.NewNavigation(userService) - session := handler.NewSession(tokenService) + v22 := system3.NewSystemConfigService(systemConfigUsecase, runtimeSettings) + systemConfig := handler.NewSystemConfig(v22, v2) + public := handler.NewPublic(authService, v22, v2) + v23 := system3.NewUserService(userUsecase, v2) + user := handler.NewUser(v23, authService) + navigation := handler.NewNavigation(v23) + session := handler.NewSession(v21) integrationConfigRepo := integration.NewIntegrationConfigRepo(dataData) runtimeconfigStore := data.NewIntegrationRuntime(dataData) connectivityTester := integration2.NewConnectivityTester(runtimeconfigStore) integrationConfigUsecase := integration3.NewIntegrationConfigUsecase(integrationConfigRepo, connectivityTester) - integrationConfigService := integration4.NewIntegrationConfigService(integrationConfigUsecase) - integrationConfig := handler.NewIntegrationConfig(integrationConfigService) + v24 := integration4.NewIntegrationConfigService(integrationConfigUsecase) + integrationConfig := handler.NewIntegrationConfig(v24) set := &handler.Set{ Authority: authority, Menu: menu, @@ -199,7 +199,7 @@ func wireApp(configServer *config.Server, store *config.Store, logger *slog.Logg cleanup() return nil, nil, err } - engine := server.NewGinEngineWithRuntime(store, accessControlService, authService, securityService, auditRecorder, logger, string2, runtime, websocketServer) + engine := server.NewGinEngineWithRuntime(store, v, authService, v2, v3, logger, string2, runtime, websocketServer) httpServer := server.NewGinServer(configServer, engine) mqReloadable, cleanup3, err := mq.New(runtimeconfigStore, logger) if err != nil { @@ -207,8 +207,7 @@ func wireApp(configServer *config.Server, store *config.Store, logger *slog.Logg cleanup() return nil, nil, err } - resourceRegistry := installGlobalResources(logger, dataData, reloadable, mqReloadable, websocketServer, taskScheduler) - kratosApp := newApp(logger, httpServer, taskScheduler, auditRecorder, reloadableLogger, mqReloadable, resourceRegistry) + kratosApp := newApp(logger, httpServer, taskScheduler, v3, reloadableLogger, mqReloadable) return kratosApp, func() { cleanup3() cleanup2() diff --git a/internal/data/config_store.go b/internal/data/config_store.go index 5d32405..dbe50ad 100644 --- a/internal/data/config_store.go +++ b/internal/data/config_store.go @@ -446,6 +446,5 @@ func (d *Data) reloadConfig(ctx context.Context) error { if d.storage != nil { d.storage.Replace(candidateStorage) } - d.notifyResources() return nil } diff --git a/internal/data/data.go b/internal/data/data.go index 0ad6db4..7646e5f 100644 --- a/internal/data/data.go +++ b/internal/data/data.go @@ -46,7 +46,6 @@ func NewIntegrationRuntime(data *Data) *runtimeconfig.Store { type Data struct { initMu sync.Mutex configMu sync.Mutex - resourceMu sync.RWMutex databaseReady atomic.Bool gormDB *reloadableDB redis *reloadableRedis @@ -65,7 +64,6 @@ type Data struct { appLogger *slog.Logger auditLog *dataScopeAuditWriter catalog module.Catalog - resourceHook func() } // DB exposes the active primary database to narrowly scoped data submodules. @@ -172,30 +170,6 @@ func (d *Data) NamedRedisClients() map[string]redis.UniversalClient { return result } -// SetResourceHook registers a composition-root callback invoked after active -// database, Redis, or Mongo handles change. A callback keeps data independent -// from the global package while allowing the root to refresh shared handles. -func (d *Data) SetResourceHook(hook func()) { - if d == nil { - return - } - d.resourceMu.Lock() - d.resourceHook = hook - d.resourceMu.Unlock() -} - -func (d *Data) notifyResources() { - if d == nil { - return - } - d.resourceMu.RLock() - hook := d.resourceHook - d.resourceMu.RUnlock() - if hook != nil { - hook() - } -} - func (d *Data) logger() *slog.Logger { if d != nil && d.appLogger != nil { return d.appLogger @@ -467,5 +441,4 @@ func (d *Data) activateDatabase(db *gorm.DB, config *config.Database) { d.gormDB.replace(db, d.enqueueDataScopeAudit) d.runtime.UpdateDatabase(config) d.databaseReady.Store(true) - d.notifyResources() } diff --git a/internal/data/initialization_backend.go b/internal/data/initialization_backend.go index 2eeb920..4531b25 100644 --- a/internal/data/initialization_backend.go +++ b/internal/data/initialization_backend.go @@ -105,7 +105,6 @@ func (d *Data) PersistRuntimeConfig(ctx context.Context, value *configpkg.Config if d.storage != nil { d.storage.Replace(candidateStorage) } - d.notifyResources() return nil } func (d *Data) ReloadConfig(ctx context.Context) error { @@ -213,6 +212,5 @@ func (d *Data) InitializeDatabase(ctx context.Context, input *system.DatabaseCon if d.integrations != nil { d.integrations.Replace(integrationConfigs) } - d.notifyResources() return nil } diff --git a/internal/global/resources.go b/internal/global/resources.go deleted file mode 100644 index 297da8e..0000000 --- a/internal/global/resources.go +++ /dev/null @@ -1,163 +0,0 @@ -package global - -import ( - "context" - "io" - "log/slog" - "reflect" - "sync" - - "github.com/redis/go-redis/v9" - "go.mongodb.org/mongo-driver/mongo" - "gorm.io/gorm" - platformmq "kra/pkg/mq" - platformws "kra/pkg/websocket" -) - -// FileStorage is the transport-neutral subset shared by every file storage -// implementation. Domain operations that return business models remain in -// the biz-owned FileStorage interface instead of making global depend on biz. -type FileStorage interface { - Open(context.Context, string) (io.ReadCloser, error) - Delete(context.Context, string) error - DeletePrefix(context.Context, string) error -} - -// Scheduler owns the lifecycle of the process-wide task scheduler. Scheduling -// business operations remain behind the task domain interfaces. -type Scheduler interface { - Start(context.Context) error - Stop(context.Context) error -} - -// Resources is an immutable snapshot from the registry's point of view. -// Replacing a resource publishes a new snapshot; it does not close the old -// resource because lifecycle ownership belongs to the composition root. -type Resources struct { - Logger *slog.Logger - DB *gorm.DB - NamedDBs map[string]*gorm.DB - Redis redis.UniversalClient - NamedRedis map[string]redis.UniversalClient - Mongo *mongo.Client - Storage FileStorage - MQ platformmq.Registry - WebSocket platformws.Hub - Scheduler Scheduler -} - -// ResourceRegistry provides concurrency-safe access to process-wide runtime -// resources. Prefer constructor injection for domain code; the registry is for -// framework callbacks and integration points that cannot be wired directly. -type ResourceRegistry struct { - mu sync.RWMutex - resources Resources -} - -func NewResourceRegistry() *ResourceRegistry { - return &ResourceRegistry{resources: normalizeResources(Resources{})} -} - -func (r *ResourceRegistry) Snapshot() Resources { - if r == nil { - return normalizeResources(Resources{}) - } - r.mu.RLock() - resources := cloneResources(r.resources) - r.mu.RUnlock() - return resources -} - -func (r *ResourceRegistry) Replace(resources Resources) { - if r == nil { - return - } - r.mu.Lock() - r.resources = normalizeResources(resources) - r.mu.Unlock() -} - -func normalizeResources(resources Resources) Resources { - if resources.Logger == nil { - resources.Logger = slog.Default() - } - resources.Redis = normalizeInterface(resources.Redis) - resources.Storage = normalizeInterface(resources.Storage) - resources.MQ = normalizeInterface(resources.MQ) - resources.WebSocket = normalizeInterface(resources.WebSocket) - resources.Scheduler = normalizeInterface(resources.Scheduler) - resources.NamedDBs = cloneDBMap(resources.NamedDBs) - resources.NamedRedis = cloneRedisMap(resources.NamedRedis) - return resources -} - -func cloneResources(resources Resources) Resources { - resources.NamedDBs = cloneDBMap(resources.NamedDBs) - resources.NamedRedis = cloneRedisMap(resources.NamedRedis) - return resources -} - -func cloneDBMap(source map[string]*gorm.DB) map[string]*gorm.DB { - if len(source) == 0 { - return nil - } - copyOf := make(map[string]*gorm.DB, len(source)) - for name, db := range source { - if name != "" && db != nil { - copyOf[name] = db - } - } - if len(copyOf) == 0 { - return nil - } - return copyOf -} - -func cloneRedisMap(source map[string]redis.UniversalClient) map[string]redis.UniversalClient { - if len(source) == 0 { - return nil - } - copyOf := make(map[string]redis.UniversalClient, len(source)) - for name, client := range source { - if name == "" { - continue - } - client = normalizeInterface(client) - if client != nil { - copyOf[name] = client - } - } - if len(copyOf) == 0 { - return nil - } - return copyOf -} - -func normalizeInterface[T any](value T) T { - reflected := reflect.ValueOf(value) - if reflected.IsValid() { - switch reflected.Kind() { - case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: - if !reflected.IsNil() { - return value - } - default: - return value - } - } - var zero T - return zero -} - -var defaultResources = NewResourceRegistry() - -// Install publishes resources assembled by the composition root and returns -// the process-wide registry for lifecycle wiring. -func Install(resources Resources) *ResourceRegistry { - defaultResources.Replace(resources) - return defaultResources -} - -func DefaultRegistry() *ResourceRegistry { return defaultResources } - -func ResourceSnapshot() Resources { return defaultResources.Snapshot() } diff --git a/internal/global/resources_test.go b/internal/global/resources_test.go deleted file mode 100644 index c38958b..0000000 --- a/internal/global/resources_test.go +++ /dev/null @@ -1,147 +0,0 @@ -package global - -import ( - "context" - "io" - "log/slog" - "sync" - "testing" - - "github.com/redis/go-redis/v9" - "go.mongodb.org/mongo-driver/mongo" - "gorm.io/gorm" - platformmq "kra/pkg/mq" - platformws "kra/pkg/websocket" -) - -type storageStub struct{} - -func (*storageStub) Open(context.Context, string) (io.ReadCloser, error) { return nil, nil } -func (*storageStub) Delete(context.Context, string) error { return nil } -func (*storageStub) DeletePrefix(context.Context, string) error { return nil } - -type schedulerStub struct{} - -func (*schedulerStub) Start(context.Context) error { return nil } -func (*schedulerStub) Stop(context.Context) error { return nil } - -type mqStub struct{} - -func (*mqStub) Register(platformmq.SubscriptionSet) error { return nil } -func (*mqStub) Unregister(string) error { return nil } -func (*mqStub) Client(string) platformmq.Client { return nil } -func (*mqStub) PublishTo(context.Context, string, string, []byte, byte, bool) error { - return nil -} -func (*mqStub) SubscribeTo(context.Context, string, string, byte, platformmq.Handler) error { - return nil -} -func (*mqStub) UnsubscribeFrom(context.Context, string, ...string) error { return nil } -func (*mqStub) ConnectedTo(string) bool { return true } - -type webSocketStub struct{} - -func (*webSocketStub) Enabled() bool { return true } -func (*webSocketStub) Broadcast([]byte) error { return nil } -func (*webSocketStub) BroadcastBinary([]byte) error { return nil } -func (*webSocketStub) Sessions() ([]*platformws.Session, error) { return nil, nil } -func (*webSocketStub) Len() int { return 0 } -func (*webSocketStub) Send(*platformws.Session, []byte) error { return nil } -func (*webSocketStub) SendBinary(*platformws.Session, []byte) error { return nil } -func (*webSocketStub) OnMessage(func(*platformws.Session, []byte)) {} -func (*webSocketStub) OnBinaryMessage(func(*platformws.Session, []byte)) {} -func (*webSocketStub) OnConnect(func(*platformws.Session)) {} -func (*webSocketStub) OnDisconnect(func(*platformws.Session)) {} - -func TestResourceRegistryDefaults(t *testing.T) { - registry := NewResourceRegistry() - resources := registry.Snapshot() - if resources.Logger == nil { - t.Fatal("default logger is nil") - } - if resources.DB != nil || len(resources.NamedDBs) != 0 || resources.Redis != nil || len(resources.NamedRedis) != 0 || resources.Mongo != nil || resources.Storage != nil || resources.MQ != nil || resources.WebSocket != nil || resources.Scheduler != nil { - t.Fatalf("unexpected initialized resource: %#v", resources) - } -} - -func TestResourceRegistryReplaceAndSnapshot(t *testing.T) { - logger := slog.Default().With("test", true) - db := &gorm.DB{} - redisClient := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"}) - t.Cleanup(func() { _ = redisClient.Close() }) - mongoClient := &mongo.Client{} - storage := &storageStub{} - mq := &mqStub{} - webSocket := &webSocketStub{} - scheduler := &schedulerStub{} - - registry := NewResourceRegistry() - inputDBs := map[string]*gorm.DB{"reporting": db} - inputRedis := map[string]redis.UniversalClient{"cache": redisClient} - registry.Replace(Resources{ - Logger: logger, DB: db, NamedDBs: inputDBs, - Redis: redisClient, NamedRedis: inputRedis, Mongo: mongoClient, - Storage: storage, MQ: mq, WebSocket: webSocket, Scheduler: scheduler, - }) - resources := registry.Snapshot() - if resources.Logger != logger || resources.DB != db || resources.Redis != redisClient || resources.Mongo != mongoClient || resources.Storage != storage || resources.MQ != mq || resources.WebSocket != webSocket || resources.Scheduler != scheduler { - t.Fatalf("replacement snapshot does not match: %#v", resources) - } - if resources.NamedDBs["reporting"] != db || resources.NamedRedis["cache"] != redisClient { - t.Fatal("named resource lookup returned the wrong resource") - } - - inputDBs["mutated"] = &gorm.DB{} - inputRedis["mutated"] = redisClient - if _, exists := registry.Snapshot().NamedDBs["mutated"]; exists { - t.Fatal("named database map was not copied on replacement") - } - if _, exists := registry.Snapshot().NamedRedis["mutated"]; exists { - t.Fatal("named redis map was not copied on replacement") - } - - resources.NamedDBs["mutated"] = db - if _, exists := registry.Snapshot().NamedDBs["mutated"]; exists { - t.Fatal("snapshot shares its named database map with the registry") - } -} - -func TestResourceRegistryNormalizesTypedNilInterfaces(t *testing.T) { - registry := NewResourceRegistry() - var redisClient *redis.Client - var storage *storageStub - var mq *mqStub - var webSocket *webSocketStub - var scheduler *schedulerStub - registry.Replace(Resources{Redis: redisClient, Storage: storage, MQ: mq, WebSocket: webSocket, Scheduler: scheduler}) - resources := registry.Snapshot() - if resources.Redis != nil || resources.Storage != nil || resources.MQ != nil || resources.WebSocket != nil || resources.Scheduler != nil { - t.Fatalf("typed nil resource was retained: %#v", resources) - } - if resources.Logger == nil { - t.Fatal("nil logger was not replaced with the default logger") - } -} - -func TestResourceRegistryConcurrentAccess(t *testing.T) { - registry := NewResourceRegistry() - const iterations = 1000 - var wait sync.WaitGroup - for index := 0; index < 8; index++ { - wait.Add(1) - go func() { - defer wait.Done() - for step := 0; step < iterations; step++ { - registry.Replace(Resources{DB: &gorm.DB{}}) - if registry.Snapshot().Logger == nil { - t.Error("concurrent snapshot lost the default logger") - return - } - } - }() - } - wait.Wait() - if registry.Snapshot().DB == nil { - t.Fatal("concurrent replacement lost the database") - } -}