Effect cluster deployment on staging

apps/web-cluster/Dockerfile

@@ -0,0 +1,28 @@
+FROM node:24-slim AS base
+RUN corepack enable
+
+FROM base AS builder
+WORKDIR /app
+COPY . .
+
+RUN corepack enable pnpm
+
+RUN echo "inject-workspace-packages=true" >> .npmrc
+RUN --mount=type=cache,id=pnpm,target=/root/.local/share/pnpm/store pnpm i
+
+RUN pnpm run --filter=@cap/web-cluster build
+RUN pnpm deploy --filter=@cap/web-cluster out
+RUN cd out && node scripts/post-deploy.ts
+
+FROM denoland/deno:2.5.3 AS runner
+WORKDIR /app
+
+COPY --from=builder --chown=deno:deno /app/out /app
+
+USER deno
+
+ENTRYPOINT ["deno", "run", "--allow-all"]
+
+EXPOSE 8080
+EXPOSE 42069
+EXPOSE 42169

apps/web-cluster/package.json

@@ -0,0 +1,31 @@
+{
+	"name": "@cap/web-cluster",
+	"type": "module",
+	"scripts": {
+		"dev": "pnpm dotenv -e ../../.env -- concurrently \"deno run --allow-all --watch ./src/runner/index.ts\" \"deno run --allow-all --watch ./src/shard-manager.ts\"",
+		"build": "pnpm run --filter @cap/web-cluster^... build",
+		"build:docker": "cd ../.. && docker build -f apps/web-cluster/Dockerfile -t ghcr.io/brendonovich/cap-web-cluster:latest ."
+	},
+	"dependencies": {
+		"@cap/web-backend": "workspace:*",
+		"@cap/web-domain": "workspace:*",
+		"@effect/cluster": "^0.49.1",
+		"@effect/opentelemetry": "^0.56.1",
+		"@effect/platform": "^0.91.1",
+		"@effect/platform-node": "^0.96.1",
+		"@effect/platform-node-shared": "^0.51.4",
+		"@effect/rpc": "^0.70.0",
+		"@effect/sql-mysql2": "^0.45.1",
+		"@effect/workflow": "^0.10.0",
+		"@opentelemetry/exporter-trace-otlp-http": "^0.203.0",
+		"@opentelemetry/sdk-trace-base": "^2.0.1",
+		"effect": "^3.17.14",
+		"rolldown": "1.0.0-beta.42",
+		"tsdown": "^0.15.6",
+		"tsup": "^8.5.0"
+	},
+	"devDependencies": {
+		"concurrently": "^9.2.1",
+		"dotenv-cli": "^10.0.0"
+	}
+}

apps/web-cluster/scripts/post-deploy.ts

@@ -0,0 +1,24 @@
+import { FileSystem } from "@effect/platform";
+import { NodeContext, NodeRuntime } from "@effect/platform-node";
+import { Effect } from "effect";
+
+Effect.gen(function* () {
+	const fs = yield* FileSystem.FileSystem;
+
+	const dotPnpm = "./node_modules/.pnpm";
+	const deps = yield* fs.readDirectory(dotPnpm);
+	const capDeps = deps.filter((dep) => dep.startsWith("@cap"));
+
+	for (const key of capDeps) {
+		const pkgName = key.split("@file")[0].replace("+", "/");
+		const pkgJsonPath = `${dotPnpm}/${key}/node_modules/${pkgName}/package.json`;
+
+		let pkgJson = JSON.parse(yield* fs.readFileString(pkgJsonPath));
+
+		if (pkgJson.publishConfig) {
+			pkgJson = { ...pkgJson, ...pkgJson.publishConfig };
+		}
+
+		yield* fs.writeFileString(pkgJsonPath, JSON.stringify(pkgJson));
+	}
+}).pipe(Effect.provide(NodeContext.layer), NodeRuntime.runMain);

apps/web-cluster/src/cluster/container-metadata.ts

@@ -0,0 +1,66 @@
+import { Config, Data, Effect, Option } from "effect";
+
+export class FetchIpError extends Data.TaggedError("FetchIpError")<{}> {}
+
+class EcsContainerMetadata extends Effect.Service<EcsContainerMetadata>()(
+	"EcsContainerMetadata",
+	{
+		effect: Effect.gen(function* () {
+			return {
+				metadataUri: yield* Config.option(
+					Config.string("ECS_CONTAINER_METADATA_URI_V4"),
+				),
+			};
+		}),
+	},
+) {}
+
+export const privateIp = EcsContainerMetadata.pipe(
+	Effect.flatMap(({ metadataUri }) =>
+		Option.match(metadataUri, {
+			onNone: () => Effect.succeed("0.0.0.0"),
+			onSome: (uri) =>
+				Effect.tryPromise({
+					try: async () => {
+						const response = await fetch(`${uri}/task`);
+						const data = await response.json();
+						return data.Containers[0].Networks[0].IPv4Addresses[0] as string;
+					},
+					catch: (error) => {
+						console.error("error", error);
+						return new FetchIpError();
+					},
+				}),
+		}),
+	),
+);
+
+export class ContainerMetadata extends Effect.Service<ContainerMetadata>()(
+	"ContainerMetadata",
+	{
+		effect: Effect.gen(function* () {
+			const containerMetadata = yield* EcsContainerMetadata;
+			const metadataUri = containerMetadata.metadataUri;
+			const ipAddress = yield* Option.match(metadataUri, {
+				onNone: () => Effect.succeed("0.0.0.0"),
+				onSome: (uri) =>
+					Effect.tryPromise({
+						try: async () => {
+							const response = await fetch(`${uri}/task`);
+							const data = await response.json();
+							return data.Containers[0].Networks[0].IPv4Addresses[0] as string;
+						},
+						catch: (error) => {
+							console.error("error", error);
+							return new FetchIpError();
+						},
+					}),
+			});
+
+			const port = yield* Config.number("PORT").pipe(Config.withDefault(42069));
+
+			return { ipAddress, port };
+		}),
+		dependencies: [EcsContainerMetadata.Default],
+	},
+) {}

