haste_fhir_search/elastic_search/
mod.rs

1use crate::{
2    IndexFailure, IndexOutcome, IndexResource, ParameterLevel, ResolvedParameter, SearchEngine,
3    SearchOptions, SearchParameterResolve, SearchReturn,
4    indexing_conversion::{self, DynamicParameterEntry, InsertableIndex},
5};
6use bytes::{Bytes, BytesMut};
7use elasticsearch::{
8    BulkOperation, BulkParts, Elasticsearch,
9    auth::Credentials,
10    cert::CertificateValidation,
11    http::{
12        Url,
13        request::Body,
14        transport::{BuildError, SingleNodeConnectionPool, TransportBuilder},
15    },
16};
17use haste_fhir_client::request::SearchRequest;
18use haste_fhir_model::r4::generated::{
19    resources::{Resource, ResourceType},
20    terminology::{BoundCode, IssueType, SearchParamType},
21};
22use haste_fhir_operation_error::{OperationOutcomeError, derive::OperationOutcomeError};
23use haste_fhirpath::FPEngine;
24use haste_jwt::{ProjectId, ResourceId, TenantId, VersionId};
25use haste_repository::types::{FHIRMethod, SupportedFHIRVersions};
26use serde::Deserialize;
27use std::{collections::HashMap, sync::Arc};
28
29mod migration;
30mod search;
31pub mod search_parameter_resolver;
32
33#[derive(Deserialize, Debug)]
34struct SearchEntryPrivate {
35    pub id: Vec<ResourceId>,
36    pub resource_type: Vec<ResourceType>,
37    pub version_id: Vec<VersionId>,
38    pub project: Vec<ProjectId>,
39}
40
41static DYNAMIC_PARAMETER_INDEX_FIELD: &str = "dynamic_parameters";
42
43/// Elasticsearch treats a literal `.` in a mapped field name as an object
44/// path separator: e.g. mapping the field
45/// `http://hl7.org/fhir/SearchParameter/Patient-name` silently produces
46/// nested objects (`http://hl7` -> `org/fhir/SearchParameter/Patient-name`),
47/// even though `/` is left alone. Search parameter canonical URLs almost
48/// always contain a dotted domain, so every active (top-level) search
49/// parameter needs its dots stripped to stay a genuinely flat field
50/// consistently in the index mapping, in the documents written to it, and in
51/// queries built against it.
52fn flatten_parameter_field_name(url: &str) -> String {
53    url.replace('.', "_")
54}
55
56#[derive(OperationOutcomeError, Debug)]
57pub enum SearchError {
58    #[fatal(
59        code = "exception",
60        diagnostic = "Failed to evaluate fhirpath expression."
61    )]
62    FHIRPathError(#[from] haste_fhirpath::FHIRPathError),
63    #[fatal(
64        code = "exception",
65        diagnostic = "Search does not support the fhir method: '{arg0:?}'"
66    )]
67    UnsupportedFHIRMethod(FHIRMethod),
68    #[fatal(
69        code = "exception",
70        diagnostic = "Failed to index resources server responded with status code: '{arg0}'"
71    )]
72    Fatal(u16),
73    #[fatal(
74        code = "exception",
75        diagnostic = "Elasticsearch server failed to index: '{arg0}'"
76    )]
77    ElasticsearchError(#[from] elasticsearch::Error),
78    #[fatal(
79        code = "exception",
80        diagnostic = "Elasticsearch server responded with an error: '{arg0}'"
81    )]
82    ElasticSearchResponseError(u16),
83    NotConnected,
84}
85
86#[derive(OperationOutcomeError, Debug)]
87pub enum SearchConfigError {
88    #[fatal(code = "exception", diagnostic = "Failed to parse URL: '{arg0}'.")]
89    UrlParseError(String),
90    #[fatal(
91        code = "exception",
92        diagnostic = "Elasticsearch client creation failed."
93    )]
94    ElasticSearchConfigError(#[from] BuildError),
95    #[fatal(
96        code = "exception",
97        diagnostic = "Unsupported FHIR version for index: '{arg0}'"
98    )]
99    UnsupportedIndex(SupportedFHIRVersions),
100}
101
102#[derive(Clone)]
103pub struct ElasticSearchEngine<SearchParameterResolver: SearchParameterResolve + 'static> {
104    parameter_resolver: Arc<SearchParameterResolver>,
105    fp_engine: Arc<FPEngine>,
106    client: Arc<Elasticsearch>,
107    /// Whether `migrate` is allowed to rebuild the index (briefly making it
108    /// unavailable) to drop columns for search parameters that no longer
109    /// exist. When `false`, `migrate` only logs which parameters would be
110    /// dropped.
111    prune_removed_search_parameters: bool,
112}
113
114/// Creates an Elasticsearch client using the provided URL and credentials.
115///
116/// # Errors
117///
118/// Returns [`SearchConfigError::UrlParseError`] if url is not a valid URL.
119///
120/// Returns a [`SearchConfigError`] if the Elasticsearch transport cannot be
121/// constructed.
122pub fn create_es_client(
123    url: &str,
124    username: String,
125    password: String,
126) -> Result<Arc<Elasticsearch>, SearchConfigError> {
127    let url = Url::parse(url).map_err(|_e| SearchConfigError::UrlParseError(url.to_string()))?;
128    let conn_pool = SingleNodeConnectionPool::new(url);
129    let transport = TransportBuilder::new(conn_pool)
130        .cert_validation(CertificateValidation::None)
131        .auth(Credentials::Basic(username, password))
132        .build()?;
133
134    let elasticsearch_client = Elasticsearch::new(transport);
135
136    Ok(Arc::new(elasticsearch_client))
137}
138
139type Tasks = tokio::task::JoinHandle<(IndexResource, Result<Bytes, OperationOutcomeError>)>;
140
141/// Target size, in bytes, for a single Elasticsearch `_bulk` request body.
142/// Kept mid-range of the recommended 5-15MB window.
143const TARGET_BULK_BATCH_BYTES: usize = 10 * 1024 * 1024;
144
145/// Serializes a bulk operation into the exact NDJSON bytes (`{"index":{...}}\n{...}\n`,
146/// or the `delete` equivalent with no second line) that would be sent to
147/// Elasticsearch. Doing this once, here, rather than letting the client
148/// serialize it later, means: (1) a document's exact size is known for
149/// byte-based batching instead of estimated, and (2) it's only JSON-encoded
150/// once -- the resulting `Bytes` is what actually gets sent, since `Bytes`
151/// implements [`Body`] as a raw passthrough (see `send_bulk_operations`).
152fn serialize_bulk_operation(
153    op: &BulkOperation<HashMap<String, InsertableIndex>>,
154) -> Result<Bytes, OperationOutcomeError> {
155    let mut buf = BytesMut::new();
156    op.write(&mut buf).map_err(SearchError::from)?;
157    Ok(buf.freeze())
158}
159
160struct CollectedOperations {
161    /// Each resource's operation, pre-serialized to the exact bytes that will
162    /// be sent to Elasticsearch. `Bytes::len()` is this batch's byte-size
163    /// accounting; the same bytes are the `_bulk` request body, so a
164    /// document is JSON-encoded exactly once.
165    bulk_lines: Vec<Bytes>,
166    /// Parallel to `bulk_lines` (same order) - the resource each line was
167    /// built from, kept so a failed Elasticsearch bulk item can be attributed
168    /// back to the resource that produced it.
169    sent_resources: Vec<IndexResource>,
170    failed: Vec<IndexFailure>,
171}
172
173/// Groups indices `0..sizes.len()` into batches whose cumulative `sizes` stay
174/// within `target_bytes`, returning each batch's length
175fn batch_lengths_by_size(sizes: &[usize], target_bytes: usize) -> Vec<usize> {
176    let mut batch_lengths = Vec::new();
177    let mut current_len = 0usize;
178    let mut current_size = 0usize;
179
180    for &size in sizes {
181        if current_len > 0 && current_size + size > target_bytes {
182            batch_lengths.push(current_len);
183            current_len = 0;
184            current_size = 0;
185        }
186
187        current_len += 1;
188        current_size += size;
189    }
190
191    if current_len > 0 {
192        batch_lengths.push(current_len);
193    }
194
195    batch_lengths
196}
197
198/// Splits collected bulk operation lines (and their matching resources) into
199/// batches sized by `batch_lengths_by_size`, so a `_bulk` request's actual
200/// document count varies with how large those documents are.
201fn batch_by_byte_size(
202    bulk_lines: Vec<Bytes>,
203    sent_resources: Vec<IndexResource>,
204    target_bytes: usize,
205) -> Vec<(Vec<Bytes>, Vec<IndexResource>)> {
206    let sizes: Vec<usize> = bulk_lines.iter().map(Bytes::len).collect();
207
208    let mut lines_iter = bulk_lines.into_iter();
209    let mut resources_iter = sent_resources.into_iter();
210
211    batch_lengths_by_size(&sizes, target_bytes)
212        .into_iter()
213        .map(|len| {
214            (
215                (&mut lines_iter).take(len).collect(),
216                (&mut resources_iter).take(len).collect(),
217            )
218        })
219        .collect()
220}
221
222/// See <https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-bulk>
223#[derive(Deserialize, Debug)]
224struct BulkResponse {
225    /// The length of time, in milliseconds, it took to process the bulk request.
226    #[allow(dead_code)]
227    took: u64,
228    /// The result of each operation in the bulk request, in the order they were submitted.
229    items: Vec<serde_json::Value>,
230    /// If true, one or more of the operations in the bulk request did not complete successfully.
231    #[allow(dead_code)]
232    errors: bool,
233}
234
235/// Processes the Elasticsearch bulk response, attributing each per-item result
236/// back to the resource that produced it. Elasticsearch indexes bulk items
237/// independently, so one bad document does not prevent the rest of the batch
238/// from succeeding - only a request-level failure (unreachable cluster, malformed response) surfaces as `Err` here.
239fn process_bulk_response(
240    bulk_response: &BulkResponse,
241    sent_resources: Vec<IndexResource>,
242) -> Result<IndexOutcome, OperationOutcomeError> {
243    if bulk_response.items.len() != sent_resources.len() {
244        return Err(OperationOutcomeError::fatal(
245            IssueType::exception(),
246            format!(
247                "Elasticsearch bulk response item count '{}' did not match request count '{}'.",
248                bulk_response.items.len(),
249                sent_resources.len()
250            ),
251        ));
252    }
253
254    let mut succeeded = 0;
255    let mut failed = Vec::new();
256
257    // Per documentation: The result of each operation in the bulk request, in the order they were submitted.
258    // See https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-bulk.
259
260    for (item, resource) in bulk_response.items.iter().zip(sent_resources) {
261        // Each item is a single-key object, e.g. `{"index": {...}}` or `{"delete": {...}}`.
262        if let Some(op_result) = item.as_object().and_then(|o| o.values().next()) {
263            match op_result["status"].as_u64() {
264                Some(status) if (200..300).contains(&status) => succeeded += 1,
265                status => {
266                    let reason = op_result["error"]["reason"]
267                        .as_str()
268                        .unwrap_or("unknown error");
269                    failed.push(IndexFailure {
270                        error: OperationOutcomeError::fatal(
271                            IssueType::exception(),
272                            format!("Elasticsearch indexing failed (status {status:?}): {reason}"),
273                        ),
274                        resource,
275                    });
276                }
277            }
278        } else {
279            failed.push(IndexFailure {
280                error: OperationOutcomeError::fatal(
281                    IssueType::exception(),
282                    format!("Unexpected Elasticsearch bulk item shape: '{item}'"),
283                ),
284                resource,
285            });
286        }
287    }
288
289    if !failed.is_empty() {
290        tracing::error!(
291            "Elasticsearch bulk index reported {} failed item(s) out of {}.",
292            failed.len(),
293            bulk_response.items.len()
294        );
295    }
296
297    Ok(IndexOutcome { succeeded, failed })
298}
299
300impl<SearchParameterResolver: SearchParameterResolve + 'static>
301    ElasticSearchEngine<SearchParameterResolver>
302{
303    pub fn new(
304        parameter_resolver: Arc<SearchParameterResolver>,
305        fp_engine: Arc<FPEngine>,
306        es_client: Arc<Elasticsearch>,
307        prune_removed_search_parameters: bool,
308    ) -> Self {
309        ElasticSearchEngine {
310            parameter_resolver,
311            fp_engine,
312            client: es_client,
313            prune_removed_search_parameters,
314        }
315    }
316
317    /// Checks whether the Elasticsearch client is connected.
318    ///
319    /// # Errors
320    ///
321    /// Returns [`SearchError::NotConnected`] if Elasticsearch responds with a
322    /// non-success status code.
323    ///
324    /// Returns a [`SearchError`] if the ping request fails.
325    pub async fn is_connected(&self) -> Result<(), SearchError> {
326        let res = self.client.ping().send().await.map_err(SearchError::from)?;
327
328        if res.status_code().is_success() {
329            Ok(())
330        } else {
331            Err(SearchError::NotConnected)
332        }
333    }
334
335    /// Sends one Elasticsearch `_bulk` request and attributes each per-item
336    /// result back to the resource that produced it. Elasticsearch indexes
337    /// bulk items independently, so one bad document does not prevent the
338    /// rest of the batch from succeeding - only a request-level failure
339    /// (unreachable cluster, malformed response) surfaces as `Err` here.
340    async fn send_bulk_operations(
341        &self,
342        search_index_name: &'static str,
343        bulk_lines: Vec<Bytes>,
344        sent_resources: Vec<IndexResource>,
345    ) -> Result<IndexOutcome, OperationOutcomeError> {
346        if bulk_lines.is_empty() {
347            return Ok(IndexOutcome {
348                succeeded: 0,
349                failed: Vec::new(),
350            });
351        }
352
353        let res = self
354            .client
355            .bulk(BulkParts::Index(search_index_name))
356            .body(bulk_lines)
357            .send()
358            .await
359            .map_err(SearchError::from)?;
360
361        let response_body = res.json::<BulkResponse>().await.map_err(|_e| {
362            OperationOutcomeError::fatal(
363                IssueType::exception(),
364                "Failed to parse response body.".to_string(),
365            )
366        })?;
367
368        process_bulk_response(&response_body, sent_resources)
369    }
370
371    fn spawn_index_tasks(
372        &self,
373        resources: Vec<IndexResource>,
374        search_index_name: &'static str,
375    ) -> Vec<Tasks> {
376        resources
377            .into_iter()
378            .map(|r| {
379                let engine = self.fp_engine.clone();
380                let parameter_resolver = self.parameter_resolver.clone();
381
382                tokio::spawn(async move {
383                    Self::build_bulk_operation(engine, parameter_resolver, r, search_index_name)
384                        .await
385                })
386            })
387            .collect()
388    }
389
390    /// Awaits every bulk-op-building task, separating resources that built
391    /// successfully from those that failed. A single resource's failure
392    /// (bad `FHIRPath` expression, unsupported method, etc.) is captured with
393    /// its identity rather than aborting the rest of the batch. A `JoinError`
394    /// (the task itself panicked) has no resource to attribute the failure
395    /// to, so it aborts the whole call.
396    async fn collect_bulk_operations(
397        &self,
398        tasks: Vec<Tasks>,
399    ) -> Result<CollectedOperations, OperationOutcomeError> {
400        tracing::trace!("Awaiting {} indexing tasks.", tasks.len());
401
402        let mut bulk_lines = Vec::with_capacity(tasks.len());
403        let mut sent_resources = Vec::with_capacity(tasks.len());
404        let mut failed = Vec::new();
405
406        for task in tasks {
407            let (resource, result) = task
408                .await
409                .map_err(|e| OperationOutcomeError::fatal(IssueType::exception(), e.to_string()))?;
410
411            match result {
412                Ok(bytes) => {
413                    sent_resources.push(resource);
414                    bulk_lines.push(bytes);
415                }
416                Err(error) => failed.push(IndexFailure { resource, error }),
417            }
418        }
419
420        Ok(CollectedOperations {
421            bulk_lines,
422            sent_resources,
423            failed,
424        })
425    }
426
427    async fn build_bulk_operation<ParameterResolver: SearchParameterResolve>(
428        engine: Arc<FPEngine>,
429        parameter_resolver: Arc<ParameterResolver>,
430        resource: IndexResource,
431        search_index_name: &'static str,
432    ) -> (IndexResource, Result<Bytes, OperationOutcomeError>) {
433        let result = match &resource.fhir_method {
434            FHIRMethod::Create | FHIRMethod::Update => {
435                Self::build_index_operation(
436                    engine,
437                    parameter_resolver,
438                    &resource,
439                    search_index_name,
440                )
441                .await
442            }
443
444            FHIRMethod::Delete => {
445                let index_id = unique_index_id(
446                    &resource.tenant,
447                    &resource.project,
448                    &resource.resource_type,
449                    &resource.id,
450                );
451                let op: BulkOperation<HashMap<String, InsertableIndex>> =
452                    BulkOperation::delete(index_id)
453                        .index(search_index_name)
454                        .into();
455
456                serialize_bulk_operation(&op)
457            }
458
459            method @ FHIRMethod::Read => Err(OperationOutcomeError::from(
460                SearchError::UnsupportedFHIRMethod((*method).clone()),
461            )),
462        };
463
464        (resource, result)
465    }
466
467    async fn build_index_operation<ParameterResolver: SearchParameterResolve>(
468        engine: Arc<FPEngine>,
469        parameter_resolver: Arc<ParameterResolver>,
470        resource: &IndexResource,
471        search_index_name: &'static str,
472    ) -> Result<Bytes, OperationOutcomeError> {
473        // Id is not sufficient because different Resourcetypes may have the same id.
474        // Additionally should be namespaced by tenant and project to avoid conflicts across tenants and projects.
475        let index_id = unique_index_id(
476            &resource.tenant,
477            &resource.project,
478            &resource.resource_type,
479            &resource.id,
480        );
481
482        let params = parameter_resolver
483            .by_resource_type(&resource.tenant, &resource.project, &resource.resource_type)
484            .await?;
485
486        let mut elastic_index =
487            resource_to_elastic_index(engine, &params, &resource.resource).await?;
488
489        Self::add_index_metadata(&mut elastic_index, resource);
490
491        let op = BulkOperation::index(elastic_index)
492            .id(index_id)
493            .index(search_index_name)
494            .into();
495
496        serialize_bulk_operation(&op)
497    }
498
499    fn add_index_metadata(
500        elastic_index: &mut HashMap<String, InsertableIndex>,
501        resource: &IndexResource,
502    ) {
503        elastic_index.insert(
504            "resource_type".to_string(),
505            InsertableIndex::Meta(resource.resource_type.as_ref().to_string()),
506        );
507
508        elastic_index.insert(
509            "id".to_string(),
510            InsertableIndex::Meta(resource.id.as_ref().to_string()),
511        );
512
513        elastic_index.insert(
514            "version_id".to_string(),
515            InsertableIndex::Meta(resource.version_id.as_ref().to_string()),
516        );
517
518        elastic_index.insert(
519            "project".to_string(),
520            InsertableIndex::Meta(resource.project.as_ref().to_string()),
521        );
522
523        elastic_index.insert(
524            "tenant".to_string(),
525            InsertableIndex::Meta(resource.tenant.as_ref().to_string()),
526        );
527    }
528}
529
530/// Whether `type_` gets a dedicated field in [`migration::create_elasticsearch_searchparameter_mappings`]
531/// (used for every `System`-level parameter's mapping). `composite`/`special`/anything
532/// else has no mapping entry there, so a `System`-level parameter of one of
533/// those types must never be written -- under `dynamic: "strict"`, writing an
534/// unmapped field name fails the whole bulk item.
535pub(crate) fn is_mapped_search_parameter_type(type_: &BoundCode<SearchParamType>) -> bool {
536    type_ == &SearchParamType::number()
537        || type_ == &SearchParamType::string()
538        || type_ == &SearchParamType::uri()
539        || type_ == &SearchParamType::token()
540        || type_ == &SearchParamType::date()
541        || type_ == &SearchParamType::reference()
542        || type_ == &SearchParamType::quantity()
543}
544
545async fn resource_to_elastic_index(
546    fp_engine: Arc<FPEngine>,
547    parameters: &[ResolvedParameter],
548    resource: &Resource,
549) -> Result<HashMap<String, InsertableIndex>, OperationOutcomeError> {
550    let mut map = HashMap::new();
551    let mut dynamic_parameters = Vec::new();
552    for param in parameters {
553        if let Some(expression) = param
554            .search_parameter
555            .expression
556            .as_ref()
557            .and_then(|e| e.value.as_ref())
558            && let Some(url) = param.search_parameter.url.value.as_ref()
559        {
560            // A `System`-level parameter of an unmapped type (composite,
561            // special, ...) has nowhere to be written -- skip it rather than
562            // evaluating FHIRPath for a value that would just get dropped
563            // (or, under strict mapping, reject the whole document).
564            if matches!(param.level, ParameterLevel::System)
565                && !is_mapped_search_parameter_type(&param.search_parameter.type_)
566            {
567                continue;
568            }
569
570            let result = fp_engine
571                .evaluate(expression, vec![resource])
572                .await
573                .map_err(SearchError::from);
574
575            if let Err(err) = result {
576                tracing::error!(
577                    "Failed to evaluate FHIRPath expression: '{}' for resource.",
578                    expression,
579                );
580
581                return Err(err.into());
582            }
583
584            let result_vec = indexing_conversion::to_insertable_index(
585                param,
586                &result?.iter().collect::<Vec<_>>(),
587            )?;
588
589            match param.level {
590                ParameterLevel::System => {
591                    map.insert(flatten_parameter_field_name(url), result_vec);
592                }
593                // Project (user-submitted) parameters are indexed under a single
594                // nested field with the URL stored as a plain value, since an
595                // arbitrary user-supplied URL can't safely become an
596                // Elasticsearch field name.
597                ParameterLevel::Project => {
598                    let type_ = param.search_parameter.type_.as_str().unwrap_or("string");
599                    if let Some(entry) =
600                        DynamicParameterEntry::from_leaf(url.clone(), type_, result_vec)
601                    {
602                        dynamic_parameters.push(entry);
603                    }
604                }
605            }
606        }
607    }
608
609    // Various project level parameters. These are indexed under a single field in elasticsearch as a nested type with url and indexed value..
610    map.insert(
611        DYNAMIC_PARAMETER_INDEX_FIELD.to_string(),
612        InsertableIndex::DynamicParameters(dynamic_parameters),
613    );
614
615    Ok(map)
616}
617
618#[allow(dead_code)]
619static R4_FHIR_INDEX_V1: &str = "r4_search_index";
620static R4_FHIR_INDEX_V2: &str = "r4_search_index_v2";
621
622#[must_use]
623pub const fn get_index_name() -> &'static str {
624    R4_FHIR_INDEX_V2
625}
626
627#[derive(serde::Deserialize, Debug)]
628struct ElasticSearchHitResult {
629    _index: String,
630    _id: String,
631    _score: Option<f64>,
632    fields: SearchEntryPrivate,
633}
634
635#[derive(serde::Deserialize, Debug)]
636struct ElasticSearchHitTotalMeta {
637    value: i64,
638    // relation: String,
639}
640
641#[derive(serde::Deserialize, Debug)]
642struct ElasticSearchHit {
643    total: Option<ElasticSearchHitTotalMeta>,
644    hits: Vec<ElasticSearchHitResult>,
645}
646
647#[derive(serde::Deserialize, Debug)]
648struct ElasticSearchResponse {
649    hits: ElasticSearchHit,
650}
651
652fn unique_index_id(
653    tenant: &TenantId,
654    project: &ProjectId,
655    resource_type: &ResourceType,
656    id: &ResourceId,
657) -> String {
658    let unique_index_id = format!(
659        "{}/{}/{}/{}",
660        tenant.as_ref(),
661        project.as_ref(),
662        resource_type.as_ref(),
663        id.as_ref()
664    );
665
666    unique_index_id
667}
668
669impl<SearchParameterResolver: SearchParameterResolve> SearchEngine
670    for ElasticSearchEngine<SearchParameterResolver>
671{
672    async fn search(
673        &self,
674        _fhir_version: &SupportedFHIRVersions,
675        tenant: &TenantId,
676        project: &ProjectId,
677        search_request: &SearchRequest,
678        options: Option<SearchOptions>,
679    ) -> Result<SearchReturn, haste_fhir_operation_error::OperationOutcomeError> {
680        search::execute_search(
681            self.client.clone(),
682            self.parameter_resolver.clone(),
683            tenant,
684            project,
685            search_request,
686            options.as_ref(),
687        )
688        .await
689    }
690
691    async fn index(
692        &self,
693        _fhir_version: SupportedFHIRVersions,
694        resources: Vec<IndexResource>,
695    ) -> Result<IndexOutcome, OperationOutcomeError> {
696        let resources_total = resources.len();
697        let search_index_name = get_index_name();
698
699        tracing::trace!(
700            "Indexing {} resources into index: '{}'",
701            resources_total,
702            search_index_name
703        );
704
705        let tasks = self.spawn_index_tasks(resources, search_index_name);
706        let CollectedOperations {
707            bulk_lines,
708            sent_resources,
709            mut failed,
710        } = self.collect_bulk_operations(tasks).await?;
711
712        let built_count = bulk_lines.len();
713        let batches = batch_by_byte_size(bulk_lines, sent_resources, TARGET_BULK_BATCH_BYTES);
714
715        tracing::trace!(
716            "Bulk indexing {} resources into index '{}' across {} byte-sized batch(es)",
717            built_count,
718            search_index_name,
719            batches.len()
720        );
721
722        let mut outcome = IndexOutcome {
723            succeeded: 0,
724            failed: Vec::new(),
725        };
726
727        for (batch_lines, batch_resources) in batches {
728            let batch_outcome = self
729                .send_bulk_operations(search_index_name, batch_lines, batch_resources)
730                .await?;
731            outcome.succeeded += batch_outcome.succeeded;
732            outcome.failed.extend(batch_outcome.failed);
733        }
734
735        outcome.failed.append(&mut failed);
736
737        Ok(outcome)
738    }
739
740    async fn migrate(
741        &self,
742        _fhir_version: &SupportedFHIRVersions,
743    ) -> Result<(), haste_fhir_operation_error::OperationOutcomeError> {
744        migration::create_mapping(
745            self.parameter_resolver.clone(),
746            &self.client,
747            get_index_name(),
748            self.prune_removed_search_parameters,
749        )
750        .await?;
751        Ok(())
752    }
753}
754
755#[cfg(test)]
756mod tests {
757    use super::*;
758
759    #[test]
760    fn flatten_parameter_field_name_strips_dots_only() {
761        assert_eq!(
762            flatten_parameter_field_name("http://hl7.org/fhir/SearchParameter/Patient-name"),
763            "http://hl7_org/fhir/SearchParameter/Patient-name"
764        );
765    }
766
767    #[test]
768    fn flatten_parameter_field_name_handles_multiple_dots() {
769        assert_eq!(
770            flatten_parameter_field_name("https://sub.acme.io/v1.2/x"),
771            "https://sub_acme_io/v1_2/x"
772        );
773    }
774
775    #[test]
776    fn batch_lengths_by_size_empty() {
777        assert_eq!(batch_lengths_by_size(&[], 100), Vec::<usize>::new());
778    }
779
780    #[test]
781    fn batch_lengths_by_size_all_fit_in_one_batch() {
782        assert_eq!(batch_lengths_by_size(&[10, 20, 30], 100), vec![3]);
783    }
784
785    #[test]
786    fn batch_lengths_by_size_splits_when_target_exceeded() {
787        // 40 + 40 = 80 fits; +40 would be 120 > 100, so it starts a new batch.
788        assert_eq!(batch_lengths_by_size(&[40, 40, 40], 100), vec![2, 1]);
789    }
790
791    #[test]
792    fn batch_lengths_by_size_oversized_single_item_gets_its_own_batch() {
793        // A single item larger than the target must still be sent, alone,
794        // rather than blocking forever waiting to "fit".
795        assert_eq!(batch_lengths_by_size(&[5, 500, 5], 100), vec![1, 1, 1]);
796    }
797
798    #[test]
799    fn batch_lengths_by_size_exact_fit_does_not_split_early() {
800        assert_eq!(batch_lengths_by_size(&[50, 50], 100), vec![2]);
801    }
802
803    #[test]
804    fn serialize_bulk_operation_index_produces_header_and_source_lines() {
805        let mut doc = HashMap::new();
806        doc.insert(
807            "resource_type".to_string(),
808            InsertableIndex::Meta("Patient".to_string()),
809        );
810
811        let op: BulkOperation<HashMap<String, InsertableIndex>> = BulkOperation::index(doc)
812            .id("t/p/Patient/1")
813            .index("r4_search_index_v2")
814            .into();
815
816        let bytes = serialize_bulk_operation(&op).unwrap();
817        let text = std::str::from_utf8(&bytes).unwrap();
818        let lines: Vec<&str> = text.lines().collect();
819
820        // Header line + source line, each newline-terminated -- this is what
821        // `batch_by_byte_size`/`send_bulk_operations` measure and send
822        // directly, with no re-serialization.
823        assert_eq!(lines.len(), 2);
824        assert!(lines[0].contains(r#""index""#));
825        assert!(lines[0].contains(r#""_id":"t/p/Patient/1""#));
826        assert!(text.ends_with('\n'));
827    }
828
829    #[test]
830    fn serialize_bulk_operation_delete_produces_only_a_header_line() {
831        let op: BulkOperation<HashMap<String, InsertableIndex>> =
832            BulkOperation::delete("t/p/Patient/1")
833                .index("r4_search_index_v2")
834                .into();
835
836        let bytes = serialize_bulk_operation(&op).unwrap();
837        let text = std::str::from_utf8(&bytes).unwrap();
838        let lines: Vec<&str> = text.lines().collect();
839
840        // No document body for a delete -- just the action line.
841        assert_eq!(lines.len(), 1);
842        assert!(lines[0].contains(r#""delete""#));
843        assert!(text.ends_with('\n'));
844    }
845}
haste_fhir_search/elastic_search/mod.rs

haste_fhir_search/elastic_search/
mod.rs
