Write to two destinations

A Process Pipeline can invoke several destination operations for one Source Item. The runtime retains one Migration Item State, while a composite Tracking Record can preserve multiple destination results.

This recipe applies that model to the combined API source: each joined item produces a directory author and a search document.

Authors API ──┐                     ┌─ Directory API
              ├─ enriched author ──┤
Profiles API ─┘                     └─ Search API

One item owns both writes. The important design decision is what happens when the directory accepts a request but the search API fails.

Define the destination contracts

The directory API accepts the author payload from API to API. The search API accepts { id, title, body } at PUT SEARCH_URL/:id. Both support deletion by ID and use the same exclusively migration-owned imported- namespace.

Both PUT operations must safely replace the same record when repeated. Confirm that behavior with your provider; an HTTP method name alone does not establish its idempotency semantics, particularly for triggers, notifications, or other side effects.

Build the process and rollback

Save the four files below together. The first is the runnable config; the others are the source and helpers from the preceding recipes. Configure SOURCE_URL, PROFILES_URL, DESTINATION_URL, and SEARCH_URL, plus their corresponding bearer tokens when needed.

import { Effect, Schema } from "effect";
import {
  DestinationChangeDescriptor,
  type DestinationError,
  MigrationDefinition,
  MigrationDefinitionRegistry,
  type ProcessPipelineFor,
  type RollbackPipelineFor,
  Tracking,
} from "migrate-sdk";
import { defineMigrationCliConfig } from "migrate-sdk/cli";
import { FileMigrationStore } from "migrate-sdk/stores/file";
import { combinedSource } from "./api-combined-source";
import { writeJson } from "./api-http";
import { destinationEndpoint, destinationId, mapAuthor } from "./api-to-api";

const searchEndpoint = {
  url: process.env.SEARCH_URL ?? "https://search.example.test/documents",
  token: process.env.SEARCH_TOKEN,
};
const SearchDocument = Schema.Struct({
  id: Schema.NonEmptyString,
  title: Schema.NonEmptyString,
  body: Schema.String,
});
const tracking = Tracking.record({
  id: "author-two-apis@v1",
  schema: Schema.Struct({
    authorId: Schema.NonEmptyString,
    documentId: Schema.NonEmptyString,
  }),
});
const authorWritten = DestinationChangeDescriptor.make(
  "directory-author-put@v1",
  Schema.Struct({ id: Schema.NonEmptyString })
);
const documentWritten = DestinationChangeDescriptor.make(
  "search-document-put@v1",
  Schema.Struct({ id: Schema.NonEmptyString })
);

type Process = ProcessPipelineFor<
  typeof combinedSource,
  DestinationError | Schema.SchemaError,
  typeof tracking
>;
const processAuthor: Process = Effect.fn("publishAuthor")(
  function* (sourceItem) {
    const { author: row, profile } = sourceItem.item.item;
    const author = yield* mapAuthor(row);
    // Validate both payloads before performing either write.
    const document = yield* Schema.decodeEffect(SearchDocument)({
      id: author.id,
      title: author.name,
      body: profile.bio,
    });
    yield* writeJson(destinationEndpoint, author.id, "PUT", author);
    yield* Tracking.recordChange(authorWritten, { id: author.id });
    yield* writeJson(searchEndpoint, document.id, "PUT", document);
    yield* Tracking.recordChange(documentWritten, { id: document.id });
    yield* Tracking.setRecord({ authorId: author.id, documentId: document.id });
  }
);

const rollbackAuthor: RollbackPipelineFor<
  typeof tracking,
  DestinationError | Schema.SchemaError
> = Effect.fn("unpublishAuthor")(function* (state) {
  const sourceId = yield* Schema.decodeUnknownEffect(Schema.NonEmptyString)(
    state.sourceIdentity.key
  );
  // Known, migration-owned IDs cover a partial write without a success record.
  const fallbackId = destinationId(sourceId);
  yield* writeJson(
    searchEndpoint,
    state.trackingRecord?.documentId ?? fallbackId,
    "DELETE"
  );
  yield* writeJson(
    destinationEndpoint,
    state.trackingRecord?.authorId ?? fallbackId,
    "DELETE"
  );
});

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

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

The mapping validates both payloads before the first destination request. The writes then happen in a clear order: directory first, search second. Each confirmed write records a journal change. Only after both succeed does the process stage its single composite tracking record.

Run and inspect

pnpm exec migrate run published-authors --config ./api-multiple-destinations.ts --plan
pnpm exec migrate run published-authors --config ./api-multiple-destinations.ts
pnpm exec migrate status published-authors --config ./api-multiple-destinations.ts

The store is .migrate-published, separate from the earlier recipe so their progress does not collide. Do not independently operate both recipes against the same destination namespace: locks apply to their own migration definitions and stores, not to every external resource they happen to write.

Recover a partial result

After repairing an item failure, run:

pnpm exec migrate run published-authors --config ./api-multiple-destinations.ts --failed

A failed-item retry executes the process again from its beginning. It does not resume at the last Tracking.recordChange. This recipe rewrites the directory record before retrying search, which is safe only under the stated destination contract.

Journal entries are staged in the process and saved with the item's outcome. They are useful evidence of confirmed writes, but they are not independent durable checkpoints. Existing process journals may be retained on later failed attempts rather than accumulating every attempted write. Read Tracking before using journals as compensation instructions.

Undo both sides

Rollback deletes search first, then the directory. It reads destination IDs from the successful tracking record when present. Without one, it derives the known IDs from the saved source identity. This covers partial attempts that did not produce a final success record, including ambiguous HTTP responses.

The ownership rule matters: deriving an ID is permission to delete only because the namespace belongs exclusively to this migration. To update preexisting records instead, save the previous values or use a provider-specific compensation strategy. Deletion would not restore those originals.

pnpm exec migrate rollback published-authors --config ./api-multiple-destinations.ts --plan
pnpm exec migrate rollback published-authors --config ./api-multiple-destinations.ts

If search deletion succeeds and directory deletion fails, the item remains tracked with rollback failure diagnostics. A later rollback can repeat search deletion (404 is accepted) and retry the directory. No distributed transaction is implied.

Choose a different boundary when needed

One definition is useful when a source item and its two destination results belong to one unit of recovery. Use separate definitions when the destinations have independent lifecycles, permissions, schedules, or operators; use dependencies and reference lookup if one result depends on the other.

For long-running hosted execution, retain these same idempotency and tracking contracts and move orchestration into the Workflow SDK cookbook.