# Go SDK

> Create jobs, run workers, define typed work and write workflows from Go.

The Go SDK uses only the standard library and needs Go 1.23 or later.

```bash
go get github.com/emit-run/sdk-go
```

This page covers version 0.2.0.

## The client

```go
import emit "github.com/emit-run/sdk-go"

c := emit.New(emit.Options{
	URL:   os.Getenv("EMIT_URL"),   // your cluster's endpoint, from its Overview page
	Token: os.Getenv("EMIT_TOKEN"), // an API token, from its Tokens tab
	Space: "orders",                // the space this client works in
})

billing := c.Space("billing") // the same client, another space
```

`c.Jobs` works with jobs in the client's space, and `c.Admin` manages spaces and tokens. Every call takes a `context.Context` first.

## Jobs

```go
created, err := c.Jobs.Create(ctx, emit.NewJob{
	Name:        "resize-image",
	Payload:     Resize{ImageID: "img_881", Width: 800},
	Key:         "resize-img_881-800",
	MaxAttempts: 5,
})

many, err := c.Jobs.CreateMany(ctx, []emit.NewJob{
	{Name: "resize-image", Payload: Resize{ImageID: "img_881", Width: 1600}},
	{Name: "resize-image", Payload: Resize{ImageID: "img_882", Width: 800}},
})

job, err := c.Jobs.Get(ctx, created.ID)            // Payload and Result are json.RawMessage
page, err := c.Jobs.List(ctx, emit.ListJobs{Name: "resize-image", Status: emit.Failed, Limit: 50})
next, err := c.Jobs.List(ctx, emit.ListJobs{Page: page.NextPage})
lines, err := c.Jobs.Lines(ctx, created.ID, emit.ReadLines{MinLevel: emit.LevelWarn})
outcomes, err := c.Jobs.Cancel(ctx, created.ID)    // needs the admin scope
```

