Setting the file. One moment. 80
function normalizeSelectionFilters
— line 80
This file
- Number
- 47.20
- Position
- 20 of 34
- Type
- JavaScript
- Size
- 11 KB
- Lines
- 313
lib/import-recovery.js
JavaScript·313 lines·11 KB
7
8const RECOVERY_SCHEMA_VERSION = 1;
9const RECOVERY_MODES = ['partial', 'missing-only', 'failed-only', 'deferred-only', 'resumed'];
10const RECOVERY_STATUSES = ['complete', 'partial', 'failed', 'blocked'];
11const DEFERRED_ATTEMPT_STATUSES = new Set(['deferred', 'needs_verification']);
12
13function executionDir(projectDir) {
14 return path.join(projectDir, 'execution');
15}
16
17function recoveryLogPath(projectDir) {
18 return path.join(executionDir(projectDir), 'recovery-log.json');
19}
20
21function liveImportSummaryPath(projectDir) {
22 return path.join(executionDir(projectDir), 'live-import-summary.json');
23}
24
25async function pathExists(filePath) {
26 try {
27 await fs.access(filePath);
28 return true;
29 } catch (error) {
30 if (error && error.code === 'ENOENT') {
31 return false;
32 }
33 throw error;
34 }
35}
36
37async function writeJsonAtomic(filePath, data) {
38 await fs.mkdir(path.dirname(filePath), { recursive: true });
39 const tempPath = `${filePath}.${process.pid}.${Date.now()}.tmp`;
40 await fs.writeFile(tempPath, `${JSON.stringify(data, null, 2)}\n`, 'utf8');
41 await fs.rename(tempPath, filePath);
42}
43
44async function readJsonIfExists(filePath, fallback) {
45 if (!(await pathExists(filePath))) {
46 return fallback;
47 }
48 return JSON.parse(await fs.readFile(filePath, 'utf8'));
49}
50
51function requireString(value, field) {
52 if (!value || typeof value !== 'string') {
53 throw new Error(`${field} must be a non-empty string`);
54 }
55 return value;
56}
57
58function count(value, field) {
59 if (value === undefined || value === null) {
60 return 0;
61 }
62 if (!Number.isInteger(value) || value < 0) {
63 throw new Error(`${field} must be a non-negative integer`);
64 }
65 return value;
66}
67
68function sourceStableKey(sourceSystem, sourceEntityType, sourceId) {
69 return `${sourceSystem}:${sourceEntityType}:${sourceId}`;
70}
71
72function defaultSourceId(record) {
73 return record && (record.id ?? record.ID ?? record.sourceId);
74}
75
76function defaultSourceType(record) {
77 return record && (record.sourceType ?? record.subtype ?? record.type);
78}
79
80function normalizeSelectionFilters(input = {}) {
81 const filters = {
82 entity: input.entity || input.sourceEntityType || null,
83 sourceType: input.sourceType || input.source_type || null,
84 missingOnly: Boolean(input.missingOnly || input['missing-only']),
85 failedOnly: Boolean(input.failedOnly || input['failed-only']),
86 deferredOnly: Boolean(input.deferredOnly || input['deferred-only']),
87 };
88 const exclusive = [filters.missingOnly, filters.failedOnly, filters.deferredOnly].filter(Boolean);
89 if (exclusive.length > 1) {
90 throw new Error('--missing-only, --failed-only, and --deferred-only are mutually exclusive');
91 }
92 return filters;
93}
94
95function foldAttemptsBySource(attemptRows = []) {
96 const bySource = {};
97 for (const row of attemptRows) {
98 if (!row || !row.sourceStableKey) {
99 continue;
100 }
101 bySource[row.sourceStableKey] = row;
102 }
103 return bySource;
104}
105
106function selectImportRecords(records, options = {}) {
107 if (!Array.isArray(records)) {
108 throw new Error('records must be an array');
109 }
110 const filters = normalizeSelectionFilters(options.filters || options);
111 const sourceSystem = options.sourceSystem || 'wordpress';
112 const sourceEntityType = requireString(filters.entity || options.sourceEntityType, 'selection entity');
113 const getSourceId = options.getSourceId || defaultSourceId;
114 const getSourceType = options.getSourceType || defaultSourceType;
115 const crosswalkBySource = options.crosswalkBySource || {};
116 const attemptsBySource = options.attemptsBySource || foldAttemptsBySource(options.attemptRows || []);
117
118 const selected = [];
119 const alreadyPresent = [];
120 const excluded = [];
121 const failed = [];
122 const deferred = [];
123
124 for (const record of records) {
125 const sourceId = getSourceId(record);
126 if (sourceId === undefined || sourceId === null || sourceId === '') {
127 excluded.push({ record, reason: 'missing_source_id' });
128 continue;
129 }
130 const stableKey = sourceStableKey(sourceSystem, sourceEntityType, String(sourceId));
131 const sourceType = getSourceType(record);
132 const existing = crosswalkBySource[stableKey] || null;
133 const latestAttempt = attemptsBySource[stableKey] || null;
134
135 if (filters.sourceType && String(sourceType) !== String(filters.sourceType)) {
136 excluded.push({ sourceStableKey: stableKey, sourceType, reason: 'source_type_filter' });
137 continue;
138 }
139 if (filters.missingOnly && existing) {
140 alreadyPresent.push({ sourceStableKey: stableKey, targetId: existing.targetId, reason: 'crosswalk' });
141 continue;
142 }
143 if (filters.failedOnly && (!latestAttempt || !String(latestAttempt.status).startsWith('failed'))) {
144 excluded.push({ sourceStableKey: stableKey, reason: 'not_failed' });
145 continue;
146 }
147 if (filters.deferredOnly && (!latestAttempt || !DEFERRED_ATTEMPT_STATUSES.has(latestAttempt.status))) {
148 excluded.push({ sourceStableKey: stableKey, reason: 'not_deferred' });
149 continue;
150 }
151
152 if (latestAttempt && String(latestAttempt.status).startsWith('failed')) {
153 failed.push(stableKey);
154 }
155 if (latestAttempt && DEFERRED_ATTEMPT_STATUSES.has(latestAttempt.status)) {
156 deferred.push(stableKey);
157 }
158 selected.push({ record, sourceStableKey: stableKey, sourceType, existing, latestAttempt });
159 }
160
161 return {
162 filters,
163 entity: sourceEntityType,
164 selected,
165 alreadyPresent,
166 excluded,
167 summary: {
168 entity: sourceEntityType,
169 sourceType: filters.sourceType || 'all',
170 recordsRead: records.length,
171 recordsSelected: selected.length,
172 alreadyPresent: alreadyPresent.length,
173 excluded: excluded.length,
174 failedCandidates: failed.length,
175 deferredCandidates: deferred.length,
176 },
177 };
178}
179
180async function selectImportRecordsFromState(projectDir, records, options = {}) {
181 const crosswalk = await loadCrosswalk(projectDir);
182 const attempts = await loadAttemptJournal(projectDir);
183 return selectImportRecords(records, {
184 ...options,
185 crosswalkBySource: crosswalk.bySource,
186 attemptRows: attempts.rows,
187 });
188}
189
190function recoveryIdFor(entry) {
191 if (entry.recoveryId) {
192 return entry.recoveryId;
193 }
194 const hash = crypto.createHash('sha256')
195 .update(JSON.stringify({
196 timestamp: entry.timestamp,
197 selectionFilters: entry.selectionFilters,
198 reason: entry.reason,
199 recordsSelected: entry.recordsSelected,
200 recordsAttempted: entry.recordsAttempted,
201 }))
202 .digest('hex')
203 .slice(0, 12);
204 const stamp = String(entry.timestamp || new Date().toISOString()).replace(/[^0-9TZ]/g, '').slice(0, 15);
205 return `recovery-${stamp}-${hash}`;
206}
207
208function normalizeRecoveryEntry(input = {}) {
209 const mode = input.mode || (input.selectionFilters && (
210 input.selectionFilters.missingOnly ? 'missing-only' :
211 input.selectionFilters.failedOnly ? 'failed-only' :
212 input.selectionFilters.deferredOnly ? 'deferred-only' : 'partial'
213 ));
214 if (!RECOVERY_MODES.includes(mode)) {
215 throw new Error(`recovery mode must be one of: ${RECOVERY_MODES.join(', ')}`);
216 }
217 const status = input.status || 'partial';
218 if (!RECOVERY_STATUSES.includes(status)) {
219 throw new Error(`recovery status must be one of: ${RECOVERY_STATUSES.join(', ')}`);
220 }
221 const timestamp = input.timestamp || new Date().toISOString();
222 if (Number.isNaN(Date.parse(timestamp))) {
223 throw new Error('recovery timestamp must be an ISO timestamp');
224 }
225 const entry = {
226 schemaVersion: RECOVERY_SCHEMA_VERSION,
227 recoveryId: input.recoveryId || null,
228 timestamp,
229 mode,
230 selectionFilters: input.selectionFilters || {},
231 reason: requireString(input.reason || 'operator_requested_recovery', 'recovery reason'),
232 recordsSelected: count(input.recordsSelected, 'recordsSelected'),
233 recordsAttempted: count(input.recordsAttempted, 'recordsAttempted'),
234 imported: count(input.imported, 'imported'),
235 alreadyPresent: count(input.alreadyPresent, 'alreadyPresent'),
236 failed: count(input.failed, 'failed'),
237 deferred: count(input.deferred, 'deferred'),
238 crosswalkChanges: input.crosswalkChanges || { before: 0, after: 0, added: 0, updated: 0 },
239 summaryChanges: input.summaryChanges || {},
240 outcome: requireString(input.outcome || status, 'recovery outcome'),
241 status,
242 logs: Array.isArray(input.logs) ? input.logs : [],
243 };
244 entry.recoveryId = recoveryIdFor(entry);
245 return entry;
246}
247
248async function loadRecoveryLog(projectDir) {
249 const rows = await readJsonIfExists(recoveryLogPath(projectDir), []);
250 if (!Array.isArray(rows)) {
251 throw new Error('execution/recovery-log.json must contain a JSON array');
252 }
253 return rows.map((row) => normalizeRecoveryEntry(row));
254}
255
256async function appendRecoveryLogEntry(projectDir, entry) {
257 const existing = await loadRecoveryLog(projectDir);
258 const normalized = normalizeRecoveryEntry(entry);
259 if (existing.some((row) => row.recoveryId === normalized.recoveryId)) {
260 throw new Error(`duplicate recoveryId: ${normalized.recoveryId}`);
261 }
262 const next = [...existing, normalized];
263 await writeJsonAtomic(recoveryLogPath(projectDir), next);
264 return normalized;
265}
266
267function diffCounts(before = {}, after = {}) {
268 before = before || {};
269 after = after || {};
270 const keys = new Set([...Object.keys(before || {}), ...Object.keys(after || {})]);
271 const diff = {};
272 for (const key of keys) {
273 if (Number.isInteger(before[key]) || Number.isInteger(after[key])) {
274 diff[key] = (after[key] || 0) - (before[key] || 0);
275 }
276 }
277 return diff;
278}
279
280async function writeLiveImportSummary(projectDir, summary, options = {}) {
281 const before = await readJsonIfExists(liveImportSummaryPath(projectDir), null);
282 const next = {
283 schemaVersion: 1,
284 generatedAt: new Date().toISOString(),
285 ...(summary || {}),
286 };
287 await writeJsonAtomic(liveImportSummaryPath(projectDir), next);
288 return {
289 path: 'execution/live-import-summary.json',
290 beforeExists: Boolean(before),
291 changes: {
292 imported: diffCounts(before && before.imported, next.imported),
293 deferred: diffCounts(before && before.deferred, next.deferred),
294 failed: diffCounts(before && before.failed, next.failed),
295 ...(options.extraChanges || {}),
296 },
297 };
298}
299
300module.exports = {
301 RECOVERY_SCHEMA_VERSION,
302 recoveryLogPath,
303 liveImportSummaryPath,
304 normalizeSelectionFilters,
305 foldAttemptsBySource,
306 sourceStableKey,
307 selectImportRecords,
308 selectImportRecordsFromState,
309 normalizeRecoveryEntry,
310 loadRecoveryLog,
311 appendRecoveryLogEntry,
312 writeLiveImportSummary,
313};