From 78e701d8bbf2ebe19fb375af9dce2ec605412974 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 24 Mar 2025 14:49:54 -0700 Subject: [PATCH] feat(metrics): add user_stats table that tracks more granular usage information per user (#177) * added user_stats table to track total number of workflow runs * added metrics for different types of triggers and runs * add total_tokens_used and total_cost metrics * 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 * fix bug with api deployment status not appearing --- sim/app/api/schedules/execute/route.ts | 27 +- sim/app/api/schedules/route.ts | 2 +- sim/app/api/user/stats/route.ts | 68 + sim/app/api/webhooks/trigger/[path]/route.ts | 19 +- sim/app/api/workflows/[id]/deploy/route.ts | 53 + sim/app/api/workflows/[id]/execute/route.ts | 19 +- sim/app/api/workflows/[id]/stats/route.ts | 89 ++ sim/app/api/workflows/middleware.ts | 2 +- .../components/control-bar/control-bar.tsx | 15 +- 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 | 54 +- sim/lib/logs/execution-logger.ts | 58 +- sim/lib/workflows.ts | 9 - sim/lib/workflows/utils.ts | 35 + sim/stores/workflows/middleware.ts | 1 - 17 files changed, 1625 insertions(+), 43 deletions(-) create mode 100644 sim/app/api/user/stats/route.ts create mode 100644 sim/app/api/workflows/[id]/stats/route.ts create mode 100644 sim/db/migrations/0021_shocking_korath.sql create mode 100644 sim/db/migrations/meta/0021_snapshot.json 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 689b0281b..2db1be647 100644 --- a/sim/app/api/schedules/execute/route.ts +++ b/sim/app/api/schedules/execute/route.ts @@ -1,16 +1,18 @@ 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' 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' -import { environment, workflow, workflowSchedule } from '@/db/schema' +import { environment, userStats, workflow, workflowSchedule } from '@/db/schema' import { Executor } from '@/executor' import { Serializer } from '@/serializer' @@ -326,6 +328,25 @@ 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) + + // 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 const { traceSpans, totalDuration } = buildTraceSpans(result) @@ -371,10 +392,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/schedules/route.ts b/sim/app/api/schedules/route.ts index 0279b0022..e2798de4b 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 000000000..7c9120039 --- /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/webhooks/trigger/[path]/route.ts b/sim/app/api/webhooks/trigger/[path]/route.ts index a55c2f93d..156f6167c 100644 --- a/sim/app/api/webhooks/trigger/[path]/route.ts +++ b/sim/app/api/webhooks/trigger/[path]/route.ts @@ -1,14 +1,15 @@ 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' 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' +import { environment, userStats, webhook, workflow } from '@/db/schema' import { Executor } from '@/executor' import { Serializer } from '@/serializer' @@ -559,6 +560,20 @@ async function processWebhook( executionTime: result.metadata?.duration, }) + // 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 const { traceSpans, totalDuration } = buildTraceSpans(result) diff --git a/sim/app/api/workflows/[id]/deploy/route.ts b/sim/app/api/workflows/[id]/deploy/route.ts index 86a4ee2a1..296435c1c 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/api/workflows/[id]/execute/route.ts b/sim/app/api/workflows/[id]/execute/route.ts index c18cb6cc1..e80217096 100644 --- a/sim/app/api/workflows/[id]/execute/route.ts +++ b/sim/app/api/workflows/[id]/execute/route.ts @@ -1,15 +1,16 @@ 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' 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' -import { environment } from '@/db/schema' +import { environment, userStats } from '@/db/schema' import { Executor } from '@/executor' import { Serializer } from '@/serializer' import { validateWorkflowAccess } from '../../middleware' @@ -159,6 +160,20 @@ 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) + + // 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 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 000000000..cc75cb83b --- /dev/null +++ b/sim/app/api/workflows/[id]/stats/route.ts @@ -0,0 +1,89 @@ +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' + +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 } + ) + } + + 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 }) + } + + // 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, + runsAdded: runs, + newTotal: workflowRecord.runCount + runs, + }) + } catch (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 74dcfeea4..9e1fe1c6b 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 75f6c9fdc..56547923c 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() @@ -313,6 +313,19 @@ 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) { + const errorData = await response.json() + logger.error(`Failed to update workflow stats: ${JSON.stringify(errorData)}`) + 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/migrations/0021_shocking_korath.sql b/sim/db/migrations/0021_shocking_korath.sql new file mode 100644 index 000000000..e1a49b17b --- /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 000000000..914cddc62 --- /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 6f908b4fa..a872aeb4c 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 85f453957..d9216957b 100644 --- a/sim/db/schema.ts +++ b/sim/db/schema.ts @@ -1,4 +1,13 @@ -import { boolean, json, pgTable, text, timestamp, uniqueIndex, integer } 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(), @@ -67,6 +76,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 +190,36 @@ 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 + 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'), + lastActive: timestamp('last_active').notNull().defaultNow(), +}) diff --git a/sim/lib/logs/execution-logger.ts b/sim/lib/logs/execution-logger.ts index 735f88796..3f8ec9a35 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 @@ -512,14 +527,39 @@ export async function persistExecutionLogs( } } - logger.info(`Workflow execution total cost: ${totalCost}`, { - workflowId, - executionId, - totalCost, - inputCost: totalInputCost, - outputCost: totalOutputCost, - models: Object.keys(modelCounts), - }) + if (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) + } + } } // Log the final execution result diff --git a/sim/lib/workflows.ts b/sim/lib/workflows.ts deleted file mode 100644 index e4e39fde8..000000000 --- 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 000000000..c4b845d7f --- /dev/null +++ b/sim/lib/workflows/utils.ts @@ -0,0 +1,35 @@ +import { eq } from 'drizzle-orm' +import { createLogger } from '@/lib/logs/console-logger' +import { db } from '@/db' +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) + return workflows[0] +} + +export async function updateWorkflowRunCounts(workflowId: string, runs: number = 1) { + try { + const workflow = await getWorkflowById(workflowId) + if (!workflow) { + logger.error(`Workflow ${workflowId} not found`) + throw new Error(`Workflow ${workflowId} not found`) + } + + 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() + } 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 43e947e7d..299919e58 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