rouchdb 0.3.2

//! Integration tests for new PouchDB parity features against CouchDB.
//! These require a running CouchDB instance:
//!   docker compose up -d
//!   cargo test -p rouchdb --test parity_http -- --ignored

mod common;

use common::{delete_remote_db, fresh_remote_db};
use rouchdb::{
    AllDocsOptions, ChangesOptions, ChangesStreamOptions, Database, FindOptions, IndexDefinition,
    ReplicationOptions, SortField,
};
use std::time::Duration;

// =========================================================================
// db.close() on HTTP
// =========================================================================

#[tokio::test]
#[ignore]
async fn close_http_db() {
    let url = fresh_remote_db("close").await;
    let db = Database::http(&url);
    db.put("doc1", serde_json::json!({})).await.unwrap();
    db.close().await.unwrap(); // No-op for HTTP, should not error
    delete_remote_db(&url).await;
}

// =========================================================================
// db.explain() on HTTP
// =========================================================================

#[tokio::test]
#[ignore]
async fn explain_on_http_without_index() {
    let url = fresh_remote_db("explain").await;
    let db = Database::http(&url);

    db.put("doc1", serde_json::json!({"name": "Alice", "age": 30}))
        .await
        .unwrap();

    let explanation = db
        .explain(FindOptions {
            selector: serde_json::json!({"age": {"$gt": 20}}),
            ..Default::default()
        })
        .await;

    // Should fall back to _all_docs since no Mango index
    assert_eq!(explanation.index.name, "_all_docs");

    delete_remote_db(&url).await;
}

#[tokio::test]
#[ignore]
async fn explain_on_http_with_index() {
    let url = fresh_remote_db("explain_idx").await;
    let db = Database::http(&url);

    db.put("doc1", serde_json::json!({"name": "Alice", "age": 30}))
        .await
        .unwrap();

    db.create_index(IndexDefinition {
        name: String::new(),
        fields: vec![SortField::Simple("age".into())],
        ddoc: None,
    })
    .await
    .unwrap();

    let explanation = db
        .explain(FindOptions {
            selector: serde_json::json!({"age": {"$gt": 20}}),
            ..Default::default()
        })
        .await;

    assert_eq!(explanation.index.name, "idx-age");
    assert_eq!(explanation.index.index_type, "json");

    delete_remote_db(&url).await;
}

// =========================================================================
// Design documents on HTTP
// =========================================================================

#[tokio::test]
#[ignore]
async fn design_doc_crud_on_http() {
    let url = fresh_remote_db("ddoc").await;
    let db = Database::http(&url);

    let ddoc = rouchdb::DesignDocument {
        id: "_design/myapp".into(),
        rev: None,
        views: {
            let mut v = std::collections::HashMap::new();
            v.insert(
                "by_type".into(),
                rouchdb::ViewDef {
                    map: "function(doc) { emit(doc.type, 1); }".into(),
                    reduce: Some("_count".into()),
                },
            );
            v
        },
        filters: std::collections::HashMap::new(),
        validate_doc_update: None,
        shows: std::collections::HashMap::new(),
        lists: std::collections::HashMap::new(),
        updates: std::collections::HashMap::new(),
        language: Some("javascript".into()),
    };

    let result = db.put_design(ddoc).await.unwrap();
    assert!(result.ok);

    let retrieved = db.get_design("myapp").await.unwrap();
    assert_eq!(retrieved.id, "_design/myapp");
    assert!(retrieved.views.contains_key("by_type"));

    // Delete
    let rev = retrieved.rev.unwrap();
    let del = db.delete_design("myapp", &rev).await.unwrap();
    assert!(del.ok);

    // Should be gone
    assert!(db.get_design("myapp").await.is_err());

    delete_remote_db(&url).await;
}

// =========================================================================
// Security document on HTTP
// =========================================================================

#[tokio::test]
#[ignore]
async fn security_document_on_http() {
    let url = fresh_remote_db("security").await;
    let db = Database::http(&url);

    let sec = db.get_security().await.unwrap();
    // CouchDB returns a security doc (may have admin set from URL auth)
    assert!(sec.admins.names.is_empty() || !sec.admins.names.is_empty());

    delete_remote_db(&url).await;
}

// =========================================================================
// Replication with since override (HTTP)
// =========================================================================

#[tokio::test]
#[ignore]
async fn replication_since_override_http() {
    let url = fresh_remote_db("repl_since").await;
    let remote = Database::http(&url);
    let local = Database::memory("local");

    remote
        .put("doc1", serde_json::json!({"v": 1}))
        .await
        .unwrap();
    remote
        .put("doc2", serde_json::json!({"v": 2}))
        .await
        .unwrap();
    remote
        .put("doc3", serde_json::json!({"v": 3}))
        .await
        .unwrap();

    // Get changes to find a since point
    let changes = remote.changes(ChangesOptions::default()).await.unwrap();
    let _since = changes.results[0].seq.clone();

    // Replicate only changes after since
    let result = local.replicate_from(&remote).await;
    // Just verify basic replication works to CouchDB
    assert!(result.is_ok());

    delete_remote_db(&url).await;
}

// =========================================================================
// Replication without checkpoint (HTTP)
// =========================================================================

