docs: ai 协作文档体系重构与多 ai 仲裁结果落地
1.AI 协作文档体系重构(objections/worklines/contracts+matrix.md) 2.coord 仲裁文档(final-decisions/cross-review/final-rulings/orchestration) 3.各服务 01/02 文档补全 4.共享包初始化(shared-ts/shared-go/hooks/ui-components/ui-tokens) 5.Proto 契约补全 6.004 架构影响地图更新 7.端口分配表 8.设计规格文档
This commit is contained in:
SpecialX
22
packages/shared-ts/src/outbox/index.ts
Normal file
22
packages/shared-ts/src/outbox/index.ts
Normal file
@@ -0,0 +1,22 @@
|
||||
export { OutboxModule } from "./outbox.module.js";
|
||||
export type { OutboxRootOptions } from "./outbox.module.js";
|
||||
export { OutboxService } from "./outbox.service.js";
|
||||
export { OutboxPublisher } from "./publisher.js";
|
||||
export { createOutboxTable } from "./schema.js";
|
||||
export type { OutboxTable, OutboxRow, NewOutboxRow } from "./schema.js";
|
||||
export {
|
||||
OUTBOX_CONFIG,
|
||||
OUTBOX_DB,
|
||||
OUTBOX_KAFKA_PRODUCER,
|
||||
OUTBOX_LOGGER,
|
||||
OUTBOX_TABLE,
|
||||
} from "./types.js";
|
||||
export type {
|
||||
OutboxConfig,
|
||||
OutboxDbClient,
|
||||
OutboxKafkaProducer,
|
||||
OutboxLogger,
|
||||
OutboxRecord,
|
||||
OutboxStatus,
|
||||
PublishOptions,
|
||||
} from "./types.js";
|
||||
75
packages/shared-ts/src/outbox/outbox.module.ts
Normal file
75
packages/shared-ts/src/outbox/outbox.module.ts
Normal file
@@ -0,0 +1,75 @@
|
||||
import { Module, type DynamicModule } from "@nestjs/common";
|
||||
import pino from "pino";
|
||||
import { OutboxService } from "./outbox.service.js";
|
||||
import { OutboxPublisher } from "./publisher.js";
|
||||
import { createOutboxTable } from "./schema.js";
|
||||
import {
|
||||
OUTBOX_CONFIG,
|
||||
OUTBOX_DB,
|
||||
OUTBOX_KAFKA_PRODUCER,
|
||||
OUTBOX_LOGGER,
|
||||
OUTBOX_TABLE,
|
||||
type OutboxConfig,
|
||||
type OutboxDbClient,
|
||||
type OutboxKafkaProducer,
|
||||
type OutboxLogger,
|
||||
} from "./types.js";
|
||||
|
||||
/**
|
||||
* forRoot 选项。
|
||||
*
|
||||
* `config` 即 OutboxConfig（表名 / topic / 轮询参数）。
|
||||
* `db` / `kafkaProducer` / `logger` 为基础设施实例：NestJS DynamicModule
|
||||
* 无法跨模块边界注入宿主已存在的具体单例，故统一通过 forRoot 注入。
|
||||
*
|
||||
* `kafkaProducer` 应由调用方以 idempotent 方式创建：
|
||||
* `kafka.producer({ idempotent: true, transactionalId: '<service>-tx' })`。
|
||||
*/
|
||||
export interface OutboxRootOptions {
|
||||
config: OutboxConfig;
|
||||
db: OutboxDbClient;
|
||||
kafkaProducer: OutboxKafkaProducer;
|
||||
/** 可选；省略时使用默认 pino logger（name=outbox） */
|
||||
logger?: OutboxLogger;
|
||||
}
|
||||
|
||||
function resolveLogger(logger: OutboxLogger | undefined): OutboxLogger {
|
||||
return logger ?? pino({ name: "outbox", level: "info" });
|
||||
}
|
||||
|
||||
/**
|
||||
* OutboxModule —— 事务性 Outbox 模式的 NestJS DynamicModule。
|
||||
*
|
||||
* 注册并启动：
|
||||
* - OutboxService：业务代码注入后调用 `publish` 写入 outbox 记录
|
||||
* - OutboxPublisher：onModuleInit 启动轮询，onModuleDestroy 停止轮询
|
||||
*
|
||||
* 用法：
|
||||
* ```ts
|
||||
* @Module({
|
||||
* imports: [OutboxModule.forRoot({ config, db, kafkaProducer })],
|
||||
* })
|
||||
* export class AppModule {}
|
||||
* ```
|
||||
*/
|
||||
@Module({})
|
||||
export class OutboxModule {
|
||||
static forRoot(options: OutboxRootOptions): DynamicModule {
|
||||
const table = createOutboxTable(options.config.tableName);
|
||||
const logger = resolveLogger(options.logger);
|
||||
|
||||
return {
|
||||
module: OutboxModule,
|
||||
providers: [
|
||||
{ provide: OUTBOX_CONFIG, useValue: options.config },
|
||||
{ provide: OUTBOX_DB, useValue: options.db },
|
||||
{ provide: OUTBOX_TABLE, useValue: table },
|
||||
{ provide: OUTBOX_KAFKA_PRODUCER, useValue: options.kafkaProducer },
|
||||
{ provide: OUTBOX_LOGGER, useValue: logger },
|
||||
OutboxService,
|
||||
OutboxPublisher,
|
||||
],
|
||||
exports: [OutboxService],
|
||||
};
|
||||
}
|
||||
}
|
||||
74
packages/shared-ts/src/outbox/outbox.service.ts
Normal file
74
packages/shared-ts/src/outbox/outbox.service.ts
Normal file
@@ -0,0 +1,74 @@
|
||||
import { createId } from "@paralleldrive/cuid2";
|
||||
import { Inject, Injectable } from "@nestjs/common";
|
||||
import type { NewOutboxRow, OutboxTable } from "./schema.js";
|
||||
import {
|
||||
OUTBOX_CONFIG,
|
||||
OUTBOX_DB,
|
||||
OUTBOX_LOGGER,
|
||||
OUTBOX_TABLE,
|
||||
type OutboxConfig,
|
||||
type OutboxDbClient,
|
||||
type OutboxLogger,
|
||||
type PublishOptions,
|
||||
} from "./types.js";
|
||||
|
||||
/**
|
||||
* OutboxService —— 事务性 Outbox 写入服务。
|
||||
*
|
||||
* 调用方在业务事务内调用 `publish`，将事件记录写入 outbox 表，
|
||||
* 与业务写在同一事务中原子提交，保证“业务变更”与“事件发布”的一致性。
|
||||
*
|
||||
* 由独立的 OutboxPublisher 负责异步轮询 pending 记录并投递到 Kafka，
|
||||
* 实现 at-least-once 投递语义。
|
||||
*/
|
||||
@Injectable()
|
||||
export class OutboxService {
|
||||
constructor(
|
||||
@Inject(OUTBOX_DB) private readonly db: OutboxDbClient,
|
||||
@Inject(OUTBOX_TABLE) private readonly table: OutboxTable,
|
||||
@Inject(OUTBOX_CONFIG) private readonly config: OutboxConfig,
|
||||
@Inject(OUTBOX_LOGGER) private readonly logger: OutboxLogger,
|
||||
) {}
|
||||
|
||||
/**
|
||||
* 写入一条 outbox 记录。
|
||||
*
|
||||
* @param eventType 事件类型（如 `ExamCreated`），用于 Kafka topic 路由
|
||||
* @param payload 事件负载，将被 JSON 序列化存入 outbox 行
|
||||
* @param options 可选：aggregateId / metadata / delayMs / tx
|
||||
* @returns event id（cuid2），供调用方追踪；同时作为 Kafka 消息 key 实现幂等去重
|
||||
*
|
||||
* 事务管理：传入 `options.tx` 则在调用方事务内写入（推荐，保证与业务表原子提交）；
|
||||
* 省略时使用模块注入的 db 执行单条插入（仅保证 outbox 行自身的原子性）。
|
||||
*/
|
||||
async publish(
|
||||
eventType: string,
|
||||
payload: unknown,
|
||||
options?: PublishOptions,
|
||||
): Promise<string> {
|
||||
const id = createId();
|
||||
const now = new Date();
|
||||
const nextRetryAt =
|
||||
options?.delayMs !== undefined && options.delayMs > 0
|
||||
? new Date(now.getTime() + options.delayMs)
|
||||
: null;
|
||||
|
||||
const record: NewOutboxRow = {
|
||||
id,
|
||||
eventType,
|
||||
payload: JSON.stringify(payload),
|
||||
aggregateId: options?.aggregateId ?? id,
|
||||
maxRetryCount: this.config.maxRetryCount,
|
||||
nextRetryAt,
|
||||
metadata: options?.metadata ?? null,
|
||||
};
|
||||
|
||||
await (options?.tx ?? this.db).insert(this.table).values(record);
|
||||
|
||||
this.logger.debug(
|
||||
{ id, eventType, aggregateId: record.aggregateId },
|
||||
"Outbox record enqueued",
|
||||
);
|
||||
return id;
|
||||
}
|
||||
}
|
||||
205
packages/shared-ts/src/outbox/publisher.ts
Normal file
205
packages/shared-ts/src/outbox/publisher.ts
Normal file
@@ -0,0 +1,205 @@
|
||||
import {
|
||||
Inject,
|
||||
Injectable,
|
||||
type OnModuleDestroy,
|
||||
type OnModuleInit,
|
||||
} from "@nestjs/common";
|
||||
import { and, eq, isNull, lte, or, type SQL } from "drizzle-orm";
|
||||
import type { OutboxRow, OutboxTable } from "./schema.js";
|
||||
import {
|
||||
OUTBOX_CONFIG,
|
||||
OUTBOX_DB,
|
||||
OUTBOX_KAFKA_PRODUCER,
|
||||
OUTBOX_LOGGER,
|
||||
OUTBOX_TABLE,
|
||||
type OutboxConfig,
|
||||
type OutboxDbClient,
|
||||
type OutboxKafkaProducer,
|
||||
type OutboxLogger,
|
||||
} from "./types.js";
|
||||
|
||||
/**
|
||||
* OutboxPublisher —— 轮询 pending 记录并投递到 Kafka。
|
||||
*
|
||||
* 实现 at-least-once 投递语义：
|
||||
* 1. 周期性（pollIntervalMs）批量拉取 status=pending 且到期的记录
|
||||
* 2. 逐条投递到 Kafka（message key = event id，保证同聚合有序）
|
||||
* 3. 投递成功 → 标记 published
|
||||
* 4. 投递失败 → 指数退避更新 nextRetryAt；重试耗尽 → 标记 failed
|
||||
*
|
||||
* 幂等性：
|
||||
* - producer 应由调用方配置为 idempotent（`kafka.producer({ idempotent: true, transactionalId })`），
|
||||
* 避免生产端重试产生重复消息
|
||||
* - message key = event id（cuid2），消费端可基于 event_id 去重（Redis SETNX 或 DB 唯一索引）
|
||||
*/
|
||||
@Injectable()
|
||||
export class OutboxPublisher implements OnModuleInit, OnModuleDestroy {
|
||||
private intervalId: ReturnType<typeof setInterval> | null = null;
|
||||
private isPolling = false;
|
||||
|
||||
constructor(
|
||||
@Inject(OUTBOX_DB) private readonly db: OutboxDbClient,
|
||||
@Inject(OUTBOX_TABLE) private readonly table: OutboxTable,
|
||||
@Inject(OUTBOX_KAFKA_PRODUCER)
|
||||
private readonly producer: OutboxKafkaProducer,
|
||||
@Inject(OUTBOX_CONFIG) private readonly config: OutboxConfig,
|
||||
@Inject(OUTBOX_LOGGER) private readonly logger: OutboxLogger,
|
||||
) {}
|
||||
|
||||
async onModuleInit(): Promise<void> {
|
||||
await this.start();
|
||||
}
|
||||
|
||||
async onModuleDestroy(): Promise<void> {
|
||||
await this.stop();
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
if (this.intervalId) return;
|
||||
this.logger.info(
|
||||
{
|
||||
topic: this.config.kafkaTopic,
|
||||
pollIntervalMs: this.config.pollIntervalMs,
|
||||
batchSize: this.config.batchSize,
|
||||
},
|
||||
"OutboxPublisher started",
|
||||
);
|
||||
this.intervalId = setInterval(() => {
|
||||
void this.poll();
|
||||
}, this.config.pollIntervalMs);
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
if (this.intervalId) {
|
||||
clearInterval(this.intervalId);
|
||||
this.intervalId = null;
|
||||
}
|
||||
this.logger.info("OutboxPublisher stopped");
|
||||
}
|
||||
|
||||
/**
|
||||
* 拉取一批到期 pending 记录并逐条投递。
|
||||
* 通过 isPolling 标志保证不会并发轮询。
|
||||
*/
|
||||
private async poll(): Promise<void> {
|
||||
if (this.isPolling) return;
|
||||
this.isPolling = true;
|
||||
try {
|
||||
const now = new Date();
|
||||
const where: SQL | undefined = and(
|
||||
eq(this.table.status, "pending"),
|
||||
or(isNull(this.table.nextRetryAt), lte(this.table.nextRetryAt, now)),
|
||||
);
|
||||
const messages: OutboxRow[] = await this.db
|
||||
.select()
|
||||
.from(this.table)
|
||||
.where(where)
|
||||
.limit(this.config.batchSize);
|
||||
|
||||
for (const message of messages) {
|
||||
await this.dispatch(message);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error({ error }, "Outbox poll failed");
|
||||
} finally {
|
||||
this.isPolling = false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 投递单条记录到 Kafka 并更新状态。
|
||||
*/
|
||||
private async dispatch(message: OutboxRow): Promise<void> {
|
||||
try {
|
||||
await this.producer.send({
|
||||
topic: this.config.kafkaTopic,
|
||||
messages: [
|
||||
{
|
||||
key: message.id,
|
||||
value: message.payload,
|
||||
headers: this.buildHeaders(message),
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
await this.db
|
||||
.update(this.table)
|
||||
.set({ status: "published", publishedAt: new Date(), lastError: null })
|
||||
.where(eq(this.table.id, message.id));
|
||||
|
||||
this.logger.info(
|
||||
{
|
||||
id: message.id,
|
||||
eventType: message.eventType,
|
||||
topic: this.config.kafkaTopic,
|
||||
},
|
||||
"Outbox message published",
|
||||
);
|
||||
} catch (error) {
|
||||
await this.handleFailure(message, this.toErrorMessage(error));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 失败处理：指数退避更新 nextRetryAt，重试耗尽则标记 failed。
|
||||
*/
|
||||
private async handleFailure(
|
||||
message: OutboxRow,
|
||||
errorMessage: string,
|
||||
): Promise<void> {
|
||||
const newRetryCount = message.retryCount + 1;
|
||||
this.logger.error(
|
||||
{
|
||||
id: message.id,
|
||||
eventType: message.eventType,
|
||||
retryCount: newRetryCount,
|
||||
error: errorMessage,
|
||||
},
|
||||
"Outbox publish failed",
|
||||
);
|
||||
|
||||
if (newRetryCount >= message.maxRetryCount) {
|
||||
await this.db
|
||||
.update(this.table)
|
||||
.set({
|
||||
status: "failed",
|
||||
retryCount: newRetryCount,
|
||||
lastError: errorMessage,
|
||||
})
|
||||
.where(eq(this.table.id, message.id));
|
||||
return;
|
||||
}
|
||||
|
||||
const backoffMs = this.config.retryBackoffMs * 2 ** newRetryCount;
|
||||
const nextRetryAt = new Date(Date.now() + backoffMs);
|
||||
await this.db
|
||||
.update(this.table)
|
||||
.set({
|
||||
retryCount: newRetryCount,
|
||||
nextRetryAt,
|
||||
lastError: errorMessage,
|
||||
})
|
||||
.where(eq(this.table.id, message.id));
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建 Kafka 消息头：eventId / eventType / aggregateId + 调用方 metadata。
|
||||
*/
|
||||
private buildHeaders(message: OutboxRow): Record<string, string> {
|
||||
const headers: Record<string, string> = {
|
||||
eventId: message.id,
|
||||
eventType: message.eventType,
|
||||
aggregateId: message.aggregateId,
|
||||
};
|
||||
if (message.metadata) {
|
||||
for (const [key, value] of Object.entries(message.metadata)) {
|
||||
headers[key] = value;
|
||||
}
|
||||
}
|
||||
return headers;
|
||||
}
|
||||
|
||||
private toErrorMessage(error: unknown): string {
|
||||
return error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
}
|
||||
56
packages/shared-ts/src/outbox/schema.ts
Normal file
56
packages/shared-ts/src/outbox/schema.ts
Normal file
@@ -0,0 +1,56 @@
|
||||
import {
|
||||
int,
|
||||
json,
|
||||
mysqlTable,
|
||||
text,
|
||||
timestamp,
|
||||
varchar,
|
||||
} from "drizzle-orm/mysql-core";
|
||||
import type { OutboxRecord, OutboxStatus } from "./types.js";
|
||||
|
||||
/**
|
||||
* 创建 outbox 表的 Drizzle schema。
|
||||
*
|
||||
* 表名由调用方通过 OutboxConfig.tableName 提供（每服务独立 outbox 表），
|
||||
* 因此以工厂函数形式返回，避免多服务共用同一表名常量。
|
||||
*
|
||||
* 字段与 OutboxRecord 接口一一对齐：
|
||||
* - id：cuid2（varchar(32)），作为 event id 与 Kafka 消息 key
|
||||
* - status：通过 $type<OutboxStatus> 收窄为联合类型
|
||||
* - metadata：通过 $type<Record<string,string> | null> 收窄
|
||||
*/
|
||||
export function createOutboxTable(tableName: string) {
|
||||
return mysqlTable(tableName, {
|
||||
id: varchar("id", { length: 32 }).notNull().primaryKey(),
|
||||
eventType: varchar("event_type", { length: 100 }).notNull(),
|
||||
payload: text("payload").notNull(),
|
||||
aggregateId: varchar("aggregate_id", { length: 32 }).notNull(),
|
||||
status: varchar("status", { length: 20 })
|
||||
.notNull()
|
||||
.default("pending")
|
||||
.$type<OutboxStatus>(),
|
||||
retryCount: int("retry_count").notNull().default(0),
|
||||
maxRetryCount: int("max_retry_count").notNull().default(5),
|
||||
nextRetryAt: timestamp("next_retry_at"),
|
||||
createdAt: timestamp("created_at").notNull().defaultNow(),
|
||||
publishedAt: timestamp("published_at"),
|
||||
lastError: text("last_error"),
|
||||
metadata: json("metadata").$type<Record<string, string> | null>(),
|
||||
});
|
||||
}
|
||||
|
||||
/** outbox 表类型 */
|
||||
export type OutboxTable = ReturnType<typeof createOutboxTable>;
|
||||
|
||||
/** select 推断行类型（应结构兼容 OutboxRecord） */
|
||||
export type OutboxRow = OutboxTable["$inferSelect"];
|
||||
|
||||
/** insert 推断类型 */
|
||||
export type NewOutboxRow = OutboxTable["$inferInsert"];
|
||||
|
||||
/**
|
||||
* 编译期断言：OutboxRow 与 OutboxRecord 结构兼容。
|
||||
* 若 schema 与接口偏离，此处会在严格模式下编译失败。
|
||||
*/
|
||||
const _assertOutboxRow: OutboxRow extends OutboxRecord ? true : never = true;
|
||||
void _assertOutboxRow;
|
||||
87
packages/shared-ts/src/outbox/types.ts
Normal file
87
packages/shared-ts/src/outbox/types.ts
Normal file
@@ -0,0 +1,87 @@
|
||||
import type { MySql2Database } from "drizzle-orm/mysql2";
|
||||
import type { Producer } from "kafkajs";
|
||||
import type { Logger } from "pino";
|
||||
|
||||
/**
|
||||
* Outbox 记录状态机：
|
||||
* - pending：已写入 outbox 表，等待 publisher 投递
|
||||
* - published：已成功投递到 Kafka
|
||||
* - failed：重试次数耗尽，需人工介入
|
||||
*/
|
||||
export type OutboxStatus = "pending" | "published" | "failed";
|
||||
|
||||
/**
|
||||
* Outbox 模块配置（由各服务在 forRoot 时提供）。
|
||||
*/
|
||||
export interface OutboxConfig {
|
||||
/** outbox 表名（每服务独立，如 core_edu_outbox） */
|
||||
tableName: string;
|
||||
/** Kafka 投递目标 topic */
|
||||
kafkaTopic: string;
|
||||
/** 轮询 pending 记录的间隔（毫秒） */
|
||||
pollIntervalMs: number;
|
||||
/** 每次轮询批量处理的记录数 */
|
||||
batchSize: number;
|
||||
/** 单条记录最大重试次数 */
|
||||
maxRetryCount: number;
|
||||
/** 重试退避基数（毫秒），实际退避 = retryBackoffMs * 2^retryCount */
|
||||
retryBackoffMs: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* publish 选项。
|
||||
*
|
||||
* `tx` 用于由调用方管理事务：调用方在 `db.transaction(async (tx) => { ... })`
|
||||
* 中同时写入业务表与 outbox 记录，保证两者原子提交（事务性 Outbox 模式）。
|
||||
*/
|
||||
export interface PublishOptions {
|
||||
/** 聚合根 ID，作为 Kafka 消息 key 用于分区与顺序保证 */
|
||||
aggregateId?: string;
|
||||
/** 附加元数据，写入 outbox 行并随消息头投递 */
|
||||
metadata?: Record<string, string>;
|
||||
/** 延迟投递毫秒数；设置后 nextRetryAt 推迟，到期前不会被轮询 */
|
||||
delayMs?: number;
|
||||
/** 调用方管理的事务客户端；省略时使用模块注入的 db（非事务单条插入） */
|
||||
tx?: OutboxDbClient;
|
||||
}
|
||||
|
||||
/**
|
||||
* Outbox 行结构（与 schema.ts 的 $inferSelect 对齐）。
|
||||
*/
|
||||
export interface OutboxRecord {
|
||||
id: string;
|
||||
eventType: string;
|
||||
/** JSON 序列化后的 payload */
|
||||
payload: string;
|
||||
aggregateId: string;
|
||||
status: OutboxStatus;
|
||||
retryCount: number;
|
||||
maxRetryCount: number;
|
||||
nextRetryAt: Date | null;
|
||||
createdAt: Date;
|
||||
publishedAt: Date | null;
|
||||
lastError: string | null;
|
||||
metadata: Record<string, string> | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Drizzle 数据库客户端类型。
|
||||
*
|
||||
* 服务以无 schema 绑定方式创建 db（`drizzle(pool)`），其类型为
|
||||
* `MySql2Database<Record<string, never>>`。由于 `MySql2Transaction extends
|
||||
* MySqlDatabase`，事务客户端同样可赋值给本类型，故 publish 可接受调用方传入的 tx。
|
||||
*/
|
||||
export type OutboxDbClient = MySql2Database<Record<string, never>>;
|
||||
|
||||
/** Kafka producer 类型别名 */
|
||||
export type OutboxKafkaProducer = Producer;
|
||||
|
||||
/** pino Logger 类型别名 */
|
||||
export type OutboxLogger = Logger;
|
||||
|
||||
// —— NestJS 注入 Token —— //
|
||||
export const OUTBOX_CONFIG = Symbol("OUTBOX_CONFIG");
|
||||
export const OUTBOX_DB = Symbol("OUTBOX_DB");
|
||||
export const OUTBOX_TABLE = Symbol("OUTBOX_TABLE");
|
||||
export const OUTBOX_KAFKA_PRODUCER = Symbol("OUTBOX_KAFKA_PRODUCER");
|
||||
export const OUTBOX_LOGGER = Symbol("OUTBOX_LOGGER");
|
||||
Reference in New Issue
Block a user