Back to Novu

Trigger.dev

docs/guides/triggerdotdev.mdx

3.18.049.3 KB
Original Source

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
<Note> If you haven't used or are unfamiliar with Trigger.dev, we recommend following [the quickstart guide in their docs](https://trigger.dev/docs/quick-start). This guide assumes you are familiar with the fundamentals. </Note>

Getting Started

<Steps> <Step> Install Novu's SDK in your project: ```bash npm i @novu/api ``` </Step> <Step> Import Novu's SDK and provide the credentials to initialize the Novu instance:
<Tabs>
  <Tab title="Node.js">
```typescript
import { Novu } from '@novu/api';

const novu = new Novu({ secretKey: process.env.NOVU_SECRET_KEY });
```
  </Tab>
  <Tab title="Python">
```python
import os
from novu_py import Novu

novu = Novu(secret_key=os.getenv("NOVU_SECRET_KEY", ""))
```
  </Tab>
  <Tab title="Go">
```go
s := novugo.New(novugo.WithSecurity(os.Getenv("NOVU_SECRET_KEY")))
```
  </Tab>
  <Tab title="PHP">
```php
$novu = novu\Novu::builder()->setSecurity(getenv('NOVU_SECRET_KEY'))->build();
```
  </Tab>
  <Tab title=".NET">
```csharp
var novu = new NovuSDK(secretKey: Environment.GetEnvironmentVariable("NOVU_SECRET_KEY")!);
```
  </Tab>
  <Tab title="Java">
```java
Novu novu = Novu.builder().secretKey(System.getenv("NOVU_SECRET_KEY")).build();
```
  </Tab>
</Tabs>
</Step> </Steps>

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:

<Steps> <Step> **Workflow ID**
`workflowId` **(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.
</Step> <Step> **Recipient Information**
`to` **(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.
</Step> </Steps>

Basic Trigger Example

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

<Tabs> <Tab title="Node.js"> ```typescript 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 });

}, });

  </Tab>
  <Tab title="Python">
```python
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={},
    ))
</Tab> <Tab title="Go"> ```go _, err := s.Trigger(ctx, components.TriggerEventRequestDto{ WorkflowID: "your-workflow-id", To: components.CreateToSubscriberPayloadDto(components.SubscriberPayloadDto{ SubscriberID: userID, Email: email, }), Payload: map[string]any{}, }, nil) ``` </Tab> <Tab title="PHP"> ```php $sdk->trigger(triggerEventRequestDto: new Components\TriggerEventRequestDto( workflowId: 'your-workflow-id', to: new Components\SubscriberPayloadDto(subscriberId: $userId, email: $email), payload: [], )); ``` </Tab> <Tab title=".NET"> ```csharp await sdk.TriggerAsync(new TriggerEventRequestDto { WorkflowId = "your-workflow-id", To = To.CreateSubscriberPayloadDto(new SubscriberPayloadDto { SubscriberId = userId, Email = email }), }); ``` </Tab> <Tab title="Java"> ```java novu.trigger().body(TriggerEventRequestDto.builder() .workflowId("your-workflow-id") .to(To2.of(SubscriberPayloadDto.builder().subscriberId(userId).email(email).build())) .build()).call(); ``` </Tab> <Tab title="cURL"> ```bash 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": {} }' ``` </Tab> </Tabs>

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.

<AccordionGroup> <Accordion title="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 }}`). </Accordion> </AccordionGroup>

Example Use Case

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

<Columns cols={2}> <Card title="Email Template Variables"> - **Subject:** `Job {{ jobName }} Completed Successfully!` - **Body:** `Hi {{ userName }}, your job '{{ jobName }}' finished processing.` </Card> </Columns>

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

Advanced Trigger Example

<Tabs> <Tab title="Node.js"> ```typescript 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 });

}, });

  </Tab>
  <Tab title="Python">
```python
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}"},
    ))
</Tab> <Tab title="Go"> ```go _, 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) ``` </Tab> <Tab title="PHP"> ```php $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}"], )); ``` </Tab> <Tab title=".NET"> ```csharp 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}" }, }, }); ``` </Tab> <Tab title="Java"> ```java 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(); ``` </Tab> <Tab title="cURL"> ```bash 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" } }' ``` </Tab> </Tabs>

Key Points about Payloads

<Columns cols={2}> <Card title="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. </Card> <Card title="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. </Card> </Columns>

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.

typescript

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
      };
    }
  },
});

