code: Observable<ComapiResult<MessagesQueryResponse>> processMessageQueryResponse(@NonNull String conversationId, @NonNull ComapiResult<MessagesQueryResponse> result) {
  
          final MessagesQueryResponse response = result.getResult();
  
          return result.isSuccessful() && response != null ?
  
                  Observable.create(emitter -> storeFactory.execute(new StoreTransaction<ChatStore>() {
  
                      @Override
                      protected void execute(ChatStore store) {
  
                          List<ChatMessage> messages = modelAdapter.adaptMessages(response.getMessages());
  
                          long updatedOn = 0;
  
                          store.beginTransaction();
  
                          if (messages != null && !messages.isEmpty()) {
                              for (ChatMessage msg : messages) {
                                  store.upsert(msg);
                                  if (msg.getSentOn() > updatedOn) {
                                      updatedOn = msg.getSentOn();
                                  }
                              }
                          }
  
                          ChatConversationBase savedConversation = store.getConversation(conversationId);
                          if (savedConversation != null) {
                              ChatConversationBase updateConversation = ChatConversationBase.baseBuilder()
                                      .setConversationId(savedConversation.getConversationId())
                                      .setETag(savedConversation.getETag())
                                      .setFirstLocalEventId(nullOrNegative(savedConversation.getFirstLocalEventId()) ? response.getEarliestEventId() : Math.min(savedConversation.getFirstLocalEventId(), response.getEarliestEventId()))
                                      .setLastLocalEventId(nullOrNegative(savedConversation.getLastLocalEventId()) ? response.getLatestEventId() : Math.max(savedConversation.getLastLocalEventId(), response.getLatestEventId()))
                                      .setLastRemoteEventId(nullOrNegative(savedConversation.getLastRemoteEventId()) ? response.getLatestEventId() : Math.max(savedConversation.getLastRemoteEventId(), response.getLatestEventId()))
                                      .setUpdatedOn(Math.max(savedConversation.getUpdatedOn(), updatedOn))
                                      .build();
  
                              store.update(updateConversation);
                          }
  
                          store.endTransaction();
  
                          emitter.onNext(result);
                          emitter.onCompleted();
                      }
                  }), Emitter.BackpressureMode.BUFFER) : Observable.fromCallable(() -> result);
      }
docs: Upsert messages from the query and updates conversation in single store transaction.
  
  @param conversationId Conversation unique id.
  @param result         Response from message query.
  @return Observable returning message query response from argument unchanged by the method.