improvement(tables): extract TablesDetail wrapper, ship trigger followups

apps/realtime/src/handlers/index.ts

@@ -2,7 +2,6 @@ import { setupConnectionHandlers } from '@/handlers/connection'
 import { setupOperationsHandlers } from '@/handlers/operations'
 import { setupPresenceHandlers } from '@/handlers/presence'
 import { setupSubblocksHandlers } from '@/handlers/subblocks'
-import { setupTableHandlers } from '@/handlers/tables'
 import { setupVariablesHandlers } from '@/handlers/variables'
 import { setupWorkflowHandlers } from '@/handlers/workflow'
 import type { AuthenticatedSocket } from '@/middleware/auth'
@@ -14,6 +13,5 @@ export function setupAllHandlers(socket: AuthenticatedSocket, roomManager: IRoom
   setupSubblocksHandlers(socket, roomManager)
   setupVariablesHandlers(socket, roomManager)
   setupPresenceHandlers(socket, roomManager)
-  setupTableHandlers(socket, roomManager)
   setupConnectionHandlers(socket, roomManager)
 }

apps/realtime/src/middleware/permissions.ts

@@ -131,51 +131,3 @@ export async function verifyWorkflowAccess(
     return { hasAccess: false }
   }
 }
-
-/**
- * Verify a user has read access to a table by virtue of workspace permission.
- * Mirrors `verifyWorkflowAccess` for the table-room socket join check.
- */
-export async function verifyTableAccess(
-  userId: string,
-  tableId: string
-): Promise<{ hasAccess: boolean; workspaceId?: string }> {
-  try {
-    const { userTableDefinitions, permissions } = await import('@sim/db')
-    const tableData = await db
-      .select({ workspaceId: userTableDefinitions.workspaceId })
-      .from(userTableDefinitions)
-      .where(and(eq(userTableDefinitions.id, tableId), isNull(userTableDefinitions.archivedAt)))
-      .limit(1)
-
-    if (!tableData.length) {
-      logger.warn(`Table ${tableId} not found`)
-      return { hasAccess: false }
-    }
-    const { workspaceId } = tableData[0]
-    if (!workspaceId) return { hasAccess: false }
-
-    const [permissionRow] = await db
-      .select({ permissionType: permissions.permissionType })
-      .from(permissions)
-      .where(
-        and(
-          eq(permissions.userId, userId),
-          eq(permissions.entityType, 'workspace'),
-          eq(permissions.entityId, workspaceId)
-        )
-      )
-      .limit(1)
-
-    if (!permissionRow?.permissionType) {
-      logger.warn(
-        `User ${userId} has no permission for workspace ${workspaceId} (table ${tableId})`
-      )
-      return { hasAccess: false }
-    }
-    return { hasAccess: true, workspaceId }
-  } catch (error) {
-    logger.error(`Error verifying table access for user ${userId}, table ${tableId}:`, error)
-    return { hasAccess: false }
-  }
-}

apps/realtime/src/rooms/memory-manager.ts

@@ -1,13 +1,6 @@
 import { createLogger } from '@sim/logger'
 import type { Server } from 'socket.io'
-import {
-  type IRoomManager,
-  type TableRowUpdatedPayload,
-  tableRoomName,
-  type UserPresence,
-  type UserSession,
-  type WorkflowRoom,
-} from '@/rooms/types'
+import type { IRoomManager, UserPresence, UserSession, WorkflowRoom } from '@/rooms/types'
 
 const logger = createLogger('MemoryRoomManager')
 
@@ -262,23 +255,4 @@ export class MemoryRoomManager implements IRoomManager {
 
     logger.info(`Notified ${room.users.size} users about workflow deployment change: ${workflowId}`)
   }
