Skip to content

Commit 73cd10c

Browse files
Sg312Sg312
committed
Jobs 2
1 parent eac8aca commit 73cd10c

File tree

27 files changed

+40668
-187
lines changed

27 files changed

+40668
-187
lines changed

‎apps/sim/app/api/copilot/chat/route.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,7 @@ const ChatMessageSchema = z.object({
7575
)
7676
.optional(),
7777
commands: z.array(z.string()).optional(),
78+
userTimezone: z.string().optional(),
7879
})
7980

8081
/**
@@ -111,6 +112,7 @@ export async function POST(req: NextRequest) {
111112
provider,
112113
contexts,
113114
commands,
115+
userTimezone,
114116
} = ChatMessageSchema.parse(body)
115117

116118
const normalizedContexts = Array.isArray(contexts)
@@ -244,6 +246,7 @@ export async function POST(req: NextRequest) {
244246
prefetch,
245247
implicitFeedback,
246248
userPermission: userPermission ?? undefined,
249+
userTimezone,
247250
},
248251
{
249252
selectedModel,

‎apps/sim/app/api/mothership/chat/route.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ const MothershipMessageSchema = z.object({
3030
chatId: z.string().optional(),
3131
createNewChat: z.boolean().optional().default(false),
3232
fileAttachments: z.array(FileAttachmentSchema).optional(),
33+
userTimezone: z.string().optional(),
3334
contexts: z
3435
.array(
3536
z.object({
@@ -80,6 +81,7 @@ export async function POST(req: NextRequest) {
8081
createNewChat,
8182
fileAttachments,
8283
contexts,
84+
userTimezone,
8385
} = MothershipMessageSchema.parse(body)
8486

8587
const userMessageId = providedMessageId || crypto.randomUUID()
@@ -150,6 +152,7 @@ export async function POST(req: NextRequest) {
150152
chatId: actualChatId,
151153
userPermission: userPermission ?? undefined,
152154
workspaceContext,
155+
userTimezone,
153156
},
154157
{ selectedModel: '' }
155158
)

‎apps/sim/app/api/schedules/execute/route.ts‎

Lines changed: 59 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,27 @@
11
import { db, workflowDeploymentVersion, workflowSchedule } from '@sim/db'
22
import { createLogger } from '@sim/logger'
3-
import { and, eq, isNull, lt, lte, not, or, sql } from 'drizzle-orm'
3+
import { and, eq, isNull, lt, lte, ne, not, or, sql } from 'drizzle-orm'
44
import { type NextRequest, NextResponse } from 'next/server'
55
import { verifyCronAuth } from '@/lib/auth/internal'
66
import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
77
import { generateRequestId } from '@/lib/core/utils/request'
8-
import { executeScheduleJob } from '@/background/schedule-execution'
8+
import { executeJobInline, executeScheduleJob } from '@/background/schedule-execution'
99

1010
export const dynamic = 'force-dynamic'
1111

1212
const logger = createLogger('ScheduledExecuteAPI')
1313

14+
const dueFilter = (queuedAt: Date) =>
15+
and(
16+
lte(workflowSchedule.nextRunAt, queuedAt),
17+
not(eq(workflowSchedule.status, 'disabled')),
18+
ne(workflowSchedule.status, 'completed'),
19+
or(
20+
isNull(workflowSchedule.lastQueuedAt),
21+
lt(workflowSchedule.lastQueuedAt, workflowSchedule.nextRunAt)
22+
)
23+
)
24+
1425
export async function GET(request: NextRequest) {
1526
const requestId = generateRequestId()
1627
logger.info(`[${requestId}] Scheduled execution triggered at ${new Date().toISOString()}`)
@@ -23,20 +34,14 @@ export async function GET(request: NextRequest) {
2334
const queuedAt = new Date()
2435

2536
try {
37+
// Workflow schedules (require active deployment)
2638
const dueSchedules = await db
2739
.update(workflowSchedule)
28-
.set({
29-
lastQueuedAt: queuedAt,
30-
updatedAt: queuedAt,
31-
})
40+
.set({ lastQueuedAt: queuedAt, updatedAt: queuedAt })
3241
.where(
3342
and(
34-
lte(workflowSchedule.nextRunAt, queuedAt),
35-
not(eq(workflowSchedule.status, 'disabled')),
36-
or(
37-
isNull(workflowSchedule.lastQueuedAt),
38-
lt(workflowSchedule.lastQueuedAt, workflowSchedule.nextRunAt)
39-
),
43+
dueFilter(queuedAt),
44+
or(eq(workflowSchedule.sourceType, 'workflow'), isNull(workflowSchedule.sourceType)),
4045
sql`${workflowSchedule.deploymentVersionId} = (select ${workflowDeploymentVersion.id} from ${workflowDeploymentVersion} where ${workflowDeploymentVersion.workflowId} = ${workflowSchedule.workflowId} and ${workflowDeploymentVersion.isActive} = true)`
4146
)
4247
)
@@ -49,18 +54,33 @@ export async function GET(request: NextRequest) {
4954
failedCount: workflowSchedule.failedCount,
5055
nextRunAt: workflowSchedule.nextRunAt,
5156
lastQueuedAt: workflowSchedule.lastQueuedAt,
57+
sourceType: workflowSchedule.sourceType,
5258
})
5359

54-
logger.info(`[${requestId}] Processing ${dueSchedules.length} due scheduled workflows`)
60+
// Jobs (no deployment, dispatch inline)
61+
const dueJobs = await db
62+
.update(workflowSchedule)
63+
.set({ lastQueuedAt: queuedAt, updatedAt: queuedAt })
64+
.where(and(dueFilter(queuedAt), eq(workflowSchedule.sourceType, 'job')))
65+
.returning({
66+
id: workflowSchedule.id,
67+
cronExpression: workflowSchedule.cronExpression,
68+
failedCount: workflowSchedule.failedCount,
69+
lastQueuedAt: workflowSchedule.lastQueuedAt,
70+
sourceType: workflowSchedule.sourceType,
71+
})
72+
73+
const totalCount = dueSchedules.length + dueJobs.length
74+
logger.info(`[${requestId}] Processing ${totalCount} due items (${dueSchedules.length} schedules, ${dueJobs.length} jobs)`)
5575

5676
const jobQueue = await getJobQueue()
5777

58-
const queuePromises = dueSchedules.map(async (schedule) => {
78+
const schedulePromises = dueSchedules.map(async (schedule) => {
5979
const queueTime = schedule.lastQueuedAt ?? queuedAt
6080

6181
const payload = {
6282
scheduleId: schedule.id,
63-
workflowId: schedule.workflowId,
83+
workflowId: schedule.workflowId!,
6484
blockId: schedule.blockId || undefined,
6585
cronExpression: schedule.cronExpression || undefined,
6686
lastRanAt: schedule.lastRanAt?.toISOString(),
@@ -111,13 +131,34 @@ export async function GET(request: NextRequest) {
111131
}
112132
})
113133

114-
await Promise.allSettled(queuePromises)
134+
// Jobs always execute inline (no TriggerDev)
135+
const jobPromises = dueJobs.map(async (job) => {
136+
const queueTime = job.lastQueuedAt ?? queuedAt
137+
const payload = {
138+
scheduleId: job.id,
139+
cronExpression: job.cronExpression || undefined,
140+
failedCount: job.failedCount || 0,
141+
now: queueTime.toISOString(),
142+
}
143+
144+
void (async () => {
145+
try {
146+
await executeJobInline(payload)
147+
} catch (error) {
148+
logger.error(`[${requestId}] Job execution failed for ${job.id}`, {
149+
error: error instanceof Error ? error.message : String(error),
150+
})
151+
}
152+
})()
153+
})
154+
155+
await Promise.allSettled([...schedulePromises, ...jobPromises])
115156

116-
logger.info(`[${requestId}] Queued ${dueSchedules.length} schedule executions`)
157+
logger.info(`[${requestId}] Processed ${totalCount} items`)
117158

118159
return NextResponse.json({
119160
message: 'Scheduled workflow executions processed',
120-
executedCount: dueSchedules.length,
161+
executedCount: totalCount,
121162
})
122163
} catch (error: any) {
123164
logger.error(`[${requestId}] Error in scheduled execution handler`, error)

‎apps/sim/app/api/schedules/route.ts‎

Lines changed: 55 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -125,43 +125,65 @@ async function handleWorkspaceSchedules(
125125

126126
logger.info(`[${requestId}] Getting all schedules for workspace ${workspaceId}`)
127127

128-
const rows = await db
129-
.select({
130-
schedule: workflowSchedule,
131-
workflowName: workflow.name,
132-
workflowColor: workflow.color,
133-
})
134-
.from(workflowSchedule)
135-
.innerJoin(workflow, eq(workflow.id, workflowSchedule.workflowId))
136-
.leftJoin(
137-
workflowDeploymentVersion,
138-
and(
139-
eq(workflowDeploymentVersion.workflowId, workflowSchedule.workflowId),
140-
eq(workflowDeploymentVersion.isActive, true)
141-
)
142-
)
143-
.where(
144-
and(
145-
eq(workflow.workspaceId, workspaceId),
146-
eq(workflowSchedule.triggerType, 'schedule'),
147-
or(
148-
eq(workflowSchedule.deploymentVersionId, workflowDeploymentVersion.id),
149-
and(isNull(workflowDeploymentVersion.id), isNull(workflowSchedule.deploymentVersionId))
128+
const [workflowRows, jobRows] = await Promise.all([
129+
db
130+
.select({
131+
schedule: workflowSchedule,
132+
workflowName: workflow.name,
133+
workflowColor: workflow.color,
134+
})
135+
.from(workflowSchedule)
136+
.innerJoin(workflow, eq(workflow.id, workflowSchedule.workflowId))
137+
.leftJoin(
138+
workflowDeploymentVersion,
139+
and(
140+
eq(workflowDeploymentVersion.workflowId, workflowSchedule.workflowId),
141+
eq(workflowDeploymentVersion.isActive, true)
150142
)
151143
)
152-
)
144+
.where(
145+
and(
146+
eq(workflow.workspaceId, workspaceId),
147+
eq(workflowSchedule.triggerType, 'schedule'),
148+
or(
149+
eq(workflowSchedule.sourceType, 'workflow'),
150+
isNull(workflowSchedule.sourceType)
151+
),
152+
or(
153+
eq(workflowSchedule.deploymentVersionId, workflowDeploymentVersion.id),
154+
and(
155+
isNull(workflowDeploymentVersion.id),
156+
isNull(workflowSchedule.deploymentVersionId)
157+
)
158+
)
159+
)
160+
),
161+
db
162+
.select({ schedule: workflowSchedule })
163+
.from(workflowSchedule)
164+
.where(
165+
and(
166+
eq(workflowSchedule.sourceWorkspaceId, workspaceId),
167+
eq(workflowSchedule.sourceType, 'job')
168+
)
169+
),
170+
])
153171

154172
const headers = new Headers()
155173
headers.set('Cache-Control', 'no-store, max-age=0')
156174

157-
return NextResponse.json(
158-
{
159-
schedules: rows.map((r) => ({
160-
...r.schedule,
161-
workflowName: r.workflowName,
162-
workflowColor: r.workflowColor,
163-
})),
164-
},
165-
{ headers }
166-
)
175+
const schedules = [
176+
...workflowRows.map((r) => ({
177+
...r.schedule,
178+
workflowName: r.workflowName,
179+
workflowColor: r.workflowColor,
180+
})),
181+
...jobRows.map((r) => ({
182+
...r.schedule,
183+
workflowName: null,
184+
workflowColor: null,
185+
})),
186+
]
187+
188+
return NextResponse.json({ schedules }, { headers })
167189
}

‎apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -365,6 +365,7 @@ export function useChat(workspaceId: string, initialChatId?: string): UseChatRet
365365
userMessageId,
366366
createNewChat: !chatIdRef.current,
367367
...(chatIdRef.current ? { chatId: chatIdRef.current } : {}),
368+
userTimezone: Intl.DateTimeFormat().resolvedOptions().timeZone,
368369
}),
369370
signal: abortController.signal,
370371
})

0 commit comments

Comments
 (0)