Feature: io.sendEvents() (#728)

* Add io.sendEvents

* Add examples to built-ins catalog entry

* Add docs

* Add changeset

---------

Co-authored-by: Matt Aitken <matt@mattaitken.com>

.changeset/purple-bulldogs-sit.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Add `io.sendEvents()`

apps/webapp/app/routes/api.v1.events.bulk.ts

@@ -0,0 +1,49 @@
+import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { SendBulkEventsBodySchema } from "@trigger.dev/core";
+import { generateErrorMessage } from "zod-error";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { IngestSendEvent } from "~/services/events/ingestSendEvent.server";
+import { eventRecordToApiJson } from "~/api.server";
+import { EventRecord } from "@trigger.dev/database";
+
+export async function action({ request }: ActionFunctionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const authenticatedEnv = authenticationResult.environment;
+
+  // Now parse the request body
+  const anyBody = await request.json();
+
+  const body = SendBulkEventsBodySchema.safeParse(anyBody);
+
+  if (!body.success) {
+    return json({ message: generateErrorMessage(body.error.issues) }, { status: 422 });
+  }
+
+  const service = new IngestSendEvent();
+
+  const events: EventRecord[] = [];
+
+  for (const event of body.data.events) {
+    const eventRecord = await service.call(authenticatedEnv, event, body.data.options);
+
+    if (!eventRecord) {
+      return json({ error: "Failed to create event during bulk ingest" }, { status: 500 });
+    }
+
+    events.push(eventRecord);
+  }
+
+  return json(events.map(eventRecordToApiJson));
+}

docs/_snippets/send-events-params.mdx

@@ -0,0 +1,52 @@
+<ParamField body="events" type="array" required>
+  <Expandable title="event properties" defaultOpen>
+    <ParamField body="name" type="string" required>
+      The `name` property must exactly match any subscriptions you want to
+      trigger.
+    </ParamField>
+    <ParamField body="payload" type="any">
+      The `payload` property will be sent to any matching Jobs and will appear
+      as the `payload` param of the `run()` function. You can leave this
+      parameter out if you just want to trigger a Job without any input data.
+    </ParamField>
+    <ParamField body="context" type="any">
+      The optional `context` property will be sent to any matching Jobs and will
+      be passed through as the `context.event.context` param of the `run()`
+      function. This is optional but can be useful if you want to pass through
+      some additional context to the Job.
+    </ParamField>
+    <ParamField body="id" type="string">
+      The `id` property uniquely identify this particular event. If unset it
+      will be set automatically using `ulid`.
+    </ParamField>
+    <ParamField body="timestamp" type="Date">
+      This is optional, it defaults to the current timestamp. Usually you would
+      only set this if you have a timestamp that you wish to pass through, e.g.
+      you receive a timestamp from a service and you want the same timestamp to
+      be used in your Job.
+    </ParamField>
+    <ParamField body="source" type="string">
+      This is optional, it defaults to "trigger.dev". It can be useful to set
+      this as you can filter events using this in the `eventTrigger()`.
+    </ParamField>
+  </Expandable>
+</ParamField>
+
+<ParamField body="options" type="object">
+  <Expandable title="properties" defaultOpen>
+    <ParamField body="deliverAt" type="Date">
+      An optional Date when you want the event to Trigger Jobs. The event will
+      be sent to the platform immediately but won't be acted upon until the
+      specified time.
+    </ParamField>
+    <ParamField body="deliverAfter" type="number">
+      An optional number of seconds you want to wait for the event to Trigger
+      any relevant Jobs. The event will be sent to the platform immediately but
+      won't be acted upon until the specified time.
+    </ParamField>
+    <ParamField body="accountId" type="string">
+      This optional param will be used by the Trigger.dev Connect feature, which
+      is coming soon.
+    </ParamField>
+  </Expandable>
+</ParamField>

docs/_snippets/send-events-return.mdx

@@ -0,0 +1,29 @@
+<ResponseField name="events" type="array">
+  <Expandable title="properties" defaultOpen>
+    <ResponseField name="id" type="string" required>
+      The `id` of the event that was sent.
+    </ResponseField>
+    <ResponseField name="name" type="string" required>
+      The `name` of the event that was sent.
+    </ResponseField>
+    <ResponseField name="payload" type="any" required>
+      The `payload` of the event that was sent
+    </ResponseField>
+    <ResponseField name="timestamp" type="Date" required>
+      The `timestamp` of the event that was sent
+    </ResponseField>
+    <ResponseField name="context" type="any">
+      The `context` of the event that was sent. Is `undefined` if no context was
+      set when sending the event.
+    </ResponseField>
+    <ResponseField name="deliverAt" type="Date">
+      The timestamp when the event will be delivered to any matching Jobs. Is
+      `undefined` if `deliverAt` or `deliverAfter` wasn't set when sending the
+      event.
+    </ResponseField>
+    <ResponseField name="deliveredAt" type="Date">
+      The timestamp when the event was delivered. Is `undefined` if `deliverAt`
+      or `deliverAfter` were set when sending the event.
+    </ResponseField>
+  </Expandable>
+</ResponseField>

docs/mint.json

@@ -302,6 +302,7 @@
               "group": "Instance methods",
               "pages": [
                 "sdk/triggerclient/instancemethods/sendevent",
+                "sdk/triggerclient/instancemethods/sendevents",
                 "sdk/triggerclient/instancemethods/getevent",
                 "sdk/triggerclient/instancemethods/cancel-event",
                 "sdk/triggerclient/instancemethods/cancel-runs-for-event",
@@ -325,6 +326,7 @@
             "sdk/io/wait",
             "sdk/io/logger",
             "sdk/io/sendevent",
+            "sdk/io/sendevents",
             "sdk/io/backgroundfetch",
             "sdk/io/random",
             "sdk/io/try",

docs/sdk/io/overview.mdx

@@ -32,10 +32,16 @@ Waits for a certain amount of time before continuing the Job. Delays works even
 
 ### [sendEvent()](/sdk/io/sendevent)
 
-`io.sendEvent()` allows you to send an event from inside a Job run. The sent even will trigger any Jobs that are listening for that event (based on the name).
+`io.sendEvent()` allows you to send an event from inside a Job run. The sent event will trigger any Jobs that are listening for that event (based on the name).
 
 If you want to send an event from outside a run (e.g. just from your backend) you can use [client.sendEvent()](/sdk/triggerclient/instancemethods/sendevent).
 
+### [sendEvents()](/sdk/io/sendevents)
+
+`io.sendEvents()` allows you to send multiple events from inside a Job run. The sent events will trigger any Jobs that are listening for those events (based on the name).
+
+If you want to send multiple events from outside a run (e.g. just from your backend) you can use [client.sendEvents()](/sdk/triggerclient/instancemethods/sendevents).
+
 ### [backgroundFetch()](/sdk/io/backgroundfetch)
 
 `io.backgroundFetch()` allows you to fetch data from a URL that can take longer that the serverless timeout. The actual `fetch` request is performed on the Trigger.dev platform, and the response is sent back to you. An example use case is fetching data from a slow API, like some AI endpoints.

docs/sdk/io/sendevent.mdx

@@ -1,13 +1,15 @@
 ---
 title: "io.sendEvent()"
 sidebarTitle: "sendEvent()"
-description: "`io.sendEvent()` allows you to send an event from inside a Job run. The sent even will trigger any Jobs that are listening for that event (based on the name)."
+description: "`io.sendEvent()` allows you to send an event from inside a Job run. The sent event will trigger any Jobs that are listening for that event (based on the name)."
 ---
 
 If you want to send an event from outside a run (e.g. just from your backend) you should use [client.sendEvent()](/sdk/triggerclient/instancemethods/sendevent) instead.
 
 Use [eventTrigger()](/sdk/eventtrigger) on a Job to listen for events.
 
+For multiple events, use [io.sendEvents()](/sdk/io/sendevents) instead.
+
 ## Parameters
 
 <Snippet file="stable-key-param.mdx" />

docs/sdk/io/sendevents.mdx

@@ -0,0 +1,71 @@
+---
+title: "io.sendEvents()"
+sidebarTitle: "sendEvents()"
+description: "`io.sendEvents()` allows you to send multiple events from inside a Job run. The sent events will trigger any Jobs that are listening for those events (based on the name)."
+---
+
+If you want to send multiple events from outside a run (e.g. just from your backend) you should use [client.sendEvents()](/sdk/triggerclient/instancemethods/sendevents) instead.
+
+Use [eventTrigger()](/sdk/eventtrigger) on a Job to listen for events.
+
+For single events, use [io.sendEvent()](/sdk/io/sendevent) instead.
+
+## Parameters
+
+<Snippet file="stable-key-param.mdx" />
+
+<Snippet file="send-events-params.mdx" />
+
+## Returns
+
+<Snippet file="send-events-return.mdx" />
+
+<RequestExample>
+
+```ts Send multiple events
+//this Job sends multiple events that triggers the second job
+client.defineJob({
+  id: "job-1",
+  name: "First job",
+  version: "0.0.1",
+  trigger: cronTrigger({
+    cron: "0 9 * * *", // 9am every day (UTC)
+  }),
+  run: async (payload, io, ctx) => {
+    //sends "new.user" events with a userId in the payload
+    await io.sendEvents("send-events", [
+      {
+        name: "new.user",
+        payload: {
+          userId: "u_12345",
+        },
+      },
+      {
+        name: "new.user",
+        payload: {
+          userId: "u_67890",
+        },
+      },
+    ]);
+  },
+});
+
+client.defineJob({
+  id: "job-2",
+  name: "Second job",
+  version: "0.0.1",
+  //subscribes to the "new.user" event
+  trigger: eventTrigger({
+    name: "new.user",
+    schema: z.object({
+      userId: z.string(),
+    }),
+  }),
+  run: async (payload, io, ctx) => {
+    await io.logger.log("New user created", { userId: payload.userId });
+    //do stuff with the new user
+  },
+});
+```
+
+</RequestExample>

docs/sdk/triggerclient/instancemethods/sendevent.mdx

@@ -8,6 +8,8 @@ You can call this function from anywhere in your backend to send an event. The o
 
 Use [eventTrigger()](/sdk/eventtrigger) on a Job to listen for events.
 
+For multiple events, use [client.sendEvents()](/sdk/triggerclient/instancemethods/sendevents) instead.
+
 ## Parameters
 
 <Snippet file="send-event-params.mdx" />

docs/sdk/triggerclient/instancemethods/sendevents.mdx

@@ -0,0 +1,83 @@
+---
+title: "TriggerClient: sendEvents() instance method"
+sidebarTitle: "sendEvents()"
+description: "The `sendEvents()` instance method send multiple events that triggers any Jobs that are listening for those events (based on the name)."
+---
+
+You can call this function from anywhere in your backend to send multiple events. The other way to send multiple events is by using [io.sendEvents()](/sdk/io/sendevents) from inside a `run()` function.
+
+Use [eventTrigger()](/sdk/eventtrigger) on a Job to listen for events.
+
+For single events, use [client.sendEvent()](/sdk/triggerclient/instancemethods/sendevent) instead.
+
+## Parameters
+
+<Snippet file="send-events-params.mdx" />
+
+## Returns
+
+<Snippet file="send-events-return.mdx" />
+
+<RequestExample>
+
+```ts Simple example with payloads
+const event = client.sendEvents([
+  {
+    name: "new.user",
+    payload: {
+      userId: "u_12345",
+    },
+  },
+  {
+    name: "new.user",
+    payload: {
+      userId: "u_67890",
+    },
+  },
+]);
+```
+
+```ts Send multiple events with an ID
+const event = client.sendEvents([
+  {
+    id: "e_12345", // You can use this to deduplicate events
+    name: "new.user",
+    payload: {
+      userId: "u_12345",
+    },
+  },
+  {
+    id: "e_67890", // You can use this to deduplicate events
+    name: "new.user",
+    payload: {
+      userId: "u_67890",
+    },
+  },
+]);
+```
+
+```ts Send multiple events to be delivered later
+const event = client.sendEvents(
+  [
+    {
+      id: "e_12345", // You can use this to deduplicate events
+      name: "new.user",
+      payload: {
+        userId: "u_12345",
+      },
+    },
+    {
+      id: "e_67890", // You can use this to deduplicate events
+      name: "new.user",
+      payload: {
+        userId: "u_67890",
+      },
+    },
+  ],
+  {
+    deliverAt: new Date("2023-12-01T00:00:00.000Z"),
+  }
+);
+```
+
+</RequestExample>

docs/sdk/triggerclient/overview.mdx

@@ -38,6 +38,12 @@ Sending an event triggers any Jobs that are listening for that event (based on t
 
 You can call this function from anywhere in your code to send an event. The other way to send an event is by using [io.sendEvent()](/sdk/io) from inside a `run()` function.
 
+#### [sendEvents()](/sdk/triggerclient/instancemethods/sendevents)
+
+Sending multiple events triggers any Jobs that are listening for those events (based on the name). Use [eventTrigger()](/sdk/eventtrigger) on a Job to listen for events.
+
+You can call this function from anywhere in your code to send multiple events. The other way to send multiple events is by using [io.sendEvents()](/sdk/io) from inside a `run()` function.
+
 #### [getEvent()](/sdk/triggerclient/instancemethods/getevent)
 
 The `getEvent()` method gets the event details for a given eventId.

packages/core/src/schemas/api.ts

@@ -448,6 +448,11 @@ export const SendEventBodySchema = z.object({
   options: SendEventOptionsSchema.optional(),
 });
 
+export const SendBulkEventsBodySchema = z.object({
+  events: RawEventSchema.array(),
+  options: SendEventOptionsSchema.optional(),
+});
+
 export type SendEventBody = z.infer<typeof SendEventBodySchema>;
 export type SendEventOptions = z.infer<typeof SendEventOptionsSchema>;
 

packages/trigger-sdk/src/apiClient.ts

@@ -205,6 +205,23 @@ export class ApiClient {
     });
   }
 
+  async sendEvents(events: SendEvent[], options: SendEventOptions = {}) {
+    const apiKey = await this.#apiKey();
+
+    this.#logger.debug("Sending multiple events", {
+      events,
+    });
+
+    return await zodfetch(ApiEventLogSchema.array(), `${this.#apiUrl}/api/v1/events/bulk`, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        Authorization: `Bearer ${apiKey}`,
+      },
+      body: JSON.stringify({ events, options }),
+    });
+  }
+
   async cancelEvent(eventId: string) {
     const apiKey = await this.#apiKey();
 

packages/trigger-sdk/src/io.ts

@@ -486,7 +486,7 @@ export class IO {
     )) as TResponseData;
   }
 
