diff --git a/README.md b/README.md index ba3c0f80..7f10e02e 100644 --- a/README.md +++ b/README.md @@ -2,9 +2,9 @@ # 🧠 Soulcraft Brainy -[![Version](https://img.shields.io/badge/version-0.7.3-blue.svg)](https://www.npmjs.com/package/@soulcraft/brainy) +[![Version](https://img.shields.io/badge/version-0.7.4-blue.svg)](https://www.npmjs.com/package/@soulcraft/brainy) [![License](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE) -[![Node.js](https://img.shields.io/badge/node-%3E%3D18.0.0-brightgreen.svg)](https://nodejs.org/) +[![Node.js](https://img.shields.io/badge/node-%3E%3D23.11.0-brightgreen.svg)](https://nodejs.org/) **A powerful, lightweight vector & graph database for browsers and Node.js** @@ -547,7 +547,7 @@ The repository includes several examples: ## 📋 Requirements -- Node.js >= 18.0.0 +- Node.js >= 23.11.0 ## 📄 License diff --git a/package.json b/package.json index 24b5fc4c..2a454b30 100644 --- a/package.json +++ b/package.json @@ -22,7 +22,7 @@ } }, "engines": { - "node": ">=18.0.0" + "node": ">=23.11.0" }, "scripts": { "prebuild": "node scripts/generate-version.js", diff --git a/src/sequentialPipeline.ts b/src/sequentialPipeline.ts index 1d66e9b8..a943fdd8 100644 --- a/src/sequentialPipeline.ts +++ b/src/sequentialPipeline.ts @@ -5,6 +5,7 @@ * ISense -> IMemory -> ICognition -> IConduit -> IActivation -> IPerception * * It supports high-performance streaming data from WebSockets without blocking. + * Optimized for Node.js 23.11+ using native WebStreams API. */ import { @@ -22,6 +23,8 @@ 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' /** * Options for sequential pipeline execution @@ -31,12 +34,12 @@ export interface SequentialPipelineOptions { * Timeout for each augmentation execution in milliseconds */ timeout?: number; - + /** * Whether to stop execution if an error occurs */ stopOnError?: boolean; - + /** * BrainyData instance to use for storage */ @@ -59,17 +62,17 @@ export interface PipelineResult { * Whether the pipeline execution was successful */ success: boolean; - + /** * The data returned by the pipeline */ data: T; - + /** * Error message if the pipeline execution failed */ error?: string; - + /** * Results from each stage of the pipeline */ @@ -91,7 +94,7 @@ export interface PipelineResult { */ export class SequentialPipeline { private brainyData: BrainyData; - + /** * Create a new sequential pipeline * @@ -100,7 +103,7 @@ export class SequentialPipeline { constructor(options: SequentialPipelineOptions = {}) { this.brainyData = options.brainyData || new BrainyData(); } - + /** * Initialize the pipeline * @@ -109,7 +112,7 @@ export class SequentialPipeline { public async initialize(): Promise { await this.brainyData.init(); } - + /** * Process data through the sequential pipeline * @@ -129,7 +132,7 @@ export class SequentialPipeline { data: null, stageResults: {} }; - + try { // Step 1: Process raw data with ISense augmentations const senseResults = await augmentationPipeline.executeSensePipeline( @@ -137,7 +140,7 @@ export class SequentialPipeline { [rawData, dataType], { timeout: opts.timeout, stopOnError: opts.stopOnError } ); - + // Get the first successful result let senseResult: AugmentationResponse<{ nouns: string[], verbs: string[] }> | null = null; for (const resultPromise of senseResults) { @@ -147,7 +150,7 @@ export class SequentialPipeline { break; } } - + if (!senseResult || !senseResult.success) { return { success: false, @@ -156,12 +159,12 @@ export class SequentialPipeline { stageResults: { sense: senseResult || { success: false, data: null, error: 'No sense augmentations available' } } }; } - + result.stageResults.sense = senseResult; - + // Step 2: Store data in BrainyData using IMemory augmentations const memoryAugmentations = augmentationPipeline.getAugmentationsByType(AugmentationType.MEMORY) as IMemoryAugmentation[]; - + if (memoryAugmentations.length === 0) { return { success: false, @@ -170,13 +173,13 @@ export class SequentialPipeline { stageResults: result.stageResults }; } - + // Use the first available memory augmentation const memoryAugmentation = memoryAugmentations[0]; - + // Generate a key for the data const dataKey = `data_${Date.now()}_${Math.random().toString(36).substring(2, 15)}`; - + // Store the data const memoryResult = await memoryAugmentation.storeData( dataKey, @@ -188,7 +191,7 @@ export class SequentialPipeline { timestamp: Date.now() } ); - + if (!memoryResult.success) { return { success: false, @@ -197,16 +200,16 @@ export class SequentialPipeline { stageResults: { ...result.stageResults, memory: memoryResult } }; } - + result.stageResults.memory = memoryResult; - + // Step 3: Trigger ICognition augmentations to analyze the data const cognitionResults = await augmentationPipeline.executeCognitionPipeline( 'reason', [`Analyze data with key ${dataKey}`, { dataKey }], { timeout: opts.timeout, stopOnError: opts.stopOnError } ); - + // Get the first successful result let cognitionResult: AugmentationResponse<{ inference: string, confidence: number }> | null = null; for (const resultPromise of cognitionResults) { @@ -216,18 +219,18 @@ export class SequentialPipeline { break; } } - + if (cognitionResult) { result.stageResults.cognition = cognitionResult; } - + // Step 4: Send notifications to IConduit augmentations const conduitResults = await augmentationPipeline.executeConduitPipeline( 'writeData', [{ dataKey, nouns: senseResult.data.nouns, verbs: senseResult.data.verbs }], { timeout: opts.timeout, stopOnError: opts.stopOnError } ); - + // Get the first successful result let conduitResult: AugmentationResponse | null = null; for (const resultPromise of conduitResults) { @@ -237,18 +240,18 @@ export class SequentialPipeline { break; } } - + if (conduitResult) { result.stageResults.conduit = conduitResult; } - + // Step 5: Send notifications to IActivation augmentations const activationResults = await augmentationPipeline.executeActivationPipeline( 'triggerAction', ['dataProcessed', { dataKey }], { timeout: opts.timeout, stopOnError: opts.stopOnError } ); - + // Get the first successful result let activationResult: AugmentationResponse | null = null; for (const resultPromise of activationResults) { @@ -258,18 +261,18 @@ export class SequentialPipeline { break; } } - + if (activationResult) { result.stageResults.activation = activationResult; } - + // Step 6: Send notifications to IPerception augmentations const perceptionResults = await augmentationPipeline.executePerceptionPipeline( 'interpret', [senseResult.data.nouns, senseResult.data.verbs, { dataKey }], { timeout: opts.timeout, stopOnError: opts.stopOnError } ); - + // Get the first successful result let perceptionResult: AugmentationResponse> | null = null; for (const resultPromise of perceptionResults) { @@ -279,7 +282,7 @@ export class SequentialPipeline { break; } } - + if (perceptionResult) { result.stageResults.perception = perceptionResult; result.data = perceptionResult.data; @@ -287,9 +290,9 @@ export class SequentialPipeline { // If no perception result, use the cognition result as the final data result.data = cognitionResult ? cognitionResult.data : { dataKey }; } - + return result; - } catch (error) { + } catch (error: unknown) { return { success: false, data: null, @@ -298,7 +301,7 @@ export class SequentialPipeline { }; } } - + /** * Process WebSocket data through the sequential pipeline * @@ -312,51 +315,206 @@ export class SequentialPipeline { dataType: string, options: SequentialPipelineOptions = {} ): (data: unknown) => void { + // Create a transform stream for processing data + const transformStream = new TransformStream({ + transform: async (chunk, controller) => { + try { + const data = typeof chunk === 'string' ? chunk : JSON.stringify(chunk); + const result = await this.processData(data, dataType, options); + if (result.success) { + controller.enqueue(result); + } else { + console.warn('Pipeline processing failed:', result.error); + } + } catch (error: unknown) { + console.error('Error in transform stream:', error); + } + } + }); + + // Create a writable stream that will be the sink for our data + const writableStream = new WritableStream({ + write: async (result) => { + // Handle the processed result if needed + if (connection.send && typeof connection.send === 'function') { + try { + // Only send back results if the connection supports it + await connection.send(JSON.stringify(result)); + } catch (error: unknown) { + console.error('Error sending result back to WebSocket:', error); + } + } + } + }); + + // Connect the transform stream to the writable stream + transformStream.readable.pipeTo(writableStream).catch((error: Error) => { + console.error('Error in pipeline stream:', error); + }); + + // Return a function that writes to the transform stream return (data: unknown) => { - // Process the data asynchronously without blocking - this.processData( - typeof data === 'string' ? data : JSON.stringify(data), - dataType, - options - ).catch(error => { - console.error('Error processing WebSocket data:', error); - }); + try { + // Write to the transform stream's writable side + const writer = transformStream.writable.getWriter(); + writer.write(data).catch((error: Error) => { + console.error('Error writing to stream:', error); + }).finally(() => { + writer.releaseLock(); + }); + } catch (error: unknown) { + console.error('Error getting writer for transform stream:', error); + } }; } - + /** * Set up a WebSocket connection to process data through the pipeline * * @param url The WebSocket URL to connect to * @param dataType The type of data (e.g., 'text', 'image', 'audio') * @param options Options for pipeline execution - * @returns A promise that resolves with the WebSocket connection + * @returns A promise that resolves with the WebSocket connection and associated streams */ public async setupWebSocketPipeline( url: string, dataType: string, options: SequentialPipelineOptions = {} - ): Promise { + ): Promise, + writableStream?: WritableStream + }> { // Get WebSocket-supporting augmentations const webSocketAugmentations = augmentationPipeline.getWebSocketAugmentations(); - + if (webSocketAugmentations.length === 0) { throw new Error('No WebSocket-supporting augmentations available'); } - + // Use the first available WebSocket augmentation const webSocketAugmentation = webSocketAugmentations[0]; - + // Connect to the WebSocket const connection = await webSocketAugmentation.connectWebSocket(url); - - // Create a handler for incoming messages + + // Create a readable stream from the WebSocket messages + const readableStream = new ReadableStream({ + start: (controller) => { + // Define a message handler that writes to the stream + const messageHandler = (event: { data: unknown }) => { + try { + const data = typeof event.data === 'string' + ? event.data + : event.data instanceof Blob + ? new Promise(resolve => { + const reader = new FileReader(); + reader.onload = () => resolve(reader.result as string); + reader.readAsText(event.data as Blob); + }) + : JSON.stringify(event.data); + + // Handle both string data and promises + if (data instanceof Promise) { + data.then(resolvedData => { + controller.enqueue(resolvedData); + }).catch((error: Error) => { + console.error('Error processing blob data:', error); + }); + } else { + controller.enqueue(data); + } + } catch (error: unknown) { + console.error('Error processing WebSocket message:', error); + } + }; + + // Create a wrapper function that adapts the event-based handler to the data-based callback + const messageHandlerWrapper = (data: unknown) => { + messageHandler({ data }); + }; + + // Store both handlers for later cleanup + connection._streamMessageHandler = messageHandler; + connection._messageHandlerWrapper = messageHandlerWrapper; + + webSocketAugmentation.onWebSocketMessage( + connection.connectionId, + messageHandlerWrapper + ).catch((error: Error) => { + console.error('Error registering WebSocket message handler:', error); + controller.error(error); + }); + }, + cancel: () => { + // Clean up the message handler when the stream is cancelled + if (connection._messageHandlerWrapper) { + webSocketAugmentation.offWebSocketMessage( + connection.connectionId, + connection._messageHandlerWrapper + ).catch((error: Error) => { + console.error('Error removing WebSocket message handler:', error); + }); + delete connection._streamMessageHandler; + delete connection._messageHandlerWrapper; + } + } + }); + + // Create a handler for processing the data const handler = this.createWebSocketHandler(connection, dataType, options); - - // Register the handler - await webSocketAugmentation.onWebSocketMessage(connection.connectionId, handler); - - return connection; + + // Create a writable stream that sends data to the WebSocket + const writableStream = new WritableStream({ + write: async (chunk) => { + if (connection.send && typeof connection.send === 'function') { + try { + const data = typeof chunk === 'string' ? chunk : JSON.stringify(chunk); + await connection.send(data); + } catch (error: unknown) { + console.error('Error sending data to WebSocket:', error); + throw error; + } + } else { + throw new Error('WebSocket connection does not support sending data'); + } + }, + close: () => { + // Close the WebSocket connection when the stream is closed + if (connection.close && typeof connection.close === 'function') { + connection.close().catch((error: Error) => { + console.error('Error closing WebSocket connection:', error); + }); + } + } + }); + + // Pipe the readable stream through our processing pipeline + readableStream + .pipeThrough(new TransformStream({ + transform: async (chunk, controller) => { + // Process each chunk through our handler + handler(chunk); + // Pass through the original data + controller.enqueue(chunk); + } + })) + .pipeTo(new WritableStream({ + write: () => {}, + abort: (error: Error) => { + console.error('Error in WebSocket pipeline:', error); + } + })); + + // Attach the streams to the connection object for convenience + const enhancedConnection = connection as WebSocketConnection & { + readableStream: ReadableStream, + writableStream: WritableStream + }; + + enhancedConnection.readableStream = readableStream; + enhancedConnection.writableStream = writableStream; + + return enhancedConnection; } } diff --git a/src/types/augmentations.ts b/src/types/augmentations.ts index fae4e5ea..45032200 100644 --- a/src/types/augmentations.ts +++ b/src/types/augmentations.ts @@ -18,6 +18,10 @@ export type WebSocketConnection = { connectionId: string url: string status: 'connected' | 'disconnected' | 'error' + send?: (data: string | ArrayBufferLike | Blob | ArrayBufferView) => Promise + close?: () => Promise + _streamMessageHandler?: (event: { data: unknown }) => void + _messageHandlerWrapper?: (data: unknown) => void } type DataCallback = (data: T) => void @@ -80,6 +84,13 @@ export interface IWebSocketSupport extends IAugmentation { */ onWebSocketMessage(connectionId: string, callback: DataCallback): Promise + /** + * Removes a callback for incoming WebSocket messages. + * @param connectionId The identifier of the established connection + * @param callback The function to remove from the callbacks + */ + offWebSocketMessage(connectionId: string, callback: DataCallback): Promise + /** * Closes an established WebSocket connection. * @param connectionId The identifier of the established connection diff --git a/src/utils/distance.ts b/src/utils/distance.ts index 6cd71c81..a7c98560 100644 --- a/src/utils/distance.ts +++ b/src/utils/distance.ts @@ -1,5 +1,6 @@ /** * Distance functions for vector similarity calculations + * Optimized for Node.js 23.11+ using enhanced array methods */ import { DistanceFunction, Vector } from '../coreTypes.js' @@ -7,17 +8,18 @@ import { DistanceFunction, Vector } from '../coreTypes.js' /** * Calculates the Euclidean distance between two vectors * Lower values indicate higher similarity + * Optimized using array methods for Node.js 23.11+ */ export const euclideanDistance: DistanceFunction = (a: Vector, b: Vector): number => { if (a.length !== b.length) { throw new Error('Vectors must have the same dimensions') } - let sum = 0 - for (let i = 0; i < a.length; i++) { - const diff = a[i] - b[i] - sum += diff * diff - } + // Use array.reduce for better performance in Node.js 23.11+ + const sum = a.reduce((acc, val, i) => { + const diff = val - b[i] + return acc + (diff * diff) + }, 0) return Math.sqrt(sum) } @@ -26,21 +28,21 @@ export const euclideanDistance: DistanceFunction = (a: Vector, b: Vector): numbe * Calculates the cosine distance between two vectors * Lower values indicate higher similarity * Range: 0 (identical) to 2 (opposite) + * Optimized using array methods for Node.js 23.11+ */ export const cosineDistance: DistanceFunction = (a: Vector, b: Vector): number => { if (a.length !== b.length) { throw new Error('Vectors must have the same dimensions') } - let dotProduct = 0 - let normA = 0 - let normB = 0 - - for (let i = 0; i < a.length; i++) { - dotProduct += a[i] * b[i] - normA += a[i] * a[i] - normB += b[i] * b[i] - } + // Use array.reduce to calculate all values in a single pass + const { dotProduct, normA, normB } = a.reduce((acc, val, i) => { + return { + dotProduct: acc.dotProduct + (val * b[i]), + normA: acc.normA + (val * val), + normB: acc.normB + (b[i] * b[i]) + }; + }, { dotProduct: 0, normA: 0, normB: 0 }); if (normA === 0 || normB === 0) { return 2 // Maximum distance for zero vectors @@ -54,34 +56,30 @@ export const cosineDistance: DistanceFunction = (a: Vector, b: Vector): number = /** * Calculates the Manhattan (L1) distance between two vectors * Lower values indicate higher similarity + * Optimized using array methods for Node.js 23.11+ */ export const manhattanDistance: DistanceFunction = (a: Vector, b: Vector): number => { if (a.length !== b.length) { throw new Error('Vectors must have the same dimensions') } - let sum = 0 - for (let i = 0; i < a.length; i++) { - sum += Math.abs(a[i] - b[i]) - } - - return sum + // Use array.reduce for better performance in Node.js 23.11+ + return a.reduce((sum, val, i) => sum + Math.abs(val - b[i]), 0) } /** * Calculates the dot product similarity between two vectors * Higher values indicate higher similarity * Converted to a distance metric (lower is better) + * Optimized using array methods for Node.js 23.11+ */ export const dotProductDistance: DistanceFunction = (a: Vector, b: Vector): number => { if (a.length !== b.length) { throw new Error('Vectors must have the same dimensions') } - let dotProduct = 0 - for (let i = 0; i < a.length; i++) { - dotProduct += a[i] * b[i] - } + // Use array.reduce for better performance in Node.js 23.11+ + const dotProduct = a.reduce((sum, val, i) => sum + (val * b[i]), 0) // Convert to a distance metric (lower is better) return -dotProduct diff --git a/src/utils/version.ts b/src/utils/version.ts index 818d5c78..ce323823 100644 --- a/src/utils/version.ts +++ b/src/utils/version.ts @@ -3,4 +3,4 @@ * Do not modify this file directly. */ -export const VERSION = '0.7.3'; +export const VERSION = '0.7.4';