174 lines
6.0 KiB
Go
174 lines
6.0 KiB
Go
package logic
|
||
|
||
import (
|
||
"context"
|
||
"strconv"
|
||
"time"
|
||
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/os/gtime"
|
||
|
||
"tool-api/internal/consts"
|
||
"tool-api/internal/model/entity"
|
||
)
|
||
|
||
// ============================================================================
|
||
// 笔记提醒扫描(gcron 回调,建议每分钟执行)
|
||
//
|
||
// 依据:架构 §4.2 图②、§8.6.4。
|
||
// 并发安全核心:条件更新抢占 —— UPDATE notes SET remind_status=2
|
||
// WHERE id=? AND remind_status=1
|
||
// 仅 rowsAffected==1 的实例继续发送,多实例/重入不会重复发送。
|
||
//
|
||
// 状态流转(remind_status):
|
||
// 1 待提醒 → (抢占) 2 已提醒 → 发送成功:记 remind_sent_at
|
||
// → 发送失败/授权不足/模板未配置:置 3
|
||
// 1 待提醒 且超时过久 → 4 已过期
|
||
// ============================================================================
|
||
|
||
// noteRemindScanBatch 单轮扫描上限,避免一次拉太多
|
||
const noteRemindScanBatch = 100
|
||
|
||
// ScanDueReminders 扫描到点的提醒并发送(gcron 回调)。
|
||
func ScanDueReminders(ctx context.Context) {
|
||
now := gtime.Now()
|
||
|
||
// 1) 先把「过期过久」的待提醒置为已过期(4),避免历史脏数据被无限重扫
|
||
expireHours := noteRemindExpireHours(ctx)
|
||
if _, err := g.Model(consts.TableNotes).
|
||
Where("remind_status", consts.NoteRemindPending).
|
||
WhereLT("remind_at", now.Add(-time.Duration(expireHours)*time.Hour)).
|
||
Data(g.Map{"remind_status": consts.NoteRemindExpired, "updated_at": now}).
|
||
Update(); err != nil {
|
||
g.Log().Warningf(ctx, "[note-remind] 标记过期失败: %v", err)
|
||
}
|
||
|
||
// 2) 取到点的待提醒
|
||
records, err := g.Model(consts.TableNotes).
|
||
Where("remind_status", consts.NoteRemindPending).
|
||
WhereLTE("remind_at", now).
|
||
OrderAsc("remind_at").
|
||
Limit(noteRemindScanBatch).
|
||
All()
|
||
if err != nil {
|
||
g.Log().Warningf(ctx, "[note-remind] 扫描到点提醒失败: %v", err)
|
||
return
|
||
}
|
||
for _, r := range records {
|
||
note := &entity.Notes{}
|
||
if err = r.Struct(note); err != nil {
|
||
g.Log().Warningf(ctx, "[note-remind] 解析笔记失败: %v", err)
|
||
continue
|
||
}
|
||
processDueRemind(ctx, note)
|
||
}
|
||
}
|
||
|
||
// processDueRemind 处理单条到点提醒:抢占 → 校验 → 扣减授权 → 发送。
|
||
func processDueRemind(ctx context.Context, note *entity.Notes) {
|
||
// 条件更新抢占:仅 rowsAffected==1 的实例继续(并发/多实例下只发一次)
|
||
result, err := g.Model(consts.TableNotes).
|
||
Where("id", note.Id).
|
||
Where("remind_status", consts.NoteRemindPending).
|
||
Data(g.Map{"remind_status": consts.NoteRemindSent}).
|
||
Update()
|
||
if err != nil {
|
||
g.Log().Warningf(ctx, "[note-remind] 抢占提醒失败 noteId=%d: %v", note.Id, err)
|
||
return
|
||
}
|
||
affected, _ := result.RowsAffected()
|
||
if !shouldClaimRemind(affected) {
|
||
// 已被其它实例/轮次抢到,跳过(保证「同一提醒只发一次」)
|
||
return
|
||
}
|
||
|
||
// 已抢占。模板未配置 → 降级:不发送,置为发送失败/未发送(站内标记由 remind_status 呈现)
|
||
if RemindTemplateId(ctx) == "" {
|
||
markRemindUnsent(ctx, note.Id, "模板未配置")
|
||
return
|
||
}
|
||
|
||
// 发送前条件扣减订阅授权次数(>=1 才减)
|
||
decResult, err := g.Model(consts.TableNotes).
|
||
Where("id", note.Id).
|
||
WhereGTE("subscribe_count", 1).
|
||
Decrement("subscribe_count", 1)
|
||
if err != nil {
|
||
markRemindUnsent(ctx, note.Id, "扣减授权异常:"+err.Error())
|
||
return
|
||
}
|
||
if decAffected, _ := decResult.RowsAffected(); decAffected != 1 {
|
||
markRemindUnsent(ctx, note.Id, "订阅授权次数不足")
|
||
return
|
||
}
|
||
|
||
user, err := getUserById(ctx, note.UserId)
|
||
if err != nil || user.Openid == "" {
|
||
markRemindUnsent(ctx, note.Id, "用户 openid 缺失")
|
||
return
|
||
}
|
||
|
||
if err = SendNoteRemind(ctx, NoteRemindSendReq{
|
||
NoteId: note.Id,
|
||
Openid: user.Openid,
|
||
Title: note.Title,
|
||
Content: note.Content,
|
||
Time: formatRemindTime(note.RemindAt),
|
||
}); err != nil {
|
||
markRemindFailed(ctx, note.Id, err)
|
||
return
|
||
}
|
||
|
||
now := gtime.Now()
|
||
if _, err = g.Model(consts.TableNotes).Where("id", note.Id).Data(g.Map{
|
||
"remind_status": consts.NoteRemindSent,
|
||
"remind_sent_at": now,
|
||
"updated_at": now,
|
||
}).Update(); err != nil {
|
||
g.Log().Warningf(ctx, "[note-remind] 记录发送时间失败 noteId=%d: %v", note.Id, err)
|
||
}
|
||
g.Log().Infof(ctx, "[note-remind] 已发送提醒 noteId=%d userId=%d", note.Id, note.UserId)
|
||
}
|
||
|
||
// markRemindUnsent 未发送(模板未配置 / 授权不足 / 其它前置失败):置 3 + 尝试次数 +1 + 站内标记日志。
|
||
func markRemindUnsent(ctx context.Context, noteId int64, reason string) {
|
||
g.Log().Warningf(ctx, "[note-remind] 未发送 noteId=%d 原因=%s", noteId, reason)
|
||
_, _ = g.Model(consts.TableNotes).Where("id", noteId).Data(g.Map{
|
||
"remind_status": consts.NoteRemindFailed,
|
||
"updated_at": gtime.Now(),
|
||
}).Update()
|
||
_, _ = g.Model(consts.TableNotes).Where("id", noteId).Increment("remind_attempts", 1)
|
||
}
|
||
|
||
// markRemindFailed 发送失败:置 3 + 尝试次数 +1,供前端/后台手动重试。
|
||
func markRemindFailed(ctx context.Context, noteId int64, cause error) {
|
||
g.Log().Warningf(ctx, "[note-remind] 发送失败 noteId=%d err=%v", noteId, cause)
|
||
_, _ = g.Model(consts.TableNotes).Where("id", noteId).Data(g.Map{
|
||
"remind_status": consts.NoteRemindFailed,
|
||
"updated_at": gtime.Now(),
|
||
}).Update()
|
||
_, _ = g.Model(consts.TableNotes).Where("id", noteId).Increment("remind_attempts", 1)
|
||
}
|
||
|
||
// shouldClaimRemind 条件更新抢占判定:仅当受影响行数为 1 时才继续发送。
|
||
func shouldClaimRemind(rowsAffected int64) bool {
|
||
return rowsAffected == 1
|
||
}
|
||
|
||
// noteRemindExpireHours 读取提醒过期阈值(小时),settings 未配置时用默认值。
|
||
func noteRemindExpireHours(ctx context.Context) int {
|
||
raw := SettingValue(ctx, consts.SettingNoteRemindExpireHours)
|
||
hours, err := strconv.Atoi(raw)
|
||
if err != nil || hours <= 0 {
|
||
return consts.DefaultNoteRemindExpireHours
|
||
}
|
||
return hours
|
||
}
|
||
|
||
func formatRemindTime(t *gtime.Time) string {
|
||
if t == nil || t.IsZero() {
|
||
return ""
|
||
}
|
||
return t.Format("Y-m-d H:i:s")
|
||
}
|