Merge pull request #1359 from trycompai/mariano/tasks-limits-release

feat: enhance onboarding tasks with increased concurrency and retry l…

Author: Marfuen

apps/app/src/jobs/tasks/onboarding/generate-risk-mitigation.ts

@@ -8,12 +8,15 @@ import {
 } from './onboard-organization-helpers';
 
 // Queues
-const riskMitigationQueue = queue({ name: 'risk-mitigations', concurrencyLimit: 10 });
-const riskMitigationFanoutQueue = queue({ name: 'risk-mitigations-fanout', concurrencyLimit: 3 });
+const riskMitigationQueue = queue({ name: 'risk-mitigations', concurrencyLimit: 100 });
+const riskMitigationFanoutQueue = queue({ name: 'risk-mitigations-fanout', concurrencyLimit: 100 });
 
 export const generateRiskMitigation = task({
   id: 'generate-risk-mitigation',
   queue: riskMitigationQueue,
+  retry: {
+    maxAttempts: 5,
+  },
   run: async (payload: { organizationId: string; riskId: string }) => {
     const { organizationId, riskId } = payload;
     logger.info(`Generating risk mitigation for risk ${riskId} in org ${organizationId}`);

apps/app/src/jobs/tasks/onboarding/generate-vendor-mitigation.ts

@@ -8,15 +8,18 @@ import {
 } from './onboard-organization-helpers';
 
 // Queues
-const vendorMitigationQueue = queue({ name: 'vendor-risk-mitigations', concurrencyLimit: 10 });
+const vendorMitigationQueue = queue({ name: 'vendor-risk-mitigations', concurrencyLimit: 100 });
 const vendorMitigationFanoutQueue = queue({
   name: 'vendor-risk-mitigations-fanout',
-  concurrencyLimit: 3,
+  concurrencyLimit: 100,
 });
 
 export const generateVendorMitigation = task({
   id: 'generate-vendor-mitigation',
   queue: vendorMitigationQueue,
+  retry: {
+    maxAttempts: 5,
+  },
   run: async (payload: { organizationId: string; vendorId: string }) => {
     const { organizationId, vendorId } = payload;
     logger.info(`Generating vendor mitigation for vendor ${vendorId} in org ${organizationId}`);

apps/app/src/jobs/tasks/onboarding/onboard-organization.ts

@@ -11,11 +11,14 @@ import {
 } from './onboard-organization-helpers';
 
 // v4 queues must be declared in advance
-const onboardOrgQueue = queue({ name: 'onboard-organization', concurrencyLimit: 10 });
+const onboardOrgQueue = queue({ name: 'onboard-organization', concurrencyLimit: 100 });
 
 export const onboardOrganization = task({
   id: 'onboard-organization',
   queue: onboardOrgQueue,
+  retry: {
+    maxAttempts: 3,
+  },
   run: async (payload: { organizationId: string }) => {
     logger.info(`Start onboarding organization ${payload.organizationId}`);
 

apps/app/src/jobs/tasks/onboarding/update-policies-helpers.ts

@@ -194,7 +194,7 @@ export async function reconcileFormatWithTemplate(
 ): Promise<{ type: 'document'; content: Record<string, unknown>[] }> {
   try {
     const { object } = await generateObject({
-      model: openai('gpt-5-mini'),
+      model: openai('gpt-5-nano'),
       mode: 'json',
       system: `You are an expert policy editor.
 Given an ORIGINAL policy TipTap JSON and a DRAFT TipTap JSON, produce a FINAL TipTap JSON that:
@@ -232,7 +232,7 @@ export async function aiCheckFormatWithTemplate(
 ): Promise<{ isConforming: boolean; reasons: string[] }> {
   try {
     const { object } = await generateObject({
-      model: openai('gpt-5-mini'),
+      model: openai('gpt-5-nano'),
       mode: 'json',
       system: `You are validating policy layout.
 Compare ORIGINAL vs DRAFT (TipTap JSON). Determine if DRAFT conforms to ORIGINAL format:
@@ -457,7 +457,7 @@ export async function generatePolicyContent(prompt: string): Promise<{
 }> {
   try {
     const { object } = await generateObject({
-      model: openai('gpt-5-mini'),
+      model: openai('gpt-5-nano'),
       mode: 'json',
       system: `You are an expert at writing security policies. Generate content directly as TipTap JSON format.
 

apps/app/src/jobs/tasks/onboarding/update-policy.ts

@@ -7,12 +7,15 @@ if (!process.env.OPENAI_API_KEY) {
 }
 
 // v4: define queue ahead of time
-export const updatePolicyQueue = queue({ name: 'update-policy', concurrencyLimit: 5 });
+export const updatePolicyQueue = queue({ name: 'update-policy', concurrencyLimit: 100 });
 
 export const updatePolicy = schemaTask({
   id: 'update-policy',
   maxDuration: 600, // 10 minutes.
   queue: updatePolicyQueue,
+  retry: {
+    maxAttempts: 5,
+  },
   schema: z.object({
     organizationId: z.string(),
     policyId: z.string(),