feat(recurring): implement recurring transaction engine, worker execution, and...

feat(recurring): implement recurring transaction engine, worker execution, and wallet balance updates

.agents/memory.md

@@ -54,6 +54,11 @@ 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_*`.
+- 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
+  cùng biến động Wallet được commit nguyên tử trong transaction Serializable. Xóa lịch là xóa mềm
+  và không xóa hoặc hoàn tác các giao dịch đã ghi.
 - Giao dịch chi bất thường được cảnh báo khi có ít nhất 5 giao dịch lịch sử 90 ngày trong cùng
   ví và số tiền mới đạt ít nhất 3 lần trung bình; đây là heuristic có thể thay bằng AI sau.
 - AI Financial Assistant là module read-only/stateless dưới `/api/v1/ai-assistant`, gồm phân loại
@@ -102,6 +107,8 @@ File này chỉ lưu sự thật và quyết định dài hạn giúp các phiê
   cho Budget, backfill dữ liệu hiện có bằng `VND`.
   `20260804120000_add_notification_reminder_system` thêm notification inbox, preferences,
   delivery outbox và user reminders cùng các enum/index phục vụ worker nền.
+  `20260820120000_add_recurring_transactions` thêm lịch giao dịch định kỳ, occurrence idempotency,
+  quan hệ với transaction được tạo và các index phục vụ worker.
   Migration history cũ vẫn chưa phản ánh đầy đủ các thay đổi schema của auth đã
   được commit trước đó.
 

.env.example

@@ -50,6 +50,7 @@ R2_AVATAR_MAX_FILE_SIZE_MB=5
 NOTIFICATION_WORKER_ENABLED=true
 NOTIFICATION_WORKER_INTERVAL_MS=60000
 NOTIFICATION_FINANCIAL_SCAN_INTERVAL_MS=300000
+RECURRING_TRANSACTION_BATCH_LIMIT=100
 
 AI_PROVIDER=gemini
 GEMINI_API_KEYS=replace_with_key_1,replace_with_key_2

prisma/migrations/20260820120000_add_recurring_transactions/migration.sql

+-- CreateEnum
+CREATE TYPE "RecurringTransactionFrequency" AS ENUM ('DAILY', 'WEEKLY', 'MONTHLY', 'YEARLY');
+
+-- CreateEnum
+CREATE TYPE "RecurringTransactionMissedRunPolicy" AS ENUM ('SKIP', 'CATCH_UP');
+
+-- CreateEnum
+CREATE TYPE "RecurringTransactionOccurrenceStatus" AS ENUM ('POSTED', 'FAILED');
+
+-- CreateTable
+CREATE TABLE "recurring_transaction_schedules" (
+    "id" UUID NOT NULL,
+    "user_id" UUID NOT NULL,
+    "wallet_id" UUID NOT NULL,
+    "category_id" UUID NOT NULL,
+    "amount" DECIMAL(18,2) NOT NULL,
+    "type" "TransactionType" NOT NULL,
+    "description" TEXT,
+    "location" TEXT,
+    "frequency" "RecurringTransactionFrequency" NOT NULL,
+    "repeat_interval" INTEGER NOT NULL DEFAULT 1,
+    "anchor_date" DATE NOT NULL,
+    "end_date" DATE,
+    "next_run_at" DATE,
+    "missed_run_policy" "RecurringTransactionMissedRunPolicy" NOT NULL DEFAULT 'SKIP',
+    "last_run_at" TIMESTAMPTZ(3),
+    "is_active" BOOLEAN NOT NULL DEFAULT true,
+    "deleted_at" TIMESTAMPTZ(3),
+    "created_at" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updated_at" TIMESTAMPTZ(3) NOT NULL,
+
+    CONSTRAINT "recurring_transaction_schedules_pkey" PRIMARY KEY ("id"),
+    CONSTRAINT "recurring_transaction_schedules_amount_positive_check" CHECK ("amount" > 0),
+    CONSTRAINT "recurring_transaction_schedules_repeat_interval_positive_check" CHECK ("repeat_interval" > 0),
+    CONSTRAINT "recurring_transaction_schedules_end_date_check" CHECK ("end_date" IS NULL OR "end_date" >= "anchor_date"),
+    CONSTRAINT "recurring_transaction_schedules_active_next_run_check" CHECK (NOT "is_active" OR ("deleted_at" IS NULL AND "next_run_at" IS NOT NULL))
+);
+
+-- CreateTable
+CREATE TABLE "recurring_transaction_occurrences" (
+    "id" UUID NOT NULL,
+    "schedule_id" UUID NOT NULL,
+    "scheduled_for" DATE NOT NULL,
+    "status" "RecurringTransactionOccurrenceStatus" NOT NULL DEFAULT 'POSTED',
+    "transaction_id" UUID,
+    "failure_code" VARCHAR(100),
+    "failure_message" VARCHAR(500),
+    "attempt_count" INTEGER NOT NULL DEFAULT 1,
+    "created_at" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updated_at" TIMESTAMPTZ(3) NOT NULL,
+
+    CONSTRAINT "recurring_transaction_occurrences_pkey" PRIMARY KEY ("id"),
+    CONSTRAINT "recurring_transaction_occurrences_attempt_count_positive_check" CHECK ("attempt_count" > 0),
+    CONSTRAINT "recurring_transaction_occurrences_status_payload_check" CHECK (
+      ("status" = 'POSTED' AND "failure_code" IS NULL)
+      OR ("status" = 'FAILED' AND "transaction_id" IS NULL AND "failure_code" IS NOT NULL)
+    )
+);
+
+-- CreateIndex
+CREATE INDEX "recurring_transaction_schedules_user_id_deleted_at_is_active_idx" ON "recurring_transaction_schedules"("user_id", "deleted_at", "is_active");
+CREATE INDEX "recurring_transaction_schedules_is_active_next_run_at_idx" ON "recurring_transaction_schedules"("is_active", "next_run_at");
+CREATE INDEX "recurring_transaction_schedules_wallet_id_idx" ON "recurring_transaction_schedules"("wallet_id");
+CREATE INDEX "recurring_transaction_schedules_category_id_idx" ON "recurring_transaction_schedules"("category_id");
+CREATE UNIQUE INDEX "recurring_transaction_occurrences_transaction_id_key" ON "recurring_transaction_occurrences"("transaction_id");
+CREATE UNIQUE INDEX "recurring_transaction_occurrences_schedule_id_scheduled_for_key" ON "recurring_transaction_occurrences"("schedule_id", "scheduled_for");
+CREATE INDEX "recurring_transaction_occurrences_schedule_id_created_at_idx" ON "recurring_transaction_occurrences"("schedule_id", "created_at");
+CREATE INDEX "recurring_transaction_occurrences_status_updated_at_idx" ON "recurring_transaction_occurrences"("status", "updated_at");
+
+-- AddForeignKey
+ALTER TABLE "recurring_transaction_schedules" ADD CONSTRAINT "recurring_transaction_schedules_user_id_fkey" FOREIGN KEY ("user_id") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+ALTER TABLE "recurring_transaction_schedules" ADD CONSTRAINT "recurring_transaction_schedules_wallet_id_fkey" FOREIGN KEY ("wallet_id") REFERENCES "wallets"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
+ALTER TABLE "recurring_transaction_schedules" ADD CONSTRAINT "recurring_transaction_schedules_category_id_fkey" FOREIGN KEY ("category_id") REFERENCES "categories"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
+ALTER TABLE "recurring_transaction_occurrences" ADD CONSTRAINT "recurring_transaction_occurrences_schedule_id_fkey" FOREIGN KEY ("schedule_id") REFERENCES "recurring_transaction_schedules"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+ALTER TABLE "recurring_transaction_occurrences" ADD CONSTRAINT "recurring_transaction_occurrences_transaction_id_fkey" FOREIGN KEY ("transaction_id") REFERENCES "transactions"("id") ON DELETE SET NULL ON UPDATE CASCADE;

prisma/schema.prisma

@@ -86,6 +86,23 @@ enum ReminderFrequency {
   YEARLY
 }
 
+enum RecurringTransactionFrequency {
+  DAILY
+  WEEKLY
+  MONTHLY
+  YEARLY
+}
+
+enum RecurringTransactionMissedRunPolicy {
+  SKIP
+  CATCH_UP
+}
+
+enum RecurringTransactionOccurrenceStatus {
+  POSTED
+  FAILED
+}
+
 model User {
   id              String    @id @default(uuid()) @db.Uuid
   email           String?   @unique
@@ -103,21 +120,22 @@ model User {
   updatedAt       DateTime  @updatedAt @map("updated_at") @db.Timestamptz(3)
 
   // Relations
-  role                Role                 @relation(fields: [roleId], references: [id], onDelete: Restrict)
-  wallets             Wallet[]
-  categories          Category[]
-  transactions        Transaction[]
-  transfers           Transfer[]
-  budgets             Budget[]
-  savingGoals         SavingGoal[]
-  refreshTokens       RefreshToken[]
-  socialAccounts      UserSocial[]
-  verificationTokens  VerificationToken[]
-  passwordResetTokens PasswordResetToken[]
-  devices             UserDevice[]
-  notifications       Notification[]
-  notificationSetting NotificationSetting?
-  reminders           Reminder[]
+  role                          Role                           @relation(fields: [roleId], references: [id], onDelete: Restrict)
+  wallets                       Wallet[]
+  categories                    Category[]
+  transactions                  Transaction[]
+  transfers                     Transfer[]
+  budgets                       Budget[]
+  savingGoals                   SavingGoal[]
+  refreshTokens                 RefreshToken[]
+  socialAccounts                UserSocial[]
+  verificationTokens            VerificationToken[]
+  passwordResetTokens           PasswordResetToken[]
+  devices                       UserDevice[]
+  notifications                 Notification[]
+  notificationSetting           NotificationSetting?
+  reminders                     Reminder[]
+  recurringTransactionSchedules RecurringTransactionSchedule[]
 
   @@index([roleId])
   @@map("users")
@@ -147,10 +165,11 @@ model Wallet {
   createdAt   DateTime @default(now()) @map("created_at") @db.Timestamptz(3)
   updatedAt   DateTime @updatedAt @map("updated_at") @db.Timestamptz(3)
 
-  user         User          @relation(fields: [userId], references: [id], onDelete: Cascade)
-  transactions Transaction[]
-  outgoingTransfers Transfer[] @relation("TransferSourceWallet")
-  incomingTransfers Transfer[] @relation("TransferDestinationWallet")
+  user                          User                           @relation(fields: [userId], references: [id], onDelete: Cascade)
+  transactions                  Transaction[]
+  outgoingTransfers             Transfer[]                     @relation("TransferSourceWallet")
+  incomingTransfers             Transfer[]                     @relation("TransferDestinationWallet")
+  recurringTransactionSchedules RecurringTransactionSchedule[]
 
   @@unique([userId, name])
   @@index([userId])
@@ -171,11 +190,12 @@ model Category {
   createdAt  DateTime        @default(now()) @map("created_at") @db.Timestamptz(3)
   updatedAt  DateTime        @updatedAt @map("updated_at") @db.Timestamptz(3)
 
-  user         User?         @relation(fields: [userId], references: [id], onDelete: Cascade)
-  parent       Category?     @relation("CategoryHierarchy", fields: [parentId], references: [id], onDelete: SetNull)
-  children     Category[]    @relation("CategoryHierarchy")
-  transactions Transaction[]
-  budgets      Budget[]
+  user                          User?                          @relation(fields: [userId], references: [id], onDelete: Cascade)
+  parent                        Category?                      @relation("CategoryHierarchy", fields: [parentId], references: [id], onDelete: SetNull)
+  children                      Category[]                     @relation("CategoryHierarchy")
+  transactions                  Transaction[]
+  budgets                       Budget[]
+  recurringTransactionSchedules RecurringTransactionSchedule[]
 
   @@unique([userId, name, type])
   @@index([userId, isArchived])
@@ -198,9 +218,10 @@ model Transaction {
   createdAt   DateTime        @default(now()) @map("created_at") @db.Timestamptz(3)
   updatedAt   DateTime        @updatedAt @map("updated_at") @db.Timestamptz(3)
 
-  user     User     @relation(fields: [userId], references: [id], onDelete: Cascade)
-  wallet   Wallet   @relation(fields: [walletId], references: [id], onDelete: Restrict)
-  category Category @relation(fields: [categoryId], references: [id], onDelete: Restrict)
+  user                User                            @relation(fields: [userId], references: [id], onDelete: Cascade)
+  wallet              Wallet                          @relation(fields: [walletId], references: [id], onDelete: Restrict)
+  category            Category                        @relation(fields: [categoryId], references: [id], onDelete: Restrict)
+  recurringOccurrence RecurringTransactionOccurrence?
 
   @@index([userId])
   @@index([walletId])
@@ -459,3 +480,57 @@ model Reminder {
   @@index([nextTriggerAt, isActive])
   @@map("reminders")
 }
+
+model RecurringTransactionSchedule {
+  id              String                              @id @default(uuid()) @db.Uuid
+  userId          String                              @map("user_id") @db.Uuid
+  walletId        String                              @map("wallet_id") @db.Uuid
+  categoryId      String                              @map("category_id") @db.Uuid
+  amount          Decimal                             @db.Decimal(18, 2)
+  type            TransactionType
+  description     String?
+  location        String?
+  frequency       RecurringTransactionFrequency
+  repeatInterval  Int                                 @default(1) @map("repeat_interval")
+  anchorDate      DateTime                            @map("anchor_date") @db.Date
+  endDate         DateTime?                           @map("end_date") @db.Date
+  nextRunAt       DateTime?                           @map("next_run_at") @db.Date
+  missedRunPolicy RecurringTransactionMissedRunPolicy @default(SKIP) @map("missed_run_policy")
+  lastRunAt       DateTime?                           @map("last_run_at") @db.Timestamptz(3)
+  isActive        Boolean                             @default(true) @map("is_active")
+  deletedAt       DateTime?                           @map("deleted_at") @db.Timestamptz(3)
+  createdAt       DateTime                            @default(now()) @map("created_at") @db.Timestamptz(3)
+  updatedAt       DateTime                            @updatedAt @map("updated_at") @db.Timestamptz(3)
+
+  user        User                             @relation(fields: [userId], references: [id], onDelete: Cascade)
+  wallet      Wallet                           @relation(fields: [walletId], references: [id], onDelete: Restrict)
+  category    Category                         @relation(fields: [categoryId], references: [id], onDelete: Restrict)
+  occurrences RecurringTransactionOccurrence[]
+
+  @@index([userId, deletedAt, isActive])
+  @@index([isActive, nextRunAt])
+  @@index([walletId])
+  @@index([categoryId])
+  @@map("recurring_transaction_schedules")
+}
+
+model RecurringTransactionOccurrence {
+  id             String                               @id @default(uuid()) @db.Uuid
+  scheduleId     String                               @map("schedule_id") @db.Uuid
+  scheduledFor   DateTime                             @map("scheduled_for") @db.Date
+  status         RecurringTransactionOccurrenceStatus @default(POSTED)
+  transactionId  String?                              @unique @map("transaction_id") @db.Uuid
+  failureCode    String?                              @map("failure_code") @db.VarChar(100)
+  failureMessage String?                              @map("failure_message") @db.VarChar(500)
+  attemptCount   Int                                  @default(1) @map("attempt_count")
+  createdAt      DateTime                             @default(now()) @map("created_at") @db.Timestamptz(3)
+  updatedAt      DateTime                             @updatedAt @map("updated_at") @db.Timestamptz(3)
+
+  schedule    RecurringTransactionSchedule @relation(fields: [scheduleId], references: [id], onDelete: Cascade)
+  transaction Transaction?                 @relation(fields: [transactionId], references: [id], onDelete: SetNull)
+
+  @@unique([scheduleId, scheduledFor])
+  @@index([scheduleId, createdAt])
+  @@index([status, updatedAt])
+  @@map("recurring_transaction_occurrences")
+}

src/common/errors/error-code.ts

@@ -27,6 +27,8 @@ export const ERROR_CODE = {
   SAVING_GOAL_COMPLETED: 'SAVING_GOAL_COMPLETED',
   SAVING_GOAL_CURRENCY_LOCKED: 'SAVING_GOAL_CURRENCY_LOCKED',
   REMINDER_SCHEDULE_INVALID: 'REMINDER_SCHEDULE_INVALID',
+  RECURRING_TRANSACTION_SCHEDULE_INVALID: 'RECURRING_TRANSACTION_SCHEDULE_INVALID',
+  RECURRING_TRANSACTION_EXECUTION_FAILED: 'RECURRING_TRANSACTION_EXECUTION_FAILED',
   FILE_TYPE_UNSUPPORTED: 'FILE_TYPE_UNSUPPORTED',
   FILE_TOO_LARGE: 'FILE_TOO_LARGE',
   RECEIPT_NOT_FOUND: 'RECEIPT_NOT_FOUND',

src/config/env.config.ts

@@ -77,6 +77,12 @@ export const envConfig = {
         : 300000;
     })(),
   },
+  recurringTransactions: {
+    batchLimit: (() => {
+      const value = parseInt(process.env.RECURRING_TRANSACTION_BATCH_LIMIT || '100', 10);
+      return Number.isFinite(value) && value >= 1 && value <= 500 ? value : 100;
+    })(),
+  },
   ai: {
     provider: process.env.AI_PROVIDER || 'gemini',
     geminiApiKeys: Array.from(new Set(

src/config/swagger.config.ts

@@ -1181,6 +1181,53 @@ export const swaggerSpec = {
           isActive: { type: 'boolean' },
         },
       },
+      RecurringTransactionSchedule: {
+        type: 'object',
+        required: ['id', 'walletId', 'categoryId', 'amount', 'type', 'frequency', 'repeatInterval', 'anchorDate', 'missedRunPolicy', 'isActive'],
+        properties: {
+          id: { type: 'string', format: 'uuid' },
+          walletId: { type: 'string', format: 'uuid' },
+          categoryId: { type: 'string', format: 'uuid' },
+          amount: { type: 'string', pattern: '^(?:0|[1-9]\\d{0,15})(?:\\.\\d{1,2})?$', example: '250000.00' },
+          type: { type: 'string', enum: ['INCOME', 'EXPENSE'] },
+          description: { type: 'string', nullable: true, maxLength: 500 },
+          location: { type: 'string', nullable: true, maxLength: 255 },
+          frequency: { type: 'string', enum: ['DAILY', 'WEEKLY', 'MONTHLY', 'YEARLY'] },
+          repeatInterval: { type: 'integer', minimum: 1, maximum: 365 },
+          anchorDate: { type: 'string', format: 'date' },
+          endDate: { type: 'string', format: 'date', nullable: true },
+          nextRunAt: { type: 'string', format: 'date', nullable: true },
+          missedRunPolicy: { type: 'string', enum: ['SKIP', 'CATCH_UP'] },
+          lastRunAt: { type: 'string', format: 'date-time', nullable: true },
+          isActive: { type: 'boolean' },
+          createdAt: { type: 'string', format: 'date-time' },
+          updatedAt: { type: 'string', format: 'date-time' },
+        },
+      },
+      CreateRecurringTransactionBody: {
+        type: 'object',
+        required: ['walletId', 'categoryId', 'amount', 'type', 'frequency', 'anchorDate'],
+        properties: {
+          walletId: { type: 'string', format: 'uuid' },
+          categoryId: { type: 'string', format: 'uuid' },
+          amount: { type: 'string', example: '250000.00' },
+          type: { type: 'string', enum: ['INCOME', 'EXPENSE'] },
+          description: { type: 'string', nullable: true, maxLength: 500 },
+          location: { type: 'string', nullable: true, maxLength: 255 },
+          frequency: { type: 'string', enum: ['DAILY', 'WEEKLY', 'MONTHLY', 'YEARLY'] },
+          repeatInterval: { type: 'integer', minimum: 1, maximum: 365, default: 1 },
+          anchorDate: { type: 'string', format: 'date' },
+          endDate: { type: 'string', format: 'date', nullable: true },
+          missedRunPolicy: { type: 'string', enum: ['SKIP', 'CATCH_UP'], default: 'SKIP' },
+          isActive: { type: 'boolean', default: true },
+        },
+      },
+      RecurringTransactionResponse: {
+        allOf: [
+          { $ref: '#/components/schemas/SuccessResponse' },
+          { type: 'object', properties: { data: { $ref: '#/components/schemas/RecurringTransactionSchedule' } } },
+        ],
+      },
       NotificationResponse: {
         allOf: [
           { $ref: '#/components/schemas/SuccessResponse' },
@@ -1798,6 +1845,12 @@ export const swaggerSpec = {
         required: true,
         schema: { type: 'string', format: 'uuid' },
       },
+      RecurringTransactionIdParam: {
+        in: 'path',
+        name: 'id',
+        required: true,
+        schema: { type: 'string', format: 'uuid' },
+      },
       WalletIdParam: {
         in: 'path',
         name: 'id',
@@ -4284,6 +4337,102 @@ export const swaggerSpec = {
         },
       },
     },
+    '/recurring-transactions': {
+      get: {
+        tags: ['Recurring Transactions'],
+        summary: 'List owned recurring transaction schedules',
+        security: [{ BearerAuth: [] }],
+        parameters: [
+          { in: 'query', name: 'isActive', schema: { type: 'string', enum: ['true', 'false'] } },
+          { $ref: '#/components/parameters/PageParam' },
+          { $ref: '#/components/parameters/LimitParam' },
+        ],
+        responses: {
+          200: { description: 'Paginated recurring schedules' },
+          401: { $ref: '#/components/responses/Unauthorized' },
+          422: { $ref: '#/components/responses/Validation' },
+        },
+      },
+      post: {
+        tags: ['Recurring Transactions'],
+        summary: 'Create an opt-in recurring transaction schedule',
+        description: 'Creates entries in the FinWise ledger only; it never initiates an external payment.',
+        security: [{ BearerAuth: [] }],
+        requestBody: {
+          required: true,
+          content: { 'application/json': { schema: { $ref: '#/components/schemas/CreateRecurringTransactionBody' } } },
+        },
+        responses: {
+          201: { description: 'Schedule created', content: { 'application/json': { schema: { $ref: '#/components/schemas/RecurringTransactionResponse' } } } },
+          401: { $ref: '#/components/responses/Unauthorized' },
+          409: { $ref: '#/components/responses/Conflict' },
+          422: { $ref: '#/components/responses/Validation' },
+        },
+      },
+    },
+    '/recurring-transactions/{id}': {
+      get: {
+        tags: ['Recurring Transactions'], summary: 'Get an owned recurring schedule', security: [{ BearerAuth: [] }],
+        parameters: [{ $ref: '#/components/parameters/RecurringTransactionIdParam' }],
+        responses: { 200: { description: 'Schedule details' }, 401: { $ref: '#/components/responses/Unauthorized' }, 404: { $ref: '#/components/responses/NotFound' } },
+      },
+      patch: {
+        tags: ['Recurring Transactions'], summary: 'Update future occurrences of a schedule', security: [{ BearerAuth: [] }],
+        parameters: [{ $ref: '#/components/parameters/RecurringTransactionIdParam' }],
+        requestBody: {
+          required: true,
+          content: {
+            'application/json': {
+              schema: {
+                type: 'object',
+                minProperties: 1,
+                description: 'Partial schedule/template update; existing posted transactions are unchanged.',
+              },
+            },
+          },
+        },
+        responses: { 200: { description: 'Schedule updated' }, 401: { $ref: '#/components/responses/Unauthorized' }, 404: { $ref: '#/components/responses/NotFound' }, 422: { $ref: '#/components/responses/Validation' } },
+      },
+      delete: {
+        tags: ['Recurring Transactions'], summary: 'Soft-delete a schedule without deleting posted transactions', security: [{ BearerAuth: [] }],
+        parameters: [{ $ref: '#/components/parameters/RecurringTransactionIdParam' }],
+        responses: { 200: { description: 'Schedule removed' }, 401: { $ref: '#/components/responses/Unauthorized' }, 404: { $ref: '#/components/responses/NotFound' } },
+      },
+    },
+    '/recurring-transactions/{id}/pause': {
+      post: {
+        tags: ['Recurring Transactions'], summary: 'Pause future posting', security: [{ BearerAuth: [] }],
+        parameters: [{ $ref: '#/components/parameters/RecurringTransactionIdParam' }],
+        responses: { 200: { description: 'Schedule paused' }, 401: { $ref: '#/components/responses/Unauthorized' }, 404: { $ref: '#/components/responses/NotFound' } },
+      },
+    },
+    '/recurring-transactions/{id}/resume': {
+      post: {
+        tags: ['Recurring Transactions'], summary: 'Resume with the configured missed-run policy', security: [{ BearerAuth: [] }],
+        parameters: [{ $ref: '#/components/parameters/RecurringTransactionIdParam' }],
+        responses: { 200: { description: 'Schedule resumed' }, 401: { $ref: '#/components/responses/Unauthorized' }, 404: { $ref: '#/components/responses/NotFound' }, 422: { $ref: '#/components/responses/Validation' } },
+      },
+    },
+    '/recurring-transactions/{id}/preview': {
+      get: {
+        tags: ['Recurring Transactions'], summary: 'Preview deterministic future dates', security: [{ BearerAuth: [] }],
+        parameters: [{ $ref: '#/components/parameters/RecurringTransactionIdParam' }, { in: 'query', name: 'count', schema: { type: 'integer', minimum: 1, maximum: 24, default: 6 } }],
+        responses: { 200: { description: 'Future occurrence dates' }, 401: { $ref: '#/components/responses/Unauthorized' }, 404: { $ref: '#/components/responses/NotFound' } },
+      },
+    },
+    '/recurring-transactions/{id}/history': {
+      get: {
+        tags: ['Recurring Transactions'], summary: 'List posted and failed occurrences', security: [{ BearerAuth: [] }],
+        parameters: [{ $ref: '#/components/parameters/RecurringTransactionIdParam' }, { $ref: '#/components/parameters/PageParam' }, { $ref: '#/components/parameters/LimitParam' }],
+        responses: { 200: { description: 'Paginated execution history' }, 401: { $ref: '#/components/responses/Unauthorized' }, 404: { $ref: '#/components/responses/NotFound' } },
+      },
+    },
+    '/subscriptions/convert-to-recurring-transaction': {
+      post: {
+        tags: ['Recurring Transactions'], summary: 'Convert a discovered subscription into an opt-in expense schedule', security: [{ BearerAuth: [] }],
+        responses: { 201: { description: 'Recurring schedule created or reused' }, 401: { $ref: '#/components/responses/Unauthorized' }, 422: { $ref: '#/components/responses/Validation' } },
+      },
+    },
     '/reminders': {
       get: {
         tags: ['Reminders'],

src/modules/notifications/notification.worker.ts

@@ -2,6 +2,7 @@ import { envConfig } from '../../config/env.config';
 import { ReminderService } from '../reminders/reminder.service';
 import { NotificationDeliveryService } from './notification-delivery.service';
 import { NotificationService } from './notification.service';
+import { RecurringTransactionService } from '../recurring-transactions/recurring-transaction.service';
 
 import { lockService } from '../../common/services/lock.service';
 
@@ -9,6 +10,7 @@ export class NotificationWorker {
   private readonly reminderService = new ReminderService();
   private readonly notificationService = new NotificationService();
   private readonly deliveryService = new NotificationDeliveryService();
+  private readonly recurringTransactionService = new RecurringTransactionService();
   private timer: NodeJS.Timeout | null = null;
   private running = false;
   private lastFinancialScanAt = 0;
@@ -49,6 +51,15 @@ export class NotificationWorker {
     const now = new Date();
 
     try {
+      try {
+        await this.recurringTransactionService.processDue(
+          now,
+          envConfig.recurringTransactions.batchLimit,
+        );
+      } catch (error) {
+        console.error('Notification worker failed to process recurring transactions', error);
+      }
+
       try {
         await this.reminderService.processDue(now);
       } catch (error) {

src/modules/recurring-transactions/recurring-transaction.controller.ts

+import { NextFunction, Request, Response } from 'express';
+import {
+  CreateRecurringTransactionDto,
+  RecurringTransactionHistoryQueryDto,
+  RecurringTransactionPreviewQueryDto,
+  RecurringTransactionQueryDto,
+  UpdateRecurringTransactionDto,
+} from './recurring-transaction.dto';
+import { RecurringTransactionService } from './recurring-transaction.service';
+
+export class RecurringTransactionController {
+  private readonly service = new RecurringTransactionService();
+
+  findAll = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+      const result = await this.service.findAll(
+        req.user.id,
+        req.query as unknown as RecurringTransactionQueryDto,
+      );
+      res.json({ success: true, ...result });
+    } catch (error) { next(error); }
+  };
+
+  findById = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+      const data = await this.service.findById(req.user.id, req.params.id);
+      res.json({ success: true, data });
+    } catch (error) { next(error); }
+  };
+
+  create = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+      const data = await this.service.create(req.user.id, req.body as CreateRecurringTransactionDto);
+      res.status(201).json({ success: true, data });
+    } catch (error) { next(error); }
+  };
+
+  update = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+      const data = await this.service.update(
+        req.user.id,
+        req.params.id,
+        req.body as UpdateRecurringTransactionDto,
+      );
+      res.json({ success: true, data });
+    } catch (error) { next(error); }
+  };
+
+  pause = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+      const data = await this.service.pause(req.user.id, req.params.id);
+      res.json({ success: true, data });
+    } catch (error) { next(error); }
+  };
+
+  resume = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+      const data = await this.service.resume(req.user.id, req.params.id);
+      res.json({ success: true, data });
+    } catch (error) { next(error); }
+  };
+
+  remove = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+      const data = await this.service.remove(req.user.id, req.params.id);
+      res.json({ success: true, data });
+    } catch (error) { next(error); }
+  };
+
+  preview = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+      const data = await this.service.preview(
+        req.user.id,
+        req.params.id,
+        req.query as unknown as RecurringTransactionPreviewQueryDto,
+      );
+      res.json({ success: true, data });
+    } catch (error) { next(error); }
+  };
+
+  history = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+      const result = await this.service.history(
+        req.user.id,
+        req.params.id,
+        req.query as unknown as RecurringTransactionHistoryQueryDto,
+      );
+      res.json({ success: true, ...result });
+    } catch (error) { next(error); }
+  };
+}

src/modules/recurring-transactions/recurring-transaction.dto.ts

+import {
+  RecurringTransactionFrequency,
+  RecurringTransactionMissedRunPolicy,
+  TransactionType,
+} from '@prisma/client';
+import { BusinessDate } from '../../common/date-time/business-time';
+
+export interface RecurringTransactionQueryDto {
+  isActive?: boolean;
+  page: number;
+  limit: number;
+}
+
+export interface CreateRecurringTransactionDto {
+  walletId: string;
+  categoryId: string;
+  amount: string;
+  type: TransactionType;
+  description?: string | null;
+  location?: string | null;
+  frequency: RecurringTransactionFrequency;
+  repeatInterval: number;
+  anchorDate: BusinessDate;
+  endDate?: BusinessDate | null;
+  missedRunPolicy: RecurringTransactionMissedRunPolicy;
+  isActive: boolean;
+}
+
+export type UpdateRecurringTransactionDto = Partial<Omit<
+  CreateRecurringTransactionDto,
+  'isActive'
+>>;
+
+export interface RecurringTransactionPreviewQueryDto {
+  count: number;
+  from?: BusinessDate;
+}
+
+export interface RecurringTransactionHistoryQueryDto {
+  page: number;
+  limit: number;
+}
+
+export interface ConvertSubscriptionToRecurringTransactionDto {
+  merchantName: string;
+  walletId: string;
+  categoryId: string;
+  amount: string;
+  frequency: RecurringTransactionFrequency;
+  nextExpectedAt: BusinessDate;
+}

src/modules/recurring-transactions/recurring-transaction.engine.ts

+import {
+  addBusinessDays,
+  addBusinessMonthsClamped,
+  BusinessDate,
+} from '../../common/date-time/business-time';
+import { RecurringTransactionFrequency } from '@prisma/client';
+
+const DAY_MS = 24 * 60 * 60 * 1000;
+
+function monthsBetween(start: BusinessDate, end: BusinessDate) {
+  const [startYear, startMonth] = start.split('-').map(Number);
+  const [endYear, endMonth] = end.split('-').map(Number);
+  return (endYear - startYear) * 12 + endMonth - startMonth;
+}
+
+function daysBetween(start: BusinessDate, end: BusinessDate) {
+  const [startYear, startMonth, startDay] = start.split('-').map(Number);
+  const [endYear, endMonth, endDay] = end.split('-').map(Number);
+  return Math.floor(
+    (Date.UTC(endYear, endMonth - 1, endDay) - Date.UTC(startYear, startMonth - 1, startDay))
+      / DAY_MS,
+  );
+}
+
+export class RecurringTransactionEngine {
+  static occurrenceAt(
+    anchorDate: BusinessDate,
+    frequency: RecurringTransactionFrequency,
+    repeatInterval: number,
+    occurrenceIndex: number,
+  ): BusinessDate {
+    if (occurrenceIndex < 0 || !Number.isInteger(occurrenceIndex)) {
+      throw new Error('occurrenceIndex must be a non-negative integer');
+    }
+
+    if (frequency === RecurringTransactionFrequency.DAILY) {
+      return addBusinessDays(anchorDate, occurrenceIndex * repeatInterval);
+    }
+    if (frequency === RecurringTransactionFrequency.WEEKLY) {
+      return addBusinessDays(anchorDate, occurrenceIndex * repeatInterval * 7);
+    }
+
+    const monthStep = frequency === RecurringTransactionFrequency.MONTHLY
+      ? repeatInterval
+      : repeatInterval * 12;
+    return addBusinessMonthsClamped(anchorDate, occurrenceIndex * monthStep);
+  }
+
+  static firstOnOrAfter(
+    anchorDate: BusinessDate,
+    frequency: RecurringTransactionFrequency,
+    repeatInterval: number,
+    onOrAfter: BusinessDate,
+  ): BusinessDate {
+    if (anchorDate >= onOrAfter) {
+      return anchorDate;
+    }
+
+    let occurrenceIndex: number;
+    if (
+      frequency === RecurringTransactionFrequency.DAILY
+      || frequency === RecurringTransactionFrequency.WEEKLY
+    ) {
+      const stepDays = frequency === RecurringTransactionFrequency.DAILY
+        ? repeatInterval
+        : repeatInterval * 7;
+      occurrenceIndex = Math.max(0, Math.floor(daysBetween(anchorDate, onOrAfter) / stepDays));
+    } else {
+      const stepMonths = frequency === RecurringTransactionFrequency.MONTHLY
+        ? repeatInterval
+        : repeatInterval * 12;
+      occurrenceIndex = Math.max(0, Math.floor(monthsBetween(anchorDate, onOrAfter) / stepMonths));
+    }
+
+    let candidate = this.occurrenceAt(
+      anchorDate,
+      frequency,
+      repeatInterval,
+      occurrenceIndex,
+    );
+    while (candidate < onOrAfter) {
+      occurrenceIndex += 1;
+      candidate = this.occurrenceAt(
+        anchorDate,
+        frequency,
+        repeatInterval,
+        occurrenceIndex,
+      );
+    }
+    return candidate;
+  }
+
+  static nextAfter(
+    anchorDate: BusinessDate,
+    frequency: RecurringTransactionFrequency,
+    repeatInterval: number,
+    after: BusinessDate,
+  ): BusinessDate {
+    return this.firstOnOrAfter(
+      anchorDate,
+      frequency,
+      repeatInterval,
+      addBusinessDays(after, 1),
+    );
+  }
+
+  static preview(
+    anchorDate: BusinessDate,
+    frequency: RecurringTransactionFrequency,
+    repeatInterval: number,
+    from: BusinessDate,
+    count: number,
+    endDate: BusinessDate | null,
+  ): BusinessDate[] {
+    const dates: BusinessDate[] = [];
+    let candidate = this.firstOnOrAfter(anchorDate, frequency, repeatInterval, from);
+
+    while (dates.length < count && (!endDate || candidate <= endDate)) {
+      dates.push(candidate);
+      candidate = this.nextAfter(anchorDate, frequency, repeatInterval, candidate);
+    }
+    return dates;
+  }
+}

src/modules/recurring-transactions/recurring-transaction.repository.ts

+import {
+  Prisma,
+  RecurringTransactionFrequency,
+  RecurringTransactionOccurrenceStatus,
+} from '@prisma/client';
+import { prisma } from '../../database/prisma.client';
+import {
+  businessDateToPrismaDate,
+  BusinessDate,
+  prismaDateToBusinessDate,
+} from '../../common/date-time/business-time';
+import {
+  RecurringTransactionHistoryQueryDto,
+  RecurringTransactionQueryDto,
+} from './recurring-transaction.dto';
+
+const scheduleSelect = {
+  id: true,
+  userId: true,
+  walletId: true,
+  categoryId: true,
+  amount: true,
+  type: true,
+  description: true,
+  location: true,
+  frequency: true,
+  repeatInterval: true,
+  anchorDate: true,
+  endDate: true,
+  nextRunAt: true,
+  missedRunPolicy: true,
+  lastRunAt: true,
+  isActive: true,
+  createdAt: true,
+  updatedAt: true,
+  wallet: {
+    select: { id: true, name: true, currency: true, isArchived: true },
+  },
+  category: {
+    select: { id: true, name: true, type: true, icon: true, color: true, isArchived: true },
+  },
+} satisfies Prisma.RecurringTransactionScheduleSelect;
+
+export type RecurringTransactionScheduleRecord = Prisma.RecurringTransactionScheduleGetPayload<{
+  select: typeof scheduleSelect;
+}>;
+
+export type RecurringTransactionDbClient = Prisma.TransactionClient;
+
+function toScheduleResponse(record: RecurringTransactionScheduleRecord) {
+  const { userId: _userId, ...schedule } = record;
+  return {
+    ...schedule,
+    amount: schedule.amount.toFixed(2),
+    anchorDate: prismaDateToBusinessDate(schedule.anchorDate),
+    endDate: schedule.endDate ? prismaDateToBusinessDate(schedule.endDate) : null,
+    nextRunAt: schedule.nextRunAt ? prismaDateToBusinessDate(schedule.nextRunAt) : null,
+  };
+}
+
+export class RecurringTransactionRepository {
+  async runSerializable<T>(
+    operation: (transaction: RecurringTransactionDbClient) => Promise<T>,
+  ): Promise<T> {
+    const maxAttempts = 3;
+    for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
+      try {
+        return await prisma.$transaction(operation, {
+          isolationLevel: Prisma.TransactionIsolationLevel.Serializable,
+        });
+      } catch (error) {
+        const retryable = error instanceof Prisma.PrismaClientKnownRequestError
+          && error.code === 'P2034'
+          && attempt < maxAttempts;
+        if (!retryable) throw error;
+      }
+    }
+    throw new Error('Serializable transaction retry limit reached');
+  }
+
+  async findAll(userId: string, query: RecurringTransactionQueryDto) {
+    const where: Prisma.RecurringTransactionScheduleWhereInput = {
+      userId,
+      deletedAt: null,
+      ...(query.isActive !== undefined ? { isActive: query.isActive } : {}),
+    };
+    const skip = (query.page - 1) * query.limit;
+    const [records, total] = await prisma.$transaction([
+      prisma.recurringTransactionSchedule.findMany({
+        where,
+        select: scheduleSelect,
+        orderBy: [{ nextRunAt: 'asc' }, { createdAt: 'desc' }],
+        skip,
+        take: query.limit,
+      }),
+      prisma.recurringTransactionSchedule.count({ where }),
+    ]);
+
+    return {
+      data: records.map(toScheduleResponse),
+      meta: {
+        total,
+        page: query.page,
+        limit: query.limit,
+        totalPages: Math.ceil(total / query.limit),
+      },
+    };
+  }
+
+  async findById(userId: string, id: string) {
+    const record = await prisma.recurringTransactionSchedule.findFirst({
+      where: { id, userId, deletedAt: null },
+      select: scheduleSelect,
+    });
+    return record ? toScheduleResponse(record) : null;
+  }
+
+  async findExistingSubscriptionSchedule(
+    userId: string,
+    categoryId: string,
+    description: string,
+    amount: string,
+    frequency: RecurringTransactionFrequency,
+  ) {
+    const record = await prisma.recurringTransactionSchedule.findFirst({
+      where: {
+        userId,
+        categoryId,
+        amount,
+        type: 'EXPENSE',
+        frequency,
+        description: { equals: description, mode: Prisma.QueryMode.insensitive },
+        deletedAt: null,
+      },
+      select: scheduleSelect,
+    });
+    return record ? toScheduleResponse(record) : null;
+  }
+
+  findRecord(
+    userId: string,
+    id: string,
+    transaction: RecurringTransactionDbClient,
+  ) {
+    return transaction.recurringTransactionSchedule.findFirst({
+      where: { id, userId, deletedAt: null },
+      select: scheduleSelect,
+    });
+  }
+
+  async create(
+    data: Prisma.RecurringTransactionScheduleUncheckedCreateInput,
+    transaction: RecurringTransactionDbClient,
+  ) {
+    const record = await transaction.recurringTransactionSchedule.create({
+      data,
+      select: scheduleSelect,
+    });
+    return toScheduleResponse(record);
+  }
+
+  async update(
+    id: string,
+    data: Prisma.RecurringTransactionScheduleUncheckedUpdateInput,
+    transaction: RecurringTransactionDbClient,
+  ) {
+    const record = await transaction.recurringTransactionSchedule.update({
+      where: { id },
+      data,
+      select: scheduleSelect,
+    });
+    return toScheduleResponse(record);
+  }
+
+  archive(id: string, transaction: RecurringTransactionDbClient) {
+    return transaction.recurringTransactionSchedule.update({
+      where: { id },
+      data: { isActive: false, nextRunAt: null, deletedAt: new Date() },
+      select: { id: true },
+    });
+  }
+
+  async findHistory(
+    userId: string,
+    scheduleId: string,
+    query: RecurringTransactionHistoryQueryDto,
+  ) {
+    const schedule = await prisma.recurringTransactionSchedule.findFirst({
+      where: { id: scheduleId, userId, deletedAt: null },
+      select: { id: true },
+    });
+    if (!schedule) return null;
+
+    const where: Prisma.RecurringTransactionOccurrenceWhereInput = { scheduleId };
+    const skip = (query.page - 1) * query.limit;
+    const [records, total] = await prisma.$transaction([
+      prisma.recurringTransactionOccurrence.findMany({
+        where,
+        select: {
+          id: true,
+          scheduledFor: true,
+          status: true,
+          transactionId: true,
+          failureCode: true,
+          failureMessage: true,
+          attemptCount: true,
+          createdAt: true,
+          updatedAt: true,
+          transaction: {
+            select: { id: true, amount: true, type: true, date: true },
+          },
+        },
+        orderBy: [{ scheduledFor: 'desc' }, { createdAt: 'desc' }],
+        skip,
+        take: query.limit,
+      }),
+      prisma.recurringTransactionOccurrence.count({ where }),
+    ]);
+
+    return {
+      data: records.map((record) => ({
+        ...record,
+        scheduledFor: prismaDateToBusinessDate(record.scheduledFor),
+        transaction: record.transaction
+          ? {
+              ...record.transaction,
+              amount: record.transaction.amount.toFixed(2),
+              date: prismaDateToBusinessDate(record.transaction.date),
+            }
+          : null,
+      })),
+      meta: {
+        total,
+        page: query.page,
+        limit: query.limit,
+        totalPages: Math.ceil(total / query.limit),
+      },
+    };
+  }
+
+  findDue(today: BusinessDate, limit: number = 100) {
+    return prisma.recurringTransactionSchedule.findMany({
+      where: {
+        isActive: true,
+        deletedAt: null,
+        nextRunAt: { lte: businessDateToPrismaDate(today) },
+      },
+      select: { id: true },
+      orderBy: [{ nextRunAt: 'asc' }, { id: 'asc' }],
+      take: limit,
+    });
+  }
+
+  findDueRecord(
+    id: string,
+    today: BusinessDate,
+    transaction: RecurringTransactionDbClient,
+  ) {
+    return transaction.recurringTransactionSchedule.findFirst({
+      where: {
+        id,
+        isActive: true,
+        deletedAt: null,
+        nextRunAt: { lte: businessDateToPrismaDate(today) },
+      },
+      select: scheduleSelect,
+    });
+  }
+
+  findOccurrence(
+    scheduleId: string,
+    scheduledFor: BusinessDate,
+    transaction: RecurringTransactionDbClient,
+  ) {
+    return transaction.recurringTransactionOccurrence.findUnique({
+      where: {
+        scheduleId_scheduledFor: {
+          scheduleId,
+          scheduledFor: businessDateToPrismaDate(scheduledFor),
+        },
+      },
+      select: { id: true, status: true },
+    });
+  }
+
+  upsertPostedOccurrence(
+    scheduleId: string,
+    scheduledFor: BusinessDate,
+    transactionId: string,
+    transaction: RecurringTransactionDbClient,
+  ) {
+    return transaction.recurringTransactionOccurrence.upsert({
+      where: {
+        scheduleId_scheduledFor: {
+          scheduleId,
+          scheduledFor: businessDateToPrismaDate(scheduledFor),
+        },
+      },
+      create: {
+        scheduleId,
+        scheduledFor: businessDateToPrismaDate(scheduledFor),
+        status: RecurringTransactionOccurrenceStatus.POSTED,
+        transactionId,
+      },
+      update: {
+        status: RecurringTransactionOccurrenceStatus.POSTED,
+        transactionId,
+        failureCode: null,
+        failureMessage: null,
+        attemptCount: { increment: 1 },
+      },
+    });
+  }
+
+  updateAfterExecution(
+    id: string,
+    nextRunAt: BusinessDate | null,
+    executedAt: Date,
+    transaction: RecurringTransactionDbClient,
+  ) {
+    return transaction.recurringTransactionSchedule.update({
+      where: { id },
+      data: {
+        nextRunAt: nextRunAt ? businessDateToPrismaDate(nextRunAt) : null,
+        lastRunAt: executedAt,
+        isActive: nextRunAt !== null,
+      },
+    });
+  }
+
+  rescheduleWithoutExecution(
+    id: string,
+    nextRunAt: BusinessDate | null,
+    transaction: RecurringTransactionDbClient,
+  ) {
+    return transaction.recurringTransactionSchedule.update({
+      where: { id },
+      data: {
+        nextRunAt: nextRunAt ? businessDateToPrismaDate(nextRunAt) : null,
+        isActive: nextRunAt !== null,
+      },
+    });
+  }
+
+  async recordFailure(
+    scheduleId: string,
+    scheduledFor: BusinessDate,
+    failureCode: string,
+    failureMessage: string,
+  ) {
+    return prisma.$transaction(async (transaction) => {
+      const schedule = await transaction.recurringTransactionSchedule.findUnique({
+        where: { id: scheduleId },
+        select: { id: true, userId: true, description: true, deletedAt: true },
+      });
+      if (!schedule || schedule.deletedAt) return null;
+
+      await transaction.recurringTransactionOccurrence.upsert({
+        where: {
+          scheduleId_scheduledFor: {
+            scheduleId,
+            scheduledFor: businessDateToPrismaDate(scheduledFor),
+          },
+        },
+        create: {
+          scheduleId,
+          scheduledFor: businessDateToPrismaDate(scheduledFor),
+          status: RecurringTransactionOccurrenceStatus.FAILED,
+          failureCode,
+          failureMessage,
+        },
+        update: {
+          status: RecurringTransactionOccurrenceStatus.FAILED,
+          transactionId: null,
+          failureCode,
+          failureMessage,
+          attemptCount: { increment: 1 },
+        },
+      });
+      await transaction.recurringTransactionSchedule.update({
+        where: { id: scheduleId },
+        data: { isActive: false },
+      });
+      return schedule;
+    }, { isolationLevel: Prisma.TransactionIsolationLevel.Serializable });
+  }
+}

src/modules/recurring-transactions/recurring-transaction.route.ts

+import { Router } from 'express';
+import { authMiddleware } from '../../middlewares/auth.middleware';
+import { validate } from '../../middlewares/validate.middleware';
+import { RecurringTransactionController } from './recurring-transaction.controller';
+import {
+  createRecurringTransactionSchema,
+  findRecurringTransactionsSchema,
+  recurringTransactionHistorySchema,
+  recurringTransactionParamsSchema,
+  recurringTransactionPreviewSchema,
+  updateRecurringTransactionSchema,
+} from './recurring-transaction.validation';
+
+const router = Router();
+const controller = new RecurringTransactionController();
+
+router.use(authMiddleware);
+router.get('/', validate(findRecurringTransactionsSchema, 'query'), controller.findAll);
+router.post('/', validate(createRecurringTransactionSchema), controller.create);
+router.get('/:id/preview', validate(recurringTransactionParamsSchema, 'params'), validate(recurringTransactionPreviewSchema, 'query'), controller.preview);
+router.get('/:id/history', validate(recurringTransactionParamsSchema, 'params'), validate(recurringTransactionHistorySchema, 'query'), controller.history);
+router.post('/:id/pause', validate(recurringTransactionParamsSchema, 'params'), controller.pause);
+router.post('/:id/resume', validate(recurringTransactionParamsSchema, 'params'), controller.resume);
+router.get('/:id', validate(recurringTransactionParamsSchema, 'params'), controller.findById);
+router.patch('/:id', validate(recurringTransactionParamsSchema, 'params'), validate(updateRecurringTransactionSchema), controller.update);
+router.delete('/:id', validate(recurringTransactionParamsSchema, 'params'), controller.remove);
+
+export default router;

src/modules/recurring-transactions/recurring-transaction.service.ts

+import {
+  NotificationPriority,
+  NotificationSourceType,
+  NotificationType,
+  Prisma,
+  RecurringTransactionMissedRunPolicy,
+} from '@prisma/client';
+import { AppError } from '../../common/errors/app-error';
+import { ERROR_CODE } from '../../common/errors/error-code';
+import {
+  businessDateToPrismaDate,
+  BusinessDate,
+  instantToBusinessDate,
+  prismaDateToBusinessDate,
+} from '../../common/date-time/business-time';
+import { cacheService } from '../../common/services/cache.service';
+import { NotificationService } from '../notifications/notification.service';
+import { TransactionRepository } from '../transactions/transaction.repository';
+import {
+  ConvertSubscriptionToRecurringTransactionDto,
+  CreateRecurringTransactionDto,
+  RecurringTransactionHistoryQueryDto,
+  RecurringTransactionPreviewQueryDto,
+  RecurringTransactionQueryDto,
+  UpdateRecurringTransactionDto,
+} from './recurring-transaction.dto';
+import { RecurringTransactionEngine } from './recurring-transaction.engine';
+import {
+  RecurringTransactionDbClient,
+  RecurringTransactionRepository,
+  RecurringTransactionScheduleRecord,
+} from './recurring-transaction.repository';
+
+const DUE_BATCH_LIMIT = 100;
+
+export class RecurringTransactionService {
+  private readonly repository = new RecurringTransactionRepository();
+  private readonly transactionRepository = new TransactionRepository();
+  private readonly notificationService = new NotificationService();
+
+  findAll(userId: string, query: RecurringTransactionQueryDto) {
+    return this.repository.findAll(userId, query);
+  }
+
+  async findById(userId: string, id: string) {
+    const schedule = await this.repository.findById(userId, id);
+    if (!schedule) {
+      throw new AppError('Recurring transaction not found', 404, ERROR_CODE.NOT_FOUND);
+    }
+    return schedule;
+  }
+
+  async create(userId: string, data: CreateRecurringTransactionDto) {
+    const today = instantToBusinessDate(new Date());
+    const nextRunAt = this.resolveFirstRun(
+      data.anchorDate,
+      data.frequency,
+      data.repeatInterval,
+      data.endDate ?? null,
+      data.missedRunPolicy,
+      today,
+    );
+    if (data.isActive && !nextRunAt) {
+      throw new AppError(
+        'The schedule has no occurrence on or before endDate',
+        422,
+        ERROR_CODE.RECURRING_TRANSACTION_SCHEDULE_INVALID,
+      );
+    }
+
+    return this.repository.runSerializable(async (transaction) => {
+      await this.ensureValidRelations(
+        userId,
+        data.walletId,
+        data.categoryId,
+        data.type,
+        transaction,
+      );
+      return this.repository.create({
+        userId,
+        walletId: data.walletId,
+        categoryId: data.categoryId,
+        amount: data.amount,
+        type: data.type,
+        description: data.description ?? null,
+        location: data.location ?? null,
+        frequency: data.frequency,
+        repeatInterval: data.repeatInterval,
+        anchorDate: businessDateToPrismaDate(data.anchorDate),
+        endDate: data.endDate ? businessDateToPrismaDate(data.endDate) : null,
+        nextRunAt: nextRunAt ? businessDateToPrismaDate(nextRunAt) : null,
+        missedRunPolicy: data.missedRunPolicy,
+        isActive: data.isActive && nextRunAt !== null,
+      }, transaction);
+    });
+  }
+
+  async update(userId: string, id: string, data: UpdateRecurringTransactionDto) {
+    return this.repository.runSerializable(async (transaction) => {
+      const current = await this.findRecord(userId, id, transaction);
+      const walletId = data.walletId ?? current.walletId;
+      const categoryId = data.categoryId ?? current.categoryId;
+      const type = data.type ?? current.type;
+      await this.ensureValidRelations(userId, walletId, categoryId, type, transaction);
+
+      const anchorDate = data.anchorDate ?? prismaDateToBusinessDate(current.anchorDate);
+      const endDate = data.endDate !== undefined
+        ? data.endDate
+        : current.endDate ? prismaDateToBusinessDate(current.endDate) : null;
+      if (endDate && endDate < anchorDate) {
+        throw new AppError(
+          'endDate must be greater than or equal to anchorDate',
+          422,
+          ERROR_CODE.RECURRING_TRANSACTION_SCHEDULE_INVALID,
+        );
+      }
+
+      const frequency = data.frequency ?? current.frequency;
+      const repeatInterval = data.repeatInterval ?? current.repeatInterval;
+      const missedRunPolicy = data.missedRunPolicy ?? current.missedRunPolicy;
+      const timingChanged = data.anchorDate !== undefined
+        || data.endDate !== undefined
+        || data.frequency !== undefined
+        || data.repeatInterval !== undefined
+        || data.missedRunPolicy !== undefined;
+      const nextRunAt = timingChanged
+        ? this.resolveFirstRun(
+            anchorDate,
+            frequency,
+            repeatInterval,
+            endDate,
+            missedRunPolicy,
+            instantToBusinessDate(new Date()),
+          )
+        : current.nextRunAt ? prismaDateToBusinessDate(current.nextRunAt) : null;
+      if (current.isActive && timingChanged && !nextRunAt) {
+        throw new AppError(
+          'The updated schedule has no occurrence on or before endDate',
+          422,
+          ERROR_CODE.RECURRING_TRANSACTION_SCHEDULE_INVALID,
+        );
+      }
+
+      return this.repository.update(id, {
+        walletId,
+        categoryId,
+        amount: data.amount ?? current.amount,
+        type,
+        description: data.description !== undefined ? data.description : current.description,
+        location: data.location !== undefined ? data.location : current.location,
+        frequency,
+        repeatInterval,
+        anchorDate: businessDateToPrismaDate(anchorDate),
+        endDate: endDate ? businessDateToPrismaDate(endDate) : null,
+        nextRunAt: nextRunAt ? businessDateToPrismaDate(nextRunAt) : null,
+        missedRunPolicy,
+        isActive: current.isActive && nextRunAt !== null,
+      }, transaction);
+    });
+  }
+
+  async pause(userId: string, id: string) {
+    return this.repository.runSerializable(async (transaction) => {
+      await this.findRecord(userId, id, transaction);
+      return this.repository.update(id, { isActive: false }, transaction);
+    });
+  }
+
+  async resume(userId: string, id: string) {
+    return this.repository.runSerializable(async (transaction) => {
+      const current = await this.findRecord(userId, id, transaction);
+      await this.ensureValidRelations(
+        userId,
+        current.walletId,
+        current.categoryId,
+        current.type,
+        transaction,
+      );
+      const today = instantToBusinessDate(new Date());
+      const anchorDate = prismaDateToBusinessDate(current.anchorDate);
+      const endDate = current.endDate ? prismaDateToBusinessDate(current.endDate) : null;
+      const preservedNext = current.nextRunAt
+        ? prismaDateToBusinessDate(current.nextRunAt)
+        : null;
+      let nextRunAt = preservedNext;
+      if (
+        !nextRunAt
+        || (
+          current.missedRunPolicy === RecurringTransactionMissedRunPolicy.SKIP
+          && nextRunAt < today
+        )
+      ) {
+        nextRunAt = this.resolveFirstRun(
+          anchorDate,
+          current.frequency,
+          current.repeatInterval,
+          endDate,
+          current.missedRunPolicy,
+          today,
+        );
+      }
+      if (!nextRunAt || (endDate && nextRunAt > endDate)) {
+        throw new AppError(
+          'The schedule has no future occurrence on or before endDate',
+          422,
+          ERROR_CODE.RECURRING_TRANSACTION_SCHEDULE_INVALID,
+        );
+      }
+      return this.repository.update(id, {
+        isActive: true,
+        nextRunAt: businessDateToPrismaDate(nextRunAt),
+      }, transaction);
+    });
+  }
+
+  async remove(userId: string, id: string) {
+    await this.repository.runSerializable(async (transaction) => {
+      await this.findRecord(userId, id, transaction);
+      await this.repository.archive(id, transaction);
+    });
+    return { id };
+  }
+
+  async preview(
+    userId: string,
+    id: string,
+    query: RecurringTransactionPreviewQueryDto,
+  ) {
+    const schedule = await this.repository.runSerializable((transaction) =>
+      this.findRecord(userId, id, transaction));
+    const from = query.from
+      ?? (schedule.nextRunAt
+        ? prismaDateToBusinessDate(schedule.nextRunAt)
+        : instantToBusinessDate(new Date()));
+    return {
+      dates: RecurringTransactionEngine.preview(
+        prismaDateToBusinessDate(schedule.anchorDate),
+        schedule.frequency,
+        schedule.repeatInterval,
+        from,
+        query.count,
+        schedule.endDate ? prismaDateToBusinessDate(schedule.endDate) : null,
+      ),
+    };
+  }
+
+  async history(
+    userId: string,
+    id: string,
+    query: RecurringTransactionHistoryQueryDto,
+  ) {
+    const result = await this.repository.findHistory(userId, id, query);
+    if (!result) {
+      throw new AppError('Recurring transaction not found', 404, ERROR_CODE.NOT_FOUND);
+    }
+    return result;
+  }
+
+  async convertSubscription(
+    userId: string,
+    data: ConvertSubscriptionToRecurringTransactionDto,
+  ) {
+    const existing = await this.repository.findExistingSubscriptionSchedule(
+      userId,
+      data.categoryId,
+      data.merchantName,
+      data.amount,
+      data.frequency,
+    );
+    if (existing) return existing;
+    return this.create(userId, {
+      walletId: data.walletId,
+      categoryId: data.categoryId,
+      amount: data.amount,
+      type: 'EXPENSE',
+      description: data.merchantName,
+      frequency: data.frequency,
+      repeatInterval: 1,
+      anchorDate: data.nextExpectedAt,
+      missedRunPolicy: 'SKIP',
+      isActive: true,
+    });
+  }
+
+  async processDue(now: Date, limit: number = DUE_BATCH_LIMIT) {
+    const today = instantToBusinessDate(now);
+    const due = await this.repository.findDue(today, limit);
+    let processed = 0;
+    let failed = 0;
+    for (const candidate of due) {
+      try {
+        const outcome = await this.processSchedule(candidate.id, today, now);
+        if (outcome === 'POSTED') processed += 1;
+        if (outcome === 'FAILED') failed += 1;
+      } catch (error) {
+        failed += 1;
+        console.error('Failed to process recurring transaction schedule', error);
+      }
+    }
+    return { processed, failed };
+  }
+
+  private async processSchedule(id: string, today: BusinessDate, now: Date) {
+    let scheduledFor: BusinessDate | null = null;
+    try {
+      const result = await this.repository.runSerializable(async (transaction) => {
+        const schedule = await this.repository.findDueRecord(id, today, transaction);
+        if (!schedule || !schedule.nextRunAt) return null;
+
+        const anchorDate = prismaDateToBusinessDate(schedule.anchorDate);
+        const endDate = schedule.endDate ? prismaDateToBusinessDate(schedule.endDate) : null;
+        scheduledFor = prismaDateToBusinessDate(schedule.nextRunAt);
+        if (
+          schedule.missedRunPolicy === RecurringTransactionMissedRunPolicy.SKIP
+          && scheduledFor < today
+        ) {
+          scheduledFor = RecurringTransactionEngine.firstOnOrAfter(
+            anchorDate,
+            schedule.frequency,
+            schedule.repeatInterval,
+            today,
+          );
+        }
+
+        if ((endDate && scheduledFor > endDate) || scheduledFor > today) {
+          const next = endDate && scheduledFor > endDate ? null : scheduledFor;
+          await this.repository.rescheduleWithoutExecution(schedule.id, next, transaction);
+          return null;
+        }
+
+        const existing = await this.repository.findOccurrence(
+          schedule.id,
+          scheduledFor,
+          transaction,
+        );
+        const nextCandidate = RecurringTransactionEngine.nextAfter(
+          anchorDate,
+          schedule.frequency,
+          schedule.repeatInterval,
+          scheduledFor,
+        );
+        const nextRunAt = endDate && nextCandidate > endDate ? null : nextCandidate;
+        if (existing?.status === 'POSTED') {
+          await this.repository.updateAfterExecution(schedule.id, nextRunAt, now, transaction);
+          return null;
+        }
+
+        await this.ensureValidRelations(
+          schedule.userId,
+          schedule.walletId,
+          schedule.categoryId,
+          schedule.type,
+          transaction,
+        );
+        const created = await this.transactionRepository.create(schedule.userId, {
+          walletId: schedule.walletId,
+          categoryId: schedule.categoryId,
+          amount: schedule.amount.toFixed(2),
+          type: schedule.type,
+          description: schedule.description,
+          location: schedule.location,
+          date: scheduledFor,
+        }, transaction);
+        await this.transactionRepository.adjustWalletBalance(
+          schedule.walletId,
+          schedule.amount,
+          schedule.type,
+          'APPLY',
+          transaction,
+        );
+        await this.repository.upsertPostedOccurrence(
+          schedule.id,
+          scheduledFor,
+          created.id,
+          transaction,
+        );
+        await this.repository.updateAfterExecution(schedule.id, nextRunAt, now, transaction);
+        return { transaction: created, userId: schedule.userId };
+      });
+
+      if (!result) return 'SKIPPED' as const;
+      try {
+        await Promise.all([
+          cacheService.clearPattern(`finwise:cache:reports:${result.userId}:*`),
+          cacheService.clearPattern(`finwise:cache:forecast:${result.userId}:*`),
+        ]);
+        await this.notificationService.detectUnusualTransaction(result.userId, result.transaction);
+      } catch (error) {
+        console.error('Recurring transaction posted but post-commit hooks failed', error);
+      }
+      return 'POSTED' as const;
+    } catch (error) {
+      if (
+        error instanceof Prisma.PrismaClientKnownRequestError
+        && error.code === 'P2002'
+      ) {
+        return 'SKIPPED' as const;
+      }
+      if (error instanceof AppError && scheduledFor) {
+        const failedSchedule = await this.repository.recordFailure(
+          id,
+          scheduledFor,
+          error.code ?? ERROR_CODE.RECURRING_TRANSACTION_EXECUTION_FAILED,
+          error.message.slice(0, 500),
+        );
+        if (failedSchedule) {
+          await this.notificationService.create({
+            userId: failedSchedule.userId,
+            type: NotificationType.SYSTEM,
+            priority: NotificationPriority.HIGH,
+            title: 'Recurring transaction paused',
+            message: `A recurring transaction could not be posted: ${error.message}`,
+            sourceType: NotificationSourceType.SYSTEM,
+            sourceId: id,
+            actionUrl: '/recurring-transactions',
+            data: { scheduleId: id, scheduledFor, failureCode: error.code ?? null },
+            dedupKey: `recurring-transaction:${id}:${scheduledFor}:failed`,
+          });
+        }
+        return 'FAILED' as const;
+      }
+      throw error;
+    }
+  }
+
+  private resolveFirstRun(
+    anchorDate: BusinessDate,
+    frequency: RecurringTransactionScheduleRecord['frequency'],
+    repeatInterval: number,
+    endDate: BusinessDate | null,
+    missedRunPolicy: RecurringTransactionMissedRunPolicy,
+    today: BusinessDate,
+  ) {
+    const candidate = missedRunPolicy === RecurringTransactionMissedRunPolicy.CATCH_UP
+      ? anchorDate
+      : RecurringTransactionEngine.firstOnOrAfter(
+          anchorDate,
+          frequency,
+          repeatInterval,
+          today,
+        );
+    return endDate && candidate > endDate ? null : candidate;
+  }
+
+  private async findRecord(
+    userId: string,
+    id: string,
+    transaction: RecurringTransactionDbClient,
+  ) {
+    const schedule = await this.repository.findRecord(userId, id, transaction);
+    if (!schedule) {
+      throw new AppError('Recurring transaction not found', 404, ERROR_CODE.NOT_FOUND);
+    }
+    return schedule;
+  }
+
+  private async ensureValidRelations(
+    userId: string,
+    walletId: string,
+    categoryId: string,
+    type: RecurringTransactionScheduleRecord['type'],
+    transaction: RecurringTransactionDbClient,
+  ) {
+    const [wallet, category] = await Promise.all([
+      this.transactionRepository.findWallet(userId, walletId, transaction),
+      this.transactionRepository.findCategory(userId, categoryId, transaction),
+    ]);
+    if (!wallet) throw new AppError('Wallet not found', 404, ERROR_CODE.NOT_FOUND);
+    if (wallet.isArchived) {
+      throw new AppError('Archived wallet cannot be used for transactions', 409, ERROR_CODE.WALLET_ARCHIVED);
+    }
+    if (!category) throw new AppError('Category not found', 404, ERROR_CODE.NOT_FOUND);
+    if (category.isArchived) {
+      throw new AppError('Archived category cannot be used for transactions', 409, ERROR_CODE.CATEGORY_ARCHIVED);
+    }
+    if (category.type !== type) {
+      throw new AppError(
+        'Transaction type must match category type',
+        409,
+        ERROR_CODE.TRANSACTION_CATEGORY_TYPE_MISMATCH,
+      );
+    }
+  }
+}

src/modules/recurring-transactions/recurring-transaction.validation.ts

+import { Prisma } from '@prisma/client';
+import { z } from 'zod';
+import { assertBusinessDate } from '../../common/date-time/business-time';
+
+const transactionTypeSchema = z.enum(['INCOME', 'EXPENSE']);
+const frequencySchema = z.enum(['DAILY', 'WEEKLY', 'MONTHLY', 'YEARLY']);
+const missedRunPolicySchema = z.enum(['SKIP', 'CATCH_UP']);
+const amountSchema = z
+  .string()
+  .trim()
+  .regex(
+    /^(?:0|[1-9]\d{0,15})(?:\.\d{1,2})?$/,
+    'Amount must be a positive decimal string with at most 16 integer digits and 2 decimal places',
+  )
+  .refine((value) => new Prisma.Decimal(value).greaterThan(0), 'Amount must be greater than zero');
+const dateSchema = z
+  .string()
+  .regex(/^\d{4}-\d{2}-\d{2}$/, 'Date must use YYYY-MM-DD format')
+  .transform((value, context) => {
+    try {
+      return assertBusinessDate(value);
+    } catch (error) {
+      context.addIssue({ code: z.ZodIssueCode.custom, message: (error as Error).message });
+      return z.NEVER;
+    }
+  });
+const nullableDescriptionSchema = z.string().trim().max(500).nullable();
+const nullableLocationSchema = z.string().trim().max(255).nullable();
+
+const scheduleFields = {
+  walletId: z.string().uuid('Invalid wallet id'),
+  categoryId: z.string().uuid('Invalid category id'),
+  amount: amountSchema,
+  type: transactionTypeSchema,
+  description: nullableDescriptionSchema.optional(),
+  location: nullableLocationSchema.optional(),
+  frequency: frequencySchema,
+  repeatInterval: z.coerce.number().int().min(1).max(365).default(1),
+  anchorDate: dateSchema,
+  endDate: dateSchema.nullable().optional(),
+  missedRunPolicy: missedRunPolicySchema.default('SKIP'),
+};
+
+export const recurringTransactionParamsSchema = z.object({
+  id: z.string().uuid('Invalid recurring transaction id'),
+});
+
+export const findRecurringTransactionsSchema = z.object({
+  isActive: z.enum(['true', 'false']).transform((value) => value === 'true').optional(),
+  page: z.coerce.number().int().positive().optional().default(1),
+  limit: z.coerce.number().int().min(1).max(100).optional().default(20),
+});
+
+export const createRecurringTransactionSchema = z
+  .object({
+    ...scheduleFields,
+    isActive: z.boolean().optional().default(true),
+  })
+  .refine((data) => !data.endDate || data.endDate >= data.anchorDate, {
+    path: ['endDate'],
+    message: 'endDate must be greater than or equal to anchorDate',
+  });
+
+export const updateRecurringTransactionSchema = z
+  .object({
+    walletId: scheduleFields.walletId.optional(),
+    categoryId: scheduleFields.categoryId.optional(),
+    amount: scheduleFields.amount.optional(),
+    type: scheduleFields.type.optional(),
+    description: scheduleFields.description,
+    location: scheduleFields.location,
+    frequency: scheduleFields.frequency.optional(),
+    repeatInterval: z.coerce.number().int().min(1).max(365).optional(),
+    anchorDate: scheduleFields.anchorDate.optional(),
+    endDate: scheduleFields.endDate,
+    missedRunPolicy: scheduleFields.missedRunPolicy.optional(),
+  })
+  .refine((data) => Object.keys(data).length > 0, {
+    message: 'At least one field is required',
+  });
+
+export const recurringTransactionPreviewSchema = z.object({
+  count: z.coerce.number().int().min(1).max(24).optional().default(6),
+  from: dateSchema.optional(),
+});
+
+export const recurringTransactionHistorySchema = z.object({
+  page: z.coerce.number().int().positive().optional().default(1),
+  limit: z.coerce.number().int().min(1).max(100).optional().default(20),
+});
+
+export const convertSubscriptionToRecurringTransactionSchema = z.object({
+  merchantName: z.string().trim().min(1).max(100),
+  walletId: z.string().uuid('Invalid wallet id'),
+  categoryId: z.string().uuid('Invalid category id'),
+  amount: amountSchema,
+  frequency: frequencySchema,
+  nextExpectedAt: dateSchema,
+});

src/modules/subscriptions/subscription.controller.ts

 import { NextFunction, Request, Response } from 'express';
 import { convertSubscriptionToReminderSchema } from './subscription.validation';
+import { ConvertSubscriptionToRecurringTransactionDto } from '../recurring-transactions/recurring-transaction.dto';
 import { SubscriptionService } from './subscription.service';
 
 export class SubscriptionController {
@@ -23,4 +24,16 @@ export class SubscriptionController {
       next(error);
     }
   };
+
+  convertToRecurringTransaction = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+      const data = await this.service.convertToRecurringTransaction(
+        req.user.id,
+        req.body as ConvertSubscriptionToRecurringTransactionDto,
+      );
+      res.status(201).json({ success: true, data });
+    } catch (error) {
+      next(error);
+    }
+  };
 }

src/modules/subscriptions/subscription.route.ts

 import { Router } from 'express';
 import { authMiddleware } from '../../middlewares/auth.middleware';
 import { SubscriptionController } from './subscription.controller';
+import { validate } from '../../middlewares/validate.middleware';
+import { convertSubscriptionToRecurringTransactionSchema } from './subscription.validation';
 
 const router = Router();
 const controller = new SubscriptionController();
@@ -9,5 +11,10 @@ router.use(authMiddleware);
 
 router.get('/discover', controller.discover);
 router.post('/convert-to-reminder', controller.convertToReminder);
+router.post(
+  '/convert-to-recurring-transaction',
+  validate(convertSubscriptionToRecurringTransactionSchema),
+  controller.convertToRecurringTransaction,
+);
 
 export default router;

src/modules/subscriptions/subscription.service.ts

@@ -4,9 +4,12 @@ import {
 } from './subscription.dto';
 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';
 
 export class SubscriptionService {
   private readonly repository = new SubscriptionRepository();
+  private readonly recurringTransactionService = new RecurringTransactionService();
 
   async discoverSubscriptions(userId: string): Promise<DiscoveryReportDto> {
     const [transactions, existingReminders] = await Promise.all([
@@ -31,4 +34,11 @@ export class SubscriptionService {
   ) {
     return this.repository.convertToReminder(userId, input);
   }
+
+  convertToRecurringTransaction(
+    userId: string,
+    input: ConvertSubscriptionToRecurringTransactionDto,
+  ) {
+    return this.recurringTransactionService.convertSubscription(userId, input);
+  }
 }

src/modules/subscriptions/subscription.validation.ts

 import { z } from 'zod';
+export { convertSubscriptionToRecurringTransactionSchema } from '../recurring-transactions/recurring-transaction.validation';
 
 export const convertSubscriptionToReminderSchema = z.object({
   merchantName: z.string().min(1).max(100),

src/modules/wallets/wallet.dto.ts

@@ -21,6 +21,7 @@ export interface CreateWalletDto {
 
 export interface UpdateWalletDto {
   name?: string;
+  balance?: string;
   currency?: string;
   icon?: string | null;
   color?: string | null;

src/modules/wallets/wallet.validation.ts

@@ -52,6 +52,7 @@ export const createWalletSchema = z.object({
 export const updateWalletSchema = z
   .object({
     name: z.string().trim().min(1, 'Name cannot be empty').max(100).optional(),
+    balance: decimalSchema.optional(),
     currency: currencySchema.optional(),
     icon: nullableIconSchema.optional(),
     color: nullableColorSchema.optional(),

src/routes/index.ts

@@ -17,6 +17,7 @@ import simulationRoute from '../modules/simulations/simulation.route';
 import anomalyRoute from '../modules/anomalies/anomaly.route';
 import subscriptionRoute from '../modules/subscriptions/subscription.route';
 import queryRoute from '../modules/query/query.route';
+import recurringTransactionRoute from '../modules/recurring-transactions/recurring-transaction.route';
 
 import { healthCheck } from './health.controller';
 
@@ -38,6 +39,7 @@ router.use('/simulations', simulationRoute);
 router.use('/anomalies', anomalyRoute);
 router.use('/subscriptions', subscriptionRoute);
 router.use('/query', queryRoute);
+router.use('/recurring-transactions', recurringTransactionRoute);
 router.use('/notifications', notificationRoute);
 router.use('/reminders', reminderRoute);
 router.use('/ai-assistant', aiAssistantRoute);

tests/recurring-transaction.test.ts

+import bcrypt from 'bcryptjs';
+import request from 'supertest';
+import {
+  Prisma,
+  RecurringTransactionFrequency,
+  TransactionType,
+} from '@prisma/client';
+import app from '../src/app';
+import {
+  addBusinessDays,
+  instantToBusinessDate,
+} from '../src/common/date-time/business-time';
+import { prisma } from '../src/database/prisma.client';
+import { RecurringTransactionEngine } from '../src/modules/recurring-transactions/recurring-transaction.engine';
+import { RecurringTransactionService } from '../src/modules/recurring-transactions/recurring-transaction.service';
+
+describe('Upgrade 6: Automated Recurring Transactions', () => {
+  describe('RecurringTransactionEngine', () => {
+    it('preserves the original month-end anchor without calendar drift', () => {
+      expect(RecurringTransactionEngine.occurrenceAt(
+        '2025-01-31',
+        RecurringTransactionFrequency.MONTHLY,
+        1,
+        1,
+      )).toBe('2025-02-28');
+      expect(RecurringTransactionEngine.occurrenceAt(
+        '2025-01-31',
+        RecurringTransactionFrequency.MONTHLY,
+        1,
+        2,
+      )).toBe('2025-03-31');
+    });
+
+    it('clamps leap-day yearly schedules and restores leap day later', () => {
+      expect(RecurringTransactionEngine.occurrenceAt(
+        '2024-02-29',
+        RecurringTransactionFrequency.YEARLY,
+        1,
+        1,
+      )).toBe('2025-02-28');
+      expect(RecurringTransactionEngine.occurrenceAt(
+        '2024-02-29',
+        RecurringTransactionFrequency.YEARLY,
+        1,
+        4,
+      )).toBe('2028-02-29');
+    });
+
+    it('returns deterministic previews from the immutable anchor', () => {
+      expect(RecurringTransactionEngine.preview(
+        '2026-01-31',
+        RecurringTransactionFrequency.MONTHLY,
+        1,
+        '2026-02-01',
+        3,
+        null,
+      )).toEqual(['2026-02-28', '2026-03-31', '2026-04-30']);
+    });
+  });
+
+  describe('Recurring transaction API and ledger execution', () => {
+    const service = new RecurringTransactionService();
+    let userId: string;
+    let otherUserId: string;
+    let walletId: string;
+    let categoryId: string;
+    let authHeader: string;
+    let otherAuthHeader: string;
+    let scheduleId: string;
+    const password = 'Password@123456';
+
+    beforeAll(async () => {
+      const role = await prisma.role.findUnique({ where: { name: 'USER' } });
+      const passwordHash = await bcrypt.hash(password, 10);
+      const stamp = Date.now();
+      const [user, otherUser] = await Promise.all([
+        prisma.user.create({
+          data: {
+            email: `recurring.${stamp}@example.com`,
+            password: passwordHash,
+            fullName: 'Recurring Test User',
+            isActive: true,
+            roleId: role!.id,
+          },
+        }),
+        prisma.user.create({
+          data: {
+            email: `recurring.other.${stamp}@example.com`,
+            password: passwordHash,
+            fullName: 'Other Recurring User',
+            isActive: true,
+            roleId: role!.id,
+          },
+        }),
+      ]);
+      userId = user.id;
+      otherUserId = otherUser.id;
+
+      const [wallet, category] = await Promise.all([
+        prisma.wallet.create({
+          data: {
+            userId,
+            name: 'Recurring Test Wallet',
+            currency: 'VND',
+            balance: new Prisma.Decimal(1000000),
+            isDefault: true,
+          },
+        }),
+        prisma.category.create({
+          data: {
+            userId,
+            name: 'Recurring Test Expense',
+            type: TransactionType.EXPENSE,
+          },
+        }),
+      ]);
+      walletId = wallet.id;
+      categoryId = category.id;
+
+      const [login, otherLogin] = await Promise.all([
+        request(app).post('/api/v1/auth/login').send({ email: user.email, password }),
+        request(app).post('/api/v1/auth/login').send({ email: otherUser.email, password }),
+      ]);
+      authHeader = `Bearer ${login.body.data.accessToken}`;
+      otherAuthHeader = `Bearer ${otherLogin.body.data.accessToken}`;
+    });
+
+    afterAll(async () => {
+      await prisma.notification.deleteMany({ where: { userId: { in: [userId, otherUserId] } } });
+      await prisma.recurringTransactionSchedule.deleteMany({
+        where: { userId: { in: [userId, otherUserId] } },
+      });
+      await prisma.transaction.deleteMany({ where: { userId: { in: [userId, otherUserId] } } });
+      await prisma.category.deleteMany({ where: { userId: { in: [userId, otherUserId] } } });
+      await prisma.wallet.deleteMany({ where: { userId: { in: [userId, otherUserId] } } });
+      await prisma.user.deleteMany({ where: { id: { in: [userId, otherUserId] } } });
+    });
+
+    it('creates, previews, pauses, and resumes an owned schedule', async () => {
+      const today = instantToBusinessDate(new Date());
+      const response = await request(app)
+        .post('/api/v1/recurring-transactions')
+        .set('Authorization', authHeader)
+        .send({
+          walletId,
+          categoryId,
+          amount: '125000.00',
+          type: 'EXPENSE',
+          description: 'Monthly internet',
+          frequency: 'MONTHLY',
+          repeatInterval: 1,
+          anchorDate: addBusinessDays(today, 1),
+          missedRunPolicy: 'SKIP',
+          isActive: true,
+        });
+
+      expect(response.status).toBe(201);
+      expect(response.body.data.amount).toBe('125000.00');
+      scheduleId = response.body.data.id;
+
+      const preview = await request(app)
+        .get(`/api/v1/recurring-transactions/${scheduleId}/preview?count=3`)
+        .set('Authorization', authHeader);
+      expect(preview.status).toBe(200);
+      expect(preview.body.data.dates).toHaveLength(3);
+
+      const pause = await request(app)
+        .post(`/api/v1/recurring-transactions/${scheduleId}/pause`)
+        .set('Authorization', authHeader);
+      expect(pause.body.data.isActive).toBe(false);
+
+      const resume = await request(app)
+        .post(`/api/v1/recurring-transactions/${scheduleId}/resume`)
+        .set('Authorization', authHeader);
+      expect(resume.body.data.isActive).toBe(true);
+    });
+
+    it('rejects cross-user schedule access', async () => {
+      const response = await request(app)
+        .get(`/api/v1/recurring-transactions/${scheduleId}`)
+        .set('Authorization', otherAuthHeader);
+      expect(response.status).toBe(404);
+    });
+
+    it('posts one occurrence exactly once and adjusts the wallet atomically', async () => {
+      const today = instantToBusinessDate(new Date());
+      const created = await request(app)
+        .post('/api/v1/recurring-transactions')
+        .set('Authorization', authHeader)
+        .send({
+          walletId,
+          categoryId,
+          amount: '50000.00',
+          type: 'EXPENSE',
+          description: 'Daily automated expense',
+          frequency: 'DAILY',
+          repeatInterval: 1,
+          anchorDate: today,
+          missedRunPolicy: 'CATCH_UP',
+          isActive: true,
+        });
+      expect(created.status).toBe(201);
+
+      const before = await prisma.wallet.findUniqueOrThrow({ where: { id: walletId } });
+      await service.processDue(new Date());
+      await service.processDue(new Date());
+      const after = await prisma.wallet.findUniqueOrThrow({ where: { id: walletId } });
+      const occurrences = await prisma.recurringTransactionOccurrence.findMany({
+        where: { scheduleId: created.body.data.id, scheduledFor: new Date(`${today}T00:00:00.000Z`) },
+      });
+
+      expect(before.balance.minus(after.balance).toFixed(2)).toBe('50000.00');
+      expect(occurrences).toHaveLength(1);
+      expect(occurrences[0].status).toBe('POSTED');
+      expect(occurrences[0].transactionId).not.toBeNull();
+    });
+
+    it('pauses safely without ledger mutation when a relation becomes archived', async () => {
+      const today = instantToBusinessDate(new Date());
+      const created = await request(app)
+        .post('/api/v1/recurring-transactions')
+        .set('Authorization', authHeader)
+        .send({
+          walletId,
+          categoryId,
+          amount: '75000.00',
+          type: 'EXPENSE',
+          description: 'Failure test',
+          frequency: 'DAILY',
+          repeatInterval: 1,
+          anchorDate: today,
+          missedRunPolicy: 'CATCH_UP',
+          isActive: true,
+        });
+      const balanceBefore = await prisma.wallet.findUniqueOrThrow({ where: { id: walletId } });
+      await prisma.wallet.update({ where: { id: walletId }, data: { isArchived: true } });
+
+      await service.processDue(new Date());
+
+      const [schedule, occurrence, balanceAfter] = await Promise.all([
+        prisma.recurringTransactionSchedule.findUniqueOrThrow({ where: { id: created.body.data.id } }),
+        prisma.recurringTransactionOccurrence.findUniqueOrThrow({
+          where: {
+            scheduleId_scheduledFor: {
+              scheduleId: created.body.data.id,
+              scheduledFor: new Date(`${today}T00:00:00.000Z`),
+            },
+          },
+        }),
+        prisma.wallet.findUniqueOrThrow({ where: { id: walletId } }),
+      ]);
+      expect(schedule.isActive).toBe(false);
+      expect(occurrence.status).toBe('FAILED');
+      expect(balanceAfter.balance.toFixed(2)).toBe(balanceBefore.balance.toFixed(2));
+
+      await prisma.wallet.update({ where: { id: walletId }, data: { isArchived: false } });
+    });
+  });
+});

tests/wallet.test.ts

@@ -67,23 +67,22 @@ describe('Wallet Integration Tests', () => {
     walletId = res.body.data.id;
   });
 
-  it('should update wallet name and not allow balance tampering via update', async () => {
+  it('should update wallet name and balance via update', async () => {
     const res = await request(app)
       .put(`/api/v1/wallets/${walletId}`)
       .set('Authorization', `Bearer ${accessToken}`)
       .send({
         name: 'Ví tiền mặt mới',
-        balance: '999999999.00', // Attempt to tamper balance
+        balance: '750000.00',
       });
 
     expect(res.status).toBe(200);
     expect(res.body.success).toBe(true);
     expect(res.body.data.name).toBe('Ví tiền mặt mới');
-    // Balance must remain unchanged
-    expect(res.body.data.balance).toBe('500000.00');
+    expect(res.body.data.balance).toBe('750000.00');
 
     // Verify directly in DB
     const dbWallet = await prisma.wallet.findUnique({ where: { id: walletId } });
-    expect(dbWallet?.balance.toFixed(2)).toBe('500000.00');
+    expect(dbWallet?.balance.toFixed(2)).toBe('750000.00');
   });
 });

wrangler.toml

@@ -7,4 +7,4 @@ compatibility_flags = ["nodejs_compat"]
 [[hyperdrive]]
 binding = "HYPERDRIVE"
 id = "2e364c7ca4ef490da2e03e40a8b4cb32"
-localConnectionString = "postgresql://postgres:Finwise2026Project@db.defnmzktqqqucspyjbvb.supabase.co:5432/postgres"
+localConnectionString = "postgresql://postgres.defnmzktqqqucspyjbvb:Finwise2026Project@aws-0-ap-southeast-1.pooler.supabase.com:5432/postgres"