entity-derive

Подписка на изменения сущностей в реальном времени через Postgres LISTEN/NOTIFY. Потоки позволяют создавать живые дашборды, мгновенные уведомления, инвалидацию кэша и событийно-ориентированные архитектуры.

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

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

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

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

Требования:

• Сущность должна реализовывать Serialize и Deserialize (для JSON-полезной нагрузки)
• Обязательны оба атрибута: events и streams
• Включите feature streams в Cargo.toml

[dependencies]
entity-derive = { version = "0.3", features = ["postgres", "streams"] }
serde = { version = "1", features = ["derive"] }

Генерируемый код

Атрибут streams генерирует:

Константа канала

impl Order {
    /// Имя канала Postgres NOTIFY.
    pub const CHANNEL: &'static str = "entity_orders";
}

Структура подписчика

/// Подписчик на изменения Order в реальном времени.
pub struct OrderSubscriber {
    listener: PgListener,
}

impl OrderSubscriber {
    /// Подключиться и подписаться на канал.
    pub async fn new(pool: &PgPool) -> Result<Self, sqlx::Error>;

    /// Ожидать следующее событие (блокирующий).
    pub async fn recv(&mut self) -> Result<OrderEvent, StreamError<sqlx::Error>>;

    /// Проверить наличие события без блокировки.
    pub async fn try_recv(&mut self) -> Result<Option<OrderEvent>, StreamError<sqlx::Error>>;
}

Автоматические уведомления

CRUD-операции автоматически отправляют события:

// В сгенерированном методе create():
async fn create(&self, dto: CreateOrderRequest) -> Result<Order, Self::Error> {
    let order = /* insert */;

    // Автоматически сгенерированное уведомление
    let event = OrderEvent::created(order.clone());
    let payload = serde_json::to_string(&event)?;
    sqlx::query("SELECT pg_notify($1, $2)")
        .bind(Order::CHANNEL)
        .bind(&payload)
        .execute(self)
        .await?;

    Ok(order)
}

Примеры использования

Базовая подписка

use entity_derive::StreamError;

async fn watch_orders(pool: &PgPool) -> Result<(), Box<dyn std::error::Error>> {
    let mut subscriber = OrderSubscriber::new(pool).await?;

    loop {
        match subscriber.recv().await {
            Ok(event) => {
                match event {
                    OrderEvent::Created(order) => {
                        println!("Новый заказ: {}", order.id);
                    }
                    OrderEvent::Updated { old, new } => {
                        println!("Заказ {} обновлён: {} -> {}", new.id, old.status, new.status);
                    }
                    OrderEvent::HardDeleted { id } => {
                        println!("Заказ {} удалён", id);
                    }
                    _ => {}
                }
            }
            Err(StreamError::Database(e)) => {
                eprintln!("Ошибка БД: {}", e);
                break;
            }
            Err(StreamError::Deserialize(e)) => {
                eprintln!("Неверная полезная нагрузка: {}", e);
            }
        }
    }

    Ok(())
}

Дашборд в реальном времени (Axum WebSocket)

use axum::{
    extract::{State, WebSocketUpgrade, ws::{Message, WebSocket}},
    response::IntoResponse,
};

async fn ws_handler(
    ws: WebSocketUpgrade,
    State(pool): State<PgPool>,
) -> impl IntoResponse {
    ws.on_upgrade(|socket| handle_socket(socket, pool))
}

async fn handle_socket(mut socket: WebSocket, pool: PgPool) {
    let mut subscriber = match OrderSubscriber::new(&pool).await {
        Ok(s) => s,
        Err(_) => return,
    };

    loop {
        match subscriber.recv().await {
            Ok(event) => {
                let json = serde_json::to_string(&event).unwrap();
                if socket.send(Message::Text(json)).await.is_err() {
                    break;
                }
            }
            Err(_) => break,
        }
    }
}

Инвалидация кэша

struct CacheInvalidator {
    cache: Redis,
    pool: PgPool,
}

impl CacheInvalidator {
    async fn run(&self) -> Result<(), StreamError<sqlx::Error>> {
        let mut subscriber = OrderSubscriber::new(&self.pool).await
            .map_err(StreamError::Database)?;

        loop {
            let event = subscriber.recv().await?;
            let key = format!("order:{}", event.entity_id());

            match event {
                OrderEvent::Created(_) | OrderEvent::Updated { .. } => {
                    self.cache.del(&key).await.ok();
                }
                OrderEvent::HardDeleted { id } | OrderEvent::SoftDeleted { id } => {
                    self.cache.del(&format!("order:{}", id)).await.ok();
                }
                _ => {}
            }
        }
    }
}

Фоновый воркер с graceful shutdown

use tokio::sync::watch;

async fn notification_worker(
    pool: PgPool,
    mut shutdown: watch::Receiver<bool>,
) {
    let mut subscriber = OrderSubscriber::new(&pool).await.unwrap();

    loop {
        tokio::select! {
            result = subscriber.recv() => {
                match result {
                    Ok(event) => process_event(event).await,
                    Err(e) => {
                        eprintln!("Ошибка потока: {:?}", e);
                        tokio::time::sleep(Duration::from_secs(1)).await;
                    }
                }
            }
            _ = shutdown.changed() => {
                println!("Завершение воркера уведомлений");
                break;
            }
        }
    }
}

Обработка ошибок

use entity_derive::StreamError;

match subscriber.recv().await {
    Ok(event) => { /* обработка */ }
    Err(StreamError::Database(sqlx_error)) => {
        // Потеря соединения, ошибка запроса и т.д.
        // Подписчик автоматически переподключится при следующем recv()
    }
    Err(StreamError::Deserialize(message)) => {
        // Неверный JSON
        // Логируем и продолжаем - не крашим цикл
    }
}

Архитектура

CRUD-операция (create/update/delete)
         │
         ▼
    pg_notify(channel, event_json)
         │
         ▼
    Postgres NOTIFY
         │
    ┌────┴────┐
    ▼         ▼
Подписчик  Подписчик  (несколько слушателей)
    │         │
    ▼         ▼
 WebSocket   Инвалидатор
 Дашборд     кэша

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

1. Переподключение — PgListener автоматически переподключается; проектируйте цикл с учётом временных сбоев
2. Идемпотентность — События могут доставляться несколько раз; обработчики должны быть идемпотентными
3. Размер полезной нагрузки — Держите сущности компактными; большие данные могут превысить лимиты Postgres
4. Отдельные пулы — Используйте выделенный пул соединений для слушателей
5. Мониторинг — Логируйте ошибки потока и отслеживайте задержку обработки событий
6. Graceful shutdown — Используйте select! с сигналом завершения для освобождения ресурсов

С мягким удалением

При включённом soft_delete доступны дополнительные события:

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

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

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

// События включают:
// - DocumentEvent::SoftDeleted { id }
// - DocumentEvent::Restored { id }
// - DocumentEvent::HardDeleted { id }

См. также

• [[События|События]] — Перечисление событий без потоковой передачи
• [[Хуки|Хуки]] — Выполнение кастомной логики при событиях жизненного цикла
• [[Лучшие практики|Лучшие-практики]] — Советы для продакшена