Webhooks and events

The problem

Other services tell you that something happened by calling a URL of yours: GitHub on a push, Stripe on a payment. This is a webhook. Handling them well is hard:

The sender waits for your answer, often only a few seconds. Slow work in the handler makes it time out, and the sender tries again.
Senders send the same webhook more than once (after a timeout, or on purpose). Your code must not do the work twice.
If your code fails half way, the work is lost, unless the sender tries again.

The shape of the solution

The engine has no webhook endpoint and gives you no hosted webhook URL. Its only way in for events is its own event API, POST /e, which takes its own event shape ({"name": ..., "data": ...}) and needs your event key. GitHub and Stripe send their own shapes, so they can not post to it.

So the webhook goes to a route in your own app:

The route checks the sender's signature, so you know the webhook is real.
It sends one event to the engine, with an event ID taken from the webhook, so a second delivery of the same webhook is the same event.
It answers 200 at once.
Functions whose trigger matches the event do the work, in steps, with retries.

The code is in the repo: examples/ts/webhooks.ts and examples/go/webhooks/main.go. It takes GitHub webhooks; the same shape works for any sender.

The webhook route

GitHub signs each delivery with a secret you choose when you set up the webhook. It sends the signature in X-Hub-Signature-256, the kind of event in X-GitHub-Event, and an ID for the delivery in X-GitHub-Delivery.

When you set up the webhook in GitHub, set Content type to application/json. With the other choice, application/x-www-form-urlencoded, GitHub sends the JSON inside a form field, and the routes below answer 415.

TypeScriptGo

const app = express();

// The engine takes one event of up to 1,000,000 bytes (as JSON, with its id, name
// and time). The body becomes the event's data, so take bodies up to 999,000 bytes
// and leave room for the rest. A bigger one gets 413 (see the error handler below).
const MAX_BODY_BYTES = 999_000;

// express.raw keeps the body as bytes: the signature is over the exact bytes sent.
app.post("/webhooks/github", express.raw({ type: "application/json", limit: MAX_BODY_BYTES }), async (req, res) => {
  // GitHub sends JSON only when the webhook's "Content type" is application/json.
  if (!req.is("application/json")) {
    res.status(415).send("set the webhook's Content type to application/json");
    return;
  }
  const body: Buffer = req.body;
  if (!validGitHubSignature(body, req.get("X-Hub-Signature-256"))) {
    res.status(401).send("bad signature");
    return;
  }

  const kind = req.get("X-GitHub-Event") ?? "unknown"; // "push", "pull_request", ...
  const delivery = req.get("X-GitHub-Delivery"); // one ID per delivery
  if (!delivery) {
    // No delivery ID, no stable event ID: refuse it rather than guess one.
    res.status(400).send("no X-GitHub-Delivery header");
    return;
  }

  let data: unknown;
  try {
    data = JSON.parse(body.toString("utf8"));
  } catch {
    res.status(400).send("the body is not JSON");
    return;
  }

  try {
    await flow.send({
      // A stable ID: a redelivery of this delivery has the same ID, so the engine
      // sees the same event, and each function runs once for it.
      id: `github-${delivery}`,
      name: `github/${kind}`,
      data,
    });
  } catch (err) {
    // Not queued: answer an error. GitHub marks the delivery as failed; it does not
    // send it again by itself, but you can redeliver it.
    console.error("could not send the event", err);
    res.status(500).send("try again");
    return;
  }

  // Queued. Answer now; the functions do the work.
  res.status(200).send("ok");
});

// express.raw fails a body over its limit with a 413 error. Answer it as plain text,
// not Express's error page.
app.use((err: { type?: string }, _req: Request, res: Response, next: NextFunction) => {
  if (err.type !== "entity.too.large") return next(err);
  res.status(413).send(`the body is over ${MAX_BODY_BYTES} bytes, too big for one event`);
});

