Initial web listeners implementation for message updates

lib/app/layouts/conversation_list/widgets/tile/conversation_tile.dart

@@ -280,8 +280,9 @@ class ChatSubtitle extends CustomStateful<ConversationTileController> {
 class _ChatSubtitleState extends CustomState<ChatSubtitle, void, ConversationTileController> {
   String subtitle = "Unknown";
   String fakeText = faker.lorem.words(1).join(" ");
-  late final StreamSubscription<Query<Message>> sub;
+  late final StreamSubscription sub;
   String? cachedLatestMessageGuid = "";
+  DateTime? cachedDateCreated;
   bool isDelivered = false;
   bool isFromMe = false;
 
@@ -332,12 +333,36 @@ class _ChatSubtitleState extends CustomState<ChatSubtitle, void, ConversationTil
           cachedLatestMessageGuid = message?.guid;
         });
       });
+    } else {
+      sub = WebListeners.newMessage.listen((tuple) {
+        final message = tuple.item1;
+        if (tuple.item2?.guid == controller.chat.guid && (cachedDateCreated == null || message.dateCreated!.isAfter(cachedDateCreated!))) {
+          isFromMe = message.isFromMe ?? false;
+          isDelivered = controller.chat.isGroup || !isFromMe || message.dateDelivered != null || message.dateRead != null;
+          if (message.guid != cachedLatestMessageGuid) {
+            String newSubtitle = MessageHelper.getNotificationText(message);
+            if (newSubtitle != subtitle) {
+              setState(() {
+                subtitle = newSubtitle;
+                fakeText = faker.lorem.words(subtitle.split(" ").length).join(" ");
+              });
+            }
+          } else if (!controller.chat.isGroup
+              && message.isFromMe!
+              && (message.dateDelivered != null || message.dateRead != null)) {
+            // update delivered status
+            setState(() {});
+          }
+          cachedDateCreated = message.dateCreated;
+          cachedLatestMessageGuid = message.guid;
+        }
+      });
     }
   }
 
   @override
   void dispose() {
-    if (!kIsWeb) sub.cancel();
+    sub.cancel();
     super.dispose();
   }
 

lib/app/layouts/conversation_list/widgets/tile/cupertino_conversation_tile.dart

@@ -124,7 +124,7 @@ class CupertinoTrailing extends CustomStateful<ConversationTileController> {
 
 class _CupertinoTrailingState extends CustomState<CupertinoTrailing, void, ConversationTileController> {
   DateTime? dateCreated;
-  late final StreamSubscription<Query<Message>> sub;
+  late final StreamSubscription sub;
   String? cachedLatestMessageGuid = "";
   Message? cachedLatestMessage;
 
@@ -162,12 +162,22 @@ class _CupertinoTrailingState extends CustomState<CupertinoTrailing, void, Conve
           cachedLatestMessageGuid = message?.guid;
         });
       });
+    } else {
+      sub = WebListeners.newMessage.listen((tuple) {
+        if (tuple.item2?.guid == controller.chat.guid && (dateCreated == null || tuple.item1.dateCreated!.isAfter(dateCreated!))) {
+          cachedLatestMessage = tuple.item1;
+          setState(() {
+            dateCreated = tuple.item1.dateCreated;
+          });
+          cachedLatestMessageGuid = tuple.item1.guid;
+        }
+      });
     }
   }
 
   @override
   void dispose() {
-    if (!kIsWeb) sub.cancel();
+    sub.cancel();
     super.dispose();
   }
 

lib/app/layouts/conversation_list/widgets/tile/material_conversation_tile.dart

@@ -151,7 +151,7 @@ class _MaterialTrailingState extends CustomState<MaterialTrailing, void, Convers
   DateTime? dateCreated;
   bool unread = false;
   String muteType = "";
-  late final StreamSubscription<Query<Message>> sub;
+  late final StreamSubscription sub;
   late final StreamSubscription<Query<Chat>> sub2;
   String? cachedLatestMessageGuid = "";
   Message? cachedLatestMessage;
