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

Add support for Run Notifications #401 (#738)

Browse Source 
* Add support for Run Notifications #401

* Fixed typo and added changeset
This commit is contained in:
Eric Allam
2023-11-16 14:19:42 +00:00
committed by GitHub
parent c47ad5e038
commit 756024da78
22 changed files with 549 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/yellow-kids-trade.md
+6
View File
@@ -0,0 +1,6 @@
---
"@trigger.dev/sdk": patch
"@trigger.dev/core": patch
---
Add support for listening to run notifications
apps/webapp/app/services/endpointApi.server.ts
+17
View File
@@ -3,6 +3,7 @@ import {
ConnectionAuth,
EndpointHeadersSchema,
ErrorWithStackSchema,
ExecuteJobHeadersSchema,
HttpSourceResponseSchema,
IndexEndpointResponseSchema,
NormalizedResponseSchema,
@@ -14,6 +15,7 @@ import {
RegisterTriggerBodyV1,
RunJobBody,
RunJobResponseSchema,
RunNotification,
ValidateResponse,
ValidateResponseSchema,
} from "@trigger.dev/core";
@@ -148,6 +150,7 @@ export class EndpointApi {
response,
parser: RunJobResponseSchema,
errorParser: ErrorWithStackSchema,
headersParser: ExecuteJobHeadersSchema,
durationInMs: Math.floor(performance.now() - startTimeInMs),
};
}
@@ -367,6 +370,20 @@ export class EndpointApi {
durationInMs: Math.floor(performance.now() - startTimeInMs),
};
}
async deliverRunNotification(notification: RunNotification<any>) {
const response = await safeFetch(this.url, {
method: "POST",
headers: {
"content-type": "application/json",
"x-trigger-api-key": this.apiKey,
"x-trigger-action": "RUN_NOTIFICATION",
},
body: JSON.stringify(notification),
});
return response;
}
}
async function safeFetch(url: string, options: RequestInit) {
apps/webapp/app/services/runs/deliverRunSubscription.server.ts
+64 -3
View File
@@ -1,6 +1,7 @@
import { RunNotification } from "@trigger.dev/core";
import { subtle } from "node:crypto";
import { PrismaClient, prisma } from "~/db.server";
import { EndpointApi } from "../endpointApi.server";
// Infer the type of the #findSubscription method
type FoundSubscription = NonNullable<
@@ -87,6 +88,34 @@ export class DeliverRunSubscriptionService {
);
}
return true;
}
case "ENDPOINT": {
const endpointId = subscription.recipient;
if (endpointId !== subscription.run.endpointId) {
return true;
}
const client = new EndpointApi(
subscription.run.environment.apiKey,
subscription.run.endpoint.url
);
const response = await client.deliverRunNotification(payload);
if (!response) {
throw new Error(
`Failed to deliver endpoint notification to ${subscription.run.endpoint.url}`
);
}
if (!response.ok) {
throw new Error(
`Failed to deliver endpoint notification to ${subscription.run.endpoint.url}: [${response.status}] ${response.statusText}`
);
}
return true;
}
}
@@ -107,6 +136,19 @@ export class DeliverRunSubscriptionService {
organization: true,
project: true,
event: true,
endpoint: true,
tasks: {
where: {
status: "ERRORED",
},
take: 1,
orderBy: {
startedAt: "desc",
},
include: {
attempts: true,
},
},
},
},
},
@@ -114,7 +156,20 @@ export class DeliverRunSubscriptionService {
}
#getPayload(run: FoundRun): RunNotification<any> {
const { id, job, version, statuses, environment, organization, project, event } = run;
const { id, job, version, statuses, environment, organization, project, event, tasks } = run;
const task = tasks[0]
? {
id: tasks[0].idempotencyKey,
cacheKey: tasks[0].displayKey,
status: tasks[0].status,
name: tasks[0].name,
icon: tasks[0].icon,
startedAt: tasks[0].startedAt,
error: tasks[0].output,
params: tasks[0].params,
}
: undefined;
const payload = {
id,
@@ -126,7 +181,7 @@ export class DeliverRunSubscriptionService {
executionDurationInMs: run.executionDuration,
executionCount: run.executionCount,
job: {
id: job.id,
id: job.slug,
version: version.version,
},
statuses: statuses.map((status) => ({
@@ -155,8 +210,14 @@ export class DeliverRunSubscriptionService {
id: event.id,
context: event.context,
timestamp: event.timestamp,
payload: event.payload,
},
...(run.status === "SUCCESS" ? { output: run.output } : { error: run.output }),
...(run.status === "SUCCESS"
? { output: run.output }
: {
error: run.output,
task,
}),
};
return payload as RunNotification<any>;
apps/webapp/app/services/runs/performRunExecutionV3.server.ts
+64 -9
View File
@@ -2,7 +2,6 @@ import {
ApiEventLog,
AutoYieldMetadata,
ConnectionAuth,
EndpointHeadersSchema,
RunJobAutoYieldWithCompletedTaskExecutionError,
RunJobBody,
RunJobError,
@@ -34,11 +33,11 @@ import { CompleteRunTaskService } from "~/routes/api.v1.runs.$runId.tasks.$id.co
import { formatError } from "~/utils/formatErrors.server";
import { safeJsonZodParse } from "~/utils/json";
import { EndpointApi } from "../endpointApi.server";
import { logger } from "../logger.server";
import { forceYieldCoordinator } from "./forceYieldCoordinator.server";
import { workerQueue } from "../worker.server";
import { ResumeTaskService } from "../tasks/resumeTask.server";
import { createExecutionEvent } from "../executions/createExecutionEvent.server";
import { logger } from "../logger.server";
import { ResumeTaskService } from "../tasks/resumeTask.server";
import { workerQueue } from "../worker.server";
import { forceYieldCoordinator } from "./forceYieldCoordinator.server";
type FoundRun = NonNullable<Awaited<ReturnType<typeof findRun>>>;
type FoundTask = FoundRun["tasks"][number];
@@ -287,9 +286,8 @@ export class PerformRunExecutionV3Service {
runId: run.id,
});
const { response, parser, errorParser, durationInMs } = await client.executeJobRequest(
executionBody
);
const { response, parser, errorParser, headersParser, durationInMs } =
await client.executeJobRequest(executionBody);
await createExecutionEvent({
eventType: "finish",
@@ -312,7 +310,7 @@ export class PerformRunExecutionV3Service {
// Update the endpoint version if it has changed
const rawHeaders = Object.fromEntries(response.headers.entries());
const headers = EndpointHeadersSchema.safeParse(rawHeaders);
const headers = headersParser.safeParse(rawHeaders);
if (
headers.success &&
@@ -329,6 +327,58 @@ export class PerformRunExecutionV3Service {
});
}
if (headers.success && headers.data["x-trigger-run-metadata"]) {
logger.debug("Endpoint responded with run metadata", {
metadata: headers.data["x-trigger-run-metadata"],
});
if (
headers.data["x-trigger-run-metadata"].successSubscription &&
!run.subscriptions.some((s) => s.event === "SUCCESS")
) {
await this.#prismaClient.jobRunSubscription.upsert({
where: {
runId_recipient_event: {
runId: run.id,
recipient: run.endpoint.id,
event: "SUCCESS",
},
},
create: {
runId: run.id,
recipient: run.endpoint.id,
recipientMethod: "ENDPOINT",
event: "SUCCESS",
status: "ACTIVE",
},
update: {},
});
}
if (
headers.data["x-trigger-run-metadata"].failedSubscription &&
!run.subscriptions.some((s) => s.event === "FAILURE")
) {
await this.#prismaClient.jobRunSubscription.upsert({
where: {
runId_recipient_event: {
runId: run.id,
recipient: run.endpoint.id,
event: "FAILURE",
},
},
create: {
runId: run.id,
recipient: run.endpoint.id,
recipientMethod: "ENDPOINT",
event: "FAILURE",
status: "ACTIVE",
},
update: {},
});
}
}
const rawBody = await response.text();
if (!response.ok) {
@@ -1257,6 +1307,11 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) {
organization: true,
},
},
subscriptions: {
where: {
recipientMethod: "ENDPOINT",
},
},
_count: {
select: {
tasks: true,
docs/_snippets/jobs/options.mdx
+6
View File
@@ -36,6 +36,12 @@
<ParamField body="enabled" type="boolean">
The `enabled` property is an optional property that specifies whether the Job is enabled or not. The Job will be enabled by default if you omit this property. When a job is disabled, no new runs will be triggered or resumed. In progress runs will continue to run until they are finished or delayed by using `io.wait`.
</ParamField>
<ParamField body="onSuccess" type="function">
The `onSuccess` property is an optional property that specifies a callback function to run when the Job finishes successfully. The callback function receives a [Run Notification](/sdk/run-notification) object as it's only parameter.
</ParamField>
<ParamField body="onFailure" type="function">
The `onFailure` property is an optional property that specifies a callback function to run when the Job fails to complete successfully. The callback function receives a [Run Notification](/sdk/run-notification) object as it's only parameter.
</ParamField>
<ParamField body="logLevel" type="log | error | warn | info | debug">
The `logLevel` property is an optional property that specifies the level of
logging for the Job. The level is inherited from the client if you omit this property.
docs/mint.json
+2 -1
View File
@@ -329,7 +329,8 @@
"sdk/triggerclient/instancemethods/define-http-endpoint",
"sdk/triggerclient/instancemethods/define-dynamic-trigger",
"sdk/triggerclient/instancemethods/define-dynamic-schedule",
"sdk/triggerclient/instancemethods/define-auth-resolver"
"sdk/triggerclient/instancemethods/define-auth-resolver",
"sdk/triggerclient/instancemethods/on"
]
}
]
docs/sdk/job.mdx
+23
View File
@@ -26,6 +26,29 @@ client.defineJob({
});
```
```ts notifications
client.defineJob({
id: "github-integration-on-issue",
name: "GitHub Integration - On Issue",
version: "0.1.0",
trigger: github.triggers.repo({
event: events.onIssue,
owner: "triggerdotdev",
repo: "empty",
}),
onSuccess: async (notification) => {
console.log("Job succeeded", notification);
},
onFailure: async (notification) => {
console.log("Job failed", notification);
},
run: async (payload, io, ctx) => {
await io.logger.info("This is a simple log info message");
return { payload, ctx };
},
});
```
</RequestExample>
# Constructor
docs/sdk/run-notification.mdx
+45
View File
@@ -47,6 +47,24 @@ This document describes the payload of a Run's completion webhook.
The number of individual function executions performed to complete the Run
</ResponseField>
<ResponseField name="invocation" type="object" required>
Metadata about the Invocation / Run
<Expandable title="properties">
<ResponseField name="id" type="string" required>
The Run ID
</ResponseField>
<ResponseField name="payload" type="any" required>
The payload that was passed to the Run
</ResponseField>
<ResponseField name="timestamp" type="Date" required>
The timestamp of the event/Run
</ResponseField>
<ResponseField name="context" type="any">
The context that was passed to the Run
</ResponseField>
</Expandable>
</ResponseField>
<ResponseField name="job" type="object" required>
Metadata about the Job
<Expandable title="properties">
@@ -59,6 +77,33 @@ This document describes the payload of a Run's completion webhook.
</Expandable>
</ResponseField>
<ResponseField name="task" type="object">
Metadata about the Task that failed, only exists if the Run failed
<Expandable title="properties">
<ResponseField name="id" type="string" required>
The Task ID
</ResponseField>
<ResponseField name="cacheKey" type="string">
The Task's cache key
</ResponseField>
<ResponseField name="error" type="Error" required>
And Error-like object that caused the Task to fail
</ResponseField>
<ResponseField name="status" type="string" required>
The Task's status
</ResponseField>
<ResponseField name="name" type="string">
The Task's name
</ResponseField>
<ResponseField name="startedAt" type="Date">
When the Task started
</ResponseField>
<ResponseField name="params" type="any">
The Task's params
</ResponseField>
</Expandable>
</ResponseField>
<ResponseField name="environment" type="object" required>
Metadata about the Environment
<Expandable title="properties">
docs/sdk/triggerclient/instancemethods/define-job.mdx
+23
View File
@@ -24,6 +24,29 @@ client.defineJob({
});
```
```ts notifications
client.defineJob({
id: "github-integration-on-issue",
name: "GitHub Integration - On Issue",
version: "0.1.0",
trigger: github.triggers.repo({
event: events.onIssue,
owner: "triggerdotdev",
repo: "empty",
}),
onSuccess: async (notification) => {
console.log("Job succeeded", notification);
},
onFailure: async (notification) => {
console.log("Job failed", notification);
},
run: async (payload, io, ctx) => {
await io.logger.info("This is a simple log info message");
return { payload, ctx };
},
});
```
</RequestExample>
## Parameters
docs/sdk/triggerclient/instancemethods/on.mdx
+46
View File
@@ -0,0 +1,46 @@
---
title: "TriggerClient: on() instance method"
sidebarTitle: "on()"
description: "Use the `on()` method to listen for run notifications across all Jobs."
---
You can subscribe to run notifications across all your Jobs using the `on()` instance method, which currently supports two events:
- `runSucceeded`: Triggered when a run succeeds.
- `runFailed`: Triggered when a run fails.
Both events receive a [Run Notification](/sdk/run-notification) object as their only parameter.
You can use the notification to determine which job the run belongs to, access the run's payload, output, errors, and more.
If you want to subscribe to just a single Job, see the [defineJob](/sdk/triggerclient/instancemethods/define-job) docs.
## Parameters
<ParamField body="event" type="runSucceeded | runFailed" required>
The notification event to listen for.
</ParamField>
<ParamField body="callback" type="function" required>
The callback function to run when the event is triggered. Receives a [Run
Notification](/sdk/run-notification) object as it's only parameter.
</ParamField>
<RequestExample>
```ts example
export const client = new TriggerClient({
id: "my-project",
apiKey: process.env.TRIGGER_API_KEY,
});
client.on("runSucceeeded", async (notification) => {
console.log(`Run on job ${notification.job.id} succeeded`);
});
client.on("runFailed", async (notification) => {
console.log(`Run on job ${notification.job.id} failed`);
});
```
</RequestExample>
docs/sdk/triggerclient/overview.mdx
+4
View File
@@ -83,3 +83,7 @@ The `defineDynamicSchedule()` method defines a new Dynamic Schedule.
#### [defineAuthResolver()](/sdk/triggerclient/instancemethods/define-auth-resolver)
The `defineAuthResolver()` method defines a new Auth Resolver.
#### [on()](/sdk/triggerclient/instancemethods/on)
Use the `on()` method to listen for run notifications across all Jobs.
packages/core/src/schemas/api.ts
+11
View File
@@ -366,6 +366,17 @@ export const EndpointHeadersSchema = z.object({
"trigger-sdk-version": z.string().optional(),
});
export const ExecuteJobRunMetadataSchema = z.object({
successSubscription: z.boolean().optional(),
failedSubscription: z.boolean().optional(),
});
export const ExecuteJobHeadersSchema = EndpointHeadersSchema.extend({
"x-trigger-run-metadata": z
.preprocess((val) => typeof val === "string" && JSON.parse(val), ExecuteJobRunMetadataSchema)
.optional(),
});
export const RawEventSchema = z.object({
/** The `name` property must exactly match any subscriptions you want to
trigger. */
packages/core/src/schemas/runs.ts
+57 -12
View File
@@ -1,8 +1,9 @@
import { z } from "zod";
import { TaskStatusSchema } from "./tasks";
import { JobRunStatusRecord, JobRunStatusRecordSchema } from "./statuses";
import { Prettify } from "../types";
import { RuntimeEnvironmentType } from "./api";
import { ErrorWithStack } from "./errors";
import { JobRunStatusRecord, JobRunStatusRecordSchema } from "./statuses";
import { TaskStatusSchema } from "./tasks";
export const RunStatusSchema = z.union([
z.literal("PENDING"),
@@ -107,7 +108,38 @@ export const GetRunsSchema = z.object({
nextCursor: z.string().optional(),
});
type RunNotificationCommon = {
export type RunNotificationJobMetadata = { id: string; version: string };
export type RunNotificationEnvMetadata = {
slug: string;
id: string;
type: RuntimeEnvironmentType;
};
export type RunNotificationOrgMetadata = { slug: string; id: string; title: string };
export type RunNotificationProjectMetadata = { slug: string; id: string; name: string };
export type RunNotificationAccountMetadata = { id: string; metadata?: any };
export type RunNotificationInvocationMetadata<T = any> = {
id: string;
context: any;
timestamp: Date;
payload: T;
};
export type RunNotificationRunMetadata = {
/** The Run id */
id: string;
/** The Run status */
statuses: JobRunStatusRecord[];
/** When the run started */
startedAt: Date;
/** When the run was last updated */
updatedAt: Date;
/** When the run was completed */
completedAt: Date;
executionDurationInMs: number;
executionCount: number;
};
type RunNotificationCommon<TPayload = any> = {
/** The Run id */
id: string;
/** The Run status */
@@ -123,20 +155,20 @@ type RunNotificationCommon = {
executionCount: number;
/** Job metadata */
job: { id: string; version: string };
job: RunNotificationJobMetadata;
/** Environment metadata */
environment: { slug: string; id: string; type: RuntimeEnvironmentType };
environment: RunNotificationEnvMetadata;
/** Organization metadata */
organization: { slug: string; id: string; title: string };
organization: RunNotificationOrgMetadata;
/** Project metadata */
project: { slug: string; id: string; name: string };
project: RunNotificationProjectMetadata;
/** Account metadata */
account?: { id: string; metadata?: any };
account?: RunNotificationAccountMetadata;
/** Invocation metadata */
invocation: { id: string; context: any; timestamp: Date };
invocation: RunNotificationInvocationMetadata<TPayload>;
};
export type SuccessfulRunNotification<TOutput> = RunNotificationCommon & {
export type SuccessfulRunNotification<TOutput, TPayload = any> = RunNotificationCommon<TPayload> & {
ok: true;
/** The Run status */
status: "SUCCESS";
@@ -144,12 +176,25 @@ export type SuccessfulRunNotification<TOutput> = RunNotificationCommon & {
output: TOutput;
};
export type FailedRunNotification = RunNotificationCommon & {
export type FailedRunNotification<TPayload = any> = RunNotificationCommon<TPayload> & {
ok: false;
/** The Run status */
status: "FAILURE" | "TIMED_OUT" | "ABORTED" | "CANCELED" | "UNRESOLVED_AUTH" | "INVALID_PAYLOAD";
/** The error of the run */
error: any;
/** The task that failed */
task?: {
id: string;
cacheKey: string | null;
status: string;
name: string;
icon: string | null;
startedAt: string;
error: ErrorWithStack;
params: any | null;
};
};
export type RunNotification<TOutput> = SuccessfulRunNotification<TOutput> | FailedRunNotification;
export type RunNotification<TOutput, TPayload = any> =
| SuccessfulRunNotification<TOutput, TPayload>
| FailedRunNotification<TPayload>;
packages/database/prisma/migrations/20231116113235_add_unique_index_on_job_run_sub/migration.sql
+11
View File
@@ -0,0 +1,11 @@
/*
Warnings:
- A unique constraint covering the columns `[runId,recipient,event]` on the table `JobRunSubscription` will be added. If there are existing duplicate values, this will fail.
*/
-- AlterEnum
ALTER TYPE "JobRunSubscriptionRecipientMethod" ADD VALUE 'ENDPOINT';
-- CreateIndex
CREATE UNIQUE INDEX "JobRunSubscription_runId_recipient_event_key" ON "JobRunSubscription"("runId", "recipient", "event");
packages/database/prisma/schema.prisma
+3
View File
@@ -815,10 +815,13 @@ model JobRunSubscription {
updatedAt DateTime @updatedAt
deliveredAt DateTime?
@@unique([runId, recipient, event])
}
enum JobRunSubscriptionRecipientMethod {
WEBHOOK
ENDPOINT
}
enum JobRunSubscriptionStatus {
packages/trigger-sdk/package.json
+1
View File
@@ -53,6 +53,7 @@
"rimraf": "^3.0.2",
"tsup": "^6.5.0",
"tsx": "^3.12.1",
"typed-emitter": "^2.1.0",
"typescript": "^4.8.4"
},
"engines": {
packages/trigger-sdk/src/job.ts
+8
View File
@@ -1,10 +1,12 @@
import {
FailedRunNotification,
IntegrationConfig,
InvokeOptions,
JobMetadata,
LogLevel,
QueueOptions,
RunNotification,
SuccessfulRunNotification,
} from "@trigger.dev/core";
import { IOWithIntegrations, TriggerIntegration } from "./integrations";
import { TriggerClient } from "./triggerClient";
@@ -75,6 +77,12 @@ export type JobOptions<
context: TriggerContext
) => Promise<TOutput>;
onSuccess?: (
notification: SuccessfulRunNotification<TOutput, TriggerEventType<TTrigger>>
) => void;
onFailure?: (notification: FailedRunNotification<TriggerEventType<TTrigger>>) => void;
// @internal
__internal?: boolean;
};
packages/trigger-sdk/src/triggerClient.ts
+88 -2
View File
@@ -4,6 +4,7 @@ import {
DeserializedJson,
EphemeralEventDispatcherRequestBody,
ErrorWithStackSchema,
FailedRunNotification,
GetRunOptionsWithTaskDetails,
GetRunsOptions,
HandleTriggerSource,
@@ -29,11 +30,13 @@ import {
RunJobBodySchema,
RunJobErrorResponse,
RunJobResponse,
RunNotification,
ScheduleMetadata,
SendEvent,
SendEventOptions,
SourceMetadataV2,
StatusUpdate,
SuccessfulRunNotification,
} from "@trigger.dev/core";
import { yellow } from "colorette";
import { ApiClient } from "./apiClient";
@@ -60,6 +63,7 @@ import { ExternalSource } from "./triggers/externalSource";
import { DynamicIntervalOptions, DynamicSchedule } from "./triggers/scheduled";
import type {
EventSpecification,
NotificationsEventEmitter,
Trigger,
TriggerContext,
TriggerPreprocessContext,
@@ -73,6 +77,7 @@ const registerSourceEvent: EventSpecification<RegisterSourceEventV2> = {
parsePayload: RegisterSourceEventSchemaV2.parse,
};
import EventEmitter from "node:events";
import * as packageJson from "../package.json";
export type TriggerClientOptions = {
@@ -130,6 +135,7 @@ export class TriggerClient {
#registeredSchedules: Record<string, Array<{ id: string; version: string }>> = {};
#registeredHttpEndpoints: Record<string, HttpEndpoint<EventSpecification<any>>> = {};
#authResolvers: Record<string, TriggerAuthResolver> = {};
#eventEmitter: NotificationsEventEmitter = new EventEmitter() as NotificationsEventEmitter;
#client: ApiClient;
#internalLogger: Logger;
@@ -145,6 +151,8 @@ export class TriggerClient {
]);
}
on = this.#eventEmitter.on.bind(this.#eventEmitter);
async handleRequest(
request: Request,
timeOrigin: number = performance.now()
@@ -340,10 +348,14 @@ export class TriggerClient {
triggerVersion,
});
const standardHeaders = this.#standardResponseHeaders(timeOrigin);
standardHeaders["x-trigger-run-metadata"] = this.#serializeRunMetadata(job);
return {
status: 200,
body: results,
headers: this.#standardResponseHeaders(timeOrigin),
headers: standardHeaders,
};
}
case "PREPROCESS_RUN": {
@@ -512,6 +524,24 @@ export class TriggerClient {
await new Promise((resolve) => setTimeout(resolve, timeout));
return {
status: 200,
body: {
ok: true,
},
headers: this.#standardResponseHeaders(timeOrigin),
};
}
case "RUN_NOTIFICATION": {
const rawJson = await request.json();
const runNotification = rawJson as RunNotification<any>;
if (runNotification.ok) {
await this.#deliverSuccessfulRunNotification(runNotification);
} else {
await this.#deliverFailedRunNotification(runNotification);
}
return {
status: 200,
body: {
@@ -1446,13 +1476,69 @@ export class TriggerClient {
});
}
#standardResponseHeaders(start: number) {
#standardResponseHeaders(start: number): Record<string, string> {
return {
"Trigger-Version": API_VERSIONS.LAZY_LOADED_CACHED_TASKS,
"Trigger-SDK-Version": packageJson.version,
"X-Trigger-Request-Timing": `dur=${performance.now() - start / 1000.0}`,
};
}
#serializeRunMetadata(job: Job<Trigger<EventSpecification<any>>, any>) {
const metadata: Record<string, any> = {};
if (
this.#eventEmitter.listenerCount("runSucceeeded") > 0 ||
typeof job.options.onSuccess === "function"
) {
metadata["successSubscription"] = true;
}
if (
this.#eventEmitter.listenerCount("runFailed") > 0 ||
typeof job.options.onFailure === "function"
) {
metadata["failedSubscription"] = true;
}
return JSON.stringify(metadata);
}
async #deliverSuccessfulRunNotification(notification: SuccessfulRunNotification<any>) {
this.#internalLogger.debug("delivering successful run notification", {
notification,
});
this.#eventEmitter.emit("runSucceeeded", notification);
const job = this.#registeredJobs[notification.job.id];
if (!job) {
return;
}
if (typeof job.options.onSuccess === "function") {
await job.options.onSuccess(notification);
}
}
async #deliverFailedRunNotification(notification: FailedRunNotification) {
this.#internalLogger.debug("delivering failed run notification", {
notification,
});
this.#eventEmitter.emit("runFailed", notification);
const job = this.#registeredJobs[notification.job.id];
if (!job) {
return;
}
if (typeof job.options.onFailure === "function") {
await job.options.onFailure(notification);
}
}
}
function dynamicTriggerRegisterSourceJobId(id: string) {
packages/trigger-sdk/src/types.ts
+13
View File
@@ -1,19 +1,25 @@
import type {
DisplayProperty,
EventFilter,
FailedRunNotification,
Logger,
OverridableRunTaskOptions,
Prettify,
RedactString,
RegisteredOptionsDiff,
RunNotificationJobMetadata,
RunNotificationRunMetadata,
RunTaskOptions,
RuntimeEnvironmentType,
ServerTask,
SourceEventOption,
SuccessfulRunNotification,
TriggerMetadata,
} from "@trigger.dev/core";
import { Job } from "./job";
import { TriggerClient } from "./triggerClient";
import { z } from "zod";
import type TypedEmitter from "typed-emitter";
export type {
DisplayProperty,
@@ -165,3 +171,10 @@ export function waitForEventSchema(schema: z.ZodTypeAny) {
accountId: z.string().optional(),
});
}
export type NotificationEvents = {
runSucceeeded: (notification: SuccessfulRunNotification<any>) => void;
runFailed: (notification: FailedRunNotification) => void;
};
export type NotificationsEventEmitter = TypedEmitter<NotificationEvents>;
pnpm-lock.yaml
Generated
+9
View File
@@ -1137,6 +1137,7 @@ importers:
terminal-link: ^3.0.0
tsup: ^6.5.0
tsx: ^3.12.1
typed-emitter: ^2.1.0
typescript: ^4.8.4
ulid: ^2.3.0
uuid: ^9.0.0
@@ -1170,6 +1171,7 @@ importers:
rimraf: 3.0.2
tsup: 6.5.0_typescript@4.9.5
tsx: 3.12.2
typed-emitter: 2.1.0
typescript: 4.9.5
perf:
@@ -29897,6 +29899,7 @@ packages:
/rxjs/7.8.1:
resolution: {integrity: sha512-AA3TVj+0A2iuIoQkWEK/tqFjBq2j+6PO6Y0zJcvzLAFhEFIO3HL0vls9hWLncZbAAbK0mar7oZ4V079I/qPMxg==}
requiresBuild: true
dependencies:
tslib: 2.6.2
@@ -32584,6 +32587,12 @@ packages:
for-each: 0.3.3
is-typed-array: 1.1.10
/typed-emitter/2.1.0:
resolution: {integrity: sha512-g/KzbYKbH5C2vPkaXGu8DJlHrGKHLsM25Zg9WuC9pMGfuvT+X25tZQWo5fK1BjBm8+UrVE9LDCvaY0CQk+fXDA==}
optionalDependencies:
rxjs: 7.8.1
dev: true
/typedarray-to-buffer/3.1.5:
resolution: {integrity: sha512-zdu8XMNEDepKKR+XYOXAVPtWui0ly0NtohUscw+UmaHiAWT8hrV1rr//H6V+0DvJ3OQ19S979M0laLfX8rm82Q==}
dependencies:
references/job-catalog/package.json
+1
View File
@@ -32,6 +32,7 @@
"invoke": "nodemon --watch src/invoke.ts -r tsconfig-paths/register -r dotenv/config src/invoke.ts",
"built-ins": "nodemon --watch src/built-ins.ts -r tsconfig-paths/register -r dotenv/config src/built-ins.ts",
"edge-cases": "nodemon --watch src/edge-cases.ts -r tsconfig-paths/register -r dotenv/config src/edge-cases.ts",
"notifications": "nodemon --watch src/notifications.ts -r tsconfig-paths/register -r dotenv/config src/notifications.ts",
"dev:trigger": "trigger-cli dev --port 8080"
},
"dependencies": {
references/job-catalog/src/notifications.ts
+47
View File
@@ -0,0 +1,47 @@
import { createExpressServer } from "@trigger.dev/express";
import { TriggerClient, invokeTrigger } from "@trigger.dev/sdk";
import { z } from "zod";
export const client = new TriggerClient({
id: "job-catalog",
apiKey: process.env["TRIGGER_API_KEY"],
apiUrl: process.env["TRIGGER_API_URL"],
verbose: false,
ioLogLocalEnabled: true,
});
client.on("runSucceeeded", async (notification) => {
console.log("[client] Run succeeded", notification);
});
client.on("runFailed", async (notification) => {
console.log("[client] Run failed", notification);
});
client.defineJob({
id: "notifications-tester",
name: "Notifications Tester",
version: "1.0.0",
trigger: invokeTrigger({
schema: z.object({
forceTaskError: z.boolean().default(false),
}),
}),
onFailure(notification) {
console.log("[job] Job failed", notification);
},
onSuccess(notification) {
console.log("[job] Job succeeded", notification);
},
run: async (payload, io, ctx) => {
await io.wait("wait-1", 1);
if (payload.forceTaskError) {
await io.runTask("task-1", async () => {
throw new Error("Task failed");
});
}
},
});
createExpressServer(client);