-  /** `io.sendEvent()` allows you to send an event from inside a Job run. The sent even will trigger any Jobs that are listening for that event (based on the name).
+  /** `io.sendEvent()` allows you to send an event from inside a Job run. The sent event will trigger any Jobs that are listening for that event (based on the name).
    * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
    * @param event The event to send. The event name must match the name of the event that your Jobs are listening for.
    * @param options Options for sending the event.
@@ -506,6 +506,32 @@ export class IO {
             text: event.name,
           },
           ...(event?.id ? [{ label: "ID", text: event.id }] : []),
+          ...sendEventOptionsProperties(options),
+        ],
+      }
+    );
+  }
+
+  /** `io.sendEvents()` allows you to send multiple events from inside a Job run. The sent events will trigger any Jobs that are listening for those events (based on the name).
+   * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+   * @param event The events to send. The event names must match the names of the events that your Jobs are listening for.
+   * @param options Options for sending the events.
+   */
+  async sendEvents(cacheKey: string | any[], events: SendEvent[], options?: SendEventOptions) {
+    return await this.runTask(
+      cacheKey,
+      async (task) => {
+        return await this._triggerClient.sendEvents(events, options);
+      },
+      {
+        name: "sendEvents",
+        params: { events, options },
+        properties: [
+          {
+            label: "Total Events",
+            text: String(events.length),
+          },
+          ...sendEventOptionsProperties(options),
         ],
       }
     );
@@ -1267,3 +1293,13 @@ async function spaceOut<T>(callback: () => Promise<T>, index: number, delay: num
 
   return await callback();
 }
+
+function sendEventOptionsProperties(options?: SendEventOptions) {
+  return [
+    ...(options?.accountId ? [{ label: "Account ID", text: options.accountId }] : []),
+    ...(options?.deliverAfter
+      ? [{ label: "Deliver After", text: `${options.deliverAfter}s` }]
+      : []),
+    ...(options?.deliverAt ? [{ label: "Deliver At", text: options.deliverAt.toISOString() }] : []),
+  ];
+}

packages/trigger-sdk/src/triggerClient.ts

@@ -742,6 +742,15 @@ export class TriggerClient {
     return this.#client.sendEvent(event, options);
   }
 
+  /** You can call this function from anywhere in your backend to send multiple events. The other way to send multiple events is by using [`io.sendEvents()`](https://trigger.dev/docs/sdk/io/sendevents) from inside a `run()` function.
+   * @param events The events to send.
+   * @param options Options for sending the events.
+   * @returns A promise that resolves to an array of event details
+   */
+  async sendEvents(events: SendEvent[], options?: SendEventOptions) {
+    return this.#client.sendEvents(events, options);
+  }
+
   async cancelEvent(eventId: string) {
     return this.#client.cancelEvent(eventId);
   }

references/job-catalog/src/built-ins.ts

@@ -100,4 +100,59 @@ client.defineJob({
   },
 });
 
+client.defineJob({
+  id: "send-event-example",
+  name: "Send Event Example",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "send.event",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.sendEvent("send-event", {
+      name: "test.event",
+    });
+  },
+});
+
+client.defineJob({
+  id: "send-events-example",
+  name: "Send Multiple Events Example",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "send.events",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.sendEvents(
+      "send-events",
+      [
+        {
+          name: "test.event",
+          payload: {
+            count: 1,
+          },
+        },
+        {
+          name: "test.event",
+          payload: {
+            count: 2,
+          },
+        },
+      ],
+      {
+        deliverAfter: 10,
+      }
+    );
+  },
+});
+
+client.defineJob({
+  id: "receive-test-events",
+  name: "Receive Test Events",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "test.event",
+  }),
+  run: async (payload, io, ctx) => {},
+});
+
 createExpressServer(client);