feat: implement subscription management module with automated discovery engine...

feat: implement subscription management module with automated discovery engine and background worker

.agents/memory.md

@@ -62,6 +62,14 @@ File này chỉ lưu sự thật và quyết định dài hạn giúp các phiê
 - Reminder hỗ trợ `ONCE`, `DAILY`, `WEEKLY`, `MONTHLY`, `YEARLY`, có khoảng lặp và ngày kết
   thúc. Worker nền trong process xử lý reminder, cảnh báo ngân sách/mục tiêu và retry delivery;
   có thể tắt hoặc chỉnh chu kỳ bằng các biến `NOTIFICATION_*`.
+- Subscription discovery dùng `SubscriptionDiscoveryEngine` phân tích 180 ngày giao dịch EXPENSE để
+  phát hiện gói cước định kỳ. Threshold `isPriceDrift` là 8% (không phải 3%) để tránh false positive
+  từ biến động tỷ giá ngoại tệ. Message nhắc nhở dùng `currency` thực tế của subscription thay vì
+  hardcode VND. Worker nền chạy `scanAndNotifyNewDiscoveries()` theo chu kỳ `subscriptionScanIntervalMs`
+  (mặc định 24h, env: `NOTIFICATION_SUBSCRIPTION_SCAN_INTERVAL_MS`), duyệt user theo cursor batch 50,
+  chỉ notify các subscription có `confidenceScore >= 0.85` và chưa được link với reminder; dedupKey
+  theo ngày tránh gửi lặp. Dùng `NotificationType.SYSTEM` — không cần thêm enum/migration mới.
+  API convert-to-reminder hỗ trợ `remindDaysBefore` (0-30 ngày, mặc định: 2 ngày cho MONTHLY, 7 ngày cho YEARLY, 0 ngày cho WEEKLY/DAILY); ngày kích hoạt thực tế `remindAt` và `nextTriggerAt` được trừ tương ứng từ ngày gia hạn gốc.
 - Giao dịch tự động định kỳ lưu template/lịch riêng với `DAILY`, `WEEKLY`, `MONTHLY`, `YEARLY`,
   hỗ trợ pause/resume, ngày kết thúc và chính sách `SKIP`/`CATCH_UP`. Worker dùng business date
   UTC+7, distributed lock và occurrence unique `(scheduleId, scheduledFor)`; bản ghi Transaction

.env.example

@@ -51,6 +51,8 @@ R2_AVATAR_MAX_FILE_SIZE_MB=5
 NOTIFICATION_WORKER_ENABLED=true
 NOTIFICATION_WORKER_INTERVAL_MS=60000
 NOTIFICATION_FINANCIAL_SCAN_INTERVAL_MS=300000
+# How often (ms) the worker scans users for newly discovered subscriptions. Default: 86400000 (24h). Range: 3600000–86400000.
+NOTIFICATION_SUBSCRIPTION_SCAN_INTERVAL_MS=86400000
 RECURRING_TRANSACTION_BATCH_LIMIT=100
 
 AI_PROVIDER=gemini

src/config/env.config.ts

@@ -92,6 +92,17 @@ export const envConfig = {
       const value = parseInt(process.env.NOTIFICATION_BACKOFF_BASE_DELAY_MS || '60000', 10);
       return Number.isFinite(value) && value >= 1000 && value <= 600000 ? value : 60000;
     })(),
