Run Replication using the factory

Author: matt-aitken

apps/webapp/app/routes/admin.api.v1.runs-replication.create.ts

@@ -2,8 +2,8 @@ import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
 import { prisma } from "~/db.server";
 import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
 import { z } from "zod";
-import { ClickHouse } from "@internal/clickhouse";
 import { env } from "~/env.server";
+import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server";
 import { RunsReplicationService } from "~/services/runsReplicationService.server";
 import {
   getRunsReplicationGlobal,
@@ -62,6 +62,8 @@ export async function action({ request }: ActionFunctionArgs) {
 
     const params = CreateRunReplicationServiceParams.parse(await request.json());
 
+    await clickhouseFactory.isReady();
+
     const service = createRunReplicationService(params);
 
     setRunsReplicationGlobal(service);
@@ -77,24 +79,23 @@ export async function action({ request }: ActionFunctionArgs) {
 }
 
 function createRunReplicationService(params: CreateRunReplicationServiceParams) {
-  const clickhouse = new ClickHouse({
-    url: env.RUN_REPLICATION_CLICKHOUSE_URL,
-    name: params.name,
-    keepAlive: {
-      enabled: params.keepAliveEnabled,
-      idleSocketTtl: params.keepAliveIdleSocketTtl,
-    },
-    logLevel: "debug",
-    compression: {
-      request: true,
-    },
-    maxOpenConnections: params.maxOpenConnections,
-  });
+  const {
+    name,
+    maxFlushConcurrency,
+    flushIntervalMs,
+    flushBatchSize,
+    leaderLockTimeoutMs,
+    leaderLockExtendIntervalMs,
+    leaderLockAcquireAdditionalTimeMs,
+    leaderLockRetryIntervalMs,
+    ackIntervalSeconds,
+    waitForAsyncInsert,
+  } = params;
 
   const service = new RunsReplicationService({
-    clickhouse: clickhouse,
+    clickhouseFactory,
     pgConnectionUrl: env.DATABASE_URL,
-    serviceName: params.name,
+    serviceName: name,
     slotName: env.RUN_REPLICATION_SLOT_NAME,
     publicationName: env.RUN_REPLICATION_PUBLICATION_NAME,
     redisOptions: {
@@ -106,16 +107,16 @@ function createRunReplicationService(params: CreateRunReplicationServiceParams)
       enableAutoPipelining: true,
       ...(env.RUN_REPLICATION_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
     },
-    maxFlushConcurrency: params.maxFlushConcurrency,
-    flushIntervalMs: params.flushIntervalMs,
-    flushBatchSize: params.flushBatchSize,
-    leaderLockTimeoutMs: params.leaderLockTimeoutMs,
-    leaderLockExtendIntervalMs: params.leaderLockExtendIntervalMs,
-    leaderLockAcquireAdditionalTimeMs: params.leaderLockAcquireAdditionalTimeMs,
-    leaderLockRetryIntervalMs: params.leaderLockRetryIntervalMs,
-    ackIntervalSeconds: params.ackIntervalSeconds,
+    maxFlushConcurrency,
+    flushIntervalMs,
+    flushBatchSize,
+    leaderLockTimeoutMs,
+    leaderLockExtendIntervalMs,
+    leaderLockAcquireAdditionalTimeMs,
+    leaderLockRetryIntervalMs,
+    ackIntervalSeconds,
     logLevel: "debug",
-    waitForAsyncInsert: params.waitForAsyncInsert,
+    waitForAsyncInsert,
   });
 
   return service;

apps/webapp/app/routes/admin.api.v1.runs-replication.start.ts

@@ -1,6 +1,7 @@
 import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
 import { prisma } from "~/db.server";
 import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server";
 import { getRunsReplicationGlobal } from "~/services/runsReplicationGlobal.server";
 import { runsReplicationInstance } from "~/services/runsReplicationInstance.server";
 
@@ -29,6 +30,8 @@ export async function action({ request }: ActionFunctionArgs) {
   try {
     const globalService = getRunsReplicationGlobal();
 
+    await clickhouseFactory.isReady();
+
     if (globalService) {
       await globalService.start();
     } else {

apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts

@@ -122,6 +122,34 @@ function initializeQueryClickhouseClient() {
   });
 }
 
+/** TaskRun replication to ClickHouse (`RUN_REPLICATION_CLICKHOUSE_URL`); not exported. */
+const defaultRunsReplicationClickhouseClient = singleton(
+  "runsReplicationClickhouseClient",
+  initializeRunsReplicationClickhouseClient
+);
+
+function initializeRunsReplicationClickhouseClient(): ClickHouse {
+  if (!env.RUN_REPLICATION_CLICKHOUSE_URL) {
+    // Runs replication worker gates on this URL; factory may still resolve "replication" for tests.
+    return defaultClickhouseClient;
+  }
+
+  const url = new URL(env.RUN_REPLICATION_CLICKHOUSE_URL);
+  url.searchParams.delete("secure");
+
+  return new ClickHouse({
+    url: url.toString(),
+    name: "runs-replication",
+    keepAlive: {
+      enabled: env.RUN_REPLICATION_KEEP_ALIVE_ENABLED === "1",
+      idleSocketTtl: env.RUN_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
+    },
+    logLevel: env.RUN_REPLICATION_CLICKHOUSE_LOG_LEVEL,
+    compression: { request: true },
+    maxOpenConnections: env.RUN_REPLICATION_MAX_OPEN_CONNECTIONS,
+  });
+}
+
 // ---------------------------------------------------------------------------
 // Helpers
 // ---------------------------------------------------------------------------
@@ -187,8 +215,9 @@ export class ClickhouseFactory {
       switch (clientType) {
         case "standard":
         case "events":
-        case "replication":
           return defaultClickhouseClient;
+        case "replication":
+          return defaultRunsReplicationClickhouseClient;
         case "logs":
           return defaultLogsClickhouseClient;
         case "query":

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

@@ -1,6 +1,6 @@
-import { ClickHouse } from "@internal/clickhouse";
 import invariant from "tiny-invariant";
 import { env } from "~/env.server";
+import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server";
 import { singleton } from "~/utils/singleton";
 import { meter, provider } from "~/v3/tracer.server";
 import { RunsReplicationService } from "./runsReplicationService.server";
@@ -22,22 +22,8 @@ function initializeRunsReplicationInstance() {
 
   console.log("🗃️  Runs replication service enabled");
 
-  const clickhouse = new ClickHouse({
-    url: env.RUN_REPLICATION_CLICKHOUSE_URL,
-    name: "runs-replication",
-    keepAlive: {
-      enabled: env.RUN_REPLICATION_KEEP_ALIVE_ENABLED === "1",
-      idleSocketTtl: env.RUN_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
-    },
-    logLevel: env.RUN_REPLICATION_CLICKHOUSE_LOG_LEVEL,
-    compression: {
-      request: true,
-    },
-    maxOpenConnections: env.RUN_REPLICATION_MAX_OPEN_CONNECTIONS,
-  });
-
   const service = new RunsReplicationService({
-    clickhouse: clickhouse,
+    clickhouseFactory,
     pgConnectionUrl: DATABASE_URL,
     serviceName: "runs-replication",
     slotName: env.RUN_REPLICATION_SLOT_NAME,
@@ -72,8 +58,9 @@ function initializeRunsReplicationInstance() {
   });
 
   if (env.RUN_REPLICATION_ENABLED === "1") {
-    service
-      .start()
+    clickhouseFactory
+      .isReady()
+      .then(() => service.start())
       .then(() => {
         console.log("🗃️ Runs replication service started");
       })