mod common;

use common::{
    add_to_cart, checkout, client, login_admin, pay, register_customer, setup_sellable, spawn_app,
    TestApp,
};
use serial_test::serial;

// Messaging suite: every test builds its own shop, customer, and orders and
// only asserts on rows it created.

struct Shop {
    owner: String,
    customer: String,
    sku_id: String,
}

async fn shop_with_customer(app: &TestApp, label: &str, price_minor: i64) -> Shop {
    let admin = login_admin(app).await;
    let (owner, _shop_id, _product, sku_id) = setup_sellable(app, &admin, label, price_minor, 50).await;
    let (customer, _) = register_customer(app, label).await;
    Shop {
        owner,
        customer,
        sku_id,
    }
}

/// Place and pay one order; returns (order_id, order_item_id, order_no).
async fn place_paid_order(app: &TestApp, shop: &Shop, qty: i32) -> (String, String, String) {
    add_to_cart(app, &shop.customer, &shop.sku_id, qty).await;
    let orders = checkout(app, &shop.customer).await;
    let order_id = orders[0]["id"].as_str().unwrap().to_string();
    let order_no = orders[0]["order_no"].as_str().unwrap().to_string();
    pay(app, &shop.customer, &order_id).await;

    let detail: serde_json::Value = client()
        .get(app.url(&format!("/api/shop/orders/{order_id}")))
        .bearer_auth(&shop.owner)
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    let item_id = detail["items"][0]["id"].as_str().unwrap().to_string();
    (order_id, item_id, order_no)
}

/// Create a shipment and mark it shipped (emits `order_shipped`).
async fn ship_order(app: &TestApp, shop: &Shop, order_id: &str, item_id: &str, qty: i32) -> String {
    let res = client()
        .post(app.url(&format!("/api/shop/orders/{order_id}/shipments")))
        .bearer_auth(&shop.owner)
        .json(&serde_json::json!({
            "carrier": "SF", "tracking_no": "T1",
            "items": [{ "order_item_id": item_id, "qty": qty }]
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 201, "{:?}", res.text().await);
    let ship_id = res.json::<serde_json::Value>().await.unwrap()["id"]
        .as_str()
        .unwrap()
        .to_string();
    let res = client()
        .post(app.url(&format!("/api/shop/shipments/{ship_id}/ship")))
        .bearer_auth(&shop.owner)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200, "{:?}", res.text().await);
    ship_id
}

/// Complete an order: ship, then the buyer confirms receipt.
async fn complete_order(app: &TestApp, shop: &Shop, order_id: &str, item_id: &str, qty: i32) {
    let ship_id = ship_order(app, shop, order_id, item_id, qty).await;
    let res = client()
        .post(app.url(&format!("/api/shipments/{ship_id}/confirm-delivered")))
        .bearer_auth(&shop.customer)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200, "{:?}", res.text().await);
}

/// Complete a refund-only after-sale (emits `refund_completed`); returns its id.
async fn refund_order(app: &TestApp, shop: &Shop, item_id: &str, amount_minor: i64) -> String {
    let res = client()
        .post(app.url("/api/aftersales"))
        .bearer_auth(&shop.customer)
        .json(&serde_json::json!({
            "order_item_id": item_id,
            "kind": "refund_only",
            "reason": { "en": "smoke refund", "zh": "冒烟退款" },
            "amount_minor": amount_minor
        }))
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 201, "{:?}", res.text().await);
    let id = res.json::<serde_json::Value>().await.unwrap()["id"]
        .as_str()
        .unwrap()
        .to_string();
    for action in ["approve", "refund"] {
        let res = client()
            .post(app.url(&format!("/api/shop/aftersales/{id}/{action}")))
            .bearer_auth(&shop.owner)
            .send()
            .await
            .unwrap();
        assert_eq!(res.status(), 200, "{action}: {:?}", res.text().await);
    }
    id
}

async fn messages(
    app: &TestApp,
    token: &str,
    query: &[(&str, String)],
) -> serde_json::Value {
    let res = client()
        .get(app.url("/api/messages"))
        .query(query)
        .bearer_auth(token)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200, "{:?}", res.text().await);
    res.json().await.unwrap()
}

async fn unread_count(app: &TestApp, token: &str) -> i64 {
    client()
        .get(app.url("/api/messages/unread-count"))
        .bearer_auth(token)
        .send()
        .await
        .unwrap()
        .json::<serde_json::Value>()
        .await
        .unwrap()["unread"]
        .as_i64()
        .unwrap()
}

fn of_kind<'a>(page: &'a serde_json::Value, kind: &str) -> Vec<&'a serde_json::Value> {
    page["items"]
        .as_array()
        .unwrap()
        .iter()
        .filter(|m| m["kind"] == kind)
        .collect()
}

