KB Server

基于 Express 的快速 Node.js 服务框架，提供简洁的 API 开发体验、标准错误码、SSE 支持、统一鉴权和完善的日志系统。

特性

• 🚀 快速创建 REST API
• 🎯 基于类的参数校验（使用 class-validator）
• 🔒 统一鉴权函数（API 级别）
• 📡 原生支持 SSE (Server-Sent Events)
• 📋 标准错误码系统
• 🔌 支持自定义 Express 中间件
• 📊 内置日志系统（基于 pino）
• 🌍 统一的响应格式

安装

npm install kb-server

要求 Node.js >= 18.0.0

快速开始

基础用法

import { createServer } from "kb-server";
import * as apis from "./apis";

// 写法一：直接创建并启动
const server = createServer({ apis });
server.listen(3000);

// 写法二：异步初始化后启动
(async () => {
  // 其他异步操作，例如：初始化数据库
  return createServer({ apis });
})().then((app) => app.listen(3000));

创建 API

使用 implementAPI 创建带有类型安全的 API：

import { implementAPI } from "kb-server";
import { IsString, IsNotEmpty } from "class-validator";

// 定义请求参数类
class CreateUserParams {
  @IsString()
  @IsNotEmpty()
  name: string;

  @IsString()
  @IsNotEmpty()
  email: string;
}

// 定义 API 函数类型
type CreateUserFn = (params: CreateUserParams) => Promise<{ id: string }>;

// 实现 API
export const createUser = implementAPI<CreateUserFn>(
  {} as CreateUserFn,
  CreateUserParams,
  async (params, ctx) => {
    // ctx.RequestId - 请求 ID
    // ctx.AuthInfo - 鉴权信息（如果配置了 authFn）
    console.log(`Request ID: ${ctx.RequestId}`);

    // 业务逻辑
    const user = await database.createUser(params);

    return { id: user.id };
  }
);

// 导出所有 API
export default {
  CreateUser: createUser,
};

配置服务器

import { createServer } from "kb-server";
import cors from "cors";
import * as apis from "./apis";
import * as sse from "./sse";

const server = createServer(
  {
    apis,                    // API 列表
    authFn,                  // 鉴权函数（可选）
    log: true,               // 是否开启请求日志
    middlewares: [cors()],   // 自定义中间件列表
    sse: {                   // SSE 配置（可选）
      handlers: sse,         // SSE 处理函数
      route: "/sse",        // SSE 路由（默认: /sse）
      timeout: 30000,        // 超时时间（毫秒，默认: 30000）
    },
  },
  {
    limit: "10mb",           // 请求体大小限制（默认: 10mb）
  }
);

server.listen(3000);

SSE 支持

创建 SSE API 实现实时数据推送：

import { implementSseAPI } from "kb-server";
import { IsString } from "class-validator";

// 定义请求参数
class StreamChatParams {
  @IsString()
  prompt: string;
}

// 定义 SSE 函数类型
type StreamChatFn = (params: StreamChatParams) => Promise<void>;

// 实现 SSE API
export const streamChat = implementSseAPI<StreamChatFn>(
  {} as StreamChatFn,
  StreamChatParams,
  async (params, sse, ctx) => {
    // sse.push(data) - 推送数据到客户端
    // sse.close() - 主动关闭连接
    // sse.abortController - 中止控制器

    try {
      // 模拟流式推送
      for (let i = 0; i < 10; i++) {
        sse.push({
          chunk: `消息片段 ${i + 1}`,
          progress: (i + 1) * 10,
        });

        // 模拟延迟
        await new Promise(resolve => setTimeout(resolve, 500));
      }

      // 完成后关闭连接
      sse.close();
    } catch (error) {
      // 可以使用 abortController 中止操作
      if (!sse.abortController.signal.aborted) {
        throw error;
      }
    }
  }
);

// 导出 SSE API
export default {
  StreamChat: streamChat,
};

统一鉴权

在 API 级别实现统一的鉴权逻辑：

import { createServer } from "kb-server";

const authFn = async (action: string, req: express.Request) => {
  // 从请求中获取 token
  const token = req.headers.authorization?.replace('Bearer ', '');

  if (!token) {
    return false; // 返回 false 表示无权限
  }

  // 验证 token
  const userInfo = await verifyToken(token);
  if (!userInfo) {
    return false;
  }

  // 返回对象会将信息注入到 ctx.AuthInfo
  return {
    userId: userInfo.id,
    role: userInfo.role,
  };
};

const server = createServer({
  apis,
  authFn,
});

// 在 API 中使用鉴权信息
const myAPI = implementAPI<SomeFn>(
  {} as SomeFn,
  ParamsClass,
  async (params, ctx) => {
    const { userId, role } = ctx.AuthInfo || {};
    console.log(`当前用户: ${userId}, 角色: ${role}`);

    // 业务逻辑...
  }
);

自定义错误码

使用 createErrors 创建业务特定的错误码：

import { createErrors } from "kb-server";

