janados-microservice-sdk

微服务通用 SDK（@janados/microservice-sdk）：鉴权、Kafka、日志、MongoDB、Redis、MinIO、错误处理与工具函数。

安装

# 首次安装
pnpm add @janados/microservice-sdk
# 更新 SDK
pnpm up @janados/microservice-sdk --latest

@janados/microservice-sdk/auth

基于 auth-service，提供 Fastify 鉴权插件。默认通过 Authorization 请求头传递 token。

{
	authroization: "...Token...";
}

verifyAccessToken 函数

封装远程调用 POST /api/auth/verify-access-token

import { verifyAccessToken } from "@janados/microservice-sdk/auth";

const payload = await verifyAccessToken(accessToken);

authPlugin

注册后装饰 request.user，并可选挂载全局 preHandler。

import Fastify from "fastify";
import { authPlugin } from "@janados/microservice-sdk/auth";
import { setupErrorHandler } from "@janados/microservice-sdk/error";

const fastify = Fastify();

setupErrorHandler(fastify);
authPlugin({ fastify }); // global 默认为 true

鉴权行为

| 场景 | 行为 | | ------------------------------------------------- | ---------------------------------------------------- | | 无 Authorization | 继续处理，request.user = null | | 校验成功 | request.user 为 auth-service 返回的 payload | | 带了 token 但校验失败 | 抛出 MicroserviceError（401，message 为 auth-service 上游原因，如 token.snapshot_expired；缺失时为 unknownReason） | | config.auth 为 true / 'required' 且无 token | 抛出 MicroserviceError（401，auth.unauthorized） |

路由级 config.auth

// 默认可选：有 token 则校验，无 token 也可访问
fastify.get("/posts", async request => {
	if (request.user) {
		// 已登录
	}
	return { items: [] };
});

// 必须登录
fastify.get("/me", { config: { auth: true } }, async request => {
	return request.user;
});

// 跳过鉴权（如 health）
fastify.get("/health", { config: { auth: false } }, async () => ({ ok: true }));

关闭全局钩子、按路由挂载

authPlugin({ fastify, global: false });

fastify.get(
	"/profile",
	{ preHandler: [fastify.authenticate({ required: true })] },
	async request => request.user
);

request.user 或 verifyAccessToken 返回值在 token 未过期时为 token 内的部分用户信息, 在过期但在保质期内时返回完整用户信息 ( 此时会通过完整用户信息刷新 token 通过 NOTIFICATION.APP_PUSH 发送给用户设备, 此时的 request.user 也将是完整用户信息 )

@janados/microservice-sdk/kafka

基于 KafkaJS 的 Fastify 集成，默认导出 kafkaSDK。

import Fastify from "fastify";
import kafkaSDK from "@janados/microservice-sdk/kafka";

const fastify = Fastify({ logger: true });

kafkaSDK.init({
	brokers: process.env.KAFKA_BROKERS.split(","),
	clientId: process.env.SERVICE_NAME,
});

await kafkaSDK.ensureTopicsReady([kafkaSDK.TOPICS.NOTIFICATION_SERVICE.EMAIL]);

await kafkaSDK.producerPlugin({
	fastify,
	serviceName: process.env.SERVICE_NAME,
});

await fastify.ready();

await kafkaSDK.sendMessage({
	topic: kafkaSDK.TOPICS.NOTIFICATION_SERVICE.EMAIL,
	message: { to: "[email protected]", subject: "hi" },
});

await fastify.listen({ port: 3000, host: "0.0.0.0" });

消费端通过 consumerPlugin 订阅 topic；handlers 的 key 必须与 topics 中的名称一一对应，消息体会自动 JSON.parse 后传入 handler。groupId 可选，默认为 `${serviceName}-group`。

import Fastify from "fastify";
import kafkaSDK from "@janados/microservice-sdk/kafka";

const fastify = Fastify({ logger: true });
const serviceName = process.env.SERVICE_NAME;

kafkaSDK.init({
	brokers: process.env.KAFKA_BROKERS.split(","),
	clientId: serviceName,
});

await kafkaSDK.ensureTopicsReady([kafkaSDK.TOPICS.NOTIFICATION_SERVICE.EMAIL]);

await kafkaSDK.consumerPlugin({
	fastify,
	serviceName,
	topics: [kafkaSDK.TOPICS.NOTIFICATION_SERVICE.EMAIL],
	handlers: {
		[kafkaSDK.TOPICS.NOTIFICATION_SERVICE.EMAIL]: async payload => {
			// payload 为 JSON 解析后的消息体
			const { to, subject } = payload;
			fastify.log.info({ to, subject }, "收到邮件通知");
			// 执行业务逻辑，例如发送邮件
		},
	},
	// groupId: `${serviceName}-email-consumer`, // 可选
});

await fastify.ready();
await fastify.listen({ port: 3000, host: "0.0.0.0" });

同一服务可同时注册 producerPlugin 与 consumerPlugin；进程关闭时 consumer 会通过 Fastify onClose 自动断开连接。

@janados/microservice-sdk/logger

将结构化日志发送到 Kafka 的 log.logs topic（需先完成 kafkaSDK 初始化并注册 producerPlugin）。

import { sendLogToKafka } from "@janados/microservice-sdk/logger";

await sendLogToKafka({
	service: "order-service",
	log: {
		orderId: "ord_123",
		message: "order.created",
	},
});

@janados/microservice-sdk/mongodb

使用环境变量连接 MongoDB（MONGO_URI 必填；MONGO_USER、MONGO_PASSWORD 可选）。

import initDatabase from "@janados/microservice-sdk/mongodb";

await initDatabase();
// 之后可使用 mongoose 模型

@janados/microservice-sdk/redis

