Setting the file. One moment. 65
function dryRunCrosswalkPath
— line 65
This file
- Number
- 21.14
- Position
- 14 of 20
- Type
- JavaScript
- Size
- 18 KB
- Lines
- 642
lib/local-state.js
JavaScript·642 lines·18 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 (errors.length && allowThrow) {
199 throw new Error(errors.join('; '));
200 }
201 return { ok: errors.length === 0, errors };
202}
203
204function validateSafeModeEmailReplacementRow(row, { allowThrow = true, label = 'safe-mode email replacement row' } = {}) {
205 const errors = [];
206 try {
207 assertObject(row, label);
208 } catch (error) {
209 if (allowThrow) {
210 throw error;
211 }
212 return { ok: false, errors: [error.message] };
213 }
214 if (row.schemaVersion !== SCHEMA_VERSION) {
215 errors.push(`${label}.schemaVersion must be ${SCHEMA_VERSION}`);
216 }
217 for (const field of [
218 'runId',
219 'sourceSystem',
220 'sourceEntityType',
221 'sourceId',
222 'sourceStableKey',
223 'targetSystem',
224 'targetEntityType',
225 'targetId',
226 'sourceEmail',
227 'targetEmail',
228 'createdAt',
229 ]) {
230 requireString(row, field, label, errors);
231 }
232 if (row.createdAt !== undefined && Number.isNaN(Date.parse(row.createdAt))) {
233 errors.push(`${label}.createdAt must be an ISO timestamp`);
234 }
235 if (errors.length && allowThrow) {
236 throw new Error(errors.join('; '));
237 }
238 return { ok: errors.length === 0, errors };
239}
240
241function validateSafeModeBlockedRecordRow(row, { allowThrow = true, label = 'safe-mode blocked record row' } = {}) {
242 const errors = [];
243 try {
244 assertObject(row, label);
245 } catch (error) {
246 if (allowThrow) {
247 throw error;
248 }
249 return { ok: false, errors: [error.message] };
250 }
251 if (row.schemaVersion !== SCHEMA_VERSION) {
252 errors.push(`${label}.schemaVersion must be ${SCHEMA_VERSION}`);
253 }
254 for (const field of [
255 'runId',
256 'sourceSystem',
257 'sourceEntityType',
258 'sourceId',
259 'sourceStableKey',
260 'targetSystem',
261 'targetEntityType',
262 'reason',
263 'createdAt',
264 ]) {
265 requireString(row, field, label, errors);
266 }
267 if (!Array.isArray(row.paths) || row.paths.some((item) => typeof item !== 'string' || !item)) {
268 errors.push(`${label}.paths must be an array of non-empty strings`);
269 }
270 if (row.reason && row.reason !== 'SAFE_MODE_SUSPICIOUS_EMAIL') {
271 errors.push(`${label}.reason must be SAFE_MODE_SUSPICIOUS_EMAIL`);
272 }
273 if (row.createdAt !== undefined && Number.isNaN(Date.parse(row.createdAt))) {
274 errors.push(`${label}.createdAt must be an ISO timestamp`);
275 }
276 if (errors.length && allowThrow) {
277 throw new Error(errors.join('; '));
278 }
279 return { ok: errors.length === 0, errors };
280}
281
282function validateWixRequestCaptureRow(row, { allowThrow = true, label = 'Wix request capture row' } = {}) {
283 const errors = [];
284 try {
285 assertObject(row, label);
286 } catch (error) {
287 if (allowThrow) {
288 throw error;
289 }
290 return { ok: false, errors: [error.message] };
291 }
292 if (row.schemaVersion !== SCHEMA_VERSION) {
293 errors.push(`${label}.schemaVersion must be ${SCHEMA_VERSION}`);
294 }
295 for (const field of ['requestCaptureId', 'timestamp', 'runId', 'phase', 'method', 'endpoint', 'result']) {
296 requireString(row, field, label, errors);
297 }
298 if (row.phase && !['setup', 'import'].includes(row.phase)) {
299 errors.push(`${label}.phase must be setup or import`);
300 }
301 if (row.result && row.result !== 'dry_run_skipped_wix_call') {
302 errors.push(`${label}.result must be dry_run_skipped_wix_call`);
303 }
304 if (row.headers !== undefined && (!row.headers || typeof row.headers !== 'object' || Array.isArray(row.headers))) {
305 errors.push(`${label}.headers must be an object when present`);
306 }
307 if (row.headers && Object.prototype.hasOwnProperty.call(row.headers, 'Authorization')) {
308 errors.push(`${label}.headers must not include Authorization`);
309 }
310 if (Number.isNaN(Date.parse(row.timestamp))) {
311 errors.push(`${label}.timestamp must be an ISO timestamp`);
312 }
313 if (errors.length && allowThrow) {
314 throw new Error(errors.join('; '));
315 }
316 return { ok: errors.length === 0, errors };
317}
318
319function validateDryRunCrosswalkRow(row, { allowThrow = true, label = 'dry-run crosswalk row' } = {}) {
320 const errors = [];
321 try {
322 assertObject(row, label);
323 } catch (error) {
324 if (allowThrow) {
325 throw error;
326 }
327 return { ok: false, errors: [error.message] };
328 }
329 if (row.schemaVersion !== SCHEMA_VERSION) {
330 errors.push(`${label}.schemaVersion must be ${SCHEMA_VERSION}`);
331 }
332 for (const field of [
333 'runId',
334 'sourceSystem',
335 'sourceEntityType',
336 'sourceId',
337 'sourceStableKey',
338 'targetSystem',
339 'targetEntityType',
340 'placeholderTargetId',
341 'operation',
342 'createdAt',
343 ]) {
344 requireString(row, field, label, errors);
345 }
346 if (row.dryRun !== true) {
347 errors.push(`${label}.dryRun must be true`);
348 }
349 if (row.placeholder !== true) {
350 errors.push(`${label}.placeholder must be true`);
351 }
352 if (row.createdAt !== undefined && Number.isNaN(Date.parse(row.createdAt))) {
353 errors.push(`${label}.createdAt must be an ISO timestamp`);
354 }
355 if (errors.length && allowThrow) {
356 throw new Error(errors.join('; '));
357 }
358 return { ok: errors.length === 0, errors };
359}
360
361function indexCrosswalkRows(rows) {
362 const bySource = {};
363 const byTarget = {};
364 for (const row of rows) {
365 bySource[row.sourceStableKey] = row;
366 byTarget[row.targetId] = row;
367 }
368 return { bySource, byTarget };
369}
370
371async function loadCrosswalk(projectDir) {
372 const rows = await readNdjson(crosswalkPath(projectDir));
373 for (const row of rows) {
374 validateCrosswalkRow(row);
375 }
376 const { bySource, byTarget } = indexCrosswalkRows(rows);
377 return { rows, bySource, byTarget };
378}
379
380async function appendCrosswalkRow(projectDir, row) {
381 validateCrosswalkRow(row);
382 await appendNdjson(crosswalkPath(projectDir), row);
383 return row;
384}
385
386async function upsertCrosswalkRow(projectDir, row) {
387 validateCrosswalkRow(row);
388 const current = await loadCrosswalk(projectDir);
389 current.bySource[row.sourceStableKey] = row;
390 const rows = Object.values(current.bySource).sort((a, b) => a.sourceStableKey.localeCompare(b.sourceStableKey));
391 const text = rows.map((item) => JSON.stringify(item)).join('\n');
392 await writeTextAtomic(crosswalkPath(projectDir), text ? `${text}\n` : '');
393 await rebuildCrosswalkIndexes(projectDir);
394 return row;
395}
396
397function foldAttempts(rows) {
398 const byAttemptId = {};
399 for (const row of rows) {
400 byAttemptId[row.attemptId] = {
401 ...(byAttemptId[row.attemptId] || {}),
402 ...row,
403 };
404 }
405 return byAttemptId;
406}
407
408async function loadAttemptJournal(projectDir) {
409 const rows = await readNdjson(attemptJournalPath(projectDir));
410 for (const row of rows) {
411 validateAttemptRow(row);
412 }
413 return { rows, byAttemptId: foldAttempts(rows) };
414}
415
416async function appendAttempt(projectDir, row) {
417 validateAttemptRow(row);
418 await appendNdjson(attemptJournalPath(projectDir), row);
419 return row;
420}
421
422async function appendSafeModeEmailReplacement(projectDir, row) {
423 validateSafeModeEmailReplacementRow(row);
424 await appendNdjson(safeModeEmailReplacementsPath(projectDir), row);
425 return row;
426}
427
428async function appendSafeModeBlockedRecord(projectDir, row) {
429 validateSafeModeBlockedRecordRow(row);
430 await appendNdjson(safeModeBlockedRecordsPath(projectDir), row);
431 return row;
432}
433
434async function appendWixRequestCapture(projectDir, row) {
435 validateWixRequestCaptureRow(row);
436 await appendNdjson(wixRequestCapturesPath(projectDir), row);
437 return row;
438}
439
440async function loadWixRequestCaptures(projectDir) {
441 const rows = await readNdjson(wixRequestCapturesPath(projectDir));
442 for (const row of rows) {
443 validateWixRequestCaptureRow(row);
444 }
445 return rows;
446}
447
448async function appendDryRunCrosswalkRow(projectDir, row) {
449 validateDryRunCrosswalkRow(row);
450 await appendNdjson(dryRunCrosswalkPath(projectDir), row);
451 return row;
452}
453
454async function loadDryRunCrosswalk(projectDir) {
455 const rows = await readNdjson(dryRunCrosswalkPath(projectDir));
456 for (const row of rows) {
457 validateDryRunCrosswalkRow(row);
458 }
459 return rows;
460}
461
462function dryRunUpsertDecision({ localCrosswalkRow = null, requiresRevision = false, hasLocalRevision = false, supportsRevisionFreeRequestBuild = false } = {}) {
463 if (localCrosswalkRow) {
464 return {
465 dryRun: true,
466 decision: 'based_on_local_crosswalk',
467 targetId: localCrosswalkRow.targetId,
468 stateKnown: true,
469 };
470 }
471 if (requiresRevision && !hasLocalRevision) {
472 return {
473 dryRun: true,
474 decision: 'would_require_live_lookup',
475 stateKnown: false,
476 canBuildRequest: Boolean(supportsRevisionFreeRequestBuild),
477 };
478 }
479 return {
480 dryRun: true,
481 decision: 'would_create_if_not_found',
482 stateKnown: false,
483 canBuildRequest: true,
484 };
485}
486
487async function markAttempt(projectDir, attemptId, patch) {
488 if (!attemptId || typeof attemptId !== 'string') {
489 throw new Error('attemptId must be a non-empty string');
490 }
491 assertObject(patch, 'attempt patch');
492 const journal = await loadAttemptJournal(projectDir);
493 const current = journal.byAttemptId[attemptId];
494 if (!current) {
495 throw new Error(`unknown attemptId: ${attemptId}`);
496 }
497 const row = {
498 ...current,
499 ...patch,
500 attemptId,
501 };
502 validateAttemptRow(row);
503 await appendNdjson(attemptJournalPath(projectDir), row);
504 return row;
505}
506
507async function rebuildCrosswalkIndexes(projectDir) {
508 const { bySource, byTarget } = await loadCrosswalk(projectDir);
509 await writeJsonAtomic(path.join(crosswalkIndexDir(projectDir), 'by-source.json'), bySource);
510 await writeJsonAtomic(path.join(crosswalkIndexDir(projectDir), 'by-target.json'), byTarget);
511 return { bySource, byTarget };
512}
513
514async function localCrosswalkStateExists(projectDir) {
515 if (!(await pathExists(crosswalkPath(projectDir)))) {
516 return false;
517 }
518 await loadCrosswalk(projectDir);
519 return true;
520}
521
522function newestRow(a, b) {
523 const aTime = Date.parse(a.updatedAt);
524 const bTime = Date.parse(b.updatedAt);
525 if (Number.isNaN(aTime) || Number.isNaN(bTime) || aTime === bTime) {
526 return null;
527 }
528 return bTime > aTime ? b : a;
529}
530
531async function seedCrosswalkFromCmsMirror(projectDir, rows) {
532 if (!Array.isArray(rows)) {
533 throw new Error('CMS mirror rows must be an array');
534 }
535 if (await localCrosswalkStateExists(projectDir)) {
536 return { seeded: false, reason: 'local_crosswalk_exists', accepted: 0, rejected: 0, conflicts: 0 };
537 }
538
539 const rejected = [];
540 const conflicts = [];
541 const bySource = {};
542
543 for (const row of rows) {
544 const validation = validateCrosswalkRow(row, { allowThrow: false, label: 'CMS mirror row' });
545 if (!validation.ok) {
546 rejected.push({ row, errors: validation.errors });
547 continue;
548 }
549
550 const current = bySource[row.sourceStableKey];
551 if (!current) {
552 bySource[row.sourceStableKey] = row;
553 continue;
554 }
555
556 const winner = newestRow(current, row);
557 if (!winner) {
558 if (current.targetId !== row.targetId) {
559 conflicts.push(current, row);
560 }
561 continue;
562 }
563 bySource[row.sourceStableKey] = winner;
564 }
565
566 await mkdirp(cmsMirrorDir(projectDir));
567 await writeTextAtomic(
568 path.join(cmsMirrorDir(projectDir), 'imported-from-cms.ndjson'),
569 rows.map((row) => JSON.stringify(row)).join('\n') + (rows.length ? '\n' : ''),
570 );
571 if (rejected.length) {
572 await writeTextAtomic(
573 path.join(cmsMirrorDir(projectDir), 'rejected.ndjson'),
574 rejected.map((item) => JSON.stringify(item)).join('\n') + '\n',
575 );
576 }
577 if (conflicts.length) {
578 await writeTextAtomic(
579 path.join(cmsMirrorDir(projectDir), 'conflicts.ndjson'),
580 conflicts.map((row) => JSON.stringify(row)).join('\n') + '\n',
581 );
582 throw new Error(`CMS mirror seed has ${conflicts.length} conflicting rows; see state/cms-mirror/conflicts.ndjson`);
583 }
584
585 const acceptedRows = Object.values(bySource).sort((a, b) => a.sourceStableKey.localeCompare(b.sourceStableKey));
586 const text = acceptedRows.map((row) => JSON.stringify(row)).join('\n');
587 await writeTextAtomic(crosswalkPath(projectDir), text ? `${text}\n` : '');
588 await rebuildCrosswalkIndexes(projectDir);
589 return { seeded: true, accepted: acceptedRows.length, rejected: rejected.length, conflicts: 0 };
590}
591
592async function withStateLock(projectDir, fn) {
593 const lockPath = path.join(stateDir(projectDir), '.lock');
594 await mkdirp(stateDir(projectDir));
595 try {
596 await fs.mkdir(lockPath);
597 } catch (error) {
598 if (error && error.code === 'EEXIST') {
599 throw new Error(`state lock already held: ${lockPath}`);
600 }
601 throw error;
602 }
603 try {
604 return await fn();
605 } finally {
606 await fs.rm(lockPath, { recursive: true, force: true });
607 }
608}
609
610module.exports = {
611 SCHEMA_VERSION,
612 stateDir,
613 crosswalkPath,
614 attemptJournalPath,
615 wixRequestCapturesPath,
616 cmsMirrorDir,
617 safeModeDir,
618 safeModeEmailReplacementsPath,
619 safeModeBlockedRecordsPath,
620 dryRunCrosswalkPath,
621 loadCrosswalk,
622 appendCrosswalkRow,
623 upsertCrosswalkRow,
624 loadAttemptJournal,
625 appendAttempt,
626 appendSafeModeEmailReplacement,
627 appendSafeModeBlockedRecord,
628 appendWixRequestCapture,
629 loadWixRequestCaptures,
630 appendDryRunCrosswalkRow,
631 loadDryRunCrosswalk,
632 dryRunUpsertDecision,
633 validateSafeModeEmailReplacementRow,
634 validateSafeModeBlockedRecordRow,
635 validateWixRequestCaptureRow,
636 validateDryRunCrosswalkRow,
637 markAttempt,
638 rebuildCrosswalkIndexes,
639 localCrosswalkStateExists,
640 seedCrosswalkFromCmsMirror,
641 withStateLock,
642};