use crate::dialect::SqliteDialect;
use openehr::base::{HierObjectId, ObjectId, ObjectRef, ObjectVersionId};
use openehr::rm::common::{CommitError, Contribution, Version};
use openehr::rm::data_types::DvDateTime;
use openehr::rm::ehr::{Composition, Ehr};
use openehr::validation::Validate as _;
use openehr_store::record::{CompositionIndexRow, StoredInstant, VersionRow};
use openehr_store::{CommitOutcome, Result, Store, StoreError, ddl_script};
use rusqlite::{Connection, OptionalExtension as _, params};
const ENGINE: &str = "SQLite";
pub struct SqliteStore {
connection: Connection,
}
impl SqliteStore {
pub fn in_memory() -> Result<Self> {
Self::from_connection(Connection::open_in_memory().map_err(|e| engine(&e))?)
}
pub fn open(path: &std::path::Path) -> Result<Self> {
Self::from_connection(Connection::open(path).map_err(|e| engine(&e))?)
}
pub fn from_connection(connection: Connection) -> Result<Self> {
connection
.execute_batch("PRAGMA foreign_keys = ON;")
.map_err(|e| engine(&e))?;
Ok(Self { connection })
}
#[must_use]
pub fn connection(&self) -> &Connection {
&self.connection
}
fn digest_column(row: &rusqlite::Row<'_>, name: &str) -> rusqlite::Result<[u8; 32]> {
let raw: Vec<u8> = row.get(name)?;
<[u8; 32]>::try_from(raw.as_slice()).map_err(|_| {
rusqlite::Error::FromSqlConversionFailure(
0,
rusqlite::types::Type::Blob,
format!("{name} is not 32 bytes").into(),
)
})
}
fn read_version(row: &rusqlite::Row<'_>) -> rusqlite::Result<VersionRow> {
Ok(VersionRow {
uid: row.get("uid")?,
versioned_object_uid: row.get("versioned_object_uid")?,
creating_system_id: row.get("creating_system_id")?,
trunk_version: row.get("trunk_version")?,
branch_number: row.get("branch_number")?,
branch_version: row.get("branch_version")?,
preceding_version_uid: row.get("preceding_version_uid")?,
lifecycle_state_code: row.get("lifecycle_state_code")?,
is_deleted: row.get::<_, i64>("is_deleted")? != 0,
contribution_uid: row.get("contribution_uid")?,
audit_system_id: row.get("audit_system_id")?,
audit_change_type_code: row.get("audit_change_type_code")?,
audit_committer_name: row.get("audit_committer_name")?,
audit_time_committed: StoredInstant {
text: row.get("audit_time_committed_text")?,
utc_seconds: row.get("audit_time_committed_utc")?,
},
data_json: row.get("data_json")?,
audit_description: row.get("audit_description")?,
signature: row.get("signature")?,
attestations_json: row.get("attestations_json")?,
other_input_version_uids_json: row.get("other_input_version_uids_json")?,
chain: openehr_store::record::ChainColumns {
previous: Self::digest_column(row, "chain_previous")?,
content: Self::digest_column(row, "chain_content")?,
digest: Self::digest_column(row, "chain_digest")?,
tag_key_id: row.get("chain_tag_key_id")?,
tag_mac: row
.get::<_, Option<Vec<u8>>>("chain_tag_mac")?
.map(|v| <[u8; 32]>::try_from(v.as_slice()))
.transpose()
.map_err(|_| {
rusqlite::Error::FromSqlConversionFailure(
0,
rusqlite::types::Type::Blob,
"chain_tag_mac is not 32 bytes".into(),
)
})?,
},
})
}
const VERSION_COLUMNS: &'static str = "uid, versioned_object_uid, creating_system_id, \
trunk_version, branch_number, branch_version, preceding_version_uid, \
lifecycle_state_code, is_deleted, contribution_uid, audit_system_id, \
audit_change_type_code, audit_committer_name, audit_time_committed_text, \
audit_time_committed_utc, data_json, audit_description, signature, \
attestations_json, other_input_version_uids_json, chain_previous, \
chain_content, chain_digest, chain_tag_key_id, chain_tag_mac";
fn check_schema_version(&self) -> Result<()> {
let recorded: Option<i64> = self
.connection
.query_row(
"SELECT version FROM openehr_schema_version LIMIT 1",
[],
|row| row.get(0),
)
.optional()
.unwrap_or(None);
if let Some(found) = recorded {
if found != openehr_store::SCHEMA_VERSION {
return Err(StoreError::SchemaVersionMismatch {
found,
expected: openehr_store::SCHEMA_VERSION,
});
}
return Ok(());
}
let legacy: Option<i64> = self
.connection
.query_row("SELECT count(*) FROM openehr_ehr", [], |row| row.get(0))
.optional()
.unwrap_or(None);
if legacy.is_some_and(|n| n > 0) {
return Err(StoreError::SchemaVersionMismatch {
found: 0,
expected: openehr_store::SCHEMA_VERSION,
});
}
Ok(())
}
fn record_schema_version(&self) -> Result<()> {
let now = StoredInstant::from_date_time(&"1970-01-01T00:00:00Z".parse().expect("literal"));
self.connection
.execute(
"INSERT OR IGNORE INTO openehr_schema_version (version, applied_text, applied_utc) \
VALUES (?1, ?2, ?3)",
params![openehr_store::SCHEMA_VERSION, now.text, now.utc_seconds],
)
.map(|_| ())
.map_err(|e| engine(&e))
}
fn chain_digest_of(&self, uid: &str) -> Result<[u8; 32]> {
let raw: Vec<u8> = self
.connection
.query_row(
"SELECT chain_digest FROM openehr_version WHERE uid = ?1",
params![uid],
|row| row.get(0),
)
.map_err(|e| engine(&e))?;
<[u8; 32]>::try_from(raw.as_slice()).map_err(|_| StoreError::Engine {
engine: ENGINE,
message: "chain_digest is not 32 bytes".to_owned(),
})
}
}
fn commit_conflict(error: &rusqlite::Error) -> Option<StoreError> {
use rusqlite::ErrorCode::ConstraintViolation;
let rusqlite::Error::SqliteFailure(code, Some(message)) = error else {
return None;
};
if code.code != ConstraintViolation {
return None;
}
if message.contains("openehr_version.uid") {
Some(StoreError::Commit(CommitError::DuplicateVersion))
} else if message.contains("ix_version_container_trunk") {
Some(StoreError::Commit(CommitError::NotLatest))
} else {
None
}
}
fn hex32(bytes: &[u8; 32]) -> String {
use std::fmt::Write as _;
let mut out = String::with_capacity(64);
for b in bytes {
let _ = write!(out, "{b:02x}");
}
out
}
fn engine(error: &rusqlite::Error) -> StoreError {
if let Some(conflict) = commit_conflict(error) {
return conflict;
}
StoreError::Engine {
engine: ENGINE,
message: error.to_string(),
}
}
impl Store for SqliteStore {
fn engine(&self) -> &'static str {
ENGINE
}
fn install(&mut self) -> Result<()> {
self.check_schema_version()?;
self.connection
.execute_batch(&ddl_script(&SqliteDialect))
.map_err(|e| engine(&e))?;
self.record_schema_version()
}
fn create_ehr(&mut self, ehr: &Ehr) -> Result<()> {
ehr.validate_ok()?;
let id = ehr.ehr_id().to_string();
let existing: Option<String> = self
.connection
.query_row(
"SELECT ehr_id FROM openehr_ehr WHERE ehr_id = ?1",
params![id],
|row| row.get(0),
)
.optional()
.map_err(|e| engine(&e))?;
if existing.is_some() {
return Err(StoreError::Conflict { kind: "ehr", id });
}
let created = StoredInstant::from_date_time(ehr.time_created().value());
self.connection
.execute(
"INSERT INTO openehr_ehr \
(ehr_id, system_id, time_created_text, time_created_utc, ehr_status_uid, ehr_access_uid) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![
id,
ehr.system_id().to_string(),
created.text,
created.utc_seconds,
ehr.ehr_status().id().to_string(),
ehr.ehr_access().id().to_string(),
],
)
.map_err(|e| engine(&e))?;
Ok(())
}
fn get_ehr(&self, ehr_id: &HierObjectId) -> Result<Ehr> {
let id = ehr_id.to_string();
let row: Option<(String, String, String, String)> = self
.connection
.query_row(
"SELECT system_id, time_created_text, ehr_status_uid, ehr_access_uid \
FROM openehr_ehr WHERE ehr_id = ?1",
params![id],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
)
.optional()
.map_err(|e| engine(&e))?;
let Some((system_id, created, status_uid, access_uid)) = row else {
return Err(StoreError::NotFound { kind: "ehr", id });
};
let reference = |uid: &str, ty: &'static str| -> Result<ObjectRef> {
Ok(ObjectRef::new(
"local",
ty,
ObjectId::HierObjectId(uid.parse()?),
)?)
};
Ok(Ehr::new(
system_id.parse()?,
ehr_id.clone(),
reference(&status_uid, "VERSIONED_EHR_STATUS")?,
reference(&access_uid, "VERSIONED_EHR_ACCESS")?,
DvDateTime::new(&created)?,
)?)
}
fn create_contribution(
&mut self,
ehr_id: &HierObjectId,
contribution: &Contribution,
) -> Result<()> {
let uid = contribution.uid().to_string();
let audit = contribution.audit();
let committed = StoredInstant::from_date_time(audit.time_committed().value());
self.connection
.execute(
"INSERT INTO openehr_contribution \
(uid, ehr_id, audit_change_type_code, audit_system_id, audit_committer_name, \
audit_time_committed_text, audit_time_committed_utc) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![
uid,
ehr_id.to_string(),
audit.change_type_code(),
audit.system_id(),
audit.committer().name(),
committed.text,
committed.utc_seconds,
],
)
.map_err(|e| match e {
rusqlite::Error::SqliteFailure(f, _)
if f.code == rusqlite::ErrorCode::ConstraintViolation =>
{
StoreError::Conflict {
kind: "contribution",
id: uid.clone(),
}
}
ref other => engine(other),
})?;
Ok(())
}
#[allow(clippy::too_many_lines)]
fn commit_composition(
&mut self,
ehr_id: &HierObjectId,
version: &Version<Composition>,
contribution_uid: &str,
) -> Result<CommitOutcome> {
version.validate_ok()?;
self.get_ehr(ehr_id)?;
let container_uid = version.uid().object_id().to_string();
let head: Option<(String, i64)> = self
.connection
.query_row(
"SELECT uid, trunk_version FROM openehr_version \
WHERE versioned_object_uid = ?1 \
ORDER BY trunk_version DESC, branch_number DESC, branch_version DESC LIMIT 1",
params![container_uid],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()
.map_err(|e| engine(&e))?;
let uid = version.uid().to_string();
let already: Option<String> = self
.connection
.query_row(
"SELECT uid FROM openehr_version WHERE uid = ?1",
params![uid],
|row| row.get(0),
)
.optional()
.map_err(|e| engine(&e))?;
if already.is_some() {
return Err(StoreError::Commit(CommitError::DuplicateVersion));
}
match (&head, version.preceding_version_uid()) {
(None, None) => {}
(None, Some(_)) | (Some(_), None) => {
return Err(StoreError::Commit(CommitError::PrecedingVersionMismatch));
}
(Some((latest, _)), Some(preceding)) => {
if latest != &preceding.to_string() {
return Err(StoreError::Commit(CommitError::NotLatest));
}
}
}
let previous_digest = head
.as_ref()
.map(|(uid, _)| self.chain_digest_of(uid))
.transpose()?;
let row = VersionRow::project(version, contribution_uid, previous_digest, None)?;
let created_container = head.is_none();
let transaction = self
.connection
.unchecked_transaction()
.map_err(|e| engine(&e))?;
if created_container {
let created =
StoredInstant::from_date_time(version.commit_audit().time_committed().value());
transaction
.execute(
"INSERT INTO openehr_versioned_object \
(uid, ehr_id, rm_type, time_created_text, time_created_utc) \
VALUES (?1, ?2, ?3, ?4, ?5)",
params![
container_uid,
ehr_id.to_string(),
"COMPOSITION",
created.text,
created.utc_seconds
],
)
.map_err(|e| engine(&e))?;
}
transaction
.execute(
"INSERT INTO openehr_version \
(uid, versioned_object_uid, creating_system_id, trunk_version, branch_number, \
branch_version, preceding_version_uid, lifecycle_state_code, is_deleted, \
contribution_uid, audit_system_id, audit_change_type_code, \
audit_committer_name, audit_time_committed_text, audit_time_committed_utc, \
data_json, audit_description, signature, attestations_json, \
other_input_version_uids_json, chain_previous, chain_content, chain_digest, \
chain_tag_key_id, chain_tag_mac) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, \
?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25)",
params![
row.uid,
row.versioned_object_uid,
row.creating_system_id,
row.trunk_version,
row.branch_number,
row.branch_version,
row.preceding_version_uid,
row.lifecycle_state_code,
i64::from(row.is_deleted),
row.contribution_uid,
row.audit_system_id,
row.audit_change_type_code,
row.audit_committer_name,
row.audit_time_committed.text,
row.audit_time_committed.utc_seconds,
row.data_json,
row.audit_description,
row.signature,
row.attestations_json,
row.other_input_version_uids_json,
row.chain.previous.as_slice(),
row.chain.content.as_slice(),
row.chain.digest.as_slice(),
row.chain.tag_key_id,
row.chain.tag_mac.map(|m| m.to_vec()),
],
)
.map_err(|e| engine(&e))?;
if let Some(composition) = version.data() {
let index = CompositionIndexRow::project(&row.uid, &ehr_id.to_string(), composition)?;
transaction
.execute(
"INSERT INTO openehr_composition_index \
(version_uid, ehr_id, archetype_id, template_id, category_code, \
composer_name, language_code, territory_code, setting_code, \
context_start_text, context_start_utc, context_end_text, context_end_utc) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
params![
index.version_uid,
index.ehr_id,
index.archetype_id,
index.template_id,
index.category_code,
index.composer_name,
index.language_code,
index.territory_code,
index.setting_code,
index.context_start.as_ref().map(|i| i.text.clone()),
index.context_start.as_ref().and_then(|i| i.utc_seconds),
index.context_end.as_ref().map(|i| i.text.clone()),
index.context_end.as_ref().and_then(|i| i.utc_seconds),
],
)
.map_err(|e| engine(&e))?;
}
transaction.commit().map_err(|e| engine(&e))?;
Ok(CommitOutcome {
version_uid: version.uid().clone(),
created_container,
})
}
fn get_version(&self, uid: &ObjectVersionId) -> Result<VersionRow> {
let id = uid.to_string();
self.connection
.query_row(
&format!(
"SELECT {} FROM openehr_version WHERE uid = ?1",
Self::VERSION_COLUMNS
),
params![id],
Self::read_version,
)
.optional()
.map_err(|e| engine(&e))?
.ok_or(StoreError::NotFound {
kind: "version",
id,
})
}
fn latest_version(&self, versioned_object_uid: &HierObjectId) -> Result<VersionRow> {
let id = versioned_object_uid.to_string();
self.connection
.query_row(
&format!(
"SELECT {} FROM openehr_version WHERE versioned_object_uid = ?1 \
ORDER BY trunk_version DESC, branch_number DESC, branch_version DESC LIMIT 1",
Self::VERSION_COLUMNS
),
params![id],
Self::read_version,
)
.optional()
.map_err(|e| engine(&e))?
.ok_or(StoreError::NotFound {
kind: "versioned_object",
id,
})
}
fn version_at_time(
&self,
versioned_object_uid: &HierObjectId,
at: &DvDateTime,
) -> Result<VersionRow> {
let id = versioned_object_uid.to_string();
let Some(at_seconds) = StoredInstant::from_date_time(at.value()).utc_seconds else {
return Err(StoreError::NotFound {
kind: "version",
id,
});
};
self.connection
.query_row(
&format!(
"SELECT {} FROM openehr_version \
WHERE versioned_object_uid = ?1 \
AND audit_time_committed_utc IS NOT NULL \
AND audit_time_committed_utc <= ?2 \
ORDER BY audit_time_committed_utc DESC, trunk_version DESC LIMIT 1",
Self::VERSION_COLUMNS
),
params![id, at_seconds],
Self::read_version,
)
.optional()
.map_err(|e| engine(&e))?
.ok_or(StoreError::NotFound {
kind: "version",
id,
})
}
fn all_versions(&self, versioned_object_uid: &HierObjectId) -> Result<Vec<VersionRow>> {
let mut statement = self
.connection
.prepare(&format!(
"SELECT {} FROM openehr_version WHERE versioned_object_uid = ?1 \
ORDER BY trunk_version ASC, branch_number ASC, branch_version ASC",
Self::VERSION_COLUMNS
))
.map_err(|e| engine(&e))?;
let rows = statement
.query_map(
params![versioned_object_uid.to_string()],
Self::read_version,
)
.map_err(|e| engine(&e))?;
rows.collect::<rusqlite::Result<Vec<_>>>()
.map_err(|e| engine(&e))
}
fn chain_checkpoint(&self, versioned_object_uid: &HierObjectId) -> Result<String> {
let versions = self.all_versions(versioned_object_uid)?;
let head = versions
.last()
.map_or_else(|| "0".repeat(64), |v| hex32(&v.chain.digest));
Ok(format!(
"entries={} head={} last_version={}",
versions.len(),
head,
versions.last().map_or("-", |v| v.uid.as_str())
))
}
fn find_compositions_by_archetype(
&self,
ehr_id: &HierObjectId,
archetype_id: &str,
) -> Result<Vec<CompositionIndexRow>> {
let mut statement = self
.connection
.prepare(
"SELECT version_uid, ehr_id, archetype_id, template_id, category_code, \
composer_name, language_code, territory_code, setting_code, \
context_start_text, context_start_utc, context_end_text, context_end_utc \
FROM openehr_composition_index \
WHERE ehr_id = ?1 AND archetype_id = ?2 \
ORDER BY version_uid",
)
.map_err(|e| engine(&e))?;
let rows = statement
.query_map(params![ehr_id.to_string(), archetype_id], |row| {
let instant = |text: Option<String>, utc: Option<i64>| {
text.map(|text| StoredInstant {
text,
utc_seconds: utc,
})
};
Ok(CompositionIndexRow {
version_uid: row.get("version_uid")?,
ehr_id: row.get("ehr_id")?,
archetype_id: row.get("archetype_id")?,
template_id: row.get("template_id")?,
category_code: row.get("category_code")?,
composer_name: row.get("composer_name")?,
language_code: row.get("language_code")?,
territory_code: row.get("territory_code")?,
setting_code: row.get("setting_code")?,
context_start: instant(
row.get("context_start_text")?,
row.get("context_start_utc")?,
),
context_end: instant(row.get("context_end_text")?, row.get("context_end_utc")?),
})
})
.map_err(|e| engine(&e))?;
rows.collect::<rusqlite::Result<Vec<_>>>()
.map_err(|e| engine(&e))
}
}