실시간 2편 — SSE로 실시간을 구현했다

사용자 B mutation
   └→ resolver → NoteSseGateway.publish
        └→ SseBus.publish("project:{id}")
             └→ NoteProjectSseContributor (권한 필터 + self-echo 차단 + wire 변환)
                  └→ SseService.merge(contributors + heartbeat)
                       └→ SseController @Sse('project/:projectId')
                            └→ 사용자 A EventSource

@Controller('sse')
@UseGuards(HttpSessionAuthGuard)
export class SseController {
  constructor(private readonly sseService: SseService) {}

  /**
   * project 단위 SSE 구독 엔드포인트. EventSource 1개당 1 project.
   * 인증·권한 필터·wire 변환은 SseService가 담당한다.
   */
  @Sse('project/:projectId')
  stream(
    @Req() req: Request & { user: RequestUser },
    @Param('projectId') projectId: string
  ): Observable<MessageEvent> {
    return this.sseService.buildProjectStream(projectId, req.user.userId);
  }
}

const HEARTBEAT_INTERVAL_MS = 25_000;

/**
 * project 단위 SSE stream 빌더 — 도메인 무지(domain-agnostic) 범용 transport.
 *
 * 권한 필터·self-echo·wire 변환 같은 도메인 정책은 각 Contributor가 가진다.
 * 그래서 이 클래스는 note/project-member 등 어떤 도메인 모듈에도 의존하지 않는다(DIP).
 */
@Injectable()
export class SseService implements OnModuleInit {
  private contributors: ProjectSseContributor[] = [];

  constructor(private readonly discovery: DiscoveryService) {}

  /**
   * @SseProjectContributor() 표식이 달린 provider들을 부트스트랩 시점에 수집한다.
   * 새 도메인 이벤트는 Contributor를 추가 등록만 하면 합류한다 — 이 파일은 수정 불필요(OCP).
   */
  onModuleInit(): void {
    this.contributors = this.discovery
      .getProviders()
      .filter(
        (wrapper) =>
          wrapper.metatype != null &&
          wrapper.instance != null &&
          Reflect.getMetadata(SSE_PROJECT_CONTRIBUTOR_METADATA, wrapper.metatype) === true
      )
      .map((wrapper) => wrapper.instance as ProjectSseContributor);
  }

  buildProjectStream(projectId: string, userId: string): Observable<MessageEvent> {
    return merge(
      ...this.contributors.map((contributor) => contributor.buildStream(projectId, userId)),
      this.heartbeat$()
    );
  }

  /**
   * 25초마다 ping 신호를 보내 프록시·LB의 idle timeout으로 연결이 끊기는 걸 막는다.
   */
  private heartbeat$(): Observable<MessageEvent> {
    return interval(HEARTBEAT_INTERVAL_MS).pipe(
      map<number, MessageEvent>(() => ({ type: 'ping', data: 'keepalive' }))
    );
  }
}

/**
 * SSE 토픽 → 구독자(Subject) 라우터. 도메인을 모르는 in-memory pub/sub helper.
 *
 * 의존성 0, pure transport. 채널 키 형식(`project:${projectUlid}` 등)은 호출자가 정의한다.
 */
@Injectable()
export class SseBus {
  private readonly channels = new Map<string, Subject<MessageEvent>>();

  subscribe(topic: string): Observable<MessageEvent> {
    const subject = this.channels.get(topic) ?? new Subject<MessageEvent>();
    this.channels.set(topic, subject);

    return subject.asObservable().pipe(
      finalize(() => {
        if (!subject.observed) {
          subject.complete();
          this.channels.delete(topic);
        }
      })
    );
  }

  publish(topic: string, payload: Record<string, unknown>): void {
    this.channels.get(topic)?.next({
      data: { ...payload, ts: Date.now() },
    });
  }
}

export const SSE_PROJECT_CONTRIBUTOR_METADATA = 'sse:project-contributor';

/**
 * project SSE 스트림에 "한 도메인의 이벤트"를 기여하는 포트(추상).
 *
 * 범용 SseService는 이 인터페이스만 알고, 구현은 각 도메인 모듈이 제공한다.
 */
export interface ProjectSseContributor {
  /**
   * 주어진 사용자에게 보낼 이 도메인의 project 이벤트 스트림을 만든다.
   * 권한 필터·self-echo 차단·wire 변환은 각 구현이 책임진다.
   */
  buildStream(projectId: string, userId: string): Observable<MessageEvent>;
}

