From 3c2ea50c7f1b7b4eb002e353a1f1d6584bd1845c Mon Sep 17 00:00:00 2001 From: SpecialX <47072643+wangxiner55@users.noreply.github.com> Date: Tue, 14 Jul 2026 23:45:32 +0800 Subject: [PATCH] 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 --- .../connectors/edu-outbox-connector.json | 37 +++ infra/debezium/register-outbox-connectors.sh | 98 ++++++ infra/docker-compose.yml | 8 +- package.json | 1 + scripts/proto-to-graphql/generator.ts | 300 ++++++++++++++++++ 5 files changed, 441 insertions(+), 3 deletions(-) create mode 100644 infra/debezium/connectors/edu-outbox-connector.json create mode 100644 infra/debezium/register-outbox-connectors.sh create mode 100644 scripts/proto-to-graphql/generator.ts diff --git a/infra/debezium/connectors/edu-outbox-connector.json b/infra/debezium/connectors/edu-outbox-connector.json new file mode 100644 index 0000000..1d0e5bd --- /dev/null +++ b/infra/debezium/connectors/edu-outbox-connector.json @@ -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" + } +} diff --git a/infra/debezium/register-outbox-connectors.sh b/infra/debezium/register-outbox-connectors.sh new file mode 100644 index 0000000..1e907a2 --- /dev/null +++ b/infra/debezium/register-outbox-connectors.sh @@ -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//status" +echo "" +echo "删除 connector：" +echo " curl -X DELETE ${DEBEZIUM_URL}/connectors/" diff --git a/infra/docker-compose.yml b/infra/docker-compose.yml index fd3fa05..a0c196b 100644 --- a/infra/docker-compose.yml +++ b/infra/docker-compose.yml @@ -136,14 +136,16 @@ services: - "16686:16686" - "4318:4318" # ============================================================ - # Debezium Connect - CDC 链路核心 + # Debezium Connect - CDC 链路核心（v2.1：Outbox 表监听 + 自动投递） # 监听 MySQL binlog → 写入 Kafka topic - # topic 命名约定：..（如 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: diff --git a/package.json b/package.json index ae7cb56..79536bb 100644 --- a/package.json +++ b/package.json @@ -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": { diff --git a/scripts/proto-to-graphql/generator.ts b/scripts/proto-to-graphql/generator.ts new file mode 100644 index 0000000..a7cfbf0 --- /dev/null +++ b/scripts/proto-to-graphql/generator.ts @@ -0,0 +1,300 @@ +/** + * proto → GraphQL SDL 生成器（v2.1 M0） + * + * 功能： + * - 解析 packages/shared-proto/proto/*.proto 文件 + * - 根据 service / message 自动生成 GraphQL Federation 子图 SDL + * - 支持 @key / @requires / @extends 指令 + * - 输出到 services//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", + int32: "Int", + int64: "String", // GraphQL 无 64 位整数，用 String 表示 + uint32: "Int", + uint64: "String", + bool: "Boolean", + float: "Float", + double: "Float", + bytes: "String", +}; + +// proto → 服务名映射（用于确定输出目录） +const SERVICE_MAP: Record = { + 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 判定：含 _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();