feat(worker): split ai workloads

Keep retried embeddings pending and validate every download redirect.

Author: nguyenvantruc92

.agents/skills/worker-development/SKILL.md

@@ -21,13 +21,22 @@ Workers run as separate Node processes in `apps/worker/`. They consume jobs from
 | integration | `integration` | `src/integration/worker.ts` |
 | chat | `chat` | `src/chat/worker.ts` |
 | ai-agent | `aiAgent` | `src/ai-agent/worker.ts` |
+| heavy | `heavy` | `src/heavy/worker.ts` |
 | default | `default` | `src/default/worker.ts` |
 | trigger | `trigger` | `src/trigger/worker.ts` |
 | webhook | `webhook` | `src/webhook/worker.ts` |
 | schedule | (cron) | `src/schedule/worker.ts` |
 | sequence-scheduler | Kafka | `src/sequence-scheduler/worker*.ts` |
 | notification | `notification` | `src/notification/worker.ts` |
 
+The `heavy` queue/worker is a **workload-class** queue, not a domain queue:
+use it for bounded but RAM/CPU/I/O/model-heavy jobs that should not occupy
+latency-sensitive domain workers. AI file processing, media generation,
+speech/text conversion, document extraction, and image analysis are current
+tenants. Future heavy workloads can join this queue with their own
+`src/heavy/handlers/<domain-or-capability>/` handler area when the same
+resource-isolation tradeoff applies.
+
 ## Creating a New Queue
 
 ### 1. Define Queue Name

apps/builder/__tests__/ai-files-actions.test.ts

