From 3e4e7ddf98ae574f660a6589a92c18c3801bf942 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 24 Mar 2025 13:12:03 -0700 Subject: [PATCH 1/5] added user_stats table to track total number of workflow runs --- sim/app/api/schedules/execute/route.ts | 15 ++++-- sim/app/api/webhooks/trigger/[path]/route.ts | 6 +++ sim/app/api/workflows/[id]/execute/route.ts | 6 +++ sim/app/api/workflows/[id]/stats/route.ts | 50 +++++++++++++++++++ .../components/control-bar/control-bar.tsx | 11 ++++ sim/db/schema.ts | 40 ++++++++++----- sim/lib/workflows.ts | 9 ---- sim/lib/workflows/utils.ts | 21 ++++++++ 8 files changed, 133 insertions(+), 25 deletions(-) create mode 100644 sim/app/api/workflows/[id]/stats/route.ts delete mode 100644 sim/lib/workflows.ts create mode 100644 sim/lib/workflows/utils.ts diff --git a/sim/app/api/schedules/execute/route.ts b/sim/app/api/schedules/execute/route.ts index 689b0281bac..1f575b3f889 100644 --- a/sim/app/api/schedules/execute/route.ts +++ b/sim/app/api/schedules/execute/route.ts @@ -7,6 +7,7 @@ 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 { updateWorkflowRunCounts } from '@/lib/workflows/utils' import { mergeSubblockState } from '@/stores/workflows/utils' import { BlockState, WorkflowState } from '@/stores/workflows/workflow/types' import { db } from '@/db' @@ -326,6 +327,16 @@ export async function GET(req: NextRequest) { ) const result = await executor.execute(schedule.workflowId) + logger.info(`[${requestId}] Workflow execution completed: ${schedule.workflowId}`, { + success: result.success, + executionTime: result.metadata?.duration, + }) + + // Update workflow run counts if execution was successful + if (result.success) { + await updateWorkflowRunCounts(schedule.workflowId) + } + // Build trace spans from execution logs const { traceSpans, totalDuration } = buildTraceSpans(result) @@ -371,10 +382,6 @@ export async function GET(req: NextRequest) { nextRunAt: nextRetryAt, }) .where(eq(workflowSchedule.id, schedule.id)) - - logger.debug( - `[${requestId}] Scheduled retry for workflow ${schedule.workflowId} at ${nextRetryAt.toISOString()}` - ) } } catch (error: any) { logger.error( diff --git a/sim/app/api/webhooks/trigger/[path]/route.ts b/sim/app/api/webhooks/trigger/[path]/route.ts index a55c2f93df1..13924e78fbb 100644 --- a/sim/app/api/webhooks/trigger/[path]/route.ts +++ b/sim/app/api/webhooks/trigger/[path]/route.ts @@ -6,6 +6,7 @@ import { persistExecutionError, persistExecutionLogs } from '@/lib/logs/executio import { buildTraceSpans } from '@/lib/logs/trace-spans' import { closeRedisConnection, hasProcessedMessage, markMessageAsProcessed } from '@/lib/redis' import { decryptSecret } from '@/lib/utils' +import { updateWorkflowRunCounts } from '@/lib/workflows/utils' import { mergeSubblockStateAsync } from '@/stores/workflows/utils' import { db } from '@/db' import { environment, webhook, workflow } from '@/db/schema' @@ -559,6 +560,11 @@ async function processWebhook( executionTime: result.metadata?.duration, }) + // Update workflow run counts if execution was successful + if (result.success) { + await updateWorkflowRunCounts(foundWorkflow.id) + } + // Build trace spans from execution logs const { traceSpans, totalDuration } = buildTraceSpans(result) diff --git a/sim/app/api/workflows/[id]/execute/route.ts b/sim/app/api/workflows/[id]/execute/route.ts index c18cb6cc1e2..239178fc2bc 100644 --- a/sim/app/api/workflows/[id]/execute/route.ts +++ b/sim/app/api/workflows/[id]/execute/route.ts @@ -6,6 +6,7 @@ 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 { updateWorkflowRunCounts } from '@/lib/workflows/utils' import { mergeSubblockState } from '@/stores/workflows/utils' import { WorkflowState } from '@/stores/workflows/workflow/types' import { db } from '@/db' @@ -159,6 +160,11 @@ async function executeWorkflow(workflow: any, requestId: string, input?: any) { executionTime: result.metadata?.duration, }) + // Update workflow run counts if execution was successful + if (result.success) { + await updateWorkflowRunCounts(workflowId) + } + // Build trace spans from execution logs const { traceSpans, totalDuration } = buildTraceSpans(result) diff --git a/sim/app/api/workflows/[id]/stats/route.ts b/sim/app/api/workflows/[id]/stats/route.ts new file mode 100644 index 00000000000..91531c0e0c8 --- /dev/null +++ b/sim/app/api/workflows/[id]/stats/route.ts @@ -0,0 +1,50 @@ +import { NextRequest, NextResponse } from 'next/server' +import { eq, sql } from 'drizzle-orm' +import { db } from '@/db' +import { userWorkflowStats, workflow } from '@/db/schema' + +export async function POST(request: NextRequest, { params }: { params: { id: string } }) { + const { id } = params + const searchParams = request.nextUrl.searchParams + const runs = parseInt(searchParams.get('runs') || '1', 10) + + if (isNaN(runs) || runs < 1 || runs > 100) { + return NextResponse.json( + { error: 'Invalid number of runs. Must be between 1 and 100.' }, + { status: 400 } + ) + } + + try { + const [workflowRecord] = await db.select().from(workflow).where(eq(workflow.id, id)).limit(1) + + if (!workflowRecord) { + return NextResponse.json({ error: `Workflow ${id} not found` }, { status: 404 }) + } + + await db + .update(workflow) + .set({ + runCount: workflowRecord.runCount + runs, + lastRunAt: new Date(), + }) + .where(eq(workflow.id, id)) + + await db + .update(userWorkflowStats) + .set({ + totalWorkflowRuns: sql`total_workflow_runs + ${runs}`, + lastUpdated: new Date(), + }) + .where(eq(userWorkflowStats.userId, workflowRecord.userId)) + + return NextResponse.json({ + success: true, + runsAdded: runs, + newTotal: workflowRecord.runCount + runs, + }) + } catch (error) { + console.error('Error updating workflow stats:', error) + return NextResponse.json({ error: 'Internal server error' }, { status: 500 }) + } +} diff --git a/sim/app/w/[id]/components/control-bar/control-bar.tsx b/sim/app/w/[id]/components/control-bar/control-bar.tsx index 75f6c9fdc90..7b310ef8a40 100644 --- a/sim/app/w/[id]/components/control-bar/control-bar.tsx +++ b/sim/app/w/[id]/components/control-bar/control-bar.tsx @@ -313,6 +313,17 @@ export function ControlBar() { await handleRunWorkflow() setCompletedRuns(i + 1) } + + // Update workflow stats after all runs are complete + if (activeWorkflowId) { + const response = await fetch(`/api/workflows/${activeWorkflowId}/stats?runs=${runCount}`, { + method: 'POST', + }) + + if (!response.ok) { + throw new Error('Failed to update workflow stats') + } + } } catch (error) { logger.error('Error during multiple workflow runs:', { error }) addNotification('error', 'Failed to complete all workflow runs', activeWorkflowId) diff --git a/sim/db/schema.ts b/sim/db/schema.ts index 85f45395711..3b833ca4c7b 100644 --- a/sim/db/schema.ts +++ b/sim/db/schema.ts @@ -1,4 +1,4 @@ -import { boolean, json, pgTable, text, timestamp, uniqueIndex, integer } from 'drizzle-orm/pg-core' +import { boolean, integer, json, pgTable, text, timestamp, uniqueIndex } from 'drizzle-orm/pg-core' export const user = pgTable('user', { id: text('id').primaryKey(), @@ -67,6 +67,8 @@ export const workflow = pgTable('workflow', { apiKey: text('api_key'), isPublished: boolean('is_published').notNull().default(false), collaborators: json('collaborators').notNull().default('[]'), + runCount: integer('run_count').notNull().default(0), + lastRunAt: timestamp('last_run_at'), }) export const waitlist = pgTable('waitlist', { @@ -179,17 +181,31 @@ export const marketplace = pgTable('marketplace', { updatedAt: timestamp('updated_at').notNull().defaultNow(), }) -export const marketplaceStar = pgTable('marketplace_star', { +export const marketplaceStar = pgTable( + 'marketplace_star', + { + id: text('id').primaryKey(), + marketplaceId: text('marketplace_id') + .notNull() + .references(() => marketplace.id, { onDelete: 'cascade' }), + userId: text('user_id') + .notNull() + .references(() => user.id), + createdAt: timestamp('created_at').notNull().defaultNow(), + }, + (table) => { + return { + userMarketplaceIdx: uniqueIndex('user_marketplace_idx').on(table.userId, table.marketplaceId), + } + } +) + +export const userStats = pgTable('user_stats', { id: text('id').primaryKey(), - marketplaceId: text('marketplace_id') - .notNull() - .references(() => marketplace.id, { onDelete: 'cascade' }), userId: text('user_id') .notNull() - .references(() => user.id), - createdAt: timestamp('created_at').notNull().defaultNow(), -}, (table) => { - return { - userMarketplaceIdx: uniqueIndex('user_marketplace_idx').on(table.userId, table.marketplaceId), - } -}) \ No newline at end of file + .references(() => user.id, { onDelete: 'cascade' }) + .unique(), // One record per user + totalWorkflowRuns: integer('total_workflow_runs').notNull().default(0), + lastUpdated: timestamp('last_updated').notNull().defaultNow(), +}) diff --git a/sim/lib/workflows.ts b/sim/lib/workflows.ts deleted file mode 100644 index e4e39fde8cf..00000000000 --- a/sim/lib/workflows.ts +++ /dev/null @@ -1,9 +0,0 @@ -import { eq } from 'drizzle-orm' -import { db } from '@/db' -import { workflow as workflowTable } from '@/db/schema' - -export async function getWorkflowById(id: string) { - const workflows = await db.select().from(workflowTable).where(eq(workflowTable.id, id)).limit(1) - - return workflows[0] -} diff --git a/sim/lib/workflows/utils.ts b/sim/lib/workflows/utils.ts new file mode 100644 index 00000000000..7a46d194673 --- /dev/null +++ b/sim/lib/workflows/utils.ts @@ -0,0 +1,21 @@ +import { eq } from 'drizzle-orm' +import { db } from '@/db' +import { workflow as workflowTable } from '@/db/schema' + +export async function getWorkflowById(id: string) { + const workflows = await db.select().from(workflowTable).where(eq(workflowTable.id, id)).limit(1) + return workflows[0] +} + +export async function updateWorkflowRunCounts(workflowId: string, runs: number = 1) { + const response = await fetch(`/api/workflows/${workflowId}/stats?runs=${runs}`, { + method: 'POST', + }) + + if (!response.ok) { + const error = await response.json() + throw new Error(error.error || 'Failed to update workflow stats') + } + + return response.json() +} From ab21d07a07c4a3c9926af81acdb9ca758c27e7bf Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 24 Mar 2025 13:18:10 -0700 Subject: [PATCH 2/5] added metrics for different types of triggers and runs --- sim/app/api/db/workflow/route.ts | 15 +++++++++++++-- sim/app/api/schedules/execute/route.ts | 12 +++++++++++- sim/app/api/webhooks/trigger/[path]/route.ts | 13 +++++++++++-- sim/app/api/workflows/[id]/execute/route.ts | 13 +++++++++++-- sim/app/api/workflows/[id]/stats/route.ts | 8 ++++---- sim/db/schema.ts | 19 +++++++++++++++++-- 6 files changed, 67 insertions(+), 13 deletions(-) diff --git a/sim/app/api/db/workflow/route.ts b/sim/app/api/db/workflow/route.ts index ef923d7946b..346cc9a1028 100644 --- a/sim/app/api/db/workflow/route.ts +++ b/sim/app/api/db/workflow/route.ts @@ -1,10 +1,10 @@ import { NextRequest, NextResponse } from 'next/server' -import { eq } from 'drizzle-orm' +import { eq, sql } from 'drizzle-orm' import { z } from 'zod' import { getSession } from '@/lib/auth' import { createLogger } from '@/lib/logs/console-logger' import { db } from '@/db' -import { workflow } from '@/db/schema' +import { userStats, workflow } from '@/db/schema' const logger = createLogger('WorkflowAPI') @@ -129,6 +129,17 @@ export async function POST(req: NextRequest) { updatedAt: now, }) ) + + // Increment workflowsCreated in userStats + operations.push( + db + .update(userStats) + .set({ + workflowsCreated: sql`workflows_created + 1`, + lastActive: now, + }) + .where(eq(userStats.userId, session.user.id)) + ) } else { // Existing workflow - update if needed const needsUpdate = diff --git a/sim/app/api/schedules/execute/route.ts b/sim/app/api/schedules/execute/route.ts index 1f575b3f889..2db1be647cb 100644 --- a/sim/app/api/schedules/execute/route.ts +++ b/sim/app/api/schedules/execute/route.ts @@ -1,6 +1,7 @@ import { NextRequest, NextResponse } from 'next/server' import { Cron } from 'croner' import { eq, lte } from 'drizzle-orm' +import { sql } from 'drizzle-orm' import { v4 as uuidv4 } from 'uuid' import { z } from 'zod' import { createLogger } from '@/lib/logs/console-logger' @@ -11,7 +12,7 @@ import { updateWorkflowRunCounts } from '@/lib/workflows/utils' import { mergeSubblockState } from '@/stores/workflows/utils' import { BlockState, WorkflowState } from '@/stores/workflows/workflow/types' import { db } from '@/db' -import { environment, workflow, workflowSchedule } from '@/db/schema' +import { environment, userStats, workflow, workflowSchedule } from '@/db/schema' import { Executor } from '@/executor' import { Serializer } from '@/serializer' @@ -335,6 +336,15 @@ export async function GET(req: NextRequest) { // Update workflow run counts if execution was successful if (result.success) { await updateWorkflowRunCounts(schedule.workflowId) + + // Track scheduled execution in user stats + await db + .update(userStats) + .set({ + totalScheduledExecutions: sql`total_scheduled_executions + 1`, + lastActive: new Date(), + }) + .where(eq(userStats.userId, workflowRecord.userId)) } // Build trace spans from execution logs diff --git a/sim/app/api/webhooks/trigger/[path]/route.ts b/sim/app/api/webhooks/trigger/[path]/route.ts index 13924e78fbb..156f6167ca8 100644 --- a/sim/app/api/webhooks/trigger/[path]/route.ts +++ b/sim/app/api/webhooks/trigger/[path]/route.ts @@ -1,5 +1,5 @@ import { NextRequest, NextResponse } from 'next/server' -import { and, eq } from 'drizzle-orm' +import { and, eq, sql } from 'drizzle-orm' import { v4 as uuidv4 } from 'uuid' import { createLogger } from '@/lib/logs/console-logger' import { persistExecutionError, persistExecutionLogs } from '@/lib/logs/execution-logger' @@ -9,7 +9,7 @@ import { decryptSecret } from '@/lib/utils' import { updateWorkflowRunCounts } from '@/lib/workflows/utils' import { mergeSubblockStateAsync } from '@/stores/workflows/utils' import { db } from '@/db' -import { environment, webhook, workflow } from '@/db/schema' +import { environment, userStats, webhook, workflow } from '@/db/schema' import { Executor } from '@/executor' import { Serializer } from '@/serializer' @@ -563,6 +563,15 @@ async function processWebhook( // Update workflow run counts if execution was successful if (result.success) { await updateWorkflowRunCounts(foundWorkflow.id) + + // Track webhook trigger in user stats + await db + .update(userStats) + .set({ + totalWebhookTriggers: sql`total_webhook_triggers + 1`, + lastActive: new Date(), + }) + .where(eq(userStats.userId, foundWorkflow.userId)) } // Build trace spans from execution logs diff --git a/sim/app/api/workflows/[id]/execute/route.ts b/sim/app/api/workflows/[id]/execute/route.ts index 239178fc2bc..e8021709659 100644 --- a/sim/app/api/workflows/[id]/execute/route.ts +++ b/sim/app/api/workflows/[id]/execute/route.ts @@ -1,5 +1,5 @@ import { NextRequest } from 'next/server' -import { eq } from 'drizzle-orm' +import { eq, sql } from 'drizzle-orm' import { v4 as uuidv4 } from 'uuid' import { z } from 'zod' import { createLogger } from '@/lib/logs/console-logger' @@ -10,7 +10,7 @@ import { updateWorkflowRunCounts } from '@/lib/workflows/utils' import { mergeSubblockState } from '@/stores/workflows/utils' import { WorkflowState } from '@/stores/workflows/workflow/types' import { db } from '@/db' -import { environment } from '@/db/schema' +import { environment, userStats } from '@/db/schema' import { Executor } from '@/executor' import { Serializer } from '@/serializer' import { validateWorkflowAccess } from '../../middleware' @@ -163,6 +163,15 @@ async function executeWorkflow(workflow: any, requestId: string, input?: any) { // Update workflow run counts if execution was successful if (result.success) { await updateWorkflowRunCounts(workflowId) + + // Track API call in user stats + await db + .update(userStats) + .set({ + totalApiCalls: sql`total_api_calls + 1`, + lastActive: new Date(), + }) + .where(eq(userStats.userId, workflow.userId)) } // Build trace spans from execution logs diff --git a/sim/app/api/workflows/[id]/stats/route.ts b/sim/app/api/workflows/[id]/stats/route.ts index 91531c0e0c8..474d14d18c6 100644 --- a/sim/app/api/workflows/[id]/stats/route.ts +++ b/sim/app/api/workflows/[id]/stats/route.ts @@ -1,7 +1,7 @@ import { NextRequest, NextResponse } from 'next/server' import { eq, sql } from 'drizzle-orm' import { db } from '@/db' -import { userWorkflowStats, workflow } from '@/db/schema' +import { userStats, workflow } from '@/db/schema' export async function POST(request: NextRequest, { params }: { params: { id: string } }) { const { id } = params @@ -31,12 +31,12 @@ export async function POST(request: NextRequest, { params }: { params: { id: str .where(eq(workflow.id, id)) await db - .update(userWorkflowStats) + .update(userStats) .set({ totalWorkflowRuns: sql`total_workflow_runs + ${runs}`, - lastUpdated: new Date(), + lastActive: new Date(), }) - .where(eq(userWorkflowStats.userId, workflowRecord.userId)) + .where(eq(userStats.userId, workflowRecord.userId)) return NextResponse.json({ success: true, diff --git a/sim/db/schema.ts b/sim/db/schema.ts index 3b833ca4c7b..279d570c1d7 100644 --- a/sim/db/schema.ts +++ b/sim/db/schema.ts @@ -1,4 +1,13 @@ -import { boolean, integer, json, pgTable, text, timestamp, uniqueIndex } from 'drizzle-orm/pg-core' +import { + boolean, + decimal, + integer, + json, + pgTable, + text, + timestamp, + uniqueIndex, +} from 'drizzle-orm/pg-core' export const user = pgTable('user', { id: text('id').primaryKey(), @@ -207,5 +216,11 @@ export const userStats = pgTable('user_stats', { .references(() => user.id, { onDelete: 'cascade' }) .unique(), // One record per user totalWorkflowRuns: integer('total_workflow_runs').notNull().default(0), - lastUpdated: timestamp('last_updated').notNull().defaultNow(), + totalApiCalls: integer('total_api_calls').notNull().default(0), + totalWebhookTriggers: integer('total_webhook_triggers').notNull().default(0), + totalScheduledExecutions: integer('total_scheduled_executions').notNull().default(0), + totalTokensUsed: integer('total_tokens_used').notNull().default(0), + totalCost: decimal('total_cost').notNull().default('0'), + workflowsCreated: integer('workflows_created').notNull().default(0), + lastActive: timestamp('last_active').notNull().defaultNow(), }) From cd9b1b5afccca96a3c95845d2f26e41685eaac74 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 24 Mar 2025 13:22:51 -0700 Subject: [PATCH 3/5] add total_tokens_used and total_cost metrics --- sim/lib/logs/execution-logger.ts | 29 ++++++++++++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/sim/lib/logs/execution-logger.ts b/sim/lib/logs/execution-logger.ts index 735f887960b..146828a64f8 100644 --- a/sim/lib/logs/execution-logger.ts +++ b/sim/lib/logs/execution-logger.ts @@ -1,7 +1,8 @@ +import { eq, sql } from 'drizzle-orm' import { v4 as uuidv4 } from 'uuid' import { createLogger } from '@/lib/logs/console-logger' import { db } from '@/db' -import { workflowLogs } from '@/db/schema' +import { userStats, workflow, workflowLogs } from '@/db/schema' import { ExecutionResult as ExecutorResult } from '@/executor/types' const logger = createLogger('ExecutionLogger') @@ -69,6 +70,20 @@ export async function persistExecutionLogs( triggerType: 'api' | 'webhook' | 'schedule' | 'manual' ) { try { + // Get the workflow record to get the userId + const [workflowRecord] = await db + .select() + .from(workflow) + .where(eq(workflow.id, workflowId)) + .limit(1) + + if (!workflowRecord) { + logger.error(`Workflow ${workflowId} not found`) + return + } + + const userId = workflowRecord.userId + // Track accumulated cost data across all agent blocks let totalCost = 0 let totalInputCost = 0 @@ -520,6 +535,18 @@ export async function persistExecutionLogs( outputCost: totalOutputCost, models: Object.keys(modelCounts), }) + + // Update user stats with cost and token information + if (userId) { + await db + .update(userStats) + .set({ + totalTokensUsed: sql`total_tokens_used + ${totalTokens}`, + totalCost: sql`total_cost + ${totalCost}`, + lastActive: new Date(), + }) + .where(eq(userStats.userId, userId)) + } } // Log the final execution result From 5f40f84e787149213ae5b06af924c42a1b895b1c Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 24 Mar 2025 14:34:52 -0700 Subject: [PATCH 4/5] ran migrations for new table, fixed build issue. added a user/stats route to fetch all user stats for the dashboard we'll eventually create --- sim/app/api/db/workflow/route.ts | 15 +- sim/app/api/schedules/route.ts | 2 +- sim/app/api/user/stats/route.ts | 68 + sim/app/api/workflows/[id]/stats/route.ts | 75 +- sim/app/api/workflows/middleware.ts | 2 +- .../components/control-bar/control-bar.tsx | 2 + sim/db/migrations/0021_shocking_korath.sql | 16 + sim/db/migrations/meta/0021_snapshot.json | 1194 +++++++++++++++++ sim/db/migrations/meta/_journal.json | 7 + sim/db/schema.ts | 3 +- sim/lib/logs/execution-logger.ts | 49 +- sim/lib/workflows/utils.ts | 32 +- sim/stores/workflows/middleware.ts | 1 - 13 files changed, 1403 insertions(+), 63 deletions(-) create mode 100644 sim/app/api/user/stats/route.ts create mode 100644 sim/db/migrations/0021_shocking_korath.sql create mode 100644 sim/db/migrations/meta/0021_snapshot.json diff --git a/sim/app/api/db/workflow/route.ts b/sim/app/api/db/workflow/route.ts index 346cc9a1028..ef923d7946b 100644 --- a/sim/app/api/db/workflow/route.ts +++ b/sim/app/api/db/workflow/route.ts @@ -1,10 +1,10 @@ import { NextRequest, NextResponse } from 'next/server' -import { eq, sql } from 'drizzle-orm' +import { eq } from 'drizzle-orm' import { z } from 'zod' import { getSession } from '@/lib/auth' import { createLogger } from '@/lib/logs/console-logger' import { db } from '@/db' -import { userStats, workflow } from '@/db/schema' +import { workflow } from '@/db/schema' const logger = createLogger('WorkflowAPI') @@ -129,17 +129,6 @@ export async function POST(req: NextRequest) { updatedAt: now, }) ) - - // Increment workflowsCreated in userStats - operations.push( - db - .update(userStats) - .set({ - workflowsCreated: sql`workflows_created + 1`, - lastActive: now, - }) - .where(eq(userStats.userId, session.user.id)) - ) } else { // Existing workflow - update if needed const needsUpdate = diff --git a/sim/app/api/schedules/route.ts b/sim/app/api/schedules/route.ts index 0279b002282..e2798de4bf9 100644 --- a/sim/app/api/schedules/route.ts +++ b/sim/app/api/schedules/route.ts @@ -6,7 +6,7 @@ import { createLogger } from '@/lib/logs/console-logger' import { db } from '@/db' import { workflowSchedule } from '@/db/schema' -const logger = createLogger('Scheduled API') +const logger = createLogger('ScheduledAPI') /** * Get schedule information for a workflow diff --git a/sim/app/api/user/stats/route.ts b/sim/app/api/user/stats/route.ts new file mode 100644 index 00000000000..7c9120039f6 --- /dev/null +++ b/sim/app/api/user/stats/route.ts @@ -0,0 +1,68 @@ +import { NextRequest, NextResponse } from 'next/server' +import { eq, sql } from 'drizzle-orm' +import { getSession } from '@/lib/auth' +import { createLogger } from '@/lib/logs/console-logger' +import { db } from '@/db' +import { userStats, workflow } from '@/db/schema' + +const logger = createLogger('UserStatsAPI') + +/** + * GET endpoint to retrieve user statistics including the count of workflows + */ +export async function GET(request: NextRequest) { + try { + // Get the user session + const session = await getSession() + if (!session?.user?.id) { + logger.warn('Unauthorized user stats access attempt') + return NextResponse.json({ error: 'Unauthorized' }, { status: 401 }) + } + + const userId = session.user.id + + // Get workflow count for user + const [workflowCountResult] = await db + .select({ count: sql`count(*)::int` }) + .from(workflow) + .where(eq(workflow.userId, userId)) + + const workflowCount = workflowCountResult?.count || 0 + + // Get user stats record + const userStatsRecords = await db.select().from(userStats).where(eq(userStats.userId, userId)) + + // If no stats record exists, create one + if (userStatsRecords.length === 0) { + const newStats = { + id: crypto.randomUUID(), + userId, + totalManualExecutions: 0, + totalApiCalls: 0, + totalWebhookTriggers: 0, + totalScheduledExecutions: 0, + totalTokensUsed: 0, + totalCost: '0.00', + lastActive: new Date(), + } + + await db.insert(userStats).values(newStats) + + // Return the newly created stats with workflow count + return NextResponse.json({ + ...newStats, + workflowCount, + }) + } + + // Return stats with workflow count + const stats = userStatsRecords[0] + return NextResponse.json({ + ...stats, + workflowCount, + }) + } catch (error) { + logger.error('Error fetching user stats:', error) + return NextResponse.json({ error: 'Failed to fetch user statistics' }, { status: 500 }) + } +} diff --git a/sim/app/api/workflows/[id]/stats/route.ts b/sim/app/api/workflows/[id]/stats/route.ts index 474d14d18c6..cc75cb83baf 100644 --- a/sim/app/api/workflows/[id]/stats/route.ts +++ b/sim/app/api/workflows/[id]/stats/route.ts @@ -1,14 +1,18 @@ import { NextRequest, NextResponse } from 'next/server' import { eq, sql } from 'drizzle-orm' +import { createLogger } from '@/lib/logs/console-logger' import { db } from '@/db' import { userStats, workflow } from '@/db/schema' -export async function POST(request: NextRequest, { params }: { params: { id: string } }) { - const { id } = params +const logger = createLogger('WorkflowStatsAPI') + +export async function POST(request: NextRequest, { params }: { params: Promise<{ id: string }> }) { + const { id } = await params const searchParams = request.nextUrl.searchParams const runs = parseInt(searchParams.get('runs') || '1', 10) if (isNaN(runs) || runs < 1 || runs > 100) { + logger.error(`Invalid number of runs: ${runs}`) return NextResponse.json( { error: 'Invalid number of runs. Must be between 1 and 100.' }, { status: 400 } @@ -16,27 +20,62 @@ export async function POST(request: NextRequest, { params }: { params: { id: str } try { + // Get workflow record const [workflowRecord] = await db.select().from(workflow).where(eq(workflow.id, id)).limit(1) if (!workflowRecord) { return NextResponse.json({ error: `Workflow ${id} not found` }, { status: 404 }) } - await db - .update(workflow) - .set({ - runCount: workflowRecord.runCount + runs, - lastRunAt: new Date(), - }) - .where(eq(workflow.id, id)) - - await db - .update(userStats) - .set({ - totalWorkflowRuns: sql`total_workflow_runs + ${runs}`, - lastActive: new Date(), - }) - .where(eq(userStats.userId, workflowRecord.userId)) + // Update workflow runCount + try { + await db + .update(workflow) + .set({ + runCount: workflowRecord.runCount + runs, + lastRunAt: new Date(), + }) + .where(eq(workflow.id, id)) + } catch (error) { + logger.error('Error updating workflow runCount:', error) + throw error + } + + // Upsert user stats record + try { + // Check if record exists + const userStatsRecords = await db + .select() + .from(userStats) + .where(eq(userStats.userId, workflowRecord.userId)) + + if (userStatsRecords.length === 0) { + // Create new record if none exists + await db.insert(userStats).values({ + id: crypto.randomUUID(), + userId: workflowRecord.userId, + totalManualExecutions: runs, + totalApiCalls: 0, + totalWebhookTriggers: 0, + totalScheduledExecutions: 0, + totalTokensUsed: 0, + totalCost: '0.00', + lastActive: new Date(), + }) + } else { + // Update existing record + await db + .update(userStats) + .set({ + totalManualExecutions: sql`total_manual_executions + ${runs}`, + lastActive: new Date(), + }) + .where(eq(userStats.userId, workflowRecord.userId)) + } + } catch (error) { + logger.error(`Error upserting userStats for userId ${workflowRecord.userId}:`, error) + // Don't rethrow - we want to continue even if this fails + } return NextResponse.json({ success: true, @@ -44,7 +83,7 @@ export async function POST(request: NextRequest, { params }: { params: { id: str newTotal: workflowRecord.runCount + runs, }) } catch (error) { - console.error('Error updating workflow stats:', error) + logger.error('Error updating workflow stats:', error) return NextResponse.json({ error: 'Internal server error' }, { status: 500 }) } } diff --git a/sim/app/api/workflows/middleware.ts b/sim/app/api/workflows/middleware.ts index 74dcfeea473..9e1fe1c6b05 100644 --- a/sim/app/api/workflows/middleware.ts +++ b/sim/app/api/workflows/middleware.ts @@ -1,6 +1,6 @@ import { NextRequest } from 'next/server' import { createLogger } from '@/lib/logs/console-logger' -import { getWorkflowById } from '@/lib/workflows' +import { getWorkflowById } from '@/lib/workflows/utils' const logger = createLogger('WorkflowMiddleware') diff --git a/sim/app/w/[id]/components/control-bar/control-bar.tsx b/sim/app/w/[id]/components/control-bar/control-bar.tsx index 7b310ef8a40..5db495700e0 100644 --- a/sim/app/w/[id]/components/control-bar/control-bar.tsx +++ b/sim/app/w/[id]/components/control-bar/control-bar.tsx @@ -321,6 +321,8 @@ export function ControlBar() { }) if (!response.ok) { + const errorData = await response.json() + logger.error(`Failed to update workflow stats: ${JSON.stringify(errorData)}`) throw new Error('Failed to update workflow stats') } } diff --git a/sim/db/migrations/0021_shocking_korath.sql b/sim/db/migrations/0021_shocking_korath.sql new file mode 100644 index 00000000000..e1a49b17b85 --- /dev/null +++ b/sim/db/migrations/0021_shocking_korath.sql @@ -0,0 +1,16 @@ +CREATE TABLE "user_stats" ( + "id" text PRIMARY KEY NOT NULL, + "user_id" text NOT NULL, + "total_manual_executions" integer DEFAULT 0 NOT NULL, + "total_api_calls" integer DEFAULT 0 NOT NULL, + "total_webhook_triggers" integer DEFAULT 0 NOT NULL, + "total_scheduled_executions" integer DEFAULT 0 NOT NULL, + "total_tokens_used" integer DEFAULT 0 NOT NULL, + "total_cost" numeric DEFAULT '0' NOT NULL, + "last_active" timestamp DEFAULT now() NOT NULL, + CONSTRAINT "user_stats_user_id_unique" UNIQUE("user_id") +); +--> statement-breakpoint +ALTER TABLE "workflow" ADD COLUMN "run_count" integer DEFAULT 0 NOT NULL;--> statement-breakpoint +ALTER TABLE "workflow" ADD COLUMN "last_run_at" timestamp;--> statement-breakpoint +ALTER TABLE "user_stats" ADD CONSTRAINT "user_stats_user_id_user_id_fk" FOREIGN KEY ("user_id") REFERENCES "public"."user"("id") ON DELETE cascade ON UPDATE no action; \ No newline at end of file diff --git a/sim/db/migrations/meta/0021_snapshot.json b/sim/db/migrations/meta/0021_snapshot.json new file mode 100644 index 00000000000..914cddc62a4 --- /dev/null +++ b/sim/db/migrations/meta/0021_snapshot.json @@ -0,0 +1,1194 @@ +{ + "id": "30efcb43-6f93-4c90-8251-b9305702c135", + "prevId": "c38e056e-3751-4d26-a496-f9b0e49e7982", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.account": { + "name": "account", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "account_id": { + "name": "account_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "provider_id": { + "name": "provider_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "access_token": { + "name": "access_token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "refresh_token": { + "name": "refresh_token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "id_token": { + "name": "id_token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "access_token_expires_at": { + "name": "access_token_expires_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "refresh_token_expires_at": { + "name": "refresh_token_expires_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "scope": { + "name": "scope", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "password": { + "name": "password", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": { + "account_user_id_user_id_fk": { + "name": "account_user_id_user_id_fk", + "tableFrom": "account", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.api_key": { + "name": "api_key", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "key": { + "name": "key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "last_used": { + "name": "last_used", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "api_key_user_id_user_id_fk": { + "name": "api_key_user_id_user_id_fk", + "tableFrom": "api_key", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "api_key_key_unique": { + "name": "api_key_key_unique", + "nullsNotDistinct": false, + "columns": ["key"] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.environment": { + "name": "environment", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "variables": { + "name": "variables", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "environment_user_id_user_id_fk": { + "name": "environment_user_id_user_id_fk", + "tableFrom": "environment", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "environment_user_id_unique": { + "name": "environment_user_id_unique", + "nullsNotDistinct": false, + "columns": ["user_id"] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.marketplace": { + "name": "marketplace", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "workflow_id": { + "name": "workflow_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "state": { + "name": "state", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "author_id": { + "name": "author_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "author_name": { + "name": "author_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "stars": { + "name": "stars", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "views": { + "name": "views", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "category": { + "name": "category", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "marketplace_workflow_id_workflow_id_fk": { + "name": "marketplace_workflow_id_workflow_id_fk", + "tableFrom": "marketplace", + "tableTo": "workflow", + "columnsFrom": ["workflow_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "marketplace_author_id_user_id_fk": { + "name": "marketplace_author_id_user_id_fk", + "tableFrom": "marketplace", + "tableTo": "user", + "columnsFrom": ["author_id"], + "columnsTo": ["id"], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.marketplace_star": { + "name": "marketplace_star", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "marketplace_id": { + "name": "marketplace_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "user_marketplace_idx": { + "name": "user_marketplace_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "marketplace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "marketplace_star_marketplace_id_marketplace_id_fk": { + "name": "marketplace_star_marketplace_id_marketplace_id_fk", + "tableFrom": "marketplace_star", + "tableTo": "marketplace", + "columnsFrom": ["marketplace_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "marketplace_star_user_id_user_id_fk": { + "name": "marketplace_star_user_id_user_id_fk", + "tableFrom": "marketplace_star", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.session": { + "name": "session", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "token": { + "name": "token", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "ip_address": { + "name": "ip_address", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "user_agent": { + "name": "user_agent", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": { + "session_user_id_user_id_fk": { + "name": "session_user_id_user_id_fk", + "tableFrom": "session", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "session_token_unique": { + "name": "session_token_unique", + "nullsNotDistinct": false, + "columns": ["token"] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.settings": { + "name": "settings", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "general": { + "name": "general", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "settings_user_id_user_id_fk": { + "name": "settings_user_id_user_id_fk", + "tableFrom": "settings", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "settings_user_id_unique": { + "name": "settings_user_id_unique", + "nullsNotDistinct": false, + "columns": ["user_id"] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.user": { + "name": "user", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "email_verified": { + "name": "email_verified", + "type": "boolean", + "primaryKey": false, + "notNull": true + }, + "image": { + "name": "image", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "user_email_unique": { + "name": "user_email_unique", + "nullsNotDistinct": false, + "columns": ["email"] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.user_stats": { + "name": "user_stats", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "total_manual_executions": { + "name": "total_manual_executions", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "total_api_calls": { + "name": "total_api_calls", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "total_webhook_triggers": { + "name": "total_webhook_triggers", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "total_scheduled_executions": { + "name": "total_scheduled_executions", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "total_tokens_used": { + "name": "total_tokens_used", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "total_cost": { + "name": "total_cost", + "type": "numeric", + "primaryKey": false, + "notNull": true, + "default": "'0'" + }, + "last_active": { + "name": "last_active", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "user_stats_user_id_user_id_fk": { + "name": "user_stats_user_id_user_id_fk", + "tableFrom": "user_stats", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "user_stats_user_id_unique": { + "name": "user_stats_user_id_unique", + "nullsNotDistinct": false, + "columns": ["user_id"] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.verification": { + "name": "verification", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "identifier": { + "name": "identifier", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "value": { + "name": "value", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.waitlist": { + "name": "waitlist", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "waitlist_email_unique": { + "name": "waitlist_email_unique", + "nullsNotDistinct": false, + "columns": ["email"] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.webhook": { + "name": "webhook", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "workflow_id": { + "name": "workflow_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "path": { + "name": "path", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "provider_config": { + "name": "provider_config", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "is_active": { + "name": "is_active", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "path_idx": { + "name": "path_idx", + "columns": [ + { + "expression": "path", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "webhook_workflow_id_workflow_id_fk": { + "name": "webhook_workflow_id_workflow_id_fk", + "tableFrom": "webhook", + "tableTo": "workflow", + "columnsFrom": ["workflow_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.workflow": { + "name": "workflow", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "state": { + "name": "state", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "color": { + "name": "color", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'#3972F6'" + }, + "last_synced": { + "name": "last_synced", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "is_deployed": { + "name": "is_deployed", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "deployed_at": { + "name": "deployed_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "api_key": { + "name": "api_key", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "is_published": { + "name": "is_published", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "collaborators": { + "name": "collaborators", + "type": "json", + "primaryKey": false, + "notNull": true, + "default": "'[]'" + }, + "run_count": { + "name": "run_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "last_run_at": { + "name": "last_run_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "workflow_user_id_user_id_fk": { + "name": "workflow_user_id_user_id_fk", + "tableFrom": "workflow", + "tableTo": "user", + "columnsFrom": ["user_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.workflow_logs": { + "name": "workflow_logs", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "workflow_id": { + "name": "workflow_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "execution_id": { + "name": "execution_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "level": { + "name": "level", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "message": { + "name": "message", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "duration": { + "name": "duration", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "trigger": { + "name": "trigger", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "metadata": { + "name": "metadata", + "type": "json", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "workflow_logs_workflow_id_workflow_id_fk": { + "name": "workflow_logs_workflow_id_workflow_id_fk", + "tableFrom": "workflow_logs", + "tableTo": "workflow", + "columnsFrom": ["workflow_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.workflow_schedule": { + "name": "workflow_schedule", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "workflow_id": { + "name": "workflow_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "cron_expression": { + "name": "cron_expression", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "next_run_at": { + "name": "next_run_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "last_ran_at": { + "name": "last_ran_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "trigger_type": { + "name": "trigger_type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "workflow_schedule_workflow_id_workflow_id_fk": { + "name": "workflow_schedule_workflow_id_workflow_id_fk", + "tableFrom": "workflow_schedule", + "tableTo": "workflow", + "columnsFrom": ["workflow_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "workflow_schedule_workflow_id_unique": { + "name": "workflow_schedule_workflow_id_unique", + "nullsNotDistinct": false, + "columns": ["workflow_id"] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} diff --git a/sim/db/migrations/meta/_journal.json b/sim/db/migrations/meta/_journal.json index 6f908b4fa00..a872aeb4c3e 100644 --- a/sim/db/migrations/meta/_journal.json +++ b/sim/db/migrations/meta/_journal.json @@ -148,6 +148,13 @@ "when": 1742552444912, "tag": "0020_clear_skreet", "breakpoints": true + }, + { + "idx": 21, + "version": "7", + "when": 1742850849852, + "tag": "0021_shocking_korath", + "breakpoints": true } ] } diff --git a/sim/db/schema.ts b/sim/db/schema.ts index 279d570c1d7..d9216957bee 100644 --- a/sim/db/schema.ts +++ b/sim/db/schema.ts @@ -215,12 +215,11 @@ export const userStats = pgTable('user_stats', { .notNull() .references(() => user.id, { onDelete: 'cascade' }) .unique(), // One record per user - totalWorkflowRuns: integer('total_workflow_runs').notNull().default(0), + totalManualExecutions: integer('total_manual_executions').notNull().default(0), totalApiCalls: integer('total_api_calls').notNull().default(0), totalWebhookTriggers: integer('total_webhook_triggers').notNull().default(0), totalScheduledExecutions: integer('total_scheduled_executions').notNull().default(0), totalTokensUsed: integer('total_tokens_used').notNull().default(0), totalCost: decimal('total_cost').notNull().default('0'), - workflowsCreated: integer('workflows_created').notNull().default(0), lastActive: timestamp('last_active').notNull().defaultNow(), }) diff --git a/sim/lib/logs/execution-logger.ts b/sim/lib/logs/execution-logger.ts index 146828a64f8..3f8ec9a35f6 100644 --- a/sim/lib/logs/execution-logger.ts +++ b/sim/lib/logs/execution-logger.ts @@ -527,25 +527,38 @@ export async function persistExecutionLogs( } } - logger.info(`Workflow execution total cost: ${totalCost}`, { - workflowId, - executionId, - totalCost, - inputCost: totalInputCost, - outputCost: totalOutputCost, - models: Object.keys(modelCounts), - }) - - // Update user stats with cost and token information if (userId) { - await db - .update(userStats) - .set({ - totalTokensUsed: sql`total_tokens_used + ${totalTokens}`, - totalCost: sql`total_cost + ${totalCost}`, - lastActive: new Date(), - }) - .where(eq(userStats.userId, userId)) + try { + const userStatsRecords = await db + .select() + .from(userStats) + .where(eq(userStats.userId, userId)) + + if (userStatsRecords.length === 0) { + await db.insert(userStats).values({ + id: crypto.randomUUID(), + userId: userId, + totalManualExecutions: 0, + totalApiCalls: 0, + totalWebhookTriggers: 0, + totalScheduledExecutions: 0, + totalTokensUsed: totalTokens, + totalCost: totalCost.toString(), + lastActive: new Date(), + }) + } else { + await db + .update(userStats) + .set({ + totalTokensUsed: sql`total_tokens_used + ${totalTokens}`, + totalCost: sql`total_cost + ${totalCost}`, + lastActive: new Date(), + }) + .where(eq(userStats.userId, userId)) + } + } catch (error) { + logger.error(`Error upserting user stats:`, error) + } } } diff --git a/sim/lib/workflows/utils.ts b/sim/lib/workflows/utils.ts index 7a46d194673..c4b845d7ffc 100644 --- a/sim/lib/workflows/utils.ts +++ b/sim/lib/workflows/utils.ts @@ -1,6 +1,9 @@ import { eq } from 'drizzle-orm' +import { createLogger } from '@/lib/logs/console-logger' import { db } from '@/db' -import { workflow as workflowTable } from '@/db/schema' +import { userStats, workflow as workflowTable } from '@/db/schema' + +const logger = createLogger('WorkflowUtils') export async function getWorkflowById(id: string) { const workflows = await db.select().from(workflowTable).where(eq(workflowTable.id, id)).limit(1) @@ -8,14 +11,25 @@ export async function getWorkflowById(id: string) { } export async function updateWorkflowRunCounts(workflowId: string, runs: number = 1) { - const response = await fetch(`/api/workflows/${workflowId}/stats?runs=${runs}`, { - method: 'POST', - }) + try { + const workflow = await getWorkflowById(workflowId) + if (!workflow) { + logger.error(`Workflow ${workflowId} not found`) + throw new Error(`Workflow ${workflowId} not found`) + } - if (!response.ok) { - const error = await response.json() - throw new Error(error.error || 'Failed to update workflow stats') - } + const response = await fetch(`/api/workflows/${workflowId}/stats?runs=${runs}`, { + method: 'POST', + }) - return response.json() + if (!response.ok) { + const error = await response.json() + throw new Error(error.error || 'Failed to update workflow stats') + } + + return response.json() + } catch (error) { + logger.error(`Error updating workflow run counts:`, error) + throw error + } } diff --git a/sim/stores/workflows/middleware.ts b/sim/stores/workflows/middleware.ts index 43e947e7dc3..299919e5871 100644 --- a/sim/stores/workflows/middleware.ts +++ b/sim/stores/workflows/middleware.ts @@ -2,7 +2,6 @@ import { StateCreator } from 'zustand' import { saveSubblockValues, saveWorkflowState } from './persistence' import { useWorkflowRegistry } from './registry/store' import { useSubBlockStore } from './subblock/store' -import { mergeSubblockState } from './utils' import { WorkflowState, WorkflowStore } from './workflow/types' // Types From 6cec4c565adc741efe986ba9f3c13838d3072209 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 24 Mar 2025 14:46:50 -0700 Subject: [PATCH 5/5] fix bug with api deployment status not appearing --- sim/app/api/workflows/[id]/deploy/route.ts | 53 +++++++++++++++++++ .../components/control-bar/control-bar.tsx | 2 +- 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/sim/app/api/workflows/[id]/deploy/route.ts b/sim/app/api/workflows/[id]/deploy/route.ts index 86a4ee2a1cb..296435c1c34 100644 --- a/sim/app/api/workflows/[id]/deploy/route.ts +++ b/sim/app/api/workflows/[id]/deploy/route.ts @@ -12,6 +12,59 @@ const logger = createLogger('WorkflowDeployAPI') export const dynamic = 'force-dynamic' export const runtime = 'nodejs' +export async function GET(request: NextRequest, { params }: { params: Promise<{ id: string }> }) { + const requestId = crypto.randomUUID().slice(0, 8) + const { id } = await params + + try { + logger.debug(`[${requestId}] Fetching deployment info for workflow: ${id}`) + const validation = await validateWorkflowAccess(request, id, false) + + if (validation.error) { + logger.warn(`[${requestId}] Failed to fetch deployment info: ${validation.error.message}`) + return createErrorResponse(validation.error.message, validation.error.status) + } + + // Fetch the workflow information including deployment details + const result = await db + .select({ + apiKey: workflow.apiKey, + isDeployed: workflow.isDeployed, + deployedAt: workflow.deployedAt, + }) + .from(workflow) + .where(eq(workflow.id, id)) + .limit(1) + + if (result.length === 0) { + logger.warn(`[${requestId}] Workflow not found: ${id}`) + return createErrorResponse('Workflow not found', 404) + } + + const workflowData = result[0] + + // If the workflow is not deployed, return appropriate response + if (!workflowData.isDeployed || !workflowData.apiKey) { + logger.info(`[${requestId}] Workflow is not deployed: ${id}`) + return createSuccessResponse({ + isDeployed: false, + apiKey: null, + deployedAt: null, + }) + } + + logger.info(`[${requestId}] Successfully retrieved deployment info: ${id}`) + return createSuccessResponse({ + apiKey: workflowData.apiKey, + isDeployed: workflowData.isDeployed, + deployedAt: workflowData.deployedAt, + }) + } catch (error: any) { + logger.error(`[${requestId}] Error fetching deployment info: ${id}`, error) + return createErrorResponse(error.message || 'Failed to fetch deployment information', 500) + } +} + export async function POST(request: NextRequest, { params }: { params: Promise<{ id: string }> }) { const requestId = crypto.randomUUID().slice(0, 8) const { id } = await params diff --git a/sim/app/w/[id]/components/control-bar/control-bar.tsx b/sim/app/w/[id]/components/control-bar/control-bar.tsx index 5db495700e0..56547923cb1 100644 --- a/sim/app/w/[id]/components/control-bar/control-bar.tsx +++ b/sim/app/w/[id]/components/control-bar/control-bar.tsx @@ -208,7 +208,7 @@ export function ControlBar() { try { setIsDeploying(true) - const response = await fetch(`/api/workflows/${activeWorkflowId}/deploy/info`) + const response = await fetch(`/api/workflows/${activeWorkflowId}/deploy`) if (!response.ok) throw new Error('Failed to fetch deployment info') const { apiKey } = await response.json()