mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-08-28 05:25:25 +00:00
[codex] fix large repository graph loading (#732)
* fix(web): stream large graph responses * fix(server): harden graph streaming * fix(ci): stabilize graph loading coverage --------- Co-authored-by: gfwangjie <gfwangjie@gf.com.cn>
This commit is contained in:
JaysonAlbert
GitHub
parent
4450a14b98
commit
338cb01ee0
7 changed files with 746 additions and 82 deletions
|
|
@ -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 });
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ export const StatusBar = () => {
|
|||
</a>
|
||||
|
||||
{/* Right - Stats */}
|
||||
<div className="flex items-center gap-3">
|
||||
<div className="flex items-center gap-3" data-testid="graph-stats">
|
||||
{graph && (
|
||||
<>
|
||||
<span>{nodeCount} nodes</span>
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<Uint8Array>({
|
||||
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<Uint8Array>({
|
||||
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<Uint8Array>({
|
||||
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',
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -637,6 +637,34 @@ export const executeQuery = async (cypher: string): Promise<any[]> => {
|
|||
return rows;
|
||||
};
|
||||
|
||||
export const streamQuery = async (
|
||||
cypher: string,
|
||||
onRow: (row: any) => void | Promise<void>,
|
||||
): Promise<number> => {
|
||||
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.
|
||||
|
|
|
|||
|
|
@ -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<void> => {
|
||||
ensureStreamIsWritable(res, signal);
|
||||
|
||||
await new Promise<void>((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<void> => {
|
||||
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<void> => {
|
||||
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 });
|
||||
}
|
||||
});
|
||||
|
||||
|
|
|
|||
235
gitnexus/test/unit/api-graph-streaming.test.ts
Normal file
235
gitnexus/test/unit/api-graph-streaming.test.ts
Normal file
|
|
@ -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<void>) => {
|
||||
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<void>) => {
|
||||
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<void>) => {
|
||||
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<void>) => {
|
||||
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,
|
||||
},
|
||||
},
|
||||
});
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Reference in a new issue