fix(payments): keep the subscription sweep cursor across passes

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>

Author: Priyanshubhartistm

src/services/subscription-service.ts

@@ -12,7 +12,16 @@ const logger = createLogger('subscription-service')
 const SWEEP_PAGE_SIZE = 10
 const SWEEP_MAX_PAGES = 10
 
+type Sweep = 'ended' | 'grace' | 'renewal'
+
 export class SubscriptionService implements ISubscriptionService {
+  /**
+   * Where each sweep starts. Rows a pass could not handle stay at the front of its query, so the next
+   * pass starts behind them. Without this, more unhandled rows than one pass reads would hide every
+   * row after them. It starts over from the front once a pass reaches the end of the rows.
+   */
+  private readonly cursors: Record<Sweep, number> = { ended: 0, grace: 0, renewal: 0 }
+
   public constructor(
     private readonly userSubscriptionRepository: IUserSubscriptionRepository,
     private readonly paymentsService: IPaymentsService,
@@ -28,14 +37,17 @@ export class SubscriptionService implements ISubscriptionService {
     // Ends first, so a subscription that lapsed is not also sent a renewal invoice.
     const changed =
       (await this.sweep(
+        'ended',
         (limit, offset) => this.userSubscriptionRepository.findPeriodEnded(now, limit, offset),
         (subscription) => this.endPeriod(subscription, currentSettings, now),
       )) +
       (await this.sweep(
+        'grace',
         (limit, offset) => this.userSubscriptionRepository.findGraceEnded(now, limit, offset),
         (subscription) => this.expire(subscription),
       )) +
       (await this.sweep(
+        'renewal',
         (limit, offset) =>
           this.userSubscriptionRepository.findDueForRenewal(
             new Date(now.getTime() + RENEWAL_LEAD_SECONDS * 1000),
@@ -52,14 +64,16 @@ export class SubscriptionService implements ISubscriptionService {
    * Each handled subscription leaves the query that found it, so the next page starts where the
    * unhandled ones end: skipping them with an offset keeps a subscription that cannot be handled,
    * such as one on a plan the operator removed, from blocking the ones behind it. The work of one
-   * pass is bounded.
+   * pass is bounded, and the next one carries on behind the rows this one skipped.
    */
   private async sweep(
+    sweep: Sweep,
     find: (limit: number, offset: number) => Promise<UserSubscription[]>,
     handle: (subscription: UserSubscription) => Promise<boolean>,
   ): Promise<number> {
     let changed = 0
-    let skipped = 0
+    let skipped = this.cursors[sweep]
+    let reachedEnd = false
 
     for (let page = 0; page < SWEEP_MAX_PAGES; page++) {
       const subscriptions = await find(SWEEP_PAGE_SIZE, skipped)
@@ -80,10 +94,13 @@ export class SubscriptionService implements ISubscriptionService {
       }
 
       if (subscriptions.length < SWEEP_PAGE_SIZE) {
+        reachedEnd = true
         break
       }
     }
 
+    this.cursors[sweep] = reachedEnd ? 0 : skipped
+
     return changed
   }
 

test/unit/app/maintenance-worker.spec.ts

@@ -373,6 +373,7 @@ describe('MaintenanceWorker', () => {
           notificationOutboxService,
           notificationDeliveryLogRepository,
           notificationOutboxRepository,
+          undefined,
           subscriptionService as any,
         )
       })

test/unit/services/subscription-service.spec.ts

@@ -270,6 +270,33 @@ describe('SubscriptionService', () => {
         expect(repository.findPeriodEnded.secondCall.args.slice(1)).to.deep.equal([10, 10])
       })
 
+      it('carries on behind the rows it skipped in the next pass, so more stuck rows than one pass reads do not hide the rest', async () => {
+        repository.findPeriodEnded.resolves(page(10, { planId: 'retired' }))
+
+        await service.processLifecycle(NOW)
+
+        // The first pass read ten pages of stuck rows and gave up at the cap; the next starts behind them.
+        expect(repository.findPeriodEnded.firstCall.args.slice(1)).to.deep.equal([10, 0])
+        repository.findPeriodEnded.resetHistory()
+        repository.findPeriodEnded.resolves([])
+
+        await service.processLifecycle(NOW)
+
+        expect(repository.findPeriodEnded.firstCall.args.slice(1)).to.deep.equal([10, 100])
+      })
+
+      it('starts over from the front once a pass reaches the end of the rows', async () => {
+        repository.findPeriodEnded.onCall(0).resolves(page(10, { planId: 'retired' }))
+        repository.findPeriodEnded.onCall(1).resolves([])
+        await service.processLifecycle(NOW)
+        repository.findPeriodEnded.resetHistory()
+        repository.findPeriodEnded.resolves([])
+
+        await service.processLifecycle(NOW)
+
+        expect(repository.findPeriodEnded.firstCall.args.slice(1)).to.deep.equal([10, 0])
+      })
+
       it('counts only the rows left behind when handled ones drop out of the next query', async () => {
         const mixed = [...page(7), ...page(3, { planId: 'retired' })]
         repository.findPeriodEnded.onCall(0).resolves(mixed)