feat: realtime (#1402)

* Denormalize run tags, increase character limit to 128

* WIP realtime subscribing to runs

* extracted the stream stuff into core, made it more reusable

* WIP tags

* Remove tags for now because it’s not support in electric

* Support async iterables, readable stream, and callback style subscription styles

* Remove tags streaming endpoint

* Add realtime rate limits and scope them to the /realtime path

* WIP rate limt per org

* Introduce per org rate limits

* WIP JWT auth

* Move migrations into new internal db package

* Resolve pnpm lock file

* Authenticating to the realtime API with JWTs are working

* realtime in the client

* Created react-hooks package and starting to move stuff in there

* Improve types for hooks

* schema tasks

* Added useBatch hook

* build uploadthing/fal demo and change how run metadata is synced to the server

* tweaks

* WIL realtime concurrency tracking

* Implement test for realtime client using testcontainers

also updated electric to latest version

* Allow customizing the expiration time of the automatic JWT created after triggering a task

* Add support for subscribing to run tags

* Improve auth types and API

* finalize the realtime API

* Fixed some example stuff

* Allow up to 10 run tags

* Remove core from docker-provider tsconfig paths to prevent it from being typechecked

* do the same for the kubernetes provider

* Fixing some typecheck errors

* Fix webapp type errors

* Update @trigger.dev/platform to 1.0.13

* Fix attw error

* Remove from/to in subscribeToRuns query params

* Add tests for the rate limit middleware and add custom JWT rate limits

* turn off webapp test parallelism

* Finish renaming jwt -> publicAccessToken and automatically give the JWT read access to the tags when using trigger

* Add changeset

* Attempt to fix unit tests in CI

* Skip running the auth rate limit middleware tests for now

* Try a beefier machine

* Try and run webapp tests separately

* Setup env vars

* Make sliding window test more reliabile

Author: ericallam

.changeset/brave-forks-compare.md

@@ -0,0 +1,7 @@
+---
+"@trigger.dev/react-hooks": minor
+"@trigger.dev/sdk": minor
+"@trigger.dev/core": minor
+---
+
+Access run status updates in realtime, from your server or from your frontend

.github/workflows/release.yml

@@ -17,7 +17,7 @@ concurrency:
 jobs:
   release:
     name: 🦋 Changesets Release
-    runs-on: buildjet-8vcpu-ubuntu-2204
+    runs-on: ubuntu-latest
     if: github.repository == 'triggerdotdev/trigger.dev'
     outputs:
       published: ${{ steps.changesets.outputs.published }}

.github/workflows/unit-tests.yml

@@ -6,7 +6,7 @@ on:
 jobs:
   unitTests:
     name: "🧪 Unit Tests"
-    runs-on: buildjet-8vcpu-ubuntu-2204
+    runs-on: buildjet-16vcpu-ubuntu-2204
     steps:
       - name: ⬇️ Checkout repo
         uses: actions/checkout@v4
@@ -30,5 +30,15 @@ jobs:
       - name: 📀 Generate Prisma Client
         run: pnpm run generate
 
-      - name: 🧪 Run Unit Tests
-        run: pnpm run test
+      - name: 🧪 Run Webapp Unit Tests
+        run: pnpm run test --filter webapp
+        env:
+          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres
+          DIRECT_URL: postgresql://postgres:postgres@localhost:5432/postgres
+          SESSION_SECRET: "secret"
+          MAGIC_LINK_SECRET: "secret"
+          ENCRYPTION_KEY: "secret"
+
+      
+      - name: 🧪 Run Internal Unit Tests
+        run: pnpm run test --filter "@internal/*"

.npmrc

@@ -1,2 +1,3 @@
 link-workspace-packages=false
-public-hoist-pattern[]=*prisma*
+public-hoist-pattern[]=*prisma*
+prefer-workspace-packages=true

.vscode/extensions.json

@@ -1,8 +1,4 @@
 {
-	"recommendations": [
-        "denoland.vscode-deno"
-    ],
-	"unwantedRecommendations": [
-		
-	]
+  "recommendations": ["bierner.comment-tagged-templates"],
+  "unwantedRecommendations": []
 }

apps/docker-provider/tsconfig.json

@@ -6,10 +6,6 @@
     "forceConsistentCasingInFileNames": true,
     "resolveJsonModule": true,
     "strict": true,
-    "skipLibCheck": true,
-    "paths": {
-      "@trigger.dev/core/v3": ["../../packages/core/src/v3"],
-      "@trigger.dev/core/v3/*": ["../../packages/core/src/v3/*"]
-    }
+    "skipLibCheck": true
   }
 }

