Optimize WorkflowChannel PubSub with workflow-specific topics
Nobody has claimed this yet.
- Dominant language
- Elixir
- Stars
- 296
- Forks
- 86
- Avg merge
- 1d 13h
- Merged PRs (30d)
- 50
Description
Details
The WorkflowChannel currently subscribes to project-level work order events (WorkOrders.subscribe(project_id)), which means every socket receives ALL work order and run events for the entire project, then filters them down to only the relevant workflow.
Current implementation (workflow_channel.ex:64):
WorkOrders.subscribe(project.id) # Gets ALL project events
This causes performance issues:
-
WorkOrder events - Filtered in handle_info (lines 349, 366):
if wo.workflow_id == socket.assigns.workflow_id do # Push update end -
Run events - Require database queries to filter (lines 385, 409):
case WorkOrders.get(run.work_order_id, include: [:workflow]) do %{workflow_id: workflow_id} when workflow_id == socket.assigns.workflow_id -> # Push update _ -> :ok end
Scale problem:
- Project with 10 workflows
- 20 users editing different workflows
- Work order created in workflow A
- Result: ALL 20 sockets receive event, 19 discard it after checking/querying
Even worse for runs: Every run event requires a database query per socket just to check if it should be ignored.
Implementation notes
Recommended approach: Multi-level topic broadcasting
The current implementation subscribes WorkflowChannel to project-level topics, then filters out 90% of events. This violates the principle: "if we're listening for it, we probably want it."
Current subscribers and their needs:
- WorkListener - Subscribes to
"all_events", needs ALL runs globally - RunLive.Index - Subscribes to
"project:#{project_id}", needs ALL events for all workflows in project - WorkflowChannel - Subscribes to
"project:#{project_id}", needs ONLY events for ONE workflow (currently filters!)
Solution: Broadcast to multiple topic levels
Add workflow-specific topics and broadcast events to all relevant levels:
# workorders/events.ex - Add workflow topic function
defp topic(project_id, workflow_id), do: "workflow:#{workflow_id}"
# Broadcast work order events to 2 levels
# WorkOrder has workflow_id field, so we can extract it
def work_order_created(project_id, work_order) do
event = %WorkOrderCreated{work_order: work_order, project_id: project_id}
Lightning.broadcast(topic(project_id), event) # RunLive.Index
Lightning.broadcast(topic(project_id, work_order.workflow_id), event) # WorkflowChannel
end
# Broadcast run events to 3 levels
# Run doesn't have workflow_id field, so must be passed as parameter
def run_created(project_id, workflow_id, run) do
event = %RunCreated{run: run, project_id: project_id}
Lightning.broadcast(topic(), event) # WorkListener (global)
Lightning.broadcast(topic(project_id), event) # RunLive.Index (project)
Lightning.broadcast(topic(project_id, workflow_id), event) # WorkflowChannel (workflow)
end
Update WorkflowChannel subscription:
# workflow_channel.ex:64 - Subscribe to workflow-specific topic
WorkOrders.subscribe(project.id, workflow.id)
# No filtering needed - all received events are relevant!
def handle_info(%WorkOrders.Events.RunCreated{run: run}, socket) do
formatted_run = format_run_for_history(run)
push(socket, "history_updated", %{...})
{:noreply, socket}
end
Benefits:
- No filtering logic in WorkflowChannel
- No database queries to check relevance
- Each subscriber only receives events it cares about
- Follows "subscribe to what you need" principle
Related code
Event broadcasting functions to update:
lib/lightning/workorders/events.ex:24-48- Add workflow_id param, broadcast to multiple topicslib/lightning/workorders/events.ex:55-57- Add subscribe(project_id, workflow_id) function
Callers to update:
lib/lightning/work_orders.ex:137-138- work_order_created + run_created (workorder has workflow_id field)lib/lightning/work_orders.ex:367-368- work_order_updated + run_created (workorder has workflow_id field)lib/lightning/work_orders.ex:733-736- work_order_updated + run_created (workorder has workflow_id field)lib/lightning/runs/handlers.ex:152- run_updated (run.workflow preloaded, use run.workflow.id)
Note: WorkOrder structs have workflow_id as a field (via belongs_to :workflow), so event functions can extract it directly from the work_order without needing it passed separately.
Subscription to update:
lib/lightning_web/channels/workflow_channel.ex:64- Subscribe to workflow-specific topic
Filtering to remove:
lib/lightning_web/channels/workflow_channel.ex:349,366- WorkOrder workflow_id checkslib/lightning_web/channels/workflow_channel.ex:379-401- RunCreated with DB querylib/lightning_web/channels/workflow_channel.ex:404-425- RunUpdated with DB query
Release notes
Optimized collaborative editor real-time updates to reduce database queries and event processing overhead when multiple workflows are being edited simultaneously.
User acceptance criteria
- Workflow-specific PubSub topics added:
"workflow:#{workflow_id}" - Event functions accept workflow_id parameter and broadcast to multiple topics
- All callers pass workflow_id (4 locations in work_orders.ex and runs/handlers.ex)
- WorkflowChannel subscribes to workflow-specific topic, not project-level
- WorkflowChannel has NO filtering logic (all received events are relevant)
- WorkflowChannel has NO database queries in handle_info for filtering
- RunLive.Index still receives all project-wide events (unchanged behavior)
- WorkListener still receives all global run events (unchanged behavior)
- Existing history panel functionality works identically
- Tests updated for new subscription patterns
- Manual test: Multiple workflows editing simultaneously shows no N+1 queries
Contributor guide
No contributing guide indexed for this repository
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start in lib/lightning/workorders/events.ex and lib/lightning_web/channels/workflow_channel.ex, then trace the listed callers in lib/lightning/work_orders.ex and lib/lightning/runs/handlers.ex. Verify the workflow-specific subscription and multi-level broadcasts preserve RunLive.Index and WorkListener behavior. Done means filtering and database queries are removed from WorkflowChannel, callers pass workflow context, tests are updated, and the history panel remains functional.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- elixir
- Domain
- backend, distributed-systems
- Issue type
- Refactor
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Clearly specified
- Newbie friendliness
- 45/100