cover/Elixir.WalletSettlement.Commands.RunReconciliation.html

1 defmodule WalletSettlement.Commands.RunReconciliation do
2 @moduledoc """
3 Command to create and enqueue a reconciliation run for async execution.
4
5 Creates a pending `ReconciliationRun`, persists it, then enqueues a
6 `ReconciliationWorker` job on the `:settlement_normal` queue.
7
8 Returns `{:ok, run}` on success or `{:error, reason}`.
9 """
10
11 alias WalletSettlement.{ReconciliationRun, ReconciliationStore, JobQueue}
12 alias WalletSettlement.Workers.ReconciliationWorker
13 alias WalletObservability.AuditEvent
14
15 @spec execute(
16 window_start :: DateTime.t(),
17 window_end :: DateTime.t(),
18 opts :: keyword()
19 ) :: {:ok, ReconciliationRun.t()} | {:error, term()}
20 10 def execute(window_start, window_end, opts \\ []) do
21 5 correlation_id = Keyword.get(opts, :correlation_id, WalletSharedKernel.Correlation.new_correlation_id())
22
23 5 run = ReconciliationRun.new(window_start, window_end,
24 correlation_id: correlation_id,
25 metadata: Keyword.get(opts, :metadata, %{})
26 )
27
28 5 with :ok <- ReconciliationStore.store(run),
29 5 {:ok, _job_id} <- JobQueue.enqueue(:settlement_normal, ReconciliationWorker, %{
30 5 "reconciliation_id" => run.reconciliation_id,
31 "correlation_id" => correlation_id
32 }) do
33 5 emit_audit(run.reconciliation_id, correlation_id)
34 {:ok, run}
35 end
36 rescue
37
:-(
error -> {:error, error}
38 end
39
40 5 defp emit_audit(rec_id, correlation_id) do
41 5 audit = AuditEvent.build(:settlement, "reconciliation_queued", "reconciliation_run", rec_id,
42 :success, correlation_id: correlation_id, metadata: %{})
43 5 :telemetry.execute([:wallet_settlement, :audit], %{}, audit)
44 rescue
45
:-(
_ -> :ok
46 end
47 end
Line Hits Source