# TypeScript SDK

> Create jobs, run workers, define typed work and write workflows from TypeScript, in Node.js, Bun, Deno or the browser.

`@emit.run/sdk` has no dependencies. It uses web-standard APIs (`fetch`, streams, Web Crypto), so it runs in Node.js 20.3 or later, Bun, Deno and the browser. Workers that run [workflows](/docs/concepts/workflows) need Node.js 20.16 or later, or Bun.

```bash
npm install @emit.run/sdk
```

This page covers version 0.2.0.

## The client

```ts
import { Emit } from "@emit.run/sdk";

const emit = new Emit({
  url: process.env.EMIT_URL!,     // your cluster's endpoint, from its Overview page
  token: process.env.EMIT_TOKEN!, // an API token, from its Tokens tab
  space: "orders",                // the space this client works in
});

const billing = emit.space("billing"); // the same client, another space
```

`token` and `space` are required, and a variable that might be `undefined` doesn't compile. Pass `null` to leave one out on purpose:

- `token: null`: no API token. The client can only [follow jobs from a browser](/docs/guides/browser-progress).
- `space: null`: no space. The client only has `admin`, for managing spaces and tokens.

## Jobs

```ts
const { id, outcome } = await emit.jobs.create("resize-image", { imageId: "img_881", width: 800 }, { key: "resize-img_881-800" });

await emit.jobs.createMany([
  { name: "resize-image", payload: { imageId: "img_881", width: 1600 } },
  { name: "resize-image", payload: { imageId: "img_882", width: 800 } },
]);

const job = await emit.jobs.get(id);                    // status, payload, result, progress, events
const page = await emit.jobs.list({ name: "resize-image", status: "failed", limit: 50 });
const next = await emit.jobs.list({ page: page.nextPage });
const lines = await emit.jobs.lines(id, { minLevel: "warn" });
await emit.jobs.cancel(id);                             // needs the admin scope
```

`create` takes the options in [Creating a job](/docs/concepts/jobs#creating-a-job): `key`, `delayMs`, `runAtMs`, `maxAttempts`, `timeoutMs`, `backoffMs`, `backoff` and `callback`. `createMany` sends every job in one request and returns the answers in the same order.

### Waiting for a result

```ts
import { JobFailed } from "@emit.run/sdk";

try {
  const result = await emit.jobs.result(id);
} catch (error) {
  if (error instanceof JobFailed) console.error(`job ${error.id} ${error.status}`, error.result);
  else throw error;
}
```

`result` follows the job until it ends and returns its result. It throws `JobFailed` if the job failed or was cancelled, with the error in `result`. If the connection drops, it reconnects.

### Following a job

```ts
for await (const event of emit.jobs.follow(id, { lines: true })) {
  if (event.progress) console.log(`${event.progress.percent}% ${event.progress.message}`);
  if (event.finished) console.log(event.finished.status, event.finished.result);
}
```

The first event is `now`, the job as it is. Then one arrives per change (`claimed`, `progress`, `event`, `lines`, `retrying`, `released`), and `finished` last. Pass `signal`, an `AbortSignal`, to stop early. `emit.jobs.followToken(id, { expiresIn: "1h" })` makes a token that lets a browser follow the job.

## Workers

```ts
import { Release, RetryAfter, Unrecoverable } from "@emit.run/sdk";

type Resize = { imageId: string; width: number };

const worker = emit.worker<Resize>({ names: ["resize-image"], concurrency: 8 }, async (job) => {
  if (job.payload.width > 10_000) throw new Unrecoverable("wider than 10,000 pixels");

  await job.progress(10, "downloading");
  await job.log("info", "downloaded", { metadata: { bytes: 52_000 } });
  await job.event({ stage: "downloaded" });
  // ...resize, passing job.signal to anything that takes one...
  return { url: `https://images.example.com/${job.payload.imageId}-${job.payload.width}.webp` };
});

