Triggers v2

[matt-aitken]

Goals

• A more flexible system that will allow us to support all Triggers (and APIs)
• A less brittle system where a new integration doesn't mess with others
• A system that will allow us to create an HTTP Trigger
• Make creating new Triggers easier for users

The core ideas

Break Triggers down into separate components

Currently Triggers are created as one asset from the user's side. This makes it hard to create new types of Triggers, hard to support more complex Triggers and makes integrations brittle to changes.

Instead let's break the functionality of Triggers into separate components. They can be created and retrieved using the Trigger.dev API/SDK. They can be versioned and updated independently. We can create new components where needed when we discover a better way of doing things, without breaking existing Triggers.

Some components might be:

• HTTP Endpoint (a URL that can receive HTTP requests)
• Secret (a secret value that is stored securely in the database)
• EventFilter (a function that filters events, based on the payload and metadata)
• A Queue (events are enqueued to this, and dequeued. It can have a rate limit, batching, etc)

Move more of the logic to the client

Generally speaking in the current system the Trigger.dev service asks for data about a Trigger and the client responds with that data.

We should change this, so the Trigger.dev service sends lifecycle events to the client. This gives the client the opportunity to perform actions using the Trigger SDK and external APIs. This reversal means that Integrations can be more flexible, allows for complicated combinations of components, and is less brittle.

The client requests the data it wants and performs the actions it wants, using the Trigger API or external ones. The Trigger.dev service sends lifecycle events to the client with the data they need to start doing stuff.

Queues

We should add a new layer to the system, a Queue. This helps decouple the production of events, the consumption of events, and the processing of events. Plus it means we can add concurrency, batching and rate limiting more easily and granularly.

graph LR;
  Producer -- Add Events --> Queue:::trigger -- Consume events --> Consumer:::trigger -. Possibly other steps  .-> Run:::user;

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;

Loading

Events would be pulled from a queue by a consumer, and then processed.

• Queues would have a unique identifier (these also would be scoped to the org, environment?, etc)
  • This would allow multiple sources to put events on a Queue.
• Queues should be granular, so for example one queue for one webhook.
• In theory you could string Queues together. Events could be pulled from one Queue and put on another. For example, batching could be achieved by doing this.

We shouldn't send Trigger payloads to the client

