refactor(projects): prepare compatible writers for the SQL backfill

Author: mzxchandra

apps/sim/lib/billing/organizations/membership.ts

@@ -557,7 +557,7 @@ async function reassignOwnedOrganizationResourcesTx({
   const ownerId = ownerMembership?.userId
   if (!ownerId || ownerId === userId) return 0
 
-  await lockProjectBackfillWrites(tx)
+  await lockProjectBackfillWrites(tx, workspaceIds)
   const ownedProjects = await tx
     .select({ id: project.id })
     .from(project)

apps/sim/lib/core/config/env.ts

@@ -669,7 +669,6 @@ export const env = createEnv({
 
     // SSO Configuration (for script-based registration)
     SSO_ENABLED:                           z.boolean().optional(),                 // Enable SSO functionality
-    PROJECT_WRITES_ENABLED:               z.boolean().optional(),                 // Activate new Project assignments after old writers drain
     PROJECT_API_ENABLED:                  z.boolean().optional(),                 // Expose Projects after backfill and contract enforcement
     SCIM_ENABLED:                          z.boolean().optional(),                 // Enable SCIM directory provisioning
     USAGE_MONITORING_ENABLED:              z.boolean().optional(),                 // Enable organization usage monitoring on self-hosted (bypasses hosted requirements)

apps/sim/lib/projects/README.md

@@ -2,14 +2,14 @@
 
 A Project groups environments. An environment is an existing `workspace` record; there is no separate environment table. Every newly created Project starts with an environment.
 
-Project APIs return HTTP 503 until the deployment enables `PROJECT_API_ENABLED`. Automatic assignment on legacy workspace creation requires `PROJECT_WRITES_ENABLED`; both server controls default off. The API requires writers to be enabled. Existing assigned Projects always retain their lifecycle protections, including fork inheritance and disconnect behavior, even if activation is disabled.
+Project APIs return HTTP 503 until the deployment enables `PROJECT_API_ENABLED`. Workspace creation always assigns a Project atomically. The API control defaults off and does not disable assignment. Existing assigned Projects always retain their lifecycle protections, including fork inheritance and disconnect behavior, even if activation is disabled.
 
 ## Choose the creation flow
 
 | Caller | Flow | Result |
 | --- | --- | --- |
 | New Project onboarding | `POST /api/projects` | Creates a Project and its first environment together, with independently supplied names. |
-| Existing workspace creation UI or caller | `POST /api/workspaces` | Creates a workspace and, when Project writers are enabled, automatically creates its Project, preserving the existing workspace response. |
+| Existing workspace creation UI or caller | `POST /api/workspaces` | Creates a workspace and automatically creates its Project, preserving the existing workspace response. |
 | Create another environment by forking | Existing workspace fork operation | Inherits the source workspace's Project; unassigned legacy families remain unassigned until backfilled. |
 
 Both POST endpoints are internal, session-authenticated APIs. `POST /api/projects` is not a public `/api/v2` endpoint and does not accept API-key principals. This foundation does not remove or deprecate existing workspace creation endpoints.
@@ -69,7 +69,7 @@ Existing callers can continue to use `createWorkspaceContract` and `POST /api/wo
 
 `skipDefaultWorkflow` remains optional and defaults to `false`. The route uses the session's active organization and existing workspace creation policy to resolve ownership and billing. It does not take the explicit Project scope or independent Project name used by `POST /api/projects`.
 
-When Project writers are enabled, the workspace and its Project are created atomically. The generated Project name is `Support workspace - Project`; long names are bounded to 100 characters while retaining the suffix. The response remains `{ "workspace": ... }` with HTTP 200, without a new Project response wrapper. Call `GET /api/projects/by-workspace/[workspaceId]` when an existing workspace caller needs its authorized Project details.
+The workspace and its Project are created atomically. The generated Project name is `Support workspace - Project`; long names are bounded to 100 characters while retaining the suffix. The response remains `{ "workspace": ... }` with HTTP 200, without a new Project response wrapper. Call `GET /api/projects/by-workspace/[workspaceId]` when an existing workspace caller needs its authorized Project details.
 
 ## Server implementation
 

apps/sim/lib/projects/__integration__/foundation.integration.ts

@@ -55,7 +55,6 @@ import { createFork } from '@/ee/workspace-forking/lib/create-fork'
 import { unlinkForkEdge } from '@/ee/workspace-forking/lib/lineage/unlink'
 
 beforeEach(() => {
-  vi.stubEnv('PROJECT_WRITES_ENABLED', 'true')
   vi.stubEnv('PROJECT_API_ENABLED', 'true')
 })
 
@@ -194,9 +193,8 @@ afterAll(async () => {
 
 describe('Project foundation at the database and application boundary', () => {
   check(
-    'disabled rollout preserves legacy creation and fork/disconnect without Project rows',
+    'workspace creation and fork/disconnect assign Projects while APIs remain disabled',
     async () => {
-      vi.stubEnv('PROJECT_WRITES_ENABLED', 'false')
       vi.stubEnv('PROJECT_API_ENABLED', 'false')
       const f = await fixture(false, 1)
       const source = await db.transaction((tx) =>
@@ -214,7 +212,7 @@ describe('Project foundation at the database and application boundary', () => {
       environments.push(source.id)
       expect(
         await db.select().from(projectWorkspace).where(eq(projectWorkspace.workspaceId, source.id))
-      ).toEqual([])
+      ).toHaveLength(1)
       const parent = await getWorkspaceWithOwner(source.id)
       if (!parent) throw new Error('Missing source fixture')
       const fork = await createFork({
@@ -229,11 +227,11 @@ describe('Project foundation at the database and application boundary', () => {
           .select()
           .from(projectWorkspace)
           .where(eq(projectWorkspace.workspaceId, fork.workspace.id))
-      ).toEqual([])
+      ).toHaveLength(1)
       await unlinkForkEdge({ parentWorkspaceId: source.id, childWorkspaceId: fork.workspace.id })
       const [child] = await db.select().from(workspace).where(eq(workspace.id, fork.workspace.id))
       expect(child.forkedFromWorkspaceId).toBeNull()
-      expect(await db.select().from(project).where(eq(project.ownerId, f.ownerId))).toHaveLength(1)
+      expect(await db.select().from(project).where(eq(project.ownerId, f.ownerId))).toHaveLength(3)
     }
   )
 
@@ -242,53 +240,49 @@ describe('Project foundation at the database and application boundary', () => {
     async () => {
       const f = await fixture(false, 1)
       vi.stubEnv('PROJECT_API_ENABLED', 'false')
-      for (const writes of ['false', 'true']) {
-        vi.stubEnv('PROJECT_WRITES_ENABLED', writes)
-        const input = { projectId: f.projectId }
-        const calls = [
-          () =>
-            createProject.execute({
-              principal: f.owner,
-              input: {
-                organizationId: null,
-                name: 'Blocked',
-                initialEnvironment: { name: 'Production' },
-              },
-              request,
-            }),
-          () => getProject.execute({ principal: f.owner, input, request }),
-          () =>
-            getWorkspaceProject.execute({
-              principal: f.owner,
-              input: { workspaceId: f.ids[0] },
-              request,
-            }),
-          () => listProjects.execute({ principal: f.owner, input: { limit: 10 }, request }),
-          () =>
-            renameProject.execute({
-              principal: f.owner,
-              input: { ...input, name: 'Blocked' },
-              request,
-            }),
-          () => archiveProject.execute({ principal: f.owner, input, request }),
-          () => getProjectIssueAccess.execute({ principal: f.owner, input, request }),
-        ]
-        for (const call of calls) await expect(call()).rejects.toMatchObject({ statusCode: 503 })
-        expect(
-          await db.select().from(workspace).where(eq(workspace.ownerId, f.ownerId))
-        ).toHaveLength(1)
-        const [record] = await db.select().from(project).where(eq(project.id, f.projectId))
-        expect(record.name).toBe('Environment 0 - Project')
-        expect(record.archivedAt).toBeNull()
-      }
+      const input = { projectId: f.projectId }
+      const calls = [
+        () =>
+          createProject.execute({
+            principal: f.owner,
+            input: {
+              organizationId: null,
+              name: 'Blocked',
+              initialEnvironment: { name: 'Production' },
+            },
+            request,
+          }),
+        () => getProject.execute({ principal: f.owner, input, request }),
+        () =>
+          getWorkspaceProject.execute({
+            principal: f.owner,
+            input: { workspaceId: f.ids[0] },
+            request,
+          }),
+        () => listProjects.execute({ principal: f.owner, input: { limit: 10 }, request }),
+        () =>
+          renameProject.execute({
+            principal: f.owner,
+            input: { ...input, name: 'Blocked' },
+            request,
+          }),
+        () => archiveProject.execute({ principal: f.owner, input, request }),
+        () => getProjectIssueAccess.execute({ principal: f.owner, input, request }),
+      ]
+      for (const call of calls) await expect(call()).rejects.toMatchObject({ statusCode: 503 })
+      expect(
+        await db.select().from(workspace).where(eq(workspace.ownerId, f.ownerId))
+      ).toHaveLength(1)
+      const [record] = await db.select().from(project).where(eq(project.id, f.projectId))
+      expect(record.name).toBe('Environment 0 - Project')
+      expect(record.archivedAt).toBeNull()
     }
   )
 
   check(
     'disabling activation preserves assigned fork membership and lifecycle protections',
     async () => {
       const f = await fixture(false, 1)
-      vi.stubEnv('PROJECT_WRITES_ENABLED', 'false')
       vi.stubEnv('PROJECT_API_ENABLED', 'false')
       const parent = await getWorkspaceWithOwner(f.ids[0])
       if (!parent) throw new Error('Missing source fixture')
@@ -316,7 +310,7 @@ describe('Project foundation at the database and application boundary', () => {
     }
   )
 
-  check('writer activation assigns new workspaces while Project APIs remain disabled', async () => {
+  check('new workspaces receive Projects while Project APIs remain disabled', async () => {
     vi.stubEnv('PROJECT_API_ENABLED', 'false')
     const f = await fixture(false, 1)
     const created = await db.transaction((tx) =>
@@ -335,26 +329,13 @@ describe('Project foundation at the database and application boundary', () => {
     expect(
       await db.select().from(projectWorkspace).where(eq(projectWorkspace.workspaceId, created.id))
     ).toHaveLength(1)
-    vi.stubEnv('PROJECT_WRITES_ENABLED', 'false')
-    vi.stubEnv('PROJECT_API_ENABLED', 'true')
-    await expect(
-      createProject.execute({
-        principal: f.owner,
-        input: { organizationId: null, name: 'Invalid', initialEnvironment: { name: 'First' } },
-        request,
-      })
-    ).rejects.toThrow('PROJECT_API_ENABLED requires PROJECT_WRITES_ENABLED')
-    expect(await db.select().from(workspace).where(eq(workspace.ownerId, f.ownerId))).toHaveLength(
-      2
-    )
   })
 
   check('legacy fork and disconnect refuse a partially assigned subtree', async () => {
     const f = await fixture(false, 3)
     await db
       .delete(projectWorkspace)
       .where(inArray(projectWorkspace.workspaceId, f.ids.slice(0, 2)))
-    vi.stubEnv('PROJECT_WRITES_ENABLED', 'false')
     vi.stubEnv('PROJECT_API_ENABLED', 'false')
     const parent = await getWorkspaceWithOwner(f.ids[1])
     if (!parent) throw new Error('Missing source fixture')
@@ -395,20 +376,23 @@ describe('Project foundation at the database and application boundary', () => {
       })
       try {
         await Promise.race([read.promise, writer])
-        await expect(
-          db.transaction(async (tx) => {
-            await tx.execute(
-              sql`LOCK TABLE workspace, project, project_workspace IN SHARE ROW EXCLUSIVE MODE NOWAIT`
-            )
-          })
-        ).rejects.toSatisfy((error: unknown) => getPostgresErrorCode(error) === '55P03')
+        await db.transaction(async (tx) => {
+          const [lock] = await tx.execute<{ acquired: boolean }>(sql`
+            SELECT pg_try_advisory_xact_lock(hashtextextended(${`project-backfill:${f.ids[0]}`}, 0)) AS acquired
+          `)
+          expect(lock.acquired).toBe(false)
+          const [unrelated] = await tx.execute<{ acquired: boolean }>(sql`
+            SELECT pg_try_advisory_xact_lock(hashtextextended('project-backfill:unrelated', 0)) AS acquired
+          `)
+          expect(unrelated.acquired).toBe(true)
+        })
       } finally {
         release.resolve()
         await writer
       }
       await db.transaction(async (tx) => {
         await tx.execute(
-          sql`LOCK TABLE workspace, project, project_workspace IN SHARE ROW EXCLUSIVE MODE NOWAIT`
+          sql`SELECT pg_advisory_xact_lock(hashtextextended(${`project-backfill:${f.ids[0]}`}, 0))`
         )
         await createProjectForWorkspace(tx, {
           workspaceId: f.ids[0],
@@ -434,7 +418,7 @@ describe('Project foundation at the database and application boundary', () => {
     const release = createDeferred<void>()
     const backfill = db.transaction(async (tx) => {
       await tx.execute(
-        sql`LOCK TABLE workspace, project, project_workspace IN SHARE ROW EXCLUSIVE MODE NOWAIT`
+        sql`SELECT pg_advisory_xact_lock(hashtextextended(${`project-backfill:${f.ids[0]}`}, 0))`
       )
       locked.resolve()
       await release.promise
@@ -459,7 +443,7 @@ describe('Project foundation at the database and application boundary', () => {
     try {
       for (let attempt = 0; attempt < 100; attempt++) {
         const rows = await db.execute<{ waiting: boolean }>(sql`SELECT EXISTS (
-          SELECT 1 FROM pg_locks WHERE relation = 'workspace'::regclass AND mode = 'RowExclusiveLock' AND NOT granted
+          SELECT 1 FROM pg_locks WHERE locktype = 'advisory' AND mode = 'ShareLock' AND NOT granted
         ) AS waiting`)
         if (rows[0]?.waiting) {
           blocked = true

apps/sim/lib/projects/account-deletion.ts

@@ -10,7 +10,14 @@ export async function prepareProjectsForAccountDeletion(
   userId: string,
   doomedWorkspaceIds: string[]
 ): Promise<void> {
-  await lockProjectBackfillWrites(tx)
+  const ownedEnvironments = await tx
+    .select({ id: workspace.id })
+    .from(workspace)
+    .where(eq(workspace.ownerId, userId))
+  await lockProjectBackfillWrites(tx, [
+    ...doomedWorkspaceIds,
+    ...ownedEnvironments.map((row) => row.id),
+  ])
   const records = await tx
     .select({ id: project.id })
     .from(project)

apps/sim/lib/projects/membership.ts

@@ -5,11 +5,21 @@ import { and, asc, eq, inArray, isNull, ne, notInArray, sql } from 'drizzle-orm'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 
-/** Prevents backfill from assigning membership between an absence read and the ensuing write. */
-export async function lockProjectBackfillWrites(tx: DbTransaction): Promise<void> {
+/** Shared per-environment gate keeps membership absence reads stable during SQL backfill. */
+export async function lockProjectBackfillWrites(
+  tx: DbTransaction,
+  workspaceIds: string[]
+): Promise<void> {
+  if (!workspaceIds.length) return
   await tx.execute(sql`SET LOCAL lock_timeout = '5s'`)
   try {
-    await tx.execute(sql`LOCK TABLE workspace IN ROW EXCLUSIVE MODE`)
+    await tx.execute(sql`
+      SELECT pg_advisory_xact_lock_shared(hashtextextended('project-backfill:' || id, 0))
+      FROM (SELECT DISTINCT unnest(ARRAY[${sql.join(
+        workspaceIds.map((id) => sql`${id}`),
+        sql`, `
+      )}]::text[]) AS id ORDER BY id) ids
+    `)
   } catch (error) {
     if (getPostgresErrorCode(error) === '55P03')
       throw new OrchestrationError('conflict', 'Project backfill is running; retry the operation')
@@ -59,9 +69,9 @@ export async function createProjectForWorkspace(
   return id
 }
 
-/** Returns null only for a legacy workspace awaiting the operator-run backfill. */
+/** Returns null only for a legacy workspace awaiting the SQL backfill. */
 export async function lockWorkspaceProject(tx: DbTransaction, workspaceId: string) {
-  await lockProjectBackfillWrites(tx)
+  await lockProjectBackfillWrites(tx, [workspaceId])
   const [membership] = await tx
     .select()
     .from(projectWorkspace)
@@ -218,7 +228,7 @@ export async function transferWorkspaceProjects(
   ownerId?: string
 ): Promise<void> {
   if (!workspaceIds.length) return
-  await lockProjectBackfillWrites(tx)
+  await lockProjectBackfillWrites(tx, workspaceIds)
   const owners = await tx
     .selectDistinct({ id: projectWorkspace.projectId })
     .from(projectWorkspace)

apps/sim/lib/projects/rollout.server.ts

@@ -1,15 +1,6 @@
 import { envBoolean, getEnv } from '@/lib/core/config/env'
 import { HttpError } from '@/lib/core/utils/http-error'
 
-/** Deployment-wide controls; existing Project lifecycle maintenance never depends on these. */
-export function getProjectRollout() {
-  const writesEnabled = envBoolean(getEnv('PROJECT_WRITES_ENABLED')) ?? false
-  const apiEnabled = envBoolean(getEnv('PROJECT_API_ENABLED')) ?? false
-  if (apiEnabled && !writesEnabled)
-    throw new Error('PROJECT_API_ENABLED requires PROJECT_WRITES_ENABLED')
-  return { writesEnabled, apiEnabled }
-}
-
 class ProjectUnavailableError extends HttpError {
   readonly statusCode = 503
   constructor() {
@@ -18,5 +9,5 @@ class ProjectUnavailableError extends HttpError {
 }
 
 export function requireProjectApiEnabled(): void {
-  if (!getProjectRollout().apiEnabled) throw new ProjectUnavailableError()
+  if (!(envBoolean(getEnv('PROJECT_API_ENABLED')) ?? false)) throw new ProjectUnavailableError()
 }