diff --git a/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts index ae99880..8e01f8b 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts @@ -997,7 +997,14 @@ describe('OfflineReplicaPullService', () => { }); projector.project.mockResolvedValue( projectionMutation === 'stale put' - ? { putRows: [oldView] } + ? { + putRows: [ + { + ...oldView, + confirmedValues: { title: 'Remote view' }, + }, + ], + } : { removeRows: [{ ...scope, sourceKey: 'test_views', identity: { kind: 'local', localId: 'view-42' } }] }, ); pull.mockResolvedValueOnce(page([itemChange(42, 'Remote truth', { serverRevision: 9 })], { nextCursor: 'cursor-v1' })); @@ -1008,7 +1015,14 @@ describe('OfflineReplicaPullService', () => { [expect.objectContaining({ commandId: 'cmd-rebase' })], expect.objectContaining({ title: 'Remote truth' }), 9, - [expect.objectContaining({ values: { title: 'Old optimistic view' } })], + projectionMutation === 'stale put' + ? [ + expect.objectContaining({ + values: { title: 'Old optimistic view' }, + confirmedValues: { title: 'Remote view' }, + }), + ] + : [], ); await expect(repository.getReplicaRow(scope, 'test_items', generatedCommandIdentity('019d-rebase'))).resolves.toMatchObject({ values: { title: 'Rebased delta' }, @@ -1034,6 +1048,91 @@ describe('OfflineReplicaPullService', () => { }, ); + it('passes only each aggregate canonical companion footprint to its rebase reducer', async () => { + const executor = TestBed.inject(OFFLINE_COMMAND_EXECUTOR); + const view = (remoteId: number, title: string): OfflineReplicaRow => ({ + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: `view-${remoteId}` }, + values: { title: `Optimistic ${title}` }, + confirmedValues: { title: `Old ${title}` }, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + }); + const views = [view(42, 'A'), view(43, 'B')]; + const command = (remoteId: number, companion: OfflineReplicaRow): OfflineCommand => ({ + ...scope, + commandId: `cmd-${remoteId}`, + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: `local-${remoteId}` }, + operation: 'test_items.delta', + payload: { delta: 1 }, + optimisticValue: { id: remoteId, title: `Local ${remoteId}` }, + optimisticCompanions: [{ key: companion, before: null, after: companion }], + payloadHash: `hash-${remoteId}`, + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: remoteId, + lastErrorCode: null, + }); + const commands = [command(42, views[0]!), command(43, views[1]!)]; + const rebase = vi.spyOn(executor, 'rebasePendingCommands').mockImplementation((related, confirmed, revision, companions) => ({ + steps: related.map((current) => ({ + optimisticValue: confirmed, + optimisticCompanions: [ + { + key: current.optimisticCompanions![0]!.key, + before: current.optimisticCompanions![0]!.before, + after: { ...companions[0]!, serverRevision: revision }, + }, + ], + })), + })); + await repository.transactReplica({ + putRows: [ + ...views, + ...[42, 43].map((remoteId) => ({ + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated' as const, localId: `local-${remoteId}`, remoteId }, + values: { id: remoteId, title: `Local ${remoteId}` }, + confirmedValues: { id: remoteId, title: `Old ${remoteId}` }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending' as const, + })), + ], + putCommands: commands, + }); + projector.project.mockResolvedValue({ + putRows: views.map((row, index) => ({ + ...row, + confirmedValues: { title: `Remote ${index === 0 ? 'A' : 'B'}` }, + })), + }); + pull.mockResolvedValueOnce( + page( + [ + itemChange(42, 'Remote 42', { serverRevision: 9 }), + itemChange(43, 'Remote 43', { serverRevision: 10 }), + ], + { nextCursor: 'cursor-v2' }, + ), + ); + + await service.pull(scope); + + expect(rebase).toHaveBeenCalledTimes(2); + expect(rebase.mock.calls.map((call) => call[3].map((row) => row.identity))).toEqual([ + [{ kind: 'local', localId: 'view-42' }], + [{ kind: 'local', localId: 'view-43' }], + ]); + }); + it.each(['conflict', 'rejected', 'blocked_auth'] as const)('%s commandは自動rebaseせずattention stateを維持する', async (state) => { const executor = TestBed.inject(OFFLINE_COMMAND_EXECUTOR); const rebase = vi.spyOn(executor, 'rebasePendingCommands'); diff --git a/projects/kit/offline/src/lib/offline-replica-pull.service.ts b/projects/kit/offline/src/lib/offline-replica-pull.service.ts index 7e23ed7..d0c26a7 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -88,6 +88,13 @@ export class OfflineReplicaPullService { const scopeCommands = await this.#repository.getCommands(scope); const userCommands = this.#repository.getCommandsForUser ? await this.#repository.getCommandsForUser(scope.userId) : scopeCommands; const changes = this.#collapseChanges(page.changes); + const projection = await this.#projector?.project({ + scope, + changes, + commands: scopeCommands, + repository: this.#repository, + }); + this.#assertProjection(scope, projection); const putRows: OfflineReplicaRow[] = []; const removeRows: OfflineReplicaRowKey[] = []; const rebasedPutRows: OfflineReplicaRow[] = []; @@ -202,7 +209,7 @@ export class OfflineReplicaPullService { related, confirmedValues, change.serverRevision, - await this.#currentCompanionRows(related), + await this.#projectedCompanionRows(related, projection), )) ?? null) : null; this.#assertRebasedSteps(related, rebase?.steps ?? null); @@ -239,13 +246,6 @@ export class OfflineReplicaPullService { } } - const projection = await this.#projector?.project({ - scope, - changes, - commands: scopeCommands, - repository: this.#repository, - }); - this.#assertProjection(scope, projection); const finalRows = this.#mergeRowMutations([ { putRows, removeRows }, { putRows: projection?.putRows ?? [], removeRows: projection?.removeRows ?? [] }, @@ -388,6 +388,25 @@ export class OfflineReplicaPullService { return rows.filter((row): row is OfflineReplicaRow => row !== null); } + async #projectedCompanionRows( + commands: readonly OfflineCommand[], + projection: { putRows?: readonly OfflineReplicaRow[]; removeRows?: readonly OfflineReplicaRowKey[] } | undefined, + ): Promise { + const footprint = new Set( + commands.flatMap((command) => (command.optimisticCompanions ?? []).map((companion) => this.#rowKey(companion.key))), + ); + const rows = new Map((await this.#currentCompanionRows(commands)).map((row) => [this.#rowKey(row), row])); + for (const row of projection?.removeRows ?? []) { + const key = this.#rowKey(row); + if (footprint.has(key)) rows.delete(key); + } + for (const row of projection?.putRows ?? []) { + const key = this.#rowKey(row); + if (footprint.has(key)) rows.set(key, row); + } + return [...rows.values()]; + } + #assertPullChange(change: unknown, index: number): void { const label = `Offline replica pull page changes[${index}]`; if (!isPlainObject(change)) {