[codex] Fix bot activity and mobile latest messages (#431)

This commit is contained in:
Wes
2026-04-29 19:25:09 -06:00
committed by GitHub
parent 028f53194e
commit 127829e0f2
8 changed files with 424 additions and 121 deletions

View File

@@ -7,6 +7,7 @@ import { mergeTimelineCacheMessages } from "@/features/messages/hooks";
import { channelMessagesKey } from "@/features/messages/lib/messageQueryKeys";
import { getChannelIdFromTags } from "@/features/messages/lib/threading";
import { relayClient } from "@/shared/api/relayClient";
import { CHANNEL_EVENT_KINDS } from "@/shared/constants/kinds";
import type { Channel, RelayEvent } from "@/shared/api/types";
export type UseLiveChannelUpdatesOptions = {
@@ -15,8 +16,8 @@ export type UseLiveChannelUpdatesOptions = {
onLiveMention?: () => void;
};
const LIVE_MENTION_SUBSCRIPTION_RETRY_BASE_MS = 1_000;
const LIVE_MENTION_SUBSCRIPTION_RETRY_MAX_MS = 30_000;
const LIVE_SUBSCRIPTION_RETRY_BASE_MS = 1_000;
const LIVE_SUBSCRIPTION_RETRY_MAX_MS = 30_000;
function getMessageTimestamp(event: RelayEvent) {
return new Date(event.created_at * 1_000).toISOString();
@@ -78,8 +79,7 @@ export function useLiveChannelUpdates(
// Effect deps use primitive keys so refetches that produce new refs with
// identical contents don't churn subscriptions. The Set/array memos are
// still handy for closure reads via useEffectEvent.
const hasLiveChannels = liveChannelIds.size > 0;
const mentionChannelIdsKey = React.useMemo(
const channelIdsKey = React.useMemo(
() => [...new Set(channels.map((channel) => channel.id))].sort().join(","),
[channels],
);
@@ -176,43 +176,89 @@ export function useLiveChannelUpdates(
});
}, [queryClient]);
const liveSubsRef = React.useRef(new Map<string, () => Promise<void>>());
React.useEffect(() => {
if (!hasLiveChannels) {
return;
}
let isCancelled = false;
let retryTimeout: number | undefined;
let retryAttempt = 0;
let isDisposed = false;
let cleanup: (() => Promise<void>) | undefined;
const syncSubs = async (): Promise<boolean> => {
const activeSubs = liveSubsRef.current;
const targetIds = new Set(channelIdsKey ? channelIdsKey.split(",") : []);
// Record the subscription start time so handleDmEvent can distinguish
// backlog replays (created_at < startedAt) from live messages.
dmSubscriptionStartedAtRef.current = Math.floor(Date.now() / 1000);
relayClient
.subscribeToAllStreamMessages((event) => {
if (!isDisposed) {
handleIncomingMessage(event);
}
})
.then((dispose) => {
if (isDisposed) {
void dispose();
return;
for (const [channelId, dispose] of activeSubs) {
if (!targetIds.has(channelId)) {
activeSubs.delete(channelId);
void dispose().catch(() => {});
}
}
cleanup = dispose;
})
.catch((error) => {
console.error("Failed to subscribe to unread channel updates", error);
});
if (targetIds.size > 0) {
// Record the subscription start time so handleDmEvent can distinguish
// backlog replays (created_at < startedAt) from live messages.
dmSubscriptionStartedAtRef.current = Math.floor(Date.now() / 1000);
}
let anyFailed = false;
const additions = Array.from(targetIds)
.filter((channelId) => !activeSubs.has(channelId))
.map(async (channelId) => {
try {
const dispose = await relayClient.subscribeLive(
{
kinds: [...CHANNEL_EVENT_KINDS],
"#h": [channelId],
limit: 1000,
since: Math.floor(Date.now() / 1_000),
},
handleIncomingMessage,
);
if (isCancelled) {
void dispose().catch(() => {});
return;
}
activeSubs.set(channelId, dispose);
} catch (err) {
anyFailed = true;
console.error(
"Failed to subscribe to live channel updates",
channelId,
err,
);
}
});
await Promise.allSettled(additions);
return !anyFailed;
};
const runSync = async () => {
const ok = await syncSubs();
if (isCancelled) return;
if (ok) {
retryAttempt = 0;
return;
}
const delayMs = Math.min(
LIVE_SUBSCRIPTION_RETRY_BASE_MS * 2 ** retryAttempt,
LIVE_SUBSCRIPTION_RETRY_MAX_MS,
);
retryAttempt += 1;
retryTimeout = window.setTimeout(() => {
retryTimeout = undefined;
void runSync();
}, delayMs);
};
void runSync();
return () => {
isDisposed = true;
if (cleanup) {
void cleanup();
isCancelled = true;
if (retryTimeout !== undefined) {
window.clearTimeout(retryTimeout);
}
};
}, [hasLiveChannels]);
}, [channelIdsKey]);
// Subscribe to mention events per channel with a diff-based manager: only
// subscribe newly-added channels and unsubscribe removed ones on each sync.
@@ -243,9 +289,7 @@ export function useLiveChannelUpdates(
}
mentionSubsPubkeyRef.current = normalizedCurrentPubkey;
const targetIds = new Set(
mentionChannelIdsKey ? mentionChannelIdsKey.split(",") : [],
);
const targetIds = new Set(channelIdsKey ? channelIdsKey.split(",") : []);
for (const [channelId, dispose] of activeSubs) {
if (!targetIds.has(channelId)) {
@@ -295,8 +339,8 @@ export function useLiveChannelUpdates(
return;
}
const delayMs = Math.min(
LIVE_MENTION_SUBSCRIPTION_RETRY_BASE_MS * 2 ** retryAttempt,
LIVE_MENTION_SUBSCRIPTION_RETRY_MAX_MS,
LIVE_SUBSCRIPTION_RETRY_BASE_MS * 2 ** retryAttempt,
LIVE_SUBSCRIPTION_RETRY_MAX_MS,
);
retryAttempt += 1;
retryTimeout = window.setTimeout(() => {
@@ -313,10 +357,15 @@ export function useLiveChannelUpdates(
window.clearTimeout(retryTimeout);
}
};
}, [mentionChannelIdsKey, normalizedCurrentPubkey, options.onLiveMention]);
}, [channelIdsKey, normalizedCurrentPubkey, options.onLiveMention]);
React.useEffect(() => {
return () => {
for (const dispose of liveSubsRef.current.values()) {
void dispose().catch(() => {});
}
liveSubsRef.current.clear();
const subs = mentionSubsRef.current;
for (const dispose of subs.values()) {
void dispose().catch(() => {});

View File

@@ -0,0 +1,58 @@
import assert from "node:assert/strict";
import test from "node:test";
import { buildMainTimelineEntries } from "./threadPanel.ts";
function message(overrides) {
return {
id: "message",
createdAt: 1,
pubkey: "author",
author: "Author",
avatarUrl: null,
role: undefined,
personaDisplayName: undefined,
time: "12:00 PM",
body: "body",
parentId: null,
rootId: null,
depth: 0,
accent: false,
pending: undefined,
edited: false,
kind: 9,
tags: [],
reactions: undefined,
...overrides,
};
}
test("buildMainTimelineEntries includes broadcast replies", () => {
const root = message({ id: "root", createdAt: 1 });
const hiddenReply = message({
id: "hidden-reply",
createdAt: 2,
parentId: "root",
rootId: "root",
depth: 1,
tags: [["e", "root", "", "reply"]],
});
const broadcastReply = message({
id: "broadcast-reply",
createdAt: 3,
parentId: "root",
rootId: "root",
depth: 1,
tags: [
["e", "root", "", "reply"],
["broadcast", "1"],
],
});
assert.deepEqual(
buildMainTimelineEntries([root, hiddenReply, broadcastReply]).map(
(entry) => entry.message.id,
),
["root", "broadcast-reply"],
);
});

View File

@@ -31,6 +31,13 @@ type ThreadDescendantStats = {
const MAX_SUMMARY_PARTICIPANTS = 3;
function isBroadcastReply(message: TimelineMessage): boolean {
return (
message.tags?.some((tag) => tag[0] === "broadcast" && tag[1] === "1") ??
false
);
}
function normalizeHeadMessage(message: TimelineMessage): TimelineMessage {
return {
...message,
@@ -215,7 +222,7 @@ export function buildMainTimelineEntries(
const descendantStatsByMessageId = buildDescendantStatsByMessageId(messages);
return messages
.filter((message) => message.parentId == null)
.filter((message) => message.parentId == null || isBroadcastReply(message))
.map((message) => {
return {
message,

View File

@@ -11,6 +11,7 @@ import {
KIND_STREAM_MESSAGE_DIFF,
KIND_TYPING_INDICATOR,
} from "@/shared/constants/kinds";
import { resolveEventAuthorPubkey } from "@/shared/lib/authors";
export type TypingIndicatorEntry = {
pubkey: string;
@@ -143,7 +144,12 @@ export function useChannelTyping(
return;
}
const authorPubkey = latestMessageEvent.pubkey.toLowerCase();
const authorPubkey = resolveEventAuthorPubkey({
pubkey: latestMessageEvent.pubkey,
tags: latestMessageEvent.tags,
preferActorTag: true,
requireChannelTagForPTags: true,
}).toLowerCase();
const threadHeadId = getTypingScopeId(latestMessageEvent);
const typingKey = getTypingStateKey(authorPubkey, threadHeadId);
latestMessageCreatedAtByPubkeyRef.current[typingKey] = Math.max(

View File

@@ -311,20 +311,51 @@ class _MessageList extends HookConsumerWidget {
required this.isArchived,
});
static const _fetchOlderThreshold = 200.0;
static const _latestThreshold = 48.0;
@override
Widget build(BuildContext context, WidgetRef ref) {
// Pagination: fetch older messages when scrolling near the top.
final scrollController = useScrollController();
final isLoadingOlder = useState(false);
final isAtLatest = useState(true);
final latestEntryId = entries.isEmpty ? null : entries.last.message.id;
final previousLatestEntryId = useRef<String?>(null);
bool nearLatest() {
if (!scrollController.hasClients) return true;
return scrollController.position.pixels <= _latestThreshold;
}
void updateLatestState() {
final next = nearLatest();
if (isAtLatest.value != next) {
isAtLatest.value = next;
}
}
Future<void> scrollToLatest() async {
if (!scrollController.hasClients) return;
await scrollController.animateTo(
0,
duration: const Duration(milliseconds: 220),
curve: Curves.easeOutCubic,
);
if (context.mounted) {
isAtLatest.value = true;
}
}
useEffect(() {
void onScroll() {
updateLatestState();
if (isLoadingOlder.value) return;
final notifier = ref.read(channelMessagesProvider(channelId).notifier);
if (notifier.reachedOldest) return;
// In a reversed ListView, maxScrollExtent is the oldest messages.
final pos = scrollController.position;
if (pos.pixels >= pos.maxScrollExtent - 200) {
if (pos.pixels >= pos.maxScrollExtent - _fetchOlderThreshold) {
isLoadingOlder.value = true;
notifier.fetchOlder().whenComplete(
() => isLoadingOlder.value = false,
@@ -334,7 +365,36 @@ class _MessageList extends HookConsumerWidget {
scrollController.addListener(onScroll);
return () => scrollController.removeListener(onScroll);
}, [scrollController]);
}, [channelId, scrollController]);
useEffect(() {
WidgetsBinding.instance.addPostFrameCallback((_) {
if (!context.mounted) return;
updateLatestState();
});
return null;
}, [entries.length, scrollController]);
useEffect(() {
final previous = previousLatestEntryId.value;
previousLatestEntryId.value = latestEntryId;
if (previous == null ||
latestEntryId == null ||
previous == latestEntryId ||
!isAtLatest.value) {
return null;
}
WidgetsBinding.instance.addPostFrameCallback((_) {
if (!context.mounted || !scrollController.hasClients) return;
scrollController.animateTo(
0,
duration: const Duration(milliseconds: 220),
curve: Curves.easeOutCubic,
);
});
return null;
}, [latestEntryId, scrollController]);
if (entries.isEmpty) {
return Center(
@@ -374,84 +434,111 @@ class _MessageList extends HookConsumerWidget {
}
});
return ListView.builder(
controller: scrollController,
reverse: true,
padding: EdgeInsets.only(
left: Grid.xs,
right: Grid.xs,
top: frostedAppBarHeight(context),
bottom: Grid.xxs,
),
itemCount: entries.length + (isLoadingOlder.value ? 1 : 0),
itemBuilder: (context, index) {
// Loading indicator at the top (last index in reversed list).
if (index >= entries.length) {
return const Padding(
padding: EdgeInsets.symmetric(vertical: Grid.xs),
return Stack(
children: [
ListView.builder(
key: const ValueKey('channel-message-list'),
controller: scrollController,
reverse: true,
padding: EdgeInsets.only(
left: Grid.xs,
right: Grid.xs,
top: frostedAppBarHeight(context),
bottom: Grid.xxs,
),
itemCount: entries.length + (isLoadingOlder.value ? 1 : 0),
itemBuilder: (context, index) {
// Loading indicator at the top (last index in reversed list).
if (index >= entries.length) {
return const Padding(
padding: EdgeInsets.symmetric(vertical: Grid.xs),
child: Center(
child: SizedBox(
width: 20,
height: 20,
child: CircularProgressIndicator(strokeWidth: 2),
),
),
);
}
// Reversed list: index 0 = newest (bottom of screen).
final chronIdx = entries.length - 1 - index;
final entry = entries[chronIdx];
final message = entry.message;
// Day boundary check — applies to all messages including system.
final prevEntry = chronIdx > 0 ? entries[chronIdx - 1] : null;
final prevMessage = prevEntry?.message;
final showDayDivider =
prevMessage == null ||
!isSameDay(prevMessage.createdAt, message.createdAt);
final showAuthor =
!message.isSystem &&
(prevMessage == null ||
prevMessage.isSystem ||
showDayDivider ||
prevMessage.pubkey.toLowerCase() !=
message.pubkey.toLowerCase() ||
(message.createdAt - prevMessage.createdAt) > 300);
return Column(
crossAxisAlignment: CrossAxisAlignment.start,
children: [
if (showDayDivider)
DayDivider(label: formatDayHeading(message.createdAt)),
if (message.isSystem)
_SystemMessageRow(message: message)
else ...[
_MessageBubble(
message: message,
showAuthor: showAuthor,
channelNames: channelNamesMap,
currentChannelId: channelId,
currentPubkey: currentPubkey,
allMessages: allMessages,
isMember: isMember,
isArchived: isArchived,
),
if (entry.summary != null)
_ThreadSummaryRow(
summary: entry.summary!,
message: message,
allMessages: allMessages,
channelId: channelId,
currentPubkey: currentPubkey,
isMember: isMember,
isArchived: isArchived,
),
],
],
);
},
),
if (!isAtLatest.value)
Positioned(
left: 0,
right: 0,
bottom: Grid.xs,
child: Center(
child: SizedBox(
width: 20,
height: 20,
child: CircularProgressIndicator(strokeWidth: 2),
child: FilledButton.icon(
key: const ValueKey('channel-jump-to-latest'),
onPressed: scrollToLatest,
style: FilledButton.styleFrom(
backgroundColor: context.colors.primaryContainer,
foregroundColor: context.colors.onPrimaryContainer,
padding: const EdgeInsets.symmetric(
horizontal: Grid.xs,
vertical: Grid.xxs,
),
),
icon: const Icon(LucideIcons.arrowDown, size: 16),
label: const Text('Latest'),
),
),
);
}
// Reversed list: index 0 = newest (bottom of screen).
final chronIdx = entries.length - 1 - index;
final entry = entries[chronIdx];
final message = entry.message;
// Day boundary check — applies to all messages including system.
final prevEntry = chronIdx > 0 ? entries[chronIdx - 1] : null;
final prevMessage = prevEntry?.message;
final showDayDivider =
prevMessage == null ||
!isSameDay(prevMessage.createdAt, message.createdAt);
final showAuthor =
!message.isSystem &&
(prevMessage == null ||
prevMessage.isSystem ||
showDayDivider ||
prevMessage.pubkey.toLowerCase() !=
message.pubkey.toLowerCase() ||
(message.createdAt - prevMessage.createdAt) > 300);
return Column(
crossAxisAlignment: CrossAxisAlignment.start,
children: [
if (showDayDivider)
DayDivider(label: formatDayHeading(message.createdAt)),
if (message.isSystem)
_SystemMessageRow(message: message)
else ...[
_MessageBubble(
message: message,
showAuthor: showAuthor,
channelNames: channelNamesMap,
currentChannelId: channelId,
currentPubkey: currentPubkey,
allMessages: allMessages,
isMember: isMember,
isArchived: isArchived,
),
if (entry.summary != null)
_ThreadSummaryRow(
summary: entry.summary!,
message: message,
allMessages: allMessages,
channelId: channelId,
currentPubkey: currentPubkey,
isMember: isMember,
isArchived: isArchived,
),
],
],
);
},
),
],
);
}
}

View File

@@ -345,7 +345,7 @@ List<MainTimelineEntry> buildMainTimelineEntries(
return [
for (final msg in messages)
if (msg.parentId == null)
if (msg.parentId == null || _isBroadcastReply(msg))
MainTimelineEntry(
message: msg,
summary: _buildSummary(msg.id, childrenByParent),
@@ -353,6 +353,12 @@ List<MainTimelineEntry> buildMainTimelineEntries(
];
}
bool _isBroadcastReply(TimelineMessage message) {
return message.tags.any(
(tag) => tag.length >= 2 && tag[0] == 'broadcast' && tag[1] == '1',
);
}
ThreadSummary? _buildSummary(
String messageId,
Map<String, List<TimelineMessage>> childrenByParent,

View File

@@ -122,15 +122,18 @@ Widget _buildTestable({
Future<List<ChannelMember>> Function()? loadMembers,
ChannelActions Function(Ref ref)? createChannelActions,
ReadStateNotifier? readStateNotifier,
_FakeMessagesNotifier? messagesNotifier,
}) {
final resolvedChannel = channel ?? _testChannel;
final fakeChannelsNotifier =
channelsNotifier ?? _FakeChannelsNotifier(channels ?? [resolvedChannel]);
final fakeMessagesNotifier =
messagesNotifier ?? _FakeMessagesNotifier(messages);
return ProviderScope(
overrides: [
channelMessagesProvider(
_channelId,
).overrideWith(() => _FakeMessagesNotifier(messages)),
).overrideWith(() => fakeMessagesNotifier),
channelTypingProvider(
_channelId,
).overrideWith(() => _FakeTypingNotifier(typing)),
@@ -421,6 +424,63 @@ void main() {
expect(find.text('Bob'), findsOneWidget);
});
testWidgets('can jump back to latest when newer messages are offscreen', (
tester,
) async {
final initialMessages = [
for (var i = 0; i < 40; i++)
_textMsg(
id: 'msg$i',
pubkey: 'alice',
content: 'Message $i',
createdAt: 1000 + i,
),
];
final messagesNotifier = _FakeMessagesNotifier(initialMessages);
await tester.pumpWidget(
_buildTestable(
messages: const [],
messagesNotifier: messagesNotifier,
users: const {
'alice': UserProfile(pubkey: 'alice', displayName: 'Alice'),
},
),
);
await tester.pumpAndSettle();
final listView = tester.widget<ListView>(
find.byKey(const ValueKey('channel-message-list')),
);
final controller = listView.controller!;
expect(controller.position.maxScrollExtent, greaterThan(0));
controller.jumpTo(controller.position.maxScrollExtent);
await tester.pump();
expect(
find.byKey(const ValueKey('channel-jump-to-latest')),
findsOneWidget,
);
messagesNotifier.setMessages([
...initialMessages,
_textMsg(
id: 'newest',
pubkey: 'alice',
content: 'Newest live update',
createdAt: 2000,
),
]);
await tester.pump();
expect(findRichText('Newest live update'), findsNothing);
await tester.tap(find.byKey(const ValueKey('channel-jump-to-latest')));
await tester.pumpAndSettle();
expect(controller.position.pixels, lessThanOrEqualTo(1));
expect(findRichText('Newest live update'), findsOneWidget);
});
testWidgets('groups consecutive messages from same author', (tester) async {
final messages = [
_textMsg(
@@ -1097,11 +1157,22 @@ void main() {
// ---------------------------------------------------------------------------
class _FakeMessagesNotifier extends ChannelMessagesNotifier {
final List<NostrEvent> _messages;
List<NostrEvent> _messages;
_FakeMessagesNotifier(this._messages) : super(_channelId);
@override
AsyncValue<List<NostrEvent>> build() => AsyncData(_messages);
@override
bool get reachedOldest => true;
@override
Future<bool> fetchOlder() async => false;
void setMessages(List<NostrEvent> messages) {
_messages = messages;
state = AsyncData(messages);
}
}
class _ErrorMessagesNotifier extends ChannelMessagesNotifier {

View File

@@ -35,6 +35,7 @@ NostrEvent _replyMsg({
String pubkey = 'alice',
String content = 'reply',
int createdAt = 2000,
List<List<String>> extraTags = const [],
}) {
final root = rootId ?? parentId;
final eTags = parentId == root
@@ -50,7 +51,7 @@ NostrEvent _replyMsg({
pubkey: pubkey,
content: content,
createdAt: createdAt,
extraTags: eTags,
extraTags: [...eTags, ...extraTags],
);
}
@@ -591,6 +592,24 @@ void main() {
expect(entries[1].message.id, 'b');
});
test('includes broadcast replies in the main timeline', () {
final messages = formatTimeline([
_textMsg(id: 'a', createdAt: 1000),
_replyMsg(id: 'hidden', parentId: 'a', createdAt: 2000),
_replyMsg(
id: 'broadcast',
parentId: 'a',
createdAt: 3000,
extraTags: const [
['broadcast', '1'],
],
),
]);
final entries = buildMainTimelineEntries(messages);
expect(entries.map((entry) => entry.message.id), ['a', 'broadcast']);
});
test('root without replies has null summary', () {
final messages = formatTimeline([_textMsg(id: 'a')]);
final entries = buildMainTimelineEntries(messages);