past-go/past-go/internal/notify/notify.go

// 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",
	"couple_bind", "smoking_consume",
}

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": "完成了所有宠物任务",
	"couple_bind": "同意了你的邀请", "smoking_consume": "记录了抽烟",
}

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",
	"couple_bind": "pages/index/index", "smoking_consume": "pages/punch/punch",
}

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)
	}
}

type Sender interface {
	SendSubscribeMessage(toUserID string, templateType string, data map[string]string, page string, activityID string, userID string) bool
}

// force 只放宽时间和次数限制，仍检查成功记录、关系和分享权限。
func Once(db *store.DB, wx Sender, force bool) {
	ph := strings.TrimSuffix(strings.Repeat("?,", len(actionTypes)), ",")
	args := make([]any, 0, len(actionTypes)+3)
	for _, action := range actionTypes {
		args = append(args, action)
	}
	query := `SELECT a.*, u.nickname, recipient.id AS to_user_id FROM activity_log a
	 JOIN users u ON u.id=a.user_id
	 JOIN couples cp ON cp.id=a.couple_id AND cp.status='active' AND u.couple_id=cp.id
	 JOIN users recipient ON recipient.id=CASE WHEN cp.user1_id=u.id THEN cp.user2_id ELSE cp.user1_id END AND recipient.couple_id=cp.id
	 JOIN subscribe_templates st ON st.user_id=recipient.id AND st.template_type='notify'
	 WHERE a.action_type IN (` + ph + `) AND u.id IN (cp.user1_id, cp.user2_id)
	 AND NOT EXISTS (SELECT 1 FROM notify_log nl WHERE nl.activity_id=a.id AND nl.to_user_id=recipient.id AND nl.status='success')
	 AND NOT EXISTS (SELECT 1 FROM subscribe_message_logs sl WHERE sl.activity_id=a.id AND sl.to_user_id=recipient.id AND sl.status='success')
	 AND (a.action_type NOT IN ('period_add','period_sex_add') OR COALESCE(u.period_visible,0)<>0)`
	if !force {
		query += ` AND a.created_at >= ? AND (SELECT COUNT(*) FROM subscribe_message_logs sl WHERE sl.activity_id=a.id AND sl.to_user_id=recipient.id) < ?`
		args = append(args, time.Now().Add(-scanMinutes*time.Minute).Format("2006-01-02 15:04:05"), maxRetries)
	}
	query += " ORDER BY a.created_at DESC, a.id DESC LIMIT ?"
	args = append(args, maxBatch)
	activities, err := db.Query(query, args...)
	if err != nil {
		slog.Error("cron_notify 查询失败", "err", err)
		return
	}
	processed, success, fail, skip := 0, 0, 0, 0
	for _, a := range activities {
		activityID, userID, actionType := a.Str("id"), a.Str("user_id"), a.Str("action_type")
		nickname := a.Str("nickname")
		if nickname == "" {
			nickname = "你的TA"
		}
		partnerID := a.Str("to_user_id")
		// 扫描后共享开关或关系可能改变；实际发送前再次检查。
		allowed, err := db.QueryOne(`SELECT u.period_visible FROM users u JOIN couples cp ON cp.id=u.couple_id AND cp.status='active'
		 JOIN users recipient ON recipient.id=? AND recipient.couple_id=cp.id
		 WHERE u.id=? AND cp.id=? AND u.id IN (cp.user1_id,cp.user2_id) AND recipient.id IN (cp.user1_id,cp.user2_id) AND recipient.id<>u.id`, partnerID, userID, a.Str("couple_id"))
		if err != nil || allowed == nil || ((actionType == "period_add" || actionType == "period_sex_add") && allowed.Int("period_visible") == 0) {
			skip++
			continue
		}
		if !force {
			count, err := db.QueryOne("SELECT COUNT(*) as cnt FROM subscribe_message_logs WHERE activity_id=? AND to_user_id=?", activityID, partnerID)
			if err != nil || count == nil || count.Int("cnt") >= maxRetries {
				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)
		processed++
		if ok {
			success++
		} else {
			fail++
		}
	}
	slog.Info("cron_notify 处理完成", "processed", processed, "success", success, "fail", fail, "skip", skip)
}