fix(app): refetch the tail when a channel window falls behind (#2565)

fluxer_app/src/features/messaging/state/ChannelMessages.ts

@@ -5,6 +5,7 @@ import {UploadingAttachment} from '@app/features/messaging/models/UploadingAttac
 import {resolveChannelIncomingMessageDecision} from '@app/features/messaging/state/ChannelIncomingMessageStateMachine';
 import {resolveChannelMessagesLoadDecision} from '@app/features/messaging/state/ChannelMessagesLoadStateMachine';
 import MessageReactions from '@app/features/messaging/state/MessageReactions';
+import {mergeAscendingById} from '@app/features/messaging/utils/MessagePaginationUtils';
 import SelectedChannel from '@app/features/navigation/state/SelectedChannel';
 import type {JumpType} from '@fluxer/constants/src/JumpConstants';
 import {JumpTypes} from '@fluxer/constants/src/JumpConstants';
@@ -48,6 +49,7 @@ interface LoadCompleteOptions {
 	hasMoreBefore?: boolean;
 	hasMoreAfter?: boolean;
 	cached?: boolean;
+	tailProbe?: boolean;
 }
 
 type MessageInput = Message | WireMessage;
@@ -281,6 +283,8 @@ class MessageBufferSegment {
 	}
 }
 
+let nextLoadGeneration = 0;
+
 export class ChannelMessages {
 	private static readonly channelCache = new Map<string, ChannelMessages>();
 	private static readonly maxChannelsInMemory = 50;
@@ -292,6 +296,8 @@ export class ChannelMessages {
 	jumpDestinationId: string | null = null;
 	jumpDestinationOffset = 0;
 	jumpTicket = 1;
+	loadGeneration = 0;
+	probeLoading = false;
 	hasJumped = false;
 	landedAtLiveEdge = false;
 	jumpHighlight = true;
@@ -605,9 +611,9 @@ export class ChannelMessages {
 		}, true);
 	}
 
