feat(projects): move Project membership to the workspace column

apps/sim/app/api/workspaces/[id]/route.ts

@@ -10,6 +10,7 @@ import { getSession } from '@/lib/auth'
 import { changeWorkspaceStoragePayerInTx } from '@/lib/billing/storage/payer-transfer'
 import { captureServerEvent } from '@/lib/posthog/server'
 import { archiveWorkspace } from '@/lib/workspaces/lifecycle'
+import { presentWorkspace } from '@/lib/workspaces/response'
 
 const logger = createLogger('WorkspaceByIdAPI')
 
@@ -51,7 +52,7 @@ export const GET = withRouteHandler(
 
     return NextResponse.json({
       workspace: {
-        ...workspaceDetails,
+        ...presentWorkspace(workspaceDetails),
         permissions: userPermission,
       },
     })
@@ -216,7 +217,7 @@ export const PATCH = withRouteHandler(
 
       return NextResponse.json({
         workspace: {
-          ...updatedWorkspace,
+          ...(updatedWorkspace ? presentWorkspace(updatedWorkspace) : {}),
           permissions: userPermission,
         },
       })

apps/sim/ee/workspace-forking/lib/create-fork.test.ts

@@ -1,4 +1,4 @@
-import { projectWorkspace, workspace } from '@sim/db/schema'
+import { workspace } from '@sim/db/schema'
 import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing'
 import { workflowsPersistenceUtilsMock } from '@sim/testing/mocks/workflows-persistence-utils.mock'
 import {
@@ -140,8 +140,9 @@ function forkParams(selection?: {
 describe('createFork storage headroom gate', () => {
   beforeEach(() => {
     resetDbChainMock()
-    queueTableRows(projectWorkspace, [{ projectId: 'project-source' }])
-    queueTableRows(projectWorkspace, [
+    dbChainMockFns.as.mockReturnValue(workspace)
+    queueTableRows(workspace, [{ projectId: 'project-source' }])
+    queueTableRows(workspace, [
       { project: { id: 'project-source', organizationId: null, archivedAt: null } },
     ])
     /**
@@ -215,8 +216,9 @@ describe('createFork storage headroom gate', () => {
 
   it('refuses when the parent changed organizations after the policy was captured', async () => {
     resetDbChainMock()
-    queueTableRows(projectWorkspace, [{ projectId: 'project-source' }])
-    queueTableRows(projectWorkspace, [
+    dbChainMockFns.as.mockReturnValue(workspace)
+    queueTableRows(workspace, [{ projectId: 'project-source' }])
+    queueTableRows(workspace, [
       { project: { id: 'project-source', organizationId: null, archivedAt: null } },
     ])
     /**
@@ -243,8 +245,9 @@ describe('createFork storage headroom gate', () => {
    */
   it('gives the child the source workspace personal API-key and fork-sync policies', async () => {
     resetDbChainMock()
-    queueTableRows(projectWorkspace, [{ projectId: 'project-source' }])
-    queueTableRows(projectWorkspace, [
+    dbChainMockFns.as.mockReturnValue(workspace)
+    queueTableRows(workspace, [{ projectId: 'project-source' }])
+    queueTableRows(workspace, [
       { project: { id: 'project-source', organizationId: null, archivedAt: null } },
     ])
     queueTableRows(workspace, [{ organizationId: null, forkSyncNewWorkflowsExcluded: true }])

apps/sim/ee/workspace-forking/lib/create-fork.ts

@@ -7,6 +7,7 @@ import { generateId } from '@sim/utils/id'
 import { and, eq } from 'drizzle-orm'
 import type { Workspace } from '@/lib/api/contracts/workspaces'
 import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
+import { getProjectMembershipPhase } from '@/lib/projects/environment-source'
 import { requireForkProject } from '@/lib/projects/membership'
 import { buildDefaultWorkflowArtifacts } from '@/lib/workflows/defaults'
 import { insertNewWorkflowRow } from '@/lib/workflows/persistence/new-workflow-row'
@@ -210,6 +211,7 @@ export async function createFork(params: CreateForkParams): Promise<CreateForkRe
     workflowMcpServers: [],
   }
   const transaction = await db.transaction(async (tx) => {
+    const phase = await getProjectMembershipPhase(tx)
     await setForkLockTimeout(tx)
     if (admission) {
       await lockWorkspaceOperationRequest(tx, admission.workspaceId, admission.requestId)
@@ -303,6 +305,7 @@ export async function createFork(params: CreateForkParams): Promise<CreateForkRe
 
     await tx.insert(workspace).values({
       id: childWorkspaceId,
+      projectId: phase === 'column' ? (parentProject?.id ?? null) : null,
       name: childName,
       ownerId: userId,
       organizationId: policy.organizationId,
@@ -316,10 +319,11 @@ export async function createFork(params: CreateForkParams): Promise<CreateForkRe
       updatedAt: now,
     })
 
-    if (parentProject)
+    if (phase === 'connector' && parentProject) {
       await tx
         .insert(projectWorkspace)
         .values({ projectId: parentProject.id, workspaceId: childWorkspaceId })
+    }
 
     const sourcePermissions = await tx
       .select({ userId: permissions.userId, permissionType: permissions.permissionType })

apps/sim/lib/api/contracts/workspaces.ts

@@ -13,6 +13,7 @@ export type WorkspacePermission = z.output<typeof workspacePermissionSchema>
 export const workspaceSchema = z.object({
   id: z.string(),
   name: z.string(),
+  color: z.string().optional(),
   logoUrl: z.string().nullable().optional(),
   ownerId: z.string(),
   organizationId: z.string().nullable(),
@@ -28,6 +29,17 @@ export const workspaceSchema = z.object({
   isOrgAdmin: z.boolean().optional(),
   billedAccountUserId: z.string().nullable().optional(),
   allowPersonalApiKeys: z.boolean().optional(),
+  storageUsedBytes: z.number().nonnegative().optional(),
+  inboxEnabled: z.boolean().optional(),
+  inboxAddress: z.string().nullable().optional(),
+  inboxProviderId: z.string().nullable().optional(),
+  inboxSecretScope: z.string().optional(),
+  inboxMountedSecrets: z.array(z.string()).optional(),
+  archivedAt: z.string().datetime().nullable().optional(),
+  organizationAssignedAt: z.string().datetime().nullable().optional(),
+  forkSyncNewWorkflowsExcluded: z.boolean().optional(),
+  createdAt: z.string().datetime().optional(),
+  updatedAt: z.string().datetime().optional(),
   inviteMembersEnabled: z.boolean().optional(),
   inviteDisabledReason: z.string().nullable().optional(),
   inviteUpgradeRequired: z.boolean().optional(),

apps/sim/lib/billing/core/usage-log.test.ts

@@ -5,7 +5,7 @@ import {
   billingSubscriptionUtilsMock,
   billingSubscriptionUtilsMockFns,
 } from '@sim/testing/mocks/billing-subscription-utils.mock'
-import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const {
   mockInsert,
@@ -455,10 +455,15 @@ describe('ledger aggregates', () => {
   ]
 
   beforeEach(() => {
+    vi.useFakeTimers({ toFake: ['Date'] })
     installSharedDbMocks()
     vi.spyOn(Date, 'now').mockReturnValue(billingPeriod.start.getTime())
   })
 
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
   for (const aggregate of aggregates) {
     it(`${aggregate.name} reads through the bounded ledger transaction`, async () => {
       const execute = vi.fn().mockResolvedValue([])

apps/sim/lib/billing/organizations/lock-order.test.ts

@@ -11,6 +11,7 @@ import {
   organization,
   outboxEvent,
   permissions,
+  projectMembershipRollout,
   subscription as subscriptionTable,
   user,
   userStats,
@@ -181,11 +182,14 @@ describe('workspace payer-change transaction lock ordering', () => {
   it('locks nonzero workspaces before join billing or aggregate payer changes', async () => {
     const ops: Array<{ op: 'lock' | 'payer-transfer' | 'update'; table: unknown }> = []
     let memberSelectCount = 0
-    const rowsForTable = (table: unknown): unknown[] => {
+    const rowsForTable = (table: unknown, fields?: Record<string, unknown>): unknown[] => {
+      if (table === projectMembershipRollout) return [{ phase: 'column' }]
       if (table === workspace) {
+        if (fields?.id === workspace.projectId) return [{ id: null }]
         return [
           {
             id: 'workspace-1',
+            projectId: null,
             billedAccountUserId: 'user-1',
             organizationId: null,
             storageUsedBytes: 128,
@@ -203,7 +207,7 @@ describe('workspace payer-change transaction lock ordering', () => {
       if (table === userStats) return [{ currentPeriodCost: '5' }]
       return []
     }
-    const select = () => {
+    const select = (fields?: Record<string, unknown>) => {
       let table: unknown
       const chain = {
         from(source: unknown) {
@@ -216,13 +220,16 @@ describe('workspace payer-change transaction lock ordering', () => {
         orderBy() {
           return chain
         },
+        as() {
+          return table
+        },
         for() {
           ops.push({ op: 'lock', table })
           return chain
         },
-        limit: async () => rowsForTable(table),
+        limit: async () => rowsForTable(table, fields),
         then(resolve: (rows: unknown[]) => unknown, reject: (error: unknown) => unknown) {
-          return Promise.resolve(rowsForTable(table)).then(resolve, reject)
+          return Promise.resolve(rowsForTable(table, fields)).then(resolve, reject)
         },
       }
       return chain

apps/sim/lib/billing/webhooks/stripe-sync-convergence.integration.ts

@@ -15,6 +15,7 @@ import {
   type InMemoryStripe,
   stripeClientMock,
 } from '@sim/testing/mocks/stripe.mock'
+import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { type BetterAuthOptions, betterAuth } from 'better-auth'
 import { createAuthMiddleware } from 'better-auth/api'
@@ -378,7 +379,7 @@ type TestTransaction = Parameters<Parameters<typeof testDatabase.transaction>[0]
  * Starts a transaction that takes its locks in `holdLocks`, then parks until released and runs
  * `finish`. `untilBlocking` resolves once another backend is waiting on one of its locks.
  */
-function startParkedTransaction(
+async function startParkedTransaction(
   holdLocks: (tx: TestTransaction) => Promise<void>,
   finish: (tx: TestTransaction) => Promise<void> = async () => {}
 ) {
@@ -399,15 +400,17 @@ function startParkedTransaction(
   })
   async function untilBlocking() {
     const pid = await holderPid
-    for (let attempt = 0; attempt < 200; attempt++) {
+    const deadline = Date.now() + 5000
+    while (Date.now() < deadline) {
       const [row] = await connection<{ blocked: number }[]>`
         select count(*)::int as blocked from pg_stat_activity
         where ${pid}::int = any(pg_blocking_pids(pid))`
       if (row.blocked > 0) return
-      await new Promise<void>((resolve) => setImmediate(resolve))
+      await sleep(10)
     }
     throw new Error('No transaction ever waited on the parked one')
   }
+  await Promise.race([holderPid, done])
   return { done, release, untilBlocking }
 }
 
@@ -1025,7 +1028,7 @@ describe('Team activation', () => {
     })
     stripe.addSubscription({ id: stripeSubscriptionId, customer: `cus_${subscriptionId}` })
 
-    const cancelling = startParkedTransaction(async (tx) => {
+    const cancelling = await startParkedTransaction(async (tx) => {
       await tx
         .update(subscription)
         .set({ cancelAtPeriodEnd: true })
@@ -1045,8 +1048,12 @@ describe('Team activation', () => {
         workspaceIdsToAttach: [],
       })
     )
-    await cancelling.untilBlocking()
-    cancelling.release()
+    try {
+      await cancelling.untilBlocking()
+    } finally {
+      cancelling.release()
+      await Promise.allSettled([cancelling.done, activating])
+    }
     await cancelling.done
     await expect(activating).resolves.toMatchObject({ success: true })
     expect((await storedSubscription(subscriptionId)).cancelAtPeriodEnd).toBe(false)
@@ -1072,7 +1079,7 @@ describe('operator retry', () => {
     )
     await deadLetter(pauseSync)
 
-    const writing = startParkedTransaction(
+    const writing = await startParkedTransaction(
       async (tx) => {
         await tx
           .update(subscription)
@@ -1089,8 +1096,12 @@ describe('operator retry', () => {
       }
     )
     const requeuing = requeueFromAdminApi(pauseSync)
-    await writing.untilBlocking()
-    writing.release()
+    try {
+      await writing.untilBlocking()
+    } finally {
+      writing.release()
+      await Promise.allSettled([writing.done, requeuing])
+    }
 
     await expect(Promise.all([writing.done, requeuing])).resolves.toBeDefined()
     await deliverUnrelatedUpdate(pro.stripeSubscriptionId)
@@ -1113,7 +1124,7 @@ describe('operator retry', () => {
     )
     await deadLetter(cancelSync)
 
-    const writing = startParkedTransaction(
+    const writing = await startParkedTransaction(
       async (tx) => {
         await tx
           .update(subscription)
@@ -1135,8 +1146,12 @@ describe('operator retry', () => {
       timing: 'period_end',
       actor,
     })
-    await writing.untilBlocking()
-    writing.release()
+    try {
+      await writing.untilBlocking()
+    } finally {
+      writing.release()
+      await Promise.allSettled([writing.done, retrying])
+    }
 
     await expect(Promise.all([writing.done, retrying])).resolves.toBeDefined()
     expect((await storedSubscription(org.subscriptionId)).cancelAtPeriodEnd).toBe(true)