Inngest

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
import (
    novugo "github.com/novuhq/novu-go"
    "os"
)

novu := novugo.New(novugo.WithSecurity(os.Getenv("NOVU_SECRET_KEY")))
```
  </Tab>
  <Tab title="PHP">
```php
use novu;

$novu = novu\Novu::builder()
    ->setSecurity(getenv('NOVU_SECRET_KEY'))
    ->build();
```
  </Tab>
  <Tab title=".NET">
```csharp
using Novu;

var novu = new NovuSDK(secretKey: Environment.GetEnvironmentVariable("NOVU_SECRET_KEY")!);
```
  </Tab>
  <Tab title="Java">
```java
import co.novu.Novu;

Novu novu = Novu.builder()
    .secretKey(System.getenv("NOVU_SECRET_KEY"))
    .build();
```
  </Tab>
</Tabs>

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

- Trigger Notification Workflow
- Add New Subscriber

return { message: `Welcome email sent to ${event.data.email}!` };

  </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="welcome-email",
        to={"subscriber_id": user_id, "email": email},
        payload={},
    ))

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

  return novuPayload;
});

  </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="inbox-test",
        to={"subscriber_id": user_id, "email": email},
        payload={
            "userName": user_name,
            "functionName": f"Data Processing Function #{event_id}",
        },
    ))

  await novu.subscribers.create(subscriberPayload);
  return subscriberPayload;
});

  </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.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,
            avatar=avatar,
            locale=locale,
            data={"userName": user_name} if user_name else None,
        )
    )

import { Inngest } from "inngest";
import { Novu } from "@novu/api";

// --- Mock/Placeholder External Services ---
// Replace these with your actual implementations
const deepgram = {
  transcribe: async (url: string): Promise<string> => {
    console.log(`Mock Transcribing: ${url}`);
    await new Promise(resolve => setTimeout(resolve, 1500)); // Simulate network delay
    return `This is a mock transcript for the video at ${url}. It contains several sentences explaining the content.`;
  }
};

const llm = {
  createCompletion: async (params: { model: string; prompt: string }): Promise<string> => {
    console.log(`Mock Summarizing with ${params.model}`);
    await new Promise(resolve => setTimeout(resolve, 1000)); // Simulate LLM processing time
    const transcriptSnippet = params.prompt.substring(params.prompt.indexOf(":") + 1, 70);
    return `Mock summary of the transcript starting with: ${transcriptSnippet}...`;
  }
};

const createSummaryPrompt = (transcript: string): string => {
  return `Please summarize the following transcript:\n\n${transcript}`;
};

const db = {
  videoSummaries: {
    upsert: async (data: { videoId: string; transcript: string; summary: string }): Promise<void> => {
      console.log(`Mock DB Upsert for videoId: ${data.videoId}`);
      // In a real scenario, save data.transcript and data.summary linked to data.videoId
      await new Promise(resolve => setTimeout(resolve, 500)); // Simulate DB write delay
    }
  }
};
// --- End Mock Services ---

// --- Inngest Client Setup ---
// Assumes you have your Inngest client initialized elsewhere (e.g., './inngest/client')
// For this example, we'll initialize it here. Replace with your actual client setup.
export const inngest = new Inngest({ id: "video-processing-app" });
// --- End Inngest Client Setup ---

// --- Novu Client Setup ---
const novu = new Novu({ secretKey: process.env.NOVU_SECRET_KEY });
// --- End Novu Client Setup ---

// --- Combined Inngest Function ---
export const processVideoAndNotify = inngest.createFunction(
  {
    id: "process-video-and-notify",
    name: "Process Uploaded Video and Notify User",
    // Apply concurrency limits per user to avoid overwhelming resources
    concurrency: {
      limit: 3, // Allow up to 3 concurrent processing jobs per user
      key: "event.data.userId"
    }
  },
  { event: "video/uploaded" }, // Triggered when a video upload event occurs
  async ({ event, step }) => {
    const { videoUrl, videoId, userId, email, userName } = event.data;

    // Validate required data from the event
    if (!videoUrl || !videoId || !userId || !email || !userName) {
        throw new Error(`Missing required data in 'video/uploaded' event payload for event ID: ${event.id}`);
    }

    // --- Step 1: Transcribe Video ---
    // step.run ensures this block retries on failure and runs exactly once on success.
    const transcript = await step.run('transcribe-video', async () => {
      console.log(`Starting transcription for videoId: ${videoId}, url: ${videoUrl}`);
      try {
        const result = await deepgram.transcribe(videoUrl);
        console.log(`Transcription successful for videoId: ${videoId}`);
        return result;
      } catch (error) {
        console.error(`Transcription failed for videoId: ${videoId}`, error);
        throw error; // Re-throw to enable Inngest retries
      }
    });

    // --- Step 2: Summarize Transcript ---
    // The 'transcript' result from the previous step is automatically available here.
    const summary = await step.run('summarize-transcript', async () => {
      console.log(`Starting summarization for videoId: ${videoId}`);
      try {
        const result = await llm.createCompletion({
          model: "gpt-4o", // Or your preferred model
          prompt: createSummaryPrompt(transcript),
        });
        console.log(`Summarization successful for videoId: ${videoId}`);
        return result;
      } catch (error) {
        console.error(`Summarization failed for videoId: ${videoId}`, error);
        throw error; // Re-throw for retries
      }
    });

    // --- Step 3: Write Results to Database ---
    await step.run('write-summary-to-db', async () => {
      console.log(`Writing transcript and summary to DB for videoId: ${videoId}`);
      try {
        await db.videoSummaries.upsert({
          videoId: videoId,
          transcript,
          summary,
        });
        console.log(`DB write successful for videoId: ${videoId}`);
      } catch (error) {
        console.error(`DB write failed for videoId: ${videoId}`, error);
        throw error; // Re-throw for retries
      }
    });

    // --- Step 4: Send Completion Notification via Novu ---
    // This step runs only after the previous steps have succeeded.
    await step.run("send-completion-notification", async () => {
      if (!novuApiKey) {
        console.warn(`Skipping notification for videoId ${videoId} as NOVU_SECRET_KEY is not set.`);
        return { skipped: true, reason: "NOVU_SECRET_KEY not set" };
      }

      console.log(`Attempting to send Novu notification for videoId: ${videoId} to user: ${userId}`);

      const novuPayload = {
        userName: userName,
        videoId: videoId,
        summarySnippet: summary.substring(0, 100) + (summary.length > 100 ? "..." : ""),
      };

      try {
        const triggerResult = await novu.trigger({
          workflowId: "video-processed-notification",
          to: { subscriberId: userId, email: email },
          payload: novuPayload,
        });

        console.log(`Novu workflow 'video-processed-notification' triggered successfully for videoId: ${videoId}`, triggerResult);
        return { success: true, novuPayload: novuPayload, triggerResult };

      } catch (error: unknown) {
        const errorMessage = error instanceof Error ? error.message : 'Unknown error triggering Novu';
        console.error(`Failed to trigger Novu workflow for videoId: ${videoId}: ${errorMessage}`, error);
        // Decide if notification failure should cause the step to fail and retry
        // For notifications, often you might want to log the error but not fail the whole function
        // throw new Error(`Novu trigger failed: ${errorMessage}`); // Uncomment to make it retry
         return { success: false, error: errorMessage }; // Log and continue
      }
    });

    // --- Function Completion ---
    // Optional: return a final status or summary
    return {
      message: `Video processing and notification steps completed for videoId: ${videoId}`,
      videoId: videoId,
      userId: userId,
      summaryLength: summary.length,
    };
  }
);

// --- How to Trigger (Example) ---
/*
Somewhere else in your application (e.g., after a file upload completes):

import { inngest } from "./path/to/inngest/client"; // Your initialized Inngest client

async function handleVideoUpload(uploadResult: { url: string, id: string, user: { id: string, email: string, name: string } }) {
  await inngest.send({
    name: "video/uploaded", // The event name our function listens to
    data: {
      videoUrl: uploadResult.url,
      videoId: uploadResult.id,
      userId: uploadResult.user.id,
      email: uploadResult.user.email,
      userName: uploadResult.user.name,
    },
    user: { // Optionally pass user context if needed for event metadata/tracking
        id: uploadResult.user.id,
        email: uploadResult.user.email,
    }
  });
  console.log(`Sent 'video/uploaded' event for videoId: ${uploadResult.id}`);
}

// Example usage:
handleVideoUpload({
    url: "https://example.com/videos/123.mp4",
    id: "vid_12345abc",
    user: { id: "user_67890def", email: "[email protected]", name: "Alice" }
});
*/