GitNexus/gitnexus/test/unit/api-graph-streaming.test.ts

import { EventEmitter } from 'node:events';
import { describe, expect, it, vi, beforeEach } from 'vitest';

const { lbugMocks } = vi.hoisted(() => ({
  lbugMocks: {
    executeQuery: vi.fn(),
    streamQuery: vi.fn(),
  },
}));

vi.mock('../../src/core/lbug/lbug-adapter.js', async (importOriginal) => {
  const actual = await importOriginal();
  return { ...actual, ...lbugMocks };
});

import { buildGraph, 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');
          expect(query).toContain('runtimeConfirmed');
          expect(query).toContain('runtimeSource');
          expect(query).toContain('runtimeStatus');
          await onRow({
            id: 'Route:/api/graph:GET',
            name: 'GET /api/graph',
            filePath: 'src/server/api.ts',
            responseKeys: ['nodes', 'relationships'],
            errorKeys: ['error'],
            middleware: ['withAuth'],
            runtimeConfirmed: true,
            runtimeSource: 'spring-actuator',
            runtimeStatus: 'runtime-confirmed',
          });
          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'],
          runtimeConfirmed: true,
          runtimeSource: 'spring-actuator',
          runtimeStatus: 'runtime-confirmed',
          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,
        },
      },
    });
  });

  it('retries Route streaming with the legacy projection when runtime columns are absent', async () => {
    let modernRouteAttempts = 0;
    let legacyRouteAttempts = 0;
    lbugMocks.streamQuery.mockImplementation(
      async (query: string, onRow: (row: any) => Promise<void>) => {
        if (!query.includes('MATCH (n:`Route`)')) return 0;
        if (query.includes('runtimeConfirmed')) {
          modernRouteAttempts++;
          throw new Error('Binder exception: Cannot find property runtimeConfirmed for n.');
        }
        legacyRouteAttempts++;
        await onRow({
          id: 'Route:GET /legacy',
          name: '/legacy',
          filePath: 'src/LegacyController.java',
          responseKeys: [],
          errorKeys: [],
          middleware: [],
        });
        return 1;
      },
    );

    const writes: string[] = [];
    const response = createMockResponse((chunk) => {
      writes.push(chunk);
      return true;
    });

    await expect(streamGraphNdjson(response, false)).resolves.toBeUndefined();

    expect(modernRouteAttempts).toBe(1);
    expect(legacyRouteAttempts).toBe(1);
    const routeRecord = writes
      .map((chunk) => JSON.parse(chunk))
      .find((record) => record.data?.id === 'Route:GET /legacy');
    expect(routeRecord).toMatchObject({
      type: 'node',
      data: { id: 'Route:GET /legacy' },
    });
    expect(routeRecord.data.properties.runtimeConfirmed).toBe(false);
    expect(routeRecord.data.properties).not.toHaveProperty('runtimeSource');
    expect(routeRecord.data.properties).not.toHaveProperty('runtimeStatus');
  });

  it('retries non-streaming Route loading with the legacy projection', async () => {
    let modernRouteAttempts = 0;
    let legacyRouteAttempts = 0;
    lbugMocks.executeQuery.mockImplementation(async (query: string) => {
      if (!query.includes('MATCH (n:`Route`)')) return [];
      if (query.includes('runtimeConfirmed')) {
        modernRouteAttempts++;
        throw new Error('Binder exception: Cannot find property runtimeConfirmed for n.');
      }
      legacyRouteAttempts++;
      return [
        {
          id: 'Route:GET /legacy',
          name: '/legacy',
          filePath: 'src/LegacyController.java',
          responseKeys: [],
          errorKeys: [],
          middleware: [],
        },
      ];
    });

    const graph = await buildGraph(false);

    expect(modernRouteAttempts).toBe(1);
    expect(legacyRouteAttempts).toBe(1);
    expect(graph.nodes).toContainEqual(
      expect.objectContaining({
        id: 'Route:GET /legacy',
        label: 'Route',
        properties: expect.objectContaining({
          runtimeConfirmed: false,
        }),
      }),
    );
  });

  // Taint/PDG substrate (#2080): BasicBlock has no name/content columns, so its
  // getNodeQuery projects none — mapGraphNodeRow must still yield a `string`
  // name (NodeProperties.name contract) or the web layer derefs undefined.
  it('emits a string name for BasicBlock nodes (no name column)', async () => {
    lbugMocks.streamQuery.mockImplementation(
      async (query: string, onRow: (row: any) => Promise<void>) => {
        if (query.includes('MATCH (n:`BasicBlock`)')) {
          expect(query).not.toContain('n.name'); // BasicBlock projects no name column
          await onRow({
            id: 'BasicBlock:src/a.ts:0',
            filePath: 'src/a.ts',
            startLine: 1,
            endLine: 3,
            text: 'const x = req.body;',
          });
          // a block with no text must still map to a string name, not undefined
          await onRow({
            id: 'BasicBlock:src/a.ts:1',
            filePath: 'src/a.ts',
            startLine: 4,
            endLine: 4,
          });
          return 2;
        }
        return 0;
      },
    );

    const writes: string[] = [];
    const response = createMockResponse((chunk) => {
      writes.push(chunk);
      return true;
    });

    await expect(streamGraphNdjson(response, false)).resolves.toBeUndefined();

    const blocks = writes
      .map((chunk) => JSON.parse(chunk))
      .filter((r) => r.type === 'node' && r.data.label === 'BasicBlock');
    expect(blocks).toHaveLength(2);
    for (const b of blocks) {
      expect(typeof b.data.properties.name).toBe('string'); // never undefined
    }
    // falls back to the block text when present, else the empty-string floor
    expect(blocks[0].data.properties.name).toBe('const x = req.body;');
    expect(blocks[0].data.properties.text).toBe('const x = req.body;');
    expect(blocks[1].data.properties.name).toBe('');
  });
});

