Features: Batching and graphile-worker upgrade to 0.16.0-rc.1

.changeset/curvy-pens-cover.md

@@ -0,0 +1,9 @@
+---
+"@trigger.dev/airtable": patch
+"@trigger.dev/shopify": patch
+"@trigger.dev/sdk": patch
+"@trigger.dev/testing": patch
+"@trigger.dev/core": patch
+---
+
+Upgrade `graphile-worker` and add batch support

apps/webapp/app/components/code/CodeBlock.tsx

@@ -191,8 +191,8 @@ export const CodeBlock = forwardRef<HTMLDivElement, CodeBlockProps>(
       Array.from({ length: end - start + 1 }, (_, i) => start + i)
     );
 
-    // if there are more than 1000 lines, don't highlight
-    const shouldHighlight = lineCount <= 1000;
+    // if there are more than 1000 lines or 20_000 characters, don't highlight
+    const shouldHighlight = lineCount <= 1000 && code.length <= 20_000;
 
     return (
       <div

apps/webapp/app/components/run/RunOverview.tsx

@@ -183,7 +183,13 @@ export function RunOverview({ run, trigger, showRerun, paths }: RunOverviewProps
                 <RunPanelHeader icon={trigger.icon} title={trigger.title} />
                 <RunPanelBody>
                   <RunPanelProperties
-                    properties={[{ label: "Event name", text: run.event.name }]
+                    properties={[
+                      { label: "Event name", text: run.event.name },
+                      { label: "Batched", text: String(run.batched) },
+                      ...(run.batched
+                        ? [{ label: "Total Events", text: String(run.eventIds.length) }]
+                        : []),
+                    ]
                       .concat(
                         run.event.externalAccount
                           ? [{ label: "Account ID", text: run.event.externalAccount.identifier }]

apps/webapp/app/components/run/TriggerDetail.tsx

@@ -15,17 +15,25 @@ import { DisplayProperty } from "@trigger.dev/core";
 
 export function TriggerDetail({
   trigger,
+  payload,
+  context,
   event,
   properties,
+  batched = false,
+  eventIds,
 }: {
   trigger: DetailedEvent;
+  payload: string;
+  context: string;
   event: {
     title: string;
     icon: string;
   };
   properties: DisplayProperty[];
+  batched?: boolean;
+  eventIds?: string[];
 }) {
-  const { id, name, payload, context, timestamp, deliveredAt } = trigger;
+  const { id, name, timestamp, deliveredAt } = trigger;
 
   return (
     <RunPanel selected={false}>
@@ -45,7 +53,18 @@ export function TriggerDetail({
             />
           )}
           <RunPanelIconProperty icon="id" label="Event name" value={name} />
-          <RunPanelIconProperty icon="account" label="Event ID" value={id} />
+          {batched ? (
+            <>
+              <RunPanelIconProperty icon="packages" label="Batched" value={String(batched)} />
+              <RunPanelIconProperty
+                icon="hash"
+                label="Total Events"
+                value={eventIds ? eventIds.length : 0}
+              />
+            </>
+          ) : (
+            <RunPanelIconProperty icon="account" label="Event ID" value={id} />
+          )}
           {trigger.externalAccount && (
             <RunPanelIconProperty
               icon="account"
@@ -66,6 +85,12 @@ export function TriggerDetail({
           <CodeBlock code={payload} />
           <Header3>Context</Header3>
           <CodeBlock code={context} />
+          {batched && eventIds && (
+            <>
+              <Header3>Event IDs</Header3>
+              <CodeBlock code={JSON.stringify(eventIds, null, 2)} />
+            </>
+          )}
         </div>
       </RunPanelBody>
     </RunPanel>

apps/webapp/app/db.server.ts

@@ -69,23 +69,22 @@ export { Prisma };
 
 export const prisma = singleton("prisma", getClient);
 
+export const $replica: Omit<PrismaClient, "$transaction"> = singleton(
+  "replica",
+  () => getReplicaClient() ?? prisma
+);
+
 function getClient() {
   const { DATABASE_URL } = process.env;
   invariant(typeof DATABASE_URL === "string", "DATABASE_URL env var not set");
 
-  const databaseUrl = new URL(DATABASE_URL);
-
   // We need to add the connection_limit and pool_timeout query params to the url, in a way that works if the DATABASE_URL already has query params
-  const query = databaseUrl.searchParams;
-  query.set("connection_limit", env.DATABASE_CONNECTION_LIMIT.toString());
-  query.set("pool_timeout", env.DATABASE_POOL_TIMEOUT.toString());
-  databaseUrl.search = query.toString();
-
-  // Remove the username:password in the url and print that to the console
-  const urlWithoutCredentials = new URL(databaseUrl.href);
-  urlWithoutCredentials.password = "";
+  const databaseUrl = extendQueryParams(DATABASE_URL, {
+    connection_limit: env.DATABASE_CONNECTION_LIMIT.toString(),
+    pool_timeout: env.DATABASE_POOL_TIMEOUT.toString(),
+  });
 
-  console.log(`🔌 setting up prisma client to ${urlWithoutCredentials.toString()}`);
+  console.log(`🔌 setting up prisma client to ${redactUrlSecrets(databaseUrl.href)}`);
 
   const client = new PrismaClient({
     datasources: {
@@ -127,6 +126,67 @@ function getClient() {
   return client;
 }
 
+function getReplicaClient() {
+  if (!env.DATABASE_READ_REPLICA_URL) {
+    return;
+  }
+
+  const replicaUrl = extendQueryParams(env.DATABASE_READ_REPLICA_URL, {
+    connection_limit: env.DATABASE_CONNECTION_LIMIT.toString(),
+    pool_timeout: env.DATABASE_POOL_TIMEOUT.toString(),
+  });
+
+  console.log(`🔌 setting up read replica connection to ${redactUrlSecrets(replicaUrl)}`);
+
+  const replicaClient = new PrismaClient({
+    datasources: {
+      db: {
+        url: replicaUrl.href,
+      },
+    },
+    log: [
+      {
+        emit: "stdout",
+        level: "error",
+      },
+      {
+        emit: "stdout",
+        level: "info",
+      },
+      {
+        emit: "stdout",
+        level: "warn",
+      },
+    ],
+  });
+
+  // connect eagerly
+  replicaClient.$connect();
+
+  console.log(`🔌 read replica connected`);
+
+  return replicaClient;
+}
+
+function extendQueryParams(hrefOrUrl: string | URL, queryParams: Record<string, string>) {
+  const url = new URL(hrefOrUrl);
+  const query = url.searchParams;
+
+  for (const [key, val] of Object.entries(queryParams)) {
+    query.set(key, val);
+  }
+
+  url.search = query.toString();
+
+  return url;
+}
+
+function redactUrlSecrets(hrefOrUrl: string | URL) {
+  const url = new URL(hrefOrUrl);
+  url.password = "";
+  return url.href;
+}
+
 export type { PrismaClient } from "@trigger.dev/database";
 
 export const PrismaErrorSchema = z.object({

apps/webapp/app/entry.server.tsx

@@ -190,7 +190,13 @@ function logError(error: unknown, request?: Request) {
       );
     }
   }
+
   console.error(error);
+
+  if (error instanceof Error && error.message.startsWith("There are locked jobs present")) {
+    console.log("⚠️  graphile-worker migration issue detected!");
+    console.log("⚠️  see: https://trigger.dev/docs/documentation/guides/self-hosting/graphile-migration");
+  }
 }
 
 const sqsEventConsumer = singleton("sqsEventConsumer", getSharedSqsEventConsumer);

apps/webapp/app/env.server.ts

@@ -5,6 +5,7 @@ import { isValidRegex } from "./utils/regex";
 const EnvironmentSchema = z.object({
   NODE_ENV: z.union([z.literal("development"), z.literal("production"), z.literal("test")]),
   DATABASE_URL: z.string(),
+  DATABASE_READ_REPLICA_URL: z.string().optional(),
   DATABASE_CONNECTION_LIMIT: z.coerce.number().int().default(10),
   DATABASE_POOL_TIMEOUT: z.coerce.number().int().default(60),
   DIRECT_URL: z.string(),