undefined

**Key Principles:**

- **Web**: tRPC procedures for UI OR public API routes for SDKs → Services → Database
- **Worker**: Queue processors → Services → Database
- **packages/shared**: Shared code for Web and Worker

See [architecture-overview.md](architecture-overview.md) for complete details.

---

web/src/
├── features/                # Feature-organized code
│   ├── [feature-name]/
│   │   ├── server/          # Backend logic
│   │   │   ├── *Router.ts   # tRPC router
│   │   │   └── service.ts   # Business logic
│   │   ├── components/      # React components
│   │   └── types/           # Feature types
├── server/
│   ├── api/
│   │   ├── routers/         # tRPC routers
│   │   ├── trpc.ts          # tRPC setup & middleware
│   │   └── root.ts          # Main router
│   ├── auth.ts              # NextAuth.js config
│   └── db.ts                # Database client
├── pages/
│   ├── api/
│   │   ├── public/          # Public REST APIs
│   │   └── trpc/            # tRPC endpoint
│   └── [routes].tsx         # Next.js pages
├── __tests__/               # Jest tests
│   └── async/               # Integration tests
├── instrumentation.ts       # OpenTelemetry (FIRST IMPORT)
└── env.mjs                  # Environment config

worker/src/
├── queues/                  # BullMQ processors
│   ├── evalQueue.ts
│   ├── ingestionQueue.ts
│   └── workerManager.ts
├── features/                # Business logic
│   └── [feature]/
│       └── service.ts
├── instrumentation.ts       # OpenTelemetry (FIRST IMPORT)
├── app.ts                   # Express setup + queue registration
├── env.ts                   # Environment config
└── index.ts                 # Server start

shared/src/
├── server/                  # Server utilities
│   ├── auth/                # Authentication helpers
│   ├── clickhouse/          # ClickHouse client & schema
│   ├── instrumentation/     # OpenTelemetry helpers
│   ├── llm/                 # LLM integration utilities
│   ├── redis/               # Redis queues & cache
│   ├── repositories/        # Data repositories
│   ├── services/            # Shared services
│   ├── utils/               # Server utilities
│   ├── logger.ts
│   └── queues.ts
├── encryption/              # Encryption utilities
├── features/                # Feature-specific code
├── tableDefinitions/        # Table schemas
├── utils/                   # Shared utilities
├── constants.ts
├── db.ts                    # Prisma client
├── env.ts                   # Environment config
└── index.ts                 # Main exports

// General imports - types, schemas, constants, interfaces
import {
  CloudConfigSchema,
  StringNoHTML,
  AnnotationQueueObjectType,
  type APIScoreV2,
  type ColumnDefinition,
  Role,
} from "@langfuse/shared";

// Database - Prisma client and types
import { prisma, Prisma, JobExecutionStatus } from "@langfuse/shared/src/db";
import { type DB as Database } from "@langfuse/shared";

// Server utilities - queues, services, auth, instrumentation
import {
  logger,
  instrumentAsync,
  traceException,
  redis,
  getTracesTable,
  StorageService,
  sendMembershipInvitationEmail,
  invalidateApiKeysForProject,
  recordIncrement,
  recordHistogram,
} from "@langfuse/shared/src/server";

// API key management (specific path)
import { createAndAddApiKeysToDb } from "@langfuse/shared/src/server/auth/apiKeys";

// Encryption utilities
import { encrypt, decrypt, sign, verify } from "@langfuse/shared/encryption";

// ❌ NEVER: Business logic in procedures
export const traceRouter = createTRPCRouter({
  byId: protectedProjectProcedure
    .input(z.object({ traceId: z.string() }))
    .query(async ({ input, ctx }) => {
      // 200 lines of logic here
    }),
});

// ✅ ALWAYS: Delegate to service
export const traceRouter = createTRPCRouter({
  byId: protectedProjectProcedure
    .input(z.object({ traceId: z.string() }))
    .query(async ({ input, ctx }) => {
      return await getTraceById(input.traceId);
    }),
});

// ❌ NEVER (except in env.mjs itself)
const dbUrl = process.env.DATABASE_URL;

