Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions src/ChannelManager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -212,6 +212,12 @@ const channelTruncatedHandler: LabeledEventHandler<EventHandlerContext> = {
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<EventHandlerContext> = {
handle: (context) => (context.event.hard_delete ? reinsertItem(context) : undefined),
id: 'ChannelManager:default-handler:message.deleted',
};

const channelVisibleHandler: LabeledEventHandler<EventHandlerContext> = {
handle: updateLists,
id: 'ChannelManager:default-handler:channel.visible',
Expand Down Expand Up @@ -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],
Expand Down
3 changes: 2 additions & 1 deletion src/EventHandlerPipeline.ts
Original file line number Diff line number Diff line change
Expand Up @@ -207,8 +207,9 @@ export class EventHandlerPipeline<CTX extends Record<string, unknown> = {}> {
ctx,
});
if (result?.action === 'stop') return;
} catch {
} catch (error) {
console.error(`[pipeline:${this.id}] handler failed`, {
error,
handlerId: handler.id ?? 'unknown',
handlerIndex: i,
});
Expand Down
10 changes: 10 additions & 0 deletions src/channel.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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));
Expand Down
30 changes: 30 additions & 0 deletions src/pagination/cursorDerivation/aroundPaginationFlags.ts
Original file line number Diff line number Diff line change
@@ -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,
};
};
72 changes: 12 additions & 60 deletions src/pagination/cursorDerivation/createdAtAroundPaginationFlags.ts
Original file line number Diff line number Diff line change
@@ -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<T, Q>,
>({
hasMoreHead,
hasMoreTail,
interval,
page,
paginator,
queryShape,
requestedPageSize,
}: CursorDeriveContext<T, Q> & { paginator: P }): PaginationFlags => {
let flags: PaginationFlags = { hasMoreHead, hasMoreTail };
if (!queryShape?.created_at_around) return flags;
}: CursorDeriveContext<T, Q>): 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,
});
};
50 changes: 11 additions & 39 deletions src/pagination/cursorDerivation/idAroundPaginationFlags.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { deriveAroundPaginationFlags } from './aroundPaginationFlags';
import type { CursorDeriveContext, PaginationFlags } from '../paginators';

export const deriveIdAroundPaginationFlags = <
Expand All @@ -11,43 +12,14 @@ export const deriveIdAroundPaginationFlags = <
queryShape,
requestedPageSize,
}: CursorDeriveContext<T, Q>): 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,
});
};
61 changes: 43 additions & 18 deletions src/pagination/paginators/BasePaginator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1968,9 +1968,12 @@ export abstract class BasePaginator<T, Q> {
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;

Expand All @@ -1982,7 +1985,7 @@ export abstract class BasePaginator<T, Q> {
this.getItemId(item1) === this.getItemId(item2),
// items in intervals are not sorted by effectiveComparator
compare: this.itemOrderComparator.bind(this),
plateauScan: true,
plateauScan,
});
}

Expand All @@ -2008,22 +2011,21 @@ export abstract class BasePaginator<T, Q> {
}

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;
}

/**
Expand Down Expand Up @@ -3064,6 +3066,11 @@ export abstract class BasePaginator<T, Q> {
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<PaginatorState<T>>,

Expand Down Expand Up @@ -3223,7 +3230,7 @@ export abstract class BasePaginator<T, Q> {
isFirstPage,
keepPreviousItems,
queryShape,
requestedPageSize: this.pageSize,
requestedPageSize: this.getRequestedPageSize(queryShape),
results,
updateState,
});
Expand Down Expand Up @@ -3332,6 +3339,21 @@ export abstract class BasePaginator<T, Q> {
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,
Expand Down Expand Up @@ -3408,7 +3430,10 @@ export abstract class BasePaginator<T, Q> {
}

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 });

Expand Down
Loading
Loading