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

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

* fix(db): keep event-rule and tag-slot reads on the primary

* fix(db): keep chunk listing and tag-usage counts on the primary

* fix(db): execution-log mention lookup stays on the primary

* fix(db): no-activity decision read stays on the primary

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.test.ts

@@ -7,7 +7,7 @@ import type { ChatContext } from '@/stores/panel'
 
 const { getSkillById } = vi.hoisted(() => ({ getSkillById: vi.fn() }))
 
-vi.mock('@sim/db', () => ({ db: {} }))
+vi.mock('@sim/db', () => ({ db: {}, dbReplica: {} }))
 vi.mock('@sim/db/schema', () => ({ document: {}, knowledgeBase: {} }))
 vi.mock('@/lib/workflows/skills/operations', () => ({ getSkillById }))
 

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

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { db, 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,7 +562,6 @@ 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
       .select({
         id: workflowExecutionLogs.id,

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/workspace-events/no-activity.ts

@@ -1,4 +1,4 @@
-import { db } from '@sim/db'
+import { db, 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))