Trigger.dev

This guide walks you through integrating Novu, the open-source notification infrastructure, with Trigger.dev, a powerful open-source background jobs framework for TypeScript. By combining Novu’s robust notification workflows with Trigger.Dev’s event-driven background job system allows you to send notifications across multiple channels (in-app, email, SMS, push, etc.) in response to events or background tasks - all within a seamless developer experience. Whether you’re processing payments, handling user onboarding, or running scheduled tasks, Trigger.dev lets you define workflows with fine-grained control and real-time observability. Novu plugs into those workflows to manage notification content and delivery, ensuring your service / product-to-user communication has a standard.

What You’ll Learn

In this guide, you’ll learn how to:

Set up both Novu and Trigger.dev in your project
Create and update subscribers via Novu’s API
Trigger Novu notification workflows from a Trigger.dev job
Pass dynamic payload data to power your notification templates

If you haven’t used or are unfamiliar with Trigger.dev, we recommend following the quickstart guide in their docs. This guide assumes you are familiar with the fundamentals.

Getting Started

Install Novu’s SDK in your project:

npm i @novu/api

Import Novu’s SDK and provide the credentials to initialize the Novu instance:

import { Novu } from '@novu/api';

const novu = new Novu({ secretKey: process.env.NOVU_SECRET_KEY });

import os
from novu_py import Novu

novu = Novu(secret_key=os.getenv("NOVU_SECRET_KEY", ""))

s := novugo.New(novugo.WithSecurity(os.Getenv("NOVU_SECRET_KEY")))

$novu = novu\Novu::builder()->setSecurity(getenv('NOVU_SECRET_KEY'))->build();

var novu = new NovuSDK(secretKey: Environment.GetEnvironmentVariable("NOVU_SECRET_KEY")!);

Novu novu = Novu.builder().secretKey(System.getenv("NOVU_SECRET_KEY")).build();

For the sake of this guide, we will create a new dedicated task for 2 common Novu actions:

Trigger Notification Workflow
Add New Subscriber

Triggering Notifications

Core Requirements

When calling Novu’s trigger method, you must provide the following information:

Workflow IDworkflowId (string): This identifies the specific notification workflow you want to execute.Note: Ensure this workflow ID corresponds to an existing, active workflow in your Novu dashboard.

Recipient Informationto (object): This object specifies the recipient of the notification. At a minimum, it requires:

subscriberId (string): A unique identifier for the notification recipient within your system (e.g., a user ID).

Note: If a subscriber with this subscriberId doesn’t already exist in Novu, Novu will automatically create one when the workflow is triggered. You can also pass other identifiers like email, phone, etc., within the to object, depending on your workflow steps.

Basic Trigger Example

Here’s a simple Trigger.dev task that triggers a Novu workflow when the task runs.

import { task, retry } from "@trigger.dev/sdk/v3";
import { Novu } from "@novu/api";

const novu = new Novu({ secretKey: process.env.NOVU_SECRET_KEY });

export const notifyUserTask = task({
  id: "notify-user-task",
  run: async (payload: { userId: string; email: string }) => {
    const { userId, email } = payload;

    await retry.onThrow(async () => {
      await novu.trigger({
        workflowId: "your-workflow-id",
        to: { subscriberId: userId, email: email },
        payload: {},
      });
    }, { maxAttempts: 3 });
  },
});

import os
import novu_py
from novu_py import Novu

with Novu(secret_key=os.getenv("NOVU_SECRET_KEY", "")) as novu:
    novu.trigger(trigger_event_request_dto=novu_py.TriggerEventRequestDto(
        workflow_id="your-workflow-id",
        to={"subscriber_id": user_id, "email": email},
        payload={},
    ))

_, err := s.Trigger(ctx, components.TriggerEventRequestDto{
    WorkflowID: "your-workflow-id",
    To: components.CreateToSubscriberPayloadDto(components.SubscriberPayloadDto{
        SubscriberID: userID, Email: email,
    }),
    Payload: map[string]any{},
}, nil)

$sdk->trigger(triggerEventRequestDto: new Components\TriggerEventRequestDto(
    workflowId: 'your-workflow-id',
    to: new Components\SubscriberPayloadDto(subscriberId: $userId, email: $email),
    payload: [],
));

await sdk.TriggerAsync(new TriggerEventRequestDto {
    WorkflowId = "your-workflow-id",
    To = To.CreateSubscriberPayloadDto(new SubscriberPayloadDto { SubscriberId = userId, Email = email }),
});

novu.trigger().body(TriggerEventRequestDto.builder()
    .workflowId("your-workflow-id")
    .to(To2.of(SubscriberPayloadDto.builder().subscriberId(userId).email(email).build()))
    .build()).call();

curl -X POST 'https://api.novu.co/v1/events/trigger' \
-H 'Content-Type: application/json' \
-H 'Authorization: ApiKey <NOVU_SECRET_KEY>' \
-d '{
  "name": "your-workflow-id",
  "to": { "subscriberId": "user_id", "email": "[email protected]" },
  "payload": {}
}'

Working with Dynamic Content

Using Payloads

Often, you’ll want your notifications to include dynamic data related to the event that triggered them. This is done using the payload property in the novu.trigger call.

Understanding Payloads

The payload is an object containing key-value pairs that you can reference within your Novu notification templates using Handlebars syntax (e.g., {{ name }}, {{ order.id }}).

Example Use Case

Imagine you want to send an email confirming a background job’s completion:

Email Template Variables

Subject: Job {{ jobName }} Completed Successfully!
Body: Hi {{ userName }}, your job '{{ jobName }}' finished processing.

