Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260// node:diagnostics_channel: pure JS pub/sub, named channels that exist whether or not// anyone is listening (frameworks such as undici and fastify `channel()` one at load,// long before any subscriber shows up). No native channel registry backs this — there// is nothing else in the runtime that publishes to one — so every channel here is the// plain JS kind Node's own Channel class falls back to.type Subscriber = (message: unknown, name: string | symbol) => void;type Transform = (data: unknown) => unknown;
function invalidArgType(name: string, expected: string, value: unknown): TypeError { const got = value === null || value === undefined ? ` Received ${value}` : typeof value === "object" ? ` Received an instance of ${(value as object).constructor?.name ?? "Object"}` : ` Received type ${typeof value}`; return Object.assign(new TypeError(`The "${name}" argument must be ${expected}.${got}`), { code: "ERR_INVALID_ARG_TYPE" });}
/** A store with the AsyncLocalStorage shape bindStore/runStores expects: `.run(store, * callback)`, duck-typed so this module need not import async_hooks.ts. */interface AlsLike { run<R>(store: unknown, callback: (...args: unknown[]) => R, ...args: unknown[]): R;}
export class Channel { readonly name: string | symbol; #subscribers: Subscriber[] = []; #stores = new Map<AlsLike, Transform>();
constructor(name: string | symbol) { this.name = name; }
static [Symbol.hasInstance](instance: unknown): boolean { return typeof instance === "object" && instance !== null && Object.getPrototypeOf(instance) === Channel.prototype; }
get hasSubscribers(): boolean { return this.#subscribers.length > 0 || this.#stores.size > 0; }
subscribe(subscription: unknown): void { if (typeof subscription !== "function") throw invalidArgType("subscription", "of type function", subscription); this.#subscribers = [...this.#subscribers, subscription as Subscriber]; }
unsubscribe(subscription: unknown): boolean { const i = this.#subscribers.indexOf(subscription as Subscriber); if (i === -1) return false; this.#subscribers = this.#subscribers.filter((_, idx) => idx !== i); return true; }
bindStore(store: AlsLike, transform: Transform = (data) => data): void { this.#stores.set(store, transform); }
unbindStore(store: AlsLike): boolean { return this.#stores.delete(store); }
publish(data?: unknown): void { for (const subscriber of this.#subscribers) { try { subscriber(data, this.name); } catch { // Node reports a subscriber's throw as an uncaught exception without // interrupting publish; there is no process-wide uncaught hook to route // this to from inside a builtin module, so it is swallowed here instead // of silently corrupting the publish loop. } } }
runStores<R>(data: unknown, fn: (...args: unknown[]) => R, thisArg?: unknown, ...args: unknown[]): R { let run = (): R => { this.publish(data); return fn.apply(thisArg, args); }; for (const [store, transform] of this.#stores) { const inner = run; run = () => store.run(transform(data), inner); } return run(); }}
const channels = new Map<string | symbol, Channel>();
export function channel(name: string | symbol): Channel { let ch = channels.get(name); if (ch) return ch; if (typeof name !== "string" && typeof name !== "symbol") throw invalidArgType("channel", "one of type string or symbol", name); ch = new Channel(name); channels.set(name, ch); return ch;}
export function subscribe(name: string | symbol, subscription: Subscriber): void { channel(name).subscribe(subscription);}
export function unsubscribe(name: string | symbol, subscription: Subscriber): boolean { return channel(name).unsubscribe(subscription);}
export function hasSubscribers(name: string | symbol): boolean { return channels.get(name)?.hasSubscribers ?? false;}
const TRACE_EVENTS = ["start", "end", "asyncStart", "asyncEnd", "error"] as const;type TraceEvent = (typeof TRACE_EVENTS)[number];
function assertChannel(value: unknown, name: string): asserts value is Channel { if (!(value instanceof Channel)) throw invalidArgType(name, "an instance of Channel", value);}
function channelFor(nameOrChannels: string | Record<string, Channel>, name: TraceEvent): Channel { if (typeof nameOrChannels === "string") return channel(`tracing:${nameOrChannels}:${name}`); if (typeof nameOrChannels === "object" && nameOrChannels !== null) { const ch = nameOrChannels[name]; assertChannel(ch, `nameOrChannels.${name}`); return ch; } throw invalidArgType("nameOrChannels", "one of type string or object", nameOrChannels);}
type TracingContext = { result?: unknown; error?: unknown };
/** diagnostics_channel.tracingChannel: start/end/asyncStart/asyncEnd/error channels * bundled with the trace{Sync,Promise,Callback} helpers that publish to them around * a call, matching Node's TracingChannel (not itself exported, only this factory). */class TracingChannel { readonly start: Channel; readonly end: Channel; readonly asyncStart: Channel; readonly asyncEnd: Channel; readonly error: Channel;
constructor(nameOrChannels: string | Record<string, Channel>) { this.start = channelFor(nameOrChannels, "start"); this.end = channelFor(nameOrChannels, "end"); this.asyncStart = channelFor(nameOrChannels, "asyncStart"); this.asyncEnd = channelFor(nameOrChannels, "asyncEnd"); this.error = channelFor(nameOrChannels, "error"); }
get hasSubscribers(): boolean { return this.start.hasSubscribers || this.end.hasSubscribers || this.asyncStart.hasSubscribers || this.asyncEnd.hasSubscribers || this.error.hasSubscribers; }
subscribe(handlers: Partial<Record<TraceEvent, Subscriber>>): void { for (const name of TRACE_EVENTS) { const handler = handlers[name]; if (handler) this[name].subscribe(handler); } }
unsubscribe(handlers: Partial<Record<TraceEvent, Subscriber>>): boolean { let done = true; for (const name of TRACE_EVENTS) { const handler = handlers[name]; if (handler && !this[name].unsubscribe(handler)) done = false; } return done; }
traceSync<R>(fn: (...args: unknown[]) => R, context: TracingContext = {}, thisArg?: unknown, ...args: unknown[]): R { if (!this.hasSubscribers) return fn.apply(thisArg, args); const { start, end, error } = this; return start.runStores(context, () => { try { const result = fn.apply(thisArg, args); context.result = result; return result; } catch (err) { context.error = err; error.publish(context); throw err; } finally { end.publish(context); } }); }
tracePromise<R>(fn: (...args: unknown[]) => Promise<R> | R, context: TracingContext = {}, thisArg?: unknown, ...args: unknown[]): Promise<R> | R { if (!this.hasSubscribers) return fn.apply(thisArg, args); const { start, end, asyncStart, asyncEnd, error } = this; const reject = (err: unknown): Promise<never> => { context.error = err; error.publish(context); asyncStart.publish(context); asyncEnd.publish(context); return Promise.reject(err); }; const resolve = (result: unknown): unknown => { context.result = result; asyncStart.publish(context); asyncEnd.publish(context); return result; }; return start.runStores(context, () => { try { const result = fn.apply(thisArg, args); const promise = result instanceof Promise ? result : Promise.resolve(result); return promise.then(resolve, reject) as Promise<R>; } catch (err) { context.error = err; error.publish(context); throw err; } finally { end.publish(context); } }); }
traceCallback<R>( fn: (...args: unknown[]) => R, position = -1, context: TracingContext = {}, thisArg?: unknown, ...args: unknown[] ): R { if (!this.hasSubscribers) return fn.apply(thisArg, args); const { start, end, asyncStart, asyncEnd, error } = this; const index = position < 0 ? args.length + position : position; const callback = args[index]; if (typeof callback !== "function") throw invalidArgType("callback", "of type function", callback); const wrapped = function (this: unknown, err: unknown, res: unknown): unknown { if (err) { context.error = err; error.publish(context); } else { context.result = res; } return asyncStart.runStores(context, () => { try { return Reflect.apply(callback as (...a: unknown[]) => unknown, this, arguments as unknown as unknown[]); } finally { asyncEnd.publish(context); } }); }; args[index] = wrapped; return start.runStores(context, () => { try { return fn.apply(thisArg, args); } catch (err) { context.error = err; error.publish(context); throw err; } finally { end.publish(context); } }); }}
export function tracingChannel(nameOrChannels: string | Record<string, Channel>): TracingChannel { return new TracingChannel(nameOrChannels);}