function validGitHubSignature(body: Buffer, header: string | undefined): boolean {
  const secret = process.env.GITHUB_WEBHOOK_SECRET;
  if (!secret || !header) return false;
  const expected = `sha256=${createHmac("sha256", secret).update(body).digest("hex")}`;
  const a = Buffer.from(header);
  const b = Buffer.from(expected);
  return a.length === b.length && timingSafeEqual(a, b);
}

// The engine takes one event of up to 1,000,000 bytes (as JSON, with its id, name
// and time). The body becomes the event's data, so take bodies up to 999,000 bytes
// and leave room for the rest.
const maxBodyBytes = 999_000

func githubWebhook(client *flow.Client) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		// GitHub sends JSON only when the webhook's "Content type" is application/json.
		if mediaType, _, _ := mime.ParseMediaType(r.Header.Get("Content-Type")); mediaType != "application/json" {
			http.Error(w, "set the webhook's Content type to application/json", http.StatusUnsupportedMediaType)
			return
		}

		// Read the exact bytes: the signature is over them. A bigger body is
		// refused, not cut short.
		body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxBodyBytes))
		var tooBig *http.MaxBytesError
		if errors.As(err, &tooBig) {
			http.Error(w, fmt.Sprintf("the body is over %d bytes, too big for one event", maxBodyBytes), http.StatusRequestEntityTooLarge)
			return
		}
		if err != nil {
			http.Error(w, "could not read the body", http.StatusBadRequest)
			return
		}
		if !validGitHubSignature(body, r.Header.Get("X-Hub-Signature-256")) {
			http.Error(w, "bad signature", http.StatusUnauthorized)
			return
		}

		kind := r.Header.Get("X-GitHub-Event")        // "push", "pull_request", ...
		delivery := r.Header.Get("X-GitHub-Delivery") // one ID per delivery
		if delivery == "" {
			// No delivery ID, no stable event ID: refuse it rather than guess one.
			http.Error(w, "no X-GitHub-Delivery header", http.StatusBadRequest)
			return
		}
		if !json.Valid(body) {
			http.Error(w, "the body is not JSON", http.StatusBadRequest)
			return
		}

		_, err = client.Send(r.Context(), flow.Event{
			// A stable ID: a redelivery of this delivery has the same ID, so the
			// engine sees the same event, and each function runs once for it.
			ID:   flow.StrPtr("github-" + delivery),
			Name: "github/" + kind,
			Data: json.RawMessage(body),
		})
		if err != nil {
			// Not queued: answer an error. GitHub marks the delivery as failed; it
			// does not send it again by itself, but you can redeliver it.
			log.Printf("could not send the event: %v", err)
			http.Error(w, "try again", http.StatusInternalServerError)
			return
		}

		// Queued. Answer now; the functions do the work.
		_, _ = w.Write([]byte("ok"))
	}
}

func validGitHubSignature(body []byte, header string) bool {
	secret := os.Getenv("GITHUB_WEBHOOK_SECRET")
	if secret == "" || header == "" {
		return false
	}
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write(body)
	expected := "sha256=" + hex.EncodeToString(mac.Sum(nil))
	return hmac.Equal([]byte(header), []byte(expected))
}

What the route does, and why:

It takes JSON only. Any other Content-Type gets 415.
It reads at most 999,000 bytes. The engine takes one event of up to 1,000,000 bytes, so this leaves room for the event's id, name and time. A bigger body gets 413 (in Express, express.raw()'s limit and a small error handler; in Go, http.MaxBytesReader), and no event is sent. See Limits and gotchas.
It checks the signature over the raw bytes. Parse the JSON only after the check. (In Express, express.raw() on this route keeps the bytes.) A bad signature gets 401; a body that is not JSON gets 400.
It names the event after the webhook: github/push, github/pull_request, and so on. Functions pick the events they want by name.
It sends the webhook's body as the event's data, so functions read event.data.repository.full_name as GitHub sent it.
It gives the event a stable id: github-<delivery ID>. GitHub keeps the same X-GitHub-Delivery when you redeliver a delivery. A delivery with no X-GitHub-Delivery gets 400 and sends no event: with no ID, every such delivery would get the same event ID, and later ones would count as repeats. Pick an ID that stays the same when the sender sends the same thing again; check your sender's docs for which one does. For Stripe, the event's own id field (evt_...) is a good one.
It answers after the event is queued, not after the work. If flow.send fails, it answers 500. GitHub does not send a failed delivery again by itself: redeliver it from the webhook's Recent deliveries in GitHub (up to 3 days back), or with GitHub's API.