apps/web-cluster/src/health-check.ts

@@ -0,0 +1,27 @@
+async function checkHealth(): Promise<boolean> {
+	try {
+		const response = await fetch(
+			`http://127.0.0.1:${process.env.HEALTH_CHECK_PORT}/health`,
+		);
+		return response.status === 200;
+	} catch (error) {
+		console.error("Health check failed:", error);
+		return false;
+	}
+}
+
+// Run the health check
+checkHealth()
+	.then((isHealthy) => {
+		if (isHealthy) {
+			console.log("✅ Service is healthy (200 OK)");
+			process.exit(0);
+		} else {
+			console.log("❌ Service is not healthy (non-200 response)");
+			process.exit(1);
+		}
+	})
+	.catch((error) => {
+		console.error("❌ Health check failed:", error);
+		process.exit(1);
+	});

apps/web-cluster/src/runner/health-server.ts

@@ -0,0 +1,22 @@
+import { createServer } from "node:http";
+import { HttpRouter, HttpServer, HttpServerResponse } from "@effect/platform";
+import { NodeHttpServer } from "@effect/platform-node";
+import { Layer } from "effect";
+
+// Define the router with a single route for the root URL
+const router = HttpRouter.empty.pipe(
+	HttpRouter.get("/health", HttpServerResponse.text("ok")),
+);
+
+// Set up the application server
+const app = router.pipe(HttpServer.serve());
+
+// Specify the port
+const port = process.env.HEALTH_CHECK_PORT
+	? parseInt(process.env.HEALTH_CHECK_PORT)
+	: 3000;
+
+// Create a server layer with the specified port
+const ServerLive = NodeHttpServer.layer(() => createServer(), { port });
+
+export const HealthServerLive = Layer.provide(app, ServerLive);

apps/web-cluster/src/runner/index.ts

@@ -0,0 +1,99 @@
+import { createServer } from "node:http";
+import { Database, S3Buckets, Videos, Workflows } from "@cap/web-backend";
+import { ClusterWorkflowEngine, RunnerAddress } from "@effect/cluster";
+import * as NodeSdk from "@effect/opentelemetry/NodeSdk";
+import {
+	FetchHttpClient,
+	HttpApiBuilder,
+	HttpMiddleware,
+	HttpRouter,
+	HttpServer,
+} from "@effect/platform";
+import {
+	NodeClusterRunnerSocket,
+	NodeHttpServer,
+	NodeRuntime,
+} from "@effect/platform-node";
+import { RpcServer } from "@effect/rpc";
+import { WorkflowProxyServer } from "@effect/workflow";
+import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
+import { BatchSpanProcessor } from "@opentelemetry/sdk-trace-base";
+import { Config, Effect, Layer, Option } from "effect";
+
+import { ContainerMetadata } from "../cluster/container-metadata.ts";
+import { DatabaseLive, ShardDatabaseLive } from "../shared/database.ts";
+import { HealthServerLive } from "./health-server.ts";
+
+const ClusterWorkflowLive = Layer.unwrapEffect(
+	Effect.gen(function* () {
+		const containerMeta = yield* ContainerMetadata;
+		return ClusterWorkflowEngine.layer.pipe(
+			Layer.provide(
+				NodeClusterRunnerSocket.layer({
+					storage: "sql",
+					shardingConfig: {
+						runnerAddress: Option.some(
+							RunnerAddress.make(containerMeta.ipAddress, containerMeta.port),
+						),
+					},
+				}),
+			),
+			Layer.provide(ShardDatabaseLive),
+		);
+	}),
+).pipe(Layer.provide(ContainerMetadata.Default));
+
+const RpcsLive = RpcServer.layer(Workflows.RpcGroup).pipe(
+	Layer.provide(WorkflowProxyServer.layerRpcHandlers(Workflows.Workflows)),
+	Layer.provide(Workflows.WorkflowsLayer),
+	Layer.provide(ClusterWorkflowLive),
+	Layer.provide(RpcServer.layerProtocolHttp({ path: "/" })),
+	Layer.provide(Workflows.RpcSerialization),
+);
+
+const TracingLayer = Layer.unwrapEffect(
+	Effect.gen(function* () {
+		const exporter = Option.match(
+			yield* Config.option(
+				Config.all([
+					Config.string("AXIOM_API_TOKEN"),
+					Config.string("AXIOM_DOMAIN"),
+					Config.string("AXIOM_DATASET"),
+				]),
+			),
+			{
+				onNone: () => new OTLPTraceExporter({}),
+				onSome: ([token, domain, dataset]) => {
+					return new OTLPTraceExporter({
+						url: `https://${domain}/v1/traces`, // Axiom API endpoint for trace data
+						headers: {
+							Authorization: `Bearer ${token}`, // Replace API_TOKEN with your actual API token
+							"X-Axiom-Dataset": dataset, // Replace DATASET_NAME with your dataset
+						},
+					});
+				},
+			},
+		);
+
+		return NodeSdk.layer(() => ({
+			resource: { serviceName: "cap-workflow-runner" },
+			spanProcessor: [new BatchSpanProcessor(exporter)],
+		}));
+	}),
+);
+
+HttpRouter.Default.serve().pipe(
+	Layer.provide(RpcsLive),
+	HttpServer.withLogAddress,
+	Layer.provide(NodeHttpServer.layer(createServer, { port: 42169 })),
+	Layer.provide(Videos.Default),
+	Layer.provide(S3Buckets.Default),
+	Layer.provide(Database.Default),
+	Layer.provide(FetchHttpClient.layer),
+	Layer.provide(DatabaseLive),
+	Layer.provide(TracingLayer),
+	Layer.launch,
+	NodeRuntime.runMain,
+);
+
+HealthServerLive.pipe(Layer.launch, NodeRuntime.runMain);