To achieve this, you would pass the userName and jobName in the payload object when triggering the workflow.

Advanced Trigger Example

import { task, retry } from "@trigger.dev/sdk/v3";
import { Novu } from "@novu/api";

const novu = new Novu({ secretKey: process.env.NOVU_SECRET_KEY });

export const notifyOnJobCompletion = task({
  id: "notify-on-job-completion",
  run: async (payload: { userId: string; email: string; name: string; jobId: string }) => {
    const { userId, email, name, jobId } = payload;

    await retry.onThrow(async () => {
      const novuPayload = {
        userName: name,
        jobName: `Data Processing Job #${jobId}`,
      };

      await novu.trigger({
        workflowId: "inbox-test",
        to: { subscriberId: userId, email: email },
        payload: { ...novuPayload },
      });

      return novuPayload;
    }, { maxAttempts: 2 });
  },
});

with Novu(secret_key=os.getenv("NOVU_SECRET_KEY", "")) as novu:
    novu.trigger(trigger_event_request_dto=novu_py.TriggerEventRequestDto(
        workflow_id="inbox-test",
        to={"subscriber_id": user_id, "email": email},
        payload={"userName": name, "jobName": f"Data Processing Job #{job_id}"},
    ))

_, err := s.Trigger(ctx, components.TriggerEventRequestDto{
    WorkflowID: "inbox-test",
    To: components.CreateToSubscriberPayloadDto(components.SubscriberPayloadDto{
        SubscriberID: userID, Email: email,
    }),
    Payload: map[string]any{
        "userName": name, "jobName": fmt.Sprintf("Data Processing Job #%s", jobID),
    },
}, nil)

$sdk->trigger(triggerEventRequestDto: new Components\TriggerEventRequestDto(
    workflowId: 'inbox-test',
    to: new Components\SubscriberPayloadDto(subscriberId: $userId, email: $email),
    payload: ['userName' => $name, 'jobName' => "Data Processing Job #{$jobId}"],
));

await sdk.TriggerAsync(new TriggerEventRequestDto {
    WorkflowId = "inbox-test",
    To = To.CreateSubscriberPayloadDto(new SubscriberPayloadDto { SubscriberId = userId, Email = email }),
    Payload = new Dictionary<string, object>() {
        { "userName", name }, { "jobName", $"Data Processing Job #{jobId}" },
    },
});

novu.trigger().body(TriggerEventRequestDto.builder()
    .workflowId("inbox-test")
    .to(To2.of(SubscriberPayloadDto.builder().subscriberId(userId).email(email).build()))
    .payload(Map.of("userName", name, "jobName", "Data Processing Job #" + jobId))
    .build()).call();

curl -X POST 'https://api.novu.co/v1/events/trigger' \
-H 'Content-Type: application/json' \
-H 'Authorization: ApiKey <NOVU_SECRET_KEY>' \
-d '{
  "name": "inbox-test",
  "to": { "subscriberId": "user_id", "email": "[email protected]" },
  "payload": { "userName": "Jane", "jobName": "Data Processing Job #123" }
}'

Key Points about Payloads

Trigger.dev Task Payload

This is the data your Trigger.dev task receives when it’s invoked. It contains the context needed for the task to run.

Novu Trigger Payload

This is the data you specifically send to Novu via the payload property in the novu.trigger() call. It’s used for populating variables in your notification templates.

Implementation Examples

AI Tasks

This example demonstrates how to build a reliable AI content generation system that keeps users informed throughout the process using Novu notifications.


import { task, event, eventTrigger, retry } from "@trigger.dev/sdk/v3";
import { Novu } from "@novu/api";
import OpenAI from "openai";

// Initialize clients
const novu = new Novu({ 
  secretKey: 'ApiKey ' + process.env['NOVU_SECRET_KEY']
});

const openai = new OpenAI({
  apiKey: process.env['OPENAI_API_KEY'],
});

// Helper functions for prompt generation
function generateTextPrompt(theme: string, description: string) {
  return [
    { role: "system", content: "You are a creative writer crafting marketing content." },
    { role: "user", content: `Write a short, engaging paragraph about ${theme}. Key points: ${description}` }
  ];
}

function generateImagePrompt(theme: string, description: string) {
  return `Create a professional marketing image that represents ${theme}. Style: modern, clean. Details: ${description}`;
}

// Event trigger for content generation
export const contentGenerationRequested = event({
  id: "content-generation-requested",
  trigger: eventTrigger({
    name: "content.generation.requested",
  }),
});

// Notification task for successful completion
export const notifyContentReady = task({
  id: "notify-content-ready",
  run: async ({ 
    userId, 
    email, 
    theme, 
    contentId,
    contentPreview,
    imageUrl
  }: { 
    userId: string;
    email: string;
    theme: string;
    contentId: string;
    contentPreview: string;
    imageUrl: string;
  }) => {
    await retry.onThrow(
      async () => {
        try {
          await novu.trigger({
            workflowId: "ai-generation-completed",
            to: {
              subscriberId: userId,
              email: email,
            },
            payload: {
              userName: email.split('@')[0],
              theme: theme,
              contentId: contentId,
              contentPreview: contentPreview,
              imageUrl: imageUrl,
              viewUrl: `https://yourapp.com/content/${contentId}`
            },
          });
          
          console.log("Content ready notification sent successfully");
        } catch (error: unknown) {
          const errorMessage = error instanceof Error ? error.message : 'Unknown error';
          console.error("Failed to send content ready notification:", errorMessage);
          throw new Error(`Final notification failed: ${errorMessage}`);
        }
      },
      { maxAttempts: 3, factor: 2, minTimeoutInMs: 1000 }
    );
  },
});

