feat(utils): add performance observer

packages/utils/src/lib/performance-observer.int.test.ts

@@ -0,0 +1,118 @@
+import { type PerformanceEntry, performance } from 'node:perf_hooks';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+import {
+  type PerformanceObserverOptions,
+  PerformanceObserverSink,
+} from './performance-observer.js';
+import type { Sink } from './sink-source.types';
+
+// @TODO remove duplicate when file-sink is implemented
+class MockSink implements Sink<string, string> {
+  private writtenItems: string[] = [];
+  private closed = false;
+
+  open(): void {
+    this.closed = false;
+  }
+
+  write(input: string): void {
+    this.writtenItems.push(input);
+  }
+
+  close(): void {
+    this.closed = true;
+  }
+
+  isClosed(): boolean {
+    return this.closed;
+  }
+
+  encode(input: string): string {
+    return `${input}-${this.constructor.name}-encoded`;
+  }
+
+  recover(): string[] {
+    return [...this.writtenItems];
+  }
+}
+
+describe('PerformanceObserverSink', () => {
+  let sink: MockSink;
+  let options: PerformanceObserverOptions<string>;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    performance.clearMeasures();
+    performance.clearMarks();
+    sink = new MockSink();
+
+    options = {
+      sink,
+      encode: vi.fn((entry: PerformanceEntry) => [
+        `${entry.name}:${entry.entryType}`,
+      ]),
+    };
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  it('creates instance with default options', () => {
+    expect(() => new PerformanceObserverSink(options)).not.toThrow();
+  });
+
+  it('creates instance with custom options', () => {
+    expect(
+      () =>
+        new PerformanceObserverSink({
+          ...options,
+          buffered: true,
+          flushThreshold: 10,
+        }),
+    ).not.toThrow();
+  });
+
+  it('should observe performance entries and write them to the sink on flush', () => {
+    const observer = new PerformanceObserverSink(options);
+
+    observer.subscribe();
+    performance.mark('test-mark');
+    observer.flush();
+    expect(sink.recover()).toHaveLength(1);
+  });
+
+  it('should observe buffered performance entries when buffered is enabled', async () => {
+    const observer = new PerformanceObserverSink({
+      ...options,
+      buffered: true,
+    });
+
+    performance.mark('test-mark-1');
+    performance.mark('test-mark-2');
+    await new Promise(resolve => setTimeout(resolve, 10));
+    observer.subscribe();
+    await new Promise(resolve => setTimeout(resolve, 10));
+    expect(performance.getEntries()).toHaveLength(2);
+    observer.flush();
+    expect(sink.recover()).toHaveLength(2);
+  });
+
+  it('handles multiple encoded items per performance entry', () => {
+    const multiEncodeFn = vi.fn(e => [
+      `${e.entryType}-item1`,
+      `${e.entryType}item2`,
+    ]);
+    const observer = new PerformanceObserverSink({
+      ...options,
+      encode: multiEncodeFn,
+    });
+
+    observer.subscribe();
+
+    performance.mark('test-mark');
+    observer.flush();
+
+    expect(sink.recover()).toHaveLength(2);
+  });
+});

packages/utils/src/lib/performance-observer.ts

@@ -0,0 +1,88 @@
+import {
+  type EntryType,
+  type PerformanceEntry,
+  PerformanceObserver,
+  type PerformanceObserverEntryList,
+  performance,
+} from 'node:perf_hooks';
+import type { Buffered, Encoder, Observer, Sink } from './sink-source.types.js';
+
+export const DEFAULT_FLUSH_THRESHOLD = 20;
+
+export type PerformanceObserverOptions<T> = {
+  sink: Sink<T, unknown>;
+  encode: (entry: PerformanceEntry) => T[];
+  buffered?: boolean;
+  flushThreshold?: number;
+};
+
+export class PerformanceObserverSink<T>
+  implements Observer, Buffered, Encoder<PerformanceEntry, T[]>
+{
+  #encode: (entry: PerformanceEntry) => T[];
+  #buffered: boolean;
+  #flushThreshold: number;
+  #sink: Sink<T, unknown>;
+  #observer: PerformanceObserver | undefined;
+  #observedTypes: EntryType[] = ['mark', 'measure'];
+  #getEntries = (list: PerformanceObserverEntryList) =>
+    this.#observedTypes.flatMap(t => list.getEntriesByType(t));
+  #observedCount: number = 0;
+
+  constructor(options: PerformanceObserverOptions<T>) {
+    this.#encode = options.encode;
+    this.#sink = options.sink;
+    this.#buffered = options.buffered ?? false;
+    this.#flushThreshold = options.flushThreshold ?? DEFAULT_FLUSH_THRESHOLD;
+  }
+
+  encode(entry: PerformanceEntry): T[] {
+    return this.#encode(entry);
+  }
+
+  subscribe(): void {
+    if (this.#observer) {
+      return;
+    }
+
+    this.#observer = new PerformanceObserver(list => {
+      const entries = this.#getEntries(list);
+      this.#observedCount += entries.length;
+      if (this.#observedCount >= this.#flushThreshold) {
+        this.flush();
+      }
+    });
+
+    this.#observer.observe({
+      entryTypes: this.#observedTypes,
+      buffered: this.#buffered,
+    });
+  }
+
+  flush(): void {
+    if (!this.#observer) {
+      return;
+    }
+
+    const entries = this.#getEntries(performance);
+    entries.forEach(entry => {
+      const encoded = this.encode(entry);
+      encoded.forEach(item => {
+        this.#sink.write(item);
+      });
+    });
+    this.#observedCount = 0;
+  }
+
+  unsubscribe(): void {
+    if (!this.#observer) {
+      return;
+    }
+    this.#observer?.disconnect();
+    this.#observer = undefined;
+  }
+
+  isSubscribed(): boolean {
+    return this.#observer !== undefined;
+  }
+}

