cleanup the types

Author: ericallam

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

@@ -1,4 +1,4 @@
-import type { ClickHouse, RawTaskRunPayloadV1, TaskRunV2 } from "@internal/clickhouse";
+import type { ClickHouse } from "@internal/clickhouse";
 import { TASK_RUN_COLUMNS, PAYLOAD_COLUMNS } from "@internal/clickhouse";
 import { type RedisOptions } from "@internal/redis";
 import {
@@ -915,12 +915,12 @@ export class RunsReplicationService {
       run.idempotencyKey ?? "", // idempotency_key
       run.ttl ?? "", // expiration_ttl
       run.isTest ?? false, // is_test
+      _version.toString(), // _version
+      event === "delete" ? 1 : 0, // _is_deleted
       run.concurrencyKey ?? "", // concurrency_key
       run.bulkActionGroupIds ?? [], // bulk_action_group_ids
       run.masterQueue ?? "", // worker_queue
       run.maxDurationInSeconds ?? 0, // max_duration_in_seconds
-      _version.toString(), // _version
-      event === "delete" ? 1 : 0, // _is_deleted
     ];
   }
 

internal-packages/clickhouse/src/client/client.ts

@@ -6,9 +6,11 @@ import {
   createClient,
   type ResultSet,
   type Row,
+  type BaseQueryParams,
+  type InsertResult,
 } from "@clickhouse/client";
 import { recordSpanError, Span, startSpan, trace, Tracer } from "@internal/tracing";
-import { flattenAttributes, tryCatch } from "@trigger.dev/core/v3";
+import { flattenAttributes, tryCatch, type Result } from "@trigger.dev/core/v3";
 import { z } from "zod";
 import { InsertError, QueryError } from "./errors.js";
 import type {
@@ -797,6 +799,105 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
       });
     };
   }
+
+  public insertCompactRaw(req: {
+    name: string;
+    table: string;
+    columns: readonly string[];
+    settings?: ClickHouseSettings;
+  }): (
+    events: readonly any[][] | any[],
+    options?: {
+      attributes?: Record<string, string | number | boolean>;
+      params?: BaseQueryParams;
+    }
+  ) => Promise<Result<InsertResult, InsertError>> {
+    return async (events, options) => {
+      const queryId = randomUUID();
+
+      return await startSpan(this.tracer, "insert", async (span) => {
+        // Check if events is a single row (array) or multiple rows (array of arrays)
+        // If first element is not an array, treat as single row
+        const isSingleRow = events.length > 0 && !Array.isArray(events[0]);
+        const eventsArray: readonly any[][] = isSingleRow
+          ? [events as any[]]
+          : (events as readonly any[][]);
+
+        this.logger.debug("Inserting into clickhouse (compact raw)", {
+          clientName: this.name,
+          name: req.name,
+          table: req.table,
+          events: eventsArray.length,
+          settings: req.settings,
+          attributes: options?.attributes,
+          options,
+          queryId,
+        });
+
+        span.setAttributes({
+          "clickhouse.clientName": this.name,
+          "clickhouse.tableName": req.table,
+          "clickhouse.operationName": req.name,
+          "clickhouse.queryId": queryId,
+          "clickhouse.format": "JSONCompactEachRowWithNames",
+          ...flattenAttributes(req.settings, "clickhouse.settings"),
+          ...flattenAttributes(options?.attributes),
+        });
+
+        // Build compact format: [columns, ...rows]
+        // Data is already in array format, no conversion needed
+        const compactData: any[] = [Array.from(req.columns), ...eventsArray];
+
+        const [clickhouseError, result] = await tryCatch(
+          this.client.insert({
+            table: req.table,
+            format: "JSONCompactEachRowWithNames",
+            values: compactData,
+            query_id: queryId,
+            ...options?.params,
+            clickhouse_settings: {
+              ...req.settings,
+              ...options?.params?.clickhouse_settings,
+            },
+          })
+        );
+
+        if (clickhouseError) {
+          this.logger.error("Error inserting into clickhouse", {
+            name: req.name,
+            error: clickhouseError,
+            table: req.table,
+          });
+
+          recordClickhouseError(span, clickhouseError);
+          return [new InsertError(clickhouseError.message), null];
+        }
+
+        this.logger.debug("Inserted into clickhouse", {
+          clientName: this.name,
+          name: req.name,
+          table: req.table,
+          result,
+          queryId,
+        });
+
+        span.setAttributes({
+          "clickhouse.query_id": result.query_id,
+          "clickhouse.executed": result.executed,
+          "clickhouse.summary.read_rows": result.summary?.read_rows,
+          "clickhouse.summary.read_bytes": result.summary?.read_bytes,
+          "clickhouse.summary.written_rows": result.summary?.written_rows,
+          "clickhouse.summary.written_bytes": result.summary?.written_bytes,
+          "clickhouse.summary.total_rows_to_read": result.summary?.total_rows_to_read,
+          "clickhouse.summary.result_rows": result.summary?.result_rows,
+          "clickhouse.summary.result_bytes": result.summary?.result_bytes,
+          "clickhouse.summary.elapsed_ns": result.summary?.elapsed_ns,
+        });
+
+        return [null, result];
+      });
+    };
+  }
 }
 
 function recordClickhouseError(span: Span, error: Error) {

internal-packages/clickhouse/src/client/noop.ts

@@ -188,4 +188,32 @@ export class NoopClient implements ClickhouseReader, ClickhouseWriter {
       ];
     };
   }