// Error notification task
export const notifyError = task({
  id: "notify-error",
  run: async ({ 
    userId, 
    email, 
    theme, 
    errorMessage 
  }: { 
    userId: string;
    email: string;
    theme: string;
    errorMessage: string;
  }) => {
    await retry.onThrow(
      async () => {
        try {
          await novu.trigger({
            workflowId: "ai-generation-error",
            to: {
              subscriberId: userId,
              email: email,
            },
            payload: {
              userName: email.split('@')[0],
              theme: theme,
              errorMessage: errorMessage,
              supportEmail: "[email protected]"
            },
          });
          
          console.log("Error notification sent successfully");
        } catch (error: unknown) {
          const errorMessage = error instanceof Error ? error.message : 'Unknown error';
          console.error("Failed to send error notification:", errorMessage);
          // Not rethrowing here to prevent infinite loops when error notifications fail
        }
      },
      { maxAttempts: 2 }
    );
  },
});

// Main AI content generation task
export const generateContent = task({
  id: "generate-content",
  retry: {
    maxAttempts: 3,
  },
  run: async ({ userId, email, theme, description }: { 
    userId: string;
    email: string;
    theme: string;
    description: string; 
  }) => {
    console.log(`Starting AI content generation for theme: ${theme}`);
    
    // Generate text content
    const textResult = await openai.chat.completions.create({
      model: "gpt-4",
      messages: [{ role: "user", content: description }]
    });

    if (!textResult.choices[0]?.message?.content) {
      throw new Error("No text content generated, retrying...");
    }
    
    // Generate image content
    const imageResult = await openai.images.generate({
      model: "dall-e-3",
      prompt: description,
      size: "1024x1024"
    });

    if (!imageResult.data[0]?.url) {
      throw new Error("No image generated, retrying...");
    }

    return {
      text: textResult.choices[0].message.content,
      imageUrl: imageResult.data[0].url,
    };
  },
});

// Main job that orchestrates the entire workflow
export const processContentRequest = task({
  id: "process-content-request",
  events: [contentGenerationRequested],
  run: async (payload: { 
    userId: string;
    email: string;
    theme: string;
    description: string;
  }) => {
    try {
      // Validate input
      if (!payload.theme || !payload.description) {
        throw new Error("Missing required parameters");
      }

      // Generate content
      const result = await generateContent.run(payload);
      
      const contentId = payload.userId + '-' + Date.now();
      
      // Send completion notification
      await notifyContentReady.run({ 
        userId,
        email,
        theme,
        contentId,
        contentPreview: result.text.substring(0, 100) + "...",
        imageUrl: result.imageUrl
      });
      
      return {
        success: true,
        contentId: contentId,
        ...result
      };
    } catch (error) {
      // If anything fails, ensure user gets notified
      const errorMessage = error instanceof Error ? error.message : "Unknown error occurred";
      console.error("Content generation failed:", errorMessage);
      
      await notifyError.run({
        userId,
        email,
        theme,
        errorMessage: errorMessage
      });
      
      return {
        success: false,
        error: errorMessage
      };
    }
  },
});

Breakdown

The Key Components

The AI Generation Task

export const generateContent = task({
id: "generate-content",
retry: { maxAttempts: 3 },
run: async ({ userId, email, theme, description }) => {
  // Generate text using GPT-4
  const textResult = await openai.chat.completions.create({
    model: "gpt-4",
    messages: [{ role: "user", content: description }]
  });
  
  // Generate image using DALL-E 3
  const imageResult = await openai.images.generate({
    model: "dall-e-3",
    prompt: description,
    size: "1024x1024"
  });
  
  return {
    text: textResult.choices[0].message.content,
    imageUrl: imageResult.data[0].url,
  };
},
});

This task handles the actual AI content generation:

It creates text content using GPT-4
It creates an image using DALL-E 3
It has built-in retry logic (will try up to 3 times if it fails)

The Two Notification Tasks

Completion Notification

export const notifyContentReady = task({
  id: "notify-content-ready",
  run: async ({ userId, email, theme, contentId, contentPreview, imageUrl }) => {
    await novu.trigger({
      workflowId: "ai-generation-completed",
      to: { subscriberId: userId, email: email },
      payload: {
        userName: email.split('@')[0],
        theme: theme,
        contentPreview: contentPreview,
        // Other details...
      },
    });
  },
});

Error Notification

export const notifyError = task({
  id: "notify-error",
  run: async ({ userId, email, theme, errorMessage }) => {
    await novu.trigger({
      workflowId: "ai-generation-error",
      to: { subscriberId: userId, email: email },
      payload: {
        userName: email.split('@')[0],
        theme: theme,
        errorMessage: errorMessage,
        // Other details...
      },
    });
  },
});

The Main Workflow Orchestrator

export const processContentRequest = task({
  id: "process-content-request",
  events: [contentGenerationRequested],
  run: async (payload) => {
    try {
      // 1. Generate the content
      const result = await generateContent.run(payload);
      
      // 2. Send success notification
      await notifyContentReady.run({
        userId: payload.userId,
        email: payload.email,
        theme: payload.theme,
        contentId: payload.userId + '-' + Date.now(),
        contentPreview: result.text.substring(0, 100) + "...",
        imageUrl: result.imageUrl
      });
      
      return { success: true, ...result };
    } catch (error) {
      // 3. Send error notification if anything fails
      await notifyError.run({
        userId: payload.userId,
        email: payload.email,
        theme: payload.theme,
        errorMessage: error instanceof Error ? error.message : "Unknown error occurred"
      });
      
      return { success: false, error: error instanceof Error ? error.message : "Unknown error occurred" };
    }
  },
});