/**
 * The Destination overlay across the API boundary.
 *
 * `getNodeQuery` has projected the five overlay columns since the phase landed,
 * but `mapGraphNodeRow` built `properties` from an explicit literal and copied
 * none of them, so every client of `/api/graph` saw a `Destination` stripped of
 * the only fields that make it one. Both entry points share that mapper, so
 * both are asserted here: a fix that reached only the streaming path would look
 * complete and leave `buildGraph` exactly as broken.
 */
const RESOLVED_DESTINATION_ROW = {
  // A resolved destination is keyed by `(broker, address)`, so `broker` is part
  // of the id and not only of the payload.
  id: 'Destination:kafka orders.v1',
  name: 'orders.v1',
  filePath: '',
  startLine: null,
  endLine: null,
  address: 'orders.v1',
  broker: 'kafka',
  resolution: 'literal',
  configKey: null,
  configDefault: null,
};

const UNRESOLVED_DESTINATION_ROW = {
  id: 'Destination:site:src/main/java/com/example/OrderConsumer.java',
  name: '${app.topic}',
  filePath: 'src/main/java/com/example/OrderConsumer.java',
  startLine: 11,
  endLine: 14,
  address: null,
  broker: 'kafka',
  resolution: 'unresolved-config-key',
  configKey: 'app.topic',
  configDefault: null,
};

describe('Destination overlay properties', () => {
  beforeEach(() => {
    vi.clearAllMocks();
  });

  it('carries the overlay through buildGraph', async () => {
    lbugMocks.executeQuery.mockImplementation(async (query: string) => {
      if (!query.includes('MATCH (n:`Destination`)')) return [];
      expect(query).toContain('n.address AS address');
      return [RESOLVED_DESTINATION_ROW, UNRESOLVED_DESTINATION_ROW];
    });

    const graph = await buildGraph(false);
    const destinations = graph.nodes.filter((node) => node.label === 'Destination');
    expect(destinations).toHaveLength(2);

    expect(destinations[0]?.properties).toMatchObject({
      name: 'orders.v1',
      address: 'orders.v1',
      broker: 'kafka',
      resolution: 'literal',
    });
    expect(destinations[1]?.properties).toMatchObject({
      name: '${app.topic}',
      broker: 'kafka',
      resolution: 'unresolved-config-key',
      configKey: 'app.topic',
    });
  });

  it('carries the overlay through streamGraphNdjson', async () => {
    lbugMocks.streamQuery.mockImplementation(
      async (query: string, onRow: (row: any) => Promise<void>) => {
        if (!query.includes('MATCH (n:`Destination`)')) return 0;
        await onRow(RESOLVED_DESTINATION_ROW);
        await onRow(UNRESOLVED_DESTINATION_ROW);
        return 2;
      },
    );

    const writes: string[] = [];
    const response = createMockResponse((chunk) => {
      writes.push(chunk);
      return true;
    });

    await expect(streamGraphNdjson(response, false)).resolves.toBeUndefined();

    const destinations = writes
      .map((chunk) => JSON.parse(chunk))
      .filter((record) => record.type === 'node' && record.data.label === 'Destination');
    expect(destinations).toHaveLength(2);
    expect(destinations[0].data.properties).toMatchObject({
      address: 'orders.v1',
      broker: 'kafka',
      resolution: 'literal',
    });
    expect(destinations[1].data.properties).toMatchObject({
      broker: 'kafka',
      resolution: 'unresolved-config-key',
      configKey: 'app.topic',
    });
  });

  it('keeps an unresolved address ABSENT rather than serializing it as null', async () => {
    // The whole feature rests on an unresolved destination having no `address`.
    // A `null` survives JSON, and two nulls are as equal as two empty strings:
    // a client grouping destinations by `address` would join every unresolved
    // one in the repository into a single connected blob. `toMatchObject`
    // cannot see this — only the serialized text can.
    lbugMocks.streamQuery.mockImplementation(
      async (query: string, onRow: (row: any) => Promise<void>) => {
        if (!query.includes('MATCH (n:`Destination`)')) return 0;
        await onRow(UNRESOLVED_DESTINATION_ROW);
        return 1;
      },
    );

    const writes: string[] = [];
    const response = createMockResponse((chunk) => {
      writes.push(chunk);
      return true;
    });
    await expect(streamGraphNdjson(response, false)).resolves.toBeUndefined();

    const [chunk] = writes.filter((written) => written.includes('"Destination"'));
    expect(chunk).toBeDefined();
    expect(chunk).not.toContain('"address"');
    const properties = JSON.parse(chunk as string).data.properties;
    expect('address' in properties).toBe(false);
    // `configDefault` is the same shape of hazard, one column over: it is NULL
    // on this row, and a serialized null is a value every other destination
    // without a default would share.
    expect('configDefault' in properties).toBe(false);
  });

  it('does not put the overlay keys on nodes of other labels', async () => {
    lbugMocks.executeQuery.mockImplementation(async (query: string) => {
      if (!query.includes('MATCH (n:`File`)')) return [];
      return [{ id: 'File:src/app.ts', name: 'app.ts', filePath: 'src/app.ts', address: 'oops' }];
    });

    const graph = await buildGraph(false);
    const [file] = graph.nodes.filter((node) => node.label === 'File');
    expect(file?.properties.address).toBeUndefined();
    expect(file?.properties.broker).toBeUndefined();
  });
});