이벤트 소싱의 이론부터 실전까지: NestJS로 구현하는 리모트 컨피그 서비스

{ "entityId": "uuid-123", "type": "SetParameter", "timestamp": "2023-11-01T12:34:56Z", "userId": "1234" "body": { "key": "hello", "value": "world" } }

{ "parameters": { "hello": "world" } }

let state; for (const event of events) { const nextState = reducer(state, event); state = nextState; } console.log(state); // 최종 상태 // Array.reduce를 쓰는게 일반적이나, 쉬운 이해를 위해 for 문 사용

// 기본 이벤트 인터페이스 export interface BaseEvent { entityId: string; timestamp: Date; type: string; userId?: string; // 감사 로그를 위한 사용자 ID }

import { NestFactory } from '@nestjs/core'; import { AppModule } from '@src/app.module'; import { ValidationPipe } from '@nestjs/common'; async function bootstrap() { const app = await NestFactory.create(AppModule); await app.listen(3000); console.log(`Application is running on: ${await app.getUrl()}`); } bootstrap();

// 기본 이벤트 인터페이스 export interface BaseEvent { entityId: string; timestamp: Date; type: string; userId?: string; // 감사 로그를 위한 사용자 ID } // 설정 생성 이벤트 export interface ConfigCreatedEvent extends BaseEvent { type: 'ConfigCreated'; body: { name: string; description?: string; }; } // 파라미터 설정 이벤트 export interface SetParameterEvent extends BaseEvent { type: 'SetParameter'; body: { key: string; value: string | number | boolean; }; } // 파라미터 삭제 이벤트 export interface DeleteParameterEvent extends BaseEvent { type: 'DeleteParameter'; body: { key: string; }; } // 모든 설정 관련 이벤트 타입 export type ConfigEvent = ConfigCreatedEvent | SetParameterEvent | DeleteParameterEvent; // 설정 상태 인터페이스 export interface ConfigState { id?: string; name?: string; description?: string; parameters: Record<string, any>; webhooks: Array<{ url: string; events: string[]; }>; version?: number; createdAt?: Date; updatedAt?: Date; }

import { Prop, Schema, SchemaFactory } from '@nestjs/mongoose'; import { Document } from 'mongoose'; @Schema({ timestamps: true }) export class Event extends Document { @Prop({ required: true }) entityId: string; @Prop({ required: true }) type: string; @Prop({ required: true, default: Date.now }) timestamp: Date; @Prop() userId?: string; @Prop({ type: Object, required: true }) body: Record<string, any>; } export const EventSchema = SchemaFactory.createForClass(Event); // timestamp를 기준으로 정렬을 위한 인덱스 추가 EventSchema.index({ entityId: 1, timestamp: 1 });

import { ConfigEvent, ConfigState } from '@src/configs/events/config-event'; // 초기 상태 정의. 기본 상태를 별도 상수로 정의하여 재사용성과 가독성을 높임 export const initialConfigState: ConfigState = { parameters: {}, webhooks: [], }; /** * 설정 리듀서 함수 * 이전 상태와 이벤트를 받아 새로운 상태를 반환합니다. */ export function configReducer( prevState: ConfigState = initialConfigState, event: ConfigEvent ): ConfigState { // 항상 새 객체를 반환하여 불변성 유지 const newState = { ...prevState }; // 이벤트 타입에 따라 상태 갱신 switch (event.type) { case 'ConfigCreated': return { ...newState, id: event.entityId, name: event.body.name, description: event.body.description, createdAt: event.timestamp, updatedAt: event.timestamp, }; case 'SetParameter': return { ...newState, parameters: { ...newState.parameters, [event.body.key]: event.body.value }, updatedAt: event.timestamp, }; case 'DeleteParameter': { const newParams = { ...newState.parameters }; delete newParams[event.body.key]; return { ...newState, parameters: newParams, updatedAt: event.timestamp, }; } default: return newState; } }

@Controller('api/v1/configs') export class ConfigsController { constructor(private readonly configsService: ConfigsService) {} @Post() @HttpCode(HttpStatus.CREATED) async createConfig( @Body() createConfigDto: { name: string; description?: string }) { return this.configsService.createConfig(createConfigDto); } @Get(':id') async getConfig(@Param('id') id: string) { return this.configsService.getConfig(id); } @Post(':id/parameters') async setParameter( @Param('id') id: string, @Body() body: { key: string; value: any } ) { return this.configsService.setParameter(id, body.key, body.value); } @Delete(':id/parameters/:key') async deleteParameter( @Param('id') id: string, @Param('key') key: string ) { return this.configsService.deleteParameter(id, key); } }

