Forem: James Ritchie

How we built a real-time service that handles 20,000 updates per second

James Ritchie — Thu, 17 Jul 2025 11:19:23 +0000

Building production-grade AI agents requires more than just calling OpenAI's API. You need long-running tasks, durability, queues, and, crucially, the ability to show your users what's happening in real-time. That last piece turned out to be harder than we thought.

When we set out to build Trigger.dev's Realtime service, we knew we wanted something that could handle serious scale. We ended up with a system that processes over 20,000 run updates per second and ingests over 500GB of data daily from Postgres. But getting there required some unexpected technical decisions.

Here's how we built it, what worked, what didn't, and why we ended up using Postgres replication over WebSockets.

The problem: Making background jobs feel foreground

Trigger.dev helps developers build reliable background tasks, from AI agents and video generation to data pipelines and automation workflows. But "background" doesn't mean "invisible." Users want to see progress bars, get real-time updates, and know their jobs are actually working.

Consider a simple app that processes CSVs.

You upload a 10,000-row file, and our system spawns tasks to process each row in parallel across many machines. Without real-time updates, users stare at a loading spinner for minutes, wondering if anything is happening. With Realtime, they see progress bars updating as each row gets processed, metadata flowing in, and confidence that their job is working.

Here's how to write the task that processes each row:

/**
 * This task is triggered for every CSV row
 */
export const handleCSVRow = task({
  id: "handle-csv-row",
  run: async (row: CSVRow) => {
    logger.info("Handling CSV row", { row });

    const result = await processRow(row);

    // Update the parent metadata
    metadata.parent.increment("totalProcessed", 1);

    // Update this task's metadata
    metadata.set("result", result);

    return { result };
  },
});

On the frontend, developers can subscribe to these changes using our React hook:

export function CSVProgress({
  runId,
  accessToken,
}: {
  runId: string;
  accessToken: string;
}) {
  const { run } = useRealtimeRun<typeof csvProcessor>(runId, {
    accessToken,
  });

  const processed = run?.metadata.get("totalProcessed") ?? 0;
  const total = run?.metadata.get("total") ?? 1;

  return <span>{Math.round((processed / total) * 100)}%</span>;
}

Architecture: Why we chose Postgres replication slots

Our first instinct was WebSockets. Everyone uses WebSockets for real-time, right? But we quickly hit on three key problems:

Historical subscriptions: We want you to be able to subscribe to runs that started before you opened the page.
Database load: Every WebSocket connection needs to query the database for initial state, then maintain a separate connection for updates.
Notification complexity: WebSockets need a separate system to know when data changes. You have to use database triggers, message queues, or polling mechanisms to notify connected clients about updates. This is challenging in a high throughput system.

Then we discovered Electric which uses Postgres replication slots. This solved these problems, kept one source of truth, and reduced database load.

Here's how it works:

Updating a run

When a task calls metadata.set("foo", "bar"), it updates the runs table in Postgres
Postgres writes this change to its Write-Ahead Log (WAL)

Subscribing to a run

An end-user loads a web page which uses Trigger.dev Realtime.
An HTTP request is sent to Trigger.dev to authenticate them (using JWT).
If authenticated we create a query for Electric to process.
ElectricSQL creates a new “Shape” (if no query param was passed) and does an initial query for the run to Postgres.
The returned row includes a transaction ID (which is sent back and used as a cursor from now on).

From now on, the client sends the shape and transaction IDs in all requests. When they're present, ElectricSQL knows it's in "live mode" and sends new changes only for that shape.

The HTTP request is a long poll which returns a new row when there's a change or 20s have passed. When the Response is received the client will do another request (unless the run is completed).

How Postgres replication slots actually work

Let's dig into how replication slots work:

A SQL query is made to the database (that modifies data).
Postgres parses the SQL, plans, then executes the query.
Data changes are cached in shared buffers.
Changes are appended to the Write-Ahead Log (WAL).
The replication slot acts as a persistent bookmark, tracking the latest WAL position processed for that slot.
ElectricSQL connects and streams changes, moving the cursor forward.
We get a reliable, ordered stream of every database change.

Combining an initial query, then subsequent long-polling requests using the replication stream we get a reliable live view of the data.

Authentication: JWTs all the way down

For auth, we use signed JSON Web Tokens instead of hitting the database on every request. This was crucial for performance.

const publicToken = await auth.createPublicToken({
  scopes: {
    read: {
      runs: ["run_1234", "run_5678"], // ✅ this token can read only these runs
    },
  },
});

The JWT is signed with our API key and includes permission scopes—whether the user can trigger tasks, read specific runs, or subscribe to runs with certain tags.

The beauty of this is that no database query is required to verify permissions. If the JWT is valid, we know they have permission.

Rate limiting: Redis Lua scripts

We limit our Realtime connections based on subscription plans. 500 concurrent connections by default on the Pro plan, for example.

We use a sliding window algorithm implemented as a Redis Lua script to achieve this.

local concurrencyKey = KEYS[1]

local timestamp = tonumber(ARGV[1])
local requestId = ARGV[2]
local expiryTime = tonumber(ARGV[3])
local cutoffTime = tonumber(ARGV[4])
local limit = tonumber(ARGV[5])

-- Remove expired entries
redis.call('ZREMRANGEBYSCORE', concurrencyKey, '-inf', cutoffTime)

-- Add the new request to the sorted set
redis.call('ZADD', concurrencyKey, timestamp, requestId)

-- Set the expiry time on the key
redis.call('EXPIRE', concurrencyKey, expiryTime)

-- Get the total number of concurrent requests
local totalRequests = redis.call('ZCARD', concurrencyKey)

-- Check if the limit has been exceeded
if totalRequests > limit then
  -- Remove the request we just added
  redis.call('ZREM', concurrencyKey, requestId)
  return 0
end

-- Return 1 to indicate success
return 1

Each request expires after 5 minutes. We count active connections in a sliding window and reject new ones if they'd exceed the limit. Lua scripts run atomically in Redis, so we don't have race conditions.