@@ -216,13 +216,23 @@ class _MaterialTrailingState extends CustomState<MaterialTrailing, void, Convers
           }
         });
       });
+    } else {
+      sub = WebListeners.newMessage.listen((tuple) {
+        if (tuple.item2?.guid == controller.chat.guid && (dateCreated == null || tuple.item1.dateCreated!.isAfter(dateCreated!))) {
+          cachedLatestMessage = tuple.item1;
+          setState(() {
+            dateCreated = tuple.item1.dateCreated;
+          });
+          cachedLatestMessageGuid = tuple.item1.guid;
+        }
+      });
     }
   }
 
   @override
   void dispose() {
+    sub.cancel();
     if (!kIsWeb) {
-      sub.cancel();
       sub2.cancel();
     }
     super.dispose();

lib/app/layouts/conversation_list/widgets/tile/pinned_tile_text_bubble.dart

@@ -32,8 +32,9 @@ class PinnedTileTextBubbleState extends CustomState<PinnedTileTextBubble, void,
   Message? lastMessage;
   String subtitle = "Unknown";
   String fakeText = faker.lorem.words(1).join(" ");
-  late final StreamSubscription<Query<Message>> sub;
+  late final StreamSubscription sub;
   String? cachedLatestMessageGuid = "";
+  DateTime? cachedDateCreated;
   late bool unread = chat.hasUnreadMessage ?? false;
   late final StreamSubscription<Query<Chat>> sub2;
 
@@ -98,13 +99,30 @@ class PinnedTileTextBubbleState extends CustomState<PinnedTileTextBubble, void,
           }
         });
       });
+    } else {
+      sub = WebListeners.newMessage.listen((tuple) {
+        final message = tuple.item1;
+        if (tuple.item2?.guid == controller.chat.guid && (cachedDateCreated == null || message.dateCreated!.isAfter(cachedDateCreated!))) {
+          if (message.guid != cachedLatestMessageGuid) {
+            String newSubtitle = MessageHelper.getNotificationText(message);
+            if (newSubtitle != subtitle) {
+              setState(() {
+                subtitle = newSubtitle;
+                fakeText = faker.lorem.words(subtitle.split(" ").length).join(" ");
+              });
+            }
+          }
+          cachedDateCreated = message.dateCreated;
+          cachedLatestMessageGuid = message.guid;
+        }
+      });
     }
   }
 
   @override
   void dispose() {
+    sub.cancel();
     if (!kIsWeb) {
-      sub.cancel();
       sub2.cancel();
     }
     super.dispose();

lib/app/layouts/conversation_list/widgets/tile/samsung_conversation_tile.dart

@@ -96,7 +96,7 @@ class _SamsungTrailingState extends CustomState<SamsungTrailing, void, Conversat
   DateTime? dateCreated;
   bool unread = false;
   String muteType = "";
-  late final StreamSubscription<Query<Message>> sub;
+  late final StreamSubscription sub;
   late final StreamSubscription<Query<Chat>> sub2;
   String? cachedLatestMessageGuid = "";
   Message? cachedLatestMessage;
@@ -160,6 +160,16 @@ class _SamsungTrailingState extends CustomState<SamsungTrailing, void, Conversat
           }
         });
       });
+    } else {
+      sub = WebListeners.newMessage.listen((tuple) {
+        if (tuple.item2?.guid == controller.chat.guid && (dateCreated == null || tuple.item1.dateCreated!.isAfter(dateCreated!))) {
+          cachedLatestMessage = tuple.item1;
+          setState(() {
+            dateCreated = tuple.item1.dateCreated;
+          });
+          cachedLatestMessageGuid = tuple.item1.guid;
+        }
+      });
     }
   }
 

lib/app/layouts/conversation_view/widgets/message/reaction/reaction.dart