Deployment platforms have aggressive limits on the amount of data you can use in a request to their functions (4.5MB in Vercel's case). But there's no limit (as far as I know) on outgoing requests from inside their functions. We can take advantage of this:

1. We send a message to trigger a Job, but don't include the Trigger payload.
2. We fetch the payload using the Trigger.dev SDK, this won't have a size restriction.
3. We invoke the run function with the payload.

Make webhook automatic registration easier

Unfortunately we need to support re-using webhooks for multiple events as some APIs only allow a limited number of webhooks to be registered. This makes it far trickier as you sometimes need to update an existing webhook rather than just creating a new one.

To make this easier for users we can create an abstraction around this, by introducing a CRUD webhook object. Integration creators would just need to implement the exposed methods. We will do the hard work of calling the right methods with the required data.

Behind-the-scenes this is still using primitives but it's just a layer of abstraction inside the SDK or integration-kit packages that make it easier to work with.

Pseudo-code

const airtableRegistration = new WebhookRegistration("airtable", {
  create: async ({ io, params }) => {
    const webhook = await io.integration.webhooks().create("create-webhook", {
      url: httpSource.url,
      baseId: params.baseId,
      options: specification,
    });

    return {
      id: webhook.id,
      url: webhook.url,
      secret: webhook.secret,
    };
  },
  read: async ({ io, params }) => {
    const listResponse = await io.integration.webhooks().list("list-webhooks", {
      baseId: params.baseId,
    });

    const existingWebhook = listResponse.webhooks.find(
      (w) => w.notificationUrl === httpSource.url
    );

    return {
      hasWebhook: true,
      webhook: {
        id: existingWebhook.id,
        url: existingWebhook.url,
        secret: existingWebhook.secret,
      },
    };
  },
  //etc
});

HTTP Trigger

This would allow users to subscribe to webhooks where we don't have an integration.

• It would have a load of options and a callback that can be invoked when the webhook is received.
• It would require the user to copy/paste the URL and secret into the webhook provider's website.

Questions

Can we make all webhooks "interactive"?

This means we would store the request, and immediately forward to the user's server. This is required for verification requests (WhatsApp) and interactive webhooks (Slack). We'd store the response from the user's server, and send it back to the webhook provider.

If the server doesn't respond then we have the payload so we can add it to a queue to retry later.

Where does filtering happen?

We need to filter before we batch events, or Trigger runs.

Some APIs, with different requirements

Some top-level examples of how we would implement some triggers.

For all of these Diagrams, we'll use these style to distinguish between Trigger.dev and the user's server.

Dashed lines mean there can be a delay between the two steps.

graph LR
  trigger[Trigger.dev service]:::trigger --- client[User's server]:::user;

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;

Loading

WhatsApp

Registration

• WhatsApp doesn't support automatic registration of webhooks.
• Instead the user must type in a URL and secret into the WhatsApp developer website.
• Then WhatsApp will send a verification GET request to the registered URL that expects a specific response with the secret.

The webhooks

• They're straight forward POST requests.
• They require verification using the secret.

Diagrams

1. Registering for webhooks

graph LR;
  indexing[Start indexing]:::trigger -- Request --> integration[WhatsApp integration]:::user -- Metadata --> indexing
  register[Register service]:::trigger -- If changed --> integration2[WhatsApp integration]:::user
  integration2 --> queue[Upsert Queue]:::trigger
  integration2 --> efilter[Upsert EventFilter]:::trigger
  integration2 --> http[Upsert HTTP endpoint]:::trigger
  integration2 --> secret[Upsert secret]:::trigger
  http --> display[Visible in Dashboard]:::trigger
  secret --> display
  display -. User enters them  .-> meta[Meta dev site]:::external;

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;
  classDef external stroke:#a3c;

Loading

2. Receiving a request from WhatsApp

graph LR;
  whatsapp[WhatsApp]:::external -- Request --> HTTPEndpoint:::trigger --> decision_method{{Is GET?}};
  Secret:::trigger --> integration;
  decision_method <-- Yes, verification request --> integration[WhatsApp integration]:::user
  decision_method <-. No, webhook -.-> integration2[WhatsApp integration]:::user -- Events --> Queue:::trigger
  secret2(Secret):::trigger --> integration2;

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;
  classDef external stroke:#a3c;

Loading

• If it's a GET, then it's a verification request. We need to send a Response back which means we need to immediately call the user's server.
• If it's a POST, then it's an actual webhook. This can be queued and processed later.

Note that the Responses to WhatsApp aren't marked on here because it screws the diagram up… 🤦‍♂️

3. Trigger a run

graph LR;
  Queue:::trigger -- Event --> EventFilter:::trigger -. Events that pass filter .-> createrun[Create run]:::trigger -. Start run .-> usersserver[A run]:::user

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;

Loading

GitHub

Registration

• GitHub has API support for registering webhooks.
• You need to provide a secret and URL when registering.

The webhooks

• They're straight forward

Diagrams

1. Registering for webhooks

We need to share webhooks when possible.

graph LR;
  indexing[Start indexing]:::trigger -- Request --> usersserver[GitHub integration]:::user -- Metadata --> indexing
  register[Register service]:::trigger -- If changed --> usersserver2[GitHub integration]:::user
  usersserver2 --> queue[Upsert Queue]:::trigger
  usersserver2 --> efilter[Upsert EventFilter]:::trigger
  usersserver2 --> http[Upsert HTTP endpoint]:::trigger
  usersserver2 --> secret[Upsert secret]:::trigger
  http --> webhooks[Upsert GitHub webhooks]:::external
  secret --> webhooks

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;
  classDef external stroke:#a3c;

Loading

A single GitHub webhook needs to be shared across many of our triggers. Those triggers can subscribe to different events.

We need to handle creating and updating webhooks from separate triggers here. This means we need the information from all the relevant triggers, and the diff since last time we registered. More details on how this can be simply achieved in the "Making webhook automatic registration easier" section.

2. Receiving a request from GitHub

graph LR;
  github[GitHub]:::external -- Request --> HTTPEndpoint:::trigger -.-> integration[GitHub integration]:::user
  Secret:::trigger --> integration;
  integration -- Events --> Queue:::trigger

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;
  classDef external stroke:#a3c;

Loading

3. Trigger a run

graph LR;
  Queue:::trigger -- Event --> EventFilter:::trigger -. Events that pass filter .-> createrun[Create run]:::trigger -. Start run .-> usersserver[A run]:::user

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;

Loading

Stripe

Registration

• Stripe has API support for registering webhooks.
• You provide a URL when registering.
• Stripe responds with a secret that you need to use to verify webhook payloads.

The webhooks

• They're straight forward

Diagrams

1. Registering for webhooks

We need to share webhooks when possible.

graph LR;
  indexing[Start indexing]:::trigger -- Request --> usersserver[Stripe integration]:::user -- Metadata --> indexing
  register[Register service]:::trigger -- If changed --> usersserver2[Stripe integration]:::user
  usersserver2 --> queue[Upsert Queue]:::trigger
  usersserver2 --> efilter[Upsert EventFilter]:::trigger
  usersserver2 --> http[Upsert HTTP endpoint]:::trigger
  http --> webhooks[Upsert Stripe webhooks]:::external
  webhooks --> usersserver2 --> secret[Upsert provided secret]:::trigger

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;
  classDef external stroke:#a3c;

Loading

A single Stripe webhook needs to be shared across many of our triggers. Those triggers can subscribe to different events.

We need to handle creating and updating webhooks from separate triggers here. This means we need the information from all the relevant triggers, and the diff since last time we registered. More details on how this can be simply achieved in the "Making webhook automatic registration easier" section.

2. Receiving a request from Stripe

graph LR;
  stripe[Stripe]:::external -- Request --> HTTPEndpoint:::trigger -.-> integration[Stripe integration]:::user;
  Secret:::trigger --> integration;
  integration -- Events --> Queue:::trigger

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;
  classDef external stroke:#a3c;

Loading

3. Trigger a run

graph LR;
  Queue:::trigger -- Event --> EventFilter:::trigger -. Events that pass filter .-> createrun[Create run]:::trigger -. Start run .-> usersserver[A run]:::user

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;

Loading

Airtable

Registration

• Airtable has API support for registering webhooks
• You provide a URL and secret when registering.

The webhooks

• They send A LOT of webhooks so they need to be batched together to limit the number of runs that happen.
• The webhook payloads aren't the data, they're just notifications
• The actual data needs to be fetched from their API
• In order to fetch the data you need to store a cursor, which is a string that represents the last time you fetched data. You can then fetch the data that has changed since that cursor.

Diagrams

1. Registering for webhooks

We need to share webhooks when possible.

graph LR;
  indexing[Start indexing]:::trigger -- Request --> usersserver[Airtable integration]:::user -- Metadata --> indexing
  register[Register service]:::trigger -- If changed --> usersserver2[Airtable integration]:::user
  usersserver2 --> queue[Upsert Queue]:::trigger
  usersserver2 --> efilter[Upsert EventFilter]:::trigger
  usersserver2 --> http[Upsert HTTP endpoint]:::trigger
  usersserver2 --> secret[Upsert provided secret]:::trigger
  http --> webhooks[Upsert Airtable webhooks]:::external
  secret --> webhooks

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;
  classDef external stroke:#a3c;

Loading

A single Airtable webhook needs to be shared across many of our triggers. Those triggers can subscribe to different events.

We need to handle creating and updating webhooks from separate triggers here. This means we need the information from all the relevant triggers, and the diff since last time we registered. More details on how this can be simply achieved in the "Making webhook automatic registration easier" section.

2. Receiving a request from Airtable

Note that we have to fetch the actual payloads, as the webhooks are just notifications. And we need to store a cursor.

graph LR;
  Airtable:::external -- Request --> HTTPEndpoint:::trigger -.-> integration[Airtable integration]:::user;
  Secret:::trigger --> integration;
  cursor[Cursor]:::trigger -- May not exist --> integration;
  integration --> getpayloads[Get payloads]:::external --> integration;
  integration -- Events --> Queue:::trigger
  integration -- Update --> Cursor:::trigger

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;
  classDef external stroke:#a3c;

Loading

3. Trigger a run

graph LR;
  Queue:::trigger -- Event --> EventFilter:::trigger -. Events that pass filter .-> BatchQueue:::trigger;
  BatchQueue -. Max 10 items & 1 min .-> createrun[Create run]:::trigger -. Start run .-> usersserver[A run]:::user

  classDef trigger fill:#6dfc7b,stroke-width:0px,color:#000;
  classDef user fill:#3ac,stroke-width:0px,color:#fff;

Loading

Notion

Notion has no support for webhooks, so we'll have to poll their API.

• Poll on a schedule, store the data.
• Diff to see if anything has changed.
• If it has, create events and trigger a run.

Discord

Registration

• AFAIK Discord doesn't support automatic registration of webhooks.
• Instead the user must type in a URL and secret into their Discord server settings.
• Discord sends a verification POST request to the URL with a specific body and expects a certain response with the secret.

The webhooks

• All Discord webhooks are "interactive", i.e. they expect a Response back with data.

More details.

Gmail

Registration

• Gmail webhooks use their Push Notifications
• Push Notification uses their Pub/Sub system
• It looks like it is possible to create Pub/Sub clients and topics programatically, but it's complex.
• A Pub/Sub topic can have a webhook URL associated with it, so that when an event is received it is sent to the webhook URL.
• A watch event must be sent to start receiving events for a particular thing, e.g. mailbox changes. This must be sent at least every 7 days, otherwise the watch will expire. They recommend daily.

The webhooks

• The actual data from the Push Notification isn't the payload of changes.
• You need to use an API call to get the actual changes.

[nicktrn]

Goals

where a new integration doesn't mess with others

An example of this would be great!

Components

HTTP Endpoint [...] Secret

How would we control how these two interact in the case of WhatsApp for example? Create an HTTP Endpoint with opts [{ on: "GET", replyWithSecret: secretId }]?

Just thinking this could get tricky as we'd probably need more complex logic, e.g. a fully fledged handler function. Not easily serializable.

But maybe (in this case) the response can be pre-computed without request data.

Queues

Events could be pulled from one Queue and put on another. For example, batching could be achieved by doing this.

That will be great! It's a bit how the current batching system (TBD) works, using push instead.

Would the idea be to use an external queueing system for this?

Payloads

We shouldn't send Trigger payloads to the client

Good thing is, it should be straightforward to do separately from the other changes.

Webhook registration

Completely agree with this 👍

Will have to make update optional, and fall back to delete + create internally if not specified.

HTTP Trigger

It would require the user to copy/paste the URL and secret into the webhook provider's website.

In many cases this could probably be automated by the user (with another task), just as we are currently doing for integrations. Maybe the new Webhook Registration abstraction could help here too.

Interactive Webhooks

We'd store the response from the user's server, and send it back to the webhook provider

Ties into my comment re HTTP Endpoint + Secret primitives.

One thing to keep in mind is that this may not be possible without keeping the incoming connection (from WhatsApp or Slack) open, effectively turning the Trigger.dev platform into a proxy.

Filtering

We need to filter before we batch events

If filters are directly tied to HTTP Endpoints, we may be able to filter during ingestion and not even store all events in the DB. We currently filter before invoking dispatchers:

trigger.dev/apps/webapp/app/services/events/deliverEvent.server.ts

Line 51 in d1ecd6b

this.#evaluateEventRule(eventDispatcher, eventRecord)

[matt-aitken]

Components

Just thinking this could get tricky as we'd probably need more complex logic, e.g. a fully fledged handler function. Not easily serializable.
I think we just pass the request through to the user's server and they return a Response. It would only be if the Request that comes through matches a filter (e.g. method: ["GET"]).

Queues

Would the idea be to use an external queueing system for this?
Do you mean outside of Graphile?

Filters

Ideally we would treat this truly like a pipe, and filters could be added at multiple positions.

• Pre-batching we need filters to determine which events go into which bucket if a webhook is receiving multiple events
• Before delivery we need to filter to determine who they go to as well
• Before ingestion we could filter to avoid storing data potentially. Although I think it might be nice to store them so users will understand what their filters are doing in the UI.

[nicktrn]

I think we just pass the request through to the user's server and they return a Response. It would only be if the Request that comes through matches a filter (e.g. method: ["GET"])

Talked about this, just writing it down for reference: The platform will be acting as a proxy. The response will come directly from the client, proxied by the platform, to e.g. WhatsApp. I expect we will be using a worker queue here as well. Will have to ensure high priority so we don't get timeouts.

Filters

👍 to pipes. Think we're good re storing all events, regardless of filters.

Queues

Do you mean outside of Graphile?

Yes and no. The Graphile worker queue will necessarily be involved at some stage(s).

As we're getting closer to implementation, I think it's important to be more specific. When talking about pull-based queues, it hasn't always been clear what is meant.

• Is it a worker queue as we have now, the only change being the last step? Instead of pushing payloads to the client, we "ping" the client to request payloads from the platform.
• Is it an independent FIFO message queue? No tasks, no workers, just payloads/messages.
• Is it a combination of both?

In this context, I think it may be the latter.

Would be great to be able to easily plug into external queueing systems when the need arises. Should also consider future plans such as long-running server support #430.

Might have to brainstorm a bit to see what's really required.