Fix CSV export connection and download lifecycle

packages/twenty-server/src/engine/core-modules/record-export/jobs/delete-record-export.job.ts

@@ -45,11 +45,12 @@ export class DeleteRecordExportJob {
           },
         },
       );
-      if (!isDefined(cleanupJobId))
+      if (!isDefined(cleanupJobId)) {
         throw new RecordExportException(
           'Export cleanup could not be queued',
           'QUEUE_UNAVAILABLE',
         );
+      }
       return;
     }
     await this.recordExportWorkspaceService.cancel({

packages/twenty-server/src/engine/core-modules/record-export/jobs/generate-record-export.job.ts

@@ -70,7 +70,9 @@ export class GenerateRecordExportJob {
       },
     });
 
-    if (!claimed) return;
+    if (!claimed) {
+      return;
+    }
 
     let processedRecordCount = 0;
     let stream: Readable | undefined;
@@ -80,7 +82,7 @@ export class GenerateRecordExportJob {
       const remainingTime =
         RECORD_EXPORT_MAX_DURATION_MS -
         (Date.now() - recordExport.createdAt.getTime());
-      if (remainingTime <= 0)
+      if (remainingTime <= 0) {
         throw new RecordExportException(
           'Export duration limit exceeded',
           'DURATION_LIMIT_EXCEEDED',
@@ -88,6 +90,7 @@ export class GenerateRecordExportJob {
             userFriendlyMessage: msg`The export took too long. Please try exporting fewer records.`,
           },
         );
+      }
 
       const requester =
         await this.recordExportQueryWorkspaceService.resolveRequester(
@@ -95,8 +98,7 @@ export class GenerateRecordExportJob {
         );
       locale = requester.workspaceMember.locale;
       const context = await this.recordExportQueryWorkspaceService.buildContext(
-        recordExport.parameters,
-        requester,
+        { parameters: recordExport.parameters, authContext: requester },
       );
       const recordExportQueryWorkspaceService =
         this.recordExportQueryWorkspaceService;
@@ -115,19 +117,20 @@ export class GenerateRecordExportJob {
           },
           changes: { processedRecordCount },
         });
-        if (!result)
+        if (!result) {
           throw new RecordExportException(
             'Export attempt was superseded',
             'ATTEMPT_SUPERSEDED',
           );
+        }
         lastProgressAt = Date.now();
       };
 
       const totalRecordCount =
-        await recordExportQueryWorkspaceService.countRecords(
-          recordExport.parameters,
+        await recordExportQueryWorkspaceService.countRecords({
+          parameters: recordExport.parameters,
           context,
-        );
+        });
       if (
         !(await recordExportCacheService.update({
           workspaceId,
@@ -135,11 +138,12 @@ export class GenerateRecordExportJob {
           condition: { attemptId, statuses: [RecordExportStatus.PROCESSING] },
           changes: { totalRecordCount },
         }))
-      )
+      ) {
         throw new RecordExportException(
           'Export attempt was superseded',
           'ATTEMPT_SUPERSEDED',
         );
+      }
       await updateProgress();
 
       async function* generateCsv() {
@@ -171,16 +175,16 @@ export class GenerateRecordExportJob {
               context.queryRunnerContext.authContext,
             );
           }
-          const { results } = await recordExportQueryWorkspaceService.readPage(
-            recordExport.parameters,
+          const { results } = await recordExportQueryWorkspaceService.readPage({
+            parameters: recordExport.parameters,
             context,
             after,
-          );
+          });
 
           for (const record of results.records) {
             const row = formatRecordExportRow(context.columns, record);
             bytes += Buffer.byteLength(row);
-            if (bytes > RECORD_EXPORT_MAX_FILE_BYTES)
+            if (bytes > RECORD_EXPORT_MAX_FILE_BYTES) {
               throw new RecordExportException(
                 'Export file size limit exceeded',
                 'FILE_SIZE_LIMIT_EXCEEDED',
@@ -188,20 +192,28 @@ export class GenerateRecordExportJob {
                   userFriendlyMessage: msg`The export file is too large. Please export fewer records.`,
                 },
               );
+            }
             yield row;
             processedRecordCount++;
           }
 
-          if (Date.now() - lastProgressAt >= RECORD_EXPORT_PROGRESS_INTERVAL_MS)
+          if (
+            Date.now() - lastProgressAt >=
+            RECORD_EXPORT_PROGRESS_INTERVAL_MS
+          ) {
             await updateProgress();
-          if (!results.pageInfo.hasNextPage) break;
+          }
+          if (!results.pageInfo.hasNextPage) {
+            break;
+          }
 
           const endCursor = results.pageInfo.endCursor;
-          if (!isDefined(endCursor) || endCursor === after)
+          if (!isDefined(endCursor) || endCursor === after) {
             throw new RecordExportException(
               'Export pagination did not advance',
               'PAGINATION_FAILED',
             );
+          }
           after = endCursor;
         } while (true);
       }
@@ -238,11 +250,12 @@ export class GenerateRecordExportJob {
         },
       });
 
