fixed types

Author: ericallam

apps/webapp/scripts/profile-runs-replication.ts

@@ -33,7 +33,7 @@ async function loadConfig(options: any): Promise<HarnessConfig> {
   if (options.config) {
     console.log(`Loading config from: ${options.config}`);
     const configFile = await fs.readFile(options.config, "utf-8");
-    const fileConfig = JSON.parse(configFile);
+    const fileConfig = JSON.parse(configFile) as Partial<HarnessConfig>;
     config = { ...config, ...fileConfig };
   }
 

apps/webapp/test/performance/clickhouse-mock.ts

@@ -1,4 +1,21 @@
-import type { RawTaskRunPayloadV1, TaskRunV2 } from "@internal/clickhouse";
+import type { ClickHouse, TaskRunInsertArray, PayloadInsertArray } from "@internal/clickhouse";
+
+// Minimal InsertResult type to avoid dependency on @clickhouse/client
+interface InsertResult {
+  executed: boolean;
+  query_id: string;
+  summary?: {
+    read_rows: string;
+    read_bytes: string;
+    written_rows: string;
+    written_bytes: string;
+    total_rows_to_read: string;
+    result_rows: string;
+    result_bytes: string;
+    elapsed_ns: string;
+  };
+  response_headers: Record<string, string>;
+}
 
 /**
  * Mock ClickHouse client for CPU-only profiling.
@@ -12,30 +29,64 @@ export class MockClickHouse {
   constructor(private readonly insertDelayMs: number = 0) {}
 
   taskRuns = {
-    insert: async (
-      runs: TaskRunV2[],
+    insertCompactArrays: async (
+      rows: TaskRunInsertArray[],
       options?: any
-    ): Promise<[Error | null, { rows: number } | null]> => {
+    ): Promise<[Error | null, InsertResult | null]> => {
       if (this.insertDelayMs > 0) {
         await new Promise((resolve) => setTimeout(resolve, this.insertDelayMs));
       }
 
-      this.insertCount += runs.length;
+      this.insertCount += rows.length;
 
-      return [null, { rows: runs.length }];
+      return [
+        null,
+        {
+          executed: true,
+          query_id: "mock",
+          summary: {
+            read_rows: "0",
+            read_bytes: "0",
+            written_rows: String(rows.length),
+            written_bytes: "0",
+            total_rows_to_read: "0",
+            result_rows: "0",
+            result_bytes: "0",
+            elapsed_ns: "0",
+          },
+          response_headers: {},
+        },
+      ];
     },
 
-    insertPayloads: async (
-      payloads: RawTaskRunPayloadV1[],
+    insertPayloadsCompactArrays: async (
+      rows: PayloadInsertArray[],
       options?: any
-    ): Promise<[Error | null, { rows: number } | null]> => {
+    ): Promise<[Error | null, InsertResult | null]> => {
       if (this.insertDelayMs > 0) {
         await new Promise((resolve) => setTimeout(resolve, this.insertDelayMs));
       }
 
-      this.payloadInsertCount += payloads.length;
+      this.payloadInsertCount += rows.length;
 
-      return [null, { rows: payloads.length }];
+      return [
+        null,
+        {
+          executed: true,
+          query_id: "mock",
+          summary: {
+            read_rows: "0",
+            read_bytes: "0",
+            written_rows: String(rows.length),
+            written_bytes: "0",
+            total_rows_to_read: "0",
+            result_rows: "0",
+            result_bytes: "0",
+            elapsed_ns: "0",
+          },
+          response_headers: {},
+        },
+      ];
     },
   };
 
@@ -51,3 +102,8 @@ export class MockClickHouse {
     this.payloadInsertCount = 0;
   }
 }
+
+// Type assertion helper for use with RunsReplicationService
+export function asMockClickHouse(mock: MockClickHouse): Pick<ClickHouse, "taskRuns"> {
+  return mock as unknown as Pick<ClickHouse, "taskRuns">;
+}

apps/webapp/test/performance/config.ts

@@ -1,5 +1,5 @@
 import { RedisOptions } from "ioredis";
-import { RuntimeEnvironmentType } from "~/database-types";
+import type { RuntimeEnvironmentType } from "@trigger.dev/database";
 import { config as loadEnv } from "dotenv";
 import path from "path";
 

apps/webapp/test/performance/consumer-runner.ts

@@ -1,7 +1,8 @@
 #!/usr/bin/env tsx
 
+import type { ClickHouse } from "@internal/clickhouse";
 import { RunsReplicationService } from "~/services/runsReplicationService.server";
-import { MockClickHouse } from "./clickhouse-mock";
+import { MockClickHouse, asMockClickHouse } from "./clickhouse-mock";
 import type { ConsumerConfig } from "./config";
 import fs from "fs";
 import path from "path";
@@ -16,7 +17,7 @@ async function main() {
   }
 
   // Parse configuration from environment variable
-  const config: ConsumerConfig = JSON.parse(process.env.CONSUMER_CONFIG!);
+  const config = JSON.parse(process.env.CONSUMER_CONFIG!) as ConsumerConfig;
 
   // Create shutdown signal file path
   const shutdownFilePath = path.join(config.outputDir || "/tmp", ".shutdown-signal");
@@ -29,9 +30,9 @@ async function main() {
   });
 
   // Create ClickHouse client (real or mocked)
-  let clickhouse;
+  let clickhouse: Pick<ClickHouse, "taskRuns">;
   if (config.useMockClickhouse) {
-    clickhouse = new MockClickHouse(config.mockClickhouseDelay);
+    clickhouse = asMockClickHouse(new MockClickHouse(config.mockClickhouseDelay));
   } else {
     // Use dynamic import to avoid module resolution issues with tsx
     const { ClickHouse } = await import("@internal/clickhouse");
@@ -47,7 +48,7 @@ async function main() {
 
   // Create replication service
   const service = new RunsReplicationService({
-    clickhouse,
+    clickhouse: clickhouse as ClickHouse,
     pgConnectionUrl: config.pgConnectionUrl,
     serviceName: "runs-replication-profiling",
     slotName: config.slotName,
@@ -89,7 +90,8 @@ async function main() {
   // Send periodic metrics to parent (if IPC available)
   const metricsInterval = setInterval(() => {
     const memUsage = process.memoryUsage();
-    const elu = performance.eventLoopUtilization();
+    const { eventLoopUtilization } = require("perf_hooks").performance;
+    const elu = eventLoopUtilization ? eventLoopUtilization() : { utilization: 0 };
 
     if (hasIPC) {
       process.send!({

apps/webapp/test/performance/consumer.ts

@@ -52,7 +52,7 @@ export class ConsumerProcessManager {
     if (isProfiling && this.process.stdout) {
       this.process.stdout.on("data", (data: Buffer) => {
         const output = data.toString();
-        process.stdout.write(data);  // Still show output
+        process.stdout.write(output); // Still show output
         if (output.includes("Consumer process ready")) {
           this.ready = true;
           console.log("Consumer process is ready (detected from stdout)");
@@ -62,7 +62,7 @@ export class ConsumerProcessManager {
 
     if (isProfiling && this.process.stderr) {
       this.process.stderr.on("data", (data: Buffer) => {
-        process.stderr.write(data);  // Show stderr
+        process.stderr.write(data.toString()); // Show stderr
       });
     }
 

apps/webapp/test/performance/data-generator.ts

@@ -1,6 +1,5 @@
-import { Prisma } from "@trigger.dev/database";
+import { Prisma, RuntimeEnvironmentType, TaskRunStatus } from "@trigger.dev/database";
 import { nanoid } from "nanoid";
-import { RuntimeEnvironmentType, TaskRunStatus } from "~/database-types";
 import superjson from "superjson";
 
 export interface DataGeneratorOptions {
@@ -41,15 +40,15 @@ export class TaskRunDataGenerator {
 
   constructor(private readonly options: DataGeneratorOptions) {}
 
-  generateBatch(count: number): Prisma.TaskRunCreateInput[] {
-    const batch: Prisma.TaskRunCreateInput[] = [];
+  generateBatch(count: number): Prisma.TaskRunCreateManyInput[] {
+    const batch: Prisma.TaskRunCreateManyInput[] = [];
     for (let i = 0; i < count; i++) {
       batch.push(this.generateInsert());
     }
     return batch;
   }
 
-  generateInsert(): Prisma.TaskRunCreateInput {
+  generateInsert(): Prisma.TaskRunCreateManyInput {
     this.counter++;
     const id = nanoid();
     const friendlyId = `run_${this.counter}_${nanoid(8)}`;
@@ -59,17 +58,19 @@ export class TaskRunDataGenerator {
     const workerQueue = this.workerQueues[Math.floor(Math.random() * this.workerQueues.length)];
 
     const payload = this.generatePayload();
-    const payloadType = this.options.includeComplexPayloads && Math.random() > 0.7
-      ? "application/super+json"
-      : "application/json";
+    const payloadType =
+      this.options.includeComplexPayloads && Math.random() > 0.7
+        ? "application/super+json"
+        : "application/json";
 
     return {
       id,
       friendlyId,
       taskIdentifier,
-      payload: payloadType === "application/super+json"
-        ? superjson.stringify(payload)
-        : JSON.stringify(payload),
+      payload:
+        payloadType === "application/super+json"
+          ? superjson.stringify(payload)
+          : JSON.stringify(payload),
       payloadType,
       traceId: nanoid(),
       spanId: nanoid(),

apps/webapp/test/performance/harness.ts

@@ -267,9 +267,10 @@ export class RunsReplicationHarness {
       const cleanProfilingUrl = this.profilingDatabaseUrl.split("?")[0];
 
       // Dump schema only (no data) from main database
-      execSync(`pg_dump "${cleanBaseUrl}" --schema-only --no-owner --no-acl | psql "${cleanProfilingUrl}" > /dev/null 2>&1`, {
-        shell: true,
-      });
+      execSync(
+        `pg_dump "${cleanBaseUrl}" --schema-only --no-owner --no-acl | psql "${cleanProfilingUrl}" > /dev/null 2>&1`,
+        { shell: "/bin/bash" }
+      );
 
       console.log("✅ Schema copied successfully");
     } catch (error) {

apps/webapp/test/performance/producer-runner.ts

@@ -11,7 +11,7 @@ async function main() {
   }
 
   // Parse configuration from environment variable
-  const config: ProducerConfig = JSON.parse(process.env.PRODUCER_CONFIG!);
+  const config = JSON.parse(process.env.PRODUCER_CONFIG!) as ProducerConfig;
 
   console.log("Producer process starting with config:", {
     targetThroughput: config.targetThroughput,