From f5b98cf103eef4df9bbcf77ee4e1c90cee89bc34 Mon Sep 17 00:00:00 2001 From: Danne Lundqvist Date: Fri, 3 Jul 2026 11:04:22 +0200 Subject: [PATCH] fix(wires): re-sort documents on subscription update so new versions count as new time Live subscription updates patched changed fields in place without re-sorting, so a wire receiving a new version kept the slot its previous version had instead of moving to the top by its new modified time. - add mergeSubscriptionUpdates helper (patch matched fields, then re-sort by the active sort) with tests - useDocuments threads sort into pollSubscriptions and uses the helper - Stream re-sorts the paginated merge branch too ELELOG-745 --- __tests__/mergeSubscriptionUpdates.test.ts | 86 +++++++++++++++++++ src/hooks/index/useDocuments/index.ts | 29 +++---- .../lib/mergeSubscriptionUpdates.ts | 67 +++++++++++++++ src/views/Wires/components/Stream.tsx | 7 +- 4 files changed, 171 insertions(+), 18 deletions(-) create mode 100644 __tests__/mergeSubscriptionUpdates.test.ts create mode 100644 src/hooks/index/useDocuments/lib/mergeSubscriptionUpdates.ts diff --git a/__tests__/mergeSubscriptionUpdates.test.ts b/__tests__/mergeSubscriptionUpdates.test.ts new file mode 100644 index 000000000..caf7dd648 --- /dev/null +++ b/__tests__/mergeSubscriptionUpdates.test.ts @@ -0,0 +1,86 @@ +import { describe, it, expect } from 'vitest' +import { SortingV1 } from '@ttab/elephant-api/index' +import type { HitV1, SubscriptionItem } from '@ttab/elephant-api/index' +import { mergeSubscriptionUpdates } from '@/hooks/index/useDocuments/lib/mergeSubscriptionUpdates' + +// Wires are sorted newest-first by the document's modified time. +const sort = [SortingV1.create({ field: 'modified', desc: true })] + +const hit = ( + id: string, + modified: string, + extra: Record = {} +): HitV1 => ({ id, fields: { modified: { values: [modified] }, ...extra } }) as unknown as HitV1 + +const matched = ( + id: string, + modified: string, + extra: Record = {} +): SubscriptionItem => + ({ id, match: true, fields: { modified: { values: [modified] }, ...extra } }) as unknown as SubscriptionItem + +const ids = (hits: HitV1[]) => hits.map((h) => h.id) + +describe('mergeSubscriptionUpdates', () => { + it('moves a wire to the top when a new version bumps its modified time', () => { + const data = [ + hit('a', '2026-07-03T10:00:00.000Z'), + hit('b', '2026-07-03T09:00:00.000Z'), + hit('c', '2026-07-03T08:00:00.000Z') + ] + + // 'c' receives a new version and is now the most recently modified. + const result = mergeSubscriptionUpdates(data, [matched('c', '2026-07-03T11:00:00.000Z')], sort) + + expect(ids(result)).toEqual(['c', 'a', 'b']) + }) + + it('re-sorts an updated wire down when its modified time is older than others', () => { + const data = [ + hit('a', '2026-07-03T10:00:00.000Z'), + hit('b', '2026-07-03T09:00:00.000Z') + ] + + // 'a' updated but still (say) with a modified between b and nothing -> stays first here; + // 'b' updated to be newest -> should jump above 'a'. + const result = mergeSubscriptionUpdates(data, [matched('b', '2026-07-03T12:00:00.000Z')], sort) + + expect(ids(result)).toEqual(['b', 'a']) + }) + + it('keeps position for a status-only update where modified is unchanged', () => { + const data = [ + hit('a', '2026-07-03T10:00:00.000Z'), + hit('b', '2026-07-03T09:00:00.000Z', { 'heads.read.version': { values: ['0'] } }) + ] + + // 'b' marked read: only heads.read.version changes, modified is unchanged. + const result = mergeSubscriptionUpdates( + data, + [matched('b', '2026-07-03T09:00:00.000Z', { 'heads.read.version': { values: ['3'] } })], + sort + ) + + expect(ids(result)).toEqual(['a', 'b']) + expect(result[1].fields['heads.read.version']?.values?.[0]).toBe('3') + }) + + it('patches the updated fields onto the matched wire', () => { + const data = [hit('a', '2026-07-03T10:00:00.000Z')] + + const result = mergeSubscriptionUpdates(data, [matched('a', '2026-07-03T12:00:00.000Z')], sort) + + expect(result[0].fields.modified?.values?.[0]).toBe('2026-07-03T12:00:00.000Z') + }) + + it('leaves order untouched when no sort is provided', () => { + const data = [ + hit('a', '2026-07-03T08:00:00.000Z'), + hit('b', '2026-07-03T10:00:00.000Z') + ] + + const result = mergeSubscriptionUpdates(data, [matched('a', '2026-07-03T12:00:00.000Z')]) + + expect(ids(result)).toEqual(['a', 'b']) + }) +}) diff --git a/src/hooks/index/useDocuments/index.ts b/src/hooks/index/useDocuments/index.ts index aac12dd83..11fcbe6f1 100644 --- a/src/hooks/index/useDocuments/index.ts +++ b/src/hooks/index/useDocuments/index.ts @@ -3,6 +3,7 @@ import type { KeyedMutator, SWRResponse } from 'swr' import useSWR from 'swr' import { useRegistry } from '@/hooks/useRegistry' import { fetch } from './lib/fetch' +import { mergeSubscriptionUpdates } from './lib/mergeSubscriptionUpdates' import { useTable } from '@/hooks/useTable' import { useEffect, useMemo, useRef, useState } from 'react' import type { SubscriptionItem, SubscriptionReference } from '@ttab/elephant-api/index' @@ -72,6 +73,7 @@ export const useDocuments = ({ documentType, query, size, pa const mutateRef = useRef | null>(null) const dataRef = useRef(undefined) const optionsRef = useRef(options) + const sortRef = useRef(sort) const { t } = useTranslation('common') // Options that change the fetch shape are part of the cache key so toggling @@ -125,7 +127,8 @@ export const useDocuments = ({ documentType, query, size, pa subscriptionsRef.current = subscriptions mutateRef.current = mutate dataRef.current = data - }, [subscriptions, mutate, data]) + sortRef.current = sort + }, [subscriptions, mutate, data, sort]) // Set table data after fetch useEffect(() => { @@ -150,7 +153,8 @@ export const useDocuments = ({ documentType, query, size, pa subscriptions: subscriptionsRef.current ?? [], mutate: mutateRef.current!, abortController, - options: optionsRef.current + options: optionsRef.current, + sort: sortRef.current }) } catch (error) { if (error instanceof AbortError) { @@ -221,7 +225,8 @@ async function pollSubscriptions({ data = [], mutate, abortController, - options + options, + sort }: { index: Index data?: T[] @@ -230,6 +235,7 @@ async function pollSubscriptions({ mutate: KeyedMutator abortController?: AbortController options?: useDocumentsFetchOptions + sort?: SortingV1[] }): Promise { try { const response: PollSubscriptionResponse = await index.pollSubscription({ @@ -271,19 +277,10 @@ async function pollSubscriptions({ await mutate() return newSubscriptions } - // Build a map of matched items by id for quick lookup - const matchedMap = new Map( - matchedItems.map((item) => [item.id, item]) - ) - // create a new array with updated fields and do a optimistic update - const updatedData = data.map((obj) => - matchedMap.has(obj.id) - ? { - ...obj, - fields: { ...obj.fields, ...matchedMap.get(obj.id)?.fields } - } - : obj - ) + // Optimistically patch the changed fields and re-sort, so a new wire + // version (which bumps `modified`) moves to its correct position instead + // of keeping the slot the previous version had. + const updatedData = mergeSubscriptionUpdates(data, matchedItems, sort) await mutate(updatedData, false) } diff --git a/src/hooks/index/useDocuments/lib/mergeSubscriptionUpdates.ts b/src/hooks/index/useDocuments/lib/mergeSubscriptionUpdates.ts new file mode 100644 index 000000000..cecafc75b --- /dev/null +++ b/src/hooks/index/useDocuments/lib/mergeSubscriptionUpdates.ts @@ -0,0 +1,67 @@ +import type { HitV1, SortingV1, SubscriptionItem } from '@ttab/elephant-api/index' + +/** + * Merge subscription field updates into the current result set. + * + * Matched items patch their fields onto the existing document (optimistic, + * in-place). The result is then re-sorted by the active `sort` so that a + * document whose sort field changed - e.g. a new wire version bumping + * `modified` - moves to its correct position instead of keeping the stale slot + * the previous version occupied. Status-only updates (read/saved/used heads) + * don't touch the sort field, so those documents keep their position. + */ +export function mergeSubscriptionUpdates( + data: T[], + matchedItems: SubscriptionItem[], + sort?: SortingV1[] +): T[] { + const matchedMap = new Map(matchedItems.map((item) => [item.id, item])) + + const merged = data.map((obj) => + matchedMap.has(obj.id) + ? { + ...obj, + fields: { ...obj.fields, ...matchedMap.get(obj.id)?.fields } + } + : obj + ) + + // Re-sort so a document whose sort field changed (e.g. a new wire version + // bumping `modified`) moves to its correct position rather than keeping the + // slot the previous version had. Status-only updates leave the sort field + // unchanged, so those documents keep their position. + return sortHits(merged, sort) +} + +export function sortHits(hits: T[], sort?: SortingV1[]): T[] { + if (!sort?.length) { + return hits + } + + // Array.prototype.sort is stable, so documents that compare equal keep their + // relative order. + return [...hits].sort((a, b) => { + for (const { field, desc } of sort) { + const cmp = compareFieldValues(fieldValue(a, field), fieldValue(b, field)) + if (cmp !== 0) { + return desc ? -cmp : cmp + } + } + return 0 + }) +} + +function fieldValue(hit: HitV1, field: string): string { + return hit.fields?.[field]?.values?.[0] ?? '' +} + +function compareFieldValues(a: string, b: string): number { + // Prefer chronological comparison for date-like fields (e.g. `modified`), + // which is robust to differing timestamp precision; fall back to string order. + const at = Date.parse(a) + const bt = Date.parse(b) + if (!Number.isNaN(at) && !Number.isNaN(bt)) { + return at === bt ? 0 : at < bt ? -1 : 1 + } + return a === b ? 0 : a < b ? -1 : 1 +} diff --git a/src/views/Wires/components/Stream.tsx b/src/views/Wires/components/Stream.tsx index 455e11d16..fa1abb181 100644 --- a/src/views/Wires/components/Stream.tsx +++ b/src/views/Wires/components/Stream.tsx @@ -2,6 +2,7 @@ import { type JSX, useMemo, useCallback, useState, useRef, useEffect, useLayoutE import { fields, type Wire, type WireFields } from '@/shared/schemas/wire' import { getWireState } from '@/lib/getWireState' import { useDocuments } from '@/hooks/index/useDocuments' +import { sortHits } from '@/hooks/index/useDocuments/lib/mergeSubscriptionUpdates' import { constructQuery } from '../lib/constructQuery' import { SortingV1 } from '@ttab/elephant-api/index' import { StreamEntry } from './StreamEntry' @@ -190,7 +191,9 @@ export const Stream = memo(({ byId.set(wire.id, existing ? { ...existing, fields: wire.fields } : wire) }) - next = Array.from(byId.values()) + // Re-sort so an updated wire (newer `modified`) moves to its correct + // position instead of keeping the slot the previous version had. + next = sortHits(Array.from(byId.values()), sort) } const wireStatusFilter = wireStream.filters.find((f) => f.type === 'wireStatus') @@ -204,7 +207,7 @@ export const Stream = memo(({ if (!isLoading) { loadingRef.current = false } - }, [data, isLoading, page, wireStream.filters]) + }, [data, isLoading, page, wireStream.filters, sort]) // Reset to page 1 when debounced filters change. // Don't clear allData here — the data effect replaces it on page 1 once the fetch completes,