|
| 1 | +/** |
| 2 | + * Async flatted JSON parsing using stream-json. |
| 3 | + * |
| 4 | + * For large execution data, synchronous flatted.parse() blocks the event loop. |
| 5 | + * This module provides an async alternative that streams the JSON in 64KB |
| 6 | + * chunks, yielding to the event loop between chunks via setImmediate. |
| 7 | + */ |
| 8 | +import { ensureError } from 'n8n-workflow'; |
| 9 | +import { Readable } from 'stream'; |
| 10 | +import { parser } from 'stream-json'; |
| 11 | +import Asm from 'stream-json/Assembler'; |
| 12 | + |
| 13 | +// 64 KB chunks — selected via benchmarks as the sweet spot for event-loop |
| 14 | +// responsiveness vs scheduling overhead (p95 lag ~4 ms at 50 MB payload). |
| 15 | +const CHUNK_SIZE = 64 * 1024; |
| 16 | + |
| 17 | +//#region Flatted reference resolution (extracted from flatted@3.2.7) |
| 18 | + |
| 19 | +const Primitive = String; |
| 20 | +const primitive = 'string'; |
| 21 | +const object = 'object'; |
| 22 | +const ignore = {}; |
| 23 | +const noop = (_: string, value: unknown) => value; |
| 24 | + |
| 25 | +const primitives = (value: unknown) => |
| 26 | + value instanceof Primitive ? Primitive(value as string) : value; |
| 27 | + |
| 28 | +const revive = ( |
| 29 | + input: unknown[], |
| 30 | + parsed: Set<unknown>, |
| 31 | + output: Record<string, unknown>, |
| 32 | + $: (key: string, value: unknown) => unknown, |
| 33 | +): Record<string, unknown> => { |
| 34 | + const lazy: Array<{ |
| 35 | + k: string; |
| 36 | + a: [unknown[], Set<unknown>, Record<string, unknown>, typeof $]; |
| 37 | + }> = []; |
| 38 | + const ke = Object.keys(output); |
| 39 | + for (let y = 0; y < ke.length; y++) { |
| 40 | + const k = ke[y]; |
| 41 | + const value = output[k]; |
| 42 | + if (value instanceof Primitive) { |
| 43 | + const tmp = (input as unknown as Record<string, unknown>)[value as unknown as string]; |
| 44 | + if (typeof tmp === object && !parsed.has(tmp)) { |
| 45 | + parsed.add(tmp); |
| 46 | + output[k] = ignore; |
| 47 | + lazy.push({ k, a: [input, parsed, tmp as Record<string, unknown>, $] }); |
| 48 | + } else { |
| 49 | + output[k] = $.call(output, k, tmp); |
| 50 | + } |
| 51 | + } else if (output[k] !== ignore) { |
| 52 | + output[k] = $.call(output, k, value); |
| 53 | + } |
| 54 | + } |
| 55 | + for (let i = 0; i < lazy.length; i++) { |
| 56 | + const { k, a } = lazy[i]; |
| 57 | + output[k] = $.call(output, k, revive(...a)); |
| 58 | + } |
| 59 | + return output; |
| 60 | +}; |
| 61 | + |
| 62 | +/** |
| 63 | + * Resolve flatted references on a pre-parsed array. |
| 64 | + * This reconstructs the original object graph including circular references. |
| 65 | + */ |
| 66 | +function resolveFlatted(rawArray: unknown[]): unknown { |
| 67 | + const input = rawArray; |
| 68 | + const value = input[0]; |
| 69 | + const $ = noop; |
| 70 | + const tmp = |
| 71 | + typeof value === object && value |
| 72 | + ? revive(input, new Set(), value as Record<string, unknown>, $) |
| 73 | + : value; |
| 74 | + return $.call({ '': tmp }, '', tmp); |
| 75 | +} |
| 76 | + |
| 77 | +//#endregion Flatted reference resolution |
| 78 | + |
| 79 | +/** |
| 80 | + * Recursively convert every string value into a String object wrapper. |
| 81 | + * This replicates what JSON.parse(text, Primitives) does so that flatted |
| 82 | + * can distinguish index references from literal string values. |
| 83 | + */ |
| 84 | +function applyPrimitivesDeep(value: unknown): unknown { |
| 85 | + if (typeof value === primitive) { |
| 86 | + return new Primitive(value); |
| 87 | + } |
| 88 | + if (Array.isArray(value)) { |
| 89 | + for (let i = 0; i < value.length; i++) { |
| 90 | + value[i] = applyPrimitivesDeep(value[i]); |
| 91 | + } |
| 92 | + return value; |
| 93 | + } |
| 94 | + if (typeof value === object && value !== null) { |
| 95 | + const ke = Object.keys(value as Record<string, unknown>); |
| 96 | + for (let i = 0; i < ke.length; i++) { |
| 97 | + const k = ke[i]; |
| 98 | + (value as Record<string, unknown>)[k] = applyPrimitivesDeep( |
| 99 | + (value as Record<string, unknown>)[k], |
| 100 | + ); |
| 101 | + } |
| 102 | + return value; |
| 103 | + } |
| 104 | + return value; |
| 105 | +} |
| 106 | + |
| 107 | +/** |
| 108 | + * Prepare raw parsed JSON for flatted reference resolution. |
| 109 | + * Applies the Primitives transformation then unwraps String objects. |
| 110 | + */ |
| 111 | +function prepareFlatted(rawParsed: unknown[]): unknown[] { |
| 112 | + return (applyPrimitivesDeep(rawParsed) as unknown[]).map(primitives); |
| 113 | +} |
| 114 | + |
| 115 | +/** |
| 116 | + * Parse a flatted JSON string asynchronously using stream-json. |
| 117 | + * Streams the string in 64KB chunks, yielding to the event loop between chunks |
| 118 | + * via setImmediate, then resolves flatted references synchronously. |
| 119 | + */ |
| 120 | +export async function parseFlattedAsync(flattedString: string): Promise<unknown> { |
| 121 | + return await new Promise((resolve, reject) => { |
| 122 | + let offset = 0; |
| 123 | + const readable = new Readable({ |
| 124 | + read() { |
| 125 | + if (offset >= flattedString.length) { |
| 126 | + this.push(null); |
| 127 | + return; |
| 128 | + } |
| 129 | + const chunk = flattedString.slice(offset, offset + CHUNK_SIZE); |
| 130 | + offset += CHUNK_SIZE; |
| 131 | + setImmediate(() => { |
| 132 | + this.push(chunk); |
| 133 | + }); |
| 134 | + }, |
| 135 | + }); |
| 136 | + |
| 137 | + const jsonParser = parser(); |
| 138 | + const asm = Asm.connectTo(jsonParser); |
| 139 | + |
| 140 | + asm.on('done', (asmResult: { current: unknown[] }) => { |
| 141 | + try { |
| 142 | + const prepared = prepareFlatted(asmResult.current); |
| 143 | + resolve(resolveFlatted(prepared)); |
| 144 | + } catch (e) { |
| 145 | + reject(ensureError(e)); |
| 146 | + } |
| 147 | + }); |
| 148 | + |
| 149 | + jsonParser.on('error', reject); |
| 150 | + readable.on('error', reject); |
| 151 | + |
| 152 | + readable.pipe(jsonParser); |
| 153 | + }); |
| 154 | +} |
0 commit comments