+    // How often (ms) the worker scans all active users for newly discovered subscriptions.
+    // Defaults to once every 24 hours. Set to 0 to disable subscription scanning.
+    subscriptionScanIntervalMs: (() => {
+      const value = parseInt(
+        process.env.NOTIFICATION_SUBSCRIPTION_SCAN_INTERVAL_MS || '86400000',
+        10,
+      );
+      return Number.isFinite(value) && value >= 3_600_000 && value <= 86_400_000
+        ? value
+        : 86_400_000;
+    })(),
   },
   webhooks: {
     concurrency: (() => {

src/modules/notifications/notification.worker.ts

@@ -4,6 +4,7 @@ import { NotificationDeliveryService } from './notification-delivery.service';
 import { NotificationService } from './notification.service';
 import { RecurringTransactionService } from '../recurring-transactions/recurring-transaction.service';
 import { BudgetService } from '../budgets/budget.service';
+import { SubscriptionService } from '../subscriptions/subscription.service';
 
 import { lockService } from '../../common/services/lock.service';
 
@@ -13,9 +14,11 @@ export class NotificationWorker {
   private readonly deliveryService = new NotificationDeliveryService();
   private readonly recurringTransactionService = new RecurringTransactionService();
   private readonly budgetService = new BudgetService();
+  private readonly subscriptionService = new SubscriptionService();
   private timer: NodeJS.Timeout | null = null;
   private running = false;
   private lastFinancialScanAt = 0;
+  private lastSubscriptionScanAt = 0;
 
   start() {
     if (!envConfig.notifications.workerEnabled || this.timer) {
@@ -91,6 +94,23 @@ export class NotificationWorker {
           console.error('Notification worker failed to scan financial alerts', error);
         }
       }
+
+      if (
+        now.getTime() - this.lastSubscriptionScanAt
+        >= envConfig.notifications.subscriptionScanIntervalMs
+      ) {
+        try {
+          const result = await this.subscriptionService.scanAndNotifyNewDiscoveries();
+          this.lastSubscriptionScanAt = now.getTime();
+          if (result.notificationsSent > 0) {
+            console.info(
+              `[NotificationWorker] Subscription scan: ${result.usersScanned} users, ${result.notificationsSent} notifications sent`,
+            );
+          }
+        } catch (error) {
+          console.error('Notification worker failed to scan subscriptions', error);
+        }
+      }
     } finally {
       this.running = false;
       await lockService.release(lockKey, lockToken);

src/modules/subscriptions/subscription-engine.ts

@@ -132,9 +132,11 @@ export class SubscriptionDiscoveryEngine {
       const countBonus = Math.min(1.0, sorted.length / 5);
       const confidenceScore = Math.min(0.99, Math.round((regularity * 0.7 + countBonus * 0.3) * 100) / 100);
 
-      // Price drift detection (> 3% increase)
+      // Price drift detection (> 8% increase).
+      // 8% guards against minor currency-conversion fluctuations for foreign-currency
+      // subscriptions while still catching real plan price increases (typically 10–30%).
       const driftPercent = ((latestAmount - avgAmount) / avgAmount) * 100;
-      const isPriceDrift = driftPercent > 3.0;
+      const isPriceDrift = driftPercent > 8.0;
 
       // Check if already linked to a user reminder
       const isLinkedToReminder = existingReminderTitles.has(cleanMerchant.toLowerCase());

src/modules/subscriptions/subscription.dto.ts

@@ -21,8 +21,11 @@ export interface DiscoveredSubscriptionDto {
 export interface ConvertSubscriptionToReminderDto {
   merchantName: string;
   amount: string;
+  currency?: string;
   frequency: ReminderFrequency;
-  remindAt: string; // ISO date string
+  /** Renewal date at 09:00 Vietnam time (ISO). Actual trigger = remindAt − remindDaysBefore days. */
+  remindAt: string;
+  remindDaysBefore?: number; // 0 = same day, default: 2 for MONTHLY, 7 for YEARLY, 0 otherwise
   categoryId?: string;
 }
 

src/modules/subscriptions/subscription.repository.ts

 import { ReminderType, TransactionType } from '@prisma/client';
 import { prisma } from '../../database/prisma.client';
 import {
+  addBusinessDays,
+  businessWallTimeToInstant,
+  instantToBusinessWallTime,
   prismaDateToBusinessDate,
 } from '../../common/date-time/business-time';
 import { ConvertSubscriptionToReminderDto } from './subscription.dto';
@@ -56,13 +59,17 @@ export class SubscriptionRepository {
   }
 
   async convertToReminder(userId: string, input: ConvertSubscriptionToReminderDto) {
-    const remindAtDate = new Date(input.remindAt);
+    const defaultDays = input.frequency === 'MONTHLY' ? 2 : input.frequency === 'YEARLY' ? 7 : 0;
+    const remindDaysBefore = Math.max(0, input.remindDaysBefore ?? defaultDays);
+    const renewalWall = instantToBusinessWallTime(new Date(input.remindAt));
+    const triggerDateStr = addBusinessDays(renewalWall.date, -remindDaysBefore);
+    const remindAtDate = businessWallTimeToInstant(triggerDateStr, renewalWall.time);
 
     return prisma.reminder.create({
       data: {
         userId,
         title: input.merchantName,
-        message: `Thanh toán gói cước định kỳ: ${input.merchantName} (${parseFloat(input.amount).toLocaleString()} VND)`,
+        message: `Thanh toán gói cước định kỳ: ${input.merchantName} (${parseFloat(input.amount).toLocaleString('vi-VN')} ${input.currency ?? 'VND'})`,
         type: ReminderType.RECURRING_PAYMENT,
         frequency: input.frequency,
         repeatInterval: 1,
@@ -72,4 +79,32 @@ export class SubscriptionRepository {
       },
     });
   }
+
+  /**
+   * Return a batch of user IDs who have had at least one EXPENSE transaction
+   * in the last `days` days. Uses cursor-based pagination so the caller can
+   * iterate without loading all users into memory at once.
+   */
+  async findActiveUserIdsBatch(
+    days: number,
+    cursor: string | undefined,
+    batchSize: number,
+  ): Promise<{ userIds: string[]; nextCursor: string | undefined }> {
+    const since = new Date(Date.now() - days * MILLISECONDS_PER_DAY);
+
+    const rows = await prisma.transaction.groupBy({
+      by: ['userId'],
+      where: {
+        type: TransactionType.EXPENSE,
+        date: { gte: since },
+        ...(cursor ? { userId: { gt: cursor } } : {}),
+      },
+      orderBy: { userId: 'asc' },
+      take: batchSize,
+    });
+
+    const userIds = rows.map((r) => r.userId);
+    const nextCursor = userIds.length === batchSize ? userIds[userIds.length - 1] : undefined;
+    return { userIds, nextCursor };
+  }
 }

src/modules/subscriptions/subscription.service.ts

+import {
+  NotificationPriority,
+  NotificationSourceType,
+  NotificationType,
+} from '@prisma/client';
 import {
   ConvertSubscriptionToReminderDto,
   DiscoveryReportDto,
@@ -6,14 +11,19 @@ import { SubscriptionDiscoveryEngine } from './subscription-engine';
 import { SubscriptionRepository } from './subscription.repository';
 import { RecurringTransactionService } from '../recurring-transactions/recurring-transaction.service';
 import { ConvertSubscriptionToRecurringTransactionDto } from '../recurring-transactions/recurring-transaction.dto';
+import { NotificationService } from '../notifications/notification.service';
+
+const SCAN_BATCH_SIZE = 50;
+const SCAN_HISTORY_DAYS = 180;
 
 export class SubscriptionService {
   private readonly repository = new SubscriptionRepository();
   private readonly recurringTransactionService = new RecurringTransactionService();
+  private readonly notificationService = new NotificationService();
 
   async discoverSubscriptions(userId: string): Promise<DiscoveryReportDto> {
     const [transactions, existingReminders] = await Promise.all([
-      this.repository.getHistoricalExpenseTransactions(userId, 180),
+      this.repository.getHistoricalExpenseTransactions(userId, SCAN_HISTORY_DAYS),
       this.repository.getExistingReminderTitles(userId),
     ]);
 
@@ -41,4 +51,80 @@ export class SubscriptionService {
   ) {
     return this.recurringTransactionService.convertSubscription(userId, input);
   }
+
+  /**
+   * Proactive background scan: iterates over all active users in cursor batches,
+   * runs subscription discovery for each, and creates an IN_APP notification for
+   * any high-confidence subscriptions that are not yet linked to a reminder.
+   *
+   * Uses dedupKey to prevent re-notifying the same merchant within a 24-hour window,
+   * so this method is safe to call on every worker tick as long as
+   * the caller gates it with `subscriptionScanIntervalMs`.
+   */
+  async scanAndNotifyNewDiscoveries(): Promise<{ usersScanned: number; notificationsSent: number }> {
+    let cursor: string | undefined;
+    let usersScanned = 0;
+    let notificationsSent = 0;
+    const scanDate = new Date().toISOString().slice(0, 10); // YYYY-MM-DD, resets dedup daily
+
+    do {
+      const { userIds, nextCursor } = await this.repository.findActiveUserIdsBatch(
+        SCAN_HISTORY_DAYS,
+        cursor,
+        SCAN_BATCH_SIZE,
+      );
+
+      for (const userId of userIds) {
+        try {
+          const report = await this.discoverSubscriptions(userId);
+          usersScanned += 1;
+
+          // Only notify for high-confidence, unlinked subscriptions
+          const candidates = report.items.filter(
+            (item) => item.confidenceScore >= 0.85 && !item.isLinkedToReminder,
+          );
+
+          for (const item of candidates) {
+            const notification = await this.notificationService.create({
+              userId,
+              type: NotificationType.SYSTEM,
+              priority: item.isPriceDrift
+                ? NotificationPriority.HIGH
+                : NotificationPriority.NORMAL,
+              title: item.isPriceDrift
+                ? `Giá ${item.merchantName} đã thay đổi`
+                : `Phát hiện gói cước định kỳ: ${item.merchantName}`,
+              message: item.isPriceDrift
+                ? `${item.merchantName} tăng giá ${item.priceDriftPercentage?.toFixed(1)}% so với trung bình. Muốn theo dõi không?`
+                : `${item.merchantName} xuất hiện ${item.occurrenceCount} lần (${item.frequency.toLowerCase()}). Thêm vào danh sách theo dõi?`,
+              actionUrl: '/subscriptions',
+              sourceType: NotificationSourceType.SYSTEM,
+              sourceId: null,
+              data: {
+                merchantName: item.merchantName,
+                averageAmount: item.averageAmount,
+                currency: item.currency,
+                frequency: item.frequency,
+                confidenceScore: item.confidenceScore,
+                isPriceDrift: item.isPriceDrift,
+              },
+              // Daily dedup key — one notification per merchant per user per day
+              dedupKey: `subscription:discovered:${userId}:${item.merchantName}:${item.currency}:${scanDate}`,
+            });
+
+            if (notification) {
+              notificationsSent += 1;
+            }
+          }
+        } catch (error) {
+          // Per-user errors must not abort the whole scan
+          console.error(`[SubscriptionService] scanAndNotify failed for user ${userId}:`, error);
+        }
+      }
+
+      cursor = nextCursor;
+    } while (cursor);
+
+    return { usersScanned, notificationsSent };
+  }
 }

src/modules/subscriptions/subscription.validation.ts

@@ -4,7 +4,10 @@ export { convertSubscriptionToRecurringTransactionSchema } from '../recurring-tr
 export const convertSubscriptionToReminderSchema = z.object({
   merchantName: z.string().min(1).max(100),
   amount: z.string().regex(/^\d+(\.\d{1,2})?$/, 'Amount must be a valid positive number'),
+  currency: z.string().min(3).max(10).optional(),
   frequency: z.enum(['DAILY', 'WEEKLY', 'MONTHLY', 'YEARLY']),
   remindAt: z.string().datetime(),
+  remindDaysBefore: z.number().int().min(0).max(30).optional(),
   categoryId: z.string().uuid().optional(),
 });
+