get it working

Author: ericallam

.configs/prometheus.yml

@@ -0,0 +1,7 @@
+global:
+  scrape_interval: 15s # how often to scrape targets
+
+scrape_configs:
+  - job_name: "trigger-dev"
+    static_configs:
+      - targets: ["localhost:3030"]

apps/webapp/app/env.server.ts

@@ -725,6 +725,35 @@ const EnvironmentSchema = z.object({
   // BetterStack
   BETTERSTACK_API_KEY: z.string().optional(),
   BETTERSTACK_STATUS_PAGE_ID: z.string().optional(),
+
+  RUN_REPLICATION_REDIS_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_HOST),
+  RUN_REPLICATION_REDIS_READER_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_READER_HOST),
+  RUN_REPLICATION_REDIS_READER_PORT: z.coerce
+    .number()
+    .optional()
+    .transform(
+      (v) =>
+        v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
+    ),
+  RUN_REPLICATION_REDIS_PORT: z.coerce
+    .number()
+    .optional()
+    .transform((v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)),
+  RUN_REPLICATION_REDIS_USERNAME: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_USERNAME),
+  RUN_REPLICATION_REDIS_PASSWORD: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_PASSWORD),
+  RUN_REPLICATION_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/metrics.server.ts

@@ -4,7 +4,9 @@ import { env } from "./env.server";
 
 export const metricsRegister = singleton("metricsRegister", initializeMetricsRegister);
 
-function initializeMetricsRegister() {
+export type MetricsRegister = Registry<OpenMetricsContentType>;
+
+function initializeMetricsRegister(): MetricsRegister {
   const registry = new Registry<OpenMetricsContentType>();
 
   register.setDefaultLabels({

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

@@ -0,0 +1,37 @@
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { prisma } from "~/db.server";
+import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { runsReplicationInstance } from "~/services/runsReplicationInstance.server";
+
+export async function action({ request }: ActionFunctionArgs) {
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequestWithPersonalAccessToken(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const user = await prisma.user.findUnique({
+    where: {
+      id: authenticationResult.userId,
+    },
+  });
+
+  if (!user) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  if (!user.admin) {
+    return json({ error: "You must be an admin to perform this action" }, { status: 403 });
+  }
+
+  try {
+    await runsReplicationInstance.start();
+
+    return json({
+      success: true,
+    });
+  } catch (error) {
+    return json({ error: error instanceof Error ? error.message : error }, { status: 400 });
+  }
+}

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

@@ -0,0 +1,37 @@
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { prisma } from "~/db.server";
+import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { runsReplicationInstance } from "~/services/runsReplicationInstance.server";
+
+export async function action({ request }: ActionFunctionArgs) {
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequestWithPersonalAccessToken(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const user = await prisma.user.findUnique({
+    where: {
+      id: authenticationResult.userId,
+    },
+  });
+
+  if (!user) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  if (!user.admin) {
+    return json({ error: "You must be an admin to perform this action" }, { status: 403 });
+  }
+
+  try {
+    await runsReplicationInstance.stop();
+
+    return json({
+      success: true,
+    });
+  } catch (error) {
+    return json({ error: error instanceof Error ? error.message : error }, { status: 400 });
+  }
+}

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

@@ -0,0 +1,37 @@
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { prisma } from "~/db.server";
+import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { runsReplicationInstance } from "~/services/runsReplicationInstance.server";
+
+export async function action({ request }: ActionFunctionArgs) {
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequestWithPersonalAccessToken(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const user = await prisma.user.findUnique({
+    where: {
+      id: authenticationResult.userId,
+    },
+  });
+
+  if (!user) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  if (!user.admin) {
+    return json({ error: "You must be an admin to perform this action" }, { status: 403 });
+  }
+
+  try {
+    await runsReplicationInstance.teardown();
+
+    return json({
+      success: true,
+    });
+  } catch (error) {
+    return json({ error: error instanceof Error ? error.message : error }, { status: 400 });
+  }
+}

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

@@ -0,0 +1,38 @@
+import { ClickHouse } from "@internal/clickhouse";
+import { RunsReplicationService } from "./runsReplicationService.server";
+import { singleton } from "~/utils/singleton";
+import invariant from "tiny-invariant";
+import { env } from "~/env.server";
+import { metricsRegister } from "~/metrics.server";
+
+export const runsReplicationInstance = singleton(
+  "runsReplicationInstance",
+  initializeRunsReplicationInstance
+);
+
+function initializeRunsReplicationInstance() {
+  const { DATABASE_URL } = process.env;
+  invariant(typeof DATABASE_URL === "string", "DATABASE_URL env var not set");
+
+  const clickhouse = ClickHouse.fromEnv();
+
+  const service = new RunsReplicationService({
+    clickhouse: clickhouse,
+    pgConnectionUrl: DATABASE_URL,
+    serviceName: "runs-replication",
+    slotName: "task_runs_to_clickhouse_v1",
+    publicationName: "task_runs_to_clickhouse_v1_publication",
+    redisOptions: {
+      keyPrefix: "runs-replication:",
+      port: env.RUN_REPLICATION_REDIS_PORT ?? undefined,
+      host: env.RUN_REPLICATION_REDIS_HOST ?? undefined,
+      username: env.RUN_REPLICATION_REDIS_USERNAME ?? undefined,
+      password: env.RUN_REPLICATION_REDIS_PASSWORD ?? undefined,
+      enableAutoPipelining: true,
+      ...(env.RUN_REPLICATION_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+    },
+    metricsRegister: metricsRegister,
+  });
+
+  return service;
+}