fix(app): derive unread from an ack timestamp floor (#2717)

This commit is contained in:
Hampus authored and GitHub committed 2026-09-12 01:40:36 +02:00
1 parent e1bc6c2f7e
commit 6d600990fe
19 files changed
+307 -460

No files matched your search

@@ -52,7 +52,6 @@ interface LoadCompleteOptions {
hasMoreBefore?: boolean;
hasMoreAfter?: boolean;
cached?: boolean;
tailProbe?: boolean;
}
type MessageInput = Message | WireMessage;
@@ -300,7 +299,6 @@ export class ChannelMessages {
jumpDestinationOffset = 0;
jumpTicket = 1;
loadGeneration = 0;
probeLoading = false;
hasJumped = false;
landedAtLiveEdge = false;
jumpHighlight = true;
@@ -835,15 +833,6 @@ 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,
@@ -868,7 +857,6 @@ 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({
@@ -887,26 +875,24 @@ export class ChannelMessages {
}
} else {
next = this.merge(records, loadDecision.prepend, true, true);
if (!tailProbe && loadDecision.trimBottom) {
if (loadDecision.trimBottom) {
next = next.trimToWindow(true, false);
} else if (!tailProbe && loadDecision.trimTop) {
} else if (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,
};
const jumpPatch = {
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,
};
const reachesLiveEdge = selectChannelMessagesLoadRestoresTrust({
mode: loadDecision.mode,
isAfter,
@@ -915,7 +901,6 @@ export class ChannelMessages {
next = next.cloneAnd({
ready: true,
loadingMore: false,
probeLoading: false,
...jumpPatch,
hasMoreBefore: loadDecision.preserveHasMoreBefore ? next.hasMoreBefore : hasMoreBefore,
hasMoreAfter: loadDecision.preserveHasMoreAfter ? next.hasMoreAfter : hasMoreAfter,
@@ -996,7 +981,6 @@ export class ChannelMessages {
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;
@@ -1020,7 +1004,6 @@ export class ChannelMessages {
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;
@@ -13,9 +13,6 @@ import {
selectChannelMessagesLoadDecision,
selectChannelMessagesLoadRestoresTrust,
selectChannelMessagesSpacerHeight,
selectChannelMessagesTailGapId,
selectChannelMessagesTailProbeId,
selectChannelMessagesTailProbeOutcome,
selectChannelMessagesWindowBar,
selectChannelMessagesWindowStatus,
transitionChannelMessagesLoadSnapshot,
@@ -394,132 +391,6 @@ describe('selectChannelMessagesFillerVisible', () => {
});
});
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');
});
});
describe('selectChannelMessagesLoadRestoresTrust', () => {
it('trusts a window a replacement load rebuilt from scratch', () => {
expect(selectChannelMessagesLoadRestoresTrust({mode: 'replace', isAfter: false, hasMoreAfter: false})).toBe(true);
@@ -1,6 +1,5 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {compare as compareSnowflakes} from '@fluxer/snowflake/src/SnowflakeUtils';
import {assign, initialTransition, type SnapshotFrom, setup, transition} from 'xstate';
export interface ChannelMessagesLoadInput {
@@ -282,25 +281,6 @@ 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 function selectChannelMessagesLoadRestoresTrust(input: {
mode: ChannelMessagesLoadMode;
isAfter: boolean;
@@ -310,28 +290,6 @@ export function selectChannelMessagesLoadRestoresTrust(input: {
return input.isAfter && !input.hasMoreAfter;
}
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';
@@ -23,6 +23,9 @@ import type {GuildMemberData} from '@fluxer/schema/src/domains/guild/GuildMember
import type {Message as WireMessage} from '@fluxer/schema/src/domains/message/MessageResponseSchemas';
import {action, makeAutoObservable, reaction} from 'mobx';
const STALE_WINDOW_REFETCH_INTERVAL_MS = 10_000;
const staleWindowRefetchedAt = new Map<string, number>();
interface GuildMemberUpdateAction {
type: 'GUILD_MEMBER_UPDATE';
guildId: string;
@@ -379,10 +382,9 @@ class Messages {
if (!isNonGuildChannel && !guildExists) {
return false;
}
const distrustedTailId = messages.ready && messages.cached ? (messages.last()?.id ?? null) : null;
this.commitMessages(messages.withPatch({loadingMore: true}));
this.notifyChange();
MessageCommands.fetchMessages(channelId, null, distrustedTailId, MAX_MESSAGES_PER_CHANNEL);
MessageCommands.fetchMessages(channelId, null, null, MAX_MESSAGES_PER_CHANNEL);
return false;
}
@@ -443,9 +445,9 @@ class Messages {
}
@action
handleLoadMessages(action: {channelId: string; jump?: JumpOptions; tailProbe?: boolean}): boolean {
handleLoadMessages(action: {channelId: string; jump?: JumpOptions}): boolean {
const messages = ChannelMessages.getOrCreate(action.channelId);
this.commitMessages(action.tailProbe ? messages.beginProbeLoad() : messages.beginLoad(action.jump));
this.commitMessages(messages.beginLoad(action.jump));
this.notifyChange();
return false;
}
@@ -499,7 +501,6 @@ class Messages {
hasMoreBefore?: boolean;
hasMoreAfter?: boolean;
cached?: boolean;
tailProbe?: boolean;
messages: Array<WireMessage>;
}): boolean {
const messages = ChannelMessages.getOrCreate(action.channelId).applyLoadedWindow({
@@ -510,13 +511,26 @@ class Messages {
hasMoreBefore: action.hasMoreBefore,
hasMoreAfter: action.hasMoreAfter,
cached: action.cached,
tailProbe: action.tailProbe,
});
this.commitMessages(messages);
this.notifyChange();
this.refetchStaleWindow(action.channelId, action.cached === true, action.jump);
return false;
}
private refetchStaleWindow(channelId: string, stale: boolean, jump?: JumpOptions): void {
if (!stale || !GatewayConnection.isConnected || SelectedChannel.currentChannelId !== channelId) {
return;
}
const lastAttemptAt = staleWindowRefetchedAt.get(channelId) ?? 0;
const now = Date.now();
if (now - lastAttemptAt < STALE_WINDOW_REFETCH_INTERVAL_MS) {
return;
}
staleWindowRefetchedAt.set(channelId, now);
MessageCommands.fetchMessages(channelId, null, null, MAX_MESSAGES_PER_CHANNEL, jump, {staleRefetch: true});
}
@action
handleLoadMessagesFailure(action: {channelId: string}): boolean {
const messages = ChannelMessages.getOrCreate(action.channelId);
@@ -525,15 +539,6 @@ 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);