merge main: resolve bootstrap and router init conflicts

This commit is contained in:
ryan
2026-06-18 10:38:55 +08:00
21 changed files with 1411 additions and 33 deletions
@@ -9,7 +9,7 @@ import (
"time"
"github.com/Rain-kl/Wavelet/internal/apps/admin/push"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/listener"
)
// AdminLogin is the metadata definition for the admin login event.
@@ -24,20 +24,15 @@ var AdminLogin = push.EventMetadata{
Description: "当管理员成功登录系统时触发此通知",
}
func init() {
push.RegisterBuiltInEvent(AdminLogin)
}
// TriggerAdminLoginEvent triggers the admin login event asynchronously.
func TriggerAdminLoginEvent(ctx context.Context, user *model.User, ip string) {
if user == nil || !user.IsAdmin {
func handleAdminLogin(ctx context.Context, event listener.AdminLoggedIn) {
if event.User == nil {
return
}
body := map[string]any{
"user": user,
"ip": ip,
"user": event.User,
"ip": event.IP,
"time": time.Now().Format("2006-01-02 15:04:05"),
}
push.DefaultTrigger.Trigger(ctx, AdminLogin, body)
}
}
@@ -0,0 +1,17 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package custom_events
import (
"github.com/Rain-kl/Wavelet/internal/apps/admin/push"
"github.com/Rain-kl/Wavelet/internal/listener"
)
// Register wires push notification handlers for domain events and registers
// built-in event metadata. Must be called once during application bootstrap
// before push.SyncEvents.
func Register() {
push.RegisterBuiltInEvent(AdminLogin)
listener.OnAdminLoggedIn(handleAdminLogin)
}
+3 -2
View File
@@ -15,8 +15,9 @@ import (
"github.com/Rain-kl/Wavelet/pkg/logger"
)
func init() {
task.OnTaskCompleted = handleTaskCompleted
// RegisterTaskListeners subscribes push notification handlers to task completion events.
func RegisterTaskListeners() {
task.OnTaskCompleted(handleTaskCompleted)
}
// handleTaskCompleted handles task completions and triggers appropriate push events.
-5
View File
@@ -13,7 +13,6 @@ import ("fmt"
"github.com/Rain-kl/Wavelet/internal/apps/admin"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/task"
taskhandlers "github.com/Rain-kl/Wavelet/internal/task/handlers"
"github.com/Rain-kl/Wavelet/internal/task/scheduler"
"github.com/Rain-kl/Wavelet/pkg/logger"
"github.com/gin-gonic/gin"
@@ -21,10 +20,6 @@ import ("fmt"
"github.com/Rain-kl/Wavelet/internal/common/response")
func init() {
taskhandlers.Register()
}
// ListTaskTypes 获取支持的任务类型列表
// @Summary 获取支持的任务类型
// @Description 返回系统支持的所有可调度任务类型列表,包括任务名称、描述、是否支持时间范围等元数据,需要管理员权限
+2 -2
View File
@@ -14,11 +14,11 @@ import (
"strings"
"time"
"github.com/Rain-kl/Wavelet/internal/apps/admin/push/custom_events"
"github.com/Rain-kl/Wavelet/internal/common"
"github.com/Rain-kl/Wavelet/internal/common/response"
"github.com/Rain-kl/Wavelet/internal/config"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/listener"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/pkg/logger"
"github.com/coreos/go-oidc/v3/oidc"
@@ -678,7 +678,7 @@ func handleCallbackLogin(ctx context.Context, c *gin.Context, source *model.Auth
logger.InfoF(ctx, "[LoginAudit] successful OAuth login via source: %s, external ID: %s, user: %s, ID: %d, IP: %s", source.Name, userInfo.Sub, user.Username, user.ID, c.ClientIP())
custom_events.TriggerAdminLoginEvent(ctx, &user, c.ClientIP())
listener.EmitAdminLoggedIn(ctx, &user, c.ClientIP())
c.JSON(http.StatusOK, response.OK(buildCallbackResult(&user, "logged_in")))
}
+2 -2
View File
@@ -8,12 +8,12 @@ import ("context"
"strings"
"time"
"github.com/Rain-kl/Wavelet/internal/apps/admin/push/custom_events"
"github.com/Rain-kl/Wavelet/internal/apps/oauth"
"github.com/Rain-kl/Wavelet/internal/common"
"github.com/Rain-kl/Wavelet/internal/config"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/db/idgen"
"github.com/Rain-kl/Wavelet/internal/listener"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/util"
"github.com/Rain-kl/Wavelet/pkg/logger"
@@ -194,7 +194,7 @@ func Login(c *gin.Context) {
logger.InfoF(ctx, "[LoginAudit] successful login for user: %s, ID: %d, IP: %s", user.Username, user.ID, c.ClientIP())
custom_events.TriggerAdminLoginEvent(ctx, &user, c.ClientIP())
listener.EmitAdminLoggedIn(ctx, &user, c.ClientIP())
c.JSON(http.StatusOK, response.OK(oauth.BuildBasicUserInfo(&user, needChangePassword)))
}
View File
+42
View File
@@ -0,0 +1,42 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
// Package listener provides domain event dispatch for cross-module integration.
// Core domains emit events here; operational modules (push, webhooks, etc.)
// subscribe at the application composition root.
package listener
import (
"context"
"github.com/Rain-kl/Wavelet/internal/model"
)
// AdminLoggedIn is emitted when an administrator successfully authenticates.
type AdminLoggedIn struct {
User *model.User
IP string
}
// AdminLoggedInHandler handles administrator login domain events.
type AdminLoggedInHandler func(ctx context.Context, event AdminLoggedIn)
var adminLoggedInHandlers []AdminLoggedInHandler
// OnAdminLoggedIn registers a handler for administrator login events.
// Handlers must be registered during application bootstrap before serving traffic.
func OnAdminLoggedIn(handler AdminLoggedInHandler) {
adminLoggedInHandlers = append(adminLoggedInHandlers, handler)
}
// EmitAdminLoggedIn dispatches an administrator login event to all registered handlers.
func EmitAdminLoggedIn(ctx context.Context, user *model.User, ip string) {
if user == nil || !user.IsAdmin {
return
}
event := AdminLoggedIn{User: user, IP: ip}
for _, handler := range adminLoggedInHandlers {
handler(ctx, event)
}
}
+21 -5
View File
@@ -26,8 +26,16 @@ import (
// handlerRegistry 已注册的任务处理器
var handlerRegistry = make(map[string]TaskHandler)
// OnTaskCompleted is a hook called when a task execution completes.
var OnTaskCompleted func(ctx context.Context, execution *model.TaskExecution, result *TaskResult, execErr error)
// CompletedHandler is called when a task execution completes.
type CompletedHandler func(ctx context.Context, execution *model.TaskExecution, result *TaskResult, execErr error)
var taskCompletedHandlers []CompletedHandler
// OnTaskCompleted registers a handler for task completion events.
// Handlers must be registered during application bootstrap before processing tasks.
func OnTaskCompleted(handler CompletedHandler) {
taskCompletedHandlers = append(taskCompletedHandlers, handler)
}
// RegisterHandler 注册任务处理器
// 传入任务类型标识(对应 constants.go 中的 AsynqTask 常量)和 TaskHandler 实现
@@ -405,9 +413,17 @@ func completeTaskExecution(ctx context.Context, execution *model.TaskExecution,
}
}
if OnTaskCompleted != nil {
asyncCtx := context.WithoutCancel(ctx)
go OnTaskCompleted(asyncCtx, execution, result, execErr)
notifyTaskCompleted(ctx, execution, result, execErr)
}
func notifyTaskCompleted(ctx context.Context, execution *model.TaskExecution, result *TaskResult, execErr error) {
if len(taskCompletedHandlers) == 0 {
return
}
asyncCtx := context.WithoutCancel(ctx)
for _, handler := range taskCompletedHandlers {
go handler(asyncCtx, execution, result, execErr)
}
}
+3
View File
@@ -11,6 +11,7 @@ import (
"syscall"
"time"
"github.com/Rain-kl/Wavelet/internal/bootstrap"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/task"
"github.com/Rain-kl/Wavelet/pkg/logger"
@@ -32,6 +33,8 @@ func GetAsynqClient() *asynq.Client {
// StartScheduler 启动调度器 (该函数阻塞,直到调度器退出)
func StartScheduler() error {
bootstrap.RegisterScheduler()
var err error
schedulerOnce.Do(func() {
quitChan = make(chan struct{})
+2 -6
View File
@@ -7,22 +7,18 @@ package worker
import (
"time"
"github.com/Rain-kl/Wavelet/internal/bootstrap"
"github.com/Rain-kl/Wavelet/internal/config"
"github.com/Rain-kl/Wavelet/internal/task"
taskhandlers "github.com/Rain-kl/Wavelet/internal/task/handlers"
"github.com/hibiken/asynq"
)
// workerShutdownTimeout Worker 优雅关闭超时时间
const workerShutdownTimeout = 3 * time.Minute
func init() {
// 注册所有任务处理器
taskhandlers.Register()
}
// StartWorker 启动任务处理服务器
func StartWorker() error {
bootstrap.RegisterWorker()
asynqServer := asynq.NewServer(
task.RedisOpt,
asynq.Config{