| 1 |
|
defmodule WalletJourney.Commands.CompensateJourney do |
| 2 |
|
@moduledoc """ |
| 3 |
|
Command handler for compensating a journey that has already initiated a transfer. |
| 4 |
|
|
| 5 |
|
Triggers cancellation of the associated transfer (if any) and marks the |
| 6 |
|
journey as :compensated. |
| 7 |
|
""" |
| 8 |
|
|
| 9 |
|
alias WalletJourney.{Journey, JourneyStore} |
| 10 |
|
alias WalletJourney.Events.JourneyCompensated |
| 11 |
|
alias WalletObservability.AuditEvent |
| 12 |
|
|
| 13 |
|
@doc """ |
| 14 |
|
Compensates the given journey by cancelling its transfer and marking it :compensated. |
| 15 |
|
|
| 16 |
|
Returns `{:ok, journey}` on success. |
| 17 |
|
Returns `{:error, :not_found}` if journey_id does not exist. |
| 18 |
|
Returns `{:error, :already_compensated}` if already in :compensating or :compensated status. |
| 19 |
|
""" |
| 20 |
|
@spec execute(journey_id :: String.t(), reason :: String.t()) :: |
| 21 |
|
{:ok, Journey.t()} | {:error, term()} |
| 22 |
6 |
def execute(journey_id, reason) do |
| 23 |
6 |
with {:ok, journey} <- JourneyStore.get(journey_id), |
| 24 |
5 |
:ok <- check_not_already_compensated(journey) do |
| 25 |
3 |
compensating = %{journey | status: :compensating} |
| 26 |
3 |
JourneyStore.update(compensating) |
| 27 |
|
|
| 28 |
3 |
if journey.transfer_id do |
| 29 |
2 |
WalletTransfers.cancel(journey.transfer_id, reason) |
| 30 |
|
end |
| 31 |
|
|
| 32 |
3 |
compensated = %{compensating | status: :compensated, failure_reason: reason} |
| 33 |
3 |
JourneyStore.update(compensated) |
| 34 |
|
|
| 35 |
3 |
emit_event( |
| 36 |
|
JourneyCompensated.build( |
| 37 |
3 |
compensated.journey_id, |
| 38 |
3 |
compensated.transfer_id, |
| 39 |
|
reason, |
| 40 |
3 |
correlation_id: compensated.correlation_id |
| 41 |
|
) |
| 42 |
|
) |
| 43 |
|
|
| 44 |
3 |
emit_audit(compensated, "journey_compensated", :failure, compensated.correlation_id) |
| 45 |
|
|
| 46 |
|
{:ok, compensated} |
| 47 |
|
end |
| 48 |
|
rescue |
| 49 |
:-( |
error -> {:error, error} |
| 50 |
|
end |
| 51 |
|
|
| 52 |
|
defp check_not_already_compensated(journey) do |
| 53 |
5 |
if journey.status in [:compensating, :compensated] do |
| 54 |
|
{:error, :already_compensated} |
| 55 |
|
else |
| 56 |
|
:ok |
| 57 |
|
end |
| 58 |
|
end |
| 59 |
|
|
| 60 |
3 |
defp emit_event(event) do |
| 61 |
3 |
pubsub = Application.get_env(:wallet_journey, :pubsub, WalletWeb.PubSub) |
| 62 |
|
|
| 63 |
3 |
apply(Phoenix.PubSub, :broadcast, [ |
| 64 |
|
pubsub, |
| 65 |
|
"wallet_journey:events", |
| 66 |
|
{:domain_event, event} |
| 67 |
|
]) |
| 68 |
|
rescue |
| 69 |
:-( |
_ -> :ok |
| 70 |
|
end |
| 71 |
|
|
| 72 |
3 |
defp emit_audit(journey, action, outcome, correlation_id) do |
| 73 |
3 |
audit = |
| 74 |
|
AuditEvent.build( |
| 75 |
|
:process, |
| 76 |
|
action, |
| 77 |
|
"journey", |
| 78 |
3 |
journey.journey_id, |
| 79 |
|
outcome, |
| 80 |
3 |
actor_id: journey.user_id, |
| 81 |
|
correlation_id: correlation_id, |
| 82 |
3 |
metadata: %{transfer_id: journey.transfer_id, status: journey.status} |
| 83 |
|
) |
| 84 |
|
|
| 85 |
3 |
:telemetry.execute([:wallet_journey, :audit], %{}, audit) |
| 86 |
|
rescue |
| 87 |
:-( |
_ -> :ok |
| 88 |
|
end |
| 89 |
|
end |