-
-  emitToTable<T = unknown>(tableId: string, event: string, payload: T): void {
-    this._io.to(tableRoomName(tableId)).emit(event, payload)
-  }
-
-  async handleTableRowUpdated(tableId: string, payload: TableRowUpdatedPayload): Promise<void> {
-    this.emitToTable(tableId, 'table-row-updated', { tableId, ...payload })
-  }
-
-  async handleTableRowDeleted(tableId: string, rowId: string): Promise<void> {
-    this.emitToTable(tableId, 'table-row-deleted', { tableId, rowId })
-  }
-
-  async handleTableDeleted(tableId: string): Promise<void> {
-    logger.info(`Handling table deletion notification for ${tableId}`)
-    this.emitToTable(tableId, 'table-deleted', { tableId, timestamp: Date.now() })
-    // Eject sockets so they don't hold a stale room. Cross-pod safe via socket.io.
-    await this._io.in(tableRoomName(tableId)).socketsLeave(tableRoomName(tableId))
-  }
 }

apps/realtime/src/rooms/redis-manager.ts

@@ -1,13 +1,7 @@
 import { createLogger } from '@sim/logger'
 import { createClient, type RedisClientType } from 'redis'
 import type { Server } from 'socket.io'
-import {
-  type IRoomManager,
-  type TableRowUpdatedPayload,
-  tableRoomName,
-  type UserPresence,
-  type UserSession,
-} from '@/rooms/types'
+import type { IRoomManager, UserPresence, UserSession } from '@/rooms/types'
 
 const logger = createLogger('RedisRoomManager')
 
@@ -463,23 +457,4 @@ export class RedisRoomManager implements IRoomManager {
     const userCount = await this.getUniqueUserCount(workflowId)
     logger.info(`Notified ${userCount} users about workflow deployment change: ${workflowId}`)
   }
-
-  emitToTable<T = unknown>(tableId: string, event: string, payload: T): void {
-    this._io.to(tableRoomName(tableId)).emit(event, payload)
-  }
-
-  async handleTableRowUpdated(tableId: string, payload: TableRowUpdatedPayload): Promise<void> {
-    this.emitToTable(tableId, 'table-row-updated', { tableId, ...payload })
-  }
-
-  async handleTableRowDeleted(tableId: string, rowId: string): Promise<void> {
-    this.emitToTable(tableId, 'table-row-deleted', { tableId, rowId })
-  }
-
-  async handleTableDeleted(tableId: string): Promise<void> {
-    logger.info(`Handling table deletion notification for ${tableId}`)
-    this.emitToTable(tableId, 'table-deleted', { tableId, timestamp: Date.now() })
-    // Eject sockets across all pods via socket.io's Redis adapter.
-    await this._io.in(tableRoomName(tableId)).socketsLeave(tableRoomName(tableId))
-  }
 }

apps/realtime/src/rooms/types.ts

@@ -143,45 +143,4 @@ export interface IRoomManager {
    * Handle workflow deployment change - notify users to refresh deployment state
    */
   handleWorkflowDeployed(workflowId: string): Promise<void>
-
-  /**
-   * Emit an event to all clients in a table room (`table:${tableId}`).
-   * Tables don't track presence/last-modified state — just pub/sub.
-   */
-  emitToTable<T = unknown>(tableId: string, event: string, payload: T): void
-
-  /**
-   * Notify all clients in a table room of a row write (insert/update/cell-state-change).
-   * Sim API calls this via the `/api/table-row-updated` HTTP bridge after every successful
-   * row commit; the client merges the delta into its React Query cache.
-   */
-  handleTableRowUpdated(tableId: string, payload: TableRowUpdatedPayload): Promise<void>
-
-  /**
-   * Notify all clients in a table room that a row has been deleted.
-   */
-  handleTableRowDeleted(tableId: string, rowId: string): Promise<void>
-
-  /**
-   * Notify all clients in a table room that the table has been deleted; eject sockets.
-   */
-  handleTableDeleted(tableId: string): Promise<void>
-}
-
-/**
- * Payload broadcast on `table-row-updated`. Mirrors the shape of `TableRow.data` so
- * the client can merge directly into its React Query rows cache. `position` and
- * `updatedAt` are included for cache reconciliation; `data` is the full row data
- * (not a per-cell delta) — see plan Notes.
- */
-export interface TableRowUpdatedPayload {
-  rowId: string
-  data: Record<string, unknown>
-  /** Per-workflow-group execution state. Keyed by `WorkflowGroup.id`. */
-  executions?: Record<string, unknown>
-  position: number
-  updatedAt: string | number
 }
