Trigger Inngest events from database changes

In your functions.ts file (typically located in src/inngest/functions.ts), create a new function:

  import { inngest } from "./client";

  export const sendWelcomeEmail = inngest.createFunction(
      { id: "send-welcome-email" },
      { event: "webhook/users.insert" },
      async ({ event }) => {
          const user = event.data;
          //function to send welcome email
          return { event, body: `Welcome email sent to ${user.email} with subject "Hi ${user.name}"` };
      },
  );

This function takes the webhook/users.insert event as input, extracts the user data, and sends a welcome email to the user.

The Inngest.serve() handler registers your function with Inngest so it can invoke the function.

In your route.ts file (typically located in src/app/api/inngest/route.ts) register the function you created above:

import { serve } from "inngest/next";
import { inngest } from "../../../inngest/client";
import { sendWelcomeEmail } from "../../../inngest/functions";

export const { GET, POST, PUT } = serve({
  client: inngest,
  functions: [sendWelcomeEmail],
});

Test the sendWelcomeEmail function by trigger a test eveng in the Inngest development server:

1. Run your Next.js project and Start the Inngest development server:

   npm run dev
   # in a new terminal, start the inngest development server
   npx inngest-cli@latest dev
   

2. Open the Inngest dashboard in your browser and navigate to the “Functions” tab.
3. You should see the send-welcome-email function. Click the “Invoke” button and enter a sample event payload. For this example, use the following event payload:

   {
       "data": {
           "name": "John Doe",
           "email": "john.doe@example.com"
       }
   }
   

4. Click the “Invoke Function” button and then navigate to the “Runs” tab. You should see the run you just triggered:

In the Inngest Cloud Dashboard, navigate to the “Webhooks” tab and click the “Create Webhook” button.

Sequin will send a payload to your webhook containing the database change:

{
  "data:" [
    {
      "record": {
        "id": 1,
        "name": "John Doe",
        "email": "john.doe@example.com"
      },
      "changes": null,
      "action": "insert",
      "metadata": {
        "table_schema": "public",
        "table_name": "users",
        "commit_timestamp": "2024-02-20T14:30:00Z",
        "consumer": {
          "id": "e2f9a3b1-7c6d-4b5a-9f8e-1d2c3b4a5e6f",
          "name": "new_user_consumer"
        }
      }
    }
  ]
}

You need to transform this payload into an Inngest event that triggers your function. To do so, use a “Transform” to extract the the event name and data from the Sequin payload:

function transform(evt, headers = {}, queryParams = {}) {
  return {
    // Rename this webhook to give the events a unique name,
    // or use a field from the incoming event as the event name.
    name: `webhook/${evt.metadata.table_name}.${evt.data[0].action}`,
    data: evt.data[0].record,
  };
};

This will create an Inngest event with the name of the table and the action (i.e. webhook/users.insert) and the data from the record field in the payload:

Navigate to the “Sinks” tab, click the “Create Sink” button, and select “Webhook Sink”.

Select the table you want to capture changes from (i.e public.users).

Specify whether you want to receive changes or rows from the table.

If you selected changes, in “Records to process”, you can indicate whether you want to receive insert, update, and/or delete changes.

In this example, you want to send welcome emails to new users, so under “Records to process” select only insert changes. No need to add a filter.

You can optionally indicate if you want your webhook endpoint to receive a backfill of all or a portion of the table’s existing data. For now, leave “Backfill” toggled off.

Under “Message grouping”, leave the default option selected to ensure events for the same row are sent to your webhook endpoint in order.

Under “Delivery configuration”, choose a conservative value for “Request timeout” and set “Batch size” to 1 since we’re just sending welcome emails.

Under “HTTP Endpoint”, enter the webhook URL you created in Inngest as the base URL. No additional headers are needed.

Give your sink a name (i.e. new_user_webhook_sink) and click “Create Webhook Sink”.

Insert a new row into your database. For example, a new user:

insert into users (name, email) values ('John Doe', 'john.doe@example.com');

In the Sequin web console, navigate to the “messages” tab for your webhook sinks and confirm that Sequin delivered the event to Inngest.

In the Inngest Cloud dashboard, navigate to the “Events” tab, select your event (i.e., webhook/users.insert), and open the “Logs” tab. You should see the event you just created:

Now, click the “Send to Dev Server” button to forward the event to your local Inngest development server.

Finally, in your local Inngest development server, navigate to the “Runs” tab and confirm that the event was received and your function was invoked: