fhir_subscription_processor/
lib.rs

1use std::sync::Arc;
2
3use haste_fhir_client::url::{ParsedParameter, ParsedParameters};
4use haste_fhir_model::r4::generated::{
5    resources::{Resource, ResourceType, Subscription},
6    terminology::IssueType,
7};
8use haste_fhir_operation_error::{OperationOutcomeError, derive::OperationOutcomeError};
9use haste_fhir_search::{
10    ResolvedParameter, SearchParameterResolve,
11    indexing_conversion::{self, InsertableIndex},
12};
13use haste_jwt::{ProjectId, TenantId};
14
15pub mod traits;
16
17#[derive(OperationOutcomeError, Debug)]
18pub enum SubscriptionFilterError {
19    #[fatal(
20        code = "exception",
21        diagnostic = "Failed to evaluate fhirpath expression."
22    )]
23    FHIRPathError(#[from] haste_fhirpath::FHIRPathError),
24}
25
26#[allow(dead_code)]
27pub struct SubscriptionParameter {
28    parameter: ResolvedParameter,
29    fp_extract_expression: String,
30    value: Vec<String>,
31    modifier: Option<String>,
32}
33
34pub enum SubscriptionTrigger {
35    // Based around simple Subscription.criteria.
36    QueryFilter {
37        resource_type: ResourceType,
38        parameters: Vec<SubscriptionParameter>,
39    },
40    // This could come from a subscriptiontopic which
41    // allows arbitrary FHIRPath expressions, or from more complex criteria in the future.
42    FHIRPathFilter {
43        expression: String,
44    },
45}
46
47/// In memory representation of a subscription filter.
48/// This is what we will use to evaluate whether a given subscription matches an incoming event.
49#[allow(dead_code)]
50pub struct MemorySubscriptionFilter {
51    fp_engine: haste_fhirpath::FPEngine,
52    triggers: Vec<SubscriptionTrigger>,
53}
54
55impl MemorySubscriptionFilter {
56    pub async fn new<Resolver: SearchParameterResolve>(
57        tenant_id: &TenantId,
58        project_id: &ProjectId,
59        resolver: Arc<Resolver>,
60        value: Subscription,
61    ) -> Result<Self, OperationOutcomeError> {
62        if let Some(criteria) = value.criteria.value {
63            let criteria_pieces = criteria.split('?').collect::<Vec<_>>();
64            let [path, parameters] = criteria_pieces.as_slice() else {
65                return Err(OperationOutcomeError::error(
66                    IssueType::Exception(None),
67                    "Invalid subscription criteria format".to_string(),
68                ));
69            };
70
71            let resource_type = ResourceType::try_from(*path).map_err(|_| {
72                OperationOutcomeError::error(
73                    IssueType::Exception(None),
74                    "Invalid resource type".to_string(),
75                )
76            })?;
77
78            let parsed_parameters = ParsedParameters::try_from(*parameters)?;
79            let mut subscription_parsed_parameters = vec![];
80
81            for parameter in parsed_parameters.owned_parameters().into_iter() {
82                match parameter {
83                    ParsedParameter::Resource(resource_param) => {
84                        let Some(parameter) = resolver
85                            .by_name(
86                                tenant_id,
87                                project_id,
88                                Some(&resource_type),
89                                &resource_param.name,
90                            )
91                            .await?
92                        else {
93                            return Err(OperationOutcomeError::error(
94                                IssueType::Exception(None),
95                                format!(
96                                    "Invalid search parameter in subscription criteria: {}",
97                                    resource_param.name
98                                ),
99                            ));
100                        };
101
102                        if resource_param.chains.is_some() {
103                            return Err(OperationOutcomeError::error(
104                                IssueType::Exception(None),
105                                format!(
106                                    "Chained parameters are not supported in subscription criteria: {}",
107                                    resource_param.name
108                                ),
109                            ));
110                        }
111
112                        let Some(fp_expression) = parameter
113                            .search_parameter
114                            .expression
115                            .as_ref()
116                            .and_then(|expr| expr.value.as_ref())
117                        else {
118                            return Err(OperationOutcomeError::error(
119                                IssueType::Exception(None),
120                                format!(
121                                    "Search parameter does not have an expression: {}",
122                                    resource_param.name
123                                ),
124                            ));
125                        };
126
127                        subscription_parsed_parameters.push(SubscriptionParameter {
128                            parameter: parameter.clone(),
129                            fp_extract_expression: fp_expression.clone(),
130                            value: resource_param.value,
131                            modifier: resource_param.modifier,
132                        });
133                    }
134                    ParsedParameter::Result(result_param) => {
135                        return Err(OperationOutcomeError::error(
136                            IssueType::Exception(None),
137                            format!(
138                                "Unsupported parameter in subscription criteria: {}",
139                                result_param.name
140                            ),
141                        ));
142                    }
143                }
144            }
145
146            Ok(MemorySubscriptionFilter {
147                fp_engine: haste_fhirpath::FPEngine::new(),
148                triggers: vec![SubscriptionTrigger::QueryFilter {
149                    resource_type,
150                    parameters: subscription_parsed_parameters,
151                }],
152            })
153        } else {
154            Err(OperationOutcomeError::error(
155                IssueType::Exception(None),
156                "SubscriptionFilter conversion not implemented".to_string(),
157            ))
158        }
159    }
160}
161
162async fn fits_subscription_parameter(
163    fp_engine: &haste_fhirpath::FPEngine,
164    subscription_parameter: &SubscriptionParameter,
165    resource: &Resource,
166) -> Result<bool, OperationOutcomeError> {
167    let result = fp_engine
168        .evaluate(
169            &subscription_parameter.fp_extract_expression,
170            vec![resource],
171        )
172        .await
173        .map_err(SubscriptionFilterError::from)?;
174
175    let conversions = indexing_conversion::to_insertable_index(
176        &subscription_parameter.parameter,
177        result.iter().collect::<Vec<_>>(),
178    )?;
179
180    match conversions {
181        InsertableIndex::String(resource_values) => {
182            Ok(resource_values.iter().any(|resource_value| {
183                subscription_parameter
184                    .value
185                    .iter()
186                    .any(|v| resource_value.to_lowercase().starts_with(&v.to_lowercase()))
187            }))
188        }
189        InsertableIndex::Number(_) => Err(OperationOutcomeError::error(
190            IssueType::Exception(None),
191            "Number search parameters are not supported in subscription criteria".to_string(),
192        ))?,
193        InsertableIndex::URI(_) => Err(OperationOutcomeError::error(
194            IssueType::Exception(None),
195            "URI search parameters are not supported in subscription criteria".to_string(),
196        ))?,
197        InsertableIndex::Token(_) => Err(OperationOutcomeError::error(
198            IssueType::Exception(None),
199            "Token search parameters are not supported in subscription criteria".to_string(),
200        ))?,
201        InsertableIndex::Date(_) => Err(OperationOutcomeError::error(
202            IssueType::Exception(None),
203            "Date search parameters are not supported in subscription criteria".to_string(),
204        ))?,
205
206        InsertableIndex::Reference(_) => Err(OperationOutcomeError::error(
207            IssueType::Exception(None),
208            "Reference search parameters are not supported in subscription criteria".to_string(),
209        ))?,
210        InsertableIndex::Quantity(_) => Err(OperationOutcomeError::error(
211            IssueType::Exception(None),
212            "Quantity search parameters are not supported in subscription criteria".to_string(),
213        ))?,
214        InsertableIndex::DynamicParameters(_) => Err(OperationOutcomeError::error(
215            IssueType::Exception(None),
216            "Dynamic search parameters are not supported in subscription criteria".to_string(),
217        ))?,
218
219        InsertableIndex::Composite(_) => Err(OperationOutcomeError::error(
220            IssueType::Exception(None),
221            "Composite search parameters are not supported in subscription criteria".to_string(),
222        ))?,
223        InsertableIndex::Special(_) => Err(OperationOutcomeError::error(
224            IssueType::Exception(None),
225            "Special search parameters are not supported in subscription criteria".to_string(),
226        ))?,
227        InsertableIndex::Meta(_) => Err(OperationOutcomeError::error(
228            IssueType::Exception(None),
229            "Meta search parameters are not supported in subscription criteria".to_string(),
230        ))?,
231    }
232}
233
234impl traits::SubscriptionFilter for MemorySubscriptionFilter {
235    async fn matches(&self, resource: &Resource) -> Result<bool, OperationOutcomeError> {
236        let resource_resource_type = resource.resource_type();
237
238        for trigger in self.triggers.iter() {
239            match trigger {
240                SubscriptionTrigger::QueryFilter {
241                    resource_type,
242                    parameters,
243                } => {
244                    if *resource_type != resource_resource_type {
245                        return Ok(false);
246                    }
247
248                    for sub_parameter in parameters {
249                        let fits_criteria =
250                            fits_subscription_parameter(&self.fp_engine, sub_parameter, resource)
251                                .await?;
252                        if !fits_criteria {
253                            return Ok(false);
254                        }
255                    }
256
257                    return Ok(true);
258                }
259                SubscriptionTrigger::FHIRPathFilter { .. } => {
260                    return Err(OperationOutcomeError::error(
261                        IssueType::Exception(None),
262                        "FHIRPathFilter triggers are not yet supported".to_string(),
263                    ))?;
264                }
265            }
266        }
267
268        Ok(false)
269    }
270}
271
272#[cfg(test)]
273mod tests {
274    use haste_fhir_model::r4::generated::{
275        resources::Patient,
276        types::{FHIRString, HumanName},
277    };
278    use haste_fhir_search::memory::R4_SEARCH_PARAMETERS_INDEX;
279
280    use crate::traits::SubscriptionFilter;
281
282    use super::*;
283
284    #[tokio::test]
285    async fn quick_test_derive() {
286        let subscription = Subscription {
287            criteria: Box::new(FHIRString {
288                value: Some("Patient?name=Smith".to_string()),
289                ..Default::default()
290            }),
291            ..Default::default()
292        };
293
294        let resolver = R4_SEARCH_PARAMETERS_INDEX.clone();
295
296        let sub_filter = MemorySubscriptionFilter::new(
297            &TenantId::System,
298            &ProjectId::System,
299            resolver,
300            subscription,
301        )
302        .await
303        .unwrap();
304
305        assert_eq!(sub_filter.triggers.len(), 1);
306
307        match &sub_filter.triggers[0] {
308            SubscriptionTrigger::QueryFilter {
309                resource_type,
310                parameters,
311            } => {
312                assert_eq!(resource_type, &ResourceType::Patient);
313                assert_eq!(parameters[0].fp_extract_expression, "Patient.name");
314                assert_eq!(parameters[0].value, vec!["Smith".to_string()]);
315            }
316            _ => panic!("Expected QueryFilter trigger"),
317        };
318    }
319
320    #[tokio::test]
321    async fn modifier_check() {
322        let subscription = Subscription {
323            criteria: Box::new(FHIRString {
324                value: Some("Observation?category:missing=true".to_string()),
325                ..Default::default()
326            }),
327            ..Default::default()
328        };
329
330        let resolver = R4_SEARCH_PARAMETERS_INDEX.clone();
331        let sub_filter = MemorySubscriptionFilter::new(
332            &TenantId::System,
333            &ProjectId::System,
334            resolver,
335            subscription,
336        )
337        .await
338        .unwrap();
339
340        assert_eq!(sub_filter.triggers.len(), 1);
341
342        match &sub_filter.triggers[0] {
343            SubscriptionTrigger::QueryFilter {
344                resource_type,
345                parameters,
346            } => {
347                assert_eq!(resource_type, &ResourceType::Observation);
348                assert_eq!(parameters[0].fp_extract_expression, "Observation.category");
349                assert_eq!(parameters[0].value, vec!["true".to_string()]);
350                assert_eq!(parameters[0].modifier, Some("missing".to_string()));
351            }
352            _ => panic!("Expected QueryFilter trigger"),
353        };
354    }
355
356    #[tokio::test]
357    async fn test_run_fhirpath() {
358        let resolver = R4_SEARCH_PARAMETERS_INDEX.clone();
359        let sub_filter = MemorySubscriptionFilter::new(
360            &TenantId::System,
361            &ProjectId::System,
362            resolver.clone(),
363            Subscription {
364                criteria: Box::new(FHIRString {
365                    value: Some("Patient?name=Smith".to_string()),
366                    ..Default::default()
367                }),
368                ..Default::default()
369            },
370        )
371        .await
372        .unwrap();
373        let patient = Resource::Patient(Patient {
374            name: Some(vec![Box::new(HumanName {
375                family: Some(Box::new(FHIRString {
376                    value: Some("Smith".to_string()),
377                    ..Default::default()
378                })),
379                ..Default::default()
380            })]),
381            ..Default::default()
382        });
383
384        assert_eq!(sub_filter.matches(&patient).await.unwrap(), true);
385
386        let sub_filter_partial = MemorySubscriptionFilter::new(
387            &TenantId::System,
388            &ProjectId::System,
389            resolver.clone(),
390            Subscription {
391                criteria: Box::new(FHIRString {
392                    value: Some("Patient?name=Sm".to_string()),
393                    ..Default::default()
394                }),
395                ..Default::default()
396            },
397        )
398        .await
399        .unwrap();
400
401        assert_eq!(sub_filter_partial.matches(&patient).await.unwrap(), true);
402
403        let sub_filter_casing = MemorySubscriptionFilter::new(
404            &TenantId::System,
405            &ProjectId::System,
406            resolver.clone(),
407            Subscription {
408                criteria: Box::new(FHIRString {
409                    value: Some("Patient?name=sm".to_string()),
410                    ..Default::default()
411                }),
412                ..Default::default()
413            },
414        )
415        .await
416        .unwrap();
417
418        assert_eq!(sub_filter_casing.matches(&patient).await.unwrap(), true);
419
420        let patient = Resource::Patient(Patient {
421            name: Some(vec![Box::new(HumanName {
422                family: Some(Box::new(FHIRString {
423                    value: Some("NotSmith".to_string()),
424                    ..Default::default()
425                })),
426                ..Default::default()
427            })]),
428            ..Default::default()
429        });
430
431        assert_eq!(sub_filter.matches(&patient).await.unwrap(), false);
432    }
433}
fhir_subscription_processor/lib.rs

fhir_subscription_processor/
lib.rs
