1import * as Cache from "effect/Cache";
2import * as Context from "effect/Context";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Exit from "effect/Exit";
6import * as Layer from "effect/Layer";
7import * as Option from "effect/Option";
8import * as Redacted from "effect/Redacted";
9import * as Result from "effect/Result";
10import * as Schema from "effect/Schema";
11import * as Semaphore from "effect/Semaphore";
12import * as Stream from "effect/Stream";
13import * as HttpClient from "effect/http/HttpClient";
14import * as HttpClientRequest from "effect/http/HttpClientRequest";
15import type * as HttpServerRequest from "effect/http/HttpServerRequest";
16import * as HttpServerResponse from "effect/http/HttpServerResponse";
17import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
18import { RelayApi, type RelayMobileTelemetrySignal } from "@t3tools/contracts/relay";
19
20import {
21 applySymbolication,
22 bundleSourceMapKey,
23 collectJsExceptions,
24 markJsFatalCrashes,
25 type SourceMapMiss,
26} from "./otlpLogs.ts";
27import { decodeSourceMap, type CompactSourceMap } from "./sourceMap.ts";
28
29/** Observe splits batches the endpoint answers 413 for, so this only bounds one request. */
30export const MOBILE_TELEMETRY_MAX_BODY_BYTES = 2 * 1024 * 1024;
31/** Per client IP. An app flushes a few batches per launch and background. */
32export const MOBILE_TELEMETRY_RATE_LIMIT = { limit: 120, periodSeconds: 60 } as const;
33/** Inside the relay's 9 s request deadline, so a slow Axiom answers as `upstream_timeout`. */
34const UPSTREAM_TIMEOUT = Duration.seconds(7);
35const MAX_UPSTREAM_RESPONSE_BYTES = 64 * 1024;
36/** A decoded production map is about 20 MB; two fit an isolate and cover iOS and Android. */
37const SOURCE_MAP_CACHE_CAPACITY = 2;
38/** Our iOS map is about 12 MB once uploads drop what the relay never reads; larger is skipped unread. */
39const MAX_SOURCE_MAP_BYTES = 16 * 1024 * 1024;
40/** A request can carry thousands of frames, each rewritten to a longer source path. */
41const MAX_SYMBOLICATION_GROWTH = 1024 * 1024;
42/** One app names one bundle, or two across an OTA update; more would evict cached maps. */
43const MAX_SOURCE_MAPS_PER_REQUEST = 2;
44/** A map can land in R2 a moment after its update ships, so a miss is retried. */
45const MISSING_SOURCE_MAP_TTL = Duration.minutes(5);
46const FAILED_SOURCE_MAP_TTL = Duration.seconds(30);
47
48const UnknownFromJsonString = Schema.fromJsonString(Schema.Unknown);
49const decodeJson = Schema.decodeUnknownOption(UnknownFromJsonString);
50const encodeJson = Schema.encodeSync(UnknownFromJsonString);
51const parseJson = (text: string): unknown => Option.getOrUndefined(decodeJson(text));
52
53interface AxiomDestination {
54 /** Axiom's OTLP/HTTP endpoint for the signal, such as `.../v1/logs`. */
55 readonly url: string;
56 readonly dataset: string;
57}
58
59export class MobileTelemetrySettings extends Context.Service<
60 MobileTelemetrySettings,
61 {
62 readonly axiomToken: Redacted.Redacted<string>;
63 readonly destinations: Readonly<Record<RelayMobileTelemetrySignal, AxiomDestination>>;
64 }
65>()("t3code-relay/mobileTelemetry/MobileTelemetry/MobileTelemetrySettings") {}
66
67export class MobileTelemetryRateLimiter extends Context.Service<
68 MobileTelemetryRateLimiter,
69 { readonly allow: (key: string) => Effect.Effect<boolean> }
70>()("t3code-relay/mobileTelemetry/MobileTelemetry/MobileTelemetryRateLimiter") {}
71
72export class SourceMapObjectError extends Schema.TaggedError<SourceMapObjectError>()(
73 "SourceMapObjectError",
74 { key: Schema.String, cause: Schema.Defect() },
75) {}
76
77/** An object in the source maps bucket, read only once its size is known to fit. */
78interface StoredSourceMap {
79 readonly size: number;
80 readonly text: Effect.Effect<string, SourceMapObjectError>;
81}
82
83interface FoundSourceMap {
84 readonly key: string;
85 readonly map: CompactSourceMap;
86}
87
88export class SourceMaps extends Context.Service<
89 SourceMaps,
90 {
91 readonly find: (key: string) => Effect.Effect<FoundSourceMap | SourceMapMiss>;
92 }
93>()("t3code-relay/mobileTelemetry/MobileTelemetry/SourceMaps") {}
94
95export class SourceMapObjects extends Context.Service<
96 SourceMapObjects,
97 {
98 readonly get: (key: string) => Effect.Effect<StoredSourceMap | undefined, SourceMapObjectError>;
99 }
100>()("t3code-relay/mobileTelemetry/MobileTelemetry/SourceMapObjects") {}
101
102const makeSourceMaps = Effect.gen(function* () {
103 const objects = yield* SourceMapObjects;
104 // The cache only joins lookups of one key; a decode holds ~40 MB, so different keys take turns.
105 const decoding = yield* Semaphore.make(1);
106 const cache = yield* Cache.makeWith(
107 (key: string) =>
108 Effect.gen(function* () {
109 const stored = yield* objects.get(key);
110 if (stored === undefined) return "source_map_missing" as const;
111 if (stored.size > MAX_SOURCE_MAP_BYTES) {
112 yield* Effect.logWarning("Stored mobile source map is too large", {
113 key,
114 size: stored.size,
115 });
116 return "source_map_too_large" as const;
117 }
118 const map = decodeSourceMap(parseJson(yield* stored.text));
119 if (typeof map !== "string") return { key, map } satisfies FoundSourceMap;
120 yield* Effect.logWarning("Stored mobile source map could not be decoded", { key, map });
121 return map === "too_large"
122 ? ("source_map_too_large" as const)
123 : ("source_map_invalid" as const);
124 }).pipe(Semaphore.withPermit(decoding)),
125 {
126 capacity: SOURCE_MAP_CACHE_CAPACITY,
127 timeToLive: (exit) =>
128 Exit.isFailure(exit)
129 ? FAILED_SOURCE_MAP_TTL
130 : typeof exit.value === "string"
131 ? MISSING_SOURCE_MAP_TTL
132 : Duration.infinity,
133 },
134 );
135 return SourceMaps.of({
136 find: Effect.fn("MobileTelemetry.findSourceMap")(function* (key: string) {
137 return yield* Cache.get(cache, key).pipe(
138 Effect.catch((error) =>
139 Effect.logWarning("Mobile source map lookup failed", { key: error.key }).pipe(
140 Effect.as("source_map_unavailable" as const),
141 ),
142 ),
143 );
144 }),
145 });
146});
147
148/** Build once per Worker isolate, not per request: decoding a production map costs ~150 ms. */
149export const layerSourceMaps = Layer.effect(SourceMaps, makeSourceMaps);
150
151export class MobileTelemetry extends Context.Service<
152 MobileTelemetry,
153 {
154 readonly handle: (
155 request: HttpServerRequest.HttpServerRequest,
156 params: { readonly projectId: string; readonly signal: RelayMobileTelemetrySignal },
157 ) => Effect.Effect<HttpServerResponse.HttpServerResponse>;
158 }
159>()("t3code-relay/mobileTelemetry/MobileTelemetry") {}
160
161class BodyTooLarge extends Schema.TaggedError<BodyTooLarge>()("BodyTooLarge", {}) {}
162
163const errorResponse = (status: number, error: string, headers?: Record<string, string>) =>
164 HttpServerResponse.jsonUnsafe({ error }, { status, ...(headers ? { headers } : {}) });
165
166const readCapped = <E>(stream: Stream.Stream<Uint8Array, E>, maxBytes: number) =>
167 Effect.suspend(() => {
168 const chunks: Array<Uint8Array> = [];
169 let total = 0;
170 return stream.pipe(
171 Stream.runForEach((chunk) => {
172 total += chunk.length;
173 if (total > maxBytes) return Effect.fail(new BodyTooLarge());
174 chunks.push(chunk);
175 return Effect.void;
176 }),
177 Effect.map(() => {
178 const body = new Uint8Array(total);
179 let offset = 0;
180 for (const chunk of chunks) {
181 body.set(chunk, offset);
182 offset += chunk.length;
183 }
184 return body;
185 }),
186 );
187 });
188
189const hasNoBody = (request: HttpServerRequest.HttpServerRequest) =>
190 request.source instanceof Request && request.source.body === null;
191
192const make = Effect.gen(function* () {
193 const settings = yield* MobileTelemetrySettings;
194 const rateLimiter = yield* MobileTelemetryRateLimiter;
195 const sourceMaps = yield* SourceMaps;
196 const httpClient = yield* HttpClient.HttpClient;
197
198 const symbolicateLogs = Effect.fn("relay.mobile_telemetry.symbolicate")(function* (
199 body: Uint8Array,
200 ) {
201 const payload = parseJson(new TextDecoder().decode(body));
202 const exceptions = collectJsExceptions(payload);
203 const jsFatalCrashes = markJsFatalCrashes(payload);
204 if (exceptions.length === 0 && jsFatalCrashes === 0) return undefined;
205 let symbolicated = 0;
206 let growthLeft = MAX_SYMBOLICATION_GROWTH;
207 const keys = new Set<string>();
208 for (const exception of exceptions) {
209 const key = bundleSourceMapKey(exception.identity);
210 if (key !== undefined && keys.size < MAX_SOURCE_MAPS_PER_REQUEST) keys.add(key);
211 const outcome = applySymbolication(
212 exception,
213 key === undefined
214 ? "no_bundle_id"
215 : keys.has(key)
216 ? yield* sourceMaps.find(key)
217 : "source_map_limit",
218 growthLeft,
219 );
220 growthLeft -= outcome.growth;
221 if (outcome.symbolicated) symbolicated += 1;
222 }
223 yield* Effect.annotateCurrentSpan({
224 "relay.mobile_telemetry.exceptions": exceptions.length,
225 "relay.mobile_telemetry.symbolicated": symbolicated,
226 "relay.mobile_telemetry.js_fatal_crashes": jsFatalCrashes,
227 });
228 return new TextEncoder().encode(encodeJson(payload));
229 });
230
231 const forward = Effect.fn("relay.mobile_telemetry.forward")(function* (
232 destination: AxiomDestination,
233 body: Uint8Array,
234 ) {
235 const response = yield* httpClient.execute(
236 HttpClientRequest.post(destination.url, {
237 headers: {
238 authorization: `Bearer ${Redacted.value(settings.axiomToken)}`,
239 "x-axiom-dataset": destination.dataset,
240 },
241 }).pipe(HttpClientRequest.bodyUint8Array(body, "application/json")),
242 );
243 const responseBody = yield* readCapped(response.stream, MAX_UPSTREAM_RESPONSE_BYTES).pipe(
244 Effect.catchIf(
245 (error) => error._tag === "HttpClientError" && error.reason._tag === "EmptyBodyError",
246 () => Effect.succeed(new Uint8Array(0)),
247 ),
248 );
249 return { response, body: responseBody };
250 }, Effect.timeoutOption(UPSTREAM_TIMEOUT));
251
252 const handle = Effect.fn("relay.mobile_telemetry.ingest")(function* (
253 request: HttpServerRequest.HttpServerRequest,
254 {
255 projectId,
256 signal,
257 }: { readonly projectId: string; readonly signal: RelayMobileTelemetrySignal },
258 ) {
259 const outcome = (value: string) =>
260 Effect.annotateCurrentSpan({ "relay.mobile_telemetry.outcome": value });
261 yield* Effect.annotateCurrentSpan({
262 "relay.mobile_telemetry.project_id": projectId,
263 "relay.mobile_telemetry.signal": signal,
264 });
265
266 const clientIp = request.headers["cf-connecting-ip"] ?? "unknown";
267 if (!(yield* rateLimiter.allow(clientIp))) {
268 yield* outcome("rate_limited");
269 return errorResponse(429, "rate_limited", {
270 "retry-after": String(MOBILE_TELEMETRY_RATE_LIMIT.periodSeconds),
271 });
272 }
273
274 const read = yield* (
275 hasNoBody(request)
276 ? Effect.succeed(new Uint8Array(0))
277 : readCapped(request.stream, MOBILE_TELEMETRY_MAX_BODY_BYTES)
278 ).pipe(Effect.result);
279 if (Result.isFailure(read)) {
280 const tooLarge = read.failure._tag === "BodyTooLarge";
281 yield* outcome(tooLarge ? "body_too_large" : "body_unreadable");
282 return errorResponse(tooLarge ? 413 : 400, tooLarge ? "body_too_large" : "invalid_body");
283 }
284 yield* Effect.annotateCurrentSpan({ "relay.mobile_telemetry.bytes": read.success.length });
285
286 // Forwarding matters more than symbolication: on any defect the payload goes through as sent.
287 const body =
288 signal === "logs"
289 ? ((yield* symbolicateLogs(read.success).pipe(
290 Effect.catchCause((cause) =>
291 Effect.logWarning("Mobile stack symbolication failed", { cause }).pipe(
292 Effect.as(undefined),
293 ),
294 ),
295 )) ?? read.success)
296 : read.success;
297
298 const upstream = yield* forward(settings.destinations[signal], body).pipe(Effect.result);
299 if (Result.isFailure(upstream)) {
300 yield* outcome("upstream_unavailable");
301 return errorResponse(502, "upstream_unavailable");
302 }
303 if (Option.isNone(upstream.success)) {
304 yield* outcome("upstream_timeout");
305 return errorResponse(504, "upstream_timeout");
306 }
307 const { response, body: responseBody } = upstream.success.value;
308 yield* outcome("forwarded");
309 yield* Effect.annotateCurrentSpan({
310 "relay.mobile_telemetry.upstream_status": response.status,
311 });
312 // Observe's retry and partial-success handling reads only these from Axiom's answer.
313 const contentType = response.headers["content-type"];
314 const retryAfter = response.headers["retry-after"];
315 const headers = retryAfter === undefined ? undefined : { "retry-after": retryAfter };
316 if (responseBody.length === 0) {
317 return HttpServerResponse.empty({ status: response.status, ...(headers ? { headers } : {}) });
318 }
319 return HttpServerResponse.uint8Array(responseBody, {
320 status: response.status,
321 ...(headers ? { headers } : {}),
322 ...(contentType ? { contentType } : {}),
323 });
324 });
325
326 return MobileTelemetry.of({ handle });
327});
328
329export const layer = Layer.effect(MobileTelemetry, make);
330
331/** Raw so the body is read once, capped. */
332export const layerApi = HttpApiBuilder.group(
333 RelayApi,
334 "mobileTelemetry",
335 Effect.fnUntraced(function* (handlers) {
336 const telemetry = yield* MobileTelemetry;
337 return handlers.handleRaw("ingest", ({ request, params }) => telemetry.handle(request, params));
338 }),
339);