feat(logs-api): expose logs as api + can subscribe to workflow execution using webhook url (#1287)

* feat(logs-api): expose logs as api + can subscribe to workflow exection using webhook url

* fix scroll

* Update apps/docs/content/docs/execution/api.mdx

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* fix rate limits

* address greptile comments

* remove unused file

* address more greptile comments

* minor UI changes

* fix atomicity to prevent races

* make search param sensible

---------

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

Author: icecrasher321

apps/docs/content/docs/execution/api.mdx

@@ -1,4 +1,4 @@
 {
   "title": "Execution",
-  "pages": ["basics", "advanced"]
+  "pages": ["basics", "advanced", "api"]
 }

apps/docs/content/docs/execution/meta.json

@@ -0,0 +1,64 @@
+import { eq } from 'drizzle-orm'
+import type { NextRequest } from 'next/server'
+import { createLogger } from '@/lib/logs/console/logger'
+import { db } from '@/db'
+import { apiKey as apiKeyTable } from '@/db/schema'
+
+const logger = createLogger('V1Auth')
+
+export interface AuthResult {
+  authenticated: boolean
+  userId?: string
+  error?: string
+}
+
+export async function authenticateApiKey(request: NextRequest): Promise<AuthResult> {
+  const apiKey = request.headers.get('x-api-key')
+
+  if (!apiKey) {
+    return {
+      authenticated: false,
+      error: 'API key required',
+    }
+  }
+
+  try {
+    const [keyRecord] = await db
+      .select({
+        userId: apiKeyTable.userId,
+        expiresAt: apiKeyTable.expiresAt,
+      })
+      .from(apiKeyTable)
+      .where(eq(apiKeyTable.key, apiKey))
+      .limit(1)
+
+    if (!keyRecord) {
+      logger.warn('Invalid API key attempted', { keyPrefix: apiKey.slice(0, 8) })
+      return {
+        authenticated: false,
+        error: 'Invalid API key',
+      }
+    }
+
+    if (keyRecord.expiresAt && keyRecord.expiresAt < new Date()) {
+      logger.warn('Expired API key attempted', { userId: keyRecord.userId })
+      return {
+        authenticated: false,
+        error: 'API key expired',
+      }
+    }
+
+    await db.update(apiKeyTable).set({ lastUsed: new Date() }).where(eq(apiKeyTable.key, apiKey))
+
+    return {
+      authenticated: true,
+      userId: keyRecord.userId,
+    }
+  } catch (error) {
+    logger.error('API key authentication error', { error })
+    return {
+      authenticated: false,
+      error: 'Authentication failed',
+    }
+  }
+}

apps/sim/app/api/v1/auth.ts

@@ -0,0 +1,106 @@
+import { and, eq } from 'drizzle-orm'
+import { type NextRequest, NextResponse } from 'next/server'
+import { createLogger } from '@/lib/logs/console/logger'
+import { createApiResponse, getUserLimits } from '@/app/api/v1/logs/meta'
+import { checkRateLimit, createRateLimitResponse } from '@/app/api/v1/middleware'
+import { db } from '@/db'
+import { permissions, workflow, workflowExecutionLogs } from '@/db/schema'
+
+const logger = createLogger('V1LogDetailsAPI')
+
+export const revalidate = 0
+
+export async function GET(request: NextRequest, { params }: { params: Promise<{ id: string }> }) {
+  const requestId = crypto.randomUUID().slice(0, 8)
+
+  try {
+    const rateLimit = await checkRateLimit(request, 'logs-detail')
+    if (!rateLimit.allowed) {
+      return createRateLimitResponse(rateLimit)
+    }
+
+    const userId = rateLimit.userId!
+    const { id } = await params
+
+    const rows = await db
+      .select({
+        id: workflowExecutionLogs.id,
+        workflowId: workflowExecutionLogs.workflowId,
+        executionId: workflowExecutionLogs.executionId,
+        stateSnapshotId: workflowExecutionLogs.stateSnapshotId,
+        level: workflowExecutionLogs.level,
+        trigger: workflowExecutionLogs.trigger,
+        startedAt: workflowExecutionLogs.startedAt,
+        endedAt: workflowExecutionLogs.endedAt,
+        totalDurationMs: workflowExecutionLogs.totalDurationMs,
+        executionData: workflowExecutionLogs.executionData,
+        cost: workflowExecutionLogs.cost,
+        files: workflowExecutionLogs.files,
+        createdAt: workflowExecutionLogs.createdAt,
+        workflowName: workflow.name,
+        workflowDescription: workflow.description,
+        workflowColor: workflow.color,
+        workflowFolderId: workflow.folderId,
+        workflowUserId: workflow.userId,
+        workflowWorkspaceId: workflow.workspaceId,
+        workflowCreatedAt: workflow.createdAt,
+        workflowUpdatedAt: workflow.updatedAt,
+      })
+      .from(workflowExecutionLogs)
+      .innerJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id))
+      .innerJoin(
+        permissions,
+        and(
+          eq(permissions.entityType, 'workspace'),
+          eq(permissions.entityId, workflow.workspaceId),
+          eq(permissions.userId, userId)
+        )
+      )
+      .where(eq(workflowExecutionLogs.id, id))
+      .limit(1)
+
+    const log = rows[0]
+    if (!log) {
+      return NextResponse.json({ error: 'Log not found' }, { status: 404 })
+    }
+
+    const workflowSummary = {
+      id: log.workflowId,
+      name: log.workflowName,
+      description: log.workflowDescription,
+      color: log.workflowColor,
+      folderId: log.workflowFolderId,
+      userId: log.workflowUserId,
+      workspaceId: log.workflowWorkspaceId,
+      createdAt: log.workflowCreatedAt,
+      updatedAt: log.workflowUpdatedAt,
+    }
+
+    const response = {
+      id: log.id,
+      workflowId: log.workflowId,
+      executionId: log.executionId,
+      level: log.level,
+      trigger: log.trigger,
+      startedAt: log.startedAt.toISOString(),
+      endedAt: log.endedAt?.toISOString() || null,
+      totalDurationMs: log.totalDurationMs,
+      files: log.files || undefined,
+      workflow: workflowSummary,
+      executionData: log.executionData as any,
+      cost: log.cost as any,
+      createdAt: log.createdAt.toISOString(),
+    }
+
+    // Get user's workflow execution limits and usage
+    const limits = await getUserLimits(userId)
+
+    // Create response with limits information
+    const apiResponse = createApiResponse({ data: response }, limits, rateLimit)
+
+    return NextResponse.json(apiResponse.body, { headers: apiResponse.headers })
+  } catch (error: any) {
+    logger.error(`[${requestId}] Log details fetch error`, { error: error.message })
+    return NextResponse.json({ error: 'Internal server error' }, { status: 500 })
+  }
+}

