MongoDB Cursor Batched Queries

use mongodb::{
    bson::{doc, Document},
    options::{AggregateOptions, FindOptions},
    Client,
};
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::with_uri_str("mongodb://localhost:27017").await?;
    let orders = client.database("shop").collection::<Document>("orders");

    // 游标按批拉取：batch_size 控制单次往返的文档数
    let find_options = FindOptions::builder()
        .batch_size(200)
        .limit(2000)
        .sort(doc! { "created_at": -1 })
        .build();

    let mut cursor = orders
        .find(doc! { "status": "paid" })
        .with_options(find_options)
        .await?;

    let mut total_value: i64 = 0;
    let mut count = 0usize;
    while cursor.advance().await? {
        for doc in cursor.deserialize_current()? {
            count += 1;
            total_value += doc.get_i64("total_cents").unwrap_or(0);
        }
    }
    println!("find: {count} orders worth {total_value} cents");

    // 聚合管道：服务端分组统计，游标同样分批消费
    let pipeline = vec![
        doc! { "$match": doc! { "status": "paid" } },
        doc! {
            "$group": {
                "_id": "$region",
                "revenue": doc! { "$sum": "$total_cents" },
                "orders": doc! { "$sum": 1 },
            }
        },
        doc! { "$sort": doc! { "revenue": -1 } },
    ];

    let aggregate_options = AggregateOptions::builder()
        .max_time(Duration::from_secs(10))
        .build();

    let mut agg_cursor = orders
        .aggregate(pipeline)
        .with_options(aggregate_options)
        .await?;

    while agg_cursor.advance().await? {
        for doc in agg_cursor.deserialize_current()? {
            println!(
                "region {:?}: {} orders, {} cents",
                doc.get_str("_id").unwrap_or("?"),
                doc.get_i64("orders").unwrap_or(0),
                doc.get_i64("revenue").unwrap_or(0)
            );
        }
    }

    client.shutdown().await;
    Ok(())
}