Time filtering and caching challenges

One tricky problem: developers want to subscribe to "all runs created in the last hour" or "all runs with tag 'org_1234' from today." But ElectricSQL Shapes need fixed WHERE clauses. If the createdAt filter changes on every React render, we'd create infinite shapes and destroy performance.

Our solution: cache the time filters against the shape ID.

const cache = createCache({
  createdAtFilter: new Namespace<string>(ctx, {
    stores: [memory, redisCacheStore],
    fresh: 60_000 * 60 * 24 * 7, // 1 week
    stale: 60_000 * 60 * 24 * 14, // 2 weeks
  }),
});

const createdAtFilterCacheResult = await this.cache.createdAtFilter.get(
  shapeId
);

We're using the excellent @unkey/cache package, which gives us memory + Redis caching with automatic freshness guarantees. This lets us have many servers running the same caching logic without stepping on each other.

Performance: The numbers that matter

After months of optimization, here's what we achieved:

20,000 updates per second during peak loads
500GB+ of Postgres inserts processed daily
Sub-100ms latency for real-time updates to reach browsers

The Postgres + ElectricSQL architecture scales surprisingly well. Postgres handles the write load because that's what it's designed for. ElectricSQL handles the fan-out to browsers.

Why ElectricSQL over alternatives

We evaluated several approaches but ElectricSQL solved our core problem: we already needed the data in Postgres for persistence. Why not use Postgres as the source of truth for real-time updates too?

The WAL-based approach means:

Users can subscribe to any run after the fact
We get strong consistency guarantees
Operational complexity stays low
We minimize load on our primary database

Want to try Trigger.dev Realtime? Check out our Realtime docs and run through our React hooks guide.

Self-hosting Trigger.dev v4 using Kubernetes

James Ritchie — Thu, 26 Jun 2025 16:59:56 +0000

Trigger.dev v4 brings self-hosting capabilities to Kubernetes environments. Alongside our Docker option, we now offer native Kubernetes deployment through our official Helm chart, which makes running the self-hosted version of Trigger.dev at scale much simpler.

This post focuses on self-hosting with Kubernetes using Helm, great for teams who need the orchestration, scaling, and reliability features that Kubernetes provides. If you're looking for a simpler setup, check out our Docker compose guide.

In this post, I'll walk you through:

Why you might want to self-host on Kubernetes
What's new in v4 for Kubernetes self-hosters
What to expect
A brief Kubernetes self-hosting overview

Why self-host Trigger.dev on Kubernetes?

For the majority of use cases, Trigger.dev Cloud provides the best experience. It's completely managed, handles scaling automatically, and includes dedicated support. However, there are situations where you might be required to self-host. If this is the case, Kubernetes is a great option.

When does Kubernetes self-hosting make sense?

You're already running Kubernetes infrastructure and want consistent deployment patterns.
Your organization has strict compliance needs (e.g. data residency) requiring Kubernetes-native security controls.
You need to deploy Trigger.dev in an air-gapped or private Kubernetes environment.
You're developing solutions that require tight integration with existing Kubernetes tooling and workflows.

When isn't it the right choice?

You're unfamiliar with Kubernetes and prefer the most straightforward setup (use Docker instead or try the Cloud).
Your team doesn't have Kubernetes experience and you want to avoid extra operational complexity.
You prefer the quickest route to production without managing Kubernetes infrastructure.

Cloud vs. self-hosted v4:

Here's how Cloud and self-hosted v4 compare feature-wise:

Feature	Cloud	Self-hosted
Warm starts	✅	❌
Auto-scaling	✅	❌
Checkpoints	✅	❌
Dedicated support	✅	❌
Community support	✅	✅
ARM support	✅	✅

What's new in v4 for Kubernetes self-hosting?

Key improvements from v3 to v4:

Helm chart with sensible defaults – eliminates the need for custom Kubernetes manifests.
Integrated Postgres, Redis, and object storage, removing the requirement to configure external services unless desired.
Streamlined, more reliable worker management using Kubernetes-native scaling capabilities.

Deployment is simplified:

Helm chart deployment is simple and straightforward.
Environment variables have improved documentation and consistency across deployment methods.
Resource limits and configurations follow standard Kubernetes conventions.

The Kubernetes self-hosting experience: what to expect

Deploying Trigger.dev on Kubernetes offers significant flexibility, but it brings additional responsibilities and considerations. Here's what to expect when choosing the Kubernetes self-hosted path:

Your responsibilities include:

Setting up and maintaining your Kubernetes cluster (EKS, GKE, AKS, or self-managed).
Managing Helm chart updates, security patches, and scaling configurations.
Overseeing cluster and application uptime and performance monitoring.

What comes included:

Helm chart containing all necessary components.
ClickHouse, Postgres, Redis, and object storage with external service support.
Complete core Trigger.dev functionality, excluding some Cloud-exclusive features.

What's missing:

Fully managed cluster scaling (though Kubernetes provides scaling capabilities).
Warm starts.
Checkpoints (non-blocking waits), meaning longer waits consume more resources.
Kubernetes expertise is your responsibility, though our self-hosting Discord community is very active and can offer great advice.

Premium support:

We do offer enterprise support for Kubernetes self-hosters. Get in touch if that's something you're interested in.

Kubernetes self-hosting overview

Self-hosting Trigger.dev v4 on Kubernetes is now available for the first time, though it still requires Kubernetes knowledge and operational experience. Here's what you need to understand before getting started:

Prerequisites

Kubernetes cluster: A cluster with version 1.19+ and Helm 3.8+. Works with managed services (EKS, GKE, AKS) or self-managed clusters.
Resources: At least 6+ vCPU and 12+ GB RAM across your cluster, plus persistent volume capability.
Database: Postgres comes with the Helm chart, though external managed databases are supported if preferred.
Email provider: For magic link authentication, configure SMTP using a service like Resend.

Scaling considerations

Begin with default resource allocations, then plan expansion based on your workload patterns.
Consider leveraging external managed services (RDS, Redis Cloud) for enhanced production reliability.
Track resource consumption and modify CPU/memory limits through Helm values as requirements change.

