jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-28 21:11:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mobile): preserve SEND scope and authority at owner 7536

Browse Source 
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-09-09 22:29:27 -04:00
parent a22b79fa8c
commit 06a016e1cb
8 changed files with 480 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
mobile/lib/features/channels/compose_bar/compose_bar_widget.dart
View File

@@ -485,12 +485,7 @@ class ComposeBar extends HookConsumerWidget {
var authorizationRevision = submittedDraftRevision;
final visit = authorizationVisit.value;
final config = ref.read(relayConfigProvider);
// Equivalent refreshes retain scope; destination/credentials do not.
bool isConfigScopeCurrent() {
final current = ref.read(relayConfigProvider);
return current.baseUrl == config.baseUrl &&
current.nsec == config.nsec;
}
bool isConfigScopeCurrent() => _isComposeConfigCurrent(ref, config);
final readSelected = ref.read(
selectedMentionAuthorizationReaderProvider,
@@ -498,29 +493,20 @@ class ComposeBar extends HookConsumerWidget {
final session = ref.read(relaySessionProvider.notifier);
final observedProfiles = <String, NostrEvent>{};
final observedKeys = <String>{};
bool profilesCurrent() => observedProfiles.entries.every((entry) {
final order = ref
.read(userCacheProvider.notifier)
.profileEventOrder(entry.key);
final event = entry.value;
return order == null ||
order.createdAt < event.createdAt ||
(order.createdAt == event.createdAt &&
order.eventId.compareTo(event.id) >= 0);
});
final evidenceChecks = <bool Function()>{};
bool profilesCurrent() => evidenceChecks.every((check) => check());
bool ownsSource() =>
context.mounted &&
visit == authorizationVisit.value &&
isConfigScopeCurrent();
bool isAuthorizationCurrent() =>
bool isAuthorizationScopeCurrent() =>
ownsSource() &&
identical(session, ref.read(relaySessionProvider.notifier)) &&
currentPubkey == ref.read(currentPubkeyProvider) &&
profilesCurrent() &&
submittedUploadGeneration == uploadGeneration.value &&
authorizationRevision == draftRevision.value &&
isConfigScopeCurrent();
void ensureAuthorizationCurrent() {
void ensureAuthorizationScopeCurrent() {
if (!context.mounted) throw const _ComposeAuthorizationCancelled();
if (!isConfigScopeCurrent()) {
throw const _ComposeCommunityChanged();
@@ -531,12 +517,26 @@ class ComposeBar extends HookConsumerWidget {
throw const _ComposeAuthorizationCancelled();
}
if (!identical(session, ref.read(relaySessionProvider.notifier)) ||
currentPubkey != ref.read(currentPubkeyProvider) ||
!profilesCurrent()) {
currentPubkey != ref.read(currentPubkeyProvider)) {
throw Exception('Mention evidence changed; retry the draft');
}
}
void ensureAuthorizationCurrent() {
ensureAuthorizationScopeCurrent();
if (!profilesCurrent()) {
throw Exception('Mention evidence changed; retry the draft');
}
}
Future<void> guardedDelivery(
String content,
List<String> keys, {
List<List<String>> mediaTags = const [],
}) => withRelayPublicationGuard(
ensureAuthorizationCurrent,
() => onSend(content, keys, mediaTags: mediaTags),
);
checkPreparationCurrent = ensureAuthorizationCurrent;
// Resolved before any await: see
@@ -566,10 +566,12 @@ class ComposeBar extends HookConsumerWidget {
priorAgentKeys: priorAgentKeys,
observedKeys: observedKeys,
observedProfiles: observedProfiles,
evidenceChecks: evidenceChecks,
currentPubkey: currentPubkey,
channelId: channelId,
ensureAuthorizationCurrent: ensureAuthorizationCurrent,
isAuthorizationCurrent: isAuthorizationCurrent,
isAuthorizationCurrent: isAuthorizationScopeCurrent,
ensureScopeCurrent: ensureAuthorizationScopeCurrent,
prepare: prepare,
);
@@ -610,44 +612,21 @@ class ComposeBar extends HookConsumerWidget {
);
final channelActions = ref.read(channelActionsProvider);
// Agent failures stop publication; the original draft keeps its keys.
Future<void> addMentionedNonMembers() async {
final keys = outgoing.pubkeys.toSet();
Future<bool> authorizeWrite(String key, String role) async {
final evidence = await authorize(keys, prepare: true);
final fresh = evidence[key]!;
if (fresh.invitationRole != role) {
throw Exception(
'Mention classification changed; retry invitation consent',
);
}
return !fresh.isMember;
}
await authorize(keys, prepare: true);
ensureAuthorizationCurrent();
invitationStarted.value = true;
await outgoing.addNonMembers(
channelActions,
scan: scan,
messenger: messenger,
ensureCurrent: ensureAuthorizationCurrent,
authorizeWrite: authorizeWrite,
);
if (!outgoing.pubkeys.toSet().containsAll(keys)) {
throw Exception(
'Mention invitation failed. Draft kept; retry or remove the mention.',
);
}
await authorize(keys);
if (queuedAttachments.isEmpty ||
(selectedKeys.isNotEmpty ||
scan.humans.isNotEmpty ||
scan.agentPubkeys.isNotEmpty)) {
ensureAuthorizationCurrent();
}
}
Future<void> addMentionedNonMembers() => _prepareMentionInvitations(
outgoing: outgoing,
scan: scan,
channelActions: channelActions,
messenger: messenger,
authorize: authorize,
ensureCurrent: ensureAuthorizationCurrent,
ensureScopeCurrent: ensureAuthorizationScopeCurrent,
onStarted: () => invitationStarted.value = true,
fenceAfterPreparation:
queuedAttachments.isEmpty ||
selectedKeys.isNotEmpty ||
scan.humans.isNotEmpty ||
scan.agentPubkeys.isNotEmpty,
);
if (queuedAttachments.isEmpty) {
if (!context.mounted) return;
await _sendTextOnlyDraft(
@@ -658,7 +637,10 @@ class ComposeBar extends HookConsumerWidget {
submittedDraftRevision: submittedDraftRevision,
ownsSource: ownsSource,
focusNode: focusNode,
clearComposer: clearComposer,
clearComposer: () {
clearComposer();
authorizationRevision = draftRevision.value;
},
addMentionedNonMembers: addMentionedNonMembers,
payload: _ComposeDraftPayload.fromDraft(
text: text,
@@ -666,7 +648,7 @@ class ComposeBar extends HookConsumerWidget {
customEmoji: customEmoji,
),
outgoing: outgoing,
onSend: onSend,
onSend: guardedDelivery,
messenger: messenger,
);
return;
@@ -691,7 +673,7 @@ class ComposeBar extends HookConsumerWidget {
final cancellation = UploadCancellationToken();
final uploadService = ref.read(mediaUploadServiceProvider);
activeUploadCancellation.value = cancellation;
final delivery = onSend;
final delivery = guardedDelivery;
unawaited(() async {
var retainedForRetry = false;
var delivered = false;

22
mobile/lib/features/channels/compose_bar/helpers.dart
View File

@@ -479,6 +479,7 @@ Future<_NonMemberAddOutcome> _addMentionedNonMembers(
required List<String> humanPubkeys,
required bool canAddMembers,
required VoidCallback ensureCurrent,
required VoidCallback ensureScopeCurrent,
required VoidCallback onAccepted,
required Future<bool> Function(String, String) authorizeWrite,
}) async {
@@ -500,23 +501,26 @@ Future<_NonMemberAddOutcome> _addMentionedNonMembers(
final notAdded = <String>[];
final errors = <String>[];
for (final (pubkeys, role) in pending) {
ensureCurrent();
ensureScopeCurrent();
if (!await authorizeWrite(pubkeys.single, role)) continue;
ensureCurrent();
try {
await channelActions.addMembers(
channelId: channelId,
pubkeys: pubkeys,
role: role,
onAccepted: (_) => onAccepted(),
await withRelayPublicationGuard(
ensureCurrent,
() => channelActions.addMembers(
channelId: channelId,
pubkeys: pubkeys,
role: role,
onAccepted: (_) => onAccepted(),
),
);
ensureCurrent();
ensureScopeCurrent();
} on _ComposeAuthorizationCancelled {
rethrow;
} on _ComposeCommunityChanged {
rethrow;
} on StateError {
ensureCurrent();
ensureScopeCurrent();
rethrow;
} catch (error) {
notAdded.addAll(
@@ -669,6 +673,7 @@ class _OutgoingMentions {
required _NonMemberMentionScan scan,
required ScaffoldMessengerState? messenger,
required VoidCallback ensureCurrent,
required VoidCallback ensureScopeCurrent,
required Future<bool> Function(String, String) authorizeWrite,
}) async {
final outcome = await _addMentionedNonMembers(
@@ -680,6 +685,7 @@ class _OutgoingMentions {
.toList(),
canAddMembers: scan.canAddMembers,
ensureCurrent: ensureCurrent,
ensureScopeCurrent: ensureScopeCurrent,
onAccepted: () => acceptedInvitations++,
authorizeWrite: authorizeWrite,
);

72
mobile/lib/features/channels/compose_bar/selected_mention_preparation.dart
View File

@@ -8,14 +8,19 @@ Future<Map<String, SelectedMentionAuthorization>> _authorizeSelectedMentions(
required Set<String> priorAgentKeys,
required Set<String> observedKeys,
required Map<String, NostrEvent> observedProfiles,
required Set<bool Function()> evidenceChecks,
required String? currentPubkey,
required String channelId,
required VoidCallback ensureAuthorizationCurrent,
required VoidCallback ensureScopeCurrent,
required bool Function() isAuthorizationCurrent,
bool prepare = false,
}) async {
ensureAuthorizationCurrent();
if (keys.isEmpty) return const {};
ensureScopeCurrent();
if (keys.isEmpty) {
evidenceChecks.clear();
return const {};
}
try {
final evidence = await readSelected(
keys,
@@ -34,11 +39,15 @@ Future<Map<String, SelectedMentionAuthorization>> _authorizeSelectedMentions(
observedProfiles.addAll(profiles);
},
);
ensureAuthorizationCurrent();
ensureScopeCurrent();
if (evidence.length != keys.length ||
!evidence.keys.toSet().containsAll(keys)) {
throw Exception('Incomplete selected mention evidence');
}
evidenceChecks
..clear()
..addAll(evidence.values.map((value) => value.isCurrent));
ensureAuthorizationCurrent();
final agents = {
for (final key in keys)
if (evidence[key]!.requiresAgentAuthorization) key,
@@ -55,7 +64,62 @@ Future<Map<String, SelectedMentionAuthorization>> _authorizeSelectedMentions(
ensureAuthorizationCurrent();
return evidence;
} catch (_) {
ensureAuthorizationCurrent();
ensureScopeCurrent();
rethrow;
}
}
// Failures preserve the exact draft audience; consent never licenses a changed
// role. Kept beside the reader owner so both boundaries use one evaluator.
Future<void> _prepareMentionInvitations({
required _OutgoingMentions outgoing,
required _NonMemberMentionScan scan,
required ChannelActions channelActions,
required ScaffoldMessengerState? messenger,
required Future<Map<String, SelectedMentionAuthorization>> Function(
Set<String> keys, {
bool prepare,
})
authorize,
required VoidCallback ensureCurrent,
required VoidCallback ensureScopeCurrent,
required VoidCallback onStarted,
required bool fenceAfterPreparation,
}) async {
final keys = outgoing.pubkeys.toSet();
Future<bool> authorizeWrite(String key, String role) async {
final evidence = await authorize(keys, prepare: true);
final fresh = evidence[key]!;
if (fresh.invitationRole != role) {
throw Exception(
'Mention classification changed; retry invitation consent',
);
}
return !fresh.isMember;
}
await authorize(keys, prepare: true);
ensureCurrent();
onStarted();
await outgoing.addNonMembers(
channelActions,
scan: scan,
messenger: messenger,
ensureCurrent: ensureCurrent,
ensureScopeCurrent: ensureScopeCurrent,
authorizeWrite: authorizeWrite,
);
if (!outgoing.pubkeys.toSet().containsAll(keys)) {
throw Exception(
'Mention invitation failed. Draft kept; retry or remove the mention.',
);
}
await authorize(keys);
if (fenceAfterPreparation) ensureCurrent();
}
// Equivalent refreshes retain scope; destination/credentials do not.
bool _isComposeConfigCurrent(WidgetRef ref, RelayConfig config) {
final current = ref.read(relayConfigProvider);
return current.baseUrl == config.baseUrl && current.nsec == config.nsec;
}

50
mobile/lib/shared/mentions/agent_policy.dart
View File

@@ -11,13 +11,16 @@ Future<List<NostrEvent>> _queryAgentFilters(
RelaySessionNotifier session,
List<NostrFilter> filters, {
void Function()? checkCurrent,
void Function(List<NostrFilter>, List<NostrEvent>?)? onQueryEvidence,
}) async {
final events = <NostrEvent>[];
onQueryEvidence?.call(filters, null);
for (var start = 0; start < filters.length; start += 10) {
checkCurrent?.call();
events.addAll(
await session.queryRelay(filters.skip(start).take(10).toList()),
);
final batch = filters.skip(start).take(10).toList();
final result = await session.queryRelay(batch);
onQueryEvidence?.call(batch, result);
events.addAll(result);
}
return events;
}
@@ -32,6 +35,7 @@ Future<List<AgentDirectoryEntry>> resolveAgentPolicies(
Set<String>? requestedKeys,
void Function()? checkCurrent,
void Function(Map<String, NostrEvent> profiles)? onProfileEvidence,
void Function(List<NostrFilter>, List<NostrEvent>?)? onQueryEvidence,
}) async {
final latest = <String, NostrEvent>{};
for (final event in runtimeEvents.where((event) => event.kind == 10100)) {
@@ -42,10 +46,15 @@ Future<List<AgentDirectoryEntry>> resolveAgentPolicies(
}
final keys = requestedKeys ?? latest.keys.toSet();
final profiles = latestProfileEvents(
await _queryAgentFilters(session, [
for (final key in keys)
NostrFilter(kinds: const [0], authors: [key], limit: 1),
], checkCurrent: checkCurrent),
await _queryAgentFilters(
session,
[
for (final key in keys)
NostrFilter(kinds: const [0], authors: [key], limit: 1),
],
checkCurrent: checkCurrent,
onQueryEvidence: onQueryEvidence,
),
);
final owners = <String, String>{};
for (final profile in profiles.values) {
@@ -55,17 +64,22 @@ Future<List<AgentDirectoryEntry>> resolveAgentPolicies(
}
}
checkCurrent?.call();
final policies = await _queryAgentFilters(session, [
for (final owner in owners.entries)
NostrFilter(
kinds: const [30177],
authors: [owner.value],
tags: {
'#d': [owner.key],
},
limit: 1,
),
], checkCurrent: checkCurrent);
final policies = await _queryAgentFilters(
session,
[
for (final owner in owners.entries)
NostrFilter(
kinds: const [30177],
authors: [owner.value],
tags: {
'#d': [owner.key],
},
limit: 1,
),
],
checkCurrent: checkCurrent,
onQueryEvidence: onQueryEvidence,
);
checkCurrent?.call();
onProfileEvidence?.call(Map.unmodifiable(profiles));
return mergeAgentPolicies(latest.values, policies, owners);

61
mobile/lib/shared/mentions/selected_mention_authorization.dart
View File

@@ -1,5 +1,7 @@
part of 'agent_identity_provider.dart';
bool _unfencedSelectedEvidence() => true;
/// Evidence requirements, not a claim that an ordinary identity is human.
enum SelectedMentionKind {
/// No fresh agent evidence for this key. Absence of agent evidence, not
@@ -36,8 +38,16 @@ class SelectedMentionAuthorization {
/// owner policy produces a deny-all entry, never runtime fallback.
final AgentDirectoryEntry? agent;
/// Local observed evidence capability, retained through actual enqueue.
final bool Function() isCurrent;
/// Construct evidence, not a publication or role-write capability.
const SelectedMentionAuthorization(this.kind, this.isMember, this.agent);
const SelectedMentionAuthorization(
this.kind,
this.isMember,
this.agent, {
this.isCurrent = _unfencedSelectedEvidence,
});
/// True when this key's evidence must be routed through the agent
/// authorization evaluator rather than the ordinary flow: agent or
@@ -73,8 +83,36 @@ readSelectedMentionAuthorization(
required bool Function() isCurrent,
void Function(Map<String, NostrEvent>)? onProfileEvidence,
}) async {
final fences = <bool Function()>[session.retainPublicationEvidence()];
bool evidenceCurrent() => fences.every((check) => check());
void check() {
if (!isCurrent()) throw StateError('Mention authorization scope changed');
if (!isCurrent() || !evidenceCurrent()) {
throw StateError('Mention authorization scope changed');
}
}
void observeQuery(List<NostrFilter> filters, List<NostrEvent>? events) {
for (final filter in filters) {
for (final kind in filter.kinds) {
for (final author in filter.authors!) {
fences.add(
events == null
? session.evidenceClock.retain(
kind,
author,
filter.tags['#d']?.single,
)
: session.evidenceClock.snapshot(
kind,
author,
filter.tags['#d']?.single,
events,
),
);
}
}
}
check();
}
check();
@@ -88,6 +126,7 @@ readSelectedMentionAuthorization(
if (requestedKeys.isEmpty) return const {};
final authority = await session.fetchRelaySelf();
check();
fences.add(session.evidenceClock.retain(39002, authority, channelId));
final membership = await _membershipPages(
session,
authority,
@@ -95,6 +134,9 @@ readSelectedMentionAuthorization(
channelId,
check,
);
fences.add(
session.evidenceClock.snapshot(39002, authority, channelId, membership),
);
check();
NostrEvent? roster;
for (final event in membership) {
@@ -120,10 +162,15 @@ readSelectedMentionAuthorization(
}
members[tag[1]] = tag.length >= 4 ? tag[3] : 'member';
}
final runtime = await _queryAgentFilters(session, [
for (final key in requestedKeys)
NostrFilter(kinds: const [10100], authors: [key], limit: 1),
], checkCurrent: check);
final runtime = await _queryAgentFilters(
session,
[
for (final key in requestedKeys)
NostrFilter(kinds: const [10100], authors: [key], limit: 1),
],
checkCurrent: check,
onQueryEvidence: observeQuery,
);
check();
Map<String, NostrEvent> profiles = const {};
final policies = await resolveAgentPolicies(
@@ -132,6 +179,7 @@ readSelectedMentionAuthorization(
requestedKeys: requestedKeys,
checkCurrent: check,
onProfileEvidence: (value) => profiles = value,
onQueryEvidence: observeQuery,
);
onProfileEvidence?.call(profiles);
check();
@@ -193,6 +241,7 @@ readSelectedMentionAuthorization(
kind,
members.containsKey(key),
agent,
isCurrent: evidenceCurrent,
);
}
check();

4
mobile/lib/shared/profile/user_cache_provider.dart
View File

@@ -31,10 +31,6 @@ class UserCacheNotifier extends Notifier<Map<String, UserProfile>> {
return {};
}
/// Latest observed profile revision; read-only fencing for publication.
({int createdAt, String eventId})? profileEventOrder(String key) =>
_profileEventOrders[key];
/// Request a profile for [pubkey]. Returns immediately from cache if
/// available, otherwise schedules a batch fetch.
UserProfile? get(String pubkey) {

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

@@ -13,6 +13,7 @@ import 'package:hooks_riverpod/hooks_riverpod.dart';
import '../auth/auth.dart';
import 'nostr_models.dart';
import 'relay_client.dart';
import 'relay_evidence_clock.dart';
import 'relay_closed_policy.dart';
import 'relay_http_query_client.dart';
import 'relay_provider.dart';
@@ -110,6 +111,9 @@ class RelaySessionNotifier extends Notifier<SessionState> {
_retryTimerFactory = retryTimerFactory,
_replayDelay = replayDelay;
/// Immediate evidence ordering shared by all reads on this session.
final evidenceClock = RelayEvidenceClock();
final RelayHttpQueryClient _httpQueryClient;
final RelaySocketFactory _socketFactory;
final DateTime Function() _now;
@@ -156,6 +160,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
// Reset disposed flag — build() may re-run on the same Notifier instance
// after a provider dependency changes (e.g. auth completing).
_disposed = false;
evidenceClock.clear();
ref.onDispose(_dispose);
@@ -194,6 +199,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
Duration timeout = const Duration(seconds: 8),
}) async {
final config = ref.read(relayConfigProvider);
final generation = _connectionGeneration;
final url = Uri.parse(config.baseUrl).resolve('/query').toString();
final bodyBytes = utf8.encode(
jsonEncode(filters.map((filter) => filter.toJson()).toList()),
@@ -223,13 +229,19 @@ class RelaySessionNotifier extends Notifier<SessionState> {
throw const FormatException('relay returned malformed query response');
}
try {
return [
final events = [
for (final eventJson in decoded)
if (eventJson is Map<String, dynamic>)
NostrEvent.fromJson(eventJson)
else
throw const FormatException('relay returned malformed query event'),
];
if (_isActiveConnection(generation)) {
for (final event in events) {
evidenceClock.observe(event);
}
}
return events;
} catch (error) {
if (error is FormatException) rethrow;
throw FormatException('relay returned malformed query event: $error');
@@ -409,6 +421,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
void debugSupersedeConnection() => _supersedeConnection();
int _supersedeConnection() {
evidenceClock.clear();
return ++_connectionGeneration;
}
@@ -700,6 +713,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
final subId = data[1] as String;
final eventJson = data[2] as Map<String, dynamic>;
final event = NostrEvent.fromJson(eventJson);
evidenceClock.observe(event);
// History subscriptions accumulate immediately.
final historySub = _historySubscriptions[subId];
@@ -1016,6 +1030,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
}
void _dispose() {
evidenceClock.clear();
_disposed = true;
_beforePauseCallbacks.clear();
_connectionGeneration++;

250
mobile/test/shared/mentions/selected_observation_test.dart Normal file
View File

@@ -0,0 +1,250 @@
import 'dart:async';
import 'dart:convert';
import 'package:buzz/shared/auth/auth.dart';
import 'package:buzz/shared/mentions/agent_identity_provider.dart';
import 'package:buzz/shared/relay/relay.dart';
import 'package:buzz/shared/relay/relay_evidence_clock.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import 'package:http/http.dart' as http;
import 'package:http/testing.dart';
import 'package:nostr/nostr.dart' as nostr;
import '../crypto/nip_oa_test.dart' show authTag, profile;
import 'agent_policy_test.dart' show signed;
void main() {
test(
'coordinate eviction retires in-flight proof without global cancellation',
() {
final clock = RelayEvidenceClock();
final owner = nostr.Keys.generate();
final retained = clock.retain(0, owner.public, null);
for (var i = 0; i < 8192; i++) {
clock.retain(30177, owner.public, '$i');
}
expect(retained(), isFalse);
final unrelated = clock.snapshot(30177, owner.public, '8191', []);
expect(unrelated(), isTrue);
},
);
for (final scenario in [
'policy',
'runtime',
'missing profile',
'roster',
'unrelated',
'retired session',
'replacement socket',
'queued policy',
'queued missing profile',
]) {
final mode = scenario.replaceFirst('queued ', '');
final queued = scenario.startsWith('queued ');
test(
'production reader rejects stale completion after observed $scenario',
() async {
final viewer = nostr.Keys.generate();
final agent = nostr.Keys.generate();
final relay = nostr.Keys.generate();
final other = nostr.Keys.generate();
NostrEvent policy(String mode, int time) => signed(
viewer,
30177,
{'name': 'Agent', 'parallelism': 1, 'respond_to': mode},
time: time,
tags: [
['d', agent.public],
],
);
NostrEvent roster(int time) => signed(
relay,
39002,
'',
time: time,
tags: [
['d', 'room'],
['p', viewer.public],
[
'p',
agent.public,
'',
mode == 'missing profile' ? 'member' : 'bot',
],
],
);
final events = [
roster(100),
if (mode != 'missing profile')
profile(agent, [authTag(viewer, agent.public)]),
if (mode != 'missing profile')
signed(agent, 10100, {'respond_to': 'anyone'}, time: 100),
policy('anyone', 100),
];
final oldQuery = Completer<void>();
final release = Completer<void>();
final blockedKind = switch (mode) {
'runtime' => 10100,
'missing profile' => 0,
'roster' => 39002,
_ => 30177,
};
final client = MockClient((request) async {
if (request.method == 'GET') {
return http.Response(jsonEncode({'self': relay.public}), 200);
}
final filters = (jsonDecode(request.body) as List)
.cast<Map<String, dynamic>>();
final results = events
.where(
(event) => filters.any(
(filter) =>
(filter['kinds'] as List).contains(event.kind) &&
(filter['authors'] as List).contains(event.pubkey) &&
(filter['#d'] == null ||
(filter['#d'] as List).contains(
event.getTagValue('d'),
)),
),
)
.toList();
if ((filters.first['kinds'] as List).contains(blockedKind)) {
oldQuery.complete();
await release.future;
}
return http.Response(
jsonEncode(results.map((e) => e.toJson()).toList()),
200,
);
});
final gate = RelayRateLimitGate();
final session = RelaySessionNotifier(
httpClient: client,
rateLimitGate: gate,
);
final container = ProviderContainer(
overrides: [
authProvider.overrideWith(_Auth.new),
relayConfigProvider.overrideWith(() => _Config(viewer.nsec)),
relaySessionProvider.overrideWith(() => session),
],
);
addTearDown(container.dispose);
await container.read(authProvider.future);
container.read(relaySessionProvider);
final socket = _Socket();
session.debugAttachSocketForTest(socket);
final subscription = session.subscribe(
const NostrFilter(kinds: [0, 10100, 30177, 39002]),
(_) {},
);
final subId = socket.messages.single[1] as String;
session.debugHandleMessage(['EOSE', subId]);
final unsubscribe = await subscription;
addTearDown(unsubscribe);
final pending = readSelectedMentionAuthorization(
session,
{agent.public},
viewer: viewer.public,
channelId: 'room',
isCurrent: () => true,
);
await oldQuery.future;
if (mode == 'retired session' || mode == 'replacement socket') {
final retained = session.evidenceClock.retain(0, agent.public, null);
if (mode == 'retired session') {
container.invalidate(relaySessionProvider);
container.read(relaySessionProvider);
} else {
session.debugSupersedeConnection();
}
expect(retained(), isFalse);
release.complete();
await expectLater(pending, throwsA(isA<StateError>()));
final clock = session.evidenceClock;
expect(
clock.snapshot(30177, viewer.public, agent.public, [])(),
isTrue,
);
return;
}
final newer = switch (mode) {
'policy' => policy('nobody', 200),
'runtime' => signed(agent, 10100, {
'respond_to': 'nobody',
}, time: 200),
'missing profile' => profile(agent, [
authTag(viewer, agent.public),
], createdAt: 200),
'roster' => roster(200),
_ => profile(other, [], createdAt: 200),
};
// Same production socket receipt handler used by discovery/user-cache
// streams. It runs immediately, before the 16ms UI batch or query finish.
if (queued) {
release.complete();
final evidence = await pending;
gate.activate(10);
final publication = withRelayPublicationGuard(
() {
if (!evidence.values.every((e) => e.isCurrent())) {
throw StateError('observed evidence changed before enqueue');
}
},
() => SignedEventRelay(session: session, nsec: viewer.nsec).submit(
kind: 9000,
content: '',
tags: [
['h', 'room'],
['p', agent.public],
],
),
);
session.debugHandleMessage(['EVENT', subId, newer.toJson()]);
gate.reset();
await expectLater(publication, throwsA(isA<StateError>()));
expect(socket.messages.where((p) => p.first == 'EVENT'), isEmpty);
session.debugFlushEventBuffer();
return;
}
session.debugHandleMessage(['EVENT', subId, newer.toJson()]);
release.complete();
if (mode == 'unrelated') {
expect((await pending).keys, {agent.public});
} else {
await expectLater(pending, throwsA(isA<StateError>()));
}
},
);
}
}
class _Config extends RelayConfigNotifier {
_Config(this.key);
final String key;
@override
RelayConfig build() =>
RelayConfig(baseUrl: 'https://relay.example', nsec: key);
}
class _Socket extends RelaySocket {
_Socket()
: super(
wsUrl: 'wss://relay.example',
nsec: null,
onMessage: (_) {},
onConnected: () {},
onDisconnected: (_) {},
);
final messages = <List<dynamic>>[];
@override
void send(List<dynamic> payload) => messages.add(payload);
}
class _Auth extends AuthNotifier {
@override
Future<AuthState> build() async =>
const AuthState(status: AuthStatus.unauthenticated);
}