mirror of
https://github.com/simstudioai/sim.git
synced 2026-02-04 19:55:08 -05:00
feat(async-jobs): async execution with job queue backends (#3134)
* feat(async-jobs): async execution with job queue backends * added migration * remove unused envvar, remove extraneous comments * ack comment * same for db * added dedicated async envvars for timeouts, updated helm * updated comment * ack comment * migrated routes to be more restful * ack comments --------- Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Waleed
@@ -21,6 +21,7 @@ const UpdateCreatorProfileSchema = z.object({
|
||||
name: z.string().min(1, 'Name is required').max(100, 'Max 100 characters').optional(),
|
||||
profileImageUrl: z.string().optional().or(z.literal('')),
|
||||
details: CreatorProfileDetailsSchema.optional(),
|
||||
verified: z.boolean().optional(), // Verification status (super users only)
|
||||
})
|
||||
|
||||
// Helper to check if user has permission to manage profile
|
||||
@@ -97,11 +98,29 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
|
||||
return NextResponse.json({ error: 'Profile not found' }, { status: 404 })
|
||||
}
|
||||
|
||||
// Check permissions
|
||||
const canEdit = await hasPermission(session.user.id, existing[0])
|
||||
if (!canEdit) {
|
||||
logger.warn(`[${requestId}] User denied permission to update profile: ${id}`)
|
||||
return NextResponse.json({ error: 'Access denied' }, { status: 403 })
|
||||
// Verification changes require super user permission
|
||||
if (data.verified !== undefined) {
|
||||
const { verifyEffectiveSuperUser } = await import('@/lib/templates/permissions')
|
||||
const { effectiveSuperUser } = await verifyEffectiveSuperUser(session.user.id)
|
||||
if (!effectiveSuperUser) {
|
||||
logger.warn(`[${requestId}] Non-super user attempted to change creator verification: ${id}`)
|
||||
return NextResponse.json(
|
||||
{ error: 'Only super users can change verification status' },
|
||||
{ status: 403 }
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// For non-verified updates, check regular permissions
|
||||
const hasNonVerifiedUpdates =
|
||||
data.name !== undefined || data.profileImageUrl !== undefined || data.details !== undefined
|
||||
|
||||
if (hasNonVerifiedUpdates) {
|
||||
const canEdit = await hasPermission(session.user.id, existing[0])
|
||||
if (!canEdit) {
|
||||
logger.warn(`[${requestId}] User denied permission to update profile: ${id}`)
|
||||
return NextResponse.json({ error: 'Access denied' }, { status: 403 })
|
||||
}
|
||||
}
|
||||
|
||||
const updateData: any = {
|
||||
@@ -111,6 +130,7 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
|
||||
if (data.name !== undefined) updateData.name = data.name
|
||||
if (data.profileImageUrl !== undefined) updateData.profileImageUrl = data.profileImageUrl
|
||||
if (data.details !== undefined) updateData.details = data.details
|
||||
if (data.verified !== undefined) updateData.verified = data.verified
|
||||
|
||||
const updated = await db
|
||||
.update(templateCreators)
|
||||
|
||||
@@ -1,113 +0,0 @@
|
||||
import { db } from '@sim/db'
|
||||
import { templateCreators } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { getSession } from '@/lib/auth'
|
||||
import { generateRequestId } from '@/lib/core/utils/request'
|
||||
import { verifyEffectiveSuperUser } from '@/lib/templates/permissions'
|
||||
|
||||
const logger = createLogger('CreatorVerificationAPI')
|
||||
|
||||
export const revalidate = 0
|
||||
|
||||
// POST /api/creators/[id]/verify - Verify a creator (super users only)
|
||||
export async function POST(request: NextRequest, { params }: { params: Promise<{ id: string }> }) {
|
||||
const requestId = generateRequestId()
|
||||
const { id } = await params
|
||||
|
||||
try {
|
||||
const session = await getSession()
|
||||
if (!session?.user?.id) {
|
||||
logger.warn(`[${requestId}] Unauthorized verification attempt for creator: ${id}`)
|
||||
return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
|
||||
}
|
||||
|
||||
// Check if user is a super user
|
||||
const { effectiveSuperUser } = await verifyEffectiveSuperUser(session.user.id)
|
||||
if (!effectiveSuperUser) {
|
||||
logger.warn(`[${requestId}] Non-super user attempted to verify creator: ${id}`)
|
||||
return NextResponse.json({ error: 'Only super users can verify creators' }, { status: 403 })
|
||||
}
|
||||
|
||||
// Check if creator exists
|
||||
const existingCreator = await db
|
||||
.select()
|
||||
.from(templateCreators)
|
||||
.where(eq(templateCreators.id, id))
|
||||
.limit(1)
|
||||
|
||||
if (existingCreator.length === 0) {
|
||||
logger.warn(`[${requestId}] Creator not found for verification: ${id}`)
|
||||
return NextResponse.json({ error: 'Creator not found' }, { status: 404 })
|
||||
}
|
||||
|
||||
// Update creator verified status to true
|
||||
await db
|
||||
.update(templateCreators)
|
||||
.set({ verified: true, updatedAt: new Date() })
|
||||
.where(eq(templateCreators.id, id))
|
||||
|
||||
logger.info(`[${requestId}] Creator verified: ${id} by super user: ${session.user.id}`)
|
||||
|
||||
return NextResponse.json({
|
||||
message: 'Creator verified successfully',
|
||||
creatorId: id,
|
||||
})
|
||||
} catch (error) {
|
||||
logger.error(`[${requestId}] Error verifying creator ${id}`, error)
|
||||
return NextResponse.json({ error: 'Internal server error' }, { status: 500 })
|
||||
}
|
||||
}
|
||||
|
||||
// DELETE /api/creators/[id]/verify - Unverify a creator (super users only)
|
||||
export async function DELETE(
|
||||
request: NextRequest,
|
||||
{ params }: { params: Promise<{ id: string }> }
|
||||
) {
|
||||
const requestId = generateRequestId()
|
||||
const { id } = await params
|
||||
|
||||
try {
|
||||
const session = await getSession()
|
||||
if (!session?.user?.id) {
|
||||
logger.warn(`[${requestId}] Unauthorized unverification attempt for creator: ${id}`)
|
||||
return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
|
||||
}
|
||||
|
||||
// Check if user is a super user
|
||||
const { effectiveSuperUser } = await verifyEffectiveSuperUser(session.user.id)
|
||||
if (!effectiveSuperUser) {
|
||||
logger.warn(`[${requestId}] Non-super user attempted to unverify creator: ${id}`)
|
||||
return NextResponse.json({ error: 'Only super users can unverify creators' }, { status: 403 })
|
||||
}
|
||||
|
||||
// Check if creator exists
|
||||
const existingCreator = await db
|
||||
.select()
|
||||
.from(templateCreators)
|
||||
.where(eq(templateCreators.id, id))
|
||||
.limit(1)
|
||||
|
||||
if (existingCreator.length === 0) {
|
||||
logger.warn(`[${requestId}] Creator not found for unverification: ${id}`)
|
||||
return NextResponse.json({ error: 'Creator not found' }, { status: 404 })
|
||||
}
|
||||
|
||||
// Update creator verified status to false
|
||||
await db
|
||||
.update(templateCreators)
|
||||
.set({ verified: false, updatedAt: new Date() })
|
||||
.where(eq(templateCreators.id, id))
|
||||
|
||||
logger.info(`[${requestId}] Creator unverified: ${id} by super user: ${session.user.id}`)
|
||||
|
||||
return NextResponse.json({
|
||||
message: 'Creator unverified successfully',
|
||||
creatorId: id,
|
||||
})
|
||||
} catch (error) {
|
||||
logger.error(`[${requestId}] Error unverifying creator ${id}`, error)
|
||||
return NextResponse.json({ error: 'Internal server error' }, { status: 500 })
|
||||
}
|
||||
}
|
||||
@@ -1,9 +1,10 @@
|
||||
import { db } from '@sim/db'
|
||||
import { asyncJobs, db } from '@sim/db'
|
||||
import { workflowExecutionLogs } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { and, eq, lt, sql } from 'drizzle-orm'
|
||||
import { and, eq, inArray, lt, sql } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { verifyCronAuth } from '@/lib/auth/internal'
|
||||
import { JOB_RETENTION_HOURS, JOB_STATUS } from '@/lib/core/async-jobs'
|
||||
import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
|
||||
|
||||
const logger = createLogger('CleanupStaleExecutions')
|
||||
@@ -80,12 +81,102 @@ export async function GET(request: NextRequest) {
|
||||
|
||||
logger.info(`Stale execution cleanup completed. Cleaned: ${cleaned}, Failed: ${failed}`)
|
||||
|
||||
// Clean up stale async jobs (stuck in processing)
|
||||
let asyncJobsMarkedFailed = 0
|
||||
|
||||
try {
|
||||
const staleAsyncJobs = await db
|
||||
.update(asyncJobs)
|
||||
.set({
|
||||
status: JOB_STATUS.FAILED,
|
||||
completedAt: new Date(),
|
||||
error: `Job terminated: stuck in processing for more than ${STALE_THRESHOLD_MINUTES} minutes`,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(eq(asyncJobs.status, JOB_STATUS.PROCESSING), lt(asyncJobs.startedAt, staleThreshold))
|
||||
)
|
||||
.returning({ id: asyncJobs.id })
|
||||
|
||||
asyncJobsMarkedFailed = staleAsyncJobs.length
|
||||
if (asyncJobsMarkedFailed > 0) {
|
||||
logger.info(`Marked ${asyncJobsMarkedFailed} stale async jobs as failed`)
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error('Failed to clean up stale async jobs:', {
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
})
|
||||
}
|
||||
|
||||
// Clean up stale pending jobs (never started, e.g., due to server crash before startJob())
|
||||
let stalePendingJobsMarkedFailed = 0
|
||||
|
||||
try {
|
||||
const stalePendingJobs = await db
|
||||
.update(asyncJobs)
|
||||
.set({
|
||||
status: JOB_STATUS.FAILED,
|
||||
completedAt: new Date(),
|
||||
error: `Job terminated: stuck in pending state for more than ${STALE_THRESHOLD_MINUTES} minutes (never started)`,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(eq(asyncJobs.status, JOB_STATUS.PENDING), lt(asyncJobs.createdAt, staleThreshold))
|
||||
)
|
||||
.returning({ id: asyncJobs.id })
|
||||
|
||||
stalePendingJobsMarkedFailed = stalePendingJobs.length
|
||||
if (stalePendingJobsMarkedFailed > 0) {
|
||||
logger.info(`Marked ${stalePendingJobsMarkedFailed} stale pending jobs as failed`)
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error('Failed to clean up stale pending jobs:', {
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
})
|
||||
}
|
||||
|
||||
// Delete completed/failed jobs older than retention period
|
||||
const retentionThreshold = new Date(Date.now() - JOB_RETENTION_HOURS * 60 * 60 * 1000)
|
||||
let asyncJobsDeleted = 0
|
||||
|
||||
try {
|
||||
const deletedJobs = await db
|
||||
.delete(asyncJobs)
|
||||
.where(
|
||||
and(
|
||||
inArray(asyncJobs.status, [JOB_STATUS.COMPLETED, JOB_STATUS.FAILED]),
|
||||
lt(asyncJobs.completedAt, retentionThreshold)
|
||||
)
|
||||
)
|
||||
.returning({ id: asyncJobs.id })
|
||||
|
||||
asyncJobsDeleted = deletedJobs.length
|
||||
if (asyncJobsDeleted > 0) {
|
||||
logger.info(
|
||||
`Deleted ${asyncJobsDeleted} old async jobs (retention: ${JOB_RETENTION_HOURS}h)`
|
||||
)
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error('Failed to delete old async jobs:', {
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
})
|
||||
}
|
||||
|
||||
return NextResponse.json({
|
||||
success: true,
|
||||
found: staleExecutions.length,
|
||||
cleaned,
|
||||
failed,
|
||||
thresholdMinutes: STALE_THRESHOLD_MINUTES,
|
||||
executions: {
|
||||
found: staleExecutions.length,
|
||||
cleaned,
|
||||
failed,
|
||||
thresholdMinutes: STALE_THRESHOLD_MINUTES,
|
||||
},
|
||||
asyncJobs: {
|
||||
staleProcessingMarkedFailed: asyncJobsMarkedFailed,
|
||||
stalePendingMarkedFailed: stalePendingJobsMarkedFailed,
|
||||
oldDeleted: asyncJobsDeleted,
|
||||
staleThresholdMinutes: STALE_THRESHOLD_MINUTES,
|
||||
retentionHours: JOB_RETENTION_HOURS,
|
||||
},
|
||||
})
|
||||
} catch (error) {
|
||||
logger.error('Error in stale execution cleanup job:', error)
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { runs } from '@trigger.dev/sdk'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { checkHybridAuth } from '@/lib/auth/hybrid'
|
||||
import { getJobQueue, JOB_STATUS } from '@/lib/core/async-jobs'
|
||||
import { generateRequestId } from '@/lib/core/utils/request'
|
||||
import { createErrorResponse } from '@/app/api/workflows/utils'
|
||||
|
||||
@@ -15,8 +15,6 @@ export async function GET(
|
||||
const requestId = generateRequestId()
|
||||
|
||||
try {
|
||||
logger.debug(`[${requestId}] Getting status for task: ${taskId}`)
|
||||
|
||||
const authResult = await checkHybridAuth(request, { requireWorkflowId: false })
|
||||
if (!authResult.success || !authResult.userId) {
|
||||
logger.warn(`[${requestId}] Unauthorized task status request`)
|
||||
@@ -25,76 +23,60 @@ export async function GET(
|
||||
|
||||
const authenticatedUserId = authResult.userId
|
||||
|
||||
const run = await runs.retrieve(taskId)
|
||||
const jobQueue = await getJobQueue()
|
||||
const job = await jobQueue.getJob(taskId)
|
||||
|
||||
logger.debug(`[${requestId}] Task ${taskId} status: ${run.status}`)
|
||||
|
||||
const payload = run.payload as any
|
||||
if (payload?.workflowId) {
|
||||
const { verifyWorkflowAccess } = await import('@/socket/middleware/permissions')
|
||||
const accessCheck = await verifyWorkflowAccess(authenticatedUserId, payload.workflowId)
|
||||
if (!accessCheck.hasAccess) {
|
||||
logger.warn(`[${requestId}] User ${authenticatedUserId} denied access to task ${taskId}`, {
|
||||
workflowId: payload.workflowId,
|
||||
})
|
||||
return createErrorResponse('Access denied', 403)
|
||||
}
|
||||
logger.debug(`[${requestId}] User ${authenticatedUserId} has access to task ${taskId}`)
|
||||
} else {
|
||||
if (payload?.userId && payload.userId !== authenticatedUserId) {
|
||||
logger.warn(
|
||||
`[${requestId}] User ${authenticatedUserId} attempted to access task ${taskId} owned by ${payload.userId}`
|
||||
)
|
||||
return createErrorResponse('Access denied', 403)
|
||||
}
|
||||
if (!payload?.userId) {
|
||||
logger.warn(
|
||||
`[${requestId}] Task ${taskId} has no ownership information in payload. Denying access for security.`
|
||||
)
|
||||
return createErrorResponse('Access denied', 403)
|
||||
}
|
||||
if (!job) {
|
||||
return createErrorResponse('Task not found', 404)
|
||||
}
|
||||
|
||||
const statusMap = {
|
||||
QUEUED: 'queued',
|
||||
WAITING_FOR_DEPLOY: 'queued',
|
||||
EXECUTING: 'processing',
|
||||
RESCHEDULED: 'processing',
|
||||
FROZEN: 'processing',
|
||||
COMPLETED: 'completed',
|
||||
CANCELED: 'cancelled',
|
||||
FAILED: 'failed',
|
||||
CRASHED: 'failed',
|
||||
INTERRUPTED: 'failed',
|
||||
SYSTEM_FAILURE: 'failed',
|
||||
EXPIRED: 'failed',
|
||||
} as const
|
||||
if (job.metadata?.workflowId) {
|
||||
const { verifyWorkflowAccess } = await import('@/socket/middleware/permissions')
|
||||
const accessCheck = await verifyWorkflowAccess(
|
||||
authenticatedUserId,
|
||||
job.metadata.workflowId as string
|
||||
)
|
||||
if (!accessCheck.hasAccess) {
|
||||
logger.warn(`[${requestId}] Access denied to workflow ${job.metadata.workflowId}`)
|
||||
return createErrorResponse('Access denied', 403)
|
||||
}
|
||||
} else if (job.metadata?.userId && job.metadata.userId !== authenticatedUserId) {
|
||||
logger.warn(`[${requestId}] Access denied to user ${job.metadata.userId}`)
|
||||
return createErrorResponse('Access denied', 403)
|
||||
} else if (!job.metadata?.userId && !job.metadata?.workflowId) {
|
||||
logger.warn(`[${requestId}] Access denied to job ${taskId}`)
|
||||
return createErrorResponse('Access denied', 403)
|
||||
}
|
||||
|
||||
const mappedStatus = statusMap[run.status as keyof typeof statusMap] || 'unknown'
|
||||
const mappedStatus = job.status === JOB_STATUS.PENDING ? 'queued' : job.status
|
||||
|
||||
const response: any = {
|
||||
success: true,
|
||||
taskId,
|
||||
status: mappedStatus,
|
||||
metadata: {
|
||||
startedAt: run.startedAt,
|
||||
startedAt: job.startedAt,
|
||||
},
|
||||
}
|
||||
|
||||
if (mappedStatus === 'completed') {
|
||||
response.output = run.output // This contains the workflow execution results
|
||||
response.metadata.completedAt = run.finishedAt
|
||||
response.metadata.duration = run.durationMs
|
||||
if (job.status === JOB_STATUS.COMPLETED) {
|
||||
response.output = job.output
|
||||
response.metadata.completedAt = job.completedAt
|
||||
if (job.startedAt && job.completedAt) {
|
||||
response.metadata.duration = job.completedAt.getTime() - job.startedAt.getTime()
|
||||
}
|
||||
}
|
||||
|
||||
if (mappedStatus === 'failed') {
|
||||
response.error = run.error
|
||||
response.metadata.completedAt = run.finishedAt
|
||||
response.metadata.duration = run.durationMs
|
||||
if (job.status === JOB_STATUS.FAILED) {
|
||||
response.error = job.error
|
||||
response.metadata.completedAt = job.completedAt
|
||||
if (job.startedAt && job.completedAt) {
|
||||
response.metadata.duration = job.completedAt.getTime() - job.startedAt.getTime()
|
||||
}
|
||||
}
|
||||
|
||||
if (mappedStatus === 'processing' || mappedStatus === 'queued') {
|
||||
response.estimatedDuration = 180000 // 3 minutes max from our config
|
||||
if (job.status === JOB_STATUS.PROCESSING || job.status === JOB_STATUS.PENDING) {
|
||||
response.estimatedDuration = 180000
|
||||
}
|
||||
|
||||
return NextResponse.json(response)
|
||||
|
||||
@@ -1,10 +1,9 @@
|
||||
import { db, workflowDeploymentVersion, workflowSchedule } from '@sim/db'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { tasks } from '@trigger.dev/sdk'
|
||||
import { and, eq, isNull, lt, lte, not, or, sql } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { verifyCronAuth } from '@/lib/auth/internal'
|
||||
import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
|
||||
import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
|
||||
import { generateRequestId } from '@/lib/core/utils/request'
|
||||
import { executeScheduleJob } from '@/background/schedule-execution'
|
||||
|
||||
@@ -55,72 +54,67 @@ export async function GET(request: NextRequest) {
|
||||
logger.debug(`[${requestId}] Successfully queried schedules: ${dueSchedules.length} found`)
|
||||
logger.info(`[${requestId}] Processing ${dueSchedules.length} due scheduled workflows`)
|
||||
|
||||
if (isTriggerDevEnabled) {
|
||||
const triggerPromises = dueSchedules.map(async (schedule) => {
|
||||
const queueTime = schedule.lastQueuedAt ?? queuedAt
|
||||
const jobQueue = await getJobQueue()
|
||||
|
||||
try {
|
||||
const payload = {
|
||||
scheduleId: schedule.id,
|
||||
workflowId: schedule.workflowId,
|
||||
blockId: schedule.blockId || undefined,
|
||||
cronExpression: schedule.cronExpression || undefined,
|
||||
lastRanAt: schedule.lastRanAt?.toISOString(),
|
||||
failedCount: schedule.failedCount || 0,
|
||||
now: queueTime.toISOString(),
|
||||
scheduledFor: schedule.nextRunAt?.toISOString(),
|
||||
}
|
||||
const queuePromises = dueSchedules.map(async (schedule) => {
|
||||
const queueTime = schedule.lastQueuedAt ?? queuedAt
|
||||
|
||||
const handle = await tasks.trigger('schedule-execution', payload)
|
||||
logger.info(
|
||||
`[${requestId}] Queued schedule execution task ${handle.id} for workflow ${schedule.workflowId}`
|
||||
)
|
||||
return handle
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
`[${requestId}] Failed to trigger schedule execution for workflow ${schedule.workflowId}`,
|
||||
error
|
||||
)
|
||||
return null
|
||||
}
|
||||
})
|
||||
const payload = {
|
||||
scheduleId: schedule.id,
|
||||
workflowId: schedule.workflowId,
|
||||
blockId: schedule.blockId || undefined,
|
||||
cronExpression: schedule.cronExpression || undefined,
|
||||
lastRanAt: schedule.lastRanAt?.toISOString(),
|
||||
failedCount: schedule.failedCount || 0,
|
||||
now: queueTime.toISOString(),
|
||||
scheduledFor: schedule.nextRunAt?.toISOString(),
|
||||
}
|
||||
|
||||
await Promise.allSettled(triggerPromises)
|
||||
|
||||
logger.info(`[${requestId}] Queued ${dueSchedules.length} schedule executions to Trigger.dev`)
|
||||
} else {
|
||||
const directExecutionPromises = dueSchedules.map(async (schedule) => {
|
||||
const queueTime = schedule.lastQueuedAt ?? queuedAt
|
||||
|
||||
const payload = {
|
||||
scheduleId: schedule.id,
|
||||
workflowId: schedule.workflowId,
|
||||
blockId: schedule.blockId || undefined,
|
||||
cronExpression: schedule.cronExpression || undefined,
|
||||
lastRanAt: schedule.lastRanAt?.toISOString(),
|
||||
failedCount: schedule.failedCount || 0,
|
||||
now: queueTime.toISOString(),
|
||||
scheduledFor: schedule.nextRunAt?.toISOString(),
|
||||
}
|
||||
|
||||
void executeScheduleJob(payload).catch((error) => {
|
||||
logger.error(
|
||||
`[${requestId}] Direct schedule execution failed for workflow ${schedule.workflowId}`,
|
||||
error
|
||||
)
|
||||
try {
|
||||
const jobId = await jobQueue.enqueue('schedule-execution', payload, {
|
||||
metadata: { workflowId: schedule.workflowId },
|
||||
})
|
||||
|
||||
logger.info(
|
||||
`[${requestId}] Queued direct schedule execution for workflow ${schedule.workflowId} (Trigger.dev disabled)`
|
||||
`[${requestId}] Queued schedule execution task ${jobId} for workflow ${schedule.workflowId}`
|
||||
)
|
||||
})
|
||||
|
||||
await Promise.allSettled(directExecutionPromises)
|
||||
if (shouldExecuteInline()) {
|
||||
void (async () => {
|
||||
try {
|
||||
await jobQueue.startJob(jobId)
|
||||
const output = await executeScheduleJob(payload)
|
||||
await jobQueue.completeJob(jobId, output)
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
logger.error(
|
||||
`[${requestId}] Schedule execution failed for workflow ${schedule.workflowId}`,
|
||||
{ jobId, error: errorMessage }
|
||||
)
|
||||
try {
|
||||
await jobQueue.markJobFailed(jobId, errorMessage)
|
||||
} catch (markFailedError) {
|
||||
logger.error(`[${requestId}] Failed to mark job as failed`, {
|
||||
jobId,
|
||||
error:
|
||||
markFailedError instanceof Error
|
||||
? markFailedError.message
|
||||
: String(markFailedError),
|
||||
})
|
||||
}
|
||||
}
|
||||
})()
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
`[${requestId}] Failed to queue schedule execution for workflow ${schedule.workflowId}`,
|
||||
error
|
||||
)
|
||||
}
|
||||
})
|
||||
|
||||
logger.info(
|
||||
`[${requestId}] Queued ${dueSchedules.length} direct schedule executions (Trigger.dev disabled)`
|
||||
)
|
||||
}
|
||||
await Promise.allSettled(queuePromises)
|
||||
|
||||
logger.info(`[${requestId}] Queued ${dueSchedules.length} schedule executions`)
|
||||
|
||||
return NextResponse.json({
|
||||
message: 'Scheduled workflow executions processed',
|
||||
|
||||
@@ -1,101 +0,0 @@
|
||||
import { db } from '@sim/db'
|
||||
import { templates } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { getSession } from '@/lib/auth'
|
||||
import { generateRequestId } from '@/lib/core/utils/request'
|
||||
import { verifyEffectiveSuperUser } from '@/lib/templates/permissions'
|
||||
|
||||
const logger = createLogger('TemplateApprovalAPI')
|
||||
|
||||
export const revalidate = 0
|
||||
|
||||
/**
|
||||
* POST /api/templates/[id]/approve - Approve a template (super users only)
|
||||
*/
|
||||
export async function POST(request: NextRequest, { params }: { params: Promise<{ id: string }> }) {
|
||||
const requestId = generateRequestId()
|
||||
const { id } = await params
|
||||
|
||||
try {
|
||||
const session = await getSession()
|
||||
if (!session?.user?.id) {
|
||||
logger.warn(`[${requestId}] Unauthorized template approval attempt for ID: ${id}`)
|
||||
return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
|
||||
}
|
||||
|
||||
const { effectiveSuperUser } = await verifyEffectiveSuperUser(session.user.id)
|
||||
if (!effectiveSuperUser) {
|
||||
logger.warn(`[${requestId}] Non-super user attempted to approve template: ${id}`)
|
||||
return NextResponse.json({ error: 'Only super users can approve templates' }, { status: 403 })
|
||||
}
|
||||
|
||||
const existingTemplate = await db.select().from(templates).where(eq(templates.id, id)).limit(1)
|
||||
if (existingTemplate.length === 0) {
|
||||
logger.warn(`[${requestId}] Template not found for approval: ${id}`)
|
||||
return NextResponse.json({ error: 'Template not found' }, { status: 404 })
|
||||
}
|
||||
|
||||
await db
|
||||
.update(templates)
|
||||
.set({ status: 'approved', updatedAt: new Date() })
|
||||
.where(eq(templates.id, id))
|
||||
|
||||
logger.info(`[${requestId}] Template approved: ${id} by super user: ${session.user.id}`)
|
||||
|
||||
return NextResponse.json({
|
||||
message: 'Template approved successfully',
|
||||
templateId: id,
|
||||
})
|
||||
} catch (error) {
|
||||
logger.error(`[${requestId}] Error approving template ${id}`, error)
|
||||
return NextResponse.json({ error: 'Internal server error' }, { status: 500 })
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* DELETE /api/templates/[id]/approve - Unapprove a template (super users only)
|
||||
*/
|
||||
export async function DELETE(
|
||||
_request: NextRequest,
|
||||
{ params }: { params: Promise<{ id: string }> }
|
||||
) {
|
||||
const requestId = generateRequestId()
|
||||
const { id } = await params
|
||||
|
||||
try {
|
||||
const session = await getSession()
|
||||
if (!session?.user?.id) {
|
||||
logger.warn(`[${requestId}] Unauthorized template rejection attempt for ID: ${id}`)
|
||||
return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
|
||||
}
|
||||
|
||||
const { effectiveSuperUser } = await verifyEffectiveSuperUser(session.user.id)
|
||||
if (!effectiveSuperUser) {
|
||||
logger.warn(`[${requestId}] Non-super user attempted to reject template: ${id}`)
|
||||
return NextResponse.json({ error: 'Only super users can reject templates' }, { status: 403 })
|
||||
}
|
||||
|
||||
const existingTemplate = await db.select().from(templates).where(eq(templates.id, id)).limit(1)
|
||||
if (existingTemplate.length === 0) {
|
||||
logger.warn(`[${requestId}] Template not found for rejection: ${id}`)
|
||||
return NextResponse.json({ error: 'Template not found' }, { status: 404 })
|
||||
}
|
||||
|
||||
await db
|
||||
.update(templates)
|
||||
.set({ status: 'rejected', updatedAt: new Date() })
|
||||
.where(eq(templates.id, id))
|
||||
|
||||
logger.info(`[${requestId}] Template rejected: ${id} by super user: ${session.user.id}`)
|
||||
|
||||
return NextResponse.json({
|
||||
message: 'Template rejected successfully',
|
||||
templateId: id,
|
||||
})
|
||||
} catch (error) {
|
||||
logger.error(`[${requestId}] Error rejecting template ${id}`, error)
|
||||
return NextResponse.json({ error: 'Internal server error' }, { status: 500 })
|
||||
}
|
||||
}
|
||||
@@ -1,55 +0,0 @@
|
||||
import { db } from '@sim/db'
|
||||
import { templates } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { getSession } from '@/lib/auth'
|
||||
import { generateRequestId } from '@/lib/core/utils/request'
|
||||
import { verifyEffectiveSuperUser } from '@/lib/templates/permissions'
|
||||
|
||||
const logger = createLogger('TemplateRejectionAPI')
|
||||
|
||||
export const revalidate = 0
|
||||
|
||||
/**
|
||||
* POST /api/templates/[id]/reject - Reject a template (super users only)
|
||||
*/
|
||||
export async function POST(request: NextRequest, { params }: { params: Promise<{ id: string }> }) {
|
||||
const requestId = generateRequestId()
|
||||
const { id } = await params
|
||||
|
||||
try {
|
||||
const session = await getSession()
|
||||
if (!session?.user?.id) {
|
||||
logger.warn(`[${requestId}] Unauthorized template rejection attempt for ID: ${id}`)
|
||||
return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
|
||||
}
|
||||
|
||||
const { effectiveSuperUser } = await verifyEffectiveSuperUser(session.user.id)
|
||||
if (!effectiveSuperUser) {
|
||||
logger.warn(`[${requestId}] Non-super user attempted to reject template: ${id}`)
|
||||
return NextResponse.json({ error: 'Only super users can reject templates' }, { status: 403 })
|
||||
}
|
||||
|
||||
const existingTemplate = await db.select().from(templates).where(eq(templates.id, id)).limit(1)
|
||||
if (existingTemplate.length === 0) {
|
||||
logger.warn(`[${requestId}] Template not found for rejection: ${id}`)
|
||||
return NextResponse.json({ error: 'Template not found' }, { status: 404 })
|
||||
}
|
||||
|
||||
await db
|
||||
.update(templates)
|
||||
.set({ status: 'rejected', updatedAt: new Date() })
|
||||
.where(eq(templates.id, id))
|
||||
|
||||
logger.info(`[${requestId}] Template rejected: ${id} by super user: ${session.user.id}`)
|
||||
|
||||
return NextResponse.json({
|
||||
message: 'Template rejected successfully',
|
||||
templateId: id,
|
||||
})
|
||||
} catch (error) {
|
||||
logger.error(`[${requestId}] Error rejecting template ${id}`, error)
|
||||
return NextResponse.json({ error: 'Internal server error' }, { status: 500 })
|
||||
}
|
||||
}
|
||||
@@ -106,6 +106,7 @@ const updateTemplateSchema = z.object({
|
||||
creatorId: z.string().optional(), // Creator profile ID
|
||||
tags: z.array(z.string()).max(10, 'Maximum 10 tags allowed').optional(),
|
||||
updateState: z.boolean().optional(), // Explicitly request state update from current workflow
|
||||
status: z.enum(['approved', 'rejected', 'pending']).optional(), // Status change (super users only)
|
||||
})
|
||||
|
||||
// PUT /api/templates/[id] - Update a template
|
||||
@@ -131,7 +132,7 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
|
||||
)
|
||||
}
|
||||
|
||||
const { name, details, creatorId, tags, updateState } = validationResult.data
|
||||
const { name, details, creatorId, tags, updateState, status } = validationResult.data
|
||||
|
||||
const existingTemplate = await db.select().from(templates).where(eq(templates.id, id)).limit(1)
|
||||
|
||||
@@ -142,21 +143,44 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
|
||||
|
||||
const template = existingTemplate[0]
|
||||
|
||||
if (!template.creatorId) {
|
||||
logger.warn(`[${requestId}] Template ${id} has no creator, denying update`)
|
||||
return NextResponse.json({ error: 'Access denied' }, { status: 403 })
|
||||
// Status changes require super user permission
|
||||
if (status !== undefined) {
|
||||
const { verifyEffectiveSuperUser } = await import('@/lib/templates/permissions')
|
||||
const { effectiveSuperUser } = await verifyEffectiveSuperUser(session.user.id)
|
||||
if (!effectiveSuperUser) {
|
||||
logger.warn(`[${requestId}] Non-super user attempted to change template status: ${id}`)
|
||||
return NextResponse.json(
|
||||
{ error: 'Only super users can change template status' },
|
||||
{ status: 403 }
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
const { verifyCreatorPermission } = await import('@/lib/templates/permissions')
|
||||
const { hasPermission, error: permissionError } = await verifyCreatorPermission(
|
||||
session.user.id,
|
||||
template.creatorId,
|
||||
'admin'
|
||||
)
|
||||
// For non-status updates, verify creator permission
|
||||
const hasNonStatusUpdates =
|
||||
name !== undefined ||
|
||||
details !== undefined ||
|
||||
creatorId !== undefined ||
|
||||
tags !== undefined ||
|
||||
updateState
|
||||
|
||||
if (!hasPermission) {
|
||||
logger.warn(`[${requestId}] User denied permission to update template ${id}`)
|
||||
return NextResponse.json({ error: permissionError || 'Access denied' }, { status: 403 })
|
||||
if (hasNonStatusUpdates) {
|
||||
if (!template.creatorId) {
|
||||
logger.warn(`[${requestId}] Template ${id} has no creator, denying update`)
|
||||
return NextResponse.json({ error: 'Access denied' }, { status: 403 })
|
||||
}
|
||||
|
||||
const { verifyCreatorPermission } = await import('@/lib/templates/permissions')
|
||||
const { hasPermission, error: permissionError } = await verifyCreatorPermission(
|
||||
session.user.id,
|
||||
template.creatorId,
|
||||
'admin'
|
||||
)
|
||||
|
||||
if (!hasPermission) {
|
||||
logger.warn(`[${requestId}] User denied permission to update template ${id}`)
|
||||
return NextResponse.json({ error: permissionError || 'Access denied' }, { status: 403 })
|
||||
}
|
||||
}
|
||||
|
||||
const updateData: any = {
|
||||
@@ -167,6 +191,7 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
|
||||
if (details !== undefined) updateData.details = details
|
||||
if (tags !== undefined) updateData.tags = tags
|
||||
if (creatorId !== undefined) updateData.creatorId = creatorId
|
||||
if (status !== undefined) updateData.status = status
|
||||
|
||||
if (updateState && template.workflowId) {
|
||||
const { verifyWorkflowAccess } = await import('@/socket/middleware/permissions')
|
||||
|
||||
@@ -1,190 +0,0 @@
|
||||
import { db, workflowDeploymentVersion } from '@sim/db'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { and, eq } from 'drizzle-orm'
|
||||
import type { NextRequest } from 'next/server'
|
||||
import { generateRequestId } from '@/lib/core/utils/request'
|
||||
import { syncMcpToolsForWorkflow } from '@/lib/mcp/workflow-mcp-sync'
|
||||
import { restorePreviousVersionWebhooks, saveTriggerWebhooksForDeploy } from '@/lib/webhooks/deploy'
|
||||
import { activateWorkflowVersion } from '@/lib/workflows/persistence/utils'
|
||||
import {
|
||||
cleanupDeploymentVersion,
|
||||
createSchedulesForDeploy,
|
||||
validateWorkflowSchedules,
|
||||
} from '@/lib/workflows/schedules'
|
||||
import { validateWorkflowPermissions } from '@/lib/workflows/utils'
|
||||
import { createErrorResponse, createSuccessResponse } from '@/app/api/workflows/utils'
|
||||
import type { BlockState } from '@/stores/workflows/workflow/types'
|
||||
|
||||
const logger = createLogger('WorkflowActivateDeploymentAPI')
|
||||
|
||||
export const dynamic = 'force-dynamic'
|
||||
export const runtime = 'nodejs'
|
||||
|
||||
export async function POST(
|
||||
request: NextRequest,
|
||||
{ params }: { params: Promise<{ id: string; version: string }> }
|
||||
) {
|
||||
const requestId = generateRequestId()
|
||||
const { id, version } = await params
|
||||
|
||||
try {
|
||||
const {
|
||||
error,
|
||||
session,
|
||||
workflow: workflowData,
|
||||
} = await validateWorkflowPermissions(id, requestId, 'admin')
|
||||
if (error) {
|
||||
return createErrorResponse(error.message, error.status)
|
||||
}
|
||||
|
||||
const actorUserId = session?.user?.id
|
||||
if (!actorUserId) {
|
||||
logger.warn(`[${requestId}] Unable to resolve actor user for deployment activation: ${id}`)
|
||||
return createErrorResponse('Unable to determine activating user', 400)
|
||||
}
|
||||
|
||||
const versionNum = Number(version)
|
||||
if (!Number.isFinite(versionNum)) {
|
||||
return createErrorResponse('Invalid version number', 400)
|
||||
}
|
||||
|
||||
const [versionRow] = await db
|
||||
.select({
|
||||
id: workflowDeploymentVersion.id,
|
||||
state: workflowDeploymentVersion.state,
|
||||
})
|
||||
.from(workflowDeploymentVersion)
|
||||
.where(
|
||||
and(
|
||||
eq(workflowDeploymentVersion.workflowId, id),
|
||||
eq(workflowDeploymentVersion.version, versionNum)
|
||||
)
|
||||
)
|
||||
.limit(1)
|
||||
|
||||
if (!versionRow?.state) {
|
||||
return createErrorResponse('Deployment version not found', 404)
|
||||
}
|
||||
|
||||
const [currentActiveVersion] = await db
|
||||
.select({ id: workflowDeploymentVersion.id })
|
||||
.from(workflowDeploymentVersion)
|
||||
.where(
|
||||
and(
|
||||
eq(workflowDeploymentVersion.workflowId, id),
|
||||
eq(workflowDeploymentVersion.isActive, true)
|
||||
)
|
||||
)
|
||||
.limit(1)
|
||||
|
||||
const previousVersionId = currentActiveVersion?.id
|
||||
|
||||
const deployedState = versionRow.state as { blocks?: Record<string, BlockState> }
|
||||
const blocks = deployedState.blocks
|
||||
if (!blocks || typeof blocks !== 'object') {
|
||||
return createErrorResponse('Invalid deployed state structure', 500)
|
||||
}
|
||||
|
||||
const scheduleValidation = validateWorkflowSchedules(blocks)
|
||||
if (!scheduleValidation.isValid) {
|
||||
return createErrorResponse(`Invalid schedule configuration: ${scheduleValidation.error}`, 400)
|
||||
}
|
||||
|
||||
const triggerSaveResult = await saveTriggerWebhooksForDeploy({
|
||||
request,
|
||||
workflowId: id,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
userId: actorUserId,
|
||||
blocks,
|
||||
requestId,
|
||||
deploymentVersionId: versionRow.id,
|
||||
previousVersionId,
|
||||
forceRecreateSubscriptions: true,
|
||||
})
|
||||
|
||||
if (!triggerSaveResult.success) {
|
||||
return createErrorResponse(
|
||||
triggerSaveResult.error?.message || 'Failed to sync trigger configuration',
|
||||
triggerSaveResult.error?.status || 500
|
||||
)
|
||||
}
|
||||
|
||||
const scheduleResult = await createSchedulesForDeploy(id, blocks, db, versionRow.id)
|
||||
|
||||
if (!scheduleResult.success) {
|
||||
await cleanupDeploymentVersion({
|
||||
workflowId: id,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
requestId,
|
||||
deploymentVersionId: versionRow.id,
|
||||
})
|
||||
if (previousVersionId) {
|
||||
await restorePreviousVersionWebhooks({
|
||||
request,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
userId: actorUserId,
|
||||
previousVersionId,
|
||||
requestId,
|
||||
})
|
||||
}
|
||||
return createErrorResponse(scheduleResult.error || 'Failed to sync schedules', 500)
|
||||
}
|
||||
|
||||
const result = await activateWorkflowVersion({ workflowId: id, version: versionNum })
|
||||
if (!result.success) {
|
||||
await cleanupDeploymentVersion({
|
||||
workflowId: id,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
requestId,
|
||||
deploymentVersionId: versionRow.id,
|
||||
})
|
||||
if (previousVersionId) {
|
||||
await restorePreviousVersionWebhooks({
|
||||
request,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
userId: actorUserId,
|
||||
previousVersionId,
|
||||
requestId,
|
||||
})
|
||||
}
|
||||
return createErrorResponse(result.error || 'Failed to activate deployment', 400)
|
||||
}
|
||||
|
||||
if (previousVersionId && previousVersionId !== versionRow.id) {
|
||||
try {
|
||||
logger.info(
|
||||
`[${requestId}] Cleaning up previous version ${previousVersionId} webhooks/schedules`
|
||||
)
|
||||
await cleanupDeploymentVersion({
|
||||
workflowId: id,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
requestId,
|
||||
deploymentVersionId: previousVersionId,
|
||||
skipExternalCleanup: true,
|
||||
})
|
||||
logger.info(`[${requestId}] Previous version cleanup completed`)
|
||||
} catch (cleanupError) {
|
||||
logger.error(
|
||||
`[${requestId}] Failed to clean up previous version ${previousVersionId}`,
|
||||
cleanupError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
await syncMcpToolsForWorkflow({
|
||||
workflowId: id,
|
||||
requestId,
|
||||
state: versionRow.state,
|
||||
context: 'activate',
|
||||
})
|
||||
|
||||
return createSuccessResponse({
|
||||
success: true,
|
||||
deployedAt: result.deployedAt,
|
||||
warnings: triggerSaveResult.warnings,
|
||||
})
|
||||
} catch (error: any) {
|
||||
logger.error(`[${requestId}] Error activating deployment for workflow: ${id}`, error)
|
||||
return createErrorResponse(error.message || 'Failed to activate deployment', 500)
|
||||
}
|
||||
}
|
||||
@@ -4,8 +4,17 @@ import { and, eq } from 'drizzle-orm'
|
||||
import type { NextRequest } from 'next/server'
|
||||
import { z } from 'zod'
|
||||
import { generateRequestId } from '@/lib/core/utils/request'
|
||||
import { syncMcpToolsForWorkflow } from '@/lib/mcp/workflow-mcp-sync'
|
||||
import { restorePreviousVersionWebhooks, saveTriggerWebhooksForDeploy } from '@/lib/webhooks/deploy'
|
||||
import { activateWorkflowVersion } from '@/lib/workflows/persistence/utils'
|
||||
import {
|
||||
cleanupDeploymentVersion,
|
||||
createSchedulesForDeploy,
|
||||
validateWorkflowSchedules,
|
||||
} from '@/lib/workflows/schedules'
|
||||
import { validateWorkflowPermissions } from '@/lib/workflows/utils'
|
||||
import { createErrorResponse, createSuccessResponse } from '@/app/api/workflows/utils'
|
||||
import type { BlockState } from '@/stores/workflows/workflow/types'
|
||||
|
||||
const logger = createLogger('WorkflowDeploymentVersionAPI')
|
||||
|
||||
@@ -23,10 +32,14 @@ const patchBodySchema = z
|
||||
.max(500, 'Description must be 500 characters or less')
|
||||
.nullable()
|
||||
.optional(),
|
||||
isActive: z.literal(true).optional(), // Set to true to activate this version
|
||||
})
|
||||
.refine((data) => data.name !== undefined || data.description !== undefined, {
|
||||
message: 'At least one of name or description must be provided',
|
||||
})
|
||||
.refine(
|
||||
(data) => data.name !== undefined || data.description !== undefined || data.isActive === true,
|
||||
{
|
||||
message: 'At least one of name, description, or isActive must be provided',
|
||||
}
|
||||
)
|
||||
|
||||
export const dynamic = 'force-dynamic'
|
||||
export const runtime = 'nodejs'
|
||||
@@ -82,7 +95,22 @@ export async function PATCH(
|
||||
const { id, version } = await params
|
||||
|
||||
try {
|
||||
const { error } = await validateWorkflowPermissions(id, requestId, 'write')
|
||||
const body = await request.json()
|
||||
const validation = patchBodySchema.safeParse(body)
|
||||
|
||||
if (!validation.success) {
|
||||
return createErrorResponse(validation.error.errors[0]?.message || 'Invalid request body', 400)
|
||||
}
|
||||
|
||||
const { name, description, isActive } = validation.data
|
||||
|
||||
// Activation requires admin permission, other updates require write
|
||||
const requiredPermission = isActive ? 'admin' : 'write'
|
||||
const {
|
||||
error,
|
||||
session,
|
||||
workflow: workflowData,
|
||||
} = await validateWorkflowPermissions(id, requestId, requiredPermission)
|
||||
if (error) {
|
||||
return createErrorResponse(error.message, error.status)
|
||||
}
|
||||
@@ -92,15 +120,193 @@ export async function PATCH(
|
||||
return createErrorResponse('Invalid version', 400)
|
||||
}
|
||||
|
||||
const body = await request.json()
|
||||
const validation = patchBodySchema.safeParse(body)
|
||||
// Handle activation
|
||||
if (isActive) {
|
||||
const actorUserId = session?.user?.id
|
||||
if (!actorUserId) {
|
||||
logger.warn(`[${requestId}] Unable to resolve actor user for deployment activation: ${id}`)
|
||||
return createErrorResponse('Unable to determine activating user', 400)
|
||||
}
|
||||
|
||||
if (!validation.success) {
|
||||
return createErrorResponse(validation.error.errors[0]?.message || 'Invalid request body', 400)
|
||||
const [versionRow] = await db
|
||||
.select({
|
||||
id: workflowDeploymentVersion.id,
|
||||
state: workflowDeploymentVersion.state,
|
||||
})
|
||||
.from(workflowDeploymentVersion)
|
||||
.where(
|
||||
and(
|
||||
eq(workflowDeploymentVersion.workflowId, id),
|
||||
eq(workflowDeploymentVersion.version, versionNum)
|
||||
)
|
||||
)
|
||||
.limit(1)
|
||||
|
||||
if (!versionRow?.state) {
|
||||
return createErrorResponse('Deployment version not found', 404)
|
||||
}
|
||||
|
||||
const [currentActiveVersion] = await db
|
||||
.select({ id: workflowDeploymentVersion.id })
|
||||
.from(workflowDeploymentVersion)
|
||||
.where(
|
||||
and(
|
||||
eq(workflowDeploymentVersion.workflowId, id),
|
||||
eq(workflowDeploymentVersion.isActive, true)
|
||||
)
|
||||
)
|
||||
.limit(1)
|
||||
|
||||
const previousVersionId = currentActiveVersion?.id
|
||||
|
||||
const deployedState = versionRow.state as { blocks?: Record<string, BlockState> }
|
||||
const blocks = deployedState.blocks
|
||||
if (!blocks || typeof blocks !== 'object') {
|
||||
return createErrorResponse('Invalid deployed state structure', 500)
|
||||
}
|
||||
|
||||
const scheduleValidation = validateWorkflowSchedules(blocks)
|
||||
if (!scheduleValidation.isValid) {
|
||||
return createErrorResponse(
|
||||
`Invalid schedule configuration: ${scheduleValidation.error}`,
|
||||
400
|
||||
)
|
||||
}
|
||||
|
||||
const triggerSaveResult = await saveTriggerWebhooksForDeploy({
|
||||
request,
|
||||
workflowId: id,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
userId: actorUserId,
|
||||
blocks,
|
||||
requestId,
|
||||
deploymentVersionId: versionRow.id,
|
||||
previousVersionId,
|
||||
forceRecreateSubscriptions: true,
|
||||
})
|
||||
|
||||
if (!triggerSaveResult.success) {
|
||||
return createErrorResponse(
|
||||
triggerSaveResult.error?.message || 'Failed to sync trigger configuration',
|
||||
triggerSaveResult.error?.status || 500
|
||||
)
|
||||
}
|
||||
|
||||
const scheduleResult = await createSchedulesForDeploy(id, blocks, db, versionRow.id)
|
||||
|
||||
if (!scheduleResult.success) {
|
||||
await cleanupDeploymentVersion({
|
||||
workflowId: id,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
requestId,
|
||||
deploymentVersionId: versionRow.id,
|
||||
})
|
||||
if (previousVersionId) {
|
||||
await restorePreviousVersionWebhooks({
|
||||
request,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
userId: actorUserId,
|
||||
previousVersionId,
|
||||
requestId,
|
||||
})
|
||||
}
|
||||
return createErrorResponse(scheduleResult.error || 'Failed to sync schedules', 500)
|
||||
}
|
||||
|
||||
const result = await activateWorkflowVersion({ workflowId: id, version: versionNum })
|
||||
if (!result.success) {
|
||||
await cleanupDeploymentVersion({
|
||||
workflowId: id,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
requestId,
|
||||
deploymentVersionId: versionRow.id,
|
||||
})
|
||||
if (previousVersionId) {
|
||||
await restorePreviousVersionWebhooks({
|
||||
request,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
userId: actorUserId,
|
||||
previousVersionId,
|
||||
requestId,
|
||||
})
|
||||
}
|
||||
return createErrorResponse(result.error || 'Failed to activate deployment', 400)
|
||||
}
|
||||
|
||||
if (previousVersionId && previousVersionId !== versionRow.id) {
|
||||
try {
|
||||
logger.info(
|
||||
`[${requestId}] Cleaning up previous version ${previousVersionId} webhooks/schedules`
|
||||
)
|
||||
await cleanupDeploymentVersion({
|
||||
workflowId: id,
|
||||
workflow: workflowData as Record<string, unknown>,
|
||||
requestId,
|
||||
deploymentVersionId: previousVersionId,
|
||||
skipExternalCleanup: true,
|
||||
})
|
||||
logger.info(`[${requestId}] Previous version cleanup completed`)
|
||||
} catch (cleanupError) {
|
||||
logger.error(
|
||||
`[${requestId}] Failed to clean up previous version ${previousVersionId}`,
|
||||
cleanupError
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
await syncMcpToolsForWorkflow({
|
||||
workflowId: id,
|
||||
requestId,
|
||||
state: versionRow.state,
|
||||
context: 'activate',
|
||||
})
|
||||
|
||||
// Apply name/description updates if provided alongside activation
|
||||
let updatedName: string | null | undefined
|
||||
let updatedDescription: string | null | undefined
|
||||
if (name !== undefined || description !== undefined) {
|
||||
const activationUpdateData: { name?: string; description?: string | null } = {}
|
||||
if (name !== undefined) {
|
||||
activationUpdateData.name = name
|
||||
}
|
||||
if (description !== undefined) {
|
||||
activationUpdateData.description = description
|
||||
}
|
||||
|
||||
const [updated] = await db
|
||||
.update(workflowDeploymentVersion)
|
||||
.set(activationUpdateData)
|
||||
.where(
|
||||
and(
|
||||
eq(workflowDeploymentVersion.workflowId, id),
|
||||
eq(workflowDeploymentVersion.version, versionNum)
|
||||
)
|
||||
)
|
||||
.returning({
|
||||
name: workflowDeploymentVersion.name,
|
||||
description: workflowDeploymentVersion.description,
|
||||
})
|
||||
|
||||
if (updated) {
|
||||
updatedName = updated.name
|
||||
updatedDescription = updated.description
|
||||
logger.info(
|
||||
`[${requestId}] Updated deployment version ${version} metadata during activation`,
|
||||
{ name: activationUpdateData.name, description: activationUpdateData.description }
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
return createSuccessResponse({
|
||||
success: true,
|
||||
deployedAt: result.deployedAt,
|
||||
warnings: triggerSaveResult.warnings,
|
||||
...(updatedName !== undefined && { name: updatedName }),
|
||||
...(updatedDescription !== undefined && { description: updatedDescription }),
|
||||
})
|
||||
}
|
||||
|
||||
const { name, description } = validation.data
|
||||
|
||||
// Handle name/description updates
|
||||
const updateData: { name?: string; description?: string | null } = {}
|
||||
if (name !== undefined) {
|
||||
updateData.name = name
|
||||
|
||||
@@ -1,286 +0,0 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { v4 as uuidv4 } from 'uuid'
|
||||
import { z } from 'zod'
|
||||
import { checkHybridAuth } from '@/lib/auth/hybrid'
|
||||
import { getTimeoutErrorMessage, isTimeoutError } from '@/lib/core/execution-limits'
|
||||
import { generateRequestId } from '@/lib/core/utils/request'
|
||||
import { SSE_HEADERS } from '@/lib/core/utils/sse'
|
||||
import { markExecutionCancelled } from '@/lib/execution/cancellation'
|
||||
import { preprocessExecution } from '@/lib/execution/preprocessing'
|
||||
import { LoggingSession } from '@/lib/logs/execution/logging-session'
|
||||
import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
|
||||
import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
|
||||
import { createSSECallbacks } from '@/lib/workflows/executor/execution-events'
|
||||
import { ExecutionSnapshot } from '@/executor/execution/snapshot'
|
||||
import type { ExecutionMetadata, SerializableExecutionState } from '@/executor/execution/types'
|
||||
import { hasExecutionResult } from '@/executor/utils/errors'
|
||||
|
||||
const logger = createLogger('ExecuteFromBlockAPI')
|
||||
|
||||
const ExecuteFromBlockSchema = z.object({
|
||||
startBlockId: z.string().min(1, 'Start block ID is required'),
|
||||
sourceSnapshot: z.object({
|
||||
blockStates: z.record(z.any()),
|
||||
executedBlocks: z.array(z.string()),
|
||||
blockLogs: z.array(z.any()),
|
||||
decisions: z.object({
|
||||
router: z.record(z.string()),
|
||||
condition: z.record(z.string()),
|
||||
}),
|
||||
completedLoops: z.array(z.string()),
|
||||
loopExecutions: z.record(z.any()).optional(),
|
||||
parallelExecutions: z.record(z.any()).optional(),
|
||||
parallelBlockMapping: z.record(z.any()).optional(),
|
||||
activeExecutionPath: z.array(z.string()),
|
||||
}),
|
||||
input: z.any().optional(),
|
||||
})
|
||||
|
||||
export const runtime = 'nodejs'
|
||||
export const dynamic = 'force-dynamic'
|
||||
|
||||
export async function POST(req: NextRequest, { params }: { params: Promise<{ id: string }> }) {
|
||||
const requestId = generateRequestId()
|
||||
const { id: workflowId } = await params
|
||||
|
||||
try {
|
||||
const auth = await checkHybridAuth(req, { requireWorkflowId: false })
|
||||
if (!auth.success || !auth.userId) {
|
||||
return NextResponse.json({ error: auth.error || 'Unauthorized' }, { status: 401 })
|
||||
}
|
||||
const userId = auth.userId
|
||||
|
||||
let body: unknown
|
||||
try {
|
||||
body = await req.json()
|
||||
} catch {
|
||||
return NextResponse.json({ error: 'Invalid JSON body' }, { status: 400 })
|
||||
}
|
||||
|
||||
const validation = ExecuteFromBlockSchema.safeParse(body)
|
||||
if (!validation.success) {
|
||||
logger.warn(`[${requestId}] Invalid request body:`, validation.error.errors)
|
||||
return NextResponse.json(
|
||||
{
|
||||
error: 'Invalid request body',
|
||||
details: validation.error.errors.map((e) => ({
|
||||
path: e.path.join('.'),
|
||||
message: e.message,
|
||||
})),
|
||||
},
|
||||
{ status: 400 }
|
||||
)
|
||||
}
|
||||
|
||||
const { startBlockId, sourceSnapshot, input } = validation.data
|
||||
const executionId = uuidv4()
|
||||
|
||||
// Run preprocessing checks (billing, rate limits, usage limits)
|
||||
const preprocessResult = await preprocessExecution({
|
||||
workflowId,
|
||||
userId,
|
||||
triggerType: 'manual',
|
||||
executionId,
|
||||
requestId,
|
||||
checkRateLimit: false, // Manual executions don't rate limit
|
||||
checkDeployment: false, // Run-from-block doesn't require deployment
|
||||
})
|
||||
|
||||
if (!preprocessResult.success) {
|
||||
const { error } = preprocessResult
|
||||
logger.warn(`[${requestId}] Preprocessing failed for run-from-block`, {
|
||||
workflowId,
|
||||
error: error?.message,
|
||||
statusCode: error?.statusCode,
|
||||
})
|
||||
return NextResponse.json(
|
||||
{ error: error?.message || 'Execution blocked' },
|
||||
{ status: error?.statusCode || 500 }
|
||||
)
|
||||
}
|
||||
|
||||
const workflowRecord = preprocessResult.workflowRecord
|
||||
if (!workflowRecord?.workspaceId) {
|
||||
return NextResponse.json({ error: 'Workflow not found or has no workspace' }, { status: 404 })
|
||||
}
|
||||
|
||||
const workspaceId = workflowRecord.workspaceId
|
||||
const workflowUserId = workflowRecord.userId
|
||||
|
||||
logger.info(`[${requestId}] Starting run-from-block execution`, {
|
||||
workflowId,
|
||||
startBlockId,
|
||||
executedBlocksCount: sourceSnapshot.executedBlocks.length,
|
||||
billingActorUserId: preprocessResult.actorUserId,
|
||||
})
|
||||
|
||||
const loggingSession = new LoggingSession(workflowId, executionId, 'manual', requestId)
|
||||
const abortController = new AbortController()
|
||||
let isStreamClosed = false
|
||||
let isTimedOut = false
|
||||
|
||||
const syncTimeout = preprocessResult.executionTimeout?.sync
|
||||
let timeoutId: NodeJS.Timeout | undefined
|
||||
if (syncTimeout) {
|
||||
timeoutId = setTimeout(() => {
|
||||
isTimedOut = true
|
||||
abortController.abort()
|
||||
}, syncTimeout)
|
||||
}
|
||||
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
async start(controller) {
|
||||
const { sendEvent, onBlockStart, onBlockComplete, onStream } = createSSECallbacks({
|
||||
executionId,
|
||||
workflowId,
|
||||
controller,
|
||||
isStreamClosed: () => isStreamClosed,
|
||||
setStreamClosed: () => {
|
||||
isStreamClosed = true
|
||||
},
|
||||
})
|
||||
|
||||
const metadata: ExecutionMetadata = {
|
||||
requestId,
|
||||
workflowId,
|
||||
userId,
|
||||
executionId,
|
||||
triggerType: 'manual',
|
||||
workspaceId,
|
||||
workflowUserId,
|
||||
useDraftState: true,
|
||||
isClientSession: true,
|
||||
startTime: new Date().toISOString(),
|
||||
}
|
||||
|
||||
const snapshot = new ExecutionSnapshot(metadata, {}, input || {}, {})
|
||||
|
||||
try {
|
||||
const startTime = new Date()
|
||||
|
||||
sendEvent({
|
||||
type: 'execution:started',
|
||||
timestamp: startTime.toISOString(),
|
||||
executionId,
|
||||
workflowId,
|
||||
data: { startTime: startTime.toISOString() },
|
||||
})
|
||||
|
||||
const result = await executeWorkflowCore({
|
||||
snapshot,
|
||||
loggingSession,
|
||||
abortSignal: abortController.signal,
|
||||
runFromBlock: {
|
||||
startBlockId,
|
||||
sourceSnapshot: sourceSnapshot as SerializableExecutionState,
|
||||
},
|
||||
callbacks: { onBlockStart, onBlockComplete, onStream },
|
||||
})
|
||||
|
||||
if (result.status === 'cancelled') {
|
||||
if (isTimedOut && syncTimeout) {
|
||||
const timeoutErrorMessage = getTimeoutErrorMessage(null, syncTimeout)
|
||||
logger.info(`[${requestId}] Run-from-block execution timed out`, {
|
||||
timeoutMs: syncTimeout,
|
||||
})
|
||||
|
||||
await loggingSession.markAsFailed(timeoutErrorMessage)
|
||||
|
||||
sendEvent({
|
||||
type: 'execution:error',
|
||||
timestamp: new Date().toISOString(),
|
||||
executionId,
|
||||
workflowId,
|
||||
data: {
|
||||
error: timeoutErrorMessage,
|
||||
duration: result.metadata?.duration || 0,
|
||||
},
|
||||
})
|
||||
} else {
|
||||
sendEvent({
|
||||
type: 'execution:cancelled',
|
||||
timestamp: new Date().toISOString(),
|
||||
executionId,
|
||||
workflowId,
|
||||
data: { duration: result.metadata?.duration || 0 },
|
||||
})
|
||||
}
|
||||
} else {
|
||||
sendEvent({
|
||||
type: 'execution:completed',
|
||||
timestamp: new Date().toISOString(),
|
||||
executionId,
|
||||
workflowId,
|
||||
data: {
|
||||
success: result.success,
|
||||
output: result.output,
|
||||
duration: result.metadata?.duration || 0,
|
||||
startTime: result.metadata?.startTime || startTime.toISOString(),
|
||||
endTime: result.metadata?.endTime || new Date().toISOString(),
|
||||
},
|
||||
})
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
const isTimeout = isTimeoutError(error) || isTimedOut
|
||||
const errorMessage = isTimeout
|
||||
? getTimeoutErrorMessage(error, syncTimeout)
|
||||
: error instanceof Error
|
||||
? error.message
|
||||
: 'Unknown error'
|
||||
|
||||
logger.error(`[${requestId}] Run-from-block execution failed: ${errorMessage}`, {
|
||||
isTimeout,
|
||||
})
|
||||
|
||||
const executionResult = hasExecutionResult(error) ? error.executionResult : undefined
|
||||
const { traceSpans, totalDuration } = executionResult
|
||||
? buildTraceSpans(executionResult)
|
||||
: { traceSpans: [], totalDuration: 0 }
|
||||
|
||||
await loggingSession.safeCompleteWithError({
|
||||
totalDurationMs: totalDuration || executionResult?.metadata?.duration,
|
||||
error: { message: errorMessage },
|
||||
traceSpans,
|
||||
})
|
||||
|
||||
sendEvent({
|
||||
type: 'execution:error',
|
||||
timestamp: new Date().toISOString(),
|
||||
executionId,
|
||||
workflowId,
|
||||
data: {
|
||||
error: executionResult?.error || errorMessage,
|
||||
duration: executionResult?.metadata?.duration || 0,
|
||||
},
|
||||
})
|
||||
} finally {
|
||||
if (timeoutId) clearTimeout(timeoutId)
|
||||
if (!isStreamClosed) {
|
||||
try {
|
||||
controller.enqueue(new TextEncoder().encode('data: [DONE]\n\n'))
|
||||
controller.close()
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
},
|
||||
cancel() {
|
||||
isStreamClosed = true
|
||||
if (timeoutId) clearTimeout(timeoutId)
|
||||
abortController.abort()
|
||||
markExecutionCancelled(executionId).catch(() => {})
|
||||
},
|
||||
})
|
||||
|
||||
return new NextResponse(stream, {
|
||||
headers: { ...SSE_HEADERS, 'X-Execution-Id': executionId },
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
const errorMessage = error instanceof Error ? error.message : 'Unknown error'
|
||||
logger.error(`[${requestId}] Failed to start run-from-block execution:`, error)
|
||||
return NextResponse.json(
|
||||
{ error: errorMessage || 'Failed to start execution' },
|
||||
{ status: 500 }
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -1,10 +1,9 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { tasks } from '@trigger.dev/sdk'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { validate as uuidValidate, v4 as uuidv4 } from 'uuid'
|
||||
import { z } from 'zod'
|
||||
import { checkHybridAuth } from '@/lib/auth/hybrid'
|
||||
import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
|
||||
import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
|
||||
import {
|
||||
createTimeoutAbortController,
|
||||
getTimeoutErrorMessage,
|
||||
@@ -31,7 +30,7 @@ import {
|
||||
} from '@/lib/workflows/persistence/utils'
|
||||
import { createStreamingResponse } from '@/lib/workflows/streaming/streaming'
|
||||
import { createHttpResponseFromBlock, workflowHasResponseBlock } from '@/lib/workflows/utils'
|
||||
import type { WorkflowExecutionPayload } from '@/background/workflow-execution'
|
||||
import { executeWorkflowJob, type WorkflowExecutionPayload } from '@/background/workflow-execution'
|
||||
import { normalizeName } from '@/executor/constants'
|
||||
import { ExecutionSnapshot } from '@/executor/execution/snapshot'
|
||||
import type { ExecutionMetadata, IterationContext } from '@/executor/execution/types'
|
||||
@@ -60,6 +59,25 @@ const ExecuteWorkflowSchema = z.object({
|
||||
})
|
||||
.optional(),
|
||||
stopAfterBlockId: z.string().optional(),
|
||||
runFromBlock: z
|
||||
.object({
|
||||
startBlockId: z.string().min(1, 'Start block ID is required'),
|
||||
sourceSnapshot: z.object({
|
||||
blockStates: z.record(z.any()),
|
||||
executedBlocks: z.array(z.string()),
|
||||
blockLogs: z.array(z.any()),
|
||||
decisions: z.object({
|
||||
router: z.record(z.string()),
|
||||
condition: z.record(z.string()),
|
||||
}),
|
||||
completedLoops: z.array(z.string()),
|
||||
loopExecutions: z.record(z.any()).optional(),
|
||||
parallelExecutions: z.record(z.any()).optional(),
|
||||
parallelBlockMapping: z.record(z.any()).optional(),
|
||||
activeExecutionPath: z.array(z.string()),
|
||||
}),
|
||||
})
|
||||
.optional(),
|
||||
})
|
||||
|
||||
export const runtime = 'nodejs'
|
||||
@@ -124,41 +142,66 @@ type AsyncExecutionParams = {
|
||||
userId: string
|
||||
input: any
|
||||
triggerType: CoreTriggerType
|
||||
executionId: string
|
||||
}
|
||||
|
||||
async function handleAsyncExecution(params: AsyncExecutionParams): Promise<NextResponse> {
|
||||
const { requestId, workflowId, userId, input, triggerType } = params
|
||||
|
||||
if (!isTriggerDevEnabled) {
|
||||
logger.warn(`[${requestId}] Async mode requested but TRIGGER_DEV_ENABLED is false`)
|
||||
return NextResponse.json(
|
||||
{ error: 'Async execution is not enabled. Set TRIGGER_DEV_ENABLED=true to use async mode.' },
|
||||
{ status: 400 }
|
||||
)
|
||||
}
|
||||
const { requestId, workflowId, userId, input, triggerType, executionId } = params
|
||||
|
||||
const payload: WorkflowExecutionPayload = {
|
||||
workflowId,
|
||||
userId,
|
||||
input,
|
||||
triggerType,
|
||||
executionId,
|
||||
}
|
||||
|
||||
try {
|
||||
const handle = await tasks.trigger('workflow-execution', payload)
|
||||
const jobQueue = await getJobQueue()
|
||||
const jobId = await jobQueue.enqueue('workflow-execution', payload, {
|
||||
metadata: { workflowId, userId },
|
||||
})
|
||||
|
||||
logger.info(`[${requestId}] Queued async workflow execution`, {
|
||||
workflowId,
|
||||
jobId: handle.id,
|
||||
jobId,
|
||||
})
|
||||
|
||||
if (shouldExecuteInline()) {
|
||||
void (async () => {
|
||||
try {
|
||||
await jobQueue.startJob(jobId)
|
||||
const output = await executeWorkflowJob(payload)
|
||||
await jobQueue.completeJob(jobId, output)
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
logger.error(`[${requestId}] Async workflow execution failed`, {
|
||||
jobId,
|
||||
error: errorMessage,
|
||||
})
|
||||
try {
|
||||
await jobQueue.markJobFailed(jobId, errorMessage)
|
||||
} catch (markFailedError) {
|
||||
logger.error(`[${requestId}] Failed to mark job as failed`, {
|
||||
jobId,
|
||||
error:
|
||||
markFailedError instanceof Error
|
||||
? markFailedError.message
|
||||
: String(markFailedError),
|
||||
})
|
||||
}
|
||||
}
|
||||
})()
|
||||
}
|
||||
|
||||
return NextResponse.json(
|
||||
{
|
||||
success: true,
|
||||
async: true,
|
||||
jobId: handle.id,
|
||||
jobId,
|
||||
executionId,
|
||||
message: 'Workflow execution queued',
|
||||
statusUrl: `${getBaseUrl()}/api/jobs/${handle.id}`,
|
||||
statusUrl: `${getBaseUrl()}/api/jobs/${jobId}`,
|
||||
},
|
||||
{ status: 202 }
|
||||
)
|
||||
@@ -226,6 +269,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
|
||||
base64MaxBytes,
|
||||
workflowStateOverride,
|
||||
stopAfterBlockId,
|
||||
runFromBlock,
|
||||
} = validation.data
|
||||
|
||||
// For API key and internal JWT auth, the entire body is the input (except for our control fields)
|
||||
@@ -242,6 +286,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
|
||||
base64MaxBytes,
|
||||
workflowStateOverride,
|
||||
stopAfterBlockId: _stopAfterBlockId,
|
||||
runFromBlock: _runFromBlock,
|
||||
workflowId: _workflowId, // Also exclude workflowId used for internal JWT auth
|
||||
...rest
|
||||
} = body
|
||||
@@ -320,6 +365,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
|
||||
userId: actorUserId,
|
||||
input,
|
||||
triggerType: loggingTriggerType,
|
||||
executionId,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -444,6 +490,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
|
||||
includeFileBase64,
|
||||
base64MaxBytes,
|
||||
stopAfterBlockId,
|
||||
runFromBlock,
|
||||
abortSignal: timeoutController.signal,
|
||||
})
|
||||
|
||||
@@ -492,6 +539,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
|
||||
|
||||
const filteredResult = {
|
||||
success: result.success,
|
||||
executionId,
|
||||
output: outputWithBase64,
|
||||
error: result.error,
|
||||
metadata: result.metadata
|
||||
@@ -783,6 +831,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
|
||||
includeFileBase64,
|
||||
base64MaxBytes,
|
||||
stopAfterBlockId,
|
||||
runFromBlock,
|
||||
})
|
||||
|
||||
if (result.status === 'paused') {
|
||||
|
||||
@@ -508,8 +508,10 @@ export default function TemplateDetails({ isWorkspaceContext = false }: Template
|
||||
|
||||
setIsApproving(true)
|
||||
try {
|
||||
const response = await fetch(`/api/templates/${template.id}/approve`, {
|
||||
method: 'POST',
|
||||
const response = await fetch(`/api/templates/${template.id}`, {
|
||||
method: 'PUT',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ status: 'approved' }),
|
||||
})
|
||||
|
||||
if (response.ok) {
|
||||
@@ -531,8 +533,10 @@ export default function TemplateDetails({ isWorkspaceContext = false }: Template
|
||||
|
||||
setIsRejecting(true)
|
||||
try {
|
||||
const response = await fetch(`/api/templates/${template.id}/reject`, {
|
||||
method: 'POST',
|
||||
const response = await fetch(`/api/templates/${template.id}`, {
|
||||
method: 'PUT',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ status: 'rejected' }),
|
||||
})
|
||||
|
||||
if (response.ok) {
|
||||
@@ -554,10 +558,11 @@ export default function TemplateDetails({ isWorkspaceContext = false }: Template
|
||||
|
||||
setIsVerifying(true)
|
||||
try {
|
||||
const endpoint = `/api/creators/${template.creator.id}/verify`
|
||||
const method = template.creator.verified ? 'DELETE' : 'POST'
|
||||
|
||||
const response = await fetch(endpoint, { method })
|
||||
const response = await fetch(`/api/creators/${template.creator.id}`, {
|
||||
method: 'PUT',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ verified: !template.creator.verified }),
|
||||
})
|
||||
|
||||
if (response.ok) {
|
||||
// Refresh page to show updated verification status
|
||||
|
||||
@@ -12,7 +12,6 @@ import {
|
||||
Tooltip,
|
||||
} from '@/components/emcn'
|
||||
import { Skeleton } from '@/components/ui'
|
||||
import { getEnv, isTruthy } from '@/lib/core/config/env'
|
||||
import { OutputSelect } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/chat/components/output-select/output-select'
|
||||
|
||||
interface WorkflowDeploymentInfo {
|
||||
@@ -78,7 +77,6 @@ export function ApiDeploy({
|
||||
async: false,
|
||||
})
|
||||
|
||||
const isAsyncEnabled = isTruthy(getEnv('NEXT_PUBLIC_TRIGGER_DEV_ENABLED'))
|
||||
const info = deploymentInfo ? { ...deploymentInfo, needsRedeployment } : null
|
||||
|
||||
const getBaseEndpoint = () => {
|
||||
@@ -272,7 +270,7 @@ response = requests.post(
|
||||
)
|
||||
|
||||
job = response.json()
|
||||
print(job) # Contains job_id for status checking`
|
||||
print(job) # Contains jobId and executionId`
|
||||
|
||||
case 'javascript':
|
||||
return `const response = await fetch("${endpoint}", {
|
||||
@@ -286,7 +284,7 @@ print(job) # Contains job_id for status checking`
|
||||
});
|
||||
|
||||
const job = await response.json();
|
||||
console.log(job); // Contains job_id for status checking`
|
||||
console.log(job); // Contains jobId and executionId`
|
||||
|
||||
case 'typescript':
|
||||
return `const response = await fetch("${endpoint}", {
|
||||
@@ -299,8 +297,8 @@ console.log(job); // Contains job_id for status checking`
|
||||
body: JSON.stringify(${JSON.stringify(payload)})
|
||||
});
|
||||
|
||||
const job: { job_id: string } = await response.json();
|
||||
console.log(job); // Contains job_id for status checking`
|
||||
const job: { jobId: string; executionId: string } = await response.json();
|
||||
console.log(job); // Contains jobId and executionId`
|
||||
|
||||
default:
|
||||
return ''
|
||||
@@ -539,55 +537,49 @@ console.log(limits);`
|
||||
/>
|
||||
</div>
|
||||
|
||||
{isAsyncEnabled && (
|
||||
<div>
|
||||
<div className='mb-[6.5px] flex items-center justify-between'>
|
||||
<Label className='block pl-[2px] font-medium text-[13px] text-[var(--text-primary)]'>
|
||||
Run workflow (async)
|
||||
</Label>
|
||||
<div className='flex items-center gap-[6px]'>
|
||||
<Tooltip.Root>
|
||||
<Tooltip.Trigger asChild>
|
||||
<Button
|
||||
variant='ghost'
|
||||
onClick={() => handleCopy('async', getAsyncCommand())}
|
||||
aria-label='Copy command'
|
||||
className='!p-1.5 -my-1.5'
|
||||
>
|
||||
{copied.async ? (
|
||||
<Check className='h-3 w-3' />
|
||||
) : (
|
||||
<Clipboard className='h-3 w-3' />
|
||||
)}
|
||||
</Button>
|
||||
</Tooltip.Trigger>
|
||||
<Tooltip.Content>
|
||||
<span>{copied.async ? 'Copied' : 'Copy'}</span>
|
||||
</Tooltip.Content>
|
||||
</Tooltip.Root>
|
||||
<Combobox
|
||||
size='sm'
|
||||
className='!w-fit !py-[2px] min-w-[100px] rounded-[6px] px-[9px]'
|
||||
options={[
|
||||
{ label: 'Execute Job', value: 'execute' },
|
||||
{ label: 'Check Status', value: 'status' },
|
||||
{ label: 'Rate Limits', value: 'rate-limits' },
|
||||
]}
|
||||
value={asyncExampleType}
|
||||
onChange={(value) => setAsyncExampleType(value as AsyncExampleType)}
|
||||
align='end'
|
||||
dropdownWidth={160}
|
||||
/>
|
||||
</div>
|
||||
<div>
|
||||
<div className='mb-[6.5px] flex items-center justify-between'>
|
||||
<Label className='block pl-[2px] font-medium text-[13px] text-[var(--text-primary)]'>
|
||||
Run workflow (async)
|
||||
</Label>
|
||||
<div className='flex items-center gap-[6px]'>
|
||||
<Tooltip.Root>
|
||||
<Tooltip.Trigger asChild>
|
||||
<Button
|
||||
variant='ghost'
|
||||
onClick={() => handleCopy('async', getAsyncCommand())}
|
||||
aria-label='Copy command'
|
||||
className='!p-1.5 -my-1.5'
|
||||
>
|
||||
{copied.async ? <Check className='h-3 w-3' /> : <Clipboard className='h-3 w-3' />}
|
||||
</Button>
|
||||
</Tooltip.Trigger>
|
||||
<Tooltip.Content>
|
||||
<span>{copied.async ? 'Copied' : 'Copy'}</span>
|
||||
</Tooltip.Content>
|
||||
</Tooltip.Root>
|
||||
<Combobox
|
||||
size='sm'
|
||||
className='!w-fit !py-[2px] min-w-[100px] rounded-[6px] px-[9px]'
|
||||
options={[
|
||||
{ label: 'Execute Job', value: 'execute' },
|
||||
{ label: 'Check Status', value: 'status' },
|
||||
{ label: 'Rate Limits', value: 'rate-limits' },
|
||||
]}
|
||||
value={asyncExampleType}
|
||||
onChange={(value) => setAsyncExampleType(value as AsyncExampleType)}
|
||||
align='end'
|
||||
dropdownWidth={160}
|
||||
/>
|
||||
</div>
|
||||
<Code.Viewer
|
||||
code={getAsyncCommand()}
|
||||
language={LANGUAGE_SYNTAX[language]}
|
||||
wrapText
|
||||
className='!min-h-0 rounded-[4px] border border-[var(--border-1)]'
|
||||
/>
|
||||
</div>
|
||||
)}
|
||||
<Code.Viewer
|
||||
code={getAsyncCommand()}
|
||||
language={LANGUAGE_SYNTAX[language]}
|
||||
wrapText
|
||||
className='!min-h-0 rounded-[4px] border border-[var(--border-1)]'
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
@@ -20,6 +20,7 @@ export type WorkflowExecutionPayload = {
|
||||
userId: string
|
||||
input?: any
|
||||
triggerType?: CoreTriggerType
|
||||
executionId?: string
|
||||
metadata?: Record<string, any>
|
||||
}
|
||||
|
||||
@@ -30,7 +31,7 @@ export type WorkflowExecutionPayload = {
|
||||
*/
|
||||
export async function executeWorkflowJob(payload: WorkflowExecutionPayload) {
|
||||
const workflowId = payload.workflowId
|
||||
const executionId = uuidv4()
|
||||
const executionId = payload.executionId || uuidv4()
|
||||
const requestId = executionId.slice(0, 8)
|
||||
|
||||
logger.info(`[${requestId}] Starting workflow execution job: ${workflowId}`, {
|
||||
|
||||
@@ -549,11 +549,12 @@ export function useActivateDeploymentVersion() {
|
||||
workflowId,
|
||||
version,
|
||||
}: ActivateVersionVariables): Promise<ActivateVersionResult> => {
|
||||
const response = await fetch(`/api/workflows/${workflowId}/deployments/${version}/activate`, {
|
||||
method: 'POST',
|
||||
const response = await fetch(`/api/workflows/${workflowId}/deployments/${version}`, {
|
||||
method: 'PATCH',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
},
|
||||
body: JSON.stringify({ isActive: true }),
|
||||
})
|
||||
|
||||
if (!response.ok) {
|
||||
|
||||
@@ -211,12 +211,16 @@ export function useExecutionStream() {
|
||||
currentExecutionRef.current = null
|
||||
|
||||
try {
|
||||
const response = await fetch(`/api/workflows/${workflowId}/execute-from-block`, {
|
||||
const response = await fetch(`/api/workflows/${workflowId}/execute`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
},
|
||||
body: JSON.stringify({ startBlockId, sourceSnapshot, input }),
|
||||
body: JSON.stringify({
|
||||
stream: true,
|
||||
input,
|
||||
runFromBlock: { startBlockId, sourceSnapshot },
|
||||
}),
|
||||
signal: abortController.signal,
|
||||
})
|
||||
|
||||
|
||||
113
apps/sim/lib/core/async-jobs/backends/database.ts
Normal file
113
apps/sim/lib/core/async-jobs/backends/database.ts
Normal file
@@ -0,0 +1,113 @@
|
||||
import { asyncJobs, db } from '@sim/db'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { eq, sql } from 'drizzle-orm'
|
||||
import {
|
||||
type EnqueueOptions,
|
||||
JOB_STATUS,
|
||||
type Job,
|
||||
type JobMetadata,
|
||||
type JobQueueBackend,
|
||||
type JobStatus,
|
||||
type JobType,
|
||||
} from '@/lib/core/async-jobs/types'
|
||||
|
||||
const logger = createLogger('DatabaseJobQueue')
|
||||
|
||||
type AsyncJobRow = typeof asyncJobs.$inferSelect
|
||||
|
||||
function rowToJob(row: AsyncJobRow): Job {
|
||||
return {
|
||||
id: row.id,
|
||||
type: row.type as JobType,
|
||||
payload: row.payload,
|
||||
status: row.status as JobStatus,
|
||||
createdAt: row.createdAt,
|
||||
startedAt: row.startedAt ?? undefined,
|
||||
completedAt: row.completedAt ?? undefined,
|
||||
attempts: row.attempts,
|
||||
maxAttempts: row.maxAttempts,
|
||||
error: row.error ?? undefined,
|
||||
output: row.output as unknown,
|
||||
metadata: (row.metadata ?? {}) as JobMetadata,
|
||||
}
|
||||
}
|
||||
|
||||
export class DatabaseJobQueue implements JobQueueBackend {
|
||||
async enqueue<TPayload>(
|
||||
type: JobType,
|
||||
payload: TPayload,
|
||||
options?: EnqueueOptions
|
||||
): Promise<string> {
|
||||
const jobId = `run_${crypto.randomUUID().replace(/-/g, '').slice(0, 20)}`
|
||||
const now = new Date()
|
||||
|
||||
await db.insert(asyncJobs).values({
|
||||
id: jobId,
|
||||
type,
|
||||
payload: payload as Record<string, unknown>,
|
||||
status: JOB_STATUS.PENDING,
|
||||
createdAt: now,
|
||||
attempts: 0,
|
||||
maxAttempts: options?.maxAttempts ?? 3,
|
||||
metadata: (options?.metadata ?? {}) as Record<string, unknown>,
|
||||
updatedAt: now,
|
||||
})
|
||||
|
||||
logger.debug('Enqueued job', { jobId, type })
|
||||
return jobId
|
||||
}
|
||||
|
||||
async getJob(jobId: string): Promise<Job | null> {
|
||||
const [row] = await db.select().from(asyncJobs).where(eq(asyncJobs.id, jobId)).limit(1)
|
||||
|
||||
return row ? rowToJob(row) : null
|
||||
}
|
||||
|
||||
async startJob(jobId: string): Promise<void> {
|
||||
const now = new Date()
|
||||
|
||||
await db
|
||||
.update(asyncJobs)
|
||||
.set({
|
||||
status: JOB_STATUS.PROCESSING,
|
||||
startedAt: now,
|
||||
attempts: sql`${asyncJobs.attempts} + 1`,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(asyncJobs.id, jobId))
|
||||
|
||||
logger.debug('Started job', { jobId })
|
||||
}
|
||||
|
||||
async completeJob(jobId: string, output: unknown): Promise<void> {
|
||||
const now = new Date()
|
||||
|
||||
await db
|
||||
.update(asyncJobs)
|
||||
.set({
|
||||
status: JOB_STATUS.COMPLETED,
|
||||
completedAt: now,
|
||||
output: output as Record<string, unknown>,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(asyncJobs.id, jobId))
|
||||
|
||||
logger.debug('Completed job', { jobId })
|
||||
}
|
||||
|
||||
async markJobFailed(jobId: string, error: string): Promise<void> {
|
||||
const now = new Date()
|
||||
|
||||
await db
|
||||
.update(asyncJobs)
|
||||
.set({
|
||||
status: JOB_STATUS.FAILED,
|
||||
completedAt: now,
|
||||
error,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(asyncJobs.id, jobId))
|
||||
|
||||
logger.debug('Marked job as failed', { jobId })
|
||||
}
|
||||
}
|
||||
3
apps/sim/lib/core/async-jobs/backends/index.ts
Normal file
3
apps/sim/lib/core/async-jobs/backends/index.ts
Normal file
@@ -0,0 +1,3 @@
|
||||
export { DatabaseJobQueue } from './database'
|
||||
export { RedisJobQueue } from './redis'
|
||||
export { TriggerDevJobQueue } from './trigger-dev'
|
||||
176
apps/sim/lib/core/async-jobs/backends/redis.test.ts
Normal file
176
apps/sim/lib/core/async-jobs/backends/redis.test.ts
Normal file
@@ -0,0 +1,176 @@
|
||||
/**
|
||||
* @vitest-environment node
|
||||
*/
|
||||
import { createMockRedis, loggerMock, type MockRedis } from '@sim/testing'
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
vi.mock('@sim/logger', () => loggerMock)
|
||||
|
||||
import {
|
||||
JOB_MAX_LIFETIME_SECONDS,
|
||||
JOB_RETENTION_SECONDS,
|
||||
JOB_STATUS,
|
||||
} from '@/lib/core/async-jobs/types'
|
||||
import { RedisJobQueue } from './redis'
|
||||
|
||||
describe('RedisJobQueue', () => {
|
||||
let mockRedis: MockRedis
|
||||
let queue: RedisJobQueue
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
mockRedis = createMockRedis()
|
||||
queue = new RedisJobQueue(mockRedis as never)
|
||||
})
|
||||
|
||||
describe('enqueue', () => {
|
||||
it.concurrent('should create a job with pending status', async () => {
|
||||
const localRedis = createMockRedis()
|
||||
const localQueue = new RedisJobQueue(localRedis as never)
|
||||
|
||||
const jobId = await localQueue.enqueue('workflow-execution', { test: 'data' })
|
||||
|
||||
expect(jobId).toMatch(/^run_/)
|
||||
expect(localRedis.hset).toHaveBeenCalledTimes(1)
|
||||
|
||||
const [key, data] = localRedis.hset.mock.calls[0]
|
||||
expect(key).toBe(`async-jobs:job:${jobId}`)
|
||||
expect(data.status).toBe(JOB_STATUS.PENDING)
|
||||
expect(data.type).toBe('workflow-execution')
|
||||
})
|
||||
|
||||
it.concurrent('should set max lifetime TTL on enqueue', async () => {
|
||||
const localRedis = createMockRedis()
|
||||
const localQueue = new RedisJobQueue(localRedis as never)
|
||||
|
||||
const jobId = await localQueue.enqueue('workflow-execution', { test: 'data' })
|
||||
|
||||
expect(localRedis.expire).toHaveBeenCalledWith(
|
||||
`async-jobs:job:${jobId}`,
|
||||
JOB_MAX_LIFETIME_SECONDS
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
describe('completeJob', () => {
|
||||
it.concurrent('should set status to completed and set TTL', async () => {
|
||||
const localRedis = createMockRedis()
|
||||
const localQueue = new RedisJobQueue(localRedis as never)
|
||||
const jobId = 'run_test123'
|
||||
|
||||
await localQueue.completeJob(jobId, { result: 'success' })
|
||||
|
||||
expect(localRedis.hset).toHaveBeenCalledWith(`async-jobs:job:${jobId}`, {
|
||||
status: JOB_STATUS.COMPLETED,
|
||||
completedAt: expect.any(String),
|
||||
output: JSON.stringify({ result: 'success' }),
|
||||
updatedAt: expect.any(String),
|
||||
})
|
||||
expect(localRedis.expire).toHaveBeenCalledWith(
|
||||
`async-jobs:job:${jobId}`,
|
||||
JOB_RETENTION_SECONDS
|
||||
)
|
||||
})
|
||||
|
||||
it.concurrent('should set TTL to 24 hours (86400 seconds)', async () => {
|
||||
const localRedis = createMockRedis()
|
||||
const localQueue = new RedisJobQueue(localRedis as never)
|
||||
|
||||
await localQueue.completeJob('run_test123', {})
|
||||
|
||||
expect(localRedis.expire).toHaveBeenCalledWith(expect.any(String), 86400)
|
||||
})
|
||||
})
|
||||
|
||||
describe('markJobFailed', () => {
|
||||
it.concurrent('should set status to failed and set TTL', async () => {
|
||||
const localRedis = createMockRedis()
|
||||
const localQueue = new RedisJobQueue(localRedis as never)
|
||||
const jobId = 'run_test456'
|
||||
const error = 'Something went wrong'
|
||||
|
||||
await localQueue.markJobFailed(jobId, error)
|
||||
|
||||
expect(localRedis.hset).toHaveBeenCalledWith(`async-jobs:job:${jobId}`, {
|
||||
status: JOB_STATUS.FAILED,
|
||||
completedAt: expect.any(String),
|
||||
error,
|
||||
updatedAt: expect.any(String),
|
||||
})
|
||||
expect(localRedis.expire).toHaveBeenCalledWith(
|
||||
`async-jobs:job:${jobId}`,
|
||||
JOB_RETENTION_SECONDS
|
||||
)
|
||||
})
|
||||
|
||||
it.concurrent('should set TTL to 24 hours (86400 seconds)', async () => {
|
||||
const localRedis = createMockRedis()
|
||||
const localQueue = new RedisJobQueue(localRedis as never)
|
||||
|
||||
await localQueue.markJobFailed('run_test456', 'error')
|
||||
|
||||
expect(localRedis.expire).toHaveBeenCalledWith(expect.any(String), 86400)
|
||||
})
|
||||
})
|
||||
|
||||
describe('startJob', () => {
|
||||
it.concurrent('should not set TTL when starting a job', async () => {
|
||||
const localRedis = createMockRedis()
|
||||
const localQueue = new RedisJobQueue(localRedis as never)
|
||||
|
||||
await localQueue.startJob('run_test789')
|
||||
|
||||
expect(localRedis.hset).toHaveBeenCalled()
|
||||
expect(localRedis.expire).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
|
||||
describe('getJob', () => {
|
||||
it.concurrent('should return null for non-existent job', async () => {
|
||||
const localRedis = createMockRedis()
|
||||
const localQueue = new RedisJobQueue(localRedis as never)
|
||||
localRedis.hgetall.mockResolvedValue({})
|
||||
|
||||
const job = await localQueue.getJob('run_nonexistent')
|
||||
|
||||
expect(job).toBeNull()
|
||||
})
|
||||
|
||||
it.concurrent('should deserialize job data correctly', async () => {
|
||||
const localRedis = createMockRedis()
|
||||
const localQueue = new RedisJobQueue(localRedis as never)
|
||||
const now = new Date()
|
||||
localRedis.hgetall.mockResolvedValue({
|
||||
id: 'run_test',
|
||||
type: 'workflow-execution',
|
||||
payload: JSON.stringify({ foo: 'bar' }),
|
||||
status: JOB_STATUS.COMPLETED,
|
||||
createdAt: now.toISOString(),
|
||||
startedAt: now.toISOString(),
|
||||
completedAt: now.toISOString(),
|
||||
attempts: '1',
|
||||
maxAttempts: '3',
|
||||
error: '',
|
||||
output: JSON.stringify({ result: 'ok' }),
|
||||
metadata: JSON.stringify({ workflowId: 'wf_123' }),
|
||||
})
|
||||
|
||||
const job = await localQueue.getJob('run_test')
|
||||
|
||||
expect(job).not.toBeNull()
|
||||
expect(job?.id).toBe('run_test')
|
||||
expect(job?.type).toBe('workflow-execution')
|
||||
expect(job?.payload).toEqual({ foo: 'bar' })
|
||||
expect(job?.status).toBe(JOB_STATUS.COMPLETED)
|
||||
expect(job?.output).toEqual({ result: 'ok' })
|
||||
expect(job?.metadata.workflowId).toBe('wf_123')
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe('JOB_RETENTION_SECONDS', () => {
|
||||
it.concurrent('should be 24 hours in seconds', async () => {
|
||||
expect(JOB_RETENTION_SECONDS).toBe(24 * 60 * 60)
|
||||
expect(JOB_RETENTION_SECONDS).toBe(86400)
|
||||
})
|
||||
})
|
||||
146
apps/sim/lib/core/async-jobs/backends/redis.ts
Normal file
146
apps/sim/lib/core/async-jobs/backends/redis.ts
Normal file
@@ -0,0 +1,146 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import type Redis from 'ioredis'
|
||||
import {
|
||||
type EnqueueOptions,
|
||||
JOB_MAX_LIFETIME_SECONDS,
|
||||
JOB_RETENTION_SECONDS,
|
||||
JOB_STATUS,
|
||||
type Job,
|
||||
type JobMetadata,
|
||||
type JobQueueBackend,
|
||||
type JobStatus,
|
||||
type JobType,
|
||||
} from '@/lib/core/async-jobs/types'
|
||||
|
||||
const logger = createLogger('RedisJobQueue')
|
||||
|
||||
const KEYS = {
|
||||
job: (id: string) => `async-jobs:job:${id}`,
|
||||
} as const
|
||||
|
||||
function serializeJob(job: Job): Record<string, string> {
|
||||
return {
|
||||
id: job.id,
|
||||
type: job.type,
|
||||
payload: JSON.stringify(job.payload),
|
||||
status: job.status,
|
||||
createdAt: job.createdAt.toISOString(),
|
||||
startedAt: job.startedAt?.toISOString() ?? '',
|
||||
completedAt: job.completedAt?.toISOString() ?? '',
|
||||
attempts: job.attempts.toString(),
|
||||
maxAttempts: job.maxAttempts.toString(),
|
||||
error: job.error ?? '',
|
||||
output: job.output !== undefined ? JSON.stringify(job.output) : '',
|
||||
metadata: JSON.stringify(job.metadata),
|
||||
updatedAt: new Date().toISOString(),
|
||||
}
|
||||
}
|
||||
|
||||
function deserializeJob(data: Record<string, string>): Job | null {
|
||||
if (!data || !data.id) return null
|
||||
|
||||
try {
|
||||
return {
|
||||
id: data.id,
|
||||
type: data.type as JobType,
|
||||
payload: JSON.parse(data.payload),
|
||||
status: data.status as JobStatus,
|
||||
createdAt: new Date(data.createdAt),
|
||||
startedAt: data.startedAt ? new Date(data.startedAt) : undefined,
|
||||
completedAt: data.completedAt ? new Date(data.completedAt) : undefined,
|
||||
attempts: Number.parseInt(data.attempts, 10),
|
||||
maxAttempts: Number.parseInt(data.maxAttempts, 10),
|
||||
error: data.error || undefined,
|
||||
output: data.output ? JSON.parse(data.output) : undefined,
|
||||
metadata: JSON.parse(data.metadata) as JobMetadata,
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error('Failed to deserialize job', { error, data })
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
export class RedisJobQueue implements JobQueueBackend {
|
||||
private redis: Redis
|
||||
|
||||
constructor(redis: Redis) {
|
||||
this.redis = redis
|
||||
}
|
||||
|
||||
async enqueue<TPayload>(
|
||||
type: JobType,
|
||||
payload: TPayload,
|
||||
options?: EnqueueOptions
|
||||
): Promise<string> {
|
||||
const jobId = `run_${crypto.randomUUID().replace(/-/g, '').slice(0, 20)}`
|
||||
const now = new Date()
|
||||
|
||||
const job: Job<TPayload> = {
|
||||
id: jobId,
|
||||
type,
|
||||
payload,
|
||||
status: JOB_STATUS.PENDING,
|
||||
createdAt: now,
|
||||
attempts: 0,
|
||||
maxAttempts: options?.maxAttempts ?? 3,
|
||||
metadata: options?.metadata ?? {},
|
||||
}
|
||||
|
||||
const key = KEYS.job(jobId)
|
||||
const serialized = serializeJob(job as Job)
|
||||
await this.redis.hset(key, serialized)
|
||||
await this.redis.expire(key, JOB_MAX_LIFETIME_SECONDS)
|
||||
|
||||
logger.debug('Enqueued job', { jobId, type })
|
||||
return jobId
|
||||
}
|
||||
|
||||
async getJob(jobId: string): Promise<Job | null> {
|
||||
const data = await this.redis.hgetall(KEYS.job(jobId))
|
||||
return deserializeJob(data)
|
||||
}
|
||||
|
||||
async startJob(jobId: string): Promise<void> {
|
||||
const now = new Date()
|
||||
const key = KEYS.job(jobId)
|
||||
|
||||
await this.redis.hset(key, {
|
||||
status: JOB_STATUS.PROCESSING,
|
||||
startedAt: now.toISOString(),
|
||||
updatedAt: now.toISOString(),
|
||||
})
|
||||
await this.redis.hincrby(key, 'attempts', 1)
|
||||
|
||||
logger.debug('Started job', { jobId })
|
||||
}
|
||||
|
||||
async completeJob(jobId: string, output: unknown): Promise<void> {
|
||||
const now = new Date()
|
||||
const key = KEYS.job(jobId)
|
||||
|
||||
await this.redis.hset(key, {
|
||||
status: JOB_STATUS.COMPLETED,
|
||||
completedAt: now.toISOString(),
|
||||
output: JSON.stringify(output),
|
||||
updatedAt: now.toISOString(),
|
||||
})
|
||||
await this.redis.expire(key, JOB_RETENTION_SECONDS)
|
||||
|
||||
logger.debug('Completed job', { jobId })
|
||||
}
|
||||
|
||||
async markJobFailed(jobId: string, error: string): Promise<void> {
|
||||
const now = new Date()
|
||||
const key = KEYS.job(jobId)
|
||||
|
||||
await this.redis.hset(key, {
|
||||
status: JOB_STATUS.FAILED,
|
||||
completedAt: now.toISOString(),
|
||||
error,
|
||||
updatedAt: now.toISOString(),
|
||||
})
|
||||
await this.redis.expire(key, JOB_RETENTION_SECONDS)
|
||||
|
||||
logger.debug('Marked job as failed', { jobId })
|
||||
}
|
||||
}
|
||||
119
apps/sim/lib/core/async-jobs/backends/trigger-dev.ts
Normal file
119
apps/sim/lib/core/async-jobs/backends/trigger-dev.ts
Normal file
@@ -0,0 +1,119 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { runs, tasks } from '@trigger.dev/sdk'
|
||||
import {
|
||||
type EnqueueOptions,
|
||||
JOB_STATUS,
|
||||
type Job,
|
||||
type JobMetadata,
|
||||
type JobQueueBackend,
|
||||
type JobStatus,
|
||||
type JobType,
|
||||
} from '@/lib/core/async-jobs/types'
|
||||
|
||||
const logger = createLogger('TriggerDevJobQueue')
|
||||
|
||||
/**
|
||||
* Maps trigger.dev task IDs to our JobType
|
||||
*/
|
||||
const JOB_TYPE_TO_TASK_ID: Record<JobType, string> = {
|
||||
'workflow-execution': 'workflow-execution',
|
||||
'schedule-execution': 'schedule-execution',
|
||||
'webhook-execution': 'webhook-execution',
|
||||
}
|
||||
|
||||
/**
|
||||
* Maps trigger.dev run status to our JobStatus
|
||||
*/
|
||||
function mapTriggerDevStatus(status: string): JobStatus {
|
||||
switch (status) {
|
||||
case 'QUEUED':
|
||||
case 'WAITING_FOR_DEPLOY':
|
||||
return JOB_STATUS.PENDING
|
||||
case 'EXECUTING':
|
||||
case 'RESCHEDULED':
|
||||
case 'FROZEN':
|
||||
return JOB_STATUS.PROCESSING
|
||||
case 'COMPLETED':
|
||||
return JOB_STATUS.COMPLETED
|
||||
case 'CANCELED':
|
||||
case 'FAILED':
|
||||
case 'CRASHED':
|
||||
case 'INTERRUPTED':
|
||||
case 'SYSTEM_FAILURE':
|
||||
case 'EXPIRED':
|
||||
return JOB_STATUS.FAILED
|
||||
default:
|
||||
return JOB_STATUS.PENDING
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Adapter that wraps the trigger.dev SDK to conform to JobQueueBackend interface.
|
||||
*/
|
||||
export class TriggerDevJobQueue implements JobQueueBackend {
|
||||
async enqueue<TPayload>(
|
||||
type: JobType,
|
||||
payload: TPayload,
|
||||
options?: EnqueueOptions
|
||||
): Promise<string> {
|
||||
const taskId = JOB_TYPE_TO_TASK_ID[type]
|
||||
if (!taskId) {
|
||||
throw new Error(`Unknown job type: ${type}`)
|
||||
}
|
||||
|
||||
const enrichedPayload =
|
||||
options?.metadata && typeof payload === 'object' && payload !== null
|
||||
? { ...payload, ...options.metadata }
|
||||
: payload
|
||||
|
||||
const handle = await tasks.trigger(taskId, enrichedPayload)
|
||||
|
||||
logger.debug('Enqueued job via trigger.dev', { jobId: handle.id, type, taskId })
|
||||
return handle.id
|
||||
}
|
||||
|
||||
async getJob(jobId: string): Promise<Job | null> {
|
||||
try {
|
||||
const run = await runs.retrieve(jobId)
|
||||
|
||||
const payload = run.payload as Record<string, unknown>
|
||||
const metadata: JobMetadata = {
|
||||
workflowId: payload?.workflowId as string | undefined,
|
||||
userId: payload?.userId as string | undefined,
|
||||
}
|
||||
|
||||
return {
|
||||
id: jobId,
|
||||
type: run.taskIdentifier as JobType,
|
||||
payload: run.payload,
|
||||
status: mapTriggerDevStatus(run.status),
|
||||
createdAt: run.createdAt ? new Date(run.createdAt) : new Date(),
|
||||
startedAt: run.startedAt ? new Date(run.startedAt) : undefined,
|
||||
completedAt: run.finishedAt ? new Date(run.finishedAt) : undefined,
|
||||
attempts: run.attemptCount ?? 1,
|
||||
maxAttempts: 3,
|
||||
error: run.error?.message,
|
||||
output: run.output as unknown,
|
||||
metadata,
|
||||
}
|
||||
} catch (error) {
|
||||
const isNotFound =
|
||||
(error instanceof Error && error.message.toLowerCase().includes('not found')) ||
|
||||
(error && typeof error === 'object' && 'status' in error && error.status === 404)
|
||||
|
||||
if (isNotFound) {
|
||||
logger.debug('Job not found in trigger.dev', { jobId })
|
||||
return null
|
||||
}
|
||||
|
||||
logger.error('Failed to get job from trigger.dev', { jobId, error })
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
async startJob(_jobId: string): Promise<void> {}
|
||||
|
||||
async completeJob(_jobId: string, _output: unknown): Promise<void> {}
|
||||
|
||||
async markJobFailed(_jobId: string, _error: string): Promise<void> {}
|
||||
}
|
||||
88
apps/sim/lib/core/async-jobs/config.ts
Normal file
88
apps/sim/lib/core/async-jobs/config.ts
Normal file
@@ -0,0 +1,88 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import type { AsyncBackendType, JobQueueBackend } from '@/lib/core/async-jobs/types'
|
||||
import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
|
||||
import { getRedisClient } from '@/lib/core/config/redis'
|
||||
|
||||
const logger = createLogger('AsyncJobsConfig')
|
||||
|
||||
let cachedBackend: JobQueueBackend | null = null
|
||||
let cachedBackendType: AsyncBackendType | null = null
|
||||
|
||||
/**
|
||||
* Determines which async backend to use based on environment configuration.
|
||||
* Follows the fallback chain: trigger.dev → redis → database
|
||||
*/
|
||||
export function getAsyncBackendType(): AsyncBackendType {
|
||||
if (isTriggerDevEnabled) {
|
||||
return 'trigger-dev'
|
||||
}
|
||||
|
||||
const redis = getRedisClient()
|
||||
if (redis) {
|
||||
return 'redis'
|
||||
}
|
||||
|
||||
return 'database'
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the job queue backend singleton.
|
||||
* Creates the appropriate backend based on environment configuration.
|
||||
*/
|
||||
export async function getJobQueue(): Promise<JobQueueBackend> {
|
||||
if (cachedBackend) {
|
||||
return cachedBackend
|
||||
}
|
||||
|
||||
const type = getAsyncBackendType()
|
||||
|
||||
switch (type) {
|
||||
case 'trigger-dev': {
|
||||
const { TriggerDevJobQueue } = await import('@/lib/core/async-jobs/backends/trigger-dev')
|
||||
cachedBackend = new TriggerDevJobQueue()
|
||||
break
|
||||
}
|
||||
case 'redis': {
|
||||
const redis = getRedisClient()
|
||||
if (!redis) {
|
||||
throw new Error('Redis client not available but redis backend was selected')
|
||||
}
|
||||
const { RedisJobQueue } = await import('@/lib/core/async-jobs/backends/redis')
|
||||
cachedBackend = new RedisJobQueue(redis)
|
||||
break
|
||||
}
|
||||
case 'database': {
|
||||
const { DatabaseJobQueue } = await import('@/lib/core/async-jobs/backends/database')
|
||||
cachedBackend = new DatabaseJobQueue()
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
cachedBackendType = type
|
||||
logger.info(`Async job backend initialized: ${type}`)
|
||||
|
||||
return cachedBackend
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the current backend type (for logging/debugging)
|
||||
*/
|
||||
export function getCurrentBackendType(): AsyncBackendType | null {
|
||||
return cachedBackendType
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if jobs should be executed inline (fire-and-forget).
|
||||
* For Redis/DB backends, we execute inline. Trigger.dev handles execution itself.
|
||||
*/
|
||||
export function shouldExecuteInline(): boolean {
|
||||
return getAsyncBackendType() !== 'trigger-dev'
|
||||
}
|
||||
|
||||
/**
|
||||
* Resets the cached backend (useful for testing)
|
||||
*/
|
||||
export function resetJobQueueCache(): void {
|
||||
cachedBackend = null
|
||||
cachedBackendType = null
|
||||
}
|
||||
22
apps/sim/lib/core/async-jobs/index.ts
Normal file
22
apps/sim/lib/core/async-jobs/index.ts
Normal file
@@ -0,0 +1,22 @@
|
||||
export {
|
||||
getAsyncBackendType,
|
||||
getCurrentBackendType,
|
||||
getJobQueue,
|
||||
resetJobQueueCache,
|
||||
shouldExecuteInline,
|
||||
} from './config'
|
||||
export type {
|
||||
AsyncBackendType,
|
||||
EnqueueOptions,
|
||||
Job,
|
||||
JobMetadata,
|
||||
JobQueueBackend,
|
||||
JobStatus,
|
||||
JobType,
|
||||
} from './types'
|
||||
export {
|
||||
JOB_MAX_LIFETIME_SECONDS,
|
||||
JOB_RETENTION_HOURS,
|
||||
JOB_RETENTION_SECONDS,
|
||||
JOB_STATUS,
|
||||
} from './types'
|
||||
32
apps/sim/lib/core/async-jobs/types.test.ts
Normal file
32
apps/sim/lib/core/async-jobs/types.test.ts
Normal file
@@ -0,0 +1,32 @@
|
||||
/**
|
||||
* @vitest-environment node
|
||||
*/
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { JOB_MAX_LIFETIME_SECONDS, JOB_RETENTION_HOURS, JOB_RETENTION_SECONDS } from './types'
|
||||
|
||||
describe('Job retention constants', () => {
|
||||
it.concurrent('JOB_RETENTION_HOURS should be 24', async () => {
|
||||
expect(JOB_RETENTION_HOURS).toBe(24)
|
||||
})
|
||||
|
||||
it.concurrent('JOB_RETENTION_SECONDS should be derived from JOB_RETENTION_HOURS', async () => {
|
||||
expect(JOB_RETENTION_SECONDS).toBe(JOB_RETENTION_HOURS * 60 * 60)
|
||||
})
|
||||
|
||||
it.concurrent('JOB_RETENTION_SECONDS should equal 86400 (24 hours)', async () => {
|
||||
expect(JOB_RETENTION_SECONDS).toBe(86400)
|
||||
})
|
||||
|
||||
it.concurrent('constants should be consistent with each other', async () => {
|
||||
const hoursToSeconds = JOB_RETENTION_HOURS * 60 * 60
|
||||
expect(JOB_RETENTION_SECONDS).toBe(hoursToSeconds)
|
||||
})
|
||||
|
||||
it.concurrent(
|
||||
'JOB_MAX_LIFETIME_SECONDS should be greater than JOB_RETENTION_SECONDS',
|
||||
async () => {
|
||||
expect(JOB_MAX_LIFETIME_SECONDS).toBeGreaterThan(JOB_RETENTION_SECONDS)
|
||||
expect(JOB_MAX_LIFETIME_SECONDS).toBe(48 * 60 * 60)
|
||||
}
|
||||
)
|
||||
})
|
||||
82
apps/sim/lib/core/async-jobs/types.ts
Normal file
82
apps/sim/lib/core/async-jobs/types.ts
Normal file
@@ -0,0 +1,82 @@
|
||||
/**
|
||||
* Types and constants for the async job queue system
|
||||
*/
|
||||
|
||||
/** Retention period for completed/failed jobs (in hours) */
|
||||
export const JOB_RETENTION_HOURS = 24
|
||||
|
||||
/** Retention period for completed/failed jobs (in seconds, for Redis TTL) */
|
||||
export const JOB_RETENTION_SECONDS = JOB_RETENTION_HOURS * 60 * 60
|
||||
|
||||
/** Max lifetime for jobs in Redis (in seconds) - cleanup for stuck pending/processing jobs */
|
||||
export const JOB_MAX_LIFETIME_SECONDS = 48 * 60 * 60
|
||||
|
||||
export const JOB_STATUS = {
|
||||
PENDING: 'pending',
|
||||
PROCESSING: 'processing',
|
||||
COMPLETED: 'completed',
|
||||
FAILED: 'failed',
|
||||
} as const
|
||||
|
||||
export type JobStatus = (typeof JOB_STATUS)[keyof typeof JOB_STATUS]
|
||||
|
||||
export type JobType = 'workflow-execution' | 'schedule-execution' | 'webhook-execution'
|
||||
|
||||
export interface Job<TPayload = unknown, TOutput = unknown> {
|
||||
id: string
|
||||
type: JobType
|
||||
payload: TPayload
|
||||
status: JobStatus
|
||||
createdAt: Date
|
||||
startedAt?: Date
|
||||
completedAt?: Date
|
||||
attempts: number
|
||||
maxAttempts: number
|
||||
error?: string
|
||||
output?: TOutput
|
||||
metadata: JobMetadata
|
||||
}
|
||||
|
||||
export interface JobMetadata {
|
||||
workflowId?: string
|
||||
userId?: string
|
||||
[key: string]: unknown
|
||||
}
|
||||
|
||||
export interface EnqueueOptions {
|
||||
maxAttempts?: number
|
||||
metadata?: JobMetadata
|
||||
}
|
||||
|
||||
/**
|
||||
* Backend interface for job queue implementations.
|
||||
* All backends must implement this interface.
|
||||
*/
|
||||
export interface JobQueueBackend {
|
||||
/**
|
||||
* Add a job to the queue
|
||||
*/
|
||||
enqueue<TPayload>(type: JobType, payload: TPayload, options?: EnqueueOptions): Promise<string>
|
||||
|
||||
/**
|
||||
* Get a job by ID
|
||||
*/
|
||||
getJob(jobId: string): Promise<Job | null>
|
||||
|
||||
/**
|
||||
* Mark a job as started/processing
|
||||
*/
|
||||
startJob(jobId: string): Promise<void>
|
||||
|
||||
/**
|
||||
* Mark a job as completed with output
|
||||
*/
|
||||
completeJob(jobId: string, output: unknown): Promise<void>
|
||||
|
||||
/**
|
||||
* Mark a job as failed with error message
|
||||
*/
|
||||
markJobFailed(jobId: string, error: string): Promise<void>
|
||||
}
|
||||
|
||||
export type AsyncBackendType = 'trigger-dev' | 'redis' | 'database'
|
||||
@@ -170,10 +170,15 @@ export const env = createEnv({
|
||||
RATE_LIMIT_ENTERPRISE_SYNC: z.string().optional().default('600'), // Enterprise tier sync API executions per minute
|
||||
RATE_LIMIT_ENTERPRISE_ASYNC: z.string().optional().default('5000'), // Enterprise tier async API executions per minute
|
||||
|
||||
EXECUTION_TIMEOUT_FREE: z.string().optional().default('300'),
|
||||
EXECUTION_TIMEOUT_PRO: z.string().optional().default('3000'),
|
||||
EXECUTION_TIMEOUT_TEAM: z.string().optional().default('3000'),
|
||||
EXECUTION_TIMEOUT_ENTERPRISE: z.string().optional().default('3000'),
|
||||
// Timeout Configuration
|
||||
EXECUTION_TIMEOUT_FREE: z.string().optional().default('300'), // 5 minutes
|
||||
EXECUTION_TIMEOUT_PRO: z.string().optional().default('3000'), // 50 minutes
|
||||
EXECUTION_TIMEOUT_TEAM: z.string().optional().default('3000'), // 50 minutes
|
||||
EXECUTION_TIMEOUT_ENTERPRISE: z.string().optional().default('3000'), // 50 minutes
|
||||
EXECUTION_TIMEOUT_ASYNC_FREE: z.string().optional().default('5400'), // 90 minutes
|
||||
EXECUTION_TIMEOUT_ASYNC_PRO: z.string().optional().default('5400'), // 90 minutes
|
||||
EXECUTION_TIMEOUT_ASYNC_TEAM: z.string().optional().default('5400'), // 90 minutes
|
||||
EXECUTION_TIMEOUT_ASYNC_ENTERPRISE: z.string().optional().default('5400'), // 90 minutes
|
||||
|
||||
// Knowledge Base Processing Configuration - Shared across all processing methods
|
||||
KB_CONFIG_MAX_DURATION: z.number().optional().default(600), // Max processing duration in seconds (10 minutes)
|
||||
@@ -345,7 +350,6 @@ export const env = createEnv({
|
||||
NEXT_PUBLIC_BRAND_BACKGROUND_COLOR: z.string().regex(/^#[0-9A-Fa-f]{6}$/).optional(), // Brand background color (hex format)
|
||||
|
||||
// Feature Flags
|
||||
NEXT_PUBLIC_TRIGGER_DEV_ENABLED: z.boolean().optional(), // Client-side gate for async executions UI
|
||||
NEXT_PUBLIC_SSO_ENABLED: z.boolean().optional(), // Enable SSO login UI components
|
||||
NEXT_PUBLIC_CREDENTIAL_SETS_ENABLED: z.boolean().optional(), // Enable credential sets (email polling) on self-hosted
|
||||
NEXT_PUBLIC_ACCESS_CONTROL_ENABLED: z.boolean().optional(), // Enable access control (permission groups) on self-hosted
|
||||
@@ -377,7 +381,6 @@ export const env = createEnv({
|
||||
NEXT_PUBLIC_BRAND_ACCENT_COLOR: process.env.NEXT_PUBLIC_BRAND_ACCENT_COLOR,
|
||||
NEXT_PUBLIC_BRAND_ACCENT_HOVER_COLOR: process.env.NEXT_PUBLIC_BRAND_ACCENT_HOVER_COLOR,
|
||||
NEXT_PUBLIC_BRAND_BACKGROUND_COLOR: process.env.NEXT_PUBLIC_BRAND_BACKGROUND_COLOR,
|
||||
NEXT_PUBLIC_TRIGGER_DEV_ENABLED: process.env.NEXT_PUBLIC_TRIGGER_DEV_ENABLED,
|
||||
NEXT_PUBLIC_SSO_ENABLED: process.env.NEXT_PUBLIC_SSO_ENABLED,
|
||||
NEXT_PUBLIC_CREDENTIAL_SETS_ENABLED: process.env.NEXT_PUBLIC_CREDENTIAL_SETS_ENABLED,
|
||||
NEXT_PUBLIC_ACCESS_CONTROL_ENABLED: process.env.NEXT_PUBLIC_ACCESS_CONTROL_ENABLED,
|
||||
|
||||
@@ -14,8 +14,12 @@ const DEFAULT_SYNC_TIMEOUTS_SECONDS = {
|
||||
enterprise: 3000,
|
||||
} as const
|
||||
|
||||
const ASYNC_MULTIPLIER = 2
|
||||
const MAX_ASYNC_TIMEOUT_SECONDS = 5400
|
||||
const DEFAULT_ASYNC_TIMEOUTS_SECONDS = {
|
||||
free: 5400,
|
||||
pro: 5400,
|
||||
team: 5400,
|
||||
enterprise: 5400,
|
||||
} as const
|
||||
|
||||
function getSyncTimeoutForPlan(plan: SubscriptionPlan): number {
|
||||
const envVarMap: Record<SubscriptionPlan, string | undefined> = {
|
||||
@@ -28,10 +32,13 @@ function getSyncTimeoutForPlan(plan: SubscriptionPlan): number {
|
||||
}
|
||||
|
||||
function getAsyncTimeoutForPlan(plan: SubscriptionPlan): number {
|
||||
const syncMs = getSyncTimeoutForPlan(plan)
|
||||
const asyncMs = syncMs * ASYNC_MULTIPLIER
|
||||
const maxAsyncMs = MAX_ASYNC_TIMEOUT_SECONDS * 1000
|
||||
return Math.min(asyncMs, maxAsyncMs)
|
||||
const envVarMap: Record<SubscriptionPlan, string | undefined> = {
|
||||
free: env.EXECUTION_TIMEOUT_ASYNC_FREE,
|
||||
pro: env.EXECUTION_TIMEOUT_ASYNC_PRO,
|
||||
team: env.EXECUTION_TIMEOUT_ASYNC_TEAM,
|
||||
enterprise: env.EXECUTION_TIMEOUT_ASYNC_ENTERPRISE,
|
||||
}
|
||||
return (Number.parseInt(envVarMap[plan] || '') || DEFAULT_ASYNC_TIMEOUTS_SECONDS[plan]) * 1000
|
||||
}
|
||||
|
||||
const EXECUTION_TIMEOUTS: Record<SubscriptionPlan, ExecutionTimeoutConfig> = {
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
import { db, webhook, workflow, workflowDeploymentVersion } from '@sim/db'
|
||||
import { credentialSet, subscription } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { tasks } from '@trigger.dev/sdk'
|
||||
import { and, eq, isNull, or } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { v4 as uuidv4 } from 'uuid'
|
||||
import { checkEnterprisePlan, checkTeamPlan } from '@/lib/billing/subscriptions/utils'
|
||||
import { isProd, isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
|
||||
import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
|
||||
import { isProd } from '@/lib/core/config/feature-flags'
|
||||
import { getEffectiveDecryptedEnv } from '@/lib/environment/utils'
|
||||
import { preprocessExecution } from '@/lib/execution/preprocessing'
|
||||
import { convertSquareBracketsToTwiML } from '@/lib/webhooks/utils'
|
||||
@@ -1015,18 +1015,39 @@ export async function queueWebhookExecution(
|
||||
...(credentialId ? { credentialId } : {}),
|
||||
}
|
||||
|
||||
if (isTriggerDevEnabled) {
|
||||
const handle = await tasks.trigger('webhook-execution', payload)
|
||||
logger.info(
|
||||
`[${options.requestId}] Queued webhook execution task ${handle.id} for ${foundWebhook.provider} webhook`
|
||||
)
|
||||
} else {
|
||||
void executeWebhookJob(payload).catch((error) => {
|
||||
logger.error(`[${options.requestId}] Direct webhook execution failed`, error)
|
||||
})
|
||||
logger.info(
|
||||
`[${options.requestId}] Queued direct webhook execution for ${foundWebhook.provider} webhook (Trigger.dev disabled)`
|
||||
)
|
||||
const jobQueue = await getJobQueue()
|
||||
const jobId = await jobQueue.enqueue('webhook-execution', payload, {
|
||||
metadata: { workflowId: foundWorkflow.id, userId: foundWorkflow.userId },
|
||||
})
|
||||
logger.info(
|
||||
`[${options.requestId}] Queued webhook execution task ${jobId} for ${foundWebhook.provider} webhook`
|
||||
)
|
||||
|
||||
if (shouldExecuteInline()) {
|
||||
void (async () => {
|
||||
try {
|
||||
await jobQueue.startJob(jobId)
|
||||
const output = await executeWebhookJob(payload)
|
||||
await jobQueue.completeJob(jobId, output)
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error)
|
||||
logger.error(`[${options.requestId}] Webhook execution failed`, {
|
||||
jobId,
|
||||
error: errorMessage,
|
||||
})
|
||||
try {
|
||||
await jobQueue.markJobFailed(jobId, errorMessage)
|
||||
} catch (markFailedError) {
|
||||
logger.error(`[${options.requestId}] Failed to mark job as failed`, {
|
||||
jobId,
|
||||
error:
|
||||
markFailedError instanceof Error
|
||||
? markFailedError.message
|
||||
: String(markFailedError),
|
||||
})
|
||||
}
|
||||
}
|
||||
})()
|
||||
}
|
||||
|
||||
if (foundWebhook.provider === 'microsoft-teams') {
|
||||
|
||||
@@ -127,6 +127,18 @@ app:
|
||||
RATE_LIMIT_WINDOW_MS: "60000" # Rate limit window duration (1 minute)
|
||||
RATE_LIMIT_FREE_SYNC: "50" # Sync API executions per minute
|
||||
RATE_LIMIT_FREE_ASYNC: "200" # Async API executions per minute
|
||||
|
||||
# Execution Timeout Configuration (in seconds)
|
||||
# Sync timeouts apply to synchronous API calls
|
||||
EXECUTION_TIMEOUT_FREE: "300" # Free tier sync timeout (5 minutes)
|
||||
EXECUTION_TIMEOUT_PRO: "3000" # Pro tier sync timeout (50 minutes)
|
||||
EXECUTION_TIMEOUT_TEAM: "3000" # Team tier sync timeout (50 minutes)
|
||||
EXECUTION_TIMEOUT_ENTERPRISE: "3000" # Enterprise tier sync timeout (50 minutes)
|
||||
# Async timeouts apply to async/background job executions
|
||||
EXECUTION_TIMEOUT_ASYNC_FREE: "5400" # Free tier async timeout (90 minutes)
|
||||
EXECUTION_TIMEOUT_ASYNC_PRO: "5400" # Pro tier async timeout (90 minutes)
|
||||
EXECUTION_TIMEOUT_ASYNC_TEAM: "5400" # Team tier async timeout (90 minutes)
|
||||
EXECUTION_TIMEOUT_ASYNC_ENTERPRISE: "5400" # Enterprise tier async timeout (90 minutes)
|
||||
|
||||
# UI Branding & Whitelabeling Configuration
|
||||
NEXT_PUBLIC_BRAND_NAME: "Sim" # Custom brand name
|
||||
|
||||
19
packages/db/migrations/0151_stale_screwball.sql
Normal file
19
packages/db/migrations/0151_stale_screwball.sql
Normal file
@@ -0,0 +1,19 @@
|
||||
CREATE TABLE "async_jobs" (
|
||||
"id" text PRIMARY KEY NOT NULL,
|
||||
"type" text NOT NULL,
|
||||
"payload" jsonb NOT NULL,
|
||||
"status" text DEFAULT 'pending' NOT NULL,
|
||||
"created_at" timestamp DEFAULT now() NOT NULL,
|
||||
"started_at" timestamp,
|
||||
"completed_at" timestamp,
|
||||
"run_at" timestamp,
|
||||
"attempts" integer DEFAULT 0 NOT NULL,
|
||||
"max_attempts" integer DEFAULT 3 NOT NULL,
|
||||
"error" text,
|
||||
"output" jsonb,
|
||||
"metadata" jsonb DEFAULT '{}' NOT NULL,
|
||||
"updated_at" timestamp DEFAULT now() NOT NULL
|
||||
);
|
||||
--> statement-breakpoint
|
||||
CREATE INDEX "async_jobs_status_started_at_idx" ON "async_jobs" USING btree ("status","started_at");--> statement-breakpoint
|
||||
CREATE INDEX "async_jobs_status_completed_at_idx" ON "async_jobs" USING btree ("status","completed_at");
|
||||
10500
packages/db/migrations/meta/0151_snapshot.json
Normal file
10500
packages/db/migrations/meta/0151_snapshot.json
Normal file
File diff suppressed because it is too large
Load Diff
@@ -1051,6 +1051,13 @@
|
||||
"when": 1769897862156,
|
||||
"tag": "0150_flimsy_hemingway",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 151,
|
||||
"version": "7",
|
||||
"when": 1770239332381,
|
||||
"tag": "0151_stale_screwball",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@@ -2124,3 +2124,34 @@ export const permissionGroupMember = pgTable(
|
||||
userIdUnique: uniqueIndex('permission_group_member_user_id_unique').on(table.userId),
|
||||
})
|
||||
)
|
||||
|
||||
/**
|
||||
* Async Jobs - Queue for background job processing (Redis/DB backends)
|
||||
* Used when trigger.dev is not available for async workflow executions
|
||||
*/
|
||||
export const asyncJobs = pgTable(
|
||||
'async_jobs',
|
||||
{
|
||||
id: text('id').primaryKey(),
|
||||
type: text('type').notNull(),
|
||||
payload: jsonb('payload').notNull(),
|
||||
status: text('status').notNull().default('pending'),
|
||||
createdAt: timestamp('created_at').notNull().defaultNow(),
|
||||
startedAt: timestamp('started_at'),
|
||||
completedAt: timestamp('completed_at'),
|
||||
runAt: timestamp('run_at'),
|
||||
attempts: integer('attempts').notNull().default(0),
|
||||
maxAttempts: integer('max_attempts').notNull().default(3),
|
||||
error: text('error'),
|
||||
output: jsonb('output'),
|
||||
metadata: jsonb('metadata').notNull().default('{}'),
|
||||
updatedAt: timestamp('updated_at').notNull().defaultNow(),
|
||||
},
|
||||
(table) => ({
|
||||
statusStartedAtIdx: index('async_jobs_status_started_at_idx').on(table.status, table.startedAt),
|
||||
statusCompletedAtIdx: index('async_jobs_status_completed_at_idx').on(
|
||||
table.status,
|
||||
table.completedAt
|
||||
),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -45,12 +45,14 @@ export * from './assertions'
|
||||
export * from './builders'
|
||||
export * from './factories'
|
||||
export {
|
||||
clearRedisMocks,
|
||||
createEnvMock,
|
||||
createMockDb,
|
||||
createMockFetch,
|
||||
createMockFormDataRequest,
|
||||
createMockGetEnv,
|
||||
createMockLogger,
|
||||
createMockRedis,
|
||||
createMockRequest,
|
||||
createMockResponse,
|
||||
createMockSocket,
|
||||
@@ -63,6 +65,7 @@ export {
|
||||
loggerMock,
|
||||
type MockAuthResult,
|
||||
type MockFetchResponse,
|
||||
type MockRedis,
|
||||
type MockUser,
|
||||
mockAuth,
|
||||
mockCommonSchemas,
|
||||
|
||||
@@ -63,6 +63,8 @@ export {
|
||||
} from './fetch.mock'
|
||||
// Logger mocks
|
||||
export { clearLoggerMocks, createMockLogger, getLoggerCalls, loggerMock } from './logger.mock'
|
||||
// Redis mocks
|
||||
export { clearRedisMocks, createMockRedis, type MockRedis } from './redis.mock'
|
||||
// Request mocks
|
||||
export { createMockFormDataRequest, createMockRequest } from './request.mock'
|
||||
// Socket mocks
|
||||
|
||||
80
packages/testing/src/mocks/redis.mock.ts
Normal file
80
packages/testing/src/mocks/redis.mock.ts
Normal file
@@ -0,0 +1,80 @@
|
||||
import { vi } from 'vitest'
|
||||
|
||||
/**
|
||||
* Creates a mock Redis client with common operations.
|
||||
*
|
||||
* @example
|
||||
* ```ts
|
||||
* const redis = createMockRedis()
|
||||
* const queue = new RedisJobQueue(redis as never)
|
||||
*
|
||||
* // After operations
|
||||
* expect(redis.hset).toHaveBeenCalled()
|
||||
* expect(redis.expire).toHaveBeenCalledWith('key', 86400)
|
||||
* ```
|
||||
*/
|
||||
export function createMockRedis() {
|
||||
return {
|
||||
// Hash operations
|
||||
hset: vi.fn().mockResolvedValue(1),
|
||||
hget: vi.fn().mockResolvedValue(null),
|
||||
hgetall: vi.fn().mockResolvedValue({}),
|
||||
hdel: vi.fn().mockResolvedValue(1),
|
||||
hmset: vi.fn().mockResolvedValue('OK'),
|
||||
hincrby: vi.fn().mockResolvedValue(1),
|
||||
|
||||
// Key operations
|
||||
get: vi.fn().mockResolvedValue(null),
|
||||
set: vi.fn().mockResolvedValue('OK'),
|
||||
del: vi.fn().mockResolvedValue(1),
|
||||
exists: vi.fn().mockResolvedValue(0),
|
||||
expire: vi.fn().mockResolvedValue(1),
|
||||
ttl: vi.fn().mockResolvedValue(-1),
|
||||
|
||||
// List operations
|
||||
lpush: vi.fn().mockResolvedValue(1),
|
||||
rpush: vi.fn().mockResolvedValue(1),
|
||||
lpop: vi.fn().mockResolvedValue(null),
|
||||
rpop: vi.fn().mockResolvedValue(null),
|
||||
lrange: vi.fn().mockResolvedValue([]),
|
||||
llen: vi.fn().mockResolvedValue(0),
|
||||
|
||||
// Set operations
|
||||
sadd: vi.fn().mockResolvedValue(1),
|
||||
srem: vi.fn().mockResolvedValue(1),
|
||||
smembers: vi.fn().mockResolvedValue([]),
|
||||
sismember: vi.fn().mockResolvedValue(0),
|
||||
|
||||
// Pub/Sub
|
||||
publish: vi.fn().mockResolvedValue(0),
|
||||
subscribe: vi.fn().mockResolvedValue(undefined),
|
||||
unsubscribe: vi.fn().mockResolvedValue(undefined),
|
||||
on: vi.fn(),
|
||||
|
||||
// Transaction
|
||||
multi: vi.fn(() => ({
|
||||
exec: vi.fn().mockResolvedValue([]),
|
||||
})),
|
||||
|
||||
// Connection
|
||||
ping: vi.fn().mockResolvedValue('PONG'),
|
||||
quit: vi.fn().mockResolvedValue('OK'),
|
||||
disconnect: vi.fn().mockResolvedValue(undefined),
|
||||
|
||||
// Status
|
||||
status: 'ready',
|
||||
}
|
||||
}
|
||||
|
||||
export type MockRedis = ReturnType<typeof createMockRedis>
|
||||
|
||||
/**
|
||||
* Clears all Redis mock calls.
|
||||
*/
|
||||
export function clearRedisMocks(redis: MockRedis) {
|
||||
Object.values(redis).forEach((value) => {
|
||||
if (typeof value === 'function' && 'mockClear' in value) {
|
||||
value.mockClear()
|
||||
}
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user