// 创建自定义错误码
const CustomErrors = createErrors({
  // 一级错误码（可选）
  // InvalidParameter: {
  //   // 扩展默认错误码
  // },
  BusinessError: {
    // 新增业务错误码
    InsufficientBalance: "余额不足",
    OrderExpired: "订单已过期",
    ProductOutOfStock: "商品库存不足",
  },
});

// 使用自定义错误
const someAPI = implementAPI<SomeFn>(
  {} as SomeFn,
  ParamsClass,
  async (params, ctx) => {
    const balance = await getBalance(ctx.AuthInfo?.userId);
    if (balance < 100) {
      throw new CustomErrors.BusinessError.InsufficientBalance();
    }

    // 业务逻辑...
  }
);

请求与响应格式

请求格式

所有 POST 请求体应包含以下结构：

{
  "Action": "API名称",
  "其他参数": "..."
}

成功响应

{
  "Response": {
    "Data": {
      // 返回的数据
    },
    "RequestId": "uuid-v4"
  }
}

错误响应

{
  "Response": {
    "Error": {
      "Code": "InvalidParameter.ValidationError",
      "Message": "参数校验失败"
    },
    "RequestId": "uuid-v4"
  }
}

内置错误码

| 一级错误码 | 二级错误码 | 说明 | |-----------|-----------|------| | InvalidParameter | EmptyParameter | 请求参数不能为空 | | InvalidParameter | EmptyAPIRequest | 未指定 API 的请求 | | InvalidParameter | ValidationError | 参数校验失败 | | InvalidParameter | RouteError | 路由错误 | | ResourceNotFound | APINotFound | 不存在的 API | | ResourceNotFound | AuthFunctionNotFound | 鉴权函数不存在 | | FailOperation | NoPermission | 无权操作 | | FailOperation | NotLogin | 未登录 | | InternalError | UnknownError | 未知错误 | | InternalError | ServiceError | 内部服务错误 | | InternalError | DatabaseError | DB 异常 |

日志系统

框架内置基于 pino 的日志系统，支持结构化日志：

import { logger } from "kb-server";

// 使用日志
logger.info({ userId: 123, action: "login" }, "用户登录");
logger.warn({ ip: "192.168.1.1" }, "异常访问");
logger.error({ error: err }, "系统错误");

环境变量：

• LOG_LEVEL - 日志级别（默认: info）
• NODE_ENV - 环境模式（非 production 时使用 pino-pretty 格式化输出）

完整示例

import { createServer, implementAPI, createErrors } from "kb-server";
import { IsString, IsEmail } from "class-validator";
import cors from "cors";

// 自定义错误码
const AppErrors = createErrors({
  UserError: {
    DuplicateEmail: "邮箱已被注册",
  },
});

// 参数类
class RegisterParams {
  @IsString()
  @IsNotEmpty()
  username: string;

  @IsString()
  @IsEmail()
  email: string;

  @IsString()
  @IsNotEmpty()
  password: string;
}

// 定义 API 类型
type RegisterFn = (params: RegisterParams) => Promise<{ id: string }>;

// 实现 API
export const register = implementAPI<RegisterFn>(
  {} as RegisterFn,
  RegisterParams,
  async (params, ctx) => {
    // 检查邮箱是否已注册
    const exists = await checkEmailExists(params.email);
    if (exists) {
      throw new AppErrors.UserError.DuplicateEmail();
    }

    // 创建用户
    const user = await createUser(params);

    return { id: user.id };
  }
);

// 导出 API
const apis = {
  User_Register: register,
};

// 鉴权函数
const authFn = async (action, req) => {
  // 公开接口不需要鉴权
  if (action.startsWith("User_")) {
    return true;
  }

  // 其他接口需要鉴权
  const token = req.headers.authorization?.replace('Bearer ', '');
  return await verifyToken(token);
};

// 创建服务器
const server = createServer({
  apis,
  authFn,
  log: true,
  middlewares: [cors()],
});

server.listen(3000, () => {
  console.log('Server is running on http://localhost:3000');
});

API 参考

createServer

创建 Express 服务器实例。

function createServer(
  params: ICreateServerParams,
  options?: ICreateServerOptions
): express.Application

参数:

• params.apis - API 列表
• params.authFn - 鉴权函数（可选）
• params.log - 是否开启日志（默认: true）
• params.middlewares - 自定义中间件列表（可选）
• params.sse - SSE 配置（可选）
• options.limit - 请求体大小限制（默认: "10mb"）

implementAPI

实现带有类型检查的 API。

function implementAPI<F, A = Record<string, any>>(
  actionStub: F,
  ParamsClass: Class<StubParam<F>>,
  execution: APIExecution<StubParam<F>, ReturnType<F>, A>
): API<StubParam<F>, ReturnType<F>>

implementSseAPI

实现 SSE API。

function implementSseAPI<F, A = Record<string, any>>(
  actionStub: F,
  ParamsClass: Class<StubParam<F>>,
  execution: SseExecution<StubParam<F>, ReturnType<F>, A>
): API<StubParam<F>, ReturnType<F>>

License

ISC