Skip to content

Commit a87ac8a

Browse files
feat(worker): split ai workloads
1 parent f92f4f8 commit a87ac8a

106 files changed

Lines changed: 6227 additions & 1364 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,13 +21,22 @@ Workers run as separate Node processes in `apps/worker/`. They consume jobs from
2121
| integration | `integration` | `src/integration/worker.ts` |
2222
| chat | `chat` | `src/chat/worker.ts` |
2323
| ai-agent | `aiAgent` | `src/ai-agent/worker.ts` |
24+
| heavy | `heavy` | `src/heavy/worker.ts` |
2425
| default | `default` | `src/default/worker.ts` |
2526
| trigger | `trigger` | `src/trigger/worker.ts` |
2627
| webhook | `webhook` | `src/webhook/worker.ts` |
2728
| schedule | (cron) | `src/schedule/worker.ts` |
2829
| sequence-scheduler | Kafka | `src/sequence-scheduler/worker*.ts` |
2930
| notification | `notification` | `src/notification/worker.ts` |
3031

32+
The `heavy` queue/worker is a **workload-class** queue, not a domain queue:
33+
use it for bounded but RAM/CPU/I/O/model-heavy jobs that should not occupy
34+
latency-sensitive domain workers. AI file processing, media generation,
35+
speech/text conversion, document extraction, and image analysis are current
36+
tenants. Future heavy workloads can join this queue with their own
37+
`src/heavy/handlers/<domain-or-capability>/` handler area when the same
38+
resource-isolation tradeoff applies.
39+
3140
## Creating a New Queue
3241

3342
### 1. Define Queue Name

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

Lines changed: 26 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,8 +46,9 @@ vi.mock("@chatbotx.io/utils", () => ({
4646
}))
4747

4848
vi.mock("@chatbotx.io/worker-config", () => ({
49-
AIJobAction: { processAIFile: "processAIFile" },
50-
aiAgentQueue: { add: mocks.queueAdd },
49+
HeavyJobAction: { processAIFile: "processAIFile" },
50+
getHeavyJobOptions: () => ({}),
51+
heavyQueue: { add: mocks.queueAdd },
5152
}))
5253