Each delivery is handled once

The engine makes the event ID from your id (and your workspace). The same id gives the same event ID, and each function gets one run per event ID. So when GitHub sends delivery d-1001 twice, both sends answer with the same event ID, and each function runs once.

We tested this on a local engine: two sends of the same delivery, one with a bad signature.

200 ok          (first delivery)
200 ok          (same delivery again: same event ID, no new runs)
401 bad signature

The app's log showed chat: 2 new commit(s) in acme/shop once.

This holds while the engine keeps the run: while it runs, and 3 days after it ends (4 days after it failed). A delivery of the same ID after that starts new runs. A function with debounce or batchEvents handles repeats its own way: see Events.

The functions

One event can start many functions. Here a push starts one run of record-push and one run of index-push. Each has its own steps and its own retries: if index-push fails, record-push is not touched.

TypeScriptGo

// Two functions on the same event: each push starts one run of each.
const recordPush = flow.createFunction(
  { id: "record-push" },
  { event: "github/push" },
  async ({ event, step }) => {
    const repo: string = event.data.repository.full_name;
    const commits: number = event.data.commits.length;

    await step.run("save-push", () => savePush(repo, event.data.after, commits));
    await step.run("tell-the-team", () => postToChat(`${commits} new commit(s) in ${repo}`));

    return { repo, commits };
  },
);

const indexPush = flow.createFunction(
  { id: "index-push" },
  { event: "github/push" },
  async ({ event, step }) => {
    await step.run("add-to-search", () => addToSearch(event.data.repository.full_name, event.data.after));
  },
);

// Push is the part of GitHub's push payload these functions read.
type Push struct {
	After      string `json:"after"`
	Repository struct {
		FullName string `json:"full_name"`
	} `json:"repository"`
	Commits []struct {
		ID string `json:"id"`
	} `json:"commits"`
}

// Two functions on the same event: each push starts one run of each.
func addPushFunctions(client *flow.Client) error {
	_, err := flow.CreateFunction(client,
		flow.FunctionOpts{ID: "record-push"},
		flow.EventTrigger("github/push", nil),
		func(ctx context.Context, in flow.Input[Push]) (any, error) {
			push := in.Event.Data
			repo, commits := push.Repository.FullName, len(push.Commits)

			if _, err := step.Run(ctx, "save-push", func(ctx context.Context) (bool, error) {
				return true, savePush(ctx, repo, push.After, commits)
			}); err != nil {
				return nil, err
			}
			if _, err := step.Run(ctx, "tell-the-team", func(ctx context.Context) (bool, error) {
				return true, postToChat(ctx, fmt.Sprintf("%d new commit(s) in %s", commits, repo))
			}); err != nil {
				return nil, err
			}
			return map[string]any{"repo": repo, "commits": commits}, nil
		},
	)
	if err != nil {
		return err
	}

	_, err = flow.CreateFunction(client,
		flow.FunctionOpts{ID: "index-push"},
		flow.EventTrigger("github/push", nil),
		func(ctx context.Context, in flow.Input[Push]) (any, error) {
			return step.Run(ctx, "add-to-search", func(ctx context.Context) (bool, error) {
				return true, addToSearch(ctx, in.Event.Data.Repository.FullName, in.Event.Data.After)
			})
		},
	)
	return err
}

