diagnostics_channel.js

'use strict';

const {
  ArrayPrototypeAt,
  ArrayPrototypeIndexOf,
  ArrayPrototypePush,
  ArrayPrototypePushApply,
  ArrayPrototypeSlice,
  ArrayPrototypeSplice,
  ObjectDefineProperty,
  ObjectGetPrototypeOf,
  ObjectSetPrototypeOf,
  PromisePrototypeThen,
  PromiseReject,
  ReflectApply,
  SafeFinalizationRegistry,
  SafeMap,
  SymbolDispose,
  SymbolHasInstance,
} = primordials;

const {
  codes: {
    ERR_INVALID_ARG_TYPE,
  },
} = require('internal/errors');
const {
  validateFunction,
} = require('internal/validators');

const { triggerUncaughtException } = internalBinding('errors');

const dc_binding = internalBinding('diagnostics_channel');
const { subscribers: subscriberCounts } = dc_binding;

const { WeakReference } = require('internal/util');
const { isPromise } = require('internal/util/types');

// Can't delete when weakref count reaches 0 as it could increment again.
// Only GC can be used as a valid time to clean up the channels map.
class WeakRefMap extends SafeMap {
  #finalizers = new SafeFinalizationRegistry((key) => {
    // Check that the key doesn't have any value before deleting, as the WeakRef for the key
    // may have been replaced since finalization callbacks aren't synchronous with GC.
    if (!this.has(key)) this.delete(key);
  });

  set(key, value) {
    this.#finalizers.register(value, key);
    return super.set(key, new WeakReference(value));
  }

  get(key) {
    return super.get(key)?.get();
  }

  has(key) {
    return !!this.get(key);
  }

  incRef(key) {
    return super.get(key)?.incRef();
  }

  decRef(key) {
    return super.get(key)?.decRef();
  }
}

function markActive(channel) {
  // eslint-disable-next-line no-use-before-define
  ObjectSetPrototypeOf(channel, ActiveChannel.prototype);
  channel._subscribers = [];
  channel._stores = new SafeMap();

  // Notify native modules that this channel just got its first subscriber.
  if (channel._index !== undefined)
    dc_binding.notifyChannelActive(channel._index);
}

function maybeMarkInactive(channel) {
  // When there are no more active subscribers or bound, restore to fast prototype.
  if (!channel._subscribers.length && !channel._stores.size) {
    // Notify native modules that this channel just lost its last subscriber.
    if (channel._index !== undefined)
      dc_binding.notifyChannelInactive(channel._index);

    // eslint-disable-next-line no-use-before-define
    ObjectSetPrototypeOf(channel, Channel.prototype);
    channel._subscribers = undefined;
    channel._stores = undefined;
  }
}

class RunStoresScope {
  #stack;

  constructor(activeChannel, data) {
    // eslint-disable-next-line no-restricted-globals
    using stack = new DisposableStack();

    // Enter stores using withScope
    if (activeChannel._stores) {
      for (const entry of activeChannel._stores.entries()) {
        const store = entry[0];
        const transform = entry[1];

        let newContext = data;
        if (transform) {
          try {
            newContext = transform(data);
          } catch (err) {
            process.nextTick(() => {
              triggerUncaughtException(err, false);
            });
            continue;
          }
        }

        stack.use(store.withScope(newContext));
      }
    }

    // Publish data
    activeChannel.publish(data);

    // Transfer ownership of the stack
    this.#stack = stack.move();
  }

  [SymbolDispose]() {
    this.#stack[SymbolDispose]();
  }
}

// TODO(qard): should there be a C++ channel interface?
class ActiveChannel {
  subscribe(subscription) {
    validateFunction(subscription, 'subscription');
    this._subscribers = ArrayPrototypeSlice(this._subscribers);
    ArrayPrototypePush(this._subscribers, subscription);
    channels.incRef(this.name);
    if (this._index !== undefined) subscriberCounts[this._index]++;
  }

