From 929f0611e0c8729737b70a48a5725a1f0261e39d Mon Sep 17 00:00:00 2001 From: CarlosGamero Date: Tue, 18 Feb 2025 10:00:38 +0100 Subject: [PATCH] Fix script options + queue id type fix --- scripts/cmd/getUserImportJobs.ts | 4 ++-- scripts/generateOpenApi.ts | 6 ++++-- scripts/utils/cliCommandWrapper.ts | 9 ++++----- src/app.ts | 2 -- src/infrastructure/commonDiConfig.ts | 12 ++---------- src/infrastructure/config.ts | 4 +--- .../UserImportJob.spec.ts | 19 ++++++++++--------- .../job-queue-processors/UserImportJob.ts | 4 ++-- src/modules/users/userDiConfig.ts | 10 +++++++++- 9 files changed, 34 insertions(+), 36 deletions(-) diff --git a/scripts/cmd/getUserImportJobs.ts b/scripts/cmd/getUserImportJobs.ts index a557034d..2dd380ff 100644 --- a/scripts/cmd/getUserImportJobs.ts +++ b/scripts/cmd/getUserImportJobs.ts @@ -10,9 +10,9 @@ const ARGUMENTS_SCHEMA = z.object({ type Arguments = z.infer const command = async (deps: Dependencies, reqContext: RequestContext, args: Arguments) => { - const userImportJob = deps.userImportJob + const queueManager = deps.bullmqQueueManager - const jobs = await userImportJob.getJobsInQueue([args.queue]) + const jobs = await queueManager.getJobsInQueue('UserImportJob', [args.queue]) reqContext.logger.info(jobs, `${args.queue} jobs`) } diff --git a/scripts/generateOpenApi.ts b/scripts/generateOpenApi.ts index 34826a5c..164b9337 100644 --- a/scripts/generateOpenApi.ts +++ b/scripts/generateOpenApi.ts @@ -11,10 +11,12 @@ const targetPath = resolve(getRootDirectory(), 'openApiSpec.yaml') async function run() { const app = await getApp({ - queuesEnabled: false, - jobsEnabled: false, healthchecksEnabled: false, monitoringEnabled: false, + arePeriodicJobsEnabled: false, + consumersEnabled: false, + enqueuedJobsEnabled: false, + backgroundQueuesEnabled: true, }) const openApiSpecResponse = await app.inject().get('/documentation/openapi.json') diff --git a/scripts/utils/cliCommandWrapper.ts b/scripts/utils/cliCommandWrapper.ts index 1a93988f..5f5b48e6 100644 --- a/scripts/utils/cliCommandWrapper.ts +++ b/scripts/utils/cliCommandWrapper.ts @@ -27,13 +27,12 @@ export const cliCommandWrapper = async argsSchema?: ArgsSchema, ): Promise => { const app = await getApp({ - queuesEnabled: false, - jobsEnabled: false, healthchecksEnabled: false, monitoringEnabled: false, - app: { - cliMode: true, - }, + arePeriodicJobsEnabled: false, + consumersEnabled: false, + enqueuedJobsEnabled: false, + backgroundQueuesEnabled: true, }) const requestId = generateMonotonicUuid() diff --git a/src/app.ts b/src/app.ts index 830e3f78..e63f2c72 100644 --- a/src/app.ts +++ b/src/app.ts @@ -1,6 +1,5 @@ import { EventEmitter } from 'node:events' import type http from 'node:http' - import type { ServerZoneType } from '@amplitude/analytics-types' import fastifyAuth from '@fastify/auth' import { fastifyAwilixPlugin } from '@fastify/awilix' @@ -33,7 +32,6 @@ import { validatorCompiler, } from 'fastify-type-provider-zod' import type { ZodTypeProvider } from 'fastify-type-provider-zod' - import { merge } from 'ts-deepmerge' import type { PartialDeep } from 'type-fest' import { type Config, getConfig, isDevelopment } from './infrastructure/config.js' diff --git a/src/infrastructure/commonDiConfig.ts b/src/infrastructure/commonDiConfig.ts index 3d575f52..bda70273 100644 --- a/src/infrastructure/commonDiConfig.ts +++ b/src/infrastructure/commonDiConfig.ts @@ -35,10 +35,7 @@ import { ToadScheduler } from 'toad-scheduler' import type z from 'zod' import { FakeStoreApiClient } from '../integrations/FakeStoreApiClient.js' import { PermissionsMessages } from '../modules/users/consumers/permissionsMessageSchemas.js' -import { - USER_IMPORT_JOB_PAYLOAD, - UserImportJob, -} from '../modules/users/job-queue-processors/UserImportJob.js' +import { userBullmqQueues } from '../modules/users/userDiConfig.js' import { getAmqpConfig, getConfig, isTest } from './config.js' import type { Config } from './config.js' import { @@ -72,12 +69,7 @@ export type PublisherManager = AmqpTopicPublisherManager< AmqpSupportedMessages > -const bullmqSupportedQueues = [ - { - queueId: UserImportJob.QUEUE_ID, - jobPayloadSchema: USER_IMPORT_JOB_PAYLOAD, - }, -] as const satisfies QueueConfiguration[] +const bullmqSupportedQueues = [...userBullmqQueues] as const satisfies QueueConfiguration[] export type BullmqSupportedQueues = typeof bullmqSupportedQueues export function resolveCommonDiConfig( diff --git a/src/infrastructure/config.ts b/src/infrastructure/config.ts index 8fd166e5..f0145280 100644 --- a/src/infrastructure/config.ts +++ b/src/infrastructure/config.ts @@ -83,7 +83,6 @@ export type AppConfig = { metrics: { isEnabled: boolean } - cliMode: boolean } export function getConfig(): Config { @@ -201,12 +200,11 @@ export function getAppConfig(): AppConfig { nodeEnv: configScope.getMandatoryOneOf('NODE_ENV', ['production', 'development', 'test']), appEnv: configScope.getMandatoryOneOf('APP_ENV', ['production', 'development', 'staging']), appVersion: configScope.getOptional('APP_VERSION', 'VERSION_NOT_SET'), - baseUrl: configScope.getOptional('BASE_URL', ``), + baseUrl: configScope.getOptional('BASE_URL', ''), gitCommitSha: configScope.getOptional('GIT_COMMIT_SHA', 'COMMIT_SHA_NOT_SET'), metrics: { isEnabled: configScope.getOptionalBoolean('METRICS_ENABLED', !configScope.isDevelopment()), }, - cliMode: false, // If a different value is needed, it should be passed to the app directly } } diff --git a/src/modules/users/job-queue-processors/UserImportJob.spec.ts b/src/modules/users/job-queue-processors/UserImportJob.spec.ts index 740afd5a..d41bd4a1 100644 --- a/src/modules/users/job-queue-processors/UserImportJob.spec.ts +++ b/src/modules/users/job-queue-processors/UserImportJob.spec.ts @@ -4,23 +4,29 @@ import { cleanRedis } from '../../../../test/RedisCleaner.js' import type { TestContext } from '../../../../test/TestContext.js' import { createTestContext, destroyTestContext } from '../../../../test/TestContext.js' +import type { QueueManager } from '@lokalise/background-jobs-common' import { user as userTable } from '../../../db/schema/user.js' +import type { BullmqSupportedQueues } from '../../../infrastructure/commonDiConfig.js' import { UserImportJob } from './UserImportJob.js' describe('UserImportJob', () => { let testContext: TestContext let userImportJob: UserImportJob + let bullmqQueueManager: QueueManager + beforeAll(async () => { testContext = await createTestContext( {}, { enqueuedJobsEnabled: [UserImportJob.QUEUE_ID], + backgroundQueuesEnabled: [UserImportJob.QUEUE_ID], }, ) await cleanRedis(testContext.diContainer.cradle.redis) await cleanTables(testContext.diContainer.cradle.drizzle, [DB_MODEL.User]) userImportJob = testContext.diContainer.cradle.userImportJob + bullmqQueueManager = testContext.diContainer.cradle.bullmqQueueManager }) afterAll(async () => { @@ -34,16 +40,11 @@ describe('UserImportJob', () => { email: 'test@email.lt', } - const jobId = await userImportJob.schedule({ - payload: userData, - metadata: { - correlationId: 'dummy', - }, + const jobId = await bullmqQueueManager.schedule('UserImportJob', { + ...userData, + metadata: { correlationId: 'dummy' }, }) - const result = await testContext.diContainer.cradle.userImportJob.spy.waitForJobWithId( - jobId, - 'completed', - ) + const result = await userImportJob.spy.waitForJobWithId(jobId, 'completed') const users = await testContext.diContainer.cradle.drizzle.select().from(userTable) expect(users).toHaveLength(1) diff --git a/src/modules/users/job-queue-processors/UserImportJob.ts b/src/modules/users/job-queue-processors/UserImportJob.ts index e54f3fc9..e79036d8 100644 --- a/src/modules/users/job-queue-processors/UserImportJob.ts +++ b/src/modules/users/job-queue-processors/UserImportJob.ts @@ -16,7 +16,7 @@ export const USER_IMPORT_JOB_PAYLOAD = BASE_JOB_PAYLOAD_SCHEMA.extend({ type UserImportJobPayload = z.infer export class UserImportJob extends AbstractEnqueuedJobProcessor<'UserImportJob'> { - public static QUEUE_ID = 'UserImportJob' + public static readonly QUEUE_ID = 'UserImportJob' private readonly userService: UserService constructor(dependencies: Dependencies) { @@ -39,7 +39,7 @@ export class UserImportJob extends AbstractEnqueuedJobProcessor<'UserImportJob'> { userId: user.id, }, - `Created new user`, + 'Created new user', ) } } diff --git a/src/modules/users/userDiConfig.ts b/src/modules/users/userDiConfig.ts index fc924fb7..bf39caae 100644 --- a/src/modules/users/userDiConfig.ts +++ b/src/modules/users/userDiConfig.ts @@ -11,10 +11,11 @@ import { } from '../../infrastructure/diConfigUtils.js' import { SINGLETON_CONFIG } from '../../infrastructure/parentDiConfig.js' +import type { QueueConfiguration } from '@lokalise/background-jobs-common' import type { User } from '../../db/schema/user.js' import { PermissionConsumer } from './consumers/PermissionConsumer.js' import { UserDataSource } from './datasources/UserDataSource.js' -import { UserImportJob } from './job-queue-processors/UserImportJob.js' +import { USER_IMPORT_JOB_PAYLOAD, UserImportJob } from './job-queue-processors/UserImportJob.js' import { DeleteOldUsersJob } from './periodic-jobs/DeleteOldUsersJob.js' import { ProcessLogFilesJob } from './periodic-jobs/ProcessLogFilesJob.js' import { SendEmailsJob } from './periodic-jobs/SendEmailsJob.js' @@ -126,3 +127,10 @@ export function resolveUsersConfig(options: DIOptions): UsersDiConfig { }), } } + +export const userBullmqQueues = [ + { + queueId: UserImportJob.QUEUE_ID, + jobPayloadSchema: USER_IMPORT_JOB_PAYLOAD, + }, +] as const satisfies QueueConfiguration[]