R524809
95 lines
3.7 KiB
TypeScript
95 lines
3.7 KiB
TypeScript
import type { Queue } from 'bullmq';
|
|
import type { Database } from '@atlas/database';
|
|
|
|
export type OutboxJob = {
|
|
eventId: string;
|
|
workspaceId: string;
|
|
eventType: string;
|
|
payload: Record<string, unknown>;
|
|
};
|
|
|
|
type EventRow = Omit<OutboxJob, 'eventId' | 'workspaceId' | 'eventType'> & {
|
|
id: string;
|
|
workspace_id: string;
|
|
event_type: string;
|
|
event_payload: OutboxJob['payload'];
|
|
};
|
|
|
|
export class OutboxDispatcher {
|
|
public constructor(
|
|
private readonly database: Database,
|
|
private readonly queue: Queue<OutboxJob>,
|
|
private readonly workerId: string,
|
|
) {}
|
|
|
|
public async dispatchBatch() {
|
|
// 业务表受 RLS 保护；先枚举工作空间，再在各自租户事务中认领事件。
|
|
const workspaces = await this.database.query<{ id: string }>(
|
|
'SELECT id FROM atlas.workspaces',
|
|
);
|
|
for (const workspace of workspaces.rows) {
|
|
const events = await this.claim(workspace.id);
|
|
for (const event of events) await this.dispatch(event);
|
|
}
|
|
}
|
|
|
|
private async claim(workspaceId: string): Promise<OutboxJob[]> {
|
|
return this.database.workspaceTransaction(workspaceId, async (tx) => {
|
|
const events = await tx.query<EventRow>(
|
|
`WITH candidates AS (
|
|
SELECT id FROM atlas.outbox_events
|
|
WHERE (status = 'pending' AND next_attempt_at <= now())
|
|
OR (status = 'dispatching' AND locked_until < now())
|
|
ORDER BY created_at
|
|
FOR UPDATE SKIP LOCKED
|
|
LIMIT 20
|
|
)
|
|
UPDATE atlas.outbox_events event
|
|
SET status = 'dispatching', locked_by = $1,
|
|
locked_until = now() + interval '2 minutes', updated_at = now()
|
|
FROM candidates
|
|
WHERE event.id = candidates.id
|
|
RETURNING event.id, event.workspace_id, event.event_type, event.event_payload`,
|
|
[this.workerId],
|
|
);
|
|
return events.rows.map((event) => ({
|
|
eventId: event.id,
|
|
workspaceId: event.workspace_id,
|
|
eventType: event.event_type,
|
|
payload: event.event_payload,
|
|
}));
|
|
});
|
|
}
|
|
|
|
private async dispatch(event: OutboxJob) {
|
|
try {
|
|
await this.queue.add(event.eventType, event, {
|
|
jobId: event.eventId,
|
|
attempts: 5,
|
|
backoff: { type: 'exponential', delay: 5_000 },
|
|
removeOnComplete: 1_000,
|
|
removeOnFail: 1_000,
|
|
});
|
|
await this.database.workspaceTransaction(event.workspaceId, (tx) =>
|
|
tx.query(
|
|
`UPDATE atlas.outbox_events SET status = 'dispatched', dispatched_at = now(),
|
|
locked_by = NULL, locked_until = NULL, updated_at = now()
|
|
WHERE id = $1 AND status = 'dispatching' AND locked_by = $2`,
|
|
[event.eventId, this.workerId],
|
|
),
|
|
);
|
|
} catch (error) {
|
|
await this.database.workspaceTransaction(event.workspaceId, (tx) =>
|
|
tx.query(
|
|
`UPDATE atlas.outbox_events SET status = 'pending', attempt_count = attempt_count + 1,
|
|
next_attempt_at = now() + interval '30 seconds', locked_by = NULL, locked_until = NULL,
|
|
last_error_code = 'QUEUE_UNAVAILABLE', last_error_message = '任务队列暂不可用', updated_at = now()
|
|
WHERE id = $1 AND status = 'dispatching' AND locked_by = $2`,
|
|
[event.eventId, this.workerId],
|
|
),
|
|
);
|
|
throw error;
|
|
}
|
|
}
|
|
}
|