n8n/packages/cli/test/unit/services/redis.service.test.ts

import Container from 'typedi';
import { Logger } from '@/Logger';
import config from '@/config';
import { RedisService } from '@/services/redis.service';
import { mockInstance } from '../../shared/mocking';

mockInstance(Logger);
const redisService = Container.get(RedisService);

function setDefaultConfig() {
	config.set('executions.mode', 'queue');
}

const PUBSUB_CHANNEL = 'testchannel';
const LIST_CHANNEL = 'testlist';
const STREAM_CHANNEL = 'teststream';

describe('RedisService', () => {
	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: {} } },
				};
			}
			// second mock for our code
			return function (...args: any) {
				return new Redis(args);
			};
		});
		setDefaultConfig();
	});

	test('should create pubsub publisher and subscriber with handler', async () => {
		const pub = await redisService.getPubSubPublisher();
		const sub = await redisService.getPubSubSubscriber();
		expect(pub).toBeDefined();
		expect(sub).toBeDefined();

		const mockHandler = jest.fn();
		mockHandler.mockImplementation((channel: string, message: string) => {});
		sub.addMessageHandler(PUBSUB_CHANNEL, mockHandler);
		await sub.subscribe(PUBSUB_CHANNEL);
		await pub.publish(PUBSUB_CHANNEL, 'test');
		await new Promise((resolve) =>
			setTimeout(async () => {
				resolve(0);
			}, 50),
		);
		expect(mockHandler).toHaveBeenCalled();
		await sub.destroy();
		await pub.destroy();
	});

	test('should create list sender and receiver', async () => {
		const sender = await redisService.getListSender();
		const receiver = await redisService.getListReceiver();
		expect(sender).toBeDefined();
		expect(receiver).toBeDefined();
		await sender.prepend(LIST_CHANNEL, 'middle');
		await sender.prepend(LIST_CHANNEL, 'first');
		await sender.append(LIST_CHANNEL, 'end');
		let popResult = await receiver.popFromHead(LIST_CHANNEL);
		expect(popResult).toBe('first');
		popResult = await receiver.popFromTail(LIST_CHANNEL);
		expect(popResult).toBe('end');
		await sender.prepend(LIST_CHANNEL, 'somevalue');
		popResult = await receiver.popFromTail(LIST_CHANNEL);
		expect(popResult).toBe('middle');
		await sender.destroy();
		await receiver.destroy();
	});

	// NOTE: This test is failing because the mock Redis client does not support streams apparently
	// eslint-disable-next-line n8n-local-rules/no-skipped-tests
	test.skip('should create stream producer and consumer', async () => {
		const consumer = await redisService.getStreamConsumer();
		const producer = await redisService.getStreamProducer();

		expect(consumer).toBeDefined();
		expect(producer).toBeDefined();

		const mockHandler = jest.fn();
		mockHandler.mockImplementation((stream: string, id: string, message: string[]) => {
			console.log('Received message', stream, id, message);
		});
		consumer.addMessageHandler('some handler', mockHandler);

		await consumer.setPollingInterval(STREAM_CHANNEL, 50);
		await consumer.listenToStream(STREAM_CHANNEL);

		let timeout;
		await new Promise((resolve) => {
			timeout = setTimeout(async () => {
				await producer.add(STREAM_CHANNEL, ['message', 'testMessage', 'event', 'testEveny']);
				resolve(0);
			}, 50);
		});

		await new Promise((resolve) =>
			setTimeout(async () => {
				resolve(0);
			}, 100),
		);

		clearInterval(timeout);

		consumer.stopListeningToStream(STREAM_CHANNEL);

		expect(mockHandler).toHaveBeenCalled();

		await consumer.destroy();
		await producer.destroy();
	});
});