apps/sim/app/api/v1/logs/[id]/route.ts

@@ -0,0 +1,100 @@
+import { and, eq } from 'drizzle-orm'
+import { type NextRequest, NextResponse } from 'next/server'
+import { createLogger } from '@/lib/logs/console/logger'
+import { createApiResponse, getUserLimits } from '@/app/api/v1/logs/meta'
+import { checkRateLimit, createRateLimitResponse } from '@/app/api/v1/middleware'
+import { db } from '@/db'
+import {
+  permissions,
+  workflow,
+  workflowExecutionLogs,
+  workflowExecutionSnapshots,
+} from '@/db/schema'
+
+const logger = createLogger('V1ExecutionAPI')
+
+export async function GET(
+  request: NextRequest,
+  { params }: { params: Promise<{ executionId: string }> }
+) {
+  try {
+    const rateLimit = await checkRateLimit(request, 'logs-detail')
+    if (!rateLimit.allowed) {
+      return createRateLimitResponse(rateLimit)
+    }
+
+    const userId = rateLimit.userId!
+    const { executionId } = await params
+
+    logger.debug(`Fetching execution data for: ${executionId}`)
+
+    const rows = await db
+      .select({
+        log: workflowExecutionLogs,
+        workflow: workflow,
+      })
+      .from(workflowExecutionLogs)
+      .innerJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id))
+      .innerJoin(
+        permissions,
+        and(
+          eq(permissions.entityType, 'workspace'),
+          eq(permissions.entityId, workflow.workspaceId),
+          eq(permissions.userId, userId)
+        )
+      )
+      .where(eq(workflowExecutionLogs.executionId, executionId))
+      .limit(1)
+
+    if (rows.length === 0) {
+      return NextResponse.json({ error: 'Workflow execution not found' }, { status: 404 })
+    }
+
+    const { log: workflowLog } = rows[0]
+
+    const [snapshot] = await db
+      .select()
+      .from(workflowExecutionSnapshots)
+      .where(eq(workflowExecutionSnapshots.id, workflowLog.stateSnapshotId))
+      .limit(1)
+
+    if (!snapshot) {
+      return NextResponse.json({ error: 'Workflow state snapshot not found' }, { status: 404 })
+    }
+
+    const response = {
+      executionId,
+      workflowId: workflowLog.workflowId,
+      workflowState: snapshot.stateData,
+      executionMetadata: {
+        trigger: workflowLog.trigger,
+        startedAt: workflowLog.startedAt.toISOString(),
+        endedAt: workflowLog.endedAt?.toISOString(),
+        totalDurationMs: workflowLog.totalDurationMs,
+        cost: workflowLog.cost || null,
+      },
+    }
+
+    logger.debug(`Successfully fetched execution data for: ${executionId}`)
+    logger.debug(
+      `Workflow state contains ${Object.keys((snapshot.stateData as any)?.blocks || {}).length} blocks`
+    )
+
+    // Get user's workflow execution limits and usage
+    const limits = await getUserLimits(userId)
+
+    // Create response with limits information
+    const apiResponse = createApiResponse(
+      {
+        ...response,
+      },
+      limits,
+      rateLimit
+    )
+
+    return NextResponse.json(apiResponse.body, { headers: apiResponse.headers })
+  } catch (error) {
+    logger.error('Error fetching execution data:', error)
+    return NextResponse.json({ error: 'Failed to fetch execution data' }, { status: 500 })
+  }
+}

