Merge branch 'main' into chore/build-toolchain-20260812
This commit is contained in:
@@ -52,6 +52,11 @@ export const buildReadModelDeltaCacheKey = (
|
|||||||
const canPatch = (value: unknown): value is Record<string, unknown> | unknown[] =>
|
const canPatch = (value: unknown): value is Record<string, unknown> | unknown[] =>
|
||||||
value !== null && typeof value === 'object';
|
value !== null && typeof value === 'object';
|
||||||
|
|
||||||
|
const snapshotByteLength = (revision: string, serialized: string): number =>
|
||||||
|
Buffer.byteLength(`{"kind":"snapshot","revision":${JSON.stringify(revision)},"data":`) +
|
||||||
|
Buffer.byteLength(serialized) +
|
||||||
|
1;
|
||||||
|
|
||||||
const storeSnapshot = async (store: ReadModelDeltaCacheStore, key: string, serialized: string): Promise<void> => {
|
const storeSnapshot = async (store: ReadModelDeltaCacheStore, key: string, serialized: string): Promise<void> => {
|
||||||
try {
|
try {
|
||||||
await store.set(key, serialized, { EX: CACHE_TTL_SECONDS });
|
await store.set(key, serialized, { EX: CACHE_TTL_SECONDS });
|
||||||
@@ -91,7 +96,7 @@ export const createReadModelDelta = async <T>(request: ReadModelDeltaRequest<T>)
|
|||||||
};
|
};
|
||||||
const snapshot = { kind: 'snapshot' as const, revision, data: canonicalValue };
|
const snapshot = { kind: 'snapshot' as const, revision, data: canonicalValue };
|
||||||
await storeSnapshot(request.store, currentKey, serialized);
|
await storeSnapshot(request.store, currentKey, serialized);
|
||||||
return Buffer.byteLength(JSON.stringify(patch)) < Buffer.byteLength(JSON.stringify(snapshot))
|
return Buffer.byteLength(JSON.stringify(patch)) < snapshotByteLength(revision, serialized)
|
||||||
? patch
|
? patch
|
||||||
: snapshot;
|
: snapshot;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -89,6 +89,31 @@ describe('createReadModelDelta', () => {
|
|||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('keeps the snapshot fallback when a positional patch is larger', async () => {
|
||||||
|
const store = new MemoryStore();
|
||||||
|
const initialValue = { values: Array.from({ length: 24 }, () => 'old') };
|
||||||
|
const initial = await createReadModelDelta({
|
||||||
|
store,
|
||||||
|
profile: 'hwe:default',
|
||||||
|
viewerId: 'user-1',
|
||||||
|
slice: 'context',
|
||||||
|
value: initialValue,
|
||||||
|
forceSnapshot: true,
|
||||||
|
});
|
||||||
|
const nextValue = { values: Array.from({ length: 24 }, () => 'new') };
|
||||||
|
|
||||||
|
const changed = await createReadModelDelta({
|
||||||
|
store,
|
||||||
|
profile: 'hwe:default',
|
||||||
|
viewerId: 'user-1',
|
||||||
|
slice: 'context',
|
||||||
|
value: nextValue,
|
||||||
|
knownRevision: initial.revision,
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(changed).toMatchObject({ kind: 'snapshot', data: nextValue });
|
||||||
|
});
|
||||||
|
|
||||||
it('falls back to a snapshot when Redis is unavailable', async () => {
|
it('falls back to a snapshot when Redis is unavailable', async () => {
|
||||||
const store: ReadModelDeltaCacheStore = {
|
const store: ReadModelDeltaCacheStore = {
|
||||||
get: async () => {
|
get: async () => {
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
import { applyPatch, createPatch, type Operation } from 'rfc6902';
|
import { applyPatch, type Operation } from 'rfc6902';
|
||||||
|
|
||||||
export interface JsonPatchOperation {
|
export interface JsonPatchOperation {
|
||||||
op: 'add' | 'remove' | 'replace' | 'move' | 'copy' | 'test';
|
op: 'add' | 'remove' | 'replace' | 'move' | 'copy' | 'test';
|
||||||
@@ -63,7 +63,76 @@ export const cloneReadModelJson = <T>(value: T): T => {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
export const createJsonPatch = (current: unknown, next: unknown): JsonPatchOperation[] => createPatch(current, next);
|
const escapeJsonPointerToken = (token: string): string => token.replaceAll('~', '~0').replaceAll('/', '~1');
|
||||||
|
|
||||||
|
const appendJsonPointer = (path: string, token: string): string => `${path}/${escapeJsonPointerToken(token)}`;
|
||||||
|
|
||||||
|
const isJsonObject = (value: unknown): value is Record<string, unknown> =>
|
||||||
|
value !== null && typeof value === 'object' && !Array.isArray(value);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Build a valid patch in one positional pass over arrays.
|
||||||
|
*
|
||||||
|
* The upstream rfc6902 generator minimizes array edit count with a Levenshtein
|
||||||
|
* matrix. Dashboard read models use stable, positional arrays and only need an
|
||||||
|
* exact reconstruction; the cache layer already rejects patches larger than a
|
||||||
|
* snapshot. Positional recursion is therefore linear: shifted positions are
|
||||||
|
* reconciled once, then excess tail items are removed or appended.
|
||||||
|
*/
|
||||||
|
const appendLinearJsonDiff = (
|
||||||
|
operations: JsonPatchOperation[],
|
||||||
|
current: unknown,
|
||||||
|
next: unknown,
|
||||||
|
path: string
|
||||||
|
): void => {
|
||||||
|
if (Object.is(current, next)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (Array.isArray(current) && Array.isArray(next)) {
|
||||||
|
const sharedLength = Math.min(current.length, next.length);
|
||||||
|
for (let index = 0; index < sharedLength; index += 1) {
|
||||||
|
appendLinearJsonDiff(operations, current[index], next[index], appendJsonPointer(path, String(index)));
|
||||||
|
}
|
||||||
|
for (let index = current.length - 1; index >= next.length; index -= 1) {
|
||||||
|
operations.push({ op: 'remove', path: appendJsonPointer(path, String(index)) });
|
||||||
|
}
|
||||||
|
for (let index = current.length; index < next.length; index += 1) {
|
||||||
|
operations.push({ op: 'add', path: appendJsonPointer(path, '-'), value: next[index] });
|
||||||
|
}
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (isJsonObject(current) && isJsonObject(next)) {
|
||||||
|
const currentKeys = Object.keys(current).filter((key) => current[key] !== undefined);
|
||||||
|
const nextKeys = Object.keys(next).filter((key) => next[key] !== undefined);
|
||||||
|
const nextKeySet = new Set(nextKeys);
|
||||||
|
const currentKeySet = new Set(currentKeys);
|
||||||
|
|
||||||
|
for (const key of currentKeys) {
|
||||||
|
if (!nextKeySet.has(key)) {
|
||||||
|
operations.push({ op: 'remove', path: appendJsonPointer(path, key) });
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for (const key of nextKeys) {
|
||||||
|
const itemPath = appendJsonPointer(path, key);
|
||||||
|
if (!currentKeySet.has(key)) {
|
||||||
|
operations.push({ op: 'add', path: itemPath, value: next[key] });
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
appendLinearJsonDiff(operations, current[key], next[key], itemPath);
|
||||||
|
}
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
operations.push({ op: 'replace', path, value: next });
|
||||||
|
};
|
||||||
|
|
||||||
|
export const createJsonPatch = (current: unknown, next: unknown): JsonPatchOperation[] => {
|
||||||
|
const operations: JsonPatchOperation[] = [];
|
||||||
|
appendLinearJsonDiff(operations, current, next, '');
|
||||||
|
return operations;
|
||||||
|
};
|
||||||
|
|
||||||
export const applyReadModelDelta = <T>(
|
export const applyReadModelDelta = <T>(
|
||||||
current: T | undefined,
|
current: T | undefined,
|
||||||
|
|||||||
@@ -1,6 +1,88 @@
|
|||||||
import { describe, expect, it } from 'vitest';
|
import { describe, expect, it } from 'vitest';
|
||||||
|
|
||||||
import { applyReadModelDelta, ReadModelDeltaApplyError, ReadModelDeltaMismatchError } from '../src/realtime/delta.js';
|
import {
|
||||||
|
applyReadModelDelta,
|
||||||
|
createJsonPatch,
|
||||||
|
ReadModelDeltaApplyError,
|
||||||
|
ReadModelDeltaMismatchError,
|
||||||
|
} from '../src/realtime/delta.js';
|
||||||
|
|
||||||
|
describe('createJsonPatch', () => {
|
||||||
|
it('diffs stable arrays positionally and preserves narrow field updates', () => {
|
||||||
|
const current = {
|
||||||
|
general: Array.from({ length: 48 }, (_, index) => ({
|
||||||
|
key: `command-${index}`,
|
||||||
|
possible: true,
|
||||||
|
status: 'available',
|
||||||
|
})),
|
||||||
|
};
|
||||||
|
const next = structuredClone(current);
|
||||||
|
const changed = next.general[0];
|
||||||
|
if (!changed) throw new Error('command fixture is empty');
|
||||||
|
changed.possible = false;
|
||||||
|
changed.status = 'blocked';
|
||||||
|
|
||||||
|
const operations = createJsonPatch(current, next);
|
||||||
|
|
||||||
|
expect(operations).toEqual([
|
||||||
|
{ op: 'replace', path: '/general/0/possible', value: false },
|
||||||
|
{ op: 'replace', path: '/general/0/status', value: 'blocked' },
|
||||||
|
]);
|
||||||
|
expect(
|
||||||
|
applyReadModelDelta(current, 'revision-1', {
|
||||||
|
kind: 'patch',
|
||||||
|
baseRevision: 'revision-1',
|
||||||
|
revision: 'revision-2',
|
||||||
|
operations,
|
||||||
|
}).data
|
||||||
|
).toEqual(next);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('resizes arrays linearly and escapes object keys as JSON Pointer tokens', () => {
|
||||||
|
const current = { 'a/b~c': { values: [1, 2] }, removed: true };
|
||||||
|
const next = { 'a/b~c': { values: [1, 2, 3] }, added: true };
|
||||||
|
|
||||||
|
const operations = createJsonPatch(current, next);
|
||||||
|
|
||||||
|
expect(operations).toEqual([
|
||||||
|
{ op: 'remove', path: '/removed' },
|
||||||
|
{ op: 'add', path: '/a~1b~0c/values/-', value: 3 },
|
||||||
|
{ op: 'add', path: '/added', value: true },
|
||||||
|
]);
|
||||||
|
expect(
|
||||||
|
applyReadModelDelta(current, 'revision-1', {
|
||||||
|
kind: 'patch',
|
||||||
|
baseRevision: 'revision-1',
|
||||||
|
revision: 'revision-2',
|
||||||
|
operations,
|
||||||
|
}).data
|
||||||
|
).toEqual(next);
|
||||||
|
});
|
||||||
|
|
||||||
|
it('reconstructs root arrays after positional insertions and removals', () => {
|
||||||
|
const inserted = ['new', 'first', 'second'];
|
||||||
|
const insertionOperations = createJsonPatch(['first', 'second'], inserted);
|
||||||
|
expect(
|
||||||
|
applyReadModelDelta(['first', 'second'], 'revision-1', {
|
||||||
|
kind: 'patch',
|
||||||
|
baseRevision: 'revision-1',
|
||||||
|
revision: 'revision-2',
|
||||||
|
operations: insertionOperations,
|
||||||
|
}).data
|
||||||
|
).toEqual(inserted);
|
||||||
|
|
||||||
|
const removed = ['second'];
|
||||||
|
const removalOperations = createJsonPatch(['first', 'second', 'third'], removed);
|
||||||
|
expect(
|
||||||
|
applyReadModelDelta(['first', 'second', 'third'], 'revision-2', {
|
||||||
|
kind: 'patch',
|
||||||
|
baseRevision: 'revision-2',
|
||||||
|
revision: 'revision-3',
|
||||||
|
operations: removalOperations,
|
||||||
|
}).data
|
||||||
|
).toEqual(removed);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe('applyReadModelDelta', () => {
|
describe('applyReadModelDelta', () => {
|
||||||
it('applies a JSON Patch without mutating the previous snapshot', () => {
|
it('applies a JSON Patch without mutating the previous snapshot', () => {
|
||||||
|
|||||||
Reference in New Issue
Block a user