import { send } from "@vercel/queue";

// Send a message to a topic
await send("my-topic", {
  message: "Hello world",
});

// With additional options
await send(
  "my-topic",
  {
    message: "Hello world",
  },
  {
    idempotencyKey: "unique-key", // Optional: prevent duplicate messages
    retentionSeconds: 3600, // Optional: override retention time (defaults to 24 hours)
    delaySeconds: 60, // Optional: delay message delivery by N seconds
  },
);

// app/api/send-message/route.ts
import { send } from "@vercel/queue";

export async function POST(request: Request) {
  const body = await request.json();

  const { messageId } = await send("my-topic", {
    message: body.message,
  });

  return Response.json({ messageId });
}

// app/api/queue/route.ts
import { handleCallback } from "@vercel/queue";

export const POST = handleCallback({
  // Single topic with one consumer
  "my-topic": {
    "my-consumer": async (message, metadata) => {
      // metadata includes: { messageId, deliveryCount, createdAt, topicName, consumerGroup }
      console.log("Processing message:", message);

      // If this throws an error, the message will be automatically retried
      await processMessage(message);
    },
  },

  // Multiple consumers for different purposes
  "order-events": {
    fulfillment: async (order, metadata) => {
      await processOrder(order);
    },
    analytics: async (order, metadata) => {
      await trackOrder(order);
    },
  },
});

// pages/api/queue.ts
import { handleCallback } from "@vercel/queue/nextjs/pages";

export default handleCallback({
  "my-topic": {
    "my-consumer": async (message, metadata) => {
      console.log("Processing message:", message);
      await processMessage(message);
    },
  },
  "order-events": {
    fulfillment: async (order, metadata) => {
      await processOrder(order);
    },
    analytics: async (order, metadata) => {
      await trackOrder(order);
    },
  },
});

import { Client } from "@vercel/queue";

const client = new Client({
  token: "my-token", // Optional: custom auth token
  headers: { "X-Custom": "header" }, // Optional: custom headers
  pinToDeployment: false, // Optional: disable deployment pinning (default: true)
});

// Send a message
await client.send("my-topic", { hello: "world" });

// Handle callbacks using the same client
export const POST = client.handleCallback({
  "my-topic": {
    "my-group": async (msg, meta) => console.log(msg),
  },
});

import { parseCallback } from "@vercel/queue";

export async function POST(request: Request) {
  const { queueName, consumerGroup, messageId } = await parseCallback(request);

  // Use the parsed information for custom processing...
  await myWorkflow.handleWebhook(queueName, consumerGroup, messageId);

  return Response.json({ status: "success" });
}

import { send, JsonTransport } from "@vercel/queue";

// JsonTransport is the default
await send("json-topic", { data: "example" });

// Explicit transport configuration
await send(
  "json-topic",
  { data: "example" },
  { transport: new JsonTransport() },
);

// JsonTransport with custom serialization
const transport = new JsonTransport({
  replacer: (key, value) => (key === "password" ? undefined : value),
  reviver: (key, value) => (key === "date" ? new Date(value) : value),
});
await send("json-topic", { data: "example" }, { transport });

import {
  BadRequestError,
  ConcurrencyLimitError,
  DuplicateMessageError,
  ForbiddenError,
  InternalServerError,
  UnauthorizedError,
} from "@vercel/queue";

try {
  await send("my-topic", payload);
} catch (error) {
  if (error instanceof UnauthorizedError) {
    console.log("Invalid token - refresh authentication");
  } else if (error instanceof ForbiddenError) {
    console.log("Environment mismatch - check configuration");
  } else if (error instanceof BadRequestError) {
    console.log("Invalid parameters:", error.message);
  } else if (error instanceof DuplicateMessageError) {
    console.log("Duplicate message:", error.idempotencyKey);
  } else if (error instanceof ConcurrencyLimitError) {
    console.log(
      "Rate limited:",
      error.currentInflight,
      "/",
      error.maxConcurrency,
    );
  } else if (error instanceof InternalServerError) {
    console.log("Server error - retry with backoff");
  }
}

import { receive } from "@vercel/queue";

// Process next available message
await receive<T>(topicName, consumerGroup, handler);

// Process specific message by ID
await receive<T>(topicName, consumerGroup, handler, {
  messageId: "message-id",
});

// Process message with options
await receive<T>(topicName, consumerGroup, handler, {
  messageId: "message-id", // Optional: process specific message by ID
  transport: new JsonTransport(), // Optional: custom transport (defaults to JsonTransport)
  visibilityTimeoutSeconds: 30, // Optional: message visibility timeout
  visibilityRefreshInterval: 10, // Optional: how often to refresh the lock
});

// Handler function signature
type MessageHandler<T = unknown> = (
  message: T,
  metadata: MessageMetadata,
) => Promise<void> | void;

// MessageMetadata type
interface MessageMetadata {
  messageId: string;
  deliveryCount: number;
  createdAt: Date;
  topicName: string;
  consumerGroup: string;
}

import { Client } from "@vercel/queue";

const client = new Client({
  // Base URL for the queue service
  // Default: "https://vercel-queue.com"
  // Env: VERCEL_QUEUE_BASE_URL
  baseUrl: "https://vercel-queue.com",

  // API path prefix
  // Default: "/api/v3/topic"
  // Env: VERCEL_QUEUE_BASE_PATH
  basePath: "/api/v3/topic",

  // Auth token (auto-fetched via OIDC if not provided)
  token: "my-token",

  // Custom headers for all requests
  headers: { "X-Custom": "value" },

  // Deployment ID for message routing
  // Default: process.env.VERCEL_DEPLOYMENT_ID
  deploymentId: "dpl_xxx",

  // Pin messages to current deployment when publishing
  // Default: true
  pinToDeployment: true,
});

await send("my-topic", payload, {
  // Deduplication key
  // Dedup window: min(retentionSeconds, 24 hours)
  idempotencyKey: "unique-key",

  // Message TTL in seconds
  // Default: 86400, Min: 60, Max: 86400
  retentionSeconds: 3600,

  // Delay before message becomes visible
  // Default: 0, Min: 0, Max: retentionSeconds
  delaySeconds: 60,

  // Custom serializer (default: JsonTransport)
  transport: new JsonTransport(),
});

await receive("my-topic", "my-consumer", handler, {
  // Specific message ID to consume (optional)
  messageId: "0-1",

  // Message lock duration in seconds
  // Default: 30, Min: 0, Max: 3600
  visibilityTimeoutSeconds: 60,

  // How often to refresh the lock during processing
  // Default: visibilityTimeoutSeconds / 3
  visibilityRefreshInterval: 15,

  // Custom deserializer (default: JsonTransport)
  transport: new JsonTransport(),
});

await receive("my-topic", "my-consumer", handler, {
  // Payload deserializer
  // Default: JsonTransport
  transport: new JsonTransport(),

  // Message lock duration
  // Default: 30, Min: 0, Max: 3600
  visibilityTimeoutSeconds: 60,

  // How often to refresh the lock during processing
  // Default: visibilityTimeoutSeconds / 3
  visibilityRefreshInterval: 20,
});

export const POST = handleCallback(
  {
    "my-topic": {
      "my-consumer": async (message, metadata) => {
        await processMessage(message);
      },
    },
  },
  {
    // Message lock duration for long-running handlers
    // Default: 30, Min: 0, Max: 3600
    // visibilityRefreshInterval defaults to visibilityTimeoutSeconds / 3
    visibilityTimeoutSeconds: 300, // 5 minutes
  },
);