feat(core): implement app profile lifecycle dispatcher and wire cli commands

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
+	}
+}

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())
+}

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
 

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)
+		}
 	},
 }

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)
+		}
 	},
 }

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
+}

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")
+		})
+	}
+}

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)
 		}
 	},
 }

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)
 		}
 	},
 }

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)
 
 	// 从配置读取队列

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,

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))
 

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
 }

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
 	})

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
 	})
 

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(),