apps/sim/app/api/v1/logs/executions/[executionId]/route.ts

@@ -0,0 +1,110 @@
+import { and, desc, eq, gte, inArray, lte, type SQL, sql } from 'drizzle-orm'
+import { workflow, workflowExecutionLogs } from '@/db/schema'
+
+export interface LogFilters {
+  workspaceId: string
+  workflowIds?: string[]
+  folderIds?: string[]
+  triggers?: string[]
+  level?: 'info' | 'error'
+  startDate?: Date
+  endDate?: Date
+  executionId?: string
+  minDurationMs?: number
+  maxDurationMs?: number
+  minCost?: number
+  maxCost?: number
+  model?: string
+  cursor?: {
+    startedAt: string
+    id: string
+  }
+  order?: 'desc' | 'asc'
+}
+
+export function buildLogFilters(filters: LogFilters): SQL<unknown> {
+  const conditions: SQL<unknown>[] = []
+
+  // Required: workspace and permissions check
+  conditions.push(eq(workflow.workspaceId, filters.workspaceId))
+
+  // Cursor-based pagination
+  if (filters.cursor) {
+    const cursorDate = new Date(filters.cursor.startedAt)
+    if (filters.order === 'desc') {
+      conditions.push(
+        sql`(${workflowExecutionLogs.startedAt}, ${workflowExecutionLogs.id}) < (${cursorDate}, ${filters.cursor.id})`
+      )
+    } else {
+      conditions.push(
+        sql`(${workflowExecutionLogs.startedAt}, ${workflowExecutionLogs.id}) > (${cursorDate}, ${filters.cursor.id})`
+      )
+    }
+  }
+
+  // Workflow IDs filter
+  if (filters.workflowIds && filters.workflowIds.length > 0) {
+    conditions.push(inArray(workflow.id, filters.workflowIds))
+  }
+
+  // Folder IDs filter
+  if (filters.folderIds && filters.folderIds.length > 0) {
+    conditions.push(inArray(workflow.folderId, filters.folderIds))
+  }
+
+  // Triggers filter
+  if (filters.triggers && filters.triggers.length > 0 && !filters.triggers.includes('all')) {
+    conditions.push(inArray(workflowExecutionLogs.trigger, filters.triggers))
+  }
+
+  // Level filter
+  if (filters.level) {
+    conditions.push(eq(workflowExecutionLogs.level, filters.level))
+  }
+
+  // Date range filters
+  if (filters.startDate) {
+    conditions.push(gte(workflowExecutionLogs.startedAt, filters.startDate))
+  }
+
+  if (filters.endDate) {
+    conditions.push(lte(workflowExecutionLogs.startedAt, filters.endDate))
+  }
+
+  // Search filter (execution ID)
+  if (filters.executionId) {
+    conditions.push(eq(workflowExecutionLogs.executionId, filters.executionId))
+  }
+
+  // Duration filters
+  if (filters.minDurationMs !== undefined) {
+    conditions.push(gte(workflowExecutionLogs.totalDurationMs, filters.minDurationMs))
+  }
+
+  if (filters.maxDurationMs !== undefined) {
+    conditions.push(lte(workflowExecutionLogs.totalDurationMs, filters.maxDurationMs))
+  }
+
+  // Cost filters
+  if (filters.minCost !== undefined) {
+    conditions.push(sql`(${workflowExecutionLogs.cost}->>'total')::numeric >= ${filters.minCost}`)
+  }
+
+  if (filters.maxCost !== undefined) {
+    conditions.push(sql`(${workflowExecutionLogs.cost}->>'total')::numeric <= ${filters.maxCost}`)
+  }
+
+  // Model filter
+  if (filters.model) {
+    conditions.push(sql`${workflowExecutionLogs.cost}->>'models' ? ${filters.model}`)
+  }
+
+  // Combine all conditions with AND
+  return conditions.length > 0 ? and(...conditions)! : sql`true`
+}
+
+export function getOrderBy(order: 'desc' | 'asc' = 'desc') {
+  return order === 'desc'
+    ? desc(workflowExecutionLogs.startedAt)
+    : sql`${workflowExecutionLogs.startedAt} ASC`
+}