feat(utils): add WriteAheadLog classes (#1210)

Author: BioPhoton

packages/utils/eslint.config.js

@@ -12,6 +12,12 @@ export default tseslint.config(
       },
     },
   },
+  {
+    files: ['packages/utils/src/lib/**/wal*.ts'],
+    rules: {
+      'n/no-sync': 'off',
+    },
+  },
   {
     files: ['**/*.json'],
     rules: {

packages/utils/mocks/sink.mock.ts

@@ -1,30 +1,61 @@
-import type { Sink } from '../src/lib/sink-source.type';
+import type { AppendableSink, Codec } from '../src/lib/wal.js';
 
-export class MockSink implements Sink<string, string> {
+export class MockFileSink implements AppendableSink<string> {
   private writtenItems: string[] = [];
   private closed = false;
 
+  constructor(options?: { file?: string; codec?: Codec<string> }) {
+    const file = options?.file || '/tmp/mock-sink.log';
+    const codec = options?.codec || {
+      encode: (input: string) => input,
+      decode: (data: string) => data,
+    };
+  }
+
+  #fd: number | null = null;
+
+  get path(): string {
+    return '/tmp/mock-sink.log';
+  }
+
+  getPath(): string {
+    return this.path;
+  }
+
   open(): void {
-    this.closed = false;
+    this.#fd = 1; // Mock file descriptor
   }
 
-  write(input: string): void {
-    this.writtenItems.push(input);
+  append(v: string): void {
+    this.writtenItems.push(v);
   }
 
   close(): void {
+    this.#fd = null;
     this.closed = true;
   }
 
   isClosed(): boolean {
-    return this.closed;
+    return this.#fd === null;
+  }
+
+  recover(): any {
+    return {
+      records: this.writtenItems,
+      errors: [],
+      partialTail: null,
+    };
   }
 
-  encode(input: string): string {
-    return `${input}-${this.constructor.name}-encoded`;
+  repack(): void {
+    // Mock implementation - do nothing
   }
 
   getWrittenItems(): string[] {
     return [...this.writtenItems];
   }
+
+  clearWrittenItems(): void {
+    this.writtenItems = [];
+  }
 }

packages/utils/src/lib/clock-epoch.unit.test.ts

@@ -3,7 +3,7 @@ import { defaultClock, epochClock } from './clock-epoch.js';
 describe('epochClock', () => {
   it('should create epoch clock with defaults', () => {
     const c = epochClock();
-    expect(c.timeOriginMs).toBe(500_000);
+    expect(c.timeOriginMs).toBe(1_700_000_000_000);
     expect(c.tid).toBe(2);
     expect(c.pid).toBe(10_001);
     expect(c.fromEpochMs).toBeFunction();
@@ -32,8 +32,8 @@ describe('epochClock', () => {
 
   it('should support performance clock by default for epochNowUs', () => {
     const c = epochClock();
-    expect(c.timeOriginMs).toBe(500_000);
-    expect(c.epochNowUs()).toBe(500_000_000);
+    expect(c.timeOriginMs).toBe(1_700_000_000_000);
+    expect(c.epochNowUs()).toBe(1_700_000_000_000_000);
   });
 
   it.each([
@@ -55,8 +55,8 @@ describe('epochClock', () => {
   });
 
   it.each([
-    [0, 500_000_000],
-    [1000, 501_000_000],
+    [0, 1_700_000_000_000_000],
+    [1000, 1_700_000_001_000_000],
   ])(
     'should convert performance milliseconds to microseconds',
     (perfMs, expected) => {

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

@@ -1,21 +1,21 @@
 import { type PerformanceEntry, performance } from 'node:perf_hooks';
 import type { MockedFunction } from 'vitest';
-import { MockSink } from '../../mocks/sink.mock';
+import { MockFileSink } from '../../mocks/sink.mock';
 import {
   type PerformanceObserverOptions,
   PerformanceObserverSink,
 } from './performance-observer.js';
 
 describe('PerformanceObserverSink', () => {
   let encode: MockedFunction<(entry: PerformanceEntry) => string[]>;
-  let sink: MockSink;
+  let sink: MockFileSink;
   let options: PerformanceObserverOptions<string>;
 
   const awaitObserverCallback = () =>
     new Promise(resolve => setTimeout(resolve, 10));
 
   beforeEach(() => {
-    sink = new MockSink();
+    sink = new MockFileSink();
     encode = vi.fn((entry: PerformanceEntry) => [
       `${entry.name}:${entry.entryType}`,
     ]);

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

@@ -4,26 +4,24 @@ import {
   type PerformanceObserverEntryList,
   performance,
 } from 'node:perf_hooks';
-import type { Buffered, Encoder, Observer, Sink } from './sink-source.type';
+import type { AppendableSink } from './wal.js';
 
 const OBSERVED_TYPES = ['mark', 'measure'] as const;
 type ObservedEntryType = 'mark' | 'measure';
 export const DEFAULT_FLUSH_THRESHOLD = 20;
 
 export type PerformanceObserverOptions<T> = {
-  sink: Sink<T, unknown>;
+  sink: AppendableSink<T>;
   encode: (entry: PerformanceEntry) => T[];
   buffered?: boolean;
   flushThreshold?: number;
 };
 
-export class PerformanceObserverSink<T>
-  implements Observer, Buffered, Encoder<PerformanceEntry, T[]>
-{
+export class PerformanceObserverSink<T> {
   #encode: (entry: PerformanceEntry) => T[];
   #buffered: boolean;
   #flushThreshold: number;
-  #sink: Sink<T, unknown>;
+  #sink: AppendableSink<T>;
   #observer: PerformanceObserver | undefined;
 
   #pendingCount = 0;
@@ -84,7 +82,7 @@ export class PerformanceObserverSink<T>
       try {
         fresh
           .flatMap(entry => this.encode(entry))
-          .forEach(item => this.#sink.write(item));
+          .forEach(item => this.#sink.append(item));
 
         this.#written.set(t, written + fresh.length);
       } catch (error) {

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

@@ -1,27 +1,28 @@
 import { type PerformanceEntry, performance } from 'node:perf_hooks';
 import type { MockedFunction } from 'vitest';
 import { MockPerformanceObserver } from '@code-pushup/test-utils';
-import { MockSink } from '../../mocks/sink.mock';
+import { MockFileSink } from '../../mocks/sink.mock';
 import {
   type PerformanceObserverOptions,
   PerformanceObserverSink,
 } from './performance-observer.js';
+import type { Codec } from './wal.js';
 
 describe('PerformanceObserverSink', () => {
   let encode: MockedFunction<(entry: PerformanceEntry) => string[]>;
-  let sink: MockSink;
+  let sink: MockFileSink;
   let options: PerformanceObserverOptions<string>;
 
   beforeEach(() => {
     vi.clearAllMocks();
-    sink = new MockSink();
+    sink = new MockFileSink();
     encode = vi.fn((entry: PerformanceEntry) => [
       `${entry.name}:${entry.entryType}`,
     ]);
     options = {
       sink,
       encode,
-      // we test buffered behavior separately
+
       flushThreshold: 1,
     };
 
@@ -43,24 +44,21 @@ describe('PerformanceObserverSink', () => {
         }),
     ).not.toThrow();
     expect(MockPerformanceObserver.instances).toHaveLength(0);
-    // Instance creation covers the default flushThreshold assignment
   });
 
   it('automatically flushes when pendingCount reaches flushThreshold', () => {
     const observer = new PerformanceObserverSink({
       sink,
       encode,
-      flushThreshold: 2, // Set threshold to 2
+      flushThreshold: 2,
     });
     observer.subscribe();
 
     const mockObserver = MockPerformanceObserver.lastInstance();
 
-    // Emit 1 entry - should not trigger flush yet (pendingCount = 1 < 2)
     mockObserver?.emitMark('first-mark');
     expect(sink.getWrittenItems()).toStrictEqual([]);
 
-    // Emit 1 more entry - should trigger flush (pendingCount = 2 >= 2)
     mockObserver?.emitMark('second-mark');
     expect(sink.getWrittenItems()).toStrictEqual([
       'first-mark:mark',
@@ -135,7 +133,7 @@ describe('PerformanceObserverSink', () => {
   it('internal PerformanceObserver should process observed entries', () => {
     const observer = new PerformanceObserverSink({
       ...options,
-      flushThreshold: 20, // Disable automatic flushing for this test
+      flushThreshold: 20,
     });
     observer.subscribe();
 
@@ -248,12 +246,22 @@ describe('PerformanceObserverSink', () => {
   });
 
   it('flush wraps sink write errors with descriptive error message', () => {
-    const failingSink = {
-      write: vi.fn(() => {
+    const failingCodec: Codec<string> = {
+      encode: () => {
         throw new Error('Sink write failed');
-      }),
+      },
+      decode: (data: string) => data,
     };
 
+    const failingSink = new MockFileSink({
+      file: '/test/path',
+      codec: failingCodec,
+    });
+
+    vi.spyOn(failingSink, 'append').mockImplementation(() => {
+      throw new Error('Sink write failed');
+    });
+
     const observer = new PerformanceObserverSink({
       sink: failingSink as any,
       encode,
@@ -298,4 +306,26 @@ describe('PerformanceObserverSink', () => {
       }),
     );
   });
+
+  it('accepts custom sinks with append method', () => {
+    const collectedItems: string[] = [];
+    const customSink = {
+      // eslint-disable-next-line functional/immutable-data
+      append: (item: string) => collectedItems.push(item),
+    };
+
+    const observer = new PerformanceObserverSink({
+      sink: customSink,
+      encode: (entry: PerformanceEntry) => [`${entry.name}:${entry.duration}`],
+    });
+
+    observer.subscribe();
+
+    const mockObserver = MockPerformanceObserver.lastInstance();
+    mockObserver?.emitMark('test-mark');
+
+    observer.flush();
+
+    expect(collectedItems).toContain('test-mark:0');
+  });
 });

packages/utils/src/lib/profiler/constants.ts

@@ -1 +1,6 @@
 export const PROFILER_ENABLED_ENV_VAR = 'CP_PROFILING';
+export const PROFILER_COORDINATOR_FLAG_ENV_VAR = 'CP_PROFILER_COORDINATOR';
+export const PROFILER_ORIGIN_PID_ENV_VAR = 'CP_PROFILER_ORIGIN_PID';
+export const PROFILER_DIRECTORY_ENV_VAR = 'CP_PROFILER_DIR';
+export const PROFILER_BASE_NAME = 'trace';
+export const PROFILER_DIRECTORY = './tmp/profiles';

packages/utils/src/lib/profiler/profiler.ts

@@ -1,4 +1,5 @@
 import process from 'node:process';
+import { threadId } from 'node:worker_threads';
 import { isEnvVarEnabled } from '../env.js';
 import {
   type ActionTrackConfigs,
@@ -16,6 +17,14 @@ import type {
 } from '../user-timing-extensibility-api.type.js';
 import { PROFILER_ENABLED_ENV_VAR } from './constants.js';
 
+/**
+ * Generates a unique profiler ID based on performance time origin, process ID, thread ID, and instance count.
+ */
+export function getProfilerId() {
+  // eslint-disable-next-line functional/immutable-data
+  return `${Math.round(performance.timeOrigin)}.${process.pid}.${threadId}.${++Profiler.instanceCount}`;
+}
+
 /**
  * Configuration options for creating a Profiler instance.
  *
@@ -59,6 +68,8 @@ export type ProfilerOptions<T extends ActionTrackConfigs = ActionTrackConfigs> =
  *
  */
 export class Profiler<T extends ActionTrackConfigs> {
+  static instanceCount = 0;
+  readonly id = getProfilerId();
   #enabled: boolean;
   readonly #defaults: ActionTrackEntryPayload;
   readonly tracks: Record<keyof T, ActionTrackEntryPayload> | undefined;

packages/utils/src/lib/profiler/profiler.unit.test.ts

@@ -1,6 +1,18 @@
 import { performance } from 'node:perf_hooks';
+import { threadId } from 'node:worker_threads';
 import type { ActionTrackEntryPayload } from '../user-timing-extensibility-api.type.js';
-import { Profiler, type ProfilerOptions } from './profiler.js';
+import { Profiler, type ProfilerOptions, getProfilerId } from './profiler.js';
+
+describe('getProfilerId', () => {
+  it('should generate a unique id per process', () => {
+    expect(getProfilerId()).toBe(
+      `${Math.round(performance.timeOrigin)}.${process.pid}.${threadId}.1`,
+    );
+    expect(getProfilerId()).toBe(
+      `${Math.round(performance.timeOrigin)}.${process.pid}.${threadId}.2`,
+    );
+  });
+});
 
 describe('Profiler', () => {
   const getProfiler = (overrides?: Partial<ProfilerOptions>) =>