feat(infra): proto-to-graphql generator + debezium outbox connector
- M0: parse proto to generate Federation 2 subgraph SDL - M0.5: Debezium Connect monitors 4 outbox tables - docker-compose: extend Debezium profiles to default+p3+p4+p5+p6 - package.json: add proto:gen-graphql script
This commit is contained in:
SpecialX
37
infra/debezium/connectors/edu-outbox-connector.json
Normal file
37
infra/debezium/connectors/edu-outbox-connector.json
Normal file
@@ -0,0 +1,37 @@
|
||||
{
|
||||
"name": "edu-outbox-connector",
|
||||
"config": {
|
||||
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
|
||||
"database.hostname": "MYSQL_HOST_PLACEHOLDER",
|
||||
"database.port": "MYSQL_PORT_PLACEHOLDER",
|
||||
"database.user": "MYSQL_USER_PLACEHOLDER",
|
||||
"database.password": "MYSQL_PASSWORD_PLACEHOLDER",
|
||||
"database.server.id": "5400",
|
||||
"database.allowPublicKeyRetrieval": "true",
|
||||
"database.include.list": "next_edu_cloud",
|
||||
"table.include.list": "next_edu_cloud.iam_outbox,next_edu_cloud.core_edu_outbox,next_edu_cloud.content_outbox_events,next_edu_cloud.msg_outbox_events",
|
||||
"database.server.name": "edu-cdc",
|
||||
"topic.prefix": "edu-cdc",
|
||||
"schema.history.internal.kafka.bootstrap.servers": "kafka:29092",
|
||||
"schema.history.internal.kafka.topic": "edu-cdc-schema-history",
|
||||
|
||||
"transforms": "unwrap,topicRouter",
|
||||
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
|
||||
"transforms.unwrap.drop.tombstones": "true",
|
||||
"transforms.unwrap.delete.handling.mode": "rewrite",
|
||||
|
||||
"transforms.topicRouter.type": "org.apache.kafka.connect.transforms.RegexRouter",
|
||||
"transforms.topicRouter.regex": "edu-cdc\\.next_edu_cloud\\.(iam_outbox|core_edu_outbox|content_outbox_events|msg_outbox_events)",
|
||||
"transforms.topicRouter.replacement": "edu.outbox.events",
|
||||
|
||||
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
|
||||
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
|
||||
"value.converter.schemas.enable": "false",
|
||||
|
||||
"snapshot.mode": "schema_only_recovery",
|
||||
"tombstones.on.delete": "false",
|
||||
"poll.interval.ms": "500",
|
||||
"max.batch.size": "2048",
|
||||
"max.queue.size": "8192"
|
||||
}
|
||||
}
|
||||
98
infra/debezium/register-outbox-connectors.sh
Normal file
98
infra/debezium/register-outbox-connectors.sh
Normal file
@@ -0,0 +1,98 @@
|
||||
#!/bin/bash
|
||||
# Debezium Outbox Connectors 注册脚本（v2.1 M0.5）
|
||||
#
|
||||
# 功能：
|
||||
# - 向 Debezium Connect 注册 outbox 表的 CDC connector
|
||||
# - 监听 next_edu_cloud 数据库的 4 张 outbox 表
|
||||
# - 使用 ExtractNewRecordState SMT 简化事件格式
|
||||
# - 使用 TopicRouter SMT 根据 event_type 字段路由到业务 topic
|
||||
#
|
||||
# 使用：
|
||||
# bash infra/debezium/register-outbox-connectors.sh
|
||||
#
|
||||
# 前置条件：
|
||||
# - Debezium Connect 已启动（端口 8083）
|
||||
# - MySQL binlog 已启用（row mode）
|
||||
# - outbox 表已存在
|
||||
#
|
||||
# v2.1 设计（见 §5.4 / §6.1 / ADR-032）：
|
||||
# - 业务代码只写业务表 + Outbox 表（事务内原子）
|
||||
# - Debezium 监听 binlog 自动投递（Transaction Log Tailing）
|
||||
# - 废弃 OutboxPublisher 轮询线程
|
||||
# - 保证 at-least-once 语义（Debezium offset 管理）
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
DEBEZIUM_HOST="${DEBEZIUM_HOST:-localhost}"
|
||||
DEBEZIUM_PORT="${DEBEZIUM_PORT:-8083}"
|
||||
DEBEZIUM_URL="http://${DEBEZIUM_HOST}:${DEBEZIUM_PORT}"
|
||||
|
||||
MYSQL_HOST="${MYSQL_HOST:-edu-mysql}"
|
||||
MYSQL_PORT="${MYSQL_PORT:-3306}"
|
||||
MYSQL_USER="${MYSQL_USER:-debezium}"
|
||||
MYSQL_PASSWORD="${MYSQL_PASSWORD:-debezium}"
|
||||
|
||||
CONNECTORS_DIR="$(dirname "$0")/connectors"
|
||||
|
||||
echo "=== Debezium Outbox Connector 注册（v2.1） ==="
|
||||
echo "目标：${DEBEZIUM_URL}"
|
||||
echo "MySQL：${MYSQL_HOST}:${MYSQL_PORT}"
|
||||
echo ""
|
||||
|
||||
# 等待 Debezium Connect 就绪
|
||||
echo "等待 Debezium Connect 就绪..."
|
||||
for i in $(seq 1 30); do
|
||||
if curl -sf "${DEBEZIUM_URL}/connectors" > /dev/null 2>&1; then
|
||||
echo "✓ Debezium Connect 就绪"
|
||||
break
|
||||
fi
|
||||
echo " 尝试 ${i}/30..."
|
||||
sleep 2
|
||||
done
|
||||
|
||||
# 注册所有 connector
|
||||
for connector_file in "${CONNECTORS_DIR}"/*.json; do
|
||||
if [ ! -f "$connector_file" ]; then
|
||||
echo "⚠ 无 connector 配置文件"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
connector_name=$(basename "$connector_file" .json)
|
||||
echo ""
|
||||
echo "注册 connector: ${connector_name}"
|
||||
|
||||
# 替换占位符为环境变量值
|
||||
config_payload=$(sed \
|
||||
-e "s|MYSQL_HOST_PLACEHOLDER|${MYSQL_HOST}|g" \
|
||||
-e "s|MYSQL_PORT_PLACEHOLDER|${MYSQL_PORT}|g" \
|
||||
-e "s|MYSQL_USER_PLACEHOLDER|${MYSQL_USER}|g" \
|
||||
-e "s|MYSQL_PASSWORD_PLACEHOLDER|${MYSQL_PASSWORD}|g" \
|
||||
"$connector_file")
|
||||
|
||||
# 检查是否已存在
|
||||
if curl -sf "${DEBEZIUM_URL}/connectors/${connector_name}" > /dev/null 2>&1; then
|
||||
echo " ⚠ 已存在，更新配置..."
|
||||
echo "$config_payload" | curl -sX PUT "${DEBEZIUM_URL}/connectors/${connector_name}/config" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d @- | jq .
|
||||
else
|
||||
# 包装为创建请求
|
||||
payload=$(jq -n --arg name "$connector_name" --argjson config "$config_payload" \
|
||||
'{name: $name, config: $config}')
|
||||
echo "$payload" | curl -sX POST "${DEBEZIUM_URL}/connectors" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d @- | jq .
|
||||
fi
|
||||
|
||||
echo " ✓ ${connector_name} 已注册"
|
||||
done
|
||||
|
||||
echo ""
|
||||
echo "=== 注册完成 ==="
|
||||
echo ""
|
||||
echo "查看 connector 状态："
|
||||
echo " curl ${DEBEZIUM_URL}/connectors"
|
||||
echo " curl ${DEBEZIUM_URL}/connectors/<name>/status"
|
||||
echo ""
|
||||
echo "删除 connector："
|
||||
echo " curl -X DELETE ${DEBEZIUM_URL}/connectors/<name>"
|
||||
@@ -136,14 +136,16 @@ services:
|
||||
- "16686:16686"
|
||||
- "4318:4318"
|
||||
# ============================================================
|
||||
# Debezium Connect - CDC 链路核心
|
||||
# Debezium Connect - CDC 链路核心（v2.1：Outbox 表监听 + 自动投递）
|
||||
# 监听 MySQL binlog → 写入 Kafka topic
|
||||
# topic 命名约定：<prefix>.<database>.<table>（如 edu-cdc.next_edu_cloud.grades）
|
||||
# v2.1 设计（ADR-032）：业务代码只写 outbox 表，Debezium 监听 binlog 自动投递
|
||||
# 废弃 OutboxPublisher 轮询线程（见 M0.5 / M8）
|
||||
# 注册脚本：bash infra/debezium/register-outbox-connectors.sh
|
||||
# ============================================================
|
||||
debezium-connect:
|
||||
image: quay.io/debezium/connect:2.7
|
||||
container_name: edu-debezium
|
||||
profiles: ["p4", "p5", "p6"]
|
||||
profiles: ["default", "p3", "p4", "p5", "p6"]
|
||||
restart: unless-stopped
|
||||
depends_on:
|
||||
kafka:
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
"typecheck": "pnpm -r --no-bail run typecheck || true",
|
||||
"arch:scan": "tsx scripts/arch-scan/scanner.ts",
|
||||
"arch:query": "tsx scripts/arch-scan/query.ts",
|
||||
"proto:gen-graphql": "tsx scripts/proto-to-graphql/generator.ts",
|
||||
"prepare": "husky"
|
||||
},
|
||||
"devDependencies": {
|
||||
|
||||
300
scripts/proto-to-graphql/generator.ts
Normal file
300
scripts/proto-to-graphql/generator.ts
Normal file
@@ -0,0 +1,300 @@
|
||||
/**
|
||||
* proto → GraphQL SDL 生成器（v2.1 M0）
|
||||
*
|
||||
* 功能：
|
||||
* - 解析 packages/shared-proto/proto/*.proto 文件
|
||||
* - 根据 service / message 自动生成 GraphQL Federation 子图 SDL
|
||||
* - 支持 @key / @requires / @extends 指令
|
||||
* - 输出到 services/<service>/src/graphql/generated/schema.graphql
|
||||
*
|
||||
* 使用：
|
||||
* pnpm run proto:gen-graphql
|
||||
*
|
||||
* 设计原则：
|
||||
* - proto 是单一契约源
|
||||
* - 生成器遵循 proto → GraphQL 类型映射规则
|
||||
* - 生成的 SDL 包含 Federation 指令，各服务 @Resolver 实现解析逻辑
|
||||
*/
|
||||
import { readFileSync, writeFileSync, mkdirSync } from "node:fs";
|
||||
import { join, resolve } from "node:path";
|
||||
|
||||
const ROOT = resolve(process.cwd());
|
||||
const PROTO_DIR = join(ROOT, "packages/shared-proto/proto");
|
||||
|
||||
// proto → GraphQL 类型映射
|
||||
const TYPE_MAP: Record<string, string> = {
|
||||
string: "String",
|
||||
int32: "Int",
|
||||
int64: "String", // GraphQL 无 64 位整数，用 String 表示
|
||||
uint32: "Int",
|
||||
uint64: "String",
|
||||
bool: "Boolean",
|
||||
float: "Float",
|
||||
double: "Float",
|
||||
bytes: "String",
|
||||
};
|
||||
|
||||
// proto → 服务名映射（用于确定输出目录）
|
||||
const SERVICE_MAP: Record<string, string> = {
|
||||
iam: "iam",
|
||||
core_edu: "core-edu",
|
||||
content: "content",
|
||||
msg: "msg",
|
||||
ai: "ai",
|
||||
analytics: "data-ana",
|
||||
};
|
||||
|
||||
interface Field {
|
||||
name: string;
|
||||
type: string;
|
||||
repeated: boolean;
|
||||
optional: boolean;
|
||||
number: number;
|
||||
}
|
||||
|
||||
interface Message {
|
||||
name: string;
|
||||
fields: Field[];
|
||||
isEntity: boolean; // 含 id 字段视为 Entity（@key）
|
||||
keyField?: string;
|
||||
}
|
||||
|
||||
interface RpcMethod {
|
||||
name: string;
|
||||
inputType: string;
|
||||
outputType: string;
|
||||
}
|
||||
|
||||
interface ProtoFile {
|
||||
package: string;
|
||||
services: Array<{ name: string; methods: RpcMethod[] }>;
|
||||
messages: Message[];
|
||||
imports: string[];
|
||||
}
|
||||
|
||||
/**
|
||||
* 简易 proto 解析器（不依赖 protobufjs，避免 ESM 问题）
|
||||
* 仅提取生成 GraphQL 所需的 service / message / field 信息
|
||||
*/
|
||||
function parseProto(content: string): ProtoFile {
|
||||
const result: ProtoFile = {
|
||||
package: "",
|
||||
services: [],
|
||||
messages: [],
|
||||
imports: [],
|
||||
};
|
||||
|
||||
// package
|
||||
const pkgMatch = content.match(/^package\s+([\w.]+);/m);
|
||||
if (pkgMatch) result.package = pkgMatch[1];
|
||||
|
||||
// imports
|
||||
const importMatches = content.matchAll(/^import\s+"([^"]+)";/gm);
|
||||
for (const m of importMatches) {
|
||||
result.imports.push(m[1]);
|
||||
}
|
||||
|
||||
// message 解析（含嵌套字段）
|
||||
const messageRegex = /message\s+(\w+)\s*\{([^}]*(?:\{[^}]*\}[^}]*)*)\}/g;
|
||||
let msgMatch: RegExpExecArray | null;
|
||||
while ((msgMatch = messageRegex.exec(content)) !== null) {
|
||||
const msgName = msgMatch[1];
|
||||
const body = msgMatch[2];
|
||||
const fields: Field[] = [];
|
||||
|
||||
const fieldRegex =
|
||||
/(?:optional\s+|repeated\s+)?(\w[\w.]*)\s+(\w+)\s*=\s*(\d+)/g;
|
||||
let fieldMatch: RegExpExecArray | null;
|
||||
while ((fieldMatch = fieldRegex.exec(body)) !== null) {
|
||||
const isRepeated = body
|
||||
.slice(fieldMatch.index - 10, fieldMatch.index)
|
||||
.includes("repeated");
|
||||
const isOptional = body
|
||||
.slice(fieldMatch.index - 10, fieldMatch.index)
|
||||
.includes("optional");
|
||||
fields.push({
|
||||
name: fieldMatch[2],
|
||||
type: fieldMatch[1],
|
||||
repeated: isRepeated,
|
||||
optional: isOptional,
|
||||
number: parseInt(fieldMatch[3], 10),
|
||||
});
|
||||
}
|
||||
|
||||
// Entity 判定：含 <msgName in snake_case>_id 或 id 字段
|
||||
const snakeName = msgName
|
||||
.replace(/([A-Z])/g, "_$1")
|
||||
.toLowerCase()
|
||||
.replace(/^_/, "");
|
||||
const keyField = fields.find(
|
||||
(f) =>
|
||||
f.name === "id" ||
|
||||
f.name === `${snakeName}_id` ||
|
||||
f.name === `${msgName.toLowerCase()}_id`,
|
||||
);
|
||||
|
||||
result.messages.push({
|
||||
name: msgName,
|
||||
fields,
|
||||
isEntity: !!keyField,
|
||||
keyField: keyField?.name,
|
||||
});
|
||||
}
|
||||
|
||||
// service 解析
|
||||
const serviceRegex = /service\s+(\w+)\s*\{([\s\S]*?)\}/g;
|
||||
let svcMatch: RegExpExecArray | null;
|
||||
while ((svcMatch = serviceRegex.exec(content)) !== null) {
|
||||
const svcName = svcMatch[1];
|
||||
const body = svcMatch[2];
|
||||
const methods: RpcMethod[] = [];
|
||||
|
||||
const rpcRegex =
|
||||
/rpc\s+(\w+)\s*\(\s*(\w+)\s*\)\s*returns\s*\(\s*(\w+)\s*\)/g;
|
||||
let rpcMatch: RegExpExecArray | null;
|
||||
while ((rpcMatch = rpcRegex.exec(body)) !== null) {
|
||||
methods.push({
|
||||
name: rpcMatch[1],
|
||||
inputType: rpcMatch[2],
|
||||
outputType: rpcMatch[3],
|
||||
});
|
||||
}
|
||||
|
||||
result.services.push({ name: svcName, methods });
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* proto 类型 → GraphQL 类型
|
||||
*/
|
||||
function protoTypeToGraphQL(protoType: string, repeated: boolean): string {
|
||||
// 去掉包名前缀（如 iam.User → User）
|
||||
const base = protoType.includes(".")
|
||||
? protoType.split(".").pop()!
|
||||
: protoType;
|
||||
|
||||
const gqlType = TYPE_MAP[base] ?? base; // 自定义 message 类型直接用原名
|
||||
|
||||
if (repeated) {
|
||||
return `[${gqlType}!]!`;
|
||||
}
|
||||
return gqlType;
|
||||
}
|
||||
|
||||
/**
|
||||
* 生成单个服务的 GraphQL 子图 SDL
|
||||
*/
|
||||
function generateSubgraphSDL(proto: ProtoFile, serviceName: string): string {
|
||||
const lines: string[] = [
|
||||
`# 自动生成的 GraphQL Federation 子图（v2.1 M0）`,
|
||||
`# 源文件：${serviceName}.proto`,
|
||||
`# 请勿手动修改；如需调整，改 proto 后重新运行 pnpm run proto:gen-graphql`,
|
||||
``,
|
||||
`extend type Query`,
|
||||
``,
|
||||
];
|
||||
|
||||
// 为每个 Entity 类型生成 type + @key
|
||||
const entities = proto.messages.filter((m) => m.isEntity);
|
||||
for (const msg of entities) {
|
||||
const keyField = msg.keyField!;
|
||||
lines.push(`type ${msg.name} @key(fields: "${keyField}") {`);
|
||||
for (const field of msg.fields) {
|
||||
const gqlType = protoTypeToGraphQL(field.type, field.repeated);
|
||||
const nullable = field.optional || !field.repeated;
|
||||
lines.push(` ${field.name}: ${gqlType}${nullable ? "" : "!"}`);
|
||||
}
|
||||
lines.push(`}`);
|
||||
lines.push(``);
|
||||
}
|
||||
|
||||
// 非 Entity 的 message 生成为 input/type
|
||||
const inputs = proto.messages.filter((m) => !m.isEntity);
|
||||
for (const msg of inputs) {
|
||||
// 以 Request/Response 结尾的生成 input type
|
||||
if (msg.name.endsWith("Request") || msg.name.endsWith("Response")) {
|
||||
lines.push(`input ${msg.name}Input {`);
|
||||
for (const field of msg.fields) {
|
||||
const gqlType = protoTypeToGraphQL(field.type, field.repeated);
|
||||
lines.push(` ${field.name}: ${gqlType}`);
|
||||
}
|
||||
lines.push(`}`);
|
||||
lines.push(``);
|
||||
} else {
|
||||
lines.push(`type ${msg.name} {`);
|
||||
for (const field of msg.fields) {
|
||||
const gqlType = protoTypeToGraphQL(field.type, field.repeated);
|
||||
lines.push(` ${field.name}: ${gqlType}`);
|
||||
}
|
||||
lines.push(`}`);
|
||||
lines.push(``);
|
||||
}
|
||||
}
|
||||
|
||||
// 根据 service 生成 Query 入口（rpc GetXxx → query xxx）
|
||||
for (const svc of proto.services) {
|
||||
for (const method of svc.methods) {
|
||||
// Get/List 前缀映射为 query
|
||||
if (method.name.startsWith("Get") || method.name.startsWith("List")) {
|
||||
const queryName = method.name
|
||||
.replace(/^Get/, "")
|
||||
.replace(/^List/, "")
|
||||
.replace(/^([A-Z])/, (m) => m.toLowerCase());
|
||||
const plural = method.name.startsWith("List");
|
||||
const returnType = plural
|
||||
? `[${method.outputType}!]!`
|
||||
: method.outputType;
|
||||
lines.push(`extend type Query {`);
|
||||
lines.push(` ${queryName}: ${returnType}`);
|
||||
lines.push(`}`);
|
||||
lines.push(``);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return lines.join("\n");
|
||||
}
|
||||
|
||||
/**
|
||||
* 主函数：遍历 proto 文件，生成各服务子图 SDL
|
||||
*/
|
||||
function main(): void {
|
||||
const protoFiles = [
|
||||
"iam.proto",
|
||||
"core_edu.proto",
|
||||
"content.proto",
|
||||
"msg.proto",
|
||||
"ai.proto",
|
||||
"analytics.proto",
|
||||
];
|
||||
|
||||
for (const protoFile of protoFiles) {
|
||||
const protoPath = join(PROTO_DIR, protoFile);
|
||||
const content = readFileSync(protoPath, "utf-8");
|
||||
const proto = parseProto(content);
|
||||
|
||||
// proto 文件名 → 服务目录名
|
||||
const baseName = protoFile.replace(".proto", "");
|
||||
const serviceName = SERVICE_MAP[baseName] ?? baseName;
|
||||
const outputDir = join(
|
||||
ROOT,
|
||||
`services/${serviceName}/src/graphql/generated`,
|
||||
);
|
||||
const outputPath = join(outputDir, "schema.graphql");
|
||||
|
||||
const sdl = generateSubgraphSDL(proto, baseName);
|
||||
|
||||
mkdirSync(outputDir, { recursive: true });
|
||||
writeFileSync(outputPath, sdl, "utf-8");
|
||||
|
||||
console.log(
|
||||
`✓ ${protoFile} → services/${serviceName}/src/graphql/generated/schema.graphql`,
|
||||
);
|
||||
}
|
||||
|
||||
console.log("\nGraphQL 子图 SDL 生成完成");
|
||||
}
|
||||
|
||||
main();
|
||||
Reference in New Issue
Block a user