| 1 |
|
defmodule WalletNotifications.Commands.QueueNotification do |
| 2 |
|
@moduledoc """ |
| 3 |
|
Command to create and queue a notification for async delivery. |
| 4 |
|
|
| 5 |
|
Creates a `Notification` struct, checks idempotency key (replay-safe), |
| 6 |
|
persists to `NotificationStore`, enqueues a `NotificationWorker` job, |
| 7 |
|
and emits `NotificationQueued` event and audit record. |
| 8 |
|
|
| 9 |
|
Returns `{:ok, notification}` on success or idempotency replay. |
| 10 |
|
""" |
| 11 |
|
|
| 12 |
|
alias WalletNotifications.{Notification, NotificationStore, JobQueue} |
| 13 |
|
alias WalletNotifications.Workers.NotificationWorker |
| 14 |
|
alias WalletNotifications.Events.NotificationQueued |
| 15 |
|
alias WalletObservability.AuditEvent |
| 16 |
|
|
| 17 |
|
@spec execute( |
| 18 |
|
user_id :: String.t(), |
| 19 |
|
type :: atom(), |
| 20 |
|
channel :: atom(), |
| 21 |
|
template_id :: String.t(), |
| 22 |
|
opts :: keyword() |
| 23 |
|
) :: {:ok, Notification.t()} | {:error, term()} |
| 24 |
62 |
def execute(user_id, type, channel, template_id, opts \\ []) do |
| 25 |
34 |
correlation_id = Keyword.get(opts, :correlation_id, WalletSharedKernel.Correlation.new_correlation_id()) |
| 26 |
|
|
| 27 |
34 |
notif = Notification.new(user_id, type, channel, template_id, |
| 28 |
|
subject: Keyword.get(opts, :subject), |
| 29 |
|
body: Keyword.get(opts, :body), |
| 30 |
|
recipient_address: Keyword.get(opts, :recipient_address), |
| 31 |
|
idempotency_key: Keyword.get(opts, :idempotency_key), |
| 32 |
|
correlation_id: correlation_id, |
| 33 |
|
max_attempts: Keyword.get(opts, :max_attempts, 3), |
| 34 |
|
payload: Keyword.get(opts, :payload, %{}), |
| 35 |
|
metadata: Keyword.get(opts, :metadata, %{}) |
| 36 |
|
) |
| 37 |
|
|
| 38 |
34 |
case NotificationStore.store(notif) do |
| 39 |
1 |
{:ok, existing} -> |
| 40 |
|
# Idempotency replay — return existing without re-queueing |
| 41 |
|
{:ok, existing} |
| 42 |
|
|
| 43 |
|
:ok -> |
| 44 |
33 |
queue = if type in [:otp, :security_alert, :transaction_alert], |
| 45 |
|
do: :notifications_high, else: :notifications_low |
| 46 |
|
|
| 47 |
33 |
JobQueue.enqueue(queue, NotificationWorker, %{ |
| 48 |
33 |
"notification_id" => notif.notification_id, |
| 49 |
|
"correlation_id" => correlation_id |
| 50 |
|
}) |
| 51 |
|
|
| 52 |
33 |
emit_event(NotificationQueued.build(notif.notification_id, |
| 53 |
|
user_id: user_id, type: type, channel: channel, |
| 54 |
|
correlation_id: correlation_id)) |
| 55 |
|
|
| 56 |
33 |
emit_audit(notif.notification_id, user_id, correlation_id) |
| 57 |
|
{:ok, notif} |
| 58 |
|
|
| 59 |
:-( |
{:error, reason} -> |
| 60 |
|
{:error, reason} |
| 61 |
|
end |
| 62 |
|
rescue |
| 63 |
:-( |
error -> {:error, error} |
| 64 |
|
end |
| 65 |
|
|
| 66 |
33 |
defp emit_event(event) do |
| 67 |
33 |
pubsub = Application.get_env(:wallet_notifications, :pubsub, WalletWeb.PubSub) |
| 68 |
33 |
apply(Phoenix.PubSub, :broadcast, [pubsub, "wallet_notifications:events", {:domain_event, event}]) |
| 69 |
|
rescue |
| 70 |
:-( |
_ -> :ok |
| 71 |
|
end |
| 72 |
|
|
| 73 |
33 |
defp emit_audit(notif_id, user_id, correlation_id) do |
| 74 |
33 |
audit = AuditEvent.build(:notifications, "notification_queued", "notification", notif_id, |
| 75 |
|
:success, actor_id: user_id, correlation_id: correlation_id, metadata: %{}) |
| 76 |
33 |
:telemetry.execute([:wallet_notifications, :audit], %{}, audit) |
| 77 |
|
rescue |
| 78 |
:-( |
_ -> :ok |
| 79 |
|
end |
| 80 |
|
end |