fix(tables): serialize explicit-position inserts with advisory lock

The `(table_id, position)` index is non-unique. Concurrent explicit-
position inserts at the same slot can both observe the slot empty, both
skip the shift, then each INSERT — producing duplicate `(table_id,
position)` rows.

Take the per-table advisory lock in the explicit-position branches of
`insertRow` and `batchInsertRows` (the auto-position branches already do
this). Updated the test that asserted the explicit path skipped the lock,
and added coverage for `batchInsertRows` with explicit positions.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: waleedlatif1

apps/sim/lib/table/__tests__/update-row.test.ts

@@ -170,7 +170,7 @@ describe('insertRow — position race safety (migration 0198 + advisory lock)',
     expect(findExecutedSqlContaining('hashtextextended')).toBe(true)
   })
 
-  it('explicit-position inserts skip the advisory lock (no max(position) race to guard)', async () => {
+  it('explicit-position inserts also acquire the advisory lock to serialize position shifts', async () => {
     dbChainMockFns.limit.mockResolvedValueOnce([])
     dbChainMockFns.returning.mockResolvedValueOnce([
       {
@@ -190,7 +190,10 @@ describe('insertRow — position race safety (migration 0198 + advisory lock)',
       'req-1'
     )
 
-    expect(findExecutedSqlContaining('pg_advisory_xact_lock')).toBe(false)
+    // `(table_id, position)` index is non-unique, so concurrent explicit-position
+    // inserts at the same slot could both skip the shift and duplicate — lock
+    // serializes them.
+    expect(findExecutedSqlContaining('pg_advisory_xact_lock')).toBe(true)
   })
 
   it('batchInsertRows acquires the advisory lock (always auto-positioned)', async () => {
@@ -205,6 +208,42 @@ describe('insertRow — position race safety (migration 0198 + advisory lock)',
     expect(findExecutedSqlContaining('pg_advisory_xact_lock')).toBe(true)
   })
 
+  it('batchInsertRows with explicit positions acquires the advisory lock', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce([
+      {
+        id: 'row-1',
+        tableId: 'tbl-1',
+        workspaceId: 'ws-1',
+        data: { name: 'a' },
+        position: 3,
+        createdAt: new Date(),
+        updatedAt: new Date(),
+      },
+      {
+        id: 'row-2',
+        tableId: 'tbl-1',
+        workspaceId: 'ws-1',
+        data: { name: 'b' },
+        position: 4,
+        createdAt: new Date(),
+        updatedAt: new Date(),
+      },
+    ])
+
+    await batchInsertRows(
+      {
+        tableId: 'tbl-1',
+        rows: [{ name: 'a' }, { name: 'b' }],
+        workspaceId: 'ws-1',
+        positions: [3, 4],
+      },
+      TABLE,
+      'req-1'
+    )
+
+    expect(findExecutedSqlContaining('pg_advisory_xact_lock')).toBe(true)
+  })
+
   it('upsertRow skips the advisory lock on the update path (match found)', async () => {
     vi.mocked(getUniqueColumns).mockReturnValue([{ name: 'name', type: 'string', unique: true }])
     dbChainMockFns.limit.mockResolvedValueOnce([

apps/sim/lib/table/service.ts

@@ -651,6 +651,13 @@ export async function insertRow(
 
     let targetPosition: number
 
+    // The `(table_id, position)` index is non-unique, so we serialize all
+    // position-aware writes (explicit and auto) through the per-table
+    // advisory lock. Without this, two concurrent explicit-position inserts
+    // at the same position can both observe an empty slot, both skip the
+    // shift, and each INSERT a row with a duplicate `(table_id, position)`.
+    await acquireTablePositionLock(trx, data.tableId)
+
     if (data.position !== undefined) {
       targetPosition = data.position
 
@@ -674,7 +681,6 @@ export async function insertRow(
           )
       }
     } else {
-      await acquireTablePositionLock(trx, data.tableId)
       const [{ maxPos }] = await trx
         .select({
           maxPos: sql<number>`coalesce(max(${userTableRows.position}), -1)`.mapWith(Number),
@@ -772,6 +778,10 @@ export async function batchInsertRows(
       ...(data.userId ? { createdBy: data.userId } : {}),
     })
 
+    // Serialize position-aware writes per-table. See `acquireTablePositionLock`
+    // for why both explicit- and auto-position paths take this lock.
+    await acquireTablePositionLock(trx, data.tableId)
+
     if (data.positions && data.positions.length > 0) {
       // Position-aware insert: shift existing rows to create gaps, then insert.
       // Process positions ascending so each shift preserves gaps created by prior shifts.
@@ -791,7 +801,6 @@ export async function batchInsertRows(
       return trx.insert(userTableRows).values(rowsToInsert).returning()
     }
 
-    await acquireTablePositionLock(trx, data.tableId)
     const [{ maxPos }] = await trx
       .select({
         maxPos: sql<number>`coalesce(max(${userTableRows.position}), -1)`.mapWith(Number),