Get Started

Basic Usage

This walkthrough shows the package in the shape it is designed to be used today: a plug client, optional flows, optional webhook wiring, and the shared runtime behind /api/khotan.

1. Scaffold the pieces you need

terminal

$ npx khotan add plug --yes
$ npx khotan add cache --yes
$ npx khotan add inflow --yes
$ npx khotan add wire --yes
$ npx khotan add catch --yes
$ npx khotan add pass --yes
$ npx khotan add mapping-browser --yes
$ npx khotan add mappings-page-1 --yes
$ npx khotan add hub --yes
$ npx khotan add runs --yes

You do not need every component in every project. This set is useful because it covers the main khotan model: a plug, a durable cache, a flow, a wire, catch and pass handlers, and UI to inspect the system, including first-class mappings.

2. Define a plug

A plug is a one-way HTTP client from your app to an external API.

// src/khotan/plugs/pollinate.ts
import { plug, bearer } from "@/khotan/plugs/plug";

export const pollinatePlug = plug({
  name: "pollinate",
  baseUrl: "https://api.pollinate.tech",
  auth: bearer(process.env.POLLINATE_TOKEN!),
  vars: [
    { key: "orgId", label: "Organization ID", type: "text" },
  ] as const,
  endpoints: {
    listProducts: {
      method: "GET",
      path: "/products",
    },
  },
});

auth helpers
retry and timeout support
pagination helpers
optional encrypted vars
optional typed endpoint metadata for the debugger and CLI

3. Define a cache

Use a cache when a flow or webhook workflow needs durable state between runs.

// src/khotan/caches/pollinate-products-snapshot.ts
import { cache } from "@/khotan/caches/cache";

export const pollinateProductsSnapshotCache = cache({
  name: "pollinate-products-snapshot",
  scope: {
    plug: "pollinate",
    resource: "products",
    flow: "pollinate-products",
  },
  ttl: "6h",
});

This gives your workflows a named store that you can access with khotanCache(ctx, "pollinate-products-snapshot").

4. Define a flow

Flows are the durable data-movement primitives. Here is a simple inflow:

// src/khotan/flows/pollinate-products.ts
import { khotanCache } from "khotan-data/factory";
import { inflow, type InflowContext } from "@/khotan/flows/inflow";
import { pollinatePlug } from "@/khotan/plugs/pollinate";

async function pollinateProductsWorkflow(ctx: InflowContext) {
  "use workflow";

  async function syncProducts() {
    "use step";
    const snapshotCache = khotanCache(ctx, "pollinate-products-snapshot");
    const previous =
      (await snapshotCache.get<Array<{ id: string; sku?: string }>>("latest")) ?? [];
    const response = await pollinatePlug.get<{ data?: Array<{ id: string; sku?: string }> }>("/products", {
      vars: ctx.vars,
    });
    const records = Array.isArray(response.data) ? response.data : [];

    await snapshotCache.set("latest", records);

    return {
      extracted: records.length,
      transformed: records.length,
      created: records.length,
      metadata: {
        source: ctx.flow.name,
        previousCount: previous.length,
      },
    };
  }

  return syncProducts();
}

export const pollinateProductsInflow = inflow({
  name: "pollinate-products",
  resource: "products",
  schedule: "0 * * * *",
  workflow: pollinateProductsWorkflow,
});

The flow runtime writes runs into khotan_runs, so the same execution can be inspected later in the Hub, Runs table, or CLI, and your workflow can keep durable snapshots or cursors in cache tables.

5. Define a wire and webhook handlers

Use a wire when the source system supports webhook subscription registration. Use catch to process the event in your app. Use pass to forward it to another plug.

// src/khotan/wires/pollinate-wire.ts
import { wire } from "@/khotan/wires/wire";

export const pollinateWire = wire({
  events: ["order.created"],
  async onSubscribe(ctx) {
    const res = await ctx.plug.post<{ id: string; secret: string }>("/webhooks", {
      body: {
        url: ctx.callbackUrl,
        events: ctx.events,
      },
    });

    await ctx.setWireVars({ webhookSecret: res.secret });
    return { remoteId: res.id };
  },
  async onUnsubscribe(ctx) {
    await ctx.plug.delete(`/webhooks/${ctx.remoteId}`);
  },
  async onVerify(ctx) {
    return Boolean(ctx.headers["x-signature"] && ctx.wireVars["webhookSecret"]);
  },
});