packages/utils/src/lib/performance-observer.unit.test.ts

@@ -0,0 +1,165 @@
+import type { PerformanceEntry } from 'node:perf_hooks';
+import {
+  type MockedFunction,
+  beforeEach,
+  describe,
+  expect,
+  it,
+  vi,
+} from 'vitest';
+import { MockPerformanceObserver } from '@code-pushup/test-utils';
+import {
+  type PerformanceObserverOptions,
+  PerformanceObserverSink,
+} from './performance-observer.js';
+import type { Sink } from './sink-source.types';
+
+// @TODO remove duplicate when file-sink is implemented
+class MockSink implements Sink<string, string> {
+  private writtenItems: string[] = [];
+  private closed = false;
+
+  open(): void {
+    this.closed = false;
+  }
+
+  write(input: string): void {
+    this.writtenItems.push(input);
+  }
+
+  close(): void {
+    this.closed = true;
+  }
+
+  isClosed(): boolean {
+    return this.closed;
+  }
+
+  encode(input: string): string {
+    return `${input}-${this.constructor.name}-encoded`;
+  }
+
+  getWrittenItems(): string[] {
+    return [...this.writtenItems];
+  }
+}
+
+describe('PerformanceObserverSink', () => {
+  let encode: MockedFunction<(entry: PerformanceEntry) => string[]>;
+  let sink: MockSink;
+  let options: PerformanceObserverOptions<string>;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    sink = new MockSink();
+    encode = vi.fn((entry: PerformanceEntry) => [
+      `${entry.name}:${entry.entryType}`,
+    ]);
+    options = {
+      sink,
+      encode,
+      flushThreshold: 1,
+    };
+  });
+
+  it('creates instance with default options', () => {
+    expect(() => new PerformanceObserverSink(options)).not.toThrow();
+  });
+
+  it('creates instance with custom options', () => {
+    expect(
+      () =>
+        new PerformanceObserverSink({
+          ...options,
+          buffered: true,
+          flushThreshold: 10,
+        }),
+    ).not.toThrow();
+  });
+
+  it('should be isomorph and create a single observer on subscribe', () => {
+    const observer = new PerformanceObserverSink(options);
+
+    expect(observer.isSubscribed()).toBe(false);
+    expect(MockPerformanceObserver.instances).toHaveLength(0);
+    observer.subscribe();
+    expect(observer.isSubscribed()).toBe(true);
+    expect(MockPerformanceObserver.instances).toHaveLength(1);
+    observer.subscribe();
+    expect(observer.isSubscribed()).toBe(true);
+    expect(MockPerformanceObserver.instances).toHaveLength(1);
+  });
+
+  it('skips non-mark and non-measure entry types', () => {
+    const observer = new PerformanceObserverSink(options);
+
+    observer.subscribe();
+
+    MockPerformanceObserver.lastInstance()?.emitNavigation('test-navigation');
+
+    expect(encode).not.toHaveBeenCalled();
+  });
+
+  it('flushes existing performance entries', () => {
+    const observer = new PerformanceObserverSink(options);
+
+    observer.subscribe(); // Create the PerformanceObserver first
+
+    MockPerformanceObserver.lastInstance()?.emitMark('test-mark');
+
+    observer.flush();
+
+    expect(encode).toHaveBeenCalledWith({
+      name: 'test-mark',
+      entryType: 'mark',
+      startTime: 0,
+      duration: 0,
+    });
+    expect(sink.getWrittenItems()).toStrictEqual(['test-mark:mark']);
+  });
+
+  it('handles flush gracefully when not connected', () => {
+    const observer = new PerformanceObserverSink(options);
+
+    observer.flush();
+
+    expect(encode).not.toHaveBeenCalled();
+    expect(sink.getWrittenItems()).toStrictEqual([]);
+  });
+
+  it('disconnects PerformanceObserver', () => {
+    const observer = new PerformanceObserverSink(options);
+
+    observer.subscribe();
+    observer.unsubscribe();
+
+    expect(observer.isSubscribed()).toBe(false);
+  });
+
+  it('handles disconnect gracefully when not connected', () => {
+    const observer = new PerformanceObserverSink(options);
+
+    observer.unsubscribe();
+
+    expect(observer.isSubscribed()).toBe(false);
+  });
+
+  it('reports connected state correctly', () => {
+    const observer = new PerformanceObserverSink(options);
+
+    expect(observer.isSubscribed()).toBe(false);
+
+    observer.subscribe();
+
+    expect(observer.isSubscribed()).toBe(true);
+  });
+
+  it('reports disconnected state correctly', () => {
+    const observer = new PerformanceObserverSink(options);
+
+    observer.subscribe();
+    observer.unsubscribe();
+
+    expect(observer.isSubscribed()).toBe(false);
+  });
+});