OpenTelemetry instrumentation for the Upstash QStash Node.js SDK. Capture spans for every QStash API call, enrich them with operation metadata, and keep an eye on message queue operations from your traces.

Visualize your message queue operations with detailed span information including message publishing, callbacks, and delivery tracking.

Installation

npm install @kubiks/otel-upstash-queues
# or
pnpm add @kubiks/otel-upstash-queues

Peer Dependencies: @opentelemetry/api >= 1.9.0, @upstash/qstash >= 2.0.0

Quick Start

Publishing Messages

import { Client } from "@upstash/qstash";
import { instrumentUpstash } from "@kubiks/otel-upstash-queues";

const client = instrumentUpstash(
  new Client({ token: process.env.QSTASH_TOKEN! })
);

await client.publishJSON({
  url: "https://your-api-endpoint.com/process-image",
  body: { imageId: "123" },
});

instrumentUpstash wraps the QStash client instance you already use — no configuration changes needed. Every SDK call creates a client span with useful attributes.

Consuming Messages

// app/api/process/route.ts
import { verifySignatureAppRouter } from "@upstash/qstash/nextjs";
import { instrumentConsumer } from "@kubiks/otel-upstash-queues";

async function handler(request: Request) {
  const data = await request.json();
  
  // Process your message
  await processImage(data.imageId);
  
  return Response.json({ success: true });
}

// Instrument first, then verify signature
export const POST = verifySignatureAppRouter(instrumentConsumer(handler));

instrumentConsumer wraps your message handler to trace message consumption, creating a SERVER span for each message received and processed.

With Body Capture

Optionally capture request/response bodies for debugging:

const client = instrumentUpstash(
  new Client({ token: process.env.QSTASH_TOKEN! }),
  {
    captureBody: true,      // Enable body capture (default: false)
    maxBodyLength: 2048,    // Max characters to capture (default: 1024)
  }
);

What Gets Traced

This instrumentation provides two main functions:

instrumentUpstash - Wraps the QStash client to trace message publishing
instrumentConsumer - Wraps your message handler to trace message consumption

Publisher Instrumentation

The instrumentUpstash function wraps the client.publishJSON method, creating a span with SpanKind.CLIENT for each message publish operation.

Consumer Instrumentation

The instrumentConsumer function wraps your message handler, creating a span with SpanKind.SERVER for each message received and processed.

Span Attributes

Publisher Spans (`instrumentUpstash`)

Attribute	Description	Example
`messaging.system`	Constant value `qstash`	`qstash`
`messaging.operation`	Operation type	`publish`
`qstash.resource`	Resource name	`messages`
`qstash.target`	Full operation target	`messages.publish`
`qstash.url`	Target URL for the message	`https://example.com/api/process`
`qstash.method`	HTTP method (default: POST)	`POST`, `PUT`, `GET`
`qstash.message_id`	Message ID returned by QStash	`msg_123`
`qstash.delay`	Delay before processing (seconds or string)	`60` or `"1h"`
`qstash.not_before`	Unix timestamp for earliest processing	`1672531200`
`qstash.deduplication_id`	Deduplication ID for idempotent operations	`unique-id-123`
`qstash.retries`	Number of retry attempts (max)	`3`
`qstash.callback_url`	Success callback URL	`https://example.com/callback`
`qstash.failure_callback_url`	Failure callback URL	`https://example.com/failure`

Consumer Spans (`instrumentConsumer`)

Attribute	Description	Example
`messaging.system`	Constant value `qstash`	`qstash`
`messaging.operation`	Operation type	`receive`
`qstash.resource`	Resource name	`messages`
`qstash.target`	Full operation target	`messages.receive`
`qstash.message_id`	Message ID from QStash	`msg_456`
`qstash.retried`	Number of times retried (actual count)	`2`
`qstash.schedule_id`	Schedule ID (if from scheduled message)	`schedule_123`
`qstash.caller_ip`	IP address of the caller	`192.168.1.1`
`http.status_code`	HTTP response status code	`200`