/**
 * DiscoveryService가 런타임에 수집할 수 있도록 다는 표식.
 * 이 데코레이터가 붙은 provider는 자동으로 project SSE 스트림에 합류한다(OCP).
 */
export const SseProjectContributor = (): ClassDecorator =>
  SetMetadata(SSE_PROJECT_CONTRIBUTOR_METADATA, true);

/**
 * 역할 → 볼 수 있는 note scope. WORKER는 FINAL만, PM·REVIEWER는 둘 다.
 */
const ALLOWED_NOTE_SCOPES_BY_ROLE: Record<ProjectRole, NoteScope[]> = {
  [ProjectRole.WORKER]: [NoteScope.FINAL],
  [ProjectRole.PM]: [NoteScope.GENERAL, NoteScope.FINAL],
  [ProjectRole.REVIEWER]: [NoteScope.GENERAL, NoteScope.FINAL],
  [ProjectRole.WATCHER]: [NoteScope.GENERAL, NoteScope.FINAL],
};

/**
 * note 도메인의 project SSE 이벤트 기여자.
 * note 권한 정책(role → scope), self-echo 차단, wire 변환을 여기에 가둔다.
 */
@Injectable()
@SseProjectContributor()
export class NoteProjectSseContributor implements ProjectSseContributor {
  constructor(
    private readonly bus: SseBus,
    private readonly projectMemberService: ProjectMemberService
  ) {}

  buildStream(projectId: string, userId: string): Observable<MessageEvent> {
    return this.allowedScopes$(projectId, userId).pipe(
      switchMap((allowedScopes) =>
        this.bus.subscribe(`project:${projectId}`).pipe(
          filter((ev) => this.isVisibleTo(ev.data as NoteSsePayload, userId, allowedScopes)),
          map((ev) => this.toNamedEvent(ev.data as NoteSsePayload))
        )
      )
    );
  }

  /**
   * 사용자의 project role 조회 → 허용 scope 목록. defer로 subscribe(연결) 시점에 1회 실행.
   */
  private allowedScopes$(projectId: string, userId: string): Observable<NoteScope[]> {
    return defer(async () => {
      const member = await this.projectMemberService.getProjectMemberByProjectIdAndUserId({
        projectId,
        userId,
      });
      return ALLOWED_NOTE_SCOPES_BY_ROLE[member.role] ?? [];
    });
  }

  /**
   * 자기 publish는 무시(self-echo), 허용 scope만 통과.
   */
  private isVisibleTo(
    payload: NoteSsePayload,
    userId: string,
    allowedScopes: NoteScope[]
  ): boolean {
    if (payload.actorId === userId) return false;
    return allowedScopes.includes(payload.scope as NoteScope);
  }

  /**
   * wire 변환. type을 `note.${scope}`로 매핑, actorId는 차단에만 쓰고 wire에서 제거.
   */
  private toNamedEvent(payload: NoteSsePayload): MessageEvent {
    return {
      type: `${payload.domain}.${payload.scope.toLowerCase()}`,
      data: { ts: payload.ts },
    };
  }
}

/**
 * HTTP 라우트용 Session 인증 Guard
 *
 * EventSource 같이 커스텀 헤더를 못 붙이는 클라이언트를 위해 sessionId 쿠키만 검증한다.
 * AccessToken 재발급은 본 가드에서 다루지 않으며, 만료 시 SSE 연결이 끊긴 후
 * 클라이언트가 GraphQL 게이트웨이를 통해 재발급받는다.
 */
@Injectable()
export class HttpSessionAuthGuard implements CanActivate {
  constructor(private readonly authHelper: AuthHelper) {}

  async canActivate(context: ExecutionContext): Promise<boolean> {
    if (context.getType() !== 'http') return true;

    const req = context.switchToHttp().getRequest();
    const sessionId: string | undefined = req.cookies?.[SESSION_COOKIE_NAME];

    if (!sessionId) {
      throw new UnauthorizedException('인증이 필요합니다');
    }

    const session = await this.authHelper.validateSession(sessionId, getClientIp(req));

    req.user = {
      userId: session.userId,
      email: '',
      role: session.role,
    };

    return true;
  }
}

type Handler = () => void;

/**
 * project 단위 SSE 연결의 구독 장부를 소유하는 도메인 무지 객체. React 밖에서 동작한다.
 */
