Implementation:Langfuse Langfuse BullMQ Queue Definitions

Knowledge Sources	Langfuse
Domains	Queues, Worker
Last Updated	2026-02-14 00:00 GMT

Overview

Defines all BullMQ queue names, job names, event payload schemas, and type-safe queue job type mappings used by the Langfuse worker and web application for asynchronous job processing.

Description

This module is the single source of truth for the Langfuse queue system. It provides:

Event Payload Schemas (Zod v4): Each queue has a corresponding Zod schema that validates its job payload:

• IngestionEvent / OtelIngestionEvent: Ingestion pipeline events with auth context and S3 file references
• TraceQueueEventSchema / TracesQueueEventSchema: Trace upsert and delete events
• ScoresQueueEventSchema: Score deletion events
• DatasetQueueEventSchema: Dataset/dataset-run deletion (discriminated union)
• ProjectQueueEventSchema: Project deletion events
• DatasetRunItemUpsertEventSchema: Dataset run item upsert events
• EvalExecutionEvent / LLMAsJudgeExecutionEventSchema: Evaluation execution events
• BatchExportJobSchema: Batch export jobs
• ExperimentCreateEventSchema: Experiment creation events
• BatchActionProcessingEventSchema: Batch actions (score-delete, trace-delete, add-to-annotation-queue, eval-create, observation-add-to-dataset) as a discriminated union
• CreateEvalQueueEventSchema: Eval creation events (union of dataset and trace triggers)
• NotificationEventSchema: Notifications (currently comment mentions)
• WebhookInputSchema / EntityChangeEventSchema: Webhook and entity change events
• Integration events: PostHog, Mixpanel, BlobStorage processing
• DataRetentionProcessingEventSchema: Data retention cleanup
• RetryBaggage: Retry metadata for jobs with retry support

Enums:

• QueueName: 30 queue name constants (e.g., TraceUpsert, IngestionQueue, EvaluationExecution, BatchExport)
• QueueJobs: 30 corresponding job name constants

Type-safe job mapping: TQueueJobTypes is a mapped type that associates each QueueName with its exact payload type, job name, and metadata shape. This enables type-safe queue producers and consumers throughout the codebase.

Usage

Use QueueName and QueueJobs when creating BullMQ queue instances and adding jobs. Use the Zod schemas to validate job payloads before enqueueing. Use TQueueJobTypes to type queue producers and consumers for compile-time safety.

Code Reference

Source Location

• Repository: Langfuse
• File: packages/shared/src/server/queues.ts
• Lines: 1-527

Signature

// Event payload schemas
export const IngestionEvent: z.ZodObject<...>;
export const OtelIngestionEvent: z.ZodObject<...>;
export const TraceQueueEventSchema: z.ZodObject<...>;
export const TracesQueueEventSchema: z.ZodObject<...>;
export const ScoresQueueEventSchema: z.ZodObject<...>;
export const DatasetQueueEventSchema: z.ZodDiscriminatedUnion<...>;
export const BatchExportJobSchema: z.ZodObject<...>;
export const EvalExecutionEvent: z.ZodObject<...>;
export const LLMAsJudgeExecutionEventSchema: z.ZodObject<...>;
export const BatchActionProcessingEventSchema: z.ZodDiscriminatedUnion<...>;
export const CreateEvalQueueEventSchema: z.ZodUnion<...>;
export const NotificationEventSchema: z.ZodDiscriminatedUnion<...>;
export const WebhookInputSchema: z.ZodObject<...>;
export const EntityChangeEventSchema: z.ZodDiscriminatedUnion<...>;
export const RetryBaggage: z.ZodObject<...>;
// ... additional schemas

// Queue and job name enums
export enum QueueName {
  TraceUpsert = "trace-upsert",
  TraceDelete = "trace-delete",
  IngestionQueue = "ingestion-queue",
  OtelIngestionQueue = "otel-ingestion-queue",
  EvaluationExecution = "evaluation-execution-queue",
  BatchExport = "batch-export-queue",
  // ... 24 more queue names
}

export enum QueueJobs {
  TraceUpsert = "trace-upsert",
  TraceDelete = "trace-delete",
  IngestionJob = "ingestion-job",
  OtelIngestionJob = "otel-ingestion-job",
  EvaluationExecution = "evaluation-execution-job",
  BatchExportJob = "batch-export-job",
  // ... 24 more job names
}

// Type-safe queue job mapping
export type TQueueJobTypes = {
  [QueueName.TraceUpsert]: {
    timestamp: Date;
    id: string;
    payload: TraceQueueEventType;
    name: QueueJobs.TraceUpsert;
  };
  [QueueName.IngestionQueue]: {
    timestamp: Date;
    id: string;
    payload: IngestionEventQueueType;
    name: QueueJobs.IngestionJob;
  };
  // ... entries for all 30 queues
};

// Inferred payload types
export type TraceQueueEventType = z.infer<typeof TraceQueueEventSchema>;
export type IngestionEventQueueType = z.infer<typeof IngestionEvent>;
export type EvalExecutionEventType = z.infer<typeof EvalExecutionEvent>;
// ... additional inferred types

Import

import {
  QueueName,
  QueueJobs,
  type TQueueJobTypes,
  type TraceQueueEventType,
  TraceQueueEventSchema,
} from "@langfuse/shared/src/server";

I/O Contract

Inputs

Name	Type	Required	Description
payload	(varies by queue)	Yes	Job-specific payload validated against the corresponding Zod schema. For example, TraceQueueEventSchema requires projectId and traceId.
timestamp	Date	Yes	Job creation timestamp, present in all queue job types
id	string	Yes	Unique job identifier, present in all queue job types
retryBaggage	RetryBaggage	No	Optional retry metadata (originalJobTimestamp, attempt) for queues that support retries

Outputs

Name	Type	Description
QueueName	enum	String enum of all queue names for BullMQ queue instantiation
QueueJobs	enum	String enum of all job names for BullMQ job naming
TQueueJobTypes	mapped type	Type-safe mapping from queue name to its complete job shape (payload, name, timestamp, id)
*EventType	inferred types	TypeScript types inferred from each Zod schema for use in producers and consumers

Usage Examples

import { QueueName, QueueJobs, type TQueueJobTypes, TraceQueueEventSchema } from "@langfuse/shared/src/server";
import { Queue } from "bullmq";

// Create a typed queue
const traceUpsertQueue = new Queue<TQueueJobTypes[QueueName.TraceUpsert]>(
  QueueName.TraceUpsert,
  { connection: redisConnection },
);

// Add a validated job
const payload = TraceQueueEventSchema.parse({
  projectId: "project-123",
  traceId: "trace-abc",
  traceEnvironment: "production",
});

await traceUpsertQueue.add(QueueJobs.TraceUpsert, {
  timestamp: new Date(),
  id: crypto.randomUUID(),
  payload,
  name: QueueJobs.TraceUpsert,
});

// Type-safe consumer
import { Worker } from "bullmq";
const worker = new Worker<TQueueJobTypes[QueueName.TraceUpsert]>(
  QueueName.TraceUpsert,
  async (job) => {
    const { projectId, traceId } = job.data.payload; // fully typed
    await processTraceUpsert(projectId, traceId);
  },
  { connection: redisConnection },
);

Related Pages

• Environment:Langfuse_Langfuse_Redis_7_Queue_Cache