perf(neo4j): stream Cypher snapshots to disk

src/build/neo4j/cypher.ts

@@ -7,30 +7,47 @@
  * it expresses the full truth. Incremental updates are the bolt writer's job.
  */
 
+import * as fs from "node:fs";
 import type { EdgeRow, GraphRows, NodeRow, Props } from "./rows";
-import { chunk, cypherMap, cypherValue } from "./rows";
+import { cypherMap, cypherValue } from "./rows";
 import { CONSTRAINTS, INDEXES } from "./schema";
 
 const BATCH = 500;
 
 export function renderCypher(rows: GraphRows, appId: string): string {
-  const out: string[] = [];
+  return [...cypherBlocks(rows, appId)].join("\n");
+}
 
-  out.push("// ── constraints & indexes ──");
-  for (const stmt of CONSTRAINTS) out.push(`${stmt};`);
-  for (const stmt of INDEXES) out.push(`${stmt};`);
+export function writeCypherFile(filePath: string, rows: GraphRows, appId: string): void {
+  const file = fs.openSync(filePath, "w");
+  try {
+    let first = true;
+    for (const block of cypherBlocks(rows, appId)) {
+      fs.writeFileSync(file, first ? block : `\n${block}`);
+      first = false;
+    }
+  } finally {
+    fs.closeSync(file);
+  }
+}
 
