init: initial commit

2026-10-07 09:28:43 +08:00
commit a68a4ef411
251 changed files with 38279 additions and 0 deletions
@@ -0,0 +1,162 @@
+// Package notify 对应 cron_notify.php：定时扫描 activity_log，补发未成功的订阅消息。
+// PHP 版本靠外部 cron 触发；Go 版本在进程内按固定间隔运行。
+package notify
+
+import (
+	"log/slog"
+	"strings"
+	"time"
+
+	"past/internal/store"
+	"past/internal/wechat"
+)
+
+const (
+	maxBatch    = 20
+	maxRetries  = 3
+	scanMinutes = 30
+)
+
+var actionTypes = []string{
+	"login", "timeline_create", "album_create", "photo_upload", "diary_create", "diary_comment_add",
+	"song_add", "checkin_do", "period_add", "period_sex_add", "pet_adopt", "pet_daily", "pet_rename", "pet_all_done",
+}
+
+var actionLabels = map[string]string{
+	"login": "登录了", "timeline_create": "新增了记忆日", "album_create": "创建了新相册", "photo_upload": "上传了新照片",
+	"diary_create": "写了新的日记", "diary_comment_add": "评论了日记", "song_add": "添加了新歌曲", "checkin_do": "完成了打卡",
+	"period_add": "记录了姨妈期", "period_sex_add": "记录了亲密时光", "pet_adopt": "领养了新宠物", "pet_daily": "照顾了宠物",
+	"pet_rename": "给宠物改了名", "pet_all_done": "完成了所有宠物任务",
+}
+
+var pageMap = map[string]string{
+	"login": "pages/index/index", "timeline_create": "pages/timeline/timeline", "album_create": "pages/albums/albums",
+	"photo_upload": "pages/albums/albums", "diary_create": "pages/diary/diary", "diary_comment_add": "pages/diary/diary",
+	"song_add": "pages/lovewords/lovewords", "checkin_do": "pages/punch/punch", "period_add": "pages/period/period",
+	"period_sex_add": "pages/period/period", "pet_adopt": "pages/pet/pet", "pet_daily": "pages/pet/pet",
+	"pet_rename": "pages/pet/pet", "pet_all_done": "pages/pet/pet",
+}
+
+func mbSub(s string, n int) string {
+	r := []rune(s)
+	if len(r) > n {
+		return string(r[:n])
+	}
+	return s
+}
+
+// Run 按 interval 周期执行一次补发（阻塞，需 go 调用）
+func Run(db *store.DB, wx *wechat.Client, interval time.Duration) {
+	for {
+		func() {
+			defer func() {
+				if r := recover(); r != nil {
+					slog.Error("cron_notify panic", "err", r)
+				}
+			}()
+			Once(db, wx, false)
+		}()
+		time.Sleep(interval)
+	}
+}
+
+// Once 执行一次扫描；force=true 时不限时间且忽略重试上限（对应 --force）
+func Once(db *store.DB, wx *wechat.Client, force bool) {
+	now := time.Now().Format("2006-01-02 15:04:05")
+	ph := strings.TrimSuffix(strings.Repeat("?,", len(actionTypes)), ",")
+	args := make([]any, 0, len(actionTypes)+1)
+	for _, a := range actionTypes {
+		args = append(args, a)
+	}
+	var sql string
+	if force {
+		sql = "SELECT a.*, u.nickname, u.couple_id as user_couple_id FROM activity_log a JOIN users u ON a.user_id = u.id WHERE a.action_type IN (" + ph + ") AND a.id NOT IN (SELECT nl.activity_id FROM notify_log nl WHERE nl.status = 'success') AND a.couple_id IS NOT NULL ORDER BY a.created_at DESC LIMIT 20"
+	} else {
+		sql = "SELECT a.*, u.nickname, u.couple_id as user_couple_id FROM activity_log a JOIN users u ON a.user_id = u.id WHERE a.action_type IN (" + ph + ") AND a.created_at >= ? AND a.id NOT IN (SELECT nl.activity_id FROM notify_log nl WHERE nl.status = 'success') AND a.couple_id IS NOT NULL ORDER BY a.created_at DESC LIMIT 20"
+		args = append(args, time.Now().Add(-scanMinutes*time.Minute).Format("2006-01-02 15:04:05"))
+	}
+	activities, err := db.Query(sql, args...)
+	if err != nil {
+		slog.Error("cron_notify 查询失败", "err", err)
+		return
+	}
+	if len(activities) == 0 {
+		return
+	}
+	processed, success, fail, skip := 0, 0, 0, 0
+	for _, a := range activities {
+		activityID, coupleID, userID, actionType := a.Int("id"), a.Int("couple_id"), a.Int("user_id"), a.Str("action_type")
+		nickname := a.Str("nickname")
+		if nickname == "" {
+			nickname = "你的TA"
+		}
+		partnerID := partnerUserID(db, userID)
+		if partnerID == 0 {
+			skip++
+			continue
+		}
+		cnt, _ := db.QueryOne("SELECT COUNT(*) as cnt FROM notify_log WHERE activity_id = ? AND to_user_id = ?", activityID, partnerID)
+		retry := int64(0)
+		if cnt != nil {
+			retry = cnt.Int("cnt")
+		}
+		if retry >= maxRetries && !force {
+			skip++
+			continue
+		}
+		// buildNotifyData
+		date := a.Str("created_at")
+		if len(date) >= 10 {
+			date = date[:10]
+		}
+		desc := a.Str("description")
+		label, ok := actionLabels[actionType]
+		if !ok {
+			label = "有了新操作"
+		}
+		page, ok := pageMap[actionType]
+		if !ok {
+			page = "pages/index/index"
+		}
+		thing1 := nickname + " " + label
+		thing2 := desc
+		if thing2 == "" {
+			thing2 = thing1
+		}
+		data := map[string]string{"thing1": mbSub(thing1, 20), "thing2": mbSub(thing2, 20), "date3": date, "thing4": "点击查看详情 💕"}
+
+		ok = wx.SendSubscribeMessage(partnerID, "notify", data, page, activityID, userID)
+		status, errMsg := "success", ""
+		if !ok {
+			status, errMsg = "failed", "发送失败（可能用户未订阅或微信接口返回错误）"
+		}
+		if retry == 0 {
+			db.Exec("INSERT INTO notify_log (activity_id, couple_id, user_id, action_type, template_type, to_user_id, status, error_msg, sent_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+				activityID, coupleID, userID, actionType, "notify", partnerID, status, errMsg, now)
+		} else {
+			db.Exec("UPDATE notify_log SET status = ?, error_msg = ?, sent_at = ? WHERE activity_id = ? AND to_user_id = ?", status, errMsg, now, activityID, partnerID)
+		}
+		processed++
+		if ok {
+			success++
+		} else {
+			fail++
+		}
+	}
+	slog.Info("cron_notify 处理完成", "processed", processed, "success", success, "fail", fail, "skip", skip)
+}
+
+func partnerUserID(db *store.DB, userID int64) int64 {
+	u, _ := db.QueryOne("SELECT couple_id FROM users WHERE id = ?", userID)
+	if u == nil || u.Int("couple_id") == 0 {
+		return 0
+	}
+	cp, _ := db.QueryOne("SELECT user1_id, user2_id FROM couples WHERE id = ?", u.Int("couple_id"))
+	if cp == nil {
+		return 0
+	}
+	if cp.Int("user1_id") == userID {
+		return cp.Int("user2_id")
+	}
+	return cp.Int("user1_id")
+}