1import * as Alchemy from "alchemy";
2import * as Axiom from "alchemy/Axiom";
3import * as Output from "alchemy/Output";
4import * as Cause from "effect/Cause";
5import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";
7import * as Option from "effect/Option";
8import * as Schema from "effect/Schema";
9import * as Tracer from "effect/Tracer";
10
11import { relayResourceNameForStage } from "./deploymentConfig.ts";
12
13const relayRecentSpansQuery = (dataset: string) =>
14 [
15 `['${dataset}']`,
16 `| where isnotnull(span_id) or isnotnull(trace_id)`,
17 `| 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({}))`,
18 `| extend userId = customAttributes['user']['id']`,
19 `| project _time, name, trace_id, span_id, duration, requestMethod, path, statusCode, endpoint, userId`,
20 `| order by _time desc`,
21 `| limit 200`,
22 ].join("\n");
23
24export const RelayObservability = Effect.gen(function* () {
25 const { stage } = yield* Alchemy.Stack;
26 const traces = yield* Axiom.Dataset("RelayTracesDataset", {
27 name: relayResourceNameForStage("t3-code-relay-traces", stage),
28 kind: "otel:traces:v1",
29 description: "T3 Code relay Worker HTTP request spans.",
30 retentionDays: 30,
31 useRetentionPeriod: true,
32 });
33
34 const workerIngestToken = yield* Axiom.ApiToken("RelayWorkerAxiomIngestToken", {
35 name: relayResourceNameForStage("t3-code-relay-otel-ingest", stage),
36 description: "Owned by Alchemy. Scoped OTLP ingest token for relay HTTP spans.",
37 datasetCapabilities: Output.map(traces.name, (dataset) => ({
38 [dataset]: { ingest: ["create" as const] },
39 })),
40 });
41
42 const mobileIngestToken = yield* Axiom.ApiToken("RelayMobileAxiomIngestToken", {
43 name: relayResourceNameForStage("t3-code-mobile-otel-ingest", stage),
44 description: "Owned by Alchemy. Scoped OTLP ingest token for T3 Code mobile spans.",
45 datasetCapabilities: Output.map(traces.name, (dataset) => ({
46 [dataset]: { ingest: ["create" as const] },
47 })),
48 });
49
50 const clientIngestToken = yield* Axiom.ApiToken("RelayClientAxiomIngestToken", {
51 name: relayResourceNameForStage("t3-code-relay-client-otel-ingest", stage),
52 description: "Owned by Alchemy. Scoped OTLP ingest token for first-party relay client spans.",
53 datasetCapabilities: Output.map(traces.name, (dataset) => ({
54 [dataset]: { ingest: ["create" as const] },
55 })),
56 });
57
58 yield* Axiom.View("RelayRecentSpansView", {
59 name: relayResourceNameForStage("t3-code-relay-recent-spans", stage),
60 description: "Recent relay HTTP request spans.",
61 datasets: [traces.name],
62 aplQuery: Output.map(traces.name, relayRecentSpansQuery),
63 });
64
65 return { traces, workerIngestToken, mobileIngestToken, clientIngestToken } as const;
66});
67
68export const withSpanAttributes =
69 (attributes: Record<string, unknown>) =>
70 <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
71 Effect.annotateCurrentSpan(attributes).pipe(
72 Effect.andThen(effect.pipe(Effect.annotateSpans(attributes))),
73 );
74
75const appendEncodedAttributes = (
76 attributes: Record<string, unknown>,
77 prefix: string,
78 value: unknown,
79): void => {
80 if (
81 value === null ||
82 typeof value === "string" ||
83 typeof value === "number" ||
84 typeof value === "boolean" ||
85 typeof value === "bigint" ||
86 Array.isArray(value)
87 ) {
88 attributes[prefix] = value;
89 return;
90 }
91 if (typeof value !== "object") {
92 return;
93 }
94 for (const [key, child] of Object.entries(value)) {
95 appendEncodedAttributes(attributes, `${prefix}.${key}`, child);
96 }
97};
98
99const schemaErrorAttributes = (error: unknown): Record<string, unknown> | undefined => {
100 if (typeof error !== "object" || error === null) {
101 return undefined;
102 }
103 const constructor = error.constructor;
104 if (!Schema.isSchema(constructor)) {
105 return undefined;
106 }
107 const encoded = Schema.encodeUnknownOption(constructor as unknown as Schema.Encoder<unknown>)(
108 error,
109 );
110 if (Option.isNone(encoded) || typeof encoded.value !== "object" || encoded.value === null) {
111 return undefined;
112 }
113 const tag = Reflect.get(encoded.value, "_tag");
114 if (typeof tag !== "string") {
115 return undefined;
116 }
117
118 const attributes: Record<string, unknown> = {
119 "error.type": tag,
120 };
121 for (const [key, value] of Object.entries(encoded.value)) {
122 if (key !== "_tag") {
123 appendEncodedAttributes(attributes, `error.${key}`, value);
124 }
125 }
126 return attributes;
127};
128
129const annotateSchemaError = (span: Tracer.Span, exit: Exit.Exit<unknown, unknown>): void => {
130 if (Exit.isSuccess(exit)) {
131 return;
132 }
133 for (const reason of exit.cause.reasons) {
134 const error = Cause.isFailReason(reason)
135 ? reason.error
136 : Cause.isDieReason(reason)
137 ? reason.defect
138 : undefined;
139 const attributes = schemaErrorAttributes(error);
140 if (attributes) {
141 for (const [key, value] of Object.entries(attributes)) {
142 span.attribute(key, value);
143 }
144 return;
145 }
146 }
147};
148
149class RelayTraceSpan implements Tracer.Span {
150 readonly _tag = "Span";
151 private readonly delegate: Tracer.Span;
152
153 constructor(delegate: Tracer.Span) {
154 this.delegate = delegate;
155 }
156
157 get name() {
158 return this.delegate.name;
159 }
160 get spanId() {
161 return this.delegate.spanId;
162 }
163 get traceId() {
164 return this.delegate.traceId;
165 }
166 get parent() {
167 return this.delegate.parent;
168 }
169 get annotations() {
170 return this.delegate.annotations;
171 }
172 get status() {
173 return this.delegate.status;
174 }
175 get attributes() {
176 return this.delegate.attributes;
177 }
178 get links() {
179 return this.delegate.links;
180 }
181 get sampled() {
182 return this.delegate.sampled;
183 }
184 get kind() {
185 return this.delegate.kind;
186 }
187
188 end(endTime: bigint, exit: Exit.Exit<unknown, unknown>): void {
189 annotateSchemaError(this.delegate, exit);
190 this.delegate.end(endTime, exit);
191 }
192
193 attribute(key: string, value: unknown): void {
194 this.delegate.attribute(key, value);
195 }
196
197 event(name: string, startTime: bigint, attributes?: Record<string, unknown>): void {
198 this.delegate.event(name, startTime, attributes);
199 }
200
201 addLinks(links: ReadonlyArray<Tracer.SpanLink>): void {
202 this.delegate.addLinks(links);
203 }
204}
205
206const withSchemaErrorAttributes = (delegate: Tracer.Tracer): Tracer.Tracer =>
207 Tracer.make({
208 span: (options) => new RelayTraceSpan(delegate.span(options)),
209 ...(delegate.context ? { context: delegate.context } : {}),
210 });
211
212/**
213 * Adds a failed span's schema error fields (`error.type`, `error.<field>`) to
214 * the span, on whichever tracer is current. Provide it around the work whose
215 * spans should carry them; the Worker's telemetry still owns export.
216 */
217export const withSchemaErrorSpanAttributes = <A, E, R>(
218 effect: Effect.Effect<A, E, R>,
219): Effect.Effect<A, E, R> =>
220 Effect.flatMap(Effect.tracer, (tracer) =>
221 effect.pipe(Effect.withTracer(withSchemaErrorAttributes(tracer))),
222 );