feat(editor): Show avatars for users currently working on the same workflow

packages/cli/src/Server.ts

@@ -116,6 +116,7 @@ import { UserService } from './services/user.service';
 import { OrchestrationController } from './controllers/orchestration.controller';
 import { WorkflowHistoryController } from './workflows/workflowHistory/workflowHistory.controller.ee';
 import { InvitationController } from './controllers/invitation.controller';
+import { CollaborationService } from './collaboration/collaboration.service';
 
 const exec = promisify(callbackExec);
 
@@ -138,6 +139,8 @@ export class Server extends AbstractServer {
 
 	private postHog: PostHogClient;
 
+	private collaborationService: CollaborationService;
+
 	constructor() {
 		super('main');
 
@@ -233,6 +236,7 @@ export class Server extends AbstractServer {
 			.then(async (workflow) =>
 				Container.get(InternalHooks).onServerStarted(diagnosticInfo, workflow?.createdAt),
 			);
+		this.collaborationService = Container.get(CollaborationService);
 	}
 
 	private async registerControllers(ignoredEndpoints: Readonly<string[]>) {

packages/cli/src/collaboration/collaboration.service.ts

@@ -1,5 +1,6 @@
 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';
@@ -8,6 +9,13 @@ 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';
+
+/**
+ * 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
@@ -28,6 +36,14 @@ export class CollaborationService {
 			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);
@@ -53,6 +69,7 @@ export class CollaborationService {
 		const { workflowId } = msg;
 
 		this.state.addActiveWorkflowUser(workflowId, userId);
+		this.state.cleanInactiveUsers(workflowId, INACTIVITY_CLEAN_UP_TIME_IN_MS);
 
 		await this.sendWorkflowUsersChangedMessage(workflowId);
 	}

packages/cli/src/collaboration/collaboration.state.ts

@@ -59,4 +59,21 @@ export class CollaborationState {
 
 		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);
+			}
+		}
+	}
 }

packages/cli/src/push/index.ts

@@ -30,14 +30,21 @@ export class Push extends EventEmitter {
 
 	private backend = useWebSockets ? Container.get(WebSocketPush) : Container.get(SSEPush);
 
+	constructor() {
+		super();
+
+		if (useWebSockets) {
+			this.backend.on('message', (msg) => this.emit('message', msg));
+		}
+	}
+
 	handleRequest(req: SSEPushRequest | WebSocketPushRequest, res: PushResponse) {
 		const {
 			userId,
 			query: { sessionId },
 		} = req;
 		if (req.ws) {
 			(this.backend as WebSocketPush).add(sessionId, userId, req.ws);
-			this.backend.on('message', (msg) => this.emit('message', msg));
 		} else if (!useWebSockets) {
 			(this.backend as SSEPush).add(sessionId, userId, { req, res });
 		} else {

packages/cli/test/unit/collaboration/collaboration.state.test.ts

@@ -0,0 +1,61 @@
+import { TIME } from '@/constants';
+import { CollaborationState } from '@/collaboration/collaboration.state';
+
+const origDate = global.Date;
+
+const mockDateFactory = (currentDate: string) => {
+	return class CustomDate extends origDate {
+		constructor() {
+			super(currentDate);
+		}
+	} as DateConstructor;
+};
+
+describe('CollaborationState', () => {
+	let collaborationState: CollaborationState;
+
+	beforeEach(() => {
+		collaborationState = new CollaborationState();
+	});
+
+	describe('cleanInactiveUsers', () => {
+		const workflowId = 'workflow';
+
+		it('should remove inactive users', () => {
+			// Setup
+			global.Date = mockDateFactory('2023-01-01T00:00:00.000Z');
+			collaborationState.addActiveWorkflowUser(workflowId, 'inactiveUser');
+
+			global.Date = mockDateFactory('2023-01-01T00:30:00.000Z');
+			collaborationState.addActiveWorkflowUser(workflowId, 'activeUser');
+
+			// Act: Clean inactive users
+			jest
+				.spyOn(global.Date, 'now')
+				.mockReturnValue(new origDate('2023-01-01T00:35:00.000Z').getTime());
+			collaborationState.cleanInactiveUsers(workflowId, 10 * TIME.MINUTE);
+
+			// Assert: The inactive user should be removed
+			expect(collaborationState.getActiveWorkflowUsers(workflowId)).toEqual([
+				{ userId: 'activeUser', lastSeen: new origDate('2023-01-01T00:30:00.000Z') },
+			]);
+		});
+
+		it('should not remove active users', () => {
+			// Setup: Add an active user to the state
+			global.Date = mockDateFactory('2023-01-01T00:30:00.000Z');
+			collaborationState.addActiveWorkflowUser(workflowId, 'activeUser');
+
+			// Act: Clean inactive users
+			jest
+				.spyOn(global.Date, 'now')
+				.mockReturnValue(new origDate('2023-01-01T00:35:00.000Z').getTime());
+			collaborationState.cleanInactiveUsers(workflowId, 10 * TIME.MINUTE);
+
+			// Assert: The active user should still be present
+			expect(collaborationState.getActiveWorkflowUsers(workflowId)).toEqual([
+				{ userId: 'activeUser', lastSeen: new origDate('2023-01-01T00:30:00.000Z') },
+			]);
+		});
+	});
+});

packages/design-system/src/components/N8nUserStack/UserStack.vue

@@ -75,26 +75,31 @@ const menuHeight = computed(() => {
 			:max-height="menuHeight"
 			popper-class="user-stack-popper"
 		>
-			<div :class="$style.avatars">
+			<div :class="$style.avatars" data-test-id="user-stack-avatars">
 				<n8n-avatar
 					v-for="user in flatUserList.slice(0, visibleAvatarCount)"
 					:key="user.id"
 					:firstName="user.firstName"
 					:lastName="user.lastName"
 					:class="$style.avatar"
+					:data-test-id="`user-stack-avatar-${user.id}`"
 					size="small"
 				/>
 				<div v-if="hiddenUsersCount > 0" :class="$style.hiddenBadge">+{{ hiddenUsersCount }}</div>
 			</div>
 			<template #dropdown>
-				<el-dropdown-menu class="user-stack-list">
+				<el-dropdown-menu class="user-stack-list" data-test-id="user-stack-list">
 					<div v-for="(groupUsers, index) in nonEmptyGroups" :key="index">
 						<div :class="$style.groupContainer">
 							<el-dropdown-item>
 								<header v-if="groupCount > 1" :class="$style.groupName">{{ index }}</header>
 							</el-dropdown-item>
 							<div :class="$style.groupUsers">
-								<el-dropdown-item v-for="user in groupUsers" :key="user.id">
+								<el-dropdown-item
+									v-for="user in groupUsers"
+									:key="user.id"
+									:data-test-id="`user-stack-info-${user.id}`"
+								>
 									<n8n-user-info
 										v-bind="user"
 										:isCurrentUser="user.email === props.currentUserEmail"
@@ -156,11 +161,12 @@ const menuHeight = computed(() => {
 </style>
 
 <style lang="scss">
-.user-stack-list {
+ul.user-stack-list {
 	border: none;
 	display: flex;
 	flex-direction: column;
-	gap: 16px;
+	gap: var(--spacing-s);
+	padding-bottom: var(--spacing-2xs);
 
 	.el-dropdown-menu__item {
 		line-height: var(--font-line-height-regular);

packages/editor-ui/src/App.vue

@@ -57,6 +57,7 @@ import {
 	useCloudPlanStore,
 	useSourceControlStore,
 	useUsageStore,
+	usePushConnectionStore,
 } from '@/stores';
 import { useHistoryHelper } from '@/composables/useHistoryHelper';
 import { useRoute } from 'vue-router';
@@ -92,6 +93,7 @@ export default defineComponent({
 			useSourceControlStore,
 			useCloudPlanStore,
 			useUsageStore,
+			usePushConnectionStore,
 		),
 		defaultLocale(): string {
 			return this.rootStore.defaultLocale;
@@ -168,6 +170,7 @@ export default defineComponent({
 		void this.onAfterAuthenticate();
 
 		void runExternalHook('app.mount');
+		this.pushStore.pushConnect();
 		this.loading = false;
 	},
 	watch: {

packages/editor-ui/src/Interface.ts

@@ -411,6 +411,16 @@ export interface IExecutionDeleteFilter {
 	ids?: string[];
 }
 
+export type PushDataUsersForWorkflow = {
+	workflowId: string;
+	activeUsers: Array<{ user: IUser; lastSeen: string }>;
+};
+
+type PushDataWorkflowUsersChanged = {
+	data: PushDataUsersForWorkflow;
+	type: 'activeWorkflowUsersChanged';
+};
+
 export type IPushData =
 	| PushDataExecutionFinished
 	| PushDataExecutionStarted
@@ -424,7 +434,8 @@ export type IPushData =
 	| PushDataWorkerStatusMessage
 	| PushDataActiveWorkflowAdded
 	| PushDataActiveWorkflowRemoved
-	| PushDataWorkflowFailedToActivate;
+	| PushDataWorkflowFailedToActivate
+	| PushDataWorkflowUsersChanged;
 
 type PushDataActiveWorkflowAdded = {
 	data: IActiveWorkflowAdded;
@@ -690,6 +701,7 @@ export interface IUser extends IUserResponse {
 	fullName?: string;
 	createdAt?: string;
 	mfaEnabled: boolean;
+	globalRoleId?: number;
 }
 
 export interface IVersionNotificationSettings {

packages/editor-ui/src/components/MainHeader/CollaborationPane.vue

@@ -0,0 +1,81 @@
+<script setup lang="ts">
+import { useUsersStore } from '@/stores/users.store';
+import { useWorkflowsStore } from '@/stores/workflows.store';
+import { useCollaborationStore } from '@/stores/collaboration.store';
+import { onBeforeUnmount } from 'vue';
+import { onMounted } from 'vue';
+import { computed, ref } from 'vue';
+import { TIME } from '@/constants';
+
+const collaborationStore = useCollaborationStore();
+const usersStore = useUsersStore();
+const workflowsStore = useWorkflowsStore();
+
+const HEARTBEAT_INTERVAL = 5 * TIME.MINUTE;
+const heartbeatTimer = ref<number | null>(null);
+
+const activeUsersSorted = computed(() => {
+	const currentWorkflowUsers = (collaborationStore.getUsersForCurrentWorkflow ?? []).map(
+		(userInfo) => userInfo.user,
+	);
+	const owner = currentWorkflowUsers.find((user) => user.globalRoleId === 1);
+	return {
+		defaultGroup: owner
+			? [owner, ...currentWorkflowUsers.filter((user) => user.id !== owner.id)]
+			: currentWorkflowUsers,
+	};
+});
+
+const currentUserEmail = computed(() => {
+	return usersStore.currentUser?.email;
+});
+
+const startHeartbeat = () => {
+	if (heartbeatTimer.value !== null) {
+		clearInterval(heartbeatTimer.value);
+		heartbeatTimer.value = null;
+	}
+	heartbeatTimer.value = window.setInterval(() => {
+		collaborationStore.notifyWorkflowOpened(workflowsStore.workflow.id);
+	}, HEARTBEAT_INTERVAL);
+};
+
+const stopHeartbeat = () => {
+	if (heartbeatTimer.value !== null) {
+		clearInterval(heartbeatTimer.value);
+	}
+};
+
+const onDocumentVisibilityChange = () => {
+	if (document.visibilityState === 'hidden') {
+		stopHeartbeat();
+	} else {
+		startHeartbeat();
+	}
+};
+
+onMounted(() => {
+	startHeartbeat();
+	document.addEventListener('visibilitychange', onDocumentVisibilityChange);
+});
+
+onBeforeUnmount(() => {
+	document.removeEventListener('visibilitychange', onDocumentVisibilityChange);
+	stopHeartbeat();
+});
+</script>
+
+<template>
+	<div
+		:class="`collaboration-pane-container ${$style.container}`"
+		data-test-id="collaboration-pane"
+	>
+		<n8n-user-stack :users="activeUsersSorted" :currentUserEmail="currentUserEmail" />
+	</div>
+</template>
+
+<style lang="scss" module>
+.container {
+	margin: 0 var(--spacing-4xs);
+}
+</style>

packages/editor-ui/src/components/MainHeader/MainHeader.vue

@@ -90,11 +90,6 @@ export default defineComponent({
 	mounted() {
 		this.dirtyState = this.uiStore.stateIsDirty;
 		this.syncTabsWithRoute(this.$route);
-		// Initialize the push connection
-		this.pushConnect();
-	},
-	beforeUnmount() {
-		this.pushDisconnect();
 	},
 	watch: {
 		$route(to, from) {