PlaidEx.Sync.TransactionSync (plaid_ex v1.0.0)

Copy Markdown View Source

Durable cursor-based transaction synchronization worker.

Implements Plaid's /transactions/sync endpoint semantics correctly, handling all documented edge cases:

  • Cursor durability: cursor is committed BEFORE handler invocation. On crash, the same page is re-delivered to the handler (which must be idempotent), ensuring no data loss.

  • Mutation during pagination: if Plaid returns TRANSACTIONS_SYNC_MUTATION_DURING_PAGINATION, the cursor is reset and a fresh sync begins. This is a documented Plaid behavior.

  • Reauthentication: ITEM_LOGIN_REQUIRED pauses the worker and emits a telemetry event so your application can redirect the user through Link update mode.

  • Institution outages: transient errors (INSTITUTION_DOWN, INSTITUTION_NOT_RESPONDING) are retried with exponential backoff. The worker does not crash — it schedules recovery automatically.

  • Full pagination: has_more: true drives immediate consecutive page fetches with no sleep between them. Only when has_more: false does the worker sleep until the next poll interval.

Starting a sync worker

{:ok, _pid} = PlaidEx.Sync.TransactionSync.start_worker(
  "access-sandbox-abc123",
  config,
  handler: fn page ->
    # page has :added, :modified, :removed lists
    MyApp.Transactions.upsert_batch(page.added)
    MyApp.Transactions.update_batch(page.modified)
    MyApp.Transactions.remove_batch(page.removed)
    :ok
  end,
  tenant_id: "acme_corp"
)

Handler contract

The handler function MUST:

  • Accept a PlaidEx.Schemas.TransactionSyncPage struct
  • Return :ok on success
  • Return {:error, reason} on failure (triggers retry of same page)
  • Be idempotent — it may be called multiple times for the same page (if the worker crashes between cursor commit and handler return)

Manual control

PlaidEx.Sync.TransactionSync.trigger_sync("access-sandbox-...")
PlaidEx.Sync.TransactionSync.pause("access-sandbox-...")
PlaidEx.Sync.TransactionSync.resume("access-sandbox-...")
PlaidEx.Sync.TransactionSync.stop_worker("access-sandbox-...")

Summary

Functions

Returns a specification to start this module under a supervisor.

Pauses the sync worker. It will not poll until resume/1 is called.

Resumes a paused sync worker and triggers an immediate sync.

Starts a new transaction sync worker for the given access token.

Returns the current status of the sync worker.

Stops the sync worker for the given access token.

Manually triggers an immediate sync cycle. Useful after a SYNC_UPDATES_AVAILABLE webhook.

Types

handler()

@type handler() :: (PlaidEx.Schemas.TransactionSyncPage.t() -> :ok | {:error, term()})

start_opts()

@type start_opts() :: [
  handler: handler(),
  tenant_id: String.t() | nil,
  poll_interval_ms: pos_integer()
]

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

pause(access_token, reason \\ :manual)

@spec pause(String.t(), atom()) :: :ok | {:error, :not_found}

Pauses the sync worker. It will not poll until resume/1 is called.

resume(access_token)

@spec resume(String.t()) :: :ok | {:error, :not_found}

Resumes a paused sync worker and triggers an immediate sync.

start_link(arg)

@spec start_link({String.t(), PlaidEx.Config.t(), keyword()}) :: GenServer.on_start()

start_worker(access_token, config, opts)

@spec start_worker(String.t(), PlaidEx.Config.t(), start_opts()) ::
  {:ok, pid()} | {:error, :already_started | term()}

Starts a new transaction sync worker for the given access token.

Workers are registered by access token in PlaidEx.SyncRegistry. Only one worker per access token is allowed.

status(access_token)

@spec status(String.t()) :: {:ok, map()} | {:error, :not_found}

Returns the current status of the sync worker.

stop_worker(access_token)

@spec stop_worker(String.t()) :: :ok | {:error, :not_found}

Stops the sync worker for the given access token.

trigger_sync(access_token)

@spec trigger_sync(String.t()) :: :ok | {:error, :not_found}

Manually triggers an immediate sync cycle. Useful after a SYNC_UPDATES_AVAILABLE webhook.