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

perf(mobile): reduce cold startup and channel rendering delays (#6996)

Browse Source 
Pinky is opening this PR on Wes’s behalf.

## Summary

Reduce two separately measured mobile delays without changing the relay
API or removing rich message rendering:

- Publish the finite channel-list snapshot without waiting for live
subscription setup.
- Batch active channel-list subscriptions into sorted, deterministic
chunks of at most 128 explicit channel IDs, retaining unchanged chunks.
- Install replacement chunks before retiring old coverage. Retain old
chunks across thrown replacement failures; filter callbacks to the
current relay/identity and still-desired channels; clean up
retired/in-flight work across disconnect and disposal.
- Scope the custom-emoji Markdown matcher to known shortcodes actually
referenced in the rendered content, rather than embedding the whole
community palette in every message’s regex. Preserve unknown literals,
shared-colon token boundaries, event-tag URL priority, content edits,
and code literals.
- Honor explicit zero retry hints without inventing a ten-second
session-wide gate, while preserving the ordinary live-subscription retry
backoff and any already-active gate.

## Matched performance results

Medians of three before and three after process-cold launches,
alternated on the same authenticated iPhone 17 Pro / iOS 26.5 simulator.
Before is mobile source at `e76c81968b65b0755b83efdd59dc3375c59ddf40`;
after is this production patch before two documentation-only comment
fixes.

First channel-list frame: 11.617s → 3.179s · 73% lower latency

Live setup duration: 8.475s → 0.185s · 98% lower latency

Channel-open first message-list frame: 2.754s → 1.230s · 55% lower
latency

Message data ready → first frame: 1.977s → 0.286s · 86% lower latency

Channel-open reveal complete: 2.845s → 1.394s · 51% lower latency

Channel-open data readiness: 0.770s → 0.944s · 23% higher latency

The gain is client-side orchestration/rendering, not a claim that the
relay became faster. First channel-list frame ranges were 10.835–11.788s
before and 2.872–3.395s after; channel-open first-frame ranges were
1.560–2.906s before and 1.149–1.317s after.

### Measurement boundaries

- Debug simulator builds, CPU sampling disabled, bounded timestamp
probes enabled identically. These are not release/physical-device
measurements.
- Startup clock starts at Dart `main`; build/install/native pre-main
time is excluded. Auth/preferences and OS/disk caches are retained
between new processes.
- Same account scale: 113 active channels. Latest-message events varied
slightly with live activity (1543–1546).
- Channel-open uses the same initial 50-row history window, 97 query
events, and 67 provider events. The 2306-entry emoji palette is
explicitly loaded before navigation on both sides; palette preparation
is excluded from the channel-open clock and happens after the startup
frame measurement.
- Both diagnostic builds temporarily disabled unused avatar segmentation
to work around the existing Google ML Kit arm64-simulator slice
limitation. The workaround, dependency/native changes, auto-navigation,
and all probes are excluded from this PR.

## Validation

- Full mobile package suite: `flutter test` — 1890 passed.
- `just mobile-check` — 506 files unchanged; analyzer clean.
- `just file-size-check` — policy tests and all client ratchets passed.
- `git diff --check` — passed.
- New lifecycle regressions cover front-sorting insertion across a chunk
boundary while replacement readiness is paused, failure
retention/departed-channel filtering, retired generation + disconnect
cleanup, disposal, chunk limits, unchanged-set reuse, and scope
switches.
- Emoji unit/widget coverage includes a 2500-unused-emoji palette,
unknown tokens, case matching at the component level, shared-colon
boundaries, rich text, event URL priority, and content edits.
- Fresh-frame source review traced the subscription queue/fences,
callback scopes, duplicate-event paths, matcher/wiring, and retry
scheduling.
- At committed/pushed head `13a83b628c8411c5885e6f76a250ba87accf6067`,
all normal pre-push hooks passed: `mobile-checks` (formatter, analyzer,
and the full 1890-test mobile suite), `file-size-check`, `branch-skew`,
and `push-head-scope`. The commit hook formatted 506 files with no
changes. Runtime measurements preceded only the two
documentation-comment fixes; no runtime source changed afterward.

## Limits / follow-ups

- `RelaySession.subscribe` still settles under its existing
EOSE/fallback/retryable-CLOSED contract. “Setup completed” is not an
unconditional EOSE or live-delivery guarantee. This PR does not add
status-aware replacement ownership.
- The channel-message provider still awaits subscribe before fetching
history; that separate serialization is not removed here.
- Oversized Huddle queries and the separate history batching path above
128 active channels remain follow-ups, as do pre-existing read-state
initialization/size warnings.
- Palette-only widget refresh and upstream Markdown uppercase-dispatch
behavior are not changed.
- A clean source build still has the existing Google ML Kit
arm64-simulator issue; the profiling workaround is not a proposed
product fix.

Originating Buzz conversation:
buzz://message?channel=793b0522-7995-4375-b1a6-fd94a96fa21d&id=6ba88afdec78ab2cfb6728afcd4a6d10f29e6aa33ff0f62f45d6750381e4d789

---------

Signed-off-by: Wes <wesbillman@users.noreply.github.com>
Co-authored-by: Pinky <5f5ab050ec58ae208332edd544ebf705221e24c1b86d82a6ca07038a7a8f6ac9@buzz.block.builderlab.xyz>
Co-authored-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz>
Co-authored-by: Carl <9d00794d3df50972eb8b615511783cab12a77a8fd5dd5edd58073ec73b54bd8b@buzz.block.builderlab.xyz>
This commit is contained in:
Wes
2026-08-28 15:38:03 -06:00
committed by GitHub
parent c432a111ca
commit b593c7d7fe
15 changed files with 1383 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
mobile/lib/features/channels/channels_provider.dart
View File

@@ -37,16 +37,18 @@ const _authoredRootIdsPrefix = 'buzz-thread-authored.v1';
///
/// The paginated kind:39000 directory is fetched separately when Browse
/// channels opens, so discovery never delays the main Conversations screen.
/// Live updates are layered on top via per-channel subscriptions on the
/// `#h` tag for any of the visible channel event kinds — incoming events
/// bump `lastMessageAt` for that channel.
/// Live updates are layered on top via chunked subscriptions on the `#h` tag
/// for any visible channel event kind. Chunks stay within the relay's explicit
/// channel cap and incoming events bump `lastMessageAt` for their channel.
class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
static const _backstopInterval = Duration(seconds: 60);
final Map<String, void Function()> _unsubscribersByChannel = {};
final Map<String, _LiveChunkSubscription> _liveSubscriptionsByChunk = {};
Future<void> _liveSubscriptionQueue = Future.value();
Set<String> _desiredLiveChannelIds = const {};
int _subscriptionVersion = 0;
int _nextLiveChunkGeneration = 0;
final Set<String> _terminallyClosedLiveChunks = {};
String? _subscriptionRelayBaseUrl;
Timer? _backstopTimer;
final Map<String, int> _latestObservedByChannel = {};
@@ -66,6 +68,10 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
late final _ChannelRefreshCoordinator _refreshCoordinator =
_ChannelRefreshCoordinator.forRef(ref);
// Expose the notifier's protected Ref only to same-library extensions used to
// keep lifecycle code below the file-size ratchet.
Ref get _lifecycleRef => ref;
/// The member snapshot already returned while loading the channel list.
///
/// Mention autocomplete can use this synchronously while its independent
@@ -113,6 +119,9 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
!connected.isCompleted) {
connected.complete();
} else if (previous?.status != SessionStatus.connected) {
// A new authenticated connection can change relay admission policy.
// Ordinary refreshes must not retry an unchanged terminal rejection.
_terminallyClosedLiveChunks.clear();
unawaited(_backstopRefresh());
}
});
@@ -391,11 +400,11 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
}
}
// Layer live delivery onto the already-built snapshot without making EOSE
// part of the provider's readiness path. Every refresh queues a generation;
// a later refresh or scope switch retires older queued/in-flight work.
if (subscribeLive) {
// Subscriptions are shared relay state, so a retired refresh must not
// install them even though its channel list is already built.
fence.ensureCurrent();
await _fenced(fence, _subscribeLive(channels, fence));
unawaited(_subscribeLive(channels, fence));
}
// Guard the provider-state write in `retryDirectory` and `build`: the
// caller assigns whatever this returns, so the last check belongs here.
@@ -516,140 +525,6 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
);
}
/// Subscribe per-channel to live events (requires `#h` tag for relay
/// channel-scoped fan-out). Also starts a 60s WS backstop poll to reconcile
/// membership changes without repeatedly downloading the global directory.
Future<void> _subscribeLive(
List<Channel> channels,
_ChannelRefreshFence fence,
) {
final channelIds = {
for (final channel in channels)
if (channel.isMember && !channel.isArchived) channel.id,
};
final relayBaseUrl = ref.read(relayConfigProvider).baseUrl;
_desiredLiveChannelIds = channelIds;
final subscriptionVersion = ++_subscriptionVersion;
final sync = _liveSubscriptionQueue.then(
(_) => _syncLiveSubscriptions(
relayBaseUrl,
subscriptionVersion,
channels,
fence,
),
);
_liveSubscriptionQueue = sync.catchError((Object error, StackTrace stack) {
if (error is! _StaleChannelRefresh) {
debugPrint(
'[ChannelsNotifier] live subscription sync failed: $error\n$stack',
);
}
});
return sync;
}
Future<void> _syncLiveSubscriptions(
String relayBaseUrl,
int subscriptionVersion,
List<Channel> channels,
_ChannelRefreshFence fence,
) async {
fence.ensureCurrent();
if (ref.read(relaySessionProvider).status != SessionStatus.connected) {
return;
}
if (subscriptionVersion != _subscriptionVersion) {
return;
}
if (_subscriptionRelayBaseUrl != relayBaseUrl) {
for (final unsubscribe in _unsubscribersByChannel.values) {
unsubscribe();
}
_unsubscribersByChannel.clear();
_subscriptionRelayBaseUrl = relayBaseUrl;
}
if (ref.read(relayConfigProvider).baseUrl != relayBaseUrl) {
return;
}
final session = ref.read(relaySessionProvider.notifier);
final channelIds = _desiredLiveChannelIds;
for (final entry in _unsubscribersByChannel.entries.toList()) {
if (channelIds.contains(entry.key)) continue;
_unsubscribersByChannel.remove(entry.key);
entry.value();
}
for (final channelId in channelIds) {
if (ref.read(relaySessionProvider).status != SessionStatus.connected) {
return;
}
if (_unsubscribersByChannel.containsKey(channelId)) continue;
try {
final unsubscribe = await session.subscribe(
NostrFilter(
kinds: EventKind.channelEventKinds,
tags: {
'#h': [channelId],
},
limit: 0,
),
_handleLiveEvent,
);
if (!fence.isCurrent) {
unsubscribe();
throw const _StaleChannelRefresh();
}
if (subscriptionVersion != _subscriptionVersion ||
ref.read(relaySessionProvider).status != SessionStatus.connected ||
!_desiredLiveChannelIds.contains(channelId) ||
ref.read(relayConfigProvider).baseUrl != relayBaseUrl ||
_subscriptionRelayBaseUrl != relayBaseUrl) {
unsubscribe();
return;
}
final replaced = _unsubscribersByChannel[channelId];
if (replaced != null) {
unsubscribe();
continue;
}
_unsubscribersByChannel[channelId] = unsubscribe;
} on _StaleChannelRefresh {
rethrow;
} catch (error) {
debugPrint(
'[ChannelsNotifier] live subscription failed for $channelId: $error',
);
}
}
if (ref.read(relaySessionProvider).status != SessionStatus.connected) {
return;
}
if (subscriptionVersion != _subscriptionVersion) {
final desiredChannelIds = _desiredLiveChannelIds;
for (final entry in _unsubscribersByChannel.entries.toList()) {
if (desiredChannelIds.contains(entry.key)) continue;
_unsubscribersByChannel.remove(entry.key);
entry.value();
}
return;
}
fence.ensureCurrent();
unawaited(_catchUpUnreadEvents(channels, fence, subscriptionVersion));
_backstopTimer?.cancel();
_backstopTimer = Timer.periodic(
_backstopInterval,
(_) => _backstopRefresh(),
);
}
/// Backfills unread badges for the channels this refresh just installed.
///
/// Runs detached from the refresh that starts it, so the lifecycle token
@@ -668,6 +543,7 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
_ChannelRefreshFence fence,
int subscriptionGeneration,
) async {
if (!ref.mounted) return;
final myPk = ref.read(myPubkeyProvider);
if (myPk == null) return;
@@ -712,7 +588,9 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
// The relay round-trip above is the window Jed's probes park in: a newer
// refresh, a community switch or an identity switch here means every
// write below belongs to a channel list the user has left.
if (_isCatchUpRetired(fence, subscriptionGeneration)) return;
if (!ref.mounted || _isCatchUpRetired(fence, subscriptionGeneration)) {
return;
}
for (final event in events) {
if (event.pubkey.toLowerCase() == myPk.toLowerCase()) {
@@ -752,6 +630,7 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
state = state.whenData((channels) => List<Channel>.of(channels));
}
} catch (error) {
if (!ref.mounted) return;
debugPrint('[ChannelsNotifier] unread catch-up failed: $error');
}
}