-      if (!completed)
+      if (!completed) {
         throw new RecordExportException(
           'Export attempt was superseded',
           'ATTEMPT_SUPERSEDED',
         );
+      }
     } catch (error) {
       stream?.destroy();
       await this.fileStorageService

packages/twenty-server/src/engine/core-modules/record-export/record-export.controller.ts

@@ -1,5 +1,8 @@
+import { RecordExportCacheService } from 'src/engine/core-modules/record-export/services/record-export-cache.service';
+import { RecordExportStatus } from 'src/engine/core-modules/record-export/enums/record-export-status.enum';
 import {
   Controller,
+  ConflictException,
   ForbiddenException,
   Get,
   Param,
@@ -29,6 +32,7 @@ import { PermissionsRestApiExceptionFilter } from 'src/engine/metadata-modules/p
 @UseFilters(PermissionsRestApiExceptionFilter)
 export class RecordExportController {
   constructor(
+    private readonly recordExportCacheService: RecordExportCacheService,
     private readonly recordExportWorkspaceService: RecordExportWorkspaceService,
     private readonly recordExportQueryWorkspaceService: RecordExportQueryWorkspaceService,
     private readonly fileStorageService: FileStorageService,
@@ -48,10 +52,11 @@ export class RecordExportController {
         payload.type !== JwtTokenTypeEnum.FILE ||
         payload.fileId !== id ||
         !isDefined(payload.workspaceId)
-      )
+      ) {
         throw new ForbiddenException(
           t`Invalid or expired export download link.`,
         );
+      }
     } catch {
       throw new ForbiddenException(t`Invalid or expired export download link.`);
     }
@@ -65,16 +70,30 @@ export class RecordExportController {
       await this.recordExportQueryWorkspaceService.resolveRequester(
         recordExport,
       );
-    const context = await this.recordExportQueryWorkspaceService.buildContext(
-      recordExport.parameters,
-      requester,
-    );
-    await this.recordExportQueryWorkspaceService.readPage(
-      recordExport.parameters,
+    const context = await this.recordExportQueryWorkspaceService.buildContext({
+      parameters: recordExport.parameters,
+      authContext: requester,
+    });
+    await this.recordExportQueryWorkspaceService.readPage({
+      parameters: recordExport.parameters,
       context,
-      undefined,
-      0,
-    );
+      first: 0,
+    });
+
+    const claimed = await this.recordExportCacheService.update({
+      workspaceId: recordExport.workspaceId,
+      id: recordExport.id,
+      condition: {
+        statuses: [RecordExportStatus.COMPLETED],
+        downloadStarted: false,
+      },
+      changes: { downloadStarted: true },
+    });
+    if (!claimed) {
+      throw new ConflictException(
+        t`This export download has already started or expired.`,
+      );
+    }
 
     const resource = this.recordExportWorkspaceService.getFileResource({
       workspaceId: recordExport.workspaceId,
@@ -82,10 +101,10 @@ export class RecordExportController {
     });
     response.setHeader('Cache-Control', 'private, no-store');
     const contentDisposition = `attachment; filename="${recordExport.filename.replace(/["\r\n\\]/g, '_')}"`;
-    const stream = await this.fileStorageService.readFile(resource);
-    response.setHeader('Content-Type', 'text/csv; charset=utf-8');
-    response.setHeader('Content-Disposition', contentDisposition);
     try {
+      const stream = await this.fileStorageService.readFile(resource);
+      response.setHeader('Content-Type', 'text/csv; charset=utf-8');
+      response.setHeader('Content-Disposition', contentDisposition);
       await pipeline(stream, response);
     } finally {
       await this.recordExportWorkspaceService.cancel({

packages/twenty-server/src/engine/core-modules/record-export/record-export.module.ts

@@ -1,3 +1,4 @@
+import { RecordExportStreamWorkspaceService } from 'src/engine/core-modules/record-export/services/record-export-stream.workspace-service';
 import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
 import { RecordExportCacheService } from 'src/engine/core-modules/record-export/services/record-export-cache.service';
 import { DeleteRecordExportJob } from 'src/engine/core-modules/record-export/jobs/delete-record-export.job';
@@ -28,6 +29,7 @@ import { WorkspaceCacheModule } from 'src/engine/workspace-cache/workspace-cache
   controllers: [RecordExportController],
   providers: [
     RecordExportResolver,
+    RecordExportStreamWorkspaceService,
     RecordExportCacheService,
     DeleteRecordExportJob,
     RecordExportWorkspaceService,

packages/twenty-server/src/engine/core-modules/record-export/record-export.resolver.ts

@@ -1,3 +1,4 @@
+import { RecordExportStreamWorkspaceService } from 'src/engine/core-modules/record-export/services/record-export-stream.workspace-service';
 import { FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
 import { PreventNestToAutoLogGraphqlErrorsFilter } from 'src/engine/core-modules/graphql/filters/prevent-nest-to-auto-log-graphql-errors.filter';
 import { ForbiddenError } from 'src/engine/core-modules/graphql/utils/graphql-errors.util';
@@ -12,7 +13,6 @@ import { getWorkspaceAuthContext } from 'src/engine/core-modules/auth/storage/wo
 import { ResolverValidationPipe } from 'src/engine/core-modules/graphql/pipes/resolver-validation.pipe';
 import { CreateRecordExportInput } from 'src/engine/core-modules/record-export/dtos/create-record-export.input';
 import { RecordExportDTO } from 'src/engine/core-modules/record-export/dtos/record-export.dto';
-import { RecordExportWorkspaceService } from 'src/engine/core-modules/record-export/services/record-export.workspace-service';
 import { SettingsPermissionGuard } from 'src/engine/guards/settings-permission.guard';
 import { UserAuthGuard } from 'src/engine/guards/user-auth.guard';
 import { WorkspaceAuthGuard } from 'src/engine/guards/workspace-auth.guard';
@@ -27,7 +27,7 @@ import { WorkspaceAuthGuard } from 'src/engine/guards/workspace-auth.guard';
 @UseFilters(PreventNestToAutoLogGraphqlErrorsFilter)
 export class RecordExportResolver {
   constructor(
-    private readonly recordExportWorkspaceService: RecordExportWorkspaceService,
+    private readonly recordExportStreamService: RecordExportStreamWorkspaceService,
     private readonly featureFlagService: FeatureFlagService,
   ) {}
 
@@ -48,7 +48,7 @@ export class RecordExportResolver {
         'Asynchronous CSV export is not enabled for this workspace',
       );
     }
-    return this.recordExportWorkspaceService.stream({
+    return this.recordExportStreamService.stream({
       parameters: input,
       authContext,
     });

packages/twenty-server/src/engine/core-modules/record-export/services/record-export-cache.service.ts

@@ -24,6 +24,7 @@ type RecordExportChanges = Partial<
     | 'jobId'
     | 'attemptId'
     | 'filePath'
+    | 'downloadStarted'
     | 'errorMessage'
     | 'totalRecordCount'
   >
@@ -31,6 +32,7 @@ type RecordExportChanges = Partial<
 type RecordExportCondition = {
   statuses?: RecordExportStatus[];
   attemptId?: string | null;
+  downloadStarted?: boolean;
 };
 
 @Injectable()
@@ -58,6 +60,7 @@ export class RecordExportCacheService {
       jobId: null,
       attemptId: null,
       filePath: null,
+      downloadStarted: false,
       errorMessage: null,
       updatedAt: now,
     };
@@ -73,10 +76,11 @@ export class RecordExportCacheService {
       recordExport.id,
       RECORD_EXPORT_CONNECTION_TTL_MS,
     );
-    if (!created)
+    if (!created) {
       throw new ConflictException(
         t`An export is already running in this workspace. Please wait for it to finish.`,
       );
+    }
     try {
       await this.cacheStorageService.set(
         this.getRecordKey(recordExport),
@@ -111,8 +115,12 @@ export class RecordExportCacheService {
     const recordExport = await this.cacheStorageService.get<RecordExport>(
       this.getRecordKey({ workspaceId, id }),
     );
-    if (!isDefined(recordExport)) return undefined;
-    if (!connected && this.isRunning(recordExport)) return undefined;
+    if (!isDefined(recordExport)) {
+      return undefined;
+    }
+    if (!connected && this.isRunning(recordExport)) {
+      return undefined;
+    }
     return {
       ...recordExport,
       createdAt: new Date(recordExport.createdAt),
@@ -139,7 +147,9 @@ export class RecordExportCacheService {
         (isDefined(condition.statuses) &&
           !condition.statuses.includes(recordExport.status)) ||
         (condition.attemptId !== undefined &&
-          condition.attemptId !== recordExport.attemptId)
+          condition.attemptId !== recordExport.attemptId) ||
+        (isDefined(condition.downloadStarted) &&
+          condition.downloadStarted !== recordExport.downloadStarted)
       ) {
         return false;
       }
@@ -157,7 +167,9 @@ export class RecordExportCacheService {
         expiresAt,
       };
       const ttl = updated.expiresAt.getTime() - Date.now();
-      if (ttl <= 0) return false;
+      if (ttl <= 0) {
+        return false;
+      }
       const updatedRecord = await this.cacheStorageService.runScript<number>({
         script: UPDATE_RECORD_EXPORT_SCRIPT,
         keys: [
@@ -172,8 +184,12 @@ export class RecordExportCacheService {
           this.isRunning(updated) ? '0' : '1',
         ],
       });
-      if (updatedRecord === 0) return false;
-      if (updatedRecord === -1) continue;
+      if (updatedRecord === 0) {
+        return false;
+      }
+      if (updatedRecord === -1) {
+        continue;
+      }
       return true;
     }
   }

packages/twenty-server/src/engine/core-modules/record-export/services/record-export-query.workspace-service.ts

@@ -117,10 +117,13 @@ export class RecordExportQueryWorkspaceService {
     );
   }
 
-  async buildContext(
-    parameters: RecordExportParameters,
-    authContext: UserWorkspaceAuthContext,
-  ): Promise<RecordExportQueryContext> {
+  async buildContext({
+    parameters,
+    authContext,
+  }: {
+    parameters: RecordExportParameters;
+    authContext: UserWorkspaceAuthContext;
+  }): Promise<RecordExportQueryContext> {
     const { flatObjectMetadataMaps, flatFieldMetadataMaps } =
       await this.workspaceCacheService.getOrRecompute(
         authContext.workspace.id,
@@ -191,10 +194,13 @@ export class RecordExportQueryWorkspaceService {
     };
   }
 
-  async countRecords(
-    parameters: RecordExportParameters,
-    context: RecordExportQueryContext,
-  ): Promise<number> {
+  async countRecords({
+    parameters,
+    context,
+  }: {
+    parameters: Pick<RecordExportParameters, 'filter'>;
+    context: RecordExportQueryContext;
+  }): Promise<number> {
     const { results } = await withWorkspaceAuthContext(
       context.queryRunnerContext.authContext,
       () =>
@@ -207,20 +213,26 @@ export class RecordExportQueryWorkspaceService {
           context.queryRunnerContext,
         ),
     );
-    if (!isDefined(results.totalCount))
+    if (!isDefined(results.totalCount)) {
       throw new RecordExportException(
         'Export record count is unavailable',
         'RECORD_COUNT_UNAVAILABLE',
       );
+    }
     return Number(results.totalCount);
   }
 
-  async readPage(
-    parameters: RecordExportParameters,
-    context: RecordExportQueryContext,
-    after?: string,
+  async readPage({
+    parameters,
+    context,
+    after,
     first = RECORD_EXPORT_PAGE_SIZE,
-  ) {
+  }: {
+    parameters: Pick<RecordExportParameters, 'filter' | 'orderBy'>;
+    context: RecordExportQueryContext;
+    after?: string;
+    first?: number;
+  }) {
     return withWorkspaceAuthContext(
       context.queryRunnerContext.authContext,
       () =>

packages/twenty-server/src/engine/core-modules/record-export/services/record-export-stream.workspace-service.ts

@@ -0,0 +1,163 @@
+import { BadRequestException, Injectable } from '@nestjs/common';
+
+import { t } from '@lingui/core/macro';
+import { setTimeout } from 'node:timers/promises';
+import { isDefined } from 'twenty-shared/utils';
+
+import { type WorkspaceAuthContext } from 'src/engine/core-modules/auth/types/workspace-auth-context.type';
+import { RECORD_EXPORT_PROGRESS_INTERVAL_MS } from 'src/engine/core-modules/record-export/constants/record-export.constants';
+import { type RecordExportDTO } from 'src/engine/core-modules/record-export/dtos/record-export.dto';
+import { RecordExportStatus } from 'src/engine/core-modules/record-export/enums/record-export-status.enum';
+import { RecordExportCacheService } from 'src/engine/core-modules/record-export/services/record-export-cache.service';
+import { RecordExportWorkspaceService } from 'src/engine/core-modules/record-export/services/record-export.workspace-service';
+import { type RecordExportParameters } from 'src/engine/core-modules/record-export/types/record-export-parameters.type';
+import { type RecordExport } from 'src/engine/core-modules/record-export/types/record-export.type';
+
+@Injectable()
+export class RecordExportStreamWorkspaceService {
+  constructor(
+    private readonly recordExportCacheService: RecordExportCacheService,
+    private readonly recordExportWorkspaceService: RecordExportWorkspaceService,
+  ) {}
+
+  async stream({
+    parameters,
+    authContext,
+  }: {
+    parameters: RecordExportParameters;
+    authContext: WorkspaceAuthContext;
+  }): Promise<AsyncIterableIterator<RecordExportDTO>> {
+    const abortController = new AbortController();
+    const service = this;
+    const recordExport = await this.recordExportWorkspaceService.create({
+      parameters,
+      authContext,
+    });
+    let downloadReady = false;
+    let connectionError: unknown;
+    let cleanup: Promise<void> | undefined;
+    const close = () => {
+      abortController.abort();
+      if (!downloadReady) {
+        cleanup ??= service.recordExportWorkspaceService.cancel(recordExport);
+      }
+      return cleanup ?? Promise.resolve();
+    };
+    const keepAlive = async (created: RecordExport) => {
+      try {
+        while (!abortController.signal.aborted) {
+          await setTimeout(RECORD_EXPORT_PROGRESS_INTERVAL_MS, undefined, {
+            signal: abortController.signal,
+          });
+          const current = await service.recordExportCacheService.findOne({
+            workspaceId: created.workspaceId,
+            id: created.id,
+            keepAlive: true,
+          });
+          if (
+            !isDefined(current) ||
+            current.expiresAt.getTime() <= Date.now()
+          ) {
+            throw new BadRequestException(
+              t`The export was interrupted. Please try again.`,
+            );
+          }
+          if (
+            current.status === RecordExportStatus.COMPLETED ||
+            current.status === RecordExportStatus.FAILED
+          ) {
+            return;
+          }
+        }
+      } catch (error) {
+        if (!abortController.signal.aborted) {
+          connectionError = error;
+          await close();
+        }
+      }
+    };
+
+    const heartbeat = keepAlive(recordExport).catch((error: unknown) => {
+      connectionError = error;
+    });
+    try {
+      await this.recordExportWorkspaceService.enqueue(recordExport);
+      if (isDefined(connectionError)) {
+        throw connectionError;
+      }
+    } catch (error) {
+      await close();
+      await heartbeat;
+      throw error;
+    }
+
+    async function* events(): AsyncGenerator<RecordExportDTO> {
+      try {
+        while (!abortController.signal.aborted) {
+          const current =
+            await service.recordExportWorkspaceService.findOrThrow(
+              recordExport,
+            );
+          const updated =
+            await service.recordExportWorkspaceService.reconcile(current);
+          if (abortController.signal.aborted) {
+            break;
+          }
+          if (updated.status === RecordExportStatus.COMPLETED) {
+            const downloadUrl =
+              await service.recordExportWorkspaceService.getDownloadUrl({
+                id: updated.id,
+                authContext,
+              });
+            if (abortController.signal.aborted) {
+              break;
+            }
+            downloadReady = true;
+            yield { ...updated, downloadUrl };
+            return;
+          }
+          yield updated;
+          if (updated.status === RecordExportStatus.FAILED) {
+            return;
+          }
+          await setTimeout(RECORD_EXPORT_PROGRESS_INTERVAL_MS, undefined, {
+            signal: abortController.signal,
+          });
+        }
+      } catch (error) {
+        if (!abortController.signal.aborted) {
+          throw error;
+        }
+      } finally {
+        await close();
+        await heartbeat;
+      }
+      if (isDefined(connectionError)) {
+        throw connectionError;
+      }
+    }
+    const iterator = events();
+    return {
+      next: () => iterator.next(),
+      return: async () => {
+        try {
+          await close();
+        } finally {
+          await iterator.return(undefined);
+        }
+        return { done: true, value: undefined };
+      },
+      throw: async (error: unknown) => {
+        try {
+          await close();
+        } finally {
+          await iterator.return(undefined);
+        }
+        throw error;
+      },
+      [Symbol.asyncIterator]() {
+        return this;
+      },
+    };
+  }
+}

packages/twenty-server/src/engine/core-modules/record-export/services/record-export.workspace-service.ts

@@ -5,10 +5,7 @@ import {
   NotFoundException,
 } from '@nestjs/common';
 
-import { setTimeout } from 'node:timers/promises';
-
 import { FileStorageService } from 'src/engine/core-modules/file-storage/services/file-storage.service';
-import { RecordExportDTO } from 'src/engine/core-modules/record-export/dtos/record-export.dto';
 import { t } from '@lingui/core/macro';
 import { TWENTY_STANDARD_APPLICATION_UNIVERSAL_IDENTIFIER } from 'twenty-shared/application';
 import { FileFolder } from 'twenty-shared/types';
@@ -23,7 +20,6 @@ import { MessageQueueService } from 'src/engine/core-modules/message-queue/servi
 import {
   RECORD_EXPORT_MAX_DURATION_MS,
   RECORD_EXPORT_CONNECTION_TTL_MS,
-  RECORD_EXPORT_PROGRESS_INTERVAL_MS,
   RECORD_EXPORT_MISSING_JOB_TIMEOUT_MS,
 } from 'src/engine/core-modules/record-export/constants/record-export.constants';
 import { RecordExportStatus } from 'src/engine/core-modules/record-export/enums/record-export-status.enum';
@@ -56,10 +52,10 @@ export class RecordExportWorkspaceService {
   }): Promise<RecordExport> {
     const requester =
       await this.recordExportQueryWorkspaceService.assertCanExport(authContext);
-    const context = await this.recordExportQueryWorkspaceService.buildContext(
+    const context = await this.recordExportQueryWorkspaceService.buildContext({
       parameters,
-      requester,
-    );
+      authContext: requester,
+    });
     const workspaceId = requester.workspace.id;
 
     const recordExport = await this.recordExportCacheService.create({
@@ -70,6 +66,11 @@ export class RecordExportWorkspaceService {
       filename: `${context.queryRunnerContext.flatObjectMetadata.nameSingular}.csv`,
     });
 
+    return recordExport;
+  }
+
+  async enqueue(recordExport: RecordExport): Promise<void> {
+    const workspaceId = recordExport.workspaceId;
     try {
       const cleanupJobId = await this.cleanupQueueService.add(
         'DeleteRecordExportJob',
@@ -84,11 +85,12 @@ export class RecordExportWorkspaceService {
           },
         },
       );
-      if (!isDefined(cleanupJobId))
+      if (!isDefined(cleanupJobId)) {
         throw new RecordExportException(
           'Export cleanup could not be queued',
           'QUEUE_UNAVAILABLE',
         );
+      }
 
       const jobId = await this.messageQueueService.add(
         'GenerateRecordExportJob',
@@ -121,89 +123,6 @@ export class RecordExportWorkspaceService {
       });
       throw error;
     }
-
-    return this.findOrThrow({ workspaceId, id: recordExport.id });
-  }
-
-  async stream({
-    parameters,
-    authContext,
-  }: {
-    parameters: RecordExportParameters;
-    authContext: WorkspaceAuthContext;
-  }): Promise<AsyncIterableIterator<RecordExportDTO>> {
-    const recordExport = await this.create({ parameters, authContext });
-    const workspaceId = recordExport.workspaceId;
-    const abortController = new AbortController();
-    const service = this;
-    let downloadReady = false;
-    let cleanup: Promise<void> | undefined;
-    const close = () => {
-      abortController.abort();
-      if (!downloadReady)
-        cleanup ??= service.cancel({ workspaceId, id: recordExport.id });
-      return cleanup ?? Promise.resolve();
-    };
-
-    async function* events(): AsyncGenerator<RecordExportDTO> {
-      try {
-        while (!abortController.signal.aborted) {
-          const current = await service.recordExportCacheService.findOne({
-            workspaceId,
-            id: recordExport.id,
-            keepAlive: true,
-          });
-          if (!isDefined(current))
-            throw new BadRequestException(
-              t`The export was interrupted. Please try again.`,
-            );
-          const updated = await service.reconcile(current);
-          if (abortController.signal.aborted) return;
-          if (updated.status === RecordExportStatus.COMPLETED) {
-            const downloadUrl = await service.getDownloadUrl({
-              id: updated.id,
-              authContext,
-            });
-            if (abortController.signal.aborted) return;
-            downloadReady = true;
-            yield { ...updated, downloadUrl };
-            return;
-          }
-          yield updated;
-          if (updated.status === RecordExportStatus.FAILED) return;
-          await setTimeout(RECORD_EXPORT_PROGRESS_INTERVAL_MS, undefined, {
-            signal: abortController.signal,
-          });
-        }
-      } catch (error) {
-        if (!abortController.signal.aborted) throw error;
-      } finally {
-        await close();
-      }
-    }
-    const iterator = events();
-    return {
-      next: () => iterator.next(),
-      return: async () => {
-        try {
-          await close();
-        } finally {
-          await iterator.return(undefined);
-        }
-        return { done: true, value: undefined };
-      },
-      throw: async (error: unknown) => {
-        try {
-          await close();
-        } finally {
-          await iterator.return(undefined);
-        }
-        throw error;
-      },
-      [Symbol.asyncIterator]() {
-        return this;
-      },
-    };
   }
 
   async cancel({
@@ -303,16 +222,15 @@ export class RecordExportWorkspaceService {
     }
 
     this.assertDownloadable(recordExport);
-    const context = await this.recordExportQueryWorkspaceService.buildContext(
-      recordExport.parameters,
-      requester,
-    );
-    await this.recordExportQueryWorkspaceService.readPage(
-      recordExport.parameters,
+    const context = await this.recordExportQueryWorkspaceService.buildContext({
+      parameters: recordExport.parameters,
+      authContext: requester,
+    });
+    await this.recordExportQueryWorkspaceService.readPage({
+      parameters: recordExport.parameters,
       context,
-      undefined,
-      0,
-    );
+      first: 0,
+    });
     const token = await this.jwtWrapperService.signAsyncOrThrow(
       {
         type: JwtTokenTypeEnum.FILE,

packages/twenty-server/src/engine/core-modules/record-export/types/record-export.type.ts

@@ -14,6 +14,7 @@ export type RecordExport = {
   jobId: string | null;
   attemptId: string | null;
   filePath: string | null;
+  downloadStarted: boolean;
   errorMessage: string | null;
   createdAt: Date;
   updatedAt: Date;

packages/twenty-server/src/engine/core-modules/record-export/utils/build-record-export-columns.util.ts

@@ -1,3 +1,4 @@
+import { isCompositeFieldMetadataType } from 'src/engine/metadata-modules/field-metadata/utils/is-composite-field-metadata-type.util';
 import { COMPOSITE_FIELD_SUB_FIELD_LABELS } from 'twenty-shared/constants';
 import { FieldMetadataType, RelationType } from 'twenty-shared/types';
 
@@ -30,10 +31,9 @@ export const buildRecordExportColumns = (
           : [];
       }
 
-      const subFields =
-        COMPOSITE_FIELD_SUB_FIELD_LABELS[
-          field.type as keyof typeof COMPOSITE_FIELD_SUB_FIELD_LABELS
-        ];
+      const subFields = isCompositeFieldMetadataType(field.type)
+        ? COMPOSITE_FIELD_SUB_FIELD_LABELS[field.type]
+        : undefined;
 
       return subFields
         ? Object.entries(subFields).map(([subFieldName, label]) => ({

packages/twenty-server/test/integration/record-export/record-export-cache.integration-spec.ts

@@ -195,7 +195,9 @@ describe('record export Redis lifetime', () => {
     let updates = 0;
     jest.spyOn(cache, 'runScript').mockImplementation(async (options) => {
       if (options.script.name === 'record-export:update' && ++updates <= 2) {
-        if (updates === 2) releaseUpdates();
+        if (updates === 2) {
+          releaseUpdates();
+        }
         await updatesReady;
       }
       return runScript(options);

packages/twenty-server/test/integration/record-export/record-export.integration-spec.ts

@@ -1,3 +1,6 @@
+import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum';
+import { type CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service';
+import { type RecordExportStreamWorkspaceService } from 'src/engine/core-modules/record-export/services/record-export-stream.workspace-service';
 import { updateFeatureFlag } from 'test/integration/metadata/suites/utils/update-feature-flag.util';
 import { type FeatureFlagService } from 'src/engine/core-modules/feature-flag/services/feature-flag.service';
 import { createClient } from 'graphql-sse';
@@ -44,7 +47,9 @@ const companies = Array.from({ length: 1005 }, (_, index) => ({
 const waitUntil = async (condition: () => Promise<boolean>) => {
   const deadline = Date.now() + 10_000;
   while (!(await condition())) {
-    if (Date.now() > deadline) throw new Error('Export did not settle');
+    if (Date.now() > deadline) {
+      throw new Error('Export did not settle');
+    }
     await setTimeout(25);
   }
 };
@@ -85,8 +90,9 @@ describe('record export lifecycle (integration)', () => {
 
   const nextExport = async (events: ReturnType<typeof subscribe>['events']) => {
     const event = await events.next();
-    if (event.done || !isDefined(event.value.data?.exportRecords))
+    if (event.done || !isDefined(event.value.data?.exportRecords)) {
       throw new Error(JSON.stringify(event.value?.errors ?? 'Export ended'));
+    }
     const recordExport = event.value.data.exportRecords;
     exportIds.add(recordExport.id);
     return recordExport;
@@ -100,8 +106,9 @@ describe('record export lifecycle (integration)', () => {
     try {
       while (true) {
         const recordExport = await nextExport(events);
-        if (recordExport.status === RecordExportStatus.FAILED)
+        if (recordExport.status === RecordExportStatus.FAILED) {
           throw new Error(recordExport.errorMessage ?? 'Export failed');
+        }
         if (recordExport.status === RecordExportStatus.COMPLETED) {
           expect(recordExport.downloadUrl).toBeDefined();
           return recordExport;
@@ -201,11 +208,12 @@ describe('record export lifecycle (integration)', () => {
     for (const id of exportIds)
       await exports.cancel({ workspaceId: SEED_APPLE_WORKSPACE_ID, id });
     exportIds.clear();
-    if (isDefined(roleId))
+    if (isDefined(roleId)) {
       await changeRole({
         canAccessAllTools: true,
         canReadAllObjectRecords: true,
       });
+    }
   });
 
   afterAll(async () => {
@@ -214,15 +222,17 @@ describe('record export lifecycle (integration)', () => {
       value: wasAsyncCsvExportEnabled,
       expectToFail: false,
     });
-    if (isDefined(originalRoleId))
+    if (isDefined(originalRoleId)) {
       await updateWorkspaceMemberRole({
         input: {
           roleId: originalRoleId,
           workspaceMemberId: WORKSPACE_MEMBER_DATA_SEED_IDS.JONY,
         },
       });
-    if (isDefined(roleId))
+    }
+    if (isDefined(roleId)) {
       await deleteOneRole({ input: { idToDelete: roleId } });
+    }
     for (let offset = 0; offset < companies.length; offset += 100) {
       await makeGraphqlAPIRequest(
         destroyManyOperationFactory({
@@ -330,14 +340,17 @@ describe('record export lifecycle (integration)', () => {
       releasePage = resolve;
     });
     jest.spyOn(query, 'readPage').mockImplementation(async (...args) => {
-      if (isDefined(args[2])) await pageGate;
+      if (isDefined(args[0].after)) {
+        await pageGate;
+      }
       return readPage(...args);
     });
     const first = subscribe();
     try {
       let progress = await nextExport(first.events);
-      while (progress.processedRecordCount < 1000)
+      while (progress.processedRecordCount < 1000) {
         progress = await nextExport(first.events);
+      }
       expect(progress.status).toBe(RecordExportStatus.PROCESSING);
       expect(progress.totalRecordCount).toBe(companies.length);
       const second = subscribe();
@@ -369,7 +382,9 @@ describe('record export lifecycle (integration)', () => {
         .spyOn(queue, 'add')
         .mockRejectedValueOnce(new Error('Queue unavailable'));
       await expect(
-        exports.create({ parameters: input, authContext: requester }),
+        exports.enqueue(
+          await exports.create({ parameters: input, authContext: requester }),
+        ),
       ).rejects.toThrow('Queue unavailable');
       expect((await getExport(failed!.id)).status).toBe(
         RecordExportStatus.FAILED,
@@ -379,6 +394,117 @@ describe('record export lifecycle (integration)', () => {
     },
   );
 
+  it('renews the lease during queue handoff and paused event consumption', async () => {
+    const ready = await exportToCompletion();
+    const requester = await query.resolveRequester(await getExport(ready.id));
+    const streams =
+      getAppProviderByClassName<RecordExportStreamWorkspaceService>(
+        'RecordExportStreamWorkspaceService',
+      );
+    const storageCache = global.app.get<CacheStorageService>(
+      CacheStorageNamespace.EngineRecordExport,
+    );
+    const enqueue = exports.enqueue.bind(exports);
+    let releaseEnqueue = () => {};
+    let notifyEnqueue = (recordExport: RecordExport) => {
+      void recordExport;
+    };
+    const enqueueGate = new Promise<void>((resolve) => {
+      releaseEnqueue = resolve;
+    });
+    const enqueueStarted = new Promise<RecordExport>((resolve) => {
+      notifyEnqueue = resolve;
+    });
+    jest.spyOn(exports, 'enqueue').mockImplementation(async (recordExport) => {
+      exportIds.add(recordExport.id);
+      notifyEnqueue(recordExport);
+      await enqueueGate;
+      await enqueue(recordExport);
+    });
+    const readPage = query.readPage.bind(query);
+    let releasePage = () => {};
+    const pageGate = new Promise<void>((resolve) => {
+      releasePage = resolve;
+    });
+    jest.spyOn(query, 'readPage').mockImplementation(async (args) => {
+      await pageGate;
+      return readPage(args);
+    });
+    const subscription = streams.stream({
+      parameters: input,
+      authContext: requester,
+    });
+    try {
+      const recordExport = await enqueueStarted;
+      const assertLeaseSurvives = async () => {
+        await storageCache.runScript({
+          script: {
+            name: 'shorten-export-test-lease',
+            source: "return redis.call('PEXPIRE', KEYS[1], 2000)",
+          },
+          keys: [`{${SEED_APPLE_WORKSPACE_ID}}:active`],
+          args: [],
+        });
+        await setTimeout(2500);
+        expect(await cache.findOne(recordExport)).toBeDefined();
+      };
+      await assertLeaseSurvives();
+      releaseEnqueue();
+      const events = await subscription;
+      expect((await events.next()).done).toBe(false);
+      await assertLeaseSurvives();
+      await events.return?.();
+      expect(await cache.findOne(recordExport)).toBeUndefined();
+    } finally {
+      releaseEnqueue();
+      releasePage();
+      const events = await subscription;
+      await events.return?.();
+    }
+  });
+
+  it('allows only one download while the file is being opened', async () => {
+    const recordExport = await exportToCompletion();
+    const stored = await getExport(recordExport.id);
+    const readFile = storage.readFile.bind(storage);
+    let releaseRead = () => {};
+    let notifyRead = () => {};
+    const readGate = new Promise<void>((resolve) => {
+      releaseRead = resolve;
+    });
+    const readStarted = new Promise<void>((resolve) => {
+      notifyRead = resolve;
+    });
+    jest.spyOn(storage, 'readFile').mockImplementationOnce(async (resource) => {
+      notifyRead();
+      await readGate;
+      return readFile(resource);
+    });
+    const first = download(recordExport)
+      .expect(200)
+      .then((response) => response);
+    try {
+      await readStarted;
+      await download(recordExport).expect(409);
+      expect(await fileExists(stored)).toBe(true);
+    } finally {
+      releaseRead();
+      expect((await first).text).toContain(companies[0].name);
+    }
+    await waitUntil(async () => !(await fileExists(stored)));
+  });
+
+  it('cleans up when storage cannot open the download', async () => {
+    const recordExport = await exportToCompletion();
+    const stored = await getExport(recordExport.id);
+    jest
+      .spyOn(storage, 'readFile')
+      .mockRejectedValueOnce(new Error('Storage unavailable'));
+    await download(recordExport).expect(500);
+    expect(await cache.findOne(stored)).toBeUndefined();
+    expect(await fileExists(stored)).toBe(false);
+  });
+
   it('rejects a download token used for another export and an invalid signature', async () => {
     const recordExport = await exportToCompletion();
     const url = new URL(recordExport.downloadUrl!);
@@ -450,7 +576,7 @@ describe('record export lifecycle (integration)', () => {
       notifyPage = resolve;
     });
     jest.spyOn(query, 'readPage').mockImplementation(async (...args) => {
-      if (!isDefined(args[2]) && !isDefined(args[3])) {
+      if (!isDefined(args[0].after) && !isDefined(args[0].first)) {
         notifyPage();
         await pageGate;
       }
@@ -462,8 +588,9 @@ describe('record export lifecycle (integration)', () => {
       await pageStarted;
       await changeRole({ canAccessAllTools: false });
       releasePage();
-      while (recordExport.status !== RecordExportStatus.FAILED)
+      while (recordExport.status !== RecordExportStatus.FAILED) {
         recordExport = await nextExport(events);
+      }
       expect(recordExport.downloadUrl).toBeNull();
       expect(recordExport.processedRecordCount).toBeLessThan(companies.length);
     } finally {
@@ -501,13 +628,16 @@ describe('record export lifecycle (integration)', () => {
       });
     const readPage = query.readPage.bind(query);
     jest.spyOn(query, 'readPage').mockImplementation(async (...args) => {
-      if (isDefined(args[2])) throw new Error('Interrupted database read');
+      if (isDefined(args[0].after)) {
+        throw new Error('Interrupted database read');
+      }
       return readPage(...args);
     });
     const { events } = subscribe();
     let recordExport = await nextExport(events);
-    while (recordExport.status !== RecordExportStatus.FAILED)
+    while (recordExport.status !== RecordExportStatus.FAILED) {
       recordExport = await nextExport(events);
+    }
     expect(recordExport.downloadUrl).toBeNull();
     expect(partialFilePath).toBeDefined();
     expect(