From 6f72a50361d279934888244796d5b5fcce39ab9f Mon Sep 17 00:00:00 2001 From: Xare123 <57245242+Xare123@users.noreply.github.com> Date: Fri, 24 Jul 2026 12:44:17 -0700 Subject: [PATCH 1/9] Harden message refresh and receive diagnostics --- .../widgets/tile/conversation_tile.dart | 45 +++++++++-- .../pages/messages_view.dart | 75 ++++++++++++++++--- lib/main.dart | 14 +++- lib/services/rustpush/rustpush_service.dart | 45 +++++++---- 4 files changed, 146 insertions(+), 33 deletions(-) diff --git a/lib/app/layouts/conversation_list/widgets/tile/conversation_tile.dart b/lib/app/layouts/conversation_list/widgets/tile/conversation_tile.dart index de10e12257..fcdeac6e1a 100644 --- a/lib/app/layouts/conversation_list/widgets/tile/conversation_tile.dart +++ b/lib/app/layouts/conversation_list/widgets/tile/conversation_tile.dart @@ -447,6 +447,26 @@ class _ChatSubtitleState extends CustomState { bool initialized = false; bool fetching = false; + bool _refreshing = false; late bool noMoreMessages = widget.customService != null; List _messages = []; RxList smartReplies = [].obs; RxMap internalSmartReplies = {}.obs; - late final messageService = widget.customService ?? ms(chat.guid) - ..init(chat, handleNewMessage, handleUpdatedMessage, handleDeletedMessage, jumpToMessage); + late MessagesService messageService; final smartReply = GoogleMlKit.nlp.smartReply(); final listKey = GlobalKey(); final RxBool dragging = false.obs; @@ -134,16 +134,12 @@ class MessagesViewState extends OptimizedState { @override void initState() { super.initState(); + messageService = widget.customService ?? ms(chat.guid); + messageService.init(chat, handleNewMessage, handleUpdatedMessage, handleDeletedMessage, jumpToMessage); eventDispatcher.stream.listen((e) async { if (e.item1 == "refresh-messagebloc" && e.item2 == chat.guid) { - // Clear state items - noMoreMessages = false; - _messages = []; - // Reload the state after refreshing - messageService.reload(); - messageService.init(chat, handleNewMessage, handleUpdatedMessage, handleDeletedMessage, jumpToMessage); - setState(() {}); + await _refreshMessageBloc(); } else if (e.item1 == "add-custom-smartreply") { if (e.item2 != null && internalSmartReplies['attach-recent'] == null) { internalSmartReplies['attach-recent'] = _buildReply("Attach recent photo", onTap: () async { @@ -198,6 +194,67 @@ class MessagesViewState extends OptimizedState { }); } + void _closeMessageControllers(Iterable messages) { + for (final message in messages) { + final guid = message.guid; + if (guid != null) getActiveMwc(guid)?.close(); + } + } + + void _bindMessageControllers(Iterable messages) { + if (!mounted) return; + for (final message in messages) { + if (message.guid == null) continue; + final messageController = mwc(message); + messageController.cvController = controller; + } + } + + Future _refreshMessageBloc() async { + if (_refreshing) return; + _refreshing = true; + try { + final staleMessages = List.from(_messages); + _closeMessageControllers(staleMessages); + for (var index = _messages.length - 1; index >= 0; index--) { + listKey.currentState?.removeItem( + index, + (context, animation) => const SizedBox.shrink(), + duration: Duration.zero, + ); + } + for (final node in messageFocusNodes.values) { + node.dispose(); + } + messageFocusNodes.clear(); + + noMoreMessages = false; + fetching = false; + _messages = []; + + // Re-acquire the service after its GetX reload so the view does not + // continue using stale callbacks while the transcript is repopulated. + messageService.reload(); + if (widget.customService == null) { + messageService = ms(chat.guid); + } + messageService.init(chat, handleNewMessage, handleUpdatedMessage, handleDeletedMessage, jumpToMessage); + await messageService.loadChunk(0, controller); + if (!mounted) return; + + _messages = List.from(messageService.struct.messages); + _messages.sort(Message.sort); + _bindMessageControllers(_messages); + _syncBottomMessageFocusNode(); + setState(() {}); + for (var index = 0; index < _messages.length; index++) { + listKey.currentState?.insertItem(index, duration: Duration.zero); + } + } finally { + _refreshing = false; + } + } + @override void dispose() { if (!kIsWeb && !kIsDesktop) smartReply.close(); diff --git a/lib/main.dart b/lib/main.dart index 1db315d28d..f86548e095 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -56,6 +56,16 @@ var usingRustPush = true; bool isAuthing = false; final systemTray = st.SystemTray(); +String _renderIncidentId() => Random.secure().nextInt(0x7fffffff).toRadixString(16).padLeft(8, '0'); + +void _logRenderError(FlutterErrorDetails details) { + final incidentId = _renderIncidentId(); + Logger.error( + "Render error incident=$incidentId exceptionType=${details.exception.runtimeType}", + trace: details.stack, + ); +} + @pragma('vm:entry-point') //ignore: prefer_void_to_null Future main(List arguments) async { @@ -82,7 +92,7 @@ Future initApp(bool bubble, List arguments) async { StackTrace? stacktrace; FlutterError.onError = (details) { - Logger.error("Rendering Error: ${details.exceptionAsString()}", error: details.exception, trace: details.stack); + _logRenderError(details); }; try { @@ -465,7 +475,7 @@ class _HomeState extends OptimizedState with WidgetsBindingObserver, TrayL } ErrorWidget.builder = (FlutterErrorDetails error) { - Logger.error("An unexpected error occurred when rendering.", error: error.exception, trace: error.stack); + _logRenderError(error); return CustomErrorWidget( "An unexpected error occurred when rendering.", ); diff --git a/lib/services/rustpush/rustpush_service.dart b/lib/services/rustpush/rustpush_service.dart index 6deb3c1e42..aaa8cfcab0 100644 --- a/lib/services/rustpush/rustpush_service.dart +++ b/lib/services/rustpush/rustpush_service.dart @@ -62,6 +62,10 @@ const rpApiRoot = "https://hw.openbubbles.app/code"; const clientId = '1041242226917-ik21n86fp43e82iu1e5soh6bu6gvuste.apps.googleusercontent.com'; const clientSecret = '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'; +String _diagnosticHash(String value) => sha256.convert(utf8.encode(value)).toString().substring(0, 12); + +String _durationMs(Stopwatch stopwatch) => stopwatch.elapsedMilliseconds.toString(); + class SyncIsolate { static void initialize() { @@ -3806,16 +3810,20 @@ class RustPushService extends GetxService { return; } } - Logger.info("Reflecting ${myMsg.id}"); + final receiveStopwatch = Stopwatch()..start(); + final receiveId = _diagnosticHash(myMsg.id); + Logger.info("rustpush_receive reflection_start id=$receiveId"); var reflected = await pushService.reflectMessageDyn(myMsg); - Logger.info("Reflect finished ${myMsg.id}"); + Logger.info("rustpush_receive reflection_complete id=$receiveId duration_ms=${_durationMs(receiveStopwatch)} reflected=${reflected != null}"); if (reflected != null) { - Logger.info("Queing"); + final queueStopwatch = Stopwatch()..start(); + Logger.info("rustpush_receive incoming_queue_enqueue id=$receiveId pending_count=${inq.items.length}"); await inq.queue(IncomingItem( chat: chat, message: reflected, type: QueueType.newMessage )); + Logger.info("rustpush_receive incoming_queue_complete id=$receiveId duration_ms=${_durationMs(queueStopwatch)} pending_count=${inq.items.length}"); } } @@ -4469,36 +4477,45 @@ class RustPushService extends GetxService { } } - Future markAsHandledAfter(String ptr) async { + Future markAsHandledAfter(String ptr, {required String eventId, required int retry}) async { + final ackStopwatch = Stopwatch()..start(); if (inq.isProcessing.value) { - Logger.info("Marking as handled processing wait $ptr"); + Logger.info("rustpush_receive ack_wait_start id=$eventId retry=$retry pending_count=${inq.items.length}"); await for (final value in inq.isProcessing.stream) { if (!value) break; } } - Logger.info("Marking as handled commit $ptr"); + Logger.info("rustpush_receive save_and_queue_drained id=$eventId retry=$retry wait_ms=${_durationMs(ackStopwatch)} pending_count=${inq.items.length}"); + Logger.info("rustpush_receive ack_commit id=$eventId retry=$retry"); await api.completeMsg(ptr: ptr); + Logger.info("rustpush_receive ack_complete id=$eventId retry=$retry duration_ms=${_durationMs(ackStopwatch)}"); } Future recievedMsgPointer(String pointer, String retry) async { + final eventId = _diagnosticHash(pointer); + final retryCount = int.tryParse(retry) ?? 3; + final receiveStopwatch = Stopwatch()..start(); var message = await api.ptrToDart(ptr: pointer); if (message == null) { - Logger.info("bad pointer $pointer $retry"); + Logger.info("rustpush_receive pointer_missing id=$eventId retry=$retryCount"); return; } - Logger.info("waitingForInit $pointer $retry"); + final initStopwatch = Stopwatch()..start(); + Logger.info("rustpush_receive aps_init_wait_start id=$eventId retry=$retryCount"); await initFuture; - var isFinal = (int.tryParse(retry) ?? 3) >= 3; + Logger.info("rustpush_receive aps_init_wait_complete id=$eventId retry=$retryCount duration_ms=${_durationMs(initStopwatch)} total_ms=${_durationMs(receiveStopwatch)}"); + var isFinal = retryCount >= 3; try { - Logger.info("Handling $pointer $retry"); + final handlingStopwatch = Stopwatch()..start(); + Logger.info("rustpush_receive handle_start id=$eventId retry=$retryCount"); await handleMsg(message, isFinal); - Logger.info("Marking as handled $pointer"); - await markAsHandledAfter(pointer); + Logger.info("rustpush_receive handle_complete id=$eventId retry=$retryCount duration_ms=${_durationMs(handlingStopwatch)} total_ms=${_durationMs(receiveStopwatch)}"); + await markAsHandledAfter(pointer, eventId: eventId, retry: retryCount); } catch (e, s) { Logger.error("Handle failed", error: e, trace: s); if (isFinal) { - Logger.info("Failed; Marking as handled anyways $pointer"); - await markAsHandledAfter(pointer); + Logger.info("rustpush_receive final_attempt_ack id=$eventId retry=$retryCount"); + await markAsHandledAfter(pointer, eventId: eventId, retry: retryCount); } rethrow; } From 36e2fd24400b97c6bf15774c3c1e441c3517429a Mon Sep 17 00:00:00 2001 From: Xare123 <57245242+Xare123@users.noreply.github.com> Date: Fri, 24 Jul 2026 12:46:20 -0700 Subject: [PATCH 2/9] Use fresh message service on transcript refresh --- .../conversation_view/pages/messages_view.dart | 15 +++++++++------ lib/main.dart | 3 ++- 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/lib/app/layouts/conversation_view/pages/messages_view.dart b/lib/app/layouts/conversation_view/pages/messages_view.dart index 06ad0a6c93..b4eb1e6c98 100644 --- a/lib/app/layouts/conversation_view/pages/messages_view.dart +++ b/lib/app/layouts/conversation_view/pages/messages_view.dart @@ -212,6 +212,10 @@ class MessagesViewState extends OptimizedState { Future _refreshMessageBloc() async { if (_refreshing) return; + if (widget.customService != null) { + Logger.info("message_refresh skipped_custom_service"); + return; + } _refreshing = true; try { final staleMessages = List.from(_messages); @@ -232,12 +236,11 @@ class MessagesViewState extends OptimizedState { fetching = false; _messages = []; - // Re-acquire the service after its GetX reload so the view does not - // continue using stale callbacks while the transcript is repopulated. - messageService.reload(); - if (widget.customService == null) { - messageService = ms(chat.guid); - } + // Get.reload rebuilds the original Get.put instance. Close it instead + // so its subscriptions and in-memory message structure are flushed + // before registering a genuinely new service for this transcript. + messageService.close(force: true); + messageService = ms(chat.guid); messageService.init(chat, handleNewMessage, handleUpdatedMessage, handleDeletedMessage, jumpToMessage); await messageService.loadChunk(0, controller); if (!mounted) return; diff --git a/lib/main.dart b/lib/main.dart index f86548e095..cebeb8db64 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -475,7 +475,8 @@ class _HomeState extends OptimizedState with WidgetsBindingObserver, TrayL } ErrorWidget.builder = (FlutterErrorDetails error) { - _logRenderError(error); + // FlutterError.onError above records the incident. Logging here would + // produce a second incident ID for the same rendering failure. return CustomErrorWidget( "An unexpected error occurred when rendering.", ); From bff1579014e4836cc73737ae8c256a9cb80a62cf Mon Sep 17 00:00:00 2001 From: Xare123 <57245242+Xare123@users.noreply.github.com> Date: Fri, 24 Jul 2026 12:47:33 -0700 Subject: [PATCH 3/9] Fix nullable reaction preview fallback --- .../widgets/tile/conversation_tile.dart | 7 +++---- lib/helpers/types/helpers/message_helper.dart | 12 ++++++++++-- test/helpers/message_helper_test.dart | 15 +++++++++++++++ 3 files changed, 28 insertions(+), 6 deletions(-) create mode 100644 test/helpers/message_helper_test.dart diff --git a/lib/app/layouts/conversation_list/widgets/tile/conversation_tile.dart b/lib/app/layouts/conversation_list/widgets/tile/conversation_tile.dart index fcdeac6e1a..8718a4dc28 100644 --- a/lib/app/layouts/conversation_list/widgets/tile/conversation_tile.dart +++ b/lib/app/layouts/conversation_list/widgets/tile/conversation_tile.dart @@ -448,13 +448,12 @@ class _ChatSubtitleState extends CustomState attachments) { Map counts = {}; diff --git a/test/helpers/message_helper_test.dart b/test/helpers/message_helper_test.dart new file mode 100644 index 0000000000..76f318f2c4 --- /dev/null +++ b/test/helpers/message_helper_test.dart @@ -0,0 +1,15 @@ +import 'package:bluebubbles/helpers/types/helpers/message_helper.dart'; +import 'package:flutter_test/flutter_test.dart'; + +void main() { + group('MessageHelper.getReactionFallbackText', () { + test('uses a human-safe reaction label for missing text', () { + expect(MessageHelper.getReactionFallbackText('Someone', null), 'Someone reacted to a message'); + expect(MessageHelper.getReactionFallbackText('Someone', ' '), 'Someone reacted to a message'); + }); + + test('preserves a populated fallback reaction text', () { + expect(MessageHelper.getReactionFallbackText('Someone', 'liked a message'), 'Someone liked a message'); + }); + }); +} From f9780fe33b9c12ffbf65d1ec2c96886ced3f87a4 Mon Sep 17 00:00:00 2001 From: Xare123 <57245242+Xare123@users.noreply.github.com> Date: Fri, 24 Jul 2026 16:34:29 -0700 Subject: [PATCH 4/9] Harden background routing and reconnect lifecycle --- .../foreground/SocketIOForegroundService.kt | 34 ++++++-- .../messaging/services/rustpush/APNService.kt | 2 - lib/services/backend/action_handler.dart | 2 +- .../method_channel_service.dart | 16 +--- .../firebase/firebase_database_service.dart | 9 +- lib/services/network/socket_service.dart | 87 ++++++++++++++----- 6 files changed, 101 insertions(+), 49 deletions(-) diff --git a/android/app/src/main/kotlin/com/bluebubbles/messaging/services/foreground/SocketIOForegroundService.kt b/android/app/src/main/kotlin/com/bluebubbles/messaging/services/foreground/SocketIOForegroundService.kt index c5a56dfe77..54b8852986 100644 --- a/android/app/src/main/kotlin/com/bluebubbles/messaging/services/foreground/SocketIOForegroundService.kt +++ b/android/app/src/main/kotlin/com/bluebubbles/messaging/services/foreground/SocketIOForegroundService.kt @@ -9,7 +9,9 @@ import android.content.Context import android.content.Intent import android.util.Log import android.os.Build +import android.os.Handler import android.os.IBinder +import android.os.Looper import androidx.core.app.NotificationCompat import androidx.core.app.ServiceCompat import com.bluebubbles.messaging.Constants @@ -47,6 +49,9 @@ class SocketIOForegroundService : Service() { private var isBeingDestroyed: Boolean = false private var hasStarted: Boolean = false + private val reconnectHandler = Handler(Looper.getMainLooper()) + private var reconnectRunnable: Runnable? = null + private var reconnectAttempt: Int = 0 private val eventBlacklist: Array = arrayOf( "typing-indicator", @@ -102,6 +107,9 @@ class SocketIOForegroundService : Service() { Log.d(Constants.logTag, "Foreground Service is connecting to: $serverUrl") val opts = IO.Options() + // Reconnects are scheduled by this service so the Socket.IO manager + // cannot race a second retry loop with our URL/service lifecycle. + opts.reconnection = false try { // Read the custom headers JSON string from preferences and parse it into a map @@ -128,6 +136,9 @@ class SocketIOForegroundService : Service() { mSocket!!.on(Socket.EVENT_CONNECT) { Log.d(Constants.logTag, "Socket.io connected to your server!") + reconnectAttempt = 0 + reconnectRunnable?.let { reconnectHandler.removeCallbacks(it) } + reconnectRunnable = null updateNotification(CONNECTED) } @@ -135,6 +146,7 @@ class SocketIOForegroundService : Service() { val error = args[0] as Exception Log.d(Constants.logTag, "Socket.io failed to connect to $serverUrl! Error: ${error.message}") updateNotification(CONNECT_FAILED + error.message) + tryReconnect() } // with reason, details args @@ -148,6 +160,7 @@ class SocketIOForegroundService : Service() { val details = args.getOrNull(1) Log.d(Constants.logTag, "Socket.io disconnected from server! Reason: $reason, Details: $details") updateNotification(DISCONNECTED + reason) + tryReconnect() } mSocket!!.on("reconnecting") { @@ -165,9 +178,7 @@ class SocketIOForegroundService : Service() { val event = args[0] as String val message = args[1] as JSONObject - Log.d(Constants.logTag, "Received event of type $event from Socket.io...") if (!eventBlacklist.contains(event)) { - Log.d(Constants.logTag, "Received event of type $event from Socket.io...") DartWorkManager.createWorker(applicationContext, "socket-event", hashMapOf("event" to event, "data" to message.toString())) {} } else { Log.d(Constants.logTag, "Ignored event of type $event from Socket.io...") @@ -190,11 +201,18 @@ class SocketIOForegroundService : Service() { private fun tryReconnect() { if (mSocket != null && !mSocket!!.connected()) { - Log.e(Constants.logTag, "Waiting 30 seconds before reconnecting...") - - // Sleep for 30 seconds before attempting to reconnect - Thread.sleep(30000) - mSocket!!.connect() + if (reconnectRunnable != null) return + val delaySeconds = 30L * (1L shl reconnectAttempt.coerceAtMost(3)) + reconnectAttempt = (reconnectAttempt + 1).coerceAtMost(3) + Log.e(Constants.logTag, "Scheduling reconnect in ${delaySeconds}s...") + val runnable = Runnable { + reconnectRunnable = null + if (!isBeingDestroyed && mSocket != null && !mSocket!!.connected()) { + mSocket!!.connect() + } + } + reconnectRunnable = runnable + reconnectHandler.postDelayed(runnable, delaySeconds * 1000L) } } @@ -261,6 +279,8 @@ class SocketIOForegroundService : Service() { override fun onDestroy() { isBeingDestroyed = true hasStarted = false + reconnectRunnable?.let { reconnectHandler.removeCallbacks(it) } + reconnectRunnable = null Log.d(Constants.logTag, "BlueBubbles Service is being destroyed!") super.onDestroy() diff --git a/android/app/src/main/kotlin/com/bluebubbles/messaging/services/rustpush/APNService.kt b/android/app/src/main/kotlin/com/bluebubbles/messaging/services/rustpush/APNService.kt index cc72a4e65e..8f531715df 100644 --- a/android/app/src/main/kotlin/com/bluebubbles/messaging/services/rustpush/APNService.kt +++ b/android/app/src/main/kotlin/com/bluebubbles/messaging/services/rustpush/APNService.kt @@ -96,12 +96,10 @@ class APNService : Service(), MsgReceiver { override fun receievedMsg(ptr: ULong, retry: ULong) { Handler(Looper.getMainLooper()).post { if (MainActivity.engine != null) { - Log.i("ugh running", "here $ptr $retry") // app is alive, deliver directly there MethodCallHandler.invokeMethod("APNMsg", mapOf("pointer" to ptr.toString(), "retry" to retry.toString())) return@post } - Log.i("ugh running", "backend $ptr $retry") CoroutineScope(Dispatchers.Main).launch { DartWorker.callMethod(this@APNService, "APNMsg", mapOf("pointer" to ptr.toString(), "retry" to retry.toString())) } diff --git a/lib/services/backend/action_handler.dart b/lib/services/backend/action_handler.dart index d0e8f9faee..514fa78540 100644 --- a/lib/services/backend/action_handler.dart +++ b/lib/services/backend/action_handler.dart @@ -498,7 +498,7 @@ class ActionHandler extends GetxService { } // should have been handled by the sanity check if (tempGuid != null) return; - Logger.info("New message: [${m.text}] - for chat [${c.guid}]", tag: "ActionHandler"); + Logger.debug("Received new message (attachments=${m.hasAttachments})", tag: "ActionHandler"); // Gets the chat from the db or server (if new) c = m.isParticipantEvent ? await handleNewOrUpdatedChat(c) : kIsWeb ? c : (Chat.findOne(guid: c.guid) ?? await handleNewOrUpdatedChat(c)); // Get the message handle diff --git a/lib/services/backend/java_dart_interop/method_channel_service.dart b/lib/services/backend/java_dart_interop/method_channel_service.dart index 6e48850a4d..ed6cd5c24a 100644 --- a/lib/services/backend/java_dart_interop/method_channel_service.dart +++ b/lib/services/backend/java_dart_interop/method_channel_service.dart @@ -141,13 +141,12 @@ class MethodChannelService extends GetxService { await Database.waitForInit(); Logger.info("Received new message from MethodChannel"); - // The socket will handle this event if the app is alive and unifiedpush is not enabled + // When the app is backgrounded, FCM is the safe fallback if the optional + // foreground socket is disconnected or still reconnecting. The message + // handler deduplicates by GUID after the first delivery is persisted. if (ls.isAlive && socket.socket.connected && ss.settings.endpointUnifiedPush.value == "") { Logger.debug("App is alive, ignoring new message..."); return Future.value(true); - } else if (!ls.isAlive && ss.settings.keepAppAlive.value) { - Logger.debug("Ignoring FCM message while app is not alive, but keepAppAlive is enabled"); - return Future.value(true); } try { @@ -174,9 +173,6 @@ class MethodChannelService extends GetxService { if (ls.isAlive && socket.socket.connected) { Logger.debug("App is alive, ignoring updated message..."); return Future.value(true); - } else if (!ls.isAlive && ss.settings.keepAppAlive.value) { - Logger.debug("Ignoring FCM message while app is not alive, but keepAppAlive is enabled"); - return Future.value(true); } try { @@ -221,9 +217,6 @@ class MethodChannelService extends GetxService { if (ls.isAlive && socket.socket.connected) { Logger.debug("App is alive, ignoring updated message..."); return Future.value(true); - } else if (!ls.isAlive && ss.settings.keepAppAlive.value) { - Logger.debug("Ignoring FCM message while app is not alive, but keepAppAlive is enabled"); - return Future.value(true); } try { @@ -246,9 +239,6 @@ class MethodChannelService extends GetxService { if (ls.isAlive && socket.socket.connected) { Logger.debug("App is alive, ignoring updated message..."); return Future.value(true); - } else if (!ls.isAlive && ss.settings.keepAppAlive.value) { - Logger.debug("Ignoring FCM message while app is not alive, but keepAppAlive is enabled"); - return Future.value(true); } try { diff --git a/lib/services/network/firebase/firebase_database_service.dart b/lib/services/network/firebase/firebase_database_service.dart index 35c7798f59..b8c9d3ccf2 100644 --- a/lib/services/network/firebase/firebase_database_service.dart +++ b/lib/services/network/firebase/firebase_database_service.dart @@ -55,7 +55,7 @@ class FirebaseDatabaseService extends GetxService { } /// Fetch the new server URL from the Firebase Database - Future fetchNewUrl() async { + Future fetchNewUrl({bool restartSocket = true, bool tryRestartForegroundService = true}) async { // Make sure setup is complete and we have valid data if (!ss.settings.finishedSetup.value) return null; if (ss.fcmData.isNull) { @@ -104,7 +104,12 @@ class FirebaseDatabaseService extends GetxService { url = sanitizeServerAddress(address: await mcs.invokeMethod("get-server-url")); } - await saveNewServerUrl(url ?? ss.settings.serverAddress.value, force: true); + await saveNewServerUrl( + url ?? ss.settings.serverAddress.value, + force: true, + restartSocket: restartSocket, + tryRestartForegroundService: tryRestartForegroundService, + ); return url; } catch (e, s) { Logger.error("Failed to fetch URL!", error: e, trace: s); diff --git a/lib/services/network/socket_service.dart b/lib/services/network/socket_service.dart index 62f56a1903..f0676ca0dd 100644 --- a/lib/services/network/socket_service.dart +++ b/lib/services/network/socket_service.dart @@ -26,6 +26,9 @@ class SocketService extends GetxService { SocketState _lastState = SocketState.disconnected; RxString lastError = "".obs; Timer? _reconnectTimer; + int _connectionGeneration = 0; + int _reconnectEpoch = 0; + int _reconnectAttempt = 0; late Socket socket; String get serverAddress => http.origin; @@ -55,29 +58,33 @@ class SocketService extends GetxService { } void startSocket() { + _cancelReconnect(); + final generation = ++_connectionGeneration; OptionBuilder options = OptionBuilder() .setQuery({"guid": password}) .setTransports(['websocket', 'polling']) .setExtraHeaders(http.headers) // Disable so that we can create the listeners first .disableAutoConnect() - .enableReconnection(); + // Reconnection is owned here so that URL refresh and socket creation + // cannot race the Socket.IO manager's own retry loop. + .disableReconnection(); socket = io(serverAddress, options.build()); // placed here so that [socket] is still initialized if (isNullOrEmpty(serverAddress)) return; - socket.onConnect((data) => handleStatusUpdate(SocketState.connected, data)); - socket.onReconnect((data) => handleStatusUpdate(SocketState.connected, data)); + socket.onConnect((data) => handleStatusUpdate(SocketState.connected, data, generation: generation)); + socket.onReconnect((data) => handleStatusUpdate(SocketState.connected, data, generation: generation)); - socket.onReconnectAttempt((data) => handleStatusUpdate(SocketState.connecting, data)); - socket.onReconnecting((data) => handleStatusUpdate(SocketState.connecting, data)); - socket.onConnecting((data) => handleStatusUpdate(SocketState.connecting, data)); + socket.onReconnectAttempt((data) => handleStatusUpdate(SocketState.connecting, data, generation: generation)); + socket.onReconnecting((data) => handleStatusUpdate(SocketState.connecting, data, generation: generation)); + socket.onConnecting((data) => handleStatusUpdate(SocketState.connecting, data, generation: generation)); - socket.onDisconnect((data) => handleStatusUpdate(SocketState.disconnected, data)); + socket.onDisconnect((data) => handleStatusUpdate(SocketState.disconnected, data, generation: generation)); - socket.onConnectError((data) => handleStatusUpdate(SocketState.error, data)); - socket.onConnectTimeout((data) => handleStatusUpdate(SocketState.error, data)); - socket.onError((data) => handleStatusUpdate(SocketState.error, data)); + socket.onConnectError((data) => handleStatusUpdate(SocketState.error, data, generation: generation)); + socket.onConnectTimeout((data) => handleStatusUpdate(SocketState.error, data, generation: generation)); + socket.onError((data) => handleStatusUpdate(SocketState.error, data, generation: generation)); // custom events // only listen to these events from socket on web/desktop (FCM handles on Android) @@ -100,6 +107,8 @@ class SocketService extends GetxService { } void disconnect() { + _cancelReconnect(); + _connectionGeneration++; if (isNullOrEmpty(serverAddress)) return; socket.disconnect(); state.value = SocketState.disconnected; @@ -107,11 +116,14 @@ class SocketService extends GetxService { void reconnect() { if (state.value == SocketState.connected || isNullOrEmpty(serverAddress)) return; + _cancelReconnect(); state.value = SocketState.connecting; socket.connect(); } void closeSocket() { + _cancelReconnect(); + _connectionGeneration++; if (isNullOrEmpty(serverAddress)) return; socket.dispose(); state.value = SocketState.disconnected; @@ -144,7 +156,8 @@ class SocketService extends GetxService { return completer.future; } - void handleStatusUpdate(SocketState status, dynamic data) { + void handleStatusUpdate(SocketState status, dynamic data, {int? generation}) { + if (generation != null && generation != _connectionGeneration) return; if (_lastState == status) return; _lastState = status; @@ -153,12 +166,15 @@ class SocketService extends GetxService { state.value = SocketState.connected; _reconnectTimer?.cancel(); _reconnectTimer = null; + _reconnectEpoch++; + _reconnectAttempt = 0; NetworkTasks.onConnect(); notif.clearSocketError(); return; case SocketState.disconnected: Logger.info("Disconnected from socket..."); state.value = SocketState.disconnected; + _scheduleReconnect(); return; case SocketState.connecting: Logger.info("Connecting to socket..."); @@ -172,25 +188,48 @@ class SocketService extends GetxService { } state.value = SocketState.error; - // After 5 seconds of an error, we should retry the connection - _reconnectTimer = Timer(const Duration(seconds: 5), () async { - if (state.value == SocketState.connected) return; - - await fdb.fetchNewUrl(); - restartSocket(); - - if (state.value == SocketState.connected) return; - - if (!ss.settings.keepAppAlive.value) { - notif.createSocketError(); - } - }); + _scheduleReconnect(); return; default: return; } } + void _cancelReconnect() { + _reconnectTimer?.cancel(); + _reconnectTimer = null; + _reconnectEpoch++; + } + + void _scheduleReconnect() { + if (_reconnectTimer != null || state.value == SocketState.connected || isNullOrEmpty(serverAddress)) return; + + final epoch = _reconnectEpoch; + final generation = _connectionGeneration; + final attempt = _reconnectAttempt > 3 ? 3 : _reconnectAttempt; + final seconds = 5 * (1 << attempt); + if (_reconnectAttempt < 3) _reconnectAttempt++; + _reconnectTimer = Timer(Duration(seconds: seconds), () async { + _reconnectTimer = null; + if (epoch != _reconnectEpoch || generation != _connectionGeneration || state.value == SocketState.connected) return; + + try { + await fdb.fetchNewUrl(restartSocket: false, tryRestartForegroundService: false); + } catch (e, s) { + Logger.warn("Failed to refresh socket URL before reconnect", error: e, trace: s); + _scheduleReconnect(); + return; + } + + if (epoch != _reconnectEpoch || generation != _connectionGeneration) return; + restartSocket(); + + if (state.value != SocketState.connected && !ss.settings.keepAppAlive.value) { + notif.createSocketError(); + } + }); + } + void handleSocketException(SocketException e) { String msg = e.message; if (msg.contains("Failed host lookup")) { From 2210a7273254bd2eaec2efb7953bf24ec57de8ce Mon Sep 17 00:00:00 2001 From: Xare123 <57245242+Xare123@users.noreply.github.com> Date: Fri, 24 Jul 2026 16:36:15 -0700 Subject: [PATCH 5/9] Back off terminal APS poll panics --- rust/src/native.rs | 37 +++++++++++++++++++++++++++++++++++-- 1 file changed, 35 insertions(+), 2 deletions(-) diff --git a/rust/src/native.rs b/rust/src/native.rs index 5ef5584215..f4b45322b0 100644 --- a/rust/src/native.rs +++ b/rust/src/native.rs @@ -154,15 +154,41 @@ pub fn plist_to_string(value: &T) -> Result)>> = LazyLock::new(|| Mutex::new((0, HashMap::new()))); +fn is_terminal_poll_panic(message: &str) -> bool { + let message = message.to_ascii_lowercase(); + message.contains("wrong phase") + || (message.contains("watcher") && message.contains("closed")) + || message.contains("channel closed") +} + +#[cfg(test)] +mod tests { + use super::is_terminal_poll_panic; + + #[test] + fn terminal_watcher_panics_stop_the_receive_loop() { + assert!(is_terminal_poll_panic("Wrong phase!")); + assert!(is_terminal_poll_panic("APS watcher is closed")); + assert!(is_terminal_poll_panic("channel closed")); + } + + #[test] + fn unrelated_panics_remain_retryable() { + assert!(!is_terminal_poll_panic("temporary network failure")); + } +} + #[uniffi::export] impl NativePushState { pub fn start_loop(self: Arc, handler: Arc) { RUNTIME.spawn(async move { let mut watcher = self.watcher.lock().await; + let mut panic_backoff_ms = 250u64; loop { match std::panic::AssertUnwindSafe(recv_wait(&mut watcher, &self.state)).catch_unwind().await { Ok(yes) => { + panic_backoff_ms = 250; match yes { PollResult::Cont(Some(msg)) => { if let PushMessage::TwoFaAuthEvent(event) = &msg { @@ -211,7 +237,14 @@ impl NativePushState { None => None, }, }; - error!("Failed {:?}", panic); + if panic.map(is_terminal_poll_panic).unwrap_or(false) { + warn!("Stopping APS receive loop after terminal watcher panic: {:?}", panic); + break; + } + + error!("Failed {:?}; backing off {}ms", panic, panic_backoff_ms); + tokio::time::sleep(Duration::from_millis(panic_backoff_ms)).await; + panic_backoff_ms = (panic_backoff_ms.saturating_mul(2)).min(5_000); } } } @@ -368,4 +401,4 @@ impl NativePushState { } }); } -} \ No newline at end of file +} From 8d6983e636609b85c90811b45e7b028e135c6f85 Mon Sep 17 00:00:00 2001 From: Xare123 <57245242+Xare123@users.noreply.github.com> Date: Fri, 24 Jul 2026 16:37:40 -0700 Subject: [PATCH 6/9] Register socket callbacks before connecting --- .../services/foreground/SocketIOForegroundService.kt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/android/app/src/main/kotlin/com/bluebubbles/messaging/services/foreground/SocketIOForegroundService.kt b/android/app/src/main/kotlin/com/bluebubbles/messaging/services/foreground/SocketIOForegroundService.kt index 54b8852986..e7d0192715 100644 --- a/android/app/src/main/kotlin/com/bluebubbles/messaging/services/foreground/SocketIOForegroundService.kt +++ b/android/app/src/main/kotlin/com/bluebubbles/messaging/services/foreground/SocketIOForegroundService.kt @@ -132,7 +132,6 @@ class SocketIOForegroundService : Service() { val encodedPw = URLEncoder.encode(storedPassword, "UTF-8") opts.query = "password=$encodedPw" mSocket = IO.socket(serverUrl, opts) - mSocket!!.connect() mSocket!!.on(Socket.EVENT_CONNECT) { Log.d(Constants.logTag, "Socket.io connected to your server!") @@ -185,6 +184,10 @@ class SocketIOForegroundService : Service() { } } } + + // Register every callback before opening the transport so an + // immediate connect or event cannot race listener setup. + mSocket!!.connect() } catch (e: Exception) { if (isBeingDestroyed) { return From 007ac1e0f6c9dd806e0f240df37a6ba3d38fda12 Mon Sep 17 00:00:00 2001 From: Xare123 <57245242+Xare123@users.noreply.github.com> Date: Fri, 24 Jul 2026 16:39:47 -0700 Subject: [PATCH 7/9] fix: preserve incoming message delivery integrity --- lib/database/io/message.dart | 34 +++--- lib/services/backend/action_handler.dart | 48 ++++++++- lib/services/backend/queue/queue_impl.dart | 111 +++++++++++--------- lib/services/rustpush/rustpush_service.dart | 13 ++- test/services/queue_impl_test.dart | 57 ++++++++++ 5 files changed, 195 insertions(+), 68 deletions(-) create mode 100644 test/services/queue_impl_test.dart diff --git a/lib/database/io/message.dart b/lib/database/io/message.dart index 4440c2ef6a..7140998410 100644 --- a/lib/database/io/message.dart +++ b/lib/database/io/message.dart @@ -400,19 +400,27 @@ class Message { var attachments = fetchAttachments()!; bool useMMS = chat.participants.length > 1 || attachments.isNotEmpty; int status; - if (useMMS) { - status = await TelephonyPlus().sendMMS( - addresses: chat.participants.map((e) => e.address).filter((e) => e.isPhoneNumber).toList(), - message: text?.trim() == "" ? null : text, - threadId: chat.telephonyId, - attachments: await Future.wait(attachments.map((e) => e!.toTelephony()).toList()) - ); - } else { - status = await TelephonyPlus().sendSMS( - address: chat.participants.first.address, - threadId: chat.telephonyId, - message: text!, - ); + try { + if (useMMS) { + status = await TelephonyPlus().sendMMS( + addresses: chat.participants.map((e) => e.address).filter((e) => e.isPhoneNumber).toList(), + message: text?.trim() == "" ? null : text, + threadId: chat.telephonyId, + attachments: await Future.wait(attachments.map((e) => e!.toTelephony()).toList()) + ); + } else { + status = await TelephonyPlus().sendSMS( + address: chat.participants.first.address, + threadId: chat.telephonyId, + message: text!, + ); + } + } catch (_) { + // No native status means the forwarding attempt did not complete. Let + // the transport retry instead of permanently suppressing forwarding. + hasBeenForwarded = false; + save(chat: chat); + rethrow; } if (status != -1) { await (backend as RustPushBackend).confirmSmsSent(this, chat, false); diff --git a/lib/services/backend/action_handler.dart b/lib/services/backend/action_handler.dart index 514fa78540..15d3498a7f 100644 --- a/lib/services/backend/action_handler.dart +++ b/lib/services/backend/action_handler.dart @@ -32,8 +32,19 @@ class ActionHandler extends GetxService { final RxList> attachmentProgress = >[].obs; final List outOfOrderTempGuids = []; final List handledNewMessages = []; + final Map> _inFlightNewMessages = {}; CancelToken? latestCancelToken; + Future _notifyNewMessageBestEffort(Message message, Chat chat) async { + try { + await MessageHelper.handleNotification(message, chat, findExisting: false); + } catch (_, stack) { + // Notification rendering must never prevent a message from being kept. + // Do not include the message or sender in diagnostics. + Logger.warn("Incoming message notification failed after persistence", tag: "Notification", trace: stack); + } + } + /// Checks if a GUID has been handled. /// After each check, before returning, trim the list of GUIDs to the last 100. bool shouldNotifyForNewMessageGuid(String guid) { @@ -487,6 +498,35 @@ class ActionHandler extends GetxService { } Future handleNewMessage(Chat c, Message m, String? tempGuid, {bool checkExisting = true}) async { + final key = m.guid; + final existingFlight = key == null ? null : _inFlightNewMessages[key]; + if (existingFlight != null) { + try { + await existingFlight; + } catch (_) { + // A failed first attempt must not prevent a concurrent fallback from retrying. + } + if (checkExisting && Message.findOne(guid: tempGuid ?? m.guid) != null) { + return await handleUpdatedMessage(c, m, tempGuid, checkExisting: false); + } + } + + if (key == null) { + return await _handleNewMessage(c, m, tempGuid, checkExisting: checkExisting); + } + + final flight = _handleNewMessage(c, m, tempGuid, checkExisting: checkExisting); + _inFlightNewMessages[key] = flight; + try { + await flight; + } finally { + if (identical(_inFlightNewMessages[key], flight)) { + _inFlightNewMessages.remove(key); + } + } + } + + Future _handleNewMessage(Chat c, Message m, String? tempGuid, {bool checkExisting = true}) async { Logger.info("handling new ${m.id}"); // sanity check if (checkExisting) { @@ -510,11 +550,13 @@ class ActionHandler extends GetxService { Logger.info("Not notifying for already handled new message with GUID ${m.guid}...", tag: "ActionHandler"); } + await c.addMessage(m); + await m.forwardIfNessesary(c, markFailed: true); + // Persistence is complete before notification work begins. Notification + // failures are isolated so they cannot make the transport drop the message. if ((!ls.isAlive || ss.settings.endpointUnifiedPush.value != "") && shouldNotify) { - await MessageHelper.handleNotification(m, c); + unawaited(_notifyNewMessageBestEffort(m, c)); } - await m.forwardIfNessesary(c, markFailed: true); - await c.addMessage(m); } Future handleUpdatedMessage(Chat c, Message m, String? tempGuid, {bool checkExisting = true}) async { diff --git a/lib/services/backend/queue/queue_impl.dart b/lib/services/backend/queue/queue_impl.dart index 38e7868de1..d76f731467 100644 --- a/lib/services/backend/queue/queue_impl.dart +++ b/lib/services/backend/queue/queue_impl.dart @@ -1,5 +1,4 @@ import 'dart:async'; -import 'dart:isolate'; import 'package:bluebubbles/helpers/helpers.dart'; import 'package:bluebubbles/database/models.dart'; @@ -10,72 +9,88 @@ import 'package:get/get.dart'; abstract class Queue extends GetxService { RxBool isProcessing = false.obs; List items = []; + bool _runnerActive = false; Future queue(QueueItem item, {bool prep = true}) async { - if (prep) { - final returned = await prepItem(item); - // we may get a link split into 2 messages - if (item is OutgoingItem && returned is List) { - items.addAll(returned.map((e) => OutgoingItem( - type: item.type, - chat: item.chat, - message: e, - completer: item.completer, - selected: item.selected, - reaction: item.reaction, - ))); + try { + if (prep) { + final returned = await prepItem(item); + // we may get a link split into 2 messages + if (item is OutgoingItem && returned is List) { + items.addAll(returned.map((e) => OutgoingItem( + type: item.type, + chat: item.chat, + message: e, + completer: item.completer, + selected: item.selected, + reaction: item.reaction, + ))); + } else { + items.add(item); + } } else { items.add(item); } - } else { - items.add(item); + } catch (ex, stacktrace) { + if (item.completer != null && !item.completer!.isCompleted) { + item.completer!.completeError(ex, stacktrace); + } + rethrow; } - if (!isProcessing.value || (items.isEmpty && item is IncomingItem)) processNextItem(); + _startRunner(); } Future prepItem(QueueItem _); + void _startRunner() { + if (_runnerActive) return; + _runnerActive = true; + unawaited(processNextItem()); + } + Future processNextItem() async { - if (items.isEmpty) { + isProcessing.value = true; + try { + while (items.isNotEmpty) { + ls.closeTimer?.cancel(); + ls.closeTimer = null; + + final queued = items.removeAt(0); + try { + await handleQueueItem(queued); + if (queued.completer != null && !queued.completer!.isCompleted) { + queued.completer!.complete(); + } + } catch (ex, stacktrace) { + Logger.error("Failed to handle queued item!", error: ex, trace: stacktrace); + if (queued is OutgoingItem && ss.settings.cancelQueuedMessages.value) { + final toCancel = List.from(items.whereType().where((e) => e.chat.guid == queued.chat.guid)); + for (final i in toCancel) { + items.remove(i); + final m = i.message; + final tempGuid = m.guid; + m.guid = m.guid!.replaceAll("temp", "error-Canceled due to previous failure"); + m.error = MessageError.BAD_REQUEST.code; + Message.replaceMessage(tempGuid, m); + } + } + if (queued.completer != null && !queued.completer!.isCompleted) { + queued.completer!.completeError(ex, stacktrace); + } + } + } + } finally { isProcessing.value = false; + _runnerActive = false; + if (items.isNotEmpty) _startRunner(); if (ls.isDead && !inq.isProcessing.value && !outq.isProcessing.value) { Logger.info("Done! waiting a bit for any stragglers"); ls.closeTimer = Timer(const Duration(seconds: 5), () { mcs.invokeMethod("engine-done"); }); } - return; - } - - ls.closeTimer?.cancel(); - ls.closeTimer = null; - - isProcessing.value = true; - QueueItem queued = items.removeAt(0); - - try { - await handleQueueItem(queued).catchError((err, trace) async { - Logger.error("Failed to handle queued item!", error: err, trace: trace); - if (queued is OutgoingItem && ss.settings.cancelQueuedMessages.value) { - final toCancel = List.from(items.whereType().where((e) => e.chat.guid == queued.chat.guid)); - for (OutgoingItem i in toCancel) { - items.remove(i); - final m = i.message; - final tempGuid = m.guid; - m.guid = m.guid!.replaceAll("temp", "error-Canceled due to previous failure"); - m.error = MessageError.BAD_REQUEST.code; - Message.replaceMessage(tempGuid, m); - } - } - }); - queued.completer?.complete(); - } catch (ex, stacktrace) { - Logger.error("Failed to handle queued item!", error: ex, trace: stacktrace); - queued.completer?.completeError(ex); } - - await processNextItem(); } Future handleQueueItem(QueueItem _); -} \ No newline at end of file +} diff --git a/lib/services/rustpush/rustpush_service.dart b/lib/services/rustpush/rustpush_service.dart index aaa8cfcab0..3f8f87137d 100644 --- a/lib/services/rustpush/rustpush_service.dart +++ b/lib/services/rustpush/rustpush_service.dart @@ -3795,9 +3795,11 @@ class RustPushService extends GetxService { myMsg.target = otherIds.map((element) => api.MessageTarget.uuid(element)).toList(); // forward to other devices await (backend as RustPushBackend).sendMsg(myMsg); } - var msg = (await pushService.reflectMessageDyn(myMsg))!; - msg.temp = true; - msg.forwardIfNessesary(chat); + final msg = await pushService.reflectMessageDyn(myMsg); + if (msg != null) { + msg.temp = true; + await msg.forwardIfNessesary(chat); + } return; } } @@ -3817,12 +3819,15 @@ class RustPushService extends GetxService { Logger.info("rustpush_receive reflection_complete id=$receiveId duration_ms=${_durationMs(receiveStopwatch)} reflected=${reflected != null}"); if (reflected != null) { final queueStopwatch = Stopwatch()..start(); + final queueCompletion = Completer(); Logger.info("rustpush_receive incoming_queue_enqueue id=$receiveId pending_count=${inq.items.length}"); await inq.queue(IncomingItem( chat: chat, message: reflected, - type: QueueType.newMessage + type: QueueType.newMessage, + completer: queueCompletion, )); + await queueCompletion.future; Logger.info("rustpush_receive incoming_queue_complete id=$receiveId duration_ms=${_durationMs(queueStopwatch)} pending_count=${inq.items.length}"); } } diff --git a/test/services/queue_impl_test.dart b/test/services/queue_impl_test.dart new file mode 100644 index 0000000000..84a63d6d78 --- /dev/null +++ b/test/services/queue_impl_test.dart @@ -0,0 +1,57 @@ +import 'dart:async'; + +import 'package:bluebubbles/database/models.dart'; +import 'package:bluebubbles/services/backend/queue/queue_impl.dart'; +import 'package:flutter_test/flutter_test.dart'; + +class _TestItem extends QueueItem { + _TestItem({Completer? completer}) + : super(type: QueueType.newMessage, completer: completer); +} + +class _TestQueue extends Queue { + int active = 0; + int maxActive = 0; + bool fail = false; + + @override + Future prepItem(QueueItem item) async {} + + @override + Future handleQueueItem(QueueItem item) async { + active++; + maxActive = active > maxActive ? active : maxActive; + await Future.delayed(const Duration(milliseconds: 10)); + active--; + if (fail) throw StateError('expected test failure'); + } +} + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + test('runs queued items with one active runner', () async { + final queue = _TestQueue(); + final first = Completer(); + final second = Completer(); + + await Future.wait([ + queue.queue(_TestItem(completer: first)), + queue.queue(_TestItem(completer: second)), + ]); + await Future.wait([first.future, second.future]); + + expect(queue.maxActive, 1); + expect(queue.isProcessing.value, isFalse); + }); + + test('completes a failed item with an error', () async { + final queue = _TestQueue()..fail = true; + final completion = Completer(); + + await queue.queue(_TestItem(completer: completion)); + + await expectLater(completion.future, throwsStateError); + expect(queue.isProcessing.value, isFalse); + }); +} From cfa9d5fa934eb25b759dc589d3daa6b105664927 Mon Sep 17 00:00:00 2001 From: Xare123 <57245242+Xare123@users.noreply.github.com> Date: Fri, 24 Jul 2026 16:47:00 -0700 Subject: [PATCH 8/9] Fix avatar resource leaks and push acknowledgement ordering --- lib/helpers/ui/ui_helpers.dart | 55 +++++++++++++-------- lib/services/rustpush/rustpush_service.dart | 30 ++++------- 2 files changed, 44 insertions(+), 41 deletions(-) diff --git a/lib/helpers/ui/ui_helpers.dart b/lib/helpers/ui/ui_helpers.dart index 083ea931fa..94b57af924 100644 --- a/lib/helpers/ui/ui_helpers.dart +++ b/lib/helpers/ui/ui_helpers.dart @@ -395,12 +395,14 @@ Future avatarAsBytes({ await paintGroupAvatar( chat: chat, participants: participants, canvas: canvas, size: quality, usingParticipantsOverride: participantsOverride != null); - ui.Picture picture = pictureRecorder.endRecording(); - ui.Image image = await picture.toImage(quality.toInt(), quality.toInt()); - - Uint8List bytes = (await image.toByteData(format: ui.ImageByteFormat.png))!.buffer.asUint8List(); - - return bytes; + final picture = pictureRecorder.endRecording(); + final image = await picture.toImage(quality.toInt(), quality.toInt()); + try { + return (await image.toByteData(format: ui.ImageByteFormat.png))!.buffer.asUint8List(); + } finally { + image.dispose(); + picture.dispose(); + } } Future paintGroupAvatar({ @@ -412,14 +414,15 @@ Future paintGroupAvatar({ }) async { late final ThemeData theme; final bool systemDark = PlatformDispatcher.instance.platformBrightness == Brightness.dark; - if (!ls.isAlive) { + final context = Get.context; + if (!ls.isAlive || context == null) { if (systemDark) { theme = ThemeStruct.getDarkTheme().data; } else { theme = ThemeStruct.getLightTheme().data; } } else { - theme = Get.context!.theme; + theme = context.theme; } if (chat.customAvatarPath != null && !usingParticipantsOverride) { @@ -430,7 +433,12 @@ Future paintGroupAvatar({ Logger.warn("Failed to load/clip custom avatar!", error: e, trace: stack); } if (customAvatar != null) { - canvas.drawImage(await loadImage(customAvatar), const Offset(0, 0), Paint()); + final avatarImage = await loadImage(customAvatar); + try { + canvas.drawImage(avatarImage, const Offset(0, 0), Paint()); + } finally { + avatarImage.dispose(); + } return; } } @@ -519,7 +527,12 @@ Future paintAvatar( if (contact?.avatar != null) { Uint8List? contactAvatar = await clip(contact!.avatar ?? contact.avatar!, size: size.toInt(), circle: kIsDesktop || inGroup); if (contactAvatar != null) { - canvas.drawImage(await loadImage(contactAvatar), offset, Paint()); + final avatarImage = await loadImage(contactAvatar); + try { + canvas.drawImage(avatarImage, offset, Paint()); + } finally { + avatarImage.dispose(); + } return; } } @@ -586,7 +599,6 @@ Future paintAvatar( } Future clip(Uint8List data, {required int size, required bool circle}) async { - ui.Image image; Uint8List _data = data; // Resize the image if it's the wrong size @@ -597,26 +609,29 @@ Future clip(Uint8List data, {required int size, required bool circle _data = img.encodePng(_image); } - image = await loadImage(_data); + final sourceImage = await loadImage(_data); ui.PictureRecorder pictureRecorder = ui.PictureRecorder(); Canvas canvas = Canvas(pictureRecorder); Paint paint = Paint(); paint.isAntiAlias = true; - Rect bounds = Rect.fromLTWH(0, 0, image.width.toDouble(), image.height.toDouble()); + Rect bounds = Rect.fromLTWH(0, 0, sourceImage.width.toDouble(), sourceImage.height.toDouble()); Path path = circle ? (Path()..addOval(bounds)) : (Path()..addRect(bounds)); canvas.clipPath(path); - canvas.drawImage(image, const Offset(0, 0), paint); - - ui.Picture picture = pictureRecorder.endRecording(); - image = await picture.toImage(image.width, image.height); + canvas.drawImage(sourceImage, const Offset(0, 0), paint); - Uint8List? bytes = (await image.toByteData(format: ui.ImageByteFormat.png))?.buffer.asUint8List(); - - return bytes; + final picture = pictureRecorder.endRecording(); + final clippedImage = await picture.toImage(sourceImage.width, sourceImage.height); + try { + return (await clippedImage.toByteData(format: ui.ImageByteFormat.png))?.buffer.asUint8List(); + } finally { + clippedImage.dispose(); + picture.dispose(); + sourceImage.dispose(); + } } Future loadImage(Uint8List data) async { diff --git a/lib/services/rustpush/rustpush_service.dart b/lib/services/rustpush/rustpush_service.dart index 3f8f87137d..07ae170d4b 100644 --- a/lib/services/rustpush/rustpush_service.dart +++ b/lib/services/rustpush/rustpush_service.dart @@ -3144,13 +3144,8 @@ class RustPushService extends GetxService { Chat.softDelete(chat); } - Future handleMsg(api.PushMessage push, bool finalAttempt) async { - try { - await handleMsgInner(push).timeout(const Duration(minutes: 3)); - } catch (e, s) { - if (finalAttempt) markCertified(push); - rethrow; - } + Future handleMsg(api.PushMessage push) async { + await handleMsgInner(push).timeout(const Duration(minutes: 3)); // if we complete successfully, mark delivery "certified" markCertified(push); } @@ -4484,13 +4479,9 @@ class RustPushService extends GetxService { Future markAsHandledAfter(String ptr, {required String eventId, required int retry}) async { final ackStopwatch = Stopwatch()..start(); - if (inq.isProcessing.value) { - Logger.info("rustpush_receive ack_wait_start id=$eventId retry=$retry pending_count=${inq.items.length}"); - await for (final value in inq.isProcessing.stream) { - if (!value) break; - } - } - Logger.info("rustpush_receive save_and_queue_drained id=$eventId retry=$retry wait_ms=${_durationMs(ackStopwatch)} pending_count=${inq.items.length}"); + // handleMsg awaits the completion for this pointer's queue item. Do not + // wait for unrelated incoming work before acknowledging this message. + Logger.info("rustpush_receive durable_work_complete id=$eventId retry=$retry pending_count=${inq.items.length}"); Logger.info("rustpush_receive ack_commit id=$eventId retry=$retry"); await api.completeMsg(ptr: ptr); Logger.info("rustpush_receive ack_complete id=$eventId retry=$retry duration_ms=${_durationMs(ackStopwatch)}"); @@ -4509,19 +4500,16 @@ class RustPushService extends GetxService { Logger.info("rustpush_receive aps_init_wait_start id=$eventId retry=$retryCount"); await initFuture; Logger.info("rustpush_receive aps_init_wait_complete id=$eventId retry=$retryCount duration_ms=${_durationMs(initStopwatch)} total_ms=${_durationMs(receiveStopwatch)}"); - var isFinal = retryCount >= 3; try { final handlingStopwatch = Stopwatch()..start(); Logger.info("rustpush_receive handle_start id=$eventId retry=$retryCount"); - await handleMsg(message, isFinal); + await handleMsg(message); Logger.info("rustpush_receive handle_complete id=$eventId retry=$retryCount duration_ms=${_durationMs(handlingStopwatch)} total_ms=${_durationMs(receiveStopwatch)}"); await markAsHandledAfter(pointer, eventId: eventId, retry: retryCount); } catch (e, s) { Logger.error("Handle failed", error: e, trace: s); - if (isFinal) { - Logger.info("rustpush_receive final_attempt_ack id=$eventId retry=$retryCount"); - await markAsHandledAfter(pointer, eventId: eventId, retry: retryCount); - } + // Leave the pointer pending so the native bounded retry loop can try + // again. A failed handler must never be acknowledged as delivered. rethrow; } } @@ -4542,7 +4530,7 @@ class RustPushService extends GetxService { if (msg == null) { continue; } - await handleMsg(msg, true); + await handleMsg(msg); } catch (e, t) { // if there was an error somewhere, log it and move on. // don't stop our loop From 742296e3b5740f484117de5e7ce5570baee09ca9 Mon Sep 17 00:00:00 2001 From: Xare123 <57245242+Xare123@users.noreply.github.com> Date: Fri, 24 Jul 2026 16:45:55 -0700 Subject: [PATCH 9/9] fix: harden group participant and reaction handling --- lib/database/global/chat_messages.dart | 42 +++++++++++- lib/helpers/group_participant_helpers.dart | 28 ++++++++ lib/services/rustpush/rustpush_service.dart | 67 +++++++++++++------ lib/services/ui/chat/chat_manager.dart | 31 +++++---- lib/services/ui/message/messages_service.dart | 19 ++++-- test/helpers/chat_messages_test.dart | 37 ++++++++++ .../group_participant_helpers_test.dart | 47 +++++++++++++ 7 files changed, 234 insertions(+), 37 deletions(-) create mode 100644 lib/helpers/group_participant_helpers.dart create mode 100644 test/helpers/chat_messages_test.dart create mode 100644 test/helpers/group_participant_helpers_test.dart diff --git a/lib/database/global/chat_messages.dart b/lib/database/global/chat_messages.dart index 131468a6a0..887a0549cc 100644 --- a/lib/database/global/chat_messages.dart +++ b/lib/database/global/chat_messages.dart @@ -1,8 +1,11 @@ import 'package:bluebubbles/database/models.dart'; class ChatMessages { + static const int _maxPendingReactionsPerMessage = 32; + static const int _maxPendingReactionParents = 128; final Map _messages = {}; final Map _reactions = {}; + final Map> _pendingReactions = {}; final Map _attachments = {}; final Map> _threads = {}; final Map> _edits = {}; @@ -21,9 +24,37 @@ class ChatMessages { if (m.associatedMessageGuid != null) { // add reactions _reactions[m.guid!] = m; + final parent = getMessage(m.associatedMessageGuid!); + if (parent != null) { + _attachReaction(parent, m); + } else { + final parentGuid = m.associatedMessageGuid!; + var pending = _pendingReactions[parentGuid]; + if (pending == null) { + if (_pendingReactions.length >= _maxPendingReactionParents) { + _pendingReactions.remove(_pendingReactions.keys.first); + } + pending = {}; + _pendingReactions[parentGuid] = pending; + } + // A malformed or delayed stream must not grow this cache forever. + // The database sync remains the source of truth if an item is + // evicted before its parent arrives. + if (pending.length >= _maxPendingReactionsPerMessage && + !pending.containsKey(m.guid)) { + pending.remove(pending.keys.first); + } + pending[m.guid!] = m; + } } else { // add regular texts _messages[m.guid!] = m; + final pending = _pendingReactions.remove(m.guid); + if (pending != null) { + for (final reaction in pending.values) { + _attachReaction(m, reaction); + } + } } if (m.threadOriginatorGuid != null && !m.guid!.startsWith("temp") && m.associatedMessageGuid == null) { // add threaded messages @@ -38,9 +69,17 @@ class ChatMessages { } } + void _attachReaction(Message parent, Message reaction) { + if (!parent.associatedMessages.any((item) => item.guid == reaction.guid)) { + parent.associatedMessages.add(reaction); + } + parent.hasReactions = true; + } + void removeMessage(String guid) { _messages.remove(guid); _reactions.remove(guid); + _pendingReactions.remove(guid); final result = _threads.remove(guid); if (result == null) { for (Map element in _threads.values) { @@ -100,8 +139,9 @@ class ChatMessages { flush() { _messages.clear(); _reactions.clear(); + _pendingReactions.clear(); _attachments.clear(); _threads.clear(); _edits.clear(); } -} \ No newline at end of file +} diff --git a/lib/helpers/group_participant_helpers.dart b/lib/helpers/group_participant_helpers.dart new file mode 100644 index 0000000000..9f76275200 --- /dev/null +++ b/lib/helpers/group_participant_helpers.dart @@ -0,0 +1,28 @@ +import 'package:bluebubbles/database/models.dart'; + +/// Reconciles a locally cached participant list with the server's latest +/// ordered list. +/// +/// The old fetch path only handled length changes and could leave a group +/// stale when one participant was replaced by another. It also added only one +/// handle when several participants were added. Preserve existing Handle +/// objects (and their contact metadata) when the identity is unchanged, while +/// applying the server list exactly once and dropping duplicate identities. +List reconcileGroupParticipants( + List current, + List incoming, +) { + final existingByIdentity = {}; + for (final handle in current) { + existingByIdentity.putIfAbsent(handle.uniqueAddressAndService, () => handle); + } + + final seen = {}; + final reconciled = []; + for (final handle in incoming) { + final identity = handle.uniqueAddressAndService; + if (!seen.add(identity)) continue; + reconciled.add(existingByIdentity[identity] ?? handle); + } + return reconciled; +} diff --git a/lib/services/rustpush/rustpush_service.dart b/lib/services/rustpush/rustpush_service.dart index 07ae170d4b..599361b834 100644 --- a/lib/services/rustpush/rustpush_service.dart +++ b/lib/services/rustpush/rustpush_service.dart @@ -1353,6 +1353,11 @@ class RustPushService extends GetxService { } Future updateChatParticipants(Chat c, api.MessageInst myMsg, List oldParticipants, List newParticipants) async { + final sender = myMsg.sender; + if (sender == null || sender.isEmpty) { + Logger.warn("Ignoring participant update without a sender"); + return; + } var myHandles = await api.getHandles(state: pushService.state!.client); var newP = newParticipants.filter((p) => !oldParticipants.contains(p) && !myHandles.contains(p)); var delP = oldParticipants.filter((p) => !newParticipants.contains(p)); @@ -1371,8 +1376,8 @@ class RustPushService extends GetxService { var bb = RustPushBBUtils.rustHandleToBB(item); var msg = Message( guid: useId ? myMsg.id : uuid.v4(), - isFromMe: myHandles.contains(myMsg.sender), - handleId: RustPushBBUtils.rustHandleToBB(myMsg.sender!).originalROWID!, + isFromMe: myHandles.contains(sender), + handleId: RustPushBBUtils.rustHandleToBB(sender).originalROWID!, dateCreated: DateTime.fromMillisecondsSinceEpoch(myMsg.sentTimestamp), itemType: 1, groupActionType: 0, @@ -1388,11 +1393,11 @@ class RustPushService extends GetxService { for (var item in delP) { var bb = RustPushBBUtils.rustHandleToBB(item); - var personDidLeave = item == myMsg.sender; + var personDidLeave = item == sender; var msg = Message( guid: useId ? myMsg.id : uuid.v4(), - isFromMe: myHandles.contains(myMsg.sender), - handleId: RustPushBBUtils.rustHandleToBB(myMsg.sender!).originalROWID!, + isFromMe: myHandles.contains(sender), + handleId: RustPushBBUtils.rustHandleToBB(sender).originalROWID!, dateCreated: DateTime.fromMillisecondsSinceEpoch(myMsg.sentTimestamp), itemType: personDidLeave ? 3 : 1, groupActionType: personDidLeave ? 0 : 1, @@ -1685,15 +1690,16 @@ class RustPushService extends GetxService { return msg; } else if (myMsg.message is api.Message_RenameMessage) { var msg = myMsg.message as api.Message_RenameMessage; - if (myMsg.verificationFailed) return null; + final sender = myMsg.sender; + if (myMsg.verificationFailed || chat == null || sender == null || sender.isEmpty) return null; - chat!.ckSyncState = false; + chat.ckSyncState = false; chat.save(updateCkSyncState: true); return Message( guid: myMsg.id, - isFromMe: myHandles.contains(myMsg.sender), - handleId: RustPushBBUtils.rustHandleToBB(myMsg.sender!).originalROWID!, + isFromMe: myHandles.contains(sender), + handleId: RustPushBBUtils.rustHandleToBB(sender).originalROWID!, dateCreated: DateTime.fromMillisecondsSinceEpoch(myMsg.sentTimestamp), itemType: 2, groupActionType: 2, @@ -1701,15 +1707,18 @@ class RustPushService extends GetxService { ); } else if (myMsg.message is api.Message_ChangeParticipants) { var msg = myMsg.message as api.Message_ChangeParticipants; - if (myMsg.verificationFailed) return null; - await updateChatParticipants(chat!, myMsg, myMsg.conversation!.participants, msg.field0.newParticipants); + final conversation = myMsg.conversation; + if (myMsg.verificationFailed || chat == null || conversation == null || myMsg.sender == null) return null; + await updateChatParticipants(chat, myMsg, conversation.participants, msg.field0.newParticipants); chat.groupVersion = msg.field0.groupVersion; chat.ckSyncState = false; chat.save(updateGroupVersion: true, updateCkSyncState: true); return null; } else if (myMsg.message is api.Message_IconChange) { var innerMsg = myMsg.message as api.Message_IconChange; - if (!chat!.lockChatIcon && (chat.groupVersion ?? 0) < innerMsg.field0.groupVersion) { + final sender = myMsg.sender; + if (chat == null || sender == null || sender.isEmpty) return null; + if (!chat.lockChatIcon && (chat.groupVersion ?? 0) < innerMsg.field0.groupVersion) { var file = innerMsg.field0.file; chat.groupVersion = innerMsg.field0.groupVersion; chat.ckSyncState = false; @@ -1728,16 +1737,21 @@ class RustPushService extends GetxService { } return Message( guid: myMsg.id, - isFromMe: myHandles.contains(myMsg.sender), - handleId: RustPushBBUtils.rustHandleToBB(myMsg.sender!).originalROWID!, + isFromMe: myHandles.contains(sender), + handleId: RustPushBBUtils.rustHandleToBB(sender).originalROWID!, dateCreated: DateTime.fromMillisecondsSinceEpoch(myMsg.sentTimestamp), itemType: 3, groupActionType: 1, ); } else if (myMsg.message is api.Message_React) { var msg = myMsg.message as api.Message_React; + final sender = myMsg.sender; + if (sender == null || sender.isEmpty) { + Logger.warn("Ignoring reaction without a sender"); + return null; + } if (msg.field0.embeddedProfile != null) { - handleSharedProfile(msg.field0.embeddedProfile!, myMsg.sender!, chat?.participants ?? []); + handleSharedProfile(msg.field0.embeddedProfile!, sender, chat?.participants ?? []); } String? reaction; @@ -1789,7 +1803,11 @@ class RustPushService extends GetxService { final messages = query.find(); query.close(); - final original = messages.firstWhere((msg) => (msg.stagingGuid ?? msg.guid) != myMsg.id); + final original = messages.firstWhereOrNull((msg) => (msg.stagingGuid ?? msg.guid) != myMsg.id); + if (original == null) { + Logger.warn("Ignoring extension update without a base message"); + return null; + } original.fetchAssociatedMessages(); @@ -1801,7 +1819,12 @@ class RustPushService extends GetxService { } // allow updating image - attributedBodyData = (attributedBodyData.$3.isEmpty ? original.attributedBody[0] : attributedBodyData.$1, original.text!, attributedBodyData.$3.isEmpty ? original.dbAttachments : attributedBodyData.$3); + final originalBody = original.attributedBody.firstOrNull; + if (attributedBodyData.$3.isEmpty && originalBody == null) { + Logger.warn("Ignoring extension update without message content"); + return null; + } + attributedBodyData = (attributedBodyData.$3.isEmpty ? originalBody! : attributedBodyData.$1, original.text ?? "", attributedBodyData.$3.isEmpty ? original.dbAttachments : attributedBodyData.$3); var tag = es.getLatest(msg.field0.toUuid); // updates cached value; we are latest if (tag.firstOrNull != myMsg.id) { @@ -1823,8 +1846,8 @@ class RustPushService extends GetxService { } var message = Message( guid: myMsg.id, - isFromMe: myHandles.contains(myMsg.sender), - handleId: RustPushBBUtils.rustHandleToBB(myMsg.sender!).originalROWID!, + isFromMe: myHandles.contains(sender), + handleId: RustPushBBUtils.rustHandleToBB(sender).originalROWID!, dateCreated: DateTime.fromMillisecondsSinceEpoch(myMsg.sentTimestamp), associatedMessagePart: msg.field0.toPart, associatedMessageGuid: reaction == null ? null : msg.field0.toUuid, @@ -1848,7 +1871,11 @@ class RustPushService extends GetxService { return message; } else if (myMsg.message is api.Message_Unsend) { var msg = myMsg.message as api.Message_Unsend; - var msgObj = Message.findOne(guid: msg.field0.tuuid)!; + var msgObj = Message.findOne(guid: msg.field0.tuuid); + if (msgObj == null) { + Logger.warn("Ignoring unsend for a missing message"); + return null; + } msgObj.verificationFailed = myMsg.verificationFailed; msgObj.dateEdited = DateTime.now(); var summaryInfo = msgObj.messageSummaryInfo.firstOrNull; diff --git a/lib/services/ui/chat/chat_manager.dart b/lib/services/ui/chat/chat_manager.dart index 11caeefe2f..6079d66d03 100644 --- a/lib/services/ui/chat/chat_manager.dart +++ b/lib/services/ui/chat/chat_manager.dart @@ -2,6 +2,7 @@ import 'dart:async'; import 'package:bluebubbles/utils/logger/logger.dart'; import 'package:bluebubbles/database/models.dart'; +import 'package:bluebubbles/helpers/group_participant_helpers.dart'; import 'package:bluebubbles/services/services.dart'; import 'package:dio/dio.dart'; import 'package:bluebubbles/services/rustpush/rustpush_service.dart'; @@ -157,18 +158,24 @@ class ChatManager extends GetxService { if (chat == null) { updatedChat.save(); chat = Chat.findOne(guid: chatGuid)!; - } else if (chat.handles.length > updatedChat.participants.length) { - final newAddresses = updatedChat.participants.map((e) => e.address); - final handlesToUse = chat.participants.where((e) => newAddresses.contains(e.address)); - chat.handles.clear(); - chat.handles.addAll(handlesToUse); - chat.handles.applyToDb(); - } else if (chat.handles.length < updatedChat.participants.length) { - final existingAddresses = chat.participants.map((e) => e.address); - final newHandle = updatedChat.participants.firstWhere((e) => !existingAddresses.contains(e.address)); - final handle = Handle.findOne(addressAndService: Tuple2(newHandle.address, chat.isIMessage ? "iMessage" : "SMS")) ?? newHandle.save(); - chat.handles.add(handle); - chat.handles.applyToDb(); + } else { + // Reconcile by address and service, not just list length. A group can + // replace one participant without changing its size, and multiple + // additions must all be reflected in the local relation. + // An incomplete server response must not erase a known participant + // list. A valid empty group is not useful for this client, so retain + // the local list when the response contains no participants. + if (updatedChat.participants.isNotEmpty || chat.participants.isEmpty) { + final handles = reconcileGroupParticipants( + chat.participants, + updatedChat.participants, + ); + if (handles.isNotEmpty) Handle.bulkSave(handles); + chat.handles.clear(); + chat.handles.addAll(handles); + chat.handles.applyToDb(); + chat.getParticipants(); + } } if (!chat.lockChatName) { chat.displayName = updatedChat.displayName; diff --git a/lib/services/ui/message/messages_service.dart b/lib/services/ui/message/messages_service.dart index 83ec39d670..74857d9927 100644 --- a/lib/services/ui/message/messages_service.dart +++ b/lib/services/ui/message/messages_service.dart @@ -120,15 +120,26 @@ class MessagesService extends GetxController { if (message.amkSessionId != null) { message.fetchAssociatedMessages(); } - // add this as a reaction if needed, update thread originators and associated messages + // Add this as a reaction if needed, update thread originators and + // associated messages. ChatMessages retains a bounded pending set when a + // reaction arrives before its base message. if (message.associatedMessageGuid != null) { - struct.getMessage(message.associatedMessageGuid!)?.associatedMessages.add(message); - getActiveMwc(message.associatedMessageGuid!)?.updateAssociatedMessage(message); + final parent = struct.getMessage(message.associatedMessageGuid!); + if (parent != null) { + getActiveMwc(message.associatedMessageGuid!)?.updateAssociatedMessage(message); + } } if (message.threadOriginatorGuid != null) { getActiveMwc(message.threadOriginatorGuid!)?.updateThreadOriginator(message); } struct.addMessages([message]); + if (message.associatedMessageGuid == null) { + // ChatMessages attaches any reactions that arrived before this message; + // refresh the active bubble after the parent is present in the struct. + for (final reaction in message.associatedMessages) { + getActiveMwc(message.guid!)?.updateAssociatedMessage(reaction); + } + } if (message.associatedMessageGuid == null) { newFunc.call(message); } @@ -271,4 +282,4 @@ class MessagesService extends GetxController { return completer.future; } -} \ No newline at end of file +} diff --git a/test/helpers/chat_messages_test.dart b/test/helpers/chat_messages_test.dart new file mode 100644 index 0000000000..734fc161da --- /dev/null +++ b/test/helpers/chat_messages_test.dart @@ -0,0 +1,37 @@ +import 'package:bluebubbles/database/global/chat_messages.dart'; +import 'package:bluebubbles/database/models.dart'; +import 'package:flutter_test/flutter_test.dart'; + +void main() { + test('attaches a reaction that arrives before its base message', () { + final messages = ChatMessages(); + final reaction = Message( + guid: 'reaction-1', + associatedMessageGuid: 'base-1', + associatedMessageType: 'like', + ); + final base = Message(guid: 'base-1', text: 'hello'); + + messages.addMessages([reaction]); + messages.addMessages([base]); + + expect(messages.getMessage('base-1'), same(base)); + expect(base.hasReactions, isTrue); + expect(base.associatedMessages.map((item) => item.guid).toList(), ['reaction-1']); + }); + + test('does not duplicate a reaction when the event is replayed', () { + final messages = ChatMessages(); + final base = Message(guid: 'base-2', text: 'hello'); + final reaction = Message( + guid: 'reaction-2', + associatedMessageGuid: 'base-2', + associatedMessageType: 'like', + ); + + messages.addMessages([base, reaction, reaction]); + + expect(base.associatedMessages, hasLength(1)); + expect(base.associatedMessages.single, same(reaction)); + }); +} diff --git a/test/helpers/group_participant_helpers_test.dart b/test/helpers/group_participant_helpers_test.dart new file mode 100644 index 0000000000..1c2d04eb5f --- /dev/null +++ b/test/helpers/group_participant_helpers_test.dart @@ -0,0 +1,47 @@ +import 'package:bluebubbles/database/models.dart'; +import 'package:bluebubbles/helpers/group_participant_helpers.dart'; +import 'package:flutter_test/flutter_test.dart'; + +void main() { + group('reconcileGroupParticipants', () { + test('replaces a participant when group size is unchanged', () { + final existing = Handle(address: 'old@example.com'); + final retained = Handle(address: 'kept@example.com'); + final replacement = Handle(address: 'new@example.com'); + + final result = reconcileGroupParticipants( + [existing, retained], + [replacement, Handle(address: 'kept@example.com')], + ); + + expect(result, hasLength(2)); + expect(result[0], same(replacement)); + expect(result[1], same(retained)); + }); + + test('keeps every new participant and removes duplicate identities', () { + final result = reconcileGroupParticipants( + [], + [ + Handle(address: 'one@example.com'), + Handle(address: 'two@example.com'), + Handle(address: 'two@example.com'), + Handle(address: 'three@example.com'), + ], + ); + + expect(result.map((handle) => handle.address), + ['one@example.com', 'two@example.com', 'three@example.com']); + }); + + test('distinguishes the same address on different services', () { + final sms = Handle(address: '+15550000001', service: 'SMS'); + final imessage = Handle(address: '+15550000001', service: 'iMessage'); + + final result = reconcileGroupParticipants([sms], [imessage]); + + expect(result, hasLength(1)); + expect(result.single, same(imessage)); + }); + }); +}