Each step writes in a way that is safe to repeat (keyed by commit, or a set), since a step can run more than once. See Background jobs.

Wait for a later event

A run can wait for another event. Here, when a pull request opens, the run waits up to 2 days for a review of that pull request. If none comes, it reminds the team.

TypeScriptGo

// When a pull request opens, wait up to 2 days for a review of it. No review:
// remind the team.
const reviewReminder = flow.createFunction(
  { id: "review-reminder" },
  { event: "github/pull_request", if: "event.data.action == 'opened'" },
  async ({ event, step }) => {
    const review = await step.waitForEvent("wait-for-review", {
      event: "github/pull_request_review",
      // `event` is the event that started this run; `async` is the new one.
      if: "async.data.pull_request.number == event.data.pull_request.number && async.data.repository.id == event.data.repository.id",
      timeout: "2d",
    });

    if (review === null) {
      await step.run("remind-the-team", () =>
        postToChat(`Pull request #${event.data.pull_request.number} has no review after 2 days`),
      );
      return { reviewed: false };
    }
    return { reviewed: true, by: review.data.review.user.login };
  },
);

// PullRequest is the part of GitHub's pull_request and pull_request_review
// payloads this function reads.
type PullRequest struct {
	Action      string `json:"action"`
	PullRequest struct {
		Number int `json:"number"`
	} `json:"pull_request"`
	Review struct {
		User struct {
			Login string `json:"login"`
		} `json:"user"`
	} `json:"review"`
}

// When a pull request opens, wait up to 2 days for a review of it. No review:
// remind the team.
func addReviewReminder(client *flow.Client) error {
	_, err := flow.CreateFunction(client,
		flow.FunctionOpts{ID: "review-reminder"},
		flow.EventTrigger("github/pull_request", flow.StrPtr("event.data.action == 'opened'")),
		func(ctx context.Context, in flow.Input[PullRequest]) (any, error) {
			review, err := step.WaitForEvent[flow.GenericEvent[PullRequest]](ctx, "wait-for-review", step.WaitForEventOpts{
				Event: "github/pull_request_review",
				// `event` is the event that started this run; `async` is the new one.
				If:      flow.StrPtr("async.data.pull_request.number == event.data.pull_request.number && async.data.repository.id == event.data.repository.id"),
				Timeout: 48 * time.Hour,
			})
			if errors.Is(err, step.ErrEventNotReceived) {
				number := in.Event.Data.PullRequest.Number
				if _, err := step.Run(ctx, "remind-the-team", func(ctx context.Context) (bool, error) {
					return true, postToChat(ctx, fmt.Sprintf("Pull request #%d has no review after 2 days", number))
				}); err != nil {
					return nil, err
				}
				return map[string]any{"reviewed": false}, nil
			}
			if err != nil {
				return nil, err
			}
			return map[string]any{"reviewed": true, "by": review.Data.Review.User.Login}, nil
		},
	)
	return err
}

The trigger's if (event.data.action == 'opened') starts runs only for opened pull requests. A closed one starts no run.
In the wait's if, event is the event that started this run, and async is the new event. For a plain "same field in both", use match: "data.orderId" instead.
waitForEvent gives back the event (TypeScript), or step.ErrEventNotReceived (Go) when the time runs out (TypeScript gives null).
An event that came a little before the wait started still counts: anything that came after the step before the wait started (or after the run was made, if no step came before), up to one hour back.

On a local engine, the review arrived and the run Completed with {"reviewed": true, "by": "grace"}.

Test it

Run the app with GITHUB_WEBHOOK_SECRET set, connect it, and post a signed delivery yourself:

bash

