perf: coalesce batch progress refreshes

api/src/send-chain/send-chain.service.spec.ts

@@ -924,6 +924,28 @@ describe('SendChainService', () => {
     }));
   });
 
+  it('coalesces concurrent batch progress refreshes and keeps a trailing refresh', async () => {
+    const { service, prisma } = createService();
+    let resolveFirst: ((value: Array<{ status: string; _count: { _all: number } }>) => void) | undefined;
+    prisma.smsMessageRecord.groupBy
+      .mockImplementationOnce(() => new Promise((resolve) => { resolveFirst = resolve; }))
+      .mockResolvedValue([{ status: 'delivered', _count: { _all: 2 } }]);
+
+    const first = service['submission'].refreshTaskProgress('task-1');
+    await Promise.resolve();
+    const second = service['submission'].refreshTaskProgress('task-1');
+
+    expect(prisma.smsMessageRecord.groupBy).toHaveBeenCalledTimes(1);
+    resolveFirst?.([{ status: 'delivered', _count: { _all: 1 } }]);
+    await Promise.all([first, second]);
+
+    expect(prisma.smsMessageRecord.groupBy).toHaveBeenCalledTimes(2);
+    expect(prisma.smsBatchTask.update).toHaveBeenLastCalledWith({
+      where: { id: 'task-1' },
+      data: expect.objectContaining({ progressTotal: 2, successTotal: 2, status: 'finished' }),
+    });
+  });
+
   it('does not expose CMPP internal tasks through client task detail or messages', async () => {
     const { service, prisma } = createService();
     prisma.smsBatchTask.findFirst.mockResolvedValue(null);

api/src/send-chain/send-gateway-submit.service.ts

@@ -24,6 +24,8 @@ export class SendGatewaySubmitService {
   private sendQueue?: Queue<SendJob, unknown, 'send-message'>;
   private gatewayQueue?: Queue;
   private worker?: Worker<SendJob>;
+  private readonly taskProgressRefreshes = new Map<string, Promise<void>>();
+  private readonly dirtyTaskProgressRefreshes = new Set<string>();
 
   constructor(
     private readonly prisma: PrismaService,
@@ -449,6 +451,29 @@ async waitForChannelRateLimit(channelId: string, tps: number) {
   }
 
 async refreshTaskProgress(batchTaskId: string) {
+    const running = this.taskProgressRefreshes.get(batchTaskId);
+    if (running) {
+      // A state transition committed after the running aggregate may not be visible
+      // to its snapshot. Mark one trailing pass instead of issuing another identical
+      // GROUP BY concurrently for every message/result/receipt callback.
+      this.dirtyTaskProgressRefreshes.add(batchTaskId);
+      return running;
+    }
+    const refresh = this.refreshTaskProgressUntilClean(batchTaskId);
+    this.taskProgressRefreshes.set(batchTaskId, refresh);
+    try {
+      await refresh;
+    } finally {
+      if (this.taskProgressRefreshes.get(batchTaskId) === refresh) {
+        this.taskProgressRefreshes.delete(batchTaskId);
+      }
+      this.dirtyTaskProgressRefreshes.delete(batchTaskId);
+    }
+  }
+
+private async refreshTaskProgressUntilClean(batchTaskId: string) {
+  do {
+    this.dirtyTaskProgressRefreshes.delete(batchTaskId);
     const groups = await this.prisma.smsMessageRecord.groupBy({
       by: ['status'],
       where: { batchTaskId },
@@ -468,6 +493,7 @@ async refreshTaskProgress(batchTaskId: string) {
       where: { id: batchTaskId },
       data: { progressTotal, submittedTotal, successTotal, failedTotal, unknownTotal, timeoutTotal, status },
     });
+  } while (this.dirtyTaskProgressRefreshes.has(batchTaskId));
   }
 
 getSendQueue(): Queue<SendJob, unknown, 'send-message'> {