artem
Commanded/CQRS and how to handle when user is trying same command when previous one is not handled yet
I am learning Commanded and CQRS in general with a small pet project that is essentially yet another wrapper over a ChatGPT conversation with the prompts/buttons specific to the use case. It is to look like:
- User: Please provide an opinion on this article
- AI: It’s a regular article on astrophysics…
- User clicks <More positive, please>
- AI: It’s great discovery on..
- User clicks <More positive, please>
- AI: Greatest discovery in 21st century..
- etc.
**Basic architecture / Control flow**
There is going to be a simple LiveView page for a conversation with initial text to analyze, buttons for human requests and texts for AI responses. AI and human messages happen in a strict order for now, i.e. [per initial business rules at least] user cannot click <More positive> two times, there’s no sense in it.
- User clicks are to generate Commands to Commanded. Click handlers disable buttons until command is fully processed under
:strongconsistency - OpenAI request should probably happen in enrichment middleware per commanded recommendations
2.1. Enriching in the aggregate could also be ok as for start at least there’s nothing else to happen to conversation while waiting for AI. However, that is likely to make event handler less testable, so middleware felt like a more proper choice - Event handler generates events such as
ConversationStarted,RequestStepAdded,ResponseStateAdded - Finally projector glues them all into a nicely denormalized Ecto
Conversationin a single Postgres table row
**The problem**
Everything should work fine until user reloads the page right after clicking the <More positive, please> button while we are waiting for an OpenAI response. After a page reload since events aren’t generated yet UI looks like user can click e.g. <More negative, please> and second command will be generated.
What happens then is that OpenAI eventually replies with a MorePositive response (that user didn’t even realize was still being generated) and then with a MoreNegatve response on top.
I can’t figure an elegant way to deal with such situation and hope for an advice.
**Ideal behavior**
In the ideal world one of these UX options could be acceptable:
- We forget about in-flight request completely like it never happened
1.1 But how to cancel that still in-flight command processing then? Via some manual messing with enrichment middleware to cancel in-flight HTTP request? - UI is still blocked waiting for response. But how would I know from just a conversation id that command handling for a particular conversation id is still in progress?
2.1 Can/should I somehow subscribe to a particular aggregate to get notified when all of its strongly consistent activities are over?
2.2. Or would it be better to keep such UI-level status outside of commanded completely, so I could e.g. store in yet another Ecto table (or cache even as we are talking about seconds or a couple of minutes max) outside Commanded the fact that particular conversation is waiting for an update?
2.3. Or would you implement more fine-grained Commands/Events so that first we [very quickly] get something likeRequestStepAddedevent (projection of which would keep UI disabled) with a process manager adding a CommandNowGoAskOpenAI. But that’s a bunch of extra commands/events just for UI and a need for some real HTTP error handling (if all enrichment happens in the original command middleware, then error doesn’t need to touch any aggregates)
**How to deal with it?**
I am puzzled. All these options don’t seem very elegant to me and feel like I am about to create lots of complexity for just a small UI candy. Yet I would love for the user not to have surprises of sudden conversation update without a visible reason.
I hope I am not the first one having situation when UI needs some progress information about a long term action initiated before-page-reload (it’s not event that much about Commanded/CQRS I suppose).
Am I missing some obvious option or are the options I was able to figure less scary that they seem to me?
What do you usually do in this sort of situation?
Most Liked
artem
Hmm, I intend to do exactly this. My current Conversation aggregate only allows commands from different parties in a strict order. Second human’s command RequestAdjustment in a row would be rejected if aggregate didn’t process the AI response yet. Enforcing this particular constraint on the aggregate level seems to be not very complex. I might be missing something, but to me it is indeed just about visualizing “AI is still thinking” part of a problem.
**CRUD ways**
In very traditional(ish) CRUD web service with synchronous request handling and no queues this could be solved by transactions in a HTTP request handling thread. When HTTP response is sent, the data is already written to DB. If user cancels the request, request handling thread would be dead and transaction would be canceled.
If you add some queues/asynchronousity on top of it, then if user reloads the page you essentially arrive to the same problem of “how do I visualize that previous order handling is still in progress”. Relatively straightforward way for solving it seems to store “in progress” status either in DB/external service such as ZooKeeper or Redis (if you really want to store important status even across server restarts) or in some local in-memory cache such as EhCache.
**What makes CQRS world special**
Maybe nothing 
Maybe I should just use something like elixir cachex to store a not-too-important in-progress status. I am new to the world of CQRS (and to elixir as well in fact) and I assumed that in a world where everything is essentially a set of commands in-flight being transformed to projections this sort of “how do I show user even after page refresh that it’s too early to retry” is a very standard situation, so there could be usual answers as well. Then I would go an read more on popular options and reasoning for it.
Like for example, if people would be saying much that it is a usual thing to have very detailed commands-events (e.g. AdjustmentRequestedByHuman event essentially meaning “in-progress period started”, then AIAskedForResponse, then AIProvidedResponse), I’d go study this approach more.
Last Post!
lud
This looks like any event based system, I don’t see the Q of CQRS.
if your main goal is to try commanded then you can disregard this comment, but otherwise this is what I would do.
In live view
- First your conversation needs an
id. - In liveview mount, when
connected?(socket)istrue, you start a GenSever under a DynamicSupervisor, that is registered on a Registry with a{:via, Registry, {YourRegistry, id}}tuple. This may return{:ok, pid}or{:error, {:already started, pid}}, in both cases you have the pid of your conversation server. - Your live view calls
checkouton that server, which sends the view’sself()pid to the server. - The view can now ask that server what its current status is, and send commands to it.
In the conversation server
-
On init you define your state with a
:awaiting_articlestatus and anilowner. -
All your init, handle_call, handle_info and handle_cast should return a timeout if the current owner is
nil. -
When receiving a
checkoutcommand with the pid of the live view, set the owner to that pid, and monitor the pid. -
The server will then receive commands like
:set_article,:more_positive. You can implement a simple FSM (or use :gen_statem instead of GenServer but that’s overkill here). Depending on the current status, the server will react appropriately.status command/event action new_status awaiting_article set_article store in state, ask AI await_ai await_ai ai_response send to owner await_user await_user more_positive ask AI await_ai await_user set_article reset state, ask AI await_aiAnd so on…
When an unexpected command is received (double click on more_positive), the server can return an error, or just ignore it.
-
To deal with AI, the server can use a Task under a TaskSupervisor so the
ai_responseevent would simply be the task response inhandle_info, matching the ref of the task stored in state. -
When receiving a
:DOWNmessage from the monitor, set the owner tonil. -
When receiving a
:timeout(returned by your callbacks), if the owner isnilthen stop. -
On most events in handle_info, if the owner is not nil, send updates to the view.
And so,
- When the user reloads the page, the server will receive a
:DOWNbut will also be checked out again, before the timeout hits, and will keep its current task running, if any. - You do not even need a DB for a base implementation.
- If you want to cancel your in-flight requests to AI when receiving a
:DOWNmessage, the conversation server can just call a function to do so. But you will need to replace the Task. Maybe just usingFinch.async_request/3andFinch.cancel_async_request/1directly from the conversation server.
Anyway, I think this model is far more simpler than using commanded. There is always the same problem with CQRS, very often there is a need to know if some events are already in the queue before sending another event. When this need arises, it means that you do not want CQRS but rather a classic architecture where your updates to the database are fully handled synchronously.
Popular in Questions
Other popular topics
Latest on Elixir Forum
Sponsor Spotlight
Catch errors, track performance, monitor hosts and more.
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #channels
- #elixirconf
- #exunit
- #discussion
- #code-sync
- #javascript
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixir-ls
- #phoenix_html
- #iex
- #blog-post
- #graphql
- #genstage
- #ai
- #websockets
- #supervisor
- #elixirconf-us
- #advent-of-code
- #distillery
- #processes
- #forms
- #api
- #metaprogramming
- #security
- #hex
Our Sponsors
Build Elixir applications with speed and confidence.
Supporting innovation across the BEAM ecosystem.
We build reliable cloud platforms for business-critical systems.
Catch errors, track performance, monitor hosts and more.
Error tracking for Elixir devs who love to ship. Start your free account.
Practical resources that improve the lives of professional developers.
Develop your skills with books, videos, and courses.
The deployment platform built for Elixir: PaaS ease, VPS control.
The team behind Membrane, Popcorn, LiveDebugger. Available for hire.
Courses that'll move you from confusion to "Aha, now I get it!"