entity-derive

Генерация доменных событий для изменений жизненного цикла сущности. События позволяют вести журнал аудита, реализовывать Event Sourcing и интегрироваться с очередями сообщений.

Быстрый старт

#[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 генерирует enum событий:

/// Сгенерировано 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();
    }
}

Паттерн Event Sourcing

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),   // Окончательное удаление
}

Лучшие практики

1. Публикация после коммита — Публикуйте события только после успешной транзакции в БД
2. Идемпотентные обработчики — Обработчики событий должны быть идемпотентными для семантики at-least-once
3. Включайте контекст — Рассмотрите добавление метаданных (user_id, timestamp, correlation_id)
4. Асинхронная обработка — Используйте фоновые воркеры для тяжёлой обработки событий
5. 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 с событиями команд
• [[Лучшие-практики|Лучшие практики]] — Советы для продакшена