Add openui cloud observability integration in react-lang

packages/react-lang/package.json

@@ -7,6 +7,13 @@
   "main": "dist/index.cjs",
   "module": "dist/index.mjs",
   "types": "dist/index.d.cts",
+  "typesVersions": {
+    "*": {
+      "observability": [
+        "dist/observability/index.d.cts"
+      ]
+    }
+  },
   "sideEffects": [
     "./dist/index.mjs",
     "./dist/index.cjs",
@@ -37,6 +44,16 @@
         "types": "./dist/index.d.cts",
         "default": "./dist/index.cjs"
       }
+    },
+    "./observability": {
+      "import": {
+        "types": "./dist/observability/index.d.mts",
+        "default": "./dist/observability/index.mjs"
+      },
+      "require": {
+        "types": "./dist/observability/index.d.cts",
+        "default": "./dist/observability/index.cjs"
+      }
     }
   },
   "react-native": "./dist/index.native.cjs",

packages/react-lang/src/hooks/useStreamingObservability.ts

@@ -1,6 +1,13 @@
 import type { OpenUIError, ParseResult } from "@openuidev/lang-core";
 import { observability } from "@openuidev/observability";
 import { useEffect, useRef } from "react";
+import {
+  STREAM_EVENT_KIND,
+  STREAM_PHASE_SETTLED,
+  STREAM_PHASE_STREAMING,
+  type SettledStreamEventDetail,
+  type StreamPhase,
+} from "../observability/events/stream";
 
 type CurrentRef<T> = { current: T };
 
