From 25f7ab73b43085e376d10ad915b48fbddc98bdce Mon Sep 17 00:00:00 2001 From: David Snelling Date: Thu, 19 Jun 2025 15:03:17 -0700 Subject: [PATCH] **feat: enhance WebStreams API compatibility and refactor BrainyData integration** ### Changes: - Introduced environment-independent WebStreams API compatibility: - Added `initializeStreamClasses` for dynamic detection of WebStreams API support in browsers and Node.js. - Implemented fallback mechanism with detailed error messages for unsupported environments. - Ensured asynchronous initialization of stream classes without blocking module execution. - Updated `SequentialPipeline`: - Added `ensureStreamClassesInitialized` method to streamline WebStreams setup across operations. - Enhanced `initialize` to parallelize stream classes and `BrainyData` initialization. - Refactored WebSocket handlers (`createWebSocketHandler` and `createWebSocketStreams`) to support async stream initialization. - Improved TypeScript typings for streams (`TransformStreamDefaultController`, `PipelineResult`). - Adjusted `ServerSearchConduitAugmentation`: - Introduced dependency on `BrainyDataInterface` instead of direct `BrainyData` usage. - Ensured `localDb` is set prior to initialization with an appropriate error message. - Updated methods (`setLocalDb`, `getLocalDb`, and vector operations) to rely on `BrainyDataInterface`. - Enhanced error handling and logging throughout streaming and augmentation processes. ### Purpose: Improved cross-environment compatibility of WebStreams API, ensuring seamless usage in both browsers and Node.js. Refactored `ServerSearchConduitAugmentation` to decouple `BrainyData` dependency, --- .../serverSearchAugmentations.ts | 17 +++-- src/sequentialPipeline.ts | 72 ++++++++++++++++--- 2 files changed, 69 insertions(+), 20 deletions(-) diff --git a/src/augmentations/serverSearchAugmentations.ts b/src/augmentations/serverSearchAugmentations.ts index 84ef93fa..93d593e3 100644 --- a/src/augmentations/serverSearchAugmentations.ts +++ b/src/augmentations/serverSearchAugmentations.ts @@ -15,7 +15,7 @@ import { } from '../types/augmentations.js' import { WebSocketConduitAugmentation } from './conduitAugmentations.js' import { v4 as uuidv4 } from 'uuid' -import { BrainyData } from '../brainyData.js' +import { BrainyDataInterface } from '../types/brainyDataInterface.js' /** * ServerSearchConduitAugmentation @@ -24,7 +24,7 @@ import { BrainyData } from '../brainyData.js' * a server-hosted Brainy instance and storing results locally. */ export class ServerSearchConduitAugmentation extends WebSocketConduitAugmentation { - private localDb: BrainyData | null = null + private localDb: BrainyDataInterface | null = null constructor(name: string = 'server-search-conduit') { super(name) @@ -43,10 +43,9 @@ export class ServerSearchConduitAugmentation extends WebSocketConduitAugmentatio // Initialize the base conduit await super.initialize() - // Initialize local Brainy instance if not provided + // Local DB must be set before initialization if (!this.localDb) { - this.localDb = new BrainyData() - await this.localDb.init() + throw new Error('Local database not set. Call setLocalDb before initializing.') } this.isInitialized = true @@ -60,7 +59,7 @@ export class ServerSearchConduitAugmentation extends WebSocketConduitAugmentatio * Set the local Brainy instance * @param db The Brainy instance to use for local storage */ - setLocalDb(db: BrainyData): void { + setLocalDb(db: BrainyDataInterface): void { this.localDb = db } @@ -68,7 +67,7 @@ export class ServerSearchConduitAugmentation extends WebSocketConduitAugmentatio * Get the local Brainy instance * @returns The local Brainy instance */ - getLocalDb(): BrainyData | null { + getLocalDb(): BrainyDataInterface | null { return this.localDb } @@ -265,7 +264,7 @@ export class ServerSearchConduitAugmentation extends WebSocketConduitAugmentatio const id = await this.localDb.add(data, metadata) // Get the vector and metadata - const noun = await this.localDb.get(id) + const noun = await this.localDb.get(id) as import('../coreTypes.js').VectorDocument if (!noun) { return { @@ -620,7 +619,7 @@ export async function createServerSearchAugmentations( conduitName?: string, activationName?: string, protocols?: string | string[], - localDb?: BrainyData + localDb?: BrainyDataInterface } = {} ): Promise<{ conduit: ServerSearchConduitAugmentation, diff --git a/src/sequentialPipeline.ts b/src/sequentialPipeline.ts index a943fdd8..7d5e727d 100644 --- a/src/sequentialPipeline.ts +++ b/src/sequentialPipeline.ts @@ -23,8 +23,39 @@ import { } from './types/augmentations.js' import { BrainyData } from './brainyData.js' import { augmentationPipeline } from './augmentationPipeline.js' -// Using Node.js 23.11+ native WebStreams API -import { TransformStream, ReadableStream, WritableStream } from 'node:stream/web' +// Use the browser's built-in WebStreams API or Node.js native WebStreams API +// This approach ensures compatibility with both environments +let TransformStream: any, ReadableStream: any, WritableStream: any; + +// Function to initialize the stream classes +const initializeStreamClasses = () => { + // Try to use the browser's built-in WebStreams API first + if (typeof globalThis.TransformStream !== 'undefined' && + typeof globalThis.ReadableStream !== 'undefined' && + typeof globalThis.WritableStream !== 'undefined') { + TransformStream = globalThis.TransformStream; + ReadableStream = globalThis.ReadableStream; + WritableStream = globalThis.WritableStream; + return Promise.resolve(); + } else { + // In Node.js environment, try to import from node:stream/web + // This will be executed in Node.js but not in browsers + return import('node:stream/web') + .then(streamWebModule => { + TransformStream = streamWebModule.TransformStream; + ReadableStream = streamWebModule.ReadableStream; + WritableStream = streamWebModule.WritableStream; + }) + .catch(error => { + console.error('Failed to import WebStreams API:', error); + // Provide fallback implementations or throw a more helpful error + throw new Error('WebStreams API is not available in this environment. Please use a modern browser or Node.js 18+.'); + }); + } +}; + +// Initialize immediately but don't block module execution +const streamClassesPromise = initializeStreamClasses(); /** * Options for sequential pipeline execution @@ -104,13 +135,25 @@ export class SequentialPipeline { this.brainyData = options.brainyData || new BrainyData(); } + /** + * Ensure stream classes are initialized + * @private + */ + private async ensureStreamClassesInitialized(): Promise { + await streamClassesPromise; + } + /** * Initialize the pipeline * * @returns A promise that resolves when initialization is complete */ public async initialize(): Promise { - await this.brainyData.init(); + // Initialize stream classes and BrainyData in parallel + await Promise.all([ + this.ensureStreamClassesInitialized(), + this.brainyData.init() + ]); } /** @@ -310,14 +353,17 @@ export class SequentialPipeline { * @param options Options for pipeline execution * @returns A function to handle incoming WebSocket messages */ - public createWebSocketHandler( + public async createWebSocketHandler( connection: WebSocketConnection, dataType: string, options: SequentialPipelineOptions = {} - ): (data: unknown) => void { + ): Promise<(data: unknown) => void> { + // Ensure stream classes are initialized + await this.ensureStreamClassesInitialized(); + // Create a transform stream for processing data const transformStream = new TransformStream({ - transform: async (chunk, controller) => { + transform: async (chunk: unknown, controller: TransformStreamDefaultController) => { try { const data = typeof chunk === 'string' ? chunk : JSON.stringify(chunk); const result = await this.processData(data, dataType, options); @@ -334,7 +380,7 @@ export class SequentialPipeline { // Create a writable stream that will be the sink for our data const writableStream = new WritableStream({ - write: async (result) => { + write: async (result: PipelineResult) => { // Handle the processed result if needed if (connection.send && typeof connection.send === 'function') { try { @@ -384,6 +430,9 @@ export class SequentialPipeline { readableStream?: ReadableStream, writableStream?: WritableStream }> { + // Ensure stream classes are initialized + await this.ensureStreamClassesInitialized(); + // Get WebSocket-supporting augmentations const webSocketAugmentations = augmentationPipeline.getWebSocketAugmentations(); @@ -399,7 +448,7 @@ export class SequentialPipeline { // Create a readable stream from the WebSocket messages const readableStream = new ReadableStream({ - start: (controller) => { + start: (controller: ReadableStreamDefaultController) => { // Define a message handler that writes to the stream const messageHandler = (event: { data: unknown }) => { try { @@ -461,11 +510,11 @@ export class SequentialPipeline { }); // Create a handler for processing the data - const handler = this.createWebSocketHandler(connection, dataType, options); + const handlerPromise = this.createWebSocketHandler(connection, dataType, options); // Create a writable stream that sends data to the WebSocket const writableStream = new WritableStream({ - write: async (chunk) => { + write: async (chunk: unknown) => { if (connection.send && typeof connection.send === 'function') { try { const data = typeof chunk === 'string' ? chunk : JSON.stringify(chunk); @@ -491,8 +540,9 @@ export class SequentialPipeline { // Pipe the readable stream through our processing pipeline readableStream .pipeThrough(new TransformStream({ - transform: async (chunk, controller) => { + transform: async (chunk: unknown, controller: TransformStreamDefaultController) => { // Process each chunk through our handler + const handler = await handlerPromise; handler(chunk); // Pass through the original data controller.enqueue(chunk);