Setting the file. One moment.
Ndjson · Rp Target Wix · wix/skills · Skills Docs
Back to the top of the page 200
async function* readSlice
— line 200
This file
Number 19.39
Position 39 of 44
Type JavaScript
Size 10 KB
Lines 260 lib/ ndjson.js
JavaScript · 260 lines · 10 KB
// product create caps products, variants AND options per request).
15 // - cursor: resume from record N is a byte offset or a line skip, not a full re-parse
16 // - append: a producer can emit records as it finds them; a crash leaves a valid prefix
17 // - a single malformed record is one bad line, not an unparseable file
18 //
19 // USE IT FOR record streams (`data/source-extract/<entity>.ndjson`, crosswalks, audit logs).
20 // Do NOT use it for single documents — a manifest, mapping plan, decisions or completion
21 // report is one object and stays `.json`. Line-delimiting a document buys nothing and makes
22 // it unreadable.
23 //
24 // Dependency-free and streaming, like the source adapters' transport modules. Codegen
25 // vendors a copy into the project (e.g. `src/lib/ndjson.js`) and imports from it.
26
27 const fs = require ( 'node:fs' );
28 const fsp = require ( 'node:fs/promises' );
29 const path = require ( 'node:path' );
30 const readline = require ( 'node:readline' );
31
32 const EXT = '.ndjson' ;
33
34 function ensureDir ( filePath ) {
35 fs. mkdirSync (path. dirname (filePath), { recursive: true });
36 }
37
38 // One record per line. JSON.stringify escapes embedded newlines, so a record can never
39 // break the line framing.
40 function encodeRecord ( record ) {
41 if (record === undefined ) throw new Error ( 'ndjson: cannot encode undefined' );
42 const line = JSON . stringify (record);
43 if (line === undefined ) throw new Error ( 'ndjson: record is not JSON-serialisable' );
44 return `${ line } \n ` ;
45 }
46
47 // --- writing ---------------------------------------------------------------
48
49 // Streaming write of a whole stream. Accepts arrays, generators and async generators, so a
50 // reader can hand over records as it produces them without materialising them all.
51 async function writeRecords ( filePath , records ) {
52 ensureDir (filePath);
53 const out = fs. createWriteStream (filePath, { flags: 'w' });
54 let count = 0 ;
55 try {
56 for await ( const record of records) {
57 count += 1 ;
58 if ( ! out. write ( encodeRecord (record))) {
59 await new Promise (( resolve ) => out. once ( 'drain' , resolve));
60 }
61 }
62 } finally {
63 await new Promise (( resolve , reject ) => {
64 out. end (( err ) => (err ? reject (err) : resolve ()));
65 });
66 }
67 return count;
68 }
69
70 // Incremental appender for producers that emit over time (crosswalk, audit log). `flush`
71 // is per-record and O(1) — the whole point of not rewriting a JSON array each time.
72 function createAppender ( filePath ) {
73 ensureDir (filePath);
74 let count = 0 ;
75 return {
76 file: filePath,
77 append ( record ) {
78 fs. appendFileSync (filePath, encodeRecord (record));
79 count += 1 ;
80 return record;
81 },
82 appendAll ( records ) {
83 let buffer = '' ;
84 for ( const record of records) buffer += encodeRecord (record);
85 if (buffer) fs. appendFileSync (filePath, buffer);
86 count += records. length ;
87 return records. length ;
88 },
89 written : () => count,
90 };
91 }
92
93 // --- reading ---------------------------------------------------------------
94
95 // Streams records without loading the file. readline handles \n, \r\n and the chunk-boundary
96 // case that a naive split('\n') gets wrong.
97 //
98 // A blank line is skipped (a trailing newline is normal). A malformed line throws, naming the
99 // line number — silently dropping records would corrupt a migration invisibly.
100 async function* readRecords ( filePath , { skipMalformed = false , onMalformed = null } = {}) {
101 if ( ! fs. existsSync (filePath)) return ;
102 const input = fs. createReadStream (filePath, { encoding: 'utf8' });
103 const rl = readline. createInterface ({ input, crlfDelay: Infinity });
104 let lineNumber = 0 ;
105 try {
106 for await ( const line of rl) {
107 lineNumber += 1 ;
108 if (line. trim () === '' ) continue ;
109 let record;
110 try {
111 record = JSON . parse (line);
112 } catch (err) {
113 const detail = `${ filePath }:${ lineNumber } is not valid JSON (${ err . message })` ;
114 if (onMalformed) onMalformed ({ file: filePath, line: lineNumber, raw: line, error: err.message });
115 if (skipMalformed) continue ;
116 throw new Error ( `ndjson: ${ detail }. Refusing to continue — a dropped record is invisible data loss.` );
117 }
118 yield record;
119 }
120 } finally {
121 rl. close ();
122 input. destroy ();
123 }
124 }
125
126 // The batching primitive bulk endpoints consume. `size` is a record count; when a target
127 // caps something else too (Wix bulk product create allows 100 products AND 1000 variants
128 // per request) use readBatchesBy instead.
129 async function* readBatches ( filePath , size = 100 , options = {}) {
130 if ( ! Number. isInteger (size) || size < 1 ) throw new Error ( 'ndjson: batch size must be a positive integer' );
131 let batch = [];
132 for await ( const record of readRecords (filePath, options)) {
133 batch. push (record);
134 if (batch. length >= size) {
135 yield batch;
136 batch = [];
137 }
138 }
139 if (batch. length ) yield batch;
140 }
141
142 // Batch against SEVERAL simultaneous caps. Real bulk endpoints rarely have just one: Wix
143 // bulk product create allows 100 products AND 1000 variants AND 100 options AND 100
144 // modifiers AND 100 infoSections per request, and exceeding any single one rejects the whole
145 // batch. Batching on record count alone silently produces requests that 428.
146 //
147 // readBatchesByLimits(file, {
148 // limits: { records: 100, variants: 1000, options: 100 },
149 // cost: (p) => ({ variants: p.variants.length, options: p.options.length }),
150 // })
151 //
152 // `records` is implicit and always counted. Any dimension absent from `limits` is ignored,
153 // so a cost function may report more than the caller caps.
154 //
155 // A single record that alone exceeds a cap is still emitted, in a batch of one, with
156 // `oversized` reported through `onOversized`. Dropping it would be silent data loss and
157 // splitting it is impossible — the caller has to decide (skip it, or fail loudly).
158 async function* readBatchesByLimits ( filePath , { limits = {}, cost = () => ({}), onOversized = null , ... options } = {}) {
159 const caps = { records: 100 , ... limits };
160 for ( const [ dim , cap ] of Object. entries (caps)) {
161 if ( ! (Number. isFinite (cap) && cap >= 1 )) throw new Error ( `ndjson: limit "${ dim }" must be a number >= 1 (got ${ cap })` );
162 }
163
164 let batch = [];
165 let running = {};
166 const costOf = ( record ) => {
167 const c = cost (record) || {};
168 return { ... c, records: 1 };
169 };
170 const wouldExceed = ( c ) => Object. keys (caps). some (( dim ) => (running[dim] || 0 ) + (c[dim] || 0 ) > caps[dim]);
171 const aloneExceeds = ( c ) => Object. keys (caps). some (( dim ) => (c[dim] || 0 ) > caps[dim]);
172
173 for await ( const record of readRecords (filePath, options)) {
174 const c = costOf (record);
175 if (batch. length && wouldExceed (c)) {
176 yield batch;
177 batch = [];
178 running = {};
179 }
180 if ( aloneExceeds (c) && onOversized) {
181 const over = Object. keys (caps). filter (( dim ) => (c[dim] || 0 ) > caps[dim]);
182 onOversized ({ record, cost: c, exceeded: over, caps });
183 }
184 batch. push (record);
185 for ( const dim of Object. keys (c)) running[dim] = (running[dim] || 0 ) + c[dim];
186 }
187 if (batch. length ) yield batch;
188 }
189
190 // Single-dimension convenience wrapper over readBatchesByLimits, kept because plenty of
191 // endpoints really do have just one cap.
192 async function* readBatchesBy ( filePath , { maxCount = 100 , maxCost = Infinity , cost = () => 1 , ... options } = {}) {
193 const limits = { records: maxCount };
194 if (Number. isFinite (maxCost)) limits.cost = maxCost;
195 yield* readBatchesByLimits (filePath, { ... options, limits, cost : ( r ) => ({ cost: cost (r) }) });
196 }
197
198 // Cursor: skip `offset` records, take at most `limit`. Used for resume — no need to parse
199 // what has already been processed into objects.
200 async function* readSlice ( filePath , { offset = 0 , limit = Infinity , ... options } = {}) {
201 let index = 0 ;
202 let taken = 0 ;
203 for await ( const record of readRecords (filePath, options)) {
204 if (index ++ < offset) continue ;
205 if (taken ++ >= limit) return ;
206 yield record;
207 }
208 }
209
210 // Cheap count — parses nothing, just counts non-empty lines.
211 async function countRecords ( filePath ) {
212 if ( ! fs. existsSync (filePath)) return 0 ;
213 const input = fs. createReadStream (filePath, { encoding: 'utf8' });
214 const rl = readline. createInterface ({ input, crlfDelay: Infinity });
215 let count = 0 ;
216 try {
217 for await ( const line of rl) if (line. trim () !== '' ) count += 1 ;
218 } finally {
219 rl. close ();
220 input. destroy ();
221 }
222 return count;
223 }
224
225 // Escape hatch for small streams that genuinely need to be in memory at once (a 6-record
226 // category list). Named so that using it on a large stream is an obvious mistake.
227 async function readAllRecords ( filePath , options = {}) {
228 const out = [];
229 for await ( const record of readRecords (filePath, options)) out. push (record);
230 return out;
231 }
232
233 // --- migration of existing projects ---------------------------------------
234
235 // Converts a legacy `{ entity, recordCount, records: [...] }` document to NDJSON. Kept so a
236 // project generated before this format change can be moved forward without re-extracting.
237 async function convertLegacyJsonFile ( jsonPath , ndjsonPath , { recordsKey = 'records' } = {}) {
238 const parsed = JSON . parse ( await fsp. readFile (jsonPath, 'utf8' ));
239 const records = Array. isArray (parsed) ? parsed : parsed[recordsKey];
240 if ( ! Array. isArray (records)) {
241 throw new Error ( `ndjson: ${ jsonPath } has no array at "${ recordsKey }" to convert` );
242 }
243 const count = await writeRecords (ndjsonPath, records);
244 return { from: jsonPath, to: ndjsonPath, count };
245 }
246
247 module . exports = {
248 EXT,
249 encodeRecord,
250 writeRecords,
251 createAppender,
252 readRecords,
253 readBatches,
254 readBatchesBy,
255 readBatchesByLimits,
256 readSlice,
257 countRecords,
258 readAllRecords,
259 convertLegacyJsonFile,
260 };