@@ -23,7 +30,7 @@ export interface StreamingObservabilityState {
 
 export interface StreamingObservabilityUpdate {
   id: string;
-  phase: "streaming" | "settled";
+  phase: StreamPhase;
   updateIndex: number;
 }
 
@@ -67,7 +74,7 @@ export function advanceStreamingObservability(
     state.hasPublishedStreamingSnapshot = true;
     state.lastResponse = response;
     state.updateIndex += 1;
-    return { id: state.id, phase: "streaming", updateIndex: state.updateIndex };
+    return { id: state.id, phase: STREAM_PHASE_STREAMING, updateIndex: state.updateIndex };
   }
 
   // A Renderer mounted only for static or historical content never starts a stream.
@@ -81,7 +88,7 @@ export function advanceStreamingObservability(
 
   state.settled = true;
   state.lastSettledErrorKey = settledErrorKey;
-  return { id: state.id, phase: "settled", updateIndex: state.updateIndex };
+  return { id: state.id, phase: STREAM_PHASE_SETTLED, updateIndex: state.updateIndex };
 }
 
 function parserMetadata(result: ParseResult | null) {
@@ -121,7 +128,7 @@ export function useStreamingObservability({
       if (update) {
         observability.info({
           id: update.id,
-          kind: "react-lang:stream",
+          kind: STREAM_EVENT_KIND,
           phase: update.phase,
           updateIndex: update.updateIndex,
           response,
@@ -133,11 +140,11 @@ export function useStreamingObservability({
       return;
     }
 
-    if (update?.phase === "settled") {
+    if (update?.phase === STREAM_PHASE_SETTLED) {
       observability(errors.length > 0 ? "error" : "info", {
         id: update.id,
-        kind: "react-lang:stream",
-        phase: update.phase,
+        kind: STREAM_EVENT_KIND,
+        phase: STREAM_PHASE_SETTLED,
         updateIndex: update.updateIndex,
         response,
         responseLength: response?.length ?? 0,
@@ -148,7 +155,7 @@ export function useStreamingObservability({
           errors.length > 0
             ? `OpenUI Lang settled with ${errors.length} error${errors.length === 1 ? "" : "s"}`
             : "OpenUI Lang settled",
-      });
+      } satisfies SettledStreamEventDetail);
     }
   }, [isStreaming, response, result, errorsRef, errorRevision]);
 }

packages/react-lang/src/observability/core/batcher.test.ts

@@ -0,0 +1,225 @@
+// @vitest-environment jsdom
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { Batcher } from "./batcher";
+import * as transport from "./transport";
+import type { WireEvent } from "./wire";
+
+const transportConfig = {
+  endpoint: "https://ingest.example.com/v1/events",
+  apiKey: "test-key",
+  debug: false,
+};
+
+function wireEvent(index: number): WireEvent {
+  return {
+    id: `event-${index}`,
+    kind: "react-lang:stream",
+    level: "info",
+    timestamp: index,
+    updateIndex: index,
+    errorCount: 0,
+  };
+}
+
+describe("Batcher", () => {
+  beforeEach(() => {
+    vi.spyOn(transport, "sendEnvelope").mockResolvedValue(true);
+    vi.spyOn(transport, "sendEnvelopeBeacon").mockImplementation(() => {});
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.restoreAllMocks();
+  });
+
+  it("flushes on interval when the queue is non-empty", async () => {
+    vi.useFakeTimers();
+    const batcher = new Batcher(transportConfig);
+    batcher.enqueue(wireEvent(1));
+
+    await vi.advanceTimersByTimeAsync(5000);
+
+    expect(transport.sendEnvelope).toHaveBeenCalledTimes(1);
+    await batcher.close();
+  });
+
+  it("flushes when the queue reaches 50 events", async () => {
+    const batcher = new Batcher(transportConfig);
+    for (let index = 0; index < 50; index++) {
+      batcher.enqueue(wireEvent(index));
+    }
+
+    await Promise.resolve();
+    expect(transport.sendEnvelope).toHaveBeenCalledTimes(1);
+    const envelope = vi.mocked(transport.sendEnvelope).mock.calls[0]?.[0];
+    expect(envelope?.events).toHaveLength(50);
+    await batcher.close();
+  });
+
+  it("explicit flush resolves true when transport accepts every batch", async () => {
+    const batcher = new Batcher(transportConfig);
+    batcher.enqueue(wireEvent(1));
+    batcher.enqueue(wireEvent(2));
+
+    await expect(batcher.flush()).resolves.toBe(true);
+    expect(transport.sendEnvelope).toHaveBeenCalledTimes(1);
+    await batcher.close();
+  });
+
+  it("explicit flush resolves false when transport drops a batch", async () => {
+    vi.mocked(transport.sendEnvelope).mockResolvedValue(false);
+    const batcher = new Batcher(transportConfig);
+    batcher.enqueue(wireEvent(1));
+
+    await expect(batcher.flush()).resolves.toBe(false);
+    await batcher.close();
+  });
+
+  it("close resolves false when transport drops during final flush", async () => {
+    vi.mocked(transport.sendEnvelope).mockResolvedValue(false);
+    const batcher = new Batcher(transportConfig);
+    batcher.enqueue(wireEvent(1));
+
+    await expect(batcher.close()).resolves.toBe(false);
+  });
+
+  it("flush resolves false within timeoutMs while send is in retry backoff", async () => {
+    vi.useFakeTimers();
+    vi.restoreAllMocks();
+    vi.spyOn(transport, "sendEnvelopeBeacon").mockImplementation(() => {});
+
+    const fetchMock = vi.fn().mockResolvedValue(new Response(null, { status: 500 }));
+    vi.stubGlobal("fetch", fetchMock);
+
+    const batcher = new Batcher(transportConfig);
+    batcher.enqueue(wireEvent(1));
+
+    const flushPromise = batcher.flush(100);
+    await vi.advanceTimersByTimeAsync(100);
+
+    await expect(flushPromise).resolves.toBe(false);
+    await vi.runAllTimersAsync();
+    await batcher.close();
+  });
+
+  it("stamps droppedEvents from queue overflow on the next envelope", async () => {
+    const flushSpy = vi.spyOn(Batcher.prototype, "flushNextBatch").mockResolvedValue({
+      accepted: true,
+      timedOut: false,
+    });
+    const batcher = new Batcher(transportConfig);
+
+    for (let index = 0; index < 501; index++) {
+      batcher.enqueue(wireEvent(index));
+    }
+
+    flushSpy.mockRestore();
+    await batcher.flush();
+
+    const envelopes = vi.mocked(transport.sendEnvelope).mock.calls.map(([payload]) => payload);
+    expect(envelopes.some((payload) => payload.droppedEvents === 1)).toBe(true);
+    await batcher.close();
+  });
+
+  it("flush awaits a threshold-initiated in-flight send and reflects its failure", async () => {
+    let resolveSend!: (accepted: boolean) => void;
+    vi.mocked(transport.sendEnvelope).mockImplementation(
+      () => new Promise<boolean>((resolve) => (resolveSend = resolve)),
+    );
+    const batcher = new Batcher(transportConfig);
+    for (let index = 0; index < 50; index++) {
+      batcher.enqueue(wireEvent(index));
+    }
+    expect(transport.sendEnvelope).toHaveBeenCalledTimes(1);
+
+    let settled = false;
+    const flushPromise = batcher.flush();
+    void flushPromise.then(() => {
+      settled = true;
+    });
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(settled).toBe(false);
+
+    resolveSend(false);
+    await expect(flushPromise).resolves.toBe(false);
+    vi.mocked(transport.sendEnvelope).mockResolvedValue(true);
+    await batcher.close();
+  });
+
+  it("flush resolves false within timeoutMs when the in-flight send is stuck", async () => {
+    vi.useFakeTimers();
+    vi.mocked(transport.sendEnvelope).mockImplementation(() => new Promise<boolean>(() => {}));
+    const batcher = new Batcher(transportConfig);
+    for (let index = 0; index < 50; index++) {
+      batcher.enqueue(wireEvent(index));
+    }
+
+    const flushPromise = batcher.flush(100);
+    await vi.advanceTimersByTimeAsync(100);
+    await expect(flushPromise).resolves.toBe(false);
+
+    const closePromise = batcher.close();
+    await vi.runAllTimersAsync();
+    await expect(closePromise).resolves.toBe(false);
+  });
+
+  it("stamps droppedEvents from a transport-dropped batch on the next accepted envelope", async () => {
+    vi.mocked(transport.sendEnvelope).mockResolvedValueOnce(false);
+    const batcher = new Batcher(transportConfig);
+    batcher.enqueue(wireEvent(1));
+    batcher.enqueue(wireEvent(2));
+    await expect(batcher.flush()).resolves.toBe(false);
+
+    batcher.enqueue(wireEvent(3));
+    await expect(batcher.flush()).resolves.toBe(true);
+
+    const envelopes = vi.mocked(transport.sendEnvelope).mock.calls.map(([payload]) => payload);
+    expect(envelopes[0]?.droppedEvents).toBeUndefined();
+    expect(envelopes[1]?.droppedEvents).toBe(2);
+    await batcher.close();
+  });
+
+  it("re-stamps queue-overflow drops carried by a failed envelope on the next accepted one", async () => {
+    const flushSpy = vi.spyOn(Batcher.prototype, "flushNextBatch").mockResolvedValue({
+      accepted: true,
+      timedOut: false,
+    });
+    const batcher = new Batcher(transportConfig);
+    for (let index = 0; index < 501; index++) {
+      batcher.enqueue(wireEvent(index));
+    }
+    flushSpy.mockRestore();
+
+    vi.mocked(transport.sendEnvelope).mockResolvedValueOnce(false);
+    await expect(batcher.flush()).resolves.toBe(false);
+
+    const envelopes = vi.mocked(transport.sendEnvelope).mock.calls.map(([payload]) => payload);
+    expect(envelopes[0]?.droppedEvents).toBe(1);
+    expect(envelopes[1]?.droppedEvents).toBe(51);
+    await batcher.close();
+  });
+
+  it("pagehide flushes through the beacon path", () => {
+    const batcher = new Batcher(transportConfig);
+    batcher.enqueue(wireEvent(1));
+
+    window.dispatchEvent(new Event("pagehide"));
+
+    expect(transport.sendEnvelopeBeacon).toHaveBeenCalledTimes(1);
+    expect(transport.sendEnvelope).not.toHaveBeenCalled();
+  });
+
+  it("visibilitychange to hidden flushes through the beacon path", () => {
+    const batcher = new Batcher(transportConfig);
+    batcher.enqueue(wireEvent(1));
+
+    Object.defineProperty(document, "visibilityState", {
+      configurable: true,
+      get: () => "hidden",
+    });
+    document.dispatchEvent(new Event("visibilitychange"));
+
+    expect(transport.sendEnvelopeBeacon).toHaveBeenCalledTimes(1);
+  });
+});