improvement(db): route additional staleness-tolerant reads to the read replica

Author: waleedlatif1

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

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { db, dbReplica } from '@sim/db'
 import { pausedExecutions, permissions, workflow, workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { and, eq, gte, inArray, isNotNull, isNull, lte, or, type SQL, sql } from 'drizzle-orm'
@@ -60,7 +60,7 @@ export const GET = withRouteHandler(
         wfWhere.push(inArray(workflow.id, wfList))
       }
 
-      const workflows = await db
+      const workflows = await dbReplica
         .select({ id: workflow.id, name: workflow.name })
         .from(workflow)
         .where(and(...wfWhere))
@@ -124,7 +124,7 @@ export const GET = withRouteHandler(
       }
 
       if (isAllTime) {
-        const boundsQuery = db
+        const boundsQuery = dbReplica
           .select({
             minDate: sql<Date>`MIN(${workflowExecutionLogs.startedAt})`,
             maxDate: sql<Date>`MAX(${workflowExecutionLogs.startedAt})`,
@@ -168,7 +168,7 @@ export const GET = withRouteHandler(
         lte(workflowExecutionLogs.startedAt, end),
       ]
 
-      const logs = await db
+      const logs = await dbReplica
         .select({
           workflowId: workflowExecutionLogs.workflowId,
           level: workflowExecutionLogs.level,

apps/sim/lib/copilot/chat/process-contents.ts

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { dbReplica } from '@sim/db'
 import { knowledgeBase } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import {
@@ -454,7 +454,7 @@ async function processKnowledgeFromDb(
     if (currentWorkspaceId) {
       conditions.push(eq(knowledgeBase.workspaceId, currentWorkspaceId))
     }
-    const kbRows = await db
+    const kbRows = await dbReplica
       .select({
         id: knowledgeBase.id,
         name: knowledgeBase.name,
@@ -562,8 +562,8 @@ async function processExecutionLogFromDb(
 ): Promise<AgentContext | null> {
   try {
     const { workflowExecutionLogs, workflow } = await import('@sim/db/schema')
-    const { db } = await import('@sim/db')
-    const rows = await db
+    const { dbReplica } = await import('@sim/db')
+    const rows = await dbReplica
       .select({
         id: workflowExecutionLogs.id,
         workflowId: workflowExecutionLogs.workflowId,

apps/sim/lib/copilot/chat/workspace-context.ts

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { dbReplica } from '@sim/db'
 import {
   knowledgeBase,
   knowledgeConnector,
@@ -302,7 +302,7 @@ export async function generateWorkspaceContext(
     ] = await Promise.all([
       getUsersWithPermissions(workspaceId),
 
-      db
+      dbReplica
         .select({
           id: workflow.id,
           name: workflow.name,
@@ -314,7 +314,7 @@ export async function generateWorkspaceContext(
         .from(workflow)
         .where(and(eq(workflow.workspaceId, workspaceId), isNull(workflow.archivedAt))),
 
-      db
+      dbReplica
         .select({
           id: workflowFolder.id,
           name: workflowFolder.name,
@@ -323,7 +323,7 @@ export async function generateWorkspaceContext(
         .from(workflowFolder)
         .where(and(eq(workflowFolder.workspaceId, workspaceId), isNull(workflowFolder.archivedAt))),
 
-      db
+      dbReplica
         .select({
           id: knowledgeBase.id,
           name: knowledgeBase.name,
@@ -332,7 +332,7 @@ export async function generateWorkspaceContext(
         .from(knowledgeBase)
         .where(and(eq(knowledgeBase.workspaceId, workspaceId), isNull(knowledgeBase.deletedAt))),
 
-      db
+      dbReplica
         .select({
           id: userTableDefinitions.id,
           name: userTableDefinitions.name,
@@ -352,7 +352,7 @@ export async function generateWorkspaceContext(
 
       listCustomTools({ userId, workspaceId }),
 
-      db
+      dbReplica
         .select({
           id: mcpServers.id,
           name: mcpServers.name,
@@ -364,7 +364,7 @@ export async function generateWorkspaceContext(
 
       listSkills({ workspaceId, includeBuiltins: false }),
 
-      db
+      dbReplica
         .select({
           id: workflowSchedule.id,
           jobTitle: workflowSchedule.jobTitle,
@@ -388,7 +388,7 @@ export async function generateWorkspaceContext(
       tables.length > 0
         ? await Promise.all(
             tables.map(async (t) => {
-              const [row] = await db
+              const [row] = await dbReplica
                 .select({ count: count() })
                 .from(userTableRows)
                 .where(eq(userTableRows.tableId, t.id))
@@ -400,7 +400,7 @@ export async function generateWorkspaceContext(
     const kbIds = kbs.map((kb) => kb.id)
     const connectorRows =
       kbIds.length > 0
-        ? await db
+        ? await dbReplica
             .select({
               knowledgeBaseId: knowledgeConnector.knowledgeBaseId,
               connectorType: knowledgeConnector.connectorType,

apps/sim/lib/knowledge/chunks/service.ts

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { db, dbReplica } from '@sim/db'
 import { document, embedding, knowledgeBase } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { sha256Hex } from '@sim/security/hash'
@@ -46,7 +46,7 @@ export async function queryChunks(
     conditions.push(ilike(embedding.content, `%${search}%`))
   }
 
-  const chunks = await db
+  const chunks = await dbReplica
     .select({
       id: embedding.id,
       chunkIndex: embedding.chunkIndex,
@@ -82,7 +82,7 @@ export async function queryChunks(
     .limit(limit)
     .offset(offset)
 
-  const totalCount = await db
+  const totalCount = await dbReplica
     .select({ count: sql`count(*)` })
     .from(embedding)
     .where(and(...conditions))

apps/sim/lib/knowledge/tags/service.ts

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { db, dbReplica } from '@sim/db'
 import { document, embedding, knowledgeBaseTagDefinitions } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { generateId } from '@sim/utils/id'
@@ -99,7 +99,7 @@ export async function getNextAvailableSlot(
 export async function getDocumentTagDefinitions(
   knowledgeBaseId: string
 ): Promise<DocumentTagDefinition[]> {
-  const definitions = await db
+  const definitions = await dbReplica
     .select({
       id: knowledgeBaseTagDefinitions.id,
       knowledgeBaseId: knowledgeBaseTagDefinitions.knowledgeBaseId,
@@ -123,7 +123,7 @@ export async function getDocumentTagDefinitions(
  * Get all tag definitions for a knowledge base (alias for compatibility)
  */
 export async function getTagDefinitions(knowledgeBaseId: string): Promise<TagDefinition[]> {
-  const tagDefinitions = await db
+  const tagDefinitions = await dbReplica
     .select({
       id: knowledgeBaseTagDefinitions.id,
       tagSlot: knowledgeBaseTagDefinitions.tagSlot,
@@ -655,7 +655,7 @@ export async function getTagUsage(
       whereConditions.push(sql`${sql.raw(tagSlot)} != ''`)
     }
 
-    const documentsWithTag = await db
+    const documentsWithTag = await dbReplica
       .select({
         id: document.id,
         filename: document.filename,
@@ -703,7 +703,7 @@ export async function getTagUsageStats(
     const tagSlot = def.tagSlot
     validateTagSlot(tagSlot)
 
-    const docCountResult = await db
+    const docCountResult = await dbReplica
       .select({ count: sql<number>`count(*)` })
       .from(document)
       .where(
@@ -716,7 +716,7 @@ export async function getTagUsageStats(
         )
       )
 
-    const chunkCountResult = await db
+    const chunkCountResult = await dbReplica
       .select({ count: sql<number>`count(*)` })
       .from(embedding)
       .innerJoin(document, eq(embedding.documentId, document.id))

apps/sim/lib/workspace-events/no-activity.ts

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { dbReplica } from '@sim/db'
 import { webhook, workflow, workflowDeploymentVersion, workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { and, asc, eq, gt, gte, inArray, isNull, ne, or, sql } from 'drizzle-orm'
@@ -42,7 +42,7 @@ export interface NoActivityPollResult {
 async function fetchNoActivitySubscriptionPage(
   afterWebhookId: string | null
 ): Promise<SimSubscription[]> {
-  const rows = await db
+  const rows = await dbReplica
     .select({ webhook, workflow })
     .from(webhook)
     .innerJoin(workflow, eq(webhook.workflowId, workflow.id))
@@ -105,7 +105,7 @@ async function fetchWatchedWorkflowPage(
     conditions.push(gt(workflow.id, afterWorkflowId))
   }
 
-  return db
+  return dbReplica
     .select({ id: workflow.id, name: workflow.name })
     .from(workflow)
     .where(and(...conditions))
@@ -120,7 +120,7 @@ async function hasRecentActivity(
 ): Promise<boolean> {
   const windowStart = new Date(Date.now() - config.inactivityHours * 60 * 60 * 1000)
 
-  const recentLogs = await db
+  const recentLogs = await dbReplica
     .select({ id: workflowExecutionLogs.id })
     .from(workflowExecutionLogs)
     .where(

apps/sim/lib/workspace-events/rules.ts

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { dbReplica } from '@sim/db'
 import { workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { and, avg, count, desc, eq, gte, ne, type SQL, sql } from 'drizzle-orm'
@@ -22,7 +22,7 @@ export function excludeSimExecutionsCondition(): SQL {
 }
 
 async function checkConsecutiveFailures(workflowId: string, threshold: number): Promise<boolean> {
-  const recentLogs = await db
+  const recentLogs = await dbReplica
     .select({ level: workflowExecutionLogs.level })
     .from(workflowExecutionLogs)
     .where(and(eq(workflowExecutionLogs.workflowId, workflowId), excludeSimExecutionsCondition()))
@@ -51,7 +51,7 @@ async function checkFailureRate(
 
   // Single DB-side aggregate: the window is user-configured and this runs on
   // the execution-completion path, so never materialize the in-window rows.
-  const result = await db
+  const result = await dbReplica
     .select({
       total: count(),
       errors: count(sql`case when ${workflowExecutionLogs.level} = 'error' then 1 end`),
@@ -82,7 +82,7 @@ async function checkLatencySpike(
 ): Promise<boolean> {
   const windowStart = new Date(Date.now() - windowHours * 60 * 60 * 1000)
 
-  const result = await db
+  const result = await dbReplica
     .select({
       avgDuration: avg(workflowExecutionLogs.totalDurationMs),
       count: count(),
@@ -114,7 +114,7 @@ async function checkErrorCount(
 ): Promise<boolean> {
   const windowStart = new Date(Date.now() - windowHours * 60 * 60 * 1000)
 
-  const result = await db
+  const result = await dbReplica
     .select({ count: count() })
     .from(workflowExecutionLogs)
     .where(