Setting the file. One moment.
Shopify Bulk · Rp Source Shopify · wix/skills · Skills Docs
ContentsBack to the top of the page function isStopping
— line 209
This file
Number 38.8
Position 8 of 51
Type JavaScript
Size 75 KB
Lines 1,588 lib/ shopify-bulk.js
JavaScript · 1,588 lines · 75 KB
14 // outcome's handler, then writes the terminal state with the handler's result. On every start each
15 // `settling` entry's handler runs again before anything else, so every handler must be idempotent.
16 //
17 // run() resolves with { status: 'complete' | 'stopped' | 'awaiting_user_action', code?, message?,
18 // details?, manifest }. Anything else it throws is a crash: the manifest on disk is the recovery
19 // point, and a fresh runner over the same files resumes it.
20
21 const fs = require ( 'node:fs' );
22 const fsp = require ( 'node:fs/promises' );
23 const http = require ( 'node:http' );
24 const https = require ( 'node:https' );
25 const path = require ( 'node:path' );
26 const readline = require ( 'node:readline' );
27 const { fork } = require ( 'node:child_process' );
28 const { finished } = require ( 'node:stream/promises' );
29 const { buildShopifyCoverageRows , upsertAdapterCoverageRows , writeJsonAtomic } = require ( '../../../lib/source-coverage.js' );
30 const { createUserAction , readUserAction } = require ( '../../../lib/user-actions.js' );
31 const { ShopifyCliError } = require ( './shopify-cli.js' );
32 const { nestedReadHook } = require ( './shopify-nested-reads.js' );
33 const { ShopifyExtractStop } = require ( './import-findings.js' );
34 const session = require ( './shopify-session.js' );
35
36 const QUERIES_DIR = path. join (__dirname, '..' , 'queries' );
37 const MANIFEST_SCHEMA_VERSION = 1 ;
38 const POLL_START_MS = 2000 ;
39 const POLL_CAP_MS = 30000 ;
40 const OVERDUE_MS = 4 * 60 * 60 * 1000 ;
41 const MAX_ATTEMPTS = 1 ;
42 const MAX_DOWNLOAD_TRIES = 3 ;
43 const MAX_LINK_RERUNS = 2 ;
44 const SNAPSHOT_PLACEHOLDER = '{{SNAPSHOT_TIME}}' ;
45 const OVERDUE_NAMESPACE = 'shopify-bulk-overdue' ;
46 const RAW_EXTRACT_DIR = 'data/source-extract' ;
47 const EXTRACT_MANIFEST_FILE = 'extract-manifest.json' ;
48 const SQLITE_WORKER = path. join (__dirname, 'shopify-bulk-sqlite-worker.js' );
49
50 const STATUS_QUERY = 'query RpBulkStatus($id: ID!) { node(id: $id) { ... on BulkOperation { id status errorCode createdAt completedAt objectCount rootObjectCount fileSize url partialDataUrl } } }' ;
51
52 // ---------------------------------------------------------------------------------------------
53 // The query catalog: the queries, in this order, and how each one's child rows fold
54 // ---------------------------------------------------------------------------------------------
55
56 const MEDIA_TYPES = [ 'MediaImage' , 'Video' , 'ExternalVideo' , 'Model3d' , 'GenericFile' ];
57 const DISCOUNT_APPLICATION_TYPES = [ 'DiscountCodeApplication' , 'AutomaticDiscountApplication' , 'ManualDiscountApplication' , 'ScriptDiscountApplication' ];
58
59 function typesTo ( types , field , extra = {}) {
60 return Object. fromEntries (types. map (( type ) => [type, { field, ... extra }]));
61 }
62
63 // `children` maps a child row's `__typename` to the field it folds into on its parent (`into` names
64 // an inline object of the parent that holds the field). `rootFields` and `childFields` are the
65 // fields initialised to [] so an empty connection reads as empty, not absent. A bulk query cannot
66 // hold a connection inside a list field, so no catalog entry has such a child: those connections
67 // are read by lib/shopify-nested-reads.js before the dataset is done.
68 // Item scopes of a discount (the products, variants and collections a customerGets or customerBuys
69 // scope names) are connections under every discount type, and one bulk query allows five
70 // connections. Each `discounts-gets-*` / `discounts-buys-*` query therefore reads one role and one
71 // kind of item, and `joinDiscountFiles` folds its arrays into the discount row at `parent`.
72 const DISCOUNT_ITEM_CHILDREN = Object. freeze ({
73 products: { type: 'Product' , owner: 'DiscountProducts' },
74 productVariants: { type: 'ProductVariant' , owner: 'DiscountProducts' },
75 collections: { type: 'Collection' , owner: 'DiscountCollections' },
76 });
77
78 function discountItemQuery ( name , side , fields ) {
79 return {
80 name,
81 entity: 'discounts' ,
82 rootFields: fields,
83 children: Object. fromEntries (fields. map (( field ) => [ DISCOUNT_ITEM_CHILDREN [field].type, { field }])),
84 supplement: { parent: [ 'discount' , side, 'items' ], fields },
85 };
86 }
87
88 const QUERY_CATALOG = Object. freeze ([
89 { name: 'products' , entity: 'products' , rootFields: [ 'variants' , 'media' , 'metafields' ], children: { ProductVariant: { field: 'variants' }, ... typesTo ( MEDIA_TYPES , 'media' ), Metafield: { field: 'metafields' } } },
90 { name: 'collections' , entity: 'collections' , rootFields: [ 'products' , 'metafields' ], children: { Product: { field: 'products' }, Metafield: { field: 'metafields' } } },
91 { name: 'customers' , entity: 'customers' , rootFields: [ 'metafields' ], children: { Metafield: { field: 'metafields' } } },
92 { name: 'orders-core' , entity: 'orders' , rootFields: [ 'lineItems' , 'shippingLines' , 'discountApplications' , 'metafields' ], children: { LineItem: { field: 'lineItems' }, ShippingLine: { field: 'shippingLines' }, ... typesTo ( DISCOUNT_APPLICATION_TYPES , 'discountApplications' ), Metafield: { field: 'metafields' } } },
93 { name: 'orders-refunds' , entity: 'orders' , children: {} },
94 { name: 'orders-fulfillments' , entity: 'orders' , children: {} },
95 { name: 'gift-cards' , entity: 'gift-cards' , children: {} },
96 { name: 'discounts' , entity: 'discounts' , children: { DiscountRedeemCode: { field: 'codes' , into: 'discount' } } },
97 discountItemQuery ( 'discounts-gets-products' , 'customerGets' , [ 'products' ]),
98 discountItemQuery ( 'discounts-gets-variants' , 'customerGets' , [ 'productVariants' ]),
99 discountItemQuery ( 'discounts-gets-collections' , 'customerGets' , [ 'collections' ]),
100 discountItemQuery ( 'discounts-buys-items' , 'customerBuys' , [ 'products' , 'collections' ]),
101 discountItemQuery ( 'discounts-buys-variants' , 'customerBuys' , [ 'productVariants' ]),
102 { name: 'delivery-profiles' , entity: 'delivery-profiles' , children: {} },
103 { name: 'inventory' , entity: 'inventory-items' , rootFields: [ 'inventoryLevels' ], children: { InventoryLevel: { field: 'inventoryLevels' } } },
104 { name: 'pages' , entity: 'pages' , children: {} },
105 { name: 'blogs' , entity: 'blogs' , children: {} },
106 { name: 'articles' , entity: 'articles' , children: {} },
107 { name: 'menus' , entity: 'menus' , children: {} },
108 { name: 'url-redirects' , entity: 'url-redirects' , children: {} },
109 { name: 'locations' , entity: 'locations' , children: {} },
110 { name: 'metaobjects' , entity: 'metaobjects' , rootFields: [ 'metaobjects' ], children: { Metaobject: { field: 'metaobjects' } } },
111 { name: 'files' , entity: 'files' , children: {} },
112 ]);
113
114 const QUERY_NAMES = Object. freeze ( QUERY_CATALOG . map (( entry ) => entry.name));
115 const CUSTOMER_QUERIES = Object. freeze ([ 'customers' ]);
116 const ORDER_QUERIES = Object. freeze ([ 'orders-core' , 'orders-refunds' , 'orders-fulfillments' ]);
117 const DENIAL_ATTRIBUTED_QUERIES = Object. freeze ([ ... CUSTOMER_QUERIES , ... ORDER_QUERIES ]);
118
119 function catalogEntry ( name ) {
120 const entry = QUERY_CATALOG . find (( candidate ) => candidate.name === name);
121 if ( ! entry) throw new Error ( `unknown Shopify bulk query ${ name }` );
122 return entry;
123 }
124
125 function queryFilePath ( name , dir = QUERIES_DIR ) {
126 return path. join (dir, `${ name }.graphql` );
127 }
128
129 function readQueryFile ( name , dir = QUERIES_DIR ) {
130 return fs. readFileSync ( queryFilePath (name, dir), 'utf8' );
131 }
132
133 // The text sent for one query: the file with the snapshot time filled in, then the caller's
134 // projection (`projectQuery`), if it passes one, for the customer and order queries.
135 function generateQueryText ( name , { snapshotTime , project = null , dir = QUERIES_DIR } = {}) {
136 let text = readQueryFile (name, dir);
137 if (text. includes ( SNAPSHOT_PLACEHOLDER )) {
138 if ( ! snapshotTime) throw new Error ( `query ${ name } needs the snapshot time` );
139 text = text. split ( SNAPSHOT_PLACEHOLDER ). join (snapshotTime);
140 }
141 if ( typeof project === 'function' && DENIAL_ATTRIBUTED_QUERIES . includes (name)) text = project (name, text);
142 return text;
143 }
144
145 // Static shape checks for the contract test: connections are the `edges { node }` blocks.
146 function stripForShape ( text ) {
147 const noComments = String (text). replace ( /# [ ^ \n] * / g , ' ' );
148 let out = '' ;
149 let depth = 0 ;
150 let inString = false ;
151 for ( let i = 0 ; i < noComments. length ; i += 1 ) {
152 const ch = noComments[i];
153 if (inString) {
154 if (ch === ' \\ ' ) i += 1 ;
155 else if (ch === '"' ) inString = false ;
156 continue ;
157 }
158 if (ch === '"' ) { inString = true ; continue ; }
159 if (ch === '(' ) { depth += 1 ; continue ; }
160 if (ch === ')' ) { depth = Math. max ( 0 , depth - 1 ); continue ; }
161 if (depth === 0 ) out += ch;
162 }
163 return out;
164 }
165
166 function connectionShape ( text ) {
167 const tokens = stripForShape (text). match ( / [{}] | [A-Za-z_][A-Za-z0-9_] * / g ) || [];
168 const stack = [];
169 let connections = 0 ;
170 let maxDepth = 0 ;
171 let open = 0 ;
172 for ( let i = 0 ; i < tokens. length ; i += 1 ) {
173 const token = tokens[i];
174 if (token === '{' ) {
175 const isEdges = tokens[i - 1 ] === 'edges' ;
176 stack. push (isEdges);
177 if (isEdges) { open += 1 ; connections += 1 ; maxDepth = Math. max (maxDepth, open); }
178 } else if (token === '}' ) {
179 if (stack. pop ()) open -= 1 ;
180 }
181 }
182 return { connections, nestedLevels: Math. max ( 0 , maxDepth - 1 ) };
183 }
184
185 // ---------------------------------------------------------------------------------------------
186 // Stops
187 // ---------------------------------------------------------------------------------------------
188
189 class BulkStop extends Error {
190 constructor ( code , message , { status = 'stopped' , details = {} } = {}) {
191 super (message);
192 this .name = 'BulkStop' ;
193 this .code = code;
194 this .status = status;
195 this .details = details;
196 }
197 }
198
199 const STOPPING_FAILED_CODE = 'shopify_bulk_failed' ;
200
201 function iso ( ms ) {
202 return new Date (ms). toISOString ();
203 }
204
205 function entityOf ( name ) {
206 return catalogEntry (name).entity;
207 }
208
209 function isStopping ( entry , caller ) {
210 if (caller !== 'extract' ) return false ;
211 if (entry.state === 'failed' ) return true ;
212 return entry.state === 'denied' && entry.handlerOutcome && entry.handlerOutcome.attribution === 'unattributed' ;
213 }
214
215 // ---------------------------------------------------------------------------------------------
216 // Files: JSONL validation, folding, ndjson writing, the default file host download
217 // ---------------------------------------------------------------------------------------------
218
219 async function removeFile ( file ) {
220 await fsp. rm (file, { force: true });
221 }
222
223 async function writeTextAtomic ( file , text ) {
224 await fsp. mkdir (path. dirname (file), { recursive: true });
225 const temp = `${ file }.${ process . pid }.${ Date . now () }.tmp` ;
226 await fsp. writeFile (temp, text, 'utf8' );
227 await fsp. rename (temp, file);
228 }
229
230 async function writeNdjsonAtomic ( file , records ) {
231 await fsp. mkdir (path. dirname (file), { recursive: true });
232 const temp = `${ file }.${ process . pid }.${ Date . now () }.tmp` ;
233 const out = fs. createWriteStream (temp, { encoding: 'utf8' });
234 let count = 0 ;
235 try {
236 for await ( const record of records) {
237 if ( ! out. write ( `${ JSON . stringify ( record ) } \n ` )) await new Promise (( resolve ) => out. once ( 'drain' , resolve));
238 count += 1 ;
239 }
240 out. end ();
241 await finished (out);
242 await fsp. rename (temp, file);
243 return count;
244 } catch (error) {
245 out. destroy ();
246 await removeFile (temp);
247 throw error;
248 }
249 }
250
251 async function* iterateNdjson ( file ) {
252 const input = fs. createReadStream (file, { encoding: 'utf8' });
253 const lines = readline. createInterface ({ input, crlfDelay: Infinity });
254 for await ( const line of lines) if (line. length ) yield JSON . parse (line);
255 }
256
257 async function countLines ( file ) {
258 try {
259 await fsp. access (file);
260 } catch (error) {
261 if (error.code === 'ENOENT' ) return null ;
262 throw error;
263 }
264 let count = 0 ;
265 for await ( const _record of iterateNdjson (file)) count += 1 ;
266 return count;
267 }
268
269 async function copyFileAtomic ( source , destination ) {
270 await fsp. mkdir (path. dirname (destination), { recursive: true });
271 const temp = `${ destination }.${ process . pid }.${ Date . now () }.tmp` ;
272 try {
273 await fsp. copyFile (source, temp);
274 await fsp. rename (temp, destination);
275 } catch (error) {
276 await removeFile (temp);
277 throw error;
278 }
279 }
280
281 // The download gate's file checks: empty or newline-terminated, and every line one
282 // JSON object. Streams the file; keeps no record in memory.
283 function validateJsonlFile ( file ) {
284 return new Promise (( resolve ) => {
285 const stream = fs. createReadStream (file, { encoding: 'utf8' });
286 let rest = '' ;
287 let lines = 0 ;
288 let last = '' ;
289 let failure = null ;
290 const checkLine = ( line ) => {
291 if (failure) return ;
292 let value;
293 try {
294 value = JSON . parse (line);
295 } catch {
296 failure = 'a line does not parse as JSON' ;
297 return ;
298 }
299 if ( ! value || typeof value !== 'object' || Array. isArray (value)) failure = 'a line is not a JSON object' ;
300 else lines += 1 ;
301 };
302 stream. on ( 'data' , ( chunk ) => {
303 if (failure) return ;
304 last = chunk[chunk. length - 1 ];
305 rest += chunk;
306 let at = rest. indexOf ( ' \n ' );
307 while (at !== - 1 && ! failure) {
308 checkLine (rest. slice ( 0 , at));
309 rest = rest. slice (at + 1 );
310 at = rest. indexOf ( ' \n ' );
311 }
312 });
313 stream. on ( 'error' , ( error ) => resolve ({ ok: false , reason: `the file could not be read: ${ error . message }` }));
314 stream. on ( 'end' , () => {
315 if ( ! failure && rest. length ) failure = 'the file does not end with a newline' ;
316 if (failure) resolve ({ ok: false , reason: failure });
317 else resolve ({ ok: true , lines, endsWithNewline: last === ' \n ' || last === '' });
318 });
319 });
320 }
321
322 // Node 22.12 ships node:sqlite behind --experimental-sqlite. Keep the documented adapter
323 // invocation flag-free: on that one supported boundary runtime, only the disk-backed transform
324 // runs in a child with the built-in module enabled. Node 22.13+ stays in-process. The worker gets
325 // only paths and the public query shape, never credentials or record values over the command line.
326 function loadDatabaseSync () {
327 try {
328 return require ( 'node:sqlite' ).DatabaseSync;
329 } catch (error) {
330 if (error && error.code === 'ERR_UNKNOWN_BUILTIN_MODULE' ) return null ;
331 throw error;
332 }
333 }
334
335 function sqliteWorkerError ( payload ) {
336 const error = new Error (payload && payload.message ? payload.message : 'the SQLite worker failed' );
337 if (payload && payload.name) error.name = payload.name;
338 for ( const field of [ 'code' , 'errno' , 'syscall' , 'path' ]) {
339 if (payload && payload[field] !== undefined ) error[field] = payload[field];
340 }
341 if (payload && payload.stack) error.stack = payload.stack;
342 return error;
343 }
344
345 function runSqliteWorker ( operation , payload ) {
346 return new Promise (( resolve , reject ) => {
347 // Preserve explicit memory ceilings used to prove the bounded-memory contract. Other Node
348 // entry-point/debug flags belong to the parent, not to this internal helper.
349 const memoryArgs = process.execArgv. filter (( arg ) => / ^ --(?:max-old-space-size | max-semi-space-size | stack-size)=/ . test (arg));
350 const child = fork ( SQLITE_WORKER , [], {
351 execArgv: [ ... memoryArgs, '--no-warnings' , '--experimental-sqlite' ],
352 stdio: [ 'ignore' , 'inherit' , 'inherit' , 'ipc' ],
353 });
354 let settled = false ;
355 const fail = ( error ) => {
356 if (settled) return ;
357 settled = true ;
358 reject (error);
359 };
360 child. once ( 'error' , fail);
361 child. once ( 'exit' , ( code , signal ) => {
362 if ( ! settled) fail ( new Error ( `the SQLite worker exited before replying (${ signal || `code ${ code }`})` ));
363 });
364 child. once ( 'message' , ( message ) => {
365 if (settled) return ;
366 settled = true ;
367 if (child.connected) child. disconnect ();
368 if (message && message.ok) resolve (message.result);
369 else reject ( sqliteWorkerError (message && message.error));
370 });
371 child. send ({ operation, payload }, ( error ) => {
372 if (error) fail (error);
373 });
374 });
375 }
376
377 // Shopify leaves `groupObjects` false, so children may precede their parents. Index child rows in
378 // a temporary SQLite database, then stream roots in source order and attach only that root's tree.
379 // Memory is therefore bounded by the largest assembled root, not by the full store export.
380 async function foldBulkJsonl ( file , shape , { outputFile } = {}) {
381 if ( ! outputFile) throw new Error ( 'foldBulkJsonl requires outputFile' );
382 const DatabaseSync = loadDatabaseSync ();
383 if ( ! DatabaseSync) return runSqliteWorker ( 'foldBulkJsonl' , { file, shape, outputFile });
384 await fsp. mkdir (path. dirname (outputFile), { recursive: true });
385 const work = `${ outputFile }.${ process . pid }.${ Date . now () }.fold.tmp` ;
386 const outputTemp = `${ outputFile }.${ process . pid }.${ Date . now () }.tmp` ;
387 const db = new DatabaseSync (work);
388 let output = null ;
389 let lines = 0 ;
390 try {
391 db. exec ( `
392 PRAGMA journal_mode = OFF;
393 PRAGMA synchronous = OFF;
394 PRAGMA temp_store = FILE;
395 CREATE TABLE roots (seq INTEGER PRIMARY KEY, json TEXT NOT NULL);
396 CREATE TABLE children (
397 seq INTEGER PRIMARY KEY,
398 parent_id TEXT NOT NULL,
399 child_id TEXT,
400 child_type TEXT,
401 json TEXT NOT NULL,
402 attached INTEGER NOT NULL DEFAULT 0,
403 orphan_reason TEXT
404 );
405 ` );
406 const addRoot = db. prepare ( 'INSERT INTO roots(seq, json) VALUES (?, ?)' );
407 const addChild = db. prepare ( 'INSERT INTO children(seq, parent_id, child_id, child_type, json) VALUES (?, ?, ?, ?, ?)' );
408 db. exec ( 'BEGIN' );
409 let inTransaction = true ;
410 try {
411 for await ( const line of readline. createInterface ({ input: fs. createReadStream (file, { encoding: 'utf8' }), crlfDelay: Infinity })) {
412 if ( ! line. length ) continue ;
413 const row = JSON . parse (line);
414 lines += 1 ;
415 if (row.__parentId === undefined ) addRoot. run (lines, line);
416 else addChild. run (lines, String (row.__parentId), typeof row.id === 'string' ? row.id : null , typeof row.__typename === 'string' ? row.__typename : null , line);
417 }
418 db. exec ( 'COMMIT' );
419 inTransaction = false ;
420 db. exec ( 'CREATE INDEX children_parent ON children(parent_id, seq)' );
421 } catch (error) {
422 if (inTransaction) db. exec ( 'ROLLBACK' );
423 throw error;
424 }
425
426 const childrenFor = db. prepare ( 'SELECT seq, json, child_type FROM children WHERE parent_id = ? AND attached = 0 ORDER BY seq' );
427 const markAttached = db. prepare ( 'UPDATE children SET attached = 1 WHERE seq = ?' );
428 const markOrphan = db. prepare ( 'UPDATE children SET orphan_reason = ? WHERE seq = ?' );
429 // Node 22.12's flag-gated StatementSync has no iterate(). Walk the integer primary key with
430 // get() instead: one root is resident at a time and later runtimes follow the identical path.
431 const nextRoot = db. prepare ( 'SELECT seq, json FROM roots WHERE seq > ? ORDER BY seq LIMIT 1' );
432 const nestedParentTypes = new Set (Object. keys (shape.childFields || {}));
433 output = fs. createWriteStream (outputTemp, { encoding: 'utf8' });
434 let recordCount = 0 ;
435 let rootSeq = 0 ;
436
437 for ( let stored = nextRoot. get (rootSeq); stored; stored = nextRoot. get (rootSeq)) {
438 rootSeq = Number (stored.seq);
439 const root = JSON . parse (stored.json);
440 const seen = new Set ();
441 for ( const field of shape.rootFields || []) if ( ! Array. isArray (root[field])) root[field] = [];
442
443 const attachChildren = ( value , mayHaveChildren ) => {
444 if ( ! value || typeof value !== 'object' ) return ;
445 if (mayHaveChildren && typeof value.id === 'string' && ! seen. has (value.id)) {
446 seen. add (value.id);
447 for ( const storedChild of childrenFor. all (value.id)) {
448 const rule = shape.children[storedChild.child_type];
449 if ( ! rule) { markOrphan. run ( 'unrecognised_child_type' , storedChild.seq); continue ; }
450 const holder = rule.into ? (value[rule.into] && typeof value[rule.into] === 'object' ? value[rule.into] : null ) : value;
451 if ( ! holder) { markOrphan. run ( 'parent_not_found' , storedChild.seq); continue ; }
452 const child = JSON . parse (storedChild.json);
453 delete child.__parentId;
454 if ( ! Array. isArray (holder[rule.field])) holder[rule.field] = [];
455 holder[rule.field]. push (child);
456 markAttached. run (storedChild.seq);
457 attachChildren (child, nestedParentTypes. has (child.__typename));
458 }
459 }
460 };
461 attachChildren (root, true );
462
463 const initialiseTree = ( value ) => {
464 if (Array. isArray (value)) { for ( const item of value) initialiseTree (item); return ; }
465 if ( ! value || typeof value !== 'object' ) return ;
466 const initialise = shape.childFields && shape.childFields[value.__typename];
467 if (initialise) for ( const field of initialise) if ( ! Array. isArray (value[field])) value[field] = [];
468 for ( const nested of Object. values (value)) initialiseTree (nested);
469 };
470 initialiseTree (root);
471 if ( ! output. write ( `${ JSON . stringify ( root ) } \n ` )) await new Promise (( resolve ) => output. once ( 'drain' , resolve));
472 recordCount += 1 ;
473 }
474 output. end ();
475 await finished (output);
476 output = null ;
477
478 const orphanCount = Number (db. prepare ( 'SELECT COUNT(*) AS count FROM children WHERE attached = 0' ). get ().count);
479 if (orphanCount) {
480 const orphans = db. prepare ( 'SELECT parent_id, child_id, child_type, orphan_reason FROM children WHERE attached = 0 ORDER BY seq LIMIT 20' ). all (). map (( row ) => ({
481 reason: row.orphan_reason || 'parent_not_found' ,
482 parentId: row.parent_id,
483 id: row.child_id || null ,
484 type: row.child_type || null ,
485 }));
486 await removeFile (outputTemp);
487 return { orphans, orphanCount, lines };
488 }
489 await fsp. rename (outputTemp, outputFile);
490 return { outputFile, recordCount, lines };
491 } finally {
492 if (output) output. destroy ();
493 db. close ();
494 await removeFile (outputTemp);
495 await removeFile (work);
496 }
497 }
498
499 // Join the three order datasets without loading any of them as a whole. Supplemental rows live
500 // in a temporary SQLite table; core orders stream through one at a time in their original order.
501 async function joinOrderFiles ({ coreFile , refundsFile , fulfillmentsFile , outputFile , workDir = path. dirname (outputFile) }) {
502 const DatabaseSync = loadDatabaseSync ();
503 if ( ! DatabaseSync) return runSqliteWorker ( 'joinOrderFiles' , { coreFile, refundsFile, fulfillmentsFile, outputFile, workDir });
504 const work = path. join (workDir, `${ path . basename ( outputFile ) }.${ process . pid }.${ Date . now () }.orders.tmp` );
505 const outputTemp = path. join (workDir, `${ path . basename ( outputFile ) }.${ process . pid }.${ Date . now () }.tmp` );
506 const db = new DatabaseSync (work);
507 let output = null ;
508 try {
509 db. exec ( `
510 PRAGMA journal_mode = OFF;
511 PRAGMA synchronous = OFF;
512 PRAGMA temp_store = FILE;
513 CREATE TABLE supplements (
514 query_name TEXT NOT NULL,
515 order_id TEXT NOT NULL,
516 value TEXT NOT NULL,
517 used INTEGER NOT NULL DEFAULT 0,
518 PRIMARY KEY(query_name, order_id)
519 );
520 CREATE INDEX supplements_order ON supplements(order_id);
521 ` );
522 const add = db. prepare ( 'INSERT OR REPLACE INTO supplements(query_name, order_id, value) VALUES (?, ?, ?)' );
523 db. exec ( 'BEGIN' );
524 try {
525 for ( const [ query , field , file ] of [[ 'orders-refunds' , 'refunds' , refundsFile], [ 'orders-fulfillments' , 'fulfillments' , fulfillmentsFile]]) {
526 if ( ! file) continue ; // this run does not carry that query
527 // A `done` supplement always has its staged file, empty when the dataset is. A missing one
528 // is lost input, never an empty dataset: fail loudly rather than write empty arrays.
529 try {
530 await fsp. access (file);
531 } catch (error) {
532 if (error.code === 'ENOENT' ) throw new Error ( `the staged ${ query } file is missing: ${ path . basename ( file ) }` );
533 throw error;
534 }
535 for await ( const row of iterateNdjson (file)) add. run (query, String (row.id), JSON . stringify (Array. isArray (row[field]) ? row[field] : []));
536 }
537 db. exec ( 'COMMIT' );
538 } catch (error) {
539 db. exec ( 'ROLLBACK' );
540 throw error;
541 }
542 const find = db. prepare ( 'SELECT query_name, value FROM supplements WHERE order_id = ?' );
543 const markUsed = db. prepare ( 'UPDATE supplements SET used = 1 WHERE order_id = ?' );
544 output = fs. createWriteStream (outputTemp, { encoding: 'utf8' });
545 for await ( const order of iterateNdjson (coreFile)) {
546 const joined = { ... order, refunds: [], fulfillments: [] };
547 for ( const row of find. all ( String (order.id))) joined[row.query_name === 'orders-refunds' ? 'refunds' : 'fulfillments' ] = JSON . parse (row.value);
548 markUsed. run ( String (order.id));
549 if ( ! output. write ( `${ JSON . stringify ( joined ) } \n ` )) await new Promise (( resolve ) => output. once ( 'drain' , resolve));
550 }
551 output. end ();
552 await finished (output);
553 output = null ;
554 await fsp. rename (outputTemp, outputFile);
555 const droppedByQuery = Object. fromEntries ([ 'orders-refunds' , 'orders-fulfillments' ]. map (( query ) => [query, Number (db. prepare ( 'SELECT COUNT(*) AS count FROM supplements WHERE query_name = ? AND used = 0' ). get (query).count)]));
556 return { droppedOrphans: droppedByQuery[ 'orders-refunds' ] + droppedByQuery[ 'orders-fulfillments' ], droppedByQuery };
557 } finally {
558 if (output) output. destroy ();
559 db. close ();
560 await removeFile (outputTemp);
561 await removeFile (work);
562 }
563 }
564
565 // Fold the discount item-scope datasets into the discount rows. A discount's rows are few, so the
566 // supplements are read into memory as ids only; the core rows stream through in their original
567 // order. A field is set only when the core row's own scope kind owns it (products and variants
568 // under DiscountProducts, collections under DiscountCollections), and only when the supplement
569 // listed that discount at all: a discount the supplement never saw keeps the field absent, which
570 // reads as "not read", never as an empty scope.
571 async function joinDiscountFiles ({ coreFile , supplements , outputFile , workDir = path. dirname (outputFile) }) {
572 const outputTemp = path. join (workDir, `${ path . basename ( outputFile ) }.${ process . pid }.${ Date . now () }.tmp` );
573 const loaded = [];
574 for ( const supplement of supplements) {
575 try {
576 await fsp. access (supplement.file);
577 } catch (error) {
578 if (error.code === 'ENOENT' ) throw new Error ( `the staged ${ supplement . query } file is missing: ${ path . basename ( supplement . file ) }` );
579 throw error;
580 }
581 const listed = new Map ();
582 for await ( const row of iterateNdjson (supplement.file)) {
583 const values = {};
584 for ( const field of supplement.fields) if (Array. isArray (row[field]) && row[field]. length ) values[field] = row[field];
585 listed. set ( String (row.id), values);
586 }
587 loaded. push ({ ... supplement, listed });
588 }
589 const coreIds = new Set ();
590 let output = null ;
591 try {
592 output = fs. createWriteStream (outputTemp, { encoding: 'utf8' });
593 for await ( const row of iterateNdjson (coreFile)) {
594 const id = String (row.id);
595 coreIds. add (id);
596 for ( const supplement of loaded) {
597 if ( ! supplement.listed. has (id)) continue ;
598 const values = supplement.listed. get (id);
599 let parent = row;
600 for ( const part of supplement.parent) parent = parent && typeof parent === 'object' ? parent[part] : null ;
601 if ( ! parent || typeof parent !== 'object' ) continue ;
602 for ( const field of supplement.fields) {
603 if (parent.__typename === DISCOUNT_ITEM_CHILDREN [field].owner) parent[field] = values[field] || [];
604 }
605 }
606 if ( ! output. write ( `${ JSON . stringify ( row ) } \n ` )) await new Promise (( resolve ) => output. once ( 'drain' , resolve));
607 }
608 output. end ();
609 await finished (output);
610 output = null ;
611 await fsp. rename (outputTemp, outputFile);
612 const droppedByQuery = {};
613 for ( const supplement of loaded) droppedByQuery[supplement.query] = [ ... supplement.listed. keys ()]. filter (( id ) => ! coreIds. has (id)). length ;
614 return { droppedOrphans: Object. values (droppedByQuery). reduce (( sum , count ) => sum + count, 0 ), droppedByQuery };
615 } finally {
616 if (output) output. destroy ();
617 await removeFile (outputTemp);
618 }
619 }
620
621 // The default file host download: plain http(s), no content decoding (identity only), bytes
622 // counted as they arrive. It reports what happened; the gate decides what it means.
623 function defaultDownload ( url , destination ) {
624 return new Promise (( resolve ) => {
625 let client;
626 try {
627 client = String (url). startsWith ( 'http:' ) ? http : https;
628 } catch {
629 client = https;
630 }
631 const request = client. get (url, { headers: { 'accept-encoding' : 'identity' } }, ( response ) => {
632 if (response.statusCode !== 200 ) {
633 response. resume ();
634 response. on ( 'end' , () => resolve ({ status: response.statusCode, bytes: 0 }));
635 response. on ( 'error' , ( error ) => resolve ({ status: response.statusCode, bytes: 0 , error: error.message }));
636 return ;
637 }
638 const out = fs. createWriteStream (destination);
639 let bytes = 0 ;
640 let failed = null ;
641 response. on ( 'data' , ( chunk ) => { bytes += chunk. length ; });
642 response. on ( 'error' , ( error ) => { failed = error.message; out. end (); });
643 response. on ( 'aborted' , () => { failed = failed || 'the connection was cut' ; });
644 response. pipe (out);
645 out. on ( 'finish' , () => resolve ({ status: 200 , bytes, ... (failed || ! response.complete ? { error: failed || 'the response did not complete' } : {}) }));
646 out. on ( 'error' , ( error ) => resolve ({ status: 200 , bytes, error: error.message }));
647 });
648 request. on ( 'error' , ( error ) => resolve ({ status: null , bytes: 0 , error: error.message }));
649 });
650 }
651
652 // ---------------------------------------------------------------------------------------------
653 // The runner
654 // ---------------------------------------------------------------------------------------------
655
656 function pendingEntry ( query ) {
657 return {
658 state: 'pending' ,
659 entity: query.entity || entityOf (query.name),
660 text: query.text,
661 attempts: 0 ,
662 reapproved: false ,
663 downloadTries: 0 ,
664 linkReruns: 0 ,
665 };
666 }
667
668 function newManifest ({ caller , store , snapshotTime , queries , previousSidecars , now }) {
669 const manifest = {
670 schemaVersion: MANIFEST_SCHEMA_VERSION ,
671 caller,
672 store,
673 snapshotTime,
674 startedAt: iso ( now ()),
675 status: 'running' ,
676 order: queries. map (( query ) => query.name),
677 queries: Object. fromEntries (queries. map (( query ) => [query.name, pendingEntry (query)])),
678 };
679 if (previousSidecars. length ) manifest.previousSidecars = [ ...new Set (previousSidecars)]. sort ();
680 return manifest;
681 }
682
683 const OPERATION_FIELDS = [ 'operationId' , 'runningAt' , 'deadlineAt' , 'overdueRound' , 'overdueActionId' , 'url' , 'fileSize' , 'objectCount' , 'rootObjectCount' , 'createdAt' , 'completedAt' , 'startedAt' ];
684
685 // The shopify extract adapters plug into the runner here and add no runner logic of their own: a denial
686 // handler, a `failed` handler per query (`failed.byQuery`), a reassembly hook per query, the
687 // `denied` reset, and `project` (the projection applied to the customer and order query text).
688 // Everything a handler registers must be idempotent. Options passed to createBulkRunner
689 // override what is registered.
690 const registered = { handlers: {}, deniedReset: null , project: null };
691
692 function registerHandlers ({ denied , failed , reassemble , deniedReset , project } = {}) {
693 if (denied) registered.handlers.denied = denied;
694 if (failed) registered.handlers.failed = { ... (registered.handlers.failed || {}), ... failed, byQuery: { ... ((registered.handlers.failed || {}).byQuery || {}), ... (failed.byQuery || {}) } };
695 if (reassemble) registered.handlers.reassemble = { ... (registered.handlers.reassemble || {}), ... reassemble };
696 if (deniedReset) registered.deniedReset = deniedReset;
697 if (project) registered.project = project;
698 }
699
700 function clearRegisteredHandlers () {
701 registered.handlers = {};
702 registered.deniedReset = null ;
703 registered.project = null ;
704 }
705
706 function mergeHandlers ( extra = {}) {
707 const base = registered.handlers;
708 return {
709 denied: extra.denied || base.denied,
710 failed: { ... (base.failed || {}), ... (extra.failed || {}), byQuery: { ... ((base.failed || {}).byQuery || {}), ... ((extra.failed || {}).byQuery || {}) } },
711 reassemble: { ... (base.reassemble || {}), ... (extra.reassemble || {}) },
712 };
713 }
714
715 function createBulkRunner ( options = {}) {
716 const {
717 projectDir ,
718 store ,
719 caller ,
720 manifestPath ,
721 stagingDir ,
722 queries = null ,
723 addQueries = null ,
724 currentText = null ,
725 snapshotTime = null ,
726 execute ,
727 handlers : handlerOptions = {},
728 deniedReset = registered.deniedReset,
729 retryFailed = false ,
730 previousSidecars = [],
731 sessionCheck = null ,
732 download = defaultDownload,
733 outputDir = null ,
734 now = () => Date. now (),
735 sleep = ( ms ) => new Promise (( resolve ) => setTimeout (resolve, ms)),
736 progress = () => {},
737 onPersist = null ,
738 overdueMs = OVERDUE_MS ,
739 } = options;
740 if (caller !== 'extract' && caller !== 'census' ) throw new Error ( 'caller must be extract or census' );
741 if ( ! projectDir || ! manifestPath || ! stagingDir) throw new Error ( 'projectDir, manifestPath and stagingDir are required' );
742 if ( typeof execute !== 'function' || typeof execute.startBulkOperation !== 'function' ) throw new Error ( 'execute (a store executor with startBulkOperation) is required' );
743 if (caller === 'extract' && ! outputDir) throw new Error ( 'an extract needs outputDir' );
744 const stage = caller === 'extract' ? 'execution' : 'discovery' ;
745 const extract = caller === 'extract' ;
746 const normalizedStore = session. normalizeDomain (store);
747 const handlers = mergeHandlers (handlerOptions);
748 let manifest = null ;
749
750 // -------------------------------------------------------------------------------------------
751 // Manifest
752 // -------------------------------------------------------------------------------------------
753
754 async function save ( label , query = null ) {
755 await writeJsonAtomic (manifestPath, manifest);
756 if ( typeof onPersist === 'function' ) await onPersist ({ label, query, state: query ? manifest.queries[query].state : null , manifest });
757 }
758
759 async function readManifest () {
760 try {
761 return JSON . parse ( await fsp. readFile (manifestPath, 'utf8' ));
762 } catch (error) {
763 if (error.code === 'ENOENT' ) return null ;
764 throw error;
765 }
766 }
767
768 const entry = ( name ) => manifest.queries[name];
769
770 async function setState ( name , state , fields = {}, dropFields = []) {
771 const target = entry (name);
772 target.state = state;
773 Object. assign (target, fields);
774 for ( const field of dropFields) delete target[field];
775 await save ( `${ name }:${ state }` , name);
776 }
777
778 function dropOperation ( name ) {
779 const target = entry (name);
780 for ( const field of OPERATION_FIELDS ) delete target[field];
781 }
782
783 // -------------------------------------------------------------------------------------------
784 // Handlers
785 // -------------------------------------------------------------------------------------------
786
787 function coverageRowFor ( name , { artifact , pointer , recoveryAction }) {
788 const entity = entityOf (name);
789 const [ row ] = buildShopifyCoverageRows ({
790 captureFailures: [{ entity, artifact, reads: [name], errors: [{ code: 'shopify_bulk' }] }],
791 rawDir: RAW_EXTRACT_DIR ,
792 });
793 return { ... row, evidence: { artifact, pointer }, recoveryAction };
794 }
795
796 function failureEvidence ( name , target ) {
797 const parts = [ `queries.${ name }` , `state ${ target . outcome || target . state }` , `errorCode ${ target . errorCode || 'unknown'}` ];
798 if (target.error && target.error.message) parts. push ( String (target.error.message). slice ( 0 , 300 ));
799 return parts. join ( '; ' );
800 }
801
802 // The default `failed` handler. Extract: the capture-failed row, upserted by coverageId.
803 // Census: the same row, carried in the entry's handlerOutcome for the discovery coverage step
804 // (the raw coverage file does not exist until discovery's last step, so nothing is upserted).
805 async function defaultFailedHandler ({ name , entry : target }) {
806 const recoveryAction = extract
807 ? `Fix the cause named in the stop message, then run shopify-extract.js --retry-failed for ${ entityOf ( name ) }.`
808 : `Fix the cause, then run Shopify discovery again for ${ entityOf ( name ) } (the extract's gated read also covers it).` ;
809 const row = coverageRowFor (name, {
810 artifact: extract ? `${ RAW_EXTRACT_DIR }/${ EXTRACT_MANIFEST_FILE }` : path. relative (projectDir, manifestPath). split (path.sep). join ( '/' ),
811 pointer: failureEvidence (name, target),
812 recoveryAction,
813 });
814 if (extract) await upsertAdapterCoverageRows (projectDir, { adapter: 'shopify' , rows: [row] });
815 return { handlerOutcome: { coverageId: row.coverageId, ... (extract ? {} : { row }) } };
816 }
817
818 // The fail-closed denial fallback, used only when the caller registers no denial handler (the
819 // customer-and-order adapter registers its own): the denial is unattributed, so the dataset is treated like a failed one.
820 async function defaultDeniedHandler ( context ) {
821 const result = await defaultFailedHandler (context);
822 return { handlerOutcome: { attribution: 'unattributed' , ... result.handlerOutcome }, stop: { code: STOPPING_FAILED_CODE , message: `Shopify denied the ${ context . name } export and the denial could not be attributed to a withheld detail.` } };
823 }
824
825 function failedHandlerFor ( name ) {
826 const byQuery = (handlers.failed && handlers.failed.byQuery) || {};
827 return byQuery[name] || (handlers.failed && handlers.failed.default) || defaultFailedHandler;
828 }
829
830 function handlerContext ( name ) {
831 return { name, query: name, entity: entityOf (name), entry: entry (name), manifest, caller, stage, projectDir, store, execute, stagingDir, outputDir, manifestPath, now, sleep };
832 }
833
834 // -------------------------------------------------------------------------------------------
835 // Settling
836 // -------------------------------------------------------------------------------------------
837
838 async function beginSettling ( name , outcome , { errorCode , error }) {
839 await setState (name, 'settling' , { outcome, errorCode, error, settlingAt: iso ( now ()) });
840 }
841
842 // `later` is a start that finds an already-terminal stopping entry: it stops with the generic
843 // `shopify_bulk_failed`; the first settle stops with the entry's own code.
844 function stopFor ( name , target , handlerStop , { later = false } = {}) {
845 const entity = entityOf (name);
846 if (handlerStop) {
847 return new BulkStop (handlerStop.code, handlerStop.message || `Shopify ${ caller } stopped (${ handlerStop . code }) at ${ name }.` , { details: { query: name, entity, ... (handlerStop.details || {}) } });
848 }
849 const specific = target.errorCode && target.state === 'failed' ? target.errorCode : STOPPING_FAILED_CODE ;
850 const code = later ? STOPPING_FAILED_CODE : specific;
851 const cause = target.error && target.error.message ? `: ${ String ( target . error . message ). slice ( 0 , 300 ) }` : '' ;
852 return new BulkStop (code, `Shopify dataset ${ entity } could not be read (${ specific })${ cause }. Requires Import Operator: fix the cause named here, run \` shopify-extract.js --retry-failed \` , and when the extract completes record the coverage decision \` provided \` for \` shopify:${ entity } \` ; otherwise show the merchant the row at mapping.` , { details: { query: name, entity, errorCode: target.errorCode || null , coverageId: target.handlerOutcome ? target.handlerOutcome.coverageId : null } });
853 }
854
855 // Runs the outcome's handler for a `settling` entry, writes the terminal state, and returns the
856 // stop the entry causes, or null.
857 async function finishSettling ( name ) {
858 const target = entry (name);
859 const context = handlerContext (name);
860 let result;
861 if (target.outcome === 'failed' ) {
862 result = await failedHandlerFor (name)({ ... context, outcome: 'failed' });
863 } else {
864 result = await (handlers.denied || defaultDeniedHandler)({ ... context, outcome: 'denied' });
865 }
866 const terminal = { handlerOutcome: (result && result.handlerOutcome) || {} };
867 if (target.outcome === 'denied' ) terminal.deniedAt = iso ( now ());
868 await setState (name, target.outcome, terminal);
869 const handlerStop = result && result.stop ? result.stop : null ;
870 if (handlerStop) return stopFor (name, entry (name), handlerStop);
871 if ( isStopping ( entry (name), caller)) return stopFor (name, entry (name), null );
872 return null ;
873 }
874
875 async function settle ( name , outcome , details ) {
876 await beginSettling (name, outcome, details);
877 const stop = await finishSettling (name);
878 if (stop) throw stop;
879 }
880
881 async function settleAll () {
882 for ( const name of manifest.order) {
883 if ( entry (name).state === 'settling' ) {
884 const stop = await finishSettling (name);
885 if (stop) throw stop;
886 }
887 }
888 }
889
890 // -------------------------------------------------------------------------------------------
891 // Session expiry, denial
892 // -------------------------------------------------------------------------------------------
893
894 async function authExpiredStop ( extraMessage = '' ) {
895 const action = await session. recordAuthExpired (projectDir, { stage, store });
896 return new BulkStop ( 'source_auth_expired' , `the Shopify connection for ${ store } expired during the ${ caller }${ extraMessage }. Re-approval request ${ action . requestId } is pending: run the Shopify session check (with --connect if asked), answer it, then run the same step again. Finished reads are not repeated.` , { details: { requestId: action.requestId } });
897 }
898
899 async function runSessionCheck () {
900 if ( typeof sessionCheck !== 'function' ) return { ok: true };
901 return sessionCheck ();
902 }
903
904 async function handleDenial ( name , evidence ) {
905 const target = entry (name);
906 const check = await runSessionCheck ();
907 if ( ! check.ok && ! target.reapproved) {
908 dropOperation (name);
909 await setState (name, 'pending' , { reapproved: true });
910 throw await authExpiredStop ( ` (Shopify denied the ${ name } export and the session check failed)` );
911 }
912 const verbatim = { message: 'Shopify refused the bulk export with ACCESS_DENIED' , ... evidence };
913 if ( DENIAL_ATTRIBUTED_QUERIES . includes (name)) {
914 await settle (name, 'denied' , { errorCode: 'ACCESS_DENIED' , error: verbatim });
915 } else {
916 await settle (name, 'failed' , { errorCode: STOPPING_FAILED_CODE , error: { ... verbatim, message: 'Shopify refused the bulk export with ACCESS_DENIED' } });
917 }
918 }
919
920 // -------------------------------------------------------------------------------------------
921 // Per query
922 // -------------------------------------------------------------------------------------------
923
924 const hasCode = ( errors , code ) => (errors || []). some (( error ) => error && error.extensions && error.extensions.code === code);
925
926 async function startQuery ( name ) {
927 const target = entry (name);
928 await setState (name, 'starting' , { startedAt: iso ( now ()) });
929 progress ( 'Starting Shopify bulk export' , { query: name });
930 const result = await execute. startBulkOperation (target.text);
931 if (result.errors && result.errors. length ) {
932 if ( hasCode (result.errors, 'ACCESS_DENIED' )) { await handleDenial (name, { graphqlErrors: result.errors }); return ; }
933 await settle (name, 'failed' , { errorCode: STOPPING_FAILED_CODE , error: { message: 'Shopify answered the bulk start with GraphQL errors' , graphqlErrors: result.errors } });
934 return ;
935 }
936 const payload = result.data && result.data.bulkOperationRunQuery;
937 if ( ! payload) {
938 await settle (name, 'failed' , { errorCode: STOPPING_FAILED_CODE , error: { message: 'Shopify answered the bulk start with no bulkOperationRunQuery result' } });
939 return ;
940 }
941 if (Array. isArray (payload.userErrors) && payload.userErrors. length ) {
942 await settle (name, 'failed' , { errorCode: STOPPING_FAILED_CODE , error: { message: 'Shopify refused the bulk start' , userErrors: payload.userErrors } });
943 return ;
944 }
945 const operation = payload.bulkOperation;
946 if ( ! operation || typeof operation.id !== 'string' ) {
947 await settle (name, 'failed' , { errorCode: STOPPING_FAILED_CODE , error: { message: 'Shopify answered the bulk start with no operation id' } });
948 return ;
949 }
950 const runningAt = now ();
951 await setState (name, 'running' , { operationId: operation.id, runningAt: iso (runningAt), deadlineAt: iso (runningAt + overdueMs), overdueRound: 1 });
952 }
953
954 function overdueRequestId ( name , round ) {
955 return session. sequencedRequestId ( OVERDUE_NAMESPACE , `${ caller } \0 ${ name } \0 ${ entry ( name ). operationId }` , round);
956 }
957
958 async function readActionIfAny ( requestId ) {
959 try {
960 return await readUserAction (projectDir, requestId);
961 } catch (error) {
962 if (error.code === 'ENOENT' ) return null ;
963 throw error;
964 }
965 }
966
967 async function raiseOverdue ( name ) {
968 const target = entry (name);
969 const round = target.overdueRound || 1 ;
970 const requestId = overdueRequestId (name, round);
971 let action = await readActionIfAny (requestId);
972 if ( ! action) {
973 action = await createUserAction (projectDir, {
974 requestId,
975 kind: 'question' ,
976 stage,
977 title: `Shopify export overdue: ${ name } (shopify_bulk_overdue)` ,
978 prompt: `Shopify has been preparing the \` ${ name } \` export (operation ${ target . operationId }) for over ${ round * 4 } hours. Wait another 4 hours, or abort this dataset?` ,
979 response: { mode: 'choice' },
980 choices: [
981 { id: 'wait' , label: 'Wait another 4 hours' },
982 { id: 'abort' , label: 'Abort this dataset' },
983 ],
984 });
985 }
986 await setState (name, 'running' , { overdueActionId: action.requestId });
987 throw new BulkStop ( 'shopify_bulk_overdue' , `Shopify has been preparing the ${ name } export for over ${ round * 4 } hours. Answer request ${ action . requestId } (wait or abort), then run the same step again.` , { status: 'awaiting_user_action' , details: { query: name, operationId: target.operationId, round, requestIds: [action.requestId] } });
988 }
989
990 // Applies an answered overdue action, or stops while it is unanswered. Returns 'waited' | 'aborted'.
991 async function applyOverdue ( name ) {
992 const target = entry (name);
993 const action = await readActionIfAny (target.overdueActionId);
994 if ( ! action || action.status !== 'answered' ) {
995 throw new BulkStop ( 'shopify_bulk_overdue' , `The ${ name } export is overdue and request ${ target . overdueActionId } is unanswered; nothing was polled or started.` , { status: 'awaiting_user_action' , details: { query: name, operationId: target.operationId, round: target.overdueRound || 1 , requestIds: [target.overdueActionId] } });
996 }
997 if (action.answer && action.answer.choice === 'abort' ) {
998 await settle (name, 'failed' , { errorCode: 'shopify_bulk_overdue_aborted' , error: { message: `The ${ name } export was aborted after it ran over ${ ( target . overdueRound || 1 ) * 4 } hours; Shopify's operation ${ target . operationId } was not cancelled and its result was never downloaded` , operationId: target.operationId, requestId: action.requestId } });
999 return 'aborted' ;
1000 }
1001 const answeredAt = Date. parse (action.answer && action.answer.answeredAt);
1002 const base = Number. isFinite (answeredAt) ? answeredAt : now ();
1003 await setState (name, 'running' , { deadlineAt: iso (base + overdueMs), overdueRound: (target.overdueRound || 1 ) + 1 }, [ 'overdueActionId' ]);
1004 return 'waited' ;
1005 }
1006
1007 function classifyOperation ( operation ) {
1008 if ( ! operation) return 'failed' ;
1009 if (operation.partialDataUrl) return 'failed' ;
1010 if (operation.status === 'COMPLETED' ) return 'completed' ;
1011 if (operation.status === 'FAILED' && operation.errorCode === 'ACCESS_DENIED' ) return 'denied' ;
1012 if ([ 'FAILED' , 'CANCELED' , 'EXPIRED' ]. includes (operation.status)) return 'failed' ;
1013 return 'running' ;
1014 }
1015
1016 async function failedResult ( name , { errorCode = STOPPING_FAILED_CODE , message , evidence = {} }) {
1017 const target = entry (name);
1018 if (target.attempts < MAX_ATTEMPTS ) {
1019 dropOperation (name);
1020 await setState (name, 'pending' , { attempts: target.attempts + 1 , downloadTries: 0 });
1021 return ;
1022 }
1023 await settle (name, 'failed' , { errorCode, error: { message, ... evidence } });
1024 }
1025
1026 async function pollOperation ( name ) {
1027 const result = await execute ( STATUS_QUERY , { id: entry (name).operationId });
1028 if (result.errors && result.errors. length ) {
1029 return { errors: result.errors };
1030 }
1031 return { operation: result.data && result.data.node ? result.data.node : null };
1032 }
1033
1034 async function processRunning ( name ) {
1035 if ( entry (name).overdueActionId) {
1036 const outcome = await applyOverdue (name);
1037 if (outcome === 'aborted' ) return ;
1038 }
1039 let wait = POLL_START_MS ;
1040 for (;;) {
1041 const polled = await pollOperation (name);
1042 if (polled.errors) {
1043 if ( hasCode (polled.errors, 'ACCESS_DENIED' )) { await handleDenial (name, { graphqlErrors: polled.errors }); return ; }
1044 await settle (name, 'failed' , { errorCode: STOPPING_FAILED_CODE , error: { message: 'Shopify answered the status poll with GraphQL errors' , graphqlErrors: polled.errors } });
1045 return ;
1046 }
1047 const operation = polled.operation;
1048 const kind = classifyOperation (operation);
1049 if (kind === 'running' ) {
1050 if ( now () >= Date. parse ( entry (name).deadlineAt)) await raiseOverdue (name);
1051 await sleep (wait);
1052 wait = Math. min (wait * 2 , POLL_CAP_MS );
1053 continue ;
1054 }
1055 if (kind === 'denied' ) { await handleDenial (name, { operation: operationEvidence (operation) }); return ; }
1056 if (kind === 'failed' ) {
1057 await failedResult (name, { message: operation ? `Shopify's bulk operation ended ${ operation . status }${ operation . errorCode ? ` (${ operation . errorCode })` : ''}${ operation . partialDataUrl ? ' with only partial data' : ''}` : 'Shopify no longer knows the bulk operation' , evidence: { operation: operationEvidence (operation) } });
1058 return ;
1059 }
1060 await setState (name, 'running' , {
1061 url: operation.url || null ,
1062 fileSize: operation.fileSize === undefined ? null : operation.fileSize,
1063 objectCount: operation.objectCount,
1064 rootObjectCount: operation.rootObjectCount,
1065 createdAt: operation.createdAt,
1066 completedAt: operation.completedAt,
1067 });
1068 await downloadGate (name);
1069 return ;
1070 }
1071 }
1072
1073 function operationEvidence ( operation ) {
1074 if ( ! operation) return null ;
1075 return { id: operation.id, status: operation.status, errorCode: operation.errorCode || null , partialDataUrl: operation.partialDataUrl || null , objectCount: operation.objectCount, fileSize: operation.fileSize === undefined ? null : operation.fileSize };
1076 }
1077
1078 // Marks the entry `downloaded` only when the whole gate holds.
1079 async function downloadGate ( name ) {
1080 const jsonl = path. join (stagingDir, `${ name }.jsonl` );
1081 const partial = `${ jsonl }.partial` ;
1082 await fsp. mkdir (stagingDir, { recursive: true });
1083 for (;;) {
1084 const target = entry (name);
1085 if ( ! target.url) {
1086 if ( String (target.objectCount) === '0' ) {
1087 await writeTextAtomic (jsonl, '' );
1088 await setState (name, 'downloaded' , { bytes: 0 , lines: 0 });
1089 return ;
1090 }
1091 await failedResult (name, { message: `Shopify completed the ${ name } export with no file for ${ target . objectCount } objects` , evidence: { operation: { objectCount: target.objectCount } } });
1092 return ;
1093 }
1094 progress ( 'Downloading Shopify bulk file' , { query: name });
1095 await removeFile (partial);
1096 const result = await download (target.url, partial);
1097 const written = result && Number. isFinite (result.bytes) ? result.bytes : 0 ;
1098 if (result && (result.status === 400 || result.status === 403 )) {
1099 await removeFile (partial);
1100 if (target.linkReruns >= MAX_LINK_RERUNS ) {
1101 await settle (name, 'failed' , { errorCode: 'shopify_bulk_download_failed' , error: { message: `The ${ name } export's download link expired ${ target . linkReruns + 1 } times (HTTP ${ result . status })` , httpStatus: result.status } });
1102 return ;
1103 }
1104 dropOperation (name);
1105 await setState (name, 'pending' , { linkReruns: target.linkReruns + 1 , downloadTries: 0 });
1106 return ;
1107 }
1108 let reason = null ;
1109 if ( ! result || result.status !== 200 ) reason = `the download answered ${ result && result . status !== null && result . status !== undefined ? `HTTP ${ result . status }` : 'no HTTP response'}${ result && result . error ? ` (${ result . error })` : ''}` ;
1110 else if (result.error) reason = `the download failed: ${ result . error }` ;
1111 else if (target.fileSize === null || target.fileSize === undefined || String (written) !== String (target.fileSize)) reason = `the download was ${ written } bytes, Shopify reported ${ target . fileSize === null || target . fileSize === undefined ? 'no size' : target . fileSize }` ;
1112 let checked = null ;
1113 if ( ! reason) {
1114 checked = await validateJsonlFile (partial);
1115 if ( ! checked.ok) reason = checked.reason;
1116 }
1117 if ( ! reason) {
1118 await fsp. rename (partial, jsonl);
1119 await setState (name, 'downloaded' , { bytes: written, lines: checked.lines });
1120 return ;
1121 }
1122 await removeFile (partial);
1123 const tries = target.downloadTries + 1 ;
1124 await setState (name, 'running' , { downloadTries: tries });
1125 if (tries < MAX_DOWNLOAD_TRIES ) continue ;
1126 await failedResult (name, { errorCode: 'shopify_bulk_download_failed' , message: `The ${ name } export could not be downloaded whole after ${ tries } tries: ${ reason }` , evidence: { reason } });
1127 return ;
1128 }
1129 }
1130
1131 async function reassemble ( name ) {
1132 const target = entry (name);
1133 const catalog = catalogEntry (name);
1134 const jsonl = path. join (stagingDir, `${ name }.jsonl` );
1135 if ( ! fs. existsSync (jsonl)) {
1136 // `downloaded` promises a complete local file. If it is gone (staging cleared by hand), the
1137 // dataset is simply read again: no attempt is spent, nothing is guessed.
1138 dropOperation (name);
1139 await setState (name, 'pending' , { downloadTries: 0 }, [ 'bytes' , 'lines' ]);
1140 return ;
1141 }
1142 const foldedFile = path. join (stagingDir, `${ name }.folded.tmp` );
1143 const folded = await foldBulkJsonl (jsonl, catalog, { outputFile: foldedFile });
1144 if (folded.orphans) {
1145 await settle (name, 'failed' , { errorCode: 'shopify_bulk_orphan_child' , error: { message: `${ folded . orphanCount } child row(s) of ${ name } match no parent row` , orphans: folded.orphans } });
1146 return ;
1147 }
1148 // A registered hook wins; otherwise the three datasets whose connections sit under a list field
1149 // (a bulk query cannot hold those) get their nested ordinary reads.
1150 const hook = (handlers.reassemble && handlers.reassemble[name]) || (extract ? nestedReadHook (name) : null );
1151 let records = null ;
1152 let recordsFile = null ;
1153 let recordCount = folded.recordCount;
1154 let outputs = {};
1155 let sidecarFiles = {};
1156 let result = {};
1157 let outputFile = null ;
1158 let entryFields = {};
1159 let hookStop = null ;
1160 if (hook) {
1161 const hooked = await hook ({ ... handlerContext (name), records: iterateNdjson (foldedFile), recordsFile: foldedFile, recordCount, lines: folded.lines });
1162 if (hooked && hooked.ok === false ) {
1163 await removeFile (foldedFile);
1164 await settle (name, 'failed' , { errorCode: hooked.errorCode || STOPPING_FAILED_CODE , error: { message: hooked.message || `The ${ name } recheck failed (${ hooked . errorCode || STOPPING_FAILED_CODE })` , ... (hooked.evidence ? { evidence: hooked.evidence } : {}) } });
1165 return ;
1166 }
1167 if (hooked && hooked.records) records = hooked.records;
1168 if (hooked && hooked.recordsFile) recordsFile = hooked.recordsFile;
1169 if (hooked && Number. isInteger (hooked.recordCount)) recordCount = hooked.recordCount;
1170 if (hooked && hooked.sidecars) outputs = hooked.sidecars;
1171 if (hooked && hooked.sidecarFiles) sidecarFiles = hooked.sidecarFiles;
1172 if (hooked && hooked.manifest) Object. assign (manifest, hooked.manifest);
1173 if (hooked && hooked.result) result = hooked.result;
1174 if (hooked && hooked.outputFile) outputFile = hooked.outputFile;
1175 if (hooked && hooked.entry) entryFields = hooked.entry;
1176 if (hooked && hooked.stop) hookStop = hooked.stop;
1177 }
1178 const readAt = { createdAt: target.createdAt || null , completedAt: target.completedAt || null };
1179 const durationMs = Number. isFinite (Date. parse (target.completedAt)) && Number. isFinite (Date. parse (target.createdAt)) ? Date. parse (target.completedAt) - Date. parse (target.createdAt) : null ;
1180 if (extract) {
1181 const stagedName = outputFile || `${ name }.ndjson` ;
1182 const staged = path. join (stagingDir, stagedName);
1183 if (recordsFile) {
1184 await removeFile (staged);
1185 await fsp. rename (recordsFile, staged);
1186 await removeFile (foldedFile);
1187 } else if (records) {
1188 recordCount = await writeNdjsonAtomic (staged, records);
1189 await removeFile (foldedFile);
1190 } else {
1191 await fsp. rename (foldedFile, staged);
1192 }
1193 const sidecarNames = [ ...new Set ([ ... Object. keys (outputs), ... Object. keys (sidecarFiles)])]. sort ();
1194 const oldSidecars = Array. isArray (target.sidecars) ? target.sidecars : [];
1195 for ( const sidecar of oldSidecars. filter (( sidecar ) => ! sidecarNames. includes (sidecar))) {
1196 await removeFile (path. join (stagingDir, `sidecar-${ sidecar }.ndjson` ));
1197 manifest.obsoleteSidecars = [ ...new Set ([ ... (manifest.obsoleteSidecars || []), sidecar])]. sort ();
1198 }
1199 for ( const sidecar of sidecarNames) {
1200 const destination = path. join (stagingDir, `sidecar-${ sidecar }.ndjson` );
1201 if (sidecarFiles[sidecar]) {
1202 await removeFile (destination);
1203 await fsp. rename (sidecarFiles[sidecar], destination);
1204 } else {
1205 await writeNdjsonAtomic (destination, outputs[sidecar]);
1206 }
1207 }
1208 manifest.sidecars = [ ...new Set (manifest.order. flatMap (( query ) => query === name ? sidecarNames : ( entry (query).sidecars || [])))]. sort ();
1209 await setState (name, 'done' , { readAt, durationMs, records: recordCount, sidecars: sidecarNames, ... (outputFile ? { outputFile } : {}), ... entryFields }, [ 'url' ]);
1210 } else {
1211 await removeFile (foldedFile);
1212 await setState (name, 'done' , { readAt, durationMs, result: { records: recordCount, ... result }, ... entryFields }, [ 'url' ]);
1213 }
1214 await removeFile (jsonl);
1215 if (hookStop) throw new BulkStop (hookStop.code, hookStop.message || `Shopify extract stopped (${ hookStop . code }) at ${ name }.` , { details: { query: name, entity: entityOf (name), ... (hookStop.details || {}) } });
1216 }
1217
1218 async function processQuery ( name ) {
1219 for (;;) {
1220 const target = entry (name);
1221 switch (target.state) {
1222 case 'done' :
1223 return ;
1224 case 'denied' :
1225 if ( isStopping (target, caller)) throw stopFor (name, target, null , { later: true });
1226 return ;
1227 case 'failed' :
1228 if (extract) throw stopFor (name, target, null , { later: true });
1229 return ;
1230 case 'settling' : {
1231 const stop = await finishSettling (name);
1232 if (stop) throw stop;
1233 break ;
1234 }
1235 case 'starting' :
1236 dropOperation (name);
1237 await setState (name, 'pending' );
1238 break ;
1239 case 'pending' :
1240 await startQuery (name);
1241 break ;
1242 case 'running' :
1243 await processRunning (name);
1244 break ;
1245 case 'downloaded' :
1246 await reassemble (name);
1247 break ;
1248 default :
1249 throw new Error ( `unknown query state ${ target . state } for ${ name }` );
1250 }
1251 }
1252 }
1253
1254 // -------------------------------------------------------------------------------------------
1255 // Resume resets
1256 // -------------------------------------------------------------------------------------------
1257
1258 async function cleanLeftovers () {
1259 let names = [];
1260 try {
1261 names = await fsp. readdir (stagingDir);
1262 } catch (error) {
1263 if (error.code !== 'ENOENT' ) throw error;
1264 }
1265 for ( const leftover of names. filter (( file ) => file. endsWith ( '.partial' ) || file. endsWith ( '.tmp' ))) await removeFile (path. join (stagingDir, leftover));
1266 // A crash between the `done` write and the JSONL delete leaves the file behind.
1267 for ( const name of manifest.order) if ( entry (name).state === 'done' ) await removeFile (path. join (stagingDir, `${ name }.jsonl` ));
1268 }
1269
1270 function resetEntry ( name , fields = {}) {
1271 const target = entry (name);
1272 dropOperation (name);
1273 for ( const field of [ 'outcome' , 'errorCode' , 'error' , 'settlingAt' , 'handlerOutcome' , 'deniedAt' , 'readAt' , 'durationMs' , 'records' , 'result' , 'bytes' , 'lines' ]) delete target[field];
1274 Object. assign (target, { state: 'pending' , attempts: 0 , downloadTries: 0 , linkReruns: 0 , reapproved: false }, fields);
1275 }
1276
1277 async function applyResets () {
1278 if ( ! extract) return ;
1279 let changed = false ;
1280 if ( typeof deniedReset === 'function' ) {
1281 for ( const reset of ( await deniedReset ({ manifest, projectDir })) || []) {
1282 resetEntry (reset.name, { text: reset.text });
1283 changed = true ;
1284 }
1285 }
1286 if (retryFailed) {
1287 for ( const name of manifest.order) {
1288 if ( isStopping ( entry (name), caller)) { resetEntry (name); changed = true ; }
1289 }
1290 if (manifest.countFailure && manifest.countFailure.entities) {
1291 for ( const failedEntity of manifest.countFailure.entities) {
1292 for ( const name of manifest.order. filter (( candidate ) => entityOf (candidate) === failedEntity)) {
1293 await removeFile (path. join (stagingDir, `${ name }.ndjson` ));
1294 resetEntry (name);
1295 }
1296 }
1297 delete manifest.countFailure;
1298 changed = true ;
1299 }
1300 }
1301 // A manifest written before a catalog query existed lists neither it nor its rows. Finalizing it
1302 // as is would report a complete extract with that dataset's data (a discount's product, variant and
1303 // collection restrictions) absent, so each catalog query the manifest lacks is added as a pending
1304 // dataset and joined at the end like any other. A finished (`complete`) manifest never reaches
1305 // here: it needs an explicit catch-up.
1306 if (Array. isArray (addQueries)) {
1307 for ( const query of addQueries) {
1308 if (manifest.queries[query.name]) continue ;
1309 manifest.order. push (query.name);
1310 manifest.queries[query.name] = pendingEntry (query);
1311 changed = true ;
1312 }
1313 }
1314 // A dataset not yet started is sent the catalog's current text, so a fixed query reaches a resumed
1315 // manifest. Started and finished datasets keep the text they were sent (the snapshot time is the
1316 // manifest's, so the filter is unchanged).
1317 if ( typeof currentText === 'function' ) {
1318 for ( const name of manifest.order) {
1319 const target = entry (name);
1320 if (target.state !== 'pending' || target.operationId) continue ;
1321 const text = currentText (name);
1322 if ( typeof text === 'string' && text !== target.text) { target.text = text; changed = true ; }
1323 }
1324 }
1325 // A `done` extract entry promises its staged output, and a resume never reads a
1326 // finished query again. When that output is gone before the final join, read the query again
1327 // without spending an attempt instead of joining the hole as an empty dataset.
1328 for ( const name of manifest.order) {
1329 const target = entry (name);
1330 if (target.state !== 'done' ) continue ;
1331 const files = [target.outputFile || `${ name }.ndjson` , ... (target.sidecars || []). map (( sidecar ) => `sidecar-${ sidecar }.ndjson` )]. map (( file ) => path. join (stagingDir, file));
1332 const present = await Promise . all (files. map (( file ) => fsp. access (file). then (() => true , ( error ) => { if (error.code === 'ENOENT' ) return false ; throw error; })));
1333 if (present. every (Boolean)) continue ;
1334 for ( const file of files) await removeFile (file);
1335 resetEntry (name);
1336 changed = true ;
1337 }
1338 if (changed) await save ( 'resets' );
1339 }
1340
1341 // -------------------------------------------------------------------------------------------
1342 // Extract: the final join and the count check
1343 // -------------------------------------------------------------------------------------------
1344
1345 function expectedFinalFiles () {
1346 const entities = [ ...new Set (manifest.order. map (entityOf))];
1347 return [
1348 ... entities. map (( entity ) => {
1349 const first = manifest.order. find (( name ) => entityOf (name) === entity);
1350 return (first && entry (first).outputFile) || `${ entity }.ndjson` ;
1351 }),
1352 ... (manifest.sidecars || []). map (( entity ) => `${ entity }.ndjson` ),
1353 ];
1354 }
1355
1356 async function joinFinalFiles () {
1357 await fsp. mkdir (outputDir, { recursive: true });
1358 const staged = ( name ) => path. join (stagingDir, `${ name }.ndjson` );
1359 let droppedOrphans = 0 ;
1360 const droppedByQuery = {};
1361 for ( const entity of [ ...new Set (manifest.order. map (entityOf))]) {
1362 const members = manifest.order. filter (( name ) => entityOf (name) === entity);
1363 if (entity === 'discounts' && members. length > 1 ) {
1364 const joined = await joinDiscountFiles ({
1365 coreFile: staged ( 'discounts' ),
1366 supplements: members. filter (( name ) => catalogEntry (name).supplement). map (( name ) => ({ query: name, file: staged (name), ... catalogEntry (name).supplement })),
1367 outputFile: path. join (outputDir, 'discounts.ndjson' ),
1368 workDir: stagingDir,
1369 });
1370 droppedOrphans += joined.droppedOrphans;
1371 Object. assign (droppedByQuery, joined.droppedByQuery);
1372 continue ;
1373 }
1374 if (entity !== 'orders' ) {
1375 const outputFile = entry (members[ 0 ]).outputFile || `${ entity }.ndjson` ;
1376 const stagedFile = entry (members[ 0 ]).outputFile || `${ members [ 0 ] }.ndjson` ;
1377 await copyFileAtomic (path. join (stagingDir, stagedFile), path. join (outputDir, outputFile));
1378 continue ;
1379 }
1380 const joined = await joinOrderFiles ({
1381 coreFile: staged ( 'orders-core' ),
1382 refundsFile: members. includes ( 'orders-refunds' ) ? staged ( 'orders-refunds' ) : null ,
1383 fulfillmentsFile: members. includes ( 'orders-fulfillments' ) ? staged ( 'orders-fulfillments' ) : null ,
1384 outputFile: path. join (outputDir, 'orders.ndjson' ),
1385 workDir: stagingDir,
1386 });
1387 droppedOrphans += joined.droppedOrphans;
1388 Object. assign (droppedByQuery, joined.droppedByQuery);
1389 }
1390 for ( const sidecar of manifest.sidecars || []) await copyFileAtomic (path. join (stagingDir, `sidecar-${ sidecar }.ndjson` ), path. join (outputDir, `${ sidecar }.ndjson` ));
1391 const currentSidecars = new Set (manifest.sidecars || []);
1392 for ( const sidecar of new Set ([ ... (manifest.previousSidecars || []), ... (manifest.obsoleteSidecars || [])])) {
1393 if ( ! currentSidecars. has (sidecar)) await removeFile (path. join (outputDir, `${ sidecar }.ndjson` ));
1394 }
1395 manifest.droppedOrphans = droppedOrphans;
1396 manifest.droppedOrphansByQuery = droppedByQuery;
1397 }
1398
1399 const COUNT_QUERIES = {
1400 customers : ( snapshot ) => `query RpExtractCustomersCount { customersCount(query: "created_at:<='${ snapshot }'", limit: null) { count precision } }` ,
1401 orders : ( snapshot ) => `query RpExtractOrdersCount { ordersCount(query: "created_at:<='${ snapshot }'", limit: null) { count precision } }` ,
1402 };
1403
1404 async function readExactCount ( entity ) {
1405 const root = entity === 'customers' ? 'customersCount' : 'ordersCount' ;
1406 let result;
1407 try {
1408 result = await execute ( COUNT_QUERIES [entity](manifest.snapshotTime));
1409 } catch (error) {
1410 if (error instanceof ShopifyCliError && error.code === 'shopify_throttled' ) return { error: { code: 'shopify_throttled' , message: error.message } };
1411 throw error;
1412 }
1413 if (result.errors && result.errors. length ) return { error: { code: 'GRAPHQL_ERROR' , message: result.errors. map (( error ) => error && error.message). filter (Boolean). join ( '; ' ). slice ( 0 , 300 ) } };
1414 const counted = result.data && result.data[root];
1415 if ( ! counted || ! Number. isInteger (counted.count)) return { error: { code: 'COUNT_MISSING' , message: 'the count read returned no count' } };
1416 return { count: counted.count, precision: counted.precision || null };
1417 }
1418
1419 async function countCheck () {
1420 const customersReceived = ( await countLines (path. join (outputDir, 'customers.ndjson' ))) || 0 ;
1421 const withheld = (manifest.sidecars || []). includes ( 'customers-withheld-identity' ) ? (( await countLines (path. join (outputDir, 'customers-withheld-identity.ndjson' ))) || 0 ) : 0 ;
1422 const received = { customers: customersReceived + withheld, orders: ( await countLines (path. join (outputDir, 'orders.ndjson' ))) || 0 };
1423 manifest.counts = {};
1424 const rows = [];
1425 const stopCodes = [];
1426 for ( const entity of [ 'customers' , 'orders' ]) {
1427 const expected = await readExactCount (entity);
1428 const record = { received: received[entity], ... (entity === 'customers' ? { withheldIdentity: withheld } : {}) };
1429 let outcome;
1430 if (expected.error) {
1431 outcome = 'inexact' ;
1432 Object. assign (record, { error: expected.error });
1433 } else if (expected.precision !== 'EXACT' ) {
1434 outcome = 'inexact' ;
1435 Object. assign (record, { expected: expected.count, precision: expected.precision });
1436 } else if (received[entity] < expected.count) {
1437 outcome = 'short' ;
1438 Object. assign (record, { expected: expected.count, precision: expected.precision });
1439 } else {
1440 outcome = 'pass' ;
1441 Object. assign (record, { expected: expected.count, precision: expected.precision });
1442 }
1443 record.outcome = outcome;
1444 manifest.counts[entity] = record;
1445 if (outcome === 'pass' ) continue ;
1446 const code = outcome === 'short' ? 'shopify_extract_count_short' : 'shopify_extract_count_inexact' ;
1447 const pointer = outcome === 'short'
1448 ? `counts.${ entity }: received ${ record . received } < Shopify's exact count ${ record . expected } (precision ${ record . precision })`
1449 : `counts.${ entity }: Shopify's count is not exact (${ record . error ? `${ record . error . code }: ${ record . error . message }` : `precision ${ record . precision }`}); received ${ record . received }` ;
1450 rows. push ( coverageRowFor ( orderQueryOf (entity), { artifact: `${ RAW_EXTRACT_DIR }/${ EXTRACT_MANIFEST_FILE }` , pointer, recoveryAction: `Run shopify-extract.js --retry-failed to read ${ entity } again, or decide the shortfall at mapping.` }));
1451 stopCodes. push ({ code, entity });
1452 }
1453 if ( ! stopCodes. length ) { await save ( 'counts' ); return ; }
1454 manifest.countFailure = { entities: stopCodes. map (( stop ) => stop.entity), codes: stopCodes. map (( stop ) => stop.code) };
1455 await save ( 'counts' );
1456 const result = await upsertAdapterCoverageRows (projectDir, { adapter: 'shopify' , rows });
1457 throw new ShopifyExtractStop (stopCodes[ 0 ].code, `Shopify extract stopped (${ stopCodes [ 0 ]. code }): ${ rows . map (( row ) => row . coverageId ). join ( ', ' ) } recorded as source-coverage row(s) for the merchant's decision at mapping. Counts: ${ JSON . stringify ( manifest . counts ) }. Requires Import Operator: after any cause is fixed, run \` shopify-extract.js --retry-failed \` to read the dataset again; when the extract completes, record the coverage decision \` provided \` for each row, otherwise show the merchant the row at mapping.` , { coverageIds: rows. map (( row ) => row.coverageId), ... result });
1458 }
1459
1460 function orderQueryOf ( entity ) {
1461 return entity === 'customers' ? 'customers' : 'orders-core' ;
1462 }
1463
1464 async function finishExtract () {
1465 if ( ! manifest.order. every (( name ) => entry (name).state === 'done' )) {
1466 const open = manifest.order. filter (( name ) => entry (name).state !== 'done' );
1467 throw new BulkStop ( STOPPING_FAILED_CODE , `The extract cannot finish: ${ open . join ( ', ' ) } ${ open . length === 1 ? 'is' : 'are'} not read (${ open . map (( name ) => `${ name } ${ entry ( name ). state }` ). join ( ', ' ) }). Nothing was marked complete.` , { details: { open } });
1468 }
1469 await joinFinalFiles ();
1470 await save ( 'joined' );
1471 await countCheck ();
1472 manifest.status = 'complete' ;
1473 manifest.completedAt = iso ( now ());
1474 delete manifest.countFailure;
1475 await save ( 'complete' );
1476 await fsp. rm (stagingDir, { recursive: true , force: true });
1477 }
1478
1479 // -------------------------------------------------------------------------------------------
1480 // run()
1481 // -------------------------------------------------------------------------------------------
1482
1483 async function filesPresent () {
1484 for ( const file of expectedFinalFiles ()) {
1485 try {
1486 await fsp. access (path. join (outputDir, file));
1487 } catch {
1488 return false ;
1489 }
1490 }
1491 return true ;
1492 }
1493
1494 async function runInner () {
1495 manifest = await readManifest ();
1496 let storeMatches = ! manifest;
1497 if (manifest && manifest.store) {
1498 try {
1499 storeMatches = session. normalizeDomain (manifest.store) === normalizedStore;
1500 } catch {
1501 storeMatches = false ;
1502 }
1503 }
1504 if (manifest && ! storeMatches) {
1505 const recorded = manifest.store || null ;
1506 manifest = null ;
1507 throw new BulkStop ( 'shopify_extract_store_mismatch' , `${ path . basename ( manifestPath ) } was read from ${ recorded || 'an unrecorded store'}, not ${ normalizedStore }. Nothing was read or changed.` , { details: { recorded, store: normalizedStore } });
1508 }
1509 if ( ! manifest) {
1510 if ( ! Array. isArray (queries) || ! queries. length ) throw new Error ( 'queries are required to start a new manifest' );
1511 manifest = newManifest ({ caller, store: normalizedStore, snapshotTime: extract ? snapshotTime : null , queries, previousSidecars, now });
1512 await save ( 'created' );
1513 }
1514 if (extract && manifest.status === 'complete' ) {
1515 if ( ! ( await filesPresent ())) {
1516 throw new BulkStop ( 'shopify_extract_files_missing' , `The extract manifest says complete but its final files are missing from ${ path . relative ( projectDir , outputDir ) }. Run shopify-extract.js --catch-up to read the store again.` );
1517 }
1518 await fsp. rm (stagingDir, { recursive: true , force: true });
1519 return { status: 'complete' , skipped: true };
1520 }
1521 if (manifest.lastStop) { delete manifest.lastStop; await save ( 'cleared-stop' ); }
1522 await settleAll ();
1523 await cleanLeftovers ();
1524 await applyResets ();
1525 for ( const name of manifest.order) {
1526 if ( isStopping ( entry (name), caller)) throw stopFor (name, entry (name), null , { later: true });
1527 }
1528 for ( const name of manifest.order) await processQuery (name);
1529 if (extract) await finishExtract ();
1530 else { manifest.status = 'complete' ; manifest.completedAt = iso ( now ()); await save ( 'complete' ); }
1531 return { status: 'complete' };
1532 }
1533
1534 async function run () {
1535 let stop = null ;
1536 try {
1537 const outcome = await runInner ();
1538 return { ... outcome, manifest };
1539 } catch (error) {
1540 if (error instanceof BulkStop ) stop = error;
1541 else if (error instanceof ShopifyExtractStop ) stop = new BulkStop (error.code, error.message, { details: error.details });
1542 else if (error instanceof ShopifyCliError && error.code === 'source_auth_expired' ) stop = await authExpiredStop ();
1543 else if (error instanceof ShopifyCliError ) stop = new BulkStop (error.code, error.message, { details: { cli: true } });
1544 else throw error;
1545 }
1546 if (manifest) {
1547 manifest.lastStop = { code: stop.code, message: stop.message, status: stop.status, at: iso ( now ()) };
1548 await save ( 'stopped' );
1549 }
1550 return { status: stop.status, code: stop.code, message: stop.message, details: stop.details, manifest };
1551 }
1552
1553 return { run, get manifest () { return manifest; } };
1554 }
1555
1556 module . exports = {
1557 QUERIES_DIR,
1558 QUERY_CATALOG,
1559 QUERY_NAMES,
1560 CUSTOMER_QUERIES,
1561 ORDER_QUERIES,
1562 DENIAL_ATTRIBUTED_QUERIES,
1563 MANIFEST_SCHEMA_VERSION,
1564 EXTRACT_MANIFEST_FILE,
1565 RAW_EXTRACT_DIR,
1566 SNAPSHOT_PLACEHOLDER,
1567 STATUS_QUERY,
1568 POLL_START_MS,
1569 POLL_CAP_MS,
1570 OVERDUE_MS,
1571 MAX_DOWNLOAD_TRIES,
1572 MAX_LINK_RERUNS,
1573 BulkStop,
1574 catalogEntry,
1575 entityOf,
1576 readQueryFile,
1577 generateQueryText,
1578 connectionShape,
1579 joinDiscountFiles,
1580 validateJsonlFile,
1581 foldBulkJsonl,
1582 joinOrderFiles,
1583 defaultDownload,
1584 registerHandlers,
1585 clearRegisteredHandlers,
1586 registeredProjection : () => registered.project,
1587 createBulkRunner,
1588 };