Mirrors/super-productivity
1
0
Fork 0
mirror of https://github.com/johannesjo/super-productivity.git synced 2026-07-18 00:46:45 +00:00
Code Issues Projects Releases Wiki Activity

refactor(sync-core): prepare core boundary for providers

Browse source
This commit is contained in:
Johannes Millan 2026-05-12 16:04:11 +02:00
parent a57323277b
commit 298928e6d2
7 changed files with 726 additions and 125 deletions
Download patch file Download diff file Expand all files Collapse all files

104
docs/long-term-plans/sync-core-extraction-plan.md
View file

@ -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

2
packages/sync-core/src/entity-key.util.ts
View file

@ -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"
*/

7
packages/sync-core/src/index.ts
View file

@ -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 {

4
packages/sync-core/src/lww-update-action-types.ts
View file

@ -22,8 +22,8 @@ export interface LwwUpdateActionTypeHelpers<TEntityType extends string = string>
* 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 = <TEntityType extends string>(
entityTypes: readonly TEntityType[],

216
packages/sync-core/src/replay-coordinator.ts Normal file
View file

@ -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<string> = Operation,
> {
op: TOperation;
processedCount: number;
error: unknown;
}
export interface OperationReplayCoordinatorOptions<
TOperation extends Operation<string> = Operation,
TBulkAction extends SyncActionLike = SyncActionLike,
TReplayAction extends SyncActionLike = SyncActionLike,
> {
ops: TOperation[];
applyOptions?: ApplyOperationsOptions;
dispatcher: ActionDispatchPort<TBulkAction>;
createBulkApplyAction: (ops: TOperation[]) => TBulkAction;
remoteApplyWindow: RemoteApplyWindowPort;
deferredLocalActions: DeferredLocalActionsPort;
archiveSideEffects?: ArchiveSideEffectPort<TReplayAction>;
operationToAction?: (op: TOperation) => TReplayAction;
isArchiveAffectingAction?: (action: TReplayAction) => boolean;
onRemoteArchiveDataApplied?: () => void;
onArchiveSideEffectError?: (
context: OperationReplayArchiveFailureContext<TOperation>,
) => void;
onPostSyncCooldownError?: (error: unknown) => void;
yieldToEventLoop?: () => Promise<void>;
}
interface ArchiveSideEffectResult<TOperation extends Operation<string> = Operation> {
appliedOps: TOperation[];
hadArchiveAffectingOp: boolean;
failedOp?: {
op: TOperation;
error: Error;
};
}
const globals = (): ReplayGlobals => globalThis as unknown as ReplayGlobals;
export const yieldToEventLoop = (): Promise<void> =>
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<string> = 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<TOperation, TBulkAction, TReplayAction>): Promise<
ApplyOperationsResult<TOperation>
> => {
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<string>,
TReplayAction extends SyncActionLike,
>({
ops,
archiveSideEffects,
operationToAction,
isArchiveAffectingAction,
onArchiveSideEffectError,
yieldToEventLoop: waitForEventLoop,
}: {
ops: TOperation[];
archiveSideEffects: ArchiveSideEffectPort<TReplayAction>;
operationToAction: (op: TOperation) => TReplayAction;
isArchiveAffectingAction: (action: TReplayAction) => boolean;
onArchiveSideEffectError?: (
context: OperationReplayArchiveFailureContext<TOperation>,
) => void;
yieldToEventLoop: () => Promise<void>;
}): Promise<ArchiveSideEffectResult<TOperation>> => {
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));

392
packages/sync-core/tests/replay-coordinator.spec.ts Normal file
View file

@ -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<string>[];
}
interface ReplayAction extends SyncActionLike {
opId: string;
archiveAffecting?: boolean;
}
const createOperation = (id: string): Operation<string> => ({
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<BulkReplayAction> = {
dispatch: vi.fn(() => {
callOrder.push('dispatchBulk');
}),
};
const archiveSideEffects: ArchiveSideEffectPort<ReplayAction> = {
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<ReplayAction> = {
handleOperation: vi.fn(async () => undefined),
};
let releaseFirstYield: (() => void) | undefined;
const yieldToEventLoop = vi.fn(
() =>
new Promise<void>((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<ReplayAction> = {
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<ReplayAction> = {
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<ReplayAction> = {
handleOperation: vi.fn(async () => undefined),
};
const dispatcher: ActionDispatchPort<BulkReplayAction> = {
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([]);
});
});

126
src/app/op-log/apply/operation-applier.service.ts
View file

@ -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<Operation> {
);
}
// 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;
}
}