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

Author: waleedlatif1

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

@@ -99,7 +99,7 @@ export async function getNextAvailableSlot(
 export async function getDocumentTagDefinitions(
   knowledgeBaseId: string
 ): Promise<DocumentTagDefinition[]> {
-  const definitions = await dbReplica
+  const definitions = await db
     .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 dbReplica
+  const tagDefinitions = await db
     .select({
       id: knowledgeBaseTagDefinitions.id,
       tagSlot: knowledgeBaseTagDefinitions.tagSlot,

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

@@ -1,4 +1,4 @@
-import { dbReplica } from '@sim/db'
+import { db } 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 dbReplica
+  const recentLogs = await db
     .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 dbReplica
+  const result = await db
     .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 dbReplica
+  const result = await db
     .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 dbReplica
+  const result = await db
     .select({ count: count() })
     .from(workflowExecutionLogs)
     .where(