`NewJob` has the fields in [Creating a job](/docs/concepts/jobs#creating-a-job): `Key`, `DelayMs`, `RunAtMs`, `MaxAttempts`, `TimeoutMs`, `BackoffMs`, `Backoff` and `Callback`. `created.Outcome` is `"new"`, or `"duplicate"` when the key already made a job.

### Waiting for a result

```go
var out struct {
	URL string `json:"url"`
}
err := c.Jobs.Result(ctx, created.ID, &out)
var failed *emit.JobFailed
if errors.As(err, &failed) {
	log.Printf("job %s %s: %s", failed.ID, failed.Status, failed.Result)
}
```

`Result` follows the job until it ends and decodes its result into `out`. If the job failed or was cancelled, the error is a `*emit.JobFailed`.

### Following a job

```go
for event, err := range c.Jobs.Follow(ctx, created.ID, emit.FollowLines()) {
	if err != nil {
		return err
	}
	if event.Progress != nil {
		log.Printf("%.0f%% %s", event.Progress.Percent, event.Progress.Message)
	}
	if event.Finished != nil {
		log.Printf("finished: %s", event.Finished.Status)
	}
}
```

The first event is `Now`, the job as it is. Then one arrives per change (`Claimed`, `Progress`, `Event`, `Lines`, `Retrying`, `Released`), and `Finished` last. `c.Jobs.FollowToken(ctx, id, time.Hour)` makes a token that lets a browser follow the job.

## Workers

```go
worker := c.Worker(ctx, emit.WorkerOptions{Names: []string{"resize-image"}, Concurrency: 8}, func(ctx context.Context, job *emit.HeldJob) (any, error) {
	var in Resize
	if err := job.Decode(&in); err != nil {
		return nil, emit.Unrecoverable(err)
	}
	if err := job.Progress(ctx, 10, "downloading"); err != nil {
		return nil, err
	}
	// ...resize, passing ctx to anything that takes one...
	return map[string]string{"url": "https://images.example.com/" + in.ImageID + ".webp"}, nil
})
```

| The handler returns | The job |
| --- | --- |
| `result, nil` | completes, with `result` as its result |
| `nil, err` | fails this attempt, and is retried while attempts are left |
| `nil, emit.Unrecoverable(err)` | fails for good |
| `nil, emit.RetryAfter(err, 30*time.Second)` | fails this attempt, and is retried after that wait (at most a day) |
| `nil, emit.Release` | goes back to the queue at once; the attempt isn't counted |

A handler that panics fails its attempt, with the panic as the error, and the worker carries on.

`job` has `ID`, `Name`, `Payload`, `Attempt`, `Decode(&v)`, and `Progress`, `Log`, `Event` and `Note`. The handler's `ctx` is cancelled when the worker loses the job, or is stopping: stop work, since nothing more from this attempt is accepted. A handler that returns `ctx.Err()` while the worker is stopping gives its job back.

A log line, a note or a progress message is at most 4,032 bytes, and an event 4 KiB as JSON: past that the call returns an error and sends nothing. 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 empty to claim any job in the space.
:::
:::field Concurrency {type="int" default="1"}
How many jobs the worker holds at once.
:::
:::field LeaseMs {type="int64"}
The lease to ask for. By default, each job's own timeout. The worker renews it every half-lease.
:::
:::field WaitMs {type="int64" default="30000"}
How long one claim waits for work before asking again.
:::
:::field Retry {type="emit.RetryOptions" default="{Attempts: 3, BackoffMs: 250}"}
How a report that fails on the way, from a network error, a 5xx or a 429, is retried before the handler hears of it. The wait doubles each time. The last report of a job is retried until its lease ends.
:::
:::field OnError {type="func(error)" default="log.Printf"}
Told of a last report that never landed.
:::
:::

`worker.Stop()` stops claiming, cancels the handlers' `ctx`, waits for them, and returns what ended the loop, if anything did. `<-worker.Done()` is closed when the loop ends by itself, for example on a revoked token, and `worker.Err()` then says why.

## Defined work

`emit.Define` names a kind of work with Go types for its input and output. Every process that defines it and calls `c.Serve` runs it, and whoever calls `Run` gets the result, wherever it ran.

```go
resize := emit.Define(c, "resize-image", func(ctx context.Context, in Resize, job *emit.HeldJob) (Sized, error) {
	if err := job.Progress(ctx, 50, ""); err != nil {
		return Sized{}, err
	}
	return Sized{URL: "https://images.example.com/" + in.ImageID + ".webp"}, nil
})

worker := c.Serve(ctx, emit.WorkerOptions{Concurrency: 8}) // define everything first

out, err := resize.Run(ctx, Resize{ImageID: "img_881", Width: 800})                            // queue it, wait for the result
id, err := resize.Enqueue(ctx, Resize{ImageID: "img_882", Width: 800}, emit.WithKey("img_882")) // queue it, return its id
here, err := resize.Here(ctx, Resize{ImageID: "img_883", Width: 800})                          // run it in this process
```

`Here` runs the job in this process under a lease: if the process dies, a worker that serves the name takes the next attempt. `emit.Ref[In, Out](c, name)` is work that another service runs: `Run` and `Enqueue` work the same, and `Serve` doesn't claim it.

Options are functions: `emit.WithKey`, `emit.WithDelayMs`, `emit.WithRunAtMs`, `emit.WithMaxAttempts`, `emit.WithBackoffMs`, `emit.WithTimeoutMs`, `emit.WithCallback`, `emit.WithSingleton`, and, for a workflow's steps, `emit.WithLabel`.

## Workflows

`emit.Workflow(c, name, fn)` defines a workflow, and `c.Serve` runs it. See [Workflows](/docs/concepts/workflows) for the whole model. The function gets the workflow's `ctx`, the input and `wf *emit.Run`:

| Call | What it does |
| --- | --- |
| `step.Run(ctx, in, opts...)` | runs a step and returns its result |
| `h := step.Start(ctx, in)`; `h.Get(ctx)`, `h.Cancel(ctx)` | starts a step and returns a handle |
| `step.Enqueue(ctx, in)` | starts a step that outlives the run |
| `emit.All(ctx, step, inputs)` / `emit.AllSettled(...)` | one step per input, all at once |
| `emit.Do(ctx, name, fn)` | runs `fn` once, here, and keeps what it returns |
| `wf.Sleep(ctx, d)`, `wf.SleepUntil(ctx, t)` | pauses the run |
| `v, ok, err := emit.Wait[T](ctx, name, timeout)` | waits for a signal; `ok` is false if the timeout passes |
| `emit.Race(ctx, emit.Case(...), emit.WaitCase(...), emit.SleepCase(...))` | whichever finishes first |
| `wf.Note(ctx, text)` | adds a line to the run's timeline |
| `wf.Now()`, `wf.Rand()` | the time and a random number, the same every time the function runs |
| `return Out{}, wf.ContinueAsNew(input)` | starts the run over with a fresh history |

`emit.Workflow` returns a `*emit.Flow[In, Out]` with `Run`, `Enqueue`, `Result`, `Signal`, `SignalOrStart` and `Cancel`. Signal a run with `emit.ByID(id)`, `emit.ByKey(key)` or `emit.BySingleton(value)`.

## Admin

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

```go
admin := emit.New(emit.Options{URL: os.Getenv("EMIT_URL"), Token: os.Getenv("EMIT_ADMIN_TOKEN")}).Admin

spaces, err := admin.Spaces.List(ctx)
counts, err := admin.Spaces.Counts(ctx, "orders")
space, err := admin.Spaces.Get(ctx, "orders")
space.MaxUnfinished = 100_000
_, err = admin.Spaces.Configure(ctx, space)

token, err := admin.Tokens.Create(ctx, emit.Token{Name: "api-server", Spaces: []string{"orders"}, Scopes: []emit.Scope{emit.ScopeCreate, emit.ScopeRead}})
log.Print(token.Secret) // shown once
err = admin.Tokens.Revoke(ctx, token.ID)
```

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

## Webhook signatures

`emit.VerifySignature(secret, header, body, tolerance)` checks the `Emit-Signature` header of a callback or a wake-up against its raw body, and returns false for a bad signature or one older than `tolerance`. See [Checking the signature](/docs/concepts/jobs#checking-the-signature).

## The firehose

The firehose streams every job change in your cluster. 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.

```go
hose := emit.NewFirehose(os.Getenv("EMIT_FIREHOSE_URL"), os.Getenv("EMIT_FIREHOSE_TOKEN"))
for batch, err := range hose.Watch(ctx, emit.Filter{Spaces: []string{"orders"}, To: emit.Failed}) {
	if err != nil {
		return err
	}
	for _, change := range batch.Changes {
		log.Printf("%s %s: %s -> %s", change.Name, change.ID, change.From, change.To)
	}
	// keep batch.Cursor, and pass it as Filter.After to resume
}
```

## Errors

Every refusal from emit is an `*emit.Error` with its `Code` (such as `NotFound`, `PermissionDenied` or `ResourceExhausted`), 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. A job that ended without completing is a `*emit.JobFailed`.

The SDK doesn't retry a `Create` for you. If you retry one, give the job a `Key`. See [When emit is busy](/docs/concepts/jobs#when-emit-is-busy).