pub mod aql_serialization;
mod form_struct;
pub mod structs;
use crate::aql_serialization::ToAql;
use crate::form_struct::{ChildElement, FormResponse};
use crate::structs::{
AQLQuery, AQLResponse, Composition, CompositionPreview, EhrDbEventTrigger, EhrDbVersion,
EhrDbView, Tag, TagsPayload,
};
use anyhow::anyhow;
use base64::Engine;
use base64::engine::general_purpose;
use chrono::Utc;
use chrono::{DateTime, Duration};
use regex::Regex;
use reqwest::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, HeaderMap, HeaderValue};
use reqwest::{Certificate, Client, StatusCode, Url};
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use serde_json::Value;
use std::error::Error;
use std::time::Duration as StdDuration;
use tracing::debug;
use uuid::Uuid;
pub struct OpenEhrClient {
url: String,
client: Client,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct UpdateContribution<'a> {
action: &'static str,
template_id: &'a str,
ehr_id: &'a str,
format: &'static str,
composition_uid: &'a str,
lifecycle_state: &'static str,
composition: &'a Value,
#[serde(skip_serializing_if = "Option::is_none")]
tags: Option<&'a Value>,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct ContributionResponse {
#[serde(default)]
commit_data: Vec<ContributionEntry>,
}
#[derive(Deserialize)]
struct ContributionEntry {
id: Option<String>,
action: Option<String>,
}
impl OpenEhrClient {
pub fn new(url: String, username: String, password: String) -> OpenEhrClient {
Self::try_new(url, username, password).expect("failed to configure EHRDB client")
}
pub fn try_new(
url: String,
username: String,
password: String,
) -> Result<OpenEhrClient, anyhow::Error> {
Self::build(url, username, password, Vec::new(), true)
}
pub fn try_new_with_ca_pem_bundle(
url: String,
username: String,
password: String,
ca_pem: &[u8],
verify_hostname: bool,
) -> Result<OpenEhrClient, anyhow::Error> {
let certificates = Certificate::from_pem_bundle(ca_pem)?;
Self::build(url, username, password, certificates, verify_hostname)
}
fn build(
url: String,
username: String,
password: String,
certificates: Vec<Certificate>,
verify_hostname: bool,
) -> Result<OpenEhrClient, anyhow::Error> {
let mut headers = HeaderMap::new();
headers.insert("wait-for-commit", HeaderValue::from_static("true"));
headers.insert("hack-time", HeaderValue::from_static("true"));
headers.insert(ACCEPT, HeaderValue::from_static("application/json"));
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
let encoded_auth =
general_purpose::STANDARD.encode(format!("{}:{}", username, password).as_str());
let auth = format!("Basic {}", encoded_auth);
let mut authorization = HeaderValue::from_str(&auth)?;
authorization.set_sensitive(true);
headers.insert(AUTHORIZATION, authorization);
let mut builder = Client::builder()
.connect_timeout(StdDuration::from_secs(10))
.timeout(StdDuration::from_secs(30))
.default_headers(headers);
for certificate in certificates {
builder = builder.add_root_certificate(certificate);
}
if !verify_hostname {
builder = builder.danger_accept_invalid_hostnames(true);
}
let client = builder.build()?;
Ok(OpenEhrClient {
url: normalize_base_url(&url)?.trim_end_matches('/').to_string(),
client,
})
}
pub async fn get_form(&self, name: &str) -> Result<String, Box<dyn Error>> {
let resp = self
.client
.get(format!("{}/form/{}", self.url, name))
.send()
.await?
.json::<FormResponse>()
.await?;
let mut all_child_elements = Vec::new();
if let Some(resource) = resp
.form
.resources
.iter()
.find(|&r| r.name == "edit-form-description")
&& let Some(content) = &resource.content
{
collect_all_children(&content.children, &mut all_child_elements);
}
for child in all_child_elements {
if let Some(view_config) = child.view_config
&& let Some(advanced) = view_config.advanced
&& !advanced.hidden
{
debug!(
"Child element name: {} {:?}",
child.name.as_deref().unwrap_or("<unnamed>"),
child.fid
);
if let Some(field) = view_config.field {
debug!("Field information: {}", field);
}
}
}
debug!("Form processed successfully.");
Ok("Form processed successfully.".to_string())
}
pub async fn get_composition_tags(
&self,
composition_uid: &String,
) -> Result<String, Box<dyn Error>> {
let resp = self
.client
.get(format!("{}/tagging/{}", self.url, composition_uid))
.send()
.await?
.text()
.await?;
Ok(resp)
}
pub async fn execute_aql_query<T: DeserializeOwned>(
&self,
aql: &str,
params: &[&(dyn ToAql + Sync)],
) -> Result<Vec<T>, Box<dyn Error + Send + Sync>> {
let aql_with_params = params
.iter()
.enumerate()
.fold(aql.to_string(), |acc, (i, param)| {
acc.replace(&format!("${}", i + 1), ¶m.to_aql_string())
});
self.execute_aql_query_with_parameters(
&aql_with_params,
&Value::Object(serde_json::Map::new()),
)
.await
}
pub async fn execute_aql_query_with_parameters<T: DeserializeOwned>(
&self,
aql: &str,
aql_parameters: &Value,
) -> Result<Vec<T>, Box<dyn Error + Send + Sync>> {
if !aql_parameters.is_object() {
return Err(anyhow!("AQL parameters must be a JSON object").into());
}
let post_data = AQLQuery {
aql,
aql_parameters: aql_parameters.clone(),
};
let response = self
.client
.post(format!("{}/query", self.url))
.json(&post_data)
.send()
.await?;
let status = response.status();
if status == StatusCode::NO_CONTENT {
return Ok(vec![]);
}
if !status.is_success() {
return Err(anyhow!("EHRDB AQL request failed with HTTP {}", status).into());
}
let body = response.text().await?;
if body.is_empty() {
return Ok(vec![]);
}
let aql_response: AQLResponse<T> = serde_json::from_str(&body)?;
Ok(aql_response.result_set)
}
pub async fn execute_aql_query_by_days<T: DeserializeOwned>(
&self,
aql: &str,
params: &[&(dyn ToAql + Sync)],
dt_beg: &DateTime<Utc>,
dt_end: &DateTime<Utc>,
) -> Result<Vec<T>, Box<dyn Error + Send + Sync>> {
let mut accumulated_results = Vec::new();
let dt_beg_p = *dt_beg;
let dt_end_p = *dt_end;
let mut current_date = dt_beg_p;
while current_date <= dt_end_p {
let mut flag = false;
let mut next_date = current_date + Duration::days(1);
if next_date > dt_end_p {
next_date = dt_end_p;
flag = true;
}
let mut daily_params = params.to_vec();
daily_params.push(¤t_date);
daily_params.push(&next_date);
let mut daily_results = self.execute_aql_query(aql, &daily_params).await?;
debug!("Got {} values in chunk", &daily_results.len());
accumulated_results.append(&mut daily_results);
if flag {
break;
}
current_date = next_date;
}
Ok(accumulated_results)
}
pub async fn execute_aql_query_by_vec<T, U>(
&self,
aql: &str,
params: &[&(dyn ToAql + Sync)],
values: &[U],
chunk_size: usize,
) -> Result<Vec<T>, Box<dyn Error + Send + Sync>>
where
T: DeserializeOwned,
U: ToAql + Sync + Clone,
{
let mut accumulated_results = Vec::new();
for chunk in values.chunks(chunk_size) {
let chunk_vec = chunk.to_vec();
let chunk_param: &(dyn ToAql + Sync) = &chunk_vec;
let mut current_params = params.to_vec();
current_params.push(chunk_param);
let mut chunk_results = self.execute_aql_query(aql, ¤t_params).await?;
debug!("Got {} values in chunk", &chunk_results.len());
accumulated_results.append(&mut chunk_results);
}
Ok(accumulated_results)
}
pub async fn get_version(&self) -> Result<EhrDbVersion, anyhow::Error> {
let resp = self
.client
.get(format!("{}/system/version", self.url))
.send()
.await?;
if resp.status().is_success() {
let version: EhrDbVersion = resp.json().await?;
Ok(version)
} else {
let response_code = resp.status();
Err(anyhow!(
"Failed to get EHRDB version: HTTP {}",
response_code
))
}
}
pub async fn get_trigger(&self, name: &String) -> Result<EhrDbEventTrigger, anyhow::Error> {
let resp = self
.client
.get(format!("{}/trigger/?name={}", self.url, name))
.send()
.await?;
if resp.status().is_success() {
let trigger: EhrDbEventTrigger = resp.json().await?;
Ok(trigger)
} else {
Err(anyhow!("Failed to get trigger {}", name))
}
}
pub async fn has_trigger(&self, name: &String) -> bool {
self.get_trigger(name).await.is_ok()
}
pub async fn create_trigger(&self, trigger: &EhrDbEventTrigger) -> Result<(), anyhow::Error> {
let resp = self
.client
.post(format!("{}/trigger/create", self.url))
.json(&trigger)
.send()
.await?;
if resp.status().is_success() {
Ok(())
} else {
let post_response_code = resp.status();
Err(anyhow!(
"Failed to create trigger {}: HTTP {}",
trigger.name,
post_response_code
))
}
}
pub async fn update_trigger(
&self,
trigger: &EhrDbEventTrigger,
id: &i32,
) -> Result<(), anyhow::Error> {
let resp = self
.client
.put(format!("{}/trigger/update/{}", self.url, id))
.json(&trigger)
.send()
.await?;
if resp.status().is_success() {
Ok(())
} else {
let put_response_code = resp.status();
Err(anyhow!(
"Failed to update trigger {}: HTTP {}",
trigger.name,
put_response_code
))
}
}
pub async fn activate_trigger(&self, id: &i32) -> Result<(), anyhow::Error> {
let resp = self
.client
.put(format!("{}/trigger/{}/status/ACTIVE", self.url, id))
.send()
.await?;
if resp.status().is_success() {
Ok(())
} else {
let put_response_code = resp.status();
Err(anyhow!(
"Failed to activate trigger with id {}: HTTP {}",
id,
put_response_code
))
}
}
pub async fn get_view(&self, name: &String) -> Result<EhrDbView, anyhow::Error> {
let resp = self
.client
.get(format!("{}/view/?name={}", self.url, name))
.send()
.await?;
if resp.status().is_success() {
let view: EhrDbView = resp.json().await?;
Ok(view)
} else {
Err(anyhow!("Failed to get view {}", name))
}
}
pub async fn get_composition(
&self,
composition_id: &String,
) -> Result<Composition, anyhow::Error> {
let resp = self
.client
.get(format!(
"{}/composition/{}?format=FLAT&meta=true",
self.url, composition_id
))
.send()
.await?;
if resp.status().is_success() {
Ok(resp.json::<Composition>().await?)
} else {
let response_code = resp.status();
Err(anyhow!("Failed to get composition: HTTP {}", response_code))
}
}
pub async fn has_view(&self, name: &String) -> bool {
self.get_view(name).await.is_ok()
}
pub async fn create_view(&self, view: EhrDbView) -> Result<(), anyhow::Error> {
let resp = self
.client
.post(format!("{}/view/create", self.url))
.json(&view)
.send()
.await?;
if resp.status().is_success() {
Ok(())
} else {
let post_response_code = resp.status();
Err(anyhow!(
"Failed to create view {}\nStatusCode: {}",
view.name,
post_response_code
))
}
}
pub async fn update_view(&self, view: EhrDbView, id: i32) -> Result<(), anyhow::Error> {
let resp = self
.client
.put(format!("{}/view/update/{}", self.url, id))
.json(&view)
.send()
.await?;
if resp.status().is_success() {
Ok(())
} else {
let post_response_code = resp.status();
Err(anyhow!(
"Failed to update view {}: HTTP {}",
view.name,
post_response_code
))
}
}
pub async fn post_composition(
&self,
composition: &Composition,
ehr_id: &Uuid,
) -> Result<String, anyhow::Error> {
debug!("posting composition");
let flat_content = composition.composition.clone();
let destination = format!(
"{}/composition?ehrId={}&format=FLAT&lifecycleState=complete&templateId={}",
self.url, ehr_id, composition.template_id
);
let resp = self
.client
.post(&destination)
.json(&flat_content)
.send()
.await?;
debug!("posted composition");
let status = resp.status();
let response_text = resp.text().await?;
debug!("returning status");
if status.is_success() {
Ok(format!("Code: {}\n{}", status, response_text.clone()))
} else {
Err(anyhow!("Failed to create composition: HTTP {}", status))
}
}
pub async fn post_tags(
&self,
composition_uid: &str,
tags: Vec<Tag>,
) -> Result<String, anyhow::Error> {
debug!("posting tags");
let tag_payload = TagsPayload {
composition_uid: composition_uid.to_string(),
tags,
};
let flat_content = serde_json::to_value(&tag_payload)?;
let destination = format!("{}/tagging", self.url);
let resp = self
.client
.post(&destination)
.json(&flat_content)
.send()
.await?;
debug!("posted tags");
let status = resp.status();
let response_text = resp.text().await?;
debug!("returning status");
if status.is_success() {
Ok(format!("Code: {}\n{}", status, response_text.clone()))
} else {
Err(anyhow!("Failed to post composition tags: HTTP {}", status))
}
}
pub async fn get_composition_list(
&self,
ehr_case_id: &Uuid,
) -> Result<Vec<CompositionPreview>, anyhow::Error> {
let parameters = serde_json::json!({ "case_id": ehr_case_id });
match self.execute_aql_query_with_parameters::<CompositionPreview>("SELECT c/uid/value AS uid, c/name/value AS name, c/archetype_details/template_id/value AS template_id, c/context/start_time/value AS start_time, c/links/target/value AS link FROM COMPOSITION c WHERE c/context/other_context[at0001]/items[openEHR-EHR-CLUSTER.composition_context_details*]/items[at0035]/value/id = $case_id LIMIT 10000", ¶meters).await {
Ok(result) => Ok(result),
Err(e) => Err(anyhow!("Failed to get composition list: {}", e))
}
}
pub async fn get_first_composition_by_template_id(
&self,
template_id: &String,
ehr_case_id: &Uuid,
) -> Result<Option<Composition>, anyhow::Error> {
let compositions = self.get_composition_list(ehr_case_id).await?;
if let Some(preview) = compositions
.into_iter()
.find(|c| &c.template_id == template_id)
{
Ok(Some(self.get_composition(&preview.uid).await?))
} else {
Ok(None)
}
}
pub async fn transfer_composition(
&self,
composition_uid: &String,
ehr_id: &Uuid,
) -> Result<(), anyhow::Error> {
debug!("Transferring composition");
let destination = format!(
"{}/composition/{}/move?targetEhrId={}",
self.url, composition_uid, ehr_id
);
let resp = self.client.post(&destination).send().await?;
debug!("Transfer request sent");
let status = resp.status();
debug!("Returning transfer status");
if status.is_success() {
Ok(())
} else {
Err(anyhow!("Failed to transfer composition: HTTP {}", status))
}
}
pub async fn update_composition(
&self,
composition: &Composition,
) -> Result<String, anyhow::Error> {
self.update_compositions(std::slice::from_ref(composition))
.await?
.pop()
.ok_or_else(|| anyhow!("EHRDB returned no updated composition UID"))
}
pub async fn update_compositions(
&self,
compositions: &[Composition],
) -> Result<Vec<String>, anyhow::Error> {
if compositions.is_empty() {
return Ok(Vec::new());
}
for composition in compositions {
if composition.deleted {
return Err(anyhow!("Cannot update a deleted EHRDB composition"));
}
if !composition.last_version {
return Err(anyhow!(
"Cannot update a non-latest EHRDB composition version"
));
}
if composition.composition_uid.is_empty()
|| composition.template_id.is_empty()
|| composition.ehr_id.as_deref().is_none_or(str::is_empty)
{
return Err(anyhow!("EHRDB composition metadata is incomplete"));
}
if !composition.composition.is_object() {
return Err(anyhow!("EHRDB FLAT composition must be a JSON object"));
}
}
let updates: Vec<_> = compositions
.iter()
.map(|composition| UpdateContribution {
action: "UPDATE",
template_id: &composition.template_id,
ehr_id: composition.ehr_id.as_deref().expect("validated above"),
format: "FLAT",
composition_uid: &composition.composition_uid,
lifecycle_state: "complete",
composition: &composition.composition,
tags: composition.tags.as_ref(),
})
.collect();
let response = self
.client
.post(format!("{}/composition/contribution", self.url))
.json(&updates)
.send()
.await?;
let status = response.status();
if !status.is_success() {
return Err(anyhow!(
"Failed to update composition contribution: HTTP {}",
status
));
}
let contribution = response.json::<ContributionResponse>().await?;
if contribution.commit_data.len() != compositions.len()
|| contribution
.commit_data
.iter()
.any(|entry| entry.action.as_deref() != Some("UPDATE") || entry.id.is_none())
{
return Err(anyhow!(
"Unexpected EHRDB composition contribution response"
));
}
Ok(contribution
.commit_data
.into_iter()
.filter_map(|entry| entry.id)
.collect())
}
}
fn normalize_base_url(raw: &str) -> Result<String, anyhow::Error> {
let mut url = Url::parse(raw.trim())?;
if !url.username().is_empty()
|| url.password().is_some()
|| url.query().is_some()
|| url.fragment().is_some()
{
return Err(anyhow!(
"EHRDB URL must not contain credentials, query, or fragment"
));
}
if url.path().is_empty() || url.path() == "/" {
url.set_path("/api/rest/v1");
}
Ok(url.to_string())
}
pub fn get_date_time(input: &Option<String>) -> Option<DateTime<Utc>> {
let datetime_regex = Regex::new(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}").unwrap();
match input {
Some(json_str) => {
if let Some(mat) = datetime_regex.find(json_str)
&& let Ok(dt) = DateTime::parse_from_rfc3339(mat.as_str())
{
return Some(dt.with_timezone(&Utc));
}
None
}
None => None,
}
}
pub fn extract_versioned_id(res: &str) -> Result<String, Box<dyn Error + Sync + Send>> {
if let Some(start) = res.find('{') {
let json_part = &res[start..];
let v: Value = serde_json::from_str(json_part)?;
if let Some(composition_uid) = v["compositionUid"].as_str() {
Ok(composition_uid.to_string())
} else {
Err("compositionUid not found".into())
}
} else {
Err("No JSON part found in the input string".into())
}
}
pub fn convert_versioned_id_to_link(versioned_id: &str) -> String {
let parts: Vec<&str> = versioned_id.split("::").collect();
let uuid_part = parts[0];
format!("ehr:compositions/{}", uuid_part)
}
fn collect_all_children(children: &[ChildElement], result: &mut Vec<ChildElement>) {
for child in children {
result.push(child.clone());
if let Some(ref children) = child.children
&& !children.is_empty()
{
collect_all_children(children, result);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn normalizes_host_only_url_to_ehrdb_rest_base() {
assert_eq!(
normalize_base_url("https://ehr.example").unwrap(),
"https://ehr.example/api/rest/v1"
);
assert_eq!(
normalize_base_url("https://ehr.example/api/rest/v1/").unwrap(),
"https://ehr.example/api/rest/v1/"
);
}
#[test]
fn rejects_credentials_in_base_url() {
assert!(normalize_base_url("https://user:secret@ehr.example").is_err());
}
#[test]
fn deserializes_ehrdb_composition_metadata() {
let composition: Composition = serde_json::from_value(serde_json::json!({
"compositionUid": "uid::system::3",
"templateId": "openEHR-EHR-COMPOSITION.test.v1",
"composition": { "flat/path": "value" },
"deleted": false,
"lastVersion": true,
"ehrId": "ehr-id",
"lifecycleState": "COMPLETE",
"tags": [{ "tag": "formname", "value": "test", "aqlPath": null }]
}))
.unwrap();
assert_eq!(composition.composition_uid, "uid::system::3");
assert!(composition.last_version);
assert!(!composition.deleted);
assert_eq!(composition.ehr_id.as_deref(), Some("ehr-id"));
}
#[test]
fn contribution_uses_full_versioned_uid_and_preserves_tags() {
let composition = Composition {
name: String::new(),
template_id: "template".to_string(),
composition_uid: "uid::system::7".to_string(),
composition: serde_json::json!({ "flat/path": "value" }),
tags: Some(serde_json::json!([{ "tag": "sign", "value": "1" }])),
deleted: false,
last_version: true,
ehr_id: Some("ehr-id".to_string()),
lifecycle_state: Some("COMPLETE".to_string()),
};
let request = UpdateContribution {
action: "UPDATE",
template_id: &composition.template_id,
ehr_id: composition.ehr_id.as_deref().unwrap(),
format: "FLAT",
composition_uid: &composition.composition_uid,
lifecycle_state: "complete",
composition: &composition.composition,
tags: composition.tags.as_ref(),
};
let json = serde_json::to_value(request).unwrap();
assert_eq!(json["compositionUid"], "uid::system::7");
assert_eq!(json["ehrId"], "ehr-id");
assert_eq!(json["action"], "UPDATE");
assert_eq!(json["format"], "FLAT");
assert_eq!(json["tags"][0]["tag"], "sign");
}
}