Langfuse Webhooks Events

// app/api/webhooks/langfuse/route.ts (Next.js App Router)
import { NextRequest, NextResponse } from "next/server";
import crypto from "crypto";

const WEBHOOK_SECRET = process.env.LANGFUSE_WEBHOOK_SECRET!;

interface LangfuseWebhookEvent {
  event: "prompt.created" | "prompt.updated" | "prompt.deleted";
  timestamp: string;
  data: {
    promptName: string;
    promptVersion: number;
    labels?: string[];
    projectId: string;
    [key: string]: any;
  };
}

// Verify HMAC SHA-256 signature
function verifySignature(payload: string, signature: string): boolean {
  const expected = crypto
    .createHmac("sha256", WEBHOOK_SECRET)
    .update(payload)
    .digest("hex");

  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expected)
  );
}

export async function POST(request: NextRequest) {
  const payload = await request.text();
  const signature = request.headers.get("x-langfuse-signature");

  // Verify webhook authenticity
  if (!signature || !verifySignature(payload, signature)) {
    return NextResponse.json({ error: "Invalid signature" }, { status: 401 });
  }

  const event: LangfuseWebhookEvent = JSON.parse(payload);
  console.log(`Langfuse webhook: ${event.event} - ${event.data.promptName}`);

  switch (event.event) {
    case "prompt.created":
      await handlePromptCreated(event.data);
      break;
    case "prompt.updated":
      await handlePromptUpdated(event.data);
      break;
    case "prompt.deleted":
      await handlePromptDeleted(event.data);
      break;
  }

  return NextResponse.json({ received: true });
}

async function handlePromptCreated(data: LangfuseWebhookEvent["data"]) {
  // Trigger CI/CD pipeline for new prompt version
  if (data.labels?.includes("production")) {
    await triggerPromptDeployPipeline(data.promptName, data.promptVersion);
  }

  await notifySlack({
    text: `New prompt version: *${data.promptName}* v${data.promptVersion}`,
    labels: data.labels,
  });
}

async function handlePromptUpdated(data: LangfuseWebhookEvent["data"]) {
  // Label change -- check if promoted to production
  if (data.labels?.includes("production")) {
    await notifySlack({
      text: `Prompt *${data.promptName}* v${data.promptVersion} promoted to production`,
    });
  }
}

async function handlePromptDeleted(data: LangfuseWebhookEvent["data"]) {
  await notifySlack({
    text: `Prompt *${data.promptName}* v${data.promptVersion} deleted`,
    level: "warning",
  });
}

// lib/slack-notify.ts

async function notifySlack(params: {
  text: string;
  labels?: string[];
  level?: "info" | "warning";
}) {
  const color = params.level === "warning" ? "#ff9800" : "#36a64f";

  await fetch(process.env.SLACK_WEBHOOK_URL!, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      attachments: [
        {
          color,
          blocks: [
            {
              type: "section",
              text: { type: "mrkdwn", text: params.text },
            },
            ...(params.labels
              ? [
                  {
                    type: "context",
                    elements: [
                      {
                        type: "mrkdwn",
                        text: `Labels: ${params.labels.join(", ")}`,
                      },
                    ],
                  },
                ]
              : []),
          ],
        },
      ],
    }),
  });
}

// Trigger GitHub Actions workflow when production prompt changes
async function triggerPromptDeployPipeline(promptName: string, version: number) {
  await fetch(
    `https://api.github.com/repos/${process.env.GITHUB_REPO}/dispatches`,
    {
      method: "POST",
      headers: {
        Authorization: `Bearer ${process.env.GITHUB_TOKEN}`,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        event_type: "prompt-updated",
        client_payload: { promptName, version },
      }),
    }
  );
}

# .github/workflows/prompt-deploy.yml
name: Deploy Updated Prompt
on:
  repository_dispatch:
    types: [prompt-updated]

jobs:
  test-and-deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-node@v4
        with: { node-version: "20", cache: "npm" }
      - run: npm ci

      - name: Test updated prompt
        env:
          LANGFUSE_PUBLIC_KEY: ${{ secrets.LANGFUSE_PUBLIC_KEY }}
          LANGFUSE_SECRET_KEY: ${{ secrets.LANGFUSE_SECRET_KEY }}
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          echo "Testing prompt: ${{ github.event.client_payload.promptName }}"
          npx vitest run tests/ai/prompt-quality.test.ts

// For production: queue webhook processing to return 200 fast
import { Queue, Worker } from "bullmq";

const webhookQueue = new Queue("langfuse-webhooks", {
  connection: { host: process.env.REDIS_HOST },
});

// In webhook handler -- enqueue and respond immediately
export async function POST(request: NextRequest) {
  const payload = await request.text();
  // ... verify signature ...

  await webhookQueue.add("process", JSON.parse(payload), {
    attempts: 3,
    backoff: { type: "exponential", delay: 1000 },
  });

  return NextResponse.json({ received: true }); // Return fast
}

// Worker processes asynchronously
const worker = new Worker(
  "langfuse-webhooks",
  async (job) => {
    const event = job.data as LangfuseWebhookEvent;
    // Process event...
  },
  { connection: { host: process.env.REDIS_HOST } }
);