fix: MongoDB ifNoneExist search correctness

[bomanaps]

Fixes #709 replace MongoDB's broken in-Rust ifNoneExist matcher with the real session-scoped search machinery so conditional bundle entries match correctly.

[TheCaffinatedDeveloper]

1. Unbounded parameter results
   collect_session_documents() still materializes every distinct matching resource ID before intersecting in Rust. For a broad filter such as Patient?active=true&identifier=ABC123, the active=true aggregation could return tens of thousands of grouped IDs and build a correspondingly large HashSet.
   Could this use bounded candidate paging instead? One approach is to probe each parameter to choose a selective driver, page that filter by search-index _id, and intersect the remaining filters using resource_id: {$in: }. Conditional create only needs zero, one, or more than one match.

2. Final resource fetch is also unbounded
   query.count = Some(1000) does not limit this find(). build_resource_filter() only constructs predicates, and the resulting cursor is collected without calling .limit().
   Since ifNoneExist only distinguishes 0, 1, and multiple matches, please limit resource materialization to two documents and stop immediately once the second live resource is found.

3. Offloaded-search regression
   Returning not-supported whenever search is offloaded changes existing transaction-bundle behavior. Elasticsearch cannot see uncommitted writes, but a later bundle entry still needs to match a resource created earlier in the same transaction.
   A transaction-local fallback should scan MongoDB resources through the active session, stream results, and stop after two matches rather than rejecting the operation.

4. Missing regression coverage
   Could we add tests for a broad-first query such as active=true&identifier=system|value, repeated parameters with AND semantics, same-transaction read-your-writes, duplicate index rows, and the two-result cap? This change currently has no tests, and those are exactly the cases where correctness and boundedness can drift.

[bomanaps]

Addressed here bc07bdf

[TheCaffinatedDeveloper]

1. The 1,001-ID cutoff produces false negatives
   The $limit: 1001 bounds client memory, but it makes the intersection incomplete. For example, if active=true matches 90,000 patients and the patient matching the identifier is not among the arbitrary first 1,001 grouped IDs, the intersection is empty and conditional create incorrectly creates a duplicate.
   Logging that the result may be incomplete does not make that behavior safe. Could we replace the cutoff with bounded candidate paging, using one parameter as the driver and intersecting each candidate batch against the remaining filters?

Also, because $group precedes $limit, MongoDB still groups the entire broad match set before returning 1,001 IDs.

2. Offloaded fallback ignores criteria
   if_none_exist_offloaded_scan currently ignores parameters other than _id and identifier, then applies .limit(2). For active=false, it could return arbitrary active patients and incorrectly suppress creation or report multiple matches.
   The fallback needs to evaluate every criterion against streamed resources before counting the match. If a criterion cannot be evaluated correctly, returning an explicit unsupported-query error is safer than treating it as matched.

3. Missing regression test
   I would add a test with more than 1,001 resources where the selective identifier match falls beyond the broad parameter’s first 1,001 IDs. That should expose the incomplete-intersection bug deterministically.

[TheCaffinatedDeveloper]

Could something like this work instead?

This keeps both the application memory and every distinct() response bounded by the candidate batch size. Unlike $limit: 1001 before intersection, paging continues until the complete driver result has been examined or two live matches are found, so broad parameters cannot cause false negatives. The production version should select the driver using a small bounded probe and retain the existing direct handling for _id and _lastUpdated.

const CANDIDATE_BATCH_SIZE: i64 = 128;

// Select this using a small bounded probe in the full implementation.
let driver_index = 0;
let driver_filter =
    self.build_search_index_filter(tenant_id, resource_type, &query.parameters[driver_index])?;

let mut last_index_id: Option<Bson> = None;
let mut matches = Vec::with_capacity(2);
let mut matched_ids = HashSet::new();

loop {
    let page_filter = match &last_index_id {
        Some(last_id) => doc! {
            "$and": [
                driver_filter.clone(),
                { "_id": { "$gt": last_id.clone() } }
            ]
        },
        None => driver_filter.clone(),
    };

    let mut cursor = search_index
        .find(page_filter)
        .sort(doc! { "_id": 1 })
        .projection(doc! { "_id": 1, "resource_id": 1 })
        .limit(CANDIDATE_BATCH_SIZE)
        .session(&mut *session)
        .await?;

    let mut candidate_ids = HashSet::new();
    let mut documents_read = 0;

    while cursor.advance(&mut *session).await? {
        let document = cursor.deserialize_current()?;
        last_index_id = document.get("_id").cloned();
        documents_read += 1;

        if let Ok(resource_id) = document.get_str("resource_id") {
            candidate_ids.insert(resource_id.to_string());
        }
    }

    if documents_read == 0 {
        break;
    }

    // Intersect only this bounded candidate batch.
    for (index, parameter) in query.parameters.iter().enumerate() {
        if index == driver_index
            || matches!(parameter.name.as_str(), "_id" | "_lastUpdated")
        {
            continue;
        }

        let parameter_filter =
            self.build_search_index_filter(tenant_id, resource_type, parameter)?;

        let bounded_filter = doc! {
            "$and": [
                parameter_filter,
                {
                    "resource_id": {
                        "$in": candidate_ids.iter().cloned().collect::<Vec<_>>()
                    }
                }
            ]
        };

        let matching_ids = search_index
            .distinct("resource_id", bounded_filter)
            .session(&mut *session)
            .await?
            .into_iter()
            .filter_map(|value| value.as_str().map(str::to_string))
            .collect::<HashSet<_>>();

        candidate_ids.retain(|id| matching_ids.contains(id));

        if candidate_ids.is_empty() {
            break;
        }
    }

    if !candidate_ids.is_empty() {
        let remaining = 2 - matches.len();

        let mut cursor = resources
            .find(doc! {
                "tenant_id": tenant_id,
                "resource_type": resource_type,
                "is_deleted": false,
                "id": { "$in": candidate_ids.into_iter().collect::<Vec<_>>() }
            })
            .limit(remaining as i64)
            .session(&mut *session)
            .await?;

        while cursor.advance(&mut *session).await? {
            let document = cursor.deserialize_current()?;
            let resource =
                document_to_stored_resource(&document, tenant, resource_type)?;

            if matched_ids.insert(resource.id().to_string()) {
                matches.push(resource);
            }
        }
    }

    if matches.len() == 2 {
        break;
    }

    if documents_read < CANDIDATE_BATCH_SIZE {
        break;
    }
}

return Ok(matches);