272
mobile/lib/features/channels/channels_provider_lifecycle.dart
View File

@@ -1,18 +1,274 @@
part of 'channels_provider.dart';
// This extension exists only to keep `channels_provider.dart` under the
// desktop/mobile file-size ratchet; `_clearLiveSubscriptions` is not a
// standalone semantic boundary and belongs to `ChannelsNotifier`.
extension _ChannelsNotifierSubscriptionCleanup on ChannelsNotifier {
// Mirrors MAX_EXPLICIT_CHANNEL_VALUES in the relay REQ handler. A larger live
// set must use multiple subscriptions or the relay rejects the whole REQ.
const _maxLiveChannelsPerSubscription = 128;
/// Owns live channel subscription reconciliation and teardown.
///
/// Keeping this extension in a part preserves access to the notifier's private
/// lifecycle state while keeping `channels_provider.dart` below the mobile
/// file-size ratchet.
extension _ChannelsNotifierLiveSubscriptions on ChannelsNotifier {
/// Subscribe to live events in deterministic chunks that stay within the
/// relay's aggregate explicit-`#h` limit. Also starts a 60s WS backstop poll
/// to reconcile membership changes without downloading the global directory.
Future<void> _subscribeLive(
List<Channel> channels,
_ChannelRefreshFence fence,
) {
final channelIds = {
for (final channel in channels)
if (channel.isMember && !channel.isArchived) channel.id,
};
final relayBaseUrl = _lifecycleRef.read(relayConfigProvider).baseUrl;
_desiredLiveChannelIds = channelIds;
// A changed filter may be admitted, but an unchanged terminal rejection
// cannot recover just because a timer or manual refresh fetched it again.
// Drop obsolete keys so this quarantine stays bounded by desired chunks.
final desiredKeys = chunkChannelIdsForLiveSubscriptions(
channelIds,
).map(_liveChunkKey).toSet();
_terminallyClosedLiveChunks.retainAll(desiredKeys);
final subscriptionVersion = ++_subscriptionVersion;
final sync = _liveSubscriptionQueue.then(
(_) => _syncLiveSubscriptions(
relayBaseUrl,
subscriptionVersion,
channels,
fence,
),
);
_liveSubscriptionQueue = sync
.whenComplete(() {
// Reconcile even if a retired generation exits before its successor
// can run (for example, a disconnect while subscribe is pending).
if (_lifecycleRef.mounted) _removeUndesiredLiveChunks();
})
.catchError((Object error, StackTrace stack) {
if (error is! _StaleChannelRefresh) {
debugPrint(
'[ChannelsNotifier] live subscription sync failed: $error\n$stack',
);
}
});
return _liveSubscriptionQueue;
}
Future<void> _syncLiveSubscriptions(
String relayBaseUrl,
int subscriptionVersion,
List<Channel> channels,
_ChannelRefreshFence fence,
) async {
if (!_lifecycleRef.mounted) return;
fence.ensureCurrent();
if (_lifecycleRef.read(relaySessionProvider).status !=
SessionStatus.connected ||
subscriptionVersion != _subscriptionVersion) {
return;
}
if (_subscriptionRelayBaseUrl != relayBaseUrl) {
_clearRetainedLiveChunks();
_subscriptionRelayBaseUrl = relayBaseUrl;
}
if (_lifecycleRef.read(relayConfigProvider).baseUrl != relayBaseUrl) return;
final desiredChunks = chunkChannelIdsForLiveSubscriptions(
_desiredLiveChannelIds,
);
// Keep existing coverage while replacements are installed. Cleanup runs
// when this generation settles, and retains old coverage on failures.
final session = _lifecycleRef.read(relaySessionProvider.notifier);
for (final chunk in desiredChunks) {
final chunkKey = _liveChunkKey(chunk);
if (_liveSubscriptionsByChunk.containsKey(chunkKey) ||
_terminallyClosedLiveChunks.contains(chunkKey)) {
continue;
}
if (_lifecycleRef.read(relaySessionProvider).status !=
SessionStatus.connected) {
return;
}
final generation = ++_nextLiveChunkGeneration;
final subscription = _LiveChunkSubscription(generation);
_liveSubscriptionsByChunk[chunkKey] = subscription;
try {
final unsubscribe = await session.subscribe(
NostrFilter(
kinds: EventKind.channelEventKinds,
tags: {'#h': chunk},
limit: 0,
),
(event) {
// Superseded chunks can overlap during replacement or recovery.
// Deliver only the current scope's still-desired channels; duplicate
// events are already idempotent in the unread/timestamp stores.
if (!_lifecycleRef.mounted ||
_refreshCoordinator.currentScope() != fence.scope ||
!_desiredLiveChannelIds.contains(event.channelId)) {
return;
}
_handleLiveEvent(event);
},
onClosed: (message) =>
_handleLiveChunkClosed(chunkKey, generation, message),
);
subscription.unsubscribe = unsubscribe;
if (!_lifecycleRef.mounted || !fence.isCurrent) {
_liveSubscriptionsByChunk.remove(chunkKey);
unsubscribe();
if (!fence.isCurrent) throw const _StaleChannelRefresh();
return;
}
if (subscriptionVersion != _subscriptionVersion ||
_lifecycleRef.read(relaySessionProvider).status !=
SessionStatus.connected ||
_lifecycleRef.read(relayConfigProvider).baseUrl != relayBaseUrl ||
_subscriptionRelayBaseUrl != relayBaseUrl ||
!chunk.every(_desiredLiveChannelIds.contains)) {
_liveSubscriptionsByChunk.remove(chunkKey);
unsubscribe();
return;
}
if (_liveSubscriptionsByChunk[chunkKey] != subscription) {
unsubscribe();
continue;
}
} on _StaleChannelRefresh {
rethrow;
} catch (error) {
if (_liveSubscriptionsByChunk[chunkKey] == subscription) {
_liveSubscriptionsByChunk.remove(chunkKey);
}
if (!_lifecycleRef.mounted) return;
debugPrint(
'[ChannelsNotifier] live subscription failed for '
'${chunk.length} channels: $error',
);
}
}
if (!_lifecycleRef.mounted ||
_lifecycleRef.read(relaySessionProvider).status !=
SessionStatus.connected ||
subscriptionVersion != _subscriptionVersion) {
return;
}
fence.ensureCurrent();
unawaited(_catchUpUnreadEvents(channels, fence, subscriptionVersion));
_backstopTimer?.cancel();
_backstopTimer = Timer.periodic(
ChannelsNotifier._backstopInterval,
(_) => _backstopRefresh(),
);
}
void _handleLiveChunkClosed(String chunkKey, int generation, String message) {
if (!_lifecycleRef.mounted) return;
final subscription = _liveSubscriptionsByChunk[chunkKey];
if (subscription == null || subscription.generation != generation) return;
_liveSubscriptionsByChunk.remove(chunkKey);
// RelaySession calls onClosed only for terminal admission failures. Its
// transient/rate-limit retries have their own backoff. Do not turn a
// terminal rejection into a fresh membership/history/REQ loop here.
if (chunkChannelIdsForLiveSubscriptions(
_desiredLiveChannelIds,
).any((chunk) => _liveChunkKey(chunk) == chunkKey)) {
_terminallyClosedLiveChunks.add(chunkKey);
}
debugPrint(
'[ChannelsNotifier] live subscription closed by relay: $message',
);
}
void _removeUndesiredLiveChunks() {
final desiredChunks = chunkChannelIdsForLiveSubscriptions(
_desiredLiveChannelIds,
);
final desiredKeys = desiredChunks.map(_liveChunkKey).toSet();
final retainedKeys = <String>{...desiredKeys};
final uncoveredIds = <String>{};
for (final chunk in desiredChunks) {
final key = _liveChunkKey(chunk);
if (!_liveSubscriptionsByChunk.containsKey(key)) {
uncoveredIds.addAll(chunk);
}
}
final obsolete = _liveSubscriptionsByChunk.entries
.where((entry) => !desiredKeys.contains(entry.key))
.toList();
while (uncoveredIds.isNotEmpty) {
MapEntry<String, _LiveChunkSubscription>? best;
var bestCoverage = 0;
for (final entry in obsolete) {
if (retainedKeys.contains(entry.key)) continue;
final coverage = entry.key
.split('\u0000')
.where(uncoveredIds.contains)
.length;
if (coverage > bestCoverage) {
best = entry;
bestCoverage = coverage;
}
}
if (best == null) break;
retainedKeys.add(best.key);
uncoveredIds.removeAll(best.key.split('\u0000'));
}
for (final entry in _liveSubscriptionsByChunk.entries.toList()) {
if (retainedKeys.contains(entry.key)) continue;
_liveSubscriptionsByChunk.remove(entry.key);
entry.value.unsubscribe?.call();
}
}
void _clearRetainedLiveChunks() {
for (final subscription in _liveSubscriptionsByChunk.values) {
subscription.unsubscribe?.call();
}
_liveSubscriptionsByChunk.clear();
}
void _clearLiveSubscriptions() {
_subscriptionVersion++;
_desiredLiveChannelIds = const {};
for (final unsubscribe in _unsubscribersByChannel.values) {
unsubscribe();
}
_unsubscribersByChannel.clear();
_terminallyClosedLiveChunks.clear();
_clearRetainedLiveChunks();
_subscriptionRelayBaseUrl = null;
_backstopTimer?.cancel();
_backstopTimer = null;
}
}
/// Returns deterministic channel-ID chunks within the relay's live REQ cap.
List<List<String>> chunkChannelIdsForLiveSubscriptions(
Iterable<String> channelIds,
) {
final sortedIds = channelIds.toList()..sort();
return [
for (
var start = 0;
start < sortedIds.length;
start += _maxLiveChannelsPerSubscription
)
sortedIds.sublist(
start,
min(start + _maxLiveChannelsPerSubscription, sortedIds.length),
),
];
}
String _liveChunkKey(List<String> channelIds) => channelIds.join('\u0000');
class _LiveChunkSubscription {
_LiveChunkSubscription(this.generation);
final int generation;
void Function()? unsubscribe;
}

6
mobile/lib/features/channels/message_content.dart
View File

@@ -287,7 +287,11 @@ class MessageContent extends HookConsumerWidget {
agentMentionPubkeys: resolvedAgentMentionPubkeys,
onMentionTap: onMentionTap,
),
CustomEmojiMd(customEmoji, size: inlineCustomEmojiSize),
CustomEmojiMd(
customEmoji,
content: finalContent,
size: inlineCustomEmojiSize,
),
_ChannelLinkMd(
channelNames: resolvedChannelNames,
onChannelTap: resolvedChannelTap,

32
mobile/lib/shared/custom_emoji/custom_emoji_render.dart
View File

@@ -57,8 +57,36 @@ class CustomEmojiMd extends InlineMd {
final double size;
late final RegExp _exp = _buildPattern(_urlByShortcode.keys);
CustomEmojiMd(List<CustomEmoji> palette, {this.size = kCustomEmojiInlineSize})
: _urlByShortcode = {for (final e in palette) e.shortcode: e.url};
/// Only include shortcodes present in the rendered [content]. gpt_markdown
/// embeds this pattern in a combined regex for every parsed text segment, so
/// unrelated community emoji must not make every message expensive to parse.
CustomEmojiMd(
List<CustomEmoji> palette, {
required String content,
this.size = kCustomEmojiInlineSize,
}) : _urlByShortcode = _referencedUrls(palette, content);
// Look ahead so adjacent tokens sharing a colon are both considered:
// :unknown:known: must still allow the known token to match.
static final _shortcodeScan = RegExp(
r'(?=:([a-z0-9_-]+):)',
caseSensitive: false,
);
static Map<String, String> _referencedUrls(
List<CustomEmoji> palette,
String content,
) {
final referenced = {
for (final match in _shortcodeScan.allMatches(content))
match.group(1)!.toLowerCase(),
};
if (referenced.isEmpty) return const {};
return {
for (final emoji in palette)
if (referenced.contains(emoji.shortcode)) emoji.shortcode: emoji.url,
};
}
@override
RegExp get exp => _exp;

20
mobile/lib/shared/relay/relay_rate_limit_gate.dart
View File

@@ -7,7 +7,7 @@ typedef RelayTimerFactory =
/// Session-owned gate that pauses relay requests after back-pressure.
class RelayRateLimitGate {
/// Default gate duration when the relay omits a positive retry hint.
/// Default gate duration when the relay omits a parseable retry hint.
static const defaultRetrySeconds = 10;
/// Longest retry hint accepted from a relay response.
@@ -32,10 +32,22 @@ class RelayRateLimitGate {
}
/// Activates or extends the gate without shrinking an existing window.
///
/// A null [retryInSeconds] means the relay sent no parseable hint, so the
/// conservative [defaultRetrySeconds] applies. An explicit hint is honored as
/// given: the relay sends `retry in 0s` to mean "retry immediately", so a
/// non-positive hint opens no window at all. Treating `0` as if it were
/// absent previously gated every relay read for [defaultRetrySeconds] — on a
/// cold start with many channels that turned the relay's own "go now" into a
/// ten-second stall of the whole session.
///
/// An already-active window is never shortened, so an immediate hint arriving
/// mid-window leaves that window intact.
void activate(int? retryInSeconds) {
final seconds = retryInSeconds != null && retryInSeconds > 0
? min(retryInSeconds, maxRetrySeconds)
: defaultRetrySeconds;
if (retryInSeconds != null && retryInSeconds <= 0) return;
final seconds = retryInSeconds == null
? defaultRetrySeconds
: min(retryInSeconds, maxRetrySeconds);
final duration = Duration(seconds: seconds);
final newExpiry = _now().add(duration);
final currentExpiry = _expiresAt;

2
mobile/lib/shared/relay/relay_session.dart
View File

@@ -737,7 +737,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
final retrySeconds = parseRateLimitRetrySeconds(message);
_rateLimitGate.activate(retrySeconds);
final fallbackMs =
(retrySeconds != null && retrySeconds > 0
(retrySeconds != null
? min(retrySeconds, RelayRateLimitGate.maxRetrySeconds)
: RelayRateLimitGate.defaultRetrySeconds) *
1000;

2
mobile/pubspec.lock
View File

@@ -354,7 +354,7 @@ packages:
source: hosted
version: "0.3.12"
fake_async:
dependency: transitive
dependency: "direct dev"
description:
name: fake_async
sha256: "5368f224a74523e8d2e7399ea1638b37aecfca824a3cc4dfdf77bf1fa905ac44"

1
mobile/pubspec.yaml
View File

@@ -51,6 +51,7 @@ dev_dependencies:
flutter_test:
sdk: flutter
flutter_lints: ^6.0.0
fake_async: ^1.3.3
camera_platform_interface: ^2.13.1
crypto: ^3.0.7
custom_lint: ^0.8.0

415
mobile/test/features/channels/channels_provider_live_cases.dart Normal file
View File

@@ -0,0 +1,415 @@
part of 'channels_provider_test.dart';
void _liveSubscriptionTests() {
const myPk = 'me';
test(
'subscribes once for joined non-archived channels without blocking snapshot',
() async {
final session = _FakeRelaySession(
memberships: [
_membership(_channelA, myPk),
_membership(_channelB, myPk),
_membership(_channelD, myPk),
],
metadata: [
_meta(id: _channelA, name: 'general'),
_meta(id: _channelB, name: 'random'),
// channelD metadata missing -> won't appear in channel list
],
);
session.pauseNextSubscribe();
final container = _buildContainer(session: session);
addTearDown(container.dispose);
final channels = await container.read(channelsProvider.future);
expect(channels.map((channel) => channel.id).toSet(), {
_channelA,
_channelB,
});
await session.nextSubscribeStarted;
expect(session.subscribeFilters, hasLength(1));
expect(session.subscribeFilters.single.tags['#h'], [
_channelA,
_channelB,
]);
expect(
session.subscribeFilters.single.kinds,
EventKind.channelEventKinds,
);
expect(session.subscribeFilters.single.limit, 0);
session.resumePausedSubscribe();
await _waitUntil(() => session.activeChannels.length == 2);
},
);
test('chunks live subscriptions at the relay explicit-channel cap', () async {
final channelIds = [for (var i = 0; i < 257; i++) _generatedChannelId(i)];
final session = _FakeRelaySession(
memberships: [for (final id in channelIds) _membership(id, myPk)],
metadata: [for (final id in channelIds) _meta(id: id, name: id)],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
final channels = await container.read(channelsProvider.future);
expect(channels, hasLength(257));
await _waitUntil(() => session.subscribeFilters.length == 3);
expect(
session.subscribeFilters.map((filter) => filter.tags['#h']!.length),
[128, 128, 1],
);
expect(
session.subscribeFilters.expand((filter) => filter.tags['#h']!).toSet(),
channelIds.toSet(),
);
});
test(
'refreshing an unchanged channel set issues zero new live REQs',
() async {
final session = _FakeRelaySession(
memberships: [
_membership(_channelA, myPk),
_membership(_channelB, myPk),
],
metadata: [
_meta(id: _channelA, name: 'general'),
_meta(id: _channelB, name: 'random'),
],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
await _waitUntil(() => session.activeSubscriptionCount == 1);
final initialSubscribeCount = session.totalSubscribeCount;
await container.read(channelsProvider.notifier).refresh();
await _settle();
expect(session.totalSubscribeCount, initialSubscribeCount);
expect(session.unsubscribeCount, 0);
expect(session.subscribeFilters, hasLength(1));
},
);
test('changing a live chunk replaces only that chunk', () async {
final channelIds = [for (var i = 0; i < 129; i++) _generatedChannelId(i)];
final addedId = _generatedChannelId(999);
final session = _FakeRelaySession(
memberships: [for (final id in channelIds) _membership(id, myPk)],
metadata: [for (final id in channelIds) _meta(id: id, name: id)],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
await _waitUntil(() => session.activeSubscriptionCount == 2);
session.memberships = [
for (final id in [...channelIds.take(128), addedId])
_membership(id, myPk),
];
session.metadata = [
for (final id in [...channelIds.take(128), addedId])
_meta(id: id, name: id),
];
await container.read(channelsProvider.notifier).refresh();
await _waitUntil(() => session.totalSubscribeCount == 3);
expect(session.activeSubscriptionCount, 2);
expect(session.unsubscribeCount, 1);
expect(session.activeChannels, {...channelIds.take(128), addedId});
});
test(
'front-sorting membership insertion retains coverage while replacement waits',
() async {
final channelIds = [
for (var i = 1; i <= 129; i++) _generatedChannelId(i),
];
final addedId = _generatedChannelId(0);
final session = _FakeRelaySession(
memberships: [for (final id in channelIds) _membership(id, myPk)],
metadata: [for (final id in channelIds) _meta(id: id, name: id)],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
await _waitUntil(() => session.activeSubscriptionCount == 2);
session.memberships.add(_membership(addedId, myPk));
session.metadata.add(_meta(id: addedId, name: addedId));
session.pauseNextSubscribe();
await container.read(channelsProvider.notifier).refresh();
await session.nextSubscribeStarted;
try {
expect(session.activeChannels, channelIds.toSet());
expect(session.unsubscribeCount, 0);
session.emit(_liveMessage(channelIds.last));
expect(
container
.read(channelsProvider)
.requireValue
.firstWhere((channel) => channel.id == channelIds.last)
.lastMessageAt
?.millisecondsSinceEpoch,
20000,
);
} finally {
session.resumePausedSubscribe();
}
await _waitUntil(() => session.unsubscribeCount == 2);
expect(session.totalSubscribeCount, 4);
expect(session.activeSubscriptionCount, 2);
expect(session.activeChannels, {...channelIds, addedId});
},
);
test(
'failed replacement keeps desired coverage and ignores departed channels',
() async {
final session = _FakeRelaySession(
memberships: [
_membership(_channelA, myPk),
_membership(_channelB, myPk),
],
metadata: [
_meta(id: _channelA, name: 'a'),
_meta(id: _channelB, name: 'b'),
],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
await _waitUntil(() => session.activeSubscriptionCount == 1);
session.memberships = [
_membership(_channelB, myPk),
_membership(_channelD, myPk),
];
session.metadata = [
_meta(id: _channelB, name: 'b'),
_meta(id: _channelD, name: 'd'),
];
session.subscribeFailures = 1;
await container.read(channelsProvider.notifier).refresh();
await _settle();
expect(session.activeChannels, {_channelA, _channelB});
expect(session.unsubscribeCount, 0);
final requests = session.membershipRequestCount;
session.emit(_liveMessage(_channelA));
await _settle();
expect(session.membershipRequestCount, requests);
session.emit(_liveMessage(_channelB));
expect(
container
.read(channelsProvider)
.requireValue
.firstWhere((channel) => channel.id == _channelB)
.lastMessageAt
?.millisecondsSinceEpoch,
20000,
);
await container.read(channelsProvider.notifier).refresh();
await _waitUntil(() => session.unsubscribeCount == 1);
expect(session.activeChannels, {_channelB, _channelD});
expect(session.activeSubscriptionCount, 1);
},
);
test(
'terminal closure preserves fallback until reconnect retries coverage',
() async {
final channelIds = [
for (var i = 1; i <= 129; i++) _generatedChannelId(i),
];
final addedId = _generatedChannelId(0);
final session = _FakeRelaySession(
memberships: [for (final id in channelIds) _membership(id, myPk)],
metadata: [for (final id in channelIds) _meta(id: id, name: id)],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
await _waitUntil(() => session.activeSubscriptionCount == 2);
session.memberships.add(_membership(addedId, myPk));
session.metadata.add(_meta(id: addedId, name: addedId));
session.subscribeFailures = 1;
await container.read(channelsProvider.notifier).refresh();
await _settle();
expect(session.activeChannels, containsAll(channelIds));
session.closeSubscriptionContaining(
channelIds.last,
'restricted: terminal closure',
);
await _settle();
// The rejected chunk itself is gone, but the previously retained
// fallback must still cover its original first 128 channels.
expect(session.activeChannels, channelIds.take(128).toSet());
expect(session.activeChannels, isNot(contains(addedId)));
session.setStatus(SessionStatus.disconnected);
session.setStatus(SessionStatus.connected);
await _waitUntil(() => session.activeChannels.contains(addedId));
expect(session.activeChannels, containsAll({...channelIds, addedId}));
expect(session.activeSubscriptionCount, 2);
},
);
test(
'partial replacement churn keeps complete coverage with bounded fallbacks',
() async {
var channelIds = [for (var i = 100; i < 356; i++) _generatedChannelId(i)];
final session = _FakeRelaySession(
memberships: [for (final id in channelIds) _membership(id, myPk)],
metadata: [for (final id in channelIds) _meta(id: id, name: id)],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
await _waitUntil(() => session.activeSubscriptionCount == 2);
for (var cycle = 0; cycle < 7; cycle++) {
final addedId = _generatedChannelId(99 - cycle);
channelIds = [addedId, ...channelIds.take(255)];
session.memberships = [
for (final id in channelIds) _membership(id, myPk),
];
session.metadata = [
for (final id in channelIds) _meta(id: id, name: id),
];
session.successfulSubscribesBeforeFailure = 1;
session.subscribeFailures = 1;
await container.read(channelsProvider.notifier).refresh();
await _settle();
expect(session.activeChannels, containsAll(channelIds));
expect(session.activeSubscriptionCount, lessThanOrEqualTo(3));
}
},
);
test(
'retired replacement cleans up even when the next generation disconnects',
() async {
final session = _FakeRelaySession(
memberships: [_membership(_channelA, myPk)],
metadata: [_meta(id: _channelA, name: 'a')],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
await _waitUntil(() => session.activeSubscriptionCount == 1);
session.memberships.add(_membership(_channelB, myPk));
session.metadata.add(_meta(id: _channelB, name: 'b'));
session.pauseNextSubscribe();
await container.read(channelsProvider.notifier).refresh();
await session.nextSubscribeStarted;
try {
expect(session.activeChannels, {_channelA});
session.memberships = [];
session.metadata = [];
await container.read(channelsProvider.notifier).refresh();
session.setStatus(SessionStatus.disconnected);
} finally {
session.resumePausedSubscribe();
}
await _waitUntil(() => session.unsubscribeCount == 2);
expect(session.activeChannels, isEmpty);
expect(session.activeSubscriptionCount, 0);
},
);
test('disposal retires retained and in-flight replacement chunks', () async {
final session = _FakeRelaySession(
memberships: [_membership(_channelA, myPk)],
metadata: [_meta(id: _channelA, name: 'a')],
);
final container = _buildContainer(session: session);
await container.read(channelsProvider.future);
await _waitUntil(() => session.activeSubscriptionCount == 1);
session.memberships.add(_membership(_channelB, myPk));
session.metadata.add(_meta(id: _channelB, name: 'b'));
session.pauseNextSubscribe();
await container.read(channelsProvider.notifier).refresh();
await session.nextSubscribeStarted;
container.dispose();
session.resumePausedSubscribe();
await _waitUntil(() => session.unsubscribeCount == 2);
expect(session.activeSubscriptionCount, 0);
});
test('empty channel refresh removes every retained live chunk', () async {
final session = _FakeRelaySession(
memberships: [_membership(_channelA, myPk), _membership(_channelB, myPk)],
metadata: [
_meta(id: _channelA, name: 'general'),
_meta(id: _channelB, name: 'random'),
],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
await _waitUntil(() => session.activeSubscriptionCount == 1);
session.memberships = [];
session.metadata = [];
await container.read(channelsProvider.notifier).refresh();
await _waitUntil(() => session.activeSubscriptionCount == 0);
expect(session.activeChannels, isEmpty);
expect(session.unsubscribeCount, 1);
});
test(
'community switch retires a detached old-scope live subscription',
() async {
final session = _FakeRelaySession(
memberships: [_membership(_channelA, myPk)],
metadata: [_meta(id: _channelA, name: 'general')],
);
session.pauseNextSubscribe();
final container = _buildContainer(session: session);
addTearDown(container.dispose);
expect(
(await container.read(channelsProvider.future)).single.id,
_channelA,
);
await session.nextSubscribeStarted;
session.memberships = [_membership(_channelB, myPk)];
session.metadata = [_meta(id: _channelB, name: 'random')];
container
.read(relayConfigProvider.notifier)
.update(baseUrl: 'https://new-community.example');
await container.read(channelsProvider.future);
session.resumePausedSubscribe();
await _waitUntil(() => session.activeChannels.contains(_channelB));
expect(session.activeChannels, {_channelB});
expect(session.activeSubscriptionCount, 1);
expect(session.unsubscribeCount, 1);
},
);
}
NostrEvent _liveMessage(String channelId) => NostrEvent(
id: 'live-$channelId',
pubkey: 'alice',
createdAt: 20,
kind: EventKind.streamMessageV2,
tags: [
['h', channelId],
],
content: 'live update',
sig: 'sig',
);

254
mobile/test/features/channels/channels_provider_terminal_cases.dart Normal file
View File

@@ -0,0 +1,254 @@
part of 'channels_provider_test.dart';
void _testWithClock(String name, void Function(FakeAsync) body) {
test(name, () => fakeAsync(body));
}
void _terminalSubscriptionTests() {
for (final beforeReady in [true, false]) {
_testWithClock(
'persistent terminal rejection ${beforeReady ? 'before' : 'after'} '
'readiness cannot amplify refreshes or retries',
(clock) {
final session = _TerminalRelaySession(beforeReady: beforeReady);
final logs = <String>[];
final originalDebugPrint = debugPrint;
debugPrint = (message, {wrapWidth}) {
if (message != null) logs.add(message);
};
final container = _buildContainer(session: session);
container.read(channelsProvider);
try {
clock.flushMicrotasks();
// The finite snapshot is usable even before live readiness settles.
expect(
container.read(channelsProvider).requireValue.single.id,
_channelA,
);
expect(session.totalSubscribeCount, 1);
final initialMemberships = session.membershipRequestCount;
clock.elapse(const Duration(milliseconds: 2));
final historyAfterClose = session.historyFilters.length;
final batchesAfterClose = session.queryBatches.length;
final logsAfterClose = logs.length;
// Virtual time permits many immediate retries on the broken path.
for (var i = 0; i < 20; i++) {
clock.elapse(const Duration(milliseconds: 10));
}
expect(session.rejectionCount, 1);
expect(session.membershipRequestCount, initialMemberships);
expect(session.historyFilters.length, historyAfterClose);
expect(session.queryBatches.length, batchesAfterClose);
expect(session.totalSubscribeCount, 1);
expect(logs.length, logsAfterClose);
expect(clock.periodicTimerCount, 1);
expect(clock.nonPeriodicTimerCount, 0);
// Ordinary polling may refresh memberships, but never re-admits an
// unchanged terminal filter, creates retry timers, or repeats logs.
final pollStart = session.membershipRequestCount;
for (var i = 0; i < 3; i++) {
clock.elapse(const Duration(seconds: 60));
expect(session.totalSubscribeCount, 1);
expect(session.activeSubscriptionCount, 0);
expect(logs.length, logsAfterClose);
expect(clock.periodicTimerCount, 1);
expect(clock.nonPeriodicTimerCount, 0);
expect(
container.read(channelsProvider).requireValue.single.id,
_channelA,
);
}
expect(session.membershipRequestCount - pollStart, 6);
// A pull-to-refresh still isn't evidence that admission changed.
unawaited(container.read(channelsProvider.notifier).refresh());
clock.flushMicrotasks();
clock.elapse(const Duration(milliseconds: 2));
expect(session.totalSubscribeCount, 1);
expect(logs.length, logsAfterClose);
} finally {
container.dispose();
debugPrint = originalDebugPrint;
}
clock.elapse(const Duration(seconds: 60));
expect(clock.pendingTimers, isEmpty);
expect(session.totalSubscribeCount, 1);
},
);
}
for (final beforeReady in [true, false]) {
_testWithClock(
'disposal fences pending terminal closure ${beforeReady ? 'before' : 'after'} readiness',
(clock) {
final session = _TerminalRelaySession(beforeReady: beforeReady);
final container = _buildContainer(session: session);
container.read(channelsProvider);
clock.flushMicrotasks();
final requests = session.membershipRequestCount;
container.dispose();
clock.elapse(const Duration(minutes: 3));
expect(session.totalSubscribeCount, 1);
expect(session.membershipRequestCount, requests);
expect(session.activeSubscriptionCount, 0);
expect(clock.pendingTimers, isEmpty);
},
);
}
_testWithClock('terminal rejection retries only after a new connection', (
clock,
) {
final session = _TerminalRelaySession();
final container = _buildContainer(session: session);
try {
container.read(channelsProvider);
clock.flushMicrotasks();
clock.elapse(const Duration(milliseconds: 2));
final oldClosed = session.closedCallbacks.single;
for (var attempt = 2; attempt <= 4; attempt++) {
session.setStatus(SessionStatus.disconnected);
clock.flushMicrotasks();
session.setStatus(SessionStatus.connected);
clock.flushMicrotasks();
clock.elapse(const Duration(milliseconds: 2));
clock.elapse(const Duration(milliseconds: 20));
expect(session.totalSubscribeCount, attempt);
expect(session.rejectionCount, attempt);
expect(session.activeSubscriptionCount, 0);
expect(
container.read(channelsProvider).requireValue.single.id,
_channelA,
);
}
session.reject = false;
session.setStatus(SessionStatus.disconnected);
clock.flushMicrotasks();
session.setStatus(SessionStatus.connected);
clock.flushMicrotasks();
expect(session.totalSubscribeCount, 5);
expect(session.activeChannels, {_channelA});
oldClosed();
clock.flushMicrotasks();
expect(session.activeChannels, {_channelA});
expect(session.totalSubscribeCount, 5);
} finally {
container.dispose();
}
clock.flushMicrotasks();
});
_testWithClock(
'changed membership admits a new filter, not a rejected old one',
(clock) {
final session = _TerminalRelaySession();
final container = _buildContainer(session: session);
try {
container.read(channelsProvider);
clock.flushMicrotasks();
clock.elapse(const Duration(milliseconds: 2));
session.reject = false;
session.memberships.add(_membership(_channelB, 'me'));
session.metadata.add(_meta(id: _channelB, name: 'b'));
unawaited(container.read(channelsProvider.notifier).refresh());
clock.flushMicrotasks();
expect(session.totalSubscribeCount, 2);
expect(session.activeChannels, {_channelA, _channelB});
// Removing then rejoining is a meaningful transition even if it returns
// to an earlier filter; obsolete quarantine keys must not leak forever.
session.memberships.removeLast();
unawaited(container.read(channelsProvider.notifier).refresh());
clock.flushMicrotasks();
expect(session.totalSubscribeCount, 3);
expect(session.activeChannels, {_channelA});
} finally {
container.dispose();
}
clock.flushMicrotasks();
},
);
for (final switchIdentity in [true, false]) {
_testWithClock(
'terminal quarantine does not cross ${switchIdentity ? 'identity' : 'community'} scope',
(clock) {
final session = _TerminalRelaySession();
final container = _buildContainer(session: session);
try {
container.read(channelsProvider);
clock.flushMicrotasks();
clock.elapse(const Duration(milliseconds: 2));
session.reject = false;
if (switchIdentity) {
session.memberships = [_membership(_channelA, _otherPk)];
container.read(_testPubkeyProvider.notifier).set(_otherPk);
} else {
container
.read(relayConfigProvider.notifier)
.update(baseUrl: 'https://new-community.example');
}
container.read(channelsProvider);
clock.flushMicrotasks();
expect(session.totalSubscribeCount, 2);
expect(session.activeChannels, {_channelA});
} finally {
container.dispose();
}
clock.flushMicrotasks();
},
);
}
}
/// Reject every attempted subscription, including every replacement, as a
/// terminal relay CLOSED. A timer models the wire turn and prevents a broken
/// immediate retry loop from hanging the test's microtask drain.
class _TerminalRelaySession extends _FakeRelaySession {
_TerminalRelaySession({this.beforeReady = false})
: super(
memberships: [_membership(_channelA, 'me')],
metadata: [_meta(id: _channelA, name: 'a')],
);
final bool beforeReady;
bool reject = true;
int rejectionCount = 0;
final List<void Function()> closedCallbacks = [];
@override
Future<void Function()> subscribe(
NostrFilter filter,
void Function(NostrEvent) onEvent, {
void Function(String message)? onClosed,
}) async {
final unsubscribe = await super.subscribe(
filter,
onEvent,
onClosed: onClosed,
);
if (!reject) return unsubscribe;
final key = _subscriptions.keys.last;
final closed = Completer<void>();
final timer = Timer(const Duration(milliseconds: 1), () {
final subscription = _subscriptions.remove(key);
if (subscription != null) {
subscribeFilters.remove(subscription.$1);
rejectionCount++;
void notify() => onClosed?.call('restricted: persistent rejection');
closedCallbacks.add(notify);
notify();
}
closed.complete();
});
if (beforeReady) {
await closed.future;
throw StateError('restricted: persistent rejection');
}
return () {
timer.cancel();
unsubscribe();
};
}
}

260
mobile/test/features/channels/channels_provider_test.dart
View File

@@ -1,5 +1,6 @@
import 'dart:async';
import 'package:fake_async/fake_async.dart';
import 'package:flutter/widgets.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
@@ -7,12 +8,15 @@ import 'package:buzz/features/channels/channel_management_provider.dart';
import 'package:buzz/features/channels/channels_provider.dart';
import 'package:buzz/shared/relay/relay.dart';
part 'channels_provider_live_cases.dart';
part 'channels_provider_terminal_cases.dart';
/// Tests for [ChannelsNotifier] in the pure-Nostr world.
///
/// The provider loads membership-backed channels first:
/// 1. paginated kind:39002 memberships tagged `#p:<my-pubkey>`
/// 2. kind:39000 metadata for those channel ids
/// then layers per-channel live subscriptions on the `#h` tag. Browse channels
/// then layers chunked live subscriptions on the `#h` tag. Browse channels
/// separately triggers paginated kind:39000 open-channel discovery.
///
/// Tests stub out the relay session by overriding [relaySessionProvider] with
@@ -545,13 +549,15 @@ void main() {
.retryDirectory();
await session.nextSubscribeStarted;
// Record every list emitted from the switch onward. The live-subscription
// queue is serialized, so community B's own subscribe waits behind the
// parked one. That makes the observable defect an emission of community
// A's channel into the current scope, not just a wrong final state.
await staleRefresh;
await _settle();
// The old scope's finite snapshot is allowed to publish before its live
// subscription becomes ready. From the actual switch onward, however,
// neither that snapshot nor its detached live work may republish.
final emitted = <List<String>>[];
container.listen(channelsProvider, (previous, next) {
final value = next.value;
final value = next.asData?.value;
if (value != null) {
emitted.add(value.map((channel) => channel.id).toList());
}
@@ -562,9 +568,9 @@ void main() {
container
.read(relayConfigProvider.notifier)
.update(baseUrl: 'https://community-b.example');
await container.read(channelsProvider.future);
session.resumePausedSubscribe();
await staleRefresh;
await _settle();
expect(
@@ -1535,6 +1541,7 @@ void main() {
expect(channels.map((channel) => channel.id), [_channelA, _channelB]);
expect(channels.first.isMember, isTrue);
expect(channels.last.isMember, isFalse);
await _waitUntil(() => session.subscribeFilters.length == 1);
expect(session.subscribeFilters, hasLength(1));
});
@@ -1574,38 +1581,8 @@ void main() {
},
);
test(
'subscribes per-channel with #h tags (only joined, non-archived)',
() async {
final session = _FakeRelaySession(
memberships: [
_membership(_channelA, myPk),
_membership(_channelB, myPk),
_membership(_channelD, myPk),
],
metadata: [
_meta(id: _channelA, name: 'general'),
_meta(id: _channelB, name: 'random'),
// channelD metadata missing -> won't appear in channel list
],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
// One subscription per joined, non-archived channel.
expect(session.subscribeFilters, hasLength(2));
expect(
session.subscribeFilters.map((f) => f.tags['#h']?.single).toSet(),
{_channelA, _channelB},
);
for (final filter in session.subscribeFilters) {
expect(filter.kinds, EventKind.channelEventKinds);
expect(filter.limit, 0);
}
},
);
_liveSubscriptionTests();
_terminalSubscriptionTests();
test('retains channel-list member snapshots for immediate reuse', () async {
final joinedAt = DateTime.fromMillisecondsSinceEpoch(1000, isUtc: true);
@@ -1626,174 +1603,6 @@ void main() {
expect(members.every((member) => member.joinedAt == joinedAt), isTrue);
});
test(
'refreshing an unchanged channel set issues zero new live REQs',
() async {
final session = _FakeRelaySession(
memberships: [
_membership(_channelA, myPk),
_membership(_channelB, myPk),
],
metadata: [
_meta(id: _channelA, name: 'general'),
_meta(id: _channelB, name: 'random'),
],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
final initialSubscribeCount = session.totalSubscribeCount;
await container.read(channelsProvider.notifier).refresh();
expect(session.totalSubscribeCount, initialSubscribeCount);
expect(session.unsubscribeCount, 0);
expect(session.subscribeFilters, hasLength(2));
},
);
test(
'live subscription diff only removes and adds changed channels',
() async {
final session = _FakeRelaySession(
memberships: [
_membership(_channelA, myPk),
_membership(_channelB, myPk),
],
metadata: [
_meta(id: _channelA, name: 'general'),
_meta(id: _channelB, name: 'random'),
],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
session.memberships = [
_membership(_channelB, myPk),
_membership(_channelD, myPk),
];
session.metadata = [
_meta(id: _channelB, name: 'random'),
_meta(id: _channelD, name: 'support'),
];
await container.read(channelsProvider.notifier).refresh();
expect(session.totalSubscribeCount, 3);
expect(session.unsubscribeCount, 1);
expect(
session.subscribeFilters
.map((filter) => filter.tags['#h']!.single)
.toSet(),
{_channelB, _channelD},
);
},
);
test(
'empty channel refresh removes every retained live subscription',
() async {
final session = _FakeRelaySession(
memberships: [
_membership(_channelA, myPk),
_membership(_channelB, myPk),
],
metadata: [
_meta(id: _channelA, name: 'general'),
_meta(id: _channelB, name: 'random'),
],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
session.memberships = [];
session.metadata = [];
await container.read(channelsProvider.notifier).refresh();
expect(session.activeChannels, isEmpty);
expect(session.activeSubscriptionCount, 0);
expect(session.unsubscribeCount, 2);
},
);
test(
'overlapping refreshes retain one live subscription per desired channel',
() async {
final session = _FakeRelaySession(
memberships: [
_membership(_channelA, myPk),
_membership(_channelB, myPk),
],
metadata: [
_meta(id: _channelA, name: 'general'),
_meta(id: _channelB, name: 'random'),
],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
session.pauseNextSubscribe();
session.memberships = [
_membership(_channelA, myPk),
_membership(_channelB, myPk),
_membership(_channelD, myPk),
];
session.metadata = [
_meta(id: _channelA, name: 'general'),
_meta(id: _channelB, name: 'random'),
_meta(id: _channelD, name: 'support'),
];
final firstRefresh = container.read(channelsProvider.notifier).refresh();
await session.nextSubscribeStarted;
final secondRefresh = container.read(channelsProvider.notifier).refresh();
session.resumePausedSubscribe();
await Future.wait([firstRefresh, secondRefresh]);
expect(session.activeChannels, {_channelA, _channelB, _channelD});
expect(session.activeSubscriptionCount, 3);
},
);
test(
'community switch replaces retained live subscriptions on the new relay',
() async {
final session = _FakeRelaySession(
memberships: [_membership(_channelA, myPk)],
metadata: [_meta(id: _channelA, name: 'general')],
);
final container = _buildContainer(session: session);
addTearDown(container.dispose);
await container.read(channelsProvider.future);
expect(session.activeChannels, {_channelA});
session.setStatus(SessionStatus.disconnected);
session.memberships = [_membership(_channelB, myPk)];
session.metadata = [_meta(id: _channelB, name: 'random')];
container
.read(relayConfigProvider.notifier)
.update(baseUrl: 'https://new-community.example');
await Future<void>.delayed(Duration.zero);
session.setStatus(SessionStatus.connected);
await container.read(channelsProvider.future);
await _waitUntil(
() =>
session.activeChannels.length == 1 &&
session.activeChannels.contains(_channelB),
);
expect(session.activeChannels, {_channelB});
expect(session.activeSubscriptionCount, 1);
expect(session.unsubscribeCount, 1);
},
);
test('live channel events update channel lastMessageAt', () async {
final session = _FakeRelaySession(
memberships: [_membership(_channelA, myPk)],
@@ -2196,6 +2005,7 @@ void main() {
final initial = await container.read(channelsProvider.future);
expect(initial.single.name, 'general');
await _waitUntil(() => session.subscribeFilters.length == 1);
expect(session.subscribeFilters, hasLength(1));
session.setStatus(SessionStatus.reconnecting);
@@ -2296,6 +2106,7 @@ void main() {
);
// And one live subscription on the resulting channel.
await _waitUntil(() => session.subscribeFilters.length == 1);
expect(session.subscribeFilters, hasLength(1));
});
}
@@ -2305,6 +2116,9 @@ const _channelB = '22222222-2222-4222-8222-222222222222';
const _channelD = '44444444-4444-4444-8444-444444444444';
const _otherPk = 'someone-else';
String _generatedChannelId(int index) =>
'${index.toString().padLeft(8, '0')}-0000-4000-8000-000000000000';
/// Build a kind:39002 membership event tagged with the channel id and member.
NostrEvent _membership(
String channelId,
@@ -2450,7 +2264,11 @@ class _FakeRelaySession extends RelaySessionNotifier {
final List<NostrFilter> directoryQueryFilters = [];
final List<NostrFilter> membershipQueryFilters = [];
final List<NostrFilter> subscribeFilters = [];
final Map<int, (NostrFilter, void Function(NostrEvent))> _subscriptions = {};
final Map<
int,
(NostrFilter, void Function(NostrEvent), void Function(String message)?)
>
_subscriptions = {};
int _nextSubscriptionKey = 0;
Completer<void>? _pausedSubscribe;
Completer<void>? _subscribeStarted;
@@ -2470,9 +2288,12 @@ class _FakeRelaySession extends RelaySessionNotifier {
Completer<void>? _claimedUnreadCatchUp;
int unsubscribeCount = 0;
int totalSubscribeCount = 0;
int subscribeFailures = 0;
int successfulSubscribesBeforeFailure = 0;
Set<String> get activeChannels => {
for (final (filter, _) in _subscriptions.values) ?filter.tags['#h']?.single,
for (final (filter, _, _) in _subscriptions.values)
...filter.tags['#h'] ?? const <String>[],
};
int get activeSubscriptionCount => _subscriptions.length;
@@ -2850,8 +2671,16 @@ class _FakeRelaySession extends RelaySessionNotifier {
_pausedSubscribe = null;
_subscribeStarted = null;
}
if (subscribeFailures > 0 && successfulSubscribesBeforeFailure == 0) {
subscribeFailures--;
subscribeFilters.remove(filter);
throw StateError('live subscription failed');
}
if (successfulSubscribesBeforeFailure > 0) {
successfulSubscribesBeforeFailure--;
}
final subscriptionKey = ++_nextSubscriptionKey;
_subscriptions[subscriptionKey] = (filter, onEvent);
_subscriptions[subscriptionKey] = (filter, onEvent, onClosed);
return () {
final subscription = _subscriptions.remove(subscriptionKey);
if (subscription == null) return;
@@ -2860,13 +2689,22 @@ class _FakeRelaySession extends RelaySessionNotifier {
};
}
void closeSubscriptionContaining(String channelId, String message) {
final entry = _subscriptions.entries.singleWhere(
(entry) => entry.value.$1.tags['#h']?.contains(channelId) ?? false,
);
_subscriptions.remove(entry.key);
subscribeFilters.remove(entry.value.$1);
entry.value.$3?.call(message);
}
void setStatus(SessionStatus status) {
state = SessionState(status: status);
}
/// Emit a live event to all subscribers.
void emit(NostrEvent event) {
for (final (_, listener) in List.of(_subscriptions.values)) {
for (final (_, listener, _) in List.of(_subscriptions.values)) {
listener(event);
}
}

121
mobile/test/features/channels/message_content_custom_emoji_test.dart Normal file
View File

@@ -0,0 +1,121 @@
import 'package:buzz/features/channels/message_content.dart';
import 'package:buzz/shared/custom_emoji/custom_emoji.dart';
import 'package:buzz/shared/custom_emoji/custom_emoji_provider.dart';
import 'package:buzz/shared/custom_emoji/custom_emoji_render.dart';
import 'package:buzz/shared/theme/theme.dart';
import 'package:flutter/material.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:gpt_markdown/gpt_markdown.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
Widget _testable(String content, {List<List<String>> tags = const []}) {
return ProviderScope(
overrides: [
customEmojiListProvider.overrideWithValue([
const CustomEmoji(
shortcode: 'wave',
url: 'https://example.com/wave.png',
),
for (var i = 0; i < 2500; i++)
CustomEmoji(
shortcode: 'unused_$i',
url: 'https://example.com/$i.png',
),
]),
],
child: MaterialApp(
theme: AppTheme.light(),
home: Scaffold(
body: MessageContent(
content: content,
tags: tags,
channelNames: const {'test': 'test-channel'},
),
),
),
);
}
void main() {
testWidgets('message wiring excludes unrelated emoji from the regex', (
tester,
) async {
await tester.pumpWidget(_testable('**hello** :unknown:'));
final markdown = tester.widget<GptMarkdown>(find.byType(GptMarkdown));
final component = markdown.inlineComponents!
.whereType<CustomEmojiMd>()
.single;
expect(component.exp.hasMatch(':unused_2499:'), isFalse);
expect(component.exp.hasMatch(':wave:'), isFalse);
expect(find.byType(CustomEmojiImage), findsNothing);
final text = tester
.widgetList<RichText>(find.byType(RichText))
.map((widget) => widget.text.toPlainText())
.join();
expect(text, contains('hello'));
expect(text, contains(':unknown:'));
expect(text, isNot(contains('**hello**')));
});
testWidgets('referenced event emoji stays available with tag URL priority', (
tester,
) async {
await tester.pumpWidget(
_testable(
':wave:',
tags: const [
['emoji', 'wave', 'https://example.com/event-wave.png'],
],
),
);
final image = tester.widget<CustomEmojiImage>(
find.byType(CustomEmojiImage),
);
expect(image.shortcode, 'wave');
expect(image.url, 'https://example.com/event-wave.png');
final markdown = tester.widget<GptMarkdown>(find.byType(GptMarkdown));
final component = markdown.inlineComponents!
.whereType<CustomEmojiMd>()
.single;
expect(component.exp.hasMatch(':unused_2499:'), isFalse);
expect(component.exp.hasMatch(':wave:'), isTrue);
});
testWidgets('content edits rebuild the scoped matcher', (tester) async {
await tester.pumpWidget(_testable('plain message'));
expect(find.byType(CustomEmojiImage), findsNothing);
await tester.pumpWidget(_testable('edited :wave:'));
expect(
tester.widget<CustomEmojiImage>(find.byType(CustomEmojiImage)).shortcode,
'wave',
);
await tester.pumpWidget(_testable('edited :unused_2499:'));
expect(
tester.widget<CustomEmojiImage>(find.byType(CustomEmojiImage)).shortcode,
'unused_2499',
);
final markdown = tester.widget<GptMarkdown>(find.byType(GptMarkdown));
final component = markdown.inlineComponents!
.whereType<CustomEmojiMd>()
.single;
expect(component.exp.hasMatch(':wave:'), isFalse);
await tester.pumpWidget(_testable('edited :unknown:'));
expect(find.byType(CustomEmojiImage), findsNothing);
});
testWidgets('code keeps literal emoji while adjacent known tokens render', (
tester,
) async {
await tester.pumpWidget(_testable('`:wave:` :unknown:wave:'));
expect(find.byType(CustomEmojiImage), findsOneWidget);
final text = tester
.widgetList<RichText>(find.byType(RichText))
.map((widget) => widget.text.toPlainText())
.join();
expect(text, contains(':wave:'));
expect(text, contains(':unknown'));
});
}

109
mobile/test/shared/custom_emoji/custom_emoji_render_test.dart Normal file
View File

@@ -0,0 +1,109 @@
import 'package:buzz/shared/custom_emoji/custom_emoji.dart';
import 'package:buzz/shared/custom_emoji/custom_emoji_render.dart';
import 'package:flutter/material.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:gpt_markdown/custom_widgets/markdown_config.dart';
const _palette = [
CustomEmoji(shortcode: 'wave', url: 'https://example.com/wave.png'),
CustomEmoji(shortcode: 'wave_long', url: 'https://example.com/long.png'),
CustomEmoji(shortcode: 'party-parrot', url: 'https://example.com/parrot.png'),
];
void main() {
test('pattern is bounded by referenced emoji, not the community palette', () {
final largePalette = [
..._palette,
for (var i = 0; i < 2500; i++)
CustomEmoji(shortcode: 'unused_$i', url: 'https://example.com/$i.png'),
];
final small = CustomEmojiMd(_palette, content: 'hello :wave:');
final large = CustomEmojiMd(largePalette, content: 'hello :wave:');
expect(large.exp.pattern, small.exp.pattern);
expect(large.exp.hasMatch(':wave:'), isTrue);
expect(large.exp.hasMatch(':wave_long:'), isFalse);
expect(large.exp.hasMatch(':unused_2499:'), isFalse);
});
test(
'no references and unknown references produce a nonmatching pattern',
() {
for (final content in [
'hello world',
':unknown:',
'https://example.com',
]) {
final matcher = CustomEmojiMd(_palette, content: content);
expect(matcher.exp.allMatches(content), isEmpty);
expect(matcher.exp.hasMatch(':wave:'), isFalse);
}
expect(
CustomEmojiMd(const [], content: ':wave:').exp.hasMatch(':wave:'),
isFalse,
);
},
);
test('selection preserves the original matcher across token boundaries', () {
final original = RegExp(
':(?:${_palette.map((e) => RegExp.escape(e.shortcode)).join('|')}):',
caseSensitive: false,
);
for (final content in [
':wave:',
':WAVE: :Wave_Long: :PARTY-PARROT:',
':unknown:wave:',
':wave:unknown:wave_long:',
':wave::wave_long:',
':::wave::: :wave_long:! (:party-parrot:)',
':wave_longer: :unknown: no match',
'`code :wave:` **bold :wave_long:**',
]) {
final selected = CustomEmojiMd(_palette, content: content);
expect(
selected.exp.allMatches(content).map((m) => m.group(0)).toList(),
original.allMatches(content).map((m) => m.group(0)).toList(),
reason: content,
);
}
});
testWidgets('known tokens keep their URL and size; unknowns remain text', (
tester,
) async {
await tester.pumpWidget(const MaterialApp(home: SizedBox()));
final context = tester.element(find.byType(SizedBox));
final matcher = CustomEmojiMd(
_palette,
content: ':WAVE: :unknown:',
size: 32,
);
final known = matcher.span(context, ':WAVE:', GptMarkdownConfig());
expect(known, isA<WidgetSpan>());
final image = (known as WidgetSpan).child as CustomEmojiImage;
expect(image.shortcode, 'wave');
expect(image.url, 'https://example.com/wave.png');
expect(image.size, 32);
final unknown = matcher.span(context, ':unknown:', GptMarkdownConfig());
expect(unknown, isA<TextSpan>());
expect((unknown as TextSpan).text, ':unknown:');
});
testWidgets('selection uses the current content and current palette URL', (
tester,
) async {
await tester.pumpWidget(const MaterialApp(home: SizedBox()));
final context = tester.element(find.byType(SizedBox));
final initial = CustomEmojiMd(_palette, content: ':wave:');
final updated = CustomEmojiMd(const [
CustomEmoji(shortcode: 'wave_long', url: 'https://example.com/new.png'),
], content: ':wave_long:');
expect(initial.exp.hasMatch(':wave_long:'), isFalse);
expect(updated.exp.hasMatch(':wave:'), isFalse);
final span = updated.span(context, ':wave_long:', GptMarkdownConfig());
expect(
((span as WidgetSpan).child as CustomEmojiImage).url,
'https://example.com/new.png',
);
});
}

65
mobile/test/shared/relay/relay_rate_limit_gate_test.dart
View File

@@ -4,7 +4,52 @@ import 'package:buzz/shared/relay/relay.dart';
import 'package:flutter_test/flutter_test.dart';
void main() {
test('uses the default for absent and non-positive hints', () {
test('uses the default only for an absent hint', () {
final timers = <_ManualTimer>[];
final gate = RelayRateLimitGate(
now: () => DateTime.utc(2026),
timerFactory: (duration, callback) {
final timer = _ManualTimer(duration, callback);
timers.add(timer);
return timer;
},
);
gate.activate(null);
expect(gate.isActive, isTrue);
expect(gate.remainingMs(), 10000);
expect(timers.single.duration, const Duration(seconds: 10));
});
test('an explicit non-positive hint opens no window', () async {
final timers = <_ManualTimer>[];
final gate = RelayRateLimitGate(
now: () => DateTime.utc(2026),
timerFactory: (duration, callback) {
final timer = _ManualTimer(duration, callback);
timers.add(timer);
return timer;
},
);
// The relay sends `retry in 0s` to mean "retry immediately".
gate.activate(0);
expect(gate.isActive, isFalse);
expect(gate.remainingMs(), 0);
expect(timers, isEmpty);
await gate.wait().timeout(const Duration(seconds: 1));
gate.activate(-5);
expect(gate.isActive, isFalse);
expect(gate.remainingMs(), 0);
expect(timers, isEmpty);
await gate.wait().timeout(const Duration(seconds: 1));
});
test('an immediate hint mid-window leaves the window intact', () async {
var now = DateTime.utc(2026);
final timers = <_ManualTimer>[];
final gate = RelayRateLimitGate(
@@ -16,13 +61,21 @@ void main() {
},
);
gate.activate(null);
expect(gate.remainingMs(), 10000);
expect(timers.single.duration, const Duration(seconds: 10));
gate.activate(30);
final wait = gate.wait();
final armedTimer = timers.single;
now = now.add(const Duration(seconds: 11));
now = now.add(const Duration(seconds: 5));
gate.activate(0);
expect(timers.last.duration, const Duration(seconds: 10));
expect(gate.isActive, isTrue);
expect(gate.remainingMs(), 25000);
expect(timers, hasLength(1));
expect(armedTimer.isActive, isTrue);
armedTimer.fire();
await wait;
expect(gate.isActive, isFalse);
});
test('clamps large hints to five minutes', () {

36
mobile/test/shared/relay/relay_session_test.dart
View File

@@ -1036,6 +1036,42 @@ void main() {
unsubscribe();
});
test('rate-limited live CLOSED honors an immediate retry hint', () async {
final retryTimers = <_ManualTimer>[];
final gateTimers = <_ManualTimer>[];
final gate = RelayRateLimitGate(
timerFactory: (duration, callback) {
final timer = _ManualTimer(duration, callback);
gateTimers.add(timer);
return timer;
},
);
final session = RelaySessionNotifier(
rateLimitGate: gate,
retryTimerFactory: (duration, callback) {
final timer = _ManualTimer(duration, callback);
retryTimers.add(timer);
return timer;
},
);
final socket = _RecordingRelaySocket();
session.debugAttachSocketForTest(socket);
final subscribe = session.subscribe(_channelFilter, (_) {});
session.debugHandleMessage(['EOSE', 'l-1']);
final unsubscribe = await subscribe;
session.debugHandleMessage([
'CLOSED',
'l-1',
'rate-limited: quota exceeded; retry in 0s',
]);
expect(retryTimers.single.duration, const Duration(seconds: 1));
expect(gateTimers, isEmpty);
expect(gate.isActive, isFalse);
unsubscribe();
});
test(
'rate-limited CLOSED retry does not survive a superseded connection',
() async {