mirror of
https://github.com/getmaxun/maxun.git
synced 2026-10-16 20:21:03 +08:00
fix: resolve merge conflicts
This commit is contained in:
Rohit Rajan
+1
-1
@@ -41,6 +41,7 @@
|
||||
"fortawesome": "^0.0.1-security",
|
||||
"google-auth-library": "^9.14.1",
|
||||
"googleapis": "^144.0.0",
|
||||
"graphile-worker": "^0.16.6",
|
||||
"i18next": "^24.0.2",
|
||||
"i18next-browser-languagedetector": "^8.0.0",
|
||||
"i18next-http-backend": "^3.0.1",
|
||||
@@ -60,7 +61,6 @@
|
||||
"node-cron": "^3.0.3",
|
||||
"pdf-parse": "^1.1.0",
|
||||
"pg": "^8.13.0",
|
||||
"pg-boss": "^10.1.6",
|
||||
"pkce-challenge": "^4.1.0",
|
||||
"playwright-core": "1.57.0",
|
||||
"posthog-node": "^4.2.1",
|
||||
|
||||
@@ -69,6 +69,7 @@ interface ScheduleConfig {
|
||||
nextRunAt?: Date;
|
||||
dayOfMonth?: string;
|
||||
cronExpression?: string;
|
||||
schedulerClaimedAt?: Date;
|
||||
}
|
||||
|
||||
interface RobotCreationAttributes extends Optional<RobotAttributes, 'id'> { }
|
||||
|
||||
+16
-156
@@ -16,7 +16,6 @@ import {
|
||||
} from '../browser-management/controller';
|
||||
import logger from "../logger";
|
||||
import { requireSignIn } from '../middlewares/auth';
|
||||
import { pgBossClient } from '../storage/pgboss';
|
||||
|
||||
export const router = Router();
|
||||
|
||||
@@ -24,102 +23,36 @@ export interface AuthenticatedRequest extends Request {
|
||||
user?: any;
|
||||
}
|
||||
|
||||
async function waitForJobCompletion(jobId: string, queueName: string, timeout = 15000): Promise<any> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const startTime = Date.now();
|
||||
|
||||
const checkJobStatus = async () => {
|
||||
if (Date.now() - startTime > timeout) {
|
||||
return reject(new Error(`Timeout waiting for job ${jobId} to complete`));
|
||||
}
|
||||
|
||||
try {
|
||||
const job = await pgBossClient.getJobById(queueName, jobId);
|
||||
|
||||
if (!job) {
|
||||
return reject(new Error(`Job ${jobId} not found`));
|
||||
}
|
||||
|
||||
if (job.state === 'completed') {
|
||||
return resolve(job.output);
|
||||
}
|
||||
|
||||
if (job.state === 'failed') {
|
||||
return reject(new Error(`Job ${jobId} failed.`));
|
||||
}
|
||||
|
||||
setTimeout(checkJobStatus, 200);
|
||||
} catch (error) {
|
||||
reject(error);
|
||||
}
|
||||
};
|
||||
|
||||
checkJobStatus();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Logs information about remote browser recording session.
|
||||
*/
|
||||
router.all('/', requireSignIn, (req, res, next) => {
|
||||
logger.log('debug', `The record API was invoked: ${req.url}`)
|
||||
next() // pass control to the next handler
|
||||
next()
|
||||
})
|
||||
|
||||
|
||||
/**
|
||||
* GET endpoint for starting the remote browser recording session
|
||||
* Waits for job completion
|
||||
*/
|
||||
router.get('/start', requireSignIn, async (req: AuthenticatedRequest, res: Response) => {
|
||||
if (!req.user) {
|
||||
return res.status(401).send('User not authenticated');
|
||||
}
|
||||
|
||||
|
||||
try {
|
||||
await pgBossClient.createQueue('initialize-browser-recording');
|
||||
|
||||
const jobId = await pgBossClient.send('initialize-browser-recording', {
|
||||
userId: req.user.id,
|
||||
timestamp: new Date().toISOString()
|
||||
});
|
||||
|
||||
if (!jobId) {
|
||||
const browserId = initializeRemoteBrowserForRecording(req.user.id);
|
||||
return res.send(browserId);
|
||||
}
|
||||
|
||||
logger.log('info', `Queued browser initialization job: ${jobId}, waiting for completion...`);
|
||||
|
||||
try {
|
||||
const result = await waitForJobCompletion(jobId, 'initialize-browser-recording', 15000);
|
||||
|
||||
if (result && result.browserId) {
|
||||
return res.send(result.browserId);
|
||||
} else {
|
||||
return res.send(jobId);
|
||||
}
|
||||
} catch (waitError: any) {
|
||||
return res.send(jobId);
|
||||
}
|
||||
const browserId = initializeRemoteBrowserForRecording(req.user.id);
|
||||
return res.send(browserId);
|
||||
} catch (error: any) {
|
||||
logger.log('error', `Failed to queue browser initialization job: ${error.message}`);
|
||||
|
||||
try {
|
||||
const browserId = initializeRemoteBrowserForRecording(req.user.id);
|
||||
return res.send( browserId );
|
||||
} catch (directError: any) {
|
||||
logger.log('error', `Direct initialization also failed: ${directError.message}`);
|
||||
return res.status(500).send('Failed to start recording');
|
||||
}
|
||||
logger.log('error', `Failed to start browser recording: ${error.message}`);
|
||||
return res.status(500).send('Failed to start recording');
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* POST endpoint for starting the remote browser recording session accepting browser launch options.
|
||||
* POST endpoint for starting the remote browser recording session.
|
||||
* returns session's id
|
||||
*/
|
||||
router.post('/start', requireSignIn, (req: AuthenticatedRequest, res:Response) => {
|
||||
router.post('/start', requireSignIn, (req: AuthenticatedRequest, res: Response) => {
|
||||
if (!req.user) {
|
||||
return res.status(401).send('User not authenticated');
|
||||
}
|
||||
@@ -129,44 +62,17 @@ router.post('/start', requireSignIn, (req: AuthenticatedRequest, res:Response) =
|
||||
|
||||
/**
|
||||
* GET endpoint for terminating the remote browser recording session.
|
||||
* returns whether the termination was successful
|
||||
*/
|
||||
router.get('/stop/:browserId', requireSignIn, async (req: AuthenticatedRequest, res) => {
|
||||
if (!req.user) {
|
||||
return res.status(401).send('User not authenticated');
|
||||
}
|
||||
|
||||
// Cancel the recording timeout synchronously before any async work
|
||||
// to prevent the timer firing during the pgBoss job queue window.
|
||||
clearRecordingTimeout(req.params.browserId);
|
||||
|
||||
try {
|
||||
await pgBossClient.createQueue('destroy-browser');
|
||||
|
||||
const jobId = await pgBossClient.send('destroy-browser', {
|
||||
browserId: req.params.browserId,
|
||||
userId: req.user.id,
|
||||
timestamp: new Date().toISOString()
|
||||
});
|
||||
|
||||
if (!jobId) {
|
||||
await destroyRemoteBrowser(req.params.browserId, req.user.id);
|
||||
return res.send(false);
|
||||
}
|
||||
|
||||
logger.log('info', `Queued browser destruction job: ${jobId}, waiting for completion...`);
|
||||
|
||||
try {
|
||||
const result = await waitForJobCompletion(jobId, 'destroy-browser', 15000);
|
||||
|
||||
if (result) {
|
||||
return res.send(result.success);
|
||||
} else {
|
||||
return res.send(false);
|
||||
}
|
||||
} catch (waitError: any) {
|
||||
return res.send(false);
|
||||
}
|
||||
const success = await destroyRemoteBrowser(req.params.browserId, req.user.id);
|
||||
return res.send(success);
|
||||
} catch (error: any) {
|
||||
logger.log('error', `Failed to stop browser: ${error.message}`);
|
||||
return res.status(500).send(false);
|
||||
@@ -235,33 +141,10 @@ router.get('/interpret', requireSignIn, async (req: AuthenticatedRequest, res) =
|
||||
}
|
||||
|
||||
try {
|
||||
await pgBossClient.createQueue('interpret-workflow');
|
||||
|
||||
const jobId = await pgBossClient.send('interpret-workflow', {
|
||||
userId: req.user.id,
|
||||
timestamp: new Date().toISOString()
|
||||
});
|
||||
|
||||
if (!jobId) {
|
||||
await interpretWholeWorkflow(req.user?.id);
|
||||
return res.send('interpretation done');
|
||||
}
|
||||
|
||||
logger.log('info', `Queued interpret workflow job: ${jobId}, waiting for completion...`);
|
||||
|
||||
try {
|
||||
const result = await waitForJobCompletion(jobId, 'interpret-workflow', 1000000);
|
||||
|
||||
if (result) {
|
||||
return res.send('interpretation done');
|
||||
} else {
|
||||
return res.send('interpretation failed');
|
||||
}
|
||||
} catch (waitError: any) {
|
||||
return res.send('interpretation failed');
|
||||
}
|
||||
await interpretWholeWorkflow(req.user?.id);
|
||||
return res.send('interpretation done');
|
||||
} catch (error: any) {
|
||||
logger.log('error', `Failed to stop interpret workflow: ${error.message}`);
|
||||
logger.log('error', `Failed to interpret workflow: ${error.message}`);
|
||||
return res.status(500).send('interpretation failed');
|
||||
}
|
||||
});
|
||||
@@ -272,35 +155,12 @@ router.get('/interpret/stop', requireSignIn, async (req: AuthenticatedRequest, r
|
||||
}
|
||||
|
||||
try {
|
||||
await pgBossClient.createQueue('stop-interpretation');
|
||||
|
||||
const jobId = await pgBossClient.send('stop-interpretation', {
|
||||
userId: req.user.id,
|
||||
timestamp: new Date().toISOString()
|
||||
});
|
||||
|
||||
if (!jobId) {
|
||||
await stopRunningInterpretation(req.user?.id);
|
||||
return res.send('interpretation stopped');
|
||||
}
|
||||
|
||||
logger.log('info', `Queued stop interpret workflow job: ${jobId}, waiting for completion...`);
|
||||
|
||||
try {
|
||||
const result = await waitForJobCompletion(jobId, 'stop-interpretation', 15000);
|
||||
|
||||
if (result) {
|
||||
return res.send('interpretation stopped');
|
||||
} else {
|
||||
return res.send('interpretation failed to stop');
|
||||
}
|
||||
} catch (waitError: any) {
|
||||
return res.send('interpretation failed to stop');
|
||||
}
|
||||
await stopRunningInterpretation(req.user?.id);
|
||||
return res.send('interpretation stopped');
|
||||
} catch (error: any) {
|
||||
logger.log('error', `Failed to stop interpretation: ${error.message}`);
|
||||
return res.status(500).send('interpretation failed to stop');
|
||||
}
|
||||
});
|
||||
|
||||
export default router;
|
||||
export default router;
|
||||
|
||||
@@ -15,7 +15,8 @@ import { capture } from "../utils/analytics";
|
||||
import { encrypt, decrypt } from '../utils/auth';
|
||||
import { WorkflowFile } from 'maxun-core';
|
||||
import { cancelScheduledWorkflow, scheduleWorkflow } from '../storage/schedule';
|
||||
import { pgBossClient } from '../storage/pgboss';
|
||||
import { addJob } from '../storage/graphileWorker';
|
||||
import { QUEUE_NAMES } from '../task-runner';
|
||||
import { WorkflowEnricher } from '../sdk/workflowEnricher';
|
||||
import sequelizeInstance from '../storage/db';
|
||||
import { Op } from 'sequelize';
|
||||
@@ -953,7 +954,7 @@ router.delete('/runs/:id', requireSignIn, async (req: AuthenticatedRequest, res)
|
||||
* PUT endpoint for starting a remote browser instance and saving run metadata to the storage.
|
||||
* Making it ready for interpretation and returning a runId.
|
||||
*
|
||||
* If the user has reached their browser limit, the run will be queued using pgBossClient.
|
||||
* If the user has reached their browser limit, the run will be queued.
|
||||
*/
|
||||
router.put('/runs/:id', requireSignIn, async (req: AuthenticatedRequest, res) => {
|
||||
try {
|
||||
@@ -1026,15 +1027,12 @@ router.put('/runs/:id', requireSignIn, async (req: AuthenticatedRequest, res) =>
|
||||
}
|
||||
|
||||
try {
|
||||
const userQueueName = `execute-run-user-${req.user.id}`;
|
||||
await pgBossClient.createQueue(userQueueName);
|
||||
|
||||
const jobId = await pgBossClient.send(userQueueName, {
|
||||
const jobId = await addJob(QUEUE_NAMES.EXECUTE_RUN, {
|
||||
userId: req.user.id,
|
||||
runId: runId,
|
||||
browserId: browserId,
|
||||
});
|
||||
|
||||
browserId: browserId,
|
||||
}, { maxAttempts: 1 });
|
||||
|
||||
logger.log('info', `Queued run execution job with ID: ${jobId} for run: ${runId}`);
|
||||
} catch (queueError: any) {
|
||||
logger.log('error', `Failed to queue run execution: ${queueError.message}`);
|
||||
@@ -1141,16 +1139,11 @@ router.post('/runs/run/:id', requireSignIn, async (req: AuthenticatedRequest, re
|
||||
}
|
||||
|
||||
try {
|
||||
const userQueueName = `execute-run-user-${req.user.id}`;
|
||||
|
||||
// Queue the execution job
|
||||
await pgBossClient.createQueue(userQueueName);
|
||||
|
||||
const jobId = await pgBossClient.send(userQueueName, {
|
||||
const jobId = await addJob(QUEUE_NAMES.EXECUTE_RUN, {
|
||||
userId: req.user.id,
|
||||
runId: req.params.id,
|
||||
browserId: plainRun.browserId
|
||||
});
|
||||
browserId: plainRun.browserId,
|
||||
}, { maxAttempts: 1 });
|
||||
|
||||
logger.log('info', `Queued run execution job with ID: ${jobId} for run: ${req.params.id}`);
|
||||
} catch (queueError: any) {
|
||||
@@ -1336,7 +1329,7 @@ router.delete('/schedule/:id', requireSignIn, async (req: AuthenticatedRequest,
|
||||
return res.status(404).json({ error: 'Robot not found' });
|
||||
}
|
||||
|
||||
// Cancel the scheduled job in pgBossClient
|
||||
// Cancel the scheduled job
|
||||
try {
|
||||
await cancelScheduledWorkflow(id);
|
||||
} catch (error) {
|
||||
@@ -1416,13 +1409,10 @@ router.post('/runs/abort/:id', requireSignIn, async (req: AuthenticatedRequest,
|
||||
logger.log('warn', `Failed to immediately stop interpreter: ${immediateStopError.message}`);
|
||||
}
|
||||
|
||||
const userQueueName = `abort-run-user-${req.user.id}`;
|
||||
await pgBossClient.createQueue(userQueueName);
|
||||
|
||||
const jobId = await pgBossClient.send(userQueueName, {
|
||||
const jobId = await addJob(QUEUE_NAMES.ABORT_RUN, {
|
||||
userId: req.user.id,
|
||||
runId: req.params.id
|
||||
});
|
||||
runId: req.params.id,
|
||||
}, { maxAttempts: 3 });
|
||||
|
||||
logger.log('info', `Abort signal sent for run ${req.params.id}, job ID: ${jobId}`);
|
||||
|
||||
@@ -1492,14 +1482,11 @@ async function processQueuedRuns() {
|
||||
log: 'Browser created and ready for execution'
|
||||
});
|
||||
|
||||
const userQueueName = `execute-run-user-${userId}`;
|
||||
await pgBossClient.createQueue(userQueueName);
|
||||
|
||||
const jobId = await pgBossClient.send(userQueueName, {
|
||||
const jobId = await addJob(QUEUE_NAMES.EXECUTE_RUN, {
|
||||
userId: userId,
|
||||
runId: queuedRun.runId,
|
||||
browserId: newBrowserId,
|
||||
});
|
||||
}, { maxAttempts: 1 });
|
||||
|
||||
logger.log('info', `Queued execution for run ${queuedRun.runId} with ready browser ${newBrowserId}, job ID: ${jobId}`);
|
||||
|
||||
|
||||
+154
-119
@@ -1,154 +1,189 @@
|
||||
/**
|
||||
* Worker process focused solely on scheduling logic
|
||||
* DB-backed schedule worker.
|
||||
* Polls the robot table for due scheduled robots and enqueues them
|
||||
* via Graphile Worker. No pgboss dependency.
|
||||
*/
|
||||
import PgBoss, { Job } from 'pg-boss';
|
||||
import { Op, QueryTypes, Sequelize } from 'sequelize';
|
||||
import logger from './logger';
|
||||
import Robot from './models/Robot';
|
||||
import { handleRunRecording } from './workflow-management/scheduler';
|
||||
import sequelize from './storage/db';
|
||||
import { computeNextRun } from './utils/schedule';
|
||||
import { addJob } from './storage/graphileWorker';
|
||||
import { QUEUE_NAMES } from './task-runner';
|
||||
|
||||
if (!process.env.DB_USER || !process.env.DB_PASSWORD || !process.env.DB_HOST || !process.env.DB_PORT || !process.env.DB_NAME) {
|
||||
throw new Error('One or more required environment variables are missing.');
|
||||
}
|
||||
const DB_SCHEDULER_BATCH_SIZE = 10;
|
||||
const DB_SCHEDULER_POLL_MS = 30000;
|
||||
const DB_SCHEDULER_CLAIM_TIMEOUT_MS = 10 * 60 * 1000;
|
||||
const DB_SCHEDULER_ADVISORY_LOCK_KEY = 43821742;
|
||||
|
||||
const pgBossConnectionString = `postgresql://${process.env.DB_USER}:${encodeURIComponent(process.env.DB_PASSWORD)}@${process.env.DB_HOST}:${process.env.DB_PORT}/${process.env.DB_NAME}`;
|
||||
let scheduleWorkerStarted = false;
|
||||
let dbScheduleTickRunning = false;
|
||||
const workerIntervals: NodeJS.Timeout[] = [];
|
||||
|
||||
export const pgBoss = new PgBoss({
|
||||
connectionString: pgBossConnectionString,
|
||||
max: 3,
|
||||
expireInHours: 23,
|
||||
...(process.env.DB_SSL === 'true' && { ssl: true }),
|
||||
});
|
||||
|
||||
const registeredQueues = new Set<string>();
|
||||
|
||||
interface ScheduledWorkflowData {
|
||||
interface DbScheduledRobot {
|
||||
id: string;
|
||||
runId: string;
|
||||
robotMetaId: string;
|
||||
userId: string;
|
||||
}
|
||||
/**
|
||||
* Process a scheduled workflow job
|
||||
*/
|
||||
async function processScheduledWorkflow(job: Job<ScheduledWorkflowData>) {
|
||||
const { id, runId, userId } = job.data;
|
||||
logger.log('info', `Processing scheduled workflow job for robotId: ${id}, runId: ${runId}, userId: ${userId}`);
|
||||
|
||||
try {
|
||||
// Execute the workflow using the existing handleRunRecording function
|
||||
await handleRunRecording(id, userId);
|
||||
|
||||
// Update the robot's schedule with last run and next run times
|
||||
const robot = await Robot.findOne({ where: { 'recording_meta.id': id } });
|
||||
if (robot && robot.schedule && robot.schedule.cronExpression && robot.schedule.timezone) {
|
||||
// Update lastRunAt to the current time
|
||||
const lastRunAt = new Date();
|
||||
|
||||
// Compute the next run date
|
||||
const nextRunAt = computeNextRun(robot.schedule.cronExpression, robot.schedule.timezone) || undefined;
|
||||
|
||||
|
||||
async function claimDueDbSchedules(): Promise<DbScheduledRobot[]> {
|
||||
const now = new Date();
|
||||
const claimExpiry = new Date(now.getTime() - DB_SCHEDULER_CLAIM_TIMEOUT_MS);
|
||||
|
||||
return sequelize.transaction(async (transaction) => {
|
||||
const lockResult = await sequelize.query<{ locked: boolean }>(
|
||||
'SELECT pg_try_advisory_xact_lock(:lockKey) AS locked',
|
||||
{ replacements: { lockKey: DB_SCHEDULER_ADVISORY_LOCK_KEY }, type: QueryTypes.SELECT, transaction }
|
||||
);
|
||||
|
||||
if (!lockResult[0]?.locked) return [];
|
||||
|
||||
const dueRobots = await Robot.findAll({
|
||||
where: {
|
||||
schedule: { [Op.ne]: null },
|
||||
[Op.and]: [
|
||||
Sequelize.literal(`schedule->>'cronExpression' IS NOT NULL`),
|
||||
Sequelize.literal(`schedule->>'timezone' IS NOT NULL`),
|
||||
Sequelize.literal(`schedule->>'nextRunAt' IS NOT NULL`),
|
||||
Sequelize.literal(`schedule->>'nextRunAt' <= '${now.toISOString()}'`),
|
||||
{
|
||||
[Op.or]: [
|
||||
Sequelize.literal(`schedule->>'schedulerClaimedAt' IS NULL`),
|
||||
Sequelize.literal(`schedule->>'schedulerClaimedAt' < '${claimExpiry.toISOString()}'`),
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
attributes: [
|
||||
'id',
|
||||
'userId',
|
||||
'schedule',
|
||||
[Sequelize.literal(`recording_meta->>'id'`), 'robotMetaId'],
|
||||
],
|
||||
limit: DB_SCHEDULER_BATCH_SIZE,
|
||||
order: [Sequelize.literal(`schedule->>'nextRunAt' ASC`)],
|
||||
lock: transaction.LOCK.UPDATE,
|
||||
skipLocked: true,
|
||||
transaction,
|
||||
});
|
||||
|
||||
const claimed: DbScheduledRobot[] = [];
|
||||
|
||||
for (const robot of dueRobots) {
|
||||
const robotMetaId = robot.get('robotMetaId') as string | undefined;
|
||||
if (!robotMetaId) continue;
|
||||
|
||||
await robot.update({
|
||||
schedule: {
|
||||
...robot.schedule,
|
||||
lastRunAt,
|
||||
nextRunAt,
|
||||
},
|
||||
});
|
||||
|
||||
logger.log('info', `Updated robot ${id} schedule - next run at: ${nextRunAt}`);
|
||||
} else {
|
||||
logger.log('error', `Robot ${id} schedule, cronExpression, or timezone is missing.`);
|
||||
schedule: { ...robot.schedule, schedulerClaimedAt: now } as any,
|
||||
}, { transaction });
|
||||
|
||||
claimed.push({ id: robot.id, robotMetaId, userId: String(robot.userId) });
|
||||
}
|
||||
|
||||
return { success: true };
|
||||
|
||||
return claimed;
|
||||
});
|
||||
}
|
||||
|
||||
async function finalizeSchedule(robotMetaId: string, executedAt: Date): Promise<void> {
|
||||
try {
|
||||
const robot = await Robot.findOne({ where: { 'recording_meta.id': robotMetaId }, attributes: ['id', 'schedule'] });
|
||||
if (!robot || !robot.schedule?.cronExpression || !robot.schedule?.timezone) return;
|
||||
|
||||
const nextRunAt = computeNextRun(robot.schedule.cronExpression, robot.schedule.timezone) || undefined;
|
||||
|
||||
await robot.update({
|
||||
schedule: {
|
||||
...robot.schedule,
|
||||
schedulerClaimedAt: undefined,
|
||||
lastRunAt: executedAt,
|
||||
nextRunAt,
|
||||
} as any,
|
||||
});
|
||||
|
||||
logger.log('info', `Updated robot ${robotMetaId} schedule - next run at: ${nextRunAt}`);
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Scheduled workflow job failed: ${errorMessage}`);
|
||||
return { success: false };
|
||||
logger.log('error', `Failed to finalize schedule for robot ${robotMetaId}: ${errorMessage}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a worker to handle scheduled workflow jobs
|
||||
*/
|
||||
async function registerScheduledWorkflowWorker() {
|
||||
async function releaseScheduleClaim(robotMetaId: string): Promise<void> {
|
||||
try {
|
||||
const jobs = await pgBoss.getSchedules();
|
||||
for (const job of jobs) {
|
||||
await pgBoss.createQueue(job.name);
|
||||
await registerWorkerForQueue(job.name);
|
||||
}
|
||||
|
||||
logger.log('info', 'Scheduled workflow workers registered successfully');
|
||||
const robot = await Robot.findOne({ where: { 'recording_meta.id': robotMetaId }, attributes: ['id', 'schedule'] });
|
||||
if (!robot?.schedule) return;
|
||||
|
||||
await robot.update({ schedule: { ...robot.schedule, schedulerClaimedAt: undefined } as any });
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Failed to register scheduled workflow workers: ${errorMessage}`);
|
||||
logger.log('error', `Failed to release schedule claim for robot ${robotMetaId}: ${errorMessage}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a worker for a specific queue
|
||||
* Exported to allow dynamic registration when new schedules are created
|
||||
*/
|
||||
export async function registerWorkerForQueue(queueName: string) {
|
||||
async function processDueSchedules(): Promise<void> {
|
||||
if (dbScheduleTickRunning) return;
|
||||
dbScheduleTickRunning = true;
|
||||
|
||||
try {
|
||||
if (registeredQueues.has(queueName)) {
|
||||
return;
|
||||
}
|
||||
|
||||
await pgBoss.work(queueName, async (job: Job<ScheduledWorkflowData> | Job<ScheduledWorkflowData>[]) => {
|
||||
const claimedRobots = await claimDueDbSchedules();
|
||||
|
||||
for (const robot of claimedRobots) {
|
||||
const executedAt = new Date();
|
||||
let dispatched = false;
|
||||
|
||||
try {
|
||||
const singleJob = Array.isArray(job) ? job[0] : job;
|
||||
return await processScheduledWorkflow(singleJob);
|
||||
logger.log('info', `Dispatching scheduled workflow for robot ${robot.robotMetaId}`);
|
||||
await addJob(QUEUE_NAMES.SCHEDULED_WORKFLOW, { robotMetaId: robot.robotMetaId, userId: robot.userId }, { maxAttempts: 6 });
|
||||
dispatched = true;
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Scheduled workflow job failed in queue ${queueName}: ${errorMessage}`);
|
||||
throw error;
|
||||
logger.log('error', `Scheduled workflow dispatch failed for robot ${robot.robotMetaId}: ${errorMessage}`);
|
||||
} finally {
|
||||
if (dispatched) {
|
||||
await finalizeSchedule(robot.robotMetaId, executedAt);
|
||||
} else {
|
||||
await releaseScheduleClaim(robot.robotMetaId);
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Failed to process DB-backed schedules: ${errorMessage}`);
|
||||
} finally {
|
||||
dbScheduleTickRunning = false;
|
||||
}
|
||||
}
|
||||
|
||||
export async function startScheduleWorker(): Promise<void> {
|
||||
if (scheduleWorkerStarted) {
|
||||
logger.log('warn', 'Schedule worker already started, skipping...');
|
||||
return;
|
||||
}
|
||||
|
||||
setImmediate(() => {
|
||||
processDueSchedules().catch((error) => {
|
||||
logger.log('error', `Initial DB schedule poll failed: ${error instanceof Error ? error.message : String(error)}`);
|
||||
});
|
||||
|
||||
registeredQueues.add(queueName);
|
||||
logger.log('info', `Registered worker for queue: ${queueName}`);
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Failed to register worker for queue ${queueName}: ${errorMessage}`);
|
||||
}
|
||||
});
|
||||
|
||||
const interval = setInterval(() => {
|
||||
processDueSchedules().catch((error) => {
|
||||
logger.log('error', `DB schedule poll failed: ${error instanceof Error ? error.message : String(error)}`);
|
||||
});
|
||||
}, DB_SCHEDULER_POLL_MS);
|
||||
|
||||
workerIntervals.push(interval);
|
||||
scheduleWorkerStarted = true;
|
||||
logger.log('info', `DB-backed schedule worker started (${DB_SCHEDULER_POLL_MS / 1000}s interval)`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize PgBoss and register scheduling workers
|
||||
*/
|
||||
async function startScheduleWorker() {
|
||||
try {
|
||||
logger.log('info', 'Starting PgBoss scheduling worker...');
|
||||
await pgBoss.start();
|
||||
logger.log('info', 'PgBoss scheduling worker started successfully');
|
||||
export async function stopScheduleWorker(): Promise<void> {
|
||||
if (!scheduleWorkerStarted) return;
|
||||
|
||||
// Register the scheduled workflow worker
|
||||
await registerScheduledWorkflowWorker();
|
||||
|
||||
logger.log('info', 'Scheduling worker registered successfully');
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Failed to start PgBoss scheduling worker: ${errorMessage}`);
|
||||
process.exit(1);
|
||||
for (const interval of workerIntervals.splice(0)) {
|
||||
clearInterval(interval);
|
||||
}
|
||||
|
||||
scheduleWorkerStarted = false;
|
||||
dbScheduleTickRunning = false;
|
||||
logger.log('info', 'Schedule worker stopped');
|
||||
}
|
||||
|
||||
startScheduleWorker();
|
||||
|
||||
pgBoss.on('error', (error) => {
|
||||
logger.log('error', `PgBoss scheduler error: ${error.message}`);
|
||||
});
|
||||
|
||||
process.on('SIGTERM', async () => {
|
||||
logger.log('info', 'SIGTERM received, shutting down PgBoss scheduler...');
|
||||
await pgBoss.stop();
|
||||
logger.log('info', 'PgBoss scheduler stopped, ready for termination');
|
||||
});
|
||||
|
||||
process.on('SIGINT', async () => {
|
||||
logger.log('info', 'SIGINT received, shutting down PgBoss scheduler...');
|
||||
await pgBoss.stop();
|
||||
logger.log('info', 'PgBoss scheduler stopped, waiting for main process cleanup...');
|
||||
});
|
||||
|
||||
+12
-78
@@ -12,7 +12,6 @@ import sequelize, { connectDB, syncDB } from './storage/db'
|
||||
import cookieParser from 'cookie-parser';
|
||||
import { SERVER_PORT } from "./constants/config";
|
||||
import { readdirSync } from "fs"
|
||||
import { fork } from 'child_process';
|
||||
import { capture } from "./utils/analytics";
|
||||
import swaggerUi from 'swagger-ui-express';
|
||||
import swaggerSpec from './swagger/config';
|
||||
@@ -20,8 +19,9 @@ import connectPgSimple from 'connect-pg-simple';
|
||||
import pg from 'pg';
|
||||
import session from 'express-session';
|
||||
import { processQueuedRuns, recoverOrphanedRuns } from './routes/storage';
|
||||
import { startWorkers } from './pgboss-worker';
|
||||
import { stopPgBossClient, startPgBossClient } from './storage/pgboss'
|
||||
import { startWorkers, stopWorkers } from './task-runner';
|
||||
import { startGraphileWorkerUtils, stopGraphileWorkerUtils } from './storage/graphileWorker';
|
||||
import { startScheduleWorker, stopScheduleWorker } from './schedule-worker';
|
||||
import Run from './models/Run';
|
||||
|
||||
const normalizeOrigin = (urlString?: string): string => {
|
||||
@@ -135,11 +135,6 @@ readdirSync(path.join(__dirname, 'api')).forEach((r) => {
|
||||
});
|
||||
|
||||
const isProduction = process.env.NODE_ENV === 'production';
|
||||
const workerPath = path.resolve(__dirname, isProduction ? './schedule-worker.js' : './schedule-worker.ts');
|
||||
const recordingWorkerPath = path.resolve(__dirname, isProduction ? './pgboss-worker.js' : './pgboss-worker.ts');
|
||||
|
||||
let workerProcess: any;
|
||||
let recordingWorkerProcess: any;
|
||||
|
||||
app.get('/', function (req, res) {
|
||||
capture(
|
||||
@@ -177,9 +172,9 @@ if (require.main === module) {
|
||||
|
||||
await recoverOrphanedRuns();
|
||||
|
||||
await startPgBossClient();
|
||||
|
||||
await startGraphileWorkerUtils();
|
||||
await startWorkers();
|
||||
await startScheduleWorker();
|
||||
|
||||
io.of('/queued-run').on('connection', (socket) => {
|
||||
const userId = socket.handshake.query.userId as string;
|
||||
@@ -206,55 +201,7 @@ if (require.main === module) {
|
||||
}
|
||||
});
|
||||
|
||||
if (!isProduction) {
|
||||
// Development mode
|
||||
if (process.platform === 'win32') {
|
||||
workerProcess = fork(workerPath, [], {
|
||||
execArgv: ['--inspect=5859'],
|
||||
});
|
||||
workerProcess.on('message', (message: any) => {
|
||||
console.log(`Message from worker: ${message}`);
|
||||
});
|
||||
workerProcess.on('error', (error: any) => {
|
||||
console.error(`Error in worker: ${error}`);
|
||||
});
|
||||
workerProcess.on('exit', (code: any) => {
|
||||
console.log(`Worker exited with code: ${code}`);
|
||||
});
|
||||
|
||||
recordingWorkerProcess = fork(recordingWorkerPath, [], {
|
||||
execArgv: ['--inspect=5860'],
|
||||
});
|
||||
recordingWorkerProcess.on('message', (message: any) => {
|
||||
console.log(`Message from recording worker: ${message}`);
|
||||
});
|
||||
recordingWorkerProcess.on('error', (error: any) => {
|
||||
console.error(`Error in recording worker: ${error}`);
|
||||
});
|
||||
recordingWorkerProcess.on('exit', (code: any) => {
|
||||
console.log(`Recording worker exited with code: ${code}`);
|
||||
});
|
||||
} else {
|
||||
// Run in same process for non-Windows development
|
||||
try {
|
||||
await import('./schedule-worker');
|
||||
await import('./pgboss-worker');
|
||||
console.log('Workers started in main process for memory sharing');
|
||||
} catch (error) {
|
||||
console.error('Failed to start workers in main process:', error);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Production mode - run workers in same process for memory sharing
|
||||
try {
|
||||
await import('./schedule-worker');
|
||||
await import('./pgboss-worker');
|
||||
logger.log('info', 'Workers started in main process');
|
||||
} catch (error: any) {
|
||||
logger.log('error', `Failed to start workers: ${error.message}`);
|
||||
process.exit(1);
|
||||
}
|
||||
}
|
||||
logger.log('info', 'All workers started in main process');
|
||||
|
||||
logger.log('info', `Server listening on port ${SERVER_PORT}`);
|
||||
} catch (error: any) {
|
||||
@@ -345,19 +292,12 @@ if (require.main === module) {
|
||||
console.error('Error during browser cleanup:', error.message);
|
||||
}
|
||||
|
||||
if (!isProduction) {
|
||||
try {
|
||||
if (workerProcess) {
|
||||
workerProcess.kill('SIGTERM');
|
||||
}
|
||||
if (recordingWorkerProcess) {
|
||||
recordingWorkerProcess.kill('SIGTERM');
|
||||
}
|
||||
} catch (workerError: any) {
|
||||
console.error('Error terminating worker processes:', workerError.message);
|
||||
}
|
||||
|
||||
await new Promise(resolve => setTimeout(resolve, 1000));
|
||||
try {
|
||||
await stopScheduleWorker();
|
||||
await stopWorkers();
|
||||
await stopGraphileWorkerUtils();
|
||||
} catch (workerError: any) {
|
||||
console.error('Error stopping workers:', workerError.message);
|
||||
}
|
||||
|
||||
try {
|
||||
@@ -392,12 +332,6 @@ if (require.main === module) {
|
||||
shutdownSuccessful = false;
|
||||
}
|
||||
|
||||
try {
|
||||
await stopPgBossClient();
|
||||
} catch (pgBossError: any) {
|
||||
console.error('Error closing PgBoss client connection:', pgBossError.message);
|
||||
shutdownSuccessful = false;
|
||||
}
|
||||
|
||||
try {
|
||||
await sequelize.close();
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
import { makeWorkerUtils, WorkerUtils } from 'graphile-worker';
|
||||
import { Pool } from 'pg';
|
||||
import logger from '../logger';
|
||||
import dotenv from 'dotenv';
|
||||
|
||||
dotenv.config();
|
||||
|
||||
if (!process.env.DB_USER || !process.env.DB_PASSWORD || !process.env.DB_HOST || !process.env.DB_PORT || !process.env.DB_NAME) {
|
||||
throw new Error('One or more required environment variables are missing.');
|
||||
}
|
||||
|
||||
const connectionString = `postgres://${process.env.DB_USER}:${encodeURIComponent(process.env.DB_PASSWORD!)}@${process.env.DB_HOST}:${process.env.DB_PORT}/${process.env.DB_NAME}`;
|
||||
const useSSL = process.env.DB_SSL === 'true';
|
||||
|
||||
let utilsPool: Pool | null = null;
|
||||
let workerUtils: WorkerUtils | null = null;
|
||||
let isStarted = false;
|
||||
|
||||
export async function startGraphileWorkerUtils(): Promise<void> {
|
||||
if (isStarted) {
|
||||
logger.log('warn', 'Graphile Worker utils already started, skipping...');
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
utilsPool = new Pool({
|
||||
connectionString,
|
||||
max: 3,
|
||||
ssl: useSSL ? true : undefined,
|
||||
});
|
||||
|
||||
workerUtils = await makeWorkerUtils({ pgPool: utilsPool });
|
||||
await workerUtils.migrate();
|
||||
isStarted = true;
|
||||
logger.log('info', 'Graphile Worker utils started');
|
||||
} catch (error: any) {
|
||||
logger.log('error', `Failed to start Graphile Worker utils: ${error.message}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
export async function stopGraphileWorkerUtils(): Promise<void> {
|
||||
if (!isStarted) return;
|
||||
|
||||
try {
|
||||
if (workerUtils) {
|
||||
await workerUtils.release();
|
||||
workerUtils = null;
|
||||
}
|
||||
if (utilsPool) {
|
||||
await utilsPool.end();
|
||||
utilsPool = null;
|
||||
}
|
||||
isStarted = false;
|
||||
logger.log('info', 'Graphile Worker utils stopped');
|
||||
} catch (error: any) {
|
||||
logger.log('error', `Failed to stop Graphile Worker utils: ${error.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
export async function addJob(
|
||||
taskIdentifier: string,
|
||||
payload: Record<string, unknown>,
|
||||
options?: { maxAttempts?: number; runAt?: Date; jobKey?: string },
|
||||
): Promise<string> {
|
||||
if (!workerUtils) {
|
||||
throw new Error('Graphile Worker utils not initialized — call startGraphileWorkerUtils first');
|
||||
}
|
||||
const job = await workerUtils.addJob(taskIdentifier, payload, options);
|
||||
return String(job.id);
|
||||
}
|
||||
|
||||
process.on('SIGTERM', async () => {
|
||||
await stopGraphileWorkerUtils();
|
||||
});
|
||||
|
||||
process.on('SIGINT', async () => {
|
||||
await stopGraphileWorkerUtils();
|
||||
});
|
||||
@@ -1,37 +1,25 @@
|
||||
/**
|
||||
* Shared scheduling utilities
|
||||
* These functions use the shared PgBoss client to avoid connection leaks
|
||||
*/
|
||||
import { v4 as uuid } from 'uuid';
|
||||
import logger from '../logger';
|
||||
import { pgBossClient } from './pgboss';
|
||||
import { registerWorkerForQueue } from '../schedule-worker';
|
||||
import Robot from '../models/Robot';
|
||||
import { computeNextRun } from '../utils/schedule';
|
||||
|
||||
/**
|
||||
* Utility function to schedule a cron job using PgBoss
|
||||
* @param id The robot ID
|
||||
* @param userId The user ID
|
||||
* @param cronExpression The cron expression for scheduling
|
||||
* @param timezone The timezone for the cron expression
|
||||
*/
|
||||
export async function scheduleWorkflow(id: string, userId: string, cronExpression: string, timezone: string): Promise<void> {
|
||||
export async function scheduleWorkflow(id: string, userId: string, cronExpression: string, timezone: string): Promise<void> {
|
||||
try {
|
||||
const runId = uuid();
|
||||
const robot = await Robot.findOne({ where: { 'recording_meta.id': id } });
|
||||
if (!robot) throw new Error(`Robot ${id} not found`);
|
||||
|
||||
const queueName = `scheduled-workflow-${id}`;
|
||||
const nextRunAt = computeNextRun(cronExpression, timezone) || undefined;
|
||||
|
||||
logger.log('info', `Scheduling workflow ${id} with cron expression ${cronExpression} in timezone ${timezone}`);
|
||||
await robot.update({
|
||||
schedule: {
|
||||
...robot.schedule,
|
||||
cronExpression,
|
||||
timezone,
|
||||
nextRunAt,
|
||||
schedulerClaimedAt: undefined,
|
||||
} as any,
|
||||
});
|
||||
|
||||
await pgBossClient.createQueue(queueName);
|
||||
|
||||
await pgBossClient.schedule(queueName, cronExpression,
|
||||
{ id, runId, userId },
|
||||
{ tz: timezone }
|
||||
);
|
||||
|
||||
await registerWorkerForQueue(queueName);
|
||||
|
||||
logger.log('info', `Scheduled workflow job for robot ${id}`);
|
||||
logger.log('info', `Scheduled robot ${id} (cron: ${cronExpression}, tz: ${timezone}, next: ${nextRunAt})`);
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Failed to schedule workflow: ${errorMessage}`);
|
||||
@@ -39,33 +27,27 @@ export async function scheduleWorkflow(id: string, userId: string, cronExpressi
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Utility function to cancel a scheduled job
|
||||
* @param robotId The robot ID
|
||||
* @returns true if successful
|
||||
*/
|
||||
export async function cancelScheduledWorkflow(robotId: string) {
|
||||
export async function cancelScheduledWorkflow(robotId: string): Promise<boolean> {
|
||||
try {
|
||||
const jobs = await pgBossClient.getSchedules();
|
||||
const robot = await Robot.findOne({ where: { 'recording_meta.id': robotId } });
|
||||
if (!robot) return true;
|
||||
|
||||
const matchingJobs = jobs.filter((job: any) => {
|
||||
try {
|
||||
const data = job.data;
|
||||
return data && data.id === robotId;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
});
|
||||
|
||||
for (const job of matchingJobs) {
|
||||
logger.log('info', `Cancelling scheduled job ${job.name} for robot ${robotId}`);
|
||||
await pgBossClient.unschedule(job.name);
|
||||
if (robot.schedule) {
|
||||
await robot.update({
|
||||
schedule: {
|
||||
...robot.schedule,
|
||||
cronExpression: undefined,
|
||||
nextRunAt: undefined,
|
||||
schedulerClaimedAt: undefined,
|
||||
} as any,
|
||||
});
|
||||
}
|
||||
|
||||
logger.log('info', `Cancelled schedule for robot ${robotId}`);
|
||||
return true;
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Failed to cancel scheduled workflow: ${errorMessage}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,672 @@
|
||||
import { run, Runner, TaskList } from 'graphile-worker';
|
||||
import { Pool } from 'pg';
|
||||
import logger from './logger';
|
||||
import {
|
||||
initializeRemoteBrowserForRecording,
|
||||
destroyRemoteBrowser,
|
||||
interpretWholeWorkflow,
|
||||
stopRunningInterpretation,
|
||||
} from './browser-management/controller';
|
||||
import { WorkflowFile } from 'maxun-core';
|
||||
import Run from './models/Run';
|
||||
import Robot from './models/Robot';
|
||||
import { browserPool } from './server';
|
||||
import { Page } from 'playwright-core';
|
||||
import { capture } from './utils/analytics';
|
||||
import { addGoogleSheetUpdateTask, processGoogleSheetUpdates } from './workflow-management/integrations/gsheet';
|
||||
import { addAirtableUpdateTask, processAirtableUpdates } from './workflow-management/integrations/airtable';
|
||||
import { io as serverIo } from './server';
|
||||
import { sendWebhook } from './routes/webhook';
|
||||
import { BinaryOutputService } from './storage/mino';
|
||||
import { convertPageToMarkdown, convertPageToHTML, convertPageToLinks, convertPageToScreenshot, convertPageToText } from './markdownify/scrape';
|
||||
import { executeBrowserAgent } from './sdk/browserAgent';
|
||||
import { processRobotOutputFormats } from './utils/output-post-processor';
|
||||
import { getInterpretationFailureReason, hasExpectedRobotOutput } from './utils/output-validation';
|
||||
import { handleRunRecording } from './workflow-management/scheduler';
|
||||
import dotenv from 'dotenv';
|
||||
|
||||
dotenv.config();
|
||||
|
||||
if (!process.env.DB_USER || !process.env.DB_PASSWORD || !process.env.DB_HOST || !process.env.DB_PORT || !process.env.DB_NAME) {
|
||||
throw new Error('Failed to start task runner: one or more required environment variables are missing.');
|
||||
}
|
||||
|
||||
const connectionString = `postgresql://${process.env.DB_USER}:${encodeURIComponent(process.env.DB_PASSWORD!)}@${process.env.DB_HOST}:${process.env.DB_PORT}/${process.env.DB_NAME}`;
|
||||
const useSSL = process.env.DB_SSL === 'true';
|
||||
|
||||
export const QUEUE_NAMES = {
|
||||
INITIALIZE_BROWSER_RECORDING: 'initialize-browser-recording',
|
||||
DESTROY_BROWSER: 'destroy-browser',
|
||||
INTERPRET_WORKFLOW: 'interpret-workflow',
|
||||
STOP_INTERPRETATION: 'stop-interpretation',
|
||||
EXECUTE_RUN: 'execute-run',
|
||||
ABORT_RUN: 'abort-run',
|
||||
SCHEDULED_WORKFLOW: 'scheduled-workflow',
|
||||
} as const;
|
||||
|
||||
interface InitializeBrowserData {
|
||||
userId: string;
|
||||
}
|
||||
|
||||
interface InterpretWorkflow {
|
||||
userId: string;
|
||||
}
|
||||
|
||||
interface StopInterpretWorkflow {
|
||||
userId: string;
|
||||
}
|
||||
|
||||
interface DestroyBrowserData {
|
||||
browserId: string;
|
||||
userId: string;
|
||||
}
|
||||
|
||||
interface ExecuteRunData {
|
||||
userId: string;
|
||||
runId: string;
|
||||
browserId: string;
|
||||
}
|
||||
|
||||
interface AbortRunData {
|
||||
userId: string;
|
||||
runId: string;
|
||||
}
|
||||
|
||||
interface ScheduledWorkflowData {
|
||||
robotMetaId: string;
|
||||
userId: string;
|
||||
}
|
||||
|
||||
const TOTAL_CONCURRENCY = Math.max(1, parseInt(process.env.WORKER_CONCURRENCY || '10', 10));
|
||||
|
||||
function AddGeneratedFlags(workflow: WorkflowFile) {
|
||||
const copy = JSON.parse(JSON.stringify(workflow));
|
||||
for (let i = 0; i < workflow.workflow.length; i++) {
|
||||
copy.workflow[i].what.unshift({
|
||||
action: 'flag',
|
||||
args: ['generated'],
|
||||
});
|
||||
}
|
||||
return copy;
|
||||
}
|
||||
|
||||
function withTimeout<T>(promise: Promise<T>, timeoutMs: number, operation: string): Promise<T> {
|
||||
return Promise.race([
|
||||
promise,
|
||||
new Promise<T>((_, reject) =>
|
||||
setTimeout(() => reject(new Error(`${operation} timed out after ${timeoutMs}ms`)), timeoutMs)
|
||||
)
|
||||
]);
|
||||
}
|
||||
|
||||
async function triggerIntegrationUpdates(runId: string, robotMetaId: string): Promise<void> {
|
||||
try {
|
||||
addGoogleSheetUpdateTask(runId, { robotId: robotMetaId, runId, status: 'pending', retries: 5 });
|
||||
addAirtableUpdateTask(runId, { robotId: robotMetaId, runId, status: 'pending', retries: 5 });
|
||||
|
||||
withTimeout(processAirtableUpdates(), 65000, 'Airtable update')
|
||||
.catch(err => logger.log('error', `Airtable update error: ${err.message}`));
|
||||
withTimeout(processGoogleSheetUpdates(), 65000, 'Google Sheets update')
|
||||
.catch(err => logger.log('error', `Google Sheets update error: ${err.message}`));
|
||||
} catch (err: any) {
|
||||
logger.log('error', `Failed to update integrations for run: ${runId}: ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
const getRobotTargetUrl = (recording: any): string => {
|
||||
const metaUrl = recording?.recording_meta?.url?.trim();
|
||||
if (metaUrl) return metaUrl;
|
||||
|
||||
const workflow = recording?.recording?.workflow || [];
|
||||
const entryPair = [...workflow].reverse().find((pair: any) =>
|
||||
pair?.what?.some((action: any) => action.action === 'goto' && typeof action.args?.[0] === 'string' && action.args[0] !== 'about:blank'),
|
||||
);
|
||||
const gotoUrl = entryPair?.what?.find((action: any) => action.action === 'goto' && typeof action.args?.[0] === 'string')?.args?.[0]?.trim();
|
||||
if (gotoUrl) return gotoUrl;
|
||||
|
||||
return workflow.find((pair: any) => typeof pair?.where?.url === 'string' && pair.where.url !== 'about:blank')?.where?.url?.trim() || '';
|
||||
};
|
||||
|
||||
async function processRunExecution(data: ExecuteRunData): Promise<void> {
|
||||
const BROWSER_INIT_TIMEOUT = 60000;
|
||||
const BROWSER_PAGE_TIMEOUT = 15000;
|
||||
|
||||
logger.log('info', `Processing run execution job for runId: ${data.runId}, browserId: ${data.browserId}`);
|
||||
|
||||
try {
|
||||
const run = await Run.findOne({ where: { runId: data.runId } });
|
||||
if (!run) {
|
||||
logger.log('error', `Run ${data.runId} not found in database`);
|
||||
return;
|
||||
}
|
||||
|
||||
if (run.status === 'aborted' || run.status === 'aborting') {
|
||||
logger.log('info', `Run ${data.runId} has status ${run.status}, skipping execution`);
|
||||
return;
|
||||
}
|
||||
|
||||
if (run.status === 'queued') {
|
||||
logger.log('info', `Run ${data.runId} has status 'queued', skipping stale execution job`);
|
||||
return;
|
||||
}
|
||||
|
||||
const plainRun = run.toJSON();
|
||||
const browserId = data.browserId || plainRun.browserId;
|
||||
|
||||
if (!browserId) throw new Error(`No browser ID available for run ${data.runId}`);
|
||||
|
||||
logger.log('info', `Looking for browser ${browserId} for run ${data.runId}`);
|
||||
|
||||
let browser = browserPool.getRemoteBrowser(browserId);
|
||||
const browserWaitStart = Date.now();
|
||||
let lastLogTime = 0;
|
||||
let pollAttempts = 0;
|
||||
const MAX_POLL_ATTEMPTS = Math.ceil(BROWSER_INIT_TIMEOUT / 2000);
|
||||
|
||||
while (!browser && (Date.now() - browserWaitStart) < BROWSER_INIT_TIMEOUT && pollAttempts < MAX_POLL_ATTEMPTS) {
|
||||
const currentTime = Date.now();
|
||||
pollAttempts++;
|
||||
|
||||
const browserStatus = browserPool.getBrowserStatus(browserId);
|
||||
if (browserStatus === null) throw new Error(`Browser slot ${browserId} does not exist in pool`);
|
||||
if (browserStatus === 'failed') throw new Error(`Browser ${browserId} initialization failed`);
|
||||
|
||||
if (currentTime - lastLogTime > 10000) {
|
||||
logger.log('info', `Browser ${browserId} not ready yet (status: ${browserStatus}), waiting... (${Math.round((currentTime - browserWaitStart) / 1000)}s elapsed)`);
|
||||
lastLogTime = currentTime;
|
||||
}
|
||||
|
||||
await new Promise(resolve => setTimeout(resolve, 2000));
|
||||
browser = browserPool.getRemoteBrowser(browserId);
|
||||
}
|
||||
|
||||
if (!browser) {
|
||||
const finalStatus = browserPool.getBrowserStatus(browserId);
|
||||
throw new Error(`Browser ${browserId} not found in pool after ${BROWSER_INIT_TIMEOUT / 1000}s timeout (final status: ${finalStatus})`);
|
||||
}
|
||||
|
||||
logger.log('info', `Browser ${browserId} found and ready for execution`);
|
||||
|
||||
try {
|
||||
const recording = await Robot.findOne({ where: { 'recording_meta.id': plainRun.robotMetaId }, raw: true });
|
||||
if (!recording) throw new Error(`Recording for run ${data.runId} not found`);
|
||||
|
||||
let currentPage = browser.getCurrentPage();
|
||||
const pageWaitStart = Date.now();
|
||||
let lastPageLogTime = 0;
|
||||
let pageAttempts = 0;
|
||||
|
||||
while (!currentPage && (Date.now() - pageWaitStart) < BROWSER_PAGE_TIMEOUT && pageAttempts < 15) {
|
||||
const currentTime = Date.now();
|
||||
pageAttempts++;
|
||||
if (currentTime - lastPageLogTime > 5000) {
|
||||
logger.log('info', `Page not ready for browser ${browserId}, waiting...`);
|
||||
lastPageLogTime = currentTime;
|
||||
}
|
||||
await new Promise(resolve => setTimeout(resolve, 1000));
|
||||
currentPage = browser.getCurrentPage();
|
||||
}
|
||||
|
||||
if (!currentPage) throw new Error(`No current page available for browser ${browserId} after ${BROWSER_PAGE_TIMEOUT / 1000}s timeout`);
|
||||
|
||||
if (recording.recording_meta.type === 'scrape') {
|
||||
logger.log('info', `Executing scrape robot for run ${data.runId}`);
|
||||
|
||||
const rawFormats = run.interpreterSettings?.formats || recording.recording_meta.formats;
|
||||
const formats = rawFormats && rawFormats.length > 0 ? rawFormats : ['markdown'];
|
||||
|
||||
await run.update({ status: 'running', log: `Converting page to ${formats.join(', ')}` });
|
||||
|
||||
try {
|
||||
const url = getRobotTargetUrl(recording);
|
||||
if (!url) throw new Error('No URL specified for scrape robot');
|
||||
|
||||
const serializableOutput: any = {};
|
||||
const binaryOutput: any = {};
|
||||
const SCRAPE_TIMEOUT = 120000;
|
||||
|
||||
if (formats.includes('screenshot-visible')) {
|
||||
const buf = await Promise.race([convertPageToScreenshot(url, currentPage, false), new Promise<never>((_, r) => setTimeout(() => r(new Error('timeout')), SCRAPE_TIMEOUT))]);
|
||||
if (!binaryOutput['screenshot-visible']) binaryOutput['screenshot-visible'] = { data: buf.toString('base64'), mimeType: 'image/png' };
|
||||
}
|
||||
|
||||
if (formats.includes('screenshot-fullpage')) {
|
||||
const buf = await Promise.race([convertPageToScreenshot(url, currentPage, true), new Promise<never>((_, r) => setTimeout(() => r(new Error('timeout')), SCRAPE_TIMEOUT))]);
|
||||
if (!binaryOutput['screenshot-fullpage']) binaryOutput['screenshot-fullpage'] = { data: buf.toString('base64'), mimeType: 'image/png' };
|
||||
}
|
||||
|
||||
if (formats.includes('text')) {
|
||||
const text = await Promise.race([convertPageToText(url, currentPage), new Promise<never>((_, r) => setTimeout(() => r(new Error('timeout')), SCRAPE_TIMEOUT))]);
|
||||
if (text) serializableOutput.text = [{ content: text }];
|
||||
}
|
||||
|
||||
if (formats.includes('markdown')) {
|
||||
const markdown = await Promise.race([convertPageToMarkdown(url, currentPage), new Promise<never>((_, r) => setTimeout(() => r(new Error('timeout')), SCRAPE_TIMEOUT))]);
|
||||
serializableOutput.markdown = [{ content: markdown }];
|
||||
}
|
||||
|
||||
if (formats.includes('html')) {
|
||||
const html = await Promise.race([convertPageToHTML(url, currentPage), new Promise<never>((_, r) => setTimeout(() => r(new Error('timeout')), SCRAPE_TIMEOUT))]);
|
||||
serializableOutput.html = [{ content: html }];
|
||||
}
|
||||
|
||||
if (formats.includes('links')) {
|
||||
try {
|
||||
const links = await Promise.race([convertPageToLinks(url, currentPage), new Promise<never>((_, r) => setTimeout(() => r(new Error('timeout')), SCRAPE_TIMEOUT))]);
|
||||
if (links && links.length > 0) serializableOutput.links = links.map((link: string) => ({ url: link }));
|
||||
} catch (error: any) {
|
||||
logger.log('warn', `Links extraction failed for run ${data.runId}: ${error.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
const promptInstructions = run.interpreterSettings?.promptInstructions || (recording.recording_meta as any).promptInstructions as string | undefined;
|
||||
if (promptInstructions) {
|
||||
try {
|
||||
const llmConfig = {
|
||||
provider: ((recording.recording_meta as any).promptLlmProvider || 'ollama') as 'anthropic' | 'openai' | 'ollama',
|
||||
model: (recording.recording_meta as any).promptLlmModel as string | undefined,
|
||||
apiKey: (recording.recording_meta as any).promptLlmApiKey as string | undefined,
|
||||
baseUrl: (recording.recording_meta as any).promptLlmBaseUrl as string | undefined,
|
||||
};
|
||||
await run.update({ log: 'Running smart query...' });
|
||||
const agentResult = await executeBrowserAgent(currentPage, promptInstructions, llmConfig);
|
||||
serializableOutput.promptResult = [{ content: agentResult.result, steps: agentResult.steps }];
|
||||
} catch (agentError: any) {
|
||||
logger.log('warn', `Smart query failed for run ${data.runId}: ${agentError.message}`);
|
||||
serializableOutput.promptResult = [{ content: `Smart query failed: ${agentError.message}`, steps: [] }];
|
||||
}
|
||||
}
|
||||
|
||||
await run.update({ status: 'success', finishedAt: new Date().toLocaleString(), log: `${formats.join(', ').toUpperCase()} conversion completed successfully`, serializableOutput, binaryOutput });
|
||||
|
||||
let uploadedBinaryOutput: Record<string, string> = {};
|
||||
if (Object.keys(binaryOutput).length > 0) {
|
||||
const svc = new BinaryOutputService('maxun-run-screenshots');
|
||||
uploadedBinaryOutput = await svc.uploadAndStoreBinaryOutput(run, binaryOutput);
|
||||
await run.update({ binaryOutput: uploadedBinaryOutput });
|
||||
}
|
||||
|
||||
try {
|
||||
const completionData = { runId: data.runId, robotMetaId: plainRun.robotMetaId, robotName: recording.recording_meta.name, status: 'success', finishedAt: new Date().toLocaleString() };
|
||||
serverIo.of(browserId).emit('run-completed', completionData);
|
||||
serverIo.of('/queued-run').to(`user-${data.userId}`).emit('run-completed', completionData);
|
||||
} catch (socketError: any) {
|
||||
logger.log('warn', `Failed to send run-completed notification for run ${data.runId}: ${socketError.message}`);
|
||||
}
|
||||
|
||||
try {
|
||||
await sendWebhook(plainRun.robotMetaId, 'run_completed', { runId: data.runId, robotId: plainRun.robotMetaId, robotName: recording.recording_meta.name, status: 'success', finishedAt: new Date().toLocaleString() });
|
||||
} catch (webhookError: any) {
|
||||
logger.log('warn', `Failed to send webhooks for run ${data.runId}: ${webhookError.message}`);
|
||||
}
|
||||
|
||||
capture('maxun-oss-run-created', { runId: data.runId, user_id: data.userId, status: 'success', robot_type: 'scrape', formats, source: 'manual' });
|
||||
await destroyRemoteBrowser(browserId, data.userId);
|
||||
return;
|
||||
|
||||
} catch (error: any) {
|
||||
logger.log('error', `Scrape conversion failed for run ${data.runId}: ${error.message}`);
|
||||
await run.update({ status: 'failed', finishedAt: new Date().toLocaleString(), log: `Conversion failed: ${error.message}` });
|
||||
try {
|
||||
const failureData = { runId: data.runId, robotMetaId: plainRun.robotMetaId, robotName: recording.recording_meta.name, status: 'failed', finishedAt: new Date().toLocaleString() };
|
||||
serverIo.of(browserId).emit('run-completed', failureData);
|
||||
serverIo.of('/queued-run').to(`user-${data.userId}`).emit('run-completed', failureData);
|
||||
} catch (_) {}
|
||||
capture('maxun-oss-run-created', { runId: data.runId, user_id: data.userId, status: 'failed', robot_type: 'scrape', source: 'manual' });
|
||||
await destroyRemoteBrowser(browserId, data.userId);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
const isRunAborted = async (): Promise<boolean> => {
|
||||
try {
|
||||
const currentRun = await Run.findOne({ where: { runId: data.runId } });
|
||||
return currentRun ? (currentRun.status === 'aborted' || currentRun.status === 'aborting') : false;
|
||||
} catch { return false; }
|
||||
};
|
||||
|
||||
logger.log('info', `Starting workflow execution for run ${data.runId}`);
|
||||
await run.update({ status: 'running', log: 'Workflow execution started' });
|
||||
|
||||
try {
|
||||
const startedData = { runId: data.runId, robotMetaId: plainRun.robotMetaId, robotName: recording.recording_meta.name, status: 'running', startedAt: new Date().toLocaleString() };
|
||||
serverIo.of(browserId).emit('run-started', startedData);
|
||||
serverIo.of('/queued-run').to(`user-${data.userId}`).emit('run-started', startedData);
|
||||
} catch (socketError: any) {
|
||||
logger.log('warn', `Failed to send run-started notification for run ${data.runId}: ${socketError.message}`);
|
||||
}
|
||||
|
||||
browser.interpreter.setRunId(data.runId);
|
||||
|
||||
const INTERPRETATION_TIMEOUT = 600000;
|
||||
const interpretationPromise = browser.interpreter.InterpretRecording(
|
||||
AddGeneratedFlags(recording.recording),
|
||||
currentPage,
|
||||
(newPage: Page) => currentPage = newPage,
|
||||
plainRun.interpreterSettings,
|
||||
);
|
||||
|
||||
const interpretationInfo = await Promise.race([
|
||||
interpretationPromise,
|
||||
new Promise<never>((_, reject) => setTimeout(() => reject(new Error(`Workflow interpretation timed out after ${INTERPRETATION_TIMEOUT / 1000}s`)), INTERPRETATION_TIMEOUT))
|
||||
]);
|
||||
|
||||
if (await isRunAborted()) {
|
||||
logger.log('info', `Run ${data.runId} was aborted during execution, not updating status`);
|
||||
try { await browser.interpreter.clearState(); } catch (_) {}
|
||||
await destroyRemoteBrowser(plainRun.browserId, data.userId);
|
||||
return;
|
||||
}
|
||||
|
||||
logger.log('info', `Workflow execution completed for run ${data.runId}`);
|
||||
|
||||
const finalRun = await Run.findByPk(run.id);
|
||||
const categorizedOutput = {
|
||||
scrapeSchema: finalRun?.serializableOutput?.scrapeSchema || {},
|
||||
scrapeList: finalRun?.serializableOutput?.scrapeList || {},
|
||||
crawl: finalRun?.serializableOutput?.crawl || {},
|
||||
search: finalRun?.serializableOutput?.search || {}
|
||||
};
|
||||
|
||||
let binaryOutput: Record<string, any> = { ...(interpretationInfo.binaryOutput || {}) };
|
||||
|
||||
const robotType = recording.recording_meta.type;
|
||||
const outputFormats = (recording.recording_meta as any).formats as string[] | undefined;
|
||||
if (robotType === 'crawl' || robotType === 'search') {
|
||||
const processedOutput = await processRobotOutputFormats({
|
||||
robotType,
|
||||
outputFormats,
|
||||
categorizedOutput: { crawl: categorizedOutput.crawl as Record<string, any>, search: categorizedOutput.search as Record<string, any> },
|
||||
currentPage,
|
||||
initialBinaryOutput: binaryOutput,
|
||||
});
|
||||
|
||||
categorizedOutput.crawl = processedOutput.categorizedOutput.crawl;
|
||||
categorizedOutput.search = processedOutput.categorizedOutput.search;
|
||||
binaryOutput = processedOutput.binaryOutput;
|
||||
|
||||
const hasOutput = hasExpectedRobotOutput(robotType, { crawl: categorizedOutput.crawl as Record<string, any>, search: categorizedOutput.search as Record<string, any> }, outputFormats, binaryOutput);
|
||||
if (!hasOutput) {
|
||||
const humanRobotType = robotType.charAt(0).toUpperCase() + robotType.slice(1);
|
||||
throw new Error(getInterpretationFailureReason(interpretationInfo.log, `${humanRobotType} run completed without producing output data`));
|
||||
}
|
||||
}
|
||||
|
||||
const binarySvc = new BinaryOutputService('maxun-run-screenshots');
|
||||
const uploadedBinaryOutput = Object.keys(binaryOutput).length > 0 ? await binarySvc.uploadAndStoreBinaryOutput(run, binaryOutput) : {};
|
||||
|
||||
if (await isRunAborted()) {
|
||||
logger.log('info', `Run ${data.runId} was aborted while processing results, not updating status`);
|
||||
return;
|
||||
}
|
||||
|
||||
await run.update({
|
||||
status: 'success',
|
||||
finishedAt: new Date().toLocaleString(),
|
||||
log: interpretationInfo.log.join('\n'),
|
||||
binaryOutput: uploadedBinaryOutput,
|
||||
serializableOutput: { ...(finalRun?.serializableOutput || {}), crawl: categorizedOutput.crawl, search: categorizedOutput.search }
|
||||
});
|
||||
|
||||
let totalSchemaItemsExtracted = 0;
|
||||
let totalListItemsExtracted = 0;
|
||||
if (categorizedOutput.scrapeSchema) Object.values(categorizedOutput.scrapeSchema).forEach((v: any) => { totalSchemaItemsExtracted += Array.isArray(v) ? v.length : (v ? 1 : 0); });
|
||||
if (categorizedOutput.scrapeList) Object.values(categorizedOutput.scrapeList).forEach((v: any) => { totalListItemsExtracted += Array.isArray(v) ? v.length : 0; });
|
||||
|
||||
capture('maxun-oss-run-created', { runId: data.runId, user_id: data.userId, created_at: new Date().toISOString(), status: 'success', totalRowsExtracted: totalSchemaItemsExtracted + totalListItemsExtracted, schemaItemsExtracted: totalSchemaItemsExtracted, listItemsExtracted: totalListItemsExtracted, extractedScreenshotsCount: Object.keys(uploadedBinaryOutput).length, is_llm: (recording.recording_meta as any).isLLM, source: 'manual' });
|
||||
|
||||
try {
|
||||
const completionData = { runId: data.runId, robotMetaId: plainRun.robotMetaId, robotName: recording.recording_meta.name, status: 'success', finishedAt: new Date().toLocaleString() };
|
||||
serverIo.of(browserId).emit('run-completed', completionData);
|
||||
serverIo.of('/queued-run').to(`user-${data.userId}`).emit('run-completed', completionData);
|
||||
} catch (socketError: any) {
|
||||
logger.log('warn', `Failed to send run-completed notification for run ${data.runId}: ${socketError.message}`);
|
||||
}
|
||||
|
||||
try {
|
||||
await sendWebhook(plainRun.robotMetaId, 'run_completed', {
|
||||
robot_id: plainRun.robotMetaId, run_id: data.runId, robot_name: recording.recording_meta.name, status: 'success',
|
||||
started_at: plainRun.startedAt, finished_at: new Date().toLocaleString(),
|
||||
extracted_data: {
|
||||
captured_texts: Object.keys(categorizedOutput.scrapeSchema || {}).length > 0 ? Object.entries(categorizedOutput.scrapeSchema).reduce((acc, [name, value]) => { acc[name] = Array.isArray(value) ? value : [value]; return acc; }, {} as Record<string, any[]>) : {},
|
||||
captured_lists: categorizedOutput.scrapeList,
|
||||
crawl_data: categorizedOutput.crawl, search_data: categorizedOutput.search,
|
||||
captured_texts_count: totalSchemaItemsExtracted, captured_lists_count: totalListItemsExtracted, screenshots_count: Object.keys(uploadedBinaryOutput).length
|
||||
},
|
||||
metadata: { browser_id: plainRun.browserId, user_id: data.userId }
|
||||
});
|
||||
} catch (webhookError: any) {
|
||||
logger.log('error', `Failed to send webhooks for run ${data.runId}: ${webhookError.message}`);
|
||||
}
|
||||
|
||||
await triggerIntegrationUpdates(plainRun.runId, plainRun.robotMetaId);
|
||||
await destroyRemoteBrowser(browserId, data.userId);
|
||||
logger.log('info', `Browser ${browserId} destroyed after successful run ${data.runId}`);
|
||||
|
||||
} catch (executionError: any) {
|
||||
logger.log('error', `Run execution failed for run ${data.runId}: ${executionError.message}`);
|
||||
|
||||
const browserId = data.browserId || (run as any).toJSON().browserId;
|
||||
let partialDataExtracted = false;
|
||||
|
||||
try {
|
||||
const hasData = (run.serializableOutput && (
|
||||
(run.serializableOutput.scrapeSchema && run.serializableOutput.scrapeSchema.length > 0) ||
|
||||
(run.serializableOutput.scrapeList && run.serializableOutput.scrapeList.length > 0) ||
|
||||
(run.serializableOutput.crawl && Object.keys(run.serializableOutput.crawl).length > 0) ||
|
||||
(run.serializableOutput.search && Object.keys(run.serializableOutput.search).length > 0))) ||
|
||||
(run.binaryOutput && Object.keys(run.binaryOutput).length > 0);
|
||||
if (hasData) {
|
||||
await triggerIntegrationUpdates((run as any).toJSON().runId, (run as any).toJSON().robotMetaId);
|
||||
partialDataExtracted = true;
|
||||
}
|
||||
} catch (_) {}
|
||||
|
||||
await run.update({ status: 'failed', finishedAt: new Date().toLocaleString(), log: `Failed: ${executionError.message}` });
|
||||
|
||||
const plainRun = (run as any).toJSON();
|
||||
const recording = await Robot.findOne({ where: { 'recording_meta.id': run.robotMetaId }, raw: true });
|
||||
|
||||
try {
|
||||
const failureData = { runId: data.runId, robotMetaId: plainRun.robotMetaId, robotName: recording ? recording.recording_meta.name : 'Unknown Robot', status: 'failed', finishedAt: new Date().toLocaleString(), hasPartialData: partialDataExtracted };
|
||||
serverIo.of(browserId).emit('run-completed', failureData);
|
||||
serverIo.of('/queued-run').to(`user-${data.userId}`).emit('run-completed', failureData);
|
||||
} catch (_) {}
|
||||
|
||||
try {
|
||||
await sendWebhook(plainRun.robotMetaId, 'run_failed', {
|
||||
robot_id: plainRun.robotMetaId, run_id: data.runId, robot_name: recording ? recording.recording_meta.name : 'Unknown Robot',
|
||||
status: 'failed', started_at: plainRun.startedAt, finished_at: new Date().toLocaleString(),
|
||||
error: { message: executionError.message, stack: executionError.stack, type: 'ExecutionError' },
|
||||
partial_data_extracted: partialDataExtracted, metadata: { browser_id: plainRun.browserId, user_id: data.userId }
|
||||
});
|
||||
} catch (_) {}
|
||||
|
||||
capture('maxun-oss-run-created', { runId: data.runId, user_id: data.userId, created_at: new Date().toISOString(), status: 'failed', error_message: executionError.message, partial_data_extracted: partialDataExtracted, source: 'manual' });
|
||||
|
||||
try { if (browser && browser.interpreter) await browser.interpreter.clearState(); } catch (_) {}
|
||||
await destroyRemoteBrowser(browserId, data.userId);
|
||||
}
|
||||
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Failed to process run execution job: ${errorMessage}`);
|
||||
|
||||
try {
|
||||
const run = await Run.findOne({ where: { runId: data.runId } });
|
||||
if (run) {
|
||||
await run.update({ status: 'failed', finishedAt: new Date().toLocaleString(), log: `Failed: ${errorMessage}` });
|
||||
const recording = await Robot.findOne({ where: { 'recording_meta.id': run.robotMetaId }, raw: true });
|
||||
try {
|
||||
await sendWebhook(run.robotMetaId, 'run_failed', {
|
||||
robot_id: run.robotMetaId, run_id: data.runId, robot_name: recording ? recording.recording_meta.name : 'Unknown Robot',
|
||||
status: 'failed', started_at: run.startedAt, finished_at: new Date().toLocaleString(),
|
||||
error: { message: errorMessage }, metadata: { browser_id: run.browserId, user_id: data.userId }
|
||||
});
|
||||
} catch (_) {}
|
||||
try {
|
||||
serverIo.of(run.browserId).emit('run-completed', { runId: data.runId, robotMetaId: run.robotMetaId, robotName: recording ? recording.recording_meta.name : 'Unknown Robot', status: 'failed', finishedAt: new Date().toLocaleString() });
|
||||
serverIo.of('/queued-run').to(`user-${data.userId}`).emit('run-completed', { runId: data.runId, status: 'failed' });
|
||||
} catch (_) {}
|
||||
}
|
||||
} catch (_) {}
|
||||
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function abortRun(runId: string, userId: string): Promise<void> {
|
||||
try {
|
||||
const run = await Run.findOne({ where: { runId } });
|
||||
if (!run) {
|
||||
logger.log('warn', `Run ${runId} not found`);
|
||||
return;
|
||||
}
|
||||
|
||||
await run.update({ status: 'aborting' });
|
||||
|
||||
const plainRun = run.toJSON();
|
||||
const recording = await Robot.findOne({ where: { 'recording_meta.id': plainRun.robotMetaId }, raw: true });
|
||||
const robotName = recording?.recording_meta?.name || 'Unknown Robot';
|
||||
|
||||
let browser;
|
||||
try { browser = browserPool.getRemoteBrowser(plainRun.browserId); } catch { browser = null; }
|
||||
|
||||
if (!browser) {
|
||||
await run.update({ status: 'aborted', finishedAt: new Date().toLocaleString(), log: 'Aborted: Browser not found or already closed' });
|
||||
try { serverIo.of(plainRun.browserId).emit('run-aborted', { runId, robotName, status: 'aborted', finishedAt: new Date().toLocaleString() }); } catch (_) {}
|
||||
return;
|
||||
}
|
||||
|
||||
await run.update({ status: 'aborted', finishedAt: new Date().toLocaleString(), log: 'Run aborted by user' });
|
||||
|
||||
const hasData = (run.serializableOutput && (
|
||||
(run.serializableOutput.scrapeSchema && run.serializableOutput.scrapeSchema.length > 0) ||
|
||||
(run.serializableOutput.scrapeList && run.serializableOutput.scrapeList.length > 0))) ||
|
||||
(run.binaryOutput && Object.keys(run.binaryOutput).length > 0);
|
||||
|
||||
if (hasData) await triggerIntegrationUpdates(runId, plainRun.robotMetaId);
|
||||
|
||||
try { serverIo.of(plainRun.browserId).emit('run-aborted', { runId, robotName, status: 'aborted', finishedAt: new Date().toLocaleString() }); } catch (_) {}
|
||||
|
||||
try {
|
||||
await new Promise(resolve => setTimeout(resolve, 500));
|
||||
await destroyRemoteBrowser(plainRun.browserId, userId);
|
||||
} catch (cleanupError) {
|
||||
logger.log('warn', `Failed to clean up browser for aborted run ${runId}: ${cleanupError}`);
|
||||
}
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Failed to abort run ${runId}: ${errorMessage}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
const taskList: TaskList = {
|
||||
[QUEUE_NAMES.INITIALIZE_BROWSER_RECORDING]: async (payload: unknown) => {
|
||||
const data = payload as InitializeBrowserData;
|
||||
logger.log('info', `Starting browser initialization job for user: ${data.userId}`);
|
||||
initializeRemoteBrowserForRecording(data.userId);
|
||||
logger.log('info', `Browser initialization triggered for user: ${data.userId}`);
|
||||
},
|
||||
|
||||
[QUEUE_NAMES.DESTROY_BROWSER]: async (payload: unknown) => {
|
||||
const data = payload as DestroyBrowserData;
|
||||
logger.log('info', `Starting browser destruction job for browser: ${data.browserId}`);
|
||||
await destroyRemoteBrowser(data.browserId, data.userId);
|
||||
logger.log('info', `Browser ${data.browserId} destroyed`);
|
||||
},
|
||||
|
||||
[QUEUE_NAMES.INTERPRET_WORKFLOW]: async (payload: unknown) => {
|
||||
const data = payload as InterpretWorkflow;
|
||||
logger.log('info', 'Starting workflow interpretation job');
|
||||
await interpretWholeWorkflow(data.userId);
|
||||
logger.log('info', 'Workflow interpretation job completed');
|
||||
},
|
||||
|
||||
[QUEUE_NAMES.STOP_INTERPRETATION]: async (payload: unknown) => {
|
||||
const data = payload as StopInterpretWorkflow;
|
||||
logger.log('info', 'Starting stop interpretation job');
|
||||
await stopRunningInterpretation(data.userId);
|
||||
logger.log('info', 'Stop interpretation job completed');
|
||||
},
|
||||
|
||||
[QUEUE_NAMES.EXECUTE_RUN]: async (payload: unknown) => {
|
||||
await processRunExecution(payload as ExecuteRunData);
|
||||
},
|
||||
|
||||
[QUEUE_NAMES.ABORT_RUN]: async (payload: unknown) => {
|
||||
const data = payload as AbortRunData;
|
||||
logger.log('info', `Processing abort request for run ${data.runId} by user ${data.userId}`);
|
||||
await abortRun(data.runId, data.userId);
|
||||
},
|
||||
|
||||
[QUEUE_NAMES.SCHEDULED_WORKFLOW]: async (payload: unknown) => {
|
||||
const data = payload as ScheduledWorkflowData;
|
||||
logger.log('info', `Processing scheduled workflow for robot ${data.robotMetaId}`);
|
||||
await handleRunRecording(data.robotMetaId, data.userId);
|
||||
},
|
||||
};
|
||||
|
||||
let runner: Runner | null = null;
|
||||
let runnerPool: Pool | null = null;
|
||||
let workersStarted = false;
|
||||
|
||||
export async function startWorkers(): Promise<void> {
|
||||
if (workersStarted) {
|
||||
logger.log('warn', 'Graphile Worker runner already started, skipping...');
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
logger.log('info', 'Starting Graphile Worker runner...');
|
||||
|
||||
runnerPool = new Pool({
|
||||
connectionString,
|
||||
max: TOTAL_CONCURRENCY + 2,
|
||||
ssl: useSSL ? true : undefined,
|
||||
});
|
||||
|
||||
runner = await run({
|
||||
pgPool: runnerPool,
|
||||
concurrency: TOTAL_CONCURRENCY,
|
||||
noHandleSignals: true,
|
||||
taskList,
|
||||
});
|
||||
|
||||
runner.events.on('job:error', ({ job, error }: { job: any; error: unknown }) => {
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Job error for job ${job?.id} (${job?.task_identifier}): ${msg}`);
|
||||
});
|
||||
|
||||
workersStarted = true;
|
||||
logger.log('info', `Graphile Worker runner started (concurrency: ${TOTAL_CONCURRENCY})`);
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.log('error', `Failed to start Graphile Worker runner: ${errorMessage}`);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
export async function stopWorkers(): Promise<void> {
|
||||
if (!workersStarted || !runner) return;
|
||||
|
||||
try {
|
||||
await runner.stop();
|
||||
runner = null;
|
||||
} catch (error: any) {
|
||||
logger.log('warn', `Error stopping Graphile Worker runner: ${error.message}`);
|
||||
}
|
||||
|
||||
try {
|
||||
if (runnerPool) {
|
||||
await runnerPool.end();
|
||||
runnerPool = null;
|
||||
}
|
||||
} catch (error: any) {
|
||||
logger.log('warn', `Error closing runner pool: ${error.message}`);
|
||||
}
|
||||
|
||||
workersStarted = false;
|
||||
logger.log('info', 'Graphile Worker runner stopped');
|
||||
}
|
||||
@@ -32,7 +32,6 @@ import { createScrapeRobot, createLLMRobot, createAndRunRecording, createCrawlRo
|
||||
import { AuthContext } from '../../../context/auth';
|
||||
import { DEFAULT_OUTPUT_FORMATS, DOC_PARSE_FORMAT_OPTIONS, OUTPUT_FORMAT_LABELS, OUTPUT_FORMAT_OPTIONS, OutputFormats } from '../../../constants/outputFormats';
|
||||
|
||||
|
||||
interface TabPanelProps {
|
||||
children?: React.ReactNode;
|
||||
index: number;
|
||||
@@ -800,7 +799,7 @@ const RobotCreate: React.FC = () => {
|
||||
/>
|
||||
|
||||
<Box sx={{ width: '100%', display: 'flex', justifyContent: 'flex-start' }}>
|
||||
<FormControl sx={{ mb: 2, width: '300px' }}>
|
||||
<FormControl fullWidth sx={{ mb: 2 }}>
|
||||
<InputLabel id="output-formats-label">Output Formats *</InputLabel>
|
||||
<Select
|
||||
labelId="output-formats-label"
|
||||
@@ -1019,7 +1018,7 @@ const RobotCreate: React.FC = () => {
|
||||
/>
|
||||
|
||||
<Box sx={{ width: '100%', display: 'flex', justifyContent: 'flex-start', mb: 2 }}>
|
||||
<FormControl sx={{ width: '300px' }}>
|
||||
<FormControl fullWidth sx={{ mb: 2 }}>
|
||||
<InputLabel id="crawl-output-formats-label">Output Formats *</InputLabel>
|
||||
<Select
|
||||
labelId="crawl-output-formats-label"
|
||||
@@ -1244,7 +1243,7 @@ const RobotCreate: React.FC = () => {
|
||||
|
||||
{searchMode === 'scrape' ? (
|
||||
<Box sx={{ width: '100%', display: 'flex', justifyContent: 'flex-start', mb: 2 }}>
|
||||
<FormControl sx={{ width: '300px' }}>
|
||||
<FormControl fullWidth sx={{ mb: 2 }}>
|
||||
<InputLabel id="search-output-formats-label">Output Formats *</InputLabel>
|
||||
<Select
|
||||
labelId="search-output-formats-label"
|
||||
|
||||
@@ -2,7 +2,7 @@ import * as React from 'react';
|
||||
import SwipeableDrawer from '@mui/material/SwipeableDrawer';
|
||||
import Typography from '@mui/material/Typography';
|
||||
import { Button, Grid, Box, TextField, IconButton, Tooltip } from '@mui/material';
|
||||
import { useCallback, useEffect, useRef, useState } from "react";
|
||||
import { useCallback, useEffect, useLayoutEffect, useRef, useState } from "react";
|
||||
import { useBrowserDimensionsStore } from "../../context/browserDimensions";
|
||||
import Table from '@mui/material/Table';
|
||||
import TableBody from '@mui/material/TableBody';
|
||||
@@ -58,7 +58,7 @@ export const InterpretationLog: React.FC<InterpretationLogProps> = ({ isOpen, se
|
||||
const previousDataLengths = useRef<Map<number, number>>(new Map());
|
||||
const hasAutoFocusedTextTab = useRef<boolean>(false);
|
||||
const previousGetText = useRef<boolean>(false);
|
||||
const autoFocusedScreenshotIndices = useRef<Set<number>>(new Set());
|
||||
const autoFocusedScreenshotIds = useRef<Set<number>>(new Set());
|
||||
|
||||
const [showPreviewData, setShowPreviewData] = useState<boolean>(false);
|
||||
const userClosedDrawer = useRef<boolean>(false);
|
||||
@@ -78,13 +78,15 @@ export const InterpretationLog: React.FC<InterpretationLogProps> = ({ isOpen, se
|
||||
)
|
||||
, [browserSteps, getText, currentTextActionId]);
|
||||
|
||||
const screenshotData = React.useMemo(() => {
|
||||
const screenshotSteps = browserSteps.filter((step): step is ScreenshotStep =>
|
||||
step.type === 'screenshot'
|
||||
);
|
||||
const screenshotSteps = React.useMemo(() =>
|
||||
browserSteps.filter((step): step is ScreenshotStep =>
|
||||
step.type === 'screenshot' && Boolean(step.screenshotData)
|
||||
)
|
||||
, [browserSteps]);
|
||||
|
||||
return screenshotSteps.filter(step => step.screenshotData).map(step => step.screenshotData!);
|
||||
}, [browserSteps]);
|
||||
const screenshotData = React.useMemo(() =>
|
||||
screenshotSteps.map(step => step.screenshotData!)
|
||||
, [screenshotSteps]);
|
||||
|
||||
const toggleDrawer = (newOpen: boolean) => (event: React.KeyboardEvent | React.MouseEvent) => {
|
||||
if (
|
||||
@@ -281,7 +283,6 @@ export const InterpretationLog: React.FC<InterpretationLogProps> = ({ isOpen, se
|
||||
const handleRemoveScreenshotAction = (screenshotId: number, actionId: string | undefined) => {
|
||||
if (!actionId) return;
|
||||
|
||||
const screenshotSteps = browserSteps.filter(step => step.type === 'screenshot' && step.screenshotData);
|
||||
const screenshotIndex = screenshotSteps.findIndex(step => step.id === screenshotId);
|
||||
const screenshotStep = screenshotSteps[screenshotIndex];
|
||||
const screenshotName = screenshotStep?.name || `Screenshot ${screenshotIndex + 1}`;
|
||||
@@ -343,7 +344,6 @@ export const InterpretationLog: React.FC<InterpretationLogProps> = ({ isOpen, se
|
||||
setActiveTab(availableTabs.findIndex(tab => tab.id === 'captureText'));
|
||||
} else if (hasNewScreenshotData && availableTabs.findIndex(tab => tab.id === 'captureScreenshot') !== -1) {
|
||||
setActiveTab(availableTabs.findIndex(tab => tab.id === 'captureScreenshot'));
|
||||
setActiveScreenshotTab(screenshotData.length - 1);
|
||||
}
|
||||
}, [captureListData.length, captureTextData.length, screenshotData.length]);
|
||||
|
||||
@@ -351,7 +351,7 @@ export const InterpretationLog: React.FC<InterpretationLogProps> = ({ isOpen, se
|
||||
browserStepsRef.current = browserSteps;
|
||||
}, [browserSteps]);
|
||||
|
||||
useEffect(() => {
|
||||
useLayoutEffect(() => {
|
||||
if (captureListData.length > 0 || captureTextData.length > 0 || screenshotData.length > 0) {
|
||||
setShowPreviewData(true);
|
||||
} else {
|
||||
@@ -377,7 +377,7 @@ export const InterpretationLog: React.FC<InterpretationLogProps> = ({ isOpen, se
|
||||
setShowPreviewData(false);
|
||||
autoFocusedListIds.current.clear();
|
||||
previousDataLengths.current.clear();
|
||||
autoFocusedScreenshotIndices.current.clear();
|
||||
autoFocusedScreenshotIds.current.clear();
|
||||
userClosedDrawer.current = false;
|
||||
lastListDataLength.current = 0;
|
||||
lastTextDataLength.current = 0;
|
||||
@@ -480,20 +480,6 @@ export const InterpretationLog: React.FC<InterpretationLogProps> = ({ isOpen, se
|
||||
const screenshotTabIndex = getAvailableTabs().findIndex(tab => tab.id === "captureScreenshot");
|
||||
if (screenshotTabIndex !== -1) {
|
||||
setActiveTab(screenshotTabIndex);
|
||||
const latestIndex = screenshotData.length - 1;
|
||||
setActiveScreenshotTab(latestIndex);
|
||||
|
||||
if (!autoFocusedScreenshotIndices.current.has(latestIndex)) {
|
||||
autoFocusedScreenshotIndices.current.add(latestIndex);
|
||||
setTimeout(() => {
|
||||
const screenshotSteps = browserSteps.filter(step => step.type === "screenshot");
|
||||
const latestScreenshotStep = screenshotSteps[latestIndex];
|
||||
if (latestScreenshotStep) {
|
||||
const screenshotName = latestScreenshotStep.name || `Screenshot ${latestIndex + 1}`;
|
||||
startEdit(latestScreenshotStep.id, 'screenshot', screenshotName);
|
||||
}
|
||||
}, 300);
|
||||
}
|
||||
}
|
||||
}
|
||||
}, 100);
|
||||
@@ -523,12 +509,22 @@ export const InterpretationLog: React.FC<InterpretationLogProps> = ({ isOpen, se
|
||||
}
|
||||
}, [captureListData.length, isOpen, captureStage]);
|
||||
|
||||
useEffect(() => {
|
||||
if (screenshotData.length > 0 && isOpen) {
|
||||
const latestScreenshotIndex = screenshotData.length - 1;
|
||||
setActiveScreenshotTab(latestScreenshotIndex);
|
||||
useLayoutEffect(() => {
|
||||
if (screenshotSteps.length > 0 && isOpen) {
|
||||
const latestScreenshotIndex = screenshotSteps.length - 1;
|
||||
const latestScreenshot = screenshotSteps[latestScreenshotIndex];
|
||||
|
||||
if (latestScreenshot && !autoFocusedScreenshotIds.current.has(latestScreenshot.id)) {
|
||||
autoFocusedScreenshotIds.current.add(latestScreenshot.id);
|
||||
setActiveScreenshotTab(latestScreenshotIndex);
|
||||
cancelEdit();
|
||||
|
||||
setTimeout(() => {
|
||||
startEdit(latestScreenshot.id, 'screenshot', latestScreenshot.name || `Screenshot ${latestScreenshotIndex + 1}`);
|
||||
}, 300);
|
||||
}
|
||||
}
|
||||
}, [screenshotData.length, isOpen]);
|
||||
}, [screenshotSteps.length, isOpen]);
|
||||
|
||||
const { darkMode } = useThemeMode();
|
||||
|
||||
@@ -940,7 +936,6 @@ export const InterpretationLog: React.FC<InterpretationLogProps> = ({ isOpen, se
|
||||
}}
|
||||
>
|
||||
{(() => {
|
||||
const screenshotSteps = browserSteps.filter(step => step.type === 'screenshot' && step.screenshotData) as Array<{ id: number; name?: string; type: 'screenshot'; fullPage: boolean; actionId?: string; screenshotData?: string }>;
|
||||
return screenshotData.map((screenshot, index) => {
|
||||
const screenshotStep = screenshotSteps[index];
|
||||
if (!screenshotStep) return null;
|
||||
|
||||
+373
-346
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user