Best practices & gotchas

Version locking: Always pin your Helm chart version to avoid surprises when updating.
Configuration: Use separate values.yaml files for different environments.
Email setup: Test your magic link email flow early to avoid login headaches. Check the webapp pod logs for any email configuration errors.
GitHub auth: You can use GitHub authentication as an alternative to email.
Security: Use Kubernetes network policies to secure your deployment. Don't expose services publicly without proper authentication.

Operations & maintenance

Updates are simple: bump your Helm chart version and run helm upgrade.
Should you find yourself investing more time in Kubernetes operations than development, you can always migrate to Trigger.dev Cloud.

Get started with Kubernetes self-hosting

Follow the Kubernetes self-hosting guide to get started.
Join the Discord self-hosting channel for support.

Self-hosting Trigger.dev v4 using Docker

James Ritchie — Thu, 19 Jun 2025 12:35:52 +0000

With Trigger.dev v4, we took everything we learned from the v3 self-hosting experience and rebuilt the process to be as smooth as possible. Our goal: you shouldn’t need to be a DevOps expert just to get up and running. If you want to run Trigger.dev yourself, it should feel approachable, well-documented, and reliable.

This post focuses on self-hosting with Docker, which is the easiest way to get started. If you’re interested in Kubernetes, we’ll have a guide for that soon as well.

In this post, I'll walk you through:

Why you might want to self-host
What’s new in v4 for self-hosters
What to expect
A brief self-hosting overview

Why self-host Trigger.dev?

For most people, I think I can safely say that using Trigger.dev Cloud is the best experience. It's fully managed, scales automatically, and you get dedicated support. But sometimes, you need more control.

When does self-hosting make sense?

You have strict compliance requirements (e.g. data residency).
You want to run Trigger.dev in an air-gapped or private environment.
You're building something that needs deep integration with your own infra or custom limits.

When doesn't it?

You don’t want to manage updates, scaling, or security patches.
You want the fastest path to production and don’t want to manage infra.

Cloud vs. self-hosted v4:

Here's a quick comparison of the features available in Cloud and self-hosted v4. I go into more detail below.

Feature	Cloud	Self-hosted
Warm starts	✅	❌
Auto-scaling	✅	❌
Checkpoints	✅	❌
Dedicated support	✅	❌
Community support	✅	✅
ARM support	✅	✅

What’s new in v4 for self-hosting?

Major changes from v3 to v4:

No more custom startup scripts – just simple Docker Compose commands.
Built-in registry and object storage, so you don’t need to wire up S3 or GCS unless you want to.
Simpler, more robust worker management.

Setup is easier:

The Docker Compose setup is streamlined. Fewer moving parts, less YAML to wrangle.
Environment variables for the webapp and supervisor are better documented and more consistent.

Horizontal scaling for workers:

You can now scale out workers easily by adding more containers.
Resource limits and security are improved – less risk of noisy neighbors.

The self-hosting experience: what to expect

Running Trigger.dev yourself gives you a lot of flexibility, but it also comes with new responsibilities and trade-offs. Here’s what you can expect if you go the self-hosted route:

You’re responsible for:

Provisioning and maintaining infrastructure (servers, storage, networking).
Handling updates, security patches, and scaling.
Monitoring uptime and performance.

What’s provided out of the box:

A ready-to-run Docker Compose setup.
Built-in registry and object storage.
All the core features of Trigger.dev, minus a few Cloud-only perks.

What’s not included:

Managed scaling.
Warm starts.
Checkpoints (non-blocking waits), so long waits use more resources.
You’re on your own for infra, but the Discord community is active and helpful.

Premium support:

We offer enterprise support for self-hosters. Get in touch if that's something you're interested in.

Self-hosting overview

Self-hosting Trigger.dev v4 is more approachable than ever, but it does ask for a bit more hands-on involvement. Here’s what you should know before you dive in:

What you’ll need

Modern infrastructure: A server or VM with Docker and Docker Compose. Linux is preferred for production, but you can experiment locally on macOS or Windows.
Database: Postgres is included in the default Docker Compose stack, but you can use your own (self-hosted or managed) if you prefer.
Email provider: For magic link logins, set up SMTP credentials using a provider such as Resend.

Planning for scale

Start small, but plan for growth. You can add more worker containers as your workload increases.
For production, consider splitting webapp and workers across different machines or VMs for reliability.
Monitor resource usage and be ready to adjust CPU/memory as needed.

Best practices & gotchas

Version locking: Always pin your Docker images to a specific version tag to avoid surprises when updating.
Environment variables: Double-check your .env file – misconfigurations are the #1 source of setup pain. For local testing, things should work out of the box with the provided defaults.
Email setup: Test your magic link email flow early to avoid login headaches. You can check the webapp logs for any errors when no email config is setup.
GitHub auth: You can now use GitHub authentication as an alternative to email.
Security: Don’t expose your webapp or registry to the public internet without authentication. Use strong secrets and keep everything patched.

Operating & upgrading

Scaling is as simple as adding more worker containers.
Upgrades are straightforward: update your image tags and restart your containers.
If you ever feel like you’re spending more time on ops than building, remember you can always move back to Trigger.dev Cloud.

Get started with self-hosting

Follow the Self-hosting Docker guide to get started.
Join the Discord self-hosting channel for support.

Building effective AI agents with Trigger.dev

James Ritchie — Thu, 06 Feb 2025 18:35:47 +0000

Anthropic published a thoughtful blog post about building AI agents, where they make a compelling case for keeping things simple: use straightforward, composable patterns instead of getting tangled up in complex frameworks. Their key insight? Focus on specific tasks rather than trying to build do-it-all agents.

I wanted to take these ideas and show how they work in practice. This article walks through each workflow with real examples using Trigger.dev, Vercel's AI SDK, and OpenAI.

Prompt chaining

Let's start with prompt chaining, which is probably the simplest pattern in the toolkit. Think of it as breaking down a complex task into a series of smaller, more manageable steps. Instead of asking an LLM to do everything at once, you guide it through a predetermined sequence.