5354
vi.mock("next-intl/server", () => ({
@@ -104,6 +105,21 @@ beforeEach(() => {
104105
})
105106

106107
describe("Knowledge tab audit messages", () => {
108+
test("allows creating a Knowledge with Gemini as the only provider", async () => {
109+
mocks.findFirstOpenai.mockResolvedValue(undefined)
110+
mocks.findFirstGemini.mockResolvedValue({ id: "gemini-1" })
111+
112+
await (
113+
createAIFileAction as unknown as ActionHandler<{ name: string }, [string]>
114+
)({
115+
parsedInput: { name: "manual.pdf" },
116+
bindArgsParsedInputs: [workspaceId],
117+
})
118+
119+
expect(mocks.insertReturning).toHaveBeenCalled()
120+
expect(mocks.queueAdd).toHaveBeenCalled()
121+
})
122+
107123
test("createAIFileAction logs created a new Knowledge by id", async () => {
108124
await (
109125
createAIFileAction as unknown as ActionHandler<{ name: string }, [string]>
@@ -117,6 +133,14 @@ describe("Knowledge tab audit messages", () => {
117133
action: "create",
118134
detail: "created a new Knowledge (#file-1)",
119135
})
136+
expect(mocks.queueAdd).toHaveBeenCalledWith(
137+
"processAIFile",
138+
{
139+
type: "processAIFile",
140+
data: { aiFileId: "file-1" },
141+
},
142+
{ jobId: "heavy-ai-file-file-1" },
143+
)
120144
})
121145

122146
test("deleteAIFile logs deleted a Knowledge by id", async () => {

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,7 @@ vi.mock("@chatbotx.io/utils", () => {
7676
return () => proxy
7777
},
7878
})
79-
return { zodBigintAsString: () => proxy }
79+
return { zodBigintAsString: () => proxy, zodUrlWithVariables: () => proxy }
8080
})
8181

8282
vi.mock("@/env", () => ({ isCloud }))

apps/builder/messages/en.json

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3586,6 +3586,7 @@
35863586
"validation": {
35873587
"maxSize": "File size exceeds maximum limit",
35883588
"maxItemsReached": "Maximum {max} {feature} allowed per workspace",
3589+
"maxCharacters": "Maximum {max} characters.",
35893590
"invalidApiKey": "Invalid API key",
35903591
"maxMustBeGreaterThanMin": "{maxField} must be greater than or equal to {minField}"
35913592
},

apps/builder/messages/vi.json

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3399,6 +3399,7 @@
33993399
"validation": {
34003400
"maxSize": "Kích thước tệp vượt quá giới hạn tối đa",
34013401
"maxItemsReached": "Tối đa {max} {feature} cho mỗi workspace",
3402+
"maxCharacters": "Tối đa {max} ký tự.",
34023403
"invalidApiKey": "API key không hợp lệ",
34033404
"maxMustBeGreaterThanMin": "{maxField} phải lớn hơn hoặc bằng {minField}"
34043405
},

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import {
77
chatQueue,
88
defaultQueue,
99
getSequenceSchedulerQueue,
10+
heavyQueue,
1011
integrationQueue,
1112
quotaQueue,
1213
scheduleQueue,
@@ -48,6 +49,7 @@ async function buildApp() {
4849
const queues = [
4950
chatQueue,
5051
aiAgentQueue,
52+
heavyQueue,
5153
triggerQueue,
5254
webhookQueue,
5355
defaultQueue,

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

Lines changed: 17 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,11 @@ import { ChatbotXException } from "@chatbotx.io/business/errors"
55
import { db } from "@chatbotx.io/database/client"
66
import { aiFileModel } from "@chatbotx.io/database/schema"
77
import { createId } from "@chatbotx.io/utils"
8-
import { AIJobAction, aiAgentQueue } from "@chatbotx.io/worker-config"
8+
import {
9+
getHeavyJobOptions,
10+
HeavyJobAction,
11+
heavyQueue,
12+
} from "@chatbotx.io/worker-config"
913
import { getTranslations } from "next-intl/server"
1014
import { workspaceIdrequestParams } from "@/features/common/schema"
1115
import { workspaceActionClient } from "@/lib/safe-action"
@@ -41,12 +45,19 @@ export const createAIFileAction = workspaceActionClient
4145
.returning({ id: aiFileModel.id })
4246

4347
// Enqueue embedding job right after creation
44-
await aiAgentQueue.add(AIJobAction.processAIFile, {
45-
type: AIJobAction.processAIFile,
46-
data: {
47-
aiFileId: created[0].id,
48+
await heavyQueue.add(
49+
HeavyJobAction.processAIFile,
50+
{
51+
type: HeavyJobAction.processAIFile,
52+
data: {
53+
aiFileId: created[0].id,
54+
},
4855
},
49-
})
56+
{
57+
...getHeavyJobOptions(HeavyJobAction.processAIFile),
58+
jobId: `heavy-ai-file-${created[0].id}`,
59+
},
60+
)
5061

5162
await auditService.record({
5263
workspaceId,

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

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,12 +36,13 @@ export const AIModelDialog = ({ parentName }: AIModelDialogProps) => {
3636
getValues: getParentValues,
3737
setValue: setParentValue,
3838
} = useFormContext()
39-
const provider = useWatch({ name: `${parentName}.provider`, control })
4039

4140
const form = useForm({
4241
resolver: zodResolver(aiGenerateImageSchema),
4342
defaultValues: getParentValues(parentName),
4443
})
44+
const provider = useWatch({ name: `${parentName}.provider`, control })
45+
const model = useWatch({ control: form.control, name: "model" })
4546

4647
useEffect(() => {
4748
if (!open) {
@@ -96,7 +97,7 @@ export const AIModelDialog = ({ parentName }: AIModelDialogProps) => {
9697

9798
{isOpenAI && <QualitySelect name="quality" />}
9899

99-
<SizeSelect name="size" provider={provider} />
100+
<SizeSelect model={model ?? ""} name="size" provider={provider} />
100101

101102
<CustomFieldSelect
102103
allowCreate={true}

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

Lines changed: 29 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -32,33 +32,45 @@ export const QualitySelect = (props: QualitySelectProps) => {
3232

3333
type SizeSelectProps = {
3434
name: string
35+
model: string
3536
required?: boolean
3637
provider: AIGenerateImageProvider
3738
}
3839

3940
export const SizeSelect = (props: SizeSelectProps) => {
40-
const { provider, ...rest } = props
41+
const { model, provider, ...rest } = props
4142
const t = useTranslations()
4243

44+
const isGPTImage =
45+
model.startsWith("gpt-image") || model.startsWith("chatgpt-image")
46+
4347
const optionsMap = useMemo<Record<AIGenerateImageProvider, SelectOption[]>>(
4448
() => ({
4549
openai: [
4650
{ label: t("fields.size.options.auto"), value: "auto" },
47-
{ label: t("fields.size.options.square1024"), value: "1024x1024" },
48-
{
49-
label: t("fields.size.options.landscape1536x1024"),
50-
value: "1536x1024",
51-
},
52-
{
53-
label: t("fields.size.options.portrait1024x1536"),
54-
value: "1024x1536",
55-
},
56-
{ label: t("fields.size.options.dalle2_256"), value: "256x256" },
57-
{ label: t("fields.size.options.dalle2_512"), value: "512x512" },
58-
{
59-
label: t("fields.size.options.dalle3_1792x1024"),
60-
value: "1792x1024",
61-
},
51+
...(isGPTImage
52+
? [
53+
{
54+
label: t("fields.size.options.square1024"),
55+
value: "1024x1024",
56+
},
57+
{
58+
label: t("fields.size.options.landscape1536x1024"),
59+
value: "1536x1024",
60+
},
61+
{
62+
label: t("fields.size.options.portrait1024x1536"),
63+
value: "1024x1536",
64+
},
65+
]
66+
: [
67+
{ label: t("fields.size.options.dalle2_256"), value: "256x256" },
68+
{ label: t("fields.size.options.dalle2_512"), value: "512x512" },
69+
{
70+
label: t("fields.size.options.dalle3_1792x1024"),
71+
value: "1792x1024",
72+
},
73+
]),
6274
],
6375
gemini: [
6476
{ label: t("fields.size.options.auto"), value: "auto" },
@@ -69,7 +81,7 @@ export const SizeSelect = (props: SizeSelectProps) => {
6981
{ label: "16:9", value: "16:9" },
7082
],
7183
}),
72-
[t],
84+
[isGPTImage, t],
7385
)
7486

7587
return (

apps/builder/src/features/flows/react-flow/steps/ai-text-to-speech/components/ai-model-dialog.tsx

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,10 @@
11
"use client"
22

33
import { openAITTSVoiceTypes } from "@chatbotx.io/ai"
4-
import { aiTextToSpeechSchema } from "@chatbotx.io/flow-config"
4+
import {
5+
AI_TEXT_TO_SPEECH_MESSAGE_MAX_LENGTH,
6+
aiTextToSpeechSchema,
7+
} from "@chatbotx.io/flow-config"
58
import { InputField } from "@chatbotx.io/ui/components/form/input-field"
69
import { SelectField } from "@chatbotx.io/ui/components/form/select-field"
710
import { Button } from "@chatbotx.io/ui/components/ui/button"
@@ -57,6 +60,22 @@ export const AIModelDialog = ({ parentName }: AIModelDialogProps) => {
5760

5861
const handleSubmit = form.handleSubmit((values) => {
5962
const currentValues = getParentValues(parentName)
63+
const currentMessage =
64+
typeof currentValues.message === "string"
65+
? currentValues.message.trim()
66+
: ""
67+
const message = values.message.trim()
68+
if (
69+
message.length > AI_TEXT_TO_SPEECH_MESSAGE_MAX_LENGTH &&
70+
message !== currentMessage
71+
) {
72+
form.setError("message", {
73+
message: t("validation.maxCharacters", {
74+
max: AI_TEXT_TO_SPEECH_MESSAGE_MAX_LENGTH,
75+
}),
76+
})
77+
return
78+
}
6079
setParentValue(parentName, {
6180
...currentValues,
6281
...values,
@@ -88,6 +107,9 @@ export const AIModelDialog = ({ parentName }: AIModelDialogProps) => {
88107
<form className="flex flex-col space-y-6" onSubmit={handleSubmit}>
89108
<div className="flex max-h-[calc(100vh-200px)] flex-col space-y-6 overflow-y-auto">
90109
<TiptapEditorField
110+
description={t("validation.maxCharacters", {
111+
max: AI_TEXT_TO_SPEECH_MESSAGE_MAX_LENGTH,
112+
})}
91113
label={t("fields.inputText.label")}
92114
name="message"
93115
required

0 commit comments

Comments
 (0)