Setting the file. One moment.
Read Back Gate · Wix Replatform · wix/skills · Skills Docs
Back to the top of the page async function runGatedEntity
— line 942
This file
Number 45.134
Position 134 of 182
Type JavaScript
Size 50 KB
Lines 1,050 lib/ read-back-gate.js
JavaScript · 1,050 lines · 50 KB
9 const {
10 compareEntity ,
11 memberProjection
12 } = require ( './read-back-comparison' );
13 const {
14 ReadBackReads
15 } = require ( './read-back-reads' );
16 const verified = require ( './write-verification' );
17 const store = require ( './write-verification-state' );
18 const {
19 hashArtifact
20 } = require ( './artifact-freshness' );
21 const writers = require ( '../resources/rp-target-wix/lib/wix-writers' );
22 const wixAuth = require ( './wix-auth.js' );
23 const {
24 deriveImportedOrderId
25 } = require ( '../resources/rp-target-wix/lib/order-history' );
26 const {
27 findMatching
28 } = require ( '../resources/rp-target-wix/lib/contact-import' );
29 const approval = require ( './read-back-approval' );
30 const API = 'https://www.wixapis.com' ;
31 const readSessions = new Map ();
32 const writeContexts = new AsyncLocalStorage ();
33 const writeTrackers = new WeakMap ();
34
35 // The generated wrapper can do reads before it attempts its mutating request. Track only
36 // non-read sends so a writer exception is classified as not-sent only when this invocation
37 // provably made no mutating request. AsyncLocalStorage keeps concurrent batch writers separate.
38 function installWriteTracker ( wix ) {
39 if ( ! wix || ! [ 'object' , 'function' ]. includes ( typeof wix) || typeof wix.send !== 'function' ) return null ;
40 const active = writeTrackers. get (wix);
41 if (active) {
42 active.references ++ ;
43 let released = false ;
44 return () => {
45 if (released) return ;
46 released = true ;
47 active.references -- ;
48 if ( ! active.references) {
49 writeTrackers. delete (wix);
50 if (wix.send === active.trackedSend) wix.send = active.original;
51 }
52 };
53 }
54 const original = wix.send;
55 const tracker = {
56 original,
57 references: 1 ,
58 trackedSend: null
59 };
60 tracker. trackedSend = function trackedSend ( request , ... args ) {
61 const context = writeContexts. getStore ();
62 // The generated writers keep their `{ read: true }` mark in a private WeakMap (`requestMark`), not on
63 // the request, so ask the writers; `request.read` still counts for hand-built requests.
64 if (context && ! (request?.read || writers. requestMark ?.(request)?.read)) context.requestSent = true ;
65 return original. call ( this , request, ... args);
66 };
67 try {
68 wix.send = tracker.trackedSend;
69 } catch {
70 return null ;
71 }
72 writeTrackers. set (wix, tracker);
73 let released = false ;
74 return () => {
75 if (released) return ;
76 released = true ;
77 tracker.references -- ;
78 if ( ! tracker.references) {
79 writeTrackers. delete (wix);
80 if (wix.send === tracker.trackedSend) wix.send = original;
81 }
82 };
83 }
84 function sharedReads ( projectDir , runId , wix ) {
85 const key = `${ path . resolve ( projectDir ) } \0 ${ runId }` ;
86 if ( ! readSessions. has (key)) readSessions. set (key, new ReadBackReads (wix));
87 return readSessions. get (key);
88 }
89 // Every shape a writer may use to report the target it wrote. The success path and the
90 // deferral check share it, so a deferral can never hide an id the success path would accept.
91 function writtenTargetId ( entity , written ) {
92 return written?.targetId || written?.id || written?.[ `${ entity }Id` ] || written?.[entity]?.id;
93 }
94 function problem ( reason ) {
95 return Object. assign ( new Error (reason), {
96 code: reason
97 });
98 }
99 // Writer guards approved elsewhere that remove a value before sending (the member
100 // phone-collision drop). Only these exact finding -> path pairs can exempt a path, and only when
101 // the writer itself reported the finding; a missing target field never implies a guard.
102 const WRITER_GUARDS = {
103 member: [{
104 path: 'member.contact.phones[]' ,
105 applies : f => f?.code === 'member-phone-collision-dropped' && f.phoneDropped === true ,
106 strip : body => {
107 delete body.member?.contact?.phones;
108 }
109 }]
110 };
111 function guardPathsFrom ( entity , findings ) {
112 return ( WRITER_GUARDS [entity] || []). filter ( g => (findings || []). some (g.applies)). map ( g => g.path);
113 }
114 function applyGuards ( entity , expected , paths ) {
115 const guards = ( WRITER_GUARDS [entity] || []). filter ( g => (paths || []). includes (g.path));
116 if ( ! guards. length ) return expected;
117 const body = structuredClone (expected);
118 for ( const g of guards) g. strip (body);
119 return body;
120 }
121 function declaration ( entity , {
122 verificationKind ,
123 reason
124 } = {}) {
125 if (Object. hasOwn ( C . DOMAINS , entity)) return {
126 entity,
127 gated: true
128 };
129 if ( ! verificationKind || ! reason?. trim ()) throw new Error ( 'Non-gated entities must declare verification and reason' );
130 return {
131 entity,
132 gated: false ,
133 verificationKind,
134 reason
135 };
136 }
137 class ReadBackGate {
138 constructor ({
139 projectDir ,
140 runId ,
141 entity ,
142 siteId ,
143 wix ,
144 buildWrite ,
145 write ,
146 repair ,
147 sourceSiteHost ,
148 marker ,
149 variants = [],
150 reads ,
151 read ,
152 lookup ,
153 now = () => new Date (). toISOString (),
154 identity
155 } = {}) {
156 if ( ! projectDir || typeof runId !== 'string' || ! / ^ [A-Za-z0-9][A-Za-z0-9_.-] {0,127}$ / . test (runId) || [ 'constructor' , 'prototype' , '__proto__' ]. includes (runId) || ! siteId || typeof buildWrite !== 'function' || typeof write !== 'function' ) throw new Error ( 'Read-back gate requires project, run, site, builder and writer' );
157 C . entityKey (entity);
158 Object. assign ( this , {
159 projectDir,
160 runId,
161 entity,
162 siteId,
163 wix,
164 buildWrite,
165 write,
166 repair,
167 sourceSiteHost,
168 marker,
169 variants,
170 now,
171 identity
172 });
173 this .reads = reads || sharedReads (projectDir, runId, wix);
174 this .startedAt = this .reads. now ();
175 this .preparationMs = 0 ;
176 this .writeMs = 0 ;
177 this .read = read || ( id => this . readTarget (id));
178 this .lookup = lookup || (( record , expected ) => this . lookupTarget (record, expected));
179 }
180 async mutate ( fn ) {
181 return state. update ( this .projectDir, g => {
182 g.runs[ this .runId] ||= {
183 entities: {}
184 };
185 g.runs[ this .runId].entities[ this .entity] ||= {
186 status: 'stopped' ,
187 canaries: [],
188 settlements: [],
189 approvedOmissions: []
190 };
191 const item = g.runs[ this .runId].entities[ this .entity],
192 result = fn (g, item);
193 item.retryStats = {
194 ... this .reads.stats,
195 remainingMs: this .reads. remaining ()
196 };
197 item.timing = {
198 elapsedMs: Math. max ( 0 , this .reads. now () - this .startedAt),
199 preparationMs: this .preparationMs,
200 writeMs: this .writeMs,
201 activeReadMs: Object. entries ( this .reads.stats). filter (([ key ]) => key. endsWith ( 'Ms' )). reduce (( n , [, value ]) => n + value, 0 )
202 };
203 return result;
204 });
205 }
206 async item () {
207 return ( await state. load ( this .projectDir)).runs[ this .runId]?.entities?.[ this .entity];
208 }
209 async prepare ( inv , {
210 ceiling ,
211 preparationMs = inv?.preparationMs || 0
212 } = {}) {
213 const preparingAt = this .reads. now ();
214 await approval. recoverAcknowledgements ( this .projectDir, this .identity ? {
215 identity: this .identity
216 } : {});
217 this .validated = await C . validate ( this .projectDir, this .entity, {
218 inv
219 });
220 if ( ! this .validated.valid) {
221 await this . stop ( 'mapping-validation-failed' );
222 throw Object. assign ( problem ( 'mapping-validation-failed' ), {
223 errors: this .validated.errors
224 });
225 }
226 const decisions = await state. readJson ( this .projectDir, 'orchestration/decisions.json' , {}),
227 chosen = decisions[ `readBackCanaryCeiling.${ this . entity }` ] || decisions.readBackCanaryCeiling?.[ this .entity] || decisions.decisions?.[ `readBackCanaryCeiling.${ this . entity }` ];
228 if (ceiling !== undefined ) throw new Error ( 'Canary ceiling must come from recorded user decision, not the caller' );
229 if (chosen && (chosen.source !== 'user' || ! Number. isInteger (chosen.value))) throw problem ( 'canary-ceiling-not-user-recorded' );
230 const history = await state. load ( this .projectDir);
231 // Only a prepared entry (schedule and bindings) is a previous run to continue. An unprepared
232 // one (completion's `gate-not-run` placeholder, a validation stop) is taken over by this run;
233 // it stays blocking until then, and this run's own gate status replaces it.
234 const others = Object. entries (history.runs). filter (([ id , run ]) => id !== this .runId && run.entities?.[ this .entity] && ! run.entities[ this .entity].supersededBy);
235 const preparedEntry = e => Boolean (e.schedule && e.bindingDigests);
236 const previousRun = [ ... others]. reverse (). find (([, run ]) => preparedEntry (run.entities[ this .entity]));
237 const unprepared = others. filter (([, run ]) => ! preparedEntry (run.entities[ this .entity])). map (([ id ]) => id);
238 const own = await this . item ();
239 const existing = own?.schedule ? own : previousRun?.[ 1 ].entities[ this .entity];
240 // A skip belongs to the run that made it: a parent may have landed since, so a later run offers
241 // the record again (and skips it again if its parent is still final). This run's own skips hold.
242 const priorSkips = existing?.skippedCanaries || [], keptSkips = priorSkips. filter ( r => r.runId === this .runId);
243 const schedule = C . cover (inv, {
244 ceiling: chosen?.value || 10 ,
245 existing: (existing?.canaries || []). map ( r => r.sourceId),
246 exclude: keptSkips. map ( r => r.sourceId)
247 });
248 this .inventory = inv;
249 this .repairContractDigest = C . digest ({
250 bindings: this .validated.bindingDigests,
251 code: hashArtifact ( this .projectDir, 'src/import' ),
252 comparator: 1
253 });
254 this .preparationMs += preparationMs + Math. max ( 0 , this .reads. now () - preparingAt);
255 const approvedOmissionRows = this .validated.fieldContractVersion === 1 ? require ( './mapping-omission-approval' ). omissions ( this .validated.decision) : null ;
256 await this . mutate (( g , item ) => {
257 if ( ! own?.schedule && existing) {
258 Object. assign (item, structuredClone (existing));
259 delete item.supersededBy;
260 if (previousRun) g.runs[previousRun[ 0 ]].entities[ this .entity].supersededBy = this .runId;
261 }
262 for ( const id of unprepared) g.runs[id].entities[ this .entity].supersededBy = this .runId;
263 if (existing && C . digest (existing.bindingDigests) !== C . digest ( this .validated.bindingDigests)) {
264 item.approvedOmissions = [];
265 delete item.pendingAcknowledgementRequest;
266 }
267 item.paths ||= {};
268 item.paths.notExercised = this .validated.contract.intendedFields. filter ( f => ! ( this .validated.decision.fieldDispositions || []). some ( d => [ 'notMigrated' , 'deferred' ]. includes (d.status) && (d.path || d.targetPath) === f.targetPath)). filter ( f => ! Object. keys (inv.writePaths). some ( p => p === f.targetPath || p. startsWith (f.targetPath + '.' ) || p. startsWith (f.targetPath + '[]' ))). map ( f => f.targetPath);
269 item.pendingFields = ( this .validated.decision.fieldDispositions || []). filter ( d => d.status === 'deferred' );
270 // The schedule is the fresh one, not a persisted one that excluded a record skipped by an earlier run.
271 const retriedSkips = priorSkips. length > keptSkips. length ;
272 if (keptSkips. length ) item.skippedCanaries = structuredClone (keptSkips); else delete item.skippedCanaries;
273 item.schedule = existing?.schedule?.universeDigest === schedule.universeDigest && ! retriedSkips ? existing.schedule : schedule;
274 item.status = 'stopped' ;
275 item.stopReason = schedule.status === 'covered' ? 'canaries-pending' : schedule.status;
276 item.retryStats = this .reads.stats;
277 item.invocationStartedAt = this . now ();
278 item.bindingDigests = this .validated.bindingDigests;
279 if (inv.heldRecords) item.heldRecords = structuredClone (inv.heldRecords); else delete item.heldRecords;
280 if ( ! inv.records. length ) item.endOfRunVerified = true ;
281 if ( ! item.approvedOmissions. length ) item.approvedOmissions = ( this .validated.decision.fieldDispositions || []). filter ( d => d.status === 'notMigrated' &&
282 ( ! approvedOmissionRows || approvedOmissionRows. some ( o => o.kind === d.kind && o.path === d.path && o.writeOperation === (d.writeOperation || null )))). map ( d => ({
283 path: d.path || d.sourcePath || d.targetPath,
284 kind: d.kind || 'source' ,
285 recordCount: d.affectedScope ? d.affectedScope.count : inv.sourcePaths[d.path || d.sourcePath] || inv.records. length ,
286 ... (d.affectedScope ? {affectedScope:d.affectedScope,writeOperation:d.writeOperation || null } : {}),
287 approvedBy: this .validated.approval?.approvedBy || null
288 }));
289 });
290 if (schedule.status !== 'covered' ) throw Object. assign ( problem (schedule.status), {
291 uncoveredPaths: schedule.uncoveredPaths
292 });
293 return schedule;
294 }
295 async protect ( sourceId , change ) {
296 return state. locked ( this .projectDir, async () => {
297 const s = await store. load ( this .projectDir, this .siteId),
298 key = `read-back:${ this . entity }:${ sourceId }` ;
299 s.rows[key] = {
300 ... (s.rows[key] || {}),
301 sourceKey: key,
302 kind: 'read-back-gate' ,
303 entity: this .entity,
304 sourceId,
305 runId: this .runId,
306 ... change
307 };
308 await store. checkpoint ( this .projectDir, s, [{
309 sourceKey: key
310 }]);
311 return s.rows[key];
312 });
313 }
314 async protected ( sourceId ) {
315 const s = await store. load ( this .projectDir, this .siteId);
316 return s.rows[ `read-back:${ this . entity }:${ sourceId }` ];
317 }
318 async stop ( reason ) {
319 await this . mutate (( g , item ) => {
320 if (reason === 'read-budget-exhausted' ) for ( const e of Object. values (g.runs[ this .runId].entities)) {
321 if (e.status !== 'passed' ) {
322 e.status = 'unknown' ;
323 e.stopReason = reason;
324 }
325 }
326 item.status = reason. includes ( 'unknown' ) || /budget | settlement | unavailable/ . test (reason) ? 'unknown' : 'stopped' ;
327 item.stopReason = reason;
328 item.retryStats = {
329 ... this .reads.stats,
330 remainingMs: this .reads. remaining ()
331 };
332 });
333 }
334 async readTarget ( id ) {
335 const w = this .reads.wix;
336 if ( this .entity === 'product' ) return writers. getStoresProductForVerification (w, id);
337 if ( this .entity === 'contact' ) return ( await w. send (writers. buildGetContactRequest (id))).contact;
338 if ( this .entity === 'order' ) return ( await w. send ({
339 method: 'GET' ,
340 url: `${ API }/ecom/v1/orders/${ encodeURIComponent ( id ) }`
341 })).order;
342 const m = ( await w. send (writers. buildGetMemberRequest (id))).member;
343 if ( ! m?.contactId) throw problem ( 'member-contact-id-unknown' );
344 const c = ( await w. send (writers. buildGetContactRequest (m.contactId))).contact;
345 if (c?.id !== m.contactId) throw problem ( 'member-contact-identity-unknown' );
346 return {
347 member: m,
348 contact: c
349 };
350 }
351 async lookupTarget ( record , body ) {
352 const e = body[ this .entity] || body,
353 w = this .reads.wix;
354 if ( this .entity === 'order' ) {
355 const id = deriveImportedOrderId ({
356 sourceSiteHost: this .sourceSiteHost,
357 sourceId: String (record.id)
358 });
359 try {
360 return [ await this . readTarget (id)];
361 } catch (err) {
362 if (err.status === 404 && err.outcomeKnown) return [];
363 throw err;
364 }
365 }
366 if ( this .entity === 'contact' ) return findMatching (w, e);
367 if ( this .entity === 'member' ) return ( await writers. queryMembersByLoginEmail (w, e.loginEmail)). filter ( m => verified. email (m.loginEmail) === verified. email (e.loginEmail));
368 const products = await writers. queryAllStoresProducts (w),
369 marker = typeof this .marker === 'function' ? this . marker (record, body) : this .marker;
370 if ( ! marker?.path) throw problem ( 'product-marker-required' );
371 const matches = products. filter ( p => C . valuesAt (p, marker.path). some ( v => v === marker.value));
372 // A product that was on the store before this import is never this record's write.
373 const baseline = require ( './catalog-baseline' ),
374 held = matches. find ( p => baseline. protectedIds ( this .projectDir).products. has (p.id));
375 if (held) {
376 baseline. recordCollision ( this .projectDir, { entity: 'stores/product' , sourceKey: `read-back:${ this . entity }:${ record . id }` , targetId: held.id, slug: held.slug ?? null });
377 throw problem ( 'target-baseline-collision' );
378 }
379 return matches;
380 }
381 checkIdentity ( expected , actual , id ) {
382 const e = expected[ this .entity] || expected,
383 a = actual[ this .entity] || actual;
384 if ( ! id || a?.id !== id || a?._dryRunPlaceholder) throw problem ( 'identity-conflict' );
385 if ( this .entity === 'contact' && ! verified. contactIdentity (e, a, id).verified) throw problem ( 'identity-conflict' );
386 if ( this .entity === 'product' && ! verified. verifyProduct ({
387 id
388 }, a, id, {
389 marker: typeof this .marker === 'function' ? this . marker ( this .activeRecord, expected) : this .marker,
390 variants: typeof this .variants === 'function' ? this . variants ( this .activeRecord, expected) : this .variants
391 }).verified) throw problem ( 'identity-conflict' );
392 if ( this .entity === 'member' && (verified. email (e.loginEmail) !== verified. email (a.loginEmail) || ! a.contactId || actual.contact?.id !== a.contactId)) throw problem ( 'member-contact-identity-unknown' );
393 if ( this .entity === 'order' && (a.number !== e.number || id !== deriveImportedOrderId ({
394 sourceSiteHost: this .sourceSiteHost,
395 sourceId: String ( this .activeSourceId)
396 }))) throw problem ( 'identity-conflict' );
397 }
398 assertApplicability ( record , expected , guarded , options = {}) {
399 if ( C . builderHold (expected)) throw problem ( 'builder-held-record' );
400 const judged = C . outputApplicability ( this .validated.rows, record, expected, { guarded: guarded || [], ... options });
401 if (judged.missing. length ) throw problem ( 'builder-missing-intended-field' );
402 if (judged.unexpected. length ) throw problem ( 'builder-emitted-inapplicable-field' );
403 }
404 async observe ( record , sourceId , targetId , expected ) {
405 // Read-back compares only applicable outputs: an expected body carrying a non-applicable (or
406 // missing a declared applicable) output is refused before any read.
407 this . assertApplicability (record, expected, ( WRITER_GUARDS [ this .entity] || []). map ( g => g.path), { declaredOnly: true });
408 // No target id means nothing was written to read: refused before any request (a read of
409 // `.../undefined` only spends the read budget on 404s).
410 if ( noTarget (targetId)) throw Object. assign ( problem ( 'target-id-missing' ), { requestSent: false });
411 this .activeSourceId = sourceId;
412 this .activeRecord = record;
413 let actual = await this . read (targetId, this .reads.wix);
414 this . checkIdentity (expected, actual, targetId);
415 let result = compareEntity ( this .entity, expected, actual),
416 media = result.checks. filter ( c => ! c.matches && c.writePath. startsWith ( 'product.media.' ));
417 if (media. length ) {
418 let settled = false ,
419 reason = 'media-settlement-exhausted' ;
420 try {
421 await this .reads. settle ();
422 actual = await this . read (targetId, this .reads.wix);
423 this . checkIdentity (expected, actual, targetId);
424 result = compareEntity ( this .entity, expected, actual);
425 settled = ! result.checks. some ( c => ! c.matches && c.writePath. startsWith ( 'product.media.' ));
426 } catch (e) {
427 if (wixAuth. isRunInterruption (e)) throw e;
428 reason = e.code || 'media-settlement-exhausted' ;
429 }
430 await this . mutate (( g , item ) => {
431 for ( const check of media) {
432 const row = this . rowFor (check.writePath),
433 entry = {
434 entity: this .entity,
435 sourceId,
436 targetId,
437 readPath: check.readPath,
438 writePath: check.writePath,
439 sourcePaths: row?.sourcePaths || [],
440 intendedTargetPath: row?.writePath || check.writePath,
441 bindingSetDigest: C . digest ( this .validated.bindingDigests),
442 policy: 'product-media-read-after-60s' ,
443 status: settled ? 'passed' : 'unknown' ,
444 reads: 2 ,
445 waitedMs: this .reads.stats.settlementMs,
446 reason: settled ? null : reason
447 };
448 item.settlements = item.settlements. filter ( s => ! (s.sourceId === sourceId && s.writePath === check.writePath));
449 item.settlements. push (entry);
450 }
451 });
452 if ( ! settled) {
453 await this . recordLosses (sourceId, result.checks. filter ( c => ! c.writePath. startsWith ( 'product.media.' )));
454 await this . stop (reason);
455 return {
456 verified: false ,
457 unknown: true ,
458 actual,
459 checks: result.checks
460 };
461 }
462 }
463 await this . recordLosses (sourceId, result.checks);
464 // A record held earlier as a baseline collision that now verifies is no longer one.
465 if (result.verified && this .entity === 'product' ) require ( './catalog-baseline' ). clearCollision ( this .projectDir, 'stores/product' , `read-back:${ this . entity }:${ sourceId }` );
466 if ( ! result.verified) await this . stop ( 'field-mismatch' ); else await this . mutate (( g , item ) => {
467 for ( const s of item.settlements. filter ( s => s.sourceId === sourceId)) s.status = 'passed' ;
468 for ( const loss of g.observedLosses) if (loss.entity === this .entity && result.checks. some ( c => c.matches && c.writePath === loss.writePath)) loss.resolvedBy = {
469 sourceId,
470 targetId,
471 at: this . now ()
472 };
473 });
474 return {
475 ... result,
476 actual
477 };
478 }
479 rowFor ( writePath ) {
480 const rows = this .validated.rows. filter ( r => r.writePath === writePath || writePath. startsWith ( `${ r . writePath }.` ) || writePath. startsWith ( `${ r . writePath }[]` ) || r.writePath. startsWith ( `${ writePath }[]` )). sort (( a , b ) => b.writePath. length - a.writePath. length );
481 return rows. length ? {
482 ... rows[ 0 ],
483 sourcePaths: [ ...new Set (rows. flatMap ( r => r.sourcePaths || []))]
484 } : null ;
485 }
486 async recordLosses ( sourceId , checks ) {
487 await this . mutate (( g , item ) => {
488 for ( const check of checks) {
489 if (check.matches) continue ;
490 const row = this . rowFor (check.writePath);
491 if ( ! g.observedLosses. some ( l => l.entity === this .entity && l.writePath === check.writePath)) g.observedLosses. push ({
492 entity: this .entity,
493 writePath: check.writePath,
494 readPath: check.readPath,
495 sourcePaths: row?.sourcePaths || [],
496 intendedTargetPath: row?.writePath || check.writePath,
497 kind: check.kind,
498 firstSourceId: sourceId,
499 bindingSetDigest: C . digest ( this .validated.bindingDigests),
500 observedAt: this . now ()
501 });
502 }
503 item.checks = checks. map ( c => ({
504 ... c
505 }));
506 for ( const c of checks) {
507 const key = c.kind === 'unverifiable' ? 'unverifiable' : c.matches ? 'verified' : c.kind;
508 (item.paths ||= {})[key] ||= [];
509 if ( ! item.paths[key]. includes (c.writePath)) item.paths[key]. push (c.writePath);
510 }
511 });
512 }
513 async timeWrite ( write ) {
514 const started = this .reads. now ();
515 const context = {
516 requestSent: false
517 };
518 try {
519 const run = this .writeTracking ? writeContexts. run (context, write) : write ();
520 return await run;
521 } catch (error) {
522 // A generated writer may fail during readiness/reference validation before it reaches
523 // Wix. The gate can safely settle that refusal only when its send tracker saw no
524 // mutating request; preserve an explicit writer classification when one exists.
525 if ( this .writeTracking && ! context.requestSent && error && typeof error === 'object' && error.requestSent === undefined ) error.requestSent = false ;
526 throw error;
527 } finally {
528 this .writeMs += Math. max ( 0 , this .reads. now () - started);
529 }
530 }
531 async withWriteTracking ( run ) {
532 const restore = installWriteTracker ( this .wix);
533 if ( ! restore) return run ();
534 this .writeTracking = true ;
535 try {
536 return await run ();
537 } finally {
538 this .writeTracking = false ;
539 restore ();
540 }
541 }
542 async executeCanaries ( getRecord ) {
543 if ( ! this .validated?.valid) throw problem ( 'gate-not-prepared' );
544 let item = await this . item ();
545 if (item.schedule.status !== 'covered' ) throw problem ( 'coverage-exceeds-ceiling' );
546 for ( const sourceId of item.schedule.canarySourceIds) {
547 if ( this .reads. remaining () <= 0 ) {
548 await this . stop ( 'read-budget-exhausted' );
549 return false ;
550 }
551 const record = await getRecord (sourceId),
552 prior = await this . protected (sourceId);
553 let expected = await this . buildWrite (record),
554 targetId = prior?.targetId,
555 newWrite = false ;
556 // A previous attempt the client refused before sending left nothing on the site: write again.
557 const notSent = prior?.outcome === 'not-sent' && ! prior.targetId;
558 await this . mutate (( g , e ) => {
559 if ( ! e.canaries. some ( c => c.sourceId === sourceId)) e.canaries. push ({
560 sourceId,
561 targetId: prior?.targetId || null ,
562 marker: prior?.marker,
563 outcome: prior?.outcome || 'pending' ,
564 repairAttempts: prior?.repairAttempts || []
565 });
566 });
567 expected = applyGuards ( this .entity, expected, prior?.guardDroppedPaths);
568 // Every applicable populated mapping must produce an expected target value, and an output whose
569 // declared appliesWhen is false must be absent. The builder's own omission proves nothing.
570 this . assertApplicability (record, expected, prior?.guardDroppedPaths);
571 if (prior && ! targetId && ! notSent) {
572 let found;
573 try {
574 found = await this . lookup (record, expected, this .reads.wix);
575 } catch (e) {
576 if (wixAuth. isRunInterruption (e)) throw e;
577 await this . stop (e.code || 'unknown-write-lookup-failed' );
578 return false ;
579 }
580 const matches = Array. isArray (found) ? found : found?.matches || [];
581 if (matches. length !== 1 ) {
582 await this . stop (matches. length > 1 ? 'identity-conflict' : 'unknown-write-no-match' );
583 return false ;
584 }
585 targetId = matches[ 0 ].id;
586 await this . protect (sourceId, {
587 targetId,
588 uncertain: false
589 });
590 }
591 if ( ! prior || notSent) {
592 await this . mutate (( g , e ) => {
593 if ( ! e.canaries. some ( c => c.sourceId === sourceId)) e.canaries. push ({
594 sourceId,
595 targetId: null ,
596 outcome: 'pending' ,
597 repairAttempts: []
598 });
599 });
600 await this . mutate (( g , e ) => {
601 e.attemptedPaths ||= [];
602 for ( const row of this .validated.rows) if ( C . valuesAt (expected, row.writePath). some ( C .populated) && ! e.attemptedPaths. some ( p => p.writePath === row.writePath)) e.attemptedPaths. push ({
603 entity: this .entity,
604 sourceId,
605 writePath: row.writePath,
606 sourcePaths: row.sourcePaths,
607 intendedTargetPath: row.writePath,
608 bindingSetDigest: C . digest ( this .validated.bindingDigests)
609 });
610 });
611 await this . protect (sourceId, {
612 expected,
613 payloadDigest: C . digest (expected),
614 contractDigest: this .repairContractDigest,
615 sentAt: this . now (),
616 outcome: 'pending' ,
617 uncertain: true
618 });
619 try {
620 const written = await this . timeWrite (() => this . write (record, expected));
621 // The writer declined before sending (`{ deferred: true, reason }`, e.g. its dependency
622 // gate deferred the record): nothing reached the site, so it is not a possible write.
623 // Only the exact documented shape is trusted. Any id alias the success path accepts, or
624 // any other key, means a write may exist: keep the protection (unknown outcome).
625 if (written?.deferred === true ) {
626 if ( writtenTargetId ( this .entity, written)) throw problem ( 'unknown-write-deferred-with-id' );
627 if (Object. keys (written). some ( key => key !== 'deferred' && key !== 'reason' )) throw problem ( 'unknown-write-deferral-malformed' );
628 await this . protect (sourceId, {
629 uncertain: false ,
630 sentAt: null ,
631 outcome: 'not-sent' ,
632 notSentReason: typeof written.reason === 'string' && written.reason ? written.reason : 'write-deferred'
633 });
634 // Its own parent record is final for this run (dependency-parent-final): this record
635 // cannot be a canary now. Another record covers its paths; it imports later with the
636 // rest of the entity once its parent lands. Only a never-written canary is replaced.
637 if (written.reason === 'dependency-parent-final' && ! prior?.targetId && this .inventory) {
638 let next;
639 await this . mutate (( g , e ) => {
640 e.skippedCanaries = [ ... (e.skippedCanaries || []). filter ( row => row.sourceId !== sourceId), { sourceId, reason: written.reason, runId: this .runId }];
641 e.canaries = e.canaries. filter ( row => row.sourceId !== sourceId);
642 next = C . cover ( this .inventory, { ceiling: e.schedule.ceiling, existing: e.canaries. map ( row => row.sourceId), exclude: e.skippedCanaries. map ( row => row.sourceId) });
643 e.schedule = next;
644 });
645 if (next.status !== 'covered' ) { await this . stop (next.status); return false ; }
646 return this . executeCanaries (getRecord);
647 }
648 await this . stop ( 'canary-write-deferred' );
649 return false ;
650 }
651 targetId = writtenTargetId ( this .entity, written);
652 const guarded = await this . recordGuard (sourceId, written?.findings);
653 if (guarded. length ) expected = applyGuards ( this .entity, expected, guarded);
654 if ( ! targetId) throw problem ( 'unknown-write-no-confirmed-id' );
655 await this . protect (sourceId, {
656 targetId,
657 uncertain: false
658 });
659 newWrite = true ;
660 } catch (e) {
661 if (wixAuth. isRunInterruption (e)) throw e;
662 // Refused before sending (a capability or readiness refusal): the outcome is known, so
663 // the record must not stay protected as possibly written.
664 if (e?.requestSent === false && ! targetId) await this . protect (sourceId, {
665 uncertain: false ,
666 sentAt: null ,
667 outcome: 'not-sent' ,
668 notSentReason: e.code || e.message
669 });
670 await this . stop (e.code || 'unknown-write-outcome' );
671 return false ;
672 }
673 }
674 let result;
675 try {
676 result = await this . observe (record, sourceId, targetId, expected);
677 } catch (e) {
678 if (wixAuth. isRunInterruption (e)) throw e;
679 await this . stop (e.code || 'unknown-read' );
680 return false ;
681 }
682 const actual = result.actual[ this .entity] || result.actual,
683 marker = actual.revision ?? actual.updatedDate;
684 if ( ! result.verified && ! result.unknown && prior) {
685 const previous = ( await this . item ()).canaries. find ( c => c.sourceId === sourceId),
686 attempted = previous?.repairAttempts?. some ( a => a.contractDigest === this .repairContractDigest);
687 if ( this .entity !== 'order' && this .repair && prior.contractDigest !== this .repairContractDigest && ! attempted && previous?.marker !== undefined && previous.marker === marker) {
688 await this . mutate (( g , e ) => {
689 e.canaries. find ( c => c.sourceId === sourceId).repairAttempts. push ({
690 contractDigest: this .repairContractDigest,
691 markerBefore: marker,
692 outcome: 'pending' ,
693 at: this . now ()
694 });
695 });
696 await this . protect (sourceId, {
697 repairAttempts: ( await this . item ()).canaries. find ( c => c.sourceId === sourceId).repairAttempts
698 });
699 try {
700 await this . timeWrite (() => this . repair ({
701 record,
702 expected,
703 targetId,
704 marker
705 }));
706 result = await this . observe (record, sourceId, targetId, expected);
707 newWrite = true ;
708 } catch (e) {
709 if (wixAuth. isRunInterruption (e)) throw e;
710 await this . stop (e.code || 'unknown-repair-outcome' );
711 return false ;
712 }
713 await this . mutate (( g , e ) => {
714 const c = e.canaries. find ( c => c.sourceId === sourceId);
715 c.repairAttempts. find ( a => a.contractDigest === this .repairContractDigest).outcome = result.unknown ? 'unknown' : result.verified ? 'repaired' : 'still-mismatching' ;
716 });
717 } else await this . mutate (( g , e ) => {
718 e.listed ||= [];
719 e.listed. push ({
720 sourceId,
721 targetId,
722 reason: this .entity === 'order' ? 'order-not-rewritten' : attempted ? 'repair-already-attempted' : 'changed-since-write'
723 });
724 });
725 }
726 await this . protect (sourceId, {
727 expected,
728 targetId,
729 ... (newWrite ? {
730 contractDigest: this .repairContractDigest
731 } : {}),
732 outcome: result.verified ? 'verified' : result.unknown ? 'unknown' : 'mismatch' ,
733 repairAttempts: ( await this . item ()).canaries. find ( c => c.sourceId === sourceId).repairAttempts,
734 ... (newWrite ? {
735 marker: (result.actual[ this .entity] || result.actual).revision ?? (result.actual[ this .entity] || result.actual).updatedDate
736 } : {})
737 });
738 await this . mutate (( g , e ) => {
739 const c = e.canaries. find ( c => c.sourceId === sourceId);
740 c.targetId = targetId;
741 c.outcome = result.verified ? 'verified' : result.unknown ? 'unknown' : 'mismatch' ;
742 if (newWrite) {
743 const v = result.actual[ this .entity] || result.actual;
744 c.marker = v.revision ?? v.updatedDate;
745 }
746 e.retryStats = {
747 ... this .reads.stats,
748 remainingMs: this .reads. remaining ()
749 };
750 });
751 if ( ! result.verified) return false ;
752 }
753 if (( await this . item ()).suspectInterval) {
754 await this . stop ( 'suspect-reconciliation-required' );
755 return false ;
756 }
757 await this . mutate (( g , e ) => {
758 e.status = 'passed' ;
759 e.stopReason = null ;
760 e.listed = [];
761 e.resumeAfter = e.schedule.canarySourceIds. at ( - 1 ) || null ;
762 });
763 await approval. createAcknowledgement ( this .projectDir, this .runId, this .entity);
764 return true ;
765 }
766 // Durably records an approved writer guard for one source record, from the findings the writer
767 // returned for that record's write. Bulk imports call this after each write that returned
768 // findings (only a recognized guard finding persists anything). Returns the guarded paths.
769 async recordGuard ( sourceId , findings ) {
770 const paths = guardPathsFrom ( this .entity, findings);
771 if ( ! paths. length ) return [];
772 sourceId = String (sourceId);
773 const prior = await this . protected (sourceId);
774 await this . protect (sourceId, {
775 guardDroppedPaths: [ ...new Set ([ ... (prior?.guardDroppedPaths || []), ... paths])]
776 });
777 await this . mutate (( g , e ) => {
778 e.guardDropped ||= [];
779 for ( const p of paths) if ( ! e.guardDropped. some ( d => d.sourceId === sourceId && d.path === p)) e.guardDropped. push ({
780 sourceId,
781 path: p,
782 kind: 'guard-dropped'
783 });
784 e.attemptedPaths = (e.attemptedPaths || []). filter ( a => ! (a.sourceId === sourceId && paths. includes (a.writePath)));
785 });
786 return paths;
787 }
788 // Expected body for a read-only check: rebuilt from the source record, minus only the paths a
789 // recorded writer guard dropped for this record. Every other field stays loss-checked.
790 async expectedFor ( record , sourceId ) {
791 return applyGuards ( this .entity, await this . buildWrite (record), ( await this . protected ( String (sourceId)))?.guardDroppedPaths);
792 }
793 async verifySample ({
794 record ,
795 sourceId ,
796 targetId ,
797 index ,
798 isLast = false ,
799 findings
800 }) {
801 if ( ! Number. isInteger (index) || index < 1 ) throw new Error ( 'Sample index is one-based stable source order' );
802 if (findings) await this . recordGuard (sourceId, findings);
803 if (index % 200 !== 0 && ! isLast) return {
804 skipped: true
805 };
806 const expected = await this . expectedFor (record, sourceId);
807 let result;
808 try {
809 result = await this . observe (record, String (sourceId), targetId, expected);
810 } catch (e) {
811 if (wixAuth. isRunInterruption (e)) throw e;
812 result = {
813 verified: false ,
814 unknown: true ,
815 reason: e.code || 'unknown-read'
816 };
817 await this . stop (result.reason);
818 }
819 await this . protect ( String (sourceId), {
820 expected,
821 actual: result.actual || null ,
822 targetId,
823 verificationType: 'sample' ,
824 outcome: result.verified ? 'verified' : 'unknown'
825 });
826 await this . mutate (( g , e ) => {
827 if (isLast) e.endOfRunVerified = result.verified === true ;
828 if (result.verified) {
829 e.lastGoodIndex = index;
830 e.lastGoodSourceId = String (sourceId);
831 } else e.suspectInterval = {
832 fromIndex: (e.lastGoodIndex || 0 ) + 1 ,
833 toIndex: index,
834 count: index - (e.lastGoodIndex || 0 ),
835 sourceIds: this .validated.inventory.records. slice (e.lastGoodIndex || 0 , index). filter ( r => ! r.held). map ( r => r.sourceId)
836 };
837 });
838 return result;
839 }
840 async reconcileInterval ( getRecord , targetIdFor ) {
841 const item = await this . item ();
842 // The interval names every non-held record between the last good sample and the failed one,
843 // landed or not. Only a record with a target can have been written while the sample was
844 // failing; it is read back and compared. A record with no target is listed, never read --
845 // unless this gate holds it as a possibly-sent write, which stops the reconcile instead.
846 const noTargetSourceIds = [];
847 for ( const id of item.suspectInterval?.sourceIds || []) {
848 const targetId = await targetIdFor (id);
849 if ( noTarget (targetId)) {
850 if (( await this . protected (id))?.uncertain) { await this . stop ( 'suspect-write-outcome-unknown' ); return false ; }
851 noTargetSourceIds. push ( String (id));
852 continue ;
853 }
854 const record = await getRecord (id);
855 try {
856 const result = await this . observe (record, id, targetId, await this . expectedFor (record, id));
857 if ( ! result.verified) return false ;
858 } catch (e) {
859 if (wixAuth. isRunInterruption (e)) throw e;
860 await this . stop (e.code || 'unknown-reconciliation' );
861 return false ;
862 }
863 }
864 await this . mutate (( g , e ) => {
865 e.reconciledInterval = { ... e.suspectInterval, noTargetSourceIds };
866 e.suspectInterval = null ;
867 e.status = 'passed' ;
868 e.stopReason = null ;
869 });
870 return true ;
871 }
872 }
873 const noTarget = id => id === undefined || id === null || id === '' || id === 'undefined' ;
874 function operatorActions ( gate ) {
875 const rows = [];
876 for ( const [ runId , run ] of Object. entries (gate.runs)) for ( const [ entity , item ] of Object. entries (run.entities || {})) {
877 if (item.supersededBy && gate.runs[item.supersededBy]?.entities?.[entity]) continue ;
878 if (item.gated === false && ! Object. hasOwn ( C . DOMAINS , entity)) { declaration (entity, item); continue ; }
879 // A run that wrote nothing of this entity defers to the runs that did. With no such
880 // run anywhere the entity was never gated, and it still blocks (not attempted is never success).
881 if (item.stopReason === 'not-run-no-writes' && Object. entries (gate.runs). some (([ id , other ]) => id !== runId && other.entities?.[entity] && other.entities[entity].stopReason !== 'not-run-no-writes' && ! (other.entities[entity].supersededBy && gate.runs[other.entities[entity].supersededBy]?.entities?.[entity]))) continue ;
882 const prompt = `Requires Import Operator: In the migration project, resume read-back verification for ${ entity } in run ${ runId }; inspect execution/read-back-gate.json. Resolve the named mapping/credential/coverage finding; for media settlement wait and resume reads of confirmed ids without rewriting. Regenerate the completion report.` ;
883 if (item.status !== 'passed' || ! item.endOfRunVerified) rows. push ({
884 code: 'read-back-gate-stopped' ,
885 severity: 'critical' ,
886 entity,
887 affectedCount: item.canaries?. length || 0 ,
888 action: prompt,
889 deadlineReason: item.status === 'passed' && ! item.canaries?. length && item.heldRecords?.count ? 'Before DNS change or launch: every record is held by the builder; nothing was imported' : 'Before DNS change or launch: entity import is unverified'
890 });
891 for ( const listed of item.listed || []) rows. push ({
892 code: 'read-back-listed-canary' ,
893 severity: 'high' ,
894 entity,
895 affectedCount: 1 ,
896 action: `${ prompt } Resolve listed canary ${ listed . sourceId } (${ listed . reason }).` ,
897 deadlineReason: 'Before DNS change or launch: this canary remains unresolved'
898 });
899 if (item.suspectInterval) rows. push ({
900 code: 'read-back-suspect-interval' ,
901 severity: 'critical' ,
902 entity,
903 affectedCount: item.suspectInterval.count,
904 action: `Requires Import Operator: Run read-only reconciliation for ${ entity } records ${ item . suspectInterval . fromIndex }–${ item . suspectInterval . toIndex }, then regenerate the completion report.` ,
905 deadlineReason: 'Before DNS change or launch: sampled records may lack intended data'
906 });
907 if (item.heldRecords?.count) rows. push ({
908 code: 'read-back-held-records' ,
909 severity: 'high' ,
910 entity,
911 affectedCount: item.heldRecords.count,
912 action: `Requires Import Operator: In the migration project, ${ item . heldRecords . count } ${ entity } records are held by the builder and were not attempted (${ Object . entries ( item . heldRecords . reasons ). slice ( 0 , 10 ). map (([ r , n ]) => `${ r }: ${ n }` ). join ( ', ' ) }${ Object . keys ( item . heldRecords . reasons ). length > 10 ? ', …' : ''}); execution/read-back-inventory/${ entity }.json lists each. Resolve the named decisions, re-run inventory and validate, then resume.` ,
913 deadlineReason: 'Before DNS change or launch: held records are not imported'
914 });
915 if (item.approvedOmissions?. some ( o => o.merchantAcknowledgement?.status !== 'acknowledged' )) rows. push ({
916 code: 'read-back-omission-ack' ,
917 severity: 'high' ,
918 entity,
919 affectedCount: item.approvedOmissions. every ( o => o.affectedScope?.sourceIdsComplete && Array. isArray (o.affectedScope.sourceIds))
920 ? new Set (item.approvedOmissions. flatMap ( o => o.affectedScope.sourceIds)).size
921 : item.approvedOmissions. some ( o => o.recordCount == null ) || item.approvedOmissions. length > 1 && item.approvedOmissions. some ( o => o.affectedScope)
922 ? null : Math. max ( ... item.approvedOmissions. map ( o => o.recordCount)),
923 action: `Tell the merchant these fields were not migrated: ${ item . approvedOmissions . slice ( 0 , 10 ). map ( o => o . path ). join ( ', ' ) }${ item . approvedOmissions . length > 10 ? `, and ${ item . approvedOmissions . length - 10 } more (${ item . approvedOmissions . length } locations; ${ ( item . pendingAcknowledgementRequest ?. artifactRefs || []). some ( r => r ?. path === approval . evidencePath ( entity , item . pendingAcknowledgementRequest . requestId )) ? approval . evidencePath ( entity , item . pendingAcknowledgementRequest . requestId ) : `execution/read-back-gate.json ${ entity } approvedOmissions`} lists each)` : ''}. Requires Import Operator: In the migration project, run replatform answer --project <project> ${ item . approvedOmissions [ 0 ]. merchantAcknowledgement ?. requestId || '<request-id>'} --acknowledge --note '<who acknowledged and when>'; regenerate completion.` ,
924 deadlineReason: 'Before DNS change or launch: merchant acknowledgement required'
925 });
926 }
927 return rows;
928 }
929 // The gate sequence for one gated entity, owned by the toolkit so generated code does
930 // not wire it by hand. Generated code supplies the reviewed builder, the gate's entity options
931 // (marker, variants, read, lookup, repair) and the writer:
932 // write(record, expected) -> one record's confirmed write ({ id|targetId, findings }) or
933 // exactly { deferred: true, reason } before anything was sent;
934 // writeBatch(records) -> optional; [{ sourceId, targetId, findings }] for the non-canary
935 // records (a bulk writer). Defaults to write() per record.
936 // Order, as rp-import-codegen describes it: load the inventory and prepare; reconcile a suspect
937 // interval carried from an earlier run; settle canaries an earlier run sent and the target refused
938 // with a readable 4xx (never created, so written again); canaries; then the remaining records in
939 // stable source order, sampled at every 200th and the last landed record; a failed sample stops
940 // new batches until reconcileInterval clears it. Gate rules are unchanged.
941 const NOT_CREATED = row => row?.status === 'failed_terminal' && ! row.targetId && Number. isInteger (row.failure?.httpStatus) && row.failure.httpStatus >= 400 && row.failure.httpStatus < 500 && ! [ 408 , 429 ]. includes (row.failure.httpStatus);
942 async function runGatedEntity ({
943 records ,
944 sourceIdOf = record => String (record.id),
945 sourceKeyOf ,
946 writeBatch ,
947 batchSize = 200 ,
948 targetIdFor ,
949 ... options
950 } = {}) {
951 if ( ! Array. isArray (records)) throw new Error ( 'runGatedEntity requires records' );
952 if ( ! Number. isInteger (batchSize) || batchSize < 1 || 200 % batchSize !== 0 ) throw new Error ( 'runGatedEntity batchSize must divide the 200-record sample interval' );
953 const gate = new ReadBackGate (options),
954 entity = options.entity;
955 const byId = new Map (records. map ( record => [ String ( sourceIdOf (record)), record]));
956 const getRecord = async id => {
957 const record = byId. get ( String (id));
958 if ( ! record) throw problem ( 'gated-record-not-in-input' );
959 return record;
960 };
961 const result = ( status , extra = {}) => ({ entity, status, ... extra });
962 return gate. withWriteTracking ( async () => {
963 const inv = await state. readJson (options.projectDir, `execution/read-back-inventory/${ entity }.json` , null );
964 if ( ! inv) {
965 await gate. stop ( 'read-back-inventory-missing' );
966 return result ( 'stopped' , { stopReason: 'read-back-inventory-missing' , written: [] });
967 }
968 try {
969 await gate. prepare (inv);
970 } catch (e) {
971 return result ( 'stopped' , { stopReason: e.code || 'gate-prepare-failed' , errors: e.errors, written: [] });
972 }
973 const landed = new Map ();
974 const targetOf = async id => landed. get ( String (id)) || (targetIdFor && await targetIdFor ( String (id))) || ( await gate. protected ( String (id)))?.targetId;
975 let item = await gate. item ();
976 if (item.suspectInterval && ! ( await gate. reconcileInterval (getRecord, targetOf))) return result ( 'stopped' , { stopReason: ( await gate. item ()).stopReason || 'suspect-reconciliation-required' , written: [] });
977 if (sourceKeyOf) {
978 const { rows } = await require ( './local-state' ). loadAttemptJournal (options.projectDir). catch ( e => { if (e.code === 'ENOENT' ) return { rows: [] }; throw e; });
979 const latest = new Map ();
980 for ( const row of rows) latest. set (row.sourceStableKey, row);
981 for ( const id of item.schedule.canarySourceIds || []) {
982 const prior = await gate. protected (id), row = latest. get ( sourceKeyOf ( String (id)));
983 if (prior && ! prior.targetId && prior.uncertain === true && prior.outcome !== 'not-sent' && NOT_CREATED (row)) await gate. protect (id, { uncertain: false , sentAt: null , outcome: 'not-sent' , notSentReason: `target-rejected-${ row . failure . httpStatus }:${ row . attemptId }` });
984 }
985 }
986 if ( ! ( await gate. executeCanaries (getRecord))) {
987 item = await gate. item ();
988 return result (item.status, { stopReason: item.stopReason, written: [] });
989 }
990 item = await gate. item ();
991 const canaries = new Set ((item.schedule.canarySourceIds || []). map (String));
992 for ( const c of item.canaries || []) if (c.targetId) landed. set ( String (c.sourceId), c.targetId);
993 // Stable source order is the inventory's; held records are never written.
994 const order = inv.records. filter ( r => ! r.held && byId. has ( String (r.sourceId))). map ( r => String (r.sourceId));
995 const position = new Map (order. map (( id , i ) => [id, i + 1 ]));
996 const rest = order. filter ( id => ! canaries. has (id));
997 const written = [];
998 const writeOne = async record => {
999 const expected = await gate. expectedFor (record, sourceIdOf (record));
1000 const out = await gate. timeWrite (() => options. write (record, expected));
1001 if (out?.deferred === true ) return { sourceId: String ( sourceIdOf (record)), deferred: true , reason: out.reason };
1002 return { sourceId: String ( sourceIdOf (record)), targetId: writtenTargetId (entity, out), findings: out?.findings };
1003 };
1004 // Batches never cross a 200-record sample boundary of the source order, so a failed sample
1005 // stops every record after it (canaries leave gaps, so slicing `rest` alone would not align).
1006 const batches = [];
1007 for ( const id of rest) {
1008 const window = Math. floor ((position. get (id) - 1 ) / 200 ), last = batches. at ( - 1 );
1009 if (last && last.window === window && last.ids. length < batchSize) last.ids. push (id);
1010 else batches. push ({ window, ids: [id] });
1011 }
1012 let lastLanded = null ;
1013 for ( const [ batchIndex , { ids }] of batches. entries ()) {
1014 const batch = await Promise . all (ids. map (getRecord));
1015 const outcomes = writeBatch ? await gate. timeWrite (() => writeBatch (batch)) : [];
1016 if ( ! writeBatch) for ( const record of batch) outcomes. push ( await writeOne (record));
1017 const isFinalBatch = batchIndex === batches. length - 1 ;
1018 const confirmed = (outcomes || []). filter ( o => o && o.targetId && byId. has ( String (o.sourceId)));
1019 for ( const o of outcomes || []) if (o) written. push (o);
1020 for ( const o of confirmed) landed. set ( String (o.sourceId), String (o.targetId));
1021 for ( const [ n , o ] of confirmed. entries ()) {
1022 const id = String (o.sourceId), isLast = isFinalBatch && n === confirmed. length - 1 ;
1023 const sample = await gate. verifySample ({ record: await getRecord (id), sourceId: id, targetId: String (o.targetId), index: position. get (id), isLast, findings: o.findings });
1024 if (isLast) lastLanded = id;
1025 if ( ! sample.skipped && ! sample.verified) {
1026 if ( ! ( await gate. reconcileInterval (getRecord, targetOf))) {
1027 item = await gate. item ();
1028 return result (item.status, { stopReason: item.stopReason || 'read-back-sample-failed' , written });
1029 }
1030 }
1031 }
1032 }
1033 // Every non-canary record landed earlier, or none remained: the end-of-run check reads the last
1034 // landed record in source order, so a resume still verifies the entity's tail.
1035 if ( ! lastLanded) {
1036 const tail = [ ... order]. reverse (). find ( id => landed. has (id));
1037 if (tail && ! canaries. has (tail)) await gate. verifySample ({ record: await getRecord (tail), sourceId: tail, targetId: landed. get (tail), index: position. get (tail), isLast: true });
1038 else if (tail) await gate. mutate (( g , e ) => { e.endOfRunVerified = e.canaries. find ( c => String (c.sourceId) === tail)?.outcome === 'verified' ; });
1039 // Nothing landed (every write deferred): no claim either way; prepare's own rule stands.
1040 }
1041 item = await gate. item ();
1042 return result (item.status, { stopReason: item.stopReason || null , endOfRunVerified: item.endOfRunVerified === true , written });
1043 });
1044 }
1045 module . exports = {
1046 ReadBackGate,
1047 declaration,
1048 operatorActions,
1049 runGatedEntity
1050 };