vmall/apps/api/tests/favorites.rs

mod common;

use common::{
    client, create_product_with_sku, create_shop, login_admin, make_shop_owner, publish_product,
    register_customer, spawn_app,
};
use serial_test::serial;
use std::time::Duration;

use uuid::Uuid;

async fn put_product(app: &common::TestApp, token: &str, product_id: &str) -> reqwest::Response {
    client()
        .put(app.url(&format!("/api/favorites/products/{product_id}")))
        .bearer_auth(token)
        .send()
        .await
        .unwrap()
}

async fn put_shop(app: &common::TestApp, token: &str, shop_id: &str) -> reqwest::Response {
    client()
        .put(app.url(&format!("/api/favorites/shops/{shop_id}")))
        .bearer_auth(token)
        .send()
        .await
        .unwrap()
}

async fn delete_product(app: &common::TestApp, token: &str, product_id: &str) -> reqwest::Response {
    client()
        .delete(app.url(&format!("/api/favorites/products/{product_id}")))
        .bearer_auth(token)
        .send()
        .await
        .unwrap()
}

async fn delete_shop(app: &common::TestApp, token: &str, shop_id: &str) -> reqwest::Response {
    client()
        .delete(app.url(&format!("/api/favorites/shops/{shop_id}")))
        .bearer_auth(token)
        .send()
        .await
        .unwrap()
}

async fn list(
    app: &common::TestApp,
    token: &str,
    kind: &str,
    target_id: Option<&str>,
    page: Option<i64>,
    per_page: Option<i64>,
) -> serde_json::Value {
    let mut req = client()
        .get(app.url("/api/favorites"))
        .query(&[("kind", kind)]);
    if let Some(id) = target_id {
        req = req.query(&[("target_id", id)]);
    }
    if let Some(p) = page {
        req = req.query(&[("page", p)]);
    }
    if let Some(n) = per_page {
        req = req.query(&[("per_page", n)]);
    }
    let res = req.bearer_auth(token).send().await.unwrap();
    assert_eq!(res.status(), 200, "list: {:?}", res.text().await);
    res.json().await.unwrap()
}

async fn sellable(
    app: &common::TestApp,
    admin: &str,
    slug: &str,
    price: i64,
) -> (String, String, String) {
    let shop_id = create_shop(app, admin, slug).await;
    let owner = make_shop_owner(app, admin, &shop_id).await;
    let (product_id, _) = create_product_with_sku(app, &owner, slug, price, 10).await;
    publish_product(app, &owner, &product_id).await;
    (owner, shop_id, product_id)
}

#[tokio::test]
#[serial]
async fn check_constraint_rejects_invalid_target_shape() {
    let app = spawn_app().await;
    let (_token, user_id) = register_customer(&app, "fav-shape").await;
    let admin = login_admin(&app).await;
    let (_owner, shop_id, product_id) = sellable(&app, &admin, "fav-shape", 1000).await;

    let both = sqlx::query(
        "INSERT INTO favorites (user_id, product_id, shop_id)
         VALUES ($1::uuid, $2::uuid, $3::uuid)",
    )
    .bind(&user_id)
    .bind(&product_id)
    .bind(&shop_id)
    .execute(&app.db)
    .await;
    assert!(both.is_err(), "both targets must be rejected");

    let neither = sqlx::query("INSERT INTO favorites (user_id) VALUES ($1::uuid)")
        .bind(&user_id)
        .execute(&app.db)
        .await;
    assert!(neither.is_err(), "neither target must be rejected");
}

#[tokio::test]
#[serial]
async fn uniqueness_is_per_customer_and_target() {
    let app = spawn_app().await;
    let (_token, user_id) = register_customer(&app, "fav-uniq").await;
    let admin = login_admin(&app).await;
    let (_owner, shop_id, product_id) = sellable(&app, &admin, "fav-uniq", 1000).await;

    sqlx::query("INSERT INTO favorites (user_id, product_id) VALUES ($1::uuid, $2::uuid)")
        .bind(&user_id)
        .bind(&product_id)
        .execute(&app.db)
        .await
        .unwrap();
    let dup_product =
        sqlx::query("INSERT INTO favorites (user_id, product_id) VALUES ($1::uuid, $2::uuid)")
            .bind(&user_id)
            .bind(&product_id)
            .execute(&app.db)
            .await;
    assert!(dup_product.is_err());

    sqlx::query("INSERT INTO favorites (user_id, shop_id) VALUES ($1::uuid, $2::uuid)")
        .bind(&user_id)
        .bind(&shop_id)
        .execute(&app.db)
        .await
        .unwrap();
    let dup_shop =
        sqlx::query("INSERT INTO favorites (user_id, shop_id) VALUES ($1::uuid, $2::uuid)")
            .bind(&user_id)
            .bind(&shop_id)
            .execute(&app.db)
            .await;
    assert!(dup_shop.is_err());
}