-
-/** Socket.IO room name for a table. Namespaced from workflow rooms. */
-export const tableRoomName = (tableId: string): string => `table:${tableId}`

apps/realtime/src/routes/http.ts

@@ -150,52 +150,6 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
       return
     }
 
-    // Handle table row write notifications from the Sim API
-    if (req.method === 'POST' && req.url === '/api/table-row-updated') {
-      try {
-        const body = await readRequestBody(req)
-        const { tableId, rowId, data, executions, position, updatedAt } = JSON.parse(body)
-        await roomManager.handleTableRowUpdated(tableId, {
-          rowId,
-          data,
-          executions,
-          position,
-          updatedAt,
-        })
-        sendSuccess(res)
-      } catch (error) {
-        logger.error('Error handling table row update notification:', error)
-        sendError(res, 'Failed to process table row update')
-      }
-      return
-    }
-
-    if (req.method === 'POST' && req.url === '/api/table-row-deleted') {
-      try {
-        const body = await readRequestBody(req)
-        const { tableId, rowId } = JSON.parse(body)
-        await roomManager.handleTableRowDeleted(tableId, rowId)
-        sendSuccess(res)
-      } catch (error) {
-        logger.error('Error handling table row deletion notification:', error)
-        sendError(res, 'Failed to process table row deletion')
-      }
-      return
-    }
-
-    if (req.method === 'POST' && req.url === '/api/table-deleted') {
-      try {
-        const body = await readRequestBody(req)
-        const { tableId } = JSON.parse(body)
-        await roomManager.handleTableDeleted(tableId)
-        sendSuccess(res)
-      } catch (error) {
-        logger.error('Error handling table deletion notification:', error)
-        sendError(res, 'Failed to process table deletion')
-      }
-      return
-    }
-
     res.writeHead(404, { 'Content-Type': 'application/json' })
     res.end(JSON.stringify({ error: 'Not found' }))
   }

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

@@ -0,0 +1,48 @@
+import { createLogger } from '@sim/logger'
+import { type NextRequest, NextResponse } from 'next/server'
+import { runColumnContract } from '@/lib/api/contracts/tables'
+import { parseRequest } from '@/lib/api/server'
+import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid'
+import { generateRequestId } from '@/lib/core/utils/request'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { runWorkflowColumn } from '@/lib/table/workflow-columns'
+import { accessError, checkAccess } from '@/app/api/table/utils'
+
+const logger = createLogger('TableRunColumnAPI')
+
+interface RouteParams {
+  params: Promise<{ tableId: string }>
+}
+
+/** POST /api/table/[tableId]/columns/run */
+export const POST = withRouteHandler(async (request: NextRequest, { params }: RouteParams) => {
+  const requestId = generateRequestId()
+  try {
+    const auth = await checkSessionOrInternalAuth(request, { requireWorkflowId: false })
+    if (!auth.success || !auth.userId) {
+      return NextResponse.json({ error: 'Authentication required' }, { status: 401 })
+    }
+    const parsed = await parseRequest(runColumnContract, request, { params })
+    if (!parsed.success) return parsed.response
+    const { tableId } = parsed.data.params
+    const { workspaceId, groupIds, runMode, rowIds } = parsed.data.body
+    const access = await checkAccess(tableId, auth.userId, 'write')
+    if (!access.ok) return accessError(access, requestId, tableId)
+
+    const { triggered } = await runWorkflowColumn({
+      tableId,
+      workspaceId,
+      groupIds,
+      mode: runMode,
+      rowIds,
+      requestId,
+    })
+    return NextResponse.json({ success: true, data: { triggered } })
+  } catch (error) {
+    if (error instanceof Error && error.message === 'Invalid workspace ID') {
+      return NextResponse.json({ error: 'Invalid workspace ID' }, { status: 400 })
+    }
+    logger.error(`run-column failed:`, error)
+    return NextResponse.json({ error: 'Failed to run columns' }, { status: 500 })
+  }
+})