| 1 |
|
defmodule WalletIntegrations.QueueConfig do |
| 2 |
|
@moduledoc """ |
| 3 |
|
Queue topology and configuration for wallet_integrations async workers. |
| 4 |
|
|
| 5 |
|
Queues: |
| 6 |
|
- `:integrations_payment` — outbound payment dispatch (high priority, bounded retries). |
| 7 |
|
- `:integrations_callback` — inbound callback processing (high priority, idempotent). |
| 8 |
|
- `:integrations_status` — provider status polling (normal priority). |
| 9 |
|
- `:integrations_refund` — refund dispatch (normal priority). |
| 10 |
|
|
| 11 |
|
Retry backoff follows exponential policy with jitter as per ADR 0008. |
| 12 |
|
""" |
| 13 |
|
|
| 14 |
|
@queues %{ |
| 15 |
|
integrations_payment: %{priority: 1, max_concurrency: 10, max_attempts: 4}, |
| 16 |
|
integrations_callback: %{priority: 1, max_concurrency: 20, max_attempts: 3}, |
| 17 |
|
integrations_status: %{priority: 2, max_concurrency: 10, max_attempts: 5}, |
| 18 |
|
integrations_refund: %{priority: 2, max_concurrency: 5, max_attempts: 3} |
| 19 |
|
} |
| 20 |
|
|
| 21 |
|
@spec queues() :: map() |
| 22 |
:-( |
def queues, do: @queues |
| 23 |
|
|
| 24 |
|
@spec queue_config(queue :: atom()) :: map() | nil |
| 25 |
:-( |
def queue_config(queue), do: Map.get(@queues, queue) |
| 26 |
|
|
| 27 |
|
@spec max_attempts(queue :: atom()) :: pos_integer() |
| 28 |
|
def max_attempts(queue) do |
| 29 |
30 |
case Map.get(@queues, queue) do |
| 30 |
30 |
%{max_attempts: n} -> n |
| 31 |
:-( |
nil -> 3 |
| 32 |
|
end |
| 33 |
|
end |
| 34 |
|
|
| 35 |
|
@doc """ |
| 36 |
|
Exponential backoff in milliseconds with jitter. |
| 37 |
|
Base: 2_000 ms; max: 60_000 ms. |
| 38 |
|
""" |
| 39 |
|
@spec backoff_ms(attempt :: non_neg_integer()) :: pos_integer() |
| 40 |
|
def backoff_ms(attempt) do |
| 41 |
:-( |
base = 2_000 |
| 42 |
:-( |
cap = 60_000 |
| 43 |
:-( |
raw = min(base * :math.pow(2, attempt) |> trunc(), cap) |
| 44 |
:-( |
jitter = :rand.uniform(div(raw, 4) + 1) |
| 45 |
:-( |
raw + jitter |
| 46 |
|
end |
| 47 |
|
end |