This task:

Responds to the content generation event
Tries to generate content
Sends appropriate notifications based on success or failure

Notification Workflows You Would Need in Novu

ai-generation-completed: Sent when content is successfully generated

Workflow payload variables: {{userName}}, {{theme}}, {{contentPreview}}, {{imageUrl}}, {{viewUrl}}

ai-generation-error: Sent when content generation fails

Workflow payload variables: {{userName}}, {{theme}}, {{errorMessage}}, {{supportEmail}}

Novu trigger calls in the AI tasks example:

await novu.trigger({
  workflowId: "ai-generation-completed",
  to: { subscriberId: userId, email: email },
  payload: {
    userName: email.split('@')[0],
    theme: theme,
    contentId: contentId,
    contentPreview: contentPreview,
    imageUrl: imageUrl,
    viewUrl: `https://yourapp.com/content/${contentId}`,
  },
});

novu.trigger(trigger_event_request_dto=novu_py.TriggerEventRequestDto(
    workflow_id="ai-generation-completed",
    to={"subscriber_id": user_id, "email": email},
    payload={
        "userName": email.split("@")[0],
        "theme": theme,
        "contentId": content_id,
        "contentPreview": content_preview,
        "imageUrl": image_url,
        "viewUrl": f"https://yourapp.com/content/{content_id}",
    },
))

_, err := s.Trigger(ctx, components.TriggerEventRequestDto{
    WorkflowID: "ai-generation-completed",
    To: components.CreateToSubscriberPayloadDto(components.SubscriberPayloadDto{
        SubscriberID: userID, Email: email,
    }),
    Payload: map[string]any{
        "userName": strings.Split(email, "@")[0],
        "theme": theme, "contentId": contentID,
        "contentPreview": contentPreview, "imageUrl": imageURL,
        "viewUrl": fmt.Sprintf("https://yourapp.com/content/%s", contentID),
    },
}, nil)

$sdk->trigger(triggerEventRequestDto: new Components\TriggerEventRequestDto(
    workflowId: 'ai-generation-completed',
    to: new Components\SubscriberPayloadDto(subscriberId: $userId, email: $email),
    payload: [
        'userName' => explode('@', $email)[0],
        'theme' => $theme,
        'contentId' => $contentId,
        'contentPreview' => $contentPreview,
        'imageUrl' => $imageUrl,
        'viewUrl' => "https://yourapp.com/content/{$contentId}",
    ],
));

await sdk.TriggerAsync(new TriggerEventRequestDto {
    WorkflowId = "ai-generation-completed",
    To = To.CreateSubscriberPayloadDto(new SubscriberPayloadDto { SubscriberId = userId, Email = email }),
    Payload = new Dictionary<string, object>() {
        { "userName", email.Split('@')[0] }, { "theme", theme },
        { "contentId", contentId }, { "contentPreview", contentPreview },
        { "imageUrl", imageUrl }, { "viewUrl", $"https://yourapp.com/content/{contentId}" },
    },
});

novu.trigger().body(TriggerEventRequestDto.builder()
    .workflowId("ai-generation-completed")
    .to(To2.of(SubscriberPayloadDto.builder().subscriberId(userId).email(email).build()))
    .payload(Map.of(
        "userName", email.split("@")[0], "theme", theme,
        "contentId", contentId, "contentPreview", contentPreview,
        "imageUrl", imageUrl, "viewUrl", "https://yourapp.com/content/" + contentId))
    .build()).call();

curl -X POST 'https://api.novu.co/v1/events/trigger' \
-H 'Content-Type: application/json' \
-H 'Authorization: ApiKey <NOVU_SECRET_KEY>' \
-d '{
  "name": "ai-generation-completed",
  "to": { "subscriberId": "user_id", "email": "[email protected]" },
  "payload": { "userName": "user", "theme": "Marketing", "contentId": "content-123" }
}'

Video processing

This example shows how to build a video transcription service that notifies users when their video has been transcribed or if an error occurs during processing.


import { task, event, eventTrigger, retry } from "@trigger.dev/sdk/v3";
import { Novu } from "@novu/api";
import fs from "fs";
import { Deepgram } from "@deepgram/sdk";

// Initialize clients
const novu = new Novu({ 
  secretKey: 'ApiKey ' + process.env['NOVU_SECRET_KEY']
});

const deepgram = new Deepgram(process.env['DEEPGRAM_API_KEY']);

// Utility functions for video processing
async function downloadFile(url: string): Promise<string> {
  // Implementation of file download logic
  console.log(`Downloading file from ${url}`);
  // This would be the actual implementation to download a file
  return "/tmp/downloaded-video.mp4";
}

async function convertToWav(filePath: string): Promise<string> {
  // Implementation of conversion logic
  console.log(`Converting ${filePath} to WAV format`);
  // This would be the actual implementation to convert video to WAV
  return "/tmp/audio-extract.wav";
}

// Event trigger for transcription request
export const transcriptionRequested = event({
  id: "transcription-requested",
  trigger: eventTrigger({
    name: "video.transcription.requested",
  }),
});

