vodorod/parcer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: server socket management for multiple browsers

Browse Source
This commit is contained in:
Rohit
2025-03-08 17:09:33 +05:30
parent a0e7404d25
commit 6de4cfcafe
6 changed files with 186 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/src/browser-management/controller.ts
View File

@@ -79,7 +79,7 @@ export const destroyRemoteBrowser = async (id: string, userId: string): Promise<
await browserSession.stopCurrentInterpretation();
await browserSession.switchOff();
}
return browserPool.deleteRemoteBrowser(userId);
return browserPool.deleteRemoteBrowser(id);
};
/**

182
server/src/browser-management/inputHandlers.ts
View File

@@ -4,6 +4,8 @@
* These functions are called by the client through socket communication.
*/
import { Socket } from 'socket.io';
import { IncomingMessage } from 'http';
import { JwtPayload } from 'jsonwebtoken';
import logger from "../logger";
import { Coordinates, ScrollDeltas, KeyboardInput, DatePickerEventData } from '../types';
@@ -12,7 +14,14 @@ import { WorkflowGenerator } from "../workflow-management/classes/Generator";
import { Page } from "playwright";
import { throttle } from "../../../src/helpers/inputHelpers";
import { CustomActions } from "../../../src/shared/types";
import { AuthenticatedRequest } from '../routes/record';
interface AuthenticatedIncomingMessage extends IncomingMessage {
user?: JwtPayload | string;
}
interface AuthenticatedSocket extends Socket {
request: AuthenticatedIncomingMessage;
}
/**
* A wrapper function for handling user input.
@@ -24,6 +33,7 @@ import { AuthenticatedRequest } from '../routes/record';
*
* @param handleCallback The callback handler to be called
* @param args - arguments to be passed to the handler
* @param socket - socket with authenticated request
* @category HelperFunctions
*/
const handleWrapper = async (
@@ -33,14 +43,14 @@ const handleWrapper = async (
args?: any
) => Promise<void>,
args?: any,
req?: AuthenticatedRequest,
socket?: AuthenticatedSocket,
) => {
if (!req || !req.user || typeof req.user === 'string') {
if (!socket || !socket.request || !socket.request.user || typeof socket.request.user === 'string') {
logger.log('warn', `User not authenticated or invalid JWT payload`);
return;
}
const userId = req.user.id;
const userId = socket.request.user.id;
if (!userId) {
logger.log('warn', `User ID is missing in JWT payload`);
return;
@@ -79,12 +89,13 @@ interface CustomActionEventData {
/**
* A wrapper function for handling custom actions.
* @param socket The socket connection
* @param customActionEventData The custom action event data
* @category HelperFunctions
*/
const onGenerateAction = async (customActionEventData: CustomActionEventData) => {
const onGenerateAction = async (socket: AuthenticatedSocket, customActionEventData: CustomActionEventData) => {
logger.log('debug', `Generating ${customActionEventData.action} action emitted from client`);
await handleWrapper(handleGenerateAction, customActionEventData);
await handleWrapper(handleGenerateAction, customActionEventData, socket);
}
/**
@@ -102,12 +113,13 @@ const handleGenerateAction =
/**
* A wrapper function for handling mousedown event.
* @param socket The socket connection
* @param coordinates - coordinates of the mouse click
* @category HelperFunctions
*/
const onMousedown = async (coordinates: Coordinates) => {
const onMousedown = async (socket: AuthenticatedSocket, coordinates: Coordinates) => {
logger.log('debug', 'Handling mousedown event emitted from client');
await handleWrapper(handleMousedown, coordinates);
await handleWrapper(handleMousedown, coordinates, socket);
}
/**
@@ -152,12 +164,13 @@ const handleMousedown = async (generator: WorkflowGenerator, page: Page, { x, y
/**
* A wrapper function for handling the wheel event.
* @param socket The socket connection
* @param scrollDeltas - the scroll deltas of the wheel event
* @category HelperFunctions
*/
const onWheel = async (scrollDeltas: ScrollDeltas) => {
const onWheel = async (socket: AuthenticatedSocket, scrollDeltas: ScrollDeltas) => {
logger.log('debug', 'Handling scroll event emitted from client');
await handleWrapper(handleWheel, scrollDeltas);
await handleWrapper(handleWheel, scrollDeltas, socket);
};
/**
@@ -178,12 +191,13 @@ const handleWheel = async (generator: WorkflowGenerator, page: Page, { deltaX, d
/**
* A wrapper function for handling the mousemove event.
* @param socket The socket connection
* @param coordinates - the coordinates of the mousemove event
* @category HelperFunctions
*/
const onMousemove = async (coordinates: Coordinates) => {
const onMousemove = async (socket: AuthenticatedSocket, coordinates: Coordinates) => {
logger.log('debug', 'Handling mousemove event emitted from client');
await handleWrapper(handleMousemove, coordinates);
await handleWrapper(handleMousemove, coordinates, socket);
}
/**
@@ -212,12 +226,13 @@ const handleMousemove = async (generator: WorkflowGenerator, page: Page, { x, y
/**
* A wrapper function for handling the keydown event.
* @param socket The socket connection
* @param keyboardInput - the keyboard input of the keydown event
* @category HelperFunctions
*/
const onKeydown = async (keyboardInput: KeyboardInput) => {
const onKeydown = async (socket: AuthenticatedSocket, keyboardInput: KeyboardInput) => {
logger.log('debug', 'Handling keydown event emitted from client');
await handleWrapper(handleKeydown, keyboardInput);
await handleWrapper(handleKeydown, keyboardInput, socket);
}
/**
@@ -248,49 +263,95 @@ const handleDateSelection = async (generator: WorkflowGenerator, page: Page, dat
logger.log('debug', `Date ${data.value} selected`);
}
const onDateSelection = async (data: DatePickerEventData) => {
/**
* A wrapper function for handling the date selection event.
* @param socket The socket connection
* @param data - the data of the date selection event
* @category HelperFunctions
*/
const onDateSelection = async (socket: AuthenticatedSocket, data: DatePickerEventData) => {
logger.log('debug', 'Handling date selection event emitted from client');
await handleWrapper(handleDateSelection, data);
await handleWrapper(handleDateSelection, data, socket);
}
/**
* Handles the dropdown selection event.
* @param generator - the workflow generator {@link Generator}
* @param page - the active page of the remote browser
* @param data - the data of the dropdown selection event
* @category BrowserManagement
*/
const handleDropdownSelection = async (generator: WorkflowGenerator, page: Page, data: { selector: string, value: string }) => {
await generator.onDropdownSelection(page, data);
logger.log('debug', `Dropdown value ${data.value} selected`);
}
const onDropdownSelection = async (data: { selector: string, value: string }) => {
/**
* A wrapper function for handling the dropdown selection event.
* @param socket The socket connection
* @param data - the data of the dropdown selection event
* @category HelperFunctions
*/
const onDropdownSelection = async (socket: AuthenticatedSocket, data: { selector: string, value: string }) => {
logger.log('debug', 'Handling dropdown selection event emitted from client');
await handleWrapper(handleDropdownSelection, data);
await handleWrapper(handleDropdownSelection, data, socket);
}
/**
* Handles the time selection event.
* @param generator - the workflow generator {@link Generator}
* @param page - the active page of the remote browser
* @param data - the data of the time selection event
* @category BrowserManagement
*/
const handleTimeSelection = async (generator: WorkflowGenerator, page: Page, data: { selector: string, value: string }) => {
await generator.onTimeSelection(page, data);
logger.log('debug', `Time value ${data.value} selected`);
}
const onTimeSelection = async (data: { selector: string, value: string }) => {
/**
* A wrapper function for handling the time selection event.
* @param socket The socket connection
* @param data - the data of the time selection event
* @category HelperFunctions
*/
const onTimeSelection = async (socket: AuthenticatedSocket, data: { selector: string, value: string }) => {
logger.log('debug', 'Handling time selection event emitted from client');
await handleWrapper(handleTimeSelection, data);
await handleWrapper(handleTimeSelection, data, socket);
}
/**
* Handles the datetime-local selection event.
* @param generator - the workflow generator {@link Generator}
* @param page - the active page of the remote browser
* @param data - the data of the datetime-local selection event
* @category BrowserManagement
*/
const handleDateTimeLocalSelection = async (generator: WorkflowGenerator, page: Page, data: { selector: string, value: string }) => {
await generator.onDateTimeLocalSelection(page, data);
logger.log('debug', `DateTime Local value ${data.value} selected`);
}
const onDateTimeLocalSelection = async (data: { selector: string, value: string }) => {
/**
* A wrapper function for handling the datetime-local selection event.
* @param socket The socket connection
* @param data - the data of the datetime-local selection event
* @category HelperFunctions
*/
const onDateTimeLocalSelection = async (socket: AuthenticatedSocket, data: { selector: string, value: string }) => {
logger.log('debug', 'Handling datetime-local selection event emitted from client');
await handleWrapper(handleDateTimeLocalSelection, data);
await handleWrapper(handleDateTimeLocalSelection, data, socket);
}
/**
* A wrapper function for handling the keyup event.
* @param socket The socket connection
* @param keyboardInput - the keyboard input of the keyup event
* @category HelperFunctions
*/
const onKeyup = async (keyboardInput: KeyboardInput) => {
const onKeyup = async (socket: AuthenticatedSocket, keyboardInput: KeyboardInput) => {
logger.log('debug', 'Handling keyup event emitted from client');
await handleWrapper(handleKeyup, keyboardInput);
await handleWrapper(handleKeyup, keyboardInput, socket);
}
/**
@@ -309,12 +370,13 @@ const handleKeyup = async (generator: WorkflowGenerator, page: Page, key: string
/**
* A wrapper function for handling the url change event.
* @param socket The socket connection
* @param url - the new url of the page
* @category HelperFunctions
*/
const onChangeUrl = async (url: string) => {
const onChangeUrl = async (socket: AuthenticatedSocket, url: string) => {
logger.log('debug', 'Handling change url event emitted from client');
await handleWrapper(handleChangeUrl, url);
await handleWrapper(handleChangeUrl, url, socket);
}
/**
@@ -342,11 +404,12 @@ const handleChangeUrl = async (generator: WorkflowGenerator, page: Page, url: st
/**
* A wrapper function for handling the refresh event.
* @param socket The socket connection
* @category HelperFunctions
*/
const onRefresh = async () => {
const onRefresh = async (socket: AuthenticatedSocket) => {
logger.log('debug', 'Handling refresh event emitted from client');
await handleWrapper(handleRefresh);
await handleWrapper(handleRefresh, undefined, socket);
}
/**
@@ -363,11 +426,12 @@ const handleRefresh = async (generator: WorkflowGenerator, page: Page) => {
/**
* A wrapper function for handling the go back event.
* @param socket The socket connection
* @category HelperFunctions
*/
const onGoBack = async () => {
logger.log('debug', 'Handling refresh event emitted from client');
await handleWrapper(handleGoBack);
const onGoBack = async (socket: AuthenticatedSocket) => {
logger.log('debug', 'Handling go back event emitted from client');
await handleWrapper(handleGoBack, undefined, socket);
}
/**
@@ -385,11 +449,12 @@ const handleGoBack = async (generator: WorkflowGenerator, page: Page) => {
/**
* A wrapper function for handling the go forward event.
* @param socket The socket connection
* @category HelperFunctions
*/
const onGoForward = async () => {
logger.log('debug', 'Handling refresh event emitted from client');
await handleWrapper(handleGoForward);
const onGoForward = async (socket: AuthenticatedSocket) => {
logger.log('debug', 'Handling go forward event emitted from client');
await handleWrapper(handleGoForward, undefined, socket);
}
/**
@@ -407,18 +472,7 @@ const handleGoForward = async (generator: WorkflowGenerator, page: Page) => {
/**
* Helper function for registering the handlers onto established websocket connection.
* Registers:
* - mousedownHandler
* - wheelHandler
* - mousemoveHandler
* - keydownHandler
* - keyupHandler
* - changeUrlHandler
* - refreshHandler
* - goBackHandler
* - goForwardHandler
* - onGenerateAction
* input handlers.
* Registers various input handlers.
*
* All these handlers first generates the workflow pair data
* and then calls the corresponding playwright's function to emulate the input.
@@ -428,21 +482,25 @@ const handleGoForward = async (generator: WorkflowGenerator, page: Page) => {
* @returns void
* @category BrowserManagement
*/
const registerInputHandlers = (socket: Socket) => {
socket.on("input:mousedown", onMousedown);
socket.on("input:wheel", onWheel);
socket.on("input:mousemove", onMousemove);
socket.on("input:keydown", onKeydown);
socket.on("input:keyup", onKeyup);
socket.on("input:url", onChangeUrl);
socket.on("input:refresh", onRefresh);
socket.on("input:back", onGoBack);
socket.on("input:forward", onGoForward);
socket.on("input:date", onDateSelection);
socket.on("input:dropdown", onDropdownSelection);
socket.on("input:time", onTimeSelection);
socket.on("input:datetime-local", onDateTimeLocalSelection);
socket.on("action", onGenerateAction);
const registerInputHandlers = (socket: Socket) => {
// Cast to our authenticated socket type
const authSocket = socket as AuthenticatedSocket;
// Register handlers with the socket
socket.on("input:mousedown", (data) => onMousedown(authSocket, data));
socket.on("input:wheel", (data) => onWheel(authSocket, data));
socket.on("input:mousemove", (data) => onMousemove(authSocket, data));
socket.on("input:keydown", (data) => onKeydown(authSocket, data));
socket.on("input:keyup", (data) => onKeyup(authSocket, data));
socket.on("input:url", (data) => onChangeUrl(authSocket, data));
socket.on("input:refresh", () => onRefresh(authSocket));
socket.on("input:back", () => onGoBack(authSocket));
socket.on("input:forward", () => onGoForward(authSocket));
socket.on("input:date", (data) => onDateSelection(authSocket, data));
socket.on("input:dropdown", (data) => onDropdownSelection(authSocket, data));
socket.on("input:time", (data) => onTimeSelection(authSocket, data));
socket.on("input:datetime-local", (data) => onDateTimeLocalSelection(authSocket, data));
socket.on("action", (data) => onGenerateAction(authSocket, data));
};
export default registerInputHandlers;
export default registerInputHandlers;

2
server/src/routes/record.ts
View File

@@ -80,7 +80,7 @@ router.get('/stop/:browserId', requireSignIn, async (req: AuthenticatedRequest,
if (!req.user) {
return res.status(401).send('User not authenticated');
}
const success = await destroyRemoteBrowser(req.user?.id, req.params.browserId);
const success = await destroyRemoteBrowser(req.params.browserId, req.user?.id);
return res.send(success);
});

2
server/src/routes/storage.ts
View File

@@ -617,7 +617,7 @@ router.post('/runs/run/:id', requireSignIn, async (req: AuthenticatedRequest, re
workflow, currentPage, (newPage: Page) => currentPage = newPage, plainRun.interpreterSettings);
const binaryOutputService = new BinaryOutputService('maxun-run-screenshots');
const uploadedBinaryOutput = await binaryOutputService.uploadAndStoreBinaryOutput(run, interpretationInfo.binaryOutput);
await destroyRemoteBrowser(req.user?.id, plainRun.browserId);
await destroyRemoteBrowser(plainRun.browserId, req.user?.id);
await run.update({
...run,
status: 'success',

62
server/src/socket-connection/connection.ts
View File

@@ -1,6 +1,60 @@
import { Namespace, Socket } from 'socket.io';
import { IncomingMessage } from 'http';
import { verify, JwtPayload } from 'jsonwebtoken';
import logger from "../logger";
import registerInputHandlers from '../browser-management/inputHandlers'
import registerInputHandlers from '../browser-management/inputHandlers';
interface AuthenticatedIncomingMessage extends IncomingMessage {
user?: JwtPayload | string;
}
interface AuthenticatedSocket extends Socket {
request: AuthenticatedIncomingMessage;
}
/**
* Socket.io middleware for authentication
* This is a socket.io specific auth handler that doesn't rely on Express middleware
*/
const socketAuthMiddleware = (socket: Socket, next: (err?: Error) => void) => {
const cookies = socket.handshake.headers.cookie;
if (!cookies) {
return next(new Error('Authentication required'));
}
const tokenMatch = cookies.split(';').find(c => c.trim().startsWith('token='));
if (!tokenMatch) {
return next(new Error('Authentication required'));
}
const token = tokenMatch.split('=')[1];
if (!token) {
return next(new Error('Authentication required'));
}
const secret = process.env.JWT_SECRET;
if (!secret) {
return next(new Error('Server configuration error'));
}
verify(token, secret, (err: any, user: any) => {
if (err) {
logger.log('warn', 'JWT verification error:', err);
return next(new Error('Authentication failed'));
}
// Normalize payload key
if (user.userId && !user.id) {
user.id = user.userId;
delete user.userId; // temporary: del the old key for clarity
}
// Attach user to socket request
const authSocket = socket as AuthenticatedSocket;
authSocket.request.user = user;
next();
});
};
/**
* Opens a websocket canal for duplex data transfer and registers all handlers for this data for the recording session.
@@ -13,6 +67,8 @@ export const createSocketConnection = (
io: Namespace,
callback: (socket: Socket) => void,
) => {
io.use(socketAuthMiddleware);
const onConnection = async (socket: Socket) => {
logger.log('info', "Client connected " + socket.id);
registerInputHandlers(socket);
@@ -34,6 +90,8 @@ export const createSocketConnectionForRun = (
io: Namespace,
callback: (socket: Socket) => void,
) => {
io.use(socketAuthMiddleware);
const onConnection = async (socket: Socket) => {
logger.log('info', "Client connected " + socket.id);
socket.on('disconnect', () => logger.log('info', "Client disconnected " + socket.id));
@@ -41,4 +99,4 @@ export const createSocketConnectionForRun = (
}
io.on('connection', onConnection);
};
};

6
server/src/workflow-management/scheduler/index.ts
View File

@@ -132,7 +132,7 @@ async function executeRun(id: string, userId: string) {
const binaryOutputService = new BinaryOutputService('maxun-run-screenshots');
const uploadedBinaryOutput = await binaryOutputService.uploadAndStoreBinaryOutput(run, interpretationInfo.binaryOutput);
await destroyRemoteBrowser(userId, plainRun.browserId);
await destroyRemoteBrowser(plainRun.browserId, userId);
await run.update({
...run,
@@ -215,14 +215,14 @@ async function readyForRunHandler(browserId: string, id: string, userId: string)
logger.log('info', `Interpretation of ${id} succeeded`);
} else {
logger.log('error', `Interpretation of ${id} failed`);
await destroyRemoteBrowser(userId, browserId);
await destroyRemoteBrowser(browserId, userId);
}
resetRecordingState(browserId, id);
} catch (error: any) {
logger.error(`Error during readyForRunHandler: ${error.message}`);
await destroyRemoteBrowser(userId, browserId);
await destroyRemoteBrowser(browserId, userId);
}
}