DoubleEntryLedger.Workers.CommandWorker.CreateTransactionCommandMapNoSaveOnError (double_entry_ledger v0.3.0)

View Source

Processes command maps for creating transactions, returning changesets on error instead of raising or saving invalid data.

This module is used for command ingestion where errors (such as invalid entry data, duplicate source keys, or OCC timeouts) should not result in partial saves or exceptions, but instead return a changeset with error details. It leverages OCC (optimistic concurrency control) and changeset validation to ensure only valid, unique, and fully processed commands are persisted.

Error Handling

  • Returns a changeset with errors for invalid input, duplicate keys, OCC timeouts, or account mismatches.
  • Errors are attached to the :input_command_map or other relevant fields in the changeset.

Summary

Types

logable()

Functions

build_multi(module, occable_item, repo)

Builds the shared Ecto.Multi pipeline for both success and error flows.

error(message, logable, changeset)
info(message, logable, schema)
process(command_map, repo \\ Repo)

Processes a command map for transaction creation, returning a changeset on error.

process_with_retry(occable_item, repo \\ Repo)

Processes a command with OCC retry logic.

retry(module, occable_item, error_map, attempts, repo)

Finalizes when retry attempts are exhausted.

warn(message, logable)
warn(message, logable, changeset)

Types

logable()

@type logable() ::
  DoubleEntryLedger.Command.t()
  | DoubleEntryLedger.Command.AccountCommandMap.t()
  | DoubleEntryLedger.Command.TransactionCommandMap.t()
  | map()

Functions

build_multi(module, occable_item, repo)

@spec build_multi(module(), DoubleEntryLedger.Occ.Occable.t(), Ecto.Repo.t()) ::
  Ecto.Multi.t()

Builds the shared Ecto.Multi pipeline for both success and error flows.

  1. :transaction_map – converts raw data to a map or returns an error tuple
  2. merges in either:
    • handle_transaction_map_error/3 when conversion fails
    • build_transaction/4 + handle_build_transaction/3 on success

Parameters

  • module - the processor module implementing the callbacks
  • occable_item - the Command or TransactionCommandMap being processed
  • repo - the Ecto repo to use for DB ops

Returns

error(message, logable, changeset)

@spec error(String.t(), logable(), any()) :: {:ok, String.t()}

info(message, logable, schema)

@spec info(String.t(), logable(), any()) :: {:ok, String.t()}

process(command_map, repo \\ Repo)

@spec process(
  DoubleEntryLedger.Command.TransactionCommandMap.t(),
  Ecto.Repo.t() | nil
) ::
  DoubleEntryLedger.Workers.CommandWorker.success_tuple()
  | {:errors,
     Ecto.Changeset.t(DoubleEntryLedger.Command.TransactionCommandMap.t())
     | String.t()}

Processes a command map for transaction creation, returning a changeset on error.

This function attempts to process the command map using OCC and entry validation. If the command is invalid (e.g., invalid entry data, duplicate source key, OCC timeout, or account mismatch), it returns an Ecto.Changeset with error details attached to the :input_command_map or other relevant fields. No partial data is saved on error.

Returns

  • {:ok, transaction, command} on success
  • {:error, changeset} if validation or OCC fails (see changeset errors for details)
  • {:error, string} for unexpected errors

process_with_retry(occable_item, repo \\ Repo)

Processes a command with OCC retry logic.

Converts command data to a transaction map, builds an Ecto.Multi, and retries on Ecto.StaleEntryError up to the configured maximum.

Parameters

  • occable_item: The command or command map to process
  • repo: The Ecto repository (defaults to Repo)

Returns

  • {:ok, %{transaction: Transaction.t(), command_success: Command.t()}} on success
  • {:ok, %{command_failure: Command.t()}} on failure
  • Ecto.Multi.failure() on unrecoverable error

retry(module, occable_item, error_map, attempts, repo)

@spec retry(
  module(),
  DoubleEntryLedger.Occ.Occable.t(),
  DoubleEntryLedger.Command.ErrorMap.t(),
  non_neg_integer(),
  Ecto.Repo.t()
) ::
  {:ok,
   %{
     transaction: DoubleEntryLedger.Transaction.t(),
     command_success: DoubleEntryLedger.Command.t()
   }}
  | {:ok, %{command_failure: DoubleEntryLedger.Command.t()}}
  | Ecto.Multi.failure()

Finalizes when retry attempts are exhausted.

Invokes Occable.timed_out/3 to mark the item as timed out, merges in handle_occ_final_timeout/2, then runs one last transaction.

Parameters

  • module - the processor module
  • occable_item - the item that timed out
  • error_map - the accumulated errors and retry count
  • repo - the Ecto repo

Returns

  • The result of the final repo.transaction/1

warn(message, logable)

@spec warn(String.t(), logable()) :: {:ok, String.t()}

warn(message, logable, changeset)

@spec warn(String.t(), logable(), any()) :: {:ok, String.t()}