The beauty of prompt chaining is that it's predictable and easy to debug since you know exactly what's happening at each step. Plus, by breaking things down, you often get better results than trying to tackle everything in one giant prompt.

Example

Generate marketing copy on a subject you provide. (LLM call 1)
Check the word count fits a target range. (Gate)
Take the output and translate it into a target language. (LLM call 2)

Notice that we're using the experimental_telemetry option from the Vercel AI SDK to enable telemetry for the task. This is useful for debugging and getting more details on each LLM call.

import { openai } from "@ai-sdk/openai";
import { task } from "@trigger.dev/sdk/v3";
import { generateText } from "ai";

export interface TranslatePayload {
  marketingSubject: string;
  targetLanguage: string;
  targetWordCount: number;
}

export const generateAndTranslateTask = task({
  id: "generate-and-translate-copy",
  maxDuration: 300, // Stop executing after 5 mins of compute
  run: async (payload: TranslatePayload) => {
    // Step 1: Generate marketing copy
    const generatedCopy = await generateText({
      model: openai("o1-mini"),
      messages: [
        {
          role: "system",
          content: "You are an expert copywriter.",
        },
        {
          role: "user",
          content: `Generate as close as possible to ${payload.targetWordCount} words of compelling marketing copy for ${payload.marketingSubject}`,
        },
      ],
      experimental_telemetry: {
        isEnabled: true,
        functionId: "generate-and-translate-copy",
      },
    });

    // Gate: Validate the generated copy meets the word count target
    const wordCount = generatedCopy.text.split(/\s+/).length;

    if (
      wordCount < payload.targetWordCount - 10 ||
      wordCount > payload.targetWordCount + 10
    ) {
      throw new Error(
        `Generated copy length (${wordCount} words) is outside acceptable range of ${
          payload.targetWordCount - 10
        }-${payload.targetWordCount + 10} words`
      );
    }

    // Step 2: Translate to target language
    const translatedCopy = await generateText({
      model: openai("o1-mini"),
      messages: [
        {
          role: "system",
          content: `You are an expert translator specializing in marketing content translation into ${payload.targetLanguage}.`,
        },
        {
          role: "user",
          content: `Translate the following marketing copy to ${payload.targetLanguage}, maintaining the same tone and marketing impact:\n\n${generatedCopy}`,
        },
      ],
      experimental_telemetry: {
        isEnabled: true,
        functionId: "generate-and-translate-copy",
      },
    });

    return {
      englishCopy: generatedCopy,
      translatedCopy,
    };
  },
});

If we trigger this task with the following payload, it generates this run for us in the Trigger.dev dashboard:

{
  marketingSubject: "The controversial new Jaguar electric concept car",
  targetLanguage: "Spanish",
  targetWordCount: 100,
}

Routing

You can think of routing as an AI traffic controller. Instead of forcing one LLM to handle everything, you first figure out what type of task you're dealing with, then send it to the right specialist.

Example

User asks a question.
Determine if the question is simple or complex.
Use the appropriate model to answer the question.
Return the answer, model used, and reasoning.

You'll notice there's a lot of prompts in these examples. As you develop your prompts, you'll likely want to iterate and refine them over time. I recommend using tools like Langfuse or Langsmith for prompt management and metrics, making it easier to track performance and make improvements.

import { openai } from "@ai-sdk/openai";
import { task } from "@trigger.dev/sdk/v3";
import { generateText } from "ai";
import { z } from "zod";

const routingSchema = z.object({
  model: z.enum(["gpt-4o", "gpt-o3-mini"]),
  reason: z.string(),
});

const ROUTER_PROMPT = `You are a routing assistant that determines the complexity of questions.
Analyze the following question and route it to the appropriate model:

- Use "gpt-4o" for simple, common, or straightforward questions
- Use "gpt-o3-mini" for complex, unusual, or questions requiring deep reasoning

Respond with a JSON object in this exact format:
{"model": "gpt-4o" or "gpt-o3-mini", "reason": "your reasoning here"}

Question: `;

export const routeAndAnswerQuestion = task({
  id: "route-and-answer-question",
  run: async (payload: { question: string }) => {
    // Step 1: Route the question
    const routingResponse = await generateText({
      model: openai("o1-mini"),
      messages: [
        {
          role: "system",
          content:
            "You must respond with a valid JSON object containing only 'model' and 'reason' fields. No markdown, no backticks, no explanation.",
        },
        {
          role: "user",
          content: ROUTER_PROMPT + payload.question,
        },
      ],
      temperature: 0.1,
      experimental_telemetry: {
        isEnabled: true,
        functionId: "route-and-answer-question",
      },
    });

   const routingResult = routingSchema.parse(JSON.parse(jsonText));

    // Step 2: Get the answer using the selected model
    const answerResult = await generateText({
      model: openai(routingResult.model),
      messages: [{ role: "user", content: payload.question }],
    });

    return {
      answer: answerResult.text,
      selectedModel: routingResult.model,
      routingReason: routingResult.reason,
    };
  },
});

Triggering our task with a simple question shows it routing to the gpt-4o model and returning the answer with reasoning:

{
  question: "How many planets are there in the solar system?"
}

Parallelization

Sometimes you need to do multiple things at once – that's where parallelization comes in. Rather than working through tasks one by one, you split them up and run them simultaneously. This is where batch.triggerByTaskAndWait shines, allowing you to execute multiple tasks in parallel and efficiently coordinate their responses.

Example

This example responds to customer questions by simultaneously generating a response and checking for inappropriate content.

3 tasks handle this:

The first generates a response to the user's question.
The second task checks for innapropriate content.
The third, main task coordinates the responses by using batch.triggerByTaskAndWait to run the two tasks in parallel. If the content is inappropriate, this task returns a message saying it can't process the request, otherwise it returns the generated response.

import { openai } from "@ai-sdk/openai";
import { batch, task } from "@trigger.dev/sdk/v3";
import { generateText } from "ai";