process.once("SIGTERM", () => worker.stop());
```

| The handler | The job |
| --- | --- |
| returns a value | completes, with that value as its result |
| throws an error | fails this attempt, and is retried while attempts are left |
| throws `new Unrecoverable(message)` | fails for good |
| throws `new RetryAfter(message, ms)` | fails this attempt, and is retried `ms` from now (at most a day) |
| throws `new Release()` | goes back to the queue at once; the attempt isn't counted |

`job` has `id`, `name`, `payload`, `attempt` (1 for the first try), `signal`, and `progress(percent, message?)`, `log(level, message, extra?)`, `event(data)` and `note(text)`. `job.signal` aborts when the worker loses the job: stop, because nothing more from this attempt is accepted.

A log or progress message, or a note, is at most 4,032 bytes, and an event 4 KiB as JSON: longer throws at the call. An error message is cut to 4 KiB. A result is at most 4 MiB as JSON; a larger one fails the attempt.

### Options

:::fields
:::field names {type="string[]"}
The job names to claim, at most 32. Leave it out to claim any job in the space.
:::
:::field concurrency {type="number" default="1"}
How many jobs the worker holds at once.
:::
:::field leaseMs {type="number"}
The lease to ask for. By default, each job's own timeout. The worker renews it while the handler runs.
:::
:::field waitMs {type="number" default="30000"}
How long one claim waits for work before asking again.
:::
:::field retry {type="{ attempts, backoffMs }" default="{ attempts: 3, backoffMs: 250 }"}
How a report (progress, a log line, an event) that fails on the way, from a network error, a 5xx or a 429, is retried before the handler sees the error. The wait doubles each time. The final report of a job is retried until its lease ends.
:::
:::field onError {type="(error) => void" default="console.error"}
Told of what the worker can't handle itself: a final report that never landed, or the loop ending on an error.
:::
:::

`await worker.stop()` stops claiming, lets the jobs in hand finish, and then throws whatever ended the loop, if anything did. `worker.done` settles when the loop ends: after `stop()`, or with the `EmitError` that ended it, such as a revoked token.

## Defined work

`emit.define` names a kind of work and the schemas of what goes in and what comes out, with any [Standard Schema](https://standardschema.dev) library: Zod, Valibot or ArkType. Every process that defines it and calls `emit.serve()` runs it, and whoever calls `run` gets a typed result, wherever it ran.

```ts
import { z } from "zod";

const resize = emit.define(
  "resize-image",
  {
    input: z.object({ imageId: z.string(), width: z.number() }),
    output: z.object({ url: z.string() }),
  },
  async (input, job) => {
    await job.progress(50);
    return { url: `https://images.example.com/${input.imageId}-${input.width}.webp` };
  },
);

const worker = emit.serve({ concurrency: 8 }); // one worker for everything defined so far

const { url } = await resize.run({ imageId: "img_881", width: 800 });   // queue it, wait for the result
const id = await resize.enqueue({ imageId: "img_882", width: 800 });    // queue it, return its id
const here = await resize.here({ imageId: "img_883", width: 800 });     // run it in this process
```

- `run(input, options?)` queues the job and waits for its result. It throws `JobFailed` if the job fails or is cancelled.
- `enqueue(input, options?)` queues the job and returns its id.
- `here(input, options?)` runs the job in this process, held under a lease (five minutes unless `leaseMs` says). If the process dies, a worker that serves the name takes the next attempt.

The input is checked against its schema before it's queued, and a mismatch throws `SchemaError`. Define everything before calling `serve()`, which takes the same options as `emit.worker` except `names`. Define work without a handler when another service runs it: `run` and `enqueue` work the same, and `serve()` doesn't claim it.

## Workflows

`emit.workflow(name, schemas, body)` defines a workflow, and `emit.serve()` runs it. See [Workflows](/docs/concepts/workflows) for the whole model. The body gets the input and `wf`:

| Call | What it does |
| --- | --- |
| `await step.run(input, options?)` | runs a step and returns its result |
| `step.start(input, options?)` | starts a step and returns a handle to await or `cancel()` |
| `step.enqueue(input, options?)` | starts a step that outlives the run; returns its id |
| `await wf.do(name, fn)` | runs `fn` once, here, and keeps what it returns |
| `await wf.sleep("2h")`, `wf.sleepUntil(date)` | pauses the run |
| `await wf.wait(name, { timeout?, schema? })` | waits for a signal; `undefined` if the timeout passes |
| `await wf.race({ a: ..., b: ... })` | whichever of several started steps, waits or sleeps finishes first |
| `await wf.note(text)` | adds a line to the run's timeline |
| `return wf.continueAsNew(input)` | starts the run over with a fresh history |

A workflow has `run`, `enqueue` and `result(id)` like defined work, plus `signal(target, name, data)` and `cancel(id)`.

## Admin

A token with the `admin` scope manages spaces and tokens:

```ts
const { admin } = new Emit({ url: process.env.EMIT_URL!, token: process.env.EMIT_ADMIN_TOKEN!, space: null });

