diff --git a/gitnexus-web/e2e/server-connect.spec.ts b/gitnexus-web/e2e/server-connect.spec.ts index eb241da10..0705b3a27 100644 --- a/gitnexus-web/e2e/server-connect.spec.ts +++ b/gitnexus-web/e2e/server-connect.spec.ts @@ -1,4 +1,4 @@ -import { test, expect, type TestInfo } from '@playwright/test'; +import { test, expect } from '@playwright/test'; /** * E2E tests for the GitNexus web UI — exploring view features. @@ -58,36 +58,41 @@ test.beforeAll(async () => { * For these tests we require at least one indexed repo, so pick the first * landing card when present and then wait for the exploring view. */ -async function waitForGraphLoaded(page: import('@playwright/test').Page, testInfo: TestInfo) { +async function waitForGraphLoaded(page: import('@playwright/test').Page) { await page.goto('/'); - const landingCard = page.locator('[data-testid="landing-repo-card"]').first(); + const landingCards = page.locator('[data-testid="landing-repo-card"]'); + const preferredLandingCard = landingCards + .filter({ hasText: /GitNexus|local-integration/ }) + .first(); try { - await landingCard.waitFor({ state: 'visible', timeout: 15_000 }); + await landingCards.first().waitFor({ state: 'visible', timeout: 15_000 }); + const landingCard = + (await preferredLandingCard.count()) > 0 ? preferredLandingCard : landingCards.first(); await landingCard.click(); } catch { // Landing screen may not appear (e.g. ?server auto-connect) } - await expect(page.locator('[data-testid="status-ready"]')).toBeVisible({ timeout: 30_000 }); - await expect(page.getByText(/\d+ nodes/).first()).toBeVisible(); - await page.screenshot({ path: testInfo.outputPath('graph-loaded.png') }); + const statusBar = page.getByRole('contentinfo'); + await expect(statusBar.getByText('Ready', { exact: true })).toBeVisible({ timeout: 45_000 }); + await expect(statusBar).toContainText(/nodes/, { + timeout: 20_000, + }); } test.describe('Server Connection & Graph Loading', () => { - test('selects a repo from landing and loads graph', async ({ page }, testInfo) => { - await waitForGraphLoaded(page, testInfo); - await page.screenshot({ path: testInfo.outputPath('graph-loaded-full.png'), fullPage: true }); + test('selects a repo from landing and loads graph', async ({ page }) => { + await waitForGraphLoaded(page); }); }); test.describe('Nexus AI', () => { - test('panel opens and agent initializes without error', async ({ page }, testInfo) => { - await waitForGraphLoaded(page, testInfo); + test('panel opens and agent initializes without error', async ({ page }) => { + await waitForGraphLoaded(page); await page.getByRole('button', { name: 'Nexus AI' }).click(); await expect(page.getByText('Ask me anything')).toBeVisible({ timeout: 15_000 }); - await page.screenshot({ path: testInfo.outputPath('nexus-ai-panel.png'), fullPage: true }); const errorBanner = page.getByText('Database not ready'); expect(await errorBanner.isVisible().catch(() => false)).toBe(false); @@ -95,8 +100,8 @@ test.describe('Nexus AI', () => { }); test.describe('Processes Panel', () => { - test('shows process list and View button works', async ({ page }, testInfo) => { - await waitForGraphLoaded(page, testInfo); + test('shows process list and View button works', async ({ page }) => { + await waitForGraphLoaded(page); await page.getByRole('button', { name: 'Nexus AI' }).click(); await page.getByText('Processes').click(); @@ -104,7 +109,6 @@ test.describe('Processes Panel', () => { await expect(page.locator('[data-testid="process-list-loaded"]')).toBeVisible({ timeout: 15_000, }); - await page.screenshot({ path: testInfo.outputPath('processes-panel.png'), fullPage: true }); const processRow = page.locator('[data-testid="process-row"]').first(); await expect(processRow).toBeVisible({ timeout: 10_000 }); @@ -114,14 +118,10 @@ test.describe('Processes Panel', () => { await viewBtn.waitFor({ state: 'visible', timeout: 5_000 }); await viewBtn.click(); await expect(page.locator('[data-testid="process-modal"]')).toBeVisible({ timeout: 5_000 }); - await page.screenshot({ - path: testInfo.outputPath('process-view-clicked.png'), - fullPage: true, - }); }); - test('lightbulb highlights nodes in graph', async ({ page }, testInfo) => { - await waitForGraphLoaded(page, testInfo); + test('lightbulb highlights nodes in graph', async ({ page }) => { + await waitForGraphLoaded(page); await page.getByRole('button', { name: 'Nexus AI' }).click(); await page.getByText('Processes').click(); @@ -137,13 +137,12 @@ test.describe('Processes Panel', () => { await lightbulb.waitFor({ state: 'visible', timeout: 5_000 }); await lightbulb.click(); await expect(processRow).toHaveClass(/bg-amber-950/, { timeout: 5_000 }); - await page.screenshot({ path: testInfo.outputPath('after-highlight.png'), fullPage: true }); }); }); test.describe('Turn Off All Highlights', () => { - test('selecting a node dims others, button clears it', async ({ page }, testInfo) => { - await waitForGraphLoaded(page, testInfo); + test('selecting a node dims others, button clears it', async ({ page }) => { + await waitForGraphLoaded(page); await expect(page.locator('canvas').first()).toBeVisible({ timeout: 10_000 }); @@ -160,6 +159,5 @@ test.describe('Turn Off All Highlights', () => { await expect(highlightToggle).toHaveAttribute('title', 'Turn on AI highlights', { timeout: 5_000, }); - await page.screenshot({ path: testInfo.outputPath('highlights-cleared.png'), fullPage: true }); }); }); diff --git a/gitnexus-web/src/components/StatusBar.tsx b/gitnexus-web/src/components/StatusBar.tsx index a618c3c1c..7468072fa 100644 --- a/gitnexus-web/src/components/StatusBar.tsx +++ b/gitnexus-web/src/components/StatusBar.tsx @@ -64,7 +64,7 @@ export const StatusBar = () => { {/* Right - Stats */} -
+
{graph && ( <> {nodeCount} nodes diff --git a/gitnexus-web/src/services/backend-client.ts b/gitnexus-web/src/services/backend-client.ts index 256219eb6..2c04b66dd 100644 --- a/gitnexus-web/src/services/backend-client.ts +++ b/gitnexus-web/src/services/backend-client.ts @@ -404,13 +404,18 @@ export const fetchGraph = async ( onProgress?: (downloaded: number, total: number | null) => void; }, ): Promise<{ nodes: GraphNode[]; relationships: GraphRelationship[] }> => { - const params = [repoParam(repo), opts?.includeContent ? 'includeContent=true' : ''] + const params = [repoParam(repo), opts?.includeContent ? 'includeContent=true' : '', 'stream=true'] .filter(Boolean) .join('&'); const url = `${_backendUrl}/api/graph${params ? `?${params}` : ''}`; const response = await fetchWithTimeout(url, { signal: opts?.signal }, 60_000); await assertOk(response); + const contentType = response.headers.get('Content-Type') || ''; + if (contentType.includes('application/x-ndjson')) { + return parseNdjsonGraphResponse(response, opts?.onProgress); + } + if (!opts?.onProgress || !response.body) { return response.json() as Promise<{ nodes: GraphNode[]; relationships: GraphRelationship[] }>; } @@ -439,6 +444,66 @@ export const fetchGraph = async ( return JSON.parse(new TextDecoder().decode(combined)); }; +const parseNdjsonGraphResponse = async ( + response: Response, + onProgress?: (downloaded: number, total: number | null) => void, +): Promise<{ nodes: GraphNode[]; relationships: GraphRelationship[] }> => { + if (!response.body) { + throw new BackendError('No response body', response.status, 'server'); + } + + const contentLength = response.headers.get('Content-Length'); + const total = contentLength ? parseInt(contentLength, 10) : null; + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + const nodes: GraphNode[] = []; + const relationships: GraphRelationship[] = []; + let buffer = ''; + let downloaded = 0; + + const parseLine = (line: string) => { + const trimmed = line.trim(); + if (!trimmed) return; + + const record = JSON.parse(trimmed) as + | { type: 'node'; data: GraphNode } + | { type: 'relationship'; data: GraphRelationship } + | { type: 'error'; error: string }; + + if (record.type === 'node') { + nodes.push(record.data); + return; + } + if (record.type === 'relationship') { + relationships.push(record.data); + return; + } + if (record.type === 'error') { + throw new BackendError(record.error, response.status || 500, 'server'); + } + }; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + downloaded += value.length; + onProgress?.(downloaded, total); + buffer += decoder.decode(value, { stream: true }); + + const lines = buffer.split('\n'); + buffer = lines.pop() || ''; + for (const line of lines) { + parseLine(line); + } + } + + buffer += decoder.decode(); + parseLine(buffer); + + return { nodes, relationships }; +}; + /** Execute a Cypher query. Returns rows. */ export const runQuery = async ( cypher: string, diff --git a/gitnexus-web/test/unit/server-connection.test.ts b/gitnexus-web/test/unit/server-connection.test.ts index a818adb13..f5ee43c53 100644 --- a/gitnexus-web/test/unit/server-connection.test.ts +++ b/gitnexus-web/test/unit/server-connection.test.ts @@ -1,5 +1,5 @@ -import { describe, expect, it } from 'vitest'; -import { normalizeServerUrl } from '../../src/services/backend-client'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { fetchGraph, normalizeServerUrl, setBackendUrl } from '../../src/services/backend-client'; describe('normalizeServerUrl', () => { it('adds http:// to localhost', () => { @@ -31,3 +31,137 @@ describe('normalizeServerUrl', () => { expect(normalizeServerUrl('https://gitnexus.example.com')).toBe('https://gitnexus.example.com'); }); }); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('fetchGraph', () => { + it('requests streamed graph responses from the backend', async () => { + setBackendUrl('http://localhost:4747'); + + const fetchMock = vi.fn().mockResolvedValue( + new Response('{"nodes":[],"relationships":[]}', { + status: 200, + headers: { + 'Content-Type': 'application/json', + }, + }), + ); + vi.stubGlobal('fetch', fetchMock); + + await fetchGraph('big-repo'); + + expect(fetchMock).toHaveBeenCalledWith( + expect.stringContaining('/api/graph?repo=big-repo&stream=true'), + expect.any(Object), + ); + }); + + it('parses NDJSON graph streams incrementally', async () => { + setBackendUrl('http://localhost:4747'); + + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue( + encoder.encode( + [ + '{"type":"node","data":{"id":"File:src/app.ts","label":"File","properties":{"name":"app.ts","filePath":"src/app.ts"}}}\n', + '{"type":"relationship","data":{"id":"File:src/app.ts_CONTAINS_Function:src/app.ts:main","type":"CONTAINS","sourceId":"File:src/app.ts","targetId":"Function:src/app.ts:main"}}\n', + ].join(''), + ), + ); + controller.close(); + }, + }); + + vi.stubGlobal( + 'fetch', + vi.fn().mockResolvedValue( + new Response(stream, { + status: 200, + headers: { + 'Content-Type': 'application/x-ndjson', + }, + }), + ), + ); + + const progress = vi.fn(); + const result = await fetchGraph('big-repo', { onProgress: progress }); + + expect(result.nodes).toHaveLength(1); + expect(result.relationships).toHaveLength(1); + expect(result.nodes[0].id).toBe('File:src/app.ts'); + expect(result.relationships[0].type).toBe('CONTAINS'); + expect(progress).toHaveBeenCalled(); + }); + + it('parses NDJSON graph lines split across chunks', async () => { + setBackendUrl('http://localhost:4747'); + + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue( + encoder.encode( + '{"type":"node","data":{"id":"File:src/app.ts","label":"File","properties":{"name":"app.ts"', + ), + ); + controller.enqueue( + encoder.encode( + ',"filePath":"src/app.ts"}}}\n{"type":"relationship","data":{"id":"File:src/app.ts_CONTAINS_Function:src/app.ts:main","type":"CONTAINS","sourceId":"File:src/app.ts","targetId":"Function:src/app.ts:main"}}\n', + ), + ); + controller.close(); + }, + }); + + vi.stubGlobal( + 'fetch', + vi.fn().mockResolvedValue( + new Response(stream, { + status: 200, + headers: { + 'Content-Type': 'application/x-ndjson', + }, + }), + ), + ); + + const result = await fetchGraph('big-repo'); + + expect(result.nodes).toHaveLength(1); + expect(result.relationships).toHaveLength(1); + expect(result.nodes[0].properties.filePath).toBe('src/app.ts'); + }); + + it('throws backend errors emitted in the NDJSON stream', async () => { + setBackendUrl('http://localhost:4747'); + + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('{"type":"error","error":"stream failed"}\n')); + controller.close(); + }, + }); + + vi.stubGlobal( + 'fetch', + vi.fn().mockResolvedValue( + new Response(stream, { + status: 200, + headers: { + 'Content-Type': 'application/x-ndjson', + }, + }), + ), + ); + + await expect(fetchGraph('big-repo')).rejects.toMatchObject({ + message: 'stream failed', + }); + }); +}); diff --git a/gitnexus/src/core/lbug/lbug-adapter.ts b/gitnexus/src/core/lbug/lbug-adapter.ts index 1363257d5..88a6e9bba 100644 --- a/gitnexus/src/core/lbug/lbug-adapter.ts +++ b/gitnexus/src/core/lbug/lbug-adapter.ts @@ -637,6 +637,34 @@ export const executeQuery = async (cypher: string): Promise => { return rows; }; +export const streamQuery = async ( + cypher: string, + onRow: (row: any) => void | Promise, +): Promise => { + if (!conn) { + throw new Error('LadybugDB not initialized. Call initLbug first.'); + } + + const queryResult = await conn.query(cypher); + const result = Array.isArray(queryResult) ? queryResult[0] : queryResult; + let rowCount = 0; + + try { + while (await result.hasNext()) { + const row = await result.getNext(); + await onRow(row); + rowCount++; + } + return rowCount; + } finally { + try { + await result.close(); + } catch { + // Best-effort cleanup only. + } + } +}; + /** * Execute a single parameterized query (prepare/execute pattern). * Prevents Cypher injection by binding values as parameters. diff --git a/gitnexus/src/server/api.ts b/gitnexus/src/server/api.ts index 8422af6ba..8111c287b 100644 --- a/gitnexus/src/server/api.ts +++ b/gitnexus/src/server/api.ts @@ -18,6 +18,7 @@ import { executeQuery, executePrepared, executeWithReusedStatement, + streamQuery, closeLbug, withLbugDb, } from '../core/lbug/lbug-adapter.js'; @@ -105,75 +106,228 @@ export const isAllowedOrigin = (origin: string | undefined): boolean => { return false; }; +type GraphStreamRecord = + | { type: 'node'; data: GraphNode } + | { type: 'relationship'; data: GraphRelationship } + | { type: 'error'; error: string }; + +export class ClientDisconnectedError extends Error { + constructor() { + super('Client disconnected during graph stream'); + this.name = 'ClientDisconnectedError'; + } +} + +export const isIgnorableGraphQueryError = (err: unknown): boolean => { + const message = err instanceof Error ? err.message : String(err); + return ( + message.includes('does not exist') || + message.includes('not found') || + message.includes('No table named') + ); +}; + +const ensureStreamIsWritable = (res: express.Response, signal?: AbortSignal): void => { + if (signal?.aborted || res.destroyed || res.writableEnded) { + throw new ClientDisconnectedError(); + } +}; + +const waitForDrain = async (res: express.Response, signal?: AbortSignal): Promise => { + ensureStreamIsWritable(res, signal); + + await new Promise((resolve, reject) => { + const cleanup = () => { + res.off('drain', onDrain); + res.off('close', onClose); + signal?.removeEventListener('abort', onAbort); + }; + + const onDrain = () => { + cleanup(); + resolve(); + }; + const onClose = () => { + cleanup(); + reject(new ClientDisconnectedError()); + }; + const onAbort = () => { + cleanup(); + reject(new ClientDisconnectedError()); + }; + + res.once('drain', onDrain); + res.once('close', onClose); + signal?.addEventListener('abort', onAbort, { once: true }); + + if (signal?.aborted || res.destroyed || res.writableEnded) { + onAbort(); + } + }); + + ensureStreamIsWritable(res, signal); +}; + +const isClientDisconnectWriteError = (err: unknown): boolean => { + if (!(err instanceof Error)) return false; + return ( + (err as NodeJS.ErrnoException).code === 'ERR_STREAM_DESTROYED' || + (err as NodeJS.ErrnoException).code === 'EPIPE' || + (err as NodeJS.ErrnoException).code === 'ECONNRESET' || + err.message.includes('write after end') + ); +}; + +export const writeNdjsonRecord = async ( + res: express.Response, + record: GraphStreamRecord, + signal?: AbortSignal, +): Promise => { + ensureStreamIsWritable(res, signal); + + try { + const canContinue = res.write(JSON.stringify(record) + '\n'); + if (!canContinue) { + await waitForDrain(res, signal); + } + } catch (err) { + if (isClientDisconnectWriteError(err)) { + throw new ClientDisconnectedError(); + } + throw err; + } +}; + const buildGraph = async ( includeContent = false, ): Promise<{ nodes: GraphNode[]; relationships: GraphRelationship[] }> => { const nodes: GraphNode[] = []; for (const table of NODE_TABLES) { try { - let query = ''; - if (table === 'File') { - query = includeContent - ? `MATCH (n:File) RETURN n.id AS id, n.name AS name, n.filePath AS filePath, n.content AS content` - : `MATCH (n:File) RETURN n.id AS id, n.name AS name, n.filePath AS filePath`; - } else if (table === 'Folder') { - query = `MATCH (n:Folder) RETURN n.id AS id, n.name AS name, n.filePath AS filePath`; - } else if (table === 'Community') { - query = `MATCH (n:Community) RETURN n.id AS id, n.label AS label, n.heuristicLabel AS heuristicLabel, n.cohesion AS cohesion, n.symbolCount AS symbolCount`; - } else if (table === 'Process') { - query = `MATCH (n:Process) RETURN n.id AS id, n.label AS label, n.heuristicLabel AS heuristicLabel, n.processType AS processType, n.stepCount AS stepCount, n.communities AS communities, n.entryPointId AS entryPointId, n.terminalId AS terminalId`; - } else { - query = includeContent - ? `MATCH (n:${table}) RETURN n.id AS id, n.name AS name, n.filePath AS filePath, n.startLine AS startLine, n.endLine AS endLine, n.content AS content` - : `MATCH (n:${table}) RETURN n.id AS id, n.name AS name, n.filePath AS filePath, n.startLine AS startLine, n.endLine AS endLine`; - } - - const rows = await executeQuery(query); + const rows = await executeQuery(getNodeQuery(table, includeContent)); for (const row of rows) { - nodes.push({ - id: row.id ?? row[0], - label: table as GraphNode['label'], - properties: { - name: row.name ?? row.label ?? row[1], - filePath: row.filePath ?? row[2], - startLine: row.startLine, - endLine: row.endLine, - content: includeContent ? row.content : undefined, - heuristicLabel: row.heuristicLabel, - cohesion: row.cohesion, - symbolCount: row.symbolCount, - processType: row.processType, - stepCount: row.stepCount, - communities: row.communities, - entryPointId: row.entryPointId, - terminalId: row.terminalId, - } as GraphNode['properties'], - }); + nodes.push(mapGraphNodeRow(table, row, includeContent)); + } + } catch (err) { + if (!isIgnorableGraphQueryError(err)) { + throw err; } - } catch { - // ignore empty tables } } const relationships: GraphRelationship[] = []; - const relRows = await executeQuery( - `MATCH (a)-[r:CodeRelation]->(b) RETURN a.id AS sourceId, b.id AS targetId, r.type AS type, r.confidence AS confidence, r.reason AS reason, r.step AS step`, - ); + const relRows = await executeQuery(GRAPH_RELATIONSHIP_QUERY); for (const row of relRows) { - relationships.push({ - id: `${row.sourceId}_${row.type}_${row.targetId}`, - type: row.type, - sourceId: row.sourceId, - targetId: row.targetId, - confidence: row.confidence, - reason: row.reason, - step: row.step, - }); + relationships.push(mapGraphRelationshipRow(row)); } return { nodes, relationships }; }; +const GRAPH_RELATIONSHIP_QUERY = + `MATCH (a)-[r:CodeRelation]->(b) RETURN a.id AS sourceId, b.id AS targetId, ` + + `r.type AS type, r.confidence AS confidence, r.reason AS reason, r.step AS step`; + +const quoteNodeTable = (table: string): string => `\`${table.replace(/`/g, '``')}\``; + +const getNodeQuery = (table: string, includeContent: boolean): string => { + const tableLabel = quoteNodeTable(table); + + if (table === 'File') { + return includeContent + ? `MATCH (n:${tableLabel}) RETURN n.id AS id, n.name AS name, n.filePath AS filePath, n.content AS content` + : `MATCH (n:${tableLabel}) RETURN n.id AS id, n.name AS name, n.filePath AS filePath`; + } + if (table === 'Folder') { + return `MATCH (n:${tableLabel}) RETURN n.id AS id, n.name AS name, n.filePath AS filePath`; + } + if (table === 'Community') { + return `MATCH (n:${tableLabel}) RETURN n.id AS id, n.label AS label, n.heuristicLabel AS heuristicLabel, n.cohesion AS cohesion, n.symbolCount AS symbolCount`; + } + if (table === 'Process') { + return `MATCH (n:${tableLabel}) RETURN n.id AS id, n.label AS label, n.heuristicLabel AS heuristicLabel, n.processType AS processType, n.stepCount AS stepCount, n.communities AS communities, n.entryPointId AS entryPointId, n.terminalId AS terminalId`; + } + if (table === 'Route') { + return `MATCH (n:${tableLabel}) RETURN n.id AS id, n.name AS name, n.filePath AS filePath, n.responseKeys AS responseKeys, n.errorKeys AS errorKeys, n.middleware AS middleware`; + } + if (table === 'Tool') { + return `MATCH (n:${tableLabel}) RETURN n.id AS id, n.name AS name, n.filePath AS filePath, n.description AS description`; + } + return includeContent + ? `MATCH (n:${tableLabel}) RETURN n.id AS id, n.name AS name, n.filePath AS filePath, n.startLine AS startLine, n.endLine AS endLine, n.content AS content` + : `MATCH (n:${tableLabel}) RETURN n.id AS id, n.name AS name, n.filePath AS filePath, n.startLine AS startLine, n.endLine AS endLine`; +}; + +const mapGraphNodeRow = (table: string, row: any, includeContent: boolean): GraphNode => ({ + id: row.id ?? row[0], + label: table as GraphNode['label'], + properties: { + name: row.name ?? row.label ?? row[1], + filePath: row.filePath ?? row[2], + startLine: row.startLine, + endLine: row.endLine, + content: includeContent ? row.content : undefined, + responseKeys: row.responseKeys, + errorKeys: row.errorKeys, + middleware: row.middleware, + heuristicLabel: row.heuristicLabel, + cohesion: row.cohesion, + symbolCount: row.symbolCount, + description: row.description, + processType: row.processType, + stepCount: row.stepCount, + communities: row.communities, + entryPointId: row.entryPointId, + terminalId: row.terminalId, + } as GraphNode['properties'], +}); + +const mapGraphRelationshipRow = (row: any): GraphRelationship => ({ + id: `${row.sourceId}_${row.type}_${row.targetId}`, + type: row.type, + sourceId: row.sourceId, + targetId: row.targetId, + confidence: row.confidence, + reason: row.reason, + step: row.step, +}); + +export const streamGraphNdjson = async ( + res: express.Response, + includeContent = false, + signal?: AbortSignal, +): Promise => { + for (const table of NODE_TABLES) { + try { + await streamQuery(getNodeQuery(table, includeContent), async (row) => { + await writeNdjsonRecord( + res, + { + type: 'node', + data: mapGraphNodeRow(table, row, includeContent), + }, + signal, + ); + }); + } catch (err) { + if (!isIgnorableGraphQueryError(err)) { + throw err; + } + } + } + + await streamQuery(GRAPH_RELATIONSHIP_QUERY, async (row) => { + await writeNdjsonRecord( + res, + { + type: 'relationship', + data: mapGraphRelationshipRow(row), + }, + signal, + ); + }); +}; + /** * Mount an SSE progress endpoint for a JobManager. * Handles: initial state, terminal events, heartbeat, event IDs, client disconnect. @@ -464,10 +618,60 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => } const lbugPath = path.join(entry.storagePath, 'lbug'); const includeContent = req.query.includeContent === 'true'; + const stream = req.query.stream === 'true'; + + if (stream) { + const abortController = new AbortController(); + let responseFinished = false; + const markFinished = () => { + responseFinished = true; + }; + const abortStreaming = () => { + if (!responseFinished) { + abortController.abort(); + } + }; + + res.setHeader('Content-Type', 'application/x-ndjson; charset=utf-8'); + res.setHeader('Cache-Control', 'no-cache'); + res.flushHeaders(); + + req.once('aborted', abortStreaming); + res.once('finish', markFinished); + res.once('close', abortStreaming); + + try { + await withLbugDb(lbugPath, async () => + streamGraphNdjson(res, includeContent, abortController.signal), + ); + if (!abortController.signal.aborted && !res.writableEnded) { + res.end(); + } + } finally { + req.off('aborted', abortStreaming); + res.off('finish', markFinished); + res.off('close', abortStreaming); + } + return; + } + const graph = await withLbugDb(lbugPath, async () => buildGraph(includeContent)); res.json(graph); } catch (err: any) { - res.status(500).json({ error: err.message || 'Failed to build graph' }); + if (err instanceof ClientDisconnectedError) { + return; + } + const message = err.message || 'Failed to build graph'; + if (res.headersSent) { + try { + res.write(JSON.stringify({ type: 'error', error: message }) + '\n'); + } catch { + // Best-effort only after streaming has started. + } + res.end(); + return; + } + res.status(500).json({ error: message }); } }); diff --git a/gitnexus/test/unit/api-graph-streaming.test.ts b/gitnexus/test/unit/api-graph-streaming.test.ts new file mode 100644 index 000000000..0cc69833c --- /dev/null +++ b/gitnexus/test/unit/api-graph-streaming.test.ts @@ -0,0 +1,235 @@ +import { EventEmitter } from 'node:events'; +import { describe, expect, it, vi, beforeEach } from 'vitest'; + +const { lbugMocks } = vi.hoisted(() => ({ + lbugMocks: { + streamQuery: vi.fn(), + }, +})); + +vi.mock('../../src/core/lbug/lbug-adapter.js', async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, ...lbugMocks }; +}); + +import { ClientDisconnectedError, streamGraphNdjson } from '../../src/server/api.js'; + +const createMockResponse = (writeImpl?: (chunk: string) => boolean) => { + const response = new EventEmitter() as any; + response.writableEnded = false; + response.destroyed = false; + response.write = vi.fn((chunk: string) => (writeImpl ? writeImpl(chunk) : true)); + return response; +}; + +describe('streamGraphNdjson', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('waits for drain when writes hit backpressure', async () => { + lbugMocks.streamQuery.mockImplementation( + async (query: string, onRow: (row: any) => Promise) => { + if (query.includes('MATCH (n:`File`)')) { + await onRow({ id: 'File:src/app.ts', name: 'app.ts', filePath: 'src/app.ts' }); + return 1; + } + if (query.includes('CodeRelation')) { + await onRow({ + sourceId: 'File:src/app.ts', + targetId: 'Function:src/app.ts:main', + type: 'CONTAINS', + }); + return 1; + } + return 0; + }, + ); + + const writes: string[] = []; + let firstWrite = true; + const response = createMockResponse((chunk) => { + writes.push(chunk); + if (firstWrite) { + firstWrite = false; + return false; + } + return true; + }); + + let settled = false; + const pending = streamGraphNdjson(response, false).then(() => { + settled = true; + }); + + await Promise.resolve(); + expect(writes).toHaveLength(1); + expect(settled).toBe(false); + + response.emit('drain'); + await pending; + + expect(writes).toHaveLength(2); + }); + + it('stops streaming when the client disconnects', async () => { + const controller = new AbortController(); + lbugMocks.streamQuery.mockImplementation( + async (query: string, onRow: (row: any) => Promise) => { + if (!query.includes('MATCH (n:`File`)')) { + return 0; + } + await onRow({ id: 'File:src/app.ts', name: 'app.ts', filePath: 'src/app.ts' }); + controller.abort(); + await onRow({ id: 'File:src/other.ts', name: 'other.ts', filePath: 'src/other.ts' }); + return 2; + }, + ); + + const response = createMockResponse(); + + await expect(streamGraphNdjson(response, false, controller.signal)).rejects.toBeInstanceOf( + ClientDisconnectedError, + ); + expect(response.write).toHaveBeenCalledTimes(1); + }); + + it('rethrows non-missing table errors', async () => { + lbugMocks.streamQuery.mockImplementation(async (query: string) => { + if (query.includes('MATCH (n:`File`)')) { + throw new Error('database unavailable'); + } + return 0; + }); + + const response = createMockResponse(); + await expect(streamGraphNdjson(response, false)).rejects.toThrow('database unavailable'); + }); + + it('ignores missing-table errors while continuing the stream', async () => { + lbugMocks.streamQuery.mockImplementation( + async (query: string, onRow: (row: any) => Promise) => { + if (query.includes('MATCH (n:`File`)')) { + throw new Error('Table File does not exist'); + } + if (query.includes('CodeRelation')) { + await onRow({ + sourceId: 'File:src/app.ts', + targetId: 'Function:src/app.ts:main', + type: 'CONTAINS', + }); + return 1; + } + return 0; + }, + ); + + const response = createMockResponse(); + await expect(streamGraphNdjson(response, false)).resolves.toBeUndefined(); + expect(response.write).toHaveBeenCalledTimes(1); + }); + + it('quotes node table names in generated Cypher queries', async () => { + lbugMocks.streamQuery.mockImplementation(async () => 0); + + const response = createMockResponse(); + await expect(streamGraphNdjson(response, false)).resolves.toBeUndefined(); + + expect(lbugMocks.streamQuery).toHaveBeenCalledWith( + expect.stringContaining('MATCH (n:`Macro`)'), + expect.any(Function), + ); + }); + + it('streams Route and Tool nodes without requiring startLine fields', async () => { + lbugMocks.streamQuery.mockImplementation( + async (query: string, onRow: (row: any) => Promise) => { + if (query.includes('MATCH (n:`Route`)')) { + expect(query).not.toContain('startLine'); + await onRow({ + id: 'Route:/api/graph:GET', + name: 'GET /api/graph', + filePath: 'src/server/api.ts', + responseKeys: ['nodes', 'relationships'], + errorKeys: ['error'], + middleware: ['withAuth'], + }); + return 1; + } + if (query.includes('MATCH (n:`Tool`)')) { + expect(query).not.toContain('startLine'); + await onRow({ + id: 'Tool:gitnexus_query', + name: 'gitnexus_query', + filePath: 'src/mcp/resources.ts', + description: 'Query the code graph', + }); + return 1; + } + return 0; + }, + ); + + const writes: string[] = []; + const response = createMockResponse((chunk) => { + writes.push(chunk); + return true; + }); + + await expect(streamGraphNdjson(response, false)).resolves.toBeUndefined(); + + const records = writes.map((chunk) => JSON.parse(chunk)); + expect(records).toContainEqual({ + type: 'node', + data: { + id: 'Route:/api/graph:GET', + label: 'Route', + properties: { + name: 'GET /api/graph', + filePath: 'src/server/api.ts', + startLine: undefined, + endLine: undefined, + content: undefined, + responseKeys: ['nodes', 'relationships'], + errorKeys: ['error'], + middleware: ['withAuth'], + heuristicLabel: undefined, + cohesion: undefined, + symbolCount: undefined, + description: undefined, + processType: undefined, + stepCount: undefined, + communities: undefined, + entryPointId: undefined, + terminalId: undefined, + }, + }, + }); + expect(records).toContainEqual({ + type: 'node', + data: { + id: 'Tool:gitnexus_query', + label: 'Tool', + properties: { + name: 'gitnexus_query', + filePath: 'src/mcp/resources.ts', + startLine: undefined, + endLine: undefined, + content: undefined, + responseKeys: undefined, + errorKeys: undefined, + middleware: undefined, + heuristicLabel: undefined, + cohesion: undefined, + symbolCount: undefined, + description: 'Query the code graph', + processType: undefined, + stepCount: undefined, + communities: undefined, + entryPointId: undefined, + terminalId: undefined, + }, + }, + }); + }); +});