  unsubscribe(subscription) {
    const index = ArrayPrototypeIndexOf(this._subscribers, subscription);
    if (index === -1) return false;

    const before = ArrayPrototypeSlice(this._subscribers, 0, index);
    const after = ArrayPrototypeSlice(this._subscribers, index + 1);
    this._subscribers = before;
    ArrayPrototypePushApply(this._subscribers, after);

    channels.decRef(this.name);
    if (this._index !== undefined) subscriberCounts[this._index]--;
    maybeMarkInactive(this);

    return true;
  }

  bindStore(store, transform) {
    const replacing = this._stores.has(store);
    if (!replacing) {
      channels.incRef(this.name);
      if (this._index !== undefined) subscriberCounts[this._index]++;
    }
    this._stores.set(store, transform);
  }

  unbindStore(store) {
    if (!this._stores.has(store)) {
      return false;
    }

    this._stores.delete(store);

    channels.decRef(this.name);
    if (this._index !== undefined) subscriberCounts[this._index]--;
    maybeMarkInactive(this);

    return true;
  }

  get hasSubscribers() {
    return true;
  }

  publish(data) {
    const subscribers = this._subscribers;
    for (let i = 0; i < (subscribers?.length || 0); i++) {
      try {
        const onMessage = subscribers[i];
        onMessage(data, this.name);
      } catch (err) {
        process.nextTick(() => {
          triggerUncaughtException(err, false);
        });
      }
    }
  }

  withStoreScope(data) {
    return new RunStoresScope(this, data);
  }

  runStores(data, fn, thisArg, ...args) {
    // eslint-disable-next-line no-unused-vars
    using scope = this.withStoreScope(data);
    return ReflectApply(fn, thisArg, args);
  }
}

class Channel {
  constructor(name) {
    this._subscribers = undefined;
    this._stores = undefined;
    this.name = name;
    if (typeof name === 'string') {
      this._index = dc_binding.getOrCreateChannelIndex(name);
    }

    channels.set(name, this);
  }

  static [SymbolHasInstance](instance) {
    const prototype = ObjectGetPrototypeOf(instance);
    return prototype === Channel.prototype ||
           prototype === ActiveChannel.prototype;
  }

  subscribe(subscription) {
    markActive(this);
    this.subscribe(subscription);
  }

  unsubscribe() {
    return false;
  }

  bindStore(store, transform) {
    markActive(this);
    this.bindStore(store, transform);
  }

  unbindStore() {
    return false;
  }

  get hasSubscribers() {
    return false;
  }

  publish() {}

  runStores(data, fn, thisArg, ...args) {
    return ReflectApply(fn, thisArg, args);
  }

  withStoreScope() {
    // Return no-op disposable for inactive channels
    return {
      [SymbolDispose]() {},
    };
  }
}

const channels = new WeakRefMap();

function channel(name) {
  const channel = channels.get(name);
  if (channel) return channel;

  if (typeof name !== 'string' && typeof name !== 'symbol') {
    throw new ERR_INVALID_ARG_TYPE('channel', ['string', 'symbol'], name);
  }

  return new Channel(name);
}

function subscribe(name, subscription) {
  return channel(name).subscribe(subscription);
}

function unsubscribe(name, subscription) {
  return channel(name).unsubscribe(subscription);
}

function hasSubscribers(name) {
  const channel = channels.get(name);
  if (!channel) return false;

  return channel.hasSubscribers;
}

const boundedEvents = [
  'start',
  'end',
];

function assertChannel(value, name) {
  if (!(value instanceof Channel)) {
    throw new ERR_INVALID_ARG_TYPE(name, ['Channel'], value);
  }
}

function emitNonThenableWarning(fn) {
  process.emitWarning(`tracePromise was called with the function '${fn.name || '<anonymous>'}', ` +
                      'which returned a non-thenable.');
}

function channelFromMap(nameOrChannels, name, className) {
  if (typeof nameOrChannels === 'string') {
    return channel(`tracing:${nameOrChannels}:${name}`);
  }

  if (typeof nameOrChannels === 'object' && nameOrChannels !== null) {
    const channel = nameOrChannels[name];
    assertChannel(channel, `nameOrChannels.${name}`);
    return channel;
  }

  throw new ERR_INVALID_ARG_TYPE('nameOrChannels',
                                 ['string', 'object', className],
                                 nameOrChannels);
}

class BoundedChannelScope {
  #context;
  #end;
  #scope;

