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

Types

handler()
start_opts()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

pause(access_token, reason \\ :manual)

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

resume(access_token)

Resumes a paused sync worker and triggers an immediate sync.

start_link(arg)
start_worker(access_token, config, opts)

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

status(access_token)

Returns the current status of the sync worker.

stop_worker(access_token)

Stops the sync worker for the given access token.

trigger_sync(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.