@Injectable() export class ConfigsService { constructor( @Inject('CONFIG_EVENT_REPOSITORY') private readonly eventRepository: EventRepository<ConfigState>, ) {} /** * 새 컨피그 생성 */ async createConfig(createConfigDto: { name: string; description?: string }): Promise<ConfigState> { const { name, description } = createConfigDto; if (!name) { throw new BadRequestException('이름은 필수 항목입니다.'); } const entityId = uuidv4(); const { entity, commit } = await this.eventRepository.dispatch(entityId, { type: 'ConfigCreated', body: { name, description } }); await commit(); return entity; } /** * 컨피그 조회 */ async getConfig(id: string): Promise<ConfigState> { const config = await this.eventRepository.findById(id); if (!config) { throw new NotFoundException('설정을 찾을 수 없습니다.'); } return config; } /** * 컨피그 내 파라미터 추가 */ async setParameter(id: string, key: string, value: any): Promise<ConfigState> { if (!key) { throw new BadRequestException('키는 필수 항목입니다.'); } const config = await this.eventRepository.findById(id); if (!config) { throw new NotFoundException('설정을 찾을 수 없습니다.'); } const { entity, commit } = await this.eventRepository.dispatch(id, { type: 'SetParameter', body: { key, value } }); await commit(); return entity; } /** * 컨피그 내 파라미터 삭제 */ async deleteParameter(id: string, key: string): Promise<ConfigState> { const config = await this.eventRepository.findById(id); if (!config) { throw new NotFoundException('설정을 찾을 수 없습니다.'); } const { entity, commit } = await this.eventRepository.dispatch(id, { type: 'DeleteParameter', body: { key } }); await commit(); return entity; } }

