| 1 |
|
defmodule WalletIntegrations.Workers.CallbackWorker do |
| 2 |
|
@moduledoc """ |
| 3 |
|
Async worker for processing verified inbound provider callbacks. |
| 4 |
|
|
| 5 |
|
Implements `perform/1` compatible with `WalletIntegrations.JobQueue` and Oban. |
| 6 |
|
|
| 7 |
|
Processing pipeline (per ADR 0002 inbox pattern): |
| 8 |
|
1. Load CallbackRecord (must be in :verified status). |
| 9 |
|
2. Transition to :processing. |
| 10 |
|
3. Check InboxStore — if already processed, treat as duplicate and return :ok. |
| 11 |
|
4. Update PaymentRequest based on event_type → PaymentRequest status update. |
| 12 |
|
5. Mark InboxRecord as :processed. |
| 13 |
|
6. Mark CallbackRecord as :processed. |
| 14 |
|
7. Emit CallbackProcessed event. |
| 15 |
|
8. Emit audit. |
| 16 |
|
|
| 17 |
|
All steps after step 3 are idempotent. A duplicate callback returns :ok without |
| 18 |
|
re-applying any side effect. |
| 19 |
|
""" |
| 20 |
|
|
| 21 |
|
alias WalletIntegrations.{CallbackRecord, CallbackStore} |
| 22 |
|
alias WalletIntegrations.{InboxRecord, InboxStore} |
| 23 |
|
alias WalletIntegrations.{PaymentRequest, PaymentRequestStore} |
| 24 |
|
alias WalletIntegrations.Events.CallbackProcessed |
| 25 |
|
alias WalletObservability.AuditEvent |
| 26 |
|
|
| 27 |
|
@spec perform(args :: map()) :: :ok | {:error, term()} |
| 28 |
|
def perform(%{"callback_id" => callback_id} = args) do |
| 29 |
1 |
correlation_id = |
| 30 |
|
Map.get(args, "correlation_id", WalletSharedKernel.Correlation.new_correlation_id()) |
| 31 |
|
|
| 32 |
1 |
with {:ok, callback} <- CallbackStore.get(callback_id), |
| 33 |
1 |
{:ok, processing} <- CallbackRecord.begin_processing(callback), |
| 34 |
1 |
:ok <- CallbackStore.update(processing) do |
| 35 |
1 |
process_callback(processing, correlation_id) |
| 36 |
|
end |
| 37 |
|
end |
| 38 |
|
|
| 39 |
:-( |
def perform(_args), do: {:error, :missing_callback_id} |
| 40 |
|
|
| 41 |
|
defp process_callback(callback, correlation_id) do |
| 42 |
|
# Check for duplicate via inbox |
| 43 |
1 |
message_id = callback.inbox_message_id |
| 44 |
|
|
| 45 |
1 |
if message_id do |
| 46 |
1 |
case InboxStore.get(message_id) do |
| 47 |
:-( |
{:ok, %InboxRecord{status: :processed}} -> |
| 48 |
|
# Already processed; idempotent noop |
| 49 |
|
:ok |
| 50 |
|
|
| 51 |
|
{:ok, inbox} -> |
| 52 |
1 |
{:ok, processing_inbox} = InboxRecord.begin_processing(inbox) |
| 53 |
1 |
InboxStore.update(processing_inbox) |
| 54 |
1 |
apply_callback_effects(callback, processing_inbox, correlation_id) |
| 55 |
|
|
| 56 |
|
{:error, :not_found} -> |
| 57 |
:-( |
apply_callback_effects(callback, nil, correlation_id) |
| 58 |
|
end |
| 59 |
|
else |
| 60 |
:-( |
apply_callback_effects(callback, nil, correlation_id) |
| 61 |
|
end |
| 62 |
|
end |
| 63 |
|
|
| 64 |
1 |
defp apply_callback_effects(callback, inbox_record, correlation_id) do |
| 65 |
|
# Find the associated PaymentRequest via provider_reference from metadata |
| 66 |
1 |
provider_ref = Map.get(callback.metadata, :provider_reference) || |
| 67 |
1 |
Map.get(callback.metadata, "provider_reference") |
| 68 |
|
|
| 69 |
1 |
payment_request_id = |
| 70 |
1 |
case provider_ref && PaymentRequestStore.get_by_provider_ref(provider_ref) do |
| 71 |
:-( |
{:ok, req} -> req.request_id |
| 72 |
1 |
_ -> nil |
| 73 |
|
end |
| 74 |
|
|
| 75 |
|
# Apply status update based on event_type |
| 76 |
1 |
if payment_request_id do |
| 77 |
:-( |
update_payment_request_from_callback(provider_ref, callback.event_type, correlation_id) |
| 78 |
|
end |
| 79 |
|
|
| 80 |
|
# Mark inbox as processed |
| 81 |
1 |
if inbox_record do |
| 82 |
1 |
{:ok, processed_inbox} = InboxRecord.mark_processed(inbox_record) |
| 83 |
1 |
InboxStore.update(processed_inbox) |
| 84 |
|
end |
| 85 |
|
|
| 86 |
|
# Mark callback as processed |
| 87 |
1 |
{:ok, processed_callback} = CallbackRecord.mark_processed(callback) |
| 88 |
1 |
CallbackStore.update(processed_callback) |
| 89 |
|
|
| 90 |
|
# Emit event |
| 91 |
1 |
emit_event( |
| 92 |
1 |
CallbackProcessed.build(callback.callback_id, |
| 93 |
|
payment_request_id: payment_request_id, |
| 94 |
1 |
event_type: callback.event_type, |
| 95 |
1 |
new_status: derive_new_status(callback.event_type), |
| 96 |
|
correlation_id: correlation_id |
| 97 |
|
) |
| 98 |
|
) |
| 99 |
|
|
| 100 |
1 |
emit_audit("callback_processed", callback.callback_id, correlation_id, :success, %{ |
| 101 |
1 |
event_type: callback.event_type, |
| 102 |
|
payment_request_id: payment_request_id |
| 103 |
|
}) |
| 104 |
|
|
| 105 |
|
:ok |
| 106 |
|
rescue |
| 107 |
:-( |
e -> |
| 108 |
:-( |
{:ok, failed} = CallbackRecord.mark_failed(callback, inspect(e)) |
| 109 |
:-( |
CallbackStore.update(failed) |
| 110 |
|
{:error, e} |
| 111 |
|
end |
| 112 |
|
|
| 113 |
|
defp update_payment_request_from_callback(provider_ref, event_type, correlation_id) do |
| 114 |
:-( |
case PaymentRequestStore.get_by_provider_ref(provider_ref) do |
| 115 |
|
{:ok, request} -> |
| 116 |
:-( |
updated = |
| 117 |
|
case normalize_event_status(event_type) do |
| 118 |
|
:completed -> |
| 119 |
:-( |
case PaymentRequest.complete(request, provider_status_code: event_type) do |
| 120 |
:-( |
{:ok, r} -> r |
| 121 |
:-( |
_ -> request |
| 122 |
|
end |
| 123 |
|
|
| 124 |
|
:failed -> |
| 125 |
:-( |
case PaymentRequest.fail(request, event_type) do |
| 126 |
:-( |
{:ok, r} -> r |
| 127 |
:-( |
_ -> request |
| 128 |
|
end |
| 129 |
|
|
| 130 |
|
:canceled -> |
| 131 |
:-( |
case PaymentRequest.cancel(request) do |
| 132 |
:-( |
{:ok, r} -> r |
| 133 |
:-( |
_ -> request |
| 134 |
|
end |
| 135 |
|
|
| 136 |
|
_ -> |
| 137 |
:-( |
request |
| 138 |
|
end |
| 139 |
|
|
| 140 |
:-( |
PaymentRequestStore.update(updated) |
| 141 |
:-( |
emit_audit("callback_applied", request.request_id, correlation_id, :success, %{event_type: event_type}) |
| 142 |
|
|
| 143 |
:-( |
_ -> |
| 144 |
|
:ok |
| 145 |
|
end |
| 146 |
|
end |
| 147 |
|
|
| 148 |
|
# Map Stripe event type strings to normalized payment status atoms |
| 149 |
1 |
defp normalize_event_status("payment_intent.succeeded"), do: :completed |
| 150 |
:-( |
defp normalize_event_status("payment_intent.payment_failed"), do: :failed |
| 151 |
:-( |
defp normalize_event_status("payment_intent.canceled"), do: :canceled |
| 152 |
:-( |
defp normalize_event_status("charge.succeeded"), do: :completed |
| 153 |
:-( |
defp normalize_event_status("charge.failed"), do: :failed |
| 154 |
:-( |
defp normalize_event_status("charge.refunded"), do: :completed |
| 155 |
:-( |
defp normalize_event_status(_event_type), do: :unknown |
| 156 |
|
|
| 157 |
|
defp derive_new_status(event_type) do |
| 158 |
1 |
case normalize_event_status(event_type) do |
| 159 |
1 |
:completed -> :completed |
| 160 |
:-( |
:failed -> :failed |
| 161 |
:-( |
:canceled -> :canceled |
| 162 |
:-( |
_ -> :unknown |
| 163 |
|
end |
| 164 |
|
end |
| 165 |
|
|
| 166 |
1 |
defp emit_event(event) do |
| 167 |
1 |
pubsub = Application.get_env(:wallet_integrations, :pubsub, WalletWeb.PubSub) |
| 168 |
1 |
apply(Phoenix.PubSub, :broadcast, [pubsub, "wallet_integrations:events", {:domain_event, event}]) |
| 169 |
|
rescue |
| 170 |
:-( |
_ -> :ok |
| 171 |
|
end |
| 172 |
|
|
| 173 |
1 |
defp emit_audit(action, aggregate_id, correlation_id, outcome, metadata) do |
| 174 |
1 |
audit = |
| 175 |
|
AuditEvent.build(:integrations, action, "callback", aggregate_id, |
| 176 |
|
outcome, |
| 177 |
|
correlation_id: correlation_id, |
| 178 |
|
metadata: metadata |
| 179 |
|
) |
| 180 |
|
|
| 181 |
1 |
:telemetry.execute([:wallet_integrations, :audit], %{}, audit) |
| 182 |
|
rescue |
| 183 |
:-( |
_ -> :ok |
| 184 |
|
end |
| 185 |
|
end |