diff --git a/src/ChannelManager.ts b/src/ChannelManager.ts index 0a9cef9989..7e862ebacd 100644 --- a/src/ChannelManager.ts +++ b/src/ChannelManager.ts @@ -212,6 +212,12 @@ const channelTruncatedHandler: LabeledEventHandler = { id: 'ChannelManager:default-handler:channel.truncated', }; +// a hard delete of the last message might move the channel's `last_message_at` back +const messageDeletedHandler: LabeledEventHandler = { + handle: (context) => (context.event.hard_delete ? reinsertItem(context) : undefined), + id: 'ChannelManager:default-handler:message.deleted', +}; + const channelVisibleHandler: LabeledEventHandler = { handle: updateLists, id: 'ChannelManager:default-handler:channel.visible', @@ -417,6 +423,7 @@ export class ChannelManager extends WithSubscriptions { 'channel.hidden': [channelHiddenHandler], 'channel.visible': [channelVisibleHandler], 'member.updated': [memberUpdatedHandler], + 'message.deleted': [messageDeletedHandler], 'message.new': [messageNewHandler], 'message.read': [readStateChangedHandler], 'message.read_locally': [readStateChangedHandler], diff --git a/src/EventHandlerPipeline.ts b/src/EventHandlerPipeline.ts index ba10fa6eb5..80e7cf4e9b 100644 --- a/src/EventHandlerPipeline.ts +++ b/src/EventHandlerPipeline.ts @@ -207,8 +207,9 @@ export class EventHandlerPipeline = {}> { ctx, }); if (result?.action === 'stop') return; - } catch { + } catch (error) { console.error(`[pipeline:${this.id}] handler failed`, { + error, handlerId: handler.id ?? 'unknown', handlerIndex: i, }); diff --git a/src/channel.ts b/src/channel.ts index 1652f3b57c..88681fb632 100644 --- a/src/channel.ts +++ b/src/channel.ts @@ -2414,6 +2414,15 @@ export class Channel extends ChannelApi { .debug(`Received an event of type "${event.type}".`, { event }); const channelState = channel.state; + + // The server unhides a channel on any new message without sending `channel.visible`. + if ( + (event.type === 'message.new' || event.type === 'notification.message_new') && + channel.data?.hidden + ) { + channel.data = { ...channel.data, hidden: false }; + } + switch (event.type) { case 'typing.start': if (event.user?.id) { @@ -2710,6 +2719,7 @@ export class Channel extends ChannelApi { } break; case 'channel.truncated': { + if (event.channel) channel.data = { ...channel.data, ...event.channel }; const truncatedAt = event.channel?.truncated_at; if (truncatedAt != null) { this._setOwnUnreadCount(this.countUnread(truncatedAt)); diff --git a/src/pagination/cursorDerivation/aroundPaginationFlags.ts b/src/pagination/cursorDerivation/aroundPaginationFlags.ts new file mode 100644 index 0000000000..1b264dd668 --- /dev/null +++ b/src/pagination/cursorDerivation/aroundPaginationFlags.ts @@ -0,0 +1,30 @@ +import type { Interval, PaginationFlags } from '../paginators'; + +/** + * Flags for a page centred on a point (`id_around` / `created_at_around`). The server splits the + * items around the point evenly and, when one side runs out, fills the rest from the other side. So + * on a full page a side that got less than the smaller half ran out. A page shorter than the limit + * holds everything on both sides. + */ +export const deriveAroundPaginationFlags = ({ + interval, + newerCount, + olderCount, + pageLength, + requestedPageSize, +}: { + interval: Interval; + newerCount: number; + olderCount: number; + pageLength: number; + requestedPageSize: number; +}): PaginationFlags => { + const isPageFull = pageLength >= requestedPageSize; + // the smaller half, so an odd slot the server gives either side does not matter + const minShare = Math.floor((olderCount + newerCount) / 2); + // an interval already known to reach the dataset head or tail stays there + return { + hasMoreHead: !interval.isHead && isPageFull && newerCount >= minShare, + hasMoreTail: !interval.isTail && isPageFull && olderCount >= minShare, + }; +}; diff --git a/src/pagination/cursorDerivation/createdAtAroundPaginationFlags.ts b/src/pagination/cursorDerivation/createdAtAroundPaginationFlags.ts index ca6e7568b4..f78132c5eb 100644 --- a/src/pagination/cursorDerivation/createdAtAroundPaginationFlags.ts +++ b/src/pagination/cursorDerivation/createdAtAroundPaginationFlags.ts @@ -1,77 +1,29 @@ -import { binarySearch } from '../sortCompiler'; +import { deriveAroundPaginationFlags } from './aroundPaginationFlags'; import { dateToNs } from '../../utils/time'; import type { TimestampNS } from '../../types'; -import type { BasePaginator, CursorDeriveContext, PaginationFlags } from '../paginators'; -import { ComparisonResult } from '../types.normalization'; +import type { CursorDeriveContext, PaginationFlags } from '../paginators'; export const deriveCreatedAtAroundPaginationFlags = < T extends { id: string; created_at: TimestampNS }, Q extends { created_at_around?: Date | string }, - P extends BasePaginator, >({ hasMoreHead, hasMoreTail, interval, page, - paginator, queryShape, requestedPageSize, -}: CursorDeriveContext & { paginator: P }): PaginationFlags => { - let flags: PaginationFlags = { hasMoreHead, hasMoreTail }; - if (!queryShape?.created_at_around) return flags; +}: CursorDeriveContext): PaginationFlags => { + if (!queryShape?.created_at_around) return { hasMoreHead, hasMoreTail }; // `created_at_around` is a REQUEST field, so it is still a `Date` (or an ISO string). Items carry // the wire unit, so bring the bound into that unit before comparing the two. const createdAtAround = dateToNs(new Date(queryShape.created_at_around)); - const [firstPageItem, lastPageItem] = [page[0], page.slice(-1)[0]]; - - // expect ASC order (from oldest to newest) - const isAboveHeadBound = - paginator.sortComparator({ created_at: createdAtAround } as T, lastPageItem) === - ComparisonResult.A_PRECEDES_B; - const isBelowTailBound = - paginator.sortComparator(firstPageItem, { created_at: createdAtAround } as T) === - ComparisonResult.A_PRECEDES_B; - - const requestedPageSizeNotMet = - requestedPageSize > interval.itemIds.length && requestedPageSize > page.length; - const noMoreMessages = - (requestedPageSize > interval.itemIds.length || - interval.itemIds.length >= page.length) && - requestedPageSize > page.length; - - if (isAboveHeadBound) { - flags.hasMoreHead = false; - if (requestedPageSizeNotMet) { - flags.hasMoreTail = false; - } - } else if (isBelowTailBound) { - flags.hasMoreTail = false; - if (requestedPageSizeNotMet) { - flags.hasMoreHead = false; - } - } else if (noMoreMessages) { - flags = { hasMoreHead: false, hasMoreTail: false }; - } else { - const [firstPageMsgIsFirstInSet, lastPageMsgIsLastInSet] = [ - firstPageItem?.id && firstPageItem.id === interval.itemIds[0], - lastPageItem?.id && lastPageItem.id === interval.itemIds.slice(-1)[0], - ]; - - const midPointByCount = Math.floor(page.length / 2); - const { insertionIndex } = binarySearch({ - needle: { created_at: createdAtAround } as T, - length: page.length, - getItemAt: (index) => page[index], - compare: (a, b) => a.created_at - b.created_at, - itemIdentityEquals: (a, b) => a.created_at === b.created_at, - plateauScan: false, - }); - - if (insertionIndex !== -1) { - if (firstPageMsgIsFirstInSet) flags.hasMoreTail = midPointByCount <= insertionIndex; - if (lastPageMsgIsLastInSet) flags.hasMoreHead = midPointByCount >= insertionIndex; - } - } - - return flags; + const olderCount = page.filter((item) => item.created_at < createdAtAround).length; + return deriveAroundPaginationFlags({ + interval, + newerCount: page.length - olderCount, + olderCount, + pageLength: page.length, + requestedPageSize, + }); }; diff --git a/src/pagination/cursorDerivation/idAroundPaginationFlags.ts b/src/pagination/cursorDerivation/idAroundPaginationFlags.ts index 55c8a08747..7b02632a8c 100644 --- a/src/pagination/cursorDerivation/idAroundPaginationFlags.ts +++ b/src/pagination/cursorDerivation/idAroundPaginationFlags.ts @@ -1,3 +1,4 @@ +import { deriveAroundPaginationFlags } from './aroundPaginationFlags'; import type { CursorDeriveContext, PaginationFlags } from '../paginators'; export const deriveIdAroundPaginationFlags = < @@ -11,43 +12,14 @@ export const deriveIdAroundPaginationFlags = < queryShape, requestedPageSize, }: CursorDeriveContext): PaginationFlags => { - let flags: PaginationFlags = { hasMoreHead, hasMoreTail }; - if (!queryShape?.id_around) return flags; - const { id_around } = queryShape; - - const [firstPageMsg, lastPageMsg] = [page[0], page.slice(-1)[0]]; - const [firstPageMsgIsFirstInSet, lastPageMsgIsLastInSet] = [ - firstPageMsg?.id === interval.itemIds[0], - lastPageMsg?.id === interval.itemIds.slice(-1)[0], - ]; - - const midPoint = Math.floor(page.length / 2); - const noMoreMessages = - (requestedPageSize > interval.itemIds.length || - interval.itemIds.length >= page.length) && - requestedPageSize > page.length; - - if (noMoreMessages) { - flags = { hasMoreHead: false, hasMoreTail: false }; - } else if (!page[midPoint]) { - return flags; - } else if (page[midPoint].id === id_around) { - flags = { hasMoreHead: true, hasMoreTail: true }; - } else { - const halves = [page.slice(0, midPoint), page.slice(midPoint)]; - if (firstPageMsgIsFirstInSet) { - const targetMsg = halves[0].find((message) => message.id === id_around); - if (targetMsg) { - flags.hasMoreTail = false; - } - } - if (lastPageMsgIsLastInSet) { - const targetMsg = halves[1].find((message) => message.id === id_around); - if (targetMsg) { - flags.hasMoreHead = false; - } - } - } - - return flags; + const targetIndex = page.findIndex((item) => item.id === queryShape?.id_around); + // Without the target in the page, the page says nothing about where it sits. + if (!queryShape?.id_around || targetIndex === -1) return { hasMoreHead, hasMoreTail }; + return deriveAroundPaginationFlags({ + interval, + newerCount: page.length - 1 - targetIndex, + olderCount: targetIndex, + pageLength: page.length, + requestedPageSize, + }); }; diff --git a/src/pagination/paginators/BasePaginator.ts b/src/pagination/paginators/BasePaginator.ts index 33df605ffe..78e54e939e 100644 --- a/src/pagination/paginators/BasePaginator.ts +++ b/src/pagination/paginators/BasePaginator.ts @@ -1968,9 +1968,12 @@ export abstract class BasePaginator { protected locateByItemInInterval({ item, interval, + plateauScan = true, }: { item: T; interval: Interval | LogicalInterval; + /** Walk the interval for an item the binary search missed. Off when it cannot be there. */ + plateauScan?: boolean; }): ItemLocation | null { const ids = interval.itemIds; @@ -1982,7 +1985,7 @@ export abstract class BasePaginator { this.getItemId(item1) === this.getItemId(item2), // items in intervals are not sorted by effectiveComparator compare: this.itemOrderComparator.bind(this), - plateauScan: true, + plateauScan, }); } @@ -2008,22 +2011,21 @@ export abstract class BasePaginator { } protected locateByItemInIntervals(item: T): ItemCoordinates['interval'] | undefined { - // Two different questions, asked in this order: - // 1. which interval's sort bounds does this item fall into (`locateIntervalForItem`)? The only - // one that can answer for an item this paginator has never stored. - // 2. which interval actually lists this id (`findIntervalHoldingItem`)? - // - // They disagree when an item is updated in place: the interval holds the very object being - // ingested, so it already carries the item's NEW sort value, and an item that moved outside its - // own window's bounds is no longer found by (1) — even though (2) still lists it. The fallback - // is what lets `ingestItem` take the old entry out before re-inserting; without it the id stays - // behind in that interval and ends up stored in two places at once. - const interval = - this.locateIntervalForItem(item) ?? this.findIntervalHoldingItem(item); - if (!interval) return undefined; - const itemLocation = this.locateByItemInInterval({ item, interval }); - if (!itemLocation) return undefined; - return { interval, ...itemLocation }; + const locateIn = (interval: AnyInterval | undefined, plateauScan?: boolean) => { + const location = + interval && this.locateByItemInInterval({ item, interval, plateauScan }); + return interval && location ? { interval, ...location } : undefined; + }; + const boundsInterval = this.locateIntervalForItem(item); + // Not held by this list, so it is in no interval. It goes where it should be located. + if (!this._itemIndex.has(this.getItemId(item))) + return locateIn(boundsInterval, false); + // Item held by some interval and has not moved. It goes where its sort value puts it, found without a scan. + const inBounds = locateIn(boundsInterval); + if (inBounds && inBounds.currentIndex > -1) return inBounds; + // The item is held by some interval but has moved. We have to scan all intervals to find it. + const holdingInterval = this.findIntervalHoldingItem(item); + return holdingInterval ? locateIn(holdingInterval) : inBounds; } /** @@ -3064,6 +3066,11 @@ export abstract class BasePaginator { return false; } + /** The page size a query's results are measured against when deriving the head/tail flags. */ + protected getRequestedPageSize(_queryShape: Q | undefined): number { + return this.pageSize; + } + protected getStateAfterQuery( stateUpdate: Partial>, @@ -3223,7 +3230,7 @@ export abstract class BasePaginator { isFirstPage, keepPreviousItems, queryShape, - requestedPageSize: this.pageSize, + requestedPageSize: this.getRequestedPageSize(queryShape), results, updateState, }); @@ -3332,6 +3339,21 @@ export abstract class BasePaginator { stateUpdate.cursor = cursor; stateUpdate.hasMoreTail = hasMoreTail; stateUpdate.hasMoreHead = hasMoreHead; + } else if (this.config.deriveCursor && direction && !isFirstPage) { + // Nothing was ingested, so there is no interval to derive from and the query only resolved + // the direction it paginated in. The other side keeps its cursor. + stateUpdate.cursor = + direction === 'headward' + ? { + headward: resolvedHeadward || null, + tailward: this.cursor?.tailward ?? null, + } + : { + headward: this.cursor?.headward ?? null, + tailward: resolvedTailward || null, + }; + if (direction === 'headward') stateUpdate.hasMoreHead = !!resolvedHeadward; + else stateUpdate.hasMoreTail = !!resolvedTailward; } else { stateUpdate.cursor = { tailward: resolvedTailward || null, @@ -3408,7 +3430,10 @@ export abstract class BasePaginator { } const state = this.getStateAfterQuery(stateUpdate, isFirstPage); + // `updateState` decides whether the results are committed, not whether the query is over: a + // caller that commits its own state must not be left with the `isLoading` executeQuery published. if (updateState) this.state.next(state); + else if (this.isLoading) this.state.partialNext({ isLoading: false }); // Only the first page is cached and a later page cannot change it. if (isFirstPage) this.populateOfflineDbAfterQuery({ items: state.items, queryShape }); diff --git a/src/pagination/paginators/ChannelPaginator.ts b/src/pagination/paginators/ChannelPaginator.ts index a89f73a914..b60df76a3e 100644 --- a/src/pagination/paginators/ChannelPaginator.ts +++ b/src/pagination/paginators/ChannelPaginator.ts @@ -32,6 +32,13 @@ import type { FieldToDataResolver, PathResolver } from '../types.normalization'; import { resolveDotPathValue } from '../utility.normalization'; import { isEqual } from '../../utils/mergeWith/mergeWithCore'; +const hasSameItems = (current: Channel[] | undefined, next: Channel[] | undefined) => { + if (current === next) return false; + if (!current || !next || current.length !== next.length) return false; + for (let i = 0; i < next.length; i++) if (current[i] !== next[i]) return false; + return true; +}; + const DEFAULT_BACKEND_SORT: SortParamRequest[] = [ { direction: -1, field: 'last_message_at' }, { direction: -1, field: 'updated_at' }, @@ -165,16 +172,9 @@ const hiddenFilterResolver: FieldToDataResolver = { const lastUpdatedFilterResolver: FieldToDataResolver = { matchesField: (field) => field === 'last_updated', - resolve: (channel) => { - // combination of last_message_at and updated_at — both already wire timestamps, so they are - // directly comparable. Deriving one of them through `new Date(...).getTime()` would put the two - // in different units and make `updated_at` win every comparison. - const lastMessageAt = channel.messagePaginator.lastMessageAt ?? null; - const updatedAt = channel.data?.updated_at; - return lastMessageAt !== null && updatedAt !== undefined - ? Math.max(lastMessageAt, updatedAt) - : (lastMessageAt ?? updatedAt); - }, + // the server uses the creation time for a channel without messages + resolve: (channel) => + channel.messagePaginator.lastMessageAt ?? channel.data?.created_at, }; const membersFilterResolver: FieldToDataResolver = { @@ -313,6 +313,8 @@ export class ChannelPaginator extends BasePaginator private _itemsLoadedFromOfflineDB = false; /** The query in flight refreshes a cache-only window, so its response replaces the window. */ private _replaceOfflineSeedOnResponse = false; + /** The query handed to the sync manager, from the cache read until it runs. See `executeQuery`. */ + private _queryAwaitingSync: ChannelQueryShape | undefined; protected _sortComparatorFactory: ChannelSortComparatorFactory | undefined; sortComparator: (a: Channel, b: Channel) => number; filterBuilder: FilterBuilder; @@ -364,6 +366,12 @@ export class ChannelPaginator extends BasePaginator memberUserNameFilterResolver, dataFieldFilterResolver, ]); + // since channels are reactive objects, we only really care about updating the + // list if their order of membership in the list changes. + this.state.addPreprocessor((next, previous) => { + if (previous && hasSameItems(previous.items, next.items)) + next.items = previous.items; + }); } /** @@ -716,11 +724,25 @@ export class ChannelPaginator extends BasePaginator if (!shouldDeferUntilSynced) return await super.executeQuery(effectiveParams); + if ( + this._queryAwaitingSync && + !this.config.hasPaginationQueryShapeChanged(this._queryAwaitingSync, queryShape) + ) { + return; + } + this._queryAwaitingSync = queryShape; + if (!this.isInitialized) { - const cachedChannels = await this.preloadFirstPageFromOfflineDb({ - ...effectiveParams, - queryShape, - }); + let cachedChannels: Channel[] | undefined; + try { + cachedChannels = await this.preloadFirstPageFromOfflineDb({ + ...effectiveParams, + queryShape, + }); + } catch (error) { + this._queryAwaitingSync = undefined; + throw error; + } if (cachedChannels?.length) { // Seed via `setItems` (which ingests the page into the interval/index storage), so that the items // actually appear within the adequate index. @@ -745,10 +767,12 @@ export class ChannelPaginator extends BasePaginator // Otherwise, the sync status change will never fire and so `executeQuery` will never really be // run. if (offlineDb.syncManager.isSynced) { + this._queryAwaitingSync = undefined; return await super.executeQuery(refreshParams); } offlineDb.syncManager.scheduleSyncStatusChangeCallback(this.id, async () => { + this._queryAwaitingSync = undefined; await this.executeQuery(refreshParams); }); } @@ -825,6 +849,7 @@ export class ChannelPaginator extends BasePaginator super.resetState(); // nothing is listed anymore, so the list holds no channels this._itemIndex.clear(); + this._queryAwaitingSync = undefined; } removeItem(params: { id?: string; item?: Channel }): ItemCoordinates { diff --git a/src/pagination/paginators/MessageIntervalPaginator.ts b/src/pagination/paginators/MessageIntervalPaginator.ts index 1a6e71b25f..f021b476f2 100644 --- a/src/pagination/paginators/MessageIntervalPaginator.ts +++ b/src/pagination/paginators/MessageIntervalPaginator.ts @@ -641,6 +641,15 @@ export class MessageIntervalPaginator extends BasePaginator< ); } + /** A jump sends its own `limit`, so its head/tail flags are derived against that, not `pageSize`. */ + protected override getRequestedPageSize( + queryShape: MessageQueryShape | undefined, + ): number { + return queryShape && this.isJumpQueryShape(queryShape) + ? (queryShape.limit ?? this.pageSize) + : this.pageSize; + } + jumpToMessage = async ( messageId: string, { @@ -688,7 +697,9 @@ export class MessageIntervalPaginator extends BasePaginator< }; } else { const result = await this.executeQuery({ - queryShape: { id_around: messageId, limit: pageSize }, + // Always send a limit, since the replies endpoint ignores `id_around` without one and returns the + // newest replies instead. + queryShape: { id_around: messageId, limit: pageSize ?? this.pageSize }, ...(watch && { requestOptions: { watch: true } }), updateState: false, }); @@ -1321,7 +1332,7 @@ export class MessageIntervalPaginator extends BasePaginator< * `isTail`/`hasMoreTail` are set; intervals entirely newer keep their flags (unloaded older * messages may still sit between them and the cutoff). The active window is re-emitted once. */ - truncate = ({ truncatedAt }: { truncatedAt: TimestampNS }) => { + truncate({ truncatedAt }: { truncatedAt: TimestampNS }) { const cutoff = truncatedAt; if (!Number.isFinite(cutoff)) return; @@ -1389,7 +1400,7 @@ export class MessageIntervalPaginator extends BasePaginator< // Nothing loaded survived the truncation. this.state.partialNext({ items: [] }); } - }; + } applyMessageDeletionForUser = ({ userId, @@ -1475,7 +1486,7 @@ export class MessageIntervalPaginator extends BasePaginator< * re-emits the active window a single time (if it held an affected message) rather than one * `ingestItem` re-emit per message. */ - reflectUserUpdate = (user: UserResponse) => { + reflectUserUpdate(user: UserResponse) { const activeIds = new Set((this.items ?? []).map((m) => this.getItemId(m))); let activeAffected = false; // Batch: a user rename can touch many messages; coalesce the shared-store fan-out to sibling @@ -1493,7 +1504,7 @@ export class MessageIntervalPaginator extends BasePaginator< items: (this.items ?? []).map((m) => this.getItem(this.getItemId(m)) ?? m), }); } - }; + } /** * Map a timestamp to a loaded message — the first message in the latest (head) window whose @@ -1595,13 +1606,8 @@ const makeDeriveCursor = if ((ctx.queryShape as MessagePaginationParams)?.created_at_around) { return injectCursor( - deriveCreatedAtAroundPaginationFlags< - LocalMessage, - MessagePaginationParams, - MessageIntervalPaginator - >({ + deriveCreatedAtAroundPaginationFlags({ ...ctx, - paginator, page: pageWithPermittedMessages, requestedPageSize: requestedPageSizeAfterAdjustment, }), diff --git a/src/pagination/paginators/MessagePaginator.ts b/src/pagination/paginators/MessagePaginator.ts index 9ffd04efd8..95322c6a9e 100644 --- a/src/pagination/paginators/MessagePaginator.ts +++ b/src/pagination/paginators/MessagePaginator.ts @@ -11,7 +11,8 @@ import { MessageIntervalPaginator, type MessageQueryShape, } from './MessageIntervalPaginator'; -import type { LocalMessage, TimestampNS } from '../../types'; +import type { LocalMessage, TimestampNS, UserResponse } from '../../types'; +import type { EntityStoreChangeBatch } from '../../entityStore/EntityStore'; import { asTimestampNS, nsToDate } from '../../utils/time'; import { StateStore } from '@stream-io/state-store'; @@ -276,22 +277,67 @@ export class MessagePaginator extends MessageIntervalPaginator { return interval; } + onEntitiesChanged(batch: EntityStoreChangeBatch) { + super.onEntitiesChanged(batch); + const { lastMessage } = this.aggregateState.getLatestValue(); + if (!lastMessage || !batch.changedIds.has(lastMessage.id)) return; + const stored = this.getItem(lastMessage.id); + if (stored) this.aggregateState.partialNext({ lastMessage: stored }); + } + + reflectUserUpdate(user: UserResponse) { + super.reflectUserUpdate(user); + const { lastMessage } = this.aggregateState.getLatestValue(); + if (lastMessage?.user?.id !== user.id) return; + this.aggregateState.partialNext({ lastMessage: { ...lastMessage, user } }); + } + removeItem( params: Parameters[0], ): ReturnType { const removedId = params.id ?? (params.item ? this.getItemId(params.item) : undefined); - const wasLatest = - !!removedId && this.aggregateState.getLatestValue().lastMessage?.id === removedId; + const { lastMessage, seededLastMessageAt } = this.aggregateState.getLatestValue(); + const wasLatest = !!removedId && lastMessage?.id === removedId; const result = super.removeItem(params); if (wasLatest) { // The tracked latest was hard-removed; fall back to the newest still-loaded message that // passes the filter. `trackLastMessage` cannot do this (it only advances), so recompute here. - this.aggregateState.partialNext({ lastMessage: this.recomputeLastMessage() }); + const previous = this.recomputeLastMessage(); + const previousAt = getMessageCreatedAtTimestamp(previous); + const removedAt = getMessageCreatedAtTimestamp(lastMessage); + const isSeedFromRemovedMessage = + seededLastMessageAt !== null && + removedAt !== null && + seededLastMessageAt <= removedAt; + // The server moves `last_message_at` back to the previous message without sending it, so the + // seed takes the previous loaded message's time. With none loaded the new value is unknown, + // and the seed stays. + this.aggregateState.partialNext({ + lastMessage: previous, + seededLastMessageAt: + isSeedFromRemovedMessage && previousAt !== null + ? previousAt + : seededLastMessageAt, + }); } return result; } + truncate(params: Parameters[0]) { + super.truncate(params); + const { lastMessage, seededLastMessageAt } = this.aggregateState.getLatestValue(); + const isTruncated = (timestamp: TimestampNS | null) => + timestamp !== null && timestamp < params.truncatedAt; + const isLastMessageTruncated = isTruncated(getMessageCreatedAtTimestamp(lastMessage)); + const isSeedTruncated = isTruncated(seededLastMessageAt); + if (!isLastMessageTruncated && !isSeedTruncated) return; + this.aggregateState.partialNext({ + lastMessage: isLastMessageTruncated ? this.recomputeLastMessage() : lastMessage, + seededLastMessageAt: isSeedTruncated ? null : seededLastMessageAt, + }); + } + /** * The still-loaded message with the greatest `created_at` that passes * {@link shouldAdvanceLastMessage} (skips system / thread-only per config), from the newest-loaded @@ -442,7 +488,7 @@ export class MessagePaginator extends MessageIntervalPaginator { queryShape: { // `created_at_around` is a request field and still takes a `Date`. created_at_around: nsToDate(lastReadBoundary), - limit: options?.pageSize, + limit: options?.pageSize ?? this.pageSize, }, updateState: false, }); diff --git a/test/unit/ChannelManager.test.ts b/test/unit/ChannelManager.test.ts index 2765581c9d..8898cf03d0 100644 --- a/test/unit/ChannelManager.test.ts +++ b/test/unit/ChannelManager.test.ts @@ -1361,6 +1361,26 @@ describe('ChannelManager', () => { }); }); + describe('event message.deleted', () => { + it('re-sorts a loaded channel on a hard delete only', async () => { + const cid = 'messaging:deleted-1'; + const channel = makeChannel(cid); + client.channelManager.getOrCreateChannel(cid, () => channel); + const channelManager = useClientManager({ + paginators: [new ChannelPaginator({ client })], + }); + channelManager.registerSubscriptions(); + const ingestChannel = vi.spyOn(channelManager, 'ingestChannel'); + + // events are handled in order, so once the hard delete is handled the soft one was too + client.dispatchEvent({ type: 'message.deleted', cid, hard_delete: false }); + client.dispatchEvent({ type: 'message.deleted', cid, hard_delete: true }); + + await vi.waitFor(() => expect(ingestChannel).toHaveBeenCalled()); + expect(ingestChannel).toHaveBeenCalledTimes(1); + }); + }); + describe('event channel.hidden', () => { const seed = (paginator: ChannelPaginator, channels: Channel[]) => paginator.setItems({ @@ -1411,6 +1431,29 @@ describe('ChannelManager', () => { expect(regular.items?.map((c) => c.cid)).toEqual([cid]); }); }); + + it.each(['message.new', 'notification.message_new'] as const)( + 're-adds the channel on %s, which unhides it on the server without channel.visible', + async (type) => { + const cid = `messaging:hidden-${type.replace('.', '-')}`; + const channel = makeChannel(cid); + client.channelManager.getOrCreateChannel(cid, () => channel); + + const regular = new ChannelPaginator({ client }); + const channelManager = useClientManager({ paginators: [regular] }); + channelManager.registerSubscriptions(); + + client.dispatchEvent({ type: 'channel.hidden', cid } as const); + await vi.waitFor(() => expect(regular.items ?? []).toEqual([])); + + client.dispatchEvent({ type, cid }); + + await vi.waitFor(() => { + expect(channel.data?.hidden).toBe(false); + expect(regular.items?.map((c) => c.cid)).toEqual([cid]); + }); + }, + ); }); describe('channel resolution from the event', () => { diff --git a/test/unit/EventHandlerPipeline.test.ts b/test/unit/EventHandlerPipeline.test.ts index 8ac43ee7cf..c126165b65 100644 --- a/test/unit/EventHandlerPipeline.test.ts +++ b/test/unit/EventHandlerPipeline.test.ts @@ -511,11 +511,12 @@ describe('EventHandlerPipeline', () => { }, }; + const error = new Error('fail'); const boom = { id: 'boom', handle: () => { order.push('boom'); - throw new Error('fail'); + throw error; }, }; @@ -534,7 +535,10 @@ describe('EventHandlerPipeline', () => { await pipeline.run(makeEvt('e'), ctx); // reverse exec: after -> boom -> before; boom throws but processing continues expect(order).toEqual(['before', 'boom', 'after']); - expect(consoleErrorSpy).toHaveBeenCalled(); // logged + expect(consoleErrorSpy).toHaveBeenCalledWith( + expect.stringContaining('handler failed'), + expect.objectContaining({ error, handlerId: 'boom' }), + ); }); it('snapshot isolation: handlers added during a run do not affect the current event', async () => { diff --git a/test/unit/channel.test.js b/test/unit/channel.test.js index c716c3c084..f115d55366 100644 --- a/test/unit/channel.test.js +++ b/test/unit/channel.test.js @@ -1660,6 +1660,20 @@ describe('Channel _handleChannelEvent', function () { expect(pinnedIds()).to.eql([]); }); + + it('takes the channel data from the event and keeps the fields it leaves out', () => { + channel.data = { ...channel.data, hidden: false, updated_at: 1 }; + const updatedAt = convertDateToTimestamp('2020-02-01T00:00:00.000Z'); + + channel._handleChannelEvent({ + type: 'channel.truncated', + channel: { truncated_at: updatedAt, updated_at: updatedAt }, + }); + + expect(channel.data.updated_at).to.equal(updatedAt); + expect(channel.data.truncated_at).to.equal(updatedAt); + expect(channel.data.hidden).to.equal(false); + }); }); describe('message.deleted', () => { diff --git a/test/unit/pagination/paginators/BasePaginator.test.ts b/test/unit/pagination/paginators/BasePaginator.test.ts index 0e2539966c..1ce6ef3291 100644 --- a/test/unit/pagination/paginators/BasePaginator.test.ts +++ b/test/unit/pagination/paginators/BasePaginator.test.ts @@ -1765,6 +1765,65 @@ describe('BasePaginator', () => { ]); }); + it('removes an item from the interval that holds it, even after it sorts into another one', async () => { + const index = new StoreBackedItemIndex({ getEntityId: ({ id }) => id }); + paginator = new Paginator({ itemIndex: index }); + paginator.sortComparator = makeComparator({ + sort: [{ field: 'age', direction: -1 }], + }); + paginator.ingestPage({ page: [b, d], setActive: true }); + // Parked below the loaded window, then updated in place so it now sorts inside it, the way a + // listed channel moves once its last message arrives. + const moved: TestItem = { id: 'moved', age: 1 }; + paginator.ingestItem(moved); + moved.age = 22; + + paginator.removeItem({ id: 'moved' }); + + // @ts-expect-error accessing protected property _itemIntervals + const intervals = Array.from(paginator._itemIntervals.values()) as { + itemIds: string[]; + }[]; + expect(intervals.some((interval) => interval.itemIds.includes('moved'))).toBe( + false, + ); + + // The next query reconciles against the intervals and has to finish. + const query = paginator.toTail(); + await sleep(0); + paginator.queryResolve({ items: [x] }); + await query.catch(() => undefined); + expect(paginator.isLoading).toBe(false); + }); + + it('locates unmoved and new items without scanning every interval', () => { + const index = new StoreBackedItemIndex({ getEntityId: ({ id }) => id }); + paginator = new Paginator({ itemIndex: index }); + paginator.sortComparator = makeComparator({ + sort: [{ field: 'age', direction: -1 }], + }); + const stored: TestItem = { id: 'stored', age: 22 }; + paginator.ingestPage({ page: [b, stored, d], setActive: true }); + // @ts-expect-error spying on a protected method + const scan = vi.spyOn(paginator, 'findIntervalHoldingItem'); + // @ts-expect-error spying on a protected method + const search = vi.spyOn(paginator, 'locateByItemInInterval'); + + // Updated in place without changing where it sorts. + stored.name = 'renamed'; + paginator.ingestItem(stored); + expect(scan).not.toHaveBeenCalled(); + + // No interval holds it, so its position is found without walking the interval for it. + search.mockClear(); + paginator.ingestItem({ id: 'new', age: 21 }); + expect(search).toHaveBeenCalledWith( + expect.objectContaining({ plateauScan: false }), + ); + expect(scan).not.toHaveBeenCalled(); + expect(paginator.items?.map(({ id }) => id)).toEqual(['b', 'stored', 'new', 'd']); + }); + it('merges all the overlapping anchored intervals, parts of logical intervals with target interval', () => { let keys: string[] = []; paginator.ingestPage({ page: [c, d], setActive: true }); diff --git a/test/unit/pagination/paginators/ChannelPaginator.test.ts b/test/unit/pagination/paginators/ChannelPaginator.test.ts index 5d5f5a5df2..d2ad492817 100644 --- a/test/unit/pagination/paginators/ChannelPaginator.test.ts +++ b/test/unit/pagination/paginators/ChannelPaginator.test.ts @@ -8,6 +8,7 @@ import { DEFAULT_PAGINATION_OPTIONS, type FilterBuilderGenerators, formatMessage, + type Interval, PaginatorCursor, type StreamChat, } from '../../../../src'; @@ -503,21 +504,28 @@ describe('ChannelPaginator', () => { client, filters: { last_updated: new Date(1000).toISOString() }, }); - channel1.data = { updated_at: undefined }; + channel1.data = { created_at: undefined }; setLastMessageAt(channel1, new Date(1000)); expect(paginator.matchesFilter(channel1)).toBeTruthy(); channel1.data = { - updated_at: convertDateToTimestamp(new Date(1000).toISOString()), + created_at: convertDateToTimestamp(new Date(1000).toISOString()), }; setLastMessageAt(channel1, null); expect(paginator.matchesFilter(channel1)).toBeTruthy(); - channel1.data = { updated_at: undefined }; + channel1.data = { created_at: undefined }; setLastMessageAt(channel1, null); expect(paginator.matchesFilter(channel1)).toBeFalsy(); + + // the server ignores `updated_at` for `last_updated` + channel1.data = { + created_at: undefined, + updated_at: convertDateToTimestamp(new Date(1000).toISOString()), + }; + expect(paginator.matchesFilter(channel1)).toBeFalsy(); }); it.each([ @@ -568,7 +576,7 @@ describe('ChannelPaginator', () => { filters: { last_updated: { [operator]: new Date(1000).toISOString() } }, }); - channel1.data = { updated_at: undefined }; + channel1.data = { created_at: undefined }; scenarios.forEach(({ val, expected }) => { setLastMessageAt(channel1, new Date(val)); expect(paginator.matchesFilter(channel1)).toBe(expected); @@ -577,12 +585,12 @@ describe('ChannelPaginator', () => { setLastMessageAt(channel1, null); scenarios.forEach(({ val, expected }) => { channel1.data = { - updated_at: convertDateToTimestamp(new Date(val).toISOString()), + created_at: convertDateToTimestamp(new Date(val).toISOString()), }; expect(paginator.matchesFilter(channel1)).toBe(expected); }); - channel1.data = { updated_at: undefined }; + channel1.data = { created_at: undefined }; setLastMessageAt(channel1, null); expect(paginator.matchesFilter(channel1)).toBe(false); }); @@ -730,6 +738,98 @@ describe('ChannelPaginator', () => { }); }); + describe('unchanged items', () => { + const setup = () => { + const newer = new Channel(client, 'type', 'unchanged-newer', {}); + setLastMessageAt(newer, new Date('1971-01-01T00:00:00.000Z')); + const older = new Channel(client, 'type', 'unchanged-older', {}); + setLastMessageAt(older, new Date('1970-01-01T00:00:00.000Z')); + const paginator = new ChannelPaginator({ client, filters: {} }); + paginator.setItems({ + valueOrFactory: [newer, older], + isFirstPage: true, + isLastPage: true, + }); + return { newer, older, paginator }; + }; + + it('keeps the items array when an ingested channel stays in place', () => { + const { newer, paginator } = setup(); + const items = paginator.items; + + paginator.ingestItem(newer); + + expect(paginator.items).toBe(items); + }); + + it('publishes a new items array when the order changes', () => { + const { older, paginator } = setup(); + const items = paginator.items; + + setLastMessageAt(older, new Date('1972-01-01T00:00:00.000Z')); + paginator.boost(older.cid); + paginator.ingestItem(older); + + expect(paginator.items).not.toBe(items); + expect(paginator.items?.map((c) => c.id)).toEqual([ + 'unchanged-older', + 'unchanged-newer', + ]); + }); + + it('keeps the items array when setItems gets the same channels', () => { + const { newer, older, paginator } = setup(); + const items = paginator.items; + + paginator.setItems({ + valueOrFactory: [newer, older], + isFirstPage: true, + isLastPage: true, + }); + + expect(paginator.items).toBe(items); + }); + + it('keeps the items array when a page with the same channels is ingested', () => { + const { newer, older, paginator } = setup(); + const items = paginator.items; + + paginator.ingestPage({ + page: [newer, older], + isHead: true, + isTail: true, + setActive: true, + }); + + expect(paginator.items).toBe(items); + }); + + it('keeps the items array when the active interval is activated again', () => { + const { paginator } = setup(); + const items = paginator.items; + + paginator.setActiveInterval(paginator.itemIntervals[0] as Interval); + + expect(paginator.items).toBe(items); + }); + + it('keeps the items array when a refresh returns the same channels, and still ends loading', async () => { + const { newer, older, paginator } = setup(); + vi.spyOn(client, 'queryChannelsAndHydrate').mockResolvedValue({ + channels: [newer, older], + duration: '0.1ms', + }); + await paginator.toTail(); + const items = paginator.items; + + // the refresh reconnect recovery runs + await paginator.toTail({ keepPreviousItems: true, reset: 'yes' }); + + expect(paginator.items).toBe(items); + expect(paginator.isLoading).toBe(false); + }); + }); + describe('empty sort', () => { // Regression: `` defaults its `sort` prop to `[]`, which `??` does not treat as // "unset". The comparator then had no fields, every comparison tied, and the id tiebreaker @@ -1117,6 +1217,37 @@ describe('ChannelPaginator', () => { ); }); + it('keeps paginating after removing a channel that moved into the loaded window from the logical tail', async () => { + const [a, b, c, d] = ['a', 'b', 'c', 'd'].map((id, index) => { + const channel = new Channel(client, 'type', id, {}); + setLastMessageAt(channel, new Date(Date.UTC(1970, 0, 10 - index * 2))); + return channel; + }); + vi.spyOn(client, 'queryChannelsAndHydrate') + .mockResolvedValueOnce({ channels: [a, b], duration: '0.1ms' }) + .mockResolvedValueOnce({ channels: [c, d], duration: '0.1ms' }); + const paginator = new ChannelPaginator({ + client, + filters: {}, + paginatorOptions: { pageSize: 2 }, + sort: [{ field: 'last_message_at', direction: -1 }], + }); + await paginator.toTail(); + + // It is added with no messages yet, so it sorts below the loaded page into the logical tail. + const added = new Channel(client, 'type', 'added', {}); + paginator.ingestItem(added); + // Its first message arrives on the same instance, so it now sorts between a and b. + setLastMessageAt(added, new Date(Date.UTC(1970, 0, 9))); + // Then it leaves the list (archived, say). + paginator.removeItem({ item: added }); + + await paginator.toTail(); + + expect(paginator.isLoading).toBe(false); + expect(paginator.items?.map((channel) => channel.id)).toEqual(['a', 'b', 'c', 'd']); + }); + it('persists only the first page, including after paginating and live reorders', async () => { await setUpOfflineDb({ isSynced: true }); const [a, b, c, d] = ['a', 'b', 'c', 'd'].map((id, index) => { @@ -1303,6 +1434,120 @@ describe('ChannelPaginator', () => { expect(paginator.items).toStrictEqual([cachedChannel]); }); + describe('when the list is requested again before the sync completes', () => { + const setUpCachedPage = async () => { + await setUpOfflineDb({ isSynced: false }); + const cachedChannel = new Channel(client, 'type', 'cached', {}); + getChannelsForQuery.mockResolvedValue({ + channels: [{ channel: cachedChannel.data }], + }); + vi.spyOn(client, 'hydrateChannels').mockReturnValue([cachedChannel]); + const queryChannels = vi + .spyOn(client, 'queryChannelsAndHydrate') + .mockResolvedValue({ channels: [cachedChannel], duration: '0.1ms' }); + return { queryChannels }; + }; + + const runScheduledQuery = async () => { + client.offlineDb!.syncManager.isSynced = true; + await scheduleSyncStatusChangeCallback.mock.calls.at(-1)![1](); + }; + + it('reads the cache and queries once for calls made in the same tick', async () => { + const { queryChannels } = await setUpCachedPage(); + const paginator = makePaginator(); + + await Promise.all([paginator.reload(), paginator.reload(), paginator.toTail()]); + await runScheduledQuery(); + + expect(getChannelsForQuery).toHaveBeenCalledTimes(1); + expect(scheduleSyncStatusChangeCallback).toHaveBeenCalledTimes(1); + expect(queryChannels).toHaveBeenCalledTimes(1); + }); + + it('reads the cache and queries once for calls made after the cache was read', async () => { + const { queryChannels } = await setUpCachedPage(); + const paginator = makePaginator(); + + await paginator.reload(); + await paginator.reload(); + await paginator.toTail(); + await runScheduledQuery(); + + expect(getChannelsForQuery).toHaveBeenCalledTimes(1); + expect(scheduleSyncStatusChangeCallback).toHaveBeenCalledTimes(1); + expect(queryChannels).toHaveBeenCalledTimes(1); + }); + + it('reads the cache again for a different query made in the same tick', async () => { + await setUpCachedPage(); + const paginator = makePaginator(); + + const first = paginator.reload(); + paginator.staticFilters = { type: 'other' } as ChannelFilters; + await Promise.all([first, paginator.reload()]); + + expect(getChannelsForQuery).toHaveBeenCalledTimes(2); + expect( + getChannelsForQuery.mock.calls[1][0].options.filter_conditions, + ).toStrictEqual({ + type: 'other', + }); + }); + + it('reads the cache again after the state was reset', async () => { + await setUpCachedPage(); + const paginator = makePaginator(); + + await paginator.reload(); + paginator.resetState(); + await paginator.reload(); + + expect(getChannelsForQuery).toHaveBeenCalledTimes(2); + }); + + it('reads the cache again after the cache read failed', async () => { + await setUpCachedPage(); + getChannelsForQuery.mockRejectedValueOnce(new Error('db failed')); + const paginator = makePaginator(); + paginator.updateConfig({ throwErrors: true }); + + await expect(paginator.reload()).rejects.toThrow('db failed'); + await paginator.reload(); + + expect(getChannelsForQuery).toHaveBeenCalledTimes(2); + expect(scheduleSyncStatusChangeCallback).toHaveBeenCalledTimes(1); + }); + + it('waits for the next sync again once the waiting query ran', async () => { + await setUpCachedPage(); + const paginator = makePaginator(); + + await paginator.reload(); + await runScheduledQuery(); + // the socket dropped, so the next sync is pending again + client.offlineDb!.syncManager.isSynced = false; + await paginator.reload(); + + expect(scheduleSyncStatusChangeCallback).toHaveBeenCalledTimes(2); + }); + + it('waits for the next sync again once the query ran right after the cache read', async () => { + await setUpCachedPage(); + getChannelsForQuery.mockImplementationOnce(async () => { + client.offlineDb!.syncManager.isSynced = true; + return null; + }); + const paginator = makePaginator(); + + await paginator.reload(); + client.offlineDb!.syncManager.isSynced = false; + await paginator.reload(); + + expect(scheduleSyncStatusChangeCallback).toHaveBeenCalledTimes(1); + }); + }); + it('seeds the preloaded channels into the index so a concurrent ingest does not collapse the list', async () => { // The cold start preload must SEED the paginator (populate the interval/index), not // just set the displayed `items`. Otherwise a channel ingested concurrently during the presync diff --git a/test/unit/pagination/paginators/MessagePaginator.test.ts b/test/unit/pagination/paginators/MessagePaginator.test.ts index 495e339747..675302fda9 100644 --- a/test/unit/pagination/paginators/MessagePaginator.test.ts +++ b/test/unit/pagination/paginators/MessagePaginator.test.ts @@ -424,7 +424,7 @@ describe('MessagePaginator', () => { await paginator.jumpToMessage('target', { watchChannel: true }); expect(channel.query).toHaveBeenCalledWith({ - messages: { id_around: 'target', limit: undefined }, + messages: { id_around: 'target', limit: paginator.pageSize }, }); }); @@ -436,7 +436,7 @@ describe('MessagePaginator', () => { await paginator.jumpToMessage('target'); expect(channel.query).toHaveBeenCalledWith({ - messages: { id_around: 'target', limit: undefined }, + messages: { id_around: 'target', limit: paginator.pageSize }, }); }); @@ -1503,6 +1503,47 @@ describe('MessagePaginator', () => { expect(paginator.items?.find((m) => m.id === 'a1')?.user?.name).toBe('Renamed A'); }); + it('renames the author of the last message', () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + paginator.ingestPage({ + page: [ + createMessage({ + id: 'a1', + user: { id: 'A' }, + created_at: convertDateToTimestamp('2021-01-01T00:00:00.000Z'), + }), + ], + isHead: true, + isTail: true, + setActive: true, + }); + + paginator.reflectUserUpdate({ id: 'A', name: 'Renamed A' }); + + expect(paginator.lastMessage?.user?.name).toBe('Renamed A'); + }); + + it('leaves the last message alone when another user is renamed', () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + paginator.ingestPage({ + page: [ + createMessage({ + id: 'b1', + user: { id: 'B', name: 'B' }, + created_at: convertDateToTimestamp('2021-01-01T00:00:00.000Z'), + }), + ], + isHead: true, + isTail: true, + setActive: true, + }); + const lastMessage = paginator.lastMessage; + + paginator.reflectUserUpdate({ id: 'A', name: 'Renamed A' }); + + expect(paginator.lastMessage).toBe(lastMessage); + }); + it('skips messages already carrying the same user object, so siblings are not dirtied back', () => { // Two collections over ONE store, as channel.messagePaginator and pinnedMessagesPaginator // are. Both run reflectUserUpdate for the same event; without the reference guard the second @@ -1551,6 +1592,41 @@ describe('MessagePaginator', () => { }); }); + describe('last message refreshed from the store', () => { + const setup = () => { + const store = new EntityStore({ getEntityId: (m) => m.id }); + const linkedChannel = { + ...channel, + getClient: () => ({ messageStore: store, userId: 'me' }), + } as unknown as Channel; + const paginator = new MessagePaginator({ channel: linkedChannel }); + const message = (id: string, day: string) => + createMessage({ + cid: linkedChannel.cid, + id, + user: { id: 'A' }, + created_at: convertDateToTimestamp(`2021-01-${day}T00:00:00.000Z`), + }); + return { message, paginator, store }; + }; + + it('reflects a write to the store that does not go through the paginator', () => { + const { message, paginator, store } = setup(); + const latest = message('m1', '01'); + paginator.ingestPage({ + page: [latest], + isHead: true, + isTail: true, + setActive: true, + }); + + // a soft delete writes the store directly + store.upsert({ ...latest, type: 'deleted' }); + + expect(paginator.lastMessage?.type).toBe('deleted'); + }); + }); + describe.todo('postQueryReconcile and deriveCursor for', () => {}); describe('linear pagination', () => { describe('updates the hasMoreTail flag only if the first message on page is the first message in interval', () => { @@ -2015,6 +2091,144 @@ describe('MessagePaginator', () => { expect(paginator.items?.map((m) => m.id)).toEqual(['m1']); expect(partialNextSpy).not.toHaveBeenCalled(); }); + + it('clears the last message and the sort key when nothing survives', () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + paginator.seedLastMessageAt(convertDateToTimestamp('2020-01-01T00:00:00.000Z')); + paginator.ingestPage({ + page: [msg('m1', '01'), msg('m2', '02')], + isHead: true, + isTail: true, + setActive: true, + }); + + paginator.truncate({ + truncatedAt: convertDateToTimestamp('2020-01-03T00:00:00.000Z'), + }); + + expect(paginator.lastMessage).toBeNull(); + expect(paginator.lastMessageAt).toBeNull(); + }); + + it('keeps the last message when it is newer than the cutoff', () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + paginator.ingestPage({ + page: [msg('m1', '01'), msg('m5', '05')], + isHead: true, + isTail: true, + setActive: true, + }); + + paginator.truncate({ + truncatedAt: convertDateToTimestamp('2020-01-03T00:00:00.000Z'), + }); + + expect(paginator.lastMessage?.id).toBe('m5'); + expect(paginator.lastMessageAt).toBe( + convertDateToTimestamp('2020-01-05T00:00:00.000Z'), + ); + }); + + it('keeps a last message newer than the cutoff that was tracked without being loaded', () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + paginator.seedLastMessageAt(convertDateToTimestamp('2020-01-01T00:00:00.000Z')); + // a replayed offline send is tracked before its message.new is ingested + paginator.trackLastMessage(msg('m5', '05')); + + paginator.truncate({ + truncatedAt: convertDateToTimestamp('2020-01-03T00:00:00.000Z'), + }); + + expect(paginator.lastMessage?.id).toBe('m5'); + }); + + it('clears a seed older than the cutoff when no message is loaded', () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + paginator.seedLastMessageAt(convertDateToTimestamp('2020-01-01T00:00:00.000Z')); + + paginator.truncate({ + truncatedAt: convertDateToTimestamp('2020-01-03T00:00:00.000Z'), + }); + + expect(paginator.lastMessageAt).toBeNull(); + }); + + it('keeps a seed newer than the cutoff', () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + paginator.seedLastMessageAt(convertDateToTimestamp('2020-01-05T00:00:00.000Z')); + paginator.ingestPage({ + page: [msg('m1', '01')], + isHead: true, + isTail: true, + setActive: true, + }); + + paginator.truncate({ + truncatedAt: convertDateToTimestamp('2020-01-03T00:00:00.000Z'), + }); + + expect(paginator.lastMessage).toBeNull(); + expect(paginator.lastMessageAt).toBe( + convertDateToTimestamp('2020-01-05T00:00:00.000Z'), + ); + }); + }); + + describe('hard-removing the last message', () => { + it('moves the sort key back to the previous message', () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + // the server's last_message_at is the message about to be removed + paginator.seedLastMessageAt(convertDateToTimestamp('2020-01-02T00:00:00.000Z')); + paginator.ingestPage({ + page: [msg('m1', '01'), msg('m2', '02')], + isHead: true, + isTail: true, + setActive: true, + }); + + paginator.removeItem({ id: 'm2' }); + + expect(paginator.lastMessage?.id).toBe('m1'); + expect(paginator.lastMessageAt).toBe( + convertDateToTimestamp('2020-01-01T00:00:00.000Z'), + ); + }); + + it('keeps a seed newer than the removed message', () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + // a newer message exists on the server that is not loaded + paginator.seedLastMessageAt(convertDateToTimestamp('2020-01-05T00:00:00.000Z')); + paginator.ingestPage({ + page: [msg('m1', '01'), msg('m2', '02')], + isHead: true, + isTail: true, + setActive: true, + }); + + paginator.removeItem({ id: 'm2' }); + + expect(paginator.lastMessageAt).toBe( + convertDateToTimestamp('2020-01-05T00:00:00.000Z'), + ); + }); + + it('keeps the seed when no previous message is loaded', () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + // older messages exist on the server, but only the latest is loaded + paginator.seedLastMessageAt(convertDateToTimestamp('2020-01-02T00:00:00.000Z')); + paginator.ingestPage({ + page: [msg('m2', '02')], + isHead: true, + setActive: true, + }); + + paginator.removeItem({ id: 'm2' }); + + expect(paginator.lastMessage).toBeNull(); + expect(paginator.lastMessageAt).toBe( + convertDateToTimestamp('2020-01-02T00:00:00.000Z'), + ); + }); }); describe('message.new routing (replaces the isUpToDate flag)', () => { @@ -4129,4 +4343,255 @@ describe('MessagePaginator', () => { expect(paginator.lastMessage?.id).toBe('m0'); }); }); + + describe('jump queries leave the list paginating', () => { + const notifications = { addError: vi.fn() }; + const m = (id: string, minute: number) => + createMessage({ + cid: 'channel-id', + id, + created_at: convertDateToTimestamp( + `2020-01-01T00:${String(minute).padStart(2, '0')}:00.000Z`, + ), + }); + const respondWith = (endpoint: 'getReplies' | 'query', messages: LocalMessage[]) => + (channel[endpoint] as unknown as ReturnType) + .mockReset() + .mockResolvedValue({ messages }); + // The latest page, seeded the way a channel opens: nothing newer to load. + const loadNewestPage = (paginator: MessagePaginator) => + paginator.seedFirstPageSync([m('m50', 50), m('m51', 51), m('m52', 52)], 3); + + beforeEach(() => { + (channel as unknown as { getClient: () => unknown }).getClient = () => ({ + getReplies: channel.getReplies, + notifications, + user: { id: 'me' }, + }); + (channel as unknown as { state: unknown }).state = { read: {} }; + }); + + it('ends loading when the jump target is not in the response', async () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + loadNewestPage(paginator); + respondWith('query', [m('m10', 10), m('m11', 11)]); + + expect(await paginator.jumpToMessage('missing')).toBe(false); + expect(paginator.isLoading).toBe(false); + + respondWith('query', [m('m49', 49)]); + await paginator.toTail(); + expect(channel.query).toHaveBeenCalledTimes(1); + }); + + it('ends loading after the first-unread jump queries around the last read time', async () => { + (channel as unknown as { state: unknown }).state = { + read: { + me: { + first_unread_message_id: null, + last_read: convertDateToTimestamp('2020-01-01T00:20:00.000Z'), + last_read_message_id: null, + }, + }, + }; + const paginator = new MessagePaginator({ channel, itemIndex }); + paginator.pageSize = 4; + // Every loaded message is newer than the last read time, so the boundary has to be queried. + loadNewestPage(paginator); + // A full page split evenly around the boundary, so there is more to load on both sides. + respondWith('query', [m('m18', 18), m('m19', 19), m('m21', 21), m('m22', 22)]); + + expect(await paginator.jumpToTheFirstUnreadMessage()).toBe(true); + expect(channel.query).toHaveBeenCalledWith({ + messages: expect.objectContaining({ + created_at_around: expect.any(Date), + limit: 4, + }), + }); + expect(paginator.items?.map((message) => message.id)).toContain('m21'); + expect(paginator.isLoading).toBe(false); + + respondWith('query', [m('m18', 18)]); + await paginator.toTail(); + expect(channel.query).toHaveBeenCalledTimes(1); + }); + + it('sends a limit with a thread jump, which the replies endpoint needs to honour id_around', async () => { + const paginator = new MessagePaginator({ + channel, + itemIndex, + parentMessageId: 'parent-1', + }); + loadNewestPage(paginator); + respondWith('getReplies', [m('m10', 10), m('m11', 11), m('m12', 12)]); + + expect(await paginator.jumpToMessage('m11')).toBe(true); + expect(channel.getReplies).toHaveBeenCalledWith( + { parent_id: 'parent-1' }, + expect.objectContaining({ id_around: 'm11', limit: paginator.pageSize }), + ); + }); + + it('derives the jump window flags from the limit the jump sent, not the page size', async () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + loadNewestPage(paginator); + // A full 5-message page centred on the target: there is more on both sides. + respondWith('query', [ + m('m9', 9), + m('m10', 10), + m('m11', 11), + m('m12', 12), + m('m13', 13), + ]); + + expect(await paginator.jumpToMessage('m11', { pageSize: 5 })).toBe(true); + expect(paginator.pageSize).toBeGreaterThan(5); + expect(paginator.hasMoreHead).toBe(true); + expect(paginator.hasMoreTail).toBe(true); + }); + + it.each([ + ['newer', 'toHead', 'id_lt', 'm9'], + ['older', 'toTail', 'id_gt', 'm13'], + ] as const)( + 'keeps paging the other way after an empty %s page', + async (_, emptyDirection, otherCursorParam, otherCursorId) => { + const paginator = new MessagePaginator({ channel, itemIndex }); + loadNewestPage(paginator); + respondWith('query', [ + m('m9', 9), + m('m10', 10), + m('m11', 11), + m('m12', 12), + m('m13', 13), + ]); + await paginator.jumpToMessage('m11', { pageSize: 5 }); + + // Nothing further this way: that side ends, the other keeps its cursor. + respondWith('query', []); + await paginator[emptyDirection](); + const otherWay = emptyDirection === 'toHead' ? 'toTail' : 'toHead'; + expect( + emptyDirection === 'toHead' ? paginator.hasMoreHead : paginator.hasMoreTail, + ).toBe(false); + + respondWith('query', []); + await paginator[otherWay](); + expect(channel.query).toHaveBeenCalledWith({ + messages: expect.objectContaining({ [otherCursorParam]: otherCursorId }), + }); + }, + ); + + it('has nothing either way when the first page of a list comes back empty', async () => { + const paginator = new MessagePaginator({ + channel, + itemIndex, + parentMessageId: 'parent-1', + }); + respondWith('getReplies', []); + + await paginator.toTail(); + + expect(paginator.hasMoreHead).toBe(false); + expect(paginator.hasMoreTail).toBe(false); + }); + + describe('jump window flags follow how the server fills an around-page', () => { + // The server centres the page on the target and fills a short side from the other one. + const page = (from: number, to: number) => + Array.from({ length: to - from + 1 }, (_, i) => m(`m${from + i}`, from + i)); + const jump = async (messageId: string, response: LocalMessage[], pageSize = 10) => { + const paginator = new MessagePaginator({ channel, itemIndex }); + loadNewestPage(paginator); + respondWith('query', response); + expect(await paginator.jumpToMessage(messageId, { pageSize })).toBe(true); + return paginator; + }; + + it.each([ + // 5 older, 4 newer, and the even split the other way round + ['an even split', 'm15', page(10, 19)], + ['the opposite even split', 'm14', page(10, 19)], + ])('has more both ways after %s', async (_, target, response) => { + const paginator = await jump(target, response); + expect(paginator.hasMoreHead).toBe(true); + expect(paginator.hasMoreTail).toBe(true); + }); + + it('has nothing older but more newer after a jump to the oldest message', async () => { + const paginator = await jump('m1', page(1, 10)); + expect(paginator.hasMoreTail).toBe(false); + expect(paginator.hasMoreHead).toBe(true); + }); + + it('has nothing older when the older side came back short', async () => { + const paginator = await jump('m3', page(1, 10)); + expect(paginator.hasMoreTail).toBe(false); + expect(paginator.hasMoreHead).toBe(true); + }); + + it('has nothing newer when the newer side came back short', async () => { + const paginator = await jump('m28', page(20, 29)); + expect(paginator.hasMoreHead).toBe(false); + expect(paginator.hasMoreTail).toBe(true); + }); + + it('has nothing either way when the page is shorter than the limit', async () => { + const paginator = await jump('m12', page(10, 14)); + expect(paginator.hasMoreHead).toBe(false); + expect(paginator.hasMoreTail).toBe(false); + }); + + it('has more newer after a first-unread jump near the oldest messages', async () => { + (channel as unknown as { state: unknown }).state = { + read: { + me: { + first_unread_message_id: null, + last_read: convertDateToTimestamp('2020-01-01T00:00:30.000Z'), + last_read_message_id: null, + }, + }, + }; + const paginator = new MessagePaginator({ channel, itemIndex }); + paginator.pageSize = 4; + loadNewestPage(paginator); + // Read up to before the first message: nothing older, the page filled from the newer side. + respondWith('query', page(1, 4)); + + expect(await paginator.jumpToTheFirstUnreadMessage()).toBe(true); + expect(paginator.hasMoreTail).toBe(false); + expect(paginator.hasMoreHead).toBe(true); + }); + + it('stays at the head when an evenly split page ends at the newest message', async () => { + // 4 older and 4 newer: the page alone would say "more newer", but its last item is the + // newest message of the loaded latest window. + const paginator = await jump('m48', page(44, 52), 9); + expect(paginator.hasMoreHead).toBe(false); + expect(paginator.hasMoreTail).toBe(true); + }); + + it('stays at the tail when an evenly split page starts at the oldest loaded message', async () => { + const paginator = new MessagePaginator({ channel, itemIndex }); + loadNewestPage(paginator); + // The oldest messages, already paged in down to the start of the channel. + paginator.ingestPage({ page: page(1, 3), isHead: false, isTail: true }); + // 4 older and 4 newer: the page alone would say "more older", but it starts at m1. + respondWith('query', page(1, 9)); + + expect(await paginator.jumpToMessage('m5', { pageSize: 9 })).toBe(true); + expect(paginator.hasMoreTail).toBe(false); + expect(paginator.hasMoreHead).toBe(true); + }); + + it('keeps the head of a loaded window the jump page merged into', async () => { + // Overlaps the loaded newest page (m50..m52), so the merged window ends at the dataset head. + const paginator = await jump('m45', page(41, 50)); + expect(paginator.items?.at(-1)?.id).toBe('m52'); + expect(paginator.hasMoreHead).toBe(false); + expect(paginator.hasMoreTail).toBe(true); + }); + }); + }); });