-  out.push("", "// ── wipe this project's prior subgraph (external targets are shared) ──");
-  out.push(wipe(appId));
+function* cypherBlocks(rows: GraphRows, appId: string): Generator<string> {
+  yield "// ── constraints & indexes ──";
+  for (const stmt of CONSTRAINTS) yield `${stmt};`;
+  for (const stmt of INDEXES) yield `${stmt};`;
 
-  out.push("", "// ── nodes ──");
-  for (const block of nodeStatements(rows.nodes)) out.push(block);
+  yield "";
+  yield "// ── wipe this project's prior subgraph (external targets are shared) ──";
+  yield wipe(appId);
 
-  out.push("", "// ── relationships ──");
-  for (const block of edgeStatements(rows.edges)) out.push(block);
+  yield "";
+  yield "// ── nodes ──";
+  yield* nodeStatements(rows.nodes);
 
-  out.push("");
-  return out.join("\n");
+  yield "";
+  yield "// ── relationships ──";
+  yield* edgeStatements(rows.edges);
+  yield "";
 }
 
 function wipe(appId: string): string {
@@ -47,63 +64,65 @@ function wipe(appId: string): string {
 // Nodes — grouped by their full label set + key property, batched into UNWIND lists.
 // ----------------------------------------------------------------------------------------------
 
-function nodeStatements(nodes: NodeRow[]): string[] {
+function* nodeStatements(nodes: NodeRow[]): Generator<string> {
   const groups = new Map<string, NodeRow[]>();
   for (const n of nodes) {
     const k = `${n.labels.join(":")}|${n.keyProp}`;
     (groups.get(k) ?? groups.set(k, []).get(k)!).push(n);
   }
 
-  const blocks: string[] = [];
   for (const group of groups.values()) {
     const { labels, keyProp } = group[0];
     const mergeLabel = labels[0];
     const extra = labels.slice(1);
     const setLabels = extra.length ? `, n:${extra.join(":")}` : "";
-    for (const batch of chunk(group, BATCH)) {
+    for (const batch of batches(group, BATCH)) {
       const list = batch
         .map((n) => `  {k: ${cypherValue(n.value)}, p: ${cypherMap(n.props)}}`)
         .join(",\n");
-      blocks.push(
+      yield (
         `UNWIND [\n${list}\n] AS row\n` +
-          `MERGE (n:${mergeLabel} {${keyProp}: row.k})\n` +
-          `SET n += row.p${setLabels};`,
+        `MERGE (n:${mergeLabel} {${keyProp}: row.k})\n` +
+        `SET n += row.p${setLabels};`
       );
     }
   }
-  return blocks;
 }
 
 // ----------------------------------------------------------------------------------------------
 // Edges — grouped by (type, endpoint labels + key props), batched.
 // ----------------------------------------------------------------------------------------------
 
-function edgeStatements(edges: EdgeRow[]): string[] {
+function* edgeStatements(edges: EdgeRow[]): Generator<string> {
   const groups = new Map<string, EdgeRow[]>();
   for (const e of edges) {
     const k = `${e.type}|${e.from.label}.${e.from.keyProp}|${e.to.label}.${e.to.keyProp}|${e.key !== undefined}`;
     (groups.get(k) ?? groups.set(k, []).get(k)!).push(e);
   }
 
-  const blocks: string[] = [];
   for (const group of groups.values()) {
     const { type, from, to } = group[0];
     // Discriminated relationships MERGE on `{_k}` — see EdgeRow.key (issue #70).
     const keyed = group[0].key !== undefined;
-    for (const batch of chunk(group, BATCH)) {
+    for (const batch of batches(group, BATCH)) {
       const list = batch
         .map((e) => `  {f: ${cypherValue(e.from.value)}, t: ${cypherValue(e.to.value)}, ${keyed ? `k: ${cypherValue(e.key!)}, ` : ""}p: ${cypherMap(e.props)}}`)
         .join(",\n");
-      blocks.push(
+      yield (
         `UNWIND [\n${list}\n] AS row\n` +
-          `MATCH (a:${from.label} {${from.keyProp}: row.f})\n` +
-          `MATCH (b:${to.label} {${to.keyProp}: row.t})\n` +
-          `MERGE (a)-[r:${type}${keyed ? " {_k: row.k}" : ""}]->(b)\n` +
-          `SET r += row.p;`,
+        `MATCH (a:${from.label} {${from.keyProp}: row.f})\n` +
+        `MATCH (b:${to.label} {${to.keyProp}: row.t})\n` +
+        `MERGE (a)-[r:${type}${keyed ? " {_k: row.k}" : ""}]->(b)\n` +
+        `SET r += row.p;`
       );
     }
   }
-  return blocks;
+}
+
+function* batches<T>(items: T[], size: number): Generator<T[]> {
+  for (let offset = 0; offset < items.length; offset += size) {
+    yield items.slice(offset, offset + size);
+  }
 }
 
 // Re-exported for the bolt writer (which batches the same rows but binds them as params).

src/build/neo4j/index.ts

@@ -1,7 +1,7 @@
 // Neo4j output: pure projection of the TSApplication IR to graph rows, plus the two writers
 // (cypher snapshot / bolt incremental). Nothing here runs unless `--emit neo4j` is selected.
 export { project } from "./project";
-export { renderCypher } from "./cypher";
+export { renderCypher, writeCypherFile } from "./cypher";
 export { boltWriter, type BoltConfig } from "./bolt";
 export { SCHEMA_VERSION, TS_PREFIX, buildSchemaDocument, NODE_LABELS, REL_TYPES, MARKER_LABELS, CONSTRAINTS, INDEXES } from "./schema";
 export type { SchemaDocument } from "./schema";

src/utils/serialize.ts

@@ -1,6 +1,6 @@
 import * as fs from "node:fs";
 import * as path from "node:path";
-import { boltWriter, buildSchemaDocument, project, renderCypher } from "../build/neo4j";
+import { boltWriter, buildSchemaDocument, project, writeCypherFile } from "../build/neo4j";
 import type { AnalysisOptions } from "../options";
 import type { TSAnalysis } from "../schema";
 import { Logger } from "./logging";
@@ -64,5 +64,5 @@ async function emitNeo4j(application: TSAnalysis, opts: AnalysisOptions): Promis
 
   const dir = opts.output ?? process.cwd();
   fs.mkdirSync(dir, { recursive: true });
-  fs.writeFileSync(path.join(dir, "graph.cypher"), renderCypher(rows, appId));
+  writeCypherFile(path.join(dir, "graph.cypher"), rows, appId);
 }

test/neo4j-schema.test.ts

@@ -16,6 +16,8 @@ import {
   REL_TYPES,
   buildSchemaDocument,
   project,
+  renderCypher,
+  writeCypherFile,
 } from "../src/build/neo4j";
 import { analyze } from "../src/core";
 import type { AnalysisOptions } from "../src/options";
@@ -188,3 +190,17 @@ describe("neo4j inheritance edges (issue #33)", () => {
     expect(nodeValues.has(shape!.value)).toBe(true);
   });
 });
+
+test("streamed Cypher snapshot matches the compatibility renderer byte for byte", () => {
+  const application = rows.nodes.find((node) => node.labels.includes("TSApplication"));
+  expect(application, "TSApplication row").toBeDefined();
+
+  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "cants-cypher-test-"));
+  const output = path.join(dir, "graph.cypher");
+  try {
+    writeCypherFile(output, rows, application!.value);
+    expect(fs.readFileSync(output, "utf8")).toBe(renderCypher(rows, application!.value));
+  } finally {
+    fs.rmSync(dir, { recursive: true, force: true });
+  }
+});