// Notification task for successful transcription
export const notifyTranscriptionComplete = task({
  id: "notify-transcription-complete",
  run: async ({ 
    userId, 
    email, 
    videoName,
    transcriptionId,
    wordCount,
    duration
  }: { 
    userId: string;
    email: string;
    videoName: string;
    transcriptionId: string;
    wordCount: number;
    duration: string;
  }) => {
    await retry.onThrow(
      async () => {
        try {
          await novu.trigger({
            workflowId: "transcription-completed",
            to: {
              subscriberId: userId,
              email: email,
            },
            payload: {
              userName: email.split('@')[0],
              videoName: videoName,
              transcriptionId: transcriptionId,
              wordCount: wordCount,
              duration: duration,
              viewUrl: `https://yourapp.com/transcriptions/${transcriptionId}`
            },
          });
          
          console.log("Transcription complete notification sent successfully");
        } catch (error: unknown) {
          const errorMessage = error instanceof Error ? error.message : 'Unknown error';
          console.error("Failed to send transcription complete notification:", errorMessage);
          throw new Error(`Notification failed: ${errorMessage}`);
        }
      },
      { maxAttempts: 3, factor: 2, minTimeoutInMs: 1000 }
    );
  },
});

// Error notification task
export const notifyTranscriptionError = task({
  id: "notify-transcription-error",
  run: async ({ 
    userId, 
    email, 
    videoName, 
    errorMessage 
  }: { 
    userId: string;
    email: string;
    videoName: string;
    errorMessage: string;
  }) => {
    await retry.onThrow(
      async () => {
        try {
          await novu.trigger({
            workflowId: "transcription-error",
            to: {
              subscriberId: userId,
              email: email,
            },
            payload: {
              userName: email.split('@')[0],
              videoName: videoName,
              errorMessage: errorMessage,
              supportEmail: "[email protected]"
            },
          });
          
          console.log("Transcription error notification sent successfully");
        } catch (error: unknown) {
          const errorMessage = error instanceof Error ? error.message : 'Unknown error';
          console.error("Failed to send error notification:", errorMessage);
          // Not rethrowing here to prevent infinite loops when error notifications fail
        }
      },
      { maxAttempts: 2 }
    );
  },
});

// Main transcription task
export const transcribeVideo = task({
  id: "transcribe",
  retry: {
    maxAttempts: 3,
  },
  machine: { preset: "large-2x" }, // Use larger machine for faster processing
  run: async (payload: { 
    id: string; 
    url: string;
    userId?: string;
    email?: string;
    videoName?: string;
  }) => {
    try {
      console.log(`Starting transcription for video: ${payload.videoName || payload.id}`);
      
      // Download and process the video
      const downloadedFile = await downloadFile(payload.url);
      const extractedAudio = await convertToWav(downloadedFile);

      // Transcribe using Deepgram
      const { result, error } = await deepgram.listen.prerecorded.transcribeFile(
        fs.createReadStream(extractedAudio),
        {
          model: "nova",
          language: "en-US",
          smart_format: true,
          diarize: true
        }
      );

      if (error || !result) {
        throw new Error(error || "Failed to transcribe video");
      }

      // Calculate some metadata from the result
      const wordCount = result.results?.utterances?.reduce((count, utterance) => count + utterance.words.length, 0) || 0;
      const duration = result.metadata?.duration 
        ? `${Math.floor(result.metadata.duration / 60)}:${Math.floor(result.metadata.duration % 60).toString().padStart(2, '0')}`
        : "Unknown";

      // Store in database
      const dbResult = await db.transcriptions.create(payload.id, result.results);
      
      return {
        transcriptionId: dbResult.id,
        wordCount,
        duration,
        transcript: result.results
      };
    } catch (error) {
      console.error("Error in transcription process:", error);
      throw error; // Rethrow to trigger retry
    }
  },
});

// Main job that orchestrates the entire workflow
export const processTranscriptionRequest = task({
  id: "process-transcription-request",
  events: [transcriptionRequested],
  run: async (payload: { 
    id: string;
    url: string;
    userId: string;
    email: string;
    videoName: string;
  }) => {
    try {
      // Validate input
      if (!payload.url) {
        throw new Error("Missing video URL");
      }

      // Transcribe video
      const result = await transcribeVideo.run(payload);
      
      // Send completion notification
      await notifyTranscriptionComplete.run({
        userId: payload.userId,
        email: payload.email,
        videoName: payload.videoName || "Your video",
        transcriptionId: result.transcriptionId,
        wordCount: result.wordCount,
        duration: result.duration
      });
      
      return {
        success: true,
        transcriptionId: result.transcriptionId,
        transcript: result.transcript
      };
    } catch (error) {
      // If anything fails, ensure user gets notified
      const errorMessage = error instanceof Error ? error.message : "Unknown error occurred";
      console.error("Transcription failed:", errorMessage);
      
      await notifyTranscriptionError.run({
        userId: payload.userId,
        email: payload.email,
        videoName: payload.videoName || "Your video",
        errorMessage: errorMessage
      });
      
      return {
        success: false,
        error: errorMessage
      };
    }
  },
});

// Type declaration for database operations (mock)
const db = {
  transcriptions: {
    create: async (id: string, results: unknown) => {
      console.log(`Storing transcription results for ${id} in database`);
      return { id, created: new Date() };
    }
  }
};

Breakdown

How It Works

A user uploads a video for transcription
The system processes the video using Deepgram’s API
The user receives a notification when transcription is complete or if an error occurs

The Key Components

Transcription Task

The core of this workflow is the transcribeVideo task that handles the actual transcription:

export const transcribeVideo = task({
  id: "transcribe",
  retry: { maxAttempts: 3 },
  machine: { preset: "large-2x" }, // Use larger machine for faster processing
  run: async (payload) => {
    // Download the video file
    const downloadedFile = await downloadFile(payload.url);
    
    // Extract and convert audio to WAV format
    const extractedAudio = await convertToWav(downloadedFile);

    // Transcribe using Deepgram
    const { result, error } = await deepgram.listen.prerecorded.transcribeFile(
      fs.createReadStream(extractedAudio),
      { model: "nova" }
    );

    // Store results in database
    return await db.transcriptions.create(payload.id, result?.results);
  },
});