// ✅ ALWAYS
import { env } from "@/src/env.mjs";
const dbUrl = env.DATABASE_URL;

import { z } from "zod/v4";

const schema = z.object({
  email: z.string().email(),
  projectId: z.string(),
});
const validated = schema.parse(input);

// Services use Prisma directly for simple CRUD
import { prisma } from "@langfuse/shared/src/db";

const dataset = await prisma.dataset.findUnique({
  where: { id: datasetId, projectId }, // Always filter by projectId for tenant isolation
});

// Or use repositories for complex queries (traces, observations, scores)
import { getTracesTable } from "@langfuse/shared/src/server";

const traces = await getTracesTable({
  projectId,
  filter: [...],
  limit: 1000,
});

// Import observability utilities
import {
  logger,          // Winston logger with OpenTelemetry/DataDog context
  traceException,  // Record exceptions to OpenTelemetry spans
  instrumentAsync, // Create instrumented spans
} from "@langfuse/shared/src/server";

// Structured logging (includes trace_id, span_id, dd.trace_id)
logger.info("Processing dataset", { datasetId, projectId });
logger.error("Failed to create dataset", { error: err.message });

// Record exceptions to OpenTelemetry (sent to DataDog)
try {
  await operation();
} catch (error) {
  traceException(error); // Records to current span
  throw error;
}

// Instrument critical operations (all API routes auto-instrumented)
const result = await instrumentAsync(
  { name: "dataset.create" },
  async (span) => {
    span.setAttributes({ datasetId, projectId });
    // Operation here
    return dataset;
  },
);

// Integration Test (Public API)
const res = await makeZodVerifiedAPICall(
  PostDatasetsV1Response, "POST", "/api/public/datasets",
  { name: "test-dataset" }, auth
);
expect(res.status).toBe(200);

// tRPC Test
const { caller } = await prepare(); // Creates session + caller
const response = await caller.automations.getAutomations({ projectId });
expect(response).toHaveLength(1);

// Service Test
const result = await getObservationsWithModelDataFromEventsTable({
  projectId, filter: [...], limit: 1000, offset: 0
});
expect(result.length).toBeGreaterThan(0);

// Worker Test (vitest)
const stream = await getObservationStream({ projectId, filter: [] });
const rows = [];
for await (const chunk of stream) rows.push(chunk);
expect(rows).toHaveLength(2);

// ✅ CORRECT: Filter by projectId for tenant isolation
const trace = await prisma.trace.findUnique({
  where: { id: traceId, projectId }, // Required for multi-tenant data isolation
});

// ✅ CORRECT: ClickHouse queries also require projectId
const traces = await queryClickhouse({
  query: `
    SELECT * FROM traces
    WHERE project_id = {projectId: String}
    AND timestamp >= {startTime: DateTime64(3)}
  `,
  params: { projectId, startTime },
});

undefined

---

// tRPC (Web)
import { z } from "zod/v4";
import {
  createTRPCRouter,
  protectedProjectProcedure,
} from "@/src/server/api/trpc";
import { TRPCError } from "@trpc/server";

// Database
import { prisma } from "@langfuse/shared/src/db";
import type { Prisma } from "@prisma/client";

// ClickHouse
import {
  queryClickhouse,
  queryClickhouseStream,
  upsertClickhouse,
} from "@langfuse/shared/src/server";

// Observability - OpenTelemetry + DataDog (NOT Sentry for backend)
import {
  logger,          // Winston logger with OTEL/DataDog trace context
  traceException,  // Record exceptions to OpenTelemetry spans
  instrumentAsync, // Create instrumented spans for operations
} from "@langfuse/shared/src/server";

// Config
import { env } from "@/src/env.mjs"; // web
// or
import { env } from "./env"; // worker

// Public API (Web)
import { withMiddlewares } from "@/src/features/public-api/server/withMiddlewares";
import { createAuthedProjectAPIRoute } from "@/src/features/public-api/server/createAuthedProjectAPIRoute";

// Queue Processing (Worker)
import { Job } from "bullmq";
import { QueueName, TQueueJobTypes } from "@langfuse/shared/src/server";