共享 Redis 客户端单例（lazyConnect: true）。配置来自 REDIS_HOST、REDIS_PORT、REDIS_PASSWORD。

import redis from "@janados/microservice-sdk/redis";

await redis.set("session:abc", JSON.stringify({ userId: "u1" }), "EX", 3600);

const raw = await redis.get("session:abc");

@janados/microservice-sdk/minio

基于 MinIO 的对象存储封装：共享客户端单例与常用 bucket / 对象操作。

客户端

import { minioClient } from "@janados/microservice-sdk/minio";

await minioClient.statObject("media", "avatars/user-1.jpg");

封装方法

import {
	putObject,
	getPresignedUrl,
	objectExists,
	removeObject,
	buildObjectUrl,
} from "@janados/microservice-sdk/minio";

// 上传（data 可为 Buffer 或本地文件路径）
const { url, size } = await putObject({
	bucketName: "media",
	objectKey: "avatars/user-1.jpg",
	data: buffer,
	contentType: "image/jpeg",
});

// 公开 bucket 下的直链（不经过预签名）
const publicUrl = buildObjectUrl("media", "avatars/user-1.jpg");

// 私有 bucket：生成临时访问 URL（默认 7 天有效）
const signedUrl = await getPresignedUrl({
	bucketName: "media",
	objectKey: "private/doc.pdf",
	expiry: 3600,
});

if (await objectExists("media", "avatars/user-1.jpg")) {
	await removeObject("media", "avatars/user-1.jpg");
}

导出说明

| 导出 | 说明 | | ------------------------- | ----------------------------------------------------- | | default / minioClient | MinIO 客户端单例 | | minioConfig | 当前连接配置（只读参考） | | getOrCreateBucket | 创建 bucket 并设置 public / private 策略 | | putObject | 上传对象，返回 { bucketName, objectKey, url, size } | | buildObjectUrl | 拼接公开访问 URL（适用于 public bucket） | | getPresignedUrl | 生成预签名 GET URL | | statObject | 获取对象元信息 | | objectExists | 检查对象是否存在 | | removeObject | 删除对象 |

文件元数据、业务状态（如软删除）请在各微服务内自行维护（例如 MongoDB 模型），SDK 仅负责对象存储本身。

@janados/microservice-sdk/utils

import {
	loadEnv,
	objectToQuery,
	queryToObject,
	randomString,
	remoteInvoke,
} from "@janados/microservice-sdk/utils";

// 生产环境将 .env.production 复制为 .env
loadEnv();

// 对象 → URL query
objectToQuery({ page: 1, active: true }); // "page=1&active=true"

// URL query → 对象（值为字符串；可带或不带前导 `?`）
queryToObject("?page=1&active=true"); // { page: "1", active: "true" }

// 随机字符串
randomString(8, { digits: true, az: true }); // 例如 "a3k9m2x1"

remoteInvoke

微服务间 HTTP 调用的类型安全客户端。根据 REMOTE_SERVICES 中预置的服务与 API 定义，自动生成对应方法；method 由 API 定义固定，调用方无需传入。

import { remoteInvoke } from "@janados/microservice-sdk/utils";

const userClient = remoteInvoke({
	service: remoteInvoke.REMOTE_SERVICES.USER_SERVICE,
});

请求行为

| 行为 | 说明 | | ----------- | ------------------------------------------------------------------------------------------ | | 默认 Header | Content-Type: application/json，可通过 headers 覆盖 | | Body 序列化 | body 为普通对象时自动 JSON.stringify；string、FormData 原样发送 | | 响应解析 | 2xx 且 Content-Type 含 application/json 时返回解析后的 JSON；非 2xx 时 reject 响应体 | | 其余选项 | 除 query 外的字段（如 signal）会透传给 fetch |

参数简写

若传入对象不包含 body、query、headers 任一字段，则视为简写：

• GET：整个对象作为 URL query
• 非 GET（POST 等）：整个参数作为 request body

若对象包含上述字段之一，则按完整配置对象解析。

示例：GET 查询用户

// 简写：{ id } → query
const user = await userClient.getUserByQuery({ id: "u1" });

// 完整配置
const user2 = await userClient.getUserByQuery({
	query: { id: "u1" },
	headers: { "X-Request-Id": "req-abc" },
});

示例：POST 注册 / 登录

// 简写：payload 直接作为 body
await userClient.registerByEmail({
	email: "[email protected]",
	password: "secret",
});

await userClient.loginByEmailPassword({
	email: "[email protected]",
	password: "secret",
});

// 完整配置：显式指定 body、headers 等
await userClient.loginByEmailVerifyCode({
	body: { email: "[email protected]", code: "123456" },
	headers: { Authorization: "Bearer <token>" },
});

错误处理

非 2xx 响应会以 Promise.reject 抛出；响应为 JSON 时 reject JSON 对象，否则 reject 文本。

try {
	await userClient.getUserByQuery({ id: "unknown" });
} catch (err) {
	// err 为服务端返回的错误体（JSON 或文本）
}

当前内置服务见 remoteInvoke.REMOTE_SERVICES（如 USER_SERVICE，默认 http://localhost:8100）。

@janados/microservice-sdk/error

Fastify 全局错误处理；MicroserviceError 可指定 HTTP 状态码，未捕获异常会写入 Kafka 日志。

import Fastify from "fastify";
import {
	MicroserviceError,
	setupErrorHandler,
} from "@janados/microservice-sdk/error";

const fastify = Fastify();

setupErrorHandler(fastify);

fastify.get("/users/:id", async request => {
	const { id } = request.params;

	if (!id) {
		throw new MicroserviceError("user.idRequired", 400);
	}

	return { id };
});

setupErrorHandler 依赖 process.env.SERVICE_NAME 与已就绪的 Kafka producer（用于 sendLogToKafka）。