mirror of https://github.com/nocodb/nocodb
Browse Source
* separated fallback and redis solutions * revised folder structure for jobs Signed-off-by: mertmit <mertmit99@gmail.com>pull/5711/head
mertmit
2 years ago
23 changed files with 365 additions and 213 deletions
@ -1,12 +1,12 @@ |
|||||||
import { Injectable } from '@nestjs/common'; |
import { Injectable } from '@nestjs/common'; |
||||||
import PQueue from 'p-queue'; |
import PQueue from 'p-queue'; |
||||||
import Emittery from 'emittery'; |
import Emittery from 'emittery'; |
||||||
import { JobStatus, JobTypes } from '../../interface/Jobs'; |
import { JobStatus, JobTypes } from '../../../interface/Jobs'; |
||||||
import { DuplicateProcessor } from './export-import/duplicate.processor'; |
import { DuplicateProcessor } from '../jobs/export-import/duplicate.processor'; |
||||||
|
import { AtImportProcessor } from '../jobs/at-import/at-import.processor'; |
||||||
import { JobsEventService } from './jobs-event.service'; |
import { JobsEventService } from './jobs-event.service'; |
||||||
import { AtImportProcessor } from './at-import/at-import.processor'; |
|
||||||
|
|
||||||
interface Job { |
export interface Job { |
||||||
id: string; |
id: string; |
||||||
name: string; |
name: string; |
||||||
status: string; |
status: string; |
||||||
@ -1,15 +1,21 @@ |
|||||||
import { Controller, HttpCode, Post, Request, UseGuards } from '@nestjs/common'; |
import { |
||||||
import { GlobalGuard } from '../../../guards/global/global.guard'; |
Controller, |
||||||
import { ExtractProjectIdMiddleware } from '../../../middlewares/extract-project-id/extract-project-id.middleware'; |
HttpCode, |
||||||
import { SyncSource } from '../../../models'; |
Inject, |
||||||
import { NcError } from '../../../helpers/catchError'; |
Post, |
||||||
import { JobsService } from '../jobs.service'; |
Request, |
||||||
import { JobTypes } from '../../../interface/Jobs'; |
UseGuards, |
||||||
|
} from '@nestjs/common'; |
||||||
|
import { GlobalGuard } from '../../../../guards/global/global.guard'; |
||||||
|
import { ExtractProjectIdMiddleware } from '../../../../middlewares/extract-project-id/extract-project-id.middleware'; |
||||||
|
import { SyncSource } from '../../../../models'; |
||||||
|
import { NcError } from '../../../../helpers/catchError'; |
||||||
|
import { JobTypes } from '../../../../interface/Jobs'; |
||||||
|
|
||||||
@Controller() |
@Controller() |
||||||
@UseGuards(ExtractProjectIdMiddleware, GlobalGuard) |
@UseGuards(ExtractProjectIdMiddleware, GlobalGuard) |
||||||
export class AtImportController { |
export class AtImportController { |
||||||
constructor(private readonly jobsService: JobsService) {} |
constructor(@Inject('JobsService') private readonly jobsService) {} |
||||||
|
|
||||||
@Post('/api/v1/db/meta/import/airtable') |
@Post('/api/v1/db/meta/import/airtable') |
||||||
@HttpCode(200) |
@HttpCode(200) |
||||||
@ -1,8 +1,8 @@ |
|||||||
/* eslint-disable no-async-promise-executor */ |
/* eslint-disable no-async-promise-executor */ |
||||||
import { RelationTypes, UITypes } from 'nocodb-sdk'; |
import { RelationTypes, UITypes } from 'nocodb-sdk'; |
||||||
import EntityMap from './EntityMap'; |
import EntityMap from './EntityMap'; |
||||||
import type { BulkDataAliasService } from '../../../../services/bulk-data-alias.service'; |
import type { BulkDataAliasService } from '../../../../../services/bulk-data-alias.service'; |
||||||
import type { TablesService } from '../../../../services/tables.service'; |
import type { TablesService } from '../../../../../services/tables.service'; |
||||||
// @ts-ignore
|
// @ts-ignore
|
||||||
import type { AirtableBase } from 'airtable/lib/airtable_base'; |
import type { AirtableBase } from 'airtable/lib/airtable_base'; |
||||||
import type { TableType } from 'nocodb-sdk'; |
import type { TableType } from 'nocodb-sdk'; |
||||||
@ -0,0 +1,16 @@ |
|||||||
|
import { Injectable } from '@nestjs/common'; |
||||||
|
import { EventEmitter2 } from '@nestjs/event-emitter'; |
||||||
|
import { JobEvents } from '../../../interface/Jobs'; |
||||||
|
import type { Job } from 'bull'; |
||||||
|
|
||||||
|
@Injectable() |
||||||
|
export class JobsLogService { |
||||||
|
constructor(private eventEmitter: EventEmitter2) {} |
||||||
|
|
||||||
|
sendLog(job: Job, data: { message: string }) { |
||||||
|
this.eventEmitter.emit(JobEvents.LOG, { |
||||||
|
id: job.id.toString(), |
||||||
|
data, |
||||||
|
}); |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,71 @@ |
|||||||
|
import { |
||||||
|
OnQueueActive, |
||||||
|
OnQueueCompleted, |
||||||
|
OnQueueFailed, |
||||||
|
Processor, |
||||||
|
} from '@nestjs/bull'; |
||||||
|
import { Job } from 'bull'; |
||||||
|
import boxen from 'boxen'; |
||||||
|
import { OnEvent } from '@nestjs/event-emitter'; |
||||||
|
import { JobEvents, JOBS_QUEUE, JobStatus } from '../../../interface/Jobs'; |
||||||
|
import { JobsRedisService } from './jobs-redis.service'; |
||||||
|
|
||||||
|
@Processor(JOBS_QUEUE) |
||||||
|
export class JobsEventService { |
||||||
|
constructor(private jobsRedisService: JobsRedisService) {} |
||||||
|
|
||||||
|
@OnQueueActive() |
||||||
|
onActive(job: Job) { |
||||||
|
this.jobsRedisService.publish(`jobs-${job.id.toString()}`, { |
||||||
|
cmd: JobEvents.STATUS, |
||||||
|
id: job.id.toString(), |
||||||
|
status: JobStatus.ACTIVE, |
||||||
|
}); |
||||||
|
} |
||||||
|
|
||||||
|
@OnQueueFailed() |
||||||
|
onFailed(job: Job, error: Error) { |
||||||
|
console.error( |
||||||
|
boxen( |
||||||
|
`---- !! JOB FAILED !! ----\nid:${job.id}\nerror:${error.name} (${error.message})\n\nstack: ${error.stack}`, |
||||||
|
{ |
||||||
|
padding: 1, |
||||||
|
borderStyle: 'double', |
||||||
|
borderColor: 'yellow', |
||||||
|
}, |
||||||
|
), |
||||||
|
); |
||||||
|
|
||||||
|
this.jobsRedisService.publish(`jobs-${job.id.toString()}`, { |
||||||
|
cmd: JobEvents.STATUS, |
||||||
|
id: job.id.toString(), |
||||||
|
status: JobStatus.FAILED, |
||||||
|
data: { |
||||||
|
error: { |
||||||
|
message: error?.message, |
||||||
|
}, |
||||||
|
}, |
||||||
|
}); |
||||||
|
} |
||||||
|
|
||||||
|
@OnQueueCompleted() |
||||||
|
onCompleted(job: Job, data: any) { |
||||||
|
this.jobsRedisService.publish(`jobs-${job.id.toString()}`, { |
||||||
|
cmd: JobEvents.STATUS, |
||||||
|
id: job.id.toString(), |
||||||
|
status: JobStatus.COMPLETED, |
||||||
|
data: { |
||||||
|
result: data, |
||||||
|
}, |
||||||
|
}); |
||||||
|
} |
||||||
|
|
||||||
|
@OnEvent(JobEvents.LOG) |
||||||
|
onLog(data: { id: string; data: { message: string } }) { |
||||||
|
this.jobsRedisService.publish(`jobs-${data.id}`, { |
||||||
|
cmd: JobEvents.LOG, |
||||||
|
id: data.id, |
||||||
|
data: data.data, |
||||||
|
}); |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,53 @@ |
|||||||
|
import { Injectable } from '@nestjs/common'; |
||||||
|
import Redis from 'ioredis'; |
||||||
|
|
||||||
|
@Injectable() |
||||||
|
export class JobsRedisService { |
||||||
|
private redisClient: Redis; |
||||||
|
private redisSubscriber: Redis; |
||||||
|
private unsubscribeCallbacks: { [key: string]: () => void } = {}; |
||||||
|
|
||||||
|
constructor() { |
||||||
|
if (process.env['NC_WORKER_CONTAINER']) { |
||||||
|
this.redisClient = new Redis(process.env.NC_REDIS_URL); |
||||||
|
return; |
||||||
|
} |
||||||
|
this.redisSubscriber = new Redis(process.env.NC_REDIS_URL); |
||||||
|
} |
||||||
|
|
||||||
|
publish(channel: string, message: string | any) { |
||||||
|
if (typeof message === 'string') { |
||||||
|
this.redisClient.publish(channel, message); |
||||||
|
} else { |
||||||
|
try { |
||||||
|
this.redisClient.publish(channel, JSON.stringify(message)); |
||||||
|
} catch (e) { |
||||||
|
console.error(e); |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
subscribe(channel: string, callback: (message: any) => void) { |
||||||
|
this.redisSubscriber.subscribe(channel); |
||||||
|
|
||||||
|
const onMessage = (_channel, message) => { |
||||||
|
try { |
||||||
|
message = JSON.parse(message); |
||||||
|
} catch (e) {} |
||||||
|
callback(message); |
||||||
|
}; |
||||||
|
|
||||||
|
this.redisSubscriber.on('message', onMessage); |
||||||
|
this.unsubscribeCallbacks[channel] = () => { |
||||||
|
this.redisSubscriber.unsubscribe(channel); |
||||||
|
this.redisSubscriber.off('message', onMessage); |
||||||
|
}; |
||||||
|
} |
||||||
|
|
||||||
|
unsubscribe(channel: string) { |
||||||
|
if (this.unsubscribeCallbacks[channel]) { |
||||||
|
this.unsubscribeCallbacks[channel](); |
||||||
|
delete this.unsubscribeCallbacks[channel]; |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,90 @@ |
|||||||
|
import { InjectQueue } from '@nestjs/bull'; |
||||||
|
import { Injectable } from '@nestjs/common'; |
||||||
|
import { Queue } from 'bull'; |
||||||
|
import { EventEmitter2 } from '@nestjs/event-emitter'; |
||||||
|
import { JobEvents, JOBS_QUEUE, JobStatus } from '../../../interface/Jobs'; |
||||||
|
import { JobsRedisService } from './jobs-redis.service'; |
||||||
|
|
||||||
|
@Injectable() |
||||||
|
export class JobsService { |
||||||
|
private localJobs: string[] = []; |
||||||
|
|
||||||
|
constructor( |
||||||
|
@InjectQueue(JOBS_QUEUE) private readonly jobsQueue: Queue, |
||||||
|
private jobsRedisService: JobsRedisService, |
||||||
|
private eventEmitter: EventEmitter2, |
||||||
|
) { |
||||||
|
if (process.env['NC_REDIS_URL'] && !process.env['NC_WORKER_CONTAINER']) { |
||||||
|
this.jobsQueue.pause(true); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
async add(name: string, data: any) { |
||||||
|
const job = await this.jobsQueue.add(name, data); |
||||||
|
this.localJobs.push(job.id.toString()); |
||||||
|
this.jobsRedisService.subscribe(`jobs-${job.id.toString()}`, (data) => { |
||||||
|
const cmd = data.cmd; |
||||||
|
delete data.cmd; |
||||||
|
switch (cmd) { |
||||||
|
case JobEvents.STATUS: |
||||||
|
this.eventEmitter.emit(JobEvents.STATUS, data); |
||||||
|
if ([JobStatus.COMPLETED, JobStatus.FAILED].includes(data.status)) { |
||||||
|
this.jobsRedisService.unsubscribe(`jobs-${data.id.toString()}`); |
||||||
|
} |
||||||
|
break; |
||||||
|
case JobEvents.LOG: |
||||||
|
this.eventEmitter.emit(JobEvents.LOG, data); |
||||||
|
break; |
||||||
|
} |
||||||
|
}); |
||||||
|
return job; |
||||||
|
} |
||||||
|
|
||||||
|
async isLocalJob(jobId: string) { |
||||||
|
return this.localJobs.includes(jobId); |
||||||
|
} |
||||||
|
|
||||||
|
async removeLocalJob(jobId: string) { |
||||||
|
this.localJobs = this.localJobs.filter((j) => j !== jobId); |
||||||
|
} |
||||||
|
|
||||||
|
async jobStatus(jobId: string) { |
||||||
|
const job = await this.jobsQueue.getJob(jobId); |
||||||
|
if (job) { |
||||||
|
return await job.getState(); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
async jobList() { |
||||||
|
return await this.jobsQueue.getJobs([ |
||||||
|
JobStatus.ACTIVE, |
||||||
|
JobStatus.WAITING, |
||||||
|
JobStatus.DELAYED, |
||||||
|
JobStatus.PAUSED, |
||||||
|
]); |
||||||
|
} |
||||||
|
|
||||||
|
async getJobWithData(data: any) { |
||||||
|
const jobs = await this.jobsQueue.getJobs([ |
||||||
|
// 'completed',
|
||||||
|
JobStatus.WAITING, |
||||||
|
JobStatus.ACTIVE, |
||||||
|
JobStatus.DELAYED, |
||||||
|
// 'failed',
|
||||||
|
JobStatus.PAUSED, |
||||||
|
]); |
||||||
|
|
||||||
|
const job = jobs.find((j) => { |
||||||
|
for (const key in data) { |
||||||
|
if (j.data[key]) { |
||||||
|
if (j.data[key] !== data[key]) return false; |
||||||
|
} else { |
||||||
|
return false; |
||||||
|
} |
||||||
|
} |
||||||
|
return true; |
||||||
|
}); |
||||||
|
|
||||||
|
return job; |
||||||
|
} |
||||||
|
} |
||||||
@ -1,78 +0,0 @@ |
|||||||
import { Inject, Module, RequestMethod } from '@nestjs/common'; |
|
||||||
import { APP_FILTER } from '@nestjs/core'; |
|
||||||
import { BullModule } from '@nestjs/bull'; |
|
||||||
import { EventEmitterModule as NestJsEventEmitter } from '@nestjs/event-emitter'; |
|
||||||
import { Connection } from './connection/connection'; |
|
||||||
import { GlobalExceptionFilter } from './filters/global-exception/global-exception.filter'; |
|
||||||
import NcPluginMgrv2 from './helpers/NcPluginMgrv2'; |
|
||||||
import { DatasModule } from './modules/datas/datas.module'; |
|
||||||
import { IEventEmitter } from './modules/event-emitter/event-emitter.interface'; |
|
||||||
import { EventEmitterModule } from './modules/event-emitter/event-emitter.module'; |
|
||||||
import { AuthService } from './services/auth.service'; |
|
||||||
import { UsersModule } from './modules/users/users.module'; |
|
||||||
import { MetaService } from './meta/meta.service'; |
|
||||||
import Noco from './Noco'; |
|
||||||
import { TestModule } from './modules/test/test.module'; |
|
||||||
import { GlobalModule } from './modules/global/global.module'; |
|
||||||
import { HookHandlerService } from './services/hook-handler.service'; |
|
||||||
import { LocalStrategy } from './strategies/local.strategy'; |
|
||||||
import { AuthTokenStrategy } from './strategies/authtoken.strategy/authtoken.strategy'; |
|
||||||
import { BaseViewStrategy } from './strategies/base-view.strategy/base-view.strategy'; |
|
||||||
import { MetasModule } from './modules/metas/metas.module'; |
|
||||||
import NocoCache from './cache/NocoCache'; |
|
||||||
import { JobsModule } from './modules/jobs/jobs.module'; |
|
||||||
import type { OnApplicationBootstrap } from '@nestjs/common'; |
|
||||||
|
|
||||||
@Module({ |
|
||||||
imports: [ |
|
||||||
GlobalModule, |
|
||||||
UsersModule, |
|
||||||
...(process.env['PLAYWRIGHT_TEST'] === 'true' ? [TestModule] : []), |
|
||||||
MetasModule, |
|
||||||
DatasModule, |
|
||||||
EventEmitterModule, |
|
||||||
JobsModule, |
|
||||||
NestJsEventEmitter.forRoot(), |
|
||||||
...(process.env['NC_REDIS_URL'] |
|
||||||
? [ |
|
||||||
BullModule.forRoot({ |
|
||||||
url: process.env.NC_REDIS_URL, |
|
||||||
}), |
|
||||||
] |
|
||||||
: []), |
|
||||||
], |
|
||||||
controllers: [], |
|
||||||
providers: [ |
|
||||||
AuthService, |
|
||||||
{ |
|
||||||
provide: APP_FILTER, |
|
||||||
useClass: GlobalExceptionFilter, |
|
||||||
}, |
|
||||||
LocalStrategy, |
|
||||||
AuthTokenStrategy, |
|
||||||
BaseViewStrategy, |
|
||||||
HookHandlerService, |
|
||||||
], |
|
||||||
}) |
|
||||||
export class AppModule implements OnApplicationBootstrap { |
|
||||||
constructor( |
|
||||||
private readonly connection: Connection, |
|
||||||
private readonly metaService: MetaService, |
|
||||||
@Inject('IEventEmitter') private readonly eventEmitter: IEventEmitter, |
|
||||||
) {} |
|
||||||
|
|
||||||
// app init
|
|
||||||
async onApplicationBootstrap(): Promise<void> { |
|
||||||
process.env.NC_VERSION = '0105004'; |
|
||||||
|
|
||||||
await NocoCache.init(); |
|
||||||
|
|
||||||
// todo: remove
|
|
||||||
// temporary hack
|
|
||||||
Noco._ncMeta = this.metaService; |
|
||||||
Noco.config = this.connection.config; |
|
||||||
Noco.eventEmitter = this.eventEmitter; |
|
||||||
|
|
||||||
await NcPluginMgrv2.init(Noco.ncMeta); |
|
||||||
} |
|
||||||
} |
|
||||||
Loading…
Reference in new issue