1import * as Cloudflare from "alchemy/Cloudflare";
2import { Stack } from "alchemy/Stack";
3import * as Effect from "effect/Effect";
4import * as Redacted from "effect/Redacted";
5import * as HttpServerRequest from "effect/http/HttpServerRequest";
6import * as HttpServerResponse from "effect/http/HttpServerResponse";
7import { Logs, Metrics, Traces } from "./Datasets.ts";
8import { IngestToken } from "./IngestToken.ts";
9
10/**
11 * Public ingest relay used by both:
12 *
13 * 1. The alchemy CLI / services posting OTLP/JSON to `/v1/{traces,logs,metrics}`,
14 * which we forward to Axiom with the ingest bearer token attached server-side.
15 * 2. The website's PostHog browser SDK, which posts to anything else
16 * (`/e`, `/s`, `/flags`, `/decide`, `/static/*`, ...). PostHog ingest needs
17 * no server secret, so we just reverse-proxy to PostHog Cloud (US region).
18 *
19 * Bound to two custom domains in prod:
20 * - `otel.alchemy.run` — primary OTLP entrypoint
21 * - `analytics.alchemy.run` — first-party PostHog ingest (defeats ad-blockers)
22 *
23 * Environment (set by `stacks/otel.ts`):
24 * - `AXIOM_TRACES_ENDPOINT` — full Axiom OTLP traces URL
25 * - `AXIOM_LOGS_ENDPOINT` — full Axiom OTLP logs URL
26 * - `AXIOM_METRICS_ENDPOINT` — full Axiom OTLP metrics URL
27 * - `AXIOM_INGEST_TOKEN` — Bearer token (Redacted)
28 */
29export default class Ingester extends Cloudflare.Worker<Ingester>()(
30 "OtelWorker",
31 Stack.useSync(({ stage }) => ({
32 main: import.meta.url,
33 observability: { enabled: true },
34 domain:
35 stage === "prod"
36 ? { name: "otel.alchemy.run", aliases: ["analytics.alchemy.run"] }
37 : undefined,
38 })),
39 Effect.gen(function* () {
40 const tokenValue = yield* (yield* IngestToken).token;
41 const traces = yield* Traces;
42 const logs = yield* Logs;
43 const metrics = yield* Metrics;
44 const tracesEndpoint = yield* traces.otelTracesEndpoint;
45 const logsEndpoint = yield* logs.otelLogsEndpoint;
46 const metricsEndpoint = yield* metrics.otelMetricsEndpoint;
47 const tracesDataset = yield* traces.name;
48 const logsDataset = yield* logs.name;
49 const metricsDataset = yield* metrics.name;
50
51 return {
52 fetch: Effect.gen(function* () {
53 const request = yield* HttpServerRequest.HttpServerRequest;
54 const url = new URL(request.url, "http://x");
55 const path = url.pathname;
56
57 // 1. OTLP → Axiom (POST /v1/{traces,logs,metrics})
58 const otlp =
59 request.method === "POST"
60 ? path === "/v1/traces"
61 ? {
62 endpoint: yield* tracesEndpoint,
63 dataset: yield* tracesDataset,
64 }
65 : path === "/v1/logs"
66 ? { endpoint: yield* logsEndpoint, dataset: yield* logsDataset }
67 : path === "/v1/metrics"
68 ? {
69 endpoint: yield* metricsEndpoint,
70 dataset: yield* metricsDataset,
71 }
72 : undefined
73 : undefined;
74
75 if (otlp) {
76 const tokenRaw = yield* tokenValue.pipe(Effect.map(Redacted.value));
77 const token = Redacted.isRedacted(tokenRaw)
78 ? Redacted.value(tokenRaw)
79 : (tokenRaw as string);
80
81 const body = yield* request.arrayBuffer;
82
83 const response = yield* Effect.tryPromise({
84 try: () =>
85 fetch(otlp.endpoint, {
86 method: "POST",
87 headers: {
88 "content-type":
89 request.headers["content-type"] ?? "application/json",
90 authorization: `Bearer ${token}`,
91 "x-axiom-dataset": otlp.dataset,
92 },
93 body,
94 }),
95 catch: (e) => (e instanceof Error ? e : new Error(String(e))),
96 });
97
98 return HttpServerResponse.fromWeb(response);
99 }
100
101 // 2. Everything else → PostHog Cloud (US region)
102 // /static/* and /array/* live on the assets host; everything else on the
103 // ingest host. Static assets are cacheable at the edge.
104 const isAsset =
105 path.startsWith("/static/") || path.startsWith("/array/");
106 const upstreamHost = isAsset
107 ? "https://us-assets.i.posthog.com"
108 : "https://us.i.posthog.com";
109 const target = `${upstreamHost}${path}${url.search}`;
110
111 const raw = yield* HttpServerRequest.toWeb(request);
112
113 // Drop hop-by-hop / Cloudflare-internal headers; preserve everything
114 // else (content-type, accept, user-agent, x-forwarded-for, etc.).
115 const headers = new Headers(raw.headers);
116 headers.delete("host");
117 headers.delete("cookie");
118 for (const key of [...headers.keys()]) {
119 if (key.startsWith("cf-")) headers.delete(key);
120 }
121 const cfIp =
122 raw.headers.get("cf-connecting-ip") ??
123 raw.headers.get("x-real-ip") ??
124 undefined;
125 if (cfIp) headers.set("x-forwarded-for", cfIp);
126
127 const upstream = yield* Effect.tryPromise({
128 try: () =>
129 fetch(target, {
130 method: raw.method,
131 headers,
132 body:
133 raw.method === "GET" || raw.method === "HEAD" ? null : raw.body,
134 redirect: "manual",
135 ...(isAsset
136 ? // Cache the SDK loader / array bundles at the edge for an hour.
137 ({ cf: { cacheTtl: 3600, cacheEverything: true } } as any)
138 : {}),
139 }),
140 catch: (e) => (e instanceof Error ? e : new Error(String(e))),
141 });
142
143 return HttpServerResponse.fromWeb(upstream);
144 }).pipe(
145 Effect.catch((err) =>
146 Effect.succeed(
147 HttpServerResponse.text(`Relay error: ${err.message}`, {
148 status: 502,
149 }),
150 ),
151 ),
152 ),
153 };
154 }),
155) {}