  constructor(boundedChannel, context) {
    // Only proceed if there are subscribers
    if (!boundedChannel.hasSubscribers) {
      return;
    }

    const { start, end } = boundedChannel;
    this.#context = context;
    this.#end = end;

    // Use RunStoresScope for the start channel
    this.#scope = new RunStoresScope(start, context);
  }

  [SymbolDispose]() {
    if (!this.#scope) {
      return;
    }

    // Publish end event
    this.#end.publish(this.#context);

    // Dispose the start scope to restore stores
    this.#scope[SymbolDispose]();
    this.#scope = undefined;
  }
}

class BoundedChannel {
  constructor(nameOrChannels) {
    for (let i = 0; i < boundedEvents.length; ++i) {
      const eventName = boundedEvents[i];
      ObjectDefineProperty(this, eventName, {
        __proto__: null,
        value: channelFromMap(nameOrChannels, eventName, 'BoundedChannel'),
      });
    }
  }

  get hasSubscribers() {
    return this.start?.hasSubscribers ||
      this.end?.hasSubscribers;
  }

  subscribe(handlers) {
    for (let i = 0; i < boundedEvents.length; ++i) {
      const name = boundedEvents[i];
      if (!handlers[name]) continue;

      this[name]?.subscribe(handlers[name]);
    }
  }

  unsubscribe(handlers) {
    let done = true;

    for (let i = 0; i < boundedEvents.length; ++i) {
      const name = boundedEvents[i];
      if (!handlers[name]) continue;

      if (!this[name]?.unsubscribe(handlers[name])) {
        done = false;
      }
    }

    return done;
  }

  withScope(context = {}) {
    return new BoundedChannelScope(this, context);
  }

  run(context, fn, thisArg, ...args) {
    context ??= {};
    // eslint-disable-next-line no-unused-vars
    using scope = this.withScope(context);
    return ReflectApply(fn, thisArg, args);
  }
}

function boundedChannel(nameOrChannels) {
  return new BoundedChannel(nameOrChannels);
}

class TracingChannel {
  #callWindow;
  #continuationWindow;

