API to API

This recipe implements a Source over HTTP, calls a destination API from a Process Pipeline, and preserves a Tracking Record for recovery. It applies the same primitives as CSV to JSON files.

The sample payload contains author records. Their fields illustrate the mapping; identity, versioning, tracking, and rollback follow the SDK contracts described in Learn.

The API contract

The endpoints are illustrative. Point the environment variables at APIs you own that implement this contract, or adapt the small transport helper to your provider.

{
  "authors": [
    { "id": "author-1040", "first_name": "Ada", "last_name": "Morgan" },
    { "id": "author-1041", "first_name": "Alex", "last_name": "Chen" }
  ]
}

The destination receives { "id": "imported-author-1040", "name": "Ada Morgan" }. Reserve the imported- namespace for this migration: rollback deletes records in it. This example does not restore preexisting records that a PUT replaced.

Define the source and migration

Install the dependencies from Getting started. Save both files below in the same directory. The first file contains the complete migration config; the second implements HTTP requests, authentication, timeout, and error wrapping.

import { Effect, Schema } from "effect";
import {
  type DestinationError,
  MigrationDefinition,
  MigrationDefinitionRegistry,
  type ProcessPipelineFor,
  type RollbackPipelineFor,
  SourceIdentity,
  Tracking,
} from "migrate-sdk";
import { defineMigrationCliConfig } from "migrate-sdk/cli";
import {
  DocumentFetchers,
  DocumentParsers,
  DocumentSource,
} from "migrate-sdk/sources/document";
import { FileMigrationStore } from "migrate-sdk/stores/file";
import { readJson, writeJson } from "./api-http";

export const AuthorRow = Schema.Struct({
  id: Schema.NonEmptyString,
  first_name: Schema.NonEmptyString,
  last_name: Schema.NonEmptyString,
});
export const AuthorDocument = Schema.Struct({
  authors: Schema.Array(AuthorRow),
});
export const Author = Schema.Struct({
  id: Schema.NonEmptyString,
  name: Schema.NonEmptyString,
});

// Point these at APIs implementing the contract described in the recipe.
export const sourceEndpoint = {
  url: process.env.SOURCE_URL ?? "https://legacy.example.test/authors",
  token: process.env.SOURCE_TOKEN,
};
export const destinationEndpoint = {
  url: process.env.DESTINATION_URL ?? "https://directory.example.test/authors",
  token: process.env.DESTINATION_TOKEN,
};

export const authorSource = DocumentSource.make({
  fetcher: DocumentFetchers.effect({
    cursorSchema: Schema.Null,
    read: () =>
      readJson(sourceEndpoint).pipe(Effect.map((resource) => ({ resource }))),
  }),
  parser: DocumentParsers.schema("authors-api", AuthorDocument),
  selector: { item: (document) => document.authors },
  identity: {
    id: "author-id@v1",
    schema: SourceIdentity.key("authorId", Schema.NonEmptyString),
    key: ({ item }) => item.id,
  },
  version: { kind: "content-hash" },
  lookup: { kind: "scan" },
});

// This namespace must belong exclusively to this migration.
export const destinationId = (sourceId: string) => `imported-${sourceId}`;
export const mapAuthor = (row: typeof AuthorRow.Type) =>
  Schema.decodeEffect(Author)({
    id: destinationId(row.id),
    name: `${row.first_name} ${row.last_name}`,
  });

const authorTracking = Tracking.record({
  id: "author-api@v1",
  schema: Schema.Struct({ authorId: Schema.NonEmptyString }),
});

type AuthorProcess = ProcessPipelineFor<
  typeof authorSource,
  DestinationError | Schema.SchemaError,
  typeof authorTracking
>;
const processAuthor: AuthorProcess = Effect.fn("processApiAuthor")(
  function* (sourceItem) {
    const author = yield* mapAuthor(sourceItem.item.item);
    yield* writeJson(destinationEndpoint, author.id, "PUT", author);
    yield* Tracking.setRecord({ authorId: author.id });
  }
);

const rollbackAuthor: RollbackPipelineFor<
  typeof authorTracking,
  DestinationError | Schema.SchemaError
> = Effect.fn("rollbackApiAuthor")(function* (state) {
  // A timed-out PUT may have succeeded before tracking was saved.
  const sourceId = yield* Schema.decodeUnknownEffect(Schema.NonEmptyString)(
    state.sourceIdentity.key
  );
  const id = state.trackingRecord?.authorId ?? destinationId(sourceId);
  yield* writeJson(destinationEndpoint, id, "DELETE");
});

export const apiAuthors = MigrationDefinition.make({
  id: "api-authors",
  source: authorSource,
  process: processAuthor,
  tracking: authorTracking,
  rollback: rollbackAuthor,
  store: FileMigrationStore.layer({ directory: "./.migrate-api" }),
});

export default defineMigrationCliConfig({
  registry: MigrationDefinitionRegistry.make({ definitions: [apiAuthors] }),
});

DocumentSource selects document.authors, validates their shape, and hashes each selected author for change detection. The fetcher supplies one complete document, so there is no pagination cursor to implement. Schema.Null declares that fetcher contract.

Document selection wraps the selected value in { item }. The process therefore reads sourceItem.item.item: the outer .item is the source payload, and the inner .item is the selected author. The Sources page explains adapter payloads.

Lookup uses a scan. A targeted retry can fetch and inspect the whole document to find one ID. Use a direct lookup or a paginated custom source when the provider and dataset justify it.

Configure and run

Set SOURCE_URL and DESTINATION_URL to the collection URLs. Set SOURCE_TOKEN and DESTINATION_TOKEN if the APIs use bearer authentication. Keep those credentials in your shell or deployment environment.

pnpm exec migrate run api-authors --config ./api-to-api.ts --plan
pnpm exec migrate run api-authors --config ./api-to-api.ts
pnpm exec migrate status api-authors --config ./api-to-api.ts

The plan checks the operation; it does not prove that the API credentials, payload contract, or destination permissions work. Start against a test environment whose imported records you can delete.

The file store saves progress under .migrate-api. Keep that directory between invocations, and keep this config pointed at the same destination. Reusing progress from one environment against another can make the SDK treat records as already migrated when the new destination is empty.

Repeat and recover

Rerun normally after changing an author's source fields. Its content hash changes and the same destination ID receives the new payload. An unchanged migrated author does not need another write.

For an item that failed, fix the endpoint, credentials, or mapping and run:

pnpm exec migrate run api-authors --config ./api-to-api.ts --failed

A timeout does not prove the destination rejected a write. The predictable PUT path allows a repeat to converge on the same record. Rollback first uses the saved tracking ID; for a failed item without a success record it derives the migration-owned ID from the saved source identity. It never needs the source API to look up that author again.

pnpm exec migrate rollback api-authors --config ./api-to-api.ts --plan
pnpm exec migrate rollback api-authors --config ./api-to-api.ts

If a process crashes before any item state is saved, that item may not yet be available to rollback. Resume or reconcile the destination first. Neither an HTTP response nor a tracking call makes the API and store one transaction.

Add resilience deliberately

The transport helper has a ten-second timeout per request and makes no automatic retries. To add bounded backoff, preserve status and transient-failure information at the HTTP boundary and use the policy in Retries and backoff. Do not retry every generic DestinationError: authentication failures and rejected data need different treatment from a 503.

For large datasets, replace the complete-document fetch with a paginated source. Its cursor must be stable enough to resume, and its discovery mode must allow updates to existing authors to be found. The next recipe first adds another source: Combine two APIs.