mohiit1502/Automata
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(core): Delete all collaboration related code (no-changelog) (#9929)

Browse Source
This commit is contained in:
कारतोफ्फेलस्क्रिप्ट™
2024-07-03 18:46:24 +02:00
committed by GitHub
parent 24091dfd9b
commit 22990342df
23 changed files with 57 additions and 911 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
packages/cli/src/Interfaces.ts
View File

@@ -312,7 +312,6 @@ export type IPushData =
| PushDataTestWebhook
| PushDataNodeDescriptionUpdated
| PushDataExecutionRecovered
| PushDataActiveWorkflowUsersChanged
| PushDataWorkerStatusMessage
| PushDataWorkflowActivated
| PushDataWorkflowDeactivated
@@ -333,11 +332,6 @@ type PushDataWorkflowDeactivated = {
type: 'workflowDeactivated';
};
type PushDataActiveWorkflowUsersChanged = {
data: IActiveWorkflowUsersChanged;
type: 'activeWorkflowUsersChanged';
};
export type PushDataExecutionRecovered = {
data: IPushDataExecutionRecovered;
type: 'executionRecovered';
@@ -393,20 +387,10 @@ export type PushDataNodeDescriptionUpdated = {
type: 'nodeDescriptionUpdated';
};
export interface IActiveWorkflowUser {
user: User;
lastSeen: Date;
}
export interface IActiveWorkflowAdded {
workflowId: Workflow['id'];
}
export interface IActiveWorkflowUsersChanged {
workflowId: Workflow['id'];
activeUsers: IActiveWorkflowUser[];
}
interface IActiveWorkflowChanged {
workflowId: Workflow['id'];
}

23
packages/cli/src/collaboration/collaboration.message.ts
View File

@@ -1,23 +0,0 @@
export type CollaborationMessage = WorkflowOpenedMessage | WorkflowClosedMessage;
export type WorkflowOpenedMessage = {
type: 'workflowOpened';
workflowId: string;
};
export type WorkflowClosedMessage = {
type: 'workflowClosed';
workflowId: string;
};
const isWorkflowMessage = (msg: unknown): msg is CollaborationMessage => {
return typeof msg === 'object' && msg !== null && 'type' in msg;
};
export const isWorkflowOpenedMessage = (msg: unknown): msg is WorkflowOpenedMessage => {
return isWorkflowMessage(msg) && msg.type === 'workflowOpened';
};
export const isWorkflowClosedMessage = (msg: unknown): msg is WorkflowClosedMessage => {
return isWorkflowMessage(msg) && msg.type === 'workflowClosed';
};

109
packages/cli/src/collaboration/collaboration.service.ts
View File

@@ -1,109 +0,0 @@
import type { Workflow } from 'n8n-workflow';
import { Service } from 'typedi';
import config from '@/config';
import { Push } from '../push';
import { Logger } from '@/Logger';
import type { WorkflowClosedMessage, WorkflowOpenedMessage } from './collaboration.message';
import { isWorkflowClosedMessage, isWorkflowOpenedMessage } from './collaboration.message';
import { UserService } from '../services/user.service';
import type { IActiveWorkflowUsersChanged } from '../Interfaces';
import type { OnPushMessageEvent } from '@/push/types';
import { CollaborationState } from '@/collaboration/collaboration.state';
import { TIME } from '@/constants';
import { UserRepository } from '@/databases/repositories/user.repository';
/**
* After how many minutes of inactivity a user should be removed
* as being an active user of a workflow.
*/
const INACTIVITY_CLEAN_UP_TIME_IN_MS = 15 * TIME.MINUTE;
/**
* Service for managing collaboration feature between users. E.g. keeping
* track of active users for a workflow.
*/
@Service()
export class CollaborationService {
constructor(
private readonly logger: Logger,
private readonly push: Push,
private readonly state: CollaborationState,
private readonly userService: UserService,
private readonly userRepository: UserRepository,
) {
if (!push.isBidirectional) {
logger.warn(
'Collaboration features are disabled because push is configured unidirectional. Use N8N_PUSH_BACKEND=websocket environment variable to enable them.',
);
return;
}
const isMultiMainSetup = config.get('multiMainSetup.enabled');
if (isMultiMainSetup) {
// TODO: We should support collaboration in multi-main setup as well
// This requires using redis as the state store instead of in-memory
logger.warn('Collaboration features are disabled because multi-main setup is enabled.');
return;
}
this.push.on('message', async (event: OnPushMessageEvent) => {
try {
await this.handleUserMessage(event.userId, event.msg);
} catch (error) {
this.logger.error('Error handling user message', {
error: error as unknown,
msg: event.msg,
userId: event.userId,
});
}
});
}
async handleUserMessage(userId: string, msg: unknown) {
if (isWorkflowOpenedMessage(msg)) {
await this.handleWorkflowOpened(userId, msg);
} else if (isWorkflowClosedMessage(msg)) {
await this.handleWorkflowClosed(userId, msg);
}
}
private async handleWorkflowOpened(userId: string, msg: WorkflowOpenedMessage) {
const { workflowId } = msg;
this.state.addActiveWorkflowUser(workflowId, userId);
this.state.cleanInactiveUsers(workflowId, INACTIVITY_CLEAN_UP_TIME_IN_MS);
await this.sendWorkflowUsersChangedMessage(workflowId);
}
private async handleWorkflowClosed(userId: string, msg: WorkflowClosedMessage) {
const { workflowId } = msg;
this.state.removeActiveWorkflowUser(workflowId, userId);
await this.sendWorkflowUsersChangedMessage(workflowId);
}
private async sendWorkflowUsersChangedMessage(workflowId: Workflow['id']) {
const activeWorkflowUsers = this.state.getActiveWorkflowUsers(workflowId);
const workflowUserIds = activeWorkflowUsers.map((user) => user.userId);
if (workflowUserIds.length === 0) {
return;
}
const users = await this.userRepository.getByIds(
this.userService.getManager(),
workflowUserIds,
);
const msgData: IActiveWorkflowUsersChanged = {
workflowId,
activeUsers: users.map((user) => ({
user,
lastSeen: activeWorkflowUsers.find((activeUser) => activeUser.userId === user.id)!.lastSeen,
})),
};
this.push.sendToUsers('activeWorkflowUsersChanged', msgData, workflowUserIds);
}
}

79
packages/cli/src/collaboration/collaboration.state.ts
View File

@@ -1,79 +0,0 @@
import type { User } from '@db/entities/User';
import type { Workflow } from 'n8n-workflow';
import { Service } from 'typedi';
type ActiveWorkflowUser = {
userId: User['id'];
lastSeen: Date;
};
type UserStateByUserId = Map<User['id'], ActiveWorkflowUser>;
type State = {
activeUsersByWorkflowId: Map<Workflow['id'], UserStateByUserId>;
};
/**
* State management for the collaboration service
*/
@Service()
export class CollaborationState {
private state: State = {
activeUsersByWorkflowId: new Map(),
};
addActiveWorkflowUser(workflowId: Workflow['id'], userId: User['id']) {
const { activeUsersByWorkflowId } = this.state;
let activeUsers = activeUsersByWorkflowId.get(workflowId);
if (!activeUsers) {
activeUsers = new Map();
activeUsersByWorkflowId.set(workflowId, activeUsers);
}
activeUsers.set(userId, {
userId,
lastSeen: new Date(),
});
}
removeActiveWorkflowUser(workflowId: Workflow['id'], userId: User['id']) {
const { activeUsersByWorkflowId } = this.state;
const activeUsers = activeUsersByWorkflowId.get(workflowId);
if (!activeUsers) {
return;
}
activeUsers.delete(userId);
if (activeUsers.size === 0) {
activeUsersByWorkflowId.delete(workflowId);
}
}
getActiveWorkflowUsers(workflowId: Workflow['id']): ActiveWorkflowUser[] {
const workflowState = this.state.activeUsersByWorkflowId.get(workflowId);
if (!workflowState) {
return [];
}
return [...workflowState.values()];
}
/**
* Removes all users that have not been seen in a given time
*/
cleanInactiveUsers(workflowId: Workflow['id'], inactivityCleanUpTimeInMs: number) {
const activeUsers = this.state.activeUsersByWorkflowId.get(workflowId);
if (!activeUsers) {
return;
}
const now = Date.now();
for (const user of activeUsers.values()) {
if (now - user.lastSeen.getTime() > inactivityCleanUpTimeInMs) {
activeUsers.delete(user.userId);
}
}
}
}

53
packages/cli/src/push/abstract.push.ts
View File

@@ -1,9 +1,6 @@
import { EventEmitter } from 'events';
import { assert, jsonStringify } from 'n8n-workflow';
import type { IPushDataType } from '@/Interfaces';
import type { Logger } from '@/Logger';
import type { User } from '@db/entities/User';
import type { OrchestrationService } from '@/services/orchestration.service';
/**
* Abstract class for two-way push communication.
@@ -11,23 +8,16 @@ import type { OrchestrationService } from '@/services/orchestration.service';
*
* @emits message when a message is received from a client
*/
export abstract class AbstractPush<T> extends EventEmitter {
export abstract class AbstractPush<T> {
protected connections: Record<string, T> = {};
protected userIdByPushRef: Record<string, string> = {};
protected abstract close(connection: T): void;
protected abstract sendToOneConnection(connection: T, data: string): void;
constructor(
protected readonly logger: Logger,
private readonly orchestrationService: OrchestrationService,
) {
super();
}
constructor(protected readonly logger: Logger) {}
protected add(pushRef: string, userId: User['id'], connection: T) {
const { connections, userIdByPushRef } = this;
protected add(pushRef: string, connection: T) {
const { connections } = this;
this.logger.debug('Add editor-UI session', { pushRef });
const existingConnection = connections[pushRef];
@@ -38,15 +28,6 @@ export abstract class AbstractPush<T> extends EventEmitter {
}
connections[pushRef] = connection;
userIdByPushRef[pushRef] = userId;
}
protected onMessageReceived(pushRef: string, msg: unknown) {
this.logger.debug('Received message from editor-UI', { pushRef, msg });
const userId = this.userIdByPushRef[pushRef];
this.emit('message', { pushRef, userId, msg });
}
protected remove(pushRef?: string) {
@@ -55,7 +36,6 @@ export abstract class AbstractPush<T> extends EventEmitter {
this.logger.debug('Removed editor-UI session', { pushRef });
delete this.connections[pushRef];
delete this.userIdByPushRef[pushRef];
}
private sendTo(type: IPushDataType, data: unknown, pushRefs: string[]) {
@@ -77,21 +57,7 @@ export abstract class AbstractPush<T> extends EventEmitter {
this.sendTo(type, data, Object.keys(this.connections));
}
sendToOneSession(type: IPushDataType, data: unknown, pushRef: string) {
/**
* Multi-main setup: In a manual webhook execution, the main process that
* handles a webhook might not be the same as the main process that created
* the webhook. If so, the handler process commands the creator process to
* relay the former's execution lifecycle events to the creator's frontend.
*/
if (this.orchestrationService.isMultiMainSetupEnabled && !this.hasPushRef(pushRef)) {
const payload = { type, args: data, pushRef };
void this.orchestrationService.publish('relay-execution-lifecycle-event', payload);
return;
}
sendToOne(type: IPushDataType, data: unknown, pushRef: string) {
if (this.connections[pushRef] === undefined) {
this.logger.error(`The session "${pushRef}" is not registered.`, { pushRef });
return;
@@ -100,15 +66,6 @@ export abstract class AbstractPush<T> extends EventEmitter {
this.sendTo(type, data, [pushRef]);
}
sendToUsers(type: IPushDataType, data: unknown, userIds: Array<User['id']>) {
const { connections } = this;
const userPushRefs = Object.keys(connections).filter((pushRef) =>
userIds.includes(this.userIdByPushRef[pushRef]),
);
this.sendTo(type, data, userPushRefs);
}
closeAllConnections() {
for (const pushRef in this.connections) {
// Signal the connection that we want to close it.

40
packages/cli/src/push/index.ts
View File

@@ -6,15 +6,17 @@ import type { Application } from 'express';
import { Server as WSServer } from 'ws';
import { parse as parseUrl } from 'url';
import { Container, Service } from 'typedi';
import config from '@/config';
import { SSEPush } from './sse.push';
import { WebSocketPush } from './websocket.push';
import type { PushResponse, SSEPushRequest, WebSocketPushRequest } from './types';
import type { IPushDataType } from '@/Interfaces';
import type { User } from '@db/entities/User';
import { OnShutdown } from '@/decorators/OnShutdown';
import { AuthService } from '@/auth/auth.service';
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
import type { IPushDataType } from '@/Interfaces';
import { OrchestrationService } from '@/services/orchestration.service';
import { SSEPush } from './sse.push';
import { WebSocketPush } from './websocket.push';
import type { PushResponse, SSEPushRequest, WebSocketPushRequest } from './types';
const useWebSockets = config.getEnv('push.backend') === 'websocket';
@@ -27,14 +29,10 @@ const useWebSockets = config.getEnv('push.backend') === 'websocket';
*/
@Service()
export class Push extends EventEmitter {
public isBidirectional = useWebSockets;
private backend = useWebSockets ? Container.get(WebSocketPush) : Container.get(SSEPush);
constructor() {
constructor(private readonly orchestrationService: OrchestrationService) {
super();
if (useWebSockets) this.backend.on('message', (msg) => this.emit('message', msg));
}
handleRequest(req: SSEPushRequest | WebSocketPushRequest, res: PushResponse) {
@@ -54,9 +52,9 @@ export class Push extends EventEmitter {
}
if (req.ws) {
(this.backend as WebSocketPush).add(pushRef, user.id, req.ws);
(this.backend as WebSocketPush).add(pushRef, req.ws);
} else if (!useWebSockets) {
(this.backend as SSEPush).add(pushRef, user.id, { req, res });
(this.backend as SSEPush).add(pushRef, { req, res });
} else {
res.status(401).send('Unauthorized');
return;
@@ -70,17 +68,25 @@ export class Push extends EventEmitter {
}
send(type: IPushDataType, data: unknown, pushRef: string) {
this.backend.sendToOneSession(type, data, pushRef);
/**
* Multi-main setup: In a manual webhook execution, the main process that
* handles a webhook might not be the same as the main process that created
* the webhook. If so, the handler process commands the creator process to
* relay the former's execution lifecycle events to the creator's frontend.
*/
if (this.orchestrationService.isMultiMainSetupEnabled && !this.backend.hasPushRef(pushRef)) {
const payload = { type, args: data, pushRef };
void this.orchestrationService.publish('relay-execution-lifecycle-event', payload);
return;
}
this.backend.sendToOne(type, data, pushRef);
}
getBackend() {
return this.backend;
}
sendToUsers(type: IPushDataType, data: unknown, userIds: Array<User['id']>) {
this.backend.sendToUsers(type, data, userIds);
}
@OnShutdown()
onShutdown() {
this.backend.closeAllConnections();

12
packages/cli/src/push/sse.push.ts
View File

@@ -1,10 +1,10 @@
import SSEChannel from 'sse-channel';
import { Service } from 'typedi';
import { Logger } from '@/Logger';
import { AbstractPush } from './abstract.push';
import type { PushRequest, PushResponse } from './types';
import type { User } from '@db/entities/User';
import { OrchestrationService } from '@/services/orchestration.service';
type Connection = { req: PushRequest; res: PushResponse };
@@ -14,16 +14,16 @@ export class SSEPush extends AbstractPush<Connection> {
readonly connections: Record<string, Connection> = {};
constructor(logger: Logger, orchestrationService: OrchestrationService) {
super(logger, orchestrationService);
constructor(logger: Logger) {
super(logger);
this.channel.on('disconnect', (_, { req }) => {
this.remove(req?.query?.pushRef);
});
}
add(pushRef: string, userId: User['id'], connection: Connection) {
super.add(pushRef, userId, connection);
add(pushRef: string, connection: Connection) {
super.add(pushRef, connection);
this.channel.addClient(connection.req, connection.res);
}

7
packages/cli/src/push/types.ts
View File

@@ -1,7 +1,6 @@
import type { Response } from 'express';
import type { WebSocket } from 'ws';
import type { User } from '@db/entities/User';
import type { AuthenticatedRequest } from '@/requests';
// TODO: move all push related types here
@@ -12,9 +11,3 @@ export type SSEPushRequest = PushRequest & { ws: undefined };
export type WebSocketPushRequest = PushRequest & { ws: WebSocket };
export type PushResponse = Response & { req: PushRequest };
export type OnPushMessageEvent = {
pushRef: string;
userId: User['id'];
msg: unknown;
};

27
packages/cli/src/push/websocket.push.ts
View File

@@ -2,8 +2,6 @@ import type WebSocket from 'ws';
import { Service } from 'typedi';
import { Logger } from '@/Logger';
import { AbstractPush } from './abstract.push';
import type { User } from '@db/entities/User';
import { OrchestrationService } from '@/services/orchestration.service';
function heartbeat(this: WebSocket) {
this.isAlive = true;
@@ -11,41 +9,24 @@ function heartbeat(this: WebSocket) {
@Service()
export class WebSocketPush extends AbstractPush<WebSocket> {
constructor(logger: Logger, orchestrationService: OrchestrationService) {
super(logger, orchestrationService);
constructor(logger: Logger) {
super(logger);
// Ping all connected clients every 60 seconds
setInterval(() => this.pingAll(), 60 * 1000);
}
add(pushRef: string, userId: User['id'], connection: WebSocket) {
add(pushRef: string, connection: WebSocket) {
connection.isAlive = true;
connection.on('pong', heartbeat);
super.add(pushRef, userId, connection);
const onMessage = (data: WebSocket.RawData) => {
try {
const buffer = Array.isArray(data) ? Buffer.concat(data) : Buffer.from(data);
this.onMessageReceived(pushRef, JSON.parse(buffer.toString('utf8')));
} catch (error) {
this.logger.error("Couldn't parse message from editor-UI", {
error: error as unknown,
pushRef,
data,
});
}
};
super.add(pushRef, connection);
// Makes sure to remove the session if the connection is closed
connection.once('close', () => {
connection.off('pong', heartbeat);
connection.off('message', onMessage);
this.remove(pushRef);
});
connection.on('message', onMessage);
}
protected close(connection: WebSocket): void {