apps/web-cluster/src/shard-manager.ts

@@ -0,0 +1,17 @@
+import {
+	NodeClusterShardManagerSocket,
+	NodeRuntime,
+} from "@effect/platform-node";
+import { Layer, Logger } from "effect";
+
+import { DatabaseLive, ShardDatabaseLive } from "./shared/database.ts";
+
+NodeClusterShardManagerSocket.layer({
+	storage: "sql",
+}).pipe(
+	Layer.provide(ShardDatabaseLive),
+	Layer.provide(DatabaseLive),
+	Layer.provide(Logger.pretty),
+	Layer.launch,
+	NodeRuntime.runMain,
+);

apps/web-cluster/src/shared/database.ts

@@ -0,0 +1,29 @@
+import { MysqlClient } from "@effect/sql-mysql2";
+import { Config, Effect, Layer, Option } from "effect";
+
+export const DatabaseLive = Layer.unwrapEffect(
+	Effect.gen(function* () {
+		const url = yield* Config.redacted(Config.string("DATABASE_URL"));
+
+		return MysqlClient.layer({ url });
+	}),
+);
+
+export const ShardDatabaseLive = Layer.unwrapEffect(
+	Effect.gen(function* () {
+		const url = yield* Config.option(
+			Config.redacted(Config.string("SHARD_DATABASE_URL")),
+		);
+
+		return yield* Option.match(url, {
+			onNone: () =>
+				Effect.gen(function* () {
+					return Layer.succeed(
+						MysqlClient.MysqlClient,
+						yield* MysqlClient.MysqlClient,
+					);
+				}),
+			onSome: (url) => Effect.succeed(MysqlClient.layer({ url })),
+		});
+	}),
+);

apps/web-cluster/tsconfig.json

@@ -0,0 +1,14 @@
+{
+	"extends": "../../packages/config/base.tsconfig.json",
+	"exclude": ["node_modules", "dist"],
+	"include": ["**/*.ts", "**/*.tsx"],
+	"compilerOptions": {
+		"moduleResolution": "bundler",
+		"composite": true,
+		"outDir": "dist",
+		"noEmit": false,
+		"emitDeclarationOnly": true,
+		"module": "es2020",
+		"target": "es2020"
+	}
+}

apps/web/app/api/[[...route]]/route.ts

@@ -0,0 +1,13 @@
+import { HttpLive } from "@cap/web-backend";
+import { HttpApiScalar } from "@effect/platform";
+import { Layer } from "effect";
+import { apiToHandler } from "@/lib/server";
+
+const handler = apiToHandler(
+	HttpApiScalar.layer({ path: "/api" }).pipe(Layer.provideMerge(HttpLive)),
+);
+
+export const GET = handler;
+export const POST = handler;
+export const HEAD = handler;
+export const OPTIONS = handler;

apps/web/app/layout.tsx

@@ -1,7 +1,6 @@
 import "@/app/globals.css";
 import { getCurrentUser } from "@cap/database/auth/session";
 import { buildEnv } from "@cap/env";
-import { S3_BUCKET_URL } from "@cap/utils";
 import { Analytics as DubAnalytics } from "@dub/analytics/react";
 import * as TooltipPrimitive from "@radix-ui/react-tooltip";
 import type { Metadata } from "next";