feat(api): added trace spans for asynchronous executions, modified api route naming convention (#154)

* improvement: added trace spans for asynchronous workflow executions, previously only appeared when workflow was executed manually

* improvement: updated api/workflow route naming convention

* improvement: updated api/schedules route naming convention

* improvement(logs/ui): add differentiator in UI for different trigger types

Author: waleedlatif1

sim/app/api/marketplace/[id]/info/route.ts

@@ -1,10 +1,10 @@
 import { NextRequest } from 'next/server'
 import { eq } from 'drizzle-orm'
 import { createLogger } from '@/lib/logs/console-logger'
+import { validateWorkflowAccess } from '@/app/api/workflows/middleware'
+import { createErrorResponse, createSuccessResponse } from '@/app/api/workflows/utils'
 import { db } from '@/db'
 import * as schema from '@/db/schema'
-import { validateWorkflowAccess } from '@/app/api/workflow/middleware'
-import { createErrorResponse, createSuccessResponse } from '@/app/api/workflow/utils'
 
 const logger = createLogger('MarketplaceInfoAPI')
 
@@ -27,7 +27,7 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
       .from(schema.marketplace)
       .where(eq(schema.marketplace.workflowId, id))
       .limit(1)
-      .then(rows => rows[0])
+      .then((rows) => rows[0])
 
     if (!marketplaceEntry) {
       logger.warn(`[${requestId}] No marketplace entry found for workflow: ${id}`)
@@ -48,7 +48,10 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
       updatedAt: marketplaceEntry.updatedAt,
     })
   } catch (error) {
-    logger.error(`[${requestId}] Error getting marketplace info for workflow: ${(await params).id}`, error)
+    logger.error(
+      `[${requestId}] Error getting marketplace info for workflow: ${(await params).id}`,
+      error
+    )
     return createErrorResponse('Failed to get marketplace information', 500)
   }
-} 
+}

sim/app/api/marketplace/[id]/star/route.ts

@@ -2,9 +2,9 @@ import { NextRequest } from 'next/server'
 import { and, eq } from 'drizzle-orm'
 import { getSession } from '@/lib/auth'
 import { createLogger } from '@/lib/logs/console-logger'
+import { createErrorResponse, createSuccessResponse } from '@/app/api/workflows/utils'
 import { db } from '@/db'
 import * as schema from '@/db/schema'
-import { createErrorResponse, createSuccessResponse } from '@/app/api/workflow/utils'
 
 const logger = createLogger('MarketplaceStarAPI')
 
@@ -15,7 +15,7 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
     const { id } = await params
     const session = await getSession()
     const userId = session?.user?.id
-    
+
     if (!userId) {
       return createErrorResponse('Unauthorized', 401)
     }
@@ -26,7 +26,7 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
       .from(schema.marketplace)
       .where(eq(schema.marketplace.id, id))
       .limit(1)
-      .then(rows => rows[0])
+      .then((rows) => rows[0])
 
     if (!marketplaceEntry) {
       logger.warn(`[${requestId}] No marketplace entry found with ID: ${id}`)
@@ -37,28 +37,30 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
     const existingStar = await db
       .select()
       .from(schema.marketplaceStar)
-      .where(and(
-        eq(schema.marketplaceStar.marketplaceId, id),
-        eq(schema.marketplaceStar.userId, userId)
-      ))
+      .where(
+        and(eq(schema.marketplaceStar.marketplaceId, id), eq(schema.marketplaceStar.userId, userId))
+      )
       .limit(1)
-      .then(rows => rows[0])
+      .then((rows) => rows[0])
 
     let action
     if (existingStar) {
       // User has already starred, so unstar it
-      await db.delete(schema.marketplaceStar).where(
-        and(
-          eq(schema.marketplaceStar.marketplaceId, id),
-          eq(schema.marketplaceStar.userId, userId)
+      await db
+        .delete(schema.marketplaceStar)
+        .where(
+          and(
+            eq(schema.marketplaceStar.marketplaceId, id),
+            eq(schema.marketplaceStar.userId, userId)
+          )
         )
-      )
-      
+
       // Decrement the star count
-      await db.update(schema.marketplace)
+      await db
+        .update(schema.marketplace)
         .set({ stars: marketplaceEntry.stars - 1 })
         .where(eq(schema.marketplace.id, id))
-      
+
       action = 'unstarred'
     } else {
       // User hasn't starred yet, add a star
@@ -68,12 +70,13 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
         userId: userId,
         createdAt: new Date(),
       })
-      
+
       // Increment the star count
-      await db.update(schema.marketplace)
+      await db
+        .update(schema.marketplace)
         .set({ stars: marketplaceEntry.stars + 1 })
         .where(eq(schema.marketplace.id, id))
-      
+
       action = 'starred'
     }
 
@@ -107,12 +110,11 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
     const existingStar = await db
       .select()
       .from(schema.marketplaceStar)
-      .where(and(
-        eq(schema.marketplaceStar.marketplaceId, id),
-        eq(schema.marketplaceStar.userId, userId)
-      ))
+      .where(
+        and(eq(schema.marketplaceStar.marketplaceId, id), eq(schema.marketplaceStar.userId, userId))
+      )
       .limit(1)
-      .then(rows => rows[0])
+      .then((rows) => rows[0])
 
     return createSuccessResponse({
       isStarred: !!existingStar,
@@ -121,4 +123,4 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
     logger.error(`[${requestId}] Error checking star status: ${(await params).id}`, error)
     return createErrorResponse('Failed to check star status', 500)
   }
-} 
+}

sim/app/api/marketplace/[id]/unpublish/route.ts

