From 619e6e4d0b4977645dbd5f17282b4c2d02108f8e Mon Sep 17 00:00:00 2001 From: rdlabo Date: Tue, 1 Sep 2026 09:07:17 +0900 Subject: [PATCH] refactor(offline): simplify repository mutations --- .../offline-replica-mutation-coordinator.ts | 2 +- .../lib/offline-replica-pull.service.spec.ts | 2 +- .../src/lib/offline-replica-pull.service.ts | 3 - .../src/lib/offline-repository.spec.ts | 10 ++- .../kit/offline/src/lib/offline-repository.ts | 34 ++++---- .../src/lib/offline-sync.service.spec.ts | 72 ++++++++--------- .../offline/src/lib/offline-sync.service.ts | 81 ++++++++----------- .../src/lib/sqlite-offline-repository.spec.ts | 8 +- .../src/lib/sqlite-offline-repository.ts | 28 +++---- tests/sqlite-offline-repository.node.spec.ts | 13 ++- 10 files changed, 117 insertions(+), 136 deletions(-) diff --git a/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.ts b/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.ts index 1259a6a3..d7c3bbf5 100644 --- a/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.ts +++ b/projects/kit/offline/src/lib/offline-replica-mutation-coordinator.ts @@ -42,7 +42,7 @@ export class OfflineReplicaMutationCoordinator { run(operation: (repository: OfflineRepository) => Promise): Promise { return this.#enqueue(() => { if (!this.#repository) throw new Error('Offline repository is not configured.'); - const atomicMutation = this.#repository?.[OFFLINE_REPOSITORY_ATOMIC_MUTATION]; + const atomicMutation = this.#repository[OFFLINE_REPOSITORY_ATOMIC_MUTATION]; return atomicMutation ? (atomicMutation.call(this.#repository, operation) as Promise) : operation(this.#repository); }); } 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 33bf4949..61b25aca 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 @@ -1528,7 +1528,7 @@ describe('OfflineReplicaPullService', () => { await expect(service.pull(scope)).rejects.toThrow('Replica remote id is immutable: current=42, incoming=43.'); await expect( - repository.getReplicaRowIncludingPendingDelete?.(scope, 'test_items', generatedCommandIdentity('019d-server-id-immutable')), + repository.getReplicaRowIncludingPendingDelete!(scope, 'test_items', generatedCommandIdentity('019d-server-id-immutable')), ).resolves.toMatchObject({ identity: { kind: 'generated', localId: '019d-server-id-immutable', remoteId: 42 }, visibility: 'pending_delete', 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 ddd6034d..9ac8374d 100644 --- a/projects/kit/offline/src/lib/offline-replica-pull.service.ts +++ b/projects/kit/offline/src/lib/offline-replica-pull.service.ts @@ -308,7 +308,6 @@ export class OfflineReplicaPullService { { putRows: projection?.putRows ?? [], removeRows: projection?.removeRows ?? [] }, ]); const rematerialized = await this.#rematerializePendingAggregates( - scope, userCommands, scopeCommands, confirmedAndProjected, @@ -488,7 +487,6 @@ export class OfflineReplicaPullService { } async #rematerializePendingAggregates( - scope: OfflineScope, userCommands: readonly OfflineCommand[], scopeCommands: readonly OfflineCommand[], currentRows: { putRows: readonly OfflineReplicaRow[]; removeRows: readonly OfflineReplicaRowKey[] }, @@ -558,7 +556,6 @@ export class OfflineReplicaPullService { putRows.push(...(mutations.putRows ?? [])); removeRows.push(...(mutations.removeRows ?? [])); } - void scope; return { putRows, removeRows }; } diff --git a/projects/kit/offline/src/lib/offline-repository.spec.ts b/projects/kit/offline/src/lib/offline-repository.spec.ts index 24d7bd06..ff5b160a 100644 --- a/projects/kit/offline/src/lib/offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/offline-repository.spec.ts @@ -810,7 +810,7 @@ describe('IonicOfflineRepository', () => { }); }); - it('同一createdAtはcommandId昇順で決定的に並べる', async () => { + it('同一createdAtを決定的に並べ、deprecated command aliasの互換性を保つ', async () => { const base: Omit = { userId: 1, aggregateType: 'test_items', @@ -829,6 +829,12 @@ describe('IonicOfflineRepository', () => { await repository.putCommand({ ...base, scopeId: '11', commandId: 'cmd-m', createdAt: 10 }); expect((await repository.getCommands({ userId: 1, scopeId: '10' })).map((item) => item.commandId)).toEqual(['cmd-a', 'cmd-z']); expect((await repository.getCommandsForUser!(1)).map((item) => item.commandId)).toEqual(['cmd-a', 'cmd-m', 'cmd-z']); + await repository.replaceCommand({ ...base, scopeId: '10', commandId: 'cmd-z', createdAt: 10, payload: { updated: true } }); + expect(await repository.getCommands({ userId: 1, scopeId: '10' })).toContainEqual( + expect.objectContaining({ commandId: 'cmd-z', payload: { updated: true } }), + ); + await repository.removeCommand('cmd-z'); + expect((await repository.getCommands({ userId: 1, scopeId: '10' })).map((item) => item.commandId)).toEqual(['cmd-a']); }); it('legacy web outboxの送信中と複数回試行済みの最終失敗をcommit不明として安全側へnormalizeする', async () => { @@ -1460,7 +1466,7 @@ describe('IonicOfflineRepository', () => { }); describe('pull attentions', () => { - it('put/get/removeとtransactionでuser+scope attentionを永続化する', async () => { + it('deprecated aliasとtransactionでuser+scope attentionをput/get/removeする', async () => { await repository.putPullAttention!({ userId: 1, scopeId: '10', diff --git a/projects/kit/offline/src/lib/offline-repository.ts b/projects/kit/offline/src/lib/offline-repository.ts index e0eccd4c..d4dcce32 100644 --- a/projects/kit/offline/src/lib/offline-repository.ts +++ b/projects/kit/offline/src/lib/offline-repository.ts @@ -258,11 +258,13 @@ export interface OfflineRepository { getCommands(scope: OfflineScope): Promise; getCommandsForUser?(userId: OfflinePrincipalId): Promise; putCommand(command: OfflineCommand): Promise; + /** @deprecated Use {@link transactReplica} with `putCommands`. */ replaceCommand(command: OfflineCommand): Promise; + /** @deprecated Use {@link transactReplica} with `removeCommandIds`. */ removeCommand(commandId: string): Promise; - /** Upserts a durable fatal-pull attention for user+scope. */ + /** @deprecated Use {@link transactReplica} with `putPullAttentions`. */ putPullAttention?(attention: OfflinePullAttention): Promise; - /** Removes a durable fatal-pull attention for user+scope when present. */ + /** @deprecated Use {@link transactReplica} with `removePullAttentions`. */ removePullAttention?(scope: OfflineScope): Promise; clearUser(userId: OfflinePrincipalId): Promise; clearScope(scope: OfflineScope): Promise; @@ -592,32 +594,24 @@ export class IonicOfflineRepository implements OfflineRepository { }); } + /** @deprecated Use {@link transactReplica} with `putCommands`. */ replaceCommand(command: OfflineCommand): Promise { return this.putCommand(command); } - async removeCommand(commandId: string): Promise { - await this.initialize(); - await this.#mutateRecord(OUTBOX_KEY, (commands) => { - delete commands[commandId]; - return commands; - }); + /** @deprecated Use {@link transactReplica} with `removeCommandIds`. */ + removeCommand(commandId: string): Promise { + return this.transactReplica({ removeCommandIds: [commandId] }); } - async putPullAttention(attention: OfflinePullAttention): Promise { - await this.initialize(); - await this.#mutateRecord(PULL_ATTENTIONS_KEY, (attentions) => { - attentions[this.#cursorKey(attention)] = attention; - return attentions; - }); + /** @deprecated Use {@link transactReplica} with `putPullAttentions`. */ + putPullAttention(attention: OfflinePullAttention): Promise { + return this.transactReplica({ putPullAttentions: [attention] }); } - async removePullAttention(scope: OfflineScope): Promise { - await this.initialize(); - await this.#mutateRecord(PULL_ATTENTIONS_KEY, (attentions) => { - delete attentions[this.#cursorKey(scope)]; - return attentions; - }); + /** @deprecated Use {@link transactReplica} with `removePullAttentions`. */ + removePullAttention(scope: OfflineScope): Promise { + return this.transactReplica({ removePullAttentions: [scope] }); } async clearUser(userId: OfflinePrincipalId): Promise { 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 994abd52..e4c59fd9 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.spec.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.spec.ts @@ -203,14 +203,6 @@ describe('OfflineSyncService', () => { return commands.filter((item) => item.userId === userId); }), putCommand, - replaceCommand: vi.fn(async (command: OfflineCommand) => { - commands = commands.filter((item) => item.commandId !== command.commandId); - commands.push(structuredClone(command)); - commands.sort((left, right) => left.createdAt - right.createdAt); - }), - removeCommand: vi.fn(async (commandId: string) => { - commands = commands.filter((item) => item.commandId !== commandId); - }), getReplicaRow: vi.fn(async (scope: OfflineScope, sourceKey: string, identity: OfflineReplicaAddress) => { await beforeGetReplicaRow?.(); return ( @@ -279,15 +271,6 @@ describe('OfflineSyncService', () => { getPullAttentions: vi.fn(async (userId: number) => pullAttentions.filter((attention) => attention.userId === userId).map((attention) => structuredClone(attention)), ), - putPullAttention: vi.fn(async (attention: OfflinePullAttention) => { - pullAttentions = pullAttentions.filter( - (candidate) => candidate.userId !== attention.userId || candidate.scopeId !== attention.scopeId, - ); - pullAttentions.push(structuredClone(attention)); - }), - removePullAttention: vi.fn(async (scope: OfflineScope) => { - pullAttentions = pullAttentions.filter((candidate) => candidate.userId !== scope.userId || candidate.scopeId !== scope.scopeId); - }), clearUser: vi.fn(async (userId: number) => { commands = commands.filter((item) => item.userId !== userId); rows = rows.filter((item) => item.userId !== userId); @@ -1656,8 +1639,6 @@ describe('OfflineSyncService', () => { getCommands: vi.fn(async () => []), getCommandsForUser: vi.fn(async () => []), putCommand: vi.fn(), - replaceCommand: vi.fn(), - removeCommand: vi.fn(), getReplicaRow: vi.fn(async () => null), getReplicaRowIncludingPendingDelete: vi.fn(async () => null), getReplicaRowByRemoteId: vi.fn(async () => null), @@ -3856,7 +3837,21 @@ describe('OfflineSyncService', () => { expect(JSON.stringify(service.pendingCommands()[1]?.payload)).toBe(JSON.stringify(second)); }); - it('JSON外payloadをrejectする', async () => { + it.each([ + ['undefined', () => ({ value: undefined })], + ['NaN', () => ({ value: Number.NaN })], + ['Infinity', () => ({ value: Number.POSITIVE_INFINITY })], + [ + '循環参照', + () => { + const value: Record = {}; + value['self'] = value; + return value; + }, + ], + ['非plain object', () => ({ value: new Date(0) })], + ['array hole', () => ({ value: Array(1) })], + ])('JSON外payload（%s）をrejectする', async (_case, payload) => { await expect( service.enqueue( { @@ -3864,13 +3859,29 @@ describe('OfflineSyncService', () => { aggregateType: 'documents', identity: { kind: 'generated', localId: '1' }, operation: 'documents.upsert', - payload: { value: undefined }, + payload: payload(), }, { flush: false }, ), ).rejects.toBeInstanceOf(OfflinePayloadValidationError); }); + it('JSON payload内の共有参照は循環参照として扱わない', async () => { + const shared = { value: 'shared' }; + await expect( + service.enqueue( + { + scopeId: '10', + aggregateType: 'documents', + identity: { kind: 'generated', localId: 'shared-reference' }, + operation: 'documents.upsert', + payload: { left: shared, right: shared }, + }, + { flush: false }, + ), + ).resolves.toEqual(expect.any(String)); + }); + it.each([ [401, 'blocked_auth', 'blocked_auth'], [409, 'conflict', 'conflict'], @@ -3973,13 +3984,6 @@ describe('OfflineSyncService', () => { commands = commands.filter((item) => item.commandId !== command.commandId); commands.push(structuredClone(command)); }), - replaceCommand: vi.fn(async (command: OfflineCommand) => { - commands = commands.filter((item) => item.commandId !== command.commandId); - commands.push(structuredClone(command)); - }), - removeCommand: vi.fn(async (commandId: string) => { - commands = commands.filter((item) => item.commandId !== commandId); - }), getReplicaRow: vi.fn(async (scope: OfflineScope, sourceKey: string, identity: OfflineReplicaAddress) => { return ( rows.find((item) => { @@ -5793,18 +5797,14 @@ describe('OfflineSyncService', () => { commands.push(structuredClone(command)); commands.sort(compareCommands); }), - replaceCommand: vi.fn(async (command: OfflineCommand) => { - commands = commands.filter((item) => item.commandId !== command.commandId); - commands.push(structuredClone(command)); - commands.sort(compareCommands); - }), - removeCommand: vi.fn(async (commandId: string) => { - commands = commands.filter((item) => item.commandId !== commandId); - }), getReplicaRow: vi.fn(async (scope: OfflineScope, sourceKey: string, identity: OfflineReplicaAddress) => { const row = findReplicaRow(scope, sourceKey, identity); return row ? projectReplicaRow(row, scope) : null; }), + getReplicaRowIncludingPendingDelete: vi.fn(async (scope: OfflineScope, sourceKey: string, identity: OfflineReplicaAddress) => { + const row = findReplicaRow(scope, sourceKey, identity); + return row ? projectReplicaRow(row, scope) : null; + }), getReplicaRowByRemoteId: vi.fn(async (scope: OfflineScope, sourceKey: string, remoteId: number) => { const row = rows.find((item) => { if ( diff --git a/projects/kit/offline/src/lib/offline-sync.service.ts b/projects/kit/offline/src/lib/offline-sync.service.ts index 1b0ba721..cb76629c 100644 --- a/projects/kit/offline/src/lib/offline-sync.service.ts +++ b/projects/kit/offline/src/lib/offline-sync.service.ts @@ -562,11 +562,11 @@ export class OfflineSyncService { const scope = { userId, scopeId: request.scopeId }; this.noteScope(scope); const commandIdentity = offlineCommandLookupIdentity(request.identity); - const normalized = await this.#normalizeEnqueueRequest(scope, request, commandIdentity, repository); const commandId = request.commandId ?? crypto.randomUUID(); if (typeof commandId !== 'string' || commandId.length === 0 || commandId.length > 255) { throw new Error('Offline command id must contain between 1 and 255 characters.'); } + assertOfflinePayload(request.payload); const sourceKey = this.#hooks.entityType(request); const localOnlyFootprint = this.#normalizedLocalOnlyFootprint(scope, request.localOnlyFootprint); if (replaced) this.#assertReplacementFootprint(replaced, localOnlyFootprint); @@ -577,9 +577,9 @@ export class OfflineSyncService { sourceKey, identity: commandIdentity, operation: request.operation, - payload: normalized.payload, + payload: request.payload, replicaMutation: request.replicaMutation ?? 'upsert', - baseRevision: normalized.baseRevision, + baseRevision: request.baseRevision ?? null, state: 'pending', attempts: 0, retryAt: null, @@ -597,21 +597,23 @@ export class OfflineSyncService { ) { throw new Error('Offline replacement command must address the same aggregate and replica identity.'); } - const entityType = command.sourceKey; - const schema = this.#entitySchema(entityType); + const schema = this.#entitySchema(sourceKey); if (schema.identity.kind === 'localOnly') { - throw new Error(`Offline replica source "${entityType}" is local-only and cannot be added to the Outbox.`); + throw new Error(`Offline replica source "${sourceKey}" is local-only and cannot be added to the Outbox.`); } if (schema.identity.kind === 'generated' && request.identity.kind !== 'generated') { - throw new Error(`Offline replica source "${entityType}" requires generated identity.`); + throw new Error(`Offline replica source "${sourceKey}" requires generated identity.`); } if (schema.identity.kind === 'naturalKey' && request.identity.kind !== 'natural') { - throw new Error(`Offline replica source "${entityType}" requires natural identity.`); + throw new Error(`Offline replica source "${sourceKey}" requires natural identity.`); } if (request.replicaMutation === 'delete' && !repository.getReplicaRowIncludingPendingDelete) { throw new Error('Offline repository does not support durable replica delete tombstones.'); } - const existing = await this.#getReplicaRowForSync(scope, entityType, commandIdentity, repository); + const existing = await this.#getReplicaRowForSync(scope, sourceKey, commandIdentity, repository); + if (existing?.serverRevision != null && existing.serverRevision !== command.baseRevision) { + command = { ...command, baseRevision: existing.serverRevision }; + } const generatedIdentity = request.identity.kind === 'generated' ? request.identity : null; const initialRemoteId = this.#initialRemoteId( schema, @@ -635,7 +637,7 @@ export class OfflineSyncService { canonicalOfflineRemoteIdentity(schema, { naturalKey: request.identity.naturalKey }) !== canonicalOfflineRemoteIdentity(schema, { naturalKey: naturalKey! }) ) { - throw new Error(`Offline command naturalKey must match replica identity for "${entityType}".`); + throw new Error(`Offline command naturalKey must match replica identity for "${sourceKey}".`); } const remoteIdentity = schema.identity.kind === 'generated' @@ -654,11 +656,11 @@ export class OfflineSyncService { existing.identity.kind === 'natural' ? existing.identity.naturalKey : offlineNaturalKeyFromValues(schema, existing.values)!, }) !== canonicalValuesKey ) { - throw new Error(`Offline replica naturalKey is immutable and must match command identity for "${entityType}".`); + throw new Error(`Offline replica naturalKey is immutable and must match command identity for "${sourceKey}".`); } } if (remoteIdentity !== null) { - const mapped = await repository.getReplicaRowByRemoteIdentity(scope, entityType, remoteIdentity); + const mapped = await repository.getReplicaRowByRemoteIdentity(scope, sourceKey, remoteIdentity); if (mapped !== null && !commandIdentityMatchesReplicaRow(schema, mapped, commandIdentity)) { if ('remoteId' in remoteIdentity) { throw new Error(`Offline replica remote id ${String(remoteIdentity.remoteId)} is already mapped to another row.`); @@ -666,7 +668,6 @@ export class OfflineSyncService { throw new Error(`Offline replica remote identity is already mapped to another row.`); } } - this.#canonicalJson(normalized.payload); const seedBaseRow = existing ? undefined : request.identity.kind === 'generated' @@ -1484,21 +1485,6 @@ export class OfflineSyncService { return true; } - async #normalizeEnqueueRequest( - scope: OfflineScope, - request: EnqueueOfflineCommand, - commandIdentity: OfflineCommandIdentity, - repository: OfflineRepository, - ): Promise<{ payload: T; baseRevision: string | number | null }> { - let baseRevision = request.baseRevision ?? null; - const sourceKey = this.#hooks.entityType(request); - const row = await this.#getReplicaRowForSync(scope, sourceKey, commandIdentity, repository); - if (row?.serverRevision != null && row.serverRevision !== baseRevision) { - baseRevision = row.serverRevision; - } - return { payload: request.payload, baseRevision }; - } - async #completeCommand( commands: OfflineCommand[], command: OfflineCommand, @@ -2102,32 +2088,29 @@ export class OfflineSyncService { await Promise.allSettled([...this.#sendingTransitions, ...this.#commandSendTransitions.values()]); } - #canonicalJson(value: unknown): string { - if (Array.isArray(value)) return `[${value.map((item) => this.#canonicalJson(item)).join(',')}]`; - if (value !== null && typeof value === 'object') { - if (Object.getPrototypeOf(value) !== Object.prototype) { - throw new OfflinePayloadValidationError(); - } - return `{${Object.entries(value) - .sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0)) - .map(([key, item]) => `${JSON.stringify(key)}:${this.#canonicalJson(item)}`) - .join(',')}}`; - } - if (typeof value === 'number' && !Number.isFinite(value)) { - throw new OfflinePayloadValidationError(); - } - const serialized = JSON.stringify(value); - if (typeof serialized !== 'string') { - throw new OfflinePayloadValidationError(); - } - return serialized; - } - #scopeKey(scope: OfflineScope): string { return `${canonicalOfflinePrincipalId(scope.userId)}:${scope.scopeId}`; } } +function assertOfflinePayload(value: unknown, ancestors = new Set()): void { + if (value === null || typeof value === 'string' || typeof value === 'boolean') return; + if (typeof value === 'number') { + if (Number.isFinite(value)) return; + throw new OfflinePayloadValidationError(); + } + if (typeof value !== 'object' || ancestors.has(value)) { + throw new OfflinePayloadValidationError(); + } + if (!Array.isArray(value) && Object.getPrototypeOf(value) !== Object.prototype) { + throw new OfflinePayloadValidationError(); + } + ancestors.add(value); + const items = Array.isArray(value) ? value : Object.values(value); + for (const item of items) assertOfflinePayload(item, ancestors); + ancestors.delete(value); +} + function compareOfflineCommands(left: OfflineCommand, right: OfflineCommand): number { return left.createdAt - right.createdAt || (left.commandId < right.commandId ? -1 : left.commandId > right.commandId ? 1 : 0); } 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 63cedf16..e30f7e87 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.spec.ts @@ -1006,7 +1006,7 @@ describe('SqliteOfflineRepository community sqlite driver', () => { ).resolves.toEqual([]); }); - it('pull attentionをput/getしtransactionでupsertする', async () => { + it('pull attentionをtransactionでput/get/upsertする', async () => { const repository = createRepository(); await repository.initialize(); expect(plugin.execute).toHaveBeenCalledWith( @@ -1014,10 +1014,8 @@ describe('SqliteOfflineRepository community sqlite driver', () => { statement: expect.stringContaining('CREATE TABLE IF NOT EXISTS offline_pull_attentions'), }), ); - await repository.putPullAttention!({ - userId: 1, - scopeId: '10', - reason: 'schema_upgrade_required', + await repository.transactReplica({ + putPullAttentions: [{ userId: 1, scopeId: '10', reason: 'schema_upgrade_required' }], }); expect(plugin.execute).toHaveBeenCalledWith( expect.objectContaining({ diff --git a/projects/kit/offline/src/lib/sqlite-offline-repository.ts b/projects/kit/offline/src/lib/sqlite-offline-repository.ts index 9dfcc1a9..7ad6c117 100644 --- a/projects/kit/offline/src/lib/sqlite-offline-repository.ts +++ b/projects/kit/offline/src/lib/sqlite-offline-repository.ts @@ -510,32 +510,24 @@ export class SqliteOfflineRepository implements OfflineRepository { await this.#queueWrite((databaseId) => this.#putCommand(databaseId, command)); } + /** @deprecated Use {@link transactReplica} with `putCommands`. */ replaceCommand(command: OfflineCommand): Promise { return this.putCommand(command); } - async removeCommand(commandId: string): Promise { - await this.#write('DELETE FROM offline_sync_commands WHERE command_id = ?', [commandId]); + /** @deprecated Use {@link transactReplica} with `removeCommandIds`. */ + removeCommand(commandId: string): Promise { + return this.transactReplica({ removeCommandIds: [commandId] }); } - async putPullAttention(attention: OfflinePullAttention): Promise { - await this.#write( - `INSERT INTO offline_pull_attentions (user_id, scope_id, reason, status) VALUES (?, ?, ?, ?) - ON CONFLICT(user_id, scope_id) DO UPDATE SET reason = excluded.reason, status = excluded.status`, - [ - canonicalOfflinePrincipalId(attention.userId), - attention.scopeId, - attention.reason, - attention.status === undefined ? null : attention.status, - ], - ); + /** @deprecated Use {@link transactReplica} with `putPullAttentions`. */ + putPullAttention(attention: OfflinePullAttention): Promise { + return this.transactReplica({ putPullAttentions: [attention] }); } - async removePullAttention(scope: OfflineScope): Promise { - await this.#write('DELETE FROM offline_pull_attentions WHERE user_id = ? AND scope_id = ?', [ - canonicalOfflinePrincipalId(scope.userId), - scope.scopeId, - ]); + /** @deprecated Use {@link transactReplica} with `removePullAttentions`. */ + removePullAttention(scope: OfflineScope): Promise { + return this.transactReplica({ removePullAttentions: [scope] }); } async clearUser(userId: OfflinePrincipalId): Promise { diff --git a/tests/sqlite-offline-repository.node.spec.ts b/tests/sqlite-offline-repository.node.spec.ts index 6c98ff14..93a3c96a 100644 --- a/tests/sqlite-offline-repository.node.spec.ts +++ b/tests/sqlite-offline-repository.node.spec.ts @@ -213,7 +213,7 @@ describe('SqliteOfflineRepository real SQLite contract', () => { await expect(repository.getCommandsForUser(8)).resolves.toEqual([otherUser]); }); - it('replaces an Outbox state without retaining optional values, then removes it', async () => { + it('keeps the deprecated Outbox mutation aliases compatible', async () => { const repository = createRepository(); await repository.initialize(); const scope = { userId: 7 as const, scopeId: 'demo' }; @@ -253,6 +253,17 @@ describe('SqliteOfflineRepository real SQLite contract', () => { await expect(repository.getCommands(scope)).resolves.toEqual([]); }); + it('keeps the deprecated pull-attention mutation aliases compatible', async () => { + const repository = createRepository(); + await repository.initialize(); + const attention = { userId: 7 as const, scopeId: 'demo', reason: 'schema_upgrade_required' as const }; + + await repository.putPullAttention(attention); + await expect(repository.getPullAttentions(7)).resolves.toEqual([attention]); + await repository.removePullAttention({ userId: 7, scopeId: 'demo' }); + await expect(repository.getPullAttentions(7)).resolves.toEqual([]); + }); + it('atomically rolls back an Outbox enqueue when a later real SQLite write fails', async () => { const repository = createRepository(); await repository.initialize();