Напреден апликативен развој (Трансакции, Pooling)

За да се разликуваат нарачките како ентитети во системот и трансакциите како теоретски поим во областа на датабазите, првите ќе ги нарекувам „нарачки", а вторите „трансакции во датабаза" во понатамошните спомнувања.

Pooling на конекции

Во Rust апликацијата се користи библиотеката sqlx со нејзиниот PgPool механизам за pooling на конекции. Pool-от се креира еднаш при стартување на апликацијата и се споделува низ сите Tauri команди преку State<'_, PgPool>.

Конфигурацијата е дефинирана во функцијата create_pool():

use sqlx::{PgPool, postgres::PgPoolOptions};

pub async fn create_pool() -> PgPool {
    let database_url =
        "postgres://db_202526z_va_prj_poshoreca_owner:4040c5ddf69@localhost:5432/db_202526z_va_prj_poshoreca";

    PgPoolOptions::new()
        .max_connections(10)
        .after_connect(|conn, _meta| {
            Box::pin(async move {
                sqlx::query("SET search_path TO project, public")
                    .execute(conn)
                    .await?;
                Ok(())
            })
        })
        .connect(database_url)
        .await
        .expect("Failed to connect to PostgreSQL")
}

Овој систем одржува до 10 паралелни конекции, што драстично ја намалува латенцијата. Дополнително, after_connect гарантира дека секоја нова конекција го има поставено search_path на project шемата.

Примери за користење на конекција од pool-от низ апликацијата:

• create_order() во src/commands/order.rs — користи pool.begin() за трансакција
• get_products() во src/commands/product.rs — користи pool.inner() за читање
• create_payment() во src/commands/payment.rs — користи pool.inner() за едноставен INSERT
• get_products_by_category() во src/commands/product.rs — користи pool.inner() за филтрирано читање

Трансакции во датабаза

За секоја операција што бара повеќекратен упис или проверка пред упис, се користи pool.begin() блок со цел да се елиминира ризикот од race conditions и делумни податоци. Ако било кој чекор врати грешка, целиот блок прави rollback.

Пример 1: Креирање на нарачка

Сместена во src/commands/order.rs, функцијата create_order ги извршува следните чекори како една атомична целина:

# Впишува нов запис во табелата orders со статус АКТИВНА. # Автоматски ја означува масата како ЗАФАТЕНА преку set_table_status. # Ако нешто не успее, целиот блок прави rollback и нарачката не се креира.

#[tauri::command]
pub async fn create_order(
    user_id: i32,
    table_id: i32,
    pool: State<'_, PgPool>
) -> Result<i32, String> {

    let mut tx = pool.begin()
        .await
        .map_err(|e| e.to_string())?;

    let order_id: i32 = sqlx::query_scalar(
        "
        INSERT INTO orders
        (
            user_id,
            table_id,
            status
        )
        VALUES
        (
            $1,
            $2,
            'АКТИВНА'
        )
        RETURNING order_id
        "
    )
        .bind(user_id)
        .bind(table_id)
        .fetch_one(&mut *tx)
        .await
        .map_err(|e| e.to_string())?;

    set_table_status(table_id, STATUS_OCCUPIED, &mut *tx)
        .await?;

    tx.commit()
        .await
        .map_err(|e| e.to_string())?;

    Ok(order_id)
}

Зошто е трансакција: Ако нарачката се креира, но set_table_status не успее, нарачката не смее да остане во базата. Трансакцијата гарантира дека или двете операции успеваат, или ниту една.

Пример 2: Процесирање на плаќање

Сместена во src/commands/order.rs, функцијата pay_order ги извршува следните чекори како една трансакција:

# Проверува дали нарачката е веќе платена. # Го зема table_id за подоцна да ја ослободи масата. # Ја пресметува вкупната сума од order_item. # Ако сумата е нула, ја отфрла трансакцијата. # Впишува запис во payment. # Ја ажурира состојбата на нарачката на ПЛАТЕНА. # Го намалува залихот на производите преку update_stock. # Ја ослободува масата преку set_table_status. # Генерира запис во invoice со број INV-{order_id}. # Ако било кој чекор не успее, целиот блок прави rollback.

