fix(db): harden query performance changes

Author: waleedlatif1

apps/sim/app/api/logs/export/route.test.ts

@@ -103,17 +103,44 @@ describe('GET /api/logs/export', () => {
     )
   })
 
-  it('materializes one bounded page at a time while preserving CSV row order', async () => {
-    queueTableRows(workflowExecutionLogs, [logRow(0), logRow(1), logRow(2)])
+  it('rejects unauthenticated exports before checking workspace access', async () => {
+    mockGetSession.mockResolvedValueOnce(null)
+
+    const response = await GET(makeRequest())
+
+    expect(response.status).toBe(401)
+    expect(mockCheckWorkspaceAccess).not.toHaveBeenCalled()
+    expect(dbChainMockFns.where).not.toHaveBeenCalled()
+    expect(mockMaterializeExecutionDataForDisplay).not.toHaveBeenCalled()
+  })
+
+  it('returns only the CSV header when workspace access is denied', async () => {
+    mockCheckWorkspaceAccess.mockResolvedValueOnce({ hasAccess: false })
+
+    const response = await GET(makeRequest())
+
+    expect(response.status).toBe(200)
+    expect(await response.text()).toBe(
+      'startedAt,level,workflow,trigger,durationMs,costTotal,workflowId,executionId,message,traceSpans\n'
+    )
+    expect(dbChainMockFns.where).not.toHaveBeenCalled()
+    expect(mockMaterializeExecutionDataForDisplay).not.toHaveBeenCalled()
+  })
+
+  it('materializes bounded chunks while preserving CSV row order', async () => {
+    queueTableRows(
+      workflowExecutionLogs,
+      Array.from({ length: 45 }, (_, index) => logRow(index))
+    )
 
     const response = await GET(makeRequest())
     const lines = (await response.text()).trimEnd().split('\n')
 
     expect(response.status).toBe(200)
-    expect(mockMapWithConcurrency.mock.calls.map(([items]) => items.length)).toEqual([1, 1, 1])
-    expect(lines).toHaveLength(4)
+    expect(mockMapWithConcurrency.mock.calls.map(([items]) => items.length)).toEqual([20, 20, 5])
+    expect(lines).toHaveLength(46)
     expect(lines[1]).toContain('execution-0')
-    expect(lines.at(-1)).toContain('execution-2')
+    expect(lines.at(-1)).toContain('execution-44')
   })
 
   it('resumes full pages by startedAt and id without using OFFSET', async () => {

apps/sim/app/api/logs/export/route.ts

@@ -5,7 +5,7 @@ import { getErrorMessage } from '@sim/utils/errors'
 import { and, desc, eq, lt, or, type SQL, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { getSession } from '@/lib/auth'
-import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
+import { MATERIALIZE_CONCURRENCY, mapWithConcurrency } from '@/lib/core/utils/concurrency'
 import { formatCsvValue, toCsvRow } from '@/lib/core/utils/csv'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { materializeExecutionDataForDisplay } from '@/lib/logs/execution/trace-store'
@@ -15,7 +15,6 @@ import { checkWorkspaceAccess } from '@/lib/workspaces/permissions/utils'
 
 const logger = createLogger('LogsExportAPI')
 const LOG_EXPORT_PAGE_SIZE = 100
-const LOG_EXPORT_MATERIALIZE_CONCURRENCY = 1
 
 export const revalidate = 0
 
@@ -120,25 +119,18 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
 
         if (!rows.length) break
 
-        for (
-          let chunkStart = 0;
-          chunkStart < rows.length;
-          chunkStart += LOG_EXPORT_MATERIALIZE_CONCURRENCY
-        ) {
-          const chunk = rows.slice(chunkStart, chunkStart + LOG_EXPORT_MATERIALIZE_CONCURRENCY)
-          const materialized = await mapWithConcurrency(
-            chunk,
-            LOG_EXPORT_MATERIALIZE_CONCURRENCY,
-            (row) =>
-              materializeExecutionDataForDisplay(
-                row.executionData as Record<string, unknown> | null,
-                {
-                  workspaceId: params.workspaceId,
-                  workflowId: row.workflowId,
-                  executionId: row.executionId,
-                  userId: session.user.id,
-                }
-              )
+        for (let chunkStart = 0; chunkStart < rows.length; chunkStart += MATERIALIZE_CONCURRENCY) {
+          const chunk = rows.slice(chunkStart, chunkStart + MATERIALIZE_CONCURRENCY)
+          const materialized = await mapWithConcurrency(chunk, MATERIALIZE_CONCURRENCY, (row) =>
+            materializeExecutionDataForDisplay(
+              row.executionData as Record<string, unknown> | null,
+              {
+                workspaceId: params.workspaceId,
+                workflowId: row.workflowId,
+                executionId: row.executionId,
+                userId: session.user.id,
+              }
+            )
           )
 
           for (let index = 0; index < chunk.length; index++) {

apps/sim/lib/billing/calculations/usage-monitor.test.ts

@@ -141,6 +141,29 @@ describe('checkUsageStatus', () => {
     expect(mockGetBillingPeriodUsageCost).not.toHaveBeenCalled()
   })
 
+  it('preserves the paid daily-refresh clamp for negative effective usage', async () => {
+    const periodStart = new Date('2026-06-01T00:00:00.000Z')
+    const periodEnd = new Date('2026-07-01T00:00:00.000Z')
+    const subscription = {
+      referenceId: 'user-1',
+      plan: 'pro',
+      status: 'active',
+      seats: 1,
+      periodStart,
+      periodEnd,
+    }
+    dbChainMockFns.limit.mockResolvedValueOnce([{ currentPeriodCost: '0' }])
+    mockComputeBillingPeriodUsageWithDailyRefresh.mockResolvedValueOnce({
+      ledgerUsage: -1,
+      refreshConsumed: 1,
+    })
+
+    await expect(checkUsageStatus('user-1', subscription)).resolves.toMatchObject({
+      currentUsage: 0,
+      scope: 'user',
+    })
+  })
+
   it('keeps unpaid personal usage on the ledger-only query', async () => {
     const periodStart = new Date('2026-06-01T00:00:00.000Z')
     const periodEnd = new Date('2026-07-01T00:00:00.000Z')
@@ -166,6 +189,28 @@ describe('checkUsageStatus', () => {
     expect(mockComputeBillingPeriodUsageWithDailyRefresh).not.toHaveBeenCalled()
   })
 
+  it('preserves negative ledger-only personal usage', async () => {
+    const periodStart = new Date('2026-06-01T00:00:00.000Z')
+    const periodEnd = new Date('2026-07-01T00:00:00.000Z')
+    const subscription = {
+      referenceId: 'user-1',
+      plan: 'free',
+      status: 'active',
+      seats: 1,
+      periodStart,
+      periodEnd,
+    }
+    dbChainMockFns.limit.mockResolvedValueOnce([{ currentPeriodCost: '0' }])
+    mockGetBillingPeriodUsageCost.mockResolvedValueOnce(-1)
+
+    await expect(checkUsageStatus('user-1', subscription)).resolves.toMatchObject({
+      currentUsage: -1,
+      scope: 'user',
+    })
+
+    expect(mockComputeBillingPeriodUsageWithDailyRefresh).not.toHaveBeenCalled()
+  })
+
   it('combines paid organization ledger usage with bounded member refresh', async () => {
     const periodStart = new Date('2026-06-01T00:00:00.000Z')
     const periodEnd = new Date('2026-07-01T00:00:00.000Z')

apps/sim/lib/billing/calculations/usage-monitor.ts

@@ -182,6 +182,7 @@ export async function checkUsageStatus(
         : defaultBillingPeriod())
     let ledgerUsage: number
     let refreshConsumed = 0
+    let appliedDailyRefresh = false
     if (sub && isPaid(sub.plan) && sub.periodStart) {
       const planDollars = getPlanTierDollars(sub.plan)
       if (planDollars > 0) {
@@ -195,16 +196,16 @@ export async function checkUsageStatus(
         })
         ledgerUsage = usage.ledgerUsage
         refreshConsumed = usage.refreshConsumed
+        appliedDailyRefresh = true
       } else {
         ledgerUsage = await getBillingPeriodUsageCost({ type: 'user', id: userId }, billingPeriod)
       }
     } else {
       ledgerUsage = await getBillingPeriodUsageCost({ type: 'user', id: userId }, billingPeriod)
     }
-    const currentUsage = Math.max(
-      0,
+    const usageBeforeRefresh =
       toNumber(toDecimal(statsRecords[0].currentPeriodCost)) + ledgerUsage - refreshConsumed
-    )
+    const currentUsage = appliedDailyRefresh ? Math.max(0, usageBeforeRefresh) : usageBeforeRefresh
 
     return buildUsageData({ currentUsage, limit, scope, organizationId })
   } catch (error) {

packages/db/migrations/0303_colorful_gressill.sql

@@ -1,15 +1,12 @@
--- migration-safe: tune the append-only ledger's vacuum thresholds without changing row data or query semantics.
+-- migration-safe: tune the append-heavy ledger's vacuum thresholds without changing row data or query semantics.
 ALTER TABLE "usage_log" SET (autovacuum_vacuum_insert_scale_factor = 0.01, autovacuum_analyze_scale_factor = 0.01);--> statement-breakpoint
 -- Concurrent index operations cannot run inside the migration runner's transaction.
 COMMIT;--> statement-breakpoint
 SET lock_timeout = 0;--> statement-breakpoint
 -- migration-safe: replay removes an invalid build created by this migration; concurrent operations preserve writes.
-DROP INDEX CONCURRENTLY IF EXISTS "doc_active_kb_filename_idx";--> statement-breakpoint
-CREATE INDEX CONCURRENTLY IF NOT EXISTS "doc_active_kb_filename_idx" ON "document" USING btree ("knowledge_base_id","filename","uploaded_at" DESC,"token_count") WHERE "document"."user_excluded" = false AND "document"."archived_at" IS NULL AND "document"."deleted_at" IS NULL;--> statement-breakpoint
--- migration-safe: replay removes an invalid build created by this migration; concurrent operations preserve writes.
-DROP INDEX CONCURRENTLY IF EXISTS "folder_active_workspace_resource_sort_idx";--> statement-breakpoint
-CREATE INDEX CONCURRENTLY IF NOT EXISTS "folder_active_workspace_resource_sort_idx" ON "folder" USING btree ("workspace_id","resource_type","sort_order","created_at") WHERE "folder"."deleted_at" IS NULL;--> statement-breakpoint
--- migration-safe: the existing connector_id index remains available while this ordered replacement expands.
+DROP INDEX CONCURRENTLY IF EXISTS "doc_active_kb_token_count_idx";--> statement-breakpoint
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "doc_active_kb_token_count_idx" ON "document" USING btree ("knowledge_base_id","token_count") WHERE "document"."user_excluded" = false AND "document"."archived_at" IS NULL AND "document"."deleted_at" IS NULL;--> statement-breakpoint
+-- migration-safe: the existing connector_id index remains available until the ordered replacement is valid.
 DROP INDEX CONCURRENTLY IF EXISTS "kcsl_connector_started_at_idx";--> statement-breakpoint
 CREATE INDEX CONCURRENTLY IF NOT EXISTS "kcsl_connector_started_at_idx" ON "knowledge_connector_sync_log" USING btree ("connector_id","started_at" DESC);--> statement-breakpoint
 -- migration-safe: replay removes an invalid build created by this migration; concurrent operations preserve writes.
@@ -18,17 +15,13 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS "table_views_workspace_created_idx" ON "
 -- migration-safe: replay removes an invalid build created by this migration; concurrent operations preserve writes.
 DROP INDEX CONCURRENTLY IF EXISTS "workflow_active_workspace_sort_idx";--> statement-breakpoint
 CREATE INDEX CONCURRENTLY IF NOT EXISTS "workflow_active_workspace_sort_idx" ON "workflow" USING btree ("workspace_id","sort_order","created_at","id") WHERE "workflow"."archived_at" IS NULL;--> statement-breakpoint
--- migration-safe: replay removes an invalid build created by this migration; concurrent operations preserve writes.
-DROP INDEX CONCURRENTLY IF EXISTS "workflow_active_workspace_folder_sort_idx";--> statement-breakpoint
-CREATE INDEX CONCURRENTLY IF NOT EXISTS "workflow_active_workspace_folder_sort_idx" ON "workflow" USING btree ("workspace_id","folder_id","sort_order","created_at","id") WHERE "workflow"."archived_at" IS NULL;--> statement-breakpoint
--- migration-safe: the existing workflow/time index remains available while the keyset replacement expands.
-DROP INDEX CONCURRENTLY IF EXISTS "workflow_execution_logs_workflow_started_at_id_idx";--> statement-breakpoint
-CREATE INDEX CONCURRENTLY IF NOT EXISTS "workflow_execution_logs_workflow_started_at_id_idx" ON "workflow_execution_logs" USING btree ("workflow_id","started_at","id");--> statement-breakpoint
--- migration-safe: each removed index duplicates a primary or unique index with the identical leading key.
+-- migration-safe: each removed index duplicates a primary or unique index with the identical key definition.
 DROP INDEX CONCURRENTLY IF EXISTS "academy_certificate_number_idx";--> statement-breakpoint
 DROP INDEX CONCURRENTLY IF EXISTS "copilot_async_tool_calls_tool_call_id_idx";--> statement-breakpoint
 DROP INDEX CONCURRENTLY IF EXISTS "mothership_settings_workspace_id_idx";--> statement-breakpoint
 DROP INDEX CONCURRENTLY IF EXISTS "permissions_user_entity_idx";--> statement-breakpoint
 DROP INDEX CONCURRENTLY IF EXISTS "session_token_idx";--> statement-breakpoint
 DROP INDEX CONCURRENTLY IF EXISTS "workspace_file_key_idx";--> statement-breakpoint
+-- migration-safe: the valid ordered connector index covers the old index's complete key definition.
+DROP INDEX CONCURRENTLY IF EXISTS "kcsl_connector_id_idx";--> statement-breakpoint
 SET lock_timeout = '5s';