mirror of
https://github.com/n8n-io/n8n.git
synced 2024-12-24 04:04:06 -08:00
perf(core): Improve caching service (#8213)
Story: https://linear.app/n8n/issue/PAY-1188 - Implement Redis hashes on the caching service, based on Micha's work in #7747, adapted from `node-cache-manager-ioredis-yet`. Optimize workflow ownership lookups and manual webhook lookups with Redis hashes. - Simplify the caching service by removing all currently unused methods and options: `enable`, `disable`, `getCache`, `keys`, `keyValues`, `refreshFunctionEach`, `refreshFunctionMany`, `refreshTtl`, etc. - Remove the flag `N8N_CACHE_ENABLED`. Currently some features on `master` are broken with caching disabled, and test webhooks now rely entirely on caching, for multi-main setup support. We originally introduced this flag to protect against excessive memory usage, but total cache usage is low enough that we decided to drop this setting. Apparently this flag was also never documented. - Overall caching service refactor: use generics, reduce branching, add discriminants for cache kinds for better type safety, type caching events, improve readability, remove outdated docs, etc. Also refactor and expand caching service tests. Follow-up to: https://github.com/n8n-io/n8n/pull/8176 --------- Co-authored-by: Michael Auerswald <michael.auerswald@gmail.com>
This commit is contained in:
Iván Ovejero
GitHub
parent
b201ff8f23
commit
f53c482939
|
|
@ -2,6 +2,16 @@
|
|||
|
||||
This list shows all the versions which include breaking changes and how to upgrade.
|
||||
|
||||
## 1.24.0
|
||||
|
||||
### What changed?
|
||||
|
||||
The flag `N8N_CACHE_ENABLED` was removed. The cache is now always enabled.
|
||||
|
||||
### When is action necessary?
|
||||
|
||||
If you are using the flag `N8N_CACHE_ENABLED`, remove it from your settings.
|
||||
|
||||
## 1.22.0
|
||||
|
||||
### What changed?
|
||||
|
|
|
|||
|
|
@ -113,7 +113,6 @@
|
|||
"bcryptjs": "2.4.3",
|
||||
"bull": "4.10.2",
|
||||
"cache-manager": "5.2.3",
|
||||
"cache-manager-ioredis-yet": "1.2.2",
|
||||
"callsites": "3.1.0",
|
||||
"change-case": "4.1.2",
|
||||
"class-transformer": "0.5.1",
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import { Service } from 'typedi';
|
||||
import { CacheService } from './services/cache.service';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import { jsonParse } from 'n8n-workflow';
|
||||
|
||||
type ActivationErrors = {
|
||||
|
|
|
|||
|
|
@ -323,6 +323,8 @@ export class TestWebhooks implements IWebhookManager {
|
|||
async deactivateWebhooks(workflow: Workflow) {
|
||||
const allRegistrations = await this.registrations.getAllRegistrations();
|
||||
|
||||
if (!allRegistrations.length) return; // nothing to deactivate
|
||||
|
||||
type WebhooksByWorkflow = { [workflowId: string]: IWebhookData[] };
|
||||
|
||||
const webhooksByWorkflow = allRegistrations.reduce<WebhooksByWorkflow>((acc, cur) => {
|
||||
|
|
|
|||
|
|
@ -1279,12 +1279,6 @@ export const schema = {
|
|||
},
|
||||
|
||||
cache: {
|
||||
enabled: {
|
||||
doc: 'Whether caching is enabled',
|
||||
format: Boolean,
|
||||
default: true,
|
||||
env: 'N8N_CACHE_ENABLED',
|
||||
},
|
||||
backend: {
|
||||
doc: 'Backend to use for caching',
|
||||
format: ['memory', 'redis', 'auto'] as const,
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ import type { UserSetupPayload } from '@/requests';
|
|||
import type { BooleanLicenseFeature, IPushDataType } from '@/Interfaces';
|
||||
import { MfaService } from '@/Mfa/mfa.service';
|
||||
import { Push } from '@/push';
|
||||
import { CacheService } from '@/services/cache.service';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import { PasswordUtility } from '@/services/password.utility';
|
||||
|
||||
if (!inE2ETests) {
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ import type { Variables } from '@db/entities/Variables';
|
|||
import { InternalHooks } from '@/InternalHooks';
|
||||
import { generateNanoId } from '@db/utils/generators';
|
||||
import { canCreateNewVariable } from './environmentHelpers';
|
||||
import { CacheService } from '@/services/cache.service';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import { VariablesRepository } from '@db/repositories/variables.repository';
|
||||
import { VariableCountLimitReachedError } from '@/errors/variable-count-limit-reached.error';
|
||||
import { VariableValidationError } from '@/errors/variable-validation.error';
|
||||
|
|
@ -17,8 +17,7 @@ export class VariablesService {
|
|||
|
||||
async getAllCached(): Promise<Variables[]> {
|
||||
const variables = await this.cacheService.get('variables', {
|
||||
async refreshFunction() {
|
||||
// TODO: log refresh cache metric
|
||||
async refreshFn() {
|
||||
return Container.get(VariablesService).findAll();
|
||||
},
|
||||
});
|
||||
|
|
|
|||
|
|
@ -0,0 +1,7 @@
|
|||
import { ApplicationError } from 'n8n-workflow';
|
||||
|
||||
export class MalformedRefreshValueError extends ApplicationError {
|
||||
constructor() {
|
||||
super('Refresh value must have the same number of values as keys');
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
import { ApplicationError } from 'n8n-workflow';
|
||||
|
||||
export class UncacheableValueError extends ApplicationError {
|
||||
constructor(key: string) {
|
||||
super('Value cannot be cached in Redis', {
|
||||
extra: { key, hint: 'Does the value contain circular references?' },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -1,317 +0,0 @@
|
|||
import { Service } from 'typedi';
|
||||
import config from '@/config';
|
||||
import { caching } from 'cache-manager';
|
||||
import type { MemoryCache } from 'cache-manager';
|
||||
import type { RedisCache } from 'cache-manager-ioredis-yet';
|
||||
import { ApplicationError, jsonStringify } from 'n8n-workflow';
|
||||
import { getDefaultRedisClient, getRedisPrefix } from './redis/RedisServiceHelper';
|
||||
import EventEmitter from 'events';
|
||||
|
||||
@Service()
|
||||
export class CacheService extends EventEmitter {
|
||||
/**
|
||||
* Keys and values:
|
||||
* - `'cache:workflow-owner:${workflowId}'`: `User`
|
||||
*/
|
||||
private cache: RedisCache | MemoryCache | undefined;
|
||||
|
||||
metricsCounterEvents = {
|
||||
cacheHit: 'metrics.cache.hit',
|
||||
|
||||
cacheMiss: 'metrics.cache.miss',
|
||||
cacheUpdate: 'metrics.cache.update',
|
||||
};
|
||||
|
||||
isRedisCache(): boolean {
|
||||
return (this.cache as RedisCache)?.store?.isCacheable !== undefined;
|
||||
}
|
||||
|
||||
isMemoryCache(): boolean {
|
||||
return !this.isRedisCache();
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the cache service.
|
||||
*
|
||||
* If the cache is enabled, it will initialize the cache from the provided config options. By default, it will use
|
||||
* the `memory` backend and create a simple in-memory cache. If running in `queue` mode, or if `redis` backend is selected,
|
||||
* it use Redis as the cache backend (either a local Redis instance or a Redis cluster, depending on the config)
|
||||
*
|
||||
* If the cache is disabled, this does nothing.
|
||||
*/
|
||||
async init(): Promise<void> {
|
||||
if (!config.getEnv('cache.enabled')) {
|
||||
return;
|
||||
}
|
||||
const backend = config.getEnv('cache.backend');
|
||||
if (
|
||||
backend === 'redis' ||
|
||||
(backend === 'auto' && config.getEnv('executions.mode') === 'queue')
|
||||
) {
|
||||
const { redisInsStore } = await import('cache-manager-ioredis-yet');
|
||||
const redisPrefix = getRedisPrefix(config.getEnv('redis.prefix'));
|
||||
const cachePrefix = config.getEnv('cache.redis.prefix');
|
||||
const keyPrefix = `${redisPrefix}:${cachePrefix}:`;
|
||||
const redisClient = await getDefaultRedisClient({ keyPrefix }, 'client(cache)');
|
||||
const redisStore = redisInsStore(redisClient, {
|
||||
ttl: config.getEnv('cache.redis.ttl'),
|
||||
});
|
||||
this.cache = await caching(redisStore);
|
||||
} else {
|
||||
// using TextEncoder to get the byte length of the string even if it contains unicode characters
|
||||
const textEncoder = new TextEncoder();
|
||||
this.cache = await caching('memory', {
|
||||
ttl: config.getEnv('cache.memory.ttl'),
|
||||
maxSize: config.getEnv('cache.memory.maxSize'),
|
||||
sizeCalculation: (item) => {
|
||||
return textEncoder.encode(jsonStringify(item, { replaceCircularRefs: true })).length;
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get a value from the cache by key.
|
||||
*
|
||||
* If the value is not in the cache or expired, the refreshFunction is called if defined,
|
||||
* which will set the key with the function's result and returns it. If no refreshFunction is set, the fallback value is returned.
|
||||
*
|
||||
* If the cache is disabled, refreshFunction's result or fallbackValue is returned.
|
||||
*
|
||||
* If cache is not hit, and neither refreshFunction nor fallbackValue are provided, `undefined` is returned.
|
||||
* @param key The key to fetch from the cache
|
||||
* @param options.refreshFunction Optional function to call to set the cache if the key is not found
|
||||
* @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 get<T = unknown>(
|
||||
key: string,
|
||||
options: {
|
||||
fallbackValue?: T;
|
||||
refreshFunction?: (key: string) => Promise<T>;
|
||||
refreshTtl?: number;
|
||||
} = {},
|
||||
): Promise<T | undefined> {
|
||||
if (!key || key.length === 0) {
|
||||
return;
|
||||
}
|
||||
const value = await this.cache?.store.get(key);
|
||||
if (value !== undefined) {
|
||||
this.emit(this.metricsCounterEvents.cacheHit);
|
||||
return value as T;
|
||||
}
|
||||
this.emit(this.metricsCounterEvents.cacheMiss);
|
||||
if (options.refreshFunction) {
|
||||
this.emit(this.metricsCounterEvents.cacheUpdate);
|
||||
const refreshValue = await options.refreshFunction(key);
|
||||
await this.set(key, refreshValue, options.refreshTtl);
|
||||
return refreshValue;
|
||||
}
|
||||
return options.fallbackValue ?? undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get many values from a list of keys.
|
||||
*
|
||||
* If a value is not in the cache or expired, the returned list will have `undefined` at that index.
|
||||
* If the cache is disabled, refreshFunction's result or fallbackValue is returned.
|
||||
* If cache is not hit, and neither refreshFunction nor fallbackValue are provided, a list of `undefined` is returned.
|
||||
* @param keys A list of keys to fetch from the cache
|
||||
* @param options.refreshFunctionEach Optional, if defined, undefined values will be replaced with the result of the refreshFunctionEach call and the cache will be updated
|
||||
* @param options.refreshFunctionMany Optional, if defined, all values will be replaced with the result of the refreshFunctionMany call and the cache will be updated
|
||||
* @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<T = unknown[]>(
|
||||
keys: string[],
|
||||
options: {
|
||||
fallbackValues?: T[];
|
||||
refreshFunctionEach?: (key: string) => Promise<T>;
|
||||
refreshFunctionMany?: (keys: string[]) => Promise<T[]>;
|
||||
refreshTtl?: number;
|
||||
} = {},
|
||||
): Promise<T[]> {
|
||||
if (keys.length === 0) {
|
||||
return [];
|
||||
}
|
||||
let values = await this.cache?.store.mget(...keys);
|
||||
if (values === undefined) {
|
||||
values = keys.map(() => undefined);
|
||||
}
|
||||
if (!values.includes(undefined)) {
|
||||
this.emit(this.metricsCounterEvents.cacheHit);
|
||||
return values as T[];
|
||||
}
|
||||
this.emit(this.metricsCounterEvents.cacheMiss);
|
||||
if (options.refreshFunctionEach) {
|
||||
for (let i = 0; i < keys.length; i++) {
|
||||
if (values[i] === undefined) {
|
||||
const key = keys[i];
|
||||
let fallback = undefined;
|
||||
if (options.fallbackValues && options.fallbackValues.length > i) {
|
||||
fallback = options.fallbackValues[i];
|
||||
}
|
||||
const refreshValue = await this.get(key, {
|
||||
refreshFunction: options.refreshFunctionEach,
|
||||
refreshTtl: options.refreshTtl,
|
||||
fallbackValue: fallback,
|
||||
});
|
||||
values[i] = refreshValue;
|
||||
}
|
||||
}
|
||||
return values as T[];
|
||||
}
|
||||
if (options.refreshFunctionMany) {
|
||||
this.emit(this.metricsCounterEvents.cacheUpdate);
|
||||
const refreshValues: unknown[] = await options.refreshFunctionMany(keys);
|
||||
if (keys.length !== refreshValues.length) {
|
||||
throw new ApplicationError(
|
||||
'refreshFunctionMany must return the same number of values as keys',
|
||||
);
|
||||
}
|
||||
const newKV: Array<[string, unknown]> = [];
|
||||
for (let i = 0; i < keys.length; i++) {
|
||||
newKV.push([keys[i], refreshValues[i]]);
|
||||
}
|
||||
await this.setMany(newKV, options.refreshTtl);
|
||||
return refreshValues as T[];
|
||||
}
|
||||
return (options.fallbackValues ?? values) as T[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Set a value in the cache by key.
|
||||
* @param key The key to set
|
||||
* @param value The value to set
|
||||
* @param ttl Optional time to live in ms
|
||||
*/
|
||||
async set(key: string, value: unknown, ttl?: number): Promise<void> {
|
||||
if (!this.cache) {
|
||||
await this.init();
|
||||
}
|
||||
if (!key || key.length === 0) {
|
||||
return;
|
||||
}
|
||||
if (value === undefined || value === null) {
|
||||
return;
|
||||
}
|
||||
if (this.isRedisCache()) {
|
||||
if (!(this.cache as RedisCache)?.store?.isCacheable(value)) {
|
||||
throw new ApplicationError('Value is not cacheable');
|
||||
}
|
||||
}
|
||||
|
||||
await this.cache?.store.set(key, value, ttl);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set a multiple values in the cache at once.
|
||||
* @param values An array of [key, value] tuples to set
|
||||
* @param ttl Optional time to live in ms
|
||||
*/
|
||||
async setMany(values: Array<[string, unknown]>, ttl?: number): Promise<void> {
|
||||
if (!this.cache) {
|
||||
await this.init();
|
||||
}
|
||||
if (values.length === 0) {
|
||||
return;
|
||||
}
|
||||
const nonNullValues = values.filter(
|
||||
([key, value]) => value !== undefined && value !== null && key && key.length > 0,
|
||||
);
|
||||
if (this.isRedisCache()) {
|
||||
nonNullValues.forEach(([_key, value]) => {
|
||||
if (!(this.cache as RedisCache)?.store?.isCacheable(value)) {
|
||||
throw new ApplicationError('Value is not cacheable');
|
||||
}
|
||||
});
|
||||
}
|
||||
await this.cache?.store.mset(nonNullValues, ttl);
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete a value from the cache by key.
|
||||
* @param key The key to delete
|
||||
*/
|
||||
async delete(key: string): Promise<void> {
|
||||
if (!key || key.length === 0) {
|
||||
return;
|
||||
}
|
||||
await this.cache?.store.del(key);
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete multiple values from the cache.
|
||||
* @param keys List of keys to delete
|
||||
*/
|
||||
async deleteMany(keys: string[]): Promise<void> {
|
||||
if (keys.length === 0) {
|
||||
return;
|
||||
}
|
||||
return this.cache?.store.mdel(...keys);
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete all values and uninitialized the cache.
|
||||
*/
|
||||
async destroy() {
|
||||
if (this.cache) {
|
||||
await this.reset();
|
||||
this.cache = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Enable and initialize the cache.
|
||||
*/
|
||||
async enable() {
|
||||
config.set('cache.enabled', true);
|
||||
await this.init();
|
||||
}
|
||||
|
||||
/**
|
||||
* Disable and destroy the cache.
|
||||
*/
|
||||
async disable() {
|
||||
config.set('cache.enabled', false);
|
||||
await this.destroy();
|
||||
}
|
||||
|
||||
async getCache(): Promise<RedisCache | MemoryCache | undefined> {
|
||||
if (!this.cache) {
|
||||
await this.init();
|
||||
}
|
||||
return this.cache;
|
||||
}
|
||||
|
||||
async reset(): Promise<void> {
|
||||
await this.cache?.store.reset();
|
||||
}
|
||||
|
||||
/**
|
||||
* 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() ?? [];
|
||||
}
|
||||
|
||||
/**
|
||||
* Return all key/value pairs in the cache. This is a potentially very expensive operation and is only meant to be used for debugging
|
||||
*/
|
||||
async keyValues(): Promise<Map<string, unknown>> {
|
||||
const keys = await this.keys();
|
||||
const values = await this.getMany(keys);
|
||||
const map = new Map<string, unknown>();
|
||||
if (keys.length === values.length) {
|
||||
for (let i = 0; i < keys.length; i++) {
|
||||
map.set(keys[i], values[i]);
|
||||
}
|
||||
return map;
|
||||
}
|
||||
throw new ApplicationError(
|
||||
'Keys and values do not match, this should not happen and appears to result from some cache corruption.',
|
||||
);
|
||||
}
|
||||
}
|
||||
344
packages/cli/src/services/cache/cache.service.ts
vendored
Normal file
344
packages/cli/src/services/cache/cache.service.ts
vendored
Normal file
|
|
@ -0,0 +1,344 @@
|
|||
import EventEmitter from 'node:events';
|
||||
|
||||
import { Service } from 'typedi';
|
||||
import { caching } from 'cache-manager';
|
||||
import { jsonStringify } from 'n8n-workflow';
|
||||
|
||||
import config from '@/config';
|
||||
import { getDefaultRedisClient, getRedisPrefix } from '@/services/redis/RedisServiceHelper';
|
||||
import { UncacheableValueError } from '@/errors/cache-errors/uncacheable-value.error';
|
||||
import { MalformedRefreshValueError } from '@/errors/cache-errors/malformed-refresh-value.error';
|
||||
import type {
|
||||
TaggedRedisCache,
|
||||
TaggedMemoryCache,
|
||||
CacheEvent,
|
||||
MaybeHash,
|
||||
Hash,
|
||||
} from '@/services/cache/cache.types';
|
||||
|
||||
@Service()
|
||||
export class CacheService extends EventEmitter {
|
||||
private cache: TaggedRedisCache | TaggedMemoryCache;
|
||||
|
||||
async init() {
|
||||
const backend = config.getEnv('cache.backend');
|
||||
const mode = config.getEnv('executions.mode');
|
||||
const ttl = config.getEnv('cache.redis.ttl');
|
||||
|
||||
const useRedis = backend === 'redis' || (backend === 'auto' && mode === 'queue');
|
||||
|
||||
if (useRedis) {
|
||||
const keyPrefix = `${getRedisPrefix()}:${config.getEnv('cache.redis.prefix')}:`;
|
||||
const redisClient = await getDefaultRedisClient({ keyPrefix }, 'client(cache)');
|
||||
|
||||
const { redisStoreUsingClient } = await import('@/services/cache/redis.cache-manager');
|
||||
const redisStore = redisStoreUsingClient(redisClient, { ttl });
|
||||
|
||||
const redisCache = await caching(redisStore);
|
||||
|
||||
this.cache = { ...redisCache, kind: 'redis' };
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
const maxSize = config.getEnv('cache.memory.maxSize');
|
||||
|
||||
const sizeCalculation = (item: unknown) => {
|
||||
const str = jsonStringify(item, { replaceCircularRefs: true });
|
||||
return new TextEncoder().encode(str).length;
|
||||
};
|
||||
|
||||
const memoryCache = await caching('memory', { ttl, maxSize, sizeCalculation });
|
||||
|
||||
this.cache = { ...memoryCache, kind: 'memory' };
|
||||
}
|
||||
|
||||
async reset() {
|
||||
await this.cache.store.reset();
|
||||
}
|
||||
|
||||
emit(event: CacheEvent, ...args: unknown[]) {
|
||||
return super.emit(event, ...args);
|
||||
}
|
||||
|
||||
isRedis() {
|
||||
return this.cache.kind === 'redis';
|
||||
}
|
||||
|
||||
isMemory() {
|
||||
return this.cache.kind === 'memory';
|
||||
}
|
||||
|
||||
// ----------------------------------
|
||||
// storing
|
||||
// ----------------------------------
|
||||
|
||||
async set(key: string, value: unknown, ttl?: number) {
|
||||
if (!this.cache) await this.init();
|
||||
|
||||
if (!key || !value) return;
|
||||
|
||||
if (this.cache.kind === 'redis' && !this.cache.store.isCacheable(value)) {
|
||||
throw new UncacheableValueError(key);
|
||||
}
|
||||
|
||||
await this.cache.store.set(key, value, ttl);
|
||||
}
|
||||
|
||||
async setMany(keysValues: Array<[key: string, value: unknown]>, ttl?: number) {
|
||||
if (!this.cache) await this.init();
|
||||
|
||||
if (keysValues.length === 0) return;
|
||||
|
||||
const truthyKeysValues = keysValues.filter(
|
||||
([key, value]) => key?.length > 0 && value !== undefined && value !== null,
|
||||
);
|
||||
|
||||
if (this.cache.kind === 'redis') {
|
||||
for (const [key, value] of truthyKeysValues) {
|
||||
if (!this.cache.store.isCacheable(value)) {
|
||||
throw new UncacheableValueError(key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
await this.cache.store.mset(truthyKeysValues, ttl);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set or append to a [Redis hash](https://redis.io/docs/data-types/hashes/)
|
||||
* stored under a key in the cache. If in-memory, the hash is a regular JS object.
|
||||
*/
|
||||
async setHash(key: string, hash: Hash) {
|
||||
if (!this.cache) await this.init();
|
||||
|
||||
if (!key?.length) return;
|
||||
|
||||
for (const hashKey in hash) {
|
||||
if (hash[hashKey] === undefined || hash[hashKey] === null) return;
|
||||
}
|
||||
|
||||
if (this.cache.kind === 'redis') {
|
||||
await this.cache.store.hset(key, hash);
|
||||
return;
|
||||
}
|
||||
|
||||
const hashObject: Hash = (await this.get(key)) ?? {};
|
||||
|
||||
Object.assign(hashObject, hash);
|
||||
|
||||
await this.set(key, hashObject);
|
||||
}
|
||||
|
||||
// ----------------------------------
|
||||
// retrieving
|
||||
// ----------------------------------
|
||||
|
||||
/**
|
||||
* Retrieve a primitive value under a key. To retrieve a hash, use `getHash`, and
|
||||
* to retrieve a primitive value in a hash, use `getHashValue`.
|
||||
*/
|
||||
async get<T = unknown>(
|
||||
key: string,
|
||||
{
|
||||
fallbackValue,
|
||||
refreshFn,
|
||||
}: { fallbackValue?: T; refreshFn?: (key: string) => Promise<T | undefined> } = {},
|
||||
) {
|
||||
if (!this.cache) await this.init();
|
||||
|
||||
if (key?.length === 0) return;
|
||||
|
||||
const value = await this.cache.store.get<T>(key);
|
||||
|
||||
if (value !== undefined) {
|
||||
this.emit('metrics.cache.hit');
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
this.emit('metrics.cache.miss');
|
||||
|
||||
if (refreshFn) {
|
||||
this.emit('metrics.cache.update');
|
||||
|
||||
const refreshValue = await refreshFn(key);
|
||||
await this.set(key, refreshValue);
|
||||
|
||||
return refreshValue;
|
||||
}
|
||||
|
||||
return fallbackValue;
|
||||
}
|
||||
|
||||
async getMany<T = unknown[]>(
|
||||
keys: string[],
|
||||
{
|
||||
fallbackValue,
|
||||
refreshFn,
|
||||
}: {
|
||||
fallbackValue?: T[];
|
||||
refreshFn?: (keys: string[]) => Promise<T[]>;
|
||||
} = {},
|
||||
) {
|
||||
if (!this.cache) await this.init();
|
||||
|
||||
if (keys.length === 0) return [];
|
||||
|
||||
const values = await this.cache.store.mget(...keys);
|
||||
|
||||
if (values !== undefined) {
|
||||
this.emit('metrics.cache.hit');
|
||||
|
||||
return values as T[];
|
||||
}
|
||||
|
||||
this.emit('metrics.cache.miss');
|
||||
|
||||
if (refreshFn) {
|
||||
this.emit('metrics.cache.update');
|
||||
|
||||
const refreshValue: T[] = await refreshFn(keys);
|
||||
|
||||
if (keys.length !== refreshValue.length) {
|
||||
throw new MalformedRefreshValueError();
|
||||
}
|
||||
|
||||
const newValue: Array<[key: string, value: unknown]> = keys.map((key, i) => [
|
||||
key,
|
||||
refreshValue[i],
|
||||
]);
|
||||
|
||||
await this.setMany(newValue);
|
||||
|
||||
return refreshValue;
|
||||
}
|
||||
|
||||
return fallbackValue;
|
||||
}
|
||||
|
||||
/**
|
||||
* Retrieve a [Redis hash](https://redis.io/docs/data-types/hashes/) under a key.
|
||||
* If in-memory, the hash is a regular JS object. To retrieve a primitive value
|
||||
* in the hash, use `getHashValue`.
|
||||
*/
|
||||
async getHash<T = unknown>(
|
||||
key: string,
|
||||
{
|
||||
fallbackValue,
|
||||
refreshFn,
|
||||
}: { fallbackValue?: T; refreshFn?: (key: string) => Promise<MaybeHash<T>> } = {},
|
||||
) {
|
||||
if (!this.cache) await this.init();
|
||||
|
||||
const hash: MaybeHash<T> =
|
||||
this.cache.kind === 'redis' ? await this.cache.store.hgetall(key) : await this.get(key);
|
||||
|
||||
if (hash !== undefined) {
|
||||
this.emit('metrics.cache.hit');
|
||||
|
||||
return hash;
|
||||
}
|
||||
|
||||
this.emit('metrics.cache.miss');
|
||||
|
||||
if (refreshFn) {
|
||||
this.emit('metrics.cache.update');
|
||||
|
||||
const refreshValue = await refreshFn(key);
|
||||
await this.set(key, refreshValue);
|
||||
|
||||
return refreshValue;
|
||||
}
|
||||
|
||||
return fallbackValue as MaybeHash<T>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Retrieve a primitive value in a [Redis hash](https://redis.io/docs/data-types/hashes/)
|
||||
* under a hash key. If in-memory, the hash is a regular JS object. To retrieve the hash
|
||||
* itself, use `getHash`.
|
||||
*/
|
||||
async getHashValue<T = unknown>(
|
||||
cacheKey: string,
|
||||
hashKey: string,
|
||||
{
|
||||
fallbackValue,
|
||||
refreshFn,
|
||||
}: { fallbackValue?: T; refreshFn?: (key: string) => Promise<T | undefined> } = {},
|
||||
) {
|
||||
if (!this.cache) await this.init();
|
||||
|
||||
let hashValue: MaybeHash<T>;
|
||||
|
||||
if (this.cache.kind === 'redis') {
|
||||
hashValue = await this.cache.store.hget(cacheKey, hashKey);
|
||||
} else {
|
||||
const hashObject = await this.cache.store.get<Hash<T>>(cacheKey);
|
||||
|
||||
hashValue = hashObject?.[hashKey] as MaybeHash<T>;
|
||||
}
|
||||
|
||||
if (hashValue !== undefined) {
|
||||
this.emit('metrics.cache.hit');
|
||||
|
||||
return hashValue as T;
|
||||
}
|
||||
|
||||
this.emit('metrics.cache.miss');
|
||||
|
||||
if (refreshFn) {
|
||||
this.emit('metrics.cache.update');
|
||||
|
||||
const refreshValue = await refreshFn(cacheKey);
|
||||
await this.set(cacheKey, refreshValue);
|
||||
|
||||
return refreshValue;
|
||||
}
|
||||
|
||||
return fallbackValue;
|
||||
}
|
||||
|
||||
// ----------------------------------
|
||||
// deleting
|
||||
// ----------------------------------
|
||||
|
||||
async delete(key: string) {
|
||||
if (!this.cache) await this.init();
|
||||
|
||||
if (!key?.length) return;
|
||||
|
||||
await this.cache.store.del(key);
|
||||
}
|
||||
|
||||
async deleteMany(keys: string[]) {
|
||||
if (!this.cache) await this.init();
|
||||
|
||||
if (keys.length === 0) return;
|
||||
|
||||
return this.cache.store.mdel(...keys);
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete a value under a key in a [Redis hash](https://redis.io/docs/data-types/hashes/).
|
||||
* If in-memory, the hash is a regular JS object. To delete the hash itself, use `delete`.
|
||||
*/
|
||||
async deleteFromHash(cacheKey: string, hashKey: string) {
|
||||
if (!this.cache) await this.init();
|
||||
|
||||
if (!cacheKey || !hashKey) return;
|
||||
|
||||
if (this.cache.kind === 'redis') {
|
||||
await this.cache.store.hdel(cacheKey, hashKey);
|
||||
return;
|
||||
}
|
||||
|
||||
const hashObject = await this.get<Hash>(cacheKey);
|
||||
|
||||
if (!hashObject) return;
|
||||
|
||||
delete hashObject[hashKey];
|
||||
|
||||
await this.cache.store.set(cacheKey, hashObject);
|
||||
}
|
||||
}
|
||||
12
packages/cli/src/services/cache/cache.types.ts
vendored
Normal file
12
packages/cli/src/services/cache/cache.types.ts
vendored
Normal file
|
|
@ -0,0 +1,12 @@
|
|||
import type { MemoryCache } from 'cache-manager';
|
||||
import type { RedisCache } from '@/services/cache/redis.cache-manager';
|
||||
|
||||
export type TaggedRedisCache = RedisCache & { kind: 'redis' };
|
||||
|
||||
export type TaggedMemoryCache = MemoryCache & { kind: 'memory' };
|
||||
|
||||
export type Hash<T = unknown> = Record<string, T>;
|
||||
|
||||
export type MaybeHash<T> = Hash<T> | undefined;
|
||||
|
||||
export type CacheEvent = `metrics.cache.${'hit' | 'miss' | 'update'}`;
|
||||
170
packages/cli/src/services/cache/redis.cache-manager.ts
vendored
Normal file
170
packages/cli/src/services/cache/redis.cache-manager.ts
vendored
Normal file
|
|
@ -0,0 +1,170 @@
|
|||
/**
|
||||
* Based on https://github.com/node-cache-manager/node-cache-manager-ioredis-yet
|
||||
*/
|
||||
|
||||
import Redis from 'ioredis';
|
||||
import type { Cluster, ClusterNode, ClusterOptions, RedisOptions } from 'ioredis';
|
||||
import type { Cache, Store, Config } from 'cache-manager';
|
||||
import { ApplicationError, jsonParse } from 'n8n-workflow';
|
||||
|
||||
export class NoCacheableError implements Error {
|
||||
name = 'NoCacheableError';
|
||||
|
||||
constructor(public message: string) {}
|
||||
}
|
||||
|
||||
export const avoidNoCacheable = async <T>(p: Promise<T>) => {
|
||||
try {
|
||||
return await p;
|
||||
} catch (e) {
|
||||
if (!(e instanceof NoCacheableError)) throw e;
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
|
||||
export interface RedisClusterConfig {
|
||||
nodes: ClusterNode[];
|
||||
options?: ClusterOptions;
|
||||
}
|
||||
|
||||
export type RedisCache = Cache<RedisStore>;
|
||||
|
||||
export interface RedisStore extends Store {
|
||||
readonly isCacheable: (value: unknown) => boolean;
|
||||
get client(): Redis | Cluster;
|
||||
hget<T>(key: string, field: string): Promise<T | undefined>;
|
||||
hgetall<T>(key: string): Promise<Record<string, T> | undefined>;
|
||||
hset(key: string, fieldValueRecord: Record<string, unknown>): Promise<void>;
|
||||
hkeys(key: string): Promise<string[]>;
|
||||
hvals<T>(key: string): Promise<T[]>;
|
||||
hexists(key: string, field: string): Promise<boolean>;
|
||||
hdel(key: string, field: string): Promise<number>;
|
||||
}
|
||||
|
||||
function builder(
|
||||
redisCache: Redis | Cluster,
|
||||
reset: () => Promise<void>,
|
||||
keys: (pattern: string) => Promise<string[]>,
|
||||
options?: Config,
|
||||
) {
|
||||
const isCacheable = options?.isCacheable ?? ((value) => value !== undefined && value !== null);
|
||||
const getVal = (value: unknown) => JSON.stringify(value) || '"undefined"';
|
||||
|
||||
return {
|
||||
async get<T>(key: string) {
|
||||
const val = await redisCache.get(key);
|
||||
if (val === undefined || val === null) return undefined;
|
||||
else return jsonParse<T>(val);
|
||||
},
|
||||
async set(key, value, ttl) {
|
||||
// eslint-disable-next-line @typescript-eslint/no-throw-literal, @typescript-eslint/restrict-template-expressions
|
||||
if (!isCacheable(value)) throw new NoCacheableError(`"${value}" is not a cacheable value`);
|
||||
const t = ttl ?? options?.ttl;
|
||||
if (t !== undefined && t !== 0) await redisCache.set(key, getVal(value), 'PX', t);
|
||||
else await redisCache.set(key, getVal(value));
|
||||
},
|
||||
async mset(args, ttl) {
|
||||
const t = ttl ?? options?.ttl;
|
||||
if (t !== undefined && t !== 0) {
|
||||
const multi = redisCache.multi();
|
||||
for (const [key, value] of args) {
|
||||
if (!isCacheable(value))
|
||||
// eslint-disable-next-line @typescript-eslint/no-throw-literal
|
||||
throw new NoCacheableError(`"${getVal(value)}" is not a cacheable value`);
|
||||
multi.set(key, getVal(value), 'PX', t);
|
||||
}
|
||||
await multi.exec();
|
||||
} else
|
||||
await redisCache.mset(
|
||||
args.flatMap(([key, value]) => {
|
||||
if (!isCacheable(value))
|
||||
throw new ApplicationError(`"${getVal(value)}" is not a cacheable value`);
|
||||
return [key, getVal(value)] as [string, string];
|
||||
}),
|
||||
);
|
||||
},
|
||||
mget: async (...args) =>
|
||||
redisCache
|
||||
.mget(args)
|
||||
.then((results) =>
|
||||
results.map((result) =>
|
||||
result === null || result === undefined ? undefined : jsonParse(result),
|
||||
),
|
||||
),
|
||||
async mdel(...args) {
|
||||
await redisCache.del(args);
|
||||
},
|
||||
async del(key) {
|
||||
await redisCache.del(key);
|
||||
},
|
||||
ttl: async (key) => redisCache.pttl(key),
|
||||
keys: async (pattern = '*') => keys(pattern),
|
||||
reset,
|
||||
isCacheable,
|
||||
get client() {
|
||||
return redisCache;
|
||||
},
|
||||
// Redis Hash functions
|
||||
async hget<T>(key: string, field: string) {
|
||||
const val = await redisCache.hget(key, field);
|
||||
if (val === undefined || val === null) return undefined;
|
||||
else return jsonParse<T>(val);
|
||||
},
|
||||
async hgetall<T>(key: string) {
|
||||
const val = await redisCache.hgetall(key);
|
||||
if (val === undefined || val === null) return undefined;
|
||||
else {
|
||||
for (const field in val) {
|
||||
const value = val[field];
|
||||
val[field] = jsonParse(value);
|
||||
}
|
||||
return val as Record<string, T>;
|
||||
}
|
||||
},
|
||||
async hset(key: string, fieldValueRecord: Record<string, unknown>) {
|
||||
for (const field in fieldValueRecord) {
|
||||
const value = fieldValueRecord[field];
|
||||
if (!isCacheable(fieldValueRecord[field])) {
|
||||
// eslint-disable-next-line @typescript-eslint/no-throw-literal, @typescript-eslint/restrict-template-expressions
|
||||
throw new NoCacheableError(`"${value}" is not a cacheable value`);
|
||||
}
|
||||
fieldValueRecord[field] = getVal(value);
|
||||
}
|
||||
await redisCache.hset(key, fieldValueRecord);
|
||||
},
|
||||
async hkeys(key: string) {
|
||||
return redisCache.hkeys(key);
|
||||
},
|
||||
async hvals<T>(key: string): Promise<T[]> {
|
||||
const values = await redisCache.hvals(key);
|
||||
return values.map((value) => jsonParse<T>(value));
|
||||
},
|
||||
async hexists(key: string, field: string): Promise<boolean> {
|
||||
return (await redisCache.hexists(key, field)) === 1;
|
||||
},
|
||||
async hdel(key: string, field: string) {
|
||||
return redisCache.hdel(key, field);
|
||||
},
|
||||
} as RedisStore;
|
||||
}
|
||||
|
||||
export function redisStoreUsingClient(redisCache: Redis | Cluster, options?: Config) {
|
||||
const reset = async () => {
|
||||
await redisCache.flushdb();
|
||||
};
|
||||
const keys = async (pattern: string) => redisCache.keys(pattern);
|
||||
|
||||
return builder(redisCache, reset, keys, options);
|
||||
}
|
||||
|
||||
export async function redisStore(
|
||||
options?: (RedisOptions | { clusterConfig: RedisClusterConfig }) & Config,
|
||||
) {
|
||||
options ||= {};
|
||||
const redisCache =
|
||||
'clusterConfig' in options
|
||||
? new Redis.Cluster(options.clusterConfig.nodes, options.clusterConfig.options)
|
||||
: new Redis(options);
|
||||
|
||||
return redisStoreUsingClient(redisCache, options);
|
||||
}
|
||||
|
|
@ -7,7 +7,7 @@ import semverParse from 'semver/functions/parse';
|
|||
import { Service } from 'typedi';
|
||||
import EventEmitter from 'events';
|
||||
|
||||
import { CacheService } from '@/services/cache.service';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import type { EventMessageTypes } from '@/eventbus/EventMessageClasses';
|
||||
import {
|
||||
METRICS_EVENT_NAME,
|
||||
|
|
@ -97,7 +97,7 @@ export class MetricsService extends EventEmitter {
|
|||
labelNames: ['cache'],
|
||||
});
|
||||
this.counters.cacheHitsTotal.inc(0);
|
||||
this.cacheService.on(this.cacheService.metricsCounterEvents.cacheHit, (amount: number = 1) => {
|
||||
this.cacheService.on('metrics.cache.hit', (amount: number = 1) => {
|
||||
this.counters.cacheHitsTotal?.inc(amount);
|
||||
});
|
||||
|
||||
|
|
@ -107,7 +107,7 @@ export class MetricsService extends EventEmitter {
|
|||
labelNames: ['cache'],
|
||||
});
|
||||
this.counters.cacheMissesTotal.inc(0);
|
||||
this.cacheService.on(this.cacheService.metricsCounterEvents.cacheMiss, (amount: number = 1) => {
|
||||
this.cacheService.on('metrics.cache.miss', (amount: number = 1) => {
|
||||
this.counters.cacheMissesTotal?.inc(amount);
|
||||
});
|
||||
|
||||
|
|
@ -117,12 +117,9 @@ export class MetricsService extends EventEmitter {
|
|||
labelNames: ['cache'],
|
||||
});
|
||||
this.counters.cacheUpdatesTotal.inc(0);
|
||||
this.cacheService.on(
|
||||
this.cacheService.metricsCounterEvents.cacheUpdate,
|
||||
(amount: number = 1) => {
|
||||
this.counters.cacheUpdatesTotal?.inc(amount);
|
||||
},
|
||||
);
|
||||
this.cacheService.on('metrics.cache.update', (amount: number = 1) => {
|
||||
this.counters.cacheUpdatesTotal?.inc(amount);
|
||||
});
|
||||
}
|
||||
|
||||
private getCounterForEvent(event: EventMessageTypes): Counter<string> | null {
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import { Service } from 'typedi';
|
||||
import { CacheService } from './cache.service';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import { SharedWorkflowRepository } from '@db/repositories/sharedWorkflow.repository';
|
||||
import type { User } from '@db/entities/User';
|
||||
import { RoleService } from './role.service';
|
||||
|
|
@ -20,7 +20,10 @@ export class OwnershipService {
|
|||
* Retrieve the user who owns the workflow. Note that workflow ownership is **immutable**.
|
||||
*/
|
||||
async getWorkflowOwnerCached(workflowId: string) {
|
||||
const cachedValue = await this.cacheService.get<User>(`cache:workflow-owner:${workflowId}`);
|
||||
const cachedValue = await this.cacheService.getHashValue<User>(
|
||||
'workflow-ownership',
|
||||
workflowId,
|
||||
);
|
||||
|
||||
if (cachedValue) return this.userRepository.create(cachedValue);
|
||||
|
||||
|
|
@ -33,7 +36,7 @@ export class OwnershipService {
|
|||
relations: ['user', 'user.globalRole'],
|
||||
});
|
||||
|
||||
void this.cacheService.set(`cache:workflow-owner:${workflowId}`, sharedWorkflow.user);
|
||||
void this.cacheService.setHash('workflow-ownership', { [workflowId]: sharedWorkflow.user });
|
||||
|
||||
return sharedWorkflow.user;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import { Service } from 'typedi';
|
||||
import { RoleRepository } from '@db/repositories/role.repository';
|
||||
import { SharedWorkflowRepository } from '@db/repositories/sharedWorkflow.repository';
|
||||
import { CacheService } from './cache.service';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import type { RoleNames, RoleScopes } from '@db/entities/Role';
|
||||
import { InvalidRoleError } from '@/errors/invalid-role.error';
|
||||
import { isSharingEnabled } from '@/UserManagement/UserManagementHelper';
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import { Service } from 'typedi';
|
||||
import { CacheService } from './cache.service';
|
||||
import { ApplicationError, type IWebhookData } from 'n8n-workflow';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import { type IWebhookData } from 'n8n-workflow';
|
||||
import type { IWorkflowDb } from '@/Interfaces';
|
||||
|
||||
export type TestWebhookRegistration = {
|
||||
|
|
@ -14,72 +14,51 @@ export type TestWebhookRegistration = {
|
|||
export class TestWebhookRegistrationsService {
|
||||
constructor(private readonly cacheService: CacheService) {}
|
||||
|
||||
private readonly cacheKey = 'test-webhook';
|
||||
private readonly cacheKey = 'test-webhooks';
|
||||
|
||||
async register(registration: TestWebhookRegistration) {
|
||||
const key = this.toKey(registration.webhook);
|
||||
const hashKey = this.toKey(registration.webhook);
|
||||
|
||||
await this.cacheService.set(key, registration);
|
||||
await this.cacheService.setHash(this.cacheKey, { [hashKey]: registration });
|
||||
}
|
||||
|
||||
async deregister(arg: IWebhookData | string) {
|
||||
if (typeof arg === 'string') {
|
||||
await this.cacheService.delete(arg);
|
||||
await this.cacheService.deleteFromHash(this.cacheKey, arg);
|
||||
} else {
|
||||
const key = this.toKey(arg);
|
||||
await this.cacheService.delete(key);
|
||||
const hashKey = this.toKey(arg);
|
||||
await this.cacheService.deleteFromHash(this.cacheKey, hashKey);
|
||||
}
|
||||
}
|
||||
|
||||
async get(key: string) {
|
||||
return this.cacheService.get<TestWebhookRegistration>(key);
|
||||
return this.cacheService.getHashValue<TestWebhookRegistration>(this.cacheKey, key);
|
||||
}
|
||||
|
||||
async getAllKeys() {
|
||||
const keys = await this.cacheService.keys();
|
||||
const hash = await this.cacheService.getHash<TestWebhookRegistration>(this.cacheKey);
|
||||
|
||||
if (this.cacheService.isMemoryCache()) {
|
||||
return keys.filter((key) => key.startsWith(this.cacheKey));
|
||||
}
|
||||
if (!hash) return [];
|
||||
|
||||
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));
|
||||
return Object.keys(hash);
|
||||
}
|
||||
|
||||
async getAllRegistrations() {
|
||||
const keys = await this.getAllKeys();
|
||||
const hash = await this.cacheService.getHash<TestWebhookRegistration>(this.cacheKey);
|
||||
|
||||
return this.cacheService.getMany<TestWebhookRegistration>(keys);
|
||||
}
|
||||
if (!hash) return [];
|
||||
|
||||
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);
|
||||
return Object.values(hash);
|
||||
}
|
||||
|
||||
async deregisterAll() {
|
||||
const testWebhookKeys = await this.getAllKeys();
|
||||
|
||||
await this.cacheService.deleteMany(testWebhookKeys);
|
||||
await this.cacheService.delete(this.cacheKey);
|
||||
}
|
||||
|
||||
toKey(webhook: Pick<IWebhookData, 'webhookId' | 'httpMethod' | 'path'>) {
|
||||
const { webhookId, httpMethod, path: webhookPath } = webhook;
|
||||
|
||||
if (!webhookId) return `${this.cacheKey}:${httpMethod}|${webhookPath}`;
|
||||
if (!webhookId) return [httpMethod, webhookPath].join('|');
|
||||
|
||||
let path = webhookPath;
|
||||
|
||||
|
|
@ -89,6 +68,6 @@ export class TestWebhookRegistrationsService {
|
|||
path = path.slice(cutFromIndex);
|
||||
}
|
||||
|
||||
return `${this.cacheKey}:${httpMethod}|${webhookId}|${path.split('/').length}`;
|
||||
return [httpMethod, webhookId, path.split('/').length].join('|');
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import { WebhookRepository } from '@db/repositories/webhook.repository';
|
||||
import { Service } from 'typedi';
|
||||
import { CacheService } from './cache.service';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import type { WebhookEntity } from '@db/entities/WebhookEntity';
|
||||
import type { IHttpRequestMethods } from 'n8n-workflow';
|
||||
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import config from '@/config';
|
|||
import { Telemetry } from '@/telemetry';
|
||||
import { ExternalSecretsManager } from '@/ExternalSecrets/ExternalSecretsManager.ee';
|
||||
import { BinaryDataService } from 'n8n-core';
|
||||
import { CacheService } from '@/services/cache.service';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import { RedisServicePubSubPublisher } from '@/services/redis/RedisServicePubSubPublisher';
|
||||
import { RedisServicePubSubSubscriber } from '@/services/redis/RedisServicePubSubSubscriber';
|
||||
import { MessageEventBus } from '@/eventbus/MessageEventBus/MessageEventBus';
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import Container from 'typedi';
|
||||
import { mock } from 'jest-mock-extended';
|
||||
import { CacheService } from '@/services/cache.service';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
|
||||
const cacheService = Container.get(CacheService);
|
||||
const store = mock<NonNullable<CacheService['cache']>['store']>({ isCacheable: () => true });
|
||||
|
|
|
|||
|
|
@ -1,355 +1,240 @@
|
|||
import Container from 'typedi';
|
||||
import { CacheService } from '@/services/cache.service';
|
||||
import type { MemoryCache } from 'cache-manager';
|
||||
import type { RedisCache } from 'cache-manager-ioredis-yet';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import config from '@/config';
|
||||
import { sleep } from 'n8n-workflow';
|
||||
|
||||
const cacheService = Container.get(CacheService);
|
||||
jest.mock('ioredis', () => {
|
||||
const Redis = require('ioredis-mock');
|
||||
|
||||
function setDefaultConfig() {
|
||||
config.set('executions.mode', 'regular');
|
||||
config.set('cache.enabled', true);
|
||||
config.set('cache.backend', 'memory');
|
||||
config.set('cache.memory.maxSize', 1 * 1024 * 1024);
|
||||
}
|
||||
return function (...args: unknown[]) {
|
||||
return new Redis(args);
|
||||
};
|
||||
});
|
||||
|
||||
interface TestObject {
|
||||
test: string;
|
||||
test2: number;
|
||||
test3?: TestObject & { test4: TestObject };
|
||||
}
|
||||
for (const backend of ['memory', 'redis'] as const) {
|
||||
describe(backend, () => {
|
||||
let cacheService: CacheService;
|
||||
|
||||
const testObject: TestObject = {
|
||||
test: 'test',
|
||||
test2: 123,
|
||||
test3: {
|
||||
test: 'test3',
|
||||
test2: 123,
|
||||
test4: {
|
||||
test: 'test4',
|
||||
test2: 123,
|
||||
},
|
||||
},
|
||||
};
|
||||
beforeAll(async () => {
|
||||
config.set('cache.backend', backend);
|
||||
cacheService = new CacheService();
|
||||
await cacheService.init();
|
||||
});
|
||||
|
||||
describe('cacheService', () => {
|
||||
beforeAll(async () => {
|
||||
jest.mock('ioredis', () => {
|
||||
const Redis = require('ioredis-mock');
|
||||
if (typeof Redis === 'object') {
|
||||
// the first mock is an ioredis shim because ioredis-mock depends on it
|
||||
// https://github.com/stipsan/ioredis-mock/blob/master/src/index.js#L101-L111
|
||||
return {
|
||||
Command: { _transformer: { argument: {}, reply: {} } },
|
||||
};
|
||||
afterEach(async () => {
|
||||
await cacheService.reset();
|
||||
config.load(config.default);
|
||||
});
|
||||
|
||||
describe('init', () => {
|
||||
test('should select backend based on config', () => {
|
||||
expect(cacheService.isMemory()).toBe(backend === 'memory');
|
||||
expect(cacheService.isRedis()).toBe(backend === 'redis');
|
||||
});
|
||||
|
||||
if (backend === 'redis') {
|
||||
test('with auto backend and queue mode, should select redis', async () => {
|
||||
config.set('executions.mode', 'queue');
|
||||
|
||||
await cacheService.init();
|
||||
|
||||
expect(cacheService.isRedis()).toBe(true);
|
||||
});
|
||||
}
|
||||
// second mock for our code
|
||||
return function (...args: any) {
|
||||
return new Redis(args);
|
||||
};
|
||||
|
||||
if (backend === 'memory') {
|
||||
test('should honor max size when enough', async () => {
|
||||
config.set('cache.memory.maxSize', 16); // enough bytes for "withoutUnicode"
|
||||
|
||||
await cacheService.init();
|
||||
await cacheService.set('key', 'withoutUnicode');
|
||||
|
||||
await expect(cacheService.get('key')).resolves.toBe('withoutUnicode');
|
||||
|
||||
// restore
|
||||
config.set('cache.memory.maxSize', 3 * 1024 * 1024);
|
||||
await cacheService.init();
|
||||
});
|
||||
|
||||
test('should honor max size when not enough', async () => {
|
||||
config.set('cache.memory.maxSize', 16); // not enough bytes for "withUnicodeԱԲԳ"
|
||||
|
||||
await cacheService.init();
|
||||
await cacheService.set('key', 'withUnicodeԱԲԳ');
|
||||
|
||||
await expect(cacheService.get('key')).resolves.toBeUndefined();
|
||||
|
||||
// restore
|
||||
config.set('cache.memory.maxSize', 3 * 1024 * 1024);
|
||||
await cacheService.init();
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
describe('set', () => {
|
||||
test('should set a string value', async () => {
|
||||
await cacheService.set('key', 'value');
|
||||
|
||||
await expect(cacheService.get('key')).resolves.toBe('value');
|
||||
});
|
||||
|
||||
test('should set a number value', async () => {
|
||||
await cacheService.set('key', 123);
|
||||
|
||||
await expect(cacheService.get('key')).resolves.toBe(123);
|
||||
});
|
||||
|
||||
test('should set an object value', async () => {
|
||||
const object = { a: { b: { c: { d: 1 } } } };
|
||||
|
||||
await cacheService.set('key', object);
|
||||
|
||||
await expect(cacheService.get('key')).resolves.toMatchObject(object);
|
||||
});
|
||||
|
||||
test('should not cache `null` or `undefined` values', async () => {
|
||||
await cacheService.set('key1', null);
|
||||
await cacheService.set('key2', undefined);
|
||||
await cacheService.set('key3', 'value');
|
||||
|
||||
await expect(cacheService.get('key1')).resolves.toBeUndefined();
|
||||
await expect(cacheService.get('key2')).resolves.toBeUndefined();
|
||||
await expect(cacheService.get('key3')).resolves.toBe('value');
|
||||
});
|
||||
|
||||
test('should disregard zero-length keys', async () => {
|
||||
await cacheService.set('', 'value');
|
||||
|
||||
await expect(cacheService.get('')).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
test('should honor ttl', async () => {
|
||||
await cacheService.set('key', 'value', 100);
|
||||
|
||||
await expect(cacheService.get('key')).resolves.toBe('value');
|
||||
|
||||
await sleep(200);
|
||||
|
||||
await expect(cacheService.get('key')).resolves.toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('get', () => {
|
||||
test('should fall back to fallback value', async () => {
|
||||
const promise = cacheService.get('key', { fallbackValue: 'fallback' });
|
||||
await expect(promise).resolves.toBe('fallback');
|
||||
});
|
||||
|
||||
test('should refresh value', async () => {
|
||||
const promise = cacheService.get('testString', {
|
||||
refreshFn: async () => 'refreshValue',
|
||||
});
|
||||
|
||||
await expect(promise).resolves.toBe('refreshValue');
|
||||
});
|
||||
|
||||
test('should handle non-ASCII key', async () => {
|
||||
const nonAsciiKey = 'ԱԲԳ';
|
||||
await cacheService.set(nonAsciiKey, 'value');
|
||||
|
||||
await expect(cacheService.get(nonAsciiKey)).resolves.toBe('value');
|
||||
});
|
||||
});
|
||||
|
||||
describe('delete', () => {
|
||||
test('should delete a key', async () => {
|
||||
await cacheService.set('key', 'value');
|
||||
|
||||
await cacheService.delete('key');
|
||||
|
||||
await expect(cacheService.get('key')).resolves.toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('setMany', () => {
|
||||
test('should set multiple string values', async () => {
|
||||
await cacheService.setMany([
|
||||
['key1', 'value1'],
|
||||
['key2', 'value2'],
|
||||
]);
|
||||
|
||||
const promise = cacheService.getMany(['key1', 'key2']);
|
||||
await expect(promise).resolves.toStrictEqual(['value1', 'value2']);
|
||||
});
|
||||
|
||||
test('should set multiple number values', async () => {
|
||||
await cacheService.setMany([
|
||||
['key1', 123],
|
||||
['key2', 456],
|
||||
]);
|
||||
|
||||
const promise = cacheService.getMany(['key1', 'key2']);
|
||||
await expect(promise).resolves.toStrictEqual([123, 456]);
|
||||
});
|
||||
|
||||
test('should disregard zero-length keys', async () => {
|
||||
await cacheService.setMany([['', 'value1']]);
|
||||
|
||||
await expect(cacheService.get('')).resolves.toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('getMany', () => {
|
||||
test('should return undefined on missing result', async () => {
|
||||
await cacheService.setMany([
|
||||
['key1', 123],
|
||||
['key2', 456],
|
||||
]);
|
||||
|
||||
const promise = cacheService.getMany(['key2', 'key3']);
|
||||
await expect(promise).resolves.toStrictEqual([456, undefined]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('delete', () => {
|
||||
test('should handle non-ASCII key', async () => {
|
||||
const nonAsciiKey = 'ԱԲԳ';
|
||||
await cacheService.set(nonAsciiKey, 'value');
|
||||
await expect(cacheService.get(nonAsciiKey)).resolves.toBe('value');
|
||||
|
||||
await cacheService.delete(nonAsciiKey);
|
||||
|
||||
await expect(cacheService.get(nonAsciiKey)).resolves.toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('setHash', () => {
|
||||
test('should set a hash if non-existing', async () => {
|
||||
await cacheService.setHash('keyW', { field: 'value' });
|
||||
|
||||
await expect(cacheService.getHash('keyW')).resolves.toStrictEqual({ field: 'value' });
|
||||
});
|
||||
|
||||
test('should add to a hash value if existing', async () => {
|
||||
await cacheService.setHash('key', { field1: 'value1' });
|
||||
await cacheService.setHash('key', { field2: 'value2' });
|
||||
|
||||
await expect(cacheService.getHash('key')).resolves.toStrictEqual({
|
||||
field1: 'value1',
|
||||
field2: 'value2',
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('deleteFromHash', () => {
|
||||
test('should delete a hash field', async () => {
|
||||
await cacheService.setHash('key', { field1: 'value1', field2: 'value2' });
|
||||
await cacheService.deleteFromHash('key', 'field1');
|
||||
|
||||
await expect(cacheService.getHash('key')).resolves.toStrictEqual({ field2: 'value2' });
|
||||
});
|
||||
});
|
||||
|
||||
describe('getHashValue', () => {
|
||||
test('should return a hash field value', async () => {
|
||||
await cacheService.setHash('key', { field1: 'value1', field2: 'value2' });
|
||||
|
||||
await expect(cacheService.getHashValue('key', 'field1')).resolves.toBe('value1');
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
setDefaultConfig();
|
||||
await Container.get(CacheService).destroy();
|
||||
});
|
||||
|
||||
test('should create a memory cache by default', async () => {
|
||||
await cacheService.init();
|
||||
await expect(cacheService.getCache()).resolves.toBeDefined();
|
||||
const candidate = (await cacheService.getCache()) as MemoryCache;
|
||||
// type guard to check that a MemoryCache is returned and not a RedisCache (which does not have a size property)
|
||||
expect(candidate.store.size).toBeDefined();
|
||||
});
|
||||
|
||||
test('should cache and retrieve a value', async () => {
|
||||
await cacheService.init();
|
||||
await expect(cacheService.getCache()).resolves.toBeDefined();
|
||||
await cacheService.set('testString', 'test');
|
||||
await cacheService.set('testNumber1', 123);
|
||||
|
||||
await expect(cacheService.get('testString')).resolves.toBe('test');
|
||||
expect(typeof (await cacheService.get('testString'))).toBe('string');
|
||||
await expect(cacheService.get('testNumber1')).resolves.toBe(123);
|
||||
expect(typeof (await cacheService.get('testNumber1'))).toBe('number');
|
||||
});
|
||||
|
||||
test('should honour ttl values', async () => {
|
||||
await cacheService.set('testString', 'test', 10);
|
||||
await cacheService.set('testNumber1', 123, 1000);
|
||||
|
||||
const store = (await cacheService.getCache())?.store;
|
||||
|
||||
expect(store).toBeDefined();
|
||||
|
||||
await expect(store!.ttl('testString')).resolves.toBeLessThanOrEqual(100);
|
||||
await expect(store!.ttl('testNumber1')).resolves.toBeLessThanOrEqual(1000);
|
||||
});
|
||||
|
||||
test('should set and remove values', async () => {
|
||||
await cacheService.set('testString', 'test');
|
||||
await expect(cacheService.get('testString')).resolves.toBe('test');
|
||||
await cacheService.delete('testString');
|
||||
await expect(cacheService.get('testString')).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
test('should calculate maxSize', async () => {
|
||||
config.set('cache.memory.maxSize', 16);
|
||||
await cacheService.destroy();
|
||||
|
||||
// 16 bytes because stringify wraps the string in quotes, so 2 bytes for the quotes
|
||||
await cacheService.set('testString', 'withoutUnicode');
|
||||
await expect(cacheService.get('testString')).resolves.toBe('withoutUnicode');
|
||||
|
||||
await cacheService.destroy();
|
||||
|
||||
// should not fit!
|
||||
await cacheService.set('testString', 'withUnicodeԱԲԳ');
|
||||
await expect(cacheService.get('testString')).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
test('should set and get complex objects', async () => {
|
||||
await cacheService.set('testObject', testObject);
|
||||
await expect(cacheService.get('testObject')).resolves.toMatchObject(testObject);
|
||||
});
|
||||
|
||||
test('should set and get multiple values', async () => {
|
||||
await cacheService.destroy();
|
||||
expect(cacheService.isRedisCache()).toBe(false);
|
||||
|
||||
await cacheService.setMany([
|
||||
['testString', 'test'],
|
||||
['testString2', 'test2'],
|
||||
]);
|
||||
await cacheService.setMany([
|
||||
['testNumber1', 123],
|
||||
['testNumber2', 456],
|
||||
]);
|
||||
await expect(cacheService.getMany(['testString', 'testString2'])).resolves.toStrictEqual([
|
||||
'test',
|
||||
'test2',
|
||||
]);
|
||||
await expect(cacheService.getMany(['testNumber1', 'testNumber2'])).resolves.toStrictEqual([
|
||||
123, 456,
|
||||
]);
|
||||
});
|
||||
|
||||
test('should create a redis in queue mode', async () => {
|
||||
config.set('cache.backend', 'auto');
|
||||
config.set('executions.mode', 'queue');
|
||||
await cacheService.destroy();
|
||||
await cacheService.init();
|
||||
|
||||
const cache = await cacheService.getCache();
|
||||
await expect(cacheService.getCache()).resolves.toBeDefined();
|
||||
const candidate = (await cacheService.getCache()) as RedisCache;
|
||||
expect(candidate.store.client).toBeDefined();
|
||||
});
|
||||
|
||||
test('should create a redis cache if asked', async () => {
|
||||
config.set('cache.backend', 'redis');
|
||||
config.set('executions.mode', 'queue');
|
||||
await cacheService.destroy();
|
||||
await cacheService.init();
|
||||
|
||||
const cache = await cacheService.getCache();
|
||||
await expect(cacheService.getCache()).resolves.toBeDefined();
|
||||
const candidate = (await cacheService.getCache()) as RedisCache;
|
||||
expect(candidate.store.client).toBeDefined();
|
||||
});
|
||||
|
||||
test('should get/set/delete redis cache', async () => {
|
||||
config.set('cache.backend', 'redis');
|
||||
config.set('executions.mode', 'queue');
|
||||
await cacheService.destroy();
|
||||
await cacheService.init();
|
||||
|
||||
await cacheService.set('testObject', testObject);
|
||||
await expect(cacheService.get('testObject')).resolves.toMatchObject(testObject);
|
||||
await cacheService.delete('testObject');
|
||||
await expect(cacheService.get('testObject')).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
// NOTE: mset and mget are not supported by ioredis-mock
|
||||
// test('should set and get multiple values with redis', async () => {
|
||||
// });
|
||||
|
||||
test('should return fallback value if key is not set', async () => {
|
||||
await cacheService.reset();
|
||||
await expect(cacheService.get('testString')).resolves.toBeUndefined();
|
||||
await expect(
|
||||
cacheService.get('testString', {
|
||||
fallbackValue: 'fallback',
|
||||
}),
|
||||
).resolves.toBe('fallback');
|
||||
});
|
||||
|
||||
test('should call refreshFunction if key is not set', async () => {
|
||||
await cacheService.reset();
|
||||
await expect(cacheService.get('testString')).resolves.toBeUndefined();
|
||||
await expect(
|
||||
cacheService.get('testString', {
|
||||
refreshFunction: async () => 'refreshed',
|
||||
fallbackValue: 'this should not be returned',
|
||||
}),
|
||||
).resolves.toBe('refreshed');
|
||||
});
|
||||
|
||||
test('should transparently handle disabled cache', async () => {
|
||||
await cacheService.disable();
|
||||
await expect(cacheService.get('testString')).resolves.toBeUndefined();
|
||||
await cacheService.set('testString', 'whatever');
|
||||
await expect(cacheService.get('testString')).resolves.toBeUndefined();
|
||||
await expect(
|
||||
cacheService.get('testString', {
|
||||
fallbackValue: 'fallback',
|
||||
}),
|
||||
).resolves.toBe('fallback');
|
||||
await expect(
|
||||
cacheService.get('testString', {
|
||||
refreshFunction: async () => 'refreshed',
|
||||
fallbackValue: 'this should not be returned',
|
||||
}),
|
||||
).resolves.toBe('refreshed');
|
||||
});
|
||||
|
||||
test('should set and get partial results', async () => {
|
||||
await cacheService.setMany([
|
||||
['testNumber1', 123],
|
||||
['testNumber2', 456],
|
||||
]);
|
||||
await expect(cacheService.getMany(['testNumber1', 'testNumber2'])).resolves.toStrictEqual([
|
||||
123, 456,
|
||||
]);
|
||||
await expect(cacheService.getMany(['testNumber3', 'testNumber2'])).resolves.toStrictEqual([
|
||||
undefined,
|
||||
456,
|
||||
]);
|
||||
});
|
||||
|
||||
test('should getMany and fix partial results and set single key', async () => {
|
||||
await cacheService.setMany([
|
||||
['testNumber1', 123],
|
||||
['testNumber2', 456],
|
||||
]);
|
||||
await expect(
|
||||
cacheService.getMany(['testNumber1', 'testNumber2', 'testNumber3']),
|
||||
).resolves.toStrictEqual([123, 456, undefined]);
|
||||
await expect(cacheService.get('testNumber3')).resolves.toBeUndefined();
|
||||
await expect(
|
||||
cacheService.getMany(['testNumber1', 'testNumber2', 'testNumber3'], {
|
||||
async refreshFunctionEach(key) {
|
||||
return key === 'testNumber3' ? 789 : undefined;
|
||||
},
|
||||
}),
|
||||
).resolves.toStrictEqual([123, 456, 789]);
|
||||
await expect(cacheService.get('testNumber3')).resolves.toBe(789);
|
||||
});
|
||||
|
||||
test('should getMany and set all keys', async () => {
|
||||
await cacheService.setMany([
|
||||
['testNumber1', 123],
|
||||
['testNumber2', 456],
|
||||
]);
|
||||
await expect(
|
||||
cacheService.getMany(['testNumber1', 'testNumber2', 'testNumber3']),
|
||||
).resolves.toStrictEqual([123, 456, undefined]);
|
||||
await expect(cacheService.get('testNumber3')).resolves.toBeUndefined();
|
||||
await expect(
|
||||
cacheService.getMany(['testNumber1', 'testNumber2', 'testNumber3'], {
|
||||
async refreshFunctionMany(keys) {
|
||||
return [111, 222, 333];
|
||||
},
|
||||
}),
|
||||
).resolves.toStrictEqual([111, 222, 333]);
|
||||
await expect(cacheService.get('testNumber1')).resolves.toBe(111);
|
||||
await expect(cacheService.get('testNumber2')).resolves.toBe(222);
|
||||
await expect(cacheService.get('testNumber3')).resolves.toBe(333);
|
||||
});
|
||||
|
||||
test('should set and get multiple values with fallbackValue', async () => {
|
||||
await cacheService.disable();
|
||||
await cacheService.setMany([
|
||||
['testNumber1', 123],
|
||||
['testNumber2', 456],
|
||||
]);
|
||||
await expect(cacheService.getMany(['testNumber1', 'testNumber2'])).resolves.toStrictEqual([
|
||||
undefined,
|
||||
undefined,
|
||||
]);
|
||||
await expect(
|
||||
cacheService.getMany(['testNumber1', 'testNumber2'], {
|
||||
fallbackValues: [123, 456],
|
||||
}),
|
||||
).resolves.toStrictEqual([123, 456]);
|
||||
await expect(
|
||||
cacheService.getMany(['testNumber1', 'testNumber2'], {
|
||||
refreshFunctionMany: async () => [123, 456],
|
||||
fallbackValues: [0, 1],
|
||||
}),
|
||||
).resolves.toStrictEqual([123, 456]);
|
||||
});
|
||||
|
||||
test('should deal with unicode keys', async () => {
|
||||
const key = '? > ":< ! withUnicodeԱԲԳ';
|
||||
await cacheService.set(key, 'test');
|
||||
await expect(cacheService.get(key)).resolves.toBe('test');
|
||||
await cacheService.delete(key);
|
||||
await expect(cacheService.get(key)).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
test('should deal with unicode keys in redis', async () => {
|
||||
config.set('cache.backend', 'redis');
|
||||
config.set('executions.mode', 'queue');
|
||||
await cacheService.destroy();
|
||||
await cacheService.init();
|
||||
const key = '? > ":< ! withUnicodeԱԲԳ';
|
||||
|
||||
expect(((await cacheService.getCache()) as RedisCache).store.client).toBeDefined();
|
||||
|
||||
await cacheService.set(key, 'test');
|
||||
await expect(cacheService.get(key)).resolves.toBe('test');
|
||||
await cacheService.delete(key);
|
||||
await expect(cacheService.get(key)).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
test('should not cache null or undefined values', async () => {
|
||||
await cacheService.set('nullValue', null);
|
||||
await cacheService.set('undefValue', undefined);
|
||||
await cacheService.set('normalValue', 'test');
|
||||
|
||||
await expect(cacheService.get('normalValue')).resolves.toBe('test');
|
||||
await expect(cacheService.get('undefValue')).resolves.toBeUndefined();
|
||||
await expect(cacheService.get('nullValue')).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
test('should handle setting empty keys', async () => {
|
||||
await cacheService.set('', null);
|
||||
await expect(cacheService.get('')).resolves.toBeUndefined();
|
||||
await cacheService.setMany([
|
||||
['', 'something'],
|
||||
['', 'something'],
|
||||
]);
|
||||
await expect(cacheService.getMany([''])).resolves.toStrictEqual([undefined]);
|
||||
await cacheService.setMany([]);
|
||||
await expect(cacheService.getMany([])).resolves.toStrictEqual([]);
|
||||
});
|
||||
|
||||
test('should handle setting empty keys (redis)', async () => {
|
||||
config.set('cache.backend', 'redis');
|
||||
config.set('executions.mode', 'queue');
|
||||
await cacheService.destroy();
|
||||
await cacheService.init();
|
||||
|
||||
await cacheService.set('', null);
|
||||
await expect(cacheService.get('')).resolves.toBeUndefined();
|
||||
await cacheService.setMany([
|
||||
['', 'something'],
|
||||
['', 'something'],
|
||||
]);
|
||||
await expect(cacheService.getMany([''])).resolves.toStrictEqual([undefined]);
|
||||
await cacheService.setMany([]);
|
||||
await expect(cacheService.getMany([])).resolves.toStrictEqual([]);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ import type { RoleNames, RoleScopes } from '@db/entities/Role';
|
|||
import { Role } from '@db/entities/Role';
|
||||
import { RoleService } from '@/services/role.service';
|
||||
import { RoleRepository } from '@db/repositories/role.repository';
|
||||
import { CacheService } from '@/services/cache.service';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import { SharedWorkflow } from '@db/entities/SharedWorkflow';
|
||||
import { mockInstance } from '../../shared/mocking';
|
||||
import { chooseRandomly } from '../../integration/shared/random';
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import type { CacheService } from '@/services/cache.service';
|
||||
import type { CacheService } from '@/services/cache/cache.service';
|
||||
import type { TestWebhookRegistration } from '@/services/test-webhook-registrations.service';
|
||||
import { TestWebhookRegistrationsService } from '@/services/test-webhook-registrations.service';
|
||||
import { mock } from 'jest-mock-extended';
|
||||
|
|
@ -11,14 +11,14 @@ describe('TestWebhookRegistrationsService', () => {
|
|||
webhook: { httpMethod: 'GET', path: 'hello', webhookId: undefined },
|
||||
});
|
||||
|
||||
const key = 'test-webhook:GET|hello';
|
||||
const fullCacheKey = `n8n:cache:${key}`;
|
||||
const webhookKey = 'GET|hello';
|
||||
const cacheKey = 'test-webhooks';
|
||||
|
||||
describe('register()', () => {
|
||||
test('should register a test webhook registration', async () => {
|
||||
await registrations.register(registration);
|
||||
|
||||
expect(cacheService.set).toHaveBeenCalledWith(key, registration);
|
||||
expect(cacheService.setHash).toHaveBeenCalledWith(cacheKey, { [webhookKey]: registration });
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -26,25 +26,25 @@ describe('TestWebhookRegistrationsService', () => {
|
|||
test('should deregister a test webhook registration', async () => {
|
||||
await registrations.register(registration);
|
||||
|
||||
await registrations.deregister(key);
|
||||
await registrations.deregister(webhookKey);
|
||||
|
||||
expect(cacheService.delete).toHaveBeenCalledWith(key);
|
||||
expect(cacheService.deleteFromHash).toHaveBeenCalledWith(cacheKey, webhookKey);
|
||||
});
|
||||
});
|
||||
|
||||
describe('get()', () => {
|
||||
test('should retrieve a test webhook registration', async () => {
|
||||
cacheService.get.mockResolvedValueOnce(registration);
|
||||
cacheService.getHashValue.mockResolvedValueOnce(registration);
|
||||
|
||||
const promise = registrations.get(key);
|
||||
const promise = registrations.get(webhookKey);
|
||||
|
||||
await expect(promise).resolves.toBe(registration);
|
||||
});
|
||||
|
||||
test('should return undefined if no such test webhook registration was found', async () => {
|
||||
cacheService.get.mockResolvedValueOnce(undefined);
|
||||
cacheService.getHashValue.mockResolvedValueOnce(undefined);
|
||||
|
||||
const promise = registrations.get(key);
|
||||
const promise = registrations.get(webhookKey);
|
||||
|
||||
await expect(promise).resolves.toBeUndefined();
|
||||
});
|
||||
|
|
@ -52,18 +52,17 @@ describe('TestWebhookRegistrationsService', () => {
|
|||
|
||||
describe('getAllKeys()', () => {
|
||||
test('should retrieve all test webhook registration keys', async () => {
|
||||
cacheService.keys.mockResolvedValueOnce([fullCacheKey]);
|
||||
cacheService.getHash.mockResolvedValueOnce({ [webhookKey]: registration });
|
||||
|
||||
const result = await registrations.getAllKeys();
|
||||
|
||||
expect(result).toEqual([key]);
|
||||
expect(result).toEqual([webhookKey]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('getAllRegistrations()', () => {
|
||||
test('should retrieve all test webhook registrations', async () => {
|
||||
cacheService.keys.mockResolvedValueOnce([fullCacheKey]);
|
||||
cacheService.getMany.mockResolvedValueOnce([registration]);
|
||||
cacheService.getHash.mockResolvedValueOnce({ [webhookKey]: registration });
|
||||
|
||||
const result = await registrations.getAllRegistrations();
|
||||
|
||||
|
|
@ -71,29 +70,11 @@ describe('TestWebhookRegistrationsService', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('updateWebhookProperties()', () => {
|
||||
test('should update the properties of a test webhook registration', async () => {
|
||||
cacheService.get.mockResolvedValueOnce(registration);
|
||||
|
||||
const newProperties = { ...registration.webhook, isTest: true };
|
||||
|
||||
await registrations.updateWebhookProperties(newProperties);
|
||||
|
||||
registration.webhook = newProperties;
|
||||
|
||||
expect(cacheService.set).toHaveBeenCalledWith(key, registration);
|
||||
|
||||
delete registration.webhook.isTest;
|
||||
});
|
||||
});
|
||||
|
||||
describe('deregisterAll()', () => {
|
||||
test('should deregister all test webhook registrations', async () => {
|
||||
cacheService.keys.mockResolvedValueOnce([fullCacheKey]);
|
||||
|
||||
await registrations.deregisterAll();
|
||||
|
||||
expect(cacheService.delete).toHaveBeenCalledWith(key);
|
||||
expect(cacheService.delete).toHaveBeenCalledWith(cacheKey);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -101,7 +82,7 @@ describe('TestWebhookRegistrationsService', () => {
|
|||
test('should convert a test webhook registration to a key', () => {
|
||||
const result = registrations.toKey(registration.webhook);
|
||||
|
||||
expect(result).toBe(key);
|
||||
expect(result).toBe(webhookKey);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import { v4 as uuid } from 'uuid';
|
||||
import config from '@/config';
|
||||
import { WebhookRepository } from '@db/repositories/webhook.repository';
|
||||
import { CacheService } from '@/services/cache.service';
|
||||
import { CacheService } from '@/services/cache/cache.service';
|
||||
import { WebhookService } from '@/services/webhook.service';
|
||||
import { WebhookEntity } from '@db/entities/WebhookEntity';
|
||||
import { mockInstance } from '../../shared/mocking';
|
||||
|
|
|
|||
|
|
@ -393,9 +393,6 @@ importers:
|
|||
cache-manager:
|
||||
specifier: 5.2.3
|
||||
version: 5.2.3
|
||||
cache-manager-ioredis-yet:
|
||||
specifier: 1.2.2
|
||||
version: 1.2.2
|
||||
callsites:
|
||||
specifier: 3.1.0
|
||||
version: 3.1.0
|
||||
|
|
@ -12615,16 +12612,6 @@ packages:
|
|||
unset-value: 1.0.0
|
||||
dev: true
|
||||
|
||||
/cache-manager-ioredis-yet@1.2.2:
|
||||
resolution: {integrity: sha512-o03N/tQxfFONZ1XLGgIxOFHuQQpjpRdnSAL1THG1YWZIVp1JMUfjU3ElSAjFN1LjbJXa55IpC8waG+VEoLUCUw==}
|
||||
engines: {node: '>= 16.17.0'}
|
||||
dependencies:
|
||||
cache-manager: 5.2.3
|
||||
ioredis: 5.3.2
|
||||
transitivePeerDependencies:
|
||||
- supports-color
|
||||
dev: false
|
||||
|
||||
/cache-manager@5.2.3:
|
||||
resolution: {integrity: sha512-9OErI8fksFkxAMJ8Mco0aiZSdphyd90HcKiOMJQncSlU1yq/9lHHxrT8PDayxrmr9IIIZPOAEfXuGSD7g29uog==}
|
||||
dependencies:
|
||||
|
|
@ -17013,23 +17000,6 @@ packages:
|
|||
transitivePeerDependencies:
|
||||
- supports-color
|
||||
|
||||
/ioredis@5.3.2:
|
||||
resolution: {integrity: sha512-1DKMMzlIHM02eBBVOFQ1+AolGjs6+xEcM4PDL7NqOS6szq7H9jSaEkIUH6/a5Hl241LzW6JLSiAbNvTQjUupUA==}
|
||||
engines: {node: '>=12.22.0'}
|
||||
dependencies:
|
||||
'@ioredis/commands': 1.2.0
|
||||
cluster-key-slot: 1.1.1
|
||||
debug: 4.3.4(supports-color@8.1.1)
|
||||
denque: 2.1.0
|
||||
lodash.defaults: 4.2.0
|
||||
lodash.isarguments: 3.1.0
|
||||
redis-errors: 1.2.0
|
||||
redis-parser: 3.0.0
|
||||
standard-as-callback: 2.1.0
|
||||
transitivePeerDependencies:
|
||||
- supports-color
|
||||
dev: false
|
||||
|
||||
/ip@2.0.0:
|
||||
resolution: {integrity: sha512-WKa+XuLG1A1R0UWhl2+1XQSi+fZWMsYKffMZTTYsiZaUD8k2yDAj5atimTUD2TZkyCkNEeYE5NhFZmupOGtjYQ==}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue