feat(core): Cache test webhook registrations (#8176)

In a multi-main setup, we have the following issue. The user's client
connects to main A and runs a test webhook, so main A starts listening
for a webhook call. A third-party service sends a request to the test
webhook URL. The request is forwarded by the load balancer to main B,
who is not listening for this webhook call. Therefore, the webhook call
is unhandled.

To start addressing this, cache test webhook registrations, using Redis
for queue mode and in-memory for regular mode. When the third-party
service sends a request to the test webhook URL, the request is
forwarded by the load balancer to main B, who fetches test webhooks from
the cache and, if it finds a match, executes the test webhook. This
should be transparent - test webhook behavior should remain the same as
so far.

Notes:
- Test webhook timeouts are not cached. A timeout is only relevant to
the process it was created in, so another process retrieving from Redis
a "foreign" timeout will be unable to act on it. A timeout also has
circular references, so `cache-manager-ioredis-yet` is unable to
serialize it.
- In a single-main scenario, the timeout remains in the single process
and is cleared on test webhook expiration, successful execution, and
manual cancellation - all as usual.
- In a multi-main scenario, we will need to have the process who
received the webhook call send a message to the process who created the
webhook directing this originating process to clear the timeout. This
will likely be implemented via execution lifecycle hooks and Redis
channel messages checking session ID. This implementation is out of
scope for this PR and will come next.
- Additional data in test webhooks is not cached. From what I can tell,
additional data is not needed for test webhooks to be executed.
Additional data also has circular references, so
`cache-manager-ioredis-yet` is unable to serialize it.

Follow-up to: #8155

packages/cli/src/services/cache.service.ts

@@ -26,6 +26,10 @@ export class CacheService extends EventEmitter {
 		return (this.cache as RedisCache)?.store?.isCacheable !== undefined;
 	}
 
+	isMemoryCache(): boolean {
+		return !this.isRedisCache();
+	}
+
 	/**
 	 * Initialize the cache service.
 	 *
@@ -118,15 +122,15 @@ export class CacheService extends EventEmitter {
 	 * @param options.refreshTtl Optional ttl for the refreshFunction's set call
 	 * @param options.fallbackValue Optional value returned is cache is not hit and refreshFunction is not provided
 	 */
-	async getMany(
+	async getMany<T = unknown[]>(
 		keys: string[],
 		options: {
-			fallbackValues?: unknown[];
-			refreshFunctionEach?: (key: string) => Promise<unknown>;
-			refreshFunctionMany?: (keys: string[]) => Promise<unknown[]>;
+			fallbackValues?: T[];
+			refreshFunctionEach?: (key: string) => Promise<T>;
+			refreshFunctionMany?: (keys: string[]) => Promise<T[]>;
 			refreshTtl?: number;
 		} = {},
-	): Promise<unknown[]> {
+	): Promise<T[]> {
 		if (keys.length === 0) {
 			return [];
 		}
@@ -136,7 +140,7 @@ export class CacheService extends EventEmitter {
 		}
 		if (!values.includes(undefined)) {
 			this.emit(this.metricsCounterEvents.cacheHit);
-			return values;
+			return values as T[];
 		}
 		this.emit(this.metricsCounterEvents.cacheMiss);
 		if (options.refreshFunctionEach) {
@@ -155,7 +159,7 @@ export class CacheService extends EventEmitter {
 					values[i] = refreshValue;
 				}
 			}
-			return values;
+			return values as T[];
 		}
 		if (options.refreshFunctionMany) {
 			this.emit(this.metricsCounterEvents.cacheUpdate);
@@ -170,9 +174,9 @@ export class CacheService extends EventEmitter {
 				newKV.push([keys[i], refreshValues[i]]);
 			}
 			await this.setMany(newKV, options.refreshTtl);
-			return refreshValues;
+			return refreshValues as T[];
 		}
-		return options.fallbackValues ?? values;
+		return (options.fallbackValues ?? values) as T[];
 	}
 
 	/**
@@ -196,6 +200,7 @@ export class CacheService extends EventEmitter {
 				throw new ApplicationError('Value is not cacheable');
 			}
 		}
+
 		await this.cache?.store.set(key, value, ttl);
 	}
 
@@ -284,7 +289,9 @@ export class CacheService extends EventEmitter {
 	}
 
 	/**
-	 * Return all keys in the cache.
+	 * Return all keys in the cache. Not recommended for production use.
+	 *
+	 * https://redis.io/commands/keys/
 	 */
 	async keys(): Promise<string[]> {
 		return this.cache?.store.keys() ?? [];

packages/cli/src/services/test-webhook-registrations.service.ts

@@ -0,0 +1,94 @@
+import { Service } from 'typedi';
+import { CacheService } from './cache.service';
+import { ApplicationError, type IWebhookData } from 'n8n-workflow';
+import type { IWorkflowDb } from '@/Interfaces';
+
+export type TestWebhookRegistration = {
+	sessionId?: string;
+	workflowEntity: IWorkflowDb;
+	destinationNode?: string;
+	webhook: IWebhookData;
+};
+
+@Service()
+export class TestWebhookRegistrationsService {
+	constructor(private readonly cacheService: CacheService) {}
+
+	private readonly cacheKey = 'test-webhook';
+
+	async register(registration: TestWebhookRegistration) {
+		const key = this.toKey(registration.webhook);
+
+		await this.cacheService.set(key, registration);
+	}
+
+	async deregister(arg: IWebhookData | string) {
+		if (typeof arg === 'string') {
+			await this.cacheService.delete(arg);
+		} else {
+			const key = this.toKey(arg);
+			await this.cacheService.delete(key);
+		}
+	}
+
+	async get(key: string) {
+		return this.cacheService.get<TestWebhookRegistration>(key);
+	}
+
+	async getAllKeys() {
+		const keys = await this.cacheService.keys();
+
+		if (this.cacheService.isMemoryCache()) {
+			return keys.filter((key) => key.startsWith(this.cacheKey));
+		}
+
+		const prefix = 'n8n:cache'; // prepended by Redis cache
+		const extendedCacheKey = `${prefix}:${this.cacheKey}`;
+
+		return keys
+			.filter((key) => key.startsWith(extendedCacheKey))
+			.map((key) => key.slice(`${prefix}:`.length));
+	}
+
+	async getAllRegistrations() {
+		const keys = await this.getAllKeys();
+
+		return this.cacheService.getMany<TestWebhookRegistration>(keys);
+	}
+
+	async updateWebhookProperties(newProperties: IWebhookData) {
+		const key = this.toKey(newProperties);
+
+		const registration = await this.cacheService.get<TestWebhookRegistration>(key);
+
+		if (!registration) {
+			throw new ApplicationError('Failed to find test webhook registration', { extra: { key } });
+		}
+
+		registration.webhook = newProperties;
+
+		await this.cacheService.set(key, registration);
+	}
+
+	async deregisterAll() {
+		const testWebhookKeys = await this.getAllKeys();
+
+		await this.cacheService.deleteMany(testWebhookKeys);
+	}
+
+	toKey(webhook: Pick<IWebhookData, 'webhookId' | 'httpMethod' | 'path'>) {
+		const { webhookId, httpMethod, path: webhookPath } = webhook;
+
+		if (!webhookId) return `${this.cacheKey}:${httpMethod}|${webhookPath}`;
+
+		let path = webhookPath;
+
+		if (path.startsWith(webhookId)) {
+			const cutFromIndex = path.indexOf('/') + 1;
+
+			path = path.slice(cutFromIndex);
+		}
+
+		return `${this.cacheKey}:${httpMethod}|${webhookId}|${path.split('/').length}`;
+	}
+}