fix(listener): resolve notification count mismatch and optimize stale…

listener/src/services/scheduled-notification-repository.ts

@@ -122,29 +122,73 @@ export class ScheduledNotificationRepository {
    */
   async recoverStaleLocks(requestId?: string): Promise<number> {
     const now = new Date();
+    let recoveredCount = 0;
 
-    const sql = `
-      UPDATE scheduled_notifications
-      SET 
-        status = ?,
-        processor_id = NULL,
-        lock_expires_at = NULL
-      WHERE status = ?
-        AND lock_expires_at IS NOT NULL
-        AND lock_expires_at < ?
-    `;
-
-    const result = await this.db.run(sql, [
-      NotificationStatus.PENDING,
-      NotificationStatus.PROCESSING,
-      now.toISOString(),
-    ]);
+    await this.db.transaction(async () => {
+      const selectSql = `
+        SELECT * FROM scheduled_notifications
+        WHERE status = ?
+          AND lock_expires_at IS NOT NULL
+          AND lock_expires_at < ?
+      `;
+
+      const rows = await this.db.all<ScheduledNotificationRow>(selectSql, [
+        NotificationStatus.PROCESSING,
+        now.toISOString(),
+      ]);
+
+      recoveredCount = rows.length;
+
+      for (const row of rows) {
+        const model = this.rowToModel(row);
+        const newRetryCount = model.retryCount + 1;
+        const isFailed = newRetryCount >= model.maxRetries;
+        const newStatus = isFailed ? NotificationStatus.FAILED : NotificationStatus.PENDING;
+
+        const updateSql = `
+          UPDATE scheduled_notifications
+          SET 
+            status = ?,
+            retry_count = ?,
+            last_error = ?,
+            error_details = ?,
+            processing_completed_at = ?,
+            processor_id = NULL,
+            lock_expires_at = NULL
+          WHERE id = ?
+        `;
+
+        const errorMsg = 'Lock expired/Processor timeout';
+        const errorDetails = JSON.stringify({
+          message: errorMsg,
+          timestamp: now.toISOString(),
+        });
+
+        await this.db.run(updateSql, [
+          newStatus,
+          newRetryCount,
+          errorMsg,
+          errorDetails,
+          isFailed ? now.toISOString() : null,
+          model.id,
+        ]);
+
+        // Log execution attempt
+        await this.logExecution({
+          scheduledNotificationId: model.id!,
+          executionAttempt: newRetryCount,
+          executionTime: now,
+          status: isFailed ? 'FAILED' : 'RETRY',
+          errorMessage: errorMsg,
+        });
+      }
+    });
 
-    if (result.changes > 0) {
-      logger.warn('Recovered stale locks', { requestId, count: result.changes });
+    if (recoveredCount > 0) {
+      logger.warn('Recovered stale locks', { requestId, count: recoveredCount });
     }
 
-    return result.changes;
+    return recoveredCount;
   }
 
   /**
@@ -283,23 +327,28 @@ export class ScheduledNotificationRepository {
     failed: number;
     overdue: number;
   }> {
+    const now = new Date().toISOString();
+
     const countBySql = `
-      SELECT status, COUNT(*) as count
+      SELECT 
+        CASE 
+          WHEN status = 'PROCESSING' AND lock_expires_at IS NOT NULL AND lock_expires_at < ? THEN 'PENDING'
+          ELSE status
+        END AS adjusted_status,
+        COUNT(*) as count
       FROM scheduled_notifications
-      GROUP BY status
+      GROUP BY adjusted_status
     `;
 
     const overdueSql = `
       SELECT COUNT(*) as count
       FROM scheduled_notifications
-      WHERE status = ? AND execute_at < ?
+      WHERE (status = 'PENDING' OR (status = 'PROCESSING' AND lock_expires_at IS NOT NULL AND lock_expires_at < ?))
+        AND execute_at < ?
     `;
 
-    const counts = await this.db.all<{ status: string; count: number }>(countBySql);
-    const overdueResult = await this.db.get<{ count: number }>(overdueSql, [
-      NotificationStatus.PENDING,
-      new Date().toISOString(),
-    ]);
+    const counts = await this.db.all<{ adjusted_status: string; count: number }>(countBySql, [now]);
+    const overdueResult = await this.db.get<{ count: number }>(overdueSql, [now, now]);
 
     const stats = {
       pending: 0,
@@ -310,7 +359,7 @@ export class ScheduledNotificationRepository {
     };
 
     counts.forEach((row) => {
-      const status = row.status.toLowerCase();
+      const status = row.adjusted_status.toLowerCase();
       if (status in stats) {
         (stats as any)[status] = row.count;
       }

listener/src/tests/notification-scheduler-refactored.test.ts

@@ -230,6 +230,111 @@ describe('NotificationScheduler (Refactored)', () => {
       expect(stats.pending).toBe(2);
       expect(stats.overdue).toBe(1);
     });
+
+    test('should increment retry_count and log attempt on lock recovery', async () => {
+      const input = NotificationFixtureBuilder
+        .aScheduledNotificationInput()
+        .forImmediateExecution()
+        .withMaxRetries(3)
+        .build();
+
+      const id = await repository.create(input);
+
+      // Lock the notification
+      await repository.fetchAndLockPendingNotifications('processor-1', 30000, 10);
+
+      // Manually expire the lock
+      const pastLock = NotificationFixtureBuilder.dates.past(1000);
+      await db.run('UPDATE scheduled_notifications SET lock_expires_at = ? WHERE id = ?', [
+        pastLock.toISOString(),
+        id,
+      ]);
+
+      const recovered = await repository.recoverStaleLocks();
+      expect(recovered).toBe(1);
+
+      const notification = await repository.getById(id);
+      expect(notification!.status).toBe(NotificationStatus.PENDING);
+      expect(notification!.retryCount).toBe(1);
+      expect(notification!.lastError).toBe('Lock expired/Processor timeout');
+
+      // Verify execution log
+      const logs = await db.all('SELECT * FROM notification_execution_log WHERE scheduled_notification_id = ?', [id]);
+      expect(logs.length).toBe(1);
+      expect(logs[0].status).toBe('RETRY');
+      expect(logs[0].error_message).toBe('Lock expired/Processor timeout');
+    });
+
+    test('should mark as failed if retry_count reaches max_retries on lock recovery', async () => {
+      const input = NotificationFixtureBuilder
+        .aScheduledNotificationInput()
+        .forImmediateExecution()
+        .withMaxRetries(1)
+        .build();
+
+      const id = await repository.create(input);
+
+      // Lock the notification
+      await repository.fetchAndLockPendingNotifications('processor-1', 30000, 10);
+
+      // Manually expire the lock
+      const pastLock = NotificationFixtureBuilder.dates.past(1000);
+      await db.run('UPDATE scheduled_notifications SET lock_expires_at = ? WHERE id = ?', [
+        pastLock.toISOString(),
+        id,
+      ]);
+
+      const recovered = await repository.recoverStaleLocks();
+      expect(recovered).toBe(1);
+
+      const notification = await repository.getById(id);
+      expect(notification!.status).toBe(NotificationStatus.FAILED);
+      expect(notification!.retryCount).toBe(1);
+
+      // Verify execution log
+      const logs = await db.all('SELECT * FROM notification_execution_log WHERE scheduled_notification_id = ?', [id]);
+      expect(logs.length).toBe(1);
+      expect(logs[0].status).toBe('FAILED');
+    });
+
+    test('should return correct statistics accounting for stale locks', async () => {
+      // 1. Create a notification in the future (pending, not overdue)
+      await repository.create(
+        NotificationFixtureBuilder
+          .aScheduledNotificationInput()
+          .withExecuteAt(new Date(Date.now() + 3600000))
+          .build()
+      );
+
+      // 2. Create a notification in the past (overdue, pending)
+      await repository.create(
+        NotificationFixtureBuilder
+          .aScheduledNotificationInput()
+          .forImmediateExecution()
+          .build()
+      );
+
+      // 3. Create a notification in the past that is currently PROCESSING but lock is expired
+      const staleId = await repository.create(
+        NotificationFixtureBuilder
+          .aScheduledNotificationInput()
+          .forImmediateExecution()
+          .build()
+      );
+      await repository.fetchAndLockPendingNotifications('processor-1', 30000, 10);
+      const pastLock = NotificationFixtureBuilder.dates.past(1000);
+      await db.run('UPDATE scheduled_notifications SET lock_expires_at = ? WHERE id = ?', [
+        pastLock.toISOString(),
+        staleId,
+      ]);
+
+      // Get stats BEFORE recovery
+      const stats = await repository.getStats();
+
+      expect(stats.pending).toBe(3);
+      expect(stats.overdue).toBe(2);
+      expect(stats.processing).toBe(0);
+    });
   });
 
   describe('NotificationAPI', () => {