@@ -1,10 +1,10 @@
 import { NextRequest } from 'next/server'
 import { eq } from 'drizzle-orm'
 import { createLogger } from '@/lib/logs/console-logger'
+import { validateWorkflowAccess } from '@/app/api/workflows/middleware'
+import { createErrorResponse, createSuccessResponse } from '@/app/api/workflows/utils'
 import { db } from '@/db'
 import * as schema from '@/db/schema'
-import { validateWorkflowAccess } from '@/app/api/workflow/middleware'
-import { createErrorResponse, createSuccessResponse } from '@/app/api/workflow/utils'
 
 const logger = createLogger('MarketplaceUnpublishAPI')
 
@@ -28,7 +28,7 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
       .from(schema.marketplace)
       .where(eq(schema.marketplace.workflowId, id))
       .limit(1)
-      .then(rows => rows[0])
+      .then((rows) => rows[0])
 
     if (!marketplaceEntry) {
       logger.warn(`[${requestId}] No marketplace entry found for workflow: ${id}`)
@@ -51,4 +51,4 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
     logger.error(`[${requestId}] Error unpublishing workflow: ${(await params).id}`, error)
     return createErrorResponse('Failed to unpublish workflow', 500)
   }
-} 
+}

sim/app/api/scheduled/[id]/route.ts sim/app/api/schedules/[id]/route.tssim/app/api/scheduled/[id]/route.ts renamed to sim/app/api/schedules/[id]/route.ts

@@ -5,6 +5,7 @@ import { v4 as uuidv4 } from 'uuid'
 import { z } from 'zod'
 import { createLogger } from '@/lib/logs/console-logger'
 import { persistExecutionError, persistExecutionLogs } from '@/lib/logs/execution-logger'
+import { buildTraceSpans } from '@/lib/logs/trace-spans'
 import { decryptSecret } from '@/lib/utils'
 import { mergeSubblockState } from '@/stores/workflows/utils'
 import { BlockState, WorkflowState } from '@/stores/workflows/workflow/types'
@@ -325,8 +326,18 @@ export async function GET(req: NextRequest) {
         )
         const result = await executor.execute(schedule.workflowId)
 
+        // Build trace spans from execution logs
+        const { traceSpans, totalDuration } = buildTraceSpans(result)
+
+        // Add trace spans to the execution result
+        const enrichedResult = {
+          ...result,
+          traceSpans,
+          totalDuration,
+        }
+
         // Log each execution step and the final result
-        await persistExecutionLogs(schedule.workflowId, executionId, result, 'schedule')
+        await persistExecutionLogs(schedule.workflowId, executionId, enrichedResult, 'schedule')
 
         // Only update next_run_at if execution was successful
         if (result.success) {

sim/app/api/scheduled/execute/route.ts sim/app/api/schedules/execute/route.tssim/app/api/scheduled/execute/route.ts renamed to sim/app/api/schedules/execute/route.ts

@@ -3,6 +3,7 @@ import { and, eq } from 'drizzle-orm'
 import { v4 as uuidv4 } from 'uuid'
 import { createLogger } from '@/lib/logs/console-logger'
 import { persistExecutionError, persistExecutionLogs } from '@/lib/logs/execution-logger'
+import { buildTraceSpans } from '@/lib/logs/trace-spans'
 import { closeRedisConnection, hasProcessedMessage, markMessageAsProcessed } from '@/lib/redis'
 import { decryptSecret } from '@/lib/utils'
 import { mergeSubblockStateAsync } from '@/stores/workflows/utils'
@@ -558,8 +559,18 @@ async function processWebhook(
       executionTime: result.metadata?.duration,
     })
 
+    // Build trace spans from execution logs
+    const { traceSpans, totalDuration } = buildTraceSpans(result)
+
+    // Add trace spans to the execution result
+    const enrichedResult = {
+      ...result,
+      traceSpans,
+      totalDuration,
+    }
+
     // Log each execution step and the final result
-    await persistExecutionLogs(foundWorkflow.id, executionId, result, 'webhook')
+    await persistExecutionLogs(foundWorkflow.id, executionId, enrichedResult, 'webhook')
 
     // Return the execution result
     return NextResponse.json(result, { status: 200 })

sim/app/api/scheduled/route.ts sim/app/api/schedules/route.tssim/app/api/scheduled/route.ts renamed to sim/app/api/schedules/route.ts

@@ -4,6 +4,7 @@ import { v4 as uuidv4 } from 'uuid'
 import { z } from 'zod'
 import { createLogger } from '@/lib/logs/console-logger'
 import { persistExecutionError, persistExecutionLogs } from '@/lib/logs/execution-logger'
+import { buildTraceSpans } from '@/lib/logs/trace-spans'
 import { decryptSecret } from '@/lib/utils'
 import { mergeSubblockState } from '@/stores/workflows/utils'
 import { WorkflowState } from '@/stores/workflows/workflow/types'
@@ -158,8 +159,18 @@ async function executeWorkflow(workflow: any, requestId: string, input?: any) {
       executionTime: result.metadata?.duration,
     })
 
+    // Build trace spans from execution logs
+    const { traceSpans, totalDuration } = buildTraceSpans(result)
+
+    // Add trace spans to the execution result
+    const enrichedResult = {
+      ...result,
+      traceSpans,
+      totalDuration,
+    }
+
     // Log each execution step and the final result
-    await persistExecutionLogs(workflowId, executionId, result, 'api')
+    await persistExecutionLogs(workflowId, executionId, enrichedResult, 'api')
 
     return result
   } catch (error: any) {