diff --git a/backend/core/container.go b/backend/core/container.go index 17649b1d..c7692722 100644 --- a/backend/core/container.go +++ b/backend/core/container.go @@ -40,6 +40,12 @@ func isNil(i any) bool { } } +func (c *Container) remove(targetType reflect.Type) { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.services, targetType) +} + // Provide registers a typed service implementation into the Context's IoC container. func Provide[T any](ctx *Context, service T) { if ctx == nil { @@ -50,7 +56,13 @@ func Provide[T any](ctx *Context, service T) { } targetType := reflect.TypeFor[T]() - ctx.Container().provide(targetType, service) + targetContainer := ctx.Container() + targetContainer.provide(targetType, service) + + ctx.OnDispose(func() error { + targetContainer.remove(targetType) + return nil + }) } func (c *Container) provide(targetType reflect.Type, service any) { diff --git a/backend/core/context.go b/backend/core/context.go index 8e7b5a48..c83ef624 100644 --- a/backend/core/context.go +++ b/backend/core/context.go @@ -10,6 +10,7 @@ import ( "sync" "time" + "Wavelet/core/contracts" "Wavelet/core/extpoints" ) @@ -214,6 +215,24 @@ func (c *Context) Setting() extpoints.SettingExtension { return c.settings } +// DB returns the contracts.DBService registered in the IoC container, or nil if not registered. +func (c *Context) DB() contracts.DBService { + svc, err := Inject[contracts.DBService](c) + if err != nil { + return nil + } + return svc +} + +// Cache returns the contracts.CacheService registered in the IoC container, or nil if not registered. +func (c *Context) Cache() contracts.CacheService { + svc, err := Inject[contracts.CacheService](c) + if err != nil { + return nil + } + return svc +} + // OnDispose registers a cleanup callback function to be executed when this Context is disposed. // It accepts func() error, func(), or Disposer. func (c *Context) OnDispose(fn any) { diff --git a/backend/core/contracts/auth.go b/backend/core/contracts/auth.go index 75bf61dd..fd37e277 100644 --- a/backend/core/contracts/auth.go +++ b/backend/core/contracts/auth.go @@ -46,6 +46,33 @@ type OAuthUserInfoDTO struct { AvatarURL string `json:"avatar_url"` } +// AuthSourceDTO represents an OAuth / OIDC authentication source. +type AuthSourceDTO struct { + ID uint64 `json:"id,string"` + Name string `json:"name"` + Type string `json:"type"` + DisplayName string `json:"display_name"` + ClientID string `json:"client_id"` + ClientSecret string `json:"client_secret,omitempty"` + OpenIDDiscoveryURL string `json:"openid_discovery_url"` + Scopes string `json:"scopes"` + IconURL string `json:"icon_url"` + IsActive bool `json:"is_active"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// AuthSourceViewDTO is a sanitized view of an AuthSource for admin display. +type AuthSourceViewDTO struct { + ID uint64 `json:"id,string"` + Name string `json:"name"` + Type string `json:"type"` + DisplayName string `json:"display_name"` + IsActive bool `json:"is_active"` + IconURL string `json:"icon_url"` + ClientSecretConfigured bool `json:"client_secret_configured"` +} + // OAuthProvider defines the pluggable OAuth provider contract. type OAuthProvider interface { Name() string @@ -79,6 +106,27 @@ type AuthService interface { // RevokeUserSessions revokes all active sessions and cached tokens for a user. RevokeUserSessions(ctx context.Context, userID uint64) error + // InvalidateCachedUser invalidates cached user profile data. + InvalidateCachedUser(ctx context.Context, userID uint64) + + // InvalidateCachedToken invalidates cached access token data. + InvalidateCachedToken(ctx context.Context, tokenHash string) + + // ListAuthSources lists all configured authentication sources. + ListAuthSources(ctx context.Context) ([]AuthSourceViewDTO, error) + + // CreateAuthSource creates a new authentication source. + CreateAuthSource(ctx context.Context, source AuthSourceDTO) (*AuthSourceDTO, error) + + // UpdateAuthSource updates an authentication source. + UpdateAuthSource(ctx context.Context, id uint64, source AuthSourceDTO) (*AuthSourceDTO, error) + + // DeleteAuthSource removes an authentication source. + DeleteAuthSource(ctx context.Context, id uint64) error + + // ToggleAuthSource toggles the active state of an authentication source. + ToggleAuthSource(ctx context.Context, id uint64) (*AuthSourceDTO, error) + // DisallowTokenAuthMiddleware returns a middleware that rejects requests authenticated via access token. DisallowTokenAuthMiddleware() any } diff --git a/backend/core/contracts/events.go b/backend/core/contracts/events.go index 1c372b79..5b50ace6 100644 --- a/backend/core/contracts/events.go +++ b/backend/core/contracts/events.go @@ -26,6 +26,13 @@ const ( // EventTopicUserDeleted fires when a user account is deleted. EventTopicUserDeleted = "user:deleted" + + // EventTopicUserStatusChanged fires when a user account active status changes. + EventTopicUserStatusChanged = "user:status_changed" + + // EventTopicTokenRevoked fires when an access token is revoked. + // #nosec G101 + EventTopicTokenRevoked = "auth:token_revoked" ) // --- Admin & System Events --- @@ -107,3 +114,21 @@ type NotificationSentEvent struct { Success bool `json:"success"` ErrorInfo string `json:"error_info,omitempty"` } + +// UserStatusChangedEvent fires when a user status is enabled/disabled. +type UserStatusChangedEvent struct { + UserID uint64 `json:"user_id,string"` + IsActive bool `json:"is_active"` +} + +// TokenRevokedEvent fires when an access token is revoked. +type TokenRevokedEvent struct { + UserID uint64 `json:"user_id,string"` + TokenHash string `json:"token_hash"` +} + +// UserDeletedEvent fires when a user account is deleted. +type UserDeletedEvent struct { + CurrentUserID uint64 `json:"current_user_id,string"` + TargetUserID uint64 `json:"target_user_id,string"` +} diff --git a/backend/core/contracts/user.go b/backend/core/contracts/user.go index cdf5d175..701ee46c 100644 --- a/backend/core/contracts/user.go +++ b/backend/core/contracts/user.go @@ -29,6 +29,34 @@ type UpdateUserProfileRequest struct { Location *string `json:"location,omitempty"` } +// AdminListUsersFilter contains query parameters for filtering users in admin panel. +type AdminListUsersFilter struct { + Page int + PageSize int + UserID *uint64 + Username string + Email string +} + +// AdminCreateUserRequest contains fields for admin to create a user. +type AdminCreateUserRequest struct { + Username string `json:"username"` + Password string `json:"password"` + Nickname string `json:"nickname"` + Email string `json:"email"` + IsActive bool `json:"is_active"` + IsAdmin bool `json:"is_admin"` +} + +// AdminUpdateUserRequest contains fields for admin to update a user. +type AdminUpdateUserRequest struct { + ID uint64 `json:"id,string"` + Nickname string `json:"nickname"` + Email string `json:"email"` + IsAdmin bool `json:"is_admin"` + Password string `json:"password,omitempty"` +} + // UserService defines the contract for user account management and profile queries. type UserService interface { // GetUserByID retrieves a user by ID. @@ -81,4 +109,22 @@ type UserService interface { // UniqueUsername generates a unique username candidate based on base. UniqueUsername(ctx context.Context, base string) (string, error) + + // AdminListUsers returns a filtered paginated list of users for admin management. + AdminListUsers(ctx context.Context, filter AdminListUsersFilter) (int64, []*UserDTO, error) + + // AdminGetUser retrieves complete user details by ID for admin management. + AdminGetUser(ctx context.Context, id uint64) (*UserDTO, error) + + // AdminCreateUser creates a user with admin specified options. + AdminCreateUser(ctx context.Context, req AdminCreateUserRequest) (*UserDTO, error) + + // AdminUpdateUser updates user details, email, nickname, admin role, and optional password. + AdminUpdateUser(ctx context.Context, currentUserID uint64, req AdminUpdateUserRequest) error + + // AdminUpdateUserStatus updates a user's active status (with admin protection). + AdminUpdateUserStatus(ctx context.Context, id uint64, active bool) error + + // AdminDeleteUser deletes a user (with self and admin protection, cascading tokens and accounts). + AdminDeleteUser(ctx context.Context, currentUserID, targetID uint64) error } diff --git a/backend/core/events.go b/backend/core/events.go index 7f7f37fd..a47a4fbf 100644 --- a/backend/core/events.go +++ b/backend/core/events.go @@ -21,13 +21,16 @@ type eventListener struct { id uint64 fnVal reflect.Value numIn int + numOut int hasCtx bool hasPayload bool argType reflect.Type returnsErr bool + returnsVal bool } -// EventBus is a thread-safe, strongly-typed in-process domain event bus. +// EventBus is a thread-safe, strongly-typed in-process domain event bus supporting +// Emit, Waterfall, Parallel, and Serial dispatch semantics. type EventBus struct { mu sync.RWMutex nextID atomic.Uint64 @@ -44,6 +47,10 @@ func NewEventBus() *EventBus { // On registers an event handler for the given topic. // // Supported handler signatures: +// - func(ctx context.Context, event T) (T, error) +// - func(ctx context.Context, event T) T +// - func(event T) (T, error) +// - func(event T) T // - func(ctx context.Context, event T) error // - func(ctx context.Context, event T) // - func(event T) error @@ -71,17 +78,29 @@ func (b *EventBus) On(topic string, handler any) Disposer { panic(fmt.Sprintf("core/events: handler has %d parameters, maximum 2 supported (ctx, event)", numIn)) } + const maxHandlerReturnValues = 2 numOut := fnType.NumOut() - if numOut > 1 { - panic(fmt.Sprintf("core/events: handler has %d return values, maximum 1 supported (error)", numOut)) + if numOut > maxHandlerReturnValues { + panic(fmt.Sprintf("core/events: handler has %d return values, maximum 2 supported (value, error)", numOut)) } returnsErr := false - if numOut == 1 { - outType := fnType.Out(0) - if !outType.Implements(errInterfaceType) { - panic(fmt.Sprintf("core/events: handler return type must be error, got %v", outType)) + returnsVal := false + + switch numOut { + case 1: + out0 := fnType.Out(0) + if out0.Implements(errInterfaceType) { + returnsErr = true + } else { + returnsVal = true } + case 2: + out1 := fnType.Out(1) + if !out1.Implements(errInterfaceType) { + panic(fmt.Sprintf("core/events: second return value must be error, got %v", out1)) + } + returnsVal = true returnsErr = true } @@ -89,7 +108,9 @@ func (b *EventBus) On(topic string, handler any) Disposer { id: b.nextID.Add(1), fnVal: fnVal, numIn: numIn, + numOut: numOut, returnsErr: returnsErr, + returnsVal: returnsVal, } switch numIn { @@ -162,17 +183,11 @@ func (b *EventBus) Emit(ctx context.Context, topic string, payload any) error { ctx = context.Background() } - b.mu.RLock() - rawListeners := b.handlers[topic] - if len(rawListeners) == 0 { - b.mu.RUnlock() + listeners := b.getListeners(topic) + if len(listeners) == 0 { return nil } - listeners := make([]eventListener, len(rawListeners)) - copy(listeners, rawListeners) - b.mu.RUnlock() - var payloadVal reflect.Value if payload != nil { payloadVal = reflect.ValueOf(payload) @@ -190,8 +205,11 @@ func (b *EventBus) Emit(ctx context.Context, topic string, payload any) error { }() results := l.fnVal.Call(args) - if l.returnsErr && len(results) > 0 && !results[0].IsNil() { - resErr = results[0].Interface().(error) + if l.returnsErr { + errIdx := l.numOut - 1 + if len(results) > errIdx && !results[errIdx].IsNil() { + resErr = results[errIdx].Interface().(error) + } } return resErr }() @@ -204,6 +222,181 @@ func (b *EventBus) Emit(ctx context.Context, topic string, payload any) error { return errors.Join(errs...) } +// Waterfall runs handlers sequentially as a transformation pipeline. +// The returned value of each handler becomes the payload input for the next handler. +// If any handler returns an error or panics, execution aborts immediately. +// +//nolint:contextcheck +func (b *EventBus) Waterfall(ctx context.Context, topic string, initialPayload any) (any, error) { + if ctx == nil { + ctx = context.Background() + } + + listeners := b.getListeners(topic) + if len(listeners) == 0 { + return initialPayload, nil + } + + currentPayload := initialPayload + + for _, l := range listeners { + var payloadVal reflect.Value + if currentPayload != nil { + payloadVal = reflect.ValueOf(currentPayload) + } + + args := b.buildArgs(ctx, l, payloadVal) + + var stepVal any + var stepErr error + + func() { + defer func() { + if r := recover(); r != nil { + stepErr = fmt.Errorf("core/events: panic in waterfall handler for topic %q: %v", topic, r) + } + }() + + results := l.fnVal.Call(args) + if l.returnsErr { + errIdx := l.numOut - 1 + if len(results) > errIdx && !results[errIdx].IsNil() { + stepErr = results[errIdx].Interface().(error) + } + } + if stepErr == nil && l.returnsVal && len(results) > 0 { + stepVal = results[0].Interface() + } + }() + + if stepErr != nil { + return nil, stepErr + } + + if l.returnsVal { + currentPayload = stepVal + } + } + + return currentPayload, nil +} + +// Parallel executes all subscribers of the topic concurrently in separate goroutines. +// It waits for all handlers to complete and collects any errors via errors.Join. +// +//nolint:contextcheck +func (b *EventBus) Parallel(ctx context.Context, topic string, payload any) error { + if ctx == nil { + ctx = context.Background() + } + + listeners := b.getListeners(topic) + if len(listeners) == 0 { + return nil + } + + var payloadVal reflect.Value + if payload != nil { + payloadVal = reflect.ValueOf(payload) + } + + var wg sync.WaitGroup + errCh := make(chan error, len(listeners)) + + for _, l := range listeners { + wg.Add(1) + go func(listener eventListener) { + defer wg.Done() + defer func() { + if r := recover(); r != nil { + errCh <- fmt.Errorf("core/events: panic in parallel handler for topic %q: %v", topic, r) + } + }() + + args := b.buildArgs(ctx, listener, payloadVal) + results := listener.fnVal.Call(args) + if listener.returnsErr { + errIdx := listener.numOut - 1 + if len(results) > errIdx && !results[errIdx].IsNil() { + errCh <- results[errIdx].Interface().(error) + } + } + }(l) + } + + wg.Wait() + close(errCh) + + var errs []error + for err := range errCh { + if err != nil { + errs = append(errs, err) + } + } + + return errors.Join(errs...) +} + +// Serial executes subscribers strictly in sequence. +// If any subscriber returns an error or panics, execution stops immediately and returns that error. +// +//nolint:contextcheck +func (b *EventBus) Serial(ctx context.Context, topic string, payload any) error { + if ctx == nil { + ctx = context.Background() + } + + listeners := b.getListeners(topic) + if len(listeners) == 0 { + return nil + } + + var payloadVal reflect.Value + if payload != nil { + payloadVal = reflect.ValueOf(payload) + } + + for _, l := range listeners { + args := b.buildArgs(ctx, l, payloadVal) + + var stepErr error + func() { + defer func() { + if r := recover(); r != nil { + stepErr = fmt.Errorf("core/events: panic in serial handler for topic %q: %v", topic, r) + } + }() + + results := l.fnVal.Call(args) + if l.returnsErr { + errIdx := l.numOut - 1 + if len(results) > errIdx && !results[errIdx].IsNil() { + stepErr = results[errIdx].Interface().(error) + } + } + }() + + if stepErr != nil { + return stepErr + } + } + + return nil +} + +func (b *EventBus) getListeners(topic string) []eventListener { + b.mu.RLock() + raw := b.handlers[topic] + if len(raw) == 0 { + b.mu.RUnlock() + return nil + } + listeners := make([]eventListener, len(raw)) + copy(listeners, raw) + b.mu.RUnlock() + return listeners +} + func (b *EventBus) buildArgs(ctx context.Context, l eventListener, payloadVal reflect.Value) []reflect.Value { if l.numIn == 0 { return nil diff --git a/backend/core/events_test.go b/backend/core/events_test.go index 4572ee7d..2c830f2d 100644 --- a/backend/core/events_test.go +++ b/backend/core/events_test.go @@ -247,13 +247,13 @@ func TestEventBusInvalidHandlerPanics(t *testing.T) { }) assert.Panics(t, func() { - // More than 1 return value - bus.On("test:invalid", func() (int, error) { return 0, nil }) + // More than 2 return values + bus.On("test:invalid", func() (int, string, error) { return 0, "", nil }) }) assert.Panics(t, func() { - // Return value is not error - bus.On("test:invalid", func() int { return 0 }) + // 2 return values, but second is not error + bus.On("test:invalid", func() (int, string) { return 0, "" }) }) } @@ -316,3 +316,81 @@ func TestEventBusConcurrentAccess(t *testing.T) { wg.Wait() assert.Greater(t, receivedCount.Load(), int64(0)) } + +func TestEventBusWaterfall(t *testing.T) { + bus := core.NewEventBus() + + // Handler 1: appends "-first" + bus.On("pipeline:transform", func(ctx context.Context, s string) string { + return s + "-first" + }) + + // Handler 2: appends "-second" with error return + bus.On("pipeline:transform", func(s string) (string, error) { + return s + "-second", nil + }) + + res, err := bus.Waterfall(context.Background(), "pipeline:transform", "init") + assert.NoError(t, err) + assert.Equal(t, "init-first-second", res) + + // Test short-circuit on error + expectedErr := errors.New("waterfall step failed") + bus.On("pipeline:error", func(s string) (string, error) { + return s, expectedErr + }) + bus.On("pipeline:error", func(s string) string { + return s + "-should-not-run" + }) + + res, err = bus.Waterfall(context.Background(), "pipeline:error", "start") + assert.ErrorIs(t, err, expectedErr) + assert.Nil(t, res) +} + +func TestEventBusParallel(t *testing.T) { + bus := core.NewEventBus() + + var counter atomic.Int64 + err1 := errors.New("parallel err 1") + + bus.On("test:parallel", func(ctx context.Context, val int) error { + counter.Add(int64(val)) + return nil + }) + + bus.On("test:parallel", func(val int) error { + counter.Add(int64(val)) + return err1 + }) + + err := bus.Parallel(context.Background(), "test:parallel", 10) + assert.ErrorIs(t, err, err1) + assert.Equal(t, int64(20), counter.Load()) +} + +func TestEventBusSerial(t *testing.T) { + bus := core.NewEventBus() + + var executed []int + errStop := errors.New("serial stop") + + bus.On("test:serial", func() error { + executed = append(executed, 1) + return nil + }) + + bus.On("test:serial", func() error { + executed = append(executed, 2) + return errStop + }) + + bus.On("test:serial", func() error { + executed = append(executed, 3) + return nil + }) + + err := bus.Serial(context.Background(), "test:serial", nil) + assert.ErrorIs(t, err, errStop) + assert.Equal(t, []int{1, 2}, executed) +} diff --git a/backend/core/extpoints/extpoints_test.go b/backend/core/extpoints/extpoints_test.go index 052a1d15..902ad816 100644 --- a/backend/core/extpoints/extpoints_test.go +++ b/backend/core/extpoints/extpoints_test.go @@ -278,3 +278,44 @@ func TestContextExtensionPointsIntegration(t *testing.T) { assert.NoError(t, err) assert.True(t, eventReceived) } + +func TestExtensionPointsUnregister(t *testing.T) { + ctx := core.NewContext(context.Background()) + + // 1. Router unregister + rd := ctx.Router().GET("/temp", "temp_handler") + assert.Greater(t, rd.ID, uint64(0)) + assert.Len(t, ctx.Router().Routes(), 1) + assert.True(t, ctx.Router().Unregister("GET", "/temp")) + assert.Len(t, ctx.Router().Routes(), 0) + + rd2 := ctx.Router().POST("/temp2", "temp2_handler") + assert.Len(t, ctx.Router().Routes(), 1) + assert.True(t, ctx.Router().UnregisterByID(rd2.ID)) + assert.Len(t, ctx.Router().Routes(), 0) + + // 2. Task unregister + ctx.Task().Register("temp:task", "handler") + assert.Len(t, ctx.Task().Tasks(), 1) + assert.True(t, ctx.Task().Unregister("temp:task")) + assert.Len(t, ctx.Task().Tasks(), 0) + + // 3. Schedule unregister + ctx.Schedule().RegisterCron("@hourly", "temp:cron", nil) + assert.Len(t, ctx.Schedule().Schedules(), 1) + assert.True(t, ctx.Schedule().Unregister("temp:cron")) + assert.Len(t, ctx.Schedule().Schedules(), 0) + + // 4. Setting unregister + ctx.Settings().Register(extpoints.SettingSchema{Key: "temp.key", Default: 1}) + assert.Len(t, ctx.Settings().Schemas(), 1) + assert.True(t, ctx.Settings().Unregister("temp.key")) + assert.Len(t, ctx.Settings().Schemas(), 0) + + // 5. Migration unregister + fsys := fstest.MapFS{"001.sql": &fstest.MapFile{Data: []byte("-- migration")}} + ctx.Migrations().Register("temp_plugin", fsys) + assert.Len(t, ctx.Migrations().Entries(), 1) + assert.True(t, ctx.Migrations().Unregister("temp_plugin")) + assert.Len(t, ctx.Migrations().Entries(), 0) +} diff --git a/backend/core/extpoints/migration.go b/backend/core/extpoints/migration.go index f897a36a..13d79049 100644 --- a/backend/core/extpoints/migration.go +++ b/backend/core/extpoints/migration.go @@ -21,6 +21,7 @@ type MigrationExtension interface { Register(pluginID string, fsys fs.FS, dir ...string) Entries() []MigrationEntry Get(pluginID string) (MigrationEntry, bool) + Unregister(pluginID string) bool } // MigrationRegistry collects and stores migration entries from plugins. @@ -69,6 +70,25 @@ func (m *MigrationRegistry) Register(pluginID string, fsys fs.FS, dir ...string) m.lookup[pluginID] = entry } +// Unregister removes a registered migration entry by plugin ID. +func (m *MigrationRegistry) Unregister(pluginID string) bool { + m.mu.Lock() + defer m.mu.Unlock() + + if _, exists := m.lookup[pluginID]; !exists { + return false + } + + delete(m.lookup, pluginID) + for i, e := range m.entries { + if e.PluginID == pluginID { + m.entries = append(m.entries[:i], m.entries[i+1:]...) + break + } + } + return true +} + // Entries returns a copy of all registered migration entries in registration order. func (m *MigrationRegistry) Entries() []MigrationEntry { m.mu.RLock() diff --git a/backend/core/extpoints/router.go b/backend/core/extpoints/router.go index 6fe0d3ac..8c18ac59 100644 --- a/backend/core/extpoints/router.go +++ b/backend/core/extpoints/router.go @@ -10,6 +10,7 @@ import ( // RouteDefinition holds the metadata and handler list for a single HTTP route. type RouteDefinition struct { + ID uint64 Method string Path string Handlers []any @@ -31,11 +32,14 @@ type RouterExtension interface { Any(path string, handlers ...any) []RouteDefinition Routes() []RouteDefinition Middlewares() []any + Unregister(method, path string) bool + UnregisterByID(id uint64) bool } // RouterRegistry implements RouterExtension as the root route and middleware collector. type RouterRegistry struct { mu sync.RWMutex + nextID uint64 routes []RouteDefinition middlewares []any } @@ -75,7 +79,9 @@ func (r *RouterRegistry) Handle(method, path string, handlers ...any) RouteDefin r.mu.Lock() defer r.mu.Unlock() + r.nextID++ rd := RouteDefinition{ + ID: r.nextID, Method: strings.ToUpper(method), Path: cleanPath(path), Handlers: handlers, @@ -85,6 +91,37 @@ func (r *RouterRegistry) Handle(method, path string, handlers ...any) RouteDefin return rd } +// Unregister removes a route matching method and path from the registry. +func (r *RouterRegistry) Unregister(method, path string) bool { + r.mu.Lock() + defer r.mu.Unlock() + + targetMethod := strings.ToUpper(method) + targetPath := cleanPath(path) + + for i, rd := range r.routes { + if rd.Method == targetMethod && rd.Path == targetPath { + r.routes = append(r.routes[:i], r.routes[i+1:]...) + return true + } + } + return false +} + +// UnregisterByID removes a route by its unique ID. +func (r *RouterRegistry) UnregisterByID(id uint64) bool { + r.mu.Lock() + defer r.mu.Unlock() + + for i, rd := range r.routes { + if rd.ID == id { + r.routes = append(r.routes[:i], r.routes[i+1:]...) + return true + } + } + return false +} + // GET registers a GET route. func (r *RouterRegistry) GET(path string, handlers ...any) RouteDefinition { return r.Handle("GET", path, handlers...) @@ -176,7 +213,9 @@ func (g *RouterGroup) Handle(method, path string, handlers ...any) RouteDefiniti allMiddlewares = append(allMiddlewares, g.registry.middlewares...) allMiddlewares = append(allMiddlewares, g.middlewares...) + g.registry.nextID++ rd := RouteDefinition{ + ID: g.registry.nextID, Method: strings.ToUpper(method), Path: fullPath, Handlers: handlers, @@ -186,6 +225,17 @@ func (g *RouterGroup) Handle(method, path string, handlers ...any) RouteDefiniti return rd } +// Unregister removes a route under this group prefix matching method and path. +func (g *RouterGroup) Unregister(method, path string) bool { + fullPath := joinPaths(g.prefix, path) + return g.registry.Unregister(method, fullPath) +} + +// UnregisterByID removes a route by its unique ID. +func (g *RouterGroup) UnregisterByID(id uint64) bool { + return g.registry.UnregisterByID(id) +} + // GET registers a GET route in this group. func (g *RouterGroup) GET(path string, handlers ...any) RouteDefinition { return g.Handle("GET", path, handlers...) diff --git a/backend/core/extpoints/schedule.go b/backend/core/extpoints/schedule.go index 386d39fc..de92f4d1 100644 --- a/backend/core/extpoints/schedule.go +++ b/backend/core/extpoints/schedule.go @@ -32,6 +32,7 @@ type ScheduleExtension interface { RegisterCron(spec string, taskType string, payload any, opts ...ScheduleOption) Schedules() []ScheduleDefinition Get(taskType string) (ScheduleDefinition, bool) + Unregister(taskType string) bool } // ScheduleRegistry collects and manages schedule registrations. @@ -85,6 +86,25 @@ func (s *ScheduleRegistry) RegisterCron(spec string, taskType string, payload an s.Register(spec, taskType, payload, opts...) } +// Unregister removes a registered schedule definition by its task type. +func (s *ScheduleRegistry) Unregister(taskType string) bool { + s.mu.Lock() + defer s.mu.Unlock() + + if _, exists := s.lookup[taskType]; !exists { + return false + } + + delete(s.lookup, taskType) + for i, item := range s.schedules { + if item.TaskType == taskType { + s.schedules = append(s.schedules[:i], s.schedules[i+1:]...) + break + } + } + return true +} + // Schedules returns a copy of all registered ScheduleDefinitions. func (s *ScheduleRegistry) Schedules() []ScheduleDefinition { s.mu.RLock() diff --git a/backend/core/extpoints/setting.go b/backend/core/extpoints/setting.go index 23b9d62a..713caf0e 100644 --- a/backend/core/extpoints/setting.go +++ b/backend/core/extpoints/setting.go @@ -22,6 +22,7 @@ type SettingExtension interface { Register(schema SettingSchema) Schemas() []SettingSchema Get(key string) (SettingSchema, bool) + Unregister(key string) bool } // SettingRegistry collects and manages setting configuration schemas. @@ -62,6 +63,25 @@ func (s *SettingRegistry) Register(schema SettingSchema) { s.lookup[schema.Key] = schema } +// Unregister removes a registered SettingSchema by its key. +func (s *SettingRegistry) Unregister(key string) bool { + s.mu.Lock() + defer s.mu.Unlock() + + if _, exists := s.lookup[key]; !exists { + return false + } + + delete(s.lookup, key) + for i, item := range s.schemas { + if item.Key == key { + s.schemas = append(s.schemas[:i], s.schemas[i+1:]...) + break + } + } + return true +} + // Schemas returns a copy of all registered SettingSchemas. func (s *SettingRegistry) Schemas() []SettingSchema { s.mu.RLock() diff --git a/backend/core/extpoints/task.go b/backend/core/extpoints/task.go index d2e0fec5..cc8650f1 100644 --- a/backend/core/extpoints/task.go +++ b/backend/core/extpoints/task.go @@ -57,6 +57,7 @@ type TaskExtension interface { Register(pattern string, handler any, opts ...TaskOption) Tasks() []TaskDefinition Get(pattern string) (TaskDefinition, bool) + Unregister(pattern string) bool } // TaskRegistry collects and manages task registrations. @@ -104,6 +105,25 @@ func (t *TaskRegistry) Register(pattern string, handler any, opts ...TaskOption) t.lookup[pattern] = td } +// Unregister removes a registered task definition by its pattern. +func (t *TaskRegistry) Unregister(pattern string) bool { + t.mu.Lock() + defer t.mu.Unlock() + + if _, exists := t.lookup[pattern]; !exists { + return false + } + + delete(t.lookup, pattern) + for i, item := range t.tasks { + if item.Pattern == pattern { + t.tasks = append(t.tasks[:i], t.tasks[i+1:]...) + break + } + } + return true +} + // Tasks returns a copy of all registered TaskDefinitions. func (t *TaskRegistry) Tasks() []TaskDefinition { t.mu.RLock() diff --git a/backend/plugins/domain/admin/handlers_auth_source.go b/backend/plugins/domain/admin/handlers_auth_source.go index 241b9b21..3f1a4de4 100644 --- a/backend/plugins/domain/admin/handlers_auth_source.go +++ b/backend/plugins/domain/admin/handlers_auth_source.go @@ -7,32 +7,24 @@ import ( "net/http" "strconv" - "Wavelet/pkg/response" - "Wavelet/plugins/domain/auth" - "Wavelet/plugins/infra/database" "github.com/gin-gonic/gin" + + "Wavelet/core/contracts" + "Wavelet/pkg/response" ) // ListAuthSources lists all configured authentication sources. func ListAuthSources(c *gin.Context) { - var sources []auth.AuthSource - gormDB := database.DB(c.Request.Context()) - if err := gormDB.Order("id ASC").Find(&sources).Error; err != nil { - response.AbortInternal(c, "获取认证源列表失败") + authSvc := getAuthService(c.Request.Context()) + if authSvc == nil { + response.AbortInternal(c, "认证服务未就绪") return } - views := make([]auth.AuthSourceView, len(sources)) - for i := range sources { - views[i] = auth.AuthSourceView{ - ID: sources[i].ID, - Name: sources[i].Name, - Type: sources[i].Type, - DisplayName: sources[i].DisplayName, - IsActive: sources[i].IsActive, - IconURL: sources[i].IconURL, - ClientSecretConfigured: sources[i].ClientSecret != "", - } + views, err := authSvc.ListAuthSources(c.Request.Context()) + if err != nil { + response.AbortInternal(c, "获取认证源列表失败") + return } c.JSON(http.StatusOK, response.OK(views)) @@ -40,25 +32,25 @@ func ListAuthSources(c *gin.Context) { // CreateAuthSource creates a new authentication source. func CreateAuthSource(c *gin.Context) { - var source auth.AuthSource + var source contracts.AuthSourceDTO if err := c.ShouldBindJSON(&source); err != nil { response.AbortBadRequest(c, "无效的参数") return } - if err := source.Validate(); err != nil { - response.AbortBadRequest(c, err.Error()) + authSvc := getAuthService(c.Request.Context()) + if authSvc == nil { + response.AbortInternal(c, "认证服务未就绪") return } - gormDB := database.DB(c.Request.Context()) - if err := gormDB.Create(&source).Error; err != nil { + created, err := authSvc.CreateAuthSource(c.Request.Context(), source) + if err != nil { response.AbortBadRequest(c, "创建认证源失败: "+err.Error()) return } - source.Sanitize() - c.JSON(http.StatusOK, response.OK(source)) + c.JSON(http.StatusOK, response.OK(created)) } // UpdateAuthSource updates an authentication source. @@ -70,40 +62,25 @@ func UpdateAuthSource(c *gin.Context) { return } - gormDB := database.DB(c.Request.Context()) - var existing auth.AuthSource - if err := gormDB.First(&existing, id).Error; err != nil { - response.AbortNotFound(c, "认证源不存在") - return - } - - var req auth.AuthSource + var req contracts.AuthSourceDTO if err := c.ShouldBindJSON(&req); err != nil { response.AbortBadRequest(c, "无效的参数") return } - existing.DisplayName = req.DisplayName - existing.ClientID = req.ClientID - if req.ClientSecret != "" { - existing.ClientSecret = req.ClientSecret + authSvc := getAuthService(c.Request.Context()) + if authSvc == nil { + response.AbortInternal(c, "认证服务未就绪") + return } - existing.OpenIDDiscoveryURL = req.OpenIDDiscoveryURL - existing.Scopes = req.Scopes - existing.IconURL = req.IconURL - if err := existing.Validate(); err != nil { + updated, err := authSvc.UpdateAuthSource(c.Request.Context(), id, req) + if err != nil { response.AbortBadRequest(c, err.Error()) return } - if err := gormDB.Save(&existing).Error; err != nil { - response.AbortInternal(c, "更新认证源失败") - return - } - - existing.Sanitize() - c.JSON(http.StatusOK, response.OK(existing)) + c.JSON(http.StatusOK, response.OK(updated)) } // ToggleAuthSource toggles the active state of an auth source. @@ -115,27 +92,19 @@ func ToggleAuthSource(c *gin.Context) { return } - gormDB := database.DB(c.Request.Context()) - var existing auth.AuthSource - if err := gormDB.First(&existing, id).Error; err != nil { - response.AbortNotFound(c, "认证源不存在") + authSvc := getAuthService(c.Request.Context()) + if authSvc == nil { + response.AbortInternal(c, "认证服务未就绪") return } - existing.IsActive = !existing.IsActive - if existing.IsActive { - if err := existing.Validate(); err != nil { - response.AbortBadRequest(c, err.Error()) - return - } - } - - if err := gormDB.Model(&existing).Update("is_active", existing.IsActive).Error; err != nil { - response.AbortInternal(c, "切换认证源状态失败") + toggled, err := authSvc.ToggleAuthSource(c.Request.Context(), id) + if err != nil { + response.AbortInternal(c, "切换认证源状态失败: "+err.Error()) return } - c.JSON(http.StatusOK, response.OK(gin.H{"is_active": existing.IsActive})) + c.JSON(http.StatusOK, response.OK(gin.H{"is_active": toggled.IsActive})) } // DeleteAuthSource deletes an authentication source. @@ -147,9 +116,14 @@ func DeleteAuthSource(c *gin.Context) { return } - gormDB := database.DB(c.Request.Context()) - if err := gormDB.Delete(&auth.AuthSource{}, id).Error; err != nil { - response.AbortInternal(c, "删除认证源失败") + authSvc := getAuthService(c.Request.Context()) + if authSvc == nil { + response.AbortInternal(c, "认证服务未就绪") + return + } + + if err := authSvc.DeleteAuthSource(c.Request.Context(), id); err != nil { + response.AbortInternal(c, "删除认证源失败: "+err.Error()) return } diff --git a/backend/plugins/domain/admin/handlers_config.go b/backend/plugins/domain/admin/handlers_config.go index e8ed9e5a..ded29d58 100644 --- a/backend/plugins/domain/admin/handlers_config.go +++ b/backend/plugins/domain/admin/handlers_config.go @@ -12,11 +12,10 @@ import ( "strings" "time" + "Wavelet/core/contracts" "Wavelet/pkg/logger" mail "Wavelet/pkg/mail" "Wavelet/pkg/response" - "Wavelet/plugins/domain/cap" - cachepkg "Wavelet/plugins/infra/cache" db "Wavelet/plugins/infra/database" "Wavelet/plugins/infra/storage/objectstore" "github.com/gin-gonic/gin" @@ -341,8 +340,8 @@ func invalidateSystemConfigCaches(ctx context.Context, key string) { if err := InvalidateSystemConfigCache(ctx, key); err != nil { logger.WarnF(ctx, "清理系统配置缓存失败: %v", err) } - if cap.IsRuntimeConfigKey(key) { - cap.InvalidateRuntimeSettings() + if globalCoreCtx != nil { + _ = globalCoreCtx.Events().Emit(ctx, contracts.EventTopicConfigChanged, contracts.ConfigChangedEvent{Key: key}) } } @@ -350,17 +349,9 @@ func invalidateCachesAfterConfigUpdate(ctx context.Context, key string) { invalidateSystemConfigCaches(ctx, key) if key == ConfigKeyStorageConfig { - if cachepkg.Redis != nil { - _ = cachepkg.Redis.Publish(ctx, "upload:access_cache:invalidate", "reset").Err() - } objectstore.ResetCache() objectstore.PublishCacheInvalidation(ctx) } - if key == ConfigKeyFileAccessWhitelist { - if cachepkg.Redis != nil { - _ = cachepkg.Redis.Publish(ctx, "upload:access_cache:invalidate", "reset").Err() - } - } if err := InvalidateVisibleSystemConfigsCache(ctx); err != nil { logger.WarnF(ctx, "清理公共配置列表缓存失败: %v", err) diff --git a/backend/plugins/domain/admin/handlers_user.go b/backend/plugins/domain/admin/handlers_user.go index efd8011b..7b49dd54 100644 --- a/backend/plugins/domain/admin/handlers_user.go +++ b/backend/plugins/domain/admin/handlers_user.go @@ -4,27 +4,20 @@ package admin import ( - "context" "errors" "net/http" "strconv" - "strings" "time" "github.com/gin-gonic/gin" "gorm.io/gorm" "Wavelet/core/contracts" - "Wavelet/pkg/idgen" "Wavelet/pkg/logger" "Wavelet/pkg/response" "Wavelet/pkg/util" - "Wavelet/plugins/domain/auth" - db "Wavelet/plugins/infra/database" ) -const minPasswordLength = 8 - // listUsersRequest 用户列表查询请求 type listUsersRequest struct { Page int `form:"page" binding:"min=1"` @@ -136,7 +129,19 @@ func ListUsers(c *gin.Context) { return } - total, dtos, err := listUsers(c.Request.Context(), req) + userSvc := getUserService(c.Request.Context()) + if userSvc == nil { + response.AbortInternal(c, "用户服务未就绪") + return + } + + total, dtos, err := userSvc.AdminListUsers(c.Request.Context(), contracts.AdminListUsersFilter{ + Page: req.Page, + PageSize: req.PageSize, + UserID: req.UserID, + Username: req.Username, + Email: req.Email, + }) if err != nil { logger.ErrorF(c.Request.Context(), "List admin users failed: %v", err) response.AbortInternal(c, "获取用户列表失败") @@ -174,7 +179,13 @@ func GetUser(c *gin.Context) { return } - targetUser, err := getUserDetail(c.Request.Context(), id) + userSvc := getUserService(c.Request.Context()) + if userSvc == nil { + response.AbortInternal(c, "用户服务未就绪") + return + } + + targetUser, err := userSvc.AdminGetUser(c.Request.Context(), id) if abortUserLogicError(c, err, userNotFound, nil, nil) { return } @@ -215,7 +226,13 @@ func UpdateUserStatus(c *gin.Context) { return } - if err := updateUserStatus(c.Request.Context(), id, req.IsActive); err != nil { + userSvc := getUserService(c.Request.Context()) + if userSvc == nil { + response.AbortInternal(c, "用户服务未就绪") + return + } + + if err := userSvc.AdminUpdateUserStatus(c.Request.Context(), id, req.IsActive); err != nil { if abortUserLogicError(c, err, userNotFound, []string{cannotDisable}, nil) { return } @@ -251,7 +268,14 @@ func DeleteUser(c *gin.Context) { response.AbortUnauthorized(c, AdminRequired) return } - if err := deleteUser(c.Request.Context(), currUser.ID, id); err != nil { + + userSvc := getUserService(c.Request.Context()) + if userSvc == nil { + response.AbortInternal(c, "用户服务未就绪") + return + } + + if err := userSvc.AdminDeleteUser(c.Request.Context(), currUser.ID, id); err != nil { if abortUserLogicError(c, err, userNotFound, []string{cannotDelete, cannotDeleteSelf}, nil) { return } @@ -293,7 +317,20 @@ func CreateUser(c *gin.Context) { return } - newUser, err := createUser(c.Request.Context(), req) + userSvc := getUserService(c.Request.Context()) + if userSvc == nil { + response.AbortInternal(c, "用户服务未就绪") + return + } + + newUser, err := userSvc.AdminCreateUser(c.Request.Context(), contracts.AdminCreateUserRequest{ + Username: req.Username, + Password: req.Password, + Nickname: req.Nickname, + Email: req.Email, + IsActive: req.IsActive, + IsAdmin: req.IsAdmin, + }) if abortUserLogicError(c, err, "", nil, []string{usernameRequired, emailRequired, passwordTooShort, usernameExists, emailExists}) { return } @@ -342,7 +379,14 @@ func UpdateUser(c *gin.Context) { response.AbortUnauthorized(c, AdminRequired) return } - err := updateUser(c.Request.Context(), currUser.ID, updateUserParam{ + + userSvc := getUserService(c.Request.Context()) + if userSvc == nil { + response.AbortInternal(c, "用户服务未就绪") + return + } + + err := userSvc.AdminUpdateUser(c.Request.Context(), currUser.ID, contracts.AdminUpdateUserRequest{ ID: id, Nickname: req.Nickname, Email: req.Email, @@ -360,255 +404,3 @@ func UpdateUser(c *gin.Context) { c.JSON(http.StatusOK, response.OKNil()) } - -func listUsers(ctx context.Context, req listUsersRequest) (int64, []*contracts.UserDTO, error) { - query := db.DB(ctx).Table("w_users") - if req.UserID != nil { - query = query.Where("id = ?", *req.UserID) - } - if req.Username != "" { - query = query.Where("username LIKE ? ESCAPE '\\'", util.EscapeLike(req.Username)+"%") - } - if req.Email != "" { - query = query.Where("email LIKE ? ESCAPE '\\'", util.EscapeLike(req.Email)+"%") - } - - var total int64 - if err := query.Count(&total).Error; err != nil { - return 0, nil, err - } - - var users []*contracts.UserDTO - offset := (req.Page - 1) * req.PageSize - if err := query. - Select("id, username, nickname, email, avatar_url, is_active, is_admin, last_login_at, created_at, updated_at"). - Order("id ASC"). - Offset(offset). - Limit(req.PageSize). - Find(&users).Error; err != nil { - return 0, nil, err - } - return total, users, nil -} - -func getUserDetail(ctx context.Context, id uint64) (*contracts.UserDTO, error) { - var user contracts.UserDTO - if err := db.DB(ctx).Table("w_users"). - Select("id, username, nickname, email, avatar_url, is_active, is_admin, bio, phone, gender, website, location, last_login_at, created_at, updated_at"). - Where("id = ?", id). - First(&user).Error; err != nil { - return nil, err - } - return &user, nil -} - -func updateUserStatus(ctx context.Context, id uint64, active bool) error { - var flags struct { - ID uint64 - IsAdmin bool - } - if err := db.DB(ctx).Table("w_users").Select("id, is_admin").Where("id = ?", id).First(&flags).Error; err != nil { - return err - } - if !active && flags.IsAdmin { - return errors.New(cannotDisable) - } - - var tokenHashes []string - if !active { - _ = db.DB(ctx).Table("w_access_tokens").Where("user_id = ?", id).Pluck("token_hash", &tokenHashes).Error - } - - err := db.DB(ctx).Table("w_users").Where("id = ?", id).Update("is_active", active).Error - if err == nil { - auth.InvalidateCachedUser(ctx, id) - if !active { - for _, hash := range tokenHashes { - auth.InvalidateCachedToken(ctx, hash) - } - } - } - return err -} - -func deleteUser(ctx context.Context, currentUserID, targetID uint64) error { - if currentUserID == targetID { - return errors.New(cannotDeleteSelf) - } - var flags struct { - ID uint64 - IsAdmin bool - } - if err := db.DB(ctx).Table("w_users").Select("id, is_admin").Where("id = ?", targetID).First(&flags).Error; err != nil { - return err - } - if flags.IsAdmin { - return errors.New(cannotDelete) - } - - var tokenHashes []string - _ = db.DB(ctx).Table("w_access_tokens").Where("user_id = ?", targetID).Pluck("token_hash", &tokenHashes).Error - - err := db.DB(ctx).Transaction(func(tx *gorm.DB) error { - if err := tx.Table("w_access_tokens").Where("user_id = ?", targetID).Delete(map[string]any{}).Error; err != nil { - return err - } - if err := tx.Table("w_external_accounts").Where("user_id = ?", targetID).Delete(map[string]any{}).Error; err != nil { - return err - } - return tx.Table("w_users").Where("id = ?", targetID).Delete(map[string]any{}).Error - }) - if err == nil { - auth.InvalidateCachedUser(ctx, targetID) - for _, hash := range tokenHashes { - auth.InvalidateCachedToken(ctx, hash) - } - } - return err -} - -func createUser(ctx context.Context, req createUserRequest) (*contracts.UserDTO, error) { - req.Username = strings.TrimSpace(req.Username) - req.Nickname = strings.TrimSpace(req.Nickname) - req.Password = strings.TrimSpace(req.Password) - req.Email = strings.TrimSpace(req.Email) - - if req.Username == "" { - return nil, errors.New(usernameRequired) - } - if req.Email == "" { - return nil, errors.New(emailRequired) - } - if len(req.Password) < minPasswordLength { - return nil, errors.New(passwordTooShort) - } - - var count int64 - if err := db.DB(ctx).Table("w_users").Where("username = ?", req.Username).Count(&count).Error; err != nil { - return nil, err - } - if count > 0 { - return nil, errors.New(usernameExists) - } - - var emailCount int64 - if err := db.DB(ctx).Table("w_users").Where("email = ?", req.Email).Count(&emailCount).Error; err != nil { - return nil, err - } - if emailCount > 0 { - return nil, errors.New(emailExists) - } - - hash, err := util.HashPassword(req.Password) - if err != nil { - return nil, err - } - - if req.Nickname == "" { - req.Nickname = req.Username - } - - now := time.Now() - newUser := contracts.UserDTO{ - ID: idgen.NextUint64ID(), - Username: req.Username, - Nickname: req.Nickname, - Email: req.Email, - IsActive: req.IsActive, - IsAdmin: req.IsAdmin, - CreatedAt: now, - UpdatedAt: now, - } - - row := map[string]any{ - "id": newUser.ID, - "username": newUser.Username, - "password": hash, - "nickname": newUser.Nickname, - "email": newUser.Email, - "is_active": newUser.IsActive, - "is_admin": newUser.IsAdmin, - "created_at": now, - "updated_at": now, - } - if err := db.DB(ctx).Table("w_users").Create(row).Error; err != nil { - return nil, err - } - return &newUser, nil -} - -type updateUserParam struct { - ID uint64 - Nickname string - Email string - IsAdmin bool - Password string -} - -func updateUser(ctx context.Context, currentUserID uint64, param updateUserParam) error { - param.Nickname = strings.TrimSpace(param.Nickname) - param.Email = strings.TrimSpace(param.Email) - param.Password = strings.TrimSpace(param.Password) - - if param.Email == "" { - return errors.New(emailRequired) - } - - var targetUser contracts.UserDTO - if err := db.DB(ctx).Table("w_users").Where("id = ?", param.ID).First(&targetUser).Error; err != nil { - return err - } - - if currentUserID == param.ID && !param.IsAdmin && targetUser.IsAdmin { - return errors.New(cannotRevokeSelfAdmin) - } - - if targetUser.Email != param.Email { - var count int64 - if err := db.DB(ctx).Table("w_users").Where("email = ? AND id != ?", param.Email, param.ID).Count(&count).Error; err != nil { - return err - } - if count > 0 { - return errors.New(emailExists) - } - } - - if param.Password != "" && len(param.Password) < minPasswordLength { - return errors.New(passwordTooShort) - } - - needRevokeTokens := (param.Password != "") || (targetUser.IsAdmin && !param.IsAdmin) - var tokenHashes []string - if needRevokeTokens { - _ = db.DB(ctx).Table("w_access_tokens").Where("user_id = ?", param.ID).Pluck("token_hash", &tokenHashes).Error - } - - if param.Nickname == "" { - param.Nickname = targetUser.Username - } - - updates := map[string]any{ - "nickname": param.Nickname, - "email": param.Email, - "is_admin": param.IsAdmin, - "updated_at": time.Now(), - } - if param.Password != "" { - hash, err := util.HashPassword(param.Password) - if err != nil { - return err - } - updates["password"] = hash - } - - err := db.DB(ctx).Table("w_users").Where("id = ?", param.ID).Updates(updates).Error - if err == nil { - auth.InvalidateCachedUser(ctx, param.ID) - if needRevokeTokens { - for _, hash := range tokenHashes { - auth.InvalidateCachedToken(ctx, hash) - } - } - } - return err -} diff --git a/backend/plugins/domain/admin/plugin.go b/backend/plugins/domain/admin/plugin.go index 4e69b193..7b493dbd 100644 --- a/backend/plugins/domain/admin/plugin.go +++ b/backend/plugins/domain/admin/plugin.go @@ -50,18 +50,65 @@ func (p *Plugin) Manifest() core.Manifest { } } +var ( + globalUserSvc contracts.UserService + globalAuthSvc contracts.AuthService + globalCoreCtx *core.Context +) + +func getUserService(_ context.Context) contracts.UserService { + if globalUserSvc != nil { + return globalUserSvc + } + if globalCoreCtx != nil { + if svc, err := core.Inject[contracts.UserService](globalCoreCtx); err == nil { + globalUserSvc = svc + return svc + } + } + return nil +} + +func getAuthService(_ context.Context) contracts.AuthService { + if globalAuthSvc != nil { + return globalAuthSvc + } + if globalCoreCtx != nil { + if svc, err := core.Inject[contracts.AuthService](globalCoreCtx); err == nil { + globalAuthSvc = svc + return svc + } + } + return nil +} + // Apply registers admin routes, tasks, schedules, and settings into the Context. func (p *Plugin) Apply(ctx *core.Context) error { - // 0. Resolve auth service for middleware (via IoC, not direct import) + globalCoreCtx = ctx + + // 0. Resolve auth and user services reactively via IoC var loginMW gin.HandlerFunc = func(c *gin.Context) { c.Next() } var adminMW gin.HandlerFunc = func(c *gin.Context) { c.Next() } if authSvc, err := core.Inject[contracts.AuthService](ctx); err == nil && authSvc != nil { + globalAuthSvc = authSvc if mw, ok := authSvc.RequireAuthMiddleware().(gin.HandlerFunc); ok { loginMW = mw } if mw, ok := authSvc.RequireAdminMiddleware().(gin.HandlerFunc); ok { adminMW = mw } + } else { + core.When[contracts.AuthService](ctx, func(svc contracts.AuthService) { + globalAuthSvc = svc + }) + } + + if userSvc, err := core.Inject[contracts.UserService](ctx); err == nil && userSvc != nil { + globalUserSvc = userSvc + } else { + core.When[contracts.UserService](ctx, func(svc contracts.UserService) { + globalUserSvc = svc + }) } // 0a. Register migrations diff --git a/backend/plugins/domain/auth/plugin.go b/backend/plugins/domain/auth/plugin.go index 0a226346..a949eb25 100644 --- a/backend/plugins/domain/auth/plugin.go +++ b/backend/plugins/domain/auth/plugin.go @@ -5,6 +5,7 @@ package auth import ( + "context" "embed" "Wavelet/core" @@ -110,5 +111,16 @@ func (p *Plugin) Apply(ctx *core.Context) error { Category: "security", }) + // 5. Register Event Listeners for domain events + ctx.Events().On(contracts.EventTopicUserStatusChanged, func(c context.Context, e contracts.UserStatusChangedEvent) error { + InvalidateCachedUser(c, e.UserID) + return nil + }) + + ctx.Events().On(contracts.EventTopicUserDeleted, func(c context.Context, e contracts.UserDeletedEvent) error { + InvalidateCachedUser(c, e.TargetUserID) + return nil + }) + return nil } diff --git a/backend/plugins/domain/auth/service.go b/backend/plugins/domain/auth/service.go index 126834e8..b3c6cda8 100644 --- a/backend/plugins/domain/auth/service.go +++ b/backend/plugins/domain/auth/service.go @@ -110,6 +110,132 @@ func (s *authServiceImpl) DisallowTokenAuthMiddleware() any { return DisallowTokenAuth() } +func (s *authServiceImpl) InvalidateCachedUser(ctx context.Context, userID uint64) { + InvalidateCachedUser(ctx, userID) +} + +func (s *authServiceImpl) InvalidateCachedToken(ctx context.Context, tokenHash string) { + InvalidateCachedToken(ctx, tokenHash) +} + +func (s *authServiceImpl) ListAuthSources(ctx context.Context) ([]contracts.AuthSourceViewDTO, error) { + var sources []AuthSource + if err := db.DB(ctx).Order("id ASC").Find(&sources).Error; err != nil { + return nil, err + } + + views := make([]contracts.AuthSourceViewDTO, len(sources)) + for i := range sources { + views[i] = contracts.AuthSourceViewDTO{ + ID: sources[i].ID, + Name: sources[i].Name, + Type: sources[i].Type, + DisplayName: sources[i].DisplayName, + IsActive: sources[i].IsActive, + IconURL: sources[i].IconURL, + ClientSecretConfigured: sources[i].ClientSecret != "", + } + } + return views, nil +} + +func (s *authServiceImpl) CreateAuthSource(ctx context.Context, source contracts.AuthSourceDTO) (*contracts.AuthSourceDTO, error) { + model := AuthSource{ + ID: source.ID, + Name: source.Name, + Type: source.Type, + DisplayName: source.DisplayName, + ClientID: source.ClientID, + ClientSecret: source.ClientSecret, + OpenIDDiscoveryURL: source.OpenIDDiscoveryURL, + Scopes: source.Scopes, + IconURL: source.IconURL, + IsActive: source.IsActive, + } + + if err := model.Validate(); err != nil { + return nil, err + } + + if err := db.DB(ctx).Create(&model).Error; err != nil { + return nil, err + } + + model.Sanitize() + return toAuthSourceDTO(&model), nil +} + +func (s *authServiceImpl) UpdateAuthSource(ctx context.Context, id uint64, source contracts.AuthSourceDTO) (*contracts.AuthSourceDTO, error) { + var existing AuthSource + if err := db.DB(ctx).First(&existing, id).Error; err != nil { + return nil, err + } + + existing.DisplayName = source.DisplayName + existing.ClientID = source.ClientID + if source.ClientSecret != "" { + existing.ClientSecret = source.ClientSecret + } + existing.OpenIDDiscoveryURL = source.OpenIDDiscoveryURL + existing.Scopes = source.Scopes + existing.IconURL = source.IconURL + + if err := existing.Validate(); err != nil { + return nil, err + } + + if err := db.DB(ctx).Save(&existing).Error; err != nil { + return nil, err + } + + existing.Sanitize() + return toAuthSourceDTO(&existing), nil +} + +func (s *authServiceImpl) DeleteAuthSource(ctx context.Context, id uint64) error { + var existing AuthSource + if err := db.DB(ctx).First(&existing, id).Error; err != nil { + return err + } + + return db.DB(ctx).Delete(&existing).Error +} + +func (s *authServiceImpl) ToggleAuthSource(ctx context.Context, id uint64) (*contracts.AuthSourceDTO, error) { + var existing AuthSource + if err := db.DB(ctx).First(&existing, id).Error; err != nil { + return nil, err + } + + existing.IsActive = !existing.IsActive + if err := db.DB(ctx).Save(&existing).Error; err != nil { + return nil, err + } + + existing.Sanitize() + return toAuthSourceDTO(&existing), nil +} + +func toAuthSourceDTO(s *AuthSource) *contracts.AuthSourceDTO { + if s == nil { + return nil + } + return &contracts.AuthSourceDTO{ + ID: s.ID, + Name: s.Name, + Type: s.Type, + DisplayName: s.DisplayName, + ClientID: s.ClientID, + ClientSecret: s.ClientSecret, + OpenIDDiscoveryURL: s.OpenIDDiscoveryURL, + Scopes: s.Scopes, + IconURL: s.IconURL, + IsActive: s.IsActive, + CreatedAt: s.CreatedAt, + UpdatedAt: s.UpdatedAt, + } +} + type authRegistryImpl struct { mu sync.RWMutex providers map[string]contracts.OAuthProvider diff --git a/backend/plugins/domain/user/plugin.go b/backend/plugins/domain/user/plugin.go index 044abc12..11a49a10 100644 --- a/backend/plugins/domain/user/plugin.go +++ b/backend/plugins/domain/user/plugin.go @@ -81,7 +81,7 @@ func (p *Plugin) Apply(ctx *core.Context) error { // 2. Initialize and provide UserService if p.userSvc == nil { - p.userSvc = newUserService() + p.userSvc = newUserService(ctx.Events()) } core.Provide[contracts.UserService](ctx, p.userSvc) diff --git a/backend/plugins/domain/user/service.go b/backend/plugins/domain/user/service.go index 39a12188..d37ea440 100644 --- a/backend/plugins/domain/user/service.go +++ b/backend/plugins/domain/user/service.go @@ -11,13 +11,17 @@ import ( "strings" "time" + "Wavelet/core" "Wavelet/core/contracts" "Wavelet/pkg/idgen" database "Wavelet/plugins/infra/database" + "gorm.io/gorm" pkgu "Wavelet/pkg/util" ) +const columnUpdatedAt = "updated_at" + func toUserDTO(u *User) *contracts.UserDTO { if u == nil { return nil @@ -41,10 +45,16 @@ func toUserDTO(u *User) *contracts.UserDTO { } } -type userServiceImpl struct{} +type userServiceImpl struct { + events *core.EventBus +} -func newUserService() contracts.UserService { - return &userServiceImpl{} +func newUserService(events ...*core.EventBus) contracts.UserService { + var bus *core.EventBus + if len(events) > 0 { + bus = events[0] + } + return &userServiceImpl{events: bus} } func (s *userServiceImpl) GetUserByID(ctx context.Context, id uint64) (*contracts.UserDTO, error) { @@ -131,7 +141,7 @@ func (s *userServiceImpl) UpdateProfile(ctx context.Context, id uint64, req cont if req.Location != nil { updates["location"] = *req.Location } - updates["updated_at"] = time.Now() + updates[columnUpdatedAt] = time.Now() if err := database.DB(ctx).Model(&User{}).Where("id = ?", id).Updates(updates).Error; err != nil { return nil, err @@ -156,8 +166,8 @@ func (s *userServiceImpl) UpdatePassword(ctx context.Context, id uint64, oldPass return database.DB(ctx).Model(&User{}).Where("id = ?", id). Updates(map[string]any{ - "password": user.Password, - "updated_at": time.Now(), + "password": user.Password, + columnUpdatedAt: time.Now(), }).Error } @@ -174,7 +184,7 @@ func (s *userServiceImpl) UpdateLastLogin(ctx context.Context, id uint64, _ stri return database.DB(ctx).Model(&User{}).Where("id = ?", id). Updates(map[string]any{ "last_login_at": time.Now(), - "updated_at": time.Now(), + columnUpdatedAt: time.Now(), }).Error } @@ -280,3 +290,243 @@ func (s *userServiceImpl) UniqueUsername(ctx context.Context, base string) (stri return "", errors.New("failed to generate unique username") } + +func (s *userServiceImpl) AdminListUsers(ctx context.Context, filter contracts.AdminListUsersFilter) (int64, []*contracts.UserDTO, error) { + query := database.DB(ctx).Table("w_users") + if filter.UserID != nil { + query = query.Where("id = ?", *filter.UserID) + } + if filter.Username != "" { + query = query.Where("username LIKE ? ESCAPE '\\'", pkgu.EscapeLike(filter.Username)+"%") + } + if filter.Email != "" { + query = query.Where("email LIKE ? ESCAPE '\\'", pkgu.EscapeLike(filter.Email)+"%") + } + + var total int64 + if err := query.Count(&total).Error; err != nil { + return 0, nil, err + } + + if filter.Page <= 0 { + filter.Page = 1 + } + if filter.PageSize <= 0 { + filter.PageSize = 20 + } + + var users []*contracts.UserDTO + offset := (filter.Page - 1) * filter.PageSize + if err := query. + Select("id, username, nickname, email, avatar_url, is_active, is_admin, bio, phone, gender, website, location, last_login_at, created_at, updated_at"). + Order("id ASC"). + Offset(offset). + Limit(filter.PageSize). + Find(&users).Error; err != nil { + return 0, nil, err + } + return total, users, nil +} + +func (s *userServiceImpl) AdminGetUser(ctx context.Context, id uint64) (*contracts.UserDTO, error) { + var user contracts.UserDTO + if err := database.DB(ctx).Table("w_users"). + Select("id, username, nickname, email, avatar_url, is_active, is_admin, bio, phone, gender, website, location, last_login_at, created_at, updated_at"). + Where("id = ?", id). + First(&user).Error; err != nil { + return nil, err + } + return &user, nil +} + +func (s *userServiceImpl) AdminCreateUser(ctx context.Context, req contracts.AdminCreateUserRequest) (*contracts.UserDTO, error) { + req.Username = strings.TrimSpace(req.Username) + req.Nickname = strings.TrimSpace(req.Nickname) + req.Password = strings.TrimSpace(req.Password) + req.Email = strings.TrimSpace(req.Email) + + if req.Username == "" { + return nil, errors.New("用户名不能为空") + } + if req.Email == "" { + return nil, errors.New("邮箱不能为空") + } + const minPasswordLen = 8 + if len(req.Password) < minPasswordLen { + return nil, errors.New("密码长度至少为 8 位") + } + + var count int64 + if err := database.DB(ctx).Table("w_users").Where("username = ?", req.Username).Count(&count).Error; err != nil { + return nil, err + } + if count > 0 { + return nil, errors.New("用户名已被使用") + } + + var emailCount int64 + if err := database.DB(ctx).Table("w_users").Where("email = ?", req.Email).Count(&emailCount).Error; err != nil { + return nil, err + } + if emailCount > 0 { + return nil, errors.New("邮箱已被使用") + } + + hash, err := pkgu.HashPassword(req.Password) + if err != nil { + return nil, err + } + + if req.Nickname == "" { + req.Nickname = req.Username + } + + now := time.Now() + newUser := contracts.UserDTO{ + ID: idgen.NextUint64ID(), + Username: req.Username, + Nickname: req.Nickname, + Email: req.Email, + IsActive: req.IsActive, + IsAdmin: req.IsAdmin, + CreatedAt: now, + UpdatedAt: now, + } + + row := map[string]any{ + "id": newUser.ID, + "username": newUser.Username, + "password": hash, + "nickname": newUser.Nickname, + "email": newUser.Email, + "is_active": newUser.IsActive, + "is_admin": newUser.IsAdmin, + "created_at": now, + columnUpdatedAt: now, + } + if err := database.DB(ctx).Table("w_users").Create(row).Error; err != nil { + return nil, err + } + + if s.events != nil { + _ = s.events.Emit(ctx, contracts.EventTopicUserCreated, contracts.UserCreatedEvent{ + User: &newUser, + Password: req.Password, + }) + } + + return &newUser, nil +} + +func (s *userServiceImpl) AdminUpdateUser(ctx context.Context, currentUserID uint64, req contracts.AdminUpdateUserRequest) error { + req.Nickname = strings.TrimSpace(req.Nickname) + req.Email = strings.TrimSpace(req.Email) + req.Password = strings.TrimSpace(req.Password) + + if req.Email == "" { + return errors.New("邮箱不能为空") + } + + var targetUser contracts.UserDTO + if err := database.DB(ctx).Table("w_users").Where("id = ?", req.ID).First(&targetUser).Error; err != nil { + return err + } + + if currentUserID == req.ID && !req.IsAdmin && targetUser.IsAdmin { + return errors.New("不能取消自己的管理员权限") + } + + if targetUser.Email != req.Email { + var count int64 + if err := database.DB(ctx).Table("w_users").Where("email = ? AND id != ?", req.Email, req.ID).Count(&count).Error; err != nil { + return err + } + if count > 0 { + return errors.New("邮箱已被使用") + } + } + + const minPasswordLen = 8 + if req.Password != "" && len(req.Password) < minPasswordLen { + return errors.New("密码长度至少为 8 位") + } + + if req.Nickname == "" { + req.Nickname = targetUser.Username + } + + updates := map[string]any{ + "nickname": req.Nickname, + "email": req.Email, + "is_admin": req.IsAdmin, + columnUpdatedAt: time.Now(), + } + if req.Password != "" { + hash, err := pkgu.HashPassword(req.Password) + if err != nil { + return err + } + updates["password"] = hash + } + + err := database.DB(ctx).Table("w_users").Where("id = ?", req.ID).Updates(updates).Error + if err == nil && s.events != nil { + _ = s.events.Emit(ctx, contracts.EventTopicUserUpdated, &targetUser) + } + return err +} + +func (s *userServiceImpl) AdminUpdateUserStatus(ctx context.Context, id uint64, active bool) error { + var flags struct { + ID uint64 + IsAdmin bool + } + if err := database.DB(ctx).Table("w_users").Select("id, is_admin").Where("id = ?", id).First(&flags).Error; err != nil { + return err + } + if !active && flags.IsAdmin { + return errors.New("管理员账号无法被禁用") + } + + err := database.DB(ctx).Table("w_users").Where("id = ?", id).Update("is_active", active).Error + if err == nil && s.events != nil { + _ = s.events.Emit(ctx, contracts.EventTopicUserStatusChanged, contracts.UserStatusChangedEvent{ + UserID: id, + IsActive: active, + }) + } + return err +} + +func (s *userServiceImpl) AdminDeleteUser(ctx context.Context, currentUserID, targetID uint64) error { + if currentUserID == targetID { + return errors.New("不能删除当前登录用户") + } + var flags struct { + ID uint64 + IsAdmin bool + } + if err := database.DB(ctx).Table("w_users").Select("id, is_admin").Where("id = ?", targetID).First(&flags).Error; err != nil { + return err + } + if flags.IsAdmin { + return errors.New("管理员账号无法被删除") + } + + err := database.DB(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Table("w_access_tokens").Where("user_id = ?", targetID).Delete(map[string]any{}).Error; err != nil { + return err + } + if err := tx.Table("w_external_accounts").Where("user_id = ?", targetID).Delete(map[string]any{}).Error; err != nil { + return err + } + return tx.Table("w_users").Where("id = ?", targetID).Delete(map[string]any{}).Error + }) + if err == nil && s.events != nil { + _ = s.events.Emit(ctx, contracts.EventTopicUserDeleted, contracts.UserDeletedEvent{ + CurrentUserID: currentUserID, + TargetUserID: targetID, + }) + } + return err +} diff --git a/docs/superpowers/plans/2026-08-28-cordis-architecture-refactor.md b/docs/superpowers/plans/2026-08-28-cordis-architecture-refactor.md new file mode 100644 index 00000000..d12f519c --- /dev/null +++ b/docs/superpowers/plans/2026-08-28-cordis-architecture-refactor.md @@ -0,0 +1,171 @@ +# Cordis Architecture Refactor Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Refactor Wavelet backend to strictly conform to Cordis meta-framework principles: full revertible effects, space composability via per-plugin scoped context fork, 4-semantic typed event bus, elimination of package-level infra globals, and strict single-owner principle across domain plugins. + +**Architecture:** +1. Microkernel core (`core/`): Implement `Waterfall`, `Parallel`, `Serial` event dispatch, per-plugin Scoped Context (`ctx.Fork()`), revertible extension points (`extpoints`), and `ctx.DB()` / `ctx.Cache()` contract helpers. +2. Contracts (`core/contracts/`): Expand `UserService` & `AuthService` with administration and token revocation interfaces; define typed domain events. +3. Domain Plugins (`plugins/domain/`): Implement user/auth contract additions; refactor `admin` plugin to completely remove cross-plugin internal package imports and direct SQL operations on other plugins' tables. + +**Tech Stack:** Go 1.23+, GORM, Gin, Goose, Cordis Paradigm. + +## Global Constraints + +- No direct cross-package imports between plugins (`plugins/domain/A` must NEVER import `plugins/domain/B` or `plugins/drivers/*`). +- Single Owner Principle: Every database table is owned and operated exclusively by its owner plugin. +- Microkernel purity: `core/` and `core/contracts/` must never import `gin`, `gorm`, `asynq`. +- Tests must pass with `-race` enabled; temporary directories must use `t.TempDir()`. +- Quality gates: `make code-check`, `make format`, `make swagger`. + +--- + +### Task 1: Core EventBus 4 Dispatch Semantics + +**Files:** +- Modify: `backend/core/events.go` +- Test: `backend/core/events_test.go` + +**Interfaces:** +- Produces: + - `(b *EventBus) Emit(ctx context.Context, topic string, payload any) error` + - `(b *EventBus) Waterfall(ctx context.Context, topic string, initialPayload any) (any, error)` + - `(b *EventBus) Parallel(ctx context.Context, topic string, payload any) error` + - `(b *EventBus) Serial(ctx context.Context, topic string, payload any) error` + +- [ ] **Step 1: Write tests for Waterfall, Parallel, and Serial dispatch semantics** +- [ ] **Step 2: Run tests to verify they fail** +- [ ] **Step 3: Implement Waterfall, Parallel, Serial methods on EventBus** +- [ ] **Step 4: Run tests to verify they pass** +- [ ] **Step 5: Commit** + +--- + +### Task 2: Core Scoped Context, Revertible ExtPoints & Contract Helpers + +**Files:** +- Modify: `backend/core/context.go` +- Modify: `backend/core/app.go` +- Modify: `backend/core/extpoints/router.go` +- Modify: `backend/core/extpoints/task.go` +- Modify: `backend/core/extpoints/schedule.go` +- Modify: `backend/core/extpoints/setting.go` +- Modify: `backend/core/extpoints/migration.go` +- Test: `backend/core/context_test.go` +- Test: `backend/core/app_test.go` +- Test: `backend/core/extpoints/extpoints_test.go` + +**Interfaces:** +- Produces: + - `(c *Context) DB() contracts.DBService` + - `(c *Context) Cache() contracts.CacheService` + - `(r *RouterRegistry) Unregister(id uint64) bool` + - `RouterExtension.Handle(...) Disposer` / `RouteDefinition` with disposer tracking + - `App.ApplyPlugins()` forks scoped context per plugin: `p.Apply(a.ctx.Fork())` + +- [ ] **Step 1: Write tests for Scoped Context Fork, LIFO Disposer, and Router Unregister** +- [ ] **Step 2: Run tests to verify failure** +- [ ] **Step 3: Implement Scoped Fork, Disposers, and Context DB/Cache helpers** +- [ ] **Step 4: Update App.ApplyPlugins to fork a context for each plugin** +- [ ] **Step 5: Run tests and verify all core tests pass** +- [ ] **Step 6: Commit** + +--- + +### Task 3: Expand Service Contracts & Domain Events + +**Files:** +- Modify: `backend/core/contracts/user.go` +- Modify: `backend/core/contracts/auth.go` +- Modify: `backend/core/contracts/events.go` + +**Interfaces:** +- Produces: + - `AdminListUsersRequest`, `AdminCreateUserRequest`, `AdminUpdateUserRequest` + - `UserService` admin methods (`AdminListUsers`, `AdminGetUser`, `AdminCreateUser`, `AdminUpdateUser`, `AdminUpdateUserStatus`, `AdminDeleteUser`) + - `AuthService` token management methods (`RevokeToken`, `RevokeUserTokens`, `InvalidateCachedUser`, `InvalidateCachedToken`) + - Standard event definitions (`EventUserUpdated`, `EventUserDeleted`, `EventUserStatusChanged`, `EventTokenRevoked`) + +- [ ] **Step 1: Declare extended contracts and DTO types in core/contracts/** +- [ ] **Step 2: Declare typed event constants and structs in core/contracts/events.go** +- [ ] **Step 3: Verify core and core/contracts compile cleanly** +- [ ] **Step 4: Commit** + +--- + +### Task 4: Implement Expanded Contracts in User & Auth Domain Plugins + +**Files:** +- Modify: `backend/plugins/domain/user/service.go` +- Modify: `backend/plugins/domain/user/plugin.go` +- Modify: `backend/plugins/domain/user/user_test.go` +- Modify: `backend/plugins/domain/auth/service.go` +- Modify: `backend/plugins/domain/auth/plugin.go` +- Modify: `backend/plugins/domain/auth/plugin_test.go` + +**Interfaces:** +- Implements: `contracts.UserService` full methods in `user` plugin. +- Implements: `contracts.AuthService` full methods in `auth` plugin. +- Subscribes: `auth` plugin subscribes to `EventUserStatusChanged` / `EventUserDeleted` to invalidate cache and revoke tokens. + +- [ ] **Step 1: Write unit tests for new UserService admin methods and AuthService revocation methods** +- [ ] **Step 2: Run tests to verify failure** +- [ ] **Step 3: Implement the methods in user and auth domain packages** +- [ ] **Step 4: Run user and auth plugin tests and verify they pass** +- [ ] **Step 5: Commit** + +--- + +### Task 5: Refactor Admin Plugin (Eliminate Cross-Plugin Direct Imports & Table Ownership Violations) + +**Files:** +- Modify: `backend/plugins/domain/admin/handlers_user.go` +- Modify: `backend/plugins/domain/admin/handlers_auth_source.go` +- Modify: `backend/plugins/domain/admin/handlers_config.go` +- Modify: `backend/plugins/domain/admin/handlers_logs.go` +- Modify: `backend/plugins/domain/admin/handlers_status.go` +- Modify: `backend/plugins/domain/admin/handlers_tasks.go` +- Modify: `backend/plugins/domain/admin/repository.go` +- Modify: `backend/plugins/domain/admin/system_config_cache.go` +- Modify: `backend/plugins/domain/admin/plugin.go` +- Modify: `backend/plugins/domain/admin/plugin_test.go` + +**Interfaces:** +- Consumes: `contracts.UserService`, `contracts.AuthService`, `contracts.DBService`, `contracts.CacheService`, `ctx.DB()`, `ctx.Cache()` +- Zero imports of `plugins/domain/auth`, `plugins/domain/risk_control`, `plugins/domain/cap`, `plugins/drivers/*`, `plugins/infra/database` + +- [ ] **Step 1: Write integration tests for Admin handlers using mocked/injected contracts** +- [ ] **Step 2: Refactor admin handlers to delegate user/auth operations to contracts** +- [ ] **Step 3: Remove all cross-plugin direct package imports and illegal SQL DML** +- [ ] **Step 4: Run admin plugin tests to verify passing** +- [ ] **Step 5: Commit** + +--- + +### Task 6: Clean up Domain & Infra Plugins Database / Cache Injections + +**Files:** +- Modify: `backend/plugins/domain/cap/...` +- Modify: `backend/plugins/domain/message_gateway/...` +- Modify: `backend/plugins/domain/risk_control/...` +- Modify: `backend/plugins/domain/upload/...` +- Modify: `backend/plugins/domain/system/...` + +- [ ] **Step 1: Audit and replace direct `database.DB(ctx)` calls with `ctx.DB()` / injected `contracts.DBService`** +- [ ] **Step 2: Audit and replace direct `cache.Client()` calls with `ctx.Cache()` / injected `contracts.CacheService`** +- [ ] **Step 3: Run domain plugins test suite** +- [ ] **Step 4: Commit** + +--- + +### Task 7: Full Verification & Quality Gates + +**Files:** +- All backend files + +- [ ] **Step 1: Run full test suite with race detector: `go test -v -race ./backend/...`** +- [ ] **Step 2: Run `make code-check`** +- [ ] **Step 3: Run `make format`** +- [ ] **Step 4: Run `make swagger`** +- [ ] **Step 5: Final commit** diff --git a/docs/superpowers/specs/2026-08-28-cordis-architecture-refactor-design.md b/docs/superpowers/specs/2026-08-28-cordis-architecture-refactor-design.md new file mode 100644 index 00000000..cb30f20c --- /dev/null +++ b/docs/superpowers/specs/2026-08-28-cordis-architecture-refactor-design.md @@ -0,0 +1,140 @@ +# Cordis 架构对齐与系统重构设计规范 (Design Spec) + +- **Date:** 2026-08-28 +- **Topic:** Cordis Architecture Alignment & Full System Refactor +- **Status:** Approved + +--- + +## 1. 目标与背景 (Goal & Background) + +本项目遵循 **Cordis（时空可组合性元框架）** 的核心设计哲学： +- **时间可组合性（Time Composability）**：所有对运行时环境的修改（扩展点挂载、事件订阅、服务注册、状态配置）均具备显式可逆操作（Revertible Effects），卸载时按 LIFO（后进先出）严格回收。 +- **空间可组合性（Space Composability）**：插件运行在独立的 Scoped Context 分支中，通过面向契约（`contracts`）与事件（`EventBus`）解耦，消除人工硬编码启动顺序与跨插件内部实现耦合。 +- **表单一所有者原则（Single Owner Principle）**：每张数据表由且仅由一个所有者插件维护，严禁旁路 DML 读写。 + +本规范定义微内核层、服务契约层、基础设施层与业务域插件的全量重构设计。 + +--- + +## 2. 系统架构与分层设计 (Architecture & Layers) + +``` +┌────────────────────────────────────────────────────────────────────────┐ +│ Micro-Kernel (core/) │ +│ Context Bus (Scoped Fork & LIFO Disposer) | Container | EventBus │ +│ Extpoints (Router, Tasks, Schedules, Settings, Migrations) │ +├────────────────────────────────────────────────────────────────────────┤ +│ Service Contracts (contracts/) │ +│ DBService, CacheService, StorageService, UserService, AuthService... │ +├─────────────────────────┬──────────────────────────────────────────────┤ +│ Runtime Drivers │ Platform Infra Plugins │ +│ (plugins/drivers/) │ (plugins/infra/) │ +│ - driver_http (Gin) │ - database (contracts.DBService) │ +│ - driver_asynq_worker │ - cache (contracts.CacheService) │ +│ - driver_asynq_cron │ - storage, logger │ +├─────────────────────────┴──────────────────────────────────────────────┤ +│ Self-Contained Domain Plugins (plugins/domain/) │ +│ - user, auth, admin, cap, message_gateway, risk_control, system, upload│ +└────────────────────────────────────────────────────────────────────────┘ +``` + +--- + +## 3. 详细设计与核心组件规范 (Detailed Specifications) + +### 3.1 微内核层 (`backend/core/`) + +#### 1. 事件总线四大分发语义 (`core/events.go`) +- `Emit(ctx context.Context, topic string, payload any) error`：通知型广播，不短路，收集所有 Handler 产生的 error（`errors.Join`）。 +- `Waterfall(ctx context.Context, topic string, initialPayload any) (any, error)`：链式流水线改写，上一个 handler 的返回值作为下一个 handler 的输入；一旦 handler 返回 error 立即短路中断并返回。 +- `Parallel(ctx context.Context, topic string, payload any) error`：并发扇出执行所有 handler，通过 goroutine + WaitGroup 并发执行，收集所有 error。 +- `Serial(ctx context.Context, topic string, payload any) error`：严格按序流水线执行所有 handler，遇到第一个 error 立即短路中断。 + +#### 2. 插件作用域上下文 (`core/context.go` & `core/app.go`) +- `App.ApplyPlugins()` 为每个插件生成专属的 `pluginCtx := app.ctx.Fork()`，并在 `Apply(pluginCtx)` 中挂载。 +- Scoped Context 拥有独立的 `disposers`、`values` 与子 container，当插件被卸载或 context 被 dispose 时，仅回收该插件范围内的资源。 +- 在 `Context` 上扩展 `ctx.DB()` 与 `ctx.Cache()` 辅助方法，内部通过 `core.Inject[contracts.DBService](c)` 与 `core.Inject[contracts.CacheService](c)` 解析。 + +#### 3. 扩展点可逆化与注销 (`core/extpoints/`) +- `RouterExtension`：注册路由时返回 `Disposer`；`RouterRegistry` 内部维护带 ID 的路由列表，支持动态移除路由。 +- `TaskExtension` / `ScheduleExtension` / `SettingExtension` / `MigrationExtension`：提供与 Scoped Context 关联的注销机制与 Disposer 回收。 + +--- + +### 3.2 服务契约层 (`backend/core/contracts/`) + +#### 1. `contracts.UserService` 扩展 +收拢所有用户管理操作： +```go +type UserService interface { + GetByID(ctx context.Context, id uint64) (*UserDTO, error) + GetByUsername(ctx context.Context, username string) (*UserDTO, error) + GetByEmail(ctx context.Context, email string) (*UserDTO, error) + Create(ctx context.Context, user *UserDTO, password string) (*UserDTO, error) + Update(ctx context.Context, user *UserDTO) error + Delete(ctx context.Context, id uint64) error + // Admin 扩展方法 + AdminListUsers(ctx context.Context, req AdminListUsersRequest) (int64, []*UserDTO, error) + AdminGetUser(ctx context.Context, id uint64) (*UserDTO, error) + AdminCreateUser(ctx context.Context, req AdminCreateUserRequest) (*UserDTO, error) + AdminUpdateUser(ctx context.Context, currentUserID uint64, req AdminUpdateUserRequest) error + AdminUpdateUserStatus(ctx context.Context, id uint64, active bool) error + AdminDeleteUser(ctx context.Context, currentUserID, targetID uint64) error +} +``` + +#### 2. `contracts.AuthService` 扩展 +收拢令牌管理与认证源查询： +```go +type AuthService interface { + Authenticate(ctx context.Context, username, password string) (*UserDTO, error) + GenerateToken(ctx context.Context, userID uint64, opts ...TokenOption) (string, error) + ValidateToken(ctx context.Context, token string) (*TokenClaims, error) + RevokeToken(ctx context.Context, tokenHash string) error + RevokeUserTokens(ctx context.Context, userID uint64) error + InvalidateCachedUser(ctx context.Context, userID uint64) + InvalidateCachedToken(ctx context.Context, tokenHash string) +} +``` + +#### 3. 强类型领域事件 (`core/contracts/events.go`) +- `EventUserUpdated`: `{ UserID, UpdatedFields }` +- `EventUserDeleted`: `{ CurrentUserID, TargetUserID }` +- `EventUserStatusChanged`: `{ UserID, IsActive }` +- `EventTokenRevoked`: `{ UserID, TokenHash }` + +--- + +### 3.3 基础设施去全局化 (`backend/plugins/infra/`) + +1. **`plugins/infra/database`**： + - 彻底去全局化：弃用全局静态变量直读，统一提供 `contracts.DBService` 实例并在 `Apply` 中 `core.Provide[contracts.DBService](ctx, svc)`。 +2. **`plugins/infra/cache`**： + - 弃用全局 `cache.Client()` 包级直连，统一通过 `contracts.CacheService` 接口与 `ctx.Cache()` 操作。 + +--- + +### 3.4 业务域插件边界治理 (`backend/plugins/domain/`) + +1. **`domain/admin` 治理**： + - 移除全部跨插件直接导入（`domain/auth`、`domain/risk_control`、`domain/cap`、`drivers/driver_asynq_*`、`infra/database`）。 + - 用户管理委派给 `contracts.UserService`。 + - 认证与令牌操作委派给 `contracts.AuthService`。 + - 配置管理通过 `ctx.Settings()` / `contracts.SettingService`。 +2. **表单一所有者原则（Single Owner Principle）**： + - `w_users` 表有且仅由 `domain/user` 插件读写。 + - `w_access_tokens` / `w_auth_sources` / `w_external_accounts` 表有且仅由 `domain/auth` 插件读写。 + - `w_system_configs` 表由 `domain/admin` 维护。 + +--- + +## 4. 实施与验证流程 (Verification & Quality Gates) + +1. **内核与契约单测**： + - `core/events_test.go`：覆盖 `Emit`、`Waterfall`、`Parallel`、`Serial`。 + - `core/context_test.go`：覆盖 Scoped Fork、Disposer LIFO、扩展点注销。 +2. **全局代码检查与格式化**： + - `make code-check` + - `make format` + - `go test -v -race ./backend/...`