Send logs from Convex to SigNoz

mkdir convex-signoz-bridge
cd convex-signoz-bridge
npm init -y
npm install express

import express from "express";
import crypto from "node:crypto";

const app = express();
const port = process.env.PORT || 3080;

// Parse all request bodies as text for signature verification
app.use(
  express.text({
    type: "*/*",
    limit: "5mb",
  })
);

app.post("/webhook", async (req, res) => {
  const textPayload = req.body;
  const signature = req.headers["x-webhook-signature"];

  if (!signature) {
    return res.status(401).send("Unauthorized: Missing signature");
  }

  if (!process.env.WEBHOOK_SECRET) {
    console.error("Missing WEBHOOK_SECRET environment variable");
    return res.status(500).send("Server configuration error");
  }

  try {
    // 1. Verify HMAC-SHA256 Signature
    const hmacSecret = await crypto.subtle.importKey(
      "raw",
      new TextEncoder().encode(process.env.WEBHOOK_SECRET),
      { name: "HMAC", hash: "SHA-256" },
      false,
      ["verify"]
    );

    const isValid = await crypto.subtle.verify(
      "HMAC",
      hmacSecret,
      Buffer.from(signature.replace("sha256=", ""), "hex"),
      new TextEncoder().encode(textPayload)
    );

    if (!isValid) {
      return res.status(401).send("Unauthorized: Invalid signature");
    }

    // 2. Parse Payload (JSON or JSONL)
    let logs;
    try {
      logs = JSON.parse(textPayload);
      if (!Array.isArray(logs)) logs = [logs];
    } catch {
      // Try JSONL format
      logs = textPayload
        .split("\n")
        .filter((line) => line.trim())
        .map((line) => JSON.parse(line));
    }

    // 3. Handle Verification Ping
    if (logs.length === 1 && logs[0].topic === "verification") {
      console.log("Received verification ping from Convex");
      return res.send("Verification successful");
    }

    // 4. Replay Attack Protection
    const firstLog = logs[0];
    if (firstLog.timestamp && firstLog.timestamp < Date.now() - 5 * 60 * 1000) {
      return res.status(403).send("Request expired");
    }

    // 5. Transform and Send to SigNoz
    await sendToSigNoz(logs);
    res.send("Success");
  } catch (error) {
    console.error("Error processing webhook:", error);
    res.status(500).send("Internal server error");
  }
});

async function sendToSigNoz(logs) {
  const ingestionKey = process.env.SIGNOZ_INGESTION_KEY;
  const serviceName = process.env.SERVICE_NAME || "convex-logs";

  if (!ingestionKey) {
    console.warn("Skipping SigNoz export: Missing SIGNOZ_INGESTION_KEY");
    return;
  }

  const url = `https://ingest.<region>.signoz.cloud:443/logs/json`;

  const transformedLogs = logs.map((log) => {
    const newLog = { ...log };

    // Set resource attributes
    newLog.resources = {
      "service.name": serviceName,
    };

    // Map Convex log_level to OTel severity_text
    const severityMap = {
      DEBUG: "DEBUG",
      INFO: "INFO",
      LOG: "INFO",
      WARN: "WARN",
      ERROR: "ERROR",
    };
    newLog.severity_text = severityMap[log.log_level] || "INFO";

    // Add a 'body' field for the log message
    if (!newLog.body) {
      if (newLog.message) {
        newLog.body = newLog.message;
      } else if (newLog.topic === "function_execution" && newLog.function) {
        newLog.body = `Function: ${newLog.function.path} (${newLog.status || "unknown"}) in ${newLog.execution_time_ms ?? "?"}ms`;
      } else if (newLog.topic === "audit_log") {
        newLog.body = `Audit: ${newLog.audit_log_action}`;
      } else {
        newLog.body = `Convex: ${newLog.topic}`;
      }
    }

    // Structure attributes for querying in SigNoz
    newLog.attributes = {
      "convex.topic": newLog.topic,
      "convex.deployment": newLog.convex?.deployment_name,
      "convex.project": newLog.convex?.project_name,
      ...newLog.attributes,
    };

    return newLog;
  });

  const response = await fetch(url, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "signoz-ingestion-key": ingestionKey,
    },
    body: JSON.stringify(transformedLogs),
  });

  if (!response.ok) {
    const text = await response.text();
    console.error(`Failed to send to SigNoz: ${response.status} ${text}`);
  } else {
    console.log(`Sent ${logs.length} logs to SigNoz`);
  }
}

app.listen(port, () => {
  console.log(`Webhook Bridge running on port ${port}`);
});

{
  "type": "module"
}

export REGION="<region>"
export SIGNOZ_INGESTION_KEY="<your-ingestion-key>"
export WEBHOOK_SECRET="<your-convex-secret>"
export SERVICE_NAME="<service-name>"  # Optional
export PORT="<port>"  # Optional

npm run start

curl -X POST http://localhost:3080/webhook -d '{"topic":"test"}' -H "x-webhook-signature: sha256=invalid"
# Expected: "Unauthorized: Invalid signature"