github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Feature: io.sendEvents() (#728)

Browse Source 
* Add io.sendEvents

* Add examples to built-ins catalog entry

* Add docs

* Add changeset

---------

Co-authored-by: Matt Aitken <matt@mattaitken.com>
This commit is contained in:
nicktrn
2023-11-09 16:42:23 +00:00
committed by GitHub
parent 9f1f59cc81
commit d02173442c
16 changed files with 433 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/purple-bulldogs-sit.md
+6
View File
@@ -0,0 +1,6 @@
---
"@trigger.dev/sdk": patch
"@trigger.dev/core": patch
---
Add `io.sendEvents()`
apps/webapp/app/routes/api.v1.events.bulk.ts
+49
View File
@@ -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
+52
View File
@@ -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
+29
View File
@@ -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
+2
View File
@@ -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
+7 -1
View File
@@ -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
+3 -1
View File
@@ -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
+71
View File
@@ -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
+2
View File
@@ -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
+83
View File
@@ -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
+6
View File
@@ -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
+5
View File
@@ -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
+17
View File
@@ -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
+37 -1
View File
@@ -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
+9
View File
@@ -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
+55
View File
@@ -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);