// src/khotan/webhooks/pollinate-catch.ts
import { khotanCache } from "khotan-data/factory";
import { catchEvent, type CatchContext } from "@/khotan/webhooks/catch";

async function pollinateCatchWorkflow(ctx: CatchContext) {
  "use workflow";

  async function handleEvent() {
    "use step";
    const dedupeCache = khotanCache(ctx, "pollinate-webhook-markers");
    const eventId = String(ctx.event["id"] ?? "");
    if (eventId && (await dedupeCache.get<boolean>(eventId))) return;

    console.log("Handled event", ctx.eventType, ctx.khotanRunId);

    if (eventId) {
      await dedupeCache.set(eventId, true);
    }
  }

  await handleEvent();
}

export const pollinateCatch = catchEvent({
  name: "pollinate-orders",
  events: ["order.created"],
  workflow: pollinateCatchWorkflow,
});

// src/khotan/webhooks/pollinate-to-slack.ts
import { pass, type PassContext } from "@/khotan/webhooks/pass";

async function pollinateToSlackWorkflow(ctx: PassContext) {
  "use workflow";

  async function forwardEvent() {
    "use step";
    console.log("Forwarding event", ctx.eventType, ctx.destVars);
  }

  await forwardEvent();
}

export const pollinateToSlack = pass({
  name: "pollinate-to-slack",
  to: "slack",
  events: ["order.created"],
  workflow: pollinateToSlackWorkflow,
});

6. Register everything in the factory config

// src/khotan/khotan.ts
import { khotan, drizzleAdapter } from "khotan-data/factory";
import { db } from "@/db";
import { pollinatePlug } from "./plugs/pollinate";
import { slackPlug } from "./plugs/slack";
import { pollinateProductsSnapshotCache } from "./caches/pollinate-products-snapshot";
import { pollinateProductsInflow } from "./flows/pollinate-products";
import { pollinateWire } from "./wires/pollinate-wire";
import { pollinateCatch } from "./webhooks/pollinate-catch";
import { pollinateToSlack } from "./webhooks/pollinate-to-slack";

const khotanData = khotan({
  adapter: drizzleAdapter(db),
  resources: [
    {
      name: "products",
      mapping: {
        connectField: "sku",
        plugs: {
          pollinate: { uniqueIdentifier: "id" },
        },
      },
      description: "Shared product identity",
    },
    {
      name: "customers",
      mapping: {
        connectField: ["tenantId", "email"],
        plugs: {
          pollinate: { uniqueIdentifier: "id" },
          slack: { uniqueIdentifier: "email" },
        },
      },
      description: "Shared customer identity across systems",
    },
  ],
  caches: [
    pollinateProductsSnapshotCache,
  ],
  plugs: [
    {
      name: "pollinate",
      plug: pollinatePlug,
      flows: [pollinateProductsInflow],
      wires: [pollinateWire],
      catches: [pollinateCatch],
      passes: [pollinateToSlack],
    },
    {
      name: "slack",
      plug: slackPlug,
    },
  ],
});

export default khotanData;

On the first real runtime request, khotan upserts the registered plugs, caches, flows, resources, and webhook handlers into its tables.

When a resource declares mapping participants, khotan also uses that contract to validate mapping mutations and to derive one canonical connectValue for lookup and browsing.

7. Operate the runtime

terminal

$ npx khotan plug --list
$ npx khotan plug pollinate --info
$ npx khotan wire pollinate connect
$ npx khotan flows list
$ npx khotan flows trigger pollinate-products --plug pollinate --variant full
$ npx khotan mappings list customers --search "alice@example.com"
$ npx khotan mappings lookup customers --connect-value "alice@example.com"

If you installed the dashboard components, you can also add config-page-1 and visit /config.

Visit /mappings if you scaffolded mappings-page-1. The browser lets you search one resource at a time, edit per-plug refs, and keep contextual metadata separate from identity fields.

8. Useful mental model

a plug talks to an external API
a cache stores durable workflow state between runs
a flow moves data on a schedule or on demand
a wire registers a callback URL with the source service
a catch handles the verified inbound event
a pass forwards a verified inbound event to another plug
a resource defines the shared entity contract used for mappings
a mapping stores one canonical shared identity plus per-plug refs
a run records every durable execution in a consistent table

Next steps

Read /docs/configuration for the factory shape and env vars.
Read /docs/cli for plug, wire, flows, and mappings.
Read /docs/agents if an agent will be helping with khotan work.