From f905bc23c6c5ac57f692b53870c9224221fb3612 Mon Sep 17 00:00:00 2001 From: morelucks Date: Wed, 29 Jul 2026 11:25:09 +0100 Subject: [PATCH 1/2] fix(indexer): implement real GraphQL network I/O for query and subscribe Fixes #177 --- src/indexer.ts | 229 ++++++++++++++++-- src/tests/graphql-indexer-memory-leak.test.ts | 63 ++++- src/tests/walletconnect-edge-cases.test.ts | 4 +- 3 files changed, 264 insertions(+), 32 deletions(-) diff --git a/src/indexer.ts b/src/indexer.ts index 4f238fb1..901115e0 100644 --- a/src/indexer.ts +++ b/src/indexer.ts @@ -1,11 +1,13 @@ export interface GraphQLQueryOptions { query: string; variables?: Record; + headers?: Record; } export interface GraphQLSubscriptionOptions { query: string; variables?: Record; + headers?: Record; onData: (data: unknown) => void; onError?: (error: Error) => void; } @@ -18,6 +20,7 @@ export class GraphQLIndexer { private endpoint: string; private activeSubscriptions: Set = 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 = { + '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 any).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,170 @@ 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 any).fetch; + if (typeof fetchFn === 'function' && typeof AbortController !== 'undefined') { + abortController = new AbortController(); + const headers: Record = { + '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: any) => { + 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: any) => { + if (!unsubscribed && !this.isDestroyed && err?.name !== 'AbortError') { + this.handleError(options.onError, err); + } + }); + } + } return subscription; } @@ -116,4 +265,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 any).WebSocket === 'function') { + return (globalThis as any).WebSocket; + } + 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 + } + } + } } diff --git a/src/tests/graphql-indexer-memory-leak.test.ts b/src/tests/graphql-indexer-memory-leak.test.ts index 9c293600..80cd5f06 100644 --- a/src/tests/graphql-indexer-memory-leak.test.ts +++ b/src/tests/graphql-indexer-memory-leak.test.ts @@ -1,14 +1,24 @@ -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'); @@ -16,8 +26,51 @@ describe('GraphQLIndexer Memory Leak & Boundary Check Regression Tests', () => { 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: () => {} }); diff --git a/src/tests/walletconnect-edge-cases.test.ts b/src/tests/walletconnect-edge-cases.test.ts index 9d1cd027..94c0f1bf 100644 --- a/src/tests/walletconnect-edge-cases.test.ts +++ b/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); }); From 0a0a6b45180c76ab580c5f7de727bbd5b742ce25 Mon Sep 17 00:00:00 2001 From: bade22brazy Date: Thu, 30 Jul 2026 00:06:46 +0100 Subject: [PATCH 2/2] fix: satisfy lint/typecheck in GraphQL indexer network I/O Replace any-typed globalThis/fetch/WebSocket access with narrow inline casts, type the fetch subscription's catch handler as unknown with a proper instanceof narrow for AbortError detection, and null-coalesce getWebSocketCtor's globalThis lookup to match its declared return type. --- src/indexer.ts | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/src/indexer.ts b/src/indexer.ts index 901115e0..ebcc8a50 100644 --- a/src/indexer.ts +++ b/src/indexer.ts @@ -51,7 +51,7 @@ export class GraphQLIndexer { ...options.headers, }; - const fetchFn = typeof fetch !== 'undefined' ? fetch : (globalThis as any).fetch; + 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'); } @@ -182,7 +182,7 @@ export class GraphQLIndexer { } }; - socket.onerror = (event: Event) => { + socket.onerror = (_event: Event) => { if (unsubscribed || this.isDestroyed) return; this.handleError(options.onError, new Error(`GraphQL subscription WebSocket error on ${this.endpoint}`)); }; @@ -196,7 +196,7 @@ export class GraphQLIndexer { this.handleError(options.onError, err); } } else { - const fetchFn = typeof fetch !== 'undefined' ? fetch : (globalThis as any).fetch; + 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 = { @@ -211,7 +211,7 @@ export class GraphQLIndexer { body: JSON.stringify({ query: options.query, variables }), signal: abortController.signal, }) - .then(async (response: any) => { + .then(async (response: Response) => { if (unsubscribed || this.isDestroyed) return; if (!response.ok) { this.handleError(options.onError, new Error(`GraphQL subscription HTTP error: ${response.status}`)); @@ -243,8 +243,9 @@ export class GraphQLIndexer { } } }) - .catch((err: any) => { - if (!unsubscribed && !this.isDestroyed && err?.name !== 'AbortError') { + .catch((err: unknown) => { + const isAbort = err instanceof Error && err.name === 'AbortError'; + if (!unsubscribed && !this.isDestroyed && !isAbort) { this.handleError(options.onError, err); } }); @@ -267,8 +268,8 @@ export class GraphQLIndexer { } private getWebSocketCtor(): (new (url: string | URL, protocols?: string | string[]) => WebSocket) | null { - if (typeof globalThis !== 'undefined' && 'WebSocket' in globalThis && typeof (globalThis as any).WebSocket === 'function') { - return (globalThis as any).WebSocket; + 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;