Files
arasystem/backend/test/fake-event-stream.ts
T
ttipo f725908053 feat(backend): 실시간 알림 GET /api/mail-events (엔진 EventSource 중계)
- 직원마다 엔진 EventSource 하나를 그 직원의 연결들이 나눠 쓴다(fan-out) — state · 받은편지함 안 읽음 · 새 메일 계산은 직원마다 한 번
- 새 메일: Email/changes → Email/get 으로 받은편지함 · 초안 아님 · 안 읽음 · 본인 아님 · 10분 넘게 앞서지 않은 메일만,
  셀 수 없거나(cannotCalculateChanges · 400 notRequest) 2,000건 넘게 바뀌면 state 만 보내고 기준을 다시 잡는다
- 연결 5개 한도, 25초 ping(upstream), 토큰 만료 30초 전 end(남은 시간이 짧으면 끝날 때), 서버 종료 때 end,
  처리 전에 끊긴 응답 정리, 처음 엔진 연결 10초 상한
- 엔진 연결: 75초 무소식이면 다시 잇기(±20%), 엔진 429 는 끊지 않고 다시 묻기, 같은 기준 3번 실패면 기준 다시 잡기, 5분 안전판
- 액세스 토큰의 만료 시각을 AuthenticatedUser 에 싣는다(선택 필드), 요청이 끝난 뒤에도 이어지는 일은 요청 컨텍스트 밖에서

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-28 13:32:08 +09:00

71 lines
2.3 KiB
TypeScript

/**
* 메일 엔진의 변화 소식 스트림(JMAP EventSource) 대역 — 단위 테스트 공용.
* fetch 본문처럼 바이트를 흘려보내고, 읽는 쪽이 닫으면(abort) 읽던 자리에 오류를 낸다.
* 테스트가 `send` · `end` 로 엔진 쪽을 흉내 낸다. 타이머를 쓰지 않는다 (promise 로만 돈다 — 가짜 타이머와 함께 써도 된다).
*/
const encoder = new TextEncoder();
export class FakeEventStream {
/** 읽는 쪽이 닫았다 */
aborted = false;
private readonly queue: (Uint8Array | Error | null)[] = [];
private wake: (() => void) | null = null;
constructor(signal: AbortSignal) {
const abort = () => {
this.aborted = true;
this.push(new Error('aborted'));
};
if (signal.aborted) abort();
else signal.addEventListener('abort', abort, { once: true });
}
/** 엔진이 글자를 보낸다 */
send(text: string): void {
this.push(encoder.encode(text));
}
/** 엔진이 스트림을 닫는다 */
end(): void {
this.push(null);
}
async *body(): AsyncGenerator<Uint8Array> {
for (;;) {
while (this.queue.length === 0) {
await new Promise<void>((resolve) => {
this.wake = resolve;
});
}
const item = this.queue.shift();
if (item === undefined || item === null) return;
if (item instanceof Error) throw item;
yield item;
}
}
private push(item: Uint8Array | Error | null): void {
this.queue.push(item);
this.wake?.();
this.wake = null;
}
}
/** 이 계정의 메일 · 메일함 변화 신호 — Stalwart 0.16.27 의 StateChange 모양 (2026-10-28 확인) */
export function stateChangeEvent(accountId: string): string {
return `event: state\ndata: ${JSON.stringify({
'@type': 'StateChange',
changed: { [accountId]: { Email: 'x', Mailbox: 'y' } },
})}\n\n`;
}
/** 엔진의 연결 유지 신호 */
export const PING_EVENT = 'event: ping\ndata: {"interval": 30}\n\n';
/** 미뤄 둔 비동기 처리(스트림 읽기 · 엔진 묻기의 결과)를 마저 돌린다 — 타이머는 움직이지 않는다 (setImmediate 는 가짜로 두지 않는다) */
export async function settleAsync(): Promise<void> {
for (let round = 0; round < 10; round += 1) {
await new Promise((resolve) => setImmediate(resolve));
}
}