subscription_expiry_service.go 6.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215
  1. package service
  2. import (
  3. "context"
  4. "database/sql"
  5. "errors"
  6. "fmt"
  7. "log"
  8. "strconv"
  9. "sync"
  10. "time"
  11. "github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
  12. "github.com/google/uuid"
  13. )
  14. const (
  15. subscriptionExpiryReminderSMTPWarningInterval = time.Minute
  16. // subscriptionExpiryReminderLeaderLockKey gates the per-cycle reminder scan so
  17. // that only one instance walks all active subscriptions and sends reminder
  18. // emails, avoiding redundant full scans and duplicate emails.
  19. subscriptionExpiryReminderLeaderLockKey = "subscription:expiry:reminder:leader"
  20. // subscriptionExpiryReminderLeaderLockTTL bounds crash recovery; the scan can
  21. // page through many subscriptions, so keep it comfortably above one cycle.
  22. subscriptionExpiryReminderLeaderLockTTL = 5 * time.Minute
  23. )
  24. // SubscriptionExpiryService periodically updates expired subscription status.
  25. type SubscriptionExpiryService struct {
  26. userSubRepo UserSubscriptionRepository
  27. settingRepo SettingRepository
  28. notificationEmailService *NotificationEmailService
  29. interval time.Duration
  30. stopCh chan struct{}
  31. stopOnce sync.Once
  32. wg sync.WaitGroup
  33. lockCache LeaderLockCache
  34. db *sql.DB
  35. instanceID string
  36. smtpWarningMu sync.Mutex
  37. lastSMTPWarning time.Time
  38. }
  39. func NewSubscriptionExpiryService(userSubRepo UserSubscriptionRepository, interval time.Duration) *SubscriptionExpiryService {
  40. return &SubscriptionExpiryService{
  41. userSubRepo: userSubRepo,
  42. interval: interval,
  43. stopCh: make(chan struct{}),
  44. instanceID: uuid.NewString(),
  45. }
  46. }
  47. // SetLeaderLock injects the leader-lock cache and DB used to elect a single
  48. // instance for the periodic expiry-reminder scan. When both are nil the scan runs
  49. // ungated (single-instance / test behavior).
  50. func (s *SubscriptionExpiryService) SetLeaderLock(lockCache LeaderLockCache, db *sql.DB) {
  51. if s == nil {
  52. return
  53. }
  54. s.lockCache = lockCache
  55. s.db = db
  56. }
  57. func (s *SubscriptionExpiryService) SetSettingRepository(settingRepo SettingRepository) {
  58. s.settingRepo = settingRepo
  59. }
  60. func (s *SubscriptionExpiryService) SetNotificationEmailService(notificationEmailService *NotificationEmailService) {
  61. s.notificationEmailService = notificationEmailService
  62. }
  63. func (s *SubscriptionExpiryService) Start() {
  64. if s == nil || s.userSubRepo == nil || s.interval <= 0 {
  65. return
  66. }
  67. s.wg.Add(1)
  68. go func() {
  69. defer s.wg.Done()
  70. ticker := time.NewTicker(s.interval)
  71. defer ticker.Stop()
  72. s.runOnce()
  73. for {
  74. select {
  75. case <-ticker.C:
  76. s.runOnce()
  77. case <-s.stopCh:
  78. return
  79. }
  80. }
  81. }()
  82. }
  83. func (s *SubscriptionExpiryService) Stop() {
  84. if s == nil {
  85. return
  86. }
  87. s.stopOnce.Do(func() {
  88. close(s.stopCh)
  89. })
  90. s.wg.Wait()
  91. }
  92. func (s *SubscriptionExpiryService) runOnce() {
  93. ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
  94. defer cancel()
  95. updated, err := s.userSubRepo.BatchUpdateExpiredStatus(ctx)
  96. if err != nil {
  97. log.Printf("[SubscriptionExpiry] Update expired subscriptions failed: %v", err)
  98. return
  99. }
  100. if updated > 0 {
  101. log.Printf("[SubscriptionExpiry] Updated %d expired subscriptions", updated)
  102. }
  103. s.sendExpiryReminders(ctx)
  104. }
  105. func (s *SubscriptionExpiryService) sendExpiryReminders(ctx context.Context) {
  106. if s == nil || s.userSubRepo == nil || s.notificationEmailService == nil {
  107. return
  108. }
  109. if !s.expiryReminderEnabled(ctx) {
  110. return
  111. }
  112. if !s.smtpConfigured(ctx) {
  113. return
  114. }
  115. // Multi-instance guard: only the leader walks every active subscription and
  116. // sends reminders, avoiding N× full scans and duplicate reminder emails.
  117. release, ok := tryAcquireSingletonLeaderLock(ctx, s.lockCache, s.db, subscriptionExpiryReminderLeaderLockKey, s.instanceID, subscriptionExpiryReminderLeaderLockTTL)
  118. if !ok {
  119. return
  120. }
  121. defer release()
  122. for page := 1; ; page++ {
  123. subs, pag, err := s.userSubRepo.List(ctx, pagination.PaginationParams{Page: page, PageSize: 200}, nil, nil, SubscriptionStatusActive, "", "expires_at", "asc")
  124. if err != nil {
  125. log.Printf("[SubscriptionExpiry] List active subscriptions for reminder failed: %v", err)
  126. return
  127. }
  128. for i := range subs {
  129. s.sendExpiryReminderIfDue(ctx, &subs[i])
  130. }
  131. if pag == nil || page >= pag.Pages || len(subs) == 0 {
  132. return
  133. }
  134. }
  135. }
  136. func (s *SubscriptionExpiryService) expiryReminderEnabled(ctx context.Context) bool {
  137. if s == nil || s.settingRepo == nil {
  138. return true
  139. }
  140. value, err := s.settingRepo.GetValue(ctx, SettingKeySubscriptionExpiryNotifyEnabled)
  141. if err != nil {
  142. if errors.Is(err, ErrSettingNotFound) {
  143. return true
  144. }
  145. log.Printf("[SubscriptionExpiry] Read expiry reminder switch failed: %v", err)
  146. return false
  147. }
  148. return !isFalseSettingValue(value)
  149. }
  150. func (s *SubscriptionExpiryService) smtpConfigured(ctx context.Context) bool {
  151. if s == nil || s.notificationEmailService == nil || s.notificationEmailService.emailService == nil {
  152. return false
  153. }
  154. _, err := s.notificationEmailService.emailService.GetSMTPConfig(ctx)
  155. if err == nil {
  156. return true
  157. }
  158. if errors.Is(err, ErrEmailNotConfigured) {
  159. s.smtpWarningMu.Lock()
  160. defer s.smtpWarningMu.Unlock()
  161. now := time.Now()
  162. if s.lastSMTPWarning.IsZero() || now.Sub(s.lastSMTPWarning) >= subscriptionExpiryReminderSMTPWarningInterval {
  163. log.Printf("[SubscriptionExpiry] SMTP is not configured; skipping expiry reminders")
  164. s.lastSMTPWarning = now
  165. }
  166. return false
  167. }
  168. log.Printf("[SubscriptionExpiry] Read SMTP configuration failed; skipping expiry reminders: %v", err)
  169. return false
  170. }
  171. func (s *SubscriptionExpiryService) sendExpiryReminderIfDue(ctx context.Context, sub *UserSubscription) {
  172. if sub == nil || sub.User == nil || sub.Group == nil || sub.User.Email == "" {
  173. return
  174. }
  175. daysRemaining := sub.DaysRemaining()
  176. if daysRemaining != 7 && daysRemaining != 3 && daysRemaining != 1 {
  177. return
  178. }
  179. if err := s.notificationEmailService.Send(ctx, NotificationEmailSendInput{
  180. Event: NotificationEmailEventSubscriptionExpiryReminder,
  181. RecipientEmail: sub.User.Email,
  182. RecipientName: firstNonEmpty(sub.User.Username, sub.User.Email),
  183. UserID: sub.UserID,
  184. SourceType: "user_subscription",
  185. SourceID: strconv.FormatInt(sub.ID, 10),
  186. ReminderKey: fmt.Sprintf("%dd", daysRemaining),
  187. Variables: map[string]string{
  188. "subscription_group": sub.Group.Name,
  189. "expiry_time": sub.ExpiresAt.Format("2006-01-02 15:04"),
  190. "days_remaining": strconv.Itoa(daysRemaining),
  191. },
  192. }); err != nil {
  193. log.Printf("[SubscriptionExpiry] Send expiry reminder failed: subscription=%d user=%d err=%v", sub.ID, sub.UserID, err)
  194. }
  195. }