1// @effect-diagnostics nodeBuiltinImport:off - one sha256 inside a synchronous Output.map.
2import * as NodeCrypto from "node:crypto";
3
4import * as Alchemy from "alchemy";
5import * as Axiom from "@alchemy.run/axiom";
6import * as Cloudflare from "@alchemy.run/cloudflare";
7import * as Output from "alchemy/Output";
8import * as Cause from "effect/Cause";
9import * as Effect from "effect/Effect";
10import * as Exit from "effect/Exit";
11import * as Option from "effect/Option";
12import * as Redacted from "effect/Redacted";
13import * as Schema from "effect/Schema";
14import * as Tracer from "effect/Tracer";
15
16import { relayResourceNameForStage } from "./deploymentConfig.ts";
17import { MobileMonitoring } from "./mobileTelemetry/monitoring.ts";
18import { RelayMonitoring } from "./monitoring.ts";
19
20const relayRecentSpansQuery = (dataset: string) =>
21 [
22 `['${dataset}']`,
23 `| where isnotnull(span_id) or isnotnull(trace_id)`,
24 `| extend requestMethod = column_ifexists('attributes.http.request.method', ''), path = column_ifexists('attributes.url.path', ''), endpoint = column_ifexists('attributes.http.route', ''), statusCode = column_ifexists('attributes.http.response.status_code', 0), customAttributes = column_ifexists('attributes.custom', dynamic({}))`,
25 `| extend userId = customAttributes['user']['id']`,
26 `| project _time, name, trace_id, span_id, duration, requestMethod, path, statusCode, endpoint, userId`,
27 `| order by _time desc`,
28 `| limit 200`,
29 ].join("\n");
30
31export const RelayObservability = Effect.gen(function* () {
32 const { stage } = yield* Alchemy.Stack;
33 const traces = yield* Axiom.Dataset("RelayTracesDataset", {
34 name: relayResourceNameForStage("t3-code-relay-traces", stage),
35 kind: "otel:traces:v1",
36 description: "T3 Code relay Worker HTTP request spans.",
37 retentionDays: 30,
38 useRetentionPeriod: true,
39 });
40
41 const workerIngestToken = yield* Axiom.ApiToken("RelayWorkerAxiomIngestToken", {
42 name: relayResourceNameForStage("t3-code-relay-otel-ingest", stage),
43 description: "Owned by Alchemy. Scoped OTLP ingest token for relay HTTP spans.",
44 datasetCapabilities: Output.map(traces.name, (dataset) => ({
45 [dataset]: { ingest: ["create" as const] },
46 })),
47 });
48
49 const mobileIngestToken = yield* Axiom.ApiToken("RelayMobileAxiomIngestToken", {
50 name: relayResourceNameForStage("t3-code-mobile-otel-ingest", stage),
51 description: "Owned by Alchemy. Scoped OTLP ingest token for T3 Code mobile spans.",
52 datasetCapabilities: Output.map(traces.name, (dataset) => ({
53 [dataset]: { ingest: ["create" as const] },
54 })),
55 });
56
57 const clientIngestToken = yield* Axiom.ApiToken("RelayClientAxiomIngestToken", {
58 name: relayResourceNameForStage("t3-code-relay-client-otel-ingest", stage),
59 description: "Owned by Alchemy. Scoped OTLP ingest token for first-party relay client spans.",
60 datasetCapabilities: Output.map(traces.name, (dataset) => ({
61 [dataset]: { ingest: ["create" as const] },
62 })),
63 });
64
65 // expo-observe data, forwarded by src/mobileTelemetry. Mobile spans share the traces dataset.
66 const mobileLogs = yield* Axiom.Dataset("MobileLogsDataset", {
67 name: relayResourceNameForStage("t3-code-mobile-logs", stage),
68 kind: "otel:logs:v1",
69 description: "T3 Code mobile app errors, crashes, and log events from expo-observe.",
70 retentionDays: 30,
71 useRetentionPeriod: true,
72 });
73 const mobileMetrics = yield* Axiom.Dataset("MobileMetricsDataset", {
74 name: relayResourceNameForStage("t3-code-mobile-metrics", stage),
75 kind: "otel:metrics:v1",
76 description: "T3 Code mobile app startup and update metrics from expo-observe.",
77 retentionDays: 30,
78 useRetentionPeriod: true,
79 });
80 const mobileTelemetryIngestToken = yield* Axiom.ApiToken("MobileTelemetryAxiomIngestToken", {
81 name: relayResourceNameForStage("t3-code-mobile-telemetry-ingest", stage),
82 description: "Owned by Alchemy. Scoped ingest token the relay forwards mobile telemetry with.",
83 datasetCapabilities: Output.map(
84 Output.all(traces.name, mobileLogs.name, mobileMetrics.name),
85 (datasets) =>
86 Object.fromEntries(datasets.map((dataset) => [dataset, { ingest: ["create" as const] }])),
87 ),
88 });
89
90 const mobileSourceMaps = yield* Cloudflare.R2.Bucket("MobileSourceMapsBucket", {
91 name: mobileSourceMapsBucketName(stage),
92 // R2 refuses to delete a non-empty bucket, which would half tear down every non-prod stage.
93 forceDestroy: stage !== "prod",
94 });
95
96 yield* Axiom.View("RelayRecentSpansView", {
97 name: relayResourceNameForStage("t3-code-relay-recent-spans", stage),
98 description: "Recent relay HTTP request spans.",
99 datasets: [traces.name],
100 aplQuery: Output.map(traces.name, relayRecentSpansQuery),
101 });
102
103 yield* RelayMonitoring(traces);
104 yield* MobileMonitoring(mobileLogs, traces);
105
106 return {
107 traces,
108 workerIngestToken,
109 mobileIngestToken,
110 clientIngestToken,
111 mobileLogs,
112 mobileMetrics,
113 mobileTelemetryIngestToken,
114 mobileSourceMaps,
115 } as const;
116});
117
118const mobileSourceMapsBucketName = (stage: string) =>
119 relayResourceNameForStage("t3-code-mobile-sourcemaps", stage);
120
121/**
122 * Write-only S3 credentials for the source maps bucket: the `MOBILE_SOURCEMAPS_R2` secret.
123 * R2's S3 access key id is the API token's id and the secret the SHA-256 of its value.
124 */
125export const MobileSourceMapsUploadCredentials = Effect.gen(function* () {
126 const { stage } = yield* Alchemy.Stack;
127 const { accountId } = yield* yield* Cloudflare.CloudflareEnvironment;
128 const bucketName = mobileSourceMapsBucketName(stage);
129 const token = yield* Cloudflare.ApiToken.AccountApiToken("MobileSourceMapsUploadToken", {
130 name: relayResourceNameForStage("t3-code-mobile-sourcemaps-upload", stage),
131 policies: [
132 {
133 effect: "allow",
134 permissionGroups: ["Workers R2 Storage Bucket Item Write"],
135 resources: { [`com.cloudflare.edge.r2.bucket.${accountId}_default_${bucketName}`]: "*" },
136 },
137 ],
138 });
139 return Output.map(Output.all(token.tokenId, token.value), ([tokenId, value]) =>
140 Redacted.make(
141 JSON.stringify({
142 endpoint: `https://${accountId}.r2.cloudflarestorage.com`,
143 bucketName,
144 accessKeyId: tokenId,
145 secretAccessKey: NodeCrypto.createHash("sha256")
146 .update(Redacted.value(value))
147 .digest("hex"),
148 }),
149 ),
150 );
151});
152
153export const withSpanAttributes =
154 (attributes: Record<string, unknown>) =>
155 <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
156 Effect.annotateCurrentSpan(attributes).pipe(
157 Effect.andThen(effect.pipe(Effect.annotateSpans(attributes))),
158 );
159
160const appendEncodedAttributes = (
161 attributes: Record<string, unknown>,
162 prefix: string,
163 value: unknown,
164): void => {
165 if (
166 value === null ||
167 typeof value === "string" ||
168 typeof value === "number" ||
169 typeof value === "boolean" ||
170 typeof value === "bigint" ||
171 Array.isArray(value)
172 ) {
173 attributes[prefix] = value;
174 return;
175 }
176 if (typeof value !== "object") {
177 return;
178 }
179 for (const [key, child] of Object.entries(value)) {
180 appendEncodedAttributes(attributes, `${prefix}.${key}`, child);
181 }
182};
183
184const schemaErrorAttributes = (error: unknown): Record<string, unknown> | undefined => {
185 if (typeof error !== "object" || error === null) {
186 return undefined;
187 }
188 const constructor = error.constructor;
189 if (!Schema.isSchema(constructor)) {
190 return undefined;
191 }
192 const encoded = Schema.encodeUnknownOption(constructor as unknown as Schema.Encoder<unknown>)(
193 error,
194 );
195 if (Option.isNone(encoded) || typeof encoded.value !== "object" || encoded.value === null) {
196 return undefined;
197 }
198 const tag = Reflect.get(encoded.value, "_tag");
199 if (typeof tag !== "string") {
200 return undefined;
201 }
202
203 const attributes: Record<string, unknown> = {
204 "error.type": tag,
205 };
206 for (const [key, value] of Object.entries(encoded.value)) {
207 if (key !== "_tag") {
208 appendEncodedAttributes(attributes, `error.${key}`, value);
209 }
210 }
211 return attributes;
212};
213
214const annotateSchemaError = (span: Tracer.Span, exit: Exit.Exit<unknown, unknown>): void => {
215 if (Exit.isSuccess(exit)) {
216 return;
217 }
218 for (const reason of exit.cause.reasons) {
219 const error = Cause.isFailReason(reason)
220 ? reason.error
221 : Cause.isDieReason(reason)
222 ? reason.defect
223 : undefined;
224 const attributes = schemaErrorAttributes(error);
225 if (attributes) {
226 for (const [key, value] of Object.entries(attributes)) {
227 span.attribute(key, value);
228 }
229 return;
230 }
231 }
232};
233
234class RelayTraceSpan implements Tracer.Span {
235 readonly _tag = "Span";
236 private readonly delegate: Tracer.Span;
237
238 constructor(delegate: Tracer.Span) {
239 this.delegate = delegate;
240 }
241
242 get name() {
243 return this.delegate.name;
244 }
245 get spanId() {
246 return this.delegate.spanId;
247 }
248 get traceId() {
249 return this.delegate.traceId;
250 }
251 get parent() {
252 return this.delegate.parent;
253 }
254 get annotations() {
255 return this.delegate.annotations;
256 }
257 get status() {
258 return this.delegate.status;
259 }
260 get attributes() {
261 return this.delegate.attributes;
262 }
263 get links() {
264 return this.delegate.links;
265 }
266 get sampled() {
267 return this.delegate.sampled;
268 }
269 get kind() {
270 return this.delegate.kind;
271 }
272
273 end(endTime: bigint, exit: Exit.Exit<unknown, unknown>): void {
274 annotateSchemaError(this.delegate, exit);
275 this.delegate.end(endTime, exit);
276 }
277
278 attribute(key: string, value: unknown): void {
279 this.delegate.attribute(key, value);
280 }
281
282 event(name: string, startTime: bigint, attributes?: Record<string, unknown>): void {
283 this.delegate.event(name, startTime, attributes);
284 }
285
286 addLinks(links: ReadonlyArray<Tracer.SpanLink>): void {
287 this.delegate.addLinks(links);
288 }
289}
290
291const withSchemaErrorAttributes = (delegate: Tracer.Tracer): Tracer.Tracer =>
292 Tracer.make({
293 span: (options) => new RelayTraceSpan(delegate.span(options)),
294 ...(delegate.context ? { context: delegate.context } : {}),
295 });
296
297export const withSchemaErrorSpanAttributes = <A, E, R>(
298 effect: Effect.Effect<A, E, R>,
299): Effect.Effect<A, E, R> =>
300 Effect.flatMap(Effect.tracer, (tracer) =>
301 effect.pipe(Effect.withTracer(withSchemaErrorAttributes(tracer))),
302 );