echo '{"after":"9f2c1e7","repository":{"id":42,"full_name":"acme/shop"},"commits":[{"id":"9f2c1e7"},{"id":"1a2b3c4"}]}' > push.json
sig=$(openssl dgst -sha256 -hmac "$GITHUB_WEBHOOK_SECRET" -hex < push.json | sed 's/^.* //')
curl -X POST http://127.0.0.1:3000/webhooks/github \
  -H "Content-Type: application/json" \
  -H "X-GitHub-Event: push" -H "X-GitHub-Delivery: d-1001" \
  -H "X-Hub-Signature-256: sha256=$sig" \
  --data-binary @push.json

To get real deliveries from GitHub on your laptop, the route must be public: use a tunnel (Local development) and give GitHub the tunnel URL plus /webhooks/github.

Send events from anywhere over HTTP

Any program in any language can send events with a plain HTTP request. Use this from your own services, scripts and jobs, not from third parties.

bash

curl -X POST https://worker.flow.clevixstudio.com/e \
  -H "Authorization: Bearer $FLOW_EVENT_KEY" \
  -H "Content-Type: application/json" \
  -d '[{"name":"order/paid","data":{"orderId":"o_1"}},{"name":"order/paid","data":{"orderId":"o_2"}}]'

json

{"ids":["evt_01M5WSJQ1WTHAFBNRS4GAMGX82","evt_01M5WSJQ1W21YNYK7WCX80TCNY"],"status":200}

The key goes in Authorization: Bearer <event key>, or in the path: POST /e/<event key>. The header keeps it out of URL logs.
The body is one event, or a list of up to 100. A request is all or nothing.
Give each sender its own event key (dashboard, Keys), so you can revoke one without cutting off the others.

The answers are listed in the HTTP API reference: 200 with the IDs, 400 for a bad event, 401 for a bad key, 402 or 403 when the workspace may start no runs, 413 when it is too big, 429 when the workspace sends too fast.

Events the engine sends

The engine makes some events itself. Their names start with inngest/, a name you can not use for your own events. Only inngest/function.failed and inngest/function.cancelled are sent out like other events, so only they can trigger your functions. The other two are only the first event of the run they start: a trigger on them starts nothing.

Name	When	`data`	Can trigger a function
`inngest/scheduled.timer`	A cron trigger fires	`cron`, `scheduledAt` (ms)	no
`inngest/function.failed`	A run fails for good	`function_id`, `run_id`, `error`, `event`	yes
`inngest/function.cancelled`	A run is cancelled	`function_id`, `run_id`, `event`, and `error` when there is one	yes
`inngest/function.invoked`	`step.invoke`, or Invoke in the dashboard, starts a run	the `data` the caller passed	no

See Errors and retries for a function that runs when another one fails.

Limits and gotchas

One event is at most 1,000,000 bytes (all its fields, as JSON); a request at most 10 MB and 100 events. The engine answers 413 to a bigger one, and the SDK gives that back as a plain error (sending events failed (413): ...). So the routes here refuse a body over 999,000 bytes themselves, with 413, before they send. GitHub sends deliveries of up to 25 MB. If yours can be bigger, raise the read limit and send only the fields your functions need.
Event names are 1 to 256 characters, and case-sensitive: github/push and GitHub/push are two names.
The ID window: a repeat of the same id is caught while the first run is kept (3 days after it ends, 4 if it failed).
A new wait name: when a run waits for an event name that no run waited for in the last day, an event that comes in the first few seconds can take about 30 seconds to reach the wait. (We saw a review resume the run 30 seconds late this way; once the name was known, the next one resumed it at once.)
Signature checks are yours. The engine only checks your event key. Your route must check the sender's signature.

Webhooks and events ​

The problem ​

The shape of the solution ​

The webhook route ​

Each delivery is handled once ​

The functions ​

Wait for a later event ​

Test it ​

Send events from anywhere over HTTP ​

Events the engine sends ​

Limits and gotchas ​

Webhooks and events

The problem

The shape of the solution

The webhook route

Each delivery is handled once

The functions

Wait for a later event

Test it

Send events from anywhere over HTTP

Events the engine sends

Limits and gotchas