#[tokio::test]
#[serial]
async fn ownership_filters_list_and_remove() {
    let app = spawn_app().await;
    let (alice, _) = register_customer(&app, "fav-alice").await;
    let (bob, _) = register_customer(&app, "fav-bob").await;
    let admin = login_admin(&app).await;
    let (_owner, shop_id, product_id) = sellable(&app, &admin, "fav-own", 2500).await;

    assert_eq!(put_product(&app, &alice, &product_id).await.status(), 200);
    assert_eq!(put_shop(&app, &alice, &shop_id).await.status(), 200);

    let bob_products = list(&app, &bob, "product", None, None, None).await;
    assert_eq!(bob_products["total"], 0);
    assert_eq!(bob_products["items"].as_array().unwrap().len(), 0);

    assert_eq!(delete_product(&app, &bob, &product_id).await.status(), 204);
    let alice_products = list(&app, &alice, "product", None, None, None).await;
    assert_eq!(alice_products["total"], 1);
    assert_eq!(alice_products["items"][0]["product"]["id"], product_id);
}

#[tokio::test]
#[serial]
async fn missing_or_unavailable_targets_are_not_found_on_add() {
    let app = spawn_app().await;
    let (token, _) = register_customer(&app, "fav-miss").await;
    let admin = login_admin(&app).await;
    let (owner, shop_id, product_id) = sellable(&app, &admin, "fav-miss", 1000).await;

    let missing = Uuid::new_v4();
    assert_eq!(
        put_product(&app, &token, &missing.to_string())
            .await
            .status(),
        404
    );
    assert_eq!(
        put_shop(&app, &token, &missing.to_string()).await.status(),
        404
    );

    let res = client()
        .post(app.url(&format!("/api/shop/products/{product_id}/unpublish")))
        .bearer_auth(&owner)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200);
    assert_eq!(put_product(&app, &token, &product_id).await.status(), 404);

    let res = client()
        .put(app.url(&format!("/api/admin/shops/{shop_id}/status")))
        .bearer_auth(&admin)
        .json(&serde_json::json!({ "status": "suspended" }))
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200);
    assert_eq!(put_shop(&app, &token, &shop_id).await.status(), 404);
}

#[tokio::test]
#[serial]
async fn repeated_add_and_remove_are_idempotent() {
    let app = spawn_app().await;
    let (token, _) = register_customer(&app, "fav-idem").await;
    let admin = login_admin(&app).await;
    let (_owner, shop_id, product_id) = sellable(&app, &admin, "fav-idem", 1800).await;

    let first = put_product(&app, &token, &product_id).await;
    assert_eq!(first.status(), 200);
    let first_id = first.json::<serde_json::Value>().await.unwrap()["id"]
        .as_str()
        .unwrap()
        .to_string();
    let second = put_product(&app, &token, &product_id).await;
    assert_eq!(second.status(), 200);
    let second_body: serde_json::Value = second.json().await.unwrap();
    assert_eq!(second_body["id"], first_id);
    assert_eq!(
        list(&app, &token, "product", None, None, None).await["total"],
        1
    );

    assert_eq!(put_shop(&app, &token, &shop_id).await.status(), 200);
    assert_eq!(put_shop(&app, &token, &shop_id).await.status(), 200);
    assert_eq!(
        list(&app, &token, "shop", None, None, None).await["total"],
        1
    );

    assert_eq!(delete_shop(&app, &token, &shop_id).await.status(), 204);
    assert_eq!(delete_shop(&app, &token, &shop_id).await.status(), 204);
    assert_eq!(
        list(&app, &token, "shop", None, None, None).await["total"],
        0
    );
    assert_eq!(
        delete_product(&app, &token, &product_id).await.status(),
        204
    );
}

