Programmatic migration API

The CLI is a thin wrapper. Every command — generate, plan, verify, apply, recover — is a function the engine exports. Embed them in your deployment scripts, your internal dashboards, or your own tooling.

#Generating and applying

28 lines


1import {
2  diffSchemas,
3  classifyRisk,
4  summarizePlan,
5  applyMigrations,
6  rollbackMigrations,
7  listAppliedMigrations,
8} from 'zanith';
9 
10// Compare two graphs → ops:
11const ops = diffSchemas(currentGraph, nextGraph);
12 
13// Score the plan — six levels, twenty reason codes:
14const summary = summarizePlan(ops);
15console.log(summary.maxRisk, summary.reasonCounts);
16 
17// Apply against an adapter:
18const result = await applyMigrations(adapter, ops, {
19  preflight: true,
20  shadow: true,
21  // …see ApplyOptions
22});
23 
24// List what's already applied:
25const applied = await listAppliedMigrations(adapter);
26 
27// Reverse an applied plan:
28await rollbackMigrations(adapter, [appliedId]);

Export	Purpose
`diffSchemas(from, to)`	Structural diff between two `SchemaGraph` instances → `MigrationOp[]`.
`reverseOps(ops)`	Best-effort reversal — returns ops that undo the input.
`classifyRisk(op)`	Risk level + reasons for a single op.
`summarizePlan(ops)`	`PlanRiskSummary` for a whole plan.
`applyMigrations(adapter, ops, opts?)`	Run a plan. Returns `ApplyResult`.
`rollbackMigrations(adapter, ids)`	Reverse previously applied plans.
`listAppliedMigrations(adapter)`	Read the applied-migrations table.

#Per-step audit — listMigrationSteps

Each op of an applied plan writes a row to _zanith_migration_steps. listMigrationSteps reads that table for a given migration id — useful for building history dashboards or debugging a partial failure.

6 lines


1import { listMigrationSteps } from 'zanith';
2 
3const steps = await listMigrationSteps(adapter, '20260504_add_orders_index');
4for (const step of steps) {
5  console.log(step.opKind, step.startedAt, step.elapsedMs, step.outcome);
6}

#MigrationStepRow

10 lines


1interface MigrationStepRow {
2  migrationId: string;
3  stepIndex: number;
4  opKind: string;            // 'addColumn' / 'dropTable' / …
5  riskLevel: number;         // 0–5
6  startedAt: Date;
7  elapsedMs: number;
8  outcome: 'ok' | 'failed' | 'skipped';
9  error?: string;
10}

#Snapshots — serializeSchemaGraph & friends

Each successful migration writes a serialized snapshot of the resulting graph to _zanith_schema_snapshots. The snapshot API lets you read those programmatically — for time-travel debugging, schema drift detection, or regenerating a graph as it existed at a point in time.

17 lines


1import {
2  serializeSchemaGraph,
3  deserializeSchemaGraph,
4  listSnapshots,
5  findSnapshot,
6} from 'zanith';
7 
8// List every snapshot:
9const all = await listSnapshots(adapter);
10// → SchemaSnapshotRow[] sorted by recordedAt DESC
11 
12// Pull one back as a SchemaGraph:
13const row = await findSnapshot(adapter, migrationId);
14const past = deserializeSchemaGraph(row!.serialized);
15 
16// Diff past vs current to see drift:
17const drift = diffSchemas(past, currentGraph);

#SchemaSnapshotRow

6 lines


1interface SchemaSnapshotRow {
2  migrationId: string;
3  phase: SnapshotPhase;          // 'before' | 'after' | 'baseline'
4  recordedAt: Date;
5  serialized: SerializedSchemaGraph;
6}

#Artifacts — recovery surface as data

Soft-drops and archives leave a row in _zanith_artifacts. The artifact API lets you list, inspect, restore, export, and purge them — see Recovery for the workflow.

28 lines


1import {
2  listArtifacts,
3  findArtifact,
4  readArtifactRows,
5  restoreSoftDropColumn,
6  restoreArchiveTable,
7  purgeArtifact,
8  purgeOlderThan,
9  rowsToCsv,
10  rowsToJsonLines,
11} from 'zanith';
12 
13const all = await listArtifacts(adapter);
14const one = await findArtifact(adapter, 'orders_total_softdrop');
15 
16// Stream the rows (returned as an async iterable):
17const rows = await readArtifactRows(adapter, one!.name);
18const csv = rowsToCsv(rows);   // or rowsToJsonLines(rows)
19 
20// Restore a soft-dropped column:
21await restoreSoftDropColumn(adapter, 'orders_total_softdrop');
22 
23// Or restore an archived table:
24await restoreArchiveTable(adapter, 'old_orders_archive_2026_03');
25 
26// Cleanup:
27await purgeArtifact(adapter, 'orders_total_softdrop');
28await purgeOlderThan(adapter, 30 * 24 * 60 * 60 * 1000);

#Shadow verification

11 lines


1import { verifyOnShadow } from 'zanith';
2 
3const report = await verifyOnShadow(adapter, plan, {
4  shadowAdapter,           // disposable adapter on a parallel DB
5  // …other ShadowVerificationOptions
6});
7 
8if (!report.ok) {
9  console.error(report.diff);   // structural diff that survived application
10  process.exit(1);
11}

The narrative version of this step is at Shadow-DB verify; this is the function call that powers it.

#When to use this instead of the CLI

You want…	Reach for
A normal generate / verify / up flow.	The [CLI](/docs/reference/cli). Don't reinvent it.
A custom dashboard that lists pending migrations and per-step audit.	`listAppliedMigrations` + `listMigrationSteps`.
Drift detection in a monitoring pipeline.	`findSnapshot` + `diffSchemas`.
Recovery wired into your admin UI.	`listArtifacts` + `readArtifactRows` + `restore*`.
Embedding migration in a deploy script that's not a shell.	`applyMigrations` directly.