From 8f0ac3773adec1434093bc75f4c73077d4d9ee12 Mon Sep 17 00:00:00 2001 From: rdlabo Date: Thu, 13 Aug 2026 14:11:37 +0900 Subject: [PATCH] Simplify offline command and projection ownership --- projects/kit/README.md | 6 +- ...offline-aggregate-intent-projector.spec.ts | 2 - .../src/lib/offline-command-executor.ts | 17 ++-- .../lib/offline-coordinator.service.spec.ts | 4 +- .../src/lib/offline-natural-key.spec.ts | 10 -- .../kit/offline/src/lib/offline-provider.ts | 1 - .../lib/offline-replica-pull.service.spec.ts | 98 +++++++++++++------ .../src/lib/offline-replica-pull.service.ts | 19 ++-- .../offline/src/lib/offline-replica-puller.ts | 7 +- .../src/lib/offline-repository.spec.ts | 15 +-- .../kit/offline/src/lib/offline-repository.ts | 2 +- .../src/lib/offline-sync.service.spec.ts | 98 ++++++++++++++----- .../offline/src/lib/offline-sync.service.ts | 48 ++------- .../src/lib/sqlite-offline-repository.spec.ts | 13 +-- .../src/lib/sqlite-offline-repository.ts | 3 +- 15 files changed, 187 insertions(+), 156 deletions(-) diff --git a/projects/kit/README.md b/projects/kit/README.md index 9a4decd..7926115 100644 --- a/projects/kit/README.md +++ b/projects/kit/README.md @@ -578,8 +578,10 @@ baseline. The server remains authoritative; SQLite is the durable local working Synchronized mode requires `aggregateIntentProjector`. Kit calls it inside `OfflineReplicaMutationCoordinator` after enqueue, batch enqueue, replacement, discard, transport success, and pull acknowledgement. The adapter must derive only from authoritative confirmed values plus the complete remaining intent chain. Do not call it from product code. -Commands persist payload/metadata and declared `localOnlyFootprint` keys only; optimistic snapshots and companion -before/after images are not durable truth. +Commands persist an immutable payload plus Kit-owned metadata (`baseRevision`, state, footprint keys). Kit updates +only `baseRevision` when a newer server revision is known, and sets it to `null` after a generated remote identity is +released; the product executor always receives the original payload. Optimistic snapshots and companion before/after +images are not durable truth. When an API response is assembled from a base replica row plus product-owned local-only view rows, use `enqueuePrepared()`. Its callback runs inside the shared replica mutation lane, so every read used to derive the diff --git a/projects/kit/offline/src/lib/offline-aggregate-intent-projector.spec.ts b/projects/kit/offline/src/lib/offline-aggregate-intent-projector.spec.ts index 024c6b2..d4ce04f 100644 --- a/projects/kit/offline/src/lib/offline-aggregate-intent-projector.spec.ts +++ b/projects/kit/offline/src/lib/offline-aggregate-intent-projector.spec.ts @@ -147,7 +147,6 @@ describe('OfflineAggregateIntentProjector', () => { identity: { kind: 'generated' as const, localId: 'item-1' }, operation: 'items.absolute', payload: {}, - payloadHash: 'hash', baseRevision: 1, state: 'pending' as const, attempts: 0, @@ -268,7 +267,6 @@ describe('OfflineAggregateIntentProjector', () => { provide: OFFLINE_COMMAND_EXECUTOR, useValue: { execute: vi.fn(async () => ({ response: null })), - withServerRevision: (command: OfflineCommand) => command, }, }, { diff --git a/projects/kit/offline/src/lib/offline-command-executor.ts b/projects/kit/offline/src/lib/offline-command-executor.ts index 3c398d8..968c7b0 100644 --- a/projects/kit/offline/src/lib/offline-command-executor.ts +++ b/projects/kit/offline/src/lib/offline-command-executor.ts @@ -35,18 +35,23 @@ export type OfflineCommandTarget = export interface OfflineCommandExecutor { /** Sends the command using `command.commandId` as its durable server-side idempotency key. */ execute(command: OfflineCommand, target: OfflineCommandTarget): Promise; - withServerRevision(command: OfflineCommand, revision: string | number): OfflineCommand; /** * Whether this transport error authoritatively proves that this idempotency * key did not commit. Returning true may clear an ambiguity retained from an * earlier response-loss attempt and expose normal conflict resolution. */ provesCommandNotCommitted?(error: unknown, command: OfflineCommand): boolean; - /** - * Removes the deleted remote row's revision from a queued recreate. - * Required only when `clearRemoteId` completes while later commands remain. - */ - withoutServerRevision?(command: OfflineCommand): OfflineCommand; +} + +/** + * Returns a command whose only changed field is `baseRevision`. + * + * Payload and every other field are copied unchanged. Kit uses this when a + * newer server revision is known, and passes `null` after a generated remote + * identity is released. + */ +export function offlineCommandWithBaseRevision(command: OfflineCommand, baseRevision: string | number | null): OfflineCommand { + return { ...command, baseRevision }; } /** DI token for the product-specific command transport adapter. */ diff --git a/projects/kit/offline/src/lib/offline-coordinator.service.spec.ts b/projects/kit/offline/src/lib/offline-coordinator.service.spec.ts index 0e591aa..1a8a758 100644 --- a/projects/kit/offline/src/lib/offline-coordinator.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-coordinator.service.spec.ts @@ -4,7 +4,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import { OfflineCoordinatorService } from './offline-coordinator.service'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; import { OfflineNetworkService } from './offline-network.service'; -import { OFFLINE_REPOSITORY, type OfflineScope } from './offline-repository'; +import { OFFLINE_REPOSITORY, OFFLINE_SCHEMA_VERSION, type OfflineScope } from './offline-repository'; import { OfflineSessionService, type OfflineSessionManifest } from './offline-session.service'; import { OfflineStorageUnavailableError } from './offline-storage'; import { OfflineSyncService } from './offline-sync.service'; @@ -263,7 +263,7 @@ describe('OfflineCoordinatorService', () => { describe('storage initialization failure', () => { const storageError = new OfflineStorageUnavailableError( 'core_schema_incompatible', - 'Unsupported offline storage schema version 999; expected 2.', + `Unsupported offline storage schema version 999; expected ${OFFLINE_SCHEMA_VERSION}.`, ); it('default fails closed and does not start session or sync', async () => { diff --git a/projects/kit/offline/src/lib/offline-natural-key.spec.ts b/projects/kit/offline/src/lib/offline-natural-key.spec.ts index 0ea8634..d765f7b 100644 --- a/projects/kit/offline/src/lib/offline-natural-key.spec.ts +++ b/projects/kit/offline/src/lib/offline-natural-key.spec.ts @@ -1,7 +1,6 @@ import { TestBed } from '@angular/core/testing'; import { KitStorageService } from '@rdlabo/ionic-angular-kit'; import { describe, expect, it, vi } from 'vitest'; -import { OFFLINE_COMMAND_EXECUTOR } from './offline-command-executor'; import { OFFLINE_COMMAND_HOOKS } from './offline-command-hooks'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; import { OFFLINE_REPLICA_PULLER, type OfflineReplicaPullPage } from './offline-replica-puller'; @@ -287,13 +286,6 @@ describe('natural-key pull reconciliation', () => { { provide: OFFLINE_REPOSITORY, useExisting: IonicOfflineRepository }, { provide: OFFLINE_REPLICA_PULLER, useValue: { pull } }, { provide: OFFLINE_COMMAND_HOOKS, useValue: { entityType: (command: OfflineCommand) => command.aggregateType } }, - { - provide: OFFLINE_COMMAND_EXECUTOR, - useValue: { - execute: vi.fn(), - withServerRevision: (command: OfflineCommand, revision: string | number) => ({ ...command, baseRevision: revision }), - }, - }, { provide: OFFLINE_AGGREGATE_INTENT_PROJECTOR, useValue: { project: rematerializeTestAggregate } }, ], }); @@ -311,7 +303,6 @@ describe('natural-key pull reconciliation', () => { identity: naturalCommandIdentity(key42), operation: 'create', payload: { favTo: '42', label: 'optimistic' }, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 1, @@ -350,7 +341,6 @@ describe('natural-key pull reconciliation', () => { identity: naturalCommandIdentity(key42), operation: 'update', payload: { favTo: '42', label: 'pending edit' }, - payloadHash: 'hash-2', baseRevision: 2, state: 'pending', attempts: 0, diff --git a/projects/kit/offline/src/lib/offline-provider.ts b/projects/kit/offline/src/lib/offline-provider.ts index 281888a..4320333 100644 --- a/projects/kit/offline/src/lib/offline-provider.ts +++ b/projects/kit/offline/src/lib/offline-provider.ts @@ -80,7 +80,6 @@ const READ_CACHE_ONLY_COMMAND_EXECUTOR: OfflineCommandExecutor = { execute: async (): Promise => { throw new Error('This offline provider is configured as a read-only cache.'); }, - withServerRevision: (command) => command, }; const READ_CACHE_ONLY_REPLICA_PULLER: OfflineReplicaPuller = { 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 7d1c637..2b51b6d 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 @@ -165,10 +165,6 @@ describe('OfflineReplicaPullService', () => { provide: OFFLINE_COMMAND_EXECUTOR, useValue: { execute: vi.fn(), - withServerRevision: (command: OfflineCommand, revision: string | number) => ({ - ...command, - baseRevision: revision, - }), }, }, { @@ -327,7 +323,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: 'shared' }, operation: 'test_items.update', payload: { title: 'Optimistic' }, - payloadHash: 'hash', baseRevision: 1, state: 'pending', attempts: 0, @@ -469,7 +464,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: 'race-local' }, operation: 'test_items.update', payload: { title: 'Optimistic edit' }, - payloadHash: 'hash', baseRevision: 2, state: 'pending', attempts: 0, @@ -848,7 +842,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-pending' }, operation: 'test_items.update', payload: { title: 'Optimistic draft' }, - payloadHash: 'hash', baseRevision: 2, state: 'pending', attempts: 0, @@ -894,7 +887,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-conflict' }, operation: 'test_items.update', payload: { title: 'Local edit' }, - payloadHash: 'hash', baseRevision: 1, state: 'pending', attempts: 0, @@ -958,7 +950,6 @@ describe('OfflineReplicaPullService', () => { operation: 'test_items.update', payload: { title: 'Local delta' }, localOnlyFootprint: [view], - payloadHash: 'hash', baseRevision: 1, state: 'pending', attempts: 0, @@ -996,6 +987,16 @@ describe('OfflineReplicaPullService', () => { }); it('remote tombstone conflictはpending commandをremote_deleted conflictへ遷移する', async () => { + const derived: OfflineReplicaRow = { + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + values: { title: 'Pending delete' }, + confirmedValues: { title: 'Confirmed' }, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + }; await repository.transactReplica({ putRows: [ { @@ -1008,6 +1009,7 @@ describe('OfflineReplicaPullService', () => { fetchedAt: 1, syncState: 'pending', }, + derived, ], putCommands: [ { @@ -1018,7 +1020,7 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-tombstone' }, operation: 'test_items.delete', payload: { title: 'Pending delete' }, - payloadHash: 'hash', + localOnlyFootprint: [derived], baseRevision: 1, state: 'pending', attempts: 0, @@ -1028,6 +1030,7 @@ describe('OfflineReplicaPullService', () => { }, ], }); + projector.project.mockResolvedValueOnce({ removeRows: [derived] }); pull.mockResolvedValueOnce(page([itemChange(42, 'Confirmed', { deleted: true, serverRevision: 2 })], { nextCursor: 'cursor-v1' })); await service.pull(scope); @@ -1046,6 +1049,12 @@ describe('OfflineReplicaPullService', () => { }), ]); expect(await repository.getReplicaRowByRemoteId(scope, 'test_items', 42)).not.toBeNull(); + await expect(repository.getReplicaRow(scope, 'test_views', derived.identity)).resolves.toMatchObject({ + values: { title: 'Pending delete' }, + confirmedValues: null, + syncState: 'conflict', + }); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v1' }); }); describe('lost ACK correlation', () => { @@ -1072,7 +1081,6 @@ describe('OfflineReplicaPullService', () => { identity: generatedCommandIdentity(localId), operation: 'test_items.create', payload: { title: 'Draft create' }, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -1162,7 +1170,6 @@ describe('OfflineReplicaPullService', () => { operation: 'test_items.delete', payload: {}, replicaMutation: 'delete', - payloadHash: 'hash', baseRevision: 1, state: 'awaiting_pull', attempts: 1, @@ -1217,7 +1224,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-update' }, operation: 'test_items.update', payload: { title: 'First edit' }, - payloadHash: 'hash-1', baseRevision: 1, state: 'pending', attempts: 0, @@ -1233,7 +1239,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-update' }, operation: 'test_items.update', payload: { title: 'Follow-up edit' }, - payloadHash: 'hash-2', baseRevision: 1, state: 'pending', attempts: 0, @@ -1286,7 +1291,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-delete-ack' }, operation: 'test_items.delete', payload: { title: 'Pending delete' }, - payloadHash: 'hash', baseRevision: 1, state: 'pending', attempts: 0, @@ -1309,6 +1313,56 @@ describe('OfflineReplicaPullService', () => { expect(await repository.getCommands(scope)).toEqual([]); }); + it('delete ACKは同じaggregateのfollowing commandをremote_deleted conflictへ遷移する', async () => { + const base: OfflineReplicaRow = { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-delete-following', remoteId: 42 }, + values: { id: 42, title: 'Following upsert' }, + confirmedValues: { id: 42, title: 'Confirmed' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending', + }; + const command = (commandId: string, operation: string, createdAt: number): OfflineCommand => ({ + ...scope, + commandId, + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-delete-following' }, + operation, + payload: { title: operation.endsWith('delete') ? 'Pending delete' : 'Following upsert' }, + replicaMutation: operation.endsWith('delete') ? 'delete' : 'upsert', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt, + lastErrorCode: null, + }); + await repository.transactReplica({ + putRows: [base], + putCommands: [command('cmd-delete-ack', 'test_items.delete', 1), command('cmd-following-upsert', 'test_items.update', 2)], + }); + pull.mockResolvedValueOnce( + page([itemChange(42, 'Deleted', { deleted: true, serverRevision: 2, acknowledgedCommandIds: ['cmd-delete-ack'] })]), + ); + + await service.pull(scope); + + await expect(repository.getReplicaRow(scope, 'test_items', generatedCommandIdentity('019d-delete-following'))).resolves.toMatchObject( + { + values: { title: 'Following upsert' }, + confirmedValues: null, + serverRevision: 2, + syncState: 'conflict', + }, + ); + await expect(repository.getCommands(scope)).resolves.toEqual([ + expect.objectContaining({ commandId: 'cmd-following-upsert', state: 'conflict', lastErrorCode: 'remote_deleted', retryAt: null }), + ]); + }); + it('同一pageでdelete ACKの後にtombstoneが続く場合はfollowing commandをconflictにして旧baselineを残さない', async () => { await repository.transactReplica({ putRows: [ @@ -1333,7 +1387,6 @@ describe('OfflineReplicaPullService', () => { operation: 'test_items.delete', payload: { title: 'Pending delete' }, replicaMutation: 'delete', - payloadHash: 'hash-delete', baseRevision: 1, state: 'pending', attempts: 1, @@ -1350,7 +1403,6 @@ describe('OfflineReplicaPullService', () => { operation: 'test_items.update', payload: { title: 'Following upsert' }, replicaMutation: 'upsert', - payloadHash: 'hash-following-upsert', baseRevision: 1, state: 'pending', attempts: 0, @@ -1407,7 +1459,6 @@ describe('OfflineReplicaPullService', () => { operation: 'test_items.delete', payload: { title: 'Pending delete' }, replicaMutation: 'delete', - payloadHash: 'hash', baseRevision: 1, state: 'pending', attempts: 0, @@ -1475,7 +1526,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-lost-update' }, operation: 'test_items.update', payload: { title: 'First edit' }, - payloadHash: 'hash-1', baseRevision: 1, state: 'pending', attempts: 1, @@ -1491,7 +1541,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-lost-update' }, operation: 'test_items.update', payload: { title: 'Follow-up edit' }, - payloadHash: 'hash-2', baseRevision: 1, state: 'pending', attempts: 0, @@ -1558,7 +1607,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-other-acks' }, operation: 'test_items.update', payload: { title: 'Local edit' }, - payloadHash: 'hash-local', baseRevision: 3, state: 'pending', attempts: 0, @@ -1619,7 +1667,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-skip' }, operation: 'test_items.update', payload: { title: 'First edit' }, - payloadHash: 'hash-1', baseRevision: 1, state: 'pending', attempts: 0, @@ -1635,7 +1682,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-skip' }, operation: 'test_items.update', payload: { title: 'Second edit' }, - payloadHash: 'hash-2', baseRevision: 1, state: 'pending', attempts: 0, @@ -1688,7 +1734,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-local-a' }, operation: 'test_items.create', payload: { title: 'Pending create A' }, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -1781,7 +1826,6 @@ describe('OfflineReplicaPullService', () => { identity: { kind: 'generated', localId: '019d-sensitive' }, operation: 'test_items.absolute', payload: { title: 'Pending absolute intent' }, - payloadHash: 'hash', baseRevision: 1, state: 'pending', attempts: 0, @@ -1881,10 +1925,6 @@ describe('OfflineReplicaPullService', () => { provide: OFFLINE_COMMAND_EXECUTOR, useValue: { execute: vi.fn(), - withServerRevision: (command: OfflineCommand, revision: string | number) => ({ - ...command, - baseRevision: revision, - }), }, }, { 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 4f017b8..644ef7f 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -2,7 +2,7 @@ import { inject, Injectable } from '@angular/core'; import { isOfflineAggregateIntentConflict, offlineAggregateIntentMutations } from './offline-aggregate-intent-projector'; import { OFFLINE_COMMAND_HOOKS } from './offline-command-hooks'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; -import { OFFLINE_COMMAND_EXECUTOR } from './offline-command-executor'; +import { offlineCommandWithBaseRevision } from './offline-command-executor'; import { commandFootprintKeys, OfflineReplicaMutationCoordinator } from './offline-replica-mutation-coordinator'; import { OFFLINE_REPLICA_PROJECTOR, @@ -72,7 +72,6 @@ export class OfflineReplicaPullService { readonly #options = inject(OFFLINE_KIT_OPTIONS); readonly #puller = inject(OFFLINE_REPLICA_PULLER); readonly #projector = inject(OFFLINE_REPLICA_PROJECTOR, { optional: true }); - readonly #executor = inject(OFFLINE_COMMAND_EXECUTOR); readonly #hooks = inject(OFFLINE_COMMAND_HOOKS); readonly #replicaMutations = inject(OfflineReplicaMutationCoordinator); #schemaHash: Promise | null = null; @@ -124,8 +123,6 @@ export class OfflineReplicaPullService { const projection = await this.#projector?.project({ scope, changes, - commands: scopeCommands, - repository: this.#repository, }); this.#assertProjection(scope, projection); const putRows: OfflineReplicaRow[] = []; @@ -223,6 +220,7 @@ export class OfflineReplicaPullService { for (const command of related) { putCommands.set(command.commandId, { ...command, state: 'conflict', retryAt: null, lastErrorCode: 'remote_deleted' }); } + rematerializeAfter.push(related[0]!); continue; } @@ -256,7 +254,7 @@ export class OfflineReplicaPullService { putRows.push(confirmedRow); if (remaining.length > 0) rematerializeAfter.push(remaining[0]!); for (const command of remaining) { - putCommands.set(command.commandId, this.#executor.withServerRevision(command, change.serverRevision)); + putCommands.set(command.commandId, offlineCommandWithBaseRevision(command, change.serverRevision)); } } @@ -660,9 +658,9 @@ export class OfflineReplicaPullService { const following = related .slice(lastAcknowledgedIndex + 1) .map((command) => - acknowledgementSuperseded + acknowledgementSuperseded || change.deleted ? { ...command, state: 'conflict' as const, retryAt: null, lastErrorCode: change.deleted ? 'remote_deleted' : 'remote_revision' } - : this.#executor.withServerRevision(command, change.serverRevision), + : offlineCommandWithBaseRevision(command, change.serverRevision), ); for (const command of following) putCommands.set(command.commandId, command); for (const command of related.slice(0, lastAcknowledgedIndex + 1)) { @@ -676,15 +674,10 @@ export class OfflineReplicaPullService { ...row, confirmedValues: null, serverRevision: change.serverRevision, - syncState: acknowledgementSuperseded ? 'conflict' : 'pending', + syncState: 'conflict', visibility: following.at(-1)!.replicaMutation === 'delete' ? 'pending_delete' : 'present', fetchedAt: Date.now(), }); - if (acknowledgementSuperseded) { - for (const command of following) { - putCommands.set(command.commandId, { ...command, state: 'conflict', lastErrorCode: 'remote_deleted' }); - } - } } else { removeRows.push({ ...row, identity: row.identity }); } diff --git a/projects/kit/offline/src/lib/offline-replica-puller.ts b/projects/kit/offline/src/lib/offline-replica-puller.ts index c6d5519..3c2db52 100644 --- a/projects/kit/offline/src/lib/offline-replica-puller.ts +++ b/projects/kit/offline/src/lib/offline-replica-puller.ts @@ -75,12 +75,7 @@ export interface OfflineReplicaPullProjection { /** Pure product adapter for local-only projections derived from server replica changes. */ export interface OfflineReplicaProjector { - project(input: { - scope: OfflineScope; - changes: readonly OfflineReplicaChange[]; - commands: readonly import('./offline-repository').OfflineCommand[]; - repository: import('./offline-repository').OfflineRepository; - }): Promise; + project(input: { scope: OfflineScope; changes: readonly OfflineReplicaChange[] }): Promise; } /** Backend response accepted by the shared pull-page normalizer. */ diff --git a/projects/kit/offline/src/lib/offline-repository.spec.ts b/projects/kit/offline/src/lib/offline-repository.spec.ts index e951f27..62dca75 100644 --- a/projects/kit/offline/src/lib/offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/offline-repository.spec.ts @@ -313,7 +313,6 @@ describe('IonicOfflineRepository', () => { identity: { kind: 'generated', localId: '019d-aaaa' }, operation: 'test_items.update', payload: { title: 'Local item' }, - payloadHash: 'hash', baseRevision: 7, state: 'pending', attempts: 0, @@ -366,7 +365,6 @@ describe('IonicOfflineRepository', () => { identity: { kind: 'generated', localId: '019d-cccc' }, operation: 'test_items.delete', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -406,7 +404,6 @@ describe('IonicOfflineRepository', () => { identity: { kind: 'generated', localId: '019d-aaaa' }, operation: 'test_items.update', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -468,7 +465,6 @@ describe('IonicOfflineRepository', () => { identity: { kind: 'generated', localId: '019d-aaaa' }, operation: 'test_items.update', payload: { title: 'Local item' }, - payloadHash: 'hash', baseRevision: 7, state: 'pending', attempts: 0, @@ -823,7 +819,6 @@ describe('IonicOfflineRepository', () => { identity: { kind: 'generated', localId: '019d-aaaa' }, operation: 'test_items.update', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'pending' as const, attempts: 0, @@ -847,7 +842,6 @@ describe('IonicOfflineRepository', () => { identity: { kind: 'generated', localId: 'legacy' }, operation: 'test_items.update', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -886,7 +880,6 @@ describe('IonicOfflineRepository', () => { identity: { kind: 'generated', localId: '019d-aaaa' }, operation: 'documents.upsert', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'pending' as const, attempts: 0, @@ -941,7 +934,6 @@ describe('IonicOfflineRepository', () => { identity: { kind: 'generated', localId: '019d-aaaa' }, operation: 'test_items.create', payload: { title: 'local' }, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -1022,7 +1014,9 @@ describe('IonicOfflineRepository', () => { storage.values.set('offline:metadata', { schemaVersion: 999, lastUserId: 1 }); storage.values.set('offline:outbox:commands', { stale: {} }); storage.values.set('firebaseToken', { token: 'keep' }); - await expect(repository.initialize()).rejects.toThrow('Unsupported offline storage schema version 999; expected 2'); + await expect(repository.initialize()).rejects.toThrow( + `Unsupported offline storage schema version 999; expected ${OFFLINE_SCHEMA_VERSION}`, + ); expect(storage.values.get('offline:outbox:commands')).toEqual({ stale: {} }); expect(storage.values.get('offline:metadata')).toEqual({ schemaVersion: 999, lastUserId: 1 }); expect(storage.values.get('firebaseToken')).toEqual({ token: 'keep' }); @@ -1054,7 +1048,6 @@ describe('IonicOfflineRepository', () => { identity: { kind: 'generated', localId: '019d-bbbb' }, operation: 'test_items.create', payload: { title: 'Local item' }, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -1091,7 +1084,6 @@ describe('IonicOfflineRepository', () => { identity: generatedCommandIdentity('delete-uuid'), operation: 'test_items.delete', payload: { id: 42 }, - payloadHash: 'delete-hash', baseRevision: 7, replicaMutation: 'delete', state: 'pending', @@ -2056,7 +2048,6 @@ describe('IonicOfflineRepository', () => { identity: generatedCommandIdentity('019d-snap'), operation: 'test_items.update', payload: {}, - payloadHash: 'hash', baseRevision: 1, state: 'pending', attempts: 0, diff --git a/projects/kit/offline/src/lib/offline-repository.ts b/projects/kit/offline/src/lib/offline-repository.ts index 2002f3a..425a9dc 100644 --- a/projects/kit/offline/src/lib/offline-repository.ts +++ b/projects/kit/offline/src/lib/offline-repository.ts @@ -68,6 +68,7 @@ interface OfflineCommandBase extends OfflineScope { /** Stable row identity. The Outbox never persists a generated server id. */ identity: OfflineCommandIdentity; operation: string; + /** Opaque product payload. Kit never mutates this after enqueue. */ payload: T; /** * Declared localOnly projection rows this intent may create, update, or remove. @@ -76,7 +77,6 @@ interface OfflineCommandBase extends OfflineScope { localOnlyFootprint?: readonly OfflineReplicaRowKey[]; /** Durable intent used to preserve a hidden tombstone across restart and replay. */ replicaMutation?: OfflineReplicaMutation; - payloadHash: string; baseRevision: string | number | null; state: OfflineCommandState; attempts: number; diff --git a/projects/kit/offline/src/lib/offline-sync.service.spec.ts b/projects/kit/offline/src/lib/offline-sync.service.spec.ts index 8e9c793..ff759c5 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.spec.ts @@ -4,6 +4,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { OFFLINE_COMMAND_EXECUTOR, OFFLINE_SYNC_CONTEXT, + offlineCommandWithBaseRevision, type OfflineCommandResult, type OfflineCommandTarget, } from './offline-command-executor'; @@ -93,6 +94,36 @@ const textReplicaSchema = defineOfflineReplicaSchema({ migrations: [], }); +describe('offlineCommandWithBaseRevision', () => { + it('baseRevisionだけを差し替えpayloadは同一参照かつ同一JSONのまま残す', () => { + const payload = { title: 'unchanged', nested: { n: 1 } }; + const command: OfflineCommand = { + userId: 1, + scopeId: '10', + commandId: 'cmd-1', + aggregateType: 'documents', + sourceKey: 'documents', + identity: { kind: 'generated', localId: 'local-1' }, + operation: 'documents.update', + payload, + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }; + const updated = offlineCommandWithBaseRevision(command, 9); + expect(updated.baseRevision).toBe(9); + expect(updated.payload).toBe(payload); + expect(JSON.stringify(updated.payload)).toBe(JSON.stringify(payload)); + const cleared = offlineCommandWithBaseRevision(updated, null); + expect(cleared.baseRevision).toBeNull(); + expect(cleared.payload).toBe(payload); + expect(JSON.stringify(cleared.payload)).toBe(JSON.stringify(payload)); + }); +}); + describe('OfflineSyncService', () => { let service: OfflineSyncService; let commands: OfflineCommand[]; @@ -109,9 +140,9 @@ describe('OfflineSyncService', () => { let handleError: ReturnType void>>; let onCommandRemoved: ReturnType Promise>>; let options: OfflineKitOptions; - const execute = vi.fn( - async (_command: OfflineCommand, _target: OfflineCommandTarget): Promise => ({ response: null }), - ); + const execute = vi.fn(async (_command: OfflineCommand, _target: OfflineCommandTarget): Promise => ({ + response: null, + })); const provesCommandNotCommitted = vi.fn((_error: unknown, _command: OfflineCommand) => false); function expectAwaitingPull(count = commands.length): void { @@ -326,8 +357,6 @@ describe('OfflineSyncService', () => { useValue: { execute, provesCommandNotCommitted, - withServerRevision: (command: OfflineCommand) => command, - withoutServerRevision: (command: OfflineCommand) => ({ ...command, baseRevision: null }), }, }, { provide: OFFLINE_AGGREGATE_INTENT_PROJECTOR, useValue: { project: rematerializeTestAggregate } }, @@ -1452,8 +1481,6 @@ describe('OfflineSyncService', () => { useValue: { execute, provesCommandNotCommitted, - withServerRevision: (command: OfflineCommand) => command, - withoutServerRevision: (command: OfflineCommand) => ({ ...command, baseRevision: null }), }, }, { provide: OFFLINE_AGGREGATE_INTENT_PROJECTOR, useValue: { project: rematerializeTestAggregate } }, @@ -2144,7 +2171,6 @@ describe('OfflineSyncService', () => { identity: { kind: 'generated', localId: '019d-aaaa' }, operation: 'documents.create', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'sending', attempts: 1, @@ -2180,7 +2206,6 @@ describe('OfflineSyncService', () => { identity: { kind: 'generated', localId: '019d-restart-unknown' }, operation: 'documents.create', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'sending', attempts: 1, @@ -2715,14 +2740,44 @@ describe('OfflineSyncService', () => { nowSpy.mockRestore(); }); - it('locale非依存のkey順で同じJSON payloadを同一hashにする', async () => { + it('replicaのserverRevisionが進んでもenqueue payloadは構造的に不変でbaseRevisionだけ更新する', async () => { + rows.push({ + userId: 1, + scopeId: '10', + sourceKey: 'documents', + identity: { kind: 'generated', localId: '019d-rebase-payload', remoteId: 42 }, + values: { name: 'confirmed' }, + confirmedValues: { name: 'confirmed' }, + serverRevision: 4, + fetchedAt: 1, + syncState: 'confirmed', + }); + const payload = { name: 'draft', expectedRevision: 1 }; + await service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: '019d-rebase-payload' }, + operation: 'documents.update', + payload, + baseRevision: 1, + }, + { flush: false }, + ); + expect(commands[0]).toMatchObject({ baseRevision: 4 }); + expect(JSON.stringify(commands[0]?.payload)).toBe(JSON.stringify(payload)); + }); + + it('JSON payloadのkey順を変えずに保持する', async () => { + const first = { あ: 3, z: 1, ä: 2 }; + const second = { ä: 2, あ: 3, z: 1 }; await service.enqueue( { scopeId: '10', aggregateType: 'documents', identity: { kind: 'generated', localId: '1' }, operation: 'documents.upsert', - payload: { あ: 3, z: 1, ä: 2 }, + payload: first, }, { flush: false }, ); @@ -2732,14 +2787,15 @@ describe('OfflineSyncService', () => { aggregateType: 'documents', identity: { kind: 'generated', localId: '2' }, operation: 'documents.upsert', - payload: { ä: 2, あ: 3, z: 1 }, + payload: second, }, { flush: false }, ); - expect(service.pendingCommands()[0]?.payloadHash).toBe(service.pendingCommands()[1]?.payloadHash); + expect(JSON.stringify(service.pendingCommands()[0]?.payload)).toBe(JSON.stringify(first)); + expect(JSON.stringify(service.pendingCommands()[1]?.payload)).toBe(JSON.stringify(second)); }); - it('JSON外payloadを衝突するhashへ変換せずrejectする', async () => { + it('JSON外payloadをrejectする', async () => { await expect( service.enqueue( { @@ -2933,7 +2989,6 @@ describe('OfflineSyncService', () => { useValue: { execute, provesCommandNotCommitted: () => false, - withServerRevision: (command: OfflineCommand) => command, }, }, { provide: OFFLINE_AGGREGATE_INTENT_PROJECTOR, useValue: { project: rematerializeTestAggregate } }, @@ -3826,8 +3881,11 @@ describe('OfflineSyncService', () => { }); connected.set(true); + const recreatePayload = { name: 'recreated', presentation: null }; await service.flush(); + expect(JSON.stringify((execute.mock.calls[1]?.[0] as OfflineCommand).payload)).toBe(JSON.stringify(recreatePayload)); + expect((execute.mock.calls[1]?.[0] as OfflineCommand).baseRevision).toBeNull(); expect(execute.mock.calls.map((call) => call[1])).toEqual([ { kind: 'generated', localId: 'stable-local-id', remoteId: 42 }, { kind: 'generated', localId: 'stable-local-id', remoteId: null }, @@ -4162,8 +4220,7 @@ describe('OfflineSyncService', () => { }, { flush: false }, ); - const rebase = vi.spyOn(TestBed.inject(OFFLINE_COMMAND_EXECUTOR), 'withServerRevision'); - rebase.mockClear(); + const followingPayload = JSON.stringify(commands[1]?.payload); execute.mockResolvedValueOnce({ removeReplica: true, clearRemoteId: true, @@ -4173,8 +4230,8 @@ describe('OfflineSyncService', () => { connected.set(true); await expect(service.flush()).rejects.toThrow('Offline command cannot return serverRevision and clearRemoteId together.'); - expect(rebase).not.toHaveBeenCalled(); expect(commands[1]).toMatchObject({ baseRevision: 4 }); + expect(JSON.stringify(commands[1]?.payload)).toBe(followingPayload); }); it('pending deleteのdiscardはconfirmed baselineとpresent visibilityを復元する', async () => { @@ -4665,10 +4722,6 @@ describe('OfflineSyncService', () => { provide: OFFLINE_COMMAND_EXECUTOR, useValue: { execute, - withServerRevision: (command: OfflineCommand, revision: string | number) => ({ - ...command, - baseRevision: revision, - }), }, }, { provide: OFFLINE_AGGREGATE_INTENT_PROJECTOR, useValue: { project: rematerializeTestAggregate } }, @@ -4726,6 +4779,7 @@ describe('OfflineSyncService', () => { commandId: secondId, baseRevision: 2, }); + expect(JSON.stringify((execute.mock.calls[1]?.[0] as OfflineCommand).payload)).toBe(JSON.stringify({ title: 'G11 edit' })); expect( findReplicaRow({ userId: 1, scopeId: '10' }, 'test_items', { kind: 'generated', diff --git a/projects/kit/offline/src/lib/offline-sync.service.ts b/projects/kit/offline/src/lib/offline-sync.service.ts index 2be54ee..de583df 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -1,9 +1,10 @@ -import { computed, effect, ErrorHandler, inject, Injectable, InjectionToken, signal } from '@angular/core'; +import { computed, effect, ErrorHandler, inject, Injectable, InjectionToken, signal, untracked } from '@angular/core'; import { OFFLINE_COMMAND_EXECUTOR, OFFLINE_SYNC_CONTEXT, offlineCommandLookupIdentity, offlineCommandTargetFromReplicaRow, + offlineCommandWithBaseRevision, type OfflineCommandResult, type EnqueueOfflineCommandIdentity, type OfflineSyncSession, @@ -195,7 +196,11 @@ export class OfflineSyncService { constructor() { effect(() => { const connected = this.#network.connected(); - if (this.#initialized && connected) this.#flushInBackground(); + // The network transition may be observed after an explicit flush has already + // persisted a fatal pull attention. Do not let that stale transition restart + // the intentionally stopped loop; auth/session refresh remains able to retry. + const hasFatalPullAttention = untracked(() => this.#pullAttentions().length > 0); + if (this.#initialized && connected && !hasFatalPullAttention) this.#flushInBackground(); }); } @@ -517,7 +522,6 @@ export class OfflineSyncService { operation: request.operation, payload: normalized.payload, replicaMutation: request.replicaMutation ?? 'upsert', - payloadHash: await this.#payloadHash(normalized.payload), baseRevision: normalized.baseRevision, state: 'pending', attempts: 0, @@ -1177,34 +1181,12 @@ export class OfflineSyncService { commandIdentity: OfflineCommandIdentity, ): Promise<{ payload: T; baseRevision: string | number | null }> { let baseRevision = request.baseRevision ?? null; - let payload = request.payload; const sourceKey = this.#hooks.entityType(request); const row = await this.#getReplicaRowForSync(scope, sourceKey, commandIdentity); if (row?.serverRevision != null && row.serverRevision !== baseRevision) { - const rebased = this.#executor.withServerRevision( - { - ...scope, - commandId: '', - aggregateType: request.aggregateType, - sourceKey, - identity: commandIdentity, - operation: request.operation, - payload, - replicaMutation: request.replicaMutation ?? 'upsert', - payloadHash: '', - baseRevision, - state: 'pending', - attempts: 0, - retryAt: null, - createdAt: 0, - lastErrorCode: null, - }, - row.serverRevision, - ); baseRevision = row.serverRevision; - payload = rebased.payload as T; } - return { payload, baseRevision }; + return { payload: request.payload, baseRevision }; } async #completeCommand( @@ -1237,15 +1219,10 @@ export class OfflineSyncService { const following = latestCommands.slice(latestIndex + 1); const rebased = result.clearRemoteId === true - ? following.map((item) => { - if (!this.#executor.withoutServerRevision) { - throw new Error('Offline command executor must implement withoutServerRevision to recreate a deleted remoteId row.'); - } - return this.#executor.withoutServerRevision(item); - }) + ? following.map((item) => offlineCommandWithBaseRevision(item, null)) : revision === undefined ? following - : following.map((item) => this.#executor.withServerRevision(item, revision)); + : following.map((item) => offlineCommandWithBaseRevision(item, revision)); const current = await this.#rowForCommand(command); if (!this.#isCurrent(generation)) return; if (!current) { @@ -1768,11 +1745,6 @@ export class OfflineSyncService { await Promise.allSettled([...this.#sendingTransitions]); } - async #payloadHash(payload: unknown): Promise { - const digest = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(this.#canonicalJson(payload))); - return [...new Uint8Array(digest)].map((byte) => byte.toString(16).padStart(2, '0')).join(''); - } - #canonicalJson(value: unknown): string { if (Array.isArray(value)) return `[${value.map((item) => this.#canonicalJson(item)).join(',')}]`; if (value !== null && typeof value === 'object') { diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts index be928c1..c87fdbb 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts @@ -370,7 +370,6 @@ describe('SqliteOfflineRepository community sqlite driver', () => { identity: { kind: 'generated', localId: '019d-aaaa' }, operation: 'test_items.update', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -387,7 +386,6 @@ describe('SqliteOfflineRepository community sqlite driver', () => { identity: { kind: 'generated', localId: '019d-aaaa' }, operation: 'test_items.update', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -413,7 +411,7 @@ describe('SqliteOfflineRepository community sqlite driver', () => { expect(userQuery?.statement).toBe('SELECT * FROM offline_sync_commands WHERE user_id = ? ORDER BY created_at ASC, command_id ASC'); }); - it('delete command persists replica_mutation in the SQLite outbox row', async () => { + it('delete command persists replica_mutation and satisfies the released v2 payload_hash column', async () => { const repository = createRepository(); await repository.initialize(); await repository.putCommand({ @@ -425,7 +423,6 @@ describe('SqliteOfflineRepository community sqlite driver', () => { identity: { kind: 'generated', localId: 'delete-uuid' }, operation: 'test_items.delete', payload: { id: 42 }, - payloadHash: 'hash', baseRevision: 4, replicaMutation: 'delete', state: 'pending', @@ -440,6 +437,8 @@ describe('SqliteOfflineRepository community sqlite driver', () => { .find(({ statement }) => statement.startsWith('INSERT INTO offline_sync_commands')); expect(insert?.statement).toContain('replica_mutation'); expect(insert?.values).toContain('delete'); + expect(insert?.statement).toContain('payload_hash'); + expect(insert?.values?.[10]).toBe(''); }); it('persists and restores declared localOnly footprint keys', async () => { @@ -461,7 +460,6 @@ describe('SqliteOfflineRepository community sqlite driver', () => { operation: 'test_items.update', payload: { title: 'Optimistic' }, localOnlyFootprint: [companion], - payloadHash: 'hash', baseRevision: 1, state: 'pending', attempts: 0, @@ -490,7 +488,6 @@ describe('SqliteOfflineRepository community sqlite driver', () => { payload_json: JSON.stringify(command.payload), local_only_footprint_json: JSON.stringify([companion]), replica_mutation: 'upsert', - payload_hash: command.payloadHash, base_revision_json: JSON.stringify(command.baseRevision), state: command.state, attempts: command.attempts, @@ -1109,7 +1106,6 @@ describe('SqliteOfflineRepository replica rows', () => { identity: { kind: 'generated', localId: '019d-bbbb' }, operation: 'test_items.create', payload: { title: 'Local item' }, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -1532,7 +1528,6 @@ describe('SqliteOfflineRepository replica rows', () => { identity: { kind: 'generated', localId: '019d-lease-read' }, operation: 'test_items.update', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -1974,7 +1969,6 @@ describe('SqliteOfflineRepository replica rows', () => { identity: { kind: 'generated', localId: '019d-aaaa' }, operation: 'test_items.update', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, @@ -2092,7 +2086,6 @@ describe('SqliteOfflineRepository replica rows', () => { identity: { kind: 'generated', localId: '019d-aaaa' }, operation: 'test_items.update', payload: {}, - payloadHash: 'hash', baseRevision: null, state: 'pending', attempts: 0, diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.ts index 7b97555..5ebcbd7 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.ts @@ -819,7 +819,6 @@ export class SqliteOfflineRepository implements OfflineRepository { ? {} : { localOnlyFootprint: this.#parse(row['local_only_footprint_json']) }), replicaMutation: this.#string(row['replica_mutation']) as 'upsert' | 'delete', - payloadHash: this.#string(row['payload_hash']), baseRevision: this.#parseNullable(row['base_revision_json']), state: this.#string(row['state']) as OfflineCommand['state'], attempts: this.#number(row['attempts']), @@ -863,7 +862,7 @@ export class SqliteOfflineRepository implements OfflineRepository { JSON.stringify(command.payload), command.localOnlyFootprint === undefined ? null : JSON.stringify(command.localOnlyFootprint), command.replicaMutation ?? 'upsert', - command.payloadHash, + '', this.#stringifyNullable(command.baseRevision), command.state, command.attempts,