Setting the file. One moment. 65
function dryRunCrosswalkPath
— line 65
This file
- Number
- 47.21
- Position
- 21 of 34
- Type
- JavaScript
- Size
- 19 KB
- Lines
- 660
lib/local-state.js
JavaScript·660 lines·19 KB
ATTEMPT_STATUS
=
new
Set
([
10 'started',
11 'succeeded_unverified',
12 'imported',
13 'failed_retryable',
14 'failed_terminal',
15 'deferred',
16 'needs_verification',
17 'skipped_already_imported',
18 'skipped_safe_mode_blocked',
19]);
20
21function stateDir(projectDir) {
22 return path.join(projectDir, 'state');
23}
24
25function crosswalkDir(projectDir) {
26 return path.join(stateDir(projectDir), 'crosswalk');
27}
28
29function crosswalkPath(projectDir) {
30 return path.join(crosswalkDir(projectDir), 'crosswalk.ndjson');
31}
32
33function crosswalkIndexDir(projectDir) {
34 return path.join(crosswalkDir(projectDir), 'indexes');
35}
36
37function attemptsDir(projectDir) {
38 return path.join(stateDir(projectDir), 'attempts');
39}
40
41function attemptJournalPath(projectDir) {
42 return path.join(attemptsDir(projectDir), 'write-attempts.ndjson');
43}
44
45function wixRequestCapturesPath(projectDir) {
46 return path.join(attemptsDir(projectDir), 'wix-request-captures.ndjson');
47}
48
49function cmsMirrorDir(projectDir) {
50 return path.join(stateDir(projectDir), 'cms-mirror');
51}
52
53function safeModeDir(projectDir) {
54 return path.join(stateDir(projectDir), 'safe-mode');
55}
56
57function safeModeEmailReplacementsPath(projectDir) {
58 return path.join(safeModeDir(projectDir), 'email-replacements.ndjson');
59}
60
61function safeModeBlockedRecordsPath(projectDir) {
62 return path.join(safeModeDir(projectDir), 'blocked-records.ndjson');
63}
64
65function dryRunCrosswalkPath(projectDir) {
66 return path.join(crosswalkDir(projectDir), 'dry-run-crosswalk.ndjson');
67}
68
69async function mkdirp(dirPath) {
70 await fs.mkdir(dirPath, { recursive: true });
71}
72
73async function pathExists(filePath) {
74 try {
75 await fs.access(filePath);
76 return true;
77 } catch (error) {
78 if (error && error.code === 'ENOENT') {
79 return false;
80 }
81 throw error;
82 }
83}
84
85async function writeJsonAtomic(filePath, data) {
86 await mkdirp(path.dirname(filePath));
87 const tempPath = `${filePath}.${process.pid}.${Date.now()}.tmp`;
88 await fs.writeFile(tempPath, `${JSON.stringify(data, null, 2)}\n`, 'utf8');
89 await fs.rename(tempPath, filePath);
90}
91
92async function writeTextAtomic(filePath, text) {
93 await mkdirp(path.dirname(filePath));
94 const tempPath = `${filePath}.${process.pid}.${Date.now()}.tmp`;
95 await fs.writeFile(tempPath, text, 'utf8');
96 await fs.rename(tempPath, filePath);
97}
98
99async function appendNdjson(filePath, row) {
100 await mkdirp(path.dirname(filePath));
101 await fs.appendFile(filePath, `${JSON.stringify(row)}\n`, 'utf8');
102}
103
104async function readNdjson(filePath) {
105 let raw;
106 try {
107 raw = await fs.readFile(filePath, 'utf8');
108 } catch (error) {
109 if (error && error.code === 'ENOENT') {
110 return [];
111 }
112 throw error;
113 }
114 const rows = [];
115 const lines = raw.split(/\r?\n/);
116 for (let index = 0; index < lines.length; index += 1) {
117 const line = lines[index].trim();
118 if (!line) {
119 continue;
120 }
121 try {
122 rows.push(JSON.parse(line));
123 } catch (error) {
124 throw new Error(`${filePath}:${index + 1} invalid NDJSON: ${error.message}`);
125 }
126 }
127 return rows;
128}
129
130function assertObject(row, label) {
131 if (!row || typeof row !== 'object' || Array.isArray(row)) {
132 throw new Error(`${label} must be an object`);
133 }
134}
135
136function requireString(row, field, label, errors) {
137 if (!row[field] || typeof row[field] !== 'string') {
138 errors.push(`${label}.${field} must be a non-empty string`);
139 }
140}
141
142function validateCrosswalkRow(row, { allowThrow = true, label = 'crosswalk row' } = {}) {
143 const errors = [];
144 try {
145 assertObject(row, label);
146 } catch (error) {
147 if (allowThrow) {
148 throw error;
149 }
150 return { ok: false, errors: [error.message] };
151 }
152 if (row.schemaVersion !== SCHEMA_VERSION) {
153 errors.push(`${label}.schemaVersion must be ${SCHEMA_VERSION}`);
154 }
155 for (const field of [
156 'sourceSystem',
157 'sourceEntityType',
158 'sourceId',
159 'sourceStableKey',
160 'targetSystem',
161 'targetEntityType',
162 'targetId',
163 'status',
164 ]) {
165 requireString(row, field, label, errors);
166 }
167 if (row.status && !CROSSWALK_STATUS.has(row.status)) {
168 errors.push(`${label}.status must be one of: ${Array.from(CROSSWALK_STATUS).join(', ')}`);
169 }
170 if (row.updatedAt !== undefined && Number.isNaN(Date.parse(row.updatedAt))) {
171 errors.push(`${label}.updatedAt must be an ISO timestamp when present`);
172 }
173 if (errors.length && allowThrow) {
174 throw new Error(errors.join('; '));
175 }
176 return { ok: errors.length === 0, errors };
177}
178
179function validateAttemptRow(row, { allowThrow = true, label = 'attempt row' } = {}) {
180 const errors = [];
181 try {
182 assertObject(row, label);
183 } catch (error) {
184 if (allowThrow) {
185 throw error;
186 }
187 return { ok: false, errors: [error.message] };
188 }
189 if (row.schemaVersion !== SCHEMA_VERSION) {
190 errors.push(`${label}.schemaVersion must be ${SCHEMA_VERSION}`);
191 }
192 for (const field of ['attemptId', 'sourceStableKey', 'writeSpecId', 'operation', 'targetEntityType', 'status']) {
193 requireString(row, field, label, errors);
194 }
195 if (row.status && !ATTEMPT_STATUS.has(row.status)) {
196 errors.push(`${label}.status must be one of: ${Array.from(ATTEMPT_STATUS).join(', ')}`);
197 }
198 if (row.blockedDataSnapshot !== undefined) {
199 const snapshot = row.blockedDataSnapshot;
200 if (!snapshot || typeof snapshot !== 'object' || Array.isArray(snapshot)) {
201 errors.push(`${label}.blockedDataSnapshot must be an object`);
202 } else {
203 for (const field of ['sourceEntityRef', 'extractedAt', 'checksum']) requireString(snapshot, field, `${label}.blockedDataSnapshot`, errors);
204 if (!Number.isInteger(snapshot.snapshotVersion) || snapshot.snapshotVersion < 1) {
205 errors.push(`${label}.blockedDataSnapshot.snapshotVersion must be a positive integer`);
206 }
207 if (snapshot.extractedAt !== undefined && Number.isNaN(Date.parse(snapshot.extractedAt))) {
208 errors.push(`${label}.blockedDataSnapshot.extractedAt must be an ISO timestamp`);
209 }
210 if (snapshot.checksum !== undefined && !/^sha256:[a-f0-9]{64}$/.test(snapshot.checksum)) {
211 errors.push(`${label}.blockedDataSnapshot.checksum must be a sha256 digest`);
212 }
213 }
214 }
215 if (errors.length && allowThrow) {
216 throw new Error(errors.join('; '));
217 }
218 return { ok: errors.length === 0, errors };
219}
220
221function validateSafeModeEmailReplacementRow(row, { allowThrow = true, label = 'safe-mode email replacement row' } = {}) {
222 const errors = [];
223 try {
224 assertObject(row, label);
225 } catch (error) {
226 if (allowThrow) {
227 throw error;
228 }
229 return { ok: false, errors: [error.message] };
230 }
231 if (row.schemaVersion !== SCHEMA_VERSION) {
232 errors.push(`${label}.schemaVersion must be ${SCHEMA_VERSION}`);
233 }
234 for (const field of [
235 'runId',
236 'sourceSystem',
237 'sourceEntityType',
238 'sourceId',
239 'sourceStableKey',
240 'targetSystem',
241 'targetEntityType',
242 'targetId',
243 'sourceEmail',
244 'targetEmail',
245 'createdAt',
246 ]) {
247 requireString(row, field, label, errors);
248 }
249 if (row.createdAt !== undefined && Number.isNaN(Date.parse(row.createdAt))) {
250 errors.push(`${label}.createdAt must be an ISO timestamp`);
251 }
252 if (errors.length && allowThrow) {
253 throw new Error(errors.join('; '));
254 }
255 return { ok: errors.length === 0, errors };
256}
257
258function validateSafeModeBlockedRecordRow(row, { allowThrow = true, label = 'safe-mode blocked record row' } = {}) {
259 const errors = [];
260 try {
261 assertObject(row, label);
262 } catch (error) {
263 if (allowThrow) {
264 throw error;
265 }
266 return { ok: false, errors: [error.message] };
267 }
268 if (row.schemaVersion !== SCHEMA_VERSION) {
269 errors.push(`${label}.schemaVersion must be ${SCHEMA_VERSION}`);
270 }
271 for (const field of [
272 'runId',
273 'sourceSystem',
274 'sourceEntityType',
275 'sourceId',
276 'sourceStableKey',
277 'targetSystem',
278 'targetEntityType',
279 'reason',
280 'createdAt',
281 ]) {
282 requireString(row, field, label, errors);
283 }
284 if (!Array.isArray(row.paths) || row.paths.some((item) => typeof item !== 'string' || !item)) {
285 errors.push(`${label}.paths must be an array of non-empty strings`);
286 }
287 if (row.reason && row.reason !== 'SAFE_MODE_SUSPICIOUS_EMAIL') {
288 errors.push(`${label}.reason must be SAFE_MODE_SUSPICIOUS_EMAIL`);
289 }
290 if (row.createdAt !== undefined && Number.isNaN(Date.parse(row.createdAt))) {
291 errors.push(`${label}.createdAt must be an ISO timestamp`);
292 }
293 if (errors.length && allowThrow) {
294 throw new Error(errors.join('; '));
295 }
296 return { ok: errors.length === 0, errors };
297}
298
299function validateWixRequestCaptureRow(row, { allowThrow = true, label = 'Wix request capture row' } = {}) {
300 const errors = [];
301 try {
302 assertObject(row, label);
303 } catch (error) {
304 if (allowThrow) {
305 throw error;
306 }
307 return { ok: false, errors: [error.message] };
308 }
309 if (row.schemaVersion !== SCHEMA_VERSION) {
310 errors.push(`${label}.schemaVersion must be ${SCHEMA_VERSION}`);
311 }
312 for (const field of ['requestCaptureId', 'timestamp', 'runId', 'phase', 'method', 'endpoint', 'result']) {
313 requireString(row, field, label, errors);
314 }
315 if (row.phase && !['setup', 'import'].includes(row.phase)) {
316 errors.push(`${label}.phase must be setup or import`);
317 }
318 if (row.result && row.result !== 'dry_run_skipped_wix_call') {
319 errors.push(`${label}.result must be dry_run_skipped_wix_call`);
320 }
321 if (row.headers !== undefined && (!row.headers || typeof row.headers !== 'object' || Array.isArray(row.headers))) {
322 errors.push(`${label}.headers must be an object when present`);
323 }
324 if (row.headers && Object.prototype.hasOwnProperty.call(row.headers, 'Authorization')) {
325 errors.push(`${label}.headers must not include Authorization`);
326 }
327 if (Number.isNaN(Date.parse(row.timestamp))) {
328 errors.push(`${label}.timestamp must be an ISO timestamp`);
329 }
330 if (errors.length && allowThrow) {
331 throw new Error(errors.join('; '));
332 }
333 return { ok: errors.length === 0, errors };
334}
335
336function validateDryRunCrosswalkRow(row, { allowThrow = true, label = 'dry-run crosswalk row' } = {}) {
337 const errors = [];
338 try {
339 assertObject(row, label);
340 } catch (error) {
341 if (allowThrow) {
342 throw error;
343 }
344 return { ok: false, errors: [error.message] };
345 }
346 if (row.schemaVersion !== SCHEMA_VERSION) {
347 errors.push(`${label}.schemaVersion must be ${SCHEMA_VERSION}`);
348 }
349 for (const field of [
350 'runId',
351 'sourceSystem',
352 'sourceEntityType',
353 'sourceId',
354 'sourceStableKey',
355 'targetSystem',
356 'targetEntityType',
357 'placeholderTargetId',
358 'operation',
359 'createdAt',
360 ]) {
361 requireString(row, field, label, errors);
362 }
363 if (row.dryRun !== true) {
364 errors.push(`${label}.dryRun must be true`);
365 }
366 if (row.placeholder !== true) {
367 errors.push(`${label}.placeholder must be true`);
368 }
369 if (row.createdAt !== undefined && Number.isNaN(Date.parse(row.createdAt))) {
370 errors.push(`${label}.createdAt must be an ISO timestamp`);
371 }
372 if (errors.length && allowThrow) {
373 throw new Error(errors.join('; '));
374 }
375 return { ok: errors.length === 0, errors };
376}
377
378function indexCrosswalkRows(rows) {
379 const bySource = {};
380 const byTarget = {};
381 for (const row of rows) {
382 bySource[row.sourceStableKey] = row;
383 byTarget[row.targetId] = row;
384 }
385 return { bySource, byTarget };
386}
387
388async function loadCrosswalk(projectDir) {
389 const rows = await readNdjson(crosswalkPath(projectDir));
390 for (const row of rows) {
391 validateCrosswalkRow(row);
392 }
393 const { bySource, byTarget } = indexCrosswalkRows(rows);
394 return { rows, bySource, byTarget };
395}
396
397async function appendCrosswalkRow(projectDir, row) {
398 validateCrosswalkRow(row);
399 await appendNdjson(crosswalkPath(projectDir), row);
400 return row;
401}
402
403async function upsertCrosswalkRow(projectDir, row) {
404 validateCrosswalkRow(row);
405 const current = await loadCrosswalk(projectDir);
406 current.bySource[row.sourceStableKey] = row;
407 const rows = Object.values(current.bySource).sort((a, b) => a.sourceStableKey.localeCompare(b.sourceStableKey));
408 const text = rows.map((item) => JSON.stringify(item)).join('\n');
409 await writeTextAtomic(crosswalkPath(projectDir), text ? `${text}\n` : '');
410 await rebuildCrosswalkIndexes(projectDir);
411 return row;
412}
413
414function foldAttempts(rows) {
415 const byAttemptId = {};
416 for (const row of rows) {
417 byAttemptId[row.attemptId] = {
418 ...(byAttemptId[row.attemptId] || {}),
419 ...row,
420 };
421 }
422 return byAttemptId;
423}
424
425async function loadAttemptJournal(projectDir) {
426 const rows = await readNdjson(attemptJournalPath(projectDir));
427 for (const row of rows) {
428 validateAttemptRow(row);
429 }
430 return { rows, byAttemptId: foldAttempts(rows) };
431}
432
433async function appendAttempt(projectDir, row) {
434 validateAttemptRow(row);
435 await appendNdjson(attemptJournalPath(projectDir), row);
436 return row;
437}
438
439async function appendSafeModeEmailReplacement(projectDir, row) {
440 validateSafeModeEmailReplacementRow(row);
441 await appendNdjson(safeModeEmailReplacementsPath(projectDir), row);
442 return row;
443}
444
445async function appendSafeModeBlockedRecord(projectDir, row) {
446 validateSafeModeBlockedRecordRow(row);
447 await appendNdjson(safeModeBlockedRecordsPath(projectDir), row);
448 return row;
449}
450
451async function appendWixRequestCapture(projectDir, row) {
452 validateWixRequestCaptureRow(row);
453 await appendNdjson(wixRequestCapturesPath(projectDir), row);
454 return row;
455}
456
457async function loadWixRequestCaptures(projectDir) {
458 const rows = await readNdjson(wixRequestCapturesPath(projectDir));
459 for (const row of rows) {
460 validateWixRequestCaptureRow(row);
461 }
462 return rows;
463}
464
465async function appendDryRunCrosswalkRow(projectDir, row) {
466 validateDryRunCrosswalkRow(row);
467 await appendNdjson(dryRunCrosswalkPath(projectDir), row);
468 return row;
469}
470
471async function loadDryRunCrosswalk(projectDir) {
472 const rows = await readNdjson(dryRunCrosswalkPath(projectDir));
473 for (const row of rows) {
474 validateDryRunCrosswalkRow(row);
475 }
476 return rows;
477}
478
479function dryRunUpsertDecision({ localCrosswalkRow = null, requiresRevision = false, hasLocalRevision = false, supportsRevisionFreeRequestBuild = false } = {}) {
480 if (localCrosswalkRow) {
481 return {
482 dryRun: true,
483 decision: 'based_on_local_crosswalk',
484 targetId: localCrosswalkRow.targetId,
485 stateKnown: true,
486 };
487 }
488 if (requiresRevision && !hasLocalRevision) {
489 return {
490 dryRun: true,
491 decision: 'would_require_live_lookup',
492 stateKnown: false,
493 canBuildRequest: Boolean(supportsRevisionFreeRequestBuild),
494 };
495 }
496 return {
497 dryRun: true,
498 decision: 'would_create_if_not_found',
499 stateKnown: false,
500 canBuildRequest: true,
501 };
502}
503
504async function markAttempt(projectDir, attemptId, patch) {
505 if (!attemptId || typeof attemptId !== 'string') {
506 throw new Error('attemptId must be a non-empty string');
507 }
508 assertObject(patch, 'attempt patch');
509 const journal = await loadAttemptJournal(projectDir);
510 const current = journal.byAttemptId[attemptId];
511 if (!current) {
512 throw new Error(`unknown attemptId: ${attemptId}`);
513 }
514 const row = {
515 ...current,
516 ...patch,
517 attemptId,
518 };
519 validateAttemptRow(row);
520 await appendNdjson(attemptJournalPath(projectDir), row);
521 return row;
522}
523
524async function rebuildCrosswalkIndexes(projectDir) {
525 const { bySource, byTarget } = await loadCrosswalk(projectDir);
526 await writeJsonAtomic(path.join(crosswalkIndexDir(projectDir), 'by-source.json'), bySource);
527 await writeJsonAtomic(path.join(crosswalkIndexDir(projectDir), 'by-target.json'), byTarget);
528 return { bySource, byTarget };
529}
530
531async function localCrosswalkStateExists(projectDir) {
532 if (!(await pathExists(crosswalkPath(projectDir)))) {
533 return false;
534 }
535 await loadCrosswalk(projectDir);
536 return true;
537}
538
539function newestRow(a, b) {
540 const aTime = Date.parse(a.updatedAt);
541 const bTime = Date.parse(b.updatedAt);
542 if (Number.isNaN(aTime) || Number.isNaN(bTime) || aTime === bTime) {
543 return null;
544 }
545 return bTime > aTime ? b : a;
546}
547
548async function seedCrosswalkFromCmsMirror(projectDir, rows) {
549 if (!Array.isArray(rows)) {
550 throw new Error('CMS mirror rows must be an array');
551 }
552 if (await localCrosswalkStateExists(projectDir)) {
553 return { seeded: false, reason: 'local_crosswalk_exists', accepted: 0, rejected: 0, conflicts: 0 };
554 }
555
556 const rejected = [];
557 const conflicts = [];
558 const bySource = {};
559
560 for (const row of rows) {
561 const validation = validateCrosswalkRow(row, { allowThrow: false, label: 'CMS mirror row' });
562 if (!validation.ok) {
563 rejected.push({ row, errors: validation.errors });
564 continue;
565 }
566
567 const current = bySource[row.sourceStableKey];
568 if (!current) {
569 bySource[row.sourceStableKey] = row;
570 continue;
571 }
572
573 const winner = newestRow(current, row);
574 if (!winner) {
575 if (current.targetId !== row.targetId) {
576 conflicts.push(current, row);
577 }
578 continue;
579 }
580 bySource[row.sourceStableKey] = winner;
581 }
582
583 await mkdirp(cmsMirrorDir(projectDir));
584 await writeTextAtomic(
585 path.join(cmsMirrorDir(projectDir), 'imported-from-cms.ndjson'),
586 rows.map((row) => JSON.stringify(row)).join('\n') + (rows.length ? '\n' : ''),
587 );
588 if (rejected.length) {
589 await writeTextAtomic(
590 path.join(cmsMirrorDir(projectDir), 'rejected.ndjson'),
591 rejected.map((item) => JSON.stringify(item)).join('\n') + '\n',
592 );
593 }
594 if (conflicts.length) {
595 await writeTextAtomic(
596 path.join(cmsMirrorDir(projectDir), 'conflicts.ndjson'),
597 conflicts.map((row) => JSON.stringify(row)).join('\n') + '\n',
598 );
599 throw new Error(`CMS mirror seed has ${conflicts.length} conflicting rows; see state/cms-mirror/conflicts.ndjson`);
600 }
601
602 const acceptedRows = Object.values(bySource).sort((a, b) => a.sourceStableKey.localeCompare(b.sourceStableKey));
603 const text = acceptedRows.map((row) => JSON.stringify(row)).join('\n');
604 await writeTextAtomic(crosswalkPath(projectDir), text ? `${text}\n` : '');
605 await rebuildCrosswalkIndexes(projectDir);
606 return { seeded: true, accepted: acceptedRows.length, rejected: rejected.length, conflicts: 0 };
607}
608
609async function withStateLock(projectDir, fn) {
610 const lockPath = path.join(stateDir(projectDir), '.lock');
611 await mkdirp(stateDir(projectDir));
612 try {
613 await fs.mkdir(lockPath);
614 } catch (error) {
615 if (error && error.code === 'EEXIST') {
616 throw new Error(`state lock already held: ${lockPath}`);
617 }
618 throw error;
619 }
620 try {
621 return await fn();
622 } finally {
623 await fs.rm(lockPath, { recursive: true, force: true });
624 }
625}
626
627module.exports = {
628 SCHEMA_VERSION,
629 stateDir,
630 crosswalkPath,
631 attemptJournalPath,
632 wixRequestCapturesPath,
633 cmsMirrorDir,
634 safeModeDir,
635 safeModeEmailReplacementsPath,
636 safeModeBlockedRecordsPath,
637 dryRunCrosswalkPath,
638 loadCrosswalk,
639 appendCrosswalkRow,
640 upsertCrosswalkRow,
641 loadAttemptJournal,
642 appendAttempt,
643 validateAttemptRow,
644 appendSafeModeEmailReplacement,
645 appendSafeModeBlockedRecord,
646 appendWixRequestCapture,
647 loadWixRequestCaptures,
648 appendDryRunCrosswalkRow,
649 loadDryRunCrosswalk,
650 dryRunUpsertDecision,
651 validateSafeModeEmailReplacementRow,
652 validateSafeModeBlockedRecordRow,
653 validateWixRequestCaptureRow,
654 validateDryRunCrosswalkRow,
655 markAttempt,
656 rebuildCrosswalkIndexes,
657 localCrosswalkStateExists,
658 seedCrosswalkFromCmsMirror,
659 withStateLock,
660};