refactor(tables): centralize row read/order/paginate/position-writes in table-wrapper

Author: TheodoreSpeaks

apps/sim/app/api/table/[tableId]/rows/route.ts

@@ -1,8 +1,5 @@
-import { db } from '@sim/db'
-import { tableRowExecutions, userTableRows } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { and, eq, inArray, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import {
   type BatchInsertTableRowsBodyInput,
@@ -17,27 +14,20 @@ import { isZodError, validationErrorResponse } from '@/lib/api/server/validation
 import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import type {
-  Filter,
-  RowData,
-  RowExecutionMetadata,
-  RowExecutions,
-  Sort,
-  TableSchema,
-} from '@/lib/table'
+import type { Filter, RowData, Sort, TableSchema } from '@/lib/table'
 import {
   batchInsertRows,
   batchUpdateRows,
   deleteRowsByFilter,
   deleteRowsByIds,
   insertRow,
-  USER_TABLE_ROWS_SQL_NAME,
   updateRowsByFilter,
   validateBatchRows,
   validateRowData,
   validateRowSize,
 } from '@/lib/table'
-import { buildFilterClause, buildSortClause, TableQueryValidationError } from '@/lib/table/sql'
+import { TableQueryValidationError } from '@/lib/table/sql'
+import { queryRows } from '@/lib/table/table-wrapper'
 import { accessError, checkAccess } from '@/app/api/table/utils'
 
 const logger = createLogger('TableRowsAPI')
@@ -268,113 +258,35 @@ export const GET = withRouteHandler(
         return NextResponse.json({ error: 'Invalid workspace ID' }, { status: 400 })
       }
 
-      const baseConditions = [
-        eq(userTableRows.tableId, tableId),
-        eq(userTableRows.workspaceId, validated.workspaceId),
-      ]
-
-      const schema = table.schema as TableSchema
-
-      if (validated.filter) {
-        const filterClause = buildFilterClause(
-          validated.filter as Filter,
-          USER_TABLE_ROWS_SQL_NAME,
-          schema.columns
-        )
-        if (filterClause) {
-          baseConditions.push(filterClause)
-        }
-      }
-
-      let query = db
-        .select({
-          id: userTableRows.id,
-          data: userTableRows.data,
-          position: userTableRows.position,
-          createdAt: userTableRows.createdAt,
-          updatedAt: userTableRows.updatedAt,
-        })
-        .from(userTableRows)
-        .where(and(...baseConditions))
-
-      if (validated.sort) {
-        const sortClause = buildSortClause(validated.sort, USER_TABLE_ROWS_SQL_NAME, schema.columns)
-        if (sortClause) {
-          query = query.orderBy(sortClause) as typeof query
-        } else {
-          query = query.orderBy(userTableRows.position) as typeof query
-        }
-      } else {
-        query = query.orderBy(userTableRows.position) as typeof query
-      }
-
-      let totalCount: number | null = null
-      if (validated.includeTotal) {
-        const [{ count }] = await db
-          .select({ count: sql<number>`count(*)` })
-          .from(userTableRows)
-          .where(and(...baseConditions))
-        totalCount = Number(count)
-      }
-
-      const rows = await query.limit(validated.limit).offset(validated.offset)
-
-      // Sidecar: fetch per-(row, group) execution state and group into a map
-      // so the response preserves the legacy `row.executions[groupId]` wire
-      // shape. One indexed-IN scan against table_row_executions.
-      const executionsByRow = new Map<string, RowExecutions>()
-      if (rows.length > 0) {
-        const execRows = await db
-          .select()
-          .from(tableRowExecutions)
-          .where(
-            inArray(
-              tableRowExecutions.rowId,
-              rows.map((r) => r.id)
-            )
-          )
-        for (const e of execRows) {
-          const existing = executionsByRow.get(e.rowId) ?? {}
-          const meta: RowExecutionMetadata = {
-            status: e.status as RowExecutionMetadata['status'],
-            executionId: e.executionId ?? null,
-            jobId: e.jobId ?? null,
-            workflowId: e.workflowId,
-            error: e.error ?? null,
-            ...(e.runningBlockIds && e.runningBlockIds.length > 0
-              ? { runningBlockIds: e.runningBlockIds }
-              : {}),
-            ...(e.blockErrors && Object.keys(e.blockErrors as Record<string, string>).length > 0
-              ? { blockErrors: e.blockErrors as Record<string, string> }
-              : {}),
-            ...(e.cancelledAt ? { cancelledAt: e.cancelledAt.toISOString() } : {}),
-          }
-          existing[e.groupId] = meta
-          executionsByRow.set(e.rowId, existing)
-        }
-      }
+      const result = await queryRows(table, {
+        filter: validated.filter as Filter | undefined,
+        sort: validated.sort,
+        limit: validated.limit,
+        offset: validated.offset,
+        includeTotal: validated.includeTotal,
+      })
 
       logger.info(
-        `[${requestId}] Queried ${rows.length} rows from table ${tableId} (total: ${totalCount ?? 'n/a'})`
+        `[${requestId}] Queried ${result.rowCount} rows from table ${tableId} (total: ${result.totalCount ?? 'n/a'})`
       )
 
       return NextResponse.json({
         success: true,
         data: {
-          rows: rows.map((r) => ({
+          rows: result.rows.map((r) => ({
             id: r.id,
             data: r.data,
-            executions: executionsByRow.get(r.id) ?? {},
+            executions: r.executions,
             position: r.position,
             createdAt:
               r.createdAt instanceof Date ? r.createdAt.toISOString() : String(r.createdAt),
             updatedAt:
               r.updatedAt instanceof Date ? r.updatedAt.toISOString() : String(r.updatedAt),
           })),
-          rowCount: rows.length,
-          totalCount,
-          limit: validated.limit,
-          offset: validated.offset,
+          rowCount: result.rowCount,
+          totalCount: result.totalCount,
+          limit: result.limit,
+          offset: result.offset,
         },
       })
     } catch (error) {

apps/sim/app/api/v1/tables/[tableId]/rows/route.ts

@@ -1,8 +1,5 @@
-import { db } from '@sim/db'
-import { userTableRows } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { and, eq, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import {
   type V1BatchInsertTableRowsBody,
@@ -24,13 +21,13 @@ import {
   deleteRowsByFilter,
   deleteRowsByIds,
   insertRow,
-  USER_TABLE_ROWS_SQL_NAME,
   updateRowsByFilter,
   validateBatchRows,
   validateRowData,
   validateRowSize,
 } from '@/lib/table'
-import { buildFilterClause, buildSortClause, TableQueryValidationError } from '@/lib/table/sql'
+import { TableQueryValidationError } from '@/lib/table/sql'
+import { queryRows } from '@/lib/table/table-wrapper'
 import { accessError, checkAccess } from '@/app/api/table/utils'
 import {
   checkRateLimit,
@@ -153,92 +150,29 @@ export const GET = withRouteHandler(async (request: NextRequest, context: TableR
       return NextResponse.json({ error: 'Invalid workspace ID' }, { status: 400 })
     }
 
-    const baseConditions = [
-      eq(userTableRows.tableId, tableId),
-      eq(userTableRows.workspaceId, validated.workspaceId),
-    ]
-
-    const schema = table.schema as TableSchema
-
-    if (validated.filter) {
-      const filterClause = buildFilterClause(
-        validated.filter as Filter,
-        USER_TABLE_ROWS_SQL_NAME,
-        schema.columns
-      )
-      if (filterClause) {
-        baseConditions.push(filterClause)
-      }
-    }
-
-    let query = db
-      .select({
-        id: userTableRows.id,
-        data: userTableRows.data,
-        position: userTableRows.position,
-        createdAt: userTableRows.createdAt,
-        updatedAt: userTableRows.updatedAt,
-      })
-      .from(userTableRows)
-      .where(and(...baseConditions))
-
-    if (validated.sort) {
-      const sortClause = buildSortClause(validated.sort, USER_TABLE_ROWS_SQL_NAME, schema.columns)
-      if (sortClause) {
-        query = query.orderBy(sortClause) as typeof query
-      } else {
-        query = query.orderBy(userTableRows.position) as typeof query
-      }
-    } else {
-      query = query.orderBy(userTableRows.position) as typeof query
-    }
-
-    const rowsPromise = query.limit(validated.limit).offset(validated.offset)
-
-    let totalCount: number | null = null
-    if (validated.includeTotal) {
-      const countQuery = db
-        .select({ count: sql<number>`count(*)` })
-        .from(userTableRows)
-        .where(and(...baseConditions))
-      const [countResult, rows] = await Promise.all([countQuery, rowsPromise])
-      totalCount = Number(countResult[0].count)
-      return NextResponse.json({
-        success: true,
-        data: {
-          rows: rows.map((r) => ({
-            id: r.id,
-            data: r.data,
-            position: r.position,
-            createdAt:
-              r.createdAt instanceof Date ? r.createdAt.toISOString() : String(r.createdAt),
-            updatedAt:
-              r.updatedAt instanceof Date ? r.updatedAt.toISOString() : String(r.updatedAt),
-          })),
-          rowCount: rows.length,
-          totalCount,
-          limit: validated.limit,
-          offset: validated.offset,
-        },
-      })
-    }
-
-    const rows = await rowsPromise
+    const result = await queryRows(table, {
+      filter: validated.filter as Filter | undefined,
+      sort: validated.sort,
+      limit: validated.limit,
+      offset: validated.offset,
+      includeTotal: validated.includeTotal,
+      withExecutions: false,
+    })
 
     return NextResponse.json({
       success: true,
       data: {
-        rows: rows.map((r) => ({
+        rows: result.rows.map((r) => ({
           id: r.id,
           data: r.data,
           position: r.position,
           createdAt: r.createdAt instanceof Date ? r.createdAt.toISOString() : String(r.createdAt),
           updatedAt: r.updatedAt instanceof Date ? r.updatedAt.toISOString() : String(r.updatedAt),
         })),
-        rowCount: rows.length,
-        totalCount,
-        limit: validated.limit,
-        offset: validated.offset,
+        rowCount: result.rowCount,
+        totalCount: result.totalCount,
+        limit: result.limit,
+        offset: result.offset,
       },
     })
   } catch (error) {

apps/sim/lib/copilot/request/tools/tables.ts

@@ -12,7 +12,9 @@ import { TraceEvent } from '@/lib/copilot/generated/trace-events-v1'
 import { TraceSpan } from '@/lib/copilot/generated/trace-spans-v1'
 import { withCopilotSpan } from '@/lib/copilot/request/otel'
 import type { ExecutionContext, ToolCallResult } from '@/lib/copilot/request/types'
+import type { RowData } from '@/lib/table'
 import { getTableById } from '@/lib/table/service'
+import { buildOrderedRowValues } from '@/lib/table/table-wrapper'
 
 const logger = createLogger('CopilotToolResultTables')
 
@@ -107,16 +109,15 @@ export async function maybeWriteOutputToTable(
               throw new Error('Request aborted before tool mutation could be applied')
             }
             const chunk = rows.slice(i, i + BATCH_CHUNK_SIZE)
-            const values = chunk.map((rowData, j) => ({
-              id: `row_${generateId().replace(/-/g, '')}`,
+            const values = buildOrderedRowValues({
               tableId: outputTable,
               workspaceId: context.workspaceId!,
-              data: rowData,
-              position: i + j,
-              createdAt: now,
-              updatedAt: now,
+              rows: chunk as RowData[],
+              startPosition: i,
+              now,
               createdBy: context.userId,
-            }))
+              makeId: () => `row_${generateId().replace(/-/g, '')}`,
+            })
             await tx.insert(userTableRows).values(values)
           }
         })
@@ -251,16 +252,15 @@ export async function maybeWriteReadCsvToTable(
               throw new Error('Request aborted before tool mutation could be applied')
             }
             const chunk = rows.slice(i, i + BATCH_CHUNK_SIZE)
-            const values = chunk.map((rowData, j) => ({
-              id: `row_${generateId().replace(/-/g, '')}`,
+            const values = buildOrderedRowValues({
               tableId: outputTable,
               workspaceId: context.workspaceId!,
-              data: rowData,
-              position: i + j,
-              createdAt: now,
-              updatedAt: now,
+              rows: chunk as RowData[],
+              startPosition: i,
+              now,
               createdBy: context.userId,
-            }))
+              makeId: () => `row_${generateId().replace(/-/g, '')}`,
+            })
             await tx.insert(userTableRows).values(values)
           }
         })