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 9f6fa06..98d9f93 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.spec.ts @@ -12,7 +12,15 @@ import { OFFLINE_KIT_OPTIONS, type OfflineKitOptions } from './offline-kit-optio import { OfflineNetworkService } from './offline-network.service'; import { OfflineReplicaPullService } from './offline-replica-pull.service'; import { OfflineReplicaMutationCoordinator } from './offline-replica-mutation-coordinator'; -import { defineOfflineReplicaSchema, defineReplicaEntity, integer, naturalKey, generatedId, text } from './offline-replica-schema'; +import { + defineOfflineReplicaSchema, + defineReplicaEntity, + integer, + localOnly, + naturalKey, + generatedId, + text, +} from './offline-replica-schema'; import { canonicalOfflineReplicaIdentity, OFFLINE_REPOSITORY, @@ -43,6 +51,13 @@ const replicaSchema = defineOfflineReplicaSchema({ title: text(), }, }), + defineReplicaEntity<{ title: string }>()({ + table: 'document_views', + sourceKey: 'document_views', + scope: 'partition', + identity: localOnly(), + fields: { title: text() }, + }), ], migrations: [], }); @@ -3404,6 +3419,13 @@ describe('OfflineSyncService', () => { name: text(), }, }), + defineReplicaEntity<{ title: string }>()({ + table: 'test_views', + sourceKey: 'test_views', + scope: 'user', + identity: localOnly(), + fields: { title: text() }, + }), ], migrations: [], }); @@ -3414,7 +3436,7 @@ describe('OfflineSyncService', () => { { userId: 1, scopeId: '11' }, ] as OfflineScope[], }; - const userScopedSourceKeys = new Set(['test_items']); + const userScopedSourceKeys = new Set(['test_items', 'test_views']); function compareCommands(left: OfflineCommand, right: OfflineCommand): number { return left.createdAt - right.createdAt || (left.commandId < right.commandId ? -1 : left.commandId > right.commandId ? 1 : 0); @@ -3679,6 +3701,106 @@ describe('OfflineSyncService', () => { }); }); + it('multi-scope batchは同じuser-scoped companionのscope alias重複をcommit前に拒否する', async () => { + const repository = TestBed.inject(OFFLINE_REPOSITORY); + const transactReplica = vi.mocked(repository.transactReplica); + const companion = (scopeId: string, title: string): OfflineReplicaRow => ({ + userId: 1, + scopeId, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'shared-view' }, + values: { title }, + confirmedValues: { title: 'Baseline' }, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + }); + + await expect( + service.enqueuePreparedBatch( + async () => [ + { + request: { + scopeId: '10', + aggregateType: 'test_items', + identity: { kind: 'generated', localId: 'batch-scope-10' }, + operation: 'test_items.create', + payload: {}, + optimisticValue: { id: 0, title: 'A' }, + }, + replicaTransaction: { putRows: [companion('10', 'A')] }, + }, + { + request: { + scopeId: '11', + aggregateType: 'test_items', + identity: { kind: 'generated', localId: 'batch-scope-11' }, + operation: 'test_items.create', + payload: {}, + optimisticValue: { id: 0, title: 'B' }, + }, + replicaTransaction: { putRows: [companion('11', 'B')] }, + }, + ], + { flush: false }, + ), + ).rejects.toThrow('overlapping replica footprints'); + + expect(transactReplica).not.toHaveBeenCalled(); + expect(commands).toEqual([]); + expect(rows).toHaveLength(1); + }); + + it('別aggregateの後続enqueueもuser-scoped companionのscope alias共有を拒否する', async () => { + const companion = (scopeId: string, title: string): OfflineReplicaRow => ({ + userId: 1, + scopeId, + sourceKey: 'test_views', + identity: { kind: 'local', localId: 'shared-view' }, + values: { title }, + confirmedValues: { title: 'Baseline' }, + serverRevision: null, + fetchedAt: 1, + syncState: 'pending', + }); + const firstId = await service.enqueuePrepared( + async () => ({ + request: { + scopeId: '10', + aggregateType: 'test_items', + identity: { kind: 'generated', localId: 'scope-10-item' }, + operation: 'test_items.create', + payload: {}, + optimisticValue: { id: 0, title: 'A' }, + }, + replicaTransaction: { putRows: [companion('10', 'A')] }, + }), + { flush: false }, + ); + + await expect( + service.enqueuePrepared( + async () => ({ + request: { + scopeId: '11', + aggregateType: 'test_items', + identity: { kind: 'generated', localId: 'scope-11-item' }, + operation: 'test_items.create', + payload: {}, + optimisticValue: { id: 0, title: 'B' }, + }, + replicaTransaction: { putRows: [companion('11', 'B')] }, + }), + { flush: false }, + ), + ).rejects.toThrow('different aggregates cannot share a replica footprint'); + + expect(commands).toHaveLength(1); + await service.discard(firstId, { flush: false }); + expect(commands).toEqual([]); + expect(findReplicaRow({ userId: 1, scopeId: '11' }, 'test_views', { kind: 'local', localId: 'shared-view' })).toBeUndefined(); + }); + it('partition-scopedの同一localIdはpartitionごとに独立aggregateのまま並列送信する', async () => { let resolveFirst!: (value: OfflineCommandResult) => void; execute.mockImplementationOnce(() => new Promise((resolve) => (resolveFirst = resolve))); diff --git a/projects/kit/offline/src/lib/offline-sync.service.ts b/projects/kit/offline/src/lib/offline-sync.service.ts index 851f902..2d69169 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -23,10 +23,9 @@ import type { OfflineRepository, OfflineScope, } from './offline-repository'; -import { OFFLINE_REPOSITORY } from './offline-repository'; +import { canonicalOfflineReplicaRowKey, OFFLINE_REPOSITORY } from './offline-repository'; import { canonicalOfflinePrincipalId, - canonicalOfflineReplicaIdentity, canonicalOfflineCommandIdentity, commandIdentityFromReplicaIdentity, commandIdentityMatchesReplicaRow, @@ -350,7 +349,10 @@ export class OfflineSyncService { const session = await this.#beginEnqueueSession(generation); this.#assertEnqueueScope(session, request.scopeId); const materialization = await this.#materializeEnqueue(session.userId, request, replicaTransaction, replaced); - await this.#assertOutboxCapacity(session.userId, [materialization.command], replaced?.commandId); + const currentCommands = await this.#commandsForUser(session.userId); + const retainedCommands = replaced ? currentCommands.filter((command) => command.commandId !== replaced.commandId) : currentCommands; + this.#assertDistinctBatchFootprints([materialization], retainedCommands); + await this.#assertOutboxCapacity(session.userId, [materialization.command], replaced?.commandId, currentCommands); await this.#commitMaterializedEnqueues([materialization], generation, options, replaced ? [replaced.commandId] : undefined); return materialization.command.commandId; } @@ -372,16 +374,10 @@ export class OfflineSyncService { for (const [index, entry] of prepared.entries()) { this.#assertEnqueueScope(session, entry.request.scopeId); materializations.push( - await this.#materializeEnqueue( - session.userId, - entry.request, - entry.replicaTransaction, - undefined, - firstCreatedAt + index, - ), + await this.#materializeEnqueue(session.userId, entry.request, entry.replicaTransaction, undefined, firstCreatedAt + index), ); } - this.#assertDistinctBatchFootprints(materializations); + this.#assertDistinctBatchFootprints(materializations, currentCommands); await this.#assertOutboxCapacity( session.userId, materializations.map((item) => item.command), @@ -542,9 +538,14 @@ export class OfflineSyncService { return { command, optimisticRow, optimisticCompanions }; } - #assertDistinctBatchFootprints(entries: readonly MaterializedOfflineEnqueue[]): void { + #assertDistinctBatchFootprints(entries: readonly MaterializedOfflineEnqueue[], existingCommands: readonly OfflineCommand[]): void { const aggregates = new Set(); const replicaKeys = new Set(); + const existingFootprints = new Map(); + for (const command of existingCommands) { + const aggregate = this.#aggregateKey(command); + for (const key of this.#commandFootprintKeys(command)) existingFootprints.set(key, aggregate); + } for (const entry of entries) { const aggregate = this.#aggregateKey(entry.command); if (aggregates.has(aggregate)) { @@ -558,11 +559,26 @@ export class OfflineSyncService { if (replicaKeys.has(key)) { throw new Error('Prepared offline batch contains overlapping replica footprints.'); } + const existingAggregate = existingFootprints.get(key); + if (existingAggregate !== undefined && existingAggregate !== aggregate) { + throw new Error('Offline commands for different aggregates cannot share a replica footprint.'); + } replicaKeys.add(key); } } } + #commandFootprintKeys(command: OfflineCommand): readonly string[] { + const identity = + command.identity.kind === 'generated' + ? offlineGeneratedReplicaIdentity(command.identity.localId, null) + : ({ kind: 'natural', naturalKey: command.identity.naturalKey } as const); + return [ + this.#replicaRowKey({ ...command, identity }), + ...(command.optimisticCompanions ?? []).map((companion) => this.#replicaRowKey(companion.key)), + ]; + } + async #commitMaterializedEnqueues( entries: readonly MaterializedOfflineEnqueue[], generation: number, @@ -678,7 +694,7 @@ export class OfflineSyncService { } #replicaRowKey(key: OfflineReplicaRowKey): string { - return `${canonicalOfflinePrincipalId(key.userId)}:${key.scopeId}:${key.sourceKey}:${canonicalOfflineReplicaIdentity(key.identity)}`; + return canonicalOfflineReplicaRowKey(this.#entitySchema(key.sourceKey), key); } async #assertOutboxCapacity(