You might want to look into MVI and how it is handling these problems. MVI is handling all the problem you raising. I can suggest to look into this: they way I would do it is:

class ChatViewModel(private val messageManager: MessageManager): ViewModel {
  private val binder: PublishProcessor<MessageEvent> = PublishProcessor.create()
  private val state =
    binder
      .flatMap{ messageManager.editMessage(it.messageId, it.newText) }
      // #1
      .replay(1)
      .autoConnect(0)

  fun editMessageClick(messageId: Long, newText: String) {
    binder.onNext(MessageEvent(messageId, newText))
  }
  override fun state(): Observable = state
}

class MessageManager {
  fun editMessage(messageId, newText) = 
    database.update(messageId, newText, State.SENDING)
      .flatMap { apiClient.editMessage(...)}
      .flatMap { database.update(messageId, newText, State.SENT )}
      .asEvents()
}

data class MessageEvent(val messageId: Long, val newText: String)

hi, reducing is not the problem, also you turned MessageManager.editMessage into a cold observable, so it will die once ChatViewModel (although im now looking at the

state

, your subscrption would I presume be in the ui, and it would not stop because of the autocnnect, but you still need to dispose of this, otherwise you create leaks afaik?)

Why do you want to represent the edit message as an endless stream? What I understand is that you want to observe the changes of the database not the request itself.

not really, for example you wanna show progressbar while editMessage Single is in flight. yes you could add it as a column to the database 'isEditing' but I really dont want to do that + you have to have error column too; and do this per every action ..edit, delete, retry, add reaction, pin, unpin etc

I think what you are looking for is an event based system where for every “edit” interaction an in cache store is updated with the messageID. This service kept alive as long as the application is alive and when the all is closed maybe pushed to a BackgroundTask to finish the execution. You can use the MVI to keep the UI always up to date.

well I think the issue is the same, how do you update the store? is it a cold observable update or regular function update returning Unit

What you are looking for is a proper architecture your your specific problem. a ViewModel can depend on other objects which keeps this state alive as long as the application exists.

well I think the issue is the same, how do you update the store?

As we described a hot observable receives events, reduce the events to state (map of elements) I would suggest a regular Unit function for propagating events.

if you have multiple edits going on, what do you emit, would you have multiple hot streams or a single one with a map of ongoing events?

Map of ongoing events. You can always filter these events if you need to.

trouble is how to keep these mutations manager or appscoped

dependency injection?

Yes I know all that, the core question is what that design of such domain level component should look like. Viewmodel is not the issue. Okay I'll think about the map. Scoping, Its not a trouble to actually do, By scoping I mean the observable should only live as long as its container (so when container goes out of scope, it should be canceled)