Sources

A Source is a pluggable capability that emits source items from a source system. It defines how data is discovered and addressed, and supplies the schemas and metadata the runtime needs before processing can begin.

A migration definition connects one configured source to its processing pipeline and migration store. The source owns reading and identifying data. The pipeline owns transformation and destination effects. The store owns durable progress.

What a source supplies

Every configured source supplies three contracts alongside its read operations:

Identity and version are established before processing. They let the runtime locate previous state and record an item failure even if its payload cannot be decoded. The payload schema may change representations: CSV text can decode into numbers or trimmed strings. A TypeScript annotation alone does not validate those incoming values.

The schema is also the source of the process-facing payload type. See Schemas and mapping for the boundary between decoding source data and constructing destination data, and Items for identity and version semantics.

Discovery and identity lookup

A source exposes two access paths to the same item contract:

Normal runs use discovery for all item states. Failed-item retries, skipped-item retries, and explicit identity targets use lookup. A scan-based lookup can read the entire source to recover one item, so targeted work is not necessarily a cheap source operation.

Lookup receives the structured identity.key; source implementations should not parse the durable string themselves. Both access paths must agree about identity, version, and payload shape. Returning a different identity from the one requested is a source error.

Cursor windows and checkpoints

A Source Cursor is a position marker. A Source Cursor Window is the batch of items returned by one cursor read. The source determines their shape, ordering, and size; a run's attempt limit does not define the source's page size.

A read can return a nextCursor. Omitting it ends discovery after the returned items settle. The runtime encodes a returned cursor using the cursor schema and saves it in the migration store only after the whole window settles, including individual failed outcomes. A partially processed window keeps its input checkpoint and can be read again.

A cursor therefore describes discovery position, not proof that every earlier item succeeded. Source versions determine whether encountered items need processing; cursors determine which items are encountered. An item behind a saved checkpoint requires identity lookup or a rescan to be revisited before a later full traversal.

Source Discovery defines the meaning of a checkpoint after traversal completes:

Incremental discovery requires every new or changed item to sort after the saved checkpoint. The source owns that guarantee, including stable ordering and tie-breakers. A cursor based only on immutable IDs cannot discover updates to earlier IDs incrementally.

A Source Rescan restarts discovery while preserving ordinary item eligibility. An Update Run restarts discovery and makes previously migrated items eligible even when their versions match. Neither makes destination effects transactional. See States and Retries.

Configured sources and adapter responsibilities

Migration authors normally use an adapter factory. The configured value carries its contracts and an Effect layer that the runtime acquires for that definition.

A Document Source composes a fetcher that retrieves resources, a parser that decodes documents, and a selector that chooses items with optional parent context. Its payload is { item } or { parent, item }, so processing accesses the selected value through sourceItem.item.item. Destination mapping remains a pipeline responsibility.

SQL callbacks return statements; the adapter executes them and normalizes SQL failures. SQL text does not infer a validated row type. The query must apply the requested limit and ordering compatible with its cursor. CSV manages its cursor internally, but bounded output windows do not make its file parser streaming.

For a new adapter, use Source.make for runtime methods with their dependencies already provided. Its factory form creates fresh implementation state. Use Source.fromLayer for setup, services, or scoped resources; source.provide(layer) binds dependencies into the configured source.

Failure boundaries

Runtime reads and lookups use the SourceError failure channel. A cursor read failure stops that definition's discovery because no reliable item set was returned. Payload validation after a valid identity and version can produce an individual failed state. Identity lookup failures can also become item failures when existing state identifies the target. These boundaries are separate from request retry policy.

Implementation examples

Use CSV to JSON for a built-in adapter, API to API for document selection, or Combine two APIs for a custom fetcher.

import { PgClient } from "@effect/sql-pg";
import { Config, Schema } from "effect";
import {
  SqlIdentity,
  SqlSource,
  type SqlSourceRead,
} from "migrate-sdk/sources/sql";

const sourceDatabase = PgClient.layerConfig({
  url: Config.Redacted("SOURCE_DATABASE_URL"),
});

const AuthorRow = Schema.Struct({
  id: Schema.NonEmptyString,
  first_name: Schema.NonEmptyString,
  last_name: Schema.NonEmptyString,
  revision: Schema.NonEmptyString,
});
const AuthorCursor = Schema.Struct({ id: Schema.NonEmptyString });

const readAuthors: SqlSourceRead<
  typeof AuthorRow.Encoded,
  typeof AuthorCursor.Type
> = (sql, cursor, limit) =>
  cursor === null
    ? sql`
        SELECT id, first_name, last_name, revision
        FROM source_authors
        ORDER BY id
        LIMIT ${limit}
      `
    : sql`
        SELECT id, first_name, last_name, revision
        FROM source_authors
        WHERE id > ${cursor.id}
        ORDER BY id
        LIMIT ${limit}
      `;

export const authorSource = SqlSource.make({
  sourceSchema: AuthorRow,
  identity: SqlIdentity.columns({
    id: "author-id@v1",
    columns: [SqlIdentity.column("id", Schema.NonEmptyString)],
  }),
  cursorSchema: AuthorCursor,
  batchSize: 100,
  read: readAuthors,
  lookup: (sql, identity) => sql`
    SELECT id, first_name, last_name, revision
    FROM source_authors
    WHERE id = ${identity.key}
  `,
  getSourceMetadata: (row) => ({
    kind: "success",
    cursor: { id: row.id },
    version: row.revision,
  }),
}).provide(sourceDatabase);