@Injectable() export class EventRepository<TState> { constructor( private readonly eventModel: Model<Event>, private readonly reducer: (state: TState, event: any) => TState, private readonly initialState: TState, ) { } /** * 엔티티 ID로 상태 조회 */ async findById(entityId: string): Promise<TState | null> { const state: TState = { ...this.initialState } as TState; const events = await this.eventModel.find( { entityId } ).sort({ timestamp: 1 }).exec(); if (events.length === 0) { return null; // 엔티티가 존재하지 않음 } // 이벤트 적용하여 최종 상태 계산 const finalState = events.reduce(this.reducer, state); return finalState; } /** * 이벤트 발행 및 상태 계산 */ async dispatch( entityId: string, eventData: { type: string; body: any; userId?: string }): Promise<{ entity: TState; commit: () => Promise<void>; }> { const currentState = await this.findById(entityId) || { ...this.initialState } as TState; const event = new this.eventModel({ ...eventData, entityId, timestamp: new Date() }); const newState = this.reducer(currentState, event); return { entity: newState, commit: async () => { await event.save(); } }; } /** * 엔티티에 대한 모든 이벤트 조회 */ async getEvents(entityId: string): Promise<Event[]> { return this.eventModel.find({ entityId }).sort({ timestamp: 1 }).exec(); } }

function createConfigEventRepository( eventModel: Model<Event>, ): EventRepository<ConfigState> { const repository = new EventRepository<ConfigState>( eventModel, configReducer, initialConfigState, ); repository.registerHandler(eventHandler); return repository; } @Module({ imports: [ MongooseModule.forFeature([ { name: Event.name, schema: EventSchema }, ]) ], controllers: [ConfigsController], providers: [ ConfigsService, { provide: 'CONFIG_EVENT_REPOSITORY', useFactory: (eventModel) => { return createConfigEventRepository(eventModel); }, inject: [getModelToken(Event.name)], }, ], exports: ['CONFIG_EVENT_REPOSITORY', ConfigsService], }) export class ConfigsModule {}

export interface EventHandler<T> { handle(event: Event): Promise<void>; } @Injectable() export class ConfigEventHandler implements EventHandler<ConfigState> { constructor(private readonly webhooksService: WebhooksService) {} async handle(event: Event): Promise<void> { const eventType = event.type as ConfigEvent['type']; if (!['ConfigCreated', 'SetParameter', 'DeleteParameter'].includes(eventType)) { return; } const configEvent = event as unknown as ConfigEvent; switch (eventType) { case 'ConfigCreated': await this.handleConfigCreated(configEvent as ConfigCreatedEvent); break; case 'SetParameter': await this.handleSetParameter(configEvent as SetParameterEvent); break; case 'DeleteParameter': await this.handleDeleteParameter(configEvent as DeleteParameterEvent); break; } } private async handleConfigCreated(event: ConfigCreatedEvent): Promise<void> { const message = `새로운 설정이 생성되었습니다.\n*설정명:* ${event.body.name}\n${event.body.description ? `*설명:* ${event.body.description}` : ''}`; await this.webhooksService.notifySlack(message); } private async handleSetParameter(event: SetParameterEvent): Promise<void> { const message = `설정 파라미터가 변경되었습니다.\n*키:* ${event.body.key}\n*값:* ${event.body.value}`; await this.webhooksService.notifySlack(message); } private async handleDeleteParameter(event: DeleteParameterEvent): Promise<void> { const message = `설정 파라미터가 삭제되었습니다.\n*키:* ${event.body.key}`; await this.webhooksService.notifySlack(message); } }

@Injectable() export class EventRepository<TState> { private snapshotFrequency: number; // EventHandler 프로퍼티 추가 private eventHandlers: EventHandler<TState>[] = []; // 이벤트 핸들러 등록 registerHandler(handler: EventHandler<TState>): void { this.eventHandlers.push(handler); } async dispatch(entityId: string, eventData: { type: string; body: any; userId?: string }): Promise<{ entity: TState; commit: () => Promise<void>; }> { // 중략 return { entity: newState, commit: async () => { await event.save(); // 이벤트 처리 요청 await Promise.all(this.eventHandlers.map(handler => handler.handle(event))); } }; }

/** * 설정 이벤트 리포지토리 생성 팩토리 함수 */ function createConfigEventRepository( eventModel: Model<Event>, eventHandler: ConfigEventHandler ): EventRepository<ConfigState> { const repository = new EventRepository<ConfigState>( eventModel, configReducer, initialConfigState, ); repository.registerHandler(eventHandler); return repository; } @Module({ imports: [ MongooseModule.forFeature([ { name: Event.name, schema: EventSchema }, { name: Snapshot.name, schema: SnapshotSchema }, ]), WebhooksModule, ], controllers: [ConfigsController], providers: [ ConfigsService, ConfigEventHandler, { provide: 'CONFIG_EVENT_REPOSITORY', useFactory: (eventModel, eventHandler) => { return createConfigEventRepository(eventModel, eventHandler); }, inject: [getModelToken(Event.name), ConfigEventHandler], }, ], exports: ['CONFIG_EVENT_REPOSITORY', ConfigsService], }) export class ConfigsModule {}

/** * 엔티티 ID로 상태 조회 */ async findById(entityId: string): Promise<TState | null> { // 최신 스냅샷 조회 const snapshot = await this.snapshotModel.findOne( { entityId }, {}, { sort: { version: -1 } } ); let state: TState; let startVersion = 0; if (snapshot) { state = snapshot.state as TState; startVersion = snapshot.version; } else { state = { ...this.initialState } as TState; } // 스냅샷 이후 이벤트 조회 (시간순 정렬) const events = await this.eventModel.find( { entityId } ).sort({ timestamp: 1 }).exec(); if (events.length === 0 && !snapshot) { return null; // 엔티티가 존재하지 않음 } // 이벤트 적용하여 최종 상태 계산 const finalState = events.reduce(this.reducer, state); return finalState; }

@Injectable() export class CachedEventRepository<TState> extends EventRepository<TState> { private cache = new Map<string, TState>(); constructor( eventModel: Model<Event>, snapshotModel: Model<Snapshot>, reducer: (state: TState, event: any) => TState, initialState: TState, options: { snapshotFrequency?: number } = {} ) { super(eventModel, snapshotModel, reducer, initialState, options); } async findById(entityId: string): Promise<TState | null> { // Check cache first if (this.cache.has(entityId)) { return this.cache.get(entityId); } // If not in cache, get from database const state = await super.findById(entityId); // Store in cache if found if (state) { this.cache.set(entityId, state); } return state; } async dispatch(entityId: string, eventData: { type: string; body: any; userId?: string }): Promise<{ entity: TState; commit: () => Promise<void>; }> { const result = await super.dispatch(entityId, eventData); // Update cache after commit const originalCommit = result.commit; result.commit = async () => { await originalCommit(); this.cache.set(entityId, result.entity); }; return result; } // Optional: Method to clear cache for a specific entity clearCache(entityId: string): void { this.cache.delete(entityId); } // Optional: Method to clear entire cache clearAllCache(): void { this.cache.clear(); } }