Body/Payload Attributes (Optional)

When captureBody is enabled in configuration:

Attribute	Description	Captured By
`qstash.request.body`	Request/message body content	Both publisher and consumer
`qstash.response.body`	Response body content	Consumer only

The instrumentation captures message metadata and configuration to help with debugging and monitoring. Body capture is disabled by default to protect sensitive data.

Usage Examples

Basic Message Publishing

import { Client } from "@upstash/qstash";
import { instrumentUpstash } from "@kubiks/otel-upstash-queues";

const client = instrumentUpstash(
  new Client({ token: process.env.QSTASH_TOKEN! })
);

// Publish a message
await client.publishJSON({
  url: "https://your-api.com/webhook",
  body: {
    userId: "user_123",
    action: "process_data",
  },
});

Delayed Message Publishing

// Delay message processing by 60 seconds
await client.publishJSON({
  url: "https://your-api.com/delayed-task",
  body: { taskId: "task_456" },
  delay: 60,
});

// Or use a human-readable delay
await client.publishJSON({
  url: "https://your-api.com/delayed-task",
  body: { taskId: "task_789" },
  delay: "1h", // 1 hour
});

Message with Callbacks

await client.publishJSON({
  url: "https://your-api.com/process",
  body: { orderId: "order_123" },
  callback: "https://your-api.com/success",
  failureCallback: "https://your-api.com/failure",
});

Message with Retries and Deduplication

await client.publishJSON({
  url: "https://your-api.com/critical-task",
  body: { taskId: "critical_123" },
  retries: 5,
  deduplicationId: "task-critical-123",
});

Scheduled Message

// Schedule for a specific time
const scheduledTime = Math.floor(Date.now() / 1000) + 3600; // 1 hour from now

await client.publishJSON({
  url: "https://your-api.com/scheduled-task",
  body: { reportId: "report_456" },
  notBefore: scheduledTime,
});

Message Consumer Instrumentation

Use instrumentConsumer to trace your message handler that receives QStash messages:

// app/api/process/route.ts
import { verifySignatureAppRouter } from "@upstash/qstash/nextjs";
import { instrumentConsumer } from "@kubiks/otel-upstash-queues";

async function handler(request: Request) {
  const data = await request.json();
  
  // Process your message
  await processImage(data.imageId);
  
  return Response.json({ success: true });
}

// Instrument first, then verify signature
export const POST = verifySignatureAppRouter(instrumentConsumer(handler));

With body capture:

export const POST = verifySignatureAppRouter(
  instrumentConsumer(handler, {
    captureBody: true,
    maxBodyLength: 2048,
  })
);

The instrumentConsumer function:

Extracts QStash headers (message ID, retry count, schedule ID, caller IP)
Creates a SERVER span for the message processing
Tracks response status codes
Captures errors during processing
Optionally captures request and response bodies (when configured)

Complete Next.js Integration Example

Publishing messages:

// app/actions.ts
"use server";
import { Client } from "@upstash/qstash";
import { instrumentUpstash } from "@kubiks/otel-upstash-queues";

const qstashClient = instrumentUpstash(
  new Client({
    token: process.env.QSTASH_TOKEN!,
  })
);

export async function startBackgroundJob(imageId: string) {
  await qstashClient.publishJSON({
    url: "https://your-app.vercel.app/api/process",
    body: { imageId },
  });
}

Receiving messages:

// app/api/process/route.ts
import { verifySignatureAppRouter } from "@upstash/qstash/nextjs";
import { instrumentConsumer } from "@kubiks/otel-upstash-queues";

async function handler(request: Request) {
  const { imageId } = await request.json();
  
  // Your processing logic
  await processImage(imageId);
  
  return Response.json({ success: true });
}

export const POST = verifySignatureAppRouter(instrumentConsumer(handler));

License

MIT

README.md

@kubiks/otel-upstash-queues