// Task to generate customer response
export const generateCustomerResponse = task({
  id: "generate-customer-response",
  run: async (payload: { question: string }) => {
    const response = await generateText({
      model: openai("o1-mini"),
      messages: [
        {
          role: "system",
          content: "You are a helpful customer service representative.",
        },
        { role: "user", content: payload.question },
      ],
      experimental_telemetry: {
        isEnabled: true,
        functionId: "generate-customer-response",
      },
    });

    return response.text;
  },
});

// Task to check for inappropriate content
export const checkInappropriateContent = task({
  id: "check-inappropriate-content",
  run: async (payload: { text: string }) => {
    const response = await generateText({
      model: openai("o1-mini"),
      messages: [
        {
          role: "system",
          content:
            "You are a content moderator. Respond with 'true' if the content is inappropriate or contains harmful, threatening, offensive, or explicit content, 'false' otherwise.",
        },
        { role: "user", content: payload.text },
      ],
      experimental_telemetry: {
        isEnabled: true,
        functionId: "check-inappropriate-content",
      },
    });

    return response.text.toLowerCase().includes("true");
  },
});

// Main task that coordinates the parallel execution
export const handleCustomerQuestion = task({
  id: "handle-customer-question",
  run: async (payload: { question: string }) => {
    const {
      runs: [responseRun, moderationRun],
    } = await batch.triggerByTaskAndWait([
      {
        task: generateCustomerResponse,
        payload: { question: payload.question },
      },
      {
        task: checkInappropriateContent,
        payload: { text: payload.question },
      },
    ]);

    // Check moderation result first
    if (moderationRun.ok && moderationRun.output === true) {
      return {
        response:
          "I apologize, but I cannot process this request as it contains inappropriate content.",
        wasInappropriate: true,
      };
    }

    // Return the generated response if everything is ok
    if (responseRun.ok) {
      return {
        response: responseRun.output,
        wasInappropriate: false,
      };
    }

    // Handle any errors
    throw new Error("Failed to process customer question");
  },
});

When we trigger our task with a question, you can see the 2 LLM calls running in parallel using batch.triggerByTaskAndWait. The main task waits for both to complete before returning a response.

{
  "question": "Can you explain 2FA?"
}

Orchestrator-workers

This pattern is like having a project manager (the orchestrator) who breaks down a big job into smaller tasks and assigns them to specialists (the workers). The orchestrator keeps track of everything and puts all the pieces back together at the end. Using batch.triggerByTaskAndWait, it efficiently coordinates multiple tasks while maintaining clear control over the entire workflow.

Example

Extracts distinct factual claims from a news article.
Verifies each claim by considering recent news sources and official statements.
Analyzes the historical context of each claim in the context of past announcements and technological feasibility.
Returns the claims, verifications, and historical analyses.

import { openai } from "@ai-sdk/openai";
import { batch, logger, task } from "@trigger.dev/sdk/v3";
import { CoreMessage, generateText } from "ai";

// Define types for our workers' outputs
interface Claim {
  id: number;
  text: string;
}

interface SourceVerification {
  claimId: number;
  isVerified: boolean;
  confidence: number;
  explanation: string;
}

interface HistoricalAnalysis {
  claimId: number;
  feasibility: number;
  historicalContext: string;
}

// Worker 1: Claim Extractor
export const extractClaims = task({
  id: "extract-claims",
  run: async ({ article }: { article: string }) => {
    try {
      const messages: CoreMessage[] = [
        {
          role: "system",
          content:
            "Extract distinct factual claims from the news article. Format as numbered claims.",
        },
        {
          role: "user",
          content: article,
        },
      ];

      const response = await generateText({
        model: openai("o1-mini"),
        messages,
      });

      const claims = response.text
        .split("\n")
        .filter((line: string) => line.trim())
        .map((claim: string, index: number) => ({
          id: index + 1,
          text: claim.replace(/^\d+\.\s*/, ""),
        }));

      logger.info("Extracted claims", { claimCount: claims.length });
      return claims;
    } catch (error) {
      logger.error("Error in claim extraction", {
        error: error instanceof Error ? error.message : "Unknown error",
      });
      throw error;
    }
  },
});

// Worker 2: Source Verifier
export const verifySource = task({
  id: "verify-source",
  run: async (claim: Claim) => {
    const response = await generateText({
      model: openai("o1-mini"),
      messages: [
        {
          role: "system",
          content:
            "Verify this claim by considering recent news sources and official statements. Assess reliability.",
        },
        {
          role: "user",
          content: claim.text,
        },
      ],
      experimental_telemetry: {
        isEnabled: true,
        functionId: "verify-source",
      },
    });

    return {
      claimId: claim.id,
      isVerified: false,
      confidence: 0.7,
      explanation: response.text,
    };
  },
});

// Worker 3: Historical Context Analyzer
export const analyzeHistory = task({
  id: "analyze-history",
  run: async (claim: Claim) => {
    const response = await generateText({
      model: openai("o1-mini"),
      messages: [
        {
          role: "system",
          content:
            "Analyze this claim in historical context, considering past announcements and technological feasibility.",
        },
        {
          role: "user",
          content: claim.text,
        },
      ],
      experimental_telemetry: {
        isEnabled: true,
        functionId: "analyze-history",
      },
    });

    return {
      claimId: claim.id,
      feasibility: 0.8,
      historicalContext: response.text,
    };
  },
});

