diff --git a/lib/app.dart b/lib/app.dart index c1c68482..74031493 100644 --- a/lib/app.dart +++ b/lib/app.dart @@ -6,6 +6,7 @@ import 'package:flutter/material.dart'; import 'package:flutter_localizations/flutter_localizations.dart'; import 'package:flutter_sharing_intent/model/sharing_file.dart' show SharedFile; import 'package:provider/provider.dart'; +import 'package:twonly/core/bridge/api.dart' as rust_api; import 'package:twonly/globals.dart'; import 'package:twonly/locator.dart'; import 'package:twonly/src/constants/keyvalue.keys.dart'; @@ -59,11 +60,16 @@ class _AppState extends State with WidgetsBindingObserver { if (_wasPaused) { AppState.isAppInBackground = false; twonlyDB.markUpdated(); - unawaited(apiService.connect()); + unawaited( + rust_api.RustApi.setBackground(inBackground: false), + ); } } else if (state == AppLifecycleState.paused) { _wasPaused = true; AppState.isAppInBackground = true; + unawaited( + rust_api.RustApi.setBackground(inBackground: true), + ); } } diff --git a/lib/core/bridge/callbacks.dart b/lib/core/bridge/callbacks.dart index 7c5ab5f6..af332b33 100644 --- a/lib/core/bridge/callbacks.dart +++ b/lib/core/bridge/callbacks.dart @@ -9,7 +9,7 @@ import 'package:flutter_rust_bridge/flutter_rust_bridge_for_generated.dart'; // These functions are ignored because they are not marked as `pub`: `get_callbacks` // These types are ignored because they are neither used by any `pub` functions nor (for structs and enums) marked `#[frb(unignore)]`: `Api`, `FlutterCallbacks`, `LegacySignal`, `Logging` -// These function are ignored because they are on traits that is not defined in current crate (put an empty `#[frb]` on it to unignore): `clone`, `clone`, `clone`, `clone`, `clone`, `clone`, `clone`, `fmt`, `fmt`, `fmt` +// These function are ignored because they are on traits that is not defined in current crate (put an empty `#[frb]` on it to unignore): `clone`, `clone`, `clone`, `clone`, `clone`, `clone`, `fmt`, `fmt` Future initFlutterCallbacks({ required int callbackId, @@ -25,8 +25,6 @@ Future initFlutterCallbacks({ Uint8List, ) legacySignalEncrypt, - required FutureOr> Function() - legacySignalGeneratePrekeys, required FutureOr Function(PlatformInt64) apiResyncSignalSession, required FutureOr Function(String, String, PlatformInt64, String) apiMediaAction, @@ -40,7 +38,6 @@ Future initFlutterCallbacks({ loggingGetStreamSink: loggingGetStreamSink, legacySignalDecrypt: legacySignalDecrypt, legacySignalEncrypt: legacySignalEncrypt, - legacySignalGeneratePrekeys: legacySignalGeneratePrekeys, apiResyncSignalSession: apiResyncSignalSession, apiMediaAction: apiMediaAction, apiVerificationProof: apiVerificationProof, @@ -95,24 +92,3 @@ class LegacySignalEncryptResult { ciphertext == other.ciphertext && messageType == other.messageType; } - -class LegacySignalPreKey { - final PlatformInt64 id; - final Uint8List publicKey; - - const LegacySignalPreKey({ - required this.id, - required this.publicKey, - }); - - @override - int get hashCode => id.hashCode ^ publicKey.hashCode; - - @override - bool operator ==(Object other) => - identical(this, other) || - other is LegacySignalPreKey && - runtimeType == other.runtimeType && - id == other.id && - publicKey == other.publicKey; -} diff --git a/lib/core/frb_generated.dart b/lib/core/frb_generated.dart index e25244f1..090a9ec4 100644 --- a/lib/core/frb_generated.dart +++ b/lib/core/frb_generated.dart @@ -149,8 +149,6 @@ abstract class RustLibApi extends BaseApi { Uint8List, ) legacySignalEncrypt, - required FutureOr> Function() - legacySignalGeneratePrekeys, required FutureOr Function(PlatformInt64) apiResyncSignalSession, required FutureOr Function(String, String, PlatformInt64, String) apiMediaAction, @@ -933,8 +931,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { Uint8List, ) legacySignalEncrypt, - required FutureOr> Function() - legacySignalGeneratePrekeys, required FutureOr Function(PlatformInt64) apiResyncSignalSession, required FutureOr Function(String, String, PlatformInt64, String) apiMediaAction, @@ -961,10 +957,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { legacySignalEncrypt, serializer, ); - sse_encode_DartFn_Inputs__Output_list_legacy_signal_pre_key_AnyhowException( - legacySignalGeneratePrekeys, - serializer, - ); sse_encode_DartFn_Inputs_i_64_Output_unit_AnyhowException( apiResyncSignalSession, serializer, @@ -1006,7 +998,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { loggingGetStreamSink, legacySignalDecrypt, legacySignalEncrypt, - legacySignalGeneratePrekeys, apiResyncSignalSession, apiMediaAction, apiVerificationProof, @@ -1027,7 +1018,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { "loggingGetStreamSink", "legacySignalDecrypt", "legacySignalEncrypt", - "legacySignalGeneratePrekeys", "apiResyncSignalSession", "apiMediaAction", "apiVerificationProof", @@ -4893,43 +4883,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { }; } - Future Function( - int, - ) - encode_DartFn_Inputs__Output_list_legacy_signal_pre_key_AnyhowException( - FutureOr> Function() raw, - ) { - return ( - callId, - ) async { - Box>? rawOutput; - Box? rawError; - try { - rawOutput = Box(await raw()); - } catch (e, s) { - rawError = Box(AnyhowException("$e\n\n$s")); - } - - final serializer = SseSerializer(generalizedFrbRustBinding); - assert((rawOutput != null) ^ (rawError != null)); - if (rawOutput != null) { - serializer.buffer.putUint8(0); - sse_encode_list_legacy_signal_pre_key(rawOutput.value, serializer); - } else { - serializer.buffer.putUint8(1); - sse_encode_AnyhowException(rawError!.value, serializer); - } - final output = serializer.intoRaw(); - - generalizedFrbRustBinding.dartFnDeliverOutput( - callId: callId, - ptr: output.ptr, - rustVecLen: output.rustVecLen, - dataLen: output.dataLen, - ); - }; - } - Future Function(int, dynamic) encode_DartFn_Inputs_i_64_Output_unit_AnyhowException( FutureOr Function(PlatformInt64) raw, @@ -5152,15 +5105,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { throw UnimplementedError(''); } - @protected - FutureOr> Function() - dco_decode_DartFn_Inputs__Output_list_legacy_signal_pre_key_AnyhowException( - dynamic raw, - ) { - // Codec=Dco (DartCObject based), see doc to use other codecs - throw UnimplementedError(''); - } - @protected FutureOr Function(PlatformInt64) dco_decode_DartFn_Inputs_i_64_Output_unit_AnyhowException(dynamic raw) { @@ -5511,18 +5455,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { ); } - @protected - LegacySignalPreKey dco_decode_legacy_signal_pre_key(dynamic raw) { - // Codec=Dco (DartCObject based), see doc to use other codecs - final arr = raw as List; - if (arr.length != 2) - throw Exception('unexpected arr length: expect 2 but see ${arr.length}'); - return LegacySignalPreKey( - id: dco_decode_i_64(arr[0]), - publicKey: dco_decode_list_prim_u_8_strict(arr[1]), - ); - } - @protected LegacyTableMigrationCount dco_decode_legacy_table_migration_count( dynamic raw, @@ -5549,14 +5481,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { return (raw as List).map(dco_decode_frb_pqc_pre_key).toList(); } - @protected - List dco_decode_list_legacy_signal_pre_key(dynamic raw) { - // Codec=Dco (DartCObject based), see doc to use other codecs - return (raw as List) - .map(dco_decode_legacy_signal_pre_key) - .toList(); - } - @protected List dco_decode_list_legacy_table_migration_count( dynamic raw, @@ -6441,16 +6365,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { ); } - @protected - LegacySignalPreKey sse_decode_legacy_signal_pre_key( - SseDeserializer deserializer, - ) { - // Codec=Sse (Serialization based), see doc to use other codecs - var var_id = sse_decode_i_64(deserializer); - var var_publicKey = sse_decode_list_prim_u_8_strict(deserializer); - return LegacySignalPreKey(id: var_id, publicKey: var_publicKey); - } - @protected LegacyTableMigrationCount sse_decode_legacy_table_migration_count( SseDeserializer deserializer, @@ -6487,20 +6401,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { return ans_; } - @protected - List sse_decode_list_legacy_signal_pre_key( - SseDeserializer deserializer, - ) { - // Codec=Sse (Serialization based), see doc to use other codecs - - var len_ = sse_decode_i_32(deserializer); - var ans_ = []; - for (var idx_ = 0; idx_ < len_; ++idx_) { - ans_.add(sse_decode_legacy_signal_pre_key(deserializer)); - } - return ans_; - } - @protected List sse_decode_list_legacy_table_migration_count( SseDeserializer deserializer, @@ -7269,21 +7169,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { ); } - @protected - void - sse_encode_DartFn_Inputs__Output_list_legacy_signal_pre_key_AnyhowException( - FutureOr> Function() self, - SseSerializer serializer, - ) { - // Codec=Sse (Serialization based), see doc to use other codecs - sse_encode_DartOpaque( - encode_DartFn_Inputs__Output_list_legacy_signal_pre_key_AnyhowException( - self, - ), - serializer, - ); - } - @protected void sse_encode_DartFn_Inputs_i_64_Output_unit_AnyhowException( FutureOr Function(PlatformInt64) self, @@ -7697,16 +7582,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { sse_encode_i_32(self.messageType, serializer); } - @protected - void sse_encode_legacy_signal_pre_key( - LegacySignalPreKey self, - SseSerializer serializer, - ) { - // Codec=Sse (Serialization based), see doc to use other codecs - sse_encode_i_64(self.id, serializer); - sse_encode_list_prim_u_8_strict(self.publicKey, serializer); - } - @protected void sse_encode_legacy_table_migration_count( LegacyTableMigrationCount self, @@ -7738,18 +7613,6 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { } } - @protected - void sse_encode_list_legacy_signal_pre_key( - List self, - SseSerializer serializer, - ) { - // Codec=Sse (Serialization based), see doc to use other codecs - sse_encode_i_32(self.length, serializer); - for (final item in self) { - sse_encode_legacy_signal_pre_key(item, serializer); - } - } - @protected void sse_encode_list_legacy_table_migration_count( List self, diff --git a/lib/core/frb_generated.io.dart b/lib/core/frb_generated.io.dart index c007a5d1..ff3602be 100644 --- a/lib/core/frb_generated.io.dart +++ b/lib/core/frb_generated.io.dart @@ -56,12 +56,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { dynamic raw, ); - @protected - FutureOr> Function() - dco_decode_DartFn_Inputs__Output_list_legacy_signal_pre_key_AnyhowException( - dynamic raw, - ); - @protected FutureOr Function(PlatformInt64) dco_decode_DartFn_Inputs_i_64_Output_unit_AnyhowException(dynamic raw); @@ -210,9 +204,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { dynamic raw, ); - @protected - LegacySignalPreKey dco_decode_legacy_signal_pre_key(dynamic raw); - @protected LegacyTableMigrationCount dco_decode_legacy_table_migration_count( dynamic raw, @@ -224,9 +215,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected List dco_decode_list_frb_pqc_pre_key(dynamic raw); - @protected - List dco_decode_list_legacy_signal_pre_key(dynamic raw); - @protected List dco_decode_list_legacy_table_migration_count( dynamic raw, @@ -562,11 +550,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { SseDeserializer deserializer, ); - @protected - LegacySignalPreKey sse_decode_legacy_signal_pre_key( - SseDeserializer deserializer, - ); - @protected LegacyTableMigrationCount sse_decode_legacy_table_migration_count( SseDeserializer deserializer, @@ -580,11 +563,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { SseDeserializer deserializer, ); - @protected - List sse_decode_list_legacy_signal_pre_key( - SseDeserializer deserializer, - ); - @protected List sse_decode_list_legacy_table_migration_count( SseDeserializer deserializer, @@ -820,13 +798,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { SseSerializer serializer, ); - @protected - void - sse_encode_DartFn_Inputs__Output_list_legacy_signal_pre_key_AnyhowException( - FutureOr> Function() self, - SseSerializer serializer, - ); - @protected void sse_encode_DartFn_Inputs_i_64_Output_unit_AnyhowException( FutureOr Function(PlatformInt64) self, @@ -1042,12 +1013,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { SseSerializer serializer, ); - @protected - void sse_encode_legacy_signal_pre_key( - LegacySignalPreKey self, - SseSerializer serializer, - ); - @protected void sse_encode_legacy_table_migration_count( LegacyTableMigrationCount self, @@ -1063,12 +1028,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { SseSerializer serializer, ); - @protected - void sse_encode_list_legacy_signal_pre_key( - List self, - SseSerializer serializer, - ); - @protected void sse_encode_list_legacy_table_migration_count( List self, diff --git a/lib/core/frb_generated.web.dart b/lib/core/frb_generated.web.dart index c6bf8050..112e04a6 100644 --- a/lib/core/frb_generated.web.dart +++ b/lib/core/frb_generated.web.dart @@ -58,12 +58,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { dynamic raw, ); - @protected - FutureOr> Function() - dco_decode_DartFn_Inputs__Output_list_legacy_signal_pre_key_AnyhowException( - dynamic raw, - ); - @protected FutureOr Function(PlatformInt64) dco_decode_DartFn_Inputs_i_64_Output_unit_AnyhowException(dynamic raw); @@ -212,9 +206,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { dynamic raw, ); - @protected - LegacySignalPreKey dco_decode_legacy_signal_pre_key(dynamic raw); - @protected LegacyTableMigrationCount dco_decode_legacy_table_migration_count( dynamic raw, @@ -226,9 +217,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { @protected List dco_decode_list_frb_pqc_pre_key(dynamic raw); - @protected - List dco_decode_list_legacy_signal_pre_key(dynamic raw); - @protected List dco_decode_list_legacy_table_migration_count( dynamic raw, @@ -564,11 +552,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { SseDeserializer deserializer, ); - @protected - LegacySignalPreKey sse_decode_legacy_signal_pre_key( - SseDeserializer deserializer, - ); - @protected LegacyTableMigrationCount sse_decode_legacy_table_migration_count( SseDeserializer deserializer, @@ -582,11 +565,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { SseDeserializer deserializer, ); - @protected - List sse_decode_list_legacy_signal_pre_key( - SseDeserializer deserializer, - ); - @protected List sse_decode_list_legacy_table_migration_count( SseDeserializer deserializer, @@ -822,13 +800,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { SseSerializer serializer, ); - @protected - void - sse_encode_DartFn_Inputs__Output_list_legacy_signal_pre_key_AnyhowException( - FutureOr> Function() self, - SseSerializer serializer, - ); - @protected void sse_encode_DartFn_Inputs_i_64_Output_unit_AnyhowException( FutureOr Function(PlatformInt64) self, @@ -1044,12 +1015,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { SseSerializer serializer, ); - @protected - void sse_encode_legacy_signal_pre_key( - LegacySignalPreKey self, - SseSerializer serializer, - ); - @protected void sse_encode_legacy_table_migration_count( LegacyTableMigrationCount self, @@ -1065,12 +1030,6 @@ abstract class RustLibApiImplPlatform extends BaseApiImpl { SseSerializer serializer, ); - @protected - void sse_encode_list_legacy_signal_pre_key( - List self, - SseSerializer serializer, - ); - @protected void sse_encode_list_legacy_table_migration_count( List self, diff --git a/lib/main.dart b/lib/main.dart index 9ad68f61..cc8fdfbf 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -153,7 +153,6 @@ void main() async { binding.addPostFrameCallback((_) async { await Future.delayed(const Duration(seconds: 1)); unawaited(postStartupTasks()); - unawaited(apiService.connect()); }); } diff --git a/lib/src/callbacks/callbacks.dart b/lib/src/callbacks/callbacks.dart index d700b2a0..7980053d 100644 --- a/lib/src/callbacks/callbacks.dart +++ b/lib/src/callbacks/callbacks.dart @@ -33,7 +33,6 @@ Future initFlutterCallbacksForRust() async { loggingGetStreamSink: LoggingCallbacks.getStreamSink, legacySignalDecrypt: LegacySignalCallbacks.decrypt, legacySignalEncrypt: LegacySignalCallbacks.encrypt, - legacySignalGeneratePrekeys: LegacySignalCallbacks.generatePrekeys, apiResyncSignalSession: handleSessionResync, apiMediaAction: _apiMediaAction, apiVerificationProof: KeyVerificationService.handleVerificationProof, diff --git a/lib/src/callbacks/legacy_signal.callbacks.dart b/lib/src/callbacks/legacy_signal.callbacks.dart index 63a12887..72bc6d1d 100644 --- a/lib/src/callbacks/legacy_signal.callbacks.dart +++ b/lib/src/callbacks/legacy_signal.callbacks.dart @@ -3,7 +3,6 @@ import 'package:twonly/core/bridge/callbacks.dart'; import 'package:twonly/src/model/protobuf/client/generated/messages.pb.dart' as pb; import 'package:twonly/src/services/signal/encryption.signal.dart'; -import 'package:twonly/src/services/signal/identity.signal.dart'; import 'package:twonly/src/utils/log.dart'; /// Flutter boundary for the legacy libsignal_protocol_dart implementation. @@ -62,25 +61,6 @@ abstract final class LegacySignalCallbacks { } } - static Future> generatePrekeys() async { - try { - final prekeys = await signalGetPreKeys(); - return prekeys - .map( - (prekey) => LegacySignalPreKey( - id: prekey.id, - publicKey: Uint8List.fromList( - prekey.getKeyPair().publicKey.serialize(), - ), - ), - ) - .toList(growable: false); - } catch (error) { - Log.error('Legacy Signal prekey callback failed: $error'); - return const []; - } - } - static bool _isLegacyMessageType(int messageType) => messageType == pb.Message_Type.CIPHERTEXT.value || messageType == pb.Message_Type.PREKEY_BUNDLE.value; diff --git a/lib/src/providers/connection.provider.dart b/lib/src/providers/connection.provider.dart index 4814a3c7..c41790b0 100644 --- a/lib/src/providers/connection.provider.dart +++ b/lib/src/providers/connection.provider.dart @@ -5,15 +5,25 @@ import 'package:twonly/locator.dart'; class CustomChangeProvider with ChangeNotifier, DiagnosticableTreeMixin { CustomChangeProvider() { // The API is connected before the subscription has started so ensure that the connection state is correct - _isConnected = apiService.isConnected; - _connSub = apiService.onConnectionStateUpdated.listen( - updateConnectionState, - ); + _isConnected = false; + unawaited(_loadConnectionState()); + _connSub = apiService.events + .where((event) => event.kind == ApiEventKind.connectionStateChanged) + .listen( + (event) => updateConnectionState( + event.state == ApiConnectionState.authenticated, + ), + ); } late bool _isConnected; - late StreamSubscription _connSub; + late StreamSubscription _connSub; bool get isConnected => _isConnected; + Future _loadConnectionState() async { + final state = await RustApi.connectionState(); + await updateConnectionState(state == ApiConnectionState.authenticated); + } + @override void dispose() { _connSub.cancel(); diff --git a/lib/src/providers/purchases.provider.dart b/lib/src/providers/purchases.provider.dart index 9700583d..706aae68 100644 --- a/lib/src/providers/purchases.provider.dart +++ b/lib/src/providers/purchases.provider.dart @@ -37,14 +37,21 @@ class PurchasesProvider with ChangeNotifier, DiagnosticableTreeMixin { onError: _updateStreamOnError, ); - _planSub = apiService.onPlanUpdated.listen(updatePlan); - _connSub = apiService.onConnectionStateUpdated.listen((_) async { - try { - if (userService.isUserCreated) { - updatePlan(planFromString(userService.currentUser.subscriptionPlan)); + _apiSub = apiService.events.listen((event) async { + if (event.kind == ApiEventKind.planUpdated) { + updatePlan(planFromString(event.message ?? '')); + } + + if (event.kind == ApiEventKind.connectionStateChanged) { + try { + if (userService.isUserCreated) { + updatePlan( + planFromString(userService.currentUser.subscriptionPlan), + ); + } + } catch (e) { + Log.error(e); } - } catch (e) { - Log.error(e); } }); @@ -63,8 +70,7 @@ class PurchasesProvider with ChangeNotifier, DiagnosticableTreeMixin { late StreamSubscription> _subscription; final InAppPurchase iapConnection = IAPConnection.instance; - late StreamSubscription _planSub; - late StreamSubscription _connSub; + late StreamSubscription _apiSub; bool _userTriggeredBuyButton = false; void updatePlan(SubscriptionPlan newPlan) { @@ -228,7 +234,8 @@ class PurchasesProvider with ChangeNotifier, DiagnosticableTreeMixin { if (currentPlan != SubscriptionPlan.Family.name && currentPlan != SubscriptionPlan.Pro.name) { for (var i = 0; i < 100; i++) { - if (apiService.isAuthenticated) { + if (await RustApi.connectionState() == + ApiConnectionState.authenticated) { Log.info( 'current user does not have a sub: ${purchaseDetails.productID}', ); @@ -251,8 +258,7 @@ class PurchasesProvider with ChangeNotifier, DiagnosticableTreeMixin { @override void dispose() { - _planSub.cancel(); - _connSub.cancel(); + _apiSub.cancel(); _subscription.cancel(); super.dispose(); } diff --git a/lib/src/services/api/api.service.dart b/lib/src/services/api/api.service.dart index 92fe8253..01b08c25 100644 --- a/lib/src/services/api/api.service.dart +++ b/lib/src/services/api/api.service.dart @@ -1,125 +1,58 @@ import 'dart:async'; import 'package:connectivity_plus/connectivity_plus.dart'; -import 'package:flutter/foundation.dart'; import 'package:twonly/core/bridge/api.dart' as rust_api; import 'package:twonly/globals.dart'; import 'package:twonly/locator.dart'; -import 'package:twonly/src/services/api/mediafiles/download.api.dart'; import 'package:twonly/src/services/api/mediafiles/upload.api.dart'; -import 'package:twonly/src/services/flame.service.dart'; -import 'package:twonly/src/services/group.service.dart'; import 'package:twonly/src/services/memories/memories_cloud.service.dart'; import 'package:twonly/src/services/notifications/fcm.notifications.dart'; -import 'package:twonly/src/services/signal/identity.signal.dart'; import 'package:twonly/src/services/signal/protocol_state.signal.dart'; -import 'package:twonly/src/services/subscription.service.dart'; -import 'package:twonly/src/services/user_discovery.service.dart'; import 'package:twonly/src/utils/log.dart'; /// The ApiProvider is responsible for communicating with the server. /// It handles errors and does automatically tries to reconnect on /// errors or network changes. class ApiService { - ApiService(); - final String apiHost = kReleaseMode ? 'api.twonly.eu' : 'dev-api.twonly.eu'; - // final String apiHost = kReleaseMode ? 'api.twonly.eu' : 'dev.twonly.eu'; - final String apiSecure = kReleaseMode ? 's' : 's'; - - String get apiEndpoint => 'http$apiSecure://$apiHost/api/'; - - final _planUpdateController = StreamController.broadcast(); - Stream get onPlanUpdated => _planUpdateController.stream; - - final _connectionStateController = StreamController.broadcast(); - Stream get onConnectionStateUpdated => - _connectionStateController.stream; - - final _appOutdatedController = StreamController.broadcast(); - Stream get onAppOutdated => _appOutdatedController.stream; - - final _newDeviceRegisteredController = StreamController.broadcast(); - Stream get onNewDeviceRegistered => - _newDeviceRegisteredController.stream; - - bool appIsOutdated = false; - bool isAuthenticated = false; - bool isConnected = false; - - // ignore: cancel_subscriptions + ApiService() { + events = rust_api.RustApi.events().asBroadcastStream(); + _apiEventSubscription = events.listen( + _handleApiEvent, + onError: (Object error, StackTrace stackTrace) { + Log.error( + 'Rust API event stream failed', + error: error, + stackTrace: stackTrace, + ); + }, + ); + } + late final Stream events; StreamSubscription>? _connectivitySubscription; + late final StreamSubscription _apiEventSubscription; + + Future _handleApiEvent(ApiEvent event) async { + if (event.kind == ApiEventKind.authenticated) { + await onAuthenticated(); + } + } // Function is called after the user is authenticated at the server Future onAuthenticated() async { await FcmNotificationService.initFCMAfterAuthenticated(); - _connectionStateController.add(true); if (AppState.isInBackgroundTask) { - await RustApi.retransmitAllMessages(); await reuploadMediaFiles(); - await tryDownloadAllMediaFiles(); } else if (!AppState.isAppInBackground) { - unawaited(RustApi.retransmitAllMessages()); - unawaited(tryDownloadAllMediaFiles()); unawaited(reuploadMediaFiles()); twonlyDB.markUpdated(); - unawaited(syncFlameCounters()); - unawaited(SignalIdentityService.onAuthenticated()); resetResyncedUsers(); // resetUserDiscoveryRequestUpdates(); - unawaited(fetchGroupStatesForUnjoinedGroups()); - unawaited(fetchMissingGroupPublicKey()); - unawaited(rust_api.RustApi.checkForDeletedUsernames()); - unawaited(RustApi.performPasswordlessRecoveryHeartbeat()); - - unawaited(UserDiscoveryService.checkForNewAnnouncedUsers()); memoriesCloudService.init(); } } - Future connect() async { - try { - await rust_api.RustApi.connect(); - for (var attempt = 0; attempt < 100; attempt++) { - final state = await rust_api.RustApi.connectionState(); - isConnected = - state == ApiConnectionState.connected || - state == ApiConnectionState.authenticated; - isAuthenticated = state == ApiConnectionState.authenticated; - if (isAuthenticated || - (!userService.isUserCreated && - state == ApiConnectionState.connected)) { - return true; - } - if (state == ApiConnectionState.permanentlyRejected || - state == ApiConnectionState.suspended) { - return false; - } - await Future.delayed(const Duration(milliseconds: 100)); - } - return false; - } catch (error) { - isConnected = false; - isAuthenticated = false; - Log.error('Rust API connection failed', error: error); - return false; - } - } - - Future close(VoidCallback? callback) async { - await rust_api.RustApi.close(); - isConnected = false; - isAuthenticated = false; - _connectionStateController.add(false); - callback?.call(); - } - - Future authenticate() async { - await rust_api.RustApi.reloadConfiguration(); - await connect(); - } - Future listenToNetworkChanges() async { if (_connectivitySubscription != null) { return; @@ -127,10 +60,14 @@ class ApiService { _connectivitySubscription = Connectivity().onConnectivityChanged.listen(( result, ) async { - if (!result.contains(ConnectivityResult.none)) { - await connect(); - } - // Received changes in available connectivity types! + await rust_api.RustApi.setNetworkAvailable( + available: !result.contains(ConnectivityResult.none), + ); }); } + + Future dispose() async { + await _connectivitySubscription?.cancel(); + await _apiEventSubscription.cancel(); + } } diff --git a/lib/src/services/api/mediafiles/upload.api.dart b/lib/src/services/api/mediafiles/upload.api.dart index e8ba7f39..04b71793 100644 --- a/lib/src/services/api/mediafiles/upload.api.dart +++ b/lib/src/services/api/mediafiles/upload.api.dart @@ -779,8 +779,7 @@ Future _uploadUploadRequest(MediaFileService media) async { return; } - final apiUrl = - 'http${apiService.apiSecure}://${apiService.apiHost}/api/upload'; + final apiUrl = '${RustApi.apiBaseUrl(protocol: 'https')}upload'; Log.info('Starting upload from ${media.mediaFile.mediaId}'); diff --git a/lib/src/services/background/callback_dispatcher.background.dart b/lib/src/services/background/callback_dispatcher.background.dart index 1b57c9d7..ec48649e 100644 --- a/lib/src/services/background/callback_dispatcher.background.dart +++ b/lib/src/services/background/callback_dispatcher.background.dart @@ -127,17 +127,20 @@ Future backgroundFetch({ final stopwatch = Stopwatch()..start(); - // Issue: Because the background isolate can be reused across multiple periodic tasks, - // the API connection state might be stale or disconnected from a previous run. - // Explicitly close it here to ensure a clean slate before connecting. - await apiService.close(null); - - if (!await apiService.connect()) { - Log.info('Could not connect to the api. Returning early.'); - return false; + var authenticated = false; + for (var attempt = 0; attempt < 100; attempt++) { + final state = await RustApi.connectionState(); + if (state == ApiConnectionState.authenticated) { + authenticated = true; + break; + } + if (state == ApiConnectionState.permanentlyRejected || + state == ApiConnectionState.suspended) { + break; + } + await Future.delayed(const Duration(milliseconds: 100)); } - - if (!apiService.isAuthenticated) { + if (!authenticated) { Log.info('Api is not authenticated. Returning early.'); return false; } @@ -164,7 +167,7 @@ Future backgroundFetch({ await Future.delayed(const Duration(milliseconds: 2000)); } } finally { - await apiService.close(() {}); + await RustApi.close(); stopwatch.stop(); } diff --git a/lib/src/services/backup.service.dart b/lib/src/services/backup.service.dart index f122ae99..91d4288a 100644 --- a/lib/src/services/backup.service.dart +++ b/lib/src/services/backup.service.dart @@ -23,10 +23,10 @@ class BackupService { static final Mutex _protected = Mutex(); static String _getIdentityBackupUrl(String backupId) => - '${apiService.apiEndpoint}/backup/identity/$backupId'; + '${RustApi.apiBaseUrl(protocol: 'https')}backup/identity/$backupId'; static String _getArchiveBackupUrl(String backupDownloadToken, int? userId) => - '${apiService.apiEndpoint}/backup/archive/${userId == null ? '' : '${userId.toRadixString(16).padLeft(16, '0').toUpperCase()}/'}$backupDownloadToken'; + '${RustApi.apiBaseUrl(protocol: 'https')}backup/archive/${userId == null ? '' : '${userId.toRadixString(16).padLeft(16, '0').toUpperCase()}/'}$backupDownloadToken'; static final _backupUpdateController = StreamController.broadcast(); static Stream get onBackupUpdated => _backupUpdateController.stream; diff --git a/lib/src/services/notifications/fcm.notifications.dart b/lib/src/services/notifications/fcm.notifications.dart index 9323408a..73fd37f1 100644 --- a/lib/src/services/notifications/fcm.notifications.dart +++ b/lib/src/services/notifications/fcm.notifications.dart @@ -97,7 +97,8 @@ class FcmNotificationService { ..updateFcmToken = true ..fcmToken = fcmToken; }); - if (apiService.isAuthenticated) { + if (await RustApi.connectionState() == + ApiConnectionState.authenticated) { if (await _uploadFcmToken(fcmToken)) { await UserService.update((u) { u.updateFcmToken = false; @@ -114,7 +115,8 @@ class FcmNotificationService { ..updateFcmToken = true ..fcmToken = fcmToken; }); - if (apiService.isAuthenticated) { + if (await RustApi.connectionState() == + ApiConnectionState.authenticated) { if (await _uploadFcmToken(fcmToken)) { await UserService.update((u) { u.updateFcmToken = false; @@ -161,7 +163,10 @@ class FcmNotificationService { // This is just a workarround until the new Rust decryption is enrolled fully. final pushDataString = message.data['push_data'] as String?; if (pushDataString != null) { - if (apiService.isConnected) { + final apiState = await RustApi.connectionState(); + if (apiState == ApiConnectionState.connected || + apiState == ApiConnectionState.authenticating || + apiState == ApiConnectionState.authenticated) { Log.info('Got FCM message, but API is connected...'); } else { Log.info('Trying to connect to the API in the background.'); diff --git a/lib/src/services/signal/identity.signal.dart b/lib/src/services/signal/identity.signal.dart index 23122682..1fc5419b 100644 --- a/lib/src/services/signal/identity.signal.dart +++ b/lib/src/services/signal/identity.signal.dart @@ -1,99 +1,13 @@ import 'dart:typed_data'; -import 'package:clock/clock.dart'; import 'package:libsignal_protocol_dart/libsignal_protocol_dart.dart'; import 'package:twonly/core/bridge/wrapper/key_manager.dart'; -import 'package:twonly/core/bridge/wrapper/signal.dart'; -import 'package:twonly/locator.dart'; import 'package:twonly/src/database/signal/signal_signed_pre_key_store.dart'; import 'package:twonly/src/model/json/signal_identity.model.dart'; import 'package:twonly/src/services/signal/consts.signal.dart'; -import 'package:twonly/src/services/signal/protocol_state.signal.dart'; import 'package:twonly/src/services/signal/utils.signal.dart'; -import 'package:twonly/src/services/user.service.dart'; import 'package:twonly/src/utils/log.dart'; -class SignalIdentityService { - static Future onAuthenticated() async { - final now = clock.now(); - final fortyEightHoursAgo = now.subtract(const Duration(hours: 48)); - final oneWeekAgo = now.subtract(const Duration(days: 7)); - - if (userService.currentUser.signalLastSignedPreKeyUpdated == null || - !userService.currentUser.signalLastSignedPreKeyUpdated!.isAfter( - fortyEightHoursAgo, - )) { - final signedPreKey = await _getNewSignalSignedPreKey(); - if (signedPreKey == null) { - Log.error('could not generate a new signed pre key!'); - } else { - await UserService.update((user) { - user.signalLastSignedPreKeyUpdated = now; - }); - final res = await rustApiResult( - RustApi.updateSignedPreKey( - id: signedPreKey.id, - key: signedPreKey.getKeyPair().publicKey.serialize(), - signature: signedPreKey.signature, - ), - ); - if (res.isError) { - Log.error('could not update the signed pre key: ${res.error}'); - await UserService.update((user) { - user.signalLastSignedPreKeyUpdated = null; - }); - } else { - Log.info('updated signed pre key'); - } - } - } - - if (userService.currentUser.signalLastPqcPreKeysUploaded == null || - !userService.currentUser.signalLastPqcPreKeysUploaded!.isAfter( - oneWeekAgo, - )) { - final bundle = await RustSignal.generateBundle(); - - final pqcRes = await rustApiResult( - RustApi.uploadPqcPreKeys( - eccSignedPrekeyId: bundle.signedPreKeyId, - eccSignedPrekey: bundle.signedPreKeyPublic, - eccSignedPrekeySignature: bundle.signedPreKeySignature, - kyberSignedPrekeyId: bundle.kyberPreKeyId, - kyberSignedPrekey: bundle.kyberPreKeyPublic, - kyberSignedPrekeySignature: bundle.kyberPreKeySignature, - prekeys: const [], - ), - ); - - if (pqcRes.isError) { - Log.warn('could not update the pqc signed pre key: ${pqcRes.error}'); - } else { - Log.info('updated pqc signed pre key'); - await UserService.update((user) { - user.signalLastPqcPreKeysUploaded = now; - }); - } - } - } -} - -Future> signalGetPreKeys() async { - return lockingSignalProtocol.protect(() async { - final start = userService.currentUser.currentPreKeyIndexStart; - await UserService.update((u) { - u.currentPreKeyIndexStart = (u.currentPreKeyIndexStart + 200) % maxValue; - }); - final preKeys = generatePreKeys(start, 200); - final signalStore = await getSignalStore(); - if (signalStore == null) return []; - for (final p in preKeys) { - await signalStore.preKeyStore.storePreKey(p.id, p); - } - return preKeys; - }); -} - Future getSignalIdentity() async { try { final identity = await RustKeyManager.getSignalIdentity(); @@ -156,30 +70,3 @@ Future createIfNotExistsSignalIdentity() async { signedPreKeyStore: const {}, ); } - -Future _getNewSignalSignedPreKey() async { - return lockingSignalProtocol.protect(() async { - var identityKeyPair = await getSignalIdentityKeyPair(); - final signalStore = await getSignalStore(); - if (identityKeyPair == null || signalStore == null) { - return null; - } - - final signedPreKeyId = - userService.currentUser.currentSignedPreKeyIndexStart; - await UserService.update((user) { - user.currentSignedPreKeyIndexStart += 1; - }); - - final signedPreKey = generateSignedPreKey( - identityKeyPair, - signedPreKeyId, - ); - - identityKeyPair = null; - - await signalStore.storeSignedPreKey(signedPreKeyId, signedPreKey); - - return signedPreKey; - }); -} diff --git a/lib/src/visual/components/app_outdated.comp.dart b/lib/src/visual/components/app_outdated.comp.dart index e3d172e5..d913eb40 100644 --- a/lib/src/visual/components/app_outdated.comp.dart +++ b/lib/src/visual/components/app_outdated.comp.dart @@ -19,33 +19,25 @@ class _AppOutdatedCompState extends State { bool appIsOutdated = false; bool newDeviceRegistered = false; - late StreamSubscription _subOutdated; - late StreamSubscription _subNewDevice; + late StreamSubscription _apiEventSubscription; @override void dispose() { - _subOutdated.cancel(); - _subNewDevice.cancel(); + _apiEventSubscription.cancel(); super.dispose(); } @override void initState() { super.initState(); - _subOutdated = apiService.onAppOutdated.listen((_) async { - if (mounted) { + _apiEventSubscription = apiService.events.listen((event) async { + if (!mounted) return; + if (event.kind == ApiEventKind.appOutdated || + event.kind == ApiEventKind.newDeviceRegistered) { await context.read().updateConnectionState(false); setState(() { - appIsOutdated = true; - }); - } - }); - - _subNewDevice = apiService.onNewDeviceRegistered.listen((_) async { - if (mounted) { - await context.read().updateConnectionState(false); - setState(() { - newDeviceRegistered = true; + appIsOutdated = event.kind == ApiEventKind.appOutdated; + newDeviceRegistered = event.kind == ApiEventKind.newDeviceRegistered; }); } }); diff --git a/lib/src/visual/views/chats/chat_list.view.dart b/lib/src/visual/views/chats/chat_list.view.dart index 9da10eda..b61c95c6 100644 --- a/lib/src/visual/views/chats/chat_list.view.dart +++ b/lib/src/visual/views/chats/chat_list.view.dart @@ -361,8 +361,8 @@ class _ChatListViewState extends State ), body: RefreshIndicator( onRefresh: () async { - await apiService.close(() {}); - await apiService.connect(); + await RustApi.close(); + await RustApi.connect(); await Future.delayed(const Duration(seconds: 1)); }, child: Column( diff --git a/lib/src/visual/views/onboarding/register.view.dart b/lib/src/visual/views/onboarding/register.view.dart index ae71e3c1..3801791d 100644 --- a/lib/src/visual/views/onboarding/register.view.dart +++ b/lib/src/visual/views/onboarding/register.view.dart @@ -176,7 +176,7 @@ class _RegisterViewState extends State { unawaited(FcmNotificationService.initAfterUserLoaded()); - await apiService.authenticate(); + await RustApi.reloadConfiguration(); widget.callbackOnSuccess(); } catch (e, stack) { Log.error('Error creating new user', error: e, stackTrace: stack); diff --git a/lib/src/visual/views/settings/help/contact_us.view.dart b/lib/src/visual/views/settings/help/contact_us.view.dart index 31e7a6a2..7caf6a8d 100644 --- a/lib/src/visual/views/settings/help/contact_us.view.dart +++ b/lib/src/visual/views/settings/help/contact_us.view.dart @@ -68,8 +68,7 @@ class _ContactUsState extends State { final uploadRequestBytes = uploadRequest.writeToBuffer(); - final apiUrl = - 'http${apiService.apiSecure}://${apiService.apiHost}/api/upload'; + final apiUrl = '${RustApi.apiBaseUrl(protocol: 'https')}upload'; final requestMultipart = http.MultipartRequest('POST', Uri.parse(apiUrl)); diff --git a/rust/src/api/messages/incoming/mod.rs b/rust/src/api/messages/incoming/mod.rs index d19bd77a..05136fb6 100644 --- a/rust/src/api/messages/incoming/mod.rs +++ b/rust/src/api/messages/incoming/mod.rs @@ -29,15 +29,6 @@ pub(crate) async fn handle_server_message( kind: server_to_client::v0::Kind, ) -> Result { let ok = match kind { - // These booleans are presence-only request markers. Their value is not part of - // the protocol; the Dart client likewise checks hasRequestNewPreKeys() only. - Kind::RequestNewPreKeys(_) => match handle_request_new_prekeys(ctx).await { - Ok(response) => response, - Err(error) => { - tracing::error!("failed to generate requested prekeys: {error}"); - ok::Ok::None(true) - } - }, Kind::RequestNewPqcPreKeys(_) => match handle_request_new_pqc_prekeys(ctx).await { Ok(response) => response, Err(error) => { @@ -107,40 +98,6 @@ pub(crate) async fn handle_sealed_message(ctx: &Arc, bytes: Vec) -> handle_decoded_server_message(ctx, payload.from_user_id, message).await } -pub(crate) async fn handle_request_new_prekeys( - ctx: &Arc, -) -> Result { - let prekeys = match get_callbacks() { - Ok(callbacks) => (callbacks.legacy_signal.generate_prekeys)() - .await - .into_iter() - .map(|key| client_to_server::response::PreKey { - id: key.id, - prekey: key.public_key, - }) - .collect(), - Err(TwonlyError::MissingCallbackInitialization) => { - let engine = ctx.get_signal_engine().lock().await; - engine - .as_ref() - .ok_or(TwonlyError::SignalIdentityNotFound)? - .generate_prekeys(200) - .await? - .into_iter() - .map(|(id, public_key)| client_to_server::response::PreKey { - id: i64::from(id), - prekey: public_key, - }) - .collect() - } - Err(error) => return Err(error), - }; - - Ok(client_to_server::response::ok::Ok::Prekeys( - client_to_server::response::Prekeys { prekeys }, - )) -} - pub(crate) async fn handle_request_new_pqc_prekeys( ctx: &Arc, ) -> Result { diff --git a/rust/src/api/runtime.rs b/rust/src/api/runtime.rs index ae2b51df..ef644abd 100644 --- a/rust/src/api/runtime.rs +++ b/rust/src/api/runtime.rs @@ -38,7 +38,7 @@ impl ApiRuntime { .ok_or(TwonlyError::Initialization)?; *slot.write().await = replacement; current.close().await; - Ok(()) + Self::client(ctx).await?.connect().await } pub async fn connect(ctx: &Arc) -> Result<()> { diff --git a/rust/src/api/runtime/client.rs b/rust/src/api/runtime/client.rs index 37b2b522..1b5130d6 100644 --- a/rust/src/api/runtime/client.rs +++ b/rust/src/api/runtime/client.rs @@ -19,6 +19,7 @@ pub(super) type PendingRequests = Arc pub(crate) static API_EVENTS: LazyLock> = LazyLock::new(|| broadcast::channel(256).0); +pub(crate) static API_PERMANENTLY_REJECTED: AtomicBool = AtomicBool::new(false); pub(crate) struct ApiClient { pub(crate) context: Weak, @@ -65,6 +66,13 @@ impl ApiClient { } pub async fn connect(self: &Arc) -> Result<()> { + if API_PERMANENTLY_REJECTED.load(Ordering::Acquire) { + self.set_state(ApiConnectionState::PermanentlyRejected) + .await; + return Err(crate::error::TwonlyError::Generic( + "API connection was permanently rejected for this process".into(), + )); + } self.deliberately_closed.store(false, Ordering::Release); let mut client_guard = self.ws_client.lock().await; @@ -139,7 +147,11 @@ impl ApiClient { } Ok(ConnectionEvent::Disconnected { .. }) => { self_clone.is_authenticated.store(false, Ordering::Release); - self_clone.set_state(ApiConnectionState::Stopped).await; + if API_PERMANENTLY_REJECTED.load(Ordering::Acquire) { + self_clone.set_state(ApiConnectionState::PermanentlyRejected).await; + } else { + self_clone.set_state(ApiConnectionState::Stopped).await; + } } Ok(ConnectionEvent::Connecting { .. }) => { self_clone.set_state(ApiConnectionState::Connecting).await; diff --git a/rust/src/api/runtime/helpers.rs b/rust/src/api/runtime/helpers.rs index 75d504b5..02ee4a62 100644 --- a/rust/src/api/runtime/helpers.rs +++ b/rust/src/api/runtime/helpers.rs @@ -3,13 +3,16 @@ * */ -use crate::api::messages::incoming::client2client::messages; +use crate::api::messages::incoming::client2client::{messages, recovery}; use crate::api::proto::server_to_client; use crate::api::runtime::ApiRuntime; +use crate::api::Server; use crate::bridge::api::ServerResult; use crate::context::Context; use crate::error::{Result, TwonlyError}; +use crate::services::groups::GroupService; +use crate::services::mediafiles::MediaFileService; use prost::Message as ProstMessage; use std::future::Future; use std::pin::Pin; @@ -37,9 +40,6 @@ pub(crate) fn response_error_code(bytes: &[u8]) -> Result> { } pub(crate) fn schedule_post_authentication(ctx: &Arc, in_background: bool) { - if in_background { - return; - } let ctx = ctx.clone(); tokio::spawn(async move { // Wait a bit to let other initial state settle @@ -50,12 +50,50 @@ pub(crate) fn schedule_post_authentication(ctx: &Arc, in_background: bo if let Err(error) = replay.await { tracing::warn!("failed to replay API outbox: {error}"); } + if let Err(error) = ApiRuntime::replay_legacy_raw_outbox(&ctx).await { tracing::warn!("failed to replay legacy raw-byte outbox: {error}"); } + if let Err(error) = messages::retransmit_queued_receipts(&ctx).await { tracing::warn!("failed to retransmit queued receipts: {error}"); } + if let Err(error) = MediaFileService::new(&ctx).download_pending().await { + tracing::warn!("failed to download pending media: {error}"); + } + + if in_background { + return; + } + + if let Err(error) = GroupService::new(&ctx).on_connected().await { + tracing::warn!("group post-connection maintenance failed: {error}"); + } + + if let Err(error) = Server::check_for_deleted_usernames(&ctx).await { + tracing::warn!("deleted-username refresh failed: {error}"); + } + + if let Err(error) = recovery::perform_heartbeat(&ctx).await { + tracing::warn!("passwordless recovery heartbeat failed: {error}"); + } + + if let Err(error) = ctx + .get_user_discovery() + .get() + .await + .on_connected(&ctx) + .await + { + tracing::warn!("user-discovery post-connection refresh failed: {error}"); + } + + let signal_engine = ctx.get_signal_engine().lock().await; + if let Some(engine) = signal_engine.as_ref() { + if let Err(error) = engine.on_connected(&ctx).await { + tracing::warn!("Signal key maintenance failed: {error}"); + } + } }); } diff --git a/rust/src/api/runtime/request.rs b/rust/src/api/runtime/request.rs index 775249b2..561f2721 100644 --- a/rust/src/api/runtime/request.rs +++ b/rust/src/api/runtime/request.rs @@ -3,7 +3,7 @@ * */ -use super::client::ApiClient; +use super::client::{ApiClient, API_PERMANENTLY_REJECTED}; use crate::api::messages::incoming::handle_server_message; use crate::api::proto::{client_to_server, server_to_client}; use crate::bridge::api::{ApiConnectionState, ApiEvent, ApiEventKind}; @@ -203,6 +203,7 @@ impl ApiClient { { self.set_state(ApiConnectionState::PermanentlyRejected) .await; + API_PERMANENTLY_REJECTED.store(true, Ordering::Release); self.deliberately_closed.store(true, Ordering::Release); let kind = if code == ErrorCode::AppVersionOutdated as i32 { ApiEventKind::AppOutdated @@ -214,8 +215,10 @@ impl ApiClient { state: Some(ApiConnectionState::PermanentlyRejected), message: None, }); - if let Some(_) = self.ws_client.lock().await.take() { - // Drop client to disconnect + if let Some(client) = self.ws_client.lock().await.take() { + if let Err(error) = client.shutdown_graceful(Duration::from_secs(5)).await { + tracing::warn!(%error, "permanently rejected WebSocket did not shut down cleanly"); + } } } if code == ErrorCode::UserIdNotFound as i32 { diff --git a/rust/src/bridge/callbacks.rs b/rust/src/bridge/callbacks.rs index 704a935a..27dd7f06 100644 --- a/rust/src/bridge/callbacks.rs +++ b/rust/src/bridge/callbacks.rs @@ -35,12 +35,6 @@ pub struct LegacySignalEncryptResult { pub message_type: i32, } -#[derive(Clone, Debug)] -pub struct LegacySignalPreKey { - pub id: i64, - pub public_key: Vec, -} - // This will also generate the function init_flutter_callbacks which MUST be called from Flutter to initialize the callbacks callback_generator! { FlutterCallbacks { @@ -49,8 +43,7 @@ callback_generator! { }, LegacySignal legacy_signal { decrypt: (i64, Vec, i32) => LegacySignalDecryptResult, - encrypt: (i64, Vec) => Option, - generate_prekeys: () => Vec + encrypt: (i64, Vec) => Option }, Api api { resync_signal_session: (i64) => (), diff --git a/rust/src/database/app/tables/groups.rs b/rust/src/database/app/tables/groups.rs index 164dc2ef..32568044 100644 --- a/rust/src/database/app/tables/groups.rs +++ b/rust/src/database/app/tables/groups.rs @@ -15,6 +15,32 @@ const MAX_FUTURE_TIMESTAMP_SKEW_SECONDS: i64 = 10 * 60; pub struct Group; impl Group { + pub async fn flame_sync_candidates(pool: &sqlx::Pool) -> Result> { + Ok(sqlx::query_as!( + FlameSyncGroup, + r#"SELECT group_id, total_media_counter, last_flame_counter_change, + last_flame_sync, flame_counter + FROM groups WHERE last_flame_counter_change IS NOT NULL"# + ) + .fetch_all(pool) + .await?) + } + + pub async fn set_last_flame_sync( + pool: &sqlx::Pool, + group_id: &str, + timestamp: i64, + ) -> Result<()> { + sqlx::query!( + "UPDATE groups SET last_flame_sync = ? WHERE group_id = ?", + timestamp, + group_id, + ) + .execute(pool) + .await?; + Ok(()) + } + pub async fn ensure_exists(tr: &mut Transaction<'_, Sqlite>, group_id: &str) -> Result<()> { let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM groups WHERE group_id = ?)", @@ -149,6 +175,14 @@ impl Group { } } +pub struct FlameSyncGroup { + pub group_id: String, + pub total_media_counter: i64, + pub last_flame_counter_change: Option, + pub last_flame_sync: Option, + pub flame_counter: i64, +} + #[derive(bon::Builder)] pub struct InsertGroup { group_id: String, @@ -462,8 +496,5 @@ impl GetMissingGroupPublicKeys { } fn current_unix_timestamp() -> Result { - Ok(std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map_err(|error| TwonlyError::Generic(error.to_string()))? - .as_secs() as i64) + Ok(crate::utils::current_time().timestamp()) } diff --git a/rust/src/frb_generated.rs b/rust/src/frb_generated.rs index addab330..e21b0fab 100644 --- a/rust/src/frb_generated.rs +++ b/rust/src/frb_generated.rs @@ -334,7 +334,6 @@ fn wire__crate__bridge__callbacks__init_flutter_callbacks_impl( let api_logging_get_stream_sink = decode_DartFn_Inputs__Output_StreamSink_String_Sse_AnyhowException(::sse_decode(&mut deserializer)); let api_legacy_signal_decrypt = decode_DartFn_Inputs_i_64_list_prim_u_8_strict_i_32_Output_legacy_signal_decrypt_result_AnyhowException(::sse_decode(&mut deserializer)); let api_legacy_signal_encrypt = decode_DartFn_Inputs_i_64_list_prim_u_8_strict_Output_opt_box_autoadd_legacy_signal_encrypt_result_AnyhowException(::sse_decode(&mut deserializer)); -let api_legacy_signal_generate_prekeys = decode_DartFn_Inputs__Output_list_legacy_signal_pre_key_AnyhowException(::sse_decode(&mut deserializer)); let api_api_resync_signal_session = decode_DartFn_Inputs_i_64_Output_unit_AnyhowException(::sse_decode(&mut deserializer)); let api_api_media_action = decode_DartFn_Inputs_String_String_i_64_String_Output_unit_AnyhowException(::sse_decode(&mut deserializer)); let api_api_verification_proof = decode_DartFn_Inputs_i_64_list_prim_u_8_strict_Output_unit_AnyhowException(::sse_decode(&mut deserializer)); @@ -342,7 +341,7 @@ let api_api_create_push_avatars = decode_DartFn_Inputs_i_64_Output_unit_AnyhowEx let api_api_media_received = decode_DartFn_Inputs_String_i_64_Output_unit_AnyhowException(::sse_decode(&mut deserializer)); let api_api_user_config_changed = decode_DartFn_Inputs_user_config_Output_unit_AnyhowException(::sse_decode(&mut deserializer));deserializer.end(); move |context| { transform_result_sse::<_, ()>((move || { - let output_ok = Result::<_,()>::Ok({ crate::bridge::callbacks::init_flutter_callbacks(api_callback_id, api_logging_get_stream_sink, api_legacy_signal_decrypt, api_legacy_signal_encrypt, api_legacy_signal_generate_prekeys, api_api_resync_signal_session, api_api_media_action, api_api_verification_proof, api_api_create_push_avatars, api_api_media_received, api_api_user_config_changed); })?; Ok(output_ok) + let output_ok = Result::<_,()>::Ok({ crate::bridge::callbacks::init_flutter_callbacks(api_callback_id, api_logging_get_stream_sink, api_legacy_signal_decrypt, api_legacy_signal_encrypt, api_api_resync_signal_session, api_api_media_action, api_api_verification_proof, api_api_create_push_avatars, api_api_media_received, api_api_user_config_changed); })?; Ok(output_ok) })()) } }) } @@ -4202,40 +4201,6 @@ fn decode_DartFn_Inputs__Output_StreamSink_String_Sse_AnyhowException( flutter_rust_bridge::for_generated::convert_into_dart_fn_future(body(dart_opaque.clone())) } } -fn decode_DartFn_Inputs__Output_list_legacy_signal_pre_key_AnyhowException( - dart_opaque: flutter_rust_bridge::DartOpaque, -) -> impl Fn() -> flutter_rust_bridge::DartFnFuture> -{ - use flutter_rust_bridge::IntoDart; - - async fn body( - dart_opaque: flutter_rust_bridge::DartOpaque, - ) -> Vec { - let args = vec![]; - let message = FLUTTER_RUST_BRIDGE_HANDLER - .dart_fn_invoke(dart_opaque, args) - .await; - - let mut deserializer = flutter_rust_bridge::for_generated::SseDeserializer::new(message); - let action = deserializer.cursor.read_u8().unwrap(); - let ans = match action { - 0 => std::result::Result::Ok( - >::sse_decode(&mut deserializer), - ), - 1 => std::result::Result::Err( - ::sse_decode(&mut deserializer), - ), - _ => unreachable!(), - }; - deserializer.end(); - let ans = ans.expect("Dart throws exception but Rust side assume it is not failable"); - ans - } - - move || { - flutter_rust_bridge::for_generated::convert_into_dart_fn_future(body(dart_opaque.clone())) - } -} fn decode_DartFn_Inputs_i_64_Output_unit_AnyhowException( dart_opaque: flutter_rust_bridge::DartOpaque, ) -> impl Fn(i64) -> flutter_rust_bridge::DartFnFuture<()> { @@ -4728,18 +4693,6 @@ impl SseDecode for crate::bridge::callbacks::LegacySignalEncryptResult { } } -impl SseDecode for crate::bridge::callbacks::LegacySignalPreKey { - // Codec=Sse (Serialization based), see doc to use other codecs - fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self { - let mut var_id = ::sse_decode(deserializer); - let mut var_publicKey = >::sse_decode(deserializer); - return crate::bridge::callbacks::LegacySignalPreKey { - id: var_id, - public_key: var_publicKey, - }; - } -} - impl SseDecode for crate::bridge::wrapper::app_database::LegacyTableMigrationCount { // Codec=Sse (Serialization based), see doc to use other codecs fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self { @@ -4778,20 +4731,6 @@ impl SseDecode for Vec { } } -impl SseDecode for Vec { - // Codec=Sse (Serialization based), see doc to use other codecs - fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self { - let mut len_ = ::sse_decode(deserializer); - let mut ans_ = Vec::with_capacity(len_ as usize); - for idx_ in 0..len_ { - ans_.push(::sse_decode( - deserializer, - )); - } - return ans_; - } -} - impl SseDecode for Vec { // Codec=Sse (Serialization based), see doc to use other codecs fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self { @@ -5913,27 +5852,6 @@ impl flutter_rust_bridge::IntoIntoDart flutter_rust_bridge::for_generated::DartAbi { - [ - self.id.into_into_dart().into_dart(), - self.public_key.into_into_dart().into_dart(), - ] - .into_dart() - } -} -impl flutter_rust_bridge::for_generated::IntoDartExceptPrimitive - for crate::bridge::callbacks::LegacySignalPreKey -{ -} -impl flutter_rust_bridge::IntoIntoDart - for crate::bridge::callbacks::LegacySignalPreKey -{ - fn into_into_dart(self) -> crate::bridge::callbacks::LegacySignalPreKey { - self - } -} -// Codec=Dco (DartCObject based), see doc to use other codecs impl flutter_rust_bridge::IntoDart for crate::bridge::wrapper::app_database::LegacyTableMigrationCount { @@ -6651,14 +6569,6 @@ impl SseEncode for crate::bridge::callbacks::LegacySignalEncryptResult { } } -impl SseEncode for crate::bridge::callbacks::LegacySignalPreKey { - // Codec=Sse (Serialization based), see doc to use other codecs - fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) { - ::sse_encode(self.id, serializer); - >::sse_encode(self.public_key, serializer); - } -} - impl SseEncode for crate::bridge::wrapper::app_database::LegacyTableMigrationCount { // Codec=Sse (Serialization based), see doc to use other codecs fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) { @@ -6687,16 +6597,6 @@ impl SseEncode for Vec { } } -impl SseEncode for Vec { - // Codec=Sse (Serialization based), see doc to use other codecs - fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) { - ::sse_encode(self.len() as _, serializer); - for item in self { - ::sse_encode(item, serializer); - } - } -} - impl SseEncode for Vec { // Codec=Sse (Serialization based), see doc to use other codecs fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) { diff --git a/rust/src/services/groups/mod.rs b/rust/src/services/groups/mod.rs index 3542ce8d..a8d28247 100644 --- a/rust/src/services/groups/mod.rs +++ b/rust/src/services/groups/mod.rs @@ -19,7 +19,7 @@ use crate::api::server::Server; use crate::bridge::api::ServerResult; use crate::context::Context; use crate::database::app::tables::{ - Contact, GetGroupPublicKey, GetMissingGroupPublicKeys, GetUnjoinedGroups, InsertGroup, + Contact, GetGroupPublicKey, GetMissingGroupPublicKeys, GetUnjoinedGroups, Group, InsertGroup, InsertGroupHistory, InsertGroupMember, UpdateContact, UpdateGroup, UpdateGroupMemberState, }; use crate::error::{Result, TwonlyError}; @@ -38,6 +38,59 @@ pub struct GroupService { } impl GroupService { + pub async fn on_connected(&self) -> Result<()> { + self.fetch_group_states_for_unjoined_groups().await?; + self.fetch_missing_group_public_keys().await?; + self.sync_flame_counters().await + } + + async fn sync_flame_counters(&self) -> Result<()> { + let db = self.ctx.get_app_database().await; + let groups = Group::flame_sync_candidates(&db.pool).await?; + + let Some(best_friend) = groups.iter().max_by_key(|group| group.total_media_counter) else { + return Ok(()); + }; + let best_friend_id = best_friend.group_id.clone(); + let now = current_time().timestamp(); + let start_today = now - now.rem_euclid(86_400); + for group in groups { + let Some(changed) = group.last_flame_counter_change else { + continue; + }; + if changed < start_today + || group + .last_flame_sync + .is_some_and(|sync| sync >= start_today) + { + continue; + } + if group.flame_counter <= 2 && group.group_id != best_friend_id { + continue; + } + MessageService::new(&self.ctx) + .send_to_group( + group.group_id.clone(), + EncryptedContent { + flame_sync: Some(encrypted_content::FlameSync { + flame_counter: group.flame_counter, + last_flame_counter_change: changed * 1000, + best_friend: group.group_id == best_friend_id, + force_update: false, + }), + ..Default::default() + } + .encode_to_vec(), + None, + false, + ) + .await?; + Group::set_last_flame_sync(&db.pool, &group.group_id, now).await?; + } + db.notify_committed(["groups"]); + Ok(()) + } + pub fn new(ctx: &Arc) -> Self { Self { ctx: ctx.clone() } } diff --git a/rust/src/signal/engine.rs b/rust/src/signal/engine.rs index bb06d129..a68d895c 100644 --- a/rust/src/signal/engine.rs +++ b/rust/src/signal/engine.rs @@ -3,7 +3,10 @@ * */ +use crate::api::server::Server; use crate::error::{Result, TwonlyError}; +use crate::user_config::UserConfig; +use chrono::{Duration, Utc}; use libsignal_protocol::{ message_encrypt, process_prekey_bundle, CiphertextMessageType, DeviceId, GenericSignedPreKey, IdentityKey, IdentityKeyPair, IdentityKeyStore, KyberPreKeyId, KyberPreKeyStore, PreKeyBundle, @@ -17,6 +20,7 @@ use tokio::sync::Mutex; use crate::bridge::get_twonly_flutter; use crate::signal::assert_send::AssertSendFutureExt; use crate::signal::store::DbSignalProtocolStore; +use crate::utils::current_time; use rand::SeedableRng; pub struct RustSignalEngine { @@ -47,6 +51,53 @@ pub struct FrbPqcPreKey { } impl RustSignalEngine { + pub async fn on_connected(&self, ctx: &Arc) -> Result<()> { + let base = UserConfig::load_required_from(ctx)?; + let now = current_time().with_timezone(&Utc); + + let refresh_signed = base + .signal_last_signed_pre_key_updated + .is_none_or(|last| last < now - Duration::hours(48)); + let refresh_pqc = base + .signal_last_pqc_pre_keys_uploaded + .is_none_or(|last| last < now - Duration::days(7)); + + if !refresh_signed && !refresh_pqc { + return Ok(()); + } + + let bundle = self.generate_bundle().await?; + if refresh_signed { + Server::update_signed_pre_key( + ctx, + i64::from(bundle.signed_pre_key_id), + bundle.signed_pre_key_public.clone(), + bundle.signed_pre_key_signature.clone(), + ) + .await?; + UserConfig::update(ctx, |config| { + config.signal_last_signed_pre_key_updated = Some(now); + })?; + } + if refresh_pqc { + Server::upload_pqc_pre_keys( + ctx, + i64::from(bundle.signed_pre_key_id), + bundle.signed_pre_key_public, + bundle.signed_pre_key_signature, + i64::from(bundle.kyber_pre_key_id), + bundle.kyber_pre_key_public, + bundle.kyber_pre_key_signature, + Vec::new(), + ) + .await?; + UserConfig::update(ctx, |config| { + config.signal_last_pqc_pre_keys_uploaded = Some(now); + })?; + } + Ok(()) + } + pub async fn new(local_name: String) -> Result { let twonly = get_twonly_flutter()?; let pool = twonly.rust_db.read().await.pool.clone(); diff --git a/rust/src/user_config.rs b/rust/src/user_config.rs index 4e64eb9b..13fb8533 100644 --- a/rust/src/user_config.rs +++ b/rust/src/user_config.rs @@ -372,6 +372,20 @@ impl UserConfig { Self::save_json_unlocked(context, json) } + /// Atomically updates the latest persisted configuration with a typed Rust + /// mutation. Unlike `update_json`, this does not need a caller snapshot: + /// loading, mutation, and saving all happen while holding the write lock. + pub(crate) fn update(context: &Context, mutate: impl FnOnce(&mut Self)) -> Result<()> { + let _guard = config_lock() + .write() + .map_err(|_| twonly_error!("user configuration lock was poisoned"))?; + let mut config = Self::load_from_unlocked(context)? + .ok_or_else(|| twonly_error!("user configuration is unavailable"))?; + mutate(&mut config); + Self::save_unlocked(context, &config)?; + Ok(()) + } + /// Applies only fields changed relative to the caller's original snapshot. /// Concurrent updates from Flutter isolates or Rust therefore do not /// overwrite unrelated fields with stale values. @@ -399,6 +413,10 @@ impl UserConfig { let config: Self = serde_json::from_str(json).map_err(|error| { TwonlyError::Generic(format!("invalid user configuration update: {error}")) })?; + Self::save_unlocked(context, &config) + } + + fn save_unlocked(context: &Context, config: &Self) -> Result { let normalized = serde_json::to_string(&config)?; let path = Self::path(context); let parent = path diff --git a/rust/src/user_discovery.rs b/rust/src/user_discovery.rs index 592dfe86..2a266909 100644 --- a/rust/src/user_discovery.rs +++ b/rust/src/user_discovery.rs @@ -4,6 +4,9 @@ */ use std::collections::HashSet; +use crate::api::server::Server; +use crate::bridge::api::ServerResult; +use crate::context::Context; use std::path::PathBuf; use std::sync::Arc; use blahaj::{Share, Sharks}; @@ -70,6 +73,55 @@ pub struct UserDiscovery { } impl UserDiscovery { + /// Refreshes server-owned data for announcements after the API connection + /// has authenticated. Cryptographic discovery state remains owned here; + /// the API runtime only invokes this lifecycle hook. + pub async fn on_connected(&self, ctx: &Arc) -> Result<()> { + let database = ctx.get_app_database().await; + let announcements = sqlx::query!( + r#"SELECT announced_user_id, announced_public_key + FROM user_discovery_announced_users WHERE username IS NULL"# + ) + .fetch_all(&database.pool) + .await?; + + for announcement in announcements { + let user = match Server::get_user_by_id(ctx, announcement.announced_user_id).await? { + ServerResult::Ok(user) => user, + ServerResult::ErrorCode(code) => { + tracing::warn!( + user_id = announcement.announced_user_id, + code, + "could not refresh announced user" + ); + continue; + } + }; + if user.public_identity_key.as_deref() + != Some(announcement.announced_public_key.as_slice()) + { + tracing::error!( + user_id = announcement.announced_user_id, + "server returned a different identity key for announced user" + ); + continue; + } + let Some(username) = user.username else { + continue; + }; + + let username = String::from_utf8(username)?; + sqlx::query!( + "UPDATE user_discovery_announced_users SET username = ? WHERE announced_user_id = ?", + username, + announcement.announced_user_id, + ) + .execute(&database.pool) + .await?; + } + database.notify_committed(["user_discovery_announced_users"]); + Ok(()) + } pub fn new( data_dir: &str, key_manager: Arc>,