This task:

Downloads the video file from a URL
Extracts audio and converts it to WAV format
Sends the audio to Deepgram for transcription
Stores the results in a database

Notification Tasks

There are two notification tasks:Success Notification

export const notifyTranscriptionComplete = task({
  id: "notify-transcription-complete",
  run: async ({ userId, email, videoName, transcriptionId, wordCount, duration }) => {
    await novu.trigger({
      workflowId: "transcription-completed",
      to: { subscriberId: userId, email },
      payload: {
        userName: email.split('@')[0],
        videoName,
        wordCount,
        duration,
        viewUrl: `https://yourapp.com/transcriptions/${transcriptionId}`
      },
    });
  },
});

Error Notification

export const notifyTranscriptionError = task({
  id: "notify-transcription-error",
  run: async ({ userId, email, videoName, errorMessage }) => {
    await novu.trigger({
      workflowId: "transcription-error",
      to: { subscriberId: userId, email },
      payload: {
        userName: email.split('@')[0],
        videoName,
        errorMessage,
        supportEmail: "[email protected]"
      },
    });
  },
});

Main Workflow Orchestrator

export const processTranscriptionRequest = task({
  id: "process-transcription-request",
  events: [transcriptionRequested],
  run: async (payload) => {
    try {
      // 1. Validate and transcribe video
      const result = await transcribeVideo.run(payload);
      
      // 2. Send success notification
      await notifyTranscriptionComplete.run({
        userId: payload.userId,
        email: payload.email,
        videoName: payload.videoName || "Your video",
        transcriptionId: result.transcriptionId,
        wordCount: result.wordCount,
        duration: result.duration
      });
      
      return { success: true, transcriptionId: result.transcriptionId };
    } catch (error) {
      // 3. Send error notification if anything fails
      await notifyTranscriptionError.run({
        userId: payload.userId,
        email: payload.email,
        videoName: payload.videoName || "Your video",
        errorMessage: error instanceof Error ? error.message : "Unknown error occurred"
      });
      
      return { success: false, error: error instanceof Error ? error.message : "Unknown error occurred" };
    }
  },
});

Usage

Create two notification workflows in Novu:

transcription-completed with payload variables: {{userName}}, {{videoName}}, {{wordCount}}, {{duration}}, {{viewUrl}}
transcription-error with payload variables: {{userName}}, {{videoName}}, {{errorMessage}}, {{supportEmail}}

Trigger the workflow by sending an event:

// Example of triggering the workflow
await client.triggerEvent({
  name: "video.transcription.requested",
  payload: {
    id: "video-123",
    url: "https://storage.example.com/videos/meeting-recording.mp4",
    userId: "user-456",
    email: "[email protected]",
    videoName: "Weekly Team Meeting"
  }
});

Example NotificationsSuccess notification email:

Subject: Your video “Weekly Team Meeting” has been transcribedHello John,Good news! We’ve finished transcribing your video “Weekly Team Meeting”.Details:

Duration: 32:15
Word count: 4,320

You can view and edit your transcription here: https://yourapp.com/transcriptions/transcript-789Thanks for using our service!

Error notification email:

Subject: Issue with transcribing “Weekly Team Meeting”Hello John,We encountered a problem while transcribing your video “Weekly Team Meeting”.Error details: The audio quality was too low for accurate transcription.Please try uploading your video again with improved audio, or contact [email protected] if you need assistance.We apologize for the inconvenience.

This workflow provides a complete solution for transcribing videos and keeping users informed about the process status, all while handling errors gracefully.

Novu trigger calls in the video processing example:

await novu.trigger({
  workflowId: "transcription-completed",
  to: { subscriberId: userId, email: email },
  payload: {
    userName: email.split('@')[0],
    videoName: videoName,
    transcriptionId: transcriptionId,
    wordCount: wordCount,
    duration: duration,
    viewUrl: `https://yourapp.com/transcriptions/${transcriptionId}`,
  },
});

novu.trigger(trigger_event_request_dto=novu_py.TriggerEventRequestDto(
    workflow_id="transcription-completed",
    to={"subscriber_id": user_id, "email": email},
    payload={
        "userName": email.split("@")[0],
        "videoName": video_name,
        "transcriptionId": transcription_id,
        "wordCount": word_count,
        "duration": duration,
        "viewUrl": f"https://yourapp.com/transcriptions/{transcription_id}",
    },
))

_, err := s.Trigger(ctx, components.TriggerEventRequestDto{
    WorkflowID: "transcription-completed",
    To: components.CreateToSubscriberPayloadDto(components.SubscriberPayloadDto{
        SubscriberID: userID, Email: email,
    }),
    Payload: map[string]any{
        "userName": strings.Split(email, "@")[0],
        "videoName": videoName, "transcriptionId": transcriptionID,
        "wordCount": wordCount, "duration": duration,
        "viewUrl": fmt.Sprintf("https://yourapp.com/transcriptions/%s", transcriptionID),
    },
}, nil)

$sdk->trigger(triggerEventRequestDto: new Components\TriggerEventRequestDto(
    workflowId: 'transcription-completed',
    to: new Components\SubscriberPayloadDto(subscriberId: $userId, email: $email),
    payload: [
        'userName' => explode('@', $email)[0],
        'videoName' => $videoName,
        'transcriptionId' => $transcriptionId,
        'wordCount' => $wordCount,
        'duration' => $duration,
        'viewUrl' => "https://yourapp.com/transcriptions/{$transcriptionId}",
    ],
));

