feat: 创建任务通知补齐信息

Author: yokowu

backend/biz/task/repo/task.go

@@ -107,6 +107,7 @@ func (t *TaskRepo) StatByIDs(ctx context.Context, ids []uuid.UUID) (map[uuid.UUI
 // GetByID implements domain.TaskRepo.
 func (t *TaskRepo) GetByID(ctx context.Context, id uuid.UUID) (*db.Task, error) {
 	return t.db.Task.Query().
+		WithUser().
 		WithProjectTasks(func(ptq *db.ProjectTaskQuery) {
 			ptq.
 				WithModel().

backend/biz/task/usecase/task.go

@@ -342,6 +342,7 @@ func (a *TaskUsecase) Create(ctx context.Context, user *domain.User, req domain.
 
 	ctx = entx.WithTaskConcurrencyLimit(ctx, limit)
 
+	var createdVm *taskflow.VirtualMachine
 	pt, err := a.repo.Create(ctx, user, req, token, func(pt *db.ProjectTask, m *db.Model, i *db.Image) (*taskflow.VirtualMachine, error) {
 		t := pt.Edges.Task
 		if t == nil {
@@ -387,6 +388,7 @@ func (a *TaskUsecase) Create(ctx context.Context, user *domain.User, req domain.
 		if vm == nil {
 			return nil, fmt.Errorf("vm is nil")
 		}
+		createdVm = vm
 
 		mcps := []taskflow.McpServerConfig{
 			{
@@ -407,23 +409,6 @@ func (a *TaskUsecase) Create(ctx context.Context, user *domain.User, req domain.
 			},
 		}
 
-		taskMeta := lifecycle.TaskMetadata{
-			TaskID: t.ID,
-			UserID: user.ID,
-		}
-		if err := a.taskLifecycle.Transition(ctx, t.ID, consts.TaskStatusPending, taskMeta); err != nil {
-			a.logger.WarnContext(ctx, "task lifecycle transition failed", "error", err)
-		}
-
-		vmMeta := lifecycle.VMMetadata{
-			VMID:   vm.ID,
-			TaskID: &t.ID,
-			UserID: user.ID,
-		}
-		if err := a.vmLifecycle.Transition(ctx, vm.ID, lifecycle.VMStatePending, vmMeta); err != nil {
-			a.logger.WarnContext(ctx, "vm lifecycle transition failed", "error", err)
-		}
-
 		// 存储 CreateTaskReq 到 Redis（10 分钟过期），供 Lifecycle Manager 消费
 		createTaskReq := &taskflow.CreateTaskReq{
 			ID:           t.ID,
@@ -456,6 +441,24 @@ func (a *TaskUsecase) Create(ctx context.Context, user *domain.User, req domain.
 		return nil, err
 	}
 	a.logger.With("req", req).InfoContext(ctx, "task created")
+	taskMeta := lifecycle.TaskMetadata{
+		TaskID: pt.TaskID,
+		UserID: user.ID,
+	}
+	if err := a.taskLifecycle.Transition(ctx, pt.TaskID, consts.TaskStatusPending, taskMeta); err != nil {
+		a.logger.WarnContext(ctx, "task lifecycle transition failed", "error", err)
+	}
+
+	if createdVm != nil {
+		vmMeta := lifecycle.VMMetadata{
+			VMID:   createdVm.ID,
+			TaskID: &pt.TaskID,
+			UserID: user.ID,
+		}
+		if err := a.vmLifecycle.Transition(ctx, createdVm.ID, lifecycle.VMStatePending, vmMeta); err != nil {
+			a.logger.WarnContext(ctx, "vm lifecycle transition failed", "error", err)
+		}
+	}
 
 	result := cvt.From(pt, &domain.ProjectTask{})
 

backend/pkg/lifecycle/tasknotifyhook.go

@@ -3,6 +3,7 @@ package lifecycle
 import (
 	"context"
 	"log/slog"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/samber/do"
@@ -16,13 +17,15 @@ import (
 type TaskNotifyHook struct {
 	notify *dispatcher.Dispatcher
 	logger *slog.Logger
+	repo   domain.TaskRepo
 }
 
 // NewTaskNotifyHook 创建任务通知 Hook
 func NewTaskNotifyHook(i *do.Injector) *TaskNotifyHook {
 	return &TaskNotifyHook{
 		notify: do.MustInvoke[*dispatcher.Dispatcher](i),
 		logger: do.MustInvoke[*slog.Logger](i).With("hook", "task-notify-hook"),
+		repo:   do.MustInvoke[domain.TaskRepo](i),
 	}
 }
 
@@ -38,17 +41,51 @@ func (h *TaskNotifyHook) OnStateChange(ctx context.Context, taskID uuid.UUID, fr
 	default:
 		return nil
 	}
+	logger := h.logger.With("task_id", taskID, "from", from, "to", to)
+
+	task, err := h.repo.GetByID(ctx, taskID)
+	if err != nil {
+		logger.With("error", err).ErrorContext(ctx, "failed to get task on state change")
+		return err
+	}
+
+	payload := domain.NotifyEventPayload{
+		TaskID:      taskID.String(),
+		TaskStatus:  string(to),
+		TaskContent: task.Content,
+		VMName:      "",
+		VMArch:      "",
+		VMCores:     0,
+		VMMemory:    0,
+		VMOS:        "",
+	}
+	if u := task.Edges.User; u != nil {
+		payload.UserName = u.Name
+	}
+	if pts := task.Edges.ProjectTasks; len(pts) > 0 {
+		pt := pts[0]
+		if m := pt.Edges.Model; m != nil {
+			payload.ModelName = m.Model
+		}
+	}
+	if vms := task.Edges.Vms; len(vms) > 0 {
+		vm := vms[0]
+		payload.VMID = vm.ID
+		payload.VMName = vm.Name
+		payload.VMArch = vm.Arch
+		payload.VMCores = vm.Cores
+		payload.VMMemory = vm.Memory
+		payload.VMOS = vm.Os
+	}
 
 	event := &domain.NotifyEvent{
 		EventType:     eventType,
 		SubjectUserID: metadata.UserID,
 		RefID:         taskID.String(),
-		Payload: domain.NotifyEventPayload{
-			TaskID:     taskID.String(),
-			TaskStatus: string(to),
-		},
+		Payload:       payload,
+		OccurredAt:    time.Now(),
 	}
 
-	h.logger.InfoContext(ctx, "publishing notify event", "event", eventType, "task_id", taskID)
+	logger.InfoContext(ctx, "publishing notify event", "event", eventType)
 	return h.notify.Publish(ctx, event)
 }