From 1635cc229c982821d93cadb91aed8d789d31b45e Mon Sep 17 00:00:00 2001 From: David Snelling Date: Fri, 6 Jun 2025 08:49:25 -0700 Subject: [PATCH] feat: add WebSocket and WebRTC conduit augmentations with examples Implemented WebSocketConduitAugmentation and WebRTCConduitAugmentation for syncing Brainy instances. Added comprehensive usage examples to showcase data synchronization via WebSocket and WebRTC. --- examples/conduitAugmentationExample.js | 236 ++++ src/augmentations/conduitAugmentations.ts | 1409 +++++++++++++++++++++ src/types/augmentations.ts | 6 +- 3 files changed, 1648 insertions(+), 3 deletions(-) create mode 100644 examples/conduitAugmentationExample.js create mode 100644 src/augmentations/conduitAugmentations.ts diff --git a/examples/conduitAugmentationExample.js b/examples/conduitAugmentationExample.js new file mode 100644 index 00000000..b0c8e739 --- /dev/null +++ b/examples/conduitAugmentationExample.js @@ -0,0 +1,236 @@ +/** + * Conduit Augmentation Example + * + * This example demonstrates how to use the conduit augmentations to sync Brainy instances: + * + * - WebSocket Conduit: For syncing between browsers and servers, or between servers. + * WebSockets cannot be used for direct browser-to-browser communication without a server in the middle. + * + * - WebRTC Conduit: For direct peer-to-peer syncing between browsers. + * This is the recommended approach for browser-to-browser communication. + */ + +import { + BrainyData, + augmentationPipeline, + createConduitAugmentation, + NounType, + VerbType +} from '@soulcraft/brainy'; + +/** + * Example of using WebSocket conduit augmentation to sync Brainy instances + */ +async function webSocketSyncExample() { + console.log('Starting WebSocket sync example...'); + + // Create and initialize the database + const db = new BrainyData(); + await db.init(); + + // Create a WebSocket conduit augmentation + const wsConduit = await createConduitAugmentation('websocket', 'websocket-sync-example'); + + // Register the augmentation with the pipeline + augmentationPipeline.register(wsConduit); + + // Add some data to the local database + const catId = await db.add("Cats are independent pets", { + noun: NounType.Thing, + category: 'animal' + }); + + const dogId = await db.add("Dogs are loyal companions", { + noun: NounType.Thing, + category: 'animal' + }); + + // Add a relationship between items + await db.addVerb(catId, dogId, undefined, { + type: VerbType.RelatedTo, + metadata: { + description: 'Both are common household pets' + } + }); + + console.log('Added sample data to local database'); + + try { + // Connect to another Brainy instance (server or browser) + // Note: You need to have a WebSocket server running at this URL + const connectionResult = await augmentationPipeline.executeConduitPipeline( + 'establishConnection', + ['wss://your-websocket-server.com/brainy-sync', { protocols: 'brainy-sync' }] + ); + + if (connectionResult[0] && (await connectionResult[0]).success) { + const connection = (await connectionResult[0]).data; + console.log('Connected to remote Brainy instance:', connection.url); + + // Read data from the remote instance + const readResult = await augmentationPipeline.executeConduitPipeline( + 'readData', + [{ connectionId: connection.connectionId, query: { type: 'getAllNouns' } }] + ); + + // Process and add the received data to the local instance + if (readResult[0] && (await readResult[0]).success) { + const remoteNouns = (await readResult[0]).data; + console.log(`Received ${remoteNouns.length} nouns from remote instance`); + + for (const noun of remoteNouns) { + await db.add(noun.vector, noun.metadata); + } + + console.log('Added remote nouns to local database'); + } + + // Set up real-time sync by monitoring the stream + await wsConduit.monitorStream(connection.connectionId, async (data) => { + console.log('Received data from stream:', data.type); + + // Handle incoming data (e.g., new nouns, verbs, updates) + if (data.type === 'newNoun') { + await db.add(data.vector, data.metadata); + console.log('Added new noun from remote instance:', data.id); + } else if (data.type === 'newVerb') { + await db.addVerb(data.sourceId, data.targetId, data.vector, data.options); + console.log('Added new verb from remote instance:', data.id); + } + }); + + // Add a new noun and send it to the remote instance + const birdId = await db.add("Birds are fascinating creatures", { + noun: NounType.Thing, + category: 'animal' + }); + + const birdData = await db.get(birdId); + + // Send the new noun to the remote instance + await augmentationPipeline.executeConduitPipeline( + 'writeData', + [ + { + connectionId: connection.connectionId, + data: { + type: 'newNoun', + id: birdId, + vector: birdData.vector, + metadata: birdData.metadata + } + } + ] + ); + + console.log('Sent new noun to remote instance:', birdId); + + // Close the connection when done + await wsConduit.closeWebSocket(connection.connectionId); + console.log('Closed connection to remote instance'); + } else { + console.error('Failed to connect to remote instance'); + } + } catch (error) { + console.error('Error in WebSocket sync example:', error); + } +} + +/** + * Example of using WebRTC conduit augmentation for peer-to-peer sync + */ +async function webRTCSyncExample() { + console.log('Starting WebRTC sync example...'); + + // Create and initialize the database + const db = new BrainyData(); + await db.init(); + + // Create a WebRTC conduit augmentation + const webrtcConduit = await createConduitAugmentation('webrtc', 'webrtc-sync-example'); + + // Register the augmentation with the pipeline + augmentationPipeline.register(webrtcConduit); + + try { + // Connect to a peer using a signaling server + // Note: You need to have a signaling server running and another peer to connect to + const connectionResult = await augmentationPipeline.executeConduitPipeline( + 'establishConnection', + [ + 'peer-id-to-connect-to', + { + signalServerUrl: 'wss://your-signal-server.com', + localPeerId: 'my-peer-id', + iceServers: [{ urls: 'stun:stun.l.google.com:19302' }] + } + ] + ); + + if (connectionResult[0] && (await connectionResult[0]).success) { + const connection = (await connectionResult[0]).data; + console.log('Connected to peer:', connection.url); + + // Set up real-time sync by monitoring the stream + await webrtcConduit.monitorStream(connection.connectionId, async (data) => { + console.log('Received data from peer:', data.type); + + // Handle incoming data (e.g., new nouns, verbs, updates) + if (data.type === 'newNoun') { + await db.add(data.vector, data.metadata); + console.log('Added new noun from peer:', data.id); + } else if (data.type === 'newVerb') { + await db.addVerb(data.sourceId, data.targetId, data.vector, data.options); + console.log('Added new verb from peer:', data.id); + } + }); + + // Add a new noun and send it to the peer + const fishId = await db.add("Fish are aquatic animals", { + noun: NounType.Thing, + category: 'animal' + }); + + const fishData = await db.get(fishId); + + // Send the new noun to the peer + await augmentationPipeline.executeConduitPipeline( + 'writeData', + [ + { + connectionId: connection.connectionId, + data: { + type: 'newNoun', + id: fishId, + vector: fishData.vector, + metadata: fishData.metadata + } + } + ] + ); + + console.log('Sent new noun to peer:', fishId); + + // Close the connection when done + await webrtcConduit.closeWebSocket(connection.connectionId); + console.log('Closed connection to peer'); + } else { + console.error('Failed to connect to peer'); + } + } catch (error) { + console.error('Error in WebRTC sync example:', error); + } +} + +// Run the examples +async function runExamples() { + try { + await webSocketSyncExample(); + console.log('\n-----------------------------------\n'); + await webRTCSyncExample(); + } catch (error) { + console.error('Error running examples:', error); + } +} + +runExamples(); diff --git a/src/augmentations/conduitAugmentations.ts b/src/augmentations/conduitAugmentations.ts new file mode 100644 index 00000000..ba94e647 --- /dev/null +++ b/src/augmentations/conduitAugmentations.ts @@ -0,0 +1,1409 @@ +import { + AugmentationType, + IConduitAugmentation, + IWebSocketSupport, + AugmentationResponse, + WebSocketConnection +} from '../types/augmentations.js' +import { v4 as uuidv4 } from 'uuid' + +/** + * Base class for conduit augmentations that provide data synchronization between Brainy instances + */ +abstract class BaseConduitAugmentation implements IConduitAugmentation { + readonly name: string + readonly description: string = 'Base conduit augmentation' + enabled: boolean = true + protected isInitialized = false + protected connections: Map = new Map() + + constructor(name: string) { + this.name = name + } + + async initialize(): Promise { + if (this.isInitialized) { + return + } + + try { + this.isInitialized = true + } catch (error) { + console.error(`Failed to initialize ${this.name}:`, error) + throw new Error(`Failed to initialize ${this.name}: ${error}`) + } + } + + async shutDown(): Promise { + // Close all connections + for (const [connectionId, connection] of this.connections.entries()) { + try { + if (connection.close) { + await connection.close() + } + } catch (error) { + console.error(`Failed to close connection ${connectionId}:`, error) + } + } + + this.connections.clear() + this.isInitialized = false + } + + async getStatus(): Promise<'active' | 'inactive' | 'error'> { + return this.isInitialized ? 'active' : 'inactive' + } + + abstract establishConnection( + targetSystemId: string, + config: Record + ): Promise> + + abstract readData( + query: Record, + options?: Record + ): Promise> + + abstract writeData( + data: Record, + options?: Record + ): Promise> + + abstract monitorStream( + streamId: string, + callback: (data: unknown) => void + ): Promise + + protected async ensureInitialized(): Promise { + if (!this.isInitialized) { + await this.initialize() + } + } +} + +/** + * WebSocket conduit augmentation for syncing Brainy instances using WebSockets + * + * This conduit is for syncing between browsers and servers, or between servers. + * WebSockets cannot be used for direct browser-to-browser communication without a server in the middle. + */ +export class WebSocketConduitAugmentation extends BaseConduitAugmentation implements IWebSocketSupport { + readonly description = 'Conduit augmentation that syncs Brainy instances using WebSockets' + private webSocketConnections: Map = new Map() + private messageCallbacks: Map void>> = new Map() + + constructor(name: string = 'websocket-conduit') { + super(name) + } + + getType(): AugmentationType { + return AugmentationType.CONDUIT + } + + /** + * Establishes a connection to another Brainy instance + * @param targetSystemId The URL or identifier of the target system + * @param config Configuration options for the connection + */ + async establishConnection( + targetSystemId: string, + config: Record + ): Promise> { + await this.ensureInitialized() + + try { + // For WebSocket connections, targetSystemId should be a WebSocket URL + const url = targetSystemId + const protocols = config.protocols as string | string[] | undefined + + // Create a WebSocket connection + const connection = await this.connectWebSocket(url, protocols) + + // Store the connection + this.connections.set(connection.connectionId, connection) + + return { + success: true, + data: connection + } + } catch (error) { + console.error(`Failed to establish connection to ${targetSystemId}:`, error) + return { + success: false, + data: null as any, + error: `Failed to establish connection: ${error}` + } + } + } + + /** + * Reads data from a connected Brainy instance + * @param query Query parameters for reading data + * @param options Additional options + */ + async readData( + query: Record, + options?: Record + ): Promise> { + await this.ensureInitialized() + + try { + const connectionId = query.connectionId as string + + if (!connectionId) { + throw new Error('connectionId is required for reading data') + } + + const connection = this.webSocketConnections.get(connectionId) + + if (!connection) { + throw new Error(`Connection ${connectionId} not found`) + } + + // Create a request message + const requestMessage = { + type: 'read', + query: query.query || {}, + requestId: uuidv4(), + options + } + + // Send the request + await this.sendWebSocketMessage(connectionId, requestMessage) + + // Return a promise that will be resolved when the response is received + return new Promise((resolve) => { + const responseHandler = (data: unknown) => { + // Check if this is the response to our request + const response = data as any + if (response && response.type === 'readResponse' && response.requestId === requestMessage.requestId) { + // Remove the handler + this.offWebSocketMessage(connectionId, responseHandler) + + // Resolve with the response data + resolve({ + success: response.success, + data: response.data, + error: response.error + }) + } + } + + // Register the response handler + this.onWebSocketMessage(connectionId, responseHandler) + + // Set a timeout to prevent hanging + setTimeout(() => { + this.offWebSocketMessage(connectionId, responseHandler) + resolve({ + success: false, + data: null, + error: 'Timeout waiting for read response' + }) + }, 30000) // 30 second timeout + }) + } catch (error) { + console.error(`Failed to read data:`, error) + return { + success: false, + data: null, + error: `Failed to read data: ${error}` + } + } + } + + /** + * Writes data to a connected Brainy instance + * @param data The data to write + * @param options Additional options + */ + async writeData( + data: Record, + options?: Record + ): Promise> { + await this.ensureInitialized() + + try { + const connectionId = data.connectionId as string + + if (!connectionId) { + throw new Error('connectionId is required for writing data') + } + + const connection = this.webSocketConnections.get(connectionId) + + if (!connection) { + throw new Error(`Connection ${connectionId} not found`) + } + + // Create a write message + const writeMessage = { + type: 'write', + data: data.data || {}, + requestId: uuidv4(), + options + } + + // Send the write message + await this.sendWebSocketMessage(connectionId, writeMessage) + + // Return a promise that will be resolved when the response is received + return new Promise((resolve) => { + const responseHandler = (data: unknown) => { + // Check if this is the response to our request + const response = data as any + if (response && response.type === 'writeResponse' && response.requestId === writeMessage.requestId) { + // Remove the handler + this.offWebSocketMessage(connectionId, responseHandler) + + // Resolve with the response data + resolve({ + success: response.success, + data: response.data, + error: response.error + }) + } + } + + // Register the response handler + this.onWebSocketMessage(connectionId, responseHandler) + + // Set a timeout to prevent hanging + setTimeout(() => { + this.offWebSocketMessage(connectionId, responseHandler) + resolve({ + success: false, + data: null, + error: 'Timeout waiting for write response' + }) + }, 30000) // 30 second timeout + }) + } catch (error) { + console.error(`Failed to write data:`, error) + return { + success: false, + data: null, + error: `Failed to write data: ${error}` + } + } + } + + /** + * Monitors a data stream from a connected Brainy instance + * @param streamId The ID of the stream to monitor (usually a connection ID) + * @param callback Function to call when new data is received + */ + async monitorStream( + streamId: string, + callback: (data: unknown) => void + ): Promise { + await this.ensureInitialized() + + try { + const connection = this.webSocketConnections.get(streamId) + + if (!connection) { + throw new Error(`Connection ${streamId} not found`) + } + + // Register the callback for all messages on this connection + await this.onWebSocketMessage(streamId, callback) + + } catch (error) { + console.error(`Failed to monitor stream ${streamId}:`, error) + throw new Error(`Failed to monitor stream: ${error}`) + } + } + + /** + * Establishes a WebSocket connection + * @param url The WebSocket server URL to connect to + * @param protocols Optional subprotocols + */ + async connectWebSocket( + url: string, + protocols?: string | string[] + ): Promise { + await this.ensureInitialized() + + return new Promise((resolve, reject) => { + try { + // Check if WebSocket is available + if (typeof WebSocket === 'undefined') { + throw new Error('WebSocket is not available in this environment') + } + + // Create a new WebSocket connection + const ws = new WebSocket(url, protocols) + const connectionId = uuidv4() + + // Create a connection object + const connection: WebSocketConnection = { + connectionId, + url, + status: 'disconnected', + send: async (data) => { + if (ws.readyState !== WebSocket.OPEN) { + throw new Error('WebSocket is not open') + } + ws.send(data) + }, + close: async () => { + ws.close() + } + } + + // Set up event handlers + ws.onopen = () => { + connection.status = 'connected' + resolve(connection) + } + + ws.onerror = (error) => { + connection.status = 'error' + console.error(`WebSocket error for ${url}:`, error) + if (ws.readyState !== WebSocket.OPEN) { + reject(new Error(`WebSocket connection failed: ${error}`)) + } + } + + ws.onclose = () => { + connection.status = 'disconnected' + // Remove from connections map + this.webSocketConnections.delete(connectionId) + // Remove all callbacks + this.messageCallbacks.delete(connectionId) + } + + // Create a message handler wrapper that will call all registered callbacks + const messageHandlerWrapper = (data: unknown) => { + const callbacks = this.messageCallbacks.get(connectionId) + if (callbacks) { + for (const callback of callbacks) { + try { + callback(data) + } catch (error) { + console.error(`Error in WebSocket message callback:`, error) + } + } + } + } + + // Store the message handler wrapper + connection._messageHandlerWrapper = messageHandlerWrapper + + // Set up the message handler + ws.onmessage = (event) => { + try { + // Parse the message if it's a string + let data = event.data + if (typeof data === 'string') { + try { + data = JSON.parse(data) + } catch { + // If parsing fails, use the raw string + } + } + + // Call the message handler wrapper + messageHandlerWrapper(data) + } catch (error) { + console.error(`Error handling WebSocket message:`, error) + } + } + + // Store the stream message handler + connection._streamMessageHandler = (event) => ws.onmessage && ws.onmessage(event as any) + + // Store the connection + this.webSocketConnections.set(connectionId, connection) + + // Initialize the callbacks set + this.messageCallbacks.set(connectionId, new Set()) + + } catch (error) { + reject(error) + } + }) + } + + /** + * Sends data through an established WebSocket connection + * @param connectionId The identifier of the established connection + * @param data The data to send (will be serialized if not a string) + */ + async sendWebSocketMessage( + connectionId: string, + data: unknown + ): Promise { + await this.ensureInitialized() + + const connection = this.webSocketConnections.get(connectionId) + + if (!connection) { + throw new Error(`WebSocket connection ${connectionId} not found`) + } + + if (!connection.send) { + throw new Error(`WebSocket connection ${connectionId} does not support sending messages`) + } + + // Serialize the data if it's not already a string or binary + let serializedData: string | ArrayBufferLike | Blob | ArrayBufferView + + if (typeof data === 'string' || + data instanceof ArrayBuffer || + data instanceof Blob || + ArrayBuffer.isView(data)) { + serializedData = data as any + } else { + // Convert to JSON string + serializedData = JSON.stringify(data) + } + + // Send the data + await connection.send(serializedData) + } + + /** + * Registers a callback for incoming WebSocket messages + * @param connectionId The identifier of the established connection + * @param callback The function to call when a message is received + */ + async onWebSocketMessage( + connectionId: string, + callback: (data: unknown) => void + ): Promise { + await this.ensureInitialized() + + const connection = this.webSocketConnections.get(connectionId) + + if (!connection) { + throw new Error(`WebSocket connection ${connectionId} not found`) + } + + // Get or create the callbacks set for this connection + let callbacks = this.messageCallbacks.get(connectionId) + + if (!callbacks) { + callbacks = new Set() + this.messageCallbacks.set(connectionId, callbacks) + } + + // Add the callback + callbacks.add(callback) + } + + /** + * Removes a callback for incoming WebSocket messages + * @param connectionId The identifier of the established connection + * @param callback The function to remove from the callbacks + */ + async offWebSocketMessage( + connectionId: string, + callback: (data: unknown) => void + ): Promise { + await this.ensureInitialized() + + const callbacks = this.messageCallbacks.get(connectionId) + + if (callbacks) { + callbacks.delete(callback) + } + } + + /** + * Closes an established WebSocket connection + * @param connectionId The identifier of the established connection + * @param code Optional close code + * @param reason Optional close reason + */ + async closeWebSocket( + connectionId: string, + code?: number, + reason?: string + ): Promise { + await this.ensureInitialized() + + const connection = this.webSocketConnections.get(connectionId) + + if (!connection) { + throw new Error(`WebSocket connection ${connectionId} not found`) + } + + if (!connection.close) { + throw new Error(`WebSocket connection ${connectionId} does not support closing`) + } + + // Close the connection + await connection.close() + + // Remove from connections map + this.webSocketConnections.delete(connectionId) + + // Remove all callbacks + this.messageCallbacks.delete(connectionId) + } +} + +/** + * WebRTC conduit augmentation for syncing Brainy instances using WebRTC + * + * This conduit is for direct peer-to-peer syncing between browsers. + * It is the recommended approach for browser-to-browser communication. + */ +export class WebRTCConduitAugmentation extends BaseConduitAugmentation implements IWebSocketSupport { + readonly description = 'Conduit augmentation that syncs Brainy instances using WebRTC' + private peerConnections: Map = new Map() + private dataChannels: Map = new Map() + private webSocketConnections: Map = new Map() + private messageCallbacks: Map void>> = new Map() + private signalServer: WebSocketConnection | null = null + + constructor(name: string = 'webrtc-conduit') { + super(name) + } + + getType(): AugmentationType { + return AugmentationType.CONDUIT + } + + async initialize(): Promise { + if (this.isInitialized) { + return + } + + try { + // Check if WebRTC is available + if (typeof RTCPeerConnection === 'undefined') { + throw new Error('WebRTC is not available in this environment') + } + + this.isInitialized = true + } catch (error) { + console.error(`Failed to initialize ${this.name}:`, error) + throw new Error(`Failed to initialize ${this.name}: ${error}`) + } + } + + /** + * Establishes a connection to another Brainy instance using WebRTC + * @param targetSystemId The peer ID or signal server URL + * @param config Configuration options for the connection + */ + async establishConnection( + targetSystemId: string, + config: Record + ): Promise> { + await this.ensureInitialized() + + try { + // For WebRTC, we need to: + // 1. Connect to a signaling server (if not already connected) + // 2. Create a peer connection + // 3. Create a data channel + // 4. Exchange ICE candidates and SDP offers/answers + + // Check if we need to connect to a signaling server + if (!this.signalServer && config.signalServerUrl) { + // Connect to the signaling server + this.signalServer = await this.connectWebSocket(config.signalServerUrl as string) + + // Set up message handling for the signaling server + await this.onWebSocketMessage(this.signalServer.connectionId, async (data) => { + // Handle signaling messages + const message = data as any + + if (message.type === 'ice-candidate' && message.targetPeerId === config.localPeerId) { + // Add ICE candidate to the appropriate peer connection + const peerConnection = this.peerConnections.get(message.sourcePeerId) + if (peerConnection) { + try { + await peerConnection.addIceCandidate(new RTCIceCandidate(message.candidate)) + } catch (error) { + console.error(`Failed to add ICE candidate:`, error) + } + } + } else if (message.type === 'offer' && message.targetPeerId === config.localPeerId) { + // Handle incoming offer + await this.handleOffer(message.sourcePeerId, message.offer, config) + } else if (message.type === 'answer' && message.targetPeerId === config.localPeerId) { + // Handle incoming answer + const peerConnection = this.peerConnections.get(message.sourcePeerId) + if (peerConnection) { + try { + await peerConnection.setRemoteDescription(new RTCSessionDescription(message.answer)) + } catch (error) { + console.error(`Failed to set remote description:`, error) + } + } + } + }) + } + + // Create a peer connection + const peerConnection = new RTCPeerConnection({ + iceServers: (config.iceServers as RTCIceServer[]) || [ + { urls: 'stun:stun.l.google.com:19302' } + ] + }) + + // Generate a connection ID + const connectionId = uuidv4() + + // Store the peer connection + this.peerConnections.set(targetSystemId, peerConnection) + + // Create a data channel + const dataChannel = peerConnection.createDataChannel('brainy-sync', { + ordered: true + }) + + // Set up data channel event handlers + dataChannel.onopen = () => { + console.log(`Data channel to ${targetSystemId} opened`) + } + + dataChannel.onclose = () => { + console.log(`Data channel to ${targetSystemId} closed`) + // Clean up + this.dataChannels.delete(targetSystemId) + this.peerConnections.delete(targetSystemId) + this.webSocketConnections.delete(connectionId) + this.messageCallbacks.delete(connectionId) + } + + dataChannel.onerror = (error) => { + console.error(`Data channel error:`, error) + } + + // Create a message handler wrapper that will call all registered callbacks + const messageHandlerWrapper = (data: unknown) => { + const callbacks = this.messageCallbacks.get(connectionId) + if (callbacks) { + for (const callback of callbacks) { + try { + callback(data) + } catch (error) { + console.error(`Error in WebRTC message callback:`, error) + } + } + } + } + + dataChannel.onmessage = (event) => { + try { + // Parse the message if it's a string + let data = event.data + if (typeof data === 'string') { + try { + data = JSON.parse(data) + } catch { + // If parsing fails, use the raw string + } + } + + // Call the message handler wrapper + messageHandlerWrapper(data) + } catch (error) { + console.error(`Error handling WebRTC message:`, error) + } + } + + // Store the data channel + this.dataChannels.set(targetSystemId, dataChannel) + + // Set up ICE candidate handling + peerConnection.onicecandidate = (event) => { + if (event.candidate && this.signalServer) { + // Send the ICE candidate to the peer via the signaling server + this.sendWebSocketMessage(this.signalServer.connectionId, { + type: 'ice-candidate', + sourcePeerId: config.localPeerId, + targetPeerId: targetSystemId, + candidate: event.candidate + }) + } + } + + // Create a WebSocket-like connection object for the WebRTC connection + const connection: WebSocketConnection = { + connectionId, + url: `webrtc://${targetSystemId}`, + status: 'disconnected', + send: async (data) => { + const dc = this.dataChannels.get(targetSystemId) + if (!dc || dc.readyState !== 'open') { + throw new Error('WebRTC data channel is not open') + } + + // Send the data + if (typeof data === 'string') { + dc.send(data) + } else if (data instanceof Blob) { + dc.send(data) + } else if (data instanceof ArrayBuffer) { + dc.send(new Uint8Array(data)) + } else if (ArrayBuffer.isView(data)) { + dc.send(data) + } else { + // Convert to JSON string + dc.send(JSON.stringify(data)) + } + }, + close: async () => { + const dc = this.dataChannels.get(targetSystemId) + if (dc) { + dc.close() + } + + const pc = this.peerConnections.get(targetSystemId) + if (pc) { + pc.close() + } + + // Clean up + this.dataChannels.delete(targetSystemId) + this.peerConnections.delete(targetSystemId) + this.webSocketConnections.delete(connectionId) + this.messageCallbacks.delete(connectionId) + }, + _messageHandlerWrapper: messageHandlerWrapper + } + + // Store the connection + this.webSocketConnections.set(connectionId, connection) + + // Initialize the callbacks set + this.messageCallbacks.set(connectionId, new Set()) + + // Create and send an offer + const offer = await peerConnection.createOffer() + await peerConnection.setLocalDescription(offer) + + // Send the offer to the peer via the signaling server + if (this.signalServer) { + await this.sendWebSocketMessage(this.signalServer.connectionId, { + type: 'offer', + sourcePeerId: config.localPeerId, + targetPeerId: targetSystemId, + offer + }) + } + + // Return the connection + return { + success: true, + data: connection + } + } catch (error) { + console.error(`Failed to establish WebRTC connection to ${targetSystemId}:`, error) + return { + success: false, + data: null as any, + error: `Failed to establish WebRTC connection: ${error}` + } + } + } + + /** + * Handles an incoming WebRTC offer + * @param peerId The ID of the peer sending the offer + * @param offer The SDP offer + * @param config Configuration options + */ + private async handleOffer( + peerId: string, + offer: RTCSessionDescriptionInit, + config: Record + ): Promise { + try { + // Create a peer connection if it doesn't exist + let peerConnection = this.peerConnections.get(peerId) + + if (!peerConnection) { + peerConnection = new RTCPeerConnection({ + iceServers: (config.iceServers as RTCIceServer[]) || [ + { urls: 'stun:stun.l.google.com:19302' } + ] + }) + + // Store the peer connection + this.peerConnections.set(peerId, peerConnection) + + // Set up ICE candidate handling + peerConnection.onicecandidate = (event) => { + if (event.candidate && this.signalServer) { + // Send the ICE candidate to the peer via the signaling server + this.sendWebSocketMessage(this.signalServer.connectionId, { + type: 'ice-candidate', + sourcePeerId: config.localPeerId, + targetPeerId: peerId, + candidate: event.candidate + }) + } + } + + // Handle data channel creation by the remote peer + peerConnection.ondatachannel = (event) => { + const dataChannel = event.channel + + // Generate a connection ID + const connectionId = uuidv4() + + // Store the data channel + this.dataChannels.set(peerId, dataChannel) + + // Set up data channel event handlers + dataChannel.onopen = () => { + console.log(`Data channel from ${peerId} opened`) + } + + dataChannel.onclose = () => { + console.log(`Data channel from ${peerId} closed`) + // Clean up + this.dataChannels.delete(peerId) + this.peerConnections.delete(peerId) + this.webSocketConnections.delete(connectionId) + this.messageCallbacks.delete(connectionId) + } + + dataChannel.onerror = (error) => { + console.error(`Data channel error:`, error) + } + + // Create a message handler wrapper that will call all registered callbacks + const messageHandlerWrapper = (data: unknown) => { + const callbacks = this.messageCallbacks.get(connectionId) + if (callbacks) { + for (const callback of callbacks) { + try { + callback(data) + } catch (error) { + console.error(`Error in WebRTC message callback:`, error) + } + } + } + } + + dataChannel.onmessage = (event) => { + try { + // Parse the message if it's a string + let data = event.data + if (typeof data === 'string') { + try { + data = JSON.parse(data) + } catch { + // If parsing fails, use the raw string + } + } + + // Call the message handler wrapper + messageHandlerWrapper(data) + } catch (error) { + console.error(`Error handling WebRTC message:`, error) + } + } + + // Create a WebSocket-like connection object for the WebRTC connection + const connection: WebSocketConnection = { + connectionId, + url: `webrtc://${peerId}`, + status: 'disconnected', + send: async (data) => { + if (dataChannel.readyState !== 'open') { + throw new Error('WebRTC data channel is not open') + } + + // Send the data + if (typeof data === 'string') { + dataChannel.send(data) + } else if (data instanceof Blob) { + dataChannel.send(data) + } else if (data instanceof ArrayBuffer) { + dataChannel.send(new Uint8Array(data)) + } else if (ArrayBuffer.isView(data)) { + dataChannel.send(data) + } else { + // Convert to JSON string + dataChannel.send(JSON.stringify(data)) + } + }, + close: async () => { + dataChannel.close() + + const pc = this.peerConnections.get(peerId) + if (pc) { + pc.close() + } + + // Clean up + this.dataChannels.delete(peerId) + this.peerConnections.delete(peerId) + this.webSocketConnections.delete(connectionId) + this.messageCallbacks.delete(connectionId) + }, + _messageHandlerWrapper: messageHandlerWrapper + } + + // Store the connection + this.webSocketConnections.set(connectionId, connection) + + // Initialize the callbacks set + this.messageCallbacks.set(connectionId, new Set()) + } + } + + // Set the remote description (the offer) + await peerConnection.setRemoteDescription(new RTCSessionDescription(offer)) + + // Create an answer + const answer = await peerConnection.createAnswer() + await peerConnection.setLocalDescription(answer) + + // Send the answer to the peer via the signaling server + if (this.signalServer) { + await this.sendWebSocketMessage(this.signalServer.connectionId, { + type: 'answer', + sourcePeerId: config.localPeerId, + targetPeerId: peerId, + answer + }) + } + } catch (error) { + console.error(`Failed to handle WebRTC offer:`, error) + throw new Error(`Failed to handle WebRTC offer: ${error}`) + } + } + + /** + * Reads data from a connected Brainy instance + * @param query Query parameters for reading data + * @param options Additional options + */ + async readData( + query: Record, + options?: Record + ): Promise> { + await this.ensureInitialized() + + try { + const connectionId = query.connectionId as string + + if (!connectionId) { + throw new Error('connectionId is required for reading data') + } + + const connection = this.webSocketConnections.get(connectionId) + + if (!connection) { + throw new Error(`Connection ${connectionId} not found`) + } + + // Create a request message + const requestMessage = { + type: 'read', + query: query.query || {}, + requestId: uuidv4(), + options + } + + // Send the request + await this.sendWebSocketMessage(connectionId, requestMessage) + + // Return a promise that will be resolved when the response is received + return new Promise((resolve) => { + const responseHandler = (data: unknown) => { + // Check if this is the response to our request + const response = data as any + if (response && response.type === 'readResponse' && response.requestId === requestMessage.requestId) { + // Remove the handler + this.offWebSocketMessage(connectionId, responseHandler) + + // Resolve with the response data + resolve({ + success: response.success, + data: response.data, + error: response.error + }) + } + } + + // Register the response handler + this.onWebSocketMessage(connectionId, responseHandler) + + // Set a timeout to prevent hanging + setTimeout(() => { + this.offWebSocketMessage(connectionId, responseHandler) + resolve({ + success: false, + data: null, + error: 'Timeout waiting for read response' + }) + }, 30000) // 30 second timeout + }) + } catch (error) { + console.error(`Failed to read data:`, error) + return { + success: false, + data: null, + error: `Failed to read data: ${error}` + } + } + } + + /** + * Writes data to a connected Brainy instance + * @param data The data to write + * @param options Additional options + */ + async writeData( + data: Record, + options?: Record + ): Promise> { + await this.ensureInitialized() + + try { + const connectionId = data.connectionId as string + + if (!connectionId) { + throw new Error('connectionId is required for writing data') + } + + const connection = this.webSocketConnections.get(connectionId) + + if (!connection) { + throw new Error(`Connection ${connectionId} not found`) + } + + // Create a write message + const writeMessage = { + type: 'write', + data: data.data || {}, + requestId: uuidv4(), + options + } + + // Send the write message + await this.sendWebSocketMessage(connectionId, writeMessage) + + // Return a promise that will be resolved when the response is received + return new Promise((resolve) => { + const responseHandler = (data: unknown) => { + // Check if this is the response to our request + const response = data as any + if (response && response.type === 'writeResponse' && response.requestId === writeMessage.requestId) { + // Remove the handler + this.offWebSocketMessage(connectionId, responseHandler) + + // Resolve with the response data + resolve({ + success: response.success, + data: response.data, + error: response.error + }) + } + } + + // Register the response handler + this.onWebSocketMessage(connectionId, responseHandler) + + // Set a timeout to prevent hanging + setTimeout(() => { + this.offWebSocketMessage(connectionId, responseHandler) + resolve({ + success: false, + data: null, + error: 'Timeout waiting for write response' + }) + }, 30000) // 30 second timeout + }) + } catch (error) { + console.error(`Failed to write data:`, error) + return { + success: false, + data: null, + error: `Failed to write data: ${error}` + } + } + } + + /** + * Monitors a data stream from a connected Brainy instance + * @param streamId The ID of the stream to monitor (usually a connection ID) + * @param callback Function to call when new data is received + */ + async monitorStream( + streamId: string, + callback: (data: unknown) => void + ): Promise { + await this.ensureInitialized() + + try { + const connection = this.webSocketConnections.get(streamId) + + if (!connection) { + throw new Error(`Connection ${streamId} not found`) + } + + // Register the callback for all messages on this connection + await this.onWebSocketMessage(streamId, callback) + + } catch (error) { + console.error(`Failed to monitor stream ${streamId}:`, error) + throw new Error(`Failed to monitor stream: ${error}`) + } + } + + /** + * Establishes a WebSocket connection (used for signaling in WebRTC) + * @param url The WebSocket server URL to connect to + * @param protocols Optional subprotocols + */ + async connectWebSocket( + url: string, + protocols?: string | string[] + ): Promise { + await this.ensureInitialized() + + return new Promise((resolve, reject) => { + try { + // Check if WebSocket is available + if (typeof WebSocket === 'undefined') { + throw new Error('WebSocket is not available in this environment') + } + + // Create a new WebSocket connection + const ws = new WebSocket(url, protocols) + const connectionId = uuidv4() + + // Create a connection object + const connection: WebSocketConnection = { + connectionId, + url, + status: 'disconnected', + send: async (data) => { + if (ws.readyState !== WebSocket.OPEN) { + throw new Error('WebSocket is not open') + } + ws.send(data) + }, + close: async () => { + ws.close() + } + } + + // Set up event handlers + ws.onopen = () => { + connection.status = 'connected' + resolve(connection) + } + + ws.onerror = (error) => { + connection.status = 'error' + console.error(`WebSocket error for ${url}:`, error) + if (ws.readyState !== WebSocket.OPEN) { + reject(new Error(`WebSocket connection failed: ${error}`)) + } + } + + ws.onclose = () => { + connection.status = 'disconnected' + // Remove from connections map + this.webSocketConnections.delete(connectionId) + // Remove all callbacks + this.messageCallbacks.delete(connectionId) + } + + // Create a message handler wrapper that will call all registered callbacks + const messageHandlerWrapper = (data: unknown) => { + const callbacks = this.messageCallbacks.get(connectionId) + if (callbacks) { + for (const callback of callbacks) { + try { + callback(data) + } catch (error) { + console.error(`Error in WebSocket message callback:`, error) + } + } + } + } + + // Store the message handler wrapper + connection._messageHandlerWrapper = messageHandlerWrapper + + // Set up the message handler + ws.onmessage = (event) => { + try { + // Parse the message if it's a string + let data = event.data + if (typeof data === 'string') { + try { + data = JSON.parse(data) + } catch { + // If parsing fails, use the raw string + } + } + + // Call the message handler wrapper + messageHandlerWrapper(data) + } catch (error) { + console.error(`Error handling WebSocket message:`, error) + } + } + + // Store the stream message handler + connection._streamMessageHandler = (event) => ws.onmessage && ws.onmessage(event as any) + + // Store the connection + this.webSocketConnections.set(connectionId, connection) + + // Initialize the callbacks set + this.messageCallbacks.set(connectionId, new Set()) + + } catch (error) { + reject(error) + } + }) + } + + /** + * Sends data through an established WebSocket or WebRTC connection + * @param connectionId The identifier of the established connection + * @param data The data to send (will be serialized if not a string) + */ + async sendWebSocketMessage( + connectionId: string, + data: unknown + ): Promise { + await this.ensureInitialized() + + const connection = this.webSocketConnections.get(connectionId) + + if (!connection) { + throw new Error(`Connection ${connectionId} not found`) + } + + if (!connection.send) { + throw new Error(`Connection ${connectionId} does not support sending messages`) + } + + // Serialize the data if it's not already a string or binary + let serializedData: string | ArrayBufferLike | Blob | ArrayBufferView + + if (typeof data === 'string' || + data instanceof ArrayBuffer || + data instanceof Blob || + ArrayBuffer.isView(data)) { + serializedData = data as any + } else { + // Convert to JSON string + serializedData = JSON.stringify(data) + } + + // Send the data + await connection.send(serializedData) + } + + /** + * Registers a callback for incoming WebSocket or WebRTC messages + * @param connectionId The identifier of the established connection + * @param callback The function to call when a message is received + */ + async onWebSocketMessage( + connectionId: string, + callback: (data: unknown) => void + ): Promise { + await this.ensureInitialized() + + const connection = this.webSocketConnections.get(connectionId) + + if (!connection) { + throw new Error(`Connection ${connectionId} not found`) + } + + // Get or create the callbacks set for this connection + let callbacks = this.messageCallbacks.get(connectionId) + + if (!callbacks) { + callbacks = new Set() + this.messageCallbacks.set(connectionId, callbacks) + } + + // Add the callback + callbacks.add(callback) + } + + /** + * Removes a callback for incoming WebSocket or WebRTC messages + * @param connectionId The identifier of the established connection + * @param callback The function to remove from the callbacks + */ + async offWebSocketMessage( + connectionId: string, + callback: (data: unknown) => void + ): Promise { + await this.ensureInitialized() + + const callbacks = this.messageCallbacks.get(connectionId) + + if (callbacks) { + callbacks.delete(callback) + } + } + + /** + * Closes an established WebSocket or WebRTC connection + * @param connectionId The identifier of the established connection + * @param code Optional close code + * @param reason Optional close reason + */ + async closeWebSocket( + connectionId: string, + code?: number, + reason?: string + ): Promise { + await this.ensureInitialized() + + const connection = this.webSocketConnections.get(connectionId) + + if (!connection) { + throw new Error(`Connection ${connectionId} not found`) + } + + if (!connection.close) { + throw new Error(`Connection ${connectionId} does not support closing`) + } + + // Close the connection + await connection.close() + + // Remove from connections map + this.webSocketConnections.delete(connectionId) + + // Remove all callbacks + this.messageCallbacks.delete(connectionId) + } +} + +/** + * Factory function to create the appropriate conduit augmentation based on the type + */ +export async function createConduitAugmentation( + type: 'websocket' | 'webrtc', + name?: string, + options: Record = {} +): Promise { + switch (type) { + case 'websocket': + const wsAugmentation = new WebSocketConduitAugmentation(name || 'websocket-conduit') + await wsAugmentation.initialize() + return wsAugmentation + case 'webrtc': + const webrtcAugmentation = new WebRTCConduitAugmentation(name || 'webrtc-conduit') + await webrtcAugmentation.initialize() + return webrtcAugmentation + default: + throw new Error(`Unknown conduit augmentation type: ${type}`) + } +} diff --git a/src/types/augmentations.ts b/src/types/augmentations.ts index 45032200..181b53f7 100644 --- a/src/types/augmentations.ts +++ b/src/types/augmentations.ts @@ -140,7 +140,7 @@ export namespace BrainyAugmentations { establishConnection( targetSystemId: string, config: Record - ): AugmentationResponse + ): Promise> /** * Reads structured data directly from Brainy's knowledge graph. @@ -150,7 +150,7 @@ export namespace BrainyAugmentations { readData( query: Record, options?: Record - ): AugmentationResponse + ): Promise> /** * Writes or updates structured data directly into Brainy's knowledge graph. @@ -160,7 +160,7 @@ export namespace BrainyAugmentations { writeData( data: Record, options?: Record - ): AugmentationResponse + ): Promise> /** * Monitors a specific data stream or event within Brainy for external systems.