apps/kubernetes-provider/tsconfig.json

@@ -6,10 +6,6 @@
     "forceConsistentCasingInFileNames": true,
     "resolveJsonModule": true,
     "strict": true,
-    "skipLibCheck": true,
-    "paths": {
-      "@trigger.dev/core/v3": ["../../packages/core/src/v3"],
-      "@trigger.dev/core/v3/*": ["../../packages/core/src/v3/*"]
-    }
+    "skipLibCheck": true
   }
 }

apps/webapp/app/env.server.ts

@@ -31,7 +31,7 @@ const EnvironmentSchema = z.object({
   REMIX_APP_PORT: z.string().optional(),
   LOGIN_ORIGIN: z.string().default("http://localhost:3030"),
   APP_ORIGIN: z.string().default("http://localhost:3030"),
-  ELECTRIC_ORIGIN: z.string(),
+  ELECTRIC_ORIGIN: z.string().default("http://localhost:3060"),
   APP_ENV: z.string().default(process.env.NODE_ENV),
   SERVICE_NAME: z.string().default("trigger.dev webapp"),
   SECRET_STORE: SecretStoreOptionsSchema.default("DATABASE"),
@@ -103,6 +103,25 @@ const EnvironmentSchema = z.object({
   API_RATE_LIMIT_REFILL_RATE: z.coerce.number().int().default(250), // refix 250 tokens every 10 seconds
   API_RATE_LIMIT_REQUEST_LOGS_ENABLED: z.string().default("0"),
   API_RATE_LIMIT_REJECTION_LOGS_ENABLED: z.string().default("1"),
+  API_RATE_LIMIT_LIMITER_LOGS_ENABLED: z.string().default("0"),
+
+  API_RATE_LIMIT_JWT_WINDOW: z.string().default("1m"),
+  API_RATE_LIMIT_JWT_TOKENS: z.coerce.number().int().default(60),
+
+  //Realtime rate limiting
+  /**
+   * @example "60s"
+   * @example "1m"
+   * @example "1h"
+   * @example "1d"
+   * @example "1000ms"
+   * @example "1000s"
+   */
+  REALTIME_RATE_LIMIT_WINDOW: z.string().default("1m"),
+  REALTIME_RATE_LIMIT_TOKENS: z.coerce.number().int().default(100),
+  REALTIME_RATE_LIMIT_REQUEST_LOGS_ENABLED: z.string().default("0"),
+  REALTIME_RATE_LIMIT_REJECTION_LOGS_ENABLED: z.string().default("1"),
+  REALTIME_RATE_LIMIT_LIMITER_LOGS_ENABLED: z.string().default("0"),
 
   //Ingesting event rate limit
   INGEST_EVENT_RATE_LIMIT_WINDOW: z.string().default("60s"),

apps/webapp/app/models/taskRunTag.server.ts

@@ -1,7 +1,7 @@
 import { prisma } from "~/db.server";
 import { generateFriendlyId } from "~/v3/friendlyIdentifiers";
 
-export const MAX_TAGS_PER_RUN = 5;
+export const MAX_TAGS_PER_RUN = 10;
 
 export async function createTag({ tag, projectId }: { tag: string; projectId: string }) {
   if (tag.trim().length === 0) return;

apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts

@@ -62,8 +62,7 @@ type CommonRelatedRun = Prisma.Result<
 export class ApiRetrieveRunPresenter extends BasePresenter {
   public async call(
     friendlyId: string,
-    env: AuthenticatedEnvironment,
-    showSecretDetails: boolean
+    env: AuthenticatedEnvironment
   ): Promise<RetrieveRunResponse | undefined> {
     return this.traceWithEnv("call", env, async (span) => {
       const taskRun = await this._replica.taskRun.findFirst({
@@ -72,11 +71,7 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
           runtimeEnvironmentId: env.id,
         },
         include: {
-          attempts: {
-            orderBy: {
-              createdAt: "desc",
-            },
-          },
+          attempts: true,
           lockedToVersion: true,
           schedule: true,
           tags: true,
@@ -111,50 +106,48 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
       let $output: any;
       let $outputPresignedUrl: string | undefined;
 
-      if (showSecretDetails) {
-        const payloadPacket = await conditionallyImportPacket({
-          data: taskRun.payload,
-          dataType: taskRun.payloadType,
-        });
+      const payloadPacket = await conditionallyImportPacket({
+        data: taskRun.payload,
+        dataType: taskRun.payloadType,
+      });
 
-        if (
-          payloadPacket.dataType === "application/store" &&
-          typeof payloadPacket.data === "string"
-        ) {
-          $payloadPresignedUrl = await generatePresignedUrl(
-            env.project.externalRef,
-            env.slug,
-            payloadPacket.data,
-            "GET"
-          );
-        } else {
-          $payload = await parsePacket(payloadPacket);
-        }
+      if (
+        payloadPacket.dataType === "application/store" &&
+        typeof payloadPacket.data === "string"
+      ) {
+        $payloadPresignedUrl = await generatePresignedUrl(
+          env.project.externalRef,
+          env.slug,
+          payloadPacket.data,
+          "GET"
+        );
+      } else {
+        $payload = await parsePacket(payloadPacket);
+      }
 
-        if (taskRun.status === "COMPLETED_SUCCESSFULLY") {
-          const completedAttempt = taskRun.attempts.find(
-            (a) => a.status === "COMPLETED" && typeof a.output !== null
-          );
+      if (taskRun.status === "COMPLETED_SUCCESSFULLY") {
+        const completedAttempt = taskRun.attempts.find(
+          (a) => a.status === "COMPLETED" && typeof a.output !== null
+        );
 
-          if (completedAttempt && completedAttempt.output) {
-            const outputPacket = await conditionallyImportPacket({
-              data: completedAttempt.output,
-              dataType: completedAttempt.outputType,
-            });
+        if (completedAttempt && completedAttempt.output) {
+          const outputPacket = await conditionallyImportPacket({
+            data: completedAttempt.output,
+            dataType: completedAttempt.outputType,
+          });
 
-            if (
-              outputPacket.dataType === "application/store" &&
-              typeof outputPacket.data === "string"
-            ) {
-              $outputPresignedUrl = await generatePresignedUrl(
-                env.project.externalRef,
-                env.slug,
-                outputPacket.data,
-                "GET"
-              );
-            } else {
-              $output = await parsePacket(outputPacket);
-            }
+          if (
+            outputPacket.dataType === "application/store" &&
+            typeof outputPacket.data === "string"
+          ) {
+            $outputPresignedUrl = await generatePresignedUrl(
+              env.project.externalRef,
+              env.slug,
+              outputPacket.data,
+              "GET"
+            );
+          } else {
+            $output = await parsePacket(outputPacket);
           }
         }
       }
@@ -165,6 +158,7 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
         payloadPresignedUrl: $payloadPresignedUrl,
         output: $output,
         outputPresignedUrl: $outputPresignedUrl,
+        error: ApiRetrieveRunPresenter.apiErrorFromError(taskRun.error),
         schedule: taskRun.schedule
           ? {
               id: taskRun.schedule.friendlyId,
@@ -179,17 +173,9 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
               },
             }
           : undefined,
-        attempts: !showSecretDetails
-          ? []
-          : taskRun.attempts.map((a) => ({
-              id: a.friendlyId,
-              status: ApiRetrieveRunPresenter.apiStatusFromAttemptStatus(a.status),
-              createdAt: a.createdAt ?? undefined,
-              updatedAt: a.updatedAt ?? undefined,
-              startedAt: a.startedAt ?? undefined,
-              completedAt: a.completedAt ?? undefined,
-              error: ApiRetrieveRunPresenter.apiErrorFromError(a.error),
-            })),
+        // We're removing attempts from the API
+        attemptCount: taskRun.attempts.length,
+        attempts: [],
         relatedRuns: {
           root: taskRun.rootTaskRun
             ? await createCommonRunStructure(taskRun.rootTaskRun)