Deduplicate

n8n-io · Oct 7, 2024 · b920fd6 · b920fd6
1 parent 989c290
commit b920fd6
Show file tree

Hide file tree

Showing 2 changed files with 58 additions and 54 deletions.
diff --git a/packages/cli/src/scaling/__tests__/pubsub-handler.test.ts b/packages/cli/src/scaling/__tests__/pubsub-handler.test.ts
@@ -29,7 +29,7 @@ describe('PubSubHandler', () => {
 
 		it('should set up handlers in webhook process', () => {
 			// @ts-expect-error Spying on private method
-			const setupWebhookHandlersSpy = jest.spyOn(PubSubHandler.prototype, 'setupWebhookHandlers');
+			const setupHandlersSpy = jest.spyOn(PubSubHandler.prototype, 'setupHandlers');
 
 			new PubSubHandler(
 				eventService,
@@ -42,7 +42,14 @@ describe('PubSubHandler', () => {
 				workerStatus,
 			).init();
 
-			expect(setupWebhookHandlersSpy).toHaveBeenCalled();
+			expect(setupHandlersSpy).toHaveBeenCalledWith({
+				'reload-license': expect.any(Function),
+				'restart-event-bus': expect.any(Function),
+				'reload-external-secrets-providers': expect.any(Function),
+				'community-package-install': expect.any(Function),
+				'community-package-update': expect.any(Function),
+				'community-package-uninstall': expect.any(Function),
+			});
 		});
 
 		it('should reload license on `reload-license` event', () => {
@@ -167,7 +174,7 @@ describe('PubSubHandler', () => {
 
 		it('should set up handlers in worker process', () => {
 			// @ts-expect-error Spying on private method
-			const setupWorkerHandlersSpy = jest.spyOn(PubSubHandler.prototype, 'setupWorkerHandlers');
+			const setupHandlersSpy = jest.spyOn(PubSubHandler.prototype, 'setupHandlers');
 
 			new PubSubHandler(
 				eventService,
@@ -180,7 +187,16 @@ describe('PubSubHandler', () => {
 				workerStatus,
 			).init();
 
-			expect(setupWorkerHandlersSpy).toHaveBeenCalled();
+			expect(setupHandlersSpy).toHaveBeenCalledWith({
+				'reload-license': expect.any(Function),
+				'restart-event-bus': expect.any(Function),
+				'reload-external-secrets-providers': expect.any(Function),
+				'community-package-install': expect.any(Function),
+				'community-package-update': expect.any(Function),
+				'community-package-uninstall': expect.any(Function),
+				'get-worker-status': expect.any(Function),
+				'get-worker-id': expect.any(Function),
+			});
 		});
 
 		it('should reload license on `reload-license` event', () => {

diff --git a/packages/cli/src/scaling/pubsub/pubsub-handler.ts b/packages/cli/src/scaling/pubsub/pubsub-handler.ts
@@ -32,10 +32,23 @@ export class PubSubHandler {
 	init() {
 		switch (this.instanceSettings.instanceType) {
 			case 'webhook':
-				this.setupWebhookHandlers();
+				this.setupHandlers(this.commonHandlers);
 				break;
 			case 'worker':
-				this.setupWorkerHandlers();
+				this.setupHandlers({
+					...this.commonHandlers,
+					'get-worker-status': async () =>
+						await this.publisher.publishWorkerResponse({
+							workerId: config.getEnv('redis.queueModeId'),
+							command: 'get-worker-status',
+							payload: this.workerStatus.generateStatus(),
+						}),
+					'get-worker-id': async () =>
+						await this.publisher.publishWorkerResponse({
+							workerId: config.getEnv('redis.queueModeId'),
+							command: 'get-worker-id',
+						}),
+				});
 				break;
 			case 'main':
 				// TODO
@@ -59,52 +72,27 @@ export class PubSubHandler {
 		}
 	}
 
-	// #region Webhook process
-
-	private setupWebhookHandlers() {
-		this.setupHandlers({
-			'reload-license': async () => await this.license.reload(),
-			'restart-event-bus': async () => await this.eventbus.restart(),
-			'reload-external-secrets-providers': async () =>
-				await this.externalSecretsManager.reloadAllProviders(),
-			'community-package-install': async ({ packageName, packageVersion }) =>
-				await this.communityPackagesService.installOrUpdateNpmPackage(packageName, packageVersion),
-			'community-package-update': async ({ packageName, packageVersion }) =>
-				await this.communityPackagesService.installOrUpdateNpmPackage(packageName, packageVersion),
-			'community-package-uninstall': async ({ packageName }) =>
-				await this.communityPackagesService.removeNpmPackage(packageName),
-		});
-	}
-
-	// #endregion
-
-	// #region Worker process
-
-	private setupWorkerHandlers() {
-		this.setupHandlers({
-			'reload-license': async () => await this.license.reload(),
-			'restart-event-bus': async () => await this.eventbus.restart(),
-			'reload-external-secrets-providers': async () =>
-				await this.externalSecretsManager.reloadAllProviders(),
-			'community-package-install': async ({ packageName, packageVersion }) =>
-				await this.communityPackagesService.installOrUpdateNpmPackage(packageName, packageVersion),
-			'community-package-update': async ({ packageName, packageVersion }) =>
-				await this.communityPackagesService.installOrUpdateNpmPackage(packageName, packageVersion),
-			'community-package-uninstall': async ({ packageName }) =>
-				await this.communityPackagesService.removeNpmPackage(packageName),
-			'get-worker-status': async () =>
-				await this.publisher.publishWorkerResponse({
-					workerId: config.getEnv('redis.queueModeId'),
-					command: 'get-worker-status',
-					payload: this.workerStatus.generateStatus(),
-				}),
-			'get-worker-id': async () =>
-				await this.publisher.publishWorkerResponse({
-					workerId: config.getEnv('redis.queueModeId'),
-					command: 'get-worker-id',
-				}),
-		});
-	}
-
-	// #endregion
+	/** Handlers shared by webhook and worker processes. */
+	private commonHandlers: {
+		[K in keyof Pick<
+			PubSubEventMap,
+			| 'reload-license'
+			| 'restart-event-bus'
+			| 'reload-external-secrets-providers'
+			| 'community-package-install'
+			| 'community-package-update'
+			| 'community-package-uninstall'
+		>]: (event: PubSubEventMap[K]) => Promise<void>;
+	} = {
+		'reload-license': async () => await this.license.reload(),
+		'restart-event-bus': async () => await this.eventbus.restart(),
+		'reload-external-secrets-providers': async () =>
+			await this.externalSecretsManager.reloadAllProviders(),
+		'community-package-install': async ({ packageName, packageVersion }) =>
+			await this.communityPackagesService.installOrUpdateNpmPackage(packageName, packageVersion),
+		'community-package-update': async ({ packageName, packageVersion }) =>
+			await this.communityPackagesService.installOrUpdateNpmPackage(packageName, packageVersion),
+		'community-package-uninstall': async ({ packageName }) =>
+			await this.communityPackagesService.removeNpmPackage(packageName),
+	};
 }