much better type safety

Author: ericallam

apps/webapp/app/services/runsReplicationService.server.ts

@@ -1,4 +1,5 @@
-import type { ClickHouse } from "@internal/clickhouse";
+import type { ClickHouse, TaskRunInsertArray, PayloadInsertArray } from "@internal/clickhouse";
+import { TASK_RUN_INDEX, PAYLOAD_INDEX } from "@internal/clickhouse";
 import { type RedisOptions } from "@internal/redis";
 import {
   LogicalReplicationClient,
@@ -80,7 +81,9 @@ type TaskRunInsert = {
 
 export type RunsReplicationServiceEvents = {
   message: [{ lsn: string; message: PgoutputMessage; service: RunsReplicationService }];
-  batchFlushed: [{ flushId: string; taskRunInserts: any[][]; payloadInserts: any[][] }];
+  batchFlushed: [
+    { flushId: string; taskRunInserts: TaskRunInsertArray[]; payloadInserts: PayloadInsertArray[] },
+  ];
 };
 
 export class RunsReplicationService {
@@ -576,35 +579,29 @@ export class RunsReplicationService {
         .filter(Boolean)
         // batch inserts in clickhouse are more performant if the items
         // are pre-sorted by the primary key
-        // Array indices: [0]=environment_id, [1]=organization_id, [2]=project_id, [3]=run_id, [5]=created_at
         .sort((a, b) => {
-          if (a[1] !== b[1]) {
-            // organization_id
-            return a[1] < b[1] ? -1 : 1;
+          if (a[TASK_RUN_INDEX.organization_id] !== b[TASK_RUN_INDEX.organization_id]) {
+            return a[TASK_RUN_INDEX.organization_id] < b[TASK_RUN_INDEX.organization_id] ? -1 : 1;
           }
-          if (a[2] !== b[2]) {
-            // project_id
-            return a[2] < b[2] ? -1 : 1;
+          if (a[TASK_RUN_INDEX.project_id] !== b[TASK_RUN_INDEX.project_id]) {
+            return a[TASK_RUN_INDEX.project_id] < b[TASK_RUN_INDEX.project_id] ? -1 : 1;
           }
-          if (a[0] !== b[0]) {
-            // environment_id
-            return a[0] < b[0] ? -1 : 1;
+          if (a[TASK_RUN_INDEX.environment_id] !== b[TASK_RUN_INDEX.environment_id]) {
+            return a[TASK_RUN_INDEX.environment_id] < b[TASK_RUN_INDEX.environment_id] ? -1 : 1;
           }
-          if (a[5] !== b[5]) {
-            // created_at
-            return a[5] - b[5];
+          if (a[TASK_RUN_INDEX.created_at] !== b[TASK_RUN_INDEX.created_at]) {
+            return a[TASK_RUN_INDEX.created_at] - b[TASK_RUN_INDEX.created_at];
           }
-          return a[3] < b[3] ? -1 : 1; // run_id
+          return a[TASK_RUN_INDEX.run_id] < b[TASK_RUN_INDEX.run_id] ? -1 : 1;
         });
 
       const payloadInserts = preparedInserts
         .map(({ payloadInsert }) => payloadInsert)
         .filter(Boolean)
         // batch inserts in clickhouse are more performant if the items
         // are pre-sorted by the primary key
-        // Array indices: [0]=run_id
         .sort((a, b) => {
-          return a[0] < b[0] ? -1 : 1; // run_id
+          return a[PAYLOAD_INDEX.run_id] < b[PAYLOAD_INDEX.run_id] ? -1 : 1;
         });
 
       span.setAttribute("task_run_inserts", taskRunInserts.length);
@@ -769,7 +766,7 @@ export class RunsReplicationService {
     };
   }
 
-  async #insertTaskRunInserts(taskRunInserts: any[][], attempt: number) {
+  async #insertTaskRunInserts(taskRunInserts: TaskRunInsertArray[], attempt: number) {
     return await startSpan(this._tracer, "insertTaskRunsInserts", async (span) => {
       const [insertError, insertResult] =
         await this.options.clickhouse.taskRuns.insertCompactArrays(taskRunInserts, {
@@ -792,7 +789,7 @@ export class RunsReplicationService {
     });
   }
 
-  async #insertPayloadInserts(payloadInserts: any[][], attempt: number) {
+  async #insertPayloadInserts(payloadInserts: PayloadInsertArray[], attempt: number) {
     return await startSpan(this._tracer, "insertPayloadInserts", async (span) => {
       const [insertError, insertResult] =
         await this.options.clickhouse.taskRuns.insertPayloadsCompactArrays(payloadInserts, {
@@ -817,7 +814,7 @@ export class RunsReplicationService {
 
   async #prepareRunInserts(
     batchedRun: TaskRunInsert
-  ): Promise<{ taskRunInsert?: any[]; payloadInsert?: any[] }> {
+  ): Promise<{ taskRunInsert?: TaskRunInsertArray; payloadInsert?: PayloadInsertArray }> {
     this.logger.debug("Preparing run", {
       batchedRun,
     });
@@ -854,7 +851,7 @@ export class RunsReplicationService {
     environmentType: string,
     event: "insert" | "update" | "delete",
     _version: bigint
-  ): Promise<any[]> {
+  ): Promise<TaskRunInsertArray> {
     const output = await this.#prepareJson(run.output, run.outputType);
 
     // Return array matching TASK_RUN_COLUMNS order
@@ -907,7 +904,7 @@ export class RunsReplicationService {
     ];
   }
 
-  async #preparePayloadInsert(run: TaskRun, _version: bigint): Promise<any[]> {
+  async #preparePayloadInsert(run: TaskRun, _version: bigint): Promise<PayloadInsertArray> {
     const payload = await this.#prepareJson(run.payload, run.payloadType);
 
     // Return array matching PAYLOAD_COLUMNS order

apps/webapp/test/runsReplicationService.part2.test.ts

@@ -1,4 +1,4 @@
-import { ClickHouse } from "@internal/clickhouse";
+import { ClickHouse, TASK_RUN_INDEX } from "@internal/clickhouse";
 import { containerTest } from "@internal/testcontainers";
 import { Logger } from "@trigger.dev/core/logger";
 import { readFile } from "node:fs/promises";
@@ -889,12 +889,15 @@ describe("RunsReplicationService (part 2/2)", () => {
       await setTimeout(1000);
 
       expect(batchFlushedEvents?.[0].taskRunInserts).toHaveLength(2);
-      // taskRunInserts are now arrays, not objects
-      // Index 3 is run_id, Index 6 is status
-      expect(batchFlushedEvents?.[0].taskRunInserts[0][3]).toEqual(run.id); // run_id
-      expect(batchFlushedEvents?.[0].taskRunInserts[0][6]).toEqual("PENDING_VERSION"); // status
-      expect(batchFlushedEvents?.[0].taskRunInserts[1][3]).toEqual(run.id); // run_id
-      expect(batchFlushedEvents?.[0].taskRunInserts[1][6]).toEqual("COMPLETED_SUCCESSFULLY"); // status
+      // Use TASK_RUN_INDEX for type-safe array access
+      expect(batchFlushedEvents?.[0].taskRunInserts[0][TASK_RUN_INDEX.run_id]).toEqual(run.id);
+      expect(batchFlushedEvents?.[0].taskRunInserts[0][TASK_RUN_INDEX.status]).toEqual(
+        "PENDING_VERSION"
+      );
+      expect(batchFlushedEvents?.[0].taskRunInserts[1][TASK_RUN_INDEX.run_id]).toEqual(run.id);
+      expect(batchFlushedEvents?.[0].taskRunInserts[1][TASK_RUN_INDEX.status]).toEqual(
+        "COMPLETED_SUCCESSFULLY"
+      );
 
       await runsReplicationService.stop();
     }

internal-packages/clickhouse/src/index.ts

@@ -31,6 +31,14 @@ export type * from "./taskRuns.js";
 export type * from "./taskEvents.js";
 export type * from "./client/queryBuilder.js";
 
+// Re-export column constants and indices for type-safe array access
+export {
+  TASK_RUN_COLUMNS,
+  TASK_RUN_INDEX,
+  PAYLOAD_COLUMNS,
+  PAYLOAD_INDEX,
+} from "./taskRuns.js";
+
 // TSQL query execution
 export {
   executeTSQL,

internal-packages/clickhouse/src/taskRuns.ts

@@ -101,6 +101,72 @@ export const TASK_RUN_COLUMNS = [
   "max_duration_in_seconds",
 ] as const;
 
+// Type-safe column indices derived from TASK_RUN_COLUMNS
+// This ensures indices stay in sync with column order
+export const TASK_RUN_INDEX = {
+  environment_id: 0,
+  organization_id: 1,
+  project_id: 2,
+  run_id: 3,
+  updated_at: 4,
+  created_at: 5,
+  status: 6,
+  environment_type: 7,
+  friendly_id: 8,
+  attempt: 9,
+  engine: 10,
+  task_identifier: 11,
+  queue: 12,
+  schedule_id: 13,
+  batch_id: 14,
+  completed_at: 15,
+  started_at: 16,
+  executed_at: 17,
+  delay_until: 18,
+  queued_at: 19,
+  expired_at: 20,
+  usage_duration_ms: 21,
+  cost_in_cents: 22,
+  base_cost_in_cents: 23,
+  output: 24,
+  error: 25,
+  tags: 26,
+  task_version: 27,
+  sdk_version: 28,
+  cli_version: 29,
+  machine_preset: 30,
+  root_run_id: 31,
+  parent_run_id: 32,
+  depth: 33,
+  span_id: 34,
+  trace_id: 35,
+  idempotency_key: 36,
+  expiration_ttl: 37,
+  is_test: 38,
+  _version: 39,
+  _is_deleted: 40,
+  concurrency_key: 41,
+  bulk_action_group_ids: 42,
+  worker_queue: 43,
+  max_duration_in_seconds: 44,
+} as const satisfies Record<(typeof TASK_RUN_COLUMNS)[number], number>;
+
+export type TaskRunColumnName = (typeof TASK_RUN_COLUMNS)[number];
+
+// Runtime assertion to verify TASK_RUN_INDEX matches TASK_RUN_COLUMNS order
+// This will throw at module load time if there's a mismatch
+(function verifyTaskRunColumnIndices() {
+  for (let i = 0; i < TASK_RUN_COLUMNS.length; i++) {
+    const column = TASK_RUN_COLUMNS[i];
+    const index = TASK_RUN_INDEX[column];
+    if (index !== i) {
+      throw new Error(
+        `TASK_RUN_INDEX mismatch: column "${column}" has index ${index} but should be ${i}`
+      );
+    }
+  }
+})();
+
 export function insertTaskRunsCompactArrays(ch: ClickhouseWriter, settings?: ClickHouseSettings) {
   return ch.insertCompactRaw({
     name: "insertTaskRunsCompactArrays",
@@ -124,6 +190,77 @@ export type RawTaskRunPayloadV1 = z.infer<typeof RawTaskRunPayloadV1>;
 
 export const PAYLOAD_COLUMNS = ["run_id", "created_at", "payload"] as const;
 
+// Type-safe column indices for payload columns
+export const PAYLOAD_INDEX = {
+  run_id: 0,
+  created_at: 1,
+  payload: 2,
+} as const satisfies Record<(typeof PAYLOAD_COLUMNS)[number], number>;
+
+export type PayloadColumnName = (typeof PAYLOAD_COLUMNS)[number];
+
+/**
+ * Type-safe tuple representing a task run insert array.
+ * Order matches TASK_RUN_COLUMNS exactly.
+ */
+export type TaskRunInsertArray = [
+  environment_id: string,
+  organization_id: string,
+  project_id: string,
+  run_id: string,
+  updated_at: number,
+  created_at: number,
+  status: string,
+  environment_type: string,
+  friendly_id: string,
+  attempt: number,
+  engine: string,
+  task_identifier: string,
+  queue: string,
+  schedule_id: string,
+  batch_id: string,
+  completed_at: number | null,
+  started_at: number | null,
+  executed_at: number | null,
+  delay_until: number | null,
+  queued_at: number | null,
+  expired_at: number | null,
+  usage_duration_ms: number,
+  cost_in_cents: number,
+  base_cost_in_cents: number,
+  output: { data: unknown },
+  error: { data: unknown },
+  tags: string[],
+  task_version: string,
+  sdk_version: string,
+  cli_version: string,
+  machine_preset: string,
+  root_run_id: string,
+  parent_run_id: string,
+  depth: number,
+  span_id: string,
+  trace_id: string,
+  idempotency_key: string,
+  expiration_ttl: string,
+  is_test: boolean,
+  _version: string,
+  _is_deleted: number,
+  concurrency_key: string,
+  bulk_action_group_ids: string[],
+  worker_queue: string,
+  max_duration_in_seconds: number | null,
+];
+
+/**
+ * Type-safe tuple representing a payload insert array.
+ * Order matches PAYLOAD_COLUMNS exactly.
+ */
+export type PayloadInsertArray = [
+  run_id: string,
+  created_at: number,
+  payload: { data: unknown },
+];
+
 export function insertRawTaskRunPayloadsCompactArrays(
   ch: ClickhouseWriter,
   settings?: ClickHouseSettings