mirror of https://github.com/nocodb/nocodb
Mert E.
3 months ago
committed by
GitHub
22 changed files with 358 additions and 174 deletions
@ -0,0 +1,79 @@
|
||||
import { Injectable } from '@nestjs/common'; |
||||
import { DuplicateProcessor } from '~/modules/jobs/jobs/export-import/duplicate.processor'; |
||||
import { AtImportProcessor } from '~/modules/jobs/jobs/at-import/at-import.processor'; |
||||
import { MetaSyncProcessor } from '~/modules/jobs/jobs/meta-sync/meta-sync.processor'; |
||||
import { SourceCreateProcessor } from '~/modules/jobs/jobs/source-create/source-create.processor'; |
||||
import { SourceDeleteProcessor } from '~/modules/jobs/jobs/source-delete/source-delete.processor'; |
||||
import { WebhookHandlerProcessor } from '~/modules/jobs/jobs/webhook-handler/webhook-handler.processor'; |
||||
import { DataExportProcessor } from '~/modules/jobs/jobs/data-export/data-export.processor'; |
||||
import { ThumbnailGeneratorProcessor } from '~/modules/jobs/jobs/thumbnail-generator/thumbnail-generator.processor'; |
||||
import { AttachmentCleanUpProcessor } from '~/modules/jobs/jobs/attachment-clean-up/attachment-clean-up'; |
||||
import { InitMigrationJobs } from '~/modules/jobs/migration-jobs/init-migration-jobs'; |
||||
import { JobTypes } from '~/interface/Jobs'; |
||||
|
||||
@Injectable() |
||||
export class JobsMap { |
||||
constructor( |
||||
protected readonly duplicateProcessor: DuplicateProcessor, |
||||
protected readonly atImportProcessor: AtImportProcessor, |
||||
protected readonly metaSyncProcessor: MetaSyncProcessor, |
||||
protected readonly sourceCreateProcessor: SourceCreateProcessor, |
||||
protected readonly sourceDeleteProcessor: SourceDeleteProcessor, |
||||
protected readonly webhookHandlerProcessor: WebhookHandlerProcessor, |
||||
protected readonly dataExportProcessor: DataExportProcessor, |
||||
protected readonly thumbnailGeneratorProcessor: ThumbnailGeneratorProcessor, |
||||
protected readonly attachmentCleanUpProcessor: AttachmentCleanUpProcessor, |
||||
protected readonly initMigrationJobs: InitMigrationJobs, |
||||
) {} |
||||
|
||||
protected _jobMap: { |
||||
[key in JobTypes]?: { |
||||
this: any; |
||||
fn?: string; |
||||
}; |
||||
} = { |
||||
[JobTypes.DuplicateBase]: { |
||||
this: this.duplicateProcessor, |
||||
fn: 'duplicateBase', |
||||
}, |
||||
[JobTypes.DuplicateModel]: { |
||||
this: this.duplicateProcessor, |
||||
fn: 'duplicateModel', |
||||
}, |
||||
[JobTypes.DuplicateColumn]: { |
||||
this: this.duplicateProcessor, |
||||
fn: 'duplicateColumn', |
||||
}, |
||||
[JobTypes.AtImport]: { |
||||
this: this.atImportProcessor, |
||||
}, |
||||
[JobTypes.MetaSync]: { |
||||
this: this.metaSyncProcessor, |
||||
}, |
||||
[JobTypes.SourceCreate]: { |
||||
this: this.sourceCreateProcessor, |
||||
}, |
||||
[JobTypes.SourceDelete]: { |
||||
this: this.sourceDeleteProcessor, |
||||
}, |
||||
[JobTypes.HandleWebhook]: { |
||||
this: this.webhookHandlerProcessor, |
||||
}, |
||||
[JobTypes.DataExport]: { |
||||
this: this.dataExportProcessor, |
||||
}, |
||||
[JobTypes.ThumbnailGenerator]: { |
||||
this: this.thumbnailGeneratorProcessor, |
||||
}, |
||||
[JobTypes.AttachmentCleanUp]: { |
||||
this: this.attachmentCleanUpProcessor, |
||||
}, |
||||
[JobTypes.InitMigrationJobs]: { |
||||
this: this.initMigrationJobs, |
||||
}, |
||||
}; |
||||
|
||||
public get jobs() { |
||||
return this._jobMap; |
||||
} |
||||
} |
||||
@ -0,0 +1,82 @@
|
||||
import { Process, Processor } from '@nestjs/bull'; |
||||
import { Inject, Logger } from '@nestjs/common'; |
||||
import { Job } from 'bull'; |
||||
import type { JobData } from '~/interface/Jobs'; |
||||
import { JOB_REQUEUE_LIMIT, JOBS_QUEUE, JobVersions } from '~/interface/Jobs'; |
||||
import { IJobsService } from '~/modules/jobs/jobs-service.interface'; |
||||
import { JobsMap } from '~/modules/jobs/jobs-map.service'; |
||||
import { JobsEventService } from '~/modules/jobs/jobs-event.service'; |
||||
import { JobStatus } from '~/interface/Jobs'; |
||||
|
||||
const NC_WORKER_CONCURRENCY = process.env.NC_WORKER_CONCURRENCY ?? 5; |
||||
|
||||
@Processor(JOBS_QUEUE) |
||||
export class JobsProcessor { |
||||
private logger = new Logger(JobsProcessor.name); |
||||
|
||||
constructor( |
||||
@Inject('JobsService') protected readonly jobsService: IJobsService, |
||||
protected readonly jobsEventService: JobsEventService, |
||||
protected readonly jobsMap: JobsMap, |
||||
) {} |
||||
|
||||
@Process({ |
||||
concurrency: +NC_WORKER_CONCURRENCY, |
||||
}) |
||||
async process(job: Job<JobData>) { |
||||
const { jobName } = job.data; |
||||
|
||||
if (!this.jobsMap.jobs[jobName]) { |
||||
this.logger.error(`Job not found for ${jobName}`); |
||||
await this.requeue(job); |
||||
return; |
||||
} |
||||
|
||||
const { this: processor, fn = 'job' } = this.jobsMap.jobs[jobName]; |
||||
|
||||
if (!processor[fn]) { |
||||
this.logger.error(`Job function not found for ${jobName}`); |
||||
await this.requeue(job); |
||||
return; |
||||
} |
||||
|
||||
if (JobVersions[jobName] || job.data?._jobVersion) { |
||||
if (JobVersions[jobName] !== job.data._jobVersion) { |
||||
this.logger.error(`Job version mismatch for ${jobName}`); |
||||
await this.requeue(job); |
||||
return; |
||||
} |
||||
} |
||||
|
||||
try { |
||||
return await processor[fn](job); |
||||
} catch (e) { |
||||
this.logger.error(`Error processing job ${jobName}`, e); |
||||
throw e; |
||||
} |
||||
} |
||||
|
||||
async requeue(job: Job<JobData>) { |
||||
// Remove the job from the queue otherwise ids will clash
|
||||
await job.releaseLock(); |
||||
await job.remove(); |
||||
|
||||
await this.jobsEventService.onCompleted(job, JobStatus.REQUEUED); |
||||
|
||||
const _jobDelay = job.data?._jobDelay ?? 0; |
||||
const _jobAttempt = job.data?._jobAttempt ?? 1; |
||||
|
||||
if (_jobAttempt > JOB_REQUEUE_LIMIT) { |
||||
this.logger.error(`Job ${job.data.jobName} failed after 10 attempts`); |
||||
return; |
||||
} |
||||
|
||||
job.data._jobDelay = _jobDelay + 5000; |
||||
job.data._jobAttempt = _jobAttempt + 1; |
||||
|
||||
return this.jobsService.add(job.data.jobName, job.data, { |
||||
jobId: job.id.toString(), |
||||
delay: job.data._jobDelay, |
||||
}); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue