Fix subscription renewal and partner notification delivery

2026-10-11 10:32:30 +08:00
parent c5cab1b945
commit 31eae87a63
13 changed files with 282 additions and 115 deletions
@@ -63,28 +63,37 @@ func Run(db *store.DB, wx *wechat.Client, interval time.Duration) {
 	}
 }

-// Once 执行一次扫描；force=true 时不限时间且忽略重试上限（对应 --force）
-func Once(db *store.DB, wx *wechat.Client, force bool) {
+type Sender interface {
+	SendSubscribeMessage(toUserID int64, templateType string, data map[string]string, page string, activityID int64, userID int64) bool
+}
+
+// force 只放宽时间和次数限制，仍检查成功记录、关系和分享权限。
+func Once(db *store.DB, wx Sender, force bool) {
 	ph := strings.TrimSuffix(strings.Repeat("?,", len(actionTypes)), ",")
-	args := make([]any, 0, len(actionTypes)+1)
-	for _, a := range actionTypes {
-		args = append(args, a)
+	args := make([]any, 0, len(actionTypes)+3)
+	for _, action := range actionTypes {
+		args = append(args, action)
 	}
-	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) 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) 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"))
+	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)
 	}
-	activities, err := db.Query(sql, args...)
+	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
 	}
-	if len(activities) == 0 {
-		return
-	}
 	processed, success, fail, skip := 0, 0, 0, 0
 	for _, a := range activities {
 		activityID, userID, actionType := a.Int("id"), a.Int("user_id"), a.Str("action_type")
@@ -92,19 +101,21 @@ func Once(db *store.DB, wx *wechat.Client, force bool) {
 		if nickname == "" {
 			nickname = "你的TA"
 		}
-		partnerID := partnerUserID(db, userID)
-		if partnerID == 0 {
+		partnerID := a.Int("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.Int("couple_id"))
+		if err != nil || allowed == nil || ((actionType == "period_add" || actionType == "period_sex_add") && allowed.Int("period_visible") == 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
+		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")
@@ -137,18 +148,3 @@ func Once(db *store.DB, wx *wechat.Client, force bool) {
 	}
 	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")
-}
@@ -0,0 +1,68 @@
+package notify
+
+import (
+	"database/sql/driver"
+	"past/internal/store"
+	"past/internal/testdb"
+	"strings"
+	"testing"
+)
+
+type testSender struct {
+	calls    int
+	to, from int64
+}
+
+func (s *testSender) SendSubscribeMessage(to int64, typ string, data map[string]string, page string, activity, from int64) bool {
+	s.calls++
+	s.to, s.from = to, from
+	return true
+}
+
+func TestRetrySelectionAndPermissionRecheck(t *testing.T) {
+	for _, tt := range []struct {
+		name    string
+		visible int64
+		count   int64
+		force   bool
+		bound   bool
+		want    int
+	}{
+		{"retry failed event", 1, 1, false, true, 1},
+		{"retry cap", 1, 3, false, true, 0},
+		{"sharing switched off after scan", 0, 0, false, true, 0},
+		{"relationship changed after scan", 1, 0, false, false, 0},
+		{"force still checks sharing", 0, 0, true, true, 0},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			steps := []testdb.Step{{Match: "SELECT a.*", Check: func(query string, _ []any) error {
+				for _, part := range []string{"nl.status='success'", "sl.status='success'", "cp.status='active'", "u.couple_id=cp.id", "COALESCE(u.period_visible,0)<>0", "JOIN subscribe_templates"} {
+					if !strings.Contains(query, part) {
+						t.Errorf("missing guard %s", part)
+					}
+				}
+				if !tt.force && (!strings.Contains(query, "COUNT(*)") || !strings.Contains(query, "a.created_at >= ?")) {
+					t.Error("missing bounded retry")
+				}
+				return nil
+			}, Columns: []string{"id", "couple_id", "user_id", "to_user_id", "action_type", "created_at"}, Rows: [][]driver.Value{{int64(9), int64(5), int64(1), int64(2), "period_add", "2026-10-11 12:00:00"}}}}
+			allowed := [][]driver.Value(nil)
+			if tt.bound {
+				allowed = [][]driver.Value{{tt.visible}}
+			}
+			steps = append(steps, testdb.Step{Match: "SELECT u.period_visible", Args: []any{int64(2), int64(1), int64(5)}, Columns: []string{"period_visible"}, Rows: allowed})
+			if tt.bound && tt.visible != 0 && !tt.force {
+				steps = append(steps, testdb.Step{Match: "SELECT COUNT(*) as cnt FROM subscribe_message_logs", Args: []any{int64(9), int64(2)}, Columns: []string{"cnt"}, Rows: [][]driver.Value{{tt.count}}})
+			}
+			db, _ := testdb.Open(t, steps)
+			sender := &testSender{}
+			Once(&store.DB{DB: db}, sender, tt.force)
+			if sender.calls != tt.want {
+				t.Fatalf("sent %d want %d", sender.calls, tt.want)
+			}
+			if sender.calls > 0 && (sender.to != 2 || sender.from != 1) {
+				t.Fatalf("message must go to partner: from=%d to=%d", sender.from, sender.to)
+			}
+		})
+	}
+}