feat: schedule routes using pgboss queue

This commit is contained in:
Rohit
2025-03-28 17:24:58 +05:30
parent d891aa14ac
commit 3977b6feb4

View File

@@ -22,7 +22,7 @@ import { encrypt, decrypt } from '../utils/auth';
import { WorkflowFile } from 'maxun-core'; import { WorkflowFile } from 'maxun-core';
import { Page } from 'playwright'; import { Page } from 'playwright';
import { airtableUpdateTasks, processAirtableUpdates } from '../workflow-management/integrations/airtable'; import { airtableUpdateTasks, processAirtableUpdates } from '../workflow-management/integrations/airtable';
import { pgBoss } from '../pgboss-worker'; import { cancelScheduledWorkflow, pgBoss, scheduleWorkflow } from '../pgboss-worker';
chromium.use(stealthPlugin()); chromium.use(stealthPlugin());
export const router = Router(); export const router = Router();
@@ -792,17 +792,13 @@ router.put('/schedule/:id/', requireSignIn, async (req: AuthenticatedRequest, re
return res.status(401).json({ error: 'Unauthorized' }); return res.status(401).json({ error: 'Unauthorized' });
} }
// Create the job in the queue with the cron expression try {
const job = await workflowQueue.add( await cancelScheduledWorkflow(id);
'run workflow', } catch (cancelError) {
{ id, runId: uuid(), userId: req.user.id }, logger.log('warn', `Failed to cancel existing schedule for robot ${id}: ${cancelError}`);
{ }
repeat: {
pattern: cronExpression, const jobId = await scheduleWorkflow(id, req.user.id, cronExpression, timezone);
tz: timezone,
},
}
);
const nextRunAt = computeNextRun(cronExpression, timezone); const nextRunAt = computeNextRun(cronExpression, timezone);
@@ -877,12 +873,12 @@ router.delete('/schedule/:id', requireSignIn, async (req: AuthenticatedRequest,
return res.status(404).json({ error: 'Robot not found' }); return res.status(404).json({ error: 'Robot not found' });
} }
// Remove existing job from queue if it exists // Cancel the scheduled job in PgBoss
const existingJobs = await workflowQueue.getJobs(['delayed', 'waiting']); try {
for (const job of existingJobs) { await cancelScheduledWorkflow(id);
if (job.data.id === id) { } catch (error) {
await job.remove(); logger.log('error', `Error cancelling scheduled job for robot ${id}: ${error}`);
} // Continue with robot update even if cancellation fails
} }
// Delete the schedule from the robot // Delete the schedule from the robot