const spaces = await admin.spaces.list();
const counts = await admin.spaces.counts("orders"); // jobs by status, and by name
const { name, ...settings } = await admin.spaces.get("orders");
await admin.spaces.configure("orders", { ...settings, maxUnfinished: 100_000 });

const created = await admin.tokens.create({ name: "api-server", spaces: ["orders"], scopes: ["create", "read"] });
console.log(created.secret); // shown once
await admin.tokens.revoke(created.id);
```

`configure` replaces the space's settings whole, so read them first, as above. See [Spaces](/docs/concepts/spaces#changing-settings).

## Webhook signatures

`verifySignature(secret, header, body, toleranceMs?)` checks the `Emit-Signature` header of a callback or a wake-up against its raw body. It returns `false` for a bad signature, or one more than five minutes old. See [Checking the signature](/docs/concepts/jobs#checking-the-signature).

## The firehose

The firehose streams every job change in your cluster, filtered, in batches. It has its own address, your cluster's endpoint followed by `/firehose`, and its own token: the firehose token from the cluster's **Tokens** tab, not an API token.

```ts
import { Firehose } from "@emit.run/sdk";

const hose = new Firehose({ url: process.env.EMIT_FIREHOSE_URL!, token: process.env.EMIT_FIREHOSE_TOKEN! });

for await (const batch of hose.watch({ spaces: ["orders"], transition: { to: "failed" } })) {
  for (const change of batch.changes) console.log(change.space, change.id, change.name, change.from, change.to);
}
```

Filter by `spaces`, `names`, `id` and `transition` (`from` and `to` a status, or `made: true` for new jobs). Keep each batch's `cursor` and pass it back as `after` to resume. `batch.gap` is `true` when changes before the batch were no longer available.

## Errors

- `EmitError` is a refusal from emit: `code`, `message`, the HTTP `status`, and `retryAfterMs` when emit asks you to wait. `permanent` is `true` when trying again unchanged won't help: any 4xx but 429.
- `JobFailed` comes from `jobs.result` and `run`: the job's `id`, its `status` (`"failed"` or `"cancelled"`) and its `result`.
- `SchemaError` is a value that doesn't fit a defined work's schema.

The SDK retries by itself in three places: a worker's claims, a worker's reports, and the streams behind `jobs.result` and `run`. Other calls aren't retried. If you retry a `create` yourself, give the job a `key`. See [When emit is busy](/docs/concepts/jobs#when-emit-is-busy).

## Types

Every request and response type is exported: `Job`, `JobEvent`, `Space`, `Token`, `CreateOptions`, `WorkerOptions` and more. Field names are camelCase, times and durations are milliseconds (`createdAtMs`, `timeoutMs`), and statuses are lower-case strings: `"pending"`, `"running"`, `"completed"`, `"failed"`, `"cancelled"`.