<AccordionGroup> <Accordion title="Breakdown">

The Key Components

  1. The AI Generation Task
typescript
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,
  };
},
});
<Note> 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) </Note>

  1. The Two Notification Tasks

Completion Notification

typescript
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

typescript
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...
      },
    });
  },
});

  1. The Main Workflow Orchestrator
typescript
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" };
    }
  },
});
<Note> This task:
  • Responds to the content generation event
  • Tries to generate content
  • Sends appropriate notifications based on success or failure </Note>

Notification Workflows You Would Need in Novu

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

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

  1. ai-generation-error: Sent when content generation fails

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

</Accordion> </AccordionGroup>

Novu trigger calls in the AI tasks example:

<Tabs> <Tab title="Node.js"> ```typescript 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}`, }, }); ``` </Tab> <Tab title="Python"> ```python 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}", }, )) ``` </Tab> <Tab title="Go"> ```go _, 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) ``` </Tab> <Tab title="PHP"> ```php $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}", ], )); ``` </Tab> <Tab title=".NET"> ```csharp 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}" }, }, }); ``` </Tab> <Tab title="Java"> ```java 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(); ``` </Tab> <Tab title="cURL"> ```bash 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" } }' ``` </Tab> </Tabs>

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.

typescript

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() };
    }
  }
};

<AccordionGroup> <Accordion title="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

  1. Transcription Task

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

typescript
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);
  },
});
<Tip> 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 </Tip>

  1. Notification Tasks

There are two notification tasks:

Success Notification

typescript
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

typescript
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]"
      },
    });
  },
});

  1. Main Workflow Orchestrator
typescript
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

  1. 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}}
  1. Trigger the workflow by sending an event:
typescript
// 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 Notifications

Success notification email:

<Note> **Subject**: Your video "Weekly Team Meeting" has been transcribed

Hello 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-789

Thanks for using our service! </Note>

Error notification email:

<Note> **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. </Note>

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

</Accordion> </AccordionGroup>

Novu trigger calls in the video processing example:

<Tabs> <Tab title="Node.js"> ```typescript 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}`, }, }); ``` </Tab> <Tab title="Python"> ```python 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}", }, )) ``` </Tab> <Tab title="Go"> ```go _, 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) ``` </Tab> <Tab title="PHP"> ```php $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}", ], )); ``` </Tab> <Tab title=".NET"> ```csharp 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}" }, }, }); ``` </Tab> <Tab title="Java"> ```java 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(); ``` </Tab> <Tab title="cURL"> ```bash 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" } }' ``` </Tab> </Tabs>

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:

<Steps> <Step> **New User Registration**
When a new user signs up or is added to your system
</Step> <Step> **Notification Eligibility**
When a user becomes eligible to receive notifications
</Step> <Step> **Data Updates**
When you want to ensure subscriber metadata (name, phone, etc.) is up-to-date
</Step> </Steps> <Note> 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. </Note>

Subscriber Creation Example

<Tabs> <Tab title="Node.js"> ```typescript 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 });

}, });

  </Tab>
  <Tab title="Python">
```python
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,
        )
    )
</Tab> <Tab title="Go"> ```go _, err := s.Subscribers.Create(ctx, components.CreateSubscriberRequestDto{ SubscriberID: userID, Email: email, FirstName: firstName, LastName: lastName, Phone: phone, Locale: locale, Avatar: avatar, Data: customData, }, nil) ``` </Tab> <Tab title="PHP"> ```php $sdk->subscribers->create( createSubscriberRequestDto: new Components\CreateSubscriberRequestDto( subscriberId: $userId, email: $email, firstName: $firstName, lastName: $lastName, phone: $phone, locale: $locale, avatar: $avatar, data: $customData, ), ); ``` </Tab> <Tab title=".NET"> ```csharp await sdk.Subscribers.CreateAsync(new CreateSubscriberRequestDto { SubscriberId = userId, Email = email, FirstName = firstName, LastName = lastName, Phone = phone, Locale = locale, Avatar = avatar, Data = customData, }); ``` </Tab> <Tab title="Java"> ```java novu.subscribers().create() .body(CreateSubscriberRequestDto.builder() .subscriberId(userId) .email(email) .firstName(firstName) .lastName(lastName) .phone(phone) .locale(locale) .avatar(avatar) .data(customData) .build()) .call(); ``` </Tab> <Tab title="cURL"> ```bash 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" }' ``` </Tab> </Tabs>

Using Subscriber Data in Templates

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

