feat(push-task): connect notification module with task module

- Add task_type to w_push_events table and GORM models. - Implement OnTaskCompleted callback hook in task executor to avoid circular dependencies. - Implement task listener in push package to trigger notifications on task completion. - Automatically resolve User objects from payload and results. - Enhance UI to select task completed events and preview default templates. - Update Swagger documentation.
2026-10-03 07:06:36 +08:00 · 2026-06-15 15:12:02 +08:00
parent de58b118b4
commit 2a3b9f8a5f
12 changed files with 540 additions and 60 deletions
@@ -15,6 +15,7 @@ import (

 	"github.com/Rain-kl/Wavelet/internal/db"
 	"github.com/Rain-kl/Wavelet/internal/model"
+	"github.com/Rain-kl/Wavelet/internal/task"
 	"github.com/Rain-kl/Wavelet/internal/util"
 	"github.com/Rain-kl/Wavelet/pkg/push"
 	"github.com/gin-gonic/gin"
@@ -84,7 +85,8 @@ func ListEvents(c *gin.Context) {

 // CreateEventRequest 创建事件请求参数
 type CreateEventRequest struct {
-	EventKey string   `json:"event_key" binding:"required"`
+	EventKey string   `json:"event_key"`
+	TaskType string   `json:"task_type"` // 关联的异步任务类型
 	Channels []string `json:"channels"`
 	Targets  []string `json:"targets"`
 	Template string   `json:"template"`
@@ -112,9 +114,49 @@ func ListBuiltInEvents(c *gin.Context) {
 	c.JSON(http.StatusOK, util.OK(BuiltInEvents))
 }

+func getEventInfo(req CreateEventRequest) (string, string, []byte, error) {
+	if req.TaskType != "" {
+		// 1. 检查关联任务是否存在
+		meta := task.GetTaskMetaByAsynqTask(req.TaskType)
+		if meta == nil {
+			return "", "", nil, errors.New("unsupported task type")
+		}
+		eventKey := "task_completed:" + req.TaskType
+		eventName := "任务完成: " + meta.Name
+
+		defaultTemplate := NotificationMessage{
+			Title:   "任务完成: " + meta.Name,
+			Content: "异步任务 {{task_name}} (ID: {{task_id}}) 已完成。状态: {{task_status}}，耗时: {{task_duration}} ms。",
+			Level:   defaultLevelInfo,
+		}
+		defaultTemplateBytes, err := json.Marshal(defaultTemplate)
+		if err != nil {
+			return "", "", nil, err
+		}
+		return eventKey, eventName, defaultTemplateBytes, nil
+	}
+
+	if req.EventKey == "" {
+		return "", "", nil, errors.New("either event_key or task_type must be provided")
+	}
+
+	// 1. 检查内置事件是否存在
+	meta, found := findBuiltInEvent(req.EventKey)
+	if !found {
+		return "", "", nil, errors.New("unsupported built-in event key")
+	}
+
+	defaultTemplateBytes, err := json.Marshal(meta.DefaultTemplate)
+	if err != nil {
+		return "", "", nil, err
+	}
+
+	return req.EventKey, meta.Name, defaultTemplateBytes, nil
+}
+
 // CreateEvent 创建通知事件
 // @Summary 创建通知事件
-// @Description 绑定系统内置事件、推送渠道、接收目标并创建通知事件配置，需要管理员权限
+// @Description 绑定系统内置事件或异步任务、推送渠道、接收目标并创建通知事件配置，需要管理员权限
 // @Tags admin-push
 // @Accept json
 // @Produce json
@@ -131,16 +173,15 @@ func CreateEvent(c *gin.Context) {

 	ctx := c.Request.Context()

-	// 1. 检查内置事件是否存在
-	meta, found := findBuiltInEvent(req.EventKey)
-	if !found {
-		c.JSON(http.StatusBadRequest, util.Err("unsupported built-in event key"))
+	eventKey, eventName, defaultTemplateBytes, err := getEventInfo(req)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, util.Err(err.Error()))
 		return
 	}

 	// 2. 检查是否已经创建过该事件的配置
 	var count int64
-	if err := db.DB(ctx).Model(&model.PushEvent{}).Where("event_key = ?", req.EventKey).Count(&count).Error; err != nil {
+	if err := db.DB(ctx).Model(&model.PushEvent{}).Where("event_key = ?", eventKey).Count(&count).Error; err != nil {
 		c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
 		return
 	}
@@ -152,11 +193,6 @@ func CreateEvent(c *gin.Context) {
 	// 3. 模板处理
 	templateStr := strings.TrimSpace(req.Template)
 	if templateStr == "" {
-		defaultTemplateBytes, err := json.Marshal(meta.DefaultTemplate)
-		if err != nil {
-			c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
-			return
-		}
 		templateStr = string(defaultTemplateBytes)
 	} else {
 		var tempMap map[string]any
@@ -177,8 +213,9 @@ func CreateEvent(c *gin.Context) {
 	}

 	event := model.PushEvent{
-		EventKey: meta.Key,
-		Name:     meta.Name,
+		EventKey: eventKey,
+		Name:     eventName,
+		TaskType: req.TaskType,
 		Channels: channels,
 		Targets:  targets,
 		Template: templateStr,
@@ -0,0 +1,156 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package push
+
+import (
+	"context"
+	"encoding/json"
+	"strconv"
+	"time"
+
+	"github.com/Rain-kl/Wavelet/internal/db"
+	"github.com/Rain-kl/Wavelet/internal/logger"
+	"github.com/Rain-kl/Wavelet/internal/model"
+	"github.com/Rain-kl/Wavelet/internal/task"
+)
+
+func init() {
+	task.OnTaskCompleted = handleTaskCompleted
+}
+
+// handleTaskCompleted handles task completions and triggers appropriate push events.
+func handleTaskCompleted(ctx context.Context, execution *model.TaskExecution, result *task.TaskResult, execErr error) {
+	// Query all active push events configured for this task type
+	var events []model.PushEvent
+	err := db.DB(ctx).Where("task_type = ? AND enabled = ?", execution.TaskType, true).Find(&events).Error
+	if err != nil {
+		logger.ErrorF(ctx, "push_task_completed_listener: failed to query push events for task type %s: %v", execution.TaskType, err)
+		return
+	}
+
+	if len(events) == 0 {
+		return
+	}
+
+	// Build the notification body context
+	body := map[string]any{
+		"task_id":       execution.TaskID,
+		"task_name":     execution.TaskName,
+		"task_type":     execution.TaskType,
+		"task_status":   string(execution.Status),
+		"task_duration": execution.Duration,
+		"time":          time.Now().Format("2006-01-02 15:04:05"),
+	}
+
+	if execErr != nil {
+		body["task_error"] = execErr.Error()
+	} else {
+		body["task_error"] = ""
+	}
+
+	if result != nil {
+		body["task_result"] = result.Message
+	} else {
+		body["task_result"] = ""
+	}
+
+	// Parse payload parameters if it is valid JSON
+	var payloadMap map[string]any
+	if execution.Payload != "" {
+		if err := json.Unmarshal([]byte(execution.Payload), &payloadMap); err == nil {
+			body["payload"] = payloadMap
+			extractUserFromMap(ctx, payloadMap, body)
+		}
+	}
+
+	// Parse result detail parameters if it is valid JSON
+	if result != nil && result.Detail != "" {
+		var detailMap map[string]any
+		if err := json.Unmarshal([]byte(result.Detail), &detailMap); err == nil {
+			body["detail"] = detailMap
+			extractUserFromMap(ctx, detailMap, body)
+		}
+	}
+
+	// Trigger notifications for all configured events
+	for _, event := range events {
+		meta := EventMetadata{
+			Key:         event.EventKey,
+			Name:        event.Name,
+			Description: "异步任务执行完毕触发的自动通知",
+		}
+		DefaultTrigger.Trigger(ctx, meta, body)
+	}
+}
+
+// extractUserFromMap tries to find user information from a map and load the full User model.
+func extractUserFromMap(ctx context.Context, data map[string]any, body map[string]any) {
+	if u, exists := body["user"]; exists && u != nil {
+		return
+	}
+
+	if uVal, ok := data["user"]; ok && uVal != nil {
+		body["user"] = uVal
+		return
+	}
+
+	if userID, ok := extractUserID(data); ok && userID > 0 {
+		var u model.User
+		if err := db.DB(ctx).Where("id = ?", userID).First(&u).Error; err == nil {
+			body["user"] = &u
+			return
+		}
+	}
+
+	if username := extractUsername(data); username != "" {
+		var u model.User
+		if err := db.DB(ctx).Where("username = ?", username).First(&u).Error; err == nil {
+			body["user"] = &u
+			return
+		}
+	}
+}
+
+// extractUserID extracts and validates a userID from map fields.
+func extractUserID(data map[string]any) (uint64, bool) {
+	for _, k := range []string{"user_id", "userId", "uid"} {
+		val, ok := data[k]
+		if !ok || val == nil {
+			continue
+		}
+		switch v := val.(type) {
+		case float64:
+			if v >= 0 {
+				return uint64(v), true
+			}
+		case int:
+			if v >= 0 {
+				return uint64(v), true
+			}
+		case int64:
+			if v >= 0 {
+				return uint64(v), true
+			}
+		case uint64:
+			return v, true
+		case string:
+			if id, err := strconv.ParseUint(v, 10, 64); err == nil {
+				return id, true
+			}
+		}
+	}
+	return 0, false
+}
+
+// extractUsername extracts a username string from map fields.
+func extractUsername(data map[string]any) string {
+	for _, k := range []string{"username", "user_name"} {
+		if val, ok := data[k]; ok && val != nil {
+			if s, ok := val.(string); ok && s != "" {
+				return s
+			}
+		}
+	}
+	return ""
+}