From 298928e6d2f1eb582c846a521255ca98981b7920 Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Tue, 12 May 2026 16:04:11 +0200 Subject: [PATCH] refactor(sync-core): prepare core boundary for providers --- .../sync-core-extraction-plan.md | 104 +++-- packages/sync-core/src/entity-key.util.ts | 2 +- packages/sync-core/src/index.ts | 7 + .../sync-core/src/lww-update-action-types.ts | 4 +- packages/sync-core/src/replay-coordinator.ts | 216 ++++++++++ .../tests/replay-coordinator.spec.ts | 392 ++++++++++++++++++ .../op-log/apply/operation-applier.service.ts | 126 ++---- 7 files changed, 726 insertions(+), 125 deletions(-) create mode 100644 packages/sync-core/src/replay-coordinator.ts create mode 100644 packages/sync-core/tests/replay-coordinator.spec.ts diff --git a/docs/long-term-plans/sync-core-extraction-plan.md b/docs/long-term-plans/sync-core-extraction-plan.md index 0283ace1ee..505093394d 100644 --- a/docs/long-term-plans/sync-core-extraction-plan.md +++ b/docs/long-term-plans/sync-core-extraction-plan.md @@ -3,9 +3,10 @@ > **Status: In progress - PR 1, PR 2 guardrails/logger adapter work, PR 3a > vector-clock ownership, full-state op classification config, PR 3b pure > helper slices, PR 4a port contracts, and the current PR 4b small -> orchestration/planning helper set are present on this branch. Remaining -> cleanup is targeted future `SyncLogger` routing for files as they move plus -> deciding whether PR 4c should extract any part of `OperationApplierService`.** +> orchestration/planning helper set are present on this branch. PR 4c now has +> the narrow operation replay coordinator extracted while the Angular shell +> stays app-side. Remaining cleanup is targeted future `SyncLogger` routing for +> files as they move plus provider/package boundary work.** **Goal:** Carve the sync engine out of `src/app/op-log/` into a reusable, framework-agnostic, **domain-agnostic** `@sp/sync-core` package, plus a sibling @@ -164,20 +165,32 @@ Current extraction state and remaining immediate debt: follow-up partitioning, download gap/full-state/encryption planning, and file-snapshot hydration skip planning. Provider calls, encryption/decryption, IndexedDB reads, UI, diagnostics, and result assembly remain app-side. +- PR 4c is present with `replayOperationBatch()` in `@sp/sync-core`. It owns + only the strict replay ordering around remote-apply windows, bulk dispatch, + the required event-loop yield, archive side-effect processing, post-sync + cooldown, and deferred local-action flushing. The Angular + `OperationApplierService` still owns NgRx action construction, + operation-to-action conversion, archive predicates, `remoteArchiveDataApplied`, + `Injector` usage, and diagnostics. +- Pre-P5 readiness cleanup is complete for this branch: movable core code no + longer depends on `OpLog`, generic prefix/error/compression helpers are + package-side with app-owned diagnostics, sync-core source comments were + rechecked for SP entity examples, and the core boundary grep was rerun with no + forbidden source imports. Suggested next order: -1. Finish PR 2 documentation and verification. -2. Continue targeted `SyncLogger` routing for files as they become movable. -3. Treat the PR 3b pure conflict-resolution and sync-import slices as complete +1. Treat PR 2 documentation/verification and the targeted `SyncLogger` routing + needed before provider extraction as complete for this branch. Continue + logger routing only when additional files actually move. +2. Treat the PR 3b pure conflict-resolution and sync-import slices as complete for this round. -4. Treat the current PR 4a/4b port and small-helper slices as complete for this - round; only revisit `OperationApplierService` under PR 4c after more - verification. -5. Continue logger/config cleanup before moving app error classes; prefix - parsing/formatting, generic error-message extraction, and generic - compression helpers now have package-side helpers with app-owned diagnostics. -6. Defer provider extraction until core boundaries and PR 4c are settled. +3. Treat the current PR 4a/4b/4c port, small-helper, and replay-coordinator + slices as complete for this round; keep the Angular `OperationApplierService` + shell app-side unless a later port proves another small extraction safe. +4. Start PR 5 provider extraction after this branch is merged. Provider-specific + `SyncLog`/`OpLog` routing should be handled inside PR 5 as provider files + move behind provider-package ports. ## PR 1 - Thin First Slice (#7546) @@ -846,6 +859,28 @@ IndexedDB reads, and result assembly remain app-side. Only after 4a/4b are stable, decide whether any part of `OperationApplierService` belongs in `@sp/sync-core`. +Status: implemented for the current branch slice. The extracted part is the +narrow `replayOperationBatch()` coordinator in `packages/sync-core/src/replay-coordinator.ts`. +It is intentionally generic and calls host-supplied ports/callbacks in a strict +order: + +1. open the remote-apply window; +2. dispatch the host-created bulk replay action; +3. yield after dispatch so host reducers finish before side effects; +4. run remote archive side effects after dispatch when configured; +5. yield around archive side effects to preserve UI responsiveness; +6. start post-sync cooldown before ending the remote-apply window; +7. end the remote-apply window and flush deferred local actions. + +Package-level Vitest coverage now asserts dispatch-yield ordering, local +hydration behavior, archive failure reporting, archive notification timing, +cooldown failure handling, and empty-batch no-op behavior. + +The Angular `OperationApplierService` delegates to this coordinator but keeps +all app-specific work app-side: `bulkApplyOperations`, `convertOpToAction`, +`isArchiveAffectingAction`, `remoteArchiveDataApplied`, `Injector` access to +`OperationLogEffects`, and `OpLog` diagnostics. + Acceptable extraction: - a small generic replay coordinator that calls ports in a strict order; @@ -870,6 +905,27 @@ Hard requirements from `CLAUDE.md`: --- +## Pre-P5 Readiness Check + +Status: complete for this branch. + +- No remaining pre-P5 `SyncLogger` routing is needed in core: files already made + movable either live in `@sp/sync-core` without app logging, accept a + `SyncLogger` port, or stay app-side because their diagnostics/recovery + behavior is still SP-specific. +- `sync-file-prefix`, generic error-message extraction, and gzip/base64 + compression helpers are package-side behind host-owned configuration/error + factories. +- `OperationApplierService` logging remains app-side intentionally; the moved + replay coordinator has no logging dependency. +- Provider-specific logging and credential diagnostics remain in + `src/app/op-log/sync-providers/` and should be handled during PR 5 when those + files move to `@sp/sync-providers`. +- Boundary verification was rerun for `packages/sync-core/src` and found no + forbidden Angular, NgRx, `src/app`, or `@sp/shared-schema` imports. + +--- + ## PR 5 - Lift Providers Into `@sp/sync-providers` Pull bundled providers out of `src/app/op-log/sync-providers/` so engine, @@ -934,17 +990,17 @@ This is now a final audit rather than the first boundary rule. ## Summary Timeline -| PR | Scope | Risk | Notes | -| ------ | ---------------------------------------------------------- | ----------- | -------------------------------------- | -| **1** | Stand up `@sp/sync-core` with generic primitives and stubs | Low | Present on branch | -| **2** | Boundary lint, registry types, privacy-aware logger port | Medium | Groundwork present; finish follow-ups | -| **3a** | Vector-clock ownership and package test harness | Medium | Present on branch | -| **3b** | Pure algorithmic core | Medium | No Angular/NgRx/IndexedDB | -| **4a** | Port contracts only | Medium | Current slice present | -| **4b** | Move small orchestration units behind ports | High | Current slice present | -| **4c** | Revisit `OperationApplierService` extraction | High | Extract only if the boundary is proven | -| **5** | Lift providers into `@sp/sync-providers` | Medium-High | Provider deps stay out of core | -| **6** | Final boundary hardening and architecture note | Low | Audit and lock down | +| PR | Scope | Risk | Notes | +| ------ | ---------------------------------------------------------- | ----------- | ------------------------------------- | +| **1** | Stand up `@sp/sync-core` with generic primitives and stubs | Low | Present on branch | +| **2** | Boundary lint, registry types, privacy-aware logger port | Medium | Groundwork present; finish follow-ups | +| **3a** | Vector-clock ownership and package test harness | Medium | Present on branch | +| **3b** | Pure algorithmic core | Medium | No Angular/NgRx/IndexedDB | +| **4a** | Port contracts only | Medium | Current slice present | +| **4b** | Move small orchestration units behind ports | High | Current slice present | +| **4c** | Revisit `OperationApplierService` extraction | High | Narrow replay coordinator present | +| **5** | Lift providers into `@sp/sync-providers` | Medium-High | Provider deps stay out of core | +| **6** | Final boundary hardening and architecture note | Low | Audit and lock down | After the final PR, `@sp/sync-core` should be the domain-agnostic sync engine and abstractions, `@sp/sync-providers` should contain bundled provider diff --git a/packages/sync-core/src/entity-key.util.ts b/packages/sync-core/src/entity-key.util.ts index 52e44ee530..6b35134485 100644 --- a/packages/sync-core/src/entity-key.util.ts +++ b/packages/sync-core/src/entity-key.util.ts @@ -2,7 +2,7 @@ * Creates a unique key for an entity by combining its type and ID. * Used for indexing and looking up entities across the operation log system. * - * @param entityType The type of the entity (e.g., 'TASK', 'PROJECT') + * @param entityType The host-defined type of the entity * @param entityId The unique ID of the entity * @returns A composite key in the format "ENTITY_TYPE:entityId" */ diff --git a/packages/sync-core/src/index.ts b/packages/sync-core/src/index.ts index 1434a28629..e0bcc71357 100644 --- a/packages/sync-core/src/index.ts +++ b/packages/sync-core/src/index.ts @@ -74,6 +74,13 @@ export type { LwwUpdateActionTypeHelpers } from './lww-update-action-types'; // Apply-operation result and option types. export type { ApplyOperationsResult, ApplyOperationsOptions } from './apply.types'; +// Generic operation replay coordinator. +export { replayOperationBatch, yieldToEventLoop } from './replay-coordinator'; +export type { + OperationReplayArchiveFailureContext, + OperationReplayCoordinatorOptions, +} from './replay-coordinator'; + // Remote operation application coordinator. export { applyRemoteOperations } from './remote-apply'; export type { diff --git a/packages/sync-core/src/lww-update-action-types.ts b/packages/sync-core/src/lww-update-action-types.ts index 9810a094c2..1bc4197035 100644 --- a/packages/sync-core/src/lww-update-action-types.ts +++ b/packages/sync-core/src/lww-update-action-types.ts @@ -22,8 +22,8 @@ export interface LwwUpdateActionTypeHelpers * Build LWW helpers for the given list of entity types. * * @example - * const lww = createLwwUpdateActionTypeHelpers(['TASK', 'PROJECT'] as const); - * lww.toLwwUpdateActionType('TASK'); // '[TASK] LWW Update' + * const lww = createLwwUpdateActionTypeHelpers(['ITEM', 'COLLECTION'] as const); + * lww.toLwwUpdateActionType('ITEM'); // '[ITEM] LWW Update' */ export const createLwwUpdateActionTypeHelpers = ( entityTypes: readonly TEntityType[], diff --git a/packages/sync-core/src/replay-coordinator.ts b/packages/sync-core/src/replay-coordinator.ts new file mode 100644 index 0000000000..6cc02b825a --- /dev/null +++ b/packages/sync-core/src/replay-coordinator.ts @@ -0,0 +1,216 @@ +import type { ApplyOperationsOptions, ApplyOperationsResult } from './apply.types'; +import type { Operation } from './operation.types'; +import type { + ActionDispatchPort, + ArchiveSideEffectPort, + DeferredLocalActionsPort, + RemoteApplyWindowPort, + SyncActionLike, +} from './ports'; + +interface ReplayGlobals { + setTimeout?: (handler: () => void, timeout?: number) => unknown; +} + +export interface OperationReplayArchiveFailureContext< + TOperation extends Operation = Operation, +> { + op: TOperation; + processedCount: number; + error: unknown; +} + +export interface OperationReplayCoordinatorOptions< + TOperation extends Operation = Operation, + TBulkAction extends SyncActionLike = SyncActionLike, + TReplayAction extends SyncActionLike = SyncActionLike, +> { + ops: TOperation[]; + applyOptions?: ApplyOperationsOptions; + dispatcher: ActionDispatchPort; + createBulkApplyAction: (ops: TOperation[]) => TBulkAction; + remoteApplyWindow: RemoteApplyWindowPort; + deferredLocalActions: DeferredLocalActionsPort; + archiveSideEffects?: ArchiveSideEffectPort; + operationToAction?: (op: TOperation) => TReplayAction; + isArchiveAffectingAction?: (action: TReplayAction) => boolean; + onRemoteArchiveDataApplied?: () => void; + onArchiveSideEffectError?: ( + context: OperationReplayArchiveFailureContext, + ) => void; + onPostSyncCooldownError?: (error: unknown) => void; + yieldToEventLoop?: () => Promise; +} + +interface ArchiveSideEffectResult = Operation> { + appliedOps: TOperation[]; + hadArchiveAffectingOp: boolean; + failedOp?: { + op: TOperation; + error: Error; + }; +} + +const globals = (): ReplayGlobals => globalThis as unknown as ReplayGlobals; + +export const yieldToEventLoop = (): Promise => + new Promise((resolve) => { + const setTimeoutFn = globals().setTimeout; + if (setTimeoutFn === undefined) { + void Promise.resolve().then(resolve); + return; + } + setTimeoutFn(resolve, 0); + }); + +/** + * Replays an operation batch through host ports in the order required by sync. + * + * Host applications keep framework-specific action construction, operation + * conversion, archive predicates, diagnostics, and UI notifications outside the + * package. This coordinator only owns the generic ordering: + * + * 1. open the remote-apply window; + * 2. dispatch the host's bulk replay action; + * 3. yield once so host state reducers finish before side effects; + * 4. run remote archive side effects after dispatch, when configured; + * 5. start post-sync cooldown before closing the remote-apply window; + * 6. close the window and flush deferred local actions. + */ +export const replayOperationBatch = async < + TOperation extends Operation = Operation, + TBulkAction extends SyncActionLike = SyncActionLike, + TReplayAction extends SyncActionLike = SyncActionLike, +>({ + ops, + applyOptions = {}, + dispatcher, + createBulkApplyAction, + remoteApplyWindow, + deferredLocalActions, + archiveSideEffects, + operationToAction, + isArchiveAffectingAction = () => false, + onRemoteArchiveDataApplied, + onArchiveSideEffectError, + onPostSyncCooldownError, + yieldToEventLoop: waitForEventLoop = yieldToEventLoop, +}: OperationReplayCoordinatorOptions): Promise< + ApplyOperationsResult +> => { + if (ops.length === 0) { + return { appliedOps: [] }; + } + + const isLocalHydration = applyOptions.isLocalHydration ?? false; + + if (!isLocalHydration && archiveSideEffects !== undefined && !operationToAction) { + throw new Error( + 'replayOperationBatch requires operationToAction when archiveSideEffects is provided.', + ); + } + + remoteApplyWindow.startApplyingRemoteOps(); + try { + dispatcher.dispatch(createBulkApplyAction(ops)); + await waitForEventLoop(); + + if (!isLocalHydration && archiveSideEffects !== undefined && operationToAction) { + const archiveResult = await processArchiveSideEffects({ + ops, + archiveSideEffects, + operationToAction, + isArchiveAffectingAction, + onArchiveSideEffectError, + yieldToEventLoop: waitForEventLoop, + }); + + if (archiveResult.failedOp) { + return { + appliedOps: archiveResult.appliedOps, + failedOp: archiveResult.failedOp, + }; + } + + if (archiveResult.hadArchiveAffectingOp) { + onRemoteArchiveDataApplied?.(); + } + } + } finally { + if (!isLocalHydration) { + try { + remoteApplyWindow.startPostSyncCooldown(); + } catch (error) { + onPostSyncCooldownError?.(error); + } + } + + remoteApplyWindow.endApplyingRemoteOps(); + await deferredLocalActions.processDeferredActions(); + } + + return { appliedOps: ops }; +}; + +const processArchiveSideEffects = async < + TOperation extends Operation, + TReplayAction extends SyncActionLike, +>({ + ops, + archiveSideEffects, + operationToAction, + isArchiveAffectingAction, + onArchiveSideEffectError, + yieldToEventLoop: waitForEventLoop, +}: { + ops: TOperation[]; + archiveSideEffects: ArchiveSideEffectPort; + operationToAction: (op: TOperation) => TReplayAction; + isArchiveAffectingAction: (action: TReplayAction) => boolean; + onArchiveSideEffectError?: ( + context: OperationReplayArchiveFailureContext, + ) => void; + yieldToEventLoop: () => Promise; +}): Promise> => { + const appliedOps: TOperation[] = []; + let hadArchiveAffectingOp = false; + + for (const op of ops) { + try { + const action = operationToAction(op); + await archiveSideEffects.handleOperation(action); + + if (isArchiveAffectingAction(action)) { + hadArchiveAffectingOp = true; + await waitForEventLoop(); + } + + appliedOps.push(op); + } catch (error) { + onArchiveSideEffectError?.({ + op, + processedCount: appliedOps.length, + error, + }); + + return { + appliedOps, + hadArchiveAffectingOp, + failedOp: { + op, + error: toError(error), + }, + }; + } + } + + await waitForEventLoop(); + + return { + appliedOps, + hadArchiveAffectingOp, + }; +}; + +const toError = (error: unknown): Error => + error instanceof Error ? error : new Error(String(error)); diff --git a/packages/sync-core/tests/replay-coordinator.spec.ts b/packages/sync-core/tests/replay-coordinator.spec.ts new file mode 100644 index 0000000000..faa8cd087e --- /dev/null +++ b/packages/sync-core/tests/replay-coordinator.spec.ts @@ -0,0 +1,392 @@ +import { describe, expect, it, vi } from 'vitest'; +import { replayOperationBatch } from '../src/replay-coordinator'; +import type { + ActionDispatchPort, + ArchiveSideEffectPort, + DeferredLocalActionsPort, + Operation, + RemoteApplyWindowPort, + SyncActionLike, +} from '../src'; + +interface BulkReplayAction extends SyncActionLike { + operations: Operation[]; +} + +interface ReplayAction extends SyncActionLike { + opId: string; + archiveAffecting?: boolean; +} + +const createOperation = (id: string): Operation => ({ + id, + actionType: '[Test] Action', + opType: 'UPD', + entityType: 'TASK', + entityId: id, + payload: {}, + clientId: 'client-1', + vectorClock: { client1: 1 }, + timestamp: 1, + schemaVersion: 1, +}); + +const createRemoteApplyWindow = (callOrder: string[]): RemoteApplyWindowPort => ({ + startApplyingRemoteOps: vi.fn(() => { + callOrder.push('startApplyingRemoteOps'); + }), + endApplyingRemoteOps: vi.fn(() => { + callOrder.push('endApplyingRemoteOps'); + }), + startPostSyncCooldown: vi.fn(() => { + callOrder.push('startPostSyncCooldown'); + }), +}); + +const createDeferredLocalActions = (callOrder: string[]): DeferredLocalActionsPort => ({ + processDeferredActions: vi.fn(async () => { + callOrder.push('processDeferredActions'); + }), +}); + +describe('replayOperationBatch', () => { + it('dispatches a bulk action, yields, runs archive effects, then closes the sync window', async () => { + const callOrder: string[] = []; + const ops = [createOperation('op-1'), createOperation('op-2')]; + const dispatcher: ActionDispatchPort = { + dispatch: vi.fn(() => { + callOrder.push('dispatchBulk'); + }), + }; + const archiveSideEffects: ArchiveSideEffectPort = { + handleOperation: vi.fn(async (action) => { + callOrder.push(`archive:${action.opId}`); + }), + }; + const yieldToEventLoop = vi.fn(async () => { + callOrder.push('yield'); + }); + const onRemoteArchiveDataApplied = vi.fn(() => { + callOrder.push('remoteArchiveDataApplied'); + }); + + const result = await replayOperationBatch({ + ops, + dispatcher, + createBulkApplyAction: (operations) => ({ + type: '[Test] Bulk Apply', + operations, + }), + remoteApplyWindow: createRemoteApplyWindow(callOrder), + deferredLocalActions: createDeferredLocalActions(callOrder), + archiveSideEffects, + operationToAction: (op) => ({ + type: '[Test] Action', + opId: op.id, + archiveAffecting: op.id === 'op-2', + }), + isArchiveAffectingAction: (action) => action.archiveAffecting === true, + onRemoteArchiveDataApplied, + yieldToEventLoop, + }); + + expect(result).toEqual({ appliedOps: ops }); + expect(dispatcher.dispatch).toHaveBeenCalledWith({ + type: '[Test] Bulk Apply', + operations: ops, + }); + expect(archiveSideEffects.handleOperation).toHaveBeenCalledTimes(2); + expect(onRemoteArchiveDataApplied).toHaveBeenCalledTimes(1); + expect(callOrder).toEqual([ + 'startApplyingRemoteOps', + 'dispatchBulk', + 'yield', + 'archive:op-1', + 'archive:op-2', + 'yield', + 'yield', + 'remoteArchiveDataApplied', + 'startPostSyncCooldown', + 'endApplyingRemoteOps', + 'processDeferredActions', + ]); + }); + + it('waits for the post-dispatch yield before archive handling starts', async () => { + const ops = [createOperation('op-1')]; + const archiveSideEffects: ArchiveSideEffectPort = { + handleOperation: vi.fn(async () => undefined), + }; + let releaseFirstYield: (() => void) | undefined; + const yieldToEventLoop = vi.fn( + () => + new Promise((resolve) => { + if (releaseFirstYield === undefined) { + releaseFirstYield = resolve; + return; + } + resolve(); + }), + ); + + const applyPromise = replayOperationBatch({ + ops, + dispatcher: { dispatch: vi.fn() }, + createBulkApplyAction: (operations) => ({ + type: '[Test] Bulk Apply', + operations, + }), + remoteApplyWindow: createRemoteApplyWindow([]), + deferredLocalActions: createDeferredLocalActions([]), + archiveSideEffects, + operationToAction: (op) => ({ type: '[Test] Action', opId: op.id }), + yieldToEventLoop, + }); + + await Promise.resolve(); + + expect(archiveSideEffects.handleOperation).not.toHaveBeenCalled(); + expect(releaseFirstYield).toBeDefined(); + + releaseFirstYield?.(); + await applyPromise; + + expect(archiveSideEffects.handleOperation).toHaveBeenCalledTimes(1); + }); + + it('skips archive handling and post-sync cooldown for local hydration', async () => { + const callOrder: string[] = []; + const op = createOperation('op-1'); + const archiveSideEffects: ArchiveSideEffectPort = { + handleOperation: vi.fn(async () => undefined), + }; + const remoteApplyWindow = createRemoteApplyWindow(callOrder); + + const result = await replayOperationBatch({ + ops: [op], + applyOptions: { isLocalHydration: true }, + dispatcher: { + dispatch: vi.fn(() => { + callOrder.push('dispatchBulk'); + }), + }, + createBulkApplyAction: (operations) => ({ + type: '[Test] Bulk Apply', + operations, + }), + remoteApplyWindow, + deferredLocalActions: createDeferredLocalActions(callOrder), + archiveSideEffects, + operationToAction: (operation) => ({ type: '[Test] Action', opId: operation.id }), + yieldToEventLoop: vi.fn(async () => { + callOrder.push('yield'); + }), + }); + + expect(result).toEqual({ appliedOps: [op] }); + expect(archiveSideEffects.handleOperation).not.toHaveBeenCalled(); + expect(remoteApplyWindow.startPostSyncCooldown).not.toHaveBeenCalled(); + expect(callOrder).toEqual([ + 'startApplyingRemoteOps', + 'dispatchBulk', + 'yield', + 'endApplyingRemoteOps', + 'processDeferredActions', + ]); + }); + + it('reports partial archive failures and still flushes deferred local actions', async () => { + const callOrder: string[] = []; + const op1 = createOperation('op-1'); + const op2 = createOperation('op-2'); + const archiveError = new Error('archive failed'); + const archiveSideEffects: ArchiveSideEffectPort = { + handleOperation: vi.fn(async (action) => { + callOrder.push(`archive:${action.opId}`); + if (action.opId === 'op-2') { + throw archiveError; + } + }), + }; + const onArchiveSideEffectError = vi.fn(); + const onRemoteArchiveDataApplied = vi.fn(); + + const result = await replayOperationBatch({ + ops: [op1, op2], + dispatcher: { + dispatch: vi.fn(() => { + callOrder.push('dispatchBulk'); + }), + }, + createBulkApplyAction: (operations) => ({ + type: '[Test] Bulk Apply', + operations, + }), + remoteApplyWindow: createRemoteApplyWindow(callOrder), + deferredLocalActions: createDeferredLocalActions(callOrder), + archiveSideEffects, + operationToAction: (op) => ({ + type: '[Test] Action', + opId: op.id, + archiveAffecting: true, + }), + isArchiveAffectingAction: (action) => action.archiveAffecting === true, + onArchiveSideEffectError, + onRemoteArchiveDataApplied, + yieldToEventLoop: vi.fn(async () => { + callOrder.push('yield'); + }), + }); + + expect(result).toEqual({ + appliedOps: [op1], + failedOp: { + op: op2, + error: archiveError, + }, + }); + expect(onArchiveSideEffectError).toHaveBeenCalledWith({ + op: op2, + processedCount: 1, + error: archiveError, + }); + expect(onRemoteArchiveDataApplied).not.toHaveBeenCalled(); + expect(callOrder).toEqual([ + 'startApplyingRemoteOps', + 'dispatchBulk', + 'yield', + 'archive:op-1', + 'yield', + 'archive:op-2', + 'startPostSyncCooldown', + 'endApplyingRemoteOps', + 'processDeferredActions', + ]); + }); + + it('fails fast when archive side effects are configured without operation conversion', async () => { + const callOrder: string[] = []; + const archiveSideEffects: ArchiveSideEffectPort = { + handleOperation: vi.fn(async () => undefined), + }; + const dispatcher: ActionDispatchPort = { + dispatch: vi.fn(), + }; + + await expect( + replayOperationBatch({ + ops: [createOperation('op-1')], + dispatcher, + createBulkApplyAction: (operations) => ({ + type: '[Test] Bulk Apply', + operations, + }), + remoteApplyWindow: createRemoteApplyWindow(callOrder), + deferredLocalActions: createDeferredLocalActions(callOrder), + archiveSideEffects, + }), + ).rejects.toThrow( + 'replayOperationBatch requires operationToAction when archiveSideEffects is provided.', + ); + + expect(dispatcher.dispatch).not.toHaveBeenCalled(); + expect(archiveSideEffects.handleOperation).not.toHaveBeenCalled(); + expect(callOrder).toEqual([]); + }); + + it('closes the sync window and flushes deferred actions when dispatch throws', async () => { + const callOrder: string[] = []; + const dispatchError = new Error('dispatch failed'); + + await expect( + replayOperationBatch({ + ops: [createOperation('op-1')], + dispatcher: { + dispatch: vi.fn(() => { + callOrder.push('dispatchBulk'); + throw dispatchError; + }), + }, + createBulkApplyAction: (operations) => ({ + type: '[Test] Bulk Apply', + operations, + }), + remoteApplyWindow: createRemoteApplyWindow(callOrder), + deferredLocalActions: createDeferredLocalActions(callOrder), + yieldToEventLoop: vi.fn(async () => { + callOrder.push('yield'); + }), + }), + ).rejects.toBe(dispatchError); + + expect(callOrder).toEqual([ + 'startApplyingRemoteOps', + 'dispatchBulk', + 'startPostSyncCooldown', + 'endApplyingRemoteOps', + 'processDeferredActions', + ]); + }); + + it('closes the sync window even when post-sync cooldown throws', async () => { + const callOrder: string[] = []; + const cooldownError = new Error('cooldown failed'); + const remoteApplyWindow = createRemoteApplyWindow(callOrder); + vi.mocked(remoteApplyWindow.startPostSyncCooldown).mockImplementation(() => { + callOrder.push('startPostSyncCooldown'); + throw cooldownError; + }); + const onPostSyncCooldownError = vi.fn(); + + await replayOperationBatch({ + ops: [createOperation('op-1')], + dispatcher: { dispatch: vi.fn() }, + createBulkApplyAction: (operations) => ({ + type: '[Test] Bulk Apply', + operations, + }), + remoteApplyWindow, + deferredLocalActions: createDeferredLocalActions(callOrder), + yieldToEventLoop: vi.fn(async () => undefined), + }); + + expect(onPostSyncCooldownError).not.toHaveBeenCalled(); + + await replayOperationBatch({ + ops: [createOperation('op-2')], + dispatcher: { dispatch: vi.fn() }, + createBulkApplyAction: (operations) => ({ + type: '[Test] Bulk Apply', + operations, + }), + remoteApplyWindow, + deferredLocalActions: createDeferredLocalActions(callOrder), + onPostSyncCooldownError, + yieldToEventLoop: vi.fn(async () => undefined), + }); + + expect(onPostSyncCooldownError).toHaveBeenCalledWith(cooldownError); + expect(callOrder.slice(-3)).toEqual([ + 'startPostSyncCooldown', + 'endApplyingRemoteOps', + 'processDeferredActions', + ]); + }); + + it('does not open a sync window for an empty operation batch', async () => { + const callOrder: string[] = []; + const result = await replayOperationBatch({ + ops: [], + dispatcher: { dispatch: vi.fn() }, + createBulkApplyAction: (operations) => ({ + type: '[Test] Bulk Apply', + operations, + }), + remoteApplyWindow: createRemoteApplyWindow(callOrder), + deferredLocalActions: createDeferredLocalActions(callOrder), + }); + + expect(result).toEqual({ appliedOps: [] }); + expect(callOrder).toEqual([]); + }); +}); diff --git a/src/app/op-log/apply/operation-applier.service.ts b/src/app/op-log/apply/operation-applier.service.ts index 00c5b0a126..c39042bf66 100644 --- a/src/app/op-log/apply/operation-applier.service.ts +++ b/src/app/op-log/apply/operation-applier.service.ts @@ -1,5 +1,6 @@ import { inject, Injectable, Injector } from '@angular/core'; import { Store } from '@ngrx/store'; +import { replayOperationBatch } from '@sp/sync-core'; import type { ActionDispatchPort, OperationApplyPort, @@ -85,112 +86,41 @@ export class OperationApplierService implements OperationApplyPort { ); } - // Mark that we're applying remote operations to suppress selector-based effects - this.hydrationState.startApplyingRemoteOps(); - try { - // STEP 1: Bulk dispatch all operations in a single NgRx update - // The bulkOperationsMetaReducer iterates through ops and applies each action. - // Effects don't see individual actions - they only see bulkApplyOperations - // which no effect listens for. - this.store.dispatch(bulkApplyOperations({ operations: ops })); - - // Yield to event loop to ensure store update is processed - await new Promise((resolve) => setTimeout(resolve, 0)); - - // STEP 2: Handle archive operations (only for remote sync, not local hydration) - // Archive data lives in IndexedDB, not NgRx state, so we need to persist it separately. - if (!isLocalHydration) { - const archiveResult = await this._processArchiveOperations(ops); - if (archiveResult.failedOp) { - return archiveResult; - } - + const result = await replayOperationBatch({ + ops, + applyOptions: options, + dispatcher: this.store, + createBulkApplyAction: (operations) => bulkApplyOperations({ operations }), + remoteApplyWindow: this.hydrationState, + deferredLocalActions: { + processDeferredActions: () => + this.injector.get(OperationLogEffects).processDeferredActions(), + }, + archiveSideEffects: this.archiveOperationHandler, + operationToAction: convertOpToAction, + isArchiveAffectingAction, + onRemoteArchiveDataApplied: () => { // Dispatch action to signal archive data was applied (for potential future use) // Note: The refreshWorklogAfterRemoteArchiveOps effect that used to listen to // this action is now disabled to prevent UI freezes during bulk archive sync. - if (archiveResult.hadArchiveAffectingOp) { - this.store.dispatch(remoteArchiveDataApplied()); - } - } - } finally { - // Start cooldown BEFORE ending remote ops flag to eliminate the timing gap - // where isInSyncWindow() returns false and selector-based effects can fire. - // Only needed for remote ops - local hydration doesn't cause the timing gap issue. - // Wrapped in try-catch so endApplyingRemoteOps() always runs even if this fails. - if (!isLocalHydration) { - try { - this.hydrationState.startPostSyncCooldown(); - } catch (e) { - OpLog.err('OperationApplierService: startPostSyncCooldown failed', e); - } - } - - this.hydrationState.endApplyingRemoteOps(); - - // Process any user actions that were buffered during sync replay. - // These get fresh vector clocks that include the newly-applied remote ops. - await this.injector.get(OperationLogEffects).processDeferredActions(); - } - - OpLog.normal('OperationApplierService: Finished applying operations.'); - return { appliedOps: ops }; - } - - /** - * Process archive operations after bulk state dispatch. - * Archive data lives in IndexedDB and needs to be persisted separately. - * - * The archive handler is called for all operations - it internally decides - * which operations need archive storage updates. - */ - private async _processArchiveOperations(ops: Operation[]): Promise<{ - appliedOps: Operation[]; - hadArchiveAffectingOp: boolean; - failedOp?: { op: Operation; error: Error }; - }> { - const appliedOps: Operation[] = []; - let hadArchiveAffectingOp = false; - - for (let i = 0; i < ops.length; i++) { - const op = ops[i]; - try { - const action = convertOpToAction(op); - - // Call handler for all operations - it internally checks if action affects archive - await this.archiveOperationHandler.handleOperation(action); - - // Track if any archive-affecting operations were processed (for UI refresh) - if (isArchiveAffectingAction(action)) { - hadArchiveAffectingOp = true; - // Yield after EACH archive-affecting operation to prevent UI freeze. - // Archive operations involve slow IndexedDB writes that can block the event loop. - await new Promise((resolve) => setTimeout(resolve, 0)); - } - - appliedOps.push(op); - } catch (e) { + this.store.dispatch(remoteArchiveDataApplied()); + }, + onArchiveSideEffectError: ({ op, processedCount, error }) => { OpLog.err( `OperationApplierService: Failed archive handling for operation ${op.id}. ` + - `${appliedOps.length} ops were processed before this failure.`, - e, + `${processedCount} ops were processed before this failure.`, + error, ); + }, + onPostSyncCooldownError: (e) => { + OpLog.err('OperationApplierService: startPostSyncCooldown failed', e); + }, + }); - return { - appliedOps, - hadArchiveAffectingOp, - failedOp: { - op, - error: e instanceof Error ? e : new Error(String(e)), - }, - }; - } + if (!result.failedOp) { + OpLog.normal('OperationApplierService: Finished applying operations.'); } - // Final yield after processing all operations to ensure last operation completes - if (ops.length > 0) { - await new Promise((resolve) => setTimeout(resolve, 0)); - } - - return { appliedOps, hadArchiveAffectingOp }; + return result; } }