fix: ack protocol logs after complete flush

api/src/protocol-logs/protocol-logs.service.spec.ts

@@ -81,13 +81,53 @@ describe('ProtocolLogsService', () => {
     const service = new ProtocolLogsService(prisma as never);
     await service.list({ createdAtFrom: '2026-08-21', createdAtTo: '2026-08-27' });
 
-    expect(prisma.protocolInteractionLog.findMany).toHaveBeenCalledWith(expect.objectContaining({
-      where: expect.objectContaining({
-        createdAt: {
-          gte: new Date('2026-08-20T16:00:00.000Z'),
-          lte: new Date('2026-08-27T15:59:59.999Z'),
-        },
+    expect(prisma.protocolInteractionLog.findMany).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: expect.objectContaining({
+          createdAt: {
+            gte: new Date('2026-08-20T16:00:00.000Z'),
+            lte: new Date('2026-08-27T15:59:59.999Z'),
+          },
+        }),
       }),
-    }));
+    );
+  });
+
+  it('waits for an active flush and persists every buffered batch before reporting success', async () => {
+    const previousBatchSize = process.env.PROTOCOL_LOG_BATCH_SIZE;
+    process.env.PROTOCOL_LOG_BATCH_SIZE = '2';
+    let releaseFirstBatch!: (value: { count: number }) => void;
+    prisma.protocolInteractionLog.createMany
+      .mockImplementationOnce(
+        () =>
+          new Promise((resolve) => {
+            releaseFirstBatch = resolve;
+          }),
+      )
+      .mockResolvedValue({ count: 1 });
+
+    try {
+      const service = new ProtocolLogsService(prisma as never);
+      service.recordMany(
+        [1, 2, 3].map((sequenceId) => ({
+          protocol: 'cmpp' as const,
+          direction: 'channel_to_platform' as const,
+          eventType: 'deliver_receipt',
+          status: 'success' as const,
+          detail: { sequenceId },
+        })),
+      );
+
+      const completed = service.flushNow();
+      expect(prisma.protocolInteractionLog.createMany).toHaveBeenCalledTimes(1);
+      releaseFirstBatch({ count: 2 });
+
+      await expect(completed).resolves.toBe(true);
+      expect(prisma.protocolInteractionLog.createMany).toHaveBeenCalledTimes(2);
+      expect(prisma.protocolInteractionLog.createMany.mock.calls.flatMap(([input]) => input.data)).toHaveLength(3);
+    } finally {
+      if (previousBatchSize === undefined) delete process.env.PROTOCOL_LOG_BATCH_SIZE;
+      else process.env.PROTOCOL_LOG_BATCH_SIZE = previousBatchSize;
+    }
   });
 });

api/src/protocol-logs/protocol-logs.service.ts

@@ -47,7 +47,7 @@ export class ProtocolLogsService implements OnModuleInit, OnModuleDestroy {
   private readonly buffer: Prisma.ProtocolInteractionLogCreateManyInput[] = [];
   private flushTimer?: ReturnType<typeof setInterval>;
   private retentionTimer?: ReturnType<typeof setInterval>;
-  private flushing = false;
+  private flushPromise?: Promise<boolean>;
 
   constructor(private readonly prisma: PrismaService) {}
 
@@ -142,22 +142,32 @@ export class ProtocolLogsService implements OnModuleInit, OnModuleDestroy {
     return { items, total, page, pageSize, eventTypes: eventTypes.map((item) => item.eventType) };
   }
 
-  private async flush(): Promise<boolean> {
-    if (this.flushing) return false;
-    if (this.buffer.length === 0) return true;
-    this.flushing = true;
-    const batch = this.buffer.splice(0, positiveEnv('PROTOCOL_LOG_BATCH_SIZE', 100));
-    try {
-      await this.prisma.protocolInteractionLog.createMany({ data: batch });
-    } catch (error) {
-	  this.buffer.unshift(...batch);
-      this.logger.error(`Protocol log batch write failed (${batch.length} events)`, error instanceof Error ? error.stack : String(error));
-	  return false;
-    } finally {
-      this.flushing = false;
-      if (this.buffer.length > 0) setImmediate(() => void this.flush());
+  private flush(): Promise<boolean> {
+    if (this.flushPromise) return this.flushPromise;
+    if (this.buffer.length === 0) return Promise.resolve(true);
+
+    const flushPromise = this.flushBufferedBatches().finally(() => {
+      if (this.flushPromise === flushPromise) this.flushPromise = undefined;
+    });
+    this.flushPromise = flushPromise;
+    return flushPromise;
+  }
+
+  private async flushBufferedBatches(): Promise<boolean> {
+    while (this.buffer.length > 0) {
+      const batch = this.buffer.splice(0, positiveEnv('PROTOCOL_LOG_BATCH_SIZE', 100));
+      try {
+        await this.prisma.protocolInteractionLog.createMany({ data: batch });
+      } catch (error) {
+        this.buffer.unshift(...batch);
+        this.logger.error(
+          `Protocol log batch write failed (${batch.length} events)`,
+          error instanceof Error ? error.stack : String(error),
+        );
+        return false;
+      }
     }
-	return true;
+    return true;
   }
 
   private async purgeExpired() {