// Orchestrator
export const newsFactChecker = task({
  id: "news-fact-checker",
  run: async ({ article }: { article: string }) => {
    // Step 1: Extract claims
    const claimsResult = await batch.triggerByTaskAndWait([
      { task: extractClaims, payload: { article } },
    ]);

    if (!claimsResult.runs[0].ok) {
      logger.error("Failed to extract claims", {
        error: claimsResult.runs[0].error,
        runId: claimsResult.runs[0].id,
      });
      throw new Error(
        `Failed to extract claims: ${claimsResult.runs[0].error}`
      );
    }

    const claims = claimsResult.runs[0].output;

    // Step 2: Process claims in parallel
    const parallelResults = await batch.triggerByTaskAndWait([
      ...claims.map((claim) => ({ task: verifySource, payload: claim })),
      ...claims.map((claim) => ({ task: analyzeHistory, payload: claim })),
    ]);

    // Split and process results
    const verifications = parallelResults.runs
      .filter(
        (run): run is typeof run & { ok: true } =>
          run.ok && run.taskIdentifier === "verify-source"
      )
      .map((run) => run.output as SourceVerification);

    const historicalAnalyses = parallelResults.runs
      .filter(
        (run): run is typeof run & { ok: true } =>
          run.ok && run.taskIdentifier === "analyze-history"
      )
      .map((run) => run.output as HistoricalAnalysis);

    return { claims, verifications, historicalAnalyses };
  },
});

When we trigger our task with the article payload below, we make 3 separate LLM calls. One for extracting claims, one for verifying each claim, and one for analyzing the historical context of each claim. The orchestrator task waits for all 3 to complete before returning a response using batch.triggerByTaskAndWait.

{
  "article": "Tesla announced a new breakthrough in battery technology today. The company claims their new batteries will have 50% more capacity and cost 30% less to produce. Elon Musk stated this development will enable electric vehicles to achieve price parity with gasoline cars by 2024. The new batteries are scheduled to enter production next quarter at the Texas Gigafactory."
}

Evaluator-optimizer

Here's where you add quality control to your AI system. The evaluator checks the output, and if it's not quite right, the optimizer suggests improvements. Think of it as having a friendly editor who reviews your work and helps make it better.

Example

Generates a translation of the text.
Evaluates the translation.
If the translation is good, returns the final result.
If the translation is not good, recursively calls the task with the translation and feedback.

import { task } from "@trigger.dev/sdk/v3";
import { generateText } from "ai";
import { openai } from "@ai-sdk/openai";

interface TranslationPayload {
  text: string;
  targetLanguage: string;
  previousTranslation?: string;
  feedback?: string;
  rejectionCount?: number;
}

export const translateAndRefine = task({
  id: "translate-and-refine",
  run: async (payload: TranslationPayload) => {
    const rejectionCount = payload.rejectionCount || 0;

    // Bail out if we've hit the maximum attempts
    if (rejectionCount >= 10) {
      return {
        finalTranslation: payload.previousTranslation,
        iterations: rejectionCount,
        status: "MAX_ITERATIONS_REACHED",
      };
    }

    // Generate translation (or refinement if we have previous feedback)
    const translationPrompt = payload.feedback
      ? `Previous translation: "${payload.previousTranslation}"\n\nFeedback received: "${payload.feedback}"\n\nPlease provide an improved translation addressing this feedback.`
      : `Translate this text into ${payload.targetLanguage}, preserving style and meaning: "${payload.text}"`;

    const translation = await generateText({
      model: openai("o1-mini"),
      messages: [
        {
          role: "system",
          content: `You are an expert literary translator into ${payload.targetLanguage}.
                   Focus on accuracy first, then style and natural flow.`,
        },
        {
          role: "user",
          content: translationPrompt,
        },
      ],
      experimental_telemetry: {
        isEnabled: true,
        functionId: "translate-and-refine",
      },
    });

    // Evaluate the translation
    const evaluation = await generateText({
      model: openai("o1-mini"),
      messages: [
        {
          role: "system",
          content: `You are an expert literary critic and translator focused on practical, high-quality translations.
                 Your goal is to ensure translations are accurate and natural, but not necessarily perfect.
                 This is iteration ${
                   rejectionCount + 1
                 } of a maximum 5 iterations.

                 RESPONSE FORMAT:
                 - If the translation meets 90%+ quality: Respond with exactly "APPROVED" (nothing else)
                 - If improvements are needed: Provide only the specific issues that must be fixed

                 Evaluation criteria:
                 - Accuracy of meaning (primary importance)
                 - Natural flow in the target language
                 - Preservation of key style elements

                 DO NOT provide detailed analysis, suggestions, or compliments.
                 DO NOT include the translation in your response.

                 IMPORTANT RULES:
                 - First iteration MUST receive feedback for improvement
                 - Be very strict on accuracy in early iterations
                 - After 3 iterations, lower quality threshold to 85%`,
        },
        {
          role: "user",
          content: `Original: "${payload.text}"
                 Translation: "${translation.text}"
                 Target Language: ${payload.targetLanguage}
                 Iteration: ${rejectionCount + 1}
                 Previous Feedback: ${
                   payload.feedback ? `"${payload.feedback}"` : "None"
                 }

                 ${
                   rejectionCount === 0
                     ? "This is the first attempt. Find aspects to improve."
                     : 'Either respond with exactly "APPROVED" or provide only critical issues that must be fixed.'
                 }`,
        },
      ],
      experimental_telemetry: {
        isEnabled: true,
        functionId: "translate-and-refine",
      },
    });

    // If approved, return the final result
    if (evaluation.text.trim() === "APPROVED") {
      return {
        finalTranslation: translation.text,
        iterations: rejectionCount,
        status: "APPROVED",
      };
    }

    // If not approved, recursively call the task with feedback
    await translateAndRefine
      .triggerAndWait({
        text: payload.text,
        targetLanguage: payload.targetLanguage,
        previousTranslation: translation.text,
        feedback: evaluation.text,
        rejectionCount: rejectionCount + 1,
      })
      .unwrap();
  },
});

When we run this example you can see how our payload text below is translated into French and then refined over a max of 10 iterations. The evaluator is satisfied with the result after the 2nd iteration in this case and returns the final translation.

{
  "text": "In the twilight of his years, the old clockmaker's hands, once steady as the timepieces he crafted, now trembled like autumn leaves in the wind.",
  "targetLanguage": "French"
}

Conclusion

Building effective AI agents doesn't have to be complicated – as Anthropic points out, simpler patterns often work better than complex frameworks. Whether you're chaining prompts together, routing tasks to specialized models, or orchestrating multiple workers, the key is to keep your approach focused and modular.

