Merge branch 'main' into v3/worker-attempt-creation

Author: nicktrn

.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:

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;

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<typeof EnvironmentSchema>;

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<string, string | number | boolean>;
+  }) {
+    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;

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);
 

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: {

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,

packages/core/tsup.config.ts

@@ -17,4 +17,5 @@ export default defineConfig({
     "./src/v3/workers/index.ts",
     "./src/v3/zodfetch.ts",
   ],
+  external: ["node:stream"],
 });

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",