export class ProjectSseConnection {
  private es: EventSource | null = null;
  /** event type → 그 type을 구독한 핸들러들. fan-out 대상. */
  private readonly handlers = new Map<string, Set<Handler>>();
  /** 네이티브 리스너를 이미 부착한 event type. 현재 연결 1개당 type별 1번만 부착한다. */
  private readonly attached = new Set<string>();

  open(projectId: string): void {
    const url = `${config.API_URL}/sse/project/${encodeURIComponent(projectId)}`;
    this.es = new EventSource(url, { withCredentials: true });
    this.attached.clear();
    // open 전에 미리 subscribe된 type들을 일괄 부착한다.
    this.handlers.forEach((_handlers, eventType) => this.attachListener(eventType));
  }

  close(): void {
    if (this.es) {
      this.es.close();
      this.es = null;
    }
    this.attached.clear();
  }

  subscribe = (eventType: string, onEvent: Handler): (() => void) => {
    const existing = this.handlers.get(eventType);
    const handlers = existing ?? new Set<Handler>();
    if (!existing) this.handlers.set(eventType, handlers);
    handlers.add(onEvent);
    this.attachListener(eventType);
    return () => {
      handlers.delete(onEvent);
    };
  };

  /** eventType에 네이티브 리스너를 1번만 붙이고, 수신 시 등록된 핸들러 전부에 fan-out한다. */
  private attachListener(eventType: string): void {
    if (!this.es) return; // 연결 전 — open()이 일괄 부착한다.
    if (this.attached.has(eventType)) return;
    this.attached.add(eventType);
    this.es.addEventListener(eventType, (event: MessageEvent) => {
      try {
        JSON.parse(event.data);
      } catch {
        return; // heartbeat 등 비-JSON 무시
      }
      const handlers = this.handlers.get(eventType);
      if (handlers) handlers.forEach((handler) => handler());
    });
  }
}

/**
 * project 단위 SSE 연결을 도메인 무지하게 소유한다. event type은 각 도메인 Provider가 subscribe로 등록한다(OCP).
 */
export const useProjectSse = (projectId: string): UseProjectSseResult => {
  const [connection] = useState(() => new ProjectSseConnection());

  useEffect(() => {
    if (!projectId) return;
    connection.open(projectId);
    return () => connection.close();
  }, [projectId, connection]);

  return { subscribe: connection.subscribe };
};

/**
 * scope ↔ wire event type 매핑. 서버가 발행하지 않는 scope(ALL)는 키가 없어 구독하지 않는다.
 */
const EVENT_TYPE_BY_SCOPE: Partial<Record<NoteScope, string>> = {
  [NoteScope.GENERAL]: 'note.general',
  [NoteScope.FINAL]: 'note.final',
};

/**
 * project SSE 연결에 note 이벤트를 등록하고, scope별 pending 신호를 소유한다.
 *
 * pending을 연결과 같은 라우트 수명에 두므로, 소비처(RefreshBar)가 탭 전환으로
 * 마운트/언마운트돼도 신호를 흘리지 않는다.
 */
export const NoteSseProvider = ({ children }: NoteSseProviderProps) => {
  const { subscribe } = useProjectSseContext();
  const [pending, setPending] = useState<Record<NoteScope, boolean>>(createInitialPending);

  useEffect(() => {
    const offs = Object.entries(EVENT_TYPE_BY_SCOPE).map(([scope, eventType]) =>
      subscribe(eventType, () =>
        setPending((prev) => ({ ...prev, [scope as NoteScope]: true }))
      )
    );
    return () => offs.forEach((off) => off());
  }, [subscribe]);

  const clearPending = (scope: NoteScope) =>
    setPending((prev) => ({ ...prev, [scope]: false }));

  return (
    <NoteSseContext.Provider value={{ pending, clearPending }}>
      {children}
    </NoteSseContext.Provider>
  );
};

export const RefreshBar = ({ scope }: RefreshBarProps) => {
  const { pending, clearPending } = useNoteSse();
  const invalidate = useInvalidateNotes();

  if (!pending[scope]) return null;

  const handleClick = () => {
    invalidate();
    clearPending(scope);
  };

  return (
    <button type="button" onClick={handleClick} className="...">
      <RefreshCw className="w-3 h-3" />
      새로운 메세지가 왔어요 · 새로고침
    </button>
  );
};