#[tokio::test]
#[ignore]
async fn replication_no_checkpoint_http() {
    let url = fresh_remote_db("repl_nockpt").await;
    let remote = Database::http(&url);
    let local = Database::memory("local");

    local
        .put("doc1", serde_json::json!({"v": 1}))
        .await
        .unwrap();

    let result = local
        .replicate_to_with_opts(
            &remote,
            ReplicationOptions {
                checkpoint: false,
                ..Default::default()
            },
        )
        .await
        .unwrap();

    assert!(result.ok);
    assert_eq!(result.docs_written, 1);

    let doc = remote.get("doc1").await.unwrap();
    assert_eq!(doc.data["v"], 1);

    delete_remote_db(&url).await;
}

// =========================================================================
// allDocs with conflicts option on HTTP
// =========================================================================

#[tokio::test]
#[ignore]
async fn all_docs_conflicts_http() {
    let url = fresh_remote_db("alldocs_conflicts").await;
    let db = Database::http(&url);

    db.put("doc1", serde_json::json!({"v": 1})).await.unwrap();
    db.put("doc2", serde_json::json!({"v": 2})).await.unwrap();

    let result = db
        .all_docs(AllDocsOptions {
            include_docs: true,
            conflicts: true,
            ..AllDocsOptions::new()
        })
        .await
        .unwrap();

    assert_eq!(result.rows.len(), 2);

    delete_remote_db(&url).await;
}

// =========================================================================
// Changes with selector filter on HTTP
// =========================================================================

#[tokio::test]
#[ignore]
async fn changes_selector_filter_http() {
    let url = fresh_remote_db("ch_sel_http").await;
    let db = Database::http(&url);

    db.put(
        "user1",
        serde_json::json!({"type": "user", "name": "Alice"}),
    )
    .await
    .unwrap();
    db.put(
        "inv1",
        serde_json::json!({"type": "invoice", "amount": 100}),
    )
    .await
    .unwrap();
    db.put("user2", serde_json::json!({"type": "user", "name": "Bob"}))
        .await
        .unwrap();

    let changes = db
        .changes(ChangesOptions {
            selector: Some(serde_json::json!({"type": "user"})),
            include_docs: true,
            ..Default::default()
        })
        .await
        .unwrap();

    assert_eq!(changes.results.len(), 2);
    for event in &changes.results {
        let doc = event.doc.as_ref().unwrap();
        assert_eq!(doc["type"], "user");
    }

    delete_remote_db(&url).await;
}

// =========================================================================
// Live changes on HTTP with events
// =========================================================================

#[tokio::test]
#[ignore]
async fn live_changes_events_http() {
    let url = fresh_remote_db("ch_events").await;
    let db = Database::http(&url);

    db.put("doc1", serde_json::json!({"v": 1})).await.unwrap();

    let (mut rx, handle) = db.live_changes_events(ChangesStreamOptions {
        include_docs: true,
        poll_interval: Duration::from_millis(200),
        ..Default::default()
    });

    let mut got_change = false;
    let timeout = tokio::time::sleep(Duration::from_secs(5));
    tokio::pin!(timeout);

    loop {
        tokio::select! {
            event = rx.recv() => {
                match event {
                    Some(rouchdb::ChangesEvent::Change(ce)) => {
                        assert_eq!(ce.id, "doc1");
                        got_change = true;
                        break;
                    }
                    Some(_) => continue,
                    None => break,
                }
            }
            _ = &mut timeout => break,
        }
    }

    assert!(got_change);
    handle.cancel();
    delete_remote_db(&url).await;
}

// =========================================================================
// Mango find with index on HTTP
// =========================================================================

#[tokio::test]
#[ignore]
async fn mango_find_with_index_http() {
    let url = fresh_remote_db("mango_idx").await;
    let db = Database::http(&url);

    db.put(
        "alice",
        serde_json::json!({"name": "Alice", "age": 30, "type": "user"}),
    )
    .await
    .unwrap();
    db.put(
        "bob",
        serde_json::json!({"name": "Bob", "age": 25, "type": "user"}),
    )
    .await
    .unwrap();
    db.put(
        "charlie",
        serde_json::json!({"name": "Charlie", "age": 35, "type": "user"}),
    )
    .await
    .unwrap();

    // Create index
    db.create_index(IndexDefinition {
        name: String::new(),
        fields: vec![SortField::Simple("age".into())],
        ddoc: None,
    })
    .await
    .unwrap();

    // Find using index
    let result = db
        .find(FindOptions {
            selector: serde_json::json!({"age": {"$gte": 30}}),
            ..Default::default()
        })
        .await
        .unwrap();

    assert_eq!(result.docs.len(), 2);

    delete_remote_db(&url).await;
}

// =========================================================================
// Replication with events on HTTP
// =========================================================================

#[tokio::test]
#[ignore]
async fn replication_events_http() {
    let url = fresh_remote_db("repl_events").await;
    let remote = Database::http(&url);
    let local = Database::memory("local");

    local
        .put("doc1", serde_json::json!({"v": 1}))
        .await
        .unwrap();
    local
        .put("doc2", serde_json::json!({"v": 2}))
        .await
        .unwrap();

    let (result, mut rx) = local
        .replicate_to_with_events(&remote, ReplicationOptions::default())
        .await
        .unwrap();

    assert!(result.ok);

    let mut events = Vec::new();
    while let Ok(event) = rx.try_recv() {
        events.push(event);
    }

    assert!(
        events
            .iter()
            .any(|e| matches!(e, rouchdb::ReplicationEvent::Active))
    );

    delete_remote_db(&url).await;
}