From 84371e2ff8d0df8bf4ad6d0eeb4490532bd4ef11 Mon Sep 17 00:00:00 2001 From: Losita <2810873701@qq.com> Date: Mon, 16 Mar 2026 20:33:33 +0800 Subject: [PATCH] =?UTF-8?q?Version:=200.6.3.dev.260316=20=E2=9C=A8=20feat(?= =?UTF-8?q?task):=20=E6=96=B0=E5=A2=9E=E5=9B=9B=E8=B1=A1=E9=99=90=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E6=87=92=E8=A7=A6=E5=8F=91=E8=87=AA=E5=8A=A8=E5=B9=B3?= =?UTF-8?q?=E7=A7=BB=E9=93=BE=E8=B7=AF=EF=BC=88=E8=AF=BB=E6=97=B6=E6=B4=BE?= =?UTF-8?q?=E7=94=9F=20+=20Outbox=20=E5=BC=82=E6=AD=A5=E6=94=B6=E6=95=9B?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 🧩 为 `Task` 模型新增 `urgency_threshold_at` 字段，并补充复合索引 `user_id,is_completed,urgency_threshold_at,priority` 及相关事件 payload - ♻️ 重构 `TaskService.GetUserTasks`：调整为“缓存/DB 读取原始任务 -> 读时派生优先级（`2 -> 1`、`4 -> 3`）-> 通过 `SETNX` 去重后发布平移事件”的处理链路 - 🚚 新增任务平移事件链路： - `service/events/task_urgency_promote.go` - 事件类型：`task.urgency.promote.requested` - 支持 `Publish` + `RegisterHandler` + `ConsumeAndMarkConsumed` 的事务化消费流程 - 🛡️ 为 `TaskDAO` 新增幂等批量更新能力 `PromoteTaskUrgencyByIDs`，采用条件更新策略，仅对“达到阈值且未完成”的任务生效 - 🔌 更新启动接线逻辑：注册任务平移 handler，并将 `eventBus` 注入 `NewTaskService` - 🧹 修复并升级任务缓存层，统一为 `[]model.Task` 原始模型缓存；同时清理误导性注释，并补充详细中文步骤化注释 - 🔗 打通 `QuickNote` 链路中的 `urgency_threshold_at` 透传与校验，覆盖 `state` / `tool` / `nodes` / `prompt` / `agent_quick_note` 全链路 - 💾 写库时补充落库 `task.UrgencyThresholdAt` - 📝 新增功能决策记录 之前画的饼正在一块块填上~这一块饼填上之后，第一批开发的后端部分基本已经搞定了。后面的功能全都是天马行空的拓展功能。 --- backend/agent/quicknote/nodes.go | 105 +++++-- backend/agent/quicknote/prompt.go | 17 +- backend/agent/quicknote/state.go | 9 +- backend/agent/quicknote/tool.go | 27 +- backend/cmd/start.go | 5 +- backend/dao/cache.go | 78 +++-- backend/dao/task.go | 65 +++++ backend/model/task.go | 54 +++- backend/service/agentsvc/agent_quick_note.go | 20 +- .../service/events/task_urgency_promote.go | 136 +++++++++ backend/service/task.go | 272 ++++++++++++++++-- .../任务四象限_自动平移_决策记录.md | 119 ++++++++ 12 files changed, 792 insertions(+), 115 deletions(-) create mode 100644 backend/service/events/task_urgency_promote.go create mode 100644 docs/功能决策记录/任务四象限_自动平移_决策记录.md diff --git a/backend/agent/quicknote/nodes.go b/backend/agent/quicknote/nodes.go index 4b52819..7742bc9 100644 --- a/backend/agent/quicknote/nodes.go +++ b/backend/agent/quicknote/nodes.go @@ -24,17 +24,19 @@ type quickNoteIntentModelOutput struct { } type quickNotePriorityModelOutput struct { - PriorityGroup int `json:"priority_group"` - Reason string `json:"reason"` + PriorityGroup int `json:"priority_group"` + Reason string `json:"reason"` + UrgencyThresholdAt string `json:"urgency_threshold_at"` } // quickNotePlanModelOutput 是“单请求聚合规划”节点的模型输出。 type quickNotePlanModelOutput struct { - Title string `json:"title"` - DeadlineAt string `json:"deadline_at"` - PriorityGroup int `json:"priority_group"` - PriorityReason string `json:"priority_reason"` - Banter string `json:"banter"` + Title string `json:"title"` + DeadlineAt string `json:"deadline_at"` + UrgencyThresholdAt string `json:"urgency_threshold_at"` + PriorityGroup int `json:"priority_group"` + PriorityReason string `json:"priority_reason"` + Banter string `json:"banter"` } // runQuickNoteIntentNode 负责“意图识别 + 聚合规划 + 时间校验”。 @@ -69,6 +71,9 @@ func runQuickNoteIntentNode(ctx context.Context, st *QuickNoteState, input Quick st.ExtractedDeadline = plan.Deadline } st.ExtractedDeadlineText = strings.TrimSpace(plan.DeadlineText) + if plan.UrgencyThreshold != nil { + st.ExtractedUrgencyThreshold = normalizeUrgencyThreshold(plan.UrgencyThreshold, plan.Deadline) + } if IsValidTaskPriority(plan.PriorityGroup) { st.ExtractedPriority = plan.PriorityGroup st.ExtractedPriorityReason = strings.TrimSpace(plan.PriorityReason) @@ -229,8 +234,15 @@ func runQuickNotePriorityNode(ctx context.Context, st *QuickNoteState, input Qui 请仅输出 JSON（不要 markdown，不要解释）： { "priority_group": 1|2|3|4, - "reason": "简短理由" -}`, + "reason": "简短理由", + "urgency_threshold_at": "yyyy-MM-dd HH:mm 或空字符串" +} + +额外约束： +1) urgency_threshold_at 表示“何时从不紧急象限自动平移到紧急象限”； +2) 若该任务不需要自动平移，可输出空字符串； +3) 若任务已在紧急象限（priority_group=1 或 3），优先输出空字符串； +4) 若输出非空时间，必须是绝对时间，且不晚于归一化截止时间（若有）。`, st.RequestNowText, st.ExtractedTitle, st.UserInput, @@ -256,6 +268,12 @@ func runQuickNotePriorityNode(ctx context.Context, st *QuickNoteState, input Qui st.ExtractedPriority = parsed.PriorityGroup st.ExtractedPriorityReason = strings.TrimSpace(parsed.Reason) + if strings.TrimSpace(parsed.UrgencyThresholdAt) != "" { + urgencyThreshold, thresholdErr := parseOptionalDeadlineWithNow(strings.TrimSpace(parsed.UrgencyThresholdAt), st.RequestNow) + if thresholdErr == nil { + st.ExtractedUrgencyThreshold = normalizeUrgencyThreshold(urgencyThreshold, st.ExtractedDeadline) + } + } return st, nil } @@ -283,12 +301,17 @@ func runQuickNotePersistNodeInternal(ctx context.Context, st *QuickNoteState, cr if st.ExtractedDeadline != nil { deadlineText = st.ExtractedDeadline.In(quickNoteLocation()).Format(time.RFC3339) } + urgencyThresholdText := "" + if st.ExtractedUrgencyThreshold != nil { + urgencyThresholdText = st.ExtractedUrgencyThreshold.In(quickNoteLocation()).Format(time.RFC3339) + } // 3. 工具参数序列化失败视作一次失败尝试，交由重试分支处理。 toolInput := QuickNoteCreateTaskToolInput{ - Title: st.ExtractedTitle, - PriorityGroup: priority, - DeadlineAt: deadlineText, + Title: st.ExtractedTitle, + PriorityGroup: priority, + DeadlineAt: deadlineText, + UrgencyThresholdAt: urgencyThresholdText, } rawInput, marshalErr := json.Marshal(toolInput) if marshalErr != nil { @@ -444,12 +467,14 @@ func callModelForJSONWithMaxTokens(ctx context.Context, chatModel *ark.ChatModel } type quickNotePlannedResult struct { - Title string - Deadline *time.Time - DeadlineText string - PriorityGroup int - PriorityReason string - Banter string + Title string + Deadline *time.Time + DeadlineText string + UrgencyThreshold *time.Time + UrgencyThresholdText string + PriorityGroup int + PriorityReason string + Banter string } // planQuickNoteInSingleCall 在一次模型调用里完成“时间/优先级/banter”聚合规划。 @@ -468,6 +493,7 @@ func planQuickNoteInSingleCall( { "title": string, "deadline_at": string, + "urgency_threshold_at": string, "priority_group": 1|2|3|4, "priority_reason": string, "banter": string @@ -475,8 +501,10 @@ func planQuickNoteInSingleCall( 约束： 1) deadline_at 只允许 "yyyy-MM-dd HH:mm" 或空字符串； -2) 若用户给了相对时间（如明天/今晚/下周一），必须换算为绝对时间； -3) banter 只允许一句中文，不超过30字，不得改动任务事实。`, +2) urgency_threshold_at 只允许 "yyyy-MM-dd HH:mm" 或空字符串； +3) 若用户给了相对时间（如明天/今晚/下周一），必须换算为绝对时间； +4) 若任务不需要自动平移，可让 urgency_threshold_at 为空； +5) banter 只允许一句中文，不超过30字，不得改动任务事实。`, nowText, strings.TrimSpace(userInput), ) @@ -493,11 +521,12 @@ func planQuickNoteInSingleCall( } result := &quickNotePlannedResult{ - Title: strings.TrimSpace(parsed.Title), - DeadlineText: strings.TrimSpace(parsed.DeadlineAt), - PriorityGroup: parsed.PriorityGroup, - PriorityReason: strings.TrimSpace(parsed.PriorityReason), - Banter: strings.TrimSpace(parsed.Banter), + Title: strings.TrimSpace(parsed.Title), + DeadlineText: strings.TrimSpace(parsed.DeadlineAt), + UrgencyThresholdText: strings.TrimSpace(parsed.UrgencyThresholdAt), + PriorityGroup: parsed.PriorityGroup, + PriorityReason: strings.TrimSpace(parsed.PriorityReason), + Banter: strings.TrimSpace(parsed.Banter), } // 4. banter 只保留首行，防止模型输出多行破坏最终回复风格。 @@ -513,6 +542,12 @@ func planQuickNoteInSingleCall( result.Deadline = deadline } } + // 6. 对 urgency_threshold_at 做本地二次校验，并与 deadline 做上界约束。 + if result.UrgencyThresholdText != "" { + if urgencyThreshold, thresholdErr := parseOptionalDeadlineWithNow(result.UrgencyThresholdText, now); thresholdErr == nil { + result.UrgencyThreshold = normalizeUrgencyThreshold(urgencyThreshold, result.Deadline) + } + } return result, nil } @@ -559,6 +594,26 @@ func extractJSONObject(text string) string { return text[start : end+1] } +// normalizeUrgencyThreshold 归一化“紧急分界线时间”。 +// +// 规则： +// 1. 分界线为空时直接返回空； +// 2. 存在 deadline 且分界线晚于 deadline 时，收敛到 deadline； +// 3. 其余情况保持原值。 +func normalizeUrgencyThreshold(threshold *time.Time, deadline *time.Time) *time.Time { + if threshold == nil { + return nil + } + if deadline == nil { + return threshold + } + if threshold.After(*deadline) { + normalized := *deadline + return &normalized + } + return threshold +} + func fallbackPriority(st *QuickNoteState) int { // 兜底规则： // 1) 有截止时间且 <=48h：重要且紧急； diff --git a/backend/agent/quicknote/prompt.go b/backend/agent/quicknote/prompt.go index a9a56f9..6226aa4 100644 --- a/backend/agent/quicknote/prompt.go +++ b/backend/agent/quicknote/prompt.go @@ -24,24 +24,26 @@ const ( 禁止输出任何其他内容。` // QuickNotePlanPrompt 用于“单请求聚合规划”： - // - 在一次调用内完成标题抽取、时间归一化、优先级评估、跟进句生成； + // - 在一次调用内完成标题抽取、时间归一化、紧急分界线评估、优先级评估、跟进句生成； // - 主要用于路由已明确命中 quick_note 的场景，以降低串行 LLM 调用次数。 // 额外说明： // 1) 强制 JSON 输出，减少后端解析分支复杂度； - // 2) deadline_at 统一分钟级，方便直接映射到数据库时间字段； + // 2) deadline_at / urgency_threshold_at 统一分钟级，方便直接映射到数据库时间字段； // 3) banter 与事实分离，避免润色文案污染结构化字段。 QuickNotePlanPrompt = `你是 SmartFlow 的任务聚合规划器。 你将基于用户输入，一次性输出任务规划结果，供后端直接写库。 -必须完成以下四件事： +必须完成以下五件事： 1) 提取任务标题 title（简洁明确）。 2) 归一化截止时间 deadline_at（若存在时间线索，必须输出绝对时间）。 -3) 评估优先级 priority_group（1~4）。 -4) 生成一句轻松跟进句 banter（不超过30字）。 +3) 评估紧急分界时间 urgency_threshold_at（何时从不紧急象限自动平移到紧急象限，可为空）。 +4) 评估优先级 priority_group（1~4）。 +5) 生成一句轻松跟进句 banter（不超过30字）。 输出要求： - 仅输出 JSON，不要 markdown，不要解释。 - deadline_at 仅允许 "yyyy-MM-dd HH:mm" 或空字符串。 +- urgency_threshold_at 仅允许 "yyyy-MM-dd HH:mm" 或空字符串。 - priority_group 仅允许 1|2|3|4。 - banter 不得新增或修改任务事实（任务名、时间、优先级）。` @@ -58,13 +60,14 @@ const ( - 若不是，请明确返回“非随口记意图”。 - 不要声称已经写入数据库。` - // QuickNotePriorityPrompt 用于第二阶段：将任务归类到四象限优先级。 + // QuickNotePriorityPrompt 用于第二阶段：将任务归类到四象限优先级，并评估紧急分界线。 // 输出会直接映射到 tasks.priority（1~4），因此要求结果必须可解释。 // 这里强调“理由必须可解释”，是为了后续日志复盘时能看懂模型为何这么判。 QuickNotePriorityPrompt = `你是 SmartFlow 的任务优先级评估器。 根据任务内容、时间约束和执行成本，输出优先级 priority_group： 1=重要且紧急，2=重要不紧急，3=简单不重要，4=不简单不重要。 -请给出简短理由，理由必须可解释。` +请给出简短理由，理由必须可解释。 +若你认为该任务需要后续自动平移，请额外输出 urgency_threshold_at（绝对时间，yyyy-MM-dd HH:mm）；否则输出空字符串。` // QuickNoteReplyBanterPrompt 用于随口记成功后的“轻松跟进句”生成。 // 约束重点： diff --git a/backend/agent/quicknote/state.go b/backend/agent/quicknote/state.go index 0ad5074..8ca0e67 100644 --- a/backend/agent/quicknote/state.go +++ b/backend/agent/quicknote/state.go @@ -75,7 +75,14 @@ type QuickNoteState struct { ExtractedTitle string ExtractedDeadline *time.Time ExtractedDeadlineText string - ExtractedPriority int + // ExtractedUrgencyThreshold 表示“进入紧急象限的分界时间”。 + // + // 语义说明： + // 1. 该时间由模型规划后给出，并在后端做解析校验； + // 2. 到达该时间后，任务可在“读时派生 + 异步落库”链路中被自动平移； + // 3. 为空表示该任务不参与自动平移。 + ExtractedUrgencyThreshold *time.Time + ExtractedPriority int // ExtractedBanter 是聚合规划阶段生成的“轻松跟进句”。 // 该字段非空时，最终回复阶段可直接复用，避免再触发一次独立润色模型调用。 ExtractedBanter string diff --git a/backend/agent/quicknote/tool.go b/backend/agent/quicknote/tool.go index a4372d8..e08c5f5 100644 --- a/backend/agent/quicknote/tool.go +++ b/backend/agent/quicknote/tool.go @@ -96,14 +96,17 @@ type QuickNoteCreateTaskRequest struct { Title string PriorityGroup int DeadlineAt *time.Time + // UrgencyThresholdAt 是“进入紧急象限”的分界时间，允许为空。 + UrgencyThresholdAt *time.Time } // QuickNoteCreateTaskResult 是业务层返回给工具层的结构化结果。 type QuickNoteCreateTaskResult struct { - TaskID int - Title string - PriorityGroup int - DeadlineAt *time.Time + TaskID int + Title string + PriorityGroup int + DeadlineAt *time.Time + UrgencyThresholdAt *time.Time } // QuickNoteCreateTaskToolInput 是提供给大模型的工具参数定义。 @@ -114,6 +117,9 @@ type QuickNoteCreateTaskToolInput struct { PriorityGroup int `json:"priority_group" jsonschema:"required,enum=1,enum=2,enum=3,enum=4,description=优先级分组(1重要且紧急,2重要不紧急,3简单不重要,4不简单不重要)"` // DeadlineAt 支持绝对时间与常见相对时间（如明天/后天/下周一/今晚），内部会归一化为绝对时间。 DeadlineAt string `json:"deadline_at,omitempty" jsonschema:"description=可选截止时间，支持RFC3339、yyyy-MM-dd HH:mm:ss、yyyy-MM-dd HH:mm 以及常见中文相对时间"` + // UrgencyThresholdAt 表示“何时从不紧急象限自动平移到紧急象限”。 + // 允许为空；非空时会走同样的时间解析与合法性校验。 + UrgencyThresholdAt string `json:"urgency_threshold_at,omitempty" jsonschema:"description=可选紧急分界时间，支持与deadline_at相同格式"` } // QuickNoteCreateTaskToolOutput 是返回给大模型的工具结果。 @@ -162,6 +168,10 @@ func BuildQuickNoteToolBundle(ctx context.Context, deps QuickNoteToolDeps) (*Qui if err != nil { return nil, err } + urgencyThresholdAt, err := parseOptionalDeadline(input.UrgencyThresholdAt) + if err != nil { + return nil, err + } // 2.3 user_id 一律来自鉴权上下文，不信任模型侧入参，防止越权写别人的任务。 userID, err := deps.ResolveUserID(ctx) @@ -174,10 +184,11 @@ func BuildQuickNoteToolBundle(ctx context.Context, deps QuickNoteToolDeps) (*Qui // 2.4 走业务层写库。 result, err := deps.CreateTask(ctx, QuickNoteCreateTaskRequest{ - UserID: userID, - Title: title, - PriorityGroup: input.PriorityGroup, - DeadlineAt: deadline, + UserID: userID, + Title: title, + PriorityGroup: input.PriorityGroup, + DeadlineAt: deadline, + UrgencyThresholdAt: urgencyThresholdAt, }) if err != nil { return nil, err diff --git a/backend/cmd/start.go b/backend/cmd/start.go index 45463b3..3e5940e 100644 --- a/backend/cmd/start.go +++ b/backend/cmd/start.go @@ -79,6 +79,9 @@ func Start() { if err = eventsvc.RegisterChatHistoryPersistHandler(eventBus, outboxRepo, manager); err != nil { log.Fatalf("Failed to register chat history event handler: %v", err) } + if err = eventsvc.RegisterTaskUrgencyPromoteHandler(eventBus, outboxRepo, manager); err != nil { + log.Fatalf("Failed to register task urgency promote event handler: %v", err) + } eventBus.Start(context.Background()) defer eventBus.Close() log.Println("Outbox event bus started") @@ -88,7 +91,7 @@ func Start() { // Service 层初始化。 userService := service.NewUserService(userRepo, cacheRepo) - taskSv := service.NewTaskService(taskRepo, cacheRepo) + taskSv := service.NewTaskService(taskRepo, cacheRepo, eventBus) courseService := service.NewCourseService(courseRepo, scheduleRepo) taskClassService := service.NewTaskClassService(taskClassRepo, cacheRepo, scheduleRepo, manager) scheduleService := service.NewScheduleService(scheduleRepo, userRepo, taskClassRepo, manager, cacheRepo) diff --git a/backend/dao/cache.go b/backend/dao/cache.go index 9aee0eb..d4b3fcf 100644 --- a/backend/dao/cache.go +++ b/backend/dao/cache.go @@ -19,44 +19,44 @@ func NewCacheDAO(client *redis.Client) *CacheDAO { return &CacheDAO{client: client} } -// SetBlacklist 把 Token 扔进黑名单 +// SetBlacklist 鎶?Token 鎵旇繘榛戝悕鍗? func (d *CacheDAO) SetBlacklist(jti string, expiration time.Duration) error { return d.client.Set(context.Background(), "blacklist:"+jti, "1", expiration).Err() } -// IsBlacklisted 检查 Token 是否在黑名单中 +// IsBlacklisted 妫€鏌?Token 鏄惁鍦ㄩ粦鍚嶅崟涓? func (d *CacheDAO) IsBlacklisted(jti string) (bool, error) { result, err := d.client.Get(context.Background(), "blacklist:"+jti).Result() if errors.Is(err, redis.Nil) { - return false, nil // 不在黑名单 + return false, nil // 涓嶅湪榛戝悕鍗? } else if err != nil { - return false, err // 其他错误 + return false, err // 鍏朵粬閿欒 } - return result == "1", nil // 在黑名单 + return result == "1", nil // 鍦ㄩ粦鍚嶅崟 } func (d *CacheDAO) AddTaskClassList(ctx context.Context, userID int, list *model.UserGetTaskClassesResponse) error { - // 1. 定义 Key，使用 userID 隔离不同用户的数据 + // 1. 瀹氫箟 Key锛屼娇鐢?userID 闅旂涓嶅悓鐢ㄦ埛鐨勬暟鎹? key := fmt.Sprintf("smartflow:task_classes:%d", userID) - // 2. 序列化：将结构体转为 []byte + // 2. 搴忓垪鍖栵細灏嗙粨鏋勪綋杞负 []byte data, err := json.Marshal(list) if err != nil { return err } - // 3. 存储：设置 30 分钟过期（根据业务灵活调整） + // 3. 瀛樺偍锛氳缃?30 鍒嗛挓杩囨湡锛堟牴鎹笟鍔＄伒娲昏皟鏁达級 return d.client.Set(ctx, key, data, 30*time.Minute).Err() } func (d *CacheDAO) GetTaskClassList(ctx context.Context, userID int) (*model.UserGetTaskClassesResponse, error) { key := fmt.Sprintf("smartflow:task_classes:%d", userID) var resp model.UserGetTaskClassesResponse - // 1. 从 Redis 获取字符串 + // 1. 浠?Redis 鑾峰彇瀛楃涓? val, err := d.client.Get(ctx, key).Result() if err != nil { - // 注意：如果是 redis.Nil，交给 Service 层处理查库逻辑 + // 娉ㄦ剰锛氬鏋滄槸 redis.Nil锛屼氦缁?Service 灞傚鐞嗘煡搴撻€昏緫 return &resp, err } - // 2. 反序列化：将 JSON 还原回结构体 + // 2. 鍙嶅簭鍒楀寲锛氬皢 JSON 杩樺師鍥炵粨鏋勪綋 err = json.Unmarshal([]byte(val), &resp) return &resp, err } @@ -69,9 +69,9 @@ func (d *CacheDAO) DeleteTaskClassList(ctx context.Context, userID int) error { func (d *CacheDAO) GetRecord(ctx context.Context, key string) (string, error) { val, err := d.client.Get(ctx, key).Result() if errors.Is(err, redis.Nil) { - return "", nil // 正常没命中的情况 + return "", nil // 姝ｅ父娌″懡涓殑鎯呭喌 } - return val, err // 真正的 Redis 报错 + return val, err // 鐪熸鐨?Redis 鎶ラ敊 } func (d *CacheDAO) SaveRecord(ctx context.Context, key string, val string, ttl time.Duration) error { @@ -86,18 +86,40 @@ func (d *CacheDAO) ReleaseLock(ctx context.Context, key string) error { return d.client.Del(ctx, key).Err() } -func (d *CacheDAO) GetUserTasksFromCache(ctx context.Context, userID int) ([]model.GetUserTaskResp, error) { +// GetUserTasksFromCache 读取用户任务缓存（内部模型版本）。 +// +// 职责边界： +// 1. 负责从 Redis 读取 `[]model.Task`，供 Service 层做“读时派生优先级”； +// 2. 不负责把模型转换成对外 DTO（该职责在 conv 层）； +// 3. 不负责缓存回填和缓存失效（回填由 Service 控制，失效由 GORM cache_deleter 统一处理）。 +// +// 输入输出语义： +// 1. 命中缓存时返回任务模型切片与 nil error； +// 2. 未命中时返回 redis.Nil，由上层决定是否回源 DB； +// 3. 反序列化失败时返回 error，避免把损坏缓存继续向后传播。 +func (d *CacheDAO) GetUserTasksFromCache(ctx context.Context, userID int) ([]model.Task, error) { key := fmt.Sprintf("smartflow:tasks:%d", userID) - var tasks []model.GetUserTaskResp + var tasks []model.Task val, err := d.client.Get(ctx, key).Result() if err != nil { - return nil, err // 注意：如果是 redis.Nil，交给 Service 层处理查库逻辑 + return nil, err // 娉ㄦ剰锛氬鏋滄槸 redis.Nil锛屼氦缁?Service 灞傚鐞嗘煡搴撻€昏緫 } err = json.Unmarshal([]byte(val), &tasks) return tasks, err } -func (d *CacheDAO) SetUserTasksToCache(ctx context.Context, userID int, tasks []model.GetUserTaskResp) error { +// SetUserTasksToCache 写入用户任务缓存（内部模型版本）。 +// +// 职责边界： +// 1. 负责把 DB 读取到的原始 `[]model.Task` 写入缓存； +// 2. 不负责对任务做“紧急性平移派生”，避免把派生结果写回缓存导致后续无法继续触发异步平移； +// 3. 不负责缓存删除，删除策略由 cache_deleter 在写库后触发。 +// +// 步骤说明： +// 1. 先把模型序列化为 JSON，确保 `urgency_threshold_at` 等字段完整保留； +// 2. 再写入固定 TTL 缓存，命中后可减少 DB 读取压力； +// 3. 若序列化失败立即返回 error，避免写入半结构化垃圾数据。 +func (d *CacheDAO) SetUserTasksToCache(ctx context.Context, userID int, tasks []model.Task) error { key := fmt.Sprintf("smartflow:tasks:%d", userID) data, err := json.Marshal(tasks) if err != nil { @@ -116,7 +138,7 @@ func (d *CacheDAO) GetUserTodayScheduleFromCache(ctx context.Context, userID int var schedules []model.UserTodaySchedule val, err := d.client.Get(ctx, key).Result() if err != nil { - return nil, err // 注意：如果是 redis.Nil，交给 Service 层处理查库逻辑 + return nil, err // 娉ㄦ剰锛氬鏋滄槸 redis.Nil锛屼氦缁?Service 灞傚鐞嗘煡搴撻€昏緫 } err = json.Unmarshal([]byte(val), &schedules) return schedules, err @@ -128,7 +150,7 @@ func (d *CacheDAO) SetUserTodayScheduleToCache(ctx context.Context, userID int, if err != nil { return err } - // 设置过期时间为当天剩余的时间，确保每天更新一次缓存 + // 璁剧疆杩囨湡鏃堕棿涓哄綋澶╁墿浣欑殑鏃堕棿锛岀‘淇濇瘡澶╂洿鏂颁竴娆＄紦瀛? return d.client.Set(ctx, key, data, time.Until(time.Date(time.Now().Year(), time.Now().Month(), time.Now().Day()+1, 0, 0, 0, 0, time.Now().Location()))).Err() } @@ -142,7 +164,7 @@ func (d *CacheDAO) GetUserWeeklyScheduleFromCache(ctx context.Context, userID in var schedules model.UserWeekSchedule val, err := d.client.Get(ctx, key).Result() if err != nil { - return nil, err // 注意：如果是 redis.Nil，交给 Service 层处理查库逻辑 + return nil, err // 娉ㄦ剰锛氬鏋滄槸 redis.Nil锛屼氦缁?Service 灞傚鐞嗘煡搴撻€昏緫 } err = json.Unmarshal([]byte(val), &schedules) return &schedules, err @@ -154,7 +176,7 @@ func (d *CacheDAO) SetUserWeeklyScheduleToCache(ctx context.Context, userID int, if err != nil { return err } - // 设置过期时间为一天 + // 璁剧疆杩囨湡鏃堕棿涓轰竴澶? return d.client.Set(ctx, key, data, 24*time.Hour).Err() } @@ -168,7 +190,7 @@ func (d *CacheDAO) GetUserRecentCompletedSchedulesFromCache(ctx context.Context, var resp model.UserRecentCompletedScheduleResponse val, err := d.client.Get(ctx, key).Result() if err != nil { - return &resp, err // 注意：如果是 redis.Nil，交给 Service 层处理查库逻辑 + return &resp, err // 娉ㄦ剰锛氬鏋滄槸 redis.Nil锛屼氦缁?Service 灞傚鐞嗘煡搴撻€昏緫 } err = json.Unmarshal([]byte(val), &resp) return &resp, err @@ -180,7 +202,7 @@ func (d *CacheDAO) SetUserRecentCompletedSchedulesToCache(ctx context.Context, u if err != nil { return err } - // 设置过期时间为30分钟 + // 璁剧疆杩囨湡鏃堕棿涓?0鍒嗛挓 return d.client.Set(ctx, key, data, 30*time.Minute).Err() } @@ -194,7 +216,7 @@ func (d *CacheDAO) DeleteUserRecentCompletedSchedulesFromCache(ctx context.Conte return err } if len(keys) > 0 { - // 用 UNLINK\(\) 异步删除，降低阻塞风险；如需强一致删除可改用 Del\(\) + // 鐢?UNLINK\(\) 寮傛鍒犻櫎锛岄檷浣庨樆濉為闄╋紱濡傞渶寮轰竴鑷村垹闄ゅ彲鏀圭敤 Del\(\) if err := d.client.Unlink(ctx, keys...).Err(); err != nil { return err } @@ -212,10 +234,10 @@ func (d *CacheDAO) GetUserOngoingScheduleFromCache(ctx context.Context, userID i var schedule model.OngoingSchedule val, err := d.client.Get(ctx, key).Result() if err != nil { - return &schedule, err // 注意：如果是 redis.Nil，交给 Service 层处理查库逻辑 + return &schedule, err // 娉ㄦ剰锛氬鏋滄槸 redis.Nil锛屼氦缁?Service 灞傚鐞嗘煡搴撻€昏緫 } if val == "null" { - return nil, nil // 之前缓存过没有正在进行的日程，直接返回 nil + return nil, nil // 涔嬪墠缂撳瓨杩囨病鏈夋鍦ㄨ繘琛岀殑鏃ョ▼锛岀洿鎺ヨ繑鍥?nil } err = json.Unmarshal([]byte(val), &schedule) return &schedule, err @@ -223,7 +245,7 @@ func (d *CacheDAO) GetUserOngoingScheduleFromCache(ctx context.Context, userID i func (d *CacheDAO) SetUserOngoingScheduleToCache(ctx context.Context, userID int, schedule *model.OngoingSchedule) error { if schedule == nil { - // 如果没有正在进行的日程，设置空值并短暂过期，避免频繁查库 + // 濡傛灉娌℃湁姝ｅ湪杩涜鐨勬棩绋嬶紝璁剧疆绌哄€煎苟鐭殏杩囨湡锛岄伩鍏嶉绻佹煡搴? key := fmt.Sprintf("smartflow:ongoing_schedule:%d", userID) return d.client.Set(ctx, key, "null", 5*time.Minute).Err() } @@ -232,7 +254,7 @@ func (d *CacheDAO) SetUserOngoingScheduleToCache(ctx context.Context, userID int if err != nil { return err } - // 设置过期时间为到 endTime 的剩余时间（若已过期则不写入缓存） + // 璁剧疆杩囨湡鏃堕棿涓哄埌 endTime 鐨勫墿浣欐椂闂达紙鑻ュ凡杩囨湡鍒欎笉鍐欏叆缂撳瓨锛? ttl := time.Until(schedule.EndTime) if ttl <= 0 { return nil diff --git a/backend/dao/task.go b/backend/dao/task.go index 17f696a..5ed4f75 100644 --- a/backend/dao/task.go +++ b/backend/dao/task.go @@ -1,6 +1,9 @@ package dao import ( + "context" + "time" + "github.com/LoveLosita/smartflow/backend/model" "github.com/LoveLosita/smartflow/backend/respond" "gorm.io/gorm" @@ -41,3 +44,65 @@ func (dao *TaskDAO) GetTasksByUserID(userID int) ([]model.Task, error) { } return tasks, nil } + +// PromoteTaskUrgencyByIDs 批量执行“任务紧急性平移”。 +// +// 职责边界： +// 1. 只负责把满足条件的任务从“不紧急象限”平移到“紧急象限”： +// 1.1 priority=2 -> 1（重要不紧急 -> 重要且紧急）； +// 1.2 priority=4 -> 3（不简单不重要 -> 简单不重要）； +// 2. 只更新本次指定 user_id + task_ids 范围内的数据； +// 3. 不负责事件发布、重试去重和缓存策略（由 Service/Outbox 负责）。 +// +// 幂等与一致性说明： +// 1. SQL 条件会限制 `is_completed=0`、`urgency_threshold_at<=now`、`priority IN (2,4)`； +// 2. 同一批任务重复调用时，已经平移过的记录不会再次更新（幂等）； +// 3. 使用 `Model(&model.Task{UserID:userID})` 是为了让 GORM 回调拿到 user_id，从而触发 cache_deleter 删除任务缓存。 +func (dao *TaskDAO) PromoteTaskUrgencyByIDs(ctx context.Context, userID int, taskIDs []int, now time.Time) (int64, error) { + // 1. 基础兜底：非法 user 或空任务列表直接无操作返回。 + if userID <= 0 || len(taskIDs) == 0 { + return 0, nil + } + + // 2. 去重并过滤非正数 ID，避免无效 where in 条件放大 SQL 噪音。 + validTaskIDs := compactPositiveIntIDs(taskIDs) + if len(validTaskIDs) == 0 { + return 0, nil + } + + // 3. 条件更新：只更新“已到紧急分界线且仍处于非紧急象限”的任务。 + result := dao.db.WithContext(ctx). + Model(&model.Task{UserID: userID}). + Where("user_id = ?", userID). + Where("id IN ?", validTaskIDs). + Where("is_completed = ?", false). + Where("urgency_threshold_at IS NOT NULL AND urgency_threshold_at <= ?", now). + Where("priority IN ?", []int{2, 4}). + Update("priority", gorm.Expr("CASE WHEN priority = 2 THEN 1 WHEN priority = 4 THEN 3 ELSE priority END")) + + if result.Error != nil { + return 0, result.Error + } + return result.RowsAffected, nil +} + +// compactPositiveIntIDs 对 int 切片做“去重 + 过滤非正数”。 +// +// 说明： +// 1. 该函数是 DAO 内部参数清洗工具，不参与任何业务判定； +// 2. 返回结果不保证稳定顺序，对当前 SQL where in 场景无影响。 +func compactPositiveIntIDs(ids []int) []int { + seen := make(map[int]struct{}, len(ids)) + result := make([]int, 0, len(ids)) + for _, id := range ids { + if id <= 0 { + continue + } + if _, exists := seen[id]; exists { + continue + } + seen[id] = struct{}{} + result = append(result, id) + } + return result +} diff --git a/backend/model/task.go b/backend/model/task.go index a208059..18a6af3 100644 --- a/backend/model/task.go +++ b/backend/model/task.go @@ -2,13 +2,43 @@ package model import "time" +// Task 是任务表的领域模型。 +// +// 职责边界： +// 1. 负责映射 tasks 表字段； +// 2. 不负责接口入参校验和业务规则判断； +// 3. 不负责“自动平移”执行（自动平移由 Service + Outbox 事件链路负责）。 type Task struct { - ID int `gorm:"primaryKey;autoIncrement"` - UserID int `gorm:"column:user_id;index"` - Title string `gorm:"type:varchar(255)"` - Priority int `gorm:"not null"` - IsCompleted bool `gorm:"column:is_completed;default:false"` - DeadlineAt *time.Time `gorm:"column:deadline_at"` + // 1. 主键。 + ID int `gorm:"primaryKey;autoIncrement"` + // 2. 归属用户 ID。 + // 2.1 单列索引用于常规按用户查任务； + // 2.2 同时参与“懒触发平移”复合索引的最左前缀。 + UserID int `gorm:"column:user_id;index;index:idx_user_done_threshold_priority,priority:1"` + // 3. 任务标题。 + Title string `gorm:"type:varchar(255)"` + // 4. 四象限优先级： + // 4.1 1=重要且紧急； + // 4.2 2=重要不紧急； + // 4.3 3=简单不重要； + // 4.4 4=不简单不重要。 + // + // 说明：该字段参与“懒触发平移”复合索引。 + Priority int `gorm:"not null;index:idx_user_done_threshold_priority,priority:4"` + // 5. 完成状态。 + // + // 说明：已完成任务不参与自动平移；该字段参与复合索引。 + IsCompleted bool `gorm:"column:is_completed;default:false;index:idx_user_done_threshold_priority,priority:2"` + // 6. 任务业务截止时间。 + DeadlineAt *time.Time `gorm:"column:deadline_at"` + // 7. 紧急分界时间（自动平移阈值）。 + // + // 规则： + // 7.1 到达该时间后，任务可从“不紧急象限”自动平移到“紧急象限”； + // 7.2 该值由上游（例如 LLM 规划）给出，不在模型层做推断； + // 7.3 为空表示该任务不参与自动平移； + // 7.4 该字段参与“懒触发平移”复合索引。 + UrgencyThresholdAt *time.Time `gorm:"column:urgency_threshold_at;index:idx_user_done_threshold_priority,priority:3"` } type UserAddTaskResponse struct { @@ -35,3 +65,15 @@ type GetUserTaskResp struct { Deadline string `json:"deadline"` IsCompleted bool `json:"is_completed"` } + +// TaskUrgencyPromoteRequestedPayload 是“任务紧急性平移请求”事件载荷。 +// +// 职责边界： +// 1. 只承载“哪个用户的哪些任务需要尝试平移”； +// 2. 不包含 outbox/kafka 协议字段（这些由基础设施层统一封装）； +// 3. TriggeredAt 只用于追踪触发时间，最终是否更新仍以消费时数据库条件为准。 +type TaskUrgencyPromoteRequestedPayload struct { + UserID int `json:"user_id"` + TaskIDs []int `json:"task_ids"` + TriggeredAt time.Time `json:"triggered_at"` +} diff --git a/backend/service/agentsvc/agent_quick_note.go b/backend/service/agentsvc/agent_quick_note.go index 516d7b6..ad41b16 100644 --- a/backend/service/agentsvc/agent_quick_note.go +++ b/backend/service/agentsvc/agent_quick_note.go @@ -124,11 +124,12 @@ func (s *AgentService) tryHandleQuickNoteWithGraph( CreateTask: func(ctx context.Context, req quicknote.QuickNoteCreateTaskRequest) (*quicknote.QuickNoteCreateTaskResult, error) { // 3.2.1 把 quick note 的工具入参映射成项目 Task 模型。 taskModel := &model.Task{ - UserID: req.UserID, - Title: req.Title, - Priority: req.PriorityGroup, - IsCompleted: false, - DeadlineAt: req.DeadlineAt, + UserID: req.UserID, + Title: req.Title, + Priority: req.PriorityGroup, + IsCompleted: false, + DeadlineAt: req.DeadlineAt, + UrgencyThresholdAt: req.UrgencyThresholdAt, } // 3.2.2 调用 DAO 写库。 @@ -139,10 +140,11 @@ func (s *AgentService) tryHandleQuickNoteWithGraph( // 3.2.3 把写库结果回填给 graph 状态，用于后续回复拼装。 return &quicknote.QuickNoteCreateTaskResult{ - TaskID: created.ID, - Title: created.Title, - PriorityGroup: created.Priority, - DeadlineAt: created.DeadlineAt, + TaskID: created.ID, + Title: created.Title, + PriorityGroup: created.Priority, + DeadlineAt: created.DeadlineAt, + UrgencyThresholdAt: created.UrgencyThresholdAt, }, nil }, }, diff --git a/backend/service/events/task_urgency_promote.go b/backend/service/events/task_urgency_promote.go new file mode 100644 index 0000000..a4a8ca0 --- /dev/null +++ b/backend/service/events/task_urgency_promote.go @@ -0,0 +1,136 @@ +package events + +import ( + "context" + "encoding/json" + "errors" + "log" + "strconv" + "time" + + "github.com/LoveLosita/smartflow/backend/dao" + kafkabus "github.com/LoveLosita/smartflow/backend/infra/kafka" + outboxinfra "github.com/LoveLosita/smartflow/backend/infra/outbox" + "github.com/LoveLosita/smartflow/backend/model" + "gorm.io/gorm" +) + +const ( + // EventTypeTaskUrgencyPromoteRequested 是“任务紧急性平移请求”事件类型。 + // + // 命名约束： + // 1. 只表达业务语义，不泄露 Kafka/outbox 技术细节； + // 2. 作为稳定路由键长期保留，后续协议演进优先走 event_version。 + EventTypeTaskUrgencyPromoteRequested = "task.urgency.promote.requested" +) + +// RegisterTaskUrgencyPromoteHandler 注册“任务紧急性平移”消费者处理器。 +// +// 职责边界： +// 1. 只负责注册 handler，不负责启动/关闭事件总线； +// 2. 只处理 `task.urgency.promote.requested` 事件，不处理其他业务事件； +// 3. 通过 `ConsumeAndMarkConsumed` 把“业务更新 + outbox consumed 推进”放进同一事务。 +func RegisterTaskUrgencyPromoteHandler( + bus *outboxinfra.EventBus, + outboxRepo *outboxinfra.Repository, + repoManager *dao.RepoManager, +) error { + // 1. 依赖校验：缺少任意关键依赖都不能安全消费消息。 + if bus == nil { + return errors.New("event bus is nil") + } + if outboxRepo == nil { + return errors.New("outbox repository is nil") + } + if repoManager == nil { + return errors.New("repo manager is nil") + } + + // 2. 定义统一处理函数。 + handler := func(ctx context.Context, envelope kafkabus.Envelope) error { + // 2.1 先解析 payload；解析失败属于不可恢复错误，直接标记 dead。 + var payload model.TaskUrgencyPromoteRequestedPayload + if unmarshalErr := json.Unmarshal(envelope.Payload, &payload); unmarshalErr != nil { + _ = outboxRepo.MarkDead(ctx, envelope.OutboxID, "解析任务紧急性平移载荷失败: "+unmarshalErr.Error()) + return nil + } + + // 2.2 做轻量参数净化，避免脏数据进入 DAO。 + payload.TaskIDs = sanitizePositiveUniqueIntIDs(payload.TaskIDs) + if payload.UserID <= 0 || len(payload.TaskIDs) == 0 { + _ = outboxRepo.MarkDead(ctx, envelope.OutboxID, "任务紧急性平移载荷无效: user_id 或 task_ids 非法") + return nil + } + + // 2.3 统一走 outbox 消费事务入口，保证“业务成功 -> consumed”原子一致。 + return outboxRepo.ConsumeAndMarkConsumed(ctx, envelope.OutboxID, func(tx *gorm.DB) error { + // 2.3.1 基于同一 tx 构造 RepoManager，复用现有跨 DAO 事务模式。 + txM := repoManager.WithTx(tx) + // 2.3.2 以消费时刻为准做条件更新，确保“到线”判定与真实落库时刻一致。 + updated, err := txM.Task.PromoteTaskUrgencyByIDs(ctx, payload.UserID, payload.TaskIDs, time.Now()) + if err != nil { + return err + } + log.Printf("任务紧急性平移消费完成: user_id=%d task_count=%d affected=%d outbox_id=%d", payload.UserID, len(payload.TaskIDs), updated, envelope.OutboxID) + return nil + }) + } + + // 3. 注册事件处理器。 + return bus.RegisterEventHandler(EventTypeTaskUrgencyPromoteRequested, handler) +} + +// PublishTaskUrgencyPromoteRequested 发布“任务紧急性平移请求”事件。 +// +// 职责边界： +// 1. 只负责把业务 DTO 发布到 outbox，不负责等待消费结果； +// 2. 若发布失败，返回 error 交给调用方决定是否降级或重试。 +func PublishTaskUrgencyPromoteRequested( + ctx context.Context, + publisher outboxinfra.EventPublisher, + payload model.TaskUrgencyPromoteRequestedPayload, +) error { + if publisher == nil { + return errors.New("event publisher is nil") + } + if payload.UserID <= 0 { + return errors.New("invalid user_id") + } + payload.TaskIDs = sanitizePositiveUniqueIntIDs(payload.TaskIDs) + if len(payload.TaskIDs) == 0 { + return errors.New("task_ids is empty") + } + if payload.TriggeredAt.IsZero() { + payload.TriggeredAt = time.Now() + } + + return publisher.Publish(ctx, outboxinfra.PublishRequest{ + EventType: EventTypeTaskUrgencyPromoteRequested, + EventVersion: outboxinfra.DefaultEventVersion, + // 这里使用 user_id 作为消息键，确保同一用户相关平移事件尽量落到同一分区，降低乱序概率。 + MessageKey: strconv.Itoa(payload.UserID), + AggregateID: strconv.Itoa(payload.UserID), + Payload: payload, + }) +} + +// sanitizePositiveUniqueIntIDs 过滤非正数并去重。 +// +// 说明： +// 1. 该函数只做参数净化，不承载业务判定； +// 2. 不保证顺序稳定，对当前 SQL where in 语义无影响。 +func sanitizePositiveUniqueIntIDs(ids []int) []int { + seen := make(map[int]struct{}, len(ids)) + result := make([]int, 0, len(ids)) + for _, id := range ids { + if id <= 0 { + continue + } + if _, exists := seen[id]; exists { + continue + } + seen[id] = struct{}{} + result = append(result, id) + } + return result +} diff --git a/backend/service/task.go b/backend/service/task.go index 599b41e..e50e18b 100644 --- a/backend/service/task.go +++ b/backend/service/task.go @@ -3,69 +3,281 @@ package service import ( "context" "errors" + "fmt" "log" + "time" "github.com/LoveLosita/smartflow/backend/conv" "github.com/LoveLosita/smartflow/backend/dao" + outboxinfra "github.com/LoveLosita/smartflow/backend/infra/outbox" "github.com/LoveLosita/smartflow/backend/model" "github.com/LoveLosita/smartflow/backend/respond" + eventsvc "github.com/LoveLosita/smartflow/backend/service/events" "github.com/go-redis/redis/v8" ) +const ( + // taskUrgencyPromoteDedupeTTL 是“同一任务平移请求”的去重锁有效期。 + // + // 设计考虑： + // 1. 太短会导致消费稍慢时被重复投递； + // 2. 太长会导致首次投递失败后恢复变慢； + // 3. 这里先取 120 秒作为折中值，后续可按线上观测再调优。 + taskUrgencyPromoteDedupeTTL = 120 * time.Second + // taskUrgencyPromoteDedupeKeyFmt 是任务平移去重键模板。 + taskUrgencyPromoteDedupeKeyFmt = "smartflow:task:promote:pending:%d:%d" +) + type TaskService struct { - // 伸出手：准备接住 DAO - dao *dao.TaskDAO + // dao 负责任务表读写。 + dao *dao.TaskDAO + // cache 负责任务列表缓存与 Redis 去重锁能力。 cache *dao.CacheDAO + // eventPublisher 负责发布 outbox 事件（可能为空：例如未启用 Kafka/总线时）。 + eventPublisher outboxinfra.EventPublisher } -// NewTaskService 创建 TaskService 实例 -func NewTaskService(dao *dao.TaskDAO, cache *dao.CacheDAO) *TaskService { +// NewTaskService 创建 TaskService 实例。 +// +// 职责边界： +// 1. 只做依赖注入，不做连接可用性探测； +// 2. 允许 eventPublisher 为空（用于本地降级场景）。 +func NewTaskService(taskDAO *dao.TaskDAO, cacheDAO *dao.CacheDAO, eventPublisher outboxinfra.EventPublisher) *TaskService { return &TaskService{ - dao: dao, - cache: cache, + dao: taskDAO, + cache: cacheDAO, + eventPublisher: eventPublisher, } } +// AddTask 新增任务。 +// +// 职责边界： +// 1. 负责参数转换、优先级合法性校验与写库； +// 2. 不负责“紧急性自动平移”逻辑（该逻辑发生在任务读取时的懒触发链路）。 func (ts *TaskService) AddTask(ctx context.Context, req *model.UserAddTaskRequest, userID int) (*model.UserAddTaskResponse, error) { - //1. 调用 conv 层进行转换 + // 1. 把用户请求转换为内部模型，避免 API 层结构直接泄漏到 DAO。 taskModel := conv.UserAddTaskRequestToModel(req, userID) - //2.检查优先级是否合法 + // 2. 优先级范围校验：当前任务体系只允许 1~4。 if taskModel.Priority < 1 || taskModel.Priority >= 5 { return nil, respond.InvalidPriority } - //3. 调用 courseDAO 层进行数据持久化 + // 3. 写库。 createdTask, err := ts.dao.AddTask(taskModel) if err != nil { return nil, err } - //4. 调用 conv 层进行响应转换 + // 4. 返回对外响应 DTO。 response := conv.ModelToUserAddTaskResponse(createdTask) return response, nil } +// GetUserTasks 获取用户任务列表（含“读时紧急性派生”与“异步平移触发”）。 +// +// 核心流程（步骤化）： +// 1. 先读缓存，未命中再回源 DB，并把“原始模型”回填缓存； +// 2. 在内存里做“读时派生”：仅用于本次返回给前端，不直接改库； +// 3. 收集“已到紧急分界线且仍处于非紧急象限”的任务 ID； +// 4. 通过 Redis SETNX 去重后，发布 outbox 事件异步落库； +// 5. 无论发布成功与否，都优先返回本次派生结果，保证用户读体验。 +// +// 一致性策略： +// 1. 缓存里存的是原始任务，不是派生后的优先级； +// 2. 真实平移由异步消费者条件更新 DB； +// 3. DB 更新后由 cache_deleter 自动删缓存，下一次读取自然拿到新状态。 func (ts *TaskService) GetUserTasks(ctx context.Context, userID int) ([]model.GetUserTaskResp, error) { - //1. 先尝试从缓存获取数据 - cachedResp, err := ts.cache.GetUserTasksFromCache(ctx, userID) - if err == nil { - // 缓存命中，直接返回 - return cachedResp, nil - } - // 如果是 redis.Nil 错误，说明缓存未命中，我们继续查库 - if !errors.Is(err, redis.Nil) { - return nil, err // 其他错误，返回错误 - } - //2. 调用 courseDAO 层获取数据 - tasks, err := ts.dao.GetTasksByUserID(userID) + // 1. 读取原始任务模型（缓存优先，DB 兜底）。 + rawTasks, err := ts.getRawUserTasks(ctx, userID) if err != nil { return nil, err } - //3. 调用 conv 层进行响应转换 - response := conv.ModelToGetUserTasksResp(tasks) - //4. 将结果存入缓存，设置合理的过期时间（24h） - err = ts.cache.SetUserTasksToCache(ctx, userID, response) - if err != nil { - // 缓存写入失败，记录日志但不影响正常返回数据 - log.Printf("Failed to cache user tasks for userID %d: %v", userID, err) - } + + // 2. 读时派生：本次请求内把“已到线任务”映射到紧急象限，同时收集待异步落库任务 ID。 + derivedTasks, duePromoteTaskIDs := deriveTaskUrgencyForRead(rawTasks, time.Now()) + + // 3. 非阻断触发异步平移事件：发布失败不影响本次查询返回。 + ts.tryEnqueueTaskUrgencyPromote(ctx, userID, duePromoteTaskIDs) + + // 4. 最后统一走 conv 转 DTO，避免 API 层直接依赖内部模型。 + response := conv.ModelToGetUserTasksResp(derivedTasks) return response, nil } + +// getRawUserTasks 读取“原始任务模型”。 +// +// 职责边界： +// 1. 负责缓存命中/回源 DB/回填缓存； +// 2. 不做优先级派生，不做异步事件投递； +// 3. 缓存写失败只记日志，不阻断主流程。 +func (ts *TaskService) getRawUserTasks(ctx context.Context, userID int) ([]model.Task, error) { + // 1. 先查缓存：命中则直接返回。 + cachedTasks, err := ts.cache.GetUserTasksFromCache(ctx, userID) + if err == nil { + return cachedTasks, nil + } + + // 2. 非 redis.Nil 错误直接返回，避免掩盖真实故障。 + if !errors.Is(err, redis.Nil) { + return nil, err + } + + // 3. 缓存未命中回源 DB。 + dbTasks, err := ts.dao.GetTasksByUserID(userID) + if err != nil { + return nil, err + } + + // 4. 回填缓存（失败不阻断主链路）。 + if setErr := ts.cache.SetUserTasksToCache(ctx, userID, dbTasks); setErr != nil { + log.Printf("写入用户任务缓存失败: user_id=%d err=%v", userID, setErr) + } + return dbTasks, nil +} + +// deriveTaskUrgencyForRead 对任务做“读时紧急性派生”，并收集需要异步落库的任务 ID。 +// +// 职责边界： +// 1. 只在内存里改本次返回值，不写 DB； +// 2. 只做“到线且未完成任务”的优先级映射； +// 3. 不处理去重锁和事件发布。 +// +// 返回语义： +// 1. 第一个返回值：可直接用于响应前端的派生任务切片； +// 2. 第二个返回值：需要发“异步平移事件”的任务 ID 列表（可能为空）。 +func deriveTaskUrgencyForRead(tasks []model.Task, now time.Time) ([]model.Task, []int) { + // 1. 拷贝切片，避免修改调用方持有的原始数据。 + derived := make([]model.Task, len(tasks)) + copy(derived, tasks) + + pendingPromoteTaskIDs := make([]int, 0, len(derived)) + + // 2. 逐条判断是否满足“自动平移”条件。 + for idx := range derived { + current := &derived[idx] + + // 2.1 已完成任务不参与平移。 + if current.IsCompleted { + continue + } + // 2.2 没有分界线的任务不参与平移。 + if current.UrgencyThresholdAt == nil { + continue + } + // 2.3 尚未到分界线，不平移。 + if current.UrgencyThresholdAt.After(now) { + continue + } + + // 2.4 到线后，仅把“不紧急象限”平移到对应“紧急象限”。 + // 2.4.1 重要不紧急(2) -> 重要且紧急(1) + // 2.4.2 不简单不重要(4) -> 简单不重要(3) + switch current.Priority { + case 2: + current.Priority = 1 + pendingPromoteTaskIDs = append(pendingPromoteTaskIDs, current.ID) + case 4: + current.Priority = 3 + pendingPromoteTaskIDs = append(pendingPromoteTaskIDs, current.ID) + default: + // 2.4.3 其他优先级不处理（包含已经是 1/3 的情况）。 + } + } + return derived, pendingPromoteTaskIDs +} + +// tryEnqueueTaskUrgencyPromote 尝试发布“任务紧急性平移请求”事件。 +// +// 职责边界： +// 1. 负责 Redis 去重锁 + outbox 发布； +// 2. 不负责真正落库（由消费者负责）； +// 3. 发布失败时要释放本次抢到的去重锁，避免任务被长时间“误判已投递”。 +func (ts *TaskService) tryEnqueueTaskUrgencyPromote(ctx context.Context, userID int, taskIDs []int) { + // 1. 基础兜底：无发布器或无候选任务时直接返回。 + if ts.eventPublisher == nil || userID <= 0 || len(taskIDs) == 0 { + return + } + + // 2. 先做任务 ID 清洗，避免无效 ID 参与去重与发布。 + validTaskIDs := compactPositiveUniqueTaskIDs(taskIDs) + if len(validTaskIDs) == 0 { + return + } + + // 3. 逐个抢 SETNX 去重锁： + // 3.1 抢到锁才允许进入本次发布； + // 3.2 抢不到说明已有请求在途，本次跳过即可； + // 3.3 抢锁失败只记录日志，不中断主流程。 + lockedTaskIDs := make([]int, 0, len(validTaskIDs)) + lockedKeys := make([]string, 0, len(validTaskIDs)) + for _, taskID := range validTaskIDs { + lockKey := fmt.Sprintf(taskUrgencyPromoteDedupeKeyFmt, userID, taskID) + locked, lockErr := ts.cache.AcquireLock(ctx, lockKey, taskUrgencyPromoteDedupeTTL) + if lockErr != nil { + log.Printf("任务平移去重锁获取失败: user_id=%d task_id=%d err=%v", userID, taskID, lockErr) + continue + } + if !locked { + continue + } + lockedTaskIDs = append(lockedTaskIDs, taskID) + lockedKeys = append(lockedKeys, lockKey) + } + if len(lockedTaskIDs) == 0 { + return + } + + // 4. 发布 outbox 事件：这里只保证“成功入 outbox 或返回错误”，不等待消费者执行完成。 + publishErr := eventsvc.PublishTaskUrgencyPromoteRequested(ctx, ts.eventPublisher, model.TaskUrgencyPromoteRequestedPayload{ + UserID: userID, + TaskIDs: lockedTaskIDs, + TriggeredAt: time.Now(), + }) + if publishErr != nil { + // 4.1 失败回滚：释放本次抢到的去重锁，避免后续请求因误锁而无法再投递。 + ts.releaseTaskPromoteLocks(lockedKeys) + log.Printf("任务平移事件发布失败: user_id=%d task_ids=%v err=%v", userID, lockedTaskIDs, publishErr) + return + } + + log.Printf("任务平移事件已发布: user_id=%d task_ids=%v", userID, lockedTaskIDs) +} + +// releaseTaskPromoteLocks 释放任务平移去重锁。 +// +// 说明： +// 1. 仅用于“发布失败回滚”场景； +// 2. 使用 Background 避免请求上下文已取消时导致锁释放失败。 +func (ts *TaskService) releaseTaskPromoteLocks(lockKeys []string) { + if len(lockKeys) == 0 { + return + } + releaseCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + for _, key := range lockKeys { + if err := ts.cache.ReleaseLock(releaseCtx, key); err != nil { + log.Printf("任务平移去重锁释放失败: key=%s err=%v", key, err) + } + } +} + +// compactPositiveUniqueTaskIDs 对任务 ID 做“过滤非正数 + 去重”。 +// +// 职责边界： +// 1. 只做参数清洗； +// 2. 不承载业务规则判断。 +func compactPositiveUniqueTaskIDs(taskIDs []int) []int { + seen := make(map[int]struct{}, len(taskIDs)) + result := make([]int, 0, len(taskIDs)) + for _, taskID := range taskIDs { + if taskID <= 0 { + continue + } + if _, exists := seen[taskID]; exists { + continue + } + seen[taskID] = struct{}{} + result = append(result, taskID) + } + return result +} diff --git a/docs/功能决策记录/任务四象限_自动平移_决策记录.md b/docs/功能决策记录/任务四象限_自动平移_决策记录.md new file mode 100644 index 0000000..a95fe21 --- /dev/null +++ b/docs/功能决策记录/任务四象限_自动平移_决策记录.md @@ -0,0 +1,119 @@ +# 功能决策记录（FDR）：任务四象限自动平移（懒触发） +## 1. 基本信息 +- 记录编号：FDR-2026-03-TASK-URGENCY-PROMOTE +- 功能名称：任务四象限自动平移（不紧急 -> 紧急） +- 记录日期：2026-03-16 +- 决策状态：已采纳 +- 负责人：项目协作实现（你 + Codex） +- 关联模块： + - `backend/service/task.go` + - `backend/service/events/task_urgency_promote.go` + - `backend/dao/task.go` + - `backend/infra/outbox/*` + - `backend/middleware/cache_deleter.go` + +## 2. 背景与问题 +- 业务背景：任务在创建时属于某个四象限，随着时间推进，部分“不紧急”任务应自动进入“紧急”象限。 +- 关键约束： + - 不引入全局定时任务（避免资源常驻扫描）。 + - 已有缓存删除机制依赖 `cache_deleter`（GORM 写库后自动删缓存）。 + - 已有 outbox 总线可承载异步更新。 +- 原始问题： + - 如果“读时直接改库 + 删缓存”，会引入写放大与重复触发； + - 如果“只靠 outbox 异步更新”，在消费延迟窗口内二次访问可能读到旧缓存； + - 需要在一致性与性能之间取平衡，且用户体验不能差。 + +## 3. 决策目标 +- 目标 1：用户访问任务列表时，优先看到“符合当前时间语义”的象限结果。 +- 目标 2：真实数据修正异步执行，避免把写库延迟绑定到读接口。 +- 目标 3：避免重复投递与重复更新，控制总线与数据库压力。 +- 非目标： + - 本次不做定时批处理平台； + - 本次不做分布式多级缓存一致性框架； + - 本次不做任务系统分库分表改造。 + +## 4. 备选方案 +### 方案 A：定时任务全量扫描并更新 +- 描述：后台周期扫描 `tasks`，到线即批量更新优先级。 +- 优点：数据收敛快、语义直观。 +- 缺点：常驻资源开销高；扫描窗口与峰值冲突风险高；运维复杂度增加。 +- 成本：中高（需要调度、监控、限流与容错）。 + +### 方案 B：读时同步改库（请求内直接 UPDATE） +- 描述：用户读任务列表时，检测到到线任务后立即写库并删缓存。 +- 优点：实现简单，数据立刻落地。 +- 缺点：读接口变“读+写”；高并发时放大锁竞争；延迟直接转嫁给用户请求。 +- 成本：中（实现快，但后续稳定性成本高）。 + +### 方案 C（采纳）：读时派生 + SETNX 去重 + outbox 异步落库 +- 描述： + - 读请求先按 `urgency_threshold_at` 做“派生展示”（仅本次响应生效）； + - 对需要平移的任务做 Redis `SETNX` 去重； + - 去重成功后发布 outbox 事件，消费者异步条件更新 DB； + - DB 更新后由 `cache_deleter` 自动删缓存，完成最终收敛。 +- 优点： + - 用户先看到正确业务语义，不被异步延迟影响体感； + - 读链路不阻塞写库； + - 复用现有 outbox 与 cache_deleter，侵入小、可扩展。 +- 缺点： + - 派生视图与真实落库存在短暂时间差； + - 需要维护去重键 TTL 与消费可观测性。 +- 成本：中（实现复杂度适中，长期收益更高）。 + +## 5. 最终决策 +- 采纳方案：方案 C（读时派生 + 异步收敛）。 +- 关键理由： + - 最符合“性能优先 + 最终一致”目标； + - 与现有 outbox、缓存失效机制天然兼容； + - 面向后续通用事件总线扩展（更多异步业务）更友好。 + +## 6. 影响范围 +- 涉及模块： + - `TaskService.GetUserTasks`：新增读时派生与异步触发。 + - `TaskDAO.PromoteTaskUrgencyByIDs`：新增幂等条件更新。 + - `service/events`：新增任务平移事件发布与消费注册。 + - `cmd/start.go`：新增 handler 注册与 service 注入。 +- 数据与存储影响： + - `tasks` 新增字段：`urgency_threshold_at`。 + - 新增复合索引：`(user_id, is_completed, urgency_threshold_at, priority)`。 + - outbox 新增事件类型：`task.urgency.promote.requested`。 +- 接口影响： + - 对外 API 协议不变； + - 行为层面变为“读时先派生展示、随后异步收敛”。 +- 监控与日志影响： + - 新增任务平移事件发布/消费日志； + - 可按 `event_type` 维度观测积压与重试。 + +## 7. 风险与应对 +- 风险 1：outbox 消费延迟导致短时间内 DB 仍是旧值。 + - 应对策略：读时派生保证用户看到新语义；最终由异步消费收敛。 +- 风险 2：同一任务被多次重复投递。 + - 应对策略：`SETNX + TTL` 去重；DAO 层条件更新保证幂等。 +- 风险 3：缓存一致性混乱（误删/抖动）。 + - 应对策略：不在读链路主动删缓存；只在真实写库后由 `cache_deleter` 自动失效。 + +## 8. 验证与回滚 +- 验证方式： + - 用到线前后样例任务验证 `GetUserTasks` 返回象限变化； + - 验证 outbox 事件是否成功入队、消费并更新 DB； + - 验证 DB 更新后缓存是否由 `cache_deleter` 自动清理。 +- 成功判定标准： + - 用户可在到线后第一时间看到“已平移”结果； + - DB 在异步窗口后收敛为相同状态； + - 无明显重复投递和异常写放大。 +- 回滚方案： + - 关闭任务平移事件发布（保留读时派生）可快速止损； + - 或回退到“只读不平移”行为（临时关闭派生逻辑）。 + +## 9. 里程碑与后续计划 +- 里程碑 1：完成懒触发主链路与事件消费者落地（已完成）。 +- 里程碑 2：完成联调与基本回归（已完成）。 +- 后续优化项： + - 增加按 `event_type=task.urgency.promote.requested` 的积压监控面板； + - 根据线上情况调优去重 TTL； + - 引入批量聚合发布（按用户或时间窗口）进一步降事件量。 + +## 10. 复盘结论（上线后补充） +- 实际效果：待补充。 +- 与预期偏差：待补充。 +- 是否需要二次决策：待补充。