Framework-agnostic Elixir CDC consumer for MySQL row-based binary-log replication, delivering committed transactions to a pluggable sink with fail-closed, transaction-granularity delivery: the checkpoint advances only after the sink has durably applied a transaction (ADR-0003). In C1's lib-owned checkpoint mode this is at-least-once — a crash between the sink write and the checkpoint re-delivers the transaction on restart (a bounded duplicate window). Effect-once delivery is the sink-owned atomic path (the sink persisting its write and the position together), which C1 validates but does not yet run — see Capstan.Sink and ADR-0004.

Starting a pipeline

Configure each pipeline explicitly at start_link/1 (no global mutable state):

Capstan.start_link(
  connection: [
    host: "replica.internal", port: 3306, username: "capstan",
    password: "…", database: "orders",
    ssl: true, ssl_opts: [cacertfile: "/etc/ssl/mysql-ca.pem"]  # TLS on by default (ADR-0002)
  ],
  server_id: 1001,                       # replica identity; MUST be unique in the topology
  sink: MyApp.OrdersSink,                # implements Capstan.Sink
  checkpoint_store: [module: MyApp.Store],   # lib-owned mode; see below
  start_position: :checkpoint,           # C1: :checkpoint (default) only — see below
  max_command_retries: 5
)

start_link/1 validates the options through Capstan.Config, enforces the per-mode Capstan.Sink callback required-ness (Capstan.Pipeline), and — for a valid configuration — starts a supervised Capstan.Supervisor wiring a checkpoint store, a Capstan.AssemblerServer, and a Capstan.Connection. A bad substrate is refused before any socket is opened (a bad config fails start-up, not first delivery).

Checkpoint mode

C1 implements lib-owned checkpoint mode. Pass checkpoint_store: [module: impl, options: keyword()], where impl is a Capstan.CheckpointStore implementation exporting start_link/1 (e.g. Capstan.CheckpointStore.InMemory for tests and ephemeral pipelines). The store is supervised as part of the pipeline. Sink-owned checkpoint mode (the sink persisting the position atomically with its own write) is validated but not yet run by C1's AssemblerServer; a sink-owned configuration is refused with :sink_owned_mode_unsupported.

Start position

C1 resumes only from the durable checkpoint (start_position: :checkpoint, the default). A populated store resumes from the persisted watermark. A fresh (never-written) store sends the dump an empty GTID set, which requests the server's full retained history — and a server that has already purged its earliest logs refuses that dump, halting :data_gap. To start "from now" today, seed the checkpoint store with the server's current @@global.gtid_executed before first start (see usage-rules.md). An explicit %Capstan.Position{} override and :current are refused fail-closed (:start_position_override_unsupported / :start_position_current_unsupported) — honoring an explicit resume position end-to-end (the AssemblerServer seeds its watermark from the store alone today) is a C1-completeness follow-up. Refusing avoids a silent checkpoint hole.

Return values

{:ok, supervisor_pid} on success, or {:error, reason} — a value-free atom from Capstan.Config (:server_id_required, :config_invalid, :tls_verification_unspecified), from Capstan.Pipeline (:invalid_sink, :sink_missing_handle_transaction, :sink_missing_checkpoint, :sink_missing_handle_schema_change), or from this module (:sink_owned_mode_unsupported, :checkpoint_store_required, :start_position_override_unsupported, :start_position_current_unsupported).