@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

• an Ecto.Multi ready for repo.transaction/1

Builds a multi-step transaction for updating a transaction record.

This function implements the OCC.Processor behavior by constructing an Ecto.Multi that:

1. Retrieves the original transaction created by a creation event.
2. Updates the transaction with the new attributes.
3. Marks the update event as processed.

Parameters

• event: The update event being processed.
• attr: The attributes to apply to the transaction.
• repo: The Ecto repo to use for database operations.

Returns

• An Ecto.Multi struct with named operations for transaction processing.

Adds the step to update the event or handle errors after transaction processing.

This function inspects the results of the previous Multi steps and determines whether to mark the event as processed, revert it to pending, schedule a retry, or move it to the dead letter queue.

Parameters

• multi: The Ecto.Multi built so far.
• event: The event being processed.
• _repo: The Ecto repository (unused).

Returns

• The updated Ecto.Multi with an :command_success or :command_failure step.

Handles the case when OCC retries are exhausted.

Delegates to DoubleEntryLedger.Workers.CommandWorker.TransactionCommandResponseHandler.handle_occ_final_timeout/2.

Parameters

• command_map: The event being processed.
• repo: The Ecto repository.

Returns

• An Ecto.Multi that updates the event as dead letter or timed out.

Handles errors that occur when converting event data to a transaction map.

Delegates to DoubleEntryLedger.Workers.CommandWorker.TransactionCommandResponseHandler.handle_transaction_map_error/3.

Parameters

• command_map: The event being processed.
• error: The error encountered during transaction map conversion.
• repo: The Ecto repository.

Returns

• An Ecto.Multi that updates the event with error information.

@spec process(DoubleEntryLedger.Command.t(), Ecto.Repo.t()) ::
  DoubleEntryLedger.Workers.CommandWorker.success_tuple()
  | DoubleEntryLedger.Workers.CommandWorker.error_tuple()

Processes an update event by modifying the corresponding transaction.

This function serves as the main entry point for processing update events. It utilizes optimistic concurrency control with retry logic to handle potential conflicts when multiple processes attempt to update the same transaction simultaneously.

Parameters

• event: The %Command{} struct with action :update to process.
• repo: The Ecto repo to use for database operations (defaults to Repo).

Returns

• {:ok, transaction, event}: Successfully processed the update event.
• {:error, event}: Failed to process the event, with the event containing error details.
• {:error, changeset}: Failed to update the event status, with changeset containing validation errors.
• {:error, reason}: Failed to process the event, with reason explaining the failure.

Error Handling

• Handles create events that are still pending.
• Handles optimistic concurrency conflicts with retries.
• Properly marks events as failed with meaningful error messages.

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

@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