@@ -33,7 +33,7 @@ class ReactionWidget extends StatefulWidget {
 
 class ReactionWidgetState extends OptimizedState<ReactionWidget> {
   late Message reaction = widget.reaction;
-  late final StreamSubscription<Query<Message>> sub;
+  late final StreamSubscription sub;
   bool hasStream = false;
 
   List<Message>? get reactions => widget.reactions;
@@ -61,13 +61,22 @@ class ReactionWidgetState extends OptimizedState<ReactionWidget> {
             } else {
               reaction = _message;
             }
-            if (widget.message != null) {
+            getActiveMwc(widget.message!.guid!)?.updateAssociatedMessage(reaction, updateHolder: false);
-              getActiveMwc(widget.message!.guid!)?.updateAssociatedMessage(reaction, updateHolder: false);
-            }
           }
         });
 
         hasStream = true;
+      } else if (kIsWeb && widget.message != null) {
+        sub = WebListeners.messageUpdate.listen((tuple) {
+          final _message = tuple.item1;
+          final tempGuid = tuple.item2;
+          if (tempGuid == reaction.guid || _message.guid == reaction.guid) {
+            setState(() {
+              reaction = _message;
+            });
+            getActiveMwc(widget.message!.guid!)?.updateAssociatedMessage(reaction, updateHolder: false);
+          }
+        });
       }
     });
   }

lib/models/html/chat.dart

@@ -171,6 +171,8 @@ class Chat {
     bool updateDisplayName = false,
     bool updateDateDeleted = false,
   }) {
+    // ignore: argument_type_not_assignable, return_of_invalid_type, invalid_assignment, for_in_of_invalid_element_type
+    WebListeners.notifyChat(this);
     return this;
   }
 

lib/models/html/message.dart

@@ -207,18 +207,30 @@ class Message {
   }
 
   Message save({Chat? chat}) {
+    // ignore: argument_type_not_assignable, return_of_invalid_type, invalid_assignment, for_in_of_invalid_element_type
+    WebListeners.notifyMessage(this, chat: chat);
     return this;
   }
 
   static Future<List<Message>> bulkSaveNewMessages(Chat chat, List<Message> messages) async {
+    for (Message m in messages) {
+      // ignore: argument_type_not_assignable, return_of_invalid_type, invalid_assignment, for_in_of_invalid_element_type
+      WebListeners.notifyMessage(m, chat: chat);
+    }
     return [];
   }
 
   static List<Message> bulkSave(List<Message> messages) {
+    for (Message m in messages) {
+      // ignore: argument_type_not_assignable, return_of_invalid_type, invalid_assignment, for_in_of_invalid_element_type
+      WebListeners.notifyMessage(m);
+    }
     return [];
   }
 
   static Future<Message> replaceMessage(String? oldGuid, Message newMessage, {bool awaitNewMessageEvent = true, Chat? chat}) async {
+    // ignore: argument_type_not_assignable, return_of_invalid_type, invalid_assignment, for_in_of_invalid_element_type
+    WebListeners.notifyMessage(newMessage, tempGuid: oldGuid, chat: chat);
     return newMessage;
   }
 

lib/services/backend/action_handler.dart

@@ -241,7 +241,7 @@ class ActionHandler extends GetxService {
     if (tempGuid != null) return;
     Logger.info("New message: [${m.text}] - for chat [${c.guid}]", tag: "ActionHandler");
     // Gets the chat from the db or server (if new)
-    c = m.isParticipantEvent ? await handleNewOrUpdatedChat(c) : (Chat.findOne(guid: c.guid) ?? await handleNewOrUpdatedChat(c));
+    c = m.isParticipantEvent ? await handleNewOrUpdatedChat(c) : kIsWeb ? c : (Chat.findOne(guid: c.guid) ?? await handleNewOrUpdatedChat(c));
     // Get the message handle
     m.handle = c.handles.firstWhereOrNull((e) => e.originalROWID == m.handleId) ?? Handle.findOne(originalROWID: m.handleId);
     // Display notification if needed and save everything to DB

lib/services/backend/notifications/notifications_service.dart

@@ -99,6 +99,13 @@ class NotificationsService extends GetxService {
         }
         currentCount = newCount;
       });