Give these patterns a try in your Trigger.dev project today. Start simple, see what works for your use case, and build from there.

Next steps

Create a Trigger.dev account
Check out our Quick Start guide
View the examples in our GitHub repo
Join our Discord community to share your implementations and get help
Read the Anthropic article
Try implementing the simplest Prompt chaining pattern in your project

How to scrape a website using Browserbase, Puppeteer, OpenAI and Trigger.dev

James Ritchie — Wed, 23 Oct 2024 18:17:31 +0000

What you'll build

In this tutorial, you'll create a Trigger.dev task that scrapes the top 3 articles from Hacker News using BrowserBase and Puppeteer, summarizes them with ChatGPT and sends a nicely formatted email summary to yourself every weekday at 9AM using Resend.

Before you begin…

Check out this 4 minute video overview of this tutorial to get an idea of what we'll be building.

Prerequisites

Create a Trigger.dev account and setup a new project
Create a BrowserBase account
Install Puppeteer on your local machine
Create an OpenAI account
Create a Resend account

⚠️ Warning

When web scraping, you MUST use a proxy to comply with Trigger.dev's terms of service. Direct scraping of third-party websites without the site owner’s permission using Trigger.dev Cloud is prohibited and will result in account suspension.

Configure your environment variables

Login to each of the services and grab the API keys. Add them to your local .env file so we can run a local test of our task later on.

BROWSERBASE_API_KEY: "<your BrowserBase API key>"
OPENAI_API_KEY: "<your OpenAI API key>"
RESEND_API_KEY: "<your Resend API key>"

Install Puppeteer

Before you can run your task locally, you need to install Puppeteer on your local machine. Check out the Puppeteer installation guide for more information.

npm i puppeteer

Write your task code

Create a new file called trigger/scrape-hacker-news.ts in the trigger folder in your project and add the following code below.

The best way to understand how the following 2 tasks work is by following the comments, but here's a quick overview:

The parent task summarizeHackerNews is set to run every weekday at 9AM using the cron property.
It connects to BrowserBase to proxy the scraping of the Hacker News articles.
It then gets the title and link of the top 3 articles on Hacker News.
Next, it triggers a child task called scrapeAndSummarizeArticle for each of our 3 articles using the batchTriggerAndWait method. You can learn more about batching in the docs.
The child task, scrapeAndSummarizeArticle, scrapes the content of each article using Puppeteer and summarizes it using ChatGPT.
The parent task waits for all of the child tasks to complete before continuing.
Finally, the parent task sends an email summary to you using Resend and React Email using the 'summaries' it's generated from the child tasks.

Ensure you replace the placeholder email addresses with your own.

import { render } from "@react-email/render";
import { logger, schedules, task, wait } from "@trigger.dev/sdk/v3";
import { OpenAI } from "openai";
import puppeteer from "puppeteer-core";
import { Resend } from "resend";
import { HNSummaryEmail } from "./summarize-hn-email";

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });
const resend = new Resend(process.env.RESEND_API_KEY);

// Parent task (scheduled to run 9AM every weekday)
export const summarizeHackerNews = schedules.task({
  id: "summarize-hacker-news",
  cron: {
    pattern: "0 9 * * 1-5",
    timezone: "Europe/London",
  }, // Run at 9 AM, Monday to Friday
  run: async () => {
    // Connect to BrowserBase to proxy the scraping of the Hacker News articles
    const browser = await puppeteer.connect({
      browserWSEndpoint: `wss://connect.browserbase.com?apiKey=${process.env.BROWSERBASE_API_KEY}`,
    });
    logger.info("Connected to Browserbase");

    const page = await browser.newPage();

    // Navigate to Hacker News and scrape top 3 articles
    await page.goto("https://news.ycombinator.com/news", {
      waitUntil: "networkidle0",
    });
    logger.info("Navigated to Hacker News");

    const articles = await page.evaluate(() => {
      const items = document.querySelectorAll(".athing");
      return Array.from(items)
        .slice(0, 3)
        .map((item) => {
          const titleElement = item.querySelector(".titleline > a");
          const link = titleElement?.getAttribute("href");
          const title = titleElement?.textContent;
          return { title, link };
        });
    });
    logger.info("Scraped top 3 articles", { articles });

    await browser.close();
    await wait.for({ seconds: 5 });

    // Use batchTriggerAndWait to process articles
    const summaries = await scrapeAndSummarizeArticle
      .batchTriggerAndWait(
        articles.map((article) => ({
          payload: { title: article.title!, link: article.link! },
          idempotencyKey: article.link,
        }))
      )
      .then((batch) =>
        batch.runs.filter((run) => run.ok).map((run) => run.output)
      );

    // Send email using Resend
    await resend.emails.send({
      from: "Hacker News Summary <from@emailaddress.com>",
      to: ["to@emailaddress.com"],
      subject: "Your morning HN summary",
      html: render(<HNSummaryEmail articles={summaries} />),
    });

    logger.info("Email sent successfully");
  },
});

// Child task for scraping and summarizing individual articles
export const scrapeAndSummarizeArticle = task({
  id: "scrape-and-summarize-articles",
  retry: {
    maxAttempts: 3,
    minTimeoutInMs: 5000,
    maxTimeoutInMs: 10000,
    factor: 2,
    randomize: true,
  },
  run: async ({ title, link }: { title: string; link: string }) => {
    logger.info(`Summarizing ${title}`);

    const browser = await puppeteer.connect({
      browserWSEndpoint: `wss://connect.browserbase.com?apiKey=${process.env.BROWSERBASE_API_KEY}`,
    });
    const page = await browser.newPage();

    // Prevent all assets from loading, images, stylesheets etc
    await page.setRequestInterception(true);
    page.on("request", (request) => {
      if (
        ["script", "stylesheet", "image", "media", "font"].includes(
          request.resourceType()
        )
      ) {
        request.abort();
      } else {
        request.continue();
      }
    });

    await page.goto(link, { waitUntil: "networkidle0" });
    logger.info(`Navigated to article: ${title}`);

    // Extract the main content of the article
    const content = await page.evaluate(() => {
      const articleElement = document.querySelector("article") || document.body;
      return articleElement.innerText.trim().slice(0, 1500); // Limit to 1500 characters
    });

    await browser.close();

    logger.info(`Extracted content for article: ${title}`, { content });

    // Summarize the content using ChatGPT
    const response = await openai.chat.completions.create({
      model: "gpt-4o",
      messages: [
        {
          role: "user",
          content: `Summarize this article in 2-3 concise sentences:\n\n${content}`,
        },
      ],
    });

    logger.info(`Generated summary for article: ${title}`);

    return {
      title,
      link,
      summary: response.choices[0].message.content,
    };
  },
});

