제20강: 대규모 프로젝트 아키텍처

이벤트 소싱 구현

// domain/aggregates/Order.ts
export class Order extends AggregateRoot {
    private orderId: OrderId;
    private customerId: CustomerId;
    private items: OrderItem[] = [];
    private status: OrderStatus;
    private totalAmount: Money;
    private version: number = 0;

    // 이벤트 소싱: 이벤트에서 상태 복원
    static fromEvents(events: DomainEvent[]): Order {
        const order = new Order();
        
        events.forEach(event => {
            order.apply(event, false);
        });
        
        return order;
    }

    // 명령 처리
    placeOrder(command: PlaceOrderCommand): void {
        // 비즈니스 규칙 검증
        if (this.items.length === 0) {
            throw new EmptyOrderException();
        }

        // 이벤트 생성 및 적용
        const event = new OrderPlacedEvent({
            orderId: this.orderId,
            customerId: this.customerId,
            items: this.items,
            totalAmount: this.calculateTotal(),
            timestamp: new Date()
        });

        this.apply(event);
    }

    cancelOrder(reason: string): void {
        if (!this.canBeCancelled()) {
            throw new OrderCannotBeCancelledException(this.status);
        }

        const event = new OrderCancelledEvent({
            orderId: this.orderId,
            reason,
            timestamp: new Date()
        });

        this.apply(event);
    }

    // 이벤트 적용
    protected when(event: DomainEvent): void {
        switch (event.constructor) {
            case OrderPlacedEvent:
                this.onOrderPlaced(event as OrderPlacedEvent);
                break;
            case OrderCancelledEvent:
                this.onOrderCancelled(event as OrderCancelledEvent);
                break;
            case OrderShippedEvent:
                this.onOrderShipped(event as OrderShippedEvent);
                break;
        }
    }

    private onOrderPlaced(event: OrderPlacedEvent): void {
        this.orderId = event.orderId;
        this.customerId = event.customerId;
        this.items = event.items;
        this.status = OrderStatus.PLACED;
        this.totalAmount = event.totalAmount;
    }

    private onOrderCancelled(event: OrderCancelledEvent): void {
        this.status = OrderStatus.CANCELLED;
    }
}

// infrastructure/eventstore/EventStore.ts
export interface EventStore {
    saveEvents(aggregateId: string, events: DomainEvent[], expectedVersion: number): Promise;
    getEvents(aggregateId: string): Promise;
    getEventsFromSnapshot(aggregateId: string, version: number): Promise;
    saveSnapshot(aggregateId: string, snapshot: AggregateSnapshot): Promise;
}

@Injectable()
export class PostgresEventStore implements EventStore {
    constructor(
        private readonly db: DatabaseConnection,
        private readonly serializer: EventSerializer
    ) {}

    async saveEvents(
        aggregateId: string, 
        events: DomainEvent[], 
        expectedVersion: number
    ): Promise {
        const client = await this.db.getClient();
        
        try {
            await client.query('BEGIN');

            // 낙관적 동시성 제어
            const result = await client.query(
                'SELECT MAX(version) as version FROM events WHERE aggregate_id = $1',
                [aggregateId]
            );

            const currentVersion = result.rows[0]?.version || 0;
            
            if (currentVersion !== expectedVersion) {
                throw new ConcurrencyException(
                    `Expected version ${expectedVersion} but was ${currentVersion}`
                );
            }

            // 이벤트 저장
            for (const event of events) {
                const eventData = this.serializer.serialize(event);
                
                await client.query(
                    `INSERT INTO events 
                     (event_id, aggregate_id, event_type, event_data, version, timestamp)
                     VALUES ($1, $2, $3, $4, $5, $6)`,
                    [
                        event.eventId,
                        aggregateId,
                        event.constructor.name,
                        eventData,
                        ++currentVersion,
                        event.timestamp
                    ]
                );
            }

            await client.query('COMMIT');
            
            // 이벤트 발행
            await this.publishEvents(events);
            
        } catch (error) {
            await client.query('ROLLBACK');
            throw error;
        } finally {
            client.release();
        }
    }

    async getEvents(aggregateId: string): Promise {
        const result = await this.db.query(
            `SELECT event_type, event_data, version, timestamp 
             FROM events 
             WHERE aggregate_id = $1 
             ORDER BY version`,
            [aggregateId]
        );

        return result.rows.map(row => 
            this.serializer.deserialize(row.event_type, row.event_data)
        );
    }
}

// application/commands/OrderCommandHandler.ts
@Injectable()
export class OrderCommandHandler {
    constructor(
        private readonly eventStore: EventStore,
        private readonly orderRepository: OrderRepository
    ) {}

    async handle(command: PlaceOrderCommand): Promise {
        // 이벤트 스토어에서 현재 상태 로드
        const events = await this.eventStore.getEvents(command.orderId);
        const order = events.length > 0 
            ? Order.fromEvents(events)
            : Order.create(command);

        // 명령 실행
        order.placeOrder(command);

        // 새 이벤트 저장
        const uncommittedEvents = order.getUncommittedEvents();
        await this.eventStore.saveEvents(
            command.orderId,
            uncommittedEvents,
            order.getVersion()
        );

        // Read Model 업데이트 (CQRS)
        await this.updateReadModel(order);
    }

    private async updateReadModel(order: Order): Promise {
        // 읽기 전용 모델 업데이트
        const orderView = OrderView.fromAggregate(order);
        await this.orderRepository.save(orderView);
    }
}

// application/queries/OrderQueryHandler.ts
@Injectable()
export class OrderQueryHandler {
    constructor(
        private readonly readDb: ReadDatabaseConnection,
        private readonly cache: CacheService
    ) {}

    async getOrderById(orderId: string): Promise {
        // 캐시 확인
        const cached = await this.cache.get(`order:${orderId}`);
        if (cached) return cached;

        // Read DB에서 조회
        const result = await this.readDb.query(
            `SELECT * FROM order_views WHERE order_id = $1`,
            [orderId]
        );

        if (!result.rows[0]) {
            throw new OrderNotFoundException(orderId);
        }

        const orderView = OrderView.fromDb(result.rows[0]);
        
        // 캐시 저장
        await this.cache.set(`order:${orderId}`, orderView, 300);
        
        return orderView;
    }

    async getOrdersByCustomer(
        customerId: string, 
        pagination: PaginationParams
    ): Promise> {
        const result = await this.readDb.query(
            `SELECT * FROM order_views 
             WHERE customer_id = $1 
             ORDER BY created_at DESC 
             LIMIT $2 OFFSET $3`,
            [customerId, pagination.limit, pagination.offset]
        );

        const countResult = await this.readDb.query(
            `SELECT COUNT(*) FROM order_views WHERE customer_id = $1`,
            [customerId]
        );

        return {
            items: result.rows.map(row => OrderView.fromDb(row)),
            total: parseInt(countResult.rows[0].count),
            page: pagination.page,
            pageSize: pagination.limit
        };
    }
}