+    } else {
+      countSub = WebListeners.newMessage.listen((tuple) {
+        final activeChatFetching = cm.activeChat != null ? ms(cm.activeChat!.chat.guid).isFetching : false;
+        if (ls.isAlive && !activeChatFetching && tuple.item2 != null) {
+          MessageHelper.handleNotification(tuple.item1, tuple.item2!, findExisting: false);
+        }
+      });
     }
   }
 

lib/services/backend/web/listeners.dart

@@ -0,0 +1,51 @@
+import 'dart:async';
+
+import 'package:bluebubbles/models/models.dart';
+import 'package:tuple/tuple.dart';
+
+/// Class to replace objectbox DB listener functionality with an old-fashioned
+/// stream based listener
+class WebListeners {
+  static final Set<String> _messageGuids = {};
+  static final Set<String> _chatGuids = {};
+
+  static final StreamController<Tuple3<Message, String?, Chat?>> _messageUpdate = StreamController.broadcast();
+  static final StreamController<Tuple2<Message, Chat?>> _newMessage = StreamController.broadcast();
+
+  static final StreamController<Chat> _chatUpdate = StreamController.broadcast();
+  static final StreamController<Chat> _newChat = StreamController.broadcast();
+
+  static Stream<Tuple3<Message, String?, Chat?>> get messageUpdate => _messageUpdate.stream;
+  static Stream<Tuple2<Message, Chat?>> get newMessage => _newMessage.stream;
+
+  static Stream<Chat> get chatUpdate => _chatUpdate.stream;
+  static Stream<Chat> get newChat => _newChat.stream;
+
+  static void notifyMessage(Message m, {Chat? chat, String? tempGuid}) {
+    if (tempGuid != null) {
+      if (_messageGuids.contains(tempGuid)) {
+        _messageGuids.add(m.guid!);
+        _messageUpdate.add(Tuple3(m, tempGuid, chat));
+      } else {
+        _messageGuids.add(tempGuid);
+        _newMessage.add(Tuple2(m, chat));
+      }
+    } else {
+      if (_messageGuids.contains(m.guid)) {
+        _messageUpdate.add(Tuple3(m, null, chat));
+      } else {
+        _messageGuids.add(m.guid!);
+        _newMessage.add(Tuple2(m, chat));
+      }
+    }
+  }
+
+  static void notifyChat(Chat c) {
+    if (_chatGuids.contains(c.guid)) {
+      _chatUpdate.add(c);
+    } else {
+      _chatGuids.add(c.guid);
+      _newChat.add(c);
+    }
+  }
+}

lib/services/services.dart

@@ -13,6 +13,7 @@ export 'backend/sync/incremental_sync_manager.dart';
 export 'backend/sync/sync_manager_impl.dart' show SyncStatus;
 export 'backend/sync/sync_service.dart';
 export 'backend/sync/tasks/sync_tasks.dart';
+export 'backend/web/listeners.dart';
 export 'backend/action_handler.dart';
 export 'backend_ui_interop/event_dispatcher.dart';
 export 'backend_ui_interop/intents.dart';

lib/services/ui/chat/chats_service.dart

@@ -47,6 +47,11 @@ class ChatsService extends GetxService {
         }
         currentCount = newCount;
       });
+    } else {
+      countSub = WebListeners.newChat.listen((chat) async {
+        if (!ss.settings.finishedSetup.value) return;
+        addChat(chat);
+      });
     }
   }
 