Create your React Email template

Install React Email and the @react-email/components package:

npm i @react-email/components

Create a new file called summarize-hn-email.tsx in your project and add the following code. This is currently a simple but nicely styled email template that you can customize to your liking.

import {
  Html,
  Head,
  Body,
  Container,
  Section,
  Heading,
  Text,
  Link,
} from "@react-email/components";

interface Article {
  title: string;
  link: string;
  summary: string | null;
}

export const HNSummaryEmail: React.FC<{ articles: Article[] }> = ({
  articles,
}) => (
  <Html>
    <Head />
    <Body style={{ fontFamily: "Arial, sans-serif", padding: "20px" }}>
      <Container>
        <Heading as="h1">Your Morning HN Summary</Heading>
        {articles.map((article, index) => (
          <Section key={index} style={{ marginBottom: "20px" }}>
            <Heading as="h3">
              <Link href={article.link}>{article.title}</Link>
            </Heading>
            <Text>{article.summary || "No summary available"}</Text>
          </Section>
        ))}
      </Container>
    </Body>
  </Html>
);

Do a test run locally

Once you've written your task code, you can do a test run locally to make sure everything is working as expected. Run the Trigger.dev dev command to start the background worker:

npx trigger.dev@latest dev

Next, go to the Trigger.dev dashboard and click Test in the left hand side menu (1). Choose DEV from the environment options at the top of the page (2), select your task (3), click Now to ensure your task runs immediately (4), and then click the Run test button to trigger your test*(5)*.

You should see your task run and an email sent to you with the Hacker News summary.

It's worth noting that some Hacker News articles might not be accessible. The tasks will each attempt 3 times before giving up and returning an error. Some reasons for this could be that the main content of the article isn't accessible via the article HTML element or that the page has a paywall or the Hacker News post links to a video file. Feel free to edit the task code to handle these cases.

Deploy your task to the Trigger.dev cloud

Once you're happy with your task code, you can deploy it to the Trigger.dev cloud. To do this, you'll first need to add Puppeteer to your build configuration in your trigger.config.ts file.

Add Puppeteer to your build configuration

import { defineConfig } from "@trigger.dev/sdk/v3";
import { puppeteer } from "@trigger.dev/build/extensions/puppeteer";

export default defineConfig({
  project: "<project ref>",
  // Your other config settings...
  build: {
    // This is required to use the Puppeteer library
    extensions: [puppeteer()],
  },
});

Add your environment variables to the Trigger.dev project

Previously, we added our environment variables to the .env file. Now we need to add them to the Trigger.dev project settings so our deployed task can access them. You can copy all of the environment variables from your .env file at once, and paste them all into the Environment variables page in the Trigger.dev dashboard.

Deploy your task

Finally, you can deploy your task to the Trigger.dev cloud by running the trigger.dev@latest deploy command.

npx trigger.dev@latest deploy

Run your task in Production

Once your task is deployed, it will run every weekday at 9AM. You can check the status of your task in the Trigger.dev dashboard by clicking on the Runs tab in the left hand side menu.

If you want to manually trigger your task in production, you can repeat the steps from your local DEV test earlier but this time select PROD from the environment options at the top of the page.

Get started with Trigger.dev for free today: trigger.dev

Trigger.dev update – September 2023

James Ritchie — Thu, 05 Oct 2023 13:41:31 +0000

September has been a busy month as usual for Trigger.dev updates. Bring-Your-Own-Auth and a new Usage Dashboard to name just a few. We've had some great community contributions too that have helped us support more frameworks. Here’s a highlight reel of some of the things we shipped during September…

Support for multiple frameworks

We've had a lot of requests for using Trigger.dev with other frameworks other than Next.js, so today we're announcing three new ones:

✅ Next.js
✅ Remix
✅ Astro
✅ Express

With many more coming very soon:

⏳ SvelteKit
⏳ RedwoodJS
⏳ Nuxt.js
⏳ Nest.js
⏳ Fastify

React Status Hooks

You can now create statuses in your Job code that lets you do some pretty cool stuff in your UI, like:

Show exactly what you want in your UI (with as many statuses as you want).
Pass arbitrary data to your UI, which you can use to render elements.
Update existing elements in your UI as the progress of the run continues.

Check out React Status Hooks in the docs.

Showcase

We've created a beautiful showcase for Jobs (and projects) that you can build using Trigger.dev. They have code you can copy and paste to get started quickly.

View the showcase

Bring-Your-Own Auth

You can now authenticate as your users!

Before, you could only use our Integrations with API Keys or OAuth authentication as yourself, the developer. Now, you can authenticate using the auth credentials of your users.

Auth Resolver allows you to implement your own custom auth resolving using a third-party service like Clerk or Nango.

Documentation

Usage Dashboard

Welcome to our new Usage Dashboard! You can now keep track of your Jobs on the 'Usage & Billing' page of the app. Here's a list of all the data you can now see at a glance:

The number of Job Runs each month
The total number of Runs this month
The total number of Jobs
The total number of Integrations
The number of team members in your Organization

Linear Integration

Streamline your project and issue tracking with our new Linear Integration.

And lots more...

Staging environment
Initial Airtable integration
Better support for generic interfaces
Redact Task outputs from the logs
Improved Documentation
New testing package
Fix for excessively deep Type instantiation

Check out the full changelog for everything we shipped in September.