  constructor(nameOrChannels) {
    // Create a BoundedChannel for start/end (call window)
    if (typeof nameOrChannels === 'string') {
      this.#callWindow = new BoundedChannel(nameOrChannels);
      this.#continuationWindow = new BoundedChannel({
        start: channel(`tracing:${nameOrChannels}:asyncStart`),
        end: channel(`tracing:${nameOrChannels}:asyncEnd`),
      });
    } else if (typeof nameOrChannels === 'object') {
      this.#callWindow = new BoundedChannel({
        start: nameOrChannels.start,
        end: nameOrChannels.end,
      });
      this.#continuationWindow = new BoundedChannel({
        start: nameOrChannels.asyncStart,
        end: nameOrChannels.asyncEnd,
      });
    }

    // Create individual channel for error
    ObjectDefineProperty(this, 'error', {
      __proto__: null,
      value: channelFromMap(nameOrChannels, 'error', 'TracingChannel'),
    });
  }

  get start() {
    return this.#callWindow.start;
  }

  get end() {
    return this.#callWindow.end;
  }

  get asyncStart() {
    return this.#continuationWindow.start;
  }

  get asyncEnd() {
    return this.#continuationWindow.end;
  }

  get hasSubscribers() {
    return this.#callWindow.hasSubscribers ||
      this.#continuationWindow.hasSubscribers ||
      this.error?.hasSubscribers;
  }

  subscribe(handlers) {
    // Subscribe to call window (start/end)
    if (handlers.start || handlers.end) {
      this.#callWindow.subscribe({
        start: handlers.start,
        end: handlers.end,
      });
    }

    // Subscribe to continuation window (asyncStart/asyncEnd)
    if (handlers.asyncStart || handlers.asyncEnd) {
      this.#continuationWindow.subscribe({
        start: handlers.asyncStart,
        end: handlers.asyncEnd,
      });
    }

    // Subscribe to error channel
    if (handlers.error) {
      this.error.subscribe(handlers.error);
    }
  }

  unsubscribe(handlers) {
    let done = true;

    // Unsubscribe from call window
    if (handlers.start || handlers.end) {
      if (!this.#callWindow.unsubscribe({
        start: handlers.start,
        end: handlers.end,
      })) {
        done = false;
      }
    }

    // Unsubscribe from continuation window
    if (handlers.asyncStart || handlers.asyncEnd) {
      if (!this.#continuationWindow.unsubscribe({
        start: handlers.asyncStart,
        end: handlers.asyncEnd,
      })) {
        done = false;
      }
    }

    // Unsubscribe from error channel
    if (handlers.error) {
      if (!this.error.unsubscribe(handlers.error)) {
        done = false;
      }
    }

    return done;
  }

  traceSync(fn, context = {}, thisArg, ...args) {
    if (!this.hasSubscribers) {
      return ReflectApply(fn, thisArg, args);
    }

    const { error } = this;

    // eslint-disable-next-line no-unused-vars
    using scope = this.#callWindow.withScope(context);
    try {
      const result = ReflectApply(fn, thisArg, args);
      context.result = result;
      return result;
    } catch (err) {
      context.error = err;
      error.publish(context);
      throw err;
    }
  }

  tracePromise(fn, context = {}, thisArg, ...args) {
    if (!this.hasSubscribers) {
      const result = ReflectApply(fn, thisArg, args);
      if (typeof result?.then !== 'function') {
        emitNonThenableWarning(fn);
      }
      return result;
    }

    const { error } = this;
    const continuationWindow = this.#continuationWindow;

    function reject(err) {
      context.error = err;
      error.publish(context);
      // Use continuation window for asyncStart/asyncEnd
      // eslint-disable-next-line no-unused-vars
      using scope = continuationWindow.withScope(context);
      // TODO: Is there a way to have asyncEnd _after_ the continuation?
      return PromiseReject(err);
    }

    function resolve(result) {
      context.result = result;
      // Use continuation window for asyncStart/asyncEnd
      // eslint-disable-next-line no-unused-vars
      using scope = continuationWindow.withScope(context);
      // TODO: Is there a way to have asyncEnd _after_ the continuation?
      return result;
    }

    // eslint-disable-next-line no-unused-vars
    using scope = this.#callWindow.withScope(context);
    try {
      const result = ReflectApply(fn, thisArg, args);
      // If the return value is not a thenable, return it directly with a warning.
      // Do not publish to asyncStart/asyncEnd.
      if (typeof result?.then !== 'function') {
        emitNonThenableWarning(fn);
        context.result = result;
        return result;
      }
      // For native Promises use PromisePrototypeThen to avoid user overrides.
      if (isPromise(result)) {
        return PromisePrototypeThen(result, resolve, reject);
      }
      // For custom thenables, call .then() directly to preserve the thenable type.
      return result.then(resolve, reject);
    } catch (err) {
      context.error = err;
      error.publish(context);
      throw err;
    }
  }

  traceCallback(fn, position = -1, context = {}, thisArg, ...args) {
    if (!this.hasSubscribers) {
      return ReflectApply(fn, thisArg, args);
    }

    const { error } = this;
    const continuationWindow = this.#continuationWindow;

    function wrappedCallback(err, res) {
      if (err) {
        context.error = err;
        error.publish(context);
      } else {
        context.result = res;
      }

      // Use continuation window for asyncStart/asyncEnd around callback
      // eslint-disable-next-line no-unused-vars
      using scope = continuationWindow.withScope(context);
      return ReflectApply(callback, this, arguments);
    }

    const callback = ArrayPrototypeAt(args, position);
    validateFunction(callback, 'callback');
    ArrayPrototypeSplice(args, position, 1, wrappedCallback);

    // eslint-disable-next-line no-unused-vars
    using scope = this.#callWindow.withScope(context);
    try {
      return ReflectApply(fn, thisArg, args);
    } catch (err) {
      context.error = err;
      error.publish(context);
      throw err;
    }
  }
}

function tracingChannel(nameOrChannels) {
  return new TracingChannel(nameOrChannels);
}

dc_binding.linkNativeChannel((name) => channel(name));

module.exports = {
  channel,
  hasSubscribers,
  subscribe,
  tracingChannel,
  unsubscribe,
  boundedChannel,
  Channel,
  BoundedChannel,
};