fix(indexer): implement real GraphQL network I/O for query and subscribe

src/indexer.ts

@@ -1,11 +1,13 @@
 export interface GraphQLQueryOptions {
   query: string;
   variables?: Record<string, unknown>;
+  headers?: Record<string, string>;
 }
 
 export interface GraphQLSubscriptionOptions {
   query: string;
   variables?: Record<string, unknown>;
+  headers?: Record<string, string>;
   onData: (data: unknown) => void;
   onError?: (error: Error) => void;
 }
@@ -18,6 +20,7 @@ export class GraphQLIndexer {
   private endpoint: string;
   private activeSubscriptions: Set<IndexerSubscription> = new Set();
   private isDestroyed = false;
+  private subCounter = 0;
 
   constructor(endpoint: string) {
     if (!endpoint || typeof endpoint !== 'string' || endpoint.trim().length === 0) {
@@ -37,22 +40,36 @@ export class GraphQLIndexer {
       throw new Error('GraphQL query string cannot be null or empty');
     }
 
-    // Boundary check on variables
     const variables = options.variables ?? {};
     if (typeof variables !== 'object' || variables === null) {
       throw new Error('GraphQL query variables must be an object');
     }
 
-    const payload = { query: options.query, variables };
+    const headers: Record<string, string> = {
+      'Content-Type': 'application/json',
+      'Accept': 'application/json',
+      ...options.headers,
+    };
 
-    // Simulate async RPC/GraphQL fetch
-    return Promise.resolve({
-      data: {
-        indexed: true,
-        endpoint: this.endpoint,
-        payload,
-      },
+    const fetchFn = typeof fetch !== 'undefined' ? fetch : (globalThis as unknown as { fetch?: typeof fetch }).fetch;
+    if (typeof fetchFn !== 'function') {
+      throw new Error('Fetch API is not available in the current environment');
+    }
+
+    const response = await fetchFn(this.endpoint, {
+      method: 'POST',
+      headers,
+      body: JSON.stringify({
+        query: options.query,
+        variables,
+      }),
     });
+
+    if (!response.ok) {
+      throw new Error(`GraphQL query failed with status ${response.status}: ${response.statusText}`);
+    }
+
+    return await response.json();
   }
 
   subscribe(options: GraphQLSubscriptionOptions): IndexerSubscription {
@@ -69,38 +86,171 @@ export class GraphQLIndexer {
       throw new Error('GraphQL subscription onData callback must be a function');
     }
 
+    const variables = options.variables ?? {};
+    if (typeof variables !== 'object' || variables === null) {
+      throw new Error('GraphQL query variables must be an object');
+    }
+
     let unsubscribed = false;
-    let timer: NodeJS.Timeout | null = null;
+    const subId = `sub_${++this.subCounter}_${Date.now()}`;
+
+    let ws: WebSocket | null = null;
+    let abortController: AbortController | null = null;
 
     const subscription: IndexerSubscription = {
       unsubscribe: () => {
         if (unsubscribed) return;
         unsubscribed = true;
-        if (timer) {
-          clearInterval(timer);
-          timer = null;
+
+        if (ws) {
+          try {
+            if (ws.readyState === 1 /* OPEN */) {
+              ws.send(JSON.stringify({ id: subId, type: 'complete' }));
+            }
+            ws.close();
+          } catch {
+            // Ignore socket closure errors
+          }
+          ws = null;
+        }
+
+        if (abortController) {
+          try {
+            abortController.abort();
+          } catch {
+            // Ignore abort error
+          }
+          abortController = null;
         }
+
         this.activeSubscriptions.delete(subscription);
       },
     };
 
     this.activeSubscriptions.add(subscription);
 
-    // Periodic subscription event callback with lifecycle boundary check
-    timer = setInterval(() => {
-      if (unsubscribed || this.isDestroyed) {
-        subscription.unsubscribe();
-        return;
-      }
-
+    const WebSocketCtor = this.getWebSocketCtor();
+    if (WebSocketCtor) {
       try {
-        options.onData({ timestamp: Date.now(), status: 'STREAM_UPDATED' });
-      } catch (err) {
-        if (options.onError && typeof options.onError === 'function') {
-          options.onError(err instanceof Error ? err : new Error(String(err)));
+        const wsUrl = this.getWsUrl(this.endpoint);
+        let socket: WebSocket;
+        try {
+          socket = new WebSocketCtor(wsUrl, 'graphql-transport-ws');
+        } catch {
+          socket = new WebSocketCtor(wsUrl);
         }
+        ws = socket;
+
+        socket.onopen = () => {
+          if (unsubscribed || this.isDestroyed) {
+            subscription.unsubscribe();
+            return;
+          }
+          try {
+            socket.send(JSON.stringify({ type: 'connection_init' }));
+            socket.send(
+              JSON.stringify({
+                id: subId,
+                type: 'subscribe',
+                payload: {
+                  query: options.query,
+                  variables,
+                },
+              })
+            );
+          } catch (err) {
+            this.handleError(options.onError, err);
+          }
+        };
+
+        socket.onmessage = (event: MessageEvent) => {
+          if (unsubscribed || this.isDestroyed) return;
+          try {
+            const data = typeof event.data === 'string' ? JSON.parse(event.data) : event.data;
+            if (!data || typeof data !== 'object') return;
+
+            if (data.type === 'next' || data.type === 'data') {
+              const payload = data.payload ?? data.data;
+              options.onData(payload);
+            } else if (data.type === 'error') {
+              const errPayload = data.payload ?? data.errors;
+              const errMsg = typeof errPayload === 'string' ? errPayload : JSON.stringify(errPayload);
+              this.handleError(options.onError, new Error(errMsg));
+            }
+          } catch (err) {
+            this.handleError(options.onError, err);
+          }
+        };
+
+        socket.onerror = (_event: Event) => {
+          if (unsubscribed || this.isDestroyed) return;
+          this.handleError(options.onError, new Error(`GraphQL subscription WebSocket error on ${this.endpoint}`));
+        };
+
+        socket.onclose = () => {
+          if (!unsubscribed && !this.isDestroyed) {
+            subscription.unsubscribe();
+          }
+        };
+      } catch (err) {
+        this.handleError(options.onError, err);
       }
-    }, 1000);
+    } else {
+      const fetchFn = typeof fetch !== 'undefined' ? fetch : (globalThis as unknown as { fetch?: typeof fetch }).fetch;
+      if (typeof fetchFn === 'function' && typeof AbortController !== 'undefined') {
+        abortController = new AbortController();
+        const headers: Record<string, string> = {
+          'Content-Type': 'application/json',
+          'Accept': 'text/event-stream, application/json',
+          ...options.headers,
+        };
+
+        fetchFn(this.endpoint, {
+          method: 'POST',
+          headers,
+          body: JSON.stringify({ query: options.query, variables }),
+          signal: abortController.signal,
+        })
+          .then(async (response: Response) => {
+            if (unsubscribed || this.isDestroyed) return;
+            if (!response.ok) {
+              this.handleError(options.onError, new Error(`GraphQL subscription HTTP error: ${response.status}`));
+              return;
+            }
+            if (response.body && typeof response.body.getReader === 'function') {
+              const reader = response.body.getReader();
+              const decoder = new TextDecoder();
+              let buffer = '';
+              while (!unsubscribed && !this.isDestroyed) {
+                const { value, done } = await reader.read();
+                if (done) break;
+                buffer += decoder.decode(value, { stream: true });
+                const lines = buffer.split('\n');
+                buffer = lines.pop() ?? '';
+                for (const line of lines) {
+                  const trimmed = line.trim();
+                  if (trimmed.startsWith('data:')) {
+                    const dataStr = trimmed.slice(5).trim();
+                    if (dataStr === '[DONE]') break;
+                    try {
+                      const parsed = JSON.parse(dataStr);
+                      options.onData(parsed.data ?? parsed);
+                    } catch {
+                      // Ignore malformed line
+                    }
+                  }
+                }
+              }
+            }
+          })
+          .catch((err: unknown) => {
+            const isAbort = err instanceof Error && err.name === 'AbortError';
+            if (!unsubscribed && !this.isDestroyed && !isAbort) {
+              this.handleError(options.onError, err);
+            }
+          });
+      }
+    }
 
     return subscription;
   }
@@ -116,4 +266,34 @@ export class GraphQLIndexer {
     }
     this.activeSubscriptions.clear();
   }
+
+  private getWebSocketCtor(): (new (url: string | URL, protocols?: string | string[]) => WebSocket) | null {
+    if (typeof globalThis !== 'undefined' && 'WebSocket' in globalThis && typeof (globalThis as unknown as { WebSocket?: typeof WebSocket }).WebSocket === 'function') {
+      return (globalThis as unknown as { WebSocket?: typeof WebSocket }).WebSocket ?? null;
+    }
+    if (typeof WebSocket !== 'undefined' && typeof WebSocket === 'function') {
+      return WebSocket;
+    }
+    return null;
+  }
+
+  private getWsUrl(endpoint: string): string {
+    if (endpoint.startsWith('https://')) {
+      return 'wss://' + endpoint.slice(8);
+    }
+    if (endpoint.startsWith('http://')) {
+      return 'ws://' + endpoint.slice(7);
+    }
+    return endpoint;
+  }
+
+  private handleError(onError: ((err: Error) => void) | undefined, err: unknown): void {
+    if (onError && typeof onError === 'function') {
+      try {
+        onError(err instanceof Error ? err : new Error(String(err)));
+      } catch {
+        // Prevent uncaught errors from user error handlers
+      }
+    }
+  }
 }

src/tests/graphql-indexer-memory-leak.test.ts

@@ -1,23 +1,76 @@
-import { describe, it, expect } from 'vitest';
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
 import { GraphQLIndexer } from '../indexer.js';
 
-describe('GraphQLIndexer Memory Leak & Boundary Check Regression Tests', () => {
+describe('GraphQLIndexer Memory Leak & Real Network I/O Tests', () => {
+  const endpoint = 'https://indexer.streamfi.io/graphql';
+
+  beforeEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
   it('throws boundary error when endpoint is empty or null', () => {
     expect(() => new GraphQLIndexer('')).toThrow('GraphQLIndexer endpoint must be a non-empty string');
     expect(() => new GraphQLIndexer(null as any)).toThrow('GraphQLIndexer endpoint must be a non-empty string');
   });
 
   it('throws boundary error when query options are null or missing query string', async () => {
-    const indexer = new GraphQLIndexer('https://indexer.streamfi.io/graphql');
+    const indexer = new GraphQLIndexer(endpoint);
 
     await expect(indexer.query(null as any)).rejects.toThrow('GraphQLQueryOptions cannot be null or undefined');
     await expect(indexer.query({ query: '' })).rejects.toThrow('GraphQL query string cannot be null or empty');
 
     indexer.cleanup();
   });
 
+  it('performs a real HTTP POST request in query() and returns response data', async () => {
+    const mockResponseData = { data: { streams: [{ id: '1', sender: 'GAAA' }] } };
+    const fetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValue({
+      ok: true,
+      status: 200,
+      json: async () => mockResponseData,
+    } as Response);
+
+    const indexer = new GraphQLIndexer(endpoint);
+    const queryStr = 'query { streams { id sender } }';
+    const variables = { limit: 10 };
+
+    const result = await indexer.query({ query: queryStr, variables });
+
+    expect(fetchSpy).toHaveBeenCalledWith(endpoint, {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+        'Accept': 'application/json',
+      },
+      body: JSON.stringify({ query: queryStr, variables }),
+    });
+    expect(result).toEqual(mockResponseData);
+
+    indexer.cleanup();
+  });
+
+  it('handles HTTP error in query() gracefully', async () => {
+    vi.spyOn(globalThis, 'fetch').mockResolvedValue({
+      ok: false,
+      status: 500,
+      statusText: 'Internal Server Error',
+    } as Response);
+
+    const indexer = new GraphQLIndexer(endpoint);
+
+    await expect(indexer.query({ query: 'query { streams { id } }' })).rejects.toThrow(
+      'GraphQL query failed with status 500: Internal Server Error'
+    );
+
+    indexer.cleanup();
+  });
+
   it('subscribes and properly cleans up active subscriptions on unsubscribe()', async () => {
-    const indexer = new GraphQLIndexer('https://indexer.streamfi.io/graphql');
+    const indexer = new GraphQLIndexer(endpoint);
 
     const sub = indexer.subscribe({
       query: 'subscription { streamUpdated { id } }',
@@ -34,7 +87,7 @@ describe('GraphQLIndexer Memory Leak & Boundary Check Regression Tests', () => {
   });
 
   it('cleans up all active subscriptions on indexer.cleanup() without memory leak', () => {
-    const indexer = new GraphQLIndexer('https://indexer.streamfi.io/graphql');
+    const indexer = new GraphQLIndexer(endpoint);
 
     indexer.subscribe({ query: 'subscription { sub1 }', onData: () => {} });
     indexer.subscribe({ query: 'subscription { sub2 }', onData: () => {} });

src/tests/walletconnect-edge-cases.test.ts

@@ -71,11 +71,11 @@ describe('WalletConnectAdapter edge cases', () => {
       // Simulates: new WalletConnectAdapter({ chainId: someReactState })
       // where someReactState is undefined before an async load completes.
       // The ?? fallback converts undefined → 'stellar:pubnet', so no crash.
-      expect(() => new WalletConnectAdapter({ chainId: undefined } as any)).not.toThrow();
+      expect(() => new WalletConnectAdapter({ chainId: undefined })).not.toThrow();
     });
 
     it('should default to stellar:pubnet when chainId is undefined', () => {
-      const adapter = new WalletConnectAdapter({ chainId: undefined } as any);
+      const adapter = new WalletConnectAdapter({ chainId: undefined });
       expect(adapter).toBeInstanceOf(WalletConnectAdapter);
     });