@@ -112,9 +117,7 @@ class ChatsService extends GetxService {
 
   @override
   void onClose() {
-    if (!kIsWeb) {
+    countSub.cancel();
-      countSub.cancel();
-    }
     super.onClose();
   }
 

lib/services/ui/message/message_widget_controller.dart

@@ -21,7 +21,6 @@ MessageWidgetController? getActiveMwc(String guid) => Get.isRegistered<MessageWi
 
 class MessageWidgetController extends StatefulController with SingleGetTickerProviderMixin {
   final RxBool showEdits = false.obs;
-  bool _init = false;
 
   List<MessagePart> parts = [];
   Message message;
@@ -29,7 +28,7 @@ class MessageWidgetController extends StatefulController with SingleGetTickerPro
   String? newMessageGuid;
   ConversationViewController? cvController;
   late final String tag;
-  late final StreamSubscription<Query<Message>> sub;
+  late final StreamSubscription sub;
 
   static const maxBubbleSizeFactor = 0.75;
 
@@ -45,7 +44,6 @@ class MessageWidgetController extends StatefulController with SingleGetTickerPro
     super.onInit();
     buildMessageParts();
     if (!kIsWeb && message.id != null) {
-      _init = true;
       final messageQuery = messageBox.query(Message_.id.equals(message.id!)).watch();
       sub = messageQuery.listen((Query<Message> query) async {
         if (message.id == null) return;
@@ -60,12 +58,20 @@ class MessageWidgetController extends StatefulController with SingleGetTickerPro
           updateMessage(_message);
         }
       });
+    } else if (kIsWeb) {
+      sub = WebListeners.messageUpdate.listen((tuple) {
+        final _message = tuple.item1;
+        final tempGuid = tuple.item2;
+        if (_message.guid == message.guid || tempGuid == message.guid) {
+          updateMessage(_message);
+        }
+      });
     }
   }
 
   @override
   void onClose() {
-    if (_init) sub.cancel();
+    sub.cancel();
     super.onClose();
   }
 
