diff --git a/packages/twenty-server/src/engine/core-modules/record-export/jobs/delete-record-export.job.ts b/packages/twenty-server/src/engine/core-modules/record-export/jobs/delete-record-export.job.ts index 12ee99986ee..f12735d6eb4 100644 --- a/packages/twenty-server/src/engine/core-modules/record-export/jobs/delete-record-export.job.ts +++ b/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({ diff --git a/packages/twenty-server/src/engine/core-modules/record-export/jobs/generate-record-export.job.ts b/packages/twenty-server/src/engine/core-modules/record-export/jobs/generate-record-export.job.ts index 481148bb4f5..76cf56bb032 100644 --- a/packages/twenty-server/src/engine/core-modules/record-export/jobs/generate-record-export.job.ts +++ b/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 diff --git a/packages/twenty-server/src/engine/core-modules/record-export/record-export.controller.ts b/packages/twenty-server/src/engine/core-modules/record-export/record-export.controller.ts index a70003362fb..d301c268758 100644 --- a/packages/twenty-server/src/engine/core-modules/record-export/record-export.controller.ts +++ b/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({ diff --git a/packages/twenty-server/src/engine/core-modules/record-export/record-export.module.ts b/packages/twenty-server/src/engine/core-modules/record-export/record-export.module.ts index b743447fca3..d5f6f74c1c0 100644 --- a/packages/twenty-server/src/engine/core-modules/record-export/record-export.module.ts +++ b/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, diff --git a/packages/twenty-server/src/engine/core-modules/record-export/record-export.resolver.ts b/packages/twenty-server/src/engine/core-modules/record-export/record-export.resolver.ts index 72e82ce60c4..ea5a3f4956d 100644 --- a/packages/twenty-server/src/engine/core-modules/record-export/record-export.resolver.ts +++ b/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, }); diff --git a/packages/twenty-server/src/engine/core-modules/record-export/services/record-export-cache.service.ts b/packages/twenty-server/src/engine/core-modules/record-export/services/record-export-cache.service.ts index b3956e41958..d4ed436d77d 100644 --- a/packages/twenty-server/src/engine/core-modules/record-export/services/record-export-cache.service.ts +++ b/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( 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({ 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; } } diff --git a/packages/twenty-server/src/engine/core-modules/record-export/services/record-export-query.workspace-service.ts b/packages/twenty-server/src/engine/core-modules/record-export/services/record-export-query.workspace-service.ts index bc0365b0b9e..c6b21879979 100644 --- a/packages/twenty-server/src/engine/core-modules/record-export/services/record-export-query.workspace-service.ts +++ b/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 { + async buildContext({ + parameters, + authContext, + }: { + parameters: RecordExportParameters; + authContext: UserWorkspaceAuthContext; + }): Promise { const { flatObjectMetadataMaps, flatFieldMetadataMaps } = await this.workspaceCacheService.getOrRecompute( authContext.workspace.id, @@ -191,10 +194,13 @@ export class RecordExportQueryWorkspaceService { }; } - async countRecords( - parameters: RecordExportParameters, - context: RecordExportQueryContext, - ): Promise { + async countRecords({ + parameters, + context, + }: { + parameters: Pick; + context: RecordExportQueryContext; + }): Promise { 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; + context: RecordExportQueryContext; + after?: string; + first?: number; + }) { return withWorkspaceAuthContext( context.queryRunnerContext.authContext, () => diff --git a/packages/twenty-server/src/engine/core-modules/record-export/services/record-export-stream.workspace-service.ts b/packages/twenty-server/src/engine/core-modules/record-export/services/record-export-stream.workspace-service.ts new file mode 100644 index 00000000000..45920e968ec --- /dev/null +++ b/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> { + const abortController = new AbortController(); + const service = this; + const recordExport = await this.recordExportWorkspaceService.create({ + parameters, + authContext, + }); + let downloadReady = false; + let connectionError: unknown; + let cleanup: Promise | 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 { + 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; + }, + }; + } +} diff --git a/packages/twenty-server/src/engine/core-modules/record-export/services/record-export.workspace-service.ts b/packages/twenty-server/src/engine/core-modules/record-export/services/record-export.workspace-service.ts index 22a78e7139f..7f9f39042d0 100644 --- a/packages/twenty-server/src/engine/core-modules/record-export/services/record-export.workspace-service.ts +++ b/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 { 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 { + 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> { - const recordExport = await this.create({ parameters, authContext }); - const workspaceId = recordExport.workspaceId; - const abortController = new AbortController(); - const service = this; - let downloadReady = false; - let cleanup: Promise | undefined; - const close = () => { - abortController.abort(); - if (!downloadReady) - cleanup ??= service.cancel({ workspaceId, id: recordExport.id }); - return cleanup ?? Promise.resolve(); - }; - - async function* events(): AsyncGenerator { - 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, diff --git a/packages/twenty-server/src/engine/core-modules/record-export/types/record-export.type.ts b/packages/twenty-server/src/engine/core-modules/record-export/types/record-export.type.ts index f2dd64b5171..bdf8e860870 100644 --- a/packages/twenty-server/src/engine/core-modules/record-export/types/record-export.type.ts +++ b/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; diff --git a/packages/twenty-server/src/engine/core-modules/record-export/utils/build-record-export-columns.util.ts b/packages/twenty-server/src/engine/core-modules/record-export/utils/build-record-export-columns.util.ts index 1a74dcdff43..41f272cc322 100644 --- a/packages/twenty-server/src/engine/core-modules/record-export/utils/build-record-export-columns.util.ts +++ b/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]) => ({ diff --git a/packages/twenty-server/test/integration/record-export/record-export-cache.integration-spec.ts b/packages/twenty-server/test/integration/record-export/record-export-cache.integration-spec.ts index 524c30a8c2b..b149a18234d 100644 --- a/packages/twenty-server/test/integration/record-export/record-export-cache.integration-spec.ts +++ b/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); diff --git a/packages/twenty-server/test/integration/record-export/record-export.integration-spec.ts b/packages/twenty-server/test/integration/record-export/record-export.integration-spec.ts index d97e1fe2e97..0de54da5f7b 100644 --- a/packages/twenty-server/test/integration/record-export/record-export.integration-spec.ts +++ b/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) => { 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['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', + ); + const storageCache = global.app.get( + CacheStorageNamespace.EngineRecordExport, + ); + const enqueue = exports.enqueue.bind(exports); + let releaseEnqueue = () => {}; + let notifyEnqueue = (recordExport: RecordExport) => { + void recordExport; + }; + const enqueueGate = new Promise((resolve) => { + releaseEnqueue = resolve; + }); + const enqueueStarted = new Promise((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((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((resolve) => { + releaseRead = resolve; + }); + const readStarted = new Promise((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(