backend-dev-guidelines

后端开发指南

目的

通过采用 Next.js 14、tRPC、BullMQ 和 TypeScript 模式，在 Langfuse 的后端包（web、worker、packages/shared）中保持一致性并遵循最佳实践。

何时使用此技能

在以下情况下自动启用：

• 创建或修改 tRPC 路由器和过程
• 创建或修改公共 API 端点（REST）
• 创建或修改 BullMQ 队列消费者和生产者
• 使用业务逻辑构建服务
• 验证 API 请求
• 根据授权访问资源
• 实现中间件（tRPC、NextAuth、公共 API）
• 使用 Prisma（PostgreSQL）或 ClickHouse 执行数据库操作
• 使用 OpenTelemetry、DataDog、logger 和 traceException 实现可观测性
• 使用 Zod v4 验证输入
• 从环境变量配置环境
• 进行后端测试和重构

---

快速入门

UI：新建 tRPC 功能检查清单（Web）

• 路由器：在 features/[feature]/server/*Router.ts 中定义
• 过程：使用合适的过程类型（受保护、公开）
• 身份验证：通过中间件使用 JWT 授权。
• 授权检查：根据资源和角色访问资源
• 验证：为输入定义 Zod v4 模式
• 服务：在服务文件中实现业务逻辑
• 错误处理：使用 traceException 包装器
• 测试：在 __tests__/ 中添加单元测试和集成测试
• 配置：通过 env.mjs 访问

SDK：新建公共 API 端点检查清单（Web）

• 路由文件：在 pages/api/public/ 中创建
• 包装器：使用 withMiddlewares + createAuthedProjectAPIRoute
• 类型：在 features/public-api/types/ 中定义
• 身份验证：通过基本身份验证进行授权
• 验证：为查询参数、请求体和响应定义 Zod 模式
• 版本控制：在 API 路径和用于查询参数、请求体及响应的 Zod 模式中进行版本控制
• 测试：在 __tests__/async/ 中添加端到端测试

新建队列处理器检查清单（Worker）

• 处理器：在 worker/src/queues/ 中创建
• 队列类型：在 packages/shared/src/server/queues 中创建队列类型
• 服务：在 features/ 或 worker/src/features/ 中实现业务逻辑
• 错误处理：区分应导致队列处理失败的错误，以及应产生成功事件的错误。
• 队列注册：在 app.ts 中添加到 WorkerManager
• 测试：在 worker 中添加 vitest 测试

---

架构概览

分层架构

# Web Package (Next.js 14)

┌─ tRPC API ──────────────────┐   ┌── Public REST API ──────────┐
│                             │   │                             │
│  HTTP Request               │   │  HTTP Request               │
│      ↓                      │   │      ↓                      │
│  tRPC Procedure             │   │  withMiddlewares +          │
│  (protectedProjectProcedure)│   │  createAuthedProjectAPIRoute│
│      ↓                      │   │      ↓                      │
│  Service (business logic)   │   │  Service (business logic)   │
│      ↓                      │   │      ↓                      │
│  Prisma / ClickHouse        │   │  Prisma / ClickHouse        │
│                             │   │                             │
└─────────────────────────────┘   └─────────────────────────────┘
                 ↓
            [optional]: Publish to Redis BullMQ queue
                 ↓
┌─ Worker Package (Express) ──────────────────────────────────┐
│                                                             │
│  BullMQ Queue Job                                           │
│      ↓                                                      │
│  Queue Processor (handles job)                              │
│      ↓                                                      │
│  Service (business logic)                                   │
│      ↓                                                      │
│  Prisma / ClickHouse                                        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

核心原则：

• Web：用于 UI 的 tRPC 过程或用于 SDK 的公共 API 路由 → 服务 → 数据库
• Worker：队列处理器 → 服务 → 数据库
• packages/shared：Web 和 Worker 共享的代码

完整详情请参阅 architecture-overview.md。

---

目录结构

Web 包（/web/）

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 包（/worker/）

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

共享包（/packages/shared/）

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

导入路径（package.json 导出）：

共享包为不同使用场景提供了特定的导入路径：

导入路径	映射到	用途
@langfuse/shared	dist/src/index.js	通用类型、架构、工具和常量
@langfuse/shared/src/db	dist/src/db.js	Prisma 客户端和数据库类型
@langfuse/shared/src/server	dist/src/server/index.js	服务端工具（队列、身份验证、服务、埋点）
@langfuse/shared/src/server/auth/apiKeys	dist/src/server/auth/apiKeys.js	API 密钥管理工具
@langfuse/shared/encryption	dist/src/encryption/index.js	加密和签名工具

使用示例：

// 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";

内容归属：

共享包提供供 Web 包和 worker 包使用的类型、工具和服务端代码。它有 5 个导出路径，用于控制前端与后端的访问权限：

导入路径	用途	包含内容
@langfuse/shared	✅ 前端 + 后端	Prisma 类型、Zod 架构、常量、表定义、领域模型、工具
@langfuse/shared/src/db	🔒 仅限后端	Prisma 客户端实例
@langfuse/shared/src/server	🔒 仅限后端	服务、存储库、队列、身份验证、ClickHouse、LLM 集成、埋点
@langfuse/shared/src/server/auth/apiKeys	🔒 仅限后端	API 密钥管理（分离以避免循环依赖）
@langfuse/shared/encryption	🔒 仅限后端	数据库字段加密/解密

命名约定：

• tRPC 路由器：camelCaseRouter.ts - datasetRouter.ts
• 服务：功能目录中的 service.ts
• 队列处理器：camelCaseQueue.ts - evalQueue.ts
• 公共 API：kebab-case.ts - dataset-items.ts

---

核心原则

1. tRPC 过程委托给服务

// ❌ 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);
    }),
});

2. 通过 env.mjs 访问配置，绝不要使用 process.env

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

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

3. 使用 Zod v4 验证所有输入

import { z } from "zod/v4";

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

4. 简单的 CRUD 由服务直接使用 Prisma，复杂查询则使用存储库

// 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,
});

6. 可观测性：OpenTelemetry + DataDog（后端不使用 Sentry）

Langfuse 使用 OpenTelemetry 实现后端可观测性，并将跟踪和日志发送到 DataDog。

// 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;
  },
);

注意：前端使用 Sentry，但后端（tRPC、API 路由、服务、worker）使用 OpenTelemetry + DataDog。

7. 必须进行全面测试

为所有新功能和错误修复编写测试。详细示例请参阅 testing-guide.md。

测试类型：

类型	框架	位置	用途
集成	Jest	web/src/__tests__/async/	完整的 API 端点测试
tRPC	Jest	web/src/__tests__/async/	带身份验证的 tRPC 过程
服务	Jest	web/src/__tests__/async/repositories/	仓库/服务函数
Worker	Vitest	worker/src/__tests__/	队列处理器和流

快速示例：

// 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);

关键原则：

• 使用唯一 ID（randomUUID()），避免测试相互干扰
• 清理测试数据或使用唯一的项目 ID
• 测试必须彼此独立，并可按任意顺序运行
• 测试中绝不可使用 pruneDatabase

8. 始终按 projectId 过滤，以隔离租户

// ✅ 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 },
});

---

常用导入

// 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";

---

快速参考

HTTP 状态码

代码	使用场景
200	成功
201	已创建
400	错误请求
401	未授权
403	禁止访问
404	未找到
500	服务器错误

可供参考的功能示例

参考 Langfuse 中已有功能的实现模式：

• 数据集（web/src/features/datasets/）- 完整功能，包含 tRPC 路由器、公共 API 和服务
• 提示词（web/src/features/prompts/）- 包含版本控制和模板的功能
• 评估（web/src/features/evals/）- 与 Worker 集成的复杂功能
• 公共 API（web/src/features/public-api/）- 中间件和路由模式

---

应避免的反模式

❌ 在路由/过程中编写业务逻辑
❌ 直接使用 process.env（始终使用 env.mjs/env.ts）
❌ 缺少错误处理
❌ 未进行输入验证（始终使用 Zod v4）
❌ 面向租户的查询缺少 projectId 过滤条件
❌ 使用 console.log，而不是 logger/traceException（OpenTelemetry）

---

导航指南

需要……	阅读此文档
了解架构	architecture-overview.md
创建路由/控制器	routing-and-controllers.md
组织业务逻辑	services-and-repositories.md
创建中间件	middleware-guide.md
访问数据库	database-patterns.md
管理配置	configuration.md
编写测试	testing-guide.md

---

资源文件

architecture-overview.md

三层架构（tRPC/公共 API → 服务 → 数据访问）、tRPC/公共 API/Worker 的请求生命周期、Next.js 14 目录结构、双数据库系统（PostgreSQL + ClickHouse）、关注点分离、复杂查询的仓库模式

routing-and-controllers.md

Next.js 基于文件的路由、tRPC 路由器模式、公共 REST API 路由、分层架构（入口 → 服务 → 仓库 → 数据库）、服务层组织方式、应避免的反模式

services-and-repositories.md

服务层概览、依赖注入模式、单例模式、数据访问的仓库模式、服务设计原则、缓存策略、服务测试

middleware-guide.md

tRPC 中间件（withErrorHandling、withOtelInstrumentation、enforceUserIsAuthed）、七种 tRPC 过程类型（publicProcedure、authenticatedProcedure、protectedProjectProcedure 等）、公共 API 中间件（withMiddlewares、createAuthedProjectAPIRoute）、身份验证模式（用于 tRPC 的 NextAuth、用于公共 API 的 Basic Auth）

database-patterns.md

双数据库架构（通过 Prisma 使用 PostgreSQL + 通过直接客户端使用 ClickHouse）、PostgreSQL CRUD 操作、ClickHouse 查询模式（queryClickhouse、queryClickhouseStream、upsertClickhouse）、复杂查询的仓储模式、通过 projectId 筛选实现租户隔离、数据库选择指南

configuration.md

使用 Zod 验证环境变量、特定于各软件包的配置（web/env.mjs 使用 t3-oss/env-nextjs、worker/env.ts、shared/env.ts）、NEXT_PUBLIC_LANGFUSE_CLOUD_REGION 的用法、用于企业功能的 LANGFUSE_EE_LICENSE_KEY、环境变量管理最佳实践

testing-guide.md

集成测试（公共 API 使用 makeZodVerifiedAPICall）、tRPC 测试（createInnerTRPCContext、appRouter.createCaller）、服务级测试（仓储/服务函数）、worker 测试（使用流的 vitest）、测试隔离原则、运行测试（web 使用 Jest，worker 使用 vitest）

---

相关技能

• database-verification - 验证列名和架构一致性
• skill-developer - 用于创建和管理技能的元技能

---

技能状态：已完成 ✅ 行数：约 540 行 渐进式披露：7 个资源文件 ✅