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

Feature: scheduled triggers

Browse Source
This commit is contained in:
Eric Allam
2023-01-15 12:58:16 -08:00
parent f045c98fa2
commit 8b7b8a81ce
20 changed files with 708 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/great-stingrays-ring.md
+5
View File
@@ -0,0 +1,5 @@
---
"@trigger.dev/sdk": minor
---
Added scheduled events
apps/webapp/app/services/externalSources/registerExternalSource.server.ts
+8 -8
View File
@@ -26,9 +26,9 @@ export class RegisterExternalSource {
return true;
}
if (!externalSource.connection) {
return true; // Somehow the connection slot was deleted, so by returning true we're saying we're done with this webhook
}
console.log("[RegisterExternalSource] registering external source", {
externalSource,
});
switch (externalSource.type) {
case "WEBHOOK": {
@@ -45,8 +45,12 @@ export class RegisterExternalSource {
async #registerWebhook(
externalSource: ExternalSource,
connection: APIConnection
connection?: APIConnection | null
) {
if (!connection) {
return true; // Somehow the connection slot was deleted, so by returning true we're saying we're done with this webhook
}
const accessInfo = await getAccessInfo(connection);
if (accessInfo == null) {
throw new Error("No access token found for webhook");
@@ -106,10 +110,6 @@ export class RegisterExternalSource {
return;
}
if (!externalSource.connection) {
return;
}
return externalSource;
}
apps/webapp/app/services/messageBroker.server.ts
+29 -1
View File
@@ -1,4 +1,7 @@
import { JsonSchema } from "@trigger.dev/common-schemas";
import {
JsonSchema,
ScheduledEventPayloadSchema,
} from "@trigger.dev/common-schemas";
import { DeliverEmailSchema } from "emails";
import type {
CommandCatalog,
@@ -38,6 +41,8 @@ import { PerformIntegrationRequest } from "./requests/performIntegrationRequest.
import { StartIntegrationRequest } from "./requests/startIntegrationRequest.server";
import { WaitForConnection } from "./requests/waitForConnection.server";
import { WorkflowRunDisconnected } from "./runs/runDisconnected.server";
import { DeliverScheduledEvent } from "./scheduler/deliverScheduledEvent.server";
import { RegisterSchedulerSource } from "./scheduler/registerSchedulerSource.server";
let pulsarClient: PulsarClient;
let triggerPublisher: ZodPublisher<TriggerCatalog>;
@@ -316,6 +321,10 @@ const taskQueueCatalog = {
data: z.object({ id: z.string() }),
properties: z.object({}),
},
SCHEDULER_SOURCE_UPSERTED: {
data: z.object({ id: z.string() }),
properties: z.object({}),
},
EXTERNAL_SERVICE_UPSERTED: {
data: z.object({ id: z.string() }),
properties: z.object({}),
@@ -336,6 +345,13 @@ const taskQueueCatalog = {
data: DeliverEmailSchema,
properties: z.object({}),
},
DELIVER_SCHEDULED_EVENT: {
data: z.object({
externalSourceId: z.string(),
payload: ScheduledEventPayloadSchema,
}),
properties: z.object({}),
},
};
function createTaskQueue() {
@@ -459,6 +475,13 @@ function createTaskQueue() {
return isRegistered; // Returning true will mean we don't retry
},
SCHEDULER_SOURCE_UPSERTED: async (id, data, properties) => {
const service = new RegisterSchedulerSource();
const isRegistered = await service.call(data.id);
return isRegistered; // Returning true will mean we don't retry
},
EXTERNAL_SERVICE_UPSERTED: async (id, data, properties) => {
const service = new HandleNewServiceConnection();
@@ -507,6 +530,11 @@ function createTaskQueue() {
await sendEmail(data);
return true;
},
DELIVER_SCHEDULED_EVENT: async (id, data, properties) => {
const service = new DeliverScheduledEvent();
return service.call(data.externalSourceId, data.payload);
},
},
});
apps/webapp/app/services/scheduler/deliverScheduledEvent.server.ts
+82
View File
@@ -0,0 +1,82 @@
import type { ScheduledEventPayload } from "@trigger.dev/common-schemas";
import { ulid } from "ulid";
import type { PrismaClient } from "~/db.server";
import { prisma } from "~/db.server";
import { DispatchWorkflowRun } from "../events/dispatch.server";
import { ScheduleNextEvent } from "./scheduleNextEvent.server";
export class DeliverScheduledEvent {
#prismaClient: PrismaClient;
#scheduleNextEventService: ScheduleNextEvent = new ScheduleNextEvent();
#dispatchWorkflowRunService: DispatchWorkflowRun = new DispatchWorkflowRun();
constructor(prismaClient: PrismaClient = prisma) {
this.#prismaClient = prismaClient;
}
async call(
schedulerSourceId: string,
payload: ScheduledEventPayload
): Promise<boolean> {
const schedulerSource = await this.#prismaClient.schedulerSource.findUnique(
{
where: {
id: schedulerSourceId,
},
include: {
organization: true,
workflow: true,
environment: true,
},
}
);
if (!schedulerSource || schedulerSource.status === "CANCELLED") {
return true;
}
const eventRule = await this.#prismaClient.eventRule.findUnique({
where: {
workflowId_environmentId: {
workflowId: schedulerSource.workflowId,
environmentId: schedulerSource.environmentId,
},
},
});
if (!eventRule) {
console.log(
`No event rule found for workflow ${schedulerSource.workflowId} and environment ${schedulerSource.environmentId}`
);
return true;
}
// 1. Create a TriggerEvent
const triggerEvent = await this.#prismaClient.triggerEvent.create({
data: {
id: ulid(),
service: "scheduler",
name: "scheduled-event",
type: "SCHEDULE",
payload: JSON.parse(JSON.stringify(payload)),
context: {},
organizationId: schedulerSource.organizationId,
environmentId: schedulerSource.environmentId,
},
});
// 2. Create a run
await this.#dispatchWorkflowRunService.call(
schedulerSource.workflow,
eventRule,
triggerEvent,
schedulerSource.environment
);
// 3. Schedule next event
await this.#scheduleNextEventService.call(schedulerSource, triggerEvent);
return true;
}
}
apps/webapp/app/services/scheduler/registerSchedulerSource.server.ts
+67
View File
@@ -0,0 +1,67 @@
import type { SchedulerSource } from ".prisma/client";
import type { PrismaClient } from "~/db.server";
import { prisma } from "~/db.server";
import { ScheduleNextEvent } from "../scheduler/scheduleNextEvent.server";
export class RegisterSchedulerSource {
#prismaClient: PrismaClient;
constructor(prismaClient: PrismaClient = prisma) {
this.#prismaClient = prismaClient;
}
public async call(id: string) {
const schedulerSource = await this.#prismaClient.schedulerSource.findUnique(
{
where: {
id,
},
}
);
if (!schedulerSource) {
return true;
}
if (schedulerSource.status !== "CREATED") {
return true;
}
console.log("[RegisterSchedulerSource] registering external source", {
schedulerSource,
});
return this.#registerScheduler(schedulerSource);
}
async #registerScheduler(schedulerSource: SchedulerSource) {
const scheduleNextEvent = new ScheduleNextEvent();
const isScheduled = await scheduleNextEvent.call(schedulerSource);
if (!isScheduled) {
return false;
}
await this.#prismaClient.schedulerSource.update({
where: {
id: schedulerSource.id,
},
data: {
status: "READY",
readyAt: new Date(),
},
});
await this.#prismaClient.workflow.updateMany({
where: {
id: schedulerSource.workflowId,
},
data: {
status: "READY",
},
});
return true;
}
}
apps/webapp/app/services/scheduler/scheduleNextEvent.server.ts
+47
View File
@@ -0,0 +1,47 @@
import type { SchedulerSource, TriggerEvent } from ".prisma/client";
import {
ScheduledEventPayloadSchema,
ScheduleSourceSchema,
} from "@trigger.dev/common-schemas";
import type { PrismaClient } from "~/db.server";
import { prisma } from "~/db.server";
import { calculateNextScheduledEvent } from "~/utils/scheduler";
import { taskQueue } from "../messageBroker.server";
export class ScheduleNextEvent {
#prismaClient: PrismaClient;
constructor(prismaClient: PrismaClient = prisma) {
this.#prismaClient = prismaClient;
}
async call(
schedulerSource: SchedulerSource,
fromEvent?: TriggerEvent
): Promise<boolean> {
if (schedulerSource.status === "CANCELLED") {
console.log(
"[ScheduleNextEvent] unable to schedule next event because the scheduler source has been cancelled"
);
return false;
}
const source = ScheduleSourceSchema.parse(schedulerSource.schedule);
const scheduledTime = calculateNextScheduledEvent(
source,
fromEvent
? ScheduledEventPayloadSchema.parse(fromEvent.payload)
: undefined
);
const messageId = await taskQueue.publish(
"DELIVER_SCHEDULED_EVENT",
{ externalSourceId: schedulerSource.id, payload: { scheduledTime } },
{},
{ deliverAt: scheduledTime.getTime() }
);
return !!messageId;
}
}
apps/webapp/app/services/workflows/registerWorkflow.server.ts
+50 -16
View File
@@ -37,21 +37,12 @@ export class RegisterWorkflow {
);
if (validation.data.trigger.service !== "trigger") {
const externalSource = await this.#upsertExternalSource(
await this.#upsertExternalSource(
validation.data,
organization
organization,
workflow,
environment
);
if (externalSource) {
await this.#prismaClient.workflow.update({
where: {
id: workflow.id,
},
data: {
externalSourceId: externalSource.id,
},
});
}
}
await this.#upsertEventRule(
@@ -81,13 +72,13 @@ export class RegisterWorkflow {
},
},
update: {
filter: payload.trigger.filter,
filter: "filter" in payload.trigger ? payload.trigger.filter : {},
},
create: {
workflowId: workflow.id,
environmentId: environment.id,
organizationId: organization.id,
filter: payload.trigger.filter,
filter: "filter" in payload.trigger ? payload.trigger.filter : {},
type: payload.trigger.type,
trigger: payload.trigger,
},
@@ -133,7 +124,9 @@ export class RegisterWorkflow {
async #upsertExternalSource(
payload: WorkflowMetadata,
organization: Organization
organization: Organization,
workflow: Workflow,
environment: RuntimeEnvironment
) {
switch (payload.trigger.type) {
case "WEBHOOK": {
@@ -179,12 +172,53 @@ export class RegisterWorkflow {
});
}
await this.#prismaClient.workflow.update({
where: {
id: workflow.id,
},
data: {
externalSourceId: externalSource.id,
},
});
await taskQueue.publish("EXTERNAL_SOURCE_UPSERTED", {
id: externalSource.id,
});
return externalSource;
}
case "SCHEDULE": {
if (!payload.trigger.source) {
return;
}
const schedulerSource = await this.#prismaClient.schedulerSource.upsert(
{
where: {
workflowId_environmentId: {
workflowId: workflow.id,
environmentId: environment.id,
},
},
update: {
schedule: payload.trigger.source,
},
create: {
organizationId: organization.id,
workflowId: workflow.id,
environmentId: environment.id,
schedule: payload.trigger.source,
status: "CREATED",
},
}
);
await taskQueue.publish("SCHEDULER_SOURCE_UPSERTED", {
id: schedulerSource.id,
});
return schedulerSource;
}
default: {
return;
}
apps/webapp/app/utils/scheduler.ts
+75
View File
@@ -0,0 +1,75 @@
import type {
ScheduleSource,
ScheduleSourceRate,
ScheduleSourceCron,
ScheduledEventPayload,
} from "@trigger.dev/common-schemas";
import { parseExpression } from "cron-parser";
export function calculateNextScheduledEvent(
source: ScheduleSource,
previousPayload?: ScheduledEventPayload
): Date {
if ("rateOf" in source) {
return calculateNextRateOfEvent(source, previousPayload);
}
if ("cron" in source) {
return calculateNextCronEvent(source, previousPayload);
}
throw new Error("Invalid schedule source");
}
function calculateNextRateOfEvent(
source: ScheduleSourceRate,
previousPayload?: ScheduledEventPayload
): Date {
const now = new Date();
if (!previousPayload) {
return new Date(now.getTime() + calculateDurationInMs(source));
}
return new Date(
previousPayload.scheduledTime.getTime() + calculateDurationInMs(source)
);
}
function calculateDurationInMs(source: ScheduleSourceRate): number {
if ("minutes" in source.rateOf) {
return source.rateOf.minutes * 60 * 1000;
}
if ("hours" in source.rateOf) {
return source.rateOf.hours * 60 * 60 * 1000;
}
if ("days" in source.rateOf) {
return source.rateOf.days * 24 * 60 * 60 * 1000;
}
throw new Error("Invalid rate of");
}
function calculateNextCronEvent(
source: ScheduleSourceCron,
previousPayload?: ScheduledEventPayload
): Date {
const now = new Date();
if (!previousPayload) {
return parseExpression(source.cron, {
currentDate: now,
})
.next()
.toDate();
}
return parseExpression(source.cron, {
currentDate: previousPayload.scheduledTime,
})
.next()
.toDate();
}
apps/webapp/package.json
+1
View File
@@ -68,6 +68,7 @@
"classnames": "^2.3.1",
"clsx": "^1.2.1",
"compression": "^1.7.4",
"cron-parser": "^4.7.1",
"cross-env": "^7.0.3",
"csstype": "^3.0.10",
"cuid": "^2.1.8",
apps/webapp/prisma/migrations/20230114154045_add_scheduler_external_source/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterEnum
ALTER TYPE "ExternalSourceType" ADD VALUE 'SCHEDULER';
apps/webapp/prisma/migrations/20230114171501_add_cancelled_external_source_status/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterEnum
ALTER TYPE "ExternalSourceStatus" ADD VALUE 'CANCELLED';
apps/webapp/prisma/migrations/20230115194010_add_scheduler_source_model/migration.sql
+41
View File
@@ -0,0 +1,41 @@
/*
Warnings:
- The values [SCHEDULER] on the enum `ExternalSourceType` will be removed. If these variants are still used in the database, this will fail.
*/
-- CreateEnum
CREATE TYPE "SchedulerSourceStatus" AS ENUM ('CREATED', 'READY', 'CANCELLED');
-- AlterEnum
BEGIN;
CREATE TYPE "ExternalSourceType_new" AS ENUM ('WEBHOOK', 'EVENT_BRIDGE', 'HTTP_POLLING');
ALTER TABLE "ExternalSource" ALTER COLUMN "type" TYPE "ExternalSourceType_new" USING ("type"::text::"ExternalSourceType_new");
ALTER TYPE "ExternalSourceType" RENAME TO "ExternalSourceType_old";
ALTER TYPE "ExternalSourceType_new" RENAME TO "ExternalSourceType";
DROP TYPE "ExternalSourceType_old";
COMMIT;
-- CreateTable
CREATE TABLE "SchedulerSource" (
"id" TEXT NOT NULL,
"organizationId" TEXT NOT NULL,
"workflowId" TEXT NOT NULL,
"environmentId" TEXT NOT NULL,
"schedule" TEXT NOT NULL,
"status" "SchedulerSourceStatus" NOT NULL DEFAULT 'CREATED',
"readyAt" TIMESTAMP(3),
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updatedAt" TIMESTAMP(3) NOT NULL,
CONSTRAINT "SchedulerSource_pkey" PRIMARY KEY ("id")
);
-- AddForeignKey
ALTER TABLE "SchedulerSource" ADD CONSTRAINT "SchedulerSource_organizationId_fkey" FOREIGN KEY ("organizationId") REFERENCES "Organization"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "SchedulerSource" ADD CONSTRAINT "SchedulerSource_workflowId_fkey" FOREIGN KEY ("workflowId") REFERENCES "Workflow"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "SchedulerSource" ADD CONSTRAINT "SchedulerSource_environmentId_fkey" FOREIGN KEY ("environmentId") REFERENCES "RuntimeEnvironment"("id") ON DELETE CASCADE ON UPDATE CASCADE;
apps/webapp/prisma/migrations/20230115194253_add_scheduler_uniq_index/migration.sql
+8
View File
@@ -0,0 +1,8 @@
/*
Warnings:
- A unique constraint covering the columns `[workflowId,environmentId]` on the table `SchedulerSource` will be added. If there are existing duplicate values, this will fail.
*/
-- CreateIndex
CREATE UNIQUE INDEX "SchedulerSource_workflowId_environmentId_key" ON "SchedulerSource"("workflowId", "environmentId");
apps/webapp/prisma/migrations/20230115194338_schedule_should_be_json/migration.sql
+9
View File
@@ -0,0 +1,9 @@
/*
Warnings:
- Changed the type of `schedule` on the `SchedulerSource` table. No cast exists, the column would be dropped and recreated, which cannot be done if there is data, since the column is required.
*/
-- AlterTable
ALTER TABLE "SchedulerSource" DROP COLUMN "schedule",
ADD COLUMN "schedule" JSONB NOT NULL;
apps/webapp/prisma/schema.prisma
+48 -15
View File
@@ -4,9 +4,9 @@ datasource db {
}
generator client {
provider = "prisma-client-js"
output = "../node_modules/.prisma/client"
binaryTargets = ["native", "debian-openssl-1.1.x"]
provider = "prisma-client-js"
output = "../node_modules/.prisma/client"
binaryTargets = ["native", "debian-openssl-1.1.x"]
previewFeatures = ["orderByNulls"]
}
@@ -44,13 +44,14 @@ model Organization {
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
users User[]
workflows Workflow[]
environments RuntimeEnvironment[]
apiConnections APIConnection[]
events TriggerEvent[]
externalSources ExternalSource[]
eventRules EventRule[]
users User[]
workflows Workflow[]
environments RuntimeEnvironment[]
apiConnections APIConnection[]
events TriggerEvent[]
externalSources ExternalSource[]
eventRules EventRule[]
schedulerSources SchedulerSource[]
}
model APIConnection {
@@ -102,9 +103,10 @@ model RuntimeEnvironment {
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
events TriggerEvent[]
runs WorkflowRun[]
eventRules EventRule[]
events TriggerEvent[]
runs WorkflowRun[]
eventRules EventRule[]
schedulerSources SchedulerSource[]
@@unique([organizationId, slug])
}
@@ -131,6 +133,7 @@ model Workflow {
runs WorkflowRun[]
rules EventRule[]
externalServices ExternalService[]
schedulerSources SchedulerSource[]
service String @default("trigger")
eventNames String[]
@@ -206,6 +209,7 @@ model ExternalSource {
enum ExternalSourceStatus {
CREATED
READY
CANCELLED
}
enum ExternalSourceType {
@@ -214,6 +218,35 @@ enum ExternalSourceType {
HTTP_POLLING
}
model SchedulerSource {
id String @id @default(cuid())
organization Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade, onUpdate: Cascade)
organizationId String
workflow Workflow @relation(fields: [workflowId], references: [id], onDelete: Cascade, onUpdate: Cascade)
workflowId String
environment RuntimeEnvironment @relation(fields: [environmentId], references: [id], onDelete: Cascade, onUpdate: Cascade)
environmentId String
schedule Json
status SchedulerSourceStatus @default(CREATED)
readyAt DateTime?
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
@@unique([workflowId, environmentId])
}
enum SchedulerSourceStatus {
CREATED
READY
CANCELLED
}
model ExternalService {
id String @id @default(cuid())
slug String
@@ -287,8 +320,8 @@ model IntegrationResponse {
request IntegrationRequest @relation(fields: [requestId], references: [id], onDelete: Cascade, onUpdate: Cascade)
requestId String
output Json
context Json
output Json
context Json
createdAt DateTime @default(now())
}
examples/playground/src/index.ts
+165 -1
View File
@@ -1,4 +1,4 @@
import { Trigger, customEvent } from "@trigger.dev/sdk";
import { Trigger, customEvent, scheduleEvent } from "@trigger.dev/sdk";
import { github, slack } from "@trigger.dev/integrations";
import { z } from "zod";
@@ -19,6 +19,113 @@ import { z } from "zod";
// await ctx.logger.info(
// "This workflow will post a message to Slack immediately and display an Error, Info, Debug, and Warning message."
// );
new Trigger({
id: "playground-1",
name: "Post to Slack immediately",
apiKey: "trigger_dev_zC25mKNn6c0q",
endpoint: "ws://localhost:8889/ws",
on: customEvent({
name: "playground",
schema: z.object({
id: z.string(),
}),
}),
run: async (event, ctx) => {
await ctx.logger.info(
"This workflow will post a message to Slack immediately and display an Error, Info, Debug, and Warning message."
);
await ctx.logger.error("Error message!", { event });
await ctx.logger.info("Info message", { event });
await ctx.logger.debug("Debug message");
await ctx.logger.warn("Warning message!");
const response = await slack.postMessage("send-to-slack", {
channel: "test-integrations",
text: `This is a message from the "Posts to Slack" workflow ${event.id}`,
});
return response.message;
},
}).listen();
// Webhook workflow that sends a message to Slack when a Github issue is created after 2 delays
new Trigger({
id: "playground-2",
name: "Posts to Slack after a GitHub Issue created",
apiKey: "trigger_dev_zC25mKNn6c0q",
endpoint: "ws://localhost:8889/ws",
on: github.events.repoIssueEvent({ repo: "triggerdotdev/trigger.dev" }),
run: async (event, ctx) => {
await ctx.logger.info(
"This workflow will post to Slack when a GitHub Issue created or modified after 2 delays."
);
await ctx.waitFor("initial-wait", { seconds: 30 });
await ctx.waitUntil("wait-until", new Date(Date.now() + 1000 * 30));
await ctx.logger.info("Both types of delay happened");
const response = await slack.postMessage("send-to-slack", {
channel: "test-integrations",
text: `This is a message posts after an Issue was created on GitHub ${event.action}`,
});
await ctx.logger.debug("Debug message");
await ctx.logger.warn("Warning message!");
return response.message;
},
}).listen();
// Workflow that shows all the log types with a 5 second delay between each
new Trigger({
id: "playground-3",
name: "All log types with a 5 second delay between each",
apiKey: "trigger_dev_zC25mKNn6c0q",
endpoint: "ws://localhost:8889/ws",
on: customEvent({
name: "playground",
schema: z.object({
id: z.string(),
}),
}),
run: async (event, ctx) => {
await ctx.logger.info(
"This workflow prints all the log types with a 5 second delay between each."
);
await ctx.logger.info(
"Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it. Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it. Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it. Hey there! This is a really long message to see how the layout handles it. If this breaks the layout, I will fix it."
);
await ctx.waitFor("first-wait", { seconds: 5 });
await ctx.logger.error(
"This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message! This is a really long Error message!",
{ event }
);
await ctx.waitFor("second-wait", { seconds: 5 });
await ctx.logger.info(
"This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message. This is a really long Info message."
);
await ctx.waitFor("third-wait", { seconds: 5 });
await ctx.logger.debug(
"This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. This is a really long Debug message. "
);
// await ctx.logger.error("Error message!", { event });
@@ -290,3 +397,60 @@ const postMessage = new Trigger({
//this workflow will now connect and start listening for events
postMessage.listen();
new Trigger({
id: "playground-7",
name: "Post to Slack many times in a loop",
apiKey: "trigger_dev_zC25mKNn6c0q",
endpoint: "ws://localhost:8889/ws",
on: customEvent({
name: "playground",
schema: z.object({
id: z.string(),
}),
}),
run: async (event, ctx) => {
for (let index = 0; index < 4; index++) {
const response = await slack.postMessage(`send-to-slack-${index}`, {
channel: "test-integrations",
text: `This is a post to Slack many times in a loop ${index} ${event.id}`,
});
}
return {};
},
}).listen();
new Trigger({
id: "scheduled-workflow",
name: "Scheduled Workflow",
apiKey: "trigger_dev_zC25mKNn6c0q",
endpoint: "ws://localhost:8889/ws",
logLevel: "debug",
on: scheduleEvent({ rateOf: { minutes: 5 } }),
run: async (event, ctx) => {
await ctx.logger.info("Received the scheduled event", {
event,
wallTime: new Date(),
});
return { foo: "bar" };
},
}).listen();
new Trigger({
id: "cron-scheduled-workflow",
name: "Cron Scheduled Workflow",
apiKey: "trigger_dev_zC25mKNn6c0q",
endpoint: "ws://localhost:8889/ws",
logLevel: "debug",
on: scheduleEvent({ cron: "0 * * * *" }),
run: async (event, ctx) => {
await ctx.logger.info("Received the cron scheduled event", {
event,
wallTime: new Date(),
});
return { foo: "bar" };
},
}).listen();
packages/common-schemas/src/events.ts
+35
View File
@@ -27,3 +27,38 @@ export type EventFilter = { [key: string]: EventMatcher | EventFilter };
export const EventFilterSchema: z.ZodType<EventFilter> = z.lazy(() =>
z.record(z.union([EventMatcherSchema, EventFilterSchema]))
);
export const ScheduledEventPayloadSchema = z.object({
scheduledTime: z.coerce.date(),
});
export type ScheduledEventPayload = z.infer<typeof ScheduledEventPayloadSchema>;
export const ScheduleSourceRateSchema = z.object({
rateOf: z.union([
z.object({
minutes: z.number().min(1).max(1440).int(),
}),
z.object({
hours: z.number().min(1).max(720).int(),
}),
z.object({
days: z.number().min(1).max(365).int(),
}),
]),
});
export type ScheduleSourceRate = z.infer<typeof ScheduleSourceRateSchema>;
export const ScheduleSourceCronSchema = z.object({
cron: z.string(),
});
export type ScheduleSourceCron = z.infer<typeof ScheduleSourceCronSchema>;
export const ScheduleSourceSchema = z.union([
ScheduleSourceRateSchema,
ScheduleSourceCronSchema,
]);
export type ScheduleSource = z.infer<typeof ScheduleSourceSchema>;
packages/common-schemas/src/triggers.ts
+2 -2
View File
@@ -29,9 +29,9 @@ export type HttpEventTrigger = z.infer<typeof HttpEventTriggerSchema>;
export const ScheduledEventTriggerSchema = z.object({
type: z.literal("SCHEDULE"),
service: z.literal("trigger"),
service: z.literal("scheduler"),
name: z.string(),
filter: EventFilterSchema,
source: JsonSchema,
});
export type ScheduledEventTrigger = z.infer<typeof ScheduledEventTriggerSchema>;
packages/trigger-sdk/src/events.ts
+18
View File
@@ -1,6 +1,8 @@
import {
EventFilterSchema,
TriggerMetadataSchema,
ScheduleSourceSchema,
ScheduledEventPayloadSchema,
} from "@trigger.dev/common-schemas";
import { z } from "zod";
@@ -29,3 +31,19 @@ export function customEvent<TSchema extends z.ZodTypeAny>(
schema: options.schema,
};
}
export type TriggerScheduleOptions = z.infer<typeof ScheduleSourceSchema>;
export function scheduleEvent(
options: TriggerScheduleOptions
): TriggerEvent<typeof ScheduledEventPayloadSchema> {
return {
metadata: {
type: "SCHEDULE",
service: "scheduler",
name: "scheduled-event",
source: options,
},
schema: ScheduledEventPayloadSchema,
};
}
pnpm-lock.yaml
Generated
+14
View File
@@ -105,6 +105,7 @@ importers:
cli-ux: ^6.0.9
clsx: ^1.2.1
compression: ^1.7.4
cron-parser: ^4.7.1
cross-env: ^7.0.3
csstype: ^3.0.10
cuid: ^2.1.8
@@ -208,6 +209,7 @@ importers:
classnames: 2.3.2
clsx: 1.2.1
compression: 1.7.4
cron-parser: 4.7.1
cross-env: 7.0.3
csstype: 3.1.1
cuid: 2.1.8
@@ -7383,6 +7385,13 @@ packages:
resolution: {integrity: sha512-+BO9wPPi+DWTDcNYhr/W90myha8ptzftZT+LwcmUbbok0rcP/fequmFYCw8NMoH7pkAZQzU78b3kYrlua5a9eA==}
dev: false
/cron-parser/4.7.1:
resolution: {integrity: sha512-WguFaoQ0hQ61SgsCZLHUcNbAvlK0lypKXu62ARguefYmjzaOXIVRNrAmyXzabTwUn4sQvQLkk6bjH+ipGfw8bA==}
engines: {node: '>=12.0.0'}
dependencies:
luxon: 3.2.1
dev: false
/cross-env/7.0.3:
resolution: {integrity: sha512-+/HKd6EgcQCJGh2PSjZuUitQBQynKor4wrFbRg4DtAgS1aWO+gU52xpH7M9ScGgXSYmAVS9bIJ8EzuaGw0oNAw==}
engines: {node: '>=10.14', npm: '>=6', yarn: '>=1'}
@@ -11506,6 +11515,11 @@ packages:
resolution: {integrity: sha512-Pn9cox5CsMYngeDbmChANltQl+5pi6XmTrraMSzhPmMBbmgcxmqWry0U3PGapCU1yB4/LqCcom7qhHZiF/jGfQ==}
dev: false
/luxon/3.2.1:
resolution: {integrity: sha512-QrwPArQCNLAKGO/C+ZIilgIuDnEnKx5QYODdDtbFaxzsbZcc/a7WFq7MhsVYgRlwawLtvOUESTlfJ+hc/USqPg==}
engines: {node: '>=12'}
dev: false
/lz-string/1.4.4:
resolution: {integrity: sha512-0ckx7ZHRPqb0oUm8zNr+90mtf9DQB60H1wMCjBtfi62Kl3a7JbHob6gA2bC+xRvZoOL+1hzUK8jeuEIQE8svEQ==}
hasBin: true