#[tauri::command]
pub async fn pay_order(
    order_id: i32,
    method: String,
    pool: State<'_, PgPool>
) -> Result<String, String> {

    let mut tx = pool.begin()
        .await
        .map_err(|e| e.to_string())?;

    let existing: Option<i32> = sqlx::query_scalar(
        "
        SELECT payment_id
        FROM payment
        WHERE order_id = $1
        "
    )
        .bind(order_id)
        .fetch_optional(&mut *tx)
        .await
        .map_err(|e| e.to_string())?;

    if existing.is_some() {
        return Err("Нарачката е веќе платена".to_string());
    }

    let table_id: i32 = sqlx::query_scalar(
        "
        SELECT table_id
        FROM orders
        WHERE order_id = $1
        "
    )
        .bind(order_id)
        .fetch_one(&mut *tx)
        .await
        .map_err(|e| e.to_string())?;

    let total: Decimal = sqlx::query_scalar(
        "
        SELECT COALESCE(SUM(quantity * unit_price), 0)
        FROM order_item
        WHERE order_id = $1
        "
    )
        .bind(order_id)
        .fetch_one(&mut *tx)
        .await
        .map_err(|e| e.to_string())?;

    if total == Decimal::ZERO {
        return Err("Нарачката е празна".to_string());
    }

    let payment_id: i32 = sqlx::query_scalar(
        "
        INSERT INTO payment
        (
            order_id,
            amount,
            method
        )
        VALUES
        (
            $1,
            $2,
            $3
        )
        RETURNING payment_id
        "
    )
        .bind(order_id)
        .bind(total)
        .bind(method)
        .fetch_one(&mut *tx)
        .await
        .map_err(|e| e.to_string())?;

    sqlx::query(
        "
        UPDATE orders
        SET status = 'ПЛАТЕНА'
        WHERE order_id = $1
        "
    )
        .bind(order_id)
        .execute(&mut *tx)
        .await
        .map_err(|e| e.to_string())?;

    update_stock(order_id, &mut tx)
        .await?;

    set_table_status(table_id, STATUS_FREE, &mut *tx)
        .await?;

    let invoice_number = format!("INV-{}", order_id);

    sqlx::query(
        "
        INSERT INTO invoice
        (
            payment_id,
            invoice_number
        )
        VALUES
        (
            $1,
            $2
        )
        "
    )
        .bind(payment_id)
        .bind(&invoice_number)
        .execute(&mut *tx)
        .await
        .map_err(|e| e.to_string())?;

    tx.commit()
        .await
        .map_err(|e| e.to_string())?;

    Ok(invoice_number)
}

Зошто е трансакција: Ова е најкомплексната операција во системот. Вклучува 6+ операции кои мора сите да успеат или ниту една:

• Ако плаќањето успее, но update_stock не успее → залихот нема да се намали.
• Ако update_stock успее, но invoice не успее → фактурата нема да се генерира.
• Ако масата не се ослободи → масата останува зафатена засекогаш.

Трансакцијата гарантира atomicity — сè или ништо.

Пример 3: Додавање на ставка во нарачка

Сместена во src/commands/order_item.rs, функцијата add_order_item користи трансакција за да провери дали производот веќе постои во нарачката пред да го вметне или ажурира.

#[tauri::command]
pub async fn add_order_item(
    token: String,
    order_id: i32,
    product_id: i32,
    quantity: i32,
    pool: State<'_, PgPool>,
    sessions: State<'_, ActiveSessions>,
) -> Result<(), String> {
    {
        let lock = sessions.0.lock().map_err(|_| "Системска грешка".to_string())?;
        if !lock.contains_key(&token) {
            return Err("Невалидна сесија. Најавете се повторно.".to_string());
        }
    }

    if quantity <= 0 {
        return Err("Количината мора да биде поголема од 0".to_string());
    }

    let mut tx = pool.begin().await.map_err(|e| e.to_string())?;

    let existing: Option<i32> = sqlx::query_scalar(
        "
        SELECT item_number
        FROM order_item
        WHERE order_id = $1
        AND product_id = $2
        "
    )
        .bind(order_id)
        .bind(product_id)
        .fetch_optional(&mut *tx)
        .await
        .map_err(|e| e.to_string())?;

    if let Some(item_number) = existing {
        sqlx::query(
            "
            UPDATE order_item
            SET quantity = quantity + $3
            WHERE order_id = $1
            AND item_number = $2
            "
        )
            .bind(order_id)
            .bind(item_number)
            .bind(quantity)
            .execute(&mut *tx)
            .await
            .map_err(|e| e.to_string())?;
    } else {
        let next_item_number: i32 = sqlx::query_scalar(
            "
            SELECT COALESCE(MAX(item_number), 0) + 1
            FROM order_item
            WHERE order_id = $1
            "
        )
            .bind(order_id)
            .fetch_one(&mut *tx)
            .await
            .map_err(|e| e.to_string())?;

        let unit_price: rust_decimal::Decimal = sqlx::query_scalar(
            "SELECT price FROM product WHERE product_id = $1"
        )
            .bind(product_id)
            .fetch_one(&mut *tx)
            .await
            .map_err(|e| e.to_string())?;

        sqlx::query(
            "
            INSERT INTO order_item (order_id, item_number, product_id, quantity, unit_price)
            VALUES ($1, $2, $3, $4, $5)
            "
        )
            .bind(order_id)
            .bind(next_item_number)
            .bind(product_id)
            .bind(quantity)
            .bind(unit_price)
            .execute(&mut *tx)
            .await
            .map_err(|e| e.to_string())?;
    }

    tx.commit().await.map_err(|e| e.to_string())?;
    Ok(())
}

