generated from ttipo/project-base
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>
This commit is contained in:
@@ -3,7 +3,7 @@ import { JwtService } from '@nestjs/jwt';
|
||||
import { Test, TestingModule } from '@nestjs/testing';
|
||||
import { createHash } from 'crypto';
|
||||
import { existsSync, unlinkSync } from 'fs';
|
||||
import type { Server } from 'http';
|
||||
import { get as httpGet, type IncomingMessage, type Server } from 'http';
|
||||
import { basename, join } from 'path';
|
||||
import { Readable } from 'stream';
|
||||
import request from 'supertest';
|
||||
@@ -34,6 +34,8 @@ import type {
|
||||
MailboxCreate,
|
||||
MailboxDeletion,
|
||||
MailVacation,
|
||||
MailWatch,
|
||||
MailWatchListener,
|
||||
MessageDeletionResult,
|
||||
MessageListQuery,
|
||||
MessageSource,
|
||||
@@ -141,6 +143,9 @@ const TAKEN_MAILBOX_NAME = '이미 있는 이름';
|
||||
/** 대역 메일 엔진의 비우기가 테스트가 문을 열 때까지 끝나지 않는 메일함 — 같은 메일함의 처리 중 잠금 확인용 */
|
||||
const SLOW_MAILBOX = 'slowbox';
|
||||
|
||||
/** 대역 메일 엔진이 실시간 알림의 처음 연결을 받지 않는 주소 (503 MAIL_001 경로) */
|
||||
const EVENTS_ENGINE_DOWN = 'events.down@e2e.arasystem.test';
|
||||
|
||||
/** 한 번도 설정하지 않은 계정의 부재중 자동 응답 (계약 2절) */
|
||||
const NEVER_CONFIGURED_VACATION: MailVacation = {
|
||||
isEnabled: false,
|
||||
@@ -196,6 +201,53 @@ function listenOnAnyPort(server: Server): Promise<void> {
|
||||
return new Promise((resolve) => server.listen(0, () => resolve()));
|
||||
}
|
||||
|
||||
/** 열린 실시간 알림 연결 — 응답 머리와 받은 글자를 모은다 */
|
||||
interface EventStreamClient {
|
||||
readonly response: IncomingMessage;
|
||||
/** 지금까지 받은 글자 */
|
||||
received(): string;
|
||||
/** 클라이언트가 끊는다 */
|
||||
close(): void;
|
||||
}
|
||||
|
||||
/**
|
||||
* 실시간 알림(SSE)을 연다 — 닫을 때까지 끝나지 않는 응답이라 supertest(본문을 다 받을 때까지 기다린다) 대신 http.get 으로 읽는다.
|
||||
* 연결을 모아 두지 않는다(agent: false) — 끊으면 바로 서버에서도 끊긴다
|
||||
*/
|
||||
function openEventStream(
|
||||
server: Server,
|
||||
cookie: string,
|
||||
): Promise<EventStreamClient> {
|
||||
const address = server.address();
|
||||
const port = typeof address === 'object' && address ? address.port : 0;
|
||||
return new Promise((resolve, reject) => {
|
||||
const request = httpGet(
|
||||
{
|
||||
host: '127.0.0.1',
|
||||
port,
|
||||
path: '/api/mail-events',
|
||||
headers: { cookie, accept: 'text/event-stream' },
|
||||
agent: false,
|
||||
},
|
||||
(response) => {
|
||||
let text = '';
|
||||
response.setEncoding('utf8');
|
||||
response.on('data', (chunk: string) => {
|
||||
text += chunk;
|
||||
});
|
||||
// 끊은 뒤의 오류(ECONNRESET 등)는 테스트와 상관없다
|
||||
response.on('error', () => undefined);
|
||||
resolve({
|
||||
response,
|
||||
received: () => text,
|
||||
close: () => request.destroy(),
|
||||
});
|
||||
},
|
||||
);
|
||||
request.on('error', reject);
|
||||
});
|
||||
}
|
||||
|
||||
/** 서버를 닫는다 — 열린 연결이 모두 끝나야 끝난다 */
|
||||
function closeServer(server: Server): Promise<void> {
|
||||
return new Promise((resolve, reject) =>
|
||||
@@ -269,6 +321,13 @@ describe('응답 계약 (e2e)', () => {
|
||||
update: MessageUpdate;
|
||||
}[] = [];
|
||||
|
||||
/** 5단계 — 대역 엔진에 열린 실시간 알림 연결 (직원마다 하나여야 한다) */
|
||||
const liveWatches: {
|
||||
user: string;
|
||||
listener: MailWatchListener;
|
||||
closed: boolean;
|
||||
}[] = [];
|
||||
|
||||
/** 이 앱의 서명 키로 토큰을 만든다 (CI 처럼 JWT_SECRET 이 없으면 임시 키) */
|
||||
const sign = (payload: object, expiresIn: number) =>
|
||||
jwt.sign(payload, {
|
||||
@@ -385,6 +444,28 @@ describe('응답 계약 (e2e)', () => {
|
||||
Promise.resolve(correspondents.get(user) ?? []),
|
||||
// 3단계 — 엔진 쪽 판단(검색 조건 · 조건부 영구 삭제)은 jmap-mail.provider.spec.ts 가 본다. 여기서는 받은 값만 기록한다
|
||||
getMailState: () => Promise.resolve({ state: MAIL_STATE }),
|
||||
// 5단계 — 엔진 연결(다시 잇기 · 새 메일 계산)은 jmap-mail-watch.spec.ts · jmap-mail.provider.spec.ts 가 본다.
|
||||
// 여기서는 처음 상태만 주고, 소식은 테스트가 listener 로 흘려 넣는다
|
||||
watchMail: (
|
||||
user: string,
|
||||
listener: MailWatchListener,
|
||||
): Promise<MailWatch> => {
|
||||
if (user === EVENTS_ENGINE_DOWN) {
|
||||
return Promise.reject(
|
||||
new BusinessException('MAIL_001', {
|
||||
context: { reason: 'e2e-engine-down' },
|
||||
}),
|
||||
);
|
||||
}
|
||||
const watch = { user, listener, closed: false };
|
||||
liveWatches.push(watch);
|
||||
return Promise.resolve({
|
||||
initial: { state: MAIL_STATE, inboxUnread: 2 },
|
||||
close: () => {
|
||||
watch.closed = true;
|
||||
},
|
||||
});
|
||||
},
|
||||
listMessages: (user: string, query: MessageListQuery) => {
|
||||
listQueries.push({ user, query });
|
||||
return Promise.resolve({
|
||||
@@ -935,6 +1016,7 @@ describe('응답 계약 (e2e)', () => {
|
||||
],
|
||||
['get', '/api/recipient-suggestions'],
|
||||
['get', '/api/mail-state'],
|
||||
['get', '/api/mail-events'],
|
||||
['get', '/api/messages?q=%ED%9A%8C%EC%9D%98'],
|
||||
['patch', '/api/messages'],
|
||||
['post', '/api/message-deletions'],
|
||||
@@ -2443,6 +2525,127 @@ describe('응답 계약 (e2e)', () => {
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* 5단계 — 실시간 알림 (계약 2절 "실시간 알림").
|
||||
* 엔진 연결(다시 잇기 · 무소식 감시 · 새 메일 계산)과 연결 관리(나눠 쓰기 · 30초 · 토큰 만료 · 종료)는 단위 테스트가 본다 —
|
||||
* 여기서는 실제 HTTP 의 머리글 · 이벤트 형식 · 연결 전 실패의 표준 JSON 을 본다.
|
||||
* 응답이 끝나지 않으므로 서버를 미리 열어 http.get 으로 읽는다 (로그인 동시 시도 테스트와 같은 이유로 테스트가 끝나면 닫는다)
|
||||
*/
|
||||
describe('5단계 — 실시간 알림 (GET /api/mail-events)', () => {
|
||||
/** 받은 글자에 빈 줄로 끝난 이벤트가 n 개 쌓일 때까지 기다린다 */
|
||||
const eventsIn = (text: string) =>
|
||||
text.split('\n\n').filter((block) => block !== '');
|
||||
const waitForEvents = (stream: EventStreamClient, count: number) =>
|
||||
waitUntil(
|
||||
() =>
|
||||
stream.received().endsWith('\n\n') &&
|
||||
eventsIn(stream.received()).length >= count,
|
||||
GATE_DEADLINE_MS,
|
||||
);
|
||||
|
||||
/** 서버를 미리 열고, 끝나면 연 연결을 모두 끊고 닫는다 */
|
||||
async function withListeningServer(
|
||||
run: (httpServer: Server, opened: EventStreamClient[]) => Promise<void>,
|
||||
): Promise<void> {
|
||||
const httpServer = app.getHttpServer();
|
||||
const startedServer = !httpServer.listening;
|
||||
if (startedServer) await listenOnAnyPort(httpServer);
|
||||
const opened: EventStreamClient[] = [];
|
||||
try {
|
||||
await run(httpServer, opened);
|
||||
} finally {
|
||||
for (const stream of opened) stream.close();
|
||||
if (startedServer) await closeServer(httpServer);
|
||||
}
|
||||
}
|
||||
|
||||
it(
|
||||
'200 text/event-stream — 머리글(charset · no-cache, no-transform · X-Accel-Buffering: no)과 처음 state, 그 뒤 엔진 소식을 이어서 보낸다',
|
||||
async () => {
|
||||
const ME = 'events.me@e2e.arasystem.test';
|
||||
await withListeningServer(async (httpServer, opened) => {
|
||||
const stream = await openEventStream(httpServer, cookieOf(ME));
|
||||
opened.push(stream);
|
||||
await waitForEvents(stream, 1);
|
||||
|
||||
const { headers, statusCode } = stream.response;
|
||||
expect(statusCode).toBe(200);
|
||||
expect(headers['content-type']).toBe(
|
||||
'text/event-stream; charset=utf-8',
|
||||
);
|
||||
expect(headers['cache-control']).toBe('no-cache, no-transform');
|
||||
expect(headers['x-accel-buffering']).toBe('no');
|
||||
expect(typeof headers['x-request-id']).toBe('string');
|
||||
// 표준 래퍼 없이 — 상태 값은 GET /api/mail-state 와 같다
|
||||
expect(stream.received()).toBe(
|
||||
`event: state\ndata: ${JSON.stringify({ state: MAIL_STATE, inboxUnread: 2 })}\n\n`,
|
||||
);
|
||||
|
||||
const watch = liveWatches.find(({ user }) => user === ME);
|
||||
expect(watch).toBeDefined();
|
||||
const newMail = {
|
||||
count: 1,
|
||||
messages: [
|
||||
{
|
||||
id: 'm1',
|
||||
from: {
|
||||
name: '박서준',
|
||||
email: 'seojun.park@e2e.arasystem.test',
|
||||
},
|
||||
subject: '줄\n바꿈이 있는 제목',
|
||||
receivedAt: '2026-10-28T05:42:00.000Z',
|
||||
},
|
||||
],
|
||||
};
|
||||
watch?.listener.state({ state: 'c2Iz.c2Fr', inboxUnread: 3 });
|
||||
watch?.listener.newMail(newMail);
|
||||
await waitForEvents(stream, 3);
|
||||
|
||||
expect(eventsIn(stream.received()).slice(1)).toEqual([
|
||||
`event: state\ndata: ${JSON.stringify({ state: 'c2Iz.c2Fr', inboxUnread: 3 })}`,
|
||||
`event: new-mail\ndata: ${JSON.stringify(newMail)}`,
|
||||
]);
|
||||
});
|
||||
},
|
||||
GATED_TEST_TIMEOUT_MS,
|
||||
);
|
||||
|
||||
it('처음 연결 때 엔진에 연결할 수 없으면 스트림을 열지 않고 503 MAIL_001 (표준 실패 JSON)', async () => {
|
||||
const res = await request(server)
|
||||
.get('/api/mail-events')
|
||||
.set('Cookie', cookieOf(EVENTS_ENGINE_DOWN))
|
||||
.expect(503);
|
||||
|
||||
expect(res.headers['content-type']).toMatch(/^application\/json/);
|
||||
expect(body(res).success).toBe(false);
|
||||
expect(body(res).error?.code).toBe('MAIL_001');
|
||||
expect(body(res).error?.traceId).toBe(res.headers['x-request-id']);
|
||||
});
|
||||
|
||||
it(
|
||||
'한 직원의 연결이 5개를 넘으면 429 REQ_001 (표준 실패 JSON) — 엔진 연결은 하나를 나눠 쓴다',
|
||||
async () => {
|
||||
const ME = 'events.limit@e2e.arasystem.test';
|
||||
await withListeningServer(async (httpServer, opened) => {
|
||||
for (let i = 0; i < 5; i += 1) {
|
||||
const stream = await openEventStream(httpServer, cookieOf(ME));
|
||||
opened.push(stream);
|
||||
await waitForEvents(stream, 1);
|
||||
expect(stream.response.statusCode).toBe(200);
|
||||
}
|
||||
|
||||
const res = await request(httpServer)
|
||||
.get('/api/mail-events')
|
||||
.set('Cookie', cookieOf(ME))
|
||||
.expect(429);
|
||||
expect(body(res).error?.code).toBe('REQ_001');
|
||||
expect(liveWatches.filter(({ user }) => user === ME)).toHaveLength(1);
|
||||
});
|
||||
},
|
||||
GATED_TEST_TIMEOUT_MS,
|
||||
);
|
||||
});
|
||||
|
||||
/**
|
||||
* 출처 검사(CSRF) — 상태를 바꾸는 요청은 Sec-Fetch-Site(same-origin·none) 또는 허용 목록의 Origin 이어야 한다.
|
||||
* 둘 다 없으면 브라우저가 아닌 클라이언트로 보고 통과시킨다 (docs/mail/api-contract.md 머리말).
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
/**
|
||||
* 메일 엔진의 변화 소식 스트림(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));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user