-	merge(records: Array<Message>, prepend = false, clearBuffer = false): ChannelMessages {
+	merge(records: Array<Message>, prepend = false, clearBuffer = false, ordered = false): ChannelMessages {
 		return this.cloneAnd((draft) => {
-			draft.mergeInto(records, prepend, clearBuffer);
+			draft.mergeInto(records, prepend, clearBuffer, ordered);
 		}, true);
 	}
 
@@ -822,8 +828,18 @@ export class ChannelMessages {
 		return this.cloneAnd({ready: true, cached: true}).merge([hydrateMessage(this, wire, 'preserve')]);
 	}
 
+	beginProbeLoad(): ChannelMessages {
+		return this.cloneAnd({loadGeneration: ++nextLoadGeneration, probeLoading: true});
+	}
+
+	endProbeLoad(): ChannelMessages {
+		if (!this.probeLoading) return this;
+		return this.cloneAnd({probeLoading: false});
+	}
+
 	beginLoad(jump?: JumpOptions): ChannelMessages {
 		return this.cloneAnd({
+			loadGeneration: ++nextLoadGeneration,
 			loadingMore: true,
 			hasJumped: jump != null,
 			landedAtLiveEdge: jump?.present ?? false,
@@ -845,6 +861,7 @@ export class ChannelMessages {
 			hasMoreBefore = false,
 			hasMoreAfter = false,
 			cached = false,
+			tailProbe = false,
 		} = options;
 		const records = [...windowMessages].reverse().map((m) => hydrateMessage(this, m, 'empty'));
 		const loadDecision = resolveChannelMessagesLoadDecision({
@@ -862,26 +879,32 @@ export class ChannelMessages {
 				next = next.merge(unsent);
 			}
 		} else {
-			next = this.merge(records, loadDecision.prepend, true);
-			if (loadDecision.trimBottom) {
+			next = this.merge(records, loadDecision.prepend, true, true);
+			if (!tailProbe && loadDecision.trimBottom) {
 				next = next.trimToWindow(true, false);
-			} else if (loadDecision.trimTop) {
+			} else if (!tailProbe && loadDecision.trimTop) {
 				next = next.trimToWindow(false, true);
 			}
 		}
+		const jumpPatch = tailProbe
+			? {}
+			: {
+					jumpType: jump?.jumpType ?? JumpTypes.ANIMATED,
+					jumpHighlight: jump?.flash ?? false,
+					hasJumped: jump != null,
+					landedAtLiveEdge: jump?.present ?? false,
+					jumpDestinationId: jump?.messageId ?? null,
+					jumpDestinationOffset: jump && jump.messageId != null && jump.offset != null ? jump.offset : 0,
+					jumpTicket: jump ? next.jumpTicket + 1 : next.jumpTicket,
+					jumpReturnMessageId: jump?.returnToMessageId ?? null,
+					jumpReturnChannelId: jump?.returnToMessageId ? (jump.returnChannelId ?? this.channelId) : null,
+					jumpReturnGuildId: jump?.returnToMessageId ? (jump.returnGuildId ?? null) : null,
+				};
 		next = next.cloneAnd({
 			ready: true,
 			loadingMore: false,
-			jumpType: jump?.jumpType ?? JumpTypes.ANIMATED,
-			jumpHighlight: jump?.flash ?? false,
-			hasJumped: jump != null,
-			landedAtLiveEdge: jump?.present ?? false,
-			jumpDestinationId: jump?.messageId ?? null,
-			jumpDestinationOffset: jump && jump.messageId != null && jump.offset != null ? jump.offset : 0,
-			jumpTicket: jump ? next.jumpTicket + 1 : next.jumpTicket,
-			jumpReturnMessageId: jump?.returnToMessageId ?? null,
-			jumpReturnChannelId: jump?.returnToMessageId ? (jump.returnChannelId ?? this.channelId) : null,
-			jumpReturnGuildId: jump?.returnToMessageId ? (jump.returnGuildId ?? null) : null,
+			probeLoading: false,
+			...jumpPatch,
 			hasMoreBefore: loadDecision.preserveHasMoreBefore ? next.hasMoreBefore : hasMoreBefore,
 			hasMoreAfter: loadDecision.preserveHasMoreAfter ? next.hasMoreAfter : hasMoreAfter,
 			cached,
@@ -914,7 +937,7 @@ export class ChannelMessages {
 		this.messageIndex = {};
 	}
 
-	private mergeInto(incoming: Array<Message>, prepend = false, clearSideBuffer = false): void {
+	private mergeInto(incoming: Array<Message>, prepend = false, clearSideBuffer = false, ordered = false): void {
 		const newItems: Array<Message> = [];
 		for (const msg of incoming) {
 			const existing = this.messageIndex[msg.id];
@@ -937,7 +960,11 @@ export class ChannelMessages {
 			buffer.clear();
 		}
 		if (newItems.length === 0) return;
-		this.messageList = prepend ? newItems.concat(this.messageList) : this.messageList.concat(newItems);
+		if (prepend) {
+			this.messageList = newItems.concat(this.messageList);
+			return;
+		}
+		this.messageList = ordered ? mergeAscendingById(this.messageList, newItems) : this.messageList.concat(newItems);
 	}
 
 	private cloneAnd(
@@ -956,6 +983,8 @@ export class ChannelMessages {
 			clone.jumpDestinationId = this.jumpDestinationId;
 			clone.jumpDestinationOffset = this.jumpDestinationOffset;
 			clone.jumpTicket = this.jumpTicket;
+			clone.loadGeneration = this.loadGeneration;
+			clone.probeLoading = this.probeLoading;
 			clone.hasJumped = this.hasJumped;
 			clone.landedAtLiveEdge = this.landedAtLiveEdge;
 			clone.jumpHighlight = this.jumpHighlight;
@@ -978,6 +1007,8 @@ export class ChannelMessages {
 			clone.jumpDestinationOffset =
 				patch.jumpDestinationOffset !== undefined ? patch.jumpDestinationOffset : this.jumpDestinationOffset;
 			clone.jumpTicket = patch.jumpTicket !== undefined ? patch.jumpTicket : this.jumpTicket;
+			clone.loadGeneration = patch.loadGeneration !== undefined ? patch.loadGeneration : this.loadGeneration;
+			clone.probeLoading = 'probeLoading' in patch ? !!patch.probeLoading : this.probeLoading;
 			clone.hasJumped = 'hasJumped' in patch ? !!patch.hasJumped : this.hasJumped;
 			clone.landedAtLiveEdge = 'landedAtLiveEdge' in patch ? !!patch.landedAtLiveEdge : this.landedAtLiveEdge;
 			clone.jumpHighlight = 'jumpHighlight' in patch ? !!patch.jumpHighlight : this.jumpHighlight;

fluxer_app/src/features/messaging/state/ChannelMessagesLoadStateMachine.test.ts

@@ -12,6 +12,9 @@ import {
 	selectChannelMessagesFillerVisible,
 	selectChannelMessagesLoadDecision,
 	selectChannelMessagesSpacerHeight,
+	selectChannelMessagesTailGapId,
+	selectChannelMessagesTailProbeId,
+	selectChannelMessagesTailProbeOutcome,
 	selectChannelMessagesWindowBar,
 	selectChannelMessagesWindowStatus,
 	transitionChannelMessagesLoadSnapshot,
@@ -389,3 +392,129 @@ describe('selectChannelMessagesFillerVisible', () => {
 		expect(streamCases).toBeGreaterThan(0);
 	});
 });
+
+describe('selectChannelMessagesTailProbeId', () => {
+	const ID = {older: '1519773906704011264', newer: '1519773906708205568'};
+
+	function tailInput(
+		overrides: Partial<ChannelMessagesWindowInput> = {},
+		tail: {
+			loading?: boolean;
+			newestLoadedMessageId?: string | null;
+			knownLatestMessageId?: string | null;
+		} = {},
+	) {
+		const windowInput: ChannelMessagesWindowInput = {
+			ready: true,
+			loading: false,
+			failed: false,
+			messageCount: 12,
+			hasMoreBefore: true,
+			hasMoreAfter: false,
+			...overrides,
+		};
+		return {
+			status: resolveChannelMessagesWindowStatus(windowInput),
+			loading: tail.loading ?? windowInput.loading,
+			newestLoadedMessageId: tail.newestLoadedMessageId === undefined ? ID.older : tail.newestLoadedMessageId,
+			knownLatestMessageId: tail.knownLatestMessageId === undefined ? ID.newer : tail.knownLatestMessageId,
+		};
+	}
+
+	it('probes from the newest loaded message when the live edge is known to be ahead', () => {
+		expect(selectChannelMessagesTailProbeId(tailInput())).toBe(ID.older);
+	});
+
+	it('stays silent for a window that already advertises a newer page', () => {
+		expect(selectChannelMessagesTailProbeId(tailInput({hasMoreAfter: true}))).toBeNull();
+	});
+
+	it('stays silent for a failed window so a failed probe cannot re-arm itself', () => {
+		expect(selectChannelMessagesTailProbeId(tailInput({failed: true}))).toBeNull();
+	});
+
+	it('stays silent while a load is in flight', () => {
+		expect(selectChannelMessagesTailProbeId(tailInput({loading: true}))).toBeNull();
+	});
+
+	it('stays silent for a window that has not loaded a page yet', () => {
+		expect(selectChannelMessagesTailProbeId(tailInput({ready: false}))).toBeNull();
+	});
+
+	it('never probes an empty window, so a channel with no readable history cannot arm it', () => {
+		expect(
+			selectChannelMessagesTailProbeId(
+				tailInput({messageCount: 0, hasMoreBefore: false}, {newestLoadedMessageId: null}),
+			),
+		).toBeNull();
+	});
+
+	it('stays silent when the watermark matches or trails the newest loaded message', () => {
+		expect(selectChannelMessagesTailProbeId(tailInput({}, {knownLatestMessageId: ID.older}))).toBeNull();
+		expect(
+			selectChannelMessagesTailProbeId(
+				tailInput({}, {newestLoadedMessageId: ID.newer, knownLatestMessageId: ID.older}),
+			),
+		).toBeNull();
+		expect(selectChannelMessagesTailProbeId(tailInput({}, {knownLatestMessageId: null}))).toBeNull();
+	});
+
+	it('keeps reporting the tail gap while the probe it armed is in flight', () => {
+		const inFlight = tailInput({loading: true});
+		expect(selectChannelMessagesTailProbeId(inFlight)).toBeNull();
+		expect(selectChannelMessagesTailGapId(inFlight)).toBe(ID.older);
+	});
+
+	it('reports no tail gap once the newest loaded message reaches the watermark', () => {
+		expect(selectChannelMessagesTailGapId(tailInput({}, {newestLoadedMessageId: ID.newer}))).toBeNull();
+	});
+
+	it('still reports the gap on a deep window so auto-ack cannot run past it', () => {
+		expect(selectChannelMessagesTailGapId(tailInput({messageCount: 200}))).toBe(ID.older);
+		expect(selectChannelMessagesTailProbeId(tailInput({messageCount: 200}))).toBe(ID.older);
+	});
+});
+
+describe('selectChannelMessagesTailProbeOutcome', () => {
+	const ID = {older: '1519773906704011264', newer: '1519773906708205568'};
+
+	function outcomeInput(overrides: Partial<Parameters<typeof selectChannelMessagesTailProbeOutcome>[0]> = {}) {
+		return {
+			probeGeneration: 4,
+			currentGeneration: 4,
+			probeJumpTicket: 2,
+			currentJumpTicket: 2,
+			ready: true,
+			hasMoreAfter: false,
+			anchorMessageId: ID.older,
+			newestLoadedMessageId: ID.older,
+			...overrides,
+		};
+	}
+
+	it('applies a probe onto the window it was issued against', () => {
+		expect(selectChannelMessagesTailProbeOutcome(outcomeInput())).toBe('apply');
+	});
+
+	it('discards without touching shared state once another load owns the window', () => {
+		expect(selectChannelMessagesTailProbeOutcome(outcomeInput({currentGeneration: 5}))).toBe('discard');
+	});
+
+	it('releases a probe whose window a jump has taken out of the ready state', () => {
+		expect(selectChannelMessagesTailProbeOutcome(outcomeInput({ready: false}))).toBe('release');
+		expect(selectChannelMessagesTailProbeOutcome(outcomeInput({currentGeneration: 9, ready: false}))).toBe('discard');
+	});
+
+	it('releases a probe whose tail moved under it', () => {
+		expect(selectChannelMessagesTailProbeOutcome(outcomeInput({newestLoadedMessageId: ID.newer}))).toBe('release');
+		expect(selectChannelMessagesTailProbeOutcome(outcomeInput({newestLoadedMessageId: null}))).toBe('release');
+	});
+
+	it('releases a probe whose window started advertising a newer page', () => {
+		expect(selectChannelMessagesTailProbeOutcome(outcomeInput({hasMoreAfter: true}))).toBe('release');
+	});
+
+	it('discards a probe whose window a cached jump took over without starting a load', () => {
+		expect(selectChannelMessagesTailProbeOutcome(outcomeInput({currentJumpTicket: 3}))).toBe('discard');
+	});
+});

fluxer_app/src/features/messaging/state/ChannelMessagesLoadStateMachine.ts

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+import {compare as compareSnowflakes} from '@fluxer/snowflake/src/SnowflakeUtils';
 import {assign, getInitialSnapshot, type SnapshotFrom, setup, transition} from 'xstate';
 
 export interface ChannelMessagesLoadInput {
@@ -281,6 +282,47 @@ export function selectChannelMessagesSpacerHeight(status: ChannelMessagesWindowS
 	return status.olderPageAvailable || status.newerPageAvailable ? fillerHeight : 0;
 }
 
+export interface ChannelMessagesTailInput {
+	status: ChannelMessagesWindowStatus;
+	loading: boolean;
+	newestLoadedMessageId: string | null;
+	knownLatestMessageId: string | null;
+}
+
+export function selectChannelMessagesTailGapId(input: ChannelMessagesTailInput): string | null {
+	if (input.status.phase !== 'stream' || input.status.retryVisible || input.status.newerPageAvailable) return null;
+	if (input.newestLoadedMessageId == null || input.knownLatestMessageId == null) return null;
+	if (compareSnowflakes(input.knownLatestMessageId, input.newestLoadedMessageId) <= 0) return null;
+	return input.newestLoadedMessageId;
+}
+
+export function selectChannelMessagesTailProbeId(input: ChannelMessagesTailInput): string | null {
+	if (input.loading) return null;
+	return selectChannelMessagesTailGapId(input);
+}
+
+export interface ChannelMessagesTailProbeResultInput {
+	probeGeneration: number;
+	currentGeneration: number;
+	probeJumpTicket: number;
+	currentJumpTicket: number;
+	ready: boolean;
+	hasMoreAfter: boolean;
+	anchorMessageId: string | null;
+	newestLoadedMessageId: string | null;
+}
+
+export type ChannelMessagesTailProbeOutcome = 'apply' | 'release' | 'discard';
+
+export function selectChannelMessagesTailProbeOutcome(
+	input: ChannelMessagesTailProbeResultInput,
+): ChannelMessagesTailProbeOutcome {
+	if (input.currentGeneration !== input.probeGeneration) return 'discard';
+	if (input.currentJumpTicket !== input.probeJumpTicket) return 'discard';
+	if (!input.ready || input.hasMoreAfter || input.newestLoadedMessageId !== input.anchorMessageId) return 'release';
+	return 'apply';
+}
+
 export function selectChannelMessagesWindowBar(status: ChannelMessagesWindowStatus): ChannelMessagesWindowBar {
 	if (status.phase === 'placeholder') return 'none';
 	if (status.phase === 'retry' || status.retryVisible) return 'retry';

fluxer_app/src/features/messaging/state/MessagingMessages.ts

@@ -449,9 +449,9 @@ class Messages {
 	}
 
 	@action
-	handleLoadMessages(action: {channelId: string; jump?: JumpOptions}): boolean {
+	handleLoadMessages(action: {channelId: string; jump?: JumpOptions; tailProbe?: boolean}): boolean {
 		const messages = ChannelMessages.getOrCreate(action.channelId);
-		this.commitMessages(messages.beginLoad(action.jump));
+		this.commitMessages(action.tailProbe ? messages.beginProbeLoad() : messages.beginLoad(action.jump));
 		this.notifyChange();
 		return false;
 	}
@@ -505,6 +505,7 @@ class Messages {
 		hasMoreBefore?: boolean;
 		hasMoreAfter?: boolean;
 		cached?: boolean;
+		tailProbe?: boolean;
 		messages: Array<WireMessage>;
 	}): boolean {
 		const messages = ChannelMessages.getOrCreate(action.channelId).applyLoadedWindow({
@@ -515,6 +516,7 @@ class Messages {
 			hasMoreBefore: action.hasMoreBefore,
 			hasMoreAfter: action.hasMoreAfter,
 			cached: action.cached,
+			tailProbe: action.tailProbe,
 		});
 		this.commitMessages(messages);
 		this.notifyChange();
@@ -529,6 +531,15 @@ class Messages {
 		return false;
 	}
 
+	@action
+	handleTailProbeSettled(action: {channelId: string}): boolean {
+		const messages = ChannelMessages.get(action.channelId);
+		if (!messages?.probeLoading) return false;
+		this.commitMessages(messages.endProbeLoad());
+		this.notifyChange();
+		return true;
+	}
+
 	@action
 	handleLoadMessagesBlocked(action: {channelId: string}): boolean {
 		const messages = ChannelMessages.getOrCreate(action.channelId);