| 1 |
|
defmodule WalletJourney.Commands.AdvanceJourneyStep do |
| 2 |
|
@moduledoc """ |
| 3 |
|
Command handler for manually advancing a journey to the next step. |
| 4 |
|
|
| 5 |
|
Used when external orchestration needs to record step completion |
| 6 |
|
outside of the synchronous coordinator pipeline. |
| 7 |
|
""" |
| 8 |
|
|
| 9 |
|
alias WalletJourney.{Journey, JourneyStore} |
| 10 |
|
alias WalletJourney.Events.{JourneyStepAdvanced, JourneyCompleted} |
| 11 |
|
alias WalletObservability.AuditEvent |
| 12 |
|
|
| 13 |
|
@doc """ |
| 14 |
|
Records a step as completed on the given journey. |
| 15 |
|
|
| 16 |
|
Returns `{:ok, journey}` on success. |
| 17 |
|
Returns `{:error, :not_found}` if journey_id does not exist. |
| 18 |
|
Returns `{:error, :already_completed}` if step is already in completed_steps. |
| 19 |
|
Returns `{:error, :invalid_step}` if step is not in the journey's step list. |
| 20 |
|
""" |
| 21 |
|
@spec execute(journey_id :: String.t(), step :: atom(), step_result :: atom()) :: |
| 22 |
|
{:ok, Journey.t()} | {:error, term()} |
| 23 |
:-( |
def execute(journey_id, step, step_result \\ :ok) do |
| 24 |
:-( |
with {:ok, journey} <- JourneyStore.get(journey_id), |
| 25 |
:-( |
:ok <- check_not_already_completed(journey, step), |
| 26 |
:-( |
:ok <- check_valid_step(journey, step) do |
| 27 |
:-( |
completed_steps = journey.completed_steps ++ [step] |
| 28 |
:-( |
updated = %{journey | completed_steps: completed_steps, status: :in_progress} |
| 29 |
|
|
| 30 |
:-( |
emit_event( |
| 31 |
|
JourneyStepAdvanced.build( |
| 32 |
:-( |
updated.journey_id, |
| 33 |
|
step, |
| 34 |
|
completed_steps, |
| 35 |
:-( |
correlation_id: updated.correlation_id, |
| 36 |
:-( |
current_step: updated.current_step, |
| 37 |
|
step_result: step_result |
| 38 |
|
) |
| 39 |
|
) |
| 40 |
|
|
| 41 |
:-( |
final = |
| 42 |
:-( |
if completed_steps == journey.steps do |
| 43 |
:-( |
done = %{updated | status: :completed, completed_at: DateTime.utc_now(), current_step: nil} |
| 44 |
:-( |
JourneyStore.update(done) |
| 45 |
|
|
| 46 |
:-( |
emit_event( |
| 47 |
|
JourneyCompleted.build( |
| 48 |
:-( |
done.journey_id, |
| 49 |
:-( |
done.transfer_id, |
| 50 |
:-( |
done.completed_steps, |
| 51 |
:-( |
correlation_id: done.correlation_id |
| 52 |
|
) |
| 53 |
|
) |
| 54 |
|
|
| 55 |
:-( |
emit_audit(done, "journey_completed", :success, done.correlation_id) |
| 56 |
:-( |
done |
| 57 |
|
else |
| 58 |
:-( |
JourneyStore.update(updated) |
| 59 |
:-( |
updated |
| 60 |
|
end |
| 61 |
|
|
| 62 |
|
{:ok, final} |
| 63 |
|
end |
| 64 |
|
rescue |
| 65 |
:-( |
error -> {:error, error} |
| 66 |
|
end |
| 67 |
|
|
| 68 |
|
defp check_not_already_completed(journey, step) do |
| 69 |
:-( |
if step in journey.completed_steps do |
| 70 |
|
{:error, :already_completed} |
| 71 |
|
else |
| 72 |
|
:ok |
| 73 |
|
end |
| 74 |
|
end |
| 75 |
|
|
| 76 |
|
defp check_valid_step(journey, step) do |
| 77 |
:-( |
if step in journey.steps do |
| 78 |
|
:ok |
| 79 |
|
else |
| 80 |
|
{:error, :invalid_step} |
| 81 |
|
end |
| 82 |
|
end |
| 83 |
|
|
| 84 |
:-( |
defp emit_event(event) do |
| 85 |
:-( |
pubsub = Application.get_env(:wallet_journey, :pubsub, WalletWeb.PubSub) |
| 86 |
|
|
| 87 |
:-( |
apply(Phoenix.PubSub, :broadcast, [ |
| 88 |
|
pubsub, |
| 89 |
|
"wallet_journey:events", |
| 90 |
|
{:domain_event, event} |
| 91 |
|
]) |
| 92 |
|
rescue |
| 93 |
:-( |
_ -> :ok |
| 94 |
|
end |
| 95 |
|
|
| 96 |
:-( |
defp emit_audit(journey, action, outcome, correlation_id) do |
| 97 |
:-( |
audit = |
| 98 |
|
AuditEvent.build( |
| 99 |
|
:process, |
| 100 |
|
action, |
| 101 |
|
"journey", |
| 102 |
:-( |
journey.journey_id, |
| 103 |
|
outcome, |
| 104 |
:-( |
actor_id: journey.user_id, |
| 105 |
|
correlation_id: correlation_id, |
| 106 |
:-( |
metadata: %{transfer_id: journey.transfer_id, status: journey.status} |
| 107 |
|
) |
| 108 |
|
|
| 109 |
:-( |
:telemetry.execute([:wallet_journey, :audit], %{}, audit) |
| 110 |
|
rescue |
| 111 |
:-( |
_ -> :ok |
| 112 |
|
end |
| 113 |
|
end |