feat(workspace-vars): add workspace scoped environment + fix cancellation of assoc. workspace invites if org invite cancelled (#1208)

* feat(env-vars): workspace scoped environment variables

* fix cascade delete or workspace invite if org invite with attached workspace invites are created

* remove redundant refetch

* feat(env-vars): workspace scoped environment variables

* fix redirect for invitation error, remove check for validated emails on workspace invitation accept

* styling improvements

* remove random migration code

* stronger typing, added helpers, parallelized envvar encryption

---------

Co-authored-by: waleedlatif1 <walif6@gmail.com>

Author: icecrasher321

apps/sim/app/api/chat/utils.ts

@@ -3,6 +3,7 @@ import { type NextRequest, NextResponse } from 'next/server'
 import { v4 as uuidv4 } from 'uuid'
 import { checkServerSideUsageLimits } from '@/lib/billing'
 import { isDev } from '@/lib/environment'
+import { getPersonalAndWorkspaceEnv } from '@/lib/environment/utils'
 import { createLogger } from '@/lib/logs/console/logger'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
@@ -12,7 +13,7 @@ import { getEmailDomain } from '@/lib/urls/utils'
 import { decryptSecret } from '@/lib/utils'
 import { getBlock } from '@/blocks'
 import { db } from '@/db'
-import { chat, environment as envTable, userStats, workflow } from '@/db/schema'
+import { chat, userStats, workflow } from '@/db/schema'
 import { Executor } from '@/executor'
 import type { BlockLog, ExecutionResult } from '@/executor/types'
 import { Serializer } from '@/serializer'
@@ -453,18 +454,21 @@ export async function executeWorkflowForChat(
     {} as Record<string, Record<string, any>>
   )
 
-  // Get user environment variables for this workflow
+  // Get user environment variables with workspace precedence
   let envVars: Record<string, string> = {}
   try {
-    const envResult = await db
-      .select()
-      .from(envTable)
-      .where(eq(envTable.userId, deployment.userId))
+    const wfWorkspaceRow = await db
+      .select({ workspaceId: workflow.workspaceId })
+      .from(workflow)
+      .where(eq(workflow.id, workflowId))
       .limit(1)
 
-    if (envResult.length > 0 && envResult[0].variables) {
-      envVars = envResult[0].variables as Record<string, string>
-    }
+    const workspaceId = wfWorkspaceRow[0]?.workspaceId || undefined
+    const { personalEncrypted, workspaceEncrypted } = await getPersonalAndWorkspaceEnv(
+      deployment.userId,
+      workspaceId
+    )
+    envVars = { ...personalEncrypted, ...workspaceEncrypted }
   } catch (error) {
     logger.warn(`[${requestId}] Could not fetch environment variables:`, error)
   }

apps/sim/app/api/environment/route.ts

@@ -31,14 +31,12 @@ export async function POST(req: NextRequest) {
       const { variables } = EnvVarSchema.parse(body)
 
       // Encrypt all variables
-      const encryptedVariables = await Object.entries(variables).reduce(
-        async (accPromise, [key, value]) => {
-          const acc = await accPromise
+      const encryptedVariables = await Promise.all(
+        Object.entries(variables).map(async ([key, value]) => {
           const { encrypted } = await encryptSecret(value)
-          return { ...acc, [key]: encrypted }
-        },
-        Promise.resolve({})
-      )
+          return [key, encrypted] as const
+        })
+      ).then((entries) => Object.fromEntries(entries))
 
       // Replace all environment variables for user
       await db

apps/sim/app/api/environment/variables/route.ts

@@ -120,14 +120,12 @@ export async function PUT(request: NextRequest) {
       }
 
       // Only encrypt the variables that are new or changed
-      const newlyEncryptedVariables = await Object.entries(variablesToEncrypt).reduce(
-        async (accPromise, [key, value]) => {
-          const acc = await accPromise
+      const newlyEncryptedVariables = await Promise.all(
+        Object.entries(variablesToEncrypt).map(async ([key, value]) => {
           const { encrypted } = await encryptSecret(value)
-          return { ...acc, [key]: encrypted }
-        },
-        Promise.resolve({})
-      )
+          return [key, encrypted] as const
+        })
+      ).then((entries) => Object.fromEntries(entries))
 
       // Merge existing encrypted variables with newly encrypted ones
       const finalEncryptedVariables = { ...existingEncryptedVariables, ...newlyEncryptedVariables }

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

@@ -1,5 +1,5 @@
 import { randomUUID } from 'crypto'
-import { and, eq, inArray } from 'drizzle-orm'
+import { and, eq, inArray, isNull } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import {
   getEmailSubject,
@@ -284,6 +284,7 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
     const workspaceInvitationIds: string[] = []
     if (isBatch && validWorkspaceInvitations.length > 0) {
       for (const email of emailsToInvite) {
+        const orgInviteForEmail = invitationsToCreate.find((inv) => inv.email === email)
         for (const wsInvitation of validWorkspaceInvitations) {
           const wsInvitationId = randomUUID()
           const token = randomUUID()
@@ -297,6 +298,7 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
             status: 'pending',
             token,
             permissions: wsInvitation.permission,
+            orgInvitationId: orgInviteForEmail?.id,
             expiresAt,
             createdAt: new Date(),
             updatedAt: new Date(),
@@ -467,9 +469,7 @@ export async function DELETE(
     // Cancel the invitation
     const result = await db
       .update(invitation)
-      .set({
-        status: 'cancelled',
-      })
+      .set({ status: 'cancelled' })
       .where(
         and(
           eq(invitation.id, invitationId),
@@ -486,6 +486,26 @@ export async function DELETE(
       )
     }
 
+    // Also cancel any linked workspace invitations created as part of the batch
+    await db
+      .update(workspaceInvitation)
+      .set({ status: 'cancelled' })
+      .where(eq(workspaceInvitation.orgInvitationId, invitationId))
+
+    // Legacy fallback: cancel any pending workspace invitations for the same email
+    // that do not have an orgInvitationId and were created by the same inviter
+    await db
+      .update(workspaceInvitation)
+      .set({ status: 'cancelled' })
+      .where(
+        and(
+          isNull(workspaceInvitation.orgInvitationId),
+          eq(workspaceInvitation.email, result[0].email),
+          eq(workspaceInvitation.status, 'pending'),
+          eq(workspaceInvitation.inviterId, session.user.id)
+        )
+      )
+
     logger.info('Organization invitation cancelled', {
       organizationId,
       invitationId,

apps/sim/app/api/organizations/invitations/accept/route.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from 'crypto'
-import { and, eq } from 'drizzle-orm'
+import { and, eq, isNull, or } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { getSession } from '@/lib/auth'
 import { env } from '@/lib/env'
@@ -82,16 +82,6 @@ export async function GET(req: NextRequest) {
       )
     }
 
-    // Check if user's email is verified
-    if (!userData[0].emailVerified) {
-      return NextResponse.redirect(
-        new URL(
-          `/invite/invite-error?reason=email-not-verified&details=${encodeURIComponent(`You must verify your email address (${userData[0].email}) before accepting invitations.`)}`,
-          env.NEXT_PUBLIC_APP_URL || 'https://sim.ai'
-        )
-      )
-    }
-
     // Verify the email matches the current user
     if (orgInvitation.email !== session.user.email) {
       return NextResponse.redirect(
@@ -137,14 +127,21 @@ export async function GET(req: NextRequest) {
       // Mark organization invitation as accepted
       await tx.update(invitation).set({ status: 'accepted' }).where(eq(invitation.id, invitationId))
 
-      // Find and accept any pending workspace invitations for the same email
+      // Find and accept any pending workspace invitations linked to this org invite.
+      // For backward compatibility, also include legacy pending invites by email with no org link.
       const workspaceInvitations = await tx
         .select()
         .from(workspaceInvitation)
         .where(
           and(
-            eq(workspaceInvitation.email, orgInvitation.email),
-            eq(workspaceInvitation.status, 'pending')
+            eq(workspaceInvitation.status, 'pending'),
+            or(
+              eq(workspaceInvitation.orgInvitationId, invitationId),
+              and(
+                isNull(workspaceInvitation.orgInvitationId),
+                eq(workspaceInvitation.email, orgInvitation.email)
+              )
+            )
           )
         )
 
@@ -264,17 +261,6 @@ export async function POST(req: NextRequest) {
       return NextResponse.json({ error: 'User not found' }, { status: 404 })
     }
 
-    // Check if user's email is verified
-    if (!userData[0].emailVerified) {
-      return NextResponse.json(
-        {
-          error: 'Email not verified',
-          message: `You must verify your email address (${userData[0].email}) before accepting invitations.`,
-        },
-        { status: 403 }
-      )
-    }
-
     if (orgInvitation.email !== session.user.email) {
       return NextResponse.json({ error: 'Email mismatch' }, { status: 403 })
     }

apps/sim/app/api/schedules/execute/route.ts

@@ -4,6 +4,7 @@ import { NextResponse } from 'next/server'
 import { v4 as uuidv4 } from 'uuid'
 import { z } from 'zod'
 import { checkServerSideUsageLimits } from '@/lib/billing'
+import { getPersonalAndWorkspaceEnv } from '@/lib/environment/utils'
 import { createLogger } from '@/lib/logs/console/logger'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
@@ -17,13 +18,7 @@ import { decryptSecret } from '@/lib/utils'
 import { loadWorkflowFromNormalizedTables } from '@/lib/workflows/db-helpers'
 import { updateWorkflowRunCounts } from '@/lib/workflows/utils'
 import { db } from '@/db'
-import {
-  environment as environmentTable,
-  subscription,
-  userStats,
-  workflow,
-  workflowSchedule,
-} from '@/db/schema'
+import { subscription, userStats, workflow, workflowSchedule } from '@/db/schema'
 import { Executor } from '@/executor'
 import { Serializer } from '@/serializer'
 import { RateLimiter } from '@/services/queue'
@@ -236,20 +231,15 @@ export async function GET() {
 
               const mergedStates = mergeSubblockState(blocks)
 
-              // Retrieve environment variables for this user (if any).
-              const [userEnv] = await db
-                .select()
-                .from(environmentTable)
-                .where(eq(environmentTable.userId, workflowRecord.userId))
-                .limit(1)
-
-              if (!userEnv) {
-                logger.debug(
-                  `[${requestId}] No environment record found for user ${workflowRecord.userId}. Proceeding with empty variables.`
-                )
-              }
-
-              const variables = EnvVarsSchema.parse(userEnv?.variables ?? {})
+              // Retrieve environment variables with workspace precedence
+              const { personalEncrypted, workspaceEncrypted } = await getPersonalAndWorkspaceEnv(
+                workflowRecord.userId,
+                workflowRecord.workspaceId || undefined
+              )
+              const variables = EnvVarsSchema.parse({
+                ...personalEncrypted,
+                ...workspaceEncrypted,
+              })
 
               const currentBlockStates = await Object.entries(mergedStates).reduce(
                 async (accPromise, [id, block]) => {

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

@@ -5,6 +5,7 @@ import { v4 as uuidv4 } from 'uuid'
 import { z } from 'zod'
 import { getSession } from '@/lib/auth'
 import { checkServerSideUsageLimits } from '@/lib/billing'
+import { getPersonalAndWorkspaceEnv } from '@/lib/environment/utils'
 import { createLogger } from '@/lib/logs/console/logger'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
@@ -18,7 +19,7 @@ import {
 import { validateWorkflowAccess } from '@/app/api/workflows/middleware'
 import { createErrorResponse, createSuccessResponse } from '@/app/api/workflows/utils'
 import { db } from '@/db'
-import { environment as environmentTable, subscription, userStats } from '@/db/schema'
+import { subscription, userStats } from '@/db/schema'
 import { Executor } from '@/executor'
 import { Serializer } from '@/serializer'
 import {
@@ -64,7 +65,12 @@ class UsageLimitError extends Error {
   }
 }
 
-async function executeWorkflow(workflow: any, requestId: string, input?: any): Promise<any> {
+async function executeWorkflow(
+  workflow: any,
+  requestId: string,
+  input?: any,
+  executingUserId?: string
+): Promise<any> {
   const workflowId = workflow.id
   const executionId = uuidv4()
 
@@ -127,23 +133,15 @@ async function executeWorkflow(workflow: any, requestId: string, input?: any): P
     // Use the same execution flow as in scheduled executions
     const mergedStates = mergeSubblockState(blocks)
 
-    // Fetch the user's environment variables (if any)
-    const [userEnv] = await db
-      .select()
-      .from(environmentTable)
-      .where(eq(environmentTable.userId, workflow.userId))
-      .limit(1)
-
-    if (!userEnv) {
-      logger.debug(
-        `[${requestId}] No environment record found for user ${workflow.userId}. Proceeding with empty variables.`
-      )
-    }
-
-    const variables = EnvVarsSchema.parse(userEnv?.variables ?? {})
+    // Load personal (for the executing user) and workspace env (workspace overrides personal)
+    const { personalEncrypted, workspaceEncrypted } = await getPersonalAndWorkspaceEnv(
+      executingUserId || workflow.userId,
+      workflow.workspaceId || undefined
+    )
+    const variables = EnvVarsSchema.parse({ ...personalEncrypted, ...workspaceEncrypted })
 
     await loggingSession.safeStart({
-      userId: workflow.userId,
+      userId: executingUserId || workflow.userId,
       workspaceId: workflow.workspaceId,
       variables,
     })
@@ -400,7 +398,13 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
         }
       }
 
-      const result = await executeWorkflow(validation.workflow, requestId, undefined)
+      const result = await executeWorkflow(
+        validation.workflow,
+        requestId,
+        undefined,
+        // Executing user (manual run): if session present, use that user for fallback
+        (await getSession())?.user?.id || undefined
+      )
 
       // Check if the workflow execution contains a response block output
       const hasResponseBlock = workflowHasResponseBlock(result)
@@ -589,7 +593,12 @@ export async function POST(
         )
       }
 
-      const result = await executeWorkflow(validation.workflow, requestId, input)
+      const result = await executeWorkflow(
+        validation.workflow,
+        requestId,
+        input,
+        authenticatedUserId
+      )
 
       const hasResponseBlock = workflowHasResponseBlock(result)
       if (hasResponseBlock) {