jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(mobile): write catch-up and visible-message read marks like the web app (#8106)

Browse Source 
🤖
## Summary
When you read a channel or thread on the phone, the web and desktop app
(buzz-app) should show it as read too, and the reverse. buzz-app no
longer saves a "read" mark for each ordinary message. It saves one
"caught up to here" mark per channel and per thread, and keeps
per-message marks only for mentions, DMs, broadcasts and replies. Mobile
already reads those marks. This PR makes mobile write them, so reading
on the phone clears unread on other devices and saved read state stays
small.

What changes on the phone:
- Reaching the bottom of a channel, and staying there for 300 ms, marks
it caught up. Reaching the end of a thread does the same for that
thread.
- A message you see fully on screen, away from the bottom, is marked
read on its own.
- Mentions, DMs, broadcasts and thread replies you see keep their own
read marks, as in buzz-app.
- Automatic reading never marks the whole channel read. Only the
explicit **Mark read** action does that. DMs and forums are the
exception: opening one marks it all read, as mobile does today.
- Reaching the bottom of a channel you marked unread ends that manual
unread. Unseen mentions and replies stay unread.
- Activity rows clear when you have read every message in the row, not
just the newest.


### Related issue
None found. buzz-app makes the same DM and manual-unread changes in
https://github.com/block/buzz-app/pull/616.

### Testing
- Unit tests cover the new write rules, Activity rows and the size limit
on saved read state.
- Widget tests cover reading on the channel page and the thread page,
opening a DM, and reaching the bottom of a channel you marked unread.
- No visual changes.

## Details
**Read marks.** "Caught up" is saved as `activity:<channel>` for a
channel and `thread-activity:<root>` for a thread. A caught-up mark
reads only ordinary top-level messages (channel) or that thread's
replies (thread). A per-message mark is `msg:<id>`.

**Caught-up time.** The mark is set to the time of the bottom row, and
never later than the current time. The time of a newer reply is not
used. So a top-level message that arrives late, with an earlier
timestamp, stays unread. A clock that runs fast, on this phone or the
sender's, cannot mark messages read before they arrive.

**Nested threads.** A thread opened on a nested reply shows only one
branch. Reaching its end writes per-message marks only, so unread
messages in other branches stay unread.

**Reading delay.** The 300 ms reading delay now restarts only when
messages arrive or leave. Before, any page rebuild restarted it, for
example a typing indicator.

**Activity rows.** A row that groups several events is done only when
each event is read. Each event is checked against the channel mark, its
own `msg:` mark, and its thread marks if it is a reply. Opening a row
goes to the oldest unread event.

**Publishing.** Mobile no longer republishes per-message marks it got
from other devices. buzz-app removes those once a caught-up mark covers
them. Broad marks (channel, thread, caught-up) are still republished, so
they stay available after the fetch window. The published blob is capped
at buzz-app's 40 KiB limit. Channel marks are kept first, then thread
marks, then caught-up marks, then the newest per-message marks. Before,
a blob over the encryption limit (about 64 KiB) turned off sync for the
rest of the session.

**DMs and manual unread.** Opening a DM reads it through the newest
loaded message, replies included, as mobile does today. Reaching the
bottom of a channel ends a manual unread. Manual unread on mobile is
local to this session, so ending it does not change what other devices
see.

**Compatibility.** Older desktop builds read only channel and
per-message marks. They may show extra unread for channels read on the
phone.

**Follow-ups, not in this PR**
- Unread catch-up fetches start after the channel mark and are capped at
1,000 events. In a busy channel, an old unread mention can drop out of
the badge. The fetch should also start from the caught-up mark, or load
more pages.
- Each mark triggers its own save and rebuild. They could be batched.
- Broadcast replies are counted slightly differently from buzz-app.
- buzz-app can also move its caught-up mark past the current time.

---------

Signed-off-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
This commit is contained in:
Logan Johnson
2026-10-14 20:04:57 -04:00
committed by GitHub
parent 71d7aa0436
commit 9ade5d1ad1
18 changed files with 2510 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
mobile/lib/features/activity/activity_page.dart
View File

@@ -253,8 +253,9 @@ class ActivityPage extends HookConsumerWidget {
// Deep-link to the represented message: oldest unread in the group,
// falling back to the latest event.
final readAt = resolveInboxItemReadAt(item, markerOf: markerOf);
final target = item.deepLinkTarget(readAt);
final target = item.deepLinkTarget(
(event) => isInboxEventRead(event, markerOf: markerOf),
);
final thread = threadReferenceOf(target.tags);
final threadRootId = isBroadcastReply(target.tags)
? null

22
mobile/lib/features/activity/inbox_item.dart
View File

@@ -113,20 +113,18 @@ class InboxItem {
return null;
}
/// The event the row should deep-link to: the oldest event in the group
/// newer than [readAt] (oldest unread), falling back to the latest event.
FeedItem deepLinkTarget(int? readAt) {
if (readAt != null) {
FeedItem? oldestUnread;
for (final candidate in groupItems) {
if (candidate.createdAt <= readAt) continue;
if (oldestUnread == null ||
candidate.createdAt < oldestUnread.createdAt) {
oldestUnread = candidate;
}
/// The event the row should deep-link to: the oldest grouped event that
/// [isRead] rejects (oldest unread), falling back to the latest event.
FeedItem deepLinkTarget(bool Function(FeedItem event) isRead) {
FeedItem? oldestUnread;
for (final candidate in groupItems) {
if (isRead(candidate)) continue;
if (oldestUnread == null ||
candidate.createdAt < oldestUnread.createdAt) {
oldestUnread = candidate;
}
if (oldestUnread != null) return oldestUnread;
}
if (oldestUnread != null) return oldestUnread;
return item;
}
}

71
mobile/lib/features/activity/inbox_read_state.dart
View File

@@ -1,30 +1,48 @@
import '../../shared/read_state/read_state_format.dart';
import 'feed_item.dart';
import 'inbox_item.dart';
/// Resolves the effective NIP-RS read marker for one inbox row, mirroring
/// desktop's `resolveInboxItemReadAt`:
/// - thread rows use `max(thread:<root>, msg:<id>)`
/// - channel rows use the channel context marker
/// - rows with no channel have no marker (caller falls back to local state)
int? resolveInboxItemReadAt(
InboxItem item, {
/// The newest read marker that reads one grouped [event], following the
/// unread badge's `observedUnreadEventReadAt`: the channel mark, the
/// event's own `msg:` mark, and for a thread reply `thread:<root>` and
/// `thread-activity:<root>`. Returns null for an event with no channel.
///
/// `activity:<channel>` never counts here. Activity rows hold only mentions,
/// needs-action and agent events addressed to the reader, and DM messages,
/// and that catch-up mark reads none of those.
int? inboxEventReadAt(
FeedItem event, {
required int? Function(String contextId) markerOf,
}) {
final channelId = item.item.channelId;
final threadRootId = item.threadRootId;
if (threadRootId != null) {
return maxReadAt([
markerOf(threadContextKey(threadRootId)),
markerOf(msgContextKey(item.item.id)),
]);
}
return channelId == null ? null : markerOf(channelId);
final channelId = event.channelId;
if (channelId == null) return null;
final rootId = isThreadReply(event.tags)
? threadReferenceOf(event.tags).rootId
: null;
return maxReadAt([
markerOf(channelId),
markerOf(msgContextKey(event.id)),
if (rootId != null) ...[
markerOf(threadContextKey(rootId)),
markerOf(threadActivityContextKey(rootId)),
],
]);
}
/// Whether the row is read ("done"), mirroring desktop's
/// `useHomeInboxReadState` projection: a local unread override always wins;
/// otherwise the row is done when no grouped activity is newer than the
/// shared read marker; channel-less rows fall back to the local done set.
/// Whether the marks read one grouped [event].
bool isInboxEventRead(
FeedItem event, {
required int? Function(String contextId) markerOf,
}) {
final readAt = inboxEventReadAt(event, markerOf: markerOf);
return readAt != null && event.createdAt <= readAt;
}
/// Whether the row is read ("done"): a local unread override always wins;
/// otherwise the row is done when the marks read every grouped event. Each
/// event is checked on its own, because reading a channel marks mentions
/// with per-message marks, so reading the newest says nothing about the
/// rest. Channel-less rows fall back to the local done set.
bool isInboxItemDone(
InboxItem item, {
required int? Function(String contextId) markerOf,
@@ -34,13 +52,16 @@ bool isInboxItemDone(
final ids = groupedInboxItemIds(item);
if (ids.any(localUnreadOverrides.contains)) return false;
final readAt = resolveInboxItemReadAt(item, markerOf: markerOf);
if (readAt != null) return item.latestActivityAt <= readAt;
if (item.threadRootId != null || item.item.channelId != null) return false;
return localDoneSet.contains(item.id);
if (item.item.channelId == null) return localDoneSet.contains(item.id);
return _groupedEvents(
item,
).every((event) => isInboxEventRead(event, markerOf: markerOf));
}
Iterable<FeedItem> _groupedEvents(InboxItem item) => {
for (final event in [item.item, ...item.groupItems]) event.id: event,
}.values;
/// All event ids identified with the row — desktop's `getGroupedInboxItemIds`.
List<String> groupedInboxItemIds(InboxItem item) {
return {item.id, item.item.id, ...item.groupItems.map((i) => i.id)}.toList();

52
mobile/lib/features/channels/channel_detail_page.dart
View File

@@ -79,6 +79,7 @@ import 'small_avatar.dart';
import 'sticky_date_header.dart';
import 'thread_detail_page.dart';
import 'thread_replies_provider.dart';
import 'reading_marks.dart';
import 'timeline_message.dart';
part 'channel_detail_page/message_list.dart';
@@ -209,29 +210,21 @@ Future<void Function()> _subscribeToDmIdentityUpdates(
}
}
int? _channelReadTimestamp({
/// The time that opening [channel] reads it through, or null when opening it
/// reads nothing. Forums read through their last activity. DMs read through
/// the newest loaded message, including replies, or the last activity before
/// messages load.
int? _openReadTimestamp({
required Channel channel,
required AsyncValue<List<NostrEvent>> messagesState,
}) {
if (channel.isForum) {
return dateTimeToUnixSeconds(channel.lastMessageAt);
if (channel.isForum) return dateTimeToUnixSeconds(channel.lastMessageAt);
if (!channel.isDm) return null;
var latest = 0;
for (final event in messagesState.value ?? const <NostrEvent>[]) {
if (event.createdAt > latest) latest = event.createdAt;
}
final events = messagesState.value;
if (events != null && events.isNotEmpty) {
var latest = 0;
for (final event in events) {
if (event.threadReference.parentId != null) continue;
if (event.createdAt > latest) {
latest = event.createdAt;
}
}
if (latest > 0) {
return latest;
}
}
return dateTimeToUnixSeconds(channel.lastMessageAt);
return latest > 0 ? latest : dateTimeToUnixSeconds(channel.lastMessageAt);
}
bool _isOneToOneAgentDm(Channel channel, Set<String> agentPubkeys) {
@@ -517,7 +510,11 @@ class ChannelDetailPage extends HookConsumerWidget {
!messagesNotifier.hasLoadedMessages;
final appBarTitleContentHeight = _twoLineAppBarTitleContentHeight(context);
final readTimestamp = _channelReadTimestamp(
// Opening a forum or a DM reads the whole channel: a forum has no
// timeline to read row by row, and a DM is all for the reader. Other
// timelines write their read marks while the reader looks at rows; see
// `readingMarks`.
final openReadTimestamp = _openReadTimestamp(
channel: resolvedChannel,
messagesState: messagesState,
);
@@ -550,19 +547,25 @@ class ChannelDetailPage extends HookConsumerWidget {
],
);
// Opening reads only while this page is in front and the app is in use.
// A covered or backgrounded DM must not read a reply that loads meanwhile;
// it reads through it when it is shown again.
final openReadActive =
isAppInUse(useAppLifecycleState()) &&
(ModalRoute.of(context)?.isCurrent ?? true);
useEffect(() {
if (!readState.isReady || readTimestamp == null) {
if (!readState.isReady || !openReadActive || openReadTimestamp == null) {
return null;
}
return deferReadStateUpdate(context, () {
ref
.read(readStateProvider.notifier)
.markContextRead(channel.id, readTimestamp);
.markContextRead(channel.id, openReadTimestamp);
ref
.read(channelsProvider.notifier)
.clearObservedUnreadCoveredByRead(channel.id, readTimestamp);
.clearObservedUnreadCoveredByRead(channel.id, openReadTimestamp);
});
}, [channel.id, readState.isReady, readTimestamp]);
}, [channel.id, readState.isReady, openReadActive, openReadTimestamp]);
final dmHeader = resolvedChannel.isDm
? _watchDmHeader(ref, resolvedChannel, currentPubkey)
@@ -797,6 +800,7 @@ class ChannelDetailPage extends HookConsumerWidget {
initialOldestOrdinaryUnreadMessageId !=
null),
channelId: channel.id,
isDm: resolvedChannel.isDm,
currentPubkey: currentPubkey,
isMember: resolvedChannel.isMember,
isArchived: resolvedChannel.isArchived,

79
mobile/lib/features/channels/channel_detail_page/message_list.dart
View File

@@ -12,6 +12,7 @@ class _MessageList extends HookConsumerWidget {
final Set<String> initialForcedUnreadMessageIds;
final bool hasInitialUnread;
final String channelId;
final bool isDm;
final String? currentPubkey;
final bool isMember;
final bool isArchived;
@@ -32,6 +33,7 @@ class _MessageList extends HookConsumerWidget {
required this.initialForcedUnreadMessageIds,
required this.hasInitialUnread,
required this.channelId,
required this.isDm,
required this.currentPubkey,
required this.isMember,
required this.isArchived,
@@ -560,6 +562,83 @@ class _MessageList extends HookConsumerWidget {
],
);
void readVisibleRows() {
final readState = ref.read(readStateProvider);
final viewportHeight = timelineViewportHeight.value;
if (!context.mounted ||
!readState.isReady ||
isAutoScrolling.value ||
viewportHeight <= 0 ||
entries.isEmpty) {
return;
}
// Rows hidden under the app bar, the composer or the keyboard are not
// read. The keyboard covers rows even when the list does not follow
// the latest message, so this uses the whole covered height, not the
// list's own bottom inset.
final bottomEdge =
(navigationBottomInset / viewportHeight).clamp(0.0, 1.0) - 0.01;
final topEdge =
1 -
frostedAppBarHeight(
context,
titleContentHeight: appBarTitleContentHeight,
) /
viewportHeight +
0.01;
final visible = <TimelineMessage>[];
for (final position in itemPositionsListener.itemPositions.value) {
if (position.index >= displayEntries.length ||
position.itemLeadingEdge < bottomEdge ||
position.itemTrailingEdge > topEdge) {
continue;
}
final group =
displayEntries[displayEntries.length - 1 - position.index];
visible.addAll(group.map((entry) => entry.message));
}
final latest = entries.last.message;
final atBottom =
latestIsAtBoundary() &&
itemPositionsListener.itemPositions.value.any(
(position) =>
position.index == 0 && position.itemLeadingEdge >= bottomEdge,
);
final marks = readingMarks(
readState: readState,
channelId: channelId,
isDm: isDm,
currentPubkey: currentPubkey,
loaded: allMessages,
visible: visible,
bottom: atBottom ? latest : null,
);
final notifier = ref.read(readStateProvider.notifier);
// Reading to the bottom ends a manual "mark unread" on the channel.
if (atBottom && readState.isForcedUnread(channelId)) {
notifier.clearForcedUnread(channelId);
}
for (final mark in marks.entries) {
notifier.markContextRead(mark.key, mark.value);
}
}
// Reading starts once the timeline holds still for the dwell while this
// page is in front and the app is in use.
final appInUse = isAppInUse(useAppLifecycleState());
final readStateReady = ref.watch(
readStateProvider.select((state) => state.isReady),
);
useReadingDwell(
positions: itemPositionsListener.itemPositions,
active:
readStateReady &&
appInUse &&
(ModalRoute.of(context)?.isCurrent ?? true),
onDwell: readVisibleRows,
keys: [channelId, readingContentKey(allMessages), navigationBottomInset],
);
useEffect(() {
stickyDayTimestamp.value = null;
return null;

185
mobile/lib/features/channels/reading_marks.dart Normal file
View File

@@ -0,0 +1,185 @@
import 'dart:async';
import 'package:flutter_hooks/flutter_hooks.dart';
import 'package:flutter/foundation.dart';
import 'package:flutter/widgets.dart';
import '../../shared/read_state/read_state_format.dart';
import '../../shared/read_state/read_state_provider.dart';
import 'timeline_message.dart';
import 'unread_badge/is_high_priority_event.dart';
/// How long a row must stay fully visible before it counts as read. This is
/// the same dwell the web and desktop app (`buzz-app`) uses.
const readingDwell = Duration(milliseconds: 300);
/// The read marks to write after a reading dwell, following `buzz-app`'s
/// rules (its `docs/unread.md`):
///
/// - When the newest message is at the bottom, a channel writes
/// `activity:<channel>` and a thread writes `thread-activity:<root>`. These
/// catch-up marks read ordinary messages and that thread's replies. They do
/// not read mentions, broadcasts or DMs.
/// - Each fully visible message gets its own `msg:` mark unless its own mark,
/// the channel mark, or (for an ordinary top-level message) `activity:`
/// already reads it. A thread mark never counts here: a reply finds its
/// thread only while the root is loaded, so `buzz-app` keeps reply marks.
/// - Automatic reading never writes the channel mark. Only an explicit
/// Mark read does that.
///
/// [loaded] is every message loaded for the channel or thread, including
/// replies. [visible] is the messages fully visible for the whole dwell.
/// [bottom] is the newest message when its bottom edge is visible at the
/// bottom of the list; otherwise null. Set [threadRootId] when reading a
/// thread, and [isRootThread] when its head is the thread root rather than a
/// nested reply. [now] is the current Unix time in seconds; it defaults to
/// the clock. The result maps each read-state context to its new time.
Map<String, int> readingMarks({
required ReadStateState readState,
required String channelId,
required bool isDm,
required String? currentPubkey,
required Iterable<TimelineMessage> loaded,
required Iterable<TimelineMessage> visible,
TimelineMessage? bottom,
String? threadRootId,
bool isRootThread = true,
int? now,
}) {
final marks = <String, int>{};
int? markerOf(String contextId) =>
maxReadAt([marks[contextId], readState.effectiveTimestamp(contextId)]);
// A nested thread view shows only one branch, so reaching its bottom does
// not read the whole thread.
if (bottom != null && (threadRootId == null || isRootThread)) {
final catchUp = _catchUpMark(
channelId: channelId,
loaded: loaded,
bottom: bottom,
threadRootId: threadRootId,
markerOf: markerOf,
now: now ?? DateTime.now().millisecondsSinceEpoch ~/ 1000,
);
if (catchUp != null) marks[catchUp.key] = catchUp.value;
}
final self = currentPubkey?.toLowerCase();
for (final message in visible) {
if (message.isSystem || message.pubkey.toLowerCase() == self) continue;
final contextId = msgContextKey(message.id);
// Automatic reading keeps a message the reader marked unread.
if (readState.isForcedUnread(contextId)) continue;
final ordinary = readByChannelCatchUp(
isDm: isDm,
isReply: message.parentId != null && !_isBroadcast(message),
highPriority: self == null || isHighPriorityEvent(message.tags, self),
);
final readAt = maxReadAt([
markerOf(contextId),
markerOf(channelId),
if (ordinary) markerOf(activityContextKey(channelId)),
]);
if (readAt != null && readAt >= message.createdAt) continue;
marks[contextId] = message.createdAt;
}
return marks;
}
/// The catch-up mark for [bottom], or null when it would read nothing new.
MapEntry<String, int>? _catchUpMark({
required String channelId,
required Iterable<TimelineMessage> loaded,
required TimelineMessage bottom,
required String? threadRootId,
required int? Function(String contextId) markerOf,
required int now,
}) {
// A loaded message newer than the bottom row means the list does not show
// the live bottom, for example after a jump into history.
final showsNewest = !loaded.any(
(message) =>
message.createdAt > bottom.createdAt &&
(threadRootId != null
? _threadRootOf(message) == threadRootId
: message.parentId == null || _isBroadcast(message)),
);
if (!showsNewest) return null;
// Cut at the bottom row. A newer reply's time would also read a top-level
// message that arrives late with an earlier time. A clock ahead of now, on
// this device or the sender's, must not read messages before they arrive.
final cut = bottom.createdAt < now ? bottom.createdAt : now;
final key = threadRootId != null
? threadActivityContextKey(threadRootId)
: activityContextKey(channelId);
final covered = maxReadAt([
markerOf(key),
markerOf(channelId),
if (threadRootId != null) markerOf(threadContextKey(threadRootId)),
]);
if (covered != null && covered >= cut) return null;
return MapEntry(key, cut);
}
String? _threadRootOf(TimelineMessage message) =>
message.parentId == null ? null : message.rootId ?? message.parentId;
bool _isBroadcast(TimelineMessage message) => message.tags.any(
(tag) => tag.length >= 2 && tag[0] == 'broadcast' && tag[1] == '1',
);
/// A stable dwell key for [messages]. Pages rebuild a new message list on
/// every frame that touches them, such as a typing indicator, so the list
/// itself would restart the dwell each time. This changes only when a
/// message arrives, leaves, or moves the newest time.
(int, int, String?) readingContentKey(Iterable<TimelineMessage> messages) {
var count = 0;
TimelineMessage? newest;
for (final message in messages) {
count++;
if (newest == null || message.createdAt > newest.createdAt) {
newest = message;
}
}
return (count, newest?.createdAt ?? 0, newest?.id);
}
/// Whether the app is in the foreground. Null means the platform has not
/// reported a state yet, which only happens before the first frame and in
/// tests.
bool isAppInUse(AppLifecycleState? state) =>
state == null || state == AppLifecycleState.resumed;
/// Runs [onDwell] once the list's item positions have stayed still for
/// [readingDwell]. Any position change restarts the wait. Changing [keys]
/// restarts it too, for changes that move no rows, such as new content or
/// the route coming back to the front. The wait is cancelled while
/// [active] is false. [onDwell] reads the current state when it runs.
void useReadingDwell({
required ValueListenable<Object?> positions,
required bool active,
required VoidCallback onDwell,
List<Object?> keys = const [],
}) {
final callback = useRef(onDwell);
callback.value = onDwell;
useEffect(() {
if (!active) return null;
Timer? timer;
void restart() {
timer?.cancel();
timer = Timer(readingDwell, () {
timer = null;
callback.value();
});
}
positions.addListener(restart);
restart();
return () {
timer?.cancel();
positions.removeListener(restart);
};
}, [positions, active, ...keys]);
}

77
mobile/lib/features/channels/thread_detail_page.dart
View File

@@ -53,11 +53,11 @@ import 'message_action_backdrop_state.dart';
import 'message_long_press_region.dart';
import 'message_content.dart';
import 'reaction_row.dart';
import '../../shared/read_state/read_state_format.dart';
import '../../shared/read_state/read_state_provider.dart';
import 'send_message_provider.dart';
import 'small_avatar.dart';
import 'sticky_date_header.dart';
import 'reading_marks.dart';
import 'timeline_message.dart';
part 'thread_detail_page/nested_thread_summary_row.dart';
@@ -68,7 +68,6 @@ part 'thread_detail_helpers.dart';
part 'thread_detail_page/tail_alignment.dart';
part 'thread_detail_page/thread_message.dart';
part 'thread_detail_page/avatar.dart';
part 'thread_detail_page/read_state.dart';
part 'thread_detail_page/app_bar.dart';
const _landingHighlightDuration = Duration(seconds: 3);
@@ -712,7 +711,79 @@ class ThreadDetailPage extends HookConsumerWidget {
);
return null;
}, [hasFetchedReplies, replies.length, settleGeometry, headHeight.value]);
_useThreadReplyReadState(ref, threadHead.id, replies);
void readVisibleReplies() {
final readState = ref.read(readStateProvider);
if (!context.mounted ||
!readState.isReady ||
!viewportHeight.isFinite ||
viewportHeight <= 0) {
return;
}
// Rows hidden under the app bar, the composer or the keyboard are not
// read. The keyboard covers rows even when the list does not follow
// the tail, so this uses the whole covered height.
final topEdge = topOverlayFraction - 0.01;
final bottomEdge =
1 - ((Grid.xs + navigationBottomInset) / viewportHeight) + 0.01;
final visible = <TimelineMessage>[];
for (final position in itemPositionsListener.itemPositions.value) {
if (position.itemLeadingEdge < topEdge ||
position.itemTrailingEdge > bottomEdge) {
continue;
}
if (position.index == headIndex) {
// The head is read like a reply. A thread opened directly, from
// a link or a notification, may be the only place it is seen.
if (!liveDeletionHidesHead) visible.add(liveHead);
continue;
}
final replyIndex = position.index - indexForReply(0);
if (replyIndex < 0 || replyIndex >= replies.length) continue;
visible.add(replies[replyIndex]);
}
final isDm =
ref
.read(channelsProvider)
.value
?.where((candidate) => candidate.id == channelId)
.firstOrNull
?.isDm ??
false;
final marks = readingMarks(
readState: readState,
channelId: channelId,
isDm: isDm,
currentPubkey: currentPubkey,
loaded: allMsgs,
visible: visible,
bottom: replies.isNotEmpty && threadTailIsVisible()
? replies.last
: null,
threadRootId: queryRootId,
isRootThread: threadHead.parentId == null,
);
final notifier = ref.read(readStateProvider.notifier);
for (final mark in marks.entries) {
notifier.markContextRead(mark.key, mark.value);
}
}
// Reading starts once the thread holds still for the dwell while this
// page is in front and the app is in use.
final appInUse = isAppInUse(useAppLifecycleState());
final readStateReady = ref.watch(
readStateProvider.select((state) => state.isReady),
);
useReadingDwell(
positions: itemPositionsListener.itemPositions,
active:
threadViewportVisible &&
readStateReady &&
appInUse &&
(ModalRoute.of(context)?.isCurrent ?? true),
onDwell: readVisibleReplies,
keys: [threadHead.id, readingContentKey(allMsgs), navigationBottomInset],
);
// Thread-scoped typing indicators (exclude self).
final allTyping = ref.watch(channelTypingProvider(channelId));

24
mobile/lib/features/channels/thread_detail_page/read_state.dart
View File

@@ -1,24 +0,0 @@
part of '../thread_detail_page.dart';
void _useThreadReplyReadState(
WidgetRef ref,
String threadHeadId,
List<TimelineMessage> replies,
) {
final readState = ref.watch(readStateProvider);
final visibleReplyReadKey = replies
.map((reply) => '${reply.id}:${reply.createdAt}')
.join(',');
useEffect(() {
if (!readState.isReady || replies.isEmpty) return null;
WidgetsBinding.instance.addPostFrameCallback((_) {
for (final reply in replies) {
ref
.read(readStateProvider.notifier)
.markContextRead(msgContextKey(reply.id), reply.createdAt);
}
});
return null;
}, [threadHeadId, readState.isReady, visibleReplyReadKey]);
}

270
mobile/lib/shared/read_state/read_state_format.dart
View File

@@ -30,6 +30,243 @@ bool readByChannelCatchUp({
required bool highPriority,
}) => !isDm && !isReply && !highPriority;
/// The plaintext budget for one published read-state slot. The web app
/// (`buzz-app`) uses the same limit, well under NIP-44's 64 KiB maximum.
const readStatePlaintextBytes = 40 * 1024;
/// The most `msg:` and `thread:` marks this device saves. The desktop app
/// uses the same limit.
const localMaxPrunableContexts = 1000;
bool _isPrunableContext(String contextId) =>
contextId.startsWith(msgContextPrefix) ||
contextId.startsWith(threadContextPrefix);
/// The marks from [contexts] that this device saves, following the desktop
/// app's `pruneStaleContexts`. It drops `msg:` and `thread:` marks older
/// than the fetch horizon, then keeps the newest [localMaxPrunableContexts]
/// of them. Each automatic read writes a `msg:` mark, so without this bound
/// the saved state, and the time to save it, would grow with every read.
/// Channel and catch-up marks are kept: there is one per channel or thread,
/// and losing one would show read messages as unread again.
Map<String, int> pruneStaleContexts(
Map<String, int> contexts, {
required int nowUnixSeconds,
}) {
final cutoff = nowUnixSeconds - readStateHorizonSeconds;
final kept = <String, int>{};
final prunable = <MapEntry<String, int>>[];
for (final entry in contexts.entries) {
if (!_isPrunableContext(entry.key)) {
kept[entry.key] = entry.value;
} else if (entry.value >= cutoff) {
prunable.add(entry);
}
}
prunable.sort((a, b) {
final byTime = b.value.compareTo(a.value);
return byTime != 0 ? byTime : a.key.compareTo(b.key);
});
for (final entry in prunable.take(localMaxPrunableContexts)) {
kept[entry.key] = entry.value;
}
return kept;
}
/// Whether [contextId] belongs to the web app's manual-unread overrides
/// (`ov_s:`, `ov_c:`, `ov_b:`) or their escaped frontiers (`esc:`). This app
/// does not read overrides. It does not take them from other slots, but it
/// carries the ones already in its own slot unchanged: NIP-RS forbids
/// dropping them, and an old version of this app copied them there.
bool isOverrideContext(String contextId) =>
contextId.startsWith('ov_') || contextId.startsWith('esc:');
const _overridePrefixes = ['ov_s:', 'ov_c:', 'ov_b:'];
/// The raw context ID of an override counter key, or null for other keys.
String? _overrideTarget(String key) {
for (final prefix in _overridePrefixes) {
if (key.startsWith(prefix)) return key.substring(prefix.length);
}
return null;
}
/// The frontier key of raw context [contextId] on the wire. NIP-RS escapes
/// a raw ID that begins with `ov_` or `esc:` by prepending `esc:`.
String overrideFrontierKey(String contextId) =>
contextId.startsWith('ov_') || contextId.startsWith('esc:')
? 'esc:$contextId'
: contextId;
/// Whether [keys] form a legal override group for [target]: exactly
/// `ov_s:`, `ov_c:` and `ov_b:`, or `ov_c:` alone as a tombstone.
bool _isOverrideShape(String target, Iterable<String> keys) {
final set = keys.toSet();
return set.length == _overridePrefixes.length ||
(set.length == 1 && set.contains('ov_c:$target'));
}
/// The override groups in [contexts] that have a legal shape, by target.
Map<String, Map<String, int>> _overrideGroups(Map<String, int> contexts) {
final groups = <String, Map<String, int>>{};
for (final entry in contexts.entries) {
if (_overrideTarget(entry.key) case final target?) {
(groups[target] ??= {})[entry.key] = entry.value;
}
}
return {
for (final MapEntry(key: target, value: group) in groups.entries)
if (_isOverrideShape(target, group.keys)) target: group,
};
}
/// The override keys of [contexts] that this app may carry. NIP-RS accepts
/// an override group only whole (see [_isOverrideShape]), and a group must
/// travel with its frontier, found in [contexts] or [frontiers]. Any other
/// group is rejected whole, so none of its keys is carried. Escaped
/// frontiers (`esc:`) are frontier keys, not counters, so each is carried
/// as it is.
Map<String, int> completeOverrideGroups(
Map<String, int> contexts, {
Map<String, int> frontiers = const {},
}) {
final kept = <String, int>{
for (final entry in contexts.entries)
if (entry.key.startsWith('esc:')) entry.key: entry.value,
};
for (final MapEntry(key: target, value: group) in _overrideGroups(
contexts,
).entries) {
final frontier = overrideFrontierKey(target);
if (contexts.containsKey(frontier) || frontiers.containsKey(frontier)) {
kept.addAll(group);
}
}
return kept;
}
/// The frontier keys that must travel with the override groups in
/// [contexts] (NIP-RS's co-location rule).
Set<String> overrideGroupFrontierKeys(Map<String, int> contexts) => {
for (final target in _overrideGroups(contexts).keys)
overrideFrontierKey(target),
};
/// Whether this device republishes a mark it merged from another device's
/// slot. Message marks are not republished: each covers one message, and the
/// web app prunes them once a catch-up mark reads the message. Broad marks
/// are republished so they outlive the fetch horizon of the slot that wrote
/// them.
bool republishesMergedContext(String contextId) =>
!contextId.startsWith(msgContextPrefix) && !isOverrideContext(contextId);
/// Keep order for a slot that is over budget, following `buzz-app`'s
/// retention: channel marks first, then thread marks, then catch-up marks,
/// then message marks.
int _retentionScope(String key) {
if (!key.contains(':')) return 0;
if (key.startsWith(threadContextPrefix)) return 1;
if (key.startsWith('activity:') || key.startsWith('thread-activity:')) {
return 2;
}
return 3;
}
/// The share of the budget that broad marks may fill before recent use
/// decides. The rest always goes to the most recently written marks, so a
/// new read is never left out because old channel or thread marks fill the
/// budget. `buzz-app` uses the same share.
const _scopedShare = 0.75;
/// The marks from [contexts] that fit one read-state slot for [clientId]
/// within [maxBytes] of plaintext, following `buzz-app`'s retention.
///
/// Up to three quarters of the budget goes to broad marks first: channel
/// marks, then thread marks, then catch-up marks, then message marks. The
/// rest goes by [recent], the time each mark was last written, newest first,
/// so reading old history still syncs. A slot over NIP-44's limit cannot be
/// encrypted, so without this cap sync would stop.
///
/// [carried] override groups are kept whole, ahead of every mark, together
/// with each group's frontier from [contexts]: NIP-RS requires a context's
/// frontier and its override keys in the same event. Incomplete groups are
/// left out (see [completeOverrideGroups]). If the groups do not fit, this
/// returns null, and the caller must leave the slot as it is instead of
/// publishing part of a group.
Map<String, int>? retainReadStateContexts(
Map<String, int> contexts, {
required String clientId,
Map<String, int> recent = const {},
Map<String, int> carried = const {},
int maxBytes = readStatePlaintextBytes,
}) {
// JSON-encoded bytes, so escaped characters are counted as published.
int bytesOf(Object? value) => utf8.encode(jsonEncode(value)).length;
final groups = completeOverrideGroups(carried, frontiers: contexts);
final reserved = <String, int>{...groups};
for (final key in overrideGroupFrontierKeys(groups)) {
final frontier = contexts[key];
if (frontier != null && frontier > (reserved[key] ?? -1)) {
reserved[key] = frontier;
}
}
final fixedBytes = bytesOf(
ReadStateBlob(clientId: clientId, contexts: reserved).toJson(),
);
if (fixedBytes > maxBytes || reserved.length > _maxContexts) return null;
// Marks share what the carried keys leave.
final markBytes = maxBytes - fixedBytes;
final markCount = _maxContexts - reserved.length;
var used = 0;
var count = 0;
bool take(MapEntry<String, int> entry, double share) {
// A key, its colon, its value and one separating comma.
final cost = bytesOf(entry.key) + 2 + '${entry.value}'.length;
if (used + cost > markBytes * share || count >= markCount * share) {
return false;
}
used += cost;
count++;
return true;
}
int byUse(MapEntry<String, int> a, MapEntry<String, int> b) {
final byRecent = (recent[b.key] ?? 0).compareTo(recent[a.key] ?? 0);
if (byRecent != 0) return byRecent;
final byTime = b.value.compareTo(a.value);
return byTime != 0 ? byTime : a.key.compareTo(b.key);
}
final retained = <String, int>{...reserved};
final scoped =
contexts.entries
.where((entry) => !reserved.containsKey(entry.key))
.toList()
..sort((a, b) {
final byScope = _retentionScope(
a.key,
).compareTo(_retentionScope(b.key));
return byScope != 0 ? byScope : byUse(a, b);
});
for (final entry in scoped) {
// Stop at the first broad mark that does not fit, so a narrower mark
// never takes the share ahead of it.
if (!take(entry, _scopedShare)) break;
retained[entry.key] = entry.value;
}
final byRecent =
contexts.entries
.where((entry) => !reserved.containsKey(entry.key))
.toList()
..sort(byUse);
for (final entry in byRecent) {
if (!retained.containsKey(entry.key) && take(entry, 1)) {
retained[entry.key] = entry.value;
}
}
return retained;
}
int? maxReadAt(Iterable<int?> markers) {
int? latest;
for (final marker in markers) {
@@ -142,16 +379,37 @@ ReadStateBlob? decodeReadStateBlob(String plaintext) {
);
}
/// The valid entries of a decoded `contexts` object. Override counters are
/// checked as whole groups first, on the raw values, as NIP-RS requires: one
/// invalid sibling rejects its whole group, so dropping it alone can never
/// turn a live group into a tombstone. Other `ov_` keys are reserved and
/// dropped. Every other entry is checked on its own.
Map<String, int> sanitizeReadStateContexts(Map<String, Object?> contexts) {
bool valid(String key, Object? value) =>
utf8.encode(key).length <= 256 &&
value is int &&
value >= 0 &&
value <= 4294967295;
final groups = <String, Map<String, Object?>>{};
for (final entry in contexts.entries) {
if (_overrideTarget(entry.key) case final target?) {
(groups[target] ??= {})[entry.key] = entry.value;
}
}
final rejected = {
for (final MapEntry(key: target, value: group) in groups.entries)
if (!_isOverrideShape(target, group.keys) ||
group.entries.any((entry) => !valid(entry.key, entry.value)))
target,
};
final sanitized = <String, int>{};
for (final entry in contexts.entries) {
if (utf8.encode(entry.key).length > 256) continue;
if (entry.key.startsWith('ov_')) {
final target = _overrideTarget(entry.key);
if (target == null || rejected.contains(target)) continue;
}
final value = entry.value;
if (value is! int) continue;
if (value < 0 || value > 4294967295) continue;
sanitized[entry.key] = value;
if (valid(entry.key, value)) sanitized[entry.key] = value as int;
}
return sanitized;
}

143
mobile/lib/shared/read_state/read_state_manager.dart
View File

@@ -55,6 +55,9 @@ class ReadStateManager {
final Map<String, int> _effectiveState = {};
final Set<String> _publishableContextIds = {};
// Override keys already in this device's own slot, carried unchanged.
// See [isOverrideContext].
final Map<String, int> _carriedOverrides = {};
Map<String, int> _lastPublishedContexts = {};
Timer? _debounceTimer;
@@ -62,6 +65,9 @@ class ReadStateManager {
bool _initialized = false;
bool _disposed = false;
bool _isPublishing = false;
// Set when a publish is asked for while one is running. The running
// publish took its snapshot first, so it publishes again when it ends.
bool _publishAgain = false;
Completer<void>? _publishCompleter;
bool _remoteUnsupported = false;
int _maxFetchedCreatedAt = 0;
@@ -116,11 +122,14 @@ class ReadStateManager {
}
void markContextRead(String contextId, int unixTimestamp) {
_advanceContext(contextId, unixTimestamp, publishable: true);
if (_disposed || isOverrideContext(contextId)) return;
// Set first: retention keeps the most recently written marks, so the
// save that follows must already see this read as the newest.
_contextSourceCreatedAt[contextId] = max(
currentUnixSeconds(),
_maxFetchedCreatedAt + 1,
);
_advanceContext(contextId, unixTimestamp, publishable: true);
}
void seedContextRead(String contextId, int unixTimestamp) {
@@ -243,6 +252,7 @@ class ReadStateManager {
}
for (final entry in decoded.blob.contexts.entries) {
if (isOverrideContext(entry.key)) continue;
final result = _applyRemoteContextTimestamp(
contextId: entry.key,
timestamp: entry.value,
@@ -250,7 +260,9 @@ class ReadStateManager {
);
if (result == _ApplyRemoteContextResult.advanced) {
_pendingSyncedAdvances.add(entry.key);
_publishableContextIds.add(entry.key);
if (republishesMergedContext(entry.key)) {
_publishableContextIds.add(entry.key);
}
}
}
@@ -262,11 +274,28 @@ class ReadStateManager {
}
if (ownBlob != null) {
_carryOwnOverrides(ownBlob.contexts);
_lastPublishedContexts = Map<String, int>.from(ownBlob.contexts);
_publishableContextIds.addAll(ownBlob.contexts.keys);
_publishableContextIds.addAll(
ownBlob.contexts.keys.where((key) => !isOverrideContext(key)),
);
}
}
/// Merges the complete override groups of this device's own slot into
/// [_carriedOverrides] by `max()`, the NIP-RS merge rule. An incomplete
/// group is rejected whole (see [completeOverrideGroups]).
bool _carryOwnOverrides(Map<String, int> contexts) {
var changed = false;
for (final entry in completeOverrideGroups(contexts).entries) {
if (entry.value > (_carriedOverrides[entry.key] ?? -1)) {
_carriedOverrides[entry.key] = entry.value;
changed = true;
}
}
return changed;
}
Future<void> _startLiveSubscription() async {
try {
final unsub = await _relaySession!.subscribe(
@@ -313,8 +342,11 @@ class ReadStateManager {
_rotateSlotId();
}
var changed = false;
var changed =
decoded.blob.clientId == _clientId &&
_carryOwnOverrides(decoded.blob.contexts);
for (final entry in decoded.blob.contexts.entries) {
if (isOverrideContext(entry.key)) continue;
final result = _applyRemoteContextTimestamp(
contextId: entry.key,
timestamp: entry.value,
@@ -324,7 +356,9 @@ class ReadStateManager {
_pendingSyncedAdvances.add(entry.key);
changed = true;
}
if (_publishableContextIds.add(entry.key)) {
if ((decoded.blob.clientId == _clientId ||
republishesMergedContext(entry.key)) &&
_publishableContextIds.add(entry.key)) {
changed = true;
}
}
@@ -385,16 +419,45 @@ class ReadStateManager {
_signedEventRelay == null) {
return;
}
if (_isPublishing) return;
if (_isPublishing) {
// The running publish may have taken its snapshot before this change.
_publishAgain = true;
return _publishCompleter?.future;
}
final completer = Completer<void>();
_publishCompleter = completer;
_isPublishing = true;
try {
do {
_publishAgain = false;
await _publishOnce(_signedEventRelay);
} while (_publishAgain &&
(allowDisposed || !_disposed) &&
!_remoteUnsupported);
} finally {
_isPublishing = false;
_publishAgain = false;
completer.complete();
if (_publishCompleter == completer) {
_publishCompleter = null;
}
}
}
Future<void> _publishOnce(SignedEventRelay signedEventRelay) async {
debugPrint('[ReadStateManager] publish starting slotId=$_slotId');
try {
await _fetchOwnBlobBeforePublish();
final contexts = _currentContexts();
if (contexts == null) {
debugPrint(
'[ReadStateManager] publish skipped: carried override keys do not '
'fit the slot, so it stays as it is.',
);
return;
}
if (_isIdenticalToLastPublished(contexts)) {
return;
}
@@ -403,7 +466,7 @@ class ReadStateManager {
final ciphertext = _crypto.encrypt(jsonEncode(blob.toJson()));
final createdAt = max(currentUnixSeconds(), _maxFetchedCreatedAt + 1);
await _signedEventRelay.submit(
await signedEventRelay.submit(
kind: EventKind.readState,
content: ciphertext,
tags: [
@@ -444,12 +507,6 @@ class ReadStateManager {
return;
}
debugPrint('[ReadStateManager] publish failed: $error');
} finally {
_isPublishing = false;
completer.complete();
if (_publishCompleter == completer) {
_publishCompleter = null;
}
}
}
@@ -477,7 +534,9 @@ class ReadStateManager {
}
}
bool _isIdenticalToLastPublished(Map<String, int> contexts) {
bool _isIdenticalToLastPublished(Map<String, int>? contexts) {
// Null: the slot cannot be published, so there is nothing to send.
if (contexts == null) return true;
if (_lastPublishedContexts.length != contexts.length) {
return false;
}
@@ -495,24 +554,50 @@ class ReadStateManager {
return drained;
}
Map<String, int> _currentContexts() {
/// The slot to publish, or null when the carried override keys alone do
/// not fit, so the current slot must stay as it is.
Map<String, int>? _currentContexts() {
final contexts = <String, int>{};
for (final entry in _effectiveState.entries) {
if (_publishableContextIds.contains(entry.key)) {
contexts[entry.key] = entry.value;
}
}
return contexts;
// A carried group's frontier travels with it, even a merged `msg:` mark
// that would not be republished on its own.
for (final key in overrideGroupFrontierKeys(_carriedOverrides)) {
if (_effectiveState[key] case final frontier?) contexts[key] = frontier;
}
return retainReadStateContexts(
contexts,
clientId: _clientId,
recent: _contextSourceCreatedAt,
carried: _carriedOverrides,
);
}
void _hydrateFromLocalStorage() {
final stored = _storage.read(pubkey);
// Earlier versions merged the web app's override keys and published
// them in this device's slot, so publishable ones are carried.
_carriedOverrides
..clear()
..addAll(
completeOverrideGroups({
for (final entry in stored.contexts.entries)
if (isOverrideContext(entry.key) &&
stored.publishableContextIds.contains(entry.key))
entry.key: entry.value,
}, frontiers: stored.contexts),
);
_effectiveState
..clear()
..addAll(stored.contexts);
..addEntries(
stored.contexts.entries.where((entry) => !isOverrideContext(entry.key)),
);
_publishableContextIds
..clear()
..addAll(stored.publishableContextIds);
..addAll(stored.publishableContextIds.where(_effectiveState.containsKey));
_contextSourceCreatedAt
..clear()
..addAll(stored.sourceCreatedAt);
@@ -520,11 +605,27 @@ class ReadStateManager {
}
void _persistLocalState() {
// Save a bounded copy. Memory keeps every mark for this session, so a
// message read from old history stays read until the app restarts.
final saved = pruneStaleContexts(
_effectiveState,
nowUnixSeconds: currentUnixSeconds(),
)..addAll(_carriedOverrides);
// A carried group's frontier is never pruned away from it.
for (final key in overrideGroupFrontierKeys(_carriedOverrides)) {
if (_effectiveState[key] case final frontier?) saved[key] = frontier;
}
_storage.write(
pubkey,
_effectiveState,
_publishableContextIds,
_contextSourceCreatedAt,
saved,
{
..._publishableContextIds.where(saved.containsKey),
..._carriedOverrides.keys,
},
{
for (final entry in _contextSourceCreatedAt.entries)
if (saved.containsKey(entry.key)) entry.key: entry.value,
},
);
}

10
mobile/lib/shared/read_state/read_state_provider.dart
View File

@@ -180,6 +180,16 @@ class ReadStateNotifier extends Notifier<ReadStateState> {
_refreshForcedState();
}
/// Clear a forced-unread flag without moving any read marker. Reading to
/// the bottom of a channel the reader marked unread uses this: it ends the
/// manual unread but must not read mentions or replies the reader has not
/// seen.
void clearForcedUnread(String contextId) {
if (_forcedUnreadContexts.remove(contextId) != null) {
_refreshForcedState();
}
}
void _refreshForcedState() {
final manager = _manager;
if (manager == null) return;