await sdk.TriggerAsync(new TriggerEventRequestDto {
    WorkflowId = "transcription-completed",
    To = To.CreateSubscriberPayloadDto(new SubscriberPayloadDto { SubscriberId = userId, Email = email }),
    Payload = new Dictionary<string, object>() {
        { "userName", email.Split('@')[0] }, { "videoName", videoName },
        { "transcriptionId", transcriptionId }, { "wordCount", wordCount },
        { "duration", duration }, { "viewUrl", $"https://yourapp.com/transcriptions/{transcriptionId}" },
    },
});

novu.trigger().body(TriggerEventRequestDto.builder()
    .workflowId("transcription-completed")
    .to(To2.of(SubscriberPayloadDto.builder().subscriberId(userId).email(email).build()))
    .payload(Map.of(
        "userName", email.split("@")[0], "videoName", videoName,
        "transcriptionId", transcriptionId, "wordCount", wordCount,
        "duration", duration, "viewUrl", "https://yourapp.com/transcriptions/" + transcriptionId))
    .build()).call();

curl -X POST 'https://api.novu.co/v1/events/trigger' \
-H 'Content-Type: application/json' \
-H 'Authorization: ApiKey <NOVU_SECRET_KEY>' \
-d '{
  "name": "transcription-completed",
  "to": { "subscriberId": "user_id", "email": "[email protected]" },
  "payload": { "videoName": "Weekly Team Meeting", "wordCount": 4320, "duration": "32:15" }
}'

Managing Subscribers

Before triggering notifications, Novu needs to know who to notify. That’s where subscribers come in. A subscriber in Novu represents a user (or entity) that can receive notifications through one or more channels (in-app, email, SMS, etc.).

When to Create Subscribers

Create or update a subscriber in Novu when:

New User RegistrationWhen a new user signs up or is added to your system

Notification EligibilityWhen a user becomes eligible to receive notifications

Data UpdatesWhen you want to ensure subscriber metadata (name, phone, etc.) is up-to-date

If you trigger a workflow with a subscriberId that doesn’t exist, Novu will auto-create the subscriber. However, doing it explicitly ensures full control over subscriber data.

Subscriber Creation Example

import { task, retry } from "@trigger.dev/sdk/v3";
import { Novu } from "@novu/api";

const novu = new Novu({ secretKey: process.env.NOVU_SECRET_KEY });

export const createSubscriberTask = task({
  id: "create-subscriber-task",
  run: async (payload: {
    userId: string;
    email?: string;
    name?: string;
    phone?: string;
    locale?: string;
    avatar?: string;
    data?: object;
  }) => {
    const { userId, email, name, phone, locale, avatar, data } = payload;
    const [firstName, lastName = ""] = (name ?? "").split(" ");

    const subscriber = { subscriberId: userId, email, firstName, lastName, phone, locale, avatar, data };

    await retry.onThrow(async () => {
      return await novu.subscribers.create(subscriber);
    }, { maxAttempts: 2 });
  },
});

with Novu(secret_key=os.getenv("NOVU_SECRET_KEY", "")) as novu:
    novu.subscribers.create(
        create_subscriber_request_dto=novu_py.CreateSubscriberRequestDto(
            subscriber_id=user_id,
            email=email,
            first_name=first_name,
            last_name=last_name,
            phone=phone,
            locale=locale,
            avatar=avatar,
            data=custom_data,
        )
    )

_, err := s.Subscribers.Create(ctx, components.CreateSubscriberRequestDto{
    SubscriberID: userID,
    Email: email,
    FirstName: firstName,
    LastName: lastName,
    Phone: phone,
    Locale: locale,
    Avatar: avatar,
    Data: customData,
}, nil)

$sdk->subscribers->create(
    createSubscriberRequestDto: new Components\CreateSubscriberRequestDto(
        subscriberId: $userId,
        email: $email,
        firstName: $firstName,
        lastName: $lastName,
        phone: $phone,
        locale: $locale,
        avatar: $avatar,
        data: $customData,
    ),
);

await sdk.Subscribers.CreateAsync(new CreateSubscriberRequestDto {
    SubscriberId = userId,
    Email = email,
    FirstName = firstName,
    LastName = lastName,
    Phone = phone,
    Locale = locale,
    Avatar = avatar,
    Data = customData,
});

novu.subscribers().create()
    .body(CreateSubscriberRequestDto.builder()
        .subscriberId(userId)
        .email(email)
        .firstName(firstName)
        .lastName(lastName)
        .phone(phone)
        .locale(locale)
        .avatar(avatar)
        .data(customData)
        .build())
    .call();

curl -X POST 'https://api.novu.co/v2/subscribers' \
-H 'Content-Type: application/json' \
-H 'Authorization: ApiKey <NOVU_SECRET_KEY>' \
-d '{
  "subscriberId": "user_id",
  "email": "[email protected]",
  "firstName": "Jane",
  "lastName": "Doe"
}'

Using Subscriber Data in Templates

Once you’ve added custom fields to a subscriber, you can use them in Novu templates using Handlebars:

Basic Fields
Contact Info
Custom Data

Hi {{subscriber.firstName}}, welcome!

We’ll reach you at {{subscriber.phone}} if needed.

Your account is set up under {{subscriber.data.userName}}.

Best Practices

Data Management Best Practices

As a best practice, maintain consistent subscriber identification:

Use your internal userId as the subscriberId in Novu
Keep this mapping consistent across all integrations
Store notification preferences and settings in your user model
Consider adding a reference field in your user table: novuSubscriberId

Example user model:

interface User {
  id: string;
  email: string;
  novuSubscriberId: string; // Same as user.id
  notificationPreferences: {
    marketing: boolean;
    updates: boolean;
    // ... other preferences
  }
}

