Broadway pipeline for high-throughput transaction synchronization.

Use this instead of TransactionSync when you need:

• Parallel processing of transactions from multiple items
• Backpressure-aware ingestion
• Batch processing with configurable batch sizes
• Integration with Broadway's observability

Architecture

Messages flow:

1. SyncProducer — GenStage producer fetching Plaid sync pages
2. Broadway concurrency layer — parallel processing
3. Your handle_message/3 implementation — business logic

Setup

defmodule MyApp.TransactionPipeline do
  use PlaidEx.Sync.BroadwayPipeline

  @impl true
  def handle_transaction(%PlaidEx.Schemas.Transaction{} = tx, _context) do
    MyApp.Transactions.upsert(tx)
  end

  @impl true
  def handle_removed(transaction_id, _context) do
    MyApp.Transactions.delete(transaction_id)
  end
end

Starting

{:ok, _pid} = MyApp.TransactionPipeline.start_link(
  access_tokens: ["access-sandbox-abc", "access-sandbox-xyz"],
  config: plaid_config(),
  concurrency: 10,
  batch_size: 100
)

Requirements

Requires :broadway and :gen_stage in your deps.