@@ -161,10 +167,11 @@ class MessageWidgetController extends StatefulController with SingleGetTickerPro
   }
 
   void updateMessage(Message newItem) {
+    final chat = message.chat.target?.guid ?? cvController?.chat.guid ?? cm.activeChat!.chat.guid;
     final oldGuid = message.guid;
     if (newItem.guid != oldGuid && oldGuid!.contains("temp")) {
       message = Message.merge(newItem, message);
-      ms(message.chat.target!.guid).updateMessage(message, oldGuid: oldGuid);
+      ms(chat).updateMessage(message, oldGuid: oldGuid);
       updateWidgets<MessageHolder>(null);
       if (message.isFromMe! && message.attachments.isNotEmpty) {
         updateWidgets<AttachmentHolder>(null);
@@ -172,9 +179,9 @@ class MessageWidgetController extends StatefulController with SingleGetTickerPro
     } else if (newItem.dateDelivered != message.dateDelivered || newItem.dateRead != message.dateRead || newItem.didNotifyRecipient != message.didNotifyRecipient) {
       final edited = newItem.dateEdited != message.dateEdited;
       message = Message.merge(newItem, message);
-      ms(message.chat.target!.guid).updateMessage(message);
+      ms(chat).updateMessage(message);
       // update the latest 2 messages in case their indicators need to go away
-      final messages = ms(message.chat.target!.guid).struct.messages
+      final messages = ms(chat).struct.messages
           .where((e) => e.isFromMe! && (e.dateDelivered != null || e.dateRead != null))
           .toList()..sort((a, b) => b.dateCreated!.compareTo(a.dateCreated!));
       for (Message m in messages.take(2)) {
@@ -190,7 +197,7 @@ class MessageWidgetController extends StatefulController with SingleGetTickerPro
       message = Message.merge(newItem, message);
       parts.clear();
       buildMessageParts();
-      ms(message.chat.target!.guid).updateMessage(message);
+      ms(chat).updateMessage(message);
       updateWidgets<MessageHolder>(null);
     }
   }

lib/services/ui/message/messages_service.dart

@@ -30,7 +30,6 @@ class MessagesService extends GetxController {
 
   int currentCount = 0;
   bool isFetching = false;
-  bool _init = false;
   String? method;
 
   Message? get mostRecentSent => (struct.messages.where((e) => e.isFromMe!).toList()
@@ -49,7 +48,6 @@ class MessagesService extends GetxController {
 
     // watch for new messages
     if (chat.id != null) {
-      _init = true;
       final countQuery = (messageBox.query(Message_.dateDeleted.isNull())
         ..link(Message_.chat, Chat_.id.equals(chat.id!))
         ..order(Message_.id, flags: Order.descending)).watch(triggerImmediately: true);
@@ -61,38 +59,23 @@ class MessagesService extends GetxController {
           final messages = event.find();
           event.limit = 0;
           for (Message message in messages) {
-            message.handle = message.getHandle();
+            await _handleNewMessage(message);
-            if (message.hasAttachments) {
-              message.attachments = List<Attachment>.from(message.dbAttachments);
-              // we may need an artificial delay in some cases since the attachment
-              // relation is initialized after message itself is saved
-              if (message.attachments.isEmpty) {
-                await Future.delayed(const Duration(milliseconds: 250));
-                message.attachments = List<Attachment>.from(message.dbAttachments);
-              }
-            }
-            // add this as a reaction if needed, update thread originators and associated messages
-            if (message.associatedMessageGuid != null) {
-              struct.getMessage(message.associatedMessageGuid!)?.associatedMessages.add(message);
-              getActiveMwc(message.associatedMessageGuid!)?.updateAssociatedMessage(message);
-            }
-            if (message.threadOriginatorGuid != null) {
-              getActiveMwc(message.threadOriginatorGuid!)?.updateThreadOriginator(message);
-            }
-            struct.addMessages([message]);
-            if (message.associatedMessageGuid == null) {
-              newFunc.call(message);
-            }
           }
         }
         currentCount = newCount;
       });
+    } else if (kIsWeb) {
+      countSub = WebListeners.newMessage.listen((tuple) {
+        if (tuple.item2?.guid == chat.guid) {
+          _handleNewMessage(tuple.item1);
+        }
+      });
     }
   }
 
   @override
   void onClose() {
-    if (_init) countSub.cancel();
+    countSub.cancel();
     super.onClose();
   }
 
@@ -108,6 +91,33 @@ class MessagesService extends GetxController {
     Get.reload<MessagesService>(tag: tag);
   }
 
+  Future<void> _handleNewMessage(Message message) async {
+    if (!kIsWeb) {
+      message.handle = message.getHandle();
+    }
+    if (message.hasAttachments && !kIsWeb) {
+      message.attachments = List<Attachment>.from(message.dbAttachments);
+      // we may need an artificial delay in some cases since the attachment
+      // relation is initialized after message itself is saved
+      if (message.attachments.isEmpty) {
+        await Future.delayed(const Duration(milliseconds: 250));
+        message.attachments = List<Attachment>.from(message.dbAttachments);
+      }
+    }
+    // add this as a reaction if needed, update thread originators and associated messages
+    if (message.associatedMessageGuid != null) {
+      struct.getMessage(message.associatedMessageGuid!)?.associatedMessages.add(message);
+      getActiveMwc(message.associatedMessageGuid!)?.updateAssociatedMessage(message);
+    }
+    if (message.threadOriginatorGuid != null) {
+      getActiveMwc(message.threadOriginatorGuid!)?.updateThreadOriginator(message);
+    }
+    struct.addMessages([message]);
+    if (message.associatedMessageGuid == null) {
+      newFunc.call(message);
+    }
+  }
+
   void updateMessage(Message updated, {String? oldGuid}) {
     final toUpdate = struct.getMessage(oldGuid ?? updated.guid!);
     if (toUpdate == null) return;