1import * as Cause from "effect/Cause";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import type * as Exit from "effect/Exit";
5import * as ExitRuntime from "effect/Exit";
6import * as Option from "effect/Option";
7import * as Queue from "effect/Queue";
8import * as Result from "effect/Result";
9import * as Semaphore from "effect/Semaphore";
10import * as Schema from "effect/Schema";
11import * as SchemaIssue from "effect/SchemaIssue";
12import * as SchemaTransformation from "effect/SchemaTransformation";
13import * as Tracer from "effect/Tracer";
14import { OtlpResource, OtlpTracer, OtlpSerialization } from "effect/observability";
15
16import { makeAsyncRotatingFileSink } from "./logging.ts";
17
18export const OtlpProtocol = Schema.Literals(["http/json", "http/protobuf"]);
19export type OtlpProtocol = typeof OtlpProtocol.Type;
20export const layerOtlpSerialization = (protocol: OtlpProtocol) =>
21 protocol === "http/protobuf" ? OtlpSerialization.layerProtobuf : OtlpSerialization.layerJson;
22
23/**
24 * How one signal is exported, once whichever source named that signal's
25 * endpoint has been resolved. Held per signal rather than per process, so a
26 * wire format or a credential cannot be paired by hand with an endpoint that
27 * came from somewhere else.
28 */
29export interface SignalExport {
30 readonly protocol: OtlpProtocol;
31 readonly headers: Readonly<Record<string, string>> | undefined;
32 readonly exportIntervalMs: number;
33}
34
35/** What T3 Code exports with when nothing configured a signal. */
36export const DEFAULT_SIGNAL_EXPORT: SignalExport = {
37 protocol: "http/json",
38 headers: undefined,
39 exportIntervalMs: 10_000,
40};
41
42const FLUSH_BUFFER_THRESHOLD = 256;
43
44export type TraceAttributes = Readonly<Record<string, unknown>>;
45
46export interface TraceRecordEvent {
47 readonly name: string;
48 readonly timeUnixNano: string;
49 readonly attributes: Readonly<Record<string, unknown>>;
50}
51
52export interface TraceRecordLink {
53 readonly traceId: string;
54 readonly spanId: string;
55 readonly attributes: Readonly<Record<string, unknown>>;
56}
57
58interface BaseTraceRecord {
59 readonly name: string;
60 readonly kind: string;
61 readonly traceId: string;
62 readonly spanId: string;
63 readonly parentSpanId?: string;
64 readonly sampled: boolean;
65 readonly startTimeUnixNano: string;
66 readonly endTimeUnixNano: string;
67 readonly durationMs: number;
68 readonly attributes: Readonly<Record<string, unknown>>;
69 readonly events: ReadonlyArray<TraceRecordEvent>;
70 readonly links: ReadonlyArray<TraceRecordLink>;
71}
72
73export interface EffectTraceRecord extends BaseTraceRecord {
74 readonly type: "effect-span";
75 readonly exit:
76 | {
77 readonly _tag: "Success";
78 }
79 | {
80 readonly _tag: "Interrupted";
81 readonly cause: string;
82 }
83 | {
84 readonly _tag: "Failure";
85 readonly cause: string;
86 };
87}
88
89export interface OtlpTraceRecord extends BaseTraceRecord {
90 readonly type: "otlp-span";
91 readonly resourceAttributes: Readonly<Record<string, unknown>>;
92 readonly scope: Readonly<{
93 readonly name?: string;
94 readonly version?: string;
95 readonly attributes: Readonly<Record<string, unknown>>;
96 }>;
97 readonly status?:
98 | {
99 readonly code?: string;
100 readonly message?: string;
101 }
102 | undefined;
103}
104
105export type TraceRecord = EffectTraceRecord | OtlpTraceRecord;
106
107function isStructuralTag(value: unknown): value is string {
108 return (
109 typeof value === "string" &&
110 value.length > 0 &&
111 value.length <= 128 &&
112 /^[A-Za-z][A-Za-z0-9._:/-]*$/.test(value)
113 );
114}
115
116export function errorTag(error: unknown): string {
117 try {
118 if (typeof error === "object" && error !== null && "_tag" in error) {
119 return isStructuralTag(error._tag) ? error._tag : "TaggedError";
120 }
121 if (error instanceof Error) {
122 return isStructuralTag(error.name) ? error.name : "Error";
123 }
124 } catch {
125 return "UnknownError";
126 }
127 return typeof error;
128}
129
130export function causeErrorTag(cause: Cause.Cause<unknown>): string {
131 const failure = Cause.findErrorOption(cause);
132 if (Option.isSome(failure)) {
133 return errorTag(failure.value);
134 }
135 return cause.reasons[0]?._tag ?? "Empty";
136}
137
138export interface TraceSinkOptions {
139 readonly filePath: string;
140 readonly maxBytes: number;
141 readonly maxFiles: number;
142 readonly batchWindowMs: number;
143 readonly onFlush?: (stats: TraceSinkFlushStats) => Effect.Effect<void>;
144}
145
146export interface TraceSinkFlushStats {
147 readonly logicalWriteBytes: number;
148 readonly count: number;
149 readonly durationMs: number;
150}
151
152export interface TraceSink {
153 readonly filePath: string;
154 push: (record: TraceRecord) => void;
155 flush: Effect.Effect<void>;
156 close: () => Effect.Effect<void>;
157}
158
159export interface LocalFileTracerOptions extends TraceSinkOptions {
160 readonly delegate?: Tracer.Tracer;
161 readonly sink?: TraceSink;
162}
163
164type OtlpSpan = OtlpTracer.ScopeSpan["spans"][number];
165type OtlpSpanEvent = OtlpSpan["events"][number];
166type OtlpSpanLink = OtlpSpan["links"][number];
167type OtlpSpanStatus = OtlpSpan["status"];
168
169interface SerializableSpan {
170 readonly name: string;
171 readonly traceId: string;
172 readonly spanId: string;
173 readonly parent: Option.Option<Tracer.AnySpan>;
174 readonly status: Tracer.SpanStatus;
175 readonly sampled: boolean;
176 readonly kind: Tracer.SpanKind;
177 readonly attributes: ReadonlyMap<string, unknown>;
178 readonly links: ReadonlyArray<Tracer.SpanLink>;
179 readonly events: ReadonlyArray<
180 readonly [name: string, startTime: bigint, attributes: Record<string, unknown>]
181 >;
182}
183
184function isPlainObject(value: unknown): value is Record<string, unknown> {
185 return typeof value === "object" && value !== null && !Array.isArray(value);
186}
187
188function markSeen(value: object, seen: WeakSet<object>): boolean {
189 if (seen.has(value)) {
190 return true;
191 }
192 seen.add(value);
193 return false;
194}
195
196function normalizeJsonValue(value: unknown, seen: WeakSet<object> = new WeakSet()): unknown {
197 if (
198 value === null ||
199 value === undefined ||
200 typeof value === "string" ||
201 typeof value === "number" ||
202 typeof value === "boolean"
203 ) {
204 return value ?? null;
205 }
206 if (typeof value === "bigint") {
207 return value.toString();
208 }
209 if (value instanceof Date) {
210 return Number.isNaN(value.getTime()) ? "Invalid Date" : value.toISOString();
211 }
212 if (value instanceof Error) {
213 return {
214 name: value.name,
215 message: value.message,
216 ...(value.stack ? { stack: value.stack } : {}),
217 };
218 }
219 if (Array.isArray(value)) {
220 if (markSeen(value, seen)) {
221 return "[Circular]";
222 }
223 return value.map((entry) => normalizeJsonValue(entry, seen));
224 }
225 if (value instanceof Map) {
226 if (markSeen(value, seen)) {
227 return "[Circular]";
228 }
229 return Object.fromEntries(
230 Array.from(value.entries(), ([key, entryValue]) => [
231 String(key),
232 normalizeJsonValue(entryValue, seen),
233 ]),
234 );
235 }
236 if (value instanceof Set) {
237 if (markSeen(value, seen)) {
238 return "[Circular]";
239 }
240 return Array.from(value.values(), (entry) => normalizeJsonValue(entry, seen));
241 }
242 if (!isPlainObject(value)) {
243 return String(value);
244 }
245 if (markSeen(value, seen)) {
246 return "[Circular]";
247 }
248 return Object.fromEntries(
249 Object.entries(value).map(([key, entryValue]) => [key, normalizeJsonValue(entryValue, seen)]),
250 );
251}
252
253export function compactTraceAttributes(
254 attributes: Readonly<Record<string, unknown>>,
255): TraceAttributes {
256 const entries: Array<[string, unknown]> = [];
257 for (const [key, value] of Object.entries(attributes)) {
258 if (value !== undefined) {
259 entries.push([key, normalizeJsonValue(value)]);
260 }
261 }
262 return Object.fromEntries(entries);
263}
264
265function formatTraceExit(exit: Exit.Exit<unknown, unknown>): EffectTraceRecord["exit"] {
266 if (ExitRuntime.isSuccess(exit)) {
267 return { _tag: "Success" };
268 }
269 if (Cause.hasInterruptsOnly(exit.cause)) {
270 return {
271 _tag: "Interrupted",
272 cause: Cause.pretty(exit.cause),
273 };
274 }
275 return {
276 _tag: "Failure",
277 cause: Cause.pretty(exit.cause),
278 };
279}
280
281const TRACE_ATTRIBUTE_MAX_LENGTH = 500;
282const TRACE_ATTRIBUTE_TRUNCATED_LENGTH = 200;
283const TRACE_ATTRIBUTE_TRUNCATION_SUFFIX = "…[truncated]";
284const ALWAYS_TRUNCATED_TRACE_ATTRIBUTES: ReadonlySet<string> = new Set(["db.query.text"]);
285
286// Clamps strings nested inside already-normalized attribute values (arrays and
287// plain objects from normalizeJsonValue, e.g. an Error's `stack`). Returns the
288// input reference when nothing was clamped.
289function truncateNestedValue(value: unknown): unknown {
290 if (typeof value === "string") {
291 return value.length <= TRACE_ATTRIBUTE_MAX_LENGTH
292 ? value
293 : `${value.slice(0, TRACE_ATTRIBUTE_MAX_LENGTH)}${TRACE_ATTRIBUTE_TRUNCATION_SUFFIX}`;
294 }
295 if (Array.isArray(value)) {
296 const truncated = value.map(truncateNestedValue);
297 return truncated.some((entry, index) => entry !== value[index]) ? truncated : value;
298 }
299 if (isPlainObject(value)) {
300 let truncated: Record<string, unknown> | undefined;
301 for (const [key, entry] of Object.entries(value)) {
302 const next = truncateNestedValue(entry);
303 if (next === entry) continue;
304 truncated ??= { ...value };
305 truncated[key] = next;
306 }
307 return truncated ?? value;
308 }
309 return value;
310}
311
312/**
313 * Clamps oversized attribute values on the serialized trace record so the file
314 * sink stays small, including strings nested inside arrays and objects (e.g.
315 * error stacks). Returns a new record when anything was clamped; never
316 * mutates the input (the live span's attributes are shared with other tracers).
317 */
318export function truncateTraceAttributes(attributes: TraceAttributes): TraceAttributes {
319 let truncated: Record<string, unknown> | undefined;
320 for (const [key, value] of Object.entries(attributes)) {
321 if (typeof value === "string" && ALWAYS_TRUNCATED_TRACE_ATTRIBUTES.has(key)) {
322 if (value.length <= TRACE_ATTRIBUTE_TRUNCATED_LENGTH) continue;
323 truncated ??= { ...attributes };
324 truncated[key] =
325 `${value.slice(0, TRACE_ATTRIBUTE_TRUNCATED_LENGTH)}${TRACE_ATTRIBUTE_TRUNCATION_SUFFIX}`;
326 continue;
327 }
328 const next = truncateNestedValue(value);
329 if (next === value) continue;
330 truncated ??= { ...attributes };
331 truncated[key] = next;
332 }
333 return truncated ?? attributes;
334}
335
336function spanToTraceRecord(span: SerializableSpan): EffectTraceRecord {
337 const status = span.status as Extract<Tracer.SpanStatus, { _tag: "Ended" }>;
338 const parentSpanId = Option.getOrUndefined(span.parent)?.spanId;
339
340 return {
341 type: "effect-span",
342 name: span.name,
343 traceId: span.traceId,
344 spanId: span.spanId,
345 ...(parentSpanId ? { parentSpanId } : {}),
346 sampled: span.sampled,
347 kind: span.kind,
348 startTimeUnixNano: String(status.startTime),
349 endTimeUnixNano: String(status.endTime),
350 durationMs: Number(status.endTime - status.startTime) / 1_000_000,
351 attributes: truncateTraceAttributes(
352 compactTraceAttributes(Object.fromEntries(span.attributes)),
353 ),
354 events: span.events.map(([name, startTime, attributes]) => ({
355 name,
356 timeUnixNano: String(startTime),
357 attributes: truncateTraceAttributes(compactTraceAttributes(attributes)),
358 })),
359 links: span.links.map((link) => ({
360 traceId: link.span.traceId,
361 spanId: link.span.spanId,
362 attributes: truncateTraceAttributes(compactTraceAttributes(link.attributes)),
363 })),
364 exit: formatTraceExit(status.exit),
365 };
366}
367
368export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: TraceSinkOptions) {
369 const sink = yield* makeAsyncRotatingFileSink({
370 filePath: options.filePath,
371 maxBytes: options.maxBytes,
372 maxFiles: options.maxFiles,
373 }).pipe(Effect.orDie);
374
375 let buffer: Array<{ readonly text: string; readonly bytes: number }> = [];
376 let bufferBytes = 0;
377 // While a write is slow, push keeps up to one trace file of records (at least
378 // 4 MiB), so the spans that time a stall still reach the file. Only a disk
379 // that stays stuck drops records, instead of growing memory or blocking push.
380 const maxBufferBytes = Math.max(options.maxBytes, 4 * 1024 * 1024);
381 const flushRequests = yield* Queue.sliding<void>(1);
382 const flushSemaphore = yield* Semaphore.make(1);
383 // A failure episode starts at the first dropped record and ends when a flush
384 // sees that the latest write succeeded. Flush checks once per window, so a
385 // disk that flaps inside one window stays one episode.
386 let writeFailing = false;
387 let droppedCount = 0;
388 let failureReported = false;
389 let pendingFlushStats: TraceSinkFlushStats = {
390 logicalWriteBytes: 0,
391 count: 0,
392 durationMs: 0,
393 };
394
395 const flushBuffered = Effect.gen(function* () {
396 if (buffer.length === 0) {
397 return;
398 }
399
400 const records = buffer;
401 buffer = [];
402 bufferBytes = 0;
403 let persistedCount = 0;
404
405 while (persistedCount < records.length) {
406 // push drops any record larger than a file, so the first one always fits.
407 const firstRecordBytes = records[persistedCount]!.bytes;
408 let nextIndex = persistedCount + 1;
409 let chunkBytes = firstRecordBytes;
410 while (nextIndex < records.length) {
411 const nextRecordBytes = records[nextIndex]!.bytes;
412 if (chunkBytes + nextRecordBytes > options.maxBytes) break;
413 chunkBytes += nextRecordBytes;
414 nextIndex += 1;
415 }
416
417 const chunk = records
418 .slice(persistedCount, nextIndex)
419 .map((record) => record.text)
420 .join("");
421 const startedAt = performance.now();
422 const written = yield* sink.write(chunk).pipe(Effect.result);
423 if (Result.isFailure(written)) {
424 // A failing disk (ENOSPC, EACCES, EIO) drops the rest of the batch.
425 // Retrying it would grow the backlog, and every later push would
426 // retry all of it.
427 writeFailing = true;
428 droppedCount += records.length - persistedCount;
429 return;
430 }
431 writeFailing = false;
432 pendingFlushStats = {
433 logicalWriteBytes: pendingFlushStats.logicalWriteBytes + chunkBytes,
434 count: pendingFlushStats.count + nextIndex - persistedCount,
435 durationMs: pendingFlushStats.durationMs + Math.max(0, performance.now() - startedAt),
436 };
437 persistedCount = nextIndex;
438 }
439 });
440
441 // Logs once when writes start failing and once when they recover, so a disk
442 // that stays broken costs one line, not one per flush.
443 const flush = flushSemaphore
444 .withPermits(1)(
445 Effect.gen(function* () {
446 const droppedAtStart = droppedCount;
447 yield* flushBuffered;
448 const stats = pendingFlushStats;
449 pendingFlushStats = {
450 logicalWriteBytes: 0,
451 count: 0,
452 durationMs: 0,
453 };
454 if (stats.count > 0 && options.onFlush) {
455 yield* options.onFlush(stats).pipe(Effect.ignore);
456 }
457 if (droppedCount > 0 && !failureReported) {
458 failureReported = true;
459 yield* Effect.logWarning(
460 "Trace writer could not keep up, dropping records until it recovers",
461 {
462 filePath: options.filePath,
463 },
464 );
465 } else if (failureReported && !writeFailing && droppedCount === droppedAtStart) {
466 yield* Effect.logInfo("Trace writes recovered", {
467 filePath: options.filePath,
468 droppedCount,
469 });
470 failureReported = false;
471 droppedCount = 0;
472 }
473 }),
474 )
475 .pipe(
476 // A cancelled write still completes in the OS. Finish it before another
477 // flush rotates the file or the scope's finalizer drains the remaining records.
478 Effect.uninterruptible,
479 Effect.withTracerEnabled(false),
480 // The timed fiber inherits the makeTraceSink span. That span has ended but
481 // lives as long as the sink, so the tracer logger must not add logs to it.
482 Effect.updateContext(Context.omit(Tracer.ParentSpan)<never>),
483 );
484
485 yield* Effect.addFinalizer(() => flush.pipe(Effect.ignore));
486 yield* Effect.forkScoped(
487 // Flush when push fills a batch or the batch window passes.
488 Queue.take(flushRequests).pipe(
489 Effect.timeoutOption(`${options.batchWindowMs} millis`),
490 Effect.andThen(flush),
491 Effect.forever,
492 ),
493 );
494
495 return {
496 filePath: options.filePath,
497 push(record) {
498 try {
499 const text = `${JSON.stringify(record)}\n`;
500 const bytes = Buffer.byteLength(text);
501 if (bytes > options.maxBytes) return;
502 if (bufferBytes + bytes > maxBufferBytes) {
503 droppedCount++;
504 return;
505 }
506 buffer.push({ text, bytes });
507 bufferBytes += bytes;
508 if (buffer.length >= FLUSH_BUFFER_THRESHOLD) {
509 Queue.offerUnsafe(flushRequests, undefined);
510 }
511 } catch {
512 return;
513 }
514 },
515 flush,
516 close: () => flush,
517 } satisfies TraceSink;
518});
519
520class LocalFileSpan implements Tracer.Span {
521 readonly _tag = "Span";
522 readonly name: string;
523 readonly spanId: string;
524 readonly traceId: string;
525 readonly parent: Option.Option<Tracer.AnySpan>;
526 readonly annotations: Tracer.Span["annotations"];
527 readonly links: Array<Tracer.SpanLink>;
528 readonly sampled: boolean;
529 readonly kind: Tracer.SpanKind;
530
531 status: Tracer.SpanStatus;
532 attributes: Map<string, unknown>;
533 events: Array<[name: string, startTime: bigint, attributes: Record<string, unknown>]>;
534 private readonly delegate: Tracer.Span;
535 private readonly push: (record: EffectTraceRecord) => void;
536
537 constructor(
538 options: Parameters<Tracer.Tracer["span"]>[0],
539 delegate: Tracer.Span,
540 push: (record: EffectTraceRecord) => void,
541 ) {
542 this.delegate = delegate;
543 this.push = push;
544 this.name = delegate.name;
545 this.spanId = delegate.spanId;
546 this.traceId = delegate.traceId;
547 this.parent = options.parent;
548 this.annotations = options.annotations;
549 this.links = [...options.links];
550 this.sampled = delegate.sampled;
551 this.kind = delegate.kind;
552 this.status = {
553 _tag: "Started",
554 startTime: options.startTime,
555 };
556 this.attributes = new Map();
557 this.events = [];
558 }
559
560 end(endTime: bigint, exit: Exit.Exit<unknown, unknown>): void {
561 // Cached lookup fibers retain parent spans. Traces need success status, not result payloads.
562 const traceExit = ExitRuntime.isSuccess(exit) ? ExitRuntime.void : exit;
563 this.status = {
564 _tag: "Ended",
565 startTime: this.status.startTime,
566 endTime,
567 exit: traceExit,
568 };
569 this.delegate.end(endTime, traceExit);
570
571 if (this.sampled) {
572 this.push(spanToTraceRecord(this));
573 }
574 }
575
576 attribute(key: string, value: unknown): void {
577 this.attributes.set(key, value);
578 this.delegate.attribute(key, value);
579 }
580
581 event(name: string, startTime: bigint, attributes?: Record<string, unknown>): void {
582 const nextAttributes = attributes ?? {};
583 this.events.push([name, startTime, nextAttributes]);
584 this.delegate.event(name, startTime, nextAttributes);
585 }
586
587 addLinks(links: ReadonlyArray<Tracer.SpanLink>): void {
588 this.links.push(...links);
589 this.delegate.addLinks(links);
590 }
591}
592
593export const makeLocalFileTracer = Effect.fn("makeLocalFileTracer")(function* (
594 options: LocalFileTracerOptions,
595) {
596 const sink =
597 options.sink ??
598 (yield* makeTraceSink({
599 filePath: options.filePath,
600 maxBytes: options.maxBytes,
601 maxFiles: options.maxFiles,
602 batchWindowMs: options.batchWindowMs,
603 ...(options.onFlush ? { onFlush: options.onFlush } : {}),
604 }));
605
606 const delegate =
607 options.delegate ??
608 Tracer.make({
609 span: (spanOptions) => new Tracer.NativeSpan(spanOptions),
610 });
611
612 return Tracer.make({
613 span(spanOptions) {
614 return new LocalFileSpan(spanOptions, delegate.span(spanOptions), sink.push);
615 },
616 ...(delegate.context ? { context: delegate.context } : {}),
617 });
618});
619
620const SPAN_KIND_MAP: Record<number, OtlpTraceRecord["kind"]> = {
621 1: "internal",
622 2: "server",
623 3: "client",
624 4: "producer",
625 5: "consumer",
626};
627
628export function decodeOtlpTraceRecords(
629 payload: OtlpTracer.TraceData,
630): ReadonlyArray<OtlpTraceRecord> {
631 const records: Array<OtlpTraceRecord> = [];
632
633 for (const resourceSpan of payload.resourceSpans) {
634 const resourceAttributes = decodeAttributes(resourceSpan.resource?.attributes ?? []);
635
636 for (const scopeSpan of resourceSpan.scopeSpans) {
637 for (const span of scopeSpan.spans) {
638 records.push(
639 otlpSpanToTraceRecord({
640 resourceAttributes,
641 scopeAttributes: decodeAttributes(
642 "attributes" in scopeSpan.scope && Array.isArray(scopeSpan.scope.attributes)
643 ? scopeSpan.scope.attributes
644 : [],
645 ),
646 scopeName: scopeSpan.scope.name,
647 scopeVersion:
648 "version" in scopeSpan.scope && typeof scopeSpan.scope.version === "string"
649 ? scopeSpan.scope.version
650 : undefined,
651 span,
652 }),
653 );
654 }
655 }
656 }
657
658 return records;
659}
660
661function otlpSpanToTraceRecord(input: {
662 readonly resourceAttributes: Readonly<Record<string, unknown>>;
663 readonly scopeAttributes: Readonly<Record<string, unknown>>;
664 readonly scopeName: string | undefined;
665 readonly scopeVersion: string | undefined;
666 readonly span: OtlpSpan;
667}): OtlpTraceRecord {
668 return {
669 type: "otlp-span",
670 name: input.span.name,
671 traceId: input.span.traceId,
672 spanId: input.span.spanId,
673 ...(input.span.parentSpanId ? { parentSpanId: input.span.parentSpanId } : {}),
674 sampled: true,
675 kind: normalizeSpanKind(input.span.kind),
676 startTimeUnixNano: input.span.startTimeUnixNano,
677 endTimeUnixNano: input.span.endTimeUnixNano,
678 durationMs:
679 Number(parseBigInt(input.span.endTimeUnixNano) - parseBigInt(input.span.startTimeUnixNano)) /
680 1_000_000,
681 attributes: decodeAttributes(input.span.attributes),
682 resourceAttributes: input.resourceAttributes,
683 scope: {
684 ...(input.scopeName ? { name: input.scopeName } : {}),
685 ...(input.scopeVersion ? { version: input.scopeVersion } : {}),
686 attributes: input.scopeAttributes,
687 },
688 events: decodeEvents(input.span.events),
689 links: decodeLinks(input.span.links),
690 status: decodeStatus(input.span.status),
691 };
692}
693
694function decodeStatus(input: OtlpSpanStatus): OtlpTraceRecord["status"] {
695 const code = String(input.code);
696 const message = input.message;
697
698 return {
699 code,
700 ...(message ? { message } : {}),
701 };
702}
703
704function decodeEvents(input: ReadonlyArray<OtlpSpanEvent>): ReadonlyArray<TraceRecordEvent> {
705 return input.map((current) => ({
706 name: current.name,
707 timeUnixNano: current.timeUnixNano,
708 attributes: decodeAttributes(current.attributes),
709 }));
710}
711
712function decodeLinks(input: ReadonlyArray<OtlpSpanLink>): ReadonlyArray<TraceRecordLink> {
713 return input.flatMap((current) => {
714 const traceId = current.traceId;
715 const spanId = current.spanId;
716 return {
717 traceId,
718 spanId,
719 attributes: decodeAttributes(current.attributes),
720 };
721 });
722}
723
724function decodeAttributes(
725 input: ReadonlyArray<OtlpResource.KeyValue>,
726): Readonly<Record<string, unknown>> {
727 const entries: Record<string, unknown> = {};
728
729 for (const attribute of input) {
730 entries[attribute.key] = decodeValue(attribute.value);
731 }
732
733 return truncateTraceAttributes(compactTraceAttributes(entries));
734}
735
736function decodeValue(input: OtlpResource.AnyValue | null | undefined): unknown {
737 if (input == null) {
738 return null;
739 }
740 if ("stringValue" in input) {
741 return input.stringValue;
742 }
743 if ("boolValue" in input) {
744 return input.boolValue;
745 }
746 if ("intValue" in input) {
747 return input.intValue;
748 }
749 if ("doubleValue" in input) {
750 return input.doubleValue;
751 }
752 if ("bytesValue" in input) {
753 return input.bytesValue;
754 }
755 if (input.arrayValue) {
756 return input.arrayValue.values.map((entry) => decodeValue(entry));
757 }
758 if (input.kvlistValue) {
759 return decodeAttributes(input.kvlistValue.values);
760 }
761 return null;
762}
763
764function normalizeSpanKind(input: number): OtlpTraceRecord["kind"] {
765 return SPAN_KIND_MAP[input] || "internal";
766}
767
768function parseBigInt(input: string): bigint {
769 try {
770 return BigInt(input);
771 } catch {
772 return 0n;
773 }
774}
775
776/**
777 * Parses the `OTEL_EXPORTER_OTLP_HEADERS` wire format used by
778 * `T3CODE_OTLP_HEADERS`: W3C Baggage `key=value` pairs joined by commas, with
779 * percent-encoded values. Each pair splits at its first `=` so an encoded or
780 * literal `=` inside a value survives, and whitespace around the separators is
781 * ignored.
782 */
783export const OtlpHeadersFromString = Schema.String.pipe(
784 Schema.decodeTo(
785 Schema.Record(Schema.String, Schema.String),
786 SchemaTransformation.transformEffect({
787 decode: (input) => {
788 const headers: Record<string, string> = {};
789 for (const pair of input.split(",")) {
790 if (pair.trim() === "") {
791 continue;
792 }
793 const separator = pair.indexOf("=");
794 const key = separator === -1 ? "" : pair.slice(0, separator).trim();
795 if (key === "") {
796 return Effect.fail(
797 new SchemaIssue.InvalidValue({
798 message: `Expected key=value but received ${JSON.stringify(pair.trim())}.`,
799 }),
800 );
801 }
802 try {
803 headers[key] = decodeURIComponent(pair.slice(separator + 1).trim());
804 } catch {
805 return Effect.fail(
806 new SchemaIssue.InvalidValue({
807 message: `Header ${JSON.stringify(key)} has a malformed percent-encoded value.`,
808 }),
809 );
810 }
811 }
812 return Effect.succeed(headers);
813 },
814 encode: (headers) =>
815 Effect.succeed(
816 Object.entries(headers)
817 .map(([key, value]) => `${key}=${encodeURIComponent(value)}`)
818 .join(","),
819 ),
820 }),
821 ),
822);