#[tokio::test]
#[serial]
async fn listing_hydrates_filters_and_paginates_visible_targets() {
    let app = spawn_app().await;
    let (token, _) = register_customer(&app, "fav-list").await;
    let admin = login_admin(&app).await;
    let (owner, shop_a, product_a) = sellable(&app, &admin, "fav-lista", 500).await;
    let (_owner_b, shop_b, product_b) = sellable(&app, &admin, "fav-listb", 1500).await;
    let (_owner_c, _shop_c, product_c) = sellable(&app, &admin, "fav-listc", 900).await;

    assert_eq!(put_product(&app, &token, &product_a).await.status(), 200);
    assert_eq!(put_product(&app, &token, &product_b).await.status(), 200);
    assert_eq!(put_product(&app, &token, &product_c).await.status(), 200);
    assert_eq!(put_shop(&app, &token, &shop_a).await.status(), 200);
    assert_eq!(put_shop(&app, &token, &shop_b).await.status(), 200);

    let products = list(&app, &token, "product", None, None, None).await;
    assert_eq!(products["total"], 3);
    let items = products["items"].as_array().unwrap();
    assert_eq!(items.len(), 3);
    assert_eq!(items[0]["kind"], "product");
    assert!(items
        .iter()
        .any(|row| { row["product"]["id"] == product_b && row["product"]["price_minor"] == 1500 }));

    let filtered = list(&app, &token, "product", Some(&product_a), None, None).await;
    assert_eq!(filtered["total"], 1);
    assert_eq!(filtered["items"][0]["product"]["id"], product_a);

    let shops = list(&app, &token, "shop", None, None, None).await;
    assert_eq!(shops["total"], 2);
    assert_eq!(shops["items"][0]["kind"], "shop");
    let shop_filter = list(&app, &token, "shop", Some(&shop_b), None, None).await;
    assert_eq!(shop_filter["total"], 1);
    assert_eq!(shop_filter["items"][0]["shop"]["id"], shop_b);

    let page1 = list(&app, &token, "product", None, Some(1), Some(1)).await;
    let page2 = list(&app, &token, "product", None, Some(2), Some(1)).await;
    assert_eq!(page1["total"], 3);
    assert_eq!(page1["items"].as_array().unwrap().len(), 1);
    assert_eq!(page2["items"].as_array().unwrap().len(), 1);
    assert_ne!(page1["items"][0]["id"], page2["items"][0]["id"]);

    let res = client()
        .post(app.url(&format!("/api/shop/products/{product_a}/unpublish")))
        .bearer_auth(&owner)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200);
    let after = list(&app, &token, "product", None, None, None).await;
    assert_eq!(after["total"], 2);
    assert!(after["items"]
        .as_array()
        .unwrap()
        .iter()
        .all(|row| row["product"]["id"] != product_a));

    let res = client()
        .put(app.url(&format!("/api/admin/shops/{shop_a}/status")))
        .bearer_auth(&admin)
        .json(&serde_json::json!({ "status": "suspended" }))
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200);
    let shops_after = list(&app, &token, "shop", None, None, None).await;
    assert_eq!(shops_after["total"], 1);
    assert_eq!(shops_after["items"][0]["shop"]["id"], shop_b);
}

#[tokio::test]
#[serial]
async fn non_customers_cannot_use_favorite_routes() {
    let app = spawn_app().await;
    let admin = login_admin(&app).await;
    let res = client()
        .get(app.url("/api/favorites"))
        .query(&[("kind", "product")])
        .bearer_auth(&admin)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 403);
}

