cover/Elixir.WalletIntegrations.Commands.IngestCallback.html

1 defmodule WalletIntegrations.Commands.IngestCallback do
2 @moduledoc """
3 Ingest an inbound provider webhook/callback.
4
5 Pipeline:
6 1. Hash payload for audit traceability.
7 2. Build and store CallbackRecord (status: :pending).
8 3. Emit CallbackReceived event.
9 4. Verify signature and freshness via CallbackVerifier.
10 5. On verification success: mark :verified, enqueue CallbackWorker.
11 6. On verification failure: mark :rejected, emit CallbackRejected, return {:error, :rejected}.
12 7. Callback dedup is enforced atomically via InboxStore before any side effect.
13
14 Returns `{:ok, CallbackRecord.t()}` on successful ingestion,
15 `{:error, :rejected}` on signature/freshness failure,
16 `{:error, :duplicate}` when inbox already has this message_id.
17 """
18
19 alias WalletIntegrations.{CallbackRecord, CallbackStore, InboxStore, JobQueue}
20 alias WalletIntegrations.{CallbackVerifier, SecretProvider}
21 alias WalletIntegrations.Workers.CallbackWorker
22 alias WalletIntegrations.Events.{CallbackReceived, CallbackRejected}
23 alias WalletIntegrations.AdapterResult
24 alias WalletObservability.AuditEvent
25
26 @spec execute(params :: map(), opts :: keyword()) ::
27 {:ok, CallbackRecord.t()} | {:error, :rejected | :duplicate | term()}
28 22 def execute(params, opts \\ []) do
29 11 correlation_id =
30 Keyword.get(opts, :correlation_id, WalletSharedKernel.Correlation.new_correlation_id())
31
32 11 provider = Map.get(params, :provider, :stripe)
33 11 raw_body = Map.get(params, :raw_body, "")
34 11 signature_header = Map.get(params, :signature_header, "")
35 11 event_type = Map.get(params, :event_type, "unknown")
36 11 message_id = Map.get(params, :message_id, derive_message_id(provider, raw_body))
37
38 11 payload_hash = AdapterResult.hash_response(raw_body)
39
40 # Dedup check via InboxStore before building any domain records
41 11 case InboxStore.claim(message_id, :wallet_integrations, callback_id: nil) do
42 2 {:duplicate, _existing} ->
43 {:error, :duplicate}
44
45 {:ok, _inbox} ->
46 9 callback =
47 CallbackRecord.new(provider, event_type, payload_hash,
48 raw_signature: signature_header,
49 inbox_message_id: message_id,
50 correlation_id: correlation_id,
51 metadata: Map.get(params, :metadata, %{})
52 )
53
54 9 with :ok <- CallbackStore.store(callback) do
55 9 emit_event(
56 9 CallbackReceived.build(callback.callback_id,
57 provider: provider,
58 event_type: event_type,
59 payload_hash: payload_hash,
60 correlation_id: correlation_id
61 )
62 )
63
64 9 case verify_callback(provider, raw_body, signature_header) do
65 :ok ->
66 6 {:ok, verified} = CallbackRecord.verify(callback)
67 6 :ok = CallbackStore.update(verified)
68
69 6 {:ok, _job_id} =
70 JobQueue.enqueue(:integrations_callback, CallbackWorker, %{
71 6 "callback_id" => verified.callback_id,
72 "correlation_id" => correlation_id
73 })
74
75 6 emit_audit(callback.callback_id, correlation_id, :success, %{provider: provider, event_type: event_type})
76 {:ok, verified}
77
78 {:error, reason} ->
79 3 {:ok, rejected} = CallbackRecord.reject(callback, to_string(reason))
80 3 :ok = CallbackStore.update(rejected)
81
82 3 emit_event(
83 3 CallbackRejected.build(callback.callback_id,
84 provider: provider,
85 3 rejection_reason: to_string(reason),
86 correlation_id: correlation_id
87 )
88 )
89
90 3 emit_audit(callback.callback_id, correlation_id, :failure, %{
91 provider: provider,
92 rejection_reason: reason
93 })
94
95 {:error, :rejected}
96 end
97 end
98 end
99 rescue
100 
:-(
 error -> {:error, error}
101 end
102
103 defp verify_callback(:stripe, raw_body, signature_header) do
104 9 case SecretProvider.get(:stripe_webhook_secret) do
105 {:ok, secret} ->
106 4 CallbackVerifier.verify_stripe(raw_body, signature_header, secret)
107
108 {:error, :secret_not_found} ->
109 # In CI/test mode with no secret configured, skip signature verification
110 5 if Application.get_env(:wallet_integrations, :skip_webhook_verification, false) do
111 :ok
112 else
113 {:error, :webhook_secret_not_configured}
114 end
115 end
116 end
117
118 
:-(
 defp verify_callback(_provider, _raw_body, _signature_header), do: :ok
119
120 defp derive_message_id(:stripe, raw_body) do
121 11 hash = :crypto.hash(:sha256, raw_body) |> Base.encode16(case: :lower)
122 11 "stripe_cb:#{hash}"
123 end
124
125 defp derive_message_id(provider, raw_body) do
126 
:-(
 hash = :crypto.hash(:sha256, raw_body) |> Base.encode16(case: :lower)
127 
:-(
 "#{provider}_cb:#{hash}"
128 end
129
130 12 defp emit_event(event) do
131 12 pubsub = Application.get_env(:wallet_integrations, :pubsub, WalletWeb.PubSub)
132 12 apply(Phoenix.PubSub, :broadcast, [pubsub, "wallet_integrations:events", {:domain_event, event}])
133 rescue
134 
:-(
 _ -> :ok
135 end
136
137 9 defp emit_audit(callback_id, correlation_id, outcome, metadata) do
138 9 audit =
139 AuditEvent.build(:integrations, "ingest_callback", "callback", callback_id,
140 outcome,
141 correlation_id: correlation_id,
142 metadata: metadata
143 )
144
145 9 :telemetry.execute([:wallet_integrations, :audit], %{}, audit)
146 rescue
147 
:-(
 _ -> :ok
148 end
149 end
Line Hits Source