TypeScript SDK

@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 need Node.js 20.16 or later, or Bun.

npm install @emit.run/sdk

This page covers version 0.2.0.

The client

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.
• space: null: no space. The client only has admin, for managing spaces and tokens.

Jobs

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: 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

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

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

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());

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

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 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.

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 for the whole model. The body gets the input and wf:

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:

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.

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.

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.

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.

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".