이벤트 - entity-derive

엔티티 생명주기 변경에 대한 도메인 이벤트를 생성합니다. 이벤트는 감사 로깅, 이벤트 소싱, 메시지 큐 통합을 가능하게 합니다.

빠른 시작

#[derive(Entity)]
#[entity(table = "orders", events)]
pub struct Order {
    #[id]
    pub id: Uuid,

    #[field(create, response)]
    pub customer_id: Uuid,

    #[field(create, update, response)]
    pub status: String,

    #[field(create, response)]
    pub total_cents: i64,

    #[field(response)]
    #[auto]
    pub created_at: DateTime<Utc>,
}

생성되는 코드

events 속성은 이벤트 열거형을 생성합니다:

/// entity-derive에 의해 생성됨
#[derive(Debug, Clone)]
pub enum OrderEvent {
    /// 엔티티가 생성됨.
    Created(Order),

    /// 엔티티가 업데이트됨.
    Updated {
        id: Uuid,
        changes: UpdateOrderRequest,
    },

    /// 엔티티가 삭제됨.
    Deleted(Uuid),
}

사용 예제

기본 이벤트 발행

use async_trait::async_trait;

#[async_trait]
pub trait EventBus: Send + Sync {
    async fn publish<E: Send + Sync>(&self, event: E);
}

async fn create_order(
    repo: &impl OrderRepository,
    bus: &impl EventBus,
    dto: CreateOrderRequest,
) -> Result<Order, sqlx::Error> {
    let order = repo.create(dto).await?;

    // 성공적인 생성 후 이벤트 발행
    bus.publish(OrderEvent::Created(order.clone())).await;

    Ok(order)
}

async fn update_order(
    repo: &impl OrderRepository,
    bus: &impl EventBus,
    id: Uuid,
    dto: UpdateOrderRequest,
) -> Result<Order, sqlx::Error> {
    let order = repo.update(id, dto.clone()).await?;

    bus.publish(OrderEvent::Updated { id, changes: dto }).await;

    Ok(order)
}

async fn delete_order(
    repo: &impl OrderRepository,
    bus: &impl EventBus,
    id: Uuid,
) -> Result<bool, sqlx::Error> {
    let deleted = repo.delete(id).await?;

    if deleted {
        bus.publish(OrderEvent::Deleted(id)).await;
    }

    Ok(deleted)
}

감사 로깅

struct AuditLogger {
    pool: PgPool,
}

#[async_trait]
impl EventHandler<OrderEvent> for AuditLogger {
    async fn handle(&self, event: OrderEvent) {
        let (action, entity_id, details) = match &event {
            OrderEvent::Created(order) => (
                "created",
                order.id,
                serde_json::to_string(order).unwrap(),
            ),
            OrderEvent::Updated { id, changes } => (
                "updated",
                *id,
                serde_json::to_string(changes).unwrap(),
            ),
            OrderEvent::Deleted(id) => (
                "deleted",
                *id,
                String::new(),
            ),
        };

        sqlx::query(
            "INSERT INTO audit_log (entity_type, entity_id, action, details, created_at)
             VALUES ('order', $1, $2, $3, NOW())"
        )
        .bind(entity_id)
        .bind(action)
        .bind(details)
        .execute(&self.pool)
        .await
        .ok();
    }
}

메시지 큐 통합

use rdkafka::producer::FutureProducer;

struct KafkaEventBus {
    producer: FutureProducer,
    topic: String,
}

#[async_trait]
impl EventBus for KafkaEventBus {
    async fn publish<E: Serialize + Send + Sync>(&self, event: E) {
        let payload = serde_json::to_vec(&event).unwrap();

        self.producer
            .send(
                FutureRecord::to(&self.topic)
                    .payload(&payload)
                    .key(&Uuid::new_v4().to_string()),
                Duration::from_secs(5),
            )
            .await
            .ok();
    }
}

이벤트 소싱 패턴

struct OrderAggregate {
    events: Vec<OrderEvent>,
    current_state: Option<Order>,
}

impl OrderAggregate {
    fn apply(&mut self, event: OrderEvent) {
        match &event {
            OrderEvent::Created(order) => {
                self.current_state = Some(order.clone());
            }
            OrderEvent::Updated { changes, .. } => {
                if let Some(ref mut order) = self.current_state {
                    if let Some(status) = &changes.status {
                        order.status = status.clone();
                    }
                }
            }
            OrderEvent::Deleted(_) => {
                self.current_state = None;
            }
        }
        self.events.push(event);
    }

    fn replay(events: Vec<OrderEvent>) -> Self {
        let mut aggregate = Self {
            events: Vec::new(),
            current_state: None,
        };
        for event in events {
            aggregate.apply(event);
        }
        aggregate
    }
}

소프트 삭제와 함께 사용

soft_delete가 활성화되면 추가 이벤트가 생성됩니다:

#[derive(Entity)]
#[entity(table = "documents", events, soft_delete)]
pub struct Document {
    #[id]
    pub id: Uuid,

    #[field(create, response)]
    pub title: String,

    #[field(skip)]
    pub deleted_at: Option<DateTime<Utc>>,
}

생성됨:

pub enum DocumentEvent {
    Created(Document),
    Updated { id: Uuid, changes: UpdateDocumentRequest },
    Deleted(Uuid),       // 소프트 삭제
    Restored(Uuid),      // 소프트 삭제에서 복원
    HardDeleted(Uuid),   // 영구 삭제
}

모범 사례

커밋 후 발행 — 데이터베이스 트랜잭션이 성공한 후에만 이벤트 발행
멱등성 핸들러 — 이벤트 핸들러는 at-least-once 전달을 위해 멱등성이어야 함
컨텍스트 포함 — 메타데이터 추가 고려 (user_id, timestamp, correlation_id)
비동기 처리 — 무거운 이벤트 처리에는 백그라운드 워커 사용
Dead letter queue — 실패한 이벤트를 우아하게 처리

훅과 결합

이벤트와 훅은 잘 함께 작동합니다:

#[derive(Entity)]
#[entity(table = "orders", events, hooks)]
pub struct Order { /* ... */ }

struct OrderService {
    repo: PgPool,
    bus: EventBus,
}

#[async_trait]
impl OrderHooks for OrderService {
    type Error = AppError;

    async fn after_create(&self, entity: &Order) -> Result<(), Self::Error> {
        // 훅에서 이벤트 발행
        self.bus.publish(OrderEvent::Created(entity.clone())).await;
        Ok(())
    }

    async fn after_update(&self, entity: &Order) -> Result<(), Self::Error> {
        // 여기서도 이벤트를 발행할 수 있음
        Ok(())
    }

    async fn after_delete(&self, id: &Uuid) -> Result<(), Self::Error> {
        self.bus.publish(OrderEvent::Deleted(*id)).await;
        Ok(())
    }
}

참고

[[훅|훅]] — 생명주기 이벤트에서 커스텀 로직 실행
[[커맨드|커맨드]] — 커맨드 이벤트를 포함한 CQRS 패턴
[[모범-사례|모범 사례]] — 프로덕션 팁

Keyboard shortcuts

entity-derive