diff --git a/package-lock.json b/package-lock.json index d5e2ef1..134b136 100644 --- a/package-lock.json +++ b/package-lock.json @@ -44,6 +44,7 @@ "three": "^0.170.0", "three-instanced-uniforms-mesh": "^0.52.0", "troika-three-text": "^0.52.2", + "turingdb": "3.0.3-dev.202611041359", "typescript": "~5.6.2", "vite": "^6.1.6", "vite-plugin-glsl": "^1.3.1", @@ -6559,6 +6560,16 @@ "dev": true, "license": "0BSD" }, + "node_modules/turingdb": { + "version": "3.0.3-dev.202611041359", + "resolved": "https://registry.npmjs.org/turingdb/-/turingdb-3.0.3-dev.202611041359.tgz", + "integrity": "sha512-9nlPeEcSWhNzSpMqAMb+SWcTlsPpZywHq2BJLzxJC34J+64F27RumgiBj4gqWwbYhb2vxvqU++8ptia2HN6kng==", + "dev": true, + "license": "Apache-2.0", + "engines": { + "node": ">=18" + } + }, "node_modules/type-is": { "version": "1.6.18", "resolved": "https://registry.npmjs.org/type-is/-/type-is-1.6.18.tgz", diff --git a/package.json b/package.json index e3f0600..c5d86ea 100644 --- a/package.json +++ b/package.json @@ -53,6 +53,7 @@ "three": "^0.170.0", "three-instanced-uniforms-mesh": "^0.52.0", "troika-three-text": "^0.52.2", + "turingdb": "3.0.3-dev.202611041359", "typescript": "~5.6.2", "vite": "^6.1.6", "vite-plugin-glsl": "^1.3.1", diff --git a/src/api/api.ts b/src/api/api.ts index dfe1e2a..bd7d510 100644 --- a/src/api/api.ts +++ b/src/api/api.ts @@ -28,10 +28,16 @@ import { type ListPropertyTypesResponse, } from './responses' +import { TuringClient, TuringQueryError } from 'turingdb' + import type { NodeEntries } from './models/nodeEntries.model' import type { NodeEntry } from './models/nodeEntry.model' import type { EdgeEntry } from './models/edgeEntry.model' +// One client for the whole app: it holds no per-query state, since every call +// passes its own graph. The first query compiles the wasm decoder. +const client = new TuringClient({ url: '/api/query' }) + // Build an `OR` chain in place of the `IN` operator (which the Cypher dialect // does not yet support). function inClause(field: string, values: number[]): string { @@ -446,20 +452,15 @@ export async function getEdges(args: GetEdgesArgs): Promise { } export async function executeCypherQuery(args: CypherQueryArgs): Promise { - return await fetch(`/api/query?graph=${args.graph}`, { - method: 'POST', - signal: args.signal ?? args.controller?.signal, - headers: { - 'Content-Type': 'text/plain', - }, - body: args.query, - }).then((res) => - res.text().then((text) => { - const json = JSON.parse(text) - if (json.error) { - throw new CypherQueryError(json.error, json.error_details || '') - } - return json.data + try { + return await client.queryData(args.query, { + graph: args.graph, + signal: args.signal ?? args.controller?.signal, }) - ) + } catch (error) { + if (error instanceof TuringQueryError) { + throw new CypherQueryError(error.status, error.message) + } + throw error + } } diff --git a/src/api/responses/cypherQuery.response.ts b/src/api/responses/cypherQuery.response.ts index 93c6d88..88872bb 100644 --- a/src/api/responses/cypherQuery.response.ts +++ b/src/api/responses/cypherQuery.response.ts @@ -2,12 +2,6 @@ // e.g. MATCH (n) RETURN n returns [[0, 1, 2, ...]] - one chunk with one column of node IDs export type CypherQueryResponse = unknown[][] -// Error response from the server -export type CypherQueryErrorResponse = { - error: string - error_details: string -} - // Custom error class to hold both error type and details export class CypherQueryError extends Error { errorType: string diff --git a/tests/cypher-query.spec.ts b/tests/cypher-query.spec.ts index dc3fca7..55526da 100644 --- a/tests/cypher-query.spec.ts +++ b/tests/cypher-query.spec.ts @@ -1,5 +1,5 @@ import { test, expect } from '@playwright/test' -import { setupLoadedGraph } from './helpers' +import { fulfillError, fulfillRows, queryOf, setupLoadedGraph } from './helpers' test.describe('Cypher query execution', () => { test('Ctrl+Enter executes the query', async ({ page }) => { @@ -7,8 +7,8 @@ test.describe('Cypher query execution', () => { await setupLoadedGraph(page, { cypherHandler: async (route) => { - querySent = await route.request().postData() ?? '' - await route.fulfill({ json: { data: [[[1, 2]]] } }) + querySent = queryOf(route) + await fulfillRows(route, [[1, 2]]) }, }) @@ -25,7 +25,7 @@ test.describe('Cypher query execution', () => { await setupLoadedGraph(page, { cypherHandler: async (route) => { queryCalled = true - await route.fulfill({ json: { data: [[[1]]] } }) + await fulfillRows(route, [[1]]) }, }) @@ -40,12 +40,7 @@ test.describe('Cypher query execution', () => { test('query error is displayed and dismissible', async ({ page }) => { await setupLoadedGraph(page, { cypherHandler: async (route) => { - await route.fulfill({ - json: { - error: 'SyntaxError', - error_details: 'Unexpected token at position 5', - }, - }) + await fulfillError(route, 'PARSE_ERROR', 'Unexpected token at position 5') }, }) @@ -56,7 +51,7 @@ test.describe('Cypher query execution', () => { await input.press('Control+Enter') // Error card should appear - await expect(page.getByText('SyntaxError')).toBeVisible() + await expect(page.getByText('PARSE_ERROR')).toBeVisible() await expect(page.getByText('Unexpected token at position 5')).toBeVisible() // Dismiss the error @@ -69,7 +64,7 @@ test.describe('Cypher query execution', () => { cypherHandler: async (route) => { // Delay response to keep pending state visible await new Promise((r) => setTimeout(r, 3000)) - await route.fulfill({ json: { data: [[[1]]] } }) + await fulfillRows(route, [[1]]) }, }) diff --git a/tests/helpers.ts b/tests/helpers.ts index 2a24df3..f7ed3b5 100644 --- a/tests/helpers.ts +++ b/tests/helpers.ts @@ -1,4 +1,5 @@ import type { Page, Route } from '@playwright/test' +import { type Cell, type QueryStatus, decodeRequestQuery, protoError, protoRows } from './proto' // The whole app now talks to a single backend endpoint, `/api/query`: catalog // and data access go through Cypher statements (`LIST AVAILABLE GRAPHS`, @@ -7,10 +8,10 @@ import type { Page, Route } from '@playwright/test' // serves the app's own statements as columnar fixture data and hands genuine // user queries to an optional `cypherHandler`. // -// TuringDB's columnar response shape is `{ data: [chunk, ...] }` where each -// chunk is `[column, ...]` and each column is a flat array of that column's -// cell values — i.e. `{ data: [[col0, col1, ...]] }` for a single chunk. The -// api.ts reshapers read the columns positionally, matching each `YIELD` clause. +// The app's `turingdb` client speaks TuringDB's binary protocol, so the mock +// answers with protocol packets (see proto.ts). A response is one chunk of +// columns, each a flat array of that column's cell values. The api.ts +// reshapers read the columns positionally, matching each `YIELD` clause. // Fixture: 3 nodes (Alice/Bob/Acme) and 2 edges (Alice-KNOWS->Bob, // Bob-WORKS_AT->Acme), mirroring tests/integration/fixture.jsonl in spirit. @@ -34,16 +35,14 @@ const IN_EDGES_IDS = [[], [[100, 1]], [[101, 2]]] const OUT_EDGE_COUNTS = ['{"0":1}', '{"1":1}', '{}'] const IN_EDGE_COUNTS = ['{}', '{"0":1}', '{"1":1}'] -/** One-chunk response from its column arrays. */ -function chunk(columns: unknown[][]) { - return { data: [columns] } -} +/** The columns of the one chunk that a mocked response carries. */ +const chunk = (columns: Cell[][]) => columns /** * Map one of the app's own statements to its columnar response, or return null * if `body` isn't a statement the app itself issues (i.e. it's a user query). */ -function internalQueryResponse(body: string, graphs: string[], loaded: boolean) { +function internalQueryResponse(body: string, graphs: string[], loaded: boolean): Cell[][] | null { // Catalog statements. Check the more specific "AVAILABLE" form first. if (body.includes('LIST AVAILABLE GRAPHS')) { return chunk([graphs, graphs.map(() => loaded), graphs.map(() => false)]) @@ -103,11 +102,11 @@ export async function mockApi(page: Page, opts: MockApiOpts = {}) { const loaded = opts.loaded ?? false await page.route('**/api/query*', async (route) => { - const body = route.request().postData() ?? '' + const body = queryOf(route) const internal = internalQueryResponse(body, graphs, loaded) if (internal !== null) { - await route.fulfill({ json: internal }) + await fulfillRows(route, internal) return } @@ -116,7 +115,25 @@ export async function mockApi(page: Page, opts: MockApiOpts = {}) { await opts.cypherHandler(route) return } - await route.fulfill({ json: { data: [] } }) + await fulfillRows(route, [[]]) + }) +} + +/** The Cypher text the app sent in this request. */ +export function queryOf(route: Route): string { + return decodeRequestQuery(route.request().postDataBuffer()) +} + +/** Answer with one chunk of columns, as a server does for a successful query. */ +export async function fulfillRows(route: Route, columns: Cell[][]) { + await route.fulfill({ contentType: 'application/turing-proto', body: protoRows(columns) }) +} + +/** Answer with a failed query: the app sees this status and message. */ +export async function fulfillError(route: Route, status: QueryStatus, message: string) { + await route.fulfill({ + contentType: 'application/turing-proto', + body: protoError(status, message), }) } diff --git a/tests/integration/README.md b/tests/integration/README.md index 48dce68..e03b972 100644 --- a/tests/integration/README.md +++ b/tests/integration/README.md @@ -37,14 +37,16 @@ change is blocked on a turingdb release. ## How it works -- `global-setup.ts` — boots `turingdb` (via `uvx turingdb@latest`, daemonized) - on a throwaway temp dir on port `6699`, waits for readiness, then imports the +- `global-setup.ts` — boots `turingdb` (via `uvx turingdb@latest`, daemonized, + with `USE_TURING_PROTO=1` for the binary protocol that the app's `turingdb` + npm client speaks) on a throwaway temp dir on port `6699`, waits for + readiness, then imports the fixture via `LOAD JSONL 'fixture.jsonl' AS vis_test`. Returns a teardown that stops the server and removes the temp dir. - `fixture.jsonl` — the deterministic fixture graph (`vis_test`: 4 nodes, 3 edges) in Neo4j APOC JSONL format. Edit this file to change the seed data. -- `setup-fetch.ts` — shims `fetch` so the relative `/api/*` URLs in `api.ts` - resolve to the test server, reproducing the Vite/Express dev proxy's +- `setup-fetch.ts` — shims `fetch` so the relative `/api/query` URL of the + client in `api.ts` resolves to the test server, reproducing the Vite/Express dev proxy's `/api`-strip. - `config.ts` — shared port / base URL / graph name. diff --git a/tests/integration/global-setup.ts b/tests/integration/global-setup.ts index 04c1d36..2a1989b 100644 --- a/tests/integration/global-setup.ts +++ b/tests/integration/global-setup.ts @@ -10,6 +10,9 @@ // TURINGDB_SPEC defaults to `turingdb@latest` (latest published, no lock). // Either way the only host requirement is uv (or the local binary). // +// The server runs with USE_TURING_PROTO=1, the binary protocol the app's +// `turingdb` npm client speaks. +// // We run the server with `-demon` (daemonized): in the foreground it shuts // down as soon as its stdin hits EOF, which happens immediately under a // non-interactive spawn. The daemon is stopped on teardown with @@ -19,6 +22,7 @@ import { copyFileSync, mkdirSync, mkdtempSync, rmSync } from 'node:fs' import { tmpdir } from 'node:os' import path from 'node:path' import { fileURLToPath } from 'node:url' +import { TuringClient } from 'turingdb' import { API_BASE, TEST_GRAPH, TEST_PORT } from './config' const here = path.dirname(fileURLToPath(import.meta.url)) @@ -31,7 +35,11 @@ const TURINGDB_SPEC = process.env.TURINGDB_SPEC ?? 'turingdb@latest' /** Run a child process to completion, rejecting on a non-zero exit. */ function run(cmd: string, args: string[]): Promise { return new Promise((resolve, reject) => { - const child = spawn(cmd, args, { cwd: here, stdio: ['ignore', 'pipe', 'pipe'] }) + const child = spawn(cmd, args, { + cwd: here, + env: { ...process.env, USE_TURING_PROTO: '1' }, + stdio: ['ignore', 'pipe', 'pipe'], + }) let out = '' child.stdout?.on('data', (d) => (out += d)) child.stderr?.on('data', (d) => (out += d)) @@ -54,7 +62,7 @@ async function waitForServer(timeoutMs = 30_000): Promise { const deadline = Date.now() + timeoutMs while (Date.now() < deadline) { try { - await fetch(`${API_BASE}/query?graph=`, { method: 'POST', body: 'LIST GRAPH' }) + await fetch(`${API_BASE}/query?graph=`, { method: 'POST' }) return } catch { await new Promise((r) => setTimeout(r, 300)) @@ -69,15 +77,10 @@ async function waitForServer(timeoutMs = 30_000): Promise { * shot — no change/commit workflow needed. */ async function loadFixture(): Promise { - const res = await fetch(`${API_BASE}/query?graph=`, { - method: 'POST', - headers: { 'Content-Type': 'text/plain' }, - body: `LOAD JSONL '${FIXTURE}' AS ${TEST_GRAPH}`, - }) - const json = (await res.json()) as { error?: string; error_details?: string } - if (json.error) { - throw new Error(`LOAD JSONL failed: ${json.error} ${json.error_details ?? ''}`) - } + // Global setup runs outside the workers, so setup-fetch.ts's `/api` rewrite + // does not apply here: give the client the server's own URL. + const client = new TuringClient({ url: `${API_BASE}/query` }) + await client.queryData(`LOAD JSONL '${FIXTURE}' AS ${TEST_GRAPH}`, { graph: '' }) } export default async function setup() { diff --git a/tests/proto.ts b/tests/proto.ts new file mode 100644 index 0000000..c4f7949 --- /dev/null +++ b/tests/proto.ts @@ -0,0 +1,191 @@ +// Test-side encoder for TuringDB's binary protocol, so the Playwright mocks can +// answer the app's `turingdb` client the way a server would. It covers the +// column kinds the app reads: integers, strings, booleans and lists of those. +// +// A response is [type u8][dataLen u32le][payload] packets: CHUNK_HEADER (the +// columns), CHUNK (the rows), END_CHUNK (the row count), END (exec time), or +// ERROR (status byte + message) followed by END. + +const MESSAGE_CHUNK_HEADER = 0 +const MESSAGE_CHUNK = 1 +const MESSAGE_END_CHUNK = 2 +const MESSAGE_END = 3 +const MESSAGE_ERROR = 4 + +const TYPE_INT64 = 1 +const TYPE_STRING = 3 +const TYPE_BOOL = 4 +const TYPE_LIST_VIEW = 6 +const ENCODING_VECTOR = 0 + +const TAG_INT = 0 +const TAG_BOOL = 3 +const TAG_STRING = 4 +const TAG_LIST_VIEW = 6 + +// Index = the status byte of an ERROR packet (base/QueryStatus.h). +const QUERY_STATUSES = [ + 'OK', + 'GRAPH_NOT_FOUND', + 'PARSE_ERROR', + 'ANALYZE_ERROR', + 'PLAN_ERROR', + 'EXEC_ERROR', + 'COMMIT_NOT_FOUND', + 'COMMIT_NOT_LOADED', + 'CHANGE_NOT_FOUND', +] as const + +export type QueryStatus = (typeof QUERY_STATUSES)[number] +export type Cell = number | string | boolean | Cell[] + +const utf8 = new TextEncoder() + +class Writer { + private bytes: number[] = [] + + u8(value: number) { + this.bytes.push(value & 0xff) + } + + u32(value: number) { + const view = new DataView(new ArrayBuffer(4)) + view.setUint32(0, value, true) + this.raw(new Uint8Array(view.buffer)) + } + + i64(value: number) { + const view = new DataView(new ArrayBuffer(8)) + view.setBigInt64(0, BigInt(value), true) + this.raw(new Uint8Array(view.buffer)) + } + + f32(value: number) { + const view = new DataView(new ArrayBuffer(4)) + view.setFloat32(0, value, true) + this.raw(new Uint8Array(view.buffer)) + } + + string(text: string) { + const encoded = utf8.encode(text) + this.u32(encoded.length) + this.raw(encoded) + } + + raw(bytes: Uint8Array) { + for (const byte of bytes) this.bytes.push(byte) + } + + packet(type: number): Uint8Array { + const out = new Writer() + out.u8(type) + out.u32(this.bytes.length) + out.raw(Uint8Array.from(this.bytes)) + return Uint8Array.from(out.bytes) + } +} + +function columnType(column: Cell[]): number { + const first = column[0] + if (typeof first === 'string') return TYPE_STRING + if (typeof first === 'boolean') return TYPE_BOOL + if (Array.isArray(first)) return TYPE_LIST_VIEW + return TYPE_INT64 +} + +// The server's in-memory size of a list's elements, which the wire carries +// ahead of them: 16 per string or nested list, 8 per integer, 1 per boolean. +function listByteSize(list: Cell[]): number { + return list.reduce((size, cell) => { + if (typeof cell === 'boolean') return size + 1 + if (typeof cell === 'number') return size + 8 + return size + 16 + }, 0) +} + +function writeList(out: Writer, list: Cell[]) { + out.u32(list.length) + out.u32(listByteSize(list)) + for (const cell of list) { + if (typeof cell === 'string') { + out.u8(TAG_STRING) + out.string(cell) + } else if (typeof cell === 'boolean') { + out.u8(TAG_BOOL) + out.u8(cell ? 1 : 0) + } else if (typeof cell === 'number') { + out.u8(TAG_INT) + out.i64(cell) + } else { + out.u8(TAG_LIST_VIEW) + writeList(out, cell) + } + } +} + +function writeColumn(out: Writer, type: number, column: Cell[]) { + out.u32(column.length) + for (const cell of column) { + if (type === TYPE_STRING) out.string(cell as string) + else if (type === TYPE_BOOL) out.u8(cell ? 1 : 0) + else if (type === TYPE_LIST_VIEW) writeList(out, cell as Cell[]) + else out.i64(cell as number) + } +} + +function concat(packets: Uint8Array[]): Buffer { + return Buffer.concat(packets) +} + +function endPacket(): Uint8Array { + const end = new Writer() + end.f32(1) + return end.packet(MESSAGE_END) +} + +/** + * A one-chunk response from its columns, each a list of row values: what the + * client's queryData returns as `[columns]`. Every column must have the same + * row count. + */ +export function protoRows(columns: Cell[][]): Buffer { + const types = columns.map(columnType) + + const header = new Writer() + header.u32(columns.length) + columns.forEach((_, index) => { + const name = utf8.encode(`col${index}`) + header.u32(name.length) + header.u32(types[index]) + header.u8(ENCODING_VECTOR) + header.raw(name) + }) + + const chunk = new Writer() + columns.forEach((column, index) => writeColumn(chunk, types[index], column)) + + const footer = new Writer() + footer.u32(columns[0]?.length ?? 0) + + return concat([ + header.packet(MESSAGE_CHUNK_HEADER), + chunk.packet(MESSAGE_CHUNK), + footer.packet(MESSAGE_END_CHUNK), + endPacket(), + ]) +} + +/** A failed query: the client rejects with this status and message. */ +export function protoError(status: QueryStatus, message: string): Buffer { + const error = new Writer() + error.u8(QUERY_STATUSES.indexOf(status)) + error.raw(utf8.encode(message)) + return concat([error.packet(MESSAGE_ERROR), endPacket()]) +} + +/** The Cypher text out of a request body: [queryLen u32le][query][paramLen u32le][params]. */ +export function decodeRequestQuery(body: Buffer | null): string { + if (!body || body.length < 4) return '' + const length = body.readUInt32LE(0) + return body.subarray(4, 4 + length).toString('utf8') +} diff --git a/vite.config.ts b/vite.config.ts index 1729e6f..1127ffd 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -18,6 +18,11 @@ export default defineConfig({ '@components': path.resolve(__dirname, './src/components'), }, }, + // The turingdb client finds its decoder with new URL('turing_wasm_decoder.wasm', + // import.meta.url). Pre-bundling would move the module away from the .wasm file. + optimizeDeps: { + exclude: ['turingdb'], + }, server: { allowedHosts: true, port: frontendPort,