@@ -308,6 +308,67 @@ export class OfflineSyncService {
308308 } ) ;
309309 }
310310
311+ /**
312+ * Atomically rematerializes every unresolved intent for one aggregate.
313+ *
314+ * This is the conflict-recovery boundary for ordered intent chains: the old
315+ * chain remains durable until every replacement has been prepared and the
316+ * complete chain can be committed in one replica transaction.
317+ */
318+ replacePreparedAggregate < T > (
319+ commandId : string ,
320+ prepare : (
321+ repository : OfflineRepository ,
322+ commands : readonly OfflineCommand [ ] ,
323+ ) => Promise < readonly PreparedOfflineCommand < T > [ ] > ,
324+ options : PreparedOfflineBatchOptions = { } ,
325+ ) : Promise < readonly string [ ] > {
326+ const generation = this . #generation;
327+ return this . #serializeReplicaMutation( async ( ) => {
328+ await this . initialize ( ) ;
329+ if ( ! this . #isCurrent( generation ) ) throw new Error ( 'Offline session changed before prepared aggregate replacement.' ) ;
330+ const knownCommands = await this . #readKnownCommands( ) ;
331+ const selected = knownCommands . find ( ( command ) => command . commandId === commandId ) ;
332+ if ( ! selected ) throw new Error ( `Offline command ${ commandId } no longer exists.` ) ;
333+ const aggregateKey = this . #aggregateKey( selected ) ;
334+ const replaced = knownCommands . filter ( ( command ) => this . #aggregateKey( command ) === aggregateKey ) ;
335+ this . #assertDiscardable( replaced ) ;
336+ const prepared = await prepare ( this . #repository, replaced ) ;
337+ if ( prepared . length !== replaced . length ) {
338+ throw new Error ( 'Offline aggregate replacement must preserve the ordered intent count.' ) ;
339+ }
340+ const session = await this . #beginEnqueueSession( generation ) ;
341+ const materializations : MaterializedOfflineEnqueue [ ] = [ ] ;
342+ for ( const [ index , entry ] of prepared . entries ( ) ) {
343+ this . #assertEnqueueScope( session , entry . request . scopeId ) ;
344+ materializations . push (
345+ await this . #materializeEnqueue(
346+ session . userId ,
347+ entry . request ,
348+ entry . replicaTransaction ,
349+ replaced [ index ] ,
350+ ) ,
351+ ) ;
352+ }
353+ const retained = knownCommands . filter ( ( command ) => ! replaced . some ( ( item ) => item . commandId === command . commandId ) ) ;
354+ this . #assertDistinctBatchFootprints( materializations , retained , true ) ;
355+ await this . #assertOutboxCapacity(
356+ session . userId ,
357+ materializations . map ( ( item ) => item . command ) ,
358+ replaced . map ( ( command ) => command . commandId ) ,
359+ knownCommands ,
360+ ) ;
361+ options . assertCurrent ?.( ) ;
362+ await this . #commitMaterializedEnqueues(
363+ materializations ,
364+ generation ,
365+ options ,
366+ replaced . map ( ( command ) => command . commandId ) ,
367+ ) ;
368+ return materializations . map ( ( item ) => item . command . commandId ) ;
369+ } ) ;
370+ }
371+
311372 /**
312373 * Serializes a product-owned replica projection with enqueue and command ACK
313374 * reconciliation. For read/derive/write cache updates, prefer
@@ -352,7 +413,7 @@ export class OfflineSyncService {
352413 const currentCommands = await this . #commandsForUser( session . userId ) ;
353414 const retainedCommands = replaced ? currentCommands . filter ( ( command ) => command . commandId !== replaced . commandId ) : currentCommands ;
354415 this . #assertDistinctBatchFootprints( [ materialization ] , retainedCommands ) ;
355- await this . #assertOutboxCapacity( session . userId , [ materialization . command ] , replaced ? .commandId , currentCommands ) ;
416+ await this . #assertOutboxCapacity( session . userId , [ materialization . command ] , replaced ? [ replaced . commandId ] : undefined , currentCommands ) ;
356417 await this . #commitMaterializedEnqueues( [ materialization ] , generation , options , replaced ? [ replaced . commandId ] : undefined ) ;
357418 return materialization . command . commandId ;
358419 }
@@ -538,7 +599,11 @@ export class OfflineSyncService {
538599 return { command, optimisticRow, optimisticCompanions } ;
539600 }
540601
541- #assertDistinctBatchFootprints( entries : readonly MaterializedOfflineEnqueue [ ] , existingCommands : readonly OfflineCommand [ ] ) : void {
602+ #assertDistinctBatchFootprints(
603+ entries : readonly MaterializedOfflineEnqueue [ ] ,
604+ existingCommands : readonly OfflineCommand [ ] ,
605+ allowOneAggregate = false ,
606+ ) : void {
542607 const aggregates = new Set < string > ( ) ;
543608 const replicaKeys = new Set < string > ( ) ;
544609 const existingFootprints = new Map < string , string > ( ) ;
@@ -548,15 +613,15 @@ export class OfflineSyncService {
548613 }
549614 for ( const entry of entries ) {
550615 const aggregate = this . #aggregateKey( entry . command ) ;
551- if ( aggregates . has ( aggregate ) ) {
616+ if ( aggregates . has ( aggregate ) && ! allowOneAggregate ) {
552617 throw new Error ( 'Prepared offline batch contains overlapping aggregate intents.' ) ;
553618 }
554619 aggregates . add ( aggregate ) ;
555620 for ( const key of [
556621 this . #replicaRowKey( entry . optimisticRow ) ,
557622 ...entry . optimisticCompanions . map ( ( companion ) => this . #replicaRowKey( companion . key ) ) ,
558623 ] ) {
559- if ( replicaKeys . has ( key ) ) {
624+ if ( replicaKeys . has ( key ) && ! allowOneAggregate ) {
560625 throw new Error ( 'Prepared offline batch contains overlapping replica footprints.' ) ;
561626 }
562627 const existingAggregate = existingFootprints . get ( key ) ;
@@ -700,12 +765,12 @@ export class OfflineSyncService {
700765 async #assertOutboxCapacity(
701766 userId : OfflinePrincipalId ,
702767 newCommands : readonly OfflineCommand [ ] ,
703- excludingCommandId ?: string ,
768+ excludingCommandIds ?: readonly string [ ] ,
704769 knownCommands ?: readonly OfflineCommand [ ] ,
705770 ) : Promise < void > {
706771 const currentCommands = knownCommands ?? ( await this . #commandsForUser( userId ) ) ;
707- const commands = excludingCommandId
708- ? currentCommands . filter ( ( candidate ) => candidate . commandId !== excludingCommandId )
772+ const commands = excludingCommandIds
773+ ? currentCommands . filter ( ( candidate ) => ! excludingCommandIds . includes ( candidate . commandId ) )
709774 : currentCommands ;
710775 const maxCommands = this . #options. outboxLimits ?. maxCommandsPerUser ?? DEFAULT_MAX_OUTBOX_COMMANDS_PER_USER ;
711776 const maxBytes = this . #options. outboxLimits ?. maxBytesPerUser ?? DEFAULT_MAX_OUTBOX_BYTES_PER_USER ;
0 commit comments