incremental chat state work

lib/chat/cubits/chat_component_cubit.dart

@@ -233,7 +233,8 @@ class ChatComponentCubit extends Cubit<ChatComponentState> {
 
     return types.User(
         id: remoteIdentityPublicKey.toString(),
-        firstName: activeConversationState.remoteConversation.profile.name,
+        firstName: activeConversationState.remoteConversation?.profile.name ??
+            '<unnamed>',
         metadata: {metadataKeyIdentityPublicKey: remoteIdentityPublicKey});
   }
 

lib/chat/cubits/single_contact_messages_cubit.dart

@@ -55,7 +55,7 @@ class SingleContactMessagesCubit extends Cubit<SingleContactMessagesState> {
     required TypedKey localConversationRecordKey,
     required TypedKey localMessagesRecordKey,
     required TypedKey remoteConversationRecordKey,
-    required TypedKey remoteMessagesRecordKey,
+    required TypedKey? remoteMessagesRecordKey,
   })  : _accountInfo = accountInfo,
         _remoteIdentityPublicKey = remoteIdentityPublicKey,
         _localConversationRecordKey = localConversationRecordKey,
@@ -147,8 +147,14 @@ class SingleContactMessagesCubit extends Cubit<SingleContactMessagesState> {
 
   // Open remote messages key
   Future<void> _initRcvdMessagesCubit() async {
+    // Don't bother if we don't have a remote messages record key yet
+    if (_remoteMessagesRecordKey == null) {
+      return;
+    }
+
+    // Open new cubit if one is desired
     _rcvdMessagesCubit = DHTLogCubit(
-        open: () async => DHTLog.openRead(_remoteMessagesRecordKey,
+        open: () async => DHTLog.openRead(_remoteMessagesRecordKey!,
             debugName: 'SingleContactMessagesCubit::_initRcvdMessagesCubit::'
                 'RcvdMessages',
             parent: _remoteConversationRecordKey,
@@ -159,6 +165,31 @@ class SingleContactMessagesCubit extends Cubit<SingleContactMessagesState> {
     _updateRcvdMessagesState(_rcvdMessagesCubit!.state);
   }
 
+  Future<void> updateRemoteMessagesRecordKey(
+      TypedKey? remoteMessagesRecordKey) async {
+    await _initWait();
+
+    _sspRemoteConversationRecordKey.updateState(remoteMessagesRecordKey,
+        (remoteMessagesRecordKey) async {
+      // Don't bother if nothing is changing
+      if (_remoteMessagesRecordKey == remoteMessagesRecordKey) {
+        return;
+      }
+
+      // Close existing cubit if we have one
+      final rcvdMessagesCubit = _rcvdMessagesCubit;
+      _rcvdMessagesCubit = null;
+      _remoteMessagesRecordKey = null;
+      await _rcvdSubscription?.cancel();
+      _rcvdSubscription = null;
+      await rcvdMessagesCubit?.close();
+
+      // Init the new cubit if we should
+      _remoteMessagesRecordKey = remoteMessagesRecordKey;
+      await _initRcvdMessagesCubit();
+    });
+  }
+
   Future<VeilidCrypto> _makeLocalMessagesCrypto() async =>
       VeilidCryptoPrivate.fromTypedKey(
           _accountInfo.userLogin!.identitySecret, 'tabledb');
@@ -452,7 +483,7 @@ class SingleContactMessagesCubit extends Cubit<SingleContactMessagesState> {
   final TypedKey _localConversationRecordKey;
   final TypedKey _localMessagesRecordKey;
   final TypedKey _remoteConversationRecordKey;
-  final TypedKey _remoteMessagesRecordKey;
+  TypedKey? _remoteMessagesRecordKey;
 
   late final VeilidCrypto _conversationCrypto;
   late final MessageIntegrity _senderMessageIntegrity;
@@ -471,4 +502,6 @@ class SingleContactMessagesCubit extends Cubit<SingleContactMessagesState> {
       _reconciledSubscription;
   final StreamController<Future<void> Function()> _commandController;
   late final Future<void> _commandRunnerFut;
+
+  final _sspRemoteConversationRecordKey = SingleStateProcessor<TypedKey?>();
 }

lib/conversation/cubits/active_conversations_bloc_map_cubit.dart

@@ -24,7 +24,7 @@ class ActiveConversationState extends Equatable {
   final TypedKey localConversationRecordKey;
   final TypedKey remoteConversationRecordKey;
   final proto.Conversation localConversation;
-  final proto.Conversation remoteConversation;
+  final proto.Conversation? remoteConversation;
 
   @override
   List<Object?> get props => [
@@ -102,7 +102,8 @@ class ActiveConversationsBlocMapCubit extends BlocMapCubit<TypedKey,
         conversationCubit.watchAccountChanges(
             _accountRecordCubit.stream, _accountRecordCubit.state);
 
-        // Transformer that only passes through completed/active conversations
+        // Transformer that only passes through conversations where the local
+        // portion is not loading
         // along with the contact that corresponds to the completed
         // conversation
         final transformedCubit = TransformerCubit<
@@ -110,12 +111,11 @@ class ActiveConversationsBlocMapCubit extends BlocMapCubit<TypedKey,
                 AsyncValue<ConversationState>,
                 ConversationCubit>(conversationCubit,
             transform: (avstate) => avstate.when(
-                data: (data) => (data.localConversation == null ||
+                data: (data) => (data.localConversation == null)
-                        data.remoteConversation == null)
                     ? const AsyncValue.loading()
                     : AsyncValue.data(ActiveConversationState(
                         localConversation: data.localConversation!,
-                        remoteConversation: data.remoteConversation!,
+                        remoteConversation: data.remoteConversation,
                         remoteIdentityPublicKey: remoteIdentityPublicKey,
                         localConversationRecordKey: localConversationRecordKey,
                         remoteConversationRecordKey:

lib/conversation/cubits/active_single_contact_chat_bloc_map_cubit.dart

@@ -25,7 +25,7 @@ class _SingleContactChatState extends Equatable {
   final TypedKey localConversationRecordKey;
   final TypedKey remoteConversationRecordKey;
   final TypedKey localMessagesRecordKey;
-  final TypedKey remoteMessagesRecordKey;
+  final TypedKey? remoteMessagesRecordKey;
 
   @override
   List<Object?> get props => [
@@ -53,8 +53,16 @@ class ActiveSingleContactChatBlocMapCubit extends BlocMapCubit<TypedKey,
     follow(activeConversationsBlocMapCubit);
   }
 
-  Future<void> _addConversationMessages(_SingleContactChatState state) async =>
+  Future<void> _addConversationMessages(_SingleContactChatState state) async {
-      add(() => MapEntry(
+    // xxx could use atomic update() function
+
+    final cubit = await tryOperateAsync<SingleContactMessagesCubit>(
+        state.localConversationRecordKey, closure: (cubit) async {
+      await cubit.updateRemoteMessagesRecordKey(state.remoteMessagesRecordKey);
+      return cubit;
+    });
+    if (cubit == null) {
+      await add(() => MapEntry(
           state.localConversationRecordKey,
           SingleContactMessagesCubit(
             accountInfo: _accountInfo,
@@ -64,6 +72,8 @@ class ActiveSingleContactChatBlocMapCubit extends BlocMapCubit<TypedKey,
             localMessagesRecordKey: state.localMessagesRecordKey,
             remoteMessagesRecordKey: state.remoteMessagesRecordKey,
           )));
+    }
+  }
 
   _SingleContactChatState? _mapStateValue(
       AsyncValue<ActiveConversationState> avInputState) {
@@ -78,7 +88,7 @@ class ActiveSingleContactChatBlocMapCubit extends BlocMapCubit<TypedKey,
         localMessagesRecordKey:
             inputState.localConversation.messages.toVeilid(),
         remoteMessagesRecordKey:
-            inputState.remoteConversation.messages.toVeilid());
+            inputState.remoteConversation?.messages.toVeilid());
   }
 
   /// StateFollower /////////////////////////

lib/conversation/cubits/conversation_cubit.dart

@@ -226,13 +226,11 @@ class ConversationCubit extends Cubit<AsyncValue<ConversationState>> {
             localConversation: conv,
             remoteConversation: _incrementalState.remoteConversation);
         // return loading still if state isn't complete
-        if ((_localConversationRecordKey != null &&
+        if (_localConversationRecordKey != null &&
-                _incrementalState.localConversation == null) ||
+            _incrementalState.localConversation == null) {
-            (_remoteConversationRecordKey != null &&
-                _incrementalState.remoteConversation == null)) {
           return const AsyncValue<ConversationState>.loading();
         }
-        // state is complete, all required keys are open
+        // local state is complete, all remote state is emitted incrementally
         return AsyncValue.data(_incrementalState);
       },
       loading: AsyncValue<ConversationState>.loading,
@@ -247,14 +245,12 @@ class ConversationCubit extends Cubit<AsyncValue<ConversationState>> {
         _incrementalState = ConversationState(
             localConversation: _incrementalState.localConversation,
             remoteConversation: conv);
-        // return loading still if state isn't complete
+        // return loading still if the local state isn't complete
-        if ((_localConversationRecordKey != null &&
+        if (_localConversationRecordKey != null &&
-                _incrementalState.localConversation == null) ||
+            _incrementalState.localConversation == null) {
-            (_remoteConversationRecordKey != null &&
-                _incrementalState.remoteConversation == null)) {
           return const AsyncValue<ConversationState>.loading();
         }
-        // state is complete, all required keys are open
+        // local state is complete, all remote state is emitted incrementally
         return AsyncValue.data(_incrementalState);
       },
       loading: AsyncValue<ConversationState>.loading,

lib/tools/loggy.dart

@@ -112,9 +112,9 @@ class CallbackPrinter extends LoggyPrinter {
   @override
   void onLog(LogRecord record) {
     final out = record.pretty();
-    if (isDesktop) {
+    //if (isDesktop) {
-      debugPrintSynchronously(out);
+    debugPrintSynchronously(out);
-    }
+    //}
     globalDebugTerminal.write('$out\n'.replaceAll('\n', '\r\n'));
     callback?.call(record);
   }

lib/tools/state_logger.dart

@@ -4,12 +4,12 @@ import 'loggy.dart';
 
 const Map<String, LogLevel> _blocChangeLogLevels = {
   'ConnectionStateCubit': LogLevel.off,
-  'ActiveSingleContactChatBlocMapCubit': LogLevel.off,
+  //'ActiveSingleContactChatBlocMapCubit': LogLevel.off,
-  'ActiveConversationsBlocMapCubit': LogLevel.off,
+  //'ActiveConversationsBlocMapCubit': LogLevel.off,
   'PersistentQueueCubit<Message>': LogLevel.off,
   'TableDBArrayProtobufCubit<ReconciledMessage>': LogLevel.off,
   'DHTLogCubit<Message>': LogLevel.off,
-  'SingleContactMessagesCubit': LogLevel.off,
+  //'SingleContactMessagesCubit': LogLevel.off,
   'ChatComponentCubit': LogLevel.off,
 };