#[tokio::test]
#[serial]
async fn add_holds_target_visible_until_favorite_commits() {
    let app = spawn_app().await;
    let (token, _) = register_customer(&app, "fav-atomic").await;
    let admin = login_admin(&app).await;
    let (_owner, shop_id, _product_id) = sellable(&app, &admin, "fav-atomic", 1000).await;

    sqlx::query("DROP TRIGGER IF EXISTS favorites_atomic_delay ON favorites")
        .execute(&app.db)
        .await
        .unwrap();
    sqlx::query("DROP FUNCTION IF EXISTS favorites_atomic_delay()")
        .execute(&app.db)
        .await
        .unwrap();
    sqlx::query(
        "CREATE FUNCTION favorites_atomic_delay() RETURNS trigger AS $$
         BEGIN
           PERFORM pg_advisory_xact_lock(915001);
           RETURN NEW;
         END
         $$ LANGUAGE plpgsql",
    )
    .execute(&app.db)
    .await
    .unwrap();
    sqlx::query(
        "CREATE TRIGGER favorites_atomic_delay BEFORE INSERT ON favorites
         FOR EACH ROW EXECUTE FUNCTION favorites_atomic_delay()",
    )
    .execute(&app.db)
    .await
    .unwrap();

    let mut lock_conn = app.db.acquire().await.unwrap();
    sqlx::query("SELECT pg_advisory_lock(915001)")
        .execute(&mut *lock_conn)
        .await
        .unwrap();

    let base = app.base.clone();
    let add_token = token.clone();
    let add_shop_id = shop_id.clone();
    let add = tokio::spawn(async move {
        client()
            .put(format!("{base}/api/favorites/shops/{add_shop_id}"))
            .bearer_auth(add_token)
            .send()
            .await
            .unwrap()
    });

    let insert_waiting = tokio::time::timeout(Duration::from_secs(2), async {
        loop {
            let waiting: bool = sqlx::query_scalar(
                "SELECT EXISTS(
                   SELECT 1 FROM pg_stat_activity
                    WHERE wait_event_type = 'Lock'
                      AND query LIKE 'INSERT INTO favorites%'
                 )",
            )
            .fetch_one(&app.db)
            .await
            .unwrap();
            if waiting {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await
    .is_ok();

    let suspend_base = app.base.clone();
    let suspend_admin = admin.clone();
    let suspend_shop_id = shop_id.clone();
    let mut suspend = tokio::spawn(async move {
        client()
            .put(format!(
                "{suspend_base}/api/admin/shops/{suspend_shop_id}/status"
            ))
            .bearer_auth(suspend_admin)
            .json(&serde_json::json!({ "status": "suspended" }))
            .send()
            .await
            .unwrap()
    });
    let status_changed_before_insert =
        tokio::time::timeout(Duration::from_millis(150), &mut suspend)
            .await
            .is_ok();

    sqlx::query("SELECT pg_advisory_unlock(915001)")
        .execute(&mut *lock_conn)
        .await
        .unwrap();
    let add_response = add.await.unwrap();
    let suspend_response = if status_changed_before_insert {
        None
    } else {
        Some(suspend.await.unwrap())
    };

    sqlx::query("DROP TRIGGER favorites_atomic_delay ON favorites")
        .execute(&app.db)
        .await
        .unwrap();
    sqlx::query("DROP FUNCTION favorites_atomic_delay()")
        .execute(&app.db)
        .await
        .unwrap();

    assert!(insert_waiting, "favorite insert never reached the trigger");
    assert!(
        !status_changed_before_insert,
        "shop status changed while favorite creation was in flight"
    );
    assert_eq!(
        add_response.status(),
        200,
        "add: {:?}",
        add_response.text().await
    );
    assert_eq!(suspend_response.unwrap().status(), 200);
}

#[tokio::test]
#[serial]
async fn repeated_add_holds_existing_favorite_until_response_is_hydrated() {
    let app = spawn_app().await;
    let (token, _) = register_customer(&app, "fav-add-remove").await;
    let admin = login_admin(&app).await;
    let (_owner, shop_id, _product_id) = sellable(&app, &admin, "fav-add-remove", 1000).await;
    assert_eq!(put_shop(&app, &token, &shop_id).await.status(), 200);

    let mut lock_conn = app.db.acquire().await.unwrap();
    sqlx::query("BEGIN").execute(&mut *lock_conn).await.unwrap();
    sqlx::query("LOCK TABLE shop_profiles IN ACCESS EXCLUSIVE MODE")
        .execute(&mut *lock_conn)
        .await
        .unwrap();

    let add_base = app.base.clone();
    let add_token = token.clone();
    let add_shop_id = shop_id.clone();
    let add = tokio::spawn(async move {
        client()
            .put(format!("{add_base}/api/favorites/shops/{add_shop_id}"))
            .bearer_auth(add_token)
            .send()
            .await
            .unwrap()
    });

    let hydration_waiting = tokio::time::timeout(Duration::from_secs(2), async {
        loop {
            let waiting: bool = sqlx::query_scalar(
                "SELECT EXISTS(
                   SELECT 1 FROM pg_stat_activity
                    WHERE wait_event_type = 'Lock'
                      AND query LIKE 'SELECT f.id, f.user_id, f.created_at,%'
                 )",
            )
            .fetch_one(&app.db)
            .await
            .unwrap();
            if waiting {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await
    .is_ok();

    let delete_base = app.base.clone();
    let delete_token = token.clone();
    let delete_shop_id = shop_id.clone();
    let mut remove = tokio::spawn(async move {
        client()
            .delete(format!(
                "{delete_base}/api/favorites/shops/{delete_shop_id}"
            ))
            .bearer_auth(delete_token)
            .send()
            .await
            .unwrap()
    });
    let removed_before_hydration = tokio::time::timeout(Duration::from_millis(150), &mut remove)
        .await
        .is_ok();

    sqlx::query("ROLLBACK")
        .execute(&mut *lock_conn)
        .await
        .unwrap();
    let add_response = add.await.unwrap();
    let remove_response = if removed_before_hydration {
        None
    } else {
        Some(remove.await.unwrap())
    };

    assert!(
        hydration_waiting,
        "favorite hydration never reached the table lock"
    );
    assert!(
        !removed_before_hydration,
        "favorite was removed before the idempotent add response was hydrated"
    );
    assert_eq!(
        add_response.status(),
        200,
        "add: {:?}",
        add_response.text().await
    );
    assert_eq!(remove_response.unwrap().status(), 204);
}