+
+  public insertCompactRaw(req: {
+    name: string;
+    table: string;
+    columns: readonly string[];
+    settings?: ClickHouseSettings;
+  }): (events: readonly any[][] | any[]) => Promise<Result<InsertResult, InsertError>> {
+    return async (events: readonly any[][] | any[]) => {
+      return [
+        null,
+        {
+          executed: true,
+          query_id: "noop",
+          summary: {
+            read_rows: "0",
+            read_bytes: "0",
+            written_rows: "0",
+            written_bytes: "0",
+            total_rows_to_read: "0",
+            result_rows: "0",
+            result_bytes: "0",
+            elapsed_ns: "0",
+          },
+          response_headers: {},
+        },
+      ];
+    };
+  }
 }

internal-packages/clickhouse/src/client/types.ts

@@ -228,5 +228,18 @@ export interface ClickhouseWriter {
     settings?: ClickHouseSettings;
   }): ClickhouseInsertFunction<TRecord>;
 
+  insertCompactRaw(req: {
+    name: string;
+    table: string;
+    columns: readonly string[];
+    settings?: ClickHouseSettings;
+  }): (
+    events: readonly any[][] | any[],
+    options?: {
+      attributes?: Record<string, string | number | boolean>;
+      params?: BaseQueryParams;
+    }
+  ) => Promise<Result<InsertResult, InsertError>>;
+
   close(): Promise<void>;
 }

internal-packages/clickhouse/src/index.ts

@@ -3,13 +3,7 @@ import { ClickhouseClient } from "./client/client.js";
 import { ClickhouseReader, ClickhouseWriter } from "./client/types.js";
 import { NoopClient } from "./client/noop.js";
 import {
-  insertTaskRuns,
-  insertTaskRunsUnsafe,
-  insertTaskRunsCompact,
   insertTaskRunsCompactArrays,
-  insertRawTaskRunPayloads,
-  insertRawTaskRunPayloadsUnsafe,
-  insertRawTaskRunPayloadsCompact,
   insertRawTaskRunPayloadsCompactArrays,
   getTaskRunsQueryBuilder,
   getTaskActivityQueryBuilder,
@@ -174,13 +168,7 @@ export class ClickHouse {
 
   get taskRuns() {
     return {
-      insert: insertTaskRuns(this.writer),
-      insertUnsafe: insertTaskRunsUnsafe(this.writer),
-      insertCompact: insertTaskRunsCompact(this.writer),
       insertCompactArrays: insertTaskRunsCompactArrays(this.writer),
-      insertPayloads: insertRawTaskRunPayloads(this.writer),
-      insertPayloadsUnsafe: insertRawTaskRunPayloadsUnsafe(this.writer),
-      insertPayloadsCompact: insertRawTaskRunPayloadsCompact(this.writer),
       insertPayloadsCompactArrays: insertRawTaskRunPayloadsCompactArrays(this.writer),
       queryBuilder: getTaskRunsQueryBuilder(this.reader),
       countQueryBuilder: getTaskRunsCountQueryBuilder(this.reader),