soulcraft/brainy
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases 335 Packages Wiki Activity Actions

**feat: enhance WebStreams API compatibility and refactor BrainyData integration**

Browse source 
### 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<unknown>`).
- 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,
This commit is contained in:
David Snelling 2025-06-19 15:03:17 -07:00
parent 5f0d9f4b0a
commit 25f7ab73b4
2 changed files with 69 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

17
src/augmentations/serverSearchAugmentations.ts
View file

@ -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<unknown>
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,

72
src/sequentialPipeline.ts
View file

@ -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<void> {
await streamClassesPromise;
}
/**
* Initialize the pipeline
*
* @returns A promise that resolves when initialization is complete
*/
public async initialize(): Promise<void> {
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<unknown>) => {
// Handle the processed result if needed
if (connection.send && typeof connection.send === 'function') {
try {
@ -384,6 +430,9 @@ export class SequentialPipeline {
readableStream?: ReadableStream<unknown>,
writableStream?: WritableStream<unknown>
}> {
// 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);