126 lines
3.6 KiB
TypeScript
126 lines
3.6 KiB
TypeScript
import { logger } from "../observability/logger.js";
|
||
import { getProducer } from "../kafka/kafka.client.js";
|
||
import { resolveTopic } from "../kafka/topic-map.js";
|
||
import {
|
||
findPending,
|
||
incrementRetry,
|
||
markFailed,
|
||
markPublished,
|
||
} from "./outbox.repository.js";
|
||
import type { OutboxEvent } from "./outbox.schema.js";
|
||
|
||
/**
|
||
* OutboxPublisher —— 轮询 pending 记录并投递到 Kafka（多 topic 路由）。
|
||
*
|
||
* 参照 core-edu OutboxPublisher 模式，支持 TOPIC_MAP 多 topic 路由：
|
||
* - notification.sent → edu.notification.sent
|
||
* - notification.read → edu.notification.read
|
||
* - notification.recalled → edu.notification.recalled
|
||
* - notification.failed → edu.notification.failed
|
||
*
|
||
* 仲裁依据：at-least-once 投递 + 指数退避重试 + 幂等（消费端 event_id 去重）。
|
||
*/
|
||
|
||
const POLL_INTERVAL_MS = 5000;
|
||
const BATCH_SIZE = 100;
|
||
const MAX_RETRY = 5;
|
||
const RETRY_BACKOFF_MS = 2000;
|
||
|
||
class OutboxPublisher {
|
||
private intervalId: ReturnType<typeof setInterval> | null = null;
|
||
private isPolling = false;
|
||
|
||
async start(): Promise<void> {
|
||
if (this.intervalId) return;
|
||
logger.info(
|
||
{ pollIntervalMs: POLL_INTERVAL_MS, batchSize: BATCH_SIZE },
|
||
"OutboxPublisher started",
|
||
);
|
||
this.intervalId = setInterval(() => {
|
||
void this.poll();
|
||
}, POLL_INTERVAL_MS);
|
||
}
|
||
|
||
async stop(): Promise<void> {
|
||
if (this.intervalId) {
|
||
clearInterval(this.intervalId);
|
||
this.intervalId = null;
|
||
}
|
||
logger.info("OutboxPublisher stopped");
|
||
}
|
||
|
||
private async poll(): Promise<void> {
|
||
if (this.isPolling) return;
|
||
this.isPolling = true;
|
||
try {
|
||
const messages = await findPending(BATCH_SIZE);
|
||
for (const message of messages) {
|
||
await this.dispatch(message);
|
||
}
|
||
} catch (error) {
|
||
logger.error({ error }, "Outbox poll failed");
|
||
} finally {
|
||
this.isPolling = false;
|
||
}
|
||
}
|
||
|
||
private async dispatch(message: OutboxEvent): Promise<void> {
|
||
const topic = resolveTopic(message.eventType);
|
||
try {
|
||
const producer = getProducer();
|
||
await producer.send({
|
||
topic,
|
||
messages: [
|
||
{
|
||
key: message.aggregateId,
|
||
value:
|
||
typeof message.payload === "string"
|
||
? message.payload
|
||
: JSON.stringify(message.payload),
|
||
headers: this.buildHeaders(message),
|
||
},
|
||
],
|
||
});
|
||
await markPublished(message.eventId);
|
||
logger.info(
|
||
{ eventId: message.eventId, eventType: message.eventType, topic },
|
||
"Outbox message published",
|
||
);
|
||
} catch (error) {
|
||
const errorMessage =
|
||
error instanceof Error ? error.message : String(error);
|
||
logger.error(
|
||
{
|
||
eventId: message.eventId,
|
||
eventType: message.eventType,
|
||
error: errorMessage,
|
||
},
|
||
"Outbox publish failed",
|
||
);
|
||
if (message.retryCount + 1 >= MAX_RETRY) {
|
||
await markFailed(message.eventId, errorMessage);
|
||
} else {
|
||
const backoffMs = RETRY_BACKOFF_MS * 2 ** message.retryCount;
|
||
await incrementRetry(message.eventId, errorMessage, backoffMs);
|
||
}
|
||
}
|
||
}
|
||
|
||
private buildHeaders(message: OutboxEvent): Record<string, string> {
|
||
const headers: Record<string, string> = {
|
||
eventId: message.eventId,
|
||
eventType: message.eventType,
|
||
aggregateType: message.aggregateType,
|
||
aggregateId: message.aggregateId,
|
||
};
|
||
if (message.metadata) {
|
||
for (const [key, value] of Object.entries(message.metadata)) {
|
||
headers[key] = value;
|
||
}
|
||
}
|
||
return headers;
|
||
}
|
||
}
|
||
|
||
export const outboxPublisher = new OutboxPublisher();
|