#[tokio::test]
#[serial]
async fn payment_emits_one_order_paid_message() {
    let app = spawn_app().await;
    let shop = shop_with_customer(&app, "ms-paid", 1000).await;
    let (order_id, _item, order_no) = place_paid_order(&app, &shop, 1).await;

    let page = messages(&app, &shop.customer, &[("per_page", "100".into())]).await;
    let paid = of_kind(&page, "order_paid");
    assert_eq!(paid.len(), 1);
    assert_eq!(paid[0]["reference_type"], "order");
    assert_eq!(paid[0]["reference_id"], order_id);
    assert_eq!(paid[0]["status"], "unread");
    assert!(paid[0]["read_at"].is_null());
    assert!(paid[0]["title"]["en"].as_str().unwrap().len() > 0);
    assert!(paid[0]["title"]["zh"].as_str().unwrap().len() > 0);
    assert!(paid[0]["body"]["en"].as_str().unwrap().contains(&order_no));
    assert!(paid[0]["body"]["zh"].as_str().unwrap().contains(&order_no));

    // A retried payment conflicts and cannot duplicate the message.
    let res = client()
        .post(app.url(&format!("/api/orders/{order_id}/pay")))
        .bearer_auth(&shop.customer)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 409, "{:?}", res.text().await);
    let page = messages(&app, &shop.customer, &[("per_page", "100".into())]).await;
    assert_eq!(of_kind(&page, "order_paid").len(), 1);
    assert_eq!(page["total"], 1);
}

#[tokio::test]
#[serial]
async fn dispatch_and_refund_emit_their_messages() {
    let app = spawn_app().await;
    let shop = shop_with_customer(&app, "ms-events", 2000).await;
    let (order_id, item_id, order_no) = place_paid_order(&app, &shop, 1).await;

    let ship_id = ship_order(&app, &shop, &order_id, &item_id, 1).await;
    // Re-shipping the same shipment conflicts; the message stays single.
    let res = client()
        .post(app.url(&format!("/api/shop/shipments/{ship_id}/ship")))
        .bearer_auth(&shop.owner)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 409, "{:?}", res.text().await);

    let aftersale_id = refund_order(&app, &shop, &item_id, 500).await;

    let page = messages(&app, &shop.customer, &[("per_page", "100".into())]).await;
    assert_eq!(of_kind(&page, "order_paid").len(), 1);

    let shipped = of_kind(&page, "order_shipped");
    assert_eq!(shipped.len(), 1);
    assert_eq!(shipped[0]["reference_type"], "order");
    assert_eq!(shipped[0]["reference_id"], order_id);
    assert!(shipped[0]["body"]["en"].as_str().unwrap().contains(&order_no));

    let refunded = of_kind(&page, "refund_completed");
    assert_eq!(refunded.len(), 1);
    assert_eq!(refunded[0]["reference_type"], "aftersale");
    assert_eq!(refunded[0]["reference_id"], aftersale_id);
    assert!(refunded[0]["body"]["en"].as_str().unwrap().contains(&order_no));
    assert!(refunded[0]["body"]["zh"].as_str().unwrap().contains(&order_no));
    assert_eq!(page["total"], 3);
}

#[tokio::test]
#[serial]
async fn read_state_machine_is_guarded_and_idempotent() {
    let app = spawn_app().await;
    let shop = shop_with_customer(&app, "ms-read", 1000).await;
    // Three paid orders -> three unread messages.
    for _ in 0..3 {
        place_paid_order(&app, &shop, 1).await;
    }
    let page = messages(&app, &shop.customer, &[("per_page", "100".into())]).await;
    assert_eq!(page["total"], 3);
    assert_eq!(unread_count(&app, &shop.customer).await, 3);
    let first_id = page["items"][0]["id"].as_str().unwrap().to_string();

    // Mark one read, then repeat: only the first call changes state.
    let res = client()
        .post(app.url(&format!("/api/messages/{first_id}/read")))
        .bearer_auth(&shop.customer)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200, "{:?}", res.text().await);
    let read: serde_json::Value = res.json().await.unwrap();
    assert_eq!(read["status"], "read");
    assert!(read["read_at"].is_string());
    let read_at = read["read_at"].as_str().unwrap().to_string();
    let res = client()
        .post(app.url(&format!("/api/messages/{first_id}/read")))
        .bearer_auth(&shop.customer)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200);
    let again: serde_json::Value = res.json().await.unwrap();
    assert_eq!(again["read_at"], read_at, "repeat marking must not rewrite");
    assert_eq!(unread_count(&app, &shop.customer).await, 2);

    // Mark all read flips exactly the two remaining unread rows.
    let res = client()
        .post(app.url("/api/messages/read-all"))
        .bearer_auth(&shop.customer)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200, "{:?}", res.text().await);
    assert_eq!(
        res.json::<serde_json::Value>().await.unwrap()["updated"],
        2
    );
    assert_eq!(unread_count(&app, &shop.customer).await, 0);
    let res = client()
        .post(app.url("/api/messages/read-all"))
        .bearer_auth(&shop.customer)
        .send()
        .await
        .unwrap();
    assert_eq!(
        res.json::<serde_json::Value>().await.unwrap()["updated"],
        0
    );

    // The unread-only filter empties out.
    let unread = messages(
        &app,
        &shop.customer,
        &[("unread_only", "true".into())],
    )
    .await;
    assert_eq!(unread["total"], 0);
}