<Tabs> <Tab title="Basic Fields">Hi `{{subscriber.firstName}}`, welcome!</Tab> <Tab title="Contact Info">We'll reach you at `{{subscriber.phone}}` if needed.</Tab> <Tab title="Custom Data">Your account is set up under `{{subscriber.data.userName}}`.</Tab> </Tabs>

Best Practices

<AccordionGroup> <Accordion title="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:
```typescript
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.
</Accordion> <Accordion title="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
```typescript
// 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
  );
}
```

<Tabs>
  <Tab title="Node.js">
```typescript
await novu.subscribers.patch(
  { email: newData.email, phone: newData.phone, data: { lastUpdated: new Date() } },
  userId
);
```
  </Tab>
  <Tab title="Python">
```python
novu.subscribers.patch(subscriber_id=user_id, patch_subscriber_request_dto={
    "email": new_data.email,
    "phone": new_data.phone,
    "data": {"lastUpdated": datetime.now().isoformat()},
})
```
  </Tab>
  <Tab title="Go">
```go
s.Subscribers.Patch(ctx, userID, components.PatchSubscriberRequestDto{
    Email: &email, Phone: &phone,
    Data: map[string]any{"lastUpdated": time.Now()},
}, nil)
```
  </Tab>
  <Tab title="PHP">
```php
$sdk->subscribers->patch($userId, new Components\PatchSubscriberRequestDto(
    email: $newData->email,
    phone: $newData->phone,
    data: ['lastUpdated' => date('c')],
));
```
  </Tab>
  <Tab title=".NET">
```csharp
await sdk.Subscribers.PatchAsync(userId, new PatchSubscriberRequestDto {
    Email = newData.Email, Phone = newData.Phone,
    Data = new Dictionary<string, object>() { { "lastUpdated", DateTime.UtcNow } },
});
```
  </Tab>
  <Tab title="Java">
```java
novu.subscribers().patch(userId).body(PatchSubscriberRequestDto.builder()
    .email(newData.getEmail()).phone(newData.getPhone())
    .data(Map.of("lastUpdated", Instant.now().toString())).build()).call();
```
  </Tab>
  <Tab title="cURL">
```bash
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" }'
```
  </Tab>
</Tabs>
</Accordion> <Accordion title="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
```typescript
await novu.subscribers.patch(
  {
    data: {
      role: 'admin',
      subscriptionTier: 'premium',
      features: ['advanced-analytics', 'priority-support'],
    },
  },
  userId
);
```

<Tabs>
  <Tab title="Node.js">
```typescript
await novu.subscribers.patch(
  { data: { role: 'admin', subscriptionTier: 'premium', features: ['advanced-analytics', 'priority-support'] } },
  userId
);
```
  </Tab>
  <Tab title="Python">
```python
novu.subscribers.patch(subscriber_id=user_id, patch_subscriber_request_dto={
    "data": {
        "role": "admin",
        "subscriptionTier": "premium",
        "features": ["advanced-analytics", "priority-support"],
    },
})
```
  </Tab>
  <Tab title="Go">
```go
s.Subscribers.Patch(ctx, userID, components.PatchSubscriberRequestDto{
    Data: map[string]any{
        "role": "admin", "subscriptionTier": "premium",
        "features": []string{"advanced-analytics", "priority-support"},
    },
}, nil)
```
  </Tab>
  <Tab title="PHP">
```php
$sdk->subscribers->patch($userId, new Components\PatchSubscriberRequestDto(
    data: [
        'role' => 'admin',
        'subscriptionTier' => 'premium',
        'features' => ['advanced-analytics', 'priority-support'],
    ],
));
```
  </Tab>
  <Tab title=".NET">
```csharp
await sdk.Subscribers.PatchAsync(userId, new PatchSubscriberRequestDto {
    Data = new Dictionary<string, object>() {
        { "role", "admin" }, { "subscriptionTier", "premium" },
        { "features", new[] { "advanced-analytics", "priority-support" } },
    },
});
```
  </Tab>
  <Tab title="Java">
```java
novu.subscribers().patch(userId).body(PatchSubscriberRequestDto.builder()
    .data(Map.of(
        "role", "admin",
        "subscriptionTier", "premium",
        "features", List.of("advanced-analytics", "priority-support")))
    .build()).call();
```
  </Tab>
  <Tab title="cURL">
```bash
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" } }'
```
  </Tab>
</Tabs>
</Accordion> </AccordionGroup>