diff --git a/.changeset/new-pants-beg.md b/.changeset/new-pants-beg.md new file mode 100644 index 000000000..e484742c9 --- /dev/null +++ b/.changeset/new-pants-beg.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/core": patch +--- + +Fix issue when using SDK in non-node environments by scoping the stream import with node: diff --git a/apps/webapp/app/consts.ts b/apps/webapp/app/consts.ts index 30dba9c7f..8b27c3c95 100644 --- a/apps/webapp/app/consts.ts +++ b/apps/webapp/app/consts.ts @@ -13,3 +13,4 @@ export const VERCEL_RESPONSE_TIMEOUT_STATUS_CODES = [408, 504]; export const MAX_BATCH_TRIGGER_ITEMS = 100; export const MAX_TASK_RUN_ATTEMPTS = 250; export const BULK_ACTION_RUN_LIMIT = 250; +export const MAX_JOB_RUN_EXECUTION_COUNT = 250; diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 71edf6a6c..07035ffb7 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -164,6 +164,8 @@ const EnvironmentSchema = z.object({ ALERT_RESEND_API_KEY: z.string().optional(), MAX_SEQUENTIAL_INDEX_FAILURE_COUNT: z.coerce.number().default(96), + + LOOPS_API_KEY: z.string().optional(), }); export type Environment = z.infer; diff --git a/apps/webapp/app/services/loops.server.ts b/apps/webapp/app/services/loops.server.ts new file mode 100644 index 000000000..75f72d027 --- /dev/null +++ b/apps/webapp/app/services/loops.server.ts @@ -0,0 +1,75 @@ +import { env } from "~/env.server"; +import { logger } from "./logger.server"; + +class LoopsClient { + constructor(private readonly apiKey: string) {} + + async userCreated({ + userId, + email, + name, + }: { + userId: string; + email: string; + name: string | null; + }) { + logger.info(`Loops send "sign-up" event`, { userId, email, name }); + return this.#sendEvent({ + email, + userId, + firstName: name?.split(" ").at(0), + eventName: "sign-up", + }); + } + + async #sendEvent({ + email, + userId, + firstName, + eventName, + eventProperties, + }: { + email: string; + userId: string; + firstName?: string; + eventName: string; + eventProperties?: Record; + }) { + const options = { + method: "POST", + headers: { Authorization: `Bearer ${this.apiKey}`, "Content-Type": "application/json" }, + body: JSON.stringify({ + email, + userId, + firstName, + eventName, + eventProperties, + }), + }; + + try { + const response = await fetch("https://app.loops.so/api/v1/events/send", options); + + if (!response.ok) { + logger.error(`Loops sendEvent ${eventName} bad status`, { status: response.status }); + return false; + } + + const responseBody = (await response.json()) as any; + + if (!responseBody.success) { + logger.error(`Loops sendEvent ${eventName} failed response`, { + message: responseBody.message, + }); + return false; + } + + return true; + } catch (error) { + logger.error(`Loops sendEvent ${eventName} failed`, { error }); + return false; + } + } +} + +export const loopsClient = env.LOOPS_API_KEY ? new LoopsClient(env.LOOPS_API_KEY) : null; diff --git a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts index 6f0ecd56c..8d26f8888 100644 --- a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts +++ b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts @@ -25,6 +25,7 @@ import { import { generateErrorMessage } from "zod-error"; import { eventRecordToApiJson } from "~/api.server"; import { + MAX_JOB_RUN_EXECUTION_COUNT, MAX_RUN_CHUNK_EXECUTION_LIMIT, MAX_RUN_YIELDED_EXECUTIONS, RUN_CHUNK_EXECUTION_BUFFER, @@ -141,6 +142,46 @@ export class PerformRunExecutionV3Service { }); } + if (run.version.status === "DISABLED") { + return await this.#failRunExecution( + this.#prismaClient, + run, + { + message: `Job version ${run.version.version} is disabled, aborting run.`, + }, + "ABORTED" + ); + } + + // If the execution duration is greater than the maximum execution time, we need to fail the run + if (run.executionDuration >= run.organization.maximumExecutionTimePerRunInMs) { + await this.#failRunExecution( + this.#prismaClient, + run, + { + message: `Execution timed out after ${ + run.organization.maximumExecutionTimePerRunInMs / 1000 + } seconds`, + }, + "TIMED_OUT", + 0 + ); + return; + } + + if (run.executionCount >= MAX_JOB_RUN_EXECUTION_COUNT) { + await this.#failRunExecution( + this.#prismaClient, + run, + { + message: `Execution timed out after ${run.executionCount} executions`, + }, + "TIMED_OUT", + 0 + ); + return; + } + const client = new EndpointApi(run.environment.apiKey, run.endpoint.url); const event = eventRecordToApiJson(run.event); diff --git a/apps/webapp/app/services/telemetry.server.ts b/apps/webapp/app/services/telemetry.server.ts index d6d83b23d..341a8094a 100644 --- a/apps/webapp/app/services/telemetry.server.ts +++ b/apps/webapp/app/services/telemetry.server.ts @@ -7,6 +7,7 @@ import type { Organization } from "~/models/organization.server"; import type { Project } from "~/models/project.server"; import type { User } from "~/models/user.server"; import { singleton } from "~/utils/singleton"; +import { loopsClient } from "./loops.server"; type Options = { postHogApiKey?: string; @@ -39,18 +40,19 @@ class Telemetry { user = { identify: ({ user, isNewUser }: { user: User; isNewUser: boolean }) => { - if (this.#posthogClient === undefined) return; - this.#posthogClient.identify({ - distinctId: user.id, - properties: { - email: user.email, - name: user.name, - authenticationMethod: user.authenticationMethod, - admin: user.admin, - createdAt: user.createdAt, - isNewUser, - }, - }); + if (this.#posthogClient) { + this.#posthogClient.identify({ + distinctId: user.id, + properties: { + email: user.email, + name: user.name, + authenticationMethod: user.authenticationMethod, + admin: user.admin, + createdAt: user.createdAt, + isNewUser, + }, + }); + } if (isNewUser) { this.#capture({ userId: user.id, @@ -64,6 +66,12 @@ class Telemetry { }, }); + loopsClient?.userCreated({ + userId: user.id, + email: user.email, + name: user.name, + }); + this.#triggerClient?.sendEvent({ name: "user.created", payload: { diff --git a/packages/core/src/v3/zodfetch.ts b/packages/core/src/v3/zodfetch.ts index e77b27f9d..ac15f38e3 100644 --- a/packages/core/src/v3/zodfetch.ts +++ b/packages/core/src/v3/zodfetch.ts @@ -4,7 +4,7 @@ import { APIConnectionError, APIError } from "./apiErrors"; import { RetryOptions } from "./schemas"; import { calculateNextRetryDelay } from "./utils/retries"; import { FormDataEncoder } from "form-data-encoder"; -import { Readable } from "stream"; +import { Readable } from "node:stream"; export const defaultRetryOptions = { maxAttempts: 3, diff --git a/packages/core/tsup.config.ts b/packages/core/tsup.config.ts index 8d6389e24..b245c1803 100644 --- a/packages/core/tsup.config.ts +++ b/packages/core/tsup.config.ts @@ -17,4 +17,5 @@ export default defineConfig({ "./src/v3/workers/index.ts", "./src/v3/zodfetch.ts", ], + external: ["node:stream"], }); diff --git a/references/job-catalog/src/stressTest.ts b/references/job-catalog/src/stressTest.ts index 46689bf26..99603e90a 100644 --- a/references/job-catalog/src/stressTest.ts +++ b/references/job-catalog/src/stressTest.ts @@ -37,6 +37,29 @@ client.defineJob({ }, }); +client.defineJob({ + id: "stress-test-disabled", + name: "Stress Test Disabled", + version: "1.0.0", + trigger: eventTrigger({ + name: "stress.test.disabled", + }), + enabled: false, + run: async (payload, io, ctx) => { + await io.wait("wait-1", 20); + + await io.runTask( + `task-1`, + async (task) => { + await new Promise((resolve) => setTimeout(resolve, 10000)); + }, + { name: `Task 1` } + ); + + await io.wait("wait-2", 5); + }, +}); + client.defineJob({ id: "stress-test-2", name: "Stress Test 2",