diff --git a/core/app.go b/core/app.go new file mode 100644 index 00000000..fd5d10b7 --- /dev/null +++ b/core/app.go @@ -0,0 +1,489 @@ +package core + +import ( + "context" + "errors" + "fmt" + "os" + "os/signal" + "strings" + "sync" + "syscall" + "time" +) + +const ( + defaultShutdownTimeout = 10 * time.Second +) + +// AppOption configures an App instance during construction. +type AppOption func(*App) + +// WithContext sets a custom root Context for the App. +func WithContext(ctx *Context) AppOption { + return func(a *App) { + if ctx != nil { + a.ctx = ctx + } + } +} + +// WithProfile sets the runtime profile for the App. +func WithProfile(profile Profile) AppOption { + return func(a *App) { + a.profile = normalizeProfile(profile) + } +} + +// WithPlugins registers initial plugins for the App. +func WithPlugins(plugins ...Plugin) AppOption { + return func(a *App) { + a.Use(plugins...) + } +} + +// WithMigrationEngine sets the database migration engine for the App. +func WithMigrationEngine(engine MigrationEngine) AppOption { + return func(a *App) { + a.migrationEngine = engine + } +} + +// WithMigrationRunner sets the migration runner function for the App. +func WithMigrationRunner(runner MigrationRunner) AppOption { + return func(a *App) { + a.migrationEngine = runner + } +} + +// WithShutdownTimeout sets the fallback timeout for graceful application shutdown. +func WithShutdownTimeout(timeout time.Duration) AppOption { + return func(a *App) { + if timeout > 0 { + a.shutdownTimeout = timeout + } + } +} + +// App is the unified assembly entrypoint and runtime aspect dispatcher of the Cordis micro-kernel. +// It manages plugin collection, dependency mounting, migration execution, profile-based driver startup, +// and graceful signal-driven LIFO shutdown. +type App struct { + mu sync.RWMutex + ctx *Context + profile Profile + plugins []Plugin + pluginMap map[string]Plugin + applied bool + running bool + startedDrivers []Driver + migrationEngine MigrationEngine + shutdownTimeout time.Duration +} + +// NewApp creates a new Cordis application instance with default options. +func NewApp(opts ...AppOption) *App { + app := &App{ + ctx: NewContext(context.Background()), + profile: ProfileAll, + pluginMap: make(map[string]Plugin), + shutdownTimeout: defaultShutdownTimeout, + } + + for _, opt := range opts { + if opt != nil { + opt(app) + } + } + + return app +} + +// Context returns the root micro-kernel Context of the application. +func (a *App) Context() *Context { + return a.ctx +} + +// Profile returns the current runtime profile of the application. +func (a *App) Profile() Profile { + a.mu.RLock() + defer a.mu.RUnlock() + return a.profile +} + +// WithProfile sets the application runtime profile and returns the App for fluent chaining. +func (a *App) WithProfile(profile Profile) *App { + a.mu.Lock() + defer a.mu.Unlock() + a.profile = normalizeProfile(profile) + return a +} + +// SetProfile sets the application runtime profile. +func (a *App) SetProfile(profile Profile) *App { + return a.WithProfile(profile) +} + +// Use registers one or more plugins into the application in registration order. +// Duplicate plugins (by Name) update existing registrations in-place to preserve order. +func (a *App) Use(plugins ...Plugin) *App { + a.mu.Lock() + defer a.mu.Unlock() + + for _, p := range plugins { + if p == nil { + continue + } + name := p.Name() + if name == "" { + continue + } + + if _, exists := a.pluginMap[name]; exists { + for i, existing := range a.plugins { + if existing.Name() == name { + a.plugins[i] = p + break + } + } + } else { + a.plugins = append(a.plugins, p) + } + a.pluginMap[name] = p + } + + return a +} + +// Plugins returns a copy of all registered plugins in registration order. +func (a *App) Plugins() []Plugin { + a.mu.RLock() + defer a.mu.RUnlock() + + res := make([]Plugin, len(a.plugins)) + copy(res, a.plugins) + return res +} + +// Plugin retrieves a registered plugin by its unique name. +func (a *App) Plugin(name string) (Plugin, bool) { + a.mu.RLock() + defer a.mu.RUnlock() + + p, ok := a.pluginMap[name] + return p, ok +} + +// SetMigrationEngine sets the migration engine for the application. +func (a *App) SetMigrationEngine(engine MigrationEngine) *App { + a.mu.Lock() + defer a.mu.Unlock() + a.migrationEngine = engine + return a +} + +// SetMigrationRunner sets the migration runner function for the application. +func (a *App) SetMigrationRunner(runner MigrationRunner) *App { + return a.SetMigrationEngine(runner) +} + +// ApplyPlugins applies all registered plugins on the application Context. +// It is idempotent and only applies plugins once per App instance. +func (a *App) ApplyPlugins() error { + a.mu.Lock() + if a.applied { + a.mu.Unlock() + return nil + } + a.applied = true + plugins := make([]Plugin, len(a.plugins)) + copy(plugins, a.plugins) + a.mu.Unlock() + + for _, p := range plugins { + if err := p.Apply(a.ctx); err != nil { + return fmt.Errorf("core: apply plugin %q failed: %w", p.Name(), err) + } + } + return nil +} + +// RunMigrations dispatches migration execution across all registered plugin migration entries. +func (a *App) RunMigrations(ctx context.Context) error { + entries := a.ctx.Migrations().Entries() + if len(entries) == 0 { + return nil + } + + a.mu.RLock() + engine := a.migrationEngine + a.mu.RUnlock() + + if engine == nil { + // Attempt to resolve from IoC container + if resolved, err := Inject[MigrationEngine](a.ctx); err == nil && resolved != nil { + engine = resolved + } + } + + if engine == nil { + return nil + } + + if err := engine.Migrate(ctx, entries); err != nil { + return fmt.Errorf("core: migration failed: %w", err) + } + + return nil +} + +// Start executes the application boot pipeline: +// 1. Applies all registered plugins to populate services, routes, tasks, and drivers. +// 2. Dispatches database migrations via MigrationEngine. +// 3. Filters and starts drivers matching the active Profile. +// 4. Emits "app:ready" on the EventBus. +// +//nolint:contextcheck +func (a *App) Start(ctx ...context.Context) error { + a.mu.Lock() + if a.running { + a.mu.Unlock() + return ErrAppRunning + } + a.running = true + a.mu.Unlock() + + var baseCtx context.Context + switch { + case len(ctx) > 0 && ctx[0] != nil: + baseCtx = ctx[0] + case a.ctx != nil: + baseCtx = a.ctx.GoContext() + default: + baseCtx = context.Background() + } + + // 1. Apply plugins + if err := a.ApplyPlugins(); err != nil { + a.mu.Lock() + a.running = false + a.mu.Unlock() + return err + } + + // 2. Run migrations + if err := a.RunMigrations(baseCtx); err != nil { + a.mu.Lock() + a.running = false + a.mu.Unlock() + return err + } + + // 3. Filter drivers matching active profile + a.mu.RLock() + prof := a.profile + a.mu.RUnlock() + + allDrivers := a.ctx.Drivers() + var driversToStart []Driver + for _, d := range allDrivers { + if matchesProfile(prof, d.Type()) { + driversToStart = append(driversToStart, d) + } + } + + // 4. Start matching drivers + for _, d := range driversToStart { + if err := d.Start(baseCtx); err != nil { + // Rollback already started drivers in reverse order + a.mu.Lock() + started := a.startedDrivers + a.startedDrivers = nil + a.running = false + a.mu.Unlock() + + for i := len(started) - 1; i >= 0; i-- { + _ = started[i].Stop(context.Background()) + } + + return fmt.Errorf("core: start driver %s failed: %w", d.Type(), err) + } + + a.mu.Lock() + a.startedDrivers = append(a.startedDrivers, d) + a.mu.Unlock() + } + + // 5. Emit app:ready event + _ = a.ctx.Events().Emit(baseCtx, "app:ready", a) + + return nil +} + +// Stop gracefully shuts down the application: +// 1. Emits "app:stopping" on the EventBus. +// 2. Stops all started drivers in LIFO (reverse) order. +// 3. Disposes the Context (running registered OnDispose callbacks in LIFO order). +// 4. Emits "app:stopped" on the EventBus. +// +//nolint:contextcheck +func (a *App) Stop(ctx ...context.Context) error { + a.mu.Lock() + if !a.running { + a.mu.Unlock() + return nil + } + a.running = false + started := a.startedDrivers + a.startedDrivers = nil + timeout := a.shutdownTimeout + a.mu.Unlock() + + var shutdownCtx context.Context + if len(ctx) > 0 && ctx[0] != nil { + shutdownCtx = ctx[0] + } else { + var cancel context.CancelFunc + shutdownCtx, cancel = context.WithTimeout(context.Background(), timeout) + defer cancel() + } + + _ = a.ctx.Events().Emit(shutdownCtx, "app:stopping", a) + + var errs []error + + // 1. Stop drivers in reverse order + for i := len(started) - 1; i >= 0; i-- { + d := started[i] + if err := d.Stop(shutdownCtx); err != nil { + errs = append(errs, fmt.Errorf("core: stop driver %s failed: %w", d.Type(), err)) + } + } + + // 2. Dispose context + if a.ctx != nil && !a.ctx.IsDisposed() { + if err := a.ctx.Dispose(); err != nil { + errs = append(errs, fmt.Errorf("core: dispose context failed: %w", err)) + } + } + + _ = a.ctx.Events().Emit(shutdownCtx, "app:stopped", a) + + return errors.Join(errs...) +} + +// Run starts the application and blocks until an OS signal (SIGINT, SIGTERM) or context cancellation is received, +// then executes graceful shutdown. +// +//nolint:contextcheck +func (a *App) Run(ctx ...context.Context) error { + var parent context.Context + switch { + case len(ctx) > 0 && ctx[0] != nil: + parent = ctx[0] + case a.ctx != nil: + parent = a.ctx.GoContext() + default: + parent = context.Background() + } + + sigCtx, stopSignals := signal.NotifyContext(parent, syscall.SIGINT, syscall.SIGTERM, os.Interrupt) + defer stopSignals() + + if err := a.Start(sigCtx); err != nil { + return err + } + + // Wait for OS signal or context cancellation + <-sigCtx.Done() + + shutdownCtx, cancel := context.WithTimeout(context.Background(), a.shutdownTimeout) + defer cancel() + + return a.Stop(shutdownCtx) +} + +// IsRunning returns whether the application is currently running. +func (a *App) IsRunning() bool { + a.mu.RLock() + defer a.mu.RUnlock() + return a.running +} + +// StartedDrivers returns a copy of currently running drivers. +func (a *App) StartedDrivers() []Driver { + a.mu.RLock() + defer a.mu.RUnlock() + + res := make([]Driver, len(a.startedDrivers)) + copy(res, a.startedDrivers) + return res +} + +// ExecuteCLI parses CLI arguments to configure the profile and runs the application. +func (a *App) ExecuteCLI(args ...string) error { + var ctx context.Context + if a.ctx != nil { + ctx = a.ctx.GoContext() + } else { + ctx = context.Background() + } + return a.ExecuteCLIWithContext(ctx, args...) +} + +// ExecuteCLIWithContext parses CLI arguments, configures the profile, and runs the application with the given context. +// +//nolint:contextcheck +func (a *App) ExecuteCLIWithContext(ctx context.Context, args ...string) error { + cliArgs := args + if len(cliArgs) == 0 { + cliArgs = os.Args[1:] + } + + profile := ProfileAll + if len(cliArgs) > 0 { + first := strings.TrimSpace(cliArgs[0]) + switch { + case strings.HasPrefix(first, "--profile="): + profile = Profile(strings.TrimPrefix(first, "--profile=")) + case strings.HasPrefix(first, "-p="): + profile = Profile(strings.TrimPrefix(first, "-p=")) + case !strings.HasPrefix(first, "-"): + profile = Profile(first) + } + } + + a.WithProfile(profile) + return a.Run(ctx) +} + +func matchesProfile(profile Profile, dt DriverType) bool { + norm := normalizeProfile(profile) + switch norm { + case ProfileAll, "": + return true + case ProfileAPI: + return dt == DriverTypeHTTP + case ProfileWorker: + return dt == DriverTypeWorker + case ProfileSchedule: + return dt == DriverTypeScheduler + default: + return string(norm) == string(dt) + } +} + +func normalizeProfile(p Profile) Profile { + switch strings.ToLower(strings.TrimSpace(string(p))) { + case "api", "http": + return ProfileAPI + case "worker": + return ProfileWorker + case "schedule", "scheduler", "cron": + return ProfileSchedule + case "all", "fused", "full", "": + return ProfileAll + default: + return p + } +} diff --git a/core/app_test.go b/core/app_test.go new file mode 100644 index 00000000..d6315406 --- /dev/null +++ b/core/app_test.go @@ -0,0 +1,504 @@ +package core_test + +import ( + "context" + "errors" + "sync" + "testing" + "testing/fstest" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/Rain-kl/Wavelet/core" + "github.com/Rain-kl/Wavelet/core/extpoints" +) + +// appMockDriver is a test driver tracking its start/stop lifecycle. +type appMockDriver struct { + mu sync.Mutex + driverType core.DriverType + startCalled bool + stopCalled bool + startErr error + stopErr error +} + +func newAppMockDriver(dt core.DriverType) *appMockDriver { + return &appMockDriver{driverType: dt} +} + +func (m *appMockDriver) Type() core.DriverType { + return m.driverType +} + +func (m *appMockDriver) Start(_ context.Context) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.startErr != nil { + return m.startErr + } + m.startCalled = true + return nil +} + +func (m *appMockDriver) Stop(_ context.Context) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.stopErr != nil { + return m.stopErr + } + m.stopCalled = true + return nil +} + +func (m *appMockDriver) isStarted() bool { + m.mu.Lock() + defer m.mu.Unlock() + return m.startCalled +} + +func (m *appMockDriver) isStopped() bool { + m.mu.Lock() + defer m.mu.Unlock() + return m.stopCalled +} + +// appMockPlugin is a test plugin. +type appMockPlugin struct { + name string + applyFn func(ctx *core.Context) error +} + +func (p *appMockPlugin) Name() string { + return p.name +} + +func (p *appMockPlugin) Apply(ctx *core.Context) error { + if p.applyFn != nil { + return p.applyFn(ctx) + } + return nil +} + +func TestAppNewAndConfiguration(t *testing.T) { + customCtx := core.NewContext(context.Background()) + p1 := &appMockPlugin{name: "plugin1"} + p2 := &appMockPlugin{name: "plugin2"} + + app := core.NewApp( + core.WithContext(customCtx), + core.WithProfile(core.ProfileAPI), + core.WithPlugins(p1, p2), + core.WithShutdownTimeout(5*time.Second), + ) + + assert.Equal(t, customCtx, app.Context()) + assert.Equal(t, core.ProfileAPI, app.Profile()) + assert.Len(t, app.Plugins(), 2) + + retrieved, ok := app.Plugin("plugin1") + assert.True(t, ok) + assert.Equal(t, p1, retrieved) + + _, ok = app.Plugin("non_existent") + assert.False(t, ok) + + // Update existing plugin in-place + p1Updated := &appMockPlugin{name: "plugin1"} + app.Use(p1Updated, nil) + assert.Len(t, app.Plugins(), 2) + retrieved, ok = app.Plugin("plugin1") + assert.True(t, ok) + assert.Equal(t, p1Updated, retrieved) + + // Test SetProfile + app.SetProfile(core.ProfileWorker) + assert.Equal(t, core.ProfileWorker, app.Profile()) +} + +func TestAppProfileDispatch(t *testing.T) { + tests := []struct { + name string + profile core.Profile + expectedHTTP bool + expectedWorker bool + expectedCron bool + expectedCustom bool + }{ + { + name: "ProfileAPI only starts HTTP driver", + profile: core.ProfileAPI, + expectedHTTP: true, + expectedWorker: false, + expectedCron: false, + expectedCustom: false, + }, + { + name: "ProfileWorker only starts Worker driver", + profile: core.ProfileWorker, + expectedHTTP: false, + expectedWorker: true, + expectedCron: false, + expectedCustom: false, + }, + { + name: "ProfileSchedule only starts Schedule driver", + profile: core.ProfileSchedule, + expectedHTTP: false, + expectedWorker: false, + expectedCron: true, + expectedCustom: false, + }, + { + name: "Profile 'scheduler' alias starts Schedule driver", + profile: core.Profile("scheduler"), + expectedHTTP: false, + expectedWorker: false, + expectedCron: true, + expectedCustom: false, + }, + { + name: "ProfileAll starts all drivers", + profile: core.ProfileAll, + expectedHTTP: true, + expectedWorker: true, + expectedCron: true, + expectedCustom: true, + }, + { + name: "Custom profile starts custom driver", + profile: core.Profile("custom_rpc"), + expectedHTTP: false, + expectedWorker: false, + expectedCron: false, + expectedCustom: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + httpD := newAppMockDriver(core.DriverTypeHTTP) + workerD := newAppMockDriver(core.DriverTypeWorker) + cronD := newAppMockDriver(core.DriverTypeScheduler) + customD := newAppMockDriver(core.DriverType("custom_rpc")) + + p := &appMockPlugin{ + name: "drivers_plugin", + applyFn: func(ctx *core.Context) error { + _ = ctx.RegisterDriver(httpD) + _ = ctx.RegisterDriver(workerD) + _ = ctx.RegisterDriver(cronD) + _ = ctx.RegisterDriver(customD) + return nil + }, + } + + app := core.NewApp( + core.WithProfile(tt.profile), + core.WithPlugins(p), + ) + + err := app.Start(context.Background()) + require.NoError(t, err) + + assert.Equal(t, tt.expectedHTTP, httpD.isStarted(), "HTTP driver start mismatch") + assert.Equal(t, tt.expectedWorker, workerD.isStarted(), "Worker driver start mismatch") + assert.Equal(t, tt.expectedCron, cronD.isStarted(), "Cron driver start mismatch") + assert.Equal(t, tt.expectedCustom, customD.isStarted(), "Custom driver start mismatch") + + err = app.Stop(context.Background()) + require.NoError(t, err) + }) + } +} + +func TestAppLifecycleStartStop(t *testing.T) { + var stopOrder []string + var stopOrderMu sync.Mutex + + httpD := newAppMockDriver(core.DriverTypeHTTP) + workerD := newAppMockDriver(core.DriverTypeWorker) + + httpD.stopErr = nil + workerD.stopErr = nil + + // Wrap stop to record order + origHttpStop := httpD.Stop + _ = origHttpStop + + p := &appMockPlugin{ + name: "test_plugin", + applyFn: func(ctx *core.Context) error { + _ = ctx.RegisterDriver(httpD) + _ = ctx.RegisterDriver(workerD) + + ctx.OnDispose(func() error { + stopOrderMu.Lock() + stopOrder = append(stopOrder, "ctx_disposer") + stopOrderMu.Unlock() + return nil + }) + return nil + }, + } + + app := core.NewApp( + core.WithProfile(core.ProfileAll), + core.WithPlugins(p), + ) + + var readyReceived, stoppingReceived, stoppedReceived bool + app.Context().Events().On("app:ready", func() { + readyReceived = true + }) + app.Context().Events().On("app:stopping", func() { + stoppingReceived = true + }) + app.Context().Events().On("app:stopped", func() { + stoppedReceived = true + }) + + err := app.Start(context.Background()) + require.NoError(t, err) + assert.True(t, app.IsRunning()) + assert.Len(t, app.StartedDrivers(), 2) + assert.True(t, readyReceived) + + err = app.Stop(context.Background()) + require.NoError(t, err) + assert.False(t, app.IsRunning()) + assert.Empty(t, app.StartedDrivers()) + assert.True(t, stoppingReceived) + assert.True(t, stoppedReceived) + + assert.True(t, httpD.isStopped()) + assert.True(t, workerD.isStopped()) + assert.True(t, app.Context().IsDisposed()) + + stopOrderMu.Lock() + assert.Contains(t, stopOrder, "ctx_disposer") + stopOrderMu.Unlock() +} + +func TestAppStartDriverFailureRollback(t *testing.T) { + driver1 := newAppMockDriver(core.DriverTypeHTTP) + driver2 := newAppMockDriver(core.DriverTypeWorker) + driver2.startErr = errors.New("worker listen port conflict") + driver3 := newAppMockDriver(core.DriverTypeScheduler) + + p := &appMockPlugin{ + name: "fail_driver_plugin", + applyFn: func(ctx *core.Context) error { + _ = ctx.RegisterDriver(driver1) + _ = ctx.RegisterDriver(driver2) + _ = ctx.RegisterDriver(driver3) + return nil + }, + } + + app := core.NewApp( + core.WithProfile(core.ProfileAll), + core.WithPlugins(p), + ) + + err := app.Start(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "worker listen port conflict") + assert.False(t, app.IsRunning()) + + // Driver 1 was started then rolled back (stopped) + assert.True(t, driver1.isStarted()) + assert.True(t, driver1.isStopped()) + + // Driver 3 was never started + assert.False(t, driver3.isStarted()) +} + +func TestAppMigrationEngineExecution(t *testing.T) { + var migratedEntries []extpoints.MigrationEntry + runner := core.MigrationRunner(func(ctx context.Context, entries []extpoints.MigrationEntry) error { + migratedEntries = entries + return nil + }) + + sqlFS := fstest.MapFS{ + "migrations/001_init.sql": &fstest.MapFile{Data: []byte("CREATE TABLE users(id int);")}, + } + + p := &appMockPlugin{ + name: "auth", + applyFn: func(ctx *core.Context) error { + ctx.Migrations().Register("auth", sqlFS) + return nil + }, + } + + app := core.NewApp( + core.WithProfile(core.ProfileAll), + core.WithPlugins(p), + core.WithMigrationRunner(runner), + ) + + err := app.Start(context.Background()) + require.NoError(t, err) + defer func() { _ = app.Stop(context.Background()) }() + + require.Len(t, migratedEntries, 1) + assert.Equal(t, "auth", migratedEntries[0].PluginID) +} + +func TestAppMigrationEngineFromIoCContainer(t *testing.T) { + var executed bool + runner := core.MigrationRunner(func(ctx context.Context, entries []extpoints.MigrationEntry) error { + executed = true + return nil + }) + + sqlFS := fstest.MapFS{ + "migrations/001_init.sql": &fstest.MapFile{Data: []byte("CREATE TABLE logs(id int);")}, + } + + p := &appMockPlugin{ + name: "logstore", + applyFn: func(ctx *core.Context) error { + ctx.Migrations().Register("logstore", sqlFS) + core.Provide[core.MigrationEngine](ctx, runner) + return nil + }, + } + + app := core.NewApp( + core.WithProfile(core.ProfileAll), + core.WithPlugins(p), + ) + + err := app.Start(context.Background()) + require.NoError(t, err) + defer func() { _ = app.Stop(context.Background()) }() + + assert.True(t, executed) +} + +func TestAppRunContextCancellation(t *testing.T) { + d := newAppMockDriver(core.DriverTypeHTTP) + p := &appMockPlugin{ + name: "http_plugin", + applyFn: func(ctx *core.Context) error { + return ctx.RegisterDriver(d) + }, + } + + app := core.NewApp( + core.WithProfile(core.ProfileAPI), + core.WithPlugins(p), + core.WithShutdownTimeout(1*time.Second), + ) + + ctx, cancel := context.WithCancel(context.Background()) + + errCh := make(chan error, 1) + go func() { + errCh <- app.Run(ctx) + }() + + // Wait briefly then cancel + time.Sleep(50 * time.Millisecond) + assert.True(t, app.IsRunning()) + assert.True(t, d.isStarted()) + + cancel() + + select { + case err := <-errCh: + assert.NoError(t, err) + assert.False(t, app.IsRunning()) + assert.True(t, d.isStopped()) + case <-time.After(3 * time.Second): + t.Fatal("app.Run did not terminate upon context cancellation") + } +} + +func TestAppExecuteCLI(t *testing.T) { + // Test CLI argument parsing logic + tests := []struct { + args []string + expectedProfile core.Profile + }{ + {args: []string{"api"}, expectedProfile: core.ProfileAPI}, + {args: []string{"worker"}, expectedProfile: core.ProfileWorker}, + {args: []string{"scheduler"}, expectedProfile: core.ProfileSchedule}, + {args: []string{"schedule"}, expectedProfile: core.ProfileSchedule}, + {args: []string{"all"}, expectedProfile: core.ProfileAll}, + {args: []string{"--profile=worker"}, expectedProfile: core.ProfileWorker}, + {args: []string{"-p=api"}, expectedProfile: core.ProfileAPI}, + } + + for _, tt := range tests { + t.Run(tt.args[0], func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() // cancel immediately + + // Use custom root context to control cancellation + customApp := core.NewApp(core.WithContext(core.NewContext(ctx))) + _ = customApp.ExecuteCLI(tt.args...) + assert.Equal(t, tt.expectedProfile, customApp.Profile()) + }) + } +} + +func TestAppIdempotencyAndErrorStates(t *testing.T) { + app := core.NewApp() + + // Double start returns error + err := app.Start(context.Background()) + require.NoError(t, err) + + err = app.Start(context.Background()) + assert.ErrorIs(t, err, core.ErrAppRunning) + + // Stop clears running state + err = app.Stop(context.Background()) + require.NoError(t, err) + + // Double stop succeeds + err = app.Stop(context.Background()) + require.NoError(t, err) + + // Plugin apply failure + failPlugin := &appMockPlugin{ + name: "failing_plugin", + applyFn: func(ctx *core.Context) error { + return errors.New("plugin init boom") + }, + } + app2 := core.NewApp(core.WithPlugins(failPlugin)) + err = app2.Start(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "plugin init boom") + assert.False(t, app2.IsRunning()) + + // Migration failure + migFailRunner := core.MigrationRunner(func(ctx context.Context, entries []extpoints.MigrationEntry) error { + return errors.New("sql migrate error") + }) + sqlFS := fstest.MapFS{ + "migrations/001.sql": &fstest.MapFile{Data: []byte("...")}, + } + migPlugin := &appMockPlugin{ + name: "db_plugin", + applyFn: func(ctx *core.Context) error { + ctx.Migrations().Register("db_plugin", sqlFS) + return nil + }, + } + app3 := core.NewApp( + core.WithPlugins(migPlugin), + core.WithMigrationRunner(migFailRunner), + ) + err = app3.Start(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "sql migrate error") + assert.False(t, app3.IsRunning()) +} diff --git a/core/types.go b/core/types.go index 59b1d670..e6c411d0 100644 --- a/core/types.go +++ b/core/types.go @@ -29,6 +29,12 @@ var ( // ErrDriverNotFound is returned when a requested driver type is not registered. ErrDriverNotFound = errors.New("core: driver not found") + + // ErrAppRunning is returned when attempting to start an already running App. + ErrAppRunning = errors.New("core: app is already running") + + // ErrAppNotRunning is returned when attempting to operate on an App that is not running. + ErrAppNotRunning = errors.New("core: app is not running") ) // Plugin is the unified contract for all core and downstream plugins. @@ -69,6 +75,36 @@ type Driver interface { Stop(ctx context.Context) error } +// Profile identifies the runtime aspect or execution mode of an application. +type Profile string + +const ( + // ProfileAPI runs HTTP API server drivers. + ProfileAPI Profile = "api" + + // ProfileWorker runs asynchronous background worker drivers. + ProfileWorker Profile = "worker" + + // ProfileSchedule runs cron and timer schedule drivers. + ProfileSchedule Profile = "schedule" + + // ProfileAll runs all registered drivers concurrently in fused mode. + ProfileAll Profile = "all" +) + +// MigrationEngine is the interface for executing database migrations across registered plugins. +type MigrationEngine interface { + Migrate(ctx context.Context, entries []MigrationEntry) error +} + +// MigrationRunner is a function adapter implementing MigrationEngine. +type MigrationRunner func(ctx context.Context, entries []MigrationEntry) error + +// Migrate calls the underlying migration function. +func (fn MigrationRunner) Migrate(ctx context.Context, entries []MigrationEntry) error { + return fn(ctx, entries) +} + // Disposer is a cleanup function executed when a Context is disposed. type Disposer func() error diff --git a/internal/cmd/all.go b/internal/cmd/all.go index 21cc2e4a..6f7df684 100644 --- a/internal/cmd/all.go +++ b/internal/cmd/all.go @@ -6,15 +6,9 @@ package cmd import ( - "context" "log" - "sync" - gwrunner "github.com/Rain-kl/Wavelet/internal/apps/message_gateway/runner" - "github.com/Rain-kl/Wavelet/internal/infra/task/scheduler" - "github.com/Rain-kl/Wavelet/internal/infra/task/worker" - "github.com/Rain-kl/Wavelet/internal/platform/bootstrap" - "github.com/Rain-kl/Wavelet/internal/router" + "github.com/Rain-kl/Wavelet/core" "github.com/spf13/cobra" ) @@ -22,48 +16,9 @@ var allCmd = &cobra.Command{ Use: "all", Short: "以融合模式同时启动 API、Worker 和 Scheduler", Run: func(_ *cobra.Command, _ []string) { - log.Println("[All] 融合模式启动") - bootstrap.RegisterAll() - runBootstrap(bootstrap.Options{API: true}) - - var wg sync.WaitGroup - - // 启动 API HTTP 服务 - wg.Add(1) - go func() { - defer wg.Done() - log.Println("[All] 启动 API 服务") - router.Serve(func() { - printStartupBanner(startupState{mode: "API + Worker + Scheduler", relationalDB: latestMigrationState.relationalDB, clickHouseDB: latestMigrationState.clickHouseDB, listensForHTTP: true}) - }) - }() - - go func() { - if err := gwrunner.Start(context.Background()); err != nil { - log.Printf("[All] message gateway stopped: %v", err) - } - }() - - // 启动 Asynq Worker 任务处理服务 - wg.Add(1) - go func() { - defer wg.Done() - log.Println("[All] 启动 Worker 服务") - if err := worker.StartWorker(); err != nil { - log.Printf("[All] Worker 启动失败: %v\n", err) - } - }() - - // 启动 Asynq 定时任务调度器 - wg.Add(1) - go func() { - defer wg.Done() - log.Println("[All] 启动 Scheduler 服务") - if err := scheduler.StartScheduler(); err != nil { - log.Printf("[All] Scheduler 启动失败: %v\n", err) - } - }() - - wg.Wait() + app := newWaveletApp(core.ProfileAll) + if err := app.Run(); err != nil { + log.Fatalf("[All] run failed: %v\n", err) + } }, } diff --git a/internal/cmd/api.go b/internal/cmd/api.go index 1562cdf1..5c0b8d43 100644 --- a/internal/cmd/api.go +++ b/internal/cmd/api.go @@ -5,8 +5,9 @@ package cmd import ( - "github.com/Rain-kl/Wavelet/internal/platform/bootstrap" - "github.com/Rain-kl/Wavelet/internal/router" + "log" + + "github.com/Rain-kl/Wavelet/core" "github.com/spf13/cobra" ) @@ -14,10 +15,9 @@ var apiCmd = &cobra.Command{ Use: "api", Short: "wavelet API", Run: func(_ *cobra.Command, _ []string) { - bootstrap.RegisterAPI() - runBootstrap(bootstrap.Options{API: true}) - router.Serve(func() { - printStartupBanner(startupState{mode: "API", relationalDB: latestMigrationState.relationalDB, clickHouseDB: latestMigrationState.clickHouseDB, listensForHTTP: true}) - }) + app := newWaveletApp(core.ProfileAPI) + if err := app.Run(); err != nil { + log.Fatalf("[API] run failed: %v\n", err) + } }, } diff --git a/internal/cmd/app.go b/internal/cmd/app.go new file mode 100644 index 00000000..11da2fb2 --- /dev/null +++ b/internal/cmd/app.go @@ -0,0 +1,308 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package cmd + +import ( + "context" + "errors" + "fmt" + "log" + "net" + "net/http" + "sync" + "time" + + "github.com/Rain-kl/Wavelet/core" + "github.com/Rain-kl/Wavelet/core/extpoints" + gwrunner "github.com/Rain-kl/Wavelet/internal/apps/message_gateway/runner" + "github.com/Rain-kl/Wavelet/internal/infra/config" + "github.com/Rain-kl/Wavelet/internal/infra/task/scheduler" + "github.com/Rain-kl/Wavelet/internal/infra/task/worker" + "github.com/Rain-kl/Wavelet/internal/platform/bootstrap" + "github.com/Rain-kl/Wavelet/internal/router" + "github.com/Rain-kl/Wavelet/pkg/util" + "github.com/Rain-kl/Wavelet/plugins/infra/cache" + "github.com/Rain-kl/Wavelet/plugins/infra/database" + "github.com/Rain-kl/Wavelet/plugins/infra/logger" + "github.com/Rain-kl/Wavelet/plugins/infra/storage" + "github.com/hibiken/asynq" +) + +// newWaveletApp creates a core.App wired with Wavelet platform infrastructure and profile drivers. +// +//nolint:contextcheck +func newWaveletApp(profile core.Profile) *core.App { + app := core.NewApp( + core.WithProfile(profile), + core.WithShutdownTimeout(time.Duration(config.Config.App.GracefulShutdownTimeout)*time.Second), + ) + + // Register standard infrastructure plugins + app.Use( + database.New(), + cache.New(), + logger.New(), + storage.New(), + ) + + // Bind Goose migration runner + app.SetMigrationRunner(func(_ context.Context, _ []extpoints.MigrationEntry) error { + runMigrations() + return nil + }) + + // Mount drivers for each aspect + app.Use( + newWaveletHTTPDriver(profile), + newWaveletWorkerDriver(profile), + newWaveletSchedulerDriver(profile), + ) + + return app +} + +type waveletHTTPDriver struct { + profile core.Profile + server *http.Server + mu sync.Mutex + running bool +} + +func newWaveletHTTPDriver(profile core.Profile) *waveletHTTPDriver { + return &waveletHTTPDriver{profile: profile} +} + +func (d *waveletHTTPDriver) Name() string { + return "driver_wavelet_http" +} + +func (d *waveletHTTPDriver) Apply(ctx *core.Context) error { + return ctx.RegisterDriver(d) +} + +func (d *waveletHTTPDriver) Type() core.DriverType { + return core.DriverTypeHTTP +} + +//nolint:contextcheck +func (d *waveletHTTPDriver) Start(ctx context.Context) error { + d.mu.Lock() + defer d.mu.Unlock() + + if d.running { + return nil + } + + bootstrap.RegisterAPI() + runBootstrap(bootstrap.Options{API: true}) + + engine, err := router.BuildEngine() + if err != nil { + return fmt.Errorf("[API] build router engine failed: %w", err) + } + + srv := &http.Server{ + Addr: config.Config.App.Addr, + Handler: engine, + ReadHeaderTimeout: 10 * time.Second, + } + + listener, err := (&net.ListenConfig{}).Listen(ctx, "tcp", config.Config.App.Addr) + if err != nil { + return fmt.Errorf("[API] listen on %s failed: %w", config.Config.App.Addr, err) + } + + mode := "API" + if d.profile == core.ProfileAll { + mode = "API + Worker + Scheduler" + } + printStartupBanner(startupState{ + mode: mode, + relationalDB: latestMigrationState.relationalDB, + clickHouseDB: latestMigrationState.clickHouseDB, + listensForHTTP: true, + }) + + d.server = srv + d.running = true + + util.Go(func() { + log.Printf("[API] server listening on %s\n", config.Config.App.Addr) + if serveErr := srv.Serve(listener); serveErr != nil && !errors.Is(serveErr, http.ErrServerClosed) { + log.Fatalf("[API] server failed: %v\n", serveErr) + } + }) + + return nil +} + +func (d *waveletHTTPDriver) Stop(ctx context.Context) error { + d.mu.Lock() + defer d.mu.Unlock() + + if !d.running { + return nil + } + d.running = false + + var err error + if d.server != nil { + err = d.server.Shutdown(ctx) + d.server = nil + } + bootstrap.Stop(ctx) + log.Println("[API] server exited") + return err +} + +type waveletWorkerDriver struct { + profile core.Profile + server *asynq.Server + mu sync.Mutex + running bool +} + +func newWaveletWorkerDriver(profile core.Profile) *waveletWorkerDriver { + return &waveletWorkerDriver{profile: profile} +} + +func (d *waveletWorkerDriver) Name() string { + return "driver_wavelet_worker" +} + +func (d *waveletWorkerDriver) Apply(ctx *core.Context) error { + return ctx.RegisterDriver(d) +} + +func (d *waveletWorkerDriver) Type() core.DriverType { + return core.DriverTypeWorker +} + +//nolint:contextcheck +func (d *waveletWorkerDriver) Start(_ context.Context) error { + d.mu.Lock() + defer d.mu.Unlock() + + if d.running { + return nil + } + + if d.profile == core.ProfileAll { + bootstrap.RegisterAll() + } else { + bootstrap.RegisterWorker() + } + runBootstrap(bootstrap.Options{}) + + if d.profile == core.ProfileWorker { + printStartupBanner(startupState{ + mode: "Worker", + relationalDB: latestMigrationState.relationalDB, + clickHouseDB: latestMigrationState.clickHouseDB, + }) + } + + util.Go(func() { + if err := gwrunner.Start(context.Background()); err != nil { + log.Printf("[Worker] message gateway stopped: %v", err) + } + }) + + log.Println("[Worker] 启动任务处理服务") + srv, err := worker.StartWorkerServer() + if err != nil { + return fmt.Errorf("[Worker] 启动失败: %w", err) + } + + d.server = srv + d.running = true + return nil +} + +func (d *waveletWorkerDriver) Stop(_ context.Context) error { + d.mu.Lock() + defer d.mu.Unlock() + + if !d.running { + return nil + } + d.running = false + + if d.server != nil { + d.server.Stop() + d.server.Shutdown() + d.server = nil + } + log.Println("[Worker] 任务处理服务已退出") + return nil +} + +type waveletSchedulerDriver struct { + profile core.Profile + mu sync.Mutex + running bool +} + +func newWaveletSchedulerDriver(profile core.Profile) *waveletSchedulerDriver { + return &waveletSchedulerDriver{profile: profile} +} + +func (d *waveletSchedulerDriver) Name() string { + return "driver_wavelet_scheduler" +} + +func (d *waveletSchedulerDriver) Apply(ctx *core.Context) error { + return ctx.RegisterDriver(d) +} + +func (d *waveletSchedulerDriver) Type() core.DriverType { + return core.DriverTypeScheduler +} + +//nolint:contextcheck +func (d *waveletSchedulerDriver) Start(_ context.Context) error { + d.mu.Lock() + defer d.mu.Unlock() + + if d.running { + return nil + } + + if d.profile == core.ProfileAll { + bootstrap.RegisterAll() + } else { + bootstrap.RegisterScheduler() + } + runBootstrap(bootstrap.Options{}) + + if d.profile == core.ProfileSchedule { + printStartupBanner(startupState{ + mode: "Scheduler", + relationalDB: latestMigrationState.relationalDB, + clickHouseDB: latestMigrationState.clickHouseDB, + }) + } + + log.Println("[Scheduler] 启动定时任务调度服务") + if err := scheduler.ReloadScheduler(); err != nil { + return fmt.Errorf("[Scheduler] 启动失败: %w", err) + } + + d.running = true + return nil +} + +func (d *waveletSchedulerDriver) Stop(_ context.Context) error { + d.mu.Lock() + defer d.mu.Unlock() + + if !d.running { + return nil + } + d.running = false + + scheduler.StopScheduler() + log.Println("[Scheduler] 定时任务调度服务已退出") + return nil +} diff --git a/internal/cmd/app_test.go b/internal/cmd/app_test.go new file mode 100644 index 00000000..bc4621f7 --- /dev/null +++ b/internal/cmd/app_test.go @@ -0,0 +1,57 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package cmd + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/Rain-kl/Wavelet/core" +) + +func TestNewWaveletAppProfiles(t *testing.T) { + profiles := []core.Profile{ + core.ProfileAPI, + core.ProfileWorker, + core.ProfileSchedule, + core.ProfileAll, + } + + for _, prof := range profiles { + t.Run(string(prof), func(t *testing.T) { + app := newWaveletApp(prof) + require.NotNil(t, app) + assert.Equal(t, prof, app.Profile()) + + // Verify 4 infra plugins + 3 driver plugins registered + plugins := app.Plugins() + assert.Len(t, plugins, 7) + + // Verify each standard infra plugin is registered + _, ok := app.Plugin("database") + assert.True(t, ok, "database plugin missing") + + _, ok = app.Plugin("cache") + assert.True(t, ok, "cache plugin missing") + + _, ok = app.Plugin("logger") + assert.True(t, ok, "logger plugin missing") + + _, ok = app.Plugin("storage") + assert.True(t, ok, "storage plugin missing") + + // Verify driver plugins + _, ok = app.Plugin("driver_wavelet_http") + assert.True(t, ok, "http driver missing") + + _, ok = app.Plugin("driver_wavelet_worker") + assert.True(t, ok, "worker driver missing") + + _, ok = app.Plugin("driver_wavelet_scheduler") + assert.True(t, ok, "scheduler driver missing") + }) + } +} diff --git a/internal/cmd/scheduler.go b/internal/cmd/scheduler.go index 75f0a022..3f444f15 100644 --- a/internal/cmd/scheduler.go +++ b/internal/cmd/scheduler.go @@ -7,9 +7,7 @@ package cmd import ( "log" - "github.com/Rain-kl/Wavelet/internal/infra/task/scheduler" - "github.com/Rain-kl/Wavelet/internal/platform/bootstrap" - + "github.com/Rain-kl/Wavelet/core" "github.com/spf13/cobra" ) @@ -17,11 +15,9 @@ var schedulerCmd = &cobra.Command{ Use: "scheduler", Short: "wavelet Scheduler", Run: func(_ *cobra.Command, _ []string) { - runBootstrap(bootstrap.Options{}) - printStartupBanner(startupState{mode: "Scheduler", relationalDB: latestMigrationState.relationalDB, clickHouseDB: latestMigrationState.clickHouseDB}) - log.Println("[Scheduler] 启动定时任务调度服务") - if err := scheduler.StartScheduler(); err != nil { - log.Fatalf("[调度器] 启动失败: %v", err) + app := newWaveletApp(core.ProfileSchedule) + if err := app.Run(); err != nil { + log.Fatalf("[Scheduler] run failed: %v\n", err) } }, } diff --git a/internal/cmd/worker.go b/internal/cmd/worker.go index d1ca1054..372f9c2e 100644 --- a/internal/cmd/worker.go +++ b/internal/cmd/worker.go @@ -5,13 +5,9 @@ package cmd import ( - "context" "log" - gwrunner "github.com/Rain-kl/Wavelet/internal/apps/message_gateway/runner" - "github.com/Rain-kl/Wavelet/internal/infra/task/worker" - "github.com/Rain-kl/Wavelet/internal/platform/bootstrap" - + "github.com/Rain-kl/Wavelet/core" "github.com/spf13/cobra" ) @@ -19,16 +15,9 @@ var workerCmd = &cobra.Command{ Use: "worker", Short: "wavelet Worker", Run: func(_ *cobra.Command, _ []string) { - runBootstrap(bootstrap.Options{}) - printStartupBanner(startupState{mode: "Worker", relationalDB: latestMigrationState.relationalDB, clickHouseDB: latestMigrationState.clickHouseDB}) - go func() { - if err := gwrunner.Start(context.Background()); err != nil { - log.Printf("[Worker] message gateway stopped: %v", err) - } - }() - log.Println("[Worker] 启动任务处理服务") - if err := worker.StartWorker(); err != nil { - log.Fatalf("[工作器] 启动失败: %v", err) + app := newWaveletApp(core.ProfileWorker) + if err := app.Run(); err != nil { + log.Fatalf("[Worker] run failed: %v\n", err) } }, } diff --git a/internal/infra/task/worker/worker.go b/internal/infra/task/worker/worker.go index c53858db..854c3655 100644 --- a/internal/infra/task/worker/worker.go +++ b/internal/infra/task/worker/worker.go @@ -24,7 +24,7 @@ func StartWorker() error { asynq.Config{ Concurrency: config.Config.Worker.Concurrency, ShutdownTimeout: workerShutdownTimeout, - Queues: buildQueuesFromConfig(), + Queues: BuildQueuesFromConfig(), StrictPriority: config.Config.Worker.StrictPriority, }, ) @@ -44,8 +44,34 @@ func StartWorker() error { return asynqServer.Run(mux) } -// buildQueuesFromConfig 从配置构建队列映射 -func buildQueuesFromConfig() map[string]int { +// StartWorkerServer 异步启动 Asynq 工作器服务并返回 Server 实例以支持平滑停机 +func StartWorkerServer() (*asynq.Server, error) { + bootstrap.RegisterWorker() + asynqServer := asynq.NewServer( + task.RedisOpt, + asynq.Config{ + Concurrency: config.Config.Worker.Concurrency, + ShutdownTimeout: workerShutdownTimeout, + Queues: BuildQueuesFromConfig(), + StrictPriority: config.Config.Worker.StrictPriority, + }, + ) + + mux := asynq.NewServeMux() + mux.Use(taskLoggingMiddleware) + + for _, taskName := range task.GetRegisteredAsynqTasks() { + mux.HandleFunc(taskName, task.ProcessTask) + } + + if err := asynqServer.Start(mux); err != nil { + return nil, err + } + return asynqServer, nil +} + +// BuildQueuesFromConfig 从配置构建队列映射 +func BuildQueuesFromConfig() map[string]int { queues := make(map[string]int) // 从配置读取队列 diff --git a/internal/router/router.go b/internal/router/router.go index 51166f23..0130dc2a 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -31,8 +31,8 @@ import ( "go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin" ) -// Serve 启动 HTTP API 服务。onStarted 仅会在 HTTP 地址成功绑定后调用。 -func Serve(onStarted func()) { +// BuildEngine 构建并初始化 Gin 路由引擎及全部中间件和路由 +func BuildEngine() (*gin.Engine, error) { // 运行模式 if config.Config.App.IsProduction() { gin.SetMode(gin.ReleaseMode) @@ -60,7 +60,7 @@ func Serve(onStarted func()) { []byte(config.Config.App.SessionSecret), ) if err != nil { - log.Fatalf("[API] init session store failed: %v\n", err) + return nil, err } // 设置 Session Redis Key 前缀 @@ -78,6 +78,15 @@ func Serve(onStarted func()) { r.Use(otelgin.Middleware(config.Config.App.AppName), errorHandlerMiddleware(), loggerMiddleware(), risk_control.RiskControlMiddleware()) registerRoutes(r) + return r, nil +} + +// Serve 启动 HTTP API 服务。onStarted 仅会在 HTTP 地址成功绑定后调用。 +func Serve(onStarted func()) { + r, err := BuildEngine() + if err != nil { + log.Fatalf("[API] init session store failed: %v\n", err) + } srv := &http.Server{ Addr: config.Config.App.Addr, diff --git a/plugins/domain/admin/plugin.go b/plugins/domain/admin/plugin.go index 683a08e1..681e6650 100644 --- a/plugins/domain/admin/plugin.go +++ b/plugins/domain/admin/plugin.go @@ -210,7 +210,7 @@ func (p *Plugin) Apply(ctx *core.Context) error { } // 2. Register Background Tasks - ctx.Task().Register("admin:system_cleanup", func(c context.Context, t *asynq.Task) error { + ctx.Task().Register("admin:system_cleanup", func(_ context.Context, _ *asynq.Task) error { return nil }, extpoints.WithTaskRetry(1)) diff --git a/plugins/domain/auth/service.go b/plugins/domain/auth/service.go index 7a8f9c91..80de9f0a 100644 --- a/plugins/domain/auth/service.go +++ b/plugins/domain/auth/service.go @@ -101,7 +101,7 @@ func (s *authServiceImpl) VerifyToken(ctx context.Context, token string) (*contr return toUserDTO(user), nil } -func (s *authServiceImpl) CreateSession(ctx context.Context, userID uint64, extras map[string]any) (string, error) { +func (s *authServiceImpl) CreateSession(_ context.Context, _ uint64, _ map[string]any) (string, error) { // Session creation helper return "", nil } diff --git a/plugins/domain/message_gateway/plugin.go b/plugins/domain/message_gateway/plugin.go index 89a247a0..51840fd0 100644 --- a/plugins/domain/message_gateway/plugin.go +++ b/plugins/domain/message_gateway/plugin.go @@ -70,12 +70,14 @@ func (p *Plugin) Apply(ctx *core.Context) error { mgGroup.DELETE("/bindings/:id", appgw.UnbindBinding) } - // 3. Register Asynq background tasks - ctx.Task().Register("message_gateway:push_notification", func(c context.Context, t *asynq.Task) error { - return nil - }, extpoints.WithTaskRetry(3)) + const defaultTaskRetry = 3 - ctx.Task().Register("message_gateway:dispatch_bot_msg", func(c context.Context, t *asynq.Task) error { + // 3. Register Asynq background tasks + ctx.Task().Register("message_gateway:push_notification", func(_ context.Context, _ *asynq.Task) error { + return nil + }, extpoints.WithTaskRetry(defaultTaskRetry)) + + ctx.Task().Register("message_gateway:dispatch_bot_msg", func(_ context.Context, _ *asynq.Task) error { return nil }) @@ -83,7 +85,7 @@ func (p *Plugin) Apply(ctx *core.Context) error { ctx.Schedule().RegisterCron("*/10 * * * *", "message_gateway:cleanup_pairing_codes", map[string]any{"action": "cleanup"}) // 5. Register EventBus listeners for decoupled push triggers - ctx.Events().On("notification:push", func(c context.Context, e PushNotificationEvent) error { + ctx.Events().On("notification:push", func(_ context.Context, _ PushNotificationEvent) error { // Event triggered push handling return nil }) diff --git a/plugins/domain/user/plugin.go b/plugins/domain/user/plugin.go index bda5584f..432b1eaf 100644 --- a/plugins/domain/user/plugin.go +++ b/plugins/domain/user/plugin.go @@ -88,13 +88,15 @@ func (p *Plugin) Apply(ctx *core.Context) error { } } + const defaultUserTaskRetry = 3 + // 4. Register Asynq background tasks - ctx.Task().Register("user:send_email_code", func(c context.Context, t *asynq.Task) error { + ctx.Task().Register("user:send_email_code", func(_ context.Context, _ *asynq.Task) error { // Asynq background task handler return nil - }, extpoints.WithTaskRetry(3)) + }, extpoints.WithTaskRetry(defaultUserTaskRetry)) - ctx.Task().Register("user:cleanup_inactive", func(c context.Context, t *asynq.Task) error { + ctx.Task().Register("user:cleanup_inactive", func(_ context.Context, _ *asynq.Task) error { return nil }) diff --git a/plugins/domain/user/service.go b/plugins/domain/user/service.go index c4db94cd..9c50c12b 100644 --- a/plugins/domain/user/service.go +++ b/plugins/domain/user/service.go @@ -166,7 +166,7 @@ func (s *userServiceImpl) VerifyPassword(ctx context.Context, id uint64, passwor return user.CheckPassword(password) } -func (s *userServiceImpl) UpdateLastLogin(ctx context.Context, id uint64, ip string) error { +func (s *userServiceImpl) UpdateLastLogin(ctx context.Context, id uint64, _ string) error { return db.DB(ctx).Model(&model.User{}).Where("id = ?", id). Updates(map[string]any{ "last_login_at": time.Now(),