improvement(billing): duplicate checks for bypasses, logger billing actor consistency, run from block (#3107)

* improvement(billing): improve against direct subscription creation bypasses

* more usage of block/unblock helpers

* address bugbot comments

* fail closed

* only run dup check for orgs

apps/sim/app/api/organizations/[id]/invitations/[invitationId]/route.ts

@@ -20,6 +20,7 @@ import { z } from 'zod'
 import { getEmailSubject, renderInvitationEmail } from '@/components/emails'
 import { getSession } from '@/lib/auth'
 import { hasAccessControlAccess } from '@/lib/billing'
+import { syncUsageLimitsFromSubscription } from '@/lib/billing/core/usage'
 import { requireStripeClient } from '@/lib/billing/stripe-client'
 import { getBaseUrl } from '@/lib/core/utils/urls'
 import { sendEmail } from '@/lib/messaging/email/mailer'
@@ -501,6 +502,18 @@ export async function PUT(
       }
     }
 
+    if (status === 'accepted') {
+      try {
+        await syncUsageLimitsFromSubscription(session.user.id)
+      } catch (syncError) {
+        logger.error('Failed to sync usage limits after joining org', {
+          userId: session.user.id,
+          organizationId,
+          error: syncError,
+        })
+      }
+    }
+
     logger.info(`Organization invitation ${status}`, {
       organizationId,
       invitationId,

apps/sim/app/api/users/me/subscription/[id]/transfer/route.ts

@@ -5,6 +5,7 @@ import { and, eq } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { z } from 'zod'
 import { getSession } from '@/lib/auth'
+import { hasActiveSubscription } from '@/lib/billing'
 
 const logger = createLogger('SubscriptionTransferAPI')
 
@@ -88,6 +89,14 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
       )
     }
 
+    // Check if org already has an active subscription (prevent duplicates)
+    if (await hasActiveSubscription(organizationId)) {
+      return NextResponse.json(
+        { error: 'Organization already has an active subscription' },
+        { status: 409 }
+      )
+    }
+
     await db
       .update(subscription)
       .set({ referenceId: organizationId })

apps/sim/app/api/v1/admin/users/[id]/billing/route.ts

@@ -203,6 +203,10 @@ export const PATCH = withAdminAuthParams<RouteParams>(async (request, context) =
       }
 
       updateData.billingBlocked = body.billingBlocked
+      // Clear the reason when unblocking
+      if (body.billingBlocked === false) {
+        updateData.billingBlockedReason = null
+      }
       updated.push('billingBlocked')
     }
 

apps/sim/app/api/workflows/[id]/execute-from-block/route.ts

@@ -1,6 +1,4 @@
-import { db, workflow as workflowTable } from '@sim/db'
 import { createLogger } from '@sim/logger'
-import { eq } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { v4 as uuidv4 } from 'uuid'
 import { z } from 'zod'
@@ -8,6 +6,7 @@ import { checkHybridAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { SSE_HEADERS } from '@/lib/core/utils/sse'
 import { markExecutionCancelled } from '@/lib/execution/cancellation'
+import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
 import { createSSECallbacks } from '@/lib/workflows/executor/execution-events'
@@ -75,12 +74,31 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
     const { startBlockId, sourceSnapshot, input } = validation.data
     const executionId = uuidv4()
 
-    const [workflowRecord] = await db
-      .select({ workspaceId: workflowTable.workspaceId, userId: workflowTable.userId })
-      .from(workflowTable)
-      .where(eq(workflowTable.id, workflowId))
-      .limit(1)
+    // Run preprocessing checks (billing, rate limits, usage limits)
+    const preprocessResult = await preprocessExecution({
+      workflowId,
+      userId,
+      triggerType: 'manual',
+      executionId,
+      requestId,
+      checkRateLimit: false, // Manual executions don't rate limit
+      checkDeployment: false, // Run-from-block doesn't require deployment
+    })
 
+    if (!preprocessResult.success) {
+      const { error } = preprocessResult
+      logger.warn(`[${requestId}] Preprocessing failed for run-from-block`, {
+        workflowId,
+        error: error?.message,
+        statusCode: error?.statusCode,
+      })
+      return NextResponse.json(
+        { error: error?.message || 'Execution blocked' },
+        { status: error?.statusCode || 500 }
+      )
+    }
+
+    const workflowRecord = preprocessResult.workflowRecord
     if (!workflowRecord?.workspaceId) {
       return NextResponse.json({ error: 'Workflow not found or has no workspace' }, { status: 404 })
     }
@@ -92,6 +110,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
       workflowId,
       startBlockId,
       executedBlocksCount: sourceSnapshot.executedBlocks.length,
+      billingActorUserId: preprocessResult.actorUserId,
     })
 
     const loggingSession = new LoggingSession(workflowId, executionId, 'manual', requestId)