diff --git a/projects/kit/offline/src/lib/offline-command-executor.ts b/projects/kit/offline/src/lib/offline-command-executor.ts index fcdd1b2..bcd86bd 100644 --- a/projects/kit/offline/src/lib/offline-command-executor.ts +++ b/projects/kit/offline/src/lib/offline-command-executor.ts @@ -1,5 +1,5 @@ import { InjectionToken } from '@angular/core'; -import type { OfflineCommand, OfflineScope } from './offline-repository'; +import type { OfflineCommand, OfflineOptimisticReplicaCompanion, OfflineReplicaRow, OfflineScope } from './offline-repository'; import type { OfflineCommandIdentity, OfflinePrincipalId, OfflineReplicaIdentity } from './offline-identity'; import type { OfflineGeneratedRemoteId, OfflineNaturalKey } from './offline-replica-schema'; @@ -31,6 +31,19 @@ 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; + /** + * Reapplies a complete aggregate's pending intents to a newer confirmed + * value. Return null when any intent is revision-sensitive. The returned + * values correspond to the original FIFO command order. Kit alone updates + * command metadata; payload and idempotency identity remain immutable. + * Without this hook, revision changes conflict by default. + */ + rebasePendingCommands?( + commands: readonly OfflineCommand[], + confirmedValues: unknown, + revision: string | number, + companionRows: readonly OfflineReplicaRow[], + ): OfflinePendingRebase | null | Promise; /** * Whether this transport error authoritatively proves that this idempotency * key did not commit. Returning true may clear an ambiguity retained from an @@ -44,6 +57,20 @@ export interface OfflineCommandExecutor { withoutServerRevision?(command: OfflineCommand): OfflineCommand; } +/** Product projection result after safely replaying pending intents onto a newer confirmed revision. */ +export interface OfflinePendingRebase { + /** Recomputed projections in the original durable FIFO order. */ + steps: readonly OfflinePendingRebaseStep[]; +} + +/** Projection state produced for one immutable durable command in FIFO order. */ +export interface OfflinePendingRebaseStep { + /** Recomputed full aggregate value after this command's intent is applied. */ + optimisticValue: unknown; + /** Same footprint as the original command, rematerialized from the new confirmed value. */ + optimisticCompanions?: readonly OfflineOptimisticReplicaCompanion[]; +} + /** DI token for the product-specific command transport adapter. */ export const OFFLINE_COMMAND_EXECUTOR = new InjectionToken('OFFLINE_COMMAND_EXECUTOR'); diff --git a/projects/kit/offline/src/lib/offline-contract.spec.ts b/projects/kit/offline/src/lib/offline-contract.spec.ts index 1493022..2d71920 100644 --- a/projects/kit/offline/src/lib/offline-contract.spec.ts +++ b/projects/kit/offline/src/lib/offline-contract.spec.ts @@ -7,8 +7,7 @@ import { offlineSessionManifestAllows } from './offline-session.service'; describe('shared offline boundary contracts', () => { it('normalizes database serverId exactly once at the pull transport boundary', () => { - expect( - normalizeOfflineReplicaPullPage({ + const normalized = normalizeOfflineReplicaPullPage({ schemaVersion: 1, schemaHash: 'hash', changes: [ @@ -29,8 +28,9 @@ describe('shared offline boundary contracts', () => { ], nextCursor: '2', hasMore: false, - }).changes, - ).toEqual([ + rebaselineRequired: true, + }); + expect(normalized.changes).toEqual([ { sourceKey: 'items', remoteId: 42, @@ -46,6 +46,7 @@ describe('shared offline boundary contracts', () => { deleted: true, }, ]); + expect(normalized.rebaselineRequired).toBe(true); }); it('narrows positive database ids without accepting coercion or zero', () => { diff --git a/projects/kit/offline/src/lib/offline-provider.ts b/projects/kit/offline/src/lib/offline-provider.ts index 206e52f..d991539 100644 --- a/projects/kit/offline/src/lib/offline-provider.ts +++ b/projects/kit/offline/src/lib/offline-provider.ts @@ -11,8 +11,8 @@ import { OfflineCoordinatorService } from './offline-coordinator.service'; import { IonicOfflineRepository, OFFLINE_REPOSITORY, selectOfflineRepository } from './offline-repository'; import type { OfflineMutationRequestPolicy, OfflineRequestPolicy } from './offline-request-policy'; import { provideOfflineMutationRequestPolicy, provideOfflineRequestPolicy } from './offline-request-policy'; -import type { OfflineReplicaPuller } from './offline-replica-puller'; -import { OFFLINE_REPLICA_PULLER } from './offline-replica-puller'; +import type { OfflineReplicaProjector, OfflineReplicaPuller } from './offline-replica-puller'; +import { OFFLINE_REPLICA_PROJECTOR, OFFLINE_REPLICA_PULLER } from './offline-replica-puller'; import { OfflineSessionService } from './offline-session.service'; import { COMMUNITY_SQLITE, @@ -29,6 +29,8 @@ interface ProvideOfflineOptionsBase extends OfflineKitOptions { commandHooks?: Type; /** Optional additional providers required by product adapters. */ providers?: readonly Provider[]; + /** Optional pure adapter for product-owned local-only projections applied inside the Kit pull transaction. */ + replicaProjector?: Type; /** Application-installed `@capacitor-community/sqlite` connection. Required only on iOS and Android. */ sqliteConnection?: CommunitySqliteConnection; } @@ -109,6 +111,9 @@ export function provideOffline(options: ProvideOfflineOptions): EnvironmentProvi ? { provide: OFFLINE_REPLICA_PULLER, useExisting: options.replicaPuller } : { provide: OFFLINE_REPLICA_PULLER, useValue: READ_CACHE_ONLY_REPLICA_PULLER }, ...(options.commandHooks ? [options.commandHooks, { provide: OFFLINE_COMMAND_HOOKS, useExisting: options.commandHooks }] : []), + ...(options.replicaProjector + ? [options.replicaProjector, { provide: OFFLINE_REPLICA_PROJECTOR, useExisting: options.replicaProjector }] + : []), ...options.requestPolicies.flatMap((policy) => provideOfflineRequestPolicy(policy)), ...(options.mutationPolicies ?? []).flatMap((policy) => provideOfflineMutationRequestPolicy(policy)), ...(options.providers ?? []), 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 90ecf02..ae99880 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 @@ -6,6 +6,7 @@ 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_PROJECTOR, OFFLINE_REPLICA_PULLER, type OfflineReplicaChange, type OfflineReplicaPullPage, @@ -19,6 +20,7 @@ import { defineReplicaEntity, integer, generatedId, + localOnly, sha256OfflineReplicaSchema, text, } from './offline-replica-schema'; @@ -44,9 +46,17 @@ const testItemEntity = defineReplicaEntity()({ }, }); +const testViewEntity = defineReplicaEntity<{ title: string }>()({ + table: 'test_views', + sourceKey: 'test_views', + scope: 'user', + identity: localOnly(), + fields: { title: text() }, +}); + const replicaSchema = defineOfflineReplicaSchema({ version: 1, - entities: [testItemEntity], + entities: [testItemEntity, testViewEntity], migrations: [], }); @@ -91,6 +101,7 @@ describe('OfflineReplicaPullService', () => { let storage: MemoryStorage; let schemaHash: string; let pull: ReturnType Promise>>; + let projector: { project: ReturnType }; function page( changes: readonly OfflineReplicaChange[], @@ -99,6 +110,7 @@ describe('OfflineReplicaPullService', () => { hasMore?: boolean; schemaVersion?: number; schemaHash?: string; + rebaselineRequired?: boolean; } = {}, ): OfflineReplicaPullPage { return { @@ -107,6 +119,7 @@ describe('OfflineReplicaPullService', () => { changes, nextCursor: options.nextCursor ?? 'cursor-v1', hasMore: options.hasMore ?? false, + ...(options.rebaselineRequired ? { rebaselineRequired: true } : {}), }; } @@ -141,6 +154,7 @@ describe('OfflineReplicaPullService', () => { { provide: OFFLINE_KIT_OPTIONS, useValue: { databaseName: 'test-offline', replicaSchema } }, { provide: OFFLINE_REPOSITORY, useExisting: IonicOfflineRepository }, { provide: OFFLINE_REPLICA_PULLER, useValue: { pull } }, + { provide: OFFLINE_REPLICA_PROJECTOR, useValue: projector }, { provide: OFFLINE_COMMAND_HOOKS, useValue: { entityType: (command: Pick) => command.aggregateType }, @@ -149,6 +163,7 @@ describe('OfflineReplicaPullService', () => { provide: OFFLINE_COMMAND_EXECUTOR, useValue: { execute: vi.fn(), + rebasePendingCommands: vi.fn(() => null), withServerRevision: (command: OfflineCommand, revision: string | number) => ({ ...command, baseRevision: revision, @@ -164,6 +179,7 @@ describe('OfflineReplicaPullService', () => { beforeEach(async () => { storage = new MemoryStorage(); pull = vi.fn(async () => page([])); + projector = { project: vi.fn(async () => ({})) }; await seedReplicaMetadata(); configureTestBed(); await repository.initialize(); @@ -183,6 +199,253 @@ describe('OfflineReplicaPullService', () => { }); }); + it('applies rebaseline reset, derived projection, base rows, and cursor in one transaction', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'stale', remoteId: 10 }, + values: { id: 10, title: 'Stale' }, + confirmedValues: { id: 10, title: 'Stale' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + putCursors: [{ ...scope, cursor: 'expired' }], + }); + projector.project.mockResolvedValue({ + putRows: [ + { + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + values: { title: 'Derived' }, + confirmedValues: { title: 'Derived' }, + serverRevision: null, + fetchedAt: 2, + syncState: 'confirmed', + }, + ], + }); + const transact = vi.spyOn(repository, 'transactReplica'); + pull.mockResolvedValueOnce(page([itemChange(42, 'Fresh')], { nextCursor: 'snapshot-1', rebaselineRequired: true })); + + await service.pull(scope); + + expect(transact).toHaveBeenCalledTimes(1); + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 10)).resolves.toBeNull(); + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 42)).resolves.toMatchObject({ + values: { title: 'Fresh' }, + }); + await expect(repository.getReplicaRow(scope, 'test_views', { kind: 'local', localId: 'view-42' })).resolves.toMatchObject({ + values: { title: 'Derived' }, + }); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'snapshot-1' }); + }); + + it('keeps snapshot and derived rows whose keys overlap the rebaseline removal set', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'existing', remoteId: 42 }, + values: { id: 42, title: 'Stale' }, + confirmedValues: { id: 42, title: 'Stale' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + { + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + values: { title: 'Stale derived' }, + confirmedValues: { title: 'Stale derived' }, + serverRevision: null, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + putCursors: [{ ...scope, cursor: 'expired' }], + }); + projector.project.mockResolvedValue({ + putRows: [ + { + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + values: { title: 'Fresh derived' }, + confirmedValues: { title: 'Fresh derived' }, + serverRevision: null, + fetchedAt: 2, + syncState: 'confirmed', + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Fresh')], { nextCursor: 'snapshot-1', rebaselineRequired: true })); + + await service.pull(scope); + + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 42)).resolves.toMatchObject({ + values: { title: 'Fresh' }, + }); + await expect(repository.getReplicaRow(scope, 'test_views', { kind: 'local', localId: 'view-42' })).resolves.toMatchObject({ + values: { title: 'Fresh derived' }, + }); + }); + + it('preserves pending companions by canonical identity across user-scoped partitions', async () => { + const otherScope = { userId: scope.userId, scopeId: '20' }; + const companion: OfflineReplicaRow = { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'shared', remoteId: 42 }, + values: { id: 42, title: 'Optimistic' }, + confirmedValues: { id: 42, title: 'Confirmed' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }; + await repository.transactReplica({ + putRows: [companion], + putCommands: [ + { + ...otherScope, + commandId: 'other-scope-command', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'other' }, + operation: 'test_items.update', + payload: {}, + optimisticValue: {}, + optimisticCompanions: [ + { + key: { + ...otherScope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'shared', remoteId: null }, + }, + before: companion, + after: companion, + }, + ], + payloadHash: 'hash', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + putCursors: [{ ...scope, cursor: 'expired' }], + }); + pull + .mockResolvedValueOnce(page([], { nextCursor: 'snapshot-1', rebaselineRequired: true, hasMore: true })) + .mockResolvedValueOnce(page([], { nextCursor: 'snapshot-1' })); + + await service.pull(scope); + + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 42)).resolves.toMatchObject({ + values: { title: 'Optimistic' }, + }); + }); + + it('keeps the previous replica durable until the first rebaseline snapshot can commit atomically', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'stale', remoteId: 10 }, + values: { id: 10, title: 'Stale' }, + confirmedValues: { id: 10, title: 'Stale' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + putCursors: [{ ...scope, cursor: 'expired' }], + }); + const transact = vi.spyOn(repository, 'transactReplica'); + transact.mockClear(); + pull + .mockResolvedValueOnce(page([], { nextCursor: '', hasMore: true, rebaselineRequired: true })) + .mockResolvedValueOnce(page([itemChange(42, 'Fresh')], { nextCursor: 'snapshot-1' })); + + await service.pull(scope); + + expect(pull.mock.calls.map(([request]) => request.cursor)).toEqual(['expired', '']); + expect(transact).toHaveBeenCalledTimes(1); + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 10)).resolves.toBeNull(); + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 42)).resolves.toMatchObject({ + values: { title: 'Fresh' }, + }); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'snapshot-1' }); + }); + + it('preserves the previous replica and cursor when the first rebaseline snapshot fails', async () => { + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'stale', remoteId: 10 }, + values: { id: 10, title: 'Stale' }, + confirmedValues: { id: 10, title: 'Stale' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'confirmed', + }, + ], + putCursors: [{ ...scope, cursor: 'expired' }], + }); + pull + .mockResolvedValueOnce(page([], { nextCursor: '', hasMore: true, rebaselineRequired: true })) + .mockRejectedValueOnce(new Error('snapshot unavailable')); + + await expect(service.pull(scope)).rejects.toThrow('snapshot unavailable'); + + await expect(repository.getReplicaRowByRemoteId(scope, 'test_items', 10)).resolves.toMatchObject({ + values: { title: 'Stale' }, + }); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'expired' }); + }); + + it('rejects a terminal rebaseline marker without clearing the previous replica', async () => { + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'expired' }] }); + pull.mockResolvedValueOnce(page([], { nextCursor: '', rebaselineRequired: true })); + + await expect(service.pull(scope)).rejects.toThrow('Offline replica rebaseline marker must lead to a snapshot page.'); + + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'expired' }); + }); + + it('rejects projector attempts to mutate synchronized base rows without advancing the cursor', async () => { + await repository.transactReplica({ putCursors: [{ ...scope, cursor: 'cursor-v0' }] }); + projector.project.mockResolvedValue({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: 'illegal', remoteId: 42 }, + values: { id: 42, title: 'Illegal' }, + confirmedValues: { id: 42, title: 'Illegal' }, + serverRevision: 2, + fetchedAt: 2, + syncState: 'confirmed', + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Fresh')], { nextCursor: 'cursor-v1' })); + + await expect(service.pull(scope)).rejects.toThrow('Offline replica projector may only mutate localOnly source "test_items".'); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v0' }); + }); + it('does not hold the mutation lane during transport and preserves an enqueue completed before stale page apply', async () => { let releasePull!: (value: OfflineReplicaPullPage) => void; pull.mockImplementationOnce(() => new Promise((resolve) => (releasePull = resolve))); @@ -387,6 +650,17 @@ describe('OfflineReplicaPullService', () => { ); }); + it('rejects a malformed rebaseline marker without advancing the cursor', async () => { + await expectPullRejectsPreservingCursor( + () => + pull.mockResolvedValueOnce({ + ...page([], { nextCursor: 'snapshot-1' }), + rebaselineRequired: 'yes', + } as unknown as OfflineReplicaPullPage), + 'Offline replica pull page rebaselineRequired must be a boolean when present.', + ); + }); + it('malformed hasMoreはrejectしcursorを進めない', async () => { await expectPullRejectsPreservingCursor( () => @@ -640,6 +914,234 @@ describe('OfflineReplicaPullService', () => { ]); }); + it.each(['stale put', 'remove'] as const)( + 'rebase overlay wins over projector %s for the same canonical row', + async (projectionMutation) => { + const executor = TestBed.inject(OFFLINE_COMMAND_EXECUTOR); + const otherScope = { userId: scope.userId, scopeId: '20' }; + const oldView: OfflineReplicaRow = { + ...scope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + values: { title: 'Old optimistic view' }, + confirmedValues: { title: 'Old confirmed view' }, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + }; + const rebase = vi.spyOn(executor, 'rebasePendingCommands').mockImplementation((commands, confirmed, revision) => ({ + steps: commands.map(() => ({ + optimisticValue: { ...(confirmed as Record), title: 'Rebased delta' }, + optimisticCompanions: [ + { + key: { + ...otherScope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + }, + before: null, + after: { + ...otherScope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'view-42' }, + values: { title: 'Rebased view' }, + confirmedValues: { title: 'Remote view' }, + serverRevision: null, + fetchedAt: 9, + syncState: 'pending', + }, + }, + ], + })), + })); + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-rebase', remoteId: 42 }, + values: { id: 42, title: 'Local delta' }, + confirmedValues: { id: 42, title: 'Old confirmed' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending', + }, + oldView, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-rebase', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-rebase' }, + operation: 'test_items.delta', + payload: { delta: 1 }, + optimisticValue: { id: 42, title: 'Local delta' }, + optimisticCompanions: [ + { + key: { ...otherScope, sourceKey: 'test_views', identity: { kind: 'local', localId: 'view-42' } }, + before: null, + after: oldView, + }, + ], + payloadHash: 'hash', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + }); + projector.project.mockResolvedValue( + projectionMutation === 'stale put' + ? { putRows: [oldView] } + : { removeRows: [{ ...scope, sourceKey: 'test_views', identity: { kind: 'local', localId: 'view-42' } }] }, + ); + pull.mockResolvedValueOnce(page([itemChange(42, 'Remote truth', { serverRevision: 9 })], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + expect(rebase).toHaveBeenCalledWith( + [expect.objectContaining({ commandId: 'cmd-rebase' })], + expect.objectContaining({ title: 'Remote truth' }), + 9, + [expect.objectContaining({ values: { title: 'Old optimistic view' } })], + ); + await expect(repository.getReplicaRow(scope, 'test_items', generatedCommandIdentity('019d-rebase'))).resolves.toMatchObject({ + values: { title: 'Rebased delta' }, + confirmedValues: { title: 'Remote truth' }, + serverRevision: 9, + syncState: 'pending', + }); + await expect(repository.getCommands(scope)).resolves.toEqual([ + expect.objectContaining({ + commandId: 'cmd-rebase', + payload: { delta: 1 }, + payloadHash: 'hash', + baseRevision: 9, + state: 'pending', + lastErrorCode: null, + }), + ]); + await expect(repository.getReplicaRow(scope, 'test_views', { kind: 'local', localId: 'view-42' })).resolves.toMatchObject({ + values: { title: 'Rebased view' }, + confirmedValues: { title: 'Remote view' }, + }); + await expect(repository.getReplicaCursor(scope)).resolves.toEqual({ ...scope, cursor: 'cursor-v1' }); + }, + ); + + 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'); + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: `019d-${state}`, remoteId: 42 }, + values: { id: 42, title: 'Local' }, + confirmedValues: { id: 42, title: 'Old' }, + serverRevision: 1, + fetchedAt: 1, + syncState: state, + }, + ], + putCommands: [ + { + ...scope, + commandId: `cmd-${state}`, + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: `019d-${state}` }, + operation: 'test_items.delta', + payload: { delta: 1 }, + optimisticValue: { id: 42, title: 'Local' }, + payloadHash: 'hash', + baseRevision: 1, + state, + attempts: 1, + retryAt: null, + createdAt: 1, + lastErrorCode: state, + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Remote', { serverRevision: 9 })], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + expect(rebase).not.toHaveBeenCalled(); + await expect(repository.getCommands(scope)).resolves.toEqual([ + expect.objectContaining({ commandId: `cmd-${state}`, state: 'conflict', lastErrorCode: 'remote_revision' }), + ]); + await expect(repository.getReplicaRow(scope, 'test_items', generatedCommandIdentity(`019d-${state}`))).resolves.toMatchObject({ + syncState: 'conflict', + }); + }); + + it('rebase reducer receives a pending-delete companion from its own scope', async () => { + const otherScope = { userId: scope.userId, scopeId: '20' }; + const companion: OfflineReplicaRow = { + ...otherScope, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'hidden-view' }, + values: { title: 'Hidden' }, + confirmedValues: { title: 'Confirmed hidden' }, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + visibility: 'pending_delete', + }; + const executor = TestBed.inject(OFFLINE_COMMAND_EXECUTOR); + const rebase = vi.spyOn(executor, 'rebasePendingCommands').mockReturnValue(null); + await repository.transactReplica({ + putRows: [ + { + ...scope, + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-cross-scope', remoteId: 42 }, + values: { id: 42, title: 'Local' }, + confirmedValues: { id: 42, title: 'Old' }, + serverRevision: 1, + fetchedAt: 1, + syncState: 'pending', + }, + companion, + ], + putCommands: [ + { + ...scope, + commandId: 'cmd-cross-scope', + aggregateType: 'test_items', + sourceKey: 'test_items', + identity: { kind: 'generated', localId: '019d-cross-scope' }, + operation: 'test_items.delta', + payload: { delta: 1 }, + optimisticValue: { id: 42, title: 'Local' }, + optimisticCompanions: [{ key: companion, before: null, after: companion }], + payloadHash: 'hash', + baseRevision: 1, + state: 'pending', + attempts: 0, + retryAt: null, + createdAt: 1, + lastErrorCode: null, + }, + ], + }); + pull.mockResolvedValueOnce(page([itemChange(42, 'Remote', { serverRevision: 9 })], { nextCursor: 'cursor-v1' })); + + await service.pull(scope); + + expect(rebase).toHaveBeenCalledWith(expect.any(Array), expect.anything(), 9, [ + expect.objectContaining({ scopeId: '20', visibility: 'pending_delete' }), + ]); + }); + it('remote tombstone conflictはpending commandをremote_deleted conflictへ遷移する', async () => { await repository.transactReplica({ putRows: [ 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 6499284..7e23ed7 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,12 @@ import { inject, Injectable } from '@angular/core'; import { OFFLINE_KIT_OPTIONS } from './offline-kit-options'; import { OFFLINE_COMMAND_EXECUTOR } from './offline-command-executor'; import { OfflineReplicaMutationCoordinator } from './offline-replica-mutation-coordinator'; -import { OFFLINE_REPLICA_PULLER, type OfflineReplicaChange, type OfflineReplicaPullPage } from './offline-replica-puller'; +import { + OFFLINE_REPLICA_PROJECTOR, + OFFLINE_REPLICA_PULLER, + type OfflineReplicaChange, + type OfflineReplicaPullPage, +} from './offline-replica-puller'; import { canonicalOfflineCommandIdentity, commandIdentityFromReplicaIdentity, @@ -21,6 +26,7 @@ import { } from './offline-replica-schema'; import { OFFLINE_REPOSITORY, + canonicalOfflineReplicaRowKey, type OfflineCommand, type OfflineReplicaRow, type OfflineReplicaRowKey, @@ -40,6 +46,7 @@ export class OfflineReplicaPullService { readonly #repository = inject(OFFLINE_REPOSITORY); 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 #replicaMutations = inject(OfflineReplicaMutationCoordinator); #schemaHash: Promise | null = null; @@ -47,34 +54,49 @@ export class OfflineReplicaPullService { async pull(scope: OfflineScope): Promise { if (this.#options.mode === 'readCacheOnly') return; const schemaHash = await (this.#schemaHash ??= sha256OfflineReplicaSchema(this.#options.replicaSchema)); - let cursor = (await this.#repository.getReplicaCursor(scope))?.cursor ?? ''; + let persistedCursor = (await this.#repository.getReplicaCursor(scope))?.cursor ?? ''; + let requestCursor = persistedCursor; + let rebaselinePending = false; for (;;) { const page = await this.#puller.pull({ scope, - cursor, + cursor: requestCursor, schemaVersion: this.#options.replicaSchema.version, schemaHash, }); this.#assertPullPage(page); this.#assertHandshake(page.schemaVersion, page.schemaHash, schemaHash); - if (page.hasMore && page.nextCursor === cursor) { + if (page.hasMore && page.nextCursor === requestCursor) { throw new Error(`Offline replica pull cursor did not advance for scope ${scope.userId}:${scope.scopeId}.`); } - if (page.changes.length === 0 && !page.hasMore && page.nextCursor === cursor) { + if (page.changes.length === 0 && !page.hasMore && page.nextCursor === requestCursor && !rebaselinePending) { return; } + if (page.rebaselineRequired && page.changes.length === 0 && !page.hasMore) { + throw new Error('Offline replica rebaseline marker must lead to a snapshot page.'); + } + if (page.rebaselineRequired && page.changes.length === 0 && page.hasMore) { + rebaselinePending = true; + requestCursor = page.nextCursor; + continue; + } const applied = await this.#replicaMutations.run(async () => { const currentCursor = (await this.#repository.getReplicaCursor(scope))?.cursor ?? ''; - if (currentCursor !== cursor) return currentCursor; + if (currentCursor !== persistedCursor) return currentCursor; 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 putRows: OfflineReplicaRow[] = []; const removeRows: OfflineReplicaRowKey[] = []; + const rebasedPutRows: OfflineReplicaRow[] = []; + const rebasedRemoveRows: OfflineReplicaRowKey[] = []; const putCommands = new Map(); const removeCommandIds = new Set(); + if (rebaselinePending || page.rebaselineRequired) { + removeRows.push(...(await this.#confirmedRowsForRebaseline(scope, userCommands))); + } for (const change of changes) { const schema = this.#entitySchema(change.sourceKey); @@ -170,10 +192,36 @@ export class OfflineReplicaPullService { continue; } - const conflicted = related.some((command) => command.baseRevision !== change.serverRevision); + const revisionChanged = related.some((command) => command.baseRevision !== change.serverRevision); + const rebaseEligible = related.every( + (command) => (command.state === 'pending' || command.state === 'retry_wait') && command.serverCommitUnknown !== true, + ); + const rebase = + revisionChanged && rebaseEligible + ? ((await this.#executor.rebasePendingCommands?.( + related, + confirmedValues, + change.serverRevision, + await this.#currentCompanionRows(related), + )) ?? null) + : null; + this.#assertRebasedSteps(related, rebase?.steps ?? null); + const conflicted = revisionChanged && rebase === null; + const rebasedCommands = rebase + ? related.map((command, index) => ({ + ...command, + baseRevision: change.serverRevision, + optimisticValue: rebase.steps[index]!.optimisticValue, + ...(rebase.steps[index]!.optimisticCompanions === undefined + ? {} + : { optimisticCompanions: rebase.steps[index]!.optimisticCompanions }), + })) + : []; + for (const command of rebasedCommands) putCommands.set(command.commandId, command); + this.#appendRebasedCompanionRows(rebasedCommands, rebasedPutRows, rebasedRemoveRows); putRows.push({ ...existing, - values: hasPending ? existing.values : confirmedValues, + values: rebasedCommands.at(-1)?.optimisticValue ?? (hasPending ? existing.values : confirmedValues), confirmedValues, serverRevision: change.serverRevision, fetchedAt: Date.now(), @@ -191,9 +239,21 @@ 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 ?? [] }, + { putRows: rebasedPutRows, removeRows: rebasedRemoveRows }, + ]); await this.#repository.transactReplica({ - putRows, - removeRows, + putRows: finalRows.putRows, + removeRows: finalRows.removeRows, putCommands: [...putCommands.values()], removeCommandIds: [...removeCommandIds], putCursors: [{ ...scope, cursor: page.nextCursor }], @@ -201,14 +261,67 @@ export class OfflineReplicaPullService { return page.nextCursor; }); if (applied !== page.nextCursor) { - cursor = applied; + persistedCursor = applied; + requestCursor = applied; + rebaselinePending = false; continue; } - cursor = page.nextCursor; + persistedCursor = page.nextCursor; + requestCursor = page.nextCursor; + rebaselinePending = false; if (!page.hasMore) return; } } + async #confirmedRowsForRebaseline(scope: OfflineScope, commands: readonly OfflineCommand[]): Promise { + const preserved = new Set( + commands.flatMap((command) => (command.optimisticCompanions ?? []).map((companion) => this.#rowKey(companion.key))), + ); + const rows = ( + await Promise.all(this.#options.replicaSchema.entities.map((entity) => this.#repository.getReplicaRows(scope, entity.sourceKey))) + ).flat(); + return rows.filter((row) => row.syncState === 'confirmed' && !preserved.has(this.#rowKey(row))); + } + + #rowKey(row: OfflineReplicaRowKey): string { + return canonicalOfflineReplicaRowKey(this.#entitySchema(row.sourceKey), row); + } + + #mergeRowMutations( + layers: readonly { + putRows: readonly OfflineReplicaRow[]; + removeRows: readonly OfflineReplicaRowKey[]; + }[], + ): { putRows: OfflineReplicaRow[]; removeRows: OfflineReplicaRowKey[] } { + const mutations = new Map(); + for (const layer of layers) { + for (const row of layer.removeRows) mutations.set(this.#rowKey(row), { kind: 'remove', row }); + for (const row of layer.putRows) mutations.set(this.#rowKey(row), { kind: 'put', row }); + } + const putRows: OfflineReplicaRow[] = []; + const removeRows: OfflineReplicaRowKey[] = []; + for (const mutation of mutations.values()) { + if (mutation.kind === 'put') putRows.push(mutation.row); + else removeRows.push(mutation.row); + } + return { putRows, removeRows }; + } + + #assertProjection( + scope: OfflineScope, + projection: { putRows?: readonly OfflineReplicaRow[]; removeRows?: readonly OfflineReplicaRowKey[] } | undefined, + ): void { + for (const row of [...(projection?.putRows ?? []), ...(projection?.removeRows ?? [])]) { + const schema = this.#entitySchema(row.sourceKey); + if (schema.identity.kind !== 'localOnly') { + throw new Error(`Offline replica projector may only mutate localOnly source "${row.sourceKey}".`); + } + if (row.userId !== scope.userId || row.scopeId !== scope.scopeId || row.identity.kind !== 'local') { + throw new Error('Offline replica projector rows must use the current scope and local identity.'); + } + } + } + #assertPullPage(page: OfflineReplicaPullPage): void { if (typeof page.nextCursor !== 'string') { throw new Error('Offline replica pull page nextCursor must be a string.'); @@ -219,11 +332,62 @@ export class OfflineReplicaPullService { if (!Array.isArray(page.changes)) { throw new Error('Offline replica pull page changes must be an array.'); } + if (page.rebaselineRequired !== undefined && typeof page.rebaselineRequired !== 'boolean') { + throw new Error('Offline replica pull page rebaselineRequired must be a boolean when present.'); + } for (const [index, change] of page.changes.entries()) { this.#assertPullChange(change, index); } } + #assertRebasedSteps( + original: readonly OfflineCommand[], + steps: readonly { optimisticCompanions?: readonly { key: OfflineReplicaRowKey }[] }[] | null, + ): void { + if (steps === null) return; + if (steps.length !== original.length) { + throw new Error('Rebased offline values must match the aggregate command count.'); + } + for (const [index, step] of steps.entries()) { + const originalKeys = (original[index]?.optimisticCompanions ?? []).map((item) => this.#rowKey(item.key)).sort(); + const rebasedKeys = (step.optimisticCompanions ?? []).map((item) => this.#rowKey(item.key)).sort(); + if (JSON.stringify(originalKeys) !== JSON.stringify(rebasedKeys)) { + throw new Error('Rebased offline companions must preserve each command footprint.'); + } + } + } + + #appendRebasedCompanionRows(commands: readonly OfflineCommand[], putRows: OfflineReplicaRow[], removeRows: OfflineReplicaRowKey[]): void { + const latest = new Map(); + for (const command of commands) { + for (const companion of command.optimisticCompanions ?? []) { + latest.set(this.#rowKey(companion.key), { key: companion.key, after: companion.after }); + } + } + for (const { key, after } of latest.values()) { + if (after) putRows.push(after); + else removeRows.push(key); + } + } + + async #currentCompanionRows(commands: readonly OfflineCommand[]): Promise { + const keys = new Map( + commands.flatMap((command) => + (command.optimisticCompanions ?? []).map((companion) => [this.#rowKey(companion.key), companion.key] as const), + ), + ); + const rows = await Promise.all( + [...keys.values()].map((key) => { + const companionScope = { userId: key.userId, scopeId: key.scopeId }; + return ( + this.#repository.getReplicaRowIncludingPendingDelete?.(companionScope, key.sourceKey, key.identity) ?? + this.#repository.getReplicaRow(companionScope, key.sourceKey, key.identity) + ); + }), + ); + return rows.filter((row): row is OfflineReplicaRow => row !== null); + } + #assertPullChange(change: unknown, index: number): void { const label = `Offline replica pull page changes[${index}]`; if (!isPlainObject(change)) { diff --git a/projects/kit/offline/src/lib/offline-replica-puller.ts b/projects/kit/offline/src/lib/offline-replica-puller.ts index 8439332..591da8d 100644 --- a/projects/kit/offline/src/lib/offline-replica-puller.ts +++ b/projects/kit/offline/src/lib/offline-replica-puller.ts @@ -1,5 +1,5 @@ import { InjectionToken } from '@angular/core'; -import type { OfflineScope } from './offline-repository'; +import type { OfflineReplicaRow, OfflineReplicaRowKey, OfflineScope } from './offline-repository'; import type { OfflineGeneratedRemoteId, OfflineNaturalKey, @@ -53,6 +53,24 @@ export interface OfflineReplicaPullPage { changes: readonly OfflineReplicaChange[]; nextCursor: string; hasMore: boolean; + /** The server can no longer continue this cursor and requires a confirmed-state snapshot rebuild. */ + rebaselineRequired?: boolean; +} + +/** Product projection derived from one collapsed authoritative pull page. */ +export interface OfflineReplicaPullProjection { + putRows?: readonly OfflineReplicaRow[]; + removeRows?: readonly OfflineReplicaRowKey[]; +} + +/** 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; } /** Backend response accepted by the shared pull-page normalizer. */ @@ -62,6 +80,8 @@ export interface OfflineReplicaWirePullPage { changes: readonly OfflineReplicaWireChange[]; nextCursor: string; hasMore: boolean; + /** The server can no longer continue this cursor and requires a confirmed-state snapshot rebuild. */ + rebaselineRequired?: boolean; } /** @@ -92,3 +112,6 @@ export interface OfflineReplicaPuller { /** DI token for the application-provided explicit replica pull transport. */ export const OFFLINE_REPLICA_PULLER = new InjectionToken('OFFLINE_REPLICA_PULLER'); + +/** Optional product adapter for local-only replica projections. */ +export const OFFLINE_REPLICA_PROJECTOR = new InjectionToken('OFFLINE_REPLICA_PROJECTOR'); diff --git a/projects/kit/offline/src/lib/offline-repository.ts b/projects/kit/offline/src/lib/offline-repository.ts index f811bc9..8634059 100644 --- a/projects/kit/offline/src/lib/offline-repository.ts +++ b/projects/kit/offline/src/lib/offline-repository.ts @@ -120,6 +120,15 @@ export interface OfflineReplicaRowKey extends OfflineScope { identity: OfflineReplicaIdentity; } +/** Canonical physical key shared by every repository and replica transaction coordinator. */ +export function canonicalOfflineReplicaRowKey( + schema: Pick>, 'scope'>, + row: OfflineReplicaRowKey, +): string { + const partition = schema.scope === 'user' ? 'user' : String(row.scopeId); + return `${canonicalOfflinePrincipalId(row.userId)}:${partition}:${row.sourceKey}:${canonicalOfflineReplicaIdentity(row.identity)}`; +} + /** Explicit one-way release of a generated remote id during a replica delete acknowledgement. */ export interface OfflineReplicaRemoteIdRelease extends OfflineReplicaRowKey { /** The current remote id being released; the matching put row must set `remoteId` to null. */ @@ -820,8 +829,7 @@ export class IonicOfflineRepository implements OfflineRepository { #rowKey(row: OfflineScope & { sourceKey: string; identity: OfflineReplicaIdentity }): string { const schema = this.#resolveReplicaEntitySchema(row.sourceKey); - const partition = schema.scope === 'user' ? 'user' : String(row.scopeId); - return `${canonicalOfflinePrincipalId(row.userId)}:${partition}:${row.sourceKey}:${canonicalOfflineReplicaIdentity(row.identity)}`; + return canonicalOfflineReplicaRowKey(schema, row); } #findRowByAddress( diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.ts index c3b4e95..6a87da0 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.ts @@ -28,6 +28,7 @@ import { } from './offline-replica-schema'; import { OFFLINE_SCHEMA_VERSION, + canonicalOfflineReplicaRowKey, type OfflineCommand, type OfflineCommandIdentity, type OfflinePrincipalId, @@ -905,7 +906,7 @@ export class SqliteOfflineRepository implements OfflineRepository { #replicaRowKey(row: OfflineReplicaRowKey): string { const schema = this.#resolveReplicaEntitySchema(row.sourceKey); - return `${canonicalOfflinePrincipalId(row.userId)}:${schema.scope === 'user' ? 'user' : row.scopeId}:${row.sourceKey}:${canonicalOfflineReplicaIdentity(row.identity)}`; + return canonicalOfflineReplicaRowKey(schema, row); } #assertReplicaIdentityAssignment(