Merge branch 'main' into main

Author: Rishi Raj Jain

.changeset/lemon-clocks-love.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/cli": patch
+---
+
+Added tests to the CLI, updated tsconfig file accordingly

CONTRIBUTING.md

@@ -27,7 +27,7 @@ branch are tagged into a release monthly.
    ```
 
    > If you are on windows, run the following command on gitbash with admin privileges:
-   > `git clone -c core.symlinks=true https://triggerdotdev/trigger.dev.git`
+   > `git clone -c core.symlinks=true https://github.com/triggerdotdev/trigger.dev.git`
 
 2. Navigate to the project folder
    ```

apps/webapp/app/routes/invitation-code.tsx

@@ -0,0 +1,67 @@
+import { LoaderArgs } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { logger } from "~/services/logger.server";
+import { sse } from "~/utils/sse";
+
+export async function loader({ request }: LoaderArgs) {
+  const url = new URL(request.url);
+  const searchParams = Object.fromEntries(url.searchParams.entries());
+
+  const options = z
+    .object({
+      minDelay: z.coerce.number().int(),
+      maxDelay: z.coerce.number().int(),
+      undefinedProbability: z.coerce.number().min(0).max(1).default(0.1),
+    })
+    .parse(searchParams);
+
+  logger.debug("Test SSE stream", { options });
+
+  let lastSignals = calculateChangeSignals(Date.now());
+
+  return sse({
+    request,
+    run: async (send, stop) => {
+      const result = await dateForUpdates(options);
+
+      if (!result) {
+        return stop();
+      }
+
+      const newSignals = calculateChangeSignals(result);
+
+      if (lastSignals.ts !== newSignals.ts) {
+        send({ data: JSON.stringify(newSignals) });
+      }
+
+      lastSignals = newSignals;
+    },
+  });
+}
+
+async function dateForUpdates(opts: {
+  minDelay: number;
+  maxDelay: number;
+  undefinedProbability: number;
+}): Promise<number | undefined> {
+  // Randomly await between minDelay and maxDelay
+  await new Promise((resolve) => {
+    setTimeout(resolve, Math.random() * (opts.maxDelay - opts.minDelay) + opts.minDelay);
+  });
+
+  // There should be about a x% chance that this returns undefined
+  if (Math.random() < opts.undefinedProbability) {
+    logger.debug("Test SSE dataForUpdates returning undefined");
+
+    return undefined;
+  }
+
+  // Randomly return true or false
+  return Date.now();
+}
+
+function calculateChangeSignals(ts: number) {
+  return {
+    ts,
+  };
+}

apps/webapp/app/routes/tests.sse.stream.ts

@@ -0,0 +1,37 @@
+import { useLoaderData } from "@remix-run/react";
+import { LoaderArgs } from "@remix-run/server-runtime";
+import { useEventSource } from "remix-utils";
+import { z } from "zod";
+
+export async function loader({ request }: LoaderArgs) {
+  const url = new URL(request.url);
+  const params = Object.fromEntries(url.searchParams.entries());
+
+  const config = z
+    .object({
+      minDelay: z.coerce.number().int().min(0).max(10000).default(1000),
+      maxDelay: z.coerce.number().int().min(0).max(10000).default(2000),
+      undefinedProbability: z.coerce.number().min(0).max(1).default(0.1),
+    })
+    .parse(params);
+
+  return config;
+}
+
+export default function SSETest() {
+  const { minDelay, maxDelay, undefinedProbability } = useLoaderData<typeof loader>();
+
+  const events = useEventSource(
+    `/tests/sse/stream?minDelay=${minDelay}&maxDelay=${maxDelay}&undefinedProbability=${undefinedProbability}`,
+    {
+      event: "message",
+    }
+  );
+
+  return (
+    <div>
+      <h2>SSE Test</h2>
+      <p>{events ?? "No events"}</p>
+    </div>
+  );
+}

apps/webapp/app/routes/tests.sse.tsx

@@ -436,7 +436,7 @@ export class RegisterJobService {
             },
           },
           create: {
-            event: SCHEDULED_EVENT,
+            event: [SCHEDULED_EVENT],
             source: "trigger.dev",
             payloadFilter: {},
             contextFilter: {},

apps/webapp/app/services/jobs/registerJob.server.ts

@@ -79,7 +79,7 @@ export class RegisterDynamicScheduleService {
         },
       },
       create: {
-        event: SCHEDULED_EVENT,
+        event: [SCHEDULED_EVENT],
         source: "trigger.dev",
         payloadFilter: {},
         contextFilter: {},

apps/webapp/app/services/triggers/registerDynamicSchedule.server.ts

@@ -76,7 +76,7 @@ export class RegisterTriggerSourceService {
           create: {
             dispatchableId: triggerSource.id,
             environmentId: environment.id,
-            event: payload.rule.event,
+            event: Array.isArray(payload.rule.event) ? payload.rule.event : [payload.rule.event], 
             source: payload.rule.source,
             payloadFilter: payload.rule.payload,
             contextFilter: payload.rule.context,
@@ -86,7 +86,7 @@ export class RegisterTriggerSourceService {
             },
           },
           update: {
-            event: payload.rule.event,
+            event: Array.isArray(payload.rule.event) ? payload.rule.event : [payload.rule.event],
             source: payload.rule.source,
             payloadFilter: payload.rule.payload,
             contextFilter: payload.rule.context,

apps/webapp/app/services/triggers/registerTriggerSource.server.ts

@@ -73,10 +73,6 @@ export function confirmBasicDetailsPath() {
   return `/confirm-basic-details`;
 }
 
-export function invitationCodePath() {
-  return `/invitation-code`;
-}
-
 export function acceptInvitePath(token: string) {
   return `/invite-accept?token=${token}`;
 }

apps/webapp/app/utils/pathBuilder.ts

@@ -1,4 +1,5 @@
 import { eventStream } from "remix-utils";
+import { logger } from "~/services/logger.server";
 
 type SseProps = {
   request: Request;
@@ -30,12 +31,38 @@ export function sse({ request, pingInterval = 1000, updateInterval = 348, run }:
   };
 
   return eventStream(request.signal, (send) => {
+    const safeSend = (args: { event?: string; data: string }) => {
+      try {
+        send(args);
+      } catch (error) {
+        if (error instanceof Error) {
+          if (error.name !== "TypeError") {
+            logger.debug("Error sending SSE, aborting", {
+              error: {
+                name: error.name,
+                message: error.message,
+                stack: error.stack,
+              },
+              args,
+            });
+          }
+        } else {
+          logger.debug("Uknown error sending SSE, aborting", {
+            error,
+            args,
+          });
+        }
+
+        abort();
+      }
+    };
+
     pinger = setInterval(() => {
-      send({ event: "ping", data: new Date().toISOString() });
+      safeSend({ event: "ping", data: new Date().toISOString() });
     }, pingInterval);
 
     updater = setInterval(async () => {
-      run(send, abort);
+      run(safeSend, abort);
     }, updateInterval);
 
     return abort;