Support safe offline intent rebasing

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<OfflineCommandResult>;
   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<OfflinePendingRebase | null>;
   /**
    * 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<OfflineCommandExecutor>('OFFLINE_COMMAND_EXECUTOR');
 

projects/kit/offline/src/lib/offline-replica-pull.service.spec.ts

@@ -19,6 +19,7 @@ import {
   defineReplicaEntity,
   integer,
   generatedId,
+  localOnly,
   sha256OfflineReplicaSchema,
   text,
 } from './offline-replica-schema';
@@ -44,9 +45,17 @@ const testItemEntity = defineReplicaEntity<TestItemSelect>()({
   },
 });
 
+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: [],
 });
 
@@ -149,6 +158,7 @@ describe('OfflineReplicaPullService', () => {
           provide: OFFLINE_COMMAND_EXECUTOR,
           useValue: {
             execute: vi.fn(),
+            rebasePendingCommands: vi.fn(() => null),
             withServerRevision: (command: OfflineCommand, revision: string | number) => ({
               ...command,
               baseRevision: revision,
@@ -640,6 +650,214 @@ describe('OfflineReplicaPullService', () => {
     ]);
   });
 
+  it('rebase policyは他端末revisionへintentを移しconflictにしない', async () => {
+    const executor = TestBed.inject(OFFLINE_COMMAND_EXECUTOR);
+    const rebase = vi.spyOn(executor, 'rebasePendingCommands').mockImplementation((commands, confirmed, revision) => ({
+      steps: commands.map(() => ({
+        optimisticValue: { ...(confirmed as Record<string, unknown>), title: 'Rebased delta' },
+        optimisticCompanions: [{
+          key: {
+            ...scope,
+            sourceKey: 'test_views',
+            identity: { kind: 'local', localId: 'view-42' },
+          },
+          before: null,
+          after: {
+          ...scope,
+          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',
+        },
+      ],
+      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: { ...scope, sourceKey: 'test_views', identity: { kind: 'local', localId: 'view-42' } },
+            before: null,
+            after: null,
+          }],
+          payloadHash: 'hash',
+          baseRevision: 1,
+          state: 'pending',
+          attempts: 0,
+          retryAt: null,
+          createdAt: 1,
+          lastErrorCode: null,
+        },
+      ],
+    });
+    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,
+      [],
+    );
+    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' } });
+  });
+
+  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: [