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

Order steps by a wss supplied timestamp

Browse Source
This commit is contained in:
Eric Allam
2023-01-05 18:06:38 +00:00
parent 47aa7f6f44
commit 6328fc3b36
18 changed files with 102 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/models/workflowRun.server.ts
+9 -3
View File
@@ -82,7 +82,8 @@ export async function failWorkflowRun(
export async function completeWorkflowRun(
output: string,
runId: string,
apiKey: string
apiKey: string,
timestamp: string
) {
const workflowRun = await findWorkflowRunScopedToApiKey(runId, apiKey);
@@ -118,6 +119,7 @@ export async function completeWorkflowRun(
context: {},
startedAt: new Date(),
finishedAt: new Date(),
ts: timestamp,
},
update: {},
});
@@ -128,7 +130,8 @@ export async function triggerEventInRun(
key: string,
event: z.infer<typeof CustomEventSchema>,
runId: string,
apiKey: string
apiKey: string,
timestamp: string
) {
const workflowRun = await findWorkflowRunScopedToApiKey(runId, apiKey);
@@ -146,6 +149,7 @@ export async function triggerEventInRun(
context: {},
startedAt: new Date(),
finishedAt: new Date(),
ts: timestamp,
});
if (step.status === "EXISTING") {
@@ -172,7 +176,8 @@ export async function logMessageInRun(
key: string,
log: z.infer<typeof LogMessageSchema>,
runId: string,
apiKey: string
apiKey: string,
timestamp: string
) {
const workflowRun = await findWorkflowRunScopedToApiKey(runId, apiKey);
@@ -189,6 +194,7 @@ export async function logMessageInRun(
status: "SUCCESS",
startedAt: new Date(),
finishedAt: new Date(),
ts: timestamp,
});
}
apps/webapp/app/models/workflowRunPresenter.server.ts
+1 -1
View File
@@ -196,7 +196,7 @@ function getWorkflowRun(prismaClient: PrismaClient, id: string) {
},
},
},
orderBy: { startedAt: "asc" },
orderBy: { ts: "asc" },
},
},
});
apps/webapp/app/services/delays/initiateDelay.server.ts
+6 -1
View File
@@ -16,7 +16,11 @@ export class InitiateDelay {
this.#prismaClient = prismaClient;
}
async call(runId: string, delay: { key: string; wait: Wait }) {
async call(
runId: string,
timestamp: string,
delay: { key: string; wait: Wait }
) {
const delayUntil = this.#calculateDelayUntil(delay.wait);
// Make sure the delay is not more than 1 year in the future
@@ -32,6 +36,7 @@ export class InitiateDelay {
context: { delayUntil: delayUntil.toISOString() },
status: "RUNNING",
startedAt: new Date(),
ts: timestamp,
});
if (idempotentStep.status === "EXISTING") {
apps/webapp/app/services/messageBroker.server.ts
+15 -7
View File
@@ -143,7 +143,8 @@ async function createTriggerSubscriber() {
data.key,
data.log,
properties["x-workflow-run-id"],
properties["x-api-key"]
properties["x-api-key"],
properties["x-timestamp"]
);
return true;
@@ -166,7 +167,8 @@ async function createTriggerSubscriber() {
await completeWorkflowRun(
data.output,
properties["x-workflow-run-id"],
properties["x-api-key"]
properties["x-api-key"],
properties["x-timestamp"]
);
return true;
@@ -185,6 +187,7 @@ async function createTriggerSubscriber() {
data.key,
properties["x-workflow-run-id"],
properties["x-api-key"],
properties["x-timestamp"],
data.request
);
@@ -195,7 +198,8 @@ async function createTriggerSubscriber() {
data.key,
data.event,
properties["x-workflow-run-id"],
properties["x-api-key"]
properties["x-api-key"],
properties["x-timestamp"]
);
return true;
@@ -203,10 +207,14 @@ async function createTriggerSubscriber() {
INITIALIZE_DELAY: async (id, data, properties) => {
const service = new InitiateDelay();
await service.call(properties["x-workflow-run-id"], {
key: data.key,
wait: data.wait,
});
await service.call(
properties["x-workflow-run-id"],
properties["x-timestamp"],
{
key: data.key,
wait: data.wait,
}
);
return true;
},
apps/webapp/app/services/requests/createIntegrationRequest.server.ts
+2
View File
@@ -16,6 +16,7 @@ export class CreateIntegrationRequest {
key: string,
runId: string,
apiKey: string,
timestamp: string,
data: {
service: string;
endpoint: string;
@@ -61,6 +62,7 @@ export class CreateIntegrationRequest {
endpoint: data.endpoint,
},
status: "PENDING",
ts: timestamp,
});
if (idempotentStep.status === "EXISTING") {
apps/webapp/prisma/migrations/20230105173643_add_timestamp_to_steps/migration.sql
+14
View File
@@ -0,0 +1,14 @@
/*
Warnings:
- Added the required column `timestamp` to the `WorkflowRunStep` table without a default value. This is not possible if the table is not empty.
*/
-- AlterTable
ALTER TABLE "WorkflowRunStep" ADD COLUMN "ts" INTEGER NULL;
-- Add timestamps to existing steps based on the step createdAt (converting to unix timestamp since timestamp is an Integer)
UPDATE "WorkflowRunStep" SET ts = extract(epoch from "createdAt") * 1000;
-- Make timestamp required
ALTER TABLE "WorkflowRunStep" ALTER COLUMN "ts" SET NOT NULL;
apps/webapp/prisma/migrations/20230105175909_make_ts_a_string/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "WorkflowRunStep" ALTER COLUMN "ts" SET DATA TYPE TEXT;
apps/webapp/prisma/schema.prisma
+1
View File
@@ -386,6 +386,7 @@ model WorkflowRunStep {
runId String
idempotencyKey String
ts String
type WorkflowRunStepType
input Json?
apps/wss/src/runController.ts
+12 -10
View File
@@ -154,25 +154,27 @@ export class WorkflowRunController {
async close() {
await this.#subscriber.close();
await this.#publisher.publish(
"WORKFLOW_RUN_DISCONNECTED",
{
id: this.#runId,
},
this.#publishProperties,
{ partitionKey: this.#runId }
);
await this.publish("WORKFLOW_RUN_DISCONNECTED", {
id: this.#runId,
});
this.#logger.debug("Workflow run closed");
}
async publish<TEventName extends keyof WSSCatalog>(
eventName: TEventName,
data: z.infer<WSSCatalog[TEventName]["data"]>
data: z.infer<WSSCatalog[TEventName]["data"]>,
timestamp: number = Date.now()
) {
this.#logger.debug(`Publishing event ${eventName} with data`, data);
return this.#publisher.publish(eventName, data, this.#publishProperties, {
const properties = {
...this.#publishProperties,
"x-timestamp": String(timestamp),
};
return this.#publisher.publish(eventName, data, properties, {
orderingKey: this.#runId,
partitionKey: this.#runId,
});
}
examples/send-to-slack/src/index.ts
-8
View File
@@ -23,12 +23,6 @@ const trigger = new Trigger({
await ctx.waitFor("initial-wait", { minutes: 1 });
await ctx.logger.error("Error message!", { event });
await ctx.waitUntil("initial-wait-until", new Date(Date.now() + 1000 * 60));
await ctx.logger.info("Info message");
const response = await slack.postMessage("send-to-slack", {
channel: "test-integrations",
text: `New domain created: ${event.domain} by customer ${event.customerId}`,
@@ -36,8 +30,6 @@ const trigger = new Trigger({
await ctx.logger.debug("Debug message");
await ctx.logger.warn("Warning message!");
return response.message;
},
});
packages/internal-bridge/src/logger.ts
+3 -1
View File
@@ -10,7 +10,9 @@ export class Logger {
constructor(name: string, level: LogLevel = "info") {
this.#name = name;
this.#level = logLevels.indexOf(level);
this.#level = logLevels.indexOf(
(process.env.TRIGGER_LOG_LEVEL ?? level) as LogLevel
);
}
log(...args: any[]) {
packages/internal-bridge/src/schemas/server.ts
+7
View File
@@ -11,6 +11,7 @@ export const ServerRPCSchema = {
runId: z.string(),
key: z.string(),
wait: WaitSchema,
timestamp: z.string(),
}),
response: z.boolean(),
},
@@ -23,6 +24,7 @@ export const ServerRPCSchema = {
endpoint: z.string(),
params: z.any(),
}),
timestamp: z.string(),
}),
response: z.boolean(),
},
@@ -35,6 +37,7 @@ export const ServerRPCSchema = {
level: z.enum(["DEBUG", "INFO", "WARN", "ERROR"]),
properties: z.string().optional(),
}),
timestamp: z.string(),
}),
response: z.boolean(),
},
@@ -43,6 +46,7 @@ export const ServerRPCSchema = {
runId: z.string(),
key: z.string(),
event: CustomEventSchema,
timestamp: z.string(),
}),
response: z.boolean(),
},
@@ -70,6 +74,7 @@ export const ServerRPCSchema = {
START_WORKFLOW_RUN: {
request: z.object({
runId: z.string(),
timestamp: z.string(),
}),
response: z.boolean(),
},
@@ -77,6 +82,7 @@ export const ServerRPCSchema = {
request: z.object({
runId: z.string(),
output: z.string(),
timestamp: z.string(),
}),
response: z.boolean(),
},
@@ -88,6 +94,7 @@ export const ServerRPCSchema = {
message: z.string(),
stackTrace: z.string().optional(),
}),
timestamp: z.string(),
}),
response: z.boolean(),
},
packages/internal-platform/src/logger.ts
+3 -1
View File
@@ -10,7 +10,9 @@ export class Logger {
constructor(name: string, level: LogLevel = "info") {
this.#name = name;
this.#level = logLevels.indexOf(level);
this.#level = logLevels.indexOf(
(process.env.TRIGGER_LOG_LEVEL ?? level) as LogLevel
);
}
log(...args: any[]) {
packages/internal-platform/src/messages/schemas/customEvents.ts
+3 -3
View File
@@ -1,6 +1,6 @@
import { z } from "zod";
import { CustomEventSchema } from "@trigger.dev/common-schemas";
import { WorkflowRunEventPropertiesSchema } from "../sharedSchemas";
import { z } from "zod";
import { WorkflowSendRunEventPropertiesSchema } from "../sharedSchemas";
export const wss = {
TRIGGER_CUSTOM_EVENT: {
@@ -8,6 +8,6 @@ export const wss = {
key: z.string(),
event: CustomEventSchema,
}),
properties: WorkflowRunEventPropertiesSchema,
properties: WorkflowSendRunEventPropertiesSchema,
},
};
packages/internal-platform/src/messages/sharedSchemas.ts
+1
View File
@@ -20,4 +20,5 @@ export const WorkflowSendEventPropertiesSchema = z.object({
export const WorkflowSendRunEventPropertiesSchema =
WorkflowSendEventPropertiesSchema.extend({
"x-workflow-run-id": z.string(),
"x-timestamp": z.string(),
});
packages/internal-platform/src/messages/zodPublisher.ts
+3
View File
@@ -13,6 +13,7 @@ export type PublishOptions = {
deliverAfter?: number;
deliverAt?: number;
partitionKey?: string;
orderingKey?: string;
};
export type ZodPublisherOptions<PublisherSchema extends MessageCatalogSchema> =
@@ -100,6 +101,7 @@ export class ZodPublisher<PublisherSchema extends MessageCatalogSchema> {
const id = ulid();
this.#logger.debug("Publishing message", {
topic: this.#config.topic,
type,
data,
properties,
@@ -123,6 +125,7 @@ export class ZodPublisher<PublisherSchema extends MessageCatalogSchema> {
deliverAfter: options?.deliverAfter,
deliverAt: options?.deliverAt,
partitionKey: options?.partitionKey,
orderingKey: options?.orderingKey,
});
return response.toString();
packages/internal-platform/src/messages/zodSubscriber.ts
+5 -6
View File
@@ -135,15 +135,14 @@ export class ZodSubscriber<SubscriberSchema extends MessageCatalogSchema> {
throw new Error(`Unknown message type: ${rawMessage.type}`);
}
this.#logger.info(
`Handling message of type ${rawMessage.type}, parsing data and properties`,
rawMessage.data,
rawProperties
);
const message = messageSchema.data.parse(rawMessage.data);
const properties = messageSchema.properties.parse(rawProperties);
this.#logger.debug("Received message, calling handler", {
message,
properties,
});
const handler = this.#handlers[typeName];
const returnValue = await handler(rawMessage.id, message, properties);
packages/trigger-sdk/src/client.ts
+15
View File
@@ -246,6 +246,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
message,
properties: JSON.stringify(properties ?? {}),
},
timestamp: String(highPrecisionTimestamp()),
});
}),
fireEvent: async (key, event) => {
@@ -253,6 +254,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
runId: data.id,
key,
event: JSON.parse(JSON.stringify(event)),
timestamp: String(highPrecisionTimestamp()),
});
},
waitFor: async (key, options) => {
@@ -273,6 +275,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
hours: options.hours,
days: options.days,
},
timestamp: String(highPrecisionTimestamp()),
});
await result;
@@ -294,6 +297,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
type: "SCHEDULE_FOR",
scheduledFor: date.toISOString(),
},
timestamp: String(highPrecisionTimestamp()),
});
await result;
@@ -327,6 +331,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
endpoint: options.endpoint,
params: options.params,
},
timestamp: String(highPrecisionTimestamp()),
});
const output = await result;
@@ -340,6 +345,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
serverRPC
.send("START_WORKFLOW_RUN", {
runId: data.id,
timestamp: String(highPrecisionTimestamp()),
})
.then(() => {
return this.#trigger.options
@@ -348,6 +354,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
return serverRPC.send("COMPLETE_WORKFLOW_RUN", {
runId: data.id,
output: JSON.stringify(output),
timestamp: String(highPrecisionTimestamp()),
});
})
.catch((anyError) => {
@@ -377,6 +384,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
return serverRPC.send("SEND_WORKFLOW_ERROR", {
runId: data.id,
error,
timestamp: String(highPrecisionTimestamp()),
});
});
})
@@ -384,6 +392,7 @@ export class TriggerClient<TSchema extends z.ZodTypeAny> {
return serverRPC.send("SEND_WORKFLOW_ERROR", {
runId: data.id,
error: anyError,
timestamp: String(highPrecisionTimestamp()),
});
});
}
@@ -461,3 +470,9 @@ export const sleep = (ms: number) =>
new Promise((resolve) => setTimeout(resolve, ms));
const messageKey = (runId: string, key: string) => `${runId}:${key}`;
function highPrecisionTimestamp() {
const [seconds, nanoseconds] = process.hrtime();
return seconds * 1e9 + nanoseconds;
}