This ensures reliable user tracking and simplifies debugging notification issues.

Proactive Data Synchronization

Keep subscriber data synchronized by:

Updating Novu whenever user contact info changes
Implementing webhooks to handle notification delivery status
Setting up retry mechanisms for failed updates

// Example of proactive update
async function updateUserProfile(userId: string, newData: UserUpdateData) {
  await db.users.update(userId, newData);
  await novu.subscribers.patch(
    { email: newData.email, phone: newData.phone, data: { lastUpdated: new Date() } },
    userId
  );
}

await novu.subscribers.patch(
  { email: newData.email, phone: newData.phone, data: { lastUpdated: new Date() } },
  userId
);

novu.subscribers.patch(subscriber_id=user_id, patch_subscriber_request_dto={
    "email": new_data.email,
    "phone": new_data.phone,
    "data": {"lastUpdated": datetime.now().isoformat()},
})

s.Subscribers.Patch(ctx, userID, components.PatchSubscriberRequestDto{
    Email: &email, Phone: &phone,
    Data: map[string]any{"lastUpdated": time.Now()},
}, nil)

$sdk->subscribers->patch($userId, new Components\PatchSubscriberRequestDto(
    email: $newData->email,
    phone: $newData->phone,
    data: ['lastUpdated' => date('c')],
));

await sdk.Subscribers.PatchAsync(userId, new PatchSubscriberRequestDto {
    Email = newData.Email, Phone = newData.Phone,
    Data = new Dictionary<string, object>() { { "lastUpdated", DateTime.UtcNow } },
});

novu.subscribers().patch(userId).body(PatchSubscriberRequestDto.builder()
    .email(newData.getEmail()).phone(newData.getPhone())
    .data(Map.of("lastUpdated", Instant.now().toString())).build()).call();

curl -X PATCH 'https://api.novu.co/v2/subscribers/user_id' \
-H 'Content-Type: application/json' \
-H 'Authorization: ApiKey <NOVU_SECRET_KEY>' \
-d '{ "email": "[email protected]", "phone": "+15551234567" }'

Enriching Subscriber Data

Enhance notifications with contextual data:

Store user preferences, roles, and settings
Include relevant business logic flags
Add metadata for analytics and tracking

await novu.subscribers.patch(
  {
    data: {
      role: 'admin',
      subscriptionTier: 'premium',
      features: ['advanced-analytics', 'priority-support'],
    },
  },
  userId
);

await novu.subscribers.patch(
  { data: { role: 'admin', subscriptionTier: 'premium', features: ['advanced-analytics', 'priority-support'] } },
  userId
);

novu.subscribers.patch(subscriber_id=user_id, patch_subscriber_request_dto={
    "data": {
        "role": "admin",
        "subscriptionTier": "premium",
        "features": ["advanced-analytics", "priority-support"],
    },
})

s.Subscribers.Patch(ctx, userID, components.PatchSubscriberRequestDto{
    Data: map[string]any{
        "role": "admin", "subscriptionTier": "premium",
        "features": []string{"advanced-analytics", "priority-support"},
    },
}, nil)

$sdk->subscribers->patch($userId, new Components\PatchSubscriberRequestDto(
    data: [
        'role' => 'admin',
        'subscriptionTier' => 'premium',
        'features' => ['advanced-analytics', 'priority-support'],
    ],
));

await sdk.Subscribers.PatchAsync(userId, new PatchSubscriberRequestDto {
    Data = new Dictionary<string, object>() {
        { "role", "admin" }, { "subscriptionTier", "premium" },
        { "features", new[] { "advanced-analytics", "priority-support" } },
    },
});

novu.subscribers().patch(userId).body(PatchSubscriberRequestDto.builder()
    .data(Map.of(
        "role", "admin",
        "subscriptionTier", "premium",
        "features", List.of("advanced-analytics", "priority-support")))
    .build()).call();

curl -X PATCH 'https://api.novu.co/v2/subscribers/user_id' \
-H 'Content-Type: application/json' \
-H 'Authorization: ApiKey <NOVU_SECRET_KEY>' \
-d '{ "data": { "role": "admin", "subscriptionTier": "premium" } }'

Use Cases

Guides

Inbound Webhooks

Recipes

Framework

Migrate to Novu

Trigger.dev

What You’ll Learn

Getting Started

Triggering Notifications

Core Requirements

Basic Trigger Example

Working with Dynamic Content

Using Payloads

Example Use Case

Email Template Variables

Advanced Trigger Example

Key Points about Payloads

Trigger.dev Task Payload

Novu Trigger Payload

Implementation Examples

AI Tasks

Video processing

Managing Subscribers

When to Create Subscribers

Subscriber Creation Example

Using Subscriber Data in Templates

Best Practices

​What You’ll Learn

​Getting Started

​Triggering Notifications

​Core Requirements

​Basic Trigger Example

​Working with Dynamic Content

​Using Payloads

​Example Use Case

Email Template Variables

​Advanced Trigger Example

​Key Points about Payloads

Trigger.dev Task Payload

Novu Trigger Payload

​Implementation Examples

​AI Tasks

​Video processing

​Managing Subscribers

​When to Create Subscribers

​Subscriber Creation Example

​Using Subscriber Data in Templates

​Best Practices

What You’ll Learn

Getting Started

Triggering Notifications

Core Requirements

Basic Trigger Example

Working with Dynamic Content

Using Payloads

Example Use Case

Advanced Trigger Example

Key Points about Payloads

Implementation Examples

AI Tasks

Video processing

Managing Subscribers

When to Create Subscribers

Subscriber Creation Example

Using Subscriber Data in Templates

Best Practices