diff --git a/mobile/lib/shared/relay/relay_http_query_client.dart b/mobile/lib/shared/relay/relay_http_query_client.dart index 71f9bc1fd9..2d461697fa 100644 --- a/mobile/lib/shared/relay/relay_http_query_client.dart +++ b/mobile/lib/shared/relay/relay_http_query_client.dart @@ -1,14 +1,31 @@ import 'dart:async'; +import 'dart:io'; import 'package:http/http.dart' as http; +import 'package:http/io_client.dart'; + +/// The default transport. `dart:io` would inflate gzip responses on the +/// calling isolate, so leave them compressed for the response decoder. +/// +/// A non-`dart:io` client supplied through `http.runWithClient`, such as a +/// `MockClient`, is used as is. An `IOClient` is replaced, not closed, since +/// it may be shared. +http.Client _gzipPassthroughClient() { + final client = http.Client(); + if (client is! IOClient) return client; + return IOClient(HttpClient()..autoUncompress = false); +} /// Owns the reusable HTTP transport used for relay `/query` requests. +/// +/// Requests advertise gzip. A response body may therefore come back +/// compressed; decode it with `relayResponseText` or `decodeRelayQueryResponse`. class RelayHttpQueryClient { RelayHttpQueryClient({ http.Client? client, http.Client Function()? clientFactory, }) : _injectedClient = client, - _clientFactory = clientFactory ?? (() => http.Client()); + _clientFactory = clientFactory ?? _gzipPassthroughClient; final http.Client? _injectedClient; final http.Client Function() _clientFactory; @@ -30,6 +47,7 @@ class RelayHttpQueryClient { try { final request = http.Request('POST', url) ..followRedirects = false + ..headers['Accept-Encoding'] = 'gzip' ..headers.addAll(headers) ..bodyBytes = body; final transport = _injectedClient ?? generation!.client; diff --git a/mobile/lib/shared/relay/relay_query_response.dart b/mobile/lib/shared/relay/relay_query_response.dart new file mode 100644 index 0000000000..d94ec3ba93 --- /dev/null +++ b/mobile/lib/shared/relay/relay_query_response.dart @@ -0,0 +1,74 @@ +import 'dart:convert'; +import 'dart:io'; +import 'dart:isolate'; +import 'dart:typed_data'; + +import 'package:http/http.dart' as http; + +import 'nostr_models.dart'; + +/// Turns a successful relay query response into events. +typedef RelayQueryDecoder = + Future> Function(http.Response response); + +/// Decodes a relay query response without blocking the UI isolate. +/// +/// Inflating, UTF-8 decoding, JSON parsing, and event parsing all run in one +/// short-lived isolate, which hands the parsed events back by ownership +/// transfer instead of copying them. A per-response isolate keeps no state to +/// dispose and is the same shape as `parseProfileEventBatch`. +/// +/// Throws the same [FormatException]s as [parseRelayQueryEvents]. +Future> decodeRelayQueryResponse(http.Response response) => + _decodeInIsolate(response.bodyBytes, response.headers['content-type']); + +// Takes only the bytes and content type so the isolate closure cannot capture +// the response, its headers, or any session state. +Future> _decodeInIsolate( + Uint8List bytes, + String? contentType, +) => Isolate.run(() => parseRelayQueryEvents(bytes, contentType)); + +/// Parses the event array a relay `/query` or `/profiles/search` answers. +/// +/// Throws a [FormatException] for a body that is not JSON, is not an array, +/// or holds an element that is not a well-formed event. +List parseRelayQueryEvents(Uint8List bytes, String? contentType) { + final decoded = jsonDecode(relayResponseText(bytes, contentType)); + if (decoded is! List) { + throw const FormatException('relay returned malformed query response'); + } + try { + return [ + for (final eventJson in decoded) + if (eventJson is Map) + NostrEvent.fromJson(eventJson) + else + throw const FormatException('relay returned malformed query event'), + ]; + } catch (error) { + if (error is FormatException) rethrow; + throw FormatException('relay returned malformed query event: $error'); + } +} + +/// The text of a relay response body, inflated when the server sent gzip. +/// +/// The query transport asks for gzip and leaves it compressed so inflating +/// can happen off the UI isolate. A body that a transport already inflated +/// passes through. JSON never starts with the gzip magic bytes, so the bytes +/// decide, not the `Content-Encoding` header a transport may leave behind. +/// The charset rules are those of [http.Response.body]. +String relayResponseText(Uint8List bytes, String? contentType) { + final inflated = _isGzip(bytes) + ? Uint8List.fromList(gzip.decode(bytes)) + : bytes; + return http.Response.bytes( + inflated, + 200, + headers: {'content-type': ?contentType}, + ).body; +} + +bool _isGzip(Uint8List bytes) => + bytes.length > 2 && bytes[0] == 0x1f && bytes[1] == 0x8b; diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index cdefed9768..bdc07b1df7 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -20,6 +20,7 @@ import 'relay_client.dart'; import 'relay_closed_policy.dart'; import 'relay_http_query_client.dart'; import 'relay_provider.dart'; +import 'relay_query_response.dart'; import 'relay_rate_limit_gate.dart'; import 'relay_reconnect_policy.dart'; import 'relay_session_types.dart'; @@ -42,6 +43,7 @@ class RelaySessionNotifier extends Notifier { Future Function(Duration) replayDelay = Future.delayed, RelayTimerFactory reconnectTimerFactory = Timer.new, Random? random, + RelayQueryDecoder queryDecoder = decodeRelayQueryResponse, }) : _httpQueryClient = RelayHttpQueryClient( client: httpClient, clientFactory: httpClientFactory, @@ -52,7 +54,8 @@ class RelaySessionNotifier extends Notifier { _retryTimerFactory = retryTimerFactory, _replayDelay = replayDelay, _reconnectTimerFactory = reconnectTimerFactory, - _random = random ?? Random(); + _random = random ?? Random(), + _queryDecoder = queryDecoder; final RelayHttpQueryClient _httpQueryClient; final RelaySocketFactory _socketFactory; @@ -62,6 +65,7 @@ class RelaySessionNotifier extends Notifier { final Future Function(Duration) _replayDelay; final RelayTimerFactory _reconnectTimerFactory; final Random _random; + final RelayQueryDecoder _queryDecoder; static const _baseReconnectDelayMs = relayReconnectBaseDelayMs; static const _maxReconnectDelayMs = relayReconnectMaxDelayMs; @@ -232,29 +236,34 @@ class RelaySessionNotifier extends Notifier { active() && !_disposed && generation == _contextGeneration, ), ); - if (_disposed || _ageRestricted || generation != _contextGeneration) { + bool retired() => + _disposed || _ageRestricted || generation != _contextGeneration; + if (retired()) { throw StateError('Relay query belongs to a retired session'); } if (response.statusCode < 200 || response.statusCode >= 300) { - _activateRateLimitGateFromHttpError(response.body); - throw RelayException(response.statusCode, response.body); - } - final decoded = jsonDecode(response.body); - if (decoded is! List) { - throw const FormatException('relay returned malformed query response'); + final body = relayResponseText( + response.bodyBytes, + response.headers['content-type'], + ); + _activateRateLimitGateFromHttpError(body); + throw RelayException(response.statusCode, body); } + // Decoding runs off the UI isolate, so the session can retire meanwhile. + // A retired session reports that, not whatever its bytes decoded to. + final List events; try { - return [ - for (final eventJson in decoded) - if (eventJson is Map) - NostrEvent.fromJson(eventJson) - else - throw const FormatException('relay returned malformed query event'), - ]; - } catch (error) { - if (error is FormatException) rethrow; - throw FormatException('relay returned malformed query event: $error'); + events = await _queryDecoder(response); + } catch (_) { + if (retired()) { + throw StateError('Relay query belongs to a retired session'); + } + rethrow; } + if (retired()) { + throw StateError('Relay query belongs to a retired session'); + } + return events; } void _activateRateLimitGateFromHttpError(String body) { diff --git a/mobile/test/helpers/fake_relay.dart b/mobile/test/helpers/fake_relay.dart index 5b67559961..b4b841bdd1 100644 --- a/mobile/test/helpers/fake_relay.dart +++ b/mobile/test/helpers/fake_relay.dart @@ -4,6 +4,7 @@ import 'dart:math'; import 'package:buzz/shared/auth/auth_provider.dart'; import 'package:buzz/shared/relay/relay.dart'; +import 'package:buzz/shared/relay/relay_query_response.dart'; import 'package:flutter_test/flutter_test.dart'; import 'package:hooks_riverpod/misc.dart'; import 'package:http/http.dart' as http; @@ -50,6 +51,12 @@ class FakeRelay { late final session = RelaySessionNotifier( socketFactory: _openSocket, httpClient: MockClient(_handleHttp), + // The production decoder runs in an isolate, whose result `fakeAsync` + // cannot deliver. Parse inline with the same function instead. + queryDecoder: (response) async => parseRelayQueryEvents( + response.bodyBytes, + response.headers['content-type'], + ), now: _now, random: _MidpointRandom(), ); diff --git a/mobile/test/shared/relay/relay_http_query_client_test.dart b/mobile/test/shared/relay/relay_http_query_client_test.dart new file mode 100644 index 0000000000..c3fa2838c1 --- /dev/null +++ b/mobile/test/shared/relay/relay_http_query_client_test.dart @@ -0,0 +1,57 @@ +import 'dart:convert'; +import 'dart:io'; + +import 'package:buzz/shared/relay/relay_http_query_client.dart'; +import 'package:buzz/shared/relay/relay_query_response.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart'; + +void main() { + test('the default transport honors http.runWithClient', () async { + final requests = []; + final response = await http.runWithClient( + () => RelayHttpQueryClient().post( + Uri.parse('https://relay.example/query'), + headers: const {'Content-Type': 'application/json'}, + body: const [], + timeout: const Duration(seconds: 5), + ), + () => MockClient((request) async { + requests.add(request); + return http.Response('[]', 200); + }), + ); + + expect(response.body, '[]'); + expect(requests.single.headers['Accept-Encoding'], 'gzip'); + }); + + test('the default transport leaves a gzip body for the decoder', () async { + final compressed = gzip.encode(utf8.encode('[]')); + final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + addTearDown(() => server.close(force: true)); + server.listen((request) async { + request.response + ..headers.set(HttpHeaders.contentEncodingHeader, 'gzip') + ..headers.contentType = ContentType.json + ..add(compressed); + await request.response.close(); + }); + + final client = RelayHttpQueryClient(); + addTearDown(client.close); + final response = await client.post( + Uri.parse('http://127.0.0.1:${server.port}/query'), + headers: const {'Content-Type': 'application/json'}, + body: const [], + timeout: const Duration(seconds: 5), + ); + + expect(response.bodyBytes, compressed); + expect( + relayResponseText(response.bodyBytes, response.headers['content-type']), + '[]', + ); + }); +} diff --git a/mobile/test/shared/relay/relay_session_test.dart b/mobile/test/shared/relay/relay_session_test.dart index 18310bc1f0..c43e0b06ed 100644 --- a/mobile/test/shared/relay/relay_session_test.dart +++ b/mobile/test/shared/relay/relay_session_test.dart @@ -1,5 +1,6 @@ import 'dart:async'; import 'dart:convert'; +import 'dart:io'; import 'package:flutter_test/flutter_test.dart'; import 'package:hooks_riverpod/hooks_riverpod.dart'; @@ -12,6 +13,7 @@ import 'package:buzz/features/channels/agent_activity/observer_subscription.dart import 'package:buzz/features/channels/agent_activity/observer_models.dart'; import 'package:buzz/shared/auth/auth_provider.dart'; import 'package:buzz/shared/relay/relay.dart'; +import 'package:buzz/shared/relay/relay_query_response.dart'; void main() { test( @@ -508,6 +510,279 @@ void main() { }, ); + group('queryRelay response decoding', () { + final eventJson = { + 'id': 'a' * 64, + 'pubkey': 'b' * 64, + 'created_at': 1700000000, + 'kind': 9, + 'tags': [ + ['h', 'channel-1'], + ['p', 'c' * 64], + ], + 'content': 'café \u{1F41D} 日本語', + 'sig': 'd' * 128, + }; + + http.Response json(String body, {int status = 200}) => http.Response.bytes( + utf8.encode(body), + status, + headers: {'content-type': 'application/json'}, + ); + + Future expectFormat( + String body, + Object message, { + int status = 200, + }) async { + final harness = _queryHarness( + client: http_testing.MockClient( + (_) async => json(body, status: status), + ), + ); + addTearDown(harness.container.dispose); + await expectLater( + harness.session.queryRelay(const []), + throwsA(isA().having((e) => e.message, 'm', message)), + ); + } + + test('parses events and keeps non-ASCII content intact', () async { + final harness = _queryHarness( + client: http_testing.MockClient( + (_) async => json(jsonEncode([eventJson, eventJson])), + ), + ); + addTearDown(harness.container.dispose); + + final events = await harness.session.queryRelay(const []); + + expect(events, hasLength(2)); + expect(events.first.toJson(), eventJson); + }); + + test('profile search decodes through the same path', () async { + final profile = {...eventJson, 'kind': 0}; + final harness = _queryHarness( + client: http_testing.MockClient( + (_) async => json(jsonEncode([profile, eventJson])), + ), + ); + addTearDown(harness.container.dispose); + + final events = await harness.session.searchProfiles('wes'); + + expect(events.map((e) => e.kind), [0]); + }); + + test('requests gzip and inflates a compressed success body', () async { + String? acceptEncoding; + final harness = _queryHarness( + client: http_testing.MockClient((request) async { + acceptEncoding = request.headers['Accept-Encoding']; + return http.Response.bytes( + gzip.encode(utf8.encode(jsonEncode([eventJson]))), + 200, + headers: { + 'content-type': 'application/json', + 'content-encoding': 'gzip', + }, + ); + }), + ); + addTearDown(harness.container.dispose); + + final events = await harness.session.queryRelay(const []); + + expect(acceptEncoding, 'gzip'); + expect(events.single.toJson(), eventJson); + }); + + test( + 'inflates a compressed error body for the gate and exception', + () async { + final gateTimers = <_ManualTimer>[]; + final gate = RelayRateLimitGate( + now: () => DateTime(2026), + timerFactory: (duration, callback) { + final timer = _ManualTimer(duration, callback); + gateTimers.add(timer); + return timer; + }, + ); + const body = '{"error":"rate-limited: quota exceeded; retry in 7s"}'; + final harness = _queryHarness( + gate: gate, + client: http_testing.MockClient( + (_) async => http.Response.bytes( + gzip.encode(utf8.encode(body)), + 429, + headers: { + 'content-type': 'application/json', + 'content-encoding': 'gzip', + }, + ), + ), + ); + addTearDown(harness.container.dispose); + + await expectLater( + harness.session.queryRelay(const []), + throwsA( + isA() + .having((e) => e.statusCode, 'statusCode', 429) + .having((e) => e.body, 'body', body), + ), + ); + expect(gateTimers.single.duration, const Duration(seconds: 7)); + }, + ); + + test('malformed JSON surfaces the parser FormatException', () async { + await expectFormat('{"oops"', isNot(startsWith('relay returned'))); + }); + + test('a non-list response is rejected', () async { + await expectFormat( + '{"events":[]}', + 'relay returned malformed query response', + ); + }); + + test('a non-object element is rejected', () async { + await expectFormat('[1]', 'relay returned malformed query event'); + }); + + test('a malformed event names the field error', () async { + await expectFormat( + '[{}]', + startsWith('relay returned malformed query event: '), + ); + }); + + test('invalid UTF-8 is a FormatException', () async { + final harness = _queryHarness( + client: http_testing.MockClient( + (_) async => http.Response.bytes( + [0x5b, 0xff, 0x5d], + 200, + headers: {'content-type': 'application/json'}, + ), + ), + ); + addTearDown(harness.container.dispose); + await expectLater( + harness.session.queryRelay(const []), + throwsA(isA()), + ); + }); + + test('an error status keeps its body and skips decoding', () async { + var decodes = 0; + const body = '{"error":"nope"}'; + final harness = _queryHarness( + client: http_testing.MockClient((_) async => json(body, status: 500)), + queryDecoder: (response) { + decodes++; + return decodeRelayQueryResponse(response); + }, + ); + addTearDown(harness.container.dispose); + + await expectLater( + harness.session.queryRelay(const []), + throwsA( + isA() + .having((e) => e.statusCode, 'statusCode', 500) + .having((e) => e.body, 'body', body), + ), + ); + expect(decodes, 0); + }); + + test('a session disposed during decoding discards the events', () async { + final decoder = _GatedDecoder(); + final harness = _queryHarness( + client: http_testing.MockClient( + (_) async => json(jsonEncode([eventJson])), + ), + queryDecoder: decoder.call, + ); + addTearDown(harness.container.dispose); + + final query = harness.session.queryRelay(const []); + final outcome = expectLater(query, throwsStateError); + await decoder.started.future; + harness.session.debugDispose(); + decoder.gate.complete(); + + await outcome; + }); + + test('a retired session reports retirement, not a decode error', () async { + final decoder = _GatedDecoder(); + final harness = _queryHarness( + client: http_testing.MockClient((_) async => json('{"oops"')), + queryDecoder: decoder.call, + ); + addTearDown(harness.container.dispose); + + final query = harness.session.queryRelay(const []); + final outcome = expectLater(query, throwsStateError); + await decoder.started.future; + harness.session.debugDispose(); + decoder.gate.complete(); + + await outcome; + }); + + test('an age restriction during decoding discards the events', () async { + final age = _MutableAgeNotifier(); + final decoder = _GatedDecoder(); + final harness = _queryHarness( + client: http_testing.MockClient( + (_) async => json(jsonEncode([eventJson])), + ), + queryDecoder: decoder.call, + age: age, + ); + addTearDown(harness.container.dispose); + final listener = harness.container.listen( + relaySessionProvider, + (_, _) {}, + ); + addTearDown(listener.close); + + final query = harness.session.queryRelay(const []); + final outcome = expectLater(query, throwsStateError); + await decoder.started.future; + age.setState(AgeSignalState.restricted); + await Future.delayed(Duration.zero); + decoder.gate.complete(); + + await outcome; + }); + + test('a session retired before decoding never decodes', () async { + final response = Completer(); + final decoder = _GatedDecoder(); + final harness = _queryHarness( + client: http_testing.MockClient((_) => response.future), + queryDecoder: decoder.call, + ); + addTearDown(harness.container.dispose); + + final query = harness.session.queryRelay(const []); + final outcome = expectLater(query, throwsStateError); + await Future.delayed(Duration.zero); + harness.session.debugDispose(); + response.complete(json(jsonEncode([eventJson]))); + + await outcome; + expect(decoder.calls, 0); + }); + }); + group('searchProfiles', () { Map profileJson(int kind, String name) => { 'id': name.padRight(64, '0'), @@ -1926,6 +2201,21 @@ class _QueuedControlledHttpClient extends http.BaseClient { void close() => closed = true; } +/// Holds the production decoder until the test releases it, so a test can +/// change the session while a real decode is outstanding. +class _GatedDecoder { + final gate = Completer(); + final started = Completer(); + int calls = 0; + + Future> call(http.Response response) async { + calls++; + started.complete(); + await gate.future; + return decodeRelayQueryResponse(response); + } +} + class _QueryHarness { final ProviderContainer container; final RelaySessionNotifier session; @@ -1934,12 +2224,19 @@ class _QueryHarness { } _QueryHarness _queryHarness({ - required RelayRateLimitGate gate, + RelayRateLimitGate? gate, required http.Client client, + RelayQueryDecoder? queryDecoder, + AgeSignalNotifier? age, }) { - final session = RelaySessionNotifier(httpClient: client, rateLimitGate: gate); + final session = RelaySessionNotifier( + httpClient: client, + rateLimitGate: gate, + queryDecoder: queryDecoder ?? decodeRelayQueryResponse, + ); final container = ProviderContainer( overrides: [ + if (age != null) ageSignalProvider.overrideWith(() => age), authProvider.overrideWith(() => _PendingAuthNotifier()), relaySessionProvider.overrideWith(() => session), relayConfigProvider.overrideWith(