PostgreSQL ve transaction · WebSocket

Pool’u DI ile paylaşın, aynı mesajın servislerini Scoped PgDbContext’e bağlayın ve transaction sınırını açıkça kurun.

lily_postgresql default single modunda DI pool servisi ve Scoped PgDbContext sağlar. WebSocket bağlantısının bütün ömrünü tek transaction'a çevirmeyin; bir mesajın iş birimini açık transaction içinde yürütün.

Bağımlılık ve config#

İlk uygulamadaki lilyrs satırını aşağıdaki feature listesiyle güncelleyin. Driver tiplerini lilyrs::postgresql::{diesel, diesel_async} re-export'larından kullanmak sürümleri uyumlu tutar.

lilyrs = { version = "0.1.0", features = ["websocket", "postgresql"] }

[postgresql]
mode = "single"
connection_string = "postgresql://lily:lily@127.0.0.1:5432/lily"
[postgresql.tls]
mode = "disable"
[postgresql.pool]
max_size = 10
acquire_timeout_secs = 5
transaction_cleanup_timeout_secs = 5

Bu local config'dir. Production için verify-full, hostname/CA doğrulaması ve gerçek secret politikası kullanılır. Schema ve migration uygulamanıza aittir; repository derive bunları oluşturmaz. Ortak kurulumun ayrıntıları HTTP PostgreSQL rehberindedir.

Mesaj scoped iş birimi#

Örnek users(id, name) tablosunu varsayar. Controller, servis ve repository'nin tamamı aşağıdadır; gerçek kullanıcı/nesne authorization'ını guard veya domain servisinde ekleyin.

use std::sync::Arc;
use lilyrs::websocket::Injectable;
use lilyrs::websocket::ServiceTrait;
use lilyrs::postgresql::{diesel, diesel_async::RunQueryDsl, PgDbContext, PgResult};
use lilyrs::websocket::{Ack, Extensions, Payload, Service, WebSocketActionError,
    WebSocketController, WebSocketControllerInitError, WebSocketControllerTrait,
    async_trait, websocket_controller};
use serde::Deserialize;

#[derive(Injectable)]
#[service(lifetime = "Scoped")]
struct UserRepository {
    #[inject]
    context: Arc<PgDbContext>,
}
impl ServiceTrait for UserRepository {}
impl UserRepository {
    async fn rename(&self, id: i64, name: String) -> PgResult<()> {
        self.context.with_connection(move |connection, _| Box::pin(async move {
            diesel::sql_query("UPDATE users SET name = $1 WHERE id = $2")
                .bind::<diesel::sql_types::Text, _>(name)
                .bind::<diesel::sql_types::BigInt, _>(id)
                .execute(connection).await?;
            Ok(())
        }), None).await
    }
}

#[derive(Injectable)]
#[service(lifetime = "Scoped")]
struct UserService {
    #[inject]
    context: Arc<PgDbContext>,
    #[inject]
    users: Arc<UserRepository>,
}
impl ServiceTrait for UserService {}
impl UserService {
    async fn rename_pair(&self, first: i64, second: i64) -> PgResult<()> {
        let users = Arc::clone(&self.users);
        self.context.transaction(move |_| async move {
            users.rename(first, "Ada".into()).await?;
            users.rename(second, "Deniz".into()).await?;
            Ok(())
        }, None).await // WS cancellation, Pg ExecutionCancellation tipine çevrilemez.
    }
}

#[derive(Deserialize)]
struct RenamePair { first: i64, second: i64 }
#[derive(WebSocketController)]
#[namespace("users")]
struct UserController;
#[async_trait]
impl WebSocketControllerTrait for UserController {
    async fn new(_: Arc<Extensions>) -> Result<Self, WebSocketControllerInitError> { Ok(Self) }
}
#[websocket_controller]
impl UserController {
    #[message("renamePair")]
    async fn rename_pair(&self, Payload(input): Payload<RenamePair>,
        Service(service): Service<UserService>) -> Result<Ack<bool>, WebSocketActionError>
    {
        // Gerçek uygulamada kullanıcı/nesne yetkisi guard veya domain servisinde doğrulanır.
        service.rename_pair(input.first, input.second).await
            .map_err(WebSocketActionError::internal)?;
        Ok(Ack::new(true))
    }
}

Tam örnek dosyasını indirin.

UserService ve UserRepository aynı mesajda aynı PgDbContext instance'ını alır. transaction callback'i içindeki iki repository çağrısı aynı transaction'a katılır. Sonraki WebSocket mesajında yeni scope açılır. DTO imzasında payload önce yazılsa da Service parts extractor'ı önce çözülür.

Cancellation tipi sınırı#

lilyrs::postgresql::ExecutionCancellation, ortak lily_cancellation tipinin re-export'udur. lilyrs::websocket::ExecutionCancellation farklı bir tiptir; aralarında public dönüşüm yoktur. HTTP örneğindeki Some(cancellation) burada derlenmez.

Örnek bu yüzden bilinçli olarak None kullanır. WebSocket sinyalinin aynı anda SQL'i iptal ettiği vaadi verilmez. Mesaj execution'ının kesilmesi bütün SQL işinin bittiğini veya commit'in geri alındığını kanıtlamaz. Pg'nin bounded operation, cleanup ve scope sözleşmesini koruyun; private cancellation API'lerine başvurmayın.

Repository ve sonuç#

Doğrudan PgDatabaseService::with_connection/transaction, Scoped context veya isteğe bağlı PgRepository kullanılabilir. postgresql feature'ı single seçer; named pool/context kurulumu için bunun yerine postgresql-factory kullanın ve config'i factory modeline göre yazın. İkisini aynı feature grafiğinde açmayın. Ortak transaction rehberi SQL iş birimini açıklar; WebSocket cancellation farkı bu sayfadaki gibidir.

Başarılı commit sonrası ACK iletimi başarısız olabilir. İstemci retry'sini uygulama idempotency anahtarına bağlayın. Örnek derlenmiştir; canlı veritabanı şeması veya transaction integration testi yapılmış sayılmaz.