Codex
70 lines
2.4 KiB
TypeScript
70 lines
2.4 KiB
TypeScript
import type { ResultSetHeader, RowDataPacket } from 'mysql2/promise';
|
|
import type { MySqlPool } from '../db/mysql.js';
|
|
|
|
export interface AppendOutboxEventInput {
|
|
tenantId: string;
|
|
aggregateType: string;
|
|
aggregateId: string;
|
|
eventType: string;
|
|
idempotencyKey: string;
|
|
payload: unknown;
|
|
availableAt?: Date;
|
|
}
|
|
|
|
type SqlExecutor = Pick<MySqlPool, 'execute'>;
|
|
|
|
export class OutboxRepository {
|
|
constructor(private readonly executor: SqlExecutor) {}
|
|
|
|
async append(input: AppendOutboxEventInput): Promise<{ id: string; created: boolean }> {
|
|
const [result] = await this.executor.execute<ResultSetHeader>(
|
|
`INSERT IGNORE INTO qipai_outbox_events
|
|
(tenant_id, aggregate_type, aggregate_id, event_type,
|
|
idempotency_key, payload, available_at)
|
|
VALUES (?, ?, ?, ?, ?, CAST(? AS JSON), ?)`,
|
|
[
|
|
input.tenantId,
|
|
input.aggregateType,
|
|
input.aggregateId,
|
|
input.eventType,
|
|
input.idempotencyKey,
|
|
JSON.stringify(input.payload),
|
|
input.availableAt ?? new Date()
|
|
]
|
|
);
|
|
if (result.insertId > 0) {
|
|
return { id: String(result.insertId), created: true };
|
|
}
|
|
const [rows] = await this.executor.execute<Array<RowDataPacket & { id: string }>>(
|
|
`SELECT id FROM qipai_outbox_events
|
|
WHERE tenant_id = ? AND idempotency_key = ?`,
|
|
[input.tenantId, input.idempotencyKey]
|
|
);
|
|
if (!rows[0]) throw new Error('Idempotent outbox lookup failed after duplicate insert.');
|
|
return { id: String(rows[0].id), created: false };
|
|
}
|
|
|
|
async markPublished(eventId: string): Promise<boolean> {
|
|
const [result] = await this.executor.execute<ResultSetHeader>(
|
|
`UPDATE qipai_outbox_events
|
|
SET status = 'PUBLISHED', published_at = UTC_TIMESTAMP(3),
|
|
attempts = attempts + 1, last_error = NULL
|
|
WHERE id = ? AND status = 'PENDING'`,
|
|
[eventId]
|
|
);
|
|
return result.affectedRows === 1;
|
|
}
|
|
|
|
async markFailed(eventId: string, error: unknown, delayMs: number): Promise<boolean> {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
const [result] = await this.executor.execute<ResultSetHeader>(
|
|
`UPDATE qipai_outbox_events
|
|
SET available_at = DATE_ADD(UTC_TIMESTAMP(3), INTERVAL ? MICROSECOND),
|
|
attempts = attempts + 1, last_error = ?
|
|
WHERE id = ? AND status = 'PENDING'`,
|
|
[delayMs * 1000, message.slice(0, 1000), eventId]
|
|
);
|
|
return result.affectedRows === 1;
|
|
}
|
|
}
|