Пример 4: Затворање на смена (Z-извештај)

Сместена во src/commands/report.rs, функцијата z_report користи трансакција за атомично пресметување и впишување на Z-извештајот.

#[tauri::command]
pub async fn z_report(
    token: String,
    pool: State<'_, PgPool>,
    sessions: State<'_, ActiveSessions>,
) -> Result<serde_json::Value, String> {
    let session = require_admin_session(&token, sessions.inner()).await?;
    let admin_id = session.user_id;

    let mut tx = pool.begin()
        .await
        .map_err(|e| e.to_string())?;

    let last_close: Option<chrono::NaiveDateTime> = sqlx::query_scalar(
        "SELECT MAX(closed_at) FROM shift_close"
    )
        .fetch_one(&mut *tx)
        .await
        .map_err(|e| e.to_string())?;

    let since = last_close.unwrap_or_else(|| {
        chrono::NaiveDate::from_ymd_opt(2000, 1, 1)
            .unwrap()
            .and_hms_opt(0, 0, 0)
            .unwrap()
    });

    let row = sqlx::query(
        "
        SELECT
            COALESCE(SUM(amount), 0) as total,
            COUNT(*) as order_count
        FROM payment
        WHERE payment_date > $1
        "
    )
        .bind(since)
        .fetch_one(&mut *tx)
        .await
        .map_err(|e| e.to_string())?;

    let total: Decimal = row.get("total");
    let order_count: i64 = row.get("order_count");

    sqlx::query(
        "
        INSERT INTO shift_close (closed_by, total, order_count)
        VALUES ($1, $2, $3)
        "
    )
        .bind(admin_id)
        .bind(total)
        .bind(order_count as i32)
        .execute(&mut *tx)
        .await
        .map_err(|e| e.to_string())?;

    tx.commit()
        .await
        .map_err(|e| e.to_string())?;

    Ok(serde_json::json!({
        "total": total.to_string(),
        "order_count": order_count,
        "since": since.to_string()
    }))
}

Пример 5: Намалување на залиха при продажба

Сместена во src/commands/inventory.rs, функцијата update_stock се повикува во истата трансакција како и pay_order, што гарантира дека залихот е секогаш синхронизиран со продажбата.

pub async fn update_stock(
    order_id: i32,
    conn: &mut sqlx::PgConnection
) -> Result<(), String> {
    let rows = sqlx::query(
        "SELECT product_id, quantity FROM order_item WHERE order_id = $1"
    )
        .bind(order_id)
        .fetch_all(&mut *conn)
        .await
        .map_err(|e| e.to_string())?;

    for row in rows {
        let p_id: i32 = row.get("product_id");
        let order_qty: i32 = row.get("quantity");

        let recipe_row = sqlx::query(
            "SELECT recipe_id FROM recipe WHERE product_id = $1"
        )
            .bind(p_id)
            .fetch_optional(&mut *conn)
            .await
            .map_err(|e| e.to_string())?;

        if let Some(r) = recipe_row {
            let r_id: i32 = r.get("recipe_id");

            let ings = sqlx::query(
                "SELECT ingredient_id, quantity_needed FROM recipe_item WHERE recipe_id = $1"
            )
                .bind(r_id)
                .fetch_all(&mut *conn)
                .await
                .map_err(|e| e.to_string())?;

            for ing in ings {
                let ing_id: i32 = ing.get("ingredient_id");
                let qty_needed: rust_decimal::Decimal = ing.get("quantity_needed");

                sqlx::query(
                    "UPDATE ingredient SET current_stock = current_stock - ($1 * $2) WHERE ingredient_id = $3"
                )
                    .bind(qty_needed)
                    .bind(order_qty as f64)
                    .bind(ing_id)
                    .execute(&mut *conn)
                    .await
                    .map_err(|e| e.to_string())?;
            }
        } else {
            sqlx::query(
                "INSERT INTO inventory (product_id, quantity_change, operation_type) VALUES ($1, -$2, 'ПРОДАЖБА')"
            )
                .bind(p_id)
                .bind(order_qty)
                .execute(&mut *conn)
                .await
                .map_err(|e| e.to_string())?;
        }
    }
    Ok(())
}

Заштитен механизам

Ако било кој чекор во pool.begin() блокот врати грешка (на пример: нарачката е веќе платена, сумата е нула, масата не постои), целиот блок прави rollback и извршувањето се прекинува. Ова значи дека ниту еден делумен податок не останува трајно зачуван во базата.

Овој механизам е критичен за:

• Да се спречи креирање на нарачка без означена маса како зафатена.
• Да се спречи двојно плаќање на иста нарачка (проверка existing.is_some()).
• Да се спречи ослободување на маса без успешно плаќање.
• Да се осигура дека залихот е секогаш конзистентен со продажбата.
• Да се осигура дека фактурата е секогаш генерирана со плаќањето.
• Да се осигура дека Z-извештајот е секогаш конзистентен со плаќањата.