@@ -46,8 +46,9 @@ vi.mock("@chatbotx.io/utils", () => ({
 }))
 
 vi.mock("@chatbotx.io/worker-config", () => ({
-  AIJobAction: { processAIFile: "processAIFile" },
-  aiAgentQueue: { add: mocks.queueAdd },
+  HeavyJobAction: { processAIFile: "processAIFile" },
+  getHeavyJobOptions: () => ({}),
+  heavyQueue: { add: mocks.queueAdd },
 }))
 
 vi.mock("next-intl/server", () => ({
@@ -104,6 +105,21 @@ beforeEach(() => {
 })
 
 describe("Knowledge tab audit messages", () => {
+  test("allows creating a Knowledge with Gemini as the only provider", async () => {
+    mocks.findFirstOpenai.mockResolvedValue(undefined)
+    mocks.findFirstGemini.mockResolvedValue({ id: "gemini-1" })
+
+    await (
+      createAIFileAction as unknown as ActionHandler<{ name: string }, [string]>
+    )({
+      parsedInput: { name: "manual.pdf" },
+      bindArgsParsedInputs: [workspaceId],
+    })
+
+    expect(mocks.insertReturning).toHaveBeenCalled()
+    expect(mocks.queueAdd).toHaveBeenCalled()
+  })
+
   test("createAIFileAction logs created a new Knowledge by id", async () => {
     await (
       createAIFileAction as unknown as ActionHandler<{ name: string }, [string]>
@@ -117,6 +133,14 @@ describe("Knowledge tab audit messages", () => {
       action: "create",
       detail: "created a new Knowledge (#file-1)",
     })
+    expect(mocks.queueAdd).toHaveBeenCalledWith(
+      "processAIFile",
+      {
+        type: "processAIFile",
+        data: { aiFileId: "file-1" },
+      },
+      { jobId: "heavy-ai-file-file-1" },
+    )
   })
 
   test("deleteAIFile logs deleted a Knowledge by id", async () => {

apps/builder/__tests__/workspace-owner-quota.test.ts

@@ -57,7 +57,7 @@ vi.mock("@chatbotx.io/utils", () => {
       return () => proxy
     },
   })
-  return { zodBigintAsString: () => proxy }
+  return { zodBigintAsString: () => proxy, zodUrlWithVariables: () => proxy }
 })
 
 vi.mock("@/env", () => ({ isCloud }))

apps/builder/src/app/developer/queues/[[...path]]/route.ts

@@ -7,6 +7,7 @@ import {
   chatQueue,
   defaultQueue,
   getSequenceSchedulerQueue,
+  heavyQueue,
   integrationQueue,
   quotaQueue,
   scheduleQueue,
@@ -48,6 +49,7 @@ async function buildApp() {
   const queues = [
     chatQueue,
     aiAgentQueue,
+    heavyQueue,
     triggerQueue,
     webhookQueue,
     defaultQueue,

apps/builder/src/features/ai-files/actions/create-ai-file.action.ts

@@ -5,7 +5,11 @@ import { ChatbotXException } from "@chatbotx.io/business/errors"
 import { db } from "@chatbotx.io/database/client"
 import { aiFileModel } from "@chatbotx.io/database/schema"
 import { createId } from "@chatbotx.io/utils"
-import { AIJobAction, aiAgentQueue } from "@chatbotx.io/worker-config"
+import {
+  getHeavyJobOptions,
+  HeavyJobAction,
+  heavyQueue,
+} from "@chatbotx.io/worker-config"
 import { getTranslations } from "next-intl/server"
 import { workspaceIdrequestParams } from "@/features/common/schema"
 import { workspaceActionClient } from "@/lib/safe-action"
@@ -41,12 +45,19 @@ export const createAIFileAction = workspaceActionClient
       .returning({ id: aiFileModel.id })
 
     // Enqueue embedding job right after creation
-    await aiAgentQueue.add(AIJobAction.processAIFile, {
-      type: AIJobAction.processAIFile,
-      data: {
-        aiFileId: created[0].id,
+    await heavyQueue.add(
+      HeavyJobAction.processAIFile,
+      {
+        type: HeavyJobAction.processAIFile,
+        data: {
+          aiFileId: created[0].id,
+        },
       },
-    })
+      {
+        ...getHeavyJobOptions(HeavyJobAction.processAIFile),
+        jobId: `heavy-ai-file-${created[0].id}`,
+      },
+    )
 
     await auditService.record({
       workspaceId,

apps/builder/src/features/flows/react-flow/steps/ai-generate-image/components/ai-model-dialog.tsx

@@ -36,12 +36,13 @@ export const AIModelDialog = ({ parentName }: AIModelDialogProps) => {
     getValues: getParentValues,
     setValue: setParentValue,
   } = useFormContext()
-  const provider = useWatch({ name: `${parentName}.provider`, control })
 
   const form = useForm({
     resolver: zodResolver(aiGenerateImageSchema),
     defaultValues: getParentValues(parentName),
   })
+  const provider = useWatch({ name: `${parentName}.provider`, control })
+  const model = useWatch({ control: form.control, name: "model" })
 
   useEffect(() => {
     if (!open) {
@@ -96,7 +97,7 @@ export const AIModelDialog = ({ parentName }: AIModelDialogProps) => {
 
               {isOpenAI && <QualitySelect name="quality" />}
 
-              <SizeSelect name="size" provider={provider} />
+              <SizeSelect model={model ?? ""} name="size" provider={provider} />
 
               <CustomFieldSelect
                 allowCreate={true}

apps/builder/src/features/flows/react-flow/steps/ai-generate-image/constants.tsx

@@ -32,33 +32,45 @@ export const QualitySelect = (props: QualitySelectProps) => {
 
 type SizeSelectProps = {
   name: string
+  model: string
   required?: boolean
   provider: AIGenerateImageProvider
 }
 
 export const SizeSelect = (props: SizeSelectProps) => {
-  const { provider, ...rest } = props
+  const { model, provider, ...rest } = props
   const t = useTranslations()
 
+  const isGPTImage =
+    model.startsWith("gpt-image") || model.startsWith("chatgpt-image")
+
   const optionsMap = useMemo<Record<AIGenerateImageProvider, SelectOption[]>>(
     () => ({
       openai: [
         { label: t("fields.size.options.auto"), value: "auto" },
-        { label: t("fields.size.options.square1024"), value: "1024x1024" },
-        {
-          label: t("fields.size.options.landscape1536x1024"),
-          value: "1536x1024",
-        },
-        {
-          label: t("fields.size.options.portrait1024x1536"),
-          value: "1024x1536",
-        },
-        { label: t("fields.size.options.dalle2_256"), value: "256x256" },
-        { label: t("fields.size.options.dalle2_512"), value: "512x512" },
-        {
-          label: t("fields.size.options.dalle3_1792x1024"),
-          value: "1792x1024",
-        },
+        ...(isGPTImage
+          ? [
+              {
+                label: t("fields.size.options.square1024"),
+                value: "1024x1024",
+              },
+              {
+                label: t("fields.size.options.landscape1536x1024"),
+                value: "1536x1024",
+              },
+              {
+                label: t("fields.size.options.portrait1024x1536"),
+                value: "1024x1536",
+              },
+            ]
+          : [
+              { label: t("fields.size.options.dalle2_256"), value: "256x256" },
+              { label: t("fields.size.options.dalle2_512"), value: "512x512" },
+              {
+                label: t("fields.size.options.dalle3_1792x1024"),
+                value: "1792x1024",
+              },
+            ]),
       ],
       gemini: [
         { label: t("fields.size.options.auto"), value: "auto" },
@@ -69,7 +81,7 @@ export const SizeSelect = (props: SizeSelectProps) => {
         { label: "16:9", value: "16:9" },
       ],
     }),
-    [t],
+    [isGPTImage, t],
   )
 
   return (