#[tokio::test]
#[serial]
async fn soft_delete_is_idempotent_and_excluded_from_list_and_count() {
    let app = spawn_app().await;
    let shop = shop_with_customer(&app, "ms-delete", 1000).await;
    for _ in 0..3 {
        place_paid_order(&app, &shop, 1).await;
    }
    let page = messages(&app, &shop.customer, &[("per_page", "100".into())]).await;
    let id = page["items"][0]["id"].as_str().unwrap().to_string();

    let res = client()
        .delete(app.url(&format!("/api/messages/{id}")))
        .bearer_auth(&shop.customer)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200, "{:?}", res.text().await);
    assert_eq!(res.json::<serde_json::Value>().await.unwrap()["deleted"], true);
    assert_eq!(unread_count(&app, &shop.customer).await, 2);
    let page = messages(&app, &shop.customer, &[("per_page", "100".into())]).await;
    assert_eq!(page["total"], 2);
    assert!(!page["items"]
        .as_array()
        .unwrap()
        .iter()
        .any(|m| m["id"] == id));

    // Soft deletion is idempotent: the row is retained for audit.
    let res = client()
        .delete(app.url(&format!("/api/messages/{id}")))
        .bearer_auth(&shop.customer)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 200);
    assert_eq!(res.json::<serde_json::Value>().await.unwrap()["deleted"], false);
    let retained: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM messages WHERE id = $1::uuid")
        .bind(&id)
        .fetch_one(&app.db)
        .await
        .unwrap();
    assert_eq!(retained, 1);
}

#[tokio::test]
#[serial]
async fn unread_only_filter_matches_the_count_endpoint() {
    let app = spawn_app().await;
    let shop = shop_with_customer(&app, "ms-filter", 1000).await;
    for _ in 0..4 {
        place_paid_order(&app, &shop, 1).await;
    }
    let unread = messages(&app, &shop.customer, &[("unread_only", "true".into())]).await;
    assert_eq!(unread["total"], 4);
    assert_eq!(unread_count(&app, &shop.customer).await, 4);

    client()
        .post(app.url("/api/messages/read-all"))
        .bearer_auth(&shop.customer)
        .send()
        .await
        .unwrap();
    assert_eq!(unread_count(&app, &shop.customer).await, 0);
    let unread = messages(&app, &shop.customer, &[("unread_only", "true".into())]).await;
    assert_eq!(unread["total"], 0);
    // The full list still shows all four, now read.
    let all = messages(&app, &shop.customer, &[("per_page", "100".into())]).await;
    assert_eq!(all["total"], 4);
}

#[tokio::test]
#[serial]
async fn foreign_messages_are_unreachable() {
    let app = spawn_app().await;
    let alice = shop_with_customer(&app, "ms-alice", 1000).await;
    let bob = shop_with_customer(&app, "ms-bob", 1000).await;
    let (_order, _item, _no) = place_paid_order(&app, &alice, 1).await;

    let page = messages(&app, &alice.customer, &[("per_page", "100".into())]).await;
    let id = page["items"][0]["id"].as_str().unwrap().to_string();

    // Bob cannot see, read, or delete Alice's message.
    let bob_page = messages(&app, &bob.customer, &[("per_page", "100".into())]).await;
    assert_eq!(bob_page["total"], 0);
    let res = client()
        .post(app.url(&format!("/api/messages/{id}/read")))
        .bearer_auth(&bob.customer)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 404, "{:?}", res.text().await);
    let res = client()
        .delete(app.url(&format!("/api/messages/{id}")))
        .bearer_auth(&bob.customer)
        .send()
        .await
        .unwrap();
    assert_eq!(res.status(), 404, "{:?}", res.text().await);

    // Alice's message is untouched and still unread.
    assert_eq!(unread_count(&app, &alice.customer).await, 1);
    let still: serde_json::Value = client()
        .get(app.url("/api/messages"))
        .query(&[("per_page", "100")])
        .bearer_